Merge remote-tracking branch 'github/master' into xtr/session-log-read-api

packages/api/session-controller/src/client/contract/session.ts

@@ -119,6 +119,15 @@ export interface ISession {
    * @returns completion; failures land in snapshot.openState/loadingOlder.
    */
   loadOlder(): Promise<void>
+  /**
+   * Page history backwards until the window covers `seq` (inclusive) — the
+   * turn-jump loader. Repeated calls while a jump is paging lower its shared
+   * target and return the in-flight completion; `snapshot.loadingOlder` is
+   * the busy signal for the whole jump.
+   * @param seq - durable event seq the window must reach (a turn's `turn/start` seq).
+   * @returns completion once covered, exhausted, superseded, or failed soft.
+   */
+  loadThrough(seq: number): Promise<void>
   /**
    * Execute one slash-command line against this session's agent — pure
    * admission semantics (the host executor durably logs the lifecycle).

packages/api/session-controller/src/client/sessions/session.ts

@@ -38,6 +38,9 @@ import { SessionQueueMirror } from './queue-mirror.ts'
 /** Messages requested per history page. */
 export const PAGE_MESSAGES = 50
 
+/** Messages requested per page while a turn jump loops backwards (fewer, larger round trips). */
+export const JUMP_PAGE_MESSAGES = 200
+
 /** Manager-owned observers of a Session object's local state edges. */
 export interface SessionOptions {
   /** Catalog-discovered address selecting non-activating subagent transport. */
@@ -78,6 +81,10 @@ export class Session implements SessionFace {
    *  passes drop all writes once the generation moves on. */
   private openGeneration = 0
   private loadingOlder = false
+  /** Shared low-water target of the running jump loop; null when no jump is paging. */
+  private jumpTargetSeq: number | null = null
+  /** The running jump loop's completion, shared by retargeting callers. */
+  private jumpPromise: Promise<void> | null = null
   /** Authoritative stream-only inbox snapshot; pending work never hits history. */
   private readonly queueMirror = new SessionQueueMirror()
   private running = false
@@ -369,6 +376,52 @@ export class Session implements SessionFace {
     }
   }
 
+  /** Jump loader: page backwards until the window covers seq (see ISession.loadThrough). */
+  loadThrough(seq: number): Promise<void> {
+    if (this.openState !== 'open' || !this.hasMore || this.baseSeq <= seq) return Promise.resolve()
+    if (this.jumpPromise !== null) {
+      // Retarget the running loop to the lowest requested seq.
+      this.jumpTargetSeq = Math.min(this.jumpTargetSeq ?? seq, seq)
+      return this.jumpPromise
+    }
+    // A plain single-page pull owns the busy flag; the jump does not queue
+    // behind it (the caller retries once it settles) and must leave no
+    // target behind — only the loop's finally clears that field, and no
+    // loop starts here.
+    if (this.loadingOlder) return Promise.resolve()
+    this.jumpTargetSeq = seq
+    this.loadingOlder = true
+    this.notifier.markDirty()
+    // Stale-pass guard (the doOpen pattern): a resync mid-loop replaces the
+    // stream generation; this pass then stops instead of paging the new
+    // generation toward its old target.
+    const generation = this.openGeneration
+    this.jumpPromise = (async () => {
+      try {
+        while (this.hasMore && this.jumpTargetSeq !== null && this.baseSeq > this.jumpTargetSeq) {
+          if (generation !== this.openGeneration) return
+          const events = this.events
+          if (events === undefined) return
+          const before = this.baseSeq
+          await events.prepend({ beforeSeq: this.baseSeq, maxMessages: JUMP_PAGE_MESSAGES })
+          // No-progress guard: an empty or dropped page that still claims more
+          // history must end the loop, not spin it.
+          if (this.baseSeq >= before) return
+        }
+      } catch (error) {
+        if (!isRemoteFailure(error)) {
+          console.error('[session-controller] loadThrough failed:', error)
+        }
+      } finally {
+        this.jumpTargetSeq = null
+        this.jumpPromise = null
+        this.loadingOlder = false
+        this.notifier.markDirty()
+      }
+    })()
+    return this.jumpPromise
+  }
+
   /** Rebuild an opened history source after address replacement.
    *  Invalidates any in-flight open first; queue state belongs to the independently
    *  reconnecting control stream and remains untouched. */