From 2a0223472e5f76500e4cf05c3c73a0c5e10967d7 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Fri, 5 Jun 2026 09:29:04 -0700 Subject: [PATCH] Address PR review feedback (#1432) - delimiter-safe foreach instance-id resolution via IR-validated candidates - restart watcher handles binding removal/defer-flip and re-keys the reverse heartbeat guard on agent change/delete - governing-node stamp owned by the pass-initiating foreach instance (race fix) - editor: policy-escalation confirm/retry handshake on save; foreach children inherit the override note; project-scoped agent fetches; picker disabled on registry fetch error; column-agent strings in the canonical i18n catalog - tests: update-tool escalation surface, PATCH escalation route, parity bound to run-derived stages, try/finally cleanup, deterministic event waits, symmetric defer surfaces, binding-release regression --- docs/workflow-steps.md | 2 +- .../__tests__/column-agent-resolver.test.ts | 22 +++ packages/core/src/column-agent-resolver.ts | 57 +++++-- .../core/src/workflow-definition-types.ts | 7 + .../app/components/WorkflowColumnPanel.tsx | 6 +- .../app/components/WorkflowNodeEditor.tsx | 67 ++++++-- .../src/__tests__/workflow-routes.test.ts | 20 +++ .../engine/src/__tests__/agent-tools.test.ts | 156 +++++++++++++----- .../executor-column-agent-principal.test.ts | 39 ++++- .../src/__tests__/executor-test-helpers.ts | 9 + .../workflow-graph-executor-parity.test.ts | 8 +- packages/engine/src/executor.ts | 151 ++++++++++++----- packages/i18n/locales/en/app.json | 16 ++ 13 files changed, 437 insertions(+), 123 deletions(-) diff --git a/docs/workflow-steps.md b/docs/workflow-steps.md index 763413955d..448644370b 100644 --- a/docs/workflow-steps.md +++ b/docs/workflow-steps.md @@ -65,7 +65,7 @@ The `workflowColumns` track introduces **IR v2** (`version: "v2"`), where a work A v2 column can optionally name a **permanent agent** from the agent registry, staffing every card that flows through it once instead of node-by-node or task-by-task. The binding is a first-class optional field on the column (not a trait — traits are board-transition policy; this is execution identity): -``` +```ts { id: "review", name: "Review", traits: [], agent: { agentId: "agent-001", mode: "defer" | "override" } } ``` diff --git a/packages/core/src/__tests__/column-agent-resolver.test.ts b/packages/core/src/__tests__/column-agent-resolver.test.ts index 5fd19903ee..e322cb4954 100644 --- a/packages/core/src/__tests__/column-agent-resolver.test.ts +++ b/packages/core/src/__tests__/column-agent-resolver.test.ts @@ -79,6 +79,15 @@ describe("resolveEffectiveAgent — precedence matrix (U2)", () => { ).toEqual({ source: "column-agent", agentId: "col-agent" }); }); + it("defer × lone modelId (incomplete pair, no agentId) → column agent wins", () => { + // Symmetric incomplete-pair surface (FN-5893: assert the invariant across + // ALL known surfaces, not only the provider-only reproduction). + expect(resolveEffectiveAgent({ binding: deferBinding, ownModelId: "claude-x" })).toEqual({ + source: "column-agent", + agentId: "col-agent", + }); + }); + it("defer × bare → column agent wins", () => { expect(resolveEffectiveAgent({ binding: deferBinding })).toEqual({ source: "column-agent", @@ -191,6 +200,19 @@ describe("resolveColumnAgentBinding — foreach instance inheritance (U2)", () = const nodeId = instanceNodeId("fe", 0, "se"); expect(resolveColumnAgentBinding(ir, nodeId)).toBeUndefined(); }); + + it("resolves bindings when the foreach node id itself contains '#'", () => { + // The instance-id format is delimiter-ambiguous; the resolver validates each + // candidate split against real foreach nodes instead of trusting the first '#' + // (PR #1432 review). + const ir = foreachIr({ foreachColumn: "review", reviewAgent: overrideBinding }); + const fe = ir.nodes.find((n) => n.id === "fe"); + if (!fe) throw new Error("fixture foreach missing"); + fe.id = "fe#a"; + const nodeId = instanceNodeId("fe#a", 0, "se"); + expect(nodeId).toBe("fe#a#0:se"); + expect(resolveColumnAgentBinding(ir, nodeId)).toEqual(overrideBinding); + }); }); describe("instanceNodeId / parseInstanceNodeId round-trip (U2)", () => { diff --git a/packages/core/src/column-agent-resolver.ts b/packages/core/src/column-agent-resolver.ts index 6ec7dc579b..639de05519 100644 --- a/packages/core/src/column-agent-resolver.ts +++ b/packages/core/src/column-agent-resolver.ts @@ -48,10 +48,20 @@ export interface ParsedInstanceNodeId { * `nodeId` is not in instance form. Defensive against `templateNodeId` itself * containing `:` — split on the FIRST `#`, then the FIRST `:` of the remainder, * and keep everything after that as the template node id. The `templateNodeId` is - * not sanitized against `:`, so a greedy/last-delimiter split would corrupt it. */ + * not sanitized against `:`, so a greedy/last-delimiter split would corrupt it. + * + * NOTE: a `foreachNodeId` that itself contains `#` is ambiguous under any single + * split. Callers that hold the IR should use {@link parseInstanceNodeIdCandidates} + * and validate each candidate's `foreachNodeId` against the graph (as + * `resolveColumnAgentBinding` does) instead of trusting one split position. */ export function parseInstanceNodeId(nodeId: string): ParsedInstanceNodeId | undefined { const hashIndex = nodeId.indexOf("#"); if (hashIndex < 0) return undefined; + return parseInstanceNodeIdAt(nodeId, hashIndex); +} + +/** Parse treating the `#` at `hashIndex` as the instance-id delimiter. */ +function parseInstanceNodeIdAt(nodeId: string, hashIndex: number): ParsedInstanceNodeId | undefined { const foreachNodeId = nodeId.slice(0, hashIndex); const remainder = nodeId.slice(hashIndex + 1); const colonIndex = remainder.indexOf(":"); @@ -65,6 +75,21 @@ export function parseInstanceNodeId(nodeId: string): ParsedInstanceNodeId | unde return { foreachNodeId, stepIndex, templateNodeId }; } +/** Every plausible parse of `nodeId` as an instance id — one candidate per `#` + * whose suffix matches the `:` shape. The id format is ambiguous when + * node ids themselves contain `#` (e.g. foreach `f#a`, instance `f#a#0:t` — both + * the first and second `#` look like delimiters), so callers with access to the + * graph validate each candidate's `foreachNodeId` against real foreach nodes + * rather than committing to a single split position. Ordered left-to-right. */ +export function parseInstanceNodeIdCandidates(nodeId: string): ParsedInstanceNodeId[] { + const candidates: ParsedInstanceNodeId[] = []; + for (let i = nodeId.indexOf("#"); i >= 0; i = nodeId.indexOf("#", i + 1)) { + const parsed = parseInstanceNodeIdAt(nodeId, i); + if (parsed) candidates.push(parsed); + } + return candidates; +} + // ── Binding lookup ─────────────────────────────────────────────────────────── /** Index a graph's top-level nodes by id (handles v1 + v2 shapes). */ @@ -104,22 +129,26 @@ export function resolveColumnAgentBinding( } // Foreach instance node: resolve against the enclosing foreach, honoring a - // template node's own declared column. - const parsed = parseInstanceNodeId(nodeId); - if (!parsed) return undefined; + // template node's own declared column. The instance-id format is ambiguous when + // node ids contain `#`, so try every plausible split and accept the first whose + // foreachNodeId names a REAL foreach node in this graph — a single fixed split + // (first-# or last-#) silently bypasses bindings for ids on the other side of + // the ambiguity (PR #1432 review). + for (const parsed of parseInstanceNodeIdCandidates(nodeId)) { + const foreachNode = nodesById.get(parsed.foreachNodeId); + if (!foreachNode || foreachNode.kind !== "foreach") continue; - const foreachNode = nodesById.get(parsed.foreachNodeId); - if (!foreachNode || foreachNode.kind !== "foreach") return undefined; + const cfg = foreachNode.config as Partial | undefined; + const templateNodes = cfg?.template?.nodes ?? []; + const templateNode = templateNodes.find((n) => n.id === parsed.templateNodeId); - const cfg = foreachNode.config as Partial | undefined; - const templateNodes = cfg?.template?.nodes ?? []; - const templateNode = templateNodes.find((n) => n.id === parsed.templateNodeId); - - // Template node's own column wins; otherwise inherit the foreach node's column. - if (templateNode?.column !== undefined) { - return bindingForColumn(templateNode.column); + // Template node's own column wins; otherwise inherit the foreach node's column. + if (templateNode?.column !== undefined) { + return bindingForColumn(templateNode.column); + } + return bindingForColumn(foreachNode.column); } - return bindingForColumn(foreachNode.column); + return undefined; } // ── Effective-agent precedence (defer / override) ──────────────────────────── diff --git a/packages/core/src/workflow-definition-types.ts b/packages/core/src/workflow-definition-types.ts index 60aee809e4..412bfca6c0 100644 --- a/packages/core/src/workflow-definition-types.ts +++ b/packages/core/src/workflow-definition-types.ts @@ -48,6 +48,13 @@ export interface WorkflowDefinitionUpdate { * the `workflowColumns` flag is ON. */ rehomeTo?: string; + /** + * Column-agent policy escalation (column-agent plan R13): set true to confirm + * binding a column agent whose permission policy is broader than the project + * default. Without it, the write surfaces (dashboard routes, fn_workflow_* + * tools) reject such bindings with a typed policy-escalation error. + */ + confirmPolicyEscalation?: boolean; /** * U11/KTD-13: when an IR update changes a custom field's type incompatibly for * tasks that already hold a value under that field, the update is blocked with diff --git a/packages/dashboard/app/components/WorkflowColumnPanel.tsx b/packages/dashboard/app/components/WorkflowColumnPanel.tsx index 74c11558d8..e712a43eba 100644 --- a/packages/dashboard/app/components/WorkflowColumnPanel.tsx +++ b/packages/dashboard/app/components/WorkflowColumnPanel.tsx @@ -142,7 +142,11 @@ export function WorkflowColumnPanel({ [columns, setColumnAgent], ); - const agentPickerDisabled = readOnly || !columnAgentsEnabled || agentsLoading; + // `!!agentsError` (PR #1432 review): when the registry fetch failed, the select + // would render enabled with only "(none)" while the bound id has no matching + // option — interacting with it could silently clear a binding. Disabled while + // the registry is unavailable, consistent with the loading guard. + const agentPickerDisabled = readOnly || !columnAgentsEnabled || agentsLoading || !!agentsError; const workflowWide = violations.filter((v) => v.columnId === null); const violationsFor = useCallback( diff --git a/packages/dashboard/app/components/WorkflowNodeEditor.tsx b/packages/dashboard/app/components/WorkflowNodeEditor.tsx index bc5f094515..b025a98e32 100644 --- a/packages/dashboard/app/components/WorkflowNodeEditor.tsx +++ b/packages/dashboard/app/components/WorkflowNodeEditor.tsx @@ -475,15 +475,42 @@ function InnerEditor({ columns.length ? columns : undefined, fields.length ? fields : undefined, ); - const updated = await updateWorkflow(activeWorkflow.id, { ir, layout }, projectId); - setWorkflows((ws) => ws.map((w) => (w.id === updated.id ? updated : w))); - // Validate by compiling — surfaces non-linear graphs as a banner. + const finishSave = async (updated: Awaited>) => { + setWorkflows((ws) => ws.map((w) => (w.id === updated.id ? updated : w))); + // Validate by compiling — surfaces non-linear graphs as a banner. + try { + await compileWorkflow(updated.id, projectId); + addToast(t("workflows.saved", "Workflow saved"), "success"); + } catch (compileErr) { + setValidationError( + getErrorMessage(compileErr) || t("workflows.savedNotCompilable", "Workflow saved but cannot be compiled"), + ); + } + }; try { - await compileWorkflow(updated.id, projectId); - addToast(t("workflows.saved", "Workflow saved"), "success"); - } catch (compileErr) { - setValidationError( - getErrorMessage(compileErr) || t("workflows.savedNotCompilable", "Workflow saved but cannot be compiled"), + await finishSave(await updateWorkflow(activeWorkflow.id, { ir, layout }, projectId)); + } catch (err) { + // Policy-escalation handshake (R13, PR #1432 review): the route rejects a + // binding to a broader-than-default agent until the author explicitly + // confirms. Surface the server's explanation, then retry with the flag — + // otherwise such bindings would be unsavable from the dashboard. + // Shape-checked rather than `instanceof ApiRequestError` so test doubles + // (and any error wrapper) that carry the details payload still route here. + const escalation = + (err as { details?: { policyEscalation?: boolean } } | null)?.details?.policyEscalation === true; + if (!escalation) throw err; + const proceed = window.confirm( + `${getErrorMessage(err)}\n\n${t( + "workflowColumns.confirmPolicyEscalation", + "Bind it anyway? The column agent will run with broader permissions than this project's default.", + )}`, + ); + if (!proceed) { + addToast(t("workflowColumns.escalationDeclined", "Save cancelled — column agent binding not confirmed"), "error"); + return; + } + await finishSave( + await updateWorkflow(activeWorkflow.id, { ir, layout, confirmPolicyEscalation: true }, projectId), ); } } catch (err) { @@ -568,12 +595,20 @@ function InnerEditor({ // column agent" note so authors don't diagnose override as a bug (R11). Keyed // on the column id + binding, not array identity. const overrideColumnBinding = useMemo(() => { - const columnId = selectedNode?.data.column; + // Foreach template children don't carry their own column in irToFlow — they + // inherit the enclosing foreach group's column at execution (R4). Mirror that + // inheritance here so a step-execute prompt inside an override-bound foreach + // still shows the note (PR #1432 review). + const columnId = + selectedNode?.data.column + ?? (selectedNode?.parentId + ? nodes.find((n) => n.id === selectedNode.parentId)?.data.column + : undefined); if (!columnId) return undefined; const col = columns.find((c) => c.id === columnId); if (!col?.agent || col.agent.mode !== "override") return undefined; return col.agent; - }, [selectedNode?.data.column, columns]); + }, [selectedNode?.data.column, selectedNode?.parentId, nodes, columns]); // Resolve the override agent's display name from the loaded registry; when the // id is stale (not in the list) fall back to the not-found treatment. @@ -596,7 +631,10 @@ function InnerEditor({ addToast(getErrorMessage(err) || "Failed to load models", "error"); }); } else if (currentExecutor === "agent" && agents.length === 0) { - fetchAgents().then(setAgents).catch((err) => { + // Project-scoped, matching WorkflowColumnPanel's fetchAgents(undefined, + // projectId) — an unscoped fetch returns the wrong registry in + // multi-project deployments (PR #1432 review). + fetchAgents(undefined, projectId).then(setAgents).catch((err) => { addToast(getErrorMessage(err) || "Failed to load agents", "error"); }); } else if (currentExecutor === "skill" && skills.length === 0) { @@ -621,7 +659,10 @@ function InnerEditor({ useEffect(() => { if (!overrideColumnBinding || agents.length > 0) return; let cancelled = false; - Promise.resolve(fetchAgents()).then((list) => { + // Project-scoped (PR #1432 review): without projectId this resolves from the + // wrong scope in multi-project deployments — the override note would show a + // false "not found" for a perfectly valid project agent. + Promise.resolve(fetchAgents(undefined, projectId)).then((list) => { if (!cancelled) setAgents(list ?? []); }).catch((err) => { if (!cancelled) addToast(getErrorMessage(err) || "Failed to load agents", "error"); @@ -629,7 +670,7 @@ function InnerEditor({ return () => { cancelled = true; }; - }, [overrideColumnBinding, agents.length, addToast]); + }, [overrideColumnBinding, agents.length, projectId, addToast]); const overlayProps = useOverlayDismiss(onClose); diff --git a/packages/dashboard/src/__tests__/workflow-routes.test.ts b/packages/dashboard/src/__tests__/workflow-routes.test.ts index f96324be06..77cd98d98c 100644 --- a/packages/dashboard/src/__tests__/workflow-routes.test.ts +++ b/packages/dashboard/src/__tests__/workflow-routes.test.ts @@ -627,4 +627,24 @@ describe("workflow routes — column agents (U6)", () => { expect(res.status).toBe(400); expect(res.body.error).toMatch(/triage/); }); + + it("PATCH enforces the policy-escalation gate the same way as POST (FN-5893)", async () => { + await store.updateSettings({ defaultAgentPermissionPolicy: { rules: { file_write_delete: "block" } } as never }); + const agentId = await makeAgent({ permissionPolicy: { presetId: "unrestricted" } }); + + const created = await post("/api/workflows", { name: "EditableEsc", ir: boundIr() }); + expect(created.status).toBe(201); + const id = (created.body as { id: string }).id; + + const denied = await patch(`/api/workflows/${id}`, { ir: boundIr({ agentId, mode: "override" }) }); + expect(denied.status).toBe(400); + expect(denied.body.error).toMatch(/broader/i); + expect((denied.body as { details?: { policyEscalation?: boolean } }).details?.policyEscalation).toBe(true); + + const ok = await patch(`/api/workflows/${id}`, { + ir: boundIr({ agentId, mode: "override" }), + confirmPolicyEscalation: true, + }); + expect(ok.status).toBe(200); + }); }); diff --git a/packages/engine/src/__tests__/agent-tools.test.ts b/packages/engine/src/__tests__/agent-tools.test.ts index c86c113c9e..54a476a61f 100644 --- a/packages/engine/src/__tests__/agent-tools.test.ts +++ b/packages/engine/src/__tests__/agent-tools.test.ts @@ -576,52 +576,124 @@ describe("createWorkflowCreateTool", () => { const rootDir = await mkdtemp(join(tmpdir(), "wf-tool-ca-root-")); const globalDir = await mkdtemp(join(tmpdir(), "wf-tool-ca-global-")); const store = new core.TaskStore(rootDir, globalDir); - await store.init(); - // Restrict the project default; the bound agent is unrestricted (broader). - await store.updateSettings({ defaultAgentPermissionPolicy: { rules: { file_write_delete: "block" } } } as any); - const agentStore = new core.AgentStore({ rootDir: store.getFusionDir() }); - await agentStore.init(); - const agent = await agentStore.createAgent({ - name: "Privileged", - role: "executor", - permissionPolicy: { presetId: "unrestricted" }, - } as any); + try { + await store.init(); + // Restrict the project default; the bound agent is unrestricted (broader). + await store.updateSettings({ defaultAgentPermissionPolicy: { rules: { file_write_delete: "block" } } } as any); + const agentStore = new core.AgentStore({ rootDir: store.getFusionDir() }); + await agentStore.init(); + const agent = await agentStore.createAgent({ + name: "Privileged", + role: "executor", + permissionPolicy: { presetId: "unrestricted" }, + } as any); - const ir = { - version: "v2", - name: "bound", - columns: [ - { id: "triage", name: "Triage", traits: [{ trait: "intake" }], agent: { agentId: agent.id, mode: "override" } }, - { id: "done", name: "Done", traits: [{ trait: "complete" }] }, - ], - nodes: [ - { id: "start", kind: "start", column: "triage" }, - { id: "work", kind: "prompt", column: "triage", config: { prompt: "do" } }, - { id: "end", kind: "end", column: "done" }, - ], - edges: [ - { from: "start", to: "work", condition: "success" }, - { from: "work", to: "end", condition: "success" }, - ], - }; - const tool = createWorkflowCreateTool(store as any); + const ir = { + version: "v2", + name: "bound", + columns: [ + { id: "triage", name: "Triage", traits: [{ trait: "intake" }], agent: { agentId: agent.id, mode: "override" } }, + { id: "done", name: "Done", traits: [{ trait: "complete" }] }, + ], + nodes: [ + { id: "start", kind: "start", column: "triage" }, + { id: "work", kind: "prompt", column: "triage", config: { prompt: "do" } }, + { id: "end", kind: "end", column: "done" }, + ], + edges: [ + { from: "start", to: "work", condition: "success" }, + { from: "work", to: "end", condition: "success" }, + ], + }; + const tool = createWorkflowCreateTool(store as any); - const denied = await tool.execute("c", { name: "Esc", ir } as any, undefined, undefined, {} as any); - expect((denied as { isError?: boolean }).isError).toBe(true); - const text = denied.content[0]?.type === "text" ? denied.content[0].text : ""; - expect(text).toMatch(/triage/); - expect(text).toMatch(/confirm_policy_escalation: true/); - expect(denied.details).toMatchObject({ columnId: "triage", agentId: agent.id, reason: "policy-escalation" }); + const denied = await tool.execute("c", { name: "Esc", ir } as any, undefined, undefined, {} as any); + expect((denied as { isError?: boolean }).isError).toBe(true); + const text = denied.content[0]?.type === "text" ? denied.content[0].text : ""; + expect(text).toMatch(/triage/); + expect(text).toMatch(/confirm_policy_escalation: true/); + expect(denied.details).toMatchObject({ columnId: "triage", agentId: agent.id, reason: "policy-escalation" }); - // With the flag set, the gate passes and the store write proceeds. - const ok = await tool.execute("c", { name: "Esc2", ir, confirm_policy_escalation: true } as any, undefined, undefined, {} as any); - expect((ok as { isError?: boolean }).isError).toBeFalsy(); - const okText = ok.content[0]?.type === "text" ? ok.content[0].text : ""; - expect(okText).toMatch(/Created workflow/); + // With the flag set, the gate passes and the store write proceeds. + const ok = await tool.execute("c", { name: "Esc2", ir, confirm_policy_escalation: true } as any, undefined, undefined, {} as any); + expect((ok as { isError?: boolean }).isError).toBeFalsy(); + const okText = ok.content[0]?.type === "text" ? ok.content[0].text : ""; + expect(okText).toMatch(/Created workflow/); + } finally { + store.close(); + await rm(rootDir, { recursive: true, force: true }); + await rm(globalDir, { recursive: true, force: true }); + } + }); - store.close(); - await rm(rootDir, { recursive: true, force: true }); - await rm(globalDir, { recursive: true, force: true }); + // FN-5893: the escalation invariant must hold on ALL workflow write surfaces — + // the update tool is the second one (the dashboard route has its own tests). + it("update tool enforces the same policy-escalation gate", async () => { + const rootDir = await mkdtemp(join(tmpdir(), "wf-tool-ca-upd-root-")); + const globalDir = await mkdtemp(join(tmpdir(), "wf-tool-ca-upd-global-")); + const store = new core.TaskStore(rootDir, globalDir); + try { + await store.init(); + await store.updateSettings({ defaultAgentPermissionPolicy: { rules: { file_write_delete: "block" } } } as any); + const agentStore = new core.AgentStore({ rootDir: store.getFusionDir() }); + await agentStore.init(); + const agent = await agentStore.createAgent({ + name: "Privileged", + role: "executor", + permissionPolicy: { presetId: "unrestricted" }, + } as any); + + const boundIr = (name: string) => ({ + version: "v2", + name, + columns: [ + { id: "triage", name: "Triage", traits: [{ trait: "intake" }], agent: { agentId: agent.id, mode: "override" } }, + { id: "done", name: "Done", traits: [{ trait: "complete" }] }, + ], + nodes: [ + { id: "start", kind: "start", column: "triage" }, + { id: "work", kind: "prompt", column: "triage", config: { prompt: "do" } }, + { id: "end", kind: "end", column: "done" }, + ], + edges: [ + { from: "start", to: "work", condition: "success" }, + { from: "work", to: "end", condition: "success" }, + ], + }); + + // Seed an unbound workflow to update. + const unbound = { ...boundIr("plain"), columns: boundIr("plain").columns.map(({ agent: _a, ...c }) => c) }; + const created = await store.createWorkflowDefinition({ name: "plain", ir: unbound as any }); + + const tool = createWorkflowUpdateTool(store as any); + const denied = await tool.execute( + "c", + { workflow_id: created.id, ir: boundIr("bound") } as any, + undefined, + undefined, + {} as any, + ); + expect((denied as { isError?: boolean }).isError).toBe(true); + const text = denied.content[0]?.type === "text" ? denied.content[0].text : ""; + expect(text).toMatch(/triage/); + expect(text).toMatch(/confirm_policy_escalation: true/); + expect(denied.details).toMatchObject({ columnId: "triage", agentId: agent.id, reason: "policy-escalation" }); + + const ok = await tool.execute( + "c", + { workflow_id: created.id, ir: boundIr("bound"), confirm_policy_escalation: true } as any, + undefined, + undefined, + {} as any, + ); + expect((ok as { isError?: boolean }).isError).toBeFalsy(); + const okText = ok.content[0]?.type === "text" ? ok.content[0].text : ""; + expect(okText).toMatch(/Updated workflow/); + } finally { + store.close(); + await rm(rootDir, { recursive: true, force: true }); + await rm(globalDir, { recursive: true, force: true }); + } }); }); diff --git a/packages/engine/src/__tests__/executor-column-agent-principal.test.ts b/packages/engine/src/__tests__/executor-column-agent-principal.test.ts index ee7f7f464a..024064dde2 100644 --- a/packages/engine/src/__tests__/executor-column-agent-principal.test.ts +++ b/packages/engine/src/__tests__/executor-column-agent-principal.test.ts @@ -453,8 +453,7 @@ describe("column-agent principal alignment (plan U5)", () => { // and lastEffectiveColumnAgentId = agent-X — matching the agent's model. const { setModel } = activeGraphSession(executor, task.id, "exec-node", OVERRIDE_COL); - store._trigger("task:updated", task); - await new Promise((r) => setTimeout(r, 0)); + await store._triggerAsync("task:updated", task); // No agent change, no model change → no hot-swap. expect(setModel).not.toHaveBeenCalled(); @@ -485,8 +484,7 @@ describe("column-agent principal alignment (plan U5)", () => { const { setModel } = activeGraphSession(executor, task.id, "exec-node", OVERRIDE_COL); - store._trigger("task:updated", task); - await new Promise((r) => setTimeout(r, 0)); + await store._triggerAsync("task:updated", task); // The legacy block is short-circuited under override: the assigned/own model // (openai/gpt-edited) is NEVER applied via setModel. @@ -499,6 +497,36 @@ describe("column-agent principal alignment (plan U5)", () => { expect((executor as any).activeSessions.get(task.id).lastEffectiveColumnAgentId).toBe("agent-X"); }); + it("binding removed by a workflow edit → session reverts to own-settings model and the reverse guard releases", async () => { + // PR #1432 review: when the binding disappears (or defer re-resolves to own + // settings) the watcher must hand the session back to normal resolution — + // hot-swap to the assigned/task model, clear the tracked column agent, and + // release isAgentEffectivelyExecuting() for the old agent. + const store = createMockStore(); + const find = vi.fn().mockReturnValue({ provider: "openai", modelId: "gpt-y" }); + const task = singleSessionTask({ assignedAgentId: "agent-Y" }); + const { executor } = makeExecutor(store, { + "agent-Y": makeAssignedAgent({ id: "agent-Y", runtimeConfig: { model: "openai/gpt-y" } }), + }); + (executor as any)._modelRegistry = { find }; + + const { setModel } = activeGraphSession(executor, task.id, "exec-node", OVERRIDE_COL); + // The workflow edit removed the binding: re-seed the resolver to yield none, + // and mark X as effectively executing so we can observe the release. + seedSeam(executor, task.id, "exec-node", undefined); + (executor as any).effectiveColumnAgentByTask.set(task.id, "agent-X"); + + await store._triggerAsync("task:updated", task); + + // Session reverted to the assigned agent's model. + expect(find).toHaveBeenCalledWith("openai", "gpt-y"); + expect(setModel).toHaveBeenCalledWith({ provider: "openai", modelId: "gpt-y" }); + // Column-agent tracking cleared; reverse heartbeat guard released. + expect((executor as any).activeSessions.get(task.id).lastEffectiveColumnAgentId).toBeNull(); + expect(executor.isAgentEffectivelyExecuting("agent-X")).toBe(false); + expect(loggedLines(store).some((l) => l.includes("binding released"))).toBe(true); + }); + it("legacy entry (no effective column agent) → the column-invalidation block is skipped", async () => { const store = createMockStore(); const find = vi.fn(); @@ -519,8 +547,7 @@ describe("column-agent principal alignment (plan U5)", () => { }); // No seam slots seeded. - store._trigger("task:updated", task); - await new Promise((r) => setTimeout(r, 0)); + await store._triggerAsync("task:updated", task); // The column-invalidation block never ran (no column-agent fetch / swap). expect(loggedLines(store).some((l) => l.includes("Column agent changed"))).toBe(false); diff --git a/packages/engine/src/__tests__/executor-test-helpers.ts b/packages/engine/src/__tests__/executor-test-helpers.ts index 2ad1be5736..a6fd6d1431 100644 --- a/packages/engine/src/__tests__/executor-test-helpers.ts +++ b/packages/engine/src/__tests__/executor-test-helpers.ts @@ -315,6 +315,15 @@ export function createMockStore() { _trigger(event: string, ...args: unknown[]) { for (const fn of listeners.get(event) || []) fn(...args); }, + /** Like `_trigger`, but awaits every (possibly async) listener — deterministic + * synchronization for tests asserting NEGATIVE outcomes after an event + * (e.g. "setModel was NOT called"), where `vi.waitFor` cannot apply and a + * bare `setTimeout(0)` is a brittle real-timer wait. */ + async _triggerAsync(event: string, ...args: unknown[]) { + await Promise.allSettled( + (listeners.get(event) || []).map((fn) => Promise.resolve(fn(...args))), + ); + }, emit: vi.fn(), listTasks: vi.fn().mockResolvedValue([]), getTask: vi.fn().mockResolvedValue({ diff --git a/packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts b/packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts index dea42fe0e5..3d7ad33a18 100644 --- a/packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts +++ b/packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts @@ -10,7 +10,7 @@ // suite `stepwise-workflow-parity.test.ts`. Keep the two concerns separate. // ───────────────────────────────────────────────────────────────────────────── import { describe, expect, it, vi } from "vitest"; -import type { TaskDetail, WorkflowIrV2 } from "@fusion/core"; +import type { TaskDetail, WorkflowIrV2, WorkflowStage } from "@fusion/core"; import { BUILTIN_CODING_WORKFLOW_IR, buildWorkflowObservation, @@ -179,6 +179,10 @@ describe("column-agent feature is invisible when unbound (U7 / R9)", () => { experimentalFeatures: { workflowGraphExecutor: true }, }); expect(result.outcome).toBe("success"); + // Bind the invariant to actual executor behavior (PR #1432 review): the + // observation below derives from the run-captured seam sequence, so seam + // drift fails here instead of being masked by a hard-coded literal. + expect(stages).toEqual(["execute", "review", "merge"]); // Legacy authoritative observation: a clean run that lands in `done`/merged. const legacyObs = buildWorkflowObservationFromTask( @@ -187,7 +191,7 @@ describe("column-agent feature is invisible when unbound (U7 / R9)", () => { ); // Interpreter (binding-free) observation assembled from the same run. const interpreterObs = buildWorkflowObservation({ - stageTransitions: ["triage", "execute", "review", "merge"], + stageTransitions: ["triage", ...stages] as WorkflowStage[], terminalColumn: "done", terminalStatus: "done", reviewVerdict: "approve", diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index f6a20aa7c8..151b059a41 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -2107,12 +2107,51 @@ export class TaskExecutor { const governingNodeId = this.graphSeamGoverningNodeId.get(task.id)!; const resolveBinding = this.graphColumnAgentResolver.get(task.id)!; const binding = resolveBinding(governingNodeId); - if (binding) { - const effective = resolveEffectiveAgent({ - binding, - ...this.extractOwnSettings(task), - }); - if (effective.source === "column-agent") { + const effective = binding + ? resolveEffectiveAgent({ binding, ...this.extractOwnSettings(task) }) + : undefined; + if (!effective || effective.source !== "column-agent") { + // Binding RELEASED (PR #1432 review): a workflow edit removed the + // binding, or `defer` now resolves to the task's own settings. Hand the + // session back to normal resolution: hot-swap to the assigned/task + // model (the same resolution the legacy block below owns), clear the + // column-agent tracking, and release the reverse heartbeat guard so + // isAgentEffectivelyExecuting() stops blocking the OLD agent. + executorLog.log(`${task.id}: column-agent binding released — reverting session to own-settings resolution`); + activeEntry.lastEffectiveColumnAgentId = null; + this.effectiveColumnAgentByTask.delete(task.id); + // Fire-and-forget audit (matches the deletion-fallback posture above). + this.store.logEntry( + task.id, + "Column-agent binding released — session reverts to its own model/agent resolution", + undefined, + this.getRunContextFor(task.id), + ).catch((err: unknown) => executorLog.warn(`${task.id}: failed to log column-agent release: ${err instanceof Error ? err.message : String(err)}`)); + const settings = await this.store.getSettings(); + const assignedRuntimeConfig = await this.getAssignedAgentRuntimeConfig(task.assignedAgentId); + const { provider: ownProvider, modelId: ownModelId } = resolveExecutorSessionModel( + task.modelProvider, + task.modelId, + settings, + assignedRuntimeConfig, + ); + const providerChanged = ownProvider !== activeEntry.lastResolvedModelProvider; + const modelIdChanged = ownModelId !== activeEntry.lastResolvedModelId; + if ((providerChanged || modelIdChanged) && ownProvider && ownModelId) { + activeEntry.lastResolvedModelProvider = ownProvider; + activeEntry.lastResolvedModelId = ownModelId; + try { + const model = this.modelRegistry.find(ownProvider, ownModelId); + if (model) { + await activeEntry.session.setModel(model); + executorLog.log(`${task.id}: binding released — model reverted to ${ownProvider}/${ownModelId}`); + } + } catch (err: unknown) { + executorLog.error(`${task.id}: failed to revert model after binding release: ${err instanceof Error ? err.message : String(err)}`); + } + } + } else { + { // Fetch the (possibly changed) effective column agent, best-effort. const newAgent = await this.options.agentStore?.getAgent(effective.agentId).catch(() => null) ?? null; if (!newAgent) { @@ -2131,6 +2170,10 @@ export class TaskExecutor { this.getRunContextFor(task.id), ).catch((err: unknown) => executorLog.warn(`${task.id}: failed to log column-agent deletion fallback: ${err instanceof Error ? err.message : String(err)}`)); activeEntry.lastEffectiveColumnAgentId = null; + // Release the reverse heartbeat guard for the deleted agent + // (PR #1432 review): isAgentEffectivelyExecuting() must not keep + // blocking an agent that no longer governs this session. + this.effectiveColumnAgentByTask.delete(task.id); } } else { const settings = await this.store.getSettings(); @@ -2145,6 +2188,9 @@ export class TaskExecutor { const modelIdChanged = newModelId !== activeEntry.lastResolvedModelId; if (agentChanged || providerChanged || modelIdChanged) { activeEntry.lastEffectiveColumnAgentId = newAgent.id; + // Re-key the reverse heartbeat guard to the NEW agent (PR #1432 + // review): the old agent stops being blocked, the new one starts. + this.effectiveColumnAgentByTask.set(task.id, newAgent.id); activeEntry.lastResolvedModelProvider = newProvider; activeEntry.lastResolvedModelId = newModelId; if (newProvider && newModelId) { @@ -4311,6 +4357,7 @@ export class TaskExecutor { task: Task, stepIndex: number, instanceId?: string, + governingNodeId?: string, ): Promise<{ success: boolean; error?: string }> { // Pin step-session physics for the run before the implementation pass. this.graphStepSessionPinned.add(task.id); @@ -4325,8 +4372,27 @@ export class TaskExecutor { // in-flight callers within a single attempt still share the one promise. let phase = this.graphStepRunOnce.get(task.id); if (!phase) { + // Column-agent governing-node ownership (PR #1432 review): the slot is + // written ONLY by the caller that CREATES the memoized pass, and cleared + // when that pass settles. One step-session pass serves every foreach + // instance, so the session-identity binding is the pass-INITIATING + // instance's — deterministic, instead of concurrent seam invocations + // racing set/delete on a shared per-task slot (parallel foreach could + // otherwise stamp another instance's node mid-build or clear it before + // the session resolved the binding). + if (typeof governingNodeId === "string") { + this.graphSeamGoverningNodeId.set(task.id, governingNodeId); + } phase = this.runImplementationPhase(task); this.graphStepRunOnce.set(task.id, phase); + void phase + .catch(() => undefined) + .finally(() => { + // Clear only our own stamp — a rework re-run may have installed a new one. + if (typeof governingNodeId === "string" && this.graphSeamGoverningNodeId.get(task.id) === governingNodeId) { + this.graphSeamGoverningNodeId.delete(task.id); + } + }); } try { await phase; @@ -4484,45 +4550,42 @@ export class TaskExecutor { // Stamp the active instance so `runGraphTaskStep` can honor // `deferDoneToReview` when judging a non-terminal step (FIX 3). this.graphStepActiveContext.set(this.graphActiveContextKey(seamTask.id, active.instanceId), active); - // Column-agent seam wiring (U4, R4): record the governing node id — the - // foreach INSTANCE node id (`#:`) stamped into - // context by createPromptLikeHandler — so the step-session implementation - // pass resolves the column-agent binding for the step-execute node's effective - // column (template-node column, else inherited foreach column). Set - // UNCONDITIONALLY per invocation (replacing the prior first-writer-wins guard) - // and clear it in a finally after runTaskStep, mirroring the execute seam. - // This makes two step-execute nodes in one template with DIFFERENT columns - // resolve correctly per-instance instead of all inheriting the first node's - // binding. + // Column-agent seam wiring (U4, R4): the governing node id — the foreach + // INSTANCE node id (`#:`) stamped into + // context by createPromptLikeHandler — threads INTO runGraphTaskStep, + // which stamps the per-task slot only when it CREATES the memoized + // implementation pass and clears it when that pass settles (PR #1432 + // review). One step-session pass serves every instance, so the + // session-identity binding is deterministically the pass-initiating + // instance's; per-invocation set/delete here would race under parallel + // foreach (overwrite mid-build, or clear while the shared pass is live). const stepGoverningNodeId = context[SEAM_GOVERNING_NODE_CONTEXT_KEY]; - if (typeof stepGoverningNodeId === "string") { - this.graphSeamGoverningNodeId.set(seamTask.id, stepGoverningNodeId); - } - let result: Awaited>; - try { - result = await runTaskStep( - { - store: this.store, - worktreePath, - // U6/U8: per-step session physics — graph-owned runs force - // step-session mode for the run (KTD-2/KTD-8) regardless of the - // runStepsInNewSessions setting. The agent authors the step's commit; - // this driver only observes (KTD-2). Thread the instanceId so the - // active-context read is per-instance (parallel-foreach safe). - runStep: (stepIndex) => this.runGraphTaskStep(seamTask, stepIndex, active.instanceId), - }, - { id: seamTask.id, steps: live.steps }, - active.stepIndex, - { - // Single-authority done-marking (U6/KTD-4): when the foreach template - // has a step-review node, leave the step in-progress so the review's - // APPROVE marks it done (the review is the single done authority). - markDoneOnSuccess: active.deferDoneToReview !== true, - }, - ); - } finally { - this.graphSeamGoverningNodeId.delete(seamTask.id); - } + const result: Awaited> = await runTaskStep( + { + store: this.store, + worktreePath, + // U6/U8: per-step session physics — graph-owned runs force + // step-session mode for the run (KTD-2/KTD-8) regardless of the + // runStepsInNewSessions setting. The agent authors the step's commit; + // this driver only observes (KTD-2). Thread the instanceId so the + // active-context read is per-instance (parallel-foreach safe). + runStep: (stepIndex) => + this.runGraphTaskStep( + seamTask, + stepIndex, + active.instanceId, + typeof stepGoverningNodeId === "string" ? stepGoverningNodeId : undefined, + ), + }, + { id: seamTask.id, steps: live.steps }, + active.stepIndex, + { + // Single-authority done-marking (U6/KTD-4): when the foreach template + // has a step-review node, leave the step in-progress so the review's + // APPROVE marks it done (the review is the single done authority). + markDoneOnSuccess: active.deferDoneToReview !== true, + }, + ); // Capture baseline/checkpoint back into the reserved active context so the // foreach sub-walk threads them to later template nodes (step-review/reset). active.baselineSha = result.baselineSha; diff --git a/packages/i18n/locales/en/app.json b/packages/i18n/locales/en/app.json index 0960526f43..62f1fcde88 100644 --- a/packages/i18n/locales/en/app.json +++ b/packages/i18n/locales/en/app.json @@ -6700,8 +6700,24 @@ }, "workflowColumns": { "add": "Add column", + "agent": "Column agent", + "agentBadgeDefer": "Column agent (defer)", + "agentBadgeOverride": "Column agent (override)", + "agentFlagHint": "Enable both experimentalFeatures.workflowColumns and experimentalFeatures.workflowGraphExecutor to staff columns with agents", + "agentLabel": "Column agent", + "agentMode": "Agent mode", + "agentModeDefer": "Defer", + "agentModeDeferHint": "Column agent applies only when the work carries no agent/model settings of its own", + "agentModeOverride": "Override", + "agentModeOverrideHint": "Column agent supersedes node- and task-level agent/model settings", + "agentNone": "(none)", + "agentNotFound": "Agent not found — {{id}}", + "agentsLoadFailed": "Failed to load agents", "compositionBlocked": "Resolve trait conflicts on highlighted columns before saving", + "confirmPolicyEscalation": "Bind it anyway? The column agent will run with broader permissions than this project's default.", "empty": "No columns yet. Add a column to place nodes into board lanes.", + "escalationDeclined": "Save cancelled — column agent binding not confirmed", + "overriddenByColumnAgent": "Overridden by column agent {{name}} — this node's executor settings are superseded.", "moveDown": "Move column down", "moveUp": "Move column up", "nameLabel": "Column name",