Stop orphaned commands, fix relative paths on macOS, and close IPC/hook gaps

README.md

@@ -490,11 +490,11 @@ done
 | `system/ready`, `system/session_start` | After `initialize`: the session id, the agent, the model, the tools |
 | `user`, `delta`, `assistant`, `tool_use`, `tool_result` | As the turn runs; `assistant` carries `usage` and `costUsd` |
 | `callback_request` (`kind`: `hook`, `tool` or `approval`) | The binary needs your answer; reply with `callback_response` and the same `id` |
-| `system/turn_end` | A turn finished, with its `outcome` and usage |
+| `system/turn_end` | A turn finished, with its `outcome` and usage (sub-agents' model calls included) |
 | `system/warning`, `system/error`, `system/hook_blocked` | Something to show; the session goes on unless the error was fatal |
-| `system/session_end` | Last message, with the session's costs |
+| `system/session_end` | Last message, with the session's costs (sub-agents' included) |
 
-With `resume: true` the conversation is stored after every turn under `.gitagent/sessions/<sessionId>/`, and a new process started with the same `sessionId` continues it.
+With `resume: true` the conversation is stored after every turn under `.gitagent/sessions/<sessionId>/`, and a new process started with the same `sessionId` continues it. One process at a time holds a stored session: a second one opened while the first runs gets `system/error` and stops.
 
 ### Rust
 

crates/cli/src/ipc.rs

@@ -30,11 +30,12 @@ use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use gitagent_core::{
-    Approval, ApprovalRequest, Approver, ExternalTools, Outcome, Resume, RunOptions, Steering,
+    Approval, ApprovalRequest, Approver, ExternalTools, Outcome, Resume, RunOptions, RunReport,
+    Steering,
 };
 use gitagent_hooks::{Decision, ToolHooks};
 use gitagent_loader::v1;
