mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-11-07 06:19:43 -10:00
jinye
b1cd4ac0c6
* docs(managed-agent): document database tables and columns * refactor(managed-agent): align shared MySQL schema conventions
6025 lines
240 KiB
TypeScript
6025 lines
240 KiB
TypeScript
import { spawn, spawnSync, type ChildProcess } from 'node:child_process';
|
|
import { createHash, randomBytes, randomUUID } from 'node:crypto';
|
|
import {
|
|
existsSync,
|
|
mkdirSync,
|
|
mkdtempSync,
|
|
readFileSync,
|
|
realpathSync,
|
|
rmSync,
|
|
writeFileSync,
|
|
} from 'node:fs';
|
|
import { createServer, request as httpRequest, type Server } from 'node:http';
|
|
import { homedir, tmpdir } from 'node:os';
|
|
import path from 'node:path';
|
|
import { inspect } from 'node:util';
|
|
import {
|
|
fakeToolCall,
|
|
startFakeOpenAIServer,
|
|
} from '../integration-tests/fake-openai-server.js';
|
|
import { SseObserver } from '../integration-tests/helpers/hosted-sse-observer.js';
|
|
import { isDeepStrictEqual } from 'node:util';
|
|
|
|
const root = process.cwd();
|
|
const databaseEpochMillisSql =
|
|
'CAST(UNIX_TIMESTAMP() AS DECIMAL(20, 0)) * 1000' +
|
|
' + FLOOR(EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(6)) / 1000)';
|
|
const reliefMarker = 'G3_STEP3_RELIEF_TURN';
|
|
const reliefResponse = 'G3_STEP3_RELIEF_COMPLETED';
|
|
let originalApprovalId: string | undefined;
|
|
interface SavedApprovalOptions {
|
|
v: number;
|
|
inputRevision: number;
|
|
policyRevision: string;
|
|
createdAt: number;
|
|
expiresAt: number;
|
|
continuationRef: { resourceId: string };
|
|
}
|
|
interface SavedApprovalPlan {
|
|
calls: {
|
|
prepareKey: string;
|
|
requestDigest: string;
|
|
runtimeCallId?: string;
|
|
partIndex?: number;
|
|
actionId?: string | null;
|
|
}[];
|
|
actionId: string;
|
|
approvalOrdinal?: number;
|
|
stage: string;
|
|
batchId: string;
|
|
runtime: {
|
|
runtimeSessionId: string;
|
|
bindingId: string;
|
|
generation: string;
|
|
workspaceGeneration: string;
|
|
};
|
|
}
|
|
let originalApprovalOptions: SavedApprovalOptions | undefined;
|
|
let originalApprovalPlan: SavedApprovalPlan | undefined;
|
|
// A3: the pending second sibling's Action, captured at the crash barrier and
|
|
// re-decided by the recovery generation.
|
|
let siblingApprovalId: string | undefined;
|
|
let siblingApprovalOptions: SavedApprovalOptions | undefined;
|
|
// B3: tool B's new Action, detected after Allow A and decided by the runner's
|
|
// second Allow; both earlier Actions must end decided under their original IDs.
|
|
let secondApprovalId: string | undefined;
|
|
let verifiedOriginalTerminal: string | undefined;
|
|
function writeEvidenceFile(
|
|
directory: string,
|
|
name: string,
|
|
body: unknown,
|
|
): void {
|
|
mkdirSync(directory, { recursive: true });
|
|
writeFileSync(
|
|
path.join(directory, name),
|
|
JSON.stringify(body, null, 2) + '\n',
|
|
);
|
|
}
|
|
function saveEvidence(name: string, body: unknown): void {
|
|
if (!verificationMode) return;
|
|
writeEvidenceFile(verificationDirectory, name, body);
|
|
}
|
|
|
|
const argumentsList = process.argv.slice(2);
|
|
let model = 'moonshot/kimi-k3';
|
|
let runtimeDelayMs = 0;
|
|
let settingsPath = path.join(homedir(), '.qwen', 'settings.json');
|
|
let sessionFailover = false;
|
|
let inflightFailover = false;
|
|
let continuationFailover = false;
|
|
let bigOutput = false;
|
|
let harnessOnly = false;
|
|
let freeze = false;
|
|
let modelRoundFailover = false;
|
|
let frozenControlPlane = false;
|
|
let g3VerifyArgument: string | undefined;
|
|
// 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',
|
|
'finished-before-commit',
|
|
'sibling-approval',
|
|
'after-earlier-tools',
|
|
'cleanup-failed',
|
|
'decision-reply-lost',
|
|
'admission-reply-lost',
|
|
'load-reply-lost',
|
|
'terminal-reply-lost',
|
|
'terminal-write-failed',
|
|
] as const;
|
|
type ApprovalFault = (typeof approvalFaults)[number];
|
|
let g3FaultArgument: string | undefined;
|
|
|
|
for (let index = 0; index < argumentsList.length; index += 1) {
|
|
const argument = argumentsList[index];
|
|
const value = argumentsList[index + 1];
|
|
if (argument === '--model' && value) {
|
|
model = value;
|
|
index += 1;
|
|
} else if (argument === '--runtime-delay-ms' && value) {
|
|
runtimeDelayMs = Number(value);
|
|
index += 1;
|
|
} else if (argument === '--settings' && value) {
|
|
settingsPath = path.resolve(value);
|
|
index += 1;
|
|
} else if (argument === '--session-failover') {
|
|
sessionFailover = true;
|
|
} else if (argument === '--inflight-failover') {
|
|
inflightFailover = true;
|
|
} else if (argument === '--big-output') {
|
|
bigOutput = true;
|
|
} else if (argument === '--continuation-failover') {
|
|
continuationFailover = true;
|
|
} else if (argument === '--harness-only') {
|
|
harnessOnly = true;
|
|
} else if (argument === '--freeze') {
|
|
freeze = true;
|
|
} else if (argument === '--model-round-failover') {
|
|
modelRoundFailover = true;
|
|
} else if (argument === '--frozen-control-plane') {
|
|
frozenControlPlane = true;
|
|
} else if (argument === '--g3-verify' && value) {
|
|
g3VerifyArgument = value;
|
|
index += 1;
|
|
} else if (argument === '--g3-fault' && value) {
|
|
g3FaultArgument = value;
|
|
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|after-result|finished-before-commit|sibling-approval|after-earlier-tools|cleanup-failed|decision-reply-lost|admission-reply-lost|load-reply-lost|terminal-reply-lost|terminal-write-failed]',
|
|
);
|
|
}
|
|
}
|
|
|
|
// The flag is what CI and the npm scripts pass; the environment spelling is
|
|
// what the Step 3 report published, so it keeps working.
|
|
const verificationMode = g3VerifyArgument ?? process.env['G3_VERIFY_MODE'];
|
|
if (
|
|
verificationMode &&
|
|
!['allow', 'deny', 'cancel', 'expiry', 'continuation'].includes(
|
|
verificationMode,
|
|
)
|
|
) {
|
|
throw new Error(
|
|
'--g3-verify (or G3_VERIFY_MODE) must be allow, deny, cancel, expiry or continuation',
|
|
);
|
|
}
|
|
const approvalRecovery =
|
|
!!verificationMode && verificationMode !== 'continuation';
|
|
const approvalFault = (g3FaultArgument ?? process.env['G3_FAULT']) as
|
|
| ApprovalFault
|
|
| undefined;
|
|
if (approvalFault && !approvalFaults.includes(approvalFault)) {
|
|
throw new Error(
|
|
`--g3-fault (or G3_FAULT) must be ${approvalFaults.join(', ')}`,
|
|
);
|
|
}
|
|
if (approvalFault && verificationMode !== 'allow') {
|
|
throw new Error(
|
|
'--g3-fault requires --g3-verify allow: Deny, Cancel and expiry have no approved execution to crash into',
|
|
);
|
|
}
|
|
const verificationDirectory = path.resolve(
|
|
root,
|
|
process.env['G3_VERIFY_REPORT_DIR'] ??
|
|
`.qwen/investigations/g3-step3-implementation/packaged-${verificationMode}-${Date.now()}`,
|
|
);
|
|
// The model-round arm has no gate mode; name its evidence explicitly instead
|
|
// of borrowing a packaged-undefined-<epoch> directory (R1-10).
|
|
const modelRoundEvidenceDirectory = path.resolve(
|
|
root,
|
|
process.env['G3_VERIFY_REPORT_DIR'] ??
|
|
`.qwen/investigations/g3-step3-implementation/model-round-${Date.now()}`,
|
|
);
|
|
|
|
if (
|
|
[
|
|
sessionFailover,
|
|
inflightFailover,
|
|
continuationFailover,
|
|
modelRoundFailover,
|
|
bigOutput,
|
|
].filter(Boolean).length > 1
|
|
) {
|
|
throw new Error('The deterministic E2E modes are exclusive');
|
|
}
|
|
if (
|
|
harnessOnly &&
|
|
!(
|
|
sessionFailover ||
|
|
inflightFailover ||
|
|
continuationFailover ||
|
|
modelRoundFailover
|
|
)
|
|
) {
|
|
throw new Error('--harness-only requires one of the failover modes');
|
|
}
|
|
if (freeze && (!continuationFailover || harnessOnly)) {
|
|
throw new Error(
|
|
'--freeze requires --continuation-failover without --harness-only',
|
|
);
|
|
}
|
|
// The model-round arm replaces either the Harness alone (twice, under a live
|
|
// Spring) or a frozen, still-alive former control plane; a kill-both variant
|
|
// would only repeat --session-failover's owner swap without a parked round.
|
|
if (modelRoundFailover && harnessOnly === frozenControlPlane) {
|
|
throw new Error(
|
|
'--model-round-failover requires exactly one of --harness-only or --frozen-control-plane',
|
|
);
|
|
}
|
|
if (frozenControlPlane && !modelRoundFailover) {
|
|
throw new Error('--frozen-control-plane requires --model-round-failover');
|
|
}
|
|
|
|
if (verificationMode && (!continuationFailover || !harnessOnly || freeze)) {
|
|
throw new Error(
|
|
'This isolated Step 3 probe requires --continuation-failover --harness-only',
|
|
);
|
|
}
|
|
if (verificationMode) mkdirSync(verificationDirectory, { recursive: true });
|
|
|
|
const durableFailover =
|
|
sessionFailover ||
|
|
inflightFailover ||
|
|
continuationFailover ||
|
|
modelRoundFailover ||
|
|
bigOutput;
|
|
const runtimeTakeover = inflightFailover || continuationFailover;
|
|
const workspaceTurns = runtimeTakeover || bigOutput;
|
|
const registerStorage = process.env['QWEN_E2E_REGISTER_STORAGE'] === '1';
|
|
if (registerStorage && process.platform !== 'linux') {
|
|
throw new Error(
|
|
'QWEN_E2E_REGISTER_STORAGE=1 requires Linux native storage identity',
|
|
);
|
|
}
|
|
// The tool-driven modes take over a dead Runtime binding, which needs the
|
|
// durable local-Worker reclaim from the W0e line — Linux-only today. With
|
|
// --harness-only the Spring and its Broker stay alive, the worker is never
|
|
// orphaned, and no reclaim is needed.
|
|
if (runtimeTakeover && !harnessOnly && process.platform !== 'linux') {
|
|
throw new Error(
|
|
`${inflightFailover ? '--inflight-failover' : '--continuation-failover'} requires Linux: the replacement owner must retire the dead worker's Runtime binding through the durable local-Worker reclaim (#12380 W0e), which only runs on Linux. Run the mode in the Hosted failover E2E CI job or a Linux container.`,
|
|
);
|
|
}
|
|
// The Stage A acceptance criterion names a 15-second Runtime delay; the
|
|
// real-provider TTFT margin under it is unrecorded (tracked in #12941).
|
|
const modelBeforeRuntimeAssertionDelayMs = 15_000;
|
|
// The frozen-arm teardown wakes the former writer by this registry key, so
|
|
// the start name and the wake-up lookup must share it (a rename drift
|
|
// would leave a stopped child wedged on the runner).
|
|
const hostedHarnessLabel = 'Hosted Harness';
|
|
|
|
if (!Number.isSafeInteger(runtimeDelayMs) || runtimeDelayMs < 0) {
|
|
throw new Error('--runtime-delay-ms must be a non-negative integer');
|
|
}
|
|
|
|
const cliBundle = path.join(root, 'dist', 'cli.js');
|
|
const springModule = path.join(root, 'packages', 'managed-agent-server');
|
|
const springTarget = path.join(springModule, 'server', 'target');
|
|
// The pom attaches classified repackage executions alongside the
|
|
// unclassified server jar, so a wildcard over target/ matches several
|
|
// artifacts. Read the project version from the
|
|
// pom and name the unclassified jar exactly — version-independent without
|
|
// assuming it is the only packaged artifact. The <version> immediately
|
|
// following the module's own <artifactId> (not the <parent> block's). The
|
|
// capture itself is whitespace-tolerant, so a formatter wrapping or
|
|
// re-indenting the element cannot flow newlines into the jar path.
|
|
const pomXml = readFileSync(path.join(springModule, 'pom.xml'), 'utf8');
|
|
const pomVersion = pomXml.match(
|
|
/<artifactId>qwen-managed-agent-parent<\/artifactId>\s*<version>\s*([^<]*?\S)\s*<\/version>/,
|
|
)?.[1];
|
|
if (!pomVersion) {
|
|
throw new Error(
|
|
'Could not read the project <version> from managed-agent-server/pom.xml',
|
|
);
|
|
}
|
|
const springJar = path.join(
|
|
springTarget,
|
|
`qwen-managed-agent-server-${pomVersion}-exec.jar`,
|
|
);
|
|
for (const required of [
|
|
cliBundle,
|
|
springJar,
|
|
...(durableFailover ? [] : [settingsPath]),
|
|
]) {
|
|
if (!existsSync(required)) {
|
|
throw new Error(`Required file is missing: ${required}`);
|
|
}
|
|
}
|
|
|
|
function command(name: string): string {
|
|
const result = spawnSync('which', [name], { encoding: 'utf8' });
|
|
if (result.status !== 0) {
|
|
throw new Error(`Required command is missing: ${name}`);
|
|
}
|
|
return realpathSync(result.stdout.trim());
|
|
}
|
|
|
|
const java = command('java');
|
|
const mysqld = command('mysqld');
|
|
const mysql = command('mysql');
|
|
const mysqladmin = command('mysqladmin');
|
|
const mysqldUserArguments = process.getuid?.() === 0 ? ['--user=root'] : [];
|
|
let receivedSignal: NodeJS.Signals | undefined;
|
|
const handleSignal = (signal: NodeJS.Signals) => {
|
|
receivedSignal = signal;
|
|
};
|
|
process.on('SIGINT', handleSignal);
|
|
process.on('SIGTERM', handleSignal);
|
|
const temporary = realpathSync(
|
|
mkdtempSync(path.join(tmpdir(), 'managed-agent-server-e2e-')),
|
|
);
|
|
const workspace = path.join(temporary, 'workspace');
|
|
const harnessHome = path.join(temporary, 'harness-home');
|
|
const replacementHarnessHome = path.join(temporary, 'replacement-harness-home');
|
|
const runtimeHome = path.join(temporary, 'runtime-home');
|
|
const replacementRuntimeHome = path.join(temporary, 'replacement-runtime-home');
|
|
const runtimeState = path.join(temporary, 'runtime-state');
|
|
// The frozen control-plane arm runs two live Brokers side by side, so the
|
|
// replacement must not share the frozen one's local Runtime state.
|
|
const replacementRuntimeState = path.join(
|
|
temporary,
|
|
'replacement-runtime-state',
|
|
);
|
|
const mysqlData = path.join(temporary, 'mysql-data');
|
|
const mysqlSocket = path.join(temporary, 'mysql.sock');
|
|
const mysqlError = path.join(temporary, 'mysql-error.log');
|
|
const trustedFolders = path.join(temporary, 'trusted-folders.json');
|
|
const delayedNode = path.join(temporary, 'delayed-node');
|
|
const sideEffectName = 'managed-agent-real-e2e.txt';
|
|
const sideEffectContent = 'managed agent real model tool execution complete';
|
|
const workspaceMount = path.join(temporary, 'workspace-mount');
|
|
const sideEffect = path.join(workspaceMount, sideEffectName);
|
|
const boundWorkspaceId = 'e2e-workspace';
|
|
const boundStorageId = 'e2e-storage';
|
|
const trustedActorHeader = 'x-qwen-e2e-trusted-actor';
|
|
const trustedActor = 'e2e-actor';
|
|
const inflightSideEffectName = 'managed-inflight-side-effect.txt';
|
|
const inflightSideEffect = path.join(workspaceMount, inflightSideEffectName);
|
|
const inflightSideEffectContent = 'MANAGED_INFLIGHT_TOOL_EXECUTED\n';
|
|
// A3 mixed-sibling arm: the second sibling writes a distinct file so each
|
|
// sibling's single execution has its own physical effect.
|
|
const siblingSideEffectName = 'managed-sibling-side-effect.txt';
|
|
const siblingSideEffect = path.join(workspaceMount, siblingSideEffectName);
|
|
const siblingSideEffectContent = 'MANAGED_SIBLING_TOOL_EXECUTED\n';
|
|
// B3 after-earlier-tools arm: a second, later tool round writes its own file so
|
|
// each earlier round's single execution has its own physical effect. The +Nk
|
|
// budget directive makes the TurnBudget snapshot non-null so its restoration is
|
|
// numerically assertable (the default null budget hides a ledger reset). A round
|
|
// reset is caught by the exactly-once / deep-equal / round-continuity assertions.
|
|
const b3SecondSideEffectName = 'managed-b3-second-side-effect.txt';
|
|
const b3SecondSideEffect = path.join(workspaceMount, b3SecondSideEffectName);
|
|
const b3SecondSideEffectContent = 'MANAGED_B3_SECOND_TOOL_EXECUTED\n';
|
|
const b3BudgetDirective = '+500k';
|
|
const workspaceId = createHash('sha256')
|
|
.update(workspace)
|
|
.digest('hex')
|
|
.slice(0, 16);
|
|
try {
|
|
for (const directory of [
|
|
workspace,
|
|
path.join(harnessHome, '.qwen'),
|
|
...(durableFailover
|
|
? [
|
|
path.join(replacementHarnessHome, '.qwen'),
|
|
path.join(replacementRuntimeHome, '.qwen'),
|
|
]
|
|
: []),
|
|
workspaceMount,
|
|
path.join(runtimeHome, '.qwen'),
|
|
runtimeState,
|
|
...(frozenControlPlane ? [replacementRuntimeState] : []),
|
|
mysqlData,
|
|
]) {
|
|
// The durable local-Runtime store refuses a directory not private to its
|
|
// owner, so the state directory gets owner-only permissions.
|
|
mkdirSync(directory, {
|
|
recursive: true,
|
|
...(directory === runtimeState && workspaceTurns ? { mode: 0o700 } : {}),
|
|
});
|
|
}
|
|
if (durableFailover) {
|
|
for (const home of [harnessHome, replacementHarnessHome]) {
|
|
writeFileSync(
|
|
path.join(home, '.qwen', 'settings.json'),
|
|
JSON.stringify({ ui: { enableFollowupSuggestions: false } }),
|
|
{ mode: 0o600 },
|
|
);
|
|
}
|
|
} else {
|
|
const sourceSettings = JSON.parse(
|
|
readFileSync(settingsPath, 'utf8'),
|
|
) as Record<string, unknown>;
|
|
const providerGroups = sourceSettings['modelProviders'] as
|
|
| Record<string, unknown>
|
|
| undefined;
|
|
let selectedProviderGroup: string | undefined;
|
|
let selectedProvider: Record<string, unknown> | undefined;
|
|
for (const [group, entries] of Object.entries(providerGroups ?? {})) {
|
|
const providers = Array.isArray(entries)
|
|
? entries
|
|
: Object.values((entries ?? {}) as Record<string, unknown>);
|
|
const match = providers.find(
|
|
(entry) =>
|
|
typeof entry === 'object' &&
|
|
entry !== null &&
|
|
((entry as Record<string, unknown>)['id'] === model ||
|
|
(entry as Record<string, unknown>)['name'] === model),
|
|
);
|
|
if (match) {
|
|
selectedProviderGroup = group;
|
|
selectedProvider = match as Record<string, unknown>;
|
|
break;
|
|
}
|
|
}
|
|
if (!selectedProviderGroup || !selectedProvider) {
|
|
throw new Error(
|
|
`Model provider is missing from ${settingsPath}: ${model}`,
|
|
);
|
|
}
|
|
const environmentKey = selectedProvider['envKey'];
|
|
const sourceEnvironment = (sourceSettings['env'] ?? {}) as Record<
|
|
string,
|
|
unknown
|
|
>;
|
|
if (
|
|
typeof environmentKey !== 'string' ||
|
|
typeof (
|
|
sourceEnvironment[environmentKey] ?? process.env[environmentKey]
|
|
) !== 'string'
|
|
) {
|
|
throw new Error(`Model credential is missing for ${model}`);
|
|
}
|
|
const harnessSettings = {
|
|
...(sourceSettings['$version'] === undefined
|
|
? {}
|
|
: { $version: sourceSettings['$version'] }),
|
|
env: {
|
|
[environmentKey]:
|
|
sourceEnvironment[environmentKey] ?? process.env[environmentKey],
|
|
},
|
|
model: {
|
|
name: model,
|
|
...(typeof selectedProvider['baseUrl'] === 'string'
|
|
? { baseUrl: selectedProvider['baseUrl'] }
|
|
: {}),
|
|
},
|
|
modelProviders: { [selectedProviderGroup]: [selectedProvider] },
|
|
...(sourceSettings['security'] === undefined
|
|
? {}
|
|
: { security: sourceSettings['security'] }),
|
|
ui: { enableFollowupSuggestions: false },
|
|
};
|
|
writeFileSync(
|
|
path.join(harnessHome, '.qwen', 'settings.json'),
|
|
JSON.stringify(harnessSettings),
|
|
{ mode: 0o600 },
|
|
);
|
|
}
|
|
writeFileSync(
|
|
path.join(runtimeHome, '.qwen', 'settings.json'),
|
|
JSON.stringify({ ui: { enableFollowupSuggestions: false } }),
|
|
{ mode: 0o600 },
|
|
);
|
|
if (durableFailover) {
|
|
writeFileSync(
|
|
path.join(replacementRuntimeHome, '.qwen', 'settings.json'),
|
|
JSON.stringify({ ui: { enableFollowupSuggestions: false } }),
|
|
{ mode: 0o600 },
|
|
);
|
|
}
|
|
writeFileSync(
|
|
trustedFolders,
|
|
JSON.stringify({ [workspace]: 'TRUST_FOLDER' }),
|
|
{ mode: 0o600 },
|
|
);
|
|
writeFileSync(
|
|
delayedNode,
|
|
`#!/bin/sh\n/bin/sleep ${runtimeDelayMs / 1000}\nexec '${process.execPath.replaceAll("'", "'\\''")}' "$@"\n`,
|
|
{ mode: 0o700 },
|
|
);
|
|
} catch (error) {
|
|
rmSync(temporary, { recursive: true, force: true });
|
|
throw error;
|
|
}
|
|
|
|
const cleanEnvironment = Object.fromEntries(
|
|
Object.entries(process.env).filter(
|
|
([key]) =>
|
|
!/^(https?|all)_proxy$/i.test(key) &&
|
|
!/^(qwen|dashscope|openai|anthropic|google|gemini|azure|aws|vertex)_/i.test(
|
|
key,
|
|
) &&
|
|
!/(api_?key|token|secret|password|credentials?)$/i.test(key),
|
|
),
|
|
);
|
|
|
|
type Child = { child: ChildProcess; log: () => string; name: string };
|
|
const children: Child[] = [];
|
|
// Children the frozen control-plane arm stopped: teardown must wake them, or
|
|
// stopChild's SIGTERM stays pending on a stopped tree for its whole stall.
|
|
const frozenChildren: Child[] = [];
|
|
|
|
function start(
|
|
executable: string,
|
|
args: string[],
|
|
options: { cwd?: string; env?: NodeJS.ProcessEnv } = {},
|
|
name: string,
|
|
): Child {
|
|
let output = '';
|
|
const child = spawn(executable, args, {
|
|
cwd: options.cwd ?? root,
|
|
env: options.env ?? process.env,
|
|
detached: process.platform !== 'win32',
|
|
stdio: ['ignore', 'pipe', 'pipe'],
|
|
});
|
|
const append = (chunk: Buffer) => {
|
|
output += chunk.toString();
|
|
if (output.length > 32_768) output = output.slice(-32_768);
|
|
};
|
|
child.stdout?.on('data', append);
|
|
child.stderr?.on('data', append);
|
|
const registered = { child, log: () => output, name };
|
|
children.push(registered);
|
|
return registered;
|
|
}
|
|
|
|
// One spelling for "has this child exited": five independent phrasings
|
|
// drifted into the pre-fix ESRCH that process.kill raised on a reaped pid.
|
|
function childExited(child: ChildProcess): boolean {
|
|
return child.exitCode !== null || child.signalCode !== null;
|
|
}
|
|
|
|
function processTreeExists(child: ChildProcess): boolean {
|
|
if (child.pid === undefined) return false;
|
|
if (process.platform === 'win32') {
|
|
return !childExited(child);
|
|
}
|
|
try {
|
|
process.kill(-child.pid, 0);
|
|
return true;
|
|
} catch (error) {
|
|
return (error as NodeJS.ErrnoException).code !== 'ESRCH';
|
|
}
|
|
}
|
|
|
|
function signalProcessTree(child: ChildProcess, signal: NodeJS.Signals): void {
|
|
if (process.platform !== 'win32' && child.pid !== undefined) {
|
|
try {
|
|
process.kill(-child.pid, signal);
|
|
return;
|
|
} catch (error) {
|
|
if ((error as NodeJS.ErrnoException).code === 'ESRCH') return;
|
|
}
|
|
}
|
|
child.kill(signal);
|
|
}
|
|
|
|
async function stopChild(child: ChildProcess): Promise<void> {
|
|
if (!processTreeExists(child)) return;
|
|
signalProcessTree(child, 'SIGTERM');
|
|
const deadline = Date.now() + 10_000;
|
|
while (processTreeExists(child) && Date.now() < deadline) {
|
|
await new Promise((resolve) => setTimeout(resolve, 50));
|
|
}
|
|
if (processTreeExists(child)) {
|
|
signalProcessTree(child, 'SIGKILL');
|
|
const killDeadline = Date.now() + 5_000;
|
|
while (processTreeExists(child) && Date.now() < killDeadline) {
|
|
await new Promise((resolve) => setTimeout(resolve, 50));
|
|
}
|
|
}
|
|
}
|
|
|
|
async function crashChild(child: ChildProcess, name: string): Promise<void> {
|
|
if (!processTreeExists(child)) {
|
|
throw new Error(`${name} exited before the crash was injected`);
|
|
}
|
|
signalProcessTree(child, 'SIGKILL');
|
|
const deadline = Date.now() + 5_000;
|
|
while (processTreeExists(child) && Date.now() < deadline) {
|
|
await new Promise((resolve) => setTimeout(resolve, 50));
|
|
}
|
|
if (processTreeExists(child)) {
|
|
throw new Error(`${name} process tree survived SIGKILL`);
|
|
}
|
|
}
|
|
|
|
async function crashProcess(child: ChildProcess, name: string): Promise<void> {
|
|
// A child killed BY signal has exitCode null but signalCode set; missing
|
|
// that branch kills a pid that no longer exists and throws raw ESRCH.
|
|
if (child.pid === undefined || childExited(child)) {
|
|
throw new Error(`${name} exited before the crash was injected`);
|
|
}
|
|
process.kill(child.pid, 'SIGKILL');
|
|
const deadline = Date.now() + 5_000;
|
|
while (!childExited(child) && Date.now() < deadline) {
|
|
await new Promise((resolve) => setTimeout(resolve, 50));
|
|
}
|
|
if (!childExited(child)) {
|
|
throw new Error(`${name} survived SIGKILL`);
|
|
}
|
|
}
|
|
|
|
const allocatedPorts = new Set<number>();
|
|
|
|
async function freePort(): Promise<number> {
|
|
const server = createServer();
|
|
await new Promise<void>((resolve, reject) => {
|
|
server.once('error', reject);
|
|
server.listen(0, '127.0.0.1', () => resolve());
|
|
});
|
|
const address = server.address();
|
|
if (!address || typeof address === 'string') {
|
|
server.close();
|
|
throw new Error('Could not allocate a loopback port');
|
|
}
|
|
await new Promise<void>((resolve, reject) =>
|
|
server.close((error) => (error ? reject(error) : resolve())),
|
|
);
|
|
if (allocatedPorts.has(address.port)) return freePort();
|
|
allocatedPorts.add(address.port);
|
|
return address.port;
|
|
}
|
|
|
|
type HeldExecutionStartProxy = {
|
|
baseUrl: string;
|
|
close: () => Promise<void>;
|
|
heldPath: () => string | undefined;
|
|
observations: () => string[];
|
|
// C3: every Runtime cleanup release is refused until allowRelease(), so
|
|
// the doomed generation cannot discharge its own cleanup debt before the
|
|
// crash lands.
|
|
releaseFailureCount: () => number;
|
|
releaseSuccessCount: () => number;
|
|
allowRelease: () => void;
|
|
};
|
|
|
|
async function startHeldExecutionStartProxy(
|
|
targetOrigin: string,
|
|
holdExecutionStart = true,
|
|
failRelease = false,
|
|
): Promise<HeldExecutionStartProxy> {
|
|
let heldPath: string | undefined;
|
|
const observations: string[] = [];
|
|
let closed = false;
|
|
let releaseAllowed = !failRelease;
|
|
let releaseFailureCount = 0;
|
|
let releaseSuccessCount = 0;
|
|
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 target = new URL(request.url ?? '/', targetOrigin);
|
|
const method = request.method ?? 'GET';
|
|
observations.push(`${method} ${target.pathname} received`);
|
|
if (
|
|
failRelease &&
|
|
method === 'POST' &&
|
|
target.pathname.endsWith(':release') &&
|
|
!target.pathname.endsWith(':release-mount')
|
|
) {
|
|
if (!releaseAllowed) {
|
|
releaseFailureCount += 1;
|
|
observations.push(`${method} ${target.pathname} refused`);
|
|
response.writeHead(500, { 'content-type': 'application/json' });
|
|
response.end('{"error":"e2e_cleanup_release_failed"}');
|
|
return;
|
|
}
|
|
releaseSuccessCount += 1;
|
|
}
|
|
if (
|
|
holdExecutionStart &&
|
|
method === 'POST' &&
|
|
target.pathname.includes('/executions/') &&
|
|
target.pathname.endsWith(':start')
|
|
) {
|
|
heldPath ??= target.pathname;
|
|
observations.push(`${method} ${target.pathname} held`);
|
|
request.socket.once('close', () => response.destroy());
|
|
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 body = Buffer.concat(chunks);
|
|
const upstream = await fetch(target, {
|
|
method,
|
|
headers,
|
|
...(['GET', 'HEAD'].includes(method) ? {} : { body }),
|
|
});
|
|
const upstreamBody = Buffer.from(await upstream.arrayBuffer());
|
|
observations.push(
|
|
`${method} ${target.pathname}${target.search} ${upstream.status}${
|
|
upstream.ok ? '' : ` ${upstreamBody.toString('utf8').slice(0, 500)}`
|
|
}`,
|
|
);
|
|
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();
|
|
});
|
|
});
|
|
const address = server.address();
|
|
if (!address || typeof address === 'string') {
|
|
server.close();
|
|
throw new Error('Held Runtime Broker proxy omitted its address');
|
|
}
|
|
return {
|
|
baseUrl: `http://127.0.0.1:${address.port}`,
|
|
heldPath: () => heldPath,
|
|
observations: () => [...observations],
|
|
releaseFailureCount: () => releaseFailureCount,
|
|
releaseSuccessCount: () => releaseSuccessCount,
|
|
allowRelease: () => {
|
|
releaseAllowed = true;
|
|
observations.push('release allowed');
|
|
},
|
|
close: async () => {
|
|
if (closed) return;
|
|
closed = true;
|
|
server.closeAllConnections();
|
|
await new Promise<void>((resolve, reject) => {
|
|
server.close((error) => (error ? reject(error) : resolve()));
|
|
});
|
|
},
|
|
};
|
|
}
|
|
|
|
// A 202 only admits the response operation (`PENDING` / `JAVA_DURABLE`);
|
|
// the Action row stays `requested` until the coordinator's delivery commits
|
|
// the decision. The after-decision barrier must therefore read the durable
|
|
// row, not the HTTP answer.
|
|
function approvalDecisionCommitted(row: string): boolean {
|
|
const [state, receipt, digest] = row.split('\t');
|
|
return state === 'decided' && receipt === '1' && digest === '1';
|
|
}
|
|
|
|
// The control plane answers 202 only after the decision or admission is
|
|
// committed, so dropping the reply here is a lost response, not a weaker
|
|
// request: the server keeps the committed row, and the caller retries the
|
|
// same Idempotency-Key and must land on that original command.
|
|
function postAndDropReply(
|
|
url: string,
|
|
body: string,
|
|
headers: Record<string, string>,
|
|
): Promise<void> {
|
|
const target = new URL(url);
|
|
return new Promise<void>((resolve, reject) => {
|
|
const request = httpRequest({
|
|
hostname: target.hostname,
|
|
port: target.port,
|
|
path: `${target.pathname}${target.search}`,
|
|
method: 'POST',
|
|
headers: { ...headers, 'content-length': Buffer.byteLength(body) },
|
|
});
|
|
request.once('response', (response) => {
|
|
// Destroy the reply unread: the caller must behave as if the answer
|
|
// never arrived, while the committed row stays on the server. A
|
|
// non-202 answer means nothing was committed, so the loss would prove
|
|
// nothing about the retry.
|
|
if (response.statusCode !== 202) {
|
|
response.destroy();
|
|
reject(
|
|
new Error(
|
|
`Lost-reply probe expected 202 from ${target.pathname}, got ${response.statusCode}`,
|
|
),
|
|
);
|
|
return;
|
|
}
|
|
response.once('error', () => {});
|
|
response.destroy();
|
|
resolve();
|
|
});
|
|
request.once('error', reject);
|
|
request.end(body);
|
|
});
|
|
}
|
|
|
|
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,
|
|
timeoutMs: number,
|
|
child?: Child | Child[],
|
|
): Promise<void> {
|
|
const deadline = Date.now() + timeoutMs;
|
|
const watched =
|
|
child === undefined ? [] : Array.isArray(child) ? child : [child];
|
|
let lastError: unknown;
|
|
let lastErrorPredatesStall = false;
|
|
// Identity, not text: matching the rendered message would let a reworded
|
|
// sentinel silently invert the error precedence in the catch below.
|
|
const stall = new Error(`${name} predicate stalled`);
|
|
while (Date.now() < deadline) {
|
|
if (receivedSignal) throw new Error(`Interrupted by ${receivedSignal}`);
|
|
for (const each of watched) {
|
|
if (childExited(each.child)) {
|
|
throw new Error(`${name} exited early: ${each.name}\n${each.log()}`);
|
|
}
|
|
}
|
|
try {
|
|
// Bound one iteration against the remaining deadline: a hung
|
|
// predicate must not outlive timeoutMs, and its last error must
|
|
// surface instead of vanishing into "did not become ready". unref the
|
|
// stall timer: a quick success must not keep the event loop (and this
|
|
// runner) alive until the discarded timeout fires. The race bounds
|
|
// only an asynchronous predicate — a synchronous one blocks the event
|
|
// loop, so the remaining budget is handed to the predicate for its
|
|
// own probe timeout.
|
|
const ready = await Promise.race([
|
|
Promise.resolve().then(() =>
|
|
predicate(Math.max(1, deadline - Date.now())),
|
|
),
|
|
new Promise<boolean>((_, reject) => {
|
|
// Floor the bound: the final iteration can start with a sliver of
|
|
// budget left, and a poll that would answer just past the deadline
|
|
// must resolve the wait, not be discarded and misreported as a
|
|
// wedged predicate. The clamp re-admits up to ~500ms of overshoot
|
|
// on a genuinely wedged poll — the same trade runMysql's probe
|
|
// timeout floor makes for the synchronous path.
|
|
const stallTimer = setTimeout(
|
|
() => reject(stall),
|
|
Math.max(500, deadline - Date.now()),
|
|
);
|
|
stallTimer.unref();
|
|
}),
|
|
]);
|
|
if (ready) return;
|
|
// The predicate answered; an error from an earlier phase no longer
|
|
// describes the state the deadline found.
|
|
lastError = undefined;
|
|
} catch (error) {
|
|
// Prefer the cause an undici fetch rejection carries (its own message
|
|
// is "fetch failed"). A real predicate error wins over the synthetic
|
|
// stall; when the stall is what the deadline found, the older error
|
|
// is reported as predating it instead of posing as the final state.
|
|
if (error instanceof Error && error.cause instanceof Error) {
|
|
lastError = error.cause;
|
|
} else if (error !== stall) {
|
|
lastError = error;
|
|
} else if (lastError === undefined) {
|
|
lastError = error;
|
|
} else {
|
|
lastErrorPredatesStall = true;
|
|
}
|
|
}
|
|
await new Promise((resolve) => setTimeout(resolve, 100));
|
|
}
|
|
// A non-Error rejection (a status string, a library object) is still
|
|
// the last thing the predicate said: inspect it so the content survives —
|
|
// String() renders a plain object as "[object Object]".
|
|
let cause = '';
|
|
if (lastError instanceof Error) {
|
|
cause = `: ${lastError.message}`;
|
|
} else if (lastError !== undefined) {
|
|
try {
|
|
cause = `: ${inspect(lastError)}`;
|
|
} catch {
|
|
cause = `: ${String(lastError)}`;
|
|
}
|
|
}
|
|
if (lastErrorPredatesStall) {
|
|
cause = `: predicate stalled; last error before the stall${cause}`;
|
|
}
|
|
throw new Error(
|
|
`${name} did not become ready${cause}\n${watched.map((each) => each.log()).join('\n')}`,
|
|
);
|
|
}
|
|
|
|
// MySQL reads $HOME/.mylogin.cnf even when option files are disabled,
|
|
// so a mysql_config_editor credential on a developer's machine would
|
|
// silently auth-connect to the scratch empty-password server — and an
|
|
// exported MYSQL_PWD (read as the password) or MYSQL_TEST_LOGIN_FILE
|
|
// (relocates .mylogin.cnf ahead of $HOME, past --no-defaults) does the
|
|
// same. Client invocations run against an isolated environment instead of
|
|
// inheriting the real one: an empty HOME under the runner's one scratch
|
|
// root, so the finally reclaims it and QWEN_MANAGED_E2E_KEEP_TMP=1 keeps
|
|
// it for inspection, with both credential variables stripped.
|
|
const mysqlClientHome = path.join(temporary, 'mysql-client-home');
|
|
mkdirSync(mysqlClientHome, { recursive: true });
|
|
const mysqlClientEnv: NodeJS.ProcessEnv = {
|
|
...process.env,
|
|
HOME: mysqlClientHome,
|
|
};
|
|
delete mysqlClientEnv['MYSQL_PWD'];
|
|
delete mysqlClientEnv['MYSQL_TEST_LOGIN_FILE'];
|
|
|
|
function runMysql(port: number, sql: string, timeoutMs = 10_000): string {
|
|
const result = spawnSync(
|
|
mysql,
|
|
[
|
|
// First argument, honored only there: a developer's ~/.my.cnf client
|
|
// options must not auth-fail a scratch empty-password server.
|
|
'--no-defaults',
|
|
'--protocol=tcp',
|
|
'--host=127.0.0.1',
|
|
`--port=${port}`,
|
|
'--user=root',
|
|
'--batch',
|
|
'--skip-column-names',
|
|
'--execute',
|
|
sql,
|
|
],
|
|
{
|
|
encoding: 'utf8',
|
|
env: mysqlClientEnv,
|
|
// waitUntil hands the lease polls its remaining budget, and the final
|
|
// iteration can shrink it below one client round trip; floor the probe
|
|
// timeout so the last poll is never handed less than it can cost. The
|
|
// clamp re-admits up to ~2s of overshoot per wedged poll, against the
|
|
// ~10s the budget threading set out to remove.
|
|
timeout: Math.max(2_000, timeoutMs),
|
|
maxBuffer: 16 * 1024 * 1024,
|
|
},
|
|
);
|
|
// A timed-out spawn sets error and leaves status null with empty stderr:
|
|
// surface the spawn-level reason, or a wedge reads as a bare
|
|
// "MySQL command failed: ".
|
|
if (result.error) {
|
|
throw new Error(`MySQL command failed: ${result.error.message}`);
|
|
}
|
|
if (result.status !== 0) {
|
|
throw new Error(`MySQL command failed: ${result.stderr}`);
|
|
}
|
|
return result.stdout.trim();
|
|
}
|
|
|
|
function assertStoredAnswer(
|
|
port: number,
|
|
tenant: string,
|
|
sessionId: string,
|
|
expected: string,
|
|
): void {
|
|
const rows = runMysql(
|
|
port,
|
|
`SELECT resource_id, kind, byte_length, sha256, HEX(inline_bytes) FROM qwen_managed_agent.ma_resource WHERE tenant_id=${sqlString(tenant)} AND session_id=${sqlString(sessionId)} AND kind IN ('managed-message', 'managed-message-chunks', 'managed-message-part')`,
|
|
).split('\n');
|
|
const bodies = new Map<string, { kind: string; bytes: Buffer }>();
|
|
for (const row of rows) {
|
|
const [id, kind, length, digest, hex] = row.split('\t');
|
|
const bytes = Buffer.from(hex ?? '', 'hex');
|
|
const actualDigest = createHash('sha256').update(bytes).digest('hex');
|
|
let reason: string | undefined;
|
|
if (!id) reason = 'missing resource id';
|
|
else if (!kind) reason = 'missing resource kind';
|
|
else if (!hex || hex === 'NULL') reason = 'missing inline bytes';
|
|
else if (bytes.length !== Number(length))
|
|
reason = `byte_length=${length} actual=${bytes.length}`;
|
|
else if (actualDigest !== digest)
|
|
reason = `sha256=${digest} actual=${actualDigest}`;
|
|
else if (bytes.length > 65_536)
|
|
reason = `inline body ${bytes.length}B exceeds 65536B`;
|
|
if (reason) {
|
|
throw new Error(
|
|
`Stored message resource failed validation: kind=${kind ?? 'unknown'} id=${id ?? 'unknown'} ${reason}`,
|
|
);
|
|
}
|
|
bodies.set(id, { kind, bytes });
|
|
}
|
|
const records = [...bodies.entries()]
|
|
.filter(([, { kind }]) => kind !== 'managed-message-part')
|
|
.map(([id, { kind, bytes }]) => {
|
|
if (kind === 'managed-message-chunks') {
|
|
let manifest: { parts: Array<{ resourceId: string }> };
|
|
try {
|
|
manifest = JSON.parse(bytes.toString('utf8')) as typeof manifest;
|
|
} catch {
|
|
throw new Error(
|
|
`Stored message manifest is invalid JSON: kind=${kind} id=${id}`,
|
|
);
|
|
}
|
|
bytes = Buffer.concat(
|
|
manifest.parts.map((part) => {
|
|
const stored = bodies.get(part.resourceId);
|
|
if (stored?.kind !== 'managed-message-part')
|
|
throw new Error(
|
|
`Stored message part is missing: kind=managed-message-part id=${part.resourceId} manifest=${id}`,
|
|
);
|
|
return stored.bytes;
|
|
}),
|
|
);
|
|
}
|
|
try {
|
|
return JSON.parse(bytes.toString('utf8')) as {
|
|
type: string;
|
|
uuid: string;
|
|
parentUuid: string;
|
|
message?: { parts?: Array<{ text?: string }> };
|
|
};
|
|
} catch {
|
|
throw new Error(
|
|
`Stored message ${kind === 'managed-message-chunks' ? 'joined-chunk' : 'inline'} record is invalid JSON: kind=${kind} id=${id}`,
|
|
);
|
|
}
|
|
});
|
|
const answers = records.filter(
|
|
(record) =>
|
|
record.type === 'assistant' &&
|
|
record.message?.parts?.map((part) => part.text ?? '').join('') ===
|
|
expected,
|
|
);
|
|
if (answers.length !== 1 || !answers[0].uuid || !answers[0].parentUuid)
|
|
throw new Error('Stored answer is incomplete or duplicated');
|
|
const manifests = [...bodies.values()].filter(
|
|
({ kind }) => kind === 'managed-message-chunks',
|
|
);
|
|
if (manifests.length !== 1)
|
|
throw new Error(
|
|
`Expected exactly one managed-message-chunks manifest, found ${manifests.length}`,
|
|
);
|
|
}
|
|
|
|
// B3a: assert the crash window from the same durable journal the recovery
|
|
// reader reads, not from the hook's log line. The window's left boundary is
|
|
// the post-tool main attempt's usage-only Finished marker (state
|
|
// output_committed); its right boundary is the assistant message.committed
|
|
// that has not landed yet. This mirrors findHostedCommittedModelOutput's own
|
|
// discriminator — a latest main attempt that is not output_committed, or an
|
|
// assistant commit at/after that attempt, both fall outside the window — so a
|
|
// hold moved above the Finished commit reds here instead of passing vacuously
|
|
// (the recovery reader also accepts a saved request whose attempt is still
|
|
// started, so request count / parity / answer equality cannot tell that
|
|
// earlier crash apart from B3a). The journal is UTF-8 JSON Lines in
|
|
// ma_session_journal_tx.record_bytes; read it back as HEX like the stored
|
|
// answer probe so embedded newlines never split a row.
|
|
function finishedBeforeCommitWindow(
|
|
port: number,
|
|
sessionFilter: string,
|
|
promptId: string,
|
|
): { attemptId: string; attemptSequence: number } {
|
|
const events = readJournalManagedEvents(port, sessionFilter)
|
|
.filter((event) => typeof event.sequence === 'number')
|
|
.map((event) => ({
|
|
kind: event.kind,
|
|
sequence: event.sequence as number,
|
|
payload: event.payload ?? {},
|
|
}));
|
|
events.sort((a, b) => a.sequence - b.sequence);
|
|
const mainAttempts = events.filter(
|
|
(event) =>
|
|
event.kind === 'model.attempt' &&
|
|
typeof event.payload['attemptId'] === 'string' &&
|
|
(event.payload['attemptId'] as string).startsWith(`${promptId}:main:`),
|
|
);
|
|
const latestAttempt = mainAttempts[mainAttempts.length - 1];
|
|
if (!latestAttempt)
|
|
throw new Error(
|
|
`Finished-before-commit crash window has no main model attempt: promptId=${promptId}`,
|
|
);
|
|
if (latestAttempt.payload['state'] !== 'output_committed')
|
|
throw new Error(
|
|
`Finished-before-commit crash landed before the model Finished marker committed: attemptState=${String(latestAttempt.payload['state'])} attemptId=${String(latestAttempt.payload['attemptId'])}`,
|
|
);
|
|
const assistantCommit = events.find(
|
|
(event) =>
|
|
event.kind === 'message.committed' &&
|
|
event.payload['role'] === 'assistant' &&
|
|
event.sequence > latestAttempt.sequence,
|
|
);
|
|
if (assistantCommit)
|
|
throw new Error(
|
|
`Finished-before-commit crash landed after the assistant commit: commitSequence=${assistantCommit.sequence} attemptSequence=${latestAttempt.sequence}`,
|
|
);
|
|
return {
|
|
attemptId: latestAttempt.payload['attemptId'] as string,
|
|
attemptSequence: latestAttempt.sequence,
|
|
};
|
|
}
|
|
|
|
// A3: read the recovered final batch plan the way recovery does — the latest
|
|
// hosted.batch.planned link for the batch, then its continuation resource —
|
|
// so the sibling order/identity assertion runs on the same durable plan the
|
|
// next generation resumed from, not on the execution store's row order.
|
|
interface SiblingFinalPlan {
|
|
stage: string;
|
|
calls: Array<{
|
|
runtimeCallId: string;
|
|
partIndex: number;
|
|
prepareKey: string;
|
|
requestDigest: string;
|
|
actionId: string | null;
|
|
}>;
|
|
}
|
|
function siblingFinalPlan(
|
|
port: number,
|
|
sessionFilter: string,
|
|
batchId: string,
|
|
): SiblingFinalPlan {
|
|
let planRef: { resourceId?: string } | undefined;
|
|
let planSequence = -1;
|
|
for (const event of readJournalManagedEvents(port, sessionFilter)) {
|
|
if (
|
|
typeof event.sequence !== 'number' ||
|
|
event.kind !== 'hosted.batch.planned'
|
|
)
|
|
continue;
|
|
const payload = event.payload ?? {};
|
|
if (payload['batchId'] !== batchId) continue;
|
|
if (event.sequence > planSequence) {
|
|
planSequence = event.sequence;
|
|
planRef = payload['planRef'] as { resourceId?: string } | undefined;
|
|
}
|
|
}
|
|
if (!planRef?.resourceId)
|
|
throw new Error(
|
|
`Sibling recovery has no final batch plan link: batchId=${batchId}`,
|
|
);
|
|
const plan = JSON.parse(
|
|
runMysql(
|
|
port,
|
|
`SELECT CONVERT(inline_bytes USING utf8mb4) FROM qwen_managed_agent.ma_resource WHERE ${sessionFilter} AND resource_id=${sqlString(planRef.resourceId)}`,
|
|
),
|
|
);
|
|
if (plan.stage !== 'final' || !Array.isArray(plan.calls))
|
|
throw new Error(
|
|
`Sibling recovered plan is not a final batch plan: stage=${String(plan.stage)}`,
|
|
);
|
|
return plan;
|
|
}
|
|
|
|
// B3: read the durable round and cost ledger the way recovery restores them.
|
|
// The round comes from the latest hosted-model-request (model.attempt.recoveryRef);
|
|
// the budget snapshot and per-model token metrics are reverse-scanned from the
|
|
// route/usage resources exactly as bindBudget does on a cold load.
|
|
interface B3RoundBudget {
|
|
round: number;
|
|
budget: unknown;
|
|
models: unknown;
|
|
}
|
|
function b3RoundAndBudget(port: number, sessionFilter: string): B3RoundBudget {
|
|
const attempts = readJournalManagedEvents(port, sessionFilter)
|
|
.filter(
|
|
(event) =>
|
|
event.kind === 'model.attempt' && typeof event.sequence === 'number',
|
|
)
|
|
.map((event) => ({
|
|
sequence: event.sequence as number,
|
|
payload: event.payload ?? {},
|
|
}));
|
|
attempts.sort((a, b) => a.sequence - b.sequence);
|
|
if (attempts.length === 0) throw new Error('B3 window has no model attempt');
|
|
const readResource = (ref: unknown): Record<string, unknown> | undefined => {
|
|
const resourceId = (ref as { resourceId?: string } | null)?.resourceId;
|
|
if (!resourceId) return undefined;
|
|
// HEX + decode avoids the mysql --batch backslash escaping that would
|
|
// corrupt the JSON's \" sequences — the saved request carries the full
|
|
// system prompt, unlike the small approval-continuation plan.
|
|
const hex = runMysql(
|
|
port,
|
|
`SELECT HEX(inline_bytes) FROM qwen_managed_agent.ma_resource WHERE ${sessionFilter} AND resource_id=${sqlString(resourceId)}`,
|
|
);
|
|
if (!hex || hex === 'NULL') return undefined;
|
|
return JSON.parse(Buffer.from(hex, 'hex').toString('utf8')) as Record<
|
|
string,
|
|
unknown
|
|
>;
|
|
};
|
|
let round = -1;
|
|
for (let i = attempts.length - 1; i >= 0; i -= 1) {
|
|
const req = readResource(attempts[i].payload['recoveryRef']);
|
|
if (req && typeof req['round'] === 'number') {
|
|
round = req['round'];
|
|
break;
|
|
}
|
|
}
|
|
let budget: unknown;
|
|
let models: unknown;
|
|
for (let i = attempts.length - 1; i >= 0; i -= 1) {
|
|
for (const field of ['usageRef', 'routeRef']) {
|
|
const saved = readResource(attempts[i].payload[field]);
|
|
if (!saved) continue;
|
|
if (budget === undefined && 'budget' in saved) budget = saved['budget'];
|
|
if (models === undefined && saved['models']) models = saved['models'];
|
|
}
|
|
if (budget !== undefined && models !== undefined) break;
|
|
}
|
|
if (round < 0) throw new Error('B3 window has no saved model-request round');
|
|
return { round, budget: budget ?? null, models: models ?? null };
|
|
}
|
|
|
|
// B3: the turn cost snapshot and per-model token ledger must survive the
|
|
// replacement. The snapshot's NUMERIC budget must be present: a non-null
|
|
// snapshot with budget: null is a legal production state (no +Nk directive
|
|
// parsed), and an object-truthy guard would let null == null pass vacuously.
|
|
// The ledger may grow as the held round completes, but every crash-boundary
|
|
// model's output-token total must survive; an unrestored ledger is a truthy
|
|
// {}, so a null check alone would let a zeroed restore pass too.
|
|
function assertB3CostContinuity(
|
|
before: B3RoundBudget,
|
|
after: B3RoundBudget,
|
|
): void {
|
|
const b3Budget = (before.budget as { budget?: number | null } | null)?.budget;
|
|
if (b3Budget === null || b3Budget === undefined)
|
|
throw new Error(
|
|
`B3 crash-boundary turn budget was not exercised: ${JSON.stringify(before.budget)}`,
|
|
);
|
|
if (!isDeepStrictEqual(after.budget, before.budget))
|
|
throw new Error(
|
|
`B3 recovery reset the turn budget snapshot: before=${JSON.stringify(before.budget)} after=${JSON.stringify(after.budget)}`,
|
|
);
|
|
const b3ModelsBefore = (before.models ?? {}) as Record<
|
|
string,
|
|
{ tokens?: { candidates?: number } }
|
|
>;
|
|
const b3ModelsAfter = (after.models ?? {}) as Record<
|
|
string,
|
|
{ tokens?: { candidates?: number } }
|
|
>;
|
|
const b3LedgerEntries = Object.entries(b3ModelsBefore);
|
|
if (
|
|
b3LedgerEntries.length === 0 ||
|
|
b3LedgerEntries.every(
|
|
([, metric]) => (metric.tokens?.candidates ?? 0) === 0,
|
|
)
|
|
)
|
|
throw new Error(
|
|
'B3 crash-boundary per-model token ledger is empty: the fixture did not exercise the ledger',
|
|
);
|
|
for (const [model, metric] of b3LedgerEntries) {
|
|
const beforeTokens = metric.tokens?.candidates ?? 0;
|
|
const afterTokens = b3ModelsAfter[model]?.tokens?.candidates ?? 0;
|
|
if (afterTokens < beforeTokens)
|
|
throw new Error(
|
|
`B3 recovery wiped the per-model token ledger: model=${model} before=${beforeTokens} after=${afterTokens}`,
|
|
);
|
|
}
|
|
}
|
|
|
|
// The text-delta selector shared by deltaSequences (which throws on an empty
|
|
// match) and b2DeltaIdentity (which tolerates it): one filter, two policies.
|
|
function textDeltas(events: PublicEvent[], text: string): PublicEvent[] {
|
|
return events.filter(
|
|
(event) =>
|
|
event.type === 'item.output_text.delta' &&
|
|
eventText(event).includes(text),
|
|
);
|
|
}
|
|
|
|
// B2: one identity group per prefix text — the item/part ids the public deltas
|
|
// carried while that prefix was live. After retraction the reload view blanks
|
|
// those deltas, so the dead generations' identities survive only in the live
|
|
// SSE capture; that is what makes the live view load-bearing.
|
|
function b2DeltaIdentity(
|
|
events: PublicEvent[],
|
|
text: string,
|
|
): { itemIds: string[]; partIds: string[]; sequences: number[] } {
|
|
const deltas = textDeltas(events, text);
|
|
const distinct = (values: Array<string | null | undefined>): string[] => [
|
|
...new Set(values.filter((id): id is string => typeof id === 'string')),
|
|
];
|
|
return {
|
|
itemIds: distinct(deltas.map((event) => event.item_id)),
|
|
partIds: distinct(deltas.map((event) => event.content_part_id)),
|
|
sequences: deltas.map((event) => event.sequence),
|
|
};
|
|
}
|
|
|
|
// The journal HEX decode shared by every window/identity reader: --batch text
|
|
// would escape backslashes and corrupt the JSON, so rows are read as HEX and
|
|
// decoded here, once.
|
|
function readJournalManagedEvents(
|
|
port: number,
|
|
sessionFilter: string,
|
|
): Array<{
|
|
kind?: unknown;
|
|
sequence?: unknown;
|
|
payload?: Record<string, unknown>;
|
|
}> {
|
|
const rows = runMysql(
|
|
port,
|
|
`SELECT HEX(record_bytes) FROM qwen_managed_agent.ma_session_journal_tx WHERE ${sessionFilter} ORDER BY first_sequence`,
|
|
).split('\n');
|
|
const events: Array<{
|
|
kind?: unknown;
|
|
sequence?: unknown;
|
|
payload?: Record<string, unknown>;
|
|
}> = [];
|
|
for (const row of rows) {
|
|
if (!row || row === 'NULL') continue;
|
|
for (const line of Buffer.from(row, 'hex').toString('utf8').split('\n')) {
|
|
if (!line) continue;
|
|
const record = JSON.parse(line) as {
|
|
subtype?: unknown;
|
|
managedSession?: {
|
|
kind?: unknown;
|
|
sequence?: unknown;
|
|
payload?: Record<string, unknown>;
|
|
};
|
|
};
|
|
if (record.subtype !== 'managed_session_event_v1') continue;
|
|
if (record.managedSession) events.push(record.managedSession);
|
|
}
|
|
}
|
|
return events;
|
|
}
|
|
|
|
// B2: the original prompt's model attempts, oldest first. Each streaming
|
|
// generation must open a new attemptId.
|
|
function b2ModelAttempts(
|
|
port: number,
|
|
sessionFilter: string,
|
|
promptId: string,
|
|
): Array<{ attemptId: string; state: string; sequence: number }> {
|
|
const attempts: Array<{
|
|
attemptId: string;
|
|
state: string;
|
|
sequence: number;
|
|
}> = [];
|
|
for (const event of readJournalManagedEvents(port, sessionFilter)) {
|
|
if (event.kind !== 'model.attempt' || typeof event.sequence !== 'number')
|
|
continue;
|
|
const attemptId = event.payload?.['attemptId'];
|
|
if (
|
|
typeof attemptId !== 'string' ||
|
|
!attemptId.startsWith(`${promptId}:main:`)
|
|
)
|
|
continue;
|
|
attempts.push({
|
|
attemptId,
|
|
state: String(event.payload?.['state'] ?? ''),
|
|
sequence: event.sequence,
|
|
});
|
|
}
|
|
return attempts.sort((a, b) => a.sequence - b.sequence);
|
|
}
|
|
|
|
interface B2SseCapture {
|
|
events: PublicEvent[];
|
|
lastId: string;
|
|
}
|
|
|
|
// Open SSE observers must not hold the process open when the arm fails early:
|
|
// the main finally aborts whatever is still pending.
|
|
const b2SseObservers = new Set<SseObserver>();
|
|
|
|
// B2: read the public SSE view through the shared SseObserver (wire-format
|
|
// assertions included). One connection stays open across the Harness crashes
|
|
// (the stream is control-plane-owned and reconciles from after=0 on connect).
|
|
// `until` is a PROGRESS signal, not the exit: after it first holds, a bounded
|
|
// drain window keeps reading so a double-projected terminal is still observed
|
|
// (R1-2). SseObserver and waitUntil are free identifiers so behavioural tests
|
|
// inject stubs.
|
|
async function readPublicSse(
|
|
storeUrl: string,
|
|
sessionId: string,
|
|
options: {
|
|
tenantId: string;
|
|
after?: number;
|
|
lastEventId?: string;
|
|
until: (events: PublicEvent[]) => boolean;
|
|
timeoutMs: number;
|
|
drainMs?: number;
|
|
},
|
|
): Promise<B2SseCapture> {
|
|
const observer = new SseObserver({
|
|
surface: 'public',
|
|
sessionId,
|
|
storeUrl,
|
|
tenantId: options.tenantId,
|
|
timeoutMs: options.timeoutMs,
|
|
});
|
|
b2SseObservers.add(observer);
|
|
const snapshot = (): PublicEvent[] =>
|
|
observer.frames.map((frame) => frame.data as unknown as PublicEvent);
|
|
const fail = (): void => {
|
|
if (observer.failure)
|
|
throw new Error(`B2 live SSE failed: ${String(observer.failure)}`);
|
|
};
|
|
try {
|
|
await observer.start(
|
|
options.lastEventId !== undefined
|
|
? Number(options.lastEventId)
|
|
: options.after,
|
|
);
|
|
await waitUntil(
|
|
'B2 SSE capture progress',
|
|
() => {
|
|
fail();
|
|
return options.until(snapshot());
|
|
},
|
|
options.timeoutMs,
|
|
);
|
|
const drainMs = options.drainMs ?? 2_000;
|
|
const drainEnd = Date.now() + drainMs;
|
|
await waitUntil(
|
|
'B2 SSE capture drain',
|
|
() => {
|
|
fail();
|
|
return Date.now() >= drainEnd;
|
|
},
|
|
drainMs + 1_000,
|
|
);
|
|
fail();
|
|
const events = snapshot();
|
|
return { events, lastId: String(observer.frames.at(-1)?.id ?? '') };
|
|
} catch (error) {
|
|
if (observer.failure)
|
|
throw new Error(`B2 live SSE failed: ${String(observer.failure)}`);
|
|
if (observer.controller.signal.aborted)
|
|
throw new Error(
|
|
`B2 live SSE aborted (budget or teardown): ${String(error)}`,
|
|
);
|
|
throw new Error(`B2 live SSE capture incomplete: ${String(error)}`);
|
|
} finally {
|
|
observer.controller.abort();
|
|
await observer.task?.catch(() => undefined);
|
|
b2SseObservers.delete(observer);
|
|
}
|
|
}
|
|
|
|
// B2: the stream-view audit, lifted out of the arm's try block so behavioural
|
|
// tests execute it (R1-6). The live capture holds the dead generations'
|
|
// original identities; reload/replay hold the blanked prefixes and the
|
|
// surviving answer. item_id is the Turn's ONE assistant item (stable across
|
|
// the repair); the content Part is the stream identity — a delta continues the
|
|
// previous Part only while that delta keeps its identity, so every retraction
|
|
// must force a new Part (pairwise distinct Parts; every group must actually
|
|
// carry the assistant item, R1-3; the replay compares Parts too, R1-4).
|
|
function assertB2StreamViews(input: {
|
|
live: PublicEvent[];
|
|
reload: PublicEvent[];
|
|
replay: PublicEvent[];
|
|
partials: string[];
|
|
answer: string;
|
|
firstPrefixSequence: number;
|
|
secondPrefixSequence: number;
|
|
originalTurnId?: string;
|
|
}): {
|
|
livePartial1: ReturnType<typeof b2DeltaIdentity>;
|
|
livePartial2: ReturnType<typeof b2DeltaIdentity>;
|
|
liveAnswer: ReturnType<typeof b2DeltaIdentity>;
|
|
reloadAnswer: ReturnType<typeof b2DeltaIdentity>;
|
|
replayAnswer: ReturnType<typeof b2DeltaIdentity>;
|
|
} {
|
|
// The public stream.reconciled frame carries no payload: a client seeing one
|
|
// reloads. The converged view is the deltas after the LAST retraction.
|
|
const converged = (events: PublicEvent[]): string => {
|
|
const reconciled = events.filter(
|
|
(event) => event.type === 'stream.reconciled',
|
|
);
|
|
const last = reconciled.length
|
|
? Math.max(...reconciled.map((event) => event.sequence))
|
|
: 0;
|
|
return events
|
|
.filter(
|
|
(event) =>
|
|
event.type === 'item.output_text.delta' && event.sequence > last,
|
|
)
|
|
.map(eventText)
|
|
.join('');
|
|
};
|
|
const terminalsOf = (events: PublicEvent[]): PublicEvent[] =>
|
|
events.filter(
|
|
(event) =>
|
|
event.terminal &&
|
|
(input.originalTurnId === undefined ||
|
|
event.turn_id === input.originalTurnId),
|
|
);
|
|
const livePartial1 = b2DeltaIdentity(input.live, input.partials[0]);
|
|
const livePartial2 = b2DeltaIdentity(input.live, input.partials[1]);
|
|
const liveAnswer = b2DeltaIdentity(input.live, input.answer);
|
|
const liveTerminals = terminalsOf(input.live);
|
|
const liveRetractions = input.live.filter(
|
|
(event) => event.type === 'stream.reconciled',
|
|
);
|
|
if (
|
|
converged(input.live) !== input.answer ||
|
|
liveTerminals.length !== 1 ||
|
|
liveTerminals[0].type !== 'turn.completed' ||
|
|
livePartial1.sequences.length === 0 ||
|
|
livePartial2.sequences.length === 0 ||
|
|
liveAnswer.sequences.length === 0 ||
|
|
liveRetractions.length < 2 ||
|
|
!reconciledBetween(
|
|
input.live,
|
|
Math.min(...livePartial1.sequences),
|
|
Math.min(...livePartial2.sequences),
|
|
) ||
|
|
!reconciledBetween(
|
|
input.live,
|
|
Math.min(...livePartial2.sequences),
|
|
Math.min(...liveAnswer.sequences),
|
|
)
|
|
)
|
|
throw new Error(
|
|
`B2 live SSE: text=${JSON.stringify(converged(input.live))} terminals=${liveTerminals.length} partial1=${livePartial1.sequences.length} partial2=${livePartial2.sequences.length} answer=${liveAnswer.sequences.length} retractions=${liveRetractions.length} frames=${input.live.length}`,
|
|
);
|
|
const reloadAnswer = b2DeltaIdentity(input.reload, input.answer);
|
|
// F5: the live answer's identity is load-bearing — the persisted views being
|
|
// correct does not excuse a live stream that delivers the surviving answer
|
|
// under a dead Part, no identity, or a foreign item.
|
|
if (liveAnswer.itemIds.length === 0 || liveAnswer.partIds.length === 0)
|
|
throw new Error(
|
|
`B2 live SSE: the live answer carries no identity: ${JSON.stringify(liveAnswer)}`,
|
|
);
|
|
if (
|
|
liveAnswer.itemIds.join(',') !== reloadAnswer.itemIds.join(',') ||
|
|
liveAnswer.partIds.join(',') !== reloadAnswer.partIds.join(',')
|
|
)
|
|
throw new Error(
|
|
`B2 live SSE: the live answer identity differs from reload: live=${JSON.stringify(liveAnswer)} reload=${JSON.stringify(reloadAnswer)}`,
|
|
);
|
|
const groups = [livePartial1, livePartial2, reloadAnswer];
|
|
const itemIds = new Set(groups.flatMap((group) => group.itemIds));
|
|
if (
|
|
groups.some(
|
|
(group) => group.partIds.length === 0 || group.itemIds.length === 0,
|
|
) ||
|
|
itemIds.size !== 1
|
|
)
|
|
throw new Error(
|
|
`B2 stream identities: expected one shared assistant item and a Part per generation: ${JSON.stringify(groups)}`,
|
|
);
|
|
for (let i = 0; i < groups.length; i += 1) {
|
|
for (let j = i + 1; j < groups.length; j += 1) {
|
|
if (groups[i].partIds.some((id) => groups[j].partIds.includes(id)))
|
|
throw new Error(
|
|
`B2 stream identities: generations ${i + 1} and ${j + 1} share a content Part (retraction did not break the continuation): ${JSON.stringify(groups)}`,
|
|
);
|
|
}
|
|
}
|
|
const blanked = (events: PublicEvent[], sequence: number): boolean => {
|
|
const delta = events.find((event) => event.sequence === sequence);
|
|
return (
|
|
!!delta &&
|
|
delta.type === 'item.output_text.delta' &&
|
|
eventText(delta) === '' &&
|
|
delta.item_id == null &&
|
|
delta.content_part_id == null
|
|
);
|
|
};
|
|
if (
|
|
!blanked(input.reload, input.firstPrefixSequence) ||
|
|
!blanked(input.reload, input.secondPrefixSequence)
|
|
)
|
|
throw new Error(
|
|
`B2 stream identities: retracted prefixes not blanked in reload at ${input.firstPrefixSequence}/${input.secondPrefixSequence}`,
|
|
);
|
|
const replayAnswer = b2DeltaIdentity(input.replay, input.answer);
|
|
const replayTerminals = terminalsOf(input.replay);
|
|
if (
|
|
converged(input.replay) !== input.answer ||
|
|
replayTerminals.length !== 1 ||
|
|
// F6: a reconnecting client shown a failure terminal while live/reload
|
|
// show success is not a passing B2.
|
|
replayTerminals[0].type !== 'turn.completed' ||
|
|
!blanked(input.replay, input.firstPrefixSequence) ||
|
|
!blanked(input.replay, input.secondPrefixSequence) ||
|
|
replayAnswer.itemIds.join(',') !== reloadAnswer.itemIds.join(',') ||
|
|
replayAnswer.partIds.join(',') !== reloadAnswer.partIds.join(',')
|
|
)
|
|
throw new Error(
|
|
`B2 replay view: text=${JSON.stringify(converged(input.replay))} terminals=${replayTerminals.length} answerIds=${replayAnswer.itemIds.join(',')} answerParts=${replayAnswer.partIds.join(',')} reloadIds=${reloadAnswer.itemIds.join(',')} reloadParts=${reloadAnswer.partIds.join(',')} frames=${input.replay.length}`,
|
|
);
|
|
return { livePartial1, livePartial2, liveAnswer, reloadAnswer, replayAnswer };
|
|
}
|
|
|
|
function assertPublicAnswer(events: PublicEvent[], expected: string): void {
|
|
const terminal = events.filter((event) => event.terminal);
|
|
const text = events
|
|
.filter((event) => event.type === 'item.output_text.delta')
|
|
.map(eventText)
|
|
.join('');
|
|
if (
|
|
terminal.length !== 1 ||
|
|
terminal[0].type !== 'turn.completed' ||
|
|
text !== expected
|
|
) {
|
|
throw new Error(
|
|
`Long-answer integrity failed: terminal=${terminal.map((event) => event.type)} expectedChars=${expected.length} actualChars=${text.length}`,
|
|
);
|
|
}
|
|
}
|
|
|
|
function sqlString(value: string): string {
|
|
return `'${value.replaceAll("'", "''")}'`;
|
|
}
|
|
|
|
async function fetchJson<T>(url: string, init?: RequestInit): Promise<T> {
|
|
const response = await fetch(url, init);
|
|
if (!response.ok) {
|
|
throw new Error(`${response.status} ${await response.text()}`);
|
|
}
|
|
return (await response.json()) as T;
|
|
}
|
|
|
|
interface PublicSession {
|
|
id: string;
|
|
last_event_id: number;
|
|
status: string;
|
|
}
|
|
|
|
interface PublicEvent {
|
|
sequence: number;
|
|
terminal: boolean;
|
|
type: string;
|
|
turn_id?: string | null;
|
|
item_id?: string | null;
|
|
content_part_id?: string | null;
|
|
data?: unknown;
|
|
}
|
|
|
|
function eventText(event: PublicEvent): string {
|
|
const data = event.data;
|
|
if (
|
|
typeof data === 'object' &&
|
|
data !== null &&
|
|
'text' in data &&
|
|
typeof data.text === 'string'
|
|
) {
|
|
return data.text;
|
|
}
|
|
return '';
|
|
}
|
|
|
|
interface PublicList<T> {
|
|
data: T[];
|
|
}
|
|
|
|
// What a reader of the public feed sees right now: retracted prefixes are
|
|
// gone from this view, so it is the reload witness the recovery arms compare
|
|
// against (the live SSE view is not read).
|
|
async function publicTranscript(
|
|
springUrl: string,
|
|
tenant: string,
|
|
sessionId: string,
|
|
): Promise<{ events: PublicEvent[]; text: string; terminals: PublicEvent[] }> {
|
|
const page = await fetchJson<PublicList<PublicEvent>>(
|
|
`${springUrl}/v1/agents/sessions/${sessionId}/events?after=0&limit=100`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
return {
|
|
events: page.data,
|
|
text: page.data
|
|
.filter((event) => event.type === 'item.output_text.delta')
|
|
.map(eventText)
|
|
.join(''),
|
|
terminals: page.data.filter((event) => event.terminal),
|
|
};
|
|
}
|
|
|
|
function deltaSequences(events: PublicEvent[], text: string): number[] {
|
|
const sequences = textDeltas(events, text).map((event) => event.sequence);
|
|
// An empty list would turn Math.min into Infinity and every "before"
|
|
// check into a vacuous pass.
|
|
if (sequences.length === 0) {
|
|
throw new Error(`No public text delta carries ${JSON.stringify(text)}`);
|
|
}
|
|
return sequences;
|
|
}
|
|
|
|
// A retraction blanks the dead owner's deltas in place and then appends
|
|
// stream.reconciled, so "retracted before the new text" is an ordering of
|
|
// sequences, not something a poll can catch in time.
|
|
function reconciledBetween(
|
|
events: PublicEvent[],
|
|
after: number,
|
|
before: number,
|
|
): boolean {
|
|
return events.some(
|
|
(event) =>
|
|
event.type === 'stream.reconciled' &&
|
|
event.sequence > after &&
|
|
event.sequence < before,
|
|
);
|
|
}
|
|
|
|
function tenantHeaders(tenant: string): Record<string, string> {
|
|
return {
|
|
'x-qwen-tenant-id': tenant,
|
|
[trustedActorHeader]: trustedActor,
|
|
};
|
|
}
|
|
|
|
async function waitForTerminal(
|
|
springUrl: string,
|
|
tenant: string,
|
|
sessionId: string,
|
|
after: number,
|
|
child: Child | Child[],
|
|
timeoutMs = 120_000,
|
|
): Promise<{ events: PublicEvent[]; lastSequence: number }> {
|
|
const events: PublicEvent[] = [];
|
|
let cursor = after;
|
|
await waitUntil(
|
|
'Managed Turn terminal event',
|
|
async () => {
|
|
const page = await fetchJson<PublicList<PublicEvent>>(
|
|
`${springUrl}/v1/agents/sessions/${sessionId}/events?after=${cursor}&limit=100`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
for (const event of page.data) {
|
|
events.push(event);
|
|
cursor = Math.max(cursor, event.sequence);
|
|
}
|
|
return page.data.some((event) => event.terminal);
|
|
},
|
|
timeoutMs,
|
|
child,
|
|
);
|
|
return { events, lastSequence: cursor };
|
|
}
|
|
|
|
const failoverFirstMarker = 'MANAGED_SESSION_FAILOVER_FIRST_TURN';
|
|
const failoverSecondMarker = 'MANAGED_SESSION_FAILOVER_SECOND_TURN';
|
|
const bigOutputSource = 'abcdefghijklmnopqrstu长😀'.repeat(
|
|
bigOutput ? 8_000 : 0,
|
|
);
|
|
const failoverFirstResponse = bigOutput
|
|
? Array.from({ length: 48 }, (_, index) => {
|
|
// Distinct prefixes keep the provider's cumulative-stream detection out
|
|
// of this storage regression without changing its character/byte sizes.
|
|
return (
|
|
String(index).padStart(4, '0') +
|
|
bigOutputSource.slice(index * 4_000 + 4, (index + 1) * 4_000)
|
|
);
|
|
}).join('')
|
|
: 'FIRST_TURN_DURABLY_COMMITTED';
|
|
const failoverSecondResponse = bigOutput
|
|
? 'CONTROL_ANSWER__'.repeat(500)
|
|
: 'SECOND_TURN_RESTORED_CONTEXT';
|
|
const failoverMissingResponse = 'SECOND_TURN_CONTEXT_MISSING';
|
|
const inflightMarker = 'MANAGED_SESSION_INFLIGHT_FAILOVER';
|
|
const inflightResponse = 'INFLIGHT_TURN_RECOVERED';
|
|
const continuationMarker = 'MANAGED_SESSION_CONTINUATION_FAILOVER';
|
|
const continuationPartial = 'CONTINUATION_PARTIAL';
|
|
const continuationResponse = 'CONTINUATION_TURN_RECOVERED';
|
|
let acceptReplacementContinuation = false;
|
|
let releaseContinuationHold = () => {};
|
|
const continuationHold = new Promise<void>((resolve) => {
|
|
releaseContinuationHold = resolve;
|
|
});
|
|
// after-result arms a fresh hold after the decision: the continuation hold
|
|
// above is already released by the time the approved call runs.
|
|
let approvalResultHold: Promise<void> | undefined;
|
|
let releaseApprovalResultHold = () => {};
|
|
const modelRoundMarker = 'MANAGED_SESSION_MODEL_ROUND_FAILOVER';
|
|
const modelRoundResponse = 'MODEL_ROUND_TURN_RECOVERED';
|
|
const modelRoundNextMarker = 'MANAGED_SESSION_MODEL_ROUND_NEXT_TURN';
|
|
const modelRoundNextResponse = 'MODEL_ROUND_NEXT_TURN_COMPLETED';
|
|
// One published, held prefix per owner generation that dies inside the first
|
|
// model round: two Harness-only replacements, or one frozen control plane.
|
|
const modelRoundPartials = frozenControlPlane
|
|
? ['MODEL_ROUND_PARTIAL_ONE']
|
|
: ['MODEL_ROUND_PARTIAL_ONE', 'MODEL_ROUND_PARTIAL_TWO'];
|
|
const modelRoundHolds = modelRoundPartials.map(() => {
|
|
let release = () => {};
|
|
const promise = new Promise<void>((resolve) => {
|
|
release = resolve;
|
|
});
|
|
return { promise, release };
|
|
});
|
|
let modelRoundGeneration = 0;
|
|
// The runner advances the round only after the owner holding it is dead or
|
|
// frozen, so a reissue always lands on the next generation's response.
|
|
function advanceModelRound(): void {
|
|
modelRoundHolds[modelRoundGeneration]?.release();
|
|
modelRoundGeneration += 1;
|
|
}
|
|
|
|
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
|
|
// the post-finally throws, so an interrupt can never put a clean pass
|
|
// 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.
|
|
let fencedWriterJson: string | undefined;
|
|
try {
|
|
const harnessToken = randomBytes(24).toString('base64url');
|
|
const brokerToken = randomBytes(24).toString('base64url');
|
|
const credentialKey = randomBytes(32).toString('base64');
|
|
const capabilityDigest = `sha256:${randomBytes(32).toString('hex')}`;
|
|
const tenant = continuationFailover
|
|
? 'managed-continuation-failover-e2e'
|
|
: inflightFailover
|
|
? 'managed-inflight-failover-e2e'
|
|
: bigOutput
|
|
? 'managed-big-output-e2e'
|
|
: sessionFailover
|
|
? 'managed-session-failover-e2e'
|
|
: frozenControlPlane
|
|
? 'managed-frozen-control-plane-e2e'
|
|
: modelRoundFailover
|
|
? 'managed-model-round-failover-e2e'
|
|
: 'real-model-e2e';
|
|
const springArguments = ['-jar', springJar];
|
|
springArguments.push(
|
|
`--qwen.managed-agent.runtime-broker.workspace-mounts[0].tenant-id=${tenant}`,
|
|
`--qwen.managed-agent.runtime-broker.workspace-mounts[0].storage-id=${boundStorageId}`,
|
|
`--qwen.managed-agent.runtime-broker.workspace-mounts[0].root=${workspaceMount}`,
|
|
);
|
|
|
|
if (durableFailover) {
|
|
fake = await startFakeOpenAIServer(({ body }) => {
|
|
const messages = Array.isArray(body['messages']) ? body['messages'] : [];
|
|
const serialized = JSON.stringify(messages);
|
|
// Match the new current input before any marker from restored history.
|
|
if (serialized.includes(reliefMarker)) return { content: reliefResponse };
|
|
if (modelRoundFailover && serialized.includes(modelRoundNextMarker)) {
|
|
return { content: modelRoundNextResponse };
|
|
}
|
|
if (modelRoundFailover && serialized.includes(modelRoundMarker)) {
|
|
const generation = modelRoundGeneration;
|
|
const partial = modelRoundPartials[generation];
|
|
return partial === undefined
|
|
? { content: modelRoundResponse }
|
|
: {
|
|
contentChunks: [partial],
|
|
holdAfterChunks: 1,
|
|
holdUntil: modelRoundHolds[generation].promise,
|
|
};
|
|
}
|
|
|
|
if (continuationFailover && serialized.includes(continuationMarker)) {
|
|
// B3: two earlier tool rounds, then a held post-tool round. The fake
|
|
// model decides by the number of tool results already in the history:
|
|
// 0 -> tool A (round 0), 1 -> tool B (round 1), >=2 -> the held final
|
|
// answer (round 2). The crash lands on the round-2 request so recovery
|
|
// must resume the saved round and cost ledger, not restart them.
|
|
if (approvalFault === 'after-earlier-tools') {
|
|
const toolResultCount = (serialized.match(/"role":"tool"/g) ?? [])
|
|
.length;
|
|
// Non-zero usage makes the token ledger numerically observable (the
|
|
// fake's default usage is all zeros). The earlier rounds outweigh the
|
|
// final round so a zeroed restore totals less than the crash-boundary
|
|
// ledger, not just the same sum re-counted.
|
|
if (toolResultCount === 0) {
|
|
return {
|
|
toolCalls: [
|
|
fakeToolCall(
|
|
'write_file',
|
|
{
|
|
file_path: inflightSideEffectName,
|
|
content: inflightSideEffectContent,
|
|
},
|
|
'call_managed_b3_first',
|
|
),
|
|
],
|
|
usage: {
|
|
prompt_tokens: 0,
|
|
completion_tokens: 300,
|
|
total_tokens: 300,
|
|
},
|
|
};
|
|
}
|
|
if (toolResultCount === 1) {
|
|
return {
|
|
toolCalls: [
|
|
fakeToolCall(
|
|
'write_file',
|
|
{
|
|
file_path: b3SecondSideEffectName,
|
|
content: b3SecondSideEffectContent,
|
|
},
|
|
'call_managed_b3_second',
|
|
),
|
|
],
|
|
usage: {
|
|
prompt_tokens: 0,
|
|
completion_tokens: 300,
|
|
total_tokens: 300,
|
|
},
|
|
};
|
|
}
|
|
if (approvalResultHold) {
|
|
return {
|
|
contentChunks: [continuationPartial],
|
|
holdAfterChunks: 1,
|
|
holdUntil: approvalResultHold,
|
|
usage: {
|
|
prompt_tokens: 0,
|
|
completion_tokens: 100,
|
|
total_tokens: 100,
|
|
},
|
|
};
|
|
}
|
|
return {
|
|
content: continuationResponse,
|
|
usage: {
|
|
prompt_tokens: 0,
|
|
completion_tokens: 100,
|
|
total_tokens: 100,
|
|
},
|
|
};
|
|
}
|
|
if (!serialized.includes('"role":"tool"')) {
|
|
// A3: a two-call batch so the replacement can be crashed in the
|
|
// approval gap between a decided sibling and a pending one.
|
|
if (approvalFault === 'sibling-approval') {
|
|
return {
|
|
toolCalls: [
|
|
fakeToolCall(
|
|
'write_file',
|
|
{
|
|
file_path: inflightSideEffectName,
|
|
content: inflightSideEffectContent,
|
|
},
|
|
'call_managed_continuation_failover',
|
|
),
|
|
fakeToolCall(
|
|
'write_file',
|
|
{
|
|
file_path: siblingSideEffectName,
|
|
content: siblingSideEffectContent,
|
|
},
|
|
'call_managed_continuation_sibling',
|
|
),
|
|
],
|
|
};
|
|
}
|
|
return {
|
|
toolCalls: [
|
|
fakeToolCall(
|
|
'write_file',
|
|
{
|
|
file_path: inflightSideEffectName,
|
|
content: inflightSideEffectContent,
|
|
},
|
|
'call_managed_continuation_failover',
|
|
),
|
|
],
|
|
};
|
|
}
|
|
if (approvalRecovery) {
|
|
if (approvalFault === 'after-result' && approvalResultHold) {
|
|
// Publish a prefix and park the round so the crash lands after
|
|
// the execution result commits but before the assistant output
|
|
// commits; the reissue must reuse this exact saved request.
|
|
return {
|
|
contentChunks: [continuationPartial],
|
|
holdAfterChunks: 1,
|
|
holdUntil: approvalResultHold,
|
|
};
|
|
}
|
|
return { content: continuationResponse };
|
|
}
|
|
if (!acceptReplacementContinuation) {
|
|
return {
|
|
contentChunks: [continuationPartial],
|
|
holdAfterChunks: 1,
|
|
holdUntil: continuationHold,
|
|
};
|
|
}
|
|
return { content: continuationResponse };
|
|
}
|
|
if (inflightFailover && serialized.includes(inflightMarker)) {
|
|
if (!serialized.includes('"role":"tool"')) {
|
|
return {
|
|
toolCalls: [
|
|
fakeToolCall(
|
|
'write_file',
|
|
{
|
|
file_path: inflightSideEffectName,
|
|
content: inflightSideEffectContent,
|
|
},
|
|
'call_managed_inflight_failover',
|
|
),
|
|
],
|
|
};
|
|
}
|
|
return { content: inflightResponse };
|
|
}
|
|
if (serialized.includes(failoverSecondMarker)) {
|
|
const restored =
|
|
serialized.includes(failoverFirstMarker) &&
|
|
serialized.includes(failoverFirstResponse);
|
|
return {
|
|
content: restored ? failoverSecondResponse : failoverMissingResponse,
|
|
};
|
|
}
|
|
if (serialized.includes(failoverFirstMarker)) {
|
|
return bigOutput
|
|
? {
|
|
contentChunks: Array.from({ length: 48 }, (_, index) =>
|
|
failoverFirstResponse.slice(index * 4_000, (index + 1) * 4_000),
|
|
),
|
|
}
|
|
: { content: failoverFirstResponse };
|
|
}
|
|
return { content: 'UNEXPECTED_FAILOVER_PROMPT' };
|
|
});
|
|
}
|
|
const fakeBaseUrl = fake?.baseUrl;
|
|
if (durableFailover && fakeBaseUrl === undefined) {
|
|
throw new Error('Fake model server did not start');
|
|
}
|
|
|
|
const mysqlPort = await freePort();
|
|
dumpPort = mysqlPort;
|
|
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,
|
|
[
|
|
'--no-defaults',
|
|
...mysqldUserArguments,
|
|
'--initialize-insecure',
|
|
`--datadir=${mysqlData}`,
|
|
],
|
|
{ encoding: 'utf8' },
|
|
);
|
|
if (initialized.status !== 0) {
|
|
throw new Error(`MySQL initialization failed: ${initialized.stderr}`);
|
|
}
|
|
const mysqlServer = start(
|
|
mysqld,
|
|
[
|
|
'--no-defaults',
|
|
...mysqldUserArguments,
|
|
`--datadir=${mysqlData}`,
|
|
`--socket=${mysqlSocket}`,
|
|
`--port=${mysqlPort}`,
|
|
'--bind-address=127.0.0.1',
|
|
'--mysqlx=0',
|
|
`--pid-file=${path.join(temporary, 'mysql.pid')}`,
|
|
`--log-error=${mysqlError}`,
|
|
],
|
|
{},
|
|
'MySQL',
|
|
);
|
|
await waitUntil(
|
|
'MySQL',
|
|
// spawnSync blocks the event loop, so the race deadline cannot fire
|
|
// inside one iteration: the probe carries its own timeout.
|
|
() =>
|
|
spawnSync(
|
|
mysqladmin,
|
|
[
|
|
'--no-defaults',
|
|
'--protocol=tcp',
|
|
'--host=127.0.0.1',
|
|
`--port=${mysqlPort}`,
|
|
'--user=root',
|
|
'ping',
|
|
],
|
|
{
|
|
stdio: 'ignore',
|
|
env: mysqlClientEnv,
|
|
timeout: 10_000,
|
|
},
|
|
).status === 0,
|
|
60_000,
|
|
mysqlServer,
|
|
);
|
|
runMysql(
|
|
mysqlPort,
|
|
'CREATE DATABASE qwen_managed_agent CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci',
|
|
);
|
|
|
|
const spring = start(
|
|
java,
|
|
springArguments,
|
|
{
|
|
env: {
|
|
...cleanEnvironment,
|
|
HOME: runtimeHome,
|
|
LANG: process.env['LANG'] ?? 'C',
|
|
LC_ALL: process.env['LC_ALL'] ?? 'C',
|
|
NO_PROXY: '127.0.0.1,localhost',
|
|
QWEN_HOME: path.join(runtimeHome, '.qwen'),
|
|
TMPDIR: temporary,
|
|
no_proxy: '127.0.0.1,localhost',
|
|
SERVER_PORT: String(springPort),
|
|
SPRING_DATASOURCE_PASSWORD: '',
|
|
SPRING_DATASOURCE_URL: `jdbc:mysql://127.0.0.1:${mysqlPort}/qwen_managed_agent?useSSL=false&allowPublicKeyRetrieval=true`,
|
|
SPRING_DATASOURCE_USERNAME: 'root',
|
|
QWEN_MANAGED_AGENT_APPROVAL_MODE: approvalRecovery ? 'default' : 'yolo',
|
|
QWEN_MANAGED_AGENT_APPROVAL_TIMEOUT:
|
|
verificationMode === 'expiry' ? '30s' : '5m',
|
|
QWEN_MANAGED_AGENT_CAPABILITY_DIGEST: capabilityDigest,
|
|
QWEN_MANAGED_AGENT_HARNESS_BASE_URL: `http://127.0.0.1:${harnessPort}`,
|
|
QWEN_MANAGED_AGENT_HARNESS_ENABLED: 'true',
|
|
QWEN_MANAGED_AGENT_HARNESS_REQUEST_TIMEOUT: '120s',
|
|
QWEN_MANAGED_AGENT_HARNESS_TOKEN: harnessToken,
|
|
// Trusted reboot recovery stays pinned off in every runner mode;
|
|
// durable local process follows runtimeTakeover. Both pins keep each
|
|
// mode's previously verified behavior and keep the runner starting
|
|
// off Linux.
|
|
QWEN_MANAGED_AGENT_RUNTIME_TRUSTED_LOCAL_REBOOT_RECOVERY: 'false',
|
|
...(registerStorage
|
|
? {
|
|
QWEN_MANAGED_AGENT_RUNTIME_VERIFIED_WORKSPACE_RECOVERY_ENABLED:
|
|
'true',
|
|
}
|
|
: {}),
|
|
QWEN_MANAGED_AGENT_TRUSTED_ACTOR_HEADER: trustedActorHeader,
|
|
QWEN_MANAGED_AGENT_WORKSPACE_FILES_ENABLED: 'true',
|
|
...(runtimeTakeover
|
|
? {
|
|
QWEN_MANAGED_AGENT_RUNTIME_DURABLE_LOCAL_PROCESS:
|
|
process.platform === 'linux' ? 'true' : 'false',
|
|
}
|
|
: {
|
|
QWEN_MANAGED_AGENT_RUNTIME_DURABLE_LOCAL_PROCESS: 'false',
|
|
}),
|
|
...(durableFailover
|
|
? {
|
|
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: sessionStoreUrl,
|
|
QWEN_MANAGED_AGENT_SESSION_STORE_ENABLED: 'true',
|
|
QWEN_MANAGED_AGENT_SESSION_STORE_WRITER_LEASE_DURATION: '5s',
|
|
QWEN_MANAGED_AGENT_WORKSPACE_ID: workspaceId,
|
|
QWEN_MANAGED_AGENT_RUNTIME_BROKER_ENABLED: 'true',
|
|
QWEN_MANAGED_AGENT_RUNTIME_BROKER_PORT: String(brokerPort),
|
|
QWEN_MANAGED_AGENT_RUNTIME_BROKER_TOKEN: brokerToken,
|
|
QWEN_MANAGED_AGENT_RUNTIME_CREDENTIAL_KEY: credentialKey,
|
|
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_NODE_EXECUTABLE: process.execPath,
|
|
QWEN_MANAGED_AGENT_WORKSPACE_CWD: workspace,
|
|
}
|
|
: {
|
|
QWEN_MANAGED_AGENT_RUNTIME_BROKER_ENABLED: 'true',
|
|
QWEN_MANAGED_AGENT_RUNTIME_BROKER_PORT: String(brokerPort),
|
|
QWEN_MANAGED_AGENT_RUNTIME_BROKER_TOKEN: brokerToken,
|
|
QWEN_MANAGED_AGENT_RUNTIME_CREDENTIAL_KEY: credentialKey,
|
|
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: sessionStoreUrl,
|
|
QWEN_MANAGED_AGENT_SESSION_STORE_ENABLED: 'true',
|
|
QWEN_MANAGED_AGENT_SESSION_STORE_WRITER_LEASE_DURATION: '60s',
|
|
QWEN_MANAGED_AGENT_WORKSPACE_ID: workspaceId,
|
|
QWEN_MANAGED_AGENT_NODE_EXECUTABLE:
|
|
runtimeDelayMs === 0 ? process.execPath : delayedNode,
|
|
QWEN_MANAGED_AGENT_WORKSPACE_CWD: workspace,
|
|
}),
|
|
},
|
|
},
|
|
'Spring Managed Agent Server',
|
|
);
|
|
const springUrl = `http://127.0.0.1:${springPort}`;
|
|
await waitUntil(
|
|
'Spring Managed Agent Server',
|
|
async () => {
|
|
const response = await fetch(`${springUrl}/actuator/health`);
|
|
return response.ok;
|
|
},
|
|
60_000,
|
|
spring,
|
|
);
|
|
runMysql(
|
|
mysqlPort,
|
|
`INSERT INTO qwen_managed_agent.ma_workspace (tenant_id, workspace_id, workspace_generation, storage_id, display_name, config_ref, policy_ref, state) VALUES (${sqlString(tenant)}, ${sqlString(boundWorkspaceId)}, 1, ${sqlString(boundStorageId)}, 'E2E', 'managed-runtime-tools/1', 'preapproved-workspace-tools/1', 'ACTIVE')`,
|
|
);
|
|
runMysql(
|
|
mysqlPort,
|
|
`INSERT INTO qwen_managed_agent.ma_workspace_access (tenant_id, workspace_id, actor_id, role) VALUES (${sqlString(tenant)}, ${sqlString(boundWorkspaceId)}, ${sqlString(trustedActor)}, 'OPERATOR')`,
|
|
);
|
|
if (registerStorage) {
|
|
// The new root's mtime must differ from its birth time; no Harness or tool can write it yet.
|
|
const touch = path.join(workspaceMount, '.storage-registration-touch');
|
|
mkdirSync(touch);
|
|
rmSync(touch, { recursive: true });
|
|
const registration = spawnSync(
|
|
java,
|
|
[
|
|
'-cp',
|
|
springJar,
|
|
'-Dloader.main=com.alibaba.qwen.code.managedagent.maintenance.WorkspaceStorageRegistrationMain',
|
|
'org.springframework.boot.loader.launch.PropertiesLauncher',
|
|
'register',
|
|
tenant,
|
|
boundStorageId,
|
|
workspaceMount,
|
|
randomUUID(),
|
|
'--offline-confirmed',
|
|
],
|
|
{
|
|
encoding: 'utf8',
|
|
timeout: 60_000,
|
|
env: {
|
|
...cleanEnvironment,
|
|
W1_JDBC_URL: `jdbc:mysql://127.0.0.1:${mysqlPort}/qwen_managed_agent?useSSL=false&allowPublicKeyRetrieval=true`,
|
|
W1_JDBC_USER: 'root',
|
|
W1_JDBC_PASSWORD: '',
|
|
},
|
|
},
|
|
);
|
|
if (registration.error || registration.status !== 0) {
|
|
throw new Error(
|
|
`Native storage registration failed: ${registration.stderr || registration.error?.message || registration.status}`,
|
|
);
|
|
}
|
|
}
|
|
if (inflightFailover) {
|
|
heldStartProxy = await startHeldExecutionStartProxy(
|
|
`http://127.0.0.1:${brokerPort}`,
|
|
);
|
|
}
|
|
|
|
const harness = start(
|
|
process.execPath,
|
|
[
|
|
cliBundle,
|
|
'serve',
|
|
'--profile',
|
|
'hosted-harness',
|
|
'--port',
|
|
String(harnessPort),
|
|
'--hostname',
|
|
'127.0.0.1',
|
|
'--require-auth',
|
|
'--no-web',
|
|
'--workspace',
|
|
workspace,
|
|
'--managed-runtime-broker-url',
|
|
heldStartProxy?.baseUrl ?? `http://127.0.0.1:${brokerPort}`,
|
|
// Joined form: a base64url token can start with '-', which argv
|
|
// would otherwise parse as another flag.
|
|
`--managed-runtime-broker-token=${brokerToken}`,
|
|
],
|
|
{
|
|
env: {
|
|
...cleanEnvironment,
|
|
HOME: harnessHome,
|
|
LANG: process.env['LANG'] ?? 'C',
|
|
LC_ALL: process.env['LC_ALL'] ?? 'C',
|
|
QWEN_HOME: path.join(harnessHome, '.qwen'),
|
|
QWEN_CODE_TRUSTED_FOLDERS_PATH: trustedFolders,
|
|
QWEN_HOSTED_HARNESS_CAPABILITY_DIGEST: capabilityDigest,
|
|
QWEN_SERVER_TOKEN: harnessToken,
|
|
...(durableFailover
|
|
? {
|
|
OPENAI_API_KEY: 'fake-key',
|
|
OPENAI_BASE_URL: fakeBaseUrl,
|
|
OPENAI_MODEL: 'fake-model',
|
|
QWEN_MODEL: 'fake-model',
|
|
}
|
|
: {}),
|
|
QWEN_RUNTIME_BROKER_TOKEN: brokerToken,
|
|
QWEN_RUNTIME_BROKER_URL:
|
|
heldStartProxy?.baseUrl ?? `http://127.0.0.1:${brokerPort}`,
|
|
},
|
|
},
|
|
hostedHarnessLabel,
|
|
);
|
|
await waitUntil(
|
|
'Hosted Harness',
|
|
async () => {
|
|
const response = await fetch(
|
|
`http://127.0.0.1:${harnessPort}/health?deep=1`,
|
|
{
|
|
headers: { authorization: `Bearer ${harnessToken}` },
|
|
},
|
|
);
|
|
return response.ok;
|
|
},
|
|
60_000,
|
|
harness,
|
|
);
|
|
|
|
if (durableFailover) {
|
|
const createResponse = await fetch(`${springUrl}/v1/agents/sessions`, {
|
|
method: 'POST',
|
|
headers: {
|
|
'content-type': 'application/json',
|
|
'idempotency-key': `failover-create-${Date.now()}`,
|
|
...tenantHeaders(tenant),
|
|
},
|
|
body: JSON.stringify({
|
|
agent_id: 'qwen-code',
|
|
input: [
|
|
{
|
|
type: 'text',
|
|
text: continuationFailover
|
|
? approvalFault === 'after-earlier-tools'
|
|
? `${continuationMarker}. ${b3BudgetDirective} Execute the requested tools and reply exactly ${continuationResponse}.`
|
|
: `${continuationMarker}. Execute the requested tool once and reply exactly ${continuationResponse}.`
|
|
: inflightFailover
|
|
? `${inflightMarker}. Execute the requested tool once and reply exactly ${inflightResponse}.`
|
|
: modelRoundFailover
|
|
? `${modelRoundMarker}. Use no tools and reply exactly ${modelRoundResponse}.`
|
|
: bigOutput
|
|
? `${failoverFirstMarker}. Produce the configured long answer.`
|
|
: `${failoverFirstMarker}. Reply exactly ${failoverFirstResponse}.`,
|
|
},
|
|
],
|
|
...(workspaceTurns
|
|
? { workspace: { workspace_id: boundWorkspaceId } }
|
|
: {}),
|
|
metadata: {
|
|
title: continuationFailover
|
|
? 'Managed continuation owner failover E2E'
|
|
: inflightFailover
|
|
? 'Managed in-flight owner failover E2E'
|
|
: frozenControlPlane
|
|
? 'Managed frozen control-plane fencing E2E'
|
|
: modelRoundFailover
|
|
? 'Managed model-round owner failover E2E'
|
|
: bigOutput
|
|
? 'Managed long-answer persistence E2E'
|
|
: 'Managed Session owner failover E2E',
|
|
},
|
|
}),
|
|
});
|
|
if (createResponse.status !== 202) {
|
|
throw new Error(
|
|
`Failover Session create returned ${createResponse.status}: ${await createResponse.text()}`,
|
|
);
|
|
}
|
|
const session = (await createResponse.json()) as PublicSession;
|
|
let firstTurnLastSequence = 0;
|
|
let formerDispatchOwner = '';
|
|
let firstPrefixSequence = 0;
|
|
// B2: the live SSE capture stays open across both Harness crashes; the
|
|
// second prefix's sequence is hoisted so the audit can check its blanking.
|
|
let b2LiveCapture: Promise<B2SseCapture> | undefined;
|
|
let b2SecondPrefixSequence = 0;
|
|
let b2AttemptIds: string[] = [];
|
|
let heldExecutionStartPath: string | undefined;
|
|
let originalExecutionCallId: string | undefined;
|
|
let originalRuntimeSessionId: string | undefined;
|
|
// The runtime_session_id column is written once at INSERT and never
|
|
// rewritten, so an equality pin cannot fail. The harness-only arms
|
|
// therefore also compare the row's own liveness metric before and
|
|
// after the crash: a surviving worker's re-attach renews it.
|
|
function runtimeSessionHeartbeat(
|
|
runtimeSessionId: string,
|
|
): [string, number] {
|
|
const row = runMysql(
|
|
mysqlPort,
|
|
`SELECT last_active_at, record_version FROM qwen_managed_agent.ma_runtime_session WHERE runtime_session_id = ${sqlString(runtimeSessionId)}`,
|
|
).split('\t');
|
|
if (row.length !== 2 || row[1].length === 0)
|
|
throw new Error(`Runtime session row missing (${runtimeSessionId})`);
|
|
return [row[0], Number(row[1])];
|
|
}
|
|
let firstRuntimeHeartbeat: [string, number] | undefined;
|
|
if (inflightFailover) {
|
|
if (heldStartProxy === undefined) {
|
|
throw new Error('In-flight failover did not start its Broker proxy');
|
|
}
|
|
try {
|
|
await waitUntil(
|
|
'Runtime execution start boundary',
|
|
() => {
|
|
const observations = heldStartProxy?.observations() ?? [];
|
|
return (
|
|
heldStartProxy?.heldPath() !== undefined ||
|
|
observations.some((observation) =>
|
|
/ [45]\d\d(?: |$)/.test(observation),
|
|
)
|
|
);
|
|
},
|
|
120_000,
|
|
harness,
|
|
);
|
|
} catch (error) {
|
|
throw new Error(
|
|
`Runtime execution start boundary failed; proxy=${heldStartProxy.observations().join(' | ') || 'no requests'}`,
|
|
{ cause: error },
|
|
);
|
|
}
|
|
heldExecutionStartPath = heldStartProxy.heldPath();
|
|
if (heldExecutionStartPath === undefined) {
|
|
throw new Error(
|
|
`Runtime execution start boundary failed; proxy=${heldStartProxy.observations().join(' | ') || 'no requests'}`,
|
|
);
|
|
}
|
|
} else if (approvalRecovery) {
|
|
await waitUntil(
|
|
'Original requested approval',
|
|
async () => {
|
|
const page = await fetchJson<PublicList<Record<string, unknown>>>(
|
|
`${springUrl}/v1/agents/sessions/${session.id}/actions?limit=100`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
const requested = page.data.filter(
|
|
(action) => action.state === 'requested',
|
|
);
|
|
if (requested.length !== 1) return false;
|
|
originalApprovalId = String(requested[0].id);
|
|
return originalApprovalId.startsWith('tool_approval_');
|
|
},
|
|
120_000,
|
|
harness,
|
|
);
|
|
const actionFilter = `tenant_id=${sqlString(tenant)} AND session_id=${sqlString(session.id)} AND record_kind='ACTION' AND action_id=${sqlString(originalApprovalId!)}`;
|
|
originalApprovalOptions = JSON.parse(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT options_json FROM qwen_managed_agent.ma_session_record WHERE ${actionFilter}`,
|
|
),
|
|
);
|
|
if (
|
|
originalApprovalOptions?.v !== 3 ||
|
|
!originalApprovalOptions.continuationRef ||
|
|
!Number.isSafeInteger(originalApprovalOptions.createdAt) ||
|
|
!Number.isSafeInteger(originalApprovalOptions.expiresAt) ||
|
|
originalApprovalOptions.expiresAt <= originalApprovalOptions.createdAt
|
|
) {
|
|
throw new Error(
|
|
`New approval omitted the recovery continuation or its deadline: version=${originalApprovalOptions?.v} createdAt=${originalApprovalOptions?.createdAt} expiresAt=${originalApprovalOptions?.expiresAt}`,
|
|
);
|
|
}
|
|
originalApprovalPlan = JSON.parse(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT CONVERT(inline_bytes USING utf8mb4) FROM qwen_managed_agent.ma_resource WHERE tenant_id=${sqlString(tenant)} AND session_id=${sqlString(session.id)} AND resource_id=${sqlString(originalApprovalOptions.continuationRef.resourceId)}`,
|
|
),
|
|
);
|
|
const expectedOriginalPlanCalls =
|
|
approvalFault === 'sibling-approval' ? 2 : 1;
|
|
if (
|
|
!originalApprovalPlan ||
|
|
originalApprovalPlan.calls?.length !== expectedOriginalPlanCalls ||
|
|
originalApprovalPlan.actionId !== originalApprovalId ||
|
|
originalApprovalPlan.stage !== 'approval' ||
|
|
(approvalFault === 'sibling-approval' &&
|
|
originalApprovalPlan.approvalOrdinal !== 0)
|
|
) {
|
|
throw new Error(
|
|
`Original saved approval plan does not identify the original ${expectedOriginalPlanCalls} call(s) at the first approval`,
|
|
);
|
|
}
|
|
originalRuntimeSessionId = originalApprovalPlan.runtime.runtimeSessionId;
|
|
const runtime = runMysql(
|
|
mysqlPort,
|
|
`SELECT binding_id, runtime_generation, workspace_generation, session_state FROM qwen_managed_agent.ma_runtime_session WHERE runtime_session_id=${sqlString(originalRuntimeSessionId!)}`,
|
|
).split('\t');
|
|
if (
|
|
runtime[0] !== originalApprovalPlan.runtime.bindingId ||
|
|
runtime[1] !== originalApprovalPlan.runtime.generation ||
|
|
runtime[2] !== originalApprovalPlan.runtime.workspaceGeneration ||
|
|
runtime[3] !== 'READY'
|
|
) {
|
|
throw new Error(
|
|
`Saved approval Runtime binding differs from the real Broker: ${runtime.join(',')}`,
|
|
);
|
|
}
|
|
firstRuntimeHeartbeat = runtimeSessionHeartbeat(
|
|
originalRuntimeSessionId!,
|
|
);
|
|
await waitUntil(
|
|
'Original durable await_action checkpoint',
|
|
() => {
|
|
const id = runMysql(
|
|
mysqlPort,
|
|
`SELECT latest_checkpoint_resource_id FROM qwen_managed_agent.ma_session_head WHERE tenant_id=${sqlString(tenant)} AND session_id=${sqlString(session.id)}`,
|
|
);
|
|
if (!id) return false;
|
|
const bytes = runMysql(
|
|
mysqlPort,
|
|
`SELECT CONVERT(inline_bytes USING utf8mb4) FROM qwen_managed_agent.ma_resource WHERE tenant_id=${sqlString(tenant)} AND session_id=${sqlString(session.id)} AND resource_id=${sqlString(id)}`,
|
|
);
|
|
if (!bytes) return false;
|
|
const checkpoint = JSON.parse(bytes);
|
|
return (
|
|
checkpoint.continuation?.phase === 'await_action' &&
|
|
checkpoint.approval?.requestId === originalApprovalId
|
|
);
|
|
},
|
|
10_000,
|
|
harness,
|
|
);
|
|
const privateHead = runMysql(
|
|
mysqlPort,
|
|
`SELECT storage_version, latest_checkpoint_resource_id FROM qwen_managed_agent.ma_session_head WHERE tenant_id=${sqlString(tenant)} AND session_id=${sqlString(session.id)}`,
|
|
).split('\t');
|
|
const checkpoint = JSON.parse(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT CONVERT(inline_bytes USING utf8mb4) FROM qwen_managed_agent.ma_resource WHERE tenant_id=${sqlString(tenant)} AND session_id=${sqlString(session.id)} AND resource_id=${sqlString(privateHead[1])}`,
|
|
),
|
|
);
|
|
if (
|
|
privateHead[0] !== '2' ||
|
|
checkpoint.continuation?.phase !== 'await_action' ||
|
|
checkpoint.approval?.requestId !== originalApprovalId
|
|
) {
|
|
throw new Error(
|
|
'Approval snapshot was not committed before the crash barrier',
|
|
);
|
|
}
|
|
const executionCount = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
'SELECT COUNT(*) FROM qwen_managed_agent.ma_tool_execution',
|
|
),
|
|
);
|
|
if (executionCount !== 0 || existsSync(inflightSideEffect))
|
|
throw new Error('Tool ran before the original approval');
|
|
saveEvidence('original-approval.json', {
|
|
actionId: originalApprovalId,
|
|
options: originalApprovalOptions,
|
|
plan: originalApprovalPlan,
|
|
originalRuntime: runtime,
|
|
storageVersion: Number(privateHead[0]),
|
|
checkpointPhase: checkpoint.continuation.phase,
|
|
executionCount,
|
|
physicalFileExists: existsSync(inflightSideEffect),
|
|
});
|
|
} else if (continuationFailover) {
|
|
await waitUntil(
|
|
'Continuation partial text',
|
|
async () => {
|
|
const page = await fetchJson<PublicList<PublicEvent>>(
|
|
`${springUrl}/v1/agents/sessions/${session.id}/events?after=0&limit=100`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
return page.data.some(
|
|
(event) =>
|
|
event.type === 'item.output_text.delta' &&
|
|
eventText(event).includes(continuationPartial),
|
|
);
|
|
},
|
|
120_000,
|
|
harness,
|
|
);
|
|
const execution = runMysql(
|
|
mysqlPort,
|
|
'SELECT execution_call_id, execution_state, dispatch_generation, runtime_session_id FROM qwen_managed_agent.ma_tool_execution',
|
|
).split('\t');
|
|
const sideEffectBytes = existsSync(inflightSideEffect)
|
|
? readFileSync(inflightSideEffect, 'utf8')
|
|
: '';
|
|
if (
|
|
execution.length !== 4 ||
|
|
execution[0]?.length === 0 ||
|
|
execution[1] !== 'SETTLED' ||
|
|
execution[2] !== '1' ||
|
|
execution[3]?.length === 0 ||
|
|
sideEffectBytes !== inflightSideEffectContent
|
|
) {
|
|
throw new Error(
|
|
`Continuation boundary was not durable: execution=${execution.join(',')} sideEffect=${JSON.stringify(sideEffectBytes)}`,
|
|
);
|
|
}
|
|
originalExecutionCallId = execution[0];
|
|
originalRuntimeSessionId = execution[3];
|
|
if (harnessOnly)
|
|
firstRuntimeHeartbeat = runtimeSessionHeartbeat(
|
|
originalRuntimeSessionId,
|
|
);
|
|
} else if (modelRoundFailover) {
|
|
// The crash point is the first model round of a Session with no
|
|
// Workspace: the provider already streamed a prefix the public feed
|
|
// shows, and nothing but the saved model request can continue it.
|
|
if (harnessOnly) {
|
|
// B2: subscribe to the live public view before the first prefix. The
|
|
// stream is control-plane-owned, so this one connection survives both
|
|
// Harness crashes; Spring reconciles from after=0 on connect, so a
|
|
// slightly late subscribe still sees every frame. The terminal is a
|
|
// progress signal; the drain window keeps reading so a double
|
|
// projection is observed (R1-2).
|
|
b2LiveCapture = readPublicSse(springUrl, session.id, {
|
|
tenantId: tenant,
|
|
after: 0,
|
|
until: (events) => events.some((event) => event.terminal),
|
|
timeoutMs: 240_000,
|
|
});
|
|
// Awaited only at the audit; without this observer a capture that
|
|
// fails — or that the teardown aborts on an early arm failure — is an
|
|
// unhandled rejection that exits before the children are stopped
|
|
// (R1-1). The audit's await still sees the real rejection.
|
|
b2LiveCapture.catch(() => undefined);
|
|
}
|
|
await waitUntil(
|
|
'First model round partial text',
|
|
async () =>
|
|
(await publicTranscript(springUrl, tenant, session.id)).text ===
|
|
modelRoundPartials[0],
|
|
120_000,
|
|
harness,
|
|
);
|
|
firstPrefixSequence = Math.min(
|
|
...deltaSequences(
|
|
(await publicTranscript(springUrl, tenant, session.id)).events,
|
|
modelRoundPartials[0],
|
|
),
|
|
);
|
|
} else {
|
|
const firstTurn = await waitForTerminal(
|
|
springUrl,
|
|
tenant,
|
|
session.id,
|
|
0,
|
|
spring,
|
|
);
|
|
firstTurnLastSequence = firstTurn.lastSequence;
|
|
const firstTerminal = firstTurn.events.find((event) => event.terminal);
|
|
if (firstTerminal?.type !== 'turn.completed') {
|
|
const storeHead = runMysql(
|
|
mysqlPort,
|
|
`SELECT state, writer_generation, writer_id, writer_lease_until, ${databaseEpochMillisSql} AS observed_now, journal_revision, committed_sequence FROM qwen_managed_agent.ma_session_head WHERE tenant_id=${sqlString(tenant)} AND session_id=${sqlString(session.id)}`,
|
|
);
|
|
throw new Error(
|
|
`First failover Turn ended with ${firstTerminal?.type ?? 'no terminal event'}; store head=${storeHead || 'missing'}`,
|
|
);
|
|
}
|
|
if (bigOutput) {
|
|
const modelRequests = fake?.requests.filter(
|
|
({ body }) => body['stream'] === true,
|
|
);
|
|
if (modelRequests?.length !== 1)
|
|
throw new Error('The long answer must complete without model retry');
|
|
assertStoredAnswer(
|
|
mysqlPort,
|
|
tenant,
|
|
session.id,
|
|
failoverFirstResponse,
|
|
);
|
|
assertPublicAnswer(firstTurn.events, failoverFirstResponse);
|
|
}
|
|
}
|
|
|
|
const sessionFilter = `tenant_id=${sqlString(tenant)} AND session_id=${sqlString(session.id)}`;
|
|
const firstBootId = runMysql(
|
|
mysqlPort,
|
|
`SELECT harness_boot_id FROM qwen_managed_agent.ma_session WHERE ${sessionFilter}`,
|
|
);
|
|
const firstHead = runMysql(
|
|
mysqlPort,
|
|
`SELECT writer_generation, journal_revision, committed_sequence FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter}`,
|
|
)
|
|
.split('\t')
|
|
.map(Number);
|
|
if (
|
|
firstBootId.length === 0 ||
|
|
firstHead.length !== 3 ||
|
|
firstHead.some((value) => !Number.isSafeInteger(value) || value < 1)
|
|
) {
|
|
throw new Error(
|
|
`First owner did not commit a durable private Session: boot=${firstBootId} head=${firstHead.join(',')}`,
|
|
);
|
|
}
|
|
if (inflightFailover) {
|
|
const execution = runMysql(
|
|
mysqlPort,
|
|
'SELECT execution_call_id, execution_state, dispatch_generation, runtime_session_id FROM qwen_managed_agent.ma_tool_execution',
|
|
).split('\t');
|
|
const latestCheckpoint = runMysql(
|
|
mysqlPort,
|
|
`SELECT latest_checkpoint_resource_id FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter}`,
|
|
);
|
|
if (
|
|
execution.length !== 4 ||
|
|
execution[0]?.length === 0 ||
|
|
execution[1] !== 'PREPARED' ||
|
|
execution[2] !== '0' ||
|
|
execution[3]?.length === 0 ||
|
|
latestCheckpoint.length === 0 ||
|
|
!heldExecutionStartPath?.includes(
|
|
encodeURIComponent(execution[0] ?? ''),
|
|
) ||
|
|
existsSync(inflightSideEffect)
|
|
) {
|
|
throw new Error(
|
|
`In-flight boundary was not durable before start: execution=${execution.join(',')} checkpoint=${latestCheckpoint || 'missing'} sideEffect=${existsSync(inflightSideEffect)}`,
|
|
);
|
|
}
|
|
originalExecutionCallId = execution[0];
|
|
originalRuntimeSessionId = execution[3];
|
|
if (harnessOnly)
|
|
firstRuntimeHeartbeat = runtimeSessionHeartbeat(
|
|
originalRuntimeSessionId,
|
|
);
|
|
}
|
|
|
|
if (freeze) {
|
|
// Freeze the writer side only: stopping the original Harness (the
|
|
// lease holder) fences it against mutation on wake. Assert liveness
|
|
// first — a setup that already died would make the freeze a no-op
|
|
// and every later wake assertion vacuous.
|
|
if (!processTreeExists(harness.child)) {
|
|
throw new Error(
|
|
'The original Harness died before the freeze: cannot freeze a dead owner',
|
|
);
|
|
}
|
|
// The original Spring must actually die — reclaiming its workspace
|
|
// binding requires death evidence from /proc liveness, and a
|
|
// SIGSTOPped JVM still reads as alive there.
|
|
await signalProcessTree(harness.child, 'SIGSTOP');
|
|
await crashProcess(
|
|
spring.child,
|
|
'Spring Managed Agent Server (original)',
|
|
);
|
|
} else if (frozenControlPlane) {
|
|
// Unlike --freeze, the former Spring survives too: no Workspace is
|
|
// bound, so no Runtime reclaim needs its death. Its claim is the stale
|
|
// authority the wake must not be able to use.
|
|
formerDispatchOwner = runMysql(
|
|
mysqlPort,
|
|
`SELECT dispatch_owner FROM qwen_managed_agent.ma_turn WHERE ${sessionFilter}`,
|
|
);
|
|
if (!/^[0-9a-f-]{36}$/.test(formerDispatchOwner)) {
|
|
throw new Error(
|
|
`The original control plane held no dispatch claim to freeze: owner=${formerDispatchOwner || 'none'}`,
|
|
);
|
|
}
|
|
const formerPair = [
|
|
[harness, `${hostedHarnessLabel} (original)`],
|
|
[spring, 'Spring Managed Agent Server (original)'],
|
|
] as const;
|
|
for (const [child, name] of formerPair) {
|
|
if (!processTreeExists(child.child)) {
|
|
throw new Error(`${name} died before the freeze`);
|
|
}
|
|
frozenChildren.push(child);
|
|
}
|
|
// A stop that lands inside a store transaction keeps its row locks for
|
|
// the whole freeze, and the replacement's claim then fails on a lock
|
|
// wait instead of meeting the fence. Only the frozen pair talks to
|
|
// MySQL at this point, so any open InnoDB transaction is theirs:
|
|
// resume them briefly and stop again.
|
|
for (let attempt = 0; ; attempt += 1) {
|
|
for (const [child] of formerPair) {
|
|
signalProcessTree(child.child, 'SIGSTOP');
|
|
}
|
|
if (
|
|
runMysql(
|
|
mysqlPort,
|
|
'SELECT COUNT(*) FROM information_schema.innodb_trx',
|
|
) === '0'
|
|
) {
|
|
break;
|
|
}
|
|
if (attempt === 100) {
|
|
throw new Error(
|
|
'The original control plane never stopped outside a store transaction',
|
|
);
|
|
}
|
|
for (const [child] of formerPair) {
|
|
signalProcessTree(child.child, 'SIGCONT');
|
|
}
|
|
// InnoDB reuses its transaction snapshot until no reader has
|
|
// queried it for 100ms; faster polling can keep a stale busy view.
|
|
await new Promise((resolve) => setTimeout(resolve, 150));
|
|
}
|
|
} else if (harnessOnly) {
|
|
// Kill only the Harness: a live control plane must adopt the next
|
|
// generation instead of failing every bound Session (G3).
|
|
await crashChild(harness.child, `${hostedHarnessLabel} (original)`);
|
|
} else {
|
|
await Promise.all([
|
|
crashChild(harness.child, `${hostedHarnessLabel} (original)`),
|
|
inflightFailover || continuationFailover
|
|
? crashProcess(spring.child, 'Spring Managed Agent Server (original)')
|
|
: crashChild(spring.child, 'Spring Managed Agent Server (original)'),
|
|
]);
|
|
}
|
|
acceptReplacementContinuation = true;
|
|
releaseContinuationHold();
|
|
if (modelRoundFailover) advanceModelRound();
|
|
await heldStartProxy?.close();
|
|
heldStartProxy = undefined;
|
|
if (!freeze && !frozenControlPlane) {
|
|
// The frozen Harness keeps its home; the killed Spring's disk is
|
|
// scrapped like any dead owner's.
|
|
rmSync(harnessHome, { recursive: true, force: true });
|
|
}
|
|
if (!harnessOnly && !frozenControlPlane) {
|
|
rmSync(runtimeHome, { recursive: true, force: true });
|
|
}
|
|
await waitUntil(
|
|
'Managed Session writer lease expiry',
|
|
(remainingMs) =>
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT IF(writer_lease_until IS NULL OR writer_lease_until < (${databaseEpochMillisSql}), 1, 0) FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter}`,
|
|
remainingMs,
|
|
) === '1',
|
|
10_000,
|
|
);
|
|
if (inflightFailover || continuationFailover || modelRoundFailover) {
|
|
await waitUntil(
|
|
'Managed Turn dispatch lease expiry',
|
|
(remainingMs) =>
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT IF(dispatch_lease_until IS NULL OR dispatch_lease_until < (${databaseEpochMillisSql}), 1, 0) FROM qwen_managed_agent.ma_turn WHERE ${sessionFilter}`,
|
|
remainingMs,
|
|
) === '1',
|
|
10_000,
|
|
);
|
|
}
|
|
|
|
let replacementSpring: typeof spring;
|
|
let replacementSpringUrl: string;
|
|
let replacementHarnessPort: number;
|
|
let replacementBrokerUrl: string;
|
|
if (harnessOnly) {
|
|
// The live Spring keeps its fixed Harness base URL, so the next
|
|
// generation must answer on the same port; the same audit selectors
|
|
// then read the original control plane.
|
|
replacementSpring = spring;
|
|
replacementSpringUrl = springUrl;
|
|
replacementHarnessPort = harnessPort;
|
|
// The two pre-dispatch crash barriers hold the approved execution's
|
|
// :start before the Broker sees it: after-prepare waits for the
|
|
// durable PREPARED row, while after-decision only needs the decision
|
|
// to commit. after-result must NOT be held here: its barrier is the
|
|
// committed result, so holding :start would keep the execution at
|
|
// PREPARED and its own wait for SETTLED could never succeed.
|
|
// cleanup-failed keeps :start live and refuses the release instead.
|
|
if (
|
|
approvalFault === 'after-decision' ||
|
|
approvalFault === 'after-prepare' ||
|
|
approvalFault === 'cleanup-failed'
|
|
) {
|
|
replacementBrokerProxy = await startHeldExecutionStartProxy(
|
|
`http://127.0.0.1:${brokerPort}`,
|
|
approvalFault !== 'cleanup-failed',
|
|
approvalFault === 'cleanup-failed',
|
|
);
|
|
}
|
|
replacementBrokerUrl =
|
|
replacementBrokerProxy?.baseUrl ?? `http://127.0.0.1:${brokerPort}`;
|
|
} else {
|
|
// Under --freeze the original Spring is dead, but the frozen Harness
|
|
// still holds the session-store URL the original Spring handed it at
|
|
// load (immutable after load). The replacement must answer on the
|
|
// original Spring port so the woken former writer's store calls meet
|
|
// a live, fencing control plane instead of a dead socket — otherwise
|
|
// every post-wake assertion would hold by disconnection, not fencing.
|
|
const replacementSpringPort = freeze ? springPort : await freePort();
|
|
replacementHarnessPort = await freePort();
|
|
const replacementBrokerPort = await freePort();
|
|
replacementSpringUrl = `http://127.0.0.1:${replacementSpringPort}`;
|
|
replacementSpring = start(
|
|
java,
|
|
springArguments,
|
|
{
|
|
env: {
|
|
...cleanEnvironment,
|
|
HOME: replacementRuntimeHome,
|
|
LANG: process.env['LANG'] ?? 'C',
|
|
LC_ALL: process.env['LC_ALL'] ?? 'C',
|
|
NO_PROXY: '127.0.0.1,localhost',
|
|
QWEN_HOME: path.join(replacementRuntimeHome, '.qwen'),
|
|
TMPDIR: temporary,
|
|
no_proxy: '127.0.0.1,localhost',
|
|
SERVER_PORT: String(replacementSpringPort),
|
|
SPRING_DATASOURCE_PASSWORD: '',
|
|
SPRING_DATASOURCE_URL: `jdbc:mysql://127.0.0.1:${mysqlPort}/qwen_managed_agent?useSSL=false&allowPublicKeyRetrieval=true`,
|
|
SPRING_DATASOURCE_USERNAME: 'root',
|
|
QWEN_MANAGED_AGENT_APPROVAL_MODE: approvalRecovery
|
|
? 'default'
|
|
: 'yolo',
|
|
QWEN_MANAGED_AGENT_APPROVAL_TIMEOUT:
|
|
verificationMode === 'expiry' ? '30s' : '5m',
|
|
QWEN_MANAGED_AGENT_CAPABILITY_DIGEST: capabilityDigest,
|
|
QWEN_MANAGED_AGENT_HARNESS_BASE_URL: `http://127.0.0.1:${replacementHarnessPort}`,
|
|
QWEN_MANAGED_AGENT_HARNESS_ENABLED: 'true',
|
|
QWEN_MANAGED_AGENT_HARNESS_REQUEST_TIMEOUT: '120s',
|
|
QWEN_MANAGED_AGENT_HARNESS_TOKEN: harnessToken,
|
|
QWEN_MANAGED_AGENT_RUNTIME_TRUSTED_LOCAL_REBOOT_RECOVERY: 'false',
|
|
...(registerStorage
|
|
? {
|
|
QWEN_MANAGED_AGENT_RUNTIME_VERIFIED_WORKSPACE_RECOVERY_ENABLED:
|
|
'true',
|
|
}
|
|
: {}),
|
|
QWEN_MANAGED_AGENT_TRUSTED_ACTOR_HEADER: trustedActorHeader,
|
|
QWEN_MANAGED_AGENT_WORKSPACE_FILES_ENABLED: 'true',
|
|
...(runtimeTakeover
|
|
? {
|
|
QWEN_MANAGED_AGENT_RUNTIME_DURABLE_LOCAL_PROCESS:
|
|
process.platform === 'linux' ? 'true' : 'false',
|
|
}
|
|
: {
|
|
QWEN_MANAGED_AGENT_RUNTIME_DURABLE_LOCAL_PROCESS: 'false',
|
|
}),
|
|
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: replacementSpringUrl,
|
|
QWEN_MANAGED_AGENT_SESSION_STORE_ENABLED: 'true',
|
|
QWEN_MANAGED_AGENT_SESSION_STORE_WRITER_LEASE_DURATION: '5s',
|
|
QWEN_MANAGED_AGENT_WORKSPACE_ID: workspaceId,
|
|
QWEN_MANAGED_AGENT_RUNTIME_BROKER_ENABLED: 'true',
|
|
QWEN_MANAGED_AGENT_RUNTIME_BROKER_PORT: String(
|
|
replacementBrokerPort,
|
|
),
|
|
QWEN_MANAGED_AGENT_RUNTIME_BROKER_TOKEN: brokerToken,
|
|
QWEN_MANAGED_AGENT_RUNTIME_CREDENTIAL_KEY: credentialKey,
|
|
QWEN_MANAGED_AGENT_RUNTIME_CREDENTIAL_KEY_ID: 'e2e-local-v1',
|
|
QWEN_MANAGED_AGENT_RUNTIME_STATE_DIRECTORY: frozenControlPlane
|
|
? replacementRuntimeState
|
|
: runtimeState,
|
|
QWEN_MANAGED_AGENT_RUNTIME_WORKER_ENTRY: cliBundle,
|
|
QWEN_MANAGED_AGENT_NODE_EXECUTABLE: process.execPath,
|
|
QWEN_MANAGED_AGENT_WORKSPACE_CWD: workspace,
|
|
},
|
|
},
|
|
'Replacement Spring Managed Agent Server',
|
|
);
|
|
await waitUntil(
|
|
'Replacement Spring Managed Agent Server',
|
|
async () => {
|
|
const response = await fetch(
|
|
`${replacementSpringUrl}/actuator/health`,
|
|
);
|
|
return response.ok;
|
|
},
|
|
60_000,
|
|
replacementSpring,
|
|
);
|
|
|
|
if (inflightFailover) {
|
|
replacementBrokerProxy = await startHeldExecutionStartProxy(
|
|
`http://127.0.0.1:${replacementBrokerPort}`,
|
|
false,
|
|
);
|
|
}
|
|
replacementBrokerUrl =
|
|
replacementBrokerProxy?.baseUrl ??
|
|
`http://127.0.0.1:${replacementBrokerPort}`;
|
|
}
|
|
|
|
// Every replacement generation answers on the same port with a fresh
|
|
// home: the Harness-only model-round arm starts a third one after the
|
|
// second dies mid-round.
|
|
const startReplacementHarness = async (
|
|
name: string,
|
|
crashBeforeAssistantCommit = false,
|
|
holdPendingSiblingApproval = false,
|
|
): Promise<Child> => {
|
|
const next = start(
|
|
process.execPath,
|
|
[
|
|
cliBundle,
|
|
'serve',
|
|
'--profile',
|
|
'hosted-harness',
|
|
'--port',
|
|
String(replacementHarnessPort),
|
|
'--hostname',
|
|
'127.0.0.1',
|
|
'--require-auth',
|
|
'--no-web',
|
|
'--workspace',
|
|
workspace,
|
|
// The original harness keeps its broker unconditionally, so the
|
|
// replacement must too — non-workspace failover arms read the same
|
|
// execution store the broker feeds; replacementBrokerUrl already
|
|
// carries the harnessOnly alias and the held-start proxy base.
|
|
'--managed-runtime-broker-url',
|
|
replacementBrokerUrl,
|
|
`--managed-runtime-broker-token=${brokerToken}`,
|
|
],
|
|
{
|
|
env: {
|
|
...cleanEnvironment,
|
|
HOME: replacementHarnessHome,
|
|
LANG: process.env['LANG'] ?? 'C',
|
|
LC_ALL: process.env['LC_ALL'] ?? 'C',
|
|
QWEN_HOME: path.join(replacementHarnessHome, '.qwen'),
|
|
QWEN_CODE_TRUSTED_FOLDERS_PATH: trustedFolders,
|
|
QWEN_HOSTED_HARNESS_CAPABILITY_DIGEST: capabilityDigest,
|
|
QWEN_SERVER_TOKEN: harnessToken,
|
|
OPENAI_API_KEY: 'fake-key',
|
|
OPENAI_BASE_URL: fakeBaseUrl,
|
|
OPENAI_MODEL: 'fake-model',
|
|
QWEN_MODEL: 'fake-model',
|
|
QWEN_RUNTIME_BROKER_TOKEN: brokerToken,
|
|
QWEN_RUNTIME_BROKER_URL: replacementBrokerUrl,
|
|
// A4: the replacement loses its first load attach reply once, so
|
|
// the control plane must redrive into the now-active driver.
|
|
...(approvalFault === 'load-reply-lost'
|
|
? { QWEN_E2E_DROP_LOAD_REPLY: session.id }
|
|
: {}),
|
|
// B3a: only the generation that must crash carries this flag; the
|
|
// post-fault recovery generation must not inherit it, or it would
|
|
// crash in the same window forever.
|
|
...(crashBeforeAssistantCommit
|
|
? { QWEN_E2E_CRASH_BEFORE_ASSISTANT_COMMIT: session.id }
|
|
: {}),
|
|
// A3: only the generation that must crash carries this flag; it
|
|
// holds the first sibling approval it asks (the pending sibling
|
|
// after the decided one). The recovery generation must not inherit
|
|
// it, or it would hold the same approval forever.
|
|
...(holdPendingSiblingApproval
|
|
? { QWEN_E2E_HOLD_PENDING_SIBLING_APPROVAL: session.id }
|
|
: {}),
|
|
},
|
|
},
|
|
name,
|
|
);
|
|
await waitUntil(
|
|
name,
|
|
async () => {
|
|
const response = await fetch(
|
|
`http://127.0.0.1:${replacementHarnessPort}/health?deep=1`,
|
|
{ headers: { authorization: `Bearer ${harnessToken}` } },
|
|
);
|
|
return response.ok;
|
|
},
|
|
60_000,
|
|
next,
|
|
);
|
|
return next;
|
|
};
|
|
let replacementHarness = await startReplacementHarness(
|
|
'Replacement Hosted Harness',
|
|
approvalFault === 'finished-before-commit',
|
|
approvalFault === 'sibling-approval',
|
|
);
|
|
|
|
if (approvalRecovery) {
|
|
await waitUntil(
|
|
'Original approval attached to replacement Harness',
|
|
() => {
|
|
const boot = runMysql(
|
|
mysqlPort,
|
|
`SELECT harness_boot_id FROM qwen_managed_agent.ma_session WHERE ${sessionFilter}`,
|
|
);
|
|
return boot.length > 0 && boot !== firstBootId;
|
|
},
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
if (approvalFault === 'load-reply-lost') {
|
|
// The fault must actually have landed: the replacement lost its first
|
|
// load attach reply (after its driver started). A redrive that never
|
|
// lost a reply would prove nothing about the active-driver handoff.
|
|
await waitUntil(
|
|
'Takeover load reply lost',
|
|
() =>
|
|
replacementHarness.log().includes('E2E dropping first load reply'),
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
}
|
|
const actionBeforeResponse = await fetchJson<Record<string, unknown>>(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/actions/${originalApprovalId}`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
if (
|
|
actionBeforeResponse.id !== originalApprovalId ||
|
|
(verificationMode !== 'expiry' &&
|
|
actionBeforeResponse.state !== 'requested')
|
|
) {
|
|
throw new Error(
|
|
'Replacement did not preserve the original requested Action',
|
|
);
|
|
}
|
|
// The replacement must hand the parked wait the deadline it already
|
|
// had; a reset deadline would extend expiry and hide a lost clock.
|
|
const optionsAfterReplacement = JSON.parse(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT options_json FROM qwen_managed_agent.ma_session_record WHERE tenant_id=${sqlString(tenant)} AND session_id=${sqlString(session.id)} AND record_kind='ACTION' AND action_id=${sqlString(originalApprovalId!)}`,
|
|
),
|
|
);
|
|
if (
|
|
actionBeforeResponse.created_at !==
|
|
originalApprovalOptions!.createdAt ||
|
|
actionBeforeResponse.expires_at !==
|
|
originalApprovalOptions!.expiresAt ||
|
|
optionsAfterReplacement.createdAt !==
|
|
originalApprovalOptions!.createdAt ||
|
|
optionsAfterReplacement.expiresAt !== originalApprovalOptions!.expiresAt
|
|
) {
|
|
throw new Error(
|
|
`Replacement reset the original approval deadline: action=${String(actionBeforeResponse.created_at)}..${String(actionBeforeResponse.expires_at)} record=${optionsAfterReplacement.createdAt}..${optionsAfterReplacement.expiresAt} saved=${originalApprovalOptions!.createdAt}..${originalApprovalOptions!.expiresAt}`,
|
|
);
|
|
}
|
|
let responseOperation: Record<string, unknown> | undefined;
|
|
if (verificationMode === 'allow' || verificationMode === 'deny') {
|
|
const decisionUrl = `${replacementSpringUrl}/v1/agents/sessions/${session.id}/actions/${originalApprovalId}/responses`;
|
|
const decisionHeaders = {
|
|
'content-type': 'application/json',
|
|
'idempotency-key': `g3-answer-${originalApprovalId}`,
|
|
...tenantHeaders(tenant),
|
|
};
|
|
const decisionBody = JSON.stringify({
|
|
kind: 'permission',
|
|
input_revision: originalApprovalOptions!.inputRevision,
|
|
policy_revision: originalApprovalOptions!.policyRevision,
|
|
option_id: verificationMode,
|
|
});
|
|
if (
|
|
approvalFault === 'after-result' ||
|
|
approvalFault === 'after-earlier-tools'
|
|
) {
|
|
approvalResultHold = new Promise<void>((resolve) => {
|
|
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);
|
|
}
|
|
const response = await fetch(decisionUrl, {
|
|
method: 'POST',
|
|
headers: decisionHeaders,
|
|
body: decisionBody,
|
|
});
|
|
if (response.status !== 202)
|
|
throw new Error(
|
|
`Original Action response returned ${response.status}: ${await response.text()}`,
|
|
);
|
|
responseOperation = await response.json();
|
|
if (approvalFault === 'decision-reply-lost') {
|
|
const operations = runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_operation WHERE ${sessionFilter} AND operation_kind='ACTION_RESPONSE'`,
|
|
);
|
|
if (operations !== '1')
|
|
throw new Error(
|
|
`Lost decision reply produced ${operations} ACTION_RESPONSE operations, not one`,
|
|
);
|
|
}
|
|
if (approvalFault === 'after-earlier-tools') {
|
|
// The first Allow drives tool A (round 0); the model then asks approval
|
|
// for tool B (round 1) under a NEW Action. Detect that requested Action
|
|
// and allow it so both earlier rounds execute before the held round 2.
|
|
await waitUntil(
|
|
'Second earlier-tool approval requested',
|
|
() => {
|
|
const rows = runMysql(
|
|
mysqlPort,
|
|
`SELECT action_id, state FROM qwen_managed_agent.ma_session_record WHERE ${sessionFilter} AND record_kind='ACTION'`,
|
|
);
|
|
const requested = (rows ? rows.split('\n') : [])
|
|
.map((row) => row.split('\t'))
|
|
.filter(
|
|
([id, state]) =>
|
|
state === 'requested' && id !== originalApprovalId,
|
|
);
|
|
if (requested.length === 1) {
|
|
secondApprovalId = requested[0][0];
|
|
return true;
|
|
}
|
|
return false;
|
|
},
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
const secondOptions = JSON.parse(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT options_json FROM qwen_managed_agent.ma_session_record WHERE ${sessionFilter} AND record_kind='ACTION' AND action_id=${sqlString(secondApprovalId!)}`,
|
|
),
|
|
);
|
|
const secondResponse = await fetch(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/actions/${secondApprovalId}/responses`,
|
|
{
|
|
method: 'POST',
|
|
headers: {
|
|
'content-type': 'application/json',
|
|
'idempotency-key': `g3-answer-${secondApprovalId}`,
|
|
...tenantHeaders(tenant),
|
|
},
|
|
body: JSON.stringify({
|
|
kind: 'permission',
|
|
input_revision: secondOptions.inputRevision,
|
|
policy_revision: secondOptions.policyRevision,
|
|
option_id: 'allow',
|
|
}),
|
|
},
|
|
);
|
|
if (secondResponse.status !== 202)
|
|
throw new Error(
|
|
`Second earlier-tool response returned ${secondResponse.status}: ${await secondResponse.text()}`,
|
|
);
|
|
}
|
|
} else if (verificationMode === 'cancel') {
|
|
const response = await fetch(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/events`,
|
|
{
|
|
method: 'POST',
|
|
headers: {
|
|
'content-type': 'application/json',
|
|
'idempotency-key': `g3-cancel-${originalApprovalId}`,
|
|
...tenantHeaders(tenant),
|
|
},
|
|
body: JSON.stringify({
|
|
type: 'agent.session.cancel',
|
|
turn_id: actionBeforeResponse.turn_id,
|
|
}),
|
|
},
|
|
);
|
|
if (response.status !== 202)
|
|
throw new Error(
|
|
`Original Turn cancel returned ${response.status}: ${await response.text()}`,
|
|
);
|
|
}
|
|
let publishedPrefixSequences: number[] = [];
|
|
let terminalAcceptsAtBarrier = 0;
|
|
let b3Before: B3RoundBudget | undefined;
|
|
if (
|
|
approvalFault === 'after-decision' ||
|
|
approvalFault === 'after-prepare' ||
|
|
approvalFault === 'after-result' ||
|
|
approvalFault === 'finished-before-commit' ||
|
|
approvalFault === 'sibling-approval' ||
|
|
approvalFault === 'after-earlier-tools' ||
|
|
approvalFault === 'cleanup-failed' ||
|
|
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
|
|
// open) so the next generation must resume the stage instead of
|
|
// replaying the approval.
|
|
const crashedBootId = runMysql(
|
|
mysqlPort,
|
|
`SELECT harness_boot_id FROM qwen_managed_agent.ma_session WHERE ${sessionFilter}`,
|
|
);
|
|
if (approvalFault === 'after-prepare') {
|
|
await waitUntil(
|
|
'Approved execution prepared behind the held start',
|
|
() => {
|
|
const execution = runMysql(
|
|
mysqlPort,
|
|
'SELECT execution_state, dispatch_generation FROM qwen_managed_agent.ma_tool_execution',
|
|
);
|
|
return (
|
|
execution === 'PREPARED\t0' &&
|
|
replacementBrokerProxy?.heldPath() !== undefined
|
|
);
|
|
},
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
} else if (approvalFault === 'after-result') {
|
|
await waitUntil(
|
|
'Approved result committed and its prefix published',
|
|
async () => {
|
|
const execution = runMysql(
|
|
mysqlPort,
|
|
'SELECT execution_state, IF(result_json IS NULL, 0, 1) FROM qwen_managed_agent.ma_tool_execution',
|
|
);
|
|
const heldResultRequest = (fake?.requests ?? []).some(
|
|
({ body }) => {
|
|
const messages = JSON.stringify(body['messages']);
|
|
return (
|
|
messages.includes(continuationMarker) &&
|
|
messages.includes('"role":"tool"')
|
|
);
|
|
},
|
|
);
|
|
if (execution !== 'SETTLED\t1' || !heldResultRequest)
|
|
return false;
|
|
// The request log is written before the fake server sends its
|
|
// response, so it alone does not prove the Harness consumed and
|
|
// published the held prefix. Require the public feed to carry
|
|
// it; otherwise a no-op retraction would pass vacuously.
|
|
const view = await publicTranscript(
|
|
replacementSpringUrl,
|
|
tenant,
|
|
session.id,
|
|
);
|
|
return view.text.includes(continuationPartial);
|
|
},
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
} else if (approvalFault === 'after-earlier-tools') {
|
|
// Both earlier tool rounds must have committed their results and the
|
|
// round-2 post-tool request must be held with its prefix published, so
|
|
// the crash lands after the earlier tools but before the final output
|
|
// commits. Recovery must then resume the saved round and cost ledger.
|
|
await waitUntil(
|
|
'Earlier tools settled and the later round held',
|
|
async () => {
|
|
const rows = runMysql(
|
|
mysqlPort,
|
|
'SELECT execution_state, IF(result_json IS NULL, 0, 1) FROM qwen_managed_agent.ma_tool_execution ORDER BY execution_call_id',
|
|
);
|
|
const execs = rows ? rows.split('\n') : [];
|
|
if (execs.length !== 2 || execs.some((r) => r !== 'SETTLED\t1'))
|
|
return false;
|
|
const heldLaterRequest = (fake?.requests ?? []).some(
|
|
({ body }) => {
|
|
const messages = JSON.stringify(body['messages']);
|
|
return (
|
|
messages.includes(continuationMarker) &&
|
|
(messages.match(/"role":"tool"/g) ?? []).length >= 2
|
|
);
|
|
},
|
|
);
|
|
if (!heldLaterRequest) return false;
|
|
const view = await publicTranscript(
|
|
replacementSpringUrl,
|
|
tenant,
|
|
session.id,
|
|
);
|
|
return view.text.includes(continuationPartial);
|
|
},
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
// Capture the durable round and cost ledger at the crash boundary so
|
|
// recovery can be proven to resume them rather than restart them.
|
|
b3Before = b3RoundAndBudget(mysqlPort, sessionFilter);
|
|
saveEvidence('b3-round-budget-before.json', b3Before);
|
|
} 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 === 'cleanup-failed') {
|
|
// C3: the release refusal is the fault; the crash must land after
|
|
// the control plane projected the terminal, or the dispatch scan
|
|
// would take a RUNNING Turn over and pay the debt through that
|
|
// path. Only a terminal Turn proves the cleanup-only retry.
|
|
await waitUntil(
|
|
'Runtime cleanup release refused',
|
|
() => (replacementBrokerProxy?.releaseFailureCount() ?? 0) > 0,
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
await waitUntil(
|
|
'Original Turn projected before the cleanup crash',
|
|
() =>
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT status FROM qwen_managed_agent.ma_turn WHERE ${sessionFilter} ORDER BY gmt_create DESC LIMIT 1`,
|
|
) === 'COMPLETED',
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
} 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(
|
|
'Approval decision durably committed',
|
|
() =>
|
|
approvalDecisionCommitted(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT state, IF(decision_receipt_id IS NULL, 0, 1), IF(decision_digest IS NULL, 0, 1) FROM qwen_managed_agent.ma_session_record WHERE ${sessionFilter} AND record_kind='ACTION' AND action_id=${sqlString(originalApprovalId!)}`,
|
|
),
|
|
),
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
} else if (approvalFault === 'finished-before-commit') {
|
|
// B3a: the replacement hooked the window after its post-tool stream
|
|
// Finished marker is durable but before the assistant commit. The
|
|
// barrier line proves the fault landed inside that window; the
|
|
// recovery parity check below proves the next generation reissued
|
|
// the saved request rather than compensating a committed terminal.
|
|
await waitUntil(
|
|
'Finished turn held before the assistant commit',
|
|
() =>
|
|
replacementHarness
|
|
.log()
|
|
.includes('E2E holding finished turn before assistant commit'),
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
} else if (approvalFault === 'sibling-approval') {
|
|
// A3: the replacement decided sibling[0] (Allow[0]) and now holds
|
|
// the first approval it asks — the pending sibling[1]. The barrier
|
|
// line proves the hook fired; the durable shape proves the crash
|
|
// lands in the approval gap (sibling[0] decided, sibling[1] still
|
|
// requested), not after the whole batch finished approving.
|
|
await waitUntil(
|
|
'Pending sibling approval held',
|
|
() =>
|
|
replacementHarness
|
|
.log()
|
|
.includes('E2E holding pending sibling approval'),
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
const siblingActionRows = runMysql(
|
|
mysqlPort,
|
|
`SELECT action_id, state FROM qwen_managed_agent.ma_session_record WHERE ${sessionFilter} AND record_kind='ACTION' ORDER BY action_id`,
|
|
);
|
|
const siblingActions = siblingActionRows
|
|
? siblingActionRows.split('\n').map((row) => row.split('\t'))
|
|
: [];
|
|
const decidedSibling = siblingActions.find(
|
|
([id]) => id === originalApprovalId,
|
|
);
|
|
const pendingSibling = siblingActions.find(
|
|
([id]) => id !== originalApprovalId,
|
|
);
|
|
if (
|
|
siblingActions.length !== 2 ||
|
|
decidedSibling?.[1] !== 'decided' ||
|
|
pendingSibling?.[1] !== 'requested'
|
|
) {
|
|
throw new Error(
|
|
`Sibling-approval crash did not land in the approval gap: ${siblingActionRows}`,
|
|
);
|
|
}
|
|
siblingApprovalId = pendingSibling![0];
|
|
siblingApprovalOptions = JSON.parse(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT options_json FROM qwen_managed_agent.ma_session_record WHERE ${sessionFilter} AND record_kind='ACTION' AND action_id=${sqlString(siblingApprovalId)}`,
|
|
),
|
|
);
|
|
const siblingPlan = JSON.parse(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT CONVERT(inline_bytes USING utf8mb4) FROM qwen_managed_agent.ma_resource WHERE tenant_id=${sqlString(tenant)} AND session_id=${sqlString(session.id)} AND resource_id=${sqlString(siblingApprovalOptions!.continuationRef.resourceId)}`,
|
|
),
|
|
) as SavedApprovalPlan;
|
|
if (
|
|
siblingPlan.stage !== 'approval' ||
|
|
siblingPlan.approvalOrdinal !== 1 ||
|
|
siblingPlan.calls?.length !== 2 ||
|
|
siblingPlan.actionId !== siblingApprovalId
|
|
) {
|
|
throw new Error(
|
|
`Sibling-approval crash plan is not the pending second sibling: ${JSON.stringify(siblingPlan)}`,
|
|
);
|
|
}
|
|
// The per-call Action ID may be null before the crash, so anchor each
|
|
// member to the first approval's saved call identities and reject any
|
|
// Action ID that is already present on the wrong member.
|
|
const expectedActionIds = [originalApprovalId!, siblingApprovalId!];
|
|
for (const [
|
|
ordinal,
|
|
originalCall,
|
|
] of originalApprovalPlan!.calls.entries()) {
|
|
const siblingCall = siblingPlan.calls[ordinal];
|
|
if (
|
|
!siblingCall ||
|
|
siblingCall.runtimeCallId !== originalCall.runtimeCallId ||
|
|
siblingCall.partIndex !== originalCall.partIndex ||
|
|
siblingCall.prepareKey !== originalCall.prepareKey ||
|
|
siblingCall.requestDigest !== originalCall.requestDigest ||
|
|
(siblingCall.actionId != null &&
|
|
siblingCall.actionId !== expectedActionIds[ordinal])
|
|
) {
|
|
throw new Error(
|
|
`Sibling-approval crash plan changed call/action binding at ordinal ${ordinal}: ${JSON.stringify(siblingCall)} != ${JSON.stringify(originalCall)}`,
|
|
);
|
|
}
|
|
}
|
|
saveEvidence('sibling-approval-barrier.json', {
|
|
fault: approvalFault,
|
|
decidedAction: decidedSibling,
|
|
pendingAction: pendingSibling,
|
|
approvalOrdinal: siblingPlan.approvalOrdinal,
|
|
calls: siblingPlan.calls,
|
|
});
|
|
} 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,
|
|
'SELECT execution_state, dispatch_generation FROM qwen_managed_agent.ma_tool_execution',
|
|
);
|
|
if (
|
|
approvalFault === 'finished-before-commit' &&
|
|
executionAtCrash !== 'SETTLED\t1'
|
|
) {
|
|
// The crash must land after the approved result settled and its
|
|
// post-tool round reached the window, not before. Assert the durable
|
|
// precondition from MySQL like the sibling crash arms, rather than
|
|
// trusting the hook's log line alone.
|
|
throw new Error(
|
|
`Finished-before-commit crash did not land after the approved result settled: executionAtCrash=${executionAtCrash}`,
|
|
);
|
|
}
|
|
// The settled execution alone does not prove the crash landed inside
|
|
// B3a: the post-tool round could have been held before its durable
|
|
// Finished marker committed. Read the window from the same journal the
|
|
// recovery reads and retain the attempt identity as evidence.
|
|
const finishedWindow =
|
|
approvalFault === 'finished-before-commit'
|
|
? finishedBeforeCommitWindow(
|
|
mysqlPort,
|
|
sessionFilter,
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT prompt_id FROM qwen_managed_agent.ma_turn WHERE ${sessionFilter} ORDER BY gmt_create DESC LIMIT 1`,
|
|
),
|
|
)
|
|
: undefined;
|
|
const publishedPrefix =
|
|
approvalFault === 'after-result' ||
|
|
approvalFault === 'after-earlier-tools'
|
|
? await publicTranscript(replacementSpringUrl, tenant, session.id)
|
|
: undefined;
|
|
publishedPrefixSequences = (publishedPrefix?.events ?? [])
|
|
.filter(
|
|
(event) =>
|
|
event.type === 'item.output_text.delta' &&
|
|
eventText(event).includes(continuationPartial),
|
|
)
|
|
.map((event) => event.sequence);
|
|
saveEvidence('approval-crash-barrier.json', {
|
|
fault: approvalFault,
|
|
crashedHarnessBootId: crashedBootId,
|
|
executionAtCrash,
|
|
finishedBeforeCommitWindow: finishedWindow,
|
|
heldExecutionStartPath: replacementBrokerProxy?.heldPath(),
|
|
releaseFailures: replacementBrokerProxy?.releaseFailureCount(),
|
|
turnStatusAtCrash:
|
|
approvalFault === 'cleanup-failed'
|
|
? runMysql(
|
|
mysqlPort,
|
|
`SELECT status FROM qwen_managed_agent.ma_turn WHERE ${sessionFilter} ORDER BY gmt_create DESC LIMIT 1`,
|
|
)
|
|
: undefined,
|
|
terminalFault: storeFaultProxy
|
|
? {
|
|
rejected: storeFaultProxy.terminalRejectCount(),
|
|
accepted: storeFaultProxy.terminalAcceptCount(),
|
|
acceptedAtBarrier: terminalAcceptsAtBarrier,
|
|
}
|
|
: undefined,
|
|
publishedPrefix:
|
|
approvalFault === 'after-result' ||
|
|
approvalFault === 'after-earlier-tools'
|
|
? {
|
|
sequences: publishedPrefixSequences,
|
|
text: publishedPrefix?.text,
|
|
}
|
|
: undefined,
|
|
});
|
|
await crashChild(
|
|
replacementHarness.child,
|
|
'Replacement Hosted Harness (fault barrier)',
|
|
);
|
|
if (
|
|
approvalFault === 'after-result' ||
|
|
approvalFault === 'after-earlier-tools'
|
|
) {
|
|
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) {
|
|
if (approvalFault === 'cleanup-failed') {
|
|
// Keep the proxy so the replacement's successful release of the
|
|
// original identity is witnessed on the same fault boundary.
|
|
replacementBrokerProxy.allowRelease();
|
|
} else {
|
|
await replacementBrokerProxy.close();
|
|
replacementBrokerProxy = undefined;
|
|
replacementBrokerUrl = `http://127.0.0.1:${brokerPort}`;
|
|
}
|
|
}
|
|
// The crashed generation's disk is scrapped like every replaced
|
|
// Harness home; the fresh one must recover from MySQL alone.
|
|
rmSync(replacementHarnessHome, { recursive: true, force: true });
|
|
mkdirSync(path.join(replacementHarnessHome, '.qwen'), {
|
|
recursive: true,
|
|
});
|
|
writeFileSync(
|
|
path.join(replacementHarnessHome, '.qwen', 'settings.json'),
|
|
JSON.stringify({ ui: { enableFollowupSuggestions: false } }),
|
|
{ mode: 0o600 },
|
|
);
|
|
await waitUntil(
|
|
'Post-fault writer lease expiry',
|
|
(remainingMs) =>
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT IF(writer_lease_until IS NULL OR writer_lease_until < (${databaseEpochMillisSql}), 1, 0) FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter}`,
|
|
remainingMs,
|
|
) === '1',
|
|
10_000,
|
|
);
|
|
replacementHarness = await startReplacementHarness(
|
|
'Post-fault recovery Harness',
|
|
);
|
|
if (approvalFault === 'cleanup-failed') {
|
|
// C3 has no Turn left to adopt: the replacement's first contact is
|
|
// the cleanup-only load the debt scan issues, and it must not
|
|
// rebind the Session's turn identity. The release it performs on
|
|
// the still-open proxy is the proof that the retry reached it.
|
|
await waitUntil(
|
|
'Post-fault cleanup retry reached the replacement',
|
|
() => (replacementBrokerProxy?.releaseSuccessCount() ?? 0) > 0,
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
} else {
|
|
await waitUntil(
|
|
'Post-fault recovery Harness adopted the Session',
|
|
() => {
|
|
const boot = runMysql(
|
|
mysqlPort,
|
|
`SELECT harness_boot_id FROM qwen_managed_agent.ma_session WHERE ${sessionFilter}`,
|
|
);
|
|
return boot.length > 0 && boot !== crashedBootId;
|
|
},
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
}
|
|
}
|
|
if (approvalFault === 'sibling-approval') {
|
|
// The recovery generation must re-hang the ORIGINAL pending sibling
|
|
// Action (same ID, same deadline) and still require its own decision:
|
|
// it must not auto-allow sibling[1] from sibling[0]'s earlier Allow,
|
|
// nor re-ask the already-decided sibling[0]. Assert the preserved
|
|
// identity/deadline before answering, then decide the pending sibling.
|
|
// Read the decided sibling HERE, before answering the pending one: a
|
|
// recovery that resumes at the wrong ordinal and re-asks sibling[0]
|
|
// would flip it back to requested, and a post-terminal check could not
|
|
// tell "never touched" from "re-decided" once the state reconverges.
|
|
const decidedBefore = await fetchJson<Record<string, unknown>>(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/actions/${originalApprovalId}`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
if (
|
|
decidedBefore.id !== originalApprovalId ||
|
|
decidedBefore.state !== 'decided' ||
|
|
decidedBefore.created_at !== originalApprovalOptions!.createdAt ||
|
|
decidedBefore.expires_at !== originalApprovalOptions!.expiresAt
|
|
) {
|
|
throw new Error(
|
|
`Recovery re-touched the decided sibling Action/deadline: ${JSON.stringify(decidedBefore)}`,
|
|
);
|
|
}
|
|
const siblingBefore = await fetchJson<Record<string, unknown>>(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/actions/${siblingApprovalId}`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
if (
|
|
siblingBefore.id !== siblingApprovalId ||
|
|
siblingBefore.state !== 'requested' ||
|
|
siblingBefore.created_at !== siblingApprovalOptions!.createdAt ||
|
|
siblingBefore.expires_at !== siblingApprovalOptions!.expiresAt
|
|
) {
|
|
throw new Error(
|
|
`Recovery did not preserve the pending sibling Action/deadline: ${JSON.stringify(siblingBefore)}`,
|
|
);
|
|
}
|
|
const siblingResponse = await fetch(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/actions/${siblingApprovalId}/responses`,
|
|
{
|
|
method: 'POST',
|
|
headers: {
|
|
'content-type': 'application/json',
|
|
'idempotency-key': `g3-answer-${siblingApprovalId}`,
|
|
...tenantHeaders(tenant),
|
|
},
|
|
body: JSON.stringify({
|
|
kind: 'permission',
|
|
input_revision: siblingApprovalOptions!.inputRevision,
|
|
policy_revision: siblingApprovalOptions!.policyRevision,
|
|
option_id: 'allow',
|
|
}),
|
|
},
|
|
);
|
|
if (siblingResponse.status !== 202)
|
|
throw new Error(
|
|
`Pending sibling response returned ${siblingResponse.status}: ${await siblingResponse.text()}`,
|
|
);
|
|
}
|
|
const recovered = await waitForTerminal(
|
|
replacementSpringUrl,
|
|
tenant,
|
|
session.id,
|
|
0,
|
|
replacementSpring,
|
|
120_000,
|
|
);
|
|
const terminals = recovered.events.filter((event) => event.terminal);
|
|
const expectedTerminal = ['allow', 'deny'].includes(verificationMode)
|
|
? 'turn.completed'
|
|
: 'turn.cancelled';
|
|
if (terminals.length !== 1 || terminals[0].type !== expectedTerminal) {
|
|
throw new Error(
|
|
`Original approval terminal differs: expected=${expectedTerminal} actual=${JSON.stringify(terminals)}`,
|
|
);
|
|
}
|
|
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' &&
|
|
approvalFault !== 'after-earlier-tools'
|
|
) {
|
|
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' ||
|
|
approvalFault === 'after-earlier-tools'
|
|
) {
|
|
// The reissue must retract the held prefix: only the recovered
|
|
// answer may remain in the public transcript.
|
|
const transcript = await publicTranscript(
|
|
replacementSpringUrl,
|
|
tenant,
|
|
session.id,
|
|
);
|
|
if (transcript.text !== continuationResponse)
|
|
throw new Error(
|
|
`Result-commit recovery left ${JSON.stringify(transcript.text)}, not the recovered answer`,
|
|
);
|
|
if (
|
|
publishedPrefixSequences.length === 0 ||
|
|
!reconciledBetween(
|
|
transcript.events,
|
|
Math.min(...publishedPrefixSequences),
|
|
Math.min(
|
|
...deltaSequences(transcript.events, continuationResponse),
|
|
),
|
|
)
|
|
)
|
|
throw new Error(
|
|
'Result-commit recovery did not reconcile the published prefix before the recovered answer',
|
|
);
|
|
}
|
|
const rows = runMysql(
|
|
mysqlPort,
|
|
`SELECT execution_call_id, idempotency_key, binding_id, runtime_generation, runtime_session_id, execution_state, dispatch_generation, request_digest, settled_at FROM qwen_managed_agent.ma_tool_execution ORDER BY execution_call_id`,
|
|
);
|
|
const executions = rows
|
|
? rows.split('\n').map((row) => row.split('\t'))
|
|
: [];
|
|
const expectedExecutions =
|
|
verificationMode === 'allow'
|
|
? approvalFault === 'sibling-approval' ||
|
|
approvalFault === 'after-earlier-tools'
|
|
? 2
|
|
: 1
|
|
: 0;
|
|
if (executions.length !== expectedExecutions)
|
|
throw new Error(
|
|
`Approval execution count ${executions.length} differs from ${expectedExecutions}`,
|
|
);
|
|
// Every approved call keeps its own original prepare key/digest and the
|
|
// original Runtime identity, settled once. Match by prepare key so the
|
|
// model's call order is checked, not the SQL row order; the A3 arm has
|
|
// two siblings and B3 two earlier rounds, every other allow arm has one.
|
|
if (approvalFault === 'after-earlier-tools') {
|
|
// B3's two earlier tools live in separate rounds, so the first approval
|
|
// plan carries only tool A. Assert both executions' Runtime identity and
|
|
// settled-once directly; the two physical file effects give per-tool
|
|
// evidence, and a re-run of either earlier tool reds the count above.
|
|
for (const [ordinal, execution] of executions.entries()) {
|
|
if (
|
|
execution[2] !== originalApprovalPlan!.runtime.bindingId ||
|
|
execution[3] !== originalApprovalPlan!.runtime.generation ||
|
|
execution[4] !== originalApprovalPlan!.runtime.runtimeSessionId ||
|
|
execution[5] !== 'SETTLED' ||
|
|
execution[6] !== '1'
|
|
) {
|
|
throw new Error(
|
|
`B3 earlier execution ${ordinal} changed its identity or re-ran: ${execution.join(',')}`,
|
|
);
|
|
}
|
|
}
|
|
} else if (verificationMode === 'allow') {
|
|
const plan = originalApprovalPlan!;
|
|
const settledAtByOrdinal: number[] = [];
|
|
for (const [ordinal, call] of plan.calls.entries()) {
|
|
const execution = executions.find(
|
|
(row) => row[1] === call.prepareKey,
|
|
);
|
|
if (
|
|
!execution ||
|
|
execution[2] !== plan.runtime.bindingId ||
|
|
execution[3] !== plan.runtime.generation ||
|
|
execution[4] !== plan.runtime.runtimeSessionId ||
|
|
execution[5] !== 'SETTLED' ||
|
|
execution[6] !== '1' ||
|
|
execution[7] !== call.requestDigest
|
|
) {
|
|
throw new Error(
|
|
`Replacement changed the original prepared execution identity for call ${ordinal}: ${(execution ?? []).join(',')}`,
|
|
);
|
|
}
|
|
settledAtByOrdinal.push(Number(execution[8]));
|
|
}
|
|
// The siblings must SETTLE in plan order, not merely exist with the
|
|
// right identities: a recovery that dispatches the batch in reverse
|
|
// keeps every identity assertion green. settled_at is epoch millis;
|
|
// require it non-decreasing along the plan so a reverse dispatch reds.
|
|
for (
|
|
let ordinal = 1;
|
|
ordinal < settledAtByOrdinal.length;
|
|
ordinal += 1
|
|
) {
|
|
if (
|
|
!Number.isFinite(settledAtByOrdinal[ordinal]) ||
|
|
settledAtByOrdinal[ordinal] < settledAtByOrdinal[ordinal - 1]
|
|
) {
|
|
throw new Error(
|
|
`Replacement settled the siblings out of plan order: settledAt=${settledAtByOrdinal.join(',')}`,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
const bytes = existsSync(inflightSideEffect)
|
|
? readFileSync(inflightSideEffect, 'utf8')
|
|
: '';
|
|
if (
|
|
verificationMode === 'allow'
|
|
? bytes !== inflightSideEffectContent
|
|
: existsSync(inflightSideEffect)
|
|
) {
|
|
throw new Error(
|
|
`Approval physical file assertion failed for ${verificationMode}`,
|
|
);
|
|
}
|
|
if (approvalFault === 'sibling-approval') {
|
|
const siblingBytes = existsSync(siblingSideEffect)
|
|
? readFileSync(siblingSideEffect, 'utf8')
|
|
: '';
|
|
if (siblingBytes !== siblingSideEffectContent)
|
|
throw new Error(
|
|
`Sibling physical file assertion failed: ${JSON.stringify(siblingBytes)}`,
|
|
);
|
|
}
|
|
if (approvalFault === 'after-earlier-tools') {
|
|
const b3Bytes = existsSync(b3SecondSideEffect)
|
|
? readFileSync(b3SecondSideEffect, 'utf8')
|
|
: '';
|
|
if (b3Bytes !== b3SecondSideEffectContent)
|
|
throw new Error(
|
|
`B3 second physical file assertion failed: ${JSON.stringify(b3Bytes)}`,
|
|
);
|
|
// Round + cost ledger continuity: recovery must resume the saved round
|
|
// and restore the budget snapshot / token ledger, not restart them.
|
|
if (!b3Before)
|
|
throw new Error('B3 crash-boundary round/budget was not captured');
|
|
const b3After = b3RoundAndBudget(mysqlPort, sessionFilter);
|
|
saveEvidence('b3-round-budget-after.json', b3After);
|
|
if (b3After.round !== b3Before.round || b3After.round < 2)
|
|
throw new Error(
|
|
`B3 recovery reset the logical round: before=${b3Before.round} after=${b3After.round}`,
|
|
);
|
|
assertB3CostContinuity(b3Before, b3After);
|
|
}
|
|
const actionRows = runMysql(
|
|
mysqlPort,
|
|
`SELECT action_id, state FROM qwen_managed_agent.ma_session_record WHERE ${sessionFilter} AND record_kind='ACTION' ORDER BY action_id`,
|
|
);
|
|
const expectedActionState =
|
|
verificationMode === 'expiry'
|
|
? 'expired'
|
|
: verificationMode === 'cancel'
|
|
? 'cancelled'
|
|
: 'decided';
|
|
if (approvalFault === 'sibling-approval') {
|
|
// Both siblings end decided under their ORIGINAL Action IDs: the
|
|
// decided one was not re-asked, the pending one earned its own Allow.
|
|
const expectedSiblingActions = [
|
|
`${originalApprovalId}\t${expectedActionState}`,
|
|
`${siblingApprovalId}\t${expectedActionState}`,
|
|
].sort();
|
|
if (actionRows !== expectedSiblingActions.join('\n'))
|
|
throw new Error(
|
|
`Sibling approval identities/states differ: ${actionRows}`,
|
|
);
|
|
} else if (approvalFault === 'after-earlier-tools') {
|
|
// Both earlier tools end decided under their ORIGINAL Action IDs: A by
|
|
// the first Allow, B by the runner's second Allow. Recovery must resume
|
|
// the saved model round, not re-ask either approval.
|
|
const expectedB3Actions = [
|
|
`${originalApprovalId}\t${expectedActionState}`,
|
|
`${secondApprovalId}\t${expectedActionState}`,
|
|
].sort();
|
|
if (actionRows !== expectedB3Actions.join('\n'))
|
|
throw new Error(
|
|
`B3 approval identities/states differ: ${actionRows}`,
|
|
);
|
|
} else if (
|
|
actionRows !== `${originalApprovalId}\t${expectedActionState}`
|
|
) {
|
|
throw new Error(
|
|
`Original approval identity/state differs: ${actionRows}`,
|
|
);
|
|
}
|
|
// State alone proves a decision was committed, not which option won.
|
|
// Read the committed decision resource back for the browser-facing
|
|
// intent, and hold it to the revision and policy the browser answered.
|
|
const decisionOption =
|
|
verificationMode === 'allow' || verificationMode === 'deny'
|
|
? JSON.parse(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT CONVERT(r.inline_bytes USING utf8mb4) FROM qwen_managed_agent.ma_resource_ref rr JOIN qwen_managed_agent.ma_resource r ON r.resource_key = rr.resource_key WHERE rr.owner_kind='SESSION_RECORD' AND rr.owner_key = (SELECT record_key FROM qwen_managed_agent.ma_session_record WHERE tenant_id=${sqlString(tenant)} AND session_id=${sqlString(session.id)} AND record_kind='ACTION' AND action_id=${sqlString(originalApprovalId!)}) AND r.kind='managed-action-decision'`,
|
|
),
|
|
)
|
|
: undefined;
|
|
if (
|
|
decisionOption &&
|
|
(decisionOption.optionId !== verificationMode ||
|
|
decisionOption.inputRevision !==
|
|
originalApprovalOptions!.inputRevision ||
|
|
decisionOption.policyRevision !==
|
|
originalApprovalOptions!.policyRevision)
|
|
) {
|
|
throw new Error(
|
|
`Stored approval decision differs from the chosen option: ${JSON.stringify(decisionOption)}`,
|
|
);
|
|
}
|
|
if (approvalFault === 'sibling-approval') {
|
|
// The pending sibling carries its OWN Allow decision bound to its own
|
|
// revisions — it was not auto-allowed from sibling[0]'s earlier Allow.
|
|
const siblingDecision = JSON.parse(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT CONVERT(r.inline_bytes USING utf8mb4) FROM qwen_managed_agent.ma_resource_ref rr JOIN qwen_managed_agent.ma_resource r ON r.resource_key = rr.resource_key WHERE rr.owner_kind='SESSION_RECORD' AND rr.owner_key = (SELECT record_key FROM qwen_managed_agent.ma_session_record WHERE tenant_id=${sqlString(tenant)} AND session_id=${sqlString(session.id)} AND record_kind='ACTION' AND action_id=${sqlString(siblingApprovalId!)}) AND r.kind='managed-action-decision'`,
|
|
),
|
|
);
|
|
if (
|
|
siblingDecision.optionId !== 'allow' ||
|
|
siblingDecision.inputRevision !==
|
|
siblingApprovalOptions!.inputRevision ||
|
|
siblingDecision.policyRevision !==
|
|
siblingApprovalOptions!.policyRevision
|
|
) {
|
|
throw new Error(
|
|
`Stored sibling decision differs from the chosen option: ${JSON.stringify(siblingDecision)}`,
|
|
);
|
|
}
|
|
// Clause 2: the recovered final plan preserves the saved sibling
|
|
// order and identities, and each sibling keeps its original Action
|
|
// ID. Compare by position so a reordered or dropped sibling reds
|
|
// here, not only in the execution count.
|
|
const finalPlan = siblingFinalPlan(
|
|
mysqlPort,
|
|
sessionFilter,
|
|
originalApprovalPlan!.batchId,
|
|
);
|
|
const savedCalls = originalApprovalPlan!.calls;
|
|
if (finalPlan.calls.length !== savedCalls.length)
|
|
throw new Error(
|
|
`Sibling recovery changed the batch size: ${finalPlan.calls.length} != ${savedCalls.length}`,
|
|
);
|
|
for (const [ordinal, saved] of savedCalls.entries()) {
|
|
const recovered = finalPlan.calls[ordinal];
|
|
if (
|
|
recovered.runtimeCallId !== saved.runtimeCallId ||
|
|
recovered.partIndex !== saved.partIndex ||
|
|
recovered.prepareKey !== saved.prepareKey ||
|
|
recovered.requestDigest !== saved.requestDigest
|
|
) {
|
|
throw new Error(
|
|
`Sibling recovery reordered or rewrote call ${ordinal}: ${JSON.stringify(recovered)} != ${JSON.stringify(saved)}`,
|
|
);
|
|
}
|
|
}
|
|
if (
|
|
finalPlan.calls[0].actionId !== originalApprovalId ||
|
|
finalPlan.calls[1].actionId !== siblingApprovalId
|
|
) {
|
|
throw new Error(
|
|
`Sibling recovery changed the original Action IDs: ${JSON.stringify(finalPlan.calls.map((call) => call.actionId))}`,
|
|
);
|
|
}
|
|
}
|
|
// Expiry may only close the wait at or after the deadline it was
|
|
// given; an early terminal would mean the replacement invented one.
|
|
const expiryTerminalAt =
|
|
verificationMode === 'expiry'
|
|
? Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT gmt_create FROM qwen_managed_agent.ma_public_event WHERE ${sessionFilter} AND stream_kind='SESSION' AND stream_id=X'' AND is_terminal=TRUE ORDER BY sequence_id DESC LIMIT 1`,
|
|
),
|
|
)
|
|
: undefined;
|
|
if (
|
|
expiryTerminalAt !== undefined &&
|
|
(!Number.isSafeInteger(expiryTerminalAt) ||
|
|
expiryTerminalAt < originalApprovalOptions!.expiresAt)
|
|
) {
|
|
throw new Error(
|
|
`Expiry closed before the original deadline: terminalAt=${expiryTerminalAt} expiresAt=${originalApprovalOptions!.expiresAt}`,
|
|
);
|
|
}
|
|
const requestBodies = (fake?.requests ?? []).filter((request) =>
|
|
JSON.stringify(request.body.messages).includes(continuationMarker),
|
|
);
|
|
const initialRequests = requestBodies.filter(
|
|
(request) =>
|
|
!JSON.stringify(request.body.messages).includes('"role":"tool"'),
|
|
);
|
|
const modelAfterResults = requestBodies.filter((request) =>
|
|
JSON.stringify(request.body.messages).includes('"role":"tool"'),
|
|
);
|
|
const expectedModelsAfterResults =
|
|
approvalFault === 'after-earlier-tools'
|
|
? 3
|
|
: approvalFault === 'after-result' ||
|
|
approvalFault === 'finished-before-commit'
|
|
? 2
|
|
: ['allow', 'deny'].includes(verificationMode)
|
|
? 1
|
|
: 0;
|
|
if (
|
|
initialRequests.length !== 1 ||
|
|
modelAfterResults.length !== expectedModelsAfterResults
|
|
)
|
|
throw new Error(
|
|
`Approval replayed inference or skipped result inference: ${initialRequests.length}+${modelAfterResults.length}`,
|
|
);
|
|
if (
|
|
approvalFault === 'after-result' ||
|
|
approvalFault === 'finished-before-commit'
|
|
) {
|
|
const parity = isDeepStrictEqual(
|
|
modelAfterResults[0]?.body,
|
|
modelAfterResults[1]?.body,
|
|
);
|
|
saveEvidence('model-request-parity.json', {
|
|
requests: modelAfterResults.map((request) => request.body),
|
|
equal: parity,
|
|
});
|
|
if (!parity)
|
|
throw new Error(
|
|
'Result-commit recovery changed the saved model request',
|
|
);
|
|
}
|
|
if (approvalFault === 'after-earlier-tools') {
|
|
// The held round-2 request (index 1) must be reissued deep-equal by the
|
|
// recovery generation (index 2); index 0 is the earlier round-1 request.
|
|
const parity = isDeepStrictEqual(
|
|
modelAfterResults[1]?.body,
|
|
modelAfterResults[2]?.body,
|
|
);
|
|
saveEvidence('model-request-parity.json', {
|
|
requests: modelAfterResults.map((request) => request.body),
|
|
equal: parity,
|
|
});
|
|
if (!parity)
|
|
throw new Error(
|
|
'B3 recovery changed the saved later-round model request',
|
|
);
|
|
}
|
|
if (responseOperation) {
|
|
await waitUntil(
|
|
'Approval response command completion',
|
|
async () => {
|
|
const operation = await fetchJson<Record<string, unknown>>(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/operations/${responseOperation!.id}`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
if (operation.status === 'failed')
|
|
throw new Error(
|
|
`Approval response operation failed: ${JSON.stringify(operation)}`,
|
|
);
|
|
return operation.status === 'completed';
|
|
},
|
|
30_000,
|
|
replacementSpring,
|
|
);
|
|
}
|
|
await waitUntil(
|
|
'Original Runtime cleanup confirmed',
|
|
() => {
|
|
const state = runMysql(
|
|
mysqlPort,
|
|
`SELECT session_state FROM qwen_managed_agent.ma_runtime_session WHERE runtime_session_id=${sqlString(originalRuntimeSessionId!)}`,
|
|
);
|
|
const confirmed = runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_session_journal_tx WHERE ${sessionFilter} AND command_id=${sqlString(`hosted-cleanup:${originalRuntimeSessionId}:${originalRuntimeSessionId}:confirmed`)}`,
|
|
);
|
|
return state === 'RELEASED' && confirmed === '1';
|
|
},
|
|
30_000,
|
|
replacementHarness,
|
|
);
|
|
const runtimeAfter = runMysql(
|
|
mysqlPort,
|
|
`SELECT binding_id, runtime_generation, workspace_generation, session_state FROM qwen_managed_agent.ma_runtime_session WHERE runtime_session_id=${sqlString(originalRuntimeSessionId!)}`,
|
|
).split('\t');
|
|
const cleanupConfirmed = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_session_journal_tx WHERE ${sessionFilter} AND command_id=${sqlString(`hosted-cleanup:${originalRuntimeSessionId}:${originalRuntimeSessionId}:confirmed`)}`,
|
|
),
|
|
);
|
|
if (
|
|
runtimeAfter[0] !== originalApprovalPlan!.runtime.bindingId ||
|
|
runtimeAfter[1] !== originalApprovalPlan!.runtime.generation ||
|
|
runtimeAfter[2] !== originalApprovalPlan!.runtime.workspaceGeneration ||
|
|
runtimeAfter[3] !== 'RELEASED' ||
|
|
cleanupConfirmed !== 1
|
|
)
|
|
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(' | ')}`,
|
|
);
|
|
}
|
|
if (approvalFault === 'cleanup-failed') {
|
|
const failures = replacementBrokerProxy?.releaseFailureCount() ?? 0;
|
|
const successes = replacementBrokerProxy?.releaseSuccessCount() ?? 0;
|
|
// The debt projection is the scan's input; a confirmed cleanup
|
|
// removes it. A surviving row here means only the Turn's own driver
|
|
// (or an operator) could still pay it — exactly the gap this arm
|
|
// exists to keep closed.
|
|
const debt = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_resource_ref WHERE ${sessionFilter} AND owner_kind='HOSTED_CLEANUP'`,
|
|
),
|
|
);
|
|
if (failures < 1 || successes < 1 || debt !== 0)
|
|
throw new Error(
|
|
`Cleanup fault was not retried after the replacement: failures=${failures} successes=${successes} debt=${debt}`,
|
|
);
|
|
}
|
|
const faultObservations = {
|
|
releaseFailures: replacementBrokerProxy?.releaseFailureCount() ?? 0,
|
|
releaseSuccesses: replacementBrokerProxy?.releaseSuccessCount() ?? 0,
|
|
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,
|
|
chosenOption: decisionOption?.optionId,
|
|
savedDeadline: {
|
|
createdAt: originalApprovalOptions!.createdAt,
|
|
expiresAt: originalApprovalOptions!.expiresAt,
|
|
},
|
|
recoveredDeadline: {
|
|
createdAt: optionsAfterReplacement.createdAt,
|
|
expiresAt: optionsAfterReplacement.expiresAt,
|
|
},
|
|
expiryTerminalAt: expiryTerminalAt ?? null,
|
|
expectedTerminal,
|
|
terminalCount: terminals.length,
|
|
executions,
|
|
physicalFileExists: existsSync(inflightSideEffect),
|
|
physicalFileBytes: bytes,
|
|
modelRequests: initialRequests.length + modelAfterResults.length,
|
|
initialRequests: initialRequests.length,
|
|
modelAfterResults: modelAfterResults.length,
|
|
runtimeAfter,
|
|
cleanupConfirmed,
|
|
});
|
|
approvalResultJson = JSON.stringify(
|
|
{
|
|
mode: verificationMode,
|
|
originalAction: originalApprovalId,
|
|
originalBatch: originalApprovalPlan!.batchId,
|
|
originalBinding: originalApprovalPlan!.runtime.bindingId,
|
|
executions: executions.length,
|
|
terminal: expectedTerminal,
|
|
cleanupConfirmed,
|
|
},
|
|
null,
|
|
2,
|
|
);
|
|
} else if (inflightFailover) {
|
|
const recoveredTurn = await waitForTerminal(
|
|
replacementSpringUrl,
|
|
tenant,
|
|
session.id,
|
|
0,
|
|
replacementSpring,
|
|
30_000,
|
|
);
|
|
const recoveredTerminal = recoveredTurn.events.find(
|
|
(event) => event.terminal,
|
|
);
|
|
const recoveredExecution = runMysql(
|
|
mysqlPort,
|
|
'SELECT execution_call_id, execution_state, dispatch_generation, IF(result_json IS NULL, 0, 1), runtime_session_id FROM qwen_managed_agent.ma_tool_execution',
|
|
).split('\t');
|
|
const executionCount = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
'SELECT COUNT(*) FROM qwen_managed_agent.ma_tool_execution',
|
|
),
|
|
);
|
|
const replacementBootId = runMysql(
|
|
mysqlPort,
|
|
`SELECT harness_boot_id FROM qwen_managed_agent.ma_session WHERE ${sessionFilter}`,
|
|
);
|
|
const replacementHead = runMysql(
|
|
mysqlPort,
|
|
`SELECT writer_generation, journal_revision, committed_sequence FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter}`,
|
|
)
|
|
.split('\t')
|
|
.map(Number);
|
|
const terminalCount = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_public_event WHERE ${sessionFilter} AND stream_kind='SESSION' AND stream_id=X'' AND is_terminal=TRUE`,
|
|
),
|
|
);
|
|
const requests = (fake?.requests ?? []).filter(({ body }) =>
|
|
JSON.stringify(body['messages']).includes(inflightMarker),
|
|
);
|
|
const initialModelRequests = requests.filter(
|
|
({ body }) =>
|
|
!JSON.stringify(body['messages']).includes('"role":"tool"'),
|
|
);
|
|
const continuationRequests = requests.filter(({ body }) =>
|
|
JSON.stringify(body['messages']).includes('"role":"tool"'),
|
|
);
|
|
const sideEffectBytes = existsSync(inflightSideEffect)
|
|
? readFileSync(inflightSideEffect, 'utf8')
|
|
: '';
|
|
if (
|
|
recoveredTerminal?.type !== 'turn.completed' ||
|
|
recoveredExecution.length !== 5 ||
|
|
recoveredExecution[0] !== originalExecutionCallId ||
|
|
recoveredExecution[1] !== 'SETTLED' ||
|
|
// A replaced Broker bumps the re-dispatch to generation 1; a live
|
|
// Broker re-dispatches on the generation it already owned (0 or 1).
|
|
(harnessOnly
|
|
? !(recoveredExecution[2] === '0' || recoveredExecution[2] === '1')
|
|
: Number(recoveredExecution[2]) !== 1) ||
|
|
// Under --harness-only the surviving Spring, Broker and durable
|
|
// Worker keep serving the same runtime session; re-provisioning
|
|
// would redefine the arm as a kill-both.
|
|
(harnessOnly && recoveredExecution[4] !== originalRuntimeSessionId) ||
|
|
// The equality above cannot fail (the column settles at INSERT):
|
|
// "kept serving" is witnessed by the row's own liveness moving.
|
|
(harnessOnly &&
|
|
!(
|
|
firstRuntimeHeartbeat !== undefined &&
|
|
(runtimeSessionHeartbeat(recoveredExecution[4])[1] >
|
|
firstRuntimeHeartbeat[1] ||
|
|
runtimeSessionHeartbeat(recoveredExecution[4])[0] >
|
|
firstRuntimeHeartbeat[0])
|
|
)) ||
|
|
recoveredExecution[3] !== '1' ||
|
|
executionCount !== 1 ||
|
|
replacementBootId.length === 0 ||
|
|
replacementBootId === firstBootId ||
|
|
replacementHead.length !== 3 ||
|
|
replacementHead[0] <= firstHead[0] ||
|
|
replacementHead[1] <= firstHead[1] ||
|
|
replacementHead[2] <= firstHead[2] ||
|
|
terminalCount !== 1 ||
|
|
initialModelRequests.length !== 1 ||
|
|
continuationRequests.length !== 1 ||
|
|
sideEffectBytes !== inflightSideEffectContent
|
|
) {
|
|
throw new Error(
|
|
`In-flight failover audit failed: terminal=${recoveredTerminal?.type ?? 'missing'} terminalData=${JSON.stringify(recoveredTerminal?.data)} execution=${recoveredExecution.join(',')} rows=${executionCount} boot=${firstBootId}->${replacementBootId} head=${firstHead.join(',')}->${replacementHead.join(',')} terminals=${terminalCount} model=${initialModelRequests.length}+${continuationRequests.length} sideEffect=${JSON.stringify(sideEffectBytes)}`,
|
|
);
|
|
}
|
|
|
|
resultJson = JSON.stringify(
|
|
{
|
|
sessionId: session.id,
|
|
executionCallId: originalExecutionCallId,
|
|
executionState: recoveredExecution[1],
|
|
dispatchGeneration: Number(recoveredExecution[2]),
|
|
firstHarnessBootId: firstBootId,
|
|
replacementHarnessBootId: replacementBootId,
|
|
writerGeneration: `${firstHead[0]} -> ${replacementHead[0]}`,
|
|
journalRevision: `${firstHead[1]} -> ${replacementHead[1]}`,
|
|
committedSequence: `${firstHead[2]} -> ${replacementHead[2]}`,
|
|
promptReplayed: false,
|
|
physicalToolExecutions: 1,
|
|
terminalTurns: terminalCount,
|
|
oldHarnessDiskDeleted: !existsSync(harnessHome),
|
|
},
|
|
null,
|
|
2,
|
|
);
|
|
} else if (continuationFailover) {
|
|
await waitForTerminal(
|
|
replacementSpringUrl,
|
|
tenant,
|
|
session.id,
|
|
0,
|
|
replacementSpring,
|
|
120_000,
|
|
);
|
|
const finalEvents = await fetchJson<PublicList<PublicEvent>>(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/events?after=0&limit=100`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
const recoveredTerminal = finalEvents.data.find(
|
|
(event) => event.terminal,
|
|
);
|
|
const textDeltas = finalEvents.data.filter(
|
|
(event) => event.type === 'item.output_text.delta',
|
|
);
|
|
const visibleText = textDeltas.map((event) => eventText(event)).join('');
|
|
const recoveredExecution = runMysql(
|
|
mysqlPort,
|
|
'SELECT execution_call_id, execution_state, dispatch_generation, IF(result_json IS NULL, 0, 1), runtime_session_id FROM qwen_managed_agent.ma_tool_execution',
|
|
).split('\t');
|
|
const executionCount = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
'SELECT COUNT(*) FROM qwen_managed_agent.ma_tool_execution',
|
|
),
|
|
);
|
|
const replacementBootId = runMysql(
|
|
mysqlPort,
|
|
`SELECT harness_boot_id FROM qwen_managed_agent.ma_session WHERE ${sessionFilter}`,
|
|
);
|
|
const terminalCount = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_public_event WHERE ${sessionFilter} AND stream_kind='SESSION' AND stream_id=X'' AND is_terminal=TRUE`,
|
|
),
|
|
);
|
|
const requests = (fake?.requests ?? []).filter(({ body }) =>
|
|
JSON.stringify(body['messages']).includes(continuationMarker),
|
|
);
|
|
const originalAndReplacementRequests = requests.filter(({ body }) =>
|
|
JSON.stringify(body['messages']).includes('"role":"tool"'),
|
|
);
|
|
if (
|
|
originalAndReplacementRequests.length === 2 &&
|
|
!isDeepStrictEqual(
|
|
originalAndReplacementRequests[0].body,
|
|
originalAndReplacementRequests[1].body,
|
|
)
|
|
) {
|
|
throw new Error(
|
|
'The replacement model request does not equal the original effective request',
|
|
);
|
|
}
|
|
saveEvidence('model-request-parity.json', {
|
|
requests: originalAndReplacementRequests.map((request) => request.body),
|
|
equal:
|
|
originalAndReplacementRequests.length === 2 &&
|
|
isDeepStrictEqual(
|
|
originalAndReplacementRequests[0].body,
|
|
originalAndReplacementRequests[1].body,
|
|
),
|
|
});
|
|
|
|
const initialModelRequests = requests.filter(
|
|
({ body }) =>
|
|
!JSON.stringify(body['messages']).includes('"role":"tool"'),
|
|
);
|
|
const continuationRequests = requests.filter(({ body }) =>
|
|
JSON.stringify(body['messages']).includes('"role":"tool"'),
|
|
);
|
|
const sideEffectBytes = existsSync(inflightSideEffect)
|
|
? readFileSync(inflightSideEffect, 'utf8')
|
|
: '';
|
|
if (
|
|
recoveredTerminal?.type !== 'turn.completed' ||
|
|
visibleText !== continuationResponse ||
|
|
visibleText.includes(continuationPartial) ||
|
|
recoveredExecution.length !== 5 ||
|
|
recoveredExecution[0] !== originalExecutionCallId ||
|
|
recoveredExecution[1] !== 'SETTLED' ||
|
|
// See the in-flight arm: a live Broker keeps the generation it owns.
|
|
(harnessOnly
|
|
? !(recoveredExecution[2] === '0' || recoveredExecution[2] === '1')
|
|
: Number(recoveredExecution[2]) !== 1) ||
|
|
// See the in-flight arm: the surviving owner keeps the durable
|
|
// runtime session; a new Worker is a kill-both in disguise.
|
|
(harnessOnly && recoveredExecution[4] !== originalRuntimeSessionId) ||
|
|
(harnessOnly &&
|
|
!(
|
|
firstRuntimeHeartbeat !== undefined &&
|
|
(runtimeSessionHeartbeat(recoveredExecution[4])[1] >
|
|
firstRuntimeHeartbeat[1] ||
|
|
runtimeSessionHeartbeat(recoveredExecution[4])[0] >
|
|
firstRuntimeHeartbeat[0])
|
|
)) ||
|
|
recoveredExecution[3] !== '1' ||
|
|
executionCount !== 1 ||
|
|
replacementBootId.length === 0 ||
|
|
replacementBootId === firstBootId ||
|
|
terminalCount !== 1 ||
|
|
initialModelRequests.length !== 1 ||
|
|
continuationRequests.length !== 2 ||
|
|
sideEffectBytes !== inflightSideEffectContent
|
|
) {
|
|
throw new Error(
|
|
`Continuation failover audit failed: terminal=${recoveredTerminal?.type ?? 'missing'} text=${JSON.stringify(visibleText)} deltas=${JSON.stringify(textDeltas)} execution=${recoveredExecution.join(',')} rows=${executionCount} boot=${firstBootId}->${replacementBootId} terminals=${terminalCount} model=${initialModelRequests.length}+${continuationRequests.length} sideEffect=${JSON.stringify(sideEffectBytes)}`,
|
|
);
|
|
}
|
|
resultJson = JSON.stringify(
|
|
{
|
|
sessionId: session.id,
|
|
executionCallId: originalExecutionCallId,
|
|
firstHarnessBootId: firstBootId,
|
|
replacementHarnessBootId: replacementBootId,
|
|
promptReplayed: initialModelRequests.length !== 1,
|
|
physicalToolExecutions: executionCount,
|
|
continuationModelRequests: continuationRequests.length,
|
|
visibleText,
|
|
terminalTurns: terminalCount,
|
|
// The frozen-owner arm keeps the home on purpose (the wake
|
|
// needs it), so one key cannot mean both arms' intent.
|
|
oldHarnessDiskDeleted: !freeze && !existsSync(harnessHome),
|
|
harnessHomeRetainedForWake: freeze && existsSync(harnessHome),
|
|
},
|
|
null,
|
|
2,
|
|
);
|
|
|
|
if (freeze) {
|
|
// Wake the frozen Harness only after the replacement finished: the
|
|
// journal writer fence the takeover installed must hold against a
|
|
// very alive former writer. The Turn is terminal by now, so nothing
|
|
// but a fencing defect could mutate the binding or the journal. The
|
|
// resumed Harness still needs its home, so it must exist too.
|
|
if (!existsSync(harnessHome)) {
|
|
throw new Error(
|
|
'Frozen owner arm lost the Harness home the wake depends on',
|
|
);
|
|
}
|
|
const headBeforeWake = runMysql(
|
|
mysqlPort,
|
|
`SELECT writer_generation, journal_revision, committed_sequence FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter}`,
|
|
);
|
|
// The rest of the wake block proves the fence holds; this line
|
|
// proves there IS a fence: the head must sit on the replacement's
|
|
// writer generation, past the frozen owner's.
|
|
if (Number(headBeforeWake.split('\t')[0]) <= firstHead[0]) {
|
|
throw new Error(
|
|
`The takeover did not advance the journal head past the frozen owner's writer generation: head=${headBeforeWake} first=${firstHead.join(',')}`,
|
|
);
|
|
}
|
|
const oldGenerationTxBeforeWake = runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_session_journal_tx WHERE ${sessionFilter} AND writer_generation < (SELECT writer_generation FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter})`,
|
|
);
|
|
// Non-vacuity, proven live rather than by signal-existence: the
|
|
// frozen Harness must stop answering /health now and answer it
|
|
// again after SIGCONT, or the wake never happened and every
|
|
// fencing assertion below reports against nothing.
|
|
const frozenSilent = await fetch(
|
|
`http://127.0.0.1:${harnessPort}/health`,
|
|
{
|
|
headers: { authorization: `Bearer ${harnessToken}` },
|
|
signal: AbortSignal.timeout(1_000),
|
|
},
|
|
).then(
|
|
() => false,
|
|
() => true,
|
|
);
|
|
if (!frozenSilent) {
|
|
throw new Error(
|
|
'Frozen Harness still answers /health before SIGCONT; the freeze did not take effect',
|
|
);
|
|
}
|
|
if (!processTreeExists(harness.child)) {
|
|
throw new Error(
|
|
'Frozen Harness did not survive the freeze: the fencing proof below would be vacuous',
|
|
);
|
|
}
|
|
signalProcessTree(harness.child, 'SIGCONT');
|
|
let resumed = false;
|
|
for (let attempt = 0; attempt < 10 && !resumed; attempt += 1) {
|
|
resumed = await fetch(`http://127.0.0.1:${harnessPort}/health`, {
|
|
headers: { authorization: `Bearer ${harnessToken}` },
|
|
signal: AbortSignal.timeout(1_000),
|
|
}).then(
|
|
(response) => response.ok,
|
|
() => false,
|
|
);
|
|
if (!resumed)
|
|
await new Promise((resolve) => setTimeout(resolve, 200));
|
|
}
|
|
if (!resumed) {
|
|
throw new Error(
|
|
'Frozen Harness never resumed /health after SIGCONT; the wake never happened and the fencing assertions would be vacuous',
|
|
);
|
|
}
|
|
await new Promise((resolve) => setTimeout(resolve, 5_000));
|
|
const headAfterWake = runMysql(
|
|
mysqlPort,
|
|
`SELECT writer_generation, journal_revision, committed_sequence FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter}`,
|
|
);
|
|
const bootAfterWake = runMysql(
|
|
mysqlPort,
|
|
`SELECT harness_boot_id FROM qwen_managed_agent.ma_session WHERE ${sessionFilter}`,
|
|
);
|
|
// The fencing claim the exit check words: no transaction out of
|
|
// the old writer generation after it wakes. Lease renewal touches
|
|
// writer_lease_until only, so the head's revision moves with real
|
|
// commits; identity, not revision, is what a fencing proof may
|
|
// freeze on.
|
|
const oldGenerationTxAfterWake = runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_session_journal_tx WHERE ${sessionFilter} AND writer_generation < (SELECT writer_generation FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter})`,
|
|
);
|
|
const awakeEvents = await fetchJson<PublicList<PublicEvent>>(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/events?after=0&limit=100`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
const awakeText = awakeEvents.data
|
|
.filter((event) => event.type === 'item.output_text.delta')
|
|
.map((event) => eventText(event))
|
|
.join('');
|
|
const awakeTerminalCount = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_public_event WHERE ${sessionFilter} AND stream_kind='SESSION' AND stream_id=X'' AND is_terminal=TRUE`,
|
|
),
|
|
);
|
|
if (
|
|
headAfterWake.split('\t')[0] !== headBeforeWake.split('\t')[0] ||
|
|
oldGenerationTxAfterWake !== oldGenerationTxBeforeWake ||
|
|
bootAfterWake !== replacementBootId ||
|
|
awakeText !== visibleText ||
|
|
awakeTerminalCount !== terminalCount
|
|
) {
|
|
throw new Error(
|
|
`Frozen former Harness mutated the takeover after waking: head=${headBeforeWake}->${headAfterWake} oldWriterTx=${oldGenerationTxBeforeWake}->${oldGenerationTxAfterWake} boot=${replacementBootId}->${bootAfterWake} text=${JSON.stringify(visibleText)}->${JSON.stringify(awakeText)} terminals=${terminalCount}->${awakeTerminalCount}`,
|
|
);
|
|
}
|
|
fencedWriterJson = JSON.stringify(
|
|
{
|
|
fencedFormerWriter: true,
|
|
journalHead: headAfterWake,
|
|
oldGenerationTx: oldGenerationTxAfterWake,
|
|
harnessBootId: bootAfterWake,
|
|
visibleText: awakeText,
|
|
},
|
|
null,
|
|
2,
|
|
);
|
|
}
|
|
} else if (modelRoundFailover) {
|
|
const terminalRows = () =>
|
|
Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_public_event WHERE ${sessionFilter} AND stream_kind='SESSION' AND stream_id=X'' AND is_terminal=TRUE`,
|
|
),
|
|
);
|
|
const currentBoot = () =>
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT harness_boot_id FROM qwen_managed_agent.ma_session WHERE ${sessionFilter}`,
|
|
);
|
|
const currentHead = () =>
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT writer_generation, journal_revision, committed_sequence FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter}`,
|
|
)
|
|
.split('\t')
|
|
.map(Number);
|
|
const boots = [firstBootId];
|
|
const heads = [firstHead];
|
|
let retractedBeforeNewDeltas = true;
|
|
let previousPrefixSequence = firstPrefixSequence;
|
|
// The terminal wait watches the live control plane and the Harness
|
|
// generation finishing the round, so either one's death fails it at
|
|
// once.
|
|
let finishingHarness = replacementHarness;
|
|
if (harnessOnly) {
|
|
// Generation two reissues the saved request: by the time its own
|
|
// prefix is public, generation one's must already be retracted.
|
|
await waitUntil(
|
|
'Second generation partial text',
|
|
async () =>
|
|
(
|
|
await publicTranscript(replacementSpringUrl, tenant, session.id)
|
|
).text.includes(modelRoundPartials[1]),
|
|
120_000,
|
|
replacementHarness,
|
|
);
|
|
const secondView = await publicTranscript(
|
|
replacementSpringUrl,
|
|
tenant,
|
|
session.id,
|
|
);
|
|
const secondPrefixSequence = Math.min(
|
|
...deltaSequences(secondView.events, modelRoundPartials[1]),
|
|
);
|
|
retractedBeforeNewDeltas =
|
|
secondView.text === modelRoundPartials[1] &&
|
|
reconciledBetween(
|
|
secondView.events,
|
|
previousPrefixSequence,
|
|
secondPrefixSequence,
|
|
);
|
|
previousPrefixSequence = secondPrefixSequence;
|
|
b2SecondPrefixSequence = secondPrefixSequence;
|
|
boots.push(currentBoot());
|
|
heads.push(currentHead());
|
|
await crashChild(
|
|
replacementHarness.child,
|
|
'Replacement Hosted Harness',
|
|
);
|
|
advanceModelRound();
|
|
// The second generation's disk is scrapped like the first one's.
|
|
rmSync(replacementHarnessHome, { recursive: true, force: true });
|
|
mkdirSync(path.join(replacementHarnessHome, '.qwen'), {
|
|
recursive: true,
|
|
});
|
|
writeFileSync(
|
|
path.join(replacementHarnessHome, '.qwen', 'settings.json'),
|
|
JSON.stringify({ ui: { enableFollowupSuggestions: false } }),
|
|
{ mode: 0o600 },
|
|
);
|
|
await waitUntil(
|
|
'Second generation writer lease expiry',
|
|
(remainingMs) =>
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT IF(writer_lease_until IS NULL OR writer_lease_until < (${databaseEpochMillisSql}), 1, 0) FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter}`,
|
|
remainingMs,
|
|
) === '1',
|
|
10_000,
|
|
);
|
|
finishingHarness = await startReplacementHarness(
|
|
'Third Hosted Harness',
|
|
);
|
|
}
|
|
await waitForTerminal(
|
|
replacementSpringUrl,
|
|
tenant,
|
|
session.id,
|
|
0,
|
|
[replacementSpring, finishingHarness],
|
|
120_000,
|
|
);
|
|
const recovered = await publicTranscript(
|
|
replacementSpringUrl,
|
|
tenant,
|
|
session.id,
|
|
);
|
|
boots.push(currentBoot());
|
|
heads.push(currentHead());
|
|
retractedBeforeNewDeltas &&= reconciledBetween(
|
|
recovered.events,
|
|
previousPrefixSequence,
|
|
Math.min(...deltaSequences(recovered.events, modelRoundResponse)),
|
|
);
|
|
// Streaming requests only: the original round and one reissue per
|
|
// replaced generation, each the same saved request.
|
|
const roundRequests = (fake?.requests ?? []).filter(({ body }) => {
|
|
const messages = JSON.stringify(body['messages']);
|
|
return (
|
|
body['stream'] === true &&
|
|
messages.includes(modelRoundMarker) &&
|
|
!messages.includes(modelRoundNextMarker)
|
|
);
|
|
});
|
|
const executionCount = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
'SELECT COUNT(*) FROM qwen_managed_agent.ma_tool_execution',
|
|
),
|
|
);
|
|
const turnCount = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_turn WHERE ${sessionFilter}`,
|
|
),
|
|
);
|
|
// B1: the replacement reissues the model request but must not admit
|
|
// the prompt again. One journal transaction under the Harness prompt
|
|
// id proves one admission and one user record; the admission resource
|
|
// is published once by that same original command. The Turn's public
|
|
// id and the prompt id are generated independently, so the journal is
|
|
// keyed by the latter.
|
|
const firstPromptId = runMysql(
|
|
mysqlPort,
|
|
`SELECT prompt_id FROM qwen_managed_agent.ma_turn WHERE ${sessionFilter} ORDER BY gmt_create LIMIT 1`,
|
|
);
|
|
const admissionCount = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_session_journal_tx WHERE ${sessionFilter} AND command_id=${sqlString(firstPromptId)}`,
|
|
),
|
|
);
|
|
const admissionResources = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_resource WHERE ${sessionFilter} AND kind='managed-admission'`,
|
|
),
|
|
);
|
|
// B2: attempt identities are checked BEFORE the shared audit — a reused
|
|
// attemptId collides the journal command ids, dedups the later
|
|
// generations' attempts and breaks recovery routing itself; naming the
|
|
// identity root cause first keeps the diagnosis specific.
|
|
if (harnessOnly) {
|
|
const b2Attempts = b2ModelAttempts(
|
|
mysqlPort,
|
|
sessionFilter,
|
|
firstPromptId,
|
|
);
|
|
b2AttemptIds = [
|
|
...new Set(b2Attempts.map((attempt) => attempt.attemptId)),
|
|
];
|
|
if (b2AttemptIds.length !== modelRoundPartials.length + 1)
|
|
throw new Error(
|
|
`B2 attempt identities: expected ${modelRoundPartials.length + 1} distinct attempts, got ${b2AttemptIds.length}: ${b2Attempts.map((attempt) => `${attempt.attemptId}:${attempt.state}`).join(',')}`,
|
|
);
|
|
}
|
|
if (
|
|
recovered.terminals.length !== 1 ||
|
|
recovered.terminals[0].type !== 'turn.completed' ||
|
|
recovered.text !== modelRoundResponse ||
|
|
!retractedBeforeNewDeltas ||
|
|
terminalRows() !== 1 ||
|
|
roundRequests.length !== modelRoundPartials.length + 1 ||
|
|
roundRequests.some(
|
|
(request) => !isDeepStrictEqual(request.body, roundRequests[0].body),
|
|
) ||
|
|
executionCount !== 0 ||
|
|
turnCount !== 1 ||
|
|
admissionCount !== 1 ||
|
|
admissionResources !== 1 ||
|
|
boots.some((boot) => boot.length === 0) ||
|
|
new Set(boots).size !== boots.length ||
|
|
// Writer generation, journal revision and committed sequence all
|
|
// advance across each takeover. The negated form also fails a
|
|
// missing column or a NULL, which read as undefined or NaN.
|
|
heads.some(
|
|
(head, index) =>
|
|
index > 0 &&
|
|
![0, 1, 2].every(
|
|
(column) => head[column] > heads[index - 1][column],
|
|
),
|
|
)
|
|
) {
|
|
throw new Error(
|
|
`Model-round failover audit failed: terminals=${JSON.stringify(recovered.terminals)} text=${JSON.stringify(recovered.text)} retractedBeforeNewDeltas=${retractedBeforeNewDeltas} terminalRows=${terminalRows()} roundRequests=${roundRequests.length} equalRequests=${roundRequests.every((request) => isDeepStrictEqual(request.body, roundRequests[0]?.body))} executions=${executionCount} turns=${turnCount} admissions=${admissionCount} admissionResources=${admissionResources} boots=${boots.join('->')} heads=${heads.map((head) => head.join(',')).join('->')}`,
|
|
);
|
|
}
|
|
|
|
if (harnessOnly) {
|
|
// B2: the live SSE view captured across both crashes (drained past the
|
|
// terminal so a double projection is observed) and a Last-Event-ID: 0
|
|
// replay; the shared view audit owns the identity/blanking/terminal
|
|
// assertions so behavioural tests execute them.
|
|
if (!b2LiveCapture)
|
|
throw new Error('B2 live SSE capture was not started');
|
|
const b2Live = await b2LiveCapture;
|
|
const b2Replay = await readPublicSse(replacementSpringUrl, session.id, {
|
|
tenantId: tenant,
|
|
lastEventId: '0',
|
|
until: (events) => events.some((event) => event.terminal),
|
|
timeoutMs: 60_000,
|
|
});
|
|
const b2Views = assertB2StreamViews({
|
|
live: b2Live.events,
|
|
reload: recovered.events,
|
|
replay: b2Replay.events,
|
|
partials: modelRoundPartials,
|
|
answer: modelRoundResponse,
|
|
firstPrefixSequence,
|
|
secondPrefixSequence: b2SecondPrefixSequence,
|
|
originalTurnId: recovered.terminals[0]?.turn_id ?? undefined,
|
|
});
|
|
writeEvidenceFile(modelRoundEvidenceDirectory, 'b2-identities.json', {
|
|
attemptIds: b2AttemptIds,
|
|
live: {
|
|
partial1: b2Views.livePartial1,
|
|
partial2: b2Views.livePartial2,
|
|
answer: b2Views.liveAnswer,
|
|
frames: b2Live.events.length,
|
|
lastId: b2Live.lastId,
|
|
},
|
|
reloadAnswer: b2Views.reloadAnswer,
|
|
replay: {
|
|
answer: b2Views.replayAnswer,
|
|
frames: b2Replay.events.length,
|
|
},
|
|
});
|
|
}
|
|
|
|
if (frozenControlPlane) {
|
|
// Q2's binding half: the former control plane is alive, its claim
|
|
// expired and its Session binding was moved by the replacement. On
|
|
// wake it must still hold its stale claim in memory and try it —
|
|
// and nothing it tries may move the binding, the Turn, the public
|
|
// feed or the journal.
|
|
const fencedProbes = 7;
|
|
const fencedState = (timeoutMs = fencedProbes * 10_000) => {
|
|
// The probes are synchronous, so waitUntil cannot cut one short: a
|
|
// budgeted call splits its budget over them (runMysql still gives
|
|
// each at least 2 s), and a one-shot call keeps 10 s per probe.
|
|
const probeMs = Math.max(1, Math.floor(timeoutMs / fencedProbes));
|
|
return {
|
|
session: runMysql(
|
|
mysqlPort,
|
|
`SELECT harness_boot_id, harness_event_epoch, harness_last_event_id, status FROM qwen_managed_agent.ma_session WHERE ${sessionFilter}`,
|
|
probeMs,
|
|
),
|
|
turns: runMysql(
|
|
mysqlPort,
|
|
`SELECT turn_id, status, version, dispatch_owner, dispatch_lease_until, retry_count, is_submission_attempted, harness_event_epoch, harness_last_event_id, error_code FROM qwen_managed_agent.ma_turn WHERE ${sessionFilter} ORDER BY turn_id`,
|
|
probeMs,
|
|
),
|
|
// Row by row: a retraction rewrites delta payloads and identities
|
|
// in place, which no count or maximum would notice.
|
|
events: runMysql(
|
|
mysqlPort,
|
|
`SELECT stream_kind, HEX(stream_id), sequence_id, event_type, is_terminal, HEX(source_key), SHA2(data_json, 256), HEX(item_id), HEX(content_part_id), SHA2(text, 256), state FROM qwen_managed_agent.ma_public_event WHERE ${sessionFilter} ORDER BY stream_kind, stream_id, sequence_id`,
|
|
probeMs,
|
|
),
|
|
// Every Spring also sweeps expired operation claims on a timer, so
|
|
// the woken one writes ma_operation too; none of it may land here.
|
|
operations: runMysql(
|
|
mysqlPort,
|
|
`SELECT HEX(operation_key), operation_kind, state, phase, delivery_state, claim_owner FROM qwen_managed_agent.ma_operation WHERE ${sessionFilter} ORDER BY 1`,
|
|
probeMs,
|
|
),
|
|
// The head's state, writer, checkpoint and recovery columns. Its
|
|
// lease, activation and revision columns move with every
|
|
// activation renewal the live writer journals.
|
|
head: runMysql(
|
|
mysqlPort,
|
|
`SELECT state, storage_version, writer_generation, HEX(writer_id), HEX(latest_checkpoint_resource_id), compacted_through_revision, recovery_status, recovery_detail_code FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter}`,
|
|
probeMs,
|
|
),
|
|
// Every journal transaction except those renewals, by writer
|
|
// generation and operation; an older generation's renewals stay
|
|
// counted in staleWriterTx below.
|
|
journal: runMysql(
|
|
mysqlPort,
|
|
`SELECT writer_generation, operation, COUNT(*), MAX(journal_revision) FROM qwen_managed_agent.ma_session_journal_tx WHERE ${sessionFilter} AND operation <> 'renewActivation' GROUP BY writer_generation, operation ORDER BY writer_generation, operation`,
|
|
probeMs,
|
|
),
|
|
staleWriterTx: runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_session_journal_tx WHERE ${sessionFilter} AND writer_generation < (SELECT writer_generation FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter})`,
|
|
probeMs,
|
|
),
|
|
};
|
|
};
|
|
// The replacement must go quiet first: a late write of its own inside
|
|
// the wake window would otherwise read as a stale mutation.
|
|
// The window matches the wait after the wake, so a replacement-side
|
|
// writer slower than the window cannot pass here and fail there.
|
|
// A dead owner would hold every snapshot still as well, so each wait
|
|
// from here on watches the replacement pair, and so does the verdict.
|
|
const owners = [replacementSpring, finishingHarness];
|
|
let beforeWake = fencedState();
|
|
await waitUntil(
|
|
'Replacement quiet before the wake',
|
|
async (remainingMs) => {
|
|
await new Promise((resolve) => setTimeout(resolve, 6_000));
|
|
const settled = fencedState(Math.max(1, remainingMs - 6_000));
|
|
const moved = (
|
|
Object.keys(settled) as Array<keyof typeof settled>
|
|
).filter(
|
|
(key) => !isDeepStrictEqual(beforeWake[key], settled[key]),
|
|
);
|
|
const previous = beforeWake;
|
|
beforeWake = settled;
|
|
if (moved.length > 0) {
|
|
throw new Error(
|
|
`still moving: ${moved.map((key) => `${key}: ${JSON.stringify(previous[key]).slice(0, 1_000)} -> ${JSON.stringify(settled[key]).slice(0, 1_000)}`).join(' | ')}`,
|
|
);
|
|
}
|
|
return true;
|
|
},
|
|
90_000,
|
|
owners,
|
|
);
|
|
if (beforeWake.session.split('\t')[0] !== boots[boots.length - 1]) {
|
|
throw new Error(
|
|
`The replacement did not move the Session binding: ${beforeWake.session}`,
|
|
);
|
|
}
|
|
const formerProbes = [
|
|
[`${springUrl}/actuator/health`, 'former Spring'],
|
|
[`http://127.0.0.1:${harnessPort}/health`, 'former Harness'],
|
|
] as const;
|
|
const answers = (url: string) =>
|
|
fetch(url, {
|
|
headers: { authorization: `Bearer ${harnessToken}` },
|
|
signal: AbortSignal.timeout(1_000),
|
|
}).then(
|
|
() => true,
|
|
() => false,
|
|
);
|
|
// Non-vacuity: alive but silent before SIGCONT, answering after.
|
|
for (const [url, name] of formerProbes) {
|
|
if (await answers(url)) {
|
|
throw new Error(
|
|
`The ${name} answered before SIGCONT; the freeze did not take effect`,
|
|
);
|
|
}
|
|
}
|
|
for (const child of frozenChildren) {
|
|
if (!processTreeExists(child.child)) {
|
|
throw new Error(`${child.name} did not survive the freeze`);
|
|
}
|
|
}
|
|
// A woken Harness with no home would leave every check below
|
|
// holding by absence instead of by fencing.
|
|
if (!existsSync(harnessHome)) {
|
|
throw new Error(
|
|
'Frozen control-plane arm lost the Harness home the wake depends on',
|
|
);
|
|
}
|
|
// The writer axis has its own witness: the woken Harness still holds
|
|
// its stale writer grant, and only the store's writer-grant check
|
|
// answers with this refusal. It must not appear before the wake.
|
|
// After the wake the Harness's whole output is kept, not the log's
|
|
// rolling tail, so a busy Harness cannot push the refusal out.
|
|
const staleWriterRefusal =
|
|
'The Managed Session writer grant is stale or unavailable';
|
|
if (harness.log().includes(staleWriterRefusal)) {
|
|
throw new Error(
|
|
`The former Harness met the writer fence before the wake:\n${harness.log()}`,
|
|
);
|
|
}
|
|
let harnessOutputSinceWake = '';
|
|
const recordSinceWake = (chunk: Buffer) => {
|
|
harnessOutputSinceWake += chunk.toString();
|
|
};
|
|
harness.child.stdout?.on('data', recordSinceWake);
|
|
harness.child.stderr?.on('data', recordSinceWake);
|
|
const staleWriterRefusals = () =>
|
|
harnessOutputSinceWake.split(staleWriterRefusal).length - 1;
|
|
// The general log names every statement the woken Spring sends, so
|
|
// a claim UPDATE carrying the former owner id proves the wake really
|
|
// exercised the stale authority instead of idling past the check.
|
|
runMysql(mysqlPort, "SET GLOBAL log_output = 'TABLE'");
|
|
runMysql(mysqlPort, "SET GLOBAL general_log = 'ON'");
|
|
const wokeAt = runMysql(mysqlPort, 'SELECT NOW(6)');
|
|
const modelRequestsBeforeWake = fake?.requests.length ?? 0;
|
|
for (const child of frozenChildren.splice(0)) {
|
|
signalProcessTree(child.child, 'SIGCONT');
|
|
}
|
|
await waitUntil(
|
|
'Former control plane wake',
|
|
async () =>
|
|
(
|
|
await Promise.all(formerProbes.map(([url]) => answers(url)))
|
|
).every(Boolean),
|
|
30_000,
|
|
owners,
|
|
);
|
|
// The witness is a renewal or retry the woken coordinator still sends
|
|
// on its stale claim; a coordinator that stopped sending a write it
|
|
// knows is stale would need another witness, not a skipped one.
|
|
const staleClaimUpdates = (timeoutMs = 10_000) =>
|
|
Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM mysql.general_log WHERE event_time >= ${sqlString(wokeAt)} AND command_type = 'Query' AND CONVERT(argument USING utf8mb4) REGEXP 'UPDATE[[:space:]]+ma_turn[[:space:]]' AND CONVERT(argument USING utf8mb4) NOT LIKE '%general_log%' AND LOCATE(${sqlString(formerDispatchOwner)}, CONVERT(argument USING utf8mb4)) > 0`,
|
|
timeoutMs,
|
|
),
|
|
);
|
|
await waitUntil(
|
|
'Woken control plane exercises its stale claim',
|
|
(remainingMs) => staleClaimUpdates(remainingMs) > 0,
|
|
30_000,
|
|
owners,
|
|
);
|
|
await waitUntil(
|
|
'Woken Harness meets the writer fence',
|
|
() => staleWriterRefusals() > 0,
|
|
30_000,
|
|
owners,
|
|
);
|
|
// Several dispatch lease periods for anything the wake set in motion.
|
|
await new Promise((resolve) => setTimeout(resolve, 6_000));
|
|
const afterWake = fencedState();
|
|
// The probes are synchronous: let the loop record an exit during
|
|
// them before the owners are checked.
|
|
await new Promise((resolve) => setImmediate(resolve));
|
|
for (const owner of owners) {
|
|
if (childExited(owner.child)) {
|
|
throw new Error(
|
|
`${owner.name} exited inside the wake window\n${owner.log()}`,
|
|
);
|
|
}
|
|
}
|
|
const moved = (
|
|
Object.keys(afterWake) as Array<keyof typeof afterWake>
|
|
).filter((key) => !isDeepStrictEqual(beforeWake[key], afterWake[key]));
|
|
const staleAttempts = staleClaimUpdates();
|
|
// Evidence, not a gate: every write the woken Spring sent on the
|
|
// connections that carried its stale owner id, by statement head.
|
|
const staleConnectionWrites = runMysql(
|
|
mysqlPort,
|
|
`SELECT LEFT(CONVERT(argument USING utf8mb4), 48), COUNT(*) FROM mysql.general_log WHERE event_time >= ${sqlString(wokeAt)} AND command_type = 'Query' AND CONVERT(argument USING utf8mb4) REGEXP '^[[:space:]]*(UPDATE|INSERT|DELETE|REPLACE)' AND thread_id IN (SELECT thread_id FROM (SELECT DISTINCT thread_id FROM mysql.general_log WHERE event_time >= ${sqlString(wokeAt)} AND CONVERT(argument USING utf8mb4) NOT LIKE '%general_log%' AND LOCATE(${sqlString(formerDispatchOwner)}, CONVERT(argument USING utf8mb4)) > 0) AS stale) GROUP BY 1 ORDER BY 1`,
|
|
)
|
|
.split('\n')
|
|
.filter(Boolean);
|
|
runMysql(mysqlPort, "SET GLOBAL general_log = 'OFF'");
|
|
const awake = await publicTranscript(
|
|
replacementSpringUrl,
|
|
tenant,
|
|
session.id,
|
|
);
|
|
// staleWriterTx counts the former writer's commits from before the
|
|
// freeze too, so it is compared across the wake, never against zero.
|
|
if (
|
|
!isDeepStrictEqual(beforeWake, afterWake) ||
|
|
awake.text !== modelRoundResponse ||
|
|
awake.terminals.length !== 1
|
|
) {
|
|
throw new Error(
|
|
`Woken former control plane mutated the takeover: moved=${moved.join(',') || 'none'} before=${JSON.stringify(beforeWake)} after=${JSON.stringify(afterWake)} text=${JSON.stringify(awake.text)} terminals=${awake.terminals.length}`,
|
|
);
|
|
}
|
|
fencedWriterJson = JSON.stringify(
|
|
{
|
|
fencedFormerControlPlane: true,
|
|
formerDispatchOwner,
|
|
staleClaimUpdates: staleAttempts,
|
|
staleWriterRefusals: staleWriterRefusals(),
|
|
staleConnectionWrites,
|
|
harnessBootId: afterWake.session.split('\t')[0],
|
|
journalHead: afterWake.head,
|
|
staleWriterTx: afterWake.staleWriterTx,
|
|
turn: afterWake.turns,
|
|
modelRequestsAfterWake:
|
|
(fake?.requests.length ?? 0) - modelRequestsBeforeWake,
|
|
visibleText: awake.text,
|
|
},
|
|
null,
|
|
2,
|
|
);
|
|
// Two live control planes over one Session would race for the next
|
|
// Turn's claim — routing that is out of scope — so the former pair
|
|
// leaves before the Session is used again.
|
|
await crashChild(harness.child, `${hostedHarnessLabel} (original)`);
|
|
await crashChild(
|
|
spring.child,
|
|
'Spring Managed Agent Server (original)',
|
|
);
|
|
}
|
|
|
|
// The Session keeps working, and what the next Turn's model sees is
|
|
// the recovered history: one copy of the original prompt and answer,
|
|
// none of the retracted prefixes.
|
|
const cursor = Math.max(
|
|
0,
|
|
...recovered.events.map((event) => event.sequence),
|
|
);
|
|
const nextResponse = await fetch(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/events`,
|
|
{
|
|
method: 'POST',
|
|
headers: {
|
|
'content-type': 'application/json',
|
|
'idempotency-key': `model-round-next-${session.id}`,
|
|
...tenantHeaders(tenant),
|
|
},
|
|
body: JSON.stringify({
|
|
type: 'agent.session.input.message',
|
|
input: [
|
|
{
|
|
type: 'text',
|
|
text: `${modelRoundNextMarker}. Reply exactly ${modelRoundNextResponse}.`,
|
|
},
|
|
],
|
|
}),
|
|
},
|
|
);
|
|
if (nextResponse.status !== 202) {
|
|
throw new Error(
|
|
`Next model-round Turn returned ${nextResponse.status}: ${await nextResponse.text()}`,
|
|
);
|
|
}
|
|
const nextTurn = await waitForTerminal(
|
|
replacementSpringUrl,
|
|
tenant,
|
|
session.id,
|
|
cursor,
|
|
replacementSpring,
|
|
60_000,
|
|
);
|
|
const nextTerminals = nextTurn.events.filter((event) => event.terminal);
|
|
const nextText = nextTurn.events
|
|
.filter((event) => event.type === 'item.output_text.delta')
|
|
.map(eventText)
|
|
.join('');
|
|
const nextRequest = [...(fake?.requests ?? [])]
|
|
.reverse()
|
|
.find(({ body }) =>
|
|
JSON.stringify(body['messages']).includes(modelRoundNextMarker),
|
|
);
|
|
const history = JSON.stringify(nextRequest?.body['messages'] ?? []);
|
|
const occurrences = (needle: string) => history.split(needle).length - 1;
|
|
// The answer string also sits inside the original prompt, so one
|
|
// prompt plus one answer is exactly two.
|
|
if (
|
|
nextTerminals.length !== 1 ||
|
|
nextTerminals[0].type !== 'turn.completed' ||
|
|
nextText !== modelRoundNextResponse ||
|
|
terminalRows() !== 2 ||
|
|
occurrences(modelRoundMarker) !== 1 ||
|
|
occurrences(modelRoundResponse) !== 2 ||
|
|
modelRoundPartials.some((partial) => history.includes(partial))
|
|
) {
|
|
throw new Error(
|
|
`Next Turn after the model-round takeover failed: terminals=${JSON.stringify(nextTerminals)} text=${JSON.stringify(nextText)} terminalRows=${terminalRows()} history=${history}`,
|
|
);
|
|
}
|
|
|
|
resultJson = JSON.stringify(
|
|
{
|
|
sessionId: session.id,
|
|
harnessBootIds: boots,
|
|
writerGenerations: heads.map((head) => head[0]),
|
|
journalRevisions: heads.map((head) => head[1]),
|
|
committedSequences: heads.map((head) => head[2]),
|
|
replacedGenerations: modelRoundPartials.length,
|
|
modelRoundRequests: roundRequests.length,
|
|
equalSavedRequests: true,
|
|
retractedPrefixes: modelRoundPartials,
|
|
visibleText: recovered.text,
|
|
physicalToolExecutions: executionCount,
|
|
terminalTurns: terminalRows(),
|
|
journalAdmissions: admissionCount,
|
|
admissionResources,
|
|
nextTurnHistoryClean: true,
|
|
// As in the frozen-owner arm, one key cannot mean both intents.
|
|
oldHarnessDiskDeleted:
|
|
!frozenControlPlane && !existsSync(harnessHome),
|
|
harnessHomeRetainedForWake:
|
|
frozenControlPlane && existsSync(harnessHome),
|
|
},
|
|
null,
|
|
2,
|
|
);
|
|
} else {
|
|
const secondResponse = await fetch(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/events`,
|
|
{
|
|
method: 'POST',
|
|
headers: {
|
|
'content-type': 'application/json',
|
|
'idempotency-key': `failover-second-${Date.now()}`,
|
|
...tenantHeaders(tenant),
|
|
},
|
|
body: JSON.stringify({
|
|
type: 'agent.session.input.message',
|
|
input: [
|
|
{
|
|
type: 'text',
|
|
text: `${failoverSecondMarker}. Use the prior conversation and reply exactly ${failoverSecondResponse}.`,
|
|
},
|
|
],
|
|
}),
|
|
},
|
|
);
|
|
if (secondResponse.status !== 202) {
|
|
throw new Error(
|
|
`Second failover Turn returned ${secondResponse.status}: ${await secondResponse.text()}`,
|
|
);
|
|
}
|
|
const secondTurn = await waitForTerminal(
|
|
replacementSpringUrl,
|
|
tenant,
|
|
session.id,
|
|
firstTurnLastSequence,
|
|
replacementSpring,
|
|
);
|
|
const secondTerminal = secondTurn.events.find((event) => event.terminal);
|
|
if (secondTerminal?.type !== 'turn.completed') {
|
|
throw new Error(
|
|
`Second failover Turn ended with ${secondTerminal?.type ?? 'no terminal event'}`,
|
|
);
|
|
}
|
|
|
|
if (bigOutput) {
|
|
assertPublicAnswer(secondTurn.events, failoverSecondResponse);
|
|
assertStoredAnswer(
|
|
mysqlPort,
|
|
tenant,
|
|
session.id,
|
|
failoverFirstResponse,
|
|
);
|
|
assertStoredAnswer(
|
|
mysqlPort,
|
|
tenant,
|
|
session.id,
|
|
failoverSecondResponse,
|
|
);
|
|
}
|
|
|
|
const secondRequest = [...(fake?.requests ?? [])]
|
|
.reverse()
|
|
.find(({ body }) =>
|
|
JSON.stringify(body['messages']).includes(failoverSecondMarker),
|
|
);
|
|
const restoredMessages = JSON.stringify(
|
|
secondRequest?.body['messages'] ?? [],
|
|
);
|
|
if (
|
|
!restoredMessages.includes(failoverFirstMarker) ||
|
|
!restoredMessages.includes(failoverFirstResponse) ||
|
|
restoredMessages.includes(failoverMissingResponse)
|
|
) {
|
|
throw new Error(
|
|
'Replacement Harness did not restore first-Turn context',
|
|
);
|
|
}
|
|
|
|
const secondBootId = runMysql(
|
|
mysqlPort,
|
|
`SELECT harness_boot_id FROM qwen_managed_agent.ma_session WHERE ${sessionFilter}`,
|
|
);
|
|
const secondHead = runMysql(
|
|
mysqlPort,
|
|
`SELECT writer_generation, journal_revision, committed_sequence FROM qwen_managed_agent.ma_session_head WHERE ${sessionFilter}`,
|
|
)
|
|
.split('\t')
|
|
.map(Number);
|
|
const terminalCount = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_public_event WHERE ${sessionFilter} AND stream_kind='SESSION' AND stream_id=X'' AND is_terminal=TRUE`,
|
|
),
|
|
);
|
|
if (
|
|
secondBootId.length === 0 ||
|
|
secondBootId === firstBootId ||
|
|
secondHead.length !== 3 ||
|
|
secondHead[0] <= firstHead[0] ||
|
|
secondHead[1] <= firstHead[1] ||
|
|
secondHead[2] <= firstHead[2] ||
|
|
terminalCount !== 2
|
|
) {
|
|
throw new Error(
|
|
`Failover audit failed: boot=${firstBootId}->${secondBootId} head=${firstHead.join(',')}->${secondHead.join(',')} terminals=${terminalCount}`,
|
|
);
|
|
}
|
|
|
|
resultJson = JSON.stringify(
|
|
{
|
|
sessionId: session.id,
|
|
firstHarnessBootId: firstBootId,
|
|
replacementHarnessBootId: secondBootId,
|
|
writerGeneration: `${firstHead[0]} -> ${secondHead[0]}`,
|
|
journalRevision: `${firstHead[1]} -> ${secondHead[1]}`,
|
|
committedSequence: `${firstHead[2]} -> ${secondHead[2]}`,
|
|
terminalTurns: terminalCount,
|
|
...(bigOutput
|
|
? {
|
|
expectedCharacters: failoverFirstResponse.length,
|
|
expectedUtf8Bytes: Buffer.byteLength(failoverFirstResponse),
|
|
shortControlCharacters: failoverSecondResponse.length,
|
|
fullTextPreserved: true,
|
|
}
|
|
: {}),
|
|
restoredFirstTurnContext: true,
|
|
oldHarnessDiskDeleted: !existsSync(harnessHome),
|
|
},
|
|
null,
|
|
2,
|
|
);
|
|
}
|
|
if (verificationMode) {
|
|
// These assertions run after every A/B primary gate, never in place of it.
|
|
await waitUntil(
|
|
'Runtime cleanup before next Turn',
|
|
() => {
|
|
const state = runMysql(
|
|
mysqlPort,
|
|
`SELECT session_state FROM qwen_managed_agent.ma_runtime_session WHERE runtime_session_id=${sqlString(originalRuntimeSessionId!)}`,
|
|
);
|
|
const confirmed = runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_session_journal_tx WHERE ${sessionFilter} AND command_id=${sqlString(`hosted-cleanup:${originalRuntimeSessionId}:${originalRuntimeSessionId}:confirmed`)}`,
|
|
);
|
|
return state === 'RELEASED' && confirmed === '1';
|
|
},
|
|
30_000,
|
|
replacementHarness,
|
|
);
|
|
const originalPage = await fetchJson<PublicList<PublicEvent>>(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/events?after=0&limit=100`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
if (originalPage.data.filter((event) => event.terminal).length !== 1)
|
|
throw new Error(
|
|
'Original Turn must have exactly one public terminal before relief',
|
|
);
|
|
const originalCursor = Math.max(
|
|
0,
|
|
...originalPage.data.map((event) => event.sequence),
|
|
);
|
|
const reliefUrl = `${replacementSpringUrl}/v1/agents/sessions/${session.id}/events`;
|
|
const reliefHeaders = {
|
|
'content-type': 'application/json',
|
|
'idempotency-key': `g3-relief-${session.id}`,
|
|
...tenantHeaders(tenant),
|
|
};
|
|
const reliefBody = JSON.stringify({
|
|
type: 'agent.session.input.message',
|
|
input: [
|
|
{
|
|
type: 'text',
|
|
text: `${reliefMarker}. Reply exactly ${reliefResponse}. Use no tools.`,
|
|
},
|
|
],
|
|
});
|
|
if (approvalFault === 'admission-reply-lost') {
|
|
await postAndDropReply(reliefUrl, reliefBody, reliefHeaders);
|
|
}
|
|
const relief = await fetch(reliefUrl, {
|
|
method: 'POST',
|
|
headers: reliefHeaders,
|
|
body: reliefBody,
|
|
});
|
|
if (relief.status !== 202)
|
|
throw new Error(
|
|
`Relief Turn admission returned ${relief.status}: ${await relief.text()}`,
|
|
);
|
|
const reliefTurn = await waitForTerminal(
|
|
replacementSpringUrl,
|
|
tenant,
|
|
session.id,
|
|
originalCursor,
|
|
replacementSpring,
|
|
60_000,
|
|
);
|
|
const reliefTerminals = reliefTurn.events.filter(
|
|
(event) => event.terminal,
|
|
);
|
|
const reliefText = reliefTurn.events
|
|
.filter((event) => event.type === 'item.output_text.delta')
|
|
.map(eventText)
|
|
.join('');
|
|
if (
|
|
reliefTerminals.length !== 1 ||
|
|
reliefTerminals[0].type !== 'turn.completed' ||
|
|
reliefText !== reliefResponse
|
|
)
|
|
throw new Error(
|
|
`Relief Turn did not finish: terminals=${JSON.stringify(reliefTerminals)} text=${JSON.stringify(reliefText)}`,
|
|
);
|
|
const reliefRequests = (fake?.requests ?? []).filter((request) =>
|
|
JSON.stringify(request.body.messages).includes(reliefMarker),
|
|
);
|
|
if (reliefRequests.length !== 1)
|
|
throw new Error(
|
|
`Relief Turn inference count differs: ${reliefRequests.length}`,
|
|
);
|
|
if (approvalFault === 'admission-reply-lost') {
|
|
// C4: the retried Idempotency-Key must land on the original
|
|
// admission. A second Turn row would mean the lost reply was
|
|
// re-admitted instead of reconciled.
|
|
const turnRows = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_turn WHERE ${sessionFilter}`,
|
|
),
|
|
);
|
|
if (turnRows !== 2)
|
|
throw new Error(
|
|
`Lost admission reply produced ${turnRows} Turn rows, not two`,
|
|
);
|
|
}
|
|
const finalTerminalCount = Number(
|
|
runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*) FROM qwen_managed_agent.ma_public_event WHERE ${sessionFilter} AND stream_kind='SESSION' AND stream_id=X'' AND is_terminal=TRUE`,
|
|
),
|
|
);
|
|
if (finalTerminalCount !== 2)
|
|
throw new Error(
|
|
`Expected one terminal per original/relief Turn, observed ${finalTerminalCount}`,
|
|
);
|
|
saveEvidence('post-turn-events.json', {
|
|
cursorBeforeRelief: originalCursor,
|
|
original: originalPage.data,
|
|
relief: reliefTurn.events,
|
|
sqlTerminalCount: finalTerminalCount,
|
|
});
|
|
const close = await fetch(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/close`,
|
|
{
|
|
method: 'POST',
|
|
headers: {
|
|
'idempotency-key': `g3-close-${session.id}`,
|
|
...tenantHeaders(tenant),
|
|
},
|
|
},
|
|
);
|
|
if (close.status !== 202)
|
|
throw new Error(
|
|
`Close admission returned ${close.status}: ${await close.text()}`,
|
|
);
|
|
const closeOperation = (await close.json()) as Record<string, unknown>;
|
|
await waitUntil(
|
|
'Reliable close operation completion',
|
|
async () => {
|
|
const operation = await fetchJson<Record<string, unknown>>(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}/operations/${closeOperation.id}`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
if (operation.status === 'failed')
|
|
throw new Error(
|
|
`Close operation failed: ${JSON.stringify(operation)}`,
|
|
);
|
|
return operation.status === 'completed';
|
|
},
|
|
60_000,
|
|
replacementSpring,
|
|
);
|
|
const closedSession = await fetchJson<Record<string, unknown>>(
|
|
`${replacementSpringUrl}/v1/agents/sessions/${session.id}`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
if (closedSession.status !== 'closed')
|
|
throw new Error(
|
|
`Reliable close completed with Session status ${closedSession.status}`,
|
|
);
|
|
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,
|
|
originalTerminal: verifiedOriginalTerminal ?? 'turn.completed',
|
|
nextTurnCompleted: true,
|
|
nextTurnRequests: reliefRequests.length,
|
|
terminalCount: finalTerminalCount,
|
|
closeAdmission: close.status,
|
|
closeCompleted: true,
|
|
oldHarnessHomeDeleted: !existsSync(harnessHome),
|
|
};
|
|
saveEvidence('summary.json', summary);
|
|
g3SummaryJson = JSON.stringify(summary, null, 2);
|
|
}
|
|
} else {
|
|
const idempotencyKey = `create-${Date.now()}`;
|
|
const prompt = [
|
|
'Before using any tool, emit the visible text MODEL_READY.',
|
|
'Then call write_file exactly once to write the exact text',
|
|
JSON.stringify(sideEffectContent),
|
|
'to the relative path',
|
|
JSON.stringify(sideEffectName),
|
|
'under the session working directory.',
|
|
'Call no other tool before or after it.',
|
|
'After the tool succeeds, reply TOOL_DONE. Do not ask a question.',
|
|
].join(' ');
|
|
const body = JSON.stringify({
|
|
agent_id: 'qwen-code',
|
|
input: [{ type: 'text', text: prompt }],
|
|
workspace: { workspace_id: boundWorkspaceId },
|
|
metadata: { title: 'Real model cold Runtime E2E' },
|
|
});
|
|
const requestStartedAt = Date.now();
|
|
const createResponse = await fetch(`${springUrl}/v1/agents/sessions`, {
|
|
method: 'POST',
|
|
headers: {
|
|
'content-type': 'application/json',
|
|
'idempotency-key': idempotencyKey,
|
|
...tenantHeaders(tenant),
|
|
},
|
|
body,
|
|
});
|
|
if (createResponse.status !== 202) {
|
|
throw new Error(
|
|
`Session create returned ${createResponse.status}: ${await createResponse.text()}`,
|
|
);
|
|
}
|
|
const createdAt = Date.now();
|
|
const session = (await createResponse.json()) as PublicSession;
|
|
const observed = new Map<
|
|
number,
|
|
{ event: PublicEvent; observedAt: number }
|
|
>();
|
|
let after = 0;
|
|
await waitUntil(
|
|
'Managed Turn terminal event',
|
|
async () => {
|
|
const page = await fetchJson<PublicList<PublicEvent>>(
|
|
`${springUrl}/v1/agents/sessions/${session.id}/events?after=${after}&limit=100`,
|
|
{ headers: tenantHeaders(tenant) },
|
|
);
|
|
const now = Date.now();
|
|
for (const event of page.data) {
|
|
observed.set(event.sequence, { event, observedAt: now });
|
|
after = Math.max(after, event.sequence);
|
|
}
|
|
return page.data.some((event) => event.terminal);
|
|
},
|
|
180_000,
|
|
spring,
|
|
);
|
|
|
|
const ordered = [...observed.values()].sort(
|
|
(left, right) => left.event.sequence - right.event.sequence,
|
|
);
|
|
const firstModel = ordered.find(({ event }) =>
|
|
['item.output_text.delta', 'item.reasoning.delta'].includes(event.type),
|
|
);
|
|
const runtimeReady = ordered.find(
|
|
({ event }) => event.type === 'environment.ready',
|
|
);
|
|
const terminal = ordered.find(({ event }) => event.terminal);
|
|
if (!firstModel || !runtimeReady || !terminal) {
|
|
throw new Error(
|
|
`Expected model, Runtime, and terminal events; got ${ordered.map(({ event }) => event.type).join(', ')}`,
|
|
);
|
|
}
|
|
if (terminal.event.type !== 'turn.completed') {
|
|
throw new Error(`Managed Turn ended with ${terminal.event.type}`);
|
|
}
|
|
if (
|
|
runtimeDelayMs >= modelBeforeRuntimeAssertionDelayMs &&
|
|
firstModel.event.sequence >= runtimeReady.event.sequence
|
|
) {
|
|
throw new Error(
|
|
'First model event did not precede Runtime readiness ' +
|
|
`(firstModelSequence=${firstModel.event.sequence}, ` +
|
|
`runtimeReadySequence=${runtimeReady.event.sequence}, ` +
|
|
`firstModelEventMs=${firstModel.observedAt - requestStartedAt}, ` +
|
|
`runtimeReadyMs=${runtimeReady.observedAt - requestStartedAt}, ` +
|
|
`runtimeDelayMs=${runtimeDelayMs})`,
|
|
);
|
|
}
|
|
if (!existsSync(sideEffect)) {
|
|
throw new Error('Tool side effect file was not created');
|
|
}
|
|
if (readFileSync(sideEffect, 'utf8') !== sideEffectContent) {
|
|
throw new Error('Tool side effect content did not match');
|
|
}
|
|
// Tool calls execute through the Broker worker, which does not publish
|
|
// item.tool_call.* public events without O2 publication; the durable
|
|
// execution record is the public-feed-independent proof.
|
|
const executions = runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*), GROUP_CONCAT(DISTINCT execution_state) FROM qwen_managed_agent.ma_tool_execution`,
|
|
).split('\t');
|
|
if (executions[0] !== '1' || executions[1] !== 'SETTLED') {
|
|
// The temporary MySQL data dir is deleted on exit, so the failure
|
|
// message is the only place the offending executions can still be
|
|
// identified: two tool_call_ids mean the model called another tool,
|
|
// one tool_call_id across two rows means a duplicate dispatch.
|
|
const rows = runMysql(
|
|
mysqlPort,
|
|
'SELECT execution_call_id, tool_call_id, execution_state FROM qwen_managed_agent.ma_tool_execution',
|
|
);
|
|
throw new Error(
|
|
`Tool execution audit failed: ${executions.join(',')}\n${rows}`,
|
|
);
|
|
}
|
|
|
|
const replay = await fetch(`${springUrl}/v1/agents/sessions`, {
|
|
method: 'POST',
|
|
headers: {
|
|
'content-type': 'application/json',
|
|
'idempotency-key': idempotencyKey,
|
|
...tenantHeaders(tenant),
|
|
},
|
|
body,
|
|
});
|
|
const replaySession = (await replay.json()) as PublicSession;
|
|
if (
|
|
replay.status !== 202 ||
|
|
replaySession.id !== session.id ||
|
|
replay.headers.get('x-qwen-idempotent-replay') !== 'true'
|
|
) {
|
|
throw new Error('Idempotent create replay did not return the Session');
|
|
}
|
|
// The probe carries an actor holding a read grant on the Session's
|
|
// Workspace, so its 404 can only come from the tenant scope: without the
|
|
// actor, a bound Session 404s through the grant check and would mask a
|
|
// tenant-scoping leak.
|
|
const crossTenant = await fetch(
|
|
`${springUrl}/v1/agents/sessions/${session.id}`,
|
|
{ headers: tenantHeaders('other-tenant') },
|
|
);
|
|
if (crossTenant.status !== 404) {
|
|
throw new Error(`Cross-tenant lookup returned ${crossTenant.status}`);
|
|
}
|
|
const durable = runMysql(
|
|
mysqlPort,
|
|
`SELECT COUNT(*), COUNT(DISTINCT sequence_id), SUM(is_terminal) FROM qwen_managed_agent.ma_public_event WHERE tenant_id='${tenant}' AND session_id='${session.id}' AND stream_kind='SESSION' AND stream_id=X''`,
|
|
).split('\t');
|
|
if (
|
|
durable.length !== 3 ||
|
|
durable[0] !== durable[1] ||
|
|
durable[2] !== '1'
|
|
) {
|
|
throw new Error(`Durable event audit failed: ${durable.join(',')}`);
|
|
}
|
|
|
|
resultJson = JSON.stringify(
|
|
{
|
|
model,
|
|
sessionId: session.id,
|
|
createAdmissionMs: createdAt - requestStartedAt,
|
|
firstModelEventMs: firstModel.observedAt - requestStartedAt,
|
|
runtimeReadyMs: runtimeReady.observedAt - requestStartedAt,
|
|
terminalMs: terminal.observedAt - requestStartedAt,
|
|
modelBeforeRuntimeReady:
|
|
firstModel.event.sequence < runtimeReady.event.sequence,
|
|
toolSideEffect: true,
|
|
idempotentReplay: true,
|
|
crossTenantStatus: crossTenant.status,
|
|
durableEventCount: Number(durable[0]),
|
|
},
|
|
null,
|
|
2,
|
|
);
|
|
}
|
|
} catch (error) {
|
|
failure = error;
|
|
console.error(error);
|
|
if (dumpPort !== undefined) {
|
|
try {
|
|
console.error(
|
|
`\n--- Durable state ---\nturns:\n${runMysql(dumpPort, 'SELECT turn_id, status, error_code, is_submission_attempted, harness_event_epoch, dispatch_owner, dispatch_lease_until FROM qwen_managed_agent.ma_turn')}\nevents:\n${runMysql(dumpPort, 'SELECT stream_kind, HEX(stream_id), sequence_id, turn_id, event_type, is_terminal, source_key FROM qwen_managed_agent.ma_public_event ORDER BY stream_kind, stream_id, sequence_id')}\nsessions:\n${runMysql(dumpPort, 'SELECT session_id, harness_boot_id, harness_event_epoch, harness_last_event_id FROM qwen_managed_agent.ma_session')}\nexecutions:\n${runMysql(dumpPort, 'SELECT execution_call_id, execution_state, dispatch_generation FROM qwen_managed_agent.ma_tool_execution')}\njournal:\n${runMysql(dumpPort, 'SELECT session_id, state, writer_generation, journal_revision, committed_sequence FROM qwen_managed_agent.ma_session_head')}`,
|
|
);
|
|
} catch (dumpError) {
|
|
console.error(`Durable state dump failed: ${String(dumpError)}`);
|
|
}
|
|
}
|
|
if (replacementBrokerProxy !== undefined) {
|
|
console.error(
|
|
`\n--- Replacement Runtime Broker requests ---\n${replacementBrokerProxy.observations().join('\n')}`,
|
|
);
|
|
}
|
|
for (const child of children) {
|
|
console.error(`\n--- ${child.name} tail ---\n${child.log()}`);
|
|
}
|
|
if (existsSync(mysqlError)) {
|
|
console.error(
|
|
`\n--- MySQL error tail ---\n${readFileSync(mysqlError, 'utf8').slice(-16_384)}`,
|
|
);
|
|
}
|
|
} finally {
|
|
// Abort any pending B2 SSE capture so an early arm failure exits promptly
|
|
// instead of waiting out the capture's budget on an open socket.
|
|
for (const observer of b2SseObservers) observer.controller.abort();
|
|
// A frozen Harness holds SIGTERM pending from stopChild; wake it before
|
|
// teardown so teardown does not burn the 10-second stall on every path,
|
|
// failure or success. `harness` is try-block scoped, so reach it through
|
|
// the children registry by its start() name — and fail loudly rather
|
|
// than leaving a wedged writer behind when the name drifts.
|
|
if (freeze) {
|
|
let woke = false;
|
|
for (const child of children) {
|
|
if (child.name === hostedHarnessLabel) {
|
|
signalProcessTree(child.child, 'SIGCONT');
|
|
woke = true;
|
|
}
|
|
}
|
|
if (!woke) {
|
|
failure ??= new Error(
|
|
'Frozen-owner arm could not find the Harness child to wake: the registry key drifted',
|
|
);
|
|
}
|
|
}
|
|
for (const child of frozenChildren.splice(0)) {
|
|
signalProcessTree(child.child, 'SIGCONT');
|
|
}
|
|
for (const child of children.reverse()) {
|
|
await stopChild(child.child);
|
|
}
|
|
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();
|
|
// An interrupt during teardown is a non-pass: the keep switch must still
|
|
// hold, or a stopped run deletes its own evidence before the post-finally
|
|
// throw reports it.
|
|
if (
|
|
(failure || receivedSignal) &&
|
|
process.env['QWEN_MANAGED_E2E_KEEP_TMP'] === '1'
|
|
) {
|
|
console.error(`Keeping temporary directory: ${temporary}`);
|
|
} else {
|
|
rmSync(temporary, { recursive: true, force: true });
|
|
}
|
|
process.removeListener('SIGINT', handleSignal);
|
|
process.removeListener('SIGTERM', handleSignal);
|
|
}
|
|
|
|
if (failure) throw failure;
|
|
// An interrupt after the last poll — or during the finally's teardown —
|
|
// must not print the success JSON and exit 0: a supervising script would
|
|
// record a clean pass for a run the operator explicitly stopped. The
|
|
// payload stays a variable until both guards have run, so the decision
|
|
// precedes the record.
|
|
if (receivedSignal) throw new Error(`Interrupted by ${receivedSignal}`);
|
|
if (approvalResultJson !== undefined) console.log(approvalResultJson);
|
|
if (g3SummaryJson !== undefined) console.log(g3SummaryJson);
|
|
if (resultJson !== undefined) {
|
|
console.log(resultJson);
|
|
if (fencedWriterJson !== undefined) console.log(fencedWriterJson);
|
|
}
|