fix(workbench): expose native Kafka turn projections
This commit is contained in:
root
+1
-2
@@ -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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<string, unknown>) => ({
|
||||
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<typeof record>[], "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<typeof record>[], endOffset: string) {
|
||||
return {
|
||||
completionReason: "end-offset",
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<any>; 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<any>; mode?: WorkbenchMode }, traceId: string) {
|
||||
async function nativeTurn(options: { snapshot?: () => Promise<any>; 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<any> }) { 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) {
|
||||
|
||||
@@ -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 }; },
|
||||
|
||||
Reference in New Issue
Block a user