fix(sessions): bound the transcript indexer (v0.461.2)

CHANGELOG.md

@@ -8,6 +8,17 @@ new version heading in the same commit.
 
 ## [Unreleased]
 
+## [0.461.2] - 2026-10-13
+### Fixed
+- **Session search's transcript indexer no longer stalls the server on long transcripts.** v0.461.0
+  parsed each conversation's whole transcript on the event loop and re-read a live run on every
+  scheduler tick. On expresstech one 89 MB transcript took 4.1 s to parse, so v0.461.0 was rolled back
+  there. The indexer now reads at most 1 MB of a transcript (head and tail, cut at line boundaries), skips
+  a file whose mtime hasn't changed, re-reads a live run at most every 10 minutes, and strips harness
+  markup (system reminders, slash-command tags) from what it indexes. Measured against expresstech's real
+  data: the 1,042-conversation backlog drains in about 180 ticks of roughly 150 ms each, and a steady-state
+  sweep takes about 11 ms. Digests written by v0.461.0 are dropped once and rebuilt under these rules.
+
 ## [0.461.1] - 2026-10-13
 ### Fixed
 - **`drift-nudge-test` no longer fails the deploy gate on a loaded box.** It waited a flat 150ms for the

CLAUDE.md

@@ -391,7 +391,9 @@ Key modules:
   handle for revise/forget). Episodic self-query (the run-history companion to semantic memory):
   `session_history` searches past sessions — with a `query`, ranked FTS5/bm25 over each run's title/task/
   report summary (`term_sessions_fts`, trigger-kept) AND what was said in it (`session_text_fts`, one row per
-  conversation, filled off the scheduler tick by `indexSessionText` — see `src/edge/session-search.ts`).
+  conversation, filled off the scheduler tick by `indexSessionText` — see `src/edge/session-search.ts`;
+  ⚠ it runs ON the event loop, so every read is bounded: ≤1 MB head+tail of the file, only when its mtime
+  moved, a live run at most every 10 min — v0.461.0 parsed whole transcripts and an 89 MB one cost 4.1 s).
   Scope `agent` (default) = own runs; `member` = everything the run-as human can see in the console
   (`canReadRow`, audited `session.searched`). `session_open` reopens one — the friendly transcript timeline
   (`readConversation`) or a `summary:true` recap — own-agent, or member-visible (audited `session.opened`).

package.json

@@ -1,6 +1,6 @@
 {
   "name": "agent-os",
-  "version": "0.461.1",
+  "version": "0.461.2",
   "description": "A generic, governed operating system for running autonomous agents safely across brands. Ships with a local web console.",
   "license": "MIT",
   "type": "commonjs",

scripts/session-search-test.cjs

@@ -46,6 +46,8 @@ const long = [turn('user', 'build the feature pages'), ...Array.from({ length: 4
 const dig = digestConversation(long, 8 * 1024);
 assert(dig.length <= 8 * 1024, 'the digest respects its cap', dig.length);
 assert(dig.includes('feature pages') && dig.includes('paypal webhook'), 'squeezing keeps BOTH the first and the last topic of a long run');
+assert(digestConversation([turn('user', '<command-name>/goal</command-name> <command-args>fix the menu</command-args><system-reminder>secret harness text</system-reminder>')]) === '/goal fix the menu',
+  'harness markup and system reminders never reach the index', digestConversation([turn('user', '<command-name>/goal</command-name> <command-args>fix the menu</command-args><system-reminder>secret harness text</system-reminder>')]));
 assert(!digestConversation([{ kind: 'activity', tool: 'Bash', label: 'Ran a command', detail: 'curl /wp-json/edd/webhooks/v1/paypal', status: 'ok' }]).includes('undefined')
   && digestConversation([{ kind: 'activity', tool: 'Bash', label: 'Ran a command', detail: 'curl /wp-json/edd/webhooks/v1/paypal', status: 'ok' }]).includes('/wp-json/edd/webhooks'),
   'an action’s hint (a path, a URL) is searchable');
@@ -112,6 +114,7 @@ assert(!digestConversation([{ kind: 'activity', tool: 'Bash', label: 'Ran a comm
 
   console.log('\n\x1b[1m3) what was said in the run — the transcript index\x1b[0m');
   const CONVO = '11111111-2222-3333-4444-555555555555';
+  let mt = 0;
   writeTranscript(CONVO, [
     { role: 'user', text: 'Build the accessible mobile menu feature pages' },
     { role: 'assistant', text: 'Built three feature pages.' },
@@ -126,18 +129,45 @@ assert(!digestConversation([{ kind: 'activity', tool: 'Bash', label: 'Ran a comm
     { role: 'user', text: 'Every PayPal IPN is failing with a 500 now, can you look?' },
     { role: 'assistant', text: 'The live PayPal client credentials were removed; restored them and the webhook retry succeeded.' },
   ]);
+  const file = path.join(projects, `${CONVO}.jsonl`);
+  const bumpMtime = () => { const t = new Date(Date.now() + (++mt) * 60_000); fs.utimesSync(file, t, t); };
+  bumpMtime();
   aos.db.prepare('UPDATE term_sessions SET last_activity = ? WHERE id = ?').run(Date.now(), longRun);
-  tm.indexSessionText(5_000);
+  assert(tm.indexSessionText(5_000) === 0, 'a run that just got read is not re-read inside the 10-minute floor, however active');
+  const LATER = Date.now() + 11 * 60_000;
+  assert(tm.indexSessionText(5_000, LATER) === 1, 'after the floor, the changed transcript is read again');
   const pp = tm.searchSessions({ query: 'paypal webhook', agent: 'ops' });
   assert(ids(pp).join() === longRun, 'a topic that exists ONLY in the transcript finds the run once it continues', ids(pp));
   assert(pp[0] && /«paypal»/i.test(pp[0].snippet || ''), 'with a snippet marking where it matched', pp[0] && pp[0].snippet);
   const resumed = mkRun({ agent: 'ops', title: 'poke: follow up', claude_session_id: CONVO, run_as: alice.id, spawned_by: alice.id });
   const both = ids(tm.searchSessions({ query: 'paypal', agent: 'ops' }));
   assert(both.length === 1 && (both[0] === longRun || both[0] === resumed), 'two runs of ONE conversation are one hit', both);
-  tm.indexSessionText(5_000); // picks up the resumed row's conversation if it looks newer
   const before = aos.db.prepare('SELECT COUNT(*) AS c FROM session_text_fts').get().c;
-  assert(tm.indexSessionText(5_000) === 0, 'a sweep with no new activity reads no transcript');
-  assert(aos.db.prepare('SELECT COUNT(*) AS c FROM session_text_fts').get().c === before, 'and one conversation is one indexed row, however many runs it has');
+  // New activity on the row, but the transcript FILE did not change: the watermark moves, nothing is read.
+  aos.db.prepare('UPDATE term_sessions SET last_activity = ? WHERE id = ?').run(Date.now() + 1000, resumed);
+  assert(tm.indexSessionText(5_000, LATER + 11 * 60_000) === 0, 'activity with an unchanged transcript file reads nothing');
+  assert(tm.indexSessionText(5_000, LATER + 22 * 60_000) === 0, 'and the next sweep has nothing left to do');
+  assert(aos.db.prepare('SELECT COUNT(*) AS c FROM session_text_fts').get().c === before, 'one conversation is one indexed row, however many runs it has');
+
+  // A days-long transcript: far over the read ceiling. Only head + tail are parsed, and both topics survive.
+  const { readConversationBounded } = require(path.join(ROOT, 'dist/edge/conversation.js'));
+  const BIG = '99999999-2222-3333-4444-555555555555';
+  const filler = 'x'.repeat(2000);
+  writeTranscript(BIG, [
+    { role: 'user', text: 'Start: audit the checkout funnel' },
+    ...Array.from({ length: 3000 }, (_, i) => ({ role: 'assistant', text: `step ${i} ${filler}` })),
+    { role: 'user', text: 'Last thing: rotate the Mailgun key' },
+  ]);
+  const bigFile = path.join(projects, `${BIG}.jsonl`);
+  const size = fs.statSync(bigFile).size;
+  const t0 = Date.now();
+  const bounded = readConversationBounded(bigFile, 256 * 1024, 64 * 1024);
+  const texts = bounded.turns.map((x) => x.text || '').join(' ');
+  assert(size > 4 * 1024 * 1024 && bounded.found, `a ${(size / 1048576).toFixed(1)} MB transcript is read bounded`);
+  assert(/checkout funnel/.test(texts) && /Mailgun/.test(texts), 'the head (first ask) and the tail (latest topic) both survive');
+  assert(bounded.turns.length < 200, 'and only a slice of the middle is parsed', bounded.turns.length);
+  assert(bounded.turns.every((x) => x.kind !== 'user' || typeof x.text === 'string'), 'no half-line was parsed into a turn');
+  assert(Date.now() - t0 < 1000, 'quickly', Date.now() - t0);
 
   console.log('\n\x1b[1m4) the loopback routes — scopes, urls, session_open\x1b[0m');
   const bobRun = mkRun({ agent: 'ops', title: 'paypal refund for bob', run_as: bob.id, spawned_by: bob.id });

src/edge/conversation.ts

@@ -278,7 +278,41 @@ export function readConversation(claudeSessionId: string): Conversation {
   } catch {
     return { turns: [], found: false };
   }
+  return parseConversationText(raw);
+}
+
+/**
+ * A transcript FILE read with a byte ceiling: under `maxBytes` the whole thing, over it the first
+ * `headBytes` and the last `maxBytes - headBytes`, cut at line boundaries (a sliced first/last line is
+ * dropped, never half-parsed). For BACKGROUND readers only — the session search indexer — which must not
+ * block the event loop: a days-long transcript is tens of MB (89 MB on a live tenant; 4.1 s to parse on
+ * a 2 GB box), and the console's own on-demand read keeps using {@link readConversation}.
+ */
+export function readConversationBounded(file: string, maxBytes: number, headBytes: number): Conversation {
+  let fd: number | undefined;
+  try {
+    fd = fs.openSync(file, 'r');
+    const size = fs.fstatSync(fd).size;
+    if (size <= maxBytes) return parseConversationText(fs.readFileSync(fd, 'utf8'));
+    const slice = (start: number, len: number): string => {
+      const buf = Buffer.alloc(len);
+      const n = fs.readSync(fd!, buf, 0, len, start);
+      return buf.subarray(0, n).toString('utf8');
+    };
+    const head = slice(0, headBytes);
+    const tail = slice(size - (maxBytes - headBytes), maxBytes - headBytes);
+    const headLines = head.slice(0, head.lastIndexOf('\n') + 1);
+    const tailLines = tail.slice(tail.indexOf('\n') + 1);
+    return parseConversationText(headLines + tailLines);
+  } catch {
+    return { turns: [], found: false };
+  } finally {
+    if (fd !== undefined) try { fs.closeSync(fd); } catch { /* already closed */ }
+  }
+}
 
+/** The line parser behind {@link readConversation}: JSONL transcript text → ordered turns. */
+export function parseConversationText(raw: string): Conversation {
   const turns: ChatTurn[] = [];
   // tool_use_id → index in `turns`, so the tool_result on the NEXT user message resolves the card.
   const activityById = new Map<string, number>();

src/edge/session-search.ts

@@ -16,6 +16,14 @@ import { consolePage } from '../governance/chat-links';
 /** Upper bound on one conversation's indexed text. Most runs are far below it; a days-long session is
  *  not, and indexing its whole transcript would make the FTS table the largest thing in the DB. */
 export const SESSION_TEXT_CAP = 32 * 1024;
+/** The indexer reads at most this much of a transcript file (head + tail), and the head part of it. A
+ *  background read runs on the server's event loop: measured on a 2 GB droplet against its real
+ *  transcripts (up to 89 MB), a 2 MB ceiling still put p95 sweeps at ~200 ms; the digest keeps 32 KB of
+ *  text either way, so the extra middle bought nothing. */
+export const SESSION_TEXT_READ_MAX = 1024 * 1024;
+export const SESSION_TEXT_READ_HEAD = 256 * 1024;
+/** A conversation that keeps changing (a live run) is re-read at most this often. */
+export const SESSION_TEXT_REREAD_MS = 10 * 60_000;
 /** Over the cap, a human prompt keeps this much and the agent's words/actions this much each. */
 const USER_PIECE = 400;
 const OTHER_PIECE = 160;
@@ -34,7 +42,7 @@ export function digestConversation(turns: ChatTurn[], cap = SESSION_TEXT_CAP): s
   const pieces: { text: string; user: boolean }[] = [];
   for (const t of turns) {
     if (t.kind === 'user' || t.kind === 'assistant') {
-      const s = t.text.replace(/\s+/g, ' ').trim();
+      const s = stripHarness(t.text).replace(/\s+/g, ' ').trim();
       if (s) pieces.push({ text: s, user: t.kind === 'user' });
     } else if (t.kind === 'activity') {
       const s = [t.label, t.detail].filter(Boolean).join(' ').replace(/\s+/g, ' ').trim();
@@ -66,6 +74,15 @@ export function digestConversation(turns: ChatTurn[], cap = SESSION_TEXT_CAP): s
   return [...keep.entries()].sort((x, y) => x[0] - y[0]).map(([, s]) => s).join('\n');
 }
 
+/** Drop what the harness injects into the user role — system reminders whole, and the bare tags around
+ *  slash-command text (`<command-name>`, `<local-command-stdout>`, …) — so they never become search hits
+ *  or snippets. The words a person typed inside a command's args are kept. */
+function stripHarness(text: string): string {
+  return text
+    .replace(/<system-reminder>[\s\S]*?<\/system-reminder>/g, ' ')
+    .replace(/<\/?(?:command|local-command)-[a-z-]+>/g, ' ');
+}
+
 /** `k` items spread evenly across `arr`, always including the first and the last. */
 function sampleEvenly<T>(arr: T[], k: number): T[] {
   if (k <= 0) return [];

src/state/db.ts

@@ -1480,6 +1480,14 @@ function migrate(db: Db): void {
   `);
   // Existing rows predate the triggers — index them once, on the boot that creates the table.
   if (!hadSessionFts) db.exec("INSERT INTO term_sessions_fts(term_sessions_fts) VALUES ('rebuild')");
+  // v0.461.2: the transcript file's mtime at the last read (unchanged file → no re-read) and when that read
+  // happened (a live run is re-read at most every 10 min). See TerminalManager.indexSessionText.
+  const preBounded = !(db.prepare('PRAGMA table_info(session_text_index)').all() as Array<{ name: string }>).some((c) => c.name === 'mtime');
+  addColumn(db, 'session_text_index', 'mtime', 'INTEGER');
+  addColumn(db, 'session_text_index', 'read_at', 'INTEGER');
+  // Rows written by v0.461.0 hold digests made under the old rules (harness markup included) and would
+  // never be re-read while their run stays quiet — start the transcript half over once.
+  if (preBounded) db.exec('DELETE FROM session_text_index; DELETE FROM session_text_fts;');
 }
 
 /** Add a column only if it isn't already present (SQLite has no ADD COLUMN IF NOT EXISTS). */