refactor(peer): give a Session's projection one owner, and make a delivery gap answerable

docs/architecture/peer-device-mode.md

@@ -90,7 +90,19 @@ boundaries; noisy tool progress is compacted.
 
 `restore_session_view` returns this additive `runtimeEventSnapshot`; the CLI
 Peer Host applies its existing Peer-owned-Turn filter before recording or
-returning the projection. During an attach, the frontend fences live events for
+returning the projection. The persisted Session record of an executing Turn is
+a lagging checkpoint, not the live projection: `loadSessionHistory` and
+`refreshPeerSessionSnapshot` must not paint that checkpoint's in-progress
+tool rows. A restore that races a live store update must still return the
+journal so attach can replay. Delivery of a live event to a product listener
+is not acceptance — a dropped ToolEvent / TextChunk marks the projection
+stale so the next attach replays instead of treating the cursor as current.
+`finish()` covers those cursors only after the painted projection has caught
+up with journal terminal tools; a matching cursor alone is not enough.
+Overlapping attach transfers the in-flight fence rather than delivering it
+onto a state machine that is about to reset. Hidden-document and
+fresh-TextChunk liveness skips apply only to the 3s poll, not to a dirty
+projection. During an attach, the frontend fences live events for
 `(DeviceSurfaceId, SessionId)`,
 replays the snapshot into an empty current-Turn projection, and then releases
 only events newer than the snapshot cursor. A different `streamId` is a new

docs/architecture/session-projection.md

