fix(workflow): tell a node the OUTPUTs its edges read

changelog.d/2669-edge-outputs.fixed.md

@@ -0,0 +1 @@
+A workflow node is now told which `OUTPUT` lines it owes from its own declared outputs **and** from the outputs its outgoing edges read, with the accepted values spelled out (`OUTPUT state: <one of: ready, blocked>`), so a node that declares no outputs no longer runs every step into `Missing OUTPUT state` and fails ([#2669](https://github.com/ThinkFlowLab/FlyBot/issues/2669)).

plugins/workflow/index.js

@@ -86,8 +86,15 @@
       for (const [name, value] of Object.entries(step.inputs ?? {})) {
         if (value) lines.push(`Input ${name}: ${value}`)
       }
-      const outputs = (step.outputs ?? []).map(port => port.name).filter(Boolean)
-      if (outputs.length > 0) lines.push(`When you finish, add one line per output: OUTPUT <name>: <value> (${outputs.join(', ')}).`)
+      // #2669: a port the outgoing edges compare against is printed as `<one of: …>`,
+      // so the step knows which values its own gate accepts.
+      const ports = (step.outputs ?? []).filter(port => port?.name)
+      if (ports.length > 0) {
+        const forms = ports.map(port => (port.values?.length
+          ? `OUTPUT ${port.name}: <one of: ${port.values.join(', ')}>`
+          : `OUTPUT ${port.name}: <value>`))
+        lines.push(`When you finish, add one line per output: ${forms.join(', ')}.`)
+      }
       lines.push('Reply with the result of this step.')
       return fb.bot.send({ from, to: step.botId, text: lines.join('\n'), role: 'request' })
     },
@@ -135,7 +142,7 @@
         task: { type: 'string', description: 'Task id, such as #1280. Omitted ids are generated.' },
       },
       // fb.tools.register hands the calling Bot to execute() as the second argument's botId (#1544).
       execute: (args, ctx) => service.runNamed({ ...(args ?? {}), ...(ctx?.botId ? { callerBotId: ctx.botId } : {}) }),
     }))
     offs.push(fb.tools.register({
       name: 'define_workflow',
@@ -147,7 +154,7 @@
         edges: { type: 'array', required: true, description: 'Hand-offs { from, to }. A fork is two edges out; a join is two edges in. A condition is when: { output, eq } or { else: true }. A loop is loop: true with max (1–8, default 3) and optional until: { output, eq }.' },
         stepTimeoutMinutes: { type: 'number', description: 'How long one step may wait for its Bot before the run fails, 1–120. Omit for the 45-minute default.' },
       },
       execute: (args, ctx) => service.define({ ...(args ?? {}), ...(ctx?.botId ? { callerBotId: ctx.botId } : {}) }),
     }))
   }
   if (typeof fb.bot?.send?.use === 'function') {

plugins/workflow/src/graph.js

@@ -645,6 +645,24 @@
   return null
 }
 
+// #2669: a node is told which OUTPUT lines to print, and the names come from its own
+// declared ports — but the edges leaving it read outputs too, so a node that declares
+// none was never told the line its gate reads. Name them, with the values those edges
+// compare against, so the step can answer the gate it is judged by.
+export function edgeOutputs(graph, nodeId) {
+  ensure(graph)
+  const values = new Map()
+  for (const edge of graph.edges) {
+    if (edge.from !== nodeId) continue
+    if (edge.when?.else === true || !edge.when?.output) continue
+    const name = edge.when.output
+    if (!values.has(name)) values.set(name, [])
+    const list = values.get(name)
+    if (edge.when.eq != null && !list.includes(edge.when.eq)) list.push(edge.when.eq)
+  }
+  return [...values].map(([name, list]) => ({ name, values: list }))
+}
+
 // #1773: the outputs a gate reads are verdicts, not prose. A `when`/`else` or `until`
 // edge decides from them, so a reply that leaves one blank is silence, and the gate
 // would take it for "not this branch". A step must not settle on that: name what it
