From d05d5f575fb21bcc59d474d8bc0005c1d63ba659 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Fri, 5 Jun 2026 00:57:49 -0700 Subject: [PATCH] fix(review): apply autofix feedback --- packages/core/src/agent-permission-policy.ts | 59 +++++ .../src/column-agent-binding-validation.ts | 104 ++++++++ packages/core/src/index.ts | 5 + .../__tests__/WorkflowNodeEditor.test.tsx | 24 ++ .../src/__tests__/workflow-routes.test.ts | 46 ++++ .../src/routes/register-workflow-routes.ts | 100 ++------ .../engine/src/__tests__/agent-tools.test.ts | 56 +++++ .../executor-column-agent-custom-node.test.ts | 28 +++ .../executor-column-agent-principal.test.ts | 112 +++++++++ .../executor-column-agent-seams.test.ts | 24 ++ packages/engine/src/agent-tools.ts | 84 ++++++- packages/engine/src/executor.ts | 237 +++++++++++------- 12 files changed, 709 insertions(+), 170 deletions(-) create mode 100644 packages/core/src/column-agent-binding-validation.ts diff --git a/packages/core/src/agent-permission-policy.ts b/packages/core/src/agent-permission-policy.ts index e7d4484960..3f0d013e0a 100644 --- a/packages/core/src/agent-permission-policy.ts +++ b/packages/core/src/agent-permission-policy.ts @@ -156,3 +156,62 @@ export function resolveEffectiveAgentPermissionPolicy( rules: policy.rules, }); } + +/** + * Disposition strictness rank for column-agent policy-escalation comparison + * (R13). A LOWER rank is *broader* (more privileged): `allow` lets an action + * through unconditionally, `require-approval` gates it, `block` denies it. An + * agent whose policy is broader than the project default on ANY action category + * is an escalation that must be explicitly confirmed at save time. + */ +const DISPOSITION_BREADTH_RANK: Record = { + allow: 0, + "require-approval": 1, + block: 2, +}; + +/** + * The broadest (most-privileged) rank — used as the fallback when a category is + * absent from a policy's rules map. Treating a missing category as the broadest + * possible disposition (`allow`) ensures an absent key can never silently + * *suppress* a genuine escalation: the comparison only flags when the agent is + * at least as broad as the default, so an unknown agent-side category errs + * toward flagging, and an unknown default-side category errs toward the most + * permissive default (the conservative direction for escalation detection). + */ +const BROADEST_RANK = DISPOSITION_BREADTH_RANK.allow; + +function dispositionRank( + rules: AgentPermissionPolicyRules, + category: (typeof AGENT_PERMISSION_POLICY_ACTION_CATEGORIES)[number], +): number { + const disposition = rules[category]; + if (disposition === undefined) { + // An absent category must not suppress escalation. Treat the agent side as + // broadest (most privileged) so a missing key never narrows the comparison. + return BROADEST_RANK; + } + return DISPOSITION_BREADTH_RANK[disposition]; +} + +/** + * True when `agentPolicy`'s effective policy is broader (more privileged) than + * the project `defaultPolicy` on at least one action category (R13). + * + * Both arguments should already be resolved via + * {@link resolveEffectiveAgentPermissionPolicy}, which fills every category. The + * defensive per-category handling here guards against a partial/custom rules + * map slipping through with a missing category key — an absent key must never + * silently suppress a genuine escalation. + */ +export function isPolicyBroaderThanDefault( + agentPolicy: AgentPermissionPolicy, + defaultPolicy: AgentPermissionPolicy, +): boolean { + for (const category of AGENT_PERMISSION_POLICY_ACTION_CATEGORIES) { + const agentRank = dispositionRank(agentPolicy.rules, category); + const defaultRank = dispositionRank(defaultPolicy.rules, category); + if (agentRank < defaultRank) return true; + } + return false; +} diff --git a/packages/core/src/column-agent-binding-validation.ts b/packages/core/src/column-agent-binding-validation.ts new file mode 100644 index 0000000000..1dd0c91544 --- /dev/null +++ b/packages/core/src/column-agent-binding-validation.ts @@ -0,0 +1,104 @@ +import type { AgentStore } from "./agent-store.js"; +import type { Settings } from "./types.js"; +import type { WorkflowIr, WorkflowIrColumn } from "./workflow-ir-types.js"; +import { + isPolicyBroaderThanDefault, + resolveEffectiveAgentPermissionPolicy, +} from "./agent-permission-policy.js"; + +/** + * Typed error raised when a workflow IR binds a column to an agent that fails a + * write-time check (existence or policy escalation, R11/R13). Carries the + * offending column id and a `reason` discriminant so each write surface can map + * it to its own transport (the dashboard route → an HTTP 400; the agent tools → + * a structured tool error) without re-deriving the message. + * + * Shared between the dashboard workflow route and the `fn_workflow_create` / + * `fn_workflow_update` agent tools so both write paths enforce the SAME gate — + * an agent must not be able to persist a binding the UI would reject. + */ +export class ColumnAgentBindingError extends Error { + readonly columnId: string; + readonly agentId: string; + readonly reason: "unknown-agent" | "policy-escalation"; + + constructor(args: { + message: string; + columnId: string; + agentId: string; + reason: "unknown-agent" | "policy-escalation"; + }) { + super(args.message); + this.name = "ColumnAgentBindingError"; + this.columnId = args.columnId; + this.agentId = args.agentId; + this.reason = args.reason; + } +} + +/** + * Write-time column-agent validation (U6, R11/R13), shared by every write + * surface. Inspects an IR's columns BEFORE it is persisted and throws a typed + * {@link ColumnAgentBindingError} naming the offending column. Never mutates the + * IR and never touches the store/scheduler. + * + * Two checks per bound column: + * 1. Existence — every `column.agent.agentId` must resolve in the agent + * registry; an unknown id throws (`reason: "unknown-agent"`) so the binding + * can't be saved and silently fall back at execution time. + * 2. Policy escalation (R13) — if the bound agent's effective permission policy + * is broader (more privileged) than the project default on any action + * category, the write requires an explicit `confirmPolicyEscalation` flag, + * else it throws (`reason: "policy-escalation"`). Override must never + * silently re-key action gates to a more-privileged agent. + * + * Config is data: bindings are accepted regardless of feature flags — flags gate + * execution, not storage. A null/non-object IR or columns array is left to the + * store's own validator (this only inspects shapes it can read). + */ +export async function validateColumnAgentBindings(args: { + ir: WorkflowIr | unknown; + agentStore: AgentStore; + settings: Pick; + confirmPolicyEscalation: boolean; +}): Promise { + const { ir, agentStore, settings, confirmPolicyEscalation } = args; + const columns = (ir as { columns?: unknown })?.columns; + if (!Array.isArray(columns)) return; + const bound = (columns as WorkflowIrColumn[]).filter( + (col) => col && typeof col === "object" && col.agent && typeof col.agent.agentId === "string", + ); + if (bound.length === 0) return; + + const defaultPolicy = resolveEffectiveAgentPermissionPolicy( + undefined, + settings.defaultAgentPermissionPolicy, + ); + + for (const col of bound) { + const agentId = col.agent!.agentId; + const agent = await agentStore.getAgent(agentId); + if (!agent) { + throw new ColumnAgentBindingError({ + message: `Column '${col.id}' binds unknown agent '${agentId}'`, + columnId: col.id, + agentId, + reason: "unknown-agent", + }); + } + const agentPolicy = resolveEffectiveAgentPermissionPolicy( + agent.permissionPolicy, + settings.defaultAgentPermissionPolicy, + ); + if (isPolicyBroaderThanDefault(agentPolicy, defaultPolicy) && !confirmPolicyEscalation) { + throw new ColumnAgentBindingError({ + message: + `Column '${col.id}' binds agent '${agentId}' whose permission policy is broader than ` + + `the project default; set confirmPolicyEscalation: true to confirm`, + columnId: col.id, + agentId, + reason: "policy-escalation", + }); + } + } +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index c14ef74fde..2745a3c8e9 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -304,8 +304,13 @@ export { normalizeAgentPermissionPolicy, resolveEffectiveAgentPermissionPolicy, isAgentPermissionPolicyPresetId, + isPolicyBroaderThanDefault, } from "./agent-permission-policy.js"; export type { BuiltInAgentPermissionPolicyPreset } from "./agent-permission-policy.js"; +export { + validateColumnAgentBindings, + ColumnAgentBindingError, +} from "./column-agent-binding-validation.js"; export { AgentStore, DEFAULT_AGENT_HEARTBEAT_INTERVAL_MS } from "./agent-store.js"; export type { AgentStoreEvents } from "./agent-store.js"; export { diff --git a/packages/dashboard/app/components/__tests__/WorkflowNodeEditor.test.tsx b/packages/dashboard/app/components/__tests__/WorkflowNodeEditor.test.tsx index a66770e2e6..b50247af1e 100644 --- a/packages/dashboard/app/components/__tests__/WorkflowNodeEditor.test.tsx +++ b/packages/dashboard/app/components/__tests__/WorkflowNodeEditor.test.tsx @@ -739,6 +739,30 @@ describe("WorkflowNodeEditor — U6 column agents", () => { expect(triage?.agent).toEqual({ agentId: "agent-001", mode: "defer" }); }); + it("toggling the mode to override saves the binding with mode: override", async () => { + // Start from a deferred binding so the mode toggle is already visible. + vi.mocked(fetchWorkflows).mockResolvedValue([boundDef("defer")]); + vi.mocked(updateWorkflow).mockImplementation(async (_id, updates) => ({ ...boundDef("defer"), ...(updates as object) })); + render( {}} addToast={() => {}} />); + const picker = (await screen.findByTestId("wf-column-agent-select-triage")) as HTMLSelectElement; + await waitFor(() => expect(picker.value).toBe("agent-001")); + + // Defer is the initial mode; flip to Override. + const deferRadio = (await screen.findByText("Defer")).closest("label")!.querySelector("input")! as HTMLInputElement; + expect(deferRadio.checked).toBe(true); + const overrideRadio = screen.getByText("Override").closest("label")!.querySelector("input")! as HTMLInputElement; + fireEvent.click(overrideRadio); + await waitFor(() => expect(overrideRadio.checked).toBe(true)); + + // Save round-trips the updated mode into the IR. + fireEvent.click(screen.getByText("Save").closest("button")!); + await waitFor(() => expect(updateWorkflow).toHaveBeenCalled()); + const [, updates] = vi.mocked(updateWorkflow).mock.calls[0]; + const cols = (updates as { ir: { columns: { id: string; agent?: { agentId: string; mode: string } }[] } }).ir.columns; + const triage = cols.find((c) => c.id === "triage"); + expect(triage?.agent).toEqual({ agentId: "agent-001", mode: "override" }); + }); + it("clearing to (none) removes the agent key entirely (no agent: null)", async () => { vi.mocked(fetchWorkflows).mockResolvedValue([boundDef("defer")]); vi.mocked(updateWorkflow).mockImplementation(async (_id, updates) => ({ ...boundDef("defer"), ...(updates as object) })); diff --git a/packages/dashboard/src/__tests__/workflow-routes.test.ts b/packages/dashboard/src/__tests__/workflow-routes.test.ts index e370800a3e..f96324be06 100644 --- a/packages/dashboard/src/__tests__/workflow-routes.test.ts +++ b/packages/dashboard/src/__tests__/workflow-routes.test.ts @@ -564,6 +564,52 @@ describe("workflow routes — column agents (U6)", () => { expect(ok.status).toBe(201); }); + it("saves without the flag when the agent policy equals the project default (no escalation)", async () => { + // Project default and the bound agent are both fully restrictive (locked-down): + // equal policies are NOT broader, so no confirmation is required. + await store.updateSettings({ + defaultAgentPermissionPolicy: { rules: { file_write_delete: "block", command_execution: "block" } } as never, + }); + const agentId = await makeAgent({ + permissionPolicy: { presetId: "custom", rules: { file_write_delete: "block", command_execution: "block" } }, + }); + const res = await post("/api/workflows", { name: "Equal", ir: boundIr({ agentId, mode: "override" }) }); + expect(res.status).toBe(201); + }); + + it("saves without the flag when the project default is unset (unrestricted) and the agent is unrestricted", async () => { + // No project default configured → effective default is `unrestricted` (allow-all). + // An unrestricted agent is equal, not broader, so no escalation. + const agentId = await makeAgent({ permissionPolicy: { presetId: "unrestricted" } }); + const res = await post("/api/workflows", { name: "Unrestricted", ir: boundIr({ agentId, mode: "override" }) }); + expect(res.status).toBe(201); + }); + + it("still detects escalation when the agent's custom rules map omits a category the default blocks", async () => { + // Default blocks two categories. The agent's custom rules map names only ONE + // of them (the other is absent → resolves to the unrestricted `allow` seed), + // so the agent is genuinely broader on the omitted category. A missing key + // must NOT silently suppress this escalation. + await store.updateSettings({ + defaultAgentPermissionPolicy: { rules: { file_write_delete: "block", command_execution: "block" } } as never, + }); + const agentId = await makeAgent({ + // Only file_write_delete declared; command_execution omitted → allow (broader). + permissionPolicy: { presetId: "custom", rules: { file_write_delete: "block" } }, + }); + const denied = await post("/api/workflows", { name: "PartialEsc", 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 post("/api/workflows", { + name: "PartialEsc2", + ir: boundIr({ agentId, mode: "override" }), + confirmPolicyEscalation: true, + }); + expect(ok.status).toBe(201); + }); + it("stores no agent key when the binding is absent (omission, R9)", async () => { const res = await post("/api/workflows", { name: "Plain", ir: boundIr() }); expect(res.status).toBe(201); diff --git a/packages/dashboard/src/routes/register-workflow-routes.ts b/packages/dashboard/src/routes/register-workflow-routes.ts index fdab82f4f5..e8f7b8b953 100644 --- a/packages/dashboard/src/routes/register-workflow-routes.ts +++ b/packages/dashboard/src/routes/register-workflow-routes.ts @@ -1,10 +1,7 @@ import type { WorkflowIr, WorkflowIrNode, - WorkflowIrColumn, TaskStore, - AgentPermissionPolicy, - AgentPermissionPolicyDisposition, } from "@fusion/core"; import { ColumnTraitValidationError, @@ -12,47 +9,18 @@ import { InvalidRehomeTargetError, WorkflowCompileError, WorkflowIrError, + ColumnAgentBindingError, compileWorkflowToSteps, listTraits, listStepParsers, AgentStore, - AGENT_PERMISSION_POLICY_ACTION_CATEGORIES, - resolveEffectiveAgentPermissionPolicy, + validateColumnAgentBindings, } from "@fusion/core"; import { validateCodeNodeSources } from "@fusion/engine"; import { ApiError, badRequest, conflict, notFound } from "../api-error.js"; import { emitWorkflowSseEvent } from "../sse.js"; import type { ApiRoutesContext } from "./types.js"; -/** - * Disposition strictness rank for column-agent policy-escalation comparison - * (R13). A LOWER rank is *broader* (more privileged): `allow` lets an action - * through unconditionally, `require-approval` gates it, `block` denies it. An - * agent whose policy is broader than the project default on ANY action category - * is an escalation that must be explicitly confirmed at save time. Kept local - * and minimal per the plan — no preset ordering helper exists in - * agent-permission-policy.ts, so we compare resolved per-category dispositions. - */ -const DISPOSITION_BREADTH_RANK: Record = { - allow: 0, - "require-approval": 1, - block: 2, -}; - -/** True when `agent`'s effective policy is broader (more privileged) than the - * project `defaultPolicy` on at least one action category. */ -function isPolicyBroaderThanDefault( - agentPolicy: AgentPermissionPolicy, - defaultPolicy: AgentPermissionPolicy, -): boolean { - for (const category of AGENT_PERMISSION_POLICY_ACTION_CATEGORIES) { - const agentRank = DISPOSITION_BREADTH_RANK[agentPolicy.rules[category]]; - const defaultRank = DISPOSITION_BREADTH_RANK[defaultPolicy.rules[category]]; - if (agentRank < defaultRank) return true; - } - return false; -} - /** * Routes for named workflow definitions, IR compilation preview, per-task * workflow selection, and the project default workflow. All state changes flow @@ -90,64 +58,36 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void { } /** - * Write-time column-agent validation (U6, R11/R13). Mirrors the - * `assertCodeNodesCompile` shape: inspects the IR's columns BEFORE persisting, - * throws a typed 400 naming the offending column, and never mutates the IR. - * - * Two checks per bound column: - * 1. Existence — every `column.agent.agentId` must resolve in the agent - * registry; an unknown id is a 400 naming the column (so the binding can't - * be saved and silently fall back at execution time). - * 2. Policy escalation (R13) — if the bound agent's effective permission - * policy is broader (more privileged) than the project default on any - * action category, the save requires an explicit `confirmPolicyEscalation` - * flag in the request body, else a 400 naming the policy gap. Override must - * never silently re-key action gates to a more-privileged agent. - * - * Config is data: bindings are accepted regardless of feature flags — flags - * gate execution, not storage. A null/non-object IR or columns array is left - * to the store's own validator (this only inspects shapes it can read). + * Write-time column-agent validation (U6, R11/R13). Delegates to the shared + * `validateColumnAgentBindings` helper in @fusion/core (the SAME gate the + * `fn_workflow_*` agent tools run), then maps its typed + * {@link ColumnAgentBindingError} onto an HTTP 400 carrying the structured + * fields the client UI consumes. Inspects columns BEFORE persisting and never + * mutates the IR. */ async function assertColumnAgentsExist( ir: unknown, store: TaskStore, confirmPolicyEscalation: boolean, ): Promise { + // Skip store/agent-registry I/O entirely when no column carries a binding. const columns = (ir as { columns?: unknown })?.columns; - if (!Array.isArray(columns)) return; - const bound = (columns as WorkflowIrColumn[]).filter( - (col) => col && typeof col === "object" && col.agent && typeof col.agent.agentId === "string", - ); - if (bound.length === 0) return; + if (!Array.isArray(columns) || !columns.some((c) => c?.agent?.agentId)) return; const agentStore = new AgentStore({ rootDir: store.getFusionDir() }); await agentStore.init(); - const settings = await store.getSettings(); - const defaultPolicy = resolveEffectiveAgentPermissionPolicy( - undefined, - settings.defaultAgentPermissionPolicy, - ); - - for (const col of bound) { - const agentId = col.agent!.agentId; - const agent = await agentStore.getAgent(agentId); - if (!agent) { - throw badRequest( - `Column '${col.id}' binds unknown agent '${agentId}'`, - { columnId: col.id, agentId }, - ); - } - const agentPolicy = resolveEffectiveAgentPermissionPolicy( - agent.permissionPolicy, - settings.defaultAgentPermissionPolicy, - ); - if (isPolicyBroaderThanDefault(agentPolicy, defaultPolicy) && !confirmPolicyEscalation) { - throw badRequest( - `Column '${col.id}' binds agent '${agentId}' whose permission policy is broader than the project default; set confirmPolicyEscalation: true to confirm`, - { columnId: col.id, agentId, policyEscalation: true }, - ); + try { + await validateColumnAgentBindings({ ir, agentStore, settings, confirmPolicyEscalation }); + } catch (err: unknown) { + if (err instanceof ColumnAgentBindingError) { + throw badRequest(err.message, { + columnId: err.columnId, + agentId: err.agentId, + ...(err.reason === "policy-escalation" ? { policyEscalation: true } : {}), + }); } + throw err; } } diff --git a/packages/engine/src/__tests__/agent-tools.test.ts b/packages/engine/src/__tests__/agent-tools.test.ts index a8155542f5..c86c113c9e 100644 --- a/packages/engine/src/__tests__/agent-tools.test.ts +++ b/packages/engine/src/__tests__/agent-tools.test.ts @@ -567,6 +567,62 @@ describe("createWorkflowCreateTool", () => { const text = result.content[0]?.type === "text" ? result.content[0].text : ""; expect(text).toMatch(/name is required/); }); + + // R13: the column-agent policy-escalation gate (shared with the dashboard + // route) must also fire on the agent-tool write path. A binding to an agent + // whose policy is broader than the project default is rejected unless the + // tool is called with confirm_policy_escalation: true. + it("rejects a binding to a more-privileged agent without confirm_policy_escalation", async () => { + 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); + + 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" }); + + // 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/); + + store.close(); + await rm(rootDir, { recursive: true, force: true }); + await rm(globalDir, { recursive: true, force: true }); + }); }); describe("createWorkflowUpdateTool", () => { diff --git a/packages/engine/src/__tests__/executor-column-agent-custom-node.test.ts b/packages/engine/src/__tests__/executor-column-agent-custom-node.test.ts index 38ab4a42fc..b41f4d7475 100644 --- a/packages/engine/src/__tests__/executor-column-agent-custom-node.test.ts +++ b/packages/engine/src/__tests__/executor-column-agent-custom-node.test.ts @@ -183,6 +183,34 @@ describe("runGraphCustomNode column-agent resolution (plan U3)", () => { ).toBe(true); }); + it("column agent lookup THROWS (store/agentStore error) → node still succeeds, 'lookup failed' logged (R8)", async () => { + // adoptColumnAgentForNode is best-effort: an agentStore.getAgent rejection must + // be swallowed and the node must fall back to node/default resolution rather + // than the graph node failing. + const store = createMockStore(); + store.getTask.mockResolvedValue({ id: "FN-001", worktree: "/tmp/wt", log: [] } as any); + const agentStore = { + getAgent: vi.fn().mockRejectedValue(new Error("agent store unavailable")), + }; + const executor = new TaskExecutor(store as any, "/tmp/test", { agentStore } as any); + const captured = spyStep(executor); + + const node = { id: "review", kind: "prompt", column: "review", config: { prompt: "Plain." } }; + const result = await (executor as any).runGraphCustomNode(node, { id: "FN-001" }, {}, OVERRIDE); + + // Node did NOT fail despite the lookup throwing. + expect(result.outcome).toBe("success"); + // No column-agent model adopted (lookup failed) → node falls back. + expect(captured.step.modelProvider).toBeUndefined(); + expect(captured.step.modelId).toBeUndefined(); + // The catch-path fallback audit fired. + expect( + loggedLines(store).some( + (l) => l.includes("column agent 'agent-col' lookup failed") && l.includes("falling back"), + ), + ).toBe(true); + }); + it("node with no declared column → untouched resolution even when a binding is passed as undefined", async () => { const store = createMockStore(); store.getTask.mockResolvedValue({ id: "FN-001", worktree: "/tmp/wt", log: [] } as any); 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 e93686e716..ee7f7f464a 100644 --- a/packages/engine/src/__tests__/executor-column-agent-principal.test.ts +++ b/packages/engine/src/__tests__/executor-column-agent-principal.test.ts @@ -277,10 +277,59 @@ describe("column-agent principal alignment (plan U5)", () => { "agent-X": makeColumnAgent(), }); const executeSpy = vi.spyOn(executor, "execute").mockResolvedValue(undefined as any); + // #12 distinguishability: spy on the pass-2 matcher to prove pass-2 was + // actually REACHED (not silently skipped) and returned false because the + // task's own complete model pair suppresses the defer column agent — rather + // than a false-pass where pass-2 never ran. + const matchSpy = vi.spyOn(executor as any, "taskEffectiveAgentMatches"); await executor.resumeTaskForAgent("agent-X"); + expect(matchSpy).toHaveBeenCalledTimes(1); + expect(matchSpy.mock.calls[0][1]).toBe("agent-X"); + await expect(matchSpy.mock.results[0].value).resolves.toBe(false); expect(executeSpy).not.toHaveBeenCalled(); }); + + it("step-execute template node binding governs → pass 2 matches a foreach-template-bound column agent (walks template subgraphs)", async () => { + // R6: step-execute seam nodes live ONLY inside a foreach template, never in + // ir.nodes. Pass 2 must walk foreach template subgraphs to find them; before + // the template-walk fix this returned false and the task was never re-dispatched. + const task = singleSessionTask({ id: "FN-STEP", assignedAgentId: "agent-Y" }); + const ir = { + version: "v2", + name: "test-wf", + columns: [ + { id: "step-col", name: "Step Col", traits: [], agent: OVERRIDE_COL }, + { id: "todo", name: "Todo", traits: [] }, + ], + nodes: [ + { + id: "foreach-1", + kind: "foreach", + column: "todo", + config: { + template: { + nodes: [ + { id: "step-exec", kind: "prompt", column: "step-col", config: { seam: "step-execute" } }, + ], + }, + }, + }, + ], + edges: [], + } as unknown as WorkflowIr; + const store = resumeStore(task, ir); + const { executor } = makeExecutor(store, { + "agent-Y": makeAssignedAgent(), + "agent-X": makeColumnAgent(), + }); + const executeSpy = vi.spyOn(executor, "execute").mockResolvedValue(undefined as any); + + await executor.resumeTaskForAgent("agent-X"); + + expect(executeSpy).toHaveBeenCalledTimes(1); + expect(executeSpy.mock.calls[0][0]).toMatchObject({ id: "FN-STEP" }); + }); }); // ── (b) Reverse direction: isAgentEffectivelyExecuting (R6) ─────────────── @@ -387,6 +436,69 @@ describe("column-agent principal alignment (plan U5)", () => { expect((executor as any).activeSessions.get(task.id).lastEffectiveColumnAgentId).toBeNull(); }); + it("no-op tick: same effective column agent + already-resolved model → setModel NOT called", async () => { + // The active session is already running as X on X's advertised model. A + // task:updated tick that changes nothing about the effective agent/model must + // not re-issue a setModel (no churn / no spurious hot-swap). + const store = createMockStore(); + const find = vi.fn().mockReturnValue({ provider: "anthropic", modelId: "claude-x" }); + const task = singleSessionTask({ assignedAgentId: "agent-Y" }); + // Column agent X advertises EXACTLY the model the session already resolved. + const { executor } = makeExecutor(store, { + "agent-X": makeColumnAgent({ runtimeConfig: { model: "anthropic/claude-x", allowParallelExecution: false } }), + }); + (executor as any)._modelRegistry = { find }; + + // activeGraphSession seeds lastResolvedModelProvider/Id = anthropic/claude-x + // 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)); + + // No agent change, no model change → no hot-swap. + expect(setModel).not.toHaveBeenCalled(); + expect(loggedLines(store).some((l) => l.includes("Column agent changed"))).toBe(false); + // The legacy task-model block must also not fire a model swap for the override session. + expect(loggedLines(store).some((l) => l.startsWith("Model changed"))).toBe(false); + }); + + it("override session + mid-flight task model/assigned-agent edit → column agent's model is preserved (legacy hot-swap does NOT clobber it)", async () => { + // R3: under an OVERRIDE column, the column agent owns the model. A user editing + // the task's modelProvider/modelId or assignedAgentId mid-flight must NOT cause + // the legacy task-model hot-swap to resolve the assigned/own model and clobber + // the column agent's model. + const store = createMockStore(); + const find = vi.fn().mockReturnValue({ provider: "openai", modelId: "gpt-edited" }); + // Edited task: now carries a complete own model pair AND a different assigned agent. + const task = singleSessionTask({ + assignedAgentId: "agent-Z", + modelProvider: "openai", + modelId: "gpt-edited", + }); + // Column agent X advertises its own (unchanged) model. + const { executor } = makeExecutor(store, { + "agent-X": makeColumnAgent({ runtimeConfig: { model: "anthropic/claude-x", allowParallelExecution: false } }), + "agent-Z": makeAssignedAgent({ id: "agent-Z", runtimeConfig: { model: "openai/gpt-edited" } }), + }); + (executor as any)._modelRegistry = { find }; + + const { setModel } = activeGraphSession(executor, task.id, "exec-node", OVERRIDE_COL); + + store._trigger("task:updated", task); + await new Promise((r) => setTimeout(r, 0)); + + // The legacy block is short-circuited under override: the assigned/own model + // (openai/gpt-edited) is NEVER applied via setModel. + expect(find).not.toHaveBeenCalledWith("openai", "gpt-edited"); + const setModelArgs = setModel.mock.calls.map((c: any[]) => c[0]); + expect(setModelArgs).not.toContainEqual({ provider: "openai", modelId: "gpt-edited" }); + // No legacy "Model changed to openai/gpt-edited" audit line either. + expect(loggedLines(store).some((l) => l.includes("openai/gpt-edited"))).toBe(false); + // The tracked effective principal stays the column agent. + expect((executor as any).activeSessions.get(task.id).lastEffectiveColumnAgentId).toBe("agent-X"); + }); + it("legacy entry (no effective column agent) → the column-invalidation block is skipped", async () => { const store = createMockStore(); const find = vi.fn(); diff --git a/packages/engine/src/__tests__/executor-column-agent-seams.test.ts b/packages/engine/src/__tests__/executor-column-agent-seams.test.ts index 08c81ae8aa..351340382a 100644 --- a/packages/engine/src/__tests__/executor-column-agent-seams.test.ts +++ b/packages/engine/src/__tests__/executor-column-agent-seams.test.ts @@ -313,6 +313,30 @@ describe("column-agent coding seams (plan U4)", () => { expect(loggedLines(store).some((l) => l.includes("running as column agent"))).toBe(false); }); + it("defer column, task with own assignedAgentId only (NO model pair) → own-settings win, column agent NOT adopted", async () => { + // KTD-5 at the seam: an own agent IDENTITY alone counts as own-settings even + // without a complete model pair, so a defer column must NOT adopt the column + // agent. (Distinct from the bare-task case below where the column agent wins.) + const store = createMockStore(); + const task = singleSessionTask({ assignedAgentId: "agent-Y" }); // no modelProvider/modelId + store.getTask.mockResolvedValue(task as any); + const { executor, agentStore } = makeExecutor(store, { + "agent-Y": makeAssignedAgent(), + "agent-col": makeColumnAgent(), + }); + installTaskDoneAgent(); + + await runExecuteSeam(executor, task, "execute-node", DEFER_COL); + + const opts = lastFnAgentOpts(); + // Fell back to the assigned agent's model — column agent's model not adopted. + expect(opts.defaultProvider).toBe("openai"); + expect(opts.defaultModelId).toBe("gpt-assigned"); + // Column agent never fetched/adopted, no adoption audit. + expect(agentStore.getAgent).not.toHaveBeenCalledWith("agent-col"); + expect(loggedLines(store).some((l) => l.includes("running as column agent"))).toBe(false); + }); + it("defer column, bare task (no own settings) → column agent adopted", async () => { const store = createMockStore(); const task = singleSessionTask(); // no assignedAgentId, no model pair diff --git a/packages/engine/src/agent-tools.ts b/packages/engine/src/agent-tools.ts index e680955baa..9ef8a78f99 100644 --- a/packages/engine/src/agent-tools.ts +++ b/packages/engine/src/agent-tools.ts @@ -11,8 +11,8 @@ import { appendFile, mkdir, readFile, readdir, stat, writeFile } from "node:fs/p import { existsSync } from "node:fs"; import { createHash } from "node:crypto"; import { join, relative, resolve } from "node:path"; -import type { AgentStore, AgentState, AgentCapability, AgentUpdateInput, TaskDocument, TaskDocumentCreateInput, TaskStore, RunMutationContext, MessageStore, Message, SourceType, Settings, ResearchRun, ResearchRunStatus, TaskCreateInput, ReflectionStore, ApprovalRequestStore, ProjectSettings, ChatStore } from "@fusion/core"; -import { listTraits, isBuiltinWorkflowId } from "@fusion/core"; +import type { AgentState, AgentCapability, AgentUpdateInput, TaskDocument, TaskDocumentCreateInput, TaskStore, RunMutationContext, MessageStore, Message, SourceType, Settings, ResearchRun, ResearchRunStatus, TaskCreateInput, ReflectionStore, ApprovalRequestStore, ProjectSettings, ChatStore } from "@fusion/core"; +import { listTraits, isBuiltinWorkflowId, AgentStore, validateColumnAgentBindings, ColumnAgentBindingError } from "@fusion/core"; import { promoteHeldTask } from "./hold-release.js"; import { DASHBOARD_USER_ID, canAgentTakeImplementationTaskForExplicitRouting, dailyMemoryPath, ensureOpenClawMemoryFiles, extractAgentProvisioningRequest, formatRoleMismatchReason, getMemoryBackendCapabilities, getProjectMemory, isEphemeralAgent, memoryLongTermPath, normalizeMessageParticipant, reconcileDeterministicDuplicate, resolveAgentProvisioningPolicy, resolveMemoryBackend, resolveResearchSettings, resolveTaskGithubTracking, resolveTitleSummarizerSettingsModel, runDeterministicDuplicateGuard, scheduleQmdProjectMemoryRefresh, searchProjectMemory, shouldSkipBackgroundQmdRefresh, summarizeTitle } from "@fusion/core"; import { ResearchOrchestrator } from "./research-orchestrator.js"; @@ -102,6 +102,14 @@ export const workflowCreateParams = Type.Object({ description: "Optional node layout map keyed by node id.", }), ), + confirm_policy_escalation: Type.Optional( + Type.Boolean({ + description: + "Set true to confirm binding a column to an agent whose permission policy is broader " + + "(more privileged) than the project default. Required when such a binding is present; " + + "the create is otherwise rejected naming the offending column.", + }), + ), }); export const workflowUpdateParams = Type.Object({ @@ -117,6 +125,14 @@ export const workflowUpdateParams = Type.Object({ "Required to resolve an OccupiedColumns conflict; the target must exist in the new IR.", }), ), + confirm_policy_escalation: Type.Optional( + Type.Boolean({ + description: + "Set true to confirm binding a column to an agent whose permission policy is broader " + + "(more privileged) than the project default. Required when such a binding is present; " + + "the update is otherwise rejected naming the offending column.", + }), + ), }); export const workflowDeleteParams = Type.Object({ @@ -1197,6 +1213,48 @@ export function createTaskPromoteTool(store: TaskStore, currentTaskId: string): }; } +/** + * Shared write-time column-agent gate for the `fn_workflow_*` tools (R11/R13). + * Runs the SAME `validateColumnAgentBindings` check the dashboard route runs, so + * an agent cannot persist a binding the UI would reject (existence + + * policy-escalation). Constructs a per-call AgentStore from the store's fusion + * dir (the connection is process-cached) and feeds it the project settings. + * + * A {@link ColumnAgentBindingError} propagates unchanged; each tool's catch + * surfaces its message (which names the column and, for an escalation, instructs + * passing `confirm_policy_escalation: true`). + */ +async function assertWorkflowColumnAgentBindings( + store: TaskStore, + ir: unknown, + confirmPolicyEscalation: boolean, +): Promise { + const columns = (ir as { columns?: unknown })?.columns; + if (!Array.isArray(columns) || !columns.some((c) => c?.agent?.agentId)) return; + const agentStore = new AgentStore({ rootDir: store.getFusionDir() }); + await agentStore.init(); + const settings = await store.getSettings(); + await validateColumnAgentBindings({ ir, agentStore, settings, confirmPolicyEscalation }); +} + +/** + * Render a {@link ColumnAgentBindingError} as a structured tool error result. + * Re-phrases the escalation guidance with the tool's snake_case flag name + * (`confirm_policy_escalation`) rather than the route's camelCase variant. + */ +function columnAgentBindingErrorResult(err: ColumnAgentBindingError) { + const text = + err.reason === "policy-escalation" + ? `Column '${err.columnId}' binds agent '${err.agentId}' whose permission policy is broader than ` + + `the project default; pass confirm_policy_escalation: true to confirm.` + : err.message; + return { + content: [{ type: "text" as const, text: `ERROR: ${text}` }], + details: { columnId: err.columnId, agentId: err.agentId, reason: err.reason }, + isError: true as const, + }; +} + /** * Create a `fn_workflow_create` tool — a thin wrapper over the store's workflow * definition create. The IR is validated server-side; a malformed graph rejects. @@ -1222,10 +1280,15 @@ export function createWorkflowCreateTool(store: TaskStore): ToolDefinition { "`code` node {source, timeoutMs?} runs sandboxed TypeScript returning {outcome?, contextPatch?, customFields?}. " + "Declare task documents via `artifacts: [{key, title?, producedBy?, role?}]` and custom task fields via " + "`fields: [{id, name, type, required?, default?, options?, render?}]` (types: string/text/number/boolean/" + - "enum/multi-enum/date/url; render.placement card|detail|detail-section, render.badge for card chips).", + "enum/multi-enum/date/url; render.placement card|detail|detail-section, render.badge for card chips).\n" + + "Bind a column to a permanent agent via `columns[].agent: { agentId, mode }`: `mode:'defer'` applies the " + + "column agent only when the work carries no own agent/model settings, while `mode:'override'` supersedes " + + "node/task settings wholesale. The bound agent must exist; if its permission policy is broader than the " + + "project default, pass `confirm_policy_escalation: true` to confirm (the create is otherwise rejected).", parameters: workflowCreateParams, execute: async (_id: string, params: Static) => { try { + await assertWorkflowColumnAgentBindings(store, params.ir, params.confirm_policy_escalation === true); const created = await store.createWorkflowDefinition({ name: params.name, description: params.description, @@ -1240,6 +1303,9 @@ export function createWorkflowCreateTool(store: TaskStore): ToolDefinition { }; // eslint-disable-next-line @typescript-eslint/no-explicit-any } catch (err: any) { + if (err instanceof ColumnAgentBindingError) { + return columnAgentBindingErrorResult(err); + } return { content: [{ type: "text" as const, text: `ERROR: Failed to create workflow: ${err?.message ?? err}` }], details: {}, @@ -1266,10 +1332,17 @@ export function createWorkflowUpdateTool(store: TaskStore): ToolDefinition { "occupied columns — retry with rehome_to set to a column id that survives in the new IR. " + "The IR accepts the same step-inversion constructs as fn_workflow_create (foreach with mode/isolation/" + "concurrency, step-execute, step-review, parse-steps, code nodes, rework edges, artifacts, fields). " + - "Editing `fields` orphans (never destroys) existing task values for removed/incompatible fields.", + "Editing `fields` orphans (never destroys) existing task values for removed/incompatible fields.\n" + + "Bind a column to a permanent agent via `columns[].agent: { agentId, mode }`: `mode:'defer'` applies the " + + "column agent only when the work carries no own agent/model settings, while `mode:'override'` supersedes " + + "node/task settings wholesale. The bound agent must exist; if its permission policy is broader than the " + + "project default, pass `confirm_policy_escalation: true` to confirm (the update is otherwise rejected).", parameters: workflowUpdateParams, execute: async (_id: string, params: Static) => { try { + if (params.ir !== undefined) { + await assertWorkflowColumnAgentBindings(store, params.ir, params.confirm_policy_escalation === true); + } const updated = await store.updateWorkflowDefinition(params.workflow_id, { name: params.name, description: params.description, @@ -1285,6 +1358,9 @@ export function createWorkflowUpdateTool(store: TaskStore): ToolDefinition { }; // eslint-disable-next-line @typescript-eslint/no-explicit-any } catch (err: any) { + if (err instanceof ColumnAgentBindingError) { + return columnAgentBindingErrorResult(err); + } // Surface the typed OccupiedColumnsError as a structured, retryable result. if (err?.name === "OccupiedColumnsError") { const occupancies = err.occupancies ?? []; diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 7423f023e5..f6a20aa7c8 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -9,8 +9,8 @@ import { delimiter, isAbsolute, join, relative, resolve as resolvePath } from "n import { existsSync, realpathSync } from "node:fs"; import { readFile, rm, writeFile } from "node:fs/promises"; import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent, AgentMemoryInclusionMode, ProjectSettings, MergeResult, WorkflowIrNode } from "@fusion/core"; -import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled, resolveWorkflowIrForTask, resolveColumnAgentBinding, resolveEffectiveAgent } from "@fusion/core"; -import type { TaskStep, WorkflowIr, WorkflowFieldDefinition, WorkflowColumnAgent } from "@fusion/core"; +import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled, resolveWorkflowIrForTask, resolveColumnAgentBinding, resolveEffectiveAgent, instanceNodeId } from "@fusion/core"; +import type { TaskStep, WorkflowIr, WorkflowFieldDefinition, WorkflowColumnAgent, EffectiveAgentInput } from "@fusion/core"; import { buildWorkflowObservationFromTask, buildWorkflowObservation, @@ -2108,12 +2108,9 @@ export class TaskExecutor { const resolveBinding = this.graphColumnAgentResolver.get(task.id)!; const binding = resolveBinding(governingNodeId); if (binding) { - const ownModelComplete = Boolean(task.modelProvider && task.modelId); const effective = resolveEffectiveAgent({ binding, - ownAgentId: (task.assignedAgentId ?? "").trim() || undefined, - ownModelProvider: ownModelComplete ? task.modelProvider : undefined, - ownModelId: ownModelComplete ? task.modelId : undefined, + ...this.extractOwnSettings(task), }); if (effective.source === "column-agent") { // Fetch the (possibly changed) effective column agent, best-effort. @@ -2124,12 +2121,15 @@ export class TaskExecutor { // tracked id so we stop probing for the missing agent every tick. if (activeEntry.lastEffectiveColumnAgentId !== null) { executorLog.log(`${task.id}: column agent '${effective.agentId}' deleted mid-session — falling back, no restart (R8)`); - await this.store.logEntry( + // Fire-and-forget audit (matches the rework-log posture at ~3582): + // a logEntry failure must not abort this task:updated tick and skip + // the model-change detection below. + this.store.logEntry( task.id, `Column agent '${effective.agentId}' deleted mid-session — falling back to current model, no restart (R8)`, undefined, 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; } } else { @@ -2160,7 +2160,10 @@ export class TaskExecutor { } catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err); executorLog.error(`${task.id}: failed to column-agent hot-swap: ${errorMessage}`); - await this.store.logEntry(task.id, `Column-agent change failed: ${errorMessage}`, undefined, this.getRunContextFor(task.id)); + // Fire-and-forget audit (see ~3582): a logEntry failure here must + // not abort the tick and skip later model-change detection. + this.store.logEntry(task.id, `Column-agent change failed: ${errorMessage}`, undefined, this.getRunContextFor(task.id)) + .catch((logErr: unknown) => executorLog.warn(`${task.id}: failed to log column-agent change failure: ${logErr instanceof Error ? logErr.message : String(logErr)}`)); } } } @@ -2172,11 +2175,28 @@ export class TaskExecutor { // Handle executor model hot-swap on active single-session executions if (this.activeSessions.has(task.id) && !task.paused) { const activeEntry = this.activeSessions.get(task.id)!; + // R3 guard: when an OVERRIDE column agent governs this running session, the + // column-agent watcher block above OWNS the model (override supersedes the + // task's own model/assigned-agent settings). The legacy task-model hot-swap + // would otherwise resolve a model from task.assignedAgentId's runtimeConfig + // and clobber the column agent's model on a mid-flight task edit. Skip it + // entirely when override governs; defer-resolved-to-own-settings (or no + // binding) keeps the legacy behavior identical. + let overrideColumnGoverns = false; + if ((activeEntry.lastEffectiveColumnAgentId ?? null) !== null) { + const governingNodeId = this.graphSeamGoverningNodeId.get(task.id); + const resolveBinding = this.graphColumnAgentResolver.get(task.id); + if (governingNodeId && resolveBinding) { + const binding = resolveBinding(governingNodeId); + if (binding?.mode === "override") overrideColumnGoverns = true; + } + } + const taskModelProviderChanged = task.modelProvider !== activeEntry.lastTaskModelProvider; const taskModelIdChanged = task.modelId !== activeEntry.lastTaskModelId; const assignedAgentChanged = (task.assignedAgentId ?? null) !== (activeEntry.lastAssignedAgentId ?? null); - if (taskModelProviderChanged || taskModelIdChanged || assignedAgentChanged) { + if (!overrideColumnGoverns && (taskModelProviderChanged || taskModelIdChanged || assignedAgentChanged)) { activeEntry.lastTaskModelProvider = task.modelProvider; activeEntry.lastTaskModelId = task.modelId; activeEntry.lastAssignedAgentId = task.assignedAgentId ?? null; @@ -3232,27 +3252,33 @@ export class TaskExecutor { const ir = await resolveWorkflowIrForTask(this.store, task.id); if (!ir || ir.version !== "v2") return false; - const ownAgentId = typeof task.assignedAgentId === "string" && task.assignedAgentId.trim() - ? task.assignedAgentId.trim() - : undefined; - const ownModelComplete = Boolean(task.modelProvider && task.modelId); + const ownSettings = this.extractOwnSettings(task); + const matchesNodeId = (nodeId: string): boolean => { + const binding = resolveColumnAgentBinding(ir, nodeId); + if (!binding) return false; + const effective = resolveEffectiveAgent({ binding, ...ownSettings }); + return effective.source === "column-agent" && effective.agentId === agentId; + }; - // Governing seam nodes: the execute-seam prompt node and any step-execute seam - // prompt nodes (the latter resolve their column via template inheritance, which - // resolveColumnAgentBinding handles by node id). + // Governing seam nodes: the execute-seam prompt node lives at the top level. for (const node of ir.nodes) { const seam = node.kind === "prompt" ? node.config?.seam : undefined; if (seam !== "execute" && seam !== "step-execute") continue; - const binding = resolveColumnAgentBinding(ir, node.id); - if (!binding) continue; - const effective = resolveEffectiveAgent({ - binding, - ownAgentId, - ownModelProvider: ownModelComplete ? task.modelProvider : undefined, - ownModelId: ownModelComplete ? task.modelId : undefined, - }); - if (effective.source === "column-agent" && effective.agentId === agentId) { - return true; + if (matchesNodeId(node.id)) return true; + } + + // step-execute seam nodes are legal ONLY inside a foreach template + // (workflow-ir.ts), so they never appear in ir.nodes above. Walk each foreach + // node's template subgraph and resolve the binding via a synthesized instance + // node id. Step index 0 is sufficient — column resolution is index-independent + // (all instances share the same template node and thus the same binding, R4). + for (const node of ir.nodes) { + if (node.kind !== "foreach") continue; + const templateNodes = (node.config as { template?: { nodes?: WorkflowIrNode[] } } | undefined)?.template?.nodes ?? []; + for (const templateNode of templateNodes) { + const seam = templateNode.kind === "prompt" ? templateNode.config?.seam : undefined; + if (seam !== "step-execute") continue; + if (matchesNodeId(instanceNodeId(node.id, 0, templateNode.id))) return true; } } return false; @@ -4460,35 +4486,43 @@ export class TaskExecutor { 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 (once-per-run, KTD-2/KTD-8) - // step-session implementation pass resolves the column-agent binding for the - // step-execute node's effective column (template-node column, else inherited - // foreach column). All instances share the same template node and thus the - // same binding, so the first instance to drive the pass sets it correctly. + // 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. const stepGoverningNodeId = context[SEAM_GOVERNING_NODE_CONTEXT_KEY]; - if (typeof stepGoverningNodeId === "string" && !this.graphSeamGoverningNodeId.has(seamTask.id)) { + if (typeof stepGoverningNodeId === "string") { this.graphSeamGoverningNodeId.set(seamTask.id, stepGoverningNodeId); } - const 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, - }, - ); + 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); + } // 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; @@ -4750,6 +4784,26 @@ export class TaskExecutor { * `"agent"` branch historically read a non-existent `customInstructions` * field (silently undefined); this is the single consistent source used by * both the node-agent and column-agent paths. */ + /** Extract a task's OWN settings for the effective-agent resolver: its assigned + * agent identity (trimmed, non-empty) and a COMPLETE model pair (an incomplete + * pair does not count — KTD-5, mirrors resolveExecutorSessionModel's both-present + * rule). Centralizes the previously-duplicated extraction so the four call sites + * (restart watcher, taskEffectiveAgentMatches, resolveSeamColumnAgent, + * resolveEffectivePrincipalId) share one normalized idiom. */ + private extractOwnSettings( + task: Pick, + ): Pick { + const ownAgentId = typeof task.assignedAgentId === "string" && task.assignedAgentId.trim() + ? task.assignedAgentId.trim() + : undefined; + const ownModelComplete = Boolean(task.modelProvider && task.modelId); + return { + ownAgentId, + ownModelProvider: ownModelComplete ? task.modelProvider : undefined, + ownModelId: ownModelComplete ? task.modelId : undefined, + }; + } + private buildAgentPersona(agent: Agent): string | undefined { const parts = [agent.soul, agent.instructionsText] .map((p) => (typeof p === "string" ? p.trim() : "")) @@ -4793,12 +4847,18 @@ export class TaskExecutor { }; } catch { // Agent lookup is best-effort; fall back to node/default resolution (R8). - await this.store.logEntry( - live.id, - `Workflow node '${node.id}': column agent '${columnAgentId}' lookup failed — falling back to node/default resolution`, - undefined, - this.getRunContextFor(live.id), - ); + // A secondary logEntry failure (DB locked / mid-recovery) must NOT propagate + // out of this error handler and escalate the node to a hard failure. + try { + await this.store.logEntry( + live.id, + `Workflow node '${node.id}': column agent '${columnAgentId}' lookup failed — falling back to node/default resolution`, + undefined, + this.getRunContextFor(live.id), + ); + } catch (logErr: unknown) { + executorLog.warn(`${live.id}: failed to log column-agent lookup failure: ${logErr instanceof Error ? logErr.message : String(logErr)}`); + } return undefined; } } @@ -4835,15 +4895,9 @@ export class TaskExecutor { // The task's OWN settings: its assigned agent identity and a COMPLETE model // pair (an incomplete pair does not count — KTD-5, mirrors // resolveExecutorSessionModel's both-present rule). - const ownAgentId = typeof detail.assignedAgentId === "string" && detail.assignedAgentId.trim() - ? detail.assignedAgentId.trim() - : undefined; - const ownModelComplete = Boolean(detail.modelProvider && detail.modelId); const effective = resolveEffectiveAgent({ binding, - ownAgentId, - ownModelProvider: ownModelComplete ? detail.modelProvider : undefined, - ownModelId: ownModelComplete ? detail.modelId : undefined, + ...this.extractOwnSettings(detail), }); if (effective.source !== "column-agent") return undefined; @@ -4855,20 +4909,30 @@ export class TaskExecutor { agent = null; } if (!agent) { + // Best-effort audit: a logEntry failure (DB locked / mid-recovery) must NOT + // escalate this graceful fallback into a hard session failure (R8). + try { + await this.store.logEntry( + task.id, + `Workflow seam node '${governingNodeId}': column agent '${effective.agentId}' not found — falling back to assigned-agent resolution`, + undefined, + this.getRunContextFor(task.id), + ); + } catch (logErr: unknown) { + executorLog.warn(`${task.id}: failed to log column-agent fallback: ${logErr instanceof Error ? logErr.message : String(logErr)}`); + } + return undefined; + } + try { await this.store.logEntry( task.id, - `Workflow seam node '${governingNodeId}': column agent '${effective.agentId}' not found — falling back to assigned-agent resolution`, + `Workflow seam node '${governingNodeId}': running as column agent '${effective.agentId}' (${binding.mode})`, undefined, this.getRunContextFor(task.id), ); - return undefined; + } catch (logErr: unknown) { + executorLog.warn(`${task.id}: failed to log column-agent adoption: ${logErr instanceof Error ? logErr.message : String(logErr)}`); } - await this.store.logEntry( - task.id, - `Workflow seam node '${governingNodeId}': running as column agent '${effective.agentId}' (${binding.mode})`, - undefined, - this.getRunContextFor(task.id), - ); return { agent, mode: binding.mode }; } @@ -4889,9 +4953,8 @@ export class TaskExecutor { task: Task, detail: Task, ): string | undefined { - const assignedAgentId = typeof detail.assignedAgentId === "string" && detail.assignedAgentId.trim() - ? detail.assignedAgentId.trim() - : undefined; + const ownSettings = this.extractOwnSettings(detail); + const assignedAgentId = ownSettings.ownAgentId; const governingNodeId = this.graphSeamGoverningNodeId.get(task.id); const resolveBinding = this.graphColumnAgentResolver.get(task.id); @@ -4900,13 +4963,7 @@ export class TaskExecutor { const binding = resolveBinding(governingNodeId); if (!binding) return assignedAgentId; - const ownModelComplete = Boolean(detail.modelProvider && detail.modelId); - const effective = resolveEffectiveAgent({ - binding, - ownAgentId: assignedAgentId, - ownModelProvider: ownModelComplete ? detail.modelProvider : undefined, - ownModelId: ownModelComplete ? detail.modelId : undefined, - }); + const effective = resolveEffectiveAgent({ binding, ...ownSettings }); if (effective.source === "column-agent") return effective.agentId; return assignedAgentId; } @@ -6178,8 +6235,9 @@ export class TaskExecutor { // memory tools adopt the column agent. The core resolver decides defer vs // override (KTD-2); a missing agent logs + falls back (R8). No binding → // `columnAgentSeam` is undefined and every line below is byte-identical to the - // assigned-agent path (characterization parity). Gating contexts still key off - // the ASSIGNED agent — principal substitution for gating/heartbeat is U5. + // assigned-agent path (characterization parity). Gating contexts key off + // `identityAgent` — the effective column agent when a binding governs, else + // the assigned agent (U5/KTD-3 principal substitution). const columnAgentSeam = await this.resolveSeamColumnAgent(task, detail); const identityAgent = columnAgentSeam?.agent ?? assignedAgent; const executorRuntimeHint = extractRuntimeHint(identityAgent?.runtimeConfig); @@ -7704,6 +7762,13 @@ export class TaskExecutor { executingTaskLock.release(task.id); // Clear run context at end of execute() lifecycle this.currentRunContexts.delete(task.id); + // U5 (R6) leak guard: effectiveColumnAgentByTask is set() in the outer execute() + // scope (execute-seam ~6191, step-session ~5674) BEFORE the session-entry try + // whose finally (deleteActiveSession / deleteActiveStepExecutor) normally clears + // it. A throw between the set() and that try would otherwise leak the entry and + // permanently block the column agent's heartbeat ticks. Deleting here in the + // outer finally covers BOTH paths since both run inside execute(). + this.effectiveColumnAgentByTask.delete(task.id); // Terminate all spawned child agents on ALL exit paths. // This must run here (in the outer finally) rather than only in agentWork's