import assert from "node:assert/strict"; import { test } from "bun:test"; import { handleJsonRpcRequest } from "../cloud/json-rpc.ts"; import { ERROR_CODES, validateResponse } from "../protocol/index.mjs"; import { RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED, RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED, RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED, RUNTIME_DURABILITY_REQUIRED_EVIDENCE, RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED, RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED, RUNTIME_STORE_KIND_POSTGRES, buildPostgresPoolConfig, createCloudRuntimeStore, createConfiguredCloudRuntimeStore } from "./runtime-store.ts"; import { CLOUD_CORE_MIGRATION_ID, CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION, CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS } from "./schema.ts"; test("memory runtime store is never marked durable", () => { const store = createCloudRuntimeStore(); const summary = store.summary(); assert.equal(summary.adapter, "memory"); assert.equal(summary.durable, false); assert.equal(summary.status, "degraded"); assert.equal(summary.ready, undefined); assert.equal(summary.adapterContract.secretMaterialRequiredInSource, false); assert.equal(summary.durabilityContract.ready, false); assert.equal(summary.durabilityContract.status, "blocked"); assert.equal(summary.durabilityContract.blockedLayer, "adapter"); assert.equal(summary.durabilityContract.adapterQueryRequired, true); assert.equal(summary.durabilityContract.dbLiveEvidenceIsDurabilityEvidence, false); assert.equal(summary.durabilityContract.secretMaterialRead, false); }); test("configured postgres runtime classifies auth blocker before schema and migration", async () => { const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient: { async query(sql) { assert.match(sql, /information_schema\.columns/u); const error = new Error("auth failed"); error.code = "28P01"; throw error; } } }); const readiness = await store.readiness(); assert.equal(readiness.adapter, RUNTIME_STORE_KIND_POSTGRES); assert.equal(readiness.durable, false); assert.equal(readiness.ready, false); assert.equal(readiness.status, "blocked"); assert.equal(readiness.blocker, RUNTIME_DURABLE_ADAPTER_AUTH_BLOCKED); assert.equal(readiness.connection.queryAttempted, true); assert.equal(readiness.connection.queryResult, "auth_blocked"); assert.equal(readiness.gates.auth.status, "blocked"); assert.equal(readiness.gates.schema.status, "not_checked"); assert.equal(readiness.gates.migration.status, "not_checked"); assert.equal(readiness.durabilityContract.ready, false); assert.equal(readiness.durabilityContract.blockedLayer, "auth"); assert.equal(readiness.durabilityContract.dbLiveEvidenceIsDurabilityEvidence, false); }); test("configured postgres runtime classifies schema driver errors separately from query failures", async () => { const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, queryClient: { async query(sql) { assert.match(sql, /information_schema\.columns/u); const error = new Error("schema missing"); error.code = "3F000"; throw error; } } }); const readiness = await store.readiness(); assert.equal(readiness.blocker, RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED); assert.equal(readiness.connection.queryAttempted, true); assert.equal(readiness.connection.queryResult, "schema_blocked"); assert.equal(readiness.gates.ssl.status, "ready"); assert.equal(readiness.gates.auth.status, "ready"); assert.equal(readiness.gates.schema.status, "blocked"); assert.equal(readiness.gates.migration.status, "not_checked"); assert.equal(readiness.durabilityContract.blockedLayer, "schema"); }); test("configured postgres runtime classifies SSL negotiation blocker separately from auth", async () => { const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, queryClient: { async query(sql) { assert.match(sql, /information_schema\.columns/u); const error = new Error("The server does not support SSL connections"); error.code = "08P01"; throw error; } } }); const readiness = await store.readiness(); assert.equal(readiness.adapter, RUNTIME_STORE_KIND_POSTGRES); assert.equal(readiness.durable, false); assert.equal(readiness.ready, false); assert.equal(readiness.status, "blocked"); assert.equal(readiness.blocker, RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED); assert.equal(readiness.connection.queryAttempted, true); assert.equal(readiness.connection.queryResult, "ssl_negotiation_blocked"); assert.equal(readiness.connection.errorCode, "08P01"); assert.equal(readiness.gates.ssl.status, "blocked"); assert.equal(readiness.gates.auth.status, "not_checked"); assert.equal(readiness.gates.schema.status, "not_checked"); assert.equal(readiness.gates.migration.status, "not_checked"); assert.equal(readiness.durabilityContract.ready, false); assert.equal(readiness.durabilityContract.blockedLayer, "ssl"); assert.equal(readiness.durabilityContract.secretMaterialRead, false); assert.equal(JSON.stringify(readiness).includes("The server does not support SSL connections"), false); }); test("postgres pool config makes HWLAB_CLOUD_DB_SSL_MODE authoritative over URL sslmode", () => { const disabled = buildPostgresPoolConfig({ dbUrl: "postgres://hwlab_user:fixture-pass@db.example.test:5432/hwlab?sslmode=require&application_name=hwlab", sslMode: "disable", timeoutMs: "2500", poolMax: "20" }); assert.equal(disabled.ssl, false); assert.equal(disabled.connectionTimeoutMillis, 2500); assert.equal(disabled.max, 20); const disabledUrl = new URL(disabled.connectionString); assert.equal(disabledUrl.searchParams.get("sslmode"), null); assert.equal(disabledUrl.searchParams.get("ssl"), null); assert.equal(disabledUrl.searchParams.get("application_name"), "hwlab"); const required = buildPostgresPoolConfig({ dbUrl: "postgres://hwlab_user:fixture-pass@db.example.test:5432/hwlab?sslmode=disable", sslMode: "require" }); assert.deepEqual(required.ssl, { rejectUnauthorized: false }); assert.equal(new URL(required.connectionString).searchParams.get("sslmode"), "require"); assert.equal(new URL(required.connectionString).searchParams.get("uselibpqcompat"), "true"); }); test("configured postgres runtime passes normalized pool config to pg", async () => { const pools = []; const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true", HWLAB_CLOUD_DB_PROBE_TIMEOUT_MS: "1800", HWLAB_CLOUD_DB_POOL_MAX: "20" }, dbUrl: "postgres://hwlab_user:fixture-pass@db.example.test:5432/hwlab?sslmode=require", sslMode: "disable", pgModuleLoader: async () => ({ Pool: class FakePool { constructor(config) { this.config = config; pools.push(this); } async query() { return { rows: [] }; } } }) }); await store.readiness(); assert.equal(pools.length, 1); assert.equal(pools[0].config.ssl, false); assert.equal(pools[0].config.connectionTimeoutMillis, 1800); assert.equal(pools[0].config.max, 20); assert.equal(new URL(pools[0].config.connectionString).searchParams.get("sslmode"), null); assert.equal(store.summary().blocker, RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED); assert.equal(store.summary().durabilityContract.blockedLayer, "schema"); }); test("configured postgres runtime consumes pg pool idle errors as degraded readiness", async () => { const pools = []; const warnings = []; const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true", HWLAB_CLOUD_DB_PROBE_TIMEOUT_MS: "1800" }, dbUrl: "postgres://hwlab_user:fixture-pass@db.example.test:5432/hwlab?sslmode=require", logger: { warn(entry) { warnings.push(entry); } }, pgModuleLoader: async () => ({ Pool: class FakePool { constructor(config) { this.config = config; this.handlers = new Map(); pools.push(this); } on(event, handler) { this.handlers.set(event, handler); return this; } async query() { return { rows: [] }; } emit(event, error) { this.handlers.get(event)?.(error); } } }) }); await store.readiness(); const error = new Error("connection timed out after fixture secret host"); error.code = "ETIMEDOUT"; pools[0].emit("error", error); const summary = store.summary(); assert.equal(summary.blocker, RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED); assert.equal(summary.connection.queryResult, "query_blocked"); assert.equal(summary.connection.errorCode, "ETIMEDOUT"); assert.equal(summary.durabilityContract.blockedLayer, "durability_query"); assert.equal(warnings.length, 1); assert.equal(warnings[0].event, "postgres_runtime_pool_error"); assert.equal(warnings[0].valuesRedacted, true); assert.equal(JSON.stringify(warnings).includes("fixture secret host"), false); }); test("configured postgres runtime classifies pg_hba no-encryption rejection as SSL before auth", async () => { const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, sslMode: "disable", queryClient: { async query(sql) { assert.match(sql, /information_schema\.columns/u); const error = new Error('no pg_hba.conf entry for host "[redacted]", user "[redacted]", database "[redacted]", no encryption'); error.code = "28000"; throw error; } } }); const readiness = await store.readiness(); assert.equal(readiness.blocker, RUNTIME_DURABLE_ADAPTER_SSL_BLOCKED); assert.equal(readiness.connection.queryAttempted, true); assert.equal(readiness.connection.queryResult, "ssl_negotiation_blocked"); assert.equal(readiness.connection.errorCode, "28000"); assert.equal(readiness.gates.ssl.status, "blocked"); assert.equal(readiness.gates.auth.status, "not_checked"); assert.equal(readiness.durabilityContract.blockedLayer, "ssl"); assert.equal(readiness.durabilityContract.dbLiveEvidenceIsDurabilityEvidence, false); assert.equal(JSON.stringify(readiness).includes("pg_hba.conf"), false); }); test("configured postgres runtime reports schema blocker without green readiness", async () => { const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient: { async query(sql) { assert.match(sql, /information_schema\.columns/u); return { rows: [] }; } } }); const readiness = await store.readiness(); assert.equal(readiness.adapter, RUNTIME_STORE_KIND_POSTGRES); assert.equal(readiness.durable, false); assert.equal(readiness.durableRequested, true); assert.equal(readiness.ready, false); assert.equal(readiness.status, "blocked"); assert.equal(readiness.blocker, RUNTIME_DURABLE_ADAPTER_SCHEMA_BLOCKED); assert.equal(readiness.liveRuntimeEvidence, false); assert.equal(readiness.fixtureEvidence, false); assert.equal(readiness.schema.checked, true); assert.ok(readiness.schema.missingTables.includes("gateway_sessions")); assert.equal(readiness.migration.checked, false); assert.equal(readiness.gates.schema.status, "blocked"); assert.equal(readiness.gates.migration.status, "not_checked"); assert.equal(readiness.durabilityContract.ready, false); assert.equal(readiness.durabilityContract.blockedLayer, "schema"); assert.equal(readiness.durabilityContract.dbLiveEvidenceIsDurabilityEvidence, false); assert.equal(JSON.stringify(readiness).includes("redacted@db.example"), false); }); test("configured postgres runtime keeps durable false when migration ledger is missing", async () => { const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient: createFakePostgresClient({ migrationReady: false }) }); const readiness = await store.readiness(); assert.equal(readiness.adapter, RUNTIME_STORE_KIND_POSTGRES); assert.equal(readiness.durable, false); assert.equal(readiness.durableRequested, true); assert.equal(readiness.ready, false); assert.equal(readiness.status, "blocked"); assert.equal(readiness.blocker, RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED); assert.equal(readiness.schema.ready, true); assert.equal(readiness.migration.checked, true); assert.equal(readiness.migration.ready, false); assert.equal(readiness.migration.missing, true); assert.equal(readiness.gates.schema.status, "ready"); assert.equal(readiness.gates.migration.status, "blocked"); assert.equal(readiness.liveRuntimeEvidence, false); assert.equal(readiness.durabilityContract.ready, false); assert.equal(readiness.durabilityContract.blockedLayer, "migration"); assert.equal(readiness.durabilityContract.dbLiveEvidenceIsDurabilityEvidence, false); }); test("configured postgres runtime classifies migration query failures separately from generic readiness queries", async () => { const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, queryClient: createFakePostgresClient({ migrationErrorCode: "57014" }) }); const readiness = await store.readiness(); assert.equal(readiness.blocker, RUNTIME_DURABLE_ADAPTER_MIGRATION_BLOCKED); assert.equal(readiness.schema.ready, true); assert.equal(readiness.migration.checked, true); assert.equal(readiness.migration.errorCode, "57014"); assert.equal(readiness.connection.queryAttempted, true); assert.equal(readiness.connection.queryResult, "migration_blocked"); assert.equal(readiness.gates.ssl.status, "ready"); assert.equal(readiness.gates.auth.status, "ready"); assert.equal(readiness.gates.schema.status, "ready"); assert.equal(readiness.gates.migration.status, "blocked"); assert.equal(readiness.gates.durability.status, "not_checked"); assert.equal(readiness.durabilityContract.blockedLayer, "migration"); }); test("configured postgres runtime classifies durable read query blocker after schema and migration", async () => { const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient: createFakePostgresClient({ migrationReady: true, countErrorCode: "57014" }) }); const readiness = await store.readiness(); assert.equal(readiness.adapter, RUNTIME_STORE_KIND_POSTGRES); assert.equal(readiness.durable, false); assert.equal(readiness.ready, false); assert.equal(readiness.status, "blocked"); assert.equal(readiness.blocker, RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED); assert.equal(readiness.schema.ready, true); assert.equal(readiness.migration.ready, true); assert.equal(readiness.connection.queryAttempted, true); assert.equal(readiness.connection.queryResult, "query_blocked"); assert.equal(readiness.gates.auth.status, "ready"); assert.equal(readiness.gates.schema.status, "ready"); assert.equal(readiness.gates.migration.status, "ready"); assert.equal(readiness.gates.durability.status, "blocked"); assert.equal(readiness.durabilityContract.ready, false); assert.equal(readiness.durabilityContract.blockedLayer, "durability_query"); assert.equal(readiness.durabilityContract.dbLiveEvidenceIsDurabilityEvidence, false); }); test("JSON-RPC audit/evidence durable queries return blocked errors instead of empty success", async () => { const secretDbUrl = "postgres://hwlab_user:super-secret-password@db.example.invalid:5432/hwlab"; const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: secretDbUrl, queryClient: createFakePostgresClient({ migrationReady: true, countErrorCode: "57014" }) }); const context = { runtimeStore: store, dbProbe: { probe: false } }; const meta = { traceId: "trc_01J00000000000000000002000", actorId: "usr_01J00000000000000000002000", serviceId: "hwlab-cloud-web", environment: "dev" }; for (const [id, method] of [ ["req_01J00000000000000000002001", "audit.event.query"], ["req_01J00000000000000000002002", "evidence.record.query"] ]) { const response = await handleJsonRpcRequest( { jsonrpc: "2.0", id, method, params: { projectId: "prj_01J00000000000000000002000" }, meta }, context ); validateResponse(response); assert.equal(Object.hasOwn(response, "result"), false); assert.equal(response.error.code, ERROR_CODES.internalError); assert.equal(response.error.data.method, method); assert.equal(response.error.data.adapter, "postgres"); assert.equal(response.error.data.status, "blocked"); assert.equal(response.error.data.blocker, RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED); assert.equal(response.error.data.durable, false); assert.equal(response.error.data.durableRequested, true); assert.equal(response.error.data.liveRuntimeEvidence, false); assert.equal(response.error.data.queryResult, "query_blocked"); assert.equal(response.error.data.blockedLayer, "durability_query"); assert.equal(response.error.data.requiredEvidence, RUNTIME_DURABILITY_REQUIRED_EVIDENCE); assert.equal(response.error.data.dbLiveEvidenceIsDurabilityEvidence, false); assert.equal(response.error.data.secretMaterialRead, false); assert.equal(response.error.data.valueRedacted, true); assert.equal(response.error.data.endpointRedacted, true); assert.equal(Object.hasOwn(response.error.data, "events"), false); assert.equal(Object.hasOwn(response.error.data, "records"), false); assert.equal(Object.hasOwn(response.error.data, "count"), false); const serialized = JSON.stringify(response); assert.equal(serialized.includes("super-secret-password"), false); assert.equal(serialized.includes(secretDbUrl), false); } }); test("JSON-RPC audit/evidence table read failures return blocked errors", async () => { const queryClient = createFakePostgresClient({ migrationReady: true, readErrorCode: "57014" }); const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_user:another-secret@db.example.invalid:5432/hwlab", queryClient }); const context = { runtimeStore: store, dbProbe: { probe: false } }; const meta = { traceId: "trc_01J00000000000000000002200", actorId: "usr_01J00000000000000000002200", serviceId: "hwlab-cloud-web", environment: "dev" }; for (const [id, method] of [ ["req_01J00000000000000000002201", "audit.event.query"], ["req_01J00000000000000000002202", "evidence.record.query"] ]) { const response = await handleJsonRpcRequest( { jsonrpc: "2.0", id, method, params: { projectId: "prj_01J00000000000000000002200" }, meta }, context ); validateResponse(response); assert.equal(Object.hasOwn(response, "result"), false); assert.equal(response.error.code, ERROR_CODES.internalError); assert.equal(response.error.data.method, method); assert.equal(response.error.data.blocker, RUNTIME_DURABLE_ADAPTER_QUERY_BLOCKED); assert.equal(response.error.data.queryResult, "query_blocked"); assert.equal(response.error.data.blockedLayer, "durability_query"); const serialized = JSON.stringify(response); assert.equal(serialized.includes("another-secret"), false); assert.equal(serialized.includes("events"), false); assert.equal(serialized.includes("records"), false); assert.equal(serialized.includes("\"count\":0"), false); } }); test("JSON-RPC durable queries return successful empty result sets for true empty tables", async () => { const queryClient = createFakePostgresClient({ migrationReady: true }); const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient }); const context = { runtimeStore: store, dbProbe: { probe: false } }; const meta = { traceId: "trc_01J00000000000000000002100", actorId: "usr_01J00000000000000000002100", serviceId: "hwlab-cloud-web", environment: "dev" }; async function rpc(id, method, params = {}) { const response = await handleJsonRpcRequest( { jsonrpc: "2.0", id, method, params, meta }, context ); validateResponse(response); assert.equal(Object.hasOwn(response, "error"), false, response.error?.message); return response.result; } const audit = await rpc("req_01J00000000000000000002101", "audit.event.query", { projectId: "prj_01J00000000000000000002100" }); assert.deepEqual(audit.events, []); assert.equal(audit.count, 0); assert.equal(audit.persistence.adapter, "postgres"); assert.equal(audit.persistence.durable, true); assert.equal(audit.persistence.ready, true); const evidence = await rpc("req_01J00000000000000000002102", "evidence.record.query", { projectId: "prj_01J00000000000000000002100" }); assert.deepEqual(evidence.records, []); assert.equal(evidence.count, 0); assert.equal(evidence.persistence.adapter, "postgres"); assert.equal(evidence.persistence.durable, true); assert.equal(evidence.persistence.ready, true); }); test("configured postgres runtime persists and queries records through query client", async () => { const queryClient = createFakePostgresClient({ migrationReady: true }); const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient, now: () => "2026-05-22T00:00:00.000Z" }); const context = { runtimeStore: store, dbProbe: { probe: false } }; const meta = { traceId: "trc_01J00000000000000000001000", actorId: "usr_01J00000000000000000001000", serviceId: "hwlab-cloud-web", environment: "dev" }; async function rpc(id, method, params = {}) { const response = await handleJsonRpcRequest( { jsonrpc: "2.0", id, method, params, meta }, context ); validateResponse(response); assert.equal(Object.hasOwn(response, "error"), false, response.error?.message); return response.result; } const health = await rpc("req_01J00000000000000000001000", "system.health"); assert.equal(health.runtime.adapter, "postgres"); assert.equal(health.runtime.durable, true); assert.equal(health.runtime.durableRequested, true); assert.equal(health.runtime.durableCapable, true); assert.equal(health.runtime.ready, true); assert.equal(health.runtime.liveRuntimeEvidence, true); assert.equal(health.runtime.fixtureEvidence, false); assert.equal(health.runtime.schema.ready, true); assert.equal(health.runtime.migration.ready, true); assert.equal(health.runtime.gates.migration.status, "ready"); assert.equal(health.runtime.gates.durability.status, "ready"); assert.equal(health.runtime.durabilityContract.ready, true); assert.equal(health.runtime.durabilityContract.blockedLayer, null); assert.equal(health.runtime.durabilityContract.dbLiveEvidenceIsDurabilityEvidence, false); assert.equal(health.readiness.components.runtime, "ready"); assert.equal(health.readiness.durability.status, "ready"); assert.equal(health.readiness.durability.dbLiveEvidenceIsDurabilityEvidence, false); await rpc("req_01J00000000000000000001001", "gateway.session.register", { projectId: "prj_01J00000000000000000001000", gatewaySessionId: "gws_01J00000000000000000001000", gatewayId: "gtw_01J00000000000000000001000", serviceId: "hwlab-gateway", endpoint: "http://127.0.0.1:7101" }); await rpc("req_01J00000000000000000001002", "box.resource.register", { projectId: "prj_01J00000000000000000001000", gatewaySessionId: "gws_01J00000000000000000001000", resourceId: "res_01J00000000000000000001000", boxId: "box_01J00000000000000000001000", resourceType: "board", state: "available" }); await rpc("req_01J00000000000000000001003", "box.capability.report", { capabilityId: "cap_01J00000000000000000001000", resourceId: "res_01J00000000000000000001000", projectId: "prj_01J00000000000000000001000", name: "shell.exec", direction: "bidirectional", valueType: "object", mutatesState: true }); const invoke = await rpc("req_01J00000000000000000001004", "hardware.invoke.shell", { projectId: "prj_01J00000000000000000001000", gatewaySessionId: "gws_01J00000000000000000001000", resourceId: "res_01J00000000000000000001000", capabilityId: "cap_01J00000000000000000001000", input: { command: "echo durable" } }); const audit = await rpc("req_01J00000000000000000001005", "audit.event.query", { projectId: "prj_01J00000000000000000001000" }); const evidence = await rpc("req_01J00000000000000000001006", "evidence.record.query", { projectId: "prj_01J00000000000000000001000" }); assert.ok(queryClient.calls.some((call) => call.sql.startsWith("INSERT INTO gateway_sessions"))); assert.ok(queryClient.calls.some((call) => call.sql.startsWith("INSERT INTO evidence_records"))); assert.ok(audit.events.some((event) => event.action === "hardware.invoke.shell")); assert.equal(evidence.count, 1); assert.equal(evidence.records[0].operationId, invoke.operationId); }); test("configured postgres runtime persists and queries Code Agent trace events", async () => { const queryClient = createFakePostgresClient({ migrationReady: true }); const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient, now: () => "2026-06-17T03:50:00.000Z" }); const traceId = "trc_01J00000000000000000002000"; const write = await store.writeAgentTraceEvent({ event: { traceId, seq: 1, type: "runner", status: "completed", label: "runner:completed", sessionId: "ses_01J00000000000000000002000", message: "turn completed", terminal: true, createdAt: "2026-06-17T03:50:01.000Z", valuesPrinted: false } }); const events = await store.queryAgentTraceEvents({ traceId }); const readiness = await store.readiness(); assert.equal(write.written, true); assert.equal(write.traceEvent.traceId, traceId); assert.equal(write.traceEvent.agentSessionId, "ses_01J00000000000000000002000"); assert.equal(events.count, 1); assert.equal(events.events[0].traceId, traceId); assert.equal(events.events[0].terminal, true); assert.equal(readiness.counts.agentTraceEvents, 1); assert.ok(queryClient.calls.some((call) => call.sql.startsWith("CREATE INDEX IF NOT EXISTS idx_agent_trace_events_trace_order"))); assert.ok(queryClient.calls.some((call) => call.sql.startsWith("INSERT INTO agent_trace_events"))); }); test("configured postgres runtime persists and queries Workbench projection state", async () => { const queryClient = createFakePostgresClient({ migrationReady: true }); const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient, now: () => "2026-06-19T13:05:00.000Z" }); const write = await store.writeWorkbenchProjectionState({ projectionState: { traceId: "trc_projection_state", sessionId: "ses_projection_state", conversationId: "cnv_projection_state", threadId: "thread-projection-state", runId: "run_projection_state", commandId: "cmd_projection_state", lastSourceSeq: 3700, lastProjectedSeq: 129, sourceLatestSeq: 3701, projectionStatus: "projecting", resultSyncState: "pending", updatedAt: "2026-06-19T13:05:00.000Z" } }); const loaded = await store.getWorkbenchProjectionState({ traceId: "trc_projection_state" }); const due = await store.queryWorkbenchProjectionStates({ projectionStatuses: ["projecting"], dueAt: "2026-06-19T13:06:00.000Z", limit: 10 }); const readiness = await store.readiness(); assert.equal(write.written, true); assert.equal(write.projectionState.lastSourceSeq, 3700); assert.equal(write.projectionState.lastAgentRunSeq, 3700); assert.equal(loaded.found, true); assert.equal(loaded.projectionState.commandId, "cmd_projection_state"); assert.equal(due.count, 1); assert.equal(due.states[0].resultSyncState, "pending"); assert.equal(readiness.counts.workbenchProjectionStates, 1); assert.ok(queryClient.calls.some((call) => call.sql.startsWith("CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_status_retry_updated"))); assert.ok(queryClient.calls.some((call) => call.sql.startsWith("INSERT INTO workbench_projection_state"))); }); test("configured postgres runtime persists and queries Workbench durable facts", async () => { const queryClient = createFakePostgresClient({ migrationReady: true }); const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient, now: () => "2026-06-20T10:00:00.000Z" }); const write = await store.writeWorkbenchFacts({ facts: { sessions: [{ sessionId: "ses_fact_1", ownerUserId: "usr_fact_1", projectId: "prj_fact_1", conversationId: "cnv_fact_1", threadId: "thread-fact-1", status: "running", lastTraceId: "trc_fact_1", projectedSeq: 11, sourceSeq: 12, sourceEventId: "evt_fact_12", updatedAt: "2026-06-20T10:00:01.000Z" }], messages: [{ messageId: "msg_fact_1", sessionId: "ses_fact_1", turnId: "turn_fact_1", traceId: "trc_fact_1", role: "assistant", status: "streaming", projectedSeq: 13, sourceSeq: 14, sourceEventId: "evt_fact_14", updatedAt: "2026-06-20T10:00:02.000Z" }], parts: [{ partId: "part_fact_1", messageId: "msg_fact_1", sessionId: "ses_fact_1", turnId: "turn_fact_1", traceId: "trc_fact_1", partIndex: 0, partType: "text", status: "streaming", updatedAt: "2026-06-20T10:00:03.000Z" }], turns: [{ turnId: "turn_fact_1", sessionId: "ses_fact_1", traceId: "trc_fact_1", messageId: "msg_fact_1", status: "completed", finalResponse: { text: "done" }, terminal: true, sealed: true, updatedAt: "2026-06-20T10:00:04.000Z" }], traceEvents: [{ id: "wte_fact_1", traceId: "trc_fact_1", sessionId: "ses_fact_1", turnId: "turn_fact_1", messageId: "msg_fact_1", sourceSeq: 15, projectedSeq: 16, eventType: "assistant.delta", occurredAt: "2026-06-20T10:00:05.000Z" }], checkpoints: [{ traceId: "trc_fact_1", sessionId: "ses_fact_1", turnId: "turn_fact_1", runId: "run_fact_1", commandId: "cmd_fact_1", projectedSeq: 16, sourceSeq: 15, projectionStatus: "caught_up", projectionHealth: "healthy", terminal: true, sealed: true, updatedAt: "2026-06-20T10:00:06.000Z" }] } }); const loaded = await store.queryWorkbenchFacts({ sessionId: "ses_fact_1", limit: 10 }); const readiness = await store.readiness(); assert.equal(write.written, true); assert.equal(write.facts.sessions[0].terminal, false); assert.equal(loaded.count, 6); assert.equal(loaded.facts.sessions[0].sessionId, "ses_fact_1"); assert.equal(loaded.facts.messages[0].messageId, "msg_fact_1"); assert.equal(loaded.facts.parts[0].partId, "part_fact_1"); assert.equal(loaded.facts.turns[0].finalResponse.text, "done"); assert.equal(loaded.facts.traceEvents[0].eventType, "assistant.delta"); assert.equal(loaded.facts.checkpoints[0].projectionStatus, "caught_up"); assert.equal(readiness.counts.workbenchSessions, 1); assert.equal(readiness.counts.workbenchMessages, 1); assert.equal(readiness.counts.workbenchParts, 1); assert.equal(readiness.counts.workbenchTurns, 1); assert.equal(readiness.counts.workbenchTraceEvents, 1); assert.equal(readiness.counts.workbenchProjectionCheckpoints, 1); assert.ok(queryClient.calls.some((call) => call.sql.startsWith("CREATE INDEX IF NOT EXISTS idx_workbench_sessions_owner_updated"))); assert.ok(queryClient.calls.some((call) => call.sql.startsWith("INSERT INTO workbench_sessions"))); assert.ok(queryClient.calls.some((call) => call.sql.startsWith("INSERT INTO workbench_projection_checkpoints"))); assert.ok(queryClient.calls.some((call) => call.sql.startsWith("SELECT session_json FROM workbench_sessions WHERE session_id = $1"))); }); test("configured postgres runtime allocates Workbench projectedSeq idempotently by source event", async () => { const queryClient = createFakePostgresClient({ migrationReady: true }); const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient, now: () => "2026-06-20T12:30:00.000Z" }); const traceId = "trc_pg_projected_seq_idempotent"; const first = await store.allocateWorkbenchProjectedSeq({ traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-a" }); await store.writeWorkbenchFacts({ facts: { traceEvents: [{ id: "wte_pg_source_a", traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-a", projectedSeq: first.projectedSeq, eventType: "backend", occurredAt: "2026-06-20T12:30:01.000Z" }], checkpoints: [{ traceId, sessionId: "ses_pg_projected_seq_idempotent", projectedSeq: first.projectedSeq, sourceSeq: 1, sourceEventId: "source-a", projectionStatus: "projecting", projectionHealth: "healthy", updatedAt: "2026-06-20T12:30:01.000Z" }] } }); const second = await store.allocateWorkbenchProjectedSeq({ traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-b" }); await store.writeWorkbenchFacts({ facts: { traceEvents: [{ id: "wte_pg_source_b", traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-b", projectedSeq: second.projectedSeq, eventType: "backend", occurredAt: "2026-06-20T12:30:02.000Z" }], checkpoints: [{ traceId, sessionId: "ses_pg_projected_seq_idempotent", projectedSeq: second.projectedSeq, sourceSeq: 1, sourceEventId: "source-b", projectionStatus: "projecting", projectionHealth: "healthy", updatedAt: "2026-06-20T12:30:02.000Z" }] } }); const repeated = await store.allocateWorkbenchProjectedSeq({ traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-a" }); await store.writeWorkbenchFacts({ facts: { checkpoints: [{ traceId, sessionId: "ses_pg_projected_seq_idempotent", projectedSeq: repeated.projectedSeq, sourceSeq: 1, sourceEventId: "source-a", projectionStatus: "projecting", projectionHealth: "healthy", updatedAt: "2026-06-20T12:30:03.000Z" }] } }); const loaded = await store.queryWorkbenchFacts({ traceId, families: ["traceEvents", "checkpoints"], limit: 10 }); assert.equal(first.projectedSeq, 1); assert.equal(second.projectedSeq, 2); assert.equal(repeated.projectedSeq, 1); assert.deepEqual(loaded.facts.traceEvents.map((event) => event.projectedSeq), [1, 2]); assert.equal(loaded.facts.checkpoints[0].projectedSeq, 2); assert.ok(queryClient.calls.some((call) => call.sql.startsWith("SELECT pg_advisory_xact_lock"))); }); test("configured postgres runtime reuses proven readiness for Workbench durable reads", async () => { const queryClient = createFakePostgresClient({ migrationReady: true }); const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient, now: () => "2026-06-20T10:00:00.000Z" }); await store.writeWorkbenchFacts({ facts: { sessions: [{ sessionId: "ses_ready_cached", ownerUserId: "usr_ready", status: "running", updatedAt: "2026-06-20T10:00:01.000Z" }] } }); queryClient.calls.length = 0; const first = await store.queryWorkbenchFacts({ families: ["sessions"], limit: 10 }); const second = await store.queryWorkbenchFacts({ families: ["sessions"], limit: 10 }); assert.equal(first.count, 1); assert.equal(second.count, 1); assert.equal(queryClient.calls.filter((call) => isReadinessQuery(call.sql)).length, 0); assert.equal(queryClient.calls.filter((call) => call.sql.startsWith("SELECT session_json FROM workbench_sessions")).length, 2); }); test("configured postgres runtime can query thin Workbench session summaries", async () => { const queryClient = createFakePostgresClient({ migrationReady: true }); const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient, now: () => "2026-06-20T10:00:00.000Z" }); await store.writeWorkbenchFacts({ facts: { sessions: [{ sessionId: "ses_summary_projection", ownerUserId: "usr_summary_projection", projectId: "prj_summary_projection", conversationId: "cnv_summary_projection", threadId: "thr_summary_projection", status: "running", lastTraceId: "trc_summary_projection", projectedSeq: 7, sourceSeq: 7, sourceEventId: "evt_summary_projection", terminal: false, sealed: false, sessionJson: { providerProfile: "codex-fast", messages: [{ role: "user", text: "large payload".repeat(500) }], prompt: "should stay out of list projection" }, createdAt: "2026-06-20T09:59:00.000Z", updatedAt: "2026-06-20T10:00:00.000Z" }] } }); queryClient.calls.length = 0; const loaded = await store.queryWorkbenchFacts({ families: ["sessions"], sessionProjection: "summary", limit: 10 }); const session = loaded.facts.sessions[0]; const readCall = queryClient.calls.find((call) => call.sql.includes("FROM workbench_sessions")); assert.equal(session.sessionId, "ses_summary_projection"); assert.equal(session.providerProfile, "codex-fast"); assert.deepEqual(session.sessionJson, { providerProfile: "codex-fast", valuesRedacted: true, secretMaterialStored: false }); assert.equal(session.sessionJson.messages, undefined); assert.ok(readCall?.sql.startsWith("SELECT session_id, owner_user_id")); assert.equal(readCall.sql.includes("SELECT session_json FROM workbench_sessions"), false); assert.ok(readCall.sql.includes("providerProfile")); }); test("configured postgres runtime orders Workbench sessions by sessionsOrder updated_desc", async () => { const queryClient = createFakePostgresClient({ migrationReady: true }); const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient, now: () => "2026-06-20T10:00:00.000Z" }); await store.writeWorkbenchFacts({ facts: { sessions: [ { sessionId: "ses_order_old", ownerUserId: "usr_order", status: "running", updatedAt: "2026-06-15T10:00:00.000Z" }, { sessionId: "ses_order_new", ownerUserId: "usr_order", status: "running", updatedAt: "2026-06-20T10:00:00.000Z" }, { sessionId: "ses_order_middle", ownerUserId: "usr_order", status: "running", updatedAt: "2026-06-18T10:00:00.000Z" } ] } }); queryClient.calls.length = 0; const loaded = await store.queryWorkbenchFacts({ families: ["sessions"], sessionsOrder: "updated_desc", limit: 10 }); const readCall = queryClient.calls.find((call) => call.sql.startsWith("SELECT session_json FROM workbench_sessions")); assert.deepEqual(loaded.facts.sessions.map((session) => session.sessionId), ["ses_order_new", "ses_order_middle", "ses_order_old"]); assert.match(readCall?.sql ?? "", /ORDER BY updated_at DESC/u); }); test("configured postgres runtime queries requested workbench fact families concurrently", async () => { const expectedReads = 4; let started = 0; let releaseReads; let resolveAllStarted; const readGate = new Promise((resolve) => { releaseReads = resolve; }); const allStarted = new Promise((resolve) => { resolveAllStarted = resolve; }); const queryClient = createFakePostgresClient({ migrationReady: true, beforeWorkbenchFactRead: async (sql) => { if (!/^SELECT (message_json|part_json|turn_json|checkpoint_json) FROM/u.test(sql)) return; started += 1; if (started === expectedReads) resolveAllStarted(); await readGate; } }); const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient, now: () => "2026-06-20T10:00:00.000Z" }); await store.writeWorkbenchFacts({ facts: { sessions: [], messages: [{ messageId: "msg_parallel", sessionId: "ses_parallel", turnId: "turn_parallel", traceId: "trc_parallel", role: "assistant", status: "completed", text: "done", updatedAt: "2026-06-20T10:00:01.000Z" }], parts: [{ partId: "part_parallel", messageId: "msg_parallel", sessionId: "ses_parallel", turnId: "turn_parallel", traceId: "trc_parallel", partIndex: 0, partType: "text", status: "completed", text: "done", updatedAt: "2026-06-20T10:00:02.000Z" }], turns: [{ turnId: "turn_parallel", sessionId: "ses_parallel", traceId: "trc_parallel", messageId: "msg_parallel", status: "completed", finalResponse: { text: "done" }, updatedAt: "2026-06-20T10:00:03.000Z" }], traceEvents: [], checkpoints: [{ traceId: "trc_parallel", sessionId: "ses_parallel", turnId: "turn_parallel", runId: "run_parallel", commandId: "cmd_parallel", projectionStatus: "caught_up", projectionHealth: "healthy", updatedAt: "2026-06-20T10:00:04.000Z" }] } }); const pending = store.queryWorkbenchFacts({ sessionId: "ses_parallel", traceId: "trc_parallel", families: ["messages", "parts", "turns", "checkpoints"], limit: 10 }); const startedConcurrently = await Promise.race([ allStarted.then(() => true), new Promise((resolve) => setTimeout(() => resolve(false), 150)) ]); releaseReads(); const loaded = await pending; assert.equal(startedConcurrently, true); assert.equal(started, expectedReads); assert.equal(loaded.count, 4); assert.equal(loaded.facts.messages[0].messageId, "msg_parallel"); assert.equal(loaded.facts.parts[0].partId, "part_parallel"); assert.equal(loaded.facts.turns[0].turnId, "turn_parallel"); assert.equal(loaded.facts.checkpoints[0].traceId, "trc_parallel"); }); test("configured postgres runtime pushes trace event query filters into SQL", async () => { const queryClient = createFakePostgresClient({ migrationReady: true }); const store = createConfiguredCloudRuntimeStore({ env: { HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", HWLAB_CLOUD_RUNTIME_DURABLE: "true" }, dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", queryClient, now: () => "2026-06-17T03:50:00.000Z" }); await store.writeAgentTraceEvent({ event: { traceId: "trc_sql_filter_a", sessionId: "ses_filter_a", seq: 1, level: "info", label: "a", createdAt: "2026-06-17T03:50:01.000Z" } }); await store.writeAgentTraceEvent({ event: { traceId: "trc_sql_filter_b", sessionId: "ses_filter_b", seq: 1, level: "info", label: "b", createdAt: "2026-06-17T03:50:02.000Z" } }); const events = await store.queryAgentTraceEvents({ sessionId: "ses_filter_b" }); assert.equal(events.count, 1); assert.equal(events.events[0].traceId, "trc_sql_filter_b"); const select = queryClient.calls.findLast((call) => call.sql.startsWith("SELECT event_json FROM agent_trace_events")); assert.match(select.sql, /WHERE agent_session_id = \$1/u); assert.deepEqual(select.params, ["ses_filter_b"]); }); function isReadinessQuery(sql) { return sql.includes("information_schema.columns") || sql.startsWith("SELECT id, schema_version FROM hwlab_schema_migrations") || sql.startsWith("SELECT COUNT(*)::int AS count FROM "); } function createFakePostgresClient({ migrationReady = true, migrationErrorCode = null, countErrorCode = null, readErrorCode = null, beforeWorkbenchFactRead = null } = {}) { const state = { gateway_sessions: new Map(), box_resources: new Map(), box_capabilities: new Map(), hardware_operations: new Map(), audit_events: new Map(), evidence_records: new Map(), worker_sessions: new Map(), agent_trace_events: new Map(), workbench_projection_state: new Map(), workbench_sessions: new Map(), workbench_messages: new Map(), workbench_parts: new Map(), workbench_turns: new Map(), workbench_trace_events: new Map(), workbench_projection_checkpoints: new Map(), hwlab_schema_migrations: new Map() }; if (migrationReady) { state.hwlab_schema_migrations.set(CLOUD_CORE_MIGRATION_ID, { id: CLOUD_CORE_MIGRATION_ID, schema_version: CLOUD_RUNTIME_DURABLE_ADAPTER_SCHEMA_VERSION }); } const calls = []; return { calls, async query(sql, params = []) { calls.push({ sql, params }); if (["BEGIN", "COMMIT", "ROLLBACK"].includes(sql)) { return { rows: [] }; } if (sql.includes("information_schema.columns")) { return { rows: schemaRows() }; } if (sql.startsWith("SELECT COUNT(*)::int AS count FROM ")) { if (countErrorCode) { const error = new Error("count query failed"); error.code = countErrorCode; throw error; } const table = sql.match(/FROM ([a-z_]+)/u)?.[1]; return { rows: [{ count: state[table]?.size ?? 0 }] }; } if (sql.startsWith("SELECT id, schema_version FROM hwlab_schema_migrations")) { if (migrationErrorCode) { const error = new Error("migration query failed"); error.code = migrationErrorCode; throw error; } const row = state.hwlab_schema_migrations.get(params[0]); return { rows: row ? [row] : [] }; } if (sql.startsWith("CREATE INDEX IF NOT EXISTS idx_agent_trace_events_")) { return { rows: [] }; } if (sql.startsWith("CREATE INDEX IF NOT EXISTS idx_workbench_projection_state_")) { return { rows: [] }; } if (sql.startsWith("CREATE INDEX IF NOT EXISTS idx_workbench_")) { return { rows: [] }; } if (sql.startsWith("SELECT pg_advisory_xact_lock")) { return { rows: [{ pg_advisory_xact_lock: null }] }; } if (sql.startsWith("SELECT projected_seq FROM workbench_trace_events WHERE trace_id = $1 AND source_event_id = $2")) { const rows = [...state.workbench_trace_events.values()] .filter((record) => record.trace_id === params[0] && record.source_event_id === params[1]) .sort((left, right) => Number(left.projected_seq) - Number(right.projected_seq)); return { rows: rows.slice(0, 1).map((record) => ({ projected_seq: record.projected_seq })) }; } if (sql.startsWith("SELECT GREATEST(COALESCE((SELECT MAX(projected_seq) FROM workbench_trace_events")) { const traceId = params[0]; const eventMax = [...state.workbench_trace_events.values()] .filter((record) => record.trace_id === traceId) .reduce((max, record) => Math.max(max, Number(record.projected_seq ?? 0)), 0); const checkpointMax = Number(state.workbench_projection_checkpoints.get(traceId)?.projected_seq ?? 0); return { rows: [{ max_projected_seq: Math.max(eventMax, checkpointMax) }] }; } if (sql.startsWith("SELECT gateway_session_json FROM gateway_sessions")) { return jsonSelect(state.gateway_sessions, params[0], "gateway_session_json"); } if (sql.startsWith("SELECT resource_json FROM box_resources")) { return jsonSelect(state.box_resources, params[0], "resource_json"); } if (sql.startsWith("SELECT capability_json FROM box_capabilities")) { return jsonSelect(state.box_capabilities, params[0], "capability_json"); } if (sql.startsWith("SELECT event_json FROM audit_events")) { if (readErrorCode) { const error = new Error("audit read query failed"); error.code = readErrorCode; throw error; } return { rows: [...state.audit_events.values()].map((record) => ({ event_json: record.event_json })) }; } if (sql.startsWith("SELECT event_json FROM agent_trace_events")) { if (readErrorCode) { const error = new Error("agent trace event read query failed"); error.code = readErrorCode; throw error; } const traceId = sql.includes("trace_id = $1") ? params[0] : null; const sessionId = sql.includes("agent_session_id = $1") ? params[0] : null; const workerSessionId = sql.includes("worker_session_id = $1") ? params[0] : null; const level = sql.includes("level = $1") ? params[0] : null; const rows = [...state.agent_trace_events.values()] .filter((record) => !traceId || record.trace_id === traceId) .filter((record) => !sessionId || record.agent_session_id === sessionId) .filter((record) => !workerSessionId || record.worker_session_id === workerSessionId) .filter((record) => !level || record.level === level) .sort((left, right) => String(left.occurred_at).localeCompare(String(right.occurred_at)) || String(left.id).localeCompare(String(right.id))) .map((record) => ({ event_json: record.event_json })); return { rows }; } if (sql.startsWith("SELECT projection_json FROM workbench_projection_state")) { if (readErrorCode) { const error = new Error("workbench projection state read query failed"); error.code = readErrorCode; throw error; } let rows = [...state.workbench_projection_state.values()]; if (sql.includes("trace_id = $1")) rows = rows.filter((record) => record.trace_id === params[0]); if (sql.includes("projection_status = ANY")) { const statuses = params.find((param) => Array.isArray(param)) ?? []; rows = rows.filter((record) => statuses.includes(record.projection_status)); } if (sql.includes("next_retry_at IS NULL")) { const dueAt = params.find((param) => typeof param === "string" && /^\d{4}-\d{2}-\d{2}T/u.test(param)); rows = rows.filter((record) => !record.next_retry_at || !dueAt || record.next_retry_at <= dueAt); } rows.sort((left, right) => String(left.updated_at).localeCompare(String(right.updated_at)) || String(left.trace_id).localeCompare(String(right.trace_id))); return { rows: rows.map((record) => ({ projection_json: record.projection_json })) }; } if (sql.startsWith("SELECT session_id, owner_user_id") && sql.includes("FROM workbench_sessions")) { if (beforeWorkbenchFactRead) await beforeWorkbenchFactRead(sql, params); return workbenchSessionSummaryRows(state.workbench_sessions, sql, params, readErrorCode); } if (sql.startsWith("SELECT session_json FROM workbench_sessions")) { if (beforeWorkbenchFactRead) await beforeWorkbenchFactRead(sql, params); return workbenchFactRows(state.workbench_sessions, "session_json", sql, params, readErrorCode); } if (sql.startsWith("SELECT message_json FROM workbench_messages")) { if (beforeWorkbenchFactRead) await beforeWorkbenchFactRead(sql, params); return workbenchFactRows(state.workbench_messages, "message_json", sql, params, readErrorCode); } if (sql.startsWith("SELECT part_json FROM workbench_parts")) { if (beforeWorkbenchFactRead) await beforeWorkbenchFactRead(sql, params); return workbenchFactRows(state.workbench_parts, "part_json", sql, params, readErrorCode); } if (sql.startsWith("SELECT turn_json FROM workbench_turns")) { if (beforeWorkbenchFactRead) await beforeWorkbenchFactRead(sql, params); return workbenchFactRows(state.workbench_turns, "turn_json", sql, params, readErrorCode); } if (sql.startsWith("SELECT event_json FROM workbench_trace_events")) { if (beforeWorkbenchFactRead) await beforeWorkbenchFactRead(sql, params); return workbenchFactRows(state.workbench_trace_events, "event_json", sql, params, readErrorCode); } if (sql.startsWith("SELECT checkpoint_json FROM workbench_projection_checkpoints")) { if (beforeWorkbenchFactRead) await beforeWorkbenchFactRead(sql, params); return workbenchFactRows(state.workbench_projection_checkpoints, "checkpoint_json", sql, params, readErrorCode); } if (sql.startsWith("SELECT metadata_json FROM evidence_records")) { if (readErrorCode) { const error = new Error("evidence read query failed"); error.code = readErrorCode; throw error; } return { rows: [...state.evidence_records.values()].map((record) => ({ metadata_json: record.metadata_json })) }; } if (sql.startsWith("INSERT INTO gateway_sessions")) { state.gateway_sessions.set(params[0], { gateway_session_json: params[6] }); return { rows: [] }; } if (sql.startsWith("INSERT INTO box_resources")) { state.box_resources.set(params[0], { resource_json: params[5] }); return { rows: [] }; } if (sql.startsWith("INSERT INTO box_capabilities")) { state.box_capabilities.set(params[0], { capability_json: params[3] }); return { rows: [] }; } if (sql.startsWith("INSERT INTO hardware_operations")) { state.hardware_operations.set(params[0], { operation_json: params[4] }); return { rows: [] }; } if (sql.startsWith("INSERT INTO audit_events")) { state.audit_events.set(params[0], { event_json: params[8] }); return { rows: [] }; } if (sql.startsWith("INSERT INTO evidence_records")) { state.evidence_records.set(params[0], { metadata_json: params[5] }); return { rows: [] }; } if (sql.startsWith("INSERT INTO agent_trace_events")) { state.agent_trace_events.set(params[0], { id: params[0], trace_id: params[1], agent_session_id: params[2], worker_session_id: params[3], level: params[4], message: params[5], event_json: params[6], occurred_at: params[7] }); return { rows: [] }; } if (sql.startsWith("INSERT INTO workbench_projection_state")) { state.workbench_projection_state.set(params[0], { trace_id: params[0], session_id: params[1], conversation_id: params[2], thread_id: params[3], run_id: params[4], command_id: params[5], last_agentrun_seq: params[6], last_projected_seq: params[7], upstream_latest_seq: params[8], projection_status: params[9], projection_health: params[10], result_sync_state: params[11], next_retry_at: params[17], projection_json: params[18], created_at: params[19], updated_at: params[20] }); return { rows: [] }; } if (sql.startsWith("INSERT INTO workbench_sessions")) { state.workbench_sessions.set(params[0], { session_id: params[0], owner_user_id: params[1], project_id: params[2], conversation_id: params[3], thread_id: params[4], status: params[5], last_trace_id: params[6], projected_seq: params[7], source_seq: params[8], source_event_id: params[9], terminal: params[10], sealed: params[11], session_json: params[12], created_at: params[13], updated_at: params[14] }); return { rows: [] }; } if (sql.startsWith("INSERT INTO workbench_messages")) { state.workbench_messages.set(params[0], { message_id: params[0], session_id: params[1], turn_id: params[2], trace_id: params[3], role: params[4], status: params[5], projected_seq: params[6], source_seq: params[7], source_event_id: params[8], terminal: params[9], sealed: params[10], message_json: params[11], created_at: params[12], updated_at: params[13] }); return { rows: [] }; } if (sql.startsWith("INSERT INTO workbench_parts")) { state.workbench_parts.set(params[0], { part_id: params[0], message_id: params[1], session_id: params[2], turn_id: params[3], trace_id: params[4], part_index: params[5], part_type: params[6], status: params[7], projected_seq: params[8], source_seq: params[9], source_event_id: params[10], terminal: params[11], sealed: params[12], part_json: params[13], created_at: params[14], updated_at: params[15] }); return { rows: [] }; } if (sql.startsWith("INSERT INTO workbench_turns")) { state.workbench_turns.set(params[0], { turn_id: params[0], session_id: params[1], trace_id: params[2], message_id: params[3], status: params[4], projected_seq: params[5], source_seq: params[6], source_event_id: params[7], terminal: params[8], sealed: params[9], final_response_json: params[10], diagnostic_json: params[11], turn_json: params[12], created_at: params[13], updated_at: params[14] }); return { rows: [] }; } if (sql.startsWith("INSERT INTO workbench_trace_events")) { state.workbench_trace_events.set(params[0], { id: params[0], trace_id: params[1], session_id: params[2], turn_id: params[3], message_id: params[4], source_seq: params[5], source_event_id: params[6], projected_seq: params[7], event_type: params[8], terminal: params[9], sealed: params[10], event_json: params[11], occurred_at: params[12], updated_at: params[13] }); return { rows: [] }; } if (sql.startsWith("INSERT INTO workbench_projection_checkpoints")) { const previous = state.workbench_projection_checkpoints.get(params[0]) ?? null; const projectedSeq = sql.includes("RETURNING projected_seq") ? Math.max(Number(previous?.projected_seq ?? 0) + (previous ? 1 : 0), Number(params[5] ?? 0)) : previous && Number(params[5] ?? 0) < Number(previous.projected_seq ?? 0) ? Number(previous.projected_seq) : Number(params[5] ?? 0); if (previous && !sql.includes("RETURNING projected_seq") && Number(params[5] ?? 0) < Number(previous.projected_seq ?? 0)) { return { rows: [] }; } state.workbench_projection_checkpoints.set(params[0], { trace_id: params[0], session_id: params[1], turn_id: params[2], run_id: params[3], command_id: params[4], projected_seq: projectedSeq, source_seq: params[6], source_event_id: params[7], projection_status: params[8], projection_health: params[9], terminal: params[10], sealed: params[11], diagnostic_json: params[12], checkpoint_json: params[13], created_at: params[14], updated_at: params[15] }); return { rows: sql.includes("RETURNING projected_seq") ? [{ projected_seq: projectedSeq }] : [] }; } throw new Error(`unexpected sql: ${sql}`); } }; } function schemaRows() { return Object.entries(CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS).flatMap(([table, columns]) => columns.map((column) => ({ table_name: table, column_name: column })) ); } function jsonSelect(map, id, column) { const row = map.get(id); return { rows: row ? [{ [column]: row[column] }] : [] }; } function workbenchFactRows(map, jsonColumn, sql, params, readErrorCode) { if (readErrorCode) { const error = new Error("workbench fact read query failed"); error.code = readErrorCode; throw error; } const direction = sql.includes("ORDER BY updated_at DESC") ? -1 : 1; const rows = [...map.values()] .filter((record) => sqlRecordMatches(record, sql, params)) .sort((left, right) => direction * (String(left.updated_at).localeCompare(String(right.updated_at)) || String(left.id ?? left.session_id ?? left.message_id ?? left.part_id ?? left.turn_id ?? left.trace_id).localeCompare(String(right.id ?? right.session_id ?? right.message_id ?? right.part_id ?? right.turn_id ?? right.trace_id)))) .map((record) => ({ [jsonColumn]: record[jsonColumn] })); const limit = sql.includes(" LIMIT $") ? Number(params.at(-1)) : null; return { rows: Number.isInteger(limit) && limit > 0 ? rows.slice(0, limit) : rows }; } function workbenchSessionSummaryRows(map, sql, params, readErrorCode) { if (readErrorCode) { const error = new Error("workbench fact read query failed"); error.code = readErrorCode; throw error; } const direction = sql.includes("ORDER BY updated_at DESC") ? -1 : 1; const rows = [...map.values()] .filter((record) => sqlRecordMatches(record, sql, params)) .sort((left, right) => direction * (String(left.updated_at).localeCompare(String(right.updated_at)) || String(left.session_id).localeCompare(String(right.session_id)))) .map((record) => { const session = JSON.parse(record.session_json || "{}"); const providerProfile = session.providerProfile ?? session.sessionJson?.providerProfile ?? null; return { session_id: record.session_id, owner_user_id: record.owner_user_id, project_id: record.project_id, conversation_id: record.conversation_id, thread_id: record.thread_id, status: record.status, last_trace_id: record.last_trace_id, projected_seq: record.projected_seq, source_seq: record.source_seq, source_event_id: record.source_event_id, terminal: record.terminal, sealed: record.sealed, created_at: record.created_at, updated_at: record.updated_at, provider_profile: providerProfile }; }); const limit = sql.includes(" LIMIT $") ? Number(params.at(-1)) : null; return { rows: Number.isInteger(limit) && limit > 0 ? rows.slice(0, limit) : rows }; } function sqlRecordMatches(record, sql, params) { for (const column of Object.keys(record)) { const match = sql.match(new RegExp(`(?:^|[^a-z_])${column} = \\$(\\d+)`, "u")); if (match && record[column] !== params[Number(match[1]) - 1]) return false; } return true; }