diff --git a/internal/cloud/server-workbench-realtime-http.ts b/internal/cloud/server-workbench-realtime-http.ts index 867c1229..cd32aa3f 100644 --- a/internal/cloud/server-workbench-realtime-http.ts +++ b/internal/cloud/server-workbench-realtime-http.ts @@ -758,17 +758,34 @@ async function handleKafkaRefreshReplayWorkbenchRealtimeHttp(request, response, liveBufferLimit: refreshReplay.liveBufferLimit, identityWindowLimit: refreshReplay.matchedEventLimit + refreshReplay.liveBufferLimit, subscribeLive: (listener) => bridge.subscribeLiveHwlabEvents(listener), - queryRetention: ({ signal }) => bridge.queryHwlabEventRetention({ - sessionId: requestedSessionId, - traceId: requestedTraceId, - limit: refreshReplay.matchedEventLimit, - scanLimit: refreshReplay.scanLimit, - timeoutMs: refreshReplay.timeoutMs, - groupIdPrefix: refreshReplay.groupIdPrefix, - partitionKey: requestedSessionId, - fromBeginning: true, - signal - }), + queryRetention: async ({ signal }) => { + const result = await bridge.queryHwlabEventRetention({ + sessionId: requestedSessionId, + traceId: requestedTraceId, + limit: refreshReplay.matchedEventLimit, + scanLimit: refreshReplay.scanLimit, + timeoutMs: refreshReplay.timeoutMs, + groupIdPrefix: refreshReplay.groupIdPrefix, + partitionKey: requestedSessionId, + fromBeginning: true, + signal + }); + (options.logger ?? console).log(JSON.stringify({ + code: "workbench-kafka-refresh-query", + component: "hwlab-workbench-realtime", + sessionId: requestedSessionId, + traceId: requestedTraceId, + completionReason: result?.completionReason ?? null, + complete: result?.completion?.complete === true, + scannedCount: result?.scannedCount ?? null, + matchedCount: result?.matchedCount ?? null, + partitionKeyScoped: result?.partitionKeyScoped === true, + targetPartition: result?.targetPartition ?? null, + timing: result?.timing ?? null, + valuesPrinted: false + })); + return result; + }, deliverEvent: (envelope, transport) => enqueueEvent("hwlab.event.v1", envelope, transport), deliverConnected: (summary) => enqueueEvent("workbench.connected", { type: "connected", diff --git a/internal/workbench/workbench.test.ts b/internal/workbench/workbench.test.ts index 51f4eec6..ff460d84 100644 --- a/internal/workbench/workbench.test.ts +++ b/internal/workbench/workbench.test.ts @@ -132,7 +132,20 @@ describe("Workbench native HTTP adapter", () => { ], {}); for (let index = 0; index < 20 && subscribers.size === 0; index += 1) await Bun.sleep(5); for (const listener of subscribers) listener({ schema: "hwlab.event.v1", sessionId: "ses_l0_sse", traceId: "trc_l0_sse", event: { type: "backend", eventType: "backend", label: "agentrun:backend:command-created" } }); - expect(await pending).toMatchObject({ ok: true, operation: "events.inspect", connected: true, eventCount: 1, eventTypes: ["backend"] }); + expect(await pending).toMatchObject({ + ok: true, + operation: "events.inspect", + connected: true, + connectedContract: { + deliverySemantics: "live-only", + realtimeSource: "hwlab.event.v1", + replay: false, + liveOnly: true, + refreshReplay: null + }, + eventCount: 1, + eventTypes: ["backend"] + }); } finally { server.stop(true); await app.close(); diff --git a/tools/src/workbench-cli.ts b/tools/src/workbench-cli.ts index a2a42ace..1226def5 100644 --- a/tools/src/workbench-cli.ts +++ b/tools/src/workbench-cli.ts @@ -89,11 +89,13 @@ async function inspectEvents(parsed: Parsed, env: Record entry.name === "workbench.connected"); + const connectedFrame = frames.find((entry) => entry.name === "workbench.connected") ?? null; + const connected = connectedFrame !== null; + const connectedContract = connectedFrame ? workbenchConnectedContract(connectedFrame.data) : null; const semantics = semanticSummary(business); const missingSemantics = waitFor.filter((semantic) => !semantics.observedSemantics.includes(semantic)); const ok = connected && inspectionSatisfied(business, minEvents, waitFor); - return { ok, operation: "events.inspect", status: ok ? "passed" : "event-timeout", transport: "api-sse", baseUrl, route: `GET ${path}`, identity: { sessionId: sessionId || null, traceId: traceId || null }, connected, eventCount: business.length, eventTypes: business.map(eventType), timeoutMs, minEvents, waitFor, ...semantics, timing: timingSummary(business, timingDetail), missingSemantics, error: ok ? undefined : { code: "workbench_kafka_sse_event_missing", message: `Workbench SSE did not deliver the required Kafka event conditions; missing=${missingSemantics.join(",") || (business.length < minEvents ? `min-events:${minEvents}` : "connection")}` }, valuesPrinted: false }; + return { ok, operation: "events.inspect", status: ok ? "passed" : "event-timeout", transport: "api-sse", baseUrl, route: `GET ${path}`, identity: { sessionId: sessionId || null, traceId: traceId || null }, connected, connectedContract, eventCount: business.length, eventTypes: business.map(eventType), timeoutMs, minEvents, waitFor, ...semantics, timing: timingSummary(business, timingDetail), missingSemantics, error: ok ? undefined : { code: "workbench_kafka_sse_event_missing", message: `Workbench SSE did not deliver the required Kafka event conditions; missing=${missingSemantics.join(",") || (business.length < minEvents ? `min-events:${minEvents}` : "connection")}` }, valuesPrinted: false }; } function commandFrom(parsed: Parsed, env: Record): WorkbenchCommand { @@ -148,6 +150,31 @@ function semanticWait(value: unknown): EventSemantic[] { return [...new Set(requested)] as EventSemantic[]; } type SseFrame = { name: string; data: Record; observedAt: string }; +function workbenchConnectedContract(data: Record) { + const refreshReplay = record(data.refreshReplay); + const counts = record(refreshReplay?.counts); + return { + deliverySemantics: String(data.deliverySemantics ?? "").trim() || null, + realtimeSource: String(data.realtimeSource ?? "").trim() || null, + replay: data.replay === true, + liveOnly: data.liveOnly === true, + lossPossible: data.lossPossible === true, + refreshReplay: refreshReplay ? { + phase: String(refreshReplay.phase ?? "").trim() || null, + topic: String(refreshReplay.topic ?? "").trim() || null, + barrier: Array.isArray(refreshReplay.barrier) ? refreshReplay.barrier : [], + counts: counts ? { + matched: Number(counts.matched ?? 0), + replayed: Number(counts.replayed ?? 0), + buffered: Number(counts.buffered ?? 0), + bufferedDelivered: Number(counts.bufferedDelivered ?? 0), + liveDelivered: Number(counts.liveDelivered ?? 0), + deduplicated: Number(counts.deduplicated ?? 0) + } : null + } : null, + valuesPrinted: false + }; +} function businessFrames(frames: SseFrame[]) { return frames.filter((entry) => entry.name === "hwlab.event.v1"); } function inspectionSatisfied(business: SseFrame[], minEvents: number, waitFor: EventSemantic[]) { if (business.length < minEvents) return false;