Make `oa coder --delegate` start children it can account for

f00c28ebbeca · AtlantisPleb · · parent fa6c121d89f9

Make `oa coder --delegate` start children it can account for

The fan-out reported work it had not done. `ChildWorkerTask::worktree_path`
was hardcoded `None` and nothing read it, so every child ran in the
parent's directory: two children told to edit the same file edited the
same file, and a fan-out asked to try three approaches produced whichever
one finished last. All four CLI harnesses called `Command::output()`,
which returns when the child is finished, so a reader watching a
four-minute child saw nothing for four minutes and then everything at
once. There was no concurrency limit, no way to stop a running fan-out —
a reader who changed their mind had to kill the terminal, and the
children carried on spending — and the final line was
`N/N children succeeded` with exit zero whatever happened.

Each child now gets a directory of its own: inside a checkout a detached
`git worktree` of `HEAD`, which is the isolation #70 asks for, laid out
under the temporary directory so a fan-out never leaves untracked
directories in the tree the reader is working in. `--isolation` takes
`directory` or `none` for callers that want the old shared cwd.
`--count` children run at once under `--max-parallel`, capped at
`MAX_DELEGATE_COUNT`. Each child's output is read as it is written and
forwarded line by line, prefixed with the child it came from; a harness
with an event stream is rendered as what it said rather than as its wire
format. `ctrl+c` sends `SIGTERM` to each child's process group and
`SIGKILL` three seconds later, because a coding agent shells out and
killing only the agent leaves its build behind. A child that fails is
reported as failed with the exit code and what it printed, and a fan-out
that lost a child exits non-zero.

The tool executor was real and reachable from one place. Children now get
a registry rooted at their own directory, so `shell` runs where the child
lives. `delegate` is declared as a tool, so the model can ask for a
fan-out mid-sentence instead of the reader having to plan it; the gate
that carries the lane and the credential is present on the session the
reader is talking to and absent on the children it starts, so a fan-out
whose children fan out cannot happen.

`capability` is not implemented and is no longer claimed. This module's
header said it was. It is not a matter of wiring: the plugins are
WebAssembly artifacts against a bespoke `packet-v0` ABI with per-manifest
mounts, host allowlists, memory ceilings and timeouts to enforce, and
this crate has no WebAssembly runtime to enforce them with. That is the
WASM half of #71 and it needs a wasm engine in this binary. A tool that
is absent is honest; one that is advertised and refuses is not.

`acp.rs` built one `initialize` request as a struct, never sent it, and
never spoke to anything. It is now a client: it starts the agent, speaks
newline-delimited JSON-RPC over its stdio, drives initialize →
session/new → session/set_mode → session/prompt, answers the permission
request a delegated child has nobody else to answer, turns `tool_call`,
`usage_update` and `agent_message_chunk` into events as they arrive, and
kills the process group on every exit path including a timeout. The
`devin` lane runs on it.

A refusal follows what the rest of the CLI now does: an unknown lane, an
unknown isolation, a count over the cap, and a workspace that could not be
prepared print `oa: <reason>` on stderr and exit 2, with no fallback value.
A fan-out that ran and lost a child exits 1 instead, because the command was
asked for correctly and it is the work that did not finish.

Verified by running it, not by reasoning about it. Two children on the
live proxy wrote `mine.txt` into two separate registered worktrees with
two different shell pids. Three children on a stand-in harness that
sleeps one second finished in 5.23s together and over 2.5s when capped
at one at a time, with three distinct pids in three distinct
directories, and their output arrived at 1.2s, 2.2s, 3.2s, 4.2s and
5.2s rather than at the end. `SIGINT` to a live fan-out stopped both
children, reported them as stopped, exited 1, and the grandchild each
had started never fired. A child exiting 7 was reported as failed beside
a sibling that succeeded, and the command exited 1. The real Devin CLI
answered over ACP with a session id, streamed text, and real token
counts; the real Claude Code CLI streamed its tool calls live. The
declared tool list, read back from a live turn, is exactly `shell`,
`skill`, `openagents`, `delegate`.

Eleven new tests make each of those claims against a real process: a
worktree whose `.git` is a file, a clock that shows streaming, a cap
that shows a queue, a grandchild that must not outlive its parent, and
an ACP server in twenty lines of Python that the client has to actually
talk to. The harness binary is substitutable through `OA_CHILD_*` for
the same reason the TypeScript harnesses take a `command`: a test that
cannot substitute the agent either costs money or does not run.

Closes #70, #85, #72. Advances #71 and #84; the WASM capability runtime
in both remains unported and is now stated as such rather than implied.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SoZMfWRSGnf6FZX2Ar9rQ2
Co-Authored-By
Claude Fable 5 <noreply@anthropic.com>
Closes
#70
Closes
#85
Closes
#72

Deploy story

What this commit did to the running system — joined from the forge receipt chain, the part a commit page elsewhere cannot show.

pushed
by user · WAL seq 183 · 2026-08-26T06:40:01.161039Z

Changed files

  • modified Cargo.lock
  • modified crates/openagents-cli/Cargo.toml
  • modified crates/openagents-cli/src/acp.rs
  • modified crates/openagents-cli/src/cli.rs
  • modified crates/openagents-cli/src/delegate.rs
  • modified crates/openagents-cli/src/lib.rs
  • added crates/openagents-cli/src/signals.rs
  • modified crates/openagents-cli/src/tools.rs
  • added crates/openagents-cli/src/workspace.rs
  • added crates/openagents-cli/tests/acp_test.rs
  • modified crates/openagents-cli/tests/cli_test.rs
  • added crates/openagents-cli/tests/delegate_test.rs

Diff

12 files changed, +2656 -122

Cargo.lock modified +1

