Merge pull request #1078 from corbitsdev/cl-7992-implement-consolidated-keeper-test-plan-post-merge

fix(approval): settle retried approvals with denial memory

Author: TheGreatAxios

src/agent/director.test.ts

@@ -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

src/exec/runner.ts

@@ -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"]);
@@ -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
@@ -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();
@@ -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 };
     }
@@ -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);
   }