Skip to content

fix(router): route synthetic wake busy through the session manager #396

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yicheng47 merged 1 commit into from
Aug 10, 2026
Merged

fix(router): route synthetic wake busy through the session manager #396

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions crates/runner-core/src/event_log/log.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,9 +141,9 @@ impl EventLog {
/// processing flow through the same channel, so a stuck flock
/// would freeze them all.
///
/// On contention the caller is expected to drop the event (status
/// transitions are observability, not load-bearing) and bump a
/// streak counter for visibility.
/// On contention the caller decides whether to drop or retry. The
/// session forwarder uses a short bounded retry because its status
/// transitions feed the router's reconciliation gate.
pub fn try_append(&self, draft: EventDraft) -> std::result::Result<Event, TryAppendError> {
let file = OpenOptions::new()
.create(true)
Expand Down Expand Up @@ -493,8 +493,8 @@ fn parse_id(line: &[u8]) -> Result<String> {
#[derive(Debug)]
pub enum TryAppendError {
/// Another writer holds the file lock right now. Caller should
/// drop the event and try again on the next transition rather
/// than blocking the producer thread.
/// either drop the event or retry with its own bounded policy
/// rather than blocking the producer thread.
Contended,
/// I/O error opening the file or any other append failure.
Failed(Error),
Expand Down
32 changes: 26 additions & 6 deletions src-tauri/src/router/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ use std::thread::JoinHandle;
use std::time::{Duration, Instant};

use runner_core::event_log::EventLog;
use runner_core::model::{Event, EventKind, SignalType};
use runner_core::model::{Event, EventDraft, EventKind, SignalType};
use serde::Serialize;
use tauri::Emitter;

Expand Down Expand Up @@ -69,6 +69,8 @@ pub trait StdinInjector: Send + Sync + 'static {

fn finish_delivery(&self, session_id: &str, token: u64);

fn synthesize_wake_busy(&self, session_id: &str, draft: EventDraft) -> Result<()>;

fn register_delivery_listener(
&self,
session_id: &str,
Expand Down Expand Up @@ -127,6 +129,10 @@ impl StdinInjector for SessionManager {
SessionManager::finish_delivery(self, session_id, token)
}

fn synthesize_wake_busy(&self, session_id: &str, draft: EventDraft) -> Result<()> {
SessionManager::synthesize_wake_busy(self, session_id, draft)
}

fn register_delivery_listener(
&self,
session_id: &str,
Expand Down Expand Up @@ -587,26 +593,40 @@ impl Router {
/// without this, a slow-to-respond agent could appear `idle` to the
/// user immediately after a nudge. Latest-wins absorbs the
/// follow-up forwarder event without churn.
///
/// Post-issue-#385: the append routes through the SessionManager
/// (not `log.append`) so the forwarder dedup key
/// (`session.activity`) updates atomically with the event — a
/// direct append left the key stale and the paired end-of-turn
/// idle was deduped away, sticking the rail on busy.
fn synthesize_wake_busy(&self, handle: &str) {
if handle == "human" {
return;
}
{
let session_id = {
let state = self.state.lock().unwrap();
if matches!(state.status.get(handle), Some(RunnerStatus::Busy)) {
return;
}
}
let draft = runner_core::model::EventDraft::signal(
state.session_by_handle.get(handle).cloned()
};
let Some(session_id) = session_id else {
log::error!(
"cannot synthesize runner_status busy for @{handle} on mission {}: no session",
self.mission_id,
);
return;
};
let draft = EventDraft::signal(
self.crew_id.clone(),
self.mission_id.clone(),
handle,
SignalType::new("runner_status"),
serde_json::json!({ "state": "busy" }),
);
if let Err(e) = self.log.append(draft) {
if let Err(e) = self.injector.synthesize_wake_busy(&session_id, draft) {
log::error!(
"failed to append synthetic runner_status busy for @{handle} on mission {}: {e}",
"failed to synthesize runner_status busy for @{handle} on mission {}: {e}",
self.mission_id,
);
return;
Expand Down
53 changes: 42 additions & 11 deletions src-tauri/src/router/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,9 @@ enum InjectKind {

/// Records every `inject` / `inject_paste_with_verify` call so
/// handler outputs can be asserted.
#[derive(Default)]
struct RecordingInjector {
status_log: Arc<EventLog>,
activity: Mutex<HashMap<String, super::RunnerStatus>>,
pushes: Mutex<Vec<(String, InjectKind, Vec<u8>)>>,
blocked_events: Mutex<Vec<DeliveryBlockedEvent>>,
/// Optional `dead_session` set — `inject` errors when called with one
Expand All @@ -54,6 +55,22 @@ struct RecordingInputState {
}

impl RecordingInjector {
fn new(status_log: Arc<EventLog>) -> Self {
Self {
status_log,
activity: Mutex::new(HashMap::new()),
pushes: Mutex::new(Vec::new()),
blocked_events: Mutex::new(Vec::new()),
dead: Mutex::new(Vec::new()),
input: Mutex::new(HashMap::new()),
listeners: Mutex::new(HashMap::new()),
}
}

fn activity_for(&self, session_id: &str) -> Option<super::RunnerStatus> {
self.activity.lock().unwrap().get(session_id).copied()
}

fn pushes_for(&self, session_id: &str) -> Vec<String> {
self.pushes
.lock()
Expand Down Expand Up @@ -333,6 +350,15 @@ impl StdinInjector for RecordingInjector {
}
}

fn synthesize_wake_busy(&self, session_id: &str, draft: EventDraft) -> Result<()> {
self.status_log.append(draft)?;
self.activity
.lock()
.unwrap()
.insert(session_id.to_string(), super::RunnerStatus::Busy);
Ok(())
}

fn register_delivery_listener(
&self,
session_id: &str,
Expand Down Expand Up @@ -403,7 +429,7 @@ fn fixture(
) {
let dir = tempfile::tempdir().unwrap();
let log = Arc::new(EventLog::open(dir.path()).unwrap());
let injector = Arc::new(RecordingInjector::default());
let injector = Arc::new(RecordingInjector::new(Arc::clone(&log)));
let injector_dyn: Arc<dyn StdinInjector> = injector.clone();
let notifier: Arc<dyn RouterUiNotifier> = injector.clone();
let router = Router::new(
Expand Down Expand Up @@ -1745,7 +1771,7 @@ fn pending_ask_map_reconstructs_from_log_on_reopen() {

// First mount handles the ask live (appends human_question).
{
let injector = Arc::new(RecordingInjector::default());
let injector = Arc::new(RecordingInjector::new(Arc::clone(&log)));
let injector_dyn: Arc<dyn StdinInjector> = injector.clone();
let router = Router::new(
"mission-1".into(),
Expand Down Expand Up @@ -1782,7 +1808,7 @@ fn pending_ask_map_reconstructs_from_log_on_reopen() {

// Reopen: build router #2, fold projection state from history. This
// is the path mission_resume / mount-on-app-restart will follow.
let injector = Arc::new(RecordingInjector::default());
let injector = Arc::new(RecordingInjector::new(Arc::clone(&log)));
let injector_dyn: Arc<dyn StdinInjector> = injector.clone();
let router2 = Router::new(
"mission-1".into(),
Expand Down Expand Up @@ -1889,7 +1915,7 @@ fn reconstruct_recovers_latest_runner_status_only() {
))
.unwrap();

let injector = Arc::new(RecordingInjector::default());
let injector = Arc::new(RecordingInjector::new(Arc::clone(&log)));
let injector_dyn: Arc<dyn StdinInjector> = injector.clone();
let router = Router::new(
"mission-1".into(),
Expand Down Expand Up @@ -1979,7 +2005,7 @@ fn fresh_mission_start_does_not_call_reconstruct() {
))
.unwrap();

let injector = Arc::new(RecordingInjector::default());
let injector = Arc::new(RecordingInjector::new(Arc::clone(&log)));
let injector_dyn: Arc<dyn StdinInjector> = injector.clone();
let router = Router::new(
"mission-1".into(),
Expand Down Expand Up @@ -2077,7 +2103,7 @@ fn reconstruct_tolerates_malformed_lines_like_the_bus() {
let roster = vec![slot_with_runner("lead", true)];
// First mount handles the ask live — appends human_question.
{
let injector = Arc::new(RecordingInjector::default());
let injector = Arc::new(RecordingInjector::new(Arc::clone(&log)));
let injector_dyn: Arc<dyn StdinInjector> = injector.clone();
let router = Router::new(
"mission-1".into(),
Expand Down Expand Up @@ -2106,7 +2132,7 @@ fn reconstruct_tolerates_malformed_lines_like_the_bus() {
.id;

// Reopen + reconstruct: must not fail despite the malformed middle line.
let injector = Arc::new(RecordingInjector::default());
let injector = Arc::new(RecordingInjector::new(Arc::clone(&log)));
let injector_dyn: Arc<dyn StdinInjector> = injector.clone();
let router2 = Router::new(
"mission-1".into(),
Expand Down Expand Up @@ -2151,7 +2177,7 @@ fn directed_wake_synthesizes_busy_and_idle_clears_it() {
// `idle` was emitted. The router now synthesizes `runner_status busy`
// (with `from = recipient`) for any wake nudge, and the existing
// worker-emitted `idle` clears it.
let (router, _injector, log, _dir) = fixture(
let (router, injector, log, _dir) = fixture(
vec![
slot_with_runner("lead", true),
slot_with_runner("impl", false),
Expand Down Expand Up @@ -2186,6 +2212,11 @@ fn directed_wake_synthesizes_busy_and_idle_clears_it() {
router.state.lock().unwrap().status.get("impl"),
Some(super::RunnerStatus::Busy),
));
assert_eq!(
injector.activity_for("S-IMPL"),
Some(super::RunnerStatus::Busy),
"synthetic busy must update the session-side activity store",
);

// A second directed wake while still busy must not churn another
// busy event into the log — the dedupe guard suppresses it.
Expand Down Expand Up @@ -2245,7 +2276,7 @@ fn synthetic_busy_replays_through_existing_runner_status_projection() {

// First mount: drive a directed message to synthesize busy.
{
let injector = Arc::new(RecordingInjector::default());
let injector = Arc::new(RecordingInjector::new(Arc::clone(&log)));
let injector_dyn: Arc<dyn StdinInjector> = injector.clone();
let router = Router::new(
"mission-1".into(),
Expand All @@ -2268,7 +2299,7 @@ fn synthetic_busy_replays_through_existing_runner_status_projection() {
}

// Reopen + reconstruct.
let injector = Arc::new(RecordingInjector::default());
let injector = Arc::new(RecordingInjector::new(Arc::clone(&log)));
let injector_dyn: Arc<dyn StdinInjector> = injector.clone();
let router2 = Router::new(
"mission-1".into(),
Expand Down
51 changes: 39 additions & 12 deletions src-tauri/src/session/manager/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,8 @@ mod tests;

const MAX_OUTPUT_BUFFER_CHUNKS: usize = 4096;
const RECENT_LOCAL_INPUT_WINDOW: Duration = Duration::from_secs(2);
const RUNNER_STATUS_APPEND_MAX_ATTEMPTS: usize = 8;
const RUNNER_STATUS_APPEND_RETRY_DELAY: Duration = Duration::from_millis(5);
pub(crate) const DEFAULT_PTY_SIZE: (u16, u16) = (80, 24);

/// How long a cols-change resize storm must stay quiet before the
Expand Down Expand Up @@ -257,24 +259,32 @@ impl ForwarderEmitCtx {
)
}

/// Non-blocking append of a forwarder-emitted `runner_status`
/// row. The consumer thread runs this on every status
/// transition; it must not block (it shares the mpsc receiver
/// with the terminal output stream and the exit-event reap, so
/// a stuck flock would freeze them too). Wire shape mirrors
/// `cli/src/signal.rs::run_status` so router / UI projections
/// can't tell the two apart except by `payload.source`.
/// Bounded append of a forwarder-emitted `runner_status` row.
/// Each flock attempt is non-blocking; brief contention is retried
/// because the router's reconciliation gate now depends on these
/// events, while persistent contention still gives up quickly so
/// terminal output and exit-event reap cannot freeze behind it.
fn try_append_runner_status(&self, state: RunnerStatus, source: &'static str) -> AppendOutcome {
match self
.event_log
.try_append(self.runner_status_draft(state, source))
{
Ok(_) => AppendOutcome::Ok,
match self.try_append_with_retry(self.runner_status_draft(state, source)) {
Ok(()) => AppendOutcome::Ok,
Err(TryAppendError::Contended) => AppendOutcome::Contended,
Err(TryAppendError::Failed(_)) => AppendOutcome::Failed,
}
}

fn try_append_with_retry(&self, draft: EventDraft) -> std::result::Result<(), TryAppendError> {
for attempt in 1..=RUNNER_STATUS_APPEND_MAX_ATTEMPTS {
match self.event_log.try_append(draft.clone()) {
Ok(_) => return Ok(()),
Err(TryAppendError::Contended) if attempt < RUNNER_STATUS_APPEND_MAX_ATTEMPTS => {
thread::sleep(RUNNER_STATUS_APPEND_RETRY_DELAY);
}
Err(error) => return Err(error),
}
}
unreachable!("bounded append loop always returns")
}

fn append_runner_status(
&self,
state: RunnerStatus,
Expand Down Expand Up @@ -988,6 +998,23 @@ impl SessionManager {
true
}

pub(crate) fn synthesize_wake_busy(&self, session_id: &str, draft: EventDraft) -> Result<()> {
let session = self
.session_state(session_id)
.ok_or_else(|| Error::msg(format!("session not found: {session_id}")))?;
let mut session = session.lock().unwrap();
let sink = session.mission_status_sink.as_ref().ok_or_else(|| {
Error::msg(format!("session has no mission status sink: {session_id}"))
})?;
match sink.try_append_with_retry(draft) {
Ok(()) => {}
Err(TryAppendError::Contended) => return Err(Error::msg("event log busy")),
Err(TryAppendError::Failed(error)) => return Err(error.into()),
}
session.activity = Some(SessionActivityState::Busy);
Ok(())
}

pub(crate) fn publish_direct_activity(
&self,
session_id: &str,
Expand Down
Loading
Loading