diff --git a/runtime/crates/muse-context/src/actor.rs b/runtime/crates/muse-context/src/actor.rs index 179b59d2..9c6c374f 100644 --- a/runtime/crates/muse-context/src/actor.rs +++ b/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 = None; let mut last_session_id: Option = None; let mut runner_error: Option = 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 diff --git a/runtime/crates/muse-context/tests/dispatch_via_runner.rs b/runtime/crates/muse-context/tests/dispatch_via_runner.rs index 17d5b7ee..7da60548 100644 --- a/runtime/crates/muse-context/tests/dispatch_via_runner.rs +++ b/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 diff --git a/runtime/crates/muse-runner/src/claude_code/fold.rs b/runtime/crates/muse-runner/src/claude_code/fold.rs index cedd0ebb..86c7fa61 100644 --- a/runtime/crates/muse-runner/src/claude_code/fold.rs +++ b/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(); diff --git a/runtime/crates/muse-runner/src/events.rs b/runtime/crates/muse-runner/src/events.rs index ff7b687c..de414cca 100644 --- a/runtime/crates/muse-runner/src/events.rs +++ b/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 { diff --git a/runtime/crates/muse-runner/tests/claude_code.rs b/runtime/crates/muse-runner/tests/claude_code.rs index 97f0caf3..ad70cf6a 100644 --- a/runtime/crates/muse-runner/tests/claude_code.rs +++ b/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", };