fix(remote,flow-chat): settle stale running sessions and keep turn settlement under event floods

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

@@ -10168,6 +10168,44 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet
         self.reconcile_restored_session(session_id, session).await
     }
 
+    /// Load a persisted user session back into this runtime before accepting
+    /// user work for it. Idle sessions may be released from memory while a
+    /// remote controller keeps showing them; its next message must find the
+    /// same session instead of being refused.
+    ///
+    /// The storage binding committed when this process admitted the session
+    /// wins. Without one (e.g. after a host restart) the persisted workspace
+    /// binding selects storage and takes runtime ownership, exactly as a
+    /// mobile remote submission does.
+    pub(crate) async fn restore_unloaded_session_for_admission(
+        &self,
+        session_id: &str,
+    ) -> OpenBitFunResult<()> {
+        if self.session_manager.get_session(session_id).is_some() {
+            return Ok(());
+        }
+        if let Some(storage_path) = self
+            .session_manager
+            .committed_session_storage_path_for(session_id)
+        {
+            self.restore_session_from_storage_path(&storage_path, session_id)
+                .await?;
+            return Ok(());
+        }
+        let binding = self
+            .session_manager
+            .resolve_session_workspace_binding(session_id)
+            .await
+            .ok_or_else(|| {
+                OpenBitFunError::NotFound(format!(
+                    "Session workspace is unavailable on this host: {session_id}"
+                ))
+            })?;
+        self.restore_session_for_workspace_binding(&binding, session_id)
+            .await?;
+        Ok(())
+    }
+
     /// Restore a persisted session through its resolved workspace binding.
     pub async fn restore_session_for_workspace_binding(
         &self,

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

@@ -369,18 +369,34 @@ impl DialogScheduler {
         // that may not be restored on this execution host yet, e.g. right after
         // a local <-> remote workspace switch. Reads may degrade: the scheduler
         // keeps its own receipts, so report that state without failing every
-        // poll. Mutations keep failing — accepting user work against a session
-        // this runtime cannot vouch for would silently lose it.
-        let session_unloaded = self.session_manager.get_session(session).is_none();
+        // poll and without loading the session.
+        //
+        // A mutation is user intent. An idle session may have been released
+        // from memory while a remote controller still shows it, so load it
+        // back from persistence first. Only a session this runtime cannot
+        // restore is refused — accepting user work against a session it cannot
+        // vouch for would silently lose it.
         if !matches!(
             request.action,
             DialogQueueAction::List | DialogQueueAction::Get { .. }
-        ) && session_unloaded
+        ) && self.session_manager.get_session(session).is_none()
         {
-            return Err(PortError::new(
-                PortErrorKind::NotFound,
-                "Session is not loaded on the execution host",
-            ));
+            if let Err(restore_error) = self
+                .coordinator
+                .restore_unloaded_session_for_admission(session)
+                .await
+            {
+                warn!(
+                    "Host queue mutation refused: session could not be restored on the execution host: session_id={}, error={}",
+                    session, restore_error
+                );
+            }
+            if self.session_manager.get_session(session).is_none() {
+                return Err(PortError::new(
+                    PortErrorKind::NotFound,
+                    "Session is not loaded on the execution host",
+                ));
+            }
         }
         let snapshot = self.queue_snapshot(session, None);
         if !matches!(request.action, DialogQueueAction::List)

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

@@ -79,7 +79,8 @@ async fn host_queue_reads_degrade_for_a_session_not_loaded_on_the_host() {
         Some("queued-a")
     );
 
-    // A mutation still refuses to accept user work for an unloaded session.
+    // Nothing was persisted here, so the session cannot be restored and a
+    // mutation still refuses to accept user work for it.
     let submit = scheduler
         .manage_host_queue(request(
             Some(&epoch),
@@ -110,6 +111,65 @@ async fn host_queue_reads_degrade_for_a_session_not_loaded_on_the_host() {
     );
 }
 
+/// An idle session released from host memory is still the session a remote
+/// controller shows. Its next queued message restores it from persistence
+/// instead of failing (and failing again on every retry).
+#[tokio::test]
+async fn host_queue_mutation_restores_an_idle_session_released_from_memory() {
+    let (scheduler, sessions, _, root) = test_scheduler_with_persistence(true);
+    let session_id = "host-queue-session";
+    let workspace = fixture_workspace_dir(root.path().join("released-workspace"));
+    sessions
+        .create_session_with_id(
+            Some(session_id.to_string()),
+            "Released".to_string(),
+            "Standard".to_string(),
+            SessionConfig {
+                tool_presentation_mode: Default::default(),
+                workspace_path: Some(workspace.to_string_lossy().into_owned()),
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("create session");
+    assert!(sessions
+        .unload_session_from_memory(session_id)
+        .await
+        .expect("unload idle session"));
+    assert!(sessions.get_session(session_id).is_none());
+
+    // Passive reads answer without loading the session.
+    let list = scheduler
+        .manage_host_queue(request(None, Action::List))
+        .await
+        .unwrap();
+    assert_eq!(list.active_turn_id, None);
+    assert!(
+        sessions.get_session(session_id).is_none(),
+        "a queue poll must not load the session"
+    );
+
+    // Keep the accepted message parked so the test does not execute a model.
+    scheduler
+        .active_turns
+        .insert(session_id.into(), desktop_active_turn("parking-turn"));
+    let accepted = scheduler
+        .manage_host_queue(request(
+            Some(&list.queue_epoch),
+            Action::Submit {
+                message: message("after-release"),
+            },
+        ))
+        .await
+        .expect("a released idle session is restored before admission");
+    assert!(sessions.get_session(session_id).is_some());
+    assert_eq!(accepted.queue_epoch, list.queue_epoch);
+    assert_eq!(
+        accepted.receipt.as_ref().map(|item| item.turn_id.as_str()),
+        Some("after-release")
+    );
+}
+
 #[tokio::test]
 async fn host_queue_subagent_conversation_admission_precedes_receipts_and_queue_mutation() {
     use openbitfun_core_types::{SessionContinuationPolicy, SessionKind};

src/crates/assembly/core/src/agentic/session/session_manager.rs

@@ -1248,6 +1248,12 @@ impl SessionManager {
             .map(|binding| binding.path.clone())
     }
 
+    /// Storage binding committed by an earlier create/restore admission, kept
+    /// after the session is unloaded from memory.
+    pub(crate) fn committed_session_storage_path_for(&self, session_id: &str) -> Option<PathBuf> {
+        Self::committed_session_storage_path(&self.session_storage_path_index, session_id)
+    }
+
     /// Reuse the storage binding admitted when this session was created/restored.
     /// A workspace path is an execution projection, never a replacement for this
     /// binding. In particular, a later same-path SSH record or runtime-layout

src/crates/execution/agent-runtime/src/event_queue.rs

@@ -128,6 +128,24 @@ enum LegacyQueuePolicy {
     AuthoritativeControl,
 }
 
+/// Turn lifecycle facts a consumer cannot reconstruct from later stream data.
+/// Losing one leaves a projection running forever, or settled without the
+/// durable history fence that makes it reconcile canonical content, until a
+/// full reload. Text and tool stream data stay best-effort: a delivered
+/// lifecycle fence repairs them. Each turn emits only a handful of these, so
+/// admitting them past the data budget stays bounded.
+fn is_turn_lifecycle_fact(event: &AgenticEvent) -> bool {
+    matches!(
+        event,
+        AgenticEvent::DialogTurnStarted { .. }
+            | AgenticEvent::DialogTurnCompleted { .. }
+            | AgenticEvent::DialogTurnCancelled { .. }
+            | AgenticEvent::DialogTurnFailed { .. }
+            | AgenticEvent::SessionStateChanged { .. }
+            | AgenticEvent::SessionHistoryChanged { .. }
+    )
+}
+
 fn interrupted_control_identity(event: &AgenticEvent) -> Option<(&str, &str, u32)> {
     match event {
         AgenticEvent::DialogTurnInterrupted {
@@ -309,6 +327,14 @@ impl EventQueue {
         priority: Option<EventPriority>,
         legacy_policy: LegacyQueuePolicy,
     ) -> EventBusResult<(String, Option<LegacyDequeueAck>)> {
+        // A best-effort producer cannot know that the queue is full; never let
+        // that silently discard a lifecycle fact (see `is_turn_lifecycle_fact`).
+        let legacy_policy = match legacy_policy {
+            LegacyQueuePolicy::BestEffort if is_turn_lifecycle_fact(&event) => {
+                LegacyQueuePolicy::AuthoritativeControl
+            }
+            policy => policy,
+        };
         let priority = priority.unwrap_or_else(|| event.default_priority());
         let envelope = EventEnvelope::new(event, priority);
         let event_id = envelope.id.clone();
@@ -1038,9 +1064,13 @@ mod tests {
         for session_id in ["first", "second"] {
             queue
                 .enqueue(
-                    AgenticEvent::SessionStateChanged {
+                    AgenticEvent::TextChunk {
                         session_id: session_id.to_string(),
-                        new_state: "idle".to_string(),
+                        turn_id: "turn".to_string(),
+                        round_id: "round".to_string(),
+                        attempt_id: None,
+                        attempt_index: None,
+                        text: "data".to_string(),
                     },
                     None,
                 )
@@ -1069,6 +1099,73 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn full_queue_keeps_turn_lifecycle_facts_of_another_session() {
+        let queue = EventQueue::new(EventQueueConfig {
+            max_queue_size: 2,
+            batch_size: 16,
+        });
+        let chunk = |session_id: &str, text: &str| AgenticEvent::TextChunk {
+            session_id: session_id.to_string(),
+            turn_id: "turn".to_string(),
+            round_id: "round".to_string(),
+            attempt_id: None,
+            attempt_index: None,
+            text: text.to_string(),
+        };
+        // Another session floods the data budget.
+        for text in ["flood-1", "flood-2"] {
+            queue.enqueue(chunk("busy", text), None).await.unwrap();
+        }
+        // This session's last chunk is best-effort data and may be skipped...
+        queue.enqueue(chunk("quiet", "tail"), None).await.unwrap();
+        // ...but its settlement and history fence must reach the consumer.
+        queue
+            .enqueue(
+                AgenticEvent::DialogTurnCompleted {
+                    session_id: "quiet".to_string(),
+                    turn_id: "turn".to_string(),
+                    total_rounds: 1,
+                    total_tools: 0,
+                    duration_ms: 1,
+                    partial_recovery_reason: None,
+                    success: Some(true),
+                    finish_reason: None,
+                    has_final_response: Some(true),
+                    background_delivery: None,
+                },
+                None,
+            )
+            .await
+            .unwrap();
+        queue
+            .enqueue(
+                AgenticEvent::SessionHistoryChanged {
+                    session_id: "quiet".to_string(),
+                    settled_turn_id: Some("turn".to_string()),
+                },
+                None,
+            )
+            .await
+            .unwrap();
+
+        let delivered = queue.dequeue_batch(16).await;
+        let quiet: Vec<_> = delivered
+            .iter()
+            .filter(|envelope| envelope.event.session_id() == Some("quiet"))
+            .map(|envelope| &envelope.event)
+            .collect();
+        assert!(quiet
+            .iter()
+            .any(|event| matches!(event, AgenticEvent::DialogTurnCompleted { .. })));
+        assert!(quiet
+            .iter()
+            .any(|event| matches!(event, AgenticEvent::SessionHistoryChanged { .. })));
+        assert!(!quiet
+            .iter()
+            .any(|event| matches!(event, AgenticEvent::TextChunk { .. })));
+    }
+
     #[tokio::test]
     async fn default_sized_broadcast_preserves_bursts_above_legacy_1024_limit() {
         let queue = EventQueue::new(EventQueueConfig::default());

src/mobile-web/src/services/SessionSynchronizer.ts

@@ -60,6 +60,9 @@ export class SessionSynchronizer {
   private async connect():Promise<void> {
     if(this.stopped||this.connecting||this.unsubscribe)return;
     this.connecting=true;
+    // A new subscription may land on a host stream rebuilt with lower revisions;
+    // its records must supersede what an earlier attempt delivered.
+    this.replica.rebaseRevisions();
     try {
       const unsubscribe=await this.sessionMgr.subscribeSessionStream(this.sessionId,{onEvent:event=>{
         if(this.stopped)return;

src/shared/dialog-queue/HostDialogQueue.ts

@@ -75,6 +75,8 @@ const newId = (): string => {
   const hex = Array.from(bytes, value => value.toString(16).padStart(2, '0')).join('');
   return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`;
 };
+/** A stable queue Turn ID a caller can hold before submitting. */
+export const newQueueTurnId = (): string => newId();
 function messageIdentity(message: Omit<QueueMessage, 'turnId'>): string {
   // Reopening the composer can allocate fresh image IDs. An ambiguous retry
   // must send the stored payload, including its original IDs, unchanged.
@@ -178,18 +180,22 @@ export class HostDialogQueue {
         || (preparedSnapshot && this.view.snapshot?.queueEpoch !== snapshot.queueEpoch)) {
         throw new Error('queue_scope_expired: refresh the host queue; do not automatically resend');
       }
-      const identity = messageIdentity(message);
-      const previous = this.view.pending.find(record => {
-        if (record.restoreIntent || record.request.action !== 'submit') return false;
-        const { turnId: _id, ...payload } = record.request.message;
-        return messageIdentity(payload) === identity;
-      });
+      const previous = this.pendingSubmission(message);
       if (previous) return this.retryRecord(previous, snapshot);
       const turnId = requestedTurnId?.trim() || newId();
       return this.transmit({ scope: this.scope, key: JSON.stringify([this.scope, turnId]), draft,
         request: { sessionId: this.sessionId, queueEpoch: snapshot.queueEpoch, action: 'submit', message: { ...message, turnId } } });
     });
   }
+  /** The unresolved outbox record that submitting this payload would retry instead. */
+  pendingSubmission(message: Omit<QueueMessage, 'turnId'>): QueueOutboxRecord | undefined {
+    const identity = messageIdentity(message);
+    return this.view.pending.find(record => {
+      if (record.restoreIntent || record.request.action !== 'submit') return false;
+      const { turnId: _id, ...payload } = record.request.message;
+      return messageIdentity(payload) === identity;
+    });
+  }
   act(item: QueueItem, action: 'cancel' | 'promote'): Promise<QueueSnapshot> {
     const observed = this.view;
     return this.serialize(async () => {

src/shared/relay-transport/SessionRecordReplica.ts

@@ -104,6 +104,40 @@ export class SessionRecordReplica {
     return turnId;
   }
 
+  /**
+   * Prepare for a new subscription of the same session. A host stream rebuilt
+   * after it expired restarts its revisions and may stamp a fresh snapshot at
+   * or below revisions held here, so a turn that settled meanwhile would be
+   * rejected and stay running. Held revisions move below any revision a stream
+   * issues (always >= 1) while keeping their relative order, so held content
+   * and tombstones keep their meaning and whatever the new subscription sends
+   * supersedes them. Ranks are recomputed each time, so repeated rebases never
+   * lose precision.
+   */
+  rebaseRevisions(): void {
+    const held = new Set<number>([...this.records.values(), ...this.deleted.values()]);
+    for (const turn of this.turns.values()) {
+      held.add(turn.header.revision);
+      for (const round of turn.rounds.values()) {
+        held.add(round.header.revision);
+        for (const item of round.items.values()) held.add(item.revision);
+      }
+    }
+    if (held.size === 0) return;
+    const ordered = [...held].sort((a, b) => a - b);
+    const rank = new Map(ordered.map((revision, index) => [revision, (index + 1) / (ordered.length + 1)]));
+    const rebase = (revision: number): number => rank.get(revision)!;
+    for (const [id, revision] of this.records) this.records.set(id, rebase(revision));
+    for (const [id, revision] of this.deleted) this.deleted.set(id, rebase(revision));
+    for (const turn of this.turns.values()) {
+      turn.header.revision = rebase(turn.header.revision);
+      for (const round of turn.rounds.values()) {
+        round.header.revision = rebase(round.header.revision);
+        for (const item of round.items.values()) item.revision = rebase(item.revision);
+      }
+    }
+  }
+
   /** The turn's current state; `turn` is null once it was tombstoned or never seen. */
   snapshot(turnId: string): SessionRecordChange {
     const turn = this.turns.get(turnId);