@@ -756,8 +774,8 @@
   const index = layerIndex(graph, nodeId)
   const layer = graph.layers[index]
   const outputs = ctx.outputs && typeof ctx.outputs === 'object' ? ctx.outputs : {}
   const loops = { ...(ctx.loops ?? {}) }
   const fired = { ...(ctx.fired ?? {}) }
 
   // Check if this completion should end the run early (#1537, #1538)
   const earlyEnd = checkEarlyEnd(outputs, ctx.text)
@@ -836,14 +854,14 @@
       started: [],
       joined: false,
       finished: true,
       failAttempts: { ...(ctx.failAttempts ?? {}) },
       recovered: false,
       reset: [],
       earlyEnd: earlyEnd.reason,
     }
   }
 
   const failAttempts = { ...(ctx.failAttempts ?? {}) }
   const recovery = failRecoveryDecision(graph, nodeId, failAttempts)
 
   if (recovery) {

plugins/workflow/src/service.js

@@ -7,8 +7,8 @@
 import { join } from 'node:path'
 import { releaseAll, spawnAll } from './ephemeral.js'
 import {
-  chainLabel, cleanLine, cleanPorts, cleanPrompt, cleanTools, completeNode, cutCodePoints, doneCount, failNode, freshStatus, layerMarks,
-  firstBranchValue, missingProviders, missingVerdicts, nodeById, normalizeGraph, orderedIds, readStepResult, resolveInputs, retryNode,
+  chainLabel, cleanLine, cleanPorts, cleanPrompt, cleanTools, completeNode, cutCodePoints, doneCount, edgeOutputs, failNode, freshStatus,
+  layerMarks, firstBranchValue, missingProviders, missingVerdicts, nodeById, normalizeGraph, orderedIds, readStepResult, resolveInputs, retryNode,
   rowPreview, startFrontier, stepArtifacts, summarizeTask, unmetRequired, validateGraph,
 } from './graph.js'
 import { matchTask, namedTarget } from './match.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)
     }
   }
@@ -722,6 +722,25 @@
     return { run: next, lines }
   }
 
+  // #2669: the outgoing edges read outputs too, so a node that declares none has to be
+  // told the lines its own gate reads. A declared port wins on every other field.
+  function stepOutputs(item, node) {
+    const ports = (node.outputs ?? []).map(port => ({ ...port }))
+    const byName = new Map(ports.map(port => [port.name, port]))
+    for (const read of edgeOutputs(item, node.id)) {
+      let port = byName.get(read.name)
+      if (!port) {
+        port = { name: read.name }
+        byName.set(read.name, port)
+        ports.push(port)
+      }
+      const values = Array.isArray(port.values) ? [...port.values] : []
+      for (const value of read.values) if (!values.includes(value)) values.push(value)
+      if (values.length > 0) port.values = values
+    }
+    return ports
+  }
+
   function stepBody(item, run, node) {
     const inputs = resolveInputs(item, node.id, run.produced, run.task)
     return {
@@ -737,7 +756,7 @@
       brief: run.task?.brief || run.task?.title || '',
       taskId: run.task?.id ?? '',
       inputs,
-      outputs: node.outputs ?? [],
+      outputs: stepOutputs(item, node),
     }
   }
 

plugins/workflow/test/graph.test.js

@@ -1,8 +1,8 @@
 import assert from 'node:assert/strict'
 import { test } from 'node:test'
 import {
-  addNode, chainLabel, cleanLine, cleanPrompt, completeNode, connectNodes, cutCodePoints, doneCount, failNode, freshStatus, layerMarks,
-  moveNode, nextDemoOp, normalizeGraph, readStepResult, reconnectEdge, removeEdge, removeNode, retryNode,
+  addNode, chainLabel, cleanLine, cleanPrompt, completeNode, connectNodes, cutCodePoints, doneCount, edgeOutputs, failNode, freshStatus,
+  layerMarks, moveNode, nextDemoOp, normalizeGraph, readStepResult, reconnectEdge, removeEdge, removeNode, retryNode,
   rowPreview, startFrontier, summarizeTask, validateGraph,
 } from '../src/graph.js'
 import { deliveryWorkflow, weeklyWorkflow } from '../src/template.js'
