diff --git a/internal/cloud/kafka-event-bridge.test.ts b/internal/cloud/kafka-event-bridge.test.ts index 6c98ea79..58644c23 100644 --- a/internal/cloud/kafka-event-bridge.test.ts +++ b/internal/cloud/kafka-event-bridge.test.ts @@ -490,6 +490,22 @@ test("composable Kafka capabilities reject consumer group collisions", () => { ); }); +test("Workbench bridge rejects direct and transactional dual authority", () => { + assert.throws( + () => startHwlabKafkaEventBridge({ + env: { + ...PROJECTOR_ENV, + HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "true", + HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true", + HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID: "hwlab-direct-authority-test", + HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID: "hwlab-live-authority-test" + }, + runtimeStore: {} + }), + (error: any) => error?.code === "hwlab_workbench_realtime_dual_authority" + ); +}); + test("Kafka refresh replay requires explicit YAML budgets and composes with live SSE", () => { const refreshEnv = { ...LIVE_ENV, diff --git a/internal/cloud/kafka-event-bridge.ts b/internal/cloud/kafka-event-bridge.ts index 3f2670ac..dc2ddf73 100644 --- a/internal/cloud/kafka-event-bridge.ts +++ b/internal/cloud/kafka-event-bridge.ts @@ -119,6 +119,7 @@ export function kafkaEventBridgeConfig(env = process.env) { export function startHwlabKafkaEventBridge({ env = process.env, logger = console, kafkaFactory = defaultKafkaFactory, runtimeStore = null, now = () => new Date().toISOString(), otelSpanEmitter = emitCodeAgentOtelSpan } = {}) { const config = kafkaEventBridgeConfig(env); if (!config) return { started: false, reason: "disabled_or_unconfigured", stop() {}, subscribeProjectionCommits() { return () => {}; }, valuesPrinted: false }; + assertSingleWorkbenchRealtimeAuthority(config.capabilities); const components = []; if (config.capabilities.directPublish || config.capabilities.liveKafkaSse) { components.push(startLiveHwlabKafkaEventBridge({ config, env, logger, kafkaFactory, otelSpanEmitter })); @@ -129,6 +130,16 @@ export function startHwlabKafkaEventBridge({ env = process.env, logger = console return components.length === 1 ? components[0] : combineKafkaEventBridgeComponents(config, components); } +function assertSingleWorkbenchRealtimeAuthority(capabilities = {}) { + const directAuthority = capabilities.directPublish || capabilities.liveKafkaSse || capabilities.kafkaRefreshReplay; + const projectionAuthority = capabilities.transactionalProjector || capabilities.projectionOutboxRelay || capabilities.projectionRealtime; + if (!directAuthority || !projectionAuthority) return; + throw contractError( + "hwlab_workbench_realtime_dual_authority", + "Workbench direct/live Kafka authority cannot run beside the transactional projector authority. Disable direct publish, live-only SSE, and Kafka retention replay before enabling the projector chain." + ); +} + function startTransactionalHwlabKafkaEventBridge({ config, logger = console, kafkaFactory = defaultKafkaFactory, runtimeStore = null, now = () => new Date().toISOString() } = {}) { requireKafkaProjectorStore(runtimeStore, config.capabilities); diff --git a/internal/cloud/server-workbench-realtime-http.test.ts b/internal/cloud/server-workbench-realtime-http.test.ts index bc5c0ab4..be4fe52d 100644 --- a/internal/cloud/server-workbench-realtime-http.test.ts +++ b/internal/cloud/server-workbench-realtime-http.test.ts @@ -90,7 +90,7 @@ test("projection outbox emits immutable assistant versions in row order", () => assert.deepEqual(events.map((item) => item.payload.cursor.outboxSeq), [10, 11]); }); -test("live Kafka SSE transparently fans out one envelope without DB, snapshot, cursor, replay, or SSE id", async () => { +test.skip("removed live-only product authority transparently fans out one envelope", async () => { const sessionId = "ses_live_kafka_sse"; const traceId = "trc_live_kafka_sse"; const subscribers = new Set(); @@ -189,7 +189,7 @@ test("live Kafka SSE transparently fans out one envelope without DB, snapshot, c } }); -test("Kafka refresh SSE replays retained user and lifecycle envelopes before one connected handoff without waiting for projector readiness", async () => { +test.skip("removed Kafka retention replay product authority replays retained envelopes", async () => { const sessionId = "ses_kafka_refresh_replay"; const traceId = "trc_kafka_refresh_replay"; const records = [ @@ -249,7 +249,7 @@ test("Kafka refresh SSE replays retained user and lifecycle envelopes before one } }); -test("Kafka refresh client abort while live readiness is pending never starts retention or live subscription", async () => { +test.skip("removed Kafka retention replay product authority handles client abort", async () => { const sessionId = "ses_kafka_refresh_ready_abort"; let releaseLiveReady: (() => void) | null = null; const liveReady = new Promise((resolve) => { releaseLiveReady = resolve; }); @@ -297,7 +297,7 @@ test("Kafka refresh client abort while live readiness is pending never starts re } }); -test("trace-only Kafka refresh binds the authorized session and rejects same-trace foreign-session retention", async () => { +test.skip("removed Kafka retention replay product authority binds trace ownership", async () => { const sessionId = "ses_kafka_refresh_trace_owner"; const foreignSessionId = "ses_kafka_refresh_trace_foreign"; const traceId = "trc_kafka_refresh_trace_owner"; @@ -329,7 +329,7 @@ test("trace-only Kafka refresh binds the authorized session and rejects same-tra } }); -test("session-scoped Kafka refresh gap emits typed workbench.error and closes without connected or fallback", async () => { +test.skip("removed Kafka retention replay product authority reports retention gaps", async () => { const sessionId = "ses_kafka_refresh_gap"; const traceId = "trc_kafka_refresh_gap"; const retained = refreshRecord(0, sessionId, traceId, "user", { userMessageId: "msg_kafka_refresh_gap", messageId: "msg_kafka_refresh_gap", text: "retained input" }); @@ -366,7 +366,7 @@ test("session-scoped Kafka refresh gap emits typed workbench.error and closes wi } }); -test("live Kafka SSE heartbeat keeps the transport open without DB, cursor, snapshot, or replay", async () => { +test.skip("removed live-only product authority keeps its transport open", async () => { const sessionId = "ses_live_kafka_heartbeat"; let dbCalls = 0; const server = createCloudApiServer({ @@ -409,7 +409,7 @@ test("live Kafka SSE heartbeat keeps the transport open without DB, cursor, snap } }); -test("live Kafka SSE rejects foreign or inconsistent ownership scopes before fanout subscription", async () => { +test.skip("removed live-only product authority rejects foreign ownership", async () => { const ownedSession = { id: "ses_live_kafka_owned", ownerUserId: ACTOR.id, lastTraceId: "trc_live_kafka_owned" }; const foreignSession = { id: "ses_live_kafka_foreign", ownerUserId: "usr_live_kafka_foreign", lastTraceId: "trc_live_kafka_foreign" }; let subscriptions = 0; @@ -460,7 +460,7 @@ test("live Kafka SSE rejects foreign or inconsistent ownership scopes before fan } }); -test("live Kafka SSE fails closed when ownership lookup is not configured", async () => { +test.skip("removed live-only product authority fails closed without ownership lookup", async () => { let subscriptions = 0; const server = createCloudApiServer({ accessController: { @@ -497,7 +497,7 @@ test("live Kafka SSE fails closed when ownership lookup is not configured", asyn } }); -test("live Kafka SSE lets admin subscribe to another owner's consistent session and trace", async () => { +test.skip("removed live-only product authority permits admin ownership scope", async () => { const session = { id: "ses_live_kafka_admin", ownerUserId: "usr_live_kafka_owner", lastTraceId: "trc_live_kafka_admin" }; const admin = { ...ACTOR, id: "usr_live_kafka_admin", role: "admin" }; let subscriptions = 0; @@ -573,7 +573,7 @@ test("workbench realtime initial connection emits current snapshot without repla } }); -test("live and projection realtime remain independently reachable when both capabilities are enabled", async () => { +test("product realtime uses projection outbox even when obsolete live capability values remain present", async () => { const sessionId = "ses_composable_realtime"; const traceId = "trc_composable_realtime"; let projectionReads = 0; @@ -610,7 +610,7 @@ test("live and projection realtime remain independently reachable when both capa }); await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); try { - const events = await getSseEvents(server.address().port, `/v1/workbench/projection-events?sessionId=${sessionId}`, 2); + const events = await getSseEvents(server.address().port, `/v1/workbench/events?sessionId=${sessionId}`, 2); assert.deepEqual(events.map((event) => event.event), ["workbench.connected", "workbench.message.snapshot"]); assert.equal(events[0].data.realtimeSource, "projection-outbox"); assert.equal(projectionReads, 1); diff --git a/internal/cloud/server-workbench-realtime-http.ts b/internal/cloud/server-workbench-realtime-http.ts index b640124d..5f07bc32 100644 --- a/internal/cloud/server-workbench-realtime-http.ts +++ b/internal/cloud/server-workbench-realtime-http.ts @@ -185,19 +185,19 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option if (request.method !== "GET") return methodNotAllowed(response, "GET"); const realtimeCapabilities = workbenchRealtimeCapabilities(options.env ?? process.env); const projectionOnly = url.pathname === "/v1/workbench/projection-events"; - if (!projectionOnly && realtimeCapabilities.liveKafkaSse) { - await handleLiveKafkaWorkbenchRealtimeHttp(request, response, url, options); + if (projectionOnly) { + sendJson(response, 410, workbenchError("workbench_projection_events_route_removed", "Use /v1/workbench/events for the single transactional projection authority.")); return; } if (!realtimeCapabilities.projectionRealtime) { - sendJson(response, 503, workbenchError("workbench_realtime_disabled", "No Workbench realtime SSE capability is enabled.")); + sendJson(response, 503, workbenchError("workbench_projection_realtime_disabled", "Workbench product realtime requires the transactional projection outbox authority.")); return; } const requestedSessionId = safeSessionId(url.searchParams.get("sessionId") ?? url.searchParams.get("includeSessionId")); const requestedTraceId = safeTraceId(url.searchParams.get("traceId")); const heartbeatMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_HEARTBEAT_MS, DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS); attachWorkbenchRealtimeOtelContext(request, { - route: projectionOnly ? "/v1/workbench/projection-events" : "/v1/workbench/events", + route: "/v1/workbench/events", sessionId: requestedSessionId, traceId: requestedTraceId, heartbeatMs, diff --git a/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts b/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts index abacb847..fa12dba2 100644 --- a/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts @@ -83,22 +83,23 @@ test("Workbench API uses metadata-only session detail and bounded messages paths assert.equal(workbenchSessionMessagesPathForTest("ses_metadata", { limit: 9 }), "/v1/workbench/sessions/ses_metadata/messages?limit=9"); }); -test("live SSE transport never sends an afterSeq cursor while projection transport can", () => { +test("only the projection authority sends the durable afterSeq cursor", () => { assert.equal(workbenchEventStreamPath({ realtimeCapabilities: { liveKafkaSse: true, kafkaRefreshReplay: false, projectionRealtime: false }, sessionId: "ses_live", traceId: "trc_live", afterSeq: 42 }), "/v1/workbench/events?sessionId=ses_live&traceId=trc_live"); assert.equal(workbenchEventStreamPath({ realtimeCapabilities: { liveKafkaSse: false, kafkaRefreshReplay: false, projectionRealtime: true }, sessionId: "ses_projection", traceId: "trc_projection", afterSeq: 42 }), "/v1/workbench/events?sessionId=ses_projection&traceId=trc_projection&afterSeq=42"); - assert.equal(workbenchEventStreamPath({ realtimeCapabilities: { liveKafkaSse: true, kafkaRefreshReplay: true, projectionRealtime: true }, sessionId: "ses_both", traceId: "trc_both", afterSeq: 42 }), "/v1/workbench/events?sessionId=ses_both&traceId=trc_both"); + assert.equal(workbenchEventStreamPath({ realtimeCapabilities: { liveKafkaSse: true, kafkaRefreshReplay: true, projectionRealtime: true }, sessionId: "ses_both", traceId: "trc_both", afterSeq: 42 }), "/v1/workbench/events?sessionId=ses_both&traceId=trc_both&afterSeq=42"); assert.equal(workbenchProjectionEventStreamPath({ sessionId: "ses_both", traceId: "trc_both", afterSeq: 42 }), "/v1/workbench/projection-events?sessionId=ses_both&traceId=trc_both&afterSeq=42"); assert.equal(workbenchRealtimeTransportEnabled({ liveKafkaSse: false, kafkaRefreshReplay: false, projectionRealtime: false }), false); - assert.equal(workbenchRealtimeTransportEnabled({ liveKafkaSse: true, kafkaRefreshReplay: false, projectionRealtime: false }), true); + assert.equal(workbenchRealtimeTransportEnabled({ liveKafkaSse: true, kafkaRefreshReplay: false, projectionRealtime: false }), false); assert.equal(workbenchRealtimeTransportEnabled({ liveKafkaSse: false, kafkaRefreshReplay: false, projectionRealtime: true }), true); assert.equal(workbenchRealtimeTransportEnabled({ liveKafkaSse: true, kafkaRefreshReplay: true, projectionRealtime: true }), true); }); -test("live Kafka keeps the pre-submit session SSE key when a turn trace becomes active", () => { +test("obsolete live flags cannot erase the active projection trace scope", () => { const capabilities = { liveKafkaSse: true, kafkaRefreshReplay: true, projectionRealtime: false }; const beforeSubmit = workbenchRealtimeScopeKey("ses_live", workbenchRealtimeTraceIdForCapabilities(capabilities, null, null)); const afterSubmit = workbenchRealtimeScopeKey("ses_live", workbenchRealtimeTraceIdForCapabilities(capabilities, "trc_current_request", "trc_message")); - assert.equal(afterSubmit, beforeSubmit); + assert.notEqual(afterSubmit, beforeSubmit); + assert.equal(afterSubmit, workbenchRealtimeScopeKey("ses_live", "trc_current_request")); assert.equal(workbenchRealtimeTraceIdForCapabilities({ liveKafkaSse: false, kafkaRefreshReplay: false, projectionRealtime: true }, "trc_current_request", "trc_message"), "trc_current_request"); }); diff --git a/web/hwlab-cloud-web/src/api/workbench-events.test.ts b/web/hwlab-cloud-web/src/api/workbench-events.test.ts index c5d302e4..de21fcd2 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.test.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import { test } from "bun:test"; -import { connectWorkbenchEvents, realtimeCoalesceKey, type WorkbenchRealtimeEvent, type WorkbenchSseIngressFrame } from "./workbench-events"; +import { connectWorkbenchEvents, realtimeCoalesceKey, workbenchEventStreamPath, type WorkbenchRealtimeEvent, type WorkbenchSseIngressFrame } from "./workbench-events"; import { createCoalescedEventQueue } from "../utils/scheduler/coalesced-event-queue"; test("turn snapshot coalescing keys by trace instead of per sequence", () => { @@ -18,6 +18,18 @@ test("trace events keep sequence-specific coalescing keys", () => { assert.notEqual(realtimeCoalesceKey(first, "workbench.trace.event"), realtimeCoalesceKey(next, "workbench.trace.event")); }); +test("product EventSource resumes the transactional projection authority with one durable outbox cursor", () => { + assert.equal( + workbenchEventStreamPath({ + realtimeCapabilities: { liveKafkaSse: true, kafkaRefreshReplay: false, projectionRealtime: true }, + sessionId: "ses_projection_authority", + traceId: "trc_projection_authority", + afterSeq: 42 + }), + "/v1/workbench/events?sessionId=ses_projection_authority&traceId=trc_projection_authority&afterSeq=42" + ); +}); + test("live Kafka envelopes from one trace retain every assistant, tool, output, and terminal event", () => { const delivered: WorkbenchRealtimeEvent[] = []; const queue = createCoalescedEventQueue({ diff --git a/web/hwlab-cloud-web/src/api/workbench-events.ts b/web/hwlab-cloud-web/src/api/workbench-events.ts index c1b0aa37..bb8db867 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.ts @@ -213,7 +213,7 @@ export function workbenchEventStreamPath(options: Pick { +test("obsolete Kafka retention flags cannot replace projection replay or durable history", () => { assert.deepEqual(workbenchHistoryAuthorityPolicy({ liveKafkaSse: true, kafkaRefreshReplay: true, projectionRealtime: true }), { - kafkaRetention: true, + kafkaRetention: false, sessionMetadataRead: true, - sessionMessagesHydrate: false, - turnStatusHydrate: false, - traceEventsHydrate: false, - syncReplay: false + sessionMessagesHydrate: true, + turnStatusHydrate: true, + traceEventsHydrate: true, + syncReplay: true }); }); -test("independent live-only and projection capabilities preserve their existing history policy", () => { +test("projection capability alone controls product replay", () => { assert.deepEqual(workbenchHistoryAuthorityPolicy({ liveKafkaSse: true, kafkaRefreshReplay: false, projectionRealtime: false }), { kafkaRetention: false, sessionMetadataRead: true, diff --git a/web/hwlab-cloud-web/src/stores/workbench-kafka-refresh-policy.ts b/web/hwlab-cloud-web/src/stores/workbench-kafka-refresh-policy.ts index 61597642..b38a5269 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-kafka-refresh-policy.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-kafka-refresh-policy.ts @@ -10,13 +10,12 @@ export interface WorkbenchHistoryAuthorityPolicy { } export function workbenchHistoryAuthorityPolicy(capabilities: WorkbenchRealtimeCapabilities): WorkbenchHistoryAuthorityPolicy { - const kafkaRetention = capabilities.liveKafkaSse && capabilities.kafkaRefreshReplay; return { - kafkaRetention, + kafkaRetention: false, sessionMetadataRead: true, - sessionMessagesHydrate: !kafkaRetention, - turnStatusHydrate: !kafkaRetention, - traceEventsHydrate: !kafkaRetention, - syncReplay: !kafkaRetention && capabilities.projectionRealtime + sessionMessagesHydrate: true, + turnStatusHydrate: true, + traceEventsHydrate: true, + syncReplay: capabilities.projectionRealtime }; } diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index fb9f05a3..cd25c89b 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -1193,7 +1193,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { } async function refreshWorkbenchSyncReplay(sessionId: string | null, traceId: string | null, sinceOutboxSeq: number | null, reason: string): Promise { - if (realtimeCapabilities.liveKafkaSse || !realtimeCapabilities.projectionRealtime) return; + if (!realtimeCapabilities.projectionRealtime) return; let cursor = firstFiniteNumber(sinceOutboxSeq) ?? 0; for (;;) { const result = await workbenchColadaQueries.fetchSyncReplay({ sessionId, traceId, since: cursor }, { timeoutMs: 8000, activityRef: () => activityRef.value }); @@ -1536,7 +1536,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { } function publishWorkbenchProjectionSignal(sessionId: string | null | undefined, traceId: string | null | undefined, reason: string): void { - if (realtimeCapabilities.liveKafkaSse || !realtimeCapabilities.projectionRealtime) return; + if (!realtimeCapabilities.projectionRealtime) return; if (typeof window === "undefined") return; const id = normalizeWorkbenchSessionId(sessionId); if (!id) return; @@ -1546,7 +1546,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { } function handleWorkbenchProjectionSignal(value: unknown): void { - if (realtimeCapabilities.liveKafkaSse || !realtimeCapabilities.projectionRealtime) return; + if (!realtimeCapabilities.projectionRealtime) return; const record = recordValue(value); if (!record || record.sourceId === workbenchProjectionSignalSourceId) return; if (firstNonEmptyString(record.type) !== "session-projection") return; diff --git a/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts b/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts index 5609c29a..8fe49afa 100644 --- a/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts +++ b/web/hwlab-cloud-web/src/utils/workbench-stream-transport.ts @@ -78,10 +78,9 @@ interface WorkbenchTransportCursor { } export function workbenchRealtimeTraceIdForCapabilities( - capabilities: WorkbenchRealtimeCapabilities, + _capabilities: WorkbenchRealtimeCapabilities, ...candidates: Array ): string | null { - if (capabilities.liveKafkaSse) return null; for (const candidate of candidates) { const traceId = typeof candidate === "string" ? candidate.trim() : ""; if (traceId) return traceId; @@ -90,7 +89,7 @@ export function workbenchRealtimeTraceIdForCapabilities( } export function workbenchRealtimeTransportEnabled(capabilities: WorkbenchRealtimeCapabilities): boolean { - return capabilities.liveKafkaSse || capabilities.projectionRealtime; + return capabilities.projectionRealtime; } export class WorkbenchStreamTransportRuntime {