diff --git a/deploy/deploy.schema.json b/deploy/deploy.schema.json index d09c0da3..c15fd5af 100644 --- a/deploy/deploy.schema.json +++ b/deploy/deploy.schema.json @@ -336,6 +336,7 @@ "runtimePath": { "type": "string" }, "imageTagMode": { "type": "string", "enum": ["short", "full"] }, "workbench": { "$ref": "#/$defs/workbenchUiPolicy" }, + "agentObserver": { "$ref": "#/$defs/agentObserverPolicy" }, "observability": { "$ref": "#/$defs/runtimeObservabilityPolicy" }, "opencode": { "$ref": "#/$defs/opencodeRuntime" }, "runtimeStore": { "$ref": "#/$defs/runtimeStore" }, @@ -610,6 +611,46 @@ }, "additionalProperties": false }, + "agentObserverPolicy": { + "type": "object", + "required": [ + "retentionEventLimit", + "retentionScanLimit", + "retentionTimeoutMs", + "retentionGroupPrefix", + "sseHeartbeatMs", + "reconnectDelayMs", + "batchMaxEvents", + "liveBufferLimit", + "reducerEventLimit", + "reducerEntityLimit", + "treeNodeLimit", + "inspectorEventTailLimit", + "virtualListWindowLimit", + "changeHighlightMs", + "staleAfterMs", + "reducedMotion" + ], + "properties": { + "retentionEventLimit": { "type": "integer", "minimum": 1 }, + "retentionScanLimit": { "type": "integer", "minimum": 1 }, + "retentionTimeoutMs": { "type": "integer", "minimum": 250 }, + "retentionGroupPrefix": { "type": "string", "minLength": 1 }, + "sseHeartbeatMs": { "type": "integer", "minimum": 1000 }, + "reconnectDelayMs": { "type": "integer", "minimum": 100 }, + "batchMaxEvents": { "type": "integer", "minimum": 1 }, + "liveBufferLimit": { "type": "integer", "minimum": 1 }, + "reducerEventLimit": { "type": "integer", "minimum": 1 }, + "reducerEntityLimit": { "type": "integer", "minimum": 1 }, + "treeNodeLimit": { "type": "integer", "minimum": 1 }, + "inspectorEventTailLimit": { "type": "integer", "minimum": 1 }, + "virtualListWindowLimit": { "type": "integer", "minimum": 1 }, + "changeHighlightMs": { "type": "integer", "minimum": 0 }, + "staleAfterMs": { "type": "integer", "minimum": 1000 }, + "reducedMotion": { "type": "string", "enum": ["respect-system"] } + }, + "additionalProperties": false + }, "externalPostgres": { "type": "object", "required": ["enabled", "serviceName", "endpointAddress", "port"], diff --git a/deploy/deploy.yaml b/deploy/deploy.yaml index de4e2375..388ab61c 100644 --- a/deploy/deploy.yaml +++ b/deploy/deploy.yaml @@ -378,6 +378,23 @@ lanes: enabled: true maxEntries: 200 maxRetainedBytes: 1048576 + agentObserver: + retentionEventLimit: 10000 + retentionScanLimit: 1000000 + retentionTimeoutMs: 30000 + retentionGroupPrefix: hwlab-v03-agent-observer-replay + sseHeartbeatMs: 15000 + reconnectDelayMs: 1500 + batchMaxEvents: 100 + liveBufferLimit: 2000 + reducerEventLimit: 2000 + reducerEntityLimit: 500 + treeNodeLimit: 500 + inspectorEventTailLimit: 100 + virtualListWindowLimit: 160 + changeHighlightMs: 1200 + staleAfterMs: 30000 + reducedMotion: respect-system observability: traceExplorerUrlTemplate: /v1/workbench/traces/{trace_id}/events prometheusOperatorResources: false diff --git a/internal/cloud/kafka-event-bridge.ts b/internal/cloud/kafka-event-bridge.ts index 3f2670ac..b8837d59 100644 --- a/internal/cloud/kafka-event-bridge.ts +++ b/internal/cloud/kafka-event-bridge.ts @@ -1317,9 +1317,17 @@ function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run, sourceSeq, runId, commandId, + taskId: firstText(payload.taskId, sourceEvent.taskId), attemptId: firstText(command.attemptId, sourceEvent.attemptId, payload.attemptId), runnerId: firstText(command.runnerId, sourceEvent.runnerId, payload.runnerId), backend: firstText(run.backendProfile, payload.backendProfile, payload.providerProfile), + providerId: firstText(run.providerId, payload.providerId), + projectId: firstText(run.projectId, payload.projectId), + workspace: firstText(payload.workspace, payload.targetWorkspace, payload.cwd), + runStatus: firstText(run.status, payload.runStatus), + terminalStatus: firstText(run.terminalStatus, payload.terminalStatus), + commandState: firstText(command.state, payload.commandState), + commandType: firstText(command.type, payload.commandType), agentRunEventType: type, createdAt: timestampValue(sourceEvent.createdAt ?? input.producedAt), valuesPrinted: false diff --git a/internal/cloud/server-agent-observer-http.ts b/internal/cloud/server-agent-observer-http.ts new file mode 100644 index 00000000..e120f0b7 --- /dev/null +++ b/internal/cloud/server-agent-observer-http.ts @@ -0,0 +1,169 @@ +// SPEC: PJ2026-010405 CONSOLE-REQ-007; MDTODO R3. +// Responsibility: expose one authenticated Kafka retention-to-live SSE stream for the AgentRun observer. + +import { createWorkbenchKafkaRefreshHandoff } from "./workbench-kafka-refresh-handoff.ts"; + +export async function handleAgentObserverHttp(request, response, options = {}) { + if (request.method !== "GET") { + response.writeHead(405, { allow: "GET", "content-type": "application/json; charset=utf-8" }); + response.end(JSON.stringify({ ok: false, error: { code: "method_not_allowed", message: "GET required." } })); + return; + } + + const bridge = options.kafkaEventBridge; + if (!bridge?.started || typeof bridge.subscribeLiveHwlabEvents !== "function" || typeof bridge.queryHwlabEventRetention !== "function") { + sendJson(response, 503, observerError("agent_observer_kafka_unavailable", "Agent observer Kafka retention/live runtime is unavailable.")); + return; + } + + let policy; + try { + policy = agentObserverPolicy(options.env ?? process.env); + } catch (error) { + sendJson(response, 503, observerError(error?.code ?? "agent_observer_config_invalid", error instanceof Error ? error.message : String(error))); + return; + } + + try { + await (bridge.liveReady ?? bridge.ready); + } catch (error) { + sendJson(response, 503, observerError(error?.code ?? "agent_observer_kafka_start_failed", error instanceof Error ? error.message : String(error))); + return; + } + + let closed = false; + const cleanup = []; + const active = () => !closed && !response.destroyed && !response.writableEnded; + const close = () => { + if (closed) return; + closed = true; + for (const dispose of cleanup.splice(0)) dispose(); + }; + response.once("close", close); + request.once?.("aborted", close); + request.socket?.once?.("close", close); + cleanup.push(() => request.off?.("aborted", close)); + cleanup.push(() => request.socket?.off?.("close", close)); + + response.writeHead(200, { + "content-type": "text/event-stream; charset=utf-8", + "cache-control": "no-store, no-transform", + connection: "keep-alive", + "x-accel-buffering": "no", + "x-content-type-options": "nosniff" + }); + response.flushHeaders?.(); + + const writeEvent = async (name, payload) => { + if (!active()) return false; + const eventId = text(payload?.eventId ?? payload?.sourceEventId); + const block = `${eventId ? `id: ${eventId}\n` : ""}event: ${name}\ndata: ${JSON.stringify(payload)}\n\n`; + if (!response.write(block)) await waitForDrain(response, active); + return active(); + }; + let writeChain = Promise.resolve(true); + const enqueue = (name, payload) => { + writeChain = writeChain.then(() => writeEvent(name, payload)); + return writeChain; + }; + const fail = async (error) => { + if (!active()) return; + await enqueue("agent-observer.error", observerError(error?.code ?? "agent_observer_stream_failed", error instanceof Error ? error.message : String(error)).error); + await writeChain; + if (!response.writableEnded) response.end(); + }; + + const handoff = createWorkbenchKafkaRefreshHandoff({ + allowUnscoped: true, + liveBufferLimit: policy.liveBufferLimit, + subscribeLive: (listener) => bridge.subscribeLiveHwlabEvents(listener), + queryRetention: ({ signal }) => bridge.queryHwlabEventRetention({ + limit: policy.retentionEventLimit, + scanLimit: policy.retentionScanLimit, + timeoutMs: policy.retentionTimeoutMs, + groupIdPrefix: policy.retentionGroupPrefix, + fromBeginning: true, + signal + }), + deliverEvent: (envelope) => enqueue("hwlab.event.v1", envelope), + deliverConnected: (summary) => enqueue("agent-observer.connected", { + type: "connected", + status: "live", + deliverySemantics: "kafka-retention-then-live", + authority: "hwlab.event.v1", + replay: summary, + serverSentAt: new Date().toISOString(), + valuesPrinted: false + }), + onFailure: fail + }); + cleanup.push(() => handoff.stop("connection-closed")); + + try { + await handoff.start(); + } catch (error) { + await fail(error); + return; + } + if (!active()) return; + + const heartbeat = setInterval(() => { + void enqueue("agent-observer.heartbeat", { + type: "heartbeat", + status: "live", + authority: "hwlab.event.v1", + serverSentAt: new Date().toISOString(), + valuesPrinted: false + }); + }, policy.heartbeatMs); + heartbeat.unref?.(); + cleanup.push(() => clearInterval(heartbeat)); +} + +function agentObserverPolicy(env) { + return { + retentionEventLimit: positiveInteger(env.HWLAB_AGENT_OBSERVER_RETENTION_EVENT_LIMIT, "HWLAB_AGENT_OBSERVER_RETENTION_EVENT_LIMIT"), + retentionScanLimit: positiveInteger(env.HWLAB_AGENT_OBSERVER_RETENTION_SCAN_LIMIT, "HWLAB_AGENT_OBSERVER_RETENTION_SCAN_LIMIT"), + retentionTimeoutMs: positiveInteger(env.HWLAB_AGENT_OBSERVER_RETENTION_TIMEOUT_MS, "HWLAB_AGENT_OBSERVER_RETENTION_TIMEOUT_MS"), + retentionGroupPrefix: requiredText(env.HWLAB_AGENT_OBSERVER_RETENTION_GROUP_PREFIX, "HWLAB_AGENT_OBSERVER_RETENTION_GROUP_PREFIX"), + liveBufferLimit: positiveInteger(env.HWLAB_AGENT_OBSERVER_LIVE_BUFFER_LIMIT, "HWLAB_AGENT_OBSERVER_LIVE_BUFFER_LIMIT"), + heartbeatMs: positiveInteger(env.HWLAB_AGENT_OBSERVER_SSE_HEARTBEAT_MS, "HWLAB_AGENT_OBSERVER_SSE_HEARTBEAT_MS") + }; +} + +function positiveInteger(value, name) { + const number = Number(value); + if (!Number.isInteger(number) || number <= 0) throw Object.assign(new Error(`${name} must be a positive integer.`), { code: "agent_observer_config_invalid" }); + return number; +} + +function requiredText(value, name) { + const result = text(value); + if (!result) throw Object.assign(new Error(`${name} is required.`), { code: "agent_observer_config_invalid" }); + return result; +} + +function text(value) { + return typeof value === "string" && value.trim() ? value.trim() : null; +} + +function observerError(code, message) { + return { ok: false, error: { code, message, authority: "hwlab.event.v1", valuesRedacted: true } }; +} + +function sendJson(response, status, payload) { + response.writeHead(status, { "content-type": "application/json; charset=utf-8", "cache-control": "no-store" }); + response.end(JSON.stringify(payload)); +} + +function waitForDrain(response, active) { + return new Promise((resolve) => { + const finish = () => { + response.off?.("drain", finish); + response.off?.("close", finish); + resolve(active()); + }; + response.once("drain", finish); + response.once("close", finish); + }); +} diff --git a/internal/cloud/server.ts b/internal/cloud/server.ts index 148283d8..c4d0f4af 100644 --- a/internal/cloud/server.ts +++ b/internal/cloud/server.ts @@ -71,6 +71,7 @@ import { drainWorkbenchRealtimeConnections, handleWorkbenchReadModelHttp, handle import { handleWorkbenchDebugFakeSseHttp } from "./workbench-debug-fake-sse.ts"; import { handleWorkbenchKafkaSseDebugHttp } from "./workbench-kafka-sse-debug.ts"; import { handleWorkbenchLaunchHttp } from "./server-workbench-launch-http.ts"; +import { handleAgentObserverHttp } from "./server-agent-observer-http.ts"; import { startWorkbenchEmptySessionGc } from "./workbench-empty-session-gc.ts"; import { startHwlabKafkaEventBridge } from "./kafka-event-bridge.ts"; import { handleM3IoControlHttp } from "./server-m3-http.ts"; @@ -758,6 +759,11 @@ async function handleRestAdapter(request, response, url, options) { return; } + if (url.pathname === "/v1/agent-observer/events") { + await handleAgentObserverHttp(request, response, options); + return; + } + if (url.pathname === "/v1/workbench/launches") { await handleWorkbenchLaunchHttp(request, response, options); return; @@ -1012,6 +1018,7 @@ function navIdForRestPath(pathname, method = "GET") { if (pathname === "/v1/users/me/profile" || pathname === "/v1/users/me/password") return "system.settings"; if (pathname === "/v1/workbench/debug/fake-sse" || pathname.startsWith("/v1/workbench/debug/fake-sse/") || pathname === "/v1/workbench/debug/kafka-sse" || pathname.startsWith("/v1/workbench/debug/kafka-sse/")) return "workbench.debug"; if (pathname === "/v1/workbench/events" || pathname === "/v1/workbench/projection-events" || pathname === "/v1/workbench/launches" || pathname === "/v1/workbench/sessions" || pathname.startsWith("/v1/workbench/sessions/") || pathname.startsWith("/v1/workbench/turns/") || pathname.startsWith("/v1/workbench/traces/")) return "workbench.code"; + if (pathname === "/v1/agent-observer/events") return "workbench.code"; if (pathname === "/v1/agent/chat" || pathname === "/v1/agent/sessions" || pathname.startsWith("/v1/agent/sessions/") || pathname === "/v1/agent/chat/inspect" || pathname.startsWith("/v1/agent/chat/result/") || pathname.startsWith("/v1/agent/turns/") || pathname.startsWith("/v1/agent/traces/") || pathname === "/v1/agent/chat/cancel" || pathname === "/v1/agent/chat/steer") return "workbench.code"; if (pathname === "/v1/admin/provider-profiles" || pathname.startsWith("/v1/admin/provider-profiles/")) return "admin.providerProfiles"; if (pathname === "/v1/admin/secrets" || pathname.startsWith("/v1/admin/secrets/")) return "admin.secrets"; diff --git a/internal/cloud/workbench-kafka-refresh-handoff.ts b/internal/cloud/workbench-kafka-refresh-handoff.ts index 316d04e9..71c664cb 100644 --- a/internal/cloud/workbench-kafka-refresh-handoff.ts +++ b/internal/cloud/workbench-kafka-refresh-handoff.ts @@ -4,8 +4,9 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { const sessionId = textValue(options.sessionId); const traceId = textValue(options.traceId); + const allowUnscoped = options.allowUnscoped === true; const liveBufferLimit = positiveInteger(options.liveBufferLimit); - if (!sessionId && !traceId) throw handoffError("workbench_kafka_refresh_scope_missing", "Kafka refresh handoff requires a session or trace scope.", "idle"); + if (!sessionId && !traceId && !allowUnscoped) throw handoffError("workbench_kafka_refresh_scope_missing", "Kafka refresh handoff requires a session or trace scope.", "idle"); if (!liveBufferLimit) throw handoffError("workbench_kafka_refresh_buffer_limit_invalid", "Kafka refresh handoff requires an explicit positive live buffer limit.", "idle"); if (typeof options.subscribeLive !== "function" || typeof options.queryRetention !== "function" || typeof options.deliverEvent !== "function" || typeof options.deliverConnected !== "function") { throw handoffError("workbench_kafka_refresh_runtime_invalid", "Kafka refresh handoff runtime callbacks are incomplete.", "idle"); @@ -169,7 +170,7 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { function validateReplayRecords(result) { const records = Array.isArray(result?.events) ? result.events : []; counts.matched = records.length; - if (traceId && records.length === 0) { + if (!allowUnscoped && traceId && records.length === 0) { throw handoffError("workbench_kafka_refresh_trace_empty", "Trace-scoped Kafka refresh found no retained events after a complete scan.", phase); } const replayPartitions = new Set(); diff --git a/internal/dev-entrypoint/cloud-web-proxy.mjs b/internal/dev-entrypoint/cloud-web-proxy.mjs index 8bb66c7a..3381d670 100644 --- a/internal/dev-entrypoint/cloud-web-proxy.mjs +++ b/internal/dev-entrypoint/cloud-web-proxy.mjs @@ -20,7 +20,8 @@ const FORWARDED_REQUEST_HEADERS = Object.freeze([ ]); export function isCloudWebSseRoute(pathname) { - return /^\/v1\/agent\/chat\/trace\/[^/]+\/stream$/u.test(String(pathname || "")); + const value = String(pathname || ""); + return value === "/v1/agent-observer/events" || /^\/v1\/agent\/chat\/trace\/[^/]+\/stream$/u.test(value); } export function isEventStreamResponse(headers = {}) { diff --git a/internal/dev-entrypoint/cloud-web-runtime.mjs b/internal/dev-entrypoint/cloud-web-runtime.mjs index 25f1e254..83c487ef 100644 --- a/internal/dev-entrypoint/cloud-web-runtime.mjs +++ b/internal/dev-entrypoint/cloud-web-runtime.mjs @@ -171,6 +171,8 @@ async function runtimeConfigFromEnv(options = {}) { const config = { displayTime: displayTimeConfigFromEnv() }; const workbench = workbenchRuntimeConfigFromEnv(); if (workbench) config.workbench = workbench; + const agentObserver = agentObserverRuntimeConfigFromEnv(); + if (agentObserver) config.agentObserver = agentObserver; const opencode = await opencodeRuntimeConfigFromEnv(options); if (opencode) config.opencode = opencode; return config; @@ -424,6 +426,52 @@ function workbenchRuntimeConfigFromEnv() { return Object.keys(result).length > 0 ? result : null; } +function agentObserverRuntimeConfigFromEnv() { + const names = [ + "HWLAB_AGENT_OBSERVER_RECONNECT_DELAY_MS", + "HWLAB_AGENT_OBSERVER_BATCH_MAX_EVENTS", + "HWLAB_AGENT_OBSERVER_LIVE_BUFFER_LIMIT", + "HWLAB_AGENT_OBSERVER_REDUCER_EVENT_LIMIT", + "HWLAB_AGENT_OBSERVER_REDUCER_ENTITY_LIMIT", + "HWLAB_AGENT_OBSERVER_TREE_NODE_LIMIT", + "HWLAB_AGENT_OBSERVER_INSPECTOR_EVENT_TAIL_LIMIT", + "HWLAB_AGENT_OBSERVER_VIRTUAL_LIST_WINDOW_LIMIT", + "HWLAB_AGENT_OBSERVER_CHANGE_HIGHLIGHT_MS", + "HWLAB_AGENT_OBSERVER_STALE_AFTER_MS", + "HWLAB_AGENT_OBSERVER_REDUCED_MOTION" + ]; + if (names.every((name) => process.env[name] === undefined || process.env[name] === "")) return null; + const value = { + nodeId: requiredRuntimeConfigValue(process.env.HWLAB_AGENT_OBSERVER_NODE_ID, "HWLAB_AGENT_OBSERVER_NODE_ID"), + lane: requiredRuntimeConfigValue(process.env.HWLAB_AGENT_OBSERVER_LANE, "HWLAB_AGENT_OBSERVER_LANE"), + reconnectDelayMs: requiredPositiveWorkbenchInteger(process.env.HWLAB_AGENT_OBSERVER_RECONNECT_DELAY_MS, "HWLAB_AGENT_OBSERVER_RECONNECT_DELAY_MS"), + batchMaxEvents: requiredPositiveWorkbenchInteger(process.env.HWLAB_AGENT_OBSERVER_BATCH_MAX_EVENTS, "HWLAB_AGENT_OBSERVER_BATCH_MAX_EVENTS"), + liveBufferLimit: requiredPositiveWorkbenchInteger(process.env.HWLAB_AGENT_OBSERVER_LIVE_BUFFER_LIMIT, "HWLAB_AGENT_OBSERVER_LIVE_BUFFER_LIMIT"), + reducerEventLimit: requiredPositiveWorkbenchInteger(process.env.HWLAB_AGENT_OBSERVER_REDUCER_EVENT_LIMIT, "HWLAB_AGENT_OBSERVER_REDUCER_EVENT_LIMIT"), + reducerEntityLimit: requiredPositiveWorkbenchInteger(process.env.HWLAB_AGENT_OBSERVER_REDUCER_ENTITY_LIMIT, "HWLAB_AGENT_OBSERVER_REDUCER_ENTITY_LIMIT"), + treeNodeLimit: requiredPositiveWorkbenchInteger(process.env.HWLAB_AGENT_OBSERVER_TREE_NODE_LIMIT, "HWLAB_AGENT_OBSERVER_TREE_NODE_LIMIT"), + inspectorEventTailLimit: requiredPositiveWorkbenchInteger(process.env.HWLAB_AGENT_OBSERVER_INSPECTOR_EVENT_TAIL_LIMIT, "HWLAB_AGENT_OBSERVER_INSPECTOR_EVENT_TAIL_LIMIT"), + virtualListWindowLimit: requiredPositiveWorkbenchInteger(process.env.HWLAB_AGENT_OBSERVER_VIRTUAL_LIST_WINDOW_LIMIT, "HWLAB_AGENT_OBSERVER_VIRTUAL_LIST_WINDOW_LIMIT"), + changeHighlightMs: requiredNonNegativeInteger(process.env.HWLAB_AGENT_OBSERVER_CHANGE_HIGHLIGHT_MS, "HWLAB_AGENT_OBSERVER_CHANGE_HIGHLIGHT_MS"), + staleAfterMs: requiredPositiveWorkbenchInteger(process.env.HWLAB_AGENT_OBSERVER_STALE_AFTER_MS, "HWLAB_AGENT_OBSERVER_STALE_AFTER_MS"), + reducedMotion: requiredRuntimeConfigValue(process.env.HWLAB_AGENT_OBSERVER_REDUCED_MOTION, "HWLAB_AGENT_OBSERVER_REDUCED_MOTION") + }; + if (value.reducedMotion !== "respect-system") throw new Error("HWLAB_AGENT_OBSERVER_REDUCED_MOTION must be respect-system"); + return value; +} + +function requiredNonNegativeInteger(value, name) { + const number = Number(value); + if (!Number.isInteger(number) || number < 0) throw new Error(`${name} is required and must be a non-negative integer`); + return number; +} + +function requiredRuntimeConfigValue(value, name) { + const result = String(value ?? "").trim(); + if (!result) throw new Error(`${name} is required`); + return result; +} + function requiredWorkbenchRealtimeFeature(value, name) { const normalized = String(value ?? "").trim().toLowerCase(); if (normalized === "true" || normalized === "1") return true; diff --git a/scripts/src/gitops-render/core.mjs b/scripts/src/gitops-render/core.mjs index 3285aaf0..5ec3a9e6 100644 --- a/scripts/src/gitops-render/core.mjs +++ b/scripts/src/gitops-render/core.mjs @@ -1147,6 +1147,13 @@ function runtimeWorkbenchRawHwlabEventWindowPolicy(deploy, profile) { }; } +function runtimeAgentObserverPolicy(deploy, profile) { + if (!isRuntimeLane(profile)) return null; + const policy = runtimeLaneConfig(deploy, profile)?.agentObserver; + if (!policy || typeof policy !== "object" || Array.isArray(policy)) return null; + return cloneJson(policy); +} + function runtimeTraceExplorerUrlTemplate(deploy, profile) { if (!isRuntimeLane(profile)) return null; const template = runtimeLaneConfig(deploy, profile)?.observability?.traceExplorerUrlTemplate; @@ -1556,6 +1563,31 @@ function transformWorkloads({ workloads, deploy, catalog, source, sourceBranch = if (serviceId === "hwlab-cloud-api" || serviceId === "hwlab-edge-proxy") { upsertEnv(container.env, "HWLAB_PUBLIC_ENDPOINT", runtimeEndpoint); } + if (runtimeLane && (serviceId === "hwlab-cloud-api" || serviceId === "hwlab-cloud-web")) { + const agentObserverPolicy = runtimeAgentObserverPolicy(deploy, profile); + if (agentObserverPolicy) { + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_RETENTION_EVENT_LIMIT", String(agentObserverPolicy.retentionEventLimit)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_RETENTION_SCAN_LIMIT", String(agentObserverPolicy.retentionScanLimit)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_RETENTION_TIMEOUT_MS", String(agentObserverPolicy.retentionTimeoutMs)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_RETENTION_GROUP_PREFIX", String(agentObserverPolicy.retentionGroupPrefix)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_SSE_HEARTBEAT_MS", String(agentObserverPolicy.sseHeartbeatMs)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_RECONNECT_DELAY_MS", String(agentObserverPolicy.reconnectDelayMs)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_BATCH_MAX_EVENTS", String(agentObserverPolicy.batchMaxEvents)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_LIVE_BUFFER_LIMIT", String(agentObserverPolicy.liveBufferLimit)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_REDUCER_EVENT_LIMIT", String(agentObserverPolicy.reducerEventLimit)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_REDUCER_ENTITY_LIMIT", String(agentObserverPolicy.reducerEntityLimit)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_TREE_NODE_LIMIT", String(agentObserverPolicy.treeNodeLimit)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_INSPECTOR_EVENT_TAIL_LIMIT", String(agentObserverPolicy.inspectorEventTailLimit)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_VIRTUAL_LIST_WINDOW_LIMIT", String(agentObserverPolicy.virtualListWindowLimit)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_CHANGE_HIGHLIGHT_MS", String(agentObserverPolicy.changeHighlightMs)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_STALE_AFTER_MS", String(agentObserverPolicy.staleAfterMs)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_REDUCED_MOTION", String(agentObserverPolicy.reducedMotion)); + if (serviceId === "hwlab-cloud-web") { + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_NODE_ID", runtimeNodeIdForProfile(deploy, profile, nodeId)); + upsertEnv(container.env, "HWLAB_AGENT_OBSERVER_LANE", profile); + } + } + } if (runtimeLane && serviceId === "hwlab-edge-proxy") { useLocalHealthProbe(container, "/health"); } diff --git a/web/hwlab-cloud-web/src/api/agentObserver.ts b/web/hwlab-cloud-web/src/api/agentObserver.ts new file mode 100644 index 00000000..cb42567a --- /dev/null +++ b/web/hwlab-cloud-web/src/api/agentObserver.ts @@ -0,0 +1,26 @@ +export interface AgentObserverStreamCallbacks { + onEnvelope: (payload: unknown) => void; + onConnected: (payload: unknown) => void; + onHeartbeat: (payload: unknown) => void; + onStreamError: (payload: unknown) => void; + onTransportError: () => void; +} + +export function openAgentObserverStream(callbacks: AgentObserverStreamCallbacks): EventSource { + const source = new EventSource("/v1/agent-observer/events", { withCredentials: true }); + source.addEventListener("hwlab.event.v1", (event) => callbacks.onEnvelope(parseEvent(event))); + source.addEventListener("agent-observer.connected", (event) => callbacks.onConnected(parseEvent(event))); + source.addEventListener("agent-observer.heartbeat", (event) => callbacks.onHeartbeat(parseEvent(event))); + source.addEventListener("agent-observer.error", (event) => callbacks.onStreamError(parseEvent(event))); + source.onerror = () => callbacks.onTransportError(); + return source; +} + +function parseEvent(event: Event): unknown { + if (!(event instanceof MessageEvent) || typeof event.data !== "string") return null; + try { + return JSON.parse(event.data); + } catch { + return null; + } +} diff --git a/web/hwlab-cloud-web/src/components/layout/AppNavigation.vue b/web/hwlab-cloud-web/src/components/layout/AppNavigation.vue index d3970f71..6bd3f313 100644 --- a/web/hwlab-cloud-web/src/components/layout/AppNavigation.vue +++ b/web/hwlab-cloud-web/src/components/layout/AppNavigation.vue @@ -4,6 +4,7 @@ + +