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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
189 changes: 189 additions & 0 deletions src/agent/director.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -426,6 +426,195 @@ describe("ChatDirector inference-error recovery (CL-6910)", () => {
);
expect(stale).toEqual([]);
});

function keeperManageTasksTurn(): ReactorInboundEvent {
return {
type: "inference.done",
turn: {
role: "assistant",
model: "test",
timestamp: 0,
content: [
{
type: "tool_call",
id: "manage-tasks",
name: "manage_tasks",
arguments: {
action: "create",
tasks: [{ id: "t1", title: "work", status: "doing" }],
},
},
],
},
usage: { input: 0, output: 0 },
source: "test",
} as unknown as ReactorInboundEvent;
}

function keeperManageTasksPlusSubmitTurn(): ReactorInboundEvent {
return {
type: "inference.done",
turn: {
role: "assistant",
model: "test",
timestamp: 0,
content: [
{
type: "tool_call",
id: "manage-tasks",
name: "manage_tasks",
arguments: {
action: "create",
tasks: [{ id: "t1", title: "work", status: "doing" }],
},
},
{
type: "tool_call",
id: "submit-1",
name: "submit_output",
arguments: { step: "step-1" },
},
],
},
usage: { input: 0, output: 0 },
source: "test",
} as unknown as ReactorInboundEvent;
}

function keeperTextTurn(): ReactorInboundEvent {
return {
type: "inference.done",
turn: {
role: "assistant",
model: "test",
timestamp: 0,
content: [{ type: "text", text: "all set" }],
},
usage: { input: 0, output: 0 },
source: "test",
} as unknown as ReactorInboundEvent;
}

function keeperTasksChanged(actions: ReactorAction[]): ReactorAction[] {
return actions.filter(
(a) =>
a.type === "emit" &&
(a as { eventType?: string }).eventType === CHAT_TASKS_CHANGED_EVENT,
);
}

// CL-7992 K1: every coordinator rail consulted on the turn boundary
// rethrows instead of degrading — a throwing rail rejects decide() and
// the next turn carries no stale task-change notifications.
test.each(["isActive", "currentStepIsGate", "currentStepId"] as const)(
"a throwing %s rail rejects the inference turn without leaking task-change emits",
async (rail) => {
const failure = new Error(`${rail} exploded`);
const coordinator = {
isActive: () => {
if (rail === "isActive") throw failure;
return true;
},
currentStepIsGate: () => {
if (rail === "currentStepIsGate") throw failure;
return false;
},
currentStepId: () => {
if (rail === "currentStepId") throw failure;
return null;
},
directive: () => null,
handleToolDone: () => false,
} as unknown as WorkflowCoordinator;
const director = createChatDirector("system", [], {});
director.setWorkflowCoordinator(coordinator);
const capabilities = makeCapabilities();

// The step-id rail only runs past a terminal base action, so it
// throws on a text turn; the earlier rails throw on a manage_tasks
// turn after it queues its task-change notification.
const triggering =
rail === "currentStepId" ? keeperTextTurn() : keeperManageTasksTurn();
await expect(
director.decide(triggering, mockState, capabilities),
).rejects.toBe(failure);

director.setWorkflowCoordinator(undefined);
const actions = actionsArray(
await director.decide(keeperTextTurn(), mockState, capabilities),
);
expect(keeperTasksChanged(actions)).toEqual([]);
},
);

// CL-7992 K2: a mid-turn failure outside the coordinator still rejects
// the turn, but already-queued notifications survive for the next turn.
test("a non-coordinator mid-turn failure preserves queued task-change emits for the next turn", async () => {
const failure = new Error("tool execution exploded");
const director = createChatDirector("system", [], {});
const capabilities = makeCapabilities();
const executing: ReactorCapabilities = {
...capabilities,
executeTools: (): ReactorAction => {
throw failure;
},
};

await expect(
director.decide(keeperManageTasksTurn(), mockState, executing),
).rejects.toBe(failure);

const actions = actionsArray(
await director.decide(keeperTextTurn(), mockState, capabilities),
);
expect(keeperTasksChanged(actions)).toHaveLength(1);
});

// CL-7992 K2 variant: a throwing handleToolDone degrades (it takes no
// rethrow parameter) instead of marking the turn stale, so a later
// non-coordinator failure still preserves the queued notifications.
test("a throwing handleToolDone degrades without dropping preserved task-change emits", async () => {
let handleToolDoneSeen = false;
const coordinator = {
isActive: () => false,
currentStepIsGate: () => false,
currentStepId: () => null,
directive: () => null,
handleToolDone: (): boolean => {
handleToolDoneSeen = true;
throw new Error("coordinator completion exploded");
},
} as unknown as WorkflowCoordinator;
const director = createChatDirector("system", [], {});
director.setWorkflowCoordinator(coordinator);
const capabilities = makeCapabilities();
const executing: ReactorCapabilities = {
...capabilities,
executeTools: (): ReactorAction => {
throw new Error("tool execution exploded");
},
};

await expect(
director.decide(keeperManageTasksPlusSubmitTurn(), mockState, executing),
).rejects.toThrow("tool execution exploded");

const brokenState = {
get turns(): never {
throw new Error("history unavailable");
},
} as unknown as ReactorState;
await expect(
director.decide(toolDoneEvent("submit-1"), brokenState, capabilities),
).rejects.toThrow("history unavailable");
expect(handleToolDoneSeen).toBe(true);

director.setWorkflowCoordinator(undefined);
const actions = actionsArray(
await director.decide(keeperTextTurn(), mockState, capabilities),
);
expect(keeperTasksChanged(actions)).toHaveLength(1);
});
});

