diff --git a/src/gateway/session-observer-notes.ts b/src/agents/session-activity-notes.ts similarity index 55% rename from src/gateway/session-observer-notes.ts rename to src/agents/session-activity-notes.ts index f1ccb77ddcf2..be18b6fb1890 100644 --- a/src/gateway/session-observer-notes.ts +++ b/src/agents/session-activity-notes.ts @@ -1,62 +1,54 @@ import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; -import { - INTERNAL_RUNTIME_CONTEXT_BEGIN, - INTERNAL_RUNTIME_CONTEXT_END, - stripInternalRuntimeContext, -} from "../agents/internal-runtime-context.js"; import { HEARTBEAT_TRANSCRIPT_PROMPT } from "../auto-reply/heartbeat.js"; import { HEARTBEAT_TOKEN } from "../auto-reply/tokens.js"; import { normalizeAgentPlanSteps } from "../channels/streaming.js"; import type { AgentEventPayload } from "../infra/agent-events.js"; import { redactToolPayloadText } from "../logging/redact.js"; +import { buildAgentRunTerminalOutcome } from "./agent-run-terminal-outcome.js"; import { - readFiniteNumber, - readString, - terminalHealthFor, - type SessionObserverState, -} from "./session-observer-model.js"; + INTERNAL_RUNTIME_CONTEXT_BEGIN, + INTERNAL_RUNTIME_CONTEXT_END, + stripInternalRuntimeContext, +} from "./internal-runtime-context.js"; + +export type SessionActivityNoteState = { + noteSequence: number; + notes: Array<{ sequence: number; text: string; bytes: number }>; + noteBytes: number; + itemStatuses: Map; + assistantBuffer: string; + lastAssistantNote?: string; + planProgress?: { completed: number; total: number }; +}; const MAX_NOTES = 40; const MAX_NOTE_BYTES = 8 * 1024; +const DEFAULT_NOTE_MAX_CHARS = 360; const ASSISTANT_NOTE_MAX_CHARS = 240; -const SESSION_OBSERVER_ASSISTANT_BUFFER_MAX_CHARS = 4096; +const ASSISTANT_BUFFER_MAX_CHARS = 4096; +const MAX_ITEM_STATUSES = 160; -/** - * Assemble streamed assistant prose: strip complete runtime-context blocks, - * then truncate without ever discarding an unmatched context BEGIN marker so - * the eventual END still closes and strips the whole block. Accepted tradeoff: - * a truncation boundary landing inside a split marker while the model echoes - * >4 KB of context is treated as ordinary prose (flush stays redacted). - */ -function assembleSessionObserverAssistantBuffer(value: string): string { - // Detect a still-open block on the RAW text: the stripper drops an - // unterminated marker together with its tail, which would leave the block - // body arriving in later deltas indistinguishable from ordinary prose. +export function createSessionActivityNoteState(): SessionActivityNoteState { + return { noteSequence: 0, notes: [], noteBytes: 0, itemStatuses: new Map(), assistantBuffer: "" }; +} + +// Preserve an unmatched BEGIN while truncating so a later END can still strip the private block. +function assembleAssistantBuffer(value: string, maxChars: number): string { + // Detect on raw text: stripping an open block would make later body deltas + // indistinguishable from ordinary prose. const openIndex = value.lastIndexOf(INTERNAL_RUNTIME_CONTEXT_BEGIN); const isOpen = openIndex !== -1 && !value.includes(INTERNAL_RUNTIME_CONTEXT_END, openIndex); if (!isOpen) { - return keepUtf16SafeTail( - stripInternalRuntimeContext(value), - SESSION_OBSERVER_ASSISTANT_BUFFER_MAX_CHARS, - ); + return keepUtf16SafeTail(stripInternalRuntimeContext(value), maxChars); } - const head = keepUtf16SafeTail( - stripInternalRuntimeContext(value.slice(0, openIndex)), - SESSION_OBSERVER_ASSISTANT_BUFFER_MAX_CHARS, - ); + const head = keepUtf16SafeTail(stripInternalRuntimeContext(value.slice(0, openIndex)), maxChars); const body = keepUtf16SafeTail( value.slice(openIndex + INTERNAL_RUNTIME_CONTEXT_BEGIN.length), - SESSION_OBSERVER_ASSISTANT_BUFFER_MAX_CHARS, + maxChars, ); return `${head}${INTERNAL_RUNTIME_CONTEXT_BEGIN}${body}`; } -/** True while the buffer holds a still-streaming runtime-context block. */ -function assistantBufferHasOpenContext(value: string): boolean { - return value.includes(INTERNAL_RUNTIME_CONTEXT_BEGIN); -} - -/** Keep the newest chars without starting on the low half of a surrogate pair. */ function keepUtf16SafeTail(value: string, maxChars: number): string { if (value.length <= maxChars) { return value; @@ -69,14 +61,14 @@ function keepUtf16SafeTail(value: string, maxChars: number): string { return value.slice(start); } -function sanitizeSessionObserverActivityText(value: string, maxChars: number): string { +function sanitizeActivityText(value: string, maxChars: number): string { const normalized = redactToolPayloadText(stripInternalRuntimeContext(value)) .replace(/\s+/gu, " ") .trim(); return truncateUtf16Safe(normalized, maxChars); } -function summarizeSessionObserverToolArgs(args: unknown): string { +function summarizeToolArgs(args: unknown): string { if (!args || typeof args !== "object") { return ""; } @@ -106,9 +98,9 @@ function summarizeSessionObserverToolArgs(args: unknown): string { } try { if (Object.keys(summary).length > 0) { - return sanitizeSessionObserverActivityText(JSON.stringify(summary), 220); + return sanitizeActivityText(JSON.stringify(summary), 220); } - return sanitizeSessionObserverActivityText( + return sanitizeActivityText( `args: ${Object.keys(record).toSorted().slice(0, 8).join(", ")}`, 220, ); @@ -117,8 +109,8 @@ function summarizeSessionObserverToolArgs(args: unknown): string { } } -function addSessionObserverNote(state: SessionObserverState, raw: string): void { - const text = sanitizeSessionObserverActivityText(raw, 360); +function addActivityNote(state: SessionActivityNoteState, raw: string, maxChars: number): void { + const text = sanitizeActivityText(raw, maxChars); if (!text) { return; } @@ -136,17 +128,36 @@ function addSessionObserverNote(state: SessionObserverState, raw: string): void } } -export function flushSessionObserverAssistantNote(state: SessionObserverState): void { - // Assistant prose is redacted only as assembled text: per-fragment sanitizing - // cannot match secrets split across stream chunks, and raw fragments must not - // count toward the digest note threshold. - if (!state.assistantBuffer || assistantBufferHasOpenContext(state.assistantBuffer)) { +function rememberItemStatus( + state: SessionActivityNoteState, + itemId: string, + status: string, + limit: number, +): boolean { + if (state.itemStatuses.get(itemId) === status) { + return false; + } + state.itemStatuses.delete(itemId); + state.itemStatuses.set(itemId, status); + while (state.itemStatuses.size > limit) { + const oldest = state.itemStatuses.keys().next().value; + if (oldest === undefined) { + break; + } + state.itemStatuses.delete(oldest); + } + return true; +} + +export function flushSessionActivityAssistantNote( + state: SessionActivityNoteState, + noteMaxChars: number = DEFAULT_NOTE_MAX_CHARS, +): void { + // Redact assembled prose so split secrets match and raw fragments do not count as notes. + if (!state.assistantBuffer || state.assistantBuffer.includes(INTERNAL_RUNTIME_CONTEXT_BEGIN)) { return; } - const sanitized = sanitizeSessionObserverActivityText( - state.assistantBuffer, - SESSION_OBSERVER_ASSISTANT_BUFFER_MAX_CHARS, - ); + const sanitized = sanitizeActivityText(state.assistantBuffer, ASSISTANT_BUFFER_MAX_CHARS); const visible = keepUtf16SafeTail(sanitized, ASSISTANT_NOTE_MAX_CHARS).trim(); if (!visible || visible === HEARTBEAT_TOKEN || visible === HEARTBEAT_TRANSCRIPT_PROMPT) { return; @@ -155,26 +166,26 @@ export function flushSessionObserverAssistantNote(state: SessionObserverState): return; } state.lastAssistantNote = visible; - addSessionObserverNote(state, `Assistant: ${visible}`); + addActivityNote(state, `Assistant: ${visible}`, noteMaxChars); } -export function noteSessionObserverEvent( - state: SessionObserverState, +export function noteSessionActivityEvent( + state: SessionActivityNoteState, event: AgentEventPayload, - rememberItemStatus: (state: SessionObserverState, itemId: string, status: string) => boolean, + noteMaxChars: number = DEFAULT_NOTE_MAX_CHARS, ): void { const data = event.data; switch (event.stream) { case "lifecycle": { const phase = data.phase; if (phase === "start") { - addSessionObserverNote(state, "Run started"); + addActivityNote(state, "Run started", noteMaxChars); } else if (phase === "finishing") { - addSessionObserverNote(state, "Run is wrapping up"); + addActivityNote(state, "Run is wrapping up", noteMaxChars); } else if (phase === "end" || phase === "error") { const health = terminalHealthFor(event); const error = readString(data.error); - addSessionObserverNote(state, error ? `Run ${health}: ${error}` : `Run ${health}`); + addActivityNote(state, error ? `Run ${health}: ${error}` : `Run ${health}`, noteMaxChars); } return; } @@ -185,8 +196,8 @@ export function noteSessionObserverEvent( return; } const name = readString(data.name) ?? "tool"; - const args = summarizeSessionObserverToolArgs(data.args); - addSessionObserverNote(state, args ? `Tool ${name}: ${args}` : `Tool ${name}`); + const args = summarizeToolArgs(data.args); + addActivityNote(state, args ? `Tool ${name}: ${args}` : `Tool ${name}`, noteMaxChars); return; } case "command_output": { @@ -196,9 +207,10 @@ export function noteSessionObserverEvent( const title = readString(data.title) ?? readString(data.name) ?? "command"; const exitCode = readFiniteNumber(data.exitCode); const status = readString(data.status) ?? (exitCode === 0 ? "completed" : "failed"); - addSessionObserverNote( + addActivityNote( state, `${title}: ${status}${exitCode === undefined ? "" : ` (exit ${exitCode})`}`, + noteMaxChars, ); return; } @@ -212,10 +224,10 @@ export function noteSessionObserverEvent( if (!["running", "completed", "failed", "blocked"].includes(status)) { return; } - if (!rememberItemStatus(state, itemId, status)) { + if (!rememberItemStatus(state, itemId, status, MAX_ITEM_STATUSES)) { return; } - addSessionObserverNote(state, `${title}: ${status}`); + addActivityNote(state, `${title}: ${status}`, noteMaxChars); return; } case "plan": { @@ -229,11 +241,11 @@ export function noteSessionObserverEvent( }; for (const [index, step] of steps.entries()) { const itemId = `plan:${index}:${step.step}`; - if (!rememberItemStatus(state, itemId, step.status)) { + if (!rememberItemStatus(state, itemId, step.status, MAX_ITEM_STATUSES)) { continue; } const status = step.status === "in_progress" ? "running" : step.status; - addSessionObserverNote(state, `Plan: ${step.step}: ${status}`); + addActivityNote(state, `Plan: ${step.step}: ${status}`, noteMaxChars); } return; } @@ -241,10 +253,11 @@ export function noteSessionObserverEvent( const full = readString(data.text); const delta = readString(data.delta); if (full) { - state.assistantBuffer = assembleSessionObserverAssistantBuffer(full); + state.assistantBuffer = assembleAssistantBuffer(full, ASSISTANT_BUFFER_MAX_CHARS); } else if (delta) { - state.assistantBuffer = assembleSessionObserverAssistantBuffer( + state.assistantBuffer = assembleAssistantBuffer( state.assistantBuffer + delta, + ASSISTANT_BUFFER_MAX_CHARS, ); } return; @@ -253,9 +266,10 @@ export function noteSessionObserverEvent( if (data.status !== "pending" && data.phase !== "requested") { return; } - addSessionObserverNote( + addActivityNote( state, `Waiting for approval: ${readString(data.title) ?? "user action"}`, + noteMaxChars, ); break; } @@ -263,3 +277,26 @@ export function noteSessionObserverEvent( break; } } + +function readString(value: unknown): string | undefined { + return typeof value === "string" && value.trim() ? value : undefined; +} + +export function readFiniteNumber(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) ? value : undefined; +} + +export function terminalHealthFor(event: AgentEventPayload): "done" | "failed" { + const phase = event.data.phase; + const outcome = buildAgentRunTerminalOutcome({ + status: phase === "end" ? "ok" : "error", + error: event.data.error, + stopReason: event.data.stopReason, + livenessState: event.data.livenessState, + timeoutPhase: event.data.timeoutPhase, + providerStarted: event.data.providerStarted, + startedAt: event.data.startedAt, + endedAt: event.data.endedAt, + }); + return outcome.reason === "completed" ? "done" : "failed"; +} diff --git a/src/auto-reply/reply/progress-narrator.test.ts b/src/auto-reply/reply/progress-narrator.test.ts index 15139757dc98..ebdf04092dba 100644 --- a/src/auto-reply/reply/progress-narrator.test.ts +++ b/src/auto-reply/reply/progress-narrator.test.ts @@ -135,7 +135,7 @@ describe("progress narration through reply options", () => { expect(generate).toHaveBeenCalledTimes(1); expect(onUpdate).toHaveBeenCalledWith({ text: "Working on the request." }); expect(inputs[0]?.userMessage).toBe("change the default model"); - expect(inputs[0]?.activityNotes.join("\n")).toContain("ls"); + expect(inputs[0]?.activityNotes).toContain('Tool exec: {"command":"ls"}'); }); it("ignores non-work tools and non-start phases", async () => { @@ -294,7 +294,7 @@ describe("progress narration through reply options", () => { await flushNarrations(); expect(generate).toHaveBeenCalledTimes(1); - expect(inputs[0]?.activityNotes).toContain("model: Checking the current configuration."); + expect(inputs[0]?.activityNotes).toContain("Assistant: Checking the current configuration."); } finally { vi.useRealTimers(); } @@ -356,18 +356,26 @@ describe("progress narration through reply options", () => { expect(generate).toHaveBeenCalledTimes(2); }); - it("narrates failures immediately", async () => { + it("narrates metadata-only command failures immediately", async () => { const { narrator, generate, onUpdate, inputs } = createNarratorHarness({ texts: ["Running a command.", "The command failed, retrying."], }); narrator.noteToolStart({ name: "exec", phase: "start" }); await flushNarrations(); - narrator.noteCommandOutput({ name: "exec", title: "pnpm test", phase: "end", exitCode: 1 }); + narrator.noteCommandOutput({ + name: "exec", + title: "pnpm test", + phase: "end", + exitCode: 1, + output: "private command output must not reach narration", + }); await flushNarrations(); expect(generate).toHaveBeenCalledTimes(2); - expect(inputs[1]?.activityNotes.join("\n")).toContain("pnpm test failed (exit 1)"); + const notes = inputs[1]?.activityNotes.join("\n") ?? ""; + expect(notes).toContain("pnpm test: failed (exit 1)"); + expect(notes).not.toContain("private command output"); expect(onUpdate).toHaveBeenLastCalledWith({ text: "The command failed, retrying." }); }); @@ -452,7 +460,7 @@ describe("progress narration through reply options", () => { const notes = inputs.at(-1)?.activityNotes.join("\n") ?? ""; expect(notes).not.toContain("/etc/hosts"); - expect(notes).toContain("exec failed (exit 1)"); + expect(notes).toContain("exec: failed (exit 1)"); }); it("normalizes narration text to one bounded plain line", async () => { diff --git a/src/auto-reply/reply/progress-narrator.ts b/src/auto-reply/reply/progress-narrator.ts index bb8886092636..9fbce838099a 100644 --- a/src/auto-reply/reply/progress-narrator.ts +++ b/src/auto-reply/reply/progress-narrator.ts @@ -1,11 +1,16 @@ // Utility-model narration for channel progress drafts. -import { formatToolSummary, resolveToolDisplay } from "../../agents/tool-display.js"; +import { + createSessionActivityNoteState, + flushSessionActivityAssistantNote, + noteSessionActivityEvent, +} from "../../agents/session-activity-notes.js"; import { resolveUtilityModelRefForAgent } from "../../agents/utility-model.js"; import { PROGRESS_STATUS_PREAMBLE_FRESH_MS } from "../../channels/progress-draft-compositor.js"; import { sanitizeProgressStatusText } from "../../channels/progress-draft-status-text.js"; import { isChannelProgressDraftWorkToolName, isCommandToolName } from "../../channels/streaming.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { logVerbose } from "../../globals.js"; +import type { AgentEventPayload, AgentEventStream } from "../../infra/agent-events.js"; import { createSubsystemLogger } from "../../logging/subsystem.js"; import type { InternalGetReplyOptions } from "./get-reply.types.js"; import { @@ -21,37 +26,15 @@ const MIN_EVENTS_PER_NARRATION = 4; const MIN_INTERVAL_MS = 12_000; const NARRATION_MAX_CHARS = 280; const NARRATION_NOTE_MAX_CHARS = 160; -const MAX_ACTIVITY_NOTES = 40; const VISIBILITY_RETRY_MS = 1_000; // Keep hidden-draft polling bounded even when a channel never exposes the draft. const MAX_VISIBILITY_RETRIES = 30; const PREAMBLE_RETRY_EPSILON_MS = 1; const MAX_NARRATIONS_PER_TURN = 30; const MAX_CONSECUTIVE_FAILURES = 2; - -type ProgressNarrator = { - beginTurn: () => void; - stopTurn: () => void; - noteToolStart: (payload: { - name?: string; - phase?: string; - args?: Record; - }) => void; - noteCommandOutput: (payload: { - name?: string; - title?: string; - phase?: string; - status?: string; - exitCode?: number | null; - }) => void; - noteItemEvent: (payload: { - kind?: string; - name?: string; - title?: string; - status?: string; - progressText?: string; - }) => void; -}; +type ToolStartPayload = Parameters>[0]; +type CommandOutputPayload = Parameters>[0]; +type ItemEventPayload = Parameters>[0]; function normalizeNarrationText(raw: string): string { const collapsed = raw @@ -79,15 +62,15 @@ function createProgressNarrator(params: { now?: () => number; setTimeoutFn?: typeof setTimeout; clearTimeoutFn?: typeof clearTimeout; -}): ProgressNarrator { +}) { const now = params.now ?? Date.now; const setTimeoutFn = params.setTimeoutFn ?? setTimeout; const clearTimeoutFn = params.clearTimeoutFn ?? clearTimeout; - const notes: string[] = []; + let activity = createSessionActivityNoteState(); let disabled = false; let inFlight = false; let pendingImmediate = false; - let notesAtLastRun = -1; + let noteSequenceAtLastRun = -1; let lastRunAt = 0; let narrationCount = 0; let consecutiveFailures = 0; @@ -117,11 +100,11 @@ function createProgressNarrator(params: { // Queued turns reuse the narrator lifecycle but not the primary request. // Empty context is safer than describing follow-up work with stale intent. userMessage = ""; - notes.splice(0); + activity = createSessionActivityNoteState(); disabled = false; inFlight = false; pendingImmediate = false; - notesAtLastRun = -1; + noteSequenceAtLastRun = -1; lastRunAt = 0; narrationCount = 0; consecutiveFailures = 0; @@ -181,24 +164,39 @@ function createProgressNarrator(params: { return outcome.text; }); - const addNote = (note: string, options?: { immediate?: boolean }) => { + const recordEvent = ( + stream: AgentEventStream, + data: Record, + options?: { immediate?: boolean; flushAssistant?: boolean }, + ): void => { if (!turnActive || disabled || params.abortSignal?.aborted) { return; } visibilityRetryCount = 0; - notes.push(truncateAtWordBoundary(note.replace(/\s+/g, " ").trim(), NARRATION_NOTE_MAX_CHARS)); - if (notes.length > MAX_ACTIVITY_NOTES) { - notes.splice(0, notes.length - MAX_ACTIVITY_NOTES); + const sequenceBefore = activity.noteSequence; + const event: AgentEventPayload = { + runId: "progress-narrator", + seq: sequenceBefore + 1, + stream, + ts: now(), + data, + }; + noteSessionActivityEvent(activity, event, NARRATION_NOTE_MAX_CHARS); + if (options?.flushAssistant) { + flushSessionActivityAssistantNote(activity, NARRATION_NOTE_MAX_CHARS); + } + const added = activity.noteSequence > sequenceBefore; + if (added) { + maybeRun(options?.immediate === true); } - maybeRun(options?.immediate === true); }; const shouldRunNow = (immediate: boolean): boolean => { - const newNotes = notes.length - Math.max(0, notesAtLastRun); + const newNotes = activity.noteSequence - Math.max(0, noteSequenceAtLastRun); if (newNotes <= 0) { return false; } - if (immediate || notesAtLastRun < 0) { + if (immediate || noteSequenceAtLastRun < 0) { return true; } if (newNotes >= MIN_EVENTS_PER_NARRATION) { @@ -265,11 +263,11 @@ function createProgressNarrator(params: { inFlight = true; const runGeneration = turnGeneration; narrationCount += 1; - notesAtLastRun = notes.length; + noteSequenceAtLastRun = activity.noteSequence; lastRunAt = now(); const input: ProgressNarrationInput = { userMessage, - activityNotes: [...notes], + activityNotes: activity.notes.map((note) => note.text), previousText: lastText, }; void (async () => { @@ -322,16 +320,19 @@ function createProgressNarrator(params: { resetTurnState(); }, stopTurn, - noteToolStart(payload) { + noteToolStart(payload: ToolStartPayload) { if (payload.phase !== "start" || !isChannelProgressDraftWorkToolName(payload.name)) { return; } - const display = resolveToolDisplay({ name: payload.name, args: payload.args }); // Same command-tool set the draft formatter uses for commandText policy. - const hideDetail = params.hideCommandText === true && isCommandToolName(display.name); - addNote(formatToolSummary(hideDetail ? { ...display, detail: undefined } : display)); + const hideDetail = params.hideCommandText === true && isCommandToolName(payload.name); + recordEvent("tool", { + phase: "start", + name: payload.name, + ...(hideDetail ? {} : { args: payload.args }), + }); }, - noteCommandOutput(payload) { + noteCommandOutput(payload: CommandOutputPayload) { if (payload.phase !== "end") { return; } @@ -343,13 +344,22 @@ function createProgressNarrator(params: { } // Command-output titles usually carry the raw command text; honor the // channel's commandText: "status" policy for the failure note too. - const subject = params.hideCommandText + const title = params.hideCommandText ? payload.name || "command" : payload.title || payload.name || "command"; - const exit = typeof payload.exitCode === "number" ? ` (exit ${payload.exitCode})` : ""; - addNote(`${subject} failed${exit}`, { immediate: true }); + recordEvent( + "command_output", + { + phase: "end", + title, + name: payload.name, + status: "failed", + exitCode: payload.exitCode, + }, + { immediate: true }, + ); }, - noteItemEvent(payload) { + noteItemEvent(payload: ItemEventPayload) { if (payload.kind === "preamble") { const preambleText = sanitizeProgressStatusText(payload.progressText ?? "") .replace(/\s+/g, " ") @@ -358,13 +368,22 @@ function createProgressNarrator(params: { return; } lastPreambleAt = now(); - addNote(`model: ${preambleText}`); + recordEvent("assistant", { text: preambleText }, { flushAssistant: true }); return; } if (payload.status !== "failed") { return; } - addNote(`${payload.title || payload.name || "step"} failed`, { immediate: true }); + const title = payload.title || payload.name || "step"; + recordEvent( + "item", + { + itemId: payload.itemId || title, + title, + status: "failed", + }, + { immediate: true }, + ); }, }; } @@ -403,8 +422,8 @@ export function attachProgressNarratorToReplyOptions(params: { hideCommandText: opts.narrationHideCommandText === true, }); opts.onProgressNarratorLifecycle?.({ - beginTurn: narrator.beginTurn, - stopTurn: narrator.stopTurn, + beginTurn: () => narrator.beginTurn(), + stopTurn: () => narrator.stopTurn(), }); return { ...opts, diff --git a/src/gateway/session-observer-ask.ts b/src/gateway/session-observer-ask.ts index e0bc7087b408..0c16cc23c189 100644 --- a/src/gateway/session-observer-ask.ts +++ b/src/gateway/session-observer-ask.ts @@ -1,5 +1,6 @@ import type { SessionObserverDigest } from "../../packages/gateway-protocol/src/schema/sessions.js"; import { resolveSessionAgentId } from "../agents/agent-scope.js"; +import { flushSessionActivityAssistantNote } from "../agents/session-activity-notes.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; import { SessionObserverAskError, @@ -10,7 +11,6 @@ import { type SessionObserverDeps, type SessionObserverState, } from "./session-observer-model.js"; -import { flushSessionObserverAssistantNote } from "./session-observer-notes.js"; const observerLog = createSubsystemLogger("gateway/session-observer"); @@ -68,7 +68,7 @@ export function createSessionObserverAskRuntime(params: SessionObserverAskRuntim const getSnapshot = (sessionKey: string): SessionObserverSnapshot => { const state = params.states.get(sessionKey); if (state) { - flushSessionObserverAssistantNote(state); + flushSessionActivityAssistantNote(state); return { agentId: state.agentId, runId: state.runId, diff --git a/src/gateway/session-observer-model.ts b/src/gateway/session-observer-model.ts index 72e5393492a9..a722bff647c7 100644 --- a/src/gateway/session-observer-model.ts +++ b/src/gateway/session-observer-model.ts @@ -6,7 +6,10 @@ import { type SessionObserverHealth, type SessionObserverPlanProgress, } from "../../packages/gateway-protocol/src/schema/sessions.js"; -import { buildAgentRunTerminalOutcome } from "../agents/agent-run-terminal-outcome.js"; +import { + terminalHealthFor, + type SessionActivityNoteState, +} from "../agents/session-activity-notes.js"; import type { completeWithPreparedSimpleCompletionModel, prepareSimpleCompletionModelForAgent, @@ -34,7 +37,7 @@ type PrepareModel = typeof prepareSimpleCompletionModelForAgent; type CompleteModel = typeof completeWithPreparedSimpleCompletionModel; export type PreparedModel = Awaited>; -export type SessionObserverState = { +export type SessionObserverState = SessionActivityNoteState & { sessionKey: string; sessionId?: string; runId: string; @@ -47,14 +50,7 @@ export type SessionObserverState = { revision: number; digestCount: number; consecutiveFailures: number; - noteSequence: number; lastDigestNoteSequence: number; - notes: Array<{ sequence: number; text: string; bytes: number }>; - noteBytes: number; - itemStatuses: Map; - assistantBuffer: string; - lastAssistantNote?: string; - planProgress?: SessionObserverPlanProgress; previousDigest?: SessionObserverDigest; preparedPromise?: Promise; activeController?: AbortController; @@ -290,50 +286,6 @@ export function isTerminalLifecycleEvent(event: AgentEventPayload): boolean { ); } -export function readString(value: unknown): string | undefined { - return typeof value === "string" && value.trim() ? value : undefined; -} - -export function readFiniteNumber(value: unknown): number | undefined { - return typeof value === "number" && Number.isFinite(value) ? value : undefined; -} - -export function rememberSessionObserverItemStatus( - statuses: Map, - itemId: string, - status: string, - limit: number, -): boolean { - if (statuses.get(itemId) === status) { - return false; - } - statuses.delete(itemId); - statuses.set(itemId, status); - while (statuses.size > limit) { - const oldest = statuses.keys().next().value; - if (oldest === undefined) { - break; - } - statuses.delete(oldest); - } - return true; -} - -export function terminalHealthFor(event: AgentEventPayload): "done" | "failed" { - const phase = event.data.phase; - const outcome = buildAgentRunTerminalOutcome({ - status: phase === "end" ? "ok" : "error", - error: event.data.error, - stopReason: event.data.stopReason, - livenessState: event.data.livenessState, - timeoutPhase: event.data.timeoutPhase, - providerStarted: event.data.providerStarted, - startedAt: event.data.startedAt, - endedAt: event.data.endedAt, - }); - return outcome.reason === "completed" ? "done" : "failed"; -} - export async function synthesizeSessionObserverTerminalDigest(params: { source: { event?: AgentEventPayload; state?: SessionObserverState }; dormant?: DormantSessionObserverRun; diff --git a/src/gateway/session-observer.ts b/src/gateway/session-observer.ts index ed0f904b0c46..f4d7eadf18da 100644 --- a/src/gateway/session-observer.ts +++ b/src/gateway/session-observer.ts @@ -1,4 +1,11 @@ import type { SessionObserverDigest } from "../../packages/gateway-protocol/src/schema/sessions.js"; +import { + createSessionActivityNoteState, + flushSessionActivityAssistantNote, + noteSessionActivityEvent, + readFiniteNumber, + terminalHealthFor, +} from "../agents/session-activity-notes.js"; import { resolveUtilityModelRefForAgent } from "../agents/utility-model.js"; import { getAgentRunContext } from "../infra/agent-events.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; @@ -14,15 +21,12 @@ import { isTerminalLifecycleEvent, markSessionObserverRunSuperseded, normalizeSessionObserverModelOutput, - readFiniteNumber, rememberSessionObserverDisabledRun, rememberSessionObserverDormantRun, - rememberSessionObserverItemStatus, rememberSessionObserverRevisionFloor, SESSION_OBSERVER_MODEL_MAX_TOKENS, SESSION_OBSERVER_SYSTEM_PROMPT, synthesizeSessionObserverTerminalDigest, - terminalHealthFor, } from "./session-observer-model.js"; import type { DormantSessionObserverRun, @@ -31,10 +35,6 @@ import type { SessionObserverRevisionFloor, SessionObserverState, } from "./session-observer-model.js"; -import { - flushSessionObserverAssistantNote, - noteSessionObserverEvent, -} from "./session-observer-notes.js"; const observerLog = createSubsystemLogger("gateway/session-observer"); @@ -44,7 +44,6 @@ const MODEL_TIMEOUT_MS = 10_000; const MAX_DIGESTS_PER_RUN = 40; const MAX_LIVE_DIGESTS_PER_RUN = MAX_DIGESTS_PER_RUN - 1; const MAX_CONSECUTIVE_FAILURES = 2; -const MAX_ITEM_STATUSES = 160; const FINAL_DIGEST_MIN_RUN_MS = 30_000; const PERSIST_INTERVAL_MS = 60_000; // The Control UI opens at most six live session subscriptions; matching that cap @@ -193,9 +192,6 @@ export function createSessionObserver(deps: SessionObserverDeps): SessionObserve return resolveUtilityModelRef({ cfg, agentId: state.agentId }) === state.utilityModelRef; }; - const rememberItemStatus = (state: SessionObserverState, itemId: string, status: string) => - rememberSessionObserverItemStatus(state.itemStatuses, itemId, status, MAX_ITEM_STATUSES); - const ensurePrepared = async (state: SessionObserverState): Promise => { state.preparedPromise ??= prepareModel({ cfg: deps.getConfig(), @@ -372,7 +368,7 @@ export function createSessionObserver(deps: SessionObserverDeps): SessionObserve if (state.digestCount >= digestLimit) { return; } - flushSessionObserverAssistantNote(state); + flushSessionActivityAssistantNote(state); const selectedNotes = pendingNotes(state); if (!final && selectedNotes.length < MIN_NOTES_PER_DIGEST) { return; @@ -501,16 +497,12 @@ export function createSessionObserver(deps: SessionObserverDeps): SessionObserve if (dormant) { dormantRuns.delete(event.runId); const state: SessionObserverState = { + ...createSessionActivityNoteState(), ...dormant, utilityModelRef, lastActivityAt: event.ts, lastRunAt: now(), - noteSequence: 0, lastDigestNoteSequence: 0, - notes: [], - noteBytes: 0, - itemStatuses: new Map(), - assistantBuffer: "", inFlight: false, finalPending: false, }; @@ -521,6 +513,7 @@ export function createSessionObserver(deps: SessionObserverDeps): SessionObserve const startedAt = readFiniteNumber(event.data.startedAt) ?? session?.startedAt ?? event.ts ?? now(); const state: SessionObserverState = { + ...createSessionActivityNoteState(), sessionKey, sessionId: event.sessionId ?? session?.sessionId, runId: event.runId, @@ -533,12 +526,7 @@ export function createSessionObserver(deps: SessionObserverDeps): SessionObserve revision: session?.observerDigest?.revision ?? 0, digestCount: 0, consecutiveFailures: 0, - noteSequence: 0, lastDigestNoteSequence: 0, - notes: [], - noteBytes: 0, - itemStatuses: new Map(), - assistantBuffer: "", previousDigest: session?.observerDigest, inFlight: false, finalPending: false, @@ -642,7 +630,7 @@ export function createSessionObserver(deps: SessionObserverDeps): SessionObserve if (eventStartedAt !== undefined) { state.startedAt = Math.min(state.startedAt, eventStartedAt); } - noteSessionObserverEvent(state, event, rememberItemStatus); + noteSessionActivityEvent(state, event); if (terminal) { state.terminalHealth = terminalHealthFor(event); const endedAt = readFiniteNumber(event.data.endedAt) ?? now();