feat: read long conversations by window with shared history writes

.agents/notes/implemented/architecture/2026-09-10-windowed-reader-integration.md

@@ -0,0 +1,47 @@
+# Windowed conversation reads
+
+Status: implemented
+Translation: current
+
+## Abstract
+
+Opening a session uses a control-plane Mirror and one reader-backed
+ConversationView. Directory rows are shallow; only leased or retained tail
+bodies are materialized. Snapshot decoding and the initial directory still
+scale with the whole conversation.
+
+The existing HistoryWriter remains the single writer. Shared planners retain
+permission, agent-output, import and editable-tail rules. Stored-copy handles
+reuse the writer's provenance; they survive source disposal during a fork.
+
+View events are structure changes or explicit changed turn ids. Derivations
+invalidate evicted facts too. Async cache reads retain identity/epoch fences.
+Storage and view events share structure/changed-id semantics. Goal, permission,
+scheduling and file-diff consumers share one reference-counted fact table.
+CLI reads are synchronous over the same storage reader; auto-seen reads only
+shallow fields and permission decisions are checked before auto-approval.
+Ordinary commands throw writer errors; only import and editable-tail replacement
+retain phased outcomes. The array adapter serves static sharing pages. There is
+no alternate session view, feature switch or complete memory command backend.
+
+## Evidence boundary
+
+Reader, writer and CLI regressions use synthetic Loro fixtures. The component
+benchmark measures the shipped reader. Device-scale cold-open, streaming frame
+time and long-session JS/WASM memory remain separate acceptance work.
+
+## Initial viewport measurement
+
+The first range promise only establishes data availability. Setting `scrollTop`
+to the estimated end does not establish that Virtua has mounted and measured the
+destination rows. Revealing at that point exposed an empty or intermediate window.
+The viewport now waits for the measured destination, visible row geometry and
+Virtua's observed offset to agree; mounted-row measurements also correct following
+before deferred spacer resizes. Corrections use the sticky library's setter so
+layout changes do not masquerade as upward user scrolling. There is no settle
+sleep, and later window hydration does not hide the viewport again.
+
+A browser regression holds the real destination-row ResizeObserver delivery:
+the previous hook reveals while held; the fixed hook stays hidden and opens at
+the measured tail after release, including remount. Unit coverage retains cached
+reading positions, user escape, composer resize suppression and row growth.

.agents/notes/implemented/architecture/2026-09-10-windowed-reader-integration.zh.md

@@ -0,0 +1,33 @@
+# 窗口读取对话历史
+
+Status: implemented
+Translation: current
+
+## 摘要
+
+打开会话时，控制面 Mirror 不读取历史正文。唯一的窗口读取实现先读浅目录，
+再读取窗口和尾部正文；快照解码、初始目录仍与总历史长度有关。
+
+HistoryWriter 仍是唯一写入者，共享 planner 保留业务规则。fork 直接持有
+writer 的可信快照句柄，源文档卸载不会使已捕获的快照失效。
+
+显示层只通知结构变化或明确的变更轮次 ID，逐出的事实也会失效。
+异步读取保留身份和版本检查；存储层和显示层使用相同的结构/变更 ID 事件。
+目标、权限、调度和文件 diff 共用一张有引用计数的事实表。
+CLI 同步读取相同存储；auto-seen 只读浅字段，权限决定先于自动批准检查。
+普通命令直接抛出 writer 错误，仅导入和编辑重发保留三相结果。
+数组适配器用于静态分享页，不再保留备用会话路径或开关。
+测试和基准使用正式 reader；真机冷开、流式帧耗时和内存仍需单独验收。
+
+## 首屏视口测量
+
+首个范围的 promise 只保证数据可用。把 `scrollTop` 写到估算底部，并不保证
+Virtua 已挂载、测量目标行；此时显示视口会暴露空白或中间位置。
+现在会等待目标行测量、可见行几何和 Virtua 观察到的偏移一致。
+已挂载行的尺寸变化也会在延后的占位容器 resize 前完成贴底修正。
+修正通过吸附库的 setter 写入，避免把布局变化误判为用户向上滚动。
+没有固定等待时间，后续窗口加载也不会重新隐藏视口。
+
+浏览器回归显式暂停目标行真实的 ResizeObserver 通知：旧 hook 在暂停时
+已经显示，修复后保持隐藏，放行测量后直接显示末尾；重新挂载也覆盖。
+单元回归保留历史阅读位置、用户脱离吸附、输入框高度抑制和行高增长检查。