-use gitagent_provider::{Message, ToolSpec};
+use gitagent_provider::{Message, ToolSpec, Usage};
 use gitagent_record::Events;
 use serde_json::{json, Map, Value};
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
@@ -401,6 +402,47 @@ struct Totals {
 }
 
 impl Totals {
+    /// `usage` from `requests` model calls, priced at `prices`.
+    fn priced(prices: Option<gitagent_sdk::Prices>, usage: &Usage, requests: u64) -> Totals {
+        Totals {
+            input: usage.input_tokens,
+            output: usage.output_tokens,
+            cache_read: usage.cache_read_tokens,
+            cache_write: usage.cache_write_tokens,
+            cost: gitagent_sdk::cost_of(
+                prices,
+                &gitagent_sdk::Usage {
+                    input_tokens: usage.input_tokens,
+                    output_tokens: usage.output_tokens,
+                    cache_read_input_tokens: usage.cache_read_tokens,
+                    cache_creation_input_tokens: usage.cache_write_tokens,
+                },
+            ),
+            requests,
+        }
+    }
+
+    fn add(&mut self, other: &Totals) {
+        self.input += other.input;
+        self.output += other.output;
+        self.cache_read += other.cache_read;
+        self.cache_write += other.cache_write;
+        self.cost += other.cost;
+        self.requests += other.requests;
+    }
+
+    /// What was added since `before`.
+    fn since(&self, before: &Totals) -> Totals {
+        Totals {
+            input: self.input - before.input,
+            output: self.output - before.output,
+            cache_read: self.cache_read - before.cache_read,
+            cache_write: self.cache_write - before.cache_write,
+            cost: self.cost - before.cost,
+            requests: self.requests - before.requests,
+        }
+    }
+
     fn json(&self) -> Value {
         json!({
             "inputTokens": self.input,
@@ -420,10 +462,31 @@ struct Stream {
     cb: Arc<Callbacks>,
     provider: String,
     prices: Option<gitagent_sdk::Prices>,
-    totals: Mutex<Totals>,
+    /// The model calls streamed so far: the parent's only, since sub-agents
+    /// run without events.
+    streamed: Mutex<Totals>,
 }
 
 impl Stream {
+    fn streamed(&self) -> Totals {
+        *self.streamed.lock().unwrap_or_else(|e| e.into_inner())
+    }
+
+    /// What a run used. A finished run's report has every model call,
+    /// its sub-agents' included; without one, only the streamed calls since
+    /// `before` are known.
+    fn run_usage(&self, report: Option<&RunReport>, before: &Totals) -> Totals {
+        let streamed = self.streamed().since(before);
+        match report {
+            Some(r) => Totals::priced(
+                self.prices,
+                &r.usage,
+                streamed.requests + r.counters.get("subagent_turns"),
+            ),
+            None => streamed,
+        }
+    }
+
     fn feed(&self, e: &Value) {
         let kind = e["event"].as_str().unwrap_or_default();
         match kind {
@@ -472,31 +535,17 @@ impl Stream {
         }
         let u = &e["usage"];
         let n = |k: &str| u[k].as_u64().unwrap_or(0);
-        let turn = Totals {
-            input: n("input_tokens"),
-            output: n("output_tokens"),
-            cache_read: n("cache_read_tokens"),
-            cache_write: n("cache_write_tokens"),
-            cost: gitagent_sdk::cost_of(
-                self.prices,
-                &gitagent_sdk::Usage {
-                    input_tokens: n("input_tokens"),
-                    output_tokens: n("output_tokens"),
-                    cache_read_input_tokens: n("cache_read_tokens"),
-                    cache_creation_input_tokens: n("cache_write_tokens"),
-                },
-            ),
-            requests: 1,
+        let usage = Usage {
+            input_tokens: n("input_tokens"),
+            output_tokens: n("output_tokens"),
+            cache_read_tokens: n("cache_read_tokens"),
+            cache_write_tokens: n("cache_write_tokens"),
         };
-        {
-            let mut t = self.totals.lock().unwrap_or_else(|e| e.into_inner());
-            t.input += turn.input;
-            t.output += turn.output;
-            t.cache_read += turn.cache_read;
-            t.cache_write += turn.cache_write;
-            t.cost += turn.cost;
-            t.requests += 1;
-        }
+        let turn = Totals::priced(self.prices, &usage, 1);
+        self.streamed
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .add(&turn);
         let (stop, error) = match (
             e["tool_calls"].as_u64().unwrap_or(0),
             e["stop_reason"].as_str(),
@@ -613,6 +662,31 @@ fn session_file(dir: &Path, id: &str) -> Result<PathBuf, String> {
         .join("messages.json"))
 }
 
+/// Locks the stored session `id` for this process (an advisory lock on a
+/// `lock` file next to the conversation). The lock lasts as long as the
+/// returned file is open; the system drops it when the process ends, so a
+/// crashed process leaves nothing to clean up.
+fn lock_session(store: &Path, id: &str) -> Result<std::fs::File, String> {
+    let path = store.with_file_name("lock");
+    let cannot = |e: std::io::Error| format!("the session lock {} failed: {e}", path.display());
+    if let Some(dir) = path.parent() {
+        std::fs::create_dir_all(dir).map_err(cannot)?;
+    }
+    let file = std::fs::OpenOptions::new()
+        .create(true)
+        .truncate(false)
+        .write(true)
+        .open(&path)
+        .map_err(cannot)?;
+    match file.try_lock() {
+        Ok(()) => Ok(file),
+        Err(std::fs::TryLockError::WouldBlock) => Err(format!(
+            "session {id:?} is open in another process; end that one first or use another sessionId"
+        )),
+        Err(std::fs::TryLockError::Error(e)) => Err(cannot(e)),
+    }
+}
+
 pub(crate) async fn run(cli: Cli) -> ! {
     let (tx, mut rx) = mpsc::unbounded_channel::<Value>();
     let out = Out(tx);
@@ -630,15 +704,39 @@ pub(crate) async fn run(cli: Cli) -> ! {
             let _ = stdout.flush().await;
         }
     });
-    // `session` drops its shells and MCP servers before returning.
-    let code = session(cli, out.clone()).await;
+    // `session` drops its shells and MCP servers before returning. A stop
+    // signal drops it too, so the commands the shells run do not outlive us.
+    let code = tokio::select! {
+        code = session(cli, out.clone()) => code,
+        signal = stop_signal() => 128 + signal,
+    };
     out.send(Value::Null);
     let _ = writer.await;
     // The stdin reader may still be waiting in a blocking read the runtime
     // would wait for (a client may keep its end open after `session_end`).
     std::process::exit(code)
 }
 
+/// Waits for SIGTERM or SIGINT and returns its number.
+async fn stop_signal() -> i32 {
+    use tokio::signal::unix::{signal, SignalKind};
+    let wait_for = |kind: SignalKind| async move {
+        match signal(kind) {
+            Ok(mut stream) => {
+                stream.recv().await;
+                kind.as_raw_value()
+            }
+            // No handler was installed, so the signal keeps its default
+            // action (ending the process) and this branch never fires.
+            Err(_) => std::future::pending().await,
+        }
+    };
+    tokio::select! {
+        number = wait_for(SignalKind::terminate()) => number,
+        number = wait_for(SignalKind::interrupt()) => number,
+    }
+}
+
 async fn session(cli: Cli, out: Out) -> i32 {
     let mut lines = BufReader::new(tokio::io::stdin()).lines();
     let fatal = |out: &Out, why: String| {
@@ -772,6 +870,16 @@ async fn session(cli: Cli, out: Out) -> i32 {
     } else {
         None
     };
+    // Held until the process ends, so a second process cannot load and
+    // overwrite the same stored conversation.
+    let _session_lock = match store
+        .as_deref()
+        .map(|p| lock_session(p, &session_id))
+        .transpose()
+    {
+        Ok(lock) => lock,
+        Err(e) => return fatal(&out, e),
+    };
     let agent_name = s
         .spec
         .agent
@@ -803,7 +911,7 @@ async fn session(cli: Cli, out: Out) -> i32 {
         cb: cb.clone(),
         provider: s.spec.model.provider.clone(),
         prices,
-        totals: Mutex::new(Totals::default()),
+        streamed: Mutex::new(Totals::default()),
     });
 
     let (mcp, warnings) = gitagent_core::connect_mcp(&s.spec).await;
@@ -956,6 +1064,8 @@ async fn session(cli: Cli, out: Out) -> i32 {
         .has("approval")
         .then(|| Arc::new(ClientApprover(cb.clone())) as Arc<dyn Approver>);
 
+    // Every model call of the session, sub-agents' included.
+    let mut session_usage = Totals::default();
     let mut queue: VecDeque<String> = VecDeque::new();
     let mut ending = false;
     let mut n = 0;
@@ -994,7 +1104,7 @@ async fn session(cli: Cli, out: Out) -> i32 {
             let stream = stream.clone();
             Events::to_callback(move |e| stream.feed(&e))
         };
-        let before = *stream.totals.lock().unwrap_or_else(|e| e.into_inner());
+        let before = stream.streamed();
         let opts = RunOptions {
             instruction: content,
             work_dir: s.work_dir.clone(),
@@ -1040,15 +1150,12 @@ async fn session(cli: Cli, out: Out) -> i32 {
         for late in ctl.steering.take().into_iter().rev() {
             queue.push_front(late);
         }
-        let after = *stream.totals.lock().unwrap_or_else(|e| e.into_inner());
-        let turn_usage = Totals {
-            input: after.input - before.input,
-            output: after.output - before.output,
-            cache_read: after.cache_read - before.cache_read,
-            cache_write: after.cache_write - before.cache_write,
-            cost: after.cost - before.cost,
-            requests: after.requests - before.requests,
-        };
+        let report = result
+            .as_ref()
+            .and_then(|r| r.as_ref().ok())
+            .map(|f| &f.report);
+        let turn_usage = stream.run_usage(report, &before);
+        session_usage.add(&turn_usage);
         match result {
             None => {
                 out.system(
@@ -1145,7 +1252,14 @@ async fn session(cli: Cli, out: Out) -> i32 {
             external_tools: external.clone(),
             throttle: None,
         };
-        match gitagent_core::teardown(&s.spec, s.provider.as_ref(), opts).await {
+        let before = stream.streamed();
+        let done = gitagent_core::teardown(&s.spec, s.provider.as_ref(), opts).await;
+        let report = done
+            .as_ref()
+            .and_then(|r| r.as_ref().ok())
+            .map(|f| &f.report);
+        session_usage.add(&stream.run_usage(report, &before));
+        match done {
             Some(Ok(done)) => out.system(
                 "teardown",
                 done.report.summary.clone().unwrap_or_default(),
@@ -1174,8 +1288,7 @@ async fn session(cli: Cli, out: Out) -> i32 {
             }
         }
     }
-    let totals = *stream.totals.lock().unwrap_or_else(|e| e.into_inner());
-    end["costs"] = totals.json();
+    end["costs"] = session_usage.json();
     out.system(
         "session_end",
         format!("Agent {agent_name} finished"),