test(managed-agent): gate G3 terminal recovery faults (#15756)

* test(managed-agent): gate G3 cleanup and terminal recovery faults

Add three packaged fault arms to hosted-failover-e2e: a refused Runtime cleanup release followed by another crash, a committed terminal whose reply is lost, and a refused terminal write over a payable checkpoint. Each keeps the original Action, execution, file effect and terminal identity, and the runner now witnesses the fault on the store or Broker boundary rather than on a client-side retry.

Part of #14481.

* test(managed-agent): harden the G3 cleanup and terminal fault arms

Drop every bounded terminal retry so the terminal reply-loss crash lands on a committed-but-undelivered terminal, and require the surviving public transcript to carry the recovered answer. Document both observations. Part of #14481.

* test(managed-agent): address the G3 terminal arm review findings

Arm C1's crash barrier only on a confirmed store 200, record that pre-crash count, and state that the replacement projects the committed terminal without a second store commit. Snapshot fault counts before the relief Turn, keep the result-commit diagnostic for after-result, and fail loudly on an unhandled crash barrier. Remove the cleanup arm: a delayed replacement reproduces a control-plane cleanup/adoption gap, so C3 stays open as a product follow-up.

Part of #14481.
This commit is contained in:
Shaojin Wen authored and GitHub committed 2026-10-28 11:38:48 +00:00
1 parent e93baf0878
commit c3e1c9bcd6
8 files changed
+479 -33

No files matched your search

+324 -8
View File
@@ -70,15 +70,18 @@ let freeze = false;
let modelRoundFailover = false;
let frozenControlPlane = false;
let g3VerifyArgument: string | undefined;
// Step 3 A5/A5a barriers and the A4/C4 lost-reply arms layer one fault point
// on the Allow arm; every other mode answers a different question and would
// make the same crash prove two things at once.
// Step 3 A5/A5a barriers, the A4/C4 lost-reply arms, and the C1-C3
// cleanup/terminal arms layer one fault point on the Allow arm; every other
// mode answers a different question and would make the same crash prove two
// things at once.
const approvalFaults = [
'after-decision',
'after-prepare',
'after-result',
'decision-reply-lost',
'admission-reply-lost',
'terminal-reply-lost',
'terminal-write-failed',
] as const;
type ApprovalFault = (typeof approvalFaults)[number];
let g3FaultArgument: string | undefined;
@@ -119,7 +122,7 @@ for (let index = 0; index < argumentsList.length; index += 1) {
index += 1;
} else {
throw new Error(
'Usage: run-managed-agent-server-e2e.ts [--model ID] [--runtime-delay-ms N] [--settings PATH] [--session-failover|--inflight-failover|--continuation-failover|--model-round-failover|--big-output] [--harness-only] [--freeze] [--frozen-control-plane] [--g3-verify allow|deny|cancel|expiry|continuation] [--g3-fault after-decision|after-prepare|decision-reply-lost|admission-reply-lost]',
'Usage: run-managed-agent-server-e2e.ts [--model ID] [--runtime-delay-ms N] [--settings PATH] [--session-failover|--inflight-failover|--continuation-failover|--model-round-failover|--big-output] [--harness-only] [--freeze] [--frozen-control-plane] [--g3-verify allow|deny|cancel|expiry|continuation] [--g3-fault after-decision|after-prepare|after-result|decision-reply-lost|admission-reply-lost|terminal-reply-lost|terminal-write-failed]',
);
}
}
@@ -754,6 +757,184 @@ function postAndDropReply(
});
}
type StoreTerminalFault = 'drop-reply' | 'reject';
type StoreFaultProxy = {
baseUrl: string;
observations: () => string[];
armTerminal: (sessionId: string, fault: StoreTerminalFault) => void;
terminalFaulted: () => boolean;
terminalRejectCount: () => number;
terminalAcceptCount: () => number;
allowTerminal: () => void;
close: () => Promise<void>;
};
// C1/C2: the Harness reaches the Session Store over the URL the control
// plane advertises. This proxy fronts that boundary so one matching
// `turn.settled` commit can lose its reply (committed, not delivered) or
// be refused before it lands (payable checkpoint, terminal missing).
function startStoreFaultProxy(targetOrigin: string): Promise<StoreFaultProxy> {
return (async () => {
let armed: { sessionId: string; fault: StoreTerminalFault } | undefined;
let terminalFaulted = false;
let terminalAllowed = false;
let terminalRejectCount = 0;
let terminalAcceptCount = 0;
let closed = false;
const observations: string[] = [];
const server: Server = createServer((request, response) => {
void (async () => {
const chunks: Buffer[] = [];
for await (const chunk of request) {
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
}
const body = Buffer.concat(chunks);
const target = new URL(request.url ?? '/', targetOrigin);
const method = request.method ?? 'GET';
const commit = target.pathname.match(
/^\/internal\/managed-session-store\/v1\/sessions\/([^/]+)\/transactions:commit(?:-v2)?$/,
);
const sessionId = commit ? decodeURIComponent(commit[1]) : undefined;
const terminal =
armed !== undefined &&
sessionId === armed.sessionId &&
commitCarriesTurnSettled(body);
if (terminal && armed?.fault === 'reject' && !terminalAllowed) {
terminalRejectCount += 1;
terminalFaulted = true;
observations.push(`${method} ${target.pathname} rejected`);
response.writeHead(500, { 'content-type': 'application/json' });
response.end('{"error":"e2e_terminal_commit_failed"}');
return;
}
const headers = new Headers();
for (const [name, value] of Object.entries(request.headers)) {
if (
value === undefined ||
[
'connection',
'content-length',
'host',
'transfer-encoding',
].includes(name)
) {
continue;
}
headers.set(name, Array.isArray(value) ? value.join(', ') : value);
}
const upstream = await fetch(target, {
method,
headers,
...(['GET', 'HEAD'].includes(method) ? {} : { body }),
});
const upstreamBody = Buffer.from(await upstream.arrayBuffer());
if (terminal) {
if (armed?.fault === 'drop-reply' && !terminalAllowed) {
// Keep dropping every bounded retry until the crash. Arm the
// barrier only on a confirmed 200: a rejected first commit did
// not land, and letting it pass would turn C1 into C2 while the
// later compensation satisfies the final accept count.
if (upstream.status === 200) {
terminalFaulted = true;
terminalAcceptCount += 1;
observations.push(
`${method} ${target.pathname} forwarded 200 dropped reply`,
);
} else {
observations.push(
`${method} ${target.pathname} forwarded ${upstream.status} dropped reply (not committed)`,
);
}
response.destroy();
return;
}
if (upstream.status === 200) terminalAcceptCount += 1;
}
observations.push(
`${method} ${target.pathname} forwarded ${upstream.status}`,
);
const responseHeaders: Record<string, string> = {};
upstream.headers.forEach((value, name) => {
if (
![
'connection',
'content-encoding',
'content-length',
'transfer-encoding',
].includes(name)
) {
responseHeaders[name] = value;
}
});
response.writeHead(upstream.status, responseHeaders);
response.end(upstreamBody);
})().catch((error: unknown) => {
if (response.destroyed) return;
if (!response.headersSent) response.writeHead(502);
response.end(String(error));
});
});
const port = await freePort();
await new Promise<void>((resolve, reject) => {
server.once('error', reject);
server.listen(port, '127.0.0.1', () => {
server.off('error', reject);
resolve();
});
});
return {
baseUrl: `http://127.0.0.1:${port}`,
observations: () => [...observations],
armTerminal: (sessionId, fault) => {
armed = { sessionId, fault };
terminalFaulted = false;
terminalAllowed = false;
observations.push(`armed ${fault} for ${sessionId}`);
},
terminalFaulted: () => terminalFaulted,
terminalRejectCount: () => terminalRejectCount,
terminalAcceptCount: () => terminalAcceptCount,
allowTerminal: () => {
terminalAllowed = true;
observations.push('terminal commits allowed');
},
close: async () => {
if (closed) return;
closed = true;
server.closeAllConnections();
await new Promise<void>((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
});
},
};
})();
}
function commitCarriesTurnSettled(body: Buffer): boolean {
try {
const parsed = JSON.parse(body.toString('utf8')) as {
recordBytesBase64?: unknown;
};
if (typeof parsed.recordBytesBase64 !== 'string') return false;
const records = Buffer.from(parsed.recordBytesBase64, 'base64').toString(
'utf8',
);
return records.split('\n').some((line) => {
if (!line.trim()) return false;
const record = JSON.parse(line) as {
subtype?: unknown;
managedSession?: { kind?: unknown };
};
return (
record.subtype === 'managed_session_event_v1' &&
record.managedSession?.kind === 'turn.settled'
);
});
} catch {
return false;
}
}
async function waitUntil(
name: string,
predicate: (remainingMs: number) => Promise<boolean> | boolean,
@@ -1196,6 +1377,7 @@ function advanceModelRound(): void {
let fake: Awaited<ReturnType<typeof startFakeOpenAIServer>> | undefined;
let heldStartProxy: HeldExecutionStartProxy | undefined;
let replacementBrokerProxy: HeldExecutionStartProxy | undefined;
let storeFaultProxy: StoreFaultProxy | undefined;
let failure: unknown;
let dumpPort: number | undefined;
// The success JSON, assembled inside the try and printed only after
@@ -1203,6 +1385,9 @@ let dumpPort: number | undefined;
// on stdout for a run the operator explicitly stopped.
let resultJson: string | undefined;
let approvalResultJson: string | undefined;
// Snapshot before the relief Turn so the summary cannot report the relief's
// store/broker traffic as part of the original fault arm.
let originalFaultObservations: Record<string, number> | undefined;
let g3SummaryJson: string | undefined;
// The freeze arm's wake record reads as a pass line too, so it defers the
// same way and prints after the main result, as in the base's ordering.
@@ -1337,6 +1522,20 @@ try {
const springPort = await freePort();
const harnessPort = await freePort();
const brokerPort = await freePort();
// C1/C2 replace the Harness's Session Store reply/write after the
// payable checkpoint, so the store URL the control plane advertises
// must point at this proxy from the first load onward. Other arms keep
// the direct URL and an unchanged path.
if (
approvalFault === 'terminal-reply-lost' ||
approvalFault === 'terminal-write-failed'
) {
storeFaultProxy = await startStoreFaultProxy(
`http://127.0.0.1:${springPort}`,
);
}
const sessionStoreUrl =
storeFaultProxy?.baseUrl ?? `http://127.0.0.1:${springPort}`;
const initialized = spawnSync(
mysqld,
@@ -1447,7 +1646,7 @@ try {
QWEN_MANAGED_AGENT_DISPATCH_LEASE_DURATION: '2s',
QWEN_MANAGED_AGENT_DISPATCH_LEASE_RENEW_INTERVAL: '500ms',
QWEN_MANAGED_AGENT_DISPATCH_SCAN_DELAY: '200ms',
QWEN_MANAGED_AGENT_SESSION_STORE_BASE_URL: `http://127.0.0.1:${springPort}`,
QWEN_MANAGED_AGENT_SESSION_STORE_BASE_URL: sessionStoreUrl,
QWEN_MANAGED_AGENT_SESSION_STORE_ENABLED: 'true',
QWEN_MANAGED_AGENT_SESSION_STORE_WRITER_LEASE_DURATION: '5s',
QWEN_MANAGED_AGENT_WORKSPACE_ID: workspaceId,
@@ -1469,7 +1668,7 @@ try {
QWEN_MANAGED_AGENT_RUNTIME_CREDENTIAL_KEY_ID: 'e2e-local-v1',
QWEN_MANAGED_AGENT_RUNTIME_STATE_DIRECTORY: runtimeState,
QWEN_MANAGED_AGENT_RUNTIME_WORKER_ENTRY: cliBundle,
QWEN_MANAGED_AGENT_SESSION_STORE_BASE_URL: `http://127.0.0.1:${springPort}`,
QWEN_MANAGED_AGENT_SESSION_STORE_BASE_URL: sessionStoreUrl,
QWEN_MANAGED_AGENT_SESSION_STORE_ENABLED: 'true',
QWEN_MANAGED_AGENT_SESSION_STORE_WRITER_LEASE_DURATION: '60s',
QWEN_MANAGED_AGENT_WORKSPACE_ID: workspaceId,
@@ -2358,6 +2557,18 @@ try {
releaseApprovalResultHold = resolve;
});
}
// C1/C2: arm the store boundary before the decision can drive the
// approved batch to its turn.settled commit. The fault must land on
// that terminal commit, not on an earlier record.
if (
approvalFault === 'terminal-reply-lost' ||
approvalFault === 'terminal-write-failed'
) {
storeFaultProxy?.armTerminal(
session.id,
approvalFault === 'terminal-reply-lost' ? 'drop-reply' : 'reject',
);
}
if (approvalFault === 'decision-reply-lost') {
await postAndDropReply(decisionUrl, decisionBody, decisionHeaders);
}
@@ -2403,10 +2614,13 @@ try {
);
}
let publishedPrefixSequences: number[] = [];
let terminalAcceptsAtBarrier = 0;
if (
approvalFault === 'after-decision' ||
approvalFault === 'after-prepare' ||
approvalFault === 'after-result'
approvalFault === 'after-result' ||
approvalFault === 'terminal-reply-lost' ||
approvalFault === 'terminal-write-failed'
) {
// A5/A5a: kill the generation that holds the committed decision (or,
// for after-result, the committed result with its model round still
@@ -2465,7 +2679,25 @@ try {
120_000,
replacementHarness,
);
} else {
} else if (
approvalFault === 'terminal-reply-lost' ||
approvalFault === 'terminal-write-failed'
) {
// C1/C2: wait until the terminal commit met the store boundary.
// C1 already committed it and lost only the reply; C2 refused the
// write, leaving the payable checkpoint without its terminal.
await waitUntil(
'Terminal commit store fault observed',
() => storeFaultProxy?.terminalFaulted() === true,
120_000,
replacementHarness,
);
// The barrier only arms on a confirmed 200, so this count proves
// the terminal was committed before the crash. The replacement
// projects it; a second store commit is not required.
terminalAcceptsAtBarrier =
storeFaultProxy?.terminalAcceptCount() ?? 0;
} else if (approvalFault === 'after-decision') {
// The 202 admitted the response operation, not the decision. Wait
// for the durable decision row and its receipt before SIGKILL.
await waitUntil(
@@ -2480,6 +2712,13 @@ try {
120_000,
replacementHarness,
);
} else {
// The enclosing condition lists every crash-type fault. A new one
// added without a barrier here must fail loudly instead of
// reaching SIGKILL with no wait at all.
throw new Error(
`Unhandled approval fault barrier: ${String(approvalFault)}`,
);
}
const executionAtCrash = runMysql(
mysqlPort,
@@ -2501,6 +2740,13 @@ try {
crashedHarnessBootId: crashedBootId,
executionAtCrash,
heldExecutionStartPath: replacementBrokerProxy?.heldPath(),
terminalFault: storeFaultProxy
? {
rejected: storeFaultProxy.terminalRejectCount(),
accepted: storeFaultProxy.terminalAcceptCount(),
acceptedAtBarrier: terminalAcceptsAtBarrier,
}
: undefined,
publishedPrefix:
approvalFault === 'after-result'
? {
@@ -2517,6 +2763,15 @@ try {
releaseApprovalResultHold();
approvalResultHold = undefined;
}
// C1/C2's doomed generation must not land or deliver the terminal after
// the injected fault; the replacement may, including a replayed commit
// for C1.
if (
approvalFault === 'terminal-reply-lost' ||
approvalFault === 'terminal-write-failed'
) {
storeFaultProxy?.allowTerminal();
}
if (replacementBrokerProxy) {
await replacementBrokerProxy.close();
replacementBrokerProxy = undefined;
@@ -2577,6 +2832,22 @@ try {
);
}
verifiedOriginalTerminal = terminals[0].type;
// A terminal is not the answer. The surviving public transcript must
// still carry the one model response the recovered Turn produced, or
// a compensation that only projects a terminal would pass. The
// result-commit arm already owns the same check with its retraction
// diagnostic, so it keeps that message.
if (verificationMode === 'allow' && approvalFault !== 'after-result') {
const transcript = await publicTranscript(
replacementSpringUrl,
tenant,
session.id,
);
if (transcript.text !== continuationResponse)
throw new Error(
`Approval recovery answer differs: ${JSON.stringify(transcript.text)}`,
);
}
if (approvalFault === 'after-result') {
// The reissue must retract the held prefix: only the recovered
// answer may remain in the public transcript.
@@ -2793,8 +3064,39 @@ try {
throw new Error(
`Original cleanup did not release the original Runtime: state=${runtimeAfter.join(',')} confirmed=${cleanupConfirmed}`,
);
if (approvalFault === 'terminal-reply-lost') {
if (
storeFaultProxy?.terminalFaulted() !== true ||
terminalAcceptsAtBarrier < 1
)
throw new Error(
`Terminal reply loss did not commit the terminal before recovery: barrierAccepts=${terminalAcceptsAtBarrier} observations=${storeFaultProxy?.observations().join(' | ')}`,
);
}
if (approvalFault === 'terminal-write-failed') {
if (
(storeFaultProxy?.terminalRejectCount() ?? 0) < 1 ||
(storeFaultProxy?.terminalAcceptCount() ?? 0) < 1
)
throw new Error(
`Terminal write fault was not compensated: reject=${storeFaultProxy?.terminalRejectCount()} accept=${storeFaultProxy?.terminalAcceptCount()} observations=${storeFaultProxy?.observations().join(' | ')}`,
);
}
const faultObservations = {
terminalRejects: storeFaultProxy?.terminalRejectCount() ?? 0,
terminalAccepts: storeFaultProxy?.terminalAcceptCount() ?? 0,
terminalAcceptsAtBarrier,
postCrashTerminalAccepts: Math.max(
0,
(storeFaultProxy?.terminalAcceptCount() ?? 0) -
terminalAcceptsAtBarrier,
),
};
originalFaultObservations = faultObservations;
saveEvidence('approval-result.json', {
mode: verificationMode,
fault: approvalFault ?? null,
faultObservations,
sessionId: session.id,
actionId: originalApprovalId,
replacementActionBeforeResponse: actionBeforeResponse,
@@ -4024,6 +4326,14 @@ try {
const summary = {
mode: verificationMode,
fault: approvalFault ?? null,
// C1/C2 must not report the later relief Turn's terminal commit as
// part of the original fault window.
faultObservations: originalFaultObservations ?? {
terminalRejects: 0,
terminalAccepts: 0,
terminalAcceptsAtBarrier: 0,
postCrashTerminalAccepts: 0,
},
sessionId: session.id,
originalRuntimeSessionId,
originalActionId: originalApprovalId,
@@ -4266,6 +4576,12 @@ try {
}
await heldStartProxy?.close();
await replacementBrokerProxy?.close();
await storeFaultProxy?.close();
// A failed G3 arm still owes the operator the model traffic that decided
// whether the recovery resumed, replayed or never inferred at all.
if (verificationMode && failure) {
saveEvidence('model-requests.json', { requests: fake?.requests ?? [] });
}
releaseContinuationHold();
for (const hold of modelRoundHolds) hold.release();
await fake?.close();
+26 -2
View File
@@ -144,6 +144,30 @@ const failoverArms = [
'admission-reply-lost',
],
],
[
'Run committed terminal reply-loss E2E',
'test:e2e:managed-harness-restart-approval-terminal-reply-lost',
[
'--continuation-failover',
'--harness-only',
'--g3-verify',
'allow',
'--g3-fault',
'terminal-reply-lost',
],
],
[
'Run missing terminal write E2E',
'test:e2e:managed-harness-restart-approval-terminal-write-failed',
[
'--continuation-failover',
'--harness-only',
'--g3-verify',
'allow',
'--g3-fault',
'terminal-write-failed',
],
],
[
'Run Harness-restart continuation closeout E2E',
'test:e2e:managed-harness-restart-continuation-closeout',
@@ -583,8 +607,8 @@ describe('Hosted real-process gates', () => {
// Step ceilings today (201 + 20 + 50 + 20): Verify, the Broker fault
// gates, admission and O4. The failover arms moved to their own job.
['hosted-harness-mysql', 291, 304],
// Twenty failover arms at 10 each.
['hosted-failover-e2e', 200, 215],
// Twenty-two failover arms at 10 each.
['hosted-failover-e2e', 220, 235],
])(
'keeps the %s job ceiling above its summed step ceilings',
(name, steps, ceiling) => {
@@ -288,6 +288,28 @@ describe('managed-agent-server e2e runner', () => {
expect(approvalDecisionCommitted('decided\t1\t1')).toBe(true);
});
it('recognizes only a turn.settled store commit as the terminal write', () => {
const { commitCarriesTurnSettled } = load(
['commitCarriesTurnSettled'],
'commitCarriesTurnSettled',
)();
const commit = (kind) =>
Buffer.from(
JSON.stringify({
recordBytesBase64: Buffer.from(
JSON.stringify({
subtype: 'managed_session_event_v1',
managedSession: { kind },
}),
).toString('base64'),
}),
);
expect(commitCarriesTurnSettled(commit('turn.settled'))).toBe(true);
expect(commitCarriesTurnSettled(commit('message.committed'))).toBe(false);
expect(commitCarriesTurnSettled(Buffer.from('{}'))).toBe(false);
expect(commitCarriesTurnSettled(Buffer.from('not json'))).toBe(false);
});
it('pins the #14481 approval fault arms to durable barriers', () => {
const source = read('scripts/run-managed-agent-server-e2e.ts');
const workflow = read('.github/workflows/sdk-java.yml');
@@ -298,6 +320,8 @@ describe('managed-agent-server e2e runner', () => {
'test:e2e:managed-harness-restart-approval-after-result',
'test:e2e:managed-harness-restart-approval-decision-reply-lost',
'test:e2e:managed-harness-restart-approval-admission-reply-lost',
'test:e2e:managed-harness-restart-approval-terminal-reply-lost',
'test:e2e:managed-harness-restart-approval-terminal-write-failed',
];
for (const script of scripts) {
expect(manifest, `${script} must be an npm script`).toContain(
@@ -377,6 +401,26 @@ describe('managed-agent-server e2e runner', () => {
expect(source.slice(proxy - 500, proxy)).not.toContain(
"approvalFault === 'after-result'",
);
// C1 arms its crash barrier only after the store confirms a 200, and
// records that pre-crash count; a non-200 first commit must not pass.
expect(source).toContain('forwarded 200 dropped reply');
expect(source).toContain('terminalAcceptsAtBarrier');
expect(source).toContain('postCrashTerminalAccepts');
expect(source).toContain('did not commit the terminal before recovery');
// C1/C2: the store proxy must be armed before the decision and must
// distinguish a lost reply (committed) from a refused write (missing).
expect(source).toContain('storeFaultProxy = await startStoreFaultProxy');
expect(source).toContain('Terminal commit store fault observed');
expect(source).toContain('storeFaultProxy?.armTerminal(');
expect(source).toContain('storeFaultProxy?.allowTerminal()');
expect(source).toContain('commitCarriesTurnSettled');
expect(source).toContain('Terminal write fault was not compensated');
// C1 must drop every bounded retry, not only the first answer, or the
// doomed generation can receive a replay before the crash.
expect(source).toContain('&& !terminalAllowed');
// A terminal alone is not the answer: the surviving transcript must
// still carry the model response.
expect(source).toContain('Approval recovery answer differs');
});
// #12941: the Stage A acceptance criterion names a 15-second Runtime delay,