Skip to content

Commit f9cbd8e

Browse files
TheGreatAxiosTheGreatAxios
committed
Fail fast when an approval delivery is never accepted
A correlated approval decision waited for reactor acceptance with no deadline, so one delivery with no observed stream event wedged the serial sessionOps tail and every later approval timed out behind it. The wait now races a 30s deadline that abandons the waiter and throws diagnostics naming the correlation, stage, outcome, and that the action may still have been applied; retries never re-deliver a handed-over decision.
1 parent bf483f8 commit f9cbd8e

5 files changed

Lines changed: 390 additions & 13 deletions

File tree

‎src/tui/approval-delivery.test.ts‎

Lines changed: 187 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,187 @@
1+
import { describe, expect, test } from "bun:test";
2+
import type { InboundMessage } from "@intx/types/runtime";
3+
4+
import { createSessionOperationQueue } from "./delivery-queue.js";
5+
import {
6+
ApprovalDeliveryTimeoutError,
7+
createApprovalDeliverer,
8+
} from "./approval-delivery.js";
9+
10+
function decisionMessage(
11+
correlationId: string | undefined,
12+
outcome: "approved" | "rejected",
13+
): InboundMessage {
14+
return {
15+
ref: { uid: 0, mailbox: "approval" },
16+
headers: {
17+
from: "approval@local",
18+
to: ["agent@local"],
19+
date: new Date().toISOString(),
20+
messageId: `approval-${correlationId ?? "uncorrelated"}`,
21+
...(correlationId !== undefined
22+
? { interchangeCorrelationId: correlationId }
23+
: {}),
24+
},
25+
flags: [],
26+
content: JSON.stringify({ outcome }),
27+
signatureStatus: "missing",
28+
} satisfies InboundMessage;
29+
}
30+
31+
function controllableAcceptance() {
32+
const waiters = new Map<string, () => void>();
33+
return {
34+
wait(correlationId: string): Promise<void> {
35+
return new Promise<void>((resolve) => {
36+
waiters.set(correlationId, () => {
37+
waiters.delete(correlationId);
38+
resolve();
39+
});
40+
});
41+
},
42+
settle(correlationId: string): void {
43+
waiters.get(correlationId)?.();
44+
},
45+
abandon(correlationId: string): void {
46+
waiters.delete(correlationId);
47+
},
48+
pendingCount(): number {
49+
return waiters.size;
50+
},
51+
};
52+
}
53+
54+
describe("approval delivery acceptance bound", () => {
55+
test("a stuck acceptance wait fails fast and the sessionOps tail advances", async () => {
56+
const acceptance = controllableAcceptance();
57+
const delivered: InboundMessage[] = [];
58+
const deliverer = createApprovalDeliverer({
59+
deliverToAgent: (message) => {
60+
delivered.push(message);
61+
},
62+
acceptance,
63+
timeoutMs: 15,
64+
});
65+
const sessionOps = createSessionOperationQueue();
66+
const started = Date.now();
67+
68+
const stuck = sessionOps.enqueue(() =>
69+
deliverer.deliver(decisionMessage("corr-stuck", "approved")),
70+
);
71+
let tailAdvanced = false;
72+
const next = sessionOps.enqueue(async () => {
73+
tailAdvanced = true;
74+
});
75+
76+
const failure = await stuck.then(
77+
() => null,
78+
(err: unknown) => err,
79+
);
80+
await next;
81+
82+
expect(failure).toBeInstanceOf(ApprovalDeliveryTimeoutError);
83+
const err = failure as ApprovalDeliveryTimeoutError;
84+
expect(err.correlationId).toBe("corr-stuck");
85+
expect(err.stage).toBe("reactor-acceptance");
86+
expect(err.timeoutMs).toBe(15);
87+
expect(err.mayStillApply).toBe(true);
88+
expect(err.message).toContain("corr-stuck");
89+
expect(err.message).toContain("reactor-acceptance");
90+
expect(err.message).toContain("may still");
91+
expect(tailAdvanced).toBe(true);
92+
expect(Date.now() - started).toBeLessThan(5000);
93+
expect(delivered).toHaveLength(1);
94+
expect(acceptance.pendingCount()).toBe(0);
95+
});
96+
97+
test("retry after an acceptance timeout does not re-deliver a handed-over decision", async () => {
98+
const acceptance = controllableAcceptance();
99+
const delivered: InboundMessage[] = [];
100+
const deliverer = createApprovalDeliverer({
101+
deliverToAgent: (message) => {
102+
delivered.push(message);
103+
},
104+
acceptance,
105+
timeoutMs: 10,
106+
});
107+
const message = decisionMessage("corr-retry", "approved");
108+
109+
await expect(deliverer.deliver(message)).rejects.toBeInstanceOf(
110+
ApprovalDeliveryTimeoutError,
111+
);
112+
expect(delivered).toHaveLength(1);
113+
114+
const retry = deliverer.deliver(message);
115+
acceptance.settle("corr-retry");
116+
await retry;
117+
118+
expect(delivered).toHaveLength(1);
119+
});
120+
121+
test("retry after a deliver throw re-delivers: recovery from a failed send", async () => {
122+
const acceptance = controllableAcceptance();
123+
const delivered: InboundMessage[] = [];
124+
let failDeliver = true;
125+
const deliverer = createApprovalDeliverer({
126+
deliverToAgent: (message) => {
127+
if (failDeliver) throw new Error("agent is done");
128+
delivered.push(message);
129+
},
130+
acceptance,
131+
timeoutMs: 50,
132+
});
133+
const message = decisionMessage("corr-failed-send", "rejected");
134+
135+
await expect(deliverer.deliver(message)).rejects.toThrow("agent is done");
136+
expect(delivered).toHaveLength(0);
137+
138+
failDeliver = false;
139+
const retry = deliverer.deliver(message);
140+
acceptance.settle("corr-failed-send");
141+
await retry;
142+
143+
expect(delivered).toHaveLength(1);
144+
});
145+
146+
test("an uncorrelated decision delivers without waiting for acceptance", async () => {
147+
const acceptance = controllableAcceptance();
148+
const delivered: InboundMessage[] = [];
149+
const deliverer = createApprovalDeliverer({
150+
deliverToAgent: (message) => {
151+
delivered.push(message);
152+
},
153+
acceptance,
154+
timeoutMs: 10,
155+
});
156+
157+
await deliverer.deliver(decisionMessage(undefined, "rejected"));
158+
159+
expect(delivered).toHaveLength(1);
160+
expect(acceptance.pendingCount()).toBe(0);
161+
});
162+
163+
test("timeout diagnostics name the rejected outcome and its uncertainty", async () => {
164+
const acceptance = controllableAcceptance();
165+
const delivered: InboundMessage[] = [];
166+
const deliverer = createApprovalDeliverer({
167+
deliverToAgent: (message) => {
168+
delivered.push(message);
169+
},
170+
acceptance,
171+
timeoutMs: 10,
172+
});
173+
174+
const failure = await deliverer
175+
.deliver(decisionMessage("corr-reject", "rejected"))
176+
.then(
177+
() => null,
178+
(err: unknown) => err,
179+
);
180+
181+
expect(failure).toBeInstanceOf(ApprovalDeliveryTimeoutError);
182+
const err = failure as ApprovalDeliveryTimeoutError;
183+
expect(err.outcome).toBe("rejected");
184+
expect(err.mayStillApply).toBe(true);
185+
expect(err.message).toContain("rejected");
186+
});
187+
});

