fix(workflow): a refused store write leaves no ghost workflow (#2598)

changelog.d/2598.fixed.md

@@ -0,0 +1 @@
+A workflow save that the store refused no longer leaves a half-created workflow in memory: the refusal is returned and the graph, its creation line and the sequence number are dropped, so the next successful save cannot persist a workflow the caller was told was not created; a refused rename keeps the name it could not save ([#2598](https://github.com/ThinkFlowLab/FlyBot/issues/2598)).

plugins/workflow/src/service.js

@@ -255,13 +255,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)
     }
   }
@@ -298,7 +298,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)
     }
@@ -571,7 +571,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 })
@@ -584,7 +584,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 }
@@ -776,7 +776,7 @@
         break
       }
       handOffTries.delete(`${current.id}:${nodeId}`)
       current = { ...current, handed: { ...current.handed, [nodeId]: true }, inputs: { ...(current.inputs ?? {}), [nodeId]: body.inputs } }
       storeRun(item.id, current)
     }
     syncTimers(item, current)
@@ -1270,12 +1270,27 @@
       if (!item) return { ok: false, error: 'No such workflow' }
       const name = cleanLine(body.name, 80)
       if (!name) return { ok: false, error: 'A workflow needs a name' }
-      item.name = name
-      if (body.description !== undefined) {
-        item.description = cleanLine(body.description, 400)
+      // #2598: the new name is memory until `write()` has landed it, so a
+      // refused write puts the old one back instead of listing a name the
+      // store never took.
+      const before = { name: item.name, description: item.description, hadDescription: Object.hasOwn(item, 'description') }
+      let written = false
+      try {
+        item.name = name
+        if (body.description !== undefined) {
+          item.description = cleanLine(body.description, 400)
+        }
+        write()
+        written = true
+        return present()
+      } catch (error) {
+        if (!written) {
+          item.name = before.name
+          if (before.hadDescription) item.description = before.description
+          else delete item.description
+        }
+        return refuse(error)
       }
-      write()
-      return present()
     },
     delete(body = {}) {
       if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' }
@@ -1324,6 +1339,14 @@
     },
     create(body = {}) {
       if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' }
+      // #2598: `write()` refuses before its first byte when the temp leaf is not
+      // writable, but the caller is told nothing was created. The push and its
+      // creation line below are memory, so remember what they change and undo
+      // them when the write refuses: the next successful write must not persist
+      // a graph the caller never got.
+      const before = { workflows: state.workflows.length, seq, stateSeq: state.seq }
+      let created = null
+      let written = false
       try {
         if (body.template) {
           const template = templateById(body.template)
@@ -1360,6 +1383,7 @@
           ...(Array.isArray(body.edges) ? { edges: body.edges } : { layers: body.layers }),
         })
         if (!item.name) throw new Error('A workflow needs a name')
+        created = item
         // #1544: the Bot that wrote the graph caps the connectors a run may reach from it.
         const authoredBy = cleanLine(body.callerBotId ?? body.authoredBy, 80) || undefined
         if (authoredBy) item.authoredBy = authoredBy
@@ -1378,8 +1402,15 @@
           text: `Workflow「${item.name}」已创建 · 成员 ${memberNames}`,
         }], now())
         write()
+        written = true
         return present()
       } catch (error) {
+        if (!written && created) {
+          state.workflows.length = before.workflows
+          delete state.logs[created.id]
+          seq = before.seq
+          state.seq = before.stateSeq
+        }
         return refuse(error)
       }
     },
@@ -2047,7 +2078,7 @@
       })
     },
     dispose() {
       for (const key of [...timers.keys()]) clearTimerAt(key)
       const ids = []
       const runs = []
       let endedLive = false

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

@@ -0,0 +1,75 @@
+// #2598: a store write that refuses before its first byte must leave memory as
+// it was. `write()` refuses when the temp leaf is a link (temp-name sweep), and
+// `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.
+import assert from 'node:assert/strict'
+import { mkdtempSync, readFileSync, rmSync, symlinkSync, writeFileSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { test } from 'node:test'
+import { createWorkflowService } from '../src/service.js'
+import { DELIVERY_ID } from '../src/template.js'
+
+const GRAPH = { name: 'Audit graph', nodes: [{ id: 'plan', name: 'Planner' }], edges: [] }
+
+// The store lands through `<dir>/workflows.json.<pid>.tmp`, so a symlink there
+// refuses the write before any byte reaches the store (tempWritable).
+function linkedStore() {
+  const dir = mkdtempSync(join(tmpdir(), 'flybot-wf-refused-'))
+  const svc = createWorkflowService({ dir, persist: true, now: () => 5, bots: () => [] })
+  const outside = join(dir, 'outside-workflows.json')
+  writeFileSync(outside, 'OUTSIDE\n')
+  const tmp = join(dir, `workflows.json.${process.pid}.tmp`)
+  symlinkSync(outside, tmp)
+  return { dir, svc, outside, tmp }
+}
+
+test('a refused create leaves no graph or creation line behind', () => {
+  const { dir, svc, outside, tmp } = linkedStore()
+  try {
+    const before = svc.state()
+    const refused = svc.create(GRAPH)
+    assert.equal(refused.ok, false, 'the write is refused')
+    assert.match(refused.error, /not a plain file/)
+    assert.equal(readFileSync(outside, 'utf8'), 'OUTSIDE\n', 'the file behind the temp link is not written through')
+
+    assert.deepEqual(svc.state().workflows.map(item => item.name), before.workflows.map(item => item.name), 'the refused create left no graph behind')
+
+    // The ghost used to land on disk with the next successful write.
+    rmSync(tmp)
+    const second = svc.create({ name: 'Second', nodes: [{ id: 'plan', name: 'Planner' }], edges: [] })
+    assert.equal(second.ok, true)
+    const stored = JSON.parse(readFileSync(join(dir, 'workflows.json'), 'utf8'))
+    assert.deepEqual(
+      stored.workflows.map(item => item.name),
+      [...before.workflows.map(item => item.name), 'Second'],
+      'the next write persists only the graph that was saved',
+    )
+    const created = new Set(stored.workflows.map(item => item.id))
+    assert.deepEqual(
+      Object.keys(stored.logs).filter(id => !created.has(id)),
+      [],
+      'a log line belongs to a workflow that is in the store',
+    )
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})
+
+test('a refused rename keeps the name it could not save', () => {
+  const { dir, svc } = linkedStore()
+  try {
+    const before = svc.state().workflows.find(item => item.id === DELIVERY_ID).name
+    const refused = svc.rename({ workflowId: DELIVERY_ID, name: 'Renamed' })
+    assert.equal(refused.ok, false, 'the write is refused')
+    assert.match(refused.error, /not a plain file/)
+    assert.equal(
+      svc.state().workflows.find(item => item.id === DELIVERY_ID).name,
+      before,
+      'the refused rename left the workflow name as it was',
+    )
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})