Skip to content

Commit 3eff6f9

Browse files
TheGreatAxiosTheGreatAxios
authored
Merge pull request #997 from corbitsdev/cl-7793-a-parked-approval-is-abandoned-when-a-different-parallel
Abandon a parked approval only on its own timeout
2 parents a93a724 + a8ee35b commit 3eff6f9

4 files changed

Lines changed: 320 additions & 18 deletions

File tree

‎src/exec/runner.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -976,6 +976,8 @@ export async function runExec(config: Config): Promise<ExecResult> {
976976
const sendResult = await activeAgent.send(operatorTaskMessage(task));
977977
// A suspension must not park silently in exec: the approval resume owns
978978
// the terminal prompt flow and delivers the decision to the reactor.
979+
// No resolveParkedCallId: the vendored reactor exposes no
980+
// correlationId-to-call lookup, so the history heuristic is the path.
979981
await createApprovalResume({
980982
getAgent: () => activeAgent,
981983
gate: permissionGate,

‎src/session/approval-resume.test.ts‎

Lines changed: 212 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,212 @@
1+
import { describe, expect, test } from "bun:test";
2+
import type { Agent, SendResult } from "@intx/agent";
3+
import type {
4+
ApprovalSnapshot,
5+
ConversationTurn,
6+
InboundMessage,
7+
} from "@intx/types/runtime";
8+
9+
import { APPROVAL_TIMEOUT_RESULT_TEXT } from "../permission/decline-markers.js";
10+
import type { PermissionGate } from "../permission/gate.js";
11+
import { createApprovalResume } from "./approval-resume.js";
12+
13+
function assistantTurn(
14+
calls: { id: string; name: string; command: string }[],
15+
): ConversationTurn {
16+
return {
17+
role: "assistant",
18+
content: calls.map((call) => ({
19+
type: "tool_call" as const,
20+
id: call.id,
21+
name: call.name,
22+
arguments: { command: call.command },
23+
})),
24+
timestamp: 1,
25+
};
26+
}
27+
28+
function timeoutTurn(callId: string): ConversationTurn {
29+
return {
30+
role: "user",
31+
content: [
32+
{
33+
type: "tool_result" as const,
34+
callId,
35+
content: [
36+
{ type: "text" as const, text: APPROVAL_TIMEOUT_RESULT_TEXT },
37+
],
38+
isError: true,
39+
},
40+
],
41+
timestamp: 2,
42+
};
43+
}
44+
45+
function suspension(
46+
correlationId: string,
47+
command: string,
48+
): Extract<SendResult, { type: "suspended" }> {
49+
const snapshot: ApprovalSnapshot = {
50+
name: "run_shell",
51+
description: "run a shell command",
52+
inputSchema: {},
53+
arguments: { command },
54+
};
55+
return { type: "suspended", correlationId, approvalSnapshot: snapshot };
56+
}
57+
58+
function setup(args: {
59+
preTurns: ConversationTurn[];
60+
onGate: (turns: ConversationTurn[]) => void;
61+
resolveParkedCallId?: (correlationId: string) => string | undefined;
62+
}) {
63+
const turns: ConversationTurn[] = [...args.preTurns];
64+
const delivered: InboundMessage[] = [];
65+
const agent = {
66+
history: async () => turns,
67+
deliver: (message: InboundMessage) => {
68+
delivered.push(message);
69+
},
70+
};
71+
const gate = {
72+
resolveSuspended: async () => {
73+
args.onGate(turns);
74+
return { allow: true };
75+
},
76+
} as unknown as PermissionGate;
77+
const resume = createApprovalResume({
78+
getAgent: () => agent as Pick<Agent, "deliver" | "history">,
79+
gate,
80+
...(args.resolveParkedCallId !== undefined
81+
? { resolveParkedCallId: args.resolveParkedCallId }
82+
: {}),
83+
});
84+
return { resume, delivered };
85+
}
86+
87+
function decisionBody(message: InboundMessage): { outcome: string } {
88+
if (message.content === undefined)
89+
throw new Error("expected a decision body");
90+
return JSON.parse(message.content) as { outcome: string };
91+
}
92+
93+
function deliveredCorrelationId(message: InboundMessage): string {
94+
const correlationId = message.headers.interchangeCorrelationId;
95+
if (correlationId === undefined)
96+
throw new Error("expected an interchange correlation id");
97+
return correlationId;
98+
}
99+
100+
describe("approval-resume parallel-parked approvals", () => {
101+
test("delivers A's decision when a different parked call's approval times out", async () => {
102+
const { resume, delivered } = setup({
103+
preTurns: [
104+
assistantTurn([
105+
{ id: "call-A", name: "run_shell", command: "echo alpha" },
106+
{ id: "call-B", name: "run_shell", command: "echo bravo" },
107+
]),
108+
],
109+
onGate: (turns) => {
110+
turns.push(timeoutTurn("call-B"));
111+
},
112+
});
113+
114+
const handled = await resume.handle(suspension("corr-A", "echo alpha"));
115+
116+
expect(handled).toBe(true);
117+
expect(delivered).toHaveLength(1);
118+
const message = delivered[0];
119+
if (message === undefined) throw new Error("expected a delivered decision");
120+
expect(deliveredCorrelationId(message)).toBe("corr-A");
121+
expect(decisionBody(message).outcome).toBe("approved");
122+
});
123+
124+
test("still drops a genuinely late decision for the same parked call", async () => {
125+
const { resume, delivered } = setup({
126+
preTurns: [
127+
assistantTurn([
128+
{ id: "call-A", name: "run_shell", command: "echo alpha" },
129+
{ id: "call-B", name: "run_shell", command: "echo bravo" },
130+
]),
131+
],
132+
onGate: (turns) => {
133+
turns.push(timeoutTurn("call-A"));
134+
},
135+
});
136+
137+
const handled = await resume.handle(suspension("corr-A", "echo alpha"));
138+
139+
expect(handled).toBe(true);
140+
expect(delivered).toHaveLength(0);
141+
});
142+
143+
test("pending-operation lookup identifies the parked call without history tool calls", async () => {
144+
const { resume, delivered } = setup({
145+
preTurns: [
146+
{
147+
role: "user",
148+
content: [{ type: "text" as const, text: "run two shell commands" }],
149+
timestamp: 1,
150+
},
151+
],
152+
onGate: (turns) => {
153+
turns.push(timeoutTurn("call-B"));
154+
},
155+
resolveParkedCallId: (correlationId) =>
156+
correlationId === "corr-A" ? "call-A" : undefined,
157+
});
158+
159+
const handled = await resume.handle(suspension("corr-A", "echo alpha"));
160+
161+
expect(handled).toBe(true);
162+
expect(delivered).toHaveLength(1);
163+
const message = delivered[0];
164+
if (message === undefined) throw new Error("expected a delivered decision");
165+
expect(deliveredCorrelationId(message)).toBe("corr-A");
166+
expect(decisionBody(message).outcome).toBe("approved");
167+
});
168+
169+
test("identical name+args twin: sibling timeout still delivers", async () => {
170+
const { resume, delivered } = setup({
171+
preTurns: [
172+
assistantTurn([
173+
{ id: "call-A", name: "run_shell", command: "echo same" },
174+
{ id: "call-B", name: "run_shell", command: "echo same" },
175+
]),
176+
],
177+
onGate: (turns) => {
178+
turns.push(timeoutTurn("call-B"));
179+
},
180+
});
181+
182+
const handled = await resume.handle(suspension("corr-A", "echo same"));
183+
184+
expect(handled).toBe(true);
185+
expect(delivered).toHaveLength(1);
186+
});
187+
188+
// Known limitation of the exact-one history derivation: with identical
189+
// name+args twins, the answered own call drops out of the candidates and
190+
// the derivation resolves to the unanswered twin, so the settled check
191+
// misses and the genuinely-late decision is delivered. Telling the twins
192+
// apart needs the resolveParkedCallId lookup; this test locks the current
193+
// shape so a future fix can flip it to a drop.
194+
test("identical name+args twin: own timeout with unanswered twin delivers", async () => {
195+
const { resume, delivered } = setup({
196+
preTurns: [
197+
assistantTurn([
198+
{ id: "call-A", name: "run_shell", command: "echo same" },
199+
{ id: "call-B", name: "run_shell", command: "echo same" },
200+
]),
201+
],
202+
onGate: (turns) => {
203+
turns.push(timeoutTurn("call-A"));
204+
},
205+
});
206+
207+
const handled = await resume.handle(suspension("corr-A", "echo same"));
208+
209+
expect(handled).toBe(true);
210+
expect(delivered).toHaveLength(1);
211+
});
212+
});

‎src/session/approval-resume.ts‎

Lines changed: 104 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -71,23 +71,85 @@ export function requestFromApprovalSnapshot(
7171
// The reactor's approval timeout answers the parked call with this exact
7272
// upstream text (see permission/decline-markers.ts) before removing the
7373
// correlation, so its presence after the suspension watermark marks the
74-
// correlation as settled.
74+
// correlation as settled — but only when the result answers this very call.
75+
// Parallel-parked calls each time out into their own tool result (callId is
76+
// the original tool-call id, not the minted correlationId), so matching text
77+
// alone abandons a still-valid sibling decision. The parked call id must come
78+
// along and match block.callId; without it the text-only scan stays as the
79+
// fallback so a genuinely late decision is still dropped.
7580

76-
function settledAfterSuspend(
81+
function timeoutResultAfterSuspend(
7782
turns: Awaited<ReturnType<Agent["history"]>>,
7883
fromIndex: number,
79-
): boolean {
80-
return turns
81-
.slice(fromIndex)
82-
.flatMap((turn) => turn.content)
83-
.some(
84-
(block) =>
85-
block.type === "tool_result" &&
86-
block.content.some(
87-
(part) =>
88-
part.type === "text" && part.text === APPROVAL_TIMEOUT_RESULT_TEXT,
89-
),
90-
);
84+
parkedCallId: string | undefined,
85+
): string | undefined {
86+
for (const block of turns.slice(fromIndex).flatMap((turn) => turn.content)) {
87+
if (
88+
block.type === "tool_result" &&
89+
(parkedCallId === undefined || block.callId === parkedCallId) &&
90+
block.content.some(
91+
(part) =>
92+
part.type === "text" && part.text === APPROVAL_TIMEOUT_RESULT_TEXT,
93+
)
94+
) {
95+
return block.callId;
96+
}
97+
}
98+
return undefined;
99+
}
100+
101+
function stableArgs(value: unknown): string {
102+
if (Array.isArray(value)) return `[${value.map(stableArgs).join(",")}]`;
103+
if (typeof value === "object" && value !== null) {
104+
const record = value as Record<string, unknown>;
105+
return `{${Object.keys(record)
106+
.sort()
107+
.map((key) => `${JSON.stringify(key)}:${stableArgs(record[key])}`)
108+
.join(",")}}`;
109+
}
110+
return JSON.stringify(value) ?? "null";
111+
}
112+
113+
// Derive this suspension's parked call id from history: the pre-watermark
114+
// tool_call matching the snapshot's name and arguments that has no tool
115+
// result anywhere yet. A parked call is neither run nor answered, so an
116+
// unanswered match is normally this suspension's own call; a timed-out
117+
// sibling is answered by its timeout result and drops out of the candidates.
118+
// Returns undefined unless exactly one candidate matches.
119+
//
120+
// Known limitation: identical name+args twins are ambiguous. When this
121+
// suspension's own timeout fires while an identical twin sits unanswered,
122+
// the twin is the exact-one survivor, so the settled check misses and the
123+
// late decision is delivered instead of dropped. The sibling-timeout mirror
124+
// (identical args, sibling answered) still resolves to the live call and
125+
// delivers. Telling identical twins apart needs the resolveParkedCallId
126+
// lookup below; the history heuristic cannot do it.
127+
function parkedCallIdFromHistory(
128+
turns: Awaited<ReturnType<Agent["history"]>>,
129+
fromIndex: number,
130+
snapshot: ApprovalSnapshot,
131+
): string | undefined {
132+
const answered = new Set<string>();
133+
for (const turn of turns) {
134+
for (const block of turn.content) {
135+
if (block.type === "tool_result") answered.add(block.callId);
136+
}
137+
}
138+
const wanted = stableArgs(snapshot.arguments ?? {});
139+
const candidates = new Set<string>();
140+
for (const turn of turns.slice(0, fromIndex)) {
141+
for (const block of turn.content) {
142+
if (
143+
block.type === "tool_call" &&
144+
block.name === snapshot.name &&
145+
stableArgs(block.arguments) === wanted &&
146+
!answered.has(block.id)
147+
) {
148+
candidates.add(block.id);
149+
}
150+
}
151+
}
152+
return candidates.size === 1 ? [...candidates][0] : undefined;
91153
}
92154

93155
function decisionMessage(
@@ -134,6 +196,21 @@ export function createApprovalResume(args: {
134196
// TUI: interrupt/clear bump this to reject the parked call on the old
135197
// agent before close/rebuild. Cleared when handle returns.
136198
registerParkedCancel?: (cancel: (() => void) | undefined) => void;
199+
// Pending-operation lookup: map this suspension's correlationId to the
200+
// parked tool-call id (PendingOperation.suspendedCall.id). The timeout
201+
// result carries the original call id while the suspension carries only
202+
// the minted correlationId, so this is what ties them together. Takes
203+
// precedence over the history derivation below; absent callers fall back
204+
// to it.
205+
//
206+
// Heuristic-only in production: neither the exec nor the TUI caller wires
207+
// this, because the vendored reactor surface ({ start, deliver, abort })
208+
// exposes no correlationId-to-call lookup (see
209+
// permission/decline-markers.ts), and the suspension snapshot carries
210+
// name+arguments without the call id. Wire this if upstream ever exports
211+
// the pending-operation lookup; until then the history derivation is the
212+
// live path.
213+
resolveParkedCallId?: (correlationId: string) => string | undefined;
137214
gate: PermissionGate;
138215
}): ApprovalResume {
139216
const { getAgent, gate } = args;
@@ -226,13 +303,22 @@ export function createApprovalResume(args: {
226303
return true;
227304
}
228305
args.registerParkedCancel?.(undefined);
229-
if (
230-
settledAfterSuspend(await requireAgent().history(), turnsAtSuspend)
231-
) {
306+
const history = await requireAgent().history();
307+
const parkedCallId =
308+
args.resolveParkedCallId?.(correlationId) ??
309+
parkedCallIdFromHistory(history, turnsAtSuspend, approvalSnapshot);
310+
const matchedTimeoutCallId = timeoutResultAfterSuspend(
311+
history,
312+
turnsAtSuspend,
313+
parkedCallId,
314+
);
315+
if (matchedTimeoutCallId !== undefined) {
232316
// The reactor already answered the parked call (its approval timeout
233317
// fired while the surface was still up). Delivering now would append
234318
// the raw decision JSON as an uncorrelated user turn — drop and log.
235-
logger.warn`late approval decision dropped correlation=${correlationId} outcome=${outcome?.allow === true ? "approved" : "rejected"}`;
319+
// The matched call id rides along so a fallback drop (parkedCallId
320+
// undefined) can still be attributed to the timeout that caused it.
321+
logger.warn`late approval decision dropped correlation=${correlationId} timeoutCall=${matchedTimeoutCallId} outcome=${outcome?.allow === true ? "approved" : "rejected"}`;
236322
return true;
237323
}
238324
if (outcome === undefined || !outcome.allow) {

‎src/tui/runner/session.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -495,6 +495,8 @@ export async function assembleTUISession(
495495
// Reload, interrupt, compaction continuation, and proxy deliver share one queue
496496
// so a rebuild never races an in-flight deliver.
497497
const sessionOps = createSessionOperationQueue();
498+
// No resolveParkedCallId: the vendored reactor exposes no
499+
// correlationId-to-call lookup, so the history heuristic is the path.
498500
const approvalResume = createApprovalResume({
499501
getAgent: () => state.currentAgent,
500502
captureGeneration: deliveryGeneration.capture,

0 commit comments

Comments
 (0)