Watermark measures the request context, not cumulative uncached input

runtime/crates/muse-context/src/actor.rs

@@ -284,6 +284,15 @@ pub struct ContextState {
     /// cold-resume TTFT crossed an environmental ~15s connection
     /// kill (2026-08-26 incident, forensics 08-27).
     pub last_cycle_input_tokens: u32,
+    /// Largest single-request context (input + cache reads + cache
+    /// creation) seen on the runner path, carried across cycles
+    /// until a cycle observes a non-zero value again. This is what
+    /// a `--resume` will send next, so it is what the 15s edge wall
+    /// is measured against; cumulative `input_tokens` excludes cache
+    /// reads and never crossed the watermark while the session grew
+    /// to 139K (2026-08-28). Feeds `last_cycle_input_tokens` on
+    /// completion and the first-failure rotation on api-unreachable.
+    pub last_cycle_context_peak: u32,
     /// `api_retry` events the CLI announced during the most recent
     /// COMPLETED cycle. Nonzero on a successful cycle means TTFT is
     /// already brushing the ~15s connection kill — the tapes show
@@ -490,6 +499,7 @@ impl Actor for Context {
             cycle_failure_streak: 0,
             compact_at_input_tokens: args.compact_at_input_tokens,
             last_cycle_input_tokens: 0,
+            last_cycle_context_peak: 0,
             last_cycle_api_retries: 0,
             reflection_due: false,
             operator_framing: args.operator_framing,
@@ -1074,7 +1084,10 @@ async fn run_wake(
                 CycleStopReason::IterationCap => CycleEndOutcome::IterationCap,
                 CycleStopReason::MaxTokens => CycleEndOutcome::MaxTokens,
             };
-            state.last_cycle_input_tokens = outcome.total_usage.input_tokens;
+            state.last_cycle_input_tokens = outcome
+                .total_usage
+                .input_tokens
+                .max(state.last_cycle_context_peak);
             (end_outcome, outcome.iterations, outcome.total_usage)
         }
         Err(err) => {
@@ -1494,6 +1507,7 @@ async fn maybe_compact(state: &mut ContextState, mut capture: Option<&mut Captur
                 info!("substrate session rotated at fold; next wake starts fresh from canonical");
             }
             state.last_cycle_input_tokens = 0;
+            state.last_cycle_context_peak = 0;
             state.last_cycle_api_retries = 0;
             true
         }
@@ -1701,6 +1715,7 @@ pub async fn dispatch_via_claude_code(
     let mut api_retries_seen: u32 = 0;
     let mut last_conversation = Conversation::new();
     let mut last_usage = TokenUsage::default();
+    let mut peak_context: u32 = 0;
     let mut last_stop: Option<muse_runner::CycleStopReason> = None;
     let mut last_session_id: Option<SessionId> = None;
     let mut runner_error: Option<muse_runner::CycleError> = None;
@@ -1772,6 +1787,9 @@ pub async fn dispatch_via_claude_code(
                         last_session_id = session_id;
                         break;
                     }
+                    CycleEvent::ContextObserved { tokens } => {
+                        peak_context = peak_context.max(tokens);
+                    }
                     // Started / Thinking / ToolUse / Usage are
                     // intermediate; nothing to accumulate here.
                     CycleEvent::Started { .. }
@@ -1793,6 +1811,14 @@ pub async fn dispatch_via_claude_code(
         }
     }
 
+    // Remember the largest request this cycle made — on success it
+    // drives the watermark; on failure it is what the next resume
+    // would send. A cycle that died before its first turn (peak 0)
+    // keeps the previous cycle's reading.
+    if peak_context > 0 {
+        state.last_cycle_context_peak = peak_context;
+    }
+
     // Persist session id (if any) so the next cycle can `--resume`.
     // Mutating requires matching on the `ClaudeCode` variant — the
     // type prevents this from being set on a substrate that doesn't
@@ -1904,19 +1930,38 @@ pub async fn dispatch_via_claude_code(
             }
             if matches!(stop_reason, muse_runner::CycleStopReason::ApiUnreachable) {
                 state.api_unreachable_streak = state.api_unreachable_streak.saturating_add(1);
-                if state.api_unreachable_streak >= API_UNREACHABLE_SESSION_HEDGE {
+                // A session already known to be at or past the
+                // watermark is the wall, not the weather: a resume
+                // sends that whole context cold and the edge kills
+                // it at 15s every time. Waiting out the streak just
+                // costs the cadence (2026-08-28: three cycles lost
+                // resuming a 139K session). Rotate on the first miss.
+                let heavy = state.compact_at_input_tokens > 0
+                    && state.last_cycle_context_peak >= state.compact_at_input_tokens;
+                if heavy || state.api_unreachable_streak >= API_UNREACHABLE_SESSION_HEDGE {
                     if let DispatchMode::ClaudeCode {
                         last_session_id, ..
                     } = &mut state.dispatch
                         && last_session_id.take().is_some()
                     {
-                        warn!(
-                            streak = state.api_unreachable_streak,
-                            "consecutive api-unreachable cycles; dropping the resumable session \
-                             as a poison hedge — next wake starts fresh from canonical"
-                        );
+                        if heavy {
+                            warn!(
+                                peak_context_tokens = state.last_cycle_context_peak,
+                                watermark = state.compact_at_input_tokens,
+                                "api-unreachable on a session at the watermark — the wall, not \
+                                 the weather; dropping the resumable session now, next wake \
+                                 starts fresh from canonical"
+                            );
+                        } else {
+                            warn!(
+                                streak = state.api_unreachable_streak,
+                                "consecutive api-unreachable cycles; dropping the resumable session \
+                                 as a poison hedge — next wake starts fresh from canonical"
+                            );
+                        }
                     }
                     state.api_unreachable_streak = 0;
+                    state.last_cycle_context_peak = 0;
                 } else {
                     warn!(
                         streak = state.api_unreachable_streak,
@@ -2016,6 +2061,7 @@ pub fn record_runner_event_to_capture(capture: &mut Capture, event: &CycleEvent)
         CycleEvent::RawLine { .. }
         | CycleEvent::Started { .. }
         | CycleEvent::Usage(_)
+        | CycleEvent::ContextObserved { .. }
         | CycleEvent::Finished { .. } => None,
     };
     if let Some(ev) = cap_event

runtime/crates/muse-context/tests/dispatch_via_runner.rs

@@ -168,6 +168,7 @@ fn build_claude_code_state(
         cycle_failure_streak: 0,
         compact_at_input_tokens: 0,
         last_cycle_input_tokens: 0,
+        last_cycle_context_peak: 0,
         last_cycle_api_retries: 0,
         reflection_due: false,
         operator_framing: String::new(),
@@ -577,6 +578,87 @@ async fn three_consecutive_api_unreachable_cycles_drop_the_session() {
     shutdown(identity, join).await;
 }
 
+/// 2026-08-28 (three cycles lost resuming a 139K session): a session
+/// already at the watermark is the wall, not the weather. The first
+/// api-unreachable on it rotates the session instead of waiting out
+/// the streak.
+#[tokio::test]
+async fn heavy_session_rotates_on_the_first_api_unreachable() {
+    let sid = SessionId::parse("2e72eba7-32fc-45ef-9608-f3f741ccd9b0").unwrap();
+    let (ctx, identity, join) = spawn_tool_ctx().await;
+    let runner = ScriptedRunner::new(vec![
+        Ok(CycleEvent::ContextObserved { tokens: 139_000 }),
+        Ok(finished_with(
+            muse_runner::CycleStopReason::ApiUnreachable,
+            None,
+        )),
+    ]);
+    let mut state = build_claude_code_state(runner.clone(), ctx, Some(sid));
+    state.compact_at_input_tokens = 80_000;
+    let config = state.cycle_config.clone();
+    let cancel = CancellationToken::new();
+
+    let _ = dispatch_via_claude_code(&mut state, runner.as_ref(), &config, &cancel, None, None)
+        .await
+        .expect_err("unreachable propagates");
+    let DispatchMode::ClaudeCode {
+        last_session_id, ..
+    } = &state.dispatch
+    else {
+        panic!("dispatch mode changed");
+    };
+    assert!(
+        last_session_id.is_none(),
+        "a session at the watermark is dropped on its first unreachable"
+    );
+    assert_eq!(state.api_unreachable_streak, 0);
+    assert_eq!(
+        state.last_cycle_context_peak, 0,
+        "peak resets with the session"
+    );
+    assert!(!state.reflection_due, "rotation is not a quarantine");
+
+    shutdown(identity, join).await;
+}
+
+/// A cycle that died before its first turn keeps the previous
+/// reading: the peak is what the next resume would send, and a
+/// wall hit on turn one says nothing new about the size.
+#[tokio::test]
+async fn context_peak_is_recorded_on_completion_and_survives_a_turnless_failure() {
+    let sid = SessionId::parse("2e72eba7-32fc-45ef-9608-f3f741ccd9b0").unwrap();
+    let (ctx, identity, join) = spawn_tool_ctx().await;
+    let runner = ScriptedRunner::new(vec![
+        Ok(CycleEvent::ContextObserved { tokens: 52_972 }),
+        Ok(CycleEvent::ContextObserved { tokens: 102_709 }),
+        Ok(finished_with(
+            muse_runner::CycleStopReason::EndTurn,
+            Some(sid.clone()),
+        )),
+    ]);
+    let mut state = build_claude_code_state(runner.clone(), ctx, Some(sid));
+    let config = state.cycle_config.clone();
+    let cancel = CancellationToken::new();
+    dispatch_via_claude_code(&mut state, runner.as_ref(), &config, &cancel, None, None)
+        .await
+        .expect("completes");
+    assert_eq!(state.last_cycle_context_peak, 102_709);
+
+    // Turn-less failure (the 23:00 shape): no ContextObserved at all.
+    let runner = ScriptedRunner::new(vec![Ok(finished_with(
+        muse_runner::CycleStopReason::ApiUnreachable,
+        None,
+    ))]);
+    let _ = dispatch_via_claude_code(&mut state, runner.as_ref(), &config, &cancel, None, None)
+        .await
+        .expect_err("unreachable propagates");
+    // compact_at is 0 in this fixture, so the heavy rule is off and
+    // the streak rule keeps the session; the reading itself survives.
+    assert_eq!(state.last_cycle_context_peak, 102_709);
+
+    shutdown(identity, join).await;
+}
+
 /// Capture translation for the runner path (2026-08-12 contract):
 /// tool events AND her words/thinking land as typed capture events —
 /// `Started` / `Usage` / `Finished` / `RawLine` have no typed-tape

runtime/crates/muse-runner/src/claude_code/fold.rs

@@ -225,7 +225,12 @@ impl ConversationFold {
         // We treat the result's usage as authoritative and prefer
         // the cumulative one, but additive per-turn is the right
         // behaviour for the local-loop substrate parity.
+        let mut context_tokens: u32 = 0;
         if let Some(usage) = message.usage {
+            context_tokens = usage
+                .input_tokens
+                .saturating_add(usage.cache_creation_input_tokens)
+                .saturating_add(usage.cache_read_input_tokens);
             accumulate_usage(
                 &mut self.total_usage,
                 usage.input_tokens,
@@ -236,6 +241,11 @@ impl ConversationFold {
         }
 
         let mut emit = Vec::new();
+        if context_tokens > 0 {
+            emit.push(CycleEvent::ContextObserved {
+                tokens: context_tokens,
+            });
+        }
         let pending = self.pending.get_or_insert_with(PendingTurn::default);
 
         for block in message.content {
@@ -533,6 +543,29 @@ mod tests {
         assert_eq!(stop_reason, CycleStopReason::ApiUnreachable);
     }
 
+    #[test]
+    fn assistant_usage_emits_the_request_context_size() {
+        // The watermark's true input (2026-08-28): what the provider
+        // billed for THIS request — input + cache creation + cache
+        // reads — not the cumulative uncached count, which stayed in
+        // the double digits while a session grew to 139K.
+        let mut fold = ConversationFold::default();
+        let ev = parse_line(
+            r#"{"type":"assistant","message":{"id":"msg_1","content":[{"type":"text","text":"checking"}],"usage":{"input_tokens":21,"output_tokens":5,"cache_creation_input_tokens":77199,"cache_read_input_tokens":25000}},"session_id":"abc"}"#,
+        )
+        .expect("ok")
+        .expect("some");
+        let FoldStep::Emit(events) = fold.step(ev) else {
+            panic!("assistant text should emit");
+        };
+        assert!(
+            events
+                .iter()
+                .any(|e| matches!(e, CycleEvent::ContextObserved { tokens } if *tokens == 102_220)),
+            "{events:?}"
+        );
+    }
+
     #[test]
     fn api_retry_rate_limit_events_accumulate_evidence() {
         let mut fold = ConversationFold::default();

runtime/crates/muse-runner/src/events.rs

@@ -78,6 +78,14 @@ pub enum CycleEvent {
     /// LLM response; Claude Code emits one per internal iteration
     /// (plus a final cumulative total in `Finished`).
     Usage(TokenUsage),
+    /// The size of one request's context, as the provider billed it
+    /// (`input + cache_creation + cache_read` on an `assistant`
+    /// event). Emitted per turn on the Claude Code substrate. This —
+    /// not cumulative usage — is what the daemon's token watermark
+    /// must compare against: the cumulative `input_tokens` excludes
+    /// cache reads and stayed in the double digits while the
+    /// session grew to 139K and hit the 15s edge wall (2026-08-28).
+    ContextObserved { tokens: u32 },
     /// Terminal event. The runner has finished and the stream will
     /// close after this event.
     Finished {

runtime/crates/muse-runner/tests/claude_code.rs

@@ -132,6 +132,7 @@ async fn claude_code_runner_round_trips_a_simple_prompt() {
                             CycleEvent::ToolUse { .. } => "ToolUse",
                             CycleEvent::ToolResult { .. } => "ToolResult",
                             CycleEvent::Usage(_) => "Usage",
+                            CycleEvent::ContextObserved { .. } => "ContextObserved",
                             CycleEvent::RawLine { .. } => "RawLine",
                             CycleEvent::Finished { .. } => "Finished",
                         };