Download src/proxy-capture/runtime.lifecycle.test.ts from SaylorTwift/openclaw: direct link, hf CLI and curl.
- Browser
- Download file 29.2 kB
-
https://huggingface.co/SaylorTwift/openclaw/resolve/main/src/proxy-capture/runtime.lifecycle.test.ts
- Command line
-
hf download hf://SaylorTwift/openclaw/src/proxy-capture/runtime.lifecycle.test.ts
-
curl -L -o runtime.lifecycle.test.ts https://huggingface.co/SaylorTwift/openclaw/resolve/main/src/proxy-capture/runtime.lifecycle.test.ts
29.2 kB
| import { spawnSync } from "node:child_process"; | |
| import fs from "node:fs"; | |
| import os from "node:os"; | |
| import path from "node:path"; | |
| import { afterEach, describe, expect, it, vi } from "vitest"; | |
| import { fetchWithSsrFGuard } from "../infra/net/fetch-guard.js"; | |
| import { registerSecretValueForRedaction } from "../logging/secret-redaction-registry.js"; | |
| import { resetSecretRedactionRegistryForTest } from "../logging/secret-redaction-registry.test-support.js"; | |
| import { createDeferredCore } from "../shared/deferred.js"; | |
| import { closeOpenClawStateDatabaseByPath } from "../state/openclaw-state-db-cache.js"; | |
| import { resolveDebugProxySettings, type DebugProxySettings } from "./env.js"; | |
| import { | |
| captureHttpExchange, | |
| captureWsEvent, | |
| finalizeDebugProxyCapture, | |
| initializeDebugProxyCapture, | |
| prepareHttpCapture, | |
| type DebugProxyCaptureRuntimeDeps, | |
| } from "./runtime.js"; | |
| import { | |
| acquireDebugProxyCaptureStore, | |
| closeDebugProxyCaptureStore, | |
| DebugProxyCaptureStore, | |
| persistEventPayload, | |
| } from "./store.sqlite.js"; | |
| import type { CaptureEventRecord } from "./types.js"; | |
| const roots: string[] = []; | |
| afterEach(() => { | |
| closeDebugProxyCaptureStore(); | |
| vi.restoreAllMocks(); | |
| vi.unstubAllEnvs(); | |
| resetSecretRedactionRegistryForTest(); | |
| for (const root of roots.splice(0)) { | |
| fs.rmSync(root, { recursive: true, force: true }); | |
| } | |
| }); | |
| function stateRoot() { | |
| const root = fs.mkdtempSync(path.join(os.tmpdir(), "capture-lifecycle-")); | |
| roots.push(root); | |
| return root; | |
| } | |
| function captureSettings(root: string, sessionId = "lifecycle"): DebugProxySettings { | |
| return { | |
| enabled: true, | |
| required: false, | |
| dbPath: path.join(root, "capture.sqlite"), | |
| blobDir: path.join(root, "blobs"), | |
| certDir: path.join(root, "certs"), | |
| sessionId, | |
| sourceProcess: "fixture", | |
| }; | |
| } | |
| function observePendingCapture(response: Response, prefixChunks: number) { | |
| const pending = createDeferredCore(); | |
| const settled = createDeferredCore(); | |
| const clone = response.clone.bind(response); | |
| vi.spyOn(response, "clone").mockImplementation(() => { | |
| const captured = clone(); | |
| const reader = captured.body!.getReader(); | |
| const read = reader.read.bind(reader); | |
| let reads = 0; | |
| vi.spyOn(reader, "read").mockImplementation(() => { | |
| reads += 1; | |
| const last = reads === prefixChunks + 1; | |
| if (last) { | |
| pending.resolve(); | |
| } | |
| return read().finally(() => { | |
| if (last) { | |
| settled.resolve(); | |
| } | |
| }); | |
| }); | |
| vi.spyOn(captured.body!, "getReader").mockReturnValue(reader); | |
| return captured; | |
| }); | |
| return { pending: pending.promise, settled: settled.promise }; | |
| } | |
| function pendingResponse(chunks: Buffer[]) { | |
| let controller: ReadableStreamDefaultController<Uint8Array>; | |
| const response = new Response( | |
| new ReadableStream<Uint8Array>({ | |
| start(value) { | |
| controller = value; | |
| for (const chunk of chunks) { | |
| controller.enqueue(chunk); | |
| } | |
| }, | |
| }), | |
| { status: 200, headers: { "content-type": "text/plain" } }, | |
| ); | |
| return { response, controller: controller!, ...observePendingCapture(response, chunks.length) }; | |
| } | |
| describe("capture store lifecycle", () => { | |
| it.each( | |
| (["shared", "legacy"] as const).flatMap((storage) => | |
| (["direct", "last-lease"] as const).map((close) => ({ storage, close })), | |
| ), | |
| )("settles bytes once before $storage $close close", async ({ storage, close }) => { | |
| const root = stateRoot(); | |
| const settings = captureSettings(root); | |
| const acquire = () => | |
| storage === "shared" | |
| ? acquireDebugProxyCaptureStore({ env: { OPENCLAW_STATE_DIR: root } }) | |
| : acquireDebugProxyCaptureStore(settings.dbPath, settings.blobDir); | |
| const lease = acquire(); | |
| const sibling = acquire(); | |
| const store = lease.store; | |
| const getStore = vi.fn(() => store); | |
| const deps: DebugProxyCaptureRuntimeDeps = { | |
| getStore, | |
| persistEventPayload: (_store, payload) => persistEventPayload(store, payload), | |
| }; | |
| registerSecretValueForRedaction("fixture-secret-value"); | |
| const bytes = Buffer.from("first fixture-secret-value caf\u00e9"); | |
| const stream = pendingResponse([ | |
| bytes.subarray(0, 15), | |
| bytes.subarray(15, -1), | |
| bytes.subarray(-1), | |
| ]); | |
| const terminals: CaptureEventRecord[] = []; | |
| const record = store.recordEvent.bind(store); | |
| const recording = vi.spyOn(store, "recordEvent").mockImplementation((event) => { | |
| expect(store.isClosed).toBe(false); | |
| record(event); | |
| if (event.kind !== "request") { | |
| terminals.push(event); | |
| } | |
| }); | |
| const end = vi.spyOn(store, "endSession"); | |
| try { | |
| store.upsertSession({ | |
| id: settings.sessionId, | |
| startedAt: Date.now(), | |
| mode: "fixture", | |
| sourceScope: "openclaw", | |
| sourceProcess: "fixture", | |
| }); | |
| captureHttpExchange( | |
| { url: "https://example.test/pending", method: "GET", response: stream.response }, | |
| settings, | |
| deps, | |
| ); | |
| await stream.pending; | |
| lease.release(); | |
| expect(store.isClosed).toBe(false); | |
| expect(terminals).toHaveLength(0); | |
| if (close === "direct") { | |
| store.close(); | |
| } else { | |
| sibling.release(); | |
| } | |
| expect(store.isClosed).toBe(true); | |
| expect(terminals).toHaveLength(1); | |
| expect(terminals[0]).toMatchObject({ | |
| kind: "response", | |
| status: 200, | |
| dataText: "first [REDACTED] caf\u00e9", | |
| }); | |
| expect(JSON.parse(terminals[0]!.metaJson!)).toMatchObject({ bodyCapture: "finalized" }); | |
| expect(end).toHaveBeenCalledExactlyOnceWith(settings.sessionId); | |
| stream.controller.enqueue(Buffer.from("-later")); | |
| stream.controller.close(); | |
| expect(await stream.response.text()).toBe(`${bytes.toString()}-later`); | |
| await stream.settled; | |
| finalizeDebugProxyCapture(settings, deps); | |
| store.close(); | |
| sibling.release(); | |
| expect(terminals).toHaveLength(1); | |
| expect(end).toHaveBeenCalledTimes(1); | |
| expect(getStore).toHaveBeenCalledTimes(1); | |
| const reopened = acquire(); | |
| try { | |
| expect(reopened.store.listSessions()[0]?.endedAt).toBeTypeOf("number"); | |
| expect(reopened.store.getSessionEvents(settings.sessionId)).toHaveLength(2); | |
| expect(reopened.store.readBlob(terminals[0]!.dataBlobId!)).toBe( | |
| "first [REDACTED] caf\u00e9", | |
| ); | |
| } finally { | |
| reopened.release(); | |
| } | |
| } finally { | |
| recording.mockRestore(); | |
| lease.release(); | |
| sibling.release(); | |
| store.close(); | |
| closeOpenClawStateDatabaseByPath(store.dbPath); | |
| } | |
| }); | |
| it.each(["eof", "error", "finalize"] as const)( | |
| "records one terminal through a pending-read %s race", | |
| async (mode) => { | |
| const root = stateRoot(); | |
| const settings = captureSettings(root); | |
| const store = new DebugProxyCaptureStore({ env: { OPENCLAW_STATE_DIR: root } }); | |
| const deps: DebugProxyCaptureRuntimeDeps = { getStore: () => store }; | |
| const stream = pendingResponse([Buffer.from("prefix")]); | |
| const terminals: CaptureEventRecord[] = []; | |
| const done = createDeferredCore(); | |
| const record = store.recordEvent.bind(store); | |
| vi.spyOn(store, "recordEvent").mockImplementation((event) => { | |
| record(event); | |
| if (event.kind !== "request") { | |
| terminals.push(event); | |
| done.resolve(); | |
| } | |
| }); | |
| registerSecretValueForRedaction("fixture-error-secret"); | |
| const reason = new Error("fixture read failure: fixture-error-secret"); | |
| try { | |
| captureHttpExchange( | |
| { url: "https://example.test/race", method: "GET", response: stream.response }, | |
| settings, | |
| deps, | |
| ); | |
| await stream.pending; | |
| if (mode === "eof") { | |
| stream.controller.close(); | |
| } else if (mode === "error") { | |
| stream.controller.error(reason); | |
| } else { | |
| finalizeDebugProxyCapture(settings, deps); | |
| stream.controller.error(reason); | |
| } | |
| await done.promise; | |
| await stream.settled; | |
| expect(terminals).toHaveLength(1); | |
| expect(terminals[0]).toMatchObject({ | |
| kind: mode === "error" ? "error" : "response", | |
| status: 200, | |
| dataText: "prefix", | |
| }); | |
| if (mode === "error") { | |
| expect(terminals[0]).toMatchObject({ errorText: "fixture read failure: [REDACTED]" }); | |
| expect(JSON.parse(terminals[0]!.metaJson!)).toMatchObject({ | |
| bodyCapture: "failed", | |
| stage: "response-body", | |
| }); | |
| } | |
| finalizeDebugProxyCapture(settings, deps); | |
| expect(terminals).toHaveLength(1); | |
| } finally { | |
| await stream.response.body?.cancel().catch(() => undefined); | |
| store.close(); | |
| closeOpenClawStateDatabaseByPath(store.dbPath); | |
| } | |
| }, | |
| ); | |
| it.each(["blob", "event"] as const)( | |
| "settles sibling reads and closes after one terminal %s persistence fails", | |
| async (surface) => { | |
| const root = stateRoot(); | |
| const settings = captureSettings(root); | |
| const store = new DebugProxyCaptureStore({ env: { OPENCLAW_STATE_DIR: root } }); | |
| const deps: DebugProxyCaptureRuntimeDeps = { getStore: () => store }; | |
| const streams = [ | |
| pendingResponse([Buffer.from("one")]), | |
| pendingResponse([Buffer.from("two")]), | |
| ]; | |
| const failure = new Error("fixture storage failure"); | |
| const attempted: string[] = []; | |
| if (surface === "blob") { | |
| const persist = store.persistPayload.bind(store); | |
| vi.spyOn(store, "persistPayload").mockImplementation((data, contentType) => { | |
| if (data.toString() === "one") { | |
| throw failure; | |
| } | |
| return persist(data, contentType); | |
| }); | |
| } | |
| const record = store.recordEvent.bind(store); | |
| vi.spyOn(store, "recordEvent").mockImplementation((event) => { | |
| if (event.kind !== "request") { | |
| attempted.push(event.flowId); | |
| if (surface === "event" && event.flowId === "one") { | |
| throw failure; | |
| } | |
| } | |
| record(event); | |
| }); | |
| try { | |
| for (const [index, stream] of streams.entries()) { | |
| captureHttpExchange( | |
| { | |
| url: "https://example.test/sibling", | |
| method: "GET", | |
| response: stream.response, | |
| flowId: index === 0 ? "one" : "two", | |
| }, | |
| settings, | |
| deps, | |
| ); | |
| } | |
| await Promise.all(streams.map((stream) => stream.pending)); | |
| let error: unknown; | |
| try { | |
| finalizeDebugProxyCapture(settings, deps); | |
| } catch (caught) { | |
| error = caught; | |
| } | |
| expect(error).toBeInstanceOf(AggregateError); | |
| expect((error as AggregateError).errors[0].errors).toContain(failure); | |
| expect(store.isClosed).toBe(true); | |
| expect(attempted).toEqual(surface === "blob" ? ["two"] : ["one", "two"]); | |
| finalizeDebugProxyCapture(settings, deps); | |
| expect(attempted).toEqual(surface === "blob" ? ["two"] : ["one", "two"]); | |
| } finally { | |
| for (const stream of streams) { | |
| stream.controller.close(); | |
| await stream.response.body?.cancel(); | |
| await stream.settled; | |
| } | |
| store.close(); | |
| closeOpenClawStateDatabaseByPath(store.dbPath); | |
| } | |
| }, | |
| ); | |
| it("does not revive a retired database while finalizing its retained owner", async () => { | |
| const root = stateRoot(); | |
| const settings = captureSettings(root); | |
| const store = new DebugProxyCaptureStore({ env: { OPENCLAW_STATE_DIR: root } }); | |
| const getStore = vi.fn(() => store); | |
| const deps: DebugProxyCaptureRuntimeDeps = { getStore }; | |
| const stream = pendingResponse([Buffer.from("prefix")]); | |
| try { | |
| captureHttpExchange( | |
| { url: "https://example.test/retired", method: "GET", response: stream.response }, | |
| settings, | |
| deps, | |
| ); | |
| await stream.pending; | |
| closeOpenClawStateDatabaseByPath(store.dbPath); | |
| expect(store.isClosed).toBe(true); | |
| expect(() => finalizeDebugProxyCapture(settings, deps)).toThrow(AggregateError); | |
| expect(() => finalizeDebugProxyCapture(settings, deps)).not.toThrow(); | |
| expect(getStore).toHaveBeenCalledTimes(1); | |
| expect(store.db.isOpen).toBe(false); | |
| } finally { | |
| stream.controller.close(); | |
| await stream.response.body?.cancel(); | |
| await stream.settled; | |
| store.close(); | |
| } | |
| }); | |
| it.each( | |
| (["shared", "legacy"] as const).flatMap((storage) => | |
| (["none", "blob", "event"] as const).map((failure) => ({ storage, failure })), | |
| ), | |
| )( | |
| "ends the $storage session and reports $failure failure before CLI exit finalization", | |
| ({ storage, failure }) => { | |
| const root = stateRoot(); | |
| const settings = captureSettings(root); | |
| const runtimeUrl = new URL("./runtime.ts", import.meta.url).href; | |
| const storeUrl = new URL("./store.sqlite.ts", import.meta.url).href; | |
| const redactionUrl = new URL("../logging/secret-redaction-registry.ts", import.meta.url).href; | |
| const script = ` | |
| import assert from "node:assert/strict"; | |
| import { captureHttpExchange, initializeDebugProxyCapture, finalizeDebugProxyCapture } from ${JSON.stringify(runtimeUrl)}; | |
| import { getDebugProxyCaptureStore } from ${JSON.stringify(storeUrl)}; | |
| import { registerSecretValueForRedaction } from ${JSON.stringify(redactionUrl)}; | |
| const settings = ${JSON.stringify(settings)}; | |
| const store = ${storage === "shared" ? "getDebugProxyCaptureStore()" : "getDebugProxyCaptureStore(settings.dbPath, settings.blobDir)"}; | |
| const failure = ${JSON.stringify(failure)}; | |
| registerSecretValueForRedaction("fixture-storage-secret"); | |
| const error = new Error("fixture " + failure + " persistence failure: fixture-storage-secret"); | |
| let acquired = 0, terminals = 0, ended = 0, failures = 0; | |
| const persist = store.persistPayload.bind(store); | |
| store.persistPayload = (data, contentType) => { | |
| if (failure === "blob" && data?.toString() === "prefix-one") { | |
| failures++; | |
| throw error; | |
| } | |
| return persist(data, contentType); | |
| }; | |
| const record = store.recordEvent.bind(store); | |
| store.recordEvent = event => { | |
| assert.equal(store.isClosed, false); | |
| if (failure === "event" && event.kind !== "request" && event.flowId === "one") { | |
| failures++; | |
| throw error; | |
| } | |
| record(event); | |
| if (event.kind !== "request") terminals++; | |
| }; | |
| const end = store.endSession.bind(store); | |
| store.endSession = id => { ended++; end(id); }; | |
| const deps = { getStore: () => { acquired++; return store; } }; | |
| initializeDebugProxyCapture("fixture", settings, deps); | |
| for (const flowId of ["one", "two"]) { | |
| const response = new Response(new ReadableStream({ | |
| start(controller) { controller.enqueue(new TextEncoder().encode("prefix-" + flowId)); } | |
| })); | |
| const clone = response.clone.bind(response); | |
| let pending, reads = 0; | |
| const ready = new Promise(resolve => pending = resolve); | |
| response.clone = () => { | |
| const cloned = clone(); | |
| const reader = cloned.body.getReader(); | |
| const read = reader.read.bind(reader); | |
| reader.read = () => { if (++reads === 2) pending(); return read(); }; | |
| cloned.body.getReader = () => reader; | |
| return cloned; | |
| }; | |
| captureHttpExchange({ url: "https://example.test/exit", method: "GET", response, flowId }, settings, deps); | |
| await ready; | |
| } | |
| process.once("exit", () => { | |
| assert.equal(store.isClosed, true); | |
| finalizeDebugProxyCapture(settings, deps); | |
| finalizeDebugProxyCapture(settings, deps); | |
| assert.equal(acquired, 1); | |
| assert.equal(terminals, failure === "none" ? 2 : 1); | |
| assert.equal(ended, 1); | |
| assert.equal(failures, failure === "none" ? 0 : 1); | |
| process.stdout.write(JSON.stringify({ acquired, terminals, ended, failures })); | |
| }); | |
| process.exit(0); | |
| `; | |
| const child = spawnSync( | |
| process.execPath, | |
| [ | |
| "--disable-warning=ExperimentalWarning", | |
| "--import", | |
| "tsx", | |
| "--input-type=module", | |
| "-e", | |
| script, | |
| ], | |
| { | |
| cwd: process.cwd(), | |
| env: { ...process.env, OPENCLAW_STATE_DIR: root }, | |
| encoding: "utf8", | |
| timeout: 20_000, | |
| }, | |
| ); | |
| expect(child.stderr).toBe( | |
| failure === "none" | |
| ? "" | |
| : `[proxy-capture] Capture persistence failed: fixture ${failure} persistence failure: [REDACTED]\n`, | |
| ); | |
| expect(child.status).toBe(0); | |
| expect(JSON.parse(child.stdout)).toEqual({ | |
| acquired: 1, | |
| terminals: failure === "none" ? 2 : 1, | |
| ended: 1, | |
| failures: failure === "none" ? 0 : 1, | |
| }); | |
| const reopened = | |
| storage === "shared" | |
| ? new DebugProxyCaptureStore({ env: { OPENCLAW_STATE_DIR: root } }) | |
| : new DebugProxyCaptureStore(settings.dbPath, settings.blobDir); | |
| try { | |
| expect(reopened.listSessions()[0]?.endedAt).toBeTypeOf("number"); | |
| const events = reopened.getSessionEvents(settings.sessionId); | |
| expect(events).toHaveLength(failure === "none" ? 4 : 3); | |
| expect(events[0]).toMatchObject({ kind: "response", dataText: "prefix-two" }); | |
| } finally { | |
| reopened.close(); | |
| closeOpenClawStateDatabaseByPath(reopened.dbPath); | |
| } | |
| }, | |
| ); | |
| }); | |
| describe("capture admission generation", () => { | |
| it.each([ | |
| [0, 1], | |
| [1, 0], | |
| ] as const)( | |
| "isolates same-session databases when finalizing %s before %s", | |
| async (first, second) => { | |
| const envs = [stateRoot(), stateRoot()].map((root) => ({ | |
| OPENCLAW_STATE_DIR: root, | |
| OPENCLAW_DEBUG_PROXY_ENABLED: "1", | |
| })); | |
| const settings = envs.map((env) => resolveDebugProxySettings(env)); | |
| expect(settings[0]!.sessionId).toBe(settings[1]!.sessionId); | |
| expect(settings[0]!.dbPath).not.toBe(settings[1]!.dbPath); | |
| const stores = envs.map((env) => new DebugProxyCaptureStore({ env })); | |
| let selected = stores[0]!; | |
| const getStore = vi.fn(() => selected); | |
| const deps = { getStore }; | |
| const recordings = stores.map((store) => vi.spyOn(store, "recordEvent")); | |
| const streams = stores.map((_, index) => pendingResponse([Buffer.from(`database-${index}`)])); | |
| const admissions = settings.map((value, index) => { | |
| selected = stores[index]!; | |
| return prepareHttpCapture(value, deps)!; | |
| }); | |
| try { | |
| for (const [index, capture] of admissions.entries()) { | |
| capture({ | |
| url: "https://example.test/identity", | |
| method: "POST", | |
| flowId: `database-${index}`, | |
| requestBody: `request-${index}`, | |
| response: streams[index]!.response, | |
| }); | |
| } | |
| await Promise.all(streams.map((stream) => stream.pending)); | |
| expect( | |
| stores.map((store, index) => | |
| store.getSessionEvents(settings[index]!.sessionId).map((event) => event.flowId), | |
| ), | |
| ).toEqual([["database-0"], ["database-1"]]); | |
| expect(getStore).toHaveBeenCalledTimes(2); | |
| getStore.mockImplementation(() => { | |
| throw new Error("Existing-owner lookup or finalization acquired a store."); | |
| }); | |
| for (const index of [first, second]) { | |
| const fresh = resolveDebugProxySettings(envs[index]); | |
| captureWsEvent( | |
| { | |
| url: "wss://example.test/identity", | |
| direction: "inbound", | |
| kind: "ws-frame", | |
| flowId: `database-${index}`, | |
| payload: "before-close", | |
| }, | |
| fresh, | |
| deps, | |
| ); | |
| expect(recordings[index]!.mock.lastCall?.[0].kind).toBe("ws-frame"); | |
| finalizeDebugProxyCapture(fresh, deps); | |
| expect(stores[index]!.isClosed).toBe(true); | |
| const terminals = recordings[index]!.mock.calls.map(([event]) => event).filter( | |
| (event) => event.kind === "response", | |
| ); | |
| expect(terminals).toHaveLength(1); | |
| expect(terminals[0]).toMatchObject({ | |
| flowId: `database-${index}`, | |
| dataText: `database-${index}`, | |
| }); | |
| const recorded = recordings[index]!.mock.calls.length; | |
| admissions[index]!({ | |
| url: "https://example.test/delayed", | |
| method: "GET", | |
| response: new Response("late"), | |
| }); | |
| expect(recordings[index]).toHaveBeenCalledTimes(recorded); | |
| if (index === first) { | |
| expect(stores[second]!.isClosed).toBe(false); | |
| expect( | |
| recordings[second]!.mock.calls.some(([event]) => event.kind === "response"), | |
| ).toBe(false); | |
| } | |
| finalizeDebugProxyCapture(resolveDebugProxySettings(envs[index]), deps); | |
| } | |
| expect(getStore).toHaveBeenCalledTimes(2); | |
| } finally { | |
| for (const [index, stream] of streams.entries()) { | |
| finalizeDebugProxyCapture(settings[index], deps); | |
| stream.controller.close(); | |
| await stream.response.body?.cancel(); | |
| await stream.settled; | |
| stores[index]!.close(); | |
| closeOpenClawStateDatabaseByPath(stores[index]!.dbPath); | |
| } | |
| } | |
| }, | |
| ); | |
| it.each(["explicit", "ambient"] as const)( | |
| "preserves fresh %s lazy sessions without reopening retired admission", | |
| (mode) => { | |
| const root = stateRoot(); | |
| let settings = captureSettings(root, "first"); | |
| vi.stubEnv("OPENCLAW_STATE_DIR", root); | |
| vi.stubEnv("OPENCLAW_DEBUG_PROXY_ENABLED", "1"); | |
| vi.stubEnv("OPENCLAW_DEBUG_PROXY_SESSION_ID", settings.sessionId); | |
| const record = vi.fn(); | |
| const getStore = vi.fn(() => ({ | |
| upsertSession() {}, | |
| endSession() {}, | |
| recordEvent: record, | |
| close() {}, | |
| })); | |
| const deps = { | |
| getStore, | |
| persistEventPayload: () => ({}), | |
| fetchTarget: { fetch: vi.fn() } as unknown as typeof globalThis, | |
| }; | |
| const resolved = () => (mode === "explicit" ? settings : undefined); | |
| const frame = { | |
| url: "wss://example.test/stream", | |
| direction: "inbound" as const, | |
| kind: "ws-frame" as const, | |
| flowId: "fixture", | |
| payload: "frame", | |
| }; | |
| const delayed = prepareHttpCapture(resolved(), deps)!; | |
| captureWsEvent(frame, resolved(), deps); | |
| finalizeDebugProxyCapture(resolved(), deps); | |
| captureWsEvent(frame, resolved(), deps); | |
| expect(getStore).toHaveBeenCalledTimes(1); | |
| expect(record).toHaveBeenCalledTimes(1); | |
| settings = captureSettings(root, "second"); | |
| vi.stubEnv("OPENCLAW_DEBUG_PROXY_SESSION_ID", settings.sessionId); | |
| captureWsEvent(frame, resolved(), deps); | |
| expect(getStore).toHaveBeenCalledTimes(2); | |
| expect(record).toHaveBeenCalledTimes(2); | |
| delayed({ | |
| url: "https://example.test/delayed", | |
| method: "GET", | |
| response: new Response("late"), | |
| }); | |
| expect(record).toHaveBeenCalledTimes(2); | |
| finalizeDebugProxyCapture(resolved(), deps); | |
| initializeDebugProxyCapture("replacement", resolved(), deps); | |
| captureWsEvent(frame, resolved(), deps); | |
| expect(getStore).toHaveBeenCalledTimes(3); | |
| expect(record).toHaveBeenCalledTimes(3); | |
| finalizeDebugProxyCapture(resolved(), deps); | |
| }, | |
| ); | |
| it("releases retired settings, store, and runtime closures while delayed admission stays fenced", () => { | |
| const root = stateRoot(); | |
| const runtimeUrl = new URL("./runtime.ts", import.meta.url).href; | |
| const script = ` | |
| import assert from "node:assert/strict"; | |
| import { initializeDebugProxyCapture, finalizeDebugProxyCapture, prepareHttpCapture } from ${JSON.stringify(runtimeUrl)}; | |
| let store, acquired = 0, recorded = 0, complete; | |
| const getStore = () => { acquired++; return store; }; | |
| const target = { fetch: () => new Promise(resolve => complete = resolve) }; | |
| const control = new WeakRef({}); | |
| function retire() { | |
| const settings = ${JSON.stringify(captureSettings(root))}; | |
| store = { upsertSession() {}, endSession() {}, recordEvent() { recorded++; }, close() {} }; | |
| const payload = { retained: Buffer.alloc(1024) }; | |
| const persist = () => { assert.ok(payload.retained); return {}; }; | |
| const refs = [new WeakRef(settings), new WeakRef(store), new WeakRef(persist)]; | |
| const deps = { getStore, persistEventPayload: persist, fetchTarget: target }; | |
| initializeDebugProxyCapture("fixture", settings, deps); | |
| const capture = prepareHttpCapture(settings, deps); | |
| const late = target.fetch("https://example.test/delayed"); | |
| finalizeDebugProxyCapture(settings, deps); | |
| store = undefined; | |
| return { refs, capture, late }; | |
| } | |
| const retired = retire(); | |
| for (let i = 0; i < 30; i++) { | |
| await new Promise(setImmediate); | |
| globalThis.gc(); | |
| } | |
| assert.equal(control.deref(), undefined, "GC control must be collected"); | |
| const released = retired.refs.map(ref => ref.deref() === undefined); | |
| const response = new Response("late"); | |
| complete(response); | |
| assert.equal(await retired.late, response); | |
| retired.capture({ url: "https://example.test/delayed", method: "GET", response }); | |
| assert.equal(acquired, 1); | |
| assert.equal(recorded, 0); | |
| process.stdout.write(JSON.stringify(released)); | |
| `; | |
| const child = spawnSync( | |
| process.execPath, | |
| [ | |
| "--disable-warning=ExperimentalWarning", | |
| "--expose-gc", | |
| "--import", | |
| "tsx", | |
| "--input-type=module", | |
| "-e", | |
| script, | |
| ], | |
| { | |
| cwd: process.cwd(), | |
| env: { ...process.env, OPENCLAW_STATE_DIR: root }, | |
| encoding: "utf8", | |
| timeout: 20_000, | |
| }, | |
| ); | |
| expect(child.stderr).toBe(""); | |
| expect(child.status).toBe(0); | |
| expect(JSON.parse(child.stdout)).toEqual([true, true, true]); | |
| }); | |
| it.each( | |
| (["patched", "guarded"] as const).flatMap((route) => | |
| (["success", "rejection"] as const).map((outcome) => ({ route, outcome })), | |
| ), | |
| )( | |
| "does not attach a delayed $route $outcome to a replacement owner", | |
| async ({ route, outcome }) => { | |
| const root = stateRoot(); | |
| const settings = captureSettings(root, `generation-${route}-${outcome}`); | |
| vi.stubEnv("OPENCLAW_STATE_DIR", root); | |
| vi.stubEnv("OPENCLAW_DEBUG_PROXY_ENABLED", "1"); | |
| vi.stubEnv("OPENCLAW_DEBUG_PROXY_SESSION_ID", settings.sessionId); | |
| const admitted = createDeferredCore(); | |
| const transport = createDeferredCore<Response>(); | |
| const fetchImpl = vi.fn(async () => { | |
| admitted.resolve(); | |
| return await transport.promise; | |
| }); | |
| const target = { ...globalThis, fetch: fetchImpl } as typeof globalThis; | |
| const deps = { fetchTarget: target }; | |
| initializeDebugProxyCapture("first", settings, deps); | |
| const firstStore = acquireDebugProxyCaptureStore(); | |
| const operation = | |
| route === "patched" | |
| ? target.fetch("https://example.test/delayed").then((response) => ({ | |
| response, | |
| release: async () => {}, | |
| })) | |
| : fetchWithSsrFGuard({ url: "https://example.test/delayed", fetchImpl }); | |
| const result = operation.then( | |
| (value) => ({ value, error: undefined }), | |
| (error: unknown) => ({ value: undefined, error }), | |
| ); | |
| await admitted.promise; | |
| finalizeDebugProxyCapture(settings, deps); | |
| expect(firstStore.store.isClosed).toBe(true); | |
| initializeDebugProxyCapture("replacement", settings, deps); | |
| const replacement = acquireDebugProxyCaptureStore(); | |
| const record = vi.spyOn(replacement.store, "recordEvent"); | |
| const response = new Response("late"); | |
| const error = new Error("fixture late transport failure"); | |
| try { | |
| if (outcome === "success") { | |
| transport.resolve(response); | |
| } else { | |
| transport.reject(error); | |
| } | |
| const completed = await result; | |
| if (outcome === "success") { | |
| expect(completed.error).toBeUndefined(); | |
| expect(completed.value?.response).toBe(response); | |
| expect(await completed.value!.response.text()).toBe("late"); | |
| await completed.value!.release(); | |
| } else { | |
| expect(completed.error).toBe(error); | |
| } | |
| expect(record).not.toHaveBeenCalled(); | |
| expect(replacement.store.getSessionEvents(settings.sessionId)).toEqual([]); | |
| } finally { | |
| finalizeDebugProxyCapture(settings, deps); | |
| firstStore.release(); | |
| replacement.release(); | |
| closeOpenClawStateDatabaseByPath(replacement.store.dbPath); | |
| } | |
| }, | |
| ); | |
| }); | |