refactor(agents): one session-activity note pipeline for observer and narrator (#112553)

* refactor(agents): one session-activity note pipeline for observer and narrator

* refactor(agents): keep note reader helper module-local

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<string, string>;
+  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";
+}

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 () => {

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<string, unknown>;
-  }) => 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<NonNullable<InternalGetReplyOptions["onToolStart"]>>[0];
+type CommandOutputPayload = Parameters<NonNullable<InternalGetReplyOptions["onCommandOutput"]>>[0];
+type ItemEventPayload = Parameters<NonNullable<InternalGetReplyOptions["onItemEvent"]>>[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<string, unknown>,
+    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,

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,

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<ReturnType<PrepareModel>>;
 
-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<string, string>;
-  assistantBuffer: string;
-  lastAssistantNote?: string;
-  planProgress?: SessionObserverPlanProgress;
   previousDigest?: SessionObserverDigest;
   preparedPromise?: Promise<PreparedModel>;
   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<string, string>,
-  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;

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<PreparedModel> => {
     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();