fix(runtime): stop journaling a vendor's own terminal-reply storm as out-of-band input

crates/protocol/src/event.rs

@@ -663,6 +663,14 @@ pub enum RuntimeEventKind {
 // from the shipped schema entirely (see NOT_WIRE_MESSAGE_ROOTS in
 // crates/xtask/src/main.rs), so it has no name a schema consumer could
 // look up; that's why the shipped description below doesn't say it.
+/// `OutOfBandInput::input_count`'s default for a row journaled before
+/// coalescing existed: such a row is exactly one input, never zero -- an
+/// `OutOfBandInput` event means "a human typed", and a row asserting zero
+/// of them would contradict its own existence.
+fn one_input() -> u64 {
+    1
+}
+
 /// A sanitized, durable runtime event. Fields are plain, already-sanitized
 /// values, so raw thinking/secret content can never reach the durable log
 /// through this type.
@@ -943,10 +951,31 @@ pub enum RuntimeEvent {
     },
     /// A human typed directly into a native pane, bypassing the
     /// adapter. Sets the run's `RunFlags.needsReconciliation` flag.
+    ///
+    /// Coalesced: one row covers every out-of-band input observed during
+    /// one idle window, not one row per keystroke read. Journaled at the
+    /// END of the window it describes, so the window's start is this
+    /// event's own envelope timestamp minus `spanMs` -- no separate start
+    /// timestamp is carried, to avoid a second place for the two to
+    /// disagree.
     OutOfBandInput {
         run_id: RunId,
         backend: DisplayBackend,
         pane_ref: String,
+        /// How many out-of-band inputs this row coalesces. Absent on
+        /// rows written before coalescing existed; such a row is exactly
+        /// one input, so absence reads as 1, never 0 -- an
+        /// `outOfBandInput` event means "a human typed", and zero of
+        /// them is not a state that can be journaled.
+        #[serde(default = "one_input")]
+        #[ts(type = "number")]
+        input_count: u64,
+        /// Milliseconds between the first and last input this row
+        /// coalesces. Absent on rows written before coalescing existed;
+        /// such a row is a single input, which spans zero milliseconds.
+        #[serde(default)]
+        #[ts(type = "number")]
+        span_ms: u64,
     },
     // The resolved backend's pane creation failed, so the
     // run fell back to a hidden pane instead of the one it actually

crates/runtime/src/adapter/event_sink.rs

@@ -134,12 +134,17 @@ pub enum AdapterEventPayload {
     /// A human typed directly into a TUI adapter's attached pane,
     /// bypassing the adapter's own input path. Carries no free text (the
     /// keystrokes themselves are never journaled) -- only that it
-    /// happened, on which pane. Maps to [`RuntimeEvent::OutOfBandInput`]
-    /// and additionally sets the run's `needsReconciliation` flag (see
-    /// [`DomainAdapterEventSink::emit`]).
+    /// happened, on which pane, how many times, and over what span.
+    /// Maps to [`RuntimeEvent::OutOfBandInput`] and additionally sets the
+    /// run's `needsReconciliation` flag (see
+    /// [`DomainAdapterEventSink::emit`]). Coalesced by the caller (see
+    /// `adapter::tui::oob_coalescer`) -- one payload per idle window, not
+    /// one per read.
     OutOfBandInput {
         backend: crew_protocol::DisplayBackend,
         pane_ref: String,
+        input_count: u64,
+        span_ms: u64,
     },
 }
 
@@ -406,13 +411,18 @@ impl DomainAdapterEventSink {
                 worker_id,
                 question: self.sanitize(text),
             },
-            AdapterEventPayload::OutOfBandInput { backend, pane_ref } => {
-                RuntimeEvent::OutOfBandInput {
-                    run_id,
-                    backend,
-                    pane_ref,
-                }
-            }
+            AdapterEventPayload::OutOfBandInput {
+                backend,
+                pane_ref,
+                input_count,
+                span_ms,
+            } => RuntimeEvent::OutOfBandInput {
+                run_id,
+                backend,
+                pane_ref,
+                input_count,
+                span_ms,
+            },
         }
     }
 }
@@ -934,6 +944,8 @@ mod out_of_band_input_tests {
             payload: AdapterEventPayload::OutOfBandInput {
                 backend: DisplayBackend::Tmux,
                 pane_ref: "%3".to_string(),
+                input_count: 7,
+                span_ms: 850,
             },
             cursor: None,
         })
@@ -946,10 +958,20 @@ mod out_of_band_input_tests {
                 run_id: got_run_id,
                 backend,
                 pane_ref,
+                input_count,
+                span_ms,
             } => {
                 assert_eq!(*got_run_id, run_id);
                 assert_eq!(*backend, DisplayBackend::Tmux);
                 assert_eq!(pane_ref, "%3");
+                assert_eq!(
+                    *input_count, 7,
+                    "the coalesced count must survive the round trip"
+                );
+                assert_eq!(
+                    *span_ms, 850,
+                    "the coalesced span must survive the round trip"
+                );
             }
             other => panic!("expected OutOfBandInput, got {other:?}"),
         }
