diff --git a/integrations/opencode/marmot/Cargo.toml b/integrations/opencode/marmot/Cargo.toml index c28399a9..4028c961 100644 --- a/integrations/opencode/marmot/Cargo.toml +++ b/integrations/opencode/marmot/Cargo.toml @@ -22,3 +22,4 @@ tracing-subscriber.workspace = true [dev-dependencies] tempfile.workspace = true +tokio = { workspace = true, features = ["test-util"] } diff --git a/integrations/opencode/marmot/README.md b/integrations/opencode/marmot/README.md index 911281ef..c6f761f1 100644 --- a/integrations/opencode/marmot/README.md +++ b/integrations/opencode/marmot/README.md @@ -90,7 +90,8 @@ Configure with environment variables: | `WN_OPENCODE_ADMIN_HEX` | unset | Legacy alias for `WN_OPENCODE_ALLOWED_SENDERS_HEX` | | `WN_OPENCODE_ACCOUNT_ID_HEX` | first local account | Specific `wn-agent` account to use | | `WN_OPENCODE_BIN` | `opencode` | OpenCode binary or executable path | -| `WN_OPENCODE_TIMEOUT_SECS` | `300` | Hard timeout for each OpenCode invocation | +| `WN_OPENCODE_IDLE_TIMEOUT_SECS` | `120` | Kill the invocation when OpenCode produces no stdout line for this long | +| `WN_OPENCODE_TIMEOUT_SECS` | `3600` | Generous total safety cap for wedge recovery; ongoing output resets only the idle timer | | `WN_OPENCODE_REQUEST_TIMEOUT_SECS` | `30` | Timeout for each control-socket request | | `WN_OPENCODE_MAX_REPLY_BYTES` | `30000` | UTF-8 byte limit for each durable Marmot reply chunk | | `WN_OPENCODE_MAX_PENDING_PER_GROUP` | `4` | Per-group in-flight/queued prompt cap | diff --git a/integrations/opencode/marmot/src/bridge.rs b/integrations/opencode/marmot/src/bridge.rs index 1d543216..b0beee53 100644 --- a/integrations/opencode/marmot/src/bridge.rs +++ b/integrations/opencode/marmot/src/bridge.rs @@ -13,7 +13,7 @@ use crate::chunking::split_reply_chunks; use crate::config::{Config, dirs_home}; use crate::control::ControlClient; use crate::error::{HarnessError, Result}; -use crate::opencode::{Invocation, Outcome, RunnerEvent}; +use crate::opencode::{Invocation, Outcome, RunFailure, RunnerEvent}; use crate::repo_picker::{parse_repo_picker, resolve_repo, validate_session_cwd}; use crate::store::{SessionRecord, SessionStore}; @@ -343,6 +343,7 @@ async fn handle_message(ctx: Arc, inbound: InboundPrompt, permit: let invocation = Invocation { bin: ctx.cfg.opencode_bin.clone(), timeout: ctx.cfg.opencode_timeout, + idle_timeout: ctx.cfg.opencode_idle_timeout, cwd: cwd.clone(), session_id, prompt, @@ -402,28 +403,28 @@ async fn handle_message(ctx: Arc, inbound: InboundPrompt, permit: ) .await; } - Ok(Err(err)) => { + Ok(Err(failure)) => { warn!( target: TRACE_TARGET, method = "opencode_run", - error_kind = err.privacy_safe_kind(), + error_kind = failure.error.privacy_safe_kind(), "opencode invocation failed" ); - let text = match err { - HarnessError::OpencodeTimedOut => { - "[wn-opencode] opencode timed out before producing a complete response." - } - HarnessError::OpencodeSpawn => { - "[wn-opencode] failed to start opencode; check WN_OPENCODE_BIN." - } - _ => "[wn-opencode] opencode failed while streaming its response.", - }; + let text = handle_opencode_run_failure( + &ctx.sessions, + &inbound.group_ref, + known_session.as_ref(), + cwd, + &failure, + ctx.cfg.opencode_idle_timeout, + ) + .await; let _ = send_reply( &ctx, &inbound.account_ref, &inbound.group_ref, &inbound.message_ref, - text, + &text, 0, ) .await; @@ -475,10 +476,14 @@ async fn finish_success( if !delivery.failed && needs_persist && let Some(session_id) = outcome.observed_session - && let Err(err) = ctx - .sessions - .set(&inbound.group_ref, SessionRecord { session_id, cwd }) - .await + && let Err(err) = persist_observed_session_if_unset( + &ctx.sessions, + &inbound.group_ref, + known_session.as_ref(), + cwd, + Some(session_id), + ) + .await { warn!( target: TRACE_TARGET, @@ -527,6 +532,64 @@ struct DeliveryReport { failure_chunk_index: usize, } +async fn handle_opencode_run_failure( + sessions: &SessionStore, + group_ref: &str, + known_session: Option<&SessionRecord>, + cwd: PathBuf, + failure: &RunFailure, + idle_timeout: Duration, +) -> String { + if let Err(store_err) = persist_observed_session_if_unset( + sessions, + group_ref, + known_session, + cwd, + failure.observed_session.clone(), + ) + .await + { + warn!( + target: TRACE_TARGET, + method = "session_store", + error_kind = store_err.privacy_safe_kind(), + "failed to persist opencode session" + ); + } + + match &failure.error { + HarnessError::OpencodeIdle => format!( + "[wn-opencode] opencode went silent for {}s without producing output; killing the invocation.", + idle_timeout.as_secs() + ), + HarnessError::OpencodeTimedOut => { + "[wn-opencode] opencode timed out before producing a complete response.".to_owned() + } + HarnessError::OpencodeSpawn => { + "[wn-opencode] failed to start opencode; check WN_OPENCODE_BIN.".to_owned() + } + _ => "[wn-opencode] opencode failed while streaming its response.".to_owned(), + } +} + +async fn persist_observed_session_if_unset( + sessions: &SessionStore, + group_ref: &str, + known_session: Option<&SessionRecord>, + cwd: PathBuf, + observed_session: Option, +) -> Result<()> { + let needs_persist = known_session + .as_ref() + .is_none_or(|record| record.session_id.is_empty()); + if needs_persist && let Some(session_id) = observed_session { + sessions + .set(group_ref, SessionRecord { session_id, cwd }) + .await?; + } + Ok(()) +} + async fn resolve_cwd_and_prompt( ctx: &BridgeContext, inbound: &InboundPrompt, @@ -753,6 +816,7 @@ impl InboundDedupe { #[cfg(test)] mod tests { use super::*; + use crate::store::SessionStore; #[tokio::test] async fn dedupe_rejects_repeated_message_refs() { @@ -784,4 +848,67 @@ mod tests { Some("AA".repeat(32)) ); } + + #[tokio::test] + async fn persist_observed_session_only_when_unset() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("sessions.json"); + let home = dir.path().to_path_buf(); + let store = SessionStore::load(path.clone(), &home).unwrap(); + let cwd = home.join("proj"); + + persist_observed_session_if_unset( + &store, + "group1", + None, + cwd.clone(), + Some("ses_new".to_owned()), + ) + .await + .unwrap(); + let record = store.get("group1").await.unwrap(); + assert_eq!(record.session_id, "ses_new"); + assert_eq!(record.cwd, cwd); + + persist_observed_session_if_unset( + &store, + "group1", + Some(&record), + cwd.clone(), + Some("ses_other".to_owned()), + ) + .await + .unwrap(); + let record = store.get("group1").await.unwrap(); + assert_eq!(record.session_id, "ses_new"); + } + + #[tokio::test] + async fn run_failure_path_persists_first_session_and_selects_idle_reply() { + let dir = tempfile::tempdir().unwrap(); + let home = dir.path().to_path_buf(); + let store = SessionStore::load(home.join("sessions.json"), &home).unwrap(); + let failure = RunFailure { + error: HarnessError::OpencodeIdle, + observed_session: Some("ses_idle".to_owned()), + }; + + let reply = handle_opencode_run_failure( + &store, + "group1", + None, + home.clone(), + &failure, + Duration::from_secs(45), + ) + .await; + + let record = store.get("group1").await.unwrap(); + assert_eq!(record.session_id, "ses_idle"); + assert_eq!(record.cwd, home); + assert_eq!( + reply, + "[wn-opencode] opencode went silent for 45s without producing output; killing the invocation." + ); + } } diff --git a/integrations/opencode/marmot/src/config.rs b/integrations/opencode/marmot/src/config.rs index 221dd3f3..e35f3dc2 100644 --- a/integrations/opencode/marmot/src/config.rs +++ b/integrations/opencode/marmot/src/config.rs @@ -8,7 +8,8 @@ use crate::error::{HarnessError, Result}; pub(crate) const DEFAULT_MAX_REPLY_BYTES: usize = 30_000; pub(crate) const MARMOT_MESSAGE_BYTES_CEILING: usize = 60_000; -const DEFAULT_OPENCODE_TIMEOUT_SECS: u64 = 300; +const DEFAULT_OPENCODE_TIMEOUT_SECS: u64 = 3600; +const DEFAULT_OPENCODE_IDLE_TIMEOUT_SECS: u64 = 120; const DEFAULT_REQUEST_TIMEOUT_SECS: u64 = 30; const DEFAULT_MAX_PENDING_PER_GROUP: usize = 4; const MIN_REPLY_BYTES: usize = 4; @@ -21,6 +22,7 @@ pub(crate) struct Config { pub(crate) account_id_hex: Option, pub(crate) opencode_bin: String, pub(crate) opencode_timeout: Duration, + pub(crate) opencode_idle_timeout: Duration, pub(crate) request_timeout: Duration, pub(crate) max_reply_bytes: usize, pub(crate) max_pending_per_group: usize, @@ -91,6 +93,11 @@ impl Config { DEFAULT_OPENCODE_TIMEOUT_SECS, "WN_OPENCODE_TIMEOUT_SECS", )?); + let opencode_idle_timeout = Duration::from_secs(parse_u64( + lookup("WN_OPENCODE_IDLE_TIMEOUT_SECS"), + DEFAULT_OPENCODE_IDLE_TIMEOUT_SECS, + "WN_OPENCODE_IDLE_TIMEOUT_SECS", + )?); let request_timeout = Duration::from_secs(parse_u64( lookup("WN_OPENCODE_REQUEST_TIMEOUT_SECS"), DEFAULT_REQUEST_TIMEOUT_SECS, @@ -129,6 +136,7 @@ impl Config { account_id_hex, opencode_bin, opencode_timeout, + opencode_idle_timeout, request_timeout, max_reply_bytes, max_pending_per_group, @@ -222,6 +230,25 @@ mod tests { assert_eq!(cfg.max_reply_bytes, DEFAULT_MAX_REPLY_BYTES); } + #[test] + fn config_defaults_idle_and_total_timeouts() { + let cfg = Config::from_pairs(&[("WN_OPENCODE_ALLOWED_SENDERS_HEX", SENDER)]).unwrap(); + assert_eq!(cfg.opencode_timeout, Duration::from_secs(3600)); + assert_eq!(cfg.opencode_idle_timeout, Duration::from_secs(120)); + } + + #[test] + fn config_accepts_custom_idle_and_total_timeouts() { + let cfg = Config::from_pairs(&[ + ("WN_OPENCODE_ALLOWED_SENDERS_HEX", SENDER), + ("WN_OPENCODE_TIMEOUT_SECS", "600"), + ("WN_OPENCODE_IDLE_TIMEOUT_SECS", "45"), + ]) + .unwrap(); + assert_eq!(cfg.opencode_timeout, Duration::from_secs(600)); + assert_eq!(cfg.opencode_idle_timeout, Duration::from_secs(45)); + } + #[test] fn config_uses_terminal_harness_home_default() { let cfg = Config::from_pairs(&[("WN_OPENCODE_ALLOWED_SENDERS_HEX", SENDER)]).unwrap(); diff --git a/integrations/opencode/marmot/src/error.rs b/integrations/opencode/marmot/src/error.rs index ca408197..c8f8e3b1 100644 --- a/integrations/opencode/marmot/src/error.rs +++ b/integrations/opencode/marmot/src/error.rs @@ -30,6 +30,8 @@ pub(crate) enum HarnessError { ControlRejected { method: &'static str, code: String }, #[error("opencode invocation timed out")] OpencodeTimedOut, + #[error("opencode went idle without producing output")] + OpencodeIdle, #[error("opencode stream error")] OpencodeStream, #[error("opencode process failed to start")] @@ -51,6 +53,7 @@ impl HarnessError { Self::UnexpectedResponse { .. } => "unexpected_response", Self::ControlRejected { .. } => "control_rejected", Self::OpencodeTimedOut => "opencode_timeout", + Self::OpencodeIdle => "opencode_idle", Self::OpencodeStream => "opencode_stream", Self::OpencodeSpawn => "opencode_spawn", Self::Join => "join", diff --git a/integrations/opencode/marmot/src/opencode.rs b/integrations/opencode/marmot/src/opencode.rs index a80a17ae..5b36dcd9 100644 --- a/integrations/opencode/marmot/src/opencode.rs +++ b/integrations/opencode/marmot/src/opencode.rs @@ -4,9 +4,10 @@ use std::time::{Duration, Instant}; use serde::Deserialize; use tokio::io::{AsyncBufReadExt, BufReader}; -use tokio::process::Command; +use tokio::process::{Child, Command}; use tokio::sync::mpsc; -use tokio::time::timeout; +use tokio::task::JoinHandle; +use tokio::time::timeout_at; use tracing::debug; use crate::bridge::TRACE_TARGET; @@ -18,6 +19,7 @@ const STDERR_CAPTURE_BYTES: usize = 4096; pub(crate) struct Invocation { pub(crate) bin: String, pub(crate) timeout: Duration, + pub(crate) idle_timeout: Duration, pub(crate) cwd: PathBuf, pub(crate) session_id: Option, pub(crate) prompt: String, @@ -33,6 +35,12 @@ pub(crate) struct Outcome { } #[derive(Debug)] +pub(crate) struct RunFailure { + pub(crate) error: HarnessError, + pub(crate) observed_session: Option, +} + +#[derive(Debug, PartialEq, Eq)] pub(crate) enum RunnerEvent { Text(String), } @@ -74,7 +82,10 @@ struct OpencodeErrorData { status_code: Option, } -pub(crate) async fn run(invocation: Invocation, tx: mpsc::Sender) -> Result { +pub(crate) async fn run( + invocation: Invocation, + tx: mpsc::Sender, +) -> std::result::Result { let mut command = Command::new(&invocation.bin); command .args(build_run_args( @@ -87,28 +98,57 @@ pub(crate) async fn run(invocation: Invocation, tx: mpsc::Sender) - .stderr(Stdio::piped()) .kill_on_drop(true); - let mut child = command.spawn().map_err(|_| HarnessError::OpencodeSpawn)?; - let stdout = child.stdout.take().ok_or(HarnessError::OpencodeSpawn)?; - let stderr = child.stderr.take().ok_or(HarnessError::OpencodeSpawn)?; + let mut child = command.spawn().map_err(|_| RunFailure { + error: HarnessError::OpencodeSpawn, + observed_session: None, + })?; + let total_deadline = tokio::time::Instant::now() + invocation.timeout; + let stdout = match child.stdout.take() { + Some(stdout) => stdout, + None => { + kill_and_reap(&mut child).await; + return Err(RunFailure { + error: HarnessError::OpencodeSpawn, + observed_session: None, + }); + } + }; + let stderr = match child.stderr.take() { + Some(stderr) => stderr, + None => { + kill_and_reap(&mut child).await; + return Err(RunFailure { + error: HarnessError::OpencodeSpawn, + observed_session: None, + }); + } + }; let mut lines = BufReader::new(stdout).lines(); - let stderr_task = tokio::spawn(capture_stderr(stderr)); + let mut stderr_task = tokio::spawn(capture_stderr(stderr)); let start = Instant::now(); let mut observed_session: Option = None; let mut error_summary: Option = None; + let mut idle_deadline = tokio::time::Instant::now() + invocation.idle_timeout; + + let lifecycle_result = timeout_at(total_deadline, async { + loop { + let Some(line) = next_stdout_line(&mut lines, idle_deadline).await? else { + break; + }; - let stream_result = timeout(invocation.timeout, async { - while let Some(line) = lines - .next_line() - .await - .map_err(|_| HarnessError::OpencodeStream)? - { if line.is_empty() { + idle_deadline = tokio::time::Instant::now() + invocation.idle_timeout; continue; } match parse_event_line(&line) { Ok(Some(ParsedEvent::Text(text))) => { - if !text.trim().is_empty() && tx.send(RunnerEvent::Text(text)).await.is_err() { - break; + if !text.trim().is_empty() { + // The idle clock measures actual stdout reads, not time intentionally + // spent applying bounded reply-channel backpressure. The total deadline + // still caps both operations. + tx.send(RunnerEvent::Text(text)) + .await + .map_err(|_| HarnessError::OpencodeStream)?; } } Ok(Some(ParsedEvent::Session(session_id))) => { @@ -120,8 +160,10 @@ pub(crate) async fn run(invocation: Invocation, tx: mpsc::Sender) - session_id, summary, })) => { - if observed_session.is_none() { - observed_session = session_id; + if let Some(session_id) = session_id + && observed_session.is_none() + { + observed_session = Some(session_id); } if error_summary.is_none() { error_summary = Some(summary); @@ -137,32 +179,70 @@ pub(crate) async fn run(invocation: Invocation, tx: mpsc::Sender) - ); } } + idle_deadline = tokio::time::Instant::now() + invocation.idle_timeout; } - Ok::<(), HarnessError>(()) + + let (status, stderr) = match timeout_at(idle_deadline, async { + let status = child.wait().await.map_err(HarnessError::from)?; + let stderr = (&mut stderr_task).await.map_err(HarnessError::from)?; + Ok::<_, HarnessError>((status, stderr)) + }) + .await + { + Err(_) => return Err(HarnessError::OpencodeIdle), + Ok(result) => result?, + }; + Ok::(Outcome { + observed_session: observed_session.clone(), + exit_code: status.code(), + error_summary, + stderr: strip_ansi(stderr.trim()), + elapsed_ms: start.elapsed().as_millis(), + }) }) .await; - match stream_result { - Ok(Ok(())) => {} - Ok(Err(err)) => { - let _ = child.kill().await; - return Err(err); + match lifecycle_result { + Ok(Ok(outcome)) => Ok(outcome), + Ok(Err(error)) => { + cleanup_failed_run(&mut child, &mut stderr_task).await; + Err(RunFailure { + error, + observed_session, + }) } Err(_) => { - let _ = child.kill().await; - return Err(HarnessError::OpencodeTimedOut); + cleanup_failed_run(&mut child, &mut stderr_task).await; + Err(RunFailure { + error: HarnessError::OpencodeTimedOut, + observed_session, + }) } } +} - let status = child.wait().await?; - let stderr = stderr_task.await?; - Ok(Outcome { - observed_session, - exit_code: status.code(), - error_summary, - stderr: strip_ansi(stderr.trim()), - elapsed_ms: start.elapsed().as_millis(), - }) +async fn next_stdout_line( + lines: &mut tokio::io::Lines, + idle_deadline: tokio::time::Instant, +) -> std::result::Result, HarnessError> { + match timeout_at(idle_deadline, lines.next_line()).await { + Err(_) => Err(HarnessError::OpencodeIdle), + Ok(Err(_)) => Err(HarnessError::OpencodeStream), + Ok(Ok(line)) => Ok(line), + } +} + +async fn cleanup_failed_run(child: &mut Child, stderr_task: &mut JoinHandle) { + stderr_task.abort(); + kill_and_reap(child).await; + if !stderr_task.is_finished() { + let _ = stderr_task.await; + } +} + +async fn kill_and_reap(child: &mut Child) { + let _ = child.start_kill(); + let _ = child.wait().await; } pub(crate) fn build_run_args(session_id: Option<&str>, prompt: &str) -> Vec { @@ -266,13 +346,25 @@ pub(crate) fn strip_ansi(input: &str) -> String { #[cfg(test)] mod tests { - use std::fs; - use std::os::unix::fs::PermissionsExt; - use tokio::sync::mpsc; use super::*; + fn mock_invocation(dir: &tempfile::TempDir, scenario: &str) -> Invocation { + Invocation { + bin: concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/fixtures/mock-opencode.sh" + ) + .to_owned(), + timeout: Duration::from_secs(10), + idle_timeout: Duration::from_millis(500), + cwd: dir.path().to_path_buf(), + session_id: None, + prompt: scenario.to_owned(), + } + } + #[test] fn build_run_args_separates_prompt_from_flags() { assert_eq!( @@ -334,32 +426,8 @@ mod tests { #[tokio::test] async fn run_streams_text_from_mock_binary() { let dir = tempfile::tempdir().unwrap(); - let script = dir.path().join("mock-opencode"); - fs::write( - &script, - r#"#!/usr/bin/env bash -printf '%s\n' '{"type":"step_start","sessionID":"ses_mock"}' -printf '%s\n' '{"type":"text","part":{"text":"hello"}}' -"#, - ) - .unwrap(); - let mut permissions = fs::metadata(&script).unwrap().permissions(); - permissions.set_mode(0o755); - fs::set_permissions(&script, permissions).unwrap(); - let (tx, mut rx) = mpsc::channel(4); - let outcome = run( - Invocation { - bin: script.display().to_string(), - timeout: Duration::from_secs(5), - cwd: dir.path().to_path_buf(), - session_id: None, - prompt: "prompt".to_owned(), - }, - tx, - ) - .await - .unwrap(); + let outcome = run(mock_invocation(&dir, "stream-text"), tx).await.unwrap(); assert_eq!(outcome.observed_session, Some("ses_mock".to_owned())); assert_eq!(outcome.exit_code, Some(0)); assert_eq!(outcome.error_summary, None); @@ -369,4 +437,171 @@ printf '%s\n' '{"type":"text","part":{"text":"hello"}}' )); assert!(rx.recv().await.is_none()); } + + #[cfg(unix)] + #[tokio::test] + async fn run_idle_timeout_fires_after_silence() { + let dir = tempfile::tempdir().unwrap(); + let (tx, _rx) = mpsc::channel(4); + let failure = run( + Invocation { + idle_timeout: Duration::from_millis(200), + ..mock_invocation(&dir, "idle") + }, + tx, + ) + .await + .unwrap_err(); + assert!(matches!(failure.error, HarnessError::OpencodeIdle)); + assert_eq!(failure.observed_session.as_deref(), Some("ses_idle")); + } + + #[tokio::test(start_paused = true)] + async fn stdout_lines_reset_idle_past_old_wall_clock_deadline() { + use tokio::io::AsyncWriteExt; + use tokio::time::sleep; + + const IDLE: Duration = Duration::from_secs(120); + const GAP: Duration = Duration::from_secs(70); + const TOTAL: Duration = Duration::from_secs(3600); + const LINE_COUNT: usize = 6; + + let (mut writer, reader) = tokio::io::duplex(1024); + let producer = tokio::spawn(async move { + for index in 0..LINE_COUNT { + if index != 0 { + sleep(GAP).await; + } + writer.write_all(b"line\n").await.unwrap(); + } + }); + let mut lines = BufReader::new(reader).lines(); + let started = tokio::time::Instant::now(); + let received = tokio::time::timeout(TOTAL, async { + let mut count = 0; + let mut idle_deadline = tokio::time::Instant::now() + IDLE; + while next_stdout_line(&mut lines, idle_deadline) + .await + .unwrap() + .is_some() + { + count += 1; + idle_deadline = tokio::time::Instant::now() + IDLE; + } + count + }) + .await + .unwrap(); + + producer.await.unwrap(); + assert_eq!(received, LINE_COUNT); + assert!(started.elapsed() > Duration::from_secs(300)); + assert!(started.elapsed() < TOTAL); + } + + #[cfg(unix)] + #[tokio::test] + async fn run_total_cap_fires_despite_ongoing_lines() { + let dir = tempfile::tempdir().unwrap(); + let (tx, _rx) = mpsc::channel(4); + let failure = run( + Invocation { + timeout: Duration::from_millis(1_500), + idle_timeout: Duration::from_secs(1), + ..mock_invocation(&dir, "total-cap") + }, + tx, + ) + .await + .unwrap_err(); + assert!( + matches!(failure.error, HarnessError::OpencodeTimedOut), + "expected total timeout, got {failure:?}" + ); + } + + #[cfg(unix)] + #[tokio::test] + async fn run_idle_timeout_fires_after_stdout_eof_with_live_child() { + let dir = tempfile::tempdir().unwrap(); + let (tx, _rx) = mpsc::channel(4); + let started = Instant::now(); + let failure = run( + Invocation { + timeout: Duration::from_secs(10), + idle_timeout: Duration::from_millis(200), + ..mock_invocation(&dir, "stdout-close-live") + }, + tx, + ) + .await + .unwrap_err(); + assert!(matches!(failure.error, HarnessError::OpencodeIdle)); + assert!(started.elapsed() < Duration::from_secs(1)); + } + + #[cfg(unix)] + #[tokio::test] + async fn run_eof_keeps_remaining_idle_budget() { + let dir = tempfile::tempdir().unwrap(); + let (tx, _rx) = mpsc::channel(4); + let started = Instant::now(); + let failure = run( + Invocation { + timeout: Duration::from_secs(10), + idle_timeout: Duration::from_secs(1), + ..mock_invocation(&dir, "stdout-close-near-idle") + }, + tx, + ) + .await + .unwrap_err(); + assert!(matches!(failure.error, HarnessError::OpencodeIdle)); + assert!( + started.elapsed() < Duration::from_millis(1_350), + "stdout EOF must not reset the existing idle deadline" + ); + } + + #[cfg(unix)] + #[tokio::test] + async fn run_total_cap_includes_child_wait_after_stdout_closes() { + let dir = tempfile::tempdir().unwrap(); + let (tx, _rx) = mpsc::channel(4); + let started = Instant::now(); + let failure = run( + Invocation { + timeout: Duration::from_millis(200), + idle_timeout: Duration::from_secs(5), + ..mock_invocation(&dir, "stdout-close-live") + }, + tx, + ) + .await + .unwrap_err(); + assert!(matches!(failure.error, HarnessError::OpencodeTimedOut)); + assert!(started.elapsed() < Duration::from_secs(1)); + } + + #[cfg(unix)] + #[tokio::test] + async fn run_failure_keeps_session_despite_channel_backpressure() { + let dir = tempfile::tempdir().unwrap(); + let (tx, _rx) = mpsc::channel(1); + let failure = run( + Invocation { + timeout: Duration::from_millis(200), + idle_timeout: Duration::from_secs(5), + ..mock_invocation(&dir, "session-backpressure") + }, + tx, + ) + .await + .unwrap_err(); + assert!(matches!(failure.error, HarnessError::OpencodeTimedOut)); + assert_eq!( + failure.observed_session.as_deref(), + Some("ses_backpressure") + ); + } } diff --git a/integrations/opencode/marmot/tests/fixtures/mock-opencode.sh b/integrations/opencode/marmot/tests/fixtures/mock-opencode.sh new file mode 100755 index 00000000..cd50f49a --- /dev/null +++ b/integrations/opencode/marmot/tests/fixtures/mock-opencode.sh @@ -0,0 +1,46 @@ +#!/usr/bin/env bash +set -euo pipefail + +scenario="${!#}" +case "$scenario" in + stream-text) + printf '%s\n' '{"type":"step_start","sessionID":"ses_mock"}' + printf '%s\n' '{"type":"text","part":{"text":"hello"}}' + ;; + idle) + printf '%s\n' '{"type":"step_start","sessionID":"ses_idle"}' + exec sleep 2 + ;; + streaming) + for i in 1 2 3 4 5; do + printf '%s\n' "{\"type\":\"text\",\"part\":{\"text\":\"chunk$i\"}}" + sleep 0.15 + done + ;; + total-cap) + printf '%s\n' '{"type":"step_start","sessionID":"ses_total"}' + for _ in $(seq 1 60); do + printf '%s\n' '{"type":"step_finish"}' + sleep 0.05 + done + ;; + stdout-close-live) + exec 1>&- + exec sleep 2 + ;; + stdout-close-near-idle) + sleep 0.6 + exec 1>&- + exec sleep 3 + ;; + session-backpressure) + printf '%s\n' '{"type":"step_start","sessionID":"ses_backpressure"}' + printf '%s\n' '{"type":"text","part":{"text":"first"}}' + printf '%s\n' '{"type":"text","part":{"text":"second"}}' + exec sleep 2 + ;; + *) + printf 'unknown mock scenario: %s\n' "$scenario" >&2 + exit 2 + ;; +esac diff --git a/integrations/opencode/marmot/tests/test_installer.sh b/integrations/opencode/marmot/tests/test_installer.sh index 6535692f..2b769b61 100755 --- a/integrations/opencode/marmot/tests/test_installer.sh +++ b/integrations/opencode/marmot/tests/test_installer.sh @@ -118,6 +118,14 @@ case "$installer_service_dry_run" in *"would require OpenCode binary: /bin/echo"* ) ;; *) echo "opencode installer service dry-run did not validate OpenCode binary" >&2; exit 1;; esac +case "$installer_service_dry_run" in + *"WN_OPENCODE_TIMEOUT_SECS"*3600* ) ;; + *) echo "opencode installer service dry-run did not set 3600s total timeout" >&2; exit 1;; +esac +case "$installer_service_dry_run" in + *"WN_OPENCODE_IDLE_TIMEOUT_SECS"*120* ) ;; + *) echo "opencode installer service dry-run did not set 120s idle timeout" >&2; exit 1;; +esac case "$(uname -s)" in Darwin) case "$installer_service_dry_run" in diff --git a/scripts/install-opencode-marmot.sh b/scripts/install-opencode-marmot.sh index 4b261ba5..1db69855 100755 --- a/scripts/install-opencode-marmot.sh +++ b/scripts/install-opencode-marmot.sh @@ -32,7 +32,8 @@ MARMOT_AGENT_SERVICE_NAME="${MARMOT_AGENT_SERVICE_NAME:-wn-agent-harnesses}" MARMOT_AGENT_LAUNCHD_LABEL="${MARMOT_AGENT_LAUNCHD_LABEL:-org.marmot.wn-agent.harnesses}" MARMOT_RELAYS="${MARMOT_RELAYS:-wss://relay.eu.whitenoise.chat,wss://relay.us.whitenoise.chat}" WN_OPENCODE_BIN="${WN_OPENCODE_BIN:-opencode}" -WN_OPENCODE_TIMEOUT_SECS="${WN_OPENCODE_TIMEOUT_SECS:-300}" +WN_OPENCODE_TIMEOUT_SECS="${WN_OPENCODE_TIMEOUT_SECS:-3600}" +WN_OPENCODE_IDLE_TIMEOUT_SECS="${WN_OPENCODE_IDLE_TIMEOUT_SECS:-120}" WN_OPENCODE_REQUEST_TIMEOUT_SECS="${WN_OPENCODE_REQUEST_TIMEOUT_SECS:-30}" WN_OPENCODE_MAX_REPLY_BYTES="${WN_OPENCODE_MAX_REPLY_BYTES:-30000}" WN_OPENCODE_MAX_PENDING_PER_GROUP="${WN_OPENCODE_MAX_PENDING_PER_GROUP:-4}" @@ -420,6 +421,7 @@ install_macos_opencode_service() { if [ "$DRY_RUN" -eq 1 ]; then log "would install LaunchAgent $plist" log "would run: launchctl bootstrap gui/$UID $plist" + log "env: WN_OPENCODE_TIMEOUT_SECS=$WN_OPENCODE_TIMEOUT_SECS WN_OPENCODE_IDLE_TIMEOUT_SECS=$WN_OPENCODE_IDLE_TIMEOUT_SECS" return 0 fi @@ -443,6 +445,7 @@ install_macos_opencode_service() { plist_env_entry "WN_OPENCODE_ALLOWED_SENDERS_HEX" "$BOOTSTRAP_ALLOWED_SENDERS_HEX" plist_env_entry "WN_OPENCODE_BIN" "$WN_OPENCODE_BIN" plist_env_entry "WN_OPENCODE_TIMEOUT_SECS" "$WN_OPENCODE_TIMEOUT_SECS" + plist_env_entry "WN_OPENCODE_IDLE_TIMEOUT_SECS" "$WN_OPENCODE_IDLE_TIMEOUT_SECS" plist_env_entry "WN_OPENCODE_REQUEST_TIMEOUT_SECS" "$WN_OPENCODE_REQUEST_TIMEOUT_SECS" plist_env_entry "WN_OPENCODE_MAX_REPLY_BYTES" "$WN_OPENCODE_MAX_REPLY_BYTES" plist_env_entry "WN_OPENCODE_MAX_PENDING_PER_GROUP" "$WN_OPENCODE_MAX_PENDING_PER_GROUP" @@ -523,6 +526,7 @@ install_linux_opencode_service() { if [ "$DRY_RUN" -eq 1 ]; then log "would install systemd user unit $service" log "would run: systemctl --user enable --now wn-opencode.service" + log "env: WN_OPENCODE_TIMEOUT_SECS=$WN_OPENCODE_TIMEOUT_SECS WN_OPENCODE_IDLE_TIMEOUT_SECS=$WN_OPENCODE_IDLE_TIMEOUT_SECS" return 0 fi @@ -544,6 +548,7 @@ install_linux_opencode_service() { printf 'Environment=%s\n' "$(systemd_quote "WN_OPENCODE_ALLOWED_SENDERS_HEX=$BOOTSTRAP_ALLOWED_SENDERS_HEX")" printf 'Environment=%s\n' "$(systemd_quote "WN_OPENCODE_BIN=$WN_OPENCODE_BIN")" printf 'Environment=%s\n' "$(systemd_quote "WN_OPENCODE_TIMEOUT_SECS=$WN_OPENCODE_TIMEOUT_SECS")" + printf 'Environment=%s\n' "$(systemd_quote "WN_OPENCODE_IDLE_TIMEOUT_SECS=$WN_OPENCODE_IDLE_TIMEOUT_SECS")" printf 'Environment=%s\n' "$(systemd_quote "WN_OPENCODE_REQUEST_TIMEOUT_SECS=$WN_OPENCODE_REQUEST_TIMEOUT_SECS")" printf 'Environment=%s\n' "$(systemd_quote "WN_OPENCODE_MAX_REPLY_BYTES=$WN_OPENCODE_MAX_REPLY_BYTES")" printf 'Environment=%s\n' "$(systemd_quote "WN_OPENCODE_MAX_PENDING_PER_GROUP=$WN_OPENCODE_MAX_PENDING_PER_GROUP")" @@ -681,6 +686,7 @@ write_opencode_env() { local env_file="$MARMOT_HOME/dev/wn-opencode.env" if [ "$DRY_RUN" -eq 1 ]; then log "would write private env file $env_file" + log "env: WN_OPENCODE_TIMEOUT_SECS=$WN_OPENCODE_TIMEOUT_SECS WN_OPENCODE_IDLE_TIMEOUT_SECS=$WN_OPENCODE_IDLE_TIMEOUT_SECS" return 0 fi run mkdir -p "$MARMOT_HOME/dev" @@ -693,6 +699,7 @@ write_opencode_env() { printf 'WN_OPENCODE_ALLOWED_SENDERS_HEX=%q\n' "$BOOTSTRAP_ALLOWED_SENDERS_HEX" printf 'WN_OPENCODE_BIN=%q\n' "$WN_OPENCODE_BIN" printf 'WN_OPENCODE_TIMEOUT_SECS=%q\n' "$WN_OPENCODE_TIMEOUT_SECS" + printf 'WN_OPENCODE_IDLE_TIMEOUT_SECS=%q\n' "$WN_OPENCODE_IDLE_TIMEOUT_SECS" printf 'WN_OPENCODE_REQUEST_TIMEOUT_SECS=%q\n' "$WN_OPENCODE_REQUEST_TIMEOUT_SECS" printf 'WN_OPENCODE_MAX_REPLY_BYTES=%q\n' "$WN_OPENCODE_MAX_REPLY_BYTES" printf 'WN_OPENCODE_MAX_PENDING_PER_GROUP=%q\n' "$WN_OPENCODE_MAX_PENDING_PER_GROUP"