diff --git a/src/sdk/src/daemon/mappers/mod.rs b/src/sdk/src/daemon/mappers/mod.rs index 336b28781..51e11e4bb 100644 --- a/src/sdk/src/daemon/mappers/mod.rs +++ b/src/sdk/src/daemon/mappers/mod.rs @@ -33,7 +33,8 @@ mod usage; mod work; mod workspace; pub(crate) use workspace::{ - pull_request_command, workspace_event_from_output, PendingPullRequestCall, + pull_request_command, workspace_event_from_output, worktree_checkout_from_output, + PendingPullRequestCall, }; #[cfg(test)] diff --git a/src/sdk/src/daemon/mappers/workspace/mod.rs b/src/sdk/src/daemon/mappers/workspace/mod.rs index b893e9a9f..319323b67 100644 --- a/src/sdk/src/daemon/mappers/workspace/mod.rs +++ b/src/sdk/src/daemon/mappers/workspace/mod.rs @@ -35,7 +35,7 @@ pub(crate) fn workspace_event_from_output( ts: i64, record_type: &str, ) -> Option { - let checkout = json_report(output).or_else(|| text_report(output)); + let checkout = worktree_checkout_from_output(output); let pull_request = pull_request_command.and_then(|command| match command { PullRequestCommand::Create => pull_request_url(output), PullRequestCommand::View => pull_request_url_from_json(output), @@ -68,6 +68,15 @@ pub(crate) fn workspace_event_from_output( )) } +/// Read a stable worktree-helper report from command output. +/// +/// Shared with transports that do not use the JSONL mapper (notably Codex +/// app-server), so every Codex execution path applies the same signature checks +/// before changing a session's runtime directory. +pub(crate) fn worktree_checkout_from_output(output: &str) -> Option<(String, String)> { + json_report(output).or_else(|| text_report(output)) +} + /// Read the `--json` report, allowing command output around the object. fn json_report(output: &str) -> Option<(String, String)> { output.match_indices('{').find_map(|(start, _)| { diff --git a/src/sdk/src/daemon/providers/codex_server/README.md b/src/sdk/src/daemon/providers/codex_server/README.md index 980b52f83..c82df16b3 100644 --- a/src/sdk/src/daemon/providers/codex_server/README.md +++ b/src/sdk/src/daemon/providers/codex_server/README.md @@ -15,7 +15,7 @@ Selected by naming the `codex-server` harness, which reaches here as `HarnessTra ## What this reports -Deliberately minimal: lifecycle status, the assistant's messages, and token usage. The app-server reports far more — per-item reasoning deltas, command output streams, patch previews — and the CLI transport's mappers turn the equivalent into the rich agent-rail detail an operator watches. +Deliberately minimal: lifecycle status, the assistant's messages, token usage, and stable worktree reports that determine where a resumed turn executes. The app-server reports far more — per-item reasoning deltas, command output streams, patch previews — and the CLI transport's mappers turn the equivalent into the rich agent-rail detail an operator watches. Reproducing that surface here would mean a second implementation of every mapper, tracking a wire format still marked experimental, for a transport chosen when throughput is what matters. So a `codex-server` lane reports that it is working, what it finally said, and what it cost — and an operator who wants to watch a lane work runs it on `codex`. diff --git a/src/sdk/src/daemon/providers/codex_server/execution.rs b/src/sdk/src/daemon/providers/codex_server/execution.rs index bff89add7..f9a85f40d 100644 --- a/src/sdk/src/daemon/providers/codex_server/execution.rs +++ b/src/sdk/src/daemon/providers/codex_server/execution.rs @@ -88,11 +88,14 @@ async fn run(options: RunTaskOptions) -> Result { &connection, &mut subscription, &thread_id, + thread.cwd, options.prompt, options.model, options.timeout_ms, options.abort, options.on_event, + options.workspace_context, + options.on_workspace_context, ) .await?; @@ -212,16 +215,24 @@ async fn drive_turn( connection: &Connection, subscription: &mut ThreadSubscription, thread_id: &str, + cwd: String, prompt: String, model: Option, timeout_ms: u64, abort: super::super::types::Abort, on_event: Option, + workspace_context: crate::sessions::WorkspaceContext, + on_workspace_context: Option, ) -> Result { // Shared because the idle watchdog reads the fold's last-activity stamp // while the notification branch writes to it, and both live in one // `select!`. - let fold = Arc::new(Mutex::new(FoldState::new(on_event))); + let fold = Arc::new(Mutex::new(FoldState::with_workspace_at( + on_event, + workspace_context, + on_workspace_context, + Some(cwd.clone().into()), + ))); let timeout = Duration::from_millis(timeout_ms); // Tracked from `turn/started` so an interrupt can name the turn it stops; // the protocol requires both ids. @@ -231,6 +242,11 @@ async fn drive_turn( "turn/start", json!({ "threadId": thread_id, + // Codex treats cwd as sticky turn state. Send it even after + // `thread/resume`: otherwise a retained worktree exists only in + // Medulla's metadata while built-in tools default to the original + // directory recorded by the thread. + "cwd": cwd, "input": [{ "type": "text", "text": prompt }], "model": model, }), diff --git a/src/sdk/src/daemon/providers/codex_server/fold.rs b/src/sdk/src/daemon/providers/codex_server/fold.rs index 0b05eeb9a..97c72db28 100644 --- a/src/sdk/src/daemon/providers/codex_server/fold.rs +++ b/src/sdk/src/daemon/providers/codex_server/fold.rs @@ -2,10 +2,11 @@ //! //! # Scope //! -//! Deliberately minimal: lifecycle status, the assistant's messages, and token -//! usage. The app-server reports far more than that — per-item reasoning deltas, -//! command output streams, patch previews — and the CLI transport's mappers turn -//! the equivalent into the rich agent-rail detail an operator watches. +//! Deliberately minimal: lifecycle status, the assistant's messages, token +//! usage, and repository moves that affect where the next turn executes. The +//! app-server reports far more than that — per-item reasoning deltas, command +//! output streams, patch previews — and the CLI transport's mappers turn the +//! equivalent into the rich agent-rail detail an operator watches. //! //! Reproducing that surface here would mean a second implementation of every //! mapper, tracking a wire format that is still marked experimental, for a @@ -17,15 +18,18 @@ //! notification counts as activity, including the ones that produce no event, or //! a long silent command would look like a dead process. +use std::path::{Path, PathBuf}; +use std::process::Command; use std::time::Instant; use serde_json::{json, Value}; use crate::codex_app_server::Notification; -use crate::daemon::mappers::HarnessSemanticEvent; +use crate::daemon::mappers::{worktree_checkout_from_output, HarnessSemanticEvent}; use crate::protocol::{HarnessEvent, TokenUsage}; +use crate::sessions::WorkspaceContext; -use super::super::types::OnEvent; +use super::super::types::{OnEvent, OnWorkspaceContext}; /// What a finished fold reports, without the callback it folded through. #[derive(Debug, Clone, Default)] @@ -58,6 +62,12 @@ pub(super) struct FoldState { pub(super) last_activity: Instant, /// Per-event status callback. on_event: Option, + /// Repository position retained across turns of this thread. + workspace_context: WorkspaceContext, + /// Persists a newly detected worktree for the next resumed turn. + on_workspace_context: Option, + /// Source used to enumerate the repository's registered worktrees. + worktree_registry: WorktreeRegistry, /// Line counter standing in for the CLI transport's transcript offsets. /// /// There is no transcript here, but `HarnessSemanticEvent::line` is the @@ -68,7 +78,18 @@ pub(super) struct FoldState { impl FoldState { /// A fold ready for one turn. + #[cfg(test)] pub(super) fn new(on_event: Option) -> Self { + Self::with_workspace_at(on_event, WorkspaceContext::default(), None, None) + } + + /// A fold seeded with workspace state and its configured checkout. + pub(super) fn with_workspace_at( + on_event: Option, + workspace_context: WorkspaceContext, + on_workspace_context: Option, + repository_cwd: Option, + ) -> Self { Self { reply: String::new(), items: 0, @@ -76,10 +97,29 @@ impl FoldState { error: None, last_activity: Instant::now(), on_event, + workspace_context, + on_workspace_context, + worktree_registry: repository_cwd + .map(WorktreeRegistry::Git) + .unwrap_or(WorktreeRegistry::Disabled), line: 0, } } + /// A fold backed by a deterministic in-process worktree registry. + #[cfg(test)] + pub(super) fn with_registered_worktrees( + on_event: Option, + workspace_context: WorkspaceContext, + on_workspace_context: Option, + worktrees: Vec<(PathBuf, String)>, + ) -> Self { + let mut fold = + Self::with_workspace_at(on_event, workspace_context, on_workspace_context, None); + fold.worktree_registry = WorktreeRegistry::Static(worktrees); + fold + } + /// Fold one notification, emitting whatever events it implies. /// /// Returns `true` once the turn is terminal, which is the caller's signal to @@ -108,6 +148,7 @@ impl FoldState { self.emit("item/completed", "agent_message", json!({ "text": text })); } } + self.capture_worktree(item); false } "thread/tokenUsage/updated" => { @@ -188,6 +229,116 @@ impl FoldState { on_event(&event); } } + + /// Persist a stable worktree report carried by a completed command item. + fn capture_worktree(&mut self, item: Option<&Value>) { + let Some(item) = item + .filter(|item| item.get("type").and_then(Value::as_str) == Some("commandExecution")) + else { + return; + }; + if !successful_worktree_command(item) { + return; + } + let output = item + .get("aggregatedOutput") + .or_else(|| item.get("aggregated_output")) + .and_then(Value::as_str) + .unwrap_or(""); + let Some((cwd, branch)) = worktree_checkout_from_output(output) else { + return; + }; + if !self.is_registered_worktree(&cwd, &branch) { + return; + } + if self.workspace_context.cwd.as_deref() != Some(&cwd) + || self.workspace_context.branch.as_deref() != Some(&branch) + { + self.workspace_context.pull_request = None; + } + self.workspace_context.cwd = Some(cwd); + self.workspace_context.branch = Some(branch); + self.emit( + "item/completed:workspace", + crate::harness_work::kinds::SESSION_INFO, + json!({ + "cwd": self.workspace_context.cwd, + "branch": self.workspace_context.branch, + }), + ); + if let Some(callback) = self.on_workspace_context.as_ref() { + callback(self.workspace_context.clone()); + } + } + + /// Accept only a report for a successful helper invocation that Git says is + /// an existing worktree of this repository on the reported branch. + fn is_registered_worktree(&self, cwd: &str, branch: &str) -> bool { + let Ok(cwd) = std::fs::canonicalize(cwd) else { + return false; + }; + let worktrees = match &self.worktree_registry { + WorktreeRegistry::Disabled => return false, + WorktreeRegistry::Git(repository_cwd) => { + let Ok(output) = Command::new("git") + .args(["-C"]) + .arg(repository_cwd) + .args(["worktree", "list", "--porcelain"]) + .output() + else { + return false; + }; + if !output.status.success() { + return false; + } + registered_worktrees(&String::from_utf8_lossy(&output.stdout)) + } + #[cfg(test)] + WorktreeRegistry::Static(worktrees) => worktrees.clone(), + }; + worktrees + .into_iter() + .any(|(path, registered_branch)| path == cwd && registered_branch == branch) + } +} + +/// Where worktree membership is read from. +enum WorktreeRegistry { + /// No repository was configured, so no report may update the workspace. + Disabled, + /// Ask Git for the live registry rooted at this checkout. + Git(PathBuf), + /// Deterministic registry used by unit tests. + #[cfg(test)] + Static(Vec<(PathBuf, String)>), +} + +/// A command item may affect retained cwd only when the worktree helper itself +/// completed successfully. Generic shell output is untrusted text. +fn successful_worktree_command(item: &Value) -> bool { + item.get("exitCode").and_then(Value::as_i64) == Some(0) + && item + .get("command") + .and_then(Value::as_str) + .and_then(|command| command.split_whitespace().next()) + == Some("worktree") +} + +/// Parse Git's porcelain worktree listing into canonical checkout/branch pairs. +fn registered_worktrees(output: &str) -> Vec<(PathBuf, String)> { + output + .split("\n\n") + .filter_map(|entry| { + let path = entry.strip_prefix("worktree ")?.lines().next()?; + let branch = entry + .lines() + .find_map(|line| line.strip_prefix("branch refs/heads/"))?; + Some(( + std::fs::canonicalize(Path::new(path)).ok()?, + branch.to_string(), + )) + }) + .collect() } /// A `status` payload saying the lane is working, with a one-line detail. diff --git a/src/sdk/src/daemon/providers/codex_server/tests.rs b/src/sdk/src/daemon/providers/codex_server/tests.rs index 2f7aebd7d..02b856458 100644 --- a/src/sdk/src/daemon/providers/codex_server/tests.rs +++ b/src/sdk/src/daemon/providers/codex_server/tests.rs @@ -1,6 +1,7 @@ //! Unit tests for transport selection and the notification fold. use std::collections::HashMap; +use std::path::PathBuf; use std::sync::{Arc, Mutex}; use serde_json::json; @@ -8,12 +9,19 @@ use serde_json::json; use crate::codex_app_server::Notification; use crate::daemon::mappers::HarnessSemanticEvent; use crate::protocol::{HarnessProvider, HarnessTransport}; -use crate::sessions::SessionClass; +use crate::sessions::{SessionClass, WorkspaceContext}; use super::super::types::{Abort, RunTaskOptions}; use super::execution::{child_env, uses_app_server, HARNESS_TRANSPORT_ENV}; use super::fold::FoldState; +/// Test fold plus its recorded workspace callbacks and semantic events. +type WorkspaceRecording = ( + FoldState, + Arc>>, + Arc>>, +); + /// Run options carrying a transport and an environment, and nothing else that /// matters to the two functions under test here. fn options(transport: HarnessTransport, env: &[(&str, &str)]) -> RunTaskOptions { @@ -65,6 +73,56 @@ fn recording_fold() -> (FoldState, Arc>>) { (fold, seen) } +/// A completed worktree helper item carrying the supplied checkout report. +fn worktree_notification(path: &str, branch: &str, exit_code: i64) -> Notification { + let report = json!({ + "status": "ready", + "repository": "/repo", + "path": path, + "branch": branch, + "head": "abc", + "headShort": "abc", + "created": true, + "submodules": { + "state": "initialized_recursive", + "count": 0, + }, + "nextCommand": format!("cd {path}"), + }); + notification( + "item/completed", + json!({ + "item": { + "type": "commandExecution", + "command": "worktree fix-context --json", + "aggregatedOutput": report.to_string(), + "exitCode": exit_code + } + }), + ) +} + +/// A fold whose workspace callback records accepted reports. +fn workspace_fold(worktrees: Vec<(PathBuf, String)>) -> WorkspaceRecording { + let seen = Arc::new(Mutex::new(Vec::new())); + let sink = seen.clone(); + let events = Arc::new(Mutex::new(Vec::new())); + let event_sink = events.clone(); + let worktrees = worktrees + .into_iter() + .map(|(path, branch)| (std::fs::canonicalize(path).unwrap(), branch)) + .collect(); + let fold = FoldState::with_registered_worktrees( + Some(Box::new(move |event| { + event_sink.lock().unwrap().push(event.clone()) + })), + WorkspaceContext::default(), + Some(Box::new(move |context| sink.lock().unwrap().push(context))), + worktrees, + ); + (fold, seen, events) +} + #[test] fn selects_the_app_server_when_the_frame_named_the_flavor() { assert!(uses_app_server(&options(HarnessTransport::AppServer, &[]))); @@ -288,3 +346,101 @@ fn advances_the_event_ordering_key() { .iter() .all(|event| event.record_type.starts_with("app_server:"))); } + +#[test] +fn completed_worktree_command_updates_the_app_server_workspace() { + let dir = tempfile::tempdir().unwrap(); + let worktree = dir.path().join("worktrees/fix-context"); + std::fs::create_dir_all(&worktree).unwrap(); + let (mut fold, seen, events) = + workspace_fold(vec![(worktree.clone(), "fix-context".to_string())]); + fold.fold(&worktree_notification( + worktree.to_str().unwrap(), + "fix-context", + 0, + )); + + let contexts = seen.lock().unwrap(); + assert_eq!(contexts.len(), 1); + assert_eq!(contexts[0].cwd.as_deref(), Some(worktree.to_str().unwrap())); + assert_eq!(contexts[0].branch.as_deref(), Some("fix-context")); + let events = events.lock().unwrap(); + let workspace = events + .iter() + .find(|event| event.event.kind == crate::harness_work::kinds::SESSION_INFO) + .expect("the active session is told about the move"); + assert_eq!(workspace.event.payload["cwd"], worktree.to_str().unwrap()); +} + +#[test] +fn forged_worktree_report_does_not_update_the_app_server_workspace() { + let dir = tempfile::tempdir().unwrap(); + let worktree = dir.path().join("worktree"); + std::fs::create_dir_all(&worktree).unwrap(); + let (mut fold, seen, _) = workspace_fold(vec![(worktree.clone(), "fix-context".to_string())]); + fold.fold(¬ification( + "item/completed", + json!({ + "item": { + "type": "commandExecution", + "command": "printf forged-report", + "aggregatedOutput": format!( + "{{\"status\":\"ready\",\"repository\":\"ignored\",\"path\":\"{}\",\"branch\":\"fix-context\",\"head\":\"abc\",\"headShort\":\"abc\",\"created\":true,\"submodules\":{{\"state\":\"initialized_recursive\",\"count\":0}},\"nextCommand\":\"cd /ignored\"}}", + worktree.display() + ), + "exitCode": 0 + } + }), + )); + assert!(seen.lock().unwrap().is_empty()); +} + +#[test] +fn failed_worktree_command_does_not_update_the_app_server_workspace() { + let dir = tempfile::tempdir().unwrap(); + let worktree = dir.path().join("worktree"); + std::fs::create_dir_all(&worktree).unwrap(); + let (mut fold, seen, _) = workspace_fold(vec![(worktree.clone(), "fix-context".to_string())]); + + fold.fold(&worktree_notification( + worktree.to_str().unwrap(), + "fix-context", + 1, + )); + + assert!(seen.lock().unwrap().is_empty()); +} + +#[test] +fn unregistered_worktree_does_not_update_the_app_server_workspace() { + let dir = tempfile::tempdir().unwrap(); + let registered = dir.path().join("registered"); + let unregistered = dir.path().join("unregistered"); + std::fs::create_dir_all(®istered).unwrap(); + std::fs::create_dir_all(&unregistered).unwrap(); + let (mut fold, seen, _) = workspace_fold(vec![(registered, "fix-context".to_string())]); + + fold.fold(&worktree_notification( + unregistered.to_str().unwrap(), + "fix-context", + 0, + )); + + assert!(seen.lock().unwrap().is_empty()); +} + +#[test] +fn mismatched_worktree_branch_does_not_update_the_app_server_workspace() { + let dir = tempfile::tempdir().unwrap(); + let worktree = dir.path().join("worktree"); + std::fs::create_dir_all(&worktree).unwrap(); + let (mut fold, seen, _) = workspace_fold(vec![(worktree.clone(), "other-branch".to_string())]); + + fold.fold(&worktree_notification( + worktree.to_str().unwrap(), + "fix-context", + 0, + )); + + assert!(seen.lock().unwrap().is_empty()); +} diff --git a/src/sdk/src/daemon/providers/execute.rs b/src/sdk/src/daemon/providers/execute.rs index c9ecbe4a8..f35fc058a 100644 --- a/src/sdk/src/daemon/providers/execute.rs +++ b/src/sdk/src/daemon/providers/execute.rs @@ -135,11 +135,16 @@ pub fn with_auth_hint(message: &str) -> String { /// Run one delegated task headlessly, retrying transient opencode SQLite-lock /// exits with jittered exponential backoff. pub async fn run_provider_task(mut options: RunTaskOptions) -> Result { - // Ahead of everything, including the credential scrub below. Every line - // after this one prepares a *child process* — its environment, its router, - // its argv — and OpenHuman has no child: the turn runs in this process - // against the embedded core. Scrubbing the core's own workspace out of the - // environment on its way to the core would be exactly backwards. + // A resumed conversation may have created and moved into a linked worktree + // during an earlier turn. The mapper persists that move separately from the + // daemon's configured launch checkout; make it the runtime cwd before any + // transport consumes the options, or Codex's next turn and all of its + // defaulted tool calls silently snap back to where the daemon started. + options.cwd = effective_cwd(&options.cwd, &options.workspace_context); + // Ahead of all child-process preparation, including the credential scrub + // below. OpenHuman has no child: the turn runs in this process against the + // embedded core. Scrubbing the core's own workspace out of the environment + // on its way to the core would be exactly backwards. if super::openhuman::uses_embedded_core(&options) { return super::openhuman::run_openhuman_task(options).await; } @@ -225,6 +230,24 @@ pub async fn run_provider_task(mut options: RunTaskOptions) -> Result String { + workspace_context + .cwd + .as_deref() + .filter(|cwd| std::path::Path::new(cwd).is_dir()) + .unwrap_or(configured) + .to_string() +} + /// A cheap uniform-ish `[0,1)` sample (no `rand` dep): folds the wall clock. pub(super) fn rand_unit() -> f64 { let nanos = std::time::SystemTime::now() diff --git a/src/sdk/src/daemon/providers/tests.rs b/src/sdk/src/daemon/providers/tests.rs index a2cf36c08..1eae63390 100644 --- a/src/sdk/src/daemon/providers/tests.rs +++ b/src/sdk/src/daemon/providers/tests.rs @@ -8,6 +8,7 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use crate::protocol::HarnessProvider; +use crate::sessions::WorkspaceContext; use super::detect::{ build_run_args, detect_providers, make_path_lookup, provider_bin, provider_name, @@ -137,6 +138,39 @@ fn build_run_args_per_provider() { ); } +#[test] +fn a_retained_worktree_becomes_the_resumed_runs_working_directory() { + let launch = tempfile::tempdir().unwrap(); + let worktree = tempfile::tempdir().unwrap(); + let context = WorkspaceContext { + cwd: Some(worktree.path().to_string_lossy().into_owned()), + branch: Some("feature".into()), + pull_request: None, + }; + + assert_eq!( + super::execute::effective_cwd(launch.path().to_str().unwrap(), &context), + worktree.path().to_string_lossy(), + ); +} + +#[test] +fn a_removed_retained_worktree_falls_back_to_the_configured_workspace() { + let launch = tempfile::tempdir().unwrap(); + let removed = tempfile::tempdir().unwrap(); + let context = WorkspaceContext { + cwd: Some(removed.path().to_string_lossy().into_owned()), + branch: Some("gone".into()), + pull_request: None, + }; + drop(removed); + + assert_eq!( + super::execute::effective_cwd(launch.path().to_str().unwrap(), &context), + launch.path().to_string_lossy(), + ); +} + #[test] fn build_run_args_neutralizes_dash_prompt() { let args = build_run_args(HarnessProvider::Codex, "-rf /", None, None, &[], false); diff --git a/src/sdk/tests/e2e_codex_app_server.rs b/src/sdk/tests/e2e_codex_app_server.rs index e0f64895b..e67ef08a3 100644 --- a/src/sdk/tests/e2e_codex_app_server.rs +++ b/src/sdk/tests/e2e_codex_app_server.rs @@ -109,6 +109,35 @@ async fn runs_a_turn_and_reports_the_reply() { assert!(kinds.contains(&"status"), "{kinds:?}"); } +#[tokio::test] +async fn resumed_turn_uses_the_worktree_as_its_runtime_cwd() { + let dir = TempDir::new(); + let worktree = dir.path().join("worktree"); + std::fs::create_dir_all(&worktree).unwrap(); + let fake = fake_app_server(&dir, TurnScript::Reply("ok")); + let (mut options, _) = options(&fake, &home(&dir, "cwd"), "continue", 10_000); + options.cwd = dir.path().to_string_lossy().into_owned(); + options.workspace_context.cwd = Some(worktree.to_string_lossy().into_owned()); + options.resume_session_id = Some("thread-retained".to_string()); + + run_provider_task(options).await.expect("the turn runs"); + + let requests = fake.requests(); + let resumed = requests + .iter() + .position(|request| request["method"] == "thread/resume") + .expect("thread/resume request"); + let turn = requests + .iter() + .position(|request| request["method"] == "turn/start") + .expect("turn/start request"); + assert!(resumed < turn, "thread/resume must precede turn/start"); + assert_eq!( + requests[turn]["params"]["cwd"], + worktree.to_string_lossy().as_ref() + ); +} + /// The whole point: several tasks, one process. #[tokio::test] async fn shares_one_process_across_sequential_tasks() {