From 4d10c59b35058bf289fb885fbfa024a0ad813e08 Mon Sep 17 00:00:00 2001 From: root Date: Wed, 15 Jul 2026 07:02:37 +0200 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E6=94=B6=E7=B4=A7?= =?UTF-8?q?=E6=99=BA=E8=83=BD=E4=BD=93=E8=A7=82=E5=AF=9F=E5=AE=9E=E6=97=B6?= =?UTF-8?q?=E8=AF=AD=E4=B9=89=E4=B8=8E=E6=9C=89=E7=95=8C=E6=81=A2=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/cloud/kafka-event-bridge.test.ts | 29 ++++++ internal/cloud/kafka-event-bridge.ts | 6 +- internal/cloud/server-agent-observer-http.ts | 12 +++ .../workbench-kafka-refresh-handoff.test.ts | 52 +++++++++- .../cloud/workbench-kafka-refresh-handoff.ts | 44 +++++++-- .../scripts/agent-observer-contract.test.ts | 65 ++++++++++++ web/hwlab-cloud-web/src/api/agentObserver.ts | 11 ++- .../src/stores/agent-observer.ts | 99 +++++++++++++------ .../src/views/agents/AgentRunsView.vue | 10 +- 9 files changed, 284 insertions(+), 44 deletions(-) create mode 100644 web/hwlab-cloud-web/scripts/agent-observer-contract.test.ts diff --git a/internal/cloud/kafka-event-bridge.test.ts b/internal/cloud/kafka-event-bridge.test.ts index 0bfd118c..463072a5 100644 --- a/internal/cloud/kafka-event-bridge.test.ts +++ b/internal/cloud/kafka-event-bridge.test.ts @@ -58,6 +58,35 @@ const LIVE_REFRESH_ENV = Object.freeze({ HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT: "2000" }); +test("projects only explicit AgentRun lifecycle timestamps", () => { + const projected = projectAgentRunKafkaEventToHwlabEvent({ + schema: "agentrun.event.v1", + eventType: "agentrun.run.event", + producedAt: "2026-07-09T10:00:09.000Z", + run: { + runId: "run_explicit_lifecycle", + sessionId: "ses_explicit_lifecycle", + startedAt: "2026-07-09T10:00:00.000Z", + finishedAt: "2026-07-09T10:00:08.000Z" + }, + command: { commandId: "cmd_explicit_lifecycle" }, + event: { seq: 1, type: "assistant_message", createdAt: "2026-07-09T10:00:05.000Z", payload: { text: "progress" } } + }, { source: "hwlab-test" }); + assert.equal(projected.event.startedAt, "2026-07-09T10:00:00.000Z"); + assert.equal(projected.event.finishedAt, "2026-07-09T10:00:08.000Z"); + + const withoutFacts = projectAgentRunKafkaEventToHwlabEvent({ + schema: "agentrun.event.v1", + eventType: "agentrun.run.event", + producedAt: "2026-07-09T10:00:09.000Z", + run: { runId: "run_unknown_lifecycle", sessionId: "ses_unknown_lifecycle" }, + command: { commandId: "cmd_unknown_lifecycle" }, + event: { seq: 1, type: "user_message", createdAt: "2026-07-09T10:00:05.000Z", payload: { text: "hello" } } + }, { source: "hwlab-test" }); + assert.equal(withoutFacts.event.startedAt, null); + assert.equal(withoutFacts.event.finishedAt, null); +}); + test("projects AgentRun assistant_message Kafka event into HWLAB trace event", () => { const projected = projectAgentRunKafkaEventToHwlabEvent({ schema: "agentrun.event.v1", diff --git a/internal/cloud/kafka-event-bridge.ts b/internal/cloud/kafka-event-bridge.ts index 12cfc7c7..8689df09 100644 --- a/internal/cloud/kafka-event-bridge.ts +++ b/internal/cloud/kafka-event-bridge.ts @@ -897,7 +897,7 @@ function assertCanonicalAgentRunEvent(value) { } if (!Number.isInteger(Number(value.sourceSeq)) || Number(value.sourceSeq) <= 0) throw contractError("workbench_kafka_schema_invalid", "Canonical AgentRun sourceSeq must be a positive integer."); if (value.valuesPrinted !== false) throw contractError("workbench_kafka_redaction_invalid", "Canonical AgentRun event must declare valuesPrinted=false."); - assertClosedObject(value.run, ["runId", "status", "terminalStatus", "failureKind", "tenantId", "projectId", "providerId", "backendProfile", "sessionId", "hwlabSessionId", "conversationId", "threadId", "valuesPrinted"], "AgentRun run"); + assertClosedObject(value.run, ["runId", "status", "terminalStatus", "failureKind", "tenantId", "projectId", "providerId", "backendProfile", "sessionId", "hwlabSessionId", "conversationId", "threadId", "startedAt", "finishedAt", "valuesPrinted"], "AgentRun run"); if (value.command !== null && value.command !== undefined) assertClosedObject(value.command, ["commandId", "runId", "seq", "type", "state", "payloadHash", "valuesPrinted"], "AgentRun command"); assertClosedObject(value.event, ["id", "runId", "seq", "type", "payload", "createdAt"], "AgentRun event"); if (value.event.id !== value.eventId || Number(value.event.seq) !== Number(value.sourceSeq)) throw contractError("workbench_kafka_identity_invalid", "Canonical AgentRun event identity does not match the envelope."); @@ -1341,6 +1341,8 @@ function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run, replyAuthority: explicitBoolean(payload.replyAuthority), final: explicitBoolean(payload.final), agentRunEventType: type, + startedAt: timestampValue(run.startedAt ?? payload.startedAt), + finishedAt: timestampValue(run.finishedAt ?? payload.finishedAt), createdAt: timestampValue(sourceEvent.createdAt ?? input.producedAt), valuesPrinted: false }; @@ -1436,7 +1438,7 @@ function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run, if (type === "terminal_status") { const terminalStatus = firstText(payload.terminalStatus, sourceEvent.terminalStatus, payload.status, sourceEvent.status) || "failed"; const normalized = terminalStatus === "cancelled" ? "canceled" : terminalStatus; - return { ...base, type: "result", eventType: "terminal", status: normalized, label: `agentrun:terminal:${terminalStatus}`, errorCode: firstText(payload.failureKind, payload.errorCode, sourceEvent.failureKind, sourceEvent.errorCode), failureKind: firstText(payload.failureKind, sourceEvent.failureKind), message: textPayload({ ...sourceEvent, ...payload }, `AgentRun terminal status ${terminalStatus}`), terminal: true }; + return { ...base, type: "result", eventType: "terminal", status: normalized, terminalStatus: normalized, label: `agentrun:terminal:${terminalStatus}`, errorCode: firstText(payload.failureKind, payload.errorCode, sourceEvent.failureKind, sourceEvent.errorCode), failureKind: firstText(payload.failureKind, sourceEvent.failureKind), message: textPayload({ ...sourceEvent, ...payload }, `AgentRun terminal status ${terminalStatus}`), terminal: true }; } if (type === "tool_call") { const toolName = firstText(payload.toolName, payload.name, payload.method) || "tool"; diff --git a/internal/cloud/server-agent-observer-http.ts b/internal/cloud/server-agent-observer-http.ts index 17abb816..065f31c5 100644 --- a/internal/cloud/server-agent-observer-http.ts +++ b/internal/cloud/server-agent-observer-http.ts @@ -23,6 +23,8 @@ export async function handleAgentObserverHttp(request, response, options = {}) { sendJson(response, 503, observerError(error?.code ?? "agent_observer_config_invalid", error instanceof Error ? error.message : String(error))); return; } + const requestUrl = new URL(request.url ?? "/v1/agent-observer/events", "http://agent-observer.local"); + const resumeAfterId = text(requestUrl.searchParams.get("afterId") ?? request.headers?.["last-event-id"]); try { await (bridge.liveReady ?? bridge.ready); @@ -75,6 +77,7 @@ export async function handleAgentObserverHttp(request, response, options = {}) { const handoff = createWorkbenchKafkaRefreshHandoff({ allowUnscoped: true, + resumeAfterId, liveBufferLimit: policy.liveBufferLimit, identityWindowLimit: policy.identityWindowLimit, subscribeLive: (listener) => bridge.subscribeLiveHwlabEvents(listener), @@ -87,6 +90,15 @@ export async function handleAgentObserverHttp(request, response, options = {}) { signal }), deliverEvent: (envelope) => enqueue("hwlab.event.v1", envelope), + deliverHandoff: (summary) => enqueue("agent-observer.handoff", { + type: "handoff", + status: "handoff", + deliverySemantics: "kafka-retention-then-live", + authority: "hwlab.event.v1", + replay: summary, + serverSentAt: new Date().toISOString(), + valuesPrinted: false + }), deliverConnected: (summary) => enqueue("agent-observer.connected", { type: "connected", status: "live", diff --git a/internal/cloud/workbench-kafka-refresh-handoff.test.ts b/internal/cloud/workbench-kafka-refresh-handoff.test.ts index 7a241f8a..4c97719f 100644 --- a/internal/cloud/workbench-kafka-refresh-handoff.test.ts +++ b/internal/cloud/workbench-kafka-refresh-handoff.test.ts @@ -41,7 +41,7 @@ test("retention handoff replays once, deduplicates overlap and late pre-barrier "buffered-live:4:hwlab:evt_0_4", "connected:1" ]); - assert.deepEqual(summary.counts, { matched: 3, replayed: 3, buffered: 3, bufferedDelivered: 2, liveDelivered: 0, deduplicated: 1 }); + assert.deepEqual(summary.counts, { matched: 3, replayed: 3, buffered: 3, bufferedDelivered: 2, liveDelivered: 0, deduplicated: 1, resumeSkipped: 0 }); publishLive(record(2).value, transport(2)); publishLive(record(5).value, transport(5)); @@ -51,6 +51,56 @@ test("retention handoff replays once, deduplicates overlap and late pre-barrier handoff.stop(); }); +test("retention resume delivers only events after a found identity and marks a missing identity as window advanced", async () => { + const foundOutput: string[] = []; + const found = createWorkbenchKafkaRefreshHandoff({ + sessionId: SESSION_ID, + resumeAfterId: "evt_0_1", + liveBufferLimit: 4, + subscribeLive() { return () => {}; }, + async queryRetention() { return queryResult([record(0), record(1), record(2)], [{ partition: 0, startOffset: "0", endOffset: "3" }]); }, + async deliverEvent(envelope: any) { foundOutput.push(envelope.sourceEventId); return true; }, + async deliverConnected() { return true; } + }); + const foundSummary = await found.start(); + assert.deepEqual(foundOutput, ["evt_0_2"]); + assert.deepEqual(foundSummary.resume, { requested: true, found: true, windowAdvanced: false }); + assert.equal(foundSummary.counts.resumeSkipped, 2); + + const missingOutput: string[] = []; + const missing = createWorkbenchKafkaRefreshHandoff({ + sessionId: SESSION_ID, + resumeAfterId: "evt_expired", + liveBufferLimit: 4, + subscribeLive() { return () => {}; }, + async queryRetention() { return queryResult([record(4), record(5)], [{ partition: 0, startOffset: "4", endOffset: "6" }]); }, + async deliverEvent(envelope: any) { missingOutput.push(envelope.sourceEventId); return true; }, + async deliverConnected() { return true; } + }); + const missingSummary = await missing.start(); + assert.deepEqual(missingOutput, ["evt_0_4", "evt_0_5"]); + assert.deepEqual(missingSummary.resume, { requested: true, found: false, windowAdvanced: true }); +}); + +test("retention emits handoff before buffered live flush and connected live", async () => { + let publishLive: any = null; + const phases: string[] = []; + const handoff = createWorkbenchKafkaRefreshHandoff({ + sessionId: SESSION_ID, + liveBufferLimit: 4, + subscribeLive(listener: any) { publishLive = listener; return () => {}; }, + async queryRetention() { + publishLive(record(2).value, transport(2)); + return queryResult([record(0), record(1)], [{ partition: 0, startOffset: "0", endOffset: "2" }]); + }, + async deliverEvent(_envelope: any, transportValue: any, delivery: string) { phases.push(`${delivery}:${transportValue.offset}`); return true; }, + async deliverHandoff(summary: any) { phases.push(summary.phase); return true; }, + async deliverConnected(summary: any) { phases.push(summary.phase); return true; } + }); + await handoff.start(); + assert.deepEqual(phases, ["replay:0", "replay:1", "handoff", "buffered-live:2", "live"]); +}); + test("retention handoff fails closed on a pre-barrier live gap", async () => { let publishLive: any = null; const handoff = createWorkbenchKafkaRefreshHandoff({ diff --git a/internal/cloud/workbench-kafka-refresh-handoff.ts b/internal/cloud/workbench-kafka-refresh-handoff.ts index 526b3e35..4d6dd614 100644 --- a/internal/cloud/workbench-kafka-refresh-handoff.ts +++ b/internal/cloud/workbench-kafka-refresh-handoff.ts @@ -7,6 +7,7 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { const allowUnscoped = options.allowUnscoped === true; const liveBufferLimit = positiveInteger(options.liveBufferLimit); const identityWindowLimit = positiveInteger(options.identityWindowLimit) ?? liveBufferLimit; + const resumeAfterId = textValue(options.resumeAfterId); 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") { @@ -37,8 +38,11 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { buffered: 0, bufferedDelivered: 0, liveDelivered: 0, - deduplicated: 0 + deduplicated: 0, + resumeSkipped: 0 }; + let resumeFound = resumeAfterId === null; + let retentionWindowAdvanced = false; function stop(reason = "connection-closed") { if (phase === "closed") return; @@ -68,6 +72,8 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { counts.replayed += 1; } + phase = "handoff"; + await deliverHandoffSerial(handoffSummary("handoff")); phase = "flushing"; while (bufferedLive.length > 0) { if (failure) throw failure; @@ -78,7 +84,7 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { counts.bufferedDelivered += 1; } - const summary = handoffSummary(); + const summary = handoffSummary("live"); const connectedDelivery = deliverConnectedSerial(summary); phase = "live"; await connectedDelivery; @@ -202,7 +208,16 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { if (priorProof && priorProof.stableKey !== stableKey) throw handoffError("workbench_kafka_refresh_transport_conflict", "One Kafka transport identity maps to conflicting stable events.", phase); const deliver = rememberIdentity(transportKey, stableIdentity, phase); replayProofByTransport.set(transportKey, stableIdentityByTransport.get(transportKey)); - normalized.push({ envelope, transport, deliver }); + normalized.push({ envelope, transport, stableIdentity, deliver }); + } + if (resumeAfterId) { + const resumeIndex = normalized.findIndex((record) => record.stableIdentity.eventId === resumeAfterId || record.stableIdentity.sourceEventId === resumeAfterId); + resumeFound = resumeIndex >= 0; + retentionWindowAdvanced = resumeIndex < 0; + if (resumeIndex >= 0) { + counts.resumeSkipped = resumeIndex + 1; + for (let index = 0; index <= resumeIndex; index += 1) normalized[index].deliver = false; + } } return normalized; } @@ -305,13 +320,30 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { return deliveryChain; } - function handoffSummary() { + function deliverHandoffSerial(summary) { + if (typeof options.deliverHandoff !== "function") return Promise.resolve(true); + const run = async () => { + if (failure || phase === "failed" || phase === "closed") throw failure ?? handoffError("workbench_kafka_refresh_stopped", "Kafka refresh delivery stopped before the handoff contract could be written.", phase); + const written = await options.deliverHandoff(summary); + if (written === false) throw handoffError("workbench_kafka_refresh_sse_write_failed", "Kafka refresh handoff contract could not be written to SSE.", phase); + return true; + }; + deliveryChain = deliveryChain.then(run); + return deliveryChain; + } + + function handoffSummary(summaryPhase = phase) { return { - phase: "live", + phase: summaryPhase, topic: scopedTopic, topicPartitions: scopedPartition === null ? [] : [scopedPartition], barrier: [...barrierByPartition.entries()].map(([partition, endOffset]) => ({ partition, endOffset })).sort((left, right) => left.partition - right.partition), counts: { ...counts }, + resume: { + requested: resumeAfterId !== null, + found: resumeFound, + windowAdvanced: retentionWindowAdvanced + }, identityWindowLimit, valuesRedacted: true }; @@ -340,7 +372,7 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) { start, stop, phase: () => phase, - summary: handoffSummary, + summary: () => handoffSummary(phase), valuesPrinted: false }; } diff --git a/web/hwlab-cloud-web/scripts/agent-observer-contract.test.ts b/web/hwlab-cloud-web/scripts/agent-observer-contract.test.ts new file mode 100644 index 00000000..5ef315aa --- /dev/null +++ b/web/hwlab-cloud-web/scripts/agent-observer-contract.test.ts @@ -0,0 +1,65 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { buildRelationWindow, mergeMonotonicBoolean, normalizeRunStatus, type AgentObserverRun } from "../src/stores/agent-observer.ts"; + +test("Agent observer keeps generic event status outside Run authority", () => { + assert.equal(normalizeRunStatus(null, null), "unknown"); + assert.equal(normalizeRunStatus(null, null, "running"), "running"); + assert.equal(normalizeRunStatus("queued", null, "unknown"), "queued"); + assert.equal(normalizeRunStatus(null, "cancelled", "running"), "canceled"); +}); + +test("Agent observer final authority is monotonic once true", () => { + assert.equal(mergeMonotonicBoolean(null, false), false); + assert.equal(mergeMonotonicBoolean(false, null), false); + assert.equal(mergeMonotonicBoolean(false, true), true); + assert.equal(mergeMonotonicBoolean(true, false), true); + assert.equal(mergeMonotonicBoolean(true, null), true); +}); + +test("Agent observer relation tree consumes the YAML node limit", () => { + const runs = [observerRun("run_1"), observerRun("run_2"), observerRun("run_3")]; + const window = buildRelationWindow(runs, null, new Set(), new Set(), 3); + assert.equal(window.limited, true); + assert.equal(window.nodeCount, 3); + assert.equal(window.omittedRunCount, 1); + assert.equal(window.rows.length, 3); +}); + +function observerRun(id: string): AgentObserverRun { + return { + id, + taskId: null, + taskStatus: null, + commandId: null, + attemptId: null, + attemptStatus: null, + runnerId: null, + runnerCapacityUsed: null, + runnerCapacityTotal: null, + nodeId: null, + lane: null, + sessionId: null, + traceId: null, + projectId: null, + workspace: null, + backend: null, + providerId: null, + runStatus: "unknown", + commandState: null, + commandType: null, + phase: "event", + result: null, + replyAuthority: null, + final: null, + errorCode: null, + lastMessage: null, + startedAt: null, + finishedAt: null, + updatedAt: null, + lastSourceSeq: null, + eventIds: [], + changedAtMs: 0 + }; +} diff --git a/web/hwlab-cloud-web/src/api/agentObserver.ts b/web/hwlab-cloud-web/src/api/agentObserver.ts index cb42567a..a8c17c2f 100644 --- a/web/hwlab-cloud-web/src/api/agentObserver.ts +++ b/web/hwlab-cloud-web/src/api/agentObserver.ts @@ -1,5 +1,7 @@ export interface AgentObserverStreamCallbacks { - onEnvelope: (payload: unknown) => void; + afterId?: string | null; + onEnvelope: (payload: unknown, eventId: string | null) => void; + onHandoff: (payload: unknown) => void; onConnected: (payload: unknown) => void; onHeartbeat: (payload: unknown) => void; onStreamError: (payload: unknown) => void; @@ -7,8 +9,11 @@ export interface AgentObserverStreamCallbacks { } 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))); + const url = new URL("/v1/agent-observer/events", window.location.origin); + if (callbacks.afterId) url.searchParams.set("afterId", callbacks.afterId); + const source = new EventSource(`${url.pathname}${url.search}`, { withCredentials: true }); + source.addEventListener("hwlab.event.v1", (event) => callbacks.onEnvelope(parseEvent(event), event instanceof MessageEvent ? event.lastEventId || null : null)); + source.addEventListener("agent-observer.handoff", (event) => callbacks.onHandoff(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))); diff --git a/web/hwlab-cloud-web/src/stores/agent-observer.ts b/web/hwlab-cloud-web/src/stores/agent-observer.ts index 6c448edb..b7f9f53e 100644 --- a/web/hwlab-cloud-web/src/stores/agent-observer.ts +++ b/web/hwlab-cloud-web/src/stores/agent-observer.ts @@ -6,7 +6,7 @@ import { asRecord, nonEmptyString } from "@/utils"; export type AgentObserverView = "table" | "cards" | "tree"; export type AgentObserverDensity = "compact" | "standard"; -export type AgentObserverPhase = "idle" | "replay" | "live" | "reconnecting" | "stale" | "error"; +export type AgentObserverPhase = "idle" | "replay" | "handoff" | "live" | "reconnecting" | "stale" | "error"; export interface AgentObserverEvent { id: string; @@ -110,6 +110,8 @@ export const useAgentObserverStore = defineStore("agentObserver", () => { const phase = ref("idle"); const lastEventAt = ref(null); const connectedAt = ref(null); + const lastConfirmedEventId = ref(null); + const retentionWindowAdvanced = ref(false); const error = ref(null); const nowMs = ref(Date.now()); const config = ref(null); @@ -117,7 +119,7 @@ export const useAgentObserverStore = defineStore("agentObserver", () => { const collapsedRelationIds = ref([]); const seenIdentity = new Map(); const identityWindow: Array<{ token: object; keys: string[] }> = []; - const queue: unknown[] = []; + const queue: Array<{ payload: unknown; eventId: string | null }> = []; let source: EventSource | null = null; let reconnectTimer: number | null = null; let staleTimer: number | null = null; @@ -172,6 +174,7 @@ export const useAgentObserverStore = defineStore("agentObserver", () => { waiting: values.filter((run) => statusGroup(run.runStatus) === "waiting").length, failed: values.filter((run) => statusGroup(run.runStatus) === "failed").length, terminal: values.filter((run) => statusGroup(run.runStatus) === "terminal").length, + unknown: values.filter((run) => statusGroup(run.runStatus) === "unknown").length, stale: values.filter((run) => run.updatedAt && nowMs.value - Date.parse(run.updatedAt) > (config.value?.staleAfterMs ?? Number.MAX_SAFE_INTEGER)).length, runners: new Set(values.map((run) => run.runnerId).filter(Boolean)).size, runnerCapacityUsed: [...capacityByRunner.values()].reduce((sum, capacity) => sum + capacity.used, 0), @@ -180,12 +183,14 @@ export const useAgentObserverStore = defineStore("agentObserver", () => { }; }); const dataAgeMs = computed(() => lastEventAt.value ? Math.max(0, nowMs.value - Date.parse(lastEventAt.value)) : null); - const relationRows = computed(() => buildRelationRows( + const relationWindow = computed(() => buildRelationWindow( visibleRuns.value, selectedRunId.value, new Set(expandedRelationIds.value), - new Set(collapsedRelationIds.value) + new Set(collapsedRelationIds.value), + config.value?.treeNodeLimit ?? 1 )); + const relationRows = computed(() => relationWindow.value.rows); function connect(): void { disconnect(false); @@ -196,11 +201,19 @@ export const useAgentObserverStore = defineStore("agentObserver", () => { return; } manuallyClosed = false; - phase.value = entities.value && orderedIds.value.length > 0 ? "reconnecting" : "replay"; + phase.value = "replay"; error.value = null; source = openAgentObserverStream({ + afterId: lastConfirmedEventId.value, onEnvelope: enqueueEnvelope, - onConnected: () => { + onHandoff: () => { + phase.value = "handoff"; + error.value = null; + }, + onConnected: (payload) => { + const replay = asRecord(asRecord(payload)?.replay); + const resume = asRecord(replay?.resume); + retentionWindowAdvanced.value = resume?.windowAdvanced === true; connectedAt.value = new Date().toISOString(); phase.value = "live"; error.value = null; @@ -249,7 +262,7 @@ export const useAgentObserverStore = defineStore("agentObserver", () => { }, config.value.reconnectDelayMs); } - function enqueueEnvelope(payload: unknown): void { + function enqueueEnvelope(payload: unknown, eventId: string | null): void { if (!config.value) return; if (queue.length >= config.value.liveBufferLimit) { phase.value = "error"; @@ -257,7 +270,7 @@ export const useAgentObserverStore = defineStore("agentObserver", () => { source?.close(); return; } - queue.push(payload); + queue.push({ payload, eventId }); if (flushScheduled) return; flushScheduled = true; queueMicrotask(flushQueue); @@ -267,15 +280,17 @@ export const useAgentObserverStore = defineStore("agentObserver", () => { flushScheduled = false; if (!config.value) return; const batch = queue.splice(0, config.value.batchMaxEvents); - for (const payload of batch) reduceEnvelope(payload); + for (const item of batch) { + if (reduceEnvelope(item.payload)) lastConfirmedEventId.value = item.eventId ?? stableEnvelopeIdentity(item.payload) ?? lastConfirmedEventId.value; + } if (queue.length > 0) { flushScheduled = true; queueMicrotask(flushQueue); } } - function reduceEnvelope(payload: unknown): void { - if (!config.value) return; + function reduceEnvelope(payload: unknown): boolean { + if (!config.value) return false; const envelope = asRecord(payload); const body = asRecord(envelope?.event); const context = asRecord(envelope?.context); @@ -283,7 +298,7 @@ export const useAgentObserverStore = defineStore("agentObserver", () => { const outerEventId = nonEmptyString(envelope?.eventId); const sourceEventId = nonEmptyString(envelope?.sourceEventId); const eventId = outerEventId ?? sourceEventId; - if (!runId || !eventId || rememberIdentity(outerEventId, sourceEventId)) return; + if (!runId || !eventId || rememberIdentity(outerEventId, sourceEventId)) return false; const createdAt = nonEmptyString(body?.createdAt) ?? nonEmptyString(envelope?.producedAt); const sourceSeq = integer(body?.sourceSeq) ?? integer(context?.sourceSeq); const event: AgentObserverEvent = { @@ -312,9 +327,9 @@ export const useAgentObserverStore = defineStore("agentObserver", () => { || (sourceSeq !== null && (previous.lastSourceSeq === null || sourceSeq > previous.lastSourceSeq)); lastEventAt.value = latestTimestamp(lastEventAt.value, createdAt) ?? new Date().toISOString(); nowMs.value = Date.now(); - if (previous && !sequenceIsCurrent) return; + if (previous && !sequenceIsCurrent) return true; const terminalStatus = nonEmptyString(body?.terminalStatus); - const runStatus = normalizeRunStatus(nonEmptyString(body?.runStatus), terminalStatus, event.status, event.terminal, previous?.runStatus); + const runStatus = normalizeRunStatus(nonEmptyString(body?.runStatus), terminalStatus, previous?.runStatus); const next: AgentObserverRun = { id: runId, taskId: nonEmptyString(body?.taskId) ?? previous?.taskId ?? null, @@ -338,12 +353,12 @@ export const useAgentObserverStore = defineStore("agentObserver", () => { commandType: nonEmptyString(body?.commandType) ?? previous?.commandType ?? null, phase: event.label.replace(/^agentrun:/u, ""), result: nonEmptyString(body?.result) ?? previous?.result ?? null, - replyAuthority: explicitBoolean(body?.replyAuthority) ?? previous?.replyAuthority ?? null, - final: explicitBoolean(body?.final) ?? previous?.final ?? null, + replyAuthority: mergeMonotonicBoolean(previous?.replyAuthority, explicitBoolean(body?.replyAuthority)), + final: mergeMonotonicBoolean(previous?.final, explicitBoolean(body?.final)), errorCode: nonEmptyString(body?.errorCode) ?? nonEmptyString(body?.failureKind) ?? previous?.errorCode ?? null, lastMessage: event.message || (previous?.lastMessage ?? null), - startedAt: previous?.startedAt ?? createdAt, - finishedAt: event.terminal ? (createdAt ?? previous?.finishedAt ?? null) : (previous?.finishedAt ?? null), + startedAt: nonEmptyString(body?.startedAt) ?? previous?.startedAt ?? null, + finishedAt: nonEmptyString(body?.finishedAt) ?? (event.terminal ? (createdAt ?? previous?.finishedAt ?? null) : (previous?.finishedAt ?? null)), updatedAt: createdAt ?? previous?.updatedAt ?? null, lastSourceSeq: sourceSeq ?? previous?.lastSourceSeq ?? null, eventIds: [...(previous?.eventIds ?? []), eventId].slice(-config.value.inspectorEventTailLimit), @@ -352,6 +367,7 @@ export const useAgentObserverStore = defineStore("agentObserver", () => { entities.value = { ...entities.value, [runId]: next }; orderedIds.value = [runId, ...orderedIds.value.filter((id) => id !== runId)]; trimEntities(); + return true; } function rememberIdentity(eventId: string | null, sourceEventId: string | null): boolean { @@ -396,17 +412,20 @@ export const useAgentObserverStore = defineStore("agentObserver", () => { return { entities, orderedIds, events, runs, visibleRuns, selectedRun, selectedEvents, selectedLogs, selectedInEventWindow, selectedRunId, view, density, search, statusFilter, projectFilter, agentFilter, timeWindow, sort, phase, lastEventAt, - connectedAt, error, nowMs, config, projects, scopeOptions, agents, stats, dataAgeMs, relationRows, + connectedAt, lastConfirmedEventId, retentionWindowAdvanced, error, nowMs, config, projects, scopeOptions, agents, stats, dataAgeMs, relationRows, + relationWindowLimited: computed(() => relationWindow.value.limited), relationNodeCount: computed(() => relationWindow.value.nodeCount), + relationOmittedRunCount: computed(() => relationWindow.value.omittedRunCount), connect, disconnect, reconnect, selectRun, toggleRelation }; }); -function buildRelationRows( +export function buildRelationWindow( runs: AgentObserverRun[], selectedRunId: string | null, expandedIds: Set, - collapsedIds: Set -): AgentObserverRelationRow[] { + collapsedIds: Set, + nodeLimit: number +): { rows: AgentObserverRelationRow[]; limited: boolean; nodeCount: number; omittedRunCount: number } { const nodes = new Map(); const rootIds: string[] = []; const selectedPath = new Set(); @@ -435,8 +454,22 @@ function buildRelationRows( status: null }); - for (const run of runs) { + let omittedRunCount = 0; + for (const [runIndex, run] of runs.entries()) { const hasCompleteContainment = Boolean((run.projectId || run.workspace) && run.sessionId && run.taskId); + const scopeId = hasCompleteContainment ? relationId("scope", `${run.projectId ?? ""}\u0000${run.workspace ?? ""}`) : unassociatedId; + const sessionId = hasCompleteContainment && run.sessionId ? relationId("session", `${scopeId}\u0000${run.sessionId}`) : null; + const taskId = sessionId && run.taskId ? relationId("task", `${sessionId}\u0000${run.taskId}`) : null; + const runParentId = taskId ?? scopeId; + const runNodeId = relationId("run", `${runParentId}\u0000${run.id}`); + const attemptId = run.attemptId ? relationId("attempt", `${runNodeId}\u0000${run.attemptId}`) : null; + const commandParentId = attemptId ?? runNodeId; + const commandId = run.commandId ? relationId("command", `${commandParentId}\u0000${run.commandId}`) : null; + const missingNodeCount = [scopeId, sessionId, taskId, runNodeId, attemptId, commandId].filter((id): id is string => typeof id === "string" && !nodes.has(id)).length; + if (nodes.size + missingNodeCount > nodeLimit) { + omittedRunCount = runs.length - runIndex; + break; + } let parent = hasCompleteContainment ? ensureNode({ id: relationId("scope", `${run.projectId ?? ""}\u0000${run.workspace ?? ""}`), @@ -537,7 +570,7 @@ function buildRelationRows( if (expanded) node.children.forEach((childId, index) => append(childId, depth + 1, index + 1, node.children.length)); }; rootIds.forEach((id, index) => append(id, 0, index + 1, rootIds.length)); - return rows; + return { rows, limited: omittedRunCount > 0, nodeCount: nodes.size, omittedRunCount }; } function relationId(kind: AgentObserverRelationKind, value: string): string { @@ -554,11 +587,15 @@ function latestTimestamp(previous: string | null, next: string | null): string | return nextMs >= previousMs ? next : previous; } -function normalizeRunStatus(runStatus: string | null, terminalStatus: string | null, eventStatus: string, terminal: boolean, previous = "unknown"): string { +export function normalizeRunStatus(runStatus: string | null, terminalStatus: string | null, previous = "unknown"): string { if (terminalStatus) return terminalStatus === "cancelled" ? "canceled" : terminalStatus; - if (terminal) return eventStatus; if (["completed", "failed", "canceled", "cancelled"].includes(previous)) return previous; - return runStatus ?? eventStatus ?? previous; + return runStatus ?? previous ?? "unknown"; +} + +export function mergeMonotonicBoolean(previous: boolean | null | undefined, next: boolean | null): boolean | null { + if (previous === true) return true; + return next ?? previous ?? null; } export function agentObserverStatusToken(status: string | null): AgentObserverStatusToken { @@ -576,7 +613,13 @@ function statusGroup(status: string): string { if (["failed", "error", "rejected"].includes(value)) return "failed"; if (["completed", "succeeded", "canceled", "cancelled"].includes(value)) return "terminal"; if (["queued", "pending", "waiting", "admitted"].includes(value)) return "waiting"; - return "running"; + if (["running", "active", "started", "in_progress"].includes(value)) return "running"; + return "unknown"; +} + +function stableEnvelopeIdentity(payload: unknown): string | null { + const envelope = asRecord(payload); + return nonEmptyString(envelope?.eventId) ?? nonEmptyString(envelope?.sourceEventId); } function compareRuns(left: AgentObserverRun, right: AgentObserverRun, sort: string): number { diff --git a/web/hwlab-cloud-web/src/views/agents/AgentRunsView.vue b/web/hwlab-cloud-web/src/views/agents/AgentRunsView.vue index eba2a830..69c2e220 100644 --- a/web/hwlab-cloud-web/src/views/agents/AgentRunsView.vue +++ b/web/hwlab-cloud-web/src/views/agents/AgentRunsView.vue @@ -26,7 +26,7 @@ const treeList = ref(null); const activeRelationId = ref(null); const treeRows = computed(() => observer.relationRows); const ageLabel = computed(() => formatAge(observer.dataAgeMs)); -const streamLabel = computed(() => ({ idle: "未连接", replay: "回放", live: "Live", reconnecting: "重连中", stale: "陈旧", error: "传输错误" }[observer.phase])); +const streamLabel = computed(() => ({ idle: "未连接", replay: "回放", handoff: "交接", live: "Live", reconnecting: "重连中", stale: "陈旧", error: "传输错误" }[observer.phase])); const selectedMissing = computed(() => Boolean(observer.selectedRunId && !observer.selectedRun)); let inspectorMedia: MediaQueryList | null = null; @@ -163,13 +163,14 @@ function compactQuery(value: Record): Record { r