From b28465853964543902a35c002c8ff17e1ab7f844 Mon Sep 17 00:00:00 2001 From: Vikas Singhal Date: Tue, 13 Oct 2026 14:46:57 +0530 Subject: [PATCH] =?UTF-8?q?fix(sessions):=20bound=20the=20transcript=20ind?= =?UTF-8?q?exer=20=E2=80=94=20capped=20head+tail=20read,=20mtime=20skip,?= =?UTF-8?q?=2010-min=20re-read=20floor=20(v0.461.2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit v0.461.0 parsed whole transcripts on the event loop and re-read a live run every tick; on expresstech an 89 MB transcript cost a 4.1s parse, so it was rolled back there. Now: <=1 MB head+tail read cut at line boundaries (readConversationBounded), no re-read while the file's mtime is unchanged, a live run re-read at most every 10 min, harness markup stripped from digests, and v0.461.0 digests dropped once. Measured on expresstech's real data: 1,042-convo backlog in ~180 ticks of ~150ms, steady sweep ~11ms. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 11 ++++++ CLAUDE.md | 4 +- package-lock.json | 4 +- package.json | 2 +- scripts/session-search-test.cjs | 38 ++++++++++++++++-- src/edge/conversation.ts | 34 ++++++++++++++++ src/edge/session-search.ts | 19 ++++++++- src/state/db.ts | 8 ++++ src/terminal.ts | 70 ++++++++++++++++++++++++--------- 9 files changed, 163 insertions(+), 27 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 00d6ff3e..74c23dc9 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CLAUDE.md b/CLAUDE.md index 0edbe4b8..f760e17e 100644 --- a/CLAUDE.md +++ b/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`). diff --git a/package-lock.json b/package-lock.json index 1636c444..d8821d8d 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "agent-os", - "version": "0.461.1", + "version": "0.461.2", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "agent-os", - "version": "0.461.1", + "version": "0.461.2", "license": "MIT", "bin": { "agent-os": "bin/agent-os" diff --git a/package.json b/package.json index 68609e73..9efd07f7 100644 --- a/package.json +++ b/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", diff --git a/scripts/session-search-test.cjs b/scripts/session-search-test.cjs index 981bc1e4..37494d83 100644 --- a/scripts/session-search-test.cjs +++ b/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', '/goal fix the menusecret harness text')]) === '/goal fix the menu', + 'harness markup and system reminders never reach the index', digestConversation([turn('user', '/goal fix the menusecret harness text')])); 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 }); diff --git a/src/edge/conversation.ts b/src/edge/conversation.ts index 9c8b3b07..c8c0d106 100644 --- a/src/edge/conversation.ts +++ b/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(); diff --git a/src/edge/session-search.ts b/src/edge/session-search.ts index 51021605..651d752a 100644 --- a/src/edge/session-search.ts +++ b/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 (``, ``, …) — 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(/[\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(arr: T[], k: number): T[] { if (k <= 0) return []; diff --git a/src/state/db.ts b/src/state/db.ts index 783f1ff9..fff7aed7 100644 --- a/src/state/db.ts +++ b/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). */ diff --git a/src/terminal.ts b/src/terminal.ts index 47132f01..7a7912c5 100644 --- a/src/terminal.ts +++ b/src/terminal.ts @@ -50,8 +50,8 @@ import { readSessionCost } from './edge/session-cost'; import { readTranscriptEnd } from './edge/outcome'; import { pendingBackgroundWork, BACKGROUND_GRACE_MS, UNATTENDED_TURN_BRIEF, WAITING_BRIEF } from './edge/background-work'; import { findCodexRollout, readCodexCost, readCodexConversation } from './edge/codex-transcript'; -import { readConversation, findTranscript, registerTranscriptRoot, Conversation } from './edge/conversation'; -import { digestConversation, sessionConsoleUrl, toSessionFtsQuery } from './edge/session-search'; +import { readConversation, readConversationBounded, findTranscript, registerTranscriptRoot, Conversation } from './edge/conversation'; +import { digestConversation, sessionConsoleUrl, toSessionFtsQuery, SESSION_TEXT_READ_HEAD, SESSION_TEXT_READ_MAX, SESSION_TEXT_REREAD_MS } from './edge/session-search'; // Video render tuning: a submitted job renders async. The in-call path polls briefly for the fast case; // the tick poller finishes the rest, bounded by a TTL + a poll ceiling so a stuck render can't linger. @@ -2063,10 +2063,16 @@ export class TerminalManager { /** * Fill `session_text_fts` — what was said in each conversation — a slice at a time, off the scheduler - * tick. A conversation is (re)read once its rows show activity after its watermark, so a finished run - * is read once and a live one is caught up as it goes; the backlog of an existing tenant drains newest - * first over the first minutes after deploy. Bounded by `budgetMs` because transcript reads are - * synchronous and a days-long transcript is megabytes. Returns how many conversations it indexed. + * tick. Runs on the server's event loop, so every step is bounded — v0.461.0 wasn't, and on a live + * tenant one 89 MB transcript cost a 4.1 s synchronous parse that repeated every tick while its run was + * active: + * - a transcript is only read once its FILE changed (mtime) — activity on a row whose transcript + * didn't move just advances the watermark; + * - a conversation that keeps changing is re-read at most every {@link SESSION_TEXT_REREAD_MS}; + * - a read is capped at {@link SESSION_TEXT_READ_MAX} bytes (head + tail, `readConversationBounded`); + * - the sweep stops starting reads once `budgetMs` is spent. + * Newest activity first, so a finished run is searchable within a tick or two and an existing tenant's + * backlog drains over the first hours after deploy. Returns how many transcripts it read. */ indexSessionText(budgetMs = 150, now = Date.now()): number { const started = Date.now(); @@ -2076,26 +2082,54 @@ export class TerminalManager { FROM term_sessions s LEFT JOIN session_text_index i ON i.convo = COALESCE(s.claude_session_id, s.id) WHERE s.archived_at IS NULL GROUP BY COALESCE(s.claude_session_id, s.id) - HAVING MAX(i.indexed_at) IS NULL OR active > MAX(i.indexed_at) + HAVING MAX(i.indexed_at) IS NULL + OR (active > MAX(i.indexed_at) AND COALESCE(MAX(i.read_at), 0) < ?) ORDER BY active DESC - LIMIT 40`).all<{ convo: string; id: string; active: number }>(); + LIMIT 40`).all<{ convo: string; id: string; active: number }>(now - SESSION_TEXT_REREAD_MS); + if (!stale.length) return 0; + const prior = this.db.prepare('SELECT mtime FROM session_text_index WHERE convo = ?'); const del = this.db.prepare('DELETE FROM session_text_fts WHERE convo = ?'); const ins = this.db.prepare('INSERT INTO session_text_fts (convo, body) VALUES (?, ?)'); - const mark = this.db.prepare(`INSERT INTO session_text_index (convo, indexed_at, bytes) VALUES (?, ?, ?) - ON CONFLICT(convo) DO UPDATE SET indexed_at = excluded.indexed_at, bytes = excluded.bytes`); - let n = 0; + const mark = this.db.prepare(`INSERT INTO session_text_index (convo, indexed_at, bytes, mtime, read_at) VALUES (?, ?, ?, ?, ?) + ON CONFLICT(convo) DO UPDATE SET indexed_at = excluded.indexed_at, bytes = excluded.bytes, + mtime = excluded.mtime, read_at = excluded.read_at`); + const touch = this.db.prepare('UPDATE session_text_index SET indexed_at = ? WHERE convo = ?'); + this.refreshTranscriptRoots(); + let reads = 0; for (const row of stale) { - if (n > 0 && Date.now() - started > budgetMs) break; + if (reads > 0 && Date.now() - started > budgetMs) break; + // Stamp the watermark at the row's own activity time, not `now`: activity that lands while we read + // is still newer than the mark, so the next sweep picks it up. + const watermark = Math.min(row.active ?? now, now); + const src = this.transcriptSource(row.id); + const was = prior.get<{ mtime: number | null }>(row.convo); + if (was && src.mtime !== undefined && was.mtime === src.mtime) { touch.run(watermark, row.convo); continue; } let body = ''; - try { body = digestConversation(this.sessionConversation(row.id).turns); } catch { /* unreadable → index nothing, retry on next activity */ } + try { + const convo = src.file ? readConversationBounded(src.file, SESSION_TEXT_READ_MAX, SESSION_TEXT_READ_HEAD) + : src.other ? this.sessionConversation(row.id) : { turns: [], found: false }; + body = digestConversation(convo.turns); + } catch { /* unreadable → index nothing, retry when the file changes */ } del.run(row.convo); if (body) ins.run(row.convo, body); - // Stamp the watermark at the row's own activity time, not `now`: activity that lands while we read - // is still newer than the mark, so the next sweep picks it up. - mark.run(row.convo, Math.min(row.active ?? now, now), body.length); - n++; + mark.run(row.convo, watermark, body.length, src.mtime ?? null, Date.now()); + reads++; } - return n; + return reads; + } + + /** Where a session's transcript lives, for the background indexer: a Claude Code JSONL `file` (read + * bounded) with its mtime, or `other` for a runtime whose transcript has its own reader (Codex). */ + private transcriptSource(sessionId: string): { file?: string; mtime?: number; other?: boolean } { + const row = this.db.prepare('SELECT agent, claude_session_id FROM term_sessions WHERE id = ?') + .get<{ agent: string; claude_session_id: string | null }>(sessionId); + if (!row) return {}; + const runtime = this.os.agents.get(row.agent)?.runtime; + if (runtime === 'codex') return { other: true }; + if (!runtimeSupports(runtime, 'transcript') || !row.claude_session_id) return {}; + const file = findTranscript(row.claude_session_id); + if (!file) return {}; + try { return { file, mtime: Math.floor(fs.statSync(file).mtimeMs) }; } catch { return {}; } } /**