diff --git a/deploy/deploy.yaml b/deploy/deploy.yaml index d5ce2bc3..caf2e0fa 100644 --- a/deploy/deploy.yaml +++ b/deploy/deploy.yaml @@ -536,6 +536,7 @@ lanes: HWLAB_CODE_AGENT_AGENTRUN_DISPATCH_RETRY_MAX: "5" HWLAB_CODE_AGENT_AGENTRUN_DISPATCH_RETRY_BASE_MS: "1000" HWLAB_CODE_AGENT_AGENTRUN_DISPATCH_RETRY_MAX_DELAY_MS: "30000" + HWLAB_USER_BILLING_CODE_AGENT_ENABLED: "0" observable: true hwlab-workbench-runtime: runtimeKind: go-service @@ -1028,7 +1029,6 @@ lanes: HWLAB_HWPOD_NODE_WS_TOKEN: secretRef:hwlab-v03-hwpod-node-auth/token HWLAB_USER_BILLING_LOGIN_RETRY_ATTEMPTS: "2" HWLAB_USER_BILLING_LOGIN_RETRY_DELAY_MS: "250" - HWLAB_USER_BILLING_CODE_AGENT_ENABLED: "1" HWLAB_USER_BILLING_CODE_AGENT_ESTIMATED_CREDITS: "1" HWLAB_USER_BILLING_CODE_AGENT_USED_CREDITS: "1" HWLAB_KEYCLOAK_ISSUER: https://auth.74-48-78-17.nip.io/realms/hwlab @@ -1329,7 +1329,6 @@ lanes: HWLAB_TASKTREE_URL: http://hwlab-tasktree-api.hwlab-production.svc.cluster.local:6673 HWLAB_USER_BILLING_LOGIN_RETRY_ATTEMPTS: "2" HWLAB_USER_BILLING_LOGIN_RETRY_DELAY_MS: "250" - HWLAB_USER_BILLING_CODE_AGENT_ENABLED: "1" HWLAB_USER_BILLING_CODE_AGENT_ESTIMATED_CREDITS: "1" HWLAB_USER_BILLING_CODE_AGENT_USED_CREDITS: "1" OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: http://otel-collector.platform-infra.svc.cluster.local:4318/v1/traces diff --git a/internal/cloud/kafka-event-bridge.ts b/internal/cloud/kafka-event-bridge.ts index 0c6eb0c6..816d095c 100644 --- a/internal/cloud/kafka-event-bridge.ts +++ b/internal/cloud/kafka-event-bridge.ts @@ -433,6 +433,22 @@ function startLiveHwlabKafkaEventBridge({ config, env = process.env, logger = co valuesRedacted: true }; }, + workbenchTurnSnapshot(traceId) { + return sessionIndex?.turnSnapshot?.(traceId) ?? { + ok: false, + turn: null, + warning: { code: "workbench_kafka_session_index_disabled", message: "Turn projection index is disabled.", blocking: false, valuesRedacted: true }, + valuesRedacted: true + }; + }, + workbenchTraceSnapshot(traceId) { + return sessionIndex?.traceSnapshot?.(traceId) ?? { + ok: false, + trace: null, + warning: { code: "workbench_kafka_session_index_disabled", message: "Trace projection index is disabled.", blocking: false, valuesRedacted: true }, + valuesRedacted: true + }; + }, liveSubscriberCount() { return subscribers.size; }, valuesPrinted: false }; @@ -516,6 +532,22 @@ function combineKafkaEventBridgeComponents(config, components) { valuesRedacted: true }; }, + workbenchTurnSnapshot(traceId) { + return liveOwner?.workbenchTurnSnapshot?.(traceId) ?? { + ok: false, + turn: null, + warning: { code: "workbench_kafka_session_index_disabled", message: "Turn projection index is disabled.", blocking: false, valuesRedacted: true }, + valuesRedacted: true + }; + }, + workbenchTraceSnapshot(traceId) { + return liveOwner?.workbenchTraceSnapshot?.(traceId) ?? { + ok: false, + trace: null, + warning: { code: "workbench_kafka_session_index_disabled", message: "Trace projection index is disabled.", blocking: false, valuesRedacted: true }, + valuesRedacted: true + }; + }, liveSubscriberCount() { const owner = components.find((component) => typeof component.liveSubscriberCount === "function"); return owner?.liveSubscriberCount() ?? 0; diff --git a/internal/cloud/workbench-kafka-session-index.test.ts b/internal/cloud/workbench-kafka-session-index.test.ts index 4b4c9f25..afb47e9b 100644 --- a/internal/cloud/workbench-kafka-session-index.test.ts +++ b/internal/cloud/workbench-kafka-session-index.test.ts @@ -277,6 +277,64 @@ test("Kafka rail projection follows actual turn delivery when a requested steer }); }); +test("Kafka retention index materializes sealed turn and trace projections before GET reads", async () => { + const sessionId = "ses_projection"; + const traceId = "trc_projection"; + const eventRecord = (offset: number, event: Record) => ({ + topic: "hwlab.event.v1", + partition: 0, + offset: String(offset), + key: sessionId, + value: { + schema: "hwlab.event.v1", + sessionId, + traceId, + sourceEventId: `src_projection_${offset}`, + event: { sessionId, traceId, sourceSeq: offset + 1, createdAt: `2026-07-20T00:00:0${offset + 1}.000Z`, ...event } + } + }); + const index = createWorkbenchKafkaSessionIndex({ + topic: "hwlab.event.v1", + maxEvents: 20, + maxEventsPerSession: 10, + bootstrapTimeoutMs: 5000, + rebuildIntervalMs: 60000, + rebuildCooldownMs: 1000, + scanLimit: 100, + logger: quietLogger(), + queryAll: async () => completeResult([ + eventRecord(0, { type: "user", text: "projection prompt" }), + eventRecord(1, { type: "assistant", text: "projection reply" }), + eventRecord(2, { type: "result", eventType: "terminal", status: "completed", terminalStatus: "completed", terminal: true }) + ] as ReturnType[], "3") + }); + + expect(await index.start()).toBe(true); + expect(index.turnSnapshot(traceId)).toMatchObject({ + ok: true, + turn: { + traceId, + sessionId, + status: "completed", + terminal: true, + projectionStatus: "caught-up", + finalResponse: { text: "projection reply", status: "completed", traceId } + } + }); + expect(index.traceSnapshot(traceId)).toMatchObject({ + ok: true, + trace: { + traceId, + status: "completed", + terminal: true, + eventCount: 3, + finalResponse: { text: "projection reply" }, + hasMore: false + } + }); + index.stop(); +}); + function completeResult(events: ReturnType[], endOffset: string) { return { completionReason: "end-offset", diff --git a/internal/cloud/workbench-kafka-session-index.ts b/internal/cloud/workbench-kafka-session-index.ts index 73abf52f..a649e717 100644 --- a/internal/cloud/workbench-kafka-session-index.ts +++ b/internal/cloud/workbench-kafka-session-index.ts @@ -3,6 +3,8 @@ import { createHash } from "node:crypto"; +import { buildWorkbenchProjectionEventFacts } from "./workbench-projection-writer.ts"; + export function createWorkbenchKafkaSessionIndex(options = {}) { const topic = text(options.topic); const maxEvents = positiveInteger(options.maxEvents); @@ -263,15 +265,85 @@ export function createWorkbenchKafkaSessionIndex(options = {}) { return { ok: true, sessions, warning: null, status: status(), valuesRedacted: true }; } + function turnSnapshot(traceIdValue) { + const traceId = text(traceIdValue); + if (!traceId) return { ok: false, turn: null, warning: warning("workbench_kafka_turn_trace_required", "Turn projection requires traceId."), status: status(), valuesRedacted: true }; + const turn = state.turnByTrace.get(traceId) ?? null; + const sessionIdValue = text(turn?.sessionId); + if (!ready || (sessionIdValue && state.incompleteSessions.has(sessionIdValue))) { + return { ok: false, turn: null, warning: warning("workbench_kafka_turn_projection_unavailable", "Turn projection is not complete in the Kafka retention index."), status: status(), valuesRedacted: true }; + } + const checkpoint = state.checkpointByTrace.get(traceId) ?? null; + const events = state.traceByTrace.get(traceId) ?? []; + return { + ok: true, + turn: turn ? { + ...turn, + eventCount: events.length, + lastProjectedSeq: checkpoint?.projectedSeq ?? turn.projectedSeq ?? null, + sourceRunId: checkpoint?.runId ?? null, + sourceCommandId: checkpoint?.commandId ?? null, + projectionStatus: checkpoint?.projectionStatus === "caught_up" ? "caught-up" : checkpoint?.projectionStatus ?? "projecting", + projectionHealth: checkpoint?.projectionHealth ?? "healthy", + blocker: null, + valuesRedacted: true + } : null, + checkpoint, + warning: null, + status: status(), + valuesRedacted: true + }; + } + + function traceSnapshot(traceIdValue) { + const traceId = text(traceIdValue); + if (!traceId) return { ok: false, trace: null, warning: warning("workbench_kafka_trace_required", "Trace projection requires traceId."), status: status(), valuesRedacted: true }; + const turnResult = turnSnapshot(traceId); + if (!turnResult.ok) return { ...turnResult, trace: null }; + const events = state.traceByTrace.get(traceId) ?? []; + const turn = turnResult.turn; + return { + ok: true, + trace: turn || events.length > 0 ? { + traceId, + sessionId: turn?.sessionId ?? events[0]?.sessionId ?? null, + turnId: turn?.turnId ?? traceId, + status: turn?.status ?? "running", + terminal: turn?.terminal === true, + finalResponse: turn?.finalResponse ?? null, + eventCount: events.length, + events: events.map((event) => ({ ...event })), + range: { + firstProjectedSeq: events[0]?.projectedSeq ?? null, + lastProjectedSeq: events.at(-1)?.projectedSeq ?? null, + count: events.length, + valuesRedacted: true + }, + hasMore: false, + nextCursor: null, + projectionStatus: turn?.projectionStatus ?? "projecting", + projectionHealth: turn?.projectionHealth ?? "healthy", + valuesRedacted: true + } : null, + warning: null, + status: status(), + valuesRedacted: true + }; + } + function emptyState() { return { records: [], byTransport: new Map(), bySession: new Map(), railBySession: new Map(), + turnByTrace: new Map(), + checkpointByTrace: new Map(), + traceByTrace: new Map(), incompleteSessions: new Set(), endOffsets: new Map(), - globalComplete: true + globalComplete: true, + projectedSeq: 0 }; } @@ -281,6 +353,7 @@ export function createWorkbenchKafkaSessionIndex(options = {}) { target.byTransport.set(transportKey, record); target.records.push(record); updateEndOffset(target.endOffsets, record.partition, record.offset); + applyWorkbenchKafkaProjectionRecord(target, record); const id = indexedSessionId(record); if (id) { const bucket = target.bySession.get(id) ?? []; @@ -334,6 +407,8 @@ export function createWorkbenchKafkaSessionIndex(options = {}) { observeLive, query, sessionSummaries, + turnSnapshot, + traceSnapshot, status, rebuild, valuesPrinted: false @@ -351,6 +426,47 @@ export function workbenchKafkaRailSummary(sessionIdValue, recordsValue) { return summary; } +function applyWorkbenchKafkaProjectionRecord(target, record) { + const envelope = record?.value; + const sourceEvent = envelope?.event && typeof envelope.event === "object" ? envelope.event : null; + const traceId = firstText(envelope?.traceId, sourceEvent?.traceId, envelope?.context?.traceId); + if (!sourceEvent || !traceId) return; + const sessionIdValue = firstText(envelope?.sessionId, envelope?.hwlabSessionId, sourceEvent?.sessionId, sourceEvent?.hwlabSessionId, envelope?.context?.sessionId); + const event = { + ...sourceEvent, + traceId, + sessionId: sessionIdValue, + turnId: firstText(envelope?.turnId, sourceEvent?.turnId, traceId), + sourceSeq: sourceEvent?.sourceSeq ?? envelope?.sourceSeq ?? sourceEvent?.seq, + sourceEventId: firstText(envelope?.sourceEventId, sourceEvent?.sourceEventId), + createdAt: firstText(sourceEvent?.createdAt, envelope?.createdAt, record?.timestamp) + }; + const projectedSeq = ++target.projectedSeq; + const previousCheckpoint = target.checkpointByTrace.get(traceId) ?? null; + try { + const projection = buildWorkbenchProjectionEventFacts({ + event, + requestMeta: { traceId, sessionId: sessionIdValue }, + previousCheckpoint, + projectedSeq, + projectedAt: event.createdAt ?? new Date().toISOString() + }); + if (projection?.written !== true) return; + const turn = projection.facts?.turns?.[0] ?? null; + const checkpoint = projection.facts?.checkpoints?.[0] ?? null; + const traceEvent = projection.facts?.traceEvents?.[0] ?? null; + if (turn) target.turnByTrace.set(traceId, turn); + if (checkpoint) target.checkpointByTrace.set(traceId, checkpoint); + if (traceEvent) { + const events = target.traceByTrace.get(traceId) ?? []; + events.push(traceEvent); + target.traceByTrace.set(traceId, events); + } + } catch (error) { + target.projectionWarning = warning(error?.code ?? "workbench_kafka_turn_projection_failed", errorMessage(error)); + } +} + function applyWorkbenchKafkaRailRecord(previous, record) { const id = text(previous?.sessionId); if (!id || indexedSessionId(record) !== id || !recordIsNewerThanRailRevision(record, previous.revision)) return previous; diff --git a/internal/workbench/http.ts b/internal/workbench/http.ts index 2157f455..97d74bc6 100644 --- a/internal/workbench/http.ts +++ b/internal/workbench/http.ts @@ -39,6 +39,12 @@ export function createWorkbenchHttpApp(options: { dispatch: (command: WorkbenchC if (sessionMatch && request.method === "GET") return nativeSessionDetail(options, decodeURIComponent(sessionMatch[1]), sessionMatch[2] === "messages"); const turnMatch = /^\/v1\/workbench\/turns\/([^/]+)$/u.exec(url.pathname); if (turnMatch && request.method === "GET") return nativeTurn(options, decodeURIComponent(turnMatch[1])); + const agentTurnMatch = /^\/v1\/agent\/turns\/([^/]+)$/u.exec(url.pathname); + if (agentTurnMatch && request.method === "GET") return nativeTurn(options, decodeURIComponent(agentTurnMatch[1])); + const agentResultMatch = /^\/v1\/agent\/chat\/result\/([^/]+)$/u.exec(url.pathname); + if (agentResultMatch && request.method === "GET") return nativeTurn(options, decodeURIComponent(agentResultMatch[1])); + const agentTraceMatch = /^\/v1\/agent\/traces\/([^/]+)$/u.exec(url.pathname); + if (agentTraceMatch && request.method === "GET") return nativeTrace(options, decodeURIComponent(agentTraceMatch[1])); } if (url.pathname === "/v1/workbench/commands" && request.method === "POST") { requireAuthorization(request, options); @@ -167,10 +173,34 @@ async function nativeSessionDetail(options: { snapshot?: () => Promise; mod if (!session) return json(404, { ok: false, error: { code: "session_not_found", message: "native session was not found" } }); return json(200, messagesOnly ? { ok: true, sessionId, messages: session.messages ?? [], count: Array.isArray(session.messages) ? session.messages.length : 0, mode: options.mode ?? "native-test" } : { ok: true, session, mode: options.mode ?? "native-test" }); } -async function nativeTurn(options: { snapshot?: () => Promise; mode?: WorkbenchMode }, traceId: string) { +async function nativeTurn(options: { snapshot?: () => Promise; mode?: WorkbenchMode; kafkaEventBridge?: any }, traceId: string) { const state = await requiredSnapshot(options); const turn = state.turns[traceId]; - return turn ? json(200, { ok: true, ...turn, mode: options.mode ?? "native-test" }) : json(404, { ok: false, error: { code: "turn_not_found", message: "native turn was not found" } }); + const projected = options.mode === "agentrun-native" ? options.kafkaEventBridge?.workbenchTurnSnapshot?.(traceId) : null; + if (!turn && !projected?.turn) return json(404, { ok: false, error: { code: "turn_not_found", message: "native turn was not found" } }); + return json(200, { + ok: true, + ...(turn ?? {}), + ...(projected?.turn ?? {}), + projectionStatus: projected?.turn?.projectionStatus ?? (projected?.ok === false ? "degraded" : turn?.terminal === true ? "caught-up" : "projecting"), + projectionWarning: projected?.warning ?? null, + mode: options.mode ?? "native-test" + }); +} +function nativeTrace(options: { mode?: WorkbenchMode; kafkaEventBridge?: any }, traceId: string) { + const projected = options.mode === "agentrun-native" ? options.kafkaEventBridge?.workbenchTraceSnapshot?.(traceId) : null; + if (!projected?.trace) { + return json(404, { + ok: false, + error: { + code: projected?.warning?.code ?? "trace_not_found", + message: projected?.warning?.message ?? "native trace was not found" + }, + projectionStatus: projected?.ok === false ? "degraded" : "unknown", + warning: projected?.warning ?? null + }); + } + return json(200, { ok: true, ...projected.trace, mode: options.mode ?? "native-test" }); } async function requiredSnapshot(options: { snapshot?: () => Promise }) { if (!options.snapshot) throw Object.assign(new Error("native projection is unavailable"), { code: "native_projection_unavailable" }); return options.snapshot(); } async function nativeEventStream(options: { mode?: WorkbenchMode; kafkaEventBridge?: any }, url: URL) { diff --git a/internal/workbench/workbench.test.ts b/internal/workbench/workbench.test.ts index 7243d3fb..f132f39e 100644 --- a/internal/workbench/workbench.test.ts +++ b/internal/workbench/workbench.test.ts @@ -229,6 +229,56 @@ describe("Workbench Temporal activities", () => { }); describe("Workbench native HTTP adapter", () => { + test("native Agent compatibility routes read the Kafka-materialized turn and trace projection", async () => { + const traceId = "trc_native_projection"; + const projectedTurn = { + traceId, + sessionId: "ses_native_projection", + status: "completed", + terminal: true, + finalResponse: { text: "native projection reply", status: "completed", traceId }, + projectionStatus: "caught-up" + }; + const app = createWorkbenchHttpApp({ + mode: "agentrun-native", + async dispatch() { return { ok: true }; }, + async snapshot() { + return { + sessions: {}, + turns: { [traceId]: { traceId, sessionId: "ses_native_projection", status: "running", terminal: false } } + }; + }, + kafkaEventBridge: { + workbenchTurnSnapshot(value: string) { + expect(value).toBe(traceId); + return { ok: true, turn: projectedTurn, warning: null }; + }, + workbenchTraceSnapshot(value: string) { + expect(value).toBe(traceId); + return { + ok: true, + trace: { + ...projectedTurn, + eventCount: 1, + events: [{ traceId, projectedSeq: 7, type: "result", status: "completed", terminal: true }], + hasMore: false, + nextCursor: null + } + }; + } + } + }); + + for (const pathname of [`/v1/workbench/turns/${traceId}`, `/v1/agent/turns/${traceId}`, `/v1/agent/chat/result/${traceId}`]) { + const response = await app.fetch(new Request(`http://native.test${pathname}`)); + expect(response.status).toBe(200); + expect(await response.json()).toMatchObject(projectedTurn); + } + const traceResponse = await app.fetch(new Request(`http://native.test/v1/agent/traces/${traceId}`)); + expect(traceResponse.status).toBe(200); + expect(await traceResponse.json()).toMatchObject({ traceId, status: "completed", terminal: true, eventCount: 1, finalResponse: { text: "native projection reply" } }); + }); + test("native session list exposes the first user preview for summary-only rail loading", async () => { const app = createWorkbenchHttpApp({ async dispatch() { return { ok: true }; },