apps/cli/AGENTS.md

@@ -7,6 +7,10 @@ Root `AGENTS.md` applies; this file adds CLI context. Build, PR-poller, and adap
 
 ## Build and packaging
 
+- The Node 22 bundle uses native top-level await. Do not run a browser TLA
+  compatibility transform over its output chunks. Validate the CLI SSR build
+  under `NODE_OPTIONS=--max-old-space-size=2048`; increasing the heap is not a fix.
+
 - The public CLI defaults to the local platform, discovers no deployment dotenv files, and must
   never initialize telemetry in local mode even if PostHog variables exist in the shell.
 - INVARIANT: the dev output layout must match production's — `index.js` plus flat sibling

apps/cli/src/commands/session-output.test.ts

@@ -1,3 +1,8 @@
+import { createSessionAgentWrites } from '../lib/loro/session-agent-writes';
+import { LoroDoc, LoroMap } from 'loro-crdt';
+import { createHistoryWriter } from '@lody/shared';
+import { createLoroSessionData } from '@lody/shared/session-data';
+import { withHistoryPort } from '../../tests/history-port-fixture';
 import { describe, expect, it, vi } from 'vitest';
 import type { SessionHistoryInput, SessionId } from '@lody/shared';
 import {
@@ -41,10 +46,12 @@ const createMirror = (initialState: MirrorState) => {
   };
 };
 