@@ -996,6 +1018,8 @@ mod out_of_band_input_tests {
             payload: AdapterEventPayload::OutOfBandInput {
                 backend: DisplayBackend::Hidden,
                 pane_ref: String::new(),
+                input_count: 1,
+                span_ms: 0,
             },
             cursor: None,
         })

crates/runtime/src/adapter/tui/adapter.rs

@@ -1024,28 +1024,38 @@ impl<V: TuiVendor> TuiAdapter<V> {
             Arc::new(StdMutex::new((DisplayBackend::Hidden, String::new())));
         let attach_sink = Arc::clone(&sink);
         let attach_identity = Arc::clone(&pane_identity);
+        // Coalesced (see `oob_coalescer`'s own doc comment): a burst of
+        // out-of-band input reads journals one row, not one per read.
+        let coalescer = super::oob_coalescer::OobCoalescer::spawn(move |input_count, span_ms| {
+            let sink = Arc::clone(&attach_sink);
+            let (backend, pane_ref) = {
+                let guard = attach_identity
+                    .lock()
+                    .expect("pane identity mutex never poisoned");
+                (guard.0, guard.1.clone())
+            };
+            tokio::spawn(async move {
+                emit(
+                    &sink,
+                    run_id,
+                    task_id,
+                    worker_id,
+                    AdapterEventPayload::OutOfBandInput {
+                        backend,
+                        pane_ref,
+                        input_count,
+                        span_ms,
+                    },
+                    None,
+                )
+                .await;
+            });
+        });
         let attach = match AttachServer::start(
             self.socket_path(run_id),
             Arc::clone(&pty) as Arc<dyn AttachTarget>,
             Box::new(move |_bytes: Vec<u8>| {
-                let sink = Arc::clone(&attach_sink);
-                let (backend, pane_ref) = {
-                    let guard = attach_identity
-                        .lock()
-                        .expect("pane identity mutex never poisoned");
-                    (guard.0, guard.1.clone())
-                };
-                tokio::spawn(async move {
-                    emit(
-                        &sink,
-                        run_id,
-                        task_id,
-                        worker_id,
-                        AdapterEventPayload::OutOfBandInput { backend, pane_ref },
-                        None,
-                    )
-                    .await;
-                });
+                coalescer.notify();
             }),
         ) {
             Ok(server) => Arc::new(server),
@@ -1289,10 +1299,10 @@ impl<V: TuiVendor> TuiAdapter<V> {
         // itself indistinguishable from a real vendor session id once
         // journaled) -- `runs.vendor_session_id` simply stays unset until
         // a real `SessionMeta` entry establishes it.
-        if let Some(initial_session_id) = self
+        let initial_session_id = self
             .vendor
-            .session_id_from_transcript_path(&transcript_path)
-        {
+            .session_id_from_transcript_path(&transcript_path);
+        if let Some(initial_session_id) = initial_session_id.clone() {
             emit(
                 &sink,
                 run_id,
@@ -1320,8 +1330,38 @@ impl<V: TuiVendor> TuiAdapter<V> {
 
         let pump_sink = Arc::clone(&sink);
         tokio::spawn(async move {
+            // A transcript format that stamps every entry with its
+            // session id (Claude's does) would otherwise re-derive and
+            // re-journal an identical `VendorSessionEstablished` on
+            // every single entry -- 41 identical rows for one id,
+            // measured on a real transcript. Only a CHANGE reaches
+            // `emit_tui_event`; seeded from the initial guess above so a
+            // first real entry that merely confirms it is a no-op too,
+            // not a second, redundant journal write for the same fact.
+            //
+            // Skipping is gated on `cursor.is_none()` too, never on the
+            // session id alone: `cursor_placements` already gives a
+            // `SessionMeta` paired with another event on its line (the
+            // common shape -- e.g. `SessionMeta` + `AssistantText`) no
+            // cursor at all, since the OTHER event carries it -- that
+            // case is free to skip. The rare "hidden SessionMeta-only
+            // line" this module's tests call out DOES carry a cursor
+            // (nothing else on the line to carry it instead), so it is
+            // still emitted even when redundant: dropping that call
+            // would drop the only commit that advances
+            // `runs.transcript_cursor` past it, which a resume must
+            // never do (see `cursor_placements`'s own doc comment).
+            let mut last_vendor_session_id = initial_session_id;
             while let Some((tagged, _new_cursor)) = batch_rx.recv().await {
                 for (event, cursor) in cursor_placements(tagged) {
+                    if let TuiEvent::SessionMeta { vendor_session_id } = &event {
+                        let unchanged =
+                            last_vendor_session_id.as_deref() == Some(vendor_session_id.as_str());
+                        if unchanged && cursor.is_none() {
+                            continue;
+                        }
+                        last_vendor_session_id = Some(vendor_session_id.clone());
+                    }
                     emit_tui_event(&pump_sink, run_id, task_id, worker_id, event, cursor).await;
                 }
             }

crates/runtime/src/adapter/tui/mod.rs

@@ -31,6 +31,7 @@ mod classify;
 mod discovery;
 mod grid;
 mod input;
+pub(crate) mod oob_coalescer;
 mod screen;
 mod tailer;
 mod verify;