Merge branch 'main' into fg/revert

This reverts commit b26cea9d3e.

CHANGELOG.md

@@ -1,5 +1,115 @@
 # Changelog
 
+## [1.537.0](https://github.com/windmill-labs/windmill/compare/v1.536.0...v1.537.0) (2025-08-28)
+
+
+### Features
+
+* **aichat:** allow reverting specific line for inline script suggestions ([#6480](https://github.com/windmill-labs/windmill/issues/6480)) ([0cc11b3](https://github.com/windmill-labs/windmill/commit/0cc11b3f31aeee60a9d4a231cea5d4285d7ab37e))
+* autovacuum or high intensity tables ([4ad0d25](https://github.com/windmill-labs/windmill/commit/4ad0d255f3eea303e97eab5325f89930b26f9e52))
+
+
+### Bug Fixes
+
+* fix okta and oauth0 sso settings ([73a3f4c](https://github.com/windmill-labs/windmill/commit/73a3f4cc73271759650e9246f4eb2e0efb7c7e37))
+* fix relative imports cache invalidation ([#6468](https://github.com/windmill-labs/windmill/issues/6468)) ([006f326](https://github.com/windmill-labs/windmill/commit/006f32602c7609b282f15135989c5f164c109c1c))
+* fix workflow as code behavior with multithread ([4973c86](https://github.com/windmill-labs/windmill/commit/4973c860f2c28d9bdc2af94530d90eb177234e5d))
+
+## [1.536.0](https://github.com/windmill-labs/windmill/compare/v1.535.0...v1.536.0) (2025-08-27)
+
+
+### Features
+
+* **aichat:** give advanced options tools to flow mode ([#6463](https://github.com/windmill-labs/windmill/issues/6463)) ([b26cea9](https://github.com/windmill-labs/windmill/commit/b26cea9d3e2f9a0acae335aad12206da491ac733))
+* email triggers extra args in 'to' header ([#6476](https://github.com/windmill-labs/windmill/issues/6476)) ([ceb9150](https://github.com/windmill-labs/windmill/commit/ceb9150f43a0ae9f8579f1984e791f69e7a05366))
+
+
+### Bug Fixes
+
+* **cli:** specific items for file resource type ([#6464](https://github.com/windmill-labs/windmill/issues/6464)) ([2066a2a](https://github.com/windmill-labs/windmill/commit/2066a2ada2f3139474527f373dc505b7e61d5182))
+* do not require locked for scheduled jobs ([41a8727](https://github.com/windmill-labs/windmill/commit/41a872725282ba4b78e8f9912bb1ac929b8557f7))
+* **frontend:** ai agent step nits ([#6469](https://github.com/windmill-labs/windmill/issues/6469)) ([2b03133](https://github.com/windmill-labs/windmill/commit/2b03133b2245bd42f3c64b915d72dd3f62eb65a4))
+* **frontend:** nats config conditional fields ([#6473](https://github.com/windmill-labs/windmill/issues/6473)) ([f90d444](https://github.com/windmill-labs/windmill/commit/f90d44469e0e4b462a5fa5160b64e99eef95c317))
+* **go:** could not read Username for 'xyz': terminal prompts disabled ([#6478](https://github.com/windmill-labs/windmill/issues/6478)) ([5808840](https://github.com/windmill-labs/windmill/commit/5808840b78e94a0b39614f37161f9def347a5352))
+* **go:** exec: "git": executable file not found ([#6475](https://github.com/windmill-labs/windmill/issues/6475)) ([475f405](https://github.com/windmill-labs/windmill/commit/475f405d0626f1c22309ee6a1b630472a89dbb30))
+* save changes made in diff mode ([#6477](https://github.com/windmill-labs/windmill/issues/6477)) ([d9ca181](https://github.com/windmill-labs/windmill/commit/d9ca181b1d8d26c175ec2a05409c45daeab887a4))
+
+## [1.535.0](https://github.com/windmill-labs/windmill/compare/v1.534.1...v1.535.0) (2025-08-25)
+
+
+### Features
+
+* **aichat:** show diff mode on inline scripts changes ([#6454](https://github.com/windmill-labs/windmill/issues/6454)) ([eca3109](https://github.com/windmill-labs/windmill/commit/eca3109ec63967e3041521bf74d34a12c70f5ff8))
+
+
+### Bug Fixes
+
+* fix opening advanced popup for run resetting tag to default ([d328894](https://github.com/windmill-labs/windmill/commit/d3288947b2d2539b2f3302059a9aad2841275a28))
+
+## [1.534.1](https://github.com/windmill-labs/windmill/compare/v1.534.0...v1.534.1) (2025-08-25)
+
+
+### Bug Fixes
+
+* add alias to subquery for older postgres versions ([#6455](https://github.com/windmill-labs/windmill/issues/6455)) ([16d233b](https://github.com/windmill-labs/windmill/commit/16d233bf466fd818ec1f9235377e4c1a8239d98c))
+* **frontend:** fix test step behavior ([#6427](https://github.com/windmill-labs/windmill/issues/6427)) ([fc20b7b](https://github.com/windmill-labs/windmill/commit/fc20b7bd91d33115aacb38cc46394f9c6465aa0f))
+
+## [1.534.0](https://github.com/windmill-labs/windmill/compare/v1.533.1...v1.534.0) (2025-08-25)
+
+
+### Features
+
+* **backend:** support unencrypted  connection to mssql ([#6453](https://github.com/windmill-labs/windmill/issues/6453)) ([8d31c2a](https://github.com/windmill-labs/windmill/commit/8d31c2ab0d34036dc8057611857a5d72aad8598f))
+
+
+### Bug Fixes
+
+* **aichat:** fix wrong current model logic ([#6451](https://github.com/windmill-labs/windmill/issues/6451)) ([e951c89](https://github.com/windmill-labs/windmill/commit/e951c896b865df48d331968953c9e44848236516))
+* **flow:** test this step preload step input evaluation ([1073eb0](https://github.com/windmill-labs/windmill/commit/1073eb0e682e7bd253c6d62225361b487d7f6d2f))
+
+## [1.533.1](https://github.com/windmill-labs/windmill/compare/v1.533.0...v1.533.1) (2025-08-23)
+
+
+### Bug Fixes
+
+* **app:** fix oneOf selected undefined freeze ([0ae8f44](https://github.com/windmill-labs/windmill/commit/0ae8f44773adb0576e1b63e858b506ba9a9fe7b3))
+
+## [1.533.0](https://github.com/windmill-labs/windmill/compare/v1.532.0...v1.533.0) (2025-08-23)
+
+
+### Features
+
+* CLI improvements ([#6446](https://github.com/windmill-labs/windmill/issues/6446)) ([a41b9e4](https://github.com/windmill-labs/windmill/commit/a41b9e47e233ebaa2baafb5cca1187bb85d6f8f4))
+
+
+### Bug Fixes
+
+* **frontend:** ai agent flow status + UI nits ([#6447](https://github.com/windmill-labs/windmill/issues/6447)) ([c13747c](https://github.com/windmill-labs/windmill/commit/c13747cda9449369288e8d078b60542ea79a49bf))
+
+## [1.532.0](https://github.com/windmill-labs/windmill/compare/v1.531.0...v1.532.0) (2025-08-22)
+
+
+### Features
+
+* **aichat:** allow adding contexts to flow mode ([#6424](https://github.com/windmill-labs/windmill/issues/6424)) ([73272f1](https://github.com/windmill-labs/windmill/commit/73272f16fddc355703b04f2c3458520753d1e19c))
+* json schema resource ([#6433](https://github.com/windmill-labs/windmill/issues/6433)) ([7da79a8](https://github.com/windmill-labs/windmill/commit/7da79a8bc525fc6b89748ad0af25c2bac4ca2ef3))
+
+## [1.531.0](https://github.com/windmill-labs/windmill/compare/v1.530.0...v1.531.0) (2025-08-22)
+
+
+### Features
+
+* ai agent steps ([#6393](https://github.com/windmill-labs/windmill/issues/6393)) ([958e8af](https://github.com/windmill-labs/windmill/commit/958e8af78290cf859f98c45c012ed41e3bada39e))
+* bump Go version from 1.22.0 to 1.25.0 [#6415](https://github.com/windmill-labs/windmill/issues/6415)  ([c92bfe6](https://github.com/windmill-labs/windmill/commit/c92bfe6601fd96f6d74860f52f9307e02961ac21))
+
+
+### Bug Fixes
+
+* **app:** fix ctrl drag for insertion into subgrids ([51ea947](https://github.com/windmill-labs/windmill/commit/51ea9473ef23c6871699e69bbe79772a4d50d3b8))
+* **frontend:** graph cache of ai agent step tools ([#6431](https://github.com/windmill-labs/windmill/issues/6431)) ([28f1d61](https://github.com/windmill-labs/windmill/commit/28f1d611643459d42531fa217c185408eb97d6d1))
+* make relevant sidebar menu items a instead of button ([06d078e](https://github.com/windmill-labs/windmill/commit/06d078ebfa8f70b66bc764eae70d33c8c57b4012))
+* s3 result presigned not working with list ([9df008b](https://github.com/windmill-labs/windmill/commit/9df008b9f8fe58692463e4b9da0538935e458b10))
+
 ## [1.530.0](https://github.com/windmill-labs/windmill/compare/v1.529.0...v1.530.0) (2025-08-20)
 
 

CLAUDE.md

@@ -4,6 +4,17 @@
 
 Windmill is an open-source developer platform for building internal tools, workflows, API integrations, background jobs, workflows, and user interfaces. See @windmill-overview.mdc for full platform details.
 
+## New Feature Implementation Guidelines
+
+When implementing new features in Windmill, follow these best practices:
+
+- **Clean Code First**: Write clean, readable, and maintainable code. Prioritize clarity over cleverness.
+- **Avoid Duplication at All Costs**: Before writing new code, thoroughly search for existing implementations that can be reused or extended.
+- **Adapt Existing Code**: Refactor and generalize existing code when necessary to avoid logic duplication. Extract common patterns into reusable utilities.
+- **Follow Established Patterns**: Study existing code patterns in the codebase and maintain consistency with established conventions.
+- **Single Responsibility**: Each function, component, and module should have a single, well-defined responsibility.
+- **Incremental Implementation**: Break large features into smaller, reviewable chunks that can be implemented and tested incrementally.
+
 ## Language-Specific Guides
 
 - Backend (Rust): @backend/rust-best-practices.mdc + @backend/summarized_schema.txt

README.md

@@ -332,40 +332,40 @@ you to have it being synced automatically everyday.
 
 ## Environment Variables
 
-| Environment Variable name           | Default                | Description                                                                                                                                                                                        | Api Server/Worker/All |
-| ----------------------------------- | ---------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------- |
-| DATABASE_URL                        |                        | The Postgres database url.                                                                                                                                                                         | All                   |
-| WORKER_GROUP                        | default                | The worker group the worker belongs to and get its configuration pulled from                                                                                                                       | Worker                |
-| MODE                                | standalone             | The mode if the binary. Possible values: standalone, worker, server, agent                                                                                                                         | All                   |
-| METRICS_ADDR                        | None                   | (ee only) The socket addr at which to expose Prometheus metrics at the /metrics path. Set to "true" to expose it on port 8001                                                                      | All                   |
-| JSON_FMT                            | false                  | Output the logs in json format instead of logfmt                                                                                                                                                   | All                   |
-| BASE_URL                            | http://localhost:8000  | The base url that is exposed publicly to access your instance. Is overriden by the instance settings if any.                                                                                       | Server                |
-| ZOMBIE_JOB_TIMEOUT                  | 30                     | The timeout after which a job is considered to be zombie if the worker did not send pings about processing the job (every server check for zombie jobs every 30s)                                  | Server                |
-| RESTART_ZOMBIE_JOBS                 | true                   | If true then a zombie job is restarted (in-place with the same uuid and some logs), if false the zombie job is failed                                                                              | Server                |
-| SLEEP_QUEUE                         | 50                     | The number of ms to sleep in between the last check for new jobs in the DB. It is multiplied by NUM_WORKERS such that in average, for one worker instance, there is one pull every SLEEP_QUEUE ms. | Worker                |
-| KEEP_JOB_DIR                        | false                  | Keep the job directory after the job is done. Useful for debugging.                                                                                                                                | Worker                |
-| LICENSE_KEY (EE only)               | None                   | License key checked at startup for the Enterprise Edition of Windmill                                                                                                                              | Worker                |
-| SLACK_SIGNING_SECRET                | None                   | The signing secret of your Slack app. See [Slack documentation](https://api.slack.com/authentication/verifying-requests-from-slack)                                                                | Server                |
-| COOKIE_DOMAIN                       | None                   | The domain of the cookie. If not set, the cookie will be set by the browser based on the full origin                                                                                               | Server                |
-| DENO_PATH                           | /usr/bin/deno          | The path to the deno binary.                                                                                                                                                                       | Worker                |
-| PYTHON_PATH                         |                        | The path to the python binary if wanting to not have it managed by uv.                                                                                                                             | Worker                |
-| GO_PATH                             | /usr/bin/go            | The path to the go binary.                                                                                                                                                                         | Worker                |
-| GOPRIVATE                           |                        | The GOPRIVATE env variable to use private go modules                                                                                                                                               | Worker                |
-| GOPROXY                             |                        | The GOPROXY env variable to use                                                                                                                                                                    | Worker                |
-| NETRC                               |                        | The netrc content to use a private go registry                                                                                                                                                     | Worker                |
-| PY_CONCURRENT_DOWNLOADS             | 20                     | Sets the maximum number of in-flight concurrent python downloads that windmill will perform at any given time.                                                                                     | Worker                |
-| PATH                                | None                   | The path environment variable, usually inherited                                                                                                                                                   | Worker                |
-| HOME                                | None                   | The home directory to use for Go and Bash , usually inherited                                                                                                                                      | Worker                |
-| DATABASE_CONNECTIONS                | 50 (Server)/3 (Worker) | The max number of connections in the database connection pool                                                                                                                                      | All                   |
-| SUPERADMIN_SECRET                   | None                   | A token that would let the caller act as a virtual superadmin superadmin@windmill.dev                                                                                                              | Server                |
-| TIMEOUT_WAIT_RESULT                 | 20                     | The number of seconds to wait before timeout on the 'run_wait_result' endpoint                                                                                                                     | Worker                |
-| QUEUE_LIMIT_WAIT_RESULT             | None                   | The number of max jobs in the queue before rejecting immediately the request in 'run_wait_result' endpoint. Takes precedence on the query arg. If none is specified, there are no limit.           | Worker                |
-| DENO_AUTH_TOKENS                    | None                   | Custom DENO_AUTH_TOKENS to pass to worker to allow the use of private modules                                                                                                                      | Worker                |
-| DISABLE_RESPONSE_LOGS               | false                  | Disable response logs                                                                                                                                                                              | Server                |
-| CREATE_WORKSPACE_REQUIRE_SUPERADMIN | true                   | If true, only superadmins can create new workspaces                                                                                                                                                | Server                |
-| MIN_FREE_DISK_SPACE_MB              | 15000                  | Minimum amount of free space on worker. Sends critical alert if worker has less free space.                                                                                                        | Worker                |
-| RUN_UPDATE_CA_CERTIFICATE_AT_START  | false                  | If true, runs CA certificate update command at startup before other initialization                                                                                                                  | All                   |
-| RUN_UPDATE_CA_CERTIFICATE_PATH      | /usr/sbin/update-ca-certificates | Path to the CA certificate update command/script to run when RUN_UPDATE_CA_CERTIFICATE_AT_START is true                                                                                             | All                   |
+| Environment Variable name           | Default                          | Description                                                                                                                                                                                        | Api Server/Worker/All |
+| ----------------------------------- | -------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------- |
+| DATABASE_URL                        |                                  | The Postgres database url.                                                                                                                                                                         | All                   |
+| WORKER_GROUP                        | default                          | The worker group the worker belongs to and get its configuration pulled from                                                                                                                       | Worker                |
+| MODE                                | standalone                       | The mode if the binary. Possible values: standalone, worker, server, agent                                                                                                                         | All                   |
+| METRICS_ADDR                        | None                             | (ee only) The socket addr at which to expose Prometheus metrics at the /metrics path. Set to "true" to expose it on port 8001                                                                      | All                   |
+| JSON_FMT                            | false                            | Output the logs in json format instead of logfmt                                                                                                                                                   | All                   |
+| BASE_URL                            | http://localhost:8000            | The base url that is exposed publicly to access your instance. Is overriden by the instance settings if any.                                                                                       | Server                |
+| ZOMBIE_JOB_TIMEOUT                  | 30                               | The timeout after which a job is considered to be zombie if the worker did not send pings about processing the job (every server check for zombie jobs every 30s)                                  | Server                |
+| RESTART_ZOMBIE_JOBS                 | true                             | If true then a zombie job is restarted (in-place with the same uuid and some logs), if false the zombie job is failed                                                                              | Server                |
+| SLEEP_QUEUE                         | 50                               | The number of ms to sleep in between the last check for new jobs in the DB. It is multiplied by NUM_WORKERS such that in average, for one worker instance, there is one pull every SLEEP_QUEUE ms. | Worker                |
+| KEEP_JOB_DIR                        | false                            | Keep the job directory after the job is done. Useful for debugging.                                                                                                                                | Worker                |
+| LICENSE_KEY (EE only)               | None                             | License key checked at startup for the Enterprise Edition of Windmill                                                                                                                              | Worker                |
+| SLACK_SIGNING_SECRET                | None                             | The signing secret of your Slack app. See [Slack documentation](https://api.slack.com/authentication/verifying-requests-from-slack)                                                                | Server                |
+| COOKIE_DOMAIN                       | None                             | The domain of the cookie. If not set, the cookie will be set by the browser based on the full origin                                                                                               | Server                |
+| DENO_PATH                           | /usr/bin/deno                    | The path to the deno binary.                                                                                                                                                                       | Worker                |
+| PYTHON_PATH                         |                                  | The path to the python binary if wanting to not have it managed by uv.                                                                                                                             | Worker                |
+| GO_PATH                             | /usr/bin/go                      | The path to the go binary.                                                                                                                                                                         | Worker                |
+| GOPRIVATE                           |                                  | The GOPRIVATE env variable to use private go modules                                                                                                                                               | Worker                |
+| GOPROXY                             |                                  | The GOPROXY env variable to use                                                                                                                                                                    | Worker                |
+| NETRC                               |                                  | The netrc content to use a private go registry                                                                                                                                                     | Worker                |
+| PY_CONCURRENT_DOWNLOADS             | 20                               | Sets the maximum number of in-flight concurrent python downloads that windmill will perform at any given time.                                                                                     | Worker                |
+| PATH                                | None                             | The path environment variable, usually inherited                                                                                                                                                   | Worker                |
+| HOME                                | None                             | The home directory to use for Go and Bash , usually inherited                                                                                                                                      | Worker                |
+| DATABASE_CONNECTIONS                | 50 (Server)/3 (Worker)           | The max number of connections in the database connection pool                                                                                                                                      | All                   |
+| SUPERADMIN_SECRET                   | None                             | A token that would let the caller act as a virtual superadmin superadmin@windmill.dev                                                                                                              | Server                |
+| TIMEOUT_WAIT_RESULT                 | 20                               | The number of seconds to wait before timeout on the 'run_wait_result' endpoint                                                                                                                     | Worker                |
+| QUEUE_LIMIT_WAIT_RESULT             | None                             | The number of max jobs in the queue before rejecting immediately the request in 'run_wait_result' endpoint. Takes precedence on the query arg. If none is specified, there are no limit.           | Worker                |
+| DENO_AUTH_TOKENS                    | None                             | Custom DENO_AUTH_TOKENS to pass to worker to allow the use of private modules                                                                                                                      | Worker                |
+| DISABLE_RESPONSE_LOGS               | false                            | Disable response logs                                                                                                                                                                              | Server                |
+| CREATE_WORKSPACE_REQUIRE_SUPERADMIN | true                             | If true, only superadmins can create new workspaces                                                                                                                                                | Server                |
+| MIN_FREE_DISK_SPACE_MB              | 15000                            | Minimum amount of free space on worker. Sends critical alert if worker has less free space.                                                                                                        | Worker                |
+| RUN_UPDATE_CA_CERTIFICATE_AT_START  | false                            | If true, runs CA certificate update command at startup before other initialization                                                                                                                 | All                   |
+| RUN_UPDATE_CA_CERTIFICATE_PATH      | /usr/sbin/update-ca-certificates | Path to the CA certificate update command/script to run when RUN_UPDATE_CA_CERTIFICATE_AT_START is true                                                                                            | All                   |
 
 ## Run a local dev setup
 
@@ -374,7 +374,6 @@ Using [Nix](./frontend/README_DEV.md#nix) (Recommended).
 See the [./frontend/README_DEV.md](./frontend/README_DEV.md) file for all
 running options.
 
-
 ### only Frontend
 
 This will use the backend of <https://app.windmill.dev> but your own frontend
@@ -400,29 +399,27 @@ npm run generate-backend-client-mac
 See the [./frontend/README_DEV.md](./frontend/README_DEV.md) file for all
 running options.
 
-1. Create a Postgres Database for Windmill and create an admin role inside your
-   Postgres setup. The easiest way to get a working db is to run
+1. Start a local Postgres database using for instance the `start-dev-db.sh` script which will make a database available at `postgres://postgres:changeme@localhost:5432/windmill`
+   Then run the migrations using the following command:
    ```
    cargo install sqlx-cli
    env DATABASE_URL=<YOUR_DATABASE_URL> sqlx migrate run
    ```
-   This will also avoid compile time issue with sqlx's `query!` macro
-2. Install [nsjail](https://github.com/google/nsjail) and have it accessible in
+   This will also avoid compile time issue with sqlx's `query!` macro.
+2. (optional, linux only) Install [nsjail](https://github.com/google/nsjail) and have it accessible in
    your PATH
-3. Install deno and python3, have the bins at `/usr/bin/deno` and
-   `/usr/local/bin/python3`
-4. Install [caddy](https://caddyserver.com)
-5. Install the [lld linker](https://lld.llvm.org/)
-6. Go to `frontend/`:
-   1. `npm install`, `npm run generate-backend-client` then `npm run dev`
+3. Install bun, deno and python3 (+ any languages you want to use), have the bins at `/usr/bin/bun`,`/usr/bin/deno`, and
+   `/usr/local/bin/python3` or set the corresponding environment variables.
+4. (optional) Install the [lld linker](https://lld.llvm.org/)
+5. Go to `frontend/`:
+   1. `npm install`, `npm run generate-backend-client` then `REMOTE=http://localhost:8000 npm run dev`
    2. You might need to set some extra heap space for the node runtime
       `export NODE_OPTIONS="--max-old-space-size=4096"`
-   3. In another shell `npm run build` otherwise the backend will not find the
-      `frontend/build` folder and will not compile.
-   4. In another shell `sudo caddy run --config Caddyfile`
-7. Go to `backend/`:
-   `env DATABASE_URL=<DATABASE_URL_TO_YOUR_WINDMILL_DB> RUST_LOG=info cargo run`
-8. Et voilà, windmill should be available at `http://localhost/`
+   3. Create an empty `frontend/build` folder using `mkdir frontend/build`
+6. Go to `backend/`:
+   1. `env DATABASE_URL=<YOUR_DATABASE_URL> RUST_LOG=info cargo run`
+   2. You can specify any feature flag you want to enable, for example `cargo run --features python` to enable the python executor.
+7. Et voilà, windmill should be available at `http://localhost:3000`
 
 ## Contributors
 

backend/.sqlx/query-124e67b0cee1baa6295846db4ad6242a39dd40186f1dbb48ad3018bd9f6913ec.json

@@ -0,0 +1,34 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            SELECT\n                w.premium,\n                COALESCE(cw.is_past_due, false) as \"is_past_due!\",\n                cw.max_tolerated_executions\n            FROM\n                workspace w\n                LEFT JOIN cloud_workspace_settings cw ON cw.workspace_id = w.id\n            WHERE\n                w.id = $1\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "premium",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 1,
+        "name": "is_past_due!",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 2,
+        "name": "max_tolerated_executions",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      null,
+      true
+    ]
+  },
+  "hash": "124e67b0cee1baa6295846db4ad6242a39dd40186f1dbb48ad3018bd9f6913ec"
+}

backend/.sqlx/query-2367e7c0f7fbafe0971a187c0909617da55251e97180babf6ac9e8068f26d73d.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "INSERT INTO deployment_metadata (workspace_id, path, flow_version, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4, $5)\n                        ON CONFLICT (workspace_id, path) DO UPDATE SET callback_job_ids = $4, deployment_msg = $5",
+  "query": "INSERT INTO deployment_metadata (workspace_id, path, flow_version, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4, $5)\n                        ON CONFLICT (workspace_id, path, flow_version) WHERE flow_version IS NOT NULL DO UPDATE SET callback_job_ids = $4, deployment_msg = $5",
   "describe": {
     "columns": [],
     "parameters": {
@@ -14,5 +14,5 @@
     },
     "nullable": []
   },
-  "hash": "434d8dfbc25cf7e92de51d763d3a2904ccc2e95ecc3d90b43a6394a7bb4d26ab"
+  "hash": "2367e7c0f7fbafe0971a187c0909617da55251e97180babf6ac9e8068f26d73d"
 }

backend/.sqlx/query-298f8609319a2928257fd5be60bb37f292c786d2348efe11d19868e5dc8fba11.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT elem FROM (SELECT unnest($1::TEXT[]) AS elem)\n        WHERE elem NOT IN (SELECT datname FROM pg_catalog.pg_database);",
+  "query": "SELECT elem FROM (SELECT unnest($1::TEXT[]) AS elem) AS e\n        WHERE elem NOT IN (SELECT datname FROM pg_catalog.pg_database);",
   "describe": {
     "columns": [
       {
@@ -18,5 +18,5 @@
       null
     ]
   },
-  "hash": "543859cf1c8d9e3bf2c2b23d21d096d01fe7a72d749229f21634b549c6b1241a"
+  "hash": "298f8609319a2928257fd5be60bb37f292c786d2348efe11d19868e5dc8fba11"
 }

backend/.sqlx/query-2d5f58dd2aff3bd49f3891ae76df23e2aa39891931516426f65b229314a0cee1.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "SELECT workspace_id as workspace, path, summary, description, schema FROM script as o WHERE created_at = (select max(created_at) from script where o.path = path and workspace_id = $1) and workspace_id = $1",
+  "query": "SELECT workspace_id as workspace, path, summary, description, schema FROM script as o \n         WHERE created_at = (select max(created_at) from script where o.path = path and workspace_id = $1 AND archived = false) \n         AND workspace_id = $1 and archived = false",
   "describe": {
     "columns": [
       {
@@ -42,5 +42,5 @@
       true
     ]
   },
-  "hash": "53e7243abd724816fb8d09c63b7ffa65f1cd622a989f5cefedbbf3c143b387c4"
+  "hash": "2d5f58dd2aff3bd49f3891ae76df23e2aa39891931516426f65b229314a0cee1"
 }

backend/.sqlx/query-4221d98d76f3cb32d6be581b0f63cf7578429009bee4f648e2c1bc3784fdbefc.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "VACUUM v2_job_queue, v2_job_runtime, v2_job_status, job_perms",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "4221d98d76f3cb32d6be581b0f63cf7578429009bee4f648e2c1bc3784fdbefc"
+}

backend/.sqlx/query-4b8132b04e454eddfe6724c6cc3a2e60c9c24decb2a6b41125247bbf741e9c25.json

@@ -1,46 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT owner, premium, usage.usage as \"usage?\", workspace_settings.customer_id, workspace_settings.plan FROM workspace LEFT JOIN workspace_settings ON workspace_settings.workspace_id = $1 LEFT JOIN usage ON usage.id = $1 AND month_ = EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date) AND usage.is_workspace IS true WHERE workspace.id = $1",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "owner",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 1,
-        "name": "premium",
-        "type_info": "Bool"
-      },
-      {
-        "ordinal": 2,
-        "name": "usage?",
-        "type_info": "Int4"
-      },
-      {
-        "ordinal": 3,
-        "name": "customer_id",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 4,
-        "name": "plan",
-        "type_info": "Varchar"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text"
-      ]
-    },
-    "nullable": [
-      false,
-      false,
-      false,
-      true,
-      true
-    ]
-  },
-  "hash": "4b8132b04e454eddfe6724c6cc3a2e60c9c24decb2a6b41125247bbf741e9c25"
-}

backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json

@@ -15,7 +15,7 @@
       ]
     },
     "nullable": [
-      null
+      true
     ]
   },
   "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"

backend/.sqlx/query-807c920bff25f56b10e88900d879cf5e8484c147e457044d6b075323b163ebaa.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "VACUUM v2_job, v2_job_completed, job_result_stream, job_stats, job_logs, concurrency_key, log_file, metrics",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "807c920bff25f56b10e88900d879cf5e8484c147e457044d6b075323b163ebaa"
+}

backend/.sqlx/query-8d119104337bf99e9aa9dcbac0a54154267a7db96cc0fb3ebaac95635e24da29.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "INSERT INTO deployment_metadata (workspace_id, path, script_hash, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4, $5) \n                        ON CONFLICT (workspace_id, path) DO UPDATE SET callback_job_ids = $4, deployment_msg = $5",
+  "query": "INSERT INTO deployment_metadata (workspace_id, path, script_hash, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4, $5) \n                        ON CONFLICT (workspace_id, script_hash) WHERE script_hash IS NOT NULL DO UPDATE SET callback_job_ids = $4, deployment_msg = $5",
   "describe": {
     "columns": [],
     "parameters": {
@@ -14,5 +14,5 @@
     },
     "nullable": []
   },
-  "hash": "e77fcf4e0d58855542605d13177df61671334418820ca942b442adfab413cbae"
+  "hash": "8d119104337bf99e9aa9dcbac0a54154267a7db96cc0fb3ebaac95635e24da29"
 }

backend/.sqlx/query-ac4a375035049304db7ae084b2d1789afb8250adde6f38f458b115e787ed876f.json

@@ -0,0 +1,58 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            SELECT \n                owner, \n                premium, \n                usage.usage as \"usage?\", \n                workspace_settings.customer_id, \n                workspace_settings.plan, \n                COALESCE(cw.is_past_due, false) as \"is_past_due!\", \n                cw.max_tolerated_executions\n            FROM workspace\n            LEFT JOIN workspace_settings \n                ON workspace_settings.workspace_id = $1\n            LEFT JOIN usage \n                ON usage.id = $1\n                AND month_ = EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date)\n                AND usage.is_workspace IS true\n            LEFT JOIN cloud_workspace_settings cw\n                ON cw.workspace_id = $1\n            WHERE workspace.id = $1\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "owner",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "premium",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 2,
+        "name": "usage?",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 3,
+        "name": "customer_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 4,
+        "name": "plan",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 5,
+        "name": "is_past_due!",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 6,
+        "name": "max_tolerated_executions",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      true,
+      true,
+      null,
+      true
+    ]
+  },
+  "hash": "ac4a375035049304db7ae084b2d1789afb8250adde6f38f458b115e787ed876f"
+}

backend/.sqlx/query-b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76.json

@@ -18,8 +18,8 @@
       "Left": []
     },
     "nullable": [
-      true,
-      false
+      false,
+      true
     ]
   },
   "hash": "b3dbdfb50ee8118bdaed3164b210cb549a34b96554ae1872355b90304f5dcb76"

backend/.sqlx/query-b9b38d63af3670d1f11d5cbb82a8008a9479bf4b3d7231371ebf26382ecde365.json

@@ -1,12 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "VACUUM v2_job_queue, v2_job_runtime, v2_job_status",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": []
-    },
-    "nullable": []
-  },
-  "hash": "b9b38d63af3670d1f11d5cbb82a8008a9479bf4b3d7231371ebf26382ecde365"
-}

backend/.sqlx/query-c1058d8816d139c63dd9c4a075ab63efc585942b30c9e853f2a5cff4cc9916cd.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "CREATE INDEX CONCURRENTLY IF NOT EXISTS script_not_archived ON script (workspace_id, path, created_at DESC) where archived = false;",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "c1058d8816d139c63dd9c4a075ab63efc585942b30c9e853f2a5cff4cc9916cd"
+}

backend/.sqlx/query-c7cae4cf872fce0a989cf89aa35929218a9d459ee1c2b36a28b110e9741ab623.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n                SELECT content FROM script WHERE path = $1 AND workspace_id = $2\n                AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND\n                workspace_id = $2)\n                ",
+  "query": "\n                SELECT content FROM script WHERE path = $1 AND workspace_id = $2\n                AND archived = false ORDER BY created_at DESC LIMIT 1\n                ",
   "describe": {
     "columns": [
       {
@@ -19,5 +19,5 @@
       false
     ]
   },
-  "hash": "443bd83bcea1d37c79cb080095343c98104529879f991c49585cd181e34aa827"
+  "hash": "c7cae4cf872fce0a989cf89aa35929218a9d459ee1c2b36a28b110e9741ab623"
 }

backend/.sqlx/query-ce6f3e803909d55c19169c77d4111bffc0fc93032943369015971373f1f2af68.json

@@ -0,0 +1,17 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n    INSERT INTO script\n    (workspace_id, hash, path, parent_hashes, summary, description, content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets) \n\n    SELECT  workspace_id, $1, path, array_prepend($2::bigint, COALESCE(parent_hashes, '{}'::bigint[])), summary, description, content, created_by, schema, is_template, extra_perms, $4, language, kind, tag, draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets \n\n    FROM script WHERE hash = $2 AND workspace_id = $3;\n            ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int8",
+        "Int8",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "ce6f3e803909d55c19169c77d4111bffc0fc93032943369015971373f1f2af68"
+}

backend/.sqlx/query-d4c963fa653652b7a3e8529cbf0d0fca091d7c1cb0924f6f9343544abb2666a5.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT value from resource WHERE path = $1 AND workspace_id = $2 AND resource_type = 'json_schema'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "value",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "d4c963fa653652b7a3e8529cbf0d0fca091d7c1cb0924f6f9343544abb2666a5"
+}

backend/.sqlx/query-d768bbc46f8a9c4289b918c88ca50aa180b29bbe931d948f6e61976f71b7cdb9.json

@@ -1,22 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT premium FROM workspace WHERE id = $1",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "premium",
-        "type_info": "Bool"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Text"
-      ]
-    },
-    "nullable": [
-      false
-    ]
-  },
-  "hash": "d768bbc46f8a9c4289b918c88ca50aa180b29bbe931d948f6e61976f71b7cdb9"
-}

backend/.sqlx/query-e64f7044c74e96c2338580562f6b087805dad2b6fb1aa194ac2a2026fa24ecd0.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script where path = $1 AND workspace_id = $2 AND\n    created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND\n    deleted = false AND archived = false)",
+  "query": "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script\n         WHERE path = $1 AND workspace_id = $2 AND archived = false AND (lock IS NOT NULL OR $3 = false)\n         ORDER BY created_at DESC LIMIT 1",
   "describe": {
     "columns": [
       {
@@ -98,7 +98,8 @@
     "parameters": {
       "Left": [
         "Text",
-        "Text"
+        "Text",
+        "Bool"
       ]
     },
     "nullable": [
@@ -116,5 +117,5 @@
       false
     ]
   },
-  "hash": "61c29d684e8e683e839a6d7210b3b9b96854e5bfd752e45922c358c42ebea0c4"
+  "hash": "e64f7044c74e96c2338580562f6b087805dad2b6fb1aa194ac2a2026fa24ecd0"
 }

backend/.sqlx/query-f37140fcdc721a8b199471b30c2baf124affa2eaf56c801c8dac3264c584f981.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "INSERT INTO deployment_metadata (workspace_id, path, app_version, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4, $5)\n                        ON CONFLICT (workspace_id, path) DO UPDATE SET callback_job_ids = $4, deployment_msg = $5",
+  "query": "INSERT INTO deployment_metadata (workspace_id, path, app_version, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4, $5)\n                        ON CONFLICT (workspace_id, path, app_version) WHERE app_version IS NOT NULL DO UPDATE SET callback_job_ids = $4, deployment_msg = $5",
   "describe": {
     "columns": [],
     "parameters": {
@@ -14,5 +14,5 @@
     },
     "nullable": []
   },
-  "hash": "f04632c3a8e0d7c5b48cdd26a99bb1dc5bd12df221f82405d663b8f15f5c0c3a"
+  "hash": "f37140fcdc721a8b199471b30c2baf124affa2eaf56c801c8dac3264c584f981"
 }

backend/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "windmill"
-version = "1.530.0"
+version = "1.537.0"
 authors.workspace = true
 edition.workspace = true
 
@@ -33,7 +33,7 @@ members = [
 ]
 
 [workspace.package]
-version = "1.530.0"
+version = "1.537.0"
 authors = ["Ruben Fiszel <ruben@windmill.dev>"]
 edition = "2021"
 

backend/ee-repo-ref.txt

@@ -1 +1 @@
-15a7592ca66b93b9760d49e58b23c090ead06fe2
+d16e52d570f10dfdabb04a9061fe7ebdbf5a2aa1

backend/migrations/20250826163254_runnable_notify_on_insert.down.sql

@@ -0,0 +1,2 @@
+-- Add down migration script here
+DROP TRIGGER script_insert_trigger ON script;

backend/migrations/20250826163254_runnable_notify_on_insert.up.sql

@@ -0,0 +1,6 @@
+-- Add up migration script here
+CREATE TRIGGER script_insert_trigger
+AFTER INSERT ON script
+FOR EACH ROW
+WHEN (NEW.lock IS NOT NULL)
+EXECUTE FUNCTION notify_runnable_version_change('script');

backend/migrations/20250827102435_add_team_plan_status.down.sql

@@ -0,0 +1,7 @@
+-- Add down migration script here
+DROP FUNCTION notify_team_plan_status_change;
+DROP TRIGGER notify_team_plan_status_change ON cloud_workspace_settings;
+
+ALTER TABLE cloud_workspace_settings 
+    DROP COLUMN is_past_due,
+    DROP COLUMN max_tolerated_executions;

backend/migrations/20250827102435_add_team_plan_status.up.sql

@@ -0,0 +1,18 @@
+-- Add up migration script here
+ALTER TABLE cloud_workspace_settings 
+    ADD COLUMN is_past_due BOOLEAN NOT NULL DEFAULT FALSE,
+    ADD COLUMN max_tolerated_executions INTEGER;
+
+CREATE OR REPLACE FUNCTION notify_team_plan_status_change()
+RETURNS TRIGGER AS $$
+BEGIN
+    PERFORM pg_notify('notify_workspace_premium_change', NEW.workspace_id); -- reuse the same channel as the one used for workspace premium change => clear cache
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+
+CREATE TRIGGER team_plan_status_change_trigger
+AFTER UPDATE OF is_past_due, max_tolerated_executions ON cloud_workspace_settings
+FOR EACH ROW
+EXECUTE FUNCTION notify_team_plan_status_change();

backend/parsers/windmill-parser-py-imports/src/lib.rs

@@ -489,8 +489,7 @@ async fn parse_python_imports_inner(
                     let code = sqlx::query_scalar!(
                         r#"
                 SELECT content FROM script WHERE path = $1 AND workspace_id = $2
-                AND created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND
-                workspace_id = $2)
+                AND archived = false ORDER BY created_at DESC LIMIT 1
                 "#,
                         &rpath,
                         w_id

backend/src/main.rs

@@ -857,7 +857,7 @@ Windmill Community Edition {GIT_VERSION}
                                                 "notify_workspace_premium_change" => {
                                                     let workspace_id = n.payload();
                                                     tracing::info!("Workspace premium change detected, invalidating workspace premium cache: {}", workspace_id);
-                                                    windmill_common::workspaces::IS_PREMIUM_CACHE.remove(workspace_id);
+                                                    windmill_common::workspaces::TEAM_PLAN_CACHE.remove(workspace_id);
                                                 },
                                                 "notify_runnable_version_change" => {
                                                     let payload = n.payload();

backend/src/monitor.rs

@@ -8,7 +8,7 @@ use std::{
         atomic::{AtomicU16, Ordering},
         Arc, Mutex,
     },
-    time::Duration,
+    time::{Duration, Instant},
 };
 
 use chrono::{DateTime, NaiveDateTime, Utc};
@@ -1530,6 +1530,20 @@ pub async fn monitor_db(
         }
     };
 
+    // run every hour
+    let vacuum_queue_f = async {
+        if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(60) {
+            if let Some(db) = conn.as_sql() {
+                let instant = Instant::now();
+                tracing::info!("vacuuming tables");
+                if let Err(e) = vacuuming_tables(&db).await {
+                    tracing::error!("Error vacuuming v2_job: {:?}", e);
+                }
+                tracing::info!("vacuum tables done in {}s", instant.elapsed().as_secs());
+            }
+        }
+    };
+
     let expired_items_f = async {
         if server_mode && !initial_load {
             if let Some(db) = conn.as_sql() {
@@ -1607,6 +1621,7 @@ pub async fn monitor_db(
         expired_items_f,
         zombie_jobs_f,
         stale_jobs_f,
+        vacuum_queue_f,
         expose_queue_metrics_f,
         verify_license_key_f,
         worker_groups_alerts_f,
@@ -1619,6 +1634,13 @@ pub async fn monitor_db(
     );
 }
 
+async fn vacuuming_tables(db: &Pool<Postgres>) -> error::Result<()> {
+    sqlx::query!("VACUUM v2_job, v2_job_completed, job_result_stream, job_stats, job_logs, concurrency_key, log_file, metrics")
+        .execute(db)
+        .await?;
+    Ok(())
+}
+
 pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
     let last_check = sqlx::query_scalar!(
             "SELECT created_at FROM metrics WHERE id LIKE 'queue_count_%' ORDER BY created_at DESC LIMIT 1"
@@ -2571,8 +2593,6 @@ pub async fn reload_app_workspaced_route_setting(conn: &DB) -> error::Result<()>
     let app_workspaced_route =
         load_value_from_global_settings(conn, APP_WORKSPACED_ROUTE_SETTING).await?;
 
-    println!("Updating...");
-
     let ws_route = match app_workspaced_route {
         Some(serde_json::Value::Bool(ws_route)) => ws_route,
         None => false,

backend/tests/worker.rs

@@ -4679,30 +4679,42 @@ mod job_payload {
         test_for_versions(VERSION_FLAGS.iter().cloned(), test).await;
     }
 
-    #[sqlx::test(fixtures("base", "hello"))]
     async fn test_dependencies_payload(db: Pool<Postgres>) {
         initialize_tracing().await;
         let server = ApiServer::start(db.clone()).await;
         let port = server.addr.port();
 
-        let test = || async {
-            let result = RunJob::from(JobPayload::Dependencies {
-                path: "f/system/hello".to_string(),
-                hash: ScriptHash(123412),
-                language: ScriptLang::Deno,
-                dedicated_worker: None,
-            })
-            .run_until_complete(&db, port)
-            .await
-            .json_result()
-            .unwrap();
+        let result = RunJob::from(JobPayload::Dependencies {
+            path: "f/system/hello".to_string(),
+            hash: ScriptHash(123412),
+            language: ScriptLang::Deno,
+            dedicated_worker: None,
+        })
+        .run_until_complete(&db, port)
+        .await
+        .json_result()
+        .unwrap();
 
-            assert_eq!(
-                result.get("status").unwrap(),
-                &json!("Successful lock file generation")
-            );
-        };
-        test_for_versions(VERSION_FLAGS.iter().cloned(), test).await;
+        assert_eq!(
+            result.get("status").unwrap(),
+            &json!("Successful lock file generation")
+        );
+    }
+
+    #[sqlx::test(fixtures("base", "hello"))]
+    async fn test_dependencies_payload_min_1_427(db: Pool<Postgres>) {
+        *MIN_VERSION_IS_AT_LEAST_1_427.write().await = true;
+        test_dependencies_payload(db).await;
+    }
+    #[sqlx::test(fixtures("base", "hello"))]
+    async fn test_dependencies_payload_min_1_432(db: Pool<Postgres>) {
+        *MIN_VERSION_IS_AT_LEAST_1_432.write().await = true;
+        test_dependencies_payload(db).await;
+    }
+    #[sqlx::test(fixtures("base", "hello"))]
+    async fn test_dependencies_payload_min_1_440(db: Pool<Postgres>) {
+        *MIN_VERSION_IS_AT_LEAST_1_440.write().await = true;
+        test_dependencies_payload(db).await;
     }
 
     // Just test that deploying a flow work as expected.

backend/windmill-api/openapi.yaml

@@ -1,7 +1,7 @@
 openapi: "3.0.3"
 
 info:
-  version: 1.530.0
+  version: 1.537.0
   title: Windmill API
 
   contact:
@@ -2034,6 +2034,11 @@ paths:
         - workspace
       parameters:
         - $ref: "#/components/parameters/WorkspaceId"
+        - name: skip_subscription_fetch
+          in: query
+          description: skip fetching subscription status from stripe
+          schema:
+            type: boolean
 
       responses:
         "200":
@@ -2051,9 +2056,14 @@ paths:
                     type: string
                   status:
                     type: string
+                  is_past_due:
+                    type: boolean
+                  max_tolerated_executions:
+                    type: number
                 required:
                   - premium
                   - owner
+                  - is_past_due
 
   /w/{workspace}/workspaces/threshold_alert:
     get:

backend/windmill-api/src/apps.rs

@@ -762,20 +762,26 @@ async fn get_public_resource(
     Path((w_id, path)): Path<(String, StripPath)>,
 ) -> JsonResult<Option<serde_json::Value>> {
     let path = path.to_path();
-    if !path.starts_with("f/app_themes/") {
-        return Err(Error::BadRequest(
-            "Only app themes are public resources".to_string(),
-        ));
-    }
-    let res = sqlx::query_scalar!(
-        "SELECT value from resource WHERE path = $1 AND workspace_id = $2",
-        path.to_owned(),
-        &w_id
-    )
-    .fetch_optional(&db)
-    .await?
-    .flatten();
-    Ok(Json(res))
+
+    let res = if path.starts_with("f/app_themes/") {
+        sqlx::query_scalar!(
+            "SELECT value from resource WHERE path = $1 AND workspace_id = $2",
+            path.to_owned(),
+            &w_id
+        )
+        .fetch_optional(&db)
+        .await?
+    } else {
+        sqlx::query_scalar!(
+            "SELECT value from resource WHERE path = $1 AND workspace_id = $2 AND resource_type = 'json_schema'",
+            path.to_owned(),
+            &w_id
+        )
+        .fetch_optional(&db)
+        .await?
+    };
+
+    Ok(Json(res.flatten()))
 }
 
 async fn get_secret_id(

backend/windmill-api/src/db.rs

@@ -6,8 +6,6 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use std::time::Duration;
-
 use futures::FutureExt;
 use sqlx::{
     migrate::{Migrate, MigrateError},
@@ -17,11 +15,11 @@ use sqlx::{
 
 use tokio::task::JoinHandle;
 use windmill_audit::audit_oss::{AuditAuthor, AuditAuthorable};
+use windmill_common::utils::generate_lock_id;
 use windmill_common::{
     db::{Authable, Authed},
     error::Error,
 };
-use windmill_common::{utils::generate_lock_id, worker::MIN_VERSION_IS_AT_LEAST_1_461};
 
 pub type DB = Pool<Postgres>;
 
@@ -59,7 +57,7 @@ lazy_static::lazy_static! {
                     ].into_iter().collect();
 }
 
-struct CustomMigrator {
+pub struct CustomMigrator {
     inner: PoolConnection<Postgres>,
 }
 impl Migrate for CustomMigrator {
@@ -243,586 +241,7 @@ pub async fn migrate(db: &DB) -> Result<Option<JoinHandle<()>>, Error> {
         Err(err) => Err(err),
     }?;
 
-    if let Err(err) = fix_flow_versioning_migration(&mut custom_migrator, db).await {
-        tracing::error!("Could not apply flow versioning fix migration: {err:#}");
-    }
-
-    let db2 = db.clone();
-    let _ = tokio::task::spawn(async move {
-        if let Err(err) = fix_job_completed_index(&db2).await {
-            tracing::error!("Could not apply job completed index fix migration: {err:#}");
-        }
-    });
-
-    let mut jh = None;
-    if !has_done_migration(db, "v2_finalize_job_completed").await {
-        let db2 = db.clone();
-        let v2jh = tokio::task::spawn(async move {
-            loop {
-                if !*MIN_VERSION_IS_AT_LEAST_1_461.read().await {
-                    tracing::info!("Waiting for all workers to be at least version 1.461 before applying v2 finalize migration, sleeping for 5s...");
-                    tokio::time::sleep(Duration::from_secs(5)).await;
-                    continue;
-                }
-                if let Err(err) = v2_finalize(&db2).await {
-                    tracing::error!(
-                        "{err:#}: Could not apply v2 finalize migration, retry in 30s.."
-                    );
-                    tokio::time::sleep(Duration::from_secs(30)).await;
-                    continue;
-                }
-                tracing::info!("v2 finalization step successfully applied.");
-                break;
-            }
-        });
-        jh = Some(v2jh)
-    }
-
-    Ok(jh)
-}
-
-async fn fix_flow_versioning_migration(
-    migrator: &mut CustomMigrator,
-    db: &DB,
-) -> Result<(), Error> {
-    let has_done_migration = sqlx::query_scalar!(
-        "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_flow_versioning_2')",
-    )
-    .fetch_one(db)
-    .await?
-    .unwrap_or(false);
-
-    if !has_done_migration {
-        migrator.lock().await?;
-
-        if migrator
-            .list_applied_migrations()
-            .await?
-            .iter()
-            .any(|x| x.version == 20240630102146)
-        {
-            let has_done_migration = sqlx::query_scalar!(
-                "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_flow_versioning_2')",
-            )
-            .fetch_one(db)
-            .await?
-            .unwrap_or(false);
-
-            if !has_done_migration {
-                let query = include_str!("../../custom_migrations/fix_flow_versioning_2.sql");
-                tracing::info!("Applying fix_flow_versioning_2.sql");
-                let mut tx: sqlx::Transaction<'_, Postgres> = db.begin().await?;
-                tx.execute(query).await?;
-                tracing::info!("Applied fix_flow_versioning_2.sql");
-                sqlx::query!(
-                    "INSERT INTO windmill_migrations (name) VALUES ('fix_flow_versioning_2')"
-                )
-                .execute(&mut *tx)
-                .await?;
-                tx.commit().await?;
-            }
-        }
-
-        migrator.unlock().await?;
-    }
-    Ok(())
-}
-
-async fn has_done_migration(db: &DB, migration_job_name: &str) -> bool {
-    sqlx::query_scalar!(
-        "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = $1)",
-        migration_job_name
-    )
-    .fetch_one(db)
-    .await
-    .ok()
-    .flatten()
-    .unwrap_or(false)
-}
-
-macro_rules! run_windmill_migration {
-    ($migration_job_name:expr, $db:expr, |$tx:ident| $code:block) => {
-        {
-            let migration_job_name = $migration_job_name;
-            let db: &Pool<Postgres> = $db;
-
-            let has_done = has_done_migration(db, migration_job_name).await;
-            if !has_done {
-                tracing::info!("Applying {migration_job_name} migration");
-                let mut $tx = db.begin().await?;
-                let mut r = false;
-                while !r {
-                    r = sqlx::query_scalar!("SELECT pg_try_advisory_lock(4242)")
-                        .fetch_one(&mut *$tx)
-                        .await
-                        .map_err(|e| {
-                            tracing::error!("Error acquiring {migration_job_name} lock: {e:#}");
-                            sqlx::migrate::MigrateError::Execute(e)
-                        })?
-                        .unwrap_or(false);
-
-                    if !r {
-                        tracing::info!("PG {migration_job_name} lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)");
-                        drop($tx);
-                        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
-                        $tx = db.begin().await?;
-                    }
-                }
-                tracing::info!("acquired lock for {migration_job_name}");
-
-                let has_done = has_done_migration(db, migration_job_name).await;
-
-                if !has_done {
-
-                    $code
-
-                    sqlx::query!(
-                        "INSERT INTO windmill_migrations (name) VALUES ($1) ON CONFLICT DO NOTHING",
-                        migration_job_name
-                    )
-                    .execute(&mut *$tx)
-                    .await?;
-                    tracing::info!("Finished applying {migration_job_name} migration");
-                } else {
-                    tracing::debug!("migration {migration_job_name} already done");
-                }
-
-                let _ = sqlx::query("SELECT pg_advisory_unlock(4242)")
-                    .execute(&mut *$tx)
-                    .await?;
-                $tx.commit().await?;
-                tracing::info!("released lock for {migration_job_name}");
-            } else {
-                tracing::debug!("migration {migration_job_name} already done");
-
-            }
-        }
-    };
-}
-
-async fn v2_finalize(db: &DB) -> Result<(), Error> {
-    run_windmill_migration!("v2_finalize_disable_sync_III", db, |tx| {
-        tx.execute(
-            r#"
-            LOCK TABLE v2_job_queue IN ACCESS EXCLUSIVE MODE;
-            ALTER TABLE v2_job_queue DISABLE ROW LEVEL SECURITY;
-            "#,
-        )
-        .await?;
-    });
-
-    run_windmill_migration!("v2_finalize_disable_sync_III_2", db, |tx| {
-        tx.execute(
-            r#"
-            LOCK TABLE v2_job_completed IN ACCESS EXCLUSIVE MODE;
-            ALTER TABLE v2_job_completed DISABLE ROW LEVEL SECURITY;
-            "#,
-        )
-        .await?;
-    });
-
-    run_windmill_migration!("v2_finalize_disable_sync_III_3", db, |tx| {
-        tx.execute(
-            r#"
-            LOCK TABLE v2_job IN ACCESS EXCLUSIVE MODE;
-            DROP FUNCTION IF EXISTS v2_job_after_update CASCADE;
-        "#,
-        )
-        .await?;
-    });
-
-    run_windmill_migration!("v2_finalize_disable_sync_III_4", db, |tx| {
-        tx.execute(
-            r#"
-            LOCK TABLE v2_job_completed IN ACCESS EXCLUSIVE MODE;
-            DROP FUNCTION IF EXISTS v2_job_completed_before_insert CASCADE;
-            DROP FUNCTION IF EXISTS v2_job_completed_before_update CASCADE;
-            "#,
-        )
-        .await?;
-    });
-
-    run_windmill_migration!("v2_finalize_disable_sync_III_5", db, |tx| {
-        tx.execute(
-            r#"
-            LOCK TABLE v2_job_queue IN ACCESS EXCLUSIVE MODE;
-            DROP FUNCTION IF EXISTS v2_job_queue_after_insert CASCADE;
-            DROP FUNCTION IF EXISTS v2_job_queue_before_insert CASCADE;
-            DROP FUNCTION IF EXISTS v2_job_queue_before_update CASCADE;
-            "#,
-        )
-        .await?;
-    });
-
-    run_windmill_migration!("v2_finalize_disable_sync_III_6", db, |tx| {
-        tx.execute(
-            r#"
-            LOCK TABLE v2_job_runtime IN ACCESS EXCLUSIVE MODE;
-            DROP FUNCTION IF EXISTS v2_job_runtime_before_insert CASCADE;
-            DROP FUNCTION IF EXISTS v2_job_runtime_before_update CASCADE;
-            "#,
-        )
-        .await?;
-    });
-
-    run_windmill_migration!("v2_finalize_disable_sync_III_7", db, |tx| {
-        tx.execute(
-            r#"
-            LOCK TABLE v2_job_status IN ACCESS EXCLUSIVE MODE;
-            DROP FUNCTION IF EXISTS v2_job_status_before_insert CASCADE;
-            DROP FUNCTION IF EXISTS v2_job_status_before_update CASCADE;
-            "#,
-        )
-        .await?;
-    });
-
-    run_windmill_migration!("v2_finalize_disable_sync_III_8", db, |tx| {
-        tx.execute(
-            r#"
-            DROP VIEW IF EXISTS completed_job, completed_job_view, job, queue, queue_view CASCADE;
-            "#,
-        )
-        .await?;
-    });
-
-    run_windmill_migration!("v2_finalize_job_queue", db, |tx| {
-        tx.execute(
-            r#"
-            LOCK TABLE v2_job_queue IN ACCESS EXCLUSIVE MODE;
-            ALTER TABLE v2_job_queue
-                DROP COLUMN IF EXISTS __parent_job CASCADE,
-                DROP COLUMN IF EXISTS __created_by CASCADE,
-                DROP COLUMN IF EXISTS __script_hash CASCADE,
-                DROP COLUMN IF EXISTS __script_path CASCADE,
-                DROP COLUMN IF EXISTS __args CASCADE,
-                DROP COLUMN IF EXISTS __logs CASCADE,
-                DROP COLUMN IF EXISTS __raw_code CASCADE,
-                DROP COLUMN IF EXISTS __canceled CASCADE,
-                DROP COLUMN IF EXISTS __last_ping CASCADE,
-                DROP COLUMN IF EXISTS __job_kind CASCADE,
-                DROP COLUMN IF EXISTS __env_id CASCADE,
-                DROP COLUMN IF EXISTS __schedule_path CASCADE,
-                DROP COLUMN IF EXISTS __permissioned_as CASCADE,
-                DROP COLUMN IF EXISTS __flow_status CASCADE,
-                DROP COLUMN IF EXISTS __raw_flow CASCADE,
-                DROP COLUMN IF EXISTS __is_flow_step CASCADE,
-                DROP COLUMN IF EXISTS __language CASCADE,
-                DROP COLUMN IF EXISTS __same_worker CASCADE,
-                DROP COLUMN IF EXISTS __raw_lock CASCADE,
-                DROP COLUMN IF EXISTS __pre_run_error CASCADE,
-                DROP COLUMN IF EXISTS __email CASCADE,
-                DROP COLUMN IF EXISTS __visible_to_owner CASCADE,
-                DROP COLUMN IF EXISTS __mem_peak CASCADE,
-                DROP COLUMN IF EXISTS __root_job CASCADE,
-                DROP COLUMN IF EXISTS __leaf_jobs CASCADE,
-                DROP COLUMN IF EXISTS __concurrent_limit CASCADE,
-                DROP COLUMN IF EXISTS __concurrency_time_window_s CASCADE,
-                DROP COLUMN IF EXISTS __timeout CASCADE,
-                DROP COLUMN IF EXISTS __flow_step_id CASCADE,
-                DROP COLUMN IF EXISTS __cache_ttl CASCADE;
-            "#,
-        )
-        .await?;
-    });
-    run_windmill_migration!("v2_finalize_job_completed", db, |tx| {
-        tx.execute(
-            r#"
-            LOCK TABLE v2_job_completed IN ACCESS EXCLUSIVE MODE;
-            ALTER TABLE v2_job_completed
-                DROP COLUMN IF EXISTS __parent_job CASCADE,
-                DROP COLUMN IF EXISTS __created_by CASCADE,
-                DROP COLUMN IF EXISTS __created_at CASCADE,
-                DROP COLUMN IF EXISTS __success CASCADE,
-                DROP COLUMN IF EXISTS __script_hash CASCADE,
-                DROP COLUMN IF EXISTS __script_path CASCADE,
-                DROP COLUMN IF EXISTS __args CASCADE,
-                DROP COLUMN IF EXISTS __logs CASCADE,
-                DROP COLUMN IF EXISTS __raw_code CASCADE,
-                DROP COLUMN IF EXISTS __canceled CASCADE,
-                DROP COLUMN IF EXISTS __job_kind CASCADE,
-                DROP COLUMN IF EXISTS __env_id CASCADE,
-                DROP COLUMN IF EXISTS __schedule_path CASCADE,
-                DROP COLUMN IF EXISTS __permissioned_as CASCADE,
-                DROP COLUMN IF EXISTS __raw_flow CASCADE,
-                DROP COLUMN IF EXISTS __is_flow_step CASCADE,
-                DROP COLUMN IF EXISTS __language CASCADE,
-                DROP COLUMN IF EXISTS __is_skipped CASCADE,
-                DROP COLUMN IF EXISTS __raw_lock CASCADE,
-                DROP COLUMN IF EXISTS __email CASCADE,
-                DROP COLUMN IF EXISTS __visible_to_owner CASCADE,
-                DROP COLUMN IF EXISTS __tag CASCADE,
-                DROP COLUMN IF EXISTS __priority CASCADE;
-            "#,
-        )
-        .await?;
-    });
-
-    Ok(())
-}
-
-async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
-    // let has_done_migration = sqlx::query_scalar!(
-    //     "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_job_completed_index')"
-    // )
-    // .fetch_one(db)
-    // .await?
-    // .unwrap_or(false);
-    // if !has_done_migration {
-    //     tracing::info!("Applying fix_job_completed_index migration");
-    //     let mut tx = db.begin().await?;
-    //     let mut r = false;
-    //     while !r {
-    //         r = sqlx::query_scalar!("SELECT pg_try_advisory_lock(4242)")
-    //             .fetch_one(&mut *tx)
-    //             .await
-    //             .map_err(|e| {
-    //                 tracing::error!("Error acquiring fix_job_completed_index lock: {e:#}");
-    //                 sqlx::migrate::MigrateError::Execute(e)
-    //             })?
-    //             .unwrap_or(false);
-    //         if !r {
-    //             tracing::info!("PG fix_job_completed_index_migration lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)");
-    //             tokio::time::sleep(std::time::Duration::from_secs(5)).await;
-    //         }
-    //     }
-    //     // sqlx::query(
-    //     //     "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new ON completed_job (workspace_id, job_kind, is_skipped, is_flow_step, created_at DESC, started_at DESC)"
-    //     // ).execute(db).await?;
-
-    //     sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at")
-    //         .execute(db)
-    //         .await?;
-
-    //     sqlx::query!("INSERT INTO windmill_migrations (name) VALUES ('fix_job_completed_index') ON CONFLICT DO NOTHING")
-    //         .execute(&mut *tx)
-    //         .await?;
-    //     let _ = sqlx::query("SELECT pg_advisory_unlock(4242)")
-    //         .execute(&mut *tx)
-    //         .await?;
-    //     tx.commit().await?;
-    // }
-
-    run_windmill_migration!("fix_job_completed_index_2", &db, |tx| {
-        //     sqlx::query(
-        //     "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new_2 ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, created_at DESC)"
-        // ).execute(db).await?;
-
-        //     sqlx::query(
-        //     "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_started_at_new ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, started_at DESC)"
-        // ).execute(db).await?;
-
-        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at")
-            .execute(db)
-            .await?;
-
-        sqlx::query(
-            "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new",
-        )
-        .execute(db)
-        .await?;
-    });
-
-    run_windmill_migration!("fix_job_completed_index_3", &db, |tx| {
-        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_job_on_schedule_path")
-            .execute(db)
-            .await?;
-
-        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS concurrency_limit_stats_queue")
-            .execute(db)
-            .await?;
-
-        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS root_job_index")
-            .execute(db)
-            .await?;
-
-        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_on_created")
-            .execute(db)
-            .await?;
-    });
-
-    run_windmill_migration!("fix_job_index_1_II", &db, |tx| {
-        let migration_job_name = "fix_job_index_1_II";
-        let mut i = 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-        sqlx::query!("create index concurrently  if not exists ix_job_workspace_id_created_at_new_3 ON v2_job  (workspace_id,  created_at DESC)")
-                .execute(db)
-                .await?;
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_8 ON v2_job  (workspace_id, created_at DESC) where kind in ('deploymentcallback') AND parent_job IS NULL")
-            .execute(db)
-            .await?;
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_9 ON v2_job  (workspace_id, created_at DESC) where kind in ('dependencies', 'flowdependencies', 'appdependencies') AND parent_job IS NULL")
-            .execute(db)
-            .await?;
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_5 ON v2_job  (workspace_id, created_at DESC) where kind in ('preview', 'flowpreview') AND parent_job IS NULL")
-                .execute(db)
-                .await?;
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query!("create index concurrently if not exists ix_completed_job_workspace_id_started_at_new_2 ON v2_job_completed  (workspace_id, started_at DESC)")
-                .execute(db)
-                .await?;
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query!("create index concurrently if not exists ix_job_root_job_index_by_path_2 ON v2_job (workspace_id, runnable_path, created_at desc) WHERE parent_job IS NULL")
-                .execute(db)
-                .await?;
-
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path_2")
-            .execute(db)
-            .await?;
-
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query!("create index concurrently if not exists ix_job_created_at ON v2_job  (created_at DESC)")
-                    .execute(db)
-                    .await?;
-
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query(
-            "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new_2",
-        )
-        .execute(db)
-        .await?;
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query(
-            "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_started_at_new",
-        )
-        .execute(db)
-        .await?;
-        i += 1;
-        tracing::info!("step {i} of {migration_job_name} migration");
-
-        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path")
-            .execute(db)
-            .await?;
-    });
-
-    run_windmill_migration!("fix_labeled_jobs_index", &db, |tx| {
-        tracing::info!("Special migration to add index concurrently on job labels 2");
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS labeled_jobs_on_jobs")
-            .execute(db)
-            .await?;
-        sqlx::query!(
-        "CREATE INDEX CONCURRENTLY labeled_jobs_on_jobs ON v2_job_completed USING GIN ((result -> 'wm_labels')) WHERE result ? 'wm_labels'"
-        ).execute(db).await?;
-    });
-
-    run_windmill_migration!("v2_labeled_jobs_index", &db, |tx| {
-        tracing::info!("Special migration to add index concurrently on job labels");
-        sqlx::query!(
-            "CREATE INDEX CONCURRENTLY ix_v2_job_labels ON v2_job
-                USING GIN (labels)
-                WHERE labels IS NOT NULL"
-        )
-        .execute(db)
-        .await?;
-    });
-
-    run_windmill_migration!("v2_jobs_rls", &db, |tx| {
-        sqlx::query!("ALTER TABLE v2_job ENABLE ROW LEVEL SECURITY")
-            .execute(db)
-            .await?;
-    });
-
-    run_windmill_migration!("v2_improve_v2_job_indices_ii", &db, |tx| {
-        sqlx::query!("create index concurrently if not exists ix_v2_job_workspace_id_created_at ON v2_job  (workspace_id, created_at DESC) where kind in ('script', 'flow', 'singlescriptflow') AND parent_job IS NULL")
-                .execute(db)
-                .await?;
-
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_6")
-            .execute(db)
-            .await?;
-
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_7")
-            .execute(db)
-            .await?;
-    });
-
-    run_windmill_migration!("v2_improve_v2_queued_jobs_indices", &db, |tx| {
-        sqlx::query!("CREATE INDEX CONCURRENTLY IF NOT EXISTS queue_sort_v2 ON v2_job_queue (priority DESC NULLS LAST, scheduled_for, tag) WHERE running = false")
-            .execute(db)
-            .await?;
-
-        // sqlx::query!("CREATE INDEX CONCURRENTLY queue_sort_2_v2 ON v2_job_queue (tag, priority DESC NULLS LAST, scheduled_for) WHERE running = false")
-        //     .execute(db)
-        //     .await?;
-
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS queue_sort")
-            .execute(db)
-            .await?;
-
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS queue_sort_2")
-            .execute(db)
-            .await?;
-    });
-
-    run_windmill_migration!("audit_timestamps", db, |tx| {
-        sqlx::query!(
-            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_audit_timestamps ON audit (timestamp DESC)"
-        )
-        .execute(db)
-        .await?;
-    });
-
-    run_windmill_migration!("job_completed_completed_at", db, |tx| {
-        sqlx::query!(
-            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_completed_completed_at ON v2_job_completed (completed_at DESC)"
-        )
-        .execute(db)
-        .await?;
-    });
-
-    run_windmill_migration!("alerts_by_workspace", db, |tx| {
-        sqlx::query!(
-            "CREATE INDEX CONCURRENTLY IF NOT EXISTS alerts_by_workspace ON alerts (workspace_id);"
-        )
-        .execute(db)
-        .await?;
-    });
-
-    run_windmill_migration!("remove_redundant_log_file_index", db, |tx| {
-        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS log_file_hostname_log_ts_idx")
-            .execute(db)
-            .await?;
-    });
-
-    run_windmill_migration!("v2_job_queue_suspend", db, |tx| {
-        sqlx::query!(
-            "CREATE INDEX CONCURRENTLY IF NOT EXISTS v2_job_queue_suspend ON v2_job_queue (workspace_id, suspend) WHERE suspend > 0;"
-        )
-        .execute(db)
-        .await?;
-    });
-
-    run_windmill_migration!("audit_recent_login_activities", db, |tx| {
-        sqlx::query!(
-            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_audit_recent_login_activities 
-ON audit (timestamp, username) 
-WHERE operation IN ('users.login', 'oauth.login', 'users.token.refresh');"
-        )
-        .execute(db)
-        .await?;
-    });
-    Ok(())
+    return crate::live_migrations::custom_migrations(&mut custom_migrator, db).await;
 }
 
 #[derive(Clone, Debug, Default, Hash, Eq, PartialEq)]

backend/windmill-api/src/lib.rs

@@ -103,6 +103,7 @@ mod inkeep_ee;
 mod inkeep_oss;
 mod inputs;
 mod integration;
+mod live_migrations;
 #[cfg(feature = "postgres_trigger")]
 mod postgres_triggers;
 
@@ -623,7 +624,7 @@ pub async fn run_server(
                         .nest("/mqtt_triggers", mqtt_triggers_service)
                         .nest("/sqs_triggers", sqs_triggers_service)
                         .nest("/gcp_triggers", gcp_triggers_service)
-                    .nest("/postgres_triggers", postgres_triggers_service),
+                        .nest("/postgres_triggers", postgres_triggers_service),
                 )
                 .nest("/workspaces", workspaces::global_service())
                 .nest(

backend/windmill-api/src/live_migrations.rs

@@ -0,0 +1,613 @@
+/*
+ * Author: Ruben Fiszel
+ * Copyright: Windmill Labs, Inc 2022
+ * This file and its contents are licensed under the AGPLv3 License.
+ * Please see the included NOTICE for copyright information and
+ * LICENSE-AGPL for a copy of the license.
+ */
+
+use sqlx::Postgres;
+use std::time::Duration;
+use tokio::task::JoinHandle;
+use windmill_common::error::Error;
+use windmill_common::worker::MIN_VERSION_IS_AT_LEAST_1_461;
+
+use crate::db::{CustomMigrator, DB};
+use sqlx::migrate::Migrate;
+use sqlx::Executor;
+
+pub async fn custom_migrations(
+    migrator: &mut CustomMigrator,
+    db: &DB,
+) -> Result<Option<JoinHandle<()>>, Error> {
+    if let Err(err) = fix_flow_versioning_migration(migrator, db).await {
+        tracing::error!("Could not apply flow versioning fix migration: {err:#}");
+    }
+
+    let db2 = db.clone();
+    let _ = tokio::task::spawn(async move {
+        if let Err(err) = fix_job_completed_index(&db2).await {
+            tracing::error!("Could not apply job completed index fix migration: {err:#}");
+        }
+    });
+
+    let mut jh = None;
+    if !has_done_migration(db, "v2_finalize_job_completed").await {
+        let db2 = db.clone();
+        let v2jh = tokio::task::spawn(async move {
+            loop {
+                if !*MIN_VERSION_IS_AT_LEAST_1_461.read().await {
+                    tracing::info!("Waiting for all workers to be at least version 1.461 before applying v2 finalize migration, sleeping for 5s...");
+                    tokio::time::sleep(Duration::from_secs(5)).await;
+                    continue;
+                }
+                if let Err(err) = v2_finalize(&db2).await {
+                    tracing::error!(
+                        "{err:#}: Could not apply v2 finalize migration, retry in 30s.."
+                    );
+                    tokio::time::sleep(Duration::from_secs(30)).await;
+                    continue;
+                }
+                tracing::info!("v2 finalization step successfully applied.");
+                break;
+            }
+        });
+        jh = Some(v2jh)
+    }
+
+    Ok(jh)
+}
+
+async fn fix_flow_versioning_migration(
+    migrator: &mut CustomMigrator,
+    db: &DB,
+) -> Result<(), Error> {
+    let has_done_migration = sqlx::query_scalar!(
+        "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_flow_versioning_2')",
+    )
+    .fetch_one(db)
+    .await?
+    .unwrap_or(false);
+
+    if !has_done_migration {
+        migrator.lock().await?;
+
+        if migrator
+            .list_applied_migrations()
+            .await?
+            .iter()
+            .any(|x| x.version == 20240630102146)
+        {
+            let has_done_migration = sqlx::query_scalar!(
+                "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_flow_versioning_2')",
+            )
+            .fetch_one(db)
+            .await?
+            .unwrap_or(false);
+
+            if !has_done_migration {
+                let query = include_str!("../../custom_migrations/fix_flow_versioning_2.sql");
+                tracing::info!("Applying fix_flow_versioning_2.sql");
+                let mut tx: sqlx::Transaction<'_, Postgres> = db.begin().await?;
+                tx.execute(query).await?;
+                tracing::info!("Applied fix_flow_versioning_2.sql");
+                sqlx::query!(
+                    "INSERT INTO windmill_migrations (name) VALUES ('fix_flow_versioning_2')"
+                )
+                .execute(&mut *tx)
+                .await?;
+                tx.commit().await?;
+            }
+        }
+
+        migrator.unlock().await?;
+    }
+    Ok(())
+}
+
+async fn has_done_migration(db: &DB, migration_job_name: &str) -> bool {
+    sqlx::query_scalar!(
+        "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = $1)",
+        migration_job_name
+    )
+    .fetch_one(db)
+    .await
+    .ok()
+    .flatten()
+    .unwrap_or(false)
+}
+
+use sqlx::Pool;
+
+macro_rules! run_windmill_migration {
+    ($migration_job_name:expr, $db:expr, |$tx:ident| $code:block) => {
+        {
+            let migration_job_name = $migration_job_name;
+            let db: &Pool<Postgres> = $db;
+
+            let has_done = has_done_migration(db, migration_job_name).await;
+            if !has_done {
+                tracing::info!("Applying {migration_job_name} migration");
+                let mut $tx = db.begin().await?;
+                let mut r = false;
+                while !r {
+                    r = sqlx::query_scalar!("SELECT pg_try_advisory_lock(4242)")
+                        .fetch_one(&mut *$tx)
+                        .await
+                        .map_err(|e| {
+                            tracing::error!("Error acquiring {migration_job_name} lock: {e:#}");
+                            sqlx::migrate::MigrateError::Execute(e)
+                        })?
+                        .unwrap_or(false);
+
+                    if !r {
+                        tracing::info!("PG {migration_job_name} lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)");
+                        drop($tx);
+                        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+                        $tx = db.begin().await?;
+                    }
+                }
+                tracing::info!("acquired lock for {migration_job_name}");
+
+                let has_done = has_done_migration(db, migration_job_name).await;
+
+                if !has_done {
+
+                    $code
+
+                    sqlx::query!(
+                        "INSERT INTO windmill_migrations (name) VALUES ($1) ON CONFLICT DO NOTHING",
+                        migration_job_name
+                    )
+                    .execute(&mut *$tx)
+                    .await?;
+                    tracing::info!("Finished applying {migration_job_name} migration");
+                } else {
+                    tracing::debug!("migration {migration_job_name} already done");
+                }
+
+                let _ = sqlx::query("SELECT pg_advisory_unlock(4242)")
+                    .execute(&mut *$tx)
+                    .await?;
+                $tx.commit().await?;
+                tracing::info!("released lock for {migration_job_name}");
+            } else {
+                tracing::debug!("migration {migration_job_name} already done");
+
+            }
+        }
+    };
+}
+
+async fn v2_finalize(db: &DB) -> Result<(), Error> {
+    run_windmill_migration!("v2_finalize_disable_sync_III", db, |tx| {
+        tx.execute(
+            r#"
+            LOCK TABLE v2_job_queue IN ACCESS EXCLUSIVE MODE;
+            ALTER TABLE v2_job_queue DISABLE ROW LEVEL SECURITY;
+            "#,
+        )
+        .await?;
+    });
+
+    run_windmill_migration!("v2_finalize_disable_sync_III_2", db, |tx| {
+        tx.execute(
+            r#"
+            LOCK TABLE v2_job_completed IN ACCESS EXCLUSIVE MODE;
+            ALTER TABLE v2_job_completed DISABLE ROW LEVEL SECURITY;
+            "#,
+        )
+        .await?;
+    });
+
+    run_windmill_migration!("v2_finalize_disable_sync_III_3", db, |tx| {
+        tx.execute(
+            r#"
+            LOCK TABLE v2_job IN ACCESS EXCLUSIVE MODE;
+            DROP FUNCTION IF EXISTS v2_job_after_update CASCADE;
+        "#,
+        )
+        .await?;
+    });
+
+    run_windmill_migration!("v2_finalize_disable_sync_III_4", db, |tx| {
+        tx.execute(
+            r#"
+            LOCK TABLE v2_job_completed IN ACCESS EXCLUSIVE MODE;
+            DROP FUNCTION IF EXISTS v2_job_completed_before_insert CASCADE;
+            DROP FUNCTION IF EXISTS v2_job_completed_before_update CASCADE;
+            "#,
+        )
+        .await?;
+    });
+
+    run_windmill_migration!("v2_finalize_disable_sync_III_5", db, |tx| {
+        tx.execute(
+            r#"
+            LOCK TABLE v2_job_queue IN ACCESS EXCLUSIVE MODE;
+            DROP FUNCTION IF EXISTS v2_job_queue_after_insert CASCADE;
+            DROP FUNCTION IF EXISTS v2_job_queue_before_insert CASCADE;
+            DROP FUNCTION IF EXISTS v2_job_queue_before_update CASCADE;
+            "#,
+        )
+        .await?;
+    });
+
+    run_windmill_migration!("v2_finalize_disable_sync_III_6", db, |tx| {
+        tx.execute(
+            r#"
+            LOCK TABLE v2_job_runtime IN ACCESS EXCLUSIVE MODE;
+            DROP FUNCTION IF EXISTS v2_job_runtime_before_insert CASCADE;
+            DROP FUNCTION IF EXISTS v2_job_runtime_before_update CASCADE;
+            "#,
+        )
+        .await?;
+    });
+
+    run_windmill_migration!("v2_finalize_disable_sync_III_7", db, |tx| {
+        tx.execute(
+            r#"
+            LOCK TABLE v2_job_status IN ACCESS EXCLUSIVE MODE;
+            DROP FUNCTION IF EXISTS v2_job_status_before_insert CASCADE;
+            DROP FUNCTION IF EXISTS v2_job_status_before_update CASCADE;
+            "#,
+        )
+        .await?;
+    });
+
+    run_windmill_migration!("v2_finalize_disable_sync_III_8", db, |tx| {
+        tx.execute(
+            r#"
+            DROP VIEW IF EXISTS completed_job, completed_job_view, job, queue, queue_view CASCADE;
+            "#,
+        )
+        .await?;
+    });
+
+    run_windmill_migration!("v2_finalize_job_queue", db, |tx| {
+        tx.execute(
+            r#"
+            LOCK TABLE v2_job_queue IN ACCESS EXCLUSIVE MODE;
+            ALTER TABLE v2_job_queue
+                DROP COLUMN IF EXISTS __parent_job CASCADE,
+                DROP COLUMN IF EXISTS __created_by CASCADE,
+                DROP COLUMN IF EXISTS __script_hash CASCADE,
+                DROP COLUMN IF EXISTS __script_path CASCADE,
+                DROP COLUMN IF EXISTS __args CASCADE,
+                DROP COLUMN IF EXISTS __logs CASCADE,
+                DROP COLUMN IF EXISTS __raw_code CASCADE,
+                DROP COLUMN IF EXISTS __canceled CASCADE,
+                DROP COLUMN IF EXISTS __last_ping CASCADE,
+                DROP COLUMN IF EXISTS __job_kind CASCADE,
+                DROP COLUMN IF EXISTS __env_id CASCADE,
+                DROP COLUMN IF EXISTS __schedule_path CASCADE,
+                DROP COLUMN IF EXISTS __permissioned_as CASCADE,
+                DROP COLUMN IF EXISTS __flow_status CASCADE,
+                DROP COLUMN IF EXISTS __raw_flow CASCADE,
+                DROP COLUMN IF EXISTS __is_flow_step CASCADE,
+                DROP COLUMN IF EXISTS __language CASCADE,
+                DROP COLUMN IF EXISTS __same_worker CASCADE,
+                DROP COLUMN IF EXISTS __raw_lock CASCADE,
+                DROP COLUMN IF EXISTS __pre_run_error CASCADE,
+                DROP COLUMN IF EXISTS __email CASCADE,
+                DROP COLUMN IF EXISTS __visible_to_owner CASCADE,
+                DROP COLUMN IF EXISTS __mem_peak CASCADE,
+                DROP COLUMN IF EXISTS __root_job CASCADE,
+                DROP COLUMN IF EXISTS __leaf_jobs CASCADE,
+                DROP COLUMN IF EXISTS __concurrent_limit CASCADE,
+                DROP COLUMN IF EXISTS __concurrency_time_window_s CASCADE,
+                DROP COLUMN IF EXISTS __timeout CASCADE,
+                DROP COLUMN IF EXISTS __flow_step_id CASCADE,
+                DROP COLUMN IF EXISTS __cache_ttl CASCADE;
+            "#,
+        )
+        .await?;
+    });
+    run_windmill_migration!("v2_finalize_job_completed", db, |tx| {
+        tx.execute(
+            r#"
+            LOCK TABLE v2_job_completed IN ACCESS EXCLUSIVE MODE;
+            ALTER TABLE v2_job_completed
+                DROP COLUMN IF EXISTS __parent_job CASCADE,
+                DROP COLUMN IF EXISTS __created_by CASCADE,
+                DROP COLUMN IF EXISTS __created_at CASCADE,
+                DROP COLUMN IF EXISTS __success CASCADE,
+                DROP COLUMN IF EXISTS __script_hash CASCADE,
+                DROP COLUMN IF EXISTS __script_path CASCADE,
+                DROP COLUMN IF EXISTS __args CASCADE,
+                DROP COLUMN IF EXISTS __logs CASCADE,
+                DROP COLUMN IF EXISTS __raw_code CASCADE,
+                DROP COLUMN IF EXISTS __canceled CASCADE,
+                DROP COLUMN IF EXISTS __job_kind CASCADE,
+                DROP COLUMN IF EXISTS __env_id CASCADE,
+                DROP COLUMN IF EXISTS __schedule_path CASCADE,
+                DROP COLUMN IF EXISTS __permissioned_as CASCADE,
+                DROP COLUMN IF EXISTS __raw_flow CASCADE,
+                DROP COLUMN IF EXISTS __is_flow_step CASCADE,
+                DROP COLUMN IF EXISTS __language CASCADE,
+                DROP COLUMN IF EXISTS __is_skipped CASCADE,
+                DROP COLUMN IF EXISTS __raw_lock CASCADE,
+                DROP COLUMN IF EXISTS __email CASCADE,
+                DROP COLUMN IF EXISTS __visible_to_owner CASCADE,
+                DROP COLUMN IF EXISTS __tag CASCADE,
+                DROP COLUMN IF EXISTS __priority CASCADE;
+            "#,
+        )
+        .await?;
+    });
+
+    Ok(())
+}
+
+async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
+    // let has_done_migration = sqlx::query_scalar!(
+    //     "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_job_completed_index')"
+    // )
+    // .fetch_one(db)
+    // .await?
+    // .unwrap_or(false);
+    // if !has_done_migration {
+    //     tracing::info!("Applying fix_job_completed_index migration");
+    //     let mut tx = db.begin().await?;
+    //     let mut r = false;
+    //     while !r {
+    //         r = sqlx::query_scalar!("SELECT pg_try_advisory_lock(4242)")
+    //             .fetch_one(&mut *tx)
+    //             .await
+    //             .map_err(|e| {
+    //                 tracing::error!("Error acquiring fix_job_completed_index lock: {e:#}");
+    //                 sqlx::migrate::MigrateError::Execute(e)
+    //             })?
+    //             .unwrap_or(false);
+    //         if !r {
+    //             tracing::info!("PG fix_job_completed_index_migration lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)");
+    //             tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+    //         }
+    //     }
+    //     // sqlx::query(
+    //     //     "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new ON completed_job (workspace_id, job_kind, is_skipped, is_flow_step, created_at DESC, started_at DESC)"
+    //     // ).execute(db).await?;
+
+    //     sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at")
+    //         .execute(db)
+    //         .await?;
+
+    //     sqlx::query!("INSERT INTO windmill_migrations (name) VALUES ('fix_job_completed_index') ON CONFLICT DO NOTHING")
+    //         .execute(&mut *tx)
+    //         .await?;
+    //     let _ = sqlx::query("SELECT pg_advisory_unlock(4242)")
+    //         .execute(&mut *tx)
+    //         .await?;
+    //     tx.commit().await?;
+    // }
+
+    run_windmill_migration!("fix_job_completed_index_2", &db, |tx| {
+        //     sqlx::query(
+        //     "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new_2 ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, created_at DESC)"
+        // ).execute(db).await?;
+
+        //     sqlx::query(
+        //     "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_started_at_new ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, started_at DESC)"
+        // ).execute(db).await?;
+
+        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at")
+            .execute(db)
+            .await?;
+
+        sqlx::query(
+            "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new",
+        )
+        .execute(db)
+        .await?;
+    });
+
+    run_windmill_migration!("fix_job_completed_index_3", &db, |tx| {
+        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_job_on_schedule_path")
+            .execute(db)
+            .await?;
+
+        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS concurrency_limit_stats_queue")
+            .execute(db)
+            .await?;
+
+        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS root_job_index")
+            .execute(db)
+            .await?;
+
+        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS index_completed_on_created")
+            .execute(db)
+            .await?;
+    });
+
+    run_windmill_migration!("fix_job_index_1_II", &db, |tx| {
+        let migration_job_name = "fix_job_index_1_II";
+        let mut i = 1;
+        tracing::info!("step {i} of {migration_job_name} migration");
+        sqlx::query!("create index concurrently  if not exists ix_job_workspace_id_created_at_new_3 ON v2_job  (workspace_id,  created_at DESC)")
+                .execute(db)
+                .await?;
+        i += 1;
+        tracing::info!("step {i} of {migration_job_name} migration");
+
+        sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_8 ON v2_job  (workspace_id, created_at DESC) where kind in ('deploymentcallback') AND parent_job IS NULL")
+            .execute(db)
+            .await?;
+        i += 1;
+        tracing::info!("step {i} of {migration_job_name} migration");
+
+        sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_9 ON v2_job  (workspace_id, created_at DESC) where kind in ('dependencies', 'flowdependencies', 'appdependencies') AND parent_job IS NULL")
+            .execute(db)
+            .await?;
+        i += 1;
+        tracing::info!("step {i} of {migration_job_name} migration");
+
+        sqlx::query!("create index concurrently if not exists ix_job_workspace_id_created_at_new_5 ON v2_job  (workspace_id, created_at DESC) where kind in ('preview', 'flowpreview') AND parent_job IS NULL")
+                .execute(db)
+                .await?;
+        i += 1;
+        tracing::info!("step {i} of {migration_job_name} migration");
+
+        sqlx::query!("create index concurrently if not exists ix_completed_job_workspace_id_started_at_new_2 ON v2_job_completed  (workspace_id, started_at DESC)")
+                .execute(db)
+                .await?;
+        i += 1;
+        tracing::info!("step {i} of {migration_job_name} migration");
+
+        sqlx::query!("create index concurrently if not exists ix_job_root_job_index_by_path_2 ON v2_job (workspace_id, runnable_path, created_at desc) WHERE parent_job IS NULL")
+                .execute(db)
+                .await?;
+
+        i += 1;
+        tracing::info!("step {i} of {migration_job_name} migration");
+
+        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path_2")
+            .execute(db)
+            .await?;
+
+        i += 1;
+        tracing::info!("step {i} of {migration_job_name} migration");
+
+        sqlx::query!("create index concurrently if not exists ix_job_created_at ON v2_job  (created_at DESC)")
+                    .execute(db)
+                    .await?;
+
+        i += 1;
+        tracing::info!("step {i} of {migration_job_name} migration");
+
+        sqlx::query(
+            "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new_2",
+        )
+        .execute(db)
+        .await?;
+        i += 1;
+        tracing::info!("step {i} of {migration_job_name} migration");
+
+        sqlx::query(
+            "DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_started_at_new",
+        )
+        .execute(db)
+        .await?;
+        i += 1;
+        tracing::info!("step {i} of {migration_job_name} migration");
+
+        sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path")
+            .execute(db)
+            .await?;
+    });
+
+    run_windmill_migration!("fix_labeled_jobs_index", &db, |tx| {
+        tracing::info!("Special migration to add index concurrently on job labels 2");
+        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS labeled_jobs_on_jobs")
+            .execute(db)
+            .await?;
+        sqlx::query!(
+        "CREATE INDEX CONCURRENTLY labeled_jobs_on_jobs ON v2_job_completed USING GIN ((result -> 'wm_labels')) WHERE result ? 'wm_labels'"
+        ).execute(db).await?;
+    });
+
+    run_windmill_migration!("v2_labeled_jobs_index", &db, |tx| {
+        tracing::info!("Special migration to add index concurrently on job labels");
+        sqlx::query!(
+            "CREATE INDEX CONCURRENTLY ix_v2_job_labels ON v2_job
+                USING GIN (labels)
+                WHERE labels IS NOT NULL"
+        )
+        .execute(db)
+        .await?;
+    });
+
+    run_windmill_migration!("v2_jobs_rls", &db, |tx| {
+        sqlx::query!("ALTER TABLE v2_job ENABLE ROW LEVEL SECURITY")
+            .execute(db)
+            .await?;
+    });
+
+    run_windmill_migration!("v2_improve_v2_job_indices_ii", &db, |tx| {
+        sqlx::query!("create index concurrently if not exists ix_v2_job_workspace_id_created_at ON v2_job  (workspace_id, created_at DESC) where kind in ('script', 'flow', 'singlescriptflow') AND parent_job IS NULL")
+                .execute(db)
+                .await?;
+
+        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_6")
+            .execute(db)
+            .await?;
+
+        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_7")
+            .execute(db)
+            .await?;
+    });
+
+    run_windmill_migration!("v2_improve_v2_queued_jobs_indices", &db, |tx| {
+        sqlx::query!("CREATE INDEX CONCURRENTLY IF NOT EXISTS queue_sort_v2 ON v2_job_queue (priority DESC NULLS LAST, scheduled_for, tag) WHERE running = false")
+            .execute(db)
+            .await?;
+
+        // sqlx::query!("CREATE INDEX CONCURRENTLY queue_sort_2_v2 ON v2_job_queue (tag, priority DESC NULLS LAST, scheduled_for) WHERE running = false")
+        //     .execute(db)
+        //     .await?;
+
+        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS queue_sort")
+            .execute(db)
+            .await?;
+
+        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS queue_sort_2")
+            .execute(db)
+            .await?;
+    });
+
+    run_windmill_migration!("audit_timestamps", db, |tx| {
+        sqlx::query!(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_audit_timestamps ON audit (timestamp DESC)"
+        )
+        .execute(db)
+        .await?;
+    });
+
+    run_windmill_migration!("job_completed_completed_at", db, |tx| {
+        sqlx::query!(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_completed_completed_at ON v2_job_completed (completed_at DESC)"
+        )
+        .execute(db)
+        .await?;
+    });
+
+    run_windmill_migration!("alerts_by_workspace", db, |tx| {
+        sqlx::query!(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS alerts_by_workspace ON alerts (workspace_id);"
+        )
+        .execute(db)
+        .await?;
+    });
+
+    run_windmill_migration!("remove_redundant_log_file_index", db, |tx| {
+        sqlx::query!("DROP INDEX CONCURRENTLY IF EXISTS log_file_hostname_log_ts_idx")
+            .execute(db)
+            .await?;
+    });
+
+    run_windmill_migration!("v2_job_queue_suspend", db, |tx| {
+        sqlx::query!(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS v2_job_queue_suspend ON v2_job_queue (workspace_id, suspend) WHERE suspend > 0;"
+        )
+        .execute(db)
+        .await?;
+    });
+
+    run_windmill_migration!("audit_recent_login_activities", db, |tx| {
+        sqlx::query!(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_audit_recent_login_activities 
+ON audit (timestamp, username) 
+WHERE operation IN ('users.login', 'oauth.login', 'users.token.refresh');"
+        )
+        .execute(db)
+        .await?;
+    });
+
+    run_windmill_migration!("v2_script_lock_index", db, |tx| {
+        sqlx::query!(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS script_not_archived ON script (workspace_id, path, created_at DESC) where archived = false;"
+        )
+        .execute(db)
+        .await?;
+    });
+    Ok(())
+}

backend/windmill-api/src/scripts.rs

@@ -33,11 +33,7 @@ use serde_json::json;
 use serde_json::value::RawValue;
 use sql_builder::prelude::*;
 use sqlx::{FromRow, Postgres, Transaction};
-use std::{
-    collections::{hash_map::DefaultHasher, HashMap},
-    hash::{Hash, Hasher},
-    sync::Arc,
-};
+use std::{collections::HashMap, sync::Arc};
 use windmill_audit::audit_oss::audit_log;
 use windmill_audit::ActionKind;
 use windmill_worker::process_relative_imports;
@@ -45,6 +41,8 @@ use windmill_worker::process_relative_imports;
 use windmill_common::{
     assets::{clear_asset_usage, insert_asset_usage, AssetUsageKind, AssetWithAltAccessType},
     error::to_anyhow,
+    scripts::hash_script,
+    utils::WarnAfterExt,
     worker::CLOUD_HOSTED,
 };
 
@@ -378,12 +376,6 @@ async fn get_top_hub_scripts(
     Ok::<_, Error>((status_code, headers, response))
 }
 
-fn hash_script(ns: &NewScript) -> i64 {
-    let mut dh = DefaultHasher::new();
-    ns.hash(&mut dh);
-    dh.finish() as i64
-}
-
 async fn create_snapshot_script(
     authed: ApiAuthed,
     Extension(user_db): Extension<UserDB>,
@@ -1447,6 +1439,7 @@ async fn raw_script_by_path_internal(
         w_id
     )
     .fetch_optional(&mut *tx)
+    .warn_after_seconds(5)
     .await?;
     tx.commit().await?;
 
@@ -1457,6 +1450,7 @@ async fn raw_script_by_path_internal(
             w_id
         )
         .fetch_one(&db)
+        .warn_after_seconds(5)
         .await?
         .unwrap_or(false);
 

backend/windmill-api/src/settings.rs

@@ -575,7 +575,7 @@ async fn databases_exist(
     Json(database_names): Json<Vec<String>>,
 ) -> JsonResult<Vec<String>> {
     let result = sqlx::query_scalar!(
-        r#"SELECT elem FROM (SELECT unnest($1::TEXT[]) AS elem)
+        r#"SELECT elem FROM (SELECT unnest($1::TEXT[]) AS elem) AS e
         WHERE elem NOT IN (SELECT datname FROM pg_catalog.pg_database);"#,
         database_names.as_slice()
     )

backend/windmill-api/src/users.rs

@@ -1327,8 +1327,9 @@ async fn update_workspace_user(
         eu.operator,
         eu.disabled,
         &mut tx,
-        Some(&authed)
-    ).await?;
+        Some(&authed),
+    )
+    .await?;
 
     let user_email = sqlx::query_scalar!(
         "SELECT email FROM usr WHERE username = $1 AND workspace_id = $2",
@@ -1589,7 +1590,14 @@ async fn delete_workspace_user(
 
     let email_to_delete = not_found_if_none(email_to_delete_o, "User", &username_to_delete)?;
 
-    delete_workspace_user_internal(&w_id, &username_to_delete, &email_to_delete, &mut tx, Some(&authed)).await?;
+    delete_workspace_user_internal(
+        &w_id,
+        &username_to_delete,
+        &email_to_delete,
+        &mut tx,
+        Some(&authed),
+    )
+    .await?;
     tx.commit().await?;
 
     handle_deployment_metadata(
@@ -2172,7 +2180,9 @@ async fn get_all_runnables(
                 .collect::<Vec<_>>(),
         );
         let scripts = sqlx::query!(
-        "SELECT workspace_id as workspace, path, summary, description, schema FROM script as o WHERE created_at = (select max(created_at) from script where o.path = path and workspace_id = $1) and workspace_id = $1", workspace
+        "SELECT workspace_id as workspace, path, summary, description, schema FROM script as o 
+         WHERE created_at = (select max(created_at) from script where o.path = path and workspace_id = $1 AND archived = false) 
+         AND workspace_id = $1 and archived = false", workspace
     )
     .fetch_all(&mut *tx)
     .await?;

backend/windmill-api/src/variables.rs

@@ -24,7 +24,10 @@ use serde_json::Value;
 use windmill_audit::audit_oss::{audit_log, AuditAuthorable};
 use windmill_audit::ActionKind;
 use windmill_common::{
-    db::UserDB, error::{Error, JsonResult, Result}, utils::{not_found_if_none, paginate, Pagination, StripPath}, variables::{
+    db::UserDB,
+    error::{Error, JsonResult, Result},
+    utils::{not_found_if_none, paginate, Pagination, StripPath, WarnAfterExt},
+    variables::{
         build_crypt, get_reserved_variables, ContextualVariable, CreateVariable, ListableVariable,
     },
     worker::CLOUD_HOSTED,
@@ -693,6 +696,7 @@ pub async fn get_value_internal<'c>(
         LEFT JOIN account ON variable.account = account.id WHERE variable.path = $1 AND variable.workspace_id = $2", path, w_id
     )
     .fetch_optional(&mut *tx)
+    .warn_after_seconds(5)
     .await?;
 
     let variable = if let Some(variable) = variable_o {

backend/windmill-api/src/workspaces.rs

@@ -415,7 +415,9 @@ async fn is_premium(
 ) -> JsonResult<bool> {
     require_admin(authed.is_admin, &authed.username)?;
     #[cfg(feature = "cloud")]
-    let premium = windmill_common::workspaces::is_premium_workspace(&_db, &_w_id).await;
+    let premium = windmill_common::workspaces::get_team_plan_status(&_db, &_w_id)
+        .await
+        .premium;
     #[cfg(not(feature = "cloud"))]
     let premium = false;
     Ok(Json(premium))

backend/windmill-common/src/cache.rs

@@ -169,6 +169,7 @@ impl<Key: Eq + Hash + Item + Clone, Val: Export, Root: AsRef<Path>> FsBackedCach
                     ),
                 }
             }
+
             // Cache path doesn't exist or import failed, generate the content.
             let data = Val::resolve(with.await?)?;
             // Try to export data to the file-system.

backend/windmill-common/src/lib.rs

@@ -631,6 +631,7 @@ pub async fn get_latest_hash_for_path<'c, E: sqlx::PgExecutor<'c>>(
     db: E,
     w_id: &str,
     script_path: &str,
+    require_locked: bool,
 ) -> error::Result<(
     scripts::ScriptHash,
     Option<Tag>,
@@ -646,11 +647,12 @@ pub async fn get_latest_hash_for_path<'c, E: sqlx::PgExecutor<'c>>(
     String,
 )> {
     let r_o = sqlx::query!(
-        "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script where path = $1 AND workspace_id = $2 AND
-    created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND
-    deleted = false AND archived = false)",
+        "select hash, tag, concurrency_key, concurrent_limit, concurrency_time_window_s, cache_ttl, language as \"language: ScriptLang\", dedicated_worker, priority, timeout, on_behalf_of_email, created_by FROM script
+         WHERE path = $1 AND workspace_id = $2 AND archived = false AND (lock IS NOT NULL OR $3 = false)
+         ORDER BY created_at DESC LIMIT 1",
         script_path,
-        w_id
+        w_id,
+        require_locked
     )
     .fetch_optional(db)
     .await?;

backend/windmill-common/src/scripts.rs

@@ -649,3 +649,9 @@ pub struct HubScript {
     pub schema: Box<serde_json::value::RawValue>,
     pub summary: Option<String>,
 }
+
+pub fn hash_script(ns: &NewScript) -> i64 {
+    let mut dh = std::hash::DefaultHasher::new();
+    ns.hash(&mut dh);
+    dh.finish() as i64
+}

backend/windmill-common/src/utils.rs

@@ -265,14 +265,18 @@ pub fn create_directory_sync(directory_path: &str) {
         .expect("could not create dir");
 }
 
+#[track_caller]
 pub fn not_found_if_none<T, U: AsRef<str>>(opt: Option<T>, kind: &str, name: U) -> Result<T> {
     if let Some(o) = opt {
         Ok(o)
     } else {
+        let loc = Location::caller();
         Err(Error::NotFound(format!(
-            "{} not found at name {}",
+            "{} not found at name {} ({}:{})",
             kind,
-            name.as_ref()
+            name.as_ref(),
+            loc.file().split("/").last().unwrap_or_default(),
+            loc.line()
         )))
     }
 }

backend/windmill-common/src/variables.rs

@@ -7,6 +7,7 @@
  */
 
 use crate::error;
+use crate::utils::WarnAfterExt;
 use crate::worker::Connection;
 use crate::{worker::WORKER_GROUP, BASE_URL, DB};
 use chrono::{SecondsFormat, Utc};
@@ -106,6 +107,7 @@ pub async fn get_workspace_key(w_id: &str, db: &DB) -> crate::error::Result<Stri
         w_id
     )
     .fetch_one(db)
+    .warn_after_seconds(5)
     .await
     .map_err(|e| crate::Error::internal_err(format!("fetching workspace key: {e:#}")))?;
     Ok(key)

backend/windmill-common/src/workspaces.rs

@@ -86,22 +86,47 @@ impl Default for GitSyncSettings {
     }
 }
 
+#[derive(Clone)]
+pub struct TeamPlanStatus {
+    pub premium: bool,
+    pub is_past_due: bool,
+    pub max_tolerated_executions: Option<i32>,
+}
+
 lazy_static::lazy_static! {
-    pub static ref IS_PREMIUM_CACHE: Cache<String, bool> = Cache::new(5000);
+    pub static ref TEAM_PLAN_CACHE: Cache<String, TeamPlanStatus> = Cache::new(5000);
 }
 
 #[cfg(feature = "cloud")]
-pub async fn is_premium_workspace(_db: &crate::DB, _w_id: &str) -> bool {
-    let cached = IS_PREMIUM_CACHE.get(_w_id);
+pub async fn get_team_plan_status(_db: &crate::DB, _w_id: &str) -> TeamPlanStatus {
+    let cached = TEAM_PLAN_CACHE.get(_w_id);
     if let Some(cached) = cached {
         return cached;
     }
-    let premium = sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", _w_id)
-        .fetch_one(_db)
-        .await
-        .unwrap_or(false);
-    IS_PREMIUM_CACHE.insert(_w_id.to_string(), premium);
-    premium
+    let team_plan_info = sqlx::query_as!(
+        TeamPlanStatus,
+        r#"
+            SELECT
+                w.premium,
+                COALESCE(cw.is_past_due, false) as "is_past_due!",
+                cw.max_tolerated_executions
+            FROM
+                workspace w
+                LEFT JOIN cloud_workspace_settings cw ON cw.workspace_id = w.id
+            WHERE
+                w.id = $1
+        "#,
+        _w_id
+    )
+    .fetch_one(_db)
+    .await
+    .unwrap_or_else(|_| TeamPlanStatus {
+        premium: false,
+        is_past_due: false,
+        max_tolerated_executions: None,
+    });
+    TEAM_PLAN_CACHE.insert(_w_id.to_string(), team_plan_info.clone());
+    team_plan_info
 }
 
 #[derive(Deserialize, Serialize, Debug)]

backend/windmill-queue/src/jobs.rs

@@ -1419,8 +1419,9 @@ fn apply_completed_job_cloud_usage(
         let email2 = email.clone();
         tokio::task::spawn(async move {
             let additional_usage = _duration / 1000;
-            let premium_workspace =
-                windmill_common::workspaces::is_premium_workspace(&db, &w_id).await;
+            let premium_workspace = windmill_common::workspaces::get_team_plan_status(&db, &w_id)
+                .await
+                .premium;
             tokio::time::timeout(std::time::Duration::from_secs(10), async move {
                 let _ = sqlx::query!(
                     "INSERT INTO usage (id, is_workspace, month_, usage) 
@@ -3636,8 +3637,8 @@ pub async fn push<'c, 'd>(
 ) -> Result<(Uuid, Transaction<'c, Postgres>), Error> {
     #[cfg(feature = "cloud")]
     if *CLOUD_HOSTED {
-        let premium_workspace =
-            windmill_common::workspaces::is_premium_workspace(_db, workspace_id).await;
+        let team_plan_status =
+            windmill_common::workspaces::get_team_plan_status(_db, workspace_id).await;
         // we track only non flow steps
         let (workspace_usage, user_usage) = if !matches!(
             job_payload,
@@ -3655,7 +3656,7 @@ pub async fn push<'c, 'd>(
                     .await
                     .map_err(|e| Error::internal_err(format!("updating usage: {e:#}")))?;
 
-                let user_usage = if !premium_workspace {
+                let user_usage = if !team_plan_status.premium {
                     Some(sqlx::query_scalar!(
                         "INSERT INTO usage (id, is_workspace, month_, usage)
                         VALUES ($1, FALSE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), 1)
@@ -3678,7 +3679,7 @@ pub async fn push<'c, 'd>(
             Ok((None, None))
         }?;
 
-        if !premium_workspace {
+        if !team_plan_status.premium || team_plan_status.is_past_due {
             let is_super_admin =
                 sqlx::query_scalar!("SELECT super_admin FROM password WHERE email = $1", email)
                     .fetch_optional(_db)
@@ -3686,7 +3687,8 @@ pub async fn push<'c, 'd>(
                     .unwrap_or(false);
 
             if !is_super_admin {
-                if email != ERROR_HANDLER_USER_EMAIL
+                if !team_plan_status.premium
+                    && email != ERROR_HANDLER_USER_EMAIL
                     && email != SCHEDULE_ERROR_HANDLER_USER_EMAIL
                     && email != SCHEDULE_RECOVERY_HANDLER_USER_EMAIL
                     && email != "worker@windmill.dev"
@@ -3765,43 +3767,53 @@ pub async fn push<'c, 'd>(
                         .flatten()
                         .unwrap_or(1)
                     };
+                    if team_plan_status.premium {
+                        // team plan is premium but past due, we check if the workspace has exceeded the max tolerated executions
+                        if team_plan_status.max_tolerated_executions.is_none()
+                            || workspace_usage > team_plan_status.max_tolerated_executions.unwrap()
+                        {
+                            return Err(error::Error::QuotaExceeded(format!(
+                                "Workspace {workspace_id} team plan is past due and isn't allowed to run any more jobs. Please fix your payment method in the workspace settings."
+                            )));
+                        }
+                    } else {
+                        if workspace_usage > MAX_FREE_EXECS
+                            && !matches!(job_payload, JobPayload::Dependencies { .. })
+                            && !matches!(job_payload, JobPayload::FlowDependencies { .. })
+                            && !matches!(job_payload, JobPayload::AppDependencies { .. })
+                        {
+                            return Err(error::Error::QuotaExceeded(format!(
+                                "Workspace {workspace_id} has exceeded the free usage limit of {MAX_FREE_EXECS} that applies outside of premium workspaces."
+                            )));
+                        }
 
-                    if workspace_usage > MAX_FREE_EXECS
-                        && !matches!(job_payload, JobPayload::Dependencies { .. })
-                        && !matches!(job_payload, JobPayload::FlowDependencies { .. })
-                        && !matches!(job_payload, JobPayload::AppDependencies { .. })
-                    {
-                        return Err(error::Error::QuotaExceeded(format!(
-                            "Workspace {workspace_id} has exceeded the free usage limit of {MAX_FREE_EXECS} that applies outside of premium workspaces."
-                        )));
-                    }
+                        let in_queue_workspace = sqlx::query_scalar!(
+                            "SELECT COUNT(id) FROM v2_job_queue WHERE workspace_id = $1",
+                            workspace_id
+                        )
+                        .fetch_one(_db)
+                        .await?
+                        .unwrap_or(0);
 
-                    let in_queue_workspace = sqlx::query_scalar!(
-                        "SELECT COUNT(id) FROM v2_job_queue WHERE workspace_id = $1",
-                        workspace_id
-                    )
-                    .fetch_one(_db)
-                    .await?
-                    .unwrap_or(0);
+                        if in_queue_workspace > MAX_FREE_EXECS as i64 {
+                            return Err(error::Error::QuotaExceeded(format!(
+                                "Workspace {workspace_id} has exceeded the jobs in queue limit of {MAX_FREE_EXECS} that applies outside of premium workspaces."
+                            )));
+                        }
 
-                    if in_queue_workspace > MAX_FREE_EXECS as i64 {
-                        return Err(error::Error::QuotaExceeded(format!(
-                            "Workspace {workspace_id} has exceeded the jobs in queue limit of {MAX_FREE_EXECS} that applies outside of premium workspaces."
-                        )));
-                    }
-
-                    let concurrent_runs_workspace = sqlx::query_scalar!(
+                        let concurrent_runs_workspace = sqlx::query_scalar!(
                         "SELECT COUNT(id) FROM v2_job_queue WHERE running = true AND workspace_id = $1",
-                        workspace_id
-                    )
-                    .fetch_one(_db)
-                    .await?
-                    .unwrap_or(0);
+                            workspace_id
+                        )
+                        .fetch_one(_db)
+                        .await?
+                        .unwrap_or(0);
 
-                    if concurrent_runs_workspace > MAX_FREE_CONCURRENT_RUNS as i64 {
-                        return Err(error::Error::QuotaExceeded(format!(
-                            "Workspace {workspace_id} has exceeded the concurrent runs limit of {MAX_FREE_CONCURRENT_RUNS} that applies outside of premium workspaces."
-                        )));
+                        if concurrent_runs_workspace > MAX_FREE_CONCURRENT_RUNS as i64 {
+                            return Err(error::Error::QuotaExceeded(format!(
+                                "Workspace {workspace_id} has exceeded the concurrent runs limit of {MAX_FREE_CONCURRENT_RUNS} that applies outside of premium workspaces."
+                            )));
+                        }
                     }
                 }
             }

backend/windmill-queue/src/schedule.rs

@@ -156,6 +156,7 @@ pub async fn push_scheduled_job<'c>(
             &mut *tx,
             &schedule.workspace_id,
             &schedule.script_path,
+            false,
         )
         .await?;
 

backend/windmill-worker/src/ai_executor.rs

@@ -118,7 +118,7 @@ struct Tool {
 #[derive(Deserialize, Debug)]
 struct AIAgentArgs {
     provider: Provider,
-    system_prompt: String,
+    system_prompt: Option<String>,
     user_message: String,
     temperature: Option<f32>,
     max_completion_tokens: Option<u32>,
@@ -608,18 +608,21 @@ async fn run_agent(
     hostname: &str,
     killpill_rx: &mut tokio::sync::broadcast::Receiver<()>,
 ) -> error::Result<Box<RawValue>> {
-    let mut messages = vec![
-        OpenAIMessage {
+    let mut messages = if let Some(system_prompt) = args.system_prompt.filter(|s| !s.is_empty()) {
+        vec![OpenAIMessage {
             role: "system".to_string(),
-            content: Some(args.system_prompt),
+            content: Some(system_prompt),
             ..Default::default()
-        },
-        OpenAIMessage {
-            role: "user".to_string(),
-            content: Some(args.user_message),
-            ..Default::default()
-        },
-    ];
+        }]
+    } else {
+        vec![]
+    };
+
+    messages.push(OpenAIMessage {
+        role: "user".to_string(),
+        content: Some(args.user_message),
+        ..Default::default()
+    });
 
     let mut actions = vec![];
 
@@ -923,7 +926,7 @@ pub async fn handle_ai_agent_job(
                             .await?;
                             Ok(Some(hub_script.schema))
                         } else {
-                            let hash = get_latest_hash_for_path(db, &job.workspace_id, path)
+                            let hash = get_latest_hash_for_path(db, &job.workspace_id, path, true)
                                 .await?
                                 .0;
                             // update module definition to use a fixed hash so all tool calls match the same schema

backend/windmill-worker/src/bun_executor.rs

@@ -758,7 +758,7 @@ pub async fn prebundle_bun_script(
 pub const BUN_BUNDLE_OBJECT_STORE_PREFIX: &str = "bun_bundle/";
 
 async fn get_script_import_updated_at(db: &DB, w_id: &str, script_path: &str) -> Result<String> {
-    let script_hash = get_latest_hash_for_path(db, w_id, script_path).await?;
+    let script_hash = get_latest_hash_for_path(db, w_id, script_path, false).await?;
     let last_updated_at = sqlx::query_scalar!(
         "SELECT created_at FROM script WHERE workspace_id = $1 AND hash = $2",
         w_id,

backend/windmill-worker/src/common.rs

@@ -680,11 +680,12 @@ pub async fn resolve_job_timeout(
     let mut warn_msg: Option<String> = None;
     #[cfg(feature = "cloud")]
     let cloud_premium_workspace = *CLOUD_HOSTED
-        && windmill_common::workspaces::is_premium_workspace(
+        && windmill_common::workspaces::get_team_plan_status(
             _conn.as_sql().expect("cloud cannot use http connection"),
             _w_id,
         )
-        .await;
+        .await
+        .premium;
     #[cfg(not(feature = "cloud"))]
     let cloud_premium_workspace = false;
 

backend/windmill-worker/src/dedicated_worker.rs

@@ -622,8 +622,7 @@ async fn spawn_dedicated_worker(
                 } else {
                     sqlx::query_as::<_, (String, Option<String>, Option<ScriptLang>, Option<Vec<String>>, bool, Option<ScriptHash>)>(
                         "SELECT content, lock, language, envs, codebase IS NOT NULL, hash FROM script WHERE path = $1 AND workspace_id = $2 AND
-                            created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND
-                            deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)",
+                            archived = false AND lock IS not NULL AND lock_error_logs IS NULL ORDER BY created_at DESC LIMIT 1",
                     )
                     .bind(&path)
                     .bind(&w_id)

backend/windmill-worker/src/go_executor.rs

@@ -571,6 +571,8 @@ pub async fn install_go_dependencies(
     child_cmd
         .current_dir(job_dir)
         .env_clear()
+        .env("HOME", HOME_ENV.as_str())
+        .env("PATH", PATH_ENV.as_str())
         .env("GOPATH", {
             #[cfg(unix)]
             {
@@ -584,6 +586,18 @@ pub async fn install_go_dependencies(
         .args(vec!["mod", mod_command])
         .stdout(Stdio::piped())
         .stderr(Stdio::piped());
+
+    if let Some(ref goprivate) = *GOPRIVATE {
+        child_cmd.env("GOPRIVATE", goprivate);
+    }
+
+    // TODO: Remove if no incidents reported
+    if !std::env::var("WMDEBUG_NO_GOPROXY_ON_TIDY").ok().is_some() {
+        if let Some(ref goproxy) = *GOPROXY {
+            child_cmd.env("GOPROXY", goproxy);
+        }
+    }
+
     // If annotation used we want to call tidy with special flag to pin go to 1.22
     // The reason for this that at some point we had to jump from go 1.22 to 1.25 and this addds backward compatibility.
     if anns.go1_22_compat && mod_command == "tidy" {

backend/windmill-worker/src/lib.rs

@@ -63,7 +63,9 @@ mod worker_lockfiles;
 mod worker_utils;
 
 pub use worker::*;
-pub use worker_lockfiles::process_relative_imports;
+pub use worker_lockfiles::{
+    process_relative_imports, trigger_dependents_to_recompute_dependencies,
+};
 
 pub use result_processor::handle_job_error;
 

backend/windmill-worker/src/mssql_executor.rs

@@ -5,7 +5,9 @@ use regex::Regex;
 use serde::Deserialize;
 use serde_json::value::RawValue;
 use serde_json::{Map, Value};
-use tiberius::{AuthMethod, Client, ColumnData, Config, FromSqlOwned, Query, Row, SqlBrowser};
+use tiberius::{
+    AuthMethod, Client, ColumnData, Config, EncryptionLevel, FromSqlOwned, Query, Row, SqlBrowser,
+};
 use tokio::net::TcpStream;
 use tokio_util::compat::TokioAsyncWriteCompatExt;
 use uuid::Uuid;
@@ -39,6 +41,7 @@ struct MssqlDatabase {
     trust_cert: Option<bool>,
     #[serde(default, deserialize_with = "empty_as_none")]
     ca_cert: Option<String>,
+    encrypt: Option<bool>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -146,6 +149,12 @@ pub async fn do_mssql(
         tracing::info!("MSSQL: using provided CA certificate for trust");
     }
 
+    config.encryption(if database.encrypt.unwrap_or(true) {
+        EncryptionLevel::Required
+    } else {
+        EncryptionLevel::NotSupported
+    });
+
     let tcp = if use_instance_name {
         TcpStream::connect_named(&config).await.map_err(to_anyhow)? // named instance
     } else {

backend/windmill-worker/src/worker_lockfiles.rs

@@ -17,7 +17,7 @@ use windmill_common::error::Result;
 use windmill_common::flows::{FlowModule, FlowModuleValue, FlowNodeId};
 use windmill_common::get_latest_deployed_hash_for_path;
 use windmill_common::jobs::JobPayload;
-use windmill_common::scripts::ScriptHash;
+use windmill_common::scripts::{hash_script, NewScript, ScriptHash};
 #[cfg(feature = "python")]
 use windmill_common::worker::PythonAnnotations;
 use windmill_common::worker::{to_raw_value, to_raw_value_owned, write_file, Connection};
@@ -38,6 +38,11 @@ use windmill_parser_py_imports::parse_relative_imports;
 use windmill_parser_ts::parse_expr_for_imports;
 use windmill_queue::{append_logs, CanceledBy, MiniPulledJob, PushIsolationLevel};
 
+lazy_static::lazy_static! {
+    // TODO: To be removed in future versions
+    static ref WMDEBUG_NO_HASH_CHANGE_ON_DJ: bool = std::env::var("WMDEBUG_NO_HASH_CHANGE_ON_DJ").is_ok();
+}
+
 use crate::common::OccupancyMetrics;
 use crate::csharp_executor::generate_nuget_lockfile;
 
@@ -251,6 +256,7 @@ pub async fn handle_dependency_job(
                 .is_some_and(|y| y.to_string().as_str() == "true")
         })
         .unwrap_or(false);
+
     let npm_mode = if job
         .script_lang
         .as_ref()
@@ -338,30 +344,140 @@ pub async fn handle_dependency_job(
                 ));
             }
 
-            let hash = job.runnable_id.unwrap_or(ScriptHash(0));
+            let current_hash = job.runnable_id.unwrap_or(ScriptHash(0));
             let w_id = &job.workspace_id;
-            sqlx::query!(
-                "UPDATE script SET lock = $1 WHERE hash = $2 AND workspace_id = $3",
-                &content,
-                &hash.0,
-                w_id
-            )
-            .execute(db)
-            .await?;
-
-            // `lock` has been updated; invalidate the cache.
-            cache::script::invalidate(hash);
-
             let (deployment_message, parent_path) =
                 get_deployment_msg_and_parent_path_from_args(job.args.clone());
 
+            let script_info = sqlx::query_as::<_, windmill_common::scripts::Script>(
+                "SELECT * FROM script WHERE hash = $1 AND workspace_id = $2",
+            )
+            .bind(&current_hash.0)
+            .bind(w_id)
+            .fetch_one(db)
+            .await?;
+
+            // DependencyJob can be triggered only from 2 places:
+            // 1. create_script function in windmill-api/src/scripts.rs
+            // 2. trigger_dependents_to_recompute_dependencies (in this file)
+            //
+            // First will **always** produce script with null in `lock`
+            // where Second will **always** do with lock being not null
+            let deployed_hash = if script_info.lock.is_some() && !*WMDEBUG_NO_HASH_CHANGE_ON_DJ {
+                let mut tx = db.begin().await?;
+                // This entire section exists to solve following problem:
+                //
+                // 2 workers, one script that depend on another in python
+                // run the original script on both workers
+                // you update the dependenecy of a relative import,
+                // run it again until you ran it on both, normally it should fail on one of those
+                //
+                // It happens because every worker has cached their own script versions.
+                // However usual dependency job does not update hash of the script (and cache is keyed by the hash).
+                // This logical branch will create new script which will update the hash and automatically invalidate cache.
+                //
+                // IMPORTANT: This will **only** be triggered by another DependencyJob. It will never be triggered by script (re)deployement
+
+                let ns = NewScript {
+                    path: script_info.path,
+                    parent_hash: Some(current_hash),
+                    summary: script_info.summary,
+                    description: script_info.description,
+                    content: script_info.content,
+                    schema: script_info.schema,
+                    is_template: Some(script_info.is_template),
+                    // TODO: Make it either None everywhere (particularely when raw reqs are calculated)
+                    // Or handle this case and conditionally make Some (only with raw reqs)
+                    lock: None,
+                    language: script_info.language,
+                    kind: Some(script_info.kind),
+                    tag: script_info.tag,
+                    draft_only: script_info.draft_only,
+                    envs: script_info.envs,
+                    concurrent_limit: script_info.concurrent_limit,
+                    concurrency_time_window_s: script_info.concurrency_time_window_s,
+                    cache_ttl: script_info.cache_ttl,
+                    dedicated_worker: script_info.dedicated_worker,
+                    ws_error_handler_muted: script_info.ws_error_handler_muted,
+                    priority: script_info.priority,
+                    timeout: script_info.timeout,
+                    delete_after_use: script_info.delete_after_use,
+                    restart_unless_cancelled: script_info.restart_unless_cancelled,
+                    deployment_message: deployment_message.clone(),
+                    concurrency_key: script_info.concurrency_key,
+                    visible_to_runner_only: script_info.visible_to_runner_only,
+                    no_main_func: script_info.no_main_func,
+                    codebase: script_info.codebase,
+                    has_preprocessor: script_info.has_preprocessor,
+                    on_behalf_of_email: script_info.on_behalf_of_email,
+                    assets: script_info.assets,
+                };
+
+                let new_hash = hash_script(&ns);
+
+                sqlx::query!("
+    INSERT INTO script
+    (workspace_id, hash, path, parent_hashes, summary, description, content, \
+    created_by, schema, is_template, extra_perms, lock, language, kind, tag, \
+    draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \
+    dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, \
+    delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, \
+    codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets) 
+
+    SELECT  workspace_id, $1, path, array_prepend($2::bigint, COALESCE(parent_hashes, '{}'::bigint[])), summary, description, \
+            content, created_by, schema, is_template, extra_perms, $4, language, kind, tag, \
+            draft_only, envs, concurrent_limit, concurrency_time_window_s, cache_ttl, \
+            dedicated_worker, ws_error_handler_muted, priority, restart_unless_cancelled, \
+            delete_after_use, timeout, concurrency_key, visible_to_runner_only, no_main_func, \
+            codebase, has_preprocessor, on_behalf_of_email, schema_validation, assets 
+
+    FROM script WHERE hash = $2 AND workspace_id = $3;
+            ",
+                new_hash, current_hash.0, w_id, &content).execute(db).await?;
+
+                // Archive current
+                sqlx::query!(
+                    "UPDATE script SET archived = true WHERE hash = $1 AND workspace_id = $2",
+                    current_hash.0,
+                    w_id
+                )
+                .execute(&mut *tx)
+                .await?;
+
+                tx.commit().await?;
+
+                ScriptHash(new_hash)
+            } else {
+                // We do not create new row for this update
+                // That means we can keep current hash and just update lock
+                sqlx::query!(
+                    "UPDATE script SET lock = $1 WHERE hash = $2 AND workspace_id = $3",
+                    &content,
+                    &current_hash.0,
+                    w_id
+                )
+                .execute(db)
+                .await?;
+
+                // `lock` has been updated; invalidate the cache.
+                // Since only worker that ran this Dependency Job has the cache
+                // we do not need to think about invalidating cache for other workers.
+                cache::script::invalidate(current_hash);
+
+                if *WMDEBUG_NO_HASH_CHANGE_ON_DJ {
+                    tracing::warn!("WMDEBUG_NO_HASH_CHANGE_ON_DJ usually should not be used. Behavior might be unstable. Please contact Windmill Team for support.")
+                }
+
+                current_hash
+            };
+
             if let Err(e) = handle_deployment_metadata(
                 &job.permissioned_as_email,
                 &job.created_by,
                 &db,
                 &w_id,
                 DeployedObject::Script {
-                    hash,
+                    hash: deployed_hash,
                     path: script_path.to_string(),
                     parent_path: parent_path.clone(),
                 },
@@ -502,7 +618,7 @@ pub async fn process_relative_imports(
     Ok(())
 }
 
-async fn trigger_dependents_to_recompute_dependencies(
+pub async fn trigger_dependents_to_recompute_dependencies(
     w_id: &str,
     script_path: &str,
     deployment_message: Option<String>,
@@ -535,6 +651,10 @@ async fn trigger_dependents_to_recompute_dependencies(
             args.insert("deployment_message".to_string(), to_raw_value(&dm));
         }
         if let Some(ref p_path) = parent_path {
+            // NOTE:
+            // it's not used but maybe one day it will be useful. allows more back-compatibility for the workers when we need it
+            // also very useful for debugging/observability
+            // it adds that information to the job args so you can see from the runs page
             args.insert("common_dependency_path".to_string(), to_raw_value(&p_path));
         }
 
@@ -542,6 +662,7 @@ async fn trigger_dependents_to_recompute_dependencies(
             "already_visited".to_string(),
             to_raw_value(&already_visited),
         );
+
         let kind = s.importer_kind.clone().unwrap_or_default();
         let job_payload = if kind == "script" {
             let r = get_latest_deployed_hash_for_path(db, w_id, s.importer_path.as_str()).await;
@@ -577,7 +698,7 @@ async fn trigger_dependents_to_recompute_dependencies(
                 Ok(Some(version)) => JobPayload::FlowDependencies {
                     path: s.importer_path.clone(),
                     dedicated_worker: None,
-                    version: version,
+                    version,
                 },
                 Ok(None) => {
                     tracing::error!(

backend/windmill-worker/src/worker_utils.rs

@@ -308,7 +308,7 @@ pub(crate) async fn queue_vacuum(conn: &Connection, worker_name: &str, hostname:
             tokio::task::spawn(
                 (async move {
                     tracing::info!(worker = %worker_name, hostname = %hostname, "vacuuming queue");
-                    if let Err(e) = sqlx::query!("VACUUM v2_job_queue, v2_job_runtime, v2_job_status")
+                    if let Err(e) = sqlx::query!("VACUUM v2_job_queue, v2_job_runtime, v2_job_status, job_perms")
                         .execute(&db2)
                         .await
                     {

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.530.0";
+export const VERSION = "v1.537.0";
 
 export async function login(email: string, password: string): Promise<string> {
   return await windmill.UserService.login({

cli/src/commands/gitsync-settings/pull.ts

@@ -3,11 +3,11 @@ import { GlobalOptions } from "../../types.ts";
 import { requireLogin } from "../../core/auth.ts";
 import { resolveWorkspace } from "../../core/context.ts";
 import * as wmill from "../../../gen/services.gen.ts";
-import { SyncOptions, readConfigFile, getEffectiveSettings, DEFAULT_SYNC_OPTIONS } from "../../core/conf.ts";
+import { SyncOptions, readConfigFile, getEffectiveSettings, DEFAULT_SYNC_OPTIONS, getWmillYamlPath } from "../../core/conf.ts";
 import { deepEqual } from "../../utils/utils.ts";
 import { getCurrentGitBranch, isGitRepository } from "../../utils/git.ts";
 
-import { GitSyncRepository, WriteMode } from "./types.ts";
+import { WriteMode } from "./types.ts";
 import { GitSyncSettingsConverter } from "./converter.ts";
 import { handleLegacyRepositoryMigration } from "./legacySettings.ts";
 import {
@@ -132,11 +132,9 @@ export async function pullGitSyncSettings(
     const backendSyncOptions: SyncOptions = GitSyncSettingsConverter.fromBackendFormat(selectedRepo.settings);
 
     // Check if wmill.yaml exists - create a default one if it doesn't exist
-    let wmillYamlExists = true;
-    try {
-      await Deno.stat("wmill.yaml");
-    } catch (error) {
-      wmillYamlExists = false;
+    const wmillYamlPath = getWmillYamlPath();
+    const wmillYamlExists = wmillYamlPath !== null;
+    if (!wmillYamlExists) {
       if (!opts.jsonOutput) {
         log.info(
           colors.yellow(
@@ -165,11 +163,11 @@ export async function pullGitSyncSettings(
       if (isGitRepository()) {
         const currentBranch = getCurrentGitBranch();
         if (currentBranch) {
-          if (!updatedConfig.git_branches) {
-            updatedConfig.git_branches = {};
+          if (!updatedConfig.gitBranches) {
+            updatedConfig.gitBranches = {};
           }
-          if (!updatedConfig.git_branches[currentBranch]) {
-            updatedConfig.git_branches[currentBranch] = { overrides: {} };
+          if (!updatedConfig.gitBranches[currentBranch]) {
+            updatedConfig.gitBranches[currentBranch] = { overrides: {} };
           }
         }
       }
@@ -358,16 +356,16 @@ export async function pullGitSyncSettings(
       let needsBranchStructure = false;
       if (isGitRepository()) {
         const currentBranch = getCurrentGitBranch();
-        if (currentBranch && (!localConfig.git_branches || !localConfig.git_branches[currentBranch])) {
+        if (currentBranch && (!localConfig.gitBranches || !localConfig.gitBranches[currentBranch])) {
           needsBranchStructure = true;
 
           // Create empty branch structure
           const updatedConfig = { ...localConfig };
-          if (!updatedConfig.git_branches) {
-            updatedConfig.git_branches = {};
+          if (!updatedConfig.gitBranches) {
+            updatedConfig.gitBranches = {};
           }
-          if (!updatedConfig.git_branches[currentBranch]) {
-            updatedConfig.git_branches[currentBranch] = { overrides: {} };
+          if (!updatedConfig.gitBranches[currentBranch]) {
+            updatedConfig.gitBranches[currentBranch] = { overrides: {} };
           }
 
           // Write updated configuration
@@ -429,11 +427,11 @@ export async function pullGitSyncSettings(
       const currentBranch = getCurrentGitBranch();
       if (currentBranch) {
         log.info(`Detected Git repository, adding empty branch structure for: ${currentBranch}`);
-        if (!updatedConfig.git_branches) {
-          updatedConfig.git_branches = {};
+        if (!updatedConfig.gitBranches) {
+          updatedConfig.gitBranches = {};
         }
-        if (!updatedConfig.git_branches[currentBranch]) {
-          updatedConfig.git_branches[currentBranch] = { overrides: {} };
+        if (!updatedConfig.gitBranches[currentBranch]) {
+          updatedConfig.gitBranches[currentBranch] = { overrides: {} };
         }
       }
     }

cli/src/commands/gitsync-settings/push.ts

@@ -3,7 +3,7 @@ import { GlobalOptions } from "../../types.ts";
 import { requireLogin } from "../../core/auth.ts";
 import { resolveWorkspace } from "../../core/context.ts";
 import * as wmill from "../../../gen/services.gen.ts";
-import { SyncOptions, readConfigFile, validateBranchConfiguration, getEffectiveSettings } from "../../core/conf.ts";
+import { SyncOptions, readConfigFile, validateBranchConfiguration, getEffectiveSettings, getWmillYamlPath } from "../../core/conf.ts";
 import { deepEqual } from "../../utils/utils.ts";
 
 import { GitSyncRepository } from "./types.ts";
@@ -44,9 +44,8 @@ export async function pushGitSyncSettings(
 
   try {
     // Check if wmill.yaml exists - require it for git-sync settings commands
-    try {
-      await Deno.stat("wmill.yaml");
-    } catch (error) {
+    const wmillYamlPath = getWmillYamlPath();
+    if (!wmillYamlPath) {
       log.error(
         colors.red(
           "No wmill.yaml file found. Please run 'wmill init' first to create the configuration file.",

cli/src/commands/gitsync-settings/utils.ts

@@ -30,12 +30,12 @@ export function getOrCreateBranchConfig(config: SyncOptions, branchName: string)
   config: SyncOptions;
   branchKey: string;
 } {
-  if (!config.git_branches) {
-    config.git_branches = {};
+  if (!config.gitBranches) {
+    config.gitBranches = {};
   }
 
-  if (!config.git_branches[branchName]) {
-    config.git_branches[branchName] = {};
+  if (!config.gitBranches[branchName]) {
+    config.gitBranches[branchName] = {};
   }
 
   return {
@@ -53,12 +53,12 @@ export function applyBackendSettingsToBranch(
   const { config: updatedConfig } = getOrCreateBranchConfig(config, branchName);
 
   // Get the base settings (top-level + defaults) to compare against
-  const { git_branches, ...topLevelSettings } = config;
+  const { gitBranches, ...topLevelSettings } = config;
   const baseSettings: Partial<SyncOptions> = { ...DEFAULT_SYNC_OPTIONS, ...topLevelSettings };
 
   // Only store fields that differ from the base settings
   Object.keys(backendSettings).forEach(key => {
-    if (key !== 'git_branches' && backendSettings[key as keyof SyncOptions] !== undefined) {
+    if (key !== 'gitBranches' && backendSettings[key as keyof SyncOptions] !== undefined) {
       const backendValue = backendSettings[key as keyof SyncOptions];
       const baseValue = baseSettings[key as keyof SyncOptions];
 
@@ -66,10 +66,10 @@ export function applyBackendSettingsToBranch(
       const isDifferent = GitSyncSettingsConverter.isDifferent(backendValue, baseValue);
 
       if (isDifferent) {
-        if (!updatedConfig.git_branches![branchName].overrides) {
-          updatedConfig.git_branches![branchName].overrides = {};
+        if (!updatedConfig.gitBranches![branchName].overrides) {
+          updatedConfig.gitBranches![branchName].overrides = {};
         }
-        (updatedConfig.git_branches![branchName].overrides as any)[key] = backendValue;
+        (updatedConfig.gitBranches![branchName].overrides as any)[key] = backendValue;
       }
     }
   });

cli/src/commands/init/init.ts

@@ -43,14 +43,14 @@ async function initAction(opts: InitOptions) {
     if (isGitRepository()) {
       const currentBranch = getCurrentGitBranch();
       if (currentBranch) {
-        initialConfig.git_branches = {
+        initialConfig.gitBranches = {
           [currentBranch]: { overrides: {} },
         };
       } else {
-        initialConfig.git_branches = {};
+        initialConfig.gitBranches = {};
       }
     } else {
-      initialConfig.git_branches = {};
+      initialConfig.gitBranches = {};
     }
 
     await Deno.writeTextFile("wmill.yaml", yamlStringify(initialConfig));
@@ -116,16 +116,16 @@ async function initAction(opts: InitOptions) {
           const currentConfig = await import("../../core/conf.ts").then((m) =>
             m.readConfigFile()
           );
-          if (!currentConfig.git_branches) {
-            currentConfig.git_branches = {};
+          if (!currentConfig.gitBranches) {
+            currentConfig.gitBranches = {};
           }
-          if (!currentConfig.git_branches[currentBranch]) {
-            currentConfig.git_branches[currentBranch] = { overrides: {} };
+          if (!currentConfig.gitBranches[currentBranch]) {
+            currentConfig.gitBranches[currentBranch] = { overrides: {} };
           }
 
-          currentConfig.git_branches[currentBranch].baseUrl =
+          currentConfig.gitBranches[currentBranch].baseUrl =
             activeWorkspace.remote;
-          currentConfig.git_branches[currentBranch].workspaceId =
+          currentConfig.gitBranches[currentBranch].workspaceId =
             activeWorkspace.workspaceId;
 
           await Deno.writeTextFile(

cli/src/commands/resource/resource.ts

@@ -11,6 +11,8 @@ import { colors, Command, log, SEP, Table } from "../../../deps.ts";
 import * as wmill from "../../../gen/services.gen.ts";
 import { Resource } from "../../../gen/types.gen.ts";
 import { readInlinePathSync } from "../../utils/utils.ts";
+import { isBranchSpecificFile } from "../../core/specific_items.ts";
+import { getCurrentGitBranch } from "../../utils/git.ts";
 
 export interface ResourceFile {
   value: any;
@@ -23,7 +25,8 @@ export async function pushResource(
   workspace: string,
   remotePath: string,
   resource: ResourceFile | Resource | undefined,
-  localResource: ResourceFile
+  localResource: ResourceFile,
+  originalLocalPath?: string
 ): Promise<void> {
   remotePath = removeType(remotePath, "resource");
   try {
@@ -35,21 +38,49 @@ export async function pushResource(
     // flow doesn't exist
   }
 
-  if (localResource.value["content"]?.startsWith("!inline ")) {
-    const basePath = localResource.value["content"].split(" ")[1];
-    localResource.value["content"] = readInlinePathSync(basePath);
-  }
+  // Helper function to resolve inline content
+  const resolveInlineContent = async () => {
+    if (localResource.value["content"]?.startsWith("!inline ")) {
+      const basePath = localResource.value["content"].split(" ")[1];
+
+      // If we're processing a branch-specific metadata file, read from branch-specific resource file
+
+      let pathToRead = basePath;
+
+      if (originalLocalPath && isBranchSpecificFile(originalLocalPath)) {
+        const currentBranch = getCurrentGitBranch();
+        if (currentBranch) {
+          // Directly construct branch-specific resource file path
+          const resourcePathSegments = basePath.split(".");
+          if (resourcePathSegments.length >= 4 && resourcePathSegments[resourcePathSegments.length - 3] === "resource" && resourcePathSegments[resourcePathSegments.length - 2] === "file") {
+            const fileBaseParts = resourcePathSegments.slice(0, -3);
+            const fileExt = resourcePathSegments.slice(-3);
+            pathToRead = [...fileBaseParts, currentBranch, ...fileExt].join(".");
+          }
+        }
+      }
+
+      localResource.value["content"] = readInlinePathSync(pathToRead);
+    }
+  };
+
   if (resource) {
     if (isSuperset(localResource, resource)) {
       return;
     }
 
+    // Only resolve inline content if we're actually updating
+    await resolveInlineContent();
+
     await wmill.updateResource({
       workspace: workspace,
       path: remotePath.replaceAll(SEP, "/"),
       requestBody: { ...localResource },
     });
   } else {
+    // New resource - resolve inline content
+    await resolveInlineContent();
+
     if (localResource.is_oauth) {
       log.info(
         colors.yellow(
@@ -89,7 +120,8 @@ async function push(opts: PushOptions, filePath: string, remotePath: string) {
     workspace.workspaceId,
     remotePath,
     undefined,
-    parseFromFile(filePath)
+    parseFromFile(filePath),
+    filePath  // Pass the local file path for branch-specific inline content resolution
   );
   log.info(colors.bold.underline.green(`Resource ${remotePath} pushed`));
 }

cli/src/commands/script/script.ts

@@ -15,6 +15,8 @@ import {
 } from "../../../deps.ts";
 import { deepEqual } from "../../utils/utils.ts";
 import * as wmill from "../../../gen/services.gen.ts";
+import * as specificItems from "../../core/specific_items.ts";
+import { getCurrentGitBranch } from "../../utils/git.ts";
 
 import {
   defaultScriptMetadata,
@@ -102,12 +104,24 @@ async function push(opts: PushOptions, filePath: string) {
 export async function findResourceFile(path: string) {
   const splitPath = path.split(".");
 
-  const contentBasePathJSON = splitPath[0] + "." + splitPath[1] + ".json";
-  const contentBasePathYAML = splitPath[0] + "." + splitPath[1] + ".yaml";
+  let contentBasePathJSON = splitPath[0] + "." + splitPath[1] + ".json";
+  let contentBasePathYAML = splitPath[0] + "." + splitPath[1] + ".yaml";
+
+  // Check for branch-specific metadata files first
+  const currentBranch = getCurrentGitBranch();
+
+  const candidates = [contentBasePathJSON, contentBasePathYAML];
+
+  if (currentBranch) {
+    // Add branch-specific candidates at the beginning (higher priority)
+    const branchSpecificJSON = specificItems.toBranchSpecificPath(contentBasePathJSON, currentBranch);
+    const branchSpecificYAML = specificItems.toBranchSpecificPath(contentBasePathYAML, currentBranch);
+    candidates.unshift(branchSpecificJSON, branchSpecificYAML);
+  }
 
   const validCandidates = (
     await Promise.all(
-      [contentBasePathJSON, contentBasePathYAML].map((x) => {
+      candidates.map((x) => {
         return Deno.stat(x)
           .catch(() => undefined)
           .then((x) => x?.isFile)
@@ -580,7 +594,7 @@ export function filePathExtensionFromContentType(
     return ".java";
   } else if (language === "ruby") {
     return ".rb";
-    // for related places search: ADD_NEW_LANG 
+    // for related places search: ADD_NEW_LANG
   } else {
     throw new Error("Invalid language: " + language);
   }
@@ -611,7 +625,7 @@ export const exts = [
   ".playbook.yml",
   ".java",
   ".rb"
-  // for related places search: ADD_NEW_LANG 
+  // for related places search: ADD_NEW_LANG
 ];
 
 export function removeExtensionToPath(path: string): string {

cli/src/commands/sync/sync.ts

@@ -42,7 +42,19 @@ import {
   readConfigFile,
   getEffectiveSettings,
   validateBranchConfiguration,
+  mergeConfigWithConfigFile,
 } from "../../core/conf.ts";
+import {
+  SpecificItemsConfig,
+  getSpecificItemsForCurrentBranch,
+  isSpecificItem,
+  getBranchSpecificPath,
+  fromBranchSpecificPath,
+  isCurrentBranchFile,
+  toBranchSpecificPath,
+  isBranchSpecificFile,
+} from "../../core/specific_items.ts";
+import { getCurrentGitBranch } from "../../utils/git.ts";
 import { Workspace } from "../workspace/workspace.ts";
 import { removePathPrefix } from "../../types.ts";
 import { SyncCodebase, listSyncCodebases } from "../../utils/codebase.ts";
@@ -67,9 +79,9 @@ function mergeCliWithEffectiveOptions<
 // Resolve effective sync options using branch-based configuration
 async function resolveEffectiveSyncOptions(
   workspace: Workspace,
+  localConfig: SyncOptions,
   promotion?: string
 ): Promise<SyncOptions> {
-  const localConfig = await readConfigFile();
   return await getEffectiveSettings(localConfig, promotion);
 }
 
@@ -631,9 +643,36 @@ export async function elementsToMap(
   els: DynFSElement,
   ignore: (path: string, isDirectory: boolean) => boolean,
   json: boolean,
-  skips: Skips
+  skips: Skips,
+  specificItems?: SpecificItemsConfig
 ): Promise<{ [key: string]: string }> {
   const map: { [key: string]: string } = {};
+  const processedBasePaths = new Set<string>();
+
+  // First pass: collect all file paths to identify branch-specific files
+  const allPaths: string[] = [];
+  for await (const entry of readDirRecursiveWithIgnore(ignore, els)) {
+    if (!entry.isDirectory && !entry.ignored) {
+      allPaths.push(entry.path);
+    }
+  }
+
+  const branchSpecificExists = new Set<string>();
+
+  if (specificItems) {
+    const currentBranch = getCurrentGitBranch();
+    if (currentBranch) {
+      for (const path of allPaths) {
+        if (isCurrentBranchFile(path)) {
+          const basePath = fromBranchSpecificPath(path, currentBranch);
+          if (isSpecificItem(basePath, specificItems)) {
+            branchSpecificExists.add(basePath);
+          }
+        }
+      }
+    }
+  }
+
   for await (const entry of readDirRecursiveWithIgnore(ignore, els)) {
     if (entry.isDirectory || entry.ignored) continue;
     const path = entry.path;
@@ -695,11 +734,21 @@ export async function elementsToMap(
         "nu",
         "java",
         "rb",
-        // for related places search: ADD_NEW_LANG 
+        // for related places search: ADD_NEW_LANG
       ].includes(path.split(".").pop() ?? "") &&
       !isFileResource(path)
     )
       continue;
+
+    // Handle branch-specific files - skip files for other branches
+    if (specificItems && isBranchSpecificFile(path)) {
+      const currentBranch = getCurrentGitBranch();
+      if (!currentBranch || !isCurrentBranchFile(path)) {
+        // Skip branch-specific files for other branches
+        continue;
+      }
+    }
+
     const content = await entry.getContentText();
 
     if (skips.skipSecrets && path.endsWith(".variable" + ext)) {
@@ -727,7 +776,33 @@ export async function elementsToMap(
         log.warn(`Error reading variable ${path} to check for secrets`);
       }
     }
-    map[entry.path] = content;
+
+    // Handle branch-specific path mapping after all filtering
+    if (specificItems) {
+      const currentBranch = getCurrentGitBranch();
+      if (currentBranch && isCurrentBranchFile(path)) {
+        // This is a branch-specific file for current branch
+        const basePath = fromBranchSpecificPath(path, currentBranch);
+        if (isSpecificItem(basePath, specificItems)) {
+          // Map to base path for push operations
+          map[basePath] = content;
+          processedBasePaths.add(basePath);
+        } else {
+          // Branch-specific file doesn't match pattern, skip it
+          continue;
+        }
+      } else if (!isBranchSpecificFile(path)) {
+        // This is a regular base file, check if we should skip it
+        if (processedBasePaths.has(path)) {
+          // Skip base file, we already processed branch-specific version
+          continue;
+        }
+        map[path] = content;
+      }
+    } else {
+      // No specific items configuration, use regular path
+      map[entry.path] = content;
+    }
   }
   return map;
 }
@@ -758,14 +833,15 @@ async function compareDynFSElement(
   skips: Skips,
   ignoreMetadataDeletion: boolean,
   codebases: SyncCodebase[],
-  ignoreCodebaseChanges: boolean
+  ignoreCodebaseChanges: boolean,
+  specificItems?: SpecificItemsConfig
 ): Promise<Change[]> {
   const [m1, m2] = els2
     ? await Promise.all([
-        elementsToMap(els1, ignore, json, skips),
-        elementsToMap(els2, ignore, json, skips),
+        elementsToMap(els1, ignore, json, skips, specificItems),
+        elementsToMap(els2, ignore, json, skips, specificItems),
       ])
-    : [await elementsToMap(els1, ignore, json, skips), {}];
+    : [await elementsToMap(els1, ignore, json, skips, specificItems), {}];
 
   const changes: Change[] = [];
 
@@ -1163,6 +1239,10 @@ export async function pull(
   opts: GlobalOptions &
     SyncOptions & { repository?: string; promotion?: string }
 ) {
+
+  const originalCliOpts = { ...opts };
+  opts = await mergeConfigWithConfigFile(opts);
+
   // Validate branch configuration early
   try {
     await validateBranchConfiguration(false, opts.yes);
@@ -1184,11 +1264,15 @@ export async function pull(
   // Resolve effective sync options with branch awareness
   const effectiveOpts = await resolveEffectiveSyncOptions(
     workspace,
+    opts,
     opts.promotion
   );
 
+  // Extract specific items configuration before merging overwrites gitBranches
+  const specificItems = getSpecificItemsForCurrentBranch(opts);
+
   // Merge CLI flags with resolved settings (CLI flags take precedence only for explicit overrides)
-  opts = mergeCliWithEffectiveOptions(opts, effectiveOpts);
+  opts = mergeCliWithEffectiveOptions(originalCliOpts, effectiveOpts);
 
   const codebases = await listSyncCodebases(opts);
 
@@ -1238,7 +1322,8 @@ export async function pull(
     opts,
     false,
     codebases,
-    true
+    true,
+    specificItems
   );
 
   log.info(
@@ -1255,6 +1340,12 @@ export async function pull(
         ...(change.name === "edited" && change.codebase
           ? { codebase_changed: true }
           : {}),
+        ...(specificItems && isSpecificItem(change.path, specificItems)
+          ? {
+              branch_specific: true,
+              branch_specific_path: getBranchSpecificPath(change.path, specificItems)
+            }
+          : {}),
       })),
       total: changes.length,
     };
@@ -1264,7 +1355,7 @@ export async function pull(
 
   if (changes.length > 0) {
     if (!opts.jsonOutput) {
-      prettyChanges(changes);
+      prettyChanges(changes, specificItems);
     }
     if (opts.dryRun) {
       log.info(colors.gray(`Dry run complete.`));
@@ -1284,8 +1375,17 @@ export async function pull(
 
     log.info(colors.gray(`Applying changes to files ...`));
     for await (const change of changes) {
-      const target = path.join(Deno.cwd(), change.path);
-      const stateTarget = path.join(Deno.cwd(), ".wmill", change.path);
+      // Determine if this file should be written to a branch-specific path
+      let targetPath = change.path;
+      if (specificItems && isSpecificItem(change.path, specificItems)) {
+        const branchSpecificPath = getBranchSpecificPath(change.path, specificItems);
+        if (branchSpecificPath) {
+          targetPath = branchSpecificPath;
+        }
+      }
+
+      const target = path.join(Deno.cwd(), targetPath);
+      const stateTarget = path.join(Deno.cwd(), ".wmill", targetPath);
       if (change.name === "edited") {
         if (opts.stateful) {
           try {
@@ -1328,12 +1428,12 @@ export async function pull(
           }
         }
         if (exts.some((e) => change.path.endsWith(e))) {
-          log.info(`Editing script content of ${change.path}`);
+          log.info(`Editing script content of ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`);
         } else if (
           change.path.endsWith(".yaml") ||
           change.path.endsWith(".json")
         ) {
-          log.info(`Editing ${getTypeStrFromPath(change.path)} ${change.path}`);
+          log.info(`Editing ${getTypeStrFromPath(change.path)} ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`);
         }
         await Deno.writeTextFile(target, change.after);
 
@@ -1345,10 +1445,10 @@ export async function pull(
         await ensureDir(path.dirname(target));
         if (opts.stateful) {
           await ensureDir(path.dirname(stateTarget));
-          log.info(`Adding ${getTypeStrFromPath(change.path)} ${change.path}`);
+          log.info(`Adding ${getTypeStrFromPath(change.path)} ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`);
         }
         await Deno.writeTextFile(target, change.content);
-        log.info(`Writing ${getTypeStrFromPath(change.path)} ${change.path}`);
+        log.info(`Writing ${getTypeStrFromPath(change.path)} ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`);
         if (opts.stateful) {
           await Deno.copyFile(target, stateTarget);
         }
@@ -1423,6 +1523,12 @@ export async function pull(
           ...(change.name === "edited" && change.codebase
             ? { codebase_changed: true }
             : {}),
+          ...(specificItems && isSpecificItem(change.path, specificItems)
+            ? {
+                branch_specific: true,
+                branch_specific_path: getBranchSpecificPath(change.path, specificItems)
+              }
+            : {}),
         })),
         total: changes.length,
       };
@@ -1445,21 +1551,33 @@ export async function pull(
   }
 }
 
-function prettyChanges(changes: Change[]) {
+function prettyChanges(changes: Change[], specificItems?: SpecificItemsConfig) {
   for (const change of changes) {
+    let displayPath = change.path;
+    let branchNote = "";
+
+    // Check if this will be written as a branch-specific file
+    if (specificItems && isSpecificItem(change.path, specificItems)) {
+      const branchSpecificPath = getBranchSpecificPath(change.path, specificItems);
+      if (branchSpecificPath) {
+        displayPath = branchSpecificPath;
+        branchNote = " (branch-specific)";
+      }
+    }
+
     if (change.name === "added") {
       log.info(
-        colors.green(`+ ${getTypeStrFromPath(change.path)} ` + change.path)
+        colors.green(`+ ${getTypeStrFromPath(change.path)} ` + displayPath + colors.gray(branchNote))
       );
     } else if (change.name === "deleted") {
       log.info(
-        colors.red(`- ${getTypeStrFromPath(change.path)} ` + change.path)
+        colors.red(`- ${getTypeStrFromPath(change.path)} ` + displayPath + colors.gray(branchNote))
       );
     } else if (change.name === "edited") {
       log.info(
         colors.yellow(
           `~ ${getTypeStrFromPath(change.path)} ` +
-            change.path +
+            displayPath + colors.gray(branchNote) +
             (change.codebase ? ` (codebase changed)` : "")
         )
       );
@@ -1499,6 +1617,12 @@ function removeSuffix(str: string, suffix: string) {
 export async function push(
   opts: GlobalOptions & SyncOptions & { repository?: string }
 ) {
+  // Save original CLI options before merging with config file
+  const originalCliOpts = { ...opts };
+
+  // Load configuration from wmill.yaml and merge with CLI options
+  opts = await mergeConfigWithConfigFile(opts);
+
   // Validate branch configuration early
   try {
     await validateBranchConfiguration(false, opts.yes);
@@ -1516,11 +1640,15 @@ export async function push(
   // Resolve effective sync options with branch awareness
   const effectiveOpts = await resolveEffectiveSyncOptions(
     workspace,
+    opts,
     opts.promotion
   );
 
+  // Extract specific items configuration BEFORE merging overwrites gitBranches
+  const specificItems = getSpecificItemsForCurrentBranch(opts);
+
   // Merge CLI flags with resolved settings (CLI flags take precedence only for explicit overrides)
-  opts = mergeCliWithEffectiveOptions(opts, effectiveOpts);
+  opts = mergeCliWithEffectiveOptions(originalCliOpts, effectiveOpts);
 
   const codebases = await listSyncCodebases(opts);
   if (opts.raw) {
@@ -1581,7 +1709,8 @@ export async function push(
     opts,
     true,
     codebases,
-    false
+    false,
+    specificItems
   );
 
   const globalDeps = await findGlobalDeps();
@@ -1660,6 +1789,12 @@ export async function push(
         ...(change.name === "edited" && change.codebase
           ? { codebase_changed: true }
           : {}),
+        ...(specificItems && isSpecificItem(change.path, specificItems)
+          ? {
+              branch_specific: true,
+              branch_specific_path: getBranchSpecificPath(change.path, specificItems)
+            }
+          : {}),
       })),
       total: changes.length,
     };
@@ -1669,7 +1804,7 @@ export async function push(
 
   if (changes.length > 0) {
     if (!opts.jsonOutput) {
-      prettyChanges(changes);
+      prettyChanges(changes, specificItems);
     }
     if (opts.dryRun) {
       log.info(colors.gray(`Dry run complete.`));
@@ -1795,11 +1930,21 @@ export async function push(
                     await Deno.readTextFile(resourceFilePath)
                   );
 
+                  // For branch-specific resources, push to the base path on the workspace server
+                  // This ensures branch-specific files are stored with their base names in the workspace
+                  let serverPath = resourceFilePath;
+                  const currentBranch = getCurrentGitBranch();
+
+                  if (currentBranch && isBranchSpecificFile(resourceFilePath)) {
+                    serverPath = fromBranchSpecificPath(resourceFilePath, currentBranch);
+                  }
+
                   await pushResource(
                     workspace.workspaceId,
-                    resourceFilePath,
+                    serverPath,
                     undefined,
-                    newObj
+                    newObj,
+                    resourceFilePath
                   );
                   if (stateTarget) {
                     await Deno.writeTextFile(stateTarget, change.after);
@@ -1810,6 +1955,12 @@ export async function push(
               const oldObj = parseFromPath(change.path, change.before);
               const newObj = parseFromPath(change.path, change.after);
 
+              // Check if this is a branch-specific item and get the original branch-specific path
+              let originalBranchSpecificPath: string | undefined;
+              if (specificItems && isSpecificItem(change.path, specificItems)) {
+                originalBranchSpecificPath = getBranchSpecificPath(change.path, specificItems);
+              }
+
               await pushObj(
                 workspace.workspaceId,
                 change.path,
@@ -1817,7 +1968,8 @@ export async function push(
                 newObj,
                 opts.plainSecrets ?? false,
                 alreadySynced,
-                opts.message
+                opts.message,
+                originalBranchSpecificPath
               );
 
               if (stateTarget) {
@@ -1851,6 +2003,17 @@ export async function push(
                 );
               }
               const obj = parseFromPath(change.path, change.content);
+
+              // Determine the actual local file path for this change
+              // For branch-specific items, we read from branch-specific files but push to base server paths
+              let localFilePath = change.path;
+              if (specificItems && isSpecificItem(change.path, specificItems)) {
+                const branchSpecificPath = getBranchSpecificPath(change.path, specificItems);
+                if (branchSpecificPath) {
+                  localFilePath = branchSpecificPath;
+                }
+              }
+
               await pushObj(
                 workspace.workspaceId,
                 change.path,
@@ -1858,7 +2021,8 @@ export async function push(
                 obj,
                 opts.plainSecrets ?? false,
                 [],
-                opts.message
+                opts.message,
+                localFilePath  // Pass the actual local file path
               );
 
               if (stateTarget) {
@@ -2041,6 +2205,12 @@ export async function push(
           ...(change.name === "edited" && change.codebase
             ? { codebase_changed: true }
             : {}),
+          ...(specificItems && isSpecificItem(change.path, specificItems)
+            ? {
+                branch_specific: true,
+                branch_specific_path: getBranchSpecificPath(change.path, specificItems)
+              }
+            : {}),
         })),
         total: changes.length,
         duration_ms: Math.round(performance.now() - start),

cli/src/commands/trigger/trigger.ts

@@ -15,7 +15,10 @@ import {
   isSuperset,
   parseFromFile,
   removeType,
+  TRIGGER_TYPES,
 } from "../../types.ts";
+import { fromBranchSpecificPath, isBranchSpecificFile } from "../../core/specific_items.ts";
+import { getCurrentGitBranch } from "../../utils/git.ts";
 import { requireLogin } from "../../core/auth.ts";
 import { validatePath, resolveWorkspace } from "../../core/context.ts";
 
@@ -222,25 +225,29 @@ async function list(opts: GlobalOptions) {
 }
 
 function checkIfValidTrigger(kind: string | undefined): kind is TriggerType {
-  if (
-    kind &&
-    [
-      "http",
-      "websocket",
-      "kafka",
-      "nats",
-      "postgres",
-      "mqtt",
-      "sqs",
-      "gcp",
-    ].includes(kind)
-  ) {
+  if (kind && (TRIGGER_TYPES as readonly string[]).includes(kind)) {
     return true;
   } else {
     return false;
   }
 }
 
+function extractTriggerKindFromPath(filePath: string): string | undefined {
+  let pathToAnalyze = filePath;
+
+  // If this is a branch-specific file, convert it to the base path first
+  if (isBranchSpecificFile(filePath)) {
+    const currentBranch = getCurrentGitBranch();
+    if (currentBranch) {
+      pathToAnalyze = fromBranchSpecificPath(filePath, currentBranch);
+    }
+  }
+
+  // Now extract trigger type from the base path: "something.kafka_trigger.yaml" -> "kafka"
+  const triggerMatch = pathToAnalyze.match(/\.(\w+)_trigger\.yaml$/);
+  return triggerMatch ? triggerMatch[1] : undefined;
+}
+
 async function push(opts: GlobalOptions, filePath: string, remotePath: string) {
   const workspace = await resolveWorkspace(opts);
   await requireLogin(opts);
@@ -256,7 +263,7 @@ async function push(opts: GlobalOptions, filePath: string, remotePath: string) {
 
   console.log(colors.bold.yellow("Pushing trigger..."));
 
-  const triggerKind = filePath.split(".")[1].split("_")[0];
+  const triggerKind = extractTriggerKindFromPath(filePath);
   if (!checkIfValidTrigger(triggerKind)) {
     throw new Error("Invalid trigger kind: " + triggerKind);
   }

cli/src/commands/workspace/workspace.ts

@@ -386,22 +386,22 @@ async function bind(
   }
 
   // For unbind, check if branch exists
-  if (!bindWorkspace && (!config.git_branches || !config.git_branches[branch])) {
-    log.error(colors.red(`Branch '${branch}' not found in wmill.yaml git_branches`));
+  if (!bindWorkspace && (!config.gitBranches || !config.gitBranches[branch])) {
+    log.error(colors.red(`Branch '${branch}' not found in wmill.yaml gitBranches`));
     return;
   }
 
   // Update the branch configuration with workspace binding
-  if (!config.git_branches) {
-    config.git_branches = {};
+  if (!config.gitBranches) {
+    config.gitBranches = {};
   }
-  if (!config.git_branches[branch]) {
-    config.git_branches[branch] = { overrides: {} };
+  if (!config.gitBranches[branch]) {
+    config.gitBranches[branch] = { overrides: {} };
   }
 
   if (bindWorkspace && activeWorkspace) {
-    config.git_branches[branch].baseUrl = activeWorkspace.remote;
-    config.git_branches[branch].workspaceId = activeWorkspace.workspaceId;
+    config.gitBranches[branch].baseUrl = activeWorkspace.remote;
+    config.gitBranches[branch].workspaceId = activeWorkspace.workspaceId;
 
     log.info(colors.green(
       `✓ Bound branch '${branch}' to workspace '${activeWorkspace.name}'\n` +
@@ -409,8 +409,8 @@ async function bind(
     ));
   } else {
     // Unbind
-    delete config.git_branches[branch].baseUrl;
-    delete config.git_branches[branch].workspaceId;
+    delete config.gitBranches[branch].baseUrl;
+    delete config.gitBranches[branch].workspaceId;
 
     log.info(colors.green(`✓ Removed workspace binding from branch '${branch}'`));
   }

cli/src/core/conf.ts

@@ -1,5 +1,8 @@
 import { log, yamlParseFile, Confirm, yamlStringify } from "../../deps.ts";
 import { getCurrentGitBranch, isGitRepository } from "../utils/git.ts";
+import { join, dirname, resolve, relative } from "node:path";
+import { existsSync } from "node:fs";
+import { execSync } from "node:child_process";
 
 export let showDiffs = false;
 export function setShowDiffs(value: boolean) {
@@ -37,13 +40,44 @@ export interface SyncOptions {
   codebases?: Codebase[];
   parallel?: number;
   jsonOutput?: boolean;
-  git_branches?: {
+  gitBranches?: {
+    commonSpecificItems?: {
+      variables?: string[];
+      resources?: string[];
+      triggers?: string[];
+    };
+  } & {
     [branchName: string]: SyncOptions & {
       overrides?: Partial<SyncOptions>;
       promotionOverrides?: Partial<SyncOptions>;
       baseUrl?: string;
       workspaceId?: string;
-    }
+      specificItems?: {
+        variables?: string[];
+        resources?: string[];
+        triggers?: string[];
+      };
+    };
+  };
+  // Legacy field - deprecated, use gitBranches instead
+  git_branches?: {
+    commonSpecificItems?: {
+      variables?: string[];
+      resources?: string[];
+      triggers?: string[];
+    };
+  } & {
+    [branchName: string]: SyncOptions & {
+      overrides?: Partial<SyncOptions>;
+      promotionOverrides?: Partial<SyncOptions>;
+      baseUrl?: string;
+      workspaceId?: string;
+      specificItems?: {
+        variables?: string[];
+        resources?: string[];
+        triggers?: string[];
+      };
+    };
   };
   promotion?: string;
 }
@@ -62,9 +96,90 @@ export interface Codebase {
   inject?: string[];
 }
 
+function getGitRepoRoot(): string | null {
+  try {
+    const result = execSync("git rev-parse --show-toplevel", {
+      encoding: "utf8",
+      stdio: "pipe"
+    });
+    return result.trim();
+  } catch (error) {
+    return null;
+  }
+}
+
+function findWmillYaml(): string | null {
+  const startDir = resolve(Deno.cwd());
+  const isInGitRepo = isGitRepository();
+
+  // If not in git repo, only check current directory
+  if (!isInGitRepo) {
+    const wmillYamlPath = join(startDir, "wmill.yaml");
+    return existsSync(wmillYamlPath) ? wmillYamlPath : null;
+  }
+
+  // If in git repo, search up to git repository root
+  const gitRoot = getGitRepoRoot();
+  let currentDir = startDir;
+  let foundPath: string | null = null;
+
+  while (true) {
+    const wmillYamlPath = join(currentDir, "wmill.yaml");
+
+    if (existsSync(wmillYamlPath)) {
+      foundPath = wmillYamlPath;
+      break;
+    }
+
+    // Check if we've reached the git repository root
+    if (gitRoot && resolve(currentDir) === resolve(gitRoot)) {
+      break;
+    }
+
+    // Check if we've reached the filesystem root
+    const parentDir = dirname(currentDir);
+    if (parentDir === currentDir) {
+      break;
+    }
+
+    currentDir = parentDir;
+  }
+
+  // If wmill.yaml was found in a parent directory, warn the user and change working directory
+  if (foundPath && resolve(dirname(foundPath)) !== resolve(startDir)) {
+    const configDir = dirname(foundPath);
+    const relativePath = relative(startDir, foundPath);
+    log.warn(`⚠️  wmill.yaml found in parent directory: ${relativePath}`);
+
+    // Change working directory to where wmill.yaml was found
+    Deno.chdir(configDir);
+    log.info(`📁 Changed working directory to: ${configDir}`);
+  }
+
+  return foundPath;
+}
+
+export function getWmillYamlPath(): string | null {
+  return findWmillYaml();
+}
+
 export async function readConfigFile(): Promise<SyncOptions> {
   try {
-    const conf = (await yamlParseFile("wmill.yaml")) as SyncOptions;
+    // First, try to find wmill.yaml recursively
+    const wmillYamlPath = findWmillYaml();
+
+    if (!wmillYamlPath) {
+      log.warn(
+        "No wmill.yaml found. Use 'wmill init' to bootstrap it. Using 'bun' as default typescript runtime."
+      );
+      return {};
+    }
+
+    const conf = (await yamlParseFile(wmillYamlPath)) as SyncOptions;
+
+    // Handle legacy format migrations (combine overrides and git_branches)
+    let needsConfigWrite = false;
+    const migrationMessages: string[] = [];
 
     // Handle obsolete overrides format
     if (conf && 'overrides' in conf) {
@@ -78,18 +193,54 @@ export async function readConfigFile(): Promise<SyncOptions> {
           "   Please delete your wmill.yaml and run 'wmill init' to recreate it with the new format."
         );
       } else {
-        // Remove empty overrides with a note
-        log.info("ℹ️  Removing empty 'overrides: {}' from wmill.yaml (migrated to git_branches format)");
+        // Remove empty overrides
         delete conf.overrides;
-        // Write the updated config back to file
-        try {
-          await Deno.writeTextFile("wmill.yaml", yamlStringify(conf));
-        } catch (error) {
-          log.warn(`Could not update wmill.yaml to remove empty overrides: ${error instanceof Error ? error.message : error}`);
-        }
+        needsConfigWrite = true;
+        migrationMessages.push("ℹ️  Removing empty 'overrides: {}' from wmill.yaml (migrated to gitBranches format)");
       }
     }
 
+    // Handle git_branches to gitBranches migration
+    if (conf && 'git_branches' in conf) {
+      if (!conf.gitBranches) {
+        // Deep copy git_branches to gitBranches (even if empty)
+        conf.gitBranches = JSON.parse(JSON.stringify(conf.git_branches));
+        needsConfigWrite = true;
+        migrationMessages.push("⚠️  Migrating 'git_branches' to 'gitBranches' (camelCase). The snake_case format is deprecated.");
+        migrationMessages.push("✅ Successfully migrated 'git_branches' to 'gitBranches' in wmill.yaml");
+      } else {
+        migrationMessages.push("⚠️  Both 'git_branches' and 'gitBranches' found in wmill.yaml. Using 'gitBranches' and ignoring 'git_branches'.");
+      }
+      // Always remove the old field from config object (both file and memory)
+      delete conf.git_branches;
+    }
+
+    // Perform single atomic write if any migrations are needed
+    if (needsConfigWrite) {
+      try {
+        await Deno.writeTextFile(wmillYamlPath, yamlStringify(conf));
+        // Log all migration messages after successful write
+        migrationMessages.forEach(msg => {
+          if (msg.startsWith('⚠️')) {
+            log.warn(msg);
+          } else {
+            log.info(msg);
+          }
+        });
+      } catch (error) {
+        log.warn(`Could not update wmill.yaml to apply migrations: ${error instanceof Error ? error.message : error}`);
+      }
+    } else if (migrationMessages.length > 0) {
+      // Log messages for non-write cases (like "both found")
+      migrationMessages.forEach(msg => {
+        if (msg.startsWith('⚠️')) {
+          log.warn(msg);
+        } else {
+          log.info(msg);
+        }
+      });
+    }
+
     if (conf?.defaultTs == undefined) {
       log.warn(
         "No defaultTs defined in your wmill.yaml. Using 'bun' as default."
@@ -100,10 +251,23 @@ export async function readConfigFile(): Promise<SyncOptions> {
     if (e instanceof Error && (e.message.includes("overrides") || e.message.includes("Obsolete configuration format"))) {
       throw e; // Re-throw the specific obsolete format error
     }
-    log.warn(
-      "No wmill.yaml found. Use 'wmill init' to bootstrap it. Using 'bun' as default typescript runtime."
-    );
-    return {};
+
+    // Since we already found the file path, this is likely a parsing or access error
+    if (e instanceof Error && e.message.includes("Error parsing yaml")) {
+      const yamlError = e.cause instanceof Error ? e.cause.message : String(e.cause);
+      throw new Error(
+        "❌ YAML syntax error in wmill.yaml:\n" +
+        "   " + yamlError + "\n" +
+        "   Please fix the YAML syntax in wmill.yaml or delete the file to start fresh."
+      );
+    } else {
+      // File exists but has other issues (permissions, etc.)
+      throw new Error(
+        "❌ Failed to read wmill.yaml:\n" +
+        "   " + (e instanceof Error ? e.message : String(e)) + "\n" +
+        "   Please check file permissions or fix the syntax."
+      );
+    }
   }
 }
 
@@ -148,26 +312,26 @@ export async function validateBranchConfiguration(skipValidation?: boolean, auto
   }
 
   const config = await readConfigFile();
-  const { git_branches } = config;
+  const { gitBranches } = config;
   const currentBranch = getCurrentGitBranch();
 
-  // In a git repository, git_branches section is recommended
-  if (!git_branches || Object.keys(git_branches).length === 0) {
+  // In a git repository, gitBranches section is recommended
+  if (!gitBranches || Object.keys(gitBranches).length === 0) {
     log.warn(
-      "⚠️  WARNING: In a Git repository, the 'git_branches' section is recommended in wmill.yaml.\n" +
-      "   Consider adding a git_branches section with configuration for your Git branches.\n" +
+      "⚠️  WARNING: In a Git repository, the 'gitBranches' section is recommended in wmill.yaml.\n" +
+      "   Consider adding a gitBranches section with configuration for your Git branches.\n" +
       "   Run 'wmill init' to recreate the configuration file with proper branch setup."
     );
     return;
   }
 
-  // Current branch must be defined in git_branches config
-  if (currentBranch && !git_branches[currentBranch]) {
+  // Current branch must be defined in gitBranches config
+  if (currentBranch && !gitBranches[currentBranch]) {
     // In interactive mode, offer to create the branch
     if (Deno.stdin.isTerminal()) {
-      const availableBranches = Object.keys(git_branches).join(', ');
+      const availableBranches = Object.keys(gitBranches).join(', ');
       log.info(
-        `Current Git branch '${currentBranch}' is not defined in the git_branches configuration.\n` +
+        `Current Git branch '${currentBranch}' is not defined in the gitBranches configuration.\n` +
         `Available branches: ${availableBranches}`
       );
 
@@ -177,13 +341,21 @@ export async function validateBranchConfiguration(skipValidation?: boolean, auto
       });
 
       if (shouldCreate) {
+        // Warn if branch name contains filesystem-unsafe characters
+        if (/[\/\\:*?"<>|.]/.test(currentBranch)) {
+          const sanitizedBranchName = currentBranch.replace(/[\/\\:*?"<>|.]/g, '_');
+          log.warn(`⚠️  WARNING: Branch name "${currentBranch}" contains filesystem-unsafe characters (/ \\ : * ? " < > | .).`);
+          log.warn(`   Branch-specific files will be saved with sanitized name: "${sanitizedBranchName}"`);
+          log.warn(`   Example: "file.variable.yaml" → "file.${sanitizedBranchName}.variable.yaml"`);
+        }
+
         // Read current config, add branch, and write it back
         const currentConfig = await readConfigFile();
 
-        if (!currentConfig.git_branches) {
-          currentConfig.git_branches = {};
+        if (!currentConfig.gitBranches) {
+          currentConfig.gitBranches = {};
         }
-        currentConfig.git_branches[currentBranch] = { overrides: {} };
+        currentConfig.gitBranches[currentBranch] = { overrides: {} };
 
         await Deno.writeTextFile("wmill.yaml", yamlStringify(currentConfig));
 
@@ -193,10 +365,17 @@ export async function validateBranchConfiguration(skipValidation?: boolean, auto
         return;
       }
     } else {
+      // Warn about filesystem-unsafe characters in branch name
+      if (/[\/\\:*?"<>|.]/.test(currentBranch)) {
+        const sanitizedBranchName = currentBranch.replace(/[\/\\:*?"<>|.]/g, '_');
+        log.warn(`⚠️  WARNING: Branch name "${currentBranch}" contains filesystem-unsafe characters (/ \\ : * ? " < > | .).`);
+        log.warn(`   Branch-specific files will use sanitized name: "${sanitizedBranchName}"`);
+      }
+      
       log.warn(
-        `⚠️  WARNING: Current Git branch '${currentBranch}' is not defined in the git_branches configuration.\n` +
-        `   Consider adding configuration for branch '${currentBranch}' in the git_branches section of wmill.yaml.\n` +
-        `   Available branches: ${Object.keys(git_branches).join(', ')}`
+        `⚠️  WARNING: Current Git branch '${currentBranch}' is not defined in the gitBranches configuration.\n` +
+        `   Consider adding configuration for branch '${currentBranch}' in the gitBranches section of wmill.yaml.\n` +
+        `   Available branches: ${Object.keys(gitBranches).join(', ')}`
       );
       return;
     }
@@ -206,15 +385,15 @@ export async function validateBranchConfiguration(skipValidation?: boolean, auto
 // Get effective settings by merging top-level settings with branch-specific overrides
 export async function getEffectiveSettings(config: SyncOptions, promotion?: string, skipBranchValidation?: boolean, suppressLogs?: boolean): Promise<SyncOptions> {
   // Start with top-level settings from config
-  const { git_branches, ...topLevelSettings } = config;
+  const { gitBranches, ...topLevelSettings } = config;
   let effective = { ...topLevelSettings };
 
   if (isGitRepository()) {
     const currentBranch = getCurrentGitBranch();
 
     // If promotion is specified, use that branch's promotionOverrides or overrides
-    if (promotion && git_branches && git_branches[promotion]) {
-      const targetBranch = git_branches[promotion];
+    if (promotion && gitBranches && gitBranches[promotion]) {
+      const targetBranch = gitBranches[promotion];
 
       // First try promotionOverrides, then fall back to overrides
       if (targetBranch.promotionOverrides) {
@@ -232,8 +411,8 @@ export async function getEffectiveSettings(config: SyncOptions, promotion?: stri
       }
     }
     // Otherwise use current branch overrides (existing behavior)
-    else if (currentBranch && git_branches && git_branches[currentBranch] && git_branches[currentBranch].overrides) {
-      Object.assign(effective, git_branches[currentBranch].overrides);
+    else if (currentBranch && gitBranches && gitBranches[currentBranch] && gitBranches[currentBranch].overrides) {
+      Object.assign(effective, gitBranches[currentBranch].overrides);
       if (!suppressLogs) {
         log.info(`Applied settings for Git branch: ${currentBranch}`);
       }

cli/src/core/context.ts

@@ -109,15 +109,11 @@ async function tryResolveWorkspace(
     return { isError: false, value: e };
   }
 
-  const defaultWorkspace = await getActiveWorkspace(opts);
-  if (!defaultWorkspace) {
-    return {
-      isError: true,
-      error: colors.red.underline("No workspace given and no default set."),
-    };
-  }
-
-  return { isError: false, value: defaultWorkspace };
+  // Only check for explicit workspace, don't fallback to active workspace here
+  return {
+    isError: true,
+    error: colors.red.underline("No explicit workspace given."),
+  };
 }
 
 async function tryResolveBranchWorkspace(
@@ -135,7 +131,7 @@ async function tryResolveBranchWorkspace(
 
   // Read wmill.yaml to check for branch workspace configuration
   const config = await readConfigFile();
-  const branchConfig = config.git_branches?.[currentBranch];
+  const branchConfig = config.gitBranches?.[currentBranch];
 
   // Check if branch has workspace configuration
   if (!branchConfig?.baseUrl || !branchConfig?.workspaceId) {
@@ -259,6 +255,9 @@ async function tryResolveBranchWorkspace(
 export async function resolveWorkspace(
   opts: GlobalOptions
 ): Promise<Workspace> {
+  const cache = (opts as any).__secret_workspace;
+  if (cache) return cache;
+
   if (opts.baseUrl) {
     if (opts.workspace && opts.token) {
       let normalizedBaseUrl: string;
@@ -328,20 +327,28 @@ export async function resolveWorkspace(
     }
   }
 
-  // Try explicit workspace flag first (should override branch-based resolution)
+  // Try explicit workspace flag first (highest priority)
   const res = await tryResolveWorkspace(opts);
   if (!res.isError) {
     return res.value;
   }
 
-  // Fall back to branch-based resolution if no explicit workspace
+  // Try branch-based resolution (medium priority)
   const branchWorkspace = await tryResolveBranchWorkspace(opts);
   if (branchWorkspace) {
+    (opts as any).__secret_workspace = branchWorkspace;
     return branchWorkspace;
   }
 
-  // If both failed, show the original error from explicit workspace resolution
-  log.info(colors.red.bold(res.error));
+  // Fall back to active workspace (lowest priority)
+  const activeWorkspace = await getActiveWorkspace(opts);
+  if (activeWorkspace) {
+    (opts as any).__secret_workspace = activeWorkspace;
+    return activeWorkspace;
+  }
+
+  // If everything failed, show error
+  log.info(colors.red.bold("No workspace given and no default set."));
   return Deno.exit(-1);
 }
 

cli/src/core/specific_items.ts

@@ -0,0 +1,282 @@
+import { minimatch } from "../../deps.ts";
+import { getCurrentGitBranch, isGitRepository } from "../utils/git.ts";
+import { isFileResource } from "../utils/utils.ts";
+import { SyncOptions } from "./conf.ts";
+import { TRIGGER_TYPES } from "../types.ts";
+
+export interface SpecificItemsConfig {
+  variables?: string[];
+  resources?: string[];
+  triggers?: string[];
+}
+
+// Define all branch-specific file types (computed lazily)
+function getBranchSpecificTypes() {
+  return {
+    variable: '.variable.yaml',
+    resource: '.resource.yaml',
+    // Generate trigger patterns from the list
+    ...Object.fromEntries(
+      TRIGGER_TYPES.map(t => [`${t}_trigger`, `.${t}_trigger.yaml`])
+    )
+  } as const;
+}
+
+/**
+ * Check if a path ends with any trigger type
+ */
+function isTriggerFile(path: string): boolean {
+  return TRIGGER_TYPES.some(type => path.endsWith(`.${type}_trigger.yaml`));
+}
+
+/**
+ * Extract the file type suffix from a path
+ */
+function getFileTypeSuffix(path: string): string | null {
+  for (const [_, suffix] of Object.entries(getBranchSpecificTypes())) {
+    if (path.endsWith(suffix)) {
+      return suffix;
+    }
+  }
+
+  const resourceFileMatch = path.match(/(\\.resource\\.file\\..+)$/);
+  if (resourceFileMatch) {
+    return resourceFileMatch[1];
+  }
+
+  return null;
+}
+
+/**
+ * Build regex pattern for all supported yaml file types
+ */
+function buildYamlTypePattern(): string {
+  const basicTypes = ['variable', 'resource'];
+  const triggerTypes = TRIGGER_TYPES.map(t => `${t}_trigger`);
+  return `((${basicTypes.join('|')})|(${triggerTypes.join('|')}))`;
+}
+
+/**
+ * Get the specific items configuration for the current git branch
+ * Merges commonSpecificItems with branch-specific specificItems
+ */
+export function getSpecificItemsForCurrentBranch(config: SyncOptions): SpecificItemsConfig | undefined {
+  if (!isGitRepository() || !config.gitBranches) {
+    return undefined;
+  }
+
+  const currentBranch = getCurrentGitBranch();
+  if (!currentBranch) {
+    return undefined;
+  }
+
+  const commonItems = config.gitBranches.commonSpecificItems;
+  const branchItems = config.gitBranches[currentBranch]?.specificItems;
+
+  // If neither common nor branch-specific items exist, return undefined
+  if (!commonItems && !branchItems) {
+    return undefined;
+  }
+
+  // Merge common and branch-specific items
+  const merged: SpecificItemsConfig = {};
+
+  // Add common items
+  if (commonItems?.variables) {
+    merged.variables = [...commonItems.variables];
+  }
+  if (commonItems?.resources) {
+    merged.resources = [...commonItems.resources];
+  }
+  if (commonItems?.triggers) {
+    merged.triggers = [...commonItems.triggers];
+  }
+
+  // Add branch-specific items (extending common items)
+  if (branchItems?.variables) {
+    merged.variables = [...(merged.variables || []), ...branchItems.variables];
+  }
+  if (branchItems?.resources) {
+    merged.resources = [...(merged.resources || []), ...branchItems.resources];
+  }
+  if (branchItems?.triggers) {
+    merged.triggers = [...(merged.triggers || []), ...branchItems.triggers];
+  }
+
+  return merged;
+}
+
+/**
+ * Check if a path matches any of the patterns in the given list
+ */
+function matchesPatterns(path: string, patterns: string[]): boolean {
+  return patterns.some(pattern => minimatch(path, pattern));
+}
+
+/**
+ * Check if a file path should be treated as branch-specific
+ */
+export function isSpecificItem(path: string, specificItems: SpecificItemsConfig | undefined): boolean {
+  if (!specificItems) {
+    return false;
+  }
+
+  // Determine the item type from the file path
+  if (path.endsWith('.variable.yaml')) {
+    return specificItems.variables ? matchesPatterns(path, specificItems.variables) : false;
+  }
+
+  if (path.endsWith('.resource.yaml')) {
+    return specificItems.resources ? matchesPatterns(path, specificItems.resources) : false;
+  }
+
+  // Check for any trigger type
+  if (isTriggerFile(path)) {
+    return specificItems.triggers ? matchesPatterns(path, specificItems.triggers) : false;
+  }
+
+  // Check for resource files using the standard detection function
+  if (isFileResource(path)) {
+    // Extract the base path without the file extension to match against patterns
+    const basePathMatch = path.match(/^(.+?)\.resource\.file\./);
+    if (basePathMatch && specificItems.resources) {
+      const basePath = basePathMatch[1] + '.resource.yaml';
+      return matchesPatterns(basePath, specificItems.resources);
+    }
+  }
+
+  return false;
+}
+
+/**
+ * Convert a base path to a branch-specific path
+ */
+export function toBranchSpecificPath(basePath: string, branchName: string): string {
+  // Check for resource file pattern (e.g., .resource.file.ini)
+  const resourceFileMatch = basePath.match(/^(.+?)(\.resource\.file\..+)$/);
+
+  let extension: string;
+  let pathWithoutExtension: string;
+
+  if (resourceFileMatch) {
+    // Handle resource files
+    extension = resourceFileMatch[2];
+    pathWithoutExtension = resourceFileMatch[1];
+  } else {
+    const suffix = getFileTypeSuffix(basePath);
+    if (!suffix) {
+      return basePath;
+    }
+    extension = suffix;
+    pathWithoutExtension = basePath.substring(0, basePath.length - extension.length);
+  }
+
+  // Sanitize branch name to be filesystem-safe
+  const sanitizedBranchName = branchName.replace(/[\/\\:*?"<>|.]/g, '_');
+
+  // Warn about potential collisions if sanitization occurred
+  if (sanitizedBranchName !== branchName) {
+    console.warn(`Warning: Branch name "${branchName}" contains filesystem-unsafe characters (/ \\ : * ? " < > | .) and was sanitized to "${sanitizedBranchName}". This may cause collisions with other similarly named branches.`);
+  }
+
+  return `${pathWithoutExtension}.${sanitizedBranchName}${extension}`;
+}
+
+/**
+ * Convert a branch-specific path back to a base path
+ */
+export function fromBranchSpecificPath(branchSpecificPath: string, branchName: string): string {
+  // Sanitize branch name the same way as in toBranchSpecificPath
+  const sanitizedBranchName = branchName.replace(/[\/\\:*?"<>|.]/g, '_');
+  const escapedBranchName = sanitizedBranchName.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
+
+  // Check for resource file pattern first
+  const resourceFilePattern = new RegExp(`\\.${escapedBranchName}(\\.resource\\.file\\..+)$`);
+  const resourceFileMatch = branchSpecificPath.match(resourceFilePattern);
+
+  if (resourceFileMatch) {
+    const extension = resourceFileMatch[1];
+    const pathWithoutBranchAndExtension = branchSpecificPath.substring(
+      0,
+      branchSpecificPath.length - `.${sanitizedBranchName}${extension}`.length
+    );
+    return `${pathWithoutBranchAndExtension}${extension}`;
+  }
+
+  const yamlPattern = new RegExp(`\\.${escapedBranchName}(\\.${buildYamlTypePattern()}\\.yaml)$`);
+  const yamlMatch = branchSpecificPath.match(yamlPattern);
+
+  if (!yamlMatch) {
+    return branchSpecificPath; // Return unchanged if not a branch-specific path
+  }
+
+  const extension = yamlMatch[1];
+  const pathWithoutBranchAndExtension = branchSpecificPath.substring(
+    0,
+    branchSpecificPath.length - `.${sanitizedBranchName}${extension}`.length
+  );
+
+  return `${pathWithoutBranchAndExtension}${extension}`;
+}
+
+/**
+ * Get the branch-specific path for the current branch if the item should be branch-specific
+ */
+export function getBranchSpecificPath(
+  basePath: string,
+  specificItems: SpecificItemsConfig | undefined
+): string | undefined {
+  if (!isGitRepository() || !specificItems) {
+    return undefined;
+  }
+
+  const currentBranch = getCurrentGitBranch();
+  if (!currentBranch) {
+    return undefined;
+  }
+
+  if (isSpecificItem(basePath, specificItems)) {
+    return toBranchSpecificPath(basePath, currentBranch);
+  }
+
+  return undefined;
+}
+
+// Cache for compiled regex patterns to avoid recompilation
+const branchPatternCache = new Map<string, RegExp>();
+
+/**
+ * Check if a path is a branch-specific file for the current branch
+ */
+export function isCurrentBranchFile(path: string): boolean {
+  if (!isGitRepository()) {
+    return false;
+  }
+
+  const currentBranch = getCurrentGitBranch();
+  if (!currentBranch) {
+    return false;
+  }
+
+  // Sanitize branch name to match what would be used in file naming
+  const sanitizedBranchName = currentBranch.replace(/[\/\\:*?"<>|.]/g, '_');
+  const escapedBranchName = sanitizedBranchName.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
+
+  // Use cached pattern or create and cache new one
+  let pattern = branchPatternCache.get(currentBranch);
+  if (!pattern) {
+    pattern = new RegExp(`\\.${escapedBranchName}\\.${buildYamlTypePattern()}\\.yaml$|\\.${escapedBranchName}\\.resource\\.file\\..+$`);
+    branchPatternCache.set(currentBranch, pattern);
+  }
+
+  return pattern.test(path);
+}
+
+/**
+ * Check if a path is a branch-specific file for ANY branch (not necessarily current)
+ * Used to identify and skip files from other branches during sync operations
+ */
+export function isBranchSpecificFile(path: string): boolean {
+  const yamlTypePattern = buildYamlTypePattern();
+  return new RegExp(`\\.[^.]+\\.${yamlTypePattern}\\.yaml$|\\.[^.]+\\.resource\\.file\\..+$`).test(path);
+}

cli/src/main.ts

@@ -68,7 +68,7 @@ export {
 //   }
 // });
 
-export const VERSION = "1.530.0";
+export const VERSION = "1.537.0";
 
 const command = new Command()
   .name("wmill")

cli/src/types.ts

@@ -45,6 +45,17 @@ export interface DifferenceChange {
 
 export type Difference = DifferenceCreate | DifferenceRemove | DifferenceChange;
 
+export const TRIGGER_TYPES = [
+  'http',
+  'websocket',
+  'kafka',
+  'nats',
+  'postgres',
+  'mqtt',
+  'sqs',
+  'gcp'
+] as const;
+
 export type GlobalOptions = {
   baseUrl: string | undefined;
   workspace: string | undefined;
@@ -111,6 +122,17 @@ export function showConflict(path: string, local: string, remote: string) {
   log.info("\n");
 }
 
+/**
+ * Pushes an object to the workspace server based on its type
+ * @param workspace - The workspace ID to push to
+ * @param p - The server path (base path for branch-specific items)
+ * @param befObj - The previous object state (for updates)
+ * @param newObj - The new object state to push
+ * @param plainSecrets - Whether to store secrets in plain text
+ * @param alreadySynced - Array to track already synced items
+ * @param message - Optional commit/update message
+ * @param originalLocalPath - The original local file path (used for branch-specific resource file resolution)
+ */
 export async function pushObj(
   workspace: string,
   p: string,
@@ -118,7 +140,8 @@ export async function pushObj(
   newObj: any,
   plainSecrets: boolean,
   alreadySynced: string[],
-  message?: string
+  message?: string,
+  originalLocalPath?: string
 ) {
   const typeEnding = getTypeStrFromPath(p);
 
@@ -135,7 +158,7 @@ export async function pushObj(
   } else if (typeEnding === "resource") {
     if (!alreadySynced.includes(p)) {
       alreadySynced.push(p);
-      await pushResource(workspace, p, befObj, newObj);
+      await pushResource(workspace, p, befObj, newObj, originalLocalPath || p);
     }
   } else if (typeEnding === "resource-type") {
     await pushResourceType(workspace, p, befObj, newObj);

cli/src/utils/utils.ts

@@ -136,10 +136,12 @@ export function sleep(ms: number) {
 
 export function isFileResource(path: string): boolean {
   const splitPath = path.split(".");
+
+  // Check for pattern: *.resource.file.* (handles both base and branch-specific)
   return (
     splitPath.length >= 4 &&
-    splitPath[1] == "resource" &&
-    splitPath[2] == "file"
+    splitPath[splitPath.length - 3] == "resource" &&
+    splitPath[splitPath.length - 2] == "file"
   );
 }
 

frontend/README_DEV.md

@@ -15,44 +15,45 @@ That's it! You are ready to go.
 > Using **direnv** is highly recommended, since it can load shell automatically based on your CWD. It also can give you hints.
 
 ### Development
+
 ```bash
 # enter a dev shell containing all necessary packages. `direnv allow` if direnv is installed.
-nix develop 
+nix develop
 ## or ignore if you have `direnv`
 
 # Start db (if not started already)
-sudo docker compose up db -d 
+./start-dev-db.sh
 
 # run the frontend.
 wm
 
 # In an other shell:
 #
-nix develop 
+nix develop
 ## or ignore if you have `direnv`
 
 cd backend
 # You don't need to install anything extra. All dependencies are already in place!
-cargo run --features all_languages 
+cargo run --features all_languages
 ```
 
 The default proxy is setup to use the local backend: <http://localhost:8000>.
 
-### wm-* Commands
+### wm-\* Commands
 
 Nix shell provides you with several helper commands prefixed with `wm-`
 
 ```bash
 # Start minio server (implements S3)
-wm-minio 
-# Note: You will need access to EE private repo in order to compile, don't forget "enterprise" and "parquet" freatures as well. 
+wm-minio
+# Note: You will need access to EE private repo in order to compile, don't forget "enterprise" and "parquet" freatures as well.
 
 # Generate keys for local dev.
-wm-minio-keys 
+wm-minio-keys
 # Minio data as well as generated keys are stored in `backend/.minio-data`
 ```
 
-You can read about all others commands individually in [flake.nix](../flake.nix). 
+You can read about all others commands individually in [flake.nix](../flake.nix).
 
 ### dev.nu
 
@@ -106,7 +107,7 @@ REMOTE=http://localhost REMOTE_LSP=http://localhost npm run dev
 
 Sometimes it is important to build docker image for your branch locally. It is crucial part of testing, since local environment may differ from the containerized one.
 
-That's why we provide [docker/dev.nu](../docker/dev.nu). It is helper that can build images locally and execute them. 
+That's why we provide [docker/dev.nu](../docker/dev.nu). It is helper that can build images locally and execute them.
 
 it can build the image and run on local repository.
 
@@ -150,7 +151,7 @@ If you develop wasm parser for new language you can also pass `--wasm-pkg <langu
 In the root folder:
 
 ```bash
-docker-compose up db
+./start-dev-db.sh
 ```
 
 In the backend folder:
@@ -218,6 +219,7 @@ nix flake update # update the lock file.
 ```
 
 Some cargo dependencies use fixed git revisions, which are also fixed in `flake.nix`:
+
 ```nix
 outputHashes = {
   "php-parser-rs-0.1.3" = "sha256-ZeI3KgUPmtjlRfq6eAYveqt8Ay35gwj6B9iOQRjQa9A=";
@@ -226,6 +228,7 @@ outputHashes = {
 ```
 
 When updating a revision, replace the incorrect `sha256` with `pkgs.lib.fakeHash`:
+
 ```diff
 -            "php-parser-rs-0.1.3" = "sha256-ZeI3KgUPmtjlRfq6eAYveqt8Ay35gwj6B9iOQRjQa9A=";
 +            "php-parser-rs-0.1.3" = pkgs.lib.fakeHash;

frontend/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "windmill-components",
-	"version": "1.530.0",
+	"version": "1.537.0",
 	"scripts": {
 		"dev": "vite dev",
 		"build": "vite build",
@@ -78,13 +78,13 @@
 	"type": "module",
 	"dependencies": {
 		"@aws-crypto/sha256-js": "^4.0.0",
-		"@codingame/monaco-vscode-configuration-service-override": "~19.1.4",
-		"@codingame/monaco-vscode-editor-api": "~19.1.4",
-		"@codingame/monaco-vscode-standalone-css-language-features": "~19.1.4",
-		"@codingame/monaco-vscode-standalone-html-language-features": "~19.1.4",
-		"@codingame/monaco-vscode-standalone-json-language-features": "~19.1.4",
-		"@codingame/monaco-vscode-standalone-languages": "~19.1.4",
-		"@codingame/monaco-vscode-standalone-typescript-language-features": "~19.1.4",
+		"@codingame/monaco-vscode-configuration-service-override": "~20.2.1",
+		"@codingame/monaco-vscode-editor-api": "~20.2.1",
+		"@codingame/monaco-vscode-standalone-css-language-features": "~20.2.1",
+		"@codingame/monaco-vscode-standalone-html-language-features": "~20.2.1",
+		"@codingame/monaco-vscode-standalone-json-language-features": "~20.2.1",
+		"@codingame/monaco-vscode-standalone-languages": "~20.2.1",
+		"@codingame/monaco-vscode-standalone-typescript-language-features": "~20.2.1",
 		"@json2csv/plainjs": "^7.0.6",
 		"@leeoniya/ufuzzy": "^1.0.8",
 		"@popperjs/core": "^2.11.6",
@@ -117,10 +117,10 @@
 		"lru-cache": "^11.1.0",
 		"lucide-svelte": "^0.399.0",
 		"minimatch": "^10.0.1",
-		"monaco-editor": "npm:@codingame/monaco-vscode-editor-api@~19.1.4",
-		"monaco-editor-wrapper": "6.10.0",
+		"monaco-editor": "npm:@codingame/monaco-vscode-editor-api@~20.2.1",
+		"monaco-editor-wrapper": "6.12.0",
 		"monaco-graphql": "=1.6.0",
-		"monaco-languageclient": "9.9.0",
+		"monaco-languageclient": "9.11.0",
 		"monaco-vim": "^0.4.1",
 		"ol": "^7.4.0",
 		"openai": "^4.87.1",
@@ -137,10 +137,10 @@
 		"svelte-infinite-loading": "^1.4.0",
 		"svelte-tiny-virtual-list": "^2.0.5",
 		"tailwind-merge": "^1.13.2",
-		"vscode": "npm:@codingame/monaco-vscode-extension-api@~19.1.4",
+		"vscode": "npm:@codingame/monaco-vscode-extension-api@~20.2.1",
 		"vscode-languageclient": "~9.0.1",
 		"vscode-uri": "~3.1.0",
-		"vscode-ws-jsonrpc": "~3.4.0",
+		"vscode-ws-jsonrpc": "~3.5.0",
 		"windmill-parser-wasm-csharp": "1.510.1",
 		"windmill-parser-wasm-go": "1.510.1",
 		"windmill-parser-wasm-java": "1.510.1",

frontend/src/lib/components/ArgInput.svelte

@@ -6,6 +6,7 @@
 	import {
 		setInputCat as computeInputCat,
 		debounce,
+		emptySchema,
 		emptyString,
 		getSchemaFromProperties,
 		type DynamicSelect
@@ -42,6 +43,8 @@
 	import { safeSelectItems } from './select/utils.svelte'
 	import S3ArgInput from './common/fileUpload/S3ArgInput.svelte'
 	import { base } from '$lib/base'
+	import { workspaceStore } from '$lib/stores'
+	import { getJsonSchemaFromResource } from './schema/jsonSchemaResource.svelte'
 
 	interface Props {
 		label?: string
@@ -658,6 +661,47 @@
 				{appPath}
 				{computeS3ForceViewerPolicies}
 			/>
+		{:else if inputCat == 'object' && format == 'json-schema'}
+			{#await import('$lib/components/EditableSchemaForm.svelte')}
+				<Loader2 class="animate-spin" />
+			{:then Module}
+				<Module.default
+					bind:schema={
+						() =>
+							value && typeof value === 'object' && !Array.isArray(value) ? value : emptySchema(),
+						(v) => {
+							value = v
+						}
+					}
+					isFlowInput
+					editTab="inputEditor"
+					noPreview
+					addPropertyInEditorTab
+				/>
+			{/await}
+		{:else if inputCat == 'object' && format?.startsWith('jsonschema-')}
+			{#await getJsonSchemaFromResource(format.substring('jsonschema-'.length), workspace ?? $workspaceStore ?? '')}
+				<Loader2 class="animate-spin" />
+			{:then schema}
+				{#if !schema || !schema.properties}
+					{#await import('$lib/components/JsonEditor.svelte')}
+						<Loader2 class="animate-spin" />
+					{:then Module}
+						<Module.default code={JSON.stringify(value, null, 2)} bind:value />
+					{/await}
+				{:else}
+					<div class="py-4 pr-2 pl-6 border rounded-md w-full">
+						<SchemaForm
+							{onlyMaskPassword}
+							{disablePortal}
+							{disabled}
+							{prettifyHeader}
+							{schema}
+							bind:args={value}
+						/>
+					</div>
+				{/if}
+			{/await}
 		{:else if inputCat == 'list' && !isListJson}
 			<div class="w-full flex gap-4">
 				<div class="w-full">
@@ -893,11 +937,21 @@
 							selected={oneOfSelected}
 							on:selected={({ detail }) => {
 								oneOfSelected = detail
-								const prevValueKeys = Object.keys(
+								const selectedObjProperties =
 									oneOf?.find((o) => o.title == detail)?.properties ?? {}
-								)
+								const newValueKeys = Object.keys(selectedObjProperties)
 								const toKeep = {}
-								for (const key of prevValueKeys) {
+								for (const key of newValueKeys) {
+									// Check if there is a select (enum) in the newly selected oneOf and if the current value is not in the enum, skip it
+									if (
+										!['kind', 'label'].includes(key) &&
+										selectedObjProperties[key]?.enum &&
+										value &&
+										value[key] !== undefined &&
+										!selectedObjProperties[key].enum.includes(value[key])
+									) {
+										continue
+									}
 									toKeep[key] = value[key]
 								}
 								const tagKey = oneOf.find((o) => Object.keys(o.properties ?? {}).includes('kind'))

frontend/src/lib/components/Auth0Setting.svelte

@@ -1,4 +1,5 @@
 <script lang="ts">
+	import { untrack } from 'svelte'
 	import CollapseLink from './CollapseLink.svelte'
 	import IconedResourceType from './IconedResourceType.svelte'
 	import Toggle from './Toggle.svelte'
@@ -32,8 +33,12 @@
 		}
 	}
 	let enabled = $derived(value != undefined)
-	$effect(() => {
-		changeDomain(value?.['domain'], value?.['custom'])
+	let lastValues = { domain: undefined, custom: undefined }
+	$effect.pre(() => {
+		if (value?.['domain'] != lastValues.domain || value?.['custom'] != lastValues.custom) {
+			lastValues = { domain: value?.['domain'], custom: value?.['custom'] }
+			untrack(() => changeDomain(value?.['domain'], value?.['custom']))
+		}
 	})
 </script>
 

frontend/src/lib/components/Dev.svelte

@@ -48,7 +48,7 @@
 	import type { FlowPropPickerConfig, PropPickerContext } from './prop_picker'
 	import type { PickableProperties } from './flows/previousResults'
 	import { Triggers } from './triggers/triggers.svelte'
-	import { TestSteps } from './flows/testSteps.svelte'
+	import { StepsInputArgs } from './flows/stepsInputArgs.svelte'
 	import { ModulesTestStates } from './modulesTest.svelte'
 	import type { GraphModuleState } from './graph'
 
@@ -457,7 +457,7 @@
 	const scriptEditorDrawer = writable(undefined)
 	const moving = writable<{ id: string } | undefined>(undefined)
 	const history = initHistory(flowStore.val)
-	const testSteps = new TestSteps()
+	const stepsInputArgs = new StepsInputArgs()
 	const selectedIdStore = writable('settings-metadata')
 	const triggersCount = writable<TriggersCount | undefined>(undefined)
 	const modulesTestStates = new ModulesTestStates((moduleId) => {
@@ -481,7 +481,7 @@
 		pathStore: writable(''),
 		flowStateStore,
 		flowStore,
-		testSteps,
+		stepsInputArgs,
 		saveDraft: () => {},
 		initialPathStore: writable(''),
 		fakeInitialPath: '',
@@ -806,7 +806,7 @@
 								noEditor
 								on:applyArgs={(ev) => {
 									if (ev.detail.kind === 'preprocessor') {
-										testSteps.setStepArgs('preprocessor', ev.detail.args ?? {})
+										stepsInputArgs.setStepArgs('preprocessor', ev.detail.args ?? {})
 										$selectedIdStore = 'preprocessor'
 									} else {
 										previewArgsStore.val = ev.detail.args ?? {}
@@ -818,6 +818,7 @@
 								isOwner={flowPreviewContent?.getIsOwner()}
 								{suspendStatus}
 								onOpenDetails={flowPreviewButtons?.openPreview}
+								previewOpen={flowPreviewButtons?.getPreviewOpen()}
 							/>
 						{/key}
 					</Pane>

frontend/src/lib/components/DiffDrawer.svelte

@@ -212,7 +212,7 @@
 										<Module.default
 											open={true}
 											automaticLayout
-											class="h-full"
+											className="h-full"
 											defaultLang={lang}
 											defaultModifiedLang={data.current.lang}
 											defaultOriginal={content}
@@ -227,7 +227,7 @@
 										<Module.default
 											open={true}
 											automaticLayout
-											class="h-full"
+											className="h-full"
 											defaultLang="yaml"
 											defaultOriginal={metadata}
 											defaultModified={data.current.metadata}

frontend/src/lib/components/DiffEditor.svelte

@@ -1,6 +1,6 @@
 <script lang="ts">
 	import { BROWSER } from 'esm-env'
-	import { createEventDispatcher, onMount } from 'svelte'
+	import { onMount } from 'svelte'
 
 	import '@codingame/monaco-vscode-standalone-languages'
 	import '@codingame/monaco-vscode-standalone-json-language-features'
@@ -10,24 +10,49 @@
 	import { initializeVscode } from './vscode'
 	import EditorTheme from './EditorTheme.svelte'
 	import Button from '$lib/components/common/button/Button.svelte'
+	import { twMerge } from 'tailwind-merge'
+	import type { ButtonType } from './common'
 
 	const SIDE_BY_SIDE_MIN_WIDTH = 700
 
-	export let automaticLayout = true
-	export let fixedOverflowWidgets = true
-	export let defaultLang: string | undefined = undefined
-	export let defaultModifiedLang: string | undefined = undefined
-	export let defaultOriginal: string | undefined = undefined
-	export let defaultModified: string | undefined = undefined
-	export let readOnly = false
-	export let showButtons = false
-	export let showHistoryButton: boolean = true
+	export interface ButtonProp {
+		text: string
+		color?: ButtonType.Color
+		onClick: () => void
+	}
 
-	let diffEditor: meditor.IStandaloneDiffEditor | undefined
-	let diffDivEl: HTMLDivElement | null = null
-	let editorWidth: number = SIDE_BY_SIDE_MIN_WIDTH
+	interface Props {
+		open?: boolean
+		className?: string
+		automaticLayout?: boolean
+		fixedOverflowWidgets?: boolean
+		defaultLang?: string
+		defaultModifiedLang?: string
+		defaultOriginal?: string
+		defaultModified?: string
+		readOnly?: boolean
+		buttons?: ButtonProp[]
+		modifiedModel?: meditor.ITextModel
+	}
+
+	let {
+		open = false,
+		className = '',
+		automaticLayout = true,
+		fixedOverflowWidgets = true,
+		defaultLang,
+		defaultModifiedLang,
+		defaultOriginal = undefined,
+		defaultModified = undefined,
+		readOnly = false,
+		buttons = [],
+		modifiedModel
+	}: Props = $props()
+
+	let diffEditor: meditor.IStandaloneDiffEditor | undefined = $state(undefined)
+	let diffDivEl: HTMLDivElement | null = $state(null)
+	let editorWidth: number = $state(SIDE_BY_SIDE_MIN_WIDTH)
 
-	export let open = false
 	async function loadDiffEditor() {
 		await initializeVscode()
 
@@ -49,11 +74,8 @@
 			lineNumbersMinChars: 2,
 			scrollbar: { alwaysConsumeMouseWheel: false }
 		})
-		if (
-			defaultOriginal !== undefined &&
-			defaultModified !== undefined &&
-			defaultLang !== undefined
-		) {
+
+		if (defaultLang !== undefined) {
 			setupModel(defaultLang, defaultOriginal, defaultModified, defaultModifiedLang)
 		}
 	}
@@ -64,16 +86,12 @@
 		modified?: string,
 		modifiedLang?: string
 	) {
+		const o = meditor.createModel(original ?? '', lang)
+		const m = modifiedModel ?? meditor.createModel(modified ?? '', modifiedLang ?? lang)
 		diffEditor?.setModel({
-			original: meditor.createModel('', lang),
-			modified: meditor.createModel('', modifiedLang ?? lang)
+			original: o,
+			modified: m
 		})
-		if (original) {
-			setOriginal(original)
-		}
-		if (modified) {
-			setModified(modified)
-		}
 	}
 
 	export function setOriginal(code: string) {
@@ -90,6 +108,15 @@
 		defaultModified = code
 	}
 
+	export function setModifiedModel(model: meditor.ITextModel) {
+		const curr = diffEditor?.getModel()
+		if (!curr) return
+		diffEditor?.setModel({
+			original: curr.original,
+			modified: model
+		})
+	}
+
 	export function getModified(): string {
 		return diffEditor?.getModel()?.modified.getValue() ?? ''
 	}
@@ -105,9 +132,15 @@
 		diffEditor?.updateOptions({ renderSideBySide: editorWidth >= SIDE_BY_SIDE_MIN_WIDTH })
 	}
 
-	$: onWidthChange(editorWidth)
+	$effect(() => {
+		if (open && diffDivEl) {
+			loadDiffEditor()
+		}
+	})
 
-	$: open && diffDivEl && loadDiffEditor()
+	$effect(() => {
+		onWidthChange(editorWidth)
+	})
 
 	onMount(() => {
 		if (BROWSER) {
@@ -116,32 +149,24 @@
 			}
 		}
 	})
-
-	const dispatch = createEventDispatcher<{
-		hideDiffMode: void
-		seeHistory: void
-	}>()
 </script>
 
 {#if open}
 	<EditorTheme />
 	<div
 		bind:this={diffDivEl}
-		class="{$$props.class} editor nonmain-editor"
+		class={twMerge('editor nonmain-editor', className)}
 		bind:clientWidth={editorWidth}
 	></div>
-	{#if showButtons}
+	{#if buttons.length > 0}
 		<div
 			class="absolute flex flex-row gap-2 bottom-10 left-1/2 z-10 -translate-x-1/2 rounded-md p-1 w-full justify-center"
 		>
-			{#if showHistoryButton}
-				<Button on:click={() => dispatch('seeHistory')} variant="contained" size="sm"
-					>See changes history</Button
+			{#each buttons as button}
+				<Button on:click={button.onClick} variant="contained" size="sm" color={button.color}
+					>{button.text}</Button
 				>
-			{/if}
-			<Button on:click={() => dispatch('hideDiffMode')} variant="contained" size="sm" color="red"
-				>Quit diff mode</Button
-			>
+			{/each}
 		</div>
 	{/if}
 {/if}

frontend/src/lib/components/DisplayResult.svelte

@@ -928,7 +928,7 @@
 										>
 									</button>
 								{:else if !s3object?.disable_download}
-									<FileDownload {s3object} />
+									<FileDownload {workspaceId} {s3object} {appPath} />
 								{:else}
 									<div class="flex text-secondary pt-2">{s3object?.s3} (download disabled)</div>
 								{/if}

frontend/src/lib/components/EditableSchemaForm.svelte

@@ -29,6 +29,7 @@
 	import type { EditableSchemaFormUi } from '$lib/components/custom_ui'
 	import Section from '$lib/components/Section.svelte'
 	import Editor from './Editor.svelte'
+	import AddPropertyV2 from './schema/AddPropertyV2.svelte'
 
 	// export let openEditTab: () => void = () => {}
 	const dispatch = createEventDispatcher()
@@ -69,6 +70,7 @@
 		dynSelectCode?: string | undefined
 		dynSelectLang?: ScriptLang | undefined
 		showDynSelectOpt?: boolean
+		addPropertyInEditorTab?: boolean
 		openEditTab?: import('svelte').Snippet
 		addProperty?: import('svelte').Snippet
 		runButton?: import('svelte').Snippet
@@ -104,6 +106,7 @@
 		dynSelectCode = $bindable(),
 		dynSelectLang = $bindable(),
 		showDynSelectOpt = false,
+		addPropertyInEditorTab = false,
 		openEditTab,
 		addProperty,
 		runButton,
@@ -509,22 +512,31 @@
 				{:else}
 					<!-- WIP -->
 					{#if jsonEnabled && customUi?.jsonOnly != true}
-						<div class="w-full p-3 flex justify-end">
-							<Toggle
-								bind:checked={jsonView}
-								label="JSON View"
-								size="xs"
-								options={{
-									right: 'JSON editor',
-									rightTooltip:
-										'Arguments can be edited either using the wizard, or by editing their JSON Schema.'
-								}}
-								lightMode
-								on:change={() => {
-									schemaString = JSON.stringify(schema, null, '\t')
-									editor?.setCode(schemaString)
-								}}
-							/>
+						<div class="w-full p-3 flex gap-4 justify-end items-center">
+							{#if addPropertyInEditorTab}
+								<AddPropertyV2 bind:schema on:change>
+									{#snippet trigger()}
+										<Button color="light" size="xs" iconOnly startIcon={{ icon: Plus }} />
+									{/snippet}
+								</AddPropertyV2>
+							{/if}
+							<div class="shrink-0">
+								<Toggle
+									bind:checked={jsonView}
+									label="JSON View"
+									size="xs"
+									options={{
+										right: 'JSON editor',
+										rightTooltip:
+											'Arguments can be edited either using the wizard, or by editing their JSON Schema.'
+									}}
+									lightMode
+									on:change={() => {
+										schemaString = JSON.stringify(schema, null, '\t')
+										editor?.setCode(schemaString)
+									}}
+								/>
+							</div>
 						</div>
 					{/if}
 
@@ -655,7 +667,6 @@
 																					const isS3 = v == 'S3'
 																					const isOneOf = v == 'oneOf'
 																					const isDynSelect = v == 'dynselect'
-
 																					const emptyProperty = {
 																						contentEncoding: undefined,
 																						enum_: undefined,

frontend/src/lib/components/Editor.svelte

@@ -150,7 +150,7 @@
 	import { writable } from 'svelte/store'
 	import { formatResourceTypes } from './copilot/chat/script/core'
 	import FakeMonacoPlaceHolder from './FakeMonacoPlaceHolder.svelte'
-	import { editorPositionMap, readFieldsRecursively } from '$lib/utils'
+	import { editorPositionMap } from '$lib/utils'
 	import { extToLang, langToExt } from '$lib/editorLangUtils'
 	import { aiChatManager } from './copilot/chat/AIChatManager.svelte'
 	import type { Selection } from 'monaco-editor'
@@ -184,6 +184,7 @@
 		loadAsync?: boolean
 		key?: string | undefined
 		class?: string | undefined
+		moduleId?: string
 	}
 
 	let {
@@ -209,7 +210,8 @@
 		changeTimeout = 500,
 		loadAsync = false,
 		key = undefined,
-		class: clazz = undefined
+		class: clazz = undefined,
+		moduleId = undefined
 	}: Props = $props()
 
 	$effect.pre(() => {
@@ -542,9 +544,7 @@
 
 	let sqlSchemaCompletor: IDisposable | undefined = undefined
 
-	async function updateSchema() {
-		const newSchemaRes = lang === 'graphql' ? args?.api : args?.database
-
+	async function updateSchema(newSchemaRes: string | undefined) {
 		if (typeof newSchemaRes === 'string') {
 			const resourcePath = newSchemaRes.replace('$res:', '')
 			dbSchema = $dbSchemas[resourcePath]
@@ -674,7 +674,21 @@
 	let selectedCode = $state('')
 
 	export function reviewAndApplyCode(code: string, applyAll: boolean = false) {
-		aiChatEditorHandler?.reviewAndApply(code, applyAll)
+		aiChatEditorHandler?.reviewChanges(code, { applyAll, mode: 'apply' })
+	}
+
+	export function reviewAppliedCode(
+		originalCode: string,
+		opts?: { onFinishedReview?: () => void }
+	) {
+		aiChatEditorHandler?.reviewChanges(originalCode, {
+			mode: 'revert',
+			onFinishedReview: opts?.onFinishedReview
+		})
+	}
+
+	export function getAiChatEditorHandler() {
+		return aiChatEditorHandler
 	}
 
 	function addChatHandler(editor: meditor.IStandaloneCodeEditor) {
@@ -1235,7 +1249,13 @@
 
 		try {
 			editor = meditor.create(divEl as HTMLDivElement, {
-				...editorConfig(code ?? '', lang, automaticLayout, fixedOverflowWidgets, $relativeLineNumbers),
+				...editorConfig(
+					code ?? '',
+					lang,
+					automaticLayout,
+					fixedOverflowWidgets,
+					$relativeLineNumbers
+				),
 				model,
 				fontSize: !small ? 14 : 12,
 				lineNumbersMinChars,
@@ -1328,7 +1348,8 @@
 					aiChatManager.addSelectedLinesToContext(
 						selectedLines,
 						selection.startLineNumber,
-						selection.endLineNumber
+						selection.endLineNumber,
+						moduleId
 					)
 				} else {
 					aiChatManager.toggleOpen()
@@ -1598,10 +1619,14 @@
 			? untrack(() => addSqlTypeCompletions())
 			: sqlTypeCompletor?.dispose()
 	})
+
+	let lastArg = undefined
 	$effect(() => {
-		console.log('updating schema', lang, $dbSchemas)
-		readFieldsRecursively(args)
-		lang && $dbSchemas && untrack(() => updateSchema())
+		let newArg = lang === 'graphql' ? args?.api : args?.database
+		if (newArg !== lastArg) {
+			lastArg = newArg
+			$dbSchemas && untrack(() => updateSchema(newArg))
+		}
 	})
 	$effect(() => {
 		console.log('updating db schema completions', dbSchema, lang)
@@ -1654,7 +1679,7 @@
 		files && model && untrack(() => onFileChanges())
 	})
 	$effect(() => {
-		editor?.updateOptions({ 
+		editor?.updateOptions({
 			lineNumbers: $relativeLineNumbers ? 'relative' : 'on'
 		})
 	})
@@ -1675,10 +1700,20 @@
 {#if $reviewingChanges}
 	<GlobalReviewButtons
 		onAcceptAll={() => {
-			aiChatEditorHandler?.acceptAll()
+			const mode = aiChatEditorHandler?.getReviewMode?.()
+			if (mode === 'revert') {
+				aiChatEditorHandler?.keepAll()
+			} else {
+				aiChatEditorHandler?.acceptAll()
+			}
 		}}
 		onRejectAll={() => {
-			aiChatEditorHandler?.rejectAll()
+			const mode = aiChatEditorHandler?.getReviewMode?.()
+			if (mode === 'revert') {
+				aiChatEditorHandler?.revertAll()
+			} else {
+				aiChatEditorHandler?.rejectAll()
+			}
 		}}
 	/>
 {/if}

frontend/src/lib/components/EditorBar.svelte

@@ -82,6 +82,7 @@
 		showHistoryDrawer?: boolean
 		right?: import('svelte').Snippet
 		openAiChat?: boolean
+		moduleId?: string
 	}
 
 	let {
@@ -105,7 +106,8 @@
 		diffMode = false,
 		showHistoryDrawer = $bindable(false),
 		right,
-		openAiChat = false
+		openAiChat = false,
+		moduleId = undefined
 	}: Props = $props()
 
 	let contextualVariablePicker: ItemPicker | undefined = $state()
@@ -964,7 +966,7 @@ JsonNode ${windmillPathToCamelCaseName(path)} = JsonNode.Parse(await client.GetS
 
 			{#if customUi?.aiGen != false}
 				{#if openAiChat}
-					<FlowInlineScriptAiButton />
+					<FlowInlineScriptAiButton {moduleId} />
 				{:else}
 					<ScriptGen {editor} {diffEditor} {lang} {iconOnly} {args} />
 				{/if}

frontend/src/lib/components/FieldHeader.svelte

@@ -40,7 +40,7 @@
 	{/if}
 
 	{#if displayType}
-		{#if format && !format.startsWith('resource')}
+		{#if format && !format.startsWith('resource') && !format.startsWith('jsonschema-')}
 			<span class="text-xs italic ml-2 text-tertiary dark:text-indigo-400">
 				{format}
 			</span>

frontend/src/lib/components/FlowBuilder.svelte

@@ -77,7 +77,7 @@
 	} from './triggers/utils'
 	import DraftTriggersConfirmationModal from './common/confirmationModal/DraftTriggersConfirmationModal.svelte'
 	import { Triggers } from './triggers/triggers.svelte'
-	import { TestSteps } from './flows/testSteps.svelte'
+	import { StepsInputArgs } from './flows/stepsInputArgs.svelte'
 	import { aiChatManager } from './copilot/chat/AIChatManager.svelte'
 	import type { GraphModuleState } from './graph'
 	import {
@@ -571,7 +571,7 @@
 		payloadData: undefined
 	})
 
-	const testSteps = new TestSteps()
+	const stepsInputArgs = new StepsInputArgs()
 
 	function select(selectedId: string) {
 		selectedIdStore.set(selectedId)
@@ -592,7 +592,7 @@
 		flowStateStore,
 		flowStore,
 		pathStore,
-		testSteps,
+		stepsInputArgs,
 		saveDraft,
 		initialPathStore,
 		fakeInitialPath,
@@ -1129,6 +1129,8 @@
 						bind:this={flowPreviewButtons}
 						{loading}
 						onRunPreview={() => {
+							// Reset manually edited args inputs when running a preview
+							stepsInputArgs.resetManuallyEditedArgs()
 							modulesTestStates.hideJobsInGraph()
 							localModuleStates = {}
 							showJobStatus = true
@@ -1170,7 +1172,7 @@
 					{newFlow}
 					on:applyArgs={(ev) => {
 						if (ev.detail.kind === 'preprocessor') {
-							testSteps.setStepArgs('preprocessor', ev.detail.args ?? {})
+							stepsInputArgs.setStepArgs('preprocessor', ev.detail.args ?? {})
 							$selectedIdStore = 'preprocessor'
 						}
 					}}
@@ -1218,6 +1220,7 @@
 						delete modulesTestStates.states[id]
 					}}
 					{flowHasChanged}
+					previewOpen={flowPreviewButtons?.getPreviewOpen()}
 				/>
 			{:else}
 				<CenteredPage>Loading...</CenteredPage>

frontend/src/lib/components/FlowPreviewContent.svelte

@@ -140,7 +140,7 @@
 			jobId = await runFlowPreview(args, newFlow, $pathStore, restartedFrom)
 			isRunning = true
 			if (inputSelected) {
-				savedArgs = previewArgs.val
+				savedArgs = $state.snapshot(previewArgs.val)
 				inputSelected = undefined
 			}
 			onRunPreview?.()
@@ -166,7 +166,7 @@
 					if (preventEscape) {
 						selectInput(undefined)
 						event.preventDefault()
-						event.stopPropagation
+						event.stopPropagation()
 					}
 					break
 			}
@@ -506,7 +506,7 @@
 									schema={flowStore.val.schema}
 									bind:args={previewArgs.val}
 									on:change={() => {
-										savedArgs = previewArgs.val
+										savedArgs = $state.snapshot(previewArgs.val)
 									}}
 									bind:isValid
 									helperScript={flowStore.val.schema?.['x-windmill-dyn-select-code'] &&

frontend/src/lib/components/FlowStatusViewer.svelte

@@ -110,6 +110,9 @@
 	let updateGlobalRefresh = (moduleId: string, updateFn: (clear, root) => Promise<void>) => {
 		globalRefreshes[moduleId] = [...(globalRefreshes[moduleId] ?? []), updateFn]
 	}
+
+	let storedToolCallJobs: Record<string, Job> = $state({})
+	let toolCallIndicesToLoad: string[] = $state([])
 </script>
 
 <FlowStatusViewerInner
@@ -141,4 +144,30 @@
 	isNodeSelected={true}
 	{refreshGlobal}
 	{updateGlobalRefresh}
+	toolCallStore={{
+		getStoredToolCallJob: (storeKey: string) => storedToolCallJobs[storeKey],
+		setStoredToolCallJob: (storeKey: string, job: Job) => {
+			storedToolCallJobs[storeKey] = job
+		},
+		getLocalToolCallJobs: (prefix: string) => {
+			// we return a map from tool call index to job
+			// to do so, we filter the storedToolCallJobs object by the prefix and we make sure what's left in the key is a tool call index: 2 part of format agentModuleId-toolCallIndex
+			// and not a further nested tool call index
+			return Object.fromEntries(
+				Object.entries(storedToolCallJobs)
+					.filter(
+						([key]) => key.startsWith(prefix) && key.replace(prefix, '').split('-').length === 2
+					)
+					.map(([key, job]) => [Number(key.replace(prefix, '').split('-').pop()), job])
+			)
+		},
+		isToolCallToBeLoaded: (storeKey: string) => {
+			return toolCallIndicesToLoad.includes(storeKey)
+		},
+		addToolCallToLoad: (storeKey: string) => {
+			if (!toolCallIndicesToLoad.includes(storeKey)) {
+				toolCallIndicesToLoad.push(storeKey)
+			}
+		}
+	}}
 />

frontend/src/lib/components/FlowStatusViewerInner.svelte

@@ -33,6 +33,7 @@
 	import { deepEqual } from 'fast-equals'
 	import FlowTimeline from './FlowTimeline.svelte'
 	import { dfs } from './flows/dfs'
+	import { dfs as dfsPreviousResults } from '$lib/components/flows/previousResults'
 	import Alert from './common/alert/Alert.svelte'
 	import FlowGraphViewerStep from './FlowGraphViewerStep.svelte'
 	import FlowGraphV2 from './graph/FlowGraphV2.svelte'
@@ -116,6 +117,13 @@
 		onStart?: () => void
 		onJobsLoaded?: ({ job, force }: { job: Job; force: boolean }) => void
 		onDone?: ({ job }: { job: CompletedJob }) => void
+		toolCallStore?: {
+			getStoredToolCallJob: (storeKey: string) => Job | undefined
+			setStoredToolCallJob: (storeKey: string, job: Job) => void
+			getLocalToolCallJobs: (prefix: string) => Record<number, Job>
+			isToolCallToBeLoaded: (storeKey: string) => boolean
+			addToolCallToLoad: (storeKey: string) => void
+		}
 	}
 
 	let {
@@ -155,7 +163,8 @@
 		loadExtraLogs = undefined,
 		onStart = undefined,
 		onJobsLoaded = undefined,
-		onDone = undefined
+		onDone = undefined,
+		toolCallStore
 	}: Props = $props()
 
 	let getTopModuleStates = $derived(topModuleStates ?? localModuleStates)
@@ -913,9 +922,7 @@
 
 	let storedListJobs: Record<number, Job> = $state({})
 
-	let storedToolCallJobs: Record<number, Job> = $state({})
-	let selectedToolCall: number | undefined = $state(undefined)
-	let toolCallIndicesToLoad: number[] = $state([])
+	let selectedToolCall: string | undefined = $state(undefined)
 
 	let wrapperHeight: number = $state(0)
 
@@ -950,8 +957,10 @@
 					let nprefix = buildPrefix(prefix, oid)
 					return fms
 						? rec(
-								dfs(fms, (x) =>
-									x.id.startsWith('subflow:') ? x.id : buildSubflowKey(x.id, nprefix)
+								dfs(
+									fms,
+									(x) => (x.id.startsWith('subflow:') ? x.id : buildSubflowKey(x.id, nprefix)),
+									{ skipToolNodes: true }
 								),
 								nprefix
 							)
@@ -1009,6 +1018,11 @@
 				selectedForLoopSetManually: false
 			})
 		}
+		if (selectedNode?.startsWith(AI_TOOL_CALL_PREFIX)) {
+			const [, agentModuleId, toolCallIndex, _] = selectedNode.split('-')
+			const parentLoopsPrefix = getParentLoopsPrefix(agentModuleId)
+			toolCallStore?.addToolCallToLoad(parentLoopsPrefix + agentModuleId + '-' + toolCallIndex)
+		}
 	}
 
 	$effect(() => {
@@ -1039,6 +1053,29 @@
 	let animateLogsTab = $state(false)
 
 	let noLogs = $derived(graphTabOpen && !isNodeSelected)
+
+	/**
+	 * Returns a string like "forloopmodid1-{iter1}-forloopmodid2-{iter2}-forloopmodid3-{iter3}-"
+	 * that can be used to prefix tool call store keys for nested tool calls.
+	 */
+	function getParentLoopsPrefix(modId: string) {
+		if (job?.raw_flow) {
+			const indices: string[] = []
+			const parents = dfsPreviousResults(modId, { value: job?.raw_flow, summary: '' }, true)
+			for (const parent of parents) {
+				if (parent.value.type === 'forloopflow' || parent.value.type === 'whileloopflow') {
+					const state = localModuleStates[parent.id]
+					if (state?.selectedForloopIndex !== undefined) {
+						indices.push(parent.id + '-' + state.selectedForloopIndex.toString())
+					}
+				}
+			}
+			indices.reverse()
+			return indices.length > 0 ? indices.join('-') + '-' : ''
+		}
+
+		return ''
+	}
 </script>
 
 <JobLoader workspaceOverride={workspaceId} {noLogs} noCode bind:this={jobLoader} />
@@ -1173,6 +1210,10 @@
 							{@const forloopIsSelected =
 								forloop_selected == loopJobId ||
 								(innerModule?.type != 'forloopflow' && innerModule?.type != 'whileloopflow')}
+							{@const forLoopStoreKeyPrefix =
+								innerModule?.type == 'forloopflow' || innerModule?.type == 'whileloopflow'
+									? (flowJobIds?.moduleId ?? '') + '-' + j + '-'
+									: ''}
 							<!-- <LogId id={loopJobId} /> -->
 							<div class="border p-6" class:hidden={forloop_selected != loopJobId}>
 								<FlowStatusViewerInner
@@ -1207,6 +1248,18 @@
 									graphTabOpen={selected == 'graph' && graphTabOpen}
 									isNodeSelected={forloop_selected == loopJobId}
 									{globalIterationBounds}
+									toolCallStore={{
+										getStoredToolCallJob: (storeKey: string) =>
+											toolCallStore?.getStoredToolCallJob(forLoopStoreKeyPrefix + storeKey),
+										setStoredToolCallJob: (storeKey: string, job: Job) =>
+											toolCallStore?.setStoredToolCallJob(forLoopStoreKeyPrefix + storeKey, job),
+										getLocalToolCallJobs: (prefix: string) =>
+											toolCallStore?.getLocalToolCallJobs(forLoopStoreKeyPrefix + prefix) ?? {},
+										addToolCallToLoad: (storeKey: string) =>
+											toolCallStore?.addToolCallToLoad(forLoopStoreKeyPrefix + storeKey),
+										isToolCallToBeLoaded: (storeKey: string) =>
+											toolCallStore?.isToolCallToBeLoaded(forLoopStoreKeyPrefix + storeKey) ?? false
+									}}
 								/>
 							</div>
 						{/if}
@@ -1366,12 +1419,17 @@
 										graphTabOpen={selected == 'graph' && graphTabOpen}
 										isNodeSelected={localModuleStates?.[selectedNode ?? '']?.job_id == mod.job}
 										{globalIterationBounds}
+										{toolCallStore}
 									/>
-									{#if mod.agent_actions && mod.agent_actions.length > 0}
+									{#if mod.agent_actions && mod.agent_actions.length > 0 && mod.id}
+										{@const storeKeyPrefix = getParentLoopsPrefix(mod.id)}
 										{#each mod.agent_actions as agentAction, j}
-											{#if agentAction.type === 'tool_call' && mod.id}
+											{#if agentAction.type === 'tool_call'}
 												{@const toolCallId = getToolCallId(j, mod.id, agentAction.module_id)}
-												{@const isSelected = selectedToolCall === j}
+												{@const localToolCallKey = mod.id + '-' + j}
+												{@const storeKey = storeKeyPrefix + localToolCallKey}
+												{@const storedToolCallJob = toolCallStore?.getStoredToolCallJob(storeKey)}
+												{@const isSelected = localToolCallKey === selectedToolCall}
 												<Button
 													variant={isSelected ? 'contained' : 'border'}
 													color={mod.agent_actions_success?.[j] === false
@@ -1381,10 +1439,10 @@
 															: 'light'}
 													btnClasses="w-full flex justify-start"
 													on:click={async () => {
-														if (selectedToolCall == j) {
+														if (isSelected) {
 															selectedToolCall = undefined
 														} else {
-															selectedToolCall = j
+															selectedToolCall = localToolCallKey
 														}
 													}}
 													endIcon={{
@@ -1396,7 +1454,7 @@
 														Tool call: {agentAction.function_name}
 													</span>
 												</Button>
-												{#if isSelected || storedToolCallJobs[j] || toolCallIndicesToLoad.includes(j)}
+												{#if isSelected || storedToolCallJob || toolCallStore?.isToolCallToBeLoaded(storeKey)}
 													<FlowStatusViewerInner
 														topModuleStates={getTopModuleStates}
 														{refreshGlobal}
@@ -1414,11 +1472,11 @@
 														{subflowParentsDurationStatuses}
 														{isSelectedBranch}
 														jobId={agentAction.job_id}
-														job={storedToolCallJobs[j]}
-														initialJob={storedToolCallJobs[j]}
+														job={storedToolCallJob}
+														initialJob={storedToolCallJob}
 														{reducedPolling}
 														onJobsLoaded={({ job, force }) => {
-															storedToolCallJobs[j] = job
+															toolCallStore?.setStoredToolCallJob(storeKey, job)
 															onJobsLoadedInner({ id: toolCallId } as FlowStatusModule, job, force)
 														}}
 														loadExtraLogs={(logs) => {
@@ -1509,11 +1567,11 @@
 										stepDetail = mod
 										selectedNode = e
 										if (e.startsWith(AI_TOOL_CALL_PREFIX)) {
-											const [_prefix, _agentModuleId, j, _toolModuleId] = e.split('-')
+											const [_prefix, agentModuleId, j, _toolModuleId] = e.split('-')
+											const parentLoopsPrefix = getParentLoopsPrefix(agentModuleId)
 											const jIdx = Number(j)
-											if (!toolCallIndicesToLoad.includes(jIdx)) {
-												toolCallIndicesToLoad.push(jIdx)
-											}
+											const storeKey = parentLoopsPrefix + agentModuleId + '-' + jIdx
+											toolCallStore?.addToolCallToLoad(storeKey)
 										}
 									}
 								} else {
@@ -1603,6 +1661,7 @@
 											stepDetail && typeof stepDetail !== 'string' ? stepDetail : undefined}
 										{@const agentTools =
 											module && module.value.type === 'aiagent' ? module.value.tools : undefined}
+										{@const parentLoopsPrefix = getParentLoopsPrefix(module?.id ?? '')}
 										{#if node.flow_jobs_results}
 											<span class="pl-1 text-tertiary"
 												>Result of step as collection of all subflows</span
@@ -1667,7 +1726,7 @@
 											logs={node.logs}
 											downloadLogs={!hideDownloadLogs}
 											aiAgentStatus={agentTools &&
-											node.job_id &&
+											node?.job_id &&
 											(node.type === 'Success' || node.type === 'Failure')
 												? {
 														tools: agentTools,
@@ -1679,9 +1738,14 @@
 															success: node.type === 'Success',
 															type: 'CompletedJob'
 														},
-														storedToolCallJobs,
+														storedToolCallJobs: module
+															? toolCallStore?.getLocalToolCallJobs(parentLoopsPrefix)
+															: undefined,
 														onToolJobLoaded: (job, idx) => {
-															storedToolCallJobs[idx] = job
+															if (module) {
+																const storeKey = parentLoopsPrefix + module.id + '-' + idx
+																toolCallStore?.setStoredToolCallJob(storeKey, job)
+															}
 														}
 													}
 												: undefined}

frontend/src/lib/components/FlowWrapper.svelte

@@ -5,12 +5,14 @@
 
 	let {
 		flowStore: oldFlowStore,
+		flowStateStore: oldFlowStateStore,
 		disableAi,
 		light,
 		...props
 	}: FlowBuilderProps & { light?: boolean } = $props()
 
 	let flowStore = $state(oldFlowStore)
+	let flowStateStore = $state(oldFlowStateStore)
 
 	let trialRender = $state(true)
 
@@ -24,7 +26,7 @@
 {#if trialRender}
 	<AiChatLayout noPadding={true} {disableAi}>
 		{#if light}<div class="bg-red-500 absolute z-10">Trial version</div>{/if}
-		<FlowBuilder {flowStore} {disableAi} {...props} />
+		<FlowBuilder {flowStore} {flowStateStore} {disableAi} {...props} />
 	</AiChatLayout>
 {:else}
 	<div class="flex flex-col items-center justify-center h-screen">

frontend/src/lib/components/ModulePreview.svelte

@@ -18,6 +18,7 @@
 		noEditor?: boolean
 		scriptProgress?: any
 		focusArg?: string
+		onJobDone?: () => void
 	}
 
 	let {
@@ -28,7 +29,8 @@
 		testIsLoading = $bindable(false),
 		noEditor = false,
 		scriptProgress = $bindable(undefined),
-		focusArg = undefined
+		focusArg = undefined,
+		onJobDone
 	}: Props = $props()
 
 	const { flowStore } = getContext<FlowEditorContext>('FlowEditorContext')
@@ -46,6 +48,7 @@
 	bind:testIsLoading
 	bind:scriptProgress
 	bind:this={moduleTest}
+	{onJobDone}
 />
 
 <div class="p-4">

frontend/src/lib/components/ModulePreviewForm.svelte

@@ -1,11 +1,11 @@
 <script lang="ts">
 	import type { Schema } from '$lib/common'
 
-	import { allTrue, sendUserToast } from '$lib/utils'
+	import { allTrue } from '$lib/utils'
 	import { RefreshCw } from 'lucide-svelte'
 	import ArgInput from './ArgInput.svelte'
 	import { Button } from './common'
-	import { getContext, onMount, untrack } from 'svelte'
+	import { getContext, untrack } from 'svelte'
 	import type { FlowEditorContext } from './flows/types'
 	import { evalValue } from './flows/utils'
 	import type { FlowModule } from '$lib/gen'
@@ -32,7 +32,7 @@
 		focusArg = undefined
 	}: Props = $props()
 
-	const { testSteps, flowStateStore, flowStore, previewArgs } =
+	const { stepsInputArgs, flowStateStore, flowStore, previewArgs } =
 		getContext<FlowEditorContext>('FlowEditorContext')
 
 	let inputCheck: { [id: string]: boolean } = $state({})
@@ -45,12 +45,12 @@
 		let lkeys = Object.keys(schema?.properties ?? {})
 		if (schema?.properties && JSON.stringify(lkeys) != JSON.stringify(keys)) {
 			keys = lkeys
-			untrack(() => testSteps?.removeExtraKey(mod.id, keys))
+			untrack(() => stepsInputArgs?.removeExtraKey(mod.id, keys))
 		}
 	})
 
 	function plugIt(argName: string) {
-		testSteps?.setEvaluatedStepArg(
+		stepsInputArgs?.setEvaluatedStepArg(
 			mod.id,
 			argName,
 			$state.snapshot(evalValue(argName, mod, pickableProperties, true))
@@ -98,70 +98,78 @@
 
 	loadResourceTypes()
 
-	onMount(() => {
-		if (!testSteps) {
-			sendUserToast('testSteps module not initialized. Preview will not work.', true)
+	let initialized = $state(false)
+
+	$effect.pre(() => {
+		if (!initialized) {
+			if (stepsInputArgs) {
+				stepsInputArgs?.updateStepArgs(mod.id, flowStateStore.val, flowStore?.val, previewArgs?.val)
+				initialized = true
+			}
 		}
-		testSteps?.updateStepArgs(mod.id, flowStateStore.val, flowStore?.val, previewArgs?.val)
 	})
 </script>
 
 <div class="w-full pt-2" data-popover>
-	{#if keys.length > 0}
-		{#each keys as argName, i (argName)}
-			{#if Object.keys(schema.properties ?? {}).includes(argName)}
-				<div
-					class={twMerge(
-						'flex gap-2',
-						animateArg === argName && 'animate-pulse ring-2 ring-offset-2 ring-blue-500 rounded'
-					)}
-					data-arg={argName}
-				>
-					{#if schema?.properties?.[argName]}
-						<ArgInput
-							{resourceTypes}
-							minW={false}
-							autofocus={autofocus && !focusArg && i == 0}
-							label={argName}
-							description={schema.properties[argName].description}
-							bind:value={
-								() => testSteps?.getStepInputArgs(mod.id, argName),
-								(v) => testSteps?.setStepInputArgs(mod.id, argName, v)
-							}
-							type={schema.properties[argName].type}
-							oneOf={schema.properties[argName].oneOf}
-							required={schema?.required?.includes(argName)}
-							pattern={schema.properties[argName].pattern}
-							bind:editor={editor[argName]}
-							bind:valid={inputCheck[argName]}
-							defaultValue={schema.properties[argName].default}
-							enum_={schema.properties[argName].enum}
-							format={schema.properties[argName].format}
-							contentEncoding={schema.properties[argName].contentEncoding}
-							properties={schema.properties[argName].properties}
-							nestedRequired={schema.properties[argName].required}
-							itemsType={schema.properties[argName].items}
-							extra={schema.properties[argName]}
-							nullable={schema.properties[argName].nullable}
-							title={schema.properties[argName].title}
-							placeholder={schema.properties[argName].placeholder}
-						/>
-					{/if}
-					{#if testSteps?.isArgManuallySet(mod.id, argName)}
-						<div class="pt-6 mt-0.5">
-							<Button
-								on:click={() => {
-									plugIt(argName)
-								}}
-								size="sm"
-								variant="border"
-								color="light"
-								title="Re-evaluate input step"><RefreshCw size={14} /></Button
-							>
-						</div>
-					{/if}
-				</div>
-			{/if}
-		{/each}
+	{#if initialized}
+		{#if keys.length > 0}
+			{#each keys as argName, i (argName)}
+				{#if Object.keys(schema.properties ?? {}).includes(argName)}
+					<div
+						class={twMerge(
+							'flex gap-2',
+							animateArg === argName && 'animate-pulse ring-2 ring-offset-2 ring-blue-500 rounded'
+						)}
+						data-arg={argName}
+					>
+						{#if schema?.properties?.[argName]}
+							<ArgInput
+								{resourceTypes}
+								minW={false}
+								autofocus={autofocus && !focusArg && i == 0}
+								label={argName}
+								description={schema.properties[argName].description}
+								bind:value={
+									() => stepsInputArgs?.getStepInputArgs(mod.id, argName),
+									(v) => stepsInputArgs?.setStepInputArgs(mod.id, argName, v)
+								}
+								type={schema.properties[argName].type}
+								oneOf={schema.properties[argName].oneOf}
+								required={schema?.required?.includes(argName)}
+								pattern={schema.properties[argName].pattern}
+								bind:editor={editor[argName]}
+								bind:valid={inputCheck[argName]}
+								defaultValue={schema.properties[argName].default}
+								enum_={schema.properties[argName].enum}
+								format={schema.properties[argName].format}
+								contentEncoding={schema.properties[argName].contentEncoding}
+								properties={schema.properties[argName].properties}
+								nestedRequired={schema.properties[argName].required}
+								itemsType={schema.properties[argName].items}
+								extra={schema.properties[argName]}
+								nullable={schema.properties[argName].nullable}
+								title={schema.properties[argName].title}
+								placeholder={schema.properties[argName].placeholder}
+							/>
+						{/if}
+						{#if stepsInputArgs?.isArgManuallySet(mod.id, argName)}
+							<div class="pt-6 mt-0.5">
+								<Button
+									on:click={() => {
+										plugIt(argName)
+									}}
+									size="sm"
+									variant="border"
+									color="light"
+									title="Re-evaluate input step"><RefreshCw size={14} /></Button
+								>
+							</div>
+						{/if}
+					</div>
+				{/if}
+			{/each}
+		{/if}
+	{:else}
+		<div class="text-center text-sm text-tertiary"> Loading test step arguments... </div>
 	{/if}
 </div>

frontend/src/lib/components/ModulePreviewResultViewer.svelte

@@ -7,7 +7,7 @@
 	import { type Script, type Job, type FlowModule } from '$lib/gen'
 	import OutputPickerInner from '$lib/components/flows/propPicker/OutputPickerInner.svelte'
 	import { Pane, Splitpanes } from 'svelte-splitpanes'
-	import type { FlowEditorContext } from './flows/types'
+	import type { FlowEditorContext, OutputViewerJob } from './flows/types'
 	import { getContext } from 'svelte'
 	import { getStringError } from './copilot/chat/utils'
 	import AiAgentLogViewer from './AIAgentLogViewer.svelte'
@@ -17,9 +17,8 @@
 		editor: Editor | undefined
 		diffEditor: DiffEditor | undefined
 		loopStatus?: { type: 'inside' | 'self'; flow: 'forloopflow' | 'whileloopflow' } | undefined
-		lastJob?: Job | undefined
+		testJob?: Job & { result_stream?: string }
 		scriptProgress?: number | undefined
-		testJob?: Job | undefined
 		mod: FlowModule
 		testIsLoading?: boolean
 		disableMock?: boolean
@@ -34,7 +33,6 @@
 		editor,
 		diffEditor,
 		loopStatus = undefined,
-		lastJob = undefined,
 		scriptProgress = $bindable(undefined),
 		testJob = undefined,
 		mod,
@@ -46,21 +44,20 @@
 		tagLabel = undefined
 	}: Props = $props()
 
-	const { testSteps } = getContext<FlowEditorContext>('FlowEditorContext')
+	const { stepsInputArgs } = getContext<FlowEditorContext>('FlowEditorContext')
 
-	let selectedJob: Job | undefined = $state(undefined)
-	let preview: 'mock' | 'job' | undefined = $state(undefined)
 	let jobProgressReset: () => void = $state(() => {})
 
-	$effect(() => {
-		if (preview != undefined && testJob) {
-			preview = undefined
-		}
-	})
-
-	let forceJson = $state(false)
+	let outputPickerInner: OutputPickerInner | undefined = $state(undefined)
+	export function getOutputPickerInner() {
+		return outputPickerInner
+	}
 
+	const selectedJob: OutputViewerJob = $derived.by(
+		() => outputPickerInner?.getSelectedJob?.() ?? undefined
+	)
 	const logJob = $derived(testJob ?? selectedJob)
+	const preview = $derived.by(() => outputPickerInner?.getPreview?.())
 </script>
 
 <Splitpanes horizontal>
@@ -75,7 +72,6 @@
 		{/if}
 
 		<OutputPickerInner
-			{lastJob}
 			{testJob}
 			fullResult
 			moduleId={mod.id}
@@ -83,24 +79,22 @@
 			getLogs
 			{onUpdateMock}
 			mock={mod.mock}
-			bind:forceJson
-			bind:selectedJob
 			isLoading={testIsLoading || loadingJob}
-			bind:preview
 			path={`path` in mod.value ? mod.value.path : ''}
 			{loopStatus}
 			{disableMock}
 			{disableHistory}
+			bind:this={outputPickerInner}
 		>
 			{#snippet copilot_fix()}
-				{#if lang && editor && diffEditor && testSteps.getStepArgs(mod.id) && selectedJob?.type === 'CompletedJob' && !selectedJob.success && getStringError(selectedJob.result)}
+				{#if lang && editor && diffEditor && stepsInputArgs.getStepArgs(mod.id) && selectedJob?.type === 'CompletedJob' && !selectedJob.success && getStringError(selectedJob.result)}
 					<ScriptFix {lang} />
 				{/if}
 			{/snippet}
 		</OutputPickerInner>
 	</Pane>
 	<Pane size={35} minSize={10}>
-		{#if (mod.mock?.enabled && preview != 'job') || preview == 'mock'}
+		{#if (mod.mock?.enabled && preview !== 'job' && testJob?.type !== 'QueuedJob') || preview === 'mock'}
 			<LogViewer
 				small
 				content={undefined}
@@ -124,9 +118,9 @@
 				jobId={logJob?.id}
 				duration={logJob?.['duration_ms']}
 				mem={logJob?.['mem_peak']}
-				content={logJob?.logs}
+				content={logJob?.['logs']}
 				isLoading={(testIsLoading && logJob?.['running'] == false) || loadingJob}
-				tag={logJob?.tag}
+				tag={logJob?.['tag']}
 				{tagLabel}
 			/>
 		{/if}

frontend/src/lib/components/ModuleTest.svelte

@@ -14,6 +14,7 @@
 		testIsLoading?: boolean
 		noEditor?: boolean
 		scriptProgress?: any
+		onJobDone?: () => void
 	}
 
 	let {
@@ -21,10 +22,11 @@
 		testJob = $bindable(undefined),
 		testIsLoading = $bindable(false),
 		noEditor = false,
-		scriptProgress = $bindable(undefined)
+		scriptProgress = $bindable(undefined),
+		onJobDone
 	}: Props = $props()
 
-	const { flowStore, flowStateStore, pathStore, testSteps, previewArgs, modulesTestStates } =
+	const { flowStore, flowStateStore, pathStore, stepsInputArgs, previewArgs, modulesTestStates } =
 		getContext<FlowEditorContext>('FlowEditorContext')
 
 	let jobLoader: JobLoader | undefined = $state(undefined)
@@ -32,12 +34,12 @@
 	let stepHistoryLoader = getStepHistoryLoaderContext()
 
 	export function runTestWithStepArgs() {
-		runTest(testSteps.getStepArgs(mod.id))
+		runTest(stepsInputArgs.getStepArgs(mod.id))
 	}
 
 	export function loadArgsAndRunTest() {
-		testSteps?.updateStepArgs(mod.id, flowStateStore.val, flowStore?.val, previewArgs?.val)
-		runTest(testSteps.getStepArgs(mod.id))
+		stepsInputArgs?.updateStepArgs(mod.id, flowStateStore.val, flowStore?.val, previewArgs?.val)
+		runTest(stepsInputArgs.getStepArgs(mod.id))
 	}
 
 	export async function runTest(args: any) {
@@ -138,6 +140,7 @@
 		if (modulesTestStates.states[mod.id]) {
 			modulesTestStates.states[mod.id].testJob = testJob
 		}
+		onJobDone?.()
 	}
 
 	export function cancelJob() {
@@ -179,7 +182,10 @@
 			}
 		}
 	}
-	bind:job={modulesTestStates.states[mod.id].testJob}
+	bind:job={
+		() => modulesTestStates.states[mod.id]?.testJob,
+		(v) => modulesTestStates.states[mod.id] && (modulesTestStates.states[mod.id].testJob = v)
+	}
 	loadPlaceholderJobOnStart={{
 		type: 'QueuedJob',
 		id: '',

frontend/src/lib/components/OktaSetting.svelte

@@ -1,4 +1,5 @@
 <script lang="ts">
+	import { untrack } from 'svelte'
 	import CollapseLink from './CollapseLink.svelte'
 	import IconedResourceType from './IconedResourceType.svelte'
 	import Toggle from './Toggle.svelte'
@@ -12,6 +13,8 @@
 
 	let { value = $bindable() }: Props = $props()
 
+	let lastValues = { domain: undefined, custom: undefined }
+
 	function changeDomain(domain, custom) {
 		if (value) {
 			let baseUrl = custom ? `https://${domain}` : `https://${domain}.okta.com`
@@ -32,8 +35,11 @@
 		}
 	}
 	let enabled = $derived(value != undefined)
-	$effect(() => {
-		changeDomain(value?.['domain'], value?.['custom'])
+	$effect.pre(() => {
+		if (value?.['domain'] != lastValues.domain || value?.['custom'] != lastValues.custom) {
+			lastValues = { domain: value?.['domain'], custom: value?.['custom'] }
+			untrack(() => changeDomain(value?.['domain'], value?.['custom']))
+		}
 	})
 </script>
 

frontend/src/lib/components/ResourceEditor.svelte

@@ -22,6 +22,7 @@
 	import TestTriggerConnection from './triggers/TestTriggerConnection.svelte'
 	import GitHubAppIntegration from './GitHubAppIntegration.svelte'
 	import Button from './common/button/Button.svelte'
+	import { clearJsonSchemaResourceCache } from './schema/jsonSchemaResource.svelte'
 
 	interface Props {
 		canSave?: boolean
@@ -94,6 +95,9 @@
 				path: resourceToEdit.path,
 				requestBody: { path, value: args, description }
 			})
+			if (resourceToEdit.resource_type === 'json_schema') {
+				clearJsonSchemaResourceCache(resourceToEdit.path, $workspaceStore!)
+			}
 			sendUserToast(`Updated resource at ${path}`)
 			dispatch('refresh', path)
 		} else {