-const createSessionDoc = (mirror: ReturnType<typeof createMirror>) => ({
-  sessionId: 'session-1' as SessionId,
-  mirror: mirror as unknown,
-});
+const createSessionDoc = (mirror: ReturnType<typeof createMirror>) =>
+  withHistoryPort({
+    sessionId: 'session-1' as SessionId,
+    readHistorySnapshot: () => mirror.getState().history ?? [],
+    subscribeAll: (listener: () => void) => mirror.subscribe(() => listener()),
+  });
 
 describe('session output helpers', () => {
   it('finds the assistant entry linked to the target user turn', () => {
@@ -72,6 +79,73 @@ describe('session output helpers', () => {
     expect(findAssistantEntryForUserTurn(history, 'user-1')?.id).toBe('assistant-target');
   });
 
+  it('streams one linked turn without materializing unrelated bodies', async () => {
+    const doc = new LoroDoc();
+    const writer = createHistoryWriter(doc);
+    for (let i = 0; i < 100; i++)
+      writer.append(
+        createHistoryEntry({
+          id: `old-${i}`,
+          items: [{ type: 'text', text: 'large old body'.repeat(100) }],
+        })
+      );
+    writer.append(createHistoryEntry({ id: 'u', role: 'user', status: 'processing' }));
+    writer.append(
+      createHistoryEntry({
+        id: 'a',
+        userTurnId: 'u',
+        finished: false,
+        items: [{ type: 'text', text: 'first' }],
+      })
+    );
+    const data = createLoroSessionData({
+      sessionId: 'session-1' as SessionId,
+      doc,
+      writer,
+    });
+    const bodyReads: string[] = [];
+    const toJSON = LoroMap.prototype.toJSON;
+    const spy = vi.spyOn(LoroMap.prototype, 'toJSON').mockImplementation(function (this: LoroMap) {
+      const id = this.get('id');
+      if (typeof id === 'string') bodyReads.push(id);
+      return toJSON.call(this);
+    });
+    const first = Promise.withResolvers<void>();
+    const events: Array<Record<string, unknown>> = [];
+    const completion = waitForTurnCompletion({
+      sessionDoc: {
+        sessionId: 'session-1' as SessionId,
+        sessionData: data,
+        subscribeAll: (notify) => doc.subscribe(() => notify()),
+      },
+      userTurnId: 'u',
+      outputMode: 'jsonl',
+      timeoutMs: 0,
+      onEvent(event) {
+        events.push(event);
+        if (event.type === 'update') first.resolve();
+      },
+    });
+    try {
+      await first.promise;
+      await createSessionAgentWrites(data.writer).setTurnField('a', 'finished', {
+        kind: 'set',
+        value: true,
+      });
+      await createSessionAgentWrites(data.writer).setTurnField('u', 'status', {
+        kind: 'set',
+        value: 'handled',
+      });
+      expect((await completion).turnId).toBe('a');
+      expect(events.map((e) => e.type)).toEqual(['update', 'done']);
+      expect(bodyReads.length).toBeGreaterThan(0);
+      expect([...new Set(bodyReads)]).toEqual(['a']);
+    } finally {
+      spy.mockRestore();
+      data.dispose();
+    }
+  });
+
   it('streams updated assistant items and resolves when the turn finishes', async () => {
     const userTurn = createHistoryEntry({
       id: 'user-1',

apps/cli/src/commands/session-output.ts

@@ -3,16 +3,7 @@ import type { SessionDocument } from '@/lib/loro/doc';
 
 export type StructuredSessionOutputMode = 'json' | 'jsonl';
 
-type SessionDocMirrorState = {
-  history?: SessionHistoryInput[];
-};
-
-type SessionDocMirror = {
-  subscribe: (listener: (next: SessionDocMirrorState) => void) => () => void;
-  getState: () => SessionDocMirrorState;
-};
-
-type SessionDocForOutput = Pick<SessionDocument, 'sessionId' | 'mirror'>;
+type SessionDocForOutput = Pick<SessionDocument, 'sessionId' | 'sessionData' | 'subscribeAll'>;
 
 export type SessionTurnOutputEvent =
   | {
@@ -138,8 +129,7 @@ export async function waitForTurnCompletion(options: {
   signal?: AbortSignal;
   onEvent?: (event: SessionTurnOutputEvent) => void;
 }): Promise<CompletedAssistantTurn> {
-  const mirror = options.sessionDoc.mirror as SessionDocMirror | null;
-  if (!mirror) {
+  if (!options.sessionDoc.sessionData || !options.sessionDoc.subscribeAll) {
     throw new Error('SessionDocument not initialized');
   }
 
@@ -175,12 +165,11 @@ export async function waitForTurnCompletion(options: {
       settle(() => reject(error));
     };
 
-    const inspect = (next: SessionDocMirrorState) => {
+    const inspect = (history: SessionHistoryInput[]) => {
       if (settled) {
         return;
       }
 
-      const history = Array.isArray(next.history) ? (next.history as SessionHistoryInput[]) : [];
       const userTurn = findUserTurn(history, options.userTurnId);
       if (userTurn?.status === 'failed') {
         rejectWith(
@@ -256,13 +245,21 @@ export async function waitForTurnCompletion(options: {
       }
     };
 
+    // The in-process reader captures and inspects one observation synchronously.
+    const refresh = () => {
+      if (settled) return;
+      try {
+        inspect(options.sessionDoc.sessionData.history.readTurnOutput(options.userTurnId));
+      } catch (error) {
+        rejectWith(error instanceof Error ? error : new Error(String(error)));
+      }
+    };
+
     const handleAbort = () => {
       rejectWith(new Error('Turn completion wait aborted.'));
     };
 
-    unsubscribe = mirror.subscribe((next) => {
-      inspect(next);
-    });
+    unsubscribe = options.sessionDoc.subscribeAll(refresh);
     options.signal?.addEventListener('abort', handleAbort, { once: true });
 
     if (options.timeoutMs > 0) {
@@ -280,6 +277,6 @@ export async function waitForTurnCompletion(options: {
       }, options.timeoutMs);
     }
 
-    inspect(mirror.getState());
+    refresh();
   });
 }