From 3e03e94e3538649e5bf7671f2dd00f51a7620f08 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 14 Jul 2026 03:55:47 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=9B=BA=E5=AE=9A=20Workbench=20transac?= =?UTF-8?q?tional=20=E6=8A=95=E5=BD=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/feature-config.schema.json | 44 ++++++ internal/cloud/kafka-event-bridge.test.ts | 77 +++++----- internal/cloud/kafka-event-bridge.ts | 138 ++++++------------ .../cloud/server-code-agent-admission-http.ts | 28 +--- .../cloud/server-workbench-realtime-http.ts | 10 +- .../cloud/workbench-projection-writer.test.ts | 83 +++++++++++ internal/cloud/workbench-projection-writer.ts | 18 +-- .../cloud/workbench-realtime-capabilities.ts | 42 ++---- internal/dev-entrypoint/cloud-web-runtime.mjs | 5 - .../dev-entrypoint/cloud-web-runtime.test.mjs | 38 +++-- .../scripts/workbench-server-state.test.ts | 6 +- web/hwlab-cloud-web/src/config/runtime.ts | 6 +- web/hwlab-cloud-web/src/types/global.d.ts | 5 - 13 files changed, 254 insertions(+), 246 deletions(-) create mode 100644 config/feature-config.schema.json diff --git a/config/feature-config.schema.json b/config/feature-config.schema.json new file mode 100644 index 00000000..e293b3a3 --- /dev/null +++ b/config/feature-config.schema.json @@ -0,0 +1,44 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/pikasTech/HWLAB/config/feature-config.schema.json", + "title": "HWLAB product feature configuration", + "type": "object", + "additionalProperties": false, + "properties": { + "tracePanel": { + "x-unidesk-feature": "workbench.trace-panel", + "type": "object", + "additionalProperties": false, + "properties": { + "autoExpandRunning": { "type": "boolean" }, + "autoCollapseTerminal": { "type": "boolean" } + } + }, + "rawEvents": { + "x-unidesk-feature": "workbench.raw-events", + "type": "object", + "additionalProperties": false, + "properties": { + "enabled": { "type": "boolean" }, + "maxEntries": { "type": "integer", "minimum": 1 }, + "maxRetainedBytes": { "type": "integer", "minimum": 1 } + } + }, + "debugReplay": { + "x-unidesk-feature": "workbench.debug-replay", + "type": "object", + "additionalProperties": false, + "properties": { + "enabled": { "type": "boolean" } + } + }, + "views": { + "x-unidesk-feature": "workbench.views", + "type": "object", + "additionalProperties": false, + "properties": { + "traceExplorerUrlTemplate": { "type": "string", "minLength": 1 } + } + } + } +} diff --git a/internal/cloud/kafka-event-bridge.test.ts b/internal/cloud/kafka-event-bridge.test.ts index 89512113..835d82c4 100644 --- a/internal/cloud/kafka-event-bridge.test.ts +++ b/internal/cloud/kafka-event-bridge.test.ts @@ -379,53 +379,42 @@ test("live Kafka OTel sampling is deterministic and bounded without payload attr assert.equal(JSON.stringify(attributes).includes("must-not-leak"), false); }); -test("composable Kafka capabilities reject consumer group collisions", () => { - assert.throws( - () => kafkaEventBridgeConfig({ - ...LIVE_ENV, - HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED: "true", - HWLAB_KAFKA_PROJECTOR_GROUP_ID: LIVE_ENV.HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID, - HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS: "1000" - }), - (error) => error?.code === "hwlab_kafka_consumer_groups_not_distinct" - ); +test("Workbench bridge ignores obsolete architecture env and fixes transactional authority", () => { + const config = kafkaEventBridgeConfig({ + ...PROJECTOR_ENV, + HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "true", + HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true", + HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED: "true", + HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED: "false", + HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED: "false", + HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false" + }); + + assert.deepEqual(config?.capabilities, { + directPublish: false, + liveKafkaSse: false, + kafkaRefreshReplay: false, + transactionalProjector: true, + projectionOutboxRelay: true, + projectionRealtime: true + }); + assert.equal(config?.directPublishGroupId, null); + assert.equal(config?.hwlabEventGroupId, null); + assert.equal(config?.refreshReplay, null); }); -test("Workbench bridge fails closed unless transactional projection authority is complete", () => { - const invalidAuthorities = [ - { name: "direct/live-only", env: LIVE_ENV }, - { - name: "projection realtime only", - env: { - HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "false", - HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "false", - HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED: "false", - HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED: "false", - HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED: "false", - HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "true" - } - }, - { name: "projector without outbox relay", env: { ...PROJECTOR_ENV, HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED: "false" } }, - { name: "projector without projection realtime", env: { ...PROJECTOR_ENV, HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false" } }, - { - name: "direct and transactional dual authority", - 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" - } - } - ]; +test("Workbench bridge does not require the six obsolete architecture env", () => { + const env = { ...PROJECTOR_ENV }; + for (const name of [ + "HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED", + "HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", + "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", + "HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED", + "HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED", + "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED" + ]) delete env[name]; - for (const authority of invalidAuthorities) { - assert.throws( - () => startHwlabKafkaEventBridge({ env: authority.env, runtimeStore: {} }), - (error: any) => error?.code === "hwlab_workbench_transactional_authority_required", - authority.name - ); - } + assert.doesNotThrow(() => kafkaEventBridgeConfig(env)); }); test("projects AgentRun run-created envelope even when no source event is present", () => { diff --git a/internal/cloud/kafka-event-bridge.ts b/internal/cloud/kafka-event-bridge.ts index 225856f0..246c09e9 100644 --- a/internal/cloud/kafka-event-bridge.ts +++ b/internal/cloud/kafka-event-bridge.ts @@ -6,10 +6,7 @@ import { createHash, randomUUID } from "node:crypto"; import { Kafka, logLevel } from "kafkajs"; import { emitCodeAgentOtelSpan } from "./otel-trace.ts"; import { buildWorkbenchProjectionEventFacts } from "./workbench-projection-writer.ts"; -import { - WORKBENCH_REALTIME_CAPABILITY_ENVS, - workbenchRealtimeCapabilities -} from "./workbench-realtime-capabilities.ts"; +import { workbenchRealtimeCapabilities } from "./workbench-realtime-capabilities.ts"; const TRUE_VALUES = new Set(["1", "true", "yes", "on"]); export const DEFAULT_AGENTRUN_EVENT_TOPIC = "agentrun.event.v1"; @@ -31,35 +28,20 @@ const REQUIRED_KAFKA_ENV = Object.freeze([ ]); export function kafkaEventBridgeConfig(env = process.env) { - const featureEnvPresent = Object.values(WORKBENCH_REALTIME_CAPABILITY_ENVS).some((name) => stringValue(env[name])); const legacyKafkaEnabled = truthy(env.HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED ?? env.HWLAB_KAFKA_AGENTRUN_CONSUME_ENABLED ?? env.HWLAB_KAFKA_ENABLED); - if (!featureEnvPresent && !legacyKafkaEnabled) return null; - const capabilities = workbenchRealtimeCapabilities(env); - if (capabilities.kafkaRefreshReplay && !capabilities.liveKafkaSse) { - throw contractError("hwlab_kafka_refresh_live_dependency_missing", "Kafka refresh replay requires the independently enabled live Kafka SSE capability for handoff."); - } - const kafkaEnabled = capabilities.directPublish || capabilities.liveKafkaSse || capabilities.kafkaRefreshReplay || capabilities.transactionalProjector || capabilities.projectionOutboxRelay; - const transactionalEnabled = capabilities.transactionalProjector || capabilities.projectionOutboxRelay || capabilities.projectionRealtime; - if (!kafkaEnabled && !transactionalEnabled) return null; - const required = [...REQUIRED_KAFKA_ENV]; - if (!kafkaEnabled) required.length = 0; - if (capabilities.directPublish) required.push("HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID"); - if (capabilities.liveKafkaSse) required.push("HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID"); - if (capabilities.kafkaRefreshReplay) required.push( - "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX", - "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS", - "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT", - "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT", - "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT" - ); - if (capabilities.transactionalProjector) required.push("HWLAB_KAFKA_PROJECTOR_GROUP_ID", "HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS"); - if (capabilities.projectionOutboxRelay) required.push( + const kafkaConfigured = REQUIRED_KAFKA_ENV.every((name) => stringValue(env[name])); + if (!legacyKafkaEnabled && !kafkaConfigured) return null; + const capabilities = workbenchRealtimeCapabilities(); + const required = [ + ...REQUIRED_KAFKA_ENV, + "HWLAB_KAFKA_PROJECTOR_GROUP_ID", + "HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS", "HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS", "HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE", "HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS", "HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS", "HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS" - ); + ]; const missing = required.filter((name) => !stringValue(env[name])); if (missing.length > 0) { const error = new Error(`HWLAB Kafka capability configuration is incomplete: ${missing.join(", ")}`); @@ -71,47 +53,29 @@ export function kafkaEventBridgeConfig(env = process.env) { const agentRunTopic = stringValue(env.HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC); const hwlabTopic = stringValue(env.HWLAB_KAFKA_EVENT_TOPIC); const clientId = stringValue(env.HWLAB_KAFKA_CLIENT_ID); - const relay = capabilities.projectionOutboxRelay ? { + const relay = { intervalMs: strictPositiveInteger(env.HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS, "HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS"), batchSize: strictPositiveInteger(env.HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE, "HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE"), leaseMs: strictPositiveInteger(env.HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS, "HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS"), sendTimeoutMs: strictPositiveInteger(env.HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS, "HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS"), retryBackoffMs: strictPositiveInteger(env.HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS, "HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS") - } : null; - const refreshReplay = capabilities.kafkaRefreshReplay ? { - groupIdPrefix: stringValue(env.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX), - timeoutMs: strictPositiveInteger(env.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS"), - scanLimit: strictPositiveInteger(env.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT"), - matchedEventLimit: strictPositiveInteger(env.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT"), - liveBufferLimit: strictPositiveInteger(env.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT") - } : null; + }; if (relay && relay.sendTimeoutMs + 5000 >= relay.leaseMs) { const error = new Error("HWLAB Kafka relay lease must exceed producer send timeout by at least 5000ms."); error.code = "hwlab_kafka_relay_lease_invalid"; throw error; } - const enabledConsumerGroups = [ - capabilities.directPublish ? stringValue(env.HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID) : null, - capabilities.transactionalProjector ? stringValue(env.HWLAB_KAFKA_PROJECTOR_GROUP_ID) : null, - capabilities.liveKafkaSse ? stringValue(env.HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID) : null - ].filter(Boolean); - if (new Set(enabledConsumerGroups).size !== enabledConsumerGroups.length) { - const error = new Error("HWLAB Kafka direct publisher, projector, and live SSE consumer groups must be distinct when enabled together."); - error.code = "hwlab_kafka_consumer_groups_not_distinct"; - error.groupIds = enabledConsumerGroups; - throw error; - } return { capabilities, brokers, agentRunTopic, hwlabTopic, clientId, - directPublishGroupId: stringValue(env.HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID), + directPublishGroupId: null, projectorGroupId: stringValue(env.HWLAB_KAFKA_PROJECTOR_GROUP_ID), - hwlabEventGroupId: stringValue(env.HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID), - refreshReplay, - projectorHeartbeatIntervalMs: capabilities.transactionalProjector ? strictPositiveInteger(env.HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS, "HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS") : null, + hwlabEventGroupId: null, + refreshReplay: null, + projectorHeartbeatIntervalMs: strictPositiveInteger(env.HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS, "HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS"), relay }; } @@ -119,29 +83,11 @@ 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 }; - assertTransactionalWorkbenchRealtimeAuthority(config.capabilities); - const components = []; - if (config.capabilities.directPublish || config.capabilities.liveKafkaSse) { - components.push(startLiveHwlabKafkaEventBridge({ config, env, logger, kafkaFactory, otelSpanEmitter })); - } - if (config.capabilities.transactionalProjector || config.capabilities.projectionOutboxRelay || config.capabilities.projectionRealtime) { - components.push(startTransactionalHwlabKafkaEventBridge({ config, logger, kafkaFactory, runtimeStore, now })); - } - return components.length === 1 ? components[0] : combineKafkaEventBridgeComponents(config, components); -} - -function assertTransactionalWorkbenchRealtimeAuthority(capabilities = {}) { - const directAuthority = capabilities.directPublish || capabilities.liveKafkaSse || capabilities.kafkaRefreshReplay; - const completeProjectionAuthority = capabilities.transactionalProjector && capabilities.projectionOutboxRelay && capabilities.projectionRealtime; - if (!directAuthority && completeProjectionAuthority) return; - throw contractError( - "hwlab_workbench_transactional_authority_required", - "Workbench Kafka bridge requires transactionalProjector, projectionOutboxRelay, and projectionRealtime together, with directPublish, liveKafkaSse, and kafkaRefreshReplay disabled." - ); + return startTransactionalHwlabKafkaEventBridge({ config, logger, kafkaFactory, runtimeStore, now }); } function startTransactionalHwlabKafkaEventBridge({ config, logger = console, kafkaFactory = defaultKafkaFactory, runtimeStore = null, now = () => new Date().toISOString() } = {}) { - requireKafkaProjectorStore(runtimeStore, config.capabilities); + requireKafkaProjectorStore(runtimeStore); let stopped = false; let consumer = null; @@ -160,16 +106,14 @@ function startTransactionalHwlabKafkaEventBridge({ config, logger = console, kaf let startupReady = false; const ready = (async () => { await runtimeStore.assertWorkbenchTransactionalRealtimeReady(); - const kafka = config.capabilities.transactionalProjector || config.capabilities.projectionOutboxRelay ? kafkaFactory(config) : null; - if (config.capabilities.transactionalProjector) consumer = kafka.consumer({ groupId: config.projectorGroupId, allowAutoTopicCreation: false }); - if (config.capabilities.projectionOutboxRelay) producer = kafka.producer({ allowAutoTopicCreation: false }); - if (producer) await producer.connect(); - if (consumer) await consumer.connect(); - if (config.capabilities.projectionRealtime && typeof runtimeStore.subscribeWorkbenchProjectionCommits === "function") { - stopProjectionNotifications = await runtimeStore.subscribeWorkbenchProjectionCommits(publishProjectionCommit); - } - if (consumer) await consumer.subscribe({ topic: config.agentRunTopic, fromBeginning: false }); - if (consumer) await consumer.run({ + const kafka = kafkaFactory(config); + consumer = kafka.consumer({ groupId: config.projectorGroupId, allowAutoTopicCreation: false }); + producer = kafka.producer({ allowAutoTopicCreation: false }); + await producer.connect(); + await consumer.connect(); + stopProjectionNotifications = await runtimeStore.subscribeWorkbenchProjectionCommits(publishProjectionCommit); + await consumer.subscribe({ topic: config.agentRunTopic, fromBeginning: false }); + await consumer.run({ autoCommit: false, eachBatchAutoResolve: false, eachBatch: async ({ batch, resolveOffset, heartbeat, isRunning, isStale }) => { @@ -196,7 +140,6 @@ function startTransactionalHwlabKafkaEventBridge({ config, logger = console, kaf } }); const runRelay = async () => { - if (!config.capabilities.projectionOutboxRelay) return; if (stopped || relayRunning) return; relayRunning = true; try { @@ -205,11 +148,9 @@ function startTransactionalHwlabKafkaEventBridge({ config, logger = console, kaf relayRunning = false; } }; - if (config.capabilities.projectionOutboxRelay) { - relayTimer = setInterval(() => { void runRelay().catch((error) => logWarn(logger, "hwlab-kafka-outbox-relay-failed", { message: errorMessage(error), valuesPrinted: false })); }, config.relay.intervalMs); - relayTimer.unref?.(); - await runRelay(); - } + relayTimer = setInterval(() => { void runRelay().catch((error) => logWarn(logger, "hwlab-kafka-outbox-relay-failed", { message: errorMessage(error), valuesPrinted: false })); }, config.relay.intervalMs); + relayTimer.unref?.(); + await runRelay(); startupReady = true; logInfo(logger, "hwlab-kafka-event-bridge-started", { agentRunTopic: config.agentRunTopic, @@ -249,9 +190,7 @@ function startTransactionalHwlabKafkaEventBridge({ config, logger = console, kaf const transactionalReadiness = typeof runtimeStore.workbenchTransactionalRealtimeReadiness === "function" ? await runtimeStore.workbenchTransactionalRealtimeReadiness() : { ready: true, valuesRedacted: true }; - const projectorStatus = config.capabilities.transactionalProjector || config.capabilities.projectionOutboxRelay - ? await runtimeStore.hwlabKafkaProjectorStatus() - : {}; + const projectorStatus = await runtimeStore.hwlabKafkaProjectorStatus(); return { status: transactionalReadiness.ready === false ? "blocked" : "ready", capabilities: config.capabilities, transactionalReadiness, ...projectorStatus, valuesRedacted: true }; }, startupStatus() { @@ -953,13 +892,18 @@ function kafkaTransport(kafkaMessage = {}) { }; } -function requireKafkaProjectorStore(runtimeStore, capabilities = {}) { - const required = []; - if (capabilities.transactionalProjector) required.push("commitAgentRunKafkaProjection", "recordFailedAgentRunKafkaMessage", "recordIgnoredAgentRunKafkaMessage"); - if (capabilities.projectionOutboxRelay) required.push("claimHwlabKafkaOutbox", "completeHwlabKafkaOutbox", "retryHwlabKafkaOutbox"); - if (capabilities.projectionRealtime) required.push("subscribeWorkbenchProjectionCommits"); - if (capabilities.transactionalProjector || capabilities.projectionOutboxRelay) required.push("hwlabKafkaProjectorStatus"); - if (capabilities.transactionalProjector || capabilities.projectionOutboxRelay || capabilities.projectionRealtime) required.push("assertWorkbenchTransactionalRealtimeReady"); +function requireKafkaProjectorStore(runtimeStore) { + const required = [ + "commitAgentRunKafkaProjection", + "recordFailedAgentRunKafkaMessage", + "recordIgnoredAgentRunKafkaMessage", + "claimHwlabKafkaOutbox", + "completeHwlabKafkaOutbox", + "retryHwlabKafkaOutbox", + "subscribeWorkbenchProjectionCommits", + "hwlabKafkaProjectorStatus", + "assertWorkbenchTransactionalRealtimeReady" + ]; const missing = required.filter((name) => typeof runtimeStore?.[name] !== "function"); if (missing.length > 0) throw contractError("hwlab_kafka_projector_store_invalid", `Kafka durable capabilities require a runtime store: ${missing.join(", ")}`); } diff --git a/internal/cloud/server-code-agent-admission-http.ts b/internal/cloud/server-code-agent-admission-http.ts index 036e9aee..8bbff398 100644 --- a/internal/cloud/server-code-agent-admission-http.ts +++ b/internal/cloud/server-code-agent-admission-http.ts @@ -1046,8 +1046,7 @@ async function submitCodeAgentChatTurn({ params, options, traceId }) { } }); if (adapterEnabled) { - const realtimeCapabilities = workbenchRealtimeCapabilities(runtimeEnv); - const transactionalAdmission = realtimeCapabilities.transactionalProjector; + const transactionalAdmission = true; const pendingPromotion = results.get(traceId); if (pendingPromotion?.admissionState === "promotion-pending" && pendingPromotion?.agentRun?.durableDispatch === true) { const retryParams = { ...params, userBillingReservation: pendingPromotion.userBillingReservation ?? null }; @@ -1324,30 +1323,7 @@ async function promoteCodeAgentTurnAdmission({ payload = {}, params = {}, option throw error; } const lifecycle = codeAgentTurnLifecycleFields(traceId, params); - const realtimeCapabilities = workbenchRealtimeCapabilities(options.env ?? process.env); - if (!realtimeCapabilities.transactionalProjector) { - const runningOwner = await recordCodeAgentSessionOwner({ payload: { ...payload, status: "running" }, params, options, status: "running" }); - if (!runningOwner) { - const error = new Error("Code Agent live turn did not durably promote the session owner to running."); - error.code = "workbench_admission_owner_not_promoted"; - error.valuesRedacted = true; - throw error; - } - traceStore.append(traceId, { - type: "request", - status: "admitted", - label: "turn:admitted-live-kafka", - message: "Code Agent turn is live after AgentRun durable dispatch; lifecycle visibility is owned only by hwlab.event.v1.", - sessionId: safeSessionId(params.sessionId) || null, - runId: payload.agentRun?.runId ?? null, - commandId: payload.agentRun?.commandId ?? null, - dispatchIntentId: payload.agentRun?.dispatchIntentId ?? null, - waitingFor: "hwlab-live-kafka-sse", - terminal: false, - valuesPrinted: false - }); - return { written: false, promoted: true, outboxCommitted: false, realtimeCapabilities, liveOnly: realtimeCapabilities.liveKafkaSse, valuesPrinted: false }; - } + const realtimeCapabilities = workbenchRealtimeCapabilities(); const inputFact = buildCodeAgentSessionInputFact({ params, options, diff --git a/internal/cloud/server-workbench-realtime-http.ts b/internal/cloud/server-workbench-realtime-http.ts index 5f07bc32..82d257ff 100644 --- a/internal/cloud/server-workbench-realtime-http.ts +++ b/internal/cloud/server-workbench-realtime-http.ts @@ -189,10 +189,6 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option 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_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); @@ -865,9 +861,9 @@ export function attachWorkbenchRealtimeOtelContext(request, fields = {}) { "workbench.thread_id": fields.threadId ?? null, "workbench.sse.heartbeat_ms": fields.heartbeatMs ?? null, "workbench.sse.realtime_source": fields.realtimeSource ?? null, - "workbench.sse.live_kafka_enabled": fields.realtimeCapabilities?.liveKafkaSse ?? null, - "workbench.sse.projection_realtime_enabled": fields.realtimeCapabilities?.projectionRealtime ?? null, - "workbench.sse.outbox_mode": fields.realtimeCapabilities?.liveKafkaSse === true ? false : true + "workbench.sse.live_kafka_enabled": false, + "workbench.sse.projection_realtime_enabled": true, + "workbench.sse.outbox_mode": true }; } diff --git a/internal/cloud/workbench-projection-writer.test.ts b/internal/cloud/workbench-projection-writer.test.ts index 3336f1c8..29d8c931 100644 --- a/internal/cloud/workbench-projection-writer.test.ts +++ b/internal/cloud/workbench-projection-writer.test.ts @@ -144,6 +144,89 @@ test("authoritative assistant final seals terminal with a final response", () => assert.equal(built.facts.checkpoints[0].projectionStatus, "caught_up"); }); +test("canonical event time owns running, terminal, replay, and duration", () => { + const running = buildWorkbenchProjectionEventFacts({ + projectedSeq: 20, + projectedAt: "2026-07-10T10:17:00.000Z", + event: { + traceId: "trc_canonical_duration", + sessionId: "ses_canonical_duration", + sourceEventId: "evt_canonical_running", + sourceSeq: 41, + type: "assistant", + eventType: "assistant_progress", + status: "running", + createdAt: "2026-07-10T10:00:01.000Z", + startedAt: "2026-07-10T09:43:00.000Z", + durationMs: 1020000 + } + }); + const runningCheckpoint = running.facts.checkpoints[0]; + assert.equal(runningCheckpoint.startedAt, "2026-07-10T10:00:01.000Z"); + assert.equal(runningCheckpoint.lastEventAt, "2026-07-10T10:00:01.000Z"); + assert.equal(runningCheckpoint.durationMs, null); + + const terminal = buildWorkbenchProjectionEventFacts({ + projectedSeq: 21, + projectedAt: "2026-07-10T10:18:00.000Z", + previousCheckpoint: runningCheckpoint, + event: { + traceId: "trc_canonical_duration", + sessionId: "ses_canonical_duration", + sourceEventId: "evt_canonical_terminal", + sourceSeq: 42, + type: "assistant", + eventType: "assistant", + status: "completed", + terminal: true, + replyAuthority: true, + text: "canonical final", + createdAt: "2026-07-10T10:00:04.000Z", + finishedAt: "2026-07-10T10:18:00.000Z", + durationMs: 1020000 + } + }); + const terminalCheckpoint = terminal.facts.checkpoints[0]; + assert.equal(terminalCheckpoint.startedAt, "2026-07-10T10:00:01.000Z"); + assert.equal(terminalCheckpoint.finishedAt, "2026-07-10T10:00:04.000Z"); + assert.equal(terminalCheckpoint.durationMs, 3000); + + const [replayed] = projectionOutboxRealtimeEvents({ + events: [{ + outboxSeq: 72, + entityFamily: "traceEvents", + entityId: terminal.facts.traceEvents[0].id, + projectedSeq: 21, + projectionRevision: 1, + traceId: "trc_canonical_duration", + sessionId: "ses_canonical_duration", + commitType: "event", + terminal: true, + sealed: true, + payload: { family: "traceEvents", fact: terminal.facts.traceEvents[0] } + }] + }); + assert.deepEqual(replayed.payload.cursor, { outboxSeq: 72, traceSeq: 21 }); + assert.equal(replayed.payload.event.durationMs, 3000); + const [refreshed] = projectionOutboxRealtimeEvents({ + events: [{ + outboxSeq: 72, + entityFamily: "traceEvents", + entityId: terminal.facts.traceEvents[0].id, + projectedSeq: 21, + projectionRevision: 1, + traceId: "trc_canonical_duration", + sessionId: "ses_canonical_duration", + commitType: "event", + terminal: true, + sealed: true, + payload: { family: "traceEvents", fact: terminal.facts.traceEvents[0] } + }] + }); + assert.deepEqual(refreshed.payload.cursor, replayed.payload.cursor); + assert.equal(refreshed.payload.event.durationMs, replayed.payload.event.durationMs); +}); + test("completed terminal without final response remains visibly unsealed", () => { const built = buildWorkbenchProjectionEventFacts({ projectedSeq: 1, diff --git a/internal/cloud/workbench-projection-writer.ts b/internal/cloud/workbench-projection-writer.ts index ab05e8d0..705da175 100644 --- a/internal/cloud/workbench-projection-writer.ts +++ b/internal/cloud/workbench-projection-writer.ts @@ -205,12 +205,9 @@ export function buildWorkbenchProjectionEventFacts({ event = {}, requestMeta = { valuesPrinted: false }; } - const explicitEventTiming = normalizeTimingProjection(event.timing) ?? normalizeTimingProjection(event); - const explicitDurationMs = durationValue(event.durationMs ?? explicitEventTiming?.durationMs); - const explicitStartedAt = optionalTimestampValue(event.startedAt ?? event.traceStartedAt ?? event.runnerStartedAt ?? explicitEventTiming?.startedAt); - const provisionalLastEventAt = latestTimestamp(previousTiming?.lastEventAt, explicitEventTiming?.lastEventAt, occurredAt); - const provisionalFinishedAt = sourceTerminal ? latestTimestamp(explicitEventTiming?.finishedAt, optionalTimestampValue(event.finishedAt), provisionalLastEventAt, occurredAt) : null; - const startedAt = previousTiming?.startedAt ?? explicitStartedAt ?? startedAtFromDuration(provisionalFinishedAt, explicitDurationMs) ?? (sourceSeq <= 1 ? occurredAt : null); + const provisionalLastEventAt = latestTimestamp(previousTiming?.lastEventAt, occurredAt); + const provisionalFinishedAt = sourceTerminal ? provisionalLastEventAt : null; + const startedAt = previousTiming?.startedAt ?? occurredAt; const lastEventAt = provisionalLastEventAt; const finishedAt = sourceTerminal ? latestTimestamp(provisionalFinishedAt, lastEventAt) : null; const sourceEventId = workbenchSourceEventId(event, { traceId, sourceSeq, occurredAt }); @@ -226,7 +223,7 @@ export function buildWorkbenchProjectionEventFacts({ event = {}, requestMeta = { const terminalSealBlocked = sourceTerminal && sourceStatus === "completed" && !messageFinalText; const effectiveTerminal = sourceTerminal && !terminalSealBlocked; const effectiveStatus = effectiveTerminal ? sourceStatus : "running"; - const timing = eventTimingProjection({ startedAt, lastEventAt, finishedAt, terminal: effectiveTerminal, durationMs: explicitDurationMs ?? previousTiming?.durationMs }); + const timing = eventTimingProjection({ startedAt, lastEventAt, finishedAt, terminal: effectiveTerminal }); const timingAuthorityIssue = terminalTimingAuthorityIssue(timing, { terminal: effectiveTerminal, traceId, source: "kafka", status: effectiveStatus, label: event.label, sourceSeq }); const projectionDiagnostic = { timingAuthorityIssue, terminalSealBlocked, sourceTerminal, sourceStatus, authority: "agentrun-kafka-projector", valuesRedacted: true }; const userMessageFact = sessionId && previousCheckpoint?.userMessage ? normalizeMessageFact(previousCheckpoint.userMessage, 0, { traceId, sessionId, turnId, terminal: false, terminalStatus: "sent", finalText: null, timestamp: projectedAt, timing }) : null; @@ -499,13 +496,6 @@ function elapsedBetween(startedAt, endedAt) { return Math.trunc(end - start); } -function startedAtFromDuration(finishedAt, durationMs) { - const end = Date.parse(String(finishedAt ?? "")); - const duration = durationValue(durationMs); - if (!Number.isFinite(end) || duration === null) return null; - return new Date(Math.max(0, end - duration)).toISOString(); -} - function terminalTimingAuthorityIssue(timing, { terminal = false, traceId = null, source = null, status = null, label = null, sourceSeq = null } = {}) { if (!terminal) return null; const startedAt = optionalTimestampValue(timing?.startedAt); diff --git a/internal/cloud/workbench-realtime-capabilities.ts b/internal/cloud/workbench-realtime-capabilities.ts index 8d87d200..01f73093 100644 --- a/internal/cloud/workbench-realtime-capabilities.ts +++ b/internal/cloud/workbench-realtime-capabilities.ts @@ -1,35 +1,15 @@ -// SPEC: PJ2026-0104010803 Workbench composable realtime capabilities. -// Responsibility: validate YAML-owned capability switches without code defaults or implied mode selection. +// SPEC: PJ2026-0104010803 Workbench fixed transactional realtime authority. +// Responsibility: expose the code-owned Workbench projector/outbox/SSE invariant. -export const WORKBENCH_REALTIME_CAPABILITY_ENVS = Object.freeze({ - directPublish: "HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED", - liveKafkaSse: "HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", - kafkaRefreshReplay: "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", - transactionalProjector: "HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED", - projectionOutboxRelay: "HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED", - projectionRealtime: "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED" +export const WORKBENCH_REALTIME_CAPABILITIES = Object.freeze({ + directPublish: false, + liveKafkaSse: false, + kafkaRefreshReplay: false, + transactionalProjector: true, + projectionOutboxRelay: true, + projectionRealtime: true }); -export function workbenchRealtimeCapabilities(env = process.env) { - return Object.fromEntries(Object.entries(WORKBENCH_REALTIME_CAPABILITY_ENVS).map(([key, name]) => [key, requiredBooleanEnv(env, name)])); -} - -export function requiredBooleanEnv(env, name) { - const value = String(env?.[name] ?? "").trim().toLowerCase(); - if (value === "true" || value === "1") return true; - if (value === "false" || value === "0") return false; - throw capabilityError( - "hwlab_workbench_realtime_capability_invalid", - `${name} is required and must be explicitly true or false in the owning YAML.`, - name - ); -} - -function capabilityError(code, message, envName) { - return Object.assign(new Error(message), { - code, - envName, - statusCode: 503, - valuesRedacted: true - }); +export function workbenchRealtimeCapabilities() { + return WORKBENCH_REALTIME_CAPABILITIES; } diff --git a/internal/dev-entrypoint/cloud-web-runtime.mjs b/internal/dev-entrypoint/cloud-web-runtime.mjs index 25f1e254..096f6cda 100644 --- a/internal/dev-entrypoint/cloud-web-runtime.mjs +++ b/internal/dev-entrypoint/cloud-web-runtime.mjs @@ -400,11 +400,6 @@ function validateDisplayLocale(locale) { function workbenchRuntimeConfigFromEnv() { const traceTimeline = {}; const result = { - realtimeFeatures: { - liveKafkaSse: requiredWorkbenchRealtimeFeature(process.env.HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED, "HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED"), - kafkaRefreshReplay: requiredWorkbenchRealtimeFeature(process.env.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED"), - projectionRealtime: requiredWorkbenchRealtimeFeature(process.env.HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED, "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED") - }, debugCapabilities: { isolatedKafka: requiredWorkbenchRealtimeFeature(process.env.HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED, "HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED"), rawHwlabEventWindow: { diff --git a/internal/dev-entrypoint/cloud-web-runtime.test.mjs b/internal/dev-entrypoint/cloud-web-runtime.test.mjs index 89504981..f2924882 100644 --- a/internal/dev-entrypoint/cloud-web-runtime.test.mjs +++ b/internal/dev-entrypoint/cloud-web-runtime.test.mjs @@ -1,5 +1,5 @@ import assert from "node:assert/strict"; -import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import { createServer } from "node:http"; import { connect } from "node:net"; import os from "node:os"; @@ -16,7 +16,25 @@ test("cloud web classifies Bun client disconnect ECONNRESET as non-fatal", () => assert.equal(isClientDisconnectError(new Error("database migration failed")), false); }); -test("serveCloudWeb rejects a missing refresh replay capability without opening health or SPA routes", async () => { +test("repo feature schema owns one unique key per product feature", async () => { + const schema = JSON.parse(await readFile(new URL("../../config/feature-config.schema.json", import.meta.url), "utf8")); + assert.equal(schema.$schema, "https://json-schema.org/draft/2020-12/schema"); + assert.deepEqual(Object.keys(schema.properties), ["tracePanel", "rawEvents", "debugReplay", "views"]); + const features = Object.values(schema.properties).map((property) => property["x-unidesk-feature"]); + assert.equal(features.every((feature) => typeof feature === "string" && feature.length > 0), true); + assert.equal(new Set(features).size, features.length); + const encoded = JSON.stringify(schema); + for (const architectureEnv of [ + "HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED", + "HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", + "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", + "HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED", + "HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED", + "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED" + ]) assert.equal(encoded.includes(architectureEnv), false); +}); + +test("serveCloudWeb starts without obsolete architecture capabilities", async () => { const restoreEnv = withEnv({ HWLAB_CLOUD_WEB_DISPLAY_TIME_ZONE: "Asia/Shanghai", HWLAB_CLOUD_WEB_DISPLAY_TIME_LOCALE: "zh-CN", @@ -29,19 +47,21 @@ test("serveCloudWeb rejects a missing refresh replay capability without opening HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_ENTRIES: "200", HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_RETAINED_BYTES: "1048576" }); - delete process.env.HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED; const port = await unusedPort(); + let server = null; try { - await assert.rejects(serveCloudWeb({ + server = await serveCloudWeb({ port, serviceId: "hwlab-cloud-web", healthPayload: () => ({ status: "ok" }), - sendJson() { - assert.fail("missing runtime configuration must fail before serving any response"); + sendJson(response, statusCode, body) { + response.writeHead(statusCode, { "content-type": "application/json" }); + response.end(JSON.stringify(body)); } - }), /HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED/u); - assert.equal(await canConnect(port), false, "serveCloudWeb must not bind before runtime configuration is valid"); + }); + assert.equal(await canConnect(port), true); } finally { + if (server) await close(server); restoreEnv(); } }); @@ -497,7 +517,7 @@ test("cloud web serves client deep links through the Vue shell", async () => { const html = await response.text(); assert.match(html, /
<\/div>/u, route); assert.match(html, /HWLAB_CLOUD_WEB_CONFIG/u, route); - assert.match(html, /"realtimeFeatures":\{"liveKafkaSse":true,"kafkaRefreshReplay":true,"projectionRealtime":false\}/u, route); + assert.doesNotMatch(html, /realtimeFeatures|liveKafkaSse|kafkaRefreshReplay|projectionRealtime/u, route); assert.match(html, /"debugCapabilities":\{"isolatedKafka":true,"rawHwlabEventWindow":\{"enabled":true,"maxEntries":200,"maxRetainedBytes":1048576\}\}/u, route); } diff --git a/web/hwlab-cloud-web/scripts/workbench-server-state.test.ts b/web/hwlab-cloud-web/scripts/workbench-server-state.test.ts index 21ee66a2..285b99d7 100644 --- a/web/hwlab-cloud-web/scripts/workbench-server-state.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-server-state.test.ts @@ -270,7 +270,7 @@ test("Workbench sealed completed final body survives stale running session messa assert.equal(selectActiveSession(state, "ses_final_seal")?.status, "completed"); }); -test("Workbench session messages repair terminal zero timing from running projection", () => { +test("Workbench session messages preserve canonical terminal zero timing", () => { let state = createWorkbenchServerState(); state = reduceWorkbenchServerState(state, { type: "session.messages", @@ -288,8 +288,8 @@ test("Workbench session messages repair terminal zero timing from running projec const agent = selectActiveMessages(state, "ses_terminal_zero_repair").find((message) => message.role === "agent"); assert.equal(agent?.status, "completed"); - assert.equal(agent?.durationMs, 6_000); - assert.equal(agent?.timing?.durationMs, 6_000); + assert.equal(agent?.durationMs, 0); + assert.equal(agent?.timing?.durationMs, 0); }); test("Workbench session messages can replace sealed terminal zero with positive timing", () => { diff --git a/web/hwlab-cloud-web/src/config/runtime.ts b/web/hwlab-cloud-web/src/config/runtime.ts index 5936ba7a..54387e08 100644 --- a/web/hwlab-cloud-web/src/config/runtime.ts +++ b/web/hwlab-cloud-web/src/config/runtime.ts @@ -98,11 +98,7 @@ export function workbenchTraceTimelinePolicy(): WorkbenchTraceTimelinePolicy { } export function workbenchRealtimeCapabilities(): WorkbenchRealtimeCapabilities { - const features = window.HWLAB_CLOUD_WEB_CONFIG?.workbench?.realtimeFeatures; - if (typeof features?.liveKafkaSse !== "boolean" || typeof features?.kafkaRefreshReplay !== "boolean" || typeof features?.projectionRealtime !== "boolean") { - throw new Error("HWLAB Cloud Web workbench.realtimeFeatures are required"); - } - return { liveKafkaSse: features.liveKafkaSse, kafkaRefreshReplay: features.kafkaRefreshReplay, projectionRealtime: features.projectionRealtime }; + return { liveKafkaSse: false, kafkaRefreshReplay: false, projectionRealtime: true }; } export function workbenchDebugCapabilities(): WorkbenchDebugCapabilities { diff --git a/web/hwlab-cloud-web/src/types/global.d.ts b/web/hwlab-cloud-web/src/types/global.d.ts index 3bafc4d3..5106f04e 100644 --- a/web/hwlab-cloud-web/src/types/global.d.ts +++ b/web/hwlab-cloud-web/src/types/global.d.ts @@ -8,11 +8,6 @@ declare global { label?: string; }; workbench?: { - realtimeFeatures?: { - liveKafkaSse?: boolean; - kafkaRefreshReplay?: boolean; - projectionRealtime?: boolean; - }; debugCapabilities?: { isolatedKafka?: boolean; rawHwlabEventWindow?: {