// CL-7973: the director's live source id (which stamps retry decisions so a
Expand Down
46 changes: 43 additions & 3 deletions src/exec/runner.ts
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,13 @@ import { ID_PREFIX, LOG_NAMESPACE_ROOT } from "../branding.js";
import type { ReactorEmittedEvent } from "@intx/inference";
import { setAgentSourceUnlessClosed } from "../tui/agent-source-sync.js";
import { ensureFreshInferenceSource } from "../subagent/refresh-inference-source.js";
import { getToolApprovalBudget } from "../tui/tool-execution-watchdog.js";
import {
MAX_TOOL_APPROVAL_PAUSE_MS,
getToolApprovalBudget,
} from "../tui/tool-execution-watchdog.js";
import { createApprovalDeliverer } from "../tui/approval-delivery.js";
import { createCorrelationAcceptance } from "../tui/correlation-acceptance.js";
import { APPROVAL_TIMEOUT_RESULT_TEXT } from "../permission/decline-markers.js";
import { WorkflowHost } from "../workflows/host.js";

const logger = getLogger([LOG_NAMESPACE_ROOT, "exec"]);
Expand Down Expand Up @@ -968,10 +974,20 @@ export async function runExec(config: Config): Promise<ExecResult> {
// Consume-once gate for the compaction continuation emit: a replayed
// duplicate of an already-answered emission must not re-deliver.
const continuationGate = createContinuationGate();
// Approval-delivery parity with the TUI: the resume routes its deliver
// through the shared acceptance-tracked deliverer, and the stream sink
// below feeds acceptances so a delivered decision settles instead of
// timing out.
const approvalAcceptance = createCorrelationAcceptance();
const approvalDeliverer = createApprovalDeliverer({
deliverToAgent: (message) => activeAgent.deliver(message),
acceptance: approvalAcceptance,
});
// Cycles persist to the context store only on inference.done; the recorder
// keeps the in-flight cycle's text so an errored or aborted turn leaves
// its partial output in partial.jsonl instead of vanishing.
const sink = (event: ReactorEmittedEvent): void => {
approvalAcceptance.observe(event);
// Chat-director reactor events (replacing the former onTasksChange /
// onActivateTools closures). Exec mode has no live task panel or task
// stdout output today (unlike the TUI's chrome zone) — debug logging
Expand Down Expand Up @@ -1062,6 +1078,10 @@ export async function runExec(config: Config): Promise<ExecResult> {
resolveParkedCallId: (correlationId) =>
resolveParkedCallIdFromStore(activeStorage, correlationId),
gate: permissionGate,
deliver: async (message, stillCurrent) => {
if (!stillCurrent()) return;
await approvalDeliverer.deliver(message);
},
}).handle(sendResult);
sendCompleted = true;
runError = runSink.getRunError();
Expand Down Expand Up @@ -1307,9 +1327,28 @@ async function promptPermission(
});
stderr.write(` [${scopes.length + 1}] Deny\n`);
const rl = createInterface({ input, output: stderr });
let timer: ReturnType<typeof setTimeout> | undefined;
try {
const line = (await rl.question("Choice: ")).trim();
const n = Number(line);
// The reactor bounds an approval suspension (default 1h); the prompt must
// settle below that so the gate answers instead of the suspension timing
// out. The deadline reuses the tool-budget approval-pause ceiling.
const answer = rl.question("Choice: ");
const expired = new Promise<undefined>((resolve) => {
timer = setTimeout(() => resolve(undefined), MAX_TOOL_APPROVAL_PAUSE_MS);
});
const line = await Promise.race([answer, expired]);
if (line === undefined) {
// The deadline won: rl.close() in the finally below does not settle a
// pending question, so abandon the loser explicitly. It can never
// resolve after close; the attached settlement keeps it contained (no
// floating promise) if a future runtime ever settles it.
void answer.then(
() => undefined,
() => undefined,
);
return { allow: false, message: APPROVAL_TIMEOUT_RESULT_TEXT };
}
const n = Number(line.trim());
if (!Number.isInteger(n) || n < 1 || n > scopes.length) {
return { allow: false };
}
Expand All @@ -1322,6 +1361,7 @@ async function promptPermission(
...(chosen.pattern !== null ? { persist: chosen } : {}),
};
} finally {
if (timer !== undefined) clearTimeout(timer);
rl.close();
if (pauseToken !== undefined) budget?.resume(pauseToken);
}
Expand Down
Loading
Loading