diff --git a/CHANGELOG.md b/CHANGELOG.md index 21bc1f8..6b27a82 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,6 +61,55 @@ 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, which combined with the issue below made + every turn past the compaction threshold fail outright. + +#### 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 + `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/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 2e22127..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)] @@ -1768,6 +1794,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-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-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 = [ 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 5b3362f..c0e6337 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, } @@ -69,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); @@ -105,11 +113,23 @@ 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, compacted_prefix, compaction, + compaction_error: _, session_start_latch, } = self; CompactionEffects { @@ -232,50 +252,41 @@ 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(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 +298,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 +317,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( @@ -351,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() { @@ -393,6 +443,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 +559,362 @@ 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, - }, - &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(), + 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 + ); + } + + /// 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( + &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 { 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 +933,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 +967,7 @@ mod tests { compacted_count: 2, summary: "squashed".to_owned(), }), + compaction_error: None, session_start_latch: None, }; @@ -658,6 +996,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) { 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(), } }