From 41f25d5c52a60789356f6cc215bca8973bf9bfbc Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Fri, 5 Jun 2026 00:19:13 -0700 Subject: [PATCH] feat(engine): effective column agent becomes the execution principal U5: action-gating contexts, the heartbeat deferral gate, a two-pass resumeTaskForAgent, and the reverse-direction agent.taskId guards (via a new isAgentEffectivelyExecuting callback wired at the in-process runtime) all consult the column-effective agent; the restart watcher re-resolves column bindings per tick for bound graph sessions, hot-swapping on agent-changed and falling back without restart on agent-deleted. --- .../workflow-column-agent-assignment.md | 2 + packages/cli/src/commands/dashboard.ts | 5 + .../executor-column-agent-principal.test.ts | 437 ++++++++++++++++++ packages/engine/src/agent-heartbeat.ts | 40 +- packages/engine/src/executor.ts | 330 +++++++++++-- .../engine/src/runtimes/in-process-runtime.ts | 8 +- 6 files changed, 786 insertions(+), 36 deletions(-) create mode 100644 packages/engine/src/__tests__/executor-column-agent-principal.test.ts diff --git a/.changeset/workflow-column-agent-assignment.md b/.changeset/workflow-column-agent-assignment.md index 249914fcce..b9c7cbc967 100644 --- a/.changeset/workflow-column-agent-assignment.md +++ b/.changeset/workflow-column-agent-assignment.md @@ -5,3 +5,5 @@ Add per-column agent assignment for workflow columns, behind the combined `experimentalFeatures.workflowColumns` + `experimentalFeatures.workflowGraphExecutor` flags. A workflow column can now name a permanent agent from the registry plus a mode — `defer` (the column agent is the default for work in that column that carries no agent/model settings of its own) or `override` (the column agent supersedes node- and task-level agent/model settings). The binding applies to all session-running work attributable to the column's nodes: custom prompt/gate/script nodes, the execute seam's coding session, and step-execute sessions. Precedence is resolved by one shared `@fusion/core` resolver (`resolveColumnAgentBinding` + `resolveEffectiveAgent`) consumed by every reader, with defer/override expressed as explicit named rules and defer granularity all-or-nothing (an own agent identity OR a complete `modelProvider`+`modelId` pair suppresses the column agent). The binding keys off the node's declared IR column; foreach template nodes inherit the enclosing foreach node's column. A missing/deleted agent at resolution time logs and falls back to normal resolution — a live session is never aborted. The built-in default workflow carries no column agents and stays byte-identical (parity oracle); with either flag off, column agents are inert. + +The effective column agent is also the principal for the subsystems that previously assumed the running agent is always `task.assignedAgentId`: action gating (`buildActionGateContext` / `buildPermanentAgentGatingContext`) is computed for the agent actually running; heartbeat serialization honors it in both directions (the execute deferral gate, a second `resumeTaskForAgent` pass that re-dispatches tasks whose effective column agent matches, and a reverse-direction heartbeat-scheduler guard so an `allowParallelExecution=false` column agent never heartbeats concurrently with its own session); and a workflow-definition edit or agent runtimeConfig change that re-keys the column-effective agent/model hot-swaps the running graph session, while an agent deleted mid-session falls back without a restart. diff --git a/packages/cli/src/commands/dashboard.ts b/packages/cli/src/commands/dashboard.ts index 0618905751..43cd1cd301 100644 --- a/packages/cli/src/commands/dashboard.ts +++ b/packages/cli/src/commands/dashboard.ts @@ -1974,6 +1974,11 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?: }); }, store, + // Dev-mode scheduler: no TaskExecutor runs here (engine not started), so + // neither `isTaskExecuting` nor the U5 reverse-direction + // `isAgentEffectivelyExecuting` guard has a source — both stay unwired (the + // guards simply never fire), matching the prior `isTaskExecuting` omission. + // The real wiring is the InProcessRuntime construction site. ); triggerScheduler.start(); diff --git a/packages/engine/src/__tests__/executor-column-agent-principal.test.ts b/packages/engine/src/__tests__/executor-column-agent-principal.test.ts new file mode 100644 index 0000000000..e93686e716 --- /dev/null +++ b/packages/engine/src/__tests__/executor-column-agent-principal.test.ts @@ -0,0 +1,437 @@ +// Column-agent PRINCIPAL alignment (plan U5, R5/R6/R7, KTD-3/KTD-4). +// +// The three subsystems that historically assumed "the running agent is +// task.assignedAgentId" must consult the EFFECTIVE column agent instead: +// (a) action gating (buildActionGateContext / buildPermanentAgentGatingContext) +// — gate for the agent actually running (R5); +// (b) heartbeat serialization in BOTH directions (R6): +// - the execute() deferral gate consults the effective principal; +// - resumeTaskForAgent re-dispatches column-effective tasks via a second +// pass the assignedAgentId-only filter would miss; +// - the heartbeat scheduler's reverse guard (isAgentEffectivelyExecuting) +// blocks a column agent from heartbeating concurrently with its own session; +// (c) the restart watcher hot-swaps when a workflow edit / agent-config change +// re-keys the column-effective agent/model mid-flight, and falls back (no +// restart storm) when the column agent is deleted (R7/KTD-4/R8). +// +// Harness mirrors executor-column-agent-seams.test.ts: a real TaskExecutor over a +// mock store with createFnAgent + StepSessionExecutor mocked. The per-run seam +// slots (graphSeamGoverningNodeId / graphColumnAgentResolver) are seeded directly, +// then runImplementationPhase drives the production session-build path. + +import { beforeEach, describe, expect, it, vi } from "vitest"; +import "./executor-test-helpers.js"; +import { TaskExecutor } from "../executor.js"; +import { + createMockStore, + mockedCreateFnAgent, + resetExecutorMocks, +} from "./executor-test-helpers.js"; +import type { WorkflowColumnAgent, WorkflowIr } from "@fusion/core"; + +const OVERRIDE_COL: WorkflowColumnAgent = { agentId: "agent-X", mode: "override" }; +const DEFER_COL: WorkflowColumnAgent = { agentId: "agent-X", mode: "defer" }; + +// agent-X = the column agent (allowParallelExecution=false unless overridden). +// agent-Y = the task's assigned agent. +function makeColumnAgent(overrides: Record = {}) { + return { + id: "agent-X", + name: "Column Agent X", + soul: "I am X.", + instructionsText: "X persona.", + memory: undefined, + permissionPolicy: { rules: {} }, + runtimeConfig: { model: "anthropic/claude-x", allowParallelExecution: false }, + ...overrides, + }; +} + +function makeAssignedAgent(overrides: Record = {}) { + return { + id: "agent-Y", + name: "Assigned Agent Y", + soul: "I am Y.", + instructionsText: "Y persona.", + memory: undefined, + permissionPolicy: { rules: {} }, + runtimeConfig: { model: "openai/gpt-y" }, + ...overrides, + }; +} + +function installTaskDoneAgent() { + mockedCreateFnAgent.mockImplementation((async (opts: any) => { + const tools = opts.customTools || []; + return { + session: { + prompt: vi.fn().mockImplementation(async () => { + const done = tools.find((t: any) => t.name === "fn_task_done"); + if (done) await done.execute("tool-1", {}); + }), + dispose: vi.fn(), + subscribe: vi.fn(), + on: vi.fn(), + setModel: vi.fn(), + sessionManager: { getLeafId: vi.fn().mockReturnValue("leaf-1") }, + state: {}, + }, + }; + }) as any); +} + +function makeExecutor( + store: ReturnType, + agentsById: Record, + heartbeatRunsByAgent: Record = {}, +) { + const agentStore = { + getAgent: vi.fn(async (id: string) => agentsById[id] ?? null), + getActiveHeartbeatRun: vi.fn(async (id: string) => heartbeatRunsByAgent[id] ?? null), + }; + const executor = new TaskExecutor(store as any, "/tmp/test", { agentStore } as any); + return { executor, agentStore }; +} + +function singleSessionTask(overrides: Record = {}) { + return { + id: "FN-001", + title: "Test", + description: "Test task", + column: "in-progress", + dependencies: [], + steps: [{ name: "Implement", status: "in-progress" }], + currentStep: 0, + log: [], + prompt: "# test\n## Steps\n### Step 0: Implement\n- [ ] implement", + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + ...overrides, + }; +} + +function seedSeam(executor: TaskExecutor, taskId: string, governingNodeId: string, binding: WorkflowColumnAgent | undefined) { + (executor as any).graphSeamGoverningNodeId.set(taskId, governingNodeId); + (executor as any).graphColumnAgentResolver.set(taskId, (nodeId: string) => + nodeId === governingNodeId ? binding : undefined, + ); +} + +function lastFnAgentOpts() { + const calls = mockedCreateFnAgent.mock.calls; + return calls[calls.length - 1]?.[0] as any; +} + +function loggedLines(store: ReturnType): string[] { + return store.logEntry.mock.calls.map((call: any[]) => String(call[1] ?? "")); +} + +/** v2 IR with an execute-seam prompt node whose column binds `binding`. */ +function irWithExecuteSeamColumn(binding: WorkflowColumnAgent): WorkflowIr { + return { + version: "v2", + name: "test-wf", + columns: [ + { id: "in-progress", name: "In Progress", traits: [], agent: binding }, + { id: "todo", name: "Todo", traits: [] }, + ], + nodes: [ + { id: "exec-node", kind: "prompt", column: "in-progress", config: { seam: "execute" } } as any, + ], + edges: [], + } as unknown as WorkflowIr; +} + +describe("column-agent principal alignment (plan U5)", () => { + beforeEach(() => { + resetExecutorMocks(); + }); + + // ── (a) Action gating principal (R5) ────────────────────────────────────── + + describe("action gating principal", () => { + it("override column governs → gating context built for X (not the assigned Y)", async () => { + const store = createMockStore(); + const task = singleSessionTask({ assignedAgentId: "agent-Y" }); + store.getTask.mockResolvedValue(task as any); + const { executor } = makeExecutor(store, { + "agent-Y": makeAssignedAgent(), + "agent-X": makeColumnAgent(), + }); + installTaskDoneAgent(); + + seedSeam(executor, task.id, "exec-node", OVERRIDE_COL); + await (executor as any).runImplementationPhase(task); + + const opts = lastFnAgentOpts(); + // R5: action gating is computed for the agent ACTUALLY running. + expect(opts.actionGateContext?.agentId).toBe("agent-X"); + expect(opts.permanentAgentGating?.requester?.actorId).toBe("agent-X"); + }); + + it("no binding → gating context built for the assigned Y (byte-identical)", async () => { + const store = createMockStore(); + const task = singleSessionTask({ assignedAgentId: "agent-Y" }); + store.getTask.mockResolvedValue(task as any); + const { executor } = makeExecutor(store, { "agent-Y": makeAssignedAgent() }); + installTaskDoneAgent(); + + // No seam slots seeded → legacy path. + await (executor as any).runImplementationPhase(task); + + const opts = lastFnAgentOpts(); + expect(opts.actionGateContext?.agentId).toBe("agent-Y"); + expect(opts.permanentAgentGating?.requester?.actorId).toBe("agent-Y"); + }); + }); + + // ── (b) Heartbeat deferral — forward direction (R6) ─────────────────────── + + describe("heartbeat deferral: effective principal", () => { + it("override column X (allowParallelExecution=false) with an active heartbeat run → resolveEffectivePrincipalId returns X and defers", async () => { + const store = createMockStore(); + const task = singleSessionTask({ assignedAgentId: "agent-Y" }); + const { executor } = makeExecutor( + store, + { "agent-Y": makeAssignedAgent(), "agent-X": makeColumnAgent() }, + { "agent-X": { id: "run-x" } }, // active heartbeat run for X + ); + + // Seam binding is known at the deferral gate (set by the seam before + // re-entering execute()). + seedSeam(executor, task.id, "exec-node", OVERRIDE_COL); + + // The effective principal for this seam is X, not the assigned Y. + const principal = (executor as any).resolveEffectivePrincipalId(task, task); + expect(principal).toBe("agent-X"); + + // X has allowParallelExecution=false AND an active run → defer. + expect(await (executor as any).shouldDeferForHeartbeat("agent-X")).toBe(true); + // Y has no such constraint → the legacy filter alone would NOT defer. + expect(await (executor as any).shouldDeferForHeartbeat("agent-Y")).toBe(false); + }); + + it("no binding → effective principal is the assigned agent (byte-identical)", () => { + const store = createMockStore(); + const task = singleSessionTask({ assignedAgentId: "agent-Y" }); + const { executor } = makeExecutor(store, { "agent-Y": makeAssignedAgent() }); + // No seam slots → legacy. + expect((executor as any).resolveEffectivePrincipalId(task, task)).toBe("agent-Y"); + }); + }); + + // ── (b) resumeTaskForAgent two-pass (R6) ────────────────────────────────── + + describe("resumeTaskForAgent: effective-agent second pass", () => { + function resumeStore(task: any, ir: WorkflowIr) { + const store = createMockStore(); + store.getSettings.mockResolvedValue({ + globalPause: false, + enginePaused: false, + experimentalFeatures: { workflowGraphExecutor: true }, + } as any); + store.listTasks.mockResolvedValue([task] as any); + store.getTaskWorkflowSelection = vi.fn().mockReturnValue({ workflowId: "wf-1", stepIds: [] }); + store.getWorkflowDefinition = vi.fn().mockResolvedValue({ ir }); + return store; + } + + it("override column re-keys an in-progress task to X → pass 2 re-dispatches it (the assignedAgentId filter alone misses it)", async () => { + // Task assigned to Y, but its execute-seam column binds X (override). + const task = singleSessionTask({ id: "FN-RES", assignedAgentId: "agent-Y" }); + const store = resumeStore(task, irWithExecuteSeamColumn(OVERRIDE_COL)); + const { executor } = makeExecutor(store, { + "agent-Y": makeAssignedAgent(), + "agent-X": makeColumnAgent(), + }); + const executeSpy = vi.spyOn(executor, "execute").mockResolvedValue(undefined as any); + + // Pass 1 (assignedAgentId === "agent-X") would NOT match — Y is assigned. + await executor.resumeTaskForAgent("agent-X"); + + // Pass 2 (effective column agent === X) re-dispatched it. + expect(executeSpy).toHaveBeenCalledTimes(1); + expect(executeSpy.mock.calls[0][0]).toMatchObject({ id: "FN-RES" }); + }); + + it("pass 1 still re-dispatches directly-assigned tasks (legacy)", async () => { + const task = singleSessionTask({ id: "FN-ASG", assignedAgentId: "agent-X" }); + const store = resumeStore(task, irWithExecuteSeamColumn(OVERRIDE_COL)); + const { executor } = makeExecutor(store, { "agent-X": makeColumnAgent() }); + const executeSpy = vi.spyOn(executor, "execute").mockResolvedValue(undefined as any); + + await executor.resumeTaskForAgent("agent-X"); + expect(executeSpy).toHaveBeenCalledTimes(1); // not double-dispatched by pass 2 + }); + + it("defer column with task own complete model pair → X is NOT the effective agent, pass 2 does not fire", async () => { + const task = singleSessionTask({ + id: "FN-DEF", + assignedAgentId: "agent-Y", + modelProvider: "task-prov", + modelId: "task-model", + }); + const store = resumeStore(task, irWithExecuteSeamColumn(DEFER_COL)); + 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).not.toHaveBeenCalled(); + }); + }); + + // ── (b) Reverse direction: isAgentEffectivelyExecuting (R6) ─────────────── + + describe("reverse-direction guard: isAgentEffectivelyExecuting", () => { + it("X executing an override-column task it is NOT assigned to → effective-executing is true for X", async () => { + const store = createMockStore(); + const task = singleSessionTask({ assignedAgentId: "agent-Y" }); + store.getTask.mockResolvedValue(task as any); + const { executor } = makeExecutor(store, { + "agent-Y": makeAssignedAgent(), + "agent-X": makeColumnAgent(), + }); + installTaskDoneAgent(); + + // Before any session: nothing effectively executing. + expect(executor.isAgentEffectivelyExecuting("agent-X")).toBe(false); + + // While the override session runs, the map is populated. We assert the map + // directly to avoid coupling to teardown timing of the mocked session. + seedSeam(executor, task.id, "exec-node", OVERRIDE_COL); + const setSpy = vi.spyOn((executor as any).effectiveColumnAgentByTask, "set"); + await (executor as any).runImplementationPhase(task); + + // The execute seam recorded X as the effective principal for the task. + expect(setSpy).toHaveBeenCalledWith(task.id, "agent-X"); + }); + + it("the heartbeat scheduler reverse guard consults the injected callback", async () => { + // Mirror the in-process-runtime wiring: the scheduler gets + // isAgentEffectivelyExecuting from the executor. Prove the guard short-circuits. + const store = createMockStore(); + store.getTask.mockResolvedValue(singleSessionTask({ assignedAgentId: "agent-Y" }) as any); + const { executor } = makeExecutor(store, {}); + // Pretend X is effectively executing some task. + (executor as any).effectiveColumnAgentByTask.set("FN-Z", "agent-X"); + const cb = (agentId: string) => executor.isAgentEffectivelyExecuting(agentId); + expect(cb("agent-X")).toBe(true); + expect(cb("agent-Y")).toBe(false); + }); + }); + + // ── (c) Restart watcher via re-resolution (R7/KTD-4) ────────────────────── + + describe("restart watcher: column-agent invalidation", () => { + function activeGraphSession(executor: TaskExecutor, taskId: string, governing: string, binding: WorkflowColumnAgent) { + const setModel = vi.fn(); + const session = { setModel, dispose: vi.fn() } as any; + seedSeam(executor, taskId, governing, binding); + (executor as any).activeSessions.set(taskId, { + session, + seenSteeringIds: new Set(), + lastResolvedModelProvider: "anthropic", + lastResolvedModelId: "claude-x", + lastTaskModelProvider: undefined, + lastTaskModelId: undefined, + lastAssignedAgentId: "agent-Y", + lastEffectiveColumnAgentId: "agent-X", + }); + return { setModel }; + } + + it("workflow edit changes the column agent's model while a session runs → restart (model hot-swap) fires", async () => { + const store = createMockStore(); + // modelRegistry.find returns a truthy model so setModel is invoked. + const find = vi.fn().mockReturnValue({ provider: "anthropic", modelId: "claude-x2" }); + const task = singleSessionTask({ assignedAgentId: "agent-Y" }); + // Column agent X now advertises a NEW model (workflow edit re-pointed / agent config changed). + const { executor } = makeExecutor(store, { + "agent-X": makeColumnAgent({ runtimeConfig: { model: "anthropic/claude-x2", allowParallelExecution: false } }), + }); + (executor as any)._modelRegistry = { find }; + + const { setModel } = activeGraphSession(executor, task.id, "exec-node", OVERRIDE_COL); + + // The watcher fires on task:updated. + store._trigger("task:updated", task); + await vi.waitFor(() => expect(setModel).toHaveBeenCalled()); + + expect(find).toHaveBeenCalledWith("anthropic", "claude-x2"); + expect(loggedLines(store).some((l) => l.includes("Column agent changed"))).toBe(true); + }); + + it("column agent deleted mid-session → no restart storm, no setModel, fallback recorded (R8)", async () => { + const store = createMockStore(); + const find = vi.fn().mockReturnValue({ provider: "anthropic", modelId: "claude-x" }); + const task = singleSessionTask({ assignedAgentId: "agent-Y" }); + // agent-X is ABSENT from the registry (deleted). + const { executor } = makeExecutor(store, {}); + (executor as any)._modelRegistry = { find }; + + const { setModel } = activeGraphSession(executor, task.id, "exec-node", OVERRIDE_COL); + + store._trigger("task:updated", task); + // Wait for the async handler to record the fallback. + await vi.waitFor(() => + expect(loggedLines(store).some((l) => l.includes("deleted mid-session") && l.includes("no restart"))).toBe(true), + ); + + // No model swap — the running session keeps its current model. + expect(setModel).not.toHaveBeenCalled(); + expect(find).not.toHaveBeenCalled(); + // Tracked id cleared so we stop probing every tick. + expect((executor as any).activeSessions.get(task.id).lastEffectiveColumnAgentId).toBeNull(); + }); + + it("legacy entry (no effective column agent) → the column-invalidation block is skipped", async () => { + const store = createMockStore(); + const find = vi.fn(); + const task = singleSessionTask({ assignedAgentId: "agent-Y" }); + const { executor } = makeExecutor(store, { "agent-X": makeColumnAgent() }); + (executor as any)._modelRegistry = { find }; + + const setModel = vi.fn(); + (executor as any).activeSessions.set(task.id, { + session: { setModel, dispose: vi.fn() }, + seenSteeringIds: new Set(), + lastResolvedModelProvider: "openai", + lastResolvedModelId: "gpt-y", + lastTaskModelProvider: undefined, + lastTaskModelId: undefined, + lastAssignedAgentId: "agent-Y", + lastEffectiveColumnAgentId: null, // legacy + }); + // No seam slots seeded. + + store._trigger("task:updated", task); + await new Promise((r) => setTimeout(r, 0)); + + // The column-invalidation block never ran (no column-agent fetch / swap). + expect(loggedLines(store).some((l) => l.includes("Column agent changed"))).toBe(false); + }); + }); + + // ── Split-branch note ───────────────────────────────────────────────────── + // Per-session principals: the executor tracks the effective principal per TASK + // (effectiveColumnAgentByTask) and per active session-build, so two distinct + // tasks bound to different columns yield two principals. Asserting TWO truly + // concurrent split-branch SESSIONS for ONE task is not cheaply expressible with + // this single-session mock harness (it pins one createFnAgent call per + // runImplementationPhase), so we assert the per-task divergence instead. + describe("per-task principal divergence (split-branch surrogate)", () => { + it("two tasks bound to different column agents resolve to different effective principals", () => { + const store = createMockStore(); + const { executor } = makeExecutor(store, {}); + const taskA = singleSessionTask({ id: "FN-A", assignedAgentId: "agent-Y" }); + const taskB = singleSessionTask({ id: "FN-B", assignedAgentId: "agent-Y" }); + seedSeam(executor, "FN-A", "exec-node", { agentId: "agent-X", mode: "override" }); + seedSeam(executor, "FN-B", "exec-node", { agentId: "agent-Z", mode: "override" }); + expect((executor as any).resolveEffectivePrincipalId(taskA, taskA)).toBe("agent-X"); + expect((executor as any).resolveEffectivePrincipalId(taskB, taskB)).toBe("agent-Z"); + }); + }); +}); diff --git a/packages/engine/src/agent-heartbeat.ts b/packages/engine/src/agent-heartbeat.ts index b85ea5e1db..307528d9c2 100644 --- a/packages/engine/src/agent-heartbeat.ts +++ b/packages/engine/src/agent-heartbeat.ts @@ -3649,17 +3649,26 @@ export class HeartbeatTriggerScheduler { private configRevisionListener: ((agentId: string, revision: AgentConfigRevision) => void) | null = null; private deletedListener: ((agentId: string) => void) | null = null; private isTaskExecuting?: (taskId: string) => boolean; + /** Column-agent principal alignment (plan U5, R6). True when the agent is the + * EFFECTIVE column-agent principal of some currently-executing task — i.e. an + * override/defer-bound column staffs it, even though the agent is not that task's + * `assignedAgentId`. The reverse-direction parallel-execution guards consult this + * in addition to `isTaskExecuting(agent.taskId)` so an `allowParallelExecution=false` + * column agent does not heartbeat concurrently with its own override session. + * Absent (legacy/no executor wiring) → treated as never effectively executing. */ + private isAgentEffectivelyExecuting?: (agentId: string) => boolean; private timerAuditIntervalHandle: ReturnType | null = null; private static readonly TIMER_AUDIT_INTERVAL_MS = 60_000; private static readonly DEFAULT_REPAIR_STALE_MULTIPLIER = 2; private static readonly DEFAULT_HEARTBEAT_TIMEOUT_MS = 60_000; - constructor(store: AgentStore, callback: TriggerCallback, taskStore?: TaskStore, options?: { isTaskExecuting?: (taskId: string) => boolean }) { + constructor(store: AgentStore, callback: TriggerCallback, taskStore?: TaskStore, options?: { isTaskExecuting?: (taskId: string) => boolean; isAgentEffectivelyExecuting?: (agentId: string) => boolean }) { this.store = store; this.callback = callback; this.taskStore = taskStore; this.isTaskExecuting = options?.isTaskExecuting; + this.isAgentEffectivelyExecuting = options?.isAgentEffectivelyExecuting; } /** @@ -3955,9 +3964,16 @@ export class HeartbeatTriggerScheduler { return; } - // Guard: when parallel execution is disabled, skip if the bound task is actively executing - if (runtimeConfig.allowParallelExecution === false && this.isTaskExecuting?.(taskId)) { - heartbeatLog.log(`Assignment tick skipped for ${agent.id} (parallel execution disabled, task ${taskId} executing)`); + // Guard: when parallel execution is disabled, skip if the bound task is + // actively executing — OR (plan U5, R6, reverse direction) if this agent is + // the EFFECTIVE column-agent principal of some other actively-executing task + // it is not assigned to. Without the second check an override-column agent + // would heartbeat concurrently with its own column-bound session. + if ( + runtimeConfig.allowParallelExecution === false + && (this.isTaskExecuting?.(taskId) || this.isAgentEffectivelyExecuting?.(agent.id)) + ) { + heartbeatLog.log(`Assignment tick skipped for ${agent.id} (parallel execution disabled, task ${taskId} or column-bound session executing)`); return; } @@ -4323,9 +4339,19 @@ export class HeartbeatTriggerScheduler { return; } - // Guard: when parallel execution is disabled, skip if the agent's bound task is actively executing - if (timerRc.allowParallelExecution === false && agent.taskId && this.isTaskExecuting?.(agent.taskId)) { - heartbeatLog.log(`Timer tick skipped for ${agentId} (parallel execution disabled, task ${agent.taskId} executing)`); + // Guard: when parallel execution is disabled, skip if the agent's bound task is + // actively executing — OR (plan U5, R6, reverse direction) if this agent is the + // EFFECTIVE column-agent principal of some actively-executing task it is not + // assigned to (override/defer column staffing). `agent.taskId` may be empty in + // the column-bound case, so the effective check is independent of it. + if ( + timerRc.allowParallelExecution === false + && ( + (agent.taskId && this.isTaskExecuting?.(agent.taskId)) + || this.isAgentEffectivelyExecuting?.(agentId) + ) + ) { + heartbeatLog.log(`Timer tick skipped for ${agentId} (parallel execution disabled, bound task ${agent.taskId ?? "—"} or column-bound session executing)`); return; } diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 521a154928..7423f023e5 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -1144,9 +1144,26 @@ export class TaskExecutor { lastTaskModelProvider?: string | null; lastTaskModelId?: string | null; lastAssignedAgentId?: string | null; + // Column-agent restart-invalidation (plan U5, R7/KTD-4). The effective + // column-agent id governing this session's seam (undefined when no binding + // governs — the legacy path). Tracked so the watcher can detect a workflow- + // definition edit or agent runtimeConfig change that re-keys the column- + // effective agent/model mid-flight and trigger the same restart path a + // task.modelProvider change does today. + lastEffectiveColumnAgentId?: string | null; }>(); /** Active step-session executors per task (mutually exclusive with activeSessions). */ private activeStepExecutors = new Map(); + /** Column-agent principal alignment (plan U5, R6): the EFFECTIVE column-agent id + * currently running each executing task's coding/step session, when an + * override/defer binding governs the in-flight seam. Keyed by task id, populated + * by the execute / step-execute seam right after `resolveSeamColumnAgent` yields a + * column agent, and cleared alongside the session (deleteActiveSession / + * deleteActiveStepExecutor). Powers `isAgentEffectivelyExecuting`, the + * reverse-direction heartbeat-scheduler guard that must know an agent is running a + * task it is not `assignedAgentId` on. Empty for the legacy/no-binding path, so + * that path is byte-identical. */ + private effectiveColumnAgentByTask = new Map(); /** Active pre-merge workflow step sessions per task. */ private activeWorkflowStepSessions = new Map(); /** Active configured-command abort controllers keyed by task. */ @@ -1194,6 +1211,7 @@ export class TaskExecutor { lastTaskModelProvider?: string | null; lastTaskModelId?: string | null; lastAssignedAgentId?: string | null; + lastEffectiveColumnAgentId?: string | null; }, worktreePath: string): void { this.activeSessions.set(taskId, sessionState); activeSessionRegistry.registerPath(worktreePath, { taskId, kind: "executor", ownerKey: taskId }); @@ -1201,6 +1219,8 @@ export class TaskExecutor { private deleteActiveSession(taskId: string, worktreePath?: string): void { this.activeSessions.delete(taskId); + // U5: drop the effective column-agent principal for this task's session. + this.effectiveColumnAgentByTask.delete(taskId); const resolvedWorktreePath = worktreePath ?? this.activeWorktrees.get(taskId); if (resolvedWorktreePath) { activeSessionRegistry.unregisterPath(resolvedWorktreePath); @@ -1214,6 +1234,8 @@ export class TaskExecutor { private deleteActiveStepExecutor(taskId: string, worktreePath?: string): void { this.activeStepExecutors.delete(taskId); + // U5: drop the effective column-agent principal for this task's step session. + this.effectiveColumnAgentByTask.delete(taskId); const resolvedWorktreePath = worktreePath ?? this.activeWorktrees.get(taskId); if (resolvedWorktreePath) { activeSessionRegistry.unregisterPath(resolvedWorktreePath); @@ -2057,6 +2079,96 @@ export class TaskExecutor { return; } + // Column-agent restart-invalidation (plan U5, R7/KTD-4). A workflow- + // definition edit (re-pointing a column's agent) or an agent runtimeConfig + // change mutates NOTHING the task-field diff below observes — the watcher + // would never see it. KTD-4's primary mechanism is event-driven invalidation, + // but no `workflow:updated`/`agent:updated` store event exists on TaskStore + // today (only task:/settings: events). Per the unit's documented fallback, we + // re-resolve the column-effective agent/model on each `task:updated` tick for + // GRAPH-MODE active entries ONLY (those whose session adopted a column agent — + // `lastEffectiveColumnAgentId != null`). This is bounded by the active session + // count, and only graph runs with a real column binding pay any cost. The + // weaker guarantee (vs an arbitrary-time diff) is that a stale session + // restarts on the next tick, not instantly — acceptable per the Risks note. + // + // agent-DELETED → fall back per R8 (no restart; the running session finishes + // on its current model). agent-CHANGED (different effective agent OR same + // agent with a new runtimeConfig model) → hot-swap, same path as a + // task.modelProvider change. + if ( + this.activeSessions.has(task.id) + && !task.paused + && (this.activeSessions.get(task.id)!.lastEffectiveColumnAgentId ?? null) !== null + && this.graphSeamGoverningNodeId.has(task.id) + && this.graphColumnAgentResolver.has(task.id) + ) { + const activeEntry = this.activeSessions.get(task.id)!; + const governingNodeId = this.graphSeamGoverningNodeId.get(task.id)!; + 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, + }); + if (effective.source === "column-agent") { + // Fetch the (possibly changed) effective column agent, best-effort. + const newAgent = await this.options.agentStore?.getAgent(effective.agentId).catch(() => null) ?? null; + if (!newAgent) { + // agent-DELETED (R8): fall back, NO restart. The running session + // keeps its current model; the NEXT resolution falls back. Update the + // 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( + task.id, + `Column agent '${effective.agentId}' deleted mid-session — falling back to current model, no restart (R8)`, + undefined, + this.getRunContextFor(task.id), + ); + activeEntry.lastEffectiveColumnAgentId = null; + } + } else { + const settings = await this.store.getSettings(); + const { provider: newProvider, modelId: newModelId } = resolveExecutorSessionModel( + task.modelProvider, + task.modelId, + settings, + (newAgent.runtimeConfig ?? undefined) as Record | undefined, + ); + const agentChanged = (activeEntry.lastEffectiveColumnAgentId ?? null) !== newAgent.id; + const providerChanged = newProvider !== activeEntry.lastResolvedModelProvider; + const modelIdChanged = newModelId !== activeEntry.lastResolvedModelId; + if (agentChanged || providerChanged || modelIdChanged) { + activeEntry.lastEffectiveColumnAgentId = newAgent.id; + activeEntry.lastResolvedModelProvider = newProvider; + activeEntry.lastResolvedModelId = newModelId; + if (newProvider && newModelId) { + try { + const model = this.modelRegistry.find(newProvider, newModelId); + if (model) { + await activeEntry.session.setModel(model); + executorLog.log(`${task.id}: column-agent hot-swap → agent '${newAgent.id}' model ${newProvider}/${newModelId}`); + await this.store.logEntry(task.id, `Column agent changed — model now ${newProvider}/${newModelId} (agent ${newAgent.id})`, undefined, this.getRunContextFor(task.id)); + } else { + executorLog.log(`${task.id}: column-agent model ${newProvider}/${newModelId} not found in registry for hot-swap`); + } + } 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)); + } + } + } + } + } + } + } + // 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)!; @@ -3050,30 +3162,100 @@ export class TaskExecutor { } /** - * Re-dispatch execute() for any unstarted in-progress task belonging to the - * given agent. Called after a heartbeat run completes to unblock tasks that - * were deferred by the allowParallelExecution=false gate. + * Re-dispatch execute() for any unstarted in-progress task whose EFFECTIVE + * principal is the given agent. Called after a heartbeat run completes to unblock + * tasks that were deferred by the allowParallelExecution=false gate. + * + * TWO-PASS (plan U5, R6) — the `assignedAgentId`-only filter alone misses tasks an + * override/defer column binding re-keys to the column agent: + * 1. Tasks directly `assignedAgentId === agentId` (legacy, byte-identical). + * 2. Tasks whose effective column agent resolves to `agentId` for their + * governing execute / step-execute seam — resolved per candidate via the core + * column-agent resolver against the task's workflow IR. Bounded: only + * not-already-executing in-progress tasks are probed, and the IR resolution is + * best-effort (failure → skip, never strands resume). + * A task re-dispatched by pass 1 is not re-dispatched by pass 2 (dedupe set). */ async resumeTaskForAgent(agentId: string): Promise { const settings = await this.store.getSettings(); if (settings.globalPause || settings.enginePaused) return; const tasks = await this.store.listTasks({ slim: true, column: "in-progress" }); + const dispatched = new Set(); + const isDispatchable = (task: Task): boolean => + !task.deletedAt + && !task.paused + && !this.executing.has(task.id) + && !this.activeSessions.has(task.id) + && !this.activeStepExecutors.has(task.id) + && !this.activeWorkflowStepSessions.has(task.id); + const dispatch = (task: Task, reason: string): void => { + if (dispatched.has(task.id)) return; + dispatched.add(task.id); + executorLog.log(`${task.id}: re-dispatching execute() after heartbeat completion for agent ${agentId} (${reason})`); + this.execute(task).catch((err) => + executorLog.error(`Failed to resume ${task.id} after heartbeat completion:`, err), + ); + }; + + // Pass 1: directly-assigned tasks (legacy behavior, byte-identical). for (const task of tasks) { - if ( - task.assignedAgentId === agentId - && !task.deletedAt - && !task.paused - && !this.executing.has(task.id) - && !this.activeSessions.has(task.id) - && !this.activeStepExecutors.has(task.id) - && !this.activeWorkflowStepSessions.has(task.id) - ) { - executorLog.log(`${task.id}: re-dispatching execute() after heartbeat completion for agent ${agentId}`); - this.execute(task).catch((err) => - executorLog.error(`Failed to resume ${task.id} after heartbeat completion:`, err), - ); + if (task.assignedAgentId === agentId && isDispatchable(task)) { + dispatch(task, "assigned"); } } + + // Pass 2: tasks whose EFFECTIVE column agent resolves to `agentId`. Only + // experimental graph-executor tasks can carry a column binding; the IR resolve + // is best-effort and skipped for tasks already dispatched/executing. + if (!isExperimentalFeatureEnabled(settings, "workflowGraphExecutor")) return; + for (const task of tasks) { + if (dispatched.has(task.id) || !isDispatchable(task)) continue; + // Skip tasks the assigned-agent filter already covers — a redundant column + // binding to the same agent would only re-confirm pass 1. + if (task.assignedAgentId === agentId) continue; + let matches = false; + try { + matches = await this.taskEffectiveAgentMatches(task, agentId); + } catch { + matches = false; + } + if (matches) dispatch(task, "effective-column-agent"); + } + } + + /** Column-agent principal alignment (plan U5, R6). True when the EFFECTIVE agent + * governing `task`'s execute or step-execute seam — resolved through the shared + * core resolver against the task's workflow IR — is `agentId`. Used by the + * `resumeTaskForAgent` second pass to re-dispatch column-bound tasks the + * `assignedAgentId` filter misses. Best-effort: an unresolvable IR yields false. */ + private async taskEffectiveAgentMatches(task: Task, agentId: string): Promise { + 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); + + // 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). + 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; + } + } + return false; } /** @@ -4690,6 +4872,63 @@ export class TaskExecutor { return { agent, mode: binding.mode }; } + /** + * Column-agent principal alignment (plan U5, R6). Resolve the EFFECTIVE + * principal id for the in-flight seam WITHOUT fetching the full Agent or + * emitting an adoption log — a light counterpart to {@link resolveSeamColumnAgent} + * used by the heartbeat-deferral gate (which only needs the id to call + * {@link shouldDeferForHeartbeat}, which itself loads the agent). + * + * Returns the column-agent id when a governing binding selects it via the shared + * core resolver (`resolveEffectiveAgent`, KTD-2/KTD-5), else `task.assignedAgentId` + * (the legacy principal). Returns `undefined` only when there is no principal at + * all (no binding AND no assigned agent) — keeping the no-binding path + * byte-identical to the prior `assignedAgentId` deferral behavior. + */ + private resolveEffectivePrincipalId( + task: Task, + detail: Task, + ): string | undefined { + const assignedAgentId = typeof detail.assignedAgentId === "string" && detail.assignedAgentId.trim() + ? detail.assignedAgentId.trim() + : undefined; + + const governingNodeId = this.graphSeamGoverningNodeId.get(task.id); + const resolveBinding = this.graphColumnAgentResolver.get(task.id); + if (!governingNodeId || !resolveBinding) return assignedAgentId; + + 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, + }); + if (effective.source === "column-agent") return effective.agentId; + return assignedAgentId; + } + + /** + * Column-agent principal alignment (plan U5, R6). True when `agentId` is the + * EFFECTIVE column-agent principal currently running some executing task's + * coding/step session — i.e. an override/defer-bound column staffs it, even + * though the agent is not the task's `assignedAgentId`. Injected into the + * heartbeat scheduler's reverse-direction parallel-execution guards + * (`agent-heartbeat.ts`) so an `allowParallelExecution=false` column agent does + * not heartbeat concurrently with its own override session. Returns false for the + * legacy/no-binding path (the map is empty), preserving prior behavior exactly. + */ + isAgentEffectivelyExecuting(agentId: string): boolean { + if (!agentId) return false; + for (const effectiveId of this.effectiveColumnAgentByTask.values()) { + if (effectiveId === agentId) return true; + } + return false; + } + /** Run a custom (non-seam) graph node on the proven WorkflowStep machinery. * * `columnBinding` (plan U3) is the agent binding governing this node's @@ -4970,9 +5209,17 @@ export class TaskExecutor { return; } - const assignedAgentId = task.assignedAgentId; - if (assignedAgentId && await this.shouldDeferForHeartbeat(assignedAgentId)) { - executorLog.log(`${task.id}: skipping execute — agent ${assignedAgentId} has active heartbeat run (allowParallelExecution=false)`); + // Column-agent principal alignment (plan U5, R6): the heartbeat-deferral gate + // must consult the EFFECTIVE principal, not blindly `assignedAgentId`. For a + // graph-routed seam the binding context (governing node id + per-run resolver) + // is already set by the time the seam re-enters execute() — so the effective + // column agent (when an override/defer binding governs) is the principal whose + // `allowParallelExecution=false` must serialize. For the legacy/no-binding path + // `resolveEffectivePrincipalId` returns `assignedAgentId`, so the gate is + // byte-identical to before. + const deferralPrincipalId = this.resolveEffectivePrincipalId(task, task); + if (deferralPrincipalId && await this.shouldDeferForHeartbeat(deferralPrincipalId)) { + executorLog.log(`${task.id}: skipping execute — agent ${deferralPrincipalId} has active heartbeat run (allowParallelExecution=false)`); // Release the slot we just claimed — we never actually ran. this.executing.delete(task.id); executingTaskLock.release(task.id); @@ -5414,11 +5661,18 @@ export class TaskExecutor { // step-execute node's declared column binds an agent that supersedes the // task's assigned agent, the per-step session's MODEL, runtime hint, and // attribution adopt the column agent. The core resolver decides defer vs - // override (KTD-2); a missing agent logs + falls back (R8). Gating contexts - // still key off the ASSIGNED agent here — principal substitution for - // gating/heartbeat is U5 (kept out of this unit deliberately). + // override (KTD-2); a missing agent logs + falls back (R8). Principal + // alignment (U5, R5/R6): the gating contexts below ALSO key off the + // effective `stepIdentityAgent`, and the effective principal is tracked for + // the reverse-direction heartbeat guard. const stepColumnAgent = await this.resolveSeamColumnAgent(task, detail); const stepIdentityAgent = stepColumnAgent?.agent ?? stepSessionAgent; + // U5 (R6): track the effective column-agent principal so the heartbeat + // scheduler's reverse guard knows this agent is executing a task it may not + // be assigned to. Cleared in deleteActiveStepExecutor. + if (stepColumnAgent?.agent) { + this.effectiveColumnAgentByTask.set(task.id, stepColumnAgent.agent.id); + } const stepSessionRuntimeHint = extractRuntimeHint(stepIdentityAgent?.runtimeConfig); let accumulatedStepTokenUsage = detail.tokenUsage; @@ -5438,8 +5692,8 @@ export class TaskExecutor { // Attribute the per-step run auditor to the column agent when it governs // (U4); absent → StepSessionExecutor falls back to assignedAgentId. effectiveAgentId: stepColumnAgent?.agent.id, - actionGateContext: this.buildActionGateContext(task.id, stepSessionAgent, settings.defaultAgentPermissionPolicy), - permanentAgentGating: this.buildPermanentAgentGatingContext(task.id, stepSessionAgent, settings.defaultAgentPermissionPolicy), + actionGateContext: this.buildActionGateContext(task.id, stepIdentityAgent, settings.defaultAgentPermissionPolicy), + permanentAgentGating: this.buildPermanentAgentGatingContext(task.id, stepIdentityAgent, settings.defaultAgentPermissionPolicy), // Pass skill selection context from the main executor session skillSelection: skillContext.skillSelectionContext, // Pass agentStore and messageStore for delegation and messaging tools @@ -5929,6 +6183,12 @@ export class TaskExecutor { const columnAgentSeam = await this.resolveSeamColumnAgent(task, detail); const identityAgent = columnAgentSeam?.agent ?? assignedAgent; const executorRuntimeHint = extractRuntimeHint(identityAgent?.runtimeConfig); + // U5 (R6): track the effective column-agent principal so the heartbeat + // scheduler's reverse guard knows this agent is executing a task it may not + // be assigned to. Cleared in deleteActiveSession. + if (columnAgentSeam?.agent) { + this.effectiveColumnAgentByTask.set(task.id, columnAgentSeam.agent.id); + } // Log fast mode status if (executionMode === "fast") { @@ -6140,8 +6400,14 @@ export class TaskExecutor { taskEnv, // Skill selection: use assigned agent skills if available, otherwise role fallback ...(skillContext.skillSelectionContext ? { skillSelection: skillContext.skillSelectionContext } : {}), - actionGateContext: this.buildActionGateContext(task.id, assignedAgent, settings.defaultAgentPermissionPolicy), - permanentAgentGating: this.buildPermanentAgentGatingContext(task.id, assignedAgent, settings.defaultAgentPermissionPolicy), + // Column-agent principal alignment (plan U5, R5): action gating is + // computed for the agent ACTUALLY RUNNING. When the governing execute + // seam's column binds an agent that supersedes the assigned agent, + // `identityAgent` is that column agent; otherwise it is `assignedAgent` + // (byte-identical to before). The builders already accept an `Agent` + // object, so this is a call-site object swap, not gating-internals surgery. + actionGateContext: this.buildActionGateContext(task.id, identityAgent, settings.defaultAgentPermissionPolicy), + permanentAgentGating: this.buildPermanentAgentGatingContext(task.id, identityAgent, settings.defaultAgentPermissionPolicy), taskId: task.id, taskTitle: detail.title, onFallbackModelUsed: createFallbackModelObserver({ @@ -6195,6 +6461,10 @@ export class TaskExecutor { lastTaskModelProvider: detail.modelProvider, lastTaskModelId: detail.modelId, lastAssignedAgentId: detail.assignedAgentId ?? null, + // U5 (R7): the effective column-agent governing this session (null when no + // binding governs — legacy path). The watcher re-resolves this for graph- + // mode entries to detect a mid-flight workflow-edit / agent-config change. + lastEffectiveColumnAgentId: columnAgentSeam?.agent.id ?? null, }, worktreePath); let leaseRenewalTimer: ReturnType | undefined; @@ -6570,8 +6840,10 @@ export class TaskExecutor { taskEnv, // Skill selection: use assigned agent skills if available, otherwise role fallback ...(skillContext.skillSelectionContext ? { skillSelection: skillContext.skillSelectionContext } : {}), - actionGateContext: this.buildActionGateContext(task.id, assignedAgent, settings.defaultAgentPermissionPolicy), - permanentAgentGating: this.buildPermanentAgentGatingContext(task.id, assignedAgent, settings.defaultAgentPermissionPolicy), + // U5 (R5): retry session re-keys gating to the effective principal, + // mirroring the primary execute-seam session above. + actionGateContext: this.buildActionGateContext(task.id, identityAgent, settings.defaultAgentPermissionPolicy), + permanentAgentGating: this.buildPermanentAgentGatingContext(task.id, identityAgent, settings.defaultAgentPermissionPolicy), }); retrySession = createdRetrySession.session; if (createdRetrySession.sessionFile) { @@ -6591,6 +6863,8 @@ export class TaskExecutor { lastTaskModelProvider: detail.modelProvider, lastTaskModelId: detail.modelId, lastAssignedAgentId: detail.assignedAgentId ?? null, + // U5 (R7): preserve the effective column-agent across the retry. + lastEffectiveColumnAgentId: columnAgentSeam?.agent.id ?? null, }, worktreePath); stuckDetector?.trackTask(task.id, retrySession); diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index da1639bcff..7e9a6d71e6 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -589,7 +589,13 @@ export class InProcessRuntime }); }, this.taskStore, - { isTaskExecuting: (taskId) => this.executor.getExecutingTaskIds().has(taskId) }, + { + isTaskExecuting: (taskId) => this.executor.getExecutingTaskIds().has(taskId), + // Column-agent principal alignment (plan U5, R6): reverse-direction guard + // — an override/defer column agent must not heartbeat concurrently with a + // column-bound session it runs but is not assigned to. + isAgentEffectivelyExecuting: (agentId) => this.executor.isAgentEffectivelyExecuting(agentId), + }, ); this.triggerScheduler.start();