Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/1862-pool-shadow.added.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Workflows can now record a shadow verdict for a handoff before anything depends on it: the pool contract of #1853 runs on the handoff a caller passes to `assign`, and the answer lands on the run record as `shadow: { verdict, code?, reason? }` and as one `shadow: <verdict>` line in the run's own stage log. Nothing reads it — the run, the card and the pull request fast lane behave exactly as they did, which is the point of the shadow step ([#1862](https://github.com/ThinkFlowLab/FlyBot/issues/1862)).
40 changes: 36 additions & 4 deletions docs/subsystems/workflow.md
Original file line number Diff line number Diff line change
Expand Up @@ -71,13 +71,45 @@ Core never reads `$DSH_HOME/bot/flybot-workflow`. The plugin registers `flybot.t
`plugins/workflow/src/pool-contract.js` is the typed handoff contract of #1550 Stage 3 P0: one pure
`validateHandoff(handoff, ctx)` running the seven checks in order — target, owner, evidence,
concurrency, idempotency, cycle, quota — plus `escalationTarget(action)`, which names the Chief for
the two escalating refusals (`cycle-budget`, `quota`) and nobody else. It does no I/O, imports no
`service.js`, and **no dispatch path calls it**: no endpoint, tool, run, `index.js` or `service.js`
mentions it, so loading the plugin behaves exactly as before. P1 is the step that will call it;
until then `plugins/workflow/test/pool-contract.test.js` is its only caller, and
the two escalating refusals (`cycle-budget`, `quota`) and nobody else. It does no I/O and imports no
`service.js`. Since P1 (#1862) one path calls it, and only through the shadow step below.
`plugins/workflow/test/pool-contract.test.js` pins the contract itself, and
`core/test/task-handoff-route.test.js` feeds the host's own board facts to its target and owner
checks so the two policies cannot drift apart silently.

## Pool shadow (#1862)

`plugins/workflow/src/pool-shadow.js` is the P1 shadow step: `shadowVerdict(handoff, ctx)` runs the
P0 contract and returns one record — `{ verdict: 'pass', key }` or `{ verdict: 'reject', code?,
reason? }` — and `shadowLine(shadow)` turns it into the run's own stage line, `shadow: <verdict> ·
<code> · <reason>`. The line never begins `Verdict:`, so no reader and no gate can take it for a
conclusion about the pull request.

- **Two sources of facts, one answer.** `runHandoff(item, task, run, facts)` builds the handoff the
run already holds — its card, the node it hands the card to, that node's producer, the pool the
workflow declares (each node role, capped by `maxConcurrent`), the exec that owns it, and no
previously spent keys — and `assign()` accepts the card's own `acceptance`/`evidence`, so a real
pull request carries a verdict without any caller learning a new contract. A caller that has
better facts passes `handoff` (card, fromRole, toRole, attempt, acceptance, evidence) and
`handoffContext` (roles, pools, execOwner, worker, pool, cycleMax, quota, used), which replace the
derived pair wholesale; either way one pair, never mixed.
- **Record only.** `beginRun()` stores the answer on the run as `shadow: { verdict, code?, reason?,
key? }` — exposed by `publicRun()` — and posts the one line. A parked task carries all four fields
(`carriedTask()`), so a queued card is shadowed when it actually starts. Nothing reads the record:
run status, `nodeStatus` and the step count are the same with or without it, the card is never
re-queued, and the pull request fast lane still decides.
- **A dry run.** The contract records an accepted key in `ctx.seen`; the shadow pass hands it a copy,
so a shadow `pass` cannot make the next real handoff look like a duplicate. A handoff that cannot
be stored (not an object, circular, over 64 KB) is dropped, and a shadow that throws records
nothing — the shadow pass cannot stop a run. A card with no handoff facts, and no acceptance or
evidence, gets no record and no line, which is every run that existed before P1.
- The facts travel over the plugin's own `assign` / `workflow/state` seam only: no host read was
needed, so `core/src/host/` is unchanged.
- Tests: `plugins/workflow/test/pool-shadow.test.js` — a rejected and an accepted handoff, a card
shadowed from its own acceptance gate, a parked card, a dropped (circular) handoff, a run that
keeps walking its graph with a `reject` on the record, and the purity assertion that the caller's
`ctx.seen` is untouched.

## Tests

`plugins/workflow/test/`: graph advance and join, graph edit, assign and persistence, the FIFO queue and concurrency cap, Bot-reply advance, a dispatch that fails without being marked handed, a bounded hand-off retry and a step deadline that fails the run and clears its timer (#1530), a per-workflow `stepTimeoutMinutes` that replaces the ten-minute constant and is refused outside 1..120 (#1580), disposing releases the Bots before the record drops their ids and a released runner name is reusable again, assign-to-bot versus assign-to-workflow, inputs and outputs along edges, a loop that breaks and an if/else branch, failure recovery (onFail edges with max attempts, exhaustion behavior, and compatibility with failOnce and loop edges), templates, `define` / `runNamed`, the matcher, ephemeral spawn and cleanup, a stopped run with its settled node map (#1424), the reply middleware, the client canvas, the settings page (including a condition and a loop), rename and delete (including factory-id rows such as `delivery`), the stage chat's scroll rule (a new visible line follows, a filtered node reply moves nothing), the assign dialog's preselected card and its failed and empty board states (#1457, #1458), a sidebar click that leaves the Workflows section in place, and the claim rules — a re-routed card moves with one claim and one route, the same target writes nothing, a trigger never invalidates, the held list is capped and omits a Bot claim, `rematch: true` drops one claim, and the panel's one re-match action goes through `ingest` (#1647), and the assigned brief reaching the step prompt while a task without one still gets its title, a parked card keeping its brief on disk, and an unknown `assign` field coming back in `ignored` (#1869). Run `npm test` in `plugins/workflow`. The host seam that applies `ownerBotId` is `core/test/task-route.test.js`, which also pins that a card written through `track_tasks` reaches the listener (#1462). The host omission of run Bots, and the grant of `stepTools` plus the hidden-approval answer, is `core/test/flybot.test.js`. Which connectors a run Bot may call is `core/test/connector-access.test.js`, and the chain from a node's `tools` through `spawn` to that guard — the built-in delivery nodes included — is `core/test/workflow-node-connectors.test.js`.
Expand Down
103 changes: 103 additions & 0 deletions plugins/workflow/src/pool-shadow.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
// The shadow pass of #1550 Stage 3 P1 (#1862): the P0 contract (#1853) runs over a handoff that
// is already in hand, and its answer is only recorded. Nothing in this file decides anything — the
// pass reads plain facts and returns a record, so the pull request fast lane still decides whether
// a card moves. A verdict of `reject` here must leave the run and the card exactly as they were.
import { validateHandoff } from './pool-contract.js'

const MAX_REASON = 200

const text = (value) => (value === undefined || value === null ? '' : String(value))
const isObject = (value) => value !== null && typeof value === 'object' && !Array.isArray(value)

// The contract records an accepted key in `ctx.seen`. The shadow pass must not, or a real handoff
// that follows it would look like a duplicate, so every mutation-capable shape is copied first.
const cloneSeen = (seen) => {
if (seen instanceof Set) return new Set(seen)
if (seen instanceof Map) return new Map(seen)
if (Array.isArray(seen)) return [...seen]
if (isObject(seen)) return { ...seen }
return seen
}

const clip = (value) => {
const body = text(value).replace(/\s+/g, ' ').trim()
return body.length > MAX_REASON ? `${body.slice(0, MAX_REASON - 1)}…` : body
}

// One record, two shapes: a refusal names the check that produced it, so the shadow can be
// compared check by check with what the human path did; a pass names the key the real contract
// would have recorded, so the two passes can be joined when the shadow is read later.
export function shadowVerdict(handoff = {}, ctx = {}) {
let result
try {
result = validateHandoff(handoff, { ...ctx, seen: cloneSeen(ctx?.seen) })
} catch (error) {
return { verdict: 'reject', code: 'contract-error', reason: clip(error?.message ?? error) }
}
if (result?.ok !== true) {
return {
verdict: 'reject',
...(result?.code ? { code: result.code } : {}),
...(result?.reason ? { reason: clip(result.reason) } : {}),
}
}
return { verdict: 'pass', ...(result?.key ? { key: String(result.key) } : {}) }
}

const roleOf = (node) => text(node?.role).trim() || text(node?.name).trim()

// The handoff a run already holds: the card it carries, the node it is handing the card to, the
// node that produced it, its acceptance gate and evidence, and the exec that owns it. A root node
// has no producer, so its handoff names no `fromRole`. This is what lets a real pull request carry
// a verdict without any caller learning a new contract.
export function runHandoff(item = {}, task = {}, run = {}, facts = {}) {
const nodes = (Array.isArray(item.nodes) ? item.nodes : []).filter(isObject)
const target = nodes.find(node => node.id === facts.nodeId) ?? nodes[0]
const edge = (Array.isArray(item.edges) ? item.edges : []).find(candidate => (
isObject(candidate) && target?.id !== undefined && candidate.to === target.id && candidate.loop !== true && candidate.onFail !== true
))
const producer = edge ? nodes.find(node => node.id === edge.from) : undefined
const handoff = {
card: text(task.id) || text(run.id),
fromRole: roleOf(producer),
toRole: roleOf(target),
attempt: 0,
...(isObject(task.acceptance) ? { acceptance: task.acceptance } : {}),
...(Array.isArray(task.evidence) ? { evidence: task.evidence } : {}),
}
const roles = [...new Set(nodes.map(roleOf).filter(name => name !== ''))]
const declared = Number(item.maxConcurrent)
const workers = Number.isInteger(facts.workers) && facts.workers > 0
? facts.workers
: Number.isInteger(declared) && declared > 0 ? declared : 1
const inflight = Number.isFinite(Number(facts.inflight)) ? Number(facts.inflight) : 0
const ctx = {
roles,
// A workflow's declared node role is its pool, and its concurrency cap is how many cards that
// pool takes at once.
pools: Object.fromEntries(roles.map(role => [role, { workers }])),
// One exec per run: the Bot whose call started it, else the Bot that saved the graph, else the
// workflow row that owns the lane.
execOwner: text(task.callerBotId) || text(item.authoredBy) || text(item.id),
// The run holds no other card while this handoff is written.
worker: { wip: 0 },
pool: { inflight, workers },
seen: new Set(),
quota: {},
used: {},
}
return { handoff, ctx }
}

// The run's own comment: one line beginning `shadow:`, never `Verdict:`, so neither a reader nor a
// gate can mistake it for a conclusion about the pull request.
export function shadowLine(shadow) {
if (!shadow || typeof shadow.verdict !== 'string' || shadow.verdict === '') return null
const detail = [shadow.code, shadow.reason].filter((part) => typeof part === 'string' && part !== '')
return {
kind: 'system',
key: 'shadow: {verdict}',
params: { verdict: shadow.verdict },
text: [`shadow: ${shadow.verdict}`, ...detail].join(' · '),
}
}
95 changes: 93 additions & 2 deletions plugins/workflow/src/service.js
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
} from './graph.js'
import { matchTask, namedTarget } from './match.js'
import { linesFor } from './narrate.js'
import { runHandoff, shadowLine, shadowVerdict } from './pool-shadow.js'
import { DELIVERY_ID, DELIVERY_STEPS, WEEKLY_ID, deliveryWorkflow, templateById, templateCatalog, weeklyWorkflow } from './template.js'

const MAX_WORKFLOWS = 24
Expand All @@ -26,7 +27,7 @@
const MAX_HELD_TASKS = 20
// #1869: the fields `workflow/assign` reads. Anything else is named back in
// `ignored`, because a typo (`brif`) used to be dropped in silence.
const ASSIGN_FIELDS = new Set(['workflowId', 'taskId', 'id', 'title', 'brief', 'callerBotId', 'projectId', 'botId'])
const ASSIGN_FIELDS = new Set(['workflowId', 'taskId', 'id', 'title', 'brief', 'callerBotId', 'projectId', 'botId', 'handoff', 'handoffContext', 'acceptance', 'evidence'])
// A brief is a work order, not a sentence: 20–40 lines is the normal case. The cap
// is far above what a dispatch can carry, so it only stops an unbounded paste.
const MAX_BRIEF = 8000
Expand Down Expand Up @@ -56,6 +57,33 @@
return `${prefix}-${randomBytes(4).toString('hex')}`
}

// A caller hands facts about a pull request over as plain data; anything else is not a handoff
// and is dropped rather than refused (#1862), because the shadow pass must never stop a run.
function plainObject(value) {
return value !== null && typeof value === 'object' && !Array.isArray(value) ? value : null
}

// The handoff is persisted with the task, so it has to survive JSON: a circular or oversized object
// is dropped here instead of breaking `write()` or bloating the store. Dropping, not refusing, for
// the same reason as above.
const MAX_HANDOFF_CHARS = 64 * 1024

function storableJson(value) {
try {
return JSON.stringify(value).length <= MAX_HANDOFF_CHARS ? value : null
} catch {
return null
}
}

function storableHandoff(value) {
return plainObject(value) === null ? null : storableJson(value)
}

function storableList(value) {
return Array.isArray(value) ? storableJson(value) : null
}

function publicNode(node, bots) {
const local = (bots ?? []).filter(bot => bot && bot.remote !== true && bot.id && bot.name)
const seated = node.botId
Expand Down Expand Up @@ -137,6 +165,8 @@
...(run.failure ? { failure: { ...run.failure } } : {}),
...(run.earlyEnd ? { earlyEnd: run.earlyEnd } : {}),
...(run.ephemeral && Object.keys(run.ephemeral).length > 0 ? { ephemeral: { ...run.ephemeral } } : {}),
// #1862: the shadow pass's record, exposed so the run can be compared with the human path.
...(run.shadow ? { shadow: { ...run.shadow } } : {}),
}
}

Expand Down Expand Up @@ -202,13 +232,13 @@
}

function clearRunTimers(runId) {
for (const [key, entry] of [...timers]) {

Check warning on line 235 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-spread)

Using a spread operator here creates a new array unnecessarily.
if (entry.runId === runId) clearTimerAt(key)
}
for (const key of [...warned]) {

Check warning on line 238 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-spread)

Using a spread operator here creates a new array unnecessarily.
if (key.endsWith(`:${runId}`)) warned.delete(key)
}
for (const key of [...handOffTries.keys()]) {

Check warning on line 241 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-spread)

Using a spread operator here creates a new array unnecessarily.
if (key.startsWith(`${runId}:`)) handOffTries.delete(key)
}
}
Expand Down Expand Up @@ -239,7 +269,7 @@
// advance(); without this the step could sit unanswered forever (#1530).
if (run?.live !== true) return
const running = new Set(orderedIds(item, run.nodeStatus, 'running'))
for (const [key, entry] of [...timers]) {

Check warning on line 272 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-spread)

Using a spread operator here creates a new array unnecessarily.
if (entry.runId !== run.id) continue
if (run.status !== 'running' || run.paused === true || !running.has(entry.nodeId)) clearTimerAt(key)
}
Expand Down Expand Up @@ -465,7 +495,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')

Check warning on line 498 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-length-check)

Found a useless array length check
if (stuck) {
status[id] = 'failed'
blocked.push({ id, missing })
Expand All @@ -478,7 +508,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 } }

Check warning on line 511 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-fallback-in-spread)

Empty fallbacks in spreads are unnecessary
}
let result
if (op.type === 'start') result = { ...startFrontier(item, working.nodeStatus, working.fired), joined: false, layerSize: 0 }
Expand Down Expand Up @@ -641,7 +671,7 @@
break
}
handOffTries.delete(`${current.id}:${nodeId}`)
current = { ...current, handed: { ...current.handed, [nodeId]: true }, inputs: { ...(current.inputs ?? {}), [nodeId]: body.inputs } }

Check warning on line 674 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-fallback-in-spread)

Empty fallbacks in spreads are unnecessary
storeRun(item.id, current)
}
syncTimers(item, current)
Expand Down Expand Up @@ -735,6 +765,22 @@
}
}

// The facts a parked run needs when it starts later: who capped it, its Project, and the handoff
// the shadow pass reads (#1544, #1856, #1862).
function carriedTask(task) {
return {
id: task.id,
title: task.title,
...(task.brief ? { brief: task.brief } : {}),
...(task.callerBotId ? { callerBotId: task.callerBotId } : {}),
...(task.projectId ? { projectId: task.projectId } : {}),
...(task.handoff ? { handoff: task.handoff } : {}),
...(task.handoffContext ? { handoffContext: task.handoffContext } : {}),
...(task.acceptance ? { acceptance: task.acceptance } : {}),
...(task.evidence ? { evidence: task.evidence } : {}),
}
}

// The oldest queued task whose own workflow still has a slot, or null.
function takeNext() {
let best = null
Expand All @@ -751,7 +797,7 @@
const task = laneOf(best.item.id).queue.shift()
// #1869: the row is rebuilt from the parked fields, so the brief has to be
// named here too or it is lost exactly when the run starts.
return { item: best.item, task: { id: task.id, title: task.title, ...(task.brief ? { brief: task.brief } : {}), ...(task.callerBotId ? { callerBotId: task.callerBotId } : {}), ...(task.projectId ? { projectId: task.projectId } : {}) } }
return { item: best.item, task: carriedTask(task) }
}

function noteUnstarted(item, error) {
Expand Down Expand Up @@ -841,6 +887,30 @@
}
}

// #1862: the shadow pass. Two sources of handoff facts, one answer: the caller's own `handoff`
// when it has the facts in hand, else the run's card and graph (`runHandoff`) when the card
// carries an acceptance gate or evidence, which is what lets a real pull request carry a verdict.
// The answer is stored on the run and posted as one line — never read by anything that decides.
function shadowOf(item, task, run) {
const handoff = storableHandoff(task?.handoff)
const acceptance = plainObject(task?.acceptance)
const evidence = Array.isArray(task?.evidence) ? task.evidence : null
if (!handoff && !acceptance && !evidence) return null
try {
if (handoff) return shadowVerdict(handoff, storableHandoff(task?.handoffContext) ?? {})
const derived = runHandoff(item, task, run, {
nodeId: orderedIds(item, run.nodeStatus, 'running')[0],
inflight: runningOf(laneOf(item.id)).length,
workers: maxOf(item),
})
return shadowVerdict(derived.handoff, derived.ctx)
} catch (error) {
// A shadow that throws is a missing record, never a failed run.
warnOnce(`shadow:${run.id}`, `flybot-workflow: the shadow pass for run ${run.id} recorded nothing: ${error?.message ?? error}`)
return null
}
}

function beginRun(item, task, ephemeral) {
const holes = missingProviders(item)
if (holes.length > 0) return { ok: false, error: `Node ${holes[0].nodeId} input ${holes[0].name} has no upstream provider` }
Expand All @@ -859,9 +929,13 @@
], at)
const stepped = applyOp(item, run, { type: 'start' })
run = stepped.run
const shadow = shadowOf(item, task, run)
if (shadow) run = { ...run, shadow }
bindRunProject(item, task, run)
laneOf(item.id).active.push(run)
push(item.id, stepped.lines, at)
const line = shadowLine(shadow)
if (line) push(item.id, [line], at)
syncTimers(item, run)
write()
return { ok: true, run }
Expand Down Expand Up @@ -917,6 +991,11 @@
...(task.projectId ? { projectId: task.projectId } : {}),
// #1869: the brief stays on the parked card, on disk and until the run starts.
...(task.brief ? { brief: task.brief } : {}),
// #1862: and the handoff the shadow pass records a verdict for.
...(task.handoff ? { handoff: task.handoff } : {}),
...(task.handoffContext ? { handoffContext: task.handoffContext } : {}),
...(task.acceptance ? { acceptance: task.acceptance } : {}),
...(task.evidence ? { evidence: task.evidence } : {}),
})
const ahead = lane.queue.length - 1
push(item.id, [{
Expand Down Expand Up @@ -1169,6 +1248,18 @@
const wanted = body.projectId !== undefined ? body.projectId : item.projectId
const projectId = cleanLine(wanted ?? '', 120)
if (projectId !== '') task.projectId = projectId
// #1862: the caller may pass the facts of a pull request it holds; the shadow pass records the
// contract's answer for them and nothing reads it. `acceptance`/`evidence` are the card's own
// gate, so a card that carries them gets a verdict from `runHandoff` with no new contract. A
// value that cannot be stored is dropped, never refused.
const handoff = storableHandoff(body.handoff)
if (handoff) task.handoff = handoff
const handoffContext = storableHandoff(body.handoffContext)
if (handoffContext) task.handoffContext = handoffContext
const acceptance = storableHandoff(body.acceptance)
if (acceptance) task.acceptance = acceptance
const evidence = storableList(body.evidence)
if (evidence) task.evidence = evidence
if (typeof spawn !== 'function') return answer(enqueue(item, task))
return hostSerial(() => Promise.resolve(enqueue(item, task)).then(answer))
},
Expand Down Expand Up @@ -1748,7 +1839,7 @@
})
},
dispose() {
for (const key of [...timers.keys()]) clearTimerAt(key)

Check warning on line 1842 in plugins/workflow/src/service.js

View workflow job for this annotation

GitHub Actions / plugins (workflow)

unicorn(no-useless-spread)

Using a spread operator here creates a new array unnecessarily.
const ids = []
const runs = []
let endedLive = false
Expand Down
Loading
Loading