From 245ff6dfc2f5854828d56197a6d6a41aab708c13 Mon Sep 17 00:00:00 2001 From: Philip Deuchler Date: Sun, 26 Jul 2026 00:45:30 -0600 Subject: [PATCH 1/3] Route ChatGPT OAuth compaction to /responses/compact MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The OAuth URL rewrite returned a single constant for every Responses-shaped path, so `/v1/responses/compact` resolved to `.../backend-api/codex/responses` with the `/compact` suffix silently dropped. Dedicated compaction bodies were therefore posted to the streaming turn endpoint, which rejects them with `Store must be set to false` — making automatic compaction unreachable for OAuth sessions. Rewrite per-endpoint instead, matching the reference Codex client's `.../backend-api/codex/responses/compact`. The explicit match subsumes `is_chatgpt_codex_rewrite_path` and `is_responses_path_or_child`, so both helpers are gone and unmapped paths still fall through to the configured base_url. `store` is deliberately not added to the compaction body: the dedicated endpoint does not take it, and the reference client omits it too. Adding it would have masked the routing defect while leaving /compact unreachable. The existing test asserted the collapsed behavior; it now asserts each endpoint keeps its own path. Also adds the wire-level coverage the OpenAI compaction codec never had — only the OpenRouter encoder was tested — including encode shape, `store`/`stream` absence, prefix carry-forward, wrong-api-kind rejection, and decode success/failure. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 13 +++ crates/halter-providers/src/openai_codec.rs | 66 ++++++++++++++ .../src/responses_transport.rs | 85 ++++++++++++------- 3 files changed, 134 insertions(+), 30 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 21bc1f8..a30abe9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,6 +61,19 @@ log. and state-only intermediate changes ride the next event-ful flush or the final turn commit. +### Compaction + +#### Fixed + +- OpenAI OAuth (ChatGPT Codex) compaction reached the wrong endpoint. The + OAuth URL rewrite collapsed every Responses-shaped path onto a single + constant, so `/v1/responses/compact` resolved to `.../codex/responses` — + the streaming turn endpoint — which rejects compaction bodies with + `Store must be set to false`. Each ChatGPT-served endpoint now keeps its + own path, matching the reference Codex client + (`.../codex/responses/compact`). Automatic compaction was therefore + unreachable for OAuth sessions. + Blank-slate review fixes on top of the provider resilience primitive (issue #183). Highlights: diff --git a/crates/halter-providers/src/openai_codec.rs b/crates/halter-providers/src/openai_codec.rs index ab13a55..0437e27 100644 --- a/crates/halter-providers/src/openai_codec.rs +++ b/crates/halter-providers/src/openai_codec.rs @@ -1921,6 +1921,72 @@ mod tests { assert!(body.get("temperature").is_none()); } + /// The dedicated `/v1/responses/compact` endpoint takes the compaction + /// input shape only. `store` and `stream` belong to the streaming turn + /// endpoint and must stay absent here, matching the reference Codex + /// client's compaction payload. + #[test] + fn openai_compaction_request_uses_compact_endpoint_shape() { + let request = sample_compaction_request(ProviderKind::OpenAi); + let body = encode_responses_compact_request(&request).expect("encode compaction request"); + let input = body["input"].as_array().expect("input"); + + assert_eq!(body["model"], "gpt-5"); + assert_eq!(body["instructions"], "Summarize the session"); + assert!(body.get("store").is_none()); + assert!(body.get("stream").is_none()); + // The carried compacted prefix leads the input so successive + // compactions fold into one another instead of restarting. + assert_eq!(input[0]["role"], "developer"); + assert!( + input[0]["content"][0]["text"] + .as_str() + .expect("prefix text") + .contains("Earlier summary") + ); + assert!(input.iter().any(|item| item["role"] == "user")); + assert!(input.iter().any(|item| item["type"] == "function_call")); + assert!( + input + .iter() + .any(|item| item["type"] == "function_call_output") + ); + } + + #[test] + fn openai_compaction_request_rejects_non_responses_api_kind() { + let mut request = sample_compaction_request(ProviderKind::OpenAi); + request.model.api_kind = ApiKind::OpenAiChat; + + let error = + encode_responses_compact_request(&request).expect_err("chat api kind must not encode"); + + assert!(error.to_string().contains("unsupported api kind")); + } + + #[test] + fn openai_compaction_response_decodes_output_items() { + let response = json!({ + "output": [{"type": "compaction", "id": "cmp_1", "encrypted_content": "opaque"}], + "usage": {"input_tokens": 120, "output_tokens": 30}, + }); + + let decoded = decode_responses_compact_response(&response).expect("decode compaction"); + + assert_eq!(decoded.output.len(), 1); + assert_eq!(decoded.output[0]["type"], "compaction"); + assert_eq!(decoded.usage.input_tokens, 120); + assert_eq!(decoded.usage.output_tokens, 30); + } + + #[test] + fn openai_compaction_response_without_output_array_is_an_error() { + let error = decode_responses_compact_response(&json!({"usage": {}})) + .expect_err("missing output must fail"); + + assert!(error.to_string().contains("missing output array")); + } + #[test] fn openrouter_compaction_request_uses_responses_shape() { let request = sample_compaction_request(ProviderKind::OpenRouter); diff --git a/crates/halter-providers/src/responses_transport.rs b/crates/halter-providers/src/responses_transport.rs index ab50374..e4fc935 100644 --- a/crates/halter-providers/src/responses_transport.rs +++ b/crates/halter-providers/src/responses_transport.rs @@ -122,7 +122,9 @@ pub(crate) fn provider_error_from_openai(error: OpenAIError) -> ProviderError { const RESPONSES_PATH: &str = "/v1/responses"; const RESPONSES_COMPACT_PATH: &str = "/v1/responses/compact"; const CHAT_COMPLETIONS_PATH: &str = "/chat/completions"; -const CHATGPT_CODEX_RESPONSES_URL: &str = "https://chatgpt.com/backend-api/codex/responses"; +const CHATGPT_CODEX_BASE_URL: &str = "https://chatgpt.com/backend-api/codex"; +const CHATGPT_CODEX_RESPONSES_PATH: &str = "/responses"; +const CHATGPT_CODEX_RESPONSES_COMPACT_PATH: &str = "/responses/compact"; /// An event sent by the OpenAI Responses streaming API that the `async-openai` /// SDK does not yet model (e.g. `keepalive` heartbeat pings). @@ -414,36 +416,30 @@ fn transport_error_to_anyhow(error: TransportError) -> anyhow::Error { anyhow::Error::new(provider_error_from_transport(error)) } +/// ChatGPT-issued OAuth tokens are not accepted by OpenAI's public Platform +/// API, so OAuth mode intentionally ignores the configured `base_url` for the +/// endpoints the ChatGPT Codex backend serves. That backend exposes each +/// endpoint at its own path — dedicated compaction lives at +/// `/responses/compact`, not `/responses` — so the mapping must preserve the +/// suffix rather than collapsing every Responses-shaped path onto one URL. +/// Chat Completions-shaped payloads are the exception: the backend accepts +/// them at the plain `/responses` endpoint. This is private ChatGPT routing, +/// not public OpenAI Platform API behavior. fn provider_url(base_url: &str, path: &str, endpoint_mode: ResponsesEndpointMode) -> String { match endpoint_mode { ResponsesEndpointMode::PublicApi => join_url(base_url, path), - ResponsesEndpointMode::ChatGptCodexOAuth => { - if is_chatgpt_codex_rewrite_path(path) { - // ChatGPT-issued OAuth tokens are not accepted by OpenAI's - // public Platform API. The ChatGPT Codex backend currently - // accepts Responses-shaped requests, including dedicated - // compaction under `/v1/responses/...`, plus Chat - // Completions-shaped payloads at this single private endpoint. - // OAuth mode therefore intentionally ignores configured - // base_url for that prefix. This is private ChatGPT routing, - // not public OpenAI Platform API behavior. - CHATGPT_CODEX_RESPONSES_URL.to_owned() - } else { - join_url(base_url, path) + ResponsesEndpointMode::ChatGptCodexOAuth => match path { + RESPONSES_PATH | CHAT_COMPLETIONS_PATH => { + join_url(CHATGPT_CODEX_BASE_URL, CHATGPT_CODEX_RESPONSES_PATH) } - } + RESPONSES_COMPACT_PATH => { + join_url(CHATGPT_CODEX_BASE_URL, CHATGPT_CODEX_RESPONSES_COMPACT_PATH) + } + _ => join_url(base_url, path), + }, } } -fn is_chatgpt_codex_rewrite_path(path: &str) -> bool { - path == CHAT_COMPLETIONS_PATH || is_responses_path_or_child(path) -} - -fn is_responses_path_or_child(path: &str) -> bool { - path.strip_prefix(RESPONSES_PATH) - .is_some_and(|suffix| suffix.is_empty() || suffix.starts_with('/')) -} - #[derive(Debug, Clone)] struct OpenAiStreamRateLimitObserver { limiter: OpenAiRateLimiter, @@ -656,25 +652,54 @@ mod tests { } } + /// Each ChatGPT-served endpoint must keep its own path. Collapsing + /// `/v1/responses/compact` onto `/responses` sent dedicated compaction + /// bodies to the streaming turn endpoint, which rejects them with + /// "Store must be set to false". #[test] fn provider_url_rewrites_chatgpt_codex_oauth_paths() { let cases = [ - ("responses", RESPONSES_PATH), - ("responses_compact", RESPONSES_COMPACT_PATH), - ("responses_child", "/v1/responses/child/path"), - ("chat_completions", CHAT_COMPLETIONS_PATH), + ( + "responses", + RESPONSES_PATH, + "https://chatgpt.com/backend-api/codex/responses", + ), + ( + "responses_compact", + RESPONSES_COMPACT_PATH, + "https://chatgpt.com/backend-api/codex/responses/compact", + ), + ( + "chat_completions", + CHAT_COMPLETIONS_PATH, + "https://chatgpt.com/backend-api/codex/responses", + ), ]; - for (name, path) in cases { + for (name, path, want) in cases { let got = provider_url( "https://api.openai.com", path, ResponsesEndpointMode::ChatGptCodexOAuth, ); - assert_eq!(got, CHATGPT_CODEX_RESPONSES_URL, "{name}"); + assert_eq!(got, want, "{name}"); } } + /// Responses-prefixed paths the ChatGPT backend does not serve must fall + /// through to the configured base URL rather than being silently + /// rewritten onto an unrelated endpoint. + #[test] + fn provider_url_leaves_unmapped_oauth_responses_children_on_base_url() { + let got = provider_url( + "https://api.openai.com", + "/v1/responses/child/path", + ResponsesEndpointMode::ChatGptCodexOAuth, + ); + + assert_eq!(got, "https://api.openai.com/v1/responses/child/path"); + } + #[test] fn provider_url_leaves_non_responses_prefix_oauth_paths_on_base_url() { let cases = [ From c0b9570e59afb9619bb676025616f59fe65b5a50 Mon Sep 17 00:00:00 2001 From: Philip Deuchler Date: Sun, 26 Jul 2026 00:48:39 -0600 Subject: [PATCH 2/3] Degrade instead of failing the turn when auto-compaction cannot run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `plan()` propagated every compaction failure, so once a session crossed the compaction threshold any provider-side problem failed the turn — and context only grows, so the next turn failed the same way. Combined with the OAuth routing defect this made affected sessions unrecoverable. Automatic compaction is now best-effort: an unsupported provider, a missing compaction window, or a failing `Provider::compact` returns the uncompacted state carrying the reason, which `run_turn` publishes as `SessionEventPayload::Warning`. Manual `compact()` still propagates — the caller asked for compaction explicitly, so a silent no-op would misreport the result. `ContextPlan` gained `compaction_warning` (serde-defaulted, so existing event logs still deserialize) and `CompactionOutcome` gained `compaction_error` to carry it out of the engine. The three failure exits now route through one `degrade_or_fail`, and the four repeated "nothing compacted" outcome literals collapse into `uncompacted_outcome`, so the engine body is shorter than before. Tests cover both sides of each branch: degrade on provider error, unsupported provider, and absent window; no warning on success; and propagation on both manual paths. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 13 +- crates/halter-protocol/src/lib.rs | 5 + crates/halter-runtime/src/context.rs | 416 ++++++++++++++++++++++----- crates/halter-runtime/src/prompt.rs | 5 + crates/halter-runtime/src/session.rs | 12 + 5 files changed, 372 insertions(+), 79 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a30abe9..235dc4c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -72,7 +72,18 @@ log. `Store must be set to false`. Each ChatGPT-served endpoint now keeps its own path, matching the reference Codex client (`.../codex/responses/compact`). Automatic compaction was therefore - unreachable for OAuth sessions. + unreachable for OAuth sessions, which combined with the issue below made + every turn past the compaction threshold fail outright. + +#### Changed + +- Automatic compaction is now best-effort. A provider that cannot compact — + failing endpoint, missing capability, or no compaction window — degrades + the turn to an uncompacted context and emits + `SessionEventPayload::Warning` instead of failing the turn. `ContextPlan` + gained `compaction_warning` to carry the reason. Manual `compact()` is + unchanged and still propagates its errors, since the caller asked for + compaction explicitly. Blank-slate review fixes on top of the provider resilience primitive (issue #183). Highlights: diff --git a/crates/halter-protocol/src/lib.rs b/crates/halter-protocol/src/lib.rs index 2e22127..0cf8488 100644 --- a/crates/halter-protocol/src/lib.rs +++ b/crates/halter-protocol/src/lib.rs @@ -1768,6 +1768,11 @@ pub struct ContextPlan { /// If the planner compacted messages this turn, the result is here. /// The caller should apply it to `SessionState` after using the plan. pub compaction: Option, + /// Set when automatic compaction was due but could not run. The plan is + /// still valid and uncompacted, so the turn proceeds; the caller should + /// surface this so a degraded context does not look like a healthy one. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub compaction_warning: Option, /// When set, the codec should chain via `previous_response_id`. #[serde(default, skip_serializing_if = "Option::is_none")] pub previous_response_id: Option, diff --git a/crates/halter-runtime/src/context.rs b/crates/halter-runtime/src/context.rs index 5b3362f..1afe137 100644 --- a/crates/halter-runtime/src/context.rs +++ b/crates/halter-runtime/src/context.rs @@ -9,7 +9,7 @@ use halter_protocol::{ use halter_providers::Provider; use serde_json::Value; use sha2::{Digest, Sha256}; -use tracing::info; +use tracing::{info, warn}; use crate::compaction::{ ContextSettings, estimate_context_tokens, prepare_compaction, render_compaction_event_summary, @@ -40,6 +40,10 @@ pub struct CompactionOutcome { pub messages: Vec, pub compacted_prefix: Vec, pub compaction: Option, + /// Why automatic compaction did not run, when it was due but failed. + /// Always `None` for manual compaction, which propagates its errors + /// instead. + pub compaction_error: Option, pub session_start_latch: Option, } @@ -110,6 +114,7 @@ impl CompactionOutcome { messages, compacted_prefix, compaction, + compaction_error: _, session_start_latch, } = self; CompactionEffects { @@ -234,48 +239,38 @@ impl DefaultContextManager { &state.messages, ); if !mode.is_forced() && !should_trigger_compaction(estimated_tokens, &self.settings) { - return Ok(CompactionOutcome { - messages: state.messages.clone(), - compacted_prefix: state.compacted_prefix.clone(), - compaction: None, - session_start_latch: mode.session_start_latch(), - }); + return Ok(uncompacted_outcome(state, mode, None)); } let capabilities = compaction_provider.capabilities(); if !capabilities.supports_compaction { - anyhow::bail!( - "failed to compact session: provider '{}' does not support compaction", - compaction_model.provider + return degrade_or_fail( + state, + mode, + format!( + "failed to compact session: provider '{}' does not support compaction", + compaction_model.provider + ), ); } let Some(window) = compaction_provider.compaction_window(&state.messages) else { - if mode.is_forced() { - anyhow::bail!( + return degrade_or_fail( + state, + mode, + format!( "failed to compact session: provider '{}' did not provide a compaction window", compaction_model.provider - ); - } - return Ok(CompactionOutcome { - messages: state.messages.clone(), - compacted_prefix: state.compacted_prefix.clone(), - compaction: None, - session_start_latch: mode.session_start_latch(), - }); + ), + ); }; let compacted_context = CompactedContext::from(state.compacted_prefix.clone()); let preparation = prepare_compaction(&self.settings, &compacted_context, window); if compacted_context.is_empty() && preparation.compact_messages.is_empty() { - return Ok(CompactionOutcome { - messages: state.messages.clone(), - compacted_prefix: state.compacted_prefix.clone(), - compaction: None, - session_start_latch: mode.session_start_latch(), - }); + return Ok(uncompacted_outcome(state, mode, None)); } - let response = compaction_provider + let response = match compaction_provider .compact( ProviderCompactionRequest { session_id: blueprint.session_id.clone(), @@ -287,7 +282,11 @@ impl DefaultContextManager { }, tokio_util::sync::CancellationToken::new(), ) - .await?; + .await + { + Ok(response) => response, + Err(error) => return degrade_or_fail(state, mode, format!("{error:#}")), + }; let summary = render_compaction_event_summary( preparation.compacted_message_count, response.output.len(), @@ -302,11 +301,45 @@ impl DefaultContextManager { compacted_count: preparation.compacted_message_count, summary, }), + compaction_error: None, session_start_latch: mode.session_start_latch(), }) } } +/// Automatic compaction is best-effort: a provider that cannot compact must +/// degrade the turn to an uncompacted context rather than fail it, because +/// the alternative is that every turn past the threshold becomes unrecoverable. +/// Manual compaction propagates instead — the caller asked for compaction +/// explicitly and needs to know it did not happen. +fn degrade_or_fail( + state: &SessionState, + mode: CompactionMode<'_>, + error: String, +) -> anyhow::Result { + if mode.is_forced() { + anyhow::bail!(error); + } + warn!(error, "automatic compaction failed; continuing uncompacted"); + Ok(uncompacted_outcome(state, mode, Some(error))) +} + +/// The turn's state left exactly as it was, carrying any reason compaction +/// did not run. +fn uncompacted_outcome( + state: &SessionState, + mode: CompactionMode<'_>, + compaction_error: Option, +) -> CompactionOutcome { + CompactionOutcome { + messages: state.messages.clone(), + compacted_prefix: state.compacted_prefix.clone(), + compaction: None, + compaction_error, + session_start_latch: mode.session_start_latch(), + } +} + #[async_trait] impl ContextManager for DefaultContextManager { async fn plan( @@ -393,6 +426,7 @@ impl ContextManager for DefaultContextManager { messages: outcome.messages, estimated_tokens, compaction: outcome.compaction, + compaction_warning: outcome.compaction_error, previous_response_id, new_messages_start, }) @@ -508,79 +542,300 @@ mod tests { use super::*; + fn sample_blueprint() -> SessionBlueprint { + SessionBlueprint { + session_id: SessionId::new(), + parent_session_id: None, + default_model: "default".into(), + subagent_model: "subagent".into(), + subagent_event_forwarding: SubagentEventForwarding::Off, + snapshot_revision: "r1".into(), + working_dir: ".".into(), + system_prompt_seed: Vec::new(), + max_turns: None, + subagent_depth: 0, + } + } + + fn sample_observed() -> ObservedState { + ObservedState { + cwd: ".".into(), + git_branch: None, + git_dirty: None, + now_utc: Utc::now(), + env_facts: Default::default(), + } + } + + fn sample_model() -> ResolvedModel { + ResolvedModel { + role: ModelRole::default(), + id: ModelId::from("default"), + provider: ProviderName::from("fake"), + provider_kind: ProviderKind::Fake, + api_kind: halter_protocol::ApiKind::Fake, + model: "fake".to_owned(), + max_input_tokens: None, + max_output_tokens: None, + reasoning: None, + tokens_per_minute: None, + } + } + + /// A manager whose threshold is low enough that every plan compacts. + fn always_compacting_manager() -> DefaultContextManager { + DefaultContextManager::new(1, 0, halter_protocol::PruneSignalThreshold::Normal) + } + + async fn plan_with( + manager: &DefaultContextManager, + state: &SessionState, + provider: &StubProvider, + ) -> anyhow::Result { + manager + .plan( + &sample_blueprint(), + state, + &sample_observed(), + &ResourceSnapshot::empty(), + &[], + &sample_model(), + provider, + ) + .await + } + #[tokio::test] async fn plan_disables_previous_response_chaining_when_compacted_prefix_exists() { - let manager = DefaultContextManager::default(); - let outcome = manager - .plan( - &SessionBlueprint { - session_id: SessionId::new(), - parent_session_id: None, - default_model: "default".into(), - subagent_model: "subagent".into(), - subagent_event_forwarding: SubagentEventForwarding::Off, - snapshot_revision: "r1".into(), - working_dir: ".".into(), - system_prompt_seed: Vec::new(), - max_turns: None, - subagent_depth: 0, - }, + let outcome = plan_with( + &DefaultContextManager::default(), + &SessionState { + compacted_prefix: vec![serde_json::json!({ + "type": "compaction", + "id": "cmp_1", + "encrypted_content": "x", + })], + summaries: vec![SummarySlice { + id: "summary-1".to_owned(), + text: "summary".to_owned(), + }], + messages: vec![Message::User(UserMessage::text("hello"))], + last_response_id: Some("resp_1".to_owned()), + messages_seen_by_provider: 1, + ..SessionState::default() + }, + &StubProvider::working(), + ) + .await + .expect("plan"); + + assert!(outcome.previous_response_id.is_none()); + } + + /// A provider that cannot compact must not take the turn down with it. + /// Before this, `plan()` propagated the error and every turn past the + /// compaction threshold failed with no way to recover. + #[tokio::test] + async fn plan_degrades_to_uncompacted_context_when_compaction_fails() { + let state = SessionState { + messages: vec![Message::User(UserMessage::text("hello"))], + ..SessionState::default() + }; + + let plan = plan_with( + &always_compacting_manager(), + &state, + &StubProvider::failing("compaction endpoint exploded"), + ) + .await + .expect("plan must survive a failed compaction"); + + assert!(plan.compaction.is_none()); + assert_eq!(plan.messages, state.messages); + assert!( + plan.compaction_warning + .as_deref() + .is_some_and(|warning| warning.contains("compaction endpoint exploded")), + "expected the provider error to be carried out, got {:?}", + plan.compaction_warning + ); + } + + #[tokio::test] + async fn plan_degrades_when_provider_does_not_support_compaction() { + let plan = plan_with( + &always_compacting_manager(), + &SessionState { + messages: vec![Message::User(UserMessage::text("hello"))], + ..SessionState::default() + }, + &StubProvider::without_compaction(), + ) + .await + .expect("plan must survive a provider that cannot compact"); + + assert!(plan.compaction.is_none()); + assert!( + plan.compaction_warning + .as_deref() + .is_some_and(|warning| warning.contains("does not support compaction")), + "got {:?}", + plan.compaction_warning + ); + } + + /// A provider advertising compaction but yielding no window used to + /// disable compaction silently on the auto path. It is a misconfiguration, + /// so it must be reported rather than swallowed. + #[tokio::test] + async fn plan_degrades_when_provider_yields_no_compaction_window() { + let plan = plan_with( + &always_compacting_manager(), + &SessionState { + messages: vec![Message::User(UserMessage::text("hello"))], + ..SessionState::default() + }, + &StubProvider::without_window(), + ) + .await + .expect("plan must survive a provider that yields no window"); + + assert!(plan.compaction.is_none()); + assert!( + plan.compaction_warning + .as_deref() + .is_some_and(|warning| warning.contains("did not provide a compaction window")), + "got {:?}", + plan.compaction_warning + ); + } + + #[tokio::test] + async fn plan_reports_no_warning_when_compaction_succeeds() { + let plan = plan_with( + &always_compacting_manager(), + &SessionState { + messages: vec![Message::User(UserMessage::text("hello"))], + ..SessionState::default() + }, + &StubProvider::working(), + ) + .await + .expect("plan"); + + assert!(plan.compaction.is_some()); + assert_eq!(plan.compaction_warning, None); + } + + /// Manual compaction keeps propagating: the caller asked for it + /// explicitly and a silent no-op would be a lie. + #[tokio::test] + async fn compact_now_propagates_provider_failures() { + let error = always_compacting_manager() + .compact_now( + &sample_blueprint(), &SessionState { - compacted_prefix: vec![serde_json::json!({ - "type": "compaction", - "id": "cmp_1", - "encrypted_content": "x", - })], - summaries: vec![SummarySlice { - id: "summary-1".to_owned(), - text: "summary".to_owned(), - }], messages: vec![Message::User(UserMessage::text("hello"))], - last_response_id: Some("resp_1".to_owned()), - messages_seen_by_provider: 1, ..SessionState::default() }, - &ObservedState { - cwd: ".".into(), - git_branch: None, - git_dirty: None, - now_utc: Utc::now(), - env_facts: Default::default(), - }, + &sample_observed(), &ResourceSnapshot::empty(), &[], - &ResolvedModel { - role: ModelRole::default(), - id: ModelId::from("default"), - provider: ProviderName::from("fake"), - provider_kind: ProviderKind::Fake, - api_kind: halter_protocol::ApiKind::Fake, - model: "fake".to_owned(), - max_input_tokens: None, - max_output_tokens: None, - reasoning: None, - tokens_per_minute: None, + &sample_model(), + &StubProvider::failing("compaction endpoint exploded"), + None, + ) + .await + .expect_err("manual compaction must surface provider failures"); + + assert!(error.to_string().contains("compaction endpoint exploded")); + } + + #[tokio::test] + async fn compact_now_propagates_unsupported_compaction() { + let error = always_compacting_manager() + .compact_now( + &sample_blueprint(), + &SessionState { + messages: vec![Message::User(UserMessage::text("hello"))], + ..SessionState::default() }, - &NoopProvider, + &sample_observed(), + &ResourceSnapshot::empty(), + &[], + &sample_model(), + &StubProvider::without_compaction(), + None, ) .await - .expect("plan"); + .expect_err("manual compaction must surface unsupported providers"); - assert!(outcome.previous_response_id.is_none()); + assert!(error.to_string().contains("does not support compaction")); } - struct NoopProvider; + /// Compaction provider stub. `supports_compaction` drives the advertised + /// capability, `offers_window` whether a compaction window is produced, + /// and `compact_error` makes the compaction call fail. + struct StubProvider { + supports_compaction: bool, + offers_window: bool, + compact_error: Option<&'static str>, + } + + impl StubProvider { + fn working() -> Self { + Self { + supports_compaction: true, + offers_window: true, + compact_error: None, + } + } + + fn failing(error: &'static str) -> Self { + Self { + compact_error: Some(error), + ..Self::working() + } + } + + fn without_compaction() -> Self { + Self { + supports_compaction: false, + ..Self::working() + } + } + + /// Advertises compaction but never yields a window — the shape a + /// provider takes when it forgets to override `compaction_window`. + fn without_window() -> Self { + Self { + offers_window: false, + ..Self::working() + } + } + } #[async_trait] - impl Provider for NoopProvider { + impl Provider for StubProvider { fn capabilities(&self) -> ProviderCapabilities { ProviderCapabilities { - supports_compaction: true, + supports_compaction: self.supports_compaction, tool_call_id_policy: ToolCallIdPolicy::ProviderSupplied, ..ProviderCapabilities::default() } } + fn compaction_window( + &self, + messages: &[Message], + ) -> Option { + self.offers_window.then(|| { + halter_protocol::CompactionWindow::preserve_latest_assistant_response_block( + messages, + ) + }) + } + async fn stream( &self, _request: halter_protocol::ProviderRequest, @@ -599,6 +854,9 @@ mod tests { _request: ProviderCompactionRequest, _cancel: tokio_util::sync::CancellationToken, ) -> anyhow::Result { + if let Some(error) = self.compact_error { + anyhow::bail!(error); + } Ok(halter_protocol::ProviderCompactionResponse { output: vec![serde_json::json!({ "type": "compaction", @@ -630,6 +888,7 @@ mod tests { compacted_count: 2, summary: "squashed".to_owned(), }), + compaction_error: None, session_start_latch: None, }; @@ -658,6 +917,7 @@ mod tests { messages: Vec::new(), compacted_prefix: Vec::new(), compaction: None, + compaction_error: None, session_start_latch: None, }; diff --git a/crates/halter-runtime/src/prompt.rs b/crates/halter-runtime/src/prompt.rs index 8e1aab1..9be509d 100644 --- a/crates/halter-runtime/src/prompt.rs +++ b/crates/halter-runtime/src/prompt.rs @@ -400,6 +400,7 @@ mod tests { messages: vec![Message::User(UserMessage::text("hello"))], estimated_tokens: 10, compaction: None, + compaction_warning: None, previous_response_id: None, new_messages_start: 0, }; @@ -459,6 +460,7 @@ mod tests { messages: vec![Message::User(UserMessage::text("hello"))], estimated_tokens: 10, compaction: None, + compaction_warning: None, previous_response_id: None, new_messages_start: 0, }; @@ -523,6 +525,7 @@ mod tests { messages: vec![Message::User(UserMessage::text("hello"))], estimated_tokens: 10, compaction: None, + compaction_warning: None, previous_response_id: None, new_messages_start: 0, }; @@ -603,6 +606,7 @@ mod tests { messages: vec![Message::User(UserMessage::text("hi"))], estimated_tokens: 0, compaction: None, + compaction_warning: None, previous_response_id: None, new_messages_start: 0, }; @@ -665,6 +669,7 @@ mod tests { messages: vec![Message::User(UserMessage::text("hi"))], estimated_tokens: 0, compaction: None, + compaction_warning: None, previous_response_id: None, new_messages_start: 0, }; diff --git a/crates/halter-runtime/src/session.rs b/crates/halter-runtime/src/session.rs index 2405038..d60b7fa 100644 --- a/crates/halter-runtime/src/session.rs +++ b/crates/halter-runtime/src/session.rs @@ -1235,10 +1235,22 @@ impl SessionHandle { ) .await?; + if let Some(warning) = plan.compaction_warning.as_ref() { + self.push_event( + &mut events, + SessionEventPayload::Warning { + message: format!( + "automatic compaction did not run; continuing with an uncompacted context: {warning}" + ), + }, + ); + } + let plan_outcome = crate::CompactionOutcome { messages: plan.messages.clone(), compacted_prefix: plan.compacted_prefix.clone(), compaction: plan.compaction.clone(), + compaction_error: None, session_start_latch: None, }; if let Some((result, effects)) = plan_outcome.apply_with_effects(&mut state) { From f59ae192f3e44faa67f1bda4e6def79e7a1e8ec9 Mon Sep 17 00:00:00 2001 From: Philip Deuchler Date: Sun, 26 Jul 2026 00:51:30 -0600 Subject: [PATCH 3/3] Anchor context estimation on provider-reported usage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Compaction triggering ran entirely off a character heuristic (chars * 10/37, documented as +/-20% for code-heavy text) applied to the whole transcript, so the error compounded with context size — exactly where the threshold decision matters most. Providers report the true context size with every assistant response. `estimate_context_tokens` now scans back for the most recent usable report, takes it as ground truth, and estimates only the messages after it, bounding heuristic error to one turn's tail. The report already covers the prompt, summaries, and compacted prefix, so those are not added on top of an anchor. Sessions with no usable report — fresh, or all turns failed — fall back to the previous whole-transcript estimate. Interrupted turns, errored turns, and zero-token reports are rejected: they describe a partial request and would peg the estimate far below the real context, stopping compaction from ever firing. Two hazards this surfaced: Compaction preserves a message tail, and the assistant message in that tail still reports the *pre*-compaction context. Anchoring on it would make every following turn see the old figure and compact again, indefinitely. `SessionState::usage_anchor_floor` bounds the scan; compaction advances it past the preserved tail in both the runtime and the event fold so replayed and resumed sessions agree. Forked subagents start with no anchor, since the parent's reports describe a different system prompt and tool set. The providers used opposite cache-token conventions: OpenAI's `input_tokens` includes cached tokens, Anthropic's excludes them. No single formula is correct for both, and plain `input_tokens` under-counts a mostly-cached Anthropic prompt badly enough to delay compaction past the context window. `Usage::input_tokens` is now defined as total input including cache traffic, with the cache counters kept as breakdown subsets, and the Anthropic decoder normalized to match. This changes reported Anthropic input totals, including accumulated `usage_so_far`, for turns recorded after this change. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 25 ++ crates/halter-protocol/src/fold.rs | 12 +- crates/halter-protocol/src/lib.rs | 26 ++ .../halter-providers/src/anthropic_codec.rs | 44 ++-- crates/halter-runtime/src/compaction.rs | 243 +++++++++++++++++- crates/halter-runtime/src/context.rs | 79 ++++++ crates/halter-runtime/src/subagent_session.rs | 5 + 7 files changed, 409 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 235dc4c..6b27a82 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -77,6 +77,31 @@ log. #### Changed +- Context estimation now anchors on provider-reported usage instead of + estimating the whole transcript with a character heuristic. When the + transcript contains a usable report from a completed assistant turn, + `estimate_context_tokens` takes that figure as ground truth and estimates + only the messages after it, bounding heuristic error (documented at ±20% + for code-heavy text) to one turn's tail rather than letting it compound + across the entire context. Interrupted turns, errored turns, and zero-usage + reports are rejected as anchors, and a session with none falls back to the + previous whole-transcript estimate. +- `SessionState` gained `usage_anchor_floor`. Compaction preserves a tail of + real messages whose usage describes the *pre*-compaction context; without a + floor, that stale figure re-triggers compaction on every following turn + indefinitely. Compaction advances the floor past the preserved tail, in + both the runtime (`CompactionEffects::apply`) and the event fold, so + replayed and resumed sessions agree. Forked subagents start with no anchor, + since the parent's reports describe a different system prompt and tool set. +- **`Usage::input_tokens` now uniformly means total input including cache + traffic.** OpenAI already reported it that way; Anthropic reports cache + reads and writes as separate counters, and its decoder now folds them in, + keeping `cache_read_input_tokens` / `cache_creation_input_tokens` as + breakdown fields. Without this, a mostly-cached Anthropic prompt reported a + small `input_tokens` and context budgeting under-counted the live context. + This changes reported Anthropic input totals — including the accumulated + `usage_so_far` — for turns recorded after the upgrade; historical values in + existing sessions are unaffected. - Automatic compaction is now best-effort. A provider that cannot compact — failing endpoint, missing capability, or no compaction window — degrades the turn to an uncompacted context and emits diff --git a/crates/halter-protocol/src/fold.rs b/crates/halter-protocol/src/fold.rs index 91bb4a5..3640dcf 100644 --- a/crates/halter-protocol/src/fold.rs +++ b/crates/halter-protocol/src/fold.rs @@ -24,9 +24,10 @@ //! provider-chaining fields) are deliberately *not* event-covered: they are //! carried by the checkpoint, which the runtime writes on every //! state-changing commit. The one exception is that `ContextCompacted` -//! effects also reset `last_response_id` / `messages_seen_by_provider`, -//! mirroring the runtime's compaction rules so a mid-replay view is not left -//! pointing at a provider response chain that predates the rewrite. +//! effects also reset `last_response_id` / `messages_seen_by_provider` and +//! advance `usage_anchor_floor`, mirroring the runtime's compaction rules so +//! a mid-replay view is not left pointing at a provider response chain — or +//! at reported token usage — that predates the rewrite. //! //! The store conformance suite locks the invariant in: after any sequence of //! commits, folding the full log over a default state must agree with the @@ -58,6 +59,8 @@ pub fn apply_event(state: &mut SessionState, payload: &SessionEventPayload) { // runtime so replayed views never chain onto pre-rewrite context. state.last_response_id = None; state.messages_seen_by_provider = 0; + // Usage reported by the preserved tail predates this rewrite. + state.usage_anchor_floor = state.messages.len(); } SessionEventPayload::ContextCompacted { effects: None, .. } | SessionEventPayload::SessionStarted @@ -216,6 +219,9 @@ mod tests { assert_eq!(state.compacted_prefix, vec![json!({"kind": "prefix"})]); assert_eq!(state.last_response_id, None); assert_eq!(state.messages_seen_by_provider, 0); + // Replay must reach the same anchor floor the runtime sets, or a + // resumed session would re-anchor on pre-compaction usage. + assert_eq!(state.usage_anchor_floor, window.len()); } #[test] diff --git a/crates/halter-protocol/src/lib.rs b/crates/halter-protocol/src/lib.rs index 0cf8488..5752673 100644 --- a/crates/halter-protocol/src/lib.rs +++ b/crates/halter-protocol/src/lib.rs @@ -314,13 +314,31 @@ pub enum PanelIsolation { #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Eq, Default)] /// Token accounting reported by providers and accumulated by sessions. pub struct Usage { + /// Total input tokens for the request, **including** tokens served from + /// or written to the prompt cache. Provider codecs normalize to this + /// convention: OpenAI already reports the total, while Anthropic reports + /// cache traffic separately and its decoder folds it in. pub input_tokens: u64, pub output_tokens: u64, + /// Portion of `input_tokens` written to the prompt cache. A breakdown of + /// `input_tokens`, not an addition to it. pub cache_creation_input_tokens: u64, + /// Portion of `input_tokens` served from the prompt cache. A breakdown of + /// `input_tokens`, not an addition to it. pub cache_read_input_tokens: u64, } impl Usage { + /// Total tokens the model had in context when it produced this response: + /// everything it read plus everything it wrote. This is the size the + /// *next* request's input starts from, which is what context budgeting + /// needs — as opposed to the per-request cost, which is what the + /// individual fields report. + #[must_use] + pub const fn context_tokens(&self) -> u64 { + self.input_tokens.saturating_add(self.output_tokens) + } + /// Accumulate `delta` into `self`, saturating at `u64::MAX` so lifetime /// counters can never overflow. Both the session runtime and the event /// fold ([`fold::apply_event`]) use this, keeping the persisted @@ -1534,6 +1552,14 @@ pub struct SessionState { /// Messages at indices `[0..messages_seen_by_provider)` don't need re-sending. #[serde(default)] pub messages_seen_by_provider: usize, + /// Index into `messages` before which reported `Usage` no longer + /// describes the live context. Compaction rewrites history but preserves + /// a tail of real messages, and any assistant message in that tail still + /// carries the usage from *before* the rewrite. Anchoring context + /// estimates on that stale figure would re-trigger compaction every turn, + /// so compaction moves this floor past the preserved tail. + #[serde(default)] + pub usage_anchor_floor: usize, } #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Eq)] diff --git a/crates/halter-providers/src/anthropic_codec.rs b/crates/halter-providers/src/anthropic_codec.rs index fdcf3fa..b13c594 100644 --- a/crates/halter-providers/src/anthropic_codec.rs +++ b/crates/halter-providers/src/anthropic_codec.rs @@ -645,24 +645,29 @@ fn decode_stop_reason(response: &Value) -> StopReason { ) } +/// Anthropic reports `input_tokens` *excluding* cache traffic, with cache +/// reads and writes as separate sibling counters. [`Usage::input_tokens`] is +/// defined as the total including cache traffic (OpenAI's convention), so +/// fold the cache counters in here and keep them as breakdown fields. Without +/// this, a mostly-cached prompt reports a tiny `input_tokens` and context +/// budgeting badly under-counts the live context. fn decode_usage(response: &Value) -> Usage { - Usage { - input_tokens: response - .pointer("/usage/input_tokens") - .and_then(Value::as_u64) - .unwrap_or_default(), - output_tokens: response - .pointer("/usage/output_tokens") - .and_then(Value::as_u64) - .unwrap_or_default(), - cache_creation_input_tokens: response - .pointer("/usage/cache_creation_input_tokens") - .and_then(Value::as_u64) - .unwrap_or_default(), - cache_read_input_tokens: response - .pointer("/usage/cache_read_input_tokens") + let field = |name: &str| { + response + .pointer(&format!("/usage/{name}")) .and_then(Value::as_u64) - .unwrap_or_default(), + .unwrap_or_default() + }; + let cache_creation_input_tokens = field("cache_creation_input_tokens"); + let cache_read_input_tokens = field("cache_read_input_tokens"); + + Usage { + input_tokens: field("input_tokens") + .saturating_add(cache_creation_input_tokens) + .saturating_add(cache_read_input_tokens), + output_tokens: field("output_tokens"), + cache_creation_input_tokens, + cache_read_input_tokens, } } @@ -1295,9 +1300,14 @@ mod tests { event, StreamEvent::ToolArgsDelta { delta, .. } if delta.contains("README.md") ))); + // input_tokens folds in cache traffic: 11 uncached + 2 written + 3 read. assert!(events.iter().any(|event| matches!( event, - StreamEvent::UsageUpdate { usage } if usage.input_tokens == 11 && usage.output_tokens == 7 + StreamEvent::UsageUpdate { usage } + if usage.input_tokens == 16 + && usage.output_tokens == 7 + && usage.cache_creation_input_tokens == 2 + && usage.cache_read_input_tokens == 3 ))); assert!(events.iter().any(|event| matches!( event, diff --git a/crates/halter-runtime/src/compaction.rs b/crates/halter-runtime/src/compaction.rs index a950811..c7e597e 100644 --- a/crates/halter-runtime/src/compaction.rs +++ b/crates/halter-runtime/src/compaction.rs @@ -86,18 +86,80 @@ pub fn prepare_compaction( } } +/// Estimate total context tokens for prompt, summaries, compacted prefix, and +/// messages. +/// +/// Prefers ground truth over the character heuristic. Providers report the +/// exact context size with every assistant response, so when the transcript +/// contains a usable report this anchors on it and estimates only the +/// messages that arrived afterwards — bounding heuristic error to one turn's +/// tail instead of letting it compound across the whole transcript. The +/// reported figure already accounts for the prompt, summaries, and compacted +/// prefix, so those are *not* added on top of an anchor. +/// +/// Falls back to estimating everything when no usable report exists: a fresh +/// session, a transcript whose assistant turns all failed, or one whose +/// reports predate the last compaction (see +/// [`SessionState::usage_anchor_floor`](halter_protocol::SessionState)). #[must_use] -/// Estimate total context tokens for prompt, summaries, compacted prefix, and messages. pub fn estimate_context_tokens( prompt_segments: &[PromptSegment], summaries: &[SummarySlice], compacted_prefix: &[Value], messages: &[Message], + usage_anchor_floor: usize, ) -> u64 { - estimate_segment_tokens(prompt_segments) - + estimate_summary_tokens(summaries) - + estimate_compacted_prefix_tokens(compacted_prefix) - + estimate_messages_tokens(messages) + match find_usage_anchor(messages, usage_anchor_floor) { + Some(anchor) => anchor + .context_tokens + .saturating_add(estimate_messages_tokens(&messages[anchor.index + 1..])), + None => { + estimate_segment_tokens(prompt_segments) + + estimate_summary_tokens(summaries) + + estimate_compacted_prefix_tokens(compacted_prefix) + + estimate_messages_tokens(messages) + } + } +} + +/// A provider-reported context measurement usable as an estimation anchor. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct UsageAnchor { + /// Index in `messages` of the assistant message that reported it. + pub index: usize, + /// Context size the provider reported at that point. + pub context_tokens: u64, +} + +/// Find the most recent assistant message at or after `floor` carrying a +/// usable context report. +/// +/// A report is usable only if the turn actually completed: interrupted and +/// errored turns can report partial or zero usage, which would anchor the +/// whole estimate to a figure far below the real context. +#[must_use] +pub fn find_usage_anchor(messages: &[Message], floor: usize) -> Option { + messages + .iter() + .enumerate() + .skip(floor) + .rev() + .find_map(|(index, message)| { + let Message::Assistant(assistant) = message else { + return None; + }; + if matches!( + assistant.stop_reason, + Some(halter_protocol::StopReason::Interrupted | halter_protocol::StopReason::Error) + ) { + return None; + } + let context_tokens = assistant.usage.as_ref()?.context_tokens(); + (context_tokens > 0).then_some(UsageAnchor { + index, + context_tokens, + }) + }) } #[must_use] @@ -417,6 +479,177 @@ mod tests { use super::*; + fn assistant( + text: &str, + stop_reason: Option, + usage: Option, + ) -> Message { + Message::Assistant(AssistantMessage { + id: MessageId::new(), + created_at: Utc::now(), + parts: vec![AssistantPart::Text { + text: text.to_owned(), + }], + stop_reason, + usage, + replay_meta: Default::default(), + }) + } + + fn reported(input_tokens: u64, output_tokens: u64) -> Option { + Some(halter_protocol::Usage { + input_tokens, + output_tokens, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }) + } + + /// The whole point of anchoring: a provider report replaces the heuristic + /// for everything up to and including the reporting message, and the + /// prompt/summaries/prefix are not re-added on top because the report + /// already covered them. + #[test] + fn estimate_context_tokens_anchors_on_reported_usage() { + let messages = vec![ + Message::User(UserMessage::text("x".repeat(4_000).as_str())), + assistant( + "done", + Some(halter_protocol::StopReason::EndTurn), + reported(50_000, 200), + ), + Message::User(UserMessage::text("follow up")), + ]; + let summaries = vec![SummarySlice { + id: "s".to_owned(), + text: "y".repeat(4_000), + }]; + + let estimated = estimate_context_tokens(&[], &summaries, &[], &messages, 0); + + // 50_000 + 200 reported, plus only the trailing user message. + assert_eq!( + estimated, + 50_200 + estimate_message_tokens(&messages[2]), + "anchor must exclude everything the report already covered" + ); + } + + #[test] + fn estimate_context_tokens_falls_back_to_heuristic_without_usage() { + let messages = vec![ + Message::User(UserMessage::text("hello")), + assistant("done", Some(halter_protocol::StopReason::EndTurn), None), + ]; + + let estimated = estimate_context_tokens(&[], &[], &[], &messages, 0); + + assert_eq!(estimated, estimate_messages_tokens(&messages)); + } + + /// Interrupted and errored turns report partial or zero usage. Anchoring + /// on those would peg the estimate far below the real context and stop + /// compaction from ever firing. + #[test] + fn find_usage_anchor_rejects_unusable_reports() { + struct Case { + name: &'static str, + message: Message, + } + + let cases = [ + Case { + name: "interrupted", + message: assistant( + "partial", + Some(halter_protocol::StopReason::Interrupted), + reported(50_000, 10), + ), + }, + Case { + name: "error", + message: assistant( + "boom", + Some(halter_protocol::StopReason::Error), + reported(50_000, 10), + ), + }, + Case { + name: "zero usage", + message: assistant( + "empty", + Some(halter_protocol::StopReason::EndTurn), + reported(0, 0), + ), + }, + Case { + name: "no usage", + message: assistant("none", Some(halter_protocol::StopReason::EndTurn), None), + }, + Case { + name: "not an assistant message", + message: Message::User(UserMessage::text("user")), + }, + ]; + + for case in cases { + assert_eq!( + find_usage_anchor(std::slice::from_ref(&case.message), 0), + None, + "{}: must not anchor", + case.name + ); + } + } + + #[test] + fn find_usage_anchor_picks_the_most_recent_usable_report() { + let messages = vec![ + assistant( + "first", + Some(halter_protocol::StopReason::EndTurn), + reported(10, 1), + ), + assistant( + "second", + Some(halter_protocol::StopReason::EndTurn), + reported(20, 2), + ), + assistant( + "third", + Some(halter_protocol::StopReason::Error), + reported(999, 0), + ), + ]; + + let anchor = find_usage_anchor(&messages, 0).expect("anchor"); + + assert_eq!(anchor.index, 1); + assert_eq!(anchor.context_tokens, 22); + } + + /// Compaction preserves a message tail whose usage describes the + /// pre-compaction context. Without the floor, that stale figure would + /// re-trigger compaction on every following turn. + #[test] + fn find_usage_anchor_ignores_reports_below_the_floor() { + let messages = vec![ + assistant( + "pre-compaction", + Some(halter_protocol::StopReason::EndTurn), + reported(80_000, 500), + ), + Message::User(UserMessage::text("after compaction")), + ]; + + assert_eq!(find_usage_anchor(&messages, 1), None); + + // ...and the estimate falls back to the heuristic rather than 80_500. + let estimated = estimate_context_tokens(&[], &[], &[], &messages, 1); + assert_eq!(estimated, estimate_messages_tokens(&messages)); + assert!(estimated < 1_000); + } + #[test] fn estimate_text_tokens_uses_character_count() { assert_eq!(estimate_text_tokens(""), 0); diff --git a/crates/halter-runtime/src/context.rs b/crates/halter-runtime/src/context.rs index 1afe137..c0e6337 100644 --- a/crates/halter-runtime/src/context.rs +++ b/crates/halter-runtime/src/context.rs @@ -73,6 +73,10 @@ impl CompactionEffects { // injected, so the next request must replay everything. state.last_response_id = None; state.messages_seen_by_provider = 0; + // Every usage report in the preserved tail describes the + // pre-compaction context and would re-trigger compaction on the + // very next turn. + state.usage_anchor_floor = state.messages.len(); } if let Some(source) = session_start_latch { state.pending_session_start_source = Some(source); @@ -109,6 +113,17 @@ impl CompactionOutcome { effects.apply(state).map(|result| (result, record)) } + /// Usage-anchor floor this outcome implies. Compaction invalidates every + /// report in the tail it preserved; otherwise the session's existing + /// floor still stands. + fn usage_anchor_floor(&self, state: &SessionState) -> usize { + if self.compaction.is_some() { + self.messages.len() + } else { + state.usage_anchor_floor + } + } + fn into_effects(self) -> CompactionEffects { let CompactionOutcome { messages, @@ -237,6 +252,7 @@ impl DefaultContextManager { &state.summaries, &state.compacted_prefix, &state.messages, + state.usage_anchor_floor, ); if !mode.is_forced() && !should_trigger_compaction(estimated_tokens, &self.settings) { return Ok(uncompacted_outcome(state, mode, None)); @@ -384,6 +400,7 @@ impl ContextManager for DefaultContextManager { &state.summaries, &outcome.compacted_prefix, &outcome.messages, + outcome.usage_anchor_floor(state), ); if let Some(compaction) = outcome.compaction.as_ref() { @@ -710,6 +727,68 @@ mod tests { ); } + /// Regression guard for a compaction loop. Compaction preserves a tail of + /// real messages, and the assistant message in that tail still reports the + /// pre-compaction context size. If the estimator kept anchoring on it, + /// every subsequent turn would see the old (large) figure and compact + /// again — forever. The floor advance is what stops that. + #[tokio::test] + async fn compaction_does_not_retrigger_on_the_following_turn() { + let manager = DefaultContextManager::new( + /*compaction_threshold*/ 1_000, + /*pre_compaction_target*/ 500, + halter_protocol::PruneSignalThreshold::Normal, + ); + let mut state = SessionState { + messages: vec![ + Message::User(UserMessage::text("hello")), + Message::Assistant(halter_protocol::AssistantMessage { + id: halter_protocol::MessageId::new(), + created_at: Utc::now(), + parts: vec![halter_protocol::AssistantPart::Text { + text: "done".to_owned(), + }], + stop_reason: Some(halter_protocol::StopReason::EndTurn), + usage: Some(Usage { + input_tokens: 80_000, + output_tokens: 500, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }), + replay_meta: Default::default(), + }), + ], + ..SessionState::default() + }; + + let first = plan_with(&manager, &state, &StubProvider::working()) + .await + .expect("first plan"); + assert!( + first.compaction.is_some(), + "80_500 reported tokens must exceed the 1_000 threshold" + ); + + let outcome = CompactionOutcome { + messages: first.messages.clone(), + compacted_prefix: first.compacted_prefix.clone(), + compaction: first.compaction.clone(), + compaction_error: None, + session_start_latch: None, + }; + outcome.apply(&mut state); + assert_eq!(state.usage_anchor_floor, state.messages.len()); + + let second = plan_with(&manager, &state, &StubProvider::working()) + .await + .expect("second plan"); + + assert!( + second.compaction.is_none(), + "stale pre-compaction usage must not re-trigger compaction" + ); + } + #[tokio::test] async fn plan_reports_no_warning_when_compaction_succeeds() { let plan = plan_with( diff --git a/crates/halter-runtime/src/subagent_session.rs b/crates/halter-runtime/src/subagent_session.rs index e57043c..f9ce720 100644 --- a/crates/halter-runtime/src/subagent_session.rs +++ b/crates/halter-runtime/src/subagent_session.rs @@ -82,6 +82,11 @@ pub fn build_subagent_state( pending_warning_messages: parent.state.pending_warning_messages.clone(), last_response_id: None, messages_seen_by_provider: 0, + // The forked transcript carries the parent's usage reports, but the + // child runs a different system prompt and tool set, so those figures + // do not describe its context. Start with no anchor and let the + // child's first response supply one. + usage_anchor_floor: parent.state.messages.len(), } }