Put a deleted workflow's records back when the store write is refused

changelog.d/2623.fixed.md

@@ -0,0 +1 @@
+Deleting a Workflow whose store write is refused now puts the workflow and its lane, run, log, route, claim and brief records back before the refusal reaches the page, so the list and `workflows.json` cannot disagree and the delete can be retried instead of answering `No such workflow` until a restart ([#2623](https://github.com/ThinkFlowLab/FlyBot/issues/2623)).

docs/subsystems/workflow.md

@@ -48,7 +48,7 @@ Everything in `plugins/workflow/`: the graph rules, the run store, the matcher,
 ## Entry points
 
 - Endpoint `workflow/state` and `workflow/templates` are reads.
-- `workflow/create`, `workflow/rename`, `workflow/delete`, `workflow/assign`, `workflow/concurrency`, `workflow/graph`, `workflow/advance`, `workflow/reset`, `workflow/ingest` write. `workflow/concurrency` sets that workflow's `maxConcurrent` (an integer from 1 to 8); the same control is in Settings → Workflow graphs. `workflow/step-timeout` sets that workflow's `stepTimeoutMinutes` (an integer from 1 to 120, or `null` to go back to the default), and `workflow/create` and `define_workflow` accept the same field, refused outside 1..120 rather than dropped (#1868). Every write lands through `workflows.json.<pid>.tmp`, and that temp name is asked as a file first (temp-name sweep): a symbolic or hard link left there refuses the write, with the temp name in the answer, instead of putting the workflows outside their folder and renaming the link into place — `lstat` reports a hard link as a plain file although another name holds the same bytes, so the guard also requires `nlink === 1`, while a crashed host's own plain leftover stays replaceable. `workflow/rename` updates a workflow's name and optional description; read-only hosts are refused. `workflow/delete` removes a workflow from state, releases its ephemeral Bots if a run exists, and cleans logs/runs/routes and every task claim that named it, so the next `ingest` of that task matches again instead of asking `assign` for a workflow that no longer exists; a load still drops a workflow claim whose workflow is gone, so a store written before this fix heals (#1632). `workflow/ingest` also takes `rematch: true`: that one task's claim and route are dropped before matching, the panel's per-task action (#1647). `workflow/assign` reads `taskId` (or `id`), `title`, `brief` (stored on the card, and the card's own brief wins on a later call that omits it), `callerBotId`, `projectId` and `botId`; any other key is named in the response's `ignored` array instead of being dropped in silence (#1869).
+- `workflow/create`, `workflow/rename`, `workflow/delete`, `workflow/assign`, `workflow/concurrency`, `workflow/graph`, `workflow/advance`, `workflow/reset`, `workflow/ingest` write. `workflow/concurrency` sets that workflow's `maxConcurrent` (an integer from 1 to 8); the same control is in Settings → Workflow graphs. `workflow/step-timeout` sets that workflow's `stepTimeoutMinutes` (an integer from 1 to 120, or `null` to go back to the default), and `workflow/create` and `define_workflow` accept the same field, refused outside 1..120 rather than dropped (#1868). Every write lands through `workflows.json.<pid>.tmp`, and that temp name is asked as a file first (temp-name sweep): a symbolic or hard link left there refuses the write, with the temp name in the answer, instead of putting the workflows outside their folder and renaming the link into place — `lstat` reports a hard link as a plain file although another name holds the same bytes, so the guard also requires `nlink === 1`, while a crashed host's own plain leftover stays replaceable. `workflow/rename` updates a workflow's name and optional description; read-only hosts are refused. `workflow/delete` removes a workflow from state, releases its ephemeral Bots if a run exists, and cleans logs/runs/routes and every task claim that named it, so the next `ingest` of that task matches again instead of asking `assign` for a workflow that no longer exists; a load still drops a workflow claim whose workflow is gone, so a store written before this fix heals (#1632). A delete whose write is refused puts the workflow and its lane, run, log, route, claim and brief records back before the refusal reaches the caller, so the page and `workflows.json` cannot disagree and the delete can be retried (#2623). `workflow/ingest` also takes `rematch: true`: that one task's claim and route are dropped before matching, the panel's per-task action (#1647). `workflow/assign` reads `taskId` (or `id`), `title`, `brief` (stored on the card, and the card's own brief wins on a later call that omits it), `callerBotId`, `projectId` and `botId`; any other key is named in the response's `ignored` array instead of being dropped in silence (#1869).
 - Tool `run_workflow` starts a named run with a brief. Tool `define_workflow` writes the same graph the canvas edits, and takes the same `stepTimeoutMinutes` (1..120) so a lane created that way does not silently inherit the 45-minute default (#1868).
 - `flybot.tasks.onWrite` calls `ingest`. A new Bot match returns `{ ownerBotId }`. A Workflow match starts the run inside the plugin and leaves the card's owner to the host. A host that offers no `tasks.onWrite` warns once and reports `tasksWired: false` in `workflow/state`, and the Workflows panel says that a board card cannot start a run there (#1462).
 

plugins/workflow/src/service.js

@@ -260,13 +260,13 @@
   }
 
   function clearRunTimers(runId) {
     for (const [key, entry] of [...timers]) {
       if (entry.runId === runId) clearTimerAt(key)
     }
     for (const key of [...warned]) {
       if (key.endsWith(`:${runId}`)) warned.delete(key)
     }
     for (const key of [...handOffTries.keys()]) {
       if (key.startsWith(`${runId}:`)) handOffTries.delete(key)
     }
   }
@@ -303,7 +303,7 @@
     // keep its deadline: without it the run holds its lane slot with nothing left
     // to free it until the owner presses Stop again (P-560).
     const stopRetry = run.status === 'running' && run.paused === true && run.releasePending === true
     for (const [key, entry] of [...timers]) {
       if (entry.runId !== run.id) continue
       if (run.status !== 'running' || (run.paused === true && !stopRetry) || !running.has(entry.nodeId)) clearTimerAt(key)
     }
@@ -591,7 +591,7 @@
       const missing = unmetRequired(item, id, run.produced, run.task)
       if (missing.length === 0) { started.push(id); continue }
       const preds = item.edges.filter(edge => edge.to === id && edge.loop !== true).map(edge => edge.from)
       const stuck = preds.length === 0 || preds.every(pred => status[pred] === 'done' || status[pred] === 'failed' || status[pred] === 'skipped')
       if (stuck) {
         status[id] = 'failed'
         blocked.push({ id, missing })
@@ -604,7 +604,7 @@
   function applyOp(item, run, op) {
     let working = run
     if (op.type === 'complete' && op.outputs && typeof op.outputs === 'object') {
       working = { ...run, produced: { ...(run.produced ?? {}), [op.nodeId]: op.outputs } }
     }
     let result
     if (op.type === 'start') result = { ...startFrontier(item, working.nodeStatus, working.fired), joined: false, layerSize: 0 }
@@ -1331,6 +1331,17 @@
         .filter(Boolean)
         .flatMap(run => Object.values(run.ephemeral ?? {})))]
       const finish = () => {
+        // #2623: everything removed below is memory until `write()` has landed
+        // it. A write that refuses puts it all back, so the page and the store
+        // cannot disagree about whether the workflow was deleted and the owner
+        // can retry instead of being told "No such workflow".
+        const index = state.workflows.indexOf(item)
+        const lane = state.lanes[workflowId]
+        const run = state.runs[workflowId]
+        const log = state.logs[workflowId]
+        const routes = Object.entries(state.routes).filter(([, route]) => route?.id === workflowId)
+        const claims = Object.entries(state.claims ?? {}).filter(([, claim]) => claim?.kind === 'workflow' && claim.id === workflowId)
+        const briefs = Object.entries(state.briefs ?? {}).filter(([, brief]) => brief?.workflowId === workflowId)
         state.workflows = state.workflows.filter(workflow => workflow.id !== workflowId)
         delete state.lanes[workflowId]
         delete state.runs[workflowId]
@@ -1349,7 +1360,18 @@
         Object.keys(state.briefs ?? {}).forEach(taskId => {
           if (state.briefs[taskId]?.workflowId === workflowId) delete state.briefs[taskId]
         })
-        write()
+        try {
+          write()
+        } catch (error) {
+          state.workflows.splice(index, 0, item)
+          if (lane !== undefined) state.lanes[workflowId] = lane
+          if (run !== undefined) state.runs[workflowId] = run
+          if (log !== undefined) state.logs[workflowId] = log
+          for (const [taskId, route] of routes) state.routes[taskId] = route
+          for (const [taskId, claim] of claims) state.claims[taskId] = claim
+          for (const [taskId, brief] of briefs) state.briefs[taskId] = brief
+          throw error
+        }
         return present()
       }
       if (typeof release !== 'function' || releasing.length === 0) return finish()

plugins/workflow/test/refused-write-rollback.test.js

@@ -3,8 +3,11 @@
 // `create()` had already pushed the graph and its creation line by then, so the
 // caller was told nothing was created while the next successful write persisted
 // it. `rename()` writes before it keeps the new name for the same reason.
+// #2623: `delete()` removed the workflow and its lane/run/log/route/claim/brief
+// records before `write()`, so a refused write left the page without a workflow
+// the store still held and a retry answered "No such workflow".
 import assert from 'node:assert/strict'
-import { mkdtempSync, readFileSync, rmSync, symlinkSync, writeFileSync } from 'node:fs'
+import { mkdirSync, mkdtempSync, readFileSync, rmSync, symlinkSync, writeFileSync } from 'node:fs'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { test } from 'node:test'
@@ -57,6 +60,47 @@ test('a refused create leaves no graph or creation line behind', () => {
   }
 })
 
+// A directory at the temp leaf refuses the write the same way a link does, and it
+// is the shape #2623 reproduces with: `delete()` had already dropped the workflow
+// and its lane, run, log, route, claim and brief records from memory.
+test('a refused delete keeps the workflow for the page and the next host', async () => {
+  const dir = mkdtempSync(join(tmpdir(), 'flybot-wf-delete-refused-'))
+  try {
+    const svc = createWorkflowService({ dir, persist: true, now: () => 5, bots: () => [] })
+    const created = svc.create(GRAPH)
+    assert.equal(created.ok, true)
+    const id = created.workflows.find(item => item.name === GRAPH.name).id
+    const tmp = join(dir, `workflows.json.${process.pid}.tmp`)
+    mkdirSync(tmp)
+    const before = svc.state().workflows.map(item => item.id)
+
+    let thrown
+    try {
+      await svc.delete({ workflowId: id })
+    } catch (error) {
+      thrown = error
+    }
+    assert.ok(thrown, 'the refused write reaches the caller instead of answering ok')
+    assert.match(thrown.message, /not a plain file/)
+    assert.deepEqual(svc.state().workflows.map(item => item.id), before, 'the page still lists the workflow the store kept')
+
+    const fresh = createWorkflowService({ dir, persist: true, now: () => 6, bots: () => [] })
+    assert.ok(fresh.state().workflows.some(item => item.id === id), 'a fresh host on the same home still lists it')
+
+    // Removing the block is what makes the retry possible at all: before this
+    // fix the workflow was gone from state, so the retry answered "No such
+    // workflow" until a restart brought it back from disk.
+    rmSync(tmp, { recursive: true, force: true })
+    const retried = await svc.delete({ workflowId: id })
+    assert.equal(retried.ok, true, 'the retry removes the workflow once the store can be written')
+    assert.equal(svc.state().workflows.some(item => item.id === id), false, 'the page no longer lists it')
+    const stored = JSON.parse(readFileSync(join(dir, 'workflows.json'), 'utf8'))
+    assert.equal(stored.workflows.some(item => item.id === id), false, 'the store no longer lists it')
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})
+
 test('a refused rename keeps the name it could not save', () => {
   const { dir, svc } = linkedStore()
   try {