From 793905e4774689a81bfc3f99b8e5454f04d47eae Mon Sep 17 00:00:00 2001 From: lsaether <25539605+lsaether@users.noreply.github.com> Date: Thu, 28 May 2026 14:24:20 -0500 Subject: [PATCH 1/3] fix(rooms): clear stranded compaction "active" state at turn settlement MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `compaction_state.active` is set by a Hermes `context compression started` stderr line and cleared by the matching `done`. If the `done` never arrives — the compaction failed, or its line was dropped by the lossy stderr pump — the flag stayed `true` forever, so /debug/sessions and session/attach snapshots reported a perpetual "compacting" state to late joiners (issue #66, item 1). Hermes compacts within a prompt turn, so by the time that turn's response settles any compaction it triggered has completed or been abandoned. Clear the stranded transient flag at turn settlement — the same signal a live client uses to clear its own "compacting" affordance (amux/turn_complete). Only the transient active/pending fields reset; compaction_count and the last_* history are durable and untouched. No frame is emitted (live clients already clear on amux/turn_complete; this only realigns snapshot state for future attachers). Tests: stranded `started` with no `done` clears on turn settlement without fabricating a count; a completed compaction is unaffected. Refs #66. --- src/room/state.rs | 148 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 148 insertions(+) diff --git a/src/room/state.rs b/src/room/state.rs index 81c39cf..ba41a54 100644 --- a/src/room/state.rs +++ b/src/room/state.rs @@ -3148,6 +3148,39 @@ impl RoomInner { } } + /// Clear a stranded "compacting" state at turn settlement. + /// + /// `compaction_state.active` is set by a `context compression started` + /// stderr line and is meant to clear on the matching `done` line. If + /// that `done` never arrives — the compaction failed, or its stderr + /// line was dropped by the lossy pump (see `STDERR_CAPACITY`) — the + /// flag would otherwise stay `true` forever, leaving `/debug/sessions` + /// and `session/attach` snapshots reporting a perpetual "compacting" + /// state to late joiners. + /// + /// Hermes compacts *within* a prompt turn (to make room while + /// answering), so by the time that turn's response settles any + /// compaction it triggered has either completed (cleared by `done`) or + /// been abandoned. Turn settlement is therefore the natural bound: + /// it's the same signal a live client uses to clear its own + /// "compacting" affordance (`amux/turn_complete`). We only reset the + /// transient `active`/`pending` fields — `compaction_count` and the + /// `last_*` history are durable and left untouched. Does not emit a + /// frame: live clients already clear on `amux/turn_complete`, and this + /// only realigns snapshot state for future attachers. + fn clear_stale_compaction_at_turn_end(&mut self) { + if !self.compaction_state.active { + return; + } + tracing::debug!( + session = %self.room_id, + "compaction marked active at turn settlement with no matching `done`; \ + clearing stranded transient compaction state", + ); + self.compaction_state.active = false; + self.compaction_state.pending_hermes_session_id = None; + } + fn handle_agent_line(&mut self, line: Vec) -> AgentLineAction { let frame = match Incoming::parse(&line) { Ok(f) => f, @@ -3431,6 +3464,10 @@ impl RoomInner { // permission timeout fires internally without writing a // response frame). self.sweep_stale_agent_pending("mux:turn-ended"); + // A compaction triggered during this turn must have settled by + // now; clear any stranded "compacting" state whose `done` line + // never arrived (failed compaction or dropped stderr line). + self.clear_stale_compaction_at_turn_end(); if let Some(turn_id) = turn_id { self.emit_turn_complete(turn_id, resp.result.as_ref()); if let Some(queue_item_id) = pr.queue_item_id.as_deref() { @@ -4517,6 +4554,117 @@ mod tests { assert!(!attach_summary.active); assert_eq!(attach_summary.last_source.as_deref(), Some("hermes_stderr")); } + + #[test] + fn stranded_compaction_active_clears_at_turn_settlement() { + // A `started` with no matching `done` (failed compaction, or the + // `done` line dropped by the lossy stderr pump) must not leave + // `active = true` forever in snapshots. Turn settlement is the + // bound: any compaction triggered during the turn has settled by + // the time its response returns. + let mut inner = test_inner(); + inner.replay_log = Some(VecDeque::new()); + inner.set_canonical_session_id_with_reason("acp-stable", EndReason::SessionLoad); + + // Simulate an in-flight prompt turn. + let mux_id = 5u64; + inner.pending.insert( + mux_id, + PendingRequest { + peer_id: "p".into(), + original_id: Id::Number(1), + handshake: None, + decorate_session_list: false, + deliver_response: false, + queue_item_id: None, + }, + ); + inner.active_turn_mux_id = Some(mux_id); + inner.active_amux_turn_id = Some(AmuxTurnId(1)); + + // Compaction starts mid-turn; the `done` line never arrives. + inner.handle_agent_stderr_line( + b"agent.conversation_compression: context compression started: session=hs messages=50" + .to_vec(), + ); + assert!(inner.compaction_state.active, "started sets active"); + assert_eq!(inner.compaction_count, 0, "no done yet"); + + // The prompt turn settles. + let resp = IncomingResponse { + jsonrpc: JsonRpcVersion, + id: Id::Number(mux_id as i64), + result: Some(json!({ "stopReason": "end_turn" })), + error: None, + }; + inner.route_agent_response(resp); + + assert!( + !inner.compaction_state.active, + "stranded compaction active must clear at turn settlement", + ); + assert_eq!( + inner.compaction_count, 0, + "clearing transient state must not fabricate a completed compaction", + ); + assert!( + inner.compaction_state.pending_hermes_session_id.is_none(), + "pending hermes session id cleared with the active flag", + ); + // Durable history is left intact for audit. + assert!(inner.compaction_state.last_started_at.is_some()); + + let snap = inner.build_snapshot(false); + let compaction = snap.compaction.expect("compaction lifecycle present"); + assert!(!compaction.active, "snapshot reflects cleared state"); + } + + #[test] + fn completed_compaction_is_unaffected_by_turn_settlement_clear() { + // Guard the happy path: a `done` already cleared `active` and + // bumped the count; turn settlement must not disturb that. + let mut inner = test_inner(); + inner.replay_log = Some(VecDeque::new()); + inner.set_canonical_session_id_with_reason("acp-stable", EndReason::SessionLoad); + + let mux_id = 9u64; + inner.pending.insert( + mux_id, + PendingRequest { + peer_id: "p".into(), + original_id: Id::Number(1), + handshake: None, + decorate_session_list: false, + deliver_response: false, + queue_item_id: None, + }, + ); + inner.active_turn_mux_id = Some(mux_id); + inner.active_amux_turn_id = Some(AmuxTurnId(1)); + + inner.handle_agent_stderr_line( + b"agent.conversation_compression: context compression started: session=hs messages=50" + .to_vec(), + ); + inner.handle_agent_stderr_line(b"agent.conversation_compression: context compression done: session=hs messages=50->9 tokens=~700".to_vec()); + assert!(!inner.compaction_state.active); + assert_eq!(inner.compaction_count, 1); + + let resp = IncomingResponse { + jsonrpc: JsonRpcVersion, + id: Id::Number(mux_id as i64), + result: Some(json!({ "stopReason": "end_turn" })), + error: None, + }; + inner.route_agent_response(resp); + + assert_eq!( + inner.compaction_count, 1, + "turn settlement must not change a completed compaction's count", + ); + assert!(!inner.compaction_state.active); + assert!(inner.compaction_state.last_completed_at.is_some()); + } } #[derive(Debug, Clone)] From 71dc7c8c88e040fea0ab9167193fde37eb5a4b54 Mon Sep 17 00:00:00 2001 From: lsaether <25539605+lsaether@users.noreply.github.com> Date: Thu, 28 May 2026 14:39:06 -0500 Subject: [PATCH 2/3] fix(rooms): make stranded-compaction clear durable across replay-store restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The turn-settlement clear added in this branch was in-memory only and emitted no frame, so it didn't survive restart: hydration rebuilds compaction_state purely from persisted frames, and a persisted context_compaction_started with no matching _done resurrected active=true after a restart — defeating the snapshot fix for persistent rooms. Teach hydration the same turn-settlement bound the live path uses: when rebuild_compaction_from_frame replays a persisted amux/turn_complete and compaction is still marked active, clear the transient active/pending fields (count + history untouched). amux/turn_complete is broadcast and therefore persisted on every prompt-turn settlement, so no new lifecycle marker frame is needed. Test: started -> turn_complete -> restart asserts restored compaction.active == false (verified it fails without the hydration arm). Refs #66. --- src/room/state.rs | 88 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 88 insertions(+) diff --git a/src/room/state.rs b/src/room/state.rs index ba41a54..3a7a69c 100644 --- a/src/room/state.rs +++ b/src/room/state.rs @@ -470,6 +470,21 @@ fn rebuild_compaction_from_frame( .unwrap_or_else(|| compaction_count.saturating_add(1)); *compaction_count = (*compaction_count).max(observed); } + "amux/turn_complete" => { + // Durable mirror of `clear_stale_compaction_at_turn_end`: a + // `started` with no matching `done` by the time a turn settles + // was abandoned. The live path clears the transient flag in + // memory and emits no frame, so without applying the same + // bound here, restart hydration would replay the persisted + // `started` and resurrect `active = true` forever. + // `amux/turn_complete` is broadcast (and therefore persisted) + // on every prompt-turn settlement, so it's a reliable durable + // bound. Count and history timestamps stay untouched. + if compaction_state.active { + compaction_state.active = false; + compaction_state.pending_hermes_session_id = None; + } + } _ => {} } } @@ -4619,6 +4634,79 @@ mod tests { assert!(!compaction.active, "snapshot reflects cleared state"); } + #[test] + fn hydration_clears_stranded_compaction_via_persisted_turn_complete() { + // Durability guard for the turn-settlement clear: a `started` + // with no `done`, followed by a settled turn, must NOT come back + // as active=true after a replay-store restart. The persisted + // amux/turn_complete is the durable bound. + let dir = std::env::temp_dir().join(format!( + "amux-hydration-stranded-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(), + )); + std::fs::create_dir_all(&dir).unwrap(); + let store = Arc::new(ReplayStore::open(&dir).unwrap()); + + // Run #1: started with no done, then the prompt turn settles + // (persists amux/turn_complete). + { + let mut inner = test_inner_with_store(store.clone()); + inner.set_canonical_session_id_with_reason("acp-stable", EndReason::SessionLoad); + let mux_id = 5u64; + inner.pending.insert( + mux_id, + PendingRequest { + peer_id: "p".into(), + original_id: Id::Number(1), + handshake: None, + decorate_session_list: false, + deliver_response: false, + queue_item_id: None, + }, + ); + inner.active_turn_mux_id = Some(mux_id); + inner.active_amux_turn_id = Some(AmuxTurnId(1)); + inner.handle_agent_stderr_line( + b"agent.conversation_compression: context compression started: session=hs messages=50" + .to_vec(), + ); + let resp = IncomingResponse { + jsonrpc: JsonRpcVersion, + id: Id::Number(mux_id as i64), + result: Some(json!({ "stopReason": "end_turn" })), + error: None, + }; + inner.route_agent_response(resp); + assert!( + !inner + .build_snapshot(false) + .compaction + .expect("live compaction state") + .active, + "live path clears at turn settlement", + ); + } + + // Run #2: rebuild from the same store. Without the turn_complete + // hydration arm this comes back active=true. + let restored = test_inner_with_store(store); + let snap = restored.build_snapshot(false); + let compaction = snap + .compaction + .expect("compaction lifecycle restored from the started frame"); + assert!( + !compaction.active, + "persisted amux/turn_complete must clear stranded active across restart", + ); + assert_eq!(snap.compaction_count, 0, "no done means no count"); + + std::fs::remove_dir_all(&dir).ok(); + } + #[test] fn completed_compaction_is_unaffected_by_turn_settlement_clear() { // Guard the happy path: a `done` already cleared `active` and From 5fd8a2ef410f1651747c76ea27b1ee42c4a331e4 Mon Sep 17 00:00:00 2001 From: lsaether <25539605+lsaether@users.noreply.github.com> Date: Thu, 28 May 2026 14:42:35 -0500 Subject: [PATCH 3/3] style: fold turn_complete hydration arm into a match guard (clippy 1.96) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit clippy::collapsible_match (newer on CI's toolchain than my local one) flagged the `if compaction_state.active` nested inside the `amux/turn_complete` match arm. Rewrite as a guarded arm `"amux/turn_complete" if compaction_state.active =>`. Behavior is identical — a turn_complete with no active compaction falls through to the no-op arm. Verified against rustc/clippy 1.96.0 locally. --- src/room/state.rs | 27 +++++++++++++-------------- 1 file changed, 13 insertions(+), 14 deletions(-) diff --git a/src/room/state.rs b/src/room/state.rs index 3a7a69c..2d396cb 100644 --- a/src/room/state.rs +++ b/src/room/state.rs @@ -470,20 +470,19 @@ fn rebuild_compaction_from_frame( .unwrap_or_else(|| compaction_count.saturating_add(1)); *compaction_count = (*compaction_count).max(observed); } - "amux/turn_complete" => { - // Durable mirror of `clear_stale_compaction_at_turn_end`: a - // `started` with no matching `done` by the time a turn settles - // was abandoned. The live path clears the transient flag in - // memory and emits no frame, so without applying the same - // bound here, restart hydration would replay the persisted - // `started` and resurrect `active = true` forever. - // `amux/turn_complete` is broadcast (and therefore persisted) - // on every prompt-turn settlement, so it's a reliable durable - // bound. Count and history timestamps stay untouched. - if compaction_state.active { - compaction_state.active = false; - compaction_state.pending_hermes_session_id = None; - } + // Durable mirror of `clear_stale_compaction_at_turn_end`: a + // `started` with no matching `done` by the time a turn settles was + // abandoned. The live path clears the transient flag in memory and + // emits no frame, so without applying the same bound here, restart + // hydration would replay the persisted `started` and resurrect + // `active = true` forever. `amux/turn_complete` is broadcast (and + // therefore persisted) on every prompt-turn settlement, so it's a + // reliable durable bound. Count and history timestamps stay + // untouched. (Guarded arm rather than a nested `if` to satisfy + // clippy::collapsible_match.) + "amux/turn_complete" if compaction_state.active => { + compaction_state.active = false; + compaction_state.pending_hermes_session_id = None; } _ => {} }