Merge origin/main

2026-10-13 08:02:47 +00:00 · 2026-09-26 22:50:43 +02:00
parent e1f7b7ba09 649c43e7c1
commit 9cd7026acb
55 changed files with 2019 additions and 779 deletions
@@ -1,3 +1,3 @@
 {
-  ".": "1.817.0"
+  ".": "1.818.0"
 }
@@ -1,5 +1,68 @@
 # Changelog

+## [1.818.0](https://github.com/windmill-labs/windmill/compare/v1.817.0...v1.818.0) (2026-09-25)
+
+
+### Features
+
+* add an instance-wide accent color setting with sidebar tint ([#11335](https://github.com/windmill-labs/windmill/issues/11335)) ([1fd729a](https://github.com/windmill-labs/windmill/commit/1fd729ac1aaec1e6a3af0d6db5fc1d6e1db6fa62))
+* add an options field to the postgresql resource ([#11223](https://github.com/windmill-labs/windmill/issues/11223)) ([d3d5392](https://github.com/windmill-labs/windmill/commit/d3d539291716baf5807a724d60d338cd3c4fe521))
+* **ai-chat:** show native script edit diffs ([#11278](https://github.com/windmill-labs/windmill/issues/11278)) ([1de54ee](https://github.com/windmill-labs/windmill/commit/1de54eeea4c4820e7291d9648a44ef3c86979c5a))
+* alert on and optionally cancel jobs stuck on unserved tags ([#11354](https://github.com/windmill-labs/windmill/issues/11354)) ([da866c5](https://github.com/windmill-labs/windmill/commit/da866c5eff77f66e45b8e3941166c695536762f5))
+* **bedrock:** add OIDC role assumption as a fourth auth mode ([#10936](https://github.com/windmill-labs/windmill/issues/10936)) ([e14da5c](https://github.com/windmill-labs/windmill/commit/e14da5c6bc8256bfe1d71ddecb886c3efc994ae4))
+* chat with a saved agent from the agent editor ([#11292](https://github.com/windmill-labs/windmill/issues/11292)) ([d8aaa73](https://github.com/windmill-labs/windmill/commit/d8aaa73573c46a8a23f6c6850d38438ad395752c))
+* **cli:** include extra paths in codebase digest for monorepo imports ([#11327](https://github.com/windmill-labs/windmill/issues/11327)) ([9b46b21](https://github.com/windmill-labs/windmill/commit/9b46b21f517a67867912092c619675d91a197313))
+* collapse the fork members setting and show its state in a badge ([#11220](https://github.com/windmill-labs/windmill/issues/11220)) ([3bd89e9](https://github.com/windmill-labs/windmill/commit/3bd89e92d8f563b564a3d2e619749f377a7d6982))
+* count enabled service accounts as operator seats ([#11297](https://github.com/windmill-labs/windmill/issues/11297)) ([318f899](https://github.com/windmill-labs/windmill/commit/318f89960e5594278d2aaad68f40db36b96918e7))
+* **db-manager:** add a schema diagram view ([#11030](https://github.com/windmill-labs/windmill/issues/11030)) ([7dec3d9](https://github.com/windmill-labs/windmill/commit/7dec3d97717406d078499ea6509cdb00c89c3480))
+* filter ai session tools to the user's workspace capabilities ([#10719](https://github.com/windmill-labs/windmill/issues/10719)) ([6cd70a8](https://github.com/windmill-labs/windmill/commit/6cd70a8e08cf4869cb52ef554bc89d5d13fadf5f))
+* infer a script's schema when a deploy (e.g. MCP) sends none ([#11339](https://github.com/windmill-labs/windmill/issues/11339)) ([d18d704](https://github.com/windmill-labs/windmill/commit/d18d7043df6536e10b22ee9f3df77e926a0460b4))
+* let a workspace withdraw operator schedule and trigger writes ([#11226](https://github.com/windmill-labs/windmill/issues/11226)) ([9a1c6e5](https://github.com/windmill-labs/windmill/commit/9a1c6e508127579dee4c1fae5c3257930e0b8aa1))
+* mount session list pages in process instead of iframes ([#11289](https://github.com/windmill-labs/windmill/issues/11289)) ([ebb3048](https://github.com/windmill-labs/windmill/commit/ebb3048ca0b6cb9f3c9130ac0871581f3c2ed307))
+* per-folder AI chat instructions via ai_instruction resources ([#11325](https://github.com/windmill-labs/windmill/issues/11325)) ([80903c9](https://github.com/windmill-labs/windmill/commit/80903c9a64125763e7514c09618e1def21a5bf97))
+* refresh MCP tools when scripts and flows change ([#11337](https://github.com/windmill-labs/windmill/issues/11337)) ([0bac766](https://github.com/windmill-labs/windmill/commit/0bac766756cf9ebcb615fe20d5a69c09c823c725))
+* render a tool's web search result as the session web search card ([#11313](https://github.com/windmill-labs/windmill/issues/11313)) ([fa0fc24](https://github.com/windmill-labs/windmill/commit/fa0fc242691ed3997c14e821b6030166223c0da0))
+* rework the db manager: native grid, tabs, sql editor, joined columns ([#11340](https://github.com/windmill-labs/windmill/issues/11340)) ([4fd7d62](https://github.com/windmill-labs/windmill/commit/4fd7d62bd0cf4970a6b155c4a3afac5fb1b79d16))
+* rework the git sync setup into a guided flow ([#11308](https://github.com/windmill-labs/windmill/issues/11308)) ([504d01a](https://github.com/windmill-labs/windmill/commit/504d01a5631e33b3a2604ef401ab8b73e4a55395))
+* schedule hub scripts ([#11330](https://github.com/windmill-labs/windmill/issues/11330)) ([733c119](https://github.com/windmill-labs/windmill/commit/733c119fd9464a951e157a8d18ec9d474430e910))
+* seed SCIM usernames from nickName ([#11346](https://github.com/windmill-labs/windmill/issues/11346)) ([9100ab9](https://github.com/windmill-labs/windmill/commit/9100ab954ab5b40cf6b9d7320e0fe435186f7eb4))
+* show restart from failed step in run page top bar ([#11317](https://github.com/windmill-labs/windmill/issues/11317)) ([62d0088](https://github.com/windmill-labs/windmill/commit/62d0088de5463d44d00a8c9204e03d20649acd07))
+* start a deferred queued job now without changing its id ([#11347](https://github.com/windmill-labs/windmill/issues/11347)) ([4b09558](https://github.com/windmill-labs/windmill/commit/4b09558e13fadca6fe604c231a5cdfb896d4d0f8))
+* support aiagent steps in test_run_step ([#11321](https://github.com/windmill-labs/windmill/issues/11321)) ([1b74939](https://github.com/windmill-labs/windmill/commit/1b749399521cc5ebaa07fc5a55773113a6a3b903))
+
+
+### Bug Fixes
+
+* always save slack/teams/email handlers as scripts ([#11345](https://github.com/windmill-labs/windmill/issues/11345)) ([421fdab](https://github.com/windmill-labs/windmill/commit/421fdab3d89fb0241a4266393d318a5f9be6ead9))
+* bill from a single service account telemetry count ([#11301](https://github.com/windmill-labs/windmill/issues/11301)) ([ceb1722](https://github.com/windmill-labs/windmill/commit/ceb17223dcb5f66532017036916e91a5323279fb))
+* build ag grid link cells as DOM nodes with a vetted href ([#11316](https://github.com/windmill-labs/windmill/issues/11316)) ([5b03c17](https://github.com/windmill-labs/windmill/commit/5b03c1732e905673480d0960b95577d47ec919b3))
+* carry labels when deploying variables, resources and folders ([#11222](https://github.com/windmill-labs/windmill/issues/11222)) ([94e4fb1](https://github.com/windmill-labs/windmill/commit/94e4fb1c842057e92f3775eac0826ac7e396d725))
+* centre the toggle knob inside its track ([#11314](https://github.com/windmill-labs/windmill/issues/11314)) ([d02ff9a](https://github.com/windmill-labs/windmill/commit/d02ff9ac24d6d720c478e8424e73eebae9ec98fe))
+* **cli:** stub the API client over its real exports in tests ([#11363](https://github.com/windmill-labs/windmill/issues/11363)) ([30bb62c](https://github.com/windmill-labs/windmill/commit/30bb62cd2516466936fc791f2ff89dd61c743dba))
+* correct the tool controls and name field of a nested AI agent ([#11221](https://github.com/windmill-labs/windmill/issues/11221)) ([1d119e6](https://github.com/windmill-labs/windmill/commit/1d119e6b25fd5c72776dfb49917387de4be065c1))
+* **frontend:** keep the session when the persisted workspace is stale ([#11344](https://github.com/windmill-labs/windmill/issues/11344)) ([2e5f6d0](https://github.com/windmill-labs/windmill/commit/2e5f6d0e76301d19d2017239f5ad4fe25d67dc57))
+* hide answer actions until the chat turn ends ([#11323](https://github.com/windmill-labs/windmill/issues/11323)) ([98d3897](https://github.com/windmill-labs/windmill/commit/98d3897b8c926cc6d47a2417b6943a5d5612b3aa))
+* keep a watched job's code when its completion update drops it ([#11290](https://github.com/windmill-labs/windmill/issues/11290)) ([592ef73](https://github.com/windmill-labs/windmill/commit/592ef736104ff63dd36978a69353a2585636da87))
+* label chat job links with the end of the job id ([#11324](https://github.com/windmill-labs/windmill/issues/11324)) ([2f2882a](https://github.com/windmill-labs/windmill/commit/2f2882adff8371e33c0fb72cbf0c61298bbcb2ca))
+* limit compare page guidance to fork deploys ([#11300](https://github.com/windmill-labs/windmill/issues/11300)) ([c232ea4](https://github.com/windmill-labs/windmill/commit/c232ea43b4bfe1ed295dd2f71cc6cefa47772bed))
+* memoize the resolved authed so one request resolves identity once ([#11299](https://github.com/windmill-labs/windmill/issues/11299)) ([42655ff](https://github.com/windmill-labs/windmill/commit/42655ff5ef3e4eae8ff90c623683bb321ae7d06b))
+* **multiplayer:** a malformed frame from one client no longer exits the server ([#11353](https://github.com/windmill-labs/windmill/issues/11353)) ([8caf414](https://github.com/windmill-labs/windmill/commit/8caf41430123318a73b76e4c202da07aecaa3c1e))
+* **multiplayer:** don't drop client messages during cold-start token verification ([#11343](https://github.com/windmill-labs/windmill/issues/11343)) ([e8c3f95](https://github.com/windmill-labs/windmill/commit/e8c3f9514e5f712a3d84e301771de748f1cc1fda))
+* pass the schedule's custom tag when using run now ([#11322](https://github.com/windmill-labs/windmill/issues/11322)) ([8525206](https://github.com/windmill-labs/windmill/commit/852520636160408cd16119184ffe9c7be92682c8))
+* relock importers on their own tag, not the bare dependency tag ([#11359](https://github.com/windmill-labs/windmill/issues/11359)) ([a1abb36](https://github.com/windmill-labs/windmill/commit/a1abb36d9f5408b5fa3b659b769bc198604804ce))
+* space chat rows evenly after thinking and answers ([#11315](https://github.com/windmill-labs/windmill/issues/11315)) ([0e53d53](https://github.com/windmill-labs/windmill/commit/0e53d53b07ee1d4611c0f5fb605a290f596de980))
+* stop the chat repeating run results the card already shows ([#11293](https://github.com/windmill-labs/windmill/issues/11293)) ([ea0b947](https://github.com/windmill-labs/windmill/commit/ea0b947644d4094304ecea26406fa6aac15dcfac))
+* trust the system CA store for SMTP TLS ([#11328](https://github.com/windmill-labs/windmill/issues/11328)) ([9375c93](https://github.com/windmill-labs/windmill/commit/9375c93fd8d55bf22e3f290f44650affdf1868c2))
+* validate app and trigger paths, refuse traversal in workspace export ([#11311](https://github.com/windmill-labs/windmill/issues/11311)) ([7593597](https://github.com/windmill-labs/windmill/commit/7593597617110fbd276f3f32ca2be7064547a080))
+
+
+### Performance Improvements
+
+* skip job-start pings and checkpoint read for short non-WAC jobs ([#11356](https://github.com/windmill-labs/windmill/issues/11356)) ([53a5cfd](https://github.com/windmill-labs/windmill/commit/53a5cfd17ad4d6f378450f7e502539c5ad8c445d))
+* skip parent status write when a parallel loop iteration starts ([#11348](https://github.com/windmill-labs/windmill/issues/11348)) ([2456282](https://github.com/windmill-labs/windmill/commit/245628210f7d75aedb9e0a289d1dee63f1513b5b))
+* skip the flow_env ancestor walk for sub-flows with nothing to inherit ([#11349](https://github.com/windmill-labs/windmill/issues/11349)) ([bc4f872](https://github.com/windmill-labs/windmill/commit/bc4f872f1093a59b9b8fbc3e11a6f00b42474f0c))
+* stop polling http trigger routes on workers and every minute ([#11351](https://github.com/windmill-labs/windmill/issues/11351)) ([d931032](https://github.com/windmill-labs/windmill/commit/d9310327063b8b51ab3fe8d27cbdab99f5552967))
+
 ## [1.817.0](https://github.com/windmill-labs/windmill/compare/v1.816.0...v1.817.0) (2026-09-22)


@@ -1,31 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "INSERT INTO v2_job_completed AS cj\n                    ( workspace_id\n                    , id\n                    , started_at\n                    , duration_ms\n                    , result\n                    , result_columns\n                    , canceled_by\n                    , canceled_reason\n                    , flow_status\n                    , workflow_as_code_status\n                    , memory_peak\n                    , status\n                    , worker\n                    )\n                SELECT q.workspace_id, q.id, started_at, COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000), $3::text::jsonb, $10, $5, $6,\n                        flow_status, workflow_as_code_status,\n                        $8, CASE WHEN $4::BOOL THEN 'canceled'::job_status\n                        WHEN $7::BOOL THEN 'skipped'::job_status\n                        WHEN $2::BOOL THEN 'success'::job_status\n                        ELSE 'failure'::job_status END AS status,\n                        q.worker\n                FROM v2_job_queue q LEFT JOIN v2_job_status USING (id) WHERE q.id = $1\n            ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3::text::jsonb RETURNING duration_ms AS \"duration_ms!\"",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "duration_ms!",
-        "type_info": "Int8"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Uuid",
-        "Bool",
-        "Text",
-        "Bool",
-        "Varchar",
-        "Text",
-        "Bool",
-        "Int4",
-        "Int8",
-        "TextArray"
-      ]
-    },
-    "nullable": [
-      false
-    ]
-  },
-  "hash": "042ff3003bf82d11a78a1074a63379c19880e4d66dbf4dd5391f053b6dfa6b01"
-}
@@ -1,14 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE v2_job_status\n                 SET flow_status = flow_status - 'retry'\n                 WHERE id = $1",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Uuid"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "06db0e720dd59a7c52c0a98ea7b316237eb1a547678858c1a1e45985035b3468"
-}
@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE v2_job_queue SET running = false,\n                    started_at = CASE WHEN canceled_by IS NULL THEN NULL ELSE started_at END\n                WHERE id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "0ad65482d8282f778f4123b4133f9f230fffd4ea1f403dd37b28d4a105c7ef58"
+}
@@ -1,23 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "INSERT INTO v2_job_completed\n            (workspace_id, id, started_at, duration_ms, result, memory_peak, status, worker)\n        SELECT q.workspace_id, q.id, q.started_at,\n            COALESCE((EXTRACT('epoch' FROM now()) - EXTRACT('epoch' FROM COALESCE(q.started_at, now()))) * 1000, 0)::bigint,\n            $2::jsonb, r.memory_peak, 'failure'::job_status, q.worker\n        FROM v2_job_queue q\n        LEFT JOIN v2_job_runtime r ON r.id = q.id\n        WHERE q.id = $1\n        ON CONFLICT (id) DO UPDATE SET status = 'failure', result = $2::jsonb\n        RETURNING duration_ms AS \"duration_ms!\"",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "duration_ms!",
-        "type_info": "Int8"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Uuid",
-        "Jsonb"
-      ]
-    },
-    "nullable": [
-      false
-    ]
-  },
-  "hash": "2abc2a5830130b2b4b32983407abeea41923ba4122fa666c6e8d8b06dcb5a71f"
-}
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "UPDATE v2_job_status\n                     SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2)\n                     WHERE id = $3",
+  "query": "UPDATE v2_job_status\n                     SET flow_leaf_jobs = JSONB_SET(coalesce(flow_leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2)\n                     WHERE id = $3",
  "describe": {
    "columns": [],
    "parameters": {
@@ -12,5 +12,5 @@
    },
    "nullable": []
  },
-  "hash": "0aaec91ab06753e46c595d82469924a98f28b0dead245df7248a9ccb8a5f20c3"
+  "hash": "3c74bcf357bb87e2ee960473d40949b5bacac241549614bd39e4b055db7fc1ea"
 }
@@ -1,14 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE v2_job_queue SET running = false, started_at = null\n                WHERE id = $1 AND canceled_by IS NULL",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Uuid"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "3e55d027327bd3c76810fbe22d3ccb1bbbf83c8cff69d8f5907d1417a2522e69"
-}
@@ -1,27 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n        WITH step_index AS (\n            SELECT idx::text AS idx\n            FROM v2_job_status,\n                 jsonb_array_elements(flow_status->'modules') WITH ORDINALITY arr(elem, idx)\n            WHERE id = $1\n              AND elem->>'id' = $5\n            LIMIT 1\n        ), completed AS (\n            INSERT INTO v2_job_completed\n                (workspace_id, id, started_at, duration_ms, result,\n                 flow_status, workflow_as_code_status, status, worker)\n            SELECT\n                q.workspace_id, q.id, q.started_at,\n                (EXTRACT('epoch' FROM now()) - EXTRACT('epoch' FROM COALESCE(q.started_at, now()))) * 1000,\n                $3::text::jsonb,\n                CASE WHEN si.idx IS NOT NULL\n                    THEN jsonb_set(\n                        s.flow_status,\n                        ARRAY['modules', (si.idx::int - 1)::text],\n                        $6::jsonb\n                    )\n                    ELSE s.flow_status\n                END,\n                s.workflow_as_code_status,\n                'skipped'::job_status,\n                q.worker\n            FROM v2_job_queue q\n            LEFT JOIN v2_job_status s ON s.id = q.id\n            LEFT JOIN step_index si ON true\n            WHERE q.id = $1\n            ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = EXCLUDED.result\n            RETURNING 1 AS x\n        ), _deleted AS (\n            DELETE FROM v2_job_queue WHERE id = $1\n        ), _logged AS (\n            INSERT INTO job_logs (logs, job_id, workspace_id)\n            VALUES ($4, $1, $2)\n            ON CONFLICT (job_id) DO UPDATE SET logs = concat(job_logs.logs, EXCLUDED.logs)\n        )\n        SELECT x FROM completed\n        ",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "x",
-        "type_info": "Int4"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Uuid",
-        "Varchar",
-        "Text",
-        "Text",
-        "Text",
-        "Jsonb"
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "4461fe84370e7f07b4423ea5e71b913dd6ee9c655f9ec7087cc0fec9b9c3099a"
-}
@@ -1,15 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE v2_job_status\n                 SET flow_status = JSONB_SET(flow_status, ARRAY['step'], $1)\n                 WHERE id = $2",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Jsonb",
-        "Uuid"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "4622d28e2fa09bc60b9d0c79397efe0ca030638ded82c2ffd2155cacdf36ec11"
-}
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "\n        SELECT\n            j.id AS \"id!\", j.workspace_id AS \"workspace_id!\", j.parent_job, j.flow_step_id IS NOT NULL AS \"is_flow_step?\",\n            COALESCE(s.flow_status, s.workflow_as_code_status)::text AS \"flow_status: Box<str>\", r.ping AS last_ping, j.same_worker AS \"same_worker?\",\n            q.worker AS \"worker?\",\n            wp.ping_at AS \"worker_last_ping?\",\n            wp.memory_usage AS \"worker_memory_usage?\",\n            wp.wm_memory_usage AS \"worker_wm_memory_usage?\",\n            wp.memory AS \"worker_memory_total?\",\n            wp.worker_group AS \"worker_group?\",\n            wp.wm_version AS \"worker_version?\",\n            wp.current_job_id AS \"worker_current_job_id?\",\n            wp.worker_instance AS \"worker_instance?\"\n        FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id)\n            LEFT JOIN worker_ping wp ON wp.worker = q.worker\n        WHERE q.running = true AND q.suspend = 0 AND q.suspend_until IS null AND q.scheduled_for <= now()\n            AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode' OR j.kind = 'singlestepflow')\n            AND r.ping IS NOT NULL AND r.ping < NOW() - ($1 || ' seconds')::interval\n            AND q.canceled_by IS NULL\n\n        ",
+  "query": "\n        SELECT\n            j.id AS \"id!\", j.workspace_id AS \"workspace_id!\", j.parent_job, j.flow_step_id IS NOT NULL AS \"is_flow_step?\",\n            COALESCE(s.flow_status, s.workflow_as_code_status)::text AS \"flow_status: Box<str>\", r.ping AS last_ping, j.same_worker AS \"same_worker?\",\n            q.worker AS \"worker?\",\n            wp.ping_at AS \"worker_last_ping?\",\n            wp.memory_usage AS \"worker_memory_usage?\",\n            wp.wm_memory_usage AS \"worker_wm_memory_usage?\",\n            wp.memory AS \"worker_memory_total?\",\n            wp.worker_group AS \"worker_group?\",\n            wp.wm_version AS \"worker_version?\",\n            wp.current_job_id AS \"worker_current_job_id?\",\n            wp.worker_instance AS \"worker_instance?\",\n            q.canceled_by AS \"canceled_by?\", q.canceled_reason AS \"canceled_reason?\"\n        FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id)\n            LEFT JOIN worker_ping wp ON wp.worker = q.worker\n        WHERE q.running = true AND q.suspend = 0 AND q.suspend_until IS null AND q.scheduled_for <= now()\n            AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode' OR j.kind = 'singlestepflow')\n            AND r.ping IS NOT NULL AND r.ping < NOW() - ($1 || ' seconds')::interval\n            AND (q.canceled_by IS NULL OR r.ping < NOW() - ($1 || ' seconds')::interval * 10)\n\n        ",
  "describe": {
    "columns": [
      {
@@ -82,6 +82,16 @@
        "ordinal": 15,
        "name": "worker_instance?",
        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 16,
+        "name": "canceled_by?",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 17,
+        "name": "canceled_reason?",
+        "type_info": "Text"
      }
    ],
    "parameters": {
@@ -105,8 +115,10 @@
      false,
      false,
      true,
-      false
+      false,
+      true,
+      true
    ]
  },
-  "hash": "1bf87fe9667f860c6089bffb803291f4ff5c979be956369fa4406d789cca92fc"
+  "hash": "4f0b93160853a64c618872abbe934e4bcee02d8a9101cc7c1518ca73459c1fca"
 }
@@ -0,0 +1,41 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT c.id IS NOT NULL AS \"completed!\",\n                c.result AS \"result: sqlx::types::Json<Box<RawValue>>\",\n                c.status = 'success' AS \"success\",\n                EXISTS(SELECT 1 FROM v2_job_queue WHERE id = $1) AS \"queued!\"\n            FROM (SELECT 1) one\n            LEFT JOIN v2_job_completed c ON c.id = $1 AND c.workspace_id = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "completed!",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 1,
+        "name": "result: sqlx::types::Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 2,
+        "name": "success",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 3,
+        "name": "queued!",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null,
+      true,
+      null,
+      null
+    ]
+  },
+  "hash": "52f9582423689f098a337f88341239a136ee5132bc97f4b06cc1ee5beaa621e5"
+}
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT COALESCE(c.duration_ms, COALESCE((EXTRACT('epoch' FROM now()) - EXTRACT('epoch' FROM COALESCE(q.started_at, now()))) * 1000, 0)::bigint) AS \"duration_ms!\"\n        FROM v2_job_queue q LEFT JOIN v2_job_completed c ON c.id = q.id WHERE q.id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "duration_ms!",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "538f12535bd1b93a3a9baac91dc0d4ee0f3c949a00f62d38686e75d22e9d7884"
+}
@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE v2_job_status\n                     SET flow_status = flow_status - 'retry'\n                     WHERE id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "5ffba9baef525e1e46e9a258e39614fe8c00a264667ce48fb1f0b45662cef43d"
+}
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT COALESCE(c.duration_ms, COALESCE($2::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(q.started_at, now()))))*1000)::bigint) AS \"duration_ms!\"\n             FROM v2_job_queue q LEFT JOIN v2_job_completed c ON c.id = q.id WHERE q.id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "duration_ms!",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "67bde191639648d1f225a046eac57f4055900fdf0b1f65979deb1c1c64092a8e"
+}
@@ -1,15 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE v2_job_runtime r SET\n                        ping = now()\n                    FROM v2_job_queue q\n                    WHERE r.id = $1 AND q.id = r.id\n                        AND q.workspace_id = $2\n                        AND canceled_by IS NULL",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Uuid",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "7d07a717533bfcaf581f6655bc387095542490fbb4aae30ec7fa75c2dae98ec8"
-}
@@ -0,0 +1,31 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH deleted AS (\n                    DELETE FROM v2_job_queue WHERE id = $1\n                    RETURNING id, workspace_id, started_at, worker, canceled_by, canceled_reason\n                ), completed AS (\n                    INSERT INTO v2_job_completed AS cj\n                        ( workspace_id\n                        , id\n                        , started_at\n                        , duration_ms\n                        , result\n                        , result_columns\n                        , canceled_by\n                        , canceled_reason\n                        , flow_status\n                        , workflow_as_code_status\n                        , memory_peak\n                        , status\n                        , worker\n                        )\n                    SELECT d.workspace_id, d.id, d.started_at,\n                        COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(d.started_at, now()))))*1000),\n                        $3::text::jsonb, $10,\n                        CASE WHEN $4::BOOL THEN $5 ELSE d.canceled_by END,\n                        CASE WHEN $4::BOOL THEN $6 WHEN d.canceled_by IS NOT NULL THEN d.canceled_reason END,\n                        s.flow_status, s.workflow_as_code_status, $8,\n                        CASE WHEN $4::BOOL OR d.canceled_by IS NOT NULL THEN 'canceled'::job_status\n                            WHEN $7::BOOL THEN 'skipped'::job_status\n                            WHEN $2::BOOL THEN 'success'::job_status\n                            ELSE 'failure'::job_status END,\n                        d.worker\n                    FROM deleted d LEFT JOIN v2_job_status s ON s.id = d.id\n                    ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3::text::jsonb,\n                        canceled_by = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL\n                            THEN EXCLUDED.canceled_by ELSE cj.canceled_by END,\n                        canceled_reason = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL\n                            THEN EXCLUDED.canceled_reason ELSE cj.canceled_reason END\n                    RETURNING duration_ms\n                )\n                SELECT duration_ms AS \"duration_ms!\" FROM completed",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "duration_ms!",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Bool",
+        "Text",
+        "Bool",
+        "Varchar",
+        "Text",
+        "Bool",
+        "Int4",
+        "Int8",
+        "TextArray"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "9212902d4eb2834ab3b429968a7a85c53c49e4ef0f42c146324d5c71fbd1e21d"
+}
@@ -0,0 +1,33 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH deleted AS (\n            DELETE FROM v2_job_queue WHERE id = $1\n            RETURNING id, workspace_id, started_at, worker, canceled_by, canceled_reason\n        ), completed AS (\n            INSERT INTO v2_job_completed AS cj\n                ( workspace_id\n                , id\n                , started_at\n                , duration_ms\n                , result\n                , result_columns\n                , canceled_by\n                , canceled_reason\n                , flow_status\n                , workflow_as_code_status\n                , memory_peak\n                , status\n                , worker\n                )\n            SELECT d.workspace_id, d.id, d.started_at,\n                COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(d.started_at, now()))))*1000),\n                $3::text::jsonb, $10,\n                CASE WHEN $4::BOOL THEN $5 ELSE d.canceled_by END,\n                CASE WHEN $4::BOOL THEN $6 WHEN d.canceled_by IS NOT NULL THEN d.canceled_reason END,\n                s.flow_status, s.workflow_as_code_status, $8,\n                CASE WHEN $4::BOOL OR d.canceled_by IS NOT NULL THEN 'canceled'::job_status\n                    WHEN $7::BOOL THEN 'skipped'::job_status\n                    WHEN $2::BOOL THEN 'success'::job_status\n                    ELSE 'failure'::job_status END,\n                d.worker\n            FROM deleted d LEFT JOIN v2_job_status s ON s.id = d.id\n            ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3::text::jsonb,\n                canceled_by = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL\n                    THEN EXCLUDED.canceled_by ELSE cj.canceled_by END,\n                canceled_reason = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL\n                    THEN EXCLUDED.canceled_reason ELSE cj.canceled_reason END\n            RETURNING duration_ms\n        ), parent_ping AS (\n            UPDATE v2_job_runtime r SET ping = now()\n            FROM v2_job_queue q\n            WHERE r.id = $11 AND q.id = r.id AND q.workspace_id = $12\n                AND EXISTS (SELECT 1 FROM completed)\n        )\n        SELECT duration_ms AS \"duration_ms!\" FROM completed",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "duration_ms!",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Bool",
+        "Text",
+        "Bool",
+        "Varchar",
+        "Text",
+        "Bool",
+        "Int4",
+        "Int8",
+        "TextArray",
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "975158ae72ec12f3f933fba1d9623bf1c925feebe746299cd1f0eda3f8b949d5"
+}
@@ -1,16 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE v2_job_completed SET status = 'canceled'::job_status, canceled_by = $2, canceled_reason = $3 WHERE id = $1",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Uuid",
-        "Varchar",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "9a2345e35c9f44578aa7dde994fdde0be9e0d192e0aee224bff78d8bd2ccbd8b"
-}
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO zombie_job_counter (job_id, counter) VALUES ($1, 1)\n                    ON CONFLICT (job_id) DO UPDATE SET counter = zombie_job_counter.counter + 1\n                    RETURNING counter",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "counter",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "9d8e3b978d746432d336900282602f0ce092d50783d8b3ea17b7e6121f207b92"
+}
@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE v2_job_status\n                     SET flow_status = JSONB_SET(flow_status, ARRAY['step'], $1)\n                     WHERE id = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Jsonb",
+        "Uuid"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "9fb1a1e4e39e0e8a0934fcda7b3e6e6180a48ab792d305d04146e81b06b0d69b"
+}
@@ -0,0 +1,325 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE v2_job_status SET\n            flow_status = JSONB_SET(\n                JSONB_SET(v2_job_status.flow_status, $2::TEXT[], $3),\n                $4::TEXT[], $5\n            ) - $6::TEXT[],\n            flow_leaf_jobs = CASE\n                WHEN $7::TEXT IS NULL\n                    OR COALESCE(v2_job.flow_innermost_root_job, v2_job_status.id) <> v2_job_status.id\n                THEN v2_job_status.flow_leaf_jobs\n                ELSE JSONB_SET(COALESCE(v2_job_status.flow_leaf_jobs, '{}'::JSONB), ARRAY[$7::TEXT], $8) END\n        FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id\n        WHERE v2_job_status.id = $1 AND v2_job_queue.id = $1\n        RETURNING\n            v2_job_queue.workspace_id,\n            v2_job_queue.id,\n            v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",\n            v2_job.parent_job,\n            v2_job.created_by,\n            v2_job_queue.started_at,\n            v2_job_queue.runnable_settings_handle,\n            v2_job_queue.scheduled_for,\n            v2_job.runnable_path,\n            v2_job.kind as \"kind: JobKind\",\n            v2_job.runnable_id as \"runnable_id: ScriptHash\",\n            v2_job_queue.canceled_reason,\n            v2_job_queue.canceled_by,\n            v2_job.permissioned_as,\n            v2_job.permissioned_as_email,\n            v2_job_status.flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",\n            v2_job.tag,\n            v2_job.script_lang as \"script_lang: ScriptLang\",\n            v2_job.same_worker,\n            v2_job.pre_run_error,\n            v2_job.concurrent_limit,\n            v2_job.concurrency_time_window_s,\n            v2_job.flow_innermost_root_job,\n            v2_job.root_job,\n            v2_job.timeout,\n            v2_job.flow_step_id,\n            v2_job.cache_ttl,\n            v2_job_queue.cache_ignore_s3_path,\n            v2_job_queue.priority,\n            v2_job.preprocessed,\n            v2_job.script_entrypoint_override,\n            v2_job.trigger,\n            v2_job.trigger_kind as \"trigger_kind: TriggerKindLabel\",\n            v2_job.visible_to_owner,\n            NULL as permissioned_as_end_user_email",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "id",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 2,
+        "name": "args: sqlx::types::Json<HashMap<String, Box<RawValue>>>",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 3,
+        "name": "parent_job",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 4,
+        "name": "created_by",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 5,
+        "name": "started_at",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 6,
+        "name": "runnable_settings_handle",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 7,
+        "name": "scheduled_for",
+        "type_info": "Timestamptz"
+      },
+      {
+        "ordinal": 8,
+        "name": "runnable_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 9,
+        "name": "kind: JobKind",
+        "type_info": {
+          "Custom": {
+            "name": "job_kind",
+            "kind": {
+              "Enum": [
+                "script",
+                "preview",
+                "flow",
+                "dependencies",
+                "flowpreview",
+                "script_hub",
+                "identity",
+                "flowdependencies",
+                "http",
+                "graphql",
+                "postgresql",
+                "noop",
+                "appdependencies",
+                "deploymentcallback",
+                "singlestepflow",
+                "flowscript",
+                "flownode",
+                "appscript",
+                "aiagent",
+                "unassigned_script",
+                "unassigned_flow",
+                "unassigned_singlestepflow"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 10,
+        "name": "runnable_id: ScriptHash",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 11,
+        "name": "canceled_reason",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 12,
+        "name": "canceled_by",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 13,
+        "name": "permissioned_as",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 14,
+        "name": "permissioned_as_email",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 15,
+        "name": "flow_status: sqlx::types::Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      },
+      {
+        "ordinal": 16,
+        "name": "tag",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 17,
+        "name": "script_lang: ScriptLang",
+        "type_info": {
+          "Custom": {
+            "name": "script_lang",
+            "kind": {
+              "Enum": [
+                "python3",
+                "deno",
+                "go",
+                "bash",
+                "postgresql",
+                "nativets",
+                "bun",
+                "mysql",
+                "bigquery",
+                "snowflake",
+                "graphql",
+                "powershell",
+                "mssql",
+                "php",
+                "bunnative",
+                "rust",
+                "ansible",
+                "csharp",
+                "oracledb",
+                "nu",
+                "java",
+                "duckdb",
+                "ruby",
+                "rlang",
+                "dbt"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 18,
+        "name": "same_worker",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 19,
+        "name": "pre_run_error",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 20,
+        "name": "concurrent_limit",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 21,
+        "name": "concurrency_time_window_s",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 22,
+        "name": "flow_innermost_root_job",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 23,
+        "name": "root_job",
+        "type_info": "Uuid"
+      },
+      {
+        "ordinal": 24,
+        "name": "timeout",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 25,
+        "name": "flow_step_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 26,
+        "name": "cache_ttl",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 27,
+        "name": "cache_ignore_s3_path",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 28,
+        "name": "priority",
+        "type_info": "Int2"
+      },
+      {
+        "ordinal": 29,
+        "name": "preprocessed",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 30,
+        "name": "script_entrypoint_override",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 31,
+        "name": "trigger",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 32,
+        "name": "trigger_kind: TriggerKindLabel",
+        "type_info": {
+          "Custom": {
+            "name": "job_trigger_kind",
+            "kind": {
+              "Enum": [
+                "webhook",
+                "http",
+                "websocket",
+                "kafka",
+                "email",
+                "nats",
+                "schedule",
+                "app",
+                "ui",
+                "postgres",
+                "sqs",
+                "gcp",
+                "mqtt",
+                "nextcloud",
+                "google",
+                "ci_test",
+                "github",
+                "azure",
+                "asset",
+                "freshness",
+                "amqp"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 33,
+        "name": "visible_to_owner",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 34,
+        "name": "permissioned_as_end_user_email",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "TextArray",
+        "Jsonb",
+        "TextArray",
+        "Jsonb",
+        "TextArray",
+        "Text",
+        "Jsonb"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      true,
+      true,
+      false,
+      true,
+      true,
+      false,
+      true,
+      false,
+      true,
+      true,
+      true,
+      false,
+      false,
+      true,
+      false,
+      true,
+      false,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      true,
+      false,
+      null
+    ]
+  },
+  "hash": "ab05c1359039ae99a2aa6156eb2a208814ce37fcde72e8ef61904a4f41fb4579"
+}
@@ -1,16 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "UPDATE v2_job_status\n                         SET flow_leaf_jobs = JSONB_SET(coalesce(flow_leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2)\n                         WHERE COALESCE((SELECT flow_innermost_root_job FROM v2_job WHERE id = $3), $3) = id",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Jsonb",
-        "Uuid"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "aed8bd751c3e988f422216e74acfb77dc03469355d2a0da0b2d6b4aeeea37d3e"
-}
@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT q.id FROM v2_job_queue q JOIN v2_job_runtime r ON r.id = q.id\n                WHERE q.id = $1 AND q.running AND r.ping = $2\n                FOR UPDATE OF q SKIP LOCKED",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Timestamptz"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "bf2a8883c3a9d57ac56b347dccd26eaa4d92e36965e7f069a0107647a2bfff8b"
+}
@@ -0,0 +1,27 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        WITH step_index AS (\n            SELECT idx::text AS idx\n            FROM v2_job_status,\n                 jsonb_array_elements(flow_status->'modules') WITH ORDINALITY arr(elem, idx)\n            WHERE id = $1\n              AND elem->>'id' = $5\n            LIMIT 1\n        ), target AS (\n            -- The queue row is locked before the completed row's key, as every completion does.\n            SELECT id FROM v2_job_queue WHERE id = $1 FOR UPDATE\n        ), completed AS (\n            INSERT INTO v2_job_completed\n                (workspace_id, id, started_at, duration_ms, result,\n                 flow_status, workflow_as_code_status, status, worker)\n            SELECT\n                q.workspace_id, q.id, q.started_at,\n                (EXTRACT('epoch' FROM now()) - EXTRACT('epoch' FROM COALESCE(q.started_at, now()))) * 1000,\n                $3::text::jsonb,\n                CASE WHEN si.idx IS NOT NULL\n                    THEN jsonb_set(\n                        s.flow_status,\n                        ARRAY['modules', (si.idx::int - 1)::text],\n                        $6::jsonb\n                    )\n                    ELSE s.flow_status\n                END,\n                s.workflow_as_code_status,\n                'skipped'::job_status,\n                q.worker\n            FROM v2_job_queue q\n            LEFT JOIN v2_job_status s ON s.id = q.id\n            LEFT JOIN step_index si ON true\n            WHERE q.id IN (SELECT id FROM target)\n            ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = EXCLUDED.result\n            RETURNING 1 AS x\n        ), _deleted AS (\n            DELETE FROM v2_job_queue WHERE id IN (SELECT id FROM target)\n        ), _logged AS (\n            INSERT INTO job_logs (logs, job_id, workspace_id)\n            VALUES ($4, $1, $2)\n            ON CONFLICT (job_id) DO UPDATE SET logs = concat(job_logs.logs, EXCLUDED.logs)\n        )\n        SELECT x FROM completed\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "x",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Varchar",
+        "Text",
+        "Text",
+        "Text",
+        "Jsonb"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "c4254f50fc6f74aae9e23b461a6c69c181eba0210cc068b64e67b0a518fe1cba"
+}
@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH deleted AS (\n            DELETE FROM v2_job_queue WHERE id = $1 RETURNING id, workspace_id, started_at, worker\n        )\n        INSERT INTO v2_job_completed\n            (workspace_id, id, started_at, duration_ms, result, memory_peak, status, worker)\n        SELECT d.workspace_id, d.id, d.started_at, $3, $2::jsonb, r.memory_peak,\n            'failure'::job_status, d.worker\n        FROM deleted d\n        LEFT JOIN v2_job_runtime r ON r.id = d.id\n        ON CONFLICT (id) DO UPDATE SET status = 'failure', result = $2::jsonb\n        RETURNING id",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Jsonb",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "d9b7e45213aebced9cebe8ce1278d38136bc73c3517d0e0fbad369666504930c"
+}
@@ -1,28 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM v2_job_queue WHERE id = $1 RETURNING canceled_by, canceled_reason",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "canceled_by",
-        "type_info": "Varchar"
-      },
-      {
-        "ordinal": 1,
-        "name": "canceled_reason",
-        "type_info": "Text"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Uuid"
-      ]
-    },
-    "nullable": [
-      true,
-      true
-    ]
-  },
-  "hash": "ed2eb32d445164e7ab5fb735e01f86505bc499320e08d1aad15a9a82c3823c0b"
-}
@@ -1,6 +1,6 @@
 [package]
 name = "windmill"
-version = "1.817.0"
+version = "1.818.0"
 authors.workspace = true
 edition.workspace = true

@@ -88,7 +88,7 @@ members = [
 exclude = ["./windmill-duckdb-ffi-internal", "./parsers/windmill-parser-wasm"]

 [workspace.package]
-version = "1.817.0"
+version = "1.818.0"
 authors = ["Ruben Fiszel <ruben@windmill.dev>"]
 edition = "2021"

@@ -1 +1 @@
-de73db2bacfdc3eaa2e63b1827178bc198d54e5c
+7a256cf353db7cf64a60a09fa0de7f3a8b27f626
@@ -6216,7 +6216,7 @@ checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f"

 [[package]]
 name = "windmill-common"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "aho-corasick",
 "anyhow",
@@ -6302,7 +6302,7 @@ dependencies = [

 [[package]]
 name = "windmill-macros"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -6314,7 +6314,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "convert_case",
 "serde",
@@ -6323,7 +6323,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-bash"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "lazy_static",
@@ -6335,7 +6335,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-csharp"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "serde_json",
@@ -6347,7 +6347,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-go"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "gosyn",
@@ -6359,7 +6359,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-graphql"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "lazy_static",
@@ -6371,7 +6371,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-java"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "serde_json",
@@ -6383,7 +6383,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-nu"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "nu-parser",
@@ -6394,7 +6394,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-php"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "itertools 0.14.0",
@@ -6405,7 +6405,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-py"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "itertools 0.14.0",
@@ -6417,7 +6417,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-py-asset"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "rustpython-ast",
@@ -6428,7 +6428,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-py-imports"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "async-recursion",
@@ -6450,7 +6450,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-r"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "serde_json",
@@ -6462,7 +6462,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-ruby"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "lazy_static",
@@ -6476,7 +6476,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-rust"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "convert_case",
@@ -6493,7 +6493,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-sql"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "lazy_static",
@@ -6506,7 +6506,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-sql-asset"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "serde",
@@ -6518,7 +6518,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-ts"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "lazy_static",
@@ -6536,7 +6536,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-ts-asset"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "serde-wasm-bindgen",
@@ -6552,7 +6552,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-wac"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "rustpython-ast",
@@ -6568,7 +6568,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-wasm"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "getrandom 0.2.17",
@@ -6600,7 +6600,7 @@ dependencies = [

 [[package]]
 name = "windmill-parser-yaml"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "lazy_static",
@@ -6614,7 +6614,7 @@ dependencies = [

 [[package]]
 name = "windmill-types"
-version = "1.817.0"
+version = "1.818.0"
 dependencies = [
 "anyhow",
 "bitflags",
@@ -12,7 +12,7 @@ resolver = "2"
 members = ["."]

 [workspace.package]
-version = "1.817.0"
+version = "1.818.0"
 edition = "2021"
 authors = ["Ruben Fiszel <ruben@windmill.dev>"]

@@ -122,9 +122,9 @@ use windmill_common::{
 #[cfg(feature = "parquet")]
 use windmill_object_store::reload_object_store_setting;
 use windmill_queue::{
-    cancel_job, get_queued_job_v2,
+    add_completed_job_error, cancel_job, canceled_result, get_queued_job_v2,
    schedule::{find_unarmed_schedules, rearm_schedule, RearmOutcome},
-    SameWorkerPayload,
+    CanceledBy, MiniCompletedJob, SameWorkerPayload,
 };
 use windmill_store::resources::MAX_RESOURCE_VERSIONS;
 use windmill_worker::{
@@ -6210,8 +6210,8 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, node_n
 }

 /// Force-complete a zombie job that handle_job_error failed to complete.
-/// This is a minimal fallback: it inserts a failed completed job and deletes
-/// from the queue in a single transaction, without schedule pushing or
+/// This is a minimal fallback: it moves the job from the queue to a failed
+/// completed job in a single transaction, without schedule pushing or
 /// error handler logic. The one thing it keeps is the WAC parent notification,
 /// deliberately inside the transaction: if that fails, the whole completion
 /// rolls back and the job waits for the next sweep, which is cheaper than a
@@ -6248,52 +6248,67 @@ async fn force_complete_zombie_job(

    let mut tx = db.begin().await?;

+    // Locks in the order of every other completion: a WAC parent's rows, then this job's queue
+    // row, then its completed row (see `record_child_completion`). A worker completing the same
+    // job concurrently would otherwise deadlock against this.
    let duration_ms = sqlx::query_scalar!(
-        "INSERT INTO v2_job_completed
-            (workspace_id, id, started_at, duration_ms, result, memory_peak, status, worker)
-        SELECT q.workspace_id, q.id, q.started_at,
-            COALESCE((EXTRACT('epoch' FROM now()) - EXTRACT('epoch' FROM COALESCE(q.started_at, now()))) * 1000, 0)::bigint,
-            $2::jsonb, r.memory_peak, 'failure'::job_status, q.worker
-        FROM v2_job_queue q
-        LEFT JOIN v2_job_runtime r ON r.id = q.id
-        WHERE q.id = $1
-        ON CONFLICT (id) DO UPDATE SET status = 'failure', result = $2::jsonb
-        RETURNING duration_ms AS \"duration_ms!\"",
+        "SELECT COALESCE(c.duration_ms, COALESCE((EXTRACT('epoch' FROM now()) - EXTRACT('epoch' FROM COALESCE(q.started_at, now()))) * 1000, 0)::bigint) AS \"duration_ms!\"
+        FROM v2_job_queue q LEFT JOIN v2_job_completed c ON c.id = q.id WHERE q.id = $1",
        job_id,
-        error_value,
    )
    .fetch_optional(&mut *tx)
    .await?;
+    let Some(duration_ms) = duration_ms else {
+        return Ok(());
+    };

    // A WAC parent parked on this job must learn of the failure here too, or it
    // waits out its whole suspend window and runs the task again.
    let mut wac_parent_ready = false;
-    if let Some(duration_ms) = duration_ms {
-        let parent = sqlx::query!(
-            "SELECT parent_job, flow_step_id FROM v2_job WHERE id = $1",
-            job_id
+    let parent = sqlx::query!(
+        "SELECT parent_job, flow_step_id FROM v2_job WHERE id = $1",
+        job_id
+    )
+    .fetch_optional(&mut *tx)
+    .await?;
+    if let Some(parent_job) = parent
+        .filter(|j| j.flow_step_id.is_none())
+        .and_then(|j| j.parent_job)
+    {
+        wac_parent_ready = windmill_common::wac::record_child_completion(
+            &mut tx,
+            &parent_job,
+            job_id,
+            false,
+            duration_ms,
+            &error_value.to_string(),
        )
-        .fetch_optional(&mut *tx)
        .await?;
-        if let Some(parent_job) = parent
-            .filter(|j| j.flow_step_id.is_none())
-            .and_then(|j| j.parent_job)
-        {
-            wac_parent_ready = windmill_common::wac::record_child_completion(
-                &mut tx,
-                &parent_job,
-                job_id,
-                false,
-                duration_ms,
-                &error_value.to_string(),
-            )
-            .await?;
-        }
    }

-    sqlx::query!("DELETE FROM v2_job_queue WHERE id = $1", job_id)
-        .execute(&mut *tx)
-        .await?;
+    let completed = sqlx::query_scalar!(
+        "WITH deleted AS (
+            DELETE FROM v2_job_queue WHERE id = $1 RETURNING id, workspace_id, started_at, worker
+        )
+        INSERT INTO v2_job_completed
+            (workspace_id, id, started_at, duration_ms, result, memory_peak, status, worker)
+        SELECT d.workspace_id, d.id, d.started_at, $3, $2::jsonb, r.memory_peak,
+            'failure'::job_status, d.worker
+        FROM deleted d
+        LEFT JOIN v2_job_runtime r ON r.id = d.id
+        ON CONFLICT (id) DO UPDATE SET status = 'failure', result = $2::jsonb
+        RETURNING id",
+        job_id,
+        error_value,
+        duration_ms,
+    )
+    .fetch_optional(&mut *tx)
+    .await?;
+    // Completed by someone else while this waited on the WAC parent: roll back, so the parent
+    // keeps what the winning completion recorded.
+    if completed.is_none() {
+        return Ok(());
+    }

    tx.commit().await?;

@@ -6486,6 +6501,9 @@ async fn find_zombie_flow_culprit_worker(
 }

 async fn handle_zombie_flows(db: &DB) -> error::Result<()> {
+    // A canceled flow gets ten times the transition timeout: requeuing it while its worker is
+    // still in a slow transition lets a second worker complete it under that transition, and
+    // which worker runs a transition is not recorded anywhere to tell a slow one from a dead one.
    // flow_status is cast ::text on purpose: decoding the jsonb column directly as Box<str>
    // yields its binary form (leading version byte) and fails serde_json parsing at column 1.
    let flows = sqlx::query!(
@@ -6501,13 +6519,14 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> {
            wp.worker_group AS "worker_group?",
            wp.wm_version AS "worker_version?",
            wp.current_job_id AS "worker_current_job_id?",
-            wp.worker_instance AS "worker_instance?"
+            wp.worker_instance AS "worker_instance?",
+            q.canceled_by AS "canceled_by?", q.canceled_reason AS "canceled_reason?"
        FROM v2_job_queue q JOIN v2_job j USING (id) LEFT JOIN v2_job_runtime r USING (id) LEFT JOIN v2_job_status s USING (id)
            LEFT JOIN worker_ping wp ON wp.worker = q.worker
        WHERE q.running = true AND q.suspend = 0 AND q.suspend_until IS null AND q.scheduled_for <= now()
            AND (j.kind = 'flow' OR j.kind = 'flowpreview' OR j.kind = 'flownode' OR j.kind = 'singlestepflow')
            AND r.ping IS NOT NULL AND r.ping < NOW() - ($1 || ' seconds')::interval
-            AND q.canceled_by IS NULL
+            AND (q.canceled_by IS NULL OR r.ping < NOW() - ($1 || ' seconds')::interval * 10)

        "#,
        FLOW_ZOMBIE_TRANSITION_TIMEOUT.as_str()
@@ -6520,21 +6539,99 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> {
            .flow_status
            .as_deref()
            .and_then(|x| serde_json::from_str::<FlowStatus>(x).ok());
-        if !flow.same_worker.unwrap_or(false)
-            && status.as_ref().is_some_and(|s| s.is_not_yet_started())
+        // A worker that pulls a canceled flow completes it as canceled, and hands that to its
+        // parent like any step's cancel, so a canceled flow whose transition was lost goes back
+        // to the queue exactly like a flow that never started.
+        if flow.canceled_by.is_some()
+            || (!flow.same_worker.unwrap_or(false)
+                && status.as_ref().is_some_and(|s| s.is_not_yet_started()))
        {
-            let error_message = format!(
-                "Zombie flow detected: {} in workspace {}. It hasn't started yet, restarting it.",
-                flow.id, flow.workspace_id
-            );
-            tracing::error!(error_message);
-            if !CRITICAL_ALERT_MUTE_ZOMBIE_JOB_RESTART.load(Ordering::Relaxed) {
-                report_critical_error(error_message, db.clone(), Some(&flow.workspace_id), None)
-                    .await;
-            }
-            // if the flow hasn't started and is a zombie, we can simply restart it
            let mut tx = db.begin().await?;

+            // Claims the flow unless another sweep holds it or it visibly moved on since it was
+            // read. This does not tell a slow transition from a lost one: the ping is written when
+            // the step completes, at the start of the transition, and nothing is locked until its
+            // end, so the grace in the query above is what keeps a live transition safe.
+            let claimed = sqlx::query_scalar!(
+                "SELECT q.id FROM v2_job_queue q JOIN v2_job_runtime r ON r.id = q.id
+                WHERE q.id = $1 AND q.running AND r.ping = $2
+                FOR UPDATE OF q SKIP LOCKED",
+                flow.id,
+                flow.last_ping,
+            )
+            .fetch_optional(&mut *tx)
+            .await?
+            .is_some();
+            if !claimed {
+                continue;
+            }
+
+            let error_message = if let Some(canceler) = flow.canceled_by.as_deref() {
+                // Bounds the requeues of a canceled flow whose completion keeps failing: past the
+                // limit it is completed on its own instead, on every sweep until that succeeds.
+                // The count is taken under the row claim, so the alert goes out once.
+                let attempt = sqlx::query_scalar!(
+                    "INSERT INTO zombie_job_counter (job_id, counter) VALUES ($1, 1)
+                    ON CONFLICT (job_id) DO UPDATE SET counter = zombie_job_counter.counter + 1
+                    RETURNING counter",
+                    flow.id
+                )
+                .fetch_one(&mut *tx)
+                .await?;
+                if attempt > RESTART_LIMIT {
+                    tx.commit().await?;
+                    if attempt == RESTART_LIMIT + 1 {
+                        let error_message = format!(
+                            "Zombie flow detected: {} in workspace {}. It was canceled by {canceler} but could not be completed after {RESTART_LIMIT} attempts, completing it as canceled without handing it to its parent.",
+                            flow.id, flow.workspace_id
+                        );
+                        tracing::error!(error_message);
+                        report_critical_error(
+                            error_message,
+                            db.clone(),
+                            Some(&flow.workspace_id),
+                            None,
+                        )
+                        .await;
+                    }
+                    let completed = match get_queued_job_v2(db, &flow.id).await {
+                        Ok(Some(job)) => add_completed_job_error(
+                            db,
+                            &MiniCompletedJob::from(job),
+                            0,
+                            Some(CanceledBy {
+                                username: Some(canceler.to_string()),
+                                reason: flow.canceled_reason.clone(),
+                            }),
+                            canceled_result(flow.canceled_reason.as_deref(), Some(canceler)),
+                            "monitor",
+                            false,
+                            None,
+                        )
+                        .await
+                        .map(|_| ()),
+                        Ok(None) => Ok(()),
+                        Err(e) => Err(e),
+                    };
+                    if let Err(e) = completed {
+                        tracing::error!(
+                            "could not complete canceled zombie flow {}, retrying on the next sweep: {e:#}",
+                            flow.id
+                        );
+                    }
+                    continue;
+                }
+                format!(
+                    "Zombie flow detected: {} in workspace {}. It was canceled by {canceler} but its worker stopped between two steps, queuing it again to complete the cancel ({attempt}/{RESTART_LIMIT}).",
+                    flow.id, flow.workspace_id
+                )
+            } else {
+                format!(
+                    "Zombie flow detected: {} in workspace {}. It hasn't started yet, restarting it.",
+                    flow.id, flow.workspace_id
+                )
+            };
+
            let concurrency_key =
                sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", flow.id)
                    .fetch_optional(&mut *tx)
@@ -6554,15 +6651,26 @@ async fn handle_zombie_flows(db: &DB) -> error::Result<()> {
                }
            }

+            // A canceled flow keeps its start: the pull only sets a missing one, and the canceled
+            // run's duration is measured from it.
            sqlx::query!(
-                "UPDATE v2_job_queue SET running = false, started_at = null
-                WHERE id = $1 AND canceled_by IS NULL",
+                "UPDATE v2_job_queue SET running = false,
+                    started_at = CASE WHEN canceled_by IS NULL THEN NULL ELSE started_at END
+                WHERE id = $1",
                flow.id
            )
            .execute(&mut *tx)
            .await?;

            tx.commit().await?;
+
+            tracing::error!(error_message);
+            if flow.canceled_by.is_some()
+                || !CRITICAL_ALERT_MUTE_ZOMBIE_JOB_RESTART.load(Ordering::Relaxed)
+            {
+                report_critical_error(error_message, db.clone(), Some(&flow.workspace_id), None)
+                    .await;
+            }
        } else {
            let id = flow.id.clone();
            let last_ping = flow.last_ping.clone();
@@ -7782,3 +7890,154 @@ mod log_file_listing_tests {
        assert_eq!(files[0].0.to_string(), "2026-08-29 06:46:00");
    }
 }
+
+#[cfg(test)]
+mod canceled_zombie_flow_tests {
+    use super::{handle_zombie_flows, DB, RESTART_LIMIT};
+    use serde_json::json;
+    use uuid::Uuid;
+
+    /// A running flow at `step` (0: not started yet), last pinged `ping_age` ago (`None`: mid-step, no ping).
+    async fn insert_flow(
+        db: &DB,
+        parent: Option<Uuid>,
+        step: i32,
+        ping_age: Option<&str>,
+        canceled_by: Option<&str>,
+    ) -> anyhow::Result<Uuid> {
+        let id = Uuid::new_v4();
+        sqlx::query(
+            "INSERT INTO v2_job (id, workspace_id, created_by, permissioned_as, permissioned_as_email,
+                kind, tag, parent_job, flow_step_id)
+             VALUES ($1, 'admins', 'admin', 'u/admin', 'admin@windmill.dev', 'flowpreview', 'flow',
+                $2, CASE WHEN $2 IS NOT NULL THEN 'sf' END)",
+        )
+        .bind(id)
+        .bind(parent)
+        .execute(db)
+        .await?;
+        sqlx::query(
+            "INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, running, started_at, tag,
+                canceled_by, canceled_reason)
+             VALUES ($1, 'admins', now(), true, now() - interval '1 hour', 'flow', $2, $2)",
+        )
+        .bind(id)
+        .bind(canceled_by)
+        .execute(db)
+        .await?;
+        sqlx::query("INSERT INTO v2_job_runtime (id, ping) VALUES ($1, now() - $2::interval)")
+            .bind(id)
+            .bind(ping_age)
+            .execute(db)
+            .await?;
+        sqlx::query("INSERT INTO v2_job_status (id, flow_status) VALUES ($1, $2)")
+            .bind(id)
+            .bind(json!({"step": step, "modules": [if step == 0 {
+                json!({"type": "WaitingForPriorSteps", "id": "a"})
+            } else {
+                json!({"type": "InProgress", "id": "a", "job": Uuid::nil()})
+            }],
+                "failure_module": {"type": "WaitingForPriorSteps", "id": "failure"}}))
+            .execute(db)
+            .await?;
+        Ok(id)
+    }
+
+    /// `(running, canceled_by, still started an hour ago)`
+    async fn queue_row(db: &DB, id: Uuid) -> anyhow::Result<(bool, Option<String>, Option<bool>)> {
+        Ok(sqlx::query_as(
+            "SELECT running, canceled_by, started_at < now() - interval '30 minutes'
+             FROM v2_job_queue WHERE id = $1",
+        )
+        .bind(id)
+        .fetch_one(db)
+        .await?)
+    }
+
+    /// A subflow canceled on its own whose worker died between two steps goes back to the queue
+    /// with its cancel and its start, for a worker to complete it and hand it to its parent. The
+    /// parent is left alone: the cancel must not reach flows the user never canceled.
+    #[sqlx::test(migrations = "./migrations")]
+    async fn requeues_a_stranded_canceled_flow_and_nothing_else(db: DB) -> anyhow::Result<()> {
+        let root = insert_flow(&db, None, 0, None, None).await?;
+        let child = insert_flow(&db, Some(root), 1, Some("1 hour"), Some("admin")).await?;
+
+        handle_zombie_flows(&db).await?;
+
+        assert_eq!(
+            queue_row(&db, child).await?,
+            (false, Some("admin".to_string()), Some(true))
+        );
+        assert_eq!(queue_row(&db, root).await?, (true, None, Some(true)));
+        Ok(())
+    }
+
+    /// A canceled flow is left to a transition that may just be slow for ten times the timeout
+    /// a flow that never started gets.
+    #[sqlx::test(migrations = "./migrations")]
+    async fn gives_a_canceled_flow_a_longer_grace(db: DB) -> anyhow::Result<()> {
+        let slow = insert_flow(&db, None, 1, Some("2 minutes"), Some("admin")).await?;
+        let stuck = insert_flow(&db, None, 1, Some("1 hour"), Some("admin")).await?;
+        let not_started = insert_flow(&db, None, 0, Some("2 minutes"), None).await?;
+
+        handle_zombie_flows(&db).await?;
+
+        assert!(
+            queue_row(&db, slow).await?.0,
+            "a recent canceled flow is left running"
+        );
+        assert!(
+            !queue_row(&db, stuck).await?.0,
+            "past the grace, it is requeued"
+        );
+        assert!(
+            !queue_row(&db, not_started).await?.0,
+            "a flow that never started is not held back"
+        );
+        Ok(())
+    }
+
+    /// A canceled flow that keeps coming back after its requeues is completed as canceled on its
+    /// own instead of being requeued (and alerted on) forever, and that completion is retried by
+    /// the next sweep if a previous one did not land. Its parent is not advanced, since the
+    /// transition that would do it is what kept failing.
+    #[sqlx::test(migrations = "./migrations")]
+    async fn completes_a_canceled_flow_requeued_too_often(db: DB) -> anyhow::Result<()> {
+        let root = insert_flow(&db, None, 0, None, None).await?;
+        let child = insert_flow(&db, Some(root), 1, Some("1 hour"), Some("admin")).await?;
+        let retried = insert_flow(&db, None, 1, Some("1 hour"), Some("admin")).await?;
+        for (id, counter) in [(child, RESTART_LIMIT), (retried, RESTART_LIMIT + 1)] {
+            sqlx::query("INSERT INTO zombie_job_counter (job_id, counter) VALUES ($1, $2)")
+                .bind(id)
+                .bind(counter)
+                .execute(&db)
+                .await?;
+        }
+
+        handle_zombie_flows(&db).await?;
+
+        for id in [child, retried] {
+            let completed: Option<(String, Option<String>)> = sqlx::query_as(
+                "SELECT status::text, canceled_by FROM v2_job_completed WHERE id = $1",
+            )
+            .bind(id)
+            .fetch_optional(&db)
+            .await?;
+            assert_eq!(
+                completed,
+                Some(("canceled".to_string(), Some("admin".to_string())))
+            );
+        }
+        // the parent is not advanced, but the completion pings it: the sweep then reaps it as a
+        // flow hanging between two steps instead of leaving it waiting
+        assert_eq!(queue_row(&db, root).await?, (true, None, Some(true)));
+        let parent_pinged: bool = sqlx::query_scalar(
+            "SELECT ping > now() - interval '1 minute' FROM v2_job_runtime WHERE id = $1",
+        )
+        .bind(root)
+        .fetch_one(&db)
+        .await?;
+        assert!(parent_pinged);
+        Ok(())
+    }
+}
@@ -0,0 +1,108 @@
+//! A flow step's completion is the progress that keeps its flow from being reaped as a zombie: it
+//! must refresh the parent's ping, canceled or not, since a canceled flow still needs its next
+//! transition to complete.
+
+use serde_json::json;
+use sqlx::{types::Json, Pool, Postgres};
+use uuid::Uuid;
+use windmill_queue::{add_completed_job, get_mini_completed_job};
+
+const W_ID: &str = "test-workspace";
+
+async fn insert_running(
+    db: &Pool<Postgres>,
+    kind: &str,
+    parent: Option<Uuid>,
+    step_id: Option<&str>,
+) -> anyhow::Result<Uuid> {
+    let id = Uuid::new_v4();
+    sqlx::query(
+        "INSERT INTO v2_job (id, workspace_id, created_by, created_at, permissioned_as, \
+            permissioned_as_email, kind, runnable_path, script_lang, tag, visible_to_owner, \
+            parent_job, flow_step_id) \
+         VALUES ($1, $2, 'test-user', now(), 'u/test-user', 'test@windmill.dev', \
+            $3::job_kind, 'u/test-user/f', 'bash', 'bash', true, $4, $5)",
+    )
+    .bind(id)
+    .bind(W_ID)
+    .bind(kind)
+    .bind(parent)
+    .bind(step_id)
+    .execute(db)
+    .await?;
+    sqlx::query(
+        "INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, running, started_at, tag) \
+         VALUES ($1, $2, now(), true, now(), 'bash')",
+    )
+    .bind(id)
+    .bind(W_ID)
+    .execute(db)
+    .await?;
+    sqlx::query("INSERT INTO v2_job_runtime (id, ping) VALUES ($1, now() - interval '1 hour')")
+        .bind(id)
+        .execute(db)
+        .await?;
+    Ok(id)
+}
+
+async fn complete_step_and_read_parent_ping_age(
+    db: &Pool<Postgres>,
+    canceled: bool,
+) -> anyhow::Result<f64> {
+    let parent = insert_running(db, "flow", None, None).await?;
+    let step = insert_running(db, "script", Some(parent), Some("a")).await?;
+    if canceled {
+        // a cancel of the flow marks the flow and each of its steps
+        sqlx::query("UPDATE v2_job_queue SET canceled_by = 'test-user' WHERE id = ANY($1)")
+            .bind(vec![parent, step])
+            .execute(db)
+            .await?;
+    }
+    let job = get_mini_completed_job(&step, W_ID, db).await?.unwrap();
+    add_completed_job(
+        db,
+        &job,
+        true,
+        false,
+        Json(&json!("done")),
+        None,
+        0,
+        None,
+        false,
+        None,
+        false,
+    )
+    .await?;
+
+    let (status, queued): (String, bool) = sqlx::query_as(
+        "SELECT status::text, EXISTS (SELECT 1 FROM v2_job_queue WHERE id = $1) \
+         FROM v2_job_completed WHERE id = $1",
+    )
+    .bind(step)
+    .fetch_one(db)
+    .await?;
+    assert_eq!(status, if canceled { "canceled" } else { "success" });
+    assert!(!queued, "the step left the queue");
+
+    Ok(sqlx::query_scalar(
+        "SELECT EXTRACT(EPOCH FROM now() - ping)::float8 FROM v2_job_runtime WHERE id = $1",
+    )
+    .bind(parent)
+    .fetch_one(db)
+    .await?)
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn a_step_completion_refreshes_its_running_flow_ping(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    let age = complete_step_and_read_parent_ping_age(&db, false).await?;
+    assert!(age < 60.0, "the parent's ping was refreshed: {age}s old");
+
+    let age = complete_step_and_read_parent_ping_age(&db, true).await?;
+    assert!(
+        age < 60.0,
+        "a canceled parent's ping was refreshed: {age}s old"
+    );
+    Ok(())
+}
@@ -1,7 +1,7 @@
 openapi: "3.0.3"

 info:
-  version: 1.817.0
+  version: 1.818.0
  title: Windmill API

  contact:
@@ -117,7 +117,8 @@ use windmill_common::{
    scripts::{ScriptHash, ScriptLang},
    users::username_to_permissioned_as,
    utils::{
-        not_found_if_none, now_from_db, paginate, require_admin, Pagination, ScheduleType, StripPath,
+        not_found_if_none, now_from_db, paginate, require_admin, Pagination, ScheduleType,
+        StripPath,
    },
 };

@@ -2762,7 +2763,7 @@ async fn send_email_with_instance_smtp(
    Json(send_email): Json<SendEmail>,
 ) -> error::Result<Json<String>> {
    use windmill_common::jobs::EMAIL_ERROR_HANDLER_USER_EMAIL;
-    use windmill_queue::SCHEDULE_ERROR_HANDLER_USER_EMAIL;
+    use windmill_queue::{ERROR_HANDLER_USER_EMAIL, SCHEDULE_ERROR_HANDLER_USER_EMAIL};

    if *CLOUD_HOSTED {
        tracing::warn!(
@@ -2771,12 +2772,15 @@ async fn send_email_with_instance_smtp(
        return Err(anyhow::anyhow!("Feature not supported in cloud hosted windmill").into());
    }

+    // Any code pushed as a workspace or schedule error handler, custom ones included, runs as
+    // one of these identities: this keeps out ad-hoc job tokens, not who authors handler code.
    let is_handler_job = authed.email == EMAIL_ERROR_HANDLER_USER_EMAIL
+        || authed.email == ERROR_HANDLER_USER_EMAIL
        || authed.email == SCHEDULE_ERROR_HANDLER_USER_EMAIL;

    if !is_handler_job && !windmill_api_auth::is_super_admin_authed(&db, &authed).await? {
        return Err(Error::NotAuthorized(
-            "Only super admin or whitelisted token can access email workspace error handler feature"
+            "Only super admin or a workspace/schedule error handler job can send emails with the instance SMTP"
                .to_string(),
        ));
    }
@@ -1177,74 +1177,76 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
    let serialized_result = result.serialized_json();
    let sanitized_result = strip_json_nul(serialized_result.as_ref());

+    let labels = result.wm_labels();
+    let is_scheduled =
+        completed_job.schedule_path().is_some() && completed_job.runnable_path.is_some();
+    let wac_parent = (!completed_job.is_flow_step())
+        .then_some(completed_job.parent_job)
+        .flatten();
+    let monitor_parent = (completed_job.is_flow_step() && flow_is_done)
+        .then_some(completed_job.parent_job)
+        .flatten();
+    let completion = Completion {
+        completed_job,
+        success,
+        skipped,
+        result: sanitized_result.as_ref(),
+        result_columns,
+        mem_peak,
+        canceled_by,
+        duration,
+    };
+
+    if labels.is_none()
+        && !has_concurrent_limit
+        && wac_parent.is_none()
+        && monitor_parent.is_none()
+        && !is_scheduled
+    {
+        let Some(duration) = completion.execute(&mut *db.acquire().await?).await? else {
+            return Err(not_in_queue_error(db, job_id).await);
+        };
+        log_completed_job(completed_job, duration, success);
+        return Ok((None, duration, false, false));
+    }
+
    let mut tx = db.begin().warn_after_seconds(10).await?;

-    let duration =  sqlx::query_scalar!(
-            "INSERT INTO v2_job_completed AS cj
-                    ( workspace_id
-                    , id
-                    , started_at
-                    , duration_ms
-                    , result
-                    , result_columns
-                    , canceled_by
-                    , canceled_reason
-                    , flow_status
-                    , workflow_as_code_status
-                    , memory_peak
-                    , status
-                    , worker
-                    )
-                SELECT q.workspace_id, q.id, started_at, COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(started_at, now()))))*1000), $3::text::jsonb, $10, $5, $6,
-                        flow_status, workflow_as_code_status,
-                        $8, CASE WHEN $4::BOOL THEN 'canceled'::job_status
-                        WHEN $7::BOOL THEN 'skipped'::job_status
-                        WHEN $2::BOOL THEN 'success'::job_status
-                        ELSE 'failure'::job_status END AS status,
-                        q.worker
-                FROM v2_job_queue q LEFT JOIN v2_job_status USING (id) WHERE q.id = $1
-            ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3::text::jsonb RETURNING duration_ms AS \"duration_ms!\"",
-            /* $1 */ completed_job.id,
-            /* $2 */ success,
-            /* $3 */ sanitized_result.as_ref(),
-            /* $4 */ canceled_by.is_some(),
-            /* $5 */ canceled_by.clone().map(|cb| cb.username).flatten(),
-            /* $6 */ canceled_by.clone().map(|cb| cb.reason).flatten(),
-            /* $7 */ skipped,
-            /* $8 */ if mem_peak > 0 { Some(mem_peak) } else { None },
-            /* $9 */ duration,
-            /* $10 */ result_columns as Option<&Vec<String>>,
+    // The parent's rows are locked ahead of the child's own queue row (see
+    // `record_child_completion` for the order this must keep), so the duration it stamps is read
+    // before the completion: the one the completed row will hold. `now()` is fixed for the
+    // transaction, and a completed row already there keeps its own duration.
+    let mut wac_parent_ready = false;
+    if let Some(parent_job) = wac_parent {
+        let Some(duration) = sqlx::query_scalar!(
+            "SELECT COALESCE(c.duration_ms, COALESCE($2::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(q.started_at, now()))))*1000)::bigint) AS \"duration_ms!\"
+             FROM v2_job_queue q LEFT JOIN v2_job_completed c ON c.id = q.id WHERE q.id = $1",
+            job_id,
+            duration,
        )
        .fetch_optional(&mut *tx)
        .warn_after_seconds(10)
-        .await
-        .map_err(|e| Error::internal_err(format!("Could not add completed job {job_id}: {e:#}")))?;
-
-    let duration = if let Some(duration) = duration {
-        duration
-    } else {
-        let already_inserted = sqlx::query_scalar!(
-            "SELECT EXISTS(SELECT 1 FROM v2_job_completed WHERE id = $1)",
-            job_id
+        .await?
+        else {
+            return Err(not_in_queue_error(&mut *tx, job_id).await);
+        };
+        wac_parent_ready = windmill_common::wac::record_child_completion(
+            &mut tx,
+            &parent_job,
+            &completed_job.id,
+            success,
+            duration,
+            sanitized_result.as_ref(),
        )
-        .fetch_one(&mut *tx)
        .warn_after_seconds(10)
-        .await
-        .map_err(|e| Error::internal_err(format!("Could not add completed job {job_id}: {e:#}")))?
-        .unwrap_or(false);
+        .await?;
+    }

-        if already_inserted {
-            return Err(Error::AlreadyCompleted(format!(
-                "The queued job {job_id} is already completed."
-            )));
-        } else {
-            return Err(Error::AlreadyCompleted(format!(
-                "There is no queued job anymore for {job_id} but there is no completed job either."
-            )));
-        }
+    let Some(duration) = completion.execute(&mut *tx).await? else {
+        return Err(not_in_queue_error(&mut *tx, job_id).await);
    };

-    if let Some(mut labels) = result.wm_labels() {
+    if let Some(mut labels) = labels {
        // A `\u0000` inside a wm_labels entry decodes to a real NUL that the
        // `text[]` column rejects, which would abort this same transaction (and
        // roll back the sanitized result insert) exactly like an unsanitized
@@ -1266,78 +1268,19 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
        .map_err(|e| Error::InternalErr(format!("Could not update job labels: {e:#}")))?;
    }

-    // Before `delete_job`: the parent's rows are locked ahead of the child's own
-    // queue row (see `record_child_completion` for the order this must keep).
-    let mut wac_parent_ready = false;
-    if !completed_job.is_flow_step() {
-        if let Some(parent_job) = completed_job.parent_job {
-            wac_parent_ready = windmill_common::wac::record_child_completion(
-                &mut tx,
-                &parent_job,
-                &completed_job.id,
-                success,
-                duration,
-                sanitized_result.as_ref(),
-            )
-            .warn_after_seconds(10)
-            .await?;
-        }
-    }
-
    let mut _skip_downstream_error_handlers = false;
-    let (ntx, canceled_at_delete) = delete_job(tx, &job_id).warn_after_seconds(10).await?;
-    tx = ntx;
-    // `canceled_by` is only what the worker last read from the queue row. Deleting the row waits
-    // for a cancel still being written, so the row it removed is the final word on whether this
-    // job was canceled.
-    if canceled_by.is_none() {
-        if let Some(canceled) = canceled_at_delete {
-            sqlx::query!(
-                "UPDATE v2_job_completed SET status = 'canceled'::job_status, canceled_by = $2, \
-                 canceled_reason = $3 WHERE id = $1",
-                job_id,
-                canceled.username,
-                canceled.reason,
-            )
-            .execute(&mut *tx)
-            .warn_after_seconds(10)
-            .await?;
-        }
-    }
-    // tracing::error!("3 {:?}", start.elapsed());
-
    if completed_job.is_flow_step() {
-        if let Some(parent_job) = completed_job.parent_job {
-            // persist the flow last progress timestamp to avoid zombie flow jobs
-            tracing::debug!(
-                "Persisting flow last progress timestamp to flow job: {:?}",
-                parent_job
-            );
-            sqlx::query!(
-                "UPDATE v2_job_runtime r SET
-                        ping = now()
-                    FROM v2_job_queue q
-                    WHERE r.id = $1 AND q.id = r.id
-                        AND q.workspace_id = $2
-                        AND canceled_by IS NULL",
+        if let Some(parent_job) = monitor_parent {
+            let r = sqlx::query_scalar!(
+                "UPDATE parallel_monitor_lock SET last_ping = now() WHERE parent_flow_id = $1 and job_id = $2 RETURNING 1",
                parent_job,
-                &completed_job.workspace_id
-            )
-            .execute(&mut *tx)
-            .warn_after_seconds(10)
-            .await?;
-            if flow_is_done {
-                let r = sqlx::query_scalar!(
-                    "UPDATE parallel_monitor_lock SET last_ping = now() WHERE parent_flow_id = $1 and job_id = $2 RETURNING 1",
-                    parent_job,
-                    &completed_job.id
-                ).fetch_optional(&mut *tx).warn_after_seconds(10).await?;
-                if r.is_some() {
-                    tracing::info!(
-                            "parallel flow iteration is done, setting parallel monitor last ping lock for job {}",
-                            &completed_job.id
-                        );
-                }
+                &completed_job.id
+            ).fetch_optional(&mut *tx).warn_after_seconds(10).await?;
+            if r.is_some() {
+                tracing::info!(
+                        "parallel flow iteration is done, setting parallel monitor last ping lock for job {}",
+                        &completed_job.id
+                    );
            }
        }
    } else {
@@ -1509,6 +1452,216 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(

    tx.commit().warn_after_seconds(10).await?;

+    log_completed_job(completed_job, duration, success);
+    // tracing::info!("completed job: {:?}", start.elapsed().as_micros());
+    Ok((
+        None,
+        duration,
+        _skip_downstream_error_handlers,
+        wac_parent_ready,
+    ))
+}
+
+/// What a job's completion writes, for `Completion::execute`.
+struct Completion<'a> {
+    completed_job: &'a MiniCompletedJob,
+    success: bool,
+    skipped: bool,
+    result: &'a str,
+    result_columns: Option<&'a Vec<String>>,
+    mem_peak: i32,
+    canceled_by: &'a Option<CanceledBy>,
+    duration: Option<i64>,
+}
+
+impl Completion<'_> {
+    /// Moves the job from the queue to the completed jobs and refreshes a flow step's parent
+    /// ping, as one statement. Returns `None` when the job was no longer in the queue.
+    ///
+    /// The completion takes its cancellation from the queue row it deletes, not only from
+    /// `canceled_by`: that is what the worker last read, and the delete waits for a cancel still
+    /// being written, so the deleted row is the final word on whether the job was canceled.
+    ///
+    /// It locks the queue row before the completed row's key. Any other writer completing a job
+    /// (the monitor's zombie fallback, debounce) must take them in the same order, or the two
+    /// deadlock.
+    async fn execute(&self, conn: &mut sqlx::PgConnection) -> error::Result<Option<i64>> {
+        let Completion {
+            completed_job,
+            success,
+            skipped,
+            result,
+            result_columns,
+            mem_peak,
+            canceled_by,
+            duration,
+        } = *self;
+        #[cfg(feature = "prometheus")]
+        if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
+            QUEUE_DELETE_COUNT.inc();
+        }
+        otel_incr_queue_delete_count();
+        let err = |e: sqlx::Error| {
+            Error::internal_err(format!(
+                "Could not add completed job {}: {e:#}",
+                completed_job.id
+            ))
+        };
+        // A step's completion is progress of its flow, and keeps the flow from being reaped as a
+        // zombie. Any other completion runs the statement without the ping: Postgres sets up every
+        // write of a plan, so an unused ping would cost about a tenth of the completion. The two
+        // statements differ only by the ping; a change to the delete or the insert goes in both.
+        let Some(parent_to_ping) = completed_job
+            .is_flow_step()
+            .then_some(completed_job.parent_job)
+            .flatten()
+        else {
+            return sqlx::query_scalar!(
+                "WITH deleted AS (
+                    DELETE FROM v2_job_queue WHERE id = $1
+                    RETURNING id, workspace_id, started_at, worker, canceled_by, canceled_reason
+                ), completed AS (
+                    INSERT INTO v2_job_completed AS cj
+                        ( workspace_id
+                        , id
+                        , started_at
+                        , duration_ms
+                        , result
+                        , result_columns
+                        , canceled_by
+                        , canceled_reason
+                        , flow_status
+                        , workflow_as_code_status
+                        , memory_peak
+                        , status
+                        , worker
+                        )
+                    SELECT d.workspace_id, d.id, d.started_at,
+                        COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(d.started_at, now()))))*1000),
+                        $3::text::jsonb, $10,
+                        CASE WHEN $4::BOOL THEN $5 ELSE d.canceled_by END,
+                        CASE WHEN $4::BOOL THEN $6 WHEN d.canceled_by IS NOT NULL THEN d.canceled_reason END,
+                        s.flow_status, s.workflow_as_code_status, $8,
+                        CASE WHEN $4::BOOL OR d.canceled_by IS NOT NULL THEN 'canceled'::job_status
+                            WHEN $7::BOOL THEN 'skipped'::job_status
+                            WHEN $2::BOOL THEN 'success'::job_status
+                            ELSE 'failure'::job_status END,
+                        d.worker
+                    FROM deleted d LEFT JOIN v2_job_status s ON s.id = d.id
+                    ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3::text::jsonb,
+                        canceled_by = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL
+                            THEN EXCLUDED.canceled_by ELSE cj.canceled_by END,
+                        canceled_reason = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL
+                            THEN EXCLUDED.canceled_reason ELSE cj.canceled_reason END
+                    RETURNING duration_ms
+                )
+                SELECT duration_ms AS \"duration_ms!\" FROM completed",
+                /* $1 */ completed_job.id,
+                /* $2 */ success,
+                /* $3 */ result,
+                /* $4 */ canceled_by.is_some(),
+                /* $5 */ canceled_by.as_ref().and_then(|cb| cb.username.as_deref()),
+                /* $6 */ canceled_by.as_ref().and_then(|cb| cb.reason.as_deref()),
+                /* $7 */ skipped,
+                /* $8 */ if mem_peak > 0 { Some(mem_peak) } else { None },
+                /* $9 */ duration,
+                /* $10 */ result_columns as Option<&Vec<String>>,
+            )
+            .fetch_optional(&mut *conn)
+            .warn_after_seconds(10)
+            .await
+            .map_err(err);
+        };
+        // A canceled flow is pinged too: it is completed by its next transition like any other
+        // flow, and the zombie flow monitor needs the ping to finish the cancel if that
+        // transition is lost.
+        sqlx::query_scalar!(
+        "WITH deleted AS (
+            DELETE FROM v2_job_queue WHERE id = $1
+            RETURNING id, workspace_id, started_at, worker, canceled_by, canceled_reason
+        ), completed AS (
+            INSERT INTO v2_job_completed AS cj
+                ( workspace_id
+                , id
+                , started_at
+                , duration_ms
+                , result
+                , result_columns
+                , canceled_by
+                , canceled_reason
+                , flow_status
+                , workflow_as_code_status
+                , memory_peak
+                , status
+                , worker
+                )
+            SELECT d.workspace_id, d.id, d.started_at,
+                COALESCE($9::bigint, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE(d.started_at, now()))))*1000),
+                $3::text::jsonb, $10,
+                CASE WHEN $4::BOOL THEN $5 ELSE d.canceled_by END,
+                CASE WHEN $4::BOOL THEN $6 WHEN d.canceled_by IS NOT NULL THEN d.canceled_reason END,
+                s.flow_status, s.workflow_as_code_status, $8,
+                CASE WHEN $4::BOOL OR d.canceled_by IS NOT NULL THEN 'canceled'::job_status
+                    WHEN $7::BOOL THEN 'skipped'::job_status
+                    WHEN $2::BOOL THEN 'success'::job_status
+                    ELSE 'failure'::job_status END,
+                d.worker
+            FROM deleted d LEFT JOIN v2_job_status s ON s.id = d.id
+            ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, result = $3::text::jsonb,
+                canceled_by = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL
+                    THEN EXCLUDED.canceled_by ELSE cj.canceled_by END,
+                canceled_reason = CASE WHEN NOT $4::BOOL AND EXCLUDED.canceled_by IS NOT NULL
+                    THEN EXCLUDED.canceled_reason ELSE cj.canceled_reason END
+            RETURNING duration_ms
+        ), parent_ping AS (
+            UPDATE v2_job_runtime r SET ping = now()
+            FROM v2_job_queue q
+            WHERE r.id = $11 AND q.id = r.id AND q.workspace_id = $12
+                AND EXISTS (SELECT 1 FROM completed)
+        )
+        SELECT duration_ms AS \"duration_ms!\" FROM completed",
+        /* $1 */ completed_job.id,
+        /* $2 */ success,
+        /* $3 */ result,
+        /* $4 */ canceled_by.is_some(),
+        /* $5 */ canceled_by.as_ref().and_then(|cb| cb.username.as_deref()),
+        /* $6 */ canceled_by.as_ref().and_then(|cb| cb.reason.as_deref()),
+        /* $7 */ skipped,
+        /* $8 */ if mem_peak > 0 { Some(mem_peak) } else { None },
+        /* $9 */ duration,
+        /* $10 */ result_columns as Option<&Vec<String>>,
+        /* $11 */ parent_to_ping,
+        /* $12 */ &completed_job.workspace_id,
+    )
+    .fetch_optional(&mut *conn)
+    .warn_after_seconds(10)
+    .await
+    .map_err(err)
+    }
+}
+
+/// The error for a completion that found no queue row to complete.
+async fn not_in_queue_error<'e>(conn: impl PgExecutor<'e>, job_id: Uuid) -> Error {
+    let already_inserted = sqlx::query_scalar!(
+        "SELECT EXISTS(SELECT 1 FROM v2_job_completed WHERE id = $1)",
+        job_id
+    )
+    .fetch_one(conn)
+    .warn_after_seconds(10)
+    .await;
+    match already_inserted {
+        Err(e) => Error::internal_err(format!("Could not add completed job {job_id}: {e:#}")),
+        Ok(Some(true)) => {
+            Error::AlreadyCompleted(format!("The queued job {job_id} is already completed."))
+        }
+        Ok(_) => Error::AlreadyCompleted(format!(
+            "There is no queued job anymore for {job_id} but there is no completed job either."
+        )),
+    }
+}
+
+fn log_completed_job(completed_job: &MiniCompletedJob, duration: i64, success: bool) {
+    let job_id = completed_job.id;
    tracing::info!(
        %job_id,
        root_job = ?completed_job.flow_innermost_root_job.map(|x| x.to_string()).unwrap_or_else(|| String::new()),
@@ -1527,13 +1680,6 @@ async fn commit_completed_job<T: Serialize + Send + Sync + ValidableJson>(
        "inserted completed job: {} (success: {success})",
        completed_job.id
    );
-    // tracing::info!("completed job: {:?}", start.elapsed().as_micros());
-    Ok((
-        None,
-        duration,
-        _skip_downstream_error_handlers,
-        wac_parent_ready,
-    ))
 }

 async fn check_result_size<T: ValidableJson>(
@@ -5190,46 +5336,6 @@ async fn extract_result_from_job_result(
    }
 }

-/// Also reports the cancellation the deleted row carried, if any. Unlike a plain read of the queue
-/// row, this waits for a cancel that is still being written, so it is the last word on one.
-pub async fn delete_job<'c>(
-    mut tx: Transaction<'c, Postgres>,
-    job_id: &Uuid,
-) -> windmill_common::error::Result<(Transaction<'c, Postgres>, Option<CanceledBy>)> {
-    #[cfg(feature = "prometheus")]
-    if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
-        QUEUE_DELETE_COUNT.inc();
-    }
-    otel_incr_queue_delete_count();
-
-    let job_removed = sqlx::query!(
-        "DELETE FROM v2_job_queue WHERE id = $1 RETURNING canceled_by, canceled_reason",
-        job_id,
-    )
-    .fetch_optional(&mut *tx)
-    .await;
-
-    let canceled = match &job_removed {
-        Err(job_removed) => {
-            tracing::error!(
-                "Job {job_id} could not be deleted: {job_removed}. This is not necessarily an error, as the job might have been deleted by another process such as in the case of cancelling"
-            );
-            None
-        }
-        Ok(None) => {
-            tracing::error!("Job {job_id} could not be deleted, no row was removed. This is not necessarily an error, as the job might have been deleted by another process such as in the case of cancelling");
-            None
-        }
-        Ok(Some(row)) => row.canceled_by.as_ref().map(|username| CanceledBy {
-            username: Some(username.clone()),
-            reason: row.canceled_reason.clone(),
-        }),
-    };
-
-    tracing::debug!("Job {job_id} deleted");
-    Ok((tx, canceled))
-}
-
 pub async fn job_is_complete(db: &DB, id: Uuid, w_id: &str) -> error::Result<bool> {
    Ok(sqlx::query_scalar!(
        "SELECT EXISTS(SELECT 1 FROM v2_job_completed WHERE id = $1 AND workspace_id = $2)",
@@ -39,8 +39,8 @@ use windmill_common::{
    worker::{to_raw_value, Connection},
 };
 use windmill_queue::{
-    add_completed_job, add_completed_job_error, get_mini_pulled_job, push, MiniCompletedJob,
-    MiniPulledJob, PushArgs, PushIsolationLevel,
+    add_completed_job, add_completed_job_error, check_tag_available_for_push, get_mini_pulled_job,
+    push, resolve_push_tag, MiniCompletedJob, MiniPulledJob, PushArgs, PushIsolationLevel,
 };

 /// Shared collection of abort handles for spawned tool tasks.
@@ -473,8 +473,7 @@ async fn execute_windmill_tool(
                ));
            }
            let path = format!("{}/tools/{}", ctx.job.runnable_path(), tool_module.id);
-            // tool jobs are pushed with the parent agent job's tag and executed inline on the
-            // same worker, so a tag override on a nested agent tool does not apply here
+            // a nested agent always runs inline on this worker, under this agent's tag
            JobPayloadWithTag {
                payload: JobPayload::AIAgent { path },
                tag: None,
@@ -492,13 +491,6 @@ async fn execute_windmill_tool(
        }
    };

-    let mut tx = ctx.db.begin().await?;
-
-    let job_perms =
-        windmill_common::auth::get_job_perms(&mut *tx, &ctx.job.id, &ctx.job.workspace_id)
-            .await?
-            .map(|x| x.into());
-
    let (email, permissioned_as) = if let Some(on_behalf_of) = job_payload.on_behalf_of.as_ref() {
        (&on_behalf_of.email, on_behalf_of.permissioned_as.clone())
    } else {
@@ -508,6 +500,55 @@ async fn execute_windmill_tool(
        )
    };

+    let push_args = PushArgs { args: &tool_call_args, extra: None };
+
+    // A tool with a tag of its own naming another queue than this agent's must run on a worker
+    // that tag selects, and like any tag the caller chose it must pass the workspace's custom tag
+    // restrictions. It runs inline when this worker serves that queue: the agent holds this
+    // worker while it waits, so dispatching to a queue it serves could wait on itself.
+    let mut resolved_tool_tag = None;
+    if let Some(tag) = job_payload.tag.as_deref() {
+        resolved_tool_tag = resolve_push_tag(tag, &push_args, &ctx.job.workspace_id, ctx.db)
+            .await
+            .filter(|resolved| *resolved != ctx.job.tag);
+    }
+    let tool_tag = match (job_payload.tag.as_deref(), resolved_tool_tag.as_ref()) {
+        (Some(tag), Some(_)) => {
+            let is_super_admin = windmill_common::auth::is_super_admin_email(ctx.db, email).await?;
+            check_tag_available_for_push(
+                ctx.db,
+                &ctx.job.workspace_id,
+                tag,
+                &push_args,
+                is_super_admin,
+                None,
+            )
+            .await
+            .map_err(|e| match e {
+                Error::BadRequest(msg) => Error::BadRequest(format!(
+                    "tool '{}' cannot run on tag '{tag}': {msg}",
+                    tool_call.function.name
+                )),
+                e => e,
+            })?;
+            Some(tag.to_string())
+        }
+        _ => None,
+    };
+    let run_inline = resolved_tool_tag.is_none_or(|resolved| {
+        windmill_common::worker::WORKER_CONFIG
+            .load()
+            .worker_tags
+            .contains(&resolved)
+    });
+
+    let mut tx = ctx.db.begin().await?;
+
+    let job_perms =
+        windmill_common::auth::get_job_perms(&mut *tx, &ctx.job.id, &ctx.job.workspace_id)
+            .await?
+            .map(|x| x.into());
+
    let job_priority = tool_module.priority.or(ctx.job.priority);

    let tx = PushIsolationLevel::Transaction(tx);
@@ -516,7 +557,7 @@ async fn execute_windmill_tool(
        tx,
        &ctx.job.workspace_id,
        job_payload.payload,
-        PushArgs { args: &tool_call_args, extra: None },
+        push_args,
        &ctx.job.created_by,
        email,
        permissioned_as,
@@ -532,12 +573,12 @@ async fn execute_windmill_tool(
        false,
        None,
        ctx.job.visible_to_owner,
-        Some(ctx.job.tag.clone()),
+        Some(tool_tag.unwrap_or_else(|| ctx.job.tag.clone())),
        job_payload.timeout,
        None,
        job_priority,
        job_perms.as_ref(),
-        true,
+        run_inline,
        None,
        None,
        None,
@@ -546,12 +587,45 @@ async fn execute_windmill_tool(

    tx.commit().await?;

+    if !run_inline {
+        // Like an inline tool's failure, a lost wait reaches the model rather than failing the
+        // agent.
+        let (result, success) =
+            match wait_for_dispatched_tool_job(ctx.db, &uuid, &ctx.job.workspace_id).await {
+                Ok(outcome) => outcome,
+                Err(e) => (
+                    to_raw_value(&serde_json::json!({ "error": { "message": e.to_string() } })),
+                    false,
+                ),
+            };
+        return report_tool_result(
+            ctx,
+            tool_call,
+            tool_module,
+            job_id,
+            &result,
+            success,
+            is_ai_agent_tool && success,
+            messages,
+            final_events_str,
+        )
+        .await;
+    }
+
    let tool_job = get_mini_pulled_job(ctx.db, &uuid).await?;

    let Some(tool_job) = tool_job else {
        return Err(Error::internal_err("Tool job not found".to_string()));
    };

+    // The tool gets a token of its own, as it would on any worker: the agent's token would
+    // identify the tool as the agent job (its OIDC path and job id).
+    let tool_client = AuthedClient::new(
+        ctx.base_internal_url.to_string(),
+        tool_job.workspace_id.clone(),
+        windmill_queue::create_token(ctx.db, &tool_job, None).await,
+        None,
+    );
    let tool_job = Arc::new(tool_job);

    let (inner_job_completed_tx, inner_job_completed_rx) = JobCompletedSender::new(ctx.conn, 1);
@@ -564,7 +638,7 @@ async fn execute_windmill_tool(
    // Clone everything needed for the spawned task
    let tool_job_spawn = tool_job.clone();
    let conn_spawn = ctx.conn.clone();
-    let client_spawn = ctx.client.clone();
+    let client_spawn = tool_client;
    let hostname_spawn = ctx.hostname.to_string();
    let worker_name_spawn = ctx.worker_name.to_string();
    let worker_dir_spawn = ctx.worker_dir.to_string();
@@ -793,12 +867,93 @@ async fn handle_tool_execution_success(
        ));
    };

+    report_tool_result(
+        ctx,
+        tool_call,
+        tool_module,
+        job_id,
+        &result,
+        success,
+        is_ai_agent_tool && job_success,
+        messages,
+        final_events_str,
+    )
+    .await
+}
+
+/// Wait for a tool job another worker runs; returns its result and whether it succeeded.
+/// Canceling the agent cancels this job along with it, as a child of the agent's job.
+async fn wait_for_dispatched_tool_job(
+    db: &DB,
+    id: &Uuid,
+    w_id: &str,
+) -> Result<(Box<RawValue>, bool), Error> {
+    const MAX_CONSECUTIVE_POLL_ERRORS: u32 = 10;
+    let mut interval = std::time::Duration::from_millis(50);
+    let mut poll_errors = 0;
+    loop {
+        // One statement reads both tables from one snapshot, so a job completing between two
+        // reads cannot look like one that vanished.
+        let state = match sqlx::query!(
+            "SELECT c.id IS NOT NULL AS \"completed!\",
+                c.result AS \"result: sqlx::types::Json<Box<RawValue>>\",
+                c.status = 'success' AS \"success\",
+                EXISTS(SELECT 1 FROM v2_job_queue WHERE id = $1) AS \"queued!\"
+            FROM (SELECT 1) one
+            LEFT JOIN v2_job_completed c ON c.id = $1 AND c.workspace_id = $2",
+            id,
+            w_id
+        )
+        .fetch_one(db)
+        .await
+        {
+            Ok(state) => {
+                poll_errors = 0;
+                state
+            }
+            Err(e) if poll_errors < MAX_CONSECUTIVE_POLL_ERRORS => {
+                poll_errors += 1;
+                tracing::warn!("polling tool job {id} failed, retrying: {e}");
+                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                continue;
+            }
+            Err(e) => return Err(e.into()),
+        };
+        if state.completed {
+            let result = state
+                .result
+                .map(|r| r.0)
+                .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
+            return Ok((result, state.success.unwrap_or(false)));
+        }
+        if !state.queued {
+            return Err(Error::internal_err(format!(
+                "tool job {id} is neither queued nor completed"
+            )));
+        }
+        tokio::time::sleep(interval).await;
+        interval = std::cmp::min(interval * 2, std::time::Duration::from_secs(1));
+    }
+}
+
+/// Hand a finished tool job's result to the model, the stream, the flow status and the chat.
+async fn report_tool_result(
+    ctx: &mut ToolExecutionContext<'_>,
+    tool_call: &OpenAIToolCall,
+    tool_module: &windmill_common::flows::FlowModule,
+    job_id: Uuid,
+    result: &RawValue,
+    success: bool,
+    is_ai_agent_output: bool,
+    messages: &mut Vec<OpenAIMessage>,
+    final_events_str: &mut String,
+) -> Result<(), Error> {
    // A nested agent returns the whole `AIAgentResult` envelope: on top of `output` it carries
    // the child's entire message history, stream log and token usage. Feeding that back would
    // grow the caller's context by the child's full transcript on every call, so the caller only
    // sees `output`. The envelope stays intact in the tool job's completed row.
-    let tool_result = if is_ai_agent_tool && job_success {
-        extract_ai_agent_output(&result).unwrap_or_else(|| result.get().to_string())
+    let tool_result = if is_ai_agent_output {
+        extract_ai_agent_output(result).unwrap_or_else(|| result.get().to_string())
    } else {
        result.get().to_string()
    };
@@ -1049,6 +1049,11 @@ pub async fn handle_ai_agent_job(
        GracefulPollOutcome::Ok(result) => Ok(result),
        GracefulPollOutcome::Timeout(ms) => {
            tracing::error!("AI agent timeout after {}s", ms / 1000);
+            // A tool dispatched to another worker group would otherwise still run later, after
+            // the agent that wanted its result is gone.
+            let reason = format!("parent AI agent {} timed out", job.id);
+            let cb = CanceledBy { username: None, reason: Some(reason) };
+            cleanup_orphaned_tool_jobs(db, &job.id, &job.workspace_id, Some(cb)).await;
            Err(Error::ExecutionErr(format!(
                "AI agent timeout after (>{}s)",
                ms / 1000
@@ -44,12 +44,13 @@ use windmill_common::flow_status::{
 };
 use windmill_common::flows::{add_virtual_items_if_necessary, Branch, FlowNodeId, StopAfterIf};
 use windmill_common::jobs::{
-    script_path_to_payload, JobKind, JobPayload, OnBehalfOf, RawCode, ENTRYPOINT_OVERRIDE,
+    script_path_to_payload, JobKind, JobPayload, OnBehalfOf, RawCode, TriggerKindLabel,
+    ENTRYPOINT_OVERRIDE,
 };
 use windmill_common::runnable_settings::{
    ConcurrencySettingsWithCustom, DebouncingSettings, RunnableSettingsTrait,
 };
-use windmill_common::scripts::{ScriptHash, ScriptRunnableSettingsInline};
+use windmill_common::scripts::{ScriptHash, ScriptLang, ScriptRunnableSettingsInline};
 use windmill_common::utils::WarnAfterExt;
 use windmill_common::worker::{error_to_value, to_raw_value, Connection};
 use windmill_common::{
@@ -1373,36 +1374,52 @@ pub async fn update_flow_status_after_job_completion_internal(
        }

        let step_counter = if inc_step_counter {
-            sqlx::query!(
-                "UPDATE v2_job_status
-                 SET flow_status = JSONB_SET(flow_status, ARRAY['step'], $1)
-                 WHERE id = $2",
-                json!(old_status.step + 1),
-                flow
-            )
-            .execute(&mut *tx)
-            .await
-            .map_err(|e| {
-                Error::internal_err(format!("error while setting flow index for {flow}: {e:#}"))
-            })?;
            old_status.step + 1
        } else {
            old_status.step
        };

-        // tracing::error!(
-        //     "step_counter: {:?} {} {inc_step_counter} {flow}",
-        //     step_counter,
-        //     old_status.step,
-        // );
-        // panic!("stop");
-
        /* is_last_step is true when the step_counter (the next step index) is an invalid index */
        let is_last_step = usize::try_from(step_counter)
            .map(|i| !(..old_status.modules.len()).contains(&i))
            .unwrap_or(true);

-        if let Some(new_status) = new_status.as_ref() {
+        let nresult = if let Some(nresult) = nresult {
+            // can be some either with early stop error or with the flow jobs results (was fetched to evaluate stop_early_after_all_iters but evaluated to false)
+            nresult
+        } else {
+            match &new_status {
+                Some(FlowStatusModule::Success { flow_jobs: Some(jobs), .. })
+                | Some(FlowStatusModule::Failure { flow_jobs: Some(jobs), .. }) => {
+                    Arc::new(retrieve_flow_jobs_results(&mut *tx, w_id, jobs).await?)
+                }
+                _ => result.clone(),
+            }
+        };
+
+        let remove_retry = old_status.retry.fail_count > 0
+            && matches!(&new_status, Some(FlowStatusModule::Success { .. }));
+
+        let special_step_status = new_status
+            .as_ref()
+            .filter(|_| is_failure_step || module_step.is_preprocessor_step());
+
+        let flow_job = if let Some(new_status) = special_step_status {
+            if inc_step_counter {
+                sqlx::query!(
+                    "UPDATE v2_job_status
+                     SET flow_status = JSONB_SET(flow_status, ARRAY['step'], $1)
+                     WHERE id = $2",
+                    json!(step_counter),
+                    flow
+                )
+                .execute(&mut *tx)
+                .await
+                .map_err(|e| {
+                    Error::internal_err(format!("error while setting flow index for {flow}: {e:#}"))
+                })?;
+            }
+
            if is_failure_step {
                let parent_module = sqlx::query_scalar!(
                     "SELECT flow_status->'failure_module'->>'parent_module' FROM v2_job_status WHERE id = $1",
@@ -1432,7 +1449,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                        "error while setting flow status in failure step: {e:#}"
                    ))
                })?;
-            } else if module_step.is_preprocessor_step() {
+            } else {
                sqlx::query!(
                    "UPDATE v2_job_status
                     SET flow_status = JSONB_SET(flow_status, ARRAY['preprocessor_module'], $1)
@@ -1447,69 +1464,71 @@ pub async fn update_flow_status_after_job_completion_internal(
                        "error while setting flow status in preprocessing step: {e:#}"
                    ))
                })?;
-            } else {
+            }
+
+            if remove_retry {
                sqlx::query!(
                    "UPDATE v2_job_status
-                     SET flow_status = JSONB_SET(flow_status, ARRAY['modules', $1::TEXT], $2)
-                     WHERE id = $3",
-                    old_status.step.to_string(),
-                    json!(new_status),
+                     SET flow_status = flow_status - 'retry'
+                     WHERE id = $1",
                    flow
                )
                .execute(&mut *tx)
                .await
-                .map_err(|e| {
-                    Error::internal_err(format!("error while setting new flow status: {e:#}"))
-                })?;
-
-                if let Some(job_result) = new_status.job_result() {
-                    sqlx::query!(
-                         "UPDATE v2_job_status
-                         SET flow_leaf_jobs = JSONB_SET(coalesce(flow_leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2)
-                         WHERE COALESCE((SELECT flow_innermost_root_job FROM v2_job WHERE id = $3), $3) = id",
-                         new_status.id(),
-                         json!(job_result),
-                         flow
-                     )
-                     .execute(&mut *tx)
-                     .await.map_err(|e| {
-                         Error::internal_err(format!(
-                             "error while setting leaf jobs: {e:#}"
-                         ))
-                     })?;
-                }
+                .context("remove flow status retry")?;
            }
-        }

-        let nresult = if let Some(nresult) = nresult {
-            // can be some either with early stop error or with the flow jobs results (was fetched to evaluate stop_early_after_all_iters but evaluated to false)
-            nresult
+            get_mini_pulled_job(&mut *tx, &flow).await?
        } else {
-            match &new_status {
-                Some(FlowStatusModule::Success { flow_jobs: Some(jobs), .. })
-                | Some(FlowStatusModule::Failure { flow_jobs: Some(jobs), .. }) => {
-                    Arc::new(retrieve_flow_jobs_results(&mut *tx, w_id, jobs).await?)
+            let module_status = new_status
+                .as_ref()
+                .map(|s| (old_status.step.to_string(), json!(s)));
+            let leaf_job = new_status
+                .as_ref()
+                .and_then(|s| s.job_result().map(|r| (s.id(), json!(r))));
+
+            let flow_job = if inc_step_counter || module_status.is_some() || remove_retry {
+                match advance_flow_status(
+                    &mut tx,
+                    flow,
+                    inc_step_counter.then_some(step_counter),
+                    module_status,
+                    leaf_job.as_ref(),
+                    remove_retry,
+                )
+                .await?
+                {
+                    Some(flow_job) => Some(flow_job),
+                    None => get_mini_pulled_job(&mut *tx, &flow).await?,
                }
-                _ => result.clone(),
+            } else {
+                get_mini_pulled_job(&mut *tx, &flow).await?
+            };
+
+            // The leaf jobs of a subflow are kept on its innermost root's row, which the
+            // statement above does not touch.
+            let innermost_root = flow_job
+                .as_ref()
+                .and_then(|j| j.flow_innermost_root_job)
+                .filter(|root| *root != flow);
+            if let (Some((leaf_id, leaf_result)), Some(root)) = (leaf_job, innermost_root) {
+                sqlx::query!(
+                    "UPDATE v2_job_status
+                     SET flow_leaf_jobs = JSONB_SET(coalesce(flow_leaf_jobs, '{}'::jsonb), ARRAY[$1::TEXT], $2)
+                     WHERE id = $3",
+                    leaf_id,
+                    leaf_result,
+                    root
+                )
+                .execute(&mut *tx)
+                .await
+                .map_err(|e| Error::internal_err(format!("error while setting leaf jobs: {e:#}")))?;
            }
+
+            flow_job
        };

-        if old_status.retry.fail_count > 0
-            && matches!(&new_status, Some(FlowStatusModule::Success { .. }))
-        {
-            sqlx::query!(
-                "UPDATE v2_job_status
-                 SET flow_status = flow_status - 'retry'
-                 WHERE id = $1",
-                flow
-            )
-            .execute(&mut *tx)
-            .await
-            .context("remove flow status retry")?;
-        }
-
-        let flow_job = get_mini_pulled_job(&mut *tx, &flow)
-            .await?
+        let flow_job = flow_job
            .ok_or_else(|| Error::internal_err(format!("requiring flow to be in the queue")))?;
        tx.commit().await?;

@@ -2304,6 +2323,100 @@ async fn set_success_and_duration_in_flow_job_success<'c>(
    Ok(())
 }

+/// Applies a step's edits to the flow's `v2_job_status` row and reads the flow back, in one
+/// statement: every UPDATE of that row writes a new row version carrying the whole
+/// `flow_status`, so the edits must not be split. `leaf_job` is only written here when the flow
+/// is its own innermost root; otherwise it belongs on the root's row and the caller writes it
+/// there. `Ok(None)` means nothing was written because the flow has no queued `v2_job_status`
+/// row; the caller then reads the flow with `get_mini_pulled_job`, whose LEFT JOIN still returns
+/// a queued flow that lacks a status row.
+async fn advance_flow_status(
+    tx: &mut Transaction<'_, Postgres>,
+    flow: Uuid,
+    step: Option<i32>,
+    module_status: Option<(String, Value)>,
+    leaf_job: Option<&(String, Value)>,
+    remove_retry: bool,
+) -> error::Result<Option<MiniPulledJob>> {
+    // An edit that does not apply gets an empty path, which JSONB_SET treats as a no-op. Its
+    // value must stay JSON `null`, never SQL NULL: JSONB_SET is strict and would null the whole
+    // `flow_status`. Each edit stays a JSONB_SET on its own path so that a malformed
+    // `flow_status` fails exactly as the equivalent separate UPDATEs would.
+    let (step_path, step) = match step {
+        Some(step) => (vec!["step"], json!(step)),
+        None => (vec![], Value::Null),
+    };
+    let (module_path, module_status) = match &module_status {
+        Some((index, status)) => (vec!["modules", index.as_str()], status),
+        None => (vec![], &Value::Null),
+    };
+    let (leaf_id, leaf_result) = leaf_job.map(|(id, r)| (id.as_str(), r)).unzip();
+    let removed_keys: &[&str] = if remove_retry { &["retry"] } else { &[] };
+
+    sqlx::query_as!(
+        MiniPulledJob,
+        "UPDATE v2_job_status SET
+            flow_status = JSONB_SET(
+                JSONB_SET(v2_job_status.flow_status, $2::TEXT[], $3),
+                $4::TEXT[], $5
+            ) - $6::TEXT[],
+            flow_leaf_jobs = CASE
+                WHEN $7::TEXT IS NULL
+                    OR COALESCE(v2_job.flow_innermost_root_job, v2_job_status.id) <> v2_job_status.id
+                THEN v2_job_status.flow_leaf_jobs
+                ELSE JSONB_SET(COALESCE(v2_job_status.flow_leaf_jobs, '{}'::JSONB), ARRAY[$7::TEXT], $8) END
+        FROM v2_job_queue INNER JOIN v2_job ON v2_job.id = v2_job_queue.id
+        WHERE v2_job_status.id = $1 AND v2_job_queue.id = $1
+        RETURNING
+            v2_job_queue.workspace_id,
+            v2_job_queue.id,
+            v2_job.args as \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\",
+            v2_job.parent_job,
+            v2_job.created_by,
+            v2_job_queue.started_at,
+            v2_job_queue.runnable_settings_handle,
+            v2_job_queue.scheduled_for,
+            v2_job.runnable_path,
+            v2_job.kind as \"kind: JobKind\",
+            v2_job.runnable_id as \"runnable_id: ScriptHash\",
+            v2_job_queue.canceled_reason,
+            v2_job_queue.canceled_by,
+            v2_job.permissioned_as,
+            v2_job.permissioned_as_email,
+            v2_job_status.flow_status as \"flow_status: sqlx::types::Json<Box<RawValue>>\",
+            v2_job.tag,
+            v2_job.script_lang as \"script_lang: ScriptLang\",
+            v2_job.same_worker,
+            v2_job.pre_run_error,
+            v2_job.concurrent_limit,
+            v2_job.concurrency_time_window_s,
+            v2_job.flow_innermost_root_job,
+            v2_job.root_job,
+            v2_job.timeout,
+            v2_job.flow_step_id,
+            v2_job.cache_ttl,
+            v2_job_queue.cache_ignore_s3_path,
+            v2_job_queue.priority,
+            v2_job.preprocessed,
+            v2_job.script_entrypoint_override,
+            v2_job.trigger,
+            v2_job.trigger_kind as \"trigger_kind: TriggerKindLabel\",
+            v2_job.visible_to_owner,
+            NULL as permissioned_as_end_user_email",
+        flow,
+        &step_path as &[&str],
+        step,
+        &module_path as &[&str],
+        module_status,
+        removed_keys as &[&str],
+        leaf_id,
+        leaf_result,
+    )
+    .fetch_optional(&mut **tx)
+    .await
+    .map_err(|e| Error::internal_err(format!("error while setting new flow status: {e:#}")))
+}
+
 async fn retrieve_flow_jobs_results<'c>(
    e: impl sqlx::PgExecutor<'c>,
    w_id: &str,
@@ -6185,7 +6298,7 @@ async fn payload_from_simple_module(
 pub fn raw_script_to_payload(
    path: String,
    content: String,
-    language: windmill_common::scripts::ScriptLang,
+    language: ScriptLang,
    lock: Option<String>,
    concurrency_settings: ConcurrencySettingsWithCustom,
    module: &FlowModule,
@@ -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.817.0";
+export const VERSION = "v1.818.0";

 export async function login(email: string, password: string): Promise<string> {
  return await windmill.UserService.login({
@@ -1,12 +1,12 @@
 {
  "name": "windmill-chat",
-  "version": "1.817.0",
+  "version": "1.818.0",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "windmill-chat",
-      "version": "1.817.0",
+      "version": "1.818.0",
      "license": "Apache-2.0",
      "devDependencies": {
        "@ai-sdk/react": "^4.0.102",
@@ -1,7 +1,7 @@
 {
  "name": "windmill-chat",
  "description": "Build chat interfaces on Windmill flows deployed in chat mode, from any frontend or raw app",
-  "version": "1.817.0",
+  "version": "1.818.0",
  "author": "Ruben Fiszel",
  "license": "Apache-2.0",
  "homepage": "https://github.com/windmill-labs/windmill/tree/main/chat-sdk#readme",
@@ -10,4 +10,4 @@ export const WM_FORK_PREFIX = "wm-fork";
 // (e.g. utils.ts) can read it without importing main.ts and creating a circular
 // dependency (main → workspace → utils → main) that triggers a TDZ.
 // Re-exported from main.ts for backwards compatibility.
-export const VERSION = "1.817.0";
+export const VERSION = "1.818.0";
@@ -1,12 +1,12 @@
 {
 	"name": "@windmill-labs/components",
-	"version": "1.817.0",
+	"version": "1.818.0",
 	"lockfileVersion": 3,
 	"requires": true,
 	"packages": {
 		"": {
 			"name": "@windmill-labs/components",
-			"version": "1.817.0",
+			"version": "1.818.0",
 			"hasInstallScript": true,
 			"license": "AGPL-3.0",
 			"dependencies": {
@@ -1,6 +1,6 @@
 {
 	"name": "@windmill-labs/components",
-	"version": "1.817.0",
+	"version": "1.818.0",
 	"scripts": {
 		"dev": "vite dev",
 		"dev:ui-builder": "mv static/ui_builder static/ui_builder.dev-disabled 2>/dev/null || true ; trap 'mv static/ui_builder.dev-disabled static/ui_builder 2>/dev/null || true' EXIT ; vite dev",
@@ -4,7 +4,7 @@ verify_ssl = true
 name = "pypi"

 [packages]
-wmill = ">=1.817.0"
+wmill = ">=1.818.0"
 sendgrid = "*"
 mysql-connector-python = "*"
 pymongo = "*"
@@ -1,7 +1,7 @@
 openapi: '3.0.3'

 info:
-  version: 1.817.0
+  version: 1.818.0
  title: OpenFlow Spec
  contact:
    name: Ruben Fiszel
@@ -12,7 +12,7 @@
    RootModule        = 'WindmillClient.psm1'

    # Version number of this module.
-    ModuleVersion     = '1.817.0'
+    ModuleVersion     = '1.818.0'

    # Supported PSEditions
    # CompatiblePSEditions = @()
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "wmill"
-version = "1.817.0"
+version = "1.818.0"
 description = "A client library for accessing Windmill server wrapping the Windmill client API"
 license = "Apache-2.0"
 homepage = "https://windmill.dev"
@@ -1,6 +1,6 @@
 {
  "name": "@windmill/windmill",
-  "version": "1.817.0",
+  "version": "1.818.0",
  "exports": "./src/index.ts",
  "publish": {
    "exclude": ["!src", "./s3Types.ts", "./sqlUtils.ts", "./client.ts", "./wacError.ts"]
@@ -1,7 +1,7 @@
 {
  "name": "windmill-client",
  "description": "Windmill SDK client for browsers and Node.js",
-  "version": "1.817.0",
+  "version": "1.818.0",
  "author": "Ruben Fiszel",
  "license": "Apache 2.0",
  "homepage": "https://github.com/windmill-labs/windmill/tree/main/typescript-client#readme",
@@ -1 +1 @@
-1.817.0
+1.818.0
@@ -1,12 +1,12 @@
 {
  "name": "windmill-yaml-validator",
-  "version": "1.817.0",
+  "version": "1.818.0",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "windmill-yaml-validator",
-      "version": "1.817.0",
+      "version": "1.818.0",
      "license": "Apache 2.0",
      "dependencies": {
        "@stoplight/yaml": "^4.3.0",
@@ -1,6 +1,6 @@
 {
  "name": "windmill-yaml-validator",
-  "version": "1.817.0",
+  "version": "1.818.0",
  "description": "YAML validator for Windmill flow, schedule, and trigger files",
  "main": "dist/index.js",
  "types": "dist/index.d.ts",
@@ -1 +1 @@
 .817.0
 .818.0