fix(peer): bound the relay history observer read and surface first-page stalls

src/crates/assembly/core/src/agentic/coordination/coordinator.rs

@@ -132,7 +132,7 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::sync::OnceLock;
 use tokio::sync::{mpsc, oneshot, watch, OwnedSemaphorePermit, RwLock, Semaphore};
-use tokio::time::{sleep, Duration, Instant};
+use tokio::time::{sleep, timeout, Duration, Instant};
 use tokio_util::sync::CancellationToken;
 
 const MANUAL_COMPACTION_COMMAND: &str = "/compact";
@@ -142,6 +142,10 @@ const DEFAULT_SUBAGENT_MAX_CONCURRENCY: usize = 5;
 const DEFAULT_SWARM_MAX_CONCURRENCY: usize = 16;
 const MAX_SUBAGENT_MAX_CONCURRENCY: usize = 64;
 const SUBAGENT_TIMEOUT_GRACE_PERIOD: Duration = Duration::from_secs(10);
+/// How long an observer read waits for the Session's mutation lease before it
+/// projects persisted records without one. Kept small: the lease owner may be a
+/// Turn that is itself waiting for the reader's answer.
+const RELAY_HISTORY_OBSERVER_LEASE_WAIT: Duration = Duration::from_millis(1500);
 const SESSION_REFERENCES_METADATA_KEY: &str = "sessionReferences";
 const MAX_SESSION_REFERENCES_PER_TURN: usize = 5;
 const SESSION_REFERENCE_ARTIFACT_STEM_LENGTH: usize = 8;
@@ -9461,12 +9465,40 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet
             .persistence_manager()
             .assert_relay_history_supported(storage, session_id)
             .await?;
-        let _mutation = self
+        // A reader here observes persisted history; it is not the Session's
+        // writer. Its owning Turn holds the mutation lease for the Turn's whole
+        // life, so an unbounded acquire would make a controller's first read
+        // depend on that Turn finishing. A Turn blocked on an unanswered
+        // question would then block the very read that lets a controller answer
+        // it. Wait briefly for the consistent window, then fall back to the
+        // persisted records so a reader is never parked behind live execution.
+        let mutation = match timeout(
+            RELAY_HISTORY_OBSERVER_LEASE_WAIT,
+            self.session_manager.acquire_session_mutation(session_id),
+        )
+        .await
+        {
+            Ok(lease) => Some(lease?),
+            Err(_) => None,
+        };
+        if mutation.is_some() {
+            self.prepare_persisted_session_read_locked(storage, session_id)
+                .await?;
+        } else if self
             .session_manager
-            .acquire_session_mutation(session_id)
-            .await?;
-        self.prepare_persisted_session_read_locked(storage, session_id)
-            .await?;
+            .persistence_manager()
+            .load_session_revert_state(storage, session_id)
+            .await?
+            .is_some_and(|state| state.phase != SessionRevertPhase::Staged)
+        {
+            // Without the lease an unfinished undo transition cannot be
+            // reconciled. Report that instead of projecting records that the
+            // transition is about to replace.
+            return Err(OpenBitFunError::OutcomeUnknown(format!(
+                "Session history is unavailable while its writer holds the Session: session_id={session_id}"
+            )));
+        }
+        let _mutation = mutation;
 
         // Persisted InProgress is not proof of a live executor after restart.
         // A loaded owner supplies runtime state; for an unloaded session an

src/web-ui/src/flow_chat/services/RelaySessionHistory.test.ts

@@ -60,6 +60,36 @@ describe('RelaySessionHistory', () => {
     expect(fixture.unsubscribe).toHaveBeenCalledWith('subscription');
   });
 
+  it('degrades to an empty surface when the host never sends the first page', async () => {
+    vi.useFakeTimers();
+    fixture.subscribe.mockResolvedValueOnce('subscription');
+    const ready = vi.fn();
+    const resumed = vi.fn(async () => {});
+    const history = new RelaySessionHistory('session', vi.fn(), ready, vi.fn(), resumed);
+
+    const opening = history.open();
+    await vi.advanceTimersByTimeAsync(0);
+    expect(ready).not.toHaveBeenCalled();
+    await vi.advanceTimersByTimeAsync(15_000);
+    await opening;
+
+    // The mailbox is read without waiting for a ready notification that may
+    // never arrive, and presentation leaves hydration.
+    expect(resumed).toHaveBeenCalledOnce();
+    expect(ready).toHaveBeenCalledWith({
+      sessionId: 'session', hasMore: false, oldestSeq: 0, cursor: 0,
+    });
+
+    // The subscription stays open, so a late first page still applies.
+    fixture.ready?.({ sessionId: 'session', hasMore: false, oldestSeq: 1, cursor: 9 });
+    expect(ready).toHaveBeenLastCalledWith({
+      sessionId: 'session', hasMore: false, oldestSeq: 1, cursor: 9,
+    });
+
+    history.close();
+    expect(fixture.unsubscribe).toHaveBeenCalledWith('subscription');
+  });
+
   it('drops stale replay and unsubscribes when the surface changes during grant', async () => {
     const apply = vi.fn();
     fixture.subscribe.mockImplementation(async () => {

src/web-ui/src/flow_chat/services/RelaySessionHistory.ts

@@ -6,6 +6,12 @@ import { resourceBudget } from '@/shared/utils/resourceBudget';
 
 const log = createLogger('RelaySessionHistory');
 
+/**
+ * How long opening may wait for the host's first page before presentation
+ * degrades. The live subscription stays open: a later page still applies.
+ */
+const FIRST_PAGE_TIMEOUT_MS = 15_000;
+
 export interface RelayHistoryReady {
   sessionId: string;
   hasMore: boolean;
@@ -33,6 +39,8 @@ export class RelaySessionHistory {
   private rejectPage: ((error: unknown) => void) | null = null;
   private visible = false;
   private prefetchCredit = 0;
+  private readonly publishReady: (ready: RelayHistoryReady) => void;
+  private readonly resumeInteraction: (() => Promise<void>) | undefined;
 
   constructor(
     readonly sessionId: string,
@@ -41,6 +49,8 @@ export class RelaySessionHistory {
     onError: (error: unknown) => void,
     onResumed?: () => Promise<void>,
   ) {
+    this.publishReady = onReady;
+    this.resumeInteraction = onResumed;
     // Register before asking Rust to replay its cache or fetch the first page.
     this.listeners.push(remoteConnectAPI.onSessionRecord(value => {
       if (!this.isCurrent()) return;
@@ -109,7 +119,7 @@ export class RelaySessionHistory {
         throw new Error('Session subscription is closed');
       }
       this.subscriptionId = id;
-      await this.firstPage;
+      await this.waitForFirstPage();
       this.scheduleHistoryPrefetch();
     } catch (error) {
       this.rejectReady(error);
@@ -118,6 +128,49 @@ export class RelaySessionHistory {
     }
   }
 
+  /**
+   * A stalled host stream must not park the surface in hydration. The
+   * subscription stays open, so a later page still replaces this empty
+   * projection; what must not happen is a permanent spinner.
+   */
+  private async waitForFirstPage(): Promise<void> {
+    let timer: ReturnType<typeof setTimeout> | null = null;
+    const overdue = new Promise<'overdue'>(resolve => {
+      timer = setTimeout(() => resolve('overdue'), FIRST_PAGE_TIMEOUT_MS);
+    });
+    try {
+      const outcome = await Promise.race([
+        this.firstPage.then(() => 'ready' as const),
+        overdue,
+      ]);
+      if (outcome === 'overdue' && this.isCurrent()) {
+        log.warn('Relay session first page overdue; presenting an empty surface', {
+          sessionId: this.sessionId,
+        });
+        // A blocking question belongs to the Runtime mailbox, not to this
+        // stream. Once the stream is overdue, read the mailbox directly instead
+        // of waiting for a ready notification that may never arrive.
+        if (this.resumeInteraction) {
+          void this.resumeInteraction().catch(error => {
+            log.warn('Session interaction mailbox refresh failed', {
+              sessionId: this.sessionId,
+              error,
+            });
+          });
+        }
+        this.resolveReady();
+        this.publishReady({
+          sessionId: this.sessionId,
+          hasMore: false,
+          oldestSeq: 0,
+          cursor: 0,
+        });
+      }
+    } finally {
+      if (timer !== null) clearTimeout(timer);
+    }
+  }
+
   // Warm one adjacent page after a visible open or reader request. Never drain
   // an entire remote transcript merely because its live subscription is open.
   private scheduleHistoryPrefetch(): void {