@@ -220,3 +220,31 @@ test('#2581: cutCodePoints counts characters, not UTF-16 units', () => {
   assert.equal(cutCodePoints('a\uD800b', 10).isWellFormed(), true)
   assert.equal(cutCodePoints(undefined, 5), '')
 })
+
+// #2669: the gate reads the outputs of a node's outgoing edges, so the list has to
+// be readable from the graph the same way missingVerdicts reads it.
+test('#2669: edgeOutputs names the OUTPUT every outgoing edge reads, with its values', () => {
+  const graph = normalizeGraph({
+    nodes: [
+      { id: 'triage', name: 'Triage' },
+      { id: 'review', name: 'Review' },
+      { id: 'claim', name: 'Claim' },
+      { id: 'escalate', name: 'Escalate' },
+      { id: 'fix', name: 'Fix' },
+      { id: 'done', name: 'Done' },
+    ],
+    edges: [
+      { from: 'triage', to: 'claim', when: { output: 'state', eq: 'ready' } },
+      { from: 'triage', to: 'fix', when: { output: 'state', eq: 'ready' } },
+      { from: 'triage', to: 'escalate', when: { output: 'state', eq: 'blocked' } },
+      { from: 'triage', to: 'done', when: { else: true } },
+      { from: 'review', to: 'claim', when: { output: 'verdict', eq: 'PASS' } },
+      { from: 'claim', to: 'done' },
+    ],
+  })
+  assert.deepEqual(edgeOutputs(graph, 'triage'), [{ name: 'state', values: ['ready', 'blocked'] }],
+    'one entry per output name: values in edge order, deduped, and `else` contributes nothing')
+  assert.deepEqual(edgeOutputs(graph, 'review'), [{ name: 'verdict', values: ['PASS'] }])
+  assert.deepEqual(edgeOutputs(graph, 'claim'), [], 'a node whose edges read no output names none')
+  assert.deepEqual(edgeOutputs(graph, 'nope'), [])
+})

plugins/workflow/test/step-body.test.js

