diff --git a/src/sdk/src/harness_hooks/tests.rs b/src/sdk/src/harness_hooks/tests.rs index 5064d0994..d996f35c5 100644 --- a/src/sdk/src/harness_hooks/tests.rs +++ b/src/sdk/src/harness_hooks/tests.rs @@ -397,7 +397,7 @@ fn every_spawn_seam_uses_the_merged_launch_builder() { let seams = [ "src/sdk/src/wrapper/run/mod.rs", "src/sdk/src/daemon/providers/execute.rs", - "src/tui/src/worker/executor/run.rs", + "src/tui/src/worker/executor/launch.rs", "src/tui/src/ui/harness_pane/spawn.rs", ]; let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) diff --git a/src/sdk/src/session_history/list.rs b/src/sdk/src/session_history/list.rs index 12d6aed31..60a21c6d0 100644 --- a/src/sdk/src/session_history/list.rs +++ b/src/sdk/src/session_history/list.rs @@ -9,7 +9,7 @@ use std::collections::HashMap; use super::scan::{ claude_sessions_dir, codex_sessions_dir, collect_session_files, is_here, safe_resolve, }; -use super::summary::read_session_summary; +use super::summary::{codex_index_map, read_session_summary}; use super::types::{RawSessionFile, RecentSession, SessionAgentKind}; /// Default number of ranked sessions returned when no limit is given. @@ -43,6 +43,10 @@ pub fn list_recent_sessions( raw.truncate(scan_limit); let here = safe_resolve(cwd); + // Load the Codex session index once into an id-to-label map so + // every session below does not re-parse the index file on each call. + let codex_labels = codex_index_map(env); + // Dedupe by agent+id, keeping the freshest file. let mut by_id: HashMap = HashMap::new(); for file in &raw { @@ -56,12 +60,20 @@ pub fn list_recent_sessions( continue; } } + let label = if file.agent == SessionAgentKind::Codex { + codex_labels + .get(&summary.id) + .cloned() + .unwrap_or(summary.label) + } else { + summary.label + }; by_id.insert( key, RecentSession { agent: file.agent, id: summary.id, - label: summary.label, + label, last_active: file.mtime_ms, path: file.path.to_string_lossy().into_owned(), cwd: summary.cwd, diff --git a/src/sdk/src/session_history/mod.rs b/src/sdk/src/session_history/mod.rs index ff666a5b2..56386db13 100644 --- a/src/sdk/src/session_history/mod.rs +++ b/src/sdk/src/session_history/mod.rs @@ -23,6 +23,7 @@ mod tests; pub use list::list_recent_sessions; pub use scan::{claude_sessions_dir, codex_sessions_dir}; +pub use summary::{codex_thread_label, codex_thread_label_for_cwd}; pub use types::{RecentSession, SessionAgentKind}; pub(crate) use scan::{collect_session_files, discover_session_file, preexisting_session_files}; diff --git a/src/sdk/src/session_history/scan.rs b/src/sdk/src/session_history/scan.rs index 02db66265..1031af382 100644 --- a/src/sdk/src/session_history/scan.rs +++ b/src/sdk/src/session_history/scan.rs @@ -163,6 +163,47 @@ pub(crate) fn discover_session_file( None } +/// Every session file for `agent` rooted at `cwd`, newest first. +/// +/// [`discover_session_file`] returns the single newest match — right for +/// binding a tailer to "the session in this folder". This is the +/// attribution-shaped variant: it returns *all* matches so a caller can prove +/// the folder maps to exactly one session before trusting the newest's name. +/// +/// Unlike `discover_session_file`, a transcript with no recorded cwd is not a +/// candidate: without one it cannot be shown to belong to this folder, so it +/// cannot anchor a label either. +pub(crate) fn session_files_for_cwd( + env: &HashMap, + agent: SessionAgentKind, + cwd: &str, +) -> Vec { + // An unresolvable `here` proves nothing. Two `None` resolves compare equal, + // so without this guard a single session whose cwd also fails to resolve + // would pass the match below and be attributed to this folder on no + // evidence at all — the very mislabeling this function exists to prevent. + let Some(here) = safe_resolve(cwd) else { + return Vec::new(); + }; + let mut files = collect_session_files(agent, &sessions_dir_for(env, agent)); + files.sort_by_key(|file| std::cmp::Reverse(file.mtime_ms)); + files + .into_iter() + .filter_map(|file| { + let canonical = std::fs::canonicalize(&file.path).unwrap_or_else(|_| file.path.clone()); + let summary = read_session_summary(agent, &file.path)?; + let session_cwd = summary.cwd?; + (safe_resolve(&session_cwd).as_deref() == Some(here.as_str())).then_some( + DiscoveredSession { + path: canonical, + id: summary.id, + cwd: Some(session_cwd), + }, + ) + }) + .collect() +} + /// Whether a session's recorded `cwd` resolves to the same path as `here`. /// Both sides must be present for a match. pub(super) fn is_here(cwd: Option<&str>, here: Option<&str>) -> bool { diff --git a/src/sdk/src/session_history/summary.rs b/src/sdk/src/session_history/summary.rs index b02e86141..ae77b5ad5 100644 --- a/src/sdk/src/session_history/summary.rs +++ b/src/sdk/src/session_history/summary.rs @@ -6,6 +6,7 @@ //! the cost of scanning many transcripts. Claude and Codex use different record //! shapes, so each has its own head reader. +use std::collections::HashMap; use std::path::Path; use serde_json::Value; @@ -162,6 +163,87 @@ pub(super) fn slug_label(text: &str) -> String { slug(text) } +/// Read Codex's persisted name for the session rooted at `cwd` — but only when +/// the folder is unambiguous about which session it belongs to. +/// +/// The id-keyed [`codex_thread_label`] is the right read once a transcript has +/// been located, because identity beats recency. This is the fallback for a +/// session nothing has located yet — one an operator created and typed into +/// directly, which never enters the transcript executor — where the best +/// identity on offer is "the Codex rollout in this working directory". +/// +/// The name is attributed only when exactly one rollout is rooted here. With +/// several — two sessions sharing a directory, or a stale rollout from a +/// finished one still newer — the cwd cannot prove which is this session's, and +/// answering with the newest would put another session's name on this row. The +/// fallback then declines and the row keeps its terminal-derived name until +/// identity is found. +pub fn codex_thread_label_for_cwd(env: &HashMap, cwd: &str) -> Option { + // `session_files_for_cwd` is cwd-strict (a transcript with no recorded cwd + // is not a candidate), so a single hit is a positive attribution, not a + // guess. + let mut candidates = super::scan::session_files_for_cwd(env, SessionAgentKind::Codex, cwd); + if candidates.len() != 1 { + return None; + } + let discovered = candidates.pop().expect("exactly one candidate"); + codex_thread_label(env, &discovered.id) +} + +/// Read Codex's persisted name for `session_id`, when it has one. +/// +/// Codex records `/rename` names in `session_index.jsonl` beside its `sessions` +/// directory rather than updating the terminal title. Claude has no equivalent +/// index, so callers use this only for Codex and retain the transcript-prompt +/// fallback when the index has not caught up yet. +/// +/// The lookup goes through [`codex_index_map`] so this single-session read and +/// the batch loader pick the same winning record when the index holds several +/// entries for an id — the newest, last one — instead of the single read +/// disagreeing with the recent-session list. +pub fn codex_thread_label(env: &HashMap, session_id: &str) -> Option { + codex_index_map(env).get(session_id).cloned() +} + +/// Load the Codex session-index into an id-to-label map. +/// +/// Callers that need to resolve thread names for many sessions (e.g. the +/// recent-session list) can load the index once and look up every session +/// against the same map, rather than re-reading and re-parsing the file +/// per session. +/// +/// `session_index.jsonl` is append-only: each `/rename` writes a new record for +/// the same id, so when several records share an id the last one — the newest +/// rename — wins. [`codex_thread_label`] routes through this map for the same +/// guarantee. +pub fn codex_index_map(env: &HashMap) -> HashMap { + let Some(index_path) = (|| { + super::scan::codex_sessions_dir(env) + .parent()? + .join("session_index.jsonl") + .into() + })() else { + return HashMap::new(); + }; + let Ok(contents) = std::fs::read_to_string(index_path) else { + return HashMap::new(); + }; + contents + .lines() + .filter_map(|line| { + let record: Value = serde_json::from_str(line).ok()?; + let object = record.as_object()?; + let id = object.get("id").and_then(Value::as_str)?; + let label = object + .get("thread_name") + .and_then(Value::as_str) + .map(slug_label) + .filter(|l| !l.is_empty())?; + Some((id.to_string(), label)) + }) + .collect() +} + /// Read the first [`HEAD_BYTES`] of `path` as UTF-8 (lossy) and split into /// non-empty lines, dropping a final partial line when the read hit the cap. fn read_head_lines(path: &Path) -> Vec { diff --git a/src/sdk/src/session_history/tests.rs b/src/sdk/src/session_history/tests.rs index 252a4dd6f..e0201a393 100644 --- a/src/sdk/src/session_history/tests.rs +++ b/src/sdk/src/session_history/tests.rs @@ -3,8 +3,8 @@ use super::scan::{collect_session_files, is_here, is_session_file, sessions_dir_for}; use super::summary::{ - as_message_content, extract_text, first_prompt_text, read_claude_summary, read_codex_summary, - slug_label, + as_message_content, codex_index_map, codex_thread_label, codex_thread_label_for_cwd, + extract_text, first_prompt_text, read_claude_summary, read_codex_summary, slug_label, }; use super::*; use crate::ui::util::SLUG_MAX_CHARS; @@ -26,7 +26,7 @@ fn write_session(dir: &Path, name: &str, contents: &str) -> PathBuf { fn ranks_current_cwd_first_then_recency() { let tmp = std::env::temp_dir().join(format!("medulla-sh-{}", std::process::id())); let claude_dir = tmp.join("claude"); - let codex_dir = tmp.join("codex"); + let codex_dir = tmp.join("codex").join("sessions"); fs::create_dir_all(&claude_dir).unwrap(); fs::create_dir_all(&codex_dir).unwrap(); @@ -53,6 +53,11 @@ fn ranks_current_cwd_first_then_recency() { serde_json::json!({"type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"do B here"}]}}) ), ); + fs::write( + tmp.join("codex").join("session_index.jsonl"), + serde_json::json!({"id":"codex-b","thread_name":"Named Codex thread"}).to_string(), + ) + .unwrap(); let mut env = HashMap::new(); env.insert( @@ -69,8 +74,8 @@ fn ranks_current_cwd_first_then_recency() { assert_eq!(sessions[0].id, "codex-b", "current-cwd session ranks first"); assert_eq!(sessions[0].agent, SessionAgentKind::Codex); assert_eq!( - sessions[0].label, "b-here", - "the prompt is slugged, filler dropped" + sessions[0].label, "named-codex-thread", + "Codex's persisted thread name takes precedence over its prompt" ); assert_eq!(sessions[1].id, "claude-a"); assert_eq!(sessions[1].label, "do-a"); @@ -254,6 +259,169 @@ fn codex_summary_uses_id_fallback_and_no_prompt_label() { assert_eq!(summary.label, "(no prompt)"); } +#[test] +fn codex_thread_label_reads_the_persisted_rename() { + let home = tempfile::tempdir().unwrap(); + let codex = home.path().join("codex"); + fs::create_dir_all(codex.join("sessions")).unwrap(); + fs::write( + codex.join("session_index.jsonl"), + serde_json::json!({"id":"codex-1","thread_name":"Ship the sidebar"}).to_string(), + ) + .unwrap(); + let mut env = HashMap::new(); + env.insert( + "MEDULLA_CODEX_SESSIONS_DIR".to_string(), + codex.join("sessions").to_string_lossy().into_owned(), + ); + + assert_eq!( + codex_thread_label(&env, "codex-1").as_deref(), + Some("ship-sidebar") + ); + assert_eq!(codex_thread_label(&env, "missing"), None); +} + +#[test] +fn codex_thread_label_and_index_map_agree_on_duplicate_ids() { + // The index is append-only: a second /rename for the same id appends a + // second record. Both the single-id lookup and the batch map must surface + // the newest (last) record, or the rail and the Sessions tab diverge. + let home = tempfile::tempdir().unwrap(); + let codex = home.path().join("codex"); + fs::create_dir_all(codex.join("sessions")).unwrap(); + fs::write( + codex.join("session_index.jsonl"), + format!( + "{}\n{}\n", + serde_json::json!({"id":"codex-1","thread_name":"Ship the sidebar"}), + serde_json::json!({"id":"codex-1","thread_name":"Land the auth flow"}) + ), + ) + .unwrap(); + let mut env = HashMap::new(); + env.insert( + "MEDULLA_CODEX_SESSIONS_DIR".to_string(), + codex.join("sessions").to_string_lossy().into_owned(), + ); + + let map = codex_index_map(&env); + assert_eq!( + map.get("codex-1").map(String::as_str), + Some("land-auth-flow") + ); + assert_eq!( + codex_thread_label(&env, "codex-1").as_deref(), + Some("land-auth-flow") + ); +} + +#[test] +fn codex_thread_label_for_cwd_finds_the_newest_rollout_in_the_folder() { + let home = tempfile::tempdir().unwrap(); + let sessions = home.path().join("codex").join("sessions"); + let project = home.path().join("project"); + fs::create_dir_all(&sessions).unwrap(); + fs::create_dir_all(&project).unwrap(); + let project_str = project.to_string_lossy().into_owned(); + + write_session( + &sessions, + "rollout-a.jsonl", + &serde_json::json!({ + "type":"session_meta", + "payload":{"session_id":"codex-a","cwd": project_str} + }) + .to_string(), + ); + fs::write( + home.path().join("codex").join("session_index.jsonl"), + serde_json::json!({"id":"codex-a","thread_name":"Ship the sidebar"}).to_string(), + ) + .unwrap(); + let mut env = HashMap::new(); + env.insert( + "MEDULLA_CODEX_SESSIONS_DIR".to_string(), + sessions.to_string_lossy().into_owned(), + ); + + assert_eq!( + codex_thread_label_for_cwd(&env, &project_str).as_deref(), + Some("ship-sidebar") + ); + // A cwd with no session in it has no label to read. + let elsewhere = home.path().join("elsewhere").to_string_lossy().into_owned(); + assert_eq!(codex_thread_label_for_cwd(&env, &elsewhere), None); +} + +#[test] +fn codex_thread_label_for_cwd_needs_an_unambiguous_folder() { + // Two sessions sharing a directory: the cwd cannot prove which rollout + // produced this label, so the fallback must decline rather than put one + // session's name on the other's row. + let home = tempfile::tempdir().unwrap(); + let sessions = home.path().join("codex").join("sessions"); + let project = home.path().join("project"); + fs::create_dir_all(&sessions).unwrap(); + fs::create_dir_all(&project).unwrap(); + let project_str = project.to_string_lossy().into_owned(); + + write_session( + &sessions, + "rollout-a.jsonl", + &serde_json::json!({ + "type":"session_meta", + "payload":{"session_id":"codex-a","cwd": project_str} + }) + .to_string(), + ); + std::thread::sleep(std::time::Duration::from_millis(20)); + write_session( + &sessions, + "rollout-b.jsonl", + &serde_json::json!({ + "type":"session_meta", + "payload":{"session_id":"codex-b","cwd": project_str} + }) + .to_string(), + ); + fs::write( + home.path().join("codex").join("session_index.jsonl"), + format!( + "{}\n{}\n", + serde_json::json!({"id":"codex-a","thread_name":"Ship the sidebar"}), + serde_json::json!({"id":"codex-b","thread_name":"Land the auth flow"}) + ), + ) + .unwrap(); + let mut env = HashMap::new(); + env.insert( + "MEDULLA_CODEX_SESSIONS_DIR".to_string(), + sessions.to_string_lossy().into_owned(), + ); + + assert_eq!(codex_thread_label_for_cwd(&env, &project_str), None); + // A transcript with no recorded cwd is not a candidate either, nor is one + // whose head window yields no summary at all. + write_session( + &sessions, + "rollout-cwdless.jsonl", + &serde_json::json!({ + "type":"session_meta", + "payload":{"session_id":"codex-c"} + }) + .to_string(), + ); + write_session( + &sessions, + "rollout-nonesummary.jsonl", + &serde_json::json!({"type":"response_item"}).to_string(), + ); + let alone = home.path().join("solo").to_string_lossy().into_owned(); + fs::create_dir_all(&alone).unwrap(); + assert_eq!(codex_thread_label_for_cwd(&env, &alone), None); +} + #[test] fn codex_summary_without_meta_is_none() { let lines = vec![serde_json::json!({"type":"response_item"}).to_string()]; diff --git a/src/tui/src/worker/executor/launch.rs b/src/tui/src/worker/executor/launch.rs new file mode 100644 index 000000000..6f0d68df5 --- /dev/null +++ b/src/tui/src/worker/executor/launch.rs @@ -0,0 +1,134 @@ +//! Session spawn: launch a fresh harness on the blocking pool and build the +//! environment it spawns with. +//! +//! Whether a task reuses an idle session, launches fresh, or queues is decided +//! up front in [`super::probe`]; this module owns the *fresh-launch* half — +//! [`super::PtySessionExecutor::launch`] opens the harness through +//! [`PtyManager`](crate::worker::pty::PtyManager), and +//! [`super::PtySessionExecutor::spawn_env`] builds the environment and extra +//! arguments it spawns with. + +use std::collections::HashMap; + +use medulla::daemon::providers::RunTaskOptions; + +use super::super::pty::LaunchSpec; +use super::types::{OpenedSession, PtySessionExecutor}; + +impl PtySessionExecutor { + /// Start a fresh harness on the blocking pool. + /// + /// [`PtyManager::open`] forks, execs, and may back off while a pty frees up + /// — all of it blocking. Calling it inline parked a tokio worker for up to + /// half a second per launch, and a burst of task frames could park most of + /// the runtime, taking the inbox drain and every screen sampler down with + /// it since they share one. So the launch goes to the blocking pool, which + /// is what it is for. + pub(super) async fn launch(&self, spec: LaunchSpec) -> Result { + let gh_repo_is_set = spec.env.contains_key("GH_REPO"); + let sessions = self.sessions.clone(); + let id = tokio::task::spawn_blocking(move || sessions.open(spec)) + .await + .map_err(|err| format!("pty launch did not complete: {err}"))??; + let harness_session_id = self.sessions.row(&id).and_then(|row| row.session_id); + Ok(OpenedSession { + id, + harness_session_id, + reused: false, + gh_repo_is_set, + }) + } + + /// The environment and extra argv a fresh launch spawns with: this + /// task-scoped environment, layered with the `[router]` injection the + /// headless executor already applies at its own spawn seam. + /// + /// Without this, switching the local host to `PtySessionExecutor` silently + /// dropped a configured router — the child spawned against its own default + /// endpoint instead of the one the operator pointed it at, with no error to + /// say so. + /// + /// # Errors + /// + /// A configured `apiKeyEnv` whose named variable is unset in this + /// executor's environment is a hard error, matching the headless path: a + /// silently-empty key would spawn the harness unauthenticated against the + /// routed endpoint. + pub(super) fn spawn_env( + &self, + options: &RunTaskOptions, + ) -> Result<(HashMap, Vec), String> { + let mut env = options.env.clone(); + // This executor launches the watched harness itself, bypassing the + // daemon's transport dispatcher. Keep the embedded core workspace out + // of that child for the same credential-store isolation as headless + // and alternate transports. + medulla::protocol::env::scrub_core_state(&mut env, options.provider); + let mut extra_args = options.extra_args.clone(); + // Commits made in a watched PTY session are just as much Medulla's work + // as headless ones, so this path carries the same attribution. + let attribution_env = medulla::attribution::attribution_env(options.attribution, &env); + env.extend(attribution_env); + // Attribution and the operator's configured hooks share Claude Code's + // single `--settings` flag, so both are built together — a watched PTY + // session runs the same lifecycle policy a headless one does. + let (launch_args, hook_notes) = medulla::harness_hooks::launch_args( + options.provider, + options.attribution, + &options.hooks, + &env, + ); + extra_args.extend(launch_args); + // Routed to the log rather than stderr: this crate draws a full-screen + // TUI, where a stray line corrupts the pane. Covers both hooks the + // harness cannot run and hooks it will not run until trusted. + if let Some(log) = &self.log { + for note in &hook_notes { + log(note); + } + } + // OpenRouter-bound runs are re-pointed at Medulla's loopback attribution + // proxy, and the real key is scrubbed from `env` here, before any of it + // reaches the child. A no-op for every other endpoint. + let mut router = options.router.clone(); + medulla::inference_proxy::route_spawn(options.provider, &mut router, &mut env)?; + if let Some(router) = &router { + let injection = medulla::protocol::env::router_env(options.provider, router); + for (key, value) in injection.env { + env.insert(key, value); + } + for (child_var, source_name) in injection.secret_env { + // Resolved from `env`, not `options.env`: when the run was routed + // through the attribution proxy the name to resolve is the token + // the routing just placed there, and the original key has been + // scrubbed. Cloned before inserting so the read does not borrow + // across the write. + let secret = env + .get(&source_name) + .filter(|value| !value.is_empty()) + .cloned(); + match secret { + Some(secret) => { + env.insert(child_var, secret); + } + None => { + return Err(format!( + "router API key env var `{source_name}` is not set; \ + export it or remove apiKeyEnv from [router]" + )); + } + } + } + extra_args.extend(injection.args); + } + // Codex needs more than an endpoint before a routed model will answer: + // a provider block, an API-key auth preference, and a catalog entry it + // is willing to describe. Read from `env`, which now holds both the + // preset's opt-in knobs and the endpoint the routing above wrote. + extra_args.extend( + medulla::codex_overrides::launch_args(options.provider, options.model.as_deref(), &env) + .map_err(|error| error.to_string())?, + ); + Ok((env, extra_args)) + } +} diff --git a/src/tui/src/worker/executor/mod.rs b/src/tui/src/worker/executor/mod.rs index bcdbe9f28..efb7519cf 100644 --- a/src/tui/src/worker/executor/mod.rs +++ b/src/tui/src/worker/executor/mod.rs @@ -1,15 +1,18 @@ //! PTY-backed execution of delegated harness tasks. //! -//! [`PtySessionExecutor`] is the public adapter. [`run`] owns its execution -//! behavior, [`probe`] owns deciding which session serves a task, [`hold`] owns -//! what happens when an operator is in the way — queue, suspend, hand back — -//! and [`types`] owns the executor and session-planning data. +//! [`PtySessionExecutor`] is the public adapter. [`run`] owns its dispatch +//! (timeout, retry, handoff), [`probe`] owns deciding which session serves a +//! task, [`launch`] owns the fresh-harness spawn, [`turn`] owns transcript +//! polling, [`hold`] owns queue/suspend/handback, and [`types`] owns the +//! executor and session-planning data. mod hold; +mod launch; mod probe; mod run; #[cfg(test)] mod tests; +mod turn; mod types; pub use run::agent_kind; diff --git a/src/tui/src/worker/executor/run.rs b/src/tui/src/worker/executor/run.rs index 374c8ec99..e38ef923c 100644 --- a/src/tui/src/worker/executor/run.rs +++ b/src/tui/src/worker/executor/run.rs @@ -31,36 +31,36 @@ use std::time::Duration; use medulla::daemon::providers::{RunTaskFn, RunTaskOptions, RunTaskResult}; use medulla::protocol::HarnessProvider; use medulla::session_history::SessionAgentKind; -use medulla::sessions::{SessionClass, TurnStream}; +use medulla::sessions::SessionClass; use medulla::wrapper::tail::SessionTailer; -use super::super::pty::{LaunchSpec, PtyManager, SessionControl}; -use super::types::{OpenedSession, PtySessionExecutor, SessionPlan, TurnSpec, WorkspaceContext}; +use super::super::pty::{PtyManager, SessionControl}; +use super::types::{PtySessionExecutor, SessionPlan, TurnSpec, WorkspaceContext}; /// How often the transcript is polled while a turn runs. /// /// Fast enough that a short turn settles promptly, slow enough that a long one /// costs almost nothing. The transcript is a file on local disk, so this is a /// stat plus a short read. -const POLL: Duration = Duration::from_millis(150); +pub(super) const POLL: Duration = Duration::from_millis(150); /// How long to keep looking for a session's transcript before giving up. /// /// A harness writes its first record only once it has started work, which on a /// cold start can take a few seconds. -const LOCATE_BUDGET: Duration = Duration::from_secs(30); +pub(super) const LOCATE_BUDGET: Duration = Duration::from_secs(30); /// Silence that settles a turn whose completion record carried no stated reason. /// /// Only reachable for the ~0.08% of claude records with no `stop_reason`; the /// watcher refuses to stall while a tool call is outstanding, so a long build is /// never mistaken for a finished turn. -const STALL_BUDGET_MS: i64 = 120_000; +pub(super) const STALL_BUDGET_MS: i64 = 120_000; /// How long to wait for the rest of a terminal message before replying with what /// arrived. The blocks of one message are written in a single burst, so this only /// has to outlast that write — it is a safety net, not the normal path. -const SETTLE_GRACE_MS: i64 = 1_500; +pub(super) const SETTLE_GRACE_MS: i64 = 1_500; impl PtySessionExecutor { /// Build an executor over the TUI's live session manager. @@ -371,7 +371,7 @@ impl PtySessionExecutor { /// harness still finishing the last one interleaves two prompts into one /// composer, which is the failure that produces confidently wrong answers /// rather than an error. - fn stop_turn(&self, id: &str) { + pub(super) fn stop_turn(&self, id: &str) { let stopped = self.sessions.stop_if_orchestrator(id); retire_stopped_workspace_context( &mut self @@ -382,375 +382,8 @@ impl PtySessionExecutor { stopped, ); } - - /// Start a fresh harness on the blocking pool. - /// - /// [`PtyManager::open`] forks, execs, and may back off while a pty frees up - /// — all of it blocking. Calling it inline parked a tokio worker for up to - /// half a second per launch, and a burst of task frames could park most of - /// the runtime, taking the inbox drain and every screen sampler down with - /// it since they share one. So the launch goes to the blocking pool, which - /// is what it is for. - async fn launch(&self, spec: LaunchSpec) -> Result { - let gh_repo_is_set = spec.env.contains_key("GH_REPO"); - let sessions = self.sessions.clone(); - let id = tokio::task::spawn_blocking(move || sessions.open(spec)) - .await - .map_err(|err| format!("pty launch did not complete: {err}"))??; - let harness_session_id = self.sessions.row(&id).and_then(|row| row.session_id); - Ok(OpenedSession { - id, - harness_session_id, - reused: false, - gh_repo_is_set, - }) - } - - /// The environment and extra argv a fresh launch spawns with: this - /// task-scoped environment, layered with the `[router]` injection the - /// headless executor already applies at its own spawn seam. - /// - /// Without this, switching the local host to `PtySessionExecutor` silently - /// dropped a configured router — the child spawned against its own default - /// endpoint instead of the one the operator pointed it at, with no error to - /// say so. - /// - /// # Errors - /// - /// A configured `apiKeyEnv` whose named variable is unset in this - /// executor's environment is a hard error, matching the headless path: a - /// silently-empty key would spawn the harness unauthenticated against the - /// routed endpoint. - pub(super) fn spawn_env( - &self, - options: &RunTaskOptions, - ) -> Result<(HashMap, Vec), String> { - let mut env = options.env.clone(); - // This executor launches the watched harness itself, bypassing the - // daemon's transport dispatcher. Keep the embedded core workspace out - // of that child for the same credential-store isolation as headless - // and alternate transports. - medulla::protocol::env::scrub_core_state(&mut env, options.provider); - let mut extra_args = options.extra_args.clone(); - // Commits made in a watched PTY session are just as much Medulla's work - // as headless ones, so this path carries the same attribution. - let attribution_env = medulla::attribution::attribution_env(options.attribution, &env); - env.extend(attribution_env); - // Attribution and the operator's configured hooks share Claude Code's - // single `--settings` flag, so both are built together — a watched PTY - // session runs the same lifecycle policy a headless one does. - let (launch_args, hook_notes) = medulla::harness_hooks::launch_args( - options.provider, - options.attribution, - &options.hooks, - &env, - ); - extra_args.extend(launch_args); - // Routed to the log rather than stderr: this crate draws a full-screen - // TUI, where a stray line corrupts the pane. Covers both hooks the - // harness cannot run and hooks it will not run until trusted. - if let Some(log) = &self.log { - for note in &hook_notes { - log(note); - } - } - // OpenRouter-bound runs are re-pointed at Medulla's loopback attribution - // proxy, and the real key is scrubbed from `env` here, before any of it - // reaches the child. A no-op for every other endpoint. - let mut router = options.router.clone(); - medulla::inference_proxy::route_spawn(options.provider, &mut router, &mut env)?; - if let Some(router) = &router { - let injection = medulla::protocol::env::router_env(options.provider, router); - for (key, value) in injection.env { - env.insert(key, value); - } - for (child_var, source_name) in injection.secret_env { - // Resolved from `env`, not `options.env`: when the run was routed - // through the attribution proxy the name to resolve is the token - // the routing just placed there, and the original key has been - // scrubbed. Cloned before inserting so the read does not borrow - // across the write. - let secret = env - .get(&source_name) - .filter(|value| !value.is_empty()) - .cloned(); - match secret { - Some(secret) => { - env.insert(child_var, secret); - } - None => { - return Err(format!( - "router API key env var `{source_name}` is not set; \ - export it or remove apiKeyEnv from [router]" - )); - } - } - } - extra_args.extend(injection.args); - } - // Codex needs more than an endpoint before a routed model will answer: - // a provider block, an API-key auth preference, and a catalog entry it - // is willing to describe. Read from `env`, which now holds both the - // preset's opt-in knobs and the endpoint the routing above wrote. - extra_args.extend( - medulla::codex_overrides::launch_args(options.provider, options.model.as_deref(), &env) - .map_err(|error| error.to_string())?, - ); - Ok((env, extra_args)) - } - - /// Fold whatever the harness has written since the last poll, and answer - /// with the turn's result if that fold completed it. - /// - /// Shared by the polling loop and by the suspend path, and shared - /// deliberately: "read what is already there before doing anything else" has - /// to mean the same thing in both, or a turn that finished microseconds - /// before an operator took the session would have its answer read by one - /// path and dropped by the other. - /// - /// `last_line_at` is advanced per line rather than per call, because it is - /// the idle watchdog's clock and a batch of lines is progress at the time - /// each of them was read, not at the time the batch was drained. - fn fold_available( - &self, - id: &str, - provider: HarnessProvider, - tailer: &mut SessionTailer, - stream: &mut TurnStream, - on_event: &mut Option, - last_line_at: &mut i64, - ) -> Option { - let poll = tailer.poll(); - // Codex cannot be told its id, so it is learned from the rollout the - // first time the tailer locates one. - if let Some(located) = &poll.located { - self.sessions - .record_session_id(id, located.harness_session_id.clone()); - } - for line in poll.lines { - *last_line_at = medulla::clock::now_millis(); - let fold = stream.observe(&line.text); - self.workspace_context - .lock() - .expect("workspace context lock poisoned") - .insert(id.to_string(), stream.workspace_context()); - // The peer watches its task through these. Dropping them would - // leave it with an ack, silence, then a reply — which is what - // this executor used to do. - if let Some(callback) = on_event.as_mut() { - for event in &fold.events { - callback(event); - } - } - if let Some(reply) = fold.reply { - return Some(RunTaskResult { - provider, - reply, - events: stream.events(), - usage: stream.usage(), - session_id: self.sessions.row(id).and_then(|row| row.session_id), - }); - } - } - None - } - - /// Poll the transcript until the harness says the turn is over. - /// - /// `timeout_ms` is the caller's configured idle watchdog (`[host] - /// .taskTimeoutMs`, mirroring the headless executor's own `timeout_ms`) — - /// the hard ceiling on how long a turn may go without producing a single - /// transcript line. It is distinct from, and can override, the two fixed - /// budgets below: [`LOCATE_BUDGET`] covers a harness that never starts a - /// turn at all, and [`STALL_BUDGET_MS`] is a soft "probably finished" - /// signal for a transcript that stops without a stated reason. A caller - /// configuring a shorter ceiling than either means it, and is honored - /// ahead of them. - async fn await_turn( - &self, - id: &str, - spec: TurnSpec, - mut tailer: SessionTailer, - abort: medulla::daemon::providers::Abort, - mut on_event: Option, - ) -> Result { - let TurnSpec { - provider, - gh_repo_is_set, - timeout_ms, - instruction, - } = spec; - let mut stream = TurnStream::new_with_gh_repo_override(provider, gh_repo_is_set); - if let Some((cwd, branch, pull_request)) = self - .workspace_context - .lock() - .expect("workspace context lock poisoned") - .get(id) - .cloned() - { - stream.set_workspace_context(cwd, branch, pull_request); - if let (Some(callback), Some(event)) = - (on_event.as_mut(), stream.retained_workspace_event()) - { - callback(&event); - } - } - let mut started = tokio::time::Instant::now(); - let mut last_line_at = medulla::clock::now_millis(); - - loop { - // Taking control is an ownership transfer, not merely a display - // preference, so it is answered before aborts or transcript output: - // from here the executor must not send Ctrl-C, report a stale - // completion, or close the PTY underneath the operator. - // - // What it does instead is **suspend** (spec §5). The turn used to - // return an error here, throwing away everything the harness had - // produced and telling the orchestrator its task had failed — for - // the entirely ordinary event of a person opening the session to - // look. Now the fold, its events, its usage and its workspace - // context all stay exactly where they are, the session keeps the - // work, and the task stays open. - if self.sessions.control(id) == Some(SessionControl::User) { - // Everything already written belongs to *this* turn — the - // takeover cannot retroactively unwrite it. Folded out before - // suspending, so a turn that finished in the instant somebody - // took the session still reports the answer it had reached. - if let Some(result) = self.fold_available( - id, - provider, - &mut tailer, - &mut stream, - &mut on_event, - &mut last_line_at, - ) { - return Ok(result); - } - super::hold::report_held(&mut on_event, provider); - self.await_handback(id, provider, &abort).await?; - // The lines the operator's own work wrote are theirs, not this - // turn's: dropped rather than folded, or the person's last - // exchange would settle the task as its answer. What they did is - // not lost — it is in the session, which is exactly what the - // hand-back turn is told to go and read. - let poll = tailer.poll(); - if let Some(located) = &poll.located { - self.sessions - .record_session_id(id, located.harness_session_id.clone()); - } - super::hold::report_resumed(&mut on_event, provider); - super::super::pty::inject_prompt( - &self.sessions, - id, - &super::hold::handback_prompt(&instruction), - ) - .await?; - // Both budgets restart with the hand-back turn, which is what - // "the watchdog is paused, not lengthened" means on this side: - // held time is excluded rather than counted, so a session held - // over lunch is not a task that timed out at the desk. - started = tokio::time::Instant::now(); - last_line_at = medulla::clock::now_millis(); - continue; - } - if abort.is_aborted() { - if abort.is_terminated() { - self.stop_turn(id); - } else { - // A requester abort is an interrupt: Ctrl-C reaches the - // harness the same way the operator's would, and the - // reusable session survives it. - let _ = self.sessions.write(id, &[0x03]); - } - return Err(format!("{} task aborted", provider.as_str())); - } - if !self - .sessions - .row(id) - .is_some_and(|row| row.state.is_running()) - { - return Err(format!( - "{} session ended before the turn did", - provider.as_str() - )); - } - - if let Some(result) = self.fold_available( - id, - provider, - &mut tailer, - &mut stream, - &mut on_event, - &mut last_line_at, - ) { - return Ok(result); - } - - if !tailer.is_located() && started.elapsed() > LOCATE_BUDGET { - // A harness writes its transcript once it starts a turn, so an - // absent one usually means it never started one — most often - // because it is still waiting on something on screen that - // `blocking_dialog` did not recognise. Say where to look; the - // bare "could not find the transcript" sent operators hunting - // through `~/.claude/projects` for a file that was never going - // to exist. - return Err(format!( - "{} never started a turn — check the session in the Sessions tab; \ - it may be waiting on a prompt", - provider.as_str() - )); - } - let idle_ms = medulla::clock::now_millis().saturating_sub(last_line_at); - // The configured idle ceiling, checked first so a caller-set budget - // shorter than the fixed ones below actually takes effect instead of - // being silently outlived by them. `timeout_ms == 0` means no - // configured ceiling (never observed from `[host]`, whose default is - // nonzero, but a defensive floor all the same). - if timeout_ms > 0 && idle_ms as u64 >= timeout_ms { - // Stop the harness before reporting the failure. A timeout is - // only silence on the *transcript* — the child is very much - // alive and may still be editing the workspace. Returning - // without stopping it tells the peer the task failed while the - // work carries on unattributed, and an unbound session would - // then be released as idle for the next task to claim, landing - // its prompt in a harness that is still mid-turn. - self.stop_turn(id); - return Err(format!( - "{} task idle for {timeout_ms}ms (no events)", - provider.as_str() - )); - } - // The turn ended, but its message is written one record per content - // block and the reply usually lives in the last one. Normally the - // records that follow close it immediately; this covers a transcript - // that simply stops, so a finished turn is never held for the full - // stall budget. - if stream.terminal_pending() && idle_ms >= SETTLE_GRACE_MS { - if let Some(reply) = stream.settle_pending() { - return Ok(RunTaskResult { - provider, - reply, - events: stream.events(), - usage: stream.usage(), - session_id: self.sessions.row(id).and_then(|row| row.session_id), - }); - } - } - if tailer.is_located() && stream.stalled_for(idle_ms, STALL_BUDGET_MS) { - return Ok(RunTaskResult { - provider, - reply: stream.settle_stalled(), - events: stream.events(), - usage: stream.usage(), - session_id: self.sessions.row(id).and_then(|row| row.session_id), - }); - } - tokio::time::sleep(POLL).await; - } - } } -/// Retain mapper state only while the PTY can serve a later turn. pub(super) fn retains_workspace_context( class: SessionClass, control: Option, diff --git a/src/tui/src/worker/executor/turn.rs b/src/tui/src/worker/executor/turn.rs new file mode 100644 index 000000000..f3f016d7c --- /dev/null +++ b/src/tui/src/worker/executor/turn.rs @@ -0,0 +1,277 @@ +//! Poll the harness transcript until the turn is settled. +//! +//! [`super::PtySessionExecutor::fold_available`] drains the tailed transcript +//! through [`medulla::sessions::TurnStream`], and +//! [`super::PtySessionExecutor::await_turn`] is the main polling loop that +//! runs until the harness states the turn is done, the turn times out, or an +//! operator takes control. + +use medulla::daemon::providers::RunTaskResult; +use medulla::protocol::HarnessProvider; +use medulla::sessions::TurnStream; +use medulla::wrapper::tail::SessionTailer; + +use super::super::pty::SessionControl; +use super::run::{LOCATE_BUDGET, POLL, SETTLE_GRACE_MS, STALL_BUDGET_MS}; +use super::types::{PtySessionExecutor, TurnSpec}; + +impl PtySessionExecutor { + fn fold_available( + &self, + id: &str, + provider: HarnessProvider, + tailer: &mut SessionTailer, + stream: &mut TurnStream, + on_event: &mut Option, + last_line_at: &mut i64, + ) -> Option { + let poll = tailer.poll(); + // Codex cannot be told its id, so it is learned from the rollout the + // first time the tailer locates one. + if let Some(located) = &poll.located { + self.sessions + .record_session_id(id, located.harness_session_id.clone()); + if provider == HarnessProvider::Codex { + if let Some(thread_name) = medulla::session_history::codex_thread_label( + &self.env, + &located.harness_session_id, + ) { + self.sessions.record_thread_name(id, thread_name); + } + } + } + for line in poll.lines { + *last_line_at = medulla::clock::now_millis(); + let fold = stream.observe(&line.text); + self.workspace_context + .lock() + .expect("workspace context lock poisoned") + .insert(id.to_string(), stream.workspace_context()); + // The peer watches its task through these. Dropping them would + // leave it with an ack, silence, then a reply — which is what + // this executor used to do. + if let Some(callback) = on_event.as_mut() { + for event in &fold.events { + callback(event); + } + } + if let Some(reply) = fold.reply { + return Some(RunTaskResult { + provider, + reply, + events: stream.events(), + usage: stream.usage(), + session_id: self.sessions.row(id).and_then(|row| row.session_id), + }); + } + } + None + } + + /// Poll the transcript until the harness says the turn is over. + /// + /// `timeout_ms` is the caller's configured idle watchdog (`[host] + /// .taskTimeoutMs`, mirroring the headless executor's own `timeout_ms`) — + /// the hard ceiling on how long a turn may go without producing a single + /// transcript line. It is distinct from, and can override, the two fixed + /// budgets below: [`LOCATE_BUDGET`] covers a harness that never starts a + /// turn at all, and [`STALL_BUDGET_MS`] is a soft "probably finished" + /// signal for a transcript that stops without a stated reason. A caller + /// configuring a shorter ceiling than either means it, and is honored + /// ahead of them. + pub(super) async fn await_turn( + &self, + id: &str, + spec: TurnSpec, + mut tailer: SessionTailer, + abort: medulla::daemon::providers::Abort, + mut on_event: Option, + ) -> Result { + let TurnSpec { + provider, + gh_repo_is_set, + timeout_ms, + instruction, + } = spec; + let mut stream = TurnStream::new_with_gh_repo_override(provider, gh_repo_is_set); + if let Some((cwd, branch, pull_request)) = self + .workspace_context + .lock() + .expect("workspace context lock poisoned") + .get(id) + .cloned() + { + stream.set_workspace_context(cwd, branch, pull_request); + if let (Some(callback), Some(event)) = + (on_event.as_mut(), stream.retained_workspace_event()) + { + callback(&event); + } + } + let mut started = tokio::time::Instant::now(); + let mut last_line_at = medulla::clock::now_millis(); + + loop { + // Taking control is an ownership transfer, not merely a display + // preference, so it is answered before aborts or transcript output: + // from here the executor must not send Ctrl-C, report a stale + // completion, or close the PTY underneath the operator. + // + // What it does instead is **suspend** (spec §5). The turn used to + // return an error here, throwing away everything the harness had + // produced and telling the orchestrator its task had failed — for + // the entirely ordinary event of a person opening the session to + // look. Now the fold, its events, its usage and its workspace + // context all stay exactly where they are, the session keeps the + // work, and the task stays open. + if self.sessions.control(id) == Some(SessionControl::User) { + // Everything already written belongs to *this* turn — the + // takeover cannot retroactively unwrite it. Folded out before + // suspending, so a turn that finished in the instant somebody + // took the session still reports the answer it had reached. + if let Some(result) = self.fold_available( + id, + provider, + &mut tailer, + &mut stream, + &mut on_event, + &mut last_line_at, + ) { + return Ok(result); + } + super::hold::report_held(&mut on_event, provider); + self.await_handback(id, provider, &abort).await?; + // The lines the operator's own work wrote are theirs, not this + // turn's: dropped rather than folded, or the person's last + // exchange would settle the task as its answer. What they did is + // not lost — it is in the session, which is exactly what the + // hand-back turn is told to go and read. + let poll = tailer.poll(); + if let Some(located) = &poll.located { + self.sessions + .record_session_id(id, located.harness_session_id.clone()); + } + super::hold::report_resumed(&mut on_event, provider); + super::super::pty::inject_prompt( + &self.sessions, + id, + &super::hold::handback_prompt(&instruction), + ) + .await?; + // Both budgets restart with the hand-back turn, which is what + // "the watchdog is paused, not lengthened" means on this side: + // held time is excluded rather than counted, so a session held + // over lunch is not a task that timed out at the desk. + started = tokio::time::Instant::now(); + last_line_at = medulla::clock::now_millis(); + continue; + } + if abort.is_aborted() { + if abort.is_terminated() { + self.stop_turn(id); + } else { + // A requester abort is an interrupt: Ctrl-C reaches the + // harness the same way the operator's would, and the + // reusable session survives it. + let _ = self.sessions.write(id, &[0x03]); + } + return Err(format!("{} task aborted", provider.as_str())); + } + if !self + .sessions + .row(id) + .is_some_and(|row| row.state.is_running()) + { + return Err(format!( + "{} session ended before the turn did", + provider.as_str() + )); + } + + if let Some(result) = self.fold_available( + id, + provider, + &mut tailer, + &mut stream, + &mut on_event, + &mut last_line_at, + ) { + return Ok(result); + } + + // Codex `/rename` names are re-read for the session's whole life by + // the manager's per-session label poller (`PtyManager::spawn_codex_label_poller`), + // which is keyed to the PTY session lifecycle and therefore also runs + // while the turn is held, idle, or retained. Nothing else re-reads + // the index here. + + if !tailer.is_located() && started.elapsed() > LOCATE_BUDGET { + // A harness writes its transcript once it starts a turn, so an + // absent one usually means it never started one — most often + // because it is still waiting on something on screen that + // `blocking_dialog` did not recognise. Say where to look; the + // bare "could not find the transcript" sent operators hunting + // through `~/.claude/projects` for a file that was never going + // to exist. + return Err(format!( + "{} never started a turn — check the session in the Sessions tab; \ + it may be waiting on a prompt", + provider.as_str() + )); + } + // `now_millis` is wall-clock (`SystemTime`), so a backward step (an + // NTP sync, a suspend/resume) makes the delta negative. Casting a + // negative i64 to u64 wraps to a huge value, which would fire the + // idle timeout below and stop a harness that is merely the victim + // of a clock change. A zero floor keeps the watchdog silent instead. + let idle_ms = medulla::clock::now_millis() + .saturating_sub(last_line_at) + .max(0); + // The configured idle ceiling, checked first so a caller-set budget + // shorter than the fixed ones below actually takes effect instead of + // being silently outlived by them. `timeout_ms == 0` means no + // configured ceiling (never observed from `[host]`, whose default is + // nonzero, but a defensive floor all the same). + if timeout_ms > 0 && idle_ms as u64 >= timeout_ms { + // Stop the harness before reporting the failure. A timeout is + // only silence on the *transcript* — the child is very much + // alive and may still be editing the workspace. Returning + // without stopping it tells the peer the task failed while the + // work carries on unattributed, and an unbound session would + // then be released as idle for the next task to claim, landing + // its prompt in a harness that is still mid-turn. + self.stop_turn(id); + return Err(format!( + "{} task idle for {timeout_ms}ms (no events)", + provider.as_str() + )); + } + // The turn ended, but its message is written one record per content + // block and the reply usually lives in the last one. Normally the + // records that follow close it immediately; this covers a transcript + // that simply stops, so a finished turn is never held for the full + // stall budget. + if stream.terminal_pending() && idle_ms >= SETTLE_GRACE_MS { + if let Some(reply) = stream.settle_pending() { + return Ok(RunTaskResult { + provider, + reply, + events: stream.events(), + usage: stream.usage(), + session_id: self.sessions.row(id).and_then(|row| row.session_id), + }); + } + } + if tailer.is_located() && stream.stalled_for(idle_ms, STALL_BUDGET_MS) { + return Ok(RunTaskResult { + provider, + reply: stream.settle_stalled(), + events: stream.events(), + usage: stream.usage(), + session_id: self.sessions.row(id).and_then(|row| row.session_id), + }); + } + tokio::time::sleep(POLL).await; + } + } +} diff --git a/src/tui/src/worker/pty/handle/lifecycle.rs b/src/tui/src/worker/pty/handle/lifecycle.rs index 400c75841..fea0a1416 100644 --- a/src/tui/src/worker/pty/handle/lifecycle.rs +++ b/src/tui/src/worker/pty/handle/lifecycle.rs @@ -62,6 +62,7 @@ impl SessionHandle { session_id, name, thread_name: None, + index_thread_name: None, // Filled in by the caller before the handle is published, and // re-read on a timer from there on: see // `PtyManager::spawn_checkout_poller`. diff --git a/src/tui/src/worker/pty/handle/screen.rs b/src/tui/src/worker/pty/handle/screen.rs index d075a12d9..1eb8bc889 100644 --- a/src/tui/src/worker/pty/handle/screen.rs +++ b/src/tui/src/worker/pty/handle/screen.rs @@ -95,7 +95,13 @@ impl SessionHandle { let title = parser.screen().title().trim().to_string(); (!title.is_empty()).then_some(title) }; - lock(&self.cold).thread_name = thread_name; + // An empty OSC title (ordinary harness output that happens to include + // no title escape) must not erase a name the session already carries — + // discovered from the Codex session index or from a prior non-empty + // title. Only a real title replaces it. + if thread_name.is_some() { + lock(&self.cold).thread_name = thread_name; + } } /// Move the emulator's scrollback by `rows`, towards the history when `up`. diff --git a/src/tui/src/worker/pty/handle/state.rs b/src/tui/src/worker/pty/handle/state.rs index 8c2aa743e..c5de0941a 100644 --- a/src/tui/src/worker/pty/handle/state.rs +++ b/src/tui/src/worker/pty/handle/state.rs @@ -152,6 +152,15 @@ impl SessionHandle { self.last_output_at.store(now, Ordering::Release); } + /// Record the thread name discovered outside the terminal stream. + /// + /// Codex persists renamed threads in its session index instead of emitting + /// an OSC window title, so the transcript executor supplies that value once + /// it has identified the session. + pub(in super::super) fn record_thread_name(&self, thread_name: String) { + lock(&self.cold).index_thread_name = Some(thread_name); + } + /// The operator-facing projection of this session, for the list pane. pub fn row(&self) -> SessionRow { let cold = lock(&self.cold); @@ -168,7 +177,10 @@ impl SessionHandle { launch_commit: self.meta.launch_commit.clone(), launch_checkout_identity: self.meta.launch_checkout_identity.clone(), session_id: cold.session_id.clone(), - thread_name: cold.thread_name.clone(), + thread_name: cold + .index_thread_name + .clone() + .or_else(|| cold.thread_name.clone()), started_at: self.meta.started_at, last_output_at: self.last_output_at(), last_error: cold.last_error.clone(), diff --git a/src/tui/src/worker/pty/handle/types.rs b/src/tui/src/worker/pty/handle/types.rs index 22bb57329..d0da17e73 100644 --- a/src/tui/src/worker/pty/handle/types.rs +++ b/src/tui/src/worker/pty/handle/types.rs @@ -109,6 +109,15 @@ pub(crate) struct ColdFields { /// operator owns and may change, where provenance is a fact about how the /// session came to exist. `None` until somebody names it. pub(super) name: Option, + /// The thread name discovered from the harness provider own session + /// state rather than from the terminal stream. + /// + /// Codex persists renamed threads in its session index instead of emitting + /// an OSC window title, so the executor supplies this separately. When + /// set, it takes precedence over [[thread_name]] so that ordinary PTY + /// output (which may carry an empty title) cannot erase the index-backed + /// name. + pub(super) index_thread_name: Option, /// The non-empty terminal title last advertised by the harness. pub(super) thread_name: Option, /// Which repository, worktree, and branch the working directory sits in. diff --git a/src/tui/src/worker/pty/manager/labels.rs b/src/tui/src/worker/pty/manager/labels.rs new file mode 100644 index 000000000..1868d9f31 --- /dev/null +++ b/src/tui/src/worker/pty/manager/labels.rs @@ -0,0 +1,65 @@ +//! Keeping each session's Codex thread name current between turns. +//! +//! Codex persists a `/rename` in `session_index.jsonl` rather than emitting an +//! OSC window title, so the terminal stream never carries the new name. The +//! transcript executor reads that file once per located turn — which leaves a +//! session the operator is holding, has been handed back, or is merely +//! retained showing the old name until the next delegated turn happens to +//! start. A timer per live session fixes that at the only cost that matters: +//! one small file read on a thread. It is the Codex counterpart of the +//! [checkout poller](super::checkout) beside it. + +use std::collections::HashMap; +use std::sync::{Arc, Weak}; + +use super::super::handle::SessionHandle; +use super::PtyManager; + +/// How often a live session's index-backed thread name is re-read. +/// +/// Two seconds, like the checkout poller: a rename happens a handful of times +/// an hour, and re-reading a small JSONL file is cheap compared to the `git` +/// the checkout poller runs at the same rate. +const LABEL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(2); + +impl PtyManager { + /// Re-read `handle`'s Codex thread name on a timer for as long as that + /// session lives. + /// + /// Both references are non-owning: a poller cannot keep either the manager + /// or a reaped session alive. + /// + /// The environment is a snapshot taken from the `LaunchSpec`, not a borrow + /// of it: the poller outlives `open`, and a handful of variables cloned once + /// per Codex session is the same shape the executor's held env already is. + /// A session a turn has located is looked up by its harness session id; one + /// nothing has located — an operator-created session, which never passes + /// through the transcript executor — falls back to the newest Codex rollout + /// rooted at this session's cwd. + pub(super) fn spawn_codex_label_poller( + &self, + handle: Weak, + env: HashMap, + ) { + let inner = Arc::downgrade(&self.inner); + std::thread::spawn(move || loop { + std::thread::sleep(LABEL_INTERVAL); + let (Some(handle), true) = (handle.upgrade(), inner.upgrade().is_some()) else { + return; + }; + // An exited session's name is frozen at whatever it last was: the + // row is kept for the operator to read, and a dead session cannot be + // renamed. + if !handle.is_running() { + return; + } + let label = match handle.session_id() { + Some(session_id) => medulla::session_history::codex_thread_label(&env, &session_id), + None => medulla::session_history::codex_thread_label_for_cwd(&env, handle.cwd()), + }; + if let Some(label) = label { + handle.record_thread_name(label); + } + }); + } +} diff --git a/src/tui/src/worker/pty/manager/mod.rs b/src/tui/src/worker/pty/manager/mod.rs index 43584ede9..8939687d9 100644 --- a/src/tui/src/worker/pty/manager/mod.rs +++ b/src/tui/src/worker/pty/manager/mod.rs @@ -20,8 +20,8 @@ //! other caller reads, [`screen`] is the emulator surface the UI renders, //! [`attention`] keeps each row's "this harness wants you" flag current, //! [`checkout`] does the same for the repository, worktree, and branch it is -//! working in, and [`clipboard`] carries a harness's own copies out to the -//! operator's terminal. +//! working in, [`labels`] for a Codex session's renamed thread, and +//! [`clipboard`] carries a harness's own copies out to the operator's terminal. //! //! Both halves of the master run on **blocking threads**, not tokio tasks: //! `portable-pty` offers only synchronous `Read`/`Write`, and parking either on @@ -201,6 +201,7 @@ impl PtyManager { mod attention; mod checkout; mod clipboard; +mod labels; mod open; mod screen; mod session; diff --git a/src/tui/src/worker/pty/manager/open.rs b/src/tui/src/worker/pty/manager/open.rs index 819667716..781d263f5 100644 --- a/src/tui/src/worker/pty/manager/open.rs +++ b/src/tui/src/worker/pty/manager/open.rs @@ -6,6 +6,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::mpsc::{channel, Receiver}; use std::sync::{Arc, Weak}; +use medulla::protocol::HarnessProvider; use portable_pty::{native_pty_system, Child, CommandBuilder, PtyPair, PtySize}; use super::super::handle::{release_queued, SessionHandle, SessionMeta}; @@ -230,6 +231,14 @@ impl PtyManager { self.spawn_writer(Arc::downgrade(&handle), writer, queued, queued_bytes); self.spawn_attention_poller(Arc::downgrade(&handle)); self.spawn_checkout_poller(Arc::downgrade(&handle)); + // Codex persists `/rename` in a file it never says anything about on the + // wire, so the rail's only way to learn one while a session is held, + // idle, or retained is to re-read that file on a timer until the + // session dies. Only Codex has an index to watch; claude names arrive + // as OSC titles. + if spec.provider == HarnessProvider::Codex { + self.spawn_codex_label_poller(Arc::downgrade(&handle), spec.env.clone()); + } self.spawn_reader(handle, reader); Ok(id) diff --git a/src/tui/src/worker/pty/manager/session.rs b/src/tui/src/worker/pty/manager/session.rs index 9af715fb8..a098f99fd 100644 --- a/src/tui/src/worker/pty/manager/session.rs +++ b/src/tui/src/worker/pty/manager/session.rs @@ -192,6 +192,13 @@ impl PtyManager { } } + /// Record a harness thread name learned from provider-owned session state. + pub(in crate::worker) fn record_thread_name(&self, id: &str, thread_name: String) { + if let Some(session) = self.handle(id) { + session.record_thread_name(thread_name); + } + } + /// Ask a session's harness to exit, then reap it. /// /// Sends the child a kill rather than typing `/exit`: the harnesses disagree diff --git a/src/tui/src/worker/pty/tests/session.rs b/src/tui/src/worker/pty/tests/session.rs index aa0de5c5f..381614cf9 100644 --- a/src/tui/src/worker/pty/tests/session.rs +++ b/src/tui/src/worker/pty/tests/session.rs @@ -54,25 +54,33 @@ fn a_terminal_title_update_surfaces_as_the_thread_name() { } #[test] -fn clearing_a_terminal_title_clears_the_thread_name() { +fn an_empty_terminal_title_preserves_the_last_non_empty_thread_name() { + // The screen layer ignores empty OSC title samples so that a thread name + // discovered from the Codex session index (or from a prior non-empty title) + // is not erased by ordinary harness output that includes no title escape. let manager = PtyManager::new(); let id = manager - .open(sh( - "printf '\\033]2;Named thread\\007'; read line; printf '\\033]2; \\007'; sleep 30", - )) + .open(sh("printf '\\033]2;Named thread\\007'; read line; \ + printf '\\033]2;\\007'; echo empty-title-done; sleep 30")) .unwrap(); wait_for("initial thread name", || { manager.row(&id).and_then(|row| row.thread_name).as_deref() == Some("Named thread") }); manager.write(&id, b"clear\n").unwrap(); - wait_for("cleared thread name", || { - manager - .row(&id) - .is_some_and(|row| row.thread_name.is_none()) + // The empty title must NOT overwrite: the name stays. The child writes its + // empty OSC sequence *then* the marker, and the reader thread feeds bytes to + // the emulator in order — so a marker on screen proves the empty sample has + // already passed through `process` before the name is checked. (A fixed + // sleep raced the reader and made this test flaky.) + wait_for("the empty title to have been emitted", || { + screen_text(&manager, &id).contains("empty-title-done") }); - - assert_eq!(manager.row(&id).unwrap().thread_name, None); + assert_eq!( + manager.row(&id).and_then(|row| row.thread_name), + Some("Named thread".to_string()), + "empty OSC title must not clear a previously set thread name" + ); manager.close(&id); } diff --git a/src/tui/tests/e2e_codex_rename.rs b/src/tui/tests/e2e_codex_rename.rs new file mode 100644 index 000000000..a94e79d9f --- /dev/null +++ b/src/tui/tests/e2e_codex_rename.rs @@ -0,0 +1,150 @@ +//! End-to-end coverage for the Codex thread-rename poller. +//! +//! Codex persists a `/rename` in `session_index.jsonl` instead of the terminal +//! stream, and the transcript executor reads that file exactly once — on the +//! poll where the session is first located. A session that is then held, idle, +//! or retained never runs another turn, so without a second reader the rail +//! would sit on the old name until a new delegated turn happened to start. +//! +//! This drives the real PTY path and renames the session *after* its turn has +//! started: the executor's one-shot read is guaranteed to have already missed +//! the new name, so only the per-session poller can put it on the row. + +#![cfg(unix)] + +use std::collections::HashMap; +use std::fs; +use std::time::{Duration, Instant}; + +use medulla::daemon::providers::{Abort, RunTaskOptions}; +use medulla::protocol::HarnessProvider; +use medulla::sessions::SessionClass; +use medulla_tui::worker::executor::PtySessionExecutor; +use medulla_tui::worker::pty::PtyManager; + +/// Maximum time for a real child process or PTY reader to make progress. +const PATIENCE: Duration = Duration::from_secs(10); + +/// Spin until `check` passes or the end-to-end deadline expires. +async fn wait_for(what: &str, mut check: impl FnMut() -> bool) { + let deadline = Instant::now() + PATIENCE; + while Instant::now() < deadline { + if check() { + return; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("timed out waiting for: {what}"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_rename_after_the_turn_started_reaches_the_row_without_a_new_turn() { + let temp = tempfile::tempdir().expect("a temporary workspace"); + let workspace = temp.path().join("workspace"); + fs::create_dir_all(&workspace).unwrap(); + let sessions_dir = temp.path().join("codex").join("sessions"); + fs::create_dir_all(&sessions_dir).unwrap(); + let cwd = workspace.to_string_lossy().into_owned(); + let sessions_str = sessions_dir.to_string_lossy().into_owned(); + // `rollout-*.jsonl` — codex transcript discovery matches on that prefix. + let rollout = sessions_dir.join("rollout-rename.jsonl"); + let rollout = rollout.to_string_lossy().into_owned(); + + let script = format!( + r#" +printf 'codex ready\r\n' +read -r prompt +printf 'task started: %s\r\n' "$prompt" +printf '{{"type":"session_meta","payload":{{"session_id":"codex-rename-e2e","cwd":"{cwd}"}}}}\n' >> '{rollout}' +printf '{{"type":"event_msg","payload":{{"type":"task_started","turn_id":"turn-1"}}}}\n' >> '{rollout}' +while read -r line; do + printf 'peered: %s\r\n' "$line" +done +"# + ); + + // A stale index is planted before the session runs: the executor's one-shot + // label read can therefore never find the rename, whenever it happens. + // Only the poller that re-reads this file on a timer can. + let index = temp.path().join("codex").join("session_index.jsonl"); + fs::write( + &index, + r#"{"id":"some-other-session","thread_name":"untouched"}"#, + ) + .unwrap(); + + let mut env = HashMap::new(); + env.insert( + "PATH".to_string(), + std::env::var("PATH").unwrap_or_default(), + ); + env.insert("TERM".to_string(), "xterm-256color".to_string()); + env.insert("MEDULLA_CODEX_SESSIONS_DIR".to_string(), sessions_str); + env.insert("MEDULLA_CODEX_BIN".to_string(), "/bin/sh".to_string()); + + let sessions = PtyManager::new(); + let run_task = + PtySessionExecutor::new(sessions.clone(), env.clone(), cwd.clone()).into_run_task(); + let (session_tx, session_rx) = tokio::sync::oneshot::channel(); + let run = tokio::spawn((run_task)(RunTaskOptions { + origin: medulla::daemon::providers::RunTaskOrigin::DelegatedTask, + hooks: medulla::harness_hooks::HooksConfig::default(), + transport: Default::default(), + provider: HarnessProvider::Codex, + prompt: "start delegated work".to_string(), + cwd, + env, + timeout_ms: 30_000, + model: None, + agent: None, + extra_args: vec!["-c".to_string(), script], + skip_permissions: false, + conversation: "medulla-orchestrator".to_string(), + session_class: SessionClass::Bounded, + resume_session_id: None, + workspace_context: Default::default(), + abort: Abort::new(), + router: None, + attribution: false, + on_event: None, + on_stdin: None, + on_session: Some(Box::new(move |id| { + let _ = session_tx.send(id); + })), + on_workspace_context: None, + })); + + let id = tokio::time::timeout(PATIENCE, session_rx) + .await + .expect("the executor reports its PTY") + .expect("the session report channel stays open"); + + // The turn has started: this id is only recorded on the fold that locates + // the session, by which point the executor's one-shot label read has run. + wait_for("the Codex session to be located", || { + sessions + .row(&id) + .is_some_and(|row| row.session_id.as_deref() == Some("codex-rename-e2e")) + }) + .await; + + // Rename it *now*, after the executor's only read has passed. The poller is + // the sole remaining reader of the index. + fs::write( + &index, + r#"{"id":"codex-rename-e2e","thread_name":"Ship the sidebar"}"#, + ) + .unwrap(); + wait_for("the per-session poller to reach the renamed thread", || { + sessions + .row(&id) + .is_some_and(|row| row.thread_name.as_deref() == Some("ship-sidebar")) + }) + .await; + assert!( + !run.is_finished(), + "the rename was picked up by the poller, not by a second turn" + ); + + sessions.close(&id); +}