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.
This commit is contained in:
gsxdsm
@@ -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.
|
||||
|
||||
@@ -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();
|
||||
|
||||
|
||||
@@ -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<string, unknown> = {}) {
|
||||
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<string, unknown> = {}) {
|
||||
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<typeof createMockStore>,
|
||||
agentsById: Record<string, unknown>,
|
||||
heartbeatRunsByAgent: Record<string, unknown> = {},
|
||||
) {
|
||||
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<string, unknown> = {}) {
|
||||
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<typeof createMockStore>): 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<string>(),
|
||||
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<string>(),
|
||||
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");
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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<typeof setInterval> | 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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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<string, StepSessionExecutor>();
|
||||
/** 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<string, string>();
|
||||
/** Active pre-merge workflow step sessions per task. */
|
||||
private activeWorkflowStepSessions = new Map<string, AgentSession>();
|
||||
/** 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<string, unknown> | 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<void> {
|
||||
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<string>();
|
||||
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<boolean> {
|
||||
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<typeof setInterval> | 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);
|
||||
|
||||
|
||||
@@ -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();
|
||||
|
||||
|
||||
Reference in New Issue
Block a user