Skip to content

Commit a24542c

Browse files
TheGreatAxiosTheGreatAxios
authored
Merge pull request #1069 from corbitsdev/cl-7995-worker-control-actions-time-out-while-answering-a-parked
Fail fast when an approval delivery is never accepted
2 parents bf483f8 + f9cbd8e commit a24542c

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)