‎src/tui/approval-delivery.ts‎

Lines changed: 167 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,167 @@
1+
/**
2+
* Bounded deliver-and-await-acceptance for approval decisions.
3+
*
4+
* The reactor accepts a correlated decision asynchronously after deliver()
5+
* returns, so the sessionOps tail waits for the correlation-acceptance signal.
6+
* That wait was unbounded: any delivery that produces no observed stream event
7+
* (reactor deliver() silently drops when done, an approved hold with no
8+
* tool.start, a missed correlation event) wedged the serial tail forever, and
9+
* every later approval — send_input answers, interrupt_agent releases,
10+
* ask_operator escalations — queued behind it until the reactor approval
11+
* timeout. The vendored reactor surface ({ start, deliver, abort }) exposes no
12+
* liveness query, so absent acceptance is treated as delivery failure: bound
13+
* the wait with a deadline race that settles the waiter, logs, and lets the
14+
* tail advance.
15+
*
16+
* Retry safety: once deliver() has returned, the reactor may hold the decision
17+
* even though no acceptance was observed. A retry for the same correlationId
18+
* must not hand the decision over twice (a duplicate approved decision could
19+
* re-dispatch the parked call), so it re-awaits acceptance only. A retry after
20+
* a deliver() throw re-delivers, because nothing was handed over.
21+
*/
22+
23+
import type { InboundMessage } from "@intx/types/runtime";
24+
25+
import { getLogger } from "@intx/log";
26+
27+
import { LOG_NAMESPACE_ROOT } from "../branding.js";
28+
29+
const logger = getLogger([LOG_NAMESPACE_ROOT, "approval-delivery"]);
30+
31+
/**
32+
* Deadline for the reactor to observably accept a delivered approval decision.
33+
* Far above normal tick latency (acceptance usually lands within a reactor
34+
* tick, including the approved hold until tool.start) and far below the
35+
* vendored reactor approval timeout, so one stuck delivery fails fast instead
36+
* of wedging the tail.
37+
*/
38+
export const APPROVAL_ACCEPTANCE_TIMEOUT_MS = 30_000;
39+
40+
export type ApprovalDecisionOutcome = "approved" | "rejected" | "unknown";
41+
42+
export class ApprovalDeliveryTimeoutError extends Error {
43+
readonly correlationId: string;
44+
readonly stage = "reactor-acceptance";
45+
readonly timeoutMs: number;
46+
readonly outcome: ApprovalDecisionOutcome;
47+
/** deliver() returned, so the reactor may still act on the decision. */
48+
readonly mayStillApply = true;
49+
50+
constructor(args: {
51+
correlationId: string;
52+
timeoutMs: number;
53+
outcome: ApprovalDecisionOutcome;
54+
}) {
55+
const action =
56+
args.outcome === "approved"
57+
? "the approved action may still run"
58+
: args.outcome === "rejected"
59+
? "the parked call may still be answered by this decision"
60+
: "the decision may still be applied";
61+
super(
62+
`approval delivery timed out waiting for reactor acceptance ` +
63+
`(correlationId=${args.correlationId}, stage=reactor-acceptance, ` +
64+
`timeoutMs=${args.timeoutMs}, outcome=${args.outcome}). ` +
65+
`The decision was handed to the reactor but no acceptance event was ` +
66+
`observed, so ${action}. Do not re-deliver: the delivery tail has ` +
67+
`advanced. Check run state, then use interrupt_agent to release the ` +
68+
`parked worker if it is still parked.`,
69+
);
70+
this.name = "ApprovalDeliveryTimeoutError";
71+
this.correlationId = args.correlationId;
72+
this.timeoutMs = args.timeoutMs;
73+
this.outcome = args.outcome;
74+
}
75+
}
76+
77+
function decisionOutcome(message: InboundMessage): ApprovalDecisionOutcome {
78+
if (message.content === undefined) return "unknown";
79+
let raw: unknown;
80+
try {
81+
raw = JSON.parse(message.content);
82+
} catch {
83+
return "unknown";
84+
}
85+
if (raw === null || typeof raw !== "object") return "unknown";
86+
const outcome = (raw as { outcome?: unknown }).outcome;
87+
return outcome === "approved" || outcome === "rejected" ? outcome : "unknown";
88+
}
89+
90+
export interface ApprovalAcceptance {
91+
wait(correlationId: string): Promise<void>;
92+
settle(correlationId: string): void;
93+
abandon(correlationId: string): void;
94+
}
95+
96+
export interface ApprovalDelivererDeps {
97+
deliverToAgent: (message: InboundMessage) => void;
98+
acceptance: ApprovalAcceptance;
99+
timeoutMs?: number;
100+
onTimeout?: (err: ApprovalDeliveryTimeoutError) => void;
101+
}
102+
103+
export interface ApprovalDeliverer {
104+
deliver: (message: InboundMessage) => Promise<void>;
105+
}
106+
107+
export function createApprovalDeliverer(
108+
deps: ApprovalDelivererDeps,
109+
): ApprovalDeliverer {
110+
const timeoutMs = deps.timeoutMs ?? APPROVAL_ACCEPTANCE_TIMEOUT_MS;
111+
const delivered = new Set<string>();
112+
113+
const awaitAcceptance = async (
114+
correlationId: string,
115+
accepted: Promise<void>,
116+
outcome: ApprovalDecisionOutcome,
117+
): Promise<void> => {
118+
let timer: ReturnType<typeof setTimeout> | undefined;
119+
try {
120+
await Promise.race([
121+
accepted,
122+
new Promise<never>((_, reject) => {
123+
timer = setTimeout(() => {
124+
deps.acceptance.abandon(correlationId);
125+
const err = new ApprovalDeliveryTimeoutError({
126+
correlationId,
127+
timeoutMs,
128+
outcome,
129+
});
130+
logger.warn`approval acceptance timed out correlation=${correlationId} timeoutMs=${timeoutMs} outcome=${outcome}`;
131+
deps.onTimeout?.(err);
132+
reject(err);
133+
}, timeoutMs);
134+
}),
135+
]);
136+
} finally {
137+
clearTimeout(timer);
138+
}
139+
};
140+
141+
return {
142+
deliver: async (message) => {
143+
const correlationId = message.headers.interchangeCorrelationId;
144+
if (correlationId === undefined) {
145+
deps.deliverToAgent(message);
146+
return;
147+
}
148+
if (delivered.has(correlationId)) {
149+
await awaitAcceptance(
150+
correlationId,
151+
deps.acceptance.wait(correlationId),
152+
decisionOutcome(message),
153+
);
154+
return;
155+
}
156+
const accepted = deps.acceptance.wait(correlationId);
157+
try {
158+
deps.deliverToAgent(message);
159+
} catch (err) {
160+
deps.acceptance.settle(correlationId);
161+
throw err;
162+
}
163+
delivered.add(correlationId);
164+
await awaitAcceptance(correlationId, accepted, decisionOutcome(message));
165+
},
166+
};
167+
}

‎src/tui/correlation-acceptance.test.ts‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,22 @@ describe("createCorrelationAcceptance", () => {
3737
acceptance.settle("missing");
3838
});
3939

40+
test("abandon drops the waiter without resolving it", async () => {
41+
const acceptance = createCorrelationAcceptance();
42+
const pending = acceptance.wait("corr-1");
43+
let settled = false;
44+
void pending.then(() => {
45+
settled = true;
46+
});
47+
acceptance.abandon("corr-1");
48+
await Promise.resolve();
49+
await new Promise((resolve) => setTimeout(resolve, 5));
50+
expect(settled).toBe(false);
51+
const retry = acceptance.wait("corr-1");
52+
acceptance.settle("corr-1");
53+
await retry;
54+
});
55+
4056
test("an approved correlation does not settle until tool.start", async () => {
4157
const acceptance = createCorrelationAcceptance();
4258
const pending = acceptance.wait("corr-1");

0 commit comments

Comments
 (0)