diff --git a/.github/workflows/sdk-java.yml b/.github/workflows/sdk-java.yml index 095bf19e91..b7a6ee9889 100644 --- a/.github/workflows/sdk-java.yml +++ b/.github/workflows/sdk-java.yml @@ -738,13 +738,14 @@ jobs: hosted-harness-mysql: name: 'Hosted process fault gates / MySQL 8.4 / Java 21' runs-on: 'ubuntu-latest' - # The step ceilings (35 + 9×10 + 20 = 145 — the Verify and O4 steps - # sit above their failsafe fork timeouts; the FG7 channel classes - # raised the hosted fork 900→1200 s) plus ~13 minutes for the + # The step ceilings (35 + 9×10 + 20 + 50 = 195 — the Verify, admission + # and O4 steps sit above their failsafe fork timeouts (the FG7 channel + # classes raised the hosted fork 900→1200 s; the admission gates share + # a 2400 s fork for both witness classes)) plus ~13 minutes for the # uncapped setup steps (Checkout..Install MySQL binaries, the report - # checks) need ~158; 160 keeps a thin margin, and each arm buys its + # checks) need ~208; 210 keeps a thin margin, and each arm buys its # own 10 instead of sharing an envelope. - timeout-minutes: 160 + timeout-minutes: 210 services: mysql: image: 'mysql:8.4.6' @@ -875,6 +876,27 @@ jobs: packages/sdk-java/managed-agent-server/target/fg7-cases/ retention-days: 7 + - name: 'Verify Hosted admission gates' + # Its clean would otherwise discard the hosted lane's failsafe + # reports and the fg7-cases evidence the hosts upload above, so it + # runs after the fixture. The hosted-harness-gates profile carries a + # 2400 s failsafe fork budget for both admission witness classes + # (@Timeout(1800) each); the same killed-after-timeout pattern as the + # hosted-harness-mysql verify applies. + timeout-minutes: 50 + env: + MAVEN_ARGS: '--settings ${{ runner.temp }}/setup-java-m2/settings.xml --toolchains ${{ runner.temp }}/setup-java-m2/toolchains.xml' + MYSQL_PORT: "${{ job.services.mysql.ports['3306'] }}" + run: >- + mvn --batch-mode --no-transfer-progress -f packages/sdk-java/managed-agent-server/pom.xml + -Phosted-harness-gates + -Dnode.executable="$(command -v node)" + -Dqwen.cli.entry="${GITHUB_WORKSPACE}/dist/cli.js" + -Dmysql.url="jdbc:mysql://127.0.0.1:${MYSQL_PORT}/hosted_harness_test?allowPublicKeyRetrieval=true&useSSL=false" + -Dmysql.user=root -Dmysql.password=hosted-fixture clean verify checkstyle:check + - name: 'Check that every Hosted admission gate class ran' + run: 'node scripts/check-failsafe-reports.js hosted-gates packages/sdk-java/managed-agent-server' + - name: 'Verify O4 filesystem process and capacity gates' id: 'o4_mysql' if: "${{ !cancelled() && steps.managed_dependencies.outcome == 'success' }}" diff --git a/integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts b/integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts new file mode 100644 index 0000000000..e12fbc57d6 --- /dev/null +++ b/integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts @@ -0,0 +1,1260 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +// Issue #13533 finding A2 reproduction driver: replay the shape the A1 +// chain left against a real managed agent server + MariaDB. Session α's +// admitted background Shell start reaches the server; the driver's relay +// then injects the reserve 400 once — that reserve never reaches the +// server — while cancel, close_not_started and the recovery-blocked turn +// all run production code. Then start session β's first turn on the SAME +// hosted harness daemon and record whether it wedges after the +// `hostedModelAttempt` journal commit with the model never called. The Java +// side (HostedRecoveryWedgeGate) owns server/DB evidence and clears +// session α's store rows; this driver then verifies β's turn completes. + +import { createHash, randomUUID } from 'node:crypto'; +import { + appendFile, + mkdir, + mkdtemp, + readFile, + rm, + writeFile, + access, +} from 'node:fs/promises'; +import { createServer, type IncomingMessage } from 'node:http'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { spawn, type ChildProcess } from 'node:child_process'; +import { fakeToolCall, startFakeOpenAIServer } from '../fake-openai-server.js'; +import { LISTENING_LINE_RE, stopDaemon } from './daemon-process.js'; +import { HOSTED_HOME_PREFIX } from '../scratch-dir.js'; +import { relayUpstream } from './hosted-relay-headers.js'; +import { waitUntil } from './hosted-harness-process.js'; + +const config = JSON.parse(await readFile(process.argv[2], 'utf8')) as { + tenantId: string; + storeUrl: string; + brokerUrl: string; + cliEntry: string; + arm?: 'wedge' | 'publication'; + driveFault: boolean; + control: boolean; + bPreloaded?: boolean; + bCaptureBytes?: number; + bToolProfile?: string; + holdAfterBlockMs?: number; + injectReserveDenyForSessionId?: string; + phaseFile: string; + clearedFile: string; + wireFile: string; + evidenceDir: string; + reportsDir: string; + sessions: Array<{ + sessionId: string; + workspaceId: string; + directory: string; + toolProfile: string; + role?: 'background' | 'monitor' | 'foreground' | 'edge' | 'probe'; + shellCommand?: string; + }>; +}; + +const HOSTED_TOKEN = 'hosted-wedge-fixture-token'; +const HOSTED_DIGEST = `sha256:${'a'.repeat(64)}`; +await mkdir(config.reportsDir, { recursive: true }); +await mkdir(config.evidenceDir, { recursive: true }); +await writeFile(config.wireFile, ''); +const aMarker = `A_TURN_${randomUUID()}`; +const bMarker = `B_TURN_${randomUUID()}`; +const cMarker = `C_TURN_${randomUUID()}`; +const aSession = config.sessions[0]; +const bSession = config.sessions[1]; +const cSession = config.sessions[2]; +const direct = config.control === true; + +async function wire(entry: Record) { + await appendFile( + config.wireFile, + JSON.stringify({ t: Date.now(), ...entry }) + '\n', + ); +} + +async function phase(name: string) { + await writeFile(config.phaseFile, `${name}\n`); + console.log(`PHASE=${name}`); +} + +class WedgeDaemon { + root = ''; + baseUrl = ''; + bootId = ''; + output = ''; + child?: ChildProcess; + + async start(modelUrl: string): Promise { + this.root = await mkdtemp(path.join(tmpdir(), HOSTED_HOME_PREFIX)); + const configDir = path.join(this.root, '.qwen'); + await mkdir(configDir, { recursive: true }); + await writeFile( + path.join(configDir, 'settings.json'), + JSON.stringify({ + security: { auth: { selectedType: 'openai' } }, + model: { name: 'hosted-wedge' }, + telemetry: { enabled: false }, + modelProviders: { + openai: [ + { + id: 'hosted-wedge', + envKey: 'OPENAI_API_KEY', + baseUrl: modelUrl, + }, + ], + }, + }), + ); + const env: NodeJS.ProcessEnv = {}; + for (const [key, value] of Object.entries(process.env)) { + if ( + /^(PATH|SYSTEMROOT|WINDIR|COMSPEC|PATHEXT|TMP|TEMP|TMPDIR|HOME|USERPROFILE)$/i.test( + key, + ) + ) + env[key] = value; + } + this.child = spawn( + process.execPath, + [ + '--report-on-signal', + '--report-signal', + 'SIGUSR2', + '--report-directory', + config.reportsDir, + config.cliEntry, + 'serve', + '--profile', + 'hosted-harness', + '--http-bridge', + '--no-web', + '--hostname', + '127.0.0.1', + '--port', + '0', + '--token', + HOSTED_TOKEN, + '--hosted-harness-capability-digest', + HOSTED_DIGEST, + '--workspace', + this.root, + '--managed-runtime-broker-url', + config.brokerUrl, + '--managed-runtime-broker-token', + 'hosted-tools-broker-token', + ], + { + cwd: this.root, + stdio: ['ignore', 'pipe', 'pipe'], + env: { + ...env, + HOME: this.root, + USERPROFILE: this.root, + QWEN_HOME: configDir, + QWEN_CODE_MODELS_DEV_REFRESH: 'off', + QWEN_CODE_SYSTEM_SETTINGS_PATH: path.join( + this.root, + 'system-settings.json', + ), + QWEN_CODE_SYSTEM_DEFAULTS_PATH: path.join( + this.root, + 'system-defaults.json', + ), + QWEN_RUNTIME_DIR: path.join(this.root, 'runtime'), + OPENAI_API_KEY: 'local-wedge-key', + OPENAI_BASE_URL: modelUrl, + OPENAI_MODEL: 'hosted-wedge', + QWEN_MODEL: 'hosted-wedge', + QWEN_SANDBOX: 'false', + NO_COLOR: '1', + }, + }, + ); + const append = (data: Buffer) => { + this.output = (this.output + data.toString()).slice(-32_768); + }; + this.child.stdout!.on('data', append); + this.child.stderr!.on('data', append); + await waitUntil(() => { + if (this.child!.exitCode !== null || this.child!.signalCode !== null) + throw new Error(`Wedge harness daemon exited: ${this.output}`); + const port = this.output.match(LISTENING_LINE_RE)?.groups?.['port']; + if (port) this.baseUrl = `http://127.0.0.1:${port}`; + return !!port; + }, 60_000); + await waitUntil(async () => { + const capabilities = await this.request('/capabilities'); + if (capabilities.status === 503) return false; + const text = await capabilities.text(); + if (!capabilities.ok) + throw new Error(`Capabilities ${capabilities.status}: ${text}`); + this.bootId = ( + JSON.parse(text) as { hostedHarness: { bootId: string } } + ).hostedHarness.bootId; + return true; + }, 60_000); + return this; + } + + headers(clientId?: string): Record { + return { + Authorization: `Bearer ${HOSTED_TOKEN}`, + 'X-Qwen-Harness-Protocol-Version': '1', + 'X-Qwen-Harness-Boot-Id': this.bootId, + ...(clientId ? { 'X-Qwen-Client-Id': clientId } : {}), + }; + } + + async json( + route: string, + clientId: string, + body?: unknown, + expected = 200, + method = body === undefined ? 'GET' : 'POST', + ) { + const response = await fetch(this.baseUrl + route, { + method, + headers: { + ...this.headers(clientId), + 'Content-Type': 'application/json', + }, + ...(body === undefined ? {} : { body: JSON.stringify(body) }), + signal: AbortSignal.timeout(30_000), + }); + const text = await response.text(); + if (response.status !== expected) + throw new Error( + `${method} ${route} -> ${response.status}: ${text}\ndaemon: ${this.output}`, + ); + return text ? JSON.parse(text) : undefined; + } + + request(route: string, init: RequestInit = {}) { + return fetch(this.baseUrl + route, { + ...init, + headers: init.headers ?? this.headers(), + signal: init.signal ?? AbortSignal.timeout(30_000), + }); + } + + keepRoot = false; + + async close() { + if (this.child) await stopDaemon(this.child); + if (!this.keepRoot) + await rm(this.root, { + recursive: true, + force: true, + maxRetries: 5, + retryDelay: 100, + }); + } +} + +async function bytes(req: IncomingMessage) { + const chunks: Buffer[] = []; + for await (const chunk of req) chunks.push(Buffer.from(chunk)); + return Buffer.concat(chunks); +} + +// Transparent store relay that records the session-store / publication wire +// so the wedge can be localized by operation name. All request headers are +// forwarded wholesale; hop-by-hop response headers follow relayUpstream. +// In-flight store requests get a requestId with a begin/end pair so a server +// side that never answers still leaves a trace (a plain end-only log cannot +// see the commit that hung mid-flight). +let storeRequestCounter = 0; +let reserveDenyArmed = config.injectReserveDenyForSessionId !== undefined; +const reserveBodies = new Map(); +const reserveHeaders = new Map>(); +const storeProxy = createServer(async (req, res) => { + const at = Date.now(); + const requestId = ++storeRequestCounter; + try { + const payload = await bytes(req); + let operation: string | undefined; + try { + const parsed = payload.length ? JSON.parse(payload.toString()) : {}; + if (req.url?.includes('/transactions:commit')) + operation = `commit:${parsed.operation}`; + else if (req.url?.includes('/grants')) + operation = `grants:${parsed.operation}`; + else if (req.url?.includes(':publish')) + operation = `publish:${parsed.kind}`; + else operation = undefined; + } catch { + operation = 'unparsed'; + } + await wire({ + kind: 'store-begin', + requestId, + at, + method: req.method, + url: req.url, + operation, + requestSnippet: payload.toString().slice(0, 600), + }); + if (operation === 'grants:reserve') { + const sidMatch = req.url?.match(/\/sessions\/([^/]+)\/grants/); + if (sidMatch) { + reserveBodies.set(sidMatch[1], payload.toString()); + reserveHeaders.set(sidMatch[1], { + 'content-type': 'application/json', + 'x-qwen-tenant-id': String(req.headers['x-qwen-tenant-id']), + 'x-qwen-managed-writer-token': String( + req.headers['x-qwen-managed-writer-token'], + ), + 'x-qwen-managed-writer-id': String( + req.headers['x-qwen-managed-writer-id'], + ), + 'x-qwen-managed-writer-generation': String( + req.headers['x-qwen-managed-writer-generation'], + ), + ...(req.headers['x-qwen-tool-publication-token'] + ? { + 'x-qwen-tool-publication-token': String( + req.headers['x-qwen-tool-publication-token'], + ), + } + : {}), + }); + } + } + // A2 wedge-arm substitute (post-A1-fix world): the A1 400-chain no longer + // exists at the contract, so α's failed-start leftovers are reproduced by + // injecting the same 400 invalid_request wire onto the admitted family's + // first reserve — the turn then runs its real cancel/cleanup path and + // lands the same recovery-blocked shape the A1 chain produced. + if ( + reserveDenyArmed && + operation === 'grants:reserve' && + req.url?.includes(`/sessions/${config.injectReserveDenyForSessionId}/`) + ) { + reserveDenyArmed = false; + const body = JSON.stringify({ + error: { + code: 'invalid_request', + message: 'The request body is invalid.', + request_id: randomUUID(), + }, + }); + await wire({ + kind: 'store-end', + requestId, + at, + method: req.method, + url: req.url, + operation, + status: 400, + injected: true, + responseSnippet: body, + }); + res.writeHead(400, { + 'Content-Type': 'application/json', + 'Cache-Control': 'no-store', + }); + res.end(body); + return; + } + const headers = { ...req.headers } as Record; + delete headers['host']; + delete headers['content-length']; + delete headers['connection']; + const response = await fetch(new URL(req.url!, config.storeUrl), { + method: req.method, + headers, + ...(payload.length ? { body: payload } : {}), + // A wedged server must not stall the relay indefinitely: 120 s is far + // past every healthy response here and past the wedge threshold. + signal: AbortSignal.timeout(120_000), + }); + const output = Buffer.from(await response.arrayBuffer()); + await wire({ + kind: 'store-end', + requestId, + at, + method: req.method, + url: req.url, + operation, + status: response.status, + responseSnippet: output.toString().slice(0, 600), + }); + relayUpstream(res, response, output); + } catch (cause) { + await wire({ + kind: 'store-end', + requestId, + at, + method: req.method, + url: req.url, + status: 598, + responseSnippet: String(cause), + }); + res.writeHead(598); + res.end(String(cause)); + } +}); +await new Promise((resolve) => + storeProxy.listen(0, '127.0.0.1', resolve), +); +const proxyAddress = storeProxy.address(); +if (!proxyAddress || typeof proxyAddress === 'string') + throw new Error('store proxy did not bind'); +const storeProxyUrl = `http://127.0.0.1:${proxyAddress.port}`; + +let modelCallsA = 0; +let modelCallsB = 0; +let modelBAnsweredAt = 0; +const faultArmed = config.driveFault === true; +// Per-session role replies for the publication arm; wedge arms keep α's +// admitted background start plus plain-text probes. +const extraMarkers = [`D_TURN_${randomUUID()}`, `E_TURN_${randomUUID()}`]; +const allMarkers = [aMarker, bMarker, cMarker, ...extraMarkers]; +const sessionByMarker = new Map( + config.sessions.map((session, index) => [allMarkers[index], session]), +); +const issued = new Set(); +const model = await startFakeOpenAIServer(({ body }) => { + const text = JSON.stringify(body); + const tools = + (body['tools'] as Array<{ function: { name: string } }> | undefined) ?? []; + const messages = + (body['messages'] as + | Array<{ role: string; content: unknown }> + | undefined) ?? []; + const entry = { + kind: 'model', + tools: tools.map((tool) => tool.function.name), + roles: messages.map((message) => message.role), + }; + if (text.includes(aMarker)) modelCallsA += 1; + for (const [marker, session] of sessionByMarker) { + if (!text.includes(marker)) continue; + const role = + session.role ?? (session === aSession ? 'background' : 'probe'); + // α owns the admitted-background reply only when a fault arm or the + // publication arm asks for it; the control arm keeps α a plain-text + // session so the healthy baseline stays healthy. + const wantsFaultShell = + role === 'background' && (faultArmed || config.arm === 'publication'); + if (wantsFaultShell) { + if (!issued.has(session.sessionId)) { + issued.add(session.sessionId); + void wire({ + ...entry, + replied: 'background-shell', + session: session.sessionId, + }); + return { + toolCalls: [ + fakeToolCall( + 'run_shell_command', + { + command: session.shellCommand ?? 'sleep 25', + timeout: 60_000, + description: 'admitted background Shell start', + is_background: true, + }, + `bg-${session.sessionId.slice(0, 8)}`, + ), + ], + }; + } + const toolContent = messages + .filter((message) => message.role === 'tool') + .map((message) => JSON.stringify(message.content).slice(0, 500)); + void wire({ ...entry, replied: 'background-second', toolContent }); + return { content: 'BACKGROUND_SECOND_CALL' }; + } + if (role === 'monitor') { + if (!issued.has(session.sessionId)) { + issued.add(session.sessionId); + void wire({ + ...entry, + replied: 'monitor-start', + session: session.sessionId, + }); + return { + toolCalls: [ + fakeToolCall( + 'monitor', + { + command: session.shellCommand ?? 'printf MON_OK\\n; sleep 1', + idle_timeout_ms: 2_000, + max_events: 5, + description: 'admitted Monitor start', + }, + `mon-${session.sessionId.slice(0, 8)}`, + ), + ], + }; + } + const toolContent = messages + .filter((message) => message.role === 'tool') + .map((message) => JSON.stringify(message.content).slice(0, 500)); + void wire({ ...entry, replied: 'monitor-second', toolContent }); + return { content: 'MONITOR_SECOND_CALL' }; + } + if (role === 'foreground') { + if (!issued.has(session.sessionId)) { + issued.add(session.sessionId); + void wire({ + ...entry, + replied: 'foreground-shell', + session: session.sessionId, + }); + return { + toolCalls: [ + fakeToolCall( + 'run_shell_command', + { + command: session.shellCommand ?? 'printf FG_OK\\n', + timeout: 60_000, + }, + `fg-${session.sessionId.slice(0, 8)}`, + ), + ], + }; + } + const toolContent = messages + .filter((message) => message.role === 'tool') + .map((message) => JSON.stringify(message.content).slice(0, 500)); + void wire({ ...entry, replied: 'foreground-second', toolContent }); + return { content: 'FOREGROUND_COMPLETE' }; + } + modelCallsB += 1; + modelBAnsweredAt = Date.now(); + void wire({ ...entry, replied: 'probe-text' }); + return { content: 'PROBE_TURN_COMPLETE' }; + } + void wire({ ...entry, replied: 'other-text' }); + return { content: 'UNEXPECTED' }; +}); + +const daemon = await new WedgeDaemon().start(model.baseUrl); + +interface WireRow { + kind?: string; + requestId?: number; + at?: number; + method?: string; + url?: string; + operation?: string; + status?: number; + requestSnippet?: string; + responseSnippet?: string; + replied?: string; +} + +async function readWire(): Promise { + try { + return (await readFile(config.wireFile, 'utf8')) + .split('\n') + .filter(Boolean) + .map((line) => JSON.parse(line) as WireRow); + } catch (cause) { + if ((cause as NodeJS.ErrnoException)?.code === 'ENOENT') return []; + throw cause; + } +} + +async function evidenceFile(name: string, content: string) { + await writeFile(path.join(config.evidenceDir, name), content); +} + +try { + const connectionFor = (workspaceId: string) => ({ + baseUrl: storeProxyUrl, + tenantId: config.tenantId, + workspaceId, + writerId: daemon.bootId, + leaseDurationMs: 60_000, + }); + + // --- Session α: drive the A1 chain into recovery-blocked ---------------- + // The A1 reserve path lives in the v3 publication lane: a Shell-profile + // Session only owns it when created with captureBytes. The publication arm + // creates every session through the generic path below instead, so this + // legacy attach runs only for wedge arms (a second POST /session for the + // same id is exactly what the daemon's already_attached guard exists for). + const createdA = + config.arm === 'publication' + ? { clientId: '' } + : ((await daemon.json('/session', '', { + sessionId: aSession.sessionId, + sessionScope: 'thread', + managedSessionStore: connectionFor(aSession.workspaceId), + toolProfile: aSession.toolProfile, + captureBytes: 64 * 1024 * 1024, + })) as { clientId: string }); + const aClient = createdA.clientId; + const promptA = async () => { + const aBlocks = [{ type: 'text', text: `Session alpha prompt ${aMarker}` }]; + await daemon.json( + `/session/${aSession.sessionId}/prompt`, + aClient, + { + promptId: randomUUID(), + prompt: aBlocks, + payloadDigest: `sha256:${createHash('sha256').update(JSON.stringify(aBlocks)).digest('hex')}`, + }, + 202, + ); + }; + const driveAFault = async () => { + await promptA(); + await waitUntil(async () => { + const status = await daemon.json( + `/session/${aSession.sessionId}/status`, + aClient, + ); + return status.recoveryBlocked === true; + }, 180_000); + const status = await daemon.json( + `/session/${aSession.sessionId}/status`, + aClient, + ); + console.log(`A_BLOCKED_STATUS=${JSON.stringify(status)}`); + const rows = await readWire(); + // The server answers contract refusals with a deliberately generic + // 400 invalid_request envelope; the exception text stays server-side. + const reserveDenied = rows.find( + (row) => + row.kind === 'store-end' && + row.operation === 'grants:reserve' && + row.status === 400 && + typeof row.responseSnippet === 'string' && + row.responseSnippet.includes('invalid_request'), + ); + const closeDenied = rows.find( + (row) => + row.kind === 'store-end' && + row.operation === 'grants:close_not_started' && + (row.status ?? 0) >= 400 && + (row.status ?? 0) < 500, + ); + await evidenceFile( + 'a1-chain.txt', + [ + `reserveDenied=${JSON.stringify(reserveDenied, null, 2)}`, + `closeDenied=${JSON.stringify(closeDenied, null, 2)}`, + `modelCallsA=${modelCallsA}`, + ].join('\n'), + ); + if (!reserveDenied || !closeDenied) + throw new Error( + `A1 chain did not materialize on the wire: reserve=${JSON.stringify( + reserveDenied, + )} close=${JSON.stringify(closeDenied)}`, + ); + await phase('A_BLOCKED'); + if (config.holdAfterBlockMs) { + console.log(`HOLDING_BLOCKED_STATE_MS=${config.holdAfterBlockMs}`); + await new Promise((resolve) => + setTimeout(resolve, config.holdAfterBlockMs), + ); + const heldRows = await readWire(); + const renewals = heldRows.filter( + (row) => + row.kind === 'store-end' && + row.status === 200 && + (row.operation?.includes('renew') ?? false), + ); + console.log(`A_RENEWALS_WHILE_BLOCKED=${renewals.length}`); + await evidenceFile( + 'a-renewals.txt', + renewals.map((row) => JSON.stringify(row)).join('\n'), + ); + } + }; + const driveAControlTurn = async () => { + await promptA(); + await waitUntil( + async () => + !(await daemon.json(`/session/${aSession.sessionId}/status`, aClient)) + .hasActivePrompt, + 120_000, + ); + const status = await daemon.json( + `/session/${aSession.sessionId}/status`, + aClient, + ); + console.log(`A_CONTROL_STATUS=${JSON.stringify(status)}`); + await phase('A_COMPLETED'); + }; + + // --- Probe turns for the non-blocked sessions on the same daemon --------- + // A probe session's creation moment is part of the matrix (preloaded vs + // fresh), so it is created where each arm decides, never earlier. + interface ProbeTurn { + outcome: 'completed' | 'wedged'; + promptId: string; + promptAt: number; + attemptAt: number; + attemptHangingAt: number; + hangingCommit: unknown; + commitMessageAt: number; + sessionId: string; + clientId: string; + } + const createSession = async ( + entry: (typeof config.sessions)[number], + toolProfile: string, + captureBytes: number, + ): Promise => { + const body: Record = { + sessionId: entry.sessionId, + sessionScope: 'thread', + managedSessionStore: connectionFor(entry.workspaceId), + toolProfile, + ...(captureBytes ? { captureBytes } : {}), + }; + const response = await fetch(daemon.baseUrl + '/session', { + method: 'POST', + headers: { ...daemon.headers(), 'Content-Type': 'application/json' }, + body: JSON.stringify(body), + signal: AbortSignal.timeout(30_000), + }); + const text = await response.text(); + if (response.status !== 200) { + await evidenceFile( + `create-session-${entry.sessionId.slice(0, 8)}.txt`, + `body=${JSON.stringify(body, null, 2)}\nstatus=${response.status}\nresponse=${text}`, + ); + throw new Error( + `create session ${entry.sessionId} -> ${response.status}: ${text}`, + ); + } + return (JSON.parse(text) as { clientId: string }).clientId; + }; + async function driveProbeTurn( + entry: (typeof config.sessions)[number], + clientId: string, + marker: string, + ): Promise { + const turn: ProbeTurn = { + outcome: 'completed', + promptId: randomUUID(), + promptAt: Date.now(), + attemptAt: 0, + attemptHangingAt: 0, + hangingCommit: undefined, + commitMessageAt: 0, + sessionId: entry.sessionId, + clientId, + }; + const bBlocks = [{ type: 'text', text: `Session probe prompt ${marker}` }]; + const modelBefore = modelBAnsweredAt; + await daemon.json( + `/session/${entry.sessionId}/prompt`, + clientId, + { + promptId: turn.promptId, + prompt: bBlocks, + payloadDigest: `sha256:${createHash('sha256').update(JSON.stringify(bBlocks)).digest('hex')}`, + }, + 202, + ); + const deadline = turn.promptAt + 120_000; + while (Date.now() < deadline) { + const rows = await readWire(); + const session = (row: WireRow) => + (row.url ?? '').includes(`/sessions/${entry.sessionId}/`) && + (row.at ?? 0) >= turn.promptAt - 500; + const attempt = rows.find( + (row) => + row.kind === 'store-end' && + row.operation === 'commit:hostedModelAttempt' && + session(row) && + row.status === 200, + ); + const commitMessage = rows.find( + (row) => + row.kind === 'store-end' && + row.operation === 'commit:commitMessage' && + session(row) && + row.status === 200, + ); + const committed = new Set( + rows + .filter((row) => row.kind === 'store-end') + .map((row) => row.requestId), + ); + const hanging = rows.find( + (row) => + row.kind === 'store-begin' && + session(row) && + !committed.has(row.requestId ?? -1) && + Date.now() - (row.at ?? 0) > 15_000, + ); + if (attempt && !turn.attemptAt) { + turn.attemptAt = attempt.at ?? 0; + console.log( + `PROBE_ATTEMPT_COMMITTED_AT=${turn.attemptAt} session=${entry.sessionId}`, + ); + } + if (hanging && !turn.attemptHangingAt) { + turn.attemptHangingAt = hanging.at ?? 0; + turn.hangingCommit = hanging; + console.log(`PROBE_HANGING_STORE_REQUEST_AT=${turn.attemptHangingAt}`); + } + if (commitMessage && !turn.commitMessageAt) + turn.commitMessageAt = commitMessage.at ?? 0; + if (modelBAnsweredAt > modelBefore) { + console.log( + `PROBE_MODEL_CALLED_AFTER=${modelBAnsweredAt - turn.promptAt}ms attempt=${turn.attemptAt}`, + ); + break; + } + if ( + (turn.attemptAt && Date.now() - turn.attemptAt > 30_000) || + (turn.attemptHangingAt && Date.now() - turn.attemptHangingAt > 45_000) + ) { + const active = ( + await daemon.json(`/session/${entry.sessionId}/status`, clientId) + ).hasActivePrompt; + if (active) { + turn.outcome = 'wedged'; + return turn; + } + // The turn ended (completed or failed) before the threshold — fall + // through to the transcript gate instead of inventing a wedge that a + // dead turn cannot have. + break; + } + await new Promise((resolve) => setTimeout(resolve, 250)); + } + if ( + turn.attemptAt === 0 && + turn.attemptHangingAt === 0 && + modelBAnsweredAt <= modelBefore + ) + throw new Error( + `Probe session never reached hostedModelAttempt within 120 s; daemon: ${daemon.output}`, + ); + await waitUntil( + async () => + !(await daemon.json(`/session/${entry.sessionId}/status`, clientId)) + .hasActivePrompt, + 120_000, + ); + const transcript = await daemon.json( + `/session/${entry.sessionId}/transcript?cursor=0&limit=256`, + clientId, + ); + const events = ( + transcript as { events: Array<{ type: string; promptId?: string }> } + ).events; + const complete = events.filter( + (event) => + event.promptId === turn.promptId && event.type === 'turn_complete', + ); + if (complete.length !== 1) + throw new Error( + `Probe turn ${turn.sessionId} ended without exactly one turn_complete for its promptId — it failed instead of completing, so it can never be journaled as not-wedged: ${JSON.stringify(events.slice(-40))}`, + ); + console.log( + `PROBE_TURN_COMPLETED session=${entry.sessionId} attemptAt=+${turn.attemptAt - turn.promptAt}ms commitMessageAt=+${turn.commitMessageAt - turn.promptAt}ms modelAt=+${modelBAnsweredAt - turn.promptAt}ms`, + ); + return turn; + } + + async function wedgeEvidence(marker: string, turn: ProbeTurn) { + console.log( + `PROBE_WEDGED attemptAt=${turn.attemptAt} hangingAt=${turn.attemptHangingAt} now=${Date.now()} modelCallsOther=${modelCallsB}`, + ); + process.kill(daemon.child!.pid!, 'SIGUSR2'); + await new Promise((resolve) => setTimeout(resolve, 2_000)); + const statusProbe = await daemon.json( + `/session/${turn.sessionId}/status`, + turn.clientId, + ); + const statusA = await daemon.json( + `/session/${aSession.sessionId}/status`, + aClient, + ); + const rows = await readWire(); + await evidenceFile( + 'probe-wedged.txt', + [ + `marker=${marker} session=${turn.sessionId} attemptAt=${turn.attemptAt} attemptHangingAt=${turn.attemptHangingAt} modelCallsA=${modelCallsA} modelCallsOther=${modelCallsB}`, + `hangingCommit=${JSON.stringify(turn.hangingCommit, null, 2)}`, + `statusA=${JSON.stringify(statusA)}`, + `statusProbe=${JSON.stringify(statusProbe)}`, + `daemonOutputTail=${daemon.output.slice(-2000)}`, + ].join('\n'), + ); + await evidenceFile( + 'wire-through-wedge.jsonl', + rows.map((row) => JSON.stringify(row)).join('\n'), + ); + await phase('B_WEDGED_CONFIRMED'); + + // The Java side clears session α's store rows now. + await waitUntil( + () => + access(config.clearedFile).then( + () => true, + () => false, + ), + 300_000, + ); + const clearedAt = Date.now(); + console.log(`CLEARED_OBSERVED_AT=${clearedAt}`); + await waitUntil( + () => Promise.resolve(modelBAnsweredAt > clearedAt), + 240_000, + ); + console.log( + `PROBE_MODEL_CALLED_AFTER_CLEAR=${modelBAnsweredAt - clearedAt}ms`, + ); + await waitUntil( + async () => + !(await daemon.json(`/session/${turn.sessionId}/status`, turn.clientId)) + .hasActivePrompt, + 240_000, + ); + const transcript = await daemon.json( + `/session/${turn.sessionId}/transcript?cursor=0&limit=256`, + turn.clientId, + ); + const events = ( + transcript as { events: Array<{ type: string; promptId?: string }> } + ).events; + const complete = events.filter( + (event) => + event.promptId === turn.promptId && event.type === 'turn_complete', + ); + if (complete.length !== 1) + throw new Error( + `Probe turn did not complete after clearing: ${JSON.stringify(events)}`, + ); + await evidenceFile( + 'probe-recovered.txt', + `modelBAnsweredAt=${modelBAnsweredAt} clearedAt=${clearedAt}\ntranscript terminal=${JSON.stringify(complete)}`, + ); + await phase('B_COMPLETED_AFTER_CLEAR'); + console.log('HOSTED_RECOVERY_WEDGE_OK'); + } + + // --- background-publication acceptance arm (post-A1-fix witness) --------- + // Three shell-profile sessions (created with captureBytes, so they own the + // v3 publication lane): α background, μ monitor, σ foreground regression; + // the Arm-3 δ1/δ2 refusal edges are driver-posted forged-digest replays of + // the captured α/μ reserves, not sessions. + if (config.arm === 'publication') { + const evidence: string[] = []; + const clients = new Map(); + for (const session of config.sessions) { + clients.set( + session.sessionId, + await createSession(session, session.toolProfile, 64 * 1024 * 1024), + ); + } + const promptFor = async ( + entry: (typeof config.sessions)[number], + index: number, + ) => { + const blocks = [ + { + type: 'text', + text: `publication probe ${allMarkers[index]} ${randomUUID()}`, + }, + ]; + await daemon.json( + `/session/${entry.sessionId}/prompt`, + clients.get(entry.sessionId)!, + { + promptId: randomUUID(), + prompt: blocks, + payloadDigest: `sha256:${createHash('sha256').update(JSON.stringify(blocks)).digest('hex')}`, + }, + 202, + ); + }; + const reservesFor = (sessionId: string) => + readWire().then((rows) => + rows.filter( + (row) => + row.kind === 'store-end' && + row.operation === 'grants:reserve' && + (row.url ?? '').includes(`/sessions/${sessionId}/`), + ), + ); + const closesFor = (sessionId: string) => + readWire().then((rows) => + rows.filter( + (row) => + row.kind === 'store-end' && + row.operation === 'grants:close_not_started' && + (row.url ?? '').includes(`/sessions/${sessionId}/`), + ), + ); + const statusOf = (entry: (typeof config.sessions)[number]) => + daemon.json( + `/session/${entry.sessionId}/status`, + clients.get(entry.sessionId)!, + ); + const waitTurnEndsBlocked = async ( + entry: (typeof config.sessions)[number], + timeoutMs: number, + ) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const status = await statusOf(entry); + if (!status.hasActivePrompt) return status; + await new Promise((resolve) => setTimeout(resolve, 400)); + } + throw new Error( + `${entry.sessionId} turn still active after ${timeoutMs}ms: ${JSON.stringify(await statusOf(entry))}`, + ); + }; + // Detached-family arm state at this slice, observed off-Linux (ephemeral + // lane): the admitted family reserves (OPEN), dispatches, the worker + // declines the start for lack of a cgroup v2 root (not_started), the + // grant closes NOT_STARTED, and the recoverable sibling records commit + // before the turn blocks. Pin exactly that; do not wait for a file the + // never-started process cannot write. + const detachedArmOf = async ( + entry: (typeof config.sessions)[number], + index: number, + label: string, + forge?: { family: string; argsDigest: string; requestDigest: string }, + ) => { + await promptFor(entry, index); + await waitUntil(async () => { + const reserves = await reservesFor(entry.sessionId); + return reserves.some( + (row) => + row.status === 200 && + (row.responseSnippet ?? '').includes('"OPEN"'), + ); + }, 120_000); + if (forge) { + // Arm 3 (refusal-edges-live) fires here: the turn's own reserve is + // captured OPEN and not yet settled, so replaying the very same + // publicationId with only its digests forged is refused at the + // placement check — the store compares the replayed binding digest + // against the persisted row before identity or payload validation. + const source = reserveBodies.get(entry.sessionId); + const headers = reserveHeaders.get(entry.sessionId); + if (!source || !headers) + throw new Error(`No captured reserve for ${forge.family} forging`); + const body = JSON.parse(source); + body.binding.reference.argsDigest = forge.argsDigest; + body.binding.requestDigest = forge.requestDigest; + const response = await fetch( + `${storeProxyUrl}/internal/managed-tool-publications/v1/sessions/${entry.sessionId}/grants?workspaceId=${entry.workspaceId}`, + { + method: 'POST', + headers, + body: JSON.stringify(body), + signal: AbortSignal.timeout(30_000), + }, + ); + const text = await response.text(); + evidence.push( + `edge ${forge.family} forged-digest replay on captured reserve -> ${response.status} ${text.slice(0, 240)}`, + ); + if ( + response.status !== 400 || + !/reservation replay conflicts/i.test(text) + ) + throw new Error( + `${forge.family} forged-digest replay must refuse with 400 naming the reservation replay conflict, got ${response.status}: ${text.slice(0, 240)}`, + ); + } + const status = await waitTurnEndsBlocked(entry, 120_000); + const reserves = await reservesFor(entry.sessionId); + const closes = await closesFor(entry.sessionId); + const mutations = (await readWire()).filter( + (row) => + row.operation === 'commit:commitChildRunRecord' || + row.operation === 'commit:commitMonitorRecord', + ); + const open = reserves.find( + (row) => + row.status === 200 && (row.responseSnippet ?? '').includes('"OPEN"'), + ); + const closure = closes.find((row) => row.status === 200); + if (!open) + throw new Error( + `${label} reserve did not answer OPEN: ${JSON.stringify(reserves)}`, + ); + if (!closure) + throw new Error( + `${label} grants close did not answer 200: ${JSON.stringify(closes)}`, + ); + if (!status.hasActivePrompt && status.recoveryBlocked !== true) + throw new Error( + `${label} turn ended without the recovery-blocked state the not_started path reports at this slice: ${JSON.stringify(status)}`, + ); + const stderr = daemon.output.match(/is recovery blocked: [^\n]*/g); + evidence.push( + `${label} reserve OPEN + close ${closure.status} + turn recovery-blocked=${status.recoveryBlocked} + childRecords=${mutations.length}`, + `${label} blockCause=${JSON.stringify(stderr?.slice(-1) ?? [])}`, + `${label} reserves=${JSON.stringify(reserves.map((row) => ({ status: row.status, snippet: (row.responseSnippet ?? '').slice(0, 160) })))}`, + `${label} closes=${JSON.stringify(closes.map((row) => ({ status: row.status, snippet: (row.responseSnippet ?? '').slice(0, 160) })))}`, + ); + return { status, reserves, closes }; + }; + + // Arm 1 — background-reserve: admitted; off-Linux the ephemeral lane has + // no cgroup root, so the start is declined not_started and the turn goes + // recovery-blocked. Full detached-handle settlement after turn + // completion stays the unlanded B1 arm — this arm pins OPEN + close + + // the blocked state and names that dependency. + const alpha = config.sessions[0]; + await detachedArmOf(alpha, 0, 'background', { + family: 'shell', + argsDigest: `sha256:${'f'.repeat(64)}`, + requestDigest: `sha256:${'e'.repeat(64)}`, + }); + evidence.push( + 'background settlement target after turn completion is the unlanded B1 arm; intermediate state pinned (OPEN + close NOT_STARTED + turn recovery-blocked, blockCause recorded above)', + ); + + // Arm 2 — monitor-reserve: admitted, same detached-family chain as the + // background arm, pinned strict like its siblings. A refused admitted + // Monitor reserve is a regression of the admission layers this witness + // exists to pin, never a passing outcome the witness may journal green. + const mu = config.sessions[1]; + await detachedArmOf(mu, 1, 'monitor', { + family: 'monitor', + argsDigest: `sha256:${'d'.repeat(64)}`, + requestDigest: `sha256:${'c'.repeat(64)}`, + }); + + // Arm 3 — refusal-edges-live: folded into the two detached arms above. + // Pure shape-level negatives can't be emitted by the turn (it + // normalizes them away) and the digest-bound protocol refuses forged + // bytes at identification, so each family replays its captured OPEN + // reserve with the same publicationId and only its + // requestDigest/argsDigest forged, fired before the blocked settlement; + // the 400 refusal reason must name the reservation replay conflict — + // the placement check stops a mismatched binding digest before + // identity or payload validation (unit fixtures cover the field-shape + // refusals). + + // Arm 4 — foreground reserve on the same v3 lane: off-Linux the worker + // declines it just like the detached families (no cgroup root, no + // capture publisher), so the turn closes NOT_STARTED and blocks; the + // byte-level foreground regression is HostedWorkspaceToolTurnIT (v2 + // lane, G5), which runs the same harness against the foreground chain + // unchanged. + const sigma = config.sessions[2]; + await detachedArmOf(sigma, 2, 'foreground-v3'); + evidence.push( + 'foreground reserve OPEN + close NOT_STARTED + turn recovery-blocked (ephemeral-lane decline, all shell-typed tools); end-to-end foreground regression held by HostedWorkspaceToolTurnIT (G5)', + ); + + await evidenceFile('publication-arms.txt', evidence.join('\n')); + await phase('PUBLICATION_DONE'); + console.log('HOSTED_BACKGROUND_PUBLICATION_OK'); + console.log(evidence.join('\n')); + } else if (direct) { + await driveAControlTurn(); + const bClient = await createSession( + bSession, + config.bToolProfile || bSession.toolProfile, + config.bCaptureBytes ?? 0, + ); + const turn = await driveProbeTurn(bSession, bClient, bMarker); + if (turn.outcome !== 'completed') + throw new Error('Control session β did not complete.'); + await phase('CONTROL_B_COMPLETED'); + console.log('HOSTED_WEDGE_CONTROL_OK'); + } else if (config.bPreloaded) { + // Overlap variant: β already lives on the daemon with a healthy first + // turn when α goes recovery-blocked; its next turn is the A2 probe. + console.log('B_PRELOADED_HEALTHY_TURN_FIRST'); + const bClient = await createSession( + bSession, + config.bToolProfile || bSession.toolProfile, + config.bCaptureBytes ?? 0, + ); + const before = await driveProbeTurn(bSession, bClient, bMarker); + if (before.outcome !== 'completed') + throw new Error('Preloaded session β first turn did not complete.'); + await driveAFault(); + const marker2 = `${bMarker}_SECOND`; + const turn = await driveProbeTurn(bSession, bClient, marker2); + if (turn.outcome === 'completed') { + await evidenceFile( + 'b-not-wedged.txt', + `preloaded B second turn completed: attemptAt=${turn.attemptAt} modelBAnsweredAt=${modelBAnsweredAt} commitMessageAt=${turn.commitMessageAt}`, + ); + await phase('B_NOT_WEDGED'); + console.log('HOSTED_WEDGE_ABSENT'); + } else { + await wedgeEvidence(marker2, turn); + } + } else { + await driveAFault(); + const bClient = await createSession( + bSession, + config.bToolProfile || bSession.toolProfile, + config.bCaptureBytes ?? 0, + ); + const turnB = await driveProbeTurn(bSession, bClient, bMarker); + if (turnB.outcome === 'wedged') { + await wedgeEvidence(bMarker, turnB); + } else { + // The rig reproduced twice in fresh sessions; probe a second fresh + // session on the same daemon as well. + const cClient = await createSession( + cSession, + config.bToolProfile || cSession.toolProfile, + config.bCaptureBytes ?? 0, + ); + const turnC = await driveProbeTurn(cSession, cClient, cMarker); + if (turnC.outcome === 'wedged') { + await wedgeEvidence(cMarker, turnC); + } else { + await evidenceFile( + 'b-not-wedged.txt', + [ + `fresh B first turn completed: attemptAt=+${turnB.attemptAt - turnB.promptAt}ms commitMessageAt=+${turnB.commitMessageAt - turnB.promptAt}ms modelBAnsweredAt=${modelBAnsweredAt}`, + `fresh C first turn completed: attemptAt=+${turnC.attemptAt - turnC.promptAt}ms commitMessageAt=+${turnC.commitMessageAt - turnC.promptAt}ms`, + ].join('\n'), + ); + await phase('B_NOT_WEDGED'); + console.log('HOSTED_WEDGE_ABSENT'); + } + } + } +} catch (cause) { + console.error('WEDGE_DRIVER_FAILURE', cause); + console.error('daemon output tail:', daemon.output.slice(-4000)); + daemon.keepRoot = true; + console.error(`daemon root kept for inspection: ${daemon.root}`); + try { + await evidenceFile( + 'driver-failure.txt', + `${String(cause)}\n${daemon.output.slice(-4000)}\ndaemonRoot=${daemon.root}`, + ); + } catch { + // best effort + } + try { + await phase('DRIVER_FAILURE'); + } catch { + // best effort + } + process.exitCode = 1; +} finally { + await daemon.close().catch(() => undefined); + storeProxy.close(); + await model.close(); +} diff --git a/integration-tests/helpers/hosted-relay-headers.test.ts b/integration-tests/helpers/hosted-relay-headers.test.ts index a61459852f..ef949cd3d5 100644 --- a/integration-tests/helpers/hosted-relay-headers.test.ts +++ b/integration-tests/helpers/hosted-relay-headers.test.ts @@ -123,7 +123,9 @@ describe('Hosted proxy header relay', () => { // the drivers relative to this file, never process.cwd(). Read source // text because importing a driver throws — each parses process.argv[2] // at module load. Positive pin only: a new driver hand-rolling its relay - // is absent from the enumerated list and still passes. + // is absent from the enumerated list and still passes. A new legitimate + // caller of the shared helper joins this list deliberately — the list + // itself is what this pin keeps countable. const dir = dirname(fileURLToPath(import.meta.url)); const callers = readdirSync(dir) .filter((name) => /^hosted-.*-driver\.ts$/.test(name)) @@ -135,9 +137,31 @@ describe('Hosted proxy header relay', () => { 'hosted-channel-fault-driver.ts', 'hosted-latency-driver.ts', 'hosted-process-crash-driver.ts', + 'hosted-recovery-blocked-wedge-driver.ts', 'hosted-shell-output-driver.ts', 'hosted-store-failure-driver.ts', 'hosted-workspace-tool-turn-driver.ts', ]); }); + + it('pins the probe driver to certify completed turns by their transcript', () => { + // R3-18 sentinel: classifying a probe turn from wall clock alone lets a + // failed turn journal green beside a wedged one's evidence; the + // transcript gate is the only honest completed-path verdict — removing + // it must turn this pin red. + const source = readFileSync( + join( + dirname(fileURLToPath(import.meta.url)), + 'hosted-recovery-blocked-wedge-driver.ts', + ), + 'utf8', + ); + const completedArm = source.slice( + source.indexOf('async function driveProbeTurn('), + source.indexOf('async function wedgeEvidence('), + ); + expect(completedArm).toContain('/transcript?cursor=0&limit=256'); + expect(completedArm).toContain("event.type === 'turn_complete'"); + expect(completedArm).toContain('event.promptId === turn.promptId'); + }); }); diff --git a/packages/core/src/managed-runtime/contracts/managed-tool-publication-v1.fixtures.json b/packages/core/src/managed-runtime/contracts/managed-tool-publication-v1.fixtures.json index 5b0902d24e..c9dd73b8c4 100644 --- a/packages/core/src/managed-runtime/contracts/managed-tool-publication-v1.fixtures.json +++ b/packages/core/src/managed-runtime/contracts/managed-tool-publication-v1.fixtures.json @@ -3,6 +3,183 @@ "payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"printf hi\",\"description\":\"capture\"}}", "canonicalInputJson": "{\"command\":\"printf hi\",\"description\":\"capture\"}", "argsJson": "{\"harnessSessionId\":\"session-1\",\"runtimeSessionId\":\"runtime-session-1\",\"payloadJson\":\"{\\\"toolName\\\":\\\"run_shell_command\\\",\\\"input\\\":{\\\"command\\\":\\\"printf hi\\\",\\\"description\\\":\\\"capture\\\"}}\"}", + "payloadVectors": [ + { + "id": "background-shell", + "valid": true, + "payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"sleep 300\",\"description\":\"background capture\",\"is_background\":true}}", + "requestDigest": "sha256:c81849f53abb0083aa249bc60c1c574b7e5de6f129ef15eb132e8aa6c9f29ef9", + "argsDigest": "sha256:2fb0a8ddc0c746a7db94c14203c34be2559405bbe79259742d6a46db938b221e" + }, + { + "id": "background-shell-timeout", + "valid": true, + "payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"sleep 300\",\"timeout\":600000,\"is_background\":true}}", + "requestDigest": "sha256:d1df5e4c8d8a3e9f326603c4c784b4d74330fcd508b0a33d79e517c58b47729e", + "argsDigest": "sha256:232a22d3e08815783a0b7cd667a15b02dbdf2487527c0aac422c024463b821cf" + }, + { + "id": "monitor", + "valid": true, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"description\":\"watch the log\",\"idle_timeout_ms\":30000,\"is_monitor\":true,\"max_events\":500}}", + "requestDigest": "sha256:eab83c00bb3a1f4571292cfdf6570c9f4a966d635c6435c77bbde92defe8c974", + "argsDigest": "sha256:7027c8316a2731ecebf5f9dd4579437df411cd39646b9efd3d24b8f75aee5797" + }, + { + "id": "monitor-minimal", + "valid": true, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true}}", + "requestDigest": "sha256:5405e318a24240a32903fa31360fec5f0f603d09179c91e1a35b976839d19e62", + "argsDigest": "sha256:911c9314170715ce5cc1e042f9d4322a2efca2e6737057444ad584ef2ac6c0b9" + }, + { + "id": "background-string-flag", + "valid": false, + "payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"sleep 300\",\"is_background\":\"true\"}}", + "requestDigest": "sha256:b5eaa0c0983387c8d4c9dd15f03844f0960c22a04b2dfe56d53eab97b90dbb3b", + "argsDigest": "sha256:e280e95a318be2942314f9189cb77a71a9e843b57387f21926361ef2a555306a" + }, + { + "id": "background-explicit-false", + "valid": true, + "payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"printf hi\",\"is_background\":false}}", + "requestDigest": "sha256:55096daad2f54b6784c7213c20902325d0ebe4b38002519655eb16e6b6178ff1", + "argsDigest": "sha256:979825f64b90d9df3850567dabc993724aa8734e9b12aed127e6bf05b86bb4f0" + }, + { + "id": "shell-unknown-key", + "valid": false, + "payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"printf hi\",\"is_background\":true,\"shell\":\"bash\"}}", + "requestDigest": "sha256:44ae05b9679d50370f3cac2eded3ab169b0024d84a37f26036310709a6c9fae9", + "argsDigest": "sha256:3e755585b8580a9210cbe6c6309aa5d759062bead8e33a8739851e280d05df87" + }, + { + "id": "background-bad-timeout", + "valid": false, + "payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"sleep 300\",\"is_background\":true,\"timeout\":600001}}", + "requestDigest": "sha256:2e361a607f8b004daeb1401a0db7fe6053379ad335048a2a284203f47a1d2b2d", + "argsDigest": "sha256:fe59bea464d33e1d18e55464a0107793e1c57f18d6b9f696af3d677bbbf9b7ea" + }, + { + "id": "monitor-no-marker", + "valid": true, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\"}}", + "requestDigest": "sha256:fb9e80ff565fce70d0b8f6cdc615192714c96f3c71ecfea2adf4bdd1c7a7dfdd", + "argsDigest": "sha256:858bba3697d3750ca35b16a3619c3279195ac60ebfc2f4b63fb2fbc5c06c12d1" + }, + { + "id": "monitor-string-marker", + "valid": false, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":\"true\"}}", + "requestDigest": "sha256:3dc630efb3b77dc056c54edf13a39f9fa7aac784774ed490febd3b24b0eb437a", + "argsDigest": "sha256:c3f0f49053360bebe684a127e65c0a3cdda1d2d96fc89a08a407a777bb7d0d70" + }, + { + "id": "monitor-unknown-key", + "valid": false, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"interval\":5}}", + "requestDigest": "sha256:c45dc7b5fa93c49a0f26dc13fa3f7e4d7107864979f125d7e5dccf0a70c0cf89", + "argsDigest": "sha256:f6d15d253892df32f6cebc3c454138bf884eaa2e2934a67c18ebd30c55fc78b3" + }, + { + "id": "monitor-bad-idle-timeout", + "valid": false, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"idle_timeout_ms\":600001}}", + "requestDigest": "sha256:ade6675ff565f468c51cfcd971c842e51983cdcaf9d0431cbf24f54b6cafde83", + "argsDigest": "sha256:a4d93e5076db20289f004061081a0f4ef4044c0641bbfb7f286f2fd42a1d160a" + }, + { + "id": "monitor-bad-max-events", + "valid": false, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"max_events\":10001}}", + "requestDigest": "sha256:dffecc5653b8a5adcfa95387dab1c1d8592c73ad49e7115f87a004d1295db44e", + "argsDigest": "sha256:0d02a5cd90187df5923d89b2b701c02f2e2e8b2098cb97de3fdfcc1d7c21db61" + }, + { + "id": "monitor-empty-command", + "valid": false, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"\",\"is_monitor\":true}}", + "requestDigest": "sha256:b1757c13f27fa16ade541fe813017af3dcad2dbd9007a97c55ae174f926c1db5", + "argsDigest": "sha256:543da28d6df0402acc09285b856f99c014b2051b3b38cd07162276552bad7868" + }, + { + "id": "unknown-tool", + "valid": false, + "payloadJson": "{\"toolName\":\"write_file\",\"input\":{\"content\":\"x\"}}", + "requestDigest": "sha256:96b7c954f6c5fe4b2fd2a7a2677366f2ca8c8e36cae21d8a76c637be0851f66d", + "argsDigest": "sha256:ee2b252d1cd491425942090e06507c7337b5279df43af31ab718b1b1b5da8708" + }, + { + "id": "monitor-idle-exponent", + "valid": true, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"idle_timeout_ms\":6e5,\"is_monitor\":true}}", + "requestDigest": "sha256:aa3d6fabd8daec37b4d3eaa2b2e60720bc4655016c3657d0b7d41660d3b2b56b", + "argsDigest": "sha256:3858b8f13deb1dfafb7a208efce5e43303c5e1a80e3eced5a41f0a71a9e01dce" + }, + { + "id": "shell-timeout-decimal", + "valid": true, + "payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"printf hi\",\"timeout\":600000.0}}", + "requestDigest": "sha256:c256a5b770823a8091e1cf7f76854d405971ca42bf8d90f599f3f40dca436e39", + "argsDigest": "sha256:23aade92a5dd8942a163d01a85ad2af48d04c1fc3b27e6e2aeab4839a6c3d8a1" + }, + { + "id": "monitor-max-events-exponent", + "valid": true, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"max_events\":1e4}}", + "requestDigest": "sha256:5bf2c66a92e675c71ef7b70bb730a8a1d24de7a46cc9eab7585117e619d166ed", + "argsDigest": "sha256:033f3e2eecb32e7937ab202290fb19d236de3642c94b9b160d4d18f535f5808f" + }, + { + "id": "monitor-idle-min-boundary", + "valid": true, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"idle_timeout_ms\":1,\"is_monitor\":true}}", + "requestDigest": "sha256:3b42f774023e574631b3f358122d6221b750e2b0c5795d0ad4eca2134df8add1", + "argsDigest": "sha256:fa1fa67d24670c6fdd987ef9b3c13dd02642edcf38485f1c57b56b485fa0ed09" + }, + { + "id": "monitor-idle-boundary", + "valid": true, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"idle_timeout_ms\":600000,\"is_monitor\":true}}", + "requestDigest": "sha256:3cffe643746f9ad88d68240d5ae4936cf9c7b2a8bbc933ae2a505e29b3e27fe3", + "argsDigest": "sha256:3858b8f13deb1dfafb7a208efce5e43303c5e1a80e3eced5a41f0a71a9e01dce" + }, + { + "id": "monitor-max-events-min-boundary", + "valid": true, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"max_events\":1}}", + "requestDigest": "sha256:7ecef2491369c33dbb963c8973bc1b7b6489ec8e4b20fc90a7bb1bcdea13b45a", + "argsDigest": "sha256:709ade433a39a2588d40a43e573492ca611e327616648bb2c2e0b2ebf4ce3c82" + }, + { + "id": "monitor-max-events-boundary", + "valid": true, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"max_events\":10000}}", + "requestDigest": "sha256:ea6855bf1318a3b80dbe18c16111b07971a67d664675c23b9adcc01aa5a94ca5", + "argsDigest": "sha256:033f3e2eecb32e7937ab202290fb19d236de3642c94b9b160d4d18f535f5808f" + }, + { + "id": "shell-timeout-fractional", + "valid": false, + "payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"printf hi\",\"timeout\":1.5}}", + "requestDigest": "sha256:cb62bf9066976e383e02a4acd6ed95c39ebc0f9a3826ae7297ce919e2e28ffe0", + "argsDigest": "sha256:ffe0129018abbd27827abf67631660619361e68d19a73d79dbfb517bacfb2d09" + }, + { + "id": "monitor-idle-timeout-fractional", + "valid": false, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"idle_timeout_ms\":1.5,\"is_monitor\":true}}", + "requestDigest": "sha256:671f6b007d5c24a526d6510336b6c2e1d8fdeabb396aa37f76a882d74d4ecfe6", + "argsDigest": "sha256:54e1edb1cb6af26ec6f8b0a290db70321871bae616ca5a5e20ae59eba678c513" + }, + { + "id": "monitor-max-events-fractional", + "valid": false, + "payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"max_events\":1.5}}", + "requestDigest": "sha256:ffdb53168a65e0e0d5835317178e5804f11cb4bb218767743fe8d18a23247de2", + "argsDigest": "sha256:86af41b3a6163755b9c0ddf30077ec33fd82a2375eacd87b909dcddd3c9d7a88" + } + ], "bindingDigest": "0e9e44f23047cf18a9347b86e3fcd05e2d58748530cac6b044fc390341f78fe8", "cases": [ { diff --git a/packages/core/src/managed-runtime/managed-tool-publication.test.ts b/packages/core/src/managed-runtime/managed-tool-publication.test.ts index 42a498ec19..93a88d9fed 100644 --- a/packages/core/src/managed-runtime/managed-tool-publication.test.ts +++ b/packages/core/src/managed-runtime/managed-tool-publication.test.ts @@ -36,6 +36,13 @@ const fixtures = JSON.parse( requestDigest: string; argsDigest: string; }>; + payloadVectors: Array<{ + id: string; + valid: boolean; + payloadJson: string; + requestDigest: string; + argsDigest: string; + }>; digestVectors: Array<{ id: string; binding: unknown; digest: string }>; tokenVector: { token: string; hash: string }; bindingDigest: string; @@ -88,6 +95,23 @@ describe('managed-tool-publication/1', () => { ).toBe(fixtures.tokenVector.hash); }); + describe('publication payload families', () => { + const binding = parseToolPublicationBinding(fixtures.cases[0].value); + for (const vector of fixtures.payloadVectors) { + it(`${vector.id}`, () => { + const bound = { + ...binding, + requestDigest: vector.requestDigest, + reference: { ...binding.reference, argsDigest: vector.argsDigest }, + }; + const check = () => + assertToolPublicationPayload(bound, vector.payloadJson); + if (vector.valid) expect(check).not.toThrow(); + else expect(check).toThrow(); + }); + } + }); + it('keeps original Runtime identity and both digests separate from model pairing', () => { const binding = parseToolPublicationBinding(fixtures.cases[0].value); expect(binding.modelCallId).not.toBe(binding.reference.callId); diff --git a/packages/sdk-java/managed-agent-server/pom.xml b/packages/sdk-java/managed-agent-server/pom.xml index 116d557bbb..a60998c576 100644 --- a/packages/sdk-java/managed-agent-server/pom.xml +++ b/packages/sdk-java/managed-agent-server/pom.xml @@ -258,6 +258,23 @@ + + hosted-harness-gates + + org.apache.maven.plugins + maven-failsafe-plugin + ${maven-failsafe-plugin.version} + + **/Hosted*Gate.java + true + + true + 2400 + + integration-testverify + + hosted-process-crashes diff --git a/packages/sdk-java/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/store/ToolPublicationContract.java b/packages/sdk-java/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/store/ToolPublicationContract.java index 924f66b9b8..d2abbda4f2 100644 --- a/packages/sdk-java/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/store/ToolPublicationContract.java +++ b/packages/sdk-java/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/store/ToolPublicationContract.java @@ -25,6 +25,12 @@ import java.util.Set; public final class ToolPublicationContract { public static final String PROTOCOL = "managed-tool-publication/1"; + /** The tool families a capture publication may bind to. Every reserve-path + * gate must take the family list from here — a second hard-coded copy is + * how the Monitor reserve came to be refused one layer below the payload + * contract (#13533, A1). */ + public static final Set PUBLISHABLE_TOOL_NAMES = Set.of( + "run_shell_command", "monitor"); public static final int MAX_BODY_BYTES = 64 * 1024; public static final long MAX_CAPTURE_BYTES = 1L << 41; public static final long PRODUCER_BYTES = 2_686_976; @@ -155,13 +161,18 @@ public final class ToolPublicationContract { String toolName = text(body, "toolName"); // H3: the detached capture family publishes background Shell and // Monitor payloads beside the foreground Shell; each family keeps its - // own closed input shape. + // own closed input shape. Admission takes the family list from one + // place — the store's checkpoint gate reads the same constant, and a + // second hard-coded copy is what refused the Monitor reserve one layer + // below this contract (#13533). + require(PUBLISHABLE_TOOL_NAMES.contains(toolName), + "Only Shell and Monitor payloads can publish"); if ("run_shell_command".equals(toolName)) { requireShellInput(input); } else if ("monitor".equals(toolName)) { requireMonitorInput(input); } else { - require(false, "Only Shell and Monitor payloads can publish"); + throw new IllegalArgumentException("Publishable tool lacks a contract arm"); } require(("sha256:" + sha256(bytes)).equals(text(binding, "requestDigest")), "Original payload digest conflicts"); @@ -176,10 +187,7 @@ public final class ToolPublicationContract { input.fieldNames().forEachRemaining(name -> require( Set.of("command", "timeout", "description", "is_background", "directory") .contains(name), "Shell input field is invalid")); - require(!input.has("timeout") || input.get("timeout").isIntegralNumber() - && input.get("timeout").canConvertToInt() - && input.get("timeout").intValue() >= 1 && input.get("timeout").intValue() <= 600_000, - "Shell timeout is invalid"); + boundedIntField(input, "timeout", 600_000, "Shell timeout is invalid"); require(!input.has("description") || input.get("description").isTextual(), "Shell description is invalid"); require(!input.has("is_background") || input.get("is_background").isBoolean(), @@ -199,19 +207,31 @@ public final class ToolPublicationContract { "Monitor description is invalid"); require(!input.has("is_monitor") || input.get("is_monitor").isBoolean(), "Monitor marker is invalid"); - require(!input.has("max_events") || input.get("max_events").isIntegralNumber() - && input.get("max_events").canConvertToInt() - && input.get("max_events").intValue() >= 1 - && input.get("max_events").intValue() <= 10_000, "Monitor max_events is invalid"); - require(!input.has("idle_timeout_ms") || input.get("idle_timeout_ms").isIntegralNumber() - && input.get("idle_timeout_ms").canConvertToInt() - && input.get("idle_timeout_ms").intValue() >= 1 - && input.get("idle_timeout_ms").intValue() <= 600_000, - "Monitor idle timeout is invalid"); + boundedIntField(input, "max_events", 10_000, "Monitor max_events is invalid"); + boundedIntField(input, "idle_timeout_ms", 600_000, "Monitor idle timeout is invalid"); require(!input.has("directory") || input.get("directory").isTextual() && !input.get("directory").textValue().isEmpty(), "Monitor directory is invalid"); } + /** + * An optional integral-valued numeric field bounded below by 1. Integral + * doubles (6e5, 600000.0) are the same integer the TypeScript mirror + * accepts after JSON.parse normalization — the wire spelling is pinned + * again by the request digest, so accepting the integral form here keeps + * the two implementations' accepted families identical. + */ + private static void boundedIntField(JsonNode input, String key, long maximum, String reason) { + JsonNode value = input.get(key); + if (value == null) { + return; + } + BigDecimal number = value.isNumber() && Double.isFinite(value.doubleValue()) + ? value.decimalValue() : null; + require(number != null && number.stripTrailingZeros().scale() <= 0 + && number.compareTo(BigDecimal.ONE) >= 0 + && number.compareTo(BigDecimal.valueOf(maximum)) <= 0, reason); + } + public static String bindingDigest(JsonNode binding) { return sha256(canonicalText(canonical(parse("binding", binding))) .getBytes(StandardCharsets.UTF_8)); diff --git a/packages/sdk-java/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/store/ToolPublicationStore.java b/packages/sdk-java/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/store/ToolPublicationStore.java index 8f4dffa0dd..0b08751322 100644 --- a/packages/sdk-java/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/store/ToolPublicationStore.java +++ b/packages/sdk-java/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/store/ToolPublicationStore.java @@ -754,8 +754,8 @@ public final class ToolPublicationStore { item = value; } } - require(item != null && ("run_shell_command".equals(text(item, "toolName")) - || "monitor".equals(text(item, "toolName"))) + require(item != null + && ToolPublicationContract.PUBLISHABLE_TOOL_NAMES.contains(text(item, "toolName")) && "in_progress".equals(text(item, "state")) && "runtime".equals(text(item, "outcomeSource")) && text(b, "modelCallId").equals(text(item, "functionCallId")) diff --git a/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedBackgroundPublicationGate.java b/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedBackgroundPublicationGate.java new file mode 100644 index 0000000000..6946ba4e6b --- /dev/null +++ b/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedBackgroundPublicationGate.java @@ -0,0 +1,266 @@ +package com.alibaba.qwen.code.managedagent; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.alibaba.qwen.code.managedagent.api.WorkspaceSelection; +import com.alibaba.qwen.code.managedagent.service.EmbeddedRuntimeBroker; +import com.alibaba.qwen.code.managedagent.store.ManagedAgentStore; +import com.alibaba.qwen.code.runtimebroker.WorkspaceExecutionProfile; +import com.alibaba.qwen.managedprobes.ProbeAdvice; +import com.alibaba.qwen.managedprobes.PublicationProbeWiring; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.attribute.FileTime; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext; +import org.springframework.jdbc.core.JdbcTemplate; + +/** + * End-to-end witness for #13533 finding A1: the background Shell and + * Monitor payload families were admitted into the publication contract + * earlier, with #13789; this PR extracts the publishable family list into + * {@code ToolPublicationContract.PUBLISHABLE_TOOL_NAMES} and widens the + * integral-valued numerics via {@code boundedIntField}, pinned in both + * languages by the shared {@code payloadVectors} corpus. Admitted + * background {@code run_shell_command} and {@code monitor} calls now + * reach a reserve instead of the removed 400. + * + *

Arms (production binaries in-process: MariaDB + embedded broker + real + * hosted-harness daemon running the bundled CLI with the H3 domains enabled + * upstream (#13856); fake model; no cgroup v2 on this host, so the + * ephemeral lane declines every shell-typed start {@code not_started}): + * + *

    + *
  • background-reserve: admitted background start reserves and the grant + * answers OPEN (200) — the pre-fix 400 no longer exists. The start is then + * declined {@code not_started} on the ephemeral lane (no cgroup v2 root), + * the grant closes NOT_STARTED (200), and the turn goes recovery-blocked; + * full detached-handle settlement after turn completion remains the unlanded + * B1 arm — pinned and named. + *
  • monitor-reserve: on the merged contract arms, this PR extracted the + * publishable list into + * {@code ToolPublicationContract.PUBLISHABLE_TOOL_NAMES} and widened the + * integral-valued numerics with {@code boundedIntField}; the admitted + * Monitor reserve answers OPEN like the background arm, then + * declines {@code not_started} on the ephemeral lane (no cgroup v2 root), + * closes NOT_STARTED (200), and the turn blocks. The store-side + * checkpoint gate that earlier refused its toolName now reads the same + * constant extracted here. + *
  • refusal-edges-live: per-family raw negatives replayed against the + * captured OPEN grant with only its digests forged — shape-level negatives + * can't be emitted by the turn and the digest-bound protocol admits no + * spoofing; each replay answers 400 whose refusal names the reservation + * replay conflict, since the placement check compares the replayed + * binding digest against the persisted row before identity or payload + * validation; the unit fixture corpus pins the field-shape refusals. + *
  • foreground (v3 lane): reserve OPEN + close NOT_STARTED + + * recovery-blocked — off-Linux the ephemeral lane declines all shell-typed + * executions alike; byte-level foreground regression is held by + * {@code HostedWorkspaceToolTurnIT}'s main workspace/shell lane (G5), run in + * the same verification pass. + *
+ * + *

Collector-less: run through the {@code hosted-harness-gates} profile + * ({@code mvn -Phosted-harness-gates -Dmysql.url=... -Dnode.executable=...}), + * which includes the {@code Hosted*Gate} test classes and sets the probe + * flag, so the hosted-harness-mysql CI lane and both roster checks stay + * untouched. The explicit source list registers {@link ProbeAdvice} so the + * refusal-diagnosis advice surfaces the contract/store refusal reasons in + * the 400 envelope and the test log (production's handler is deliberately + * generic; the wedge witness observes it as-is). + */ +class HostedBackgroundPublicationGate { + + @TempDir + private Path temporary; + + @Test + @Timeout(1800) + void backgroundAndMonitorReservesAnswerOpen() throws Exception { + org.junit.jupiter.api.Assumptions.assumeTrue( + "true".equals(System.getProperty("qwen.wedge.probe")), + "background publication witness: skipped unless -Dqwen.wedge.probe=true"); + assertThat(System.getProperty("mysql.url")).as("witness requires -Dmysql.url") + .startsWith("jdbc:mysql:"); + assertThat(System.getProperty("mysql.user")).as("witness requires -Dmysql.user").isNotBlank(); + String node = System.getProperty("node.executable"); + assertThat(node).as("Pass -Dnode.executable with an absolute Node.js 22+ path").isNotBlank(); + Path cli = Path.of(System.getProperty("qwen.cli.entry", "../../../dist/cli.js")) + .toAbsolutePath().normalize(); + assertThat(cli).as("CLI entry -Dqwen.cli.entry=%s", cli).isRegularFile(); + Path moduleDir = Path.of(System.getProperty("user.dir")).toAbsolutePath(); + Path repoRoot = moduleDir.resolve("../../..").normalize(); + assertThat(repoRoot.resolve("integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts")) + .isRegularFile(); + Path evidence = (System.getProperty("qwen.wedge.evidence") != null + ? Path.of(System.getProperty("qwen.wedge.evidence")).toAbsolutePath() + : moduleDir.resolve("target/wedge-evidence")).resolve("publication"); + Path moduleTarget = moduleDir.resolve("target").normalize(); + assertThat(evidence.normalize().startsWith(moduleTarget)) + .as("-Dqwen.wedge.evidence must resolve under %s before the recursive wipe", moduleTarget) + .isTrue(); + Files.createDirectories(evidence); + // Runs share this fixed evidence directory; a stale wedged run's + // artifacts must not survive into a later green run's bundle, so the + // subtree this invocation owns is wiped before any write. + try (var stale = Files.walk(evidence)) { + stale.sorted(java.util.Comparator.reverseOrder()).forEach(entry -> { + try { + Files.deleteIfExists(entry); + } catch (java.io.IOException ignored) { + } + }); + } + Files.createDirectories(evidence); + Path reportsDir = evidence.resolve("node-reports"); + Files.createDirectories(reportsDir); + temporary = temporary.toRealPath(); + Files.createDirectory(temporary.resolve("runtime")); + + String tenant = "publication-" + UUID.randomUUID(); + record SessionSpec(String name, String role, String shellCommand) {} + List specs = List.of( + new SessionSpec("alpha", "background", + "mkdir -p probe && echo BG_OK > probe/bg.txt"), + new SessionSpec("mu", "monitor", "printf MON_OK\\n; sleep 1"), + new SessionSpec("sigma", "foreground", "echo FG_OK > fg-proof.txt")); + List workspaces = new ArrayList<>(); + for (SessionSpec spec : specs) { + Path dir = Files.createDirectory(temporary.resolve(spec.name())); + Files.createDirectory(dir.resolve("child")); + Files.setLastModifiedTime(dir, FileTime.fromMillis(1)); + workspaces.add(dir); + } + Path wrapper = temporary.resolve("node-report-wrapper.sh"); + for (String path : List.of(node, reportsDir.toString())) { + assertThat(path).as("wrapper paths must not contain a quote or newline") + .doesNotContain("\"").doesNotContain("\n"); + } + Files.writeString(wrapper, "#!/bin/sh\nexec \"" + node + + "\" --report-on-signal --report-signal=SIGUSR2 --report-directory=\"" + + reportsDir + "\" \"$@\"\n"); + wrapper.toFile().setExecutable(true); + + var arguments = new ArrayList<>(List.of( + "--server.address=127.0.0.1", "--server.port=0", + "--spring.datasource.url=" + System.getProperty("mysql.url"), + "--spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver", + "--spring.datasource.username=" + System.getProperty("mysql.user"), + "--spring.datasource.password=" + System.getProperty("mysql.password", ""), + "--qwen.managed-agent.session-store.enabled=true", + "--qwen.managed-agent.tool-publication.entry-concurrency=32", + "--qwen.managed-agent.harness.enabled=false", + "--qwen.managed-agent.harness.capability-digest=sha256:" + "a".repeat(64), + "--qwen.managed-agent.runtime-broker.enabled=true", + "--qwen.managed-agent.runtime-broker.port=0", + "--qwen.managed-agent.runtime-broker.token=hosted-tools-broker-token", + "--qwen.managed-agent.runtime-broker.durable-local-process=false", + "--qwen.managed-agent.runtime-broker.trusted-local-reboot-recovery=false", + "--qwen.managed-agent.runtime-broker.workspace-cwd=" + temporary, + "--qwen.managed-agent.runtime-broker.state-directory=" + temporary.resolve("runtime"), + "--qwen.managed-agent.runtime-broker.credential-key-id=test", + "--qwen.managed-agent.runtime-broker.credential-key=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=", + "--qwen.managed-agent.runtime-broker.node-executable=" + wrapper, + "--qwen.managed-agent.runtime-broker.worker-entry=" + cli, + "--qwen.managed-agent.runtime-broker.verified-workspace-recovery-enabled=false", + "--qwen.managed-agent.runtime-broker.cli-entry=" + cli)); + for (int index = 0; index < workspaces.size(); index++) { + String prefix = "--qwen.managed-agent.runtime-broker.workspace-mounts[" + index + "]."; + arguments.add(prefix + "tenant-id=" + tenant); + arguments.add(prefix + "storage-id=storage-" + index); + arguments.add(prefix + "root=" + workspaces.get(index)); + } + var application = new SpringApplicationBuilder(ManagedAgentServerApplication.class, + PublicationProbeWiring.class, ProbeAdvice.class); + try (var spring = (ServletWebServerApplicationContext) application.run(arguments.toArray(String[]::new))) { + JdbcTemplate jdbc = spring.getBean(JdbcTemplate.class); + var metadata = jdbc.queryForMap("SELECT VERSION() AS version, @@version_comment AS engine"); + System.out.println("PUBLICATION_DATABASE " + metadata); + assertThat(metadata.toString().toLowerCase()) + .as("witness requires MariaDB/MySQL, got %s", metadata) + .containsAnyOf("mysql", "mariadb"); + ManagedAgentStore store = spring.getBean(ManagedAgentStore.class); + var sessions = new ArrayList>(); + for (int index = 0; index < specs.size(); index++) { + String workspaceId = "workspace-" + index; + jdbc.update("INSERT INTO managed_workspace_registry (tenant_id, workspace_id," + + " workspace_generation, storage_id, display_name, config_ref, policy_ref," + + " state) VALUES (?, ?, 1, ?, 'Workspace', ?, ?, 'ACTIVE')", + tenant, workspaceId, "storage-" + index, + WorkspaceExecutionProfile.CONFIG_REF, WorkspaceExecutionProfile.POLICY_REF); + jdbc.update("INSERT INTO managed_workspace_access (tenant_id, workspace_id, actor_id," + + " role) VALUES (?, ?, ?, 'OPERATOR')", + tenant, workspaceId, "actor".getBytes(StandardCharsets.UTF_8)); + var created = store.insertWorkspaceSessionCommand(tenant, "actor", "create-" + index, + "sha256:" + "a".repeat(64), "qwen-code", null, null, List.of(), null, + new WorkspaceSelection(workspaceId, "child")); + var entry = new LinkedHashMap(); + entry.put("sessionId", created.sessionId()); + entry.put("workspaceId", workspaceId); + entry.put("directory", workspaces.get(index).resolve("child").toString()); + entry.put("toolProfile", "hosted-workspace-shell/1"); + entry.put("role", specs.get(index).role()); + if (specs.get(index).shellCommand() != null) + entry.put("shellCommand", specs.get(index).shellCommand()); + sessions.add(entry); + } + EmbeddedRuntimeBroker broker = spring.getBean(EmbeddedRuntimeBroker.class); + Path phaseFile = temporary.resolve("phase"); + Path config = temporary.resolve("driver.json"); + var driverConfig = new LinkedHashMap(); + driverConfig.put("tenantId", tenant); + driverConfig.put("storeUrl", "http://127.0.0.1:" + spring.getWebServer().getPort()); + driverConfig.put("brokerUrl", broker.getBaseUri().toString()); + driverConfig.put("cliEntry", cli.toString()); + driverConfig.put("arm", "publication"); + driverConfig.put("driveFault", true); + driverConfig.put("control", false); + driverConfig.put("phaseFile", phaseFile.toString()); + driverConfig.put("clearedFile", temporary.resolve("cleared").toString()); + driverConfig.put("wireFile", temporary.resolve("wire.jsonl").toString()); + driverConfig.put("evidenceDir", evidence.toString()); + driverConfig.put("reportsDir", reportsDir.toString()); + driverConfig.put("sessions", sessions); + new ObjectMapper().writeValue(config.toFile(), driverConfig); + Files.writeString(evidence.resolve("sessions.txt"), + sessions.stream().map(s -> s.get("role") + "=" + s.get("sessionId")) + .reduce(tenant, (out, line) -> out + "\n" + line) + "\n"); + Path driverLog = temporary.resolve("driver.log"); + Process driver = new ProcessBuilder(node, "--import", "tsx", + "integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts", + config.toString()) + .directory(repoRoot.toFile()) + .redirectErrorStream(true).redirectOutput(driverLog.toFile()).start(); + try { + assertThat(driver.waitFor(1500, TimeUnit.SECONDS)) + .as("Driver timeout:\n%s", Files.readString(driverLog)).isTrue(); + if (Files.exists(temporary.resolve("wire.jsonl"))) { + Files.copy(temporary.resolve("wire.jsonl"), evidence.resolve("wire.jsonl"), + java.nio.file.StandardCopyOption.REPLACE_EXISTING); + } + Files.copy(driverLog, evidence.resolve("driver-final.log"), + java.nio.file.StandardCopyOption.REPLACE_EXISTING); + String log = Files.readString(driverLog); + System.out.println(log); + assertThat(driver.exitValue()).as("Driver output:\n%s", log).isZero(); + assertThat(log).contains("HOSTED_BACKGROUND_PUBLICATION_OK"); + } finally { + driver.descendants().forEach(ProcessHandle::destroyForcibly); + if (driver.isAlive()) { + driver.destroyForcibly(); + } + } + } + } +} diff --git a/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedRecoveryWedgeGate.java b/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedRecoveryWedgeGate.java new file mode 100644 index 0000000000..919bb040c9 --- /dev/null +++ b/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedRecoveryWedgeGate.java @@ -0,0 +1,482 @@ +package com.alibaba.qwen.code.managedagent; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.alibaba.qwen.code.managedagent.api.WorkspaceSelection; +import com.alibaba.qwen.code.managedagent.service.EmbeddedRuntimeBroker; +import com.alibaba.qwen.code.managedagent.store.ManagedAgentStore; +import com.alibaba.qwen.code.runtimebroker.WorkspaceExecutionProfile; +import com.alibaba.qwen.managedprobes.PublicationProbeWiring; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.attribute.FileTime; +import java.sql.Connection; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext; +import org.springframework.jdbc.core.JdbcTemplate; + +/** + * Issue #13533 finding A2: one recovery-blocked session wedges later turns of + * other sessions on the same hosted harness daemon. + * + *

Session α is put into the same recovery-blocked shape the A1 chain left + * (admitted background start → failed reserve → cancel → + * {@code close_not_started} → recovery-blocked turn, session renewing). The + * A1 fix admitted the background Shell/Monitor families into the publication + * contract, so the original 400-chain no longer exists at the endpoint; the + * substitute used here is an injected reserve refusal on α's admitted + * background start, injected once at the driver's relay with the same 400 + * {@code invalid_request} wire the pre-fix server answered — every downstream + * step (cancel, close, recovery-block) then runs production code, and the + * leftover shape matches what A1 produced. The substitute is documented in + * the reproduction report for this issue. + * + *

Session β (and facultatively γ, or β as a preloaded session, per + * -Dqwen.wedge.preloaded) then runs its turn(s) on the same daemon. The + * regression expectation at this tree is absent of wedge: every + * probe turn completes. If a probe ever wedges again, this test captures the + * JVM thread dump (the managed-agent-server runs in-process), the MariaDB + * transaction/lock state and the session rows, deletes session α's store + * rows, asserts the probe recovers, and then fails as the A2 regression. + * + *

Collector-less: run through the {@code hosted-harness-gates} profile + * ({@code mvn -Phosted-harness-gates -Dmysql.url=... -Dnode.executable=...}), + * which includes the {@code Hosted*Gate} test classes and sets the probe + * flag. The H3 domains are enabled upstream (#13856), so no bundle patch + * or test switch is needed. Run with -Dqwen.wedge.control=true for the + * no-fault control run. + */ +class HostedRecoveryWedgeGate { + + @TempDir + private Path temporary; + + @Test + @Timeout(1800) + void recoveryBlockedSessionMustNotWedgeOtherSessionTurns() throws Exception { + org.junit.jupiter.api.Assumptions.assumeTrue( + "true".equals(System.getProperty("qwen.wedge.probe")), + "A2 reproduction probe: skipped unless -Dqwen.wedge.probe=true"); + assertThat(System.getProperty("mysql.url")).as("A2 reproduction requires -Dmysql.url") + .startsWith("jdbc:mysql:"); + assertThat(System.getProperty("mysql.user")).as("A2 reproduction requires -Dmysql.user").isNotBlank(); + String node = System.getProperty("node.executable"); + assertThat(node).as("Pass -Dnode.executable with an absolute Node.js 22+ path").isNotBlank(); + Path cli = Path.of(System.getProperty("qwen.cli.entry", "../../../dist/cli.js")) + .toAbsolutePath().normalize(); + assertThat(cli).as("CLI entry -Dqwen.cli.entry=%s", cli).isRegularFile(); + boolean control = "true".equals(System.getProperty("qwen.wedge.control")); + Path moduleDir = Path.of(System.getProperty("user.dir")).toAbsolutePath(); + Path repoRoot = moduleDir.resolve("../../..").normalize(); + assertThat(repoRoot.resolve("integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts")) + .isRegularFile(); + Path evidence = (System.getProperty("qwen.wedge.evidence") != null + ? Path.of(System.getProperty("qwen.wedge.evidence")).toAbsolutePath() + : moduleDir.resolve("target/wedge-evidence")).resolve("wedge"); + Path moduleTarget = moduleDir.resolve("target").normalize(); + assertThat(evidence.normalize().startsWith(moduleTarget)) + .as("-Dqwen.wedge.evidence must resolve under %s before the recursive wipe", moduleTarget) + .isTrue(); + Files.createDirectories(evidence); + // Runs share this fixed evidence directory; a stale wedged run's + // artifacts must not survive into a later green run's bundle, so the + // subtree this invocation owns is wiped before any write. + try (var stale = Files.walk(evidence)) { + stale.sorted(java.util.Comparator.reverseOrder()).forEach(entry -> { + try { + Files.deleteIfExists(entry); + } catch (java.io.IOException ignored) { + } + }); + } + Files.createDirectories(evidence); + Path reportsDir = evidence.resolve("node-reports"); + Files.createDirectories(reportsDir); + temporary = temporary.toRealPath(); + Files.createDirectory(temporary.resolve("runtime")); + + String tenant = "wedge-" + UUID.randomUUID(); + List workspaces = new ArrayList<>(); + List workspaceIds = List.of("workspace-a", "workspace-b", "workspace-c"); + for (String name : List.of("alpha", "beta", "gamma")) { + Path dir = Files.createDirectory(temporary.resolve(name)); + Files.createDirectory(dir.resolve("child")); + Files.setLastModifiedTime(dir, FileTime.fromMillis(1)); + workspaces.add(dir); + } + Path wrapper = temporary.resolve("node-report-wrapper.sh"); + for (String path : List.of(node, reportsDir.toString())) { + assertThat(path).as("wrapper paths must not contain a quote or newline") + .doesNotContain("\"").doesNotContain("\n"); + } + Files.writeString(wrapper, "#!/bin/sh\nexec \"" + node + + "\" --report-on-signal --report-signal=SIGUSR2 --report-directory=\"" + + reportsDir + "\" \"$@\"\n"); + wrapper.toFile().setExecutable(true); + + var arguments = new ArrayList<>(List.of( + "--server.address=127.0.0.1", "--server.port=0", + "--spring.datasource.url=" + System.getProperty("mysql.url"), + "--spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver", + "--spring.datasource.username=" + System.getProperty("mysql.user"), + "--spring.datasource.password=" + System.getProperty("mysql.password", ""), + "--qwen.managed-agent.session-store.enabled=true", + // The conditional OSS-backed publication stack cannot boot + // outside a provisioned rig, so the real controller and + // stores are registered programmatically below with a stub + // object store (A1's reserve is refused before any object + // call), with only the entry-concurrency property bound. + "--qwen.managed-agent.tool-publication.entry-concurrency=32", + "--qwen.managed-agent.harness.enabled=false", + "--qwen.managed-agent.harness.capability-digest=sha256:" + "a".repeat(64), + "--qwen.managed-agent.runtime-broker.enabled=true", + "--qwen.managed-agent.runtime-broker.port=0", + "--qwen.managed-agent.runtime-broker.token=hosted-tools-broker-token", + "--qwen.managed-agent.runtime-broker.durable-local-process=false", + "--qwen.managed-agent.runtime-broker.trusted-local-reboot-recovery=false", + "--qwen.managed-agent.runtime-broker.workspace-cwd=" + temporary, + "--qwen.managed-agent.runtime-broker.state-directory=" + temporary.resolve("runtime"), + "--qwen.managed-agent.runtime-broker.credential-key-id=test", + "--qwen.managed-agent.runtime-broker.credential-key=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=", + "--qwen.managed-agent.runtime-broker.node-executable=" + wrapper, + "--qwen.managed-agent.runtime-broker.worker-entry=" + cli, + "--qwen.managed-agent.runtime-broker.verified-workspace-recovery-enabled=false", + "--qwen.managed-agent.runtime-broker.cli-entry=" + cli)); + for (int index = 0; index < workspaces.size(); index++) { + String prefix = "--qwen.managed-agent.runtime-broker.workspace-mounts[" + index + "]."; + arguments.add(prefix + "tenant-id=" + tenant); + arguments.add(prefix + "storage-id=storage-" + index); + arguments.add(prefix + "root=" + workspaces.get(index)); + } + var application = new SpringApplicationBuilder(ManagedAgentServerApplication.class, + PublicationProbeWiring.class); + try (var spring = (ServletWebServerApplicationContext) application.run(arguments.toArray(String[]::new))) { + JdbcTemplate jdbc = spring.getBean(JdbcTemplate.class); + var metadata = jdbc.queryForMap("SELECT VERSION() AS version, @@version_comment AS engine"); + System.out.println("A2_DATABASE " + metadata); + assertThat(metadata.toString().toLowerCase()) + .as("A2 reproduction requires MariaDB/MySQL, got %s", metadata) + .containsAnyOf("mysql", "mariadb"); + ManagedAgentStore store = spring.getBean(ManagedAgentStore.class); + var sessions = new ArrayList>(); + String[] profiles = {"hosted-workspace-shell/1", "hosted-workspace-files/1", + "hosted-workspace-files/1"}; + for (int index = 0; index < 3; index++) { + String workspaceId = workspaceIds.get(index); + jdbc.update("INSERT INTO managed_workspace_registry (tenant_id, workspace_id," + + " workspace_generation, storage_id, display_name, config_ref, policy_ref," + + " state) VALUES (?, ?, 1, ?, 'Workspace', ?, ?, 'ACTIVE')", + tenant, workspaceId, "storage-" + index, + WorkspaceExecutionProfile.CONFIG_REF, WorkspaceExecutionProfile.POLICY_REF); + jdbc.update("INSERT INTO managed_workspace_access (tenant_id, workspace_id, actor_id," + + " role) VALUES (?, ?, ?, 'OPERATOR')", + tenant, workspaceId, "actor".getBytes(StandardCharsets.UTF_8)); + var created = store.insertWorkspaceSessionCommand(tenant, "actor", "create-" + index, + "sha256:" + "a".repeat(64), "qwen-code", null, null, List.of(), null, + new WorkspaceSelection(workspaceId, "child")); + sessions.add(Map.of( + "sessionId", created.sessionId(), + "workspaceId", workspaceId, + "directory", workspaces.get(index).resolve("child").toString(), + "toolProfile", profiles[index], + // α's admitted background start is quick-exit, so the + // leftover execution is cheap to leave unresolved. + "shellCommand", index == 0 ? "sleep 2" : "printf PROBE_OK\\n")); + } + EmbeddedRuntimeBroker broker = spring.getBean(EmbeddedRuntimeBroker.class); + Path phaseFile = temporary.resolve("phase"); + Path clearedFile = temporary.resolve("cleared"); + Path config = temporary.resolve("driver.json"); + var driverConfig = new java.util.LinkedHashMap(); + driverConfig.put("tenantId", tenant); + driverConfig.put("storeUrl", "http://127.0.0.1:" + spring.getWebServer().getPort()); + driverConfig.put("brokerUrl", broker.getBaseUri().toString()); + driverConfig.put("cliEntry", cli.toString()); + driverConfig.put("driveFault", !control); + driverConfig.put("control", control); + if (!control) { + driverConfig.put("injectReserveDenyForSessionId", + sessions.get(0).get("sessionId").toString()); + } + driverConfig.put("bPreloaded", "true".equals(System.getProperty("qwen.wedge.preloaded"))); + driverConfig.put("bCaptureBytes", Long.getLong("qwen.wedge.bCaptureBytes", 0L)); + driverConfig.put("bToolProfile", System.getProperty("qwen.wedge.bToolProfile", "")); + driverConfig.put("holdAfterBlockMs", Long.getLong("qwen.wedge.holdMs", 0L)); + driverConfig.put("phaseFile", phaseFile.toString()); + driverConfig.put("clearedFile", clearedFile.toString()); + driverConfig.put("wireFile", temporary.resolve("wire.jsonl").toString()); + driverConfig.put("evidenceDir", evidence.toString()); + driverConfig.put("reportsDir", reportsDir.toString()); + driverConfig.put("sessions", sessions); + new ObjectMapper().writeValue(config.toFile(), driverConfig); + Path driverLog = temporary.resolve("driver.log"); + Process driver = new ProcessBuilder(node, "--import", "tsx", + "integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts", + config.toString()) + .directory(repoRoot.toFile()) + .redirectErrorStream(true).redirectOutput(driverLog.toFile()).start(); + Files.writeString(evidence.resolve("sessions.txt"), + "tenant=" + tenant + "\nalpha=" + sessions.get(0).get("sessionId") + + "\nbeta=" + sessions.get(1).get("sessionId") + + "\ngamma=" + sessions.get(2).get("sessionId") + "\n"); + try { + if (control) { + assertThat(driver.waitFor(300, TimeUnit.SECONDS)) + .as("Control driver timeout:\n%s", Files.readString(driverLog)).isTrue(); + Files.copy(driverLog, evidence.resolve("driver-final.log"), + java.nio.file.StandardCopyOption.REPLACE_EXISTING); + String controlLog = Files.readString(driverLog); + System.out.println(controlLog); + assertThat(driver.exitValue()) + .as("Control driver output:\n%s", controlLog).isZero(); + assertThat(controlLog).contains("HOSTED_WEDGE_CONTROL_OK"); + return; + } + boolean wedged = false; + long watchDeadline = System.currentTimeMillis() + 900_000; + while (System.currentTimeMillis() < watchDeadline) { + if (Files.exists(phaseFile) + && Files.readString(phaseFile).trim().equals("B_WEDGED_CONFIRMED")) { + wedged = true; + break; + } + if (!driver.isAlive()) break; + Thread.sleep(400); + } + Files.copy(driverLog, evidence.resolve("driver.log"), + java.nio.file.StandardCopyOption.REPLACE_EXISTING); + if (!wedged) { + String log = Files.readString(driverLog); + if (!driver.isAlive()) { + assertThat(driver.waitFor(60, TimeUnit.SECONDS)) + .as("Driver exit wait:\n%s", log).isTrue(); + if (log.contains("HOSTED_WEDGE_ABSENT") && driver.exitValue() == 0) { + // Regression expectation: with α recovery-blocked, the + // probes' turns complete. Anything else is the A2 + // wedge returning. + dumpAll(jdbc, evidence, sessions, tenant, "not-wedged"); + return; + } + if (log.contains("HOSTED_RECOVERY_WEDGE_OK")) { + dumpAll(jdbc, evidence, sessions, tenant, "wedge-regression"); + throw new AssertionError( + "A2 REGRESSION: a probe turn wedged with α recovery-blocked" + + " and only completed after α's rows were cleared." + + " See " + evidence + " and driver log:\n" + log); + } + throw new AssertionError("Driver exited unexpectedly (wedged=" + wedged + + "). Log:\n" + log); + } + throw new AssertionError("Watch deadline expired with the driver still" + + " running (wedged=" + wedged + "). Log:\n" + log); + } + + // --- A2 reproduced: capture the full park-point evidence -------- + Files.writeString(evidence.resolve("java-threads.txt"), threadDump()); + dumpAll(jdbc, evidence, sessions, tenant, "before-clear"); + + // --- Clear session α's store rows, like the rig operator did --- + String alpha = sessions.get(0).get("sessionId").toString(); + String clearReport = clearSessionRows(jdbc, alpha); + Files.writeString(evidence.resolve("clear-alpha.txt"), clearReport); + Files.writeString(clearedFile, clearReport); + dumpAll(jdbc, evidence, sessions, tenant, "after-clear"); + + // The driver's post-clear budget is two sequential 240 s + // windows plus a 30 s transcript fetch; exceeding it turns a + // legitimate recovery into an opaque timeout and discards the + // clear-and-recover evidence this witness exists to collect. + assertThat(driver.waitFor(600, TimeUnit.SECONDS)) + .as("Driver timeout:\n%s", Files.readString(driverLog)).isTrue(); + Files.copy(driverLog, evidence.resolve("driver-final.log"), + java.nio.file.StandardCopyOption.REPLACE_EXISTING); + String log = Files.readString(driverLog); + System.out.println(log); + assertThat(driver.exitValue()).as("Driver output:\n%s", log).isZero(); + throw new AssertionError( + "A2 REGRESSION: a probe turn wedged with α recovery-blocked and only" + + " completed after α's rows were cleared. See " + evidence + + " and driver log:\n" + log); + } finally { + driver.descendants().forEach(ProcessHandle::destroyForcibly); + if (driver.isAlive()) { + driver.destroyForcibly(); + } + } + } + } + + private static String threadDump() { + ThreadMXBean bean = ManagementFactory.getThreadMXBean(); + StringBuilder out = new StringBuilder(); + out.append("threadCount=").append(bean.getThreadCount()).append('\n'); + for (ThreadInfo info : bean.dumpAllThreads(true, true)) { + out.append('\n').append('"').append(info.getThreadName()).append('"') + .append(" id=").append(info.getThreadId()) + .append(" state=").append(info.getThreadState()) + .append(" lock=").append(info.getLockName()) + .append(" lockOwner=").append(info.getLockOwnerName()) + .append('\n'); + for (StackTraceElement frame : info.getStackTrace()) { + out.append(" at ").append(frame).append('\n'); + } + for (var monitor : info.getLockedMonitors()) { + out.append(" - locked ").append(monitor).append('\n'); + } + for (var ownable : info.getLockedSynchronizers()) { + out.append(" - sync ").append(ownable).append('\n'); + } + } + return out.toString(); + } + + private void dumpAll(JdbcTemplate jdbc, Path evidence, List> sessions, + String tenant, String tag) throws Exception { + StringBuilder out = new StringBuilder(); + query(out, jdbc, "PROCESSLIST", + "SELECT ID, USER, HOST, DB, COMMAND, TIME, STATE, LEFT(INFO, 400) AS INFO" + + " FROM information_schema.PROCESSLIST ORDER BY ID"); + query(out, jdbc, "INNODB_TRX", + "SELECT TRX_ID, TRX_STATE, TRX_STARTED, TRX_MYSQL_THREAD_ID, TRX_ROWS_LOCKED," + + " TRX_ISOLATION_LEVEL, LEFT(TRX_QUERY, 400) AS TRX_QUERY" + + " FROM information_schema.INNODB_TRX"); + query(out, jdbc, "DATA_LOCKS", + "SELECT ENGINE, ENGINE_LOCK_ID, ENGINE_TRANSACTION_ID, THREAD_ID, OBJECT_SCHEMA," + + " OBJECT_NAME, LOCK_TYPE, LOCK_MODE, LOCK_STATUS, LEFT(LOCK_DATA, 120) AS LOCK_DATA" + + " FROM performance_schema.data_locks"); + query(out, jdbc, "INNODB_LOCKS_LEGACY", "SELECT * FROM information_schema.INNODB_LOCKS"); + query(out, jdbc, "INNODB_LOCK_WAITS_LEGACY", "SELECT * FROM information_schema.INNODB_LOCK_WAITS"); + query(out, jdbc, "METADATA_LOCKS", + "SELECT OBJECT_SCHEMA, OBJECT_NAME, LOCK_TYPE, LOCK_DURATION, LOCK_STATUS," + + " SOURCE, OWNER_THREAD_ID FROM performance_schema.metadata_locks"); + try { + var status = jdbc.queryForMap("SHOW ENGINE INNODB STATUS"); + out.append("=== SHOW ENGINE INNODB STATUS ===\n").append(status.get("Status")).append('\n'); + } catch (Exception error) { + out.append("INNODB STATUS failed: ").append(error).append('\n'); + } + for (Map session : sessions) { + String id = session.get("sessionId").toString(); + query(out, jdbc, "journal_head " + id, + "SELECT tenant_id, session_id, workspace_id, state, recovery_status," + + " recovery_detail_code, writer_id, writer_generation," + + " journal_revision, committed_sequence, activation_epoch," + + " writer_lease_until, updated_at FROM" + + " qwen_managed_session_journal_head WHERE tenant_id = ? AND" + + " session_id = ?", tenant, id); + query(out, jdbc, "journal_tx " + id, + "SELECT session_id, operation, command_id, journal_revision, first_sequence," + + " last_sequence, created_at FROM qwen_managed_session_journal_tx" + + " WHERE tenant_id = ? AND session_id = ? ORDER BY journal_revision" + + " DESC LIMIT 25", tenant, id); + query(out, jdbc, "publications " + id, + "SELECT publication_id, tenant_id, session_id, retention_state, state," + + " capture_used_bytes FROM qwen_tool_publication WHERE tenant_id = ?" + + " AND session_id = ?", tenant, id); + } + dumpSessionRows(jdbc, out, sessions); + Files.writeString(evidence.resolve("db-" + tag + ".txt"), out.toString()); + } + + private static void query(StringBuilder out, JdbcTemplate jdbc, String label, String sql, + Object... args) { + out.append("=== ").append(label).append(" ===\n"); + try { + for (Map row : jdbc.queryForList(sql, args)) { + out.append(row).append('\n'); + } + } catch (Exception error) { + out.append("query failed: ").append(error.getMessage()).append('\n'); + } + } + + /** Every table row keyed by any probe session id, through any session-ish column. */ + private static void dumpSessionRows(JdbcTemplate jdbc, StringBuilder out, + List> sessions) throws Exception { + List ids = sessions.stream() + .map(session -> session.get("sessionId").toString()).toList(); + String placeholders = String.join(", ", java.util.Collections.nCopies(ids.size(), "?")); + for (String table : sessionTables(jdbc)) { + for (String column : sessionColumns(jdbc, table)) { + out.append("=== rows ").append(table).append('.').append(column).append(" ===\n"); + try { + List> rows = jdbc.queryForList( + "SELECT * FROM " + table + " WHERE " + column + " IN (" + + placeholders + ") LIMIT 40", ids.toArray()); + for (Map row : rows) out.append(row).append('\n'); + out.append("(count=").append(rows.size()); + if (rows.size() == 40) { + out.append(", sample capped at 40"); + } + out.append(")\n"); + } catch (Exception error) { + out.append("query failed: ").append(error.getMessage()).append('\n'); + } + } + } + } + + private static List sessionTables(JdbcTemplate jdbc) { + return jdbc.queryForList( + "SELECT DISTINCT table_name FROM information_schema.columns" + + " WHERE table_schema = DATABASE() AND (column_name = 'session_id'" + + " OR column_name = 'harness_session_id') ORDER BY table_name", + String.class); + } + + private static List sessionColumns(JdbcTemplate jdbc, String table) { + return jdbc.queryForList( + "SELECT column_name FROM information_schema.columns WHERE table_schema = DATABASE()" + + " AND table_name = ? AND (column_name = 'session_id'" + + " OR column_name = 'harness_session_id') ORDER BY column_name", + String.class, table); + } + + private static String clearSessionRows(JdbcTemplate jdbc, String sessionId) throws Exception { + StringBuilder out = new StringBuilder(); + String database; + try (Connection connection = jdbc.getDataSource().getConnection()) { + database = connection.getCatalog(); + } + out.append("clearing session ").append(sessionId).append(" in ").append(database).append('\n'); + for (int round = 0; round < 8; round++) { + int errors = 0; + out.append("--- round ").append(round).append(" ---\n"); + for (String table : sessionTables(jdbc)) { + for (String column : sessionColumns(jdbc, table)) { + try { + int deleted = jdbc.update("DELETE FROM " + table + " WHERE " + column + " = ?", + sessionId); + if (deleted > 0) + out.append("deleted ").append(deleted).append(" from ").append(table) + .append('.').append(column).append('\n'); + } catch (Exception error) { + errors++; + out.append("FAILED ").append(table).append('.').append(column).append(' ') + .append(error.getMessage()).append('\n'); + } + } + } + if (errors == 0) { + out.append("clear complete in round ").append(round).append('\n'); + return out.toString(); + } + Thread.sleep(500); + } + out.append("CLEAR INCOMPLETE after 8 rounds\n"); + return out.toString(); + } +} diff --git a/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ToolPublicationContractTest.java b/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ToolPublicationContractTest.java index 2a264823fa..07120c466e 100644 --- a/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ToolPublicationContractTest.java +++ b/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ToolPublicationContractTest.java @@ -91,6 +91,32 @@ class ToolPublicationContractTest { new byte[65537])).isInstanceOf(IllegalArgumentException.class); } + /** + * Replays the shared payload vectors — the digests are pinned from the + * TypeScript canonicalizer, so an accepted vector also proves the two + * implementations canonicalize the boolean marker fields byte-identically. + */ + @Test + void acceptsAndRefusesThePublicationPayloadFamilies() throws Exception { + JsonNode suite = JSON.readTree(contracts().resolve( + "managed-tool-publication-v1.fixtures.json").toFile()); + JsonNode base = suite.required("cases").get(0).required("value"); + for (JsonNode vector : suite.required("payloadVectors")) { + ObjectNode binding = (ObjectNode) base.deepCopy(); + binding.put("requestDigest", vector.required("requestDigest").asText()); + ((ObjectNode) binding.get("reference")).put("argsDigest", + vector.required("argsDigest").asText()); + String payload = vector.required("payloadJson").asText(); + if (vector.required("valid").asBoolean()) { + ToolPublicationContract.requirePayload(binding, payload); + } else { + assertThatThrownBy(() -> ToolPublicationContract.requirePayload(binding, payload)) + .as(vector.required("id").asText()) + .isInstanceOf(IllegalArgumentException.class); + } + } + } + @Test void acceptsEcmascriptCanonicalShellInputWithControlCharacters() throws Exception { JsonNode suite = JSON.readTree(contracts().resolve( diff --git a/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/managedprobes/ProbeAdvice.java b/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/managedprobes/ProbeAdvice.java new file mode 100644 index 0000000000..078e27e071 --- /dev/null +++ b/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/managedprobes/ProbeAdvice.java @@ -0,0 +1,37 @@ +package com.alibaba.qwen.managedprobes; + +import jakarta.servlet.http.HttpServletRequest; +import java.util.Map; +import org.springframework.core.Ordered; +import org.springframework.core.annotation.Order; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.ExceptionHandler; +import org.springframework.web.bind.annotation.RestControllerAdvice; + +/** + * Probe-lane diagnosis: the production 400 handler answers with a + * deliberately generic envelope and no log, so contract refusals are + * unreadable on the wire. This advice logs the full cause and embeds it in + * the envelope. It lives outside the scanned {@code com.alibaba.qwen.code} + * package, so only a probe witness that lists it as an explicit application + * source registers it; {@code HostedRecoveryWedgeGate} deliberately omits it + * and observes the production envelope as it is. + */ +@RestControllerAdvice +@Order(Ordered.HIGHEST_PRECEDENCE) +public final class ProbeAdvice { + + @ExceptionHandler(IllegalArgumentException.class) + ResponseEntity> probeInvalid(IllegalArgumentException error, + HttpServletRequest request) { + System.out.println("PROBE_CONTRACT_REFUSAL " + request.getMethod() + " " + + request.getRequestURI() + " query=" + request.getQueryString()); + error.printStackTrace(System.out); + return ResponseEntity.status(HttpStatus.BAD_REQUEST).body(Map.of( + "error", Map.of( + "code", "invalid_request", + "message", "The request body is invalid.", + "diagnostic", String.valueOf(error.getMessage())))); + } +} diff --git a/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/managedprobes/PublicationProbeWiring.java b/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/managedprobes/PublicationProbeWiring.java new file mode 100644 index 0000000000..74d6ed8f4f --- /dev/null +++ b/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/managedprobes/PublicationProbeWiring.java @@ -0,0 +1,116 @@ +package com.alibaba.qwen.managedprobes; + +import com.alibaba.qwen.code.managedagent.api.ToolPublicationController; +import com.alibaba.qwen.code.managedagent.config.ManagedAgentProperties; +import com.alibaba.qwen.code.managedagent.store.ManagedSessionStore; +import com.alibaba.qwen.code.managedagent.store.ToolPublicationAdmissionStore; +import com.alibaba.qwen.code.managedagent.store.ToolPublicationContract; +import com.alibaba.qwen.code.managedagent.store.ToolPublicationDataStore; +import com.alibaba.qwen.code.managedagent.store.ToolPublicationObjectStore; +import com.alibaba.qwen.code.managedagent.store.ToolPublicationStore; +import com.alibaba.qwen.code.runtimebroker.RuntimeBindingRepository; +import com.alibaba.qwen.code.runtimebroker.ToolExecutionRepository; +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.time.Duration; +import java.util.concurrent.ConcurrentHashMap; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.transaction.PlatformTransactionManager; + +/** + * The real publication controller and grant/data/admission stores with an + * in-memory object store, mirroring the conditional OSS-backed + * {@code ToolPublicationConfiguration} that only provisioned rigs can boot. + * Object bytes are never touched by the probe ITs' reserve/close arms, so + * the store can be a stable in-memory stub. + * + *

Shared by {@code HostedBackgroundPublicationGate} and + * {@code HostedRecoveryWedgeGate}. It deliberately contains no request + * advice: the refusal-diagnosis advice (which changes the 400 envelope) must + * exist only in the publication witness's context, since the wedge witness + * observes the production envelope as it is. + */ +@Configuration(proxyBeanMethods = false) +public final class PublicationProbeWiring { + + private static final long EXECUTION_BYTES = 268_435_456L; + private static final long SESSION_BYTES = 536_870_912L; + private static final long TENANT_BYTES = 1_073_741_824L; + private static final long ACTIVE_CAPTURES = 8; + + @Bean + ToolPublicationObjectStore publicationObjectStore() { + return new InMemoryObjectStore(); + } + + @Bean + ToolPublicationStore publicationStore(JdbcTemplate jdbc, + PlatformTransactionManager manager, ManagedSessionStore sessions, + ToolExecutionRepository executions, RuntimeBindingRepository bindings) { + return new ToolPublicationStore(jdbc, manager, sessions, executions, + bindings, new ToolPublicationStore.Capacity(EXECUTION_BYTES, + SESSION_BYTES, TENANT_BYTES, ACTIVE_CAPTURES), false); + } + + @Bean + ToolPublicationDataStore publicationDataStore(JdbcTemplate jdbc, + PlatformTransactionManager manager, ToolPublicationStore grants, + ManagedSessionStore sessions, ToolPublicationObjectStore objects) { + var operationTimeout = Duration.ofSeconds(60); + var budget = new ToolPublicationDataStore.VerificationBudget( + 1_048_576L, Duration.ofMinutes(10)); + budget.timeout(operationTimeout, Math.addExact( + EXECUTION_BYTES, ToolPublicationContract.PRODUCER_BYTES)); + return new ToolPublicationDataStore(jdbc, manager, grants, sessions, + objects, operationTimeout, Duration.ofSeconds(30), budget); + } + + @Bean + ToolPublicationAdmissionStore publicationAdmissionStore(JdbcTemplate jdbc, + PlatformTransactionManager manager, ManagedSessionStore sessions, + ToolPublicationDataStore data) { + return new ToolPublicationAdmissionStore(jdbc, manager, sessions, data); + } + + @Bean + ToolPublicationController publicationController(ToolPublicationStore grants, + ToolPublicationDataStore data, ToolPublicationAdmissionStore admissions, + ManagedAgentProperties properties) { + return new ToolPublicationController(grants, data, admissions, properties); + } + + static final class InMemoryObjectStore implements ToolPublicationObjectStore { + private final ConcurrentHashMap objects = new ConcurrentHashMap<>(); + + @Override + public void requireUnversioned() { + } + + @Override + public void putIfAbsent(String key, byte[] bytes) { + objects.putIfAbsent(key, bytes); + } + + @Override + public void deleteIfPresent(String key) { + objects.remove(key); + } + + @Override + public InputStream open(String key) { + byte[] bytes = objects.get(key); + if (bytes == null) { + throw new IllegalStateException("Unknown object " + key); + } + return new ByteArrayInputStream(bytes); + } + + @Override + public InputStream open(String key, Runnable guard) { + guard.run(); + return open(key); + } + } +} diff --git a/scripts/check-failsafe-reports.js b/scripts/check-failsafe-reports.js index c89cfc0b21..84192cd218 100644 --- a/scripts/check-failsafe-reports.js +++ b/scripts/check-failsafe-reports.js @@ -20,17 +20,19 @@ import { existsSync, readdirSync, readFileSync } from 'node:fs'; import path from 'node:path'; const isHosted = (className) => /^Hosted.*IT$/.test(className); +const isGate = (className) => /^Hosted.*Gate$/.test(className); const [family, ...modules] = process.argv.slice(2); const optIn = family === 'o4-mysql' || family === 'o4-oss'; const inFamily = new Map([ ['hosted', isHosted], ['non-hosted', (className) => /IT$/.test(className) && !isHosted(className)], + ['hosted-gates', isGate], ['o4-mysql', (className) => className === 'O4MySqlGate'], ['o4-oss', (className) => className === 'O4OssGate'], ]).get(family); if (!inFamily || modules.length === 0) { console.error( - 'usage: node scripts/check-failsafe-reports.js hosted|non-hosted|o4-mysql|o4-oss ...', + 'usage: node scripts/check-failsafe-reports.js hosted|non-hosted|hosted-gates|o4-mysql|o4-oss ...', ); process.exit(2); } diff --git a/scripts/tests/check-failsafe-reports.test.js b/scripts/tests/check-failsafe-reports.test.js index b9c7238c05..7e23450d12 100644 --- a/scripts/tests/check-failsafe-reports.test.js +++ b/scripts/tests/check-failsafe-reports.test.js @@ -256,6 +256,28 @@ describe('check-failsafe-reports', () => { expect(check('hosted', noFamily).status).toBe(0); }); + it('hosts admission gates as their own family and rejects drift in both directions', () => { + const gates = module('gates', ['com.example.HostedAdmissionGate'], { + 'com.example.HostedAdmissionGate': [1, 0], + }); + expect(check('hosted-gates', gates).status).toBe(0); + expect(check('hosted-gates', gates).output).toContain( + 'gates: com.example.HostedAdmissionGate ran 1 test(s)', + ); + const outside = module('outsiders', ['com.example.HostedFlowIT'], { + 'com.example.HostedFlowIT': [1, 0], + }); + expect(check('hosted-gates', outside).output).toContain( + 'com.example.HostedFlowIT ran outside the hosted-gates family', + ); + expect(check('hosted-gates', outside).status).toBe(1); + const unrun = module('unrun-gate', ['com.example.HostedWedgeGate']); + expect(check('hosted-gates', unrun).output).toContain( + 'com.example.HostedWedgeGate ran no test in this job', + ); + expect(check('hosted-gates', unrun).status).toBe(1); + }); + it('refuses an unknown family or a missing module', () => { expect(check('mariadb', root).status).toBe(2); expect(check('constructor', root).status).toBe(2); diff --git a/scripts/tests/hosted-process-ci.test.js b/scripts/tests/hosted-process-ci.test.js index 030885137f..23b7bbae4f 100644 --- a/scripts/tests/hosted-process-ci.test.js +++ b/scripts/tests/hosted-process-ci.test.js @@ -262,12 +262,12 @@ describe('Hosted real-process gates', () => { (total, step) => total + (step['timeout-minutes'] ?? 0), 0, ); - // Step ceilings today (35 + 9x10 + 20); the uncapped setup steps need - // their own allowance — the job comment estimates it at ~13, so the - // enforced floor moves with that estimate, not the older 10. - expect(summed).toBe(145); + // Step ceilings today (35 + 9x10 + 20 + 50); the uncapped setup steps + // need their own allowance — the job comment estimates it at ~13, so + // the enforced floor moves with that estimate, not the older 10. + expect(summed).toBe(195); expect(job['timeout-minutes']).toBeGreaterThanOrEqual(summed + 13); - expect(job['timeout-minutes']).toBe(160); + expect(job['timeout-minutes']).toBe(210); }); it.each([