diff --git a/packages/server/src/api/types.ts b/packages/server/src/api/types.ts index 31ffed5..852d5a2 100644 --- a/packages/server/src/api/types.ts +++ b/packages/server/src/api/types.ts @@ -661,15 +661,58 @@ export interface MessagesLiveTail { fragments: OmniMessage[]; } +/** + * Pagination envelope of a windowed `GET /messages` (`tailLimit` / `before` requests + * only; the parameterless full read never carries it). A window is a run of whole + * message-bearing units — one unit = one Task in the Web reducer's sense, opened by a + * main-session user prompt — cut so that no pairing (tool_call/output), compaction span + * or steering group ever splits across windows. + */ +export interface MessagesPageInfo { + /** + * Cursor of this window's first unit (`:`): pass it back as + * `before=` to fetch the previous window. Stable across requests and compaction — + * rotation opens a NEW shard and closed shards are immutable. Absent = this window + * reaches the very beginning of the transcript (no older history). + */ + before?: string; + /** + * Outline turns (the Web conversation outline's entry rule) opened BEFORE this + * window: the client offsets its global `第 N 轮` numbering by this, so a partial + * window never mis-numbers. 0 when the window starts at the beginning. + */ + earlierTurns: number; + /** + * Cumulative stats accrued before this window, seeded into the client's stats + * tracker so header chips and per-turn cumulative rows equal a full load: + * finished-Task elapsed, subagent token totals, and the last main-session + * session/context token readings. + */ + prior: { + subagentTokens: number; + elapsedMs: number; + sessionTokens: number; + contextTokens: number; + }; +} + /** Message history: the full messages and events from concatenating all of this Session's Trace files in order (excludes partial_*). */ export interface MessagesResponse { messages: OmniMessage[]; /** * Present only while the Session is running/compacting: the in-progress stream tail * (open streaming fragments + the channel cursor they cover), so a client joining - * mid-stream can render the currently streaming message. Omitted when idle. + * mid-stream can render the currently streaming message. Omitted when idle. On + * windowed requests it rides TAIL pages only — a `before` page is immutable history + * and never carries it. */ live?: MessagesLiveTail; + /** + * Present exactly on windowed requests (`tailLimit` / `before`): `messages` is then + * the requested window (subagent pointers inside it expanded as usual) rather than + * the full transcript. See MessagesPageInfo. + */ + page?: MessagesPageInfo; } // --------------------------------------------------------------------------- diff --git a/packages/server/src/db/database.ts b/packages/server/src/db/database.ts index fc83ca3..0734110 100644 --- a/packages/server/src/db/database.ts +++ b/packages/server/src/db/database.ts @@ -30,6 +30,7 @@ export function openDatabase(dbPath: string): DatabaseSync { ensureColumn(db, "sessions", "client", "TEXT"); ensureColumn(db, "sessions", "has_trace", "INTEGER NOT NULL DEFAULT 0"); ensureColumn(db, "auth_sessions", "via", "TEXT"); + ensureColumn(db, "trace_files", "page_stats", "TEXT"); return db; } diff --git a/packages/server/src/db/repos/trace-index.ts b/packages/server/src/db/repos/trace-index.ts index 76db3b1..cd26944 100644 --- a/packages/server/src/db/repos/trace-index.ts +++ b/packages/server/src/db/repos/trace-index.ts @@ -62,17 +62,22 @@ export class TraceIndexRepo { constructor(private readonly db: DatabaseSync) {} upsertFile(row: TraceFileRow): void { + // page_stats is a derivation of the shard's CONTENT: any observed size change voids + // it (an unchanged size keeps the cached scan — reconcile passes re-upsert rows). this.db .prepare( `INSERT INTO trace_files (project_id, agent_id, session_id, file_index, date, size_bytes) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(project_id, agent_id, session_id, file_index) - DO UPDATE SET date = excluded.date, size_bytes = excluded.size_bytes`, + DO UPDATE SET date = excluded.date, + page_stats = CASE WHEN excluded.size_bytes = trace_files.size_bytes + THEN trace_files.page_stats ELSE NULL END, + size_bytes = excluded.size_bytes`, ) .run(row.projectId, row.agentId, row.sessionId, row.fileIndex, row.date, row.sizeBytes); } - /** Refreshes one shard's last-observed size (listings write back their page stats). */ + /** Refreshes one shard's last-observed size (listings write back their page stats); a changed size voids the cached window scan. */ updateFileSize( projectId: string, agentId: string, @@ -82,10 +87,45 @@ export class TraceIndexRepo { ): void { this.db .prepare( - `UPDATE trace_files SET size_bytes = ? + `UPDATE trace_files SET + page_stats = CASE WHEN size_bytes = ? THEN page_stats ELSE NULL END, + size_bytes = ? WHERE project_id = ? AND agent_id = ? AND session_id = ? AND file_index = ?`, ) - .run(sizeBytes, projectId, agentId, sessionId, fileIndex); + .run(sizeBytes, sizeBytes, projectId, agentId, sessionId, fileIndex); + } + + /** Cached message-window prefix record of one shard (see services/message-window.ts); null = none. */ + getPageStats( + projectId: string, + agentId: string, + sessionId: string, + fileIndex: number, + ): string | null { + const r = this.db + .prepare( + `SELECT page_stats FROM trace_files + WHERE project_id = ? AND agent_id = ? AND session_id = ? AND file_index = ?`, + ) + .get(projectId, agentId, sessionId, fileIndex); + return r ? ((r.page_stats as string | null) ?? null) : null; + } + + /** Stores a shard's message-window prefix record, guarded on the size it was computed for (a concurrent append voids the write). */ + setPageStats( + projectId: string, + agentId: string, + sessionId: string, + fileIndex: number, + sizeBytes: number, + pageStats: string, + ): void { + this.db + .prepare( + `UPDATE trace_files SET page_stats = ? + WHERE project_id = ? AND agent_id = ? AND session_id = ? AND file_index = ? AND size_bytes = ?`, + ) + .run(pageStats, projectId, agentId, sessionId, fileIndex, sizeBytes); } deleteFile(projectId: string, agentId: string, sessionId: string, fileIndex: number): void { diff --git a/packages/server/src/db/schema.ts b/packages/server/src/db/schema.ts index 7b504f3..a1f5412 100644 --- a/packages/server/src/db/schema.ts +++ b/packages/server/src/db/schema.ts @@ -128,6 +128,7 @@ CREATE TABLE IF NOT EXISTS trace_files ( -- DERIVED CACHE of the on-disk T file_index INTEGER NOT NULL, -- shard index NNN of _NNN.jsonl (name/path are reconstructed from the row, never stored: the data root may move) date TEXT NOT NULL, -- date directory name (local yyyy-mm-dd) size_bytes INTEGER NOT NULL, -- last observed size (listings stat-refresh the returned page and write back; an actively-appended shard may lag in between) + page_stats TEXT, -- cached message-window scan state at this shard's END (services/message-window.ts ShardPrefixRecord JSON); immutable shards only, NULL = not yet computed, nulled whenever size_bytes changes PRIMARY KEY (project_id, agent_id, session_id, file_index) ); CREATE INDEX IF NOT EXISTS idx_trace_files_agent_date ON trace_files(project_id, agent_id, date); diff --git a/packages/server/src/http/routes/sessions.ts b/packages/server/src/http/routes/sessions.ts index 435ed2c..8a9c522 100644 --- a/packages/server/src/http/routes/sessions.ts +++ b/packages/server/src/http/routes/sessions.ts @@ -17,6 +17,7 @@ import type { FilesStatResponse, GoalResponse, MessagesLiveTail, + MessagesPageInfo, MessagesResponse, ServerEvent, SessionCategory, @@ -26,6 +27,8 @@ import type { RetryNowResponse, TaskCreateResponse, } from "../../api/types.js"; +import { decodeCursor } from "../../services/message-window.js"; +import type { MessagesPageRequest } from "../../services/trace-service.js"; import { PREVIEW_TOKEN_TTL_MS, resolvePreviewTarget } from "../../services/preview-token.js"; import type { AppEnv } from "../../auth/middleware.js"; import type { SessionRow } from "../../db/repos/sessions.js"; @@ -72,6 +75,51 @@ export const APPROVAL_MODES: readonly ApprovalMode[] = [ /** The five valid per-turn thinking level names (TaskCreateRequest.thinkingLevel). */ const THINKING_LEVELS: readonly ThinkingLevelName[] = ["none", "low", "medium", "high", "xhigh"]; +/** Unit-count bounds for windowed history reads (`tailLimit` / `limit`), and the `before` page's default. */ +const MESSAGES_PAGE_LIMIT_MAX = 1000; +const MESSAGES_PAGE_LIMIT_DEFAULT = 200; + +/** Parse one windowed-read unit-count param (positive integer, capped). */ +function pageLimit(raw: string, name: string): number { + if (!/^\d{1,4}$/.test(raw)) throw badRequest(`${name} must be a positive integer.`); + const v = Number.parseInt(raw, 10); + if (v < 1 || v > MESSAGES_PAGE_LIMIT_MAX) { + throw badRequest(`${name} must be an integer between 1 and ${MESSAGES_PAGE_LIMIT_MAX}.`); + } + return v; +} + +/** + * Parse GET /messages windowed-read params. No params → null: the legacy full-transcript + * read, byte-identical to the pre-pagination response (other consumers depend on it). + * `tailLimit=` → the newest n units; `before=[&limit=]` → the n units + * preceding the cursor. The two forms are mutually exclusive, and `limit` belongs to + * `before` alone — mixing them is a caller bug worth a loud 400 rather than a guess. + */ +function messagesPageQuery(c: Context): MessagesPageRequest | null { + const rawTail = c.req.query("tailLimit"); + const rawBefore = c.req.query("before"); + const rawLimit = c.req.query("limit"); + if (rawTail === undefined && rawBefore === undefined) { + if (rawLimit !== undefined) throw badRequest("limit requires before."); + return null; + } + if (rawTail !== undefined) { + if (rawBefore !== undefined) throw badRequest("tailLimit and before are mutually exclusive."); + if (rawLimit !== undefined) throw badRequest("limit only applies to before requests."); + return { kind: "tail", limit: pageLimit(rawTail, "tailLimit") }; + } + const cursor = decodeCursor(rawBefore!); + if (cursor === null) { + throw badRequest("before must be a cursor of the form :."); + } + return { + kind: "before", + cursor, + limit: rawLimit !== undefined ? pageLimit(rawLimit, "limit") : MESSAGES_PAGE_LIMIT_DEFAULT, + }; +} + /** Accepted `category` query values of the list endpoint (SessionCategory, spelled out for validation). */ const SESSION_CATEGORIES: readonly SessionCategory[] = [ "active", @@ -494,6 +542,7 @@ export function sessionsRoutes(deps: AppDeps): Hono { app.get("/:sessionId/messages", async (c) => { const row = resolveSession(c); + const page = messagesPageQuery(c); // Live tail (running/compacting sessions only): capture the channel cursor and the // open-fragment snapshot together, synchronously — no await between the two, and both // BEFORE the trace read starts. That ordering is what makes the client contract safe @@ -504,13 +553,41 @@ export function sessionsRoutes(deps: AppDeps): Hono { // cursor — the client's overlap dedup against `messages` decides for them — so a // complete message whose trace append is still in flight when the read starts is not // lost either. + // + // Windowed reads keep the same contract on TAIL pages only: a tail page ends at the + // transcript's live edge, so the attachment's semantics are identical. A `before` + // page is immutable history — attaching in-flight fragments to it would seed them at + // the wrong position — so it never carries `live`. let live: MessagesLiveTail | undefined; - if (deps.manager.statusOf(row.sessionId) !== "idle") { + if (page?.kind !== "before" && deps.manager.statusOf(row.sessionId) !== "idle") { live = { cursor: deps.channels.get(row.sessionId).lastEventId, fragments: deps.manager.liveFragments(row.sessionId), }; } + if (page !== null) { + const result = await deps.traceService.readMessagesPage( + row.projectId, + row.agentId, + row.sessionId, + page, + ); + const info: MessagesPageInfo = { + ...(result.before !== undefined ? { before: result.before } : {}), + earlierTurns: result.prior.turns, + prior: { + subagentTokens: result.prior.subagentTokens, + elapsedMs: result.prior.elapsedMs, + sessionTokens: result.prior.sessionTokens, + contextTokens: result.prior.contextTokens, + }, + }; + return c.json({ + messages: result.messages, + ...(live !== undefined ? { live } : {}), + page: info, + } satisfies MessagesResponse); + } const messages = await deps.traceService.readMessages( row.projectId, row.agentId, diff --git a/packages/server/src/services/message-window.ts b/packages/server/src/services/message-window.ts new file mode 100644 index 0000000..e255119 --- /dev/null +++ b/packages/server/src/services/message-window.ts @@ -0,0 +1,430 @@ +/** + * Message-window scanner: the pure logic behind cursor pagination of + * `GET /api/sessions/:id/messages` (TraceService.readMessagesPage). + * + * A window **unit** is one Task in the Web reducer's sense: it opens at a main-session + * user prompt (text or image) and runs until the next such prompt. Cutting only at these + * boundaries is what keeps every stream-model invariant intact inside a window — a + * tool_call is never separated from its tool_call_output (outputs land before the next + * user prompt), a compaction_begin/end span never splits (compaction-internal messages + * stay skippable), a steering chip keeps the images that ride behind it, and a + * mid-Task shard rotation (auto compaction with carry-over) stays glued to its Task. + * + * The scanner walks one shard's RAW messages (no subagent expansion — origin-carrying + * messages never appear in a shard) and mirrors, decision for decision, the Web reducer + * in web/src/lib/omni/stream-model.ts (pushMessage/startTask/finalizeOpenTask) plus the + * outline-entry rule in web/src/features/chat/outline-model.ts (buildOutline) — the four + * implementations of "what is one turn" (this file, stream-model, outline-model, and + * trace-service.analyze) must stay in step; tests on both sides pin the shared cases. + * + * Two distinct counts come out of one pass: + * - **unit boundaries** — the safe cut points described above (banner-only prompts and + * goal-round re-sends included: they start Tasks in the reducer); + * - **outline turns** — the subset that opens an entry in the Web's conversation + * outline (`第 N 轮` numbering). Machine-only prompts (handoff / model-switch + * blocks), goal rounds past 1, steering chips and compaction injections do NOT open + * entries, and consecutive user messages of one send merge into one entry — the + * count must match buildOutline exactly, or a paginated outline would mis-number. + * + * The pass also accumulates the **prior stats** a partial window needs seeded into the + * Web's stats tracker so header chips and per-turn cumulative rows keep telling the + * truth (see task-stats.ts `seedPriorStats`): elapsed time of finished Tasks, subagent + * token totals (via the caller-provided child expander), and the last main-session + * session/context token readings. + * + * Scan state is carried across shards (a Task can span a rotation). Cached per-shard + * prefix records (trace_files.page_stats) persist the carry so old shards are read at + * most once ever; bump CACHE_VERSION whenever any rule in this file changes. + */ +import { parseUserSteeringText } from "@prismshadow/penguin-core"; +import { + parseGoalMessage, + parseHandoffMessage, + parseModelSwitchMessage, +} from "@prismshadow/penguin-core/markers"; +import type { OmniMessage } from "@prismshadow/penguin-core"; + +/** Bump when any counting/boundary rule changes: cached page_stats records with an older version are recomputed. */ +export const CACHE_VERSION = 1; + +/** Cumulative totals at a point in the trace (all values are "before this point"). */ +export interface WindowPriorStats { + /** Outline entries opened before this point (the Web outline's global numbering offset). */ + turns: number; + /** Sum of subagent token_usage request totals (all descendant sessions) before this point. */ + subagentTokens: number; + /** Sum of finished Tasks' elapsed time before this point (the Web's sessionElapsedMs basis). */ + elapsedMs: number; + /** The last main-session token_usage `session.total` seen before this point (0 = none). */ + sessionTokens: number; + /** The last main-session NON-compaction token_usage `request.total` before this point (context occupancy basis). */ + contextTokens: number; +} + +/** Open-Task bookkeeping carried across messages (mirrors StreamModel.task* fields). */ +interface TaskCarry { + open: boolean; + /** Task first/latest message timestamps (ms; the degenerate-round elapsed fallback). */ + firstTsMs: number; + lastTsMs: number; + /** Last non-compaction request_end (ms); the Task's true end when present. */ + lastReqEndMs: number | null; + /** Whether this Task saw main usage / non-blank assistant text — decides whether a stats row would be emitted at its close (which breaks a user-item run). */ + sawUsage: boolean; + sawReply: boolean; + /** Compaction usage held pending (committed to the Task by a later non-compaction request_end — mirrors task-stats.ts pendingCompaction*). */ + pendingCompactionUsage: boolean; +} + +/** Full scanner state, serializable into a shard's cached prefix record. */ +export interface ScanState { + totals: WindowPriorStats; + task: TaskCarry; + /** Between a main-session compaction_begin and its compaction_end. */ + compactionActive: boolean; + /** A steering chip is still collecting its trailing images (mirrors StreamModel.openSteering). */ + steeringOpen: boolean; + /** + * Inside an unbroken, entry-carrying run of user items — buildOutline's `lastWasUser`, + * tracked at the item level: true after an entry-eligible user item, false after any + * non-user item (a stats row included) AND after banner/goal-round texts (which open no + * entry and break the run in buildOutline). Both the window-cut and the merge decision + * key off it, so a cut can never split what the outline merges. + */ + runOpen: boolean; +} + +export function initialScanState(): ScanState { + return { + totals: { turns: 0, subagentTokens: 0, elapsedMs: 0, sessionTokens: 0, contextTokens: 0 }, + task: { + open: false, + firstTsMs: 0, + lastTsMs: 0, + lastReqEndMs: null, + sawUsage: false, + sawReply: false, + pendingCompactionUsage: false, + }, + compactionActive: false, + steeringOpen: false, + runOpen: false, + }; +} + +/** One safe cut point: `ordinal` indexes into the shard's parsed message array; `stats` is the cumulative prior AT the cut (the unit itself not included). */ +export interface UnitBoundary { + ordinal: number; + stats: WindowPriorStats; +} + +/** + * Aggregate a subagent pointer's child subtree without materializing its messages: + * the token sum feeds `subagentTokens`, the max timestamp advances the open Task's + * lastTs exactly as routeNested's touchTask would for every expanded child message. + * Null = child trace missing (the pointer event stays a plain event, contributing nothing). + */ +export interface ChildAggregate { + requestTokens: number; + maxTsMs: number | null; +} + +function tsMs(timestamp: string): number | null { + const ms = Date.parse(timestamp); + return Number.isFinite(ms) ? ms : null; +} + +/** Mirrors stream-model touchTask: advance the open Task's latest timestamp (compaction-internal messages excluded). */ +function touchTask(state: ScanState, ms: number | null): void { + if (!state.task.open || state.compactionActive || ms === null) return; + if (ms > state.task.lastTsMs) state.task.lastTsMs = ms; +} + +/** Mirrors stream-model finalizeOpenTask + task-stats endTask: settle the open Task's elapsed into the cumulative. */ +function finalizeTask(state: ScanState): void { + const t = state.task; + if (!t.open) return; + t.open = false; + const endMs = t.lastReqEndMs ?? t.lastTsMs; + state.totals.elapsedMs += Math.max(0, endMs - t.firstTsMs); + t.pendingCompactionUsage = false; +} + +/** Mirrors stream-model startTask: finalize the previous Task and open a new one at `ms`. */ +function startTask(state: ScanState, ms: number | null): void { + finalizeTask(state); + const t = state.task; + t.open = true; + t.firstTsMs = ms ?? 0; + t.lastTsMs = t.firstTsMs; + t.lastReqEndMs = null; + t.sawUsage = false; + t.sawReply = false; + t.pendingCompactionUsage = false; +} + +/** A non-user item entered the stream: the user-item run breaks (buildOutline's lastWasUser = false). */ +function breakRuns(state: ScanState): void { + state.runOpen = false; +} + +/** + * Handle one Task-starting user message (prompt text or non-steering image). + * `entryEligible` says whether the message would open an outline entry (buildOutline's + * rule); ineligible messages (banner blocks, goal rounds > 1) still start Tasks — and + * still cut when the run is broken — but never count and always break the run, exactly + * as buildOutline resets lastWasUser for them. + */ +function onTaskStart( + state: ScanState, + ms: number | null, + entryEligible: boolean, + onBoundary: (stats: WindowPriorStats) => void, +): void { + // A stats row emitted while closing the previous Task lands BEFORE this user item and + // breaks the item run (finalizeOpenTask inserts it ahead of the new prompt) — mirror + // that so two sends merged by the outline are never cut apart, while two sends + // separated by a stats row cut (and count) as two. + if (state.task.open && (state.task.sawUsage || state.task.sawReply)) breakRuns(state); + const boundary = !state.runOpen; + startTask(state, ms); + if (boundary) onBoundary({ ...state.totals }); + if (entryEligible) { + if (boundary) state.totals.turns += 1; + state.runOpen = true; + } else { + state.runOpen = false; + } +} + +/** buildOutline's eligibility for a user prompt TEXT: machine-only source blocks and goal rounds past 1 open no entry. */ +function outlineEligibleText(text: string): boolean { + if (parseHandoffMessage(text) || parseModelSwitchMessage(text)) return false; + const goal = parseGoalMessage(text); + return !(goal !== null && goal.round > 1); +} + +/** + * Scan one shard's raw messages, mutating `state` in place and reporting every unit + * boundary through `onBoundary` (with the cumulative priors AT the cut). `expandChild` + * resolves a subagent pointer's aggregate (may hit a per-request memo); pass null to + * skip child reads when the caller does not need subagent totals for this span. + */ +export async function scanMessages( + state: ScanState, + messages: readonly OmniMessage[], + onBoundary: (ordinal: number, stats: WindowPriorStats) => void, + expandChild: ((sessionId: string) => Promise) | null, + fromOrdinal = 0, + toOrdinal = messages.length, +): Promise { + for (let i = fromOrdinal; i < toOrdinal; i++) { + const msg = messages[i]!; + // Shards never contain origin-carrying messages (core's Writer filters them); + // defensively skip any that appear rather than mis-shaping the counts. + if (msg.origin !== undefined && msg.origin.length > 0) continue; + const ms = tsMs(msg.timestamp); + const p = msg.payload as Record & { type?: string; role?: string }; + + // Mirrors pushMessage's first step: anything that is not a complete user image + // closes the steering-images window (session_meta and events included). + const isUserImage = msg.type === "model_msg" && p.type === "image_url"; + if (!isUserImage) state.steeringOpen = false; + + if (msg.type === "model_msg") { + // Compaction-internal model messages: never rendered, never counted (stream-model + // returns before any item/Task logic; only touchTask advances). + if (state.compactionActive) { + touchTask(state, ms); + continue; + } + if (p.type === "text" && p.role === "user" && typeof p.text === "string") { + const text = p.text; + // Compaction-summary injection: no item, no Task, no run change. + if (text.startsWith("[context_summary]") || text.startsWith("")) { + touchTask(state, ms); + continue; + } + // Mid-run steering: rendered as a user_steering item (not user_text) — it breaks + // the outline's user-item run but never starts a Task or cuts a window. + if (parseUserSteeringText(text) !== null) { + touchTask(state, ms); + breakRuns(state); + state.steeringOpen = true; + continue; + } + onTaskStart(state, ms, outlineEligibleText(text), (stats) => onBoundary(i, stats)); + continue; + } + if (p.type === "image_url") { + // An image riding behind a steering chip folds into it: no item, no Task. + if (state.steeringOpen) { + touchTask(state, ms); + continue; + } + onTaskStart(state, ms, true, (stats) => onBoundary(i, stats)); + continue; + } + if (p.type === "text" && p.role === "assistant" && typeof p.text === "string") { + touchTask(state, ms); + // Blank fidelity-only messages produce no item (stream-model discards them). + if (p.text.trim() !== "") { + state.task.sawReply = true; + breakRuns(state); + } + continue; + } + if (p.type === "thinking") { + touchTask(state, ms); + if (typeof p.thinking === "string" && p.thinking.trim() !== "") breakRuns(state); + continue; + } + if (p.type === "tool_call") { + touchTask(state, ms); + breakRuns(state); + continue; + } + // tool_call_output updates an existing card (no new item); inline_* render nothing. + touchTask(state, ms); + continue; + } + + if (msg.type === "event_msg") { + touchTask(state, ms); + const t = p.type; + if (t === "compaction_begin") { + state.compactionActive = true; + breakRuns(state); // the compaction banner is an item + continue; + } + if (t === "compaction_end") { + // Closes the banner opened by the begin (no new item mid-window). + state.compactionActive = false; + continue; + } + if (t === "abort") { + breakRuns(state); // the interruption marker is an item + continue; + } + if (t === "request_end") { + if (state.compactionActive) continue; // compaction requests render nothing + if (ms !== null) state.task.lastReqEndMs = ms; + // Pending compaction usage commits at a later non-compaction request_end + // (compaction mid-Task) — from then on the Task WILL show a stats row. + if (state.task.pendingCompactionUsage) { + state.task.sawUsage = true; + state.task.pendingCompactionUsage = false; + } + const status = p.status; + // A retryable end renders a reconnect-hint item. + if (status === "failed" || status === "timeout" || status === "malformed") { + breakRuns(state); + } + continue; + } + if (t === "token_usage") { + const request = p.request as { total?: number } | undefined; + const session = p.session as { total?: number } | undefined; + // The session cumulative tracks the provider even during compaction + // (task-stats trackMainUsage does the same). + if (typeof session?.total === "number") state.totals.sessionTokens = session.total; + if (state.compactionActive) { + state.task.pendingCompactionUsage = true; + } else { + if (typeof request?.total === "number") state.totals.contextTokens = request.total; + state.task.sawUsage = true; + } + continue; + } + if (t === "subagent" && typeof p.session_id === "string" && expandChild !== null) { + // The pointer expands to the child's messages in the served transcript: their + // token_usage feeds the parent's subagent totals at every depth, and their + // timestamps advance the open Task exactly as routeNested's touchTask would. + const agg = await expandChild(p.session_id); + if (agg !== null) { + state.totals.subagentTokens += agg.requestTokens; + touchTask(state, agg.maxTsMs); + } + continue; + } + // approval_decision / request_begin / unexpanded pointers: no items, no run change. + continue; + } + // session_meta: no item (steering window already closed above). + } +} + +/** Finalize the trailing open Task (end of the whole trace): the cumulative then covers every finished Task. */ +export function finalizeScan(state: ScanState): void { + finalizeTask(state); +} + +// --------------------------------------------------------------------------- +// Cursor encoding +// --------------------------------------------------------------------------- + +/** A window cursor: shard file index + message ordinal within that shard's parsed array. */ +export interface MessageCursor { + fileIndex: number; + ordinal: number; +} + +/** + * `:` — stable across requests and compaction: rotation always + * opens a NEW shard, closed shards are immutable, and the active shard is append-only, + * so a (shard, ordinal) pair never moves. + */ +export function encodeCursor(c: MessageCursor): string { + return `${c.fileIndex}:${c.ordinal}`; +} + +/** Strict parse of a cursor string; null when malformed (callers turn that into a 400). */ +export function decodeCursor(raw: string): MessageCursor | null { + const m = /^(\d{1,9}):(\d{1,9})$/.exec(raw); + if (!m) return null; + return { fileIndex: Number(m[1]), ordinal: Number(m[2]) }; +} + +// --------------------------------------------------------------------------- +// Cached per-shard prefix records +// --------------------------------------------------------------------------- + +/** + * The persisted shape of trace_files.page_stats: the scan state at the END of a shard + * (cumulative from the very beginning of the session). Only immutable shards are cached + * — the newest shard still grows. `v` gates rule evolution: a record from an older + * CACHE_VERSION is recomputed as if absent. + */ +export interface ShardPrefixRecord { + v: number; + state: ScanState; +} + +export function serializePrefix(state: ScanState): string { + return JSON.stringify({ v: CACHE_VERSION, state } satisfies ShardPrefixRecord); +} + +/** Parse a cached record; null when absent, unparseable, or from another CACHE_VERSION. */ +export function deserializePrefix(raw: string | null): ScanState | null { + if (raw === null) return null; + try { + const rec = JSON.parse(raw) as ShardPrefixRecord; + if (rec.v !== CACHE_VERSION || typeof rec.state !== "object" || rec.state === null) { + return null; + } + return rec.state; + } catch { + return null; + } +} + +/** Deep-copy a scan state (cached records must not be mutated by a later scan). */ +export function cloneScanState(state: ScanState): ScanState { + return { + totals: { ...state.totals }, + task: { ...state.task }, + compactionActive: state.compactionActive, + steeringOpen: state.steeringOpen, + runOpen: state.runOpen, + }; +} diff --git a/packages/server/src/services/trace-service.ts b/packages/server/src/services/trace-service.ts index fa446ac..8125f64 100644 --- a/packages/server/src/services/trace-service.ts +++ b/packages/server/src/services/trace-service.ts @@ -40,6 +40,21 @@ import type { TraceFileRow, TraceSessionRow } from "../db/repos/trace-index.js"; import { HttpError } from "../http/errors.js"; import { formatLocalDate } from "../internal/dates.js"; import type { SessionSources } from "../runtime/session-sources.js"; +import { + cloneScanState, + deserializePrefix, + encodeCursor, + finalizeScan, + initialScanState, + scanMessages, + serializePrefix, +} from "./message-window.js"; +import type { + ChildAggregate, + MessageCursor, + ScanState, + WindowPriorStats, +} from "./message-window.js"; import { TraceIndexService, traceFilePath } from "./trace-index.js"; const TRACE_FILE_RE = /^(.+)_(\d{3})\.jsonl$/; @@ -58,6 +73,34 @@ interface LocatedFile { path: string; date: string; index: number; + /** Last observed size from the index row (the page-stats cache write guard). */ + sizeBytes: number; +} + +/** A windowed history read's request shape (see readMessagesPage). */ +export type MessagesPageRequest = + { kind: "tail"; limit: number } | { kind: "before"; cursor: MessageCursor; limit: number }; + +/** A windowed history read's result (route maps it onto MessagesResponse.page). */ +export interface MessagesPageResult { + messages: OmniMessage[]; + /** Cursor of the window's first unit; absent = the window reaches the beginning. */ + before?: string; + /** Cumulative stats before the window (earlierTurns = prior.turns). */ + prior: WindowPriorStats; +} + +/** + * Per-request expansion context: `projectScanned` caps the miss-path force reconcile at + * one per request; `ancestry` guards cycles; `raw` memoizes each child session's + * concatenated shard messages so the window expansion and the subagent-token + * aggregation never read the same child twice within one request. + */ +interface ExpandCtx { + projectScanned: boolean; + ancestry: Set; + depth: number; + raw: Map; } /** @@ -92,6 +135,12 @@ export interface TraceServiceDeps { sessions?: TraceSessionIndex; /** Optional (narrow tests may omit): the shared in-process Session-origin registry (single source of truth for `source`). */ sources?: SessionSources; + /** + * Test observability: called with the path of every Trace shard this service reads + * from disk (windowed-read tests assert an old-window request never touches the + * newest shard and vice versa). Production wiring omits it. + */ + observeShardRead?: (path: string) => void; } /** One Session's classification result (see classify): its sidebar category + Workspace path ("" = unknown). */ @@ -127,9 +176,16 @@ export class TraceService { path: traceFilePath(this.root, r), date: r.date, index: r.fileIndex, + sizeBytes: r.sizeBytes, })); } + /** All shard reads funnel through here (deps.observeShardRead is the windowed-read tests' proof of which files were touched). */ + private async readShard(path: string): Promise { + this.deps.observeShardRead?.(path); + return readTraceTolerant(path); + } + /** Deletes all of this Session's Trace files (called when the Session is deleted); the index rows go with them. */ async deleteSessionTraces(projectId: string, agentId: string, sessionId: string): Promise { const files = await this.locateAll(projectId, agentId, sessionId); @@ -160,11 +216,18 @@ export class TraceService { agentId: string, sessionId: string, ): Promise { - return this.readMessagesExpanded(projectId, agentId, sessionId, { + const ctx: ExpandCtx = { projectScanned: false, ancestry: new Set([sessionId]), depth: 0, - }); + raw: new Map(), + }; + const files = await this.locateAll(projectId, agentId, sessionId); + const out: OmniMessage[] = []; + for (const file of files) { + out.push(...(await this.expandMessages(projectId, await this.readShard(file.path), ctx))); + } + return out; } /** @@ -186,42 +249,247 @@ export class TraceService { return this.deps.index.repo.findAgentBySession(projectId, childSid); } - private async readMessagesExpanded( + /** A child session's concatenated raw shard messages, memoized per request; null = no Trace located. */ + private async readChildRaw( + projectId: string, + childSid: string, + ctx: ExpandCtx, + ): Promise { + const cached = ctx.raw.get(childSid); + if (cached !== undefined) return cached; + const childAgent = await this.resolveChildAgent(projectId, childSid, ctx); + let raw: OmniMessage[] | null = null; + if (childAgent) { + const files = await this.locateAll(projectId, childAgent, childSid); + if (files.length > 0) { + raw = []; + for (const file of files) raw.push(...(await this.readShard(file.path))); + } + } + ctx.raw.set(childSid, raw); + return raw; + } + + /** + * Expand subagent pointers within one message span (the whole transcript on the full + * path, one window on the paged path — same recursive shape either way). + */ + private async expandMessages( + projectId: string, + messages: readonly OmniMessage[], + ctx: ExpandCtx, + ): Promise { + const out: OmniMessage[] = []; + for (const msg of messages) { + // The depth cap guards against runaway recursion; ancestry guards against a + // cyclic pointer (a tampered Trace pointing to itself/an ancestor is not expanded). + const childSid = ctx.depth < MAX_SUBAGENT_DEPTH ? subagentPointer(msg) : null; + if (!childSid || ctx.ancestry.has(childSid)) { + out.push(msg); + continue; + } + const raw = await this.readChildRaw(projectId, childSid, ctx); + let nested: OmniMessage[] = []; + if (raw !== null) { + ctx.ancestry.add(childSid); + nested = await this.expandMessages(projectId, raw, { ...ctx, depth: ctx.depth + 1 }); + ctx.ancestry.delete(childSid); + } + // Child Trace missing (deleted): keep the pointer event, since the sub-session's content can't be recovered. + if (nested.length === 0) { + out.push(msg); + continue; + } + for (const m of nested) out.push({ ...m, origin: [childSid, ...(m.origin ?? [])] }); + } + return out; + } + + /** + * A subagent pointer's subtree aggregate for the window scanner (message-window.ts): + * descendant token_usage request totals plus the subtree's max timestamp — the same + * contributions the expanded messages make to the Web's parent-level stats tracker. + * The recursion mirrors expandMessages' depth/ancestry rules exactly, so an + * unexpandable pointer contributes nothing on both paths. + */ + private async aggregateChild( + projectId: string, + childSid: string, + ctx: ExpandCtx, + ): Promise { + if (ctx.depth >= MAX_SUBAGENT_DEPTH || ctx.ancestry.has(childSid)) return null; + const raw = await this.readChildRaw(projectId, childSid, ctx); + if (raw === null || raw.length === 0) return null; + let requestTokens = 0; + let maxTsMs: number | null = null; + ctx.ancestry.add(childSid); + for (const msg of raw) { + const ms = Date.parse(msg.timestamp); + if (Number.isFinite(ms) && (maxTsMs === null || ms > maxTsMs)) maxTsMs = ms; + const p = msg.payload as { type?: string; request?: { total?: number } }; + if (msg.type === "event_msg" && p.type === "token_usage") { + requestTokens += typeof p.request?.total === "number" ? p.request.total : 0; + continue; + } + const grandchild = subagentPointer(msg); + if (grandchild !== null) { + const agg = await this.aggregateChild(projectId, grandchild, { + ...ctx, + depth: ctx.depth + 1, + }); + if (agg !== null) { + requestTokens += agg.requestTokens; + if (agg.maxTsMs !== null && (maxTsMs === null || agg.maxTsMs > maxTsMs)) { + maxTsMs = agg.maxTsMs; + } + } + } + } + ctx.ancestry.delete(childSid); + return { requestTokens, maxTsMs }; + } + + /** + * End-of-shard scan states for files[0..upto] (cumulative from the transcript's + * beginning), served from the trace_files.page_stats cache. A missing/stale record + * costs one read of THAT shard — once ever: every shard here is immutable (rotation + * opens a new shard; the newest shard never appears in a prefix, because windows are + * suffixes and their first shard is always read anyway). The write-back is guarded on + * the indexed size, so an externally rewritten shard can only invalidate, never + * poison, the cache. + */ + private async prefixStates( projectId: string, agentId: string, sessionId: string, - ctx: { projectScanned: boolean; ancestry: Set; depth: number }, - ): Promise { - const files = await this.locateAll(projectId, agentId, sessionId); - const out: OmniMessage[] = []; - for (const file of files) { - for (const msg of await readTraceTolerant(file.path)) { - // The depth cap guards against runaway recursion; ancestry guards against a - // cyclic pointer (a tampered Trace pointing to itself/an ancestor is not expanded). - const childSid = ctx.depth < MAX_SUBAGENT_DEPTH ? subagentPointer(msg) : null; - if (!childSid || ctx.ancestry.has(childSid)) { - out.push(msg); - continue; - } - const childAgent = await this.resolveChildAgent(projectId, childSid, ctx); - let nested: OmniMessage[] = []; - if (childAgent) { - ctx.ancestry.add(childSid); - nested = await this.readMessagesExpanded(projectId, childAgent, childSid, { - ...ctx, - depth: ctx.depth + 1, - }); - ctx.ancestry.delete(childSid); - } - // Child Trace missing (deleted): keep the pointer event, since the sub-session's content can't be recovered. - if (nested.length === 0) { - out.push(msg); - continue; - } - for (const m of nested) out.push({ ...m, origin: [childSid, ...(m.origin ?? [])] }); + files: LocatedFile[], + upto: number, + ctx: ExpandCtx, + ): Promise { + const states: ScanState[] = []; + for (let j = 0; j <= upto; j++) { + const file = files[j]!; + const cached = deserializePrefix( + this.deps.index.repo.getPageStats(projectId, agentId, sessionId, file.index), + ); + if (cached !== null) { + states.push(cached); + continue; } + const state = cloneScanState(j === 0 ? initialScanState() : states[j - 1]!); + const messages = await this.readShard(file.path); + await scanMessages( + state, + messages, + () => {}, + (sid) => this.aggregateChild(projectId, sid, ctx), + ); + this.deps.index.repo.setPageStats( + projectId, + agentId, + sessionId, + file.index, + file.sizeBytes, + serializePrefix(state), + ); + states.push(state); } - return out; + return states; + } + + /** + * Windowed history read (the `tailLimit` / `before` forms of GET /messages). The + * window is a run of whole units — cut points and unit semantics live in + * message-window.ts — assembled by reading ONLY the shards the window overlaps + * (plus, once ever per old shard, the prefix-cache backfill above). Subagent + * pointers are expanded exactly as the full path expands them, but only within the + * window: children referenced by older windows load when those windows do. + */ + async readMessagesPage( + projectId: string, + agentId: string, + sessionId: string, + req: MessagesPageRequest, + ): Promise { + const files = await this.locateAll(projectId, agentId, sessionId); + const empty = (): MessagesPageResult => ({ + messages: [], + prior: initialScanState().totals, + }); + if (files.length === 0) return empty(); + const ctx: ExpandCtx = { + projectScanned: false, + ancestry: new Set([sessionId]), + depth: 0, + raw: new Map(), + }; + + // The window's exclusive end: the newest shard's end (tail), or the cursor (before). + let endPos: number; + let endOrdinal: number | null = null; // null = to the shard's end + if (req.kind === "tail") { + endPos = files.length - 1; + } else { + endPos = files.findIndex((f) => f.index === req.cursor.fileIndex); + // Cursor shard no longer on disk (external deletion — locateAll already + // force-retried): the history it pointed into is gone; report end-of-history + // rather than a guess at what used to precede it. + if (endPos < 0) return empty(); + endOrdinal = req.cursor.ordinal; + } + + const prefixes = await this.prefixStates(projectId, agentId, sessionId, files, endPos - 1, ctx); + + // Walk backward from the end, scanning whole shards (each from its cached carry-in) + // until the window has more units than requested or the beginning is reached. + const shardMessages = new Map(); + let boundaries: Array<{ pos: number; ordinal: number; stats: WindowPriorStats }> = []; + let startPos = endPos + 1; + while (boundaries.length <= req.limit && startPos > 0) { + startPos -= 1; + const messages = await this.readShard(files[startPos]!.path); + shardMessages.set(startPos, messages); + const state = cloneScanState(startPos === 0 ? initialScanState() : prefixes[startPos - 1]!); + const shardBoundaries: Array<{ pos: number; ordinal: number; stats: WindowPriorStats }> = []; + const to = startPos === endPos && endOrdinal !== null ? endOrdinal : messages.length; + await scanMessages( + state, + messages, + (ordinal, stats) => shardBoundaries.push({ pos: startPos, ordinal, stats }), + (sid) => this.aggregateChild(projectId, sid, ctx), + 0, + Math.min(to, messages.length), + ); + boundaries = [...shardBoundaries, ...boundaries]; + } + + // Window start: the last `limit` units, or the very beginning (preamble included) + // when the whole remaining history fits — then there is no `before` cursor. + let start: { pos: number; ordinal: number }; + let before: string | undefined; + let prior: WindowPriorStats; + if (boundaries.length > req.limit) { + const wb = boundaries[boundaries.length - req.limit]!; + start = { pos: wb.pos, ordinal: wb.ordinal }; + before = encodeCursor({ fileIndex: files[wb.pos]!.index, ordinal: wb.ordinal }); + prior = wb.stats; + } else { + start = { pos: 0, ordinal: 0 }; + prior = initialScanState().totals; + } + + const windowRaw: OmniMessage[] = []; + for (let pos = start.pos; pos <= endPos; pos++) { + const messages = shardMessages.get(pos) ?? (await this.readShard(files[pos]!.path)); + const from = pos === start.pos ? start.ordinal : 0; + const to = + pos === endPos && endOrdinal !== null + ? Math.min(endOrdinal, messages.length) + : messages.length; + for (let i = from; i < to; i++) windowRaw.push(messages[i]!); + } + const expanded = await this.expandMessages(projectId, windowRaw, ctx); + return { messages: expanded, ...(before !== undefined ? { before } : {}), prior }; } /** List of Trace files (index / date / size / mtime). */ diff --git a/packages/server/test/helpers.ts b/packages/server/test/helpers.ts index 3155d42..12e092b 100644 --- a/packages/server/test/helpers.ts +++ b/packages/server/test/helpers.ts @@ -164,17 +164,21 @@ export function makeTraceHarness( traceIndex: TraceIndexService; service: TraceService; sources: SessionSources; + /** Paths of every Trace shard the service read from disk (windowed-read IO assertions); reset freely between calls. */ + shardReads: string[]; close: () => void; } { const db = openDatabase(":memory:"); const sources = opts.sources ?? new SessionSources(); const traceIndex = new TraceIndexService(root, new TraceIndexRepo(db), sources); + const shardReads: string[] = []; const service = new TraceService(root, { index: traceIndex, ...(opts.sessions !== undefined ? { sessions: opts.sessions } : {}), sources, + observeShardRead: (p) => shardReads.push(p), }); - return { traceIndex, service, sources, close: () => db.close() }; + return { traceIndex, service, sources, shardReads, close: () => db.close() }; } /** Writes a Trace JSONL file directly (for building historical / discovery scenarios). */ diff --git a/packages/server/test/messages-live.test.ts b/packages/server/test/messages-live.test.ts index 05ed2b8..6ef8633 100644 --- a/packages/server/test/messages-live.test.ts +++ b/packages/server/test/messages-live.test.ts @@ -85,8 +85,8 @@ describe("messages live tail", () => { await t.cleanup(); }); - const getMessages = async (): Promise => { - const res = await apiClient(t.app, cookie).get(`/api/sessions/${SID}/messages`); + const getMessages = async (query = ""): Promise => { + const res = await apiClient(t.app, cookie).get(`/api/sessions/${SID}/messages${query}`); expect(res.status).toBe(200); return (await res.json()) as MessagesResponse; }; @@ -134,4 +134,58 @@ describe("messages live tail", () => { const after = await getMessages(); expect(after.live).toBeUndefined(); }); + + it("windowed reads: `live` rides TAIL pages with identical semantics and never rides `before` pages", async () => { + t.deps.manager.adopt(row, midStreamFakeSession(SID)); + await t.deps.manager.startTask(SID, [userText("go")]); + await waitFor(() => t.deps.manager.pendingApprovalCount(SID) === 1); + + const full = await getMessages(); + const tail = await getMessages("?tailLimit=5"); + expect(tail.page).toBeDefined(); + expect(tail.live).toBeDefined(); + // Same capture as the full read: cursor shape and open-fragment snapshot. + expect(tail.live!.cursor).toMatch(/^[0-9a-f]{8}-\d+$/); + expect(tail.live!.fragments.map((f) => (f.payload as { type: string }).type)).toEqual( + full.live!.fragments.map((f) => (f.payload as { type: string }).type), + ); + + // A `before` page is immutable history: never a live attachment, even mid-run. + const before = await getMessages("?before=1:0&limit=5"); + expect(before.page).toBeDefined(); + expect(before.live).toBeUndefined(); + + expect(t.deps.manager.decideApproval(SID, "tc-lv", "deny")).toBe(true); + await waitFor(() => t.deps.manager.statusOf(SID) === "idle"); + }); + + it("the parameterless read stays byte-identical to the pre-pagination contract (no page envelope)", async () => { + t.deps.manager.adopt(row, midStreamFakeSession(SID)); + await t.deps.manager.startTask(SID, [userText("go")]); + await waitFor(() => t.deps.manager.pendingApprovalCount(SID) === 1); + expect(t.deps.manager.decideApproval(SID, "tc-lv", "deny")).toBe(true); + await waitFor(() => t.deps.manager.statusOf(SID) === "idle"); + + const res = await apiClient(t.app, cookie).get(`/api/sessions/${SID}/messages`); + expect(res.status).toBe(200); + const raw = await res.text(); + const body = JSON.parse(raw) as MessagesResponse; + expect(Object.keys(body)).toEqual(["messages"]); + // Exactly the legacy serialization: the full transcript and nothing else. + const expected = JSON.stringify({ + messages: await t.deps.traceService.readMessages(row.projectId, row.agentId, row.sessionId), + }); + expect(raw).toBe(expected); + // Windowed-param validation stays loud rather than guessy. + const bad = await apiClient(t.app, cookie).get( + `/api/sessions/${SID}/messages?tailLimit=5&before=1:0`, + ); + expect(bad.status).toBe(400); + const badLimit = await apiClient(t.app, cookie).get(`/api/sessions/${SID}/messages?limit=5`); + expect(badLimit.status).toBe(400); + const badCursor = await apiClient(t.app, cookie).get( + `/api/sessions/${SID}/messages?before=xyz`, + ); + expect(badCursor.status).toBe(400); + }); }); diff --git a/packages/server/test/messages-page.test.ts b/packages/server/test/messages-page.test.ts new file mode 100644 index 0000000..15fb6df --- /dev/null +++ b/packages/server/test/messages-page.test.ts @@ -0,0 +1,454 @@ +/** + * Windowed history reads (TraceService.readMessagesPage + GET /messages paging params). + * + * Covers: tail/before window slicing mid-shard and across shard boundaries, cursor + * round-trips, pairing-safe cut points (a window never separates a tool_call from its + * output, a compaction span, or a steering group), subagent expansion happening only + * for pointers inside the window, outline-turn counting (`earlierTurns`) staying in + * step with the Web's buildOutline rule (its twin fixtures live in + * web/test/outline-model.test.ts and stream-model.test.ts — keep the two sides + * agreeing), prior stats (elapsed / subagent tokens / session tokens / context), the + * shard-read discipline (an old-window request never reads the newest shard and vice + * versa, once the per-shard prefix cache is primed), and the no-params full read + * staying byte-identical with no `page` envelope. + */ +import fs from "node:fs/promises"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { + assistantText, + compactionBegin, + compactionEnd, + imageUrlMessage, + requestBegin, + requestEnd, + sessionMeta, + subagentEvent, + tokenUsage, + toolCall, + toolCallOutput, + userText, +} from "@prismshadow/penguin-core"; +import { buildHandoffMessage } from "@prismshadow/penguin-core/markers"; +import type { OmniMessage, SessionMetaPayload, TokenCounts } from "@prismshadow/penguin-core"; +import { decodeCursor, encodeCursor } from "../src/services/message-window.js"; +import type { TraceService } from "../src/services/trace-service.js"; +import { makeTempRoot, makeTraceHarness, writeTraceFile } from "./helpers.js"; + +const P = "project-w"; +const A = "agent-w"; +const S = "session-2026-07-20-10-00-00-aabb0001"; +const CHILD = "session-2026-07-20-10-00-05-ccdd0002"; + +function at(ts: string, msg: OmniMessage): OmniMessage { + return { ...msg, timestamp: ts }; +} + +function counts(total: number): TokenCounts { + return { cache_read: 0, cache_write: 0, output: total, total }; +} + +function metaPayload(over: Partial = {}): SessionMetaPayload { + return { + session_id: S, + model_id: "m1", + provider: "custom", + model_context_window: 1000, + system_prompt: "sp", + tools: [], + agent_state: "/tmp/a", + workspace: "/tmp/w", + ...over, + }; +} + +/** One complete turn (prompt → request → reply → usage), 4 messages, spaced inside one minute `mm`. */ +function turn(mm: number, n: number, sessionTotal: number): OmniMessage[] { + const t = (s: string) => `2026-07-20T10:${String(mm).padStart(2, "0")}:${s}Z`; + return [ + at(t("00.000"), userText(`q${n}`)), + at(t("01.000"), requestBegin()), + at(t("02.000"), assistantText(`a${n}`)), + at(t("03.000"), requestEnd("completed")), + at(t("03.500"), tokenUsage(counts(sessionTotal), counts(100 + n))), + ]; +} + +const textOf = (m: OmniMessage): string | undefined => (m.payload as { text?: string }).text; +const userTexts = (ms: OmniMessage[]): string[] => + ms + .filter( + (m) => + m.type === "model_msg" && + (m.payload as { type?: string; role?: string }).type === "text" && + (m.payload as { role?: string }).role === "user", + ) + .map((m) => textOf(m)!); + +describe("messages windowed reads", () => { + let root: string; + let service: TraceService; + let harness: ReturnType; + + beforeEach(async () => { + root = await makeTempRoot(); + harness = makeTraceHarness(root); + service = harness.service; + }); + afterEach(async () => { + harness.close(); + await fs.rm(root, { recursive: true, force: true }); + }); + + it("tail mid-shard: newest N units, before cursor round-trips to the previous window, beginning ends the chain", async () => { + await writeTraceFile(root, P, A, "2026-07-20", S, 1, [ + sessionMeta(metaPayload()), + ...turn(0, 1, 1000), + ...turn(1, 2, 2000), + ...turn(2, 3, 3000), + ...turn(3, 4, 4000), + ]); + + const tail = await service.readMessagesPage(P, A, S, { kind: "tail", limit: 2 }); + expect(userTexts(tail.messages)).toEqual(["q3", "q4"]); + expect(tail.messages.some((m) => m.type === "session_meta")).toBe(false); + expect(tail.before).toBeDefined(); + expect(tail.prior.turns).toBe(2); + // The cursor names the window's first unit: shard 1, ordinal of q3 (meta + 2×5 turns). + expect(decodeCursor(tail.before!)).toEqual({ fileIndex: 1, ordinal: 11 }); + + const mid = await service.readMessagesPage(P, A, S, { + kind: "before", + cursor: decodeCursor(tail.before!)!, + limit: 1, + }); + expect(userTexts(mid.messages)).toEqual(["q2"]); + expect(mid.prior.turns).toBe(1); + expect(mid.before).toBeDefined(); + + // The oldest window absorbs the preamble (session_meta) and carries no cursor. + const first = await service.readMessagesPage(P, A, S, { + kind: "before", + cursor: decodeCursor(mid.before!)!, + limit: 5, + }); + expect(userTexts(first.messages)).toEqual(["q1"]); + expect(first.messages[0]!.type).toBe("session_meta"); + expect(first.before).toBeUndefined(); + expect(first.prior).toEqual({ + turns: 0, + subagentTokens: 0, + elapsedMs: 0, + sessionTokens: 0, + contextTokens: 0, + }); + + // Windows tile the transcript exactly: concatenated they equal the full read. + const full = await service.readMessages(P, A, S); + expect([...first.messages, ...mid.messages, ...tail.messages]).toEqual(full); + }); + + it("windows span shard boundaries; a mid-Task rotation stays glued to its Task's window", async () => { + // Shard 1: turn 1, then turn 2 begins and rotates mid-Task (auto compaction with + // carry-over): the continuation lives in shard 2 with no new user prompt. + await writeTraceFile(root, P, A, "2026-07-20", S, 1, [ + sessionMeta(metaPayload()), + ...turn(0, 1, 1000), + at("2026-07-20T10:01:00.000Z", userText("q2")), + at("2026-07-20T10:01:01.000Z", requestBegin()), + at("2026-07-20T10:01:02.000Z", requestEnd("completed")), + at("2026-07-20T10:01:02.100Z", tokenUsage(counts(2000), counts(900))), + at( + "2026-07-20T10:01:03.000Z", + compactionBegin({ reason: "context", mode: "summarize", context: 900, turns: 2 }), + ), + at("2026-07-20T10:01:04.000Z", requestBegin()), + at("2026-07-20T10:01:05.000Z", assistantText("[summary]s[/summary]")), + at("2026-07-20T10:01:06.000Z", requestEnd("completed")), + at( + "2026-07-20T10:01:07.000Z", + compactionEnd({ reason: "context", mode: "summarize", status: "completed" }), + ), + ]); + await writeTraceFile(root, P, A, "2026-07-20", S, 2, [ + sessionMeta(metaPayload()), + at("2026-07-20T10:01:08.000Z", userText("[context_summary]\ns\n[/context_summary]")), + at("2026-07-20T10:01:09.000Z", requestBegin()), + at("2026-07-20T10:01:10.000Z", assistantText("a2")), + at("2026-07-20T10:01:11.000Z", requestEnd("completed")), + at("2026-07-20T10:01:11.500Z", tokenUsage(counts(2500), counts(300))), + ...turn(2, 3, 3000), + ]); + + // Tail of 2 units = turn 2 (which spans BOTH shards, compaction included) + turn 3. + const tail = await service.readMessagesPage(P, A, S, { kind: "tail", limit: 2 }); + expect(userTexts(tail.messages)[0]).toBe("q2"); + expect(userTexts(tail.messages)).toContain("q3"); + expect( + tail.messages.some((m) => (m.payload as { type?: string }).type === "compaction_begin"), + ).toBe(true); + // The rotated shard's rewritten session_meta and summary injection ride along inside the window. + expect(tail.messages.filter((m) => m.type === "session_meta")).toHaveLength(1); + expect(decodeCursor(tail.before!)).toEqual({ fileIndex: 1, ordinal: 6 }); + expect(tail.prior.turns).toBe(1); + + const older = await service.readMessagesPage(P, A, S, { + kind: "before", + cursor: decodeCursor(tail.before!)!, + limit: 10, + }); + expect(userTexts(older.messages)).toEqual(["q1"]); + expect(older.before).toBeUndefined(); + const full = await service.readMessages(P, A, S); + expect([...older.messages, ...tail.messages]).toEqual(full); + }); + + it("cuts are pairing-safe: a tool_call and its output never split, steering images stay with their chip", async () => { + await writeTraceFile(root, P, A, "2026-07-20", S, 1, [ + sessionMeta(metaPayload()), + ...turn(0, 1, 1000), + // Turn 2: tool call whose output lands after request_end, steering + image mid-run. + at("2026-07-20T10:01:00.000Z", userText("q2")), + at("2026-07-20T10:01:01.000Z", requestBegin()), + at("2026-07-20T10:01:02.000Z", toolCall({ name: "exec", arguments: "{}", toolCallId: "t1" })), + at("2026-07-20T10:01:03.000Z", requestEnd("completed")), + at("2026-07-20T10:01:04.000Z", toolCallOutput({ output: "out", toolCallId: "t1" })), + at("2026-07-20T10:01:05.000Z", userText("[user_steering]\nfaster\n[/user_steering]")), + at("2026-07-20T10:01:05.100Z", imageUrlMessage("data:image/png;base64,AAAA")), + at("2026-07-20T10:01:06.000Z", requestBegin()), + at("2026-07-20T10:01:07.000Z", assistantText("a2")), + at("2026-07-20T10:01:08.000Z", requestEnd("completed")), + at("2026-07-20T10:01:08.500Z", tokenUsage(counts(2000), counts(200))), + ]); + + // limit 1 must take the WHOLE second turn: the cut lands at q2, never at the + // steering text, its image, or between t1's call and output. + const tail = await service.readMessagesPage(P, A, S, { kind: "tail", limit: 1 }); + expect(userTexts(tail.messages)[0]).toBe("q2"); + const kinds = tail.messages.map((m) => (m.payload as { type?: string }).type); + expect(kinds).toContain("tool_call"); + expect(kinds).toContain("tool_call_output"); + expect(kinds).toContain("image_url"); + // Steering opened no new unit: the previous window holds exactly turn 1. + const older = await service.readMessagesPage(P, A, S, { + kind: "before", + cursor: decodeCursor(tail.before!)!, + limit: 10, + }); + expect(userTexts(older.messages)).toEqual(["q1"]); + }); + + it("a text+images send is one unit and one outline turn; banner and goal-round prompts cut but never count", async () => { + await writeTraceFile(root, P, A, "2026-07-20", S, 1, [ + sessionMeta(metaPayload()), + // Send 1: text + two images (one entry, one unit). + at("2026-07-20T10:00:00.000Z", userText("look")), + at("2026-07-20T10:00:00.100Z", imageUrlMessage("data:image/png;base64,AAAA")), + at("2026-07-20T10:00:00.200Z", imageUrlMessage("data:image/png;base64,BBBB")), + at("2026-07-20T10:00:01.000Z", requestBegin()), + at("2026-07-20T10:00:02.000Z", assistantText("a1")), + at("2026-07-20T10:00:03.000Z", requestEnd("completed")), + at("2026-07-20T10:00:03.500Z", tokenUsage(counts(1000), counts(100))), + // A handoff banner send: machine-only, no outline entry (buildOutline returns null body). + at( + "2026-07-20T10:01:00.000Z", + userText(buildHandoffMessage({ agentId: "worker", workspace: "/tmp/w" })), + ), + at("2026-07-20T10:01:01.000Z", requestBegin()), + at("2026-07-20T10:01:02.000Z", assistantText("a2")), + at("2026-07-20T10:01:03.000Z", requestEnd("completed")), + at("2026-07-20T10:01:03.500Z", tokenUsage(counts(2000), counts(200))), + // A goal round-2 re-send: starts a Task but opens no outline entry. + at("2026-07-20T10:02:00.000Z", userText("[goal]\nround: 2\nbudget: unlimited\n[/goal]\ngo")), + at("2026-07-20T10:02:01.000Z", requestBegin()), + at("2026-07-20T10:02:02.000Z", assistantText("a3")), + at("2026-07-20T10:02:03.000Z", requestEnd("completed")), + at("2026-07-20T10:02:03.500Z", tokenUsage(counts(3000), counts(300))), + ...turn(3, 4, 4000), + ]); + + // Four units in total (each send cuts), but only ONE outline entry precedes q4: + // the text+images send counts once, the banner and the goal round count zero — + // exactly buildOutline's rule, so q4 renders as global turn 2 with offset 1. + const tail = await service.readMessagesPage(P, A, S, { kind: "tail", limit: 1 }); + expect(userTexts(tail.messages)).toEqual(["q4"]); + expect(tail.prior.turns).toBe(1); + + // Walk one more window back: the goal-round unit alone; still 1 entry before it. + const goalWin = await service.readMessagesPage(P, A, S, { + kind: "before", + cursor: decodeCursor(tail.before!)!, + limit: 1, + }); + expect(userTexts(goalWin.messages)[0]).toContain("[goal]"); + expect(goalWin.prior.turns).toBe(1); + + const bannerWin = await service.readMessagesPage(P, A, S, { + kind: "before", + cursor: decodeCursor(goalWin.before!)!, + limit: 1, + }); + expect(bannerWin.prior.turns).toBe(1); // only the image send precedes the banner + const firstWin = await service.readMessagesPage(P, A, S, { + kind: "before", + cursor: decodeCursor(bannerWin.before!)!, + limit: 1, + }); + expect(firstWin.prior.turns).toBe(0); + expect(firstWin.before).toBeUndefined(); + // The text and its two images stayed one unit. + expect(userTexts(firstWin.messages)).toEqual(["look"]); + expect( + firstWin.messages.filter((m) => (m.payload as { type?: string }).type === "image_url"), + ).toHaveLength(2); + }); + + it("prior stats: finished-Task elapsed, session/context token readings", async () => { + await writeTraceFile(root, P, A, "2026-07-20", S, 1, [ + sessionMeta(metaPayload()), + ...turn(0, 1, 1000), // request_begin 10:00:01 → request_end 10:00:03 = 2s... elapsed = lastReqEnd - firstMsg(q1 at 10:00:00) = 3s + ...turn(1, 2, 2000), + ...turn(2, 3, 3000), + ]); + const tail = await service.readMessagesPage(P, A, S, { kind: "tail", limit: 1 }); + // Web semantics: a turn's elapsed = last request_end − its first message = 3s each. + expect(tail.prior.elapsedMs).toBe(2 * 3000); + expect(tail.prior.sessionTokens).toBe(2000); // last session.total before the window + expect(tail.prior.contextTokens).toBe(102); // last request.total before the window (turn(…, 2) writes 100+2) + expect(tail.prior.subagentTokens).toBe(0); + }); + + it("subagent pointers expand only inside the window; earlier children feed prior.subagentTokens without appearing", async () => { + const child2 = "session-2026-07-20-10-02-00-eeff0003"; + await writeTraceFile(root, P, A, "2026-07-20", S, 1, [ + sessionMeta(metaPayload()), + // Turn 1 spawns CHILD. + at("2026-07-20T10:00:00.000Z", userText("q1")), + at("2026-07-20T10:00:01.000Z", requestBegin()), + at( + "2026-07-20T10:00:02.000Z", + toolCall({ name: "run_subagent", arguments: "{}", toolCallId: "c1" }), + ), + at("2026-07-20T10:00:03.000Z", subagentEvent(CHILD)), + at("2026-07-20T10:00:04.000Z", requestEnd("completed")), + at("2026-07-20T10:00:05.000Z", toolCallOutput({ output: "done", toolCallId: "c1" })), + at("2026-07-20T10:00:06.000Z", tokenUsage(counts(1000), counts(100))), + // Turn 2 spawns child2. + at("2026-07-20T10:01:00.000Z", userText("q2")), + at("2026-07-20T10:01:01.000Z", requestBegin()), + at( + "2026-07-20T10:01:02.000Z", + toolCall({ name: "run_subagent", arguments: "{}", toolCallId: "c2" }), + ), + at("2026-07-20T10:01:03.000Z", subagentEvent(child2)), + at("2026-07-20T10:01:04.000Z", requestEnd("completed")), + at("2026-07-20T10:01:05.000Z", toolCallOutput({ output: "done", toolCallId: "c2" })), + at("2026-07-20T10:01:06.000Z", tokenUsage(counts(2000), counts(120))), + ]); + await writeTraceFile(root, P, A, "2026-07-20", CHILD, 1, [ + sessionMeta(metaPayload({ session_id: CHILD })), + at("2026-07-20T10:00:03.500Z", assistantText("child says")), + at("2026-07-20T10:00:03.800Z", tokenUsage(counts(400), counts(400))), + ]); + await writeTraceFile(root, P, A, "2026-07-20", child2, 1, [ + sessionMeta(metaPayload({ session_id: child2 })), + at("2026-07-20T10:01:03.500Z", assistantText("child2 says")), + at("2026-07-20T10:01:03.800Z", tokenUsage(counts(70), counts(70))), + ]); + + const tail = await service.readMessagesPage(P, A, S, { kind: "tail", limit: 1 }); + // The window (turn 2) expands child2 in place, origin-tagged, pointer replaced. + const origins = tail.messages.filter((m) => m.origin !== undefined); + expect(origins.length).toBeGreaterThan(0); + for (const m of origins) expect(m.origin).toEqual([child2]); + expect(tail.messages.some((m) => (m.payload as { type?: string }).type === "subagent")).toBe( + false, + ); + // CHILD (turn 1) is NOT in the window — its usage rides prior.subagentTokens instead. + expect(tail.messages.some((m) => textOf(m) === "child says")).toBe(false); + expect(tail.prior.subagentTokens).toBe(400); + + // The older window then expands CHILD with the exact shape the full path produces. + const older = await service.readMessagesPage(P, A, S, { + kind: "before", + cursor: decodeCursor(tail.before!)!, + limit: 10, + }); + const full = await service.readMessages(P, A, S); + expect([...older.messages, ...tail.messages]).toEqual(full); + }); + + it("shard-read discipline: after priming, old-window requests never read the newest shard and tail requests never read old shards", async () => { + await writeTraceFile(root, P, A, "2026-07-20", S, 1, [ + sessionMeta(metaPayload()), + ...turn(0, 1, 1000), + ...turn(1, 2, 2000), + ]); + await writeTraceFile(root, P, A, "2026-07-21", S, 2, [ + sessionMeta(metaPayload()), + ...turn(2, 3, 3000), + ]); + await writeTraceFile(root, P, A, "2026-07-21", S, 3, [ + sessionMeta(metaPayload()), + ...turn(3, 4, 4000), + ...turn(4, 5, 5000), + ]); + + // Priming: the first windowed read backfills the per-shard prefix cache (reads old shards once). + const primed = await service.readMessagesPage(P, A, S, { kind: "tail", limit: 1 }); + expect(userTexts(primed.messages)).toEqual(["q5"]); + + // A tail request now touches ONLY the newest shard. + harness.shardReads.length = 0; + await service.readMessagesPage(P, A, S, { kind: "tail", limit: 1 }); + expect(harness.shardReads).toHaveLength(1); + expect(harness.shardReads[0]).toContain(`${S}_003.jsonl`); + + // An old-window request touches ONLY its own shard — never the newest one. + harness.shardReads.length = 0; + const older = await service.readMessagesPage(P, A, S, { + kind: "before", + cursor: { fileIndex: 2, ordinal: 1 }, + limit: 1, + }); + expect(userTexts(older.messages)).toEqual(["q2"]); + expect(harness.shardReads.every((p) => !p.includes(`${S}_003.jsonl`))).toBe(true); + expect(harness.shardReads.some((p) => p.includes(`${S}_001.jsonl`))).toBe(true); + }); + + it("cursor edge cases: unknown shard yields an empty end-of-history page; empty sessions page cleanly", async () => { + expect(await service.readMessagesPage(P, A, S, { kind: "tail", limit: 5 })).toEqual({ + messages: [], + prior: { turns: 0, subagentTokens: 0, elapsedMs: 0, sessionTokens: 0, contextTokens: 0 }, + }); + await writeTraceFile(root, P, A, "2026-07-20", S, 1, [ + sessionMeta(metaPayload()), + ...turn(0, 1, 1000), + ]); + const gone = await service.readMessagesPage(P, A, S, { + kind: "before", + cursor: { fileIndex: 9, ordinal: 4 }, + limit: 5, + }); + expect(gone.messages).toEqual([]); + expect(gone.before).toBeUndefined(); + // Round-trip of the cursor encoding itself. + expect(decodeCursor(encodeCursor({ fileIndex: 12, ordinal: 345 }))).toEqual({ + fileIndex: 12, + ordinal: 345, + }); + expect(decodeCursor("12:")).toBeNull(); + expect(decodeCursor("a:b")).toBeNull(); + }); + + it("tail covering the whole transcript returns everything with no cursor and equals the full read", async () => { + await writeTraceFile(root, P, A, "2026-07-20", S, 1, [ + sessionMeta(metaPayload()), + ...turn(0, 1, 1000), + ...turn(1, 2, 2000), + ]); + const tail = await service.readMessagesPage(P, A, S, { kind: "tail", limit: 50 }); + expect(tail.before).toBeUndefined(); + expect(tail.prior.turns).toBe(0); + expect(tail.messages).toEqual(await service.readMessages(P, A, S)); + }); +}); diff --git a/packages/web/src/api/endpoints.ts b/packages/web/src/api/endpoints.ts index fc6ebce..19fdb27 100644 --- a/packages/web/src/api/endpoints.ts +++ b/packages/web/src/api/endpoints.ts @@ -306,17 +306,32 @@ export const patchSession = (sessionId: string, body: SessionPatchRequest) => export const deleteSession = (sessionId: string) => apiFetch(`/api/sessions/${encodeURIComponent(sessionId)}`, { method: "DELETE" }); +/** Windowed history request: the newest N units (tail), or the N units before a cursor. */ +export type MessagesPageQuery = + { kind: "tail"; limit: number } | { kind: "before"; cursor: string; limit: number }; + /** * History rebuild. Carries the server's clock at read time (see ApiFetchMeta.serverNowMs) * alongside the messages: a Task still running has no Trace entry for the event currently in * flight, so its elapsed can only be measured by differencing this against the Task's first * message timestamp — both server-side values, so no client clock offset enters the result * (see pushMessages). + * + * With `page`, requests a WINDOW instead of the full transcript (tail-first loading / + * scroll-up backfill — see stream-controller): the response then carries + * `MessagesResponse.page`. Omitted = the legacy full read (the resync fallback path). */ -export const getMessages = (sessionId: string) => - apiFetchWithMeta( - `/api/sessions/${encodeURIComponent(sessionId)}/messages`, +export const getMessages = (sessionId: string, page?: MessagesPageQuery) => { + const qs = + page === undefined + ? "" + : page.kind === "tail" + ? `?tailLimit=${page.limit}` + : `?before=${encodeURIComponent(page.cursor)}&limit=${page.limit}`; + return apiFetchWithMeta( + `/api/sessions/${encodeURIComponent(sessionId)}/messages${qs}`, ).then(({ data, serverNowMs }) => ({ ...data, serverNowMs })); +}; // Task execution, approval, abort, compaction ------------------------------------------------------ diff --git a/packages/web/src/features/chat/chat-page.tsx b/packages/web/src/features/chat/chat-page.tsx index 19e0989..f8fd042 100644 --- a/packages/web/src/features/chat/chat-page.tsx +++ b/packages/web/src/features/chat/chat-page.tsx @@ -289,16 +289,43 @@ export function ChatPage() { discard: discardSessionDraft, } = useSessionDraft(selected?.sessionId ?? null); + // The rendered transcript: backfilled older windows (frozen items, negative ids) ahead + // of the live tail model's items. Version keys the memo — a prepend bumps it. + const allItems = useMemo( + () => + stream.prefixItems.length > 0 + ? [...stream.prefixItems, ...stream.model.items] + : stream.model.items, + // eslint-disable-next-line react-hooks/exhaustive-deps + [stream.version, routeSessionId], + ); // Derivations over the stream items (the model mutates in place, so `version` — its own // repaint signal — keys the memos; the session id covers a switch racing a same-valued - // version): the composer's ↑/↓ recall list and the left outline's entries. + // version): the composer's ↑/↓ recall list and the left outline's entries. Both read the + // LOADED transcript (prefix included), so backfilling extends recall and the index alike. const inputHistory = useMemo( - () => buildInputHistory(stream.model.items), + () => buildInputHistory(allItems), // eslint-disable-next-line react-hooks/exhaustive-deps [stream.version, routeSessionId], ); const outline = useMemo( - () => buildOutline(stream.model.items), + () => buildOutline(allItems), + // eslint-disable-next-line react-hooks/exhaustive-deps + [stream.version, routeSessionId], + ); + // The subagents panel's model view: the live model, with backfilled windows' items and + // nested subagent models merged in — a chip clicked on a backfilled turn must still + // resolve its historical Task slice and child conversation. Scalar fields snapshot per + // version, which is exactly as fresh as everything else the panel renders. + const panelModel = useMemo( + () => + stream.prefixItems.length > 0 || stream.prefixSubagents.size > 0 + ? { + ...stream.model, + items: [...allItems], + subagents: new Map([...stream.prefixSubagents, ...stream.model.subagents]), + } + : stream.model, // eslint-disable-next-line react-hooks/exhaustive-deps [stream.version, routeSessionId], ); @@ -347,6 +374,10 @@ export function ChatPage() { // dot still signal), and never re-triggers an already-open panel (that would yank a pinned // historical graph back to the latest Task); the tracker consumes the attempt regardless. const panelTaskScopeRef = useRef(createPanelTaskScope()); + // Deliberately the LIVE model's items only (never the backfilled prefix): the tracker + // reads an INCREASE as "the user started a new Task", and a scroll-up backfill growing + // the count would spuriously re-arm the auto-open mid-conversation. The latest Task + // always lives in the live tail window, so live-only loses nothing. const taskCount = taskStartCount(stream.model.items); const liveSpawn = stream.taskState !== "idle" && latestTaskHasSubagent(stream.model); useEffect(() => { @@ -1087,6 +1118,7 @@ export function ChatPage() { {!railFit.shown && ( @@ -1240,16 +1272,27 @@ export function ChatPage() { ) : ( 0 keeps the + * "more above" edge dots on — numbering never lies about unloaded turns, and the rail + * signals that scrolling up will reveal them. + */ + turnOffset?: number; /** Stream repaint signal: re-runs the scrollspy as content grows or the stream remounts. */ version: number; /** MessageStream's scroll container (null while the stream isn't mounted, e.g. the empty greeting). */ @@ -227,7 +236,7 @@ export function ConversationOutline({ // cheap, and keying on version also re-binds after the stream remounts on a session switch. useEffect(() => { const el = scrollRef.current; - if (!fit.shown || !el || entries.length < OUTLINE_MIN_TURNS) return; + if (!fit.shown || !el || !outlineVisible(turnOffset, entries.length)) return; const ids = new Set(entries.map((entry) => entry.anchorId)); let raf: number | null = null; const compute = () => { @@ -246,9 +255,11 @@ export function ConversationOutline({ }; // `entries` is rebuilt per version; length + version cover it. // eslint-disable-next-line react-hooks/exhaustive-deps - }, [fit.shown, scrollRef, entries.length, version]); + }, [fit.shown, scrollRef, entries.length, turnOffset, version]); - if (!fit.shown || entries.length < OUTLINE_MIN_TURNS) return null; + // The gate counts the WHOLE conversation (loaded + unloaded turns): a long session + // whose tail window happens to hold few entries still deserves its index. + if (!fit.shown || !outlineVisible(turnOffset, entries.length)) return null; /** Tick center Y relative to the rail overlay (the preview card anchors to it, clamped in render). */ const tickTop = (e: MouseEvent | FocusEvent) => { @@ -291,7 +302,9 @@ export function ConversationOutline({ inside the rail by construction, and this guarantees a mismeasure still can't push ticks (which take pointer events) out over the toolbar or composer. */}
- {start > 0 && } + {/* "More above" also covers turns not yet LOADED (turnOffset > 0): the dots tell + the same story either way — earlier turns exist beyond the rendered ticks. */} + {(start > 0 || turnOffset > 0) && } {visible.map((entry, i) => { const active = entry.anchorId === activeId; return ( @@ -300,10 +313,10 @@ export function ConversationOutline({ type="button" data-outline-tick={entry.anchorId} aria-current={active || undefined} - // Global turn number (start + i): the window changes which ticks render, - // never how a turn is numbered. + // Global turn number (turnOffset + start + i): neither the sliding window + // nor a partially-loaded history changes how a turn is numbered. aria-label={S.chat.outlineTickLabel( - start + i + 1, + globalTurnNumber(turnOffset, start + i), entry.question || S.chat.outlineNoText, )} style={{ height: pitch }} @@ -373,17 +386,20 @@ export function ConversationOutline({ */ export function OutlineMenuButton({ entries, + turnOffset = 0, scrollRef, running, }: { entries: OutlineEntry[]; + /** Turns before the loaded window (windowed loading): gates visibility on the WHOLE conversation; the list itself shows loaded turns only. */ + turnOffset?: number; scrollRef: RefObject; running: boolean; }) { const [open, setOpen] = useState(false); const [activeId, setActiveId] = useState(null); // Same gate as the rail: with this few turns neither shape earns its place. - if (entries.length < OUTLINE_MIN_TURNS) return null; + if (!outlineVisible(turnOffset, entries.length)) return null; const setOpenComputing = (next: boolean) => { if (next && scrollRef.current) { diff --git a/packages/web/src/features/chat/message-stream.tsx b/packages/web/src/features/chat/message-stream.tsx index 8b2019b..7effe7d 100644 --- a/packages/web/src/features/chat/message-stream.tsx +++ b/packages/web/src/features/chat/message-stream.tsx @@ -155,12 +155,29 @@ export function MessageItems({ items, ctx }: { items: ChatItem[]; ctx: StreamRen return <>{nodes}; } +/** Scroll-up backfill wiring (windowed history): state + trigger for the top-of-stream affordance. */ +export interface OlderHistoryControls { + /** Older windows exist beyond the loaded history (scrolling near the top triggers onLoad). */ + hasMore: boolean; + /** A backfill request is in flight (spinner row). */ + loading: boolean; + /** The last backfill failed (retry row); null = fine. */ + error: string | null; + /** Number of windows already prepended: the prepend signal for scroll anchoring, and >0 gates the beginning-of-history marker (a session that fit one window shows no extra chrome). */ + prependedCount: number; + onLoad: () => void; +} + +/** Distance from the top (px) at which scrolling starts fetching the previous history window. */ +const OLDER_TRIGGER_PX = 300; + export function MessageStream({ items, version, ctx, scrollElRef, outline, + older, }: { items: ChatItem[]; /** View-model version number (a repaint signal for in-place updates that also drives auto-scroll). */ @@ -174,6 +191,8 @@ export function MessageStream({ * and anchor its absolute positioning to this wrapper, which only this component owns. */ outline?: ReactNode; + /** Scroll-up backfill of older history windows; omitted = the whole transcript is loaded (no top affordance). */ + older?: OlderHistoryControls; }) { const scrollRef = useRef(null); // An upward-swipe intent immediately exits auto-follow; scrolling back near the bottom resumes it — see stream-follow.ts (#75) for the exact rule. @@ -220,6 +239,19 @@ export function MessageStream({ ); }; + /** Held refs for prepend scroll anchoring (see the layout effect below). */ + const olderRef = useRef(older); + olderRef.current = older; + const lastPrependedRef = useRef(older?.prependedCount ?? 0); + const lastHeightRef = useRef(0); + + /** Near the top of loaded history: fetch the previous window (loading/error states gate re-triggering; the retry row is click-driven). */ + const maybeLoadOlder = (el: HTMLDivElement) => { + const o = olderRef.current; + if (!o || !o.hasMore || o.loading || o.error !== null) return; + if (el.scrollTop < OLDER_TRIGGER_PX) o.onLoad(); + }; + const onScroll = () => { const el = scrollRef.current; if (!el) return; @@ -234,6 +266,7 @@ export function MessageStream({ scrollHeight: el.scrollHeight, clientHeight: el.clientHeight, }); + maybeLoadOlder(el); syncJump(); }; @@ -242,11 +275,24 @@ export function MessageStream({ // during the animated return — the glide owns the scroll position until it arrives. useLayoutEffect(() => { const el = scrollRef.current; + // Prepend scroll anchoring: when older windows land ABOVE the viewport, keep the + // message the user was reading exactly where it was by offsetting scrollTop by the + // content growth (same pre-paint timing as the stick snap, so nothing flashes). + // Keyed on the prepend count — ordinary streaming growth at the bottom must not + // shift the view. lastHeightRef is refreshed every commit, so at the prepend commit + // it still holds the pre-prepend height. Skipped while sticking (the snap below + // owns the position; a prepend while stuck at the bottom cannot move the tail). + const prepended = older?.prependedCount ?? 0; + if (el && prepended > lastPrependedRef.current && !follow.stick && !returningRef.current) { + el.scrollTop += el.scrollHeight - lastHeightRef.current; + } + lastPrependedRef.current = prepended; + if (el) lastHeightRef.current = el.scrollHeight; if (el && follow.stick && !returningRef.current) el.scrollTop = el.scrollHeight; syncJump(); // syncJump is recreated per render; the effect intentionally keys on stream growth only. // eslint-disable-next-line react-hooks/exhaustive-deps - }, [version, follow]); + }, [version, follow, older?.prependedCount]); /** Back-to-bottom: glide down to the live bottom (reduced motion gets an instant jump); follow re-engages on arrival. */ const jumpToLatest = () => { @@ -316,6 +362,32 @@ export function MessageStream({ className="anim-fade h-full overflow-y-auto px-4 py-4 md:px-6" >
+ {/* Top-of-history affordance: spinner while the previous window loads, a click-to-retry + row after a failure, and — once at least one window was backfilled — a quiet + beginning-of-conversation marker when there is nothing older. Idle-with-more shows + nothing: scrolling near the top triggers the fetch by itself. */} + {older && items.length > 0 && ( +
+ {older.loading ? ( + + + {S.chat.loadingEarlier} + + ) : older.error !== null ? ( + + ) : !older.hasMore && older.prependedCount > 0 ? ( + + {S.chat.historyBeginning} + + ) : null} +
+ )} {items.length === 0 ? ( ) : ( diff --git a/packages/web/src/features/chat/outline-model.ts b/packages/web/src/features/chat/outline-model.ts index 08e2512..4fdfd14 100644 --- a/packages/web/src/features/chat/outline-model.ts +++ b/packages/web/src/features/chat/outline-model.ts @@ -36,6 +36,24 @@ const ANSWER_CAP = 500; */ export const OUTLINE_MIN_TURNS = 5; +/** + * With windowed history loading, entries built from the loaded items are only the tail of + * the conversation: `turnOffset` (MessagesPageInfo.earlierTurns — the server counts with + * the same entry rule buildOutline applies; the two must stay in step) is the number of + * turns that exist before them. These two helpers are the ONLY places that combine the + * offset with loaded-entry indices, so numbering can never drift between the shapes. + */ + +/** Global 1-based turn number of loaded entry `index` (its position in the loaded entries array). */ +export function globalTurnNumber(turnOffset: number, index: number): number { + return turnOffset + index + 1; +} + +/** Whether the outline earns its place: the gate counts the WHOLE conversation — unloaded earlier turns included — never just the loaded window. */ +export function outlineVisible(turnOffset: number, loadedCount: number): boolean { + return turnOffset + loadedCount >= OUTLINE_MIN_TURNS; +} + /** Rail window half-widths: at most this many ticks render before/after the active one. */ export const OUTLINE_WINDOW_BEFORE = 20; export const OUTLINE_WINDOW_AFTER = 20; diff --git a/packages/web/src/features/chat/use-session-stream.ts b/packages/web/src/features/chat/use-session-stream.ts index 7ba5932..3f0e9a6 100644 --- a/packages/web/src/features/chat/use-session-stream.ts +++ b/packages/web/src/features/chat/use-session-stream.ts @@ -24,19 +24,37 @@ import type { import { getGoal, getMe, getMessages } from "../../api/endpoints"; import { openSessionStream } from "../../api/sse"; import { createStreamController } from "../../lib/omni/stream-controller"; -import type { PendingApproval, StreamController } from "../../lib/omni/stream-controller"; +import type { + OlderHistoryState, + PendingApproval, + StreamController, +} from "../../lib/omni/stream-controller"; import { createStreamModel } from "../../lib/omni/stream-model"; -import type { StreamModel } from "../../lib/omni/stream-model"; +import type { ChatItem, StreamModel } from "../../lib/omni/stream-model"; import type { GoalBannerState } from "./goal-use"; -export type { PendingApproval } from "../../lib/omni/stream-controller"; +export type { OlderHistoryState, PendingApproval } from "../../lib/omni/stream-controller"; /** Minimum spacing between version commits: below one frame at 8fps, invisible as staleness, but ~8× fewer full re-parses of the growing message during a fast large-code stream. */ const BUMP_MIN_INTERVAL_MS = 120; export interface SessionStreamState { - /** View model (updated in place; version bump triggers re-render). */ + /** View model (updated in place; version bump triggers re-render): the LIVE tail window. */ model: StreamModel; + /** + * Backfilled older-window items, oldest first — the transcript renders + * `[...prefixItems, ...model.items]`. Empty until the user scrolls up past the tail + * window (see loadOlder); ids are negative and unique, so keys/anchors stay clean. + */ + prefixItems: readonly ChatItem[]; + /** Nested subagent models owned by backfilled windows (the subagents panel merges them with model.subagents). */ + prefixSubagents: ReadonlyMap; + /** Outline entries existing before the oldest loaded window (global turn-numbering offset). */ + outlineOffset: number; + /** Scroll-up backfill state (top affordance: spinner / retry / beginning-of-history). */ + older: OlderHistoryState; + /** Prepend the previous history window (triggered near the top of the loaded transcript). */ + loadOlder: () => void; version: number; /** True until history finishes loading. */ loading: boolean; @@ -79,6 +97,9 @@ export interface SessionStreamState { } const EMPTY_PENDING: ReadonlyMap = new Map(); +const EMPTY_PREFIX: readonly ChatItem[] = []; +const EMPTY_SUBAGENTS: ReadonlyMap = new Map(); +const IDLE_OLDER: OlderHistoryState = { hasMore: false, loading: false, error: null }; export function useSessionStream( sessionId: string | null, @@ -208,8 +229,9 @@ export function useSessionStream( const controller = createStreamController({ // The whole response rides through: `live` (in-progress stream tail) lets the - // controller seed the currently streaming message after a reload (see stream-controller). - loadMessages: () => getMessages(sessionId), + // controller seed the currently streaming message after a reload, and `page` (the + // windowed-history envelope) drives tail-first loading (see stream-controller). + loadMessages: (page) => getMessages(sessionId, page), onTaskState: setTaskState, onQueuedFollowUps: setQueuedFollowUps, onPendingSteering: setPendingSteering, @@ -269,6 +291,10 @@ export function useSessionStream( void controllerRef.current?.retry(); }, []); + const loadOlder = useCallback(() => { + void controllerRef.current?.loadOlder(); + }, []); + const dismissModelAuthDead = useCallback(() => { const m = controllerRef.current?.model; if (m && m.lastAuthFailureMs !== null) { @@ -282,6 +308,11 @@ export function useSessionStream( return { model: controllerRef.current?.model ?? placeholderRef.current, + prefixItems: controllerRef.current?.prefixItems ?? EMPTY_PREFIX, + prefixSubagents: controllerRef.current?.prefixSubagents ?? EMPTY_SUBAGENTS, + outlineOffset: controllerRef.current?.outlineOffset ?? 0, + older: controllerRef.current?.older ?? IDLE_OLDER, + loadOlder, version, loading, taskState, diff --git a/packages/web/src/lib/omni/stream-controller.ts b/packages/web/src/lib/omni/stream-controller.ts index f725e9f..a220010 100644 --- a/packages/web/src/lib/omni/stream-controller.ts +++ b/packages/web/src/lib/omni/stream-controller.ts @@ -34,6 +34,7 @@ import type { OmniMessage, ToolCallPayload } from "@prismshadow/penguin-core/omn import type { GoalServerEvent, MessagesLiveTail, + MessagesPageInfo, PendingSteeringInfo, ServerEvent, SessionStatus, @@ -51,7 +52,8 @@ import { pushMessages, registerLocalDecision, } from "./stream-model"; -import type { StreamModel } from "./stream-model"; +import type { ChatItem, StreamModel } from "./stream-model"; +import { seedPriorStats } from "./task-stats"; /** A single pending approval (keyed by approvalKey(origin, toolCallId)). */ export interface PendingApproval { @@ -73,17 +75,45 @@ function parseEventId(id: string): { epoch: string; seq: number } | null { return { epoch: id.slice(0, sep), seq }; } +/** A windowed history request (mirrors the server's tailLimit / before params). */ +export type MessagesPageQuery = + { kind: "tail"; limit: number } | { kind: "before"; cursor: string; limit: number }; + +/** + * Initial (tail) window size, in message-bearing units — one unit = one Task, opened by + * a user prompt (the server's cut rule; see MessagesPageInfo). 200 covers the vast + * majority of real sessions in a single request, so ordinary conversations still load + * whole exactly as before — only the pathological long tail (months-long sessions, + * agentic marathons) starts windowed, which is the point: their full-transcript reads + * were the unbounded memory/disk cost this pagination removes. + */ +export const TAIL_UNITS = 200; + +/** Scroll-up backfill window size: smaller than the tail so each prepend stays snappy. */ +export const OLDER_UNITS = 100; + +/** + * Item-id space reserved per prepended window. The live model numbers its items upward + * from 1; each prepended window numbers upward from its own NEGATIVE base, so ids stay + * unique across the concatenated view (React keys, outline anchors) without ever + * renumbering already-mounted items. A window holds at most a few thousand items — + * far under the span. + */ +const PREPEND_ID_SPAN = 1_000_000; + export interface StreamControllerDeps { /** * Fetch history messages (GET /api/sessions/:id/messages), including the live tail while * running. `serverNowMs` is the server's clock at read time (the response's `Date` header); * omitted/null just costs a running Task's header the time its in-flight event has taken so - * far, which falls back to the Trace's own span (see pushMessages). + * far, which falls back to the Trace's own span (see pushMessages). `page` requests a + * WINDOW (the response then carries `page`); omitted = the legacy full transcript. */ - loadMessages: () => Promise<{ + loadMessages: (page?: MessagesPageQuery) => Promise<{ messages: OmniMessage[]; live?: MessagesLiveTail; serverNowMs?: number | null; + page?: MessagesPageInfo; }>; /** Authoritative running state from the stream (covers both the subscription snapshot and transition events). */ onTaskState: (state: SessionStatus) => void; @@ -108,14 +138,37 @@ export interface StreamControllerDeps { now?: () => number; } +/** Scroll-up backfill state (drives the stream's top affordance). */ +export interface OlderHistoryState { + /** Older windows exist beyond the loaded prefix. */ + hasMore: boolean; + /** A backfill request is in flight. */ + loading: boolean; + /** The last backfill failed (the affordance offers a retry); null = fine. */ + error: string | null; +} + export interface StreamController { - /** The current view model (a resync rebuild swaps in a new object). */ + /** The current view model (a resync rebuild swaps in a new object): the LIVE tail window. */ readonly model: StreamModel; + /** + * Items of the backfilled older windows, oldest first — render them immediately BEFORE + * `model.items`. Frozen once built (their Tasks are complete); item ids are negative + * and unique across windows, so the concatenated list keys/anchors cleanly. + */ + readonly prefixItems: readonly ChatItem[]; + /** Nested subagent models of the backfilled windows (merged view for the subagents panel; disjoint from model.subagents — a child session lives in exactly one window). */ + readonly prefixSubagents: ReadonlyMap; + /** Outline entries that exist before the OLDEST loaded window: the outline's global numbering offset. */ + readonly outlineOffset: number; + readonly older: OlderHistoryState; readonly pendingApprovals: ReadonlyMap; - /** Load history for the first time (called once after connect-first). */ + /** Load history for the first time (called once after connect-first): fetches the TAIL window. */ load: () => Promise; /** Retry entry point after a history load failure (keeps the buffer, refetches history). */ retry: () => Promise; + /** Prepend the previous window (scroll-up backfill); no-op while loading, failed, at the beginning, or before the initial load settled. */ + loadOlder: () => Promise; /** SSE OmniMessage entry point (`eventId`: the SSE event id, used for live-tail cursor alignment). */ handleOmni: (msg: OmniMessage, eventId?: string | null) => void; /** SSE server-event entry point (`eventId`: same as handleOmni). */ @@ -146,6 +199,49 @@ export function createStreamController(deps: StreamControllerDeps): StreamContro /** Whether the most recent load failed (retry only takes effect after a failure, to avoid mistakenly replaying history). */ let failed = false; + // —— Windowed-history state (tail-first load + scroll-up backfill) —— + /** Items of backfilled older windows, oldest first (frozen; rendered before model.items). */ + let prefixItems: ChatItem[] = []; + /** Nested subagent models owned by backfilled windows. */ + let prefixSubagents = new Map(); + /** How many windows have been prepended (derives each one's negative item-id base). */ + let prependCount = 0; + /** Cursor for the NEXT older window (= the oldest loaded window's start); null = beginning reached or full transcript loaded. */ + let nextBefore: string | null = null; + /** + * The LIVE tail window's start cursor, as returned by the last tail fetch — the resync + * continuity anchor: a refetched tail whose start cursor equals this provably abuts the + * retained prefix. Null = the tail reaches the beginning (or the transcript was loaded + * whole), in which case there is no prefix to splice against. + */ + let tailStartCursor: string | null = null; + const older: OlderHistoryState = { hasMore: false, loading: false, error: null }; + /** Outline entries before the OLDEST loaded window (the outline's numbering offset). */ + let outlineOffset = 0; + + /** Reset all windowed-history bookkeeping to "everything loaded from the beginning". */ + const resetPaging = (): void => { + prefixItems = []; + prefixSubagents = new Map(); + prependCount = 0; + nextBefore = null; + tailStartCursor = null; + older.hasMore = false; + older.loading = false; + older.error = null; + outlineOffset = 0; + }; + + /** Adopt a TAIL page's pagination envelope as the fresh baseline (initial load / prefix-dropping rebuild). */ + const adoptTailPage = (page: MessagesPageInfo | undefined): void => { + resetPaging(); + if (page === undefined) return; // full transcript: nothing older exists by definition + nextBefore = page.before ?? null; + tailStartCursor = page.before ?? null; + older.hasMore = page.before !== undefined; + outlineOffset = page.earlierTurns; + }; + const clearPending = (): void => { if (pending.size === 0) return; pending.clear(); @@ -222,6 +318,32 @@ export function createStreamController(deps: StreamControllerDeps): StreamContro } }; + /** + * resync_required decision tree — which refetch rebuilds the model. Resync means the + * SSE buffer was evicted and the transcript state is suspect, so every branch below + * chooses correctness over cleverness (ANY doubt falls back to the full read): + * + * 1. No backfilled prefix → refetch the TAIL window. Identical in shape to the + * initial load: the window is a disk-true suffix, the buffered events replay + * with overlap dedup, and the live attachment weaves in under the existing + * channel-epoch guard (weaveLiveTail skips seeding when the cursor's epoch + * doesn't match the events seen on this connection). + * 2. Prefix retained → refetch the TAIL window and splice ONLY when continuity is + * provable: the refetched window's start cursor must EQUAL the recorded start + * of the current tail window (cursors are (shard, ordinal) positions on + * immutable storage, so equality proves the new tail abuts the prefix exactly — + * no gap, no overlap). Equality holds precisely when no new unit started since + * the last tail fetch, the common mid-Task resync. + * 3. Prefix retained but the refetched tail reaches the very beginning (no cursor) + * → the tail alone provably covers everything: drop the prefix and use it. + * 4. Anything else — the start cursor moved (new units arrived), the response + * carried no page envelope, or the tail fetch itself failed mid-decision — + * is doubt: fall back to the legacy FULL refetch (one complete transcript, no + * prefix, offsets zeroed). Slow but beyond suspicion. + * + * The decision runs inside load() (it needs the response); this entry only picks the + * request shape. + */ const rebuild = async (): Promise => { epoch += 1; phase = "buffering"; @@ -238,7 +360,10 @@ export function createStreamController(deps: StreamControllerDeps): StreamContro // the input for a skeleton, losing scroll position, expanded tool cards, and composer // focus/draft.) Deltas arriving during the refetch are buffered and replayed on swap; the brief // no-new-text pause is invisible next to a full teardown. - await load(epoch, createStreamModel(localDecisions)); + await load(epoch, createStreamModel(localDecisions), { + page: { kind: "tail", limit: TAIL_UNITS }, + splice: prefixItems.length > 0, + }); }; /** @@ -274,15 +399,45 @@ export function createStreamController(deps: StreamControllerDeps): StreamContro return [...pre, ...seeds, ...post]; }; - const load = async (currentEpoch: number, freshModel?: StreamModel): Promise => { + const load = async ( + currentEpoch: number, + freshModel?: StreamModel, + opts: { page?: MessagesPageQuery; splice?: boolean } = {}, + ): Promise => { try { - const { messages, live, serverNowMs } = await deps.loadMessages(); + let res = await deps.loadMessages(opts.page); if (disposed || currentEpoch !== epoch) return; + if (opts.splice === true) { + // The resync decision tree's prefix-retained branches (see rebuild): splice only + // on exact cursor continuity; a beginning-reaching tail supersedes the prefix; + // everything else falls back to the full read within this same epoch (events + // keep buffering meanwhile). + const start = res.page?.before ?? null; + if (res.page !== undefined && start !== null && start === tailStartCursor) { + // Continuity proven: keep prefix and paging state exactly as they are. + } else if (res.page !== undefined && start === null) { + adoptTailPage(res.page); // tail covers everything: prefix dropped, provably complete + } else { + res = await deps.loadMessages(); + if (disposed || currentEpoch !== epoch) return; + adoptTailPage(undefined); // full transcript: no prefix, no cursors + } + } else if (opts.page !== undefined) { + // Fresh tail baseline (initial load / retry): any previously-loaded prefix is + // superseded by the new window chain. + adoptTailPage(res.page); + } else { + adoptTailPage(undefined); + } + const { messages, live, serverNowMs } = res; // Rebuild path: make the freshly-built model visible only now, atomically — the old model // stayed on screen throughout the refetch above (see rebuild). Initial load / retry pass no // freshModel and keep operating on the current model. if (freshModel) model = freshModel; const target = model; + // Windowed loads seed the stats accrued before the window, so header chips and + // per-turn cumulative rows equal a full load (see seedPriorStats). + if (res.page !== undefined) seedPriorStats(target.stats, res.page.prior); pushMessages(target, messages, now(), serverNowMs ?? null); const dedup = buildDedupIndex(messages, 100); // Replay the buffer (events that arrived while fetching history), with dedup; while a @@ -322,16 +477,79 @@ export function createStreamController(deps: StreamControllerDeps): StreamContro } }; + /** + * Scroll-up backfill: fetch the window before the oldest loaded one and prepend it. + * The window's messages build a FRESH model (its own negative id base, priors seeded, + * finalizeHistory closing its last Task — complete by construction, since a newer + * window follows), whose items freeze into the prefix. Guarded to the live phase: a + * rebuild in flight owns the loading pipeline, and its epoch bump discards any + * backfill that raced it. + */ + const loadOlder = async (): Promise => { + if (disposed || phase !== "live" || failed) return; + if (older.loading || !older.hasMore || nextBefore === null) return; + const currentEpoch = epoch; + older.loading = true; + older.error = null; + deps.onModelChange(); + try { + const res = await deps.loadMessages({ + kind: "before", + cursor: nextBefore, + limit: OLDER_UNITS, + }); + if (disposed || currentEpoch !== epoch) return; + // A before-request against a server without windowing support would return the + // full transcript with no envelope; prepending that would duplicate history. + if (res.page === undefined) throw new Error("windowed history not supported"); + prependCount += 1; + const m = createStreamModel(localDecisions); + m.nextItemId = -prependCount * PREPEND_ID_SPAN; + seedPriorStats(m.stats, res.page.prior); + pushMessages(m, res.messages, now(), null); + finalizeHistory(m); + prefixItems = [...m.items, ...prefixItems]; + // Child sessions live in exactly one window (a spawn is contained in its Task), + // so the merge is disjoint; newer windows' entries win defensively on a clash. + const mergedSubagents = new Map(m.subagents); + for (const [sid, sub] of prefixSubagents) mergedSubagents.set(sid, sub); + prefixSubagents = mergedSubagents; + nextBefore = res.page.before ?? null; + older.hasMore = res.page.before !== undefined; + outlineOffset = res.page.earlierTurns; + } catch (e) { + if (disposed || currentEpoch !== epoch) return; + older.error = e instanceof Error ? e.message : String(e); + } finally { + if (!disposed && currentEpoch === epoch) { + older.loading = false; + deps.onModelChange(); + } + } + }; + return { get model() { return model; }, + get prefixItems(): readonly ChatItem[] { + return prefixItems; + }, + get prefixSubagents(): ReadonlyMap { + return prefixSubagents; + }, + get outlineOffset() { + return outlineOffset; + }, + get older(): OlderHistoryState { + return older; + }, get pendingApprovals(): ReadonlyMap { return pending; }, load: () => { epoch += 1; - return load(epoch); + return load(epoch, undefined, { page: { kind: "tail", limit: TAIL_UNITS } }); }, retry: async () => { if (disposed || !failed) return; @@ -341,11 +559,15 @@ export function createStreamController(deps: StreamControllerDeps): StreamContro // history straight onto it would duplicate the entire conversation (pushMessages appends with // no id-dedup). load() swaps the fresh model in only once the refetch succeeds, then replays // the still-accumulating buffer into it; localDecisions carry over via the shared set. + // The refetch is a fresh TAIL baseline: any retained prefix is superseded on success. deps.onError(null); deps.onLoading(true); epoch += 1; - await load(epoch, createStreamModel(localDecisions)); + await load(epoch, createStreamModel(localDecisions), { + page: { kind: "tail", limit: TAIL_UNITS }, + }); }, + loadOlder, handleOmni: (msg, eventId = null) => { if (disposed) return; if (eventId !== null) lastEventId = eventId; diff --git a/packages/web/src/lib/omni/task-stats.ts b/packages/web/src/lib/omni/task-stats.ts index 8a5ecbc..96c7c12 100644 --- a/packages/web/src/lib/omni/task-stats.ts +++ b/packages/web/src/lib/omni/task-stats.ts @@ -139,6 +139,33 @@ export function createTaskStatsTracker(): TaskStatsTracker { }; } +/** + * Seed the tracker with the cumulative stats accrued BEFORE a partial history window + * (MessagesPageInfo.prior from a windowed GET /messages): finished-Task elapsed, + * subagent token totals, and the last session/context token readings. Applied before + * the window's messages replay, so header chips and per-turn cumulative rows land on + * the same figures a full-transcript load computes. sessionTotal/contextNow are only + * "last seen" fallbacks — any token_usage inside the window overwrites them; the + * elapsed/subagent/delta baselines genuinely accumulate on top. + */ +export function seedPriorStats( + t: TaskStatsTracker, + prior: { + subagentTokens: number; + elapsedMs: number; + sessionTokens: number; + contextTokens: number; + }, +): void { + t.subagentTotal = prior.subagentTokens; + t.sessionElapsedMs = prior.elapsedMs; + t.sessionTotal = prior.sessionTokens; + t.contextNow = prior.contextTokens; + // The first in-window stats row's context delta measures against the pre-window + // occupancy, not against zero. + t.contextAtLastStats = prior.contextTokens; +} + /** This Task's bucketed accumulation (shared by main + subagent sessions; for real-time cost calc). */ function addBuckets(t: TaskStatsTracker, p: TokenUsagePayload): void { t.taskCacheRead += p.request.cache_read; diff --git a/packages/web/src/lib/strings-en.ts b/packages/web/src/lib/strings-en.ts index 1975b55..7d806f9 100644 --- a/packages/web/src/lib/strings-en.ts +++ b/packages/web/src/lib/strings-en.ts @@ -758,6 +758,12 @@ Scenarios: statusCompacting: "Compacting", pendingApprovals: (n: number) => `${n} pending approval${n > 1 ? "s" : ""}`, jumpToLatest: "Jump to latest", + /** Top-of-stream affordance while the previous history window is being fetched (scroll-up backfill). */ + loadingEarlier: "Loading earlier messages…", + /** Top-of-stream affordance after a backfill failure: click to retry fetching the previous window. */ + loadEarlierRetry: "Failed to load earlier messages — click to retry", + /** Top-of-stream marker once the loaded history reaches the very beginning (shown only after a backfill happened). */ + historyBeginning: "Beginning of conversation", /** Conversation minimap (tick rail over the stream's left gutter): rail aria-label. */ outlineTitle: "Outline", /** Tick accessible name: turn number + the question (or the no-text placeholder). */ diff --git a/packages/web/src/lib/strings.ts b/packages/web/src/lib/strings.ts index 806a0a4..3f288d3 100644 --- a/packages/web/src/lib/strings.ts +++ b/packages/web/src/lib/strings.ts @@ -736,6 +736,12 @@ Benchmark： statusCompacting: "压缩中", pendingApprovals: (n: number) => `${n} 个待审批`, jumpToLatest: "回到最新消息", + /** Top-of-stream affordance while the previous history window is being fetched (scroll-up backfill). */ + loadingEarlier: "正在加载更早的对话…", + /** Top-of-stream affordance after a backfill failure: click to retry fetching the previous window. */ + loadEarlierRetry: "更早的对话加载失败，点击重试", + /** Top-of-stream marker once the loaded history reaches the very beginning (shown only after a backfill happened). */ + historyBeginning: "已是对话开头", /** Conversation minimap (tick rail over the stream's left gutter): rail aria-label. */ outlineTitle: "对话索引", /** Tick accessible name: turn number + the question (or the no-text placeholder). */ diff --git a/packages/web/test/outline-model.test.ts b/packages/web/test/outline-model.test.ts index 0cab3cd..2936360 100644 --- a/packages/web/test/outline-model.test.ts +++ b/packages/web/test/outline-model.test.ts @@ -8,11 +8,14 @@ import type { ChatItem } from "../src/lib/omni/stream-model"; import { handoffMessage } from "../src/features/chat/agent-handoff"; import { buildScheduledMessage } from "@prismshadow/penguin-core/markers"; import { + OUTLINE_MIN_TURNS, OUTLINE_WINDOW_AFTER, OUTLINE_WINDOW_BEFORE, TICK_PITCH_MAX, TICK_PITCH_MIN, buildOutline, + globalTurnNumber, + outlineVisible, previewText, railTickPitch, railWindowHalf, @@ -162,6 +165,35 @@ describe("rail fit (pitch and height-adaptive window half-width)", () => { }); }); +describe("windowed-history offset (globalTurnNumber / outlineVisible)", () => { + it("numbers loaded entries globally: offset (unloaded earlier turns) + loaded index + 1", () => { + // A conversation of 10 turns loaded from turn 8 on (server earlierTurns = 7): the + // three loaded entries must read 第 8/9/10 轮 — the window never renumbers. + const items: ChatItem[] = [ + user("q8"), + assistant("a8"), + stats(), + user("q9"), + assistant("a9"), + stats(), + user("q10"), + ]; + const entries = buildOutline(items); + expect(entries.map((_, i) => globalTurnNumber(7, i))).toEqual([8, 9, 10]); + // Without an offset the numbering degenerates to the classic 1-based index. + expect(globalTurnNumber(0, 0)).toBe(1); + }); + + it("the visibility gate counts the WHOLE conversation, not the loaded window", () => { + // Two loaded entries alone would hide the outline; 7 earlier turns make it a long + // conversation that deserves its index (with the earlier-turns hint shown). + expect(outlineVisible(0, 2)).toBe(false); + expect(outlineVisible(7, 2)).toBe(true); + expect(outlineVisible(0, OUTLINE_MIN_TURNS)).toBe(true); + expect(outlineVisible(OUTLINE_MIN_TURNS, 0)).toBe(true); + }); +}); + describe("previewText", () => { it("flattens markdown to one plain line", () => { const md = diff --git a/packages/web/test/stream-controller.test.ts b/packages/web/test/stream-controller.test.ts index ab8933c..4013e2f 100644 --- a/packages/web/test/stream-controller.test.ts +++ b/packages/web/test/stream-controller.test.ts @@ -19,12 +19,13 @@ import { import type { OmniMessage, TokenCounts } from "@prismshadow/penguin-core/omnimessage"; import type { MessagesLiveTail, + MessagesPageInfo, PendingSteeringInfo, ServerEvent, SessionStatus, } from "@prismshadow/penguin-server/api"; -import { createStreamController } from "../src/lib/omni/stream-controller"; -import type { StreamController } from "../src/lib/omni/stream-controller"; +import { OLDER_UNITS, TAIL_UNITS, createStreamController } from "../src/lib/omni/stream-controller"; +import type { MessagesPageQuery, StreamController } from "../src/lib/omni/stream-controller"; import { approvalKey, findToolCard } from "../src/lib/omni/stream-model"; import type { AssistantTextItem, TaskStatsItem, ToolCallItem } from "../src/lib/omni/stream-model"; @@ -47,10 +48,13 @@ interface Harness { errors: Array; loadings: boolean[]; loadCalls: () => number; + /** The `page` argument of each loadMessages call, in order (undefined = the legacy full read). */ + pageArgs: Array; resolveLoad: ( messages: OmniMessage[], live?: MessagesLiveTail, serverNowMs?: number | null, + page?: MessagesPageInfo, ) => void; rejectLoad: (err: Error) => void; } @@ -61,6 +65,7 @@ function createHarness(): Harness { messages: OmniMessage[]; live?: MessagesLiveTail; serverNowMs?: number | null; + page?: MessagesPageInfo; }) => void; reject: (e: unknown) => void; }> = []; @@ -68,15 +73,18 @@ function createHarness(): Harness { const pendingSteering: PendingSteeringInfo[][] = []; const errors: Array = []; const loadings: boolean[] = []; + const pageArgs: Array = []; let calls = 0; const controller = createStreamController({ - loadMessages: () => + loadMessages: (page) => new Promise<{ messages: OmniMessage[]; live?: MessagesLiveTail; serverNowMs?: number | null; + page?: MessagesPageInfo; }>((resolve, reject) => { calls += 1; + pageArgs.push(page); pendingLoads.push({ resolve, reject }); }), onTaskState: (s) => states.push(s), @@ -94,16 +102,27 @@ function createHarness(): Harness { errors, loadings, loadCalls: () => calls, - resolveLoad: (messages, live, serverNowMs) => + pageArgs, + resolveLoad: (messages, live, serverNowMs, page) => pendingLoads.shift()!.resolve({ messages, ...(live !== undefined ? { live } : {}), ...(serverNowMs !== undefined ? { serverNowMs } : {}), + ...(page !== undefined ? { page } : {}), }), rejectLoad: (err) => pendingLoads.shift()!.reject(err), }; } +/** Shorthand for a page envelope (zeroed priors unless overridden). */ +function pageInfo(over: Partial = {}): MessagesPageInfo { + return { + earlierTurns: 0, + prior: { subagentTokens: 0, elapsedMs: 0, sessionTokens: 0, contextTokens: 0 }, + ...over, + }; +} + const HISTORY_TASK: OmniMessage[] = [ at(userText("question"), "2026-07-05T00:00:00.000Z"), at(assistantText("answer"), "2026-07-05T00:00:03.000Z"), @@ -452,6 +471,198 @@ describe("live-tail seeding (reload mid-stream)", () => { }); }); +describe("windowed history: tail-first load + scroll-up backfill", () => { + const OLD_TURN: OmniMessage[] = [ + at(userText("old question"), "2026-07-04T00:00:00.000Z"), + at(assistantText("old answer"), "2026-07-04T00:00:02.000Z"), + at(tokenUsage(counts(400), counts(400)), "2026-07-04T00:00:04.000Z"), + ]; + + it("initial load requests the TAIL window and seeds the prior stats into the tracker", async () => { + const h = createHarness(); + const p = h.controller.load(); + expect(h.pageArgs[0]).toEqual({ kind: "tail", limit: TAIL_UNITS }); + h.controller.handleServer({ type: "task_state", state: "idle" }); + h.resolveLoad( + HISTORY_TASK, + undefined, + null, + pageInfo({ + before: "2:10", + earlierTurns: 7, + prior: { subagentTokens: 500, elapsedMs: 60_000, sessionTokens: 900, contextTokens: 800 }, + }), + ); + await p; + expect(h.controller.outlineOffset).toBe(7); + expect(h.controller.older).toEqual({ hasMore: true, loading: false, error: null }); + // Header basis: prior elapsed + the loaded turn's own span (usage at +5s of a turn + // starting at 0s); token cumulative = in-window session.total + prior subagent total. + expect(h.controller.model.stats.sessionElapsedMs).toBe(60_000 + 5_000); + const stats = h.controller.model.items.find((i) => i.kind === "task_stats") as TaskStatsItem; + expect(stats.stats!.tokens).toBe(1000 + 500); + }); + + it("loadOlder prepends the previous window: frozen items ahead of the live model, unique ids, closed stats row", async () => { + const h = createHarness(); + const p = h.controller.load(); + h.controller.handleServer({ type: "task_state", state: "idle" }); + h.resolveLoad(HISTORY_TASK, undefined, null, pageInfo({ before: "1:8", earlierTurns: 1 })); + await p; + + const older = h.controller.loadOlder(); + expect(h.pageArgs[1]).toEqual({ kind: "before", cursor: "1:8", limit: OLDER_UNITS }); + // Reaches the beginning: no cursor, offset drops to 0. + h.resolveLoad(OLD_TURN, undefined, null, pageInfo({ earlierTurns: 0 })); + await older; + + // The prepended window renders BEFORE the live model, with its last Task closed + // (finalizeHistory — a newer window follows, so the Task is complete by construction). + expect(h.controller.prefixItems.map((i) => i.kind)).toEqual([ + "user_text", + "assistant_text", + "task_stats", + ]); + expect(h.controller.older).toEqual({ hasMore: false, loading: false, error: null }); + expect(h.controller.outlineOffset).toBe(0); + // Ids stay unique across the concatenated view (negative prefix base vs. positive live ids). + const ids = [...h.controller.prefixItems, ...h.controller.model.items].map((i) => i.id); + expect(new Set(ids).size).toBe(ids.length); + expect(h.controller.prefixItems.every((i) => i.id < 0)).toBe(true); + // Its cumulative stats column carries on into the live window's figures. + const oldStats = h.controller.prefixItems.find((i) => i.kind === "task_stats") as TaskStatsItem; + expect(oldStats.stats!.tokens).toBe(400); + }); + + it("loadOlder without more history, while loading, or before the initial load is a no-op", async () => { + const h = createHarness(); + const early = h.controller.loadOlder(); // buffering phase: ignored + await early; + const p = h.controller.load(); + h.controller.handleServer({ type: "task_state", state: "idle" }); + h.resolveLoad(HISTORY_TASK, undefined, null, pageInfo()); // no before cursor: beginning already loaded + await p; + await h.controller.loadOlder(); + expect(h.loadCalls()).toBe(1); + }); + + it("a failed backfill surfaces on older.error and can be retried", async () => { + const h = createHarness(); + const p = h.controller.load(); + h.controller.handleServer({ type: "task_state", state: "idle" }); + h.resolveLoad(HISTORY_TASK, undefined, null, pageInfo({ before: "1:8", earlierTurns: 1 })); + await p; + const older = h.controller.loadOlder(); + h.rejectLoad(new Error("boom")); + await older; + expect(h.controller.older).toEqual({ hasMore: true, loading: false, error: "boom" }); + expect(h.controller.prefixItems).toHaveLength(0); + const again = h.controller.loadOlder(); + h.resolveLoad(OLD_TURN, undefined, null, pageInfo()); + await again; + expect(h.controller.older.error).toBeNull(); + expect(h.controller.prefixItems.length).toBeGreaterThan(0); + }); +}); + +describe("resync decision tree (windowed history)", () => { + const tailPage = pageInfo({ before: "2:10", earlierTurns: 1 }); + + /** Boots a controller with a tail window + one backfilled prefix window. */ + async function withPrefix(): Promise { + const h = createHarness(); + const p = h.controller.load(); + h.controller.handleServer({ type: "task_state", state: "idle" }); + h.resolveLoad(HISTORY_TASK, undefined, null, tailPage); + await p; + const older = h.controller.loadOlder(); + h.resolveLoad( + [at(userText("old question"), "2026-07-04T00:00:00.000Z")], + undefined, + null, + pageInfo(), + ); + await older; + expect(h.controller.prefixItems.length).toBeGreaterThan(0); + return h; + } + + it("no prefix: resync refetches the TAIL window (initial-load shape)", async () => { + const h = createHarness(); + const p = h.controller.load(); + h.controller.handleServer({ type: "task_state", state: "idle" }); + h.resolveLoad(HISTORY_TASK, undefined, null, pageInfo({ before: "2:10", earlierTurns: 3 })); + await p; + h.controller.handleServer({ type: "resync_required" }); + expect(h.pageArgs[1]).toEqual({ kind: "tail", limit: TAIL_UNITS }); + h.resolveLoad(HISTORY_TASK, undefined, null, pageInfo({ before: "2:10", earlierTurns: 3 })); + await flush(); + expect(h.loadCalls()).toBe(2); + expect(h.controller.outlineOffset).toBe(3); + expect(h.controller.model.items.some((i) => i.kind === "user_text")).toBe(true); + }); + + it("prefix + provable continuity (identical window-start cursor): splice — prefix retained, one fetch", async () => { + const h = await withPrefix(); + h.controller.handleServer({ type: "resync_required" }); + // The refetched tail starts at the SAME cursor the current tail started at: no new + // units since — the new window abuts the prefix exactly. + h.resolveLoad(HISTORY_TASK, undefined, null, tailPage); + await flush(); + expect(h.loadCalls()).toBe(3); // initial + backfill + one resync fetch + expect(h.controller.prefixItems.length).toBeGreaterThan(0); + expect(h.controller.older.hasMore).toBe(false); // backfill already reached the beginning + expect(h.controller.model.items.some((i) => i.kind === "user_text")).toBe(true); + }); + + it("prefix + moved cursor (new units arrived): doubt — falls back to the FULL refetch, prefix dropped", async () => { + const h = await withPrefix(); + h.controller.handleServer({ type: "resync_required" }); + // The tail window slid forward: splicing would leave a gap between prefix and tail. + h.resolveLoad([], undefined, null, pageInfo({ before: "3:0", earlierTurns: 9 })); + await flush(); + // The fallback full read (no page argument) is issued within the same rebuild. + expect(h.pageArgs[h.pageArgs.length - 1]).toBeUndefined(); + h.resolveLoad([at(userText("old question"), "2026-07-04T00:00:00.000Z"), ...HISTORY_TASK]); + await flush(); + expect(h.controller.prefixItems).toHaveLength(0); + expect(h.controller.outlineOffset).toBe(0); + expect(h.controller.older.hasMore).toBe(false); + // The full transcript lives in the single model now. + expect(h.controller.model.items.filter((i) => i.kind === "user_text")).toHaveLength(2); + }); + + it("prefix + tail reaching the beginning: prefix superseded without a second fetch", async () => { + const h = await withPrefix(); + h.controller.handleServer({ type: "resync_required" }); + h.resolveLoad( + [at(userText("old question"), "2026-07-04T00:00:00.000Z"), ...HISTORY_TASK], + undefined, + null, + pageInfo({ earlierTurns: 0 }), + ); + await flush(); + expect(h.loadCalls()).toBe(3); + expect(h.controller.prefixItems).toHaveLength(0); + expect(h.controller.model.items.filter((i) => i.kind === "user_text")).toHaveLength(2); + }); + + it("a legacy full response during resync (no page envelope) resets the windowed state", async () => { + const h = await withPrefix(); + h.controller.handleServer({ type: "resync_required" }); + // A server without windowing support answers the tail request with the full + // transcript and no envelope: doubt — the full-fallback branch also covers it + // (the second fetch returns the same full transcript). + h.resolveLoad([...HISTORY_TASK]); + await flush(); + h.resolveLoad([at(userText("old question"), "2026-07-04T00:00:00.000Z"), ...HISTORY_TASK]); + await flush(); + expect(h.controller.prefixItems).toHaveLength(0); + expect(h.controller.older.hasMore).toBe(false); + expect(h.controller.outlineOffset).toBe(0); + }); +}); + describe("history load failure and retry (#6)", () => { it("failure surfaces the error and stops loading; retry keeps the buffer (snapshot and initial events are not lost)", async () => { const h = createHarness();