diff --git a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs index dea1101aed..0b6f0afb06 100644 --- a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs +++ b/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 diff --git a/src/web-ui/src/flow_chat/services/RelaySessionHistory.test.ts b/src/web-ui/src/flow_chat/services/RelaySessionHistory.test.ts index 399e9605ad..14ee4282f0 100644 --- a/src/web-ui/src/flow_chat/services/RelaySessionHistory.test.ts +++ b/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 () => { diff --git a/src/web-ui/src/flow_chat/services/RelaySessionHistory.ts b/src/web-ui/src/flow_chat/services/RelaySessionHistory.ts index b5698eb10a..98209c4607 100644 --- a/src/web-ui/src/flow_chat/services/RelaySessionHistory.ts +++ b/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) | undefined; constructor( readonly sessionId: string, @@ -41,6 +49,8 @@ export class RelaySessionHistory { onError: (error: unknown) => void, onResumed?: () => Promise, ) { + 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 { + let timer: ReturnType | 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 {