From 4500fb6486b0523d85088e715cbaa03f73b23d6b Mon Sep 17 00:00:00 2001 From: lyon Date: Wed, 17 Jun 2026 11:58:07 +0800 Subject: [PATCH] =?UTF-8?q?feat(workbench):=20=E6=8C=81=E4=B9=85=E5=8C=96?= =?UTF-8?q?=20Code=20Agent=20trace=20=E6=8A=95=E5=BD=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/cloud/code-agent-trace-store.test.ts | 28 ++- internal/cloud/code-agent-trace-store.ts | 35 ++++ internal/cloud/server-workbench-http.test.ts | 71 +++++++ internal/cloud/server-workbench-http.ts | 72 ++++++- internal/cloud/server.ts | 9 +- internal/db/runtime-store.test.ts | 68 +++++++ internal/db/runtime-store.ts | 175 +++++++++++++++++- internal/db/schema.ts | 22 +++ 8 files changed, 466 insertions(+), 14 deletions(-) diff --git a/internal/cloud/code-agent-trace-store.test.ts b/internal/cloud/code-agent-trace-store.test.ts index 9a6c0ca5..0a0e2578 100644 --- a/internal/cloud/code-agent-trace-store.test.ts +++ b/internal/cloud/code-agent-trace-store.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import { test } from "bun:test"; -import { createCodeAgentTraceStore } from "./code-agent-trace-store.ts"; +import { createCodeAgentTraceStore, createDurableCodeAgentTraceStore } from "./code-agent-trace-store.ts"; test("code agent trace store keeps assistant deltas outside event count", () => { const traceStore = createCodeAgentTraceStore({ maxEvents: 6000 }); @@ -42,6 +42,32 @@ test("code agent trace store dedupes repeated AgentRun source events", () => { assert.deepEqual(snapshot.events.map((item) => item.label), ["agentrun:backend:resource-bundle-materialized", "agentrun:assistant:message"]); }); +test("durable code agent trace store projects appended events without counting assistant deltas", () => { + const writes = []; + const baseStore = createCodeAgentTraceStore({ maxEvents: 20 }); + const traceStore = createDurableCodeAgentTraceStore({ + traceStore: baseStore, + traceEventStore: { + writeAgentTraceEvent(params, requestMeta) { + writes.push({ params, requestMeta }); + return { written: true }; + } + } + }); + const traceId = "trc_trace-store-durable-projection"; + + traceStore.append(traceId, { type: "request", status: "accepted", label: "request:accepted", sessionId: "ses_trace_store_durable" }); + traceStore.appendAssistantDelta(traceId, { itemId: "assistant", chunk: "hello" }); + + const snapshot = traceStore.snapshot(traceId); + assert.equal(snapshot.eventCount, 1); + assert.equal(snapshot.assistantStreams.length, 1); + assert.equal(writes.length, 1); + assert.equal(writes[0].params.event.traceId, traceId); + assert.equal(writes[0].params.event.seq, 1); + assert.equal(writes[0].requestMeta.agentSessionId, "ses_trace_store_durable"); +}); + test("code agent trace store retains six thousand regular events", () => { const traceStore = createCodeAgentTraceStore({ maxEvents: 6000 }); const traceId = "trc_trace-store-regular-events"; diff --git a/internal/cloud/code-agent-trace-store.ts b/internal/cloud/code-agent-trace-store.ts index b6a70f14..51b4fa01 100644 --- a/internal/cloud/code-agent-trace-store.ts +++ b/internal/cloud/code-agent-trace-store.ts @@ -1,3 +1,7 @@ +/* + * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0 + * 职责: Code Agent trace capture。实时内存 trace 可加速 SSE/详情面板，但 append 事件必须同步投影到 durable store。 + */ import { randomUUID } from "node:crypto"; const DEFAULT_MAX_TRACES = 256; @@ -209,6 +213,37 @@ export function createCodeAgentTraceStore(options = {}) { export const defaultCodeAgentTraceStore = createCodeAgentTraceStore(); +export function createDurableCodeAgentTraceStore({ traceStore = defaultCodeAgentTraceStore, traceEventStore = null } = {}) { + if (!traceEventStore || typeof traceEventStore.writeAgentTraceEvent !== "function") return traceStore; + + function append(traceId, event = {}, meta = {}) { + const normalized = traceStore.append(traceId, event, meta); + if (normalized?.traceId) persistTraceEvent(traceEventStore, normalized, meta); + return normalized; + } + + return { + ensure: (...args) => traceStore.ensure(...args), + append, + appendAssistantDelta: (...args) => traceStore.appendAssistantDelta(...args), + snapshot: (...args) => traceStore.snapshot(...args), + subscribe: (...args) => traceStore.subscribe(...args), + clear: (...args) => traceStore.clear(...args) + }; +} + +function persistTraceEvent(traceEventStore, event, meta = {}) { + try { + Promise.resolve(traceEventStore.writeAgentTraceEvent({ event }, { + traceId: event.traceId, + agentSessionId: event.agentSessionId ?? event.sessionId ?? meta.agentSessionId ?? meta.sessionId, + sessionId: event.sessionId ?? meta.sessionId + })).catch(() => {}); + } catch { + // Trace capture must not fail the user turn when the durable projection is temporarily unavailable. + } +} + export function createCodeAgentTraceRecorder({ traceStore = defaultCodeAgentTraceStore, traceId, diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index d0a9f5c9..b0c1c63a 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -102,6 +102,77 @@ test("workbench read model exposes session, messages, turn, and trace without wr } }); +test("workbench read model recovers trace events from durable projection without duplicating final response", async () => { + const traceStore = createCodeAgentTraceStore(); + const traceId = "trc_workbench_durable_reload"; + const session = { + id: "ses_workbench_durable_reload", + projectId: "prj_hwpod_workbench", + agentId: "hwlab-code-agent", + status: "completed", + startedAt: "2026-06-17T01:00:00.000Z", + endedAt: "2026-06-17T01:00:04.000Z", + ownerUserId: ACTOR.id, + conversationId: "cnv_workbench_durable_reload", + threadId: "thread-workbench-durable-reload", + lastTraceId: traceId, + updatedAt: "2026-06-17T01:00:04.000Z", + session: { + sessionStatus: "completed", + lastTraceId: traceId, + finalResponse: "durable pong", + messages: [ + { role: "user", text: "durable ping", traceId, createdAt: "2026-06-17T01:00:00.000Z" }, + { role: "agent", text: "durable pong", traceId, createdAt: "2026-06-17T01:00:04.000Z" } + ], + valuesRedacted: true, + secretMaterialStored: false + } + }; + const durableEvents = [ + { traceId, seq: 1, type: "request", status: "accepted", label: "request:accepted", createdAt: "2026-06-17T01:00:00.000Z", valuesPrinted: false }, + { traceId, seq: 2, type: "result", status: "completed", label: "result:completed", terminal: true, createdAt: "2026-06-17T01:00:04.000Z", valuesPrinted: false } + ]; + const accessController = { + store: { + async getAgentSession(sessionId) { return sessionId === session.id ? session : null; }, + async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; } + }, + async ensureBootstrap() {}, + async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; } + }; + const runtimeStore = { + async queryAgentTraceEvents(params = {}) { + assert.equal(params.traceId, traceId); + return { events: durableEvents, count: durableEvents.length }; + } + }; + const server = createCloudApiServer({ accessController, traceStore, runtimeStore, codeAgentChatResults: createCodeAgentChatResultStore() }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + + try { + const { port } = server.address(); + const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=10`); + assert.equal(messages.status, 200); + assert.equal(messages.body.total, 2); + assert.deepEqual(messages.body.messages.map((message) => message.role), ["user", "agent"]); + + const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`); + assert.equal(turn.status, 200); + assert.equal(turn.body.turn.assistantMessageId, messages.body.messages[1].messageId); + assert.equal(turn.body.turn.trace.eventCount, 2); + assert.equal(turn.body.turn.trace.status, "completed"); + + const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=10`); + assert.equal(trace.status, 200); + assert.equal(trace.body.events.length, 2); + assert.equal(trace.body.traceStatus, "completed"); + assert.equal(trace.body.hasMore, false); + } finally { + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +}); + test("workbench read model does not expose trace-only memory without visible session or result owner", async () => { const traceStore = createCodeAgentTraceStore(); const traceId = "trc_workbench_trace_only"; diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 7b226e8b..619bde01 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -157,7 +157,7 @@ async function handleWorkbenchTurnSnapshot(response, url, options, actor, rawTur return sendJson(response, 403, workbenchError("agent_session_owner_required", "Only the session owner or admin can read this Workbench turn.", { traceId })); } const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId)); - const trace = traceSnapshot(options, traceId); + const trace = await traceSnapshot(options, traceId); const status = turnStatus(result, session, trace); const found = Boolean(resultVisible || session); if (!found) return sendJson(response, 404, workbenchError("workbench_turn_not_found", "Workbench turn is not visible to the current actor.", { turnId, traceId })); @@ -183,7 +183,8 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT if (!resultVisible && !session) { return sendJson(response, 404, workbenchError("workbench_trace_not_found", "Workbench trace is not visible to the current actor.", { traceId })); } - const page = traceEventPage(traceSnapshot(options, traceId), tracePageOptions(url)); + const trace = await traceSnapshot(options, traceId); + const page = traceEventPage(trace, tracePageOptions(url)); sendJson(response, 200, { ok: true, status: "succeeded", @@ -213,7 +214,7 @@ function sessionSummary(session, options) { if (!session?.id) return null; const snapshot = objectValue(session.session); const traceId = safeTraceId(session.lastTraceId ?? snapshot.lastTraceId ?? snapshot.currentTraceId) ?? null; - const trace = traceId ? traceSnapshot(options, traceId) : null; + const trace = traceId ? traceSnapshotSync(options, traceId) : null; const messages = sessionMessages(session); const status = normalizeStatus(snapshot.sessionStatus ?? session.status ?? trace?.status); return { @@ -275,7 +276,9 @@ function sessionMessages(session) { const snapshot = objectValue(session.session); const raw = Array.isArray(snapshot.messages) ? snapshot.messages : Array.isArray(snapshot.chatMessages) ? snapshot.chatMessages : []; const messages = raw.map((message, index) => messageFact(message, index, session, snapshot)).filter(Boolean); - if (!messages.some((message) => message.role === "assistant") && textValue(snapshot.finalResponse)) { + const finalTraceId = safeTraceId(session.lastTraceId ?? snapshot.lastTraceId ?? snapshot.currentTraceId) ?? null; + const hasAssistantLikeFinal = messages.some((message) => isAssistantLikeRole(message.role) && (!finalTraceId || message.traceId === finalTraceId)); + if (!hasAssistantLikeFinal && textValue(snapshot.finalResponse)) { messages.push(messageFact({ role: "assistant", text: snapshot.finalResponse, traceId: session.lastTraceId ?? snapshot.lastTraceId, source: "finalResponse" }, messages.length, session, snapshot)); } return messages; @@ -297,7 +300,7 @@ function messageFact(message, index, session, snapshot) { conversationId: safeConversationId(session.conversationId) ?? null, traceId, turnId: safeTurnId(message?.turnId) || traceId, - status: normalizeStatus(message?.status ?? (role === "assistant" ? session.status : "completed")), + status: normalizeStatus(message?.status ?? (isAssistantLikeRole(role) ? session.status : "completed")), parts, textPreview: text ? text.slice(0, 240) : null, createdAt: textValue(message?.createdAt ?? message?.timestamp ?? message?.at) || session.startedAt || session.updatedAt || null, @@ -325,7 +328,7 @@ function partFact(part, index, messageId, traceId) { function turnSnapshot({ turnId, traceId, status, result, session, trace }) { const messages = session ? sessionMessages(session) : []; const userMessage = messages.find((message) => message.role === "user") ?? null; - const assistantMessage = [...messages].reverse().find((message) => message.role === "assistant") ?? null; + const assistantMessage = [...messages].reverse().find((message) => isAssistantLikeRole(message.role)) ?? null; return { turnId, traceId, @@ -391,10 +394,61 @@ function tracePageOptions(url) { return { afterSeq, limit: boundedLimit(url.searchParams.get("limit")) }; } -function traceSnapshot(options, traceId) { +async function traceSnapshot(options, traceId) { + const memory = traceSnapshotSync(options, traceId); + if (memory?.status !== "missing" && (memory.eventCount > 0 || memory.lastEvent)) return memory; + const durable = await durableTraceSnapshot(options, traceId); + return durable ?? memory; +} + +function traceSnapshotSync(options, traceId) { return (options.traceStore ?? defaultCodeAgentTraceStore).snapshot(traceId); } +async function durableTraceSnapshot(options, traceId) { + const store = options.runtimeStore; + if (!store || typeof store.queryAgentTraceEvents !== "function") return null; + let result; + try { + result = await store.queryAgentTraceEvents({ traceId }); + } catch { + return null; + } + const events = Array.isArray(result?.events) ? result.events : []; + if (events.length === 0) return null; + const normalizedEvents = events.map((event, index) => ({ ...event, traceId, seq: eventSeq(event, index) })); + const firstEvent = normalizedEvents[0] ?? null; + const lastEvent = normalizedEvents.at(-1) ?? null; + const status = durableTraceStatus(normalizedEvents); + return { + traceId, + status, + createdAt: firstEvent?.createdAt ?? null, + updatedAt: lastEvent?.createdAt ?? firstEvent?.createdAt ?? null, + startedAt: firstEvent?.createdAt ?? null, + finishedAt: TERMINAL_STATUSES.has(status) ? lastEvent?.createdAt ?? null : null, + eventCount: normalizedEvents.length, + events: normalizedEvents, + assistantStreams: [], + eventLabels: normalizedEvents.map((event) => event.label).filter(Boolean), + lastEvent, + outputTruncated: normalizedEvents.some((event) => event.outputTruncated === true), + valuesPrinted: false + }; +} + +function durableTraceStatus(events) { + const lastEvent = events.at(-1) ?? null; + const normalized = normalizeStatus(lastEvent?.status ?? lastEvent?.type); + if (lastEvent?.terminal === true) { + if (normalized === "observed" || normalized === "event" || normalized === "unknown") return "completed"; + return normalized; + } + if (TERMINAL_STATUSES.has(normalized) || RUNNING_STATUSES.has(normalized)) return normalized; + if (["failed", "error", "timeout"].includes(normalized)) return normalized; + return "running"; +} + function turnStatus(result, session, trace) { return normalizeStatus( result?.status ?? @@ -469,6 +523,10 @@ function firstUserPreview(messages) { return messages.find((message) => message.role === "user")?.textPreview ?? null; } +function isAssistantLikeRole(role) { + return role === "assistant" || role === "agent"; +} + function objectValue(value) { return value && typeof value === "object" && !Array.isArray(value) ? value : {}; } diff --git a/internal/cloud/server.ts b/internal/cloud/server.ts index c8160962..d162e4c3 100644 --- a/internal/cloud/server.ts +++ b/internal/cloud/server.ts @@ -27,7 +27,7 @@ import { describeCodeAgentAvailability, handleCodeAgentChat } from "./code-agent-chat.ts"; -import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; +import { createDurableCodeAgentTraceStore, defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; import { createCodeAgentSessionRegistry } from "./code-agent-session-registry.ts"; import { codeAgentSessionLifecycleSummary } from "./code-agent-session-lifecycle.ts"; import { createCodexStdioSessionManager } from "./codex-stdio-session.ts"; @@ -94,13 +94,16 @@ export function createCloudApiServer(options = {}) { const env = options.env ?? process.env; ensureCodeAgentRuntimeBase(env); const sessionRegistry = options.sessionRegistry || createCodeAgentSessionRegistry(); - const traceStore = options.traceStore || defaultCodeAgentTraceStore; + const runtimeStore = options.runtimeStore || createConfiguredCloudRuntimeStore({ ...options, env }); + const traceStore = createDurableCodeAgentTraceStore({ + traceStore: options.traceStore || defaultCodeAgentTraceStore, + traceEventStore: runtimeStore + }); const codexStdioManager = options.codexStdioManager || createCodexStdioSessionManager({ traceStore }); const codeAgentChatResults = options.codeAgentChatResults || createCodeAgentChatResultStore({ maxResults: parsePositiveInteger(env.HWLAB_CODE_AGENT_CHAT_RESULT_CACHE_SIZE, 256) }); const webPerformanceStore = options.webPerformanceStore || createWebPerformanceStore({ env }); - const runtimeStore = options.runtimeStore || createConfiguredCloudRuntimeStore({ ...options, env }); const gatewayRegistry = options.gatewayRegistry || createGatewayDemoRegistry({ staleMs: parsePositiveInteger(env.HWLAB_GATEWAY_DEMO_STALE_MS, 30000), dispatchTimeoutMs: parsePositiveInteger(env.HWLAB_GATEWAY_DISPATCH_TIMEOUT_MS, 120000) diff --git a/internal/db/runtime-store.test.ts b/internal/db/runtime-store.test.ts index 7160a84a..ec15734d 100644 --- a/internal/db/runtime-store.test.ts +++ b/internal/db/runtime-store.test.ts @@ -605,6 +605,46 @@ test("configured postgres runtime persists and queries records through query cli 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("INSERT INTO agent_trace_events"))); +}); + function createFakePostgresClient({ migrationReady = true, migrationErrorCode = null, @@ -618,6 +658,8 @@ function createFakePostgresClient({ hardware_operations: new Map(), audit_events: new Map(), evidence_records: new Map(), + worker_sessions: new Map(), + agent_trace_events: new Map(), hwlab_schema_migrations: new Map() }; if (migrationReady) { @@ -670,6 +712,19 @@ function createFakePostgresClient({ } 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("WHERE trace_id = $1") ? params[0] : null; + const rows = [...state.agent_trace_events.values()] + .filter((record) => !traceId || record.trace_id === traceId) + .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 metadata_json FROM evidence_records")) { if (readErrorCode) { const error = new Error("evidence read query failed"); @@ -702,6 +757,19 @@ function createFakePostgresClient({ 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: [] }; + } throw new Error(`unexpected sql: ${sql}`); } }; diff --git a/internal/db/runtime-store.ts b/internal/db/runtime-store.ts index cc2af0b3..4c3017d9 100644 --- a/internal/db/runtime-store.ts +++ b/internal/db/runtime-store.ts @@ -1,3 +1,7 @@ +/* + * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0 + * 职责: Cloud runtime durable store。Code Agent trace/session 投影事实必须从同一持久化适配器写入和恢复。 + */ import { createHash, randomUUID } from "node:crypto"; import { @@ -45,7 +49,9 @@ const postgresCountTables = Object.freeze([ "audit_events", "evidence_records", "agent_sessions", - "account_workspaces" + "account_workspaces", + "worker_sessions", + "agent_trace_events" ]); export function createCloudRuntimeStore(options = {}) { @@ -90,6 +96,7 @@ export class CloudRuntimeStore { this.hardwareOperations = new Map(); this.auditEvents = new Map(); this.evidenceRecords = new Map(); + this.agentTraceEvents = new Map(); } summary() { @@ -110,7 +117,8 @@ export class CloudRuntimeStore { boxCapabilities: this.boxCapabilities.size, hardwareOperations: this.hardwareOperations.size, auditEvents: this.auditEvents.size, - evidenceRecords: this.evidenceRecords.size + evidenceRecords: this.evidenceRecords.size, + agentTraceEvents: this.agentTraceEvents.size } }); } @@ -473,6 +481,32 @@ export class CloudRuntimeStore { }; } + writeAgentTraceEvent(params = {}, requestMeta = {}) { + const traceEvent = normalizeAgentTraceEvent(params.traceEvent ?? params.event ?? params, requestMeta, this.now()); + this.agentTraceEvents.set(traceEvent.id, traceEvent); + return { + written: true, + traceEvent, + persistence: this.summary() + }; + } + + queryAgentTraceEvents(params = {}) { + const records = [...this.agentTraceEvents.values()].filter((record) => matchesQuery(record, params, [ + "id", + "traceId", + "agentSessionId", + "workerSessionId", + "level" + ])); + records.sort(compareTraceEventRecords); + return { + events: limitResults(records.map((record) => record.event), params.limit), + count: records.length, + persistence: this.summary() + }; + } + recordAuditEvent(input) { return this.writeAuditEvent(this.buildAuditEventFromInput(input, input.requestMeta), input.requestMeta).auditEvent; } @@ -901,6 +935,34 @@ export class PostgresCloudRuntimeStore { }; } + async writeAgentTraceEvent(params = {}, requestMeta = {}) { + const readiness = this.summary(); + if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) { + await this.assertReadyForWrites(); + } + const result = this.memory.writeAgentTraceEvent(params, requestMeta); + await this.persistAgentTraceEvent(result.traceEvent); + return withPersistence(result, this.summary()); + } + + async queryAgentTraceEvents(params = {}) { + await this.assertReadyForDurableReads("agent.trace-events.query"); + const traceId = textOr(params.traceId, ""); + const sql = traceId + ? "SELECT event_json FROM agent_trace_events WHERE trace_id = $1 ORDER BY occurred_at ASC, id ASC" + : "SELECT event_json FROM agent_trace_events ORDER BY occurred_at ASC, id ASC"; + const result = await this.queryDurableReadRows("agent.trace-events.query", sql, traceId ? [traceId] : []); + const events = result.rows + .map((row) => parseJsonColumn(row.event_json, null)) + .filter(Boolean) + .filter((event) => matchesQuery(event, params, ["traceId", "sessionId", "workerSessionId", "level"])); + return { + events: limitResults(events, params.limit), + count: events.length, + persistence: this.summary() + }; + } + async hydrateOperationRefs(params = {}) { await this.hydrateGatewaySession(params.gatewaySessionId); await this.hydrateBoxResource(params.resourceId); @@ -959,6 +1021,9 @@ export class PostgresCloudRuntimeStore { for (const record of newRecords(this.memory.evidenceRecords, before.evidenceRecords)) { await this.persistEvidenceRecord(record); } + for (const record of newRecords(this.memory.agentTraceEvents, before.agentTraceEvents)) { + await this.persistAgentTraceEvent(record); + } } async assertReadyForWrites() { @@ -1089,6 +1154,22 @@ export class PostgresCloudRuntimeStore { ); } + async persistAgentTraceEvent(record) { + await this.query( + "INSERT INTO agent_trace_events (id, trace_id, agent_session_id, worker_session_id, level, message, event_json, occurred_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (id) DO UPDATE SET trace_id = EXCLUDED.trace_id, agent_session_id = EXCLUDED.agent_session_id, worker_session_id = EXCLUDED.worker_session_id, level = EXCLUDED.level, message = EXCLUDED.message, event_json = EXCLUDED.event_json, occurred_at = EXCLUDED.occurred_at", + [ + record.id, + record.traceId, + record.agentSessionId, + record.workerSessionId, + record.level, + record.message, + stableJson(record.event), + record.occurredAt + ] + ); + } + async readCounts() { const counts = {}; for (const table of postgresCountTables) { @@ -1238,6 +1319,91 @@ function normalizeShellInput(params) { }; } +function normalizeAgentTraceEvent(value, requestMeta = {}, now) { + const input = normalizeJsonObject(value); + const traceId = textOr(input.traceId ?? requestMeta.traceId, ""); + if (!traceId) { + throw new HwlabProtocolError("agent trace event requires traceId", { + code: ERROR_CODES.invalidParams, + data: { required: ["traceId"] } + }); + } + const seq = positiveIntegerOrNull(input.seq); + const occurredAt = timestampOr(input.occurredAt ?? input.createdAt ?? input.timestamp, now); + const status = textOr(input.status, "observed"); + const type = textOr(input.type ?? input.kind, "event"); + const level = traceEventLevel(input.level ?? status ?? type); + const message = textOr(input.message ?? input.label ?? type, ""); + const agentSessionId = textOr(input.agentSessionId ?? input.sessionId ?? requestMeta.agentSessionId ?? requestMeta.sessionId, ""); + const workerSessionId = textOr(input.workerSessionId ?? input.workerSession?.id ?? input.workerId, ""); + const id = textOr(input.id ?? input.eventId, "") || `tev_${sha256Hex({ + traceId, + seq, + source: input.source ?? null, + sourceSeq: input.sourceSeq ?? null, + type, + status, + label: input.label ?? null, + message, + occurredAt + }).slice(0, 32)}`; + const event = pruneUndefined({ + ...input, + traceId, + seq: seq ?? undefined, + agentSessionId: agentSessionId || undefined, + workerSessionId: workerSessionId || undefined, + level, + createdAt: timestampOr(input.createdAt ?? occurredAt, occurredAt), + valuesPrinted: false + }); + return { + id, + traceId, + agentSessionId: agentSessionId || null, + workerSessionId: workerSessionId || null, + level, + message, + occurredAt, + event + }; +} + +function compareTraceEventRecords(left, right) { + const leftSeq = positiveIntegerOrNull(left?.event?.seq); + const rightSeq = positiveIntegerOrNull(right?.event?.seq); + if (leftSeq && rightSeq && leftSeq !== rightSeq) return leftSeq - rightSeq; + return String(left?.occurredAt ?? "").localeCompare(String(right?.occurredAt ?? "")) || + String(left?.id ?? "").localeCompare(String(right?.id ?? "")); +} + +function traceEventLevel(value) { + const text = textOr(value, "info").toLowerCase(); + if (["failed", "failure", "error", "timeout"].includes(text)) return "error"; + if (["warn", "warning", "blocked", "canceled", "cancelled"].includes(text)) return "warn"; + if (["debug", "trace"].includes(text)) return "debug"; + return "info"; +} + +function positiveIntegerOrNull(value) { + const parsed = Number.parseInt(value ?? "", 10); + return Number.isInteger(parsed) && parsed > 0 ? parsed : null; +} + +function timestampOr(value, fallback) { + const text = textOr(value, ""); + if (text) { + const ms = Date.parse(text); + if (Number.isFinite(ms)) return new Date(ms).toISOString(); + } + return textOr(fallback, "") || new Date().toISOString(); +} + +function textOr(value, fallback) { + const text = String(value ?? "").trim(); + return text || fallback; +} + function asObject(value, label) { if (!value || typeof value !== "object" || Array.isArray(value)) { throw new HwlabProtocolError(`${label} must be a JSON object`, { @@ -1720,7 +1886,8 @@ function snapshotMemory(memory) { boxCapabilities: new Map(memory.boxCapabilities), hardwareOperations: new Map(memory.hardwareOperations), auditEvents: new Map(memory.auditEvents), - evidenceRecords: new Map(memory.evidenceRecords) + evidenceRecords: new Map(memory.evidenceRecords), + agentTraceEvents: new Map(memory.agentTraceEvents) }; } @@ -1764,6 +1931,8 @@ function toCountKey(table) { if (table === "evidence_records") return "evidenceRecords"; if (table === "agent_sessions") return "agentSessions"; if (table === "account_workspaces") return "accountWorkspaces"; + if (table === "worker_sessions") return "workerSessions"; + if (table === "agent_trace_events") return "agentTraceEvents"; return table; } diff --git a/internal/db/schema.ts b/internal/db/schema.ts index 3d8b2f5b..7c019f32 100644 --- a/internal/db/schema.ts +++ b/internal/db/schema.ts @@ -1,3 +1,7 @@ +/* + * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0 + * 职责: Cloud runtime durable schema contract。Workbench/Code Agent 投影表必须纳入启动 readiness 和迁移一致性检查。 + */ import { TABLES } from "../protocol/index.mjs"; export const CLOUD_CORE_MIGRATION_ID = "0001_cloud_core_skeleton"; @@ -119,6 +123,24 @@ export const CLOUD_RUNTIME_DURABLE_TABLE_COLUMNS = Object.freeze({ "updated_by_client", "created_at", "updated_at" + ]), + worker_sessions: Object.freeze([ + "id", + "agent_session_id", + "worker_id", + "status", + "started_at", + "ended_at" + ]), + agent_trace_events: Object.freeze([ + "id", + "trace_id", + "agent_session_id", + "worker_session_id", + "level", + "message", + "event_json", + "occurred_at" ]) });