@@ -0,0 +1,118 @@
+// #2669: a step is only handed the outputs its node declares, while the outgoing
+// edges read outputs too. A graph whose `triage` node declares nothing and whose
+// edges branch on `state` therefore never told its node to print that line, and
+// every run of that shape died on `Missing OUTPUT state`.
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import { apply } from '../index.js'
+import { createWorkflowService } from '../src/service.js'
+
+const NOW = 1_700_000_000_000
+
+// `wf-1cdf6ec5`'s shape: triage declares no outputs; its edges read `state`.
+function fastLineGraph() {
+  return {
+    nodes: [
+      { id: 'triage', name: 'Triage', prompt: 'Triage the issue.' },
+      { id: 'claim', name: 'Claim', prompt: 'Claim the issue.' },
+      { id: 'escalate', name: 'Escalate', prompt: 'Report where it stopped.' },
+    ],
+    edges: [
+      { from: 'triage', to: 'claim', when: { output: 'state', eq: 'ready' } },
+      { from: 'triage', to: 'escalate', when: { else: true } },
+    ],
+  }
+}
+
+function serviceHarness(graph = fastLineGraph()) {
+  const dispatched = []
+  let seq = 0
+  const svc = createWorkflowService({
+    persist: false,
+    now: () => NOW,
+    bots: () => [],
+    spawn: async () => `eph-${++seq}`,
+    release: async () => {},
+    dispatch: async (step) => {
+      dispatched.push(step)
+      return { ok: true }
+    },
+  })
+  const made = svc.define({ name: 'Fast line', ...graph }).workflows.find(item => item.name === 'Fast line')
+  return { svc, id: made.id, dispatched }
+}
+
+const laneOf = (views, id) => views.workflows.find(item => item.id === id)
+
+test('#2669: a node that declares no outputs is handed the ones its edges read', async () => {
+  const { svc, dispatched } = serviceHarness()
+  await svc.runNamed({ workflow: 'Fast line', brief: 'i2669-impl' })
+  const triage = dispatched.find(step => step.nodeId === 'triage')
+  const state = triage.outputs.find(port => port.name === 'state')
+  assert.ok(state, 'the gate-read OUTPUT reaches the step even though the node declares none')
+  assert.deepEqual(state.values, ['ready'], 'and it carries the value the edge tests')
+})
+
+test('#2669: the handoff names the OUTPUT and the values it may take', async () => {
+  const sent = []
+  const roster = [
+    { id: 'bot-main', name: 'Chief', role: 'lead', isMain: true, remote: false },
+    { id: 'bot-worker', name: 'Worker', remote: false },
+  ]
+  const fb = {
+    bot: {
+      list: () => roster,
+      async create() { return 'bot-step-1' },
+      async remove() { return true },
+      async send(message) { sent.push(message); return { action: 'allow' } },
+    },
+    readOnly: () => false,
+    api: { handle() { return () => {} } },
+    tools: { register() { return () => {} } },
+    warn() {},
+  }
+  const started = apply({ flybot: { scope: () => fb }, effect(fn) { fn() } }, { persist: false })
+  started.service.define({ name: 'Fast line', ...fastLineGraph() })
+  await started.service.runNamed({ workflow: 'Fast line', brief: 'i2669-impl' })
+
+  const text = sent.map(message => message.text).join('\n')
+  assert.match(text, /OUTPUT state/, 'the node is told which OUTPUT line the gate reads')
+  assert.match(text, /one of: ready/, 'and which value satisfies the edge')
+})
+
+test('#2669: declared outputs are kept once, with the values their edges read', async () => {
+  const graph = {
+    nodes: [
+      {
+        id: 'triage',
+        name: 'Triage',
+        prompt: 'Triage the issue.',
+        outputs: [{ name: 'state', type: 'text' }, { name: 'summary', type: 'text' }],
+      },
+      { id: 'claim', name: 'Claim', prompt: 'Claim the issue.' },
+      { id: 'escalate', name: 'Escalate', prompt: 'Report where it stopped.' },
+    ],
+    edges: [
+      { from: 'triage', to: 'claim', when: { output: 'state', eq: 'ready' } },
+      { from: 'triage', to: 'escalate', when: { output: 'verdict', eq: 'PASS' } },
+    ],
+  }
+  const { svc, dispatched } = serviceHarness(graph)
+  await svc.runNamed({ workflow: 'Fast line', brief: 'i2669-impl' })
+  const triage = dispatched.find(step => step.nodeId === 'triage')
+  assert.deepEqual(triage.outputs.map(port => port.name), ['state', 'summary', 'verdict'], 'no declared port is lost or repeated')
+  assert.equal(triage.outputs.find(port => port.name === 'summary').type, 'text', 'a declared port keeps its own shape')
+  assert.deepEqual(triage.outputs.find(port => port.name === 'state').values, ['ready'], 'a declared port gains the values its edge reads')
+  assert.deepEqual(triage.outputs.find(port => port.name === 'verdict').values, ['PASS'])
+})
+
+test('#2669: the run walks the branch once the node prints the OUTPUT it was told about', async () => {
+  const { svc, id, dispatched } = serviceHarness()
+  let views = await svc.runNamed({ workflow: 'Fast line', brief: 'i2669-impl' })
+  views = await svc.stepResult({ botId: dispatched[0].botId, text: 'OUTPUT state: ready' })
+  const run = laneOf(views, id).run
+  assert.equal(run.nodeStatus.triage, 'done', 'the gate the node now knows about is satisfied')
+  assert.equal(run.nodeStatus.claim, 'running', 'the ready branch starts')
+  assert.equal(run.nodeStatus.escalate, 'skipped', 'the else route is not taken')
+  assert.equal(run.status, 'running')
+})