diff --git a/.github/workflows/git-commands.yaml b/.github/workflows/git-commands.yaml index b866d9b0f9..7c7a05e81e 100644 --- a/.github/workflows/git-commands.yaml +++ b/.github/workflows/git-commands.yaml @@ -40,11 +40,12 @@ jobs: with: github-token: ${{ steps.app.outputs.token }} script: | + const runUrl = `https://github.com/${context.repo.owner}/${context.repo.repo}/actions/runs/${context.runId}`; github.rest.issues.createComment({ issue_number: context.issue.number, owner: context.repo.owner, repo: context.repo.repo, - body: 'Starting sqlx update...' + body: `Starting sqlx update...\n\n[View workflow run](${runUrl})` }) - name: Checkout repository diff --git a/.github/workflows/weekly-pr-summary.yml b/.github/workflows/weekly-pr-summary.yml index ea0d115919..821fd86b05 100644 --- a/.github/workflows/weekly-pr-summary.yml +++ b/.github/workflows/weekly-pr-summary.yml @@ -49,14 +49,14 @@ jobs: - **Bug Fixes**: PRs with titles starting with "fix:", "bug:", or containing "fix", "resolve", "patch" - **Other**: All remaining PRs (improvements, refactors, docs, chores, etc.) - 4. **Gather Details**: For each merged PR, include: + 4. **Gather Details**: For each feature and bug fix merged PR, include: - Full PR title (NO truncation, NO links) - Author (extract login from author.login in JSON) - Brief summary: Use `gh pr view --json body` to get PR description, then extract first paragraph or key points (1-2 sentences max) 5. **Character Limit Enforcement**: - - The final summary MUST be under 6000 characters - - If the summary exceeds 6000 characters, truncate PR descriptions (NOT titles) and add at the end: "_and X more PRs_" where X is the count of omitted PRs + - The final summary MUST be under 5000 characters + - If the summary exceeds 5000 characters, truncate PR descriptions (NOT titles) and add at the end: "_and X more PRs_" where X is the count of omitted PRs 6. **Save Summary to Markdown File**: Write the summary to a file for webhook delivery: - Save the complete formatted markdown to: `summary.md` @@ -65,23 +65,19 @@ jobs: ## Output Format: ```markdown - #### 📈 Weekly overview + ### 📈 Weekly overview - **Total merged**: X - **Features**: Y - **Bug Fixes**: Z - **Other**: W - #### ✨ Features (Y) - • **[Full PR Title]** by @username - [brief impact description] - • **[Full PR Title]** by @username - [brief impact description] + ### ✨ Features (Y) + - **[Full PR Title]** by @username - [brief impact description] + - **[Full PR Title]** by @username - [brief impact description] - #### 🐛 Bug Fixes (Z) - • **[Full PR Title]** by @username - [brief impact description] - • **[Full PR Title]** by @username - [brief impact description] - - #### 🔧 Other (W) - • **[Full PR Title]** by @username - [brief impact description] - • **[Full PR Title]** by @username - [brief impact description] + ### 🐛 Bug Fixes (Z) + - **[Full PR Title]** by @username - [brief impact description] + - **[Full PR Title]** by @username - [brief impact description] _and X more PRs_ ``` @@ -89,9 +85,9 @@ jobs: ## Important Notes: - **CRITICAL**: ONLY include PRs with state "merged" from the last 7 days - **CRITICAL**: EXCLUDE all PRs with titles starting with "chore: release" or "chore(release)" - - **CRITICAL**: Total character count MUST be under 6000 characters - - Only use #### markdown headers for major sections and emoji indicators - - Use bullet points (•) for individual PR entries - more compact than paragraphs + - **CRITICAL**: Total character count MUST be under 5000 characters + - Count the number of "Other" PRs but do not include a section for them in the output + - Only use ### markdown headers for major sections and emoji indicators - NO links to PRs - NO merged date in output - NEVER truncate PR titles - show full titles @@ -100,7 +96,7 @@ jobs: - If a PR has no description, write "(No description provided)" - Extract meaningful summary from PR body - look for the first paragraph or key bullet points - Parse JSON responses carefully using `jq` or similar tools - - If summary exceeds 6000 chars, shorten PR descriptions and add "_and X more PRs_" at the end + - If summary exceeds 5000 chars, shorten PR descriptions and add "_and X more PRs_" at the end - Count PRs in each category and display in both overview and section headers ## Saving the Markdown Output: diff --git a/CHANGELOG.md b/CHANGELOG.md index df8800289b..e20b7f4631 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,23 @@ # Changelog +## [1.573.5](https://github.com/windmill-labs/windmill/compare/v1.573.4...v1.573.5) (2025-11-07) + + +### Bug Fixes + +* dep jobs improvements ([#7081](https://github.com/windmill-labs/windmill/issues/7081)) ([ad43680](https://github.com/windmill-labs/windmill/commit/ad43680bb979f80db3bdb182ccb5bc61ca385ceb)) +* modal action in App AgGrid ([#7085](https://github.com/windmill-labs/windmill/issues/7085)) ([b07c35f](https://github.com/windmill-labs/windmill/commit/b07c35fe1c9dbc4955ce47766ef92bd5cfdac4c8)) + +## [1.573.4](https://github.com/windmill-labs/windmill/compare/v1.573.3...v1.573.4) (2025-11-07) + + +### Bug Fixes + +* **bun:** support relative imports without .ts but starting with /u/ or /f/ ([9e3ea76](https://github.com/windmill-labs/windmill/commit/9e3ea7657568d6561abcdfd23cc8a210432e4050)) +* create git branch right before creating the workspace fork to catch errors and have a coherent fork point ([#7073](https://github.com/windmill-labs/windmill/issues/7073)) ([192fecc](https://github.com/windmill-labs/windmill/commit/192fecc86f54cf2a69ec7dddc466139b40933ac9)) +* improve priority for child jobs ran from api ([5f7e0b7](https://github.com/windmill-labs/windmill/commit/5f7e0b7244e27533d9a8d1551e8affeb17eb94c0)) +* make it impossible to have non linear script history even in edge condition ([3bda6da](https://github.com/windmill-labs/windmill/commit/3bda6da1683589180cad0780dd87b391a04851f8)) + ## [1.573.3](https://github.com/windmill-labs/windmill/compare/v1.573.2...v1.573.3) (2025-11-06) diff --git a/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json b/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json index d29a18c691..e7ed0aee65 100644 --- a/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json +++ b/backend/.sqlx/query-07168aaf14cb6beff0ad4274b441f7f387f5055c47f493271d26731336257384.json @@ -46,11 +46,11 @@ ] }, "nullable": [ - true, - true, - true, - true, - true, + false, + false, + false, + false, + false, true, true ] diff --git a/backend/.sqlx/query-0c0f3909b80c35210fc64c685905308621f9135c2c45a2fa0531ea750387da1f.json b/backend/.sqlx/query-0c0f3909b80c35210fc64c685905308621f9135c2c45a2fa0531ea750387da1f.json new file mode 100644 index 0000000000..cec59be959 --- /dev/null +++ b/backend/.sqlx/query-0c0f3909b80c35210fc64c685905308621f9135c2c45a2fa0531ea750387da1f.json @@ -0,0 +1,25 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT \n CASE \n WHEN flow_version.id IS NOT NULL THEN\n (flow_version.value -> 'flow_env' -> $3) #> $4\n ELSE\n (root_job.raw_flow -> 'flow_env' -> $3) #> $4\n END AS \"flow_env: sqlx::types::Json>\"\n FROM \n v2_job current_job\n JOIN \n v2_job root_job ON root_job.id = COALESCE(current_job.root_job, current_job.flow_innermost_root_job, current_job.parent_job, current_job.id)\n AND root_job.workspace_id = current_job.workspace_id\n LEFT JOIN\n flow_version ON flow_version.id = root_job.runnable_id\n AND flow_version.path = root_job.runnable_path\n AND flow_version.workspace_id = root_job.workspace_id\n WHERE \n current_job.id = $1 AND \n current_job.workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "flow_env: sqlx::types::Json>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text", + "Text", + "TextArray" + ] + }, + "nullable": [ + null + ] + }, + "hash": "0c0f3909b80c35210fc64c685905308621f9135c2c45a2fa0531ea750387da1f" +} diff --git a/backend/.sqlx/query-1fc43faf3a7ea15be82b1666273836901cc6ba17c7d5e194b50b9204f264de27.json b/backend/.sqlx/query-1fc43faf3a7ea15be82b1666273836901cc6ba17c7d5e194b50b9204f264de27.json new file mode 100644 index 0000000000..d20ff3ba93 --- /dev/null +++ b/backend/.sqlx/query-1fc43faf3a7ea15be82b1666273836901cc6ba17c7d5e194b50b9204f264de27.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT 1 FROM script WHERE hash = $1 AND workspace_id = $2 FOR UPDATE", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "?column?", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Int8", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "1fc43faf3a7ea15be82b1666273836901cc6ba17c7d5e194b50b9204f264de27" +} diff --git a/backend/.sqlx/query-38baa6cf7d1c2532d38486a01e6d27ac1a58540a51e97dec5613e9bc0791e890.json b/backend/.sqlx/query-38baa6cf7d1c2532d38486a01e6d27ac1a58540a51e97dec5613e9bc0791e890.json new file mode 100644 index 0000000000..eeb0418109 --- /dev/null +++ b/backend/.sqlx/query-38baa6cf7d1c2532d38486a01e6d27ac1a58540a51e97dec5613e9bc0791e890.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO ai_agent_memory (workspace_id, conversation_id, step_id, messages, created_at, updated_at)\n VALUES ($1, $2, $3, $4, NOW(), NOW())\n ON CONFLICT (workspace_id, conversation_id, step_id)\n DO UPDATE SET\n messages = EXCLUDED.messages,\n updated_at = NOW()", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Uuid", + "Varchar", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "38baa6cf7d1c2532d38486a01e6d27ac1a58540a51e97dec5613e9bc0791e890" +} diff --git a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json index 713ccb9dd3..36ddb8ab9f 100644 --- a/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json +++ b/backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json @@ -15,7 +15,7 @@ ] }, "nullable": [ - null + true ] }, "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55" diff --git a/backend/.sqlx/query-6a1005b8ae5326c5d5955534e9228b6bd524253971ae09cbe100c0d6d42e6cdc.json b/backend/.sqlx/query-6a1005b8ae5326c5d5955534e9228b6bd524253971ae09cbe100c0d6d42e6cdc.json new file mode 100644 index 0000000000..647e56619c --- /dev/null +++ b/backend/.sqlx/query-6a1005b8ae5326c5d5955534e9228b6bd524253971ae09cbe100c0d6d42e6cdc.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT messages FROM ai_agent_memory\n WHERE workspace_id = $1 AND conversation_id = $2 AND step_id = $3", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "messages", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text", + "Uuid", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "6a1005b8ae5326c5d5955534e9228b6bd524253971ae09cbe100c0d6d42e6cdc" +} diff --git a/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json b/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json index 99269c9851..54e94cfb8f 100644 --- a/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json +++ b/backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json @@ -18,8 +18,8 @@ "Left": [] }, "nullable": [ - true, - false + false, + true ] }, "hash": "b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76" diff --git a/backend/.sqlx/query-d1d9cb0a7043760564171f481dd73f04eaa2c7d5c1ddc0501bc70c3fb34a07be.json b/backend/.sqlx/query-d1d9cb0a7043760564171f481dd73f04eaa2c7d5c1ddc0501bc70c3fb34a07be.json new file mode 100644 index 0000000000..2243c6f719 --- /dev/null +++ b/backend/.sqlx/query-d1d9cb0a7043760564171f481dd73f04eaa2c7d5c1ddc0501bc70c3fb34a07be.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM ai_agent_memory\n WHERE workspace_id = $1 AND conversation_id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "d1d9cb0a7043760564171f481dd73f04eaa2c7d5c1ddc0501bc70c3fb34a07be" +} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index d3a95f582b..0485c7aa07 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -788,9 +788,9 @@ dependencies = [ [[package]] name = "aws-credential-types" -version = "1.2.8" +version = "1.2.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "faf26925f4a5b59eb76722b63c2892b1d70d06fa053c72e4a100ec308c1d47bc" +checksum = "86590e57ea40121d47d3f2e131bfd873dea15d78dc2f4604f4734537ad9e56c4" dependencies = [ "aws-smithy-async", "aws-smithy-runtime-api", @@ -1945,9 +1945,9 @@ dependencies = [ [[package]] name = "cc" -version = "1.2.44" +version = "1.2.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "37521ac7aabe3d13122dc382493e20c9416f299d2ccd5b3a5340a2570cdeb0f3" +checksum = "35900b6c8d709fb1d854671ae27aeaa9eec2f8b01b364e1619a40da3e6fe2afe" dependencies = [ "find-msvc-tools", "jobserver", @@ -10296,9 +10296,9 @@ dependencies = [ [[package]] name = "quote" -version = "1.0.41" +version = "1.0.42" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ce25767e7b499d1b604768e7cde645d14cc8584231ea6b295e9c9eb22c02e1d1" +checksum = "a338cc41d27e6cc6dce6cefc13a0729dfbb81c262b1f519331575dd80ef3067f" dependencies = [ "proc-macro2", ] @@ -15137,7 +15137,7 @@ checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" [[package]] name = "windmill" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "aws-sdk-config", @@ -15197,7 +15197,7 @@ dependencies = [ [[package]] name = "windmill-api" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "argon2", @@ -15317,7 +15317,7 @@ dependencies = [ [[package]] name = "windmill-api-client" -version = "1.573.3" +version = "1.573.5" dependencies = [ "base64 0.22.1", "chrono", @@ -15332,7 +15332,7 @@ dependencies = [ [[package]] name = "windmill-audit" -version = "1.573.3" +version = "1.573.5" dependencies = [ "chrono", "lazy_static", @@ -15346,7 +15346,7 @@ dependencies = [ [[package]] name = "windmill-autoscaling" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "axum", @@ -15365,7 +15365,7 @@ dependencies = [ [[package]] name = "windmill-common" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "async-recursion", @@ -15450,7 +15450,7 @@ dependencies = [ [[package]] name = "windmill-git-sync" -version = "1.573.3" +version = "1.573.5" dependencies = [ "regex", "serde", @@ -15465,7 +15465,7 @@ dependencies = [ [[package]] name = "windmill-indexer" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "bytes", @@ -15489,7 +15489,7 @@ dependencies = [ [[package]] name = "windmill-macros" -version = "1.573.3" +version = "1.573.5" dependencies = [ "itertools 0.14.0", "lazy_static", @@ -15501,7 +15501,7 @@ dependencies = [ [[package]] name = "windmill-parser" -version = "1.573.3" +version = "1.573.5" dependencies = [ "convert_case 0.6.0", "serde", @@ -15510,7 +15510,7 @@ dependencies = [ [[package]] name = "windmill-parser-bash" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "lazy_static", @@ -15522,7 +15522,7 @@ dependencies = [ [[package]] name = "windmill-parser-csharp" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "serde_json", @@ -15534,7 +15534,7 @@ dependencies = [ [[package]] name = "windmill-parser-go" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "gosyn", @@ -15546,7 +15546,7 @@ dependencies = [ [[package]] name = "windmill-parser-graphql" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "lazy_static", @@ -15558,7 +15558,7 @@ dependencies = [ [[package]] name = "windmill-parser-java" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "serde_json", @@ -15570,7 +15570,7 @@ dependencies = [ [[package]] name = "windmill-parser-nu" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "nu-parser", @@ -15581,7 +15581,7 @@ dependencies = [ [[package]] name = "windmill-parser-php" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "itertools 0.14.0", @@ -15592,7 +15592,7 @@ dependencies = [ [[package]] name = "windmill-parser-py" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "itertools 0.14.0", @@ -15604,7 +15604,7 @@ dependencies = [ [[package]] name = "windmill-parser-py-imports" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "async-recursion", @@ -15627,7 +15627,7 @@ dependencies = [ [[package]] name = "windmill-parser-ruby" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "lazy_static", @@ -15641,7 +15641,7 @@ dependencies = [ [[package]] name = "windmill-parser-rust" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "convert_case 0.6.0", @@ -15658,7 +15658,7 @@ dependencies = [ [[package]] name = "windmill-parser-sql" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "lazy_static", @@ -15672,7 +15672,7 @@ dependencies = [ [[package]] name = "windmill-parser-ts" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "lazy_static", @@ -15690,7 +15690,7 @@ dependencies = [ [[package]] name = "windmill-parser-yaml" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "serde", @@ -15701,7 +15701,7 @@ dependencies = [ [[package]] name = "windmill-queue" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "async-recursion", @@ -15735,7 +15735,7 @@ dependencies = [ [[package]] name = "windmill-sql-datatype-parser-wasm" -version = "1.573.3" +version = "1.573.5" dependencies = [ "wasm-bindgen", "wasm-bindgen-test", @@ -15745,7 +15745,7 @@ dependencies = [ [[package]] name = "windmill-worker" -version = "1.573.3" +version = "1.573.5" dependencies = [ "anyhow", "async-once-cell", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 68e5165f25..26cacfdb8a 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "windmill" -version = "1.573.3" +version = "1.573.5" authors.workspace = true edition.workspace = true @@ -33,7 +33,7 @@ members = [ exclude = ["./windmill-duckdb-ffi-internal"] [workspace.package] -version = "1.573.3" +version = "1.573.5" authors = ["Ruben Fiszel "] edition = "2021" diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 3629450170..b9762186b4 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -ab6116235f7869cef306996059f8663f7e831b8a +5b7afe50da442441747e7a8f6ef461c96faa9dc2 diff --git a/backend/migrations/20251104135208_ai_agent_memory.down.sql b/backend/migrations/20251104135208_ai_agent_memory.down.sql new file mode 100644 index 0000000000..6b659bcadc --- /dev/null +++ b/backend/migrations/20251104135208_ai_agent_memory.down.sql @@ -0,0 +1,4 @@ +-- Add down migration script here + +-- Drop table +DROP TABLE IF EXISTS ai_agent_memory; diff --git a/backend/migrations/20251104135208_ai_agent_memory.up.sql b/backend/migrations/20251104135208_ai_agent_memory.up.sql new file mode 100644 index 0000000000..168659eb72 --- /dev/null +++ b/backend/migrations/20251104135208_ai_agent_memory.up.sql @@ -0,0 +1,16 @@ +-- Add up migration script here + +-- Create ai_agent_memory table for storing AI agent step memory when S3 is unavailable +CREATE TABLE ai_agent_memory ( + workspace_id VARCHAR(50) NOT NULL, + conversation_id UUID NOT NULL, + step_id VARCHAR(255) NOT NULL, + messages JSONB NOT NULL, + created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(), + updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(), + PRIMARY KEY (workspace_id, conversation_id, step_id) +); + +-- Grant permissions +GRANT ALL ON ai_agent_memory TO windmill_admin; +GRANT ALL ON ai_agent_memory TO windmill_user; diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 3fc7fc0788..b8a72ff3ab 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -1,7 +1,7 @@ openapi: "3.0.3" info: - version: 1.573.3 + version: 1.573.5 title: Windmill API contact: diff --git a/backend/windmill-api/src/flow_conversations.rs b/backend/windmill-api/src/flow_conversations.rs index 6f046c4738..0cbdb7aa55 100644 --- a/backend/windmill-api/src/flow_conversations.rs +++ b/backend/windmill-api/src/flow_conversations.rs @@ -11,7 +11,7 @@ use uuid::Uuid; use crate::db::ApiAuthed; use windmill_common::{ - db::UserDB, + db::{UserDB, DB}, error::{JsonResult, Result}, flow_conversations::MessageType, utils::{not_found_if_none, paginate, Pagination}, @@ -154,6 +154,7 @@ pub async fn get_or_create_conversation_with_id( async fn delete_conversation( authed: ApiAuthed, Extension(user_db): Extension, + Extension(db): Extension, Path((w_id, conversation_id)): Path<(String, Uuid)>, ) -> Result { let mut tx = user_db.clone().begin(&authed).await?; @@ -185,10 +186,14 @@ async fn delete_conversation( // Delete associated memory in background (non-blocking cleanup) let w_id_clone = w_id.clone(); + let db_clone = db.clone(); tokio::spawn(async move { - if let Err(e) = - windmill_worker::memory_oss::delete_conversation_memory(&w_id_clone, conversation_id) - .await + if let Err(e) = windmill_worker::memory_oss::delete_conversation_memory( + &db_clone, + &w_id_clone, + conversation_id, + ) + .await { tracing::error!( "Failed to delete memory for conversation {} in workspace {}: {:?}", diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 0f4e8db038..9952ca1ddd 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -1634,6 +1634,7 @@ mod tests { early_return: None, concurrency_key: None, chat_input_enabled: None, + flow_env: None, debounce_key: None, debounce_delay_s: None, }; diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 33beb32ec6..d2de1a8f49 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -281,6 +281,10 @@ pub fn workspaced_service() -> Router { "/result_by_id/:job_id/:node_id", get(get_result_by_id).layer(cors.clone()), ) + .route( + "/flow_env_by_flow_job_id/:flow_job_id/:var_name", + get(get_flow_env_by_flow_job_id).layer(cors.clone()), + ) .route("/run/dependencies", post(run_dependencies_job)) .route("/run/flow_dependencies", post(run_flow_dependencies_job)) .route( @@ -376,6 +380,59 @@ async fn get_root_job( Ok(Json(res)) } +async fn get_flow_env_by_flow_job_id( + authed: ApiAuthed, + tokened: Tokened, + Extension(db): Extension, + Path((w_id, flow_job_id, var_name)): Path<(String, Uuid, String)>, + Query(JsonPath { json_path, .. }): Query, +) -> windmill_common::error::JsonResult> { + let flow_env = sqlx::query_scalar!( + r#" + SELECT + CASE + WHEN flow_version.id IS NOT NULL THEN + (flow_version.value -> 'flow_env' -> $3) #> $4 + ELSE + (root_job.raw_flow -> 'flow_env' -> $3) #> $4 + END AS "flow_env: sqlx::types::Json>" + FROM + v2_job current_job + JOIN + v2_job root_job ON root_job.id = COALESCE(current_job.root_job, current_job.flow_innermost_root_job, current_job.parent_job, current_job.id) + AND root_job.workspace_id = current_job.workspace_id + LEFT JOIN + flow_version ON flow_version.id = root_job.runnable_id + AND flow_version.path = root_job.runnable_path + AND flow_version.workspace_id = root_job.workspace_id + WHERE + current_job.id = $1 AND + current_job.workspace_id = $2"#, + flow_job_id, + w_id, + var_name, + json_path + .as_ref() + .map(|x| x.split(".").collect::>()) + .unwrap_or_default() as Vec<&str>, + ) + .fetch_optional(&db) + .await? + .map(|r| r.map(|x| x.0)) + .flatten() + .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null)); + + log_job_view( + &db, + Some(&authed), + Some(&tokened.token), + &w_id, + &flow_job_id, + ) + .await?; + Ok(Json(flow_env)) +} + async fn compute_root_job_for_flow(db: &DB, w_id: &str, job_id: Uuid) -> error::Result { let root_job = sqlx::query_scalar!( r#"SELECT COALESCE(root_job, flow_innermost_root_job, parent_job, id) as "root_job!" FROM v2_job WHERE id = $1 AND workspace_id = $2"#, @@ -3769,6 +3826,7 @@ async fn batch_rerun_handle_job( batch_rerun_compute_js_expression(expr.clone(), job.clone()).await?, ); } + InputTransform::Ai => {} } } @@ -4226,7 +4284,8 @@ pub async fn run_script_by_path_inner( tag, timeout, None, - None, + // If the job has a parent job, set priority to 2 as it may be ran synchronously and block a current worker until being executed. Flow steps have a priority of 1 so this is higher. + if run_query.parent_job.is_some() || run_query.root_job.is_some() { Some(2) } else { None }, push_authed.as_ref(), false, None, diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index d491d46792..a639c58f6e 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -607,8 +607,10 @@ async fn create_script_internal<'c>( Ok(None) } (Some(p_hash), o) => { + // Lock the parent row to prevent concurrent updates with the same parent_hash + // This ensures linear lineage - only one script can have a given parent at a time if sqlx::query_scalar!( - "SELECT 1 FROM script WHERE hash = $1 AND workspace_id = $2", + "SELECT 1 FROM script WHERE hash = $1 AND workspace_id = $2 FOR UPDATE", p_hash.0, &w_id ) diff --git a/backend/windmill-common/src/client.rs b/backend/windmill-common/src/client.rs index b05fdfa2f7..146cc98e68 100644 --- a/backend/windmill-common/src/client.rs +++ b/backend/windmill-common/src/client.rs @@ -49,14 +49,7 @@ impl AuthedClient { "{}/api/w/{}/oidc/token/{}", self.base_internal_url, self.workspace, audience ); - let response = self.get(&url, vec![]).await?; - match response.status().as_u16() { - 200u16 => Ok(response - .json::() - .await - .context("decoding oidc token as json string")?), - _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default())), - } + make_basic_get_request(self, &url, None, Some("decoding oidc token as json string")).await } pub async fn get_resource_value(&self, path: &str) -> anyhow::Result { @@ -64,14 +57,7 @@ impl AuthedClient { "{}/api/w/{}/resources/get_value/{}", self.base_internal_url, self.workspace, path ); - let response = self.get(&url, vec![]).await?; - match response.status().as_u16() { - 200u16 => Ok(response - .json::() - .await - .context("decoding resource value as json")?), - _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default())), - } + make_basic_get_request(self, &url, None, Some("decoding resource value as json")).await } pub async fn get_variable_value(&self, path: &str) -> anyhow::Result { @@ -79,14 +65,7 @@ impl AuthedClient { "{}/api/w/{}/variables/get_value/{}", self.base_internal_url, self.workspace, path ); - let response = self.get(&url, vec![]).await?; - match response.status().as_u16() { - 200u16 => Ok(response - .json::() - .await - .context("decoding variable value as json")?), - _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default())), - } + make_basic_get_request(self, &url, None, Some("decoding variable value as json")).await } pub async fn get_resource_value_interpolated( @@ -121,19 +100,34 @@ impl AuthedClient { "{}/api/w/{}/jobs_u/completed/get_result/{}", self.base_internal_url, self.workspace, path ); - let query = if let Some(json_path) = json_path { - vec![("json_path", json_path)] - } else { - vec![] - }; - let response = self.get(&url, query).await?; - match response.status().as_u16() { - 200u16 => Ok(response - .json::() - .await - .context("decoding completed job result as json")?), - _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default())), - } + let query = query_from_json_path(json_path); + make_basic_get_request( + self, + &url, + Some(query), + Some("decoding completed job result as json"), + ) + .await + } + + pub async fn get_flow_env_by_flow_job_id( + &self, + root_job_id: &str, + var_name: &str, + json_path: Option, + ) -> anyhow::Result { + let url = format!( + "{}/api/w/{}/jobs/flow_env_by_flow_job_id/{}/{}", + self.base_internal_url, self.workspace, root_job_id, var_name + ); + let query = query_from_json_path(json_path); + make_basic_get_request( + self, + &url, + Some(query), + Some("decoding flow env variable as json"), + ) + .await } pub async fn get_result_by_id( @@ -146,19 +140,14 @@ impl AuthedClient { "{}/api/w/{}/jobs/result_by_id/{}/{}", self.base_internal_url, self.workspace, flow_job_id, node_id ); - let query = if let Some(json_path) = json_path { - vec![("json_path", json_path)] - } else { - vec![] - }; - let response = self.get(&url, query).await?; - match response.status().as_u16() { - 200u16 => Ok(response - .json::() - .await - .context("decoding result by id as json")?), - _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default())), - } + let query = query_from_json_path(json_path); + make_basic_get_request( + self, + &url, + Some(query), + Some("decoding result by id as json"), + ) + .await } pub async fn upload_s3_file( @@ -245,3 +234,33 @@ impl AuthedClient { } } } + +#[inline] +fn query_from_json_path(json_path: Option) -> Vec<(&'static str, String)> { + json_path + .map(|json_path| vec![("json_path", json_path)]) + .unwrap_or_else(|| Vec::new()) +} + +#[inline] +async fn make_basic_get_request( + client: &AuthedClient, + url: &str, + query: Option>, + context: Option<&'static str>, +) -> anyhow::Result { + let response = client + .get(&url, query.unwrap_or_else(|| Vec::new())) + .await?; + + match response.status().as_u16() { + 200u16 => { + let json_body = response + .json::() + .await + .context(context.unwrap_or("error decoding body as json"))?; + Ok(json_body) + } + _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default())), + } +} diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index fcfac158d3..929736eff5 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -192,6 +192,8 @@ pub struct FlowValue { pub priority: Option, #[serde(skip_serializing_if = "Option::is_none")] pub chat_input_enabled: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub flow_env: Option>> } impl FlowValue { @@ -643,6 +645,7 @@ pub enum InputTransform { #[serde(default = "default_empty_string")] expr: String, }, + Ai, } impl InputTransform { @@ -661,6 +664,7 @@ impl TryFrom for InputTransform { let input_transform = match value.type_.as_str() { "static" => InputTransform::new_static_value(value.value.unwrap_or_else(default_null)), "javascript" => InputTransform::new_javascript_expr(&value.expr.unwrap_or_default()), + "ai" => InputTransform::Ai, other => { return Err(anyhow::anyhow!( "got value: {other} for field `type`, expected value: `static` or `javascript`" @@ -816,8 +820,7 @@ pub struct McpToolValue { pub exclude_tools: Vec, } -fn is_none_or_empty_vec(expr: &Option>) -> bool -{ +fn is_none_or_empty_vec(expr: &Option>) -> bool { expr.is_none() || expr.as_ref().unwrap().is_empty() } diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index a4be34aa4b..b0a3f5e4ea 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -53,6 +53,7 @@ pub mod job_metrics; pub mod job_s3_helpers_ee; #[cfg(feature = "parquet")] pub mod job_s3_helpers_oss; +pub mod lockfiles; #[cfg(feature = "private")] pub mod git_sync_ee; diff --git a/backend/windmill-common/src/lockfiles.rs b/backend/windmill-common/src/lockfiles.rs new file mode 100644 index 0000000000..481caa70c3 --- /dev/null +++ b/backend/windmill-common/src/lockfiles.rs @@ -0,0 +1,17 @@ +use crate::scripts::ScriptLang; + +pub const LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT: &str = "# from requirements.txt"; + +pub fn is_generated_from_raw_requirements( + lang: &Option, + lock: &Option, +) -> bool { + (lang.is_some_and(|v| v == ScriptLang::Bun) + && lock + .as_ref() + .is_some_and(|v| v.contains("generatedFromPackageJson"))) + || (lang.is_some_and(|v| v == ScriptLang::Python3) + && lock + .as_ref() + .is_some_and(|v| v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT))) +} diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 34ba5a1694..11d6758caa 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -732,18 +732,32 @@ pub fn hash_script(ns: &NewScript) -> i64 { dh.finish() as i64 } +pub struct ClonedScript { + pub old_script: NewScript, + pub new_hash: i64, +} pub async fn clone_script<'c>( base_hash: ScriptHash, w_id: &str, deployment_message: Option, tx: &mut sqlx::Transaction<'c, sqlx::Postgres>, -) -> crate::error::Result { - let s = - sqlx::query_as::<_, Script>("SELECT * FROM script WHERE hash = $1 AND workspace_id = $2") - .bind(base_hash.0) - .bind(w_id) - .fetch_one(&mut **tx) - .await?; +) -> crate::error::Result { + let s = sqlx::query_as::<_, Script>( + "SELECT * FROM script WHERE hash = $1 AND workspace_id = $2 AND archived = false FOR UPDATE", + ) + .bind(base_hash.0) + .bind(w_id) + .fetch_optional(&mut **tx) + .await?; + + let s = if let Some(s) = s { + s + } else { + return Err(crate::error::Error::NotFound(format!( + "Non-archived script with hash {} not found", + base_hash.0 + ))); + }; let ns = NewScript { path: s.path.clone(), @@ -819,5 +833,5 @@ pub async fn clone_script<'c>( .execute(&mut **tx) .await?; - Ok(new_hash) + Ok(ClonedScript { old_script: ns, new_hash }) } diff --git a/backend/windmill-common/src/variables.rs b/backend/windmill-common/src/variables.rs index 9dd9edc64d..1edd292cbe 100644 --- a/backend/windmill-common/src/variables.rs +++ b/backend/windmill-common/src/variables.rs @@ -401,8 +401,7 @@ pub async fn get_reserved_variables( value, description: "Custom workspace environment variable".to_string(), is_custom: true, -}) -).collect() +})).collect() } async fn get_cached_workspace_envs(conn: &Connection, w_id: &str) -> Vec<(String, String)> { @@ -439,7 +438,11 @@ async fn get_cached_workspace_envs(conn: &Connection, w_id: &str) -> Vec<(String custom_envs } -pub async fn get_variable_or_self(path: String, db: &DB, w_id: &str) -> crate::error::Result { +pub async fn get_variable_or_self( + path: String, + db: &DB, + w_id: &str, +) -> crate::error::Result { if !path.starts_with("$var:") { return Ok(path); } diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index eafd8cd894..29d5488be8 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -498,7 +498,6 @@ pub async fn store_pull_query(wc: &WorkerConfig) { pub const TMP_DIR: &str = "/tmp/windmill"; pub const TMP_LOGS_DIR: &str = concatcp!(TMP_DIR, "/logs"); -pub const TMP_MEMORY_DIR: &str = concatcp!(TMP_DIR, "/memory"); pub const HUB_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "hub"); diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index d46a87ce0c..a6370163d6 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -35,6 +35,7 @@ use windmill_common::add_time; use windmill_common::auth::JobPerms; #[cfg(feature = "benchmark")] use windmill_common::bench::BenchmarkIter; +use windmill_common::lockfiles::is_generated_from_raw_requirements; use windmill_common::jobs::{JobTriggerKind, EMAIL_ERROR_HANDLER_USER_EMAIL}; use windmill_common::utils::{configure_client, now_from_db}; use windmill_common::worker::{Connection, MIN_VERSION_SUPPORTS_DEBOUNCING, SCRIPT_TOKEN_EXPIRY}; @@ -4328,6 +4329,7 @@ pub async fn push<'c, 'd>( skip_expr: None, preprocessor_module: None, chat_input_enabled: None, + flow_env: None, }; // this is a new flow being pushed, flow_status is set to flow_value: let flow_status: FlowStatus = FlowStatus::new(&flow_value); @@ -5660,15 +5662,20 @@ pub async fn preprocess_dependency_job(job: &mut PulledJob, db: &DB) -> error::R args.insert("base_hash".to_owned(), to_raw_value(&*base_hash)) }); - let new_hash = windmill_common::scripts::clone_script( + let cloned_script = windmill_common::scripts::clone_script( base_hash, &job.workspace_id, deployment_message, &mut tx, ) .await?; - - new_hash + if is_generated_from_raw_requirements(&Some(cloned_script.old_script.language), &cloned_script.old_script.lock.map(|v| v.to_string())) { + return Err(Error::BadRequest(format!( + "Script at path {} is generated from raw requirements, not overriding", + job.runnable_path() + ))); + } + cloned_script.new_hash } JobKind::FlowDependencies => { sqlx::query_scalar!( diff --git a/backend/windmill-worker/loader.bun.js b/backend/windmill-worker/loader.bun.js index 289fac0a9a..a697ba1d53 100644 --- a/backend/windmill-worker/loader.bun.js +++ b/backend/windmill-worker/loader.bun.js @@ -32,7 +32,12 @@ const p = { const imports = transpiler.scanImports(code); for (const imp of imports) { if (imp.kind == "import-statement") { - if (imp.path.startsWith(".") && !imp.path.endsWith(".ts")) { + if ( + (imp.path.startsWith(".") || + imp.path.startsWith("/u/") || + imp.path.startsWith("/f/")) && + !imp.path.endsWith(".ts") + ) { code = code.replaceAll(imp.path, imp.path + ".ts"); } } diff --git a/backend/windmill-worker/src/ai/tools.rs b/backend/windmill-worker/src/ai/tools.rs index d1280f0053..243a67fb4e 100644 --- a/backend/windmill-worker/src/ai/tools.rs +++ b/backend/windmill-worker/src/ai/tools.rs @@ -322,16 +322,23 @@ async fn execute_windmill_tool( // Evaluate each input transform and merge with AI-provided args for (key, transform) in input_transforms.iter() { // We skip static empty / null values, those are the one the AI will fill in - if let InputTransform::Static { value } = transform { - let val = value.get().trim(); - if val.is_empty() || val == "null" { + match transform { + InputTransform::Static { value } => { + let val = value.get().trim(); + if val.is_empty() || val == "null" { + continue; + } + } + InputTransform::Ai => { continue; } + _ => (), } let result = evaluate_input_transform::>( transform, last_result.clone(), flow_inputs.clone(), + None, Some(ctx.client), ctx.id_context.as_ref(), ) diff --git a/backend/windmill-worker/src/ai/utils.rs b/backend/windmill-worker/src/ai/utils.rs index fb27c419ba..947b56f549 100644 --- a/backend/windmill-worker/src/ai/utils.rs +++ b/backend/windmill-worker/src/ai/utils.rs @@ -78,6 +78,7 @@ pub fn filter_schema_by_input_transforms( !val.is_empty() && val != "null" } InputTransform::Javascript { expr } => !expr.trim().is_empty(), + InputTransform::Ai => false, }; if is_completed { Some(key.clone()) diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index f9207e23a5..481717a207 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -405,7 +405,7 @@ pub async fn run_agent( .and_then(|fs| fs.memory_id) { // Read messages from memory - match read_from_memory(&job.workspace_id, memory_id, step_id).await { + match read_from_memory(db, &job.workspace_id, memory_id, step_id).await { Ok(Some(loaded_messages)) => { // Take the last n messages let start_idx = loaded_messages.len().saturating_sub(context_length); @@ -856,6 +856,7 @@ pub async fn run_agent( if let Some(memory_id) = flow_context.flow_status.and_then(|fs| fs.memory_id) { if let Err(e) = write_to_memory( + db, &job.workspace_id, memory_id, step_id, diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 892a3c8b7e..259b12ec84 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1068,7 +1068,6 @@ pub fn build_http_client(timeout_duration: std::time::Duration) -> error::Result } pub fn get_root_job_id(job: &MiniPulledJob) -> uuid::Uuid { - // fallback to flow_innermost_root_job and parent_job as root_job is not set if equal to innermost root job or parent job job.root_job .or(job.flow_innermost_root_job) .or(job.parent_job) diff --git a/backend/windmill-worker/src/js_eval.rs b/backend/windmill-worker/src/js_eval.rs index c87b506686..ec28b6f509 100644 --- a/backend/windmill-worker/src/js_eval.rs +++ b/backend/windmill-worker/src/js_eval.rs @@ -165,10 +165,98 @@ impl NetPermissions for PermissionsContainer { #[cfg(feature = "deno_core")] pub struct OptAuthedClient(Option); +const FLOW_INPUT_PREFIX: &'static str = "flow_input"; +const ENV_KEY_PREFIX: &'static str = "flow_env"; +const DOT_PATTERN: &'static str = "."; +const START_BRACKET_PATTERN: &'static str = "[\""; +const END_BRACKET_PATTERN: &'static str = "\"]"; + +fn try_exact_property_access( + expr: &str, + flow_input: Option<&mappable_rc::Marc>>>, + flow_env: Option<&HashMap>>, +) -> Option> { + let obj = if expr.starts_with(FLOW_INPUT_PREFIX) { + Some(( + FLOW_INPUT_PREFIX, + flow_input.as_ref().map(|obj| obj.as_ref()), + )) + } else if expr.starts_with(ENV_KEY_PREFIX) { + Some((ENV_KEY_PREFIX, flow_env)) + } else { + None + }; + + if let Some((prefix, obj)) = obj { + let access_pattern_pos = prefix.len(); + let suffix = &expr[access_pattern_pos..]; + let maybe_key_name = if suffix.starts_with(DOT_PATTERN) { + let key_name_pos = DOT_PATTERN.len(); + Some(&expr[key_name_pos..]) + } else if suffix.starts_with(START_BRACKET_PATTERN) { + let key_name_pos = START_BRACKET_PATTERN.len(); + let suffix = &suffix[key_name_pos..]; + + let flow_arg_name = suffix + .ends_with(END_BRACKET_PATTERN) + .then(|| { + let start_key_name_pos = access_pattern_pos + key_name_pos; + let end_key_name_pos = expr.len() - END_BRACKET_PATTERN.len(); + &expr[start_key_name_pos..end_key_name_pos] + }) + .filter(|s| s.len() > 0); + flow_arg_name + } else { + None + }; + + if let Some(key_name) = maybe_key_name { + if let Some(key_value) = obj.and_then(|obj| obj.get(key_name)) { + return Some(key_value.clone()); + } + } + } + None +} + +async fn handle_full_regex( + captures: regex::Captures<'_>, + authed_client: &AuthedClient, + by_id: &IdContext, +) -> anyhow::Result> { + let obj_name = captures.get(1).unwrap().as_str(); + let obj_key = captures.get(2).unwrap().as_str(); + let idx_o = captures.get(3).map(|y| y.as_str()); + let rest = captures.get(4).map(|y| y.as_str()); + let query = if let Some(idx) = idx_o { + match rest { + Some(rest) => Some(format!("{}{}", idx, rest)), + None => Some(idx.to_string()), + } + } else { + rest.map(|x| x.trim_start_matches('.').to_string()) + }; + + let result = if obj_name == "results" { + authed_client + .get_result_by_id(&by_id.flow_job.to_string(), obj_key, query) + .await + } else if obj_name == "flow_env" { + authed_client + .get_flow_env_by_flow_job_id(&by_id.flow_job.to_string(), obj_key, query) + .await + } else { + unreachable!(); + }; + + return result; +} + pub async fn eval_timeout( expr: String, transform_context: HashMap>>, flow_input: Option>>>, + flow_env: Option<&HashMap>>, authed_client: Option<&AuthedClient>, by_id: Option<&IdContext>, #[allow(unused_variables)] ctx: Option>, @@ -180,21 +268,13 @@ pub async fn eval_timeout( expr, transform_context ); - for (k, v) in transform_context.iter() { - if k == &expr { - return Ok(v.as_ref().clone()); - } + + if let Some(value) = transform_context.get(&expr) { + return Ok(value.as_ref().to_owned()); } - if expr.starts_with("flow_input.") || expr.starts_with("flow_input[") { - if let Some(ref flow_input) = flow_input { - for (k, v) in flow_input.iter() { - if &format!("flow_input.{k}") == &expr || &format!("flow_input[\"{k}\"]") == &expr { - // tracing::error!("FLOW_INPUT"); - return Ok(v.clone()); - } - } - } + if let Some(value) = try_exact_property_access(&expr, flow_input.as_ref(), flow_env) { + return Ok(value); } let p_ids = by_id.map(|x| { @@ -219,24 +299,8 @@ pub async fn eval_timeout( } if let (Some(by_id), Some(authed_client)) = (by_id, authed_client) { - if let Some((id, idx_o, rest)) = RE_FULL.captures(&expr).map(|x| { - ( - x.get(1).unwrap().as_str(), - x.get(2).map(|y| y.as_str()), - x.get(3).map(|y| y.as_str()), - ) - }) { - let query = if let Some(idx) = idx_o { - match rest { - Some(rest) => Some(format!("{}{}", idx, rest)), - None => Some(idx.to_string()), - } - } else { - rest.map(|x| x.trim_start_matches('.').to_string()) - }; - return authed_client - .get_result_by_id(&by_id.flow_job.to_string(), id, query) - .await; + if let Some(captures) = RE_FULL.captures(&expr) { + return handle_full_regex(captures, authed_client, by_id).await; } } @@ -271,8 +335,8 @@ pub async fn eval_timeout( if by_id.is_some() && authed_client.is_some() { ops.push(op_get_result()); ops.push(op_get_id()); + ops.push(op_get_flow_env()); } - let ext = Extension { name: "js_eval", ops: ops.into(), ..Default::default() }; let exts = vec![ext]; // Use our snapshot to provision our new runtime @@ -307,13 +371,15 @@ pub async fn eval_timeout( let mut client = authed_client.clone(); if let Some(client) = client.as_mut() { client.force_client = Some( - configure_client(reqwest::ClientBuilder::new() - .user_agent("windmill/beta") - .danger_accept_invalid_certs( - std::env::var("ACCEPT_INVALID_CERTS").is_ok(), - )) - .build() - .unwrap(), + configure_client( + reqwest::ClientBuilder::new() + .user_agent("windmill/beta") + .danger_accept_invalid_certs( + std::env::var("ACCEPT_INVALID_CERTS").is_ok(), + ), + ) + .build() + .unwrap(), ); } op_state.put(OptAuthedClient(client)); @@ -323,7 +389,7 @@ pub async fn eval_timeout( .into_iter() .filter(|(a, _)| context_keys.contains(a)) .collect(), - }) + }); } sender @@ -379,10 +445,12 @@ fn replace_with_await(expr: String, fn_name: &str) -> String { } lazy_static! { static ref RE: Regex = - Regex::new(r#"(?m)(?Presults(?:\?)?(?:(?:\.[a-zA-Z_0-9]+)|(?:\[\".*?\"\])))"#).unwrap(); - static ref RE_FULL: Regex = - Regex::new(r"(?m)^results(?:\?)?\.([a-zA-Z_0-9]+)(?:\[(\d+)\])?((?:\.[a-zA-Z_0-9]+)+)?$") + Regex::new(r#"(?m)(?P(?:results|flow_env)(?:\?)?(?:(?:\.[a-zA-Z_0-9]+)|(?:\[\".*?\"\])))"#) .unwrap(); + static ref RE_FULL: Regex = Regex::new( + r"(?m)^(results|flow_env)(?:\?)?\.([a-zA-Z_0-9]+)(?:\[(\d+)\])?((?:\.[a-zA-Z_0-9]+)+)?$" + ) + .unwrap(); static ref RE_PROXY: Regex = Regex::new(r"^(https?)://(([^:@\s]+):([^:@\s]+)@)?([^:@\s]+)(:(\d+))?$").unwrap(); } @@ -453,6 +521,17 @@ const results = new Proxy({{}}, {{ }} }}); +async function flow_env_by_var_name(var_name) {{ + let root_job_id = "{}"; + return JSON.parse(await Deno.core.ops.op_get_flow_env(root_job_id, var_name, null)); +}} + +const flow_env = new Proxy({{}}, {{ + get: function(target, name, receiver) {{ + return flow_env_by_var_name(name); + }} +}}); + "#, by_id .steps_results @@ -469,6 +548,7 @@ const results = new Proxy({{}}, {{ .join(","), by_id.previous_id, by_id.flow_job, + by_id.flow_job ) } else { String::new() @@ -641,6 +721,33 @@ async fn op_resource( } } +#[cfg(feature = "deno_core")] +#[op2(async)] +#[string] +async fn op_get_flow_env( + op_state: Rc>, + #[string] root_job_id: String, + #[string] var_name: String, + #[string] json_path: Option, +) -> Result, deno_error::JsErrorBox> { + let client = op_state.borrow().borrow::().0.clone(); + if let Some(client) = client { + client + .get_flow_env_by_flow_job_id::>>( + &root_job_id, + &var_name, + json_path, + ) + .await + .map(|value| value.map(|val| val.get().to_string())) + .map_err(|e| deno_error::JsErrorBox::generic(e.to_string())) + } else { + Err(deno_error::JsErrorBox::generic( + "No client found in op state", + )) + } +} + #[cfg(feature = "deno_core")] pub struct TransformContext { pub envs: HashMap>>, @@ -1318,7 +1425,7 @@ multiline template`"; op_state.put(TransformContext { flow_input: None, envs: env.clone() }) } - let res = eval_timeout(code.to_string(), env, None, None, None, None).await?; + let res = eval_timeout(code.to_string(), env, None, None, None, None, None).await?; assert_eq!(res.get(), "2"); Ok(()) } diff --git a/backend/windmill-worker/src/memory_common.rs b/backend/windmill-worker/src/memory_common.rs index 16821c555b..440199702f 100644 --- a/backend/windmill-worker/src/memory_common.rs +++ b/backend/windmill-worker/src/memory_common.rs @@ -1,72 +1,123 @@ use crate::ai::types::OpenAIMessage; -use std::path::PathBuf; -use tokio::{fs, io::AsyncWriteExt}; use uuid::Uuid; -use windmill_common::worker::TMP_MEMORY_DIR; +use windmill_common::{db::DB, error::Error}; -/// Get the file path for storing memory for a specific AI agent step -pub fn path_for(workspace_id: &str, conversation_id: Uuid, step_id: &str) -> PathBuf { - PathBuf::from(TMP_MEMORY_DIR) - .join(workspace_id) - .join(conversation_id.to_string()) - .join(format!("{step_id}.json")) -} +pub const MAX_MEMORY_SIZE_BYTES: usize = 100_000; // 100KB per memory entry in database -/// Read messages from disk storage -pub async fn read_from_disk( +/// Read AI agent memory from database +pub async fn read_from_db( + db: &DB, workspace_id: &str, conversation_id: Uuid, step_id: &str, -) -> anyhow::Result>> { - let path = path_for(workspace_id, conversation_id, step_id); - if !fs::try_exists(&path).await? { - return Ok(None); - } +) -> Result>, Error> { + let result = sqlx::query!( + "SELECT messages FROM ai_agent_memory + WHERE workspace_id = $1 AND conversation_id = $2 AND step_id = $3", + workspace_id, + conversation_id, + step_id + ) + .fetch_optional(db) + .await?; - let bytes = fs::read(&path).await?; - let messages: Vec = serde_json::from_slice(&bytes)?; - Ok(Some(messages)) + match result { + Some(row) => { + let messages: Vec = serde_json::from_value(row.messages)?; + Ok(Some(messages)) + } + None => Ok(None), + } } -/// Write messages to disk storage -pub async fn write_to_disk( +/// Write AI agent memory to database with size checking and truncation +pub async fn write_to_db( + db: &DB, workspace_id: &str, conversation_id: Uuid, step_id: &str, messages: &[OpenAIMessage], -) -> anyhow::Result<()> { - let path = path_for(workspace_id, conversation_id, step_id); - - // Ensure parent directories exist - if let Some(dir) = path.parent() { - fs::create_dir_all(dir).await?; +) -> Result<(), Error> { + if messages.is_empty() { + return Ok(()); } - // Write atomically using a temporary file - let tmp = path.with_extension("json.tmp"); - let mut f = fs::File::create(&tmp).await?; - f.write_all(&serde_json::to_vec(messages)?).await?; - f.flush().await?; - drop(f); + // Serialize messages and check size + let mut messages_to_store = messages.to_vec(); + let mut json_value = serde_json::to_value(&messages_to_store)?; + let size_bytes = json_value.to_string().len(); - // Atomic rename - fs::rename(tmp, &path).await?; + // Truncate if necessary + if size_bytes > MAX_MEMORY_SIZE_BYTES { + tracing::warn!( + "Memory size ({} bytes) exceeds limit ({} bytes) for workspace={} conversation={} step={}. Truncating messages. Use S3 storage in workspace settings to store full conversation history.", + size_bytes, + MAX_MEMORY_SIZE_BYTES, + workspace_id, + conversation_id, + step_id + ); + + messages_to_store = truncate_messages(messages, MAX_MEMORY_SIZE_BYTES)?; + json_value = serde_json::to_value(&messages_to_store)?; + } + + // Insert or update using UPSERT + sqlx::query!( + "INSERT INTO ai_agent_memory (workspace_id, conversation_id, step_id, messages, created_at, updated_at) + VALUES ($1, $2, $3, $4, NOW(), NOW()) + ON CONFLICT (workspace_id, conversation_id, step_id) + DO UPDATE SET + messages = EXCLUDED.messages, + updated_at = NOW()", + workspace_id, + conversation_id, + step_id, + json_value + ) + .execute(db) + .await?; Ok(()) } -/// Delete all memory for a conversation from disk storage -pub async fn delete_conversation_from_disk( +/// Delete all memory for a conversation from database +pub async fn delete_conversation_from_db( + db: &DB, workspace_id: &str, conversation_id: Uuid, -) -> anyhow::Result<()> { - let conversation_path = PathBuf::from(TMP_MEMORY_DIR) - .join(workspace_id) - .join(conversation_id.to_string()); - - if fs::try_exists(&conversation_path).await? { - fs::remove_dir_all(&conversation_path).await?; - } +) -> Result<(), Error> { + sqlx::query!( + "DELETE FROM ai_agent_memory + WHERE workspace_id = $1 AND conversation_id = $2", + workspace_id, + conversation_id + ) + .execute(db) + .await?; Ok(()) } + +/// Truncate messages to fit within the size limit +fn truncate_messages( + messages: &[OpenAIMessage], + max_size_bytes: usize, +) -> Result, Error> { + let mut result = messages.to_vec(); + + // Keep removing oldest messages until we're under the threshold + while !result.is_empty() { + let test_json = serde_json::to_value(&result)?; + let test_size = test_json.to_string().len(); + + if test_size <= max_size_bytes { + break; + } + + // Remove the first (oldest) message + result.remove(0); + } + + Ok(result) +} diff --git a/backend/windmill-worker/src/memory_oss.rs b/backend/windmill-worker/src/memory_oss.rs index 2444cd0ebf..74771a79e8 100644 --- a/backend/windmill-worker/src/memory_oss.rs +++ b/backend/windmill-worker/src/memory_oss.rs @@ -3,23 +3,27 @@ pub use crate::memory_ee::*; #[cfg(not(all(feature = "private", feature = "enterprise")))] -use {crate::ai::types::OpenAIMessage, crate::memory_common, uuid::Uuid}; +use {crate::ai::types::OpenAIMessage, crate::memory_common, uuid::Uuid, windmill_common::db::DB}; /// Read AI agent memory from storage -/// In OSS: always reads from disk +/// In OSS: always reads from database #[cfg(not(all(feature = "private", feature = "enterprise")))] pub async fn read_from_memory( + db: &DB, workspace_id: &str, conversation_id: Uuid, step_id: &str, ) -> anyhow::Result>> { - memory_common::read_from_disk(workspace_id, conversation_id, step_id).await + memory_common::read_from_db(db, workspace_id, conversation_id, step_id) + .await + .map_err(|e| anyhow::anyhow!("Database read failed: {e:?}")) } /// Write AI agent memory to storage -/// In OSS: always writes to disk +/// In OSS: always writes to database #[cfg(not(all(feature = "private", feature = "enterprise")))] pub async fn write_to_memory( + db: &DB, workspace_id: &str, conversation_id: Uuid, step_id: &str, @@ -29,15 +33,20 @@ pub async fn write_to_memory( return Ok(()); } - memory_common::write_to_disk(workspace_id, conversation_id, step_id, messages).await + memory_common::write_to_db(db, workspace_id, conversation_id, step_id, messages) + .await + .map_err(|e| anyhow::anyhow!("Database write failed: {e:?}")) } /// Delete all memory for a conversation from storage -/// In OSS: always deletes from disk +/// In OSS: always deletes from database #[cfg(not(all(feature = "private", feature = "enterprise")))] pub async fn delete_conversation_memory( + db: &DB, workspace_id: &str, conversation_id: Uuid, ) -> anyhow::Result<()> { - memory_common::delete_conversation_from_disk(workspace_id, conversation_id).await + memory_common::delete_conversation_from_db(db, workspace_id, conversation_id) + .await + .map_err(|e| anyhow::anyhow!("Database delete failed: {e:?}")) } diff --git a/backend/windmill-worker/src/python_versions.rs b/backend/windmill-worker/src/python_versions.rs index 2c5b828313..b4b39a0c46 100644 --- a/backend/windmill-worker/src/python_versions.rs +++ b/backend/windmill-worker/src/python_versions.rs @@ -12,6 +12,7 @@ use tokio::{fs::DirBuilder, process::Command, sync::RwLock}; use uuid::Uuid; use windmill_common::{ error::{self, Error}, + lockfiles::LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT, worker::Connection, }; @@ -22,7 +23,6 @@ use crate::{ common::{start_child_process, OccupancyMetrics}, handle_child::handle_child, python_executor::{PYTHON_PATH, UV_PATH}, - worker_lockfiles::LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT, HOME_ENV, INSTANCE_PYTHON_VERSION, PATH_ENV, PROXY_ENVS, PY_INSTALL_DIR, WIN_ENVS, }; diff --git a/backend/windmill-worker/src/scoped_dependency_map.rs b/backend/windmill-worker/src/scoped_dependency_map.rs index 09436bed3b..072fedc8a6 100644 --- a/backend/windmill-worker/src/scoped_dependency_map.rs +++ b/backend/windmill-worker/src/scoped_dependency_map.rs @@ -5,15 +5,12 @@ use windmill_common::{ cache, error::{Error, Result}, flows::{FlowModuleValue, FlowValue}, - scripts::ScriptLang, }; use std::collections::HashSet; -use crate::worker_lockfiles::{ - extract_relative_imports, is_generated_from_raw_requirements, - LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT, -}; +use crate::worker_lockfiles::extract_relative_imports; +use windmill_common::lockfiles::is_generated_from_raw_requirements; // TODO: To be removed in future versions lazy_static::lazy_static! { @@ -295,16 +292,7 @@ SELECT importer_node_id, imported_path let mut dmap = ScopedDependencyMap::fetch(w_id, &r.path, "script", db).await?; let mut tx = db.begin().await?; - if (smd.language.is_some_and(|v| v == ScriptLang::Bun) - && sd - .lock - .as_ref() - .is_some_and(|v| v.contains("generatedFromPackageJson"))) - || (smd.language.is_some_and(|v| v == ScriptLang::Python3) - && sd.lock.as_ref().is_some_and(|v| { - v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT) - })) - { + if is_generated_from_raw_requirements(&smd.language, &sd.lock) { // if the lock file is generated from a package.json/requirements.txt, we need to clear the dependency map // because we do not want to have dependencies be recomputed automatically. Empty relative imports passed // to update_script_dependency_map will clear the dependency map. @@ -350,7 +338,7 @@ SELECT importer_node_id, imported_path match fmv { // Since we fetched from flow_version it is safe to assume all inline scripts are in form of RawScript. FlowModuleValue::RawScript { content, language, lock ,.. } => { - if !is_generated_from_raw_requirements(Some(*language), lock) { + if !is_generated_from_raw_requirements(&Some(*language), lock) { to_process.push(( extract_relative_imports( content, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index db568652e5..c457dc5343 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -877,7 +877,7 @@ pub fn start_interactive_worker_shell( token, precomputed_agent_info: precomputed_bundle, } = extract_job_and_perms(job, &conn).await; - + let authed_client = AuthedClient::new( base_internal_url.to_owned(), job.workspace_id.clone(), @@ -886,7 +886,7 @@ pub fn start_interactive_worker_shell( ); let arc_job = Arc::new(job); - + let _ = handle_queued_job( arc_job.clone(), raw_code, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index dd34f4dc24..f9f8709f5c 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -245,6 +245,7 @@ async fn evaluate_stop_after_all_iters_if( let stop_early_after_all_iters = compute_bool_from_expr( &stop_after_all_iters_if.expr, Marc::new(args), + None, iters_result.clone(), None, None, @@ -427,17 +428,20 @@ pub async fn update_flow_status_after_job_completion_internal( ) .fetch_one(db) .await; + let args = + args.map(|flow_args| flow_args.map(|flow_args| flow_args.0).unwrap_or_default()); + args })); - let from_result_to_args = - |args: &Result>>>, sqlx::Error>| { - let args = args.as_ref().map_err(|e| { - Error::internal_err(format!("retrieval of args from state: {e:#}")) - })?; - Ok::<_, Error>(args.clone().unwrap_or_default().0) - }; + let from_result_to_args = |args: &Result>, sqlx::Error>| { + let args = args + .as_ref() + .map_err(|e| Error::internal_err(format!("retrieval of args from state: {e:#}")))?; + + Ok::<_, Error>(args.clone()) + }; let (mut stop_early, mut stop_early_err_msg, mut skip_if_stop_early, continue_on_error) = if stop_early_override.is_some() @@ -469,9 +473,11 @@ pub async fn update_flow_status_after_job_completion_internal( _ => None, }; let args = from_result_to_args(args.as_ref().await.get_ref())?; + compute_bool_from_expr( &expr, Marc::new(args), + None, result.clone(), all_iters, None, @@ -736,6 +742,7 @@ pub async fn update_flow_status_after_job_completion_internal( &mut stop_early_err_msg, &mut nresult, args, + ) .await?; } @@ -924,6 +931,7 @@ pub async fn update_flow_status_after_job_completion_internal( .and_then(|x| x.stop_after_all_iters_if.as_ref()) { let args = from_result_to_args(args.as_ref().await.get_ref())?; + evaluate_stop_after_all_iters_if( db, stop_after_all_iters_if, @@ -1016,6 +1024,7 @@ pub async fn update_flow_status_after_job_completion_internal( &old_status.retry, result.clone(), Marc::new(args), + None, Some(client), ) .await? @@ -1334,6 +1343,7 @@ pub async fn update_flow_status_after_job_completion_internal( &old_status.retry, result.clone(), Marc::new(args), + None, Some(client), ) .await? @@ -1824,6 +1834,7 @@ async fn evaluate_retry( status: &RetryStatus, result: Arc>, flow_args: Marc>>, + flow_env: Option<&HashMap>>, client: Option<&AuthedClient>, ) -> anyhow::Result> { if status.fail_count > MAX_RETRY_ATTEMPTS { @@ -1834,6 +1845,7 @@ async fn evaluate_retry( let should_retry = compute_bool_from_expr( &retry_if.expr, flow_args, + flow_env, result, None, None, @@ -1857,6 +1869,7 @@ async fn evaluate_retry( async fn compute_bool_from_expr( expr: &str, flow_args: Marc>>, + flow_env: Option<&HashMap>>, result: Arc>, all_iters: Option>>, by_id: Option<&IdContext>, @@ -1881,6 +1894,7 @@ async fn compute_bool_from_expr( format!("Boolean({expr})"), context, Some(flow_args), + flow_env, client, by_id, ctx, @@ -1906,15 +1920,13 @@ pub async fn evaluate_input_transform( transform: &InputTransform, last_result: Arc>, flow_args: Option>>>, + flow_env: Option<&HashMap>>, authed_client: Option<&AuthedClient>, by_id: Option<&IdContext>, ) -> error::Result where - T: for<'de> serde::Deserialize<'de> + Send, + T: for<'de> serde::Deserialize<'de> + Send + Default, { - let mut context = HashMap::with_capacity(2); - context.insert("result".to_string(), last_result.clone()); - context.insert("previous_result".to_string(), last_result.clone()); match transform { InputTransform::Static { value } => serde_json::from_str(value.get()).map_err(|e| { Error::ExecutionErr(format!( @@ -1923,10 +1935,14 @@ where )) }), InputTransform::Javascript { expr } => { + let mut context = HashMap::with_capacity(2); + context.insert("result".to_string(), last_result.clone()); + context.insert("previous_result".to_string(), last_result.clone()); let result = eval_timeout( expr.to_string(), context, flow_args, + flow_env, authed_client, by_id, None, @@ -1947,6 +1963,7 @@ where )) }) } + InputTransform::Ai => Ok(T::default()), } } @@ -1954,6 +1971,7 @@ where #[instrument(level = "trace", skip_all)] async fn transform_input( flow_args: Marc>>, + flow_env: Option<&HashMap>>, last_result: Arc>, input_transforms: &HashMap, resumes: Arc>, @@ -1995,6 +2013,7 @@ async fn transform_input( expr.to_string(), env.clone(), Some(flow_args.clone()), + flow_env, Some(client), Some(by_id), None, @@ -2007,6 +2026,7 @@ async fn transform_input( })?; mapped.insert(key.to_string(), v); } + InputTransform::Ai => (), } } @@ -2067,6 +2087,7 @@ pub async fn handle_flow( ); } } + let mut rec = PushNextFlowJobRec { flow_job: flow_job, status: status }; loop { let PushNextFlowJobRec { flow_job, status } = rec; @@ -2202,13 +2223,14 @@ async fn push_next_flow_job( // tracing::error!("status_module: {status_module:#?}"); let fj: mappable_rc::Marc = flow_job.clone().into(); - let arc_flow_job_args: Marc>> = Marc::map(fj, |x| { - if let Some(args) = &x.args { - &args.0 - } else { - &EHM - } - }); + let arc_flow_job_args: Marc>> = + Marc::map(fj, |x: &MiniPulledJob| { + if let Some(args) = &x.args { + &args.0 + } else { + &EHM + } + }); // if this is an empty module without preprocessor of if the module has already been completed, successfully, update the parent flow if (flow.modules.is_empty() && !step.is_preprocessor_step()) @@ -2298,6 +2320,7 @@ async fn push_next_flow_job( let skip = compute_bool_from_expr( &skip_expr, arc_flow_job_args.clone(), + flow.flow_env.as_ref(), Arc::new(to_raw_value(&json!("{}"))), None, None, @@ -2419,6 +2442,7 @@ async fn push_next_flow_job( expr.to_string(), context, Some(arc_flow_job_args.clone()), + flow.flow_env.as_ref(), None, None, None @@ -2441,6 +2465,9 @@ async fn push_next_flow_job( ))); } } + InputTransform::Ai => { + user_groups_required = Vec::new(); + } } } else { user_groups_required = Vec::new(); @@ -2676,6 +2703,7 @@ async fn push_next_flow_job( &input_transform, arc_last_job_result.clone(), Some(arc_flow_job_args.clone()), + flow.flow_env.as_ref(), Some(client), None, ) @@ -2713,6 +2741,7 @@ async fn push_next_flow_job( &status.retry, arc_last_job_result.clone(), arc_flow_job_args.clone(), + flow.flow_env.as_ref(), Some(client), ) .await? @@ -2802,6 +2831,7 @@ async fn push_next_flow_job( compute_bool_from_expr( &skip_if.expr, arc_flow_job_args.clone(), + flow.flow_env.as_ref(), arc_last_job_result.clone(), None, Some(&idcontext), @@ -2893,6 +2923,7 @@ async fn push_next_flow_job( }; transform_input( arc_flow_job_args.clone(), + flow.flow_env.as_ref(), arc_last_job_result.clone(), input_transforms, resumes.clone(), @@ -2919,6 +2950,7 @@ async fn push_next_flow_job( let next_flow_transform = compute_next_flow_transform( arc_flow_job_args.clone(), arc_last_job_result.clone(), + flow.flow_env.as_ref(), &flow_job, &flow, transform_context, @@ -3067,6 +3099,7 @@ async fn push_next_flow_job( .await?; let ti = transform_input( Marc::new(args), + flow.flow_env.as_ref(), arc_last_job_result.clone(), input_transforms, resumes.clone(), @@ -3117,6 +3150,7 @@ async fn push_next_flow_job( .await?; let ti = transform_input( Marc::new(hm), + flow.flow_env.as_ref(), arc_last_job_result.clone(), input_transforms, resumes.clone(), @@ -3220,6 +3254,7 @@ async fn push_next_flow_job( timeout_transform, arc_last_job_result.clone(), Some(arc_flow_job_args.clone()), + flow.flow_env.as_ref(), Some(client), Some(&ctx), ) @@ -3299,6 +3334,7 @@ async fn push_next_flow_job( parallelism_transform, arc_last_job_result.clone(), Some(arc_flow_job_args.clone()), + flow.flow_env.as_ref(), Some(client), Some(&ctx), ) @@ -3756,6 +3792,7 @@ pub fn get_path(flow_job: &MiniPulledJob, status: &FlowStatus, module: &FlowModu async fn compute_next_flow_transform( arc_flow_job_args: Marc>>, arc_last_job_result: Arc>, + flow_env: Option<&HashMap>>, flow_job: &MiniPulledJob, flow: &FlowValue, by_id: Option, @@ -3974,6 +4011,7 @@ async fn compute_next_flow_transform( resume, approvers, arc_flow_job_args, + flow_env, client, ¶llel, ) @@ -4069,6 +4107,7 @@ async fn compute_next_flow_transform( let pred = compute_bool_from_expr( &b.expr, arc_flow_job_args.clone(), + flow.flow_env.as_ref(), arc_last_job_result.clone(), None, Some(&idcontext), @@ -4333,6 +4372,7 @@ async fn next_forloop_status( resume: Arc>, approvers: Arc>, arc_flow_job_args: Marc>>, + flow_env: Option<&HashMap>>, client: &AuthedClient, parallel: &bool, ) -> Result { @@ -4348,6 +4388,11 @@ async fn next_forloop_status( /* Iterator is an InputTransform, evaluate it into an array. */ let itered_raw = match iterator { InputTransform::Static { value } => to_raw_value(value), + InputTransform::Ai => { + return Err(Error::ExecutionErr(format!( + "AI input transform not supported for iterator" + )))? + } InputTransform::Javascript { expr } => { let mut context = HashMap::with_capacity(5); context.insert("result".to_string(), arc_last_job_result.clone()); @@ -4360,6 +4405,7 @@ async fn next_forloop_status( expr.to_string(), context, Some(arc_flow_job_args), + flow_env, Some(client), Some(&by_id), None, @@ -4423,12 +4469,18 @@ async fn next_forloop_status( expr.to_string(), context, Some(arc_flow_job_args), + flow_env, Some(client), Some(&by_id), None, ) .await? } + InputTransform::Ai => { + return Err(Error::ExecutionErr(format!( + "AI input transform not supported for iterator" + )))? + } }; serde_json::from_str::>>(itered_raw.get()).map_err( |not_array| { diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 7b1acc19e8..44b68948a7 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -20,6 +20,7 @@ use windmill_common::error::Error; use windmill_common::error::Result; use windmill_common::flows::{FlowModule, FlowModuleValue, FlowNodeId}; use windmill_common::jobs::JobPayload; +use windmill_common::lockfiles::is_generated_from_raw_requirements; use windmill_common::scripts::ScriptHash; use windmill_common::utils::WarnAfterExt; #[cfg(feature = "python")] @@ -373,15 +374,7 @@ pub async fn process_relative_imports( db, ) .await?; - if (script_lang.is_some_and(|v| v == ScriptLang::Bun) - && lock - .as_ref() - .is_some_and(|v| v.contains("generatedFromPackageJson"))) - || (script_lang.is_some_and(|v| v == ScriptLang::Python3) - && lock - .as_ref() - .is_some_and(|v| v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT))) - { + if is_generated_from_raw_requirements(script_lang, &lock) { // if the lock file is generated from a package.json/requirements.txt, we need to clear the dependency map // because we do not want to have dependencies be recomputed automatically. Empty relative imports passed // to update_script_dependency_map will clear the dependency map. @@ -445,17 +438,6 @@ pub async fn process_relative_imports( Ok(()) } -pub fn is_generated_from_raw_requirements(lang: Option, lock: &Option) -> bool { - (lang.is_some_and(|v| v == ScriptLang::Bun) - && lock - .as_ref() - .is_some_and(|v| v.contains("generatedFromPackageJson"))) - || (lang.is_some_and(|v| v == ScriptLang::Python3) - && lock - .as_ref() - .is_some_and(|v| v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT))) -} - pub async fn trigger_dependents_to_recompute_dependencies( w_id: &str, script_path: &str, @@ -1450,7 +1432,7 @@ async fn lock_modules<'c>( if let Some(locks_to_reload) = locks_to_reload { if !locks_to_reload.contains(&e.id) { - if !is_generated_from_raw_requirements(Some(language), &lock) { + if !is_generated_from_raw_requirements(&Some(language), &lock) { let relative_imports = get_imports(); tx = dependency_map .patch(relative_imports.clone(), e.id.clone(), tx) @@ -1463,7 +1445,7 @@ async fn lock_modules<'c>( if lock.as_ref().is_some_and(|x| !x.trim().is_empty()) { let skip_creating_new_lock = skip_creating_new_lock(&language, &content); if skip_creating_new_lock { - if !is_generated_from_raw_requirements(Some(language), &lock) { + if !is_generated_from_raw_requirements(&Some(language), &lock) { let relative_imports = get_imports(); tx = dependency_map .patch(relative_imports.clone(), e.id.clone(), tx) @@ -2582,8 +2564,6 @@ async fn ansible_dep( serde_json::to_string(&ansible_lockfile).map_err(|e| e.into()) } -pub const LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT: &str = "# from requirements.txt"; - async fn capture_dependency_job( job_id: &Uuid, job_language: &ScriptLang, @@ -2672,7 +2652,11 @@ async fn capture_dependency_job( .await .map(|res| { if raw_deps { - format!("{}\n{}", LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT, res) + format!( + "{}\n{}", + windmill_common::lockfiles::LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT, + res + ) } else { res } diff --git a/benchmarks/lib.ts b/benchmarks/lib.ts index d8ac299e57..5dfc44b9ae 100644 --- a/benchmarks/lib.ts +++ b/benchmarks/lib.ts @@ -2,7 +2,7 @@ import { sleep } from "https://deno.land/x/sleep@v1.2.1/mod.ts"; import * as windmill from "https://deno.land/x/windmill@v1.174.0/mod.ts"; import * as api from "https://deno.land/x/windmill@v1.174.0/windmill-api/index.ts"; -export const VERSION = "v1.573.3"; +export const VERSION = "v1.573.5"; export async function login(email: string, password: string): Promise { return await windmill.UserService.login({ diff --git a/cli/src/main.ts b/cli/src/main.ts index c0ed2b651b..413cd2a59a 100644 --- a/cli/src/main.ts +++ b/cli/src/main.ts @@ -68,7 +68,7 @@ export { // } // }); -export const VERSION = "1.573.3"; +export const VERSION = "1.573.5"; export const WM_FORK_PREFIX = "wm-fork"; diff --git a/docker-compose.yml b/docker-compose.yml index 1770a1e4db..ea67377ffc 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -73,8 +73,6 @@ services: - /var/run/docker.sock:/var/run/docker.sock - worker_dependency_cache:/tmp/windmill/cache - worker_logs:/tmp/windmill/logs - # for AI agent memory - - worker_memory:/tmp/windmill/memory logging: *default-logging diff --git a/frontend/package-lock.json b/frontend/package-lock.json index 56d162adee..48eea0020c 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -1,12 +1,12 @@ { "name": "windmill-components", - "version": "1.573.3", + "version": "1.573.5", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "windmill-components", - "version": "1.573.3", + "version": "1.573.5", "hasInstallScript": true, "license": "AGPL-3.0", "dependencies": { diff --git a/frontend/package.json b/frontend/package.json index e3efa90d53..47b0b78bc8 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "windmill-components", - "version": "1.573.3", + "version": "1.573.5", "scripts": { "dev": "vite dev", "build": "vite build", diff --git a/frontend/src/lib/common.ts b/frontend/src/lib/common.ts index e5220a2655..5f30b6056e 100644 --- a/frontend/src/lib/common.ts +++ b/frontend/src/lib/common.ts @@ -50,6 +50,7 @@ export interface SchemaProperty { oneOf?: SchemaProperty[] originalType?: string disabled?: boolean + 'x-no-s3-storage-workspace-warning'?: string } export interface ModalSchemaProperty { diff --git a/frontend/src/lib/components/ArgInput.svelte b/frontend/src/lib/components/ArgInput.svelte index 1c631b0d9b..3aa15b98f8 100644 --- a/frontend/src/lib/components/ArgInput.svelte +++ b/frontend/src/lib/components/ArgInput.svelte @@ -14,7 +14,7 @@ import { DollarSign, Plus, X, Check, Loader2, ExternalLink } from 'lucide-svelte' import { createEventDispatcher, onDestroy, onMount, tick, untrack } from 'svelte' import { fade } from 'svelte/transition' - import { Button, SecondsInput } from './common' + import { Alert, Button, SecondsInput } from './common' import FieldHeader from './FieldHeader.svelte' import type ItemPicker from './ItemPicker.svelte' import ObjectResourceInput from './ObjectResourceInput.svelte' @@ -126,6 +126,7 @@ | undefined) | undefined workspace?: string | undefined + s3StorageConfigured?: boolean actions?: import('svelte').Snippet innerBottomSnippet?: import('svelte').Snippet fieldHeaderActions?: import('svelte').Snippet @@ -185,6 +186,7 @@ appPath = undefined, computeS3ForceViewerPolicies = undefined, workspace = undefined, + s3StorageConfigured = true, actions, innerBottomSnippet, fieldHeaderActions, @@ -1455,6 +1457,15 @@ {:else if !noMargin}
{/if} + + {#if !s3StorageConfigured && extra['x-no-s3-storage-workspace-warning']} + + {/if}