@@ -1573,6 +1573,7 @@ dependencies = [

1573 1573
 "crossterm",
1574 1574
 "eventsource-stream",
1575 1575
 "futures",
1576
 "libc",
1576 1577
 "openagents-all-work-contract",
1577 1578
 "openagents-cloud-contract",
1578 1579
 "ratatui",
crates/openagents-cli/Cargo.toml modified +1

@@ -24,6 +24,7 @@ clap = { version = "4", features = ["derive", "cargo"] }

24 24
crossterm = { version = "0.28", features = ["event-stream"] }
25 25
ratatui = { version = "0.29", default-features = false, features = ["crossterm"] }
26 26
futures = "0.3"
27
libc = "0.2"
27 28
async-trait = "0.1"
28 29
eventsource-stream = "0.2.3"
29 30
unicode-width = "0.2"
crates/openagents-cli/src/acp.rs modified +467 -28

@@ -1,6 +1,35 @@

1
//! Agent Client Protocol (ACP) & Devin / external harness integration
1
//! An Agent Client Protocol client, and the Devin harness built on it.
2
//!
3
//! What was here built one `initialize` request as a struct, never sent it,
4
//! and never spoke to anything. No process was started, no socket or pipe was
5
//! opened, and `handle_response` was called by nothing. This is the client
6
//! OpenAgentsInc/openagents#72 asks for: it starts the agent, speaks the
7
//! protocol over its stdio, and returns what the agent said.
8
//!
9
//! Devin's print mode (`devin -p`) writes nothing until the very end, so a
10
//! child doing four minutes of work reports nothing for four minutes and a
11
//! reader cannot tell it from a hang. `devin acp` is the same agent as an ACP
12
//! server over stdio, and it streams: `tool_call` with a title,
13
//! `tool_call_update` with a status, `usage_update` with token counts, and
14
//! `agent_message_chunk` with the answer as it is written.
15
//!
16
//! Newline-delimited JSON-RPC, one server per child. A shared server would
17
//! save a process and cost a lifecycle nobody asked for: one crash would take
18
//! every child with it.
19
//!
20
//! Devin logs heavily to stderr and none of it is protocol. It is drained and
21
//! dropped rather than parsed.
22
23
use std::path::Path;
24
use std::process::Stdio;
25
use std::time::Duration;
2 26
3 27
use serde::{Deserialize, Serialize};
28
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
29
use tokio::process::{ChildStdin, ChildStdout, Command};
30
use tokio::sync::watch;
31
32
use crate::signals::stop_tree;
4 33
5 34
#[derive(Debug, Clone, Serialize, Deserialize)]
6 35
pub struct JsonRpcRequest {

@@ -18,6 +47,11 @@ pub struct JsonRpcResponse {

18 47
    pub error: Option<serde_json::Value>,
19 48
}
20 49
50
/// How much the child is allowed to do without being asked.
51
///
52
/// The names are this CLI's; the wire carries the agent's own. Devin calls the
53
/// permissive one `bypass`. A build of the agent that does not know a mode is
54
/// not a reason to lose the child, so setting it is best effort.
21 55
#[derive(Debug, Clone, PartialEq, Eq)]
22 56
pub enum PermissionMode {
23 57
    Dangerous,

@@ -25,41 +59,446 @@ pub enum PermissionMode {

25 59
    ReadOnly,
26 60
}
27 61
28
pub struct DevinAcpClient {
29
    pub mode: PermissionMode,
30
    pub seq: u64,
62
impl PermissionMode {
63
    pub fn parse(name: &str) -> Option<Self> {
64
        match name.trim().to_lowercase().as_str() {
65
            "dangerous" | "bypass" => Some(PermissionMode::Dangerous),
66
            "prompt" | "default" | "ask" => Some(PermissionMode::Prompt),
67
            "read-only" | "readonly" => Some(PermissionMode::ReadOnly),
68
            _ => None,
69
        }
70
    }
71
72
    /// The mode id sent in `session/set_mode`.
73
    pub fn mode_id(&self) -> &'static str {
74
        match self {
75
            PermissionMode::Dangerous => "bypass",
76
            PermissionMode::Prompt => "default",
77
            PermissionMode::ReadOnly => "read-only",
78
        }
79
    }
80
}
81
82
/// What a running ACP child reports as it works.
83
#[derive(Debug, Clone)]
84
pub enum AcpEvent {
85
    Session { id: String },
86
    /// A tool the agent ran. `title` is Devin's own phrase — "Ran ls", "Read
87
    /// src/a.ts" — so it is the activity rather than a name to look up.
88
    Tool { kind: String, title: String },
89
    Tokens { input: u64, output: u64 },
90
    /// A piece of the answer, as it is written.
91
    Text { chunk: String },
92
}
93
94
/// How an ACP child is started.
95
#[derive(Debug, Clone)]
96
pub struct AcpHarness {
97
    /// The binary. Defaults to `devin`; a test points it at a stand-in.
98
    pub command: String,
99
    /// The subcommand that puts it in ACP mode.
100
    pub args: Vec<String>,
101
    pub mode: Option<PermissionMode>,
102
}
103
104
impl Default for AcpHarness {
105
    fn default() -> Self {
106
        Self {
107
            command: "devin".to_string(),
108
            args: vec!["acp".to_string()],
109
            mode: Some(PermissionMode::Dangerous),
110
        }
111
    }
112
}
113
114
/// How long the client waits for the agent to answer one request.
115
///
116
/// `session/prompt` is the whole turn, so this is the child's own ceiling
117
/// rather than a network timeout.
118
const REQUEST_TIMEOUT: Duration = Duration::from_secs(900);
119
120
/// How long the client waits for the agent's first protocol line.
121
///
122
/// An agent that needs a credential and is waiting for a terminal never writes
123
/// one, and without this the child hangs for as long as the reader leaves it.
124
const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(60);
125
126
/// Why a run ended other than by answering.
127
#[derive(Debug)]
128
pub enum AcpFailure {
129
    /// The binary is not on `PATH`, or would not start.
130
    Unstartable(String),
131
    /// The agent refused, exited early, or broke the protocol.
132
    Refused(String),
133
    /// The reader stopped the fan-out.
134
    Cancelled,
31 135
}
32 136
33
impl DevinAcpClient {
34
    pub fn new(mode: PermissionMode) -> Self {
35
        Self { mode, seq: 0 }
137
impl std::fmt::Display for AcpFailure {
138
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
139
        match self {
140
            AcpFailure::Unstartable(why) => write!(f, "{why}"),
141
            AcpFailure::Refused(why) => write!(f, "{why}"),
142
            AcpFailure::Cancelled => write!(f, "stopped before finishing"),
143
        }
36 144
    }
145
}
146
147
impl AcpHarness {
148
    /// Run one prompt to completion and return what the agent said.
149
    ///
150
    /// `on_event` is called as the agent works, so a caller can stream rather
151
    /// than wait for the end. The child is spawned into a process group of its
152
    /// own and the group is killed on every exit path, including cancellation
153
    /// and a timeout — an agent that shells out must not outlive the run that
154
    /// started it.
155
    pub async fn run<F>(
156
        &self,
157
        prompt: &str,
158
        cwd: &Path,
159
        mut on_event: F,
160
        cancel: &mut watch::Receiver<bool>,
161
    ) -> Result<String, AcpFailure>
162
    where
163
        F: FnMut(AcpEvent) + Send,
164
    {
165
        let mut child = Command::new(&self.command)
166
            .args(&self.args)
167
            .current_dir(cwd)
168
            .stdin(Stdio::piped())
169
            .stdout(Stdio::piped())
170
            .stderr(Stdio::piped())
171
            // Its own process group, so stopping the child stops what the
172
            // child started.
173
            .process_group(0)
174
            .spawn()
175
            .map_err(|error| {
176
                AcpFailure::Unstartable(if error.kind() == std::io::ErrorKind::NotFound {
177
                    format!("the `{}` command is not on PATH", self.command)
178
                } else {
179
                    format!("the `{}` command would not start: {error}", self.command)
180
                })
181
            })?;
182
183
        // Devin's own logging. Not protocol, and there is a lot of it.
184
        if let Some(stderr) = child.stderr.take() {
185
            tokio::spawn(async move {
186
                let mut lines = BufReader::new(stderr).lines();
187
                while let Ok(Some(_)) = lines.next_line().await {}
188
            });
189
        }
190
191
        let stdin = child.stdin.take();
192
        let stdout = child.stdout.take();
193
        let (Some(mut stdin), Some(stdout)) = (stdin, stdout) else {
194
            stop_tree(&mut child).await;
195
            return Err(AcpFailure::Refused(
196
                "the agent's standard streams could not be opened".to_string(),
197
            ));
198
        };
199
        let mut lines = BufReader::new(stdout).lines();
200
201
        let outcome = self
202
            .converse(prompt, cwd, &mut stdin, &mut lines, &mut on_event, cancel)
203
            .await;
204
205
        stop_tree(&mut child).await;
206
        outcome
207
    }
208
209
    /// The conversation itself: handshake, session, prompt, answer.
210
    async fn converse<F>(
211
        &self,
212
        prompt: &str,
213
        cwd: &Path,
214
        stdin: &mut ChildStdin,
215
        lines: &mut tokio::io::Lines<BufReader<ChildStdout>>,
216
        on_event: &mut F,
217
        cancel: &mut watch::Receiver<bool>,
218
    ) -> Result<String, AcpFailure>
219
    where
220
        F: FnMut(AcpEvent) + Send,
221
    {
222
        let mut seq: u64 = 0;
223
        let mut answer = String::new();
224
225
        request(
226
            stdin,
227
            lines,
228
            &mut seq,
229
            "initialize",
230
            serde_json::json!({
231
                "protocolVersion": 1,
232
                "clientCapabilities": {"fs": {"readTextFile": false, "writeTextFile": false}}
233
            }),
234
            HANDSHAKE_TIMEOUT,
235
            &mut answer,
236
            on_event,
237
            cancel,
238
        )
239
        .await?;
240
241
        let opened = request(
242
            stdin,
243
            lines,
244
            &mut seq,
245
            "session/new",
246
            serde_json::json!({"cwd": cwd.to_string_lossy(), "mcpServers": []}),
247
            REQUEST_TIMEOUT,
248
            &mut answer,
249
            on_event,
250
            cancel,
251
        )
252
        .await?;
253
254
        let session_id = opened
255
            .get("sessionId")
256
            .and_then(|v| v.as_str())
257
            .ok_or_else(|| AcpFailure::Refused("the agent opened no session".to_string()))?
258
            .to_string();
259
        on_event(AcpEvent::Session {
260
            id: session_id.clone(),
261
        });
262
263
        if let Some(mode) = &self.mode {
264
            // Best effort: a build of the agent without this mode should not
265
            // cost the child over the name of a permission setting.
266
            let _ = request(
267
                stdin,
268
                lines,
269
                &mut seq,
270
                "session/set_mode",
271
                serde_json::json!({"sessionId": session_id, "modeId": mode.mode_id()}),
272
                REQUEST_TIMEOUT,
273
                &mut answer,
274
                on_event,
275
                cancel,
276
            )
277
            .await;
278
        }
37 279
38
    pub fn build_initialize_request(&mut self) -> JsonRpcRequest {
39
        self.seq += 1;
40
        JsonRpcRequest {
41
            jsonrpc: "2.0".to_string(),
42
            id: self.seq,
43
            method: "initialize".to_string(),
44
            params: serde_json::json!({
45
                "protocolVersion": "2024-11-05",
46
                "clientInfo": {
47
                    "name": "openagents-cli-rust",
48
                    "version": "0.1.0"
49
                },
50
                "capabilities": {
51
                    "tools": true,
52
                    "prompts": true
53
                }
280
        request(
281
            stdin,
282
            lines,
283
            &mut seq,
284
            "session/prompt",
285
            serde_json::json!({
286
                "sessionId": session_id,
287
                "prompt": [{"type": "text", "text": prompt}]
54 288
            }),
289
            REQUEST_TIMEOUT,
290
            &mut answer,
291
            on_event,
292
            cancel,
293
        )
294
        .await?;
295
296
        Ok(answer.trim().to_string())
297
    }
298
}
299
300
/// Send one request and pump the stream until its reply arrives.
301
///
302
/// Everything else that comes down the pipe while waiting is handled on the
303
/// way past: notifications become events, and a permission request is answered
304
/// here because a delegated child has nobody to ask. An unanswered permission
305
/// request hangs the agent for as long as the reader leaves it.
306
#[allow(clippy::too_many_arguments)]
307
async fn request<F>(
308
    stdin: &mut ChildStdin,
309
    lines: &mut tokio::io::Lines<BufReader<ChildStdout>>,
310
    seq: &mut u64,
311
    method: &str,
312
    params: serde_json::Value,
313
    limit: Duration,
314
    answer: &mut String,
315
    on_event: &mut F,
316
    cancel: &mut watch::Receiver<bool>,
317
) -> Result<serde_json::Value, AcpFailure>
318
where
319
    F: FnMut(AcpEvent) + Send,
320
{
321
    *seq += 1;
322
    let id = *seq;
323
    let line = serde_json::json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params});
324
    write_line(stdin, &line).await?;
325
326
    let deadline = tokio::time::Instant::now() + limit;
327
328
    loop {
329
        let next = tokio::select! {
330
            biased;
331
            _ = cancel.changed() => return Err(AcpFailure::Cancelled),
332
            read = lines.next_line() => read,
333
            _ = tokio::time::sleep_until(deadline) => {
334
                return Err(AcpFailure::Refused(format!(
335
                    "the agent did not answer `{method}` within {}s", limit.as_secs()
336
                )));
337
            }
338
        };
339
340
        let raw = match next {
341
            Ok(Some(raw)) => raw,
342
            Ok(None) => {
343
                return Err(AcpFailure::Refused(
344
                    "the agent exited before it answered".to_string(),
345
                ))
346
            }
347
            Err(error) => {
348
                return Err(AcpFailure::Refused(format!(
349
                    "the agent's output could not be read: {error}"
350
                )))
351
            }
352
        };
353
354
        let trimmed = raw.trim();
355
        if trimmed.is_empty() {
356
            continue;
357
        }
358
        // Not protocol. Agents write plain lines to stdout too.
359
        let Ok(message) = serde_json::from_str::<serde_json::Value>(trimmed) else {
360
            continue;
361
        };
362
363
        // A reply to something asked for.
364
        let is_reply = message.get("id").and_then(|v| v.as_u64()).is_some()
365
            && message.get("method").is_none();
366
        if is_reply {
367
            if message.get("id").and_then(|v| v.as_u64()) != Some(id) {
368
                continue;
369
            }
370
            if let Some(error) = message.get("error") {
371
                let text = serde_json::to_string(error).unwrap_or_default();
372
                return Err(AcpFailure::Refused(format!(
373
                    "the agent refused `{method}`: {}",
374
                    &text[..text.len().min(200)]
375
                )));
376
            }
377
            return Ok(message
378
                .get("result")
379
                .cloned()
380
                .unwrap_or(serde_json::json!({})));
55 381
        }
382
383
        handle_incoming(&message, stdin, answer, on_event).await?;
384
    }
385
}
386
387
/// A notification, or a request from the agent.
388
async fn handle_incoming<F>(
389
    message: &serde_json::Value,
390
    stdin: &mut ChildStdin,
391
    answer: &mut String,
392
    on_event: &mut F,
393
) -> Result<(), AcpFailure>
394
where
395
    F: FnMut(AcpEvent) + Send,
396
{
397
    let method = message.get("method").and_then(|v| v.as_str()).unwrap_or("");
398
399
    if method == "session/request_permission" {
400
        let Some(id) = message.get("id").and_then(|v| v.as_u64()) else {
401
            return Ok(());
402
        };
403
        let params = message
404
            .get("params")
405
            .cloned()
406
            .unwrap_or(serde_json::json!({}));
407
        let outcome = match first_allow_option(&params) {
408
            Some(option) => serde_json::json!({"outcome": "selected", "optionId": option}),
409
            None => serde_json::json!({"outcome": "cancelled"}),
410
        };
411
        write_line(
412
            stdin,
413
            &serde_json::json!({"jsonrpc": "2.0", "id": id, "result": {"outcome": outcome}}),
414
        )
415
        .await?;
416
        return Ok(());
417
    }
418
419
    if method != "session/update" {
420
        return Ok(());
56 421
    }
57 422
58
    pub fn handle_response(&self, res: JsonRpcResponse) -> Result<serde_json::Value, String> {
59
        if let Some(err) = res.error {
60
            Err(format!("ACP error: {:?}", err))
61
        } else {
62
            Ok(res.result.unwrap_or(serde_json::Value::Null))
423
    let update = message
424
        .get("params")
425
        .and_then(|p| p.get("update"))
426
        .cloned()
427
        .unwrap_or(serde_json::json!({}));
428
429
    match update.get("sessionUpdate").and_then(|v| v.as_str()) {
430
        Some("tool_call") => {
431
            on_event(AcpEvent::Tool {
432
                kind: update
433
                    .get("kind")
434
                    .and_then(|v| v.as_str())
435
                    .unwrap_or("tool")
436
                    .to_string(),
437
                title: update
438
                    .get("title")
439
                    .and_then(|v| v.as_str())
440
                    .unwrap_or("")
441
                    .to_string(),
442
            });
63 443
        }
444
        Some("usage_update") => {
445
            let meta = update.get("_meta").cloned().unwrap_or(serde_json::json!({}));
446
            let input = meta.get("cognition.ai/inputTokens").and_then(|v| v.as_u64());
447
            let output = meta.get("cognition.ai/outputTokens").and_then(|v| v.as_u64());
448
            if let (Some(input), Some(output)) = (input, output) {
449
                on_event(AcpEvent::Tokens { input, output });
450
            }
451
        }
452
        Some("agent_message_chunk") => {
453
            if let Some(piece) = update
454
                .get("content")
455
                .and_then(|c| c.get("text"))
456
                .and_then(|v| v.as_str())
457
            {
458
                answer.push_str(piece);
459
                on_event(AcpEvent::Text {
460
                    chunk: piece.to_string(),
461
                });
462
            }
463
        }
464
        _ => {}
64 465
    }
466
467
    Ok(())
468
}
469
470
async fn write_line(
471
    stdin: &mut ChildStdin,
472
    value: &serde_json::Value,
473
) -> Result<(), AcpFailure> {
474
    let mut line = serde_json::to_string(value).unwrap_or_default();
475
    line.push('\n');
476
    stdin
477
        .write_all(line.as_bytes())
478
        .await
479
        .map_err(|error| AcpFailure::Refused(format!("the agent stopped reading: {error}")))?;
480
    stdin
481
        .flush()
482
        .await
483
        .map_err(|error| AcpFailure::Refused(format!("the agent stopped reading: {error}")))
484
}
485
486
/// The option a permission request offers that lets the work continue.
487
pub fn first_allow_option(params: &serde_json::Value) -> Option<String> {
488
    let options = params.get("options")?.as_array()?;
489
    let named: Vec<&serde_json::Value> = options
490
        .iter()
491
        .filter(|option| option.get("optionId").and_then(|v| v.as_str()).is_some())
492
        .collect();
493
    let allow = named.iter().find(|option| {
494
        option
495
            .get("kind")
496
            .and_then(|v| v.as_str())
497
            .is_some_and(|kind| kind.starts_with("allow"))
498
    });
499
    allow
500
        .or(named.first())
501
        .and_then(|option| option.get("optionId"))
502
        .and_then(|v| v.as_str())
503
        .map(String::from)
65 504
}
crates/openagents-cli/src/cli.rs modified +27 -4

@@ -202,9 +202,21 @@ pub struct CoderArgs {

202 202
    #[arg(long, help = "Delegate prompt to parallel child agents")]
203 203
    pub delegate: bool,
204 204
205
    #[arg(long, default_value_t = 1, help = "Parallel child worker count")]
205
    #[arg(long, default_value_t = 1, help = "How many child agents run the prompt")]
206 206
    pub count: usize,
207 207
208
    #[arg(long, help = "How many children run at once. Defaults to all of them")]
209
    pub max_parallel: Option<usize>,
210
211
    #[arg(
212
        long,
213
        help = "Working directory each child gets: worktree (default, a detached git worktree of HEAD), directory, or none"
214
    )]
215
    pub isolation: Option<String>,
216
217
    #[arg(long, help = "Leave the children's worktrees on disk so their work can be read")]
218
    pub keep_workspaces: bool,
219
208 220
    #[arg(long, help = "Target harness lane (e.g. ox-alpha, gemini, devin, claude, codex)")]
209 221
    pub lane: Option<String>,
210 222

@@ -450,8 +462,19 @@ pub async fn run(cli: Cli) -> Result<(), Box<dyn std::error::Error>> {

450 462
            } else if coder.headless {
451 463
                let prompt = coder.prompt.unwrap_or_else(|| "Analyze workspace and run tests".to_string());
452 464
                println!("Executing coder prompt headlessly: {}", prompt);
453
                let tools = crate::tools::HarnessToolRegistry::new(None);
454
                let lane = crate::runtime::Lane::from_str(&coder.lane.unwrap_or_else(|| "ox-alpha".to_string()));
465
                let lane_name = coder.lane.unwrap_or_else(|| "ox-alpha".to_string());
466
                // A headless session may start children. They run on the same
467
                // lane and the same credential, and they do not get the tool
468
                // themselves.
469
                let tools = crate::tools::HarnessToolRegistry::with_delegation(
470
                    None,
471
                    crate::tools::DelegationGate {
472
                        lane: lane_name.clone(),
473
                        user_token: token.clone(),
474
                        max_count: crate::delegate::MAX_DELEGATE_COUNT,
475
                    },
476
                );
477
                let lane = crate::runtime::Lane::from_str(&lane_name);
455 478
                let mut runtime = crate::runtime::CoderRuntimeSession::new(lane, None, token, tools);
456 479
                let result = runtime.execute_turn(&prompt, |chunk| {
457 480
                    print!("{}", chunk);

@@ -572,7 +595,7 @@ pub async fn run(cli: Cli) -> Result<(), Box<dyn std::error::Error>> {

572 595
/// Exit code 2 is what the TypeScript CLI returns for an input or configuration
573 596
/// error, and the point of this whole path: a command that cannot reach its data
574 597
/// says so and exits non-zero rather than returning something plausible.
575
fn fail(message: &str) -> ! {
598
pub(crate) fn fail(message: &str) -> ! {
576 599
    eprintln!("oa: {}", message);
577 600
    std::process::exit(2)
578 601
}
crates/openagents-cli/src/delegate.rs modified +1089 -84

@@ -1,21 +1,63 @@

1
//! Child agent delegation engine with live CLI harnesses
2
//! Supports ox-alpha, opencode, devin (via ACP/CLI), claude, and codex
1
//! Child agent delegation: a fan-out of real coding agents, each in its own
2
//! working directory, streaming as they go.
3
//!
4
//! What was here reported a fan-out it had not performed. `worktree_path` was
5
//! hardcoded `None` and nothing read it, so every child ran in the parent's
6
//! directory and two children told to edit the same file edited the same file.
7
//! There was no concurrency limit, no way to stop a running fan-out, and no
8
//! streaming: all four CLI harnesses called `Command::output()`, which returns
9
//! when the child is finished, so a reader watching a four-minute child saw
10
//! nothing for four minutes and then everything at once.
11
//!
12
//! Now:
13
//!
14
//! - Each child gets its own directory ([`crate::workspace`]), a detached git
15
//!   worktree of `HEAD` inside a checkout.
16
//! - `--count` children run at once, under a cap (`--max-parallel`), bounded
17
//!   by [`MAX_DELEGATE_COUNT`].
18
//! - Every child's output is forwarded line by line as it arrives, prefixed
19
//!   with the child it came from.
20
//! - `ctrl+c` stops the fan-out: children are signalled `SIGTERM` by process
21
//!   group, then `SIGKILL` after a grace period, so a child's own subprocesses
22
//!   go with it.
23
//! - A child that fails is reported as failed, and a fan-out with any failed
24
//!   child is a failed command.
25
//!
26
//! The lanes follow `coder-delegate.ts`: `ox-alpha` runs in this process on
27
//! the OpenAgents inference proxy with this session's tools; `claude`,
28
//! `codex`, and `gemini`/`opencode/*` run the corresponding CLI on this
29
//! machine; `devin` runs the Devin CLI as an ACP server ([`crate::acp`]).
3 30
4
use crate::cli::CoderArgs;
5
use crate::runtime::{CoderRuntimeSession, Lane};
6
use crate::tools::HarnessToolRegistry;
7
use futures::future::join_all;
8
use serde::{Deserialize, Serialize};
9 31
use std::path::PathBuf;
10 32
use std::process::Stdio;
33
use std::sync::Arc;
11 34
use std::time::Instant;
35
36
use serde::{Deserialize, Serialize};
37
use tokio::io::{AsyncBufReadExt, BufReader};
12 38
use tokio::process::Command;
39
use tokio::sync::{mpsc, watch, Semaphore};
40
41
use crate::acp::{AcpEvent, AcpFailure, AcpHarness, PermissionMode};
42
use crate::cli::{fail, CoderArgs};
43
use crate::runtime::{CoderRuntimeSession, Lane};
44
use crate::signals::stop_tree;
45
use crate::tools::HarnessToolRegistry;
46
use crate::workspace::{ChildWorkspace, Isolation, WorkspacePlan};
47
48
/// The most children one fan-out may run. Matches `MAX_DELEGATE_COUNT` in
49
/// `coder-delegate.ts`.
50
pub const MAX_DELEGATE_COUNT: usize = 32;
51
52
/// How much of a child's answer is kept for the report.
53
pub const CHILD_RESULT_LIMIT: usize = 30_000;
13 54
14 55
#[derive(Debug, Clone, Serialize, Deserialize)]
15 56
pub struct ChildWorkerTask {
16 57
    pub id: usize,
17 58
    pub prompt: String,
18 59
    pub lane: String,
60
    /// Where this child works. Populated now, and read.
19 61
    pub worktree_path: Option<PathBuf>,
20 62
}
21 63

@@ -23,138 +65,1101 @@ pub struct ChildWorkerTask {

23 65
pub struct ChildWorkerResult {
24 66
    pub id: usize,
25 67
    pub success: bool,
68
    /// The child's answer when it succeeded, or why it did not.
26 69
    pub output: String,
27 70
    pub duration_ms: u128,
71
    /// The operating system process, for a child that is one. `ox-alpha` runs
72
    /// in this process and reports `None`; what it starts are its `shell` tool
73
    /// subprocesses.
74
    pub pid: Option<u32>,
75
    pub workspace: Option<PathBuf>,
76
    /// Set when the child did not answer, so a caller does not have to read
77
    /// `output` to find out.
78
    pub failure: Option<String>,
79
}
80
81
/// What a child reports while it works.
82
#[derive(Debug, Clone)]
83
pub enum ChildEvent {
84
    Started {
85
        id: usize,
86
        lane: String,
87
        workspace: String,
88
        pid: Option<u32>,
89
    },
90
    /// A piece of what the child wrote, exactly as it arrived.
91
    Output { id: usize, text: String },
92
    /// Something the child did that is not its answer: a tool call, a token
93
    /// count, a session id.
94
    Activity { id: usize, text: String },
95
    Finished(Box<ChildWorkerResult>),
96
}
97
98
/// Which harness and model a child runs on.
99
#[derive(Debug, Clone, PartialEq, Eq)]
100
pub enum ChildLane {
101
    /// This process, on the OpenAgents inference proxy, with this session's
102
    /// tools.
103
    OxAlpha,
104
    /// The `opencode` CLI on this machine, with its own tools.
105
    Opencode { model: String },
106
    /// The Devin CLI as an ACP server.
107
    Devin,
108
    /// The Claude Code CLI in print mode.
109
    Claude,
110
    /// The OpenAI Codex CLI in exec mode.
111
    Codex,
112
}
113
114
impl ChildLane {
115
    pub fn parse(name: &str) -> Self {
116
        let lowered = name.trim().to_lowercase();
117
        match lowered.as_str() {
118
            "gemini" | "gemini-flash" => ChildLane::Opencode {
119
                model: "gemini-3.7-flash".to_string(),
120
            },
121
            "devin" => ChildLane::Devin,
122
            "claude" => ChildLane::Claude,
123
            "codex" => ChildLane::Codex,
124
            "ox-alpha" | "ox" | "openagents" => ChildLane::OxAlpha,
125
            other if other.starts_with("opencode/") => ChildLane::Opencode {
126
                model: other.trim_start_matches("opencode/").to_string(),
127
            },
128
            // An unknown name used to fall through to `ox-alpha` in silence, so
129
            // a typo spent this account's budget on a lane the caller did not
130
            // ask for. It still runs there, but the caller is told.
131
            _ => ChildLane::OxAlpha,
132
        }
133
    }
134
135
    /// Whether [`ChildLane::parse`] recognised the name it was given.
136
    pub fn known(name: &str) -> bool {
137
        let lowered = name.trim().to_lowercase();
138
        matches!(
139
            lowered.as_str(),
140
            "gemini" | "gemini-flash" | "devin" | "claude" | "codex" | "ox-alpha" | "ox" | "openagents"
141
        ) || lowered.starts_with("opencode/")
142
    }
143
144
    pub fn label(&self) -> String {
145
        match self {
146
            ChildLane::OxAlpha => "ox-alpha (this process, the OpenAgents proxy)".to_string(),
147
            ChildLane::Opencode { model } => format!("opencode ({model})"),
148
            ChildLane::Devin => "devin (ACP over the Devin CLI)".to_string(),
149
            ChildLane::Claude => "claude (Claude Code print mode)".to_string(),
150
            ChildLane::Codex => "codex (Codex exec)".to_string(),
151
        }
152
    }
153
154
    /// The binary this lane needs on `PATH`, if it needs one.
155
    pub fn binary(&self) -> Option<&'static str> {
156
        match self {
157
            ChildLane::OxAlpha => None,
158
            ChildLane::Opencode { .. } => Some("opencode"),
159
            ChildLane::Devin => Some("devin"),
160
            ChildLane::Claude => Some("claude"),
161
            ChildLane::Codex => Some("codex"),
162
        }
163
    }
28 164
}
29 165
30 166
pub struct DelegationSupervisor {
31 167
    pub count: usize,
32 168
    pub lane: String,
33 169
    pub user_token: Option<String>,
170
    /// How much of a directory each child gets to itself.
171
    pub isolation: Isolation,
172
    /// How many children run at once. Defaults to all of them.
173
    pub max_parallel: usize,
174
    /// Leave the children's worktrees on disk when the fan-out is over, so
175
    /// what they wrote can be read or merged.
176
    pub keep_workspaces: bool,
34 177
}
35 178
36 179
impl DelegationSupervisor {
37 180
    pub fn new(count: usize, lane: &str, user_token: Option<String>) -> Self {
181
        let count = count.clamp(1, MAX_DELEGATE_COUNT);
38 182
        Self {
39 183
            count,
40 184
            lane: lane.to_string(),
41 185
            user_token,
186
            isolation: Isolation::Worktree,
187
            max_parallel: count,
188
            keep_workspaces: false,
42 189
        }
43 190
    }
44 191
192
    pub fn with_isolation(mut self, isolation: Isolation) -> Self {
193
        self.isolation = isolation;
194
        self
195
    }
196
197
    pub fn with_max_parallel(mut self, max_parallel: usize) -> Self {
198
        self.max_parallel = max_parallel.clamp(1, self.count);
199
        self
200
    }
201
202
    pub fn keeping_workspaces(mut self, keep: bool) -> Self {
203
        self.keep_workspaces = keep;
204
        self
205
    }
206
207
    /// Run the fan-out and return every child's outcome.
208
    ///
209
    /// Convenience over [`DelegationSupervisor::dispatch_streaming`] for a
210
    /// caller that has nowhere to stream to.
45 211
    pub async fn dispatch(&self, prompt: &str) -> Vec<ChildWorkerResult> {
46
        let mut handles = Vec::new();
47
        for id in 1..=self.count {
212
        let (events, mut drain) = mpsc::unbounded_channel();
213
        let sink = tokio::spawn(async move { while drain.recv().await.is_some() {} });
214
        let (_stop, cancel) = watch::channel(false);
215
        let results = self.dispatch_streaming(prompt, events, cancel).await;
216
        let _ = sink.await;
217
        results.unwrap_or_default()
218
    }
219
220
    /// Run the fan-out, reporting each child as it goes.
221
    ///
222
    /// Returns `Err` only when no child could be started at all — a workspace
223
    /// that could not be prepared. A child that fails is a result with
224
    /// `success: false`, because the other children's answers are still worth
225
    /// having.
226
    pub async fn dispatch_streaming(
227
        &self,
228
        prompt: &str,
229
        events: mpsc::UnboundedSender<ChildEvent>,
230
        cancel: watch::Receiver<bool>,
231
    ) -> Result<Vec<ChildWorkerResult>, String> {
232
        let lane = ChildLane::parse(&self.lane);
233
        let cwd = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
234
        let plan = WorkspacePlan::resolve(cwd, self.isolation).await;
235
        let workspaces = plan.prepare(self.count).await?;
236
237
        let gate = Arc::new(Semaphore::new(self.max_parallel.max(1)));
238
        let mut handles = Vec::with_capacity(self.count);
239
240
        for workspace in workspaces.clone() {
48 241
            let task = ChildWorkerTask {
49
                id,
50
                prompt: prompt.to_string(),
242
                id: workspace.id,
243
                prompt: identify(prompt, workspace.id, self.count),
51 244
                lane: self.lane.clone(),
52
                worktree_path: None,
245
                worktree_path: Some(workspace.path.clone()),
53 246
            };
247
            let lane = lane.clone();
54 248
            let token = self.user_token.clone();
249
            let events = events.clone();
250
            let cancel = cancel.clone();
251
            let gate = Arc::clone(&gate);
252
55 253
            handles.push(tokio::spawn(async move {
56
                Self::execute_worker(task, token).await
254
                // The cap is here rather than around the spawn so a child that
255
                // is waiting for a slot still exists and still reports.
256
                let _slot = gate.acquire().await;
257
                let result = run_child(task, lane, workspace, token, &events, cancel).await;
258
                let _ = events.send(ChildEvent::Finished(Box::new(result.clone())));
259
                result
57 260
            }));
58 261
        }
59 262
60
        let mut results = Vec::new();
61
        for handle in join_all(handles).await {
62
            if let Ok(res) = handle {
63
                results.push(res);
263
        let mut results = Vec::with_capacity(handles.len());
264
        for handle in handles {
265
            match handle.await {
266
                Ok(result) => results.push(result),
267
                // A panic in a child's task is a failed child, not a lost one.
268
                Err(error) => results.push(ChildWorkerResult {
269
                    id: 0,
270
                    success: false,
271
                    output: format!("the child's task ended abnormally: {error}"),
272
                    duration_ms: 0,
273
                    pid: None,
274
                    workspace: None,
275
                    failure: Some(format!("the child's task ended abnormally: {error}")),
276
                }),
64 277
            }
65 278
        }
66
        results
67
    }
68
69
    async fn execute_worker(task: ChildWorkerTask, user_token: Option<String>) -> ChildWorkerResult {
70
        let start = Instant::now();
71
        let lane_str = task.lane.to_lowercase();
72
73
        let (success, output) = match lane_str.as_str() {
74
            "claude" => run_claude_cli(&task.prompt).await,
75
            "codex" => run_codex_cli(&task.prompt).await,
76
            "gemini" => run_opencode_cli(&task.prompt, "gemini-3.7-flash").await,
77
            "devin" => run_devin_cli(&task.prompt).await,
78
            _ => {
79
                // Default ox-alpha via live CoderRuntimeSession
80
                let tools = HarnessToolRegistry::new(None);
81
                let mut runtime = CoderRuntimeSession::new(Lane::OxAlpha, None, user_token, tools);
82
                match runtime.execute_turn(&task.prompt, |_| {}).await {
83
                    Ok(out) => (true, out),
84
                    Err(e) => (false, format!("Inference error: {}", e)),
279
        results.sort_by_key(|result| result.id);
280
281
        if !self.keep_workspaces {
282
            for workspace in &workspaces {
283
                if let Some(problem) = workspace.release().await {
284
                    let _ = events.send(ChildEvent::Activity {
285
                        id: workspace.id,
286
                        text: problem,
287
                    });
85 288
                }
86 289
            }
87
        };
290
        }
291
292
        Ok(results)
293
    }
294
}
295
296
/// Tell a child which of the fan-out it is.
297
///
298
/// Every child gets the same prompt, so a prompt that says "your own file"
299
/// otherwise has no way to mean anything and the whole fleet writes the same
300
/// one. A single child is told nothing, because there is nothing to
301
/// distinguish.
302
pub fn identify(prompt: &str, index: usize, count: usize) -> String {
303
    if count == 1 {
304
        return prompt.to_string();
305
    }
306
    format!("You are child {index} of {count}.\n\n{prompt}")
307
}
308
309
async fn run_child(
310
    task: ChildWorkerTask,
311
    lane: ChildLane,
312
    workspace: ChildWorkspace,
313
    user_token: Option<String>,
314
    events: &mpsc::UnboundedSender<ChildEvent>,
315
    cancel: watch::Receiver<bool>,
316
) -> ChildWorkerResult {
317
    let start = Instant::now();
318
    let id = task.id;
88 319
89
        ChildWorkerResult {
90
            id: task.id,
91
            success,
92
            output,
93
            duration_ms: start.elapsed().as_millis(),
320
    let outcome = match &lane {
321
        ChildLane::OxAlpha => {
322
            let _ = events.send(ChildEvent::Started {
323
                id,
324
                lane: lane.label(),
325
                workspace: workspace.describe(),
326
                pid: None,
327
            });
328
            run_proxy_child(&task, &workspace, user_token, events, cancel).await
329
        }
330
        ChildLane::Devin => {
331
            run_devin_child(&task, &lane, &workspace, events, cancel).await
332
        }
333
        ChildLane::Claude | ChildLane::Codex | ChildLane::Opencode { .. } => {
334
            run_cli_child(&task, &lane, &workspace, events, cancel).await
94 335
        }
336
    };
337
338
    let duration_ms = start.elapsed().as_millis();
339
    match outcome {
340
        Ok(ChildAnswer { text, pid }) => ChildWorkerResult {
341
            id,
342
            success: true,
343
            output: clip(&text),
344
            duration_ms,
345
            pid,
346
            workspace: Some(workspace.path.clone()),
347
            failure: None,
348
        },
349
        Err(ChildFailure { why, pid }) => ChildWorkerResult {
350
            id,
351
            success: false,
352
            output: why.clone(),
353
            duration_ms,
354
            pid,
355
            workspace: Some(workspace.path.clone()),
356
            failure: Some(why),
357
        },
95 358
    }
96 359
}
97 360
98
async fn run_claude_cli(prompt: &str) -> (bool, String) {
99
    let mut cmd = Command::new("claude");
100
    cmd.args(["-p", prompt]);
101
    cmd.stdout(Stdio::piped());
102
    cmd.stderr(Stdio::piped());
361
struct ChildAnswer {
362
    text: String,
363
    pid: Option<u32>,
364
}
365
366
struct ChildFailure {
367
    why: String,
368
    pid: Option<u32>,
369
}
370
371
/// A child on this process's own runtime, over the inference proxy.
372
///
373
/// It gets a tool registry rooted at its own directory, so its `shell` tool
374
/// runs there and what it writes lands there. It does not get the `delegate`
375
/// tool: a fan-out whose children fan out is a fan-out with no ceiling.
376
async fn run_proxy_child(
377
    task: &ChildWorkerTask,
378
    workspace: &ChildWorkspace,
379
    user_token: Option<String>,
380
    events: &mpsc::UnboundedSender<ChildEvent>,
381
    mut cancel: watch::Receiver<bool>,
382
) -> Result<ChildAnswer, ChildFailure> {
383
    let tools = HarnessToolRegistry::child(Some(workspace.path.clone()));
384
    let mut runtime = CoderRuntimeSession::new(Lane::OxAlpha, None, user_token, tools);
385
386
    let id = task.id;
387
    let sink = events.clone();
388
    let turn = runtime.execute_turn(&task.prompt, move |chunk| {
389
        let _ = sink.send(ChildEvent::Output {
390
            id,
391
            text: chunk.to_string(),
392
        });
393
    });
103 394
104
    match cmd.output().await {
105
        Ok(res) => (res.status.success(), String::from_utf8_lossy(&res.stdout).trim().to_string()),
106
        Err(e) => (false, format!("Failed to spawn claude: {}", e)),
395
    tokio::select! {
396
        biased;
397
        _ = cancel.changed() => Err(ChildFailure {
398
            why: "stopped before finishing".to_string(),
399
            pid: None,
400
        }),
401
        answered = turn => match answered {
402
            Ok(text) => Ok(ChildAnswer { text, pid: None }),
403
            Err(error) => Err(ChildFailure { why: error.to_string(), pid: None }),
404
        },
107 405
    }
108 406
}
109 407
110
async fn run_codex_cli(prompt: &str) -> (bool, String) {
111
    let mut cmd = Command::new("codex");
112
    cmd.args(["exec", prompt]);
113
    cmd.stdout(Stdio::piped());
114
    cmd.stderr(Stdio::piped());
408
/// A child on the Devin CLI, over the Agent Client Protocol.
409
async fn run_devin_child(
410
    task: &ChildWorkerTask,
411
    lane: &ChildLane,
412
    workspace: &ChildWorkspace,
413
    events: &mpsc::UnboundedSender<ChildEvent>,
414
    mut cancel: watch::Receiver<bool>,
415
) -> Result<ChildAnswer, ChildFailure> {
416
    let id = task.id;
417
    let _ = events.send(ChildEvent::Started {
418
        id,
419
        lane: lane.label(),
420
        workspace: workspace.describe(),
421
        pid: None,
422
    });
115 423
116
    match cmd.output().await {
117
        Ok(res) => (res.status.success(), String::from_utf8_lossy(&res.stdout).trim().to_string()),
118
        Err(e) => (false, format!("Failed to spawn codex: {}", e)),
424
    let harness = AcpHarness {
425
        command: harness_binary(lane),
426
        mode: Some(PermissionMode::Dangerous),
427
        ..AcpHarness::default()
428
    };
429
    let sink = events.clone();
430
    let answered = harness
431
        .run(
432
            &task.prompt,
433
            &workspace.path,
434
            move |event| {
435
                let text = match event {
436
                    AcpEvent::Session { id: session } => format!("session {session}"),
437
                    AcpEvent::Tool { kind, title } => {
438
                        if title.is_empty() {
439
                            kind
440
                        } else {
441
                            title
442
                        }
443
                    }
444
                    AcpEvent::Tokens { input, output } => {
445
                        format!("{input} in / {output} out tokens")
446
                    }
447
                    AcpEvent::Text { chunk } => {
448
                        let _ = sink.send(ChildEvent::Output { id, text: chunk });
449
                        return;
450
                    }
451
                };
452
                let _ = sink.send(ChildEvent::Activity { id, text });
453
            },
454
            &mut cancel,
455
        )
456
        .await;
457
458
    match answered {
459
        Ok(text) => Ok(ChildAnswer { text, pid: None }),
460
        Err(AcpFailure::Cancelled) => Err(ChildFailure {
461
            why: "stopped before finishing".to_string(),
462
            pid: None,
463
        }),
464
        Err(other) => Err(ChildFailure {
465
            why: other.to_string(),
466
            pid: None,
467
        }),
119 468
    }
120 469
}
121 470
122
async fn run_opencode_cli(prompt: &str, model: &str) -> (bool, String) {
123
    let mut cmd = Command::new("opencode");
124
    cmd.args(["run", "--model", model, prompt]);
125
    cmd.stdout(Stdio::piped());
126
    cmd.stderr(Stdio::piped());
471
/// A child on another coding CLI: `claude`, `codex`, or `opencode`.
472
///
473
/// The child is spawned into a process group of its own, its two output
474
/// streams are read as they are written rather than at the end, and both are
475
/// forwarded upward line by line. The answer is pulled out of the harness's
476
/// own event stream where the harness has one, and is the tail of what it
477
/// printed where it does not.
478
async fn run_cli_child(
479
    task: &ChildWorkerTask,
480
    lane: &ChildLane,
481
    workspace: &ChildWorkspace,
482
    events: &mpsc::UnboundedSender<ChildEvent>,
483
    mut cancel: watch::Receiver<bool>,
484
) -> Result<ChildAnswer, ChildFailure> {
485
    let id = task.id;
486
    let (command, args) = harness_command(lane, &task.prompt, &workspace.path);
487
488
    let mut child = match Command::new(&command)
489
        .args(&args)
490
        .current_dir(&workspace.path)
491
        // No terminal, so a harness that would prompt gets end-of-file rather
492
        // than a wait nobody can see.
493
        .stdin(Stdio::null())
494
        .stdout(Stdio::piped())
495
        .stderr(Stdio::piped())
496
        .process_group(0)
497
        .spawn()
498
    {
499
        Ok(child) => child,
500
        Err(error) => {
501
            let why = if error.kind() == std::io::ErrorKind::NotFound {
502
                format!("the `{command}` command is not on PATH")
503
            } else {
504
                format!("the `{command}` command would not start: {error}")
505
            };
506
            let _ = events.send(ChildEvent::Started {
507
                id,
508
                lane: lane.label(),
509
                workspace: workspace.describe(),
510
                pid: None,
511
            });
512
            return Err(ChildFailure { why, pid: None });
513
        }
514
    };
515
516
    let pid = child.id();
517
    let _ = events.send(ChildEvent::Started {
518
        id,
519
        lane: lane.label(),
520
        workspace: workspace.describe(),
521
        pid,
522
    });
127 523
128
    match cmd.output().await {
129
        Ok(res) => (res.status.success(), String::from_utf8_lossy(&res.stdout).trim().to_string()),
130
        Err(e) => (false, format!("Failed to spawn opencode: {}", e)),
524
    // Both streams are drained by their own task and merged here, so neither
525
    // can fill its pipe and stall the child while the other is being read.
526
    let (lines_tx, mut lines_rx) = mpsc::unbounded_channel::<(bool, String)>();
527
    if let Some(stdout) = child.stdout.take() {
528
        let tx = lines_tx.clone();
529
        tokio::spawn(async move {
530
            let mut reader = BufReader::new(stdout).lines();
531
            while let Ok(Some(line)) = reader.next_line().await {
532
                if tx.send((true, line)).is_err() {
533
                    break;
534
                }
535
            }
536
        });
537
    }
538
    if let Some(stderr) = child.stderr.take() {
539
        let tx = lines_tx.clone();
540
        tokio::spawn(async move {
541
            let mut reader = BufReader::new(stderr).lines();
542
            while let Ok(Some(line)) = reader.next_line().await {
543
                if tx.send((false, line)).is_err() {
544
                    break;
545
                }
546
            }
547
        });
131 548
    }
549
    drop(lines_tx);
550
551
    let mut harvest = Harvest::new(lane);
552
    let mut stopped = false;
553
554
    loop {
555
        tokio::select! {
556
            biased;
557
            _ = cancel.changed() => {
558
                stopped = true;
559
                stop_tree(&mut child).await;
560
                break;
561
            }
562
            line = lines_rx.recv() => {
563
                match line {
564
                    Some((from_stdout, line)) => {
565
                        if from_stdout {
566
                            // A harness with an event stream is forwarded as
567
                            // what it said, not as its wire format: five lines
568
                            // of `{"type":"assistant","message":{"content":…`
569
                            // is not a reader watching a child work.
570
                            for rendered in harvest.take(&line) {
571
                                let _ = events.send(match rendered {
572
                                    Rendered::Text(text) => ChildEvent::Output { id, text },
573
                                    Rendered::Note(text) => ChildEvent::Activity { id, text },
574
                                });
575
                            }
576
                        } else {
577
                            let _ = events.send(ChildEvent::Activity { id, text: line.clone() });
578
                            harvest.note_stderr(&line);
579
                        }
580
                    }
581
                    // Both streams are closed, so the child has finished
582
                    // writing even if it has not yet been reaped.
583
                    None => break,
584
                }
585
            }
586
        }
587
    }
588
589
    if stopped {
590
        return Err(ChildFailure {
591
            why: "stopped before finishing".to_string(),
592
            pid,
593
        });
594
    }
595
596
    let status = match child.wait().await {
597
        Ok(status) => status,
598
        Err(error) => {
599
            return Err(ChildFailure {
600
                why: format!("the `{command}` child could not be reaped: {error}"),
601
                pid,
602
            })
603
        }
604
    };
605
606
    if !status.success() {
607
        let code = status.code().unwrap_or(-1);
608
        return Err(ChildFailure {
609
            why: format!(
610
                "the `{command}` child exited with code {code}.\n\n{}",
611
                harvest.tail()
612
            ),
613
            pid,
614
        });
615
    }
616
617
    if let Some(reported) = harvest.reported_error() {
618
        return Err(ChildFailure {
619
            why: format!("the `{command}` child reported an error: {reported}"),
620
            pid,
621
        });
622
    }
623
624
    Ok(ChildAnswer {
625
        text: harvest.answer(),
626
        pid,
627
    })
132 628
}
133 629
134
async fn run_devin_cli(prompt: &str) -> (bool, String) {
135
    let mut cmd = Command::new("devin");
136
    cmd.args(["--prompt", prompt]);
137
    cmd.stdout(Stdio::piped());
138
    cmd.stderr(Stdio::piped());
630
/// The binary a lane runs, or the stand-in a test points it at.
631
///
632
/// The TypeScript harnesses each take a `command` for the same reason: a test
633
/// that cannot substitute the agent can only assert against a real one, which
634
/// means it either costs money or does not run.
635
pub fn harness_binary(lane: &ChildLane) -> String {
636
    let (variable, default) = match lane {
637
        ChildLane::Claude => ("OA_CHILD_CLAUDE", "claude"),
638
        ChildLane::Codex => ("OA_CHILD_CODEX", "codex"),
639
        ChildLane::Opencode { .. } => ("OA_CHILD_OPENCODE", "opencode"),
640
        ChildLane::Devin => ("OA_CHILD_DEVIN", "devin"),
641
        ChildLane::OxAlpha => ("", ""),
642
    };
643
    std::env::var(variable)
644
        .ok()
645
        .filter(|value| !value.trim().is_empty())
646
        .unwrap_or_else(|| default.to_string())
647
}
139 648
140
    match cmd.output().await {
141
        Ok(res) => (res.status.success(), String::from_utf8_lossy(&res.stdout).trim().to_string()),
142
        Err(e) => (false, format!("Failed to spawn devin: {}", e)),
649
/// The binary and arguments each CLI lane runs, following `coder-delegate.ts`.
650
fn harness_command(lane: &ChildLane, prompt: &str, cwd: &std::path::Path) -> (String, Vec<String>) {
651
    match lane {
652
        ChildLane::Claude => (
653
            harness_binary(lane),
654
            vec![
655
                "-p".to_string(),
656
                prompt.to_string(),
657
                "--output-format".to_string(),
658
                "stream-json".to_string(),
659
                // `stream-json` requires it.
660
                "--verbose".to_string(),
661
                // A delegated child has nobody to ask.
662
                "--permission-mode".to_string(),
663
                "acceptEdits".to_string(),
664
            ],
665
        ),
666
        ChildLane::Codex => (
667
            harness_binary(lane),
668
            vec![
669
                "exec".to_string(),
670
                "--json".to_string(),
671
                // A child's worktree is a checkout but not one Codex has been
672
                // told to trust, and without this it refuses before it starts.
673
                "--skip-git-repo-check".to_string(),
674
                // The child may edit the checkout it was pointed at and
675
                // nothing outside it.
676
                "--sandbox".to_string(),
677
                "workspace-write".to_string(),
678
                prompt.to_string(),
679
            ],
680
        ),
681
        ChildLane::Opencode { model } => (
682
            harness_binary(lane),
683
            vec![
684
                "run".to_string(),
685
                "--format".to_string(),
686
                "json".to_string(),
687
                "--model".to_string(),
688
                model.clone(),
689
                "--dir".to_string(),
690
                cwd.to_string_lossy().to_string(),
691
                prompt.to_string(),
692
            ],
693
        ),
694
        // Handled by their own runners; unreachable through this function.
695
        ChildLane::OxAlpha => ("".to_string(), Vec::new()),
696
        ChildLane::Devin => (harness_binary(lane), vec!["acp".to_string()]),
143 697
    }
144 698
}
145 699
146
pub async fn run_delegation(args: CoderArgs, user_token: Option<String>) -> Result<(), Box<dyn std::error::Error>> {
147
    let count = args.count.max(1);
148
    let lane = args.lane.unwrap_or_else(|| "ox-alpha".to_string());
149
    let prompt = args.prompt.unwrap_or_else(|| "Analyze workspace and run tests".to_string());
700
/// Pulls a child's answer out of whatever its harness prints.
701
///
702
/// A harness with a JSON event stream is read as one; anything that is not
703
/// JSON, or is JSON of a shape this does not know, is kept as text. So a
704
/// harness that changes its schema degrades to "the tail of what it printed"
705
/// rather than to an empty answer that reads like a child with nothing to say.
706
/// What one line of a child's output is: its answer, or what it is doing.
707
enum Rendered {
708
    Text(String),
709
    Note(String),
710
}
150 711
151
    println!("Starting parallel delegation across {} child workers on lane {}...", count, lane);
152
    let supervisor = DelegationSupervisor::new(count, &lane, user_token);
153
    let results = supervisor.dispatch(&prompt).await;
712
struct Harvest {
713
    lane: ChildLane,
714
    assistant: String,
715
    result: Option<String>,
716
    error: Option<String>,
717
    plain: String,
718
    stderr_tail: String,
719
}
154 720
155
    for res in &results {
156
        println!("Child {}: status={}, duration={}ms, output={}", res.id, if res.success { "ok" } else { "err" }, res.duration_ms, res.output);
721
impl Harvest {
722
    fn new(lane: &ChildLane) -> Self {
723
        Self {
724
            lane: lane.clone(),
725
            assistant: String::new(),
726
            result: None,
727
            error: None,
728
            plain: String::new(),
729
            stderr_tail: String::new(),
730
        }
731
    }
732
733
    /// Read one line and say what the reader should see of it.
734
    fn take(&mut self, line: &str) -> Vec<Rendered> {
735
        let trimmed = line.trim();
736
        if trimmed.is_empty() {
737
            return Vec::new();
738
        }
739
        // Not the harness's event stream. Whatever it is, it is what the child
740
        // printed, so it is passed through as written.
741
        let Ok(event) = serde_json::from_str::<serde_json::Value>(trimmed) else {
742
            push_bounded(&mut self.plain, line);
743
            return vec![Rendered::Text(format!("{line}\n"))];
744
        };
745
746
        let mut shown: Vec<Rendered> = Vec::new();
747
748
        // Claude Code print mode: `assistant` messages carry the text and the
749
        // last event carries the whole result.
750
        if let Some(text) = event
751
            .get("message")
752
            .and_then(|m| m.get("content"))
753
            .and_then(|c| c.as_array())
754
        {
755
            for part in text {
756
                if let Some(said) = part.get("text").and_then(|v| v.as_str()) {
757
                    self.assistant.push_str(said);
758
                    shown.push(Rendered::Text(said.to_string()));
759
                }
760
                // A tool call is what the child is doing, not what it said.
761
                if part.get("type").and_then(|v| v.as_str()) == Some("tool_use") {
762
                    shown.push(Rendered::Note(format!(
763
                        "tool {}",
764
                        part.get("name").and_then(|v| v.as_str()).unwrap_or("?")
765
                    )));
766
                }
767
            }
768
        }
769
        if let Some(result) = event.get("result").and_then(|v| v.as_str()) {
770
            self.result = Some(result.to_string());
771
        }
772
        if event.get("is_error").and_then(|v| v.as_bool()) == Some(true) {
773
            self.error = Some(
774
                event
775
                    .get("result")
776
                    .and_then(|v| v.as_str())
777
                    .unwrap_or("the harness reported an error")
778
                    .to_string(),
779
            );
780
        }
781
782
        // Codex exec: the final assistant message arrives as an item.
783
        if let Some(item) = event.get("item") {
784
            match item.get("type").and_then(|v| v.as_str()) {
785
                Some("agent_message") => {
786
                    if let Some(said) = item.get("text").and_then(|v| v.as_str()) {
787
                        self.assistant.push_str(said);
788
                        shown.push(Rendered::Text(format!("{said}\n")));
789
                    }
790
                }
791
                Some("command_execution") => {
792
                    if let Some(command) = item.get("command").and_then(|v| v.as_str()) {
793
                        shown.push(Rendered::Note(format!("ran {command}")));
794
                    }
795
                }
796
                _ => {}
797
            }
798
        }
799
        // Codex's older wire shape, and opencode's.
800
        if let Some(msg) = event.get("msg") {
801
            if let Some(said) = msg.get("message").and_then(|v| v.as_str()) {
802
                self.assistant.push_str(said);
803
                shown.push(Rendered::Text(format!("{said}\n")));
804
            }
805
            if msg.get("type").and_then(|v| v.as_str()) == Some("error") {
806
                if let Some(said) = msg.get("message").and_then(|v| v.as_str()) {
807
                    self.error = Some(said.to_string());
808
                }
809
            }
810
        }
811
        if let Some(said) = event
812
            .get("parts")
813
            .and_then(|p| p.as_array())
814
            .map(|parts| {
815
                parts
816
                    .iter()
817
                    .filter_map(|part| part.get("text").and_then(|v| v.as_str()))
818
                    .collect::<Vec<_>>()
819
                    .join("")
820
            })
821
            .filter(|said| !said.is_empty())
822
        {
823
            self.assistant.push_str(&said);
824
            shown.push(Rendered::Text(said));
825
        }
826
827
        if shown.is_empty() {
828
            // A known wire format, an event this does not render. Named rather
829
            // than dropped, so a silent stretch is a silent child and not a
830
            // parser looking the other way.
831
            if let Some(kind) = event
832
                .get("type")
833
                .or_else(|| event.get("msg").and_then(|m| m.get("type")))
834
                .and_then(|v| v.as_str())
835
            {
836
                if kind != "result" {
837
                    shown.push(Rendered::Note(kind.to_string()));
838
                }
839
            }
840
        }
841
        shown
842
    }
843
844
    fn note_stderr(&mut self, line: &str) {
845
        push_bounded(&mut self.stderr_tail, line);
846
    }
847
848
    fn reported_error(&self) -> Option<&str> {
849
        self.error.as_deref()
850
    }
851
852
    fn answer(&self) -> String {
853
        let said = self
854
            .result
855
            .clone()
856
            .filter(|text| !text.trim().is_empty())
857
            .or_else(|| Some(self.assistant.clone()).filter(|text| !text.trim().is_empty()))
858
            .or_else(|| Some(self.plain.clone()).filter(|text| !text.trim().is_empty()));
859
        match said {
860
            Some(text) => text.trim().to_string(),
861
            None => format!(
862
                "The {} child finished and printed no answer this harness could read.",
863
                match self.lane {
864
                    ChildLane::Claude => "claude",
865
                    ChildLane::Codex => "codex",
866
                    ChildLane::Opencode { .. } => "opencode",
867
                    ChildLane::Devin => "devin",
868
                    ChildLane::OxAlpha => "ox-alpha",
869
                }
870
            ),
871
        }
872
    }
873
874
    fn tail(&self) -> String {
875
        let mut both = String::new();
876
        if !self.plain.trim().is_empty() {
877
            both.push_str(self.plain.trim());
878
        }
879
        if !self.stderr_tail.trim().is_empty() {
880
            if !both.is_empty() {
881
                both.push('\n');
882
            }
883
            both.push_str(self.stderr_tail.trim());
884
        }
885
        if both.is_empty() {
886
            "The child printed nothing.".to_string()
887
        } else {
888
            both
889
        }
890
    }
891
}
892
893
/// Keep the last [`CHILD_RESULT_LIMIT`] characters, so an hour of build output
894
/// cannot grow without bound in memory.
895
fn push_bounded(buffer: &mut String, line: &str) {
896
    buffer.push_str(line);
897
    buffer.push('\n');
898
    if buffer.len() > CHILD_RESULT_LIMIT * 2 {
899
        let keep = buffer.len() - CHILD_RESULT_LIMIT;
900
        let at = buffer
901
            .char_indices()
902
            .map(|(at, _)| at)
903
            .find(|at| *at >= keep)
904
            .unwrap_or(0);
905
        buffer.drain(..at);
906
    }
907
}
908
909
/// A child's answer, cut to what the reader is shown.
910
///
911
/// The cut says its own size. A child that reported ten findings and was shown
912
/// as three reads exactly like a child that found three.
913
fn clip(text: &str) -> String {
914
    if text.len() <= CHILD_RESULT_LIMIT {
915
        return text.to_string();
916
    }
917
    let mut at = CHILD_RESULT_LIMIT;
918
    while at > 0 && !text.is_char_boundary(at) {
919
        at -= 1;
920
    }
921
    format!(
922
        "{}\n…[{} of {} characters cut from the end of this child's answer]",
923
        &text[..at],
924
        text.len() - at,
925
        text.len()
926
    )
927
}
928
929
/// Turns the event stream into prefixed lines on standard output.
930
///
931
/// A child's output arrives in whatever pieces the harness or the model
932
/// produced it in, which for a streamed model is a few characters at a time.
933
/// Each child gets its own buffer so a prefix is printed once per line rather
934
/// than once per chunk, and two children writing at once do not interleave
935
/// mid-word.
936
struct Printer {
937
    pending: std::collections::BTreeMap<usize, String>,
938
}
939
940
impl Printer {
941
    fn new() -> Self {
942
        Self {
943
            pending: std::collections::BTreeMap::new(),
944
        }
945
    }
946
947
    fn feed(&mut self, id: usize, text: &str) {
948
        let buffer = self.pending.entry(id).or_default();
949
        buffer.push_str(text);
950
        while let Some(at) = buffer.find('\n') {
951
            let line: String = buffer.drain(..=at).collect();
952
            println!("[child {id}] {}", line.trim_end_matches('\n'));
953
        }
954
    }
955
956
    fn flush(&mut self, id: usize) {
957
        if let Some(buffer) = self.pending.get_mut(&id) {
958
            if !buffer.trim().is_empty() {
959
                println!("[child {id}] {}", buffer.trim_end());
960
            }
961
            buffer.clear();
962
        }
963
    }
964
}
965
966
/// `oa coder --delegate`.
967
pub async fn run_delegation(
968
    args: CoderArgs,
969
    user_token: Option<String>,
970
) -> Result<(), Box<dyn std::error::Error>> {
971
    let requested = args.count.max(1);
972
    if requested > MAX_DELEGATE_COUNT {
973
        fail(&format!(
974
            "{requested} children were asked for and this command runs at most {MAX_DELEGATE_COUNT}."
975
        ));
976
    }
977
    let lane_name = args.lane.clone().unwrap_or_else(|| "ox-alpha".to_string());
978
    let prompt = args
979
        .prompt
980
        .clone()
981
        .unwrap_or_else(|| "Analyze workspace and run tests".to_string());
982
983
    if !ChildLane::known(&lane_name) {
984
        fail(&format!(
985
            "there is no `{lane_name}` lane. This command runs children on: ox-alpha, gemini, opencode/<model>, devin, claude, codex."
986
        ));
987
    }
988
    let lane = ChildLane::parse(&lane_name);
989
990
    let isolation = match args.isolation.as_deref() {
991
        None => Isolation::Worktree,
992
        Some(named) => match Isolation::parse(named) {
993
            Some(isolation) => isolation,
994
            None => fail(&format!(
995
                "`{named}` is not an isolation this command knows. Use worktree, directory, or none."
996
            )),
997
        },
998
    };
999
1000
    let supervisor = DelegationSupervisor::new(requested, &lane_name, user_token)
1001
        .with_isolation(isolation)
1002
        .with_max_parallel(args.max_parallel.unwrap_or(requested))
1003
        .keeping_workspaces(args.keep_workspaces);
1004
1005
    println!(
1006
        "Delegating to {} {} on {}, {} at a time, isolation: {}.",
1007
        supervisor.count,
1008
        if supervisor.count == 1 { "child" } else { "children" },
1009
        lane.label(),
1010
        supervisor.max_parallel,
1011
        isolation.name(),
1012
    );
1013
1014
    // `ctrl+c` is the only stop signal a running fan-out has. Without it a
1015
    // reader who changed their mind had to kill the terminal, and the
1016
    // children — which are their own process groups — carried on spending.
1017
    let (stop, cancel) = watch::channel(false);
1018
    let interrupt = tokio::spawn(async move {
1019
        if tokio::signal::ctrl_c().await.is_ok() {
1020
            eprintln!("\nStopping the fan-out; children are being signalled.");
1021
            let _ = stop.send(true);
1022
        }
1023
    });
1024
1025
    let (events, mut incoming) = mpsc::unbounded_channel();
1026
    let printing = tokio::spawn(async move {
1027
        let mut printer = Printer::new();
1028
        while let Some(event) = incoming.recv().await {
1029
            match event {
1030
                ChildEvent::Started {
1031
                    id,
1032
                    lane,
1033
                    workspace,
1034
                    pid,
1035
                } => {
1036
                    println!(
1037
                        "[child {id}] started on {lane} in {workspace}{}",
1038
                        match pid {
1039
                            Some(pid) => format!(" as pid {pid}"),
1040
                            None => " in this process".to_string(),
1041
                        }
1042
                    );
1043
                }
1044
                ChildEvent::Output { id, text } => printer.feed(id, &text),
1045
                ChildEvent::Activity { id, text } => {
1046
                    printer.flush(id);
1047
                    println!("[child {id}] · {text}");
1048
                }
1049
                ChildEvent::Finished(result) => {
1050
                    printer.flush(result.id);
1051
                    println!(
1052
                        "[child {}] {} after {}ms",
1053
                        result.id,
1054
                        if result.success { "finished" } else { "FAILED" },
1055
                        result.duration_ms
1056
                    );
1057
                }
1058
            }
1059
        }
1060
    });
1061
1062
    let results = match supervisor.dispatch_streaming(&prompt, events, cancel).await {
1063
        Ok(results) => results,
1064
        // No child ran at all, so there is nothing to report but the reason.
1065
        Err(error) => fail(&format!("no children were started: {error}")),
1066
    };
1067
    let _ = printing.await;
1068
    interrupt.abort();
1069
1070
    println!();
1071
    let succeeded = results.iter().filter(|result| result.success).count();
1072
    for result in &results {
1073
        println!(
1074
            "child {}: {} in {}ms{}{}",
1075
            result.id,
1076
            if result.success { "ok" } else { "failed" },
1077
            result.duration_ms,
1078
            match result.pid {
1079
                Some(pid) => format!(", pid {pid}"),
1080
                None => String::new(),
1081
            },
1082
            match &result.workspace {
1083
                Some(path) => format!(", in {}", path.display()),
1084
                None => String::new(),
1085
            }
1086
        );
1087
        if let Some(why) = &result.failure {
1088
            println!("  {why}");
1089
        }
1090
    }
1091
    println!(
1092
        "{succeeded} of {} {} completed on {}.",
1093
        results.len(),
1094
        if results.len() == 1 { "child" } else { "children" },
1095
        lane.label()
1096
    );
1097
1098
    if succeeded < results.len() {
1099
        // A fan-out that lost a child is not a command that worked. This used
1100
        // to print `2/2 children succeeded` and exit zero whatever happened.
1101
        // Exit 1 rather than the 2 an input error gets: the command was asked
1102
        // for correctly and the work is what did not finish.
1103
        eprintln!(
1104
            "oa: {} of {} children did not finish.",
1105
            results.len() - succeeded,
1106
            results.len()
1107
        );
1108
        std::process::exit(1);
157 1109
    }
158
    println!("Delegation fan-out complete. {}/{} children succeeded.", results.iter().filter(|r| r.success).count(), results.len());
159 1110
    Ok(())
160 1111
}
1112
1113
/// The `delegate` tool's fan-out, rendered for a model to read.
1114
///
1115
/// Awaited rather than launched and forgotten: a model told "three children
1116
/// are running" has nothing to say next and will either invent their findings
1117
/// or ask the reader to wait.
1118
///
1119
/// Returned as a boxed `Send` future on purpose. The call graph is a cycle —
1120
/// a session runs a tool, the `delegate` tool starts a child, and the child is
1121
/// a session that runs tools — and the compiler cannot infer `Send` around a
1122
/// cycle: it asks whether this future is `Send` in order to answer whether it
1123
/// is `Send`. Naming the bound here is what breaks it.
1124
pub fn fanout_for_tool(
1125
    prompt: &str,
1126
    count: usize,
1127
    lane: &str,
1128
    user_token: Option<String>,
1129
) -> std::pin::Pin<Box<dyn std::future::Future<Output = String> + Send>> {
1130
    let prompt = prompt.to_string();
1131
    let lane = lane.to_string();
1132
    Box::pin(async move {
1133
    let prompt = prompt.as_str();
1134
    let lane = lane.as_str();
1135
    let count = count.clamp(1, MAX_DELEGATE_COUNT);
1136
    let supervisor = DelegationSupervisor::new(count, lane, user_token);
1137
    let results = supervisor.dispatch(prompt).await;
1138
1139
    let succeeded = results.iter().filter(|result| result.success).count();
1140
    let mut lines = vec![format!(
1141
        "{succeeded} of {} {} completed on {}.",
1142
        results.len(),
1143
        if results.len() == 1 { "child" } else { "children" },
1144
        ChildLane::parse(lane).label()
1145
    )];
1146
    lines.push(String::new());
1147
    for result in &results {
1148
        if result.success {
1149
            lines.push(format!(
1150
                "child {} completed in {}ms:\n{}",
1151
                result.id,
1152
                result.duration_ms,
1153
                if result.output.trim().is_empty() {
1154
                    "(no output)"
1155
                } else {
1156
                    result.output.trim()
1157
                }
1158
            ));
1159
        } else {
1160
            lines.push(format!("child {} failed: {}", result.id, result.output));
1161
        }
1162
    }
1163
    lines.join("\n")
1164
    })
1165
}
crates/openagents-cli/src/lib.rs modified +2

@@ -25,8 +25,10 @@ pub mod interactive;

25 25
pub mod memory_client;
26 26
pub mod repo;
27 27
pub mod runtime;
28
pub mod signals;
28 29
pub mod tools;
29 30
pub mod trace;
30 31
pub mod tracker;
31 32
pub mod tui;
32 33
pub mod update;
34
pub mod workspace;
crates/openagents-cli/src/signals.rs added +60

@@ -0,0 +1,60 @@

1
//! Stopping a child and everything the child started.
2
//!
3
//! A coding agent shells out. Killing only the agent leaves its build, its
4
//! test run, or its `sleep` behind with nothing left to stop them, and with a
5
//! fan-out that is one orphan per child every time a run is cancelled. So a
6
//! child is spawned into a process group of its own and the group is what gets
7
//! signalled.
8
//!
9
//! `SIGTERM` first, then `SIGKILL` after a grace period, so an agent that
10
//! writes a transcript on the way out gets to write it.
11
12
use std::time::Duration;
13
14
/// How long a stopped child has to leave on its own before it is killed.
15
pub const KILL_GRACE: Duration = Duration::from_secs(3);
16
17
/// Signal a process group. `pid` is the group leader, which is the child
18
/// itself because it was spawned with `process_group(0)`.
19
///
20
/// Returns whether the signal was delivered. A group that has already exited
21
/// reports `false`, which is the outcome that was wanted.
22
#[cfg(unix)]
23
pub fn signal_group(pid: u32, signal: i32) -> bool {
24
    // Negative pid means "the group led by pid" — the whole point of the
25
    // exercise. Safe because the only inputs are an integer we were handed by
26
    // the spawn and a constant from libc.
27
    unsafe { libc::kill(-(pid as i32), signal) == 0 }
28
}
29
30
#[cfg(not(unix))]
31
pub fn signal_group(_pid: u32, _signal: i32) -> bool {
32
    false
33
}
34
35
#[cfg(unix)]
36
pub const SIGTERM: i32 = libc::SIGTERM;
37
#[cfg(unix)]
38
pub const SIGKILL: i32 = libc::SIGKILL;
39
40
#[cfg(not(unix))]
41
pub const SIGTERM: i32 = 15;
42
#[cfg(not(unix))]
43
pub const SIGKILL: i32 = 9;
44
45
/// Ask a child's whole group to stop, then insist.
46
///
47
/// The direct child is killed as well as the group: on a platform where the
48
/// group signal does not land, the agent itself still goes.
49
pub async fn stop_tree(child: &mut tokio::process::Child) {
50
    let Some(pid) = child.id() else {
51
        return;
52
    };
53
    signal_group(pid, SIGTERM);
54
    tokio::select! {
55
        _ = child.wait() => return,
56
        _ = tokio::time::sleep(KILL_GRACE) => {}
57
    }
58
    signal_group(pid, SIGKILL);
59
    let _ = child.start_kill();
60
}
crates/openagents-cli/src/tools.rs modified +155 -5

@@ -1,5 +1,26 @@

1
//! Real tool execution runtime for OpenAgents Coder
2
//! Implements `shell`, `skill`, `openagents`, `capability`, and delegation hooks
1
//! The tools a session declares to the model, and what running them does.
2
//!
3
//! Four tools: `shell`, `skill`, `openagents`, and `delegate`. Each is
4
//! declared to the model and each has an implementation in
5
//! [`HarnessToolRegistry::execute_tool`]; the list and the match arms are the
6
//! same four, which is the only property that keeps a declared tool from being
7
//! a promise nothing keeps.
8
//!
9
//! `capability` — the standing tool that searches the local plugin catalog and
10
//! loads a digest-pinned WebAssembly plugin — is **not** implemented here and
11
//! is **not** declared. This module's own header used to claim it was. It is
12
//! not a matter of wiring: the plugins in `plugins/` are WebAssembly artifacts
13
//! against a bespoke `packet-v0` ABI, with per-manifest mounts, host
14
//! allowlists, memory ceilings, and timeouts to enforce, and this crate has no
15
//! WebAssembly runtime to enforce them with. That is the "WASM capability
16
//! runtime integration" half of OpenAgentsInc/openagents#71, and porting it
17
//! means bringing a wasm engine into this binary. Until that happens the
18
//! honest state is a tool that is absent rather than one that is advertised
19
//! and refuses.
20
//!
21
//! The tool runtime is the client's. The inference proxy forwards the
22
//! declarations and returns the calls the model asks for; nothing runs
23
//! server-side.
3 24
4 25
use serde::{Deserialize, Serialize};
5 26
use std::collections::HashMap;

@@ -42,17 +63,51 @@ pub struct SkillInfo {

42 63
    pub body: String,
43 64
}
44 65
66
/// What the `delegate` tool is allowed to start.
67
///
68
/// Present on the session the reader is talking to and absent on the children
69
/// it starts. A fan-out whose children fan out has no ceiling: three children
70
/// each starting three is nine agents on one grant, and none of them told the
71
/// reader.
72
#[derive(Debug, Clone)]
73
pub struct DelegationGate {
74
    /// The lane children run on.
75
    pub lane: String,
76
    /// The credential children spend against.
77
    pub user_token: Option<String>,
78
    /// The most children one call may start.
79
    pub max_count: usize,
80
}
81
45 82
pub struct HarnessToolRegistry {
46 83
    pub cwd: PathBuf,
47 84
    pub skills: HashMap<String, SkillInfo>,
85
    /// `None` on a delegated child, so it cannot delegate further.
86
    pub delegation: Option<DelegationGate>,
48 87
}
49 88
50 89
impl HarnessToolRegistry {
51 90
    pub fn new(cwd: Option<PathBuf>) -> Self {
91
        Self::build(cwd, None)
92
    }
93
94
    /// The registry a session gets when it may start children.
95
    pub fn with_delegation(cwd: Option<PathBuf>, gate: DelegationGate) -> Self {
96
        Self::build(cwd, Some(gate))
97
    }
98
99
    /// The registry a delegated child gets: rooted at the child's own
100
    /// directory, and with no `delegate` tool.
101
    pub fn child(cwd: Option<PathBuf>) -> Self {
102
        Self::build(cwd, None)
103
    }
104
105
    fn build(cwd: Option<PathBuf>, delegation: Option<DelegationGate>) -> Self {
52 106
        let root = cwd.unwrap_or_else(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")));
53 107
        let mut registry = Self {
54 108
            cwd: root,
55 109
            skills: HashMap::new(),
110
            delegation,
56 111
        };
57 112
        registry.load_local_skills();
58 113
        registry

@@ -102,10 +157,16 @@ impl HarnessToolRegistry {

102 157
            skill_list.push_str(&format!("\n- `{}`: {}", name, info.description));
103 158
        }
104 159
105
        vec![
160
        let mut tools = vec![
106 161
            ToolDefinition {
107 162
                name: "shell".to_string(),
108
                description: "Run a shell command on this machine. Returns combined stdout and stderr with exit code. Paths are relative to the working directory. Batch independent commands with &&.".to_string(),
163
                description: format!(
164
                    "Run a shell command on this machine. The working directory is {}, so paths are \
165
                    relative to it and you do not need to ask where you are. Returns combined stdout \
166
                    and stderr with the exit code. Batch independent commands into one call with && \
167
                    instead of one call each: every call replays the conversation so far.",
168
                    self.cwd.display()
169
                ),
109 170
                parameters: serde_json::json!({
110 171
                    "type": "object",
111 172
                    "properties": {

@@ -141,7 +202,49 @@ impl HarnessToolRegistry {

141 202
                    "required": ["args"]
142 203
                }),
143 204
            },
144
        ]
205
        ];
206
207
        // Declared only where it can be run. A child's registry has no gate,
208
        // so a child neither sees the tool nor can call it.
209
        if let Some(gate) = &self.delegation {
210
            tools.push(ToolDefinition {
211
                name: "delegate".to_string(),
212
                description: format!(
213
                    "Run one prompt on independent child coding agents in parallel and return what \
214
                    each one found or did. Use it when work splits into parts that do not depend on \
215
                    each other: several files to change the same way, several hypotheses to check, \
216
                    several tests to run down. Each child is a full coding agent with its own shell \
217
                    tool, working in a git worktree of its own so children cannot overwrite each \
218
                    other, and it starts with no context from this conversation and cannot ask \
219
                    questions — so the prompt has to be self-contained. Every child runs the same \
220
                    prompt and each is told separately which number it is, so write the prompt for \
221
                    whichever child reads it: say \"read the file at your own number\" rather than \
222
                    naming one child. Children run on {} and on this session's budget. Prefer one \
223
                    call with a count over several calls, and prefer `shell` over this for a single \
224
                    command — a child agent is for work worth a whole agent, not one line of \
225
                    output. At most {} children.",
226
                    gate.lane, gate.max_count
227
                ),
228
                parameters: serde_json::json!({
229
                    "type": "object",
230
                    "properties": {
231
                        "prompt": {
232
                            "type": "string",
233
                            "description": "The complete, self-contained instruction every child performs. Name the files, the command, and what to report back."
234
                        },
235
                        "count": {
236
                            "type": "integer",
237
                            "minimum": 1,
238
                            "maximum": gate.max_count,
239
                            "description": "How many children run this prompt. Defaults to 1."
240
                        }
241
                    },
242
                    "required": ["prompt"]
243
                }),
244
            });
245
        }
246
247
        tools
145 248
    }
146 249
147 250
    pub async fn execute_tool(&self, call: &ToolCall) -> ToolOutput {

@@ -197,6 +300,53 @@ impl HarnessToolRegistry {

197 300
                    is_error: false,
198 301
                }
199 302
            }
303
            "delegate" => {
304
                let Some(gate) = &self.delegation else {
305
                    // Reachable only if a model invents the name, since the
306
                    // tool is not declared without a gate.
307
                    return ToolOutput {
308
                        call_id: call.id.clone(),
309
                        output: "This session cannot start child agents.".to_string(),
310
                        is_error: true,
311
                    };
312
                };
313
314
                let prompt = call
315
                    .arguments
316
                    .get("prompt")
317
                    .and_then(|v| v.as_str())
318
                    .unwrap_or("")
319
                    .trim()
320
                    .to_string();
321
                if prompt.is_empty() {
322
                    return ToolOutput {
323
                        call_id: call.id.clone(),
324
                        output: "No children were started: `prompt` is required and must say what the child does.".to_string(),
325
                        is_error: true,
326
                    };
327
                }
328
329
                let count = call
330
                    .arguments
331
                    .get("count")
332
                    .and_then(|v| v.as_u64())
333
                    .unwrap_or(1)
334
                    .clamp(1, gate.max_count as u64) as usize;
335
336
                let report = crate::delegate::fanout_for_tool(
337
                    &prompt,
338
                    count,
339
                    &gate.lane,
340
                    gate.user_token.clone(),
341
                )
342
                .await;
343
344
                ToolOutput {
345
                    call_id: call.id.clone(),
346
                    output: report,
347
                    is_error: false,
348
                }
349
            }
200 350
            _ => ToolOutput {
201 351
                call_id: call.id.clone(),
202 352
                output: format!("Unknown tool: {}", call.name),
crates/openagents-cli/src/workspace.rs added +274

@@ -0,0 +1,274 @@

1
//! Where a delegated child works.
2
//!
3
//! Children ran in the parent's directory. Two of them told to edit the same
4
//! file edited the same file, and a fan-out asked to try three approaches
5
//! produced whichever one finished last. `ChildWorkerTask::worktree_path` was
6
//! hardcoded `None` and nothing read it.
7
//!
8
//! A child now gets a directory of its own. In a git checkout that is a
9
//! detached worktree of `HEAD`, which is the isolation
10
//! OpenAgentsInc/openagents#70 asks for: the child has the whole tree, its own
11
//! index, and its own branchless checkout, so what it writes is visible on
12
//! disk and cannot collide with a sibling. Outside a checkout it is a plain
13
//! empty directory. Either can be turned off with `--isolation none`, which is
14
//! what the TypeScript CLI does today.
15
//!
16
//! The worktrees are laid out under the system temporary directory rather than
17
//! inside the repository, so a fan-out never leaves untracked directories in
18
//! the tree the reader is working in.
19
20
use std::path::{Path, PathBuf};
21
use std::sync::atomic::{AtomicU64, Ordering};
22
use std::process::Stdio;
23
use std::time::{SystemTime, UNIX_EPOCH};
24
25
use tokio::process::Command;
26
27
/// How many plans this process has made, so no two share a directory.
28
static PLANS: AtomicU64 = AtomicU64::new(0);
29
30
/// How much of a directory a child gets to itself.
31
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
32
pub enum Isolation {
33
    /// A detached `git worktree` of `HEAD`, one per child.
34
    Worktree,
35
    /// An empty directory, one per child. Used outside a git checkout.
36
    Directory,
37
    /// The parent's own working directory, shared by every child.
38
    None,
39
}
40
41
impl Isolation {
42
    pub fn parse(name: &str) -> Option<Self> {
43
        match name.trim().to_lowercase().as_str() {
44
            "worktree" | "git" => Some(Isolation::Worktree),
45
            "dir" | "directory" | "temp" => Some(Isolation::Directory),
46
            "none" | "off" | "shared" => Some(Isolation::None),
47
            _ => None,
48
        }
49
    }
50
51
    pub fn name(self) -> &'static str {
52
        match self {
53
            Isolation::Worktree => "worktree",
54
            Isolation::Directory => "directory",
55
            Isolation::None => "none",
56
        }
57
    }
58
}
59
60
/// One child's working directory, and what it takes to put it back.
61
#[derive(Debug, Clone)]
62
pub struct ChildWorkspace {
63
    pub id: usize,
64
    pub path: PathBuf,
65
    pub kind: Isolation,
66
    /// The repository the worktree is registered in, when there is one to
67
    /// unregister it from.
68
    repo: Option<PathBuf>,
69
}
70
71
impl ChildWorkspace {
72
    /// A one-line account of where this child is working, for the header.
73
    pub fn describe(&self) -> String {
74
        match self.kind {
75
            Isolation::Worktree => format!("git worktree {}", self.path.display()),
76
            Isolation::Directory => format!("directory {}", self.path.display()),
77
            Isolation::None => format!("shared directory {}", self.path.display()),
78
        }
79
    }
80
81
    /// Unregister and delete the worktree.
82
    ///
83
    /// A failure here is reported and not raised: the fan-out's answers are
84
    /// worth more than a tidy temporary directory, and `git worktree prune`
85
    /// clears whatever is left behind.
86
    pub async fn release(&self) -> Option<String> {
87
        match (self.kind, &self.repo) {
88
            (Isolation::Worktree, Some(repo)) => {
89
                let done = Command::new("git")
90
                    .arg("-C")
91
                    .arg(repo)
92
                    .args(["worktree", "remove", "--force"])
93
                    .arg(&self.path)
94
                    .stdout(Stdio::null())
95
                    .stderr(Stdio::piped())
96
                    .output()
97
                    .await;
98
                match done {
99
                    Ok(out) if out.status.success() => None,
100
                    Ok(out) => Some(format!(
101
                        "could not remove the worktree at {}: {}",
102
                        self.path.display(),
103
                        String::from_utf8_lossy(&out.stderr).trim()
104
                    )),
105
                    Err(error) => Some(format!(
106
                        "could not remove the worktree at {}: {error}",
107
                        self.path.display()
108
                    )),
109
                }
110
            }
111
            (Isolation::Directory, _) => match tokio::fs::remove_dir_all(&self.path).await {
112
                Ok(()) => None,
113
                Err(error) => Some(format!(
114
                    "could not remove {}: {error}",
115
                    self.path.display()
116
                )),
117
            },
118
            _ => None,
119
        }
120
    }
121
}
122
123
/// Prepares one working directory per child, ahead of the fan-out.
124
///
125
/// Sequentially, on purpose. `git worktree add` takes the repository's lock,
126
/// and three of them started at once fail on `index.lock` rather than
127
/// producing three worktrees.
128
pub struct WorkspacePlan {
129
    cwd: PathBuf,
130
    repo: Option<PathBuf>,
131
    base: PathBuf,
132
    isolation: Isolation,
133
}
134
135
impl WorkspacePlan {
136
    /// Work out what isolation this directory can actually support.
137
    ///
138
    /// `Worktree` in a directory that is not a git checkout is not an error;
139
    /// it is a directory, and the header says which was used.
140
    pub async fn resolve(cwd: PathBuf, asked: Isolation) -> Self {
141
        let repo = if asked == Isolation::None {
142
            None
143
        } else {
144
            git_toplevel(&cwd).await
145
        };
146
        let isolation = match (asked, &repo) {
147
            (Isolation::Worktree, None) => Isolation::Directory,
148
            (other, _) => other,
149
        };
150
        // Unique per plan, not per millisecond. Two plans made in the same
151
        // millisecond of the same process shared a base directory, and each
152
        // child was `<base>/child-N` in both — so one plan's cleanup deleted
153
        // the other plan's first child while it was still working in it.
154
        let stamp = SystemTime::now()
155
            .duration_since(UNIX_EPOCH)
156
            .map(|d| d.as_nanos())
157
            .unwrap_or(0);
158
        let ordinal = PLANS.fetch_add(1, Ordering::Relaxed);
159
        let base = std::env::temp_dir().join(format!(
160
            "oa-delegate-{}-{stamp}-{ordinal}",
161
            std::process::id()
162
        ));
163
        Self {
164
            cwd,
165
            repo,
166
            base,
167
            isolation,
168
        }
169
    }
170
171
    pub fn isolation(&self) -> Isolation {
172
        self.isolation
173
    }
174
175
    /// Build every child's directory, in order.
176
    pub async fn prepare(&self, count: usize) -> Result<Vec<ChildWorkspace>, String> {
177
        let mut made: Vec<ChildWorkspace> = Vec::with_capacity(count);
178
        for id in 1..=count {
179
            match self.one(id).await {
180
                Ok(workspace) => made.push(workspace),
181
                Err(error) => {
182
                    // Half a fan-out's worktrees left on disk is worse than
183
                    // none, so what was built is torn down before reporting.
184
                    for built in &made {
185
                        let _ = built.release().await;
186
                    }
187
                    return Err(error);
188
                }
189
            }
190
        }
191
        Ok(made)
192
    }
193
194
    async fn one(&self, id: usize) -> Result<ChildWorkspace, String> {
195
        match self.isolation {
196
            Isolation::None => Ok(ChildWorkspace {
197
                id,
198
                path: self.cwd.clone(),
199
                kind: Isolation::None,
200
                repo: None,
201
            }),
202
            Isolation::Directory => {
203
                let path = self.base.join(format!("child-{id}"));
204
                tokio::fs::create_dir_all(&path)
205
                    .await
206
                    .map_err(|error| format!("could not create {}: {error}", path.display()))?;
207
                Ok(ChildWorkspace {
208
                    id,
209
                    path,
210
                    kind: Isolation::Directory,
211
                    repo: None,
212
                })
213
            }
214
            Isolation::Worktree => {
215
                let repo = self
216
                    .repo
217
                    .clone()
218
                    .ok_or_else(|| "no git checkout to take a worktree from".to_string())?;
219
                let path = self.base.join(format!("child-{id}"));
220
                tokio::fs::create_dir_all(&self.base)
221
                    .await
222
                    .map_err(|error| format!("could not create {}: {error}", self.base.display()))?;
223
224
                let out = Command::new("git")
225
                    .arg("-C")
226
                    .arg(&repo)
227
                    .args(["worktree", "add", "--detach"])
228
                    .arg(&path)
229
                    .arg("HEAD")
230
                    .stdout(Stdio::piped())
231
                    .stderr(Stdio::piped())
232
                    .output()
233
                    .await
234
                    .map_err(|error| format!("could not run git: {error}"))?;
235
236
                if !out.status.success() {
237
                    return Err(format!(
238
                        "git worktree add refused child {id}: {}",
239
                        String::from_utf8_lossy(&out.stderr).trim()
240
                    ));
241
                }
242
243
                Ok(ChildWorkspace {
244
                    id,
245
                    path,
246
                    kind: Isolation::Worktree,
247
                    repo: Some(repo),
248
                })
249
            }
250
        }
251
    }
252
}
253
254
/// The root of the checkout `cwd` is in, or nothing if it is not in one.
255
async fn git_toplevel(cwd: &Path) -> Option<PathBuf> {
256
    let out = Command::new("git")
257
        .arg("-C")
258
        .arg(cwd)
259
        .args(["rev-parse", "--show-toplevel"])
260
        .stdout(Stdio::piped())
261
        .stderr(Stdio::null())
262
        .output()
263
        .await
264
        .ok()?;
265
    if !out.status.success() {
266
        return None;
267
    }
268
    let path = String::from_utf8_lossy(&out.stdout).trim().to_string();
269
    if path.is_empty() {
270
        None
271
    } else {
272
        Some(PathBuf::from(path))
273
    }
274
}
crates/openagents-cli/tests/acp_test.rs added +242

@@ -0,0 +1,242 @@

1
//! The Agent Client Protocol client, against a server that speaks it.
2
//!
3
//! The stand-in is an ACP server in twenty lines of Python: it reads
4
//! newline-delimited JSON-RPC on stdin and writes it on stdout, exactly as
5
//! `devin acp` does. What is under test is this crate's half of that
6
//! conversation — the handshake order, the session, the permission answer the
7
//! child has nobody else to give, the update parsing, and the kill.
8
//!
9
//! The client this replaces built one `initialize` request as a struct and
10
//! never sent it anywhere.
11
12
use std::path::PathBuf;
13
use std::time::{Duration, Instant};
14
15
use openagents_cli::acp::{first_allow_option, AcpEvent, AcpFailure, AcpHarness, PermissionMode};
16
use tokio::sync::watch;
17
18
const SERVER: &str = r#"#!/usr/bin/env python3
19
import json, sys, time
20
21
def send(obj):
22
    sys.stdout.write(json.dumps(obj) + "\n")
23
    sys.stdout.flush()
24
25
sys.stderr.write("this line is not protocol and must be ignored\n")
26
sys.stdout.write("neither is this one\n")
27
sys.stdout.flush()
28
29
for line in sys.stdin:
30
    line = line.strip()
31
    if not line:
32
        continue
33
    message = json.loads(line)
34
    method = message.get("method")
35
    if method == "initialize":
36
        send({"jsonrpc": "2.0", "id": message["id"], "result": {"protocolVersion": 1}})
37
    elif method == "session/new":
38
        send({"jsonrpc": "2.0", "id": message["id"],
39
              "result": {"sessionId": "sess_" + message["params"]["cwd"].split("/")[-1]}})
40
    elif method == "session/set_mode":
41
        send({"jsonrpc": "2.0", "id": message["id"],
42
              "result": {"modeId": message["params"]["modeId"]}})
43
    elif method == "session/prompt":
44
        sid = message["params"]["sessionId"]
45
        send({"jsonrpc": "2.0", "method": "session/update", "params": {"sessionId": sid,
46
              "update": {"sessionUpdate": "tool_call", "toolCallId": "t1",
47
                         "kind": "execute", "title": "Ran ls"}}})
48
        # The agent asks. A delegated child has nobody to ask, so the client
49
        # answers, and the server proves it by only continuing once it has.
50
        send({"jsonrpc": "2.0", "id": 9001, "method": "session/request_permission",
51
              "params": {"sessionId": sid, "options": [
52
                  {"optionId": "reject-once", "kind": "reject_once"},
53
                  {"optionId": "allow-always", "kind": "allow_always"}]}})
54
        answer = json.loads(sys.stdin.readline())
55
        chosen = answer["result"]["outcome"]["optionId"]
56
        send({"jsonrpc": "2.0", "method": "session/update", "params": {"sessionId": sid,
57
              "update": {"sessionUpdate": "usage_update",
58
                         "_meta": {"cognition.ai/inputTokens": 120,
59
                                   "cognition.ai/outputTokens": 34}}}})
60
        for piece in ["the answer ", "in two ", "pieces, permitted by " + chosen]:
61
            send({"jsonrpc": "2.0", "method": "session/update", "params": {"sessionId": sid,
62
                  "update": {"sessionUpdate": "agent_message_chunk",
63
                             "content": {"type": "text", "text": piece}}}})
64
            time.sleep(0.2)
65
        send({"jsonrpc": "2.0", "id": message["id"], "result": {"stopReason": "end_turn"}})
66
    else:
67
        send({"jsonrpc": "2.0", "id": message.get("id", 0), "result": {}})
68
"#;
69
70
fn stand_in(name: &str, body: &str) -> PathBuf {
71
    let dir = std::env::temp_dir().join(format!("oa-acp-test-{}", std::process::id()));
72
    std::fs::create_dir_all(&dir).unwrap();
73
    let path = dir.join(name);
74
    std::fs::write(&path, body).unwrap();
75
    #[cfg(unix)]
76
    {
77
        use std::os::unix::fs::PermissionsExt;
78
        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
79
    }
80
    path
81
}
82
83
fn harness(name: &str, body: &str) -> AcpHarness {
84
    AcpHarness {
85
        command: stand_in(name, body).to_string_lossy().to_string(),
86
        args: Vec::new(),
87
        mode: Some(PermissionMode::Dangerous),
88
    }
89
}
90
91
/// A whole turn: handshake, session, permission, updates, answer.
92
#[tokio::test]
93
async fn a_turn_over_acp_streams_and_answers() {
94
    let harness = harness("acp-server", SERVER);
95
    let cwd = std::env::temp_dir().join(format!("oa-acp-cwd-{}", std::process::id()));
96
    std::fs::create_dir_all(&cwd).unwrap();
97
98
    let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::<(Instant, AcpEvent)>::new()));
99
    let sink = std::sync::Arc::clone(&seen);
100
    let (_stop, mut cancel) = watch::channel(false);
101
102
    let started = Instant::now();
103
    let answer = harness
104
        .run(
105
            "do the thing",
106
            &cwd,
107
            move |event| sink.lock().unwrap().push((Instant::now(), event)),
108
            &mut cancel,
109
        )
110
        .await
111
        .expect("the turn failed");
112
113
    // The permission answer is in the text, so the answer proves the client
114
    // both replied to the request and picked the option that allows.
115
    assert_eq!(answer, "the answer in two pieces, permitted by allow-always");
116
117
    let events = seen.lock().unwrap();
118
    let session = events
119
        .iter()
120
        .find_map(|(_, event)| match event {
121
            AcpEvent::Session { id } => Some(id.clone()),
122
            _ => None,
123
        })
124
        .expect("no session was reported");
125
    assert!(session.starts_with("sess_"), "{session}");
126
127
    assert!(
128
        events.iter().any(|(_, event)| matches!(
129
            event,
130
            AcpEvent::Tool { title, .. } if title == "Ran ls"
131
        )),
132
        "the tool call was not reported"
133
    );
134
    assert!(
135
        events.iter().any(|(_, event)| matches!(
136
            event,
137
            AcpEvent::Tokens { input: 120, output: 34 }
138
        )),
139
        "the token counts were not reported"
140
    );
141
142
    // Streamed, not delivered at the end: the first piece arrives well before
143
    // the last, because the server sleeps between them.
144
    let chunks: Vec<Instant> = events
145
        .iter()
146
        .filter_map(|(at, event)| match event {
147
            AcpEvent::Text { .. } => Some(*at),
148
            _ => None,
149
        })
150
        .collect();
151
    assert_eq!(chunks.len(), 3, "the answer did not arrive in pieces");
152
    let spread = chunks[2].duration_since(chunks[0]);
153
    assert!(
154
        spread >= Duration::from_millis(300),
155
        "three pieces arrived {spread:?} apart, which is one delivery and not three"
156
    );
157
    assert!(started.elapsed() < Duration::from_secs(30));
158
}
159
160
/// An agent that is not there is a failure that says so.
161
#[tokio::test]
162
async fn a_missing_agent_is_reported_as_missing() {
163
    let harness = AcpHarness {
164
        command: "/nonexistent/no-such-agent".to_string(),
165
        args: Vec::new(),
166
        mode: None,
167
    };
168
    let (_stop, mut cancel) = watch::channel(false);
169
    let failure = harness
170
        .run("hello", &std::env::temp_dir(), |_| {}, &mut cancel)
171
        .await
172
        .expect_err("a missing binary was reported as a finished turn");
173
    assert!(
174
        matches!(&failure, AcpFailure::Unstartable(why) if why.contains("not on PATH")),
175
        "{failure}"
176
    );
177
}
178
179
/// An agent that exits without answering is a failure, not an empty answer.
180
#[tokio::test]
181
async fn an_agent_that_exits_early_is_a_failure() {
182
    let harness = harness(
183
        "acp-quitter",
184
        "#!/bin/sh\necho 'not json'\nexit 0\n",
185
    );
186
    let (_stop, mut cancel) = watch::channel(false);
187
    let failure = harness
188
        .run("hello", &std::env::temp_dir(), |_| {}, &mut cancel)
189
        .await
190
        .expect_err("an agent that said nothing reported an answer");
191
    assert!(
192
        matches!(&failure, AcpFailure::Refused(why) if why.contains("exited before it answered")),
193
        "{failure}"
194
    );
195
}
196
197
/// Stopping the fan-out stops the agent, and does not wait for it.
198
#[tokio::test]
199
async fn a_cancelled_turn_stops_the_agent() {
200
    let harness = harness(
201
        "acp-slow",
202
        r#"#!/bin/sh
203
read -r line
204
echo '{"jsonrpc":"2.0","id":1,"result":{}}'
205
sleep 120
206
"#,
207
    );
208
    let (stop, mut cancel) = watch::channel(false);
209
    tokio::spawn(async move {
210
        tokio::time::sleep(Duration::from_millis(600)).await;
211
        let _ = stop.send(true);
212
    });
213
214
    let at = Instant::now();
215
    let failure = harness
216
        .run("hello", &std::env::temp_dir(), |_| {}, &mut cancel)
217
        .await
218
        .expect_err("a cancelled turn reported an answer");
219
    assert!(matches!(failure, AcpFailure::Cancelled), "{failure}");
220
    assert!(
221
        at.elapsed() < Duration::from_secs(20),
222
        "cancelling took {:?}; the stand-in sleeps for two minutes",
223
        at.elapsed()
224
    );
225
}
226
227
/// The permission answer prefers an option that allows.
228
#[test]
229
fn the_allowing_option_is_the_one_chosen() {
230
    let asked = serde_json::json!({"options": [
231
        {"optionId": "no", "kind": "reject_once"},
232
        {"optionId": "yes", "kind": "allow_always"}
233
    ]});
234
    assert_eq!(first_allow_option(&asked).as_deref(), Some("yes"));
235
236
    // Nothing that allows: the first named option, so the agent is answered
237
    // rather than left waiting.
238
    let none = serde_json::json!({"options": [{"optionId": "only", "kind": "reject_once"}]});
239
    assert_eq!(first_allow_option(&none).as_deref(), Some("only"));
240
241
    assert_eq!(first_allow_option(&serde_json::json!({})), None);
242
}
crates/openagents-cli/tests/cli_test.rs modified +3 -1

@@ -4,6 +4,7 @@ mod support;

4 4
mod tests {
5 5
    use openagents_cli::runtime::{CoderRuntimeSession, Lane};
6 6
    use openagents_cli::delegate::DelegationSupervisor;
7
    use openagents_cli::workspace::Isolation;
7 8
    use openagents_cli::tools::{HarnessToolRegistry, ToolCall};
8 9
    use openagents_cli::auth::CredentialStore;
9 10
    use openagents_cli::identity::{derive_seed_identity, SeedStore};

@@ -201,7 +202,8 @@ mod tests {

201 202
        let stub = crate::support::start(vec!["child ", "did the work"], None).await;
202 203
        std::env::set_var("OPENAGENTS_API_BASE", &stub.base);
203 204
204
        let supervisor = DelegationSupervisor::new(1, "ox-alpha", None);
205
        let supervisor = DelegationSupervisor::new(1, "ox-alpha", None)
206
            .with_isolation(Isolation::None);
205 207
        let results = supervisor.dispatch("test task").await;
206 208
207 209
        std::env::remove_var("OPENAGENTS_API_BASE");
crates/openagents-cli/tests/delegate_test.rs added +335

@@ -0,0 +1,335 @@

1
//! Delegation, against real child processes.
2
//!
3
//! Every claim here is made by running something. The children are stand-in
4
//! harnesses — shell scripts that speak the wire format of the real ones —
5
//! reached through the same `OA_CHILD_*` seam the TypeScript harnesses expose
6
//! for the same reason: a test that cannot substitute the agent can only
7
//! assert against a real one, which means it costs money or does not run.
8
//!
9
//! What the stand-in cannot fake is the part under test: the process is a real
10
//! process with a real pid in a real directory, its output is read as it is
11
//! written, and killing it is a real signal to a real process group.
12
13
use std::path::{Path, PathBuf};
14
use std::time::{Duration, Instant};
15
16
use openagents_cli::delegate::{ChildEvent, ChildLane, DelegationSupervisor};
17
use openagents_cli::workspace::{Isolation, WorkspacePlan};
18
use tokio::sync::{mpsc, watch};
19
20
/// The environment is one variable per lane and the tests share a process, so
21
/// two of them setting `OA_CHILD_CLAUDE` at once run each other's stand-in.
22
/// Held for as long as the variable is set rather than only while setting it.
23
static ENVIRONMENT: std::sync::Mutex<()> = std::sync::Mutex::new(());
24
25
fn exclusive() -> std::sync::MutexGuard<'static, ()> {
26
    // A test that panics while holding it poisons it; the next test still
27
    // wants the lock, and what it protects has no invariant to have broken.
28
    ENVIRONMENT.lock().unwrap_or_else(|held| held.into_inner())
29
}
30
31
/// Write an executable stand-in and return its path.
32
fn stand_in(name: &str, body: &str) -> PathBuf {
33
    let dir = std::env::temp_dir().join(format!("oa-delegate-test-{}", std::process::id()));
34
    std::fs::create_dir_all(&dir).unwrap();
35
    let path = dir.join(name);
36
    std::fs::write(&path, body).unwrap();
37
    #[cfg(unix)]
38
    {
39
        use std::os::unix::fs::PermissionsExt;
40
        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
41
    }
42
    path
43
}
44
45
/// Run a fan-out and collect both the outcomes and everything it reported.
46
async fn run(
47
    supervisor: &DelegationSupervisor,
48
    prompt: &str,
49
    stop_after: Option<Duration>,
50
) -> (
51
    Vec<openagents_cli::delegate::ChildWorkerResult>,
52
    Vec<(Instant, ChildEvent)>,
53
) {
54
    let (events, mut incoming) = mpsc::unbounded_channel();
55
    let collected = tokio::spawn(async move {
56
        let mut seen = Vec::new();
57
        while let Some(event) = incoming.recv().await {
58
            seen.push((Instant::now(), event));
59
        }
60
        seen
61
    });
62
63
    let (stop, cancel) = watch::channel(false);
64
    if let Some(after) = stop_after {
65
        tokio::spawn(async move {
66
            tokio::time::sleep(after).await;
67
            let _ = stop.send(true);
68
        });
69
    } else {
70
        // Held so the channel stays open for the life of the run.
71
        std::mem::forget(stop);
72
    }
73
74
    let results = supervisor
75
        .dispatch_streaming(prompt, events, cancel)
76
        .await
77
        .expect("no child could be started");
78
    (results, collected.await.unwrap())
79
}
80
81
/// Every child gets a git worktree of its own, and giving it back leaves
82
/// nothing behind.
83
#[tokio::test]
84
async fn each_child_gets_a_worktree_of_its_own() {
85
    let here = std::env::current_dir().unwrap();
86
    let plan = WorkspacePlan::resolve(here.clone(), Isolation::Worktree).await;
87
    assert_eq!(
88
        plan.isolation(),
89
        Isolation::Worktree,
90
        "the test runs inside a checkout, so a worktree is available"
91
    );
92
93
    let workspaces = plan.prepare(2).await.expect("the worktrees were not made");
94
    assert_eq!(workspaces.len(), 2);
95
    assert_ne!(
96
        workspaces[0].path, workspaces[1].path,
97
        "two children in one directory is the defect this replaces"
98
    );
99
100
    for workspace in &workspaces {
101
        assert!(workspace.path.is_dir(), "{}", workspace.path.display());
102
        // A worktree's `.git` is a file pointing back at the repository, not a
103
        // directory: this is a worktree and not a copied tree.
104
        assert!(
105
            workspace.path.join(".git").is_file(),
106
            "{} is not a git worktree",
107
            workspace.path.display()
108
        );
109
        // Each child has its own index, so what one writes the other does not
110
        // see.
111
        std::fs::write(workspace.path.join("only-mine.txt"), format!("{}", workspace.id)).unwrap();
112
    }
113
    assert!(!workspaces[0].path.join("only-mine.txt").is_symlink());
114
    assert_eq!(
115
        std::fs::read_to_string(workspaces[0].path.join("only-mine.txt")).unwrap(),
116
        "1"
117
    );
118
    assert_eq!(
119
        std::fs::read_to_string(workspaces[1].path.join("only-mine.txt")).unwrap(),
120
        "2"
121
    );
122
123
    for workspace in &workspaces {
124
        assert_eq!(workspace.release().await, None);
125
        assert!(
126
            !workspace.path.exists(),
127
            "{} outlived the fan-out",
128
            workspace.path.display()
129
        );
130
    }
131
}
132
133
/// A child's output reaches the caller while the child is still running.
134
///
135
/// The version this replaces called `Command::output()`, which returns when
136
/// the child is finished. The assertion is a clock: the first line has to
137
/// arrive well before the last one, and the child does not exit until it has
138
/// written both.
139
#[tokio::test]
140
async fn a_child_streams_while_it_is_still_running() {
141
    let harness = stand_in(
142
        "slow-claude",
143
        r#"#!/bin/sh
144
echo '{"type":"assistant","message":{"content":[{"type":"text","text":"first"}]}}'
145
sleep 2
146
echo '{"type":"assistant","message":{"content":[{"type":"text","text":"last"}]}}'
147
echo '{"type":"result","is_error":false,"result":"first last"}'
148
"#,
149
    );
150
    let _exclusive = exclusive();
151
    std::env::set_var("OA_CHILD_CLAUDE", &harness);
152
153
    let supervisor = DelegationSupervisor::new(1, "claude", None)
154
        .with_isolation(Isolation::Directory);
155
    let started = Instant::now();
156
    let (results, events) = run(&supervisor, "ignored by the stand-in", None).await;
157
    std::env::remove_var("OA_CHILD_CLAUDE");
158
159
    assert!(results[0].success, "{}", results[0].output);
160
    let total = started.elapsed();
161
    assert!(
162
        total >= Duration::from_secs(2),
163
        "the stand-in sleeps for two seconds; this run took {total:?}"
164
    );
165
166
    let first_said = events
167
        .iter()
168
        .find_map(|(at, event)| match event {
169
            ChildEvent::Output { text, .. } if text.contains("first") => Some(*at),
170
            _ => None,
171
        })
172
        .expect("nothing was streamed at all");
173
    let gap = first_said.duration_since(started);
174
    assert!(
175
        gap < Duration::from_secs(1),
176
        "the first line arrived after {gap:?}, which is the end of the run rather than the start of it"
177
    );
178
}
179
180
/// Children run at once, and the cap is what decides how many.
181
///
182
/// Three children that each sleep for a second finish in about a second when
183
/// they run together and about three when they are made to queue.
184
#[tokio::test]
185
async fn count_is_real_concurrency_and_the_cap_is_real_too() {
186
    let harness = stand_in(
187
        "sleepy-claude",
188
        r#"#!/bin/sh
189
sleep 1
190
echo "{\"type\":\"result\",\"is_error\":false,\"result\":\"pid $$\"}"
191
"#,
192
    );
193
    let _exclusive = exclusive();
194
    std::env::set_var("OA_CHILD_CLAUDE", &harness);
195
196
    let together = DelegationSupervisor::new(3, "claude", None)
197
        .with_isolation(Isolation::Directory);
198
    let at = Instant::now();
199
    let (parallel, events) = run(&together, "ignored", None).await;
200
    let parallel_took = at.elapsed();
201
202
    let queued = DelegationSupervisor::new(3, "claude", None)
203
        .with_isolation(Isolation::Directory)
204
        .with_max_parallel(1);
205
    let at = Instant::now();
206
    let (serial, _) = run(&queued, "ignored", None).await;
207
    let serial_took = at.elapsed();
208
    std::env::remove_var("OA_CHILD_CLAUDE");
209
210
    assert_eq!(parallel.iter().filter(|r| r.success).count(), 3);
211
    assert_eq!(serial.iter().filter(|r| r.success).count(), 3);
212
    assert!(
213
        parallel_took < Duration::from_millis(2_500),
214
        "three one-second children took {parallel_took:?} together, which is not together"
215
    );
216
    assert!(
217
        serial_took > Duration::from_millis(2_500),
218
        "three one-second children capped at one at a time took {serial_took:?}, which is not a cap"
219
    );
220
221
    // Three real processes, three different pids, three different directories.
222
    let pids: Vec<u32> = parallel.iter().filter_map(|result| result.pid).collect();
223
    assert_eq!(pids.len(), 3, "a child with no pid is not a child process");
224
    assert_eq!(
225
        pids.iter().collect::<std::collections::BTreeSet<_>>().len(),
226
        3,
227
        "{pids:?} are not three separate processes"
228
    );
229
    let homes: std::collections::BTreeSet<&Path> = parallel
230
        .iter()
231
        .filter_map(|result| result.workspace.as_deref())
232
        .collect();
233
    assert_eq!(homes.len(), 3, "{homes:?} is not one directory per child");
234
235
    let announced = events
236
        .iter()
237
        .filter(|(_, event)| matches!(event, ChildEvent::Started { .. }))
238
        .count();
239
    assert_eq!(announced, 3);
240
}
241
242
/// A child that fails is reported as failed.
243
///
244
/// Both shapes: one that runs and exits non-zero, and one whose binary is not
245
/// there at all. The version this replaces reported `2/2 children succeeded`
246
/// in both cases.
247
#[tokio::test]
248
async fn a_failing_child_is_reported_as_failed() {
249
    let harness = stand_in(
250
        "doomed-claude",
251
        r#"#!/bin/sh
252
echo "the tests did not pass" >&2
253
exit 3
254
"#,
255
    );
256
    let _exclusive = exclusive();
257
    std::env::set_var("OA_CHILD_CLAUDE", &harness);
258
    let supervisor = DelegationSupervisor::new(1, "claude", None)
259
        .with_isolation(Isolation::Directory);
260
    let (results, _) = run(&supervisor, "ignored", None).await;
261
    std::env::remove_var("OA_CHILD_CLAUDE");
262
263
    assert!(!results[0].success, "a child that exited 3 reported success");
264
    assert!(results[0].failure.is_some());
265
    let why = results[0].failure.clone().unwrap();
266
    assert!(why.contains("code 3"), "{why}");
267
    assert!(why.contains("the tests did not pass"), "{why}");
268
269
    std::env::set_var("OA_CHILD_CLAUDE", "/nonexistent/no-such-agent");
270
    let supervisor = DelegationSupervisor::new(1, "claude", None)
271
        .with_isolation(Isolation::Directory);
272
    let (missing, _) = run(&supervisor, "ignored", None).await;
273
    std::env::remove_var("OA_CHILD_CLAUDE");
274
275
    assert!(!missing[0].success);
276
    assert!(
277
        missing[0].failure.clone().unwrap().contains("not on PATH"),
278
        "{:?}",
279
        missing[0].failure
280
    );
281
}
282
283
/// Stopping a fan-out stops the children, and the children's own children.
284
///
285
/// The stand-in starts a background process that writes a file after five
286
/// seconds. If the group was signalled, the file never appears; if only the
287
/// direct child was killed, it does.
288
#[tokio::test]
289
async fn stopping_a_fanout_kills_the_whole_group() {
290
    let witness = std::env::temp_dir().join(format!("oa-orphan-{}.txt", std::process::id()));
291
    let _ = std::fs::remove_file(&witness);
292
293
    let harness = stand_in(
294
        "runaway-claude",
295
        &format!(
296
            r#"#!/bin/sh
297
sh -c 'sleep 5; echo orphaned > {}' &
298
sleep 30
299
"#,
300
            witness.display()
301
        ),
302
    );
303
    let _exclusive = exclusive();
304
    std::env::set_var("OA_CHILD_CLAUDE", &harness);
305
306
    let supervisor = DelegationSupervisor::new(1, "claude", None)
307
        .with_isolation(Isolation::Directory);
308
    let at = Instant::now();
309
    let (results, _) = run(&supervisor, "ignored", Some(Duration::from_millis(700))).await;
310
    let took = at.elapsed();
311
    std::env::remove_var("OA_CHILD_CLAUDE");
312
313
    assert!(!results[0].success);
314
    assert_eq!(results[0].failure.as_deref(), Some("stopped before finishing"));
315
    assert!(
316
        took < Duration::from_secs(20),
317
        "a stopped child ran for {took:?}; the stand-in sleeps for thirty seconds"
318
    );
319
320
    // Long enough for the grandchild to have fired if it were still alive.
321
    tokio::time::sleep(Duration::from_secs(6)).await;
322
    assert!(
323
        !witness.exists(),
324
        "the child's own subprocess outlived the fan-out that started it"
325
    );
326
}
327
328
/// A lane name that is not a lane is refused rather than quietly redirected.
329
#[test]
330
fn an_unknown_lane_is_not_silently_ox_alpha() {
331
    assert!(ChildLane::known("claude"));
332
    assert!(ChildLane::known("opencode/x-preview-f-free"));
333
    assert!(!ChildLane::known("gemni"));
334
    assert!(!ChildLane::known(""));
335
}

This page updates live while a promote is in flight · changelog