修复：收紧智能体观察实时语义与有界恢复
This commit is contained in:
root
@@ -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",
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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
|
||||
};
|
||||
}
|
||||
@@ -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)));
|
||||
|
||||
@@ -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<AgentObserverPhase>("idle");
|
||||
const lastEventAt = ref<string | null>(null);
|
||||
const connectedAt = ref<string | null>(null);
|
||||
const lastConfirmedEventId = ref<string | null>(null);
|
||||
const retentionWindowAdvanced = ref(false);
|
||||
const error = ref<string | null>(null);
|
||||
const nowMs = ref(Date.now());
|
||||
const config = ref<AgentObserverRuntimeConfig | null>(null);
|
||||
@@ -117,7 +119,7 @@ export const useAgentObserverStore = defineStore("agentObserver", () => {
|
||||
const collapsedRelationIds = ref<string[]>([]);
|
||||
const seenIdentity = new Map<string, object>();
|
||||
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<string>,
|
||||
collapsedIds: Set<string>
|
||||
): AgentObserverRelationRow[] {
|
||||
collapsedIds: Set<string>,
|
||||
nodeLimit: number
|
||||
): { rows: AgentObserverRelationRow[]; limited: boolean; nodeCount: number; omittedRunCount: number } {
|
||||
const nodes = new Map<string, AgentObserverRelationNode>();
|
||||
const rootIds: string[] = [];
|
||||
const selectedPath = new Set<string>();
|
||||
@@ -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 {
|
||||
|
||||
@@ -26,7 +26,7 @@ const treeList = ref<VirtualListHandle | null>(null);
|
||||
const activeRelationId = ref<string | null>(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<string, string>): Record<string, string> { r
|
||||
<PageCommandBar eyebrow="智能体管理" title="AgentRun 运行观察" description="纯 Kafka retention replay → live；三种视图与统一 Inspector 共享同一 orderedIds、筛选、排序和选择。" sticky>
|
||||
<template #actions>
|
||||
<span class="agent-observer-stream" :data-phase="observer.phase" aria-live="polite"><Activity :size="15" aria-hidden="true" /><strong>{{ streamLabel }}</strong><span>{{ ageLabel }}</span></span>
|
||||
<span v-if="observer.retentionWindowAdvanced" class="agent-observer-target">当前 retention 窗口已前移</span>
|
||||
<span v-if="observer.config" class="agent-observer-target mono">{{ observer.config.nodeId }}/{{ observer.config.lane }}</span>
|
||||
<button class="btn btn-secondary" type="button" @click="observer.reconnect"><RefreshCw :size="15" aria-hidden="true" />重连</button>
|
||||
</template>
|
||||
<template #filters>
|
||||
<label class="agent-observer-search"><Search :size="16" aria-hidden="true" /><span class="sr-only">搜索 run、task 或 session</span><input v-model="observer.search" type="search" placeholder="搜索 run / task / session / error" /><button v-if="observer.search" type="button" title="清除搜索" aria-label="清除搜索" @click="observer.search = ''"><X :size="14" /></button></label>
|
||||
<div class="agent-observer-secondary-filters">
|
||||
<select v-model="observer.statusFilter" aria-label="运行状态"><option value="all">全部状态</option><option value="running">运行中</option><option value="waiting">等待</option><option value="failed">失败</option><option value="terminal">终态</option></select>
|
||||
<select v-model="observer.statusFilter" aria-label="运行状态"><option value="all">全部状态</option><option value="running">运行中</option><option value="waiting">等待</option><option value="failed">失败</option><option value="terminal">终态</option><option value="unknown">未知</option></select>
|
||||
<select v-model="observer.projectFilter" aria-label="项目或 Workspace"><option value="all">全部 Project / Workspace</option><option v-for="item in observer.scopeOptions" :key="item.value" :value="item.value">{{ item.label }}</option></select>
|
||||
<select v-model="observer.agentFilter" aria-label="Agent 或 Profile"><option value="all">全部 Agent / Profile</option><option v-for="item in observer.agents" :key="item" :value="item">{{ item }}</option></select>
|
||||
<select v-model="observer.timeWindow" aria-label="时间窗口"><option value="all">全部窗口</option><option value="15m">15 分钟</option><option value="1h">1 小时</option><option value="6h">6 小时</option></select>
|
||||
@@ -182,11 +183,11 @@ function compactQuery(value: Record<string, string>): Record<string, string> { r
|
||||
</PageCommandBar>
|
||||
|
||||
<BaseDrawer :open="filtersOpen" title="AgentRun 筛选" description="与桌面命令栏共享同一查询状态。" surface-class="agent-observer-filter-drawer" @close="filtersOpen = false">
|
||||
<div class="agent-observer-drawer-fields"><label>运行状态<select v-model="observer.statusFilter"><option value="all">全部状态</option><option value="running">运行中</option><option value="waiting">等待</option><option value="failed">失败</option><option value="terminal">终态</option></select></label><label>项目 / Workspace<select v-model="observer.projectFilter"><option value="all">全部</option><option v-for="item in observer.scopeOptions" :key="item.value" :value="item.value">{{ item.label }}</option></select></label><label>Agent / Profile<select v-model="observer.agentFilter"><option value="all">全部</option><option v-for="item in observer.agents" :key="item" :value="item">{{ item }}</option></select></label><label>时间窗口<select v-model="observer.timeWindow"><option value="all">全部</option><option value="15m">15 分钟</option><option value="1h">1 小时</option><option value="6h">6 小时</option></select></label></div>
|
||||
<div class="agent-observer-drawer-fields"><label>运行状态<select v-model="observer.statusFilter"><option value="all">全部状态</option><option value="running">运行中</option><option value="waiting">等待</option><option value="failed">失败</option><option value="terminal">终态</option><option value="unknown">未知</option></select></label><label>项目 / Workspace<select v-model="observer.projectFilter"><option value="all">全部</option><option v-for="item in observer.scopeOptions" :key="item.value" :value="item.value">{{ item.label }}</option></select></label><label>Agent / Profile<select v-model="observer.agentFilter"><option value="all">全部</option><option v-for="item in observer.agents" :key="item" :value="item">{{ item }}</option></select></label><label>时间窗口<select v-model="observer.timeWindow"><option value="all">全部</option><option value="15m">15 分钟</option><option value="1h">1 小时</option><option value="6h">6 小时</option></select></label></div>
|
||||
</BaseDrawer>
|
||||
|
||||
<div class="agent-observer-status-strip" aria-label="运行状态摘要">
|
||||
<span><strong>{{ observer.stats.running }}</strong>运行</span><span><strong>{{ observer.stats.waiting }}</strong>等待</span><span data-tone="danger"><strong>{{ observer.stats.failed }}</strong>失败</span><span><strong>{{ observer.stats.stale }}</strong>陈旧</span><span><strong>{{ observer.stats.terminal }}</strong>终态</span><span><strong>{{ observer.stats.runners }}</strong>Runner</span><span><strong>{{ observer.stats.runnerCapacityKnown ? `${observer.stats.runnerCapacityUsed}/${observer.stats.runnerCapacityTotal}` : "未知" }}</strong>容量</span><span><strong>{{ observer.stats.total }}</strong>结果</span>
|
||||
<span><strong>{{ observer.stats.running }}</strong>运行</span><span><strong>{{ observer.stats.waiting }}</strong>等待</span><span data-tone="danger"><strong>{{ observer.stats.failed }}</strong>失败</span><span><strong>{{ observer.stats.unknown }}</strong>未知</span><span><strong>{{ observer.stats.stale }}</strong>陈旧</span><span><strong>{{ observer.stats.terminal }}</strong>终态</span><span><strong>{{ observer.stats.runners }}</strong>Runner</span><span><strong>{{ observer.stats.runnerCapacityKnown ? `${observer.stats.runnerCapacityUsed}/${observer.stats.runnerCapacityTotal}` : "未知" }}</strong>容量</span><span><strong>{{ observer.stats.total }}</strong>结果</span>
|
||||
</div>
|
||||
<div v-if="observer.error" class="agent-observer-inline-error" role="status">{{ observer.error }}</div>
|
||||
<div v-if="selectedMissing" class="agent-observer-window-warning" role="status">深链所选实体不在当前 reducer 实体窗口；选择保持不变，等待正式 Kafka 事件重新进入窗口。</div>
|
||||
@@ -210,6 +211,7 @@ function compactQuery(value: Record<string, string>): Record<string, string> { r
|
||||
</section>
|
||||
|
||||
<section v-else class="agent-observer-tree" role="tree" aria-label="AgentRun 关系树">
|
||||
<p v-if="observer.relationWindowLimited" class="agent-observer-window-note">当前事件窗口：关系树按 YAML 上限显示 {{ observer.relationNodeCount }} 个节点，另有 {{ observer.relationOmittedRunCount }} 个运行未展开。</p>
|
||||
<VirtualList ref="treeList" :items="treeRows" :item-key="relationKey" :item-height="44" height="100%" :overscan="8" :window-limit="observer.config?.virtualListWindowLimit" root-role="presentation" item-role="presentation" aria-label="AgentRun 虚拟关系树">
|
||||
<template #default="{ item: node, index }"><div class="agent-observer-tree-row" role="treeitem" :data-relation-id="node.id" :data-tree-index="index" :aria-level="node.depth + 1" :aria-posinset="node.position" :aria-setsize="node.setSize" :aria-expanded="node.childCount > 0 ? node.expanded : undefined" :aria-selected="observer.selectedRunId === node.runId" :tabindex="activeRelationId === node.id || (!activeRelationId && index === 0) ? 0 : -1" :style="{ '--tree-depth': node.depth }" @focus="activeRelationId = node.id" @click="activateRelation(node)" @keydown="onRelationKeydown($event, node, index)"><button v-if="node.childCount > 0" class="agent-observer-tree-toggle" type="button" :aria-label="node.expanded ? '折叠' : '展开'" tabindex="-1" @click.stop="observer.toggleRelation(node.id, !node.expanded)"><ChevronRight :size="15" :data-expanded="node.expanded" /></button><span v-else class="agent-observer-tree-spacer" aria-hidden="true" /><StatusBadge v-if="node.status" :status="status(node.status).tone" :label="status(node.status).label" /><span v-else class="agent-observer-relation-kind">{{ node.kind }}</span><strong class="mono" :title="node.label">{{ shortId(node.label) }}</strong><small :title="node.detail || ''">{{ node.detail || `${node.childCount} 个子节点` }}</small></div></template>
|
||||
</VirtualList>
|
||||
|
||||
Reference in New Issue
Block a user