From fe816d69ef5dcea1620f44dcb256f0d5eb98ea14 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 18 Aug 2026 17:14:43 -0700 Subject: [PATCH] fix(gateway): route detached announce by instance (#125946) * test(qa): cover worker generation reload * test(qa): anchor worker generation fixture * test(qa): strengthen worker generation proof * test(qa): stabilize terminal reply smoke waits * test(qa): widen terminal reply CI budget * test(qa): drop superseded timeout workaround * fix(gateway): route detached announce by instance --- .../src/scenario-catalog.inference.test.ts | 41 ++ .../worker-inference-generation-reload.yaml | 32 + ...subagent-announce-delivery.runtime.test.ts | 34 +- .../announce/subagent-announce.runtime.ts | 2 +- .../index.js | 138 ++++ .../openclaw.plugin.json | 20 + .../worker-inference-generation-reload.ts | 649 ++++++++++++++++++ 7 files changed, 893 insertions(+), 23 deletions(-) create mode 100644 qa/scenarios/runtime/worker-inference-generation-reload.yaml create mode 100644 test/e2e/qa-lab/runtime/fixtures/worker-inference-generation-provider/index.js create mode 100644 test/e2e/qa-lab/runtime/fixtures/worker-inference-generation-provider/openclaw.plugin.json create mode 100644 test/e2e/qa-lab/runtime/worker-inference-generation-reload.ts diff --git a/extensions/qa-lab/src/scenario-catalog.inference.test.ts b/extensions/qa-lab/src/scenario-catalog.inference.test.ts index cefd57b04e62..c6701f59f7a5 100644 --- a/extensions/qa-lab/src/scenario-catalog.inference.test.ts +++ b/extensions/qa-lab/src/scenario-catalog.inference.test.ts @@ -1,3 +1,5 @@ +import fs from "node:fs"; +import path from "node:path"; import { describe, expect, it } from "vitest"; import { readQaScenarioById, readQaScenarioExecutionConfig } from "./scenario-catalog.js"; import { requireFlowScenario } from "./scenario-catalog.test-utils.js"; @@ -53,4 +55,43 @@ describe("QA inference scenario catalog", () => { }); expect(scenario.execution.flow?.steps).toHaveLength(1); }); + + it("keeps worker inference provider stages on one admitted generation", () => { + const scenario = readQaScenarioById("worker-inference-generation-reload"); + const scriptPath = path.join( + import.meta.dirname, + "../../../test/e2e/qa-lab/runtime/worker-inference-generation-reload.ts", + ); + const fixturePath = path.join( + import.meta.dirname, + "../../../test/e2e/qa-lab/runtime/fixtures/worker-inference-generation-provider/index.js", + ); + const script = fs.readFileSync(scriptPath, "utf8"); + const fixture = fs.readFileSync(fixturePath, "utf8"); + + expect(scenario.execution).toMatchObject({ + kind: "script", + path: "test/e2e/qa-lab/runtime/worker-inference-generation-reload.ts", + timeoutMs: 900_000, + }); + expect(scenario.successCriteria).toContain( + "Releasing B overrides the placement's stale A lifecycle scope and constructs the provider factory, stream policy, wrapper, and stream execution entirely from generation B with B's prepared model and runtime credential.", + ); + expect(script).toContain("createPairedNodeWorkerHost"); + expect(script).toContain("startQaMockOpenAiServer"); + expect(script).toContain( + 'const OWNERSHIP_STAGES = ["factory", "policy", "wrapper", "execution"]', + ); + expect(script).toContain("hotPublishGeneration"); + expect(script).toContain("startAuthInspectingProxy"); + expect(script).toContain("runtimeCredentialGenerations"); + expect(script).toContain("replyCounts[reply] !== 1"); + expect(script).not.toContain("createWorkerInferenceExecutor"); + expect(fixture).toContain('from "openclaw/plugin-sdk/llm"'); + expect(fixture).toContain("prepareRuntimeAuth"); + expect(fixture).toContain("createStreamFn"); + expect(fixture).toContain("prepareExtraParams"); + expect(fixture).toContain("wrapStreamFn"); + expect(fixture).not.toContain("src/"); + }); }); diff --git a/qa/scenarios/runtime/worker-inference-generation-reload.yaml b/qa/scenarios/runtime/worker-inference-generation-reload.yaml new file mode 100644 index 000000000000..58f1136159d2 --- /dev/null +++ b/qa/scenarios/runtime/worker-inference-generation-reload.yaml @@ -0,0 +1,32 @@ +title: Worker inference generation reload + +scenario: + id: worker-inference-generation-reload + surface: gateway + category: gateway.session-apis + coverage: + secondary: + - plugins.activation-hot-reload + - gateway.config-reload-hot-reload + objective: Prove Gateway-owned worker inference keeps its admitted provider generation when a newer generation hot-publishes after model and auth preparation but before provider stream construction. + successCriteria: + - A managed-worktree session dispatches to a real paired-node worker through an isolated Gateway, establishes generation A's placement lifecycle, and uses mock-openai for provider transport. + - After generation B publishes, a second turn resolves B's model and source credential, then pauses in its final provider-owned runtime-auth preparation while generation C hot-publishes in the same Gateway process. + - Releasing B overrides the placement's stale A lifecycle scope and constructs the provider factory, stream policy, wrapper, and stream execution entirely from generation B with B's prepared model and runtime credential. + - A subsequent turn on the same worker placement constructs the same four stages entirely from generation C, and all three turns persist their exact replies once. + docsRefs: + - docs/gateway/cloud-workers.md + - docs/concepts/qa-e2e-automation.md + codeRefs: + - src/gateway/worker-environments/inference-runtime.ts + - src/agents/simple-completion-runtime.ts + - test/e2e/qa-lab/runtime/worker-inference-generation-reload.ts + - test/e2e/qa-lab/runtime/fixtures/worker-inference-generation-provider/index.js + execution: + kind: script + path: test/e2e/qa-lab/runtime/worker-inference-generation-reload.ts + summary: Establishes a generation A paired-node worker placement, hot-publishes C while a generation B turn is paused in runtime-auth preparation, and proves B overrides the stale placement scope before a fresh C turn. + timeoutMs: 900000 + args: + - --artifact-base + - ${outputDir} diff --git a/src/agents/subagents/announce/subagent-announce-delivery.runtime.test.ts b/src/agents/subagents/announce/subagent-announce-delivery.runtime.test.ts index 9a1b27c0c1e7..5ebbc75ce806 100644 --- a/src/agents/subagents/announce/subagent-announce-delivery.runtime.test.ts +++ b/src/agents/subagents/announce/subagent-announce-delivery.runtime.test.ts @@ -1,17 +1,17 @@ -import { afterEach, describe, expect, it, vi } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { WRITE_SCOPE } from "../../../gateway/method-scopes.js"; import { createGatewayMethodRegistry } from "../../../gateway/methods/registry.js"; -import { createGatewayInstanceRuntime } from "../../../gateway/server-instance-runtime.js"; import type { GatewayRequestContext, GatewayRequestHandlers, } from "../../../gateway/server-methods/types.js"; import { dispatchSubagentAnnounceAgent } from "./subagent-announce-delivery.runtime.js"; -function createContext(): GatewayRequestContext { +function createContext(handlers: GatewayRequestHandlers): GatewayRequestContext { return { deps: {}, getRuntimeConfig: () => ({}), + getGatewayMethodRegistry: () => createRegistry(handlers), logGateway: { warn: vi.fn(), error: vi.fn(), @@ -34,30 +34,16 @@ function createRegistry(handlers: GatewayRequestHandlers) { } describe("subagent announce Gateway instance dispatch", () => { - let closeRuntime: (() => void) | undefined; - - afterEach(() => { - closeRuntime?.(); - closeRuntime = undefined; - }); - - it("delivers a detached announce after the originating request scope has ended", async () => { - const context = createContext(); + it("delivers a detached announce through its explicit instance resolver", async () => { + const context = createContext({ + agent: ({ respond }) => respond(true, { raw: true }), + }); const idempotencyKey = "detached-subagent-announce"; context.dedupe.set(`agent:${idempotencyKey}`, { ts: Date.now(), ok: true, payload: { runId: "announce-run", status: "ok", summary: "delivered" }, }); - const runtime = createGatewayInstanceRuntime({ - getContext: () => context, - getMethodRegistry: () => - createRegistry({ - agent: ({ respond }) => respond(true, { raw: true }), - }), - isDispatchAvailable: () => true, - }); - closeRuntime = runtime.close; await expect( dispatchSubagentAnnounceAgent( @@ -65,7 +51,11 @@ describe("subagent announce Gateway instance dispatch", () => { message: "Process one completed child result.", idempotencyKey, }, - { expectFinal: true, forceSyntheticClient: true }, + { + expectFinal: true, + forceSyntheticClient: true, + resolveGatewayContext: () => context, + }, ), ).resolves.toEqual({ runId: "announce-run", status: "ok", summary: "delivered" }); }); diff --git a/src/agents/subagents/announce/subagent-announce.runtime.ts b/src/agents/subagents/announce/subagent-announce.runtime.ts index 05700a27d0b3..5b3f5bdc4f2b 100644 --- a/src/agents/subagents/announce/subagent-announce.runtime.ts +++ b/src/agents/subagents/announce/subagent-announce.runtime.ts @@ -5,7 +5,7 @@ * IO without changing the announce logic itself. */ import { loadSessionEntry } from "../../../config/sessions/session-accessor.js"; -export { dispatchGatewayLifecycleMethod as dispatchGatewayMethodInProcess } from "../../../gateway/server-recovery-runtime-context.js"; +export { dispatchGatewayMethodInProcess } from "../../../gateway/server-plugin-in-process-dispatch.js"; export { getRuntimeConfig } from "../../../config/config.js"; export { resolveAgentIdFromSessionKey, diff --git a/test/e2e/qa-lab/runtime/fixtures/worker-inference-generation-provider/index.js b/test/e2e/qa-lab/runtime/fixtures/worker-inference-generation-provider/index.js new file mode 100644 index 000000000000..12b01e22184d --- /dev/null +++ b/test/e2e/qa-lab/runtime/fixtures/worker-inference-generation-provider/index.js @@ -0,0 +1,138 @@ +import { appendFileSync, mkdirSync } from "node:fs"; +import { readFile } from "node:fs/promises"; +import path from "node:path"; +import { setTimeout as sleep } from "node:timers/promises"; +import { streamSimple } from "openclaw/plugin-sdk/llm"; +import { definePluginEntry } from "openclaw/plugin-sdk/plugin-entry"; + +const PLUGIN_ID = "qa-worker-generation"; +const PROVIDER_ID = PLUGIN_ID; +const MODEL_ID = "qa-worker-generation-model"; + +function requireConfig(pluginConfig) { + const generation = pluginConfig?.generation; + const tracePath = pluginConfig?.tracePath; + const barrierPath = pluginConfig?.barrierPath; + const mockProviderBaseUrl = pluginConfig?.mockProviderBaseUrl; + if ( + (generation !== "A" && generation !== "B" && generation !== "C") || + typeof tracePath !== "string" || + !tracePath || + typeof barrierPath !== "string" || + !barrierPath || + typeof mockProviderBaseUrl !== "string" || + !mockProviderBaseUrl + ) { + throw new Error("qa worker generation fixture config is incomplete"); + } + return { generation, tracePath, barrierPath, mockProviderBaseUrl }; +} + +function createTraceWriter(tracePath, generation) { + mkdirSync(path.dirname(tracePath), { recursive: true }); + return (event, facts = {}) => { + appendFileSync( + tracePath, + `${JSON.stringify({ event, generation, ...facts, at: Date.now() })}\n`, + "utf8", + ); + }; +} + +async function readBarrierState(barrierPath) { + return await readFile(barrierPath, "utf8") + .then((value) => value.trim()) + .catch(() => ""); +} + +async function waitForBarrierRelease(barrierPath) { + const deadline = Date.now() + 120_000; + while ((await readBarrierState(barrierPath)) !== "released") { + if (Date.now() >= deadline) { + throw new Error("qa worker generation barrier release timed out"); + } + await sleep(25); + } +} + +export default definePluginEntry({ + id: PLUGIN_ID, + name: "QA worker generation fixture", + description: "Exercises worker inference provider ownership across plugin reload.", + register(api) { + const { generation, tracePath, barrierPath, mockProviderBaseUrl } = requireConfig( + api.pluginConfig, + ); + const trace = createTraceWriter(tracePath, generation); + const sourceCredential = api.config.models?.providers?.[PROVIDER_ID]?.apiKey; + if (typeof sourceCredential !== "string" || !sourceCredential) { + throw new Error("qa worker generation provider requires a direct credential"); + } + const runtimeCredential = `qa-worker-runtime-${generation}`; + + trace("registered", { registrationMode: api.registrationMode }); + api.registerReload({ + hotPrefixes: [`plugins.entries.${PLUGIN_ID}.config`], + }); + api.registerProvider({ + id: PROVIDER_ID, + label: "QA worker generation provider", + auth: [], + normalizeResolvedModel: ({ model }) => { + trace("model", { modelMatches: model.id === MODEL_ID }); + return { + ...model, + baseUrl: mockProviderBaseUrl, + params: { + ...model.params, + fixtureBaseUrl: mockProviderBaseUrl, + fixtureGeneration: generation, + }, + }; + }, + prepareRuntimeAuth: async ({ apiKey, model }) => { + const shouldWait = (await readBarrierState(barrierPath)) === "armed"; + trace("auth-prepare", { + modelGenerationMatches: model.params?.fixtureGeneration === generation, + sourceAuthMatchesGeneration: apiKey === sourceCredential, + waited: shouldWait, + }); + if (shouldWait) { + await waitForBarrierRelease(barrierPath); + trace("auth-prepare-released", { waited: true }); + } + trace("auth-ready", { runtimeCredentialGeneration: generation }); + return { apiKey: runtimeCredential }; + }, + createStreamFn: ({ model }) => { + trace("factory", { + modelGenerationMatches: model.params?.fixtureGeneration === generation, + }); + return streamSimple; + }, + prepareExtraParams: ({ model }) => { + trace("policy", { + modelGenerationMatches: model?.params?.fixtureGeneration === generation, + }); + return {}; + }, + wrapStreamFn: ({ model, streamFn }) => { + trace("wrapper", { + modelGenerationMatches: model?.params?.fixtureGeneration === generation, + streamPresent: Boolean(streamFn), + }); + if (!streamFn) { + return undefined; + } + return (streamModel, context, options) => { + trace("execution", { + authPresent: typeof options?.apiKey === "string" && options.apiKey.length > 0, + baseUrlMatchesGeneration: streamModel.baseUrl === mockProviderBaseUrl, + modelGenerationMatches: streamModel.params?.fixtureGeneration === generation, + }); + return streamFn(streamModel, context, options); + }; + }, + }); + }, +}); diff --git a/test/e2e/qa-lab/runtime/fixtures/worker-inference-generation-provider/openclaw.plugin.json b/test/e2e/qa-lab/runtime/fixtures/worker-inference-generation-provider/openclaw.plugin.json new file mode 100644 index 000000000000..df56d36857e6 --- /dev/null +++ b/test/e2e/qa-lab/runtime/fixtures/worker-inference-generation-provider/openclaw.plugin.json @@ -0,0 +1,20 @@ +{ + "id": "qa-worker-generation", + "name": "QA worker generation fixture", + "description": "Exercises worker inference provider ownership across plugin reload.", + "activation": { + "onStartup": true + }, + "providers": ["qa-worker-generation"], + "configSchema": { + "type": "object", + "additionalProperties": false, + "required": ["generation", "tracePath", "barrierPath", "mockProviderBaseUrl"], + "properties": { + "generation": { "type": "string", "enum": ["A", "B", "C"] }, + "tracePath": { "type": "string", "minLength": 1 }, + "barrierPath": { "type": "string", "minLength": 1 }, + "mockProviderBaseUrl": { "type": "string", "minLength": 1 } + } + } +} diff --git a/test/e2e/qa-lab/runtime/worker-inference-generation-reload.ts b/test/e2e/qa-lab/runtime/worker-inference-generation-reload.ts new file mode 100644 index 000000000000..16ef40512adc --- /dev/null +++ b/test/e2e/qa-lab/runtime/worker-inference-generation-reload.ts @@ -0,0 +1,649 @@ +import { randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import { createServer } from "node:http"; +import os from "node:os"; +import path from "node:path"; +import { pathToFileURL } from "node:url"; +import { GatewayClient } from "openclaw/plugin-sdk/gateway-runtime"; +import { + QA_EVIDENCE_FILENAME, + startQaGatewayChild, + startQaMockOpenAiServer, + type QaEvidenceSummaryJson, +} from "../../../../extensions/qa-lab/api.js"; +import type { OpenClawConfig } from "../../../../src/config/types.openclaw.js"; +import { + MODEL_REF as DEFAULT_MOCK_MODEL_REF, + PROOF_TIMEOUT_MS, + waitFor, +} from "./cloud-worker-midturn-loss-fixture.js"; +import workerGenerationProviderFixture from "./fixtures/worker-inference-generation-provider/index.js"; +import { + closeWireServer, + connectWireClient, + createPairedNodeWorkerHost, + createPublishedWireWorkspace, + type PairedNodeWorkerHost, + type PublishedWireWorkspace, + type WireGateway, + wireMessageText, +} from "./paired-node-worker-wire-fixture.js"; +import { createQaScriptEvidenceWriter } from "./script-evidence.js"; + +const SCENARIO_ID = "worker-inference-generation-reload"; +const VERDICT_FILE = `${SCENARIO_ID}-verdict.json`; +const PLUGIN_ID = "qa-worker-generation"; +const PROVIDER_ID = PLUGIN_ID; +const MODEL_ID = "qa-worker-generation-model"; +const MODEL_REF = `${PROVIDER_ID}/${MODEL_ID}`; +const SESSION_KEY = "agent:qa:worker-inference-generation-reload"; +const SOURCE_CREDENTIAL_PREFIX = "qa-worker-source"; +const GENERATION_A_REPLY = "WORKER-GENERATION-A-OK"; +const GENERATION_B_REPLY = "WORKER-GENERATION-B-OK"; +const GENERATION_C_REPLY = "WORKER-GENERATION-C-OK"; +const OWNERSHIP_STAGES = ["factory", "policy", "wrapper", "execution"] as const; + +type Generation = "A" | "B" | "C"; +type ProducerOptions = { artifactBase: string; repoRoot: string }; +type TraceEvent = { + event: string; + generation: Generation; + at: number; + waited?: boolean; + authPresent?: boolean; + baseUrlMatchesGeneration?: boolean; + modelGenerationMatches?: boolean; + sourceAuthMatchesGeneration?: boolean; + streamPresent?: boolean; +}; +type TurnResult = { runId?: string; status?: string; summary?: string }; + +function parseOptions(argv: readonly string[]): ProducerOptions { + const index = argv.indexOf("--artifact-base"); + const artifactBase = index >= 0 ? argv[index + 1] : undefined; + if (!artifactBase) { + throw new Error("--artifact-base is required"); + } + return { artifactBase: path.resolve(artifactBase), repoRoot: process.cwd() }; +} + +async function readTrace(tracePath: string): Promise { + return await fs + .readFile(tracePath, "utf8") + .then((text) => + text + .split(/\r?\n/u) + .filter(Boolean) + .map((line) => JSON.parse(line) as TraceEvent), + ) + .catch(() => []); +} + +function classifyAuthorization(value: string | undefined): Generation | "unknown" { + const credential = value?.replace(/^Bearer\s+/iu, ""); + for (const generation of ["A", "B", "C"] as const) { + if (credential === `qa-worker-runtime-${generation}`) { + return generation; + } + } + return "unknown"; +} + +async function startAuthInspectingProxy(targetBaseUrl: string) { + const authGenerations: Array = []; + const server = createServer((request, response) => { + void (async () => { + const chunks: Buffer[] = []; + for await (const chunk of request) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk))); + } + const body = Buffer.concat(chunks); + if (request.method === "POST") { + authGenerations.push(classifyAuthorization(request.headers.authorization)); + } + const headers = new Headers(); + for (const [name, value] of Object.entries(request.headers)) { + if (value === undefined || ["connection", "content-length", "host"].includes(name)) { + continue; + } + headers.set(name, Array.isArray(value) ? value.join(", ") : value); + } + const upstream = await fetch(new URL(request.url ?? "/", targetBaseUrl), { + method: request.method, + headers, + ...(body.length > 0 ? { body } : {}), + }); + const responseHeaders: Record = {}; + upstream.headers.forEach((value, name) => { + if (!["connection", "content-length", "transfer-encoding"].includes(name)) { + responseHeaders[name] = value; + } + }); + response.writeHead(upstream.status, responseHeaders); + response.end(Buffer.from(await upstream.arrayBuffer())); + })().catch((error) => { + response.writeHead(502, { "content-type": "text/plain; charset=utf-8" }); + response.end(error instanceof Error ? error.message : String(error)); + }); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("worker auth inspection proxy did not bind"); + } + return { + authGenerations, + baseUrl: `http://127.0.0.1:${address.port}`, + async stop() { + server.closeAllConnections(); + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + }, + }; +} + +function buildGenerationConfig(params: { + config: OpenClawConfig; + generation: Generation; + pluginDir: string; + tracePath: string; + barrierPath: string; + mockProviderBaseUrl: string; +}): OpenClawConfig { + const { config, generation, pluginDir, tracePath, barrierPath, mockProviderBaseUrl } = params; + const providerConfig = config.models?.providers?.[PROVIDER_ID]; + return { + ...config, + plugins: { + ...config.plugins, + enabled: true, + allow: [...new Set([...(config.plugins?.allow ?? []), PLUGIN_ID])], + load: { + ...config.plugins?.load, + paths: [...new Set([...(config.plugins?.load?.paths ?? []), pluginDir])], + }, + entries: { + ...config.plugins?.entries, + [PLUGIN_ID]: { + ...config.plugins?.entries?.[PLUGIN_ID], + enabled: true, + config: { + generation, + tracePath, + barrierPath, + mockProviderBaseUrl, + }, + }, + }, + }, + models: { + ...config.models, + providers: { + ...config.models?.providers, + [PROVIDER_ID]: { + ...providerConfig, + api: "openai-responses", + apiKey: `${SOURCE_CREDENTIAL_PREFIX}-${generation}`, + baseUrl: mockProviderBaseUrl, + request: { ...providerConfig?.request, allowPrivateNetwork: true }, + models: [ + { + id: MODEL_ID, + name: "QA worker generation model", + api: "openai-responses", + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 32_768, + contextTokens: 32_768, + maxTokens: 256, + }, + ], + }, + }, + }, + agents: { + ...config.agents, + defaults: { + ...config.agents?.defaults, + model: { primary: MODEL_REF, fallbacks: [] }, + models: { + ...config.agents?.defaults?.models, + [MODEL_REF]: { agentRuntime: { id: "openclaw" } }, + }, + }, + }, + nodeHost: { + ...config.nodeHost, + workerRuns: { enabled: true }, + }, + }; +} + +async function hotPublishGeneration(params: { + gateway: WireGateway; + tracePath: string; + generation: Exclude; +}): Promise<{ pidBefore: number; pidAfter: number }> { + const before = (await params.gateway.call("system.info", {})) as { pid?: number }; + const config = JSON.parse(await fs.readFile(params.gateway.configPath, "utf8")) as OpenClawConfig; + const pluginEntry = config.plugins?.entries?.[PLUGIN_ID]; + const providerConfig = config.models?.providers?.[PROVIDER_ID]; + if (!pluginEntry?.config || !providerConfig) { + throw new Error("generation A config was not installed before hot publish"); + } + const next: OpenClawConfig = { + ...config, + plugins: { + ...config.plugins, + entries: { + ...config.plugins?.entries, + [PLUGIN_ID]: { + ...pluginEntry, + config: { ...pluginEntry.config, generation: params.generation }, + }, + }, + }, + models: { + ...config.models, + providers: { + ...config.models?.providers, + [PROVIDER_ID]: { + ...providerConfig, + apiKey: `${SOURCE_CREDENTIAL_PREFIX}-${params.generation}`, + }, + }, + }, + }; + await fs.writeFile(params.gateway.configPath, `${JSON.stringify(next, null, 2)}\n`, "utf8"); + await waitFor(`generation ${params.generation} provider registration`, async () => { + const registrations = (await readTrace(params.tracePath)).filter( + (event) => event.event === "registered" && event.generation === params.generation, + ); + return registrations.length > 0 ? registrations : undefined; + }); + const after = (await params.gateway.call("system.info", {})) as { pid?: number }; + if (!Number.isSafeInteger(before.pid) || after.pid !== before.pid) { + throw new Error( + `generation ${params.generation} hot publish replaced the Gateway: ${before.pid} -> ${after.pid}`, + ); + } + return { pidBefore: before.pid!, pidAfter: after.pid! }; +} + +async function startTurn(operator: GatewayClient, reply: string): Promise { + const runId = `${SCENARIO_ID}-${randomUUID()}`; + const started = await operator.request("chat.send", { + sessionKey: SESSION_KEY, + message: `Reply exactly: ${reply}`, + deliver: false, + idempotencyKey: runId, + }); + if (started.status !== "started" || started.runId !== runId) { + throw new Error(`chat.send did not start ${reply}: ${JSON.stringify(started)}`); + } + return runId; +} + +async function waitForTurn(operator: GatewayClient, runId: string): Promise { + const result = await operator.request( + "agent.wait", + { runId, timeoutMs: PROOF_TIMEOUT_MS }, + { timeoutMs: PROOF_TIMEOUT_MS + 5_000 }, + ); + if (result.status !== "ok") { + throw new Error(`worker turn failed: ${JSON.stringify(result)}`); + } +} + +async function readHistoryReplyCounts(operator: GatewayClient): Promise> { + const history = await operator.request<{ messages?: unknown[] }>("chat.history", { + sessionKey: SESSION_KEY, + limit: 100, + }); + return Object.fromEntries( + [GENERATION_A_REPLY, GENERATION_B_REPLY, GENERATION_C_REPLY].map((reply) => [ + reply, + (history.messages ?? []).filter( + (message) => + (message as { role?: unknown }).role === "assistant" && + wireMessageText(message) === reply, + ).length, + ]), + ); +} + +async function waitForHistoryReply(operator: GatewayClient, reply: string): Promise { + await waitFor(`one worker history reply ${reply}`, async () => { + const count = (await readHistoryReplyCounts(operator))[reply] ?? 0; + if (count > 1) { + throw new Error(`worker history persisted ${count} copies of ${reply}`); + } + return count === 1 ? true : undefined; + }); +} + +function requireGenerationOwnership(params: { + events: readonly TraceEvent[]; + generation: Generation; + waited: boolean; +}) { + const authEvents = params.events.filter((event) => event.event === "auth-prepare"); + const authReadyEvents = params.events.filter((event) => event.event === "auth-ready"); + const stages = params.events.filter((event) => + OWNERSHIP_STAGES.includes(event.event as (typeof OWNERSHIP_STAGES)[number]), + ); + if ( + authEvents.length !== 1 || + authEvents[0]?.generation !== params.generation || + authEvents[0]?.waited !== params.waited || + authEvents[0]?.modelGenerationMatches !== true || + authEvents[0]?.sourceAuthMatchesGeneration !== true || + authReadyEvents.length !== 1 || + authReadyEvents[0]?.generation !== params.generation + ) { + throw new Error( + `unexpected ${params.generation} auth preparation: ${JSON.stringify(authEvents)}`, + ); + } + if ( + stages.length !== OWNERSHIP_STAGES.length || + stages.some( + (event, index) => + event.event !== OWNERSHIP_STAGES[index] || event.generation !== params.generation, + ) + ) { + throw new Error(`worker inference crossed provider generations: ${JSON.stringify(stages)}`); + } + const [factory, policy, wrapper, execution] = stages; + if ( + factory?.modelGenerationMatches !== true || + policy?.modelGenerationMatches !== true || + wrapper?.modelGenerationMatches !== true || + wrapper?.streamPresent !== true || + execution?.modelGenerationMatches !== true || + execution?.authPresent !== true || + execution?.baseUrlMatchesGeneration !== true + ) { + throw new Error(`worker inference ownership facts were incomplete: ${JSON.stringify(stages)}`); + } + return stages.map((event) => `${event.event}:${event.generation}`); +} + +async function readMockRequests(baseUrl: string) { + const response = await fetch(`${baseUrl}/debug/requests?after=0`); + if (!response.ok) { + throw new Error(`mock request inspection failed: HTTP ${response.status}`); + } + return (await response.json()) as Array<{ + model?: string; + outcome?: string; + allInputText?: string; + }>; +} + +async function runProof(options: ProducerOptions) { + if (workerGenerationProviderFixture.id !== PLUGIN_ID) { + throw new Error("worker generation fixture id does not match its configured plugin id"); + } + // openclaw-temp-dir: standalone QA producer owns and removes this fixture root. + const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-worker-generation-")); + const tracePath = path.join(options.artifactBase, `${SCENARIO_ID}-trace.jsonl`); + const barrierPath = path.join(root, "auth-preparation-barrier"); + const pluginDir = path.join( + options.repoRoot, + "test/e2e/qa-lab/runtime/fixtures/worker-inference-generation-provider", + ); + const mock = await startQaMockOpenAiServer({ modelRefs: [MODEL_REF] }); + const authProxy = await startAuthInspectingProxy(mock.baseUrl); + let gateway: WireGateway | undefined; + let operator: GatewayClient | undefined; + let worker: PairedNodeWorkerHost | undefined; + let published: PublishedWireWorkspace | undefined; + let proofError: unknown; + let verdict: Record | undefined; + try { + await fs.mkdir(options.artifactBase, { recursive: true }); + await fs.rm(tracePath, { force: true }); + await fs.writeFile(barrierPath, "released\n", "utf8"); + published = await createPublishedWireWorkspace(root); + gateway = await startQaGatewayChild({ + repoRoot: options.repoRoot, + useRepoCli: true, + providerBaseUrl: `${authProxy.baseUrl}/v1`, + providerMode: "mock-openai", + primaryModel: MODEL_REF, + alternateModel: DEFAULT_MOCK_MODEL_REF, + transportBaseUrl: "http://127.0.0.1", + controlUiEnabled: false, + mutateConfig: (config) => + buildGenerationConfig({ + config, + generation: "A", + pluginDir, + tracePath, + barrierPath, + mockProviderBaseUrl: `${authProxy.baseUrl}/v1`, + }), + }); + await waitFor("generation A provider registration", async () => { + const registrations = (await readTrace(tracePath)).filter( + (event) => event.event === "registered" && event.generation === "A", + ); + return registrations.length > 0 ? registrations : undefined; + }); + operator = await connectWireClient({ gateway, role: "operator", identity: null }); + worker = await createPairedNodeWorkerHost({ gateway, operator, root, bundlePrewarm: true }); + await operator.request("sessions.create", { + key: SESSION_KEY, + agentId: "qa", + worktree: true, + worktreeName: `worker-generation-${randomUUID().slice(0, 8)}`, + worktreeBaseRef: "main", + cwd: published.source, + }); + const dispatched = (await gateway.call( + "sessions.dispatch", + { key: SESSION_KEY, deviceId: worker.identity.deviceId }, + { timeoutMs: PROOF_TIMEOUT_MS }, + )) as { placement?: { state?: string; environmentId?: string; generation?: number } }; + if (dispatched.placement?.state !== "active") { + throw new Error(`worker session did not dispatch: ${JSON.stringify(dispatched)}`); + } + + const traceCursorA = (await readTrace(tracePath)).length; + const runA = await startTurn(operator, GENERATION_A_REPLY); + await waitForTurn(operator, runA); + await waitForHistoryReply(operator, GENERATION_A_REPLY); + const eventsA = (await readTrace(tracePath)).slice(traceCursorA); + const stagesA = requireGenerationOwnership({ events: eventsA, generation: "A", waited: false }); + + const hotPublishB = await hotPublishGeneration({ gateway, tracePath, generation: "B" }); + await fs.writeFile(barrierPath, "armed\n", "utf8"); + const traceCursorB = (await readTrace(tracePath)).length; + const runB = await startTurn(operator, GENERATION_B_REPLY); + await waitFor("generation B auth preparation barrier", async () => { + const events = (await readTrace(tracePath)).slice(traceCursorB); + return events.some((event) => event.event === "auth-prepare" && event.waited === true) + ? events + : undefined; + }); + const hotPublishC = await hotPublishGeneration({ gateway, tracePath, generation: "C" }); + await fs.writeFile(barrierPath, "released\n", "utf8"); + await waitForTurn(operator, runB); + await waitForHistoryReply(operator, GENERATION_B_REPLY); + const eventsB = (await readTrace(tracePath)).slice(traceCursorB); + const stagesB = requireGenerationOwnership({ events: eventsB, generation: "B", waited: true }); + if ( + !eventsB.some((event) => event.event === "auth-prepare-released" && event.generation === "B") + ) { + throw new Error("generation B auth preparation did not resume after generation C published"); + } + + const traceCursorC = (await readTrace(tracePath)).length; + const runC = await startTurn(operator, GENERATION_C_REPLY); + await waitForTurn(operator, runC); + await waitForHistoryReply(operator, GENERATION_C_REPLY); + const eventsC = (await readTrace(tracePath)).slice(traceCursorC); + const stagesC = requireGenerationOwnership({ events: eventsC, generation: "C", waited: false }); + + const requests = await readMockRequests(mock.baseUrl); + const requestFacts = requests.map((request) => ({ + model: request.model, + outcome: request.outcome, + generationA: String(request.allInputText ?? "").includes(GENERATION_A_REPLY), + generationB: String(request.allInputText ?? "").includes(GENERATION_B_REPLY), + generationC: String(request.allInputText ?? "").includes(GENERATION_C_REPLY), + })); + if ( + requests.length !== 3 || + requests.some((request) => request.outcome !== "success") || + requestFacts[0]?.generationA !== true || + requestFacts[0]?.generationB !== false || + requestFacts[0]?.generationC !== false || + requestFacts[1]?.generationA !== true || + requestFacts[1]?.generationB !== true || + requestFacts[1]?.generationC !== false || + requestFacts[2]?.generationA !== true || + requestFacts[2]?.generationB !== true || + requestFacts[2]?.generationC !== true + ) { + throw new Error(`unexpected mock-openai worker requests: ${JSON.stringify(requestFacts)}`); + } + if (JSON.stringify(authProxy.authGenerations) !== JSON.stringify(["A", "B", "C"])) { + throw new Error( + `worker inference used unexpected runtime credential generations: ${JSON.stringify(authProxy.authGenerations)}`, + ); + } + const replyCounts = await readHistoryReplyCounts(operator); + if ( + [GENERATION_A_REPLY, GENERATION_B_REPLY, GENERATION_C_REPLY].some( + (reply) => replyCounts[reply] !== 1, + ) + ) { + throw new Error(`worker history reply counts were not exact: ${JSON.stringify(replyCounts)}`); + } + + verdict = { + status: "pass", + providerMode: "mock-openai", + sessionKey: SESSION_KEY, + placement: dispatched.placement, + hotPublishes: { generationB: hotPublishB, generationC: hotPublishC }, + generationA: { reply: GENERATION_A_REPLY, stages: stagesA }, + generationB: { reply: GENERATION_B_REPLY, stages: stagesB }, + generationC: { reply: GENERATION_C_REPLY, stages: stagesC }, + runtimeCredentialGenerations: authProxy.authGenerations, + replyCounts, + requestFacts, + tracePath, + }; + await fs.writeFile( + path.join(options.artifactBase, VERDICT_FILE), + `${JSON.stringify(verdict, null, 2)}\n`, + "utf8", + ); + } catch (error) { + proofError = error; + } + + const cleanup = await Promise.allSettled([ + operator?.stopAndWait({ timeoutMs: 1_000 }) ?? Promise.resolve(), + worker?.stop() ?? Promise.resolve(), + gateway?.stop() ?? Promise.resolve(), + published ? closeWireServer(published.server) : Promise.resolve(), + authProxy.stop(), + mock.stop(), + fs.rm(root, { recursive: true, force: true }), + ]); + const cleanupFailures = cleanup.flatMap((result) => + result.status === "rejected" ? [result.reason] : [], + ); + if (cleanupFailures.length > 0) { + proofError = new AggregateError( + proofError ? [proofError, ...cleanupFailures] : cleanupFailures, + "worker inference generation proof cleanup failed", + proofError ? { cause: proofError } : undefined, + ); + } + if (proofError) { + throw proofError; + } + if (!verdict) { + throw new Error("worker inference generation proof produced no verdict"); + } + return verdict; +} + +async function runProducer(options: ProducerOptions): Promise { + const writer = createQaScriptEvidenceWriter({ + artifactBase: options.artifactBase, + logFileName: `${SCENARIO_ID}.log`, + primaryModel: MODEL_REF, + providerMode: "mock-openai", + repoRoot: options.repoRoot, + target: { + id: SCENARIO_ID, + title: "Worker inference generation reload", + sourcePath: `qa/scenarios/runtime/${SCENARIO_ID}.yaml`, + docsRefs: ["docs/gateway/cloud-workers.md", "docs/concepts/qa-e2e-automation.md"], + codeRefs: [ + "src/gateway/worker-environments/inference-runtime.ts", + "src/agents/simple-completion-runtime.ts", + "test/e2e/qa-lab/runtime/worker-inference-generation-reload.ts", + ], + }, + }); + const startedAt = Date.now(); + try { + const verdict = await runProof(options); + writer.appendLog(`pass: ${JSON.stringify(verdict)}\n`); + return await writer.write({ + artifacts: [ + { filePath: VERDICT_FILE, kind: "verdict" }, + { filePath: `${SCENARIO_ID}-trace.jsonl`, kind: "trace" }, + ], + details: + "A baseline worker turn established the placement lifecycle; generation B then retained factory, policy, wrapper, and execution ownership while C hot-published, and the next turn used C", + durationMs: Math.max(1, Date.now() - startedAt), + status: "pass", + }); + } catch (error) { + const details = error instanceof Error ? error.message : String(error); + writer.appendLog(`fail: ${details}\n`); + return await writer.write({ + details, + durationMs: Math.max(1, Date.now() - startedAt), + status: "fail", + }); + } +} + +async function main(argv: readonly string[]) { + const options = parseOptions(argv); + const evidence = await runProducer(options); + const status = evidence.entries[0]?.result.status; + console.log(`Worker inference generation evidence: ${QA_EVIDENCE_FILENAME}`); + console.log( + `Worker inference generation verdict: ${path.join(options.artifactBase, VERDICT_FILE)}`, + ); + if (status === "pass") { + console.log((await fs.readFile(path.join(options.artifactBase, VERDICT_FILE), "utf8")).trim()); + } + return status === "pass" ? 0 : 1; +} + +if (import.meta.url === pathToFileURL(process.argv[1] ?? "").href) { + main(process.argv.slice(2)) + .then((exitCode) => { + process.exitCode = exitCode; + }) + .catch((error: unknown) => { + console.error(error instanceof Error ? error.message : String(error)); + process.exitCode = 1; + }); +}