Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170/** Read-only session JSONL projections used by daemon sync and replay. */import { join } from 'node:path';import { existsSync, readFileSync } from 'node:fs';import { splitJsonl } from '@/protocol/framing/jsonl';import type { Message } from '@/types/message';import type { SessionEntry, MessageEntry, CompactionEntry } from '@/types/session-entry';import { SessionStore } from '@/core/session/store';import { activeChain, latestEntryId } from '@/core/session/lineage';
/** Hard wire-side bounds for explicit replay. */export const MAX_REPLAY_ENTRIES = 10_000;/** UTF-8 bytes of serialized entries; replay envelope overhead is not included. */export const MAX_REPLAY_BYTES = 4 * 1024 * 1024;
export interface ReplaySnapshot { entries: SessionEntry[]; /** The active lineage snapshot pinned by the initial request. */ tail_id?: string; /** The oldest returned entry; pass it as before_entry_id for the next page. */ before_entry_id?: string; has_older: boolean; /** Compatibility alias retained for older protocol clients. */ truncated: boolean;}
export interface ReplayPageOptions { limit?: number; tail_id?: string; before_entry_id?: string;}
export class ReplayCursorError extends Error { readonly code = 'INVALID_CURSOR'; constructor(message: string) { super(message); this.name = 'ReplayCursorError'; }}
export class ReplayEntryTooLargeError extends Error { readonly code = 'REPLAY_ENTRY_TOO_LARGE'; readonly entryId: string; readonly byteCount: number; readonly maxBytes: number;
constructor(entryId: string, byteCount: number, maxBytes = MAX_REPLAY_BYTES) { super(`Replay entry ${entryId} is ${byteCount} bytes, exceeding the ${maxBytes}-byte cap`); this.name = 'ReplayEntryTooLargeError'; this.entryId = entryId; this.byteCount = byteCount; this.maxBytes = maxBytes; }}
/** * Read one backwards page from a stable active-lineage snapshot. Results are oldest-first. * An initial request omits cursors; subsequent requests must carry both returned `tail_id` * and `before_entry_id`. Appends after the pinned tail cannot enter later pages. */export function readReplayPage(path: string, options: ReplayPageOptions = {}): ReplaySnapshot { const limit = boundedLimit(options.limit); let all: SessionEntry[]; try { all = SessionStore.openReadOnly(path).readAll(); } catch { return emptySnapshot(); } const currentTail = latestEntryId(all); const requestedTail = options.tail_id; if (options.before_entry_id !== undefined && requestedTail === undefined) { throw new ReplayCursorError('before_entry_id requires tail_id'); } if (requestedTail !== undefined && !all.some((entry) => entry.id === requestedTail)) { throw new ReplayCursorError('tail_id does not exist'); } const tailId = requestedTail ?? currentTail; const chain = activeChain(all, tailId).reverse(); if (tailId !== undefined && chain.length === 0) throw new ReplayCursorError('tail_id is not readable');
let end = chain.length; if (options.before_entry_id !== undefined) { const index = chain.findIndex((entry) => entry.id === options.before_entry_id); if (index < 0) throw new ReplayCursorError('before_entry_id is not on the pinned lineage'); end = index; } const startByCount = Math.max(0, end - limit); const candidates = chain.slice(startByCount, end); const entries = takeNewestWithinBytes(candidates); const firstIndex = entries.length === 0 ? end : chain.findIndex((entry) => entry.id === entries[0]!.id); const hasOlder = firstIndex > 0; return { entries, ...(tailId !== undefined ? { tail_id: tailId } : {}), ...(entries[0] !== undefined ? { before_entry_id: entries[0].id } : {}), has_older: hasOlder, truncated: hasOlder, };}
/** Compatibility wrapper for the original newest-page API. */export function readReplay(path: string, limit = MAX_REPLAY_ENTRIES): ReplaySnapshot { return readReplayPage(path, { limit });}
function boundedLimit(limit: number | undefined): number { if (limit === undefined || !Number.isFinite(limit)) return MAX_REPLAY_ENTRIES; return Math.max(1, Math.min(Math.floor(limit), MAX_REPLAY_ENTRIES));}
/** Keep the newest suffix under the byte cap, rejecting an unrepresentable row. */function takeNewestWithinBytes(entries: SessionEntry[]): SessionEntry[] { const selected: SessionEntry[] = []; let bytes = 0; for (let index = entries.length - 1; index >= 0; index--) { const entry = entries[index]!; const size = Buffer.byteLength(JSON.stringify(entry), 'utf8'); if (selected.length > 0 && bytes + size > MAX_REPLAY_BYTES) break; if (size > MAX_REPLAY_BYTES) throw new ReplayEntryTooLargeError(entry.id, size); selected.unshift(entry); bytes += size; if (bytes >= MAX_REPLAY_BYTES) break; } return selected;}
function emptySnapshot(): ReplaySnapshot { return { entries: [], has_older: false, truncated: false };}
/** Deterministic per-agent session JSONL path. */export function agentSessionPath(home: string, agentId: string): string { return join(home, 'agents', agentId, 'session.jsonl');}
export interface LiveWindow { messages: Message[]; compactionIndex?: number;}
/** Read the post-compaction message window, tolerating missing/corrupt rows. */export function readLiveWindow(path: string): LiveWindow { if (!existsSync(path)) return { messages: [] }; let raw: string; try { raw = readFileSync(path, 'utf-8'); } catch { return { messages: [] }; } let lastCompactionIndex: number | undefined; const messages: Message[] = []; for (const line of splitJsonl(raw)) { let entry: SessionEntry; try { entry = JSON.parse(line) as SessionEntry; } catch { continue; } if (entry.type === 'compaction') { lastCompactionIndex = (entry as CompactionEntry).details.index; messages.length = 0; } else if (entry.type === 'message') { messages.push((entry as MessageEntry).details.message); } } return { messages, ...(lastCompactionIndex !== undefined ? { compactionIndex: lastCompactionIndex } : {}), };}