77 * compositing into the live transcript's scrollback. It renders a parked
88 * subagent / advisor / collab-guest transcript that has no live in-view session.
99 *
10- * The transcript is rebuilt from scratch on every refresh ({@link ChatTranscriptBuilder.rebuild})
11- * rather than synced incrementally, so a growing file-backed transcript (the
12- * advisor appends while you watch) can never duplicate or misorder rows. Scroll
13- * is owned end-to-end by a single {@link ScrollView}; the viewer follows the tail
14- * until the reader scrolls up.
10+ * The viewer tails append-only JSONL when possible and falls back to a full
11+ * compaction-aware rebuild when file identity changes, content is replaced, or a
12+ * structural session entry arrives. Scroll is owned end-to-end by a single
13+ * {@link ScrollView}; the viewer follows the tail until the reader scrolls up.
1514 *
16- * Local agents re- read the whole session file whenever its size or mtime changes
17- * (covering SessionManager's in-place rewrites, not just appends) . Collab guests
18- * keep the incremental byte cursor the host's capped `readTranscript` requires
19- * and rebuild components from the accumulated entries .
15+ * Local agents read only newly appended bytes for normal writes while preserving
16+ * an incomplete trailing JSONL line across polls . Collab guests keep the
17+ * incremental byte cursor the host's capped `readTranscript` requires and clear
18+ * stale rows when the host reports rotation .
2019 */
2120import * as fs from "node:fs" ;
22- import type { AgentTool } from "@oh-my-pi/pi-agent-core" ;
21+ import type { AgentMessage , AgentTool } from "@oh-my-pi/pi-agent-core" ;
2322import { type Component , Editor , matchesKey , parseSgrMouse , ScrollView , type TUI } from "@oh-my-pi/pi-tui" ;
2423import { formatDuration , formatNumber , logger } from "@oh-my-pi/pi-utils" ;
2524import type { KeyId } from "../../config/keybindings" ;
2625import type { MessageRenderer } from "../../extensibility/extensions/types" ;
2726import type { AgentLifecycleManager } from "../../registry/agent-lifecycle" ;
2827import type { AgentRegistry , AgentStatus } from "../../registry/agent-registry" ;
29- import type { FileEntry , SessionMessageEntry } from "../../session/session-entries" ;
28+ import { buildSessionContext } from "../../session/session-context" ;
29+ import type { FileEntry , SessionEntry } from "../../session/session-entries" ;
3030import { parseSessionEntries } from "../../session/session-loader" ;
3131import type { ObservableSession , SessionObserverRegistry } from "../session-observer-registry" ;
3232import { getEditorTheme , theme } from "../theme/theme" ;
@@ -64,6 +64,27 @@ export interface AgentTranscriptViewerDeps {
6464/** How often to re-stat a file-backed transcript for growth (advisor/live tail). */
6565const POLL_MS = 250 ;
6666
67+ type LocalTailState = {
68+ path : string ;
69+ dev : number ;
70+ ino : number ;
71+ size : number ;
72+ mtimeMs : number ;
73+ ctimeMs : number ;
74+ offset : number ;
75+ pending : string ;
76+ } ;
77+
78+ function splitCompleteJsonl ( text : string ) : { complete : string ; pending : string } {
79+ const lastNewline = text . lastIndexOf ( "\n" ) ;
80+ if ( lastNewline < 0 ) return { complete : "" , pending : text } ;
81+ return { complete : text . slice ( 0 , lastNewline + 1 ) , pending : text . slice ( lastNewline + 1 ) } ;
82+ }
83+
84+ function isSessionEntry ( entry : FileEntry ) : entry is SessionEntry {
85+ return entry . type !== "session" ;
86+ }
87+
6788function statusBadge ( status : AgentStatus ) : string {
6889 switch ( status ) {
6990 case "running" :
@@ -85,10 +106,12 @@ export class AgentTranscriptViewer implements Component {
85106 #notice: string | undefined ;
86107 #expanded = false ;
87108
88- // Local file transcript state: re-read when the file size or mtime changes.
89- #lastSignature = "" ;
109+ // Local file transcript state: append-tail same-inode growth; rebuild on replacement.
110+ #localState: LocalTailState | undefined ;
111+ #localEmptyReason: "none" | "missing" | undefined ;
90112 // Remote transcript state (incremental; the host caps each read).
91- #remoteEntries: SessionMessageEntry [ ] = [ ] ;
113+ #remoteEntries: FileEntry [ ] = [ ] ;
114+ #remotePending = "" ;
92115 #remoteBytes = 0 ;
93116 #remoteFetchInFlight = false ;
94117 #remoteToken = 0 ;
@@ -145,7 +168,7 @@ export class AgentTranscriptViewer implements Component {
145168 // Transcript loading
146169 // ========================================================================
147170
148- /** Re-read the transcript and rebuild components when it changed . */
171+ /** Tail the transcript and rebuild components only when necessary . */
149172 #refresh( ) : void {
150173 if ( this . #disposed) return ;
151174 if ( this . deps . remote ) {
@@ -154,39 +177,99 @@ export class AgentTranscriptViewer implements Component {
154177 }
155178 const sessionFile = this . deps . registry . get ( this . deps . agentId ) ?. sessionFile ;
156179 if ( ! sessionFile ) {
157- if ( this . #lastSignature !== "none" ) {
158- this . #lastSignature = "none" ;
159- this . #rebuild( [ ] ) ;
180+ if ( this . #localEmptyReason !== "none" ) {
181+ this . #localState = undefined ;
182+ this . #localEmptyReason = "none" ;
183+ this . #model = undefined ;
184+ this . #rebuildMessages( [ ] ) ;
160185 }
161186 return ;
162187 }
163- let signature : string ;
188+ let stat : fs . Stats ;
164189 try {
165- const stat = fs . statSync ( sessionFile ) ;
166- // Include the path: a different file with the same size/mtime must not alias.
167- signature = `${ sessionFile } :${ stat . size } :${ stat . mtimeMs } ` ;
190+ stat = fs . statSync ( sessionFile ) ;
168191 } catch {
169- // File deleted/rotated while open (e.g. the owning session was dropped):
170- // clear stale content once instead of freezing on it forever.
171- if ( this . #lastSignature !== "missing" ) {
172- this . #lastSignature = "missing" ;
192+ if ( this . #localEmptyReason !== "missing" ) {
193+ this . #localState = undefined ;
194+ this . #localEmptyReason = "missing" ;
173195 this . #model = undefined ;
174- this . #rebuild ( [ ] ) ;
196+ this . #rebuildMessages ( [ ] ) ;
175197 }
176198 return ;
177199 }
178- if ( signature === this . #lastSignature) return ;
179- let text : string ;
200+ this . #localEmptyReason = undefined ;
201+ const state = this . #localState;
202+ const identityChanged = ! state || state . path !== sessionFile || state . dev !== stat . dev || state . ino !== stat . ino ;
203+ const contentReplaced =
204+ state &&
205+ state . path === sessionFile &&
206+ state . dev === stat . dev &&
207+ state . ino === stat . ino &&
208+ stat . size === state . size &&
209+ ( stat . mtimeMs !== state . mtimeMs || stat . ctimeMs !== state . ctimeMs ) ;
210+ if ( identityChanged || stat . size < ( state ?. offset ?? 0 ) || contentReplaced ) {
211+ this . #loadLocalFull( sessionFile , stat ) ;
212+ return ;
213+ }
214+ if ( ! state || stat . size === state . offset ) return ;
215+ let fd : number | undefined ;
180216 try {
181- text = fs . readFileSync ( sessionFile , "utf-8" ) ;
217+ fd = fs . openSync ( sessionFile , "r" ) ;
218+ const length = stat . size - state . offset ;
219+ const buffer = Buffer . allocUnsafe ( length ) ;
220+ fs . readSync ( fd , buffer , 0 , length , state . offset ) ;
221+ const { complete, pending } = splitCompleteJsonl ( state . pending + buffer . toString ( "utf-8" ) ) ;
222+ const entries = complete ? parseSessionEntries ( complete ) : [ ] ;
223+ this . #localState = {
224+ ...state ,
225+ size : stat . size ,
226+ mtimeMs : stat . mtimeMs ,
227+ ctimeMs : stat . ctimeMs ,
228+ offset : stat . size ,
229+ pending,
230+ } ;
231+ const incremental = this . #incrementalMessages( entries ) ;
232+ if ( incremental ) {
233+ this . #appendMessages( incremental ) ;
234+ } else {
235+ this . #loadLocalFull( sessionFile , stat ) ;
236+ }
237+ } catch ( err ) {
238+ logger . debug ( "transcript viewer: append read failed" , { err : String ( err ) } ) ;
239+ } finally {
240+ if ( fd !== undefined ) fs . closeSync ( fd ) ;
241+ }
242+ }
243+
244+ #loadLocalFull( sessionFile : string , stat : fs . Stats ) : void {
245+ let data : Buffer ;
246+ try {
247+ data = fs . readFileSync ( sessionFile ) ;
182248 } catch ( err ) {
183- // Leave #lastSignature unchanged so a transient read error retries next poll.
184249 logger . debug ( "transcript viewer: read failed" , { err : String ( err ) } ) ;
185250 return ;
186251 }
187- this . #lastSignature = signature ;
252+ const { complete, pending } = splitCompleteJsonl ( data . toString ( "utf-8" ) ) ;
253+ const entries = complete ? parseSessionEntries ( complete ) : [ ] ;
188254 this . #model = undefined ;
189- this . #rebuild( this . #extractMessages( parseSessionEntries ( text ) ) ) ;
255+ this . #scanModel( entries ) ;
256+ this . #rebuildMessages( this . #messagesFromEntries( entries ) ) ;
257+ let nextStat = stat ;
258+ try {
259+ nextStat = fs . statSync ( sessionFile ) ;
260+ } catch {
261+ nextStat = stat ;
262+ }
263+ this . #localState = {
264+ path : sessionFile ,
265+ dev : nextStat . dev ,
266+ ino : nextStat . ino ,
267+ size : data . byteLength ,
268+ mtimeMs : nextStat . mtimeMs ,
269+ ctimeMs : nextStat . ctimeMs ,
270+ offset : data . byteLength ,
271+ pending,
272+ } ;
190273 }
191274
192275 #fetchRemote( ) : void {
@@ -209,27 +292,39 @@ export class AgentTranscriptViewer implements Component {
209292 return ;
210293 }
211294 if ( result . newSize < fromByte ) {
212- // Host transcript rotated/truncated — restart from 0.
295+ // Host transcript rotated/truncated — clear stale rows and restart from 0.
213296 this . #remoteBytes = 0 ;
297+ this . #remotePending = "" ;
214298 this . #remoteEntries = [ ] ;
299+ this . #hasRemoteData = false ;
300+ this . #rebuildMessages( [ ] ) ;
215301 this . #fetchRemote( ) ;
216302 return ;
217303 }
218304 this . #remoteUnavailable = false ;
219305 const firstData = ! this . #hasRemoteData;
220306 this . #hasRemoteData = true ;
221- const lastNewline = result . text . lastIndexOf ( "\n" ) ;
222- if ( lastNewline >= 0 ) {
223- const completeChunk = result . text . slice ( 0 , lastNewline + 1 ) ;
224- this . #remoteBytes = fromByte + Buffer . byteLength ( completeChunk , "utf-8" ) ;
225- const parsed = this . #extractMessages( parseSessionEntries ( completeChunk ) ) ;
226- if ( parsed . length > 0 ) {
227- this . #remoteEntries. push ( ...parsed ) ;
228- this . #rebuild( this . #remoteEntries) ;
229- return ;
307+ const { complete, pending } = splitCompleteJsonl ( this . #remotePending + result . text ) ;
308+ const parsed = complete ? parseSessionEntries ( complete ) : [ ] ;
309+ this . #remotePending = pending ;
310+ this . #remoteBytes = result . newSize ;
311+ if ( parsed . length > 0 ) {
312+ this . #remoteEntries. push ( ...parsed ) ;
313+ const incremental = this . #incrementalMessages( parsed ) ;
314+ if ( incremental ) {
315+ if ( incremental . length > 0 ) {
316+ this . #appendMessages( incremental ) ;
317+ } else if ( firstData ) {
318+ this . deps . requestRender ( ) ;
319+ }
320+ } else {
321+ this . #model = undefined ;
322+ this . #scanModel( this . #remoteEntries) ;
323+ this . #rebuildMessages( this . #messagesFromEntries( this . #remoteEntries) ) ;
230324 }
325+ return ;
231326 }
232- // First completed fetch (even empty) clears the "Loading…" placeholder.
327+ // First completed fetch (even empty/header-only ) clears the "Loading…" placeholder.
233328 if ( firstData ) this . deps . requestRender ( ) ;
234329 } )
235330 . catch ( ( error : unknown ) => {
@@ -238,22 +333,49 @@ export class AgentTranscriptViewer implements Component {
238333 } ) ;
239334 }
240335
241- /** Filter to message entries, tracking the model from the first assistant / a model_change. */
242- #extractMessages( entries : FileEntry [ ] ) : SessionMessageEntry [ ] {
243- const messages : SessionMessageEntry [ ] = [ ] ;
336+ #messagesFromEntries( entries : readonly FileEntry [ ] ) : AgentMessage [ ] {
337+ const sessionEntries = entries . filter ( isSessionEntry ) ;
338+ return buildSessionContext ( sessionEntries , undefined , undefined , {
339+ transcript : true ,
340+ collapseCompactedHistory : true ,
341+ } ) . messages ;
342+ }
343+
344+ /** Return appendable messages, or undefined when a structural entry requires rebuild. */
345+ #incrementalMessages( entries : readonly FileEntry [ ] ) : AgentMessage [ ] | undefined {
346+ const messages : AgentMessage [ ] = [ ] ;
244347 for ( const entry of entries ) {
348+ if ( entry . type === "session" ) continue ;
245349 if ( entry . type === "message" ) {
246- messages . push ( entry ) ;
350+ messages . push ( entry . message ) ;
247351 if ( ! this . #model && entry . message . role === "assistant" ) this . #model = entry . message . model ;
248352 } else if ( entry . type === "model_change" ) {
249353 this . #model = entry . model ;
354+ } else {
355+ return undefined ;
250356 }
251357 }
252358 return messages ;
253359 }
254360
255- #rebuild( entries : SessionMessageEntry [ ] ) : void {
256- this . #builder. rebuild ( entries ) ;
361+ #scanModel( entries : readonly FileEntry [ ] ) : void {
362+ for ( const entry of entries ) {
363+ if ( entry . type === "message" && ! this . #model && entry . message . role === "assistant" ) {
364+ this . #model = entry . message . model ;
365+ } else if ( entry . type === "model_change" ) {
366+ this . #model = entry . model ;
367+ }
368+ }
369+ }
370+
371+ #rebuildMessages( messages : readonly AgentMessage [ ] ) : void {
372+ this . #builder. rebuildMessages ( messages ) ;
373+ this . deps . requestRender ( ) ;
374+ }
375+
376+ #appendMessages( messages : readonly AgentMessage [ ] ) : void {
377+ if ( messages . length === 0 ) return ;
378+ this . #builder. appendMessages ( messages ) ;
257379 this . deps . requestRender ( ) ;
258380 }
259381
@@ -457,6 +579,7 @@ export class AgentTranscriptViewer implements Component {
457579 #placeholder( ) : string {
458580 if ( this . deps . remote && this . #remoteUnavailable) return "Transcript lives on the host — not available." ;
459581 if ( this . deps . remote && ! this . #hasRemoteData) return "Loading transcript from host…" ;
582+ if ( this . deps . remote ) return "No messages yet." ;
460583 if ( ! this . deps . registry . get ( this . deps . agentId ) ?. sessionFile ) return "No session file available yet." ;
461584 return "No messages yet." ;
462585 }
0 commit comments