@@ -0,0 +1,173 @@
+# Session Projection
+
+What a client shows for a Session is a **projection of an ordered stream**. This
+document is the contract that projection obeys, and the migration that brings
+the existing writers under it.
+
+Read [`peer-device-mode.md`](peer-device-mode.md) for how a controller reaches
+another device. This document is about what happens to the data once it
+arrives, and applies identically on the local surface.
+
+## The problem this replaces
+
+Seven independent writers currently produce a Session's on-screen state:
+
+| # | Writer | Entry point |
+|---|---|---|
+| 1 | Live agentic events | `AgenticEventListener` → `eventBatcher` |
+| 2 | Disk hydrate | `loadSessionHistory` → `restore_session_view` |
+| 3 | Windowed history | `load_session_turn_window` |
+| 4 | Snapshot reconcile | `refreshPeerSessionSnapshot`, `replaceRunningSnapshot` |
+| 5 | Journal snapshot replay | `dispatchExternal(snapshot.events)` |
+| 6 | Interaction mailbox | `reconcilePendingUserQuestions` |
+| 7 | Backfill delta | `load_session_event_backfill` |
+
+None of them carries a position that the others can compare against, so every
+pair needs its own conflict rule. Those rules are the fourteen invariants in
+[`peer-device/README.md`](../../src/web-ui/src/infrastructure/peer-device/README.md),
+and they are written in terms of painted content rather than ordering:
+`snapshotDropsProjectedTurnContent` decides whether a write is safe by counting
+rounds and progress entries; `runtimeProjectionCaughtUp` decides whether a
+cursor may be trusted by looking for tool cards on screen.
+
+Counting pixels to decide whether a write is safe is what a missing position
+looks like. The rule table also grows quadratically: adding writer 7 required
+two new pairwise rules (7↔1, 7↔6), and shipping without them produced exactly
+two defects — the live stream stalled behind writer 7's fence, and a blocking
+interaction came back unanswerable because writer 6 never ran.
+
+## Contract
+
+### 1. Every write carries a position, and the projection never regresses
+
+A write whose position is not ahead of what has been applied is **dropped, not
+merged**. There is no operation that replaces projected content, so no writer
+needs to prove it is not about to lose any.
+
+Positions are per `(surface, session)`:
+
+- **Runtime positions** — `(streamId, cursor)`, minted by the Host journal as
+  each event enters its ordered delivery stream. `streamId` identifies the
+  Runtime process; cursors from different `streamId`s are never comparable.
+- **History positions** — turn ordinal within the persisted record. Immutable
+  and totally ordered.
+
+The two are not compared with each other. They cannot conflict, because of
+invariant 2.
+
+### 2. A Turn has exactly one writer, decided by whether it is executing
+
+- An **executing** Turn is owned by the runtime stream. No persisted record,
+  checkpoint, or snapshot of that Turn may write it.
+- A **settled** Turn is owned by the persisted record. No live event may
+  write it.
+- Ownership transfers **once**, on the Turn's terminal event, driven by
+  position — never by inspecting what is on screen.
+
+This is why history and live events cannot race: they are never both
+authoritative for the same Turn. The persisted checkpoint of an executing Turn
+is identity only; it names the Turn and carries none of its content.
+
+### 3. Identity is `(surface, session)` by construction
+
+One `SessionStream` object owns the position, the pending queue, and the
+projection for one `(DeviceSurfaceId, SessionId)`. Workspace paths and session
+ids repeat across machines, so surface is part of identity, not an extra
+argument each feature remembers to thread through.
+
+Any state that is per-Session is reached through its stream. A feature cannot
+hold Session state that is not surface-scoped, because there is nowhere to
+put it.
+
+## Sources are not writers
+
+Every source above becomes a way of **obtaining positioned events**, applied
+through one path:
+
+| Source | Produces |
+|---|---|
+| Live DeviceEvent / local emit | events at `(streamId, cursor)` |
+| `load_session_event_backfill` | events after a position, or `snapshotRequired` |
+| `restore_session_view` runtime snapshot | a compacted prefix ending at a position |
+| `restore_session_view` turns / `load_session_turn_window` | settled Turns at history positions |
+| Interaction mailbox | revisioned state of an executing Turn, applied at its position |
+
+A snapshot is a prefix. A delta is a suffix. History is the older part of the
+same order. None of them is a distinct kind of write.
+
+## What this deletes
+
+Each item disappears when its writer migrates. This list is the acceptance
+criteria — a migration step that does not remove its entry has not finished.
+
+| Removed | Replaced by |
+|---|---|
+| `replaceRunningSnapshot` | there is no replace operation (contract 1) |
+| `runtimeProjectionCaughtUp` | the applied position is the answer (contract 1) |
+| `prepareRuntimeTurnReplay` / `asRuntimeReplayTurn` | an executing Turn has one writer (contract 2) |
+| `hasGap` / `projectionStale` / `markRuntimeSessionProjectionStale` | a position discontinuity is the gap |
+| `beginRuntimeSessionAttachment` fence | the stream's own queue (contract 3) |
+| manual `(DeviceSurfaceId, …)` threading | stream identity (contract 3) |
+
+The 3s poll is **not** on this list. Events that never arrive advance no
+cursor, so no discontinuity is observable; the poll remains the liveness probe
+that notices a stream has gone quiet. It stops being a repair mechanism.
+
+### Two gaps the contract does not yet close
+
+Both were found by deleting a heuristic and watching a behavioural test fail.
+They are why `snapshotDropsProjectedTurnContent` and
+`isRunningSnapshotForwardProgress` survive, inside
+`persistedReadMayReplaceTurn`, as the last content comparison in the merge:
+
+- **A Host that serves no runtime projection.** Contract 2 hands an executing
+  Turn to the runtime stream, but an older Host has no such stream. Its
+  persisted checkpoint is the only progress that exists, so forward progress
+  from it is still admitted when `runtimeEventSnapshot` is absent.
+- **A partial history read.** History positions are turn ordinals, and a
+  windowed or not-yet-checkpointed read can name a Turn while carrying none of
+  its work. Such a read holds no position for the content it omitted, so
+  writing the Turn from it is lossy rather than advancing. Closing this needs
+  the read to report its own completeness; until then "would this write lose
+  content" is the only question available.
+
+Deleting either guard without first closing its gap reintroduces a real defect,
+not just a test failure.
+
+## Migration
+
+Ordered so that each step is separately verifiable and deletes its own rules.
+
+1. **Position algebra + `SessionStream`** — the contract as a tested module,
+   with no writer on it yet.
+2. **Runtime-stream writers (1, 5, 7)** — live events, snapshot replay, and
+   backfill are already positioned; move them onto the stream and delete the
+   fence, `hasGap`, and `runtimeProjectionCaughtUp`.
+3. **Snapshot reconcile (4)** — becomes "apply a prefix"; deletes
+   `replaceRunningSnapshot` and `snapshotDropsProjectedTurnContent`.
+4. **Interaction mailbox (6)** — applied at the executing Turn's position
+   rather than as a separate reconcile pass.
+5. **History (2, 3)** — settled Turns at history positions; deletes the
+   executing/settled overlap rules and `prepareRuntimeTurnReplay`.
+
+Steps 2–5 each remove entries from the peer-device README's invariant list.
+That list shrinking is the measure of progress; if it is not shrinking, the
+step reintroduced a pairwise rule instead of removing one.
+
+## Host contract
+
+Hosts expose exactly two reads over a Session's stream, both already present:
+
+- `restore_session_view` — a prefix (compacted projection + settled Turns +
+  mailbox), ending at a position.
+- `load_session_event_backfill` — the suffix after a position, or
+  `snapshotRequired` when contiguity cannot be proven.
+
+`SessionEventJournal` owns both. The compacted projection answers "what does
+this Turn look like now"; the append-only tail answers "what came after
+position N". Neither is allowed to answer the other's question — that
+conflation is what made a gap something to infer.
+
+Peer ownership is a cancellation and bookkeeping boundary and never filters
+either read. A Turn started in a Host's own TUI is part of the Session every
+attached surface projects.

src/apps/cli/src/peer_host/commands/dialog.rs

@@ -1,16 +1,76 @@
 //! Dialog HostInvoke handlers.
 
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex, OnceLock};
+use std::time::{Duration, Instant};
+
 use serde_json::{json, Value};
+use tokio::sync::Mutex as AsyncMutex;
 
 use bitfun_runtime_ports::{
     AgentDialogTurnRequest, AgentSubmissionSource, AgentTurnCancellationRequest,
     DialogSubmissionPolicy, DialogTriggerSource,
 };
 
 use crate::peer_host::args::{get_string, optional_string, request_value};
-use crate::peer_host::control::{attached_controller_lease, is_controller_lease_current};
 use crate::peer_host::state::{PeerHostState, PeerTurnKey};
 
+/// How long a settled submission stays replayable. The controller's own
+/// bounded retry window is far shorter; this only has to outlive it.
+const DIALOG_SUBMISSION_TTL: Duration = Duration::from_secs(120);
+const MAX_CACHED_DIALOG_SUBMISSIONS: usize = 128;
+
+type DialogSubmissionOutcome = Arc<AsyncMutex<Option<Result<Value, String>>>>;
+
+struct CachedDialogSubmission {
+    expires_at: Instant,
+    outcome: DialogSubmissionOutcome,
+}
+
+fn dialog_submissions() -> &'static Mutex<HashMap<String, CachedDialogSubmission>> {
+    static DIALOG_SUBMISSIONS: OnceLock<Mutex<HashMap<String, CachedDialogSubmission>>> =
+        OnceLock::new();
+    DIALOG_SUBMISSIONS.get_or_init(|| Mutex::new(HashMap::new()))
+}
+
+/// Reserve the slot that records the outcome of one `(sessionId, turnId)`
+/// submission.
+///
+/// A Relay timeout leaves the controller unable to tell "never arrived" from
+/// "already running". Holding the outcome behind a per-identity async lock
+/// makes a replay wait for the original attempt and then observe its result,
+/// instead of starting the prompt a second time. Desktop Peer Hosts get this
+/// from the webview bridge's idempotency cache; a CLI Host has no webview, so
+/// it owns the same contract here.
+fn dialog_submission_slot(
+    session_id: &str,
+    turn_id: &str,
+) -> Result<DialogSubmissionOutcome, String> {
+    let mut submissions = dialog_submissions()
+        .lock()
+        .map_err(|_| "Peer dialog submission cache is unavailable".to_string())?;
+    let now = Instant::now();
+    submissions.retain(|_, entry| entry.expires_at > now);
+    while submissions.len() >= MAX_CACHED_DIALOG_SUBMISSIONS {
+        let Some(oldest) = submissions
+            .iter()
+            .min_by_key(|(_, entry)| entry.expires_at)
+            .map(|(key, _)| key.clone())
+        else {
+            break;
+        };
+        submissions.remove(&oldest);
+    }
+    Ok(submissions
+        .entry(format!("{session_id}:{turn_id}"))
+        .or_insert_with(|| CachedDialogSubmission {
+            expires_at: now + DIALOG_SUBMISSION_TTL,
+            outcome: Arc::new(AsyncMutex::new(None)),
+        })
+        .outcome
+        .clone())
+}
+
 fn peer_dialog_metadata(request: &Value) -> Result<serde_json::Map<String, Value>, String> {
     let mut metadata = match request.get("userMessageMetadata") {
         Some(Value::Object(metadata)) => metadata.clone(),
@@ -35,6 +95,25 @@ pub(crate) async fn start_dialog_turn(
     state: &PeerHostState,
     args: &Value,
 ) -> Result<Value, String> {
+    let request = request_value(args);
+    let session_id = get_string(request, "sessionId")?;
+    // Only a client-supplied turn id gives the submission a stable identity.
+    // Without one there is nothing to deduplicate against — and the controller
+    // stays single-shot for exactly that reason.
+    let Some(turn_id) = optional_string(request, "turnId") else {
+        return submit_dialog_turn(state, args).await;
+    };
+    let outcome_slot = dialog_submission_slot(&session_id, &turn_id)?;
+    let mut outcome = outcome_slot.lock().await;
+    if let Some(settled) = outcome.as_ref() {
+        return settled.clone();
+    }
+    let result = submit_dialog_turn(state, args).await;
+    *outcome = Some(result.clone());
+    result
+}
+
+async fn submit_dialog_turn(state: &PeerHostState, args: &Value) -> Result<Value, String> {
     let request = request_value(args);
     let session_id = get_string(request, "sessionId")?;
     let user_input = get_string(request, "userInput")?;
@@ -44,22 +123,23 @@ pub(crate) async fn start_dialog_turn(
         .or_else(|| optional_string(request, "workspacePath"));
     let remote_connection_id = optional_string(request, "remoteConnectionId");
     let remote_ssh_host = optional_string(request, "remoteSshHost");
-    let controller_lease = attached_controller_lease()?;
     let turn_id =
         optional_string(request, "turnId").unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
     let metadata = peer_dialog_metadata(request)?;
     let turn = PeerTurnKey::new(session_id.clone(), turn_id.clone());
     let stream_generation = state.turns.register_root(turn.clone())?;
+    // Controller presence is not an admission requirement. This host owns and
+    // keeps executing the Turn whether or not a controller is watching, exactly
+    // like a Turn submitted in its own TUI; a controller that leaves and comes
+    // back re-attaches to the same Runtime projection. Only this host's own
+    // event-stream continuity still gates submission, because a Turn nobody can
+    // observe is a Turn nobody can attach to.
     if !state
         .turns
         .is_event_stream_generation_current(stream_generation)
-        || !is_controller_lease_current(controller_lease)
     {
         state.turns.finish_turn(&turn);
-        return Err(
-            "Peer controller or event stream continuity was lost before dialog submission"
-                .to_string(),
-        );
+        return Err("Peer event stream continuity was lost before dialog submission".to_string());
     }
 
     let policy = DialogSubmissionPolicy::for_source(DialogTriggerSource::DesktopUi);
@@ -124,9 +204,10 @@ pub(crate) async fn cancel_dialog_turn(
     let request = request_value(args);
     let session_id = get_string(request, "sessionId")?;
     let dialog_turn_id = get_string(request, "dialogTurnId")?;
-    if !state.turns.owns(&session_id, Some(&dialog_turn_id)) {
-        return Err("The dialog turn is not owned by the Peer controller".to_string());
-    }
+    // Cancellation follows visibility: a controller renders every Turn in the
+    // Session, including ones this host started, so it must be able to stop
+    // them. A Desktop Peer Host reaches the same handler the local UI does and
+    // has never had an ownership gate here.
     state
         .agent_runtime
         .cancel_turn(AgentTurnCancellationRequest {
@@ -147,7 +228,7 @@ pub(crate) async fn cancel_dialog_turn(
 mod tests {
     use serde_json::json;
 
-    use super::peer_dialog_metadata;
+    use super::{dialog_submission_slot, peer_dialog_metadata};
 
     #[test]
     fn peer_metadata_removes_reserved_runtime_fields() {
@@ -196,4 +277,30 @@ mod tests {
         assert_eq!(metadata.get("kind"), Some(&json!("manual_compaction")));
         assert_eq!(metadata.get("sourceKind"), Some(&json!("user")));
     }
+
+    #[tokio::test]
+    async fn a_replayed_submission_observes_the_first_attempt_instead_of_running_twice() {
+        let first = dialog_submission_slot("session-1", "turn-1").expect("first slot");
+        let mut outcome = first.lock().await;
+        assert!(outcome.is_none(), "a fresh identity has nothing to replay");
+        *outcome = Some(Ok(json!({ "success": true })));
+        drop(outcome);
+
+        // The Relay timed out, so the controller replays the exact payload. It
+        // must observe the recorded outcome, not submit the prompt again.
+        let replay = dialog_submission_slot("session-1", "turn-1").expect("replay slot");
+        assert_eq!(
+            replay.lock().await.clone(),
+            Some(Ok(json!({ "success": true }))),
+        );
+
+        // A different turn in the same Session is a different submission.
+        let other_turn = dialog_submission_slot("session-1", "turn-2").expect("other turn slot");
+        assert!(other_turn.lock().await.is_none());
+
+        // As is the same turn id under a different Session.
+        let other_session =
+            dialog_submission_slot("session-2", "turn-1").expect("other session slot");
+        assert!(other_session.lock().await.is_none());
+    }
 }

src/apps/cli/src/peer_host/commands/mod.rs

@@ -77,6 +77,7 @@ pub(crate) async fn dispatch(
         }
         "load_session_turn_window" => session::load_session_turn_window(state, args).await,
         "load_session_turns" => session::load_session_turns(state, args).await,
+        "load_session_event_backfill" => session::load_session_event_backfill(state, args),
         "restore_session_view" => session::restore_session_view(state, args).await,
         "restore_session_with_turns" => session::restore_session_with_turns(state, args).await,
         "restore_session" => session::restore_session(state, args).await,