diff --git a/.agentworkforce/trajectories/fleet-startup-task-summary.json b/.agentworkforce/trajectories/fleet-startup-task-summary.json new file mode 100644 index 000000000..f986444e4 --- /dev/null +++ b/.agentworkforce/trajectories/fleet-startup-task-summary.json @@ -0,0 +1,43 @@ +{ + "id": "compact_vh1unpxve3p7", + "version": 1, + "type": "compacted", + "compactedAt": "2026-10-09T00:06:51.822Z", + "sourceTrajectories": [ + "traj_q94ajuue54aw" + ], + "dateRange": { + "start": "2026-10-08T23:57:27.648Z", + "end": "2026-10-09T00:06:30.759Z" + }, + "summary": { + "totalDecisions": 2, + "totalEvents": 3, + "uniqueAgents": [ + "default" + ] + }, + "decisionGroups": [ + { + "category": "other", + "decisions": [ + { + "question": "Pass Claude and Codex startup briefs through native argv prompts", + "chosen": "Pass Claude and Codex startup briefs through native argv prompts", + "reasoning": "PTY startup typing can lose a prefix; argv preserves the complete task and existing argv bookkeeping prevents duplicate injection.", + "fromTrajectory": "traj_q94ajuue54aw" + }, + { + "question": "Preserve task whitespace and recognize active argv startup turns", + "chosen": "Preserve task whitespace and recognize active argv startup turns", + "reasoning": "Fleet task parsing previously trimmed bytes; native prompts may start working before an idle composer appears, so readiness must accept a busy harness while still vetoing trust dialogs.", + "fromTrajectory": "traj_q94ajuue54aw" + } + ] + } + ], + "keyLearnings": [], + "keyFindings": [], + "filesAffected": [], + "commits": [] +} \ No newline at end of file diff --git a/.agentworkforce/trajectories/fleet-startup-task-summary.md b/.agentworkforce/trajectories/fleet-startup-task-summary.md new file mode 100644 index 000000000..c9ea41f78 --- /dev/null +++ b/.agentworkforce/trajectories/fleet-startup-task-summary.md @@ -0,0 +1,26 @@ +# Trajectory Compaction: Oct 8, 2026 - Oct 9, 2026 + +## Summary +- Sessions: 1 +- Decisions: 2 +- Events: 3 +- Agents: default +- Files: 0 +- Commits: 0 + +## Other +- Pass Claude and Codex startup briefs through native argv prompts -> Pass Claude and Codex startup briefs through native argv prompts (traj_q94ajuue54aw) +- Preserve task whitespace and recognize active argv startup turns -> Preserve task whitespace and recognize active argv startup turns (traj_q94ajuue54aw) + +## Key Learnings +- No additional findings recorded. + +## Key Findings +- No additional findings recorded. + +## Validation + +- Real broker/PTY fleet fixtures: 7 passed, including Claude/Codex 400-character and 8-KiB tasks and explicit PTY configs. Exact first task and absence of duplicate PTY injection were checked. No provider credentials were used. +- Broker unit suite: 1,412 passed, 5 ignored, zero failures after removing inherited Git config overrides from the test process. +- Production Clippy, TypeScript compilation, Rust formatting and changed-file Prettier checks passed. Strict all-target Clippy has seven existing test-code lints under Rust 1.99; compatibility lint allowances pass. +- Initial argv tasks containing NUL or exceeding 16 KiB fail before registration. Deployments must upgrade and restart brokers to receive this fix. diff --git a/CHANGELOG.md b/CHANGELOG.md index 21f3330a1..e05788c5e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,11 @@ All notable changes to Agent Relay will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Patch] + +### Fixed + +- `fleet spawn --task` passes Claude and Codex initial briefs directly to the harness at startup, preventing task prefixes from being lost while the terminal composer loads. Startup prompts containing NUL or exceeding 16 KiB fail explicitly before registration. ## [13.2.0] - 2026-10-08 diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index 6ee5866da..f9bf5aba2 100644 --- a/crates/broker/src/pty_worker.rs +++ b/crates/broker/src/pty_worker.rs @@ -655,6 +655,20 @@ fn startup_gate_block_reason(resolved_cli: &str, screen: &str) -> Option bool { + match cli_basename(cli).to_ascii_lowercase().as_str() { + "codex" | "codex.exe" => screen.lines().any(is_codex_busy_status_line), + "claude" | "claude.exe" => screen.lines().any(|line| { + let lower = line.to_ascii_lowercase(); + lower.contains("esc to interrupt") + && line + .trim_start() + .starts_with(['✻', '✽', '✶', '✳', '✢', '·']) + }), + _ => false, + } +} + /// Classify the harness's startup state from a single grid render, so the /// readiness and blocking verdicts cannot straddle two repaints. fn startup_gate( @@ -663,8 +677,19 @@ fn startup_gate( startup_total_bytes: usize, output_quiet: Duration, pty: &PtySession, + startup_task_in_argv: bool, ) -> StartupGate { let screen = pty.screen_text(); + if let Some(blocked) = startup_gate_block_reason(resolved_cli, &screen) { + return blocked; + } + // A native argv task can already be running before an idle composer is + // ever drawn. Its busy status proves startup without waiting for that + // first (potentially long) turn to finish. PTY-injected tasks still require + // the idle prompt, and trust/auth dialogs always veto this path. + if startup_task_in_argv && native_startup_turn_running(resolved_cli, &screen) { + return StartupGate::Ready; + } if evaluate_startup_gate( resolved_cli, startup_output, @@ -1066,6 +1091,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { let mut startup_total_bytes = 0usize; let mut init_request_id: Option = None; let mut init_received_at: Option = None; + let mut startup_task_in_argv = false; let mut startup_readiness = StartupReadinessState::default(); let suppress_multiline_mcp_reminder = cli_basename(&resolved_cli).eq_ignore_ascii_case("agent") || cli_basename(&resolved_cli).eq_ignore_ascii_case("cursor-agent") @@ -1234,6 +1260,10 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { .map(ToOwned::to_owned) }) .unwrap_or_else(|| "pty-worker".to_string()); + startup_task_in_argv = frame.payload + .get("startup_task_in_argv") + .and_then(Value::as_bool) + .unwrap_or(false); init_request_id = frame.request_id; init_received_at = Some(Instant::now()); // Process liveness and safe input readiness are @@ -1255,6 +1285,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { startup_total_bytes, last_pty_output_time.elapsed(), &pty, + startup_task_in_argv, ); try_emit_worker_ready( &out_tx, @@ -1683,6 +1714,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { startup_total_bytes, last_pty_output_time.elapsed(), &pty, + startup_task_in_argv, ); try_emit_worker_ready( &out_tx, @@ -2390,6 +2422,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { startup_total_bytes, last_pty_output_time.elapsed(), &pty, + startup_task_in_argv, ); try_emit_worker_ready( &out_tx, @@ -2639,6 +2672,56 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { mod tests { use super::*; + #[cfg(unix)] + #[tokio::test] + async fn native_startup_busy_gate_requires_argv_task_and_respects_trust() { + for (cli, screen) in [ + ("codex", "Working (1s • esc to interrupt)"), + ("claude", "✻ Thinking… (esc to interrupt)"), + ("codex", "Do you trust the contents of this directory?\n› 1. Yes, continue\n2. No, quit\nWorking (esc to interrupt)"), + ] { + let blocked = screen.contains("Do you trust"); + let script = format!("printf '%s' '{}' ; sleep 5", screen); + let (pty, mut rx) = PtySession::spawn("/bin/sh", &["-c".into(), script], 24, 120).unwrap(); + let drain = tokio::spawn(async move { while rx.recv().await.is_some() {} }); + for _ in 0..100 { + if pty.screen_text().contains("esc to interrupt") { break; } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(pty.screen_text().contains("esc to interrupt")); + let expected = if blocked { StartupGate::Blocked } else { StartupGate::Unrecognised }; + assert_eq!(startup_gate(cli, "", 0, Duration::ZERO, &pty, false), expected); + let expected = if blocked { StartupGate::Blocked } else { StartupGate::Ready }; + assert_eq!(startup_gate(cli, "", 0, Duration::ZERO, &pty, true), expected); + pty.shutdown().unwrap(); + drain.abort(); + } + } + + #[test] + fn native_startup_turn_requires_harness_specific_busy_status() { + assert!(native_startup_turn_running( + "codex", + "Working (1s • esc to interrupt)" + )); + assert!(native_startup_turn_running( + "claude", + "✻ Thinking… (esc to interrupt)" + )); + assert!(!native_startup_turn_running( + "claude", + "Task says esc to interrupt" + )); + assert!(!native_startup_turn_running( + "codex", + "› Ask Codex to do anything" + )); + assert!(!native_startup_turn_running( + "gemini", + "Working (esc to interrupt)" + )); + } + #[test] fn completed_delivery_replay_retains_the_original_event_identity() { let mut completed = CompletedWorkerDeliveries::default(); diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index ee2bd9d91..07f2b4265 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -572,7 +572,7 @@ impl BrokerRuntime { ); } } - if let Err(error) = crate::worker::validate_muse_startup_prompt_for_spec( + if let Err(error) = crate::worker::validate_pty_startup_prompt_for_spec( &spec, effective_task.as_deref(), ) { diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index a94631822..577eeec32 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -1548,7 +1548,7 @@ impl BrokerRuntime { return; } }; - let task = action_invoke_string(&invoke.input, &["task", "initial_task", "prompt"]); + let task = action_invoke_task(&invoke.input); let channel = action_invoke_string(&invoke.input, &["channel"]); let model = action_invoke_string(&invoke.input, &["model"]); @@ -3075,6 +3075,21 @@ fn action_invoke_agent_name(invoke: &ActionInvoke) -> Option { }) } +/// Task text is opaque: use trimming only to reject empty values, never to +/// change the bytes that become the harness's first turn. +fn action_invoke_task(input: &Value) -> Option { + for source in [Some(input), input.get("agent")].into_iter().flatten() { + for key in ["task", "initial_task", "prompt"] { + if let Some(value) = source.get(key).and_then(Value::as_str) { + if !value.trim().is_empty() { + return Some(value.to_string()); + } + } + } + } + None +} + /// Read the first non-empty string at any of the given top-level keys of an /// `action.invoke` input object (also checks under a nested `agent` object, /// mirroring the firehose payload shape). @@ -3759,6 +3774,30 @@ mod tests { ); } + #[test] + fn action_invoke_task_preserves_whitespace_and_alias_precedence() { + let task = " implement issue #34\n\t trailing "; + for key in ["task", "initial_task", "prompt"] { + assert_eq!( + action_invoke_task(&json!({key: task})), + Some(task.to_string()) + ); + assert_eq!( + action_invoke_task(&json!({"agent": {key: task}})), + Some(task.to_string()) + ); + } + assert_eq!( + action_invoke_task(&json!({"task": " ", "agent": {"task": task}})), + Some(task.to_string()) + ); + assert_eq!( + action_invoke_task(&json!({"task": "top", "agent": {"task": "nested"}})), + Some("top".to_string()) + ); + assert_eq!(action_invoke_task(&json!({"task": "\n\t"})), None); + } + #[test] fn action_invoke_string_reads_top_level_and_nested_agent() { let input = json!({"cli": "codex", "agent": {"model": "gpt-5"}}); diff --git a/crates/broker/src/runtime/relaycast_events.rs b/crates/broker/src/runtime/relaycast_events.rs index cb1c002b2..b048dda08 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -797,7 +797,7 @@ pub(super) async fn spawn_worker_from_request( "prepared relay skill prefix before Relaycast worker startup" ); } - crate::worker::validate_muse_startup_prompt_for_spec(&spec, effective_task.as_deref())?; + crate::worker::validate_pty_startup_prompt_for_spec(&spec, effective_task.as_deref())?; // Pre-register an agent token for every spawned worker. // The Agent Relay MCP server needs RELAY_AGENT_TOKEN + diff --git a/crates/broker/src/worker.rs b/crates/broker/src/worker.rs index ee2d45a63..359ade1d3 100644 --- a/crates/broker/src/worker.rs +++ b/crates/broker/src/worker.rs @@ -906,7 +906,7 @@ impl WorkerRegistry { if self.workers.contains_key(&spec.name) { anyhow::bail!("agent '{}' already exists", spec.name); } - validate_muse_startup_prompt_for_spec(&spec, initial_task.as_deref())?; + validate_pty_startup_prompt_for_spec(&spec, initial_task.as_deref())?; tracing::info!( target = "broker::spawn", @@ -1134,7 +1134,7 @@ impl WorkerRegistry { let startup_prompt = muse_startup_prompt(&cli_lower, initial_task.as_deref()); initial_task_in_argv = startup_prompt.is_some(); - let pty_cli_args = ordered_pty_cli_args( + let mut pty_cli_args = ordered_pty_cli_args( bypass_flag, muse_flag, model_flag.as_deref(), @@ -1143,6 +1143,11 @@ impl WorkerRegistry { &harness_session_args, startup_prompt, ); + initial_task_in_argv |= append_native_startup_prompt( + &mut pty_cli_args, + &resolved_cli, + initial_task.as_deref(), + ); if !pty_cli_args.is_empty() { command.arg("--"); for arg in &pty_cli_args { @@ -1392,7 +1397,7 @@ impl WorkerRegistry { let startup_prompt = muse_startup_prompt(&cli_lower, initial_task.as_deref()); initial_task_in_argv = startup_prompt.is_some(); - let pty_cli_args = ordered_pty_cli_args( + let mut pty_cli_args = ordered_pty_cli_args( bypass_flag, muse_flag, model_flag.as_deref(), @@ -1401,6 +1406,11 @@ impl WorkerRegistry { &harness_session_args, startup_prompt, ); + initial_task_in_argv |= append_native_startup_prompt( + &mut pty_cli_args, + &resolved_cli, + initial_task.as_deref(), + ); if !pty_cli_args.is_empty() { command.arg("--"); for arg in &pty_cli_args { @@ -1734,6 +1744,7 @@ impl WorkerRegistry { None, json!({ "agent": spec, + "startup_task_in_argv": initial_task_in_argv, }), ) .await @@ -1901,9 +1912,9 @@ impl WorkerRegistry { } /// Remove a startup task once the harness is ready and return it only when - /// the broker still needs to inject it through the PTY. Muse receives its - /// startup task as an argv prompt, but remains present in `initial_tasks` - /// until readiness so follow-up deliveries cannot race the assigned work. + /// the broker still needs to inject it through the PTY. Claude, Codex and + /// Muse receive their startup task as an argv prompt, which remains in + /// `initial_tasks` until readiness so follow-up deliveries cannot race it. pub(crate) fn take_initial_task_for_injection(&mut self, name: &str) -> Option { let task = self.initial_tasks.remove(name); if self.argv_initial_tasks.remove(name) { @@ -2443,10 +2454,33 @@ fn muse_yolo_flag(cli_lower: &str, effective_args: &[String]) -> Option<&'static Some("--yolo") } -/// Muse only begins a broker-assigned task deterministically when the prompt is -/// present at process startup. Keep the prompt as one argv value (including -/// newlines) and leave every other harness on the established post-ready PTY -/// injection path. +/// Native startup prompts are parsed by the harness before its TUI starts. +/// Never type the initial brief into a booting composer. The option terminator +/// also keeps leading dashes and variadic options (e.g. Codex --image) from +/// consuming the task. For Codex resume, this follows the session ID. +fn append_native_startup_prompt(args: &mut Vec, cli: &str, task: Option<&str>) -> bool { + if !native_startup_prompt_cli(cli) { + return false; + } + let Some(task) = task.filter(|task| !task.trim().is_empty()) else { + return false; + }; + args.push("--".to_string()); + args.push(task.to_string()); + true +} + +fn native_startup_prompt_cli(cli: &str) -> bool { + matches!( + cli.rsplit(['/', '\\']) + .next() + .unwrap_or(cli) + .to_ascii_lowercase() + .as_str(), + "claude" | "claude.exe" | "codex" | "codex.exe" + ) +} + fn muse_startup_prompt<'a>(cli_lower: &str, initial_task: Option<&'a str>) -> Option<&'a str> { if !is_muse_executable(cli_lower) { return None; @@ -2454,16 +2488,16 @@ fn muse_startup_prompt<'a>(cli_lower: &str, initial_task: Option<&'a str>) -> Op initial_task.filter(|task| !task.trim().is_empty()) } -/// Portable ceiling for Muse's single-argument startup prompt. Windows limits +/// Portable ceiling for a single-argument startup prompt. Windows limits /// the complete command line to roughly 32 Ki UTF-16 code units; reserving half /// for the executable and broker/user flags keeps accepted prompts portable. -pub(crate) const MUSE_STARTUP_PROMPT_MAX_BYTES: usize = 16 * 1024; +pub(crate) const PTY_STARTUP_PROMPT_MAX_BYTES: usize = 16 * 1024; -/// Validate text that must cross Muse's argv startup boundary. Callers perform +/// Validate text that must cross the harness argv startup boundary. Callers perform /// this check before registering a remote worker identity; `spawn` repeats it /// as a final defense for restart and direct registry callers. -pub(crate) fn validate_muse_startup_prompt(cli: &str, task: Option<&str>) -> Result<()> { - if !is_muse_executable(cli) { +pub(crate) fn validate_pty_startup_prompt(cli: &str, task: Option<&str>) -> Result<()> { + if !is_muse_executable(cli) && !native_startup_prompt_cli(cli) { return Ok(()); } let Some(task) = task.filter(|task| !task.trim().is_empty()) else { @@ -2471,23 +2505,23 @@ pub(crate) fn validate_muse_startup_prompt(cli: &str, task: Option<&str>) -> Res }; if task.contains('\0') { anyhow::bail!( - "Muse startup task contains a NUL byte and cannot be passed as a process argument" + "PTY startup task contains a NUL byte and cannot be passed as a process argument" ); } - if task.len() > MUSE_STARTUP_PROMPT_MAX_BYTES { + if task.len() > PTY_STARTUP_PROMPT_MAX_BYTES { anyhow::bail!( - "Muse startup task is {} bytes; the portable argv limit is {} bytes", + "PTY startup task is {} bytes; the portable argv limit is {} bytes", task.len(), - MUSE_STARTUP_PROMPT_MAX_BYTES + PTY_STARTUP_PROMPT_MAX_BYTES ); } Ok(()) } /// Resolve the command that a PTY spec will actually launch before validating -/// its Muse prompt. Explicit PTY harness configs own the executable; ordinary +/// its startup prompt. Explicit PTY harness configs own the executable; ordinary /// PTY specs use `spec.cli`. -pub(crate) fn validate_muse_startup_prompt_for_spec( +pub(crate) fn validate_pty_startup_prompt_for_spec( spec: &AgentSpec, task: Option<&str>, ) -> Result<()> { @@ -2508,7 +2542,7 @@ pub(crate) fn validate_muse_startup_prompt_for_spec( _ => None, }; match cli { - Some(cli) => validate_muse_startup_prompt(&cli, task), + Some(cli) => validate_pty_startup_prompt(&cli, task), None => Ok(()), } } @@ -4759,6 +4793,57 @@ sleep 30 ); } + #[test] + fn native_startup_prompt_preserves_task_bytes_after_options_and_resume() { + for cli in ["claude", "/bin/claude", "codex", r"C:\tools\codex.exe"] { + for size in [400, 8192] { + let task = format!("--{}\n é🙂\t ", "x".repeat(size)); + let mut args = ordered_pty_cli_args( + None, + None, + None, + &[], + &["--image".into(), "screenshot.png".into()], + &["resume".into(), "thread-id".into()], + None, + ); + assert!(append_native_startup_prompt(&mut args, cli, Some(&task))); + assert_eq!( + args, + vec![ + "resume", + "thread-id", + "--image", + "screenshot.png", + "--", + &task + ] + ); + assert!(validate_pty_startup_prompt(cli, Some(&task)).is_ok()); + assert!(validate_pty_startup_prompt(cli, Some("bad\0task")).is_err()); + assert!(validate_pty_startup_prompt( + cli, + Some(&"x".repeat(PTY_STARTUP_PROMPT_MAX_BYTES + 1)) + ) + .is_err()); + } + } + } + + #[test] + fn native_startup_prompt_does_not_change_other_harnesses_or_empty_tasks() { + for (cli, task) in [ + ("gemini", Some("task")), + ("muse", Some("task")), + ("claude", None), + ("codex", Some(" \n")), + ] { + let mut args = vec!["--model".into(), "existing".into()]; + assert!(!append_native_startup_prompt(&mut args, cli, task)); + assert_eq!(args, vec!["--model", "existing"]); + } + } + #[test] fn muse_startup_prompt_is_not_used_for_other_harnesses_or_empty_tasks() { assert_eq!(muse_startup_prompt("codex", Some("task")), None); @@ -4768,22 +4853,22 @@ sleep 30 #[test] fn muse_startup_prompt_rejects_nonportable_argv_text() { - assert!(validate_muse_startup_prompt("muse", Some("valid task")).is_ok()); - assert!(validate_muse_startup_prompt("codex", Some("nul\0is fine off argv")).is_ok()); + assert!(validate_pty_startup_prompt("muse", Some("valid task")).is_ok()); + assert!(validate_pty_startup_prompt("gemini", Some("nul\0is fine off argv")).is_ok()); - let nul = validate_muse_startup_prompt("muse", Some("invalid\0task")) + let nul = validate_pty_startup_prompt("muse", Some("invalid\0task")) .expect_err("Muse argv cannot contain NUL") .to_string(); assert!(nul.contains("NUL byte"), "{nul}"); - let maximum = "x".repeat(MUSE_STARTUP_PROMPT_MAX_BYTES); - assert!(validate_muse_startup_prompt("muse", Some(&maximum)).is_ok()); + let maximum = "x".repeat(PTY_STARTUP_PROMPT_MAX_BYTES); + assert!(validate_pty_startup_prompt("muse", Some(&maximum)).is_ok()); let oversized = format!("{maximum}x"); - let error = validate_muse_startup_prompt("muse", Some(&oversized)) + let error = validate_pty_startup_prompt("muse", Some(&oversized)) .expect_err("oversized Muse argv must fail before process spawn") .to_string(); assert!( - error.contains(&MUSE_STARTUP_PROMPT_MAX_BYTES.to_string()), + error.contains(&PTY_STARTUP_PROMPT_MAX_BYTES.to_string()), "{error}" ); } @@ -4806,9 +4891,9 @@ sleep 30 channels: Vec::new(), restart_policy: None, }; - let oversized = "x".repeat(MUSE_STARTUP_PROMPT_MAX_BYTES + 1); + let oversized = "x".repeat(PTY_STARTUP_PROMPT_MAX_BYTES + 1); assert!( - validate_muse_startup_prompt_for_spec(&spec, Some(&oversized)).is_err(), + validate_pty_startup_prompt_for_spec(&spec, Some(&oversized)).is_err(), "inline Muse commands must not bypass argv prompt validation" ); } diff --git a/tests/integration/broker/fleet-spawn-readiness.test.ts b/tests/integration/broker/fleet-spawn-readiness.test.ts index e8d15e052..a74acbb35 100644 --- a/tests/integration/broker/fleet-spawn-readiness.test.ts +++ b/tests/integration/broker/fleet-spawn-readiness.test.ts @@ -1,7 +1,7 @@ -/** Real broker/PTY, loopback engine, deterministic Claude stub; no live credentials. */ +/** Real broker/PTY, loopback engine, deterministic Claude/Codex stubs; no live credentials. */ import assert from 'node:assert/strict'; import { spawn } from 'node:child_process'; -import { mkdtemp, mkdir, rm, writeFile } from 'node:fs/promises'; +import { mkdtemp, mkdir, readFile, rm, writeFile } from 'node:fs/promises'; import http from 'node:http'; import { tmpdir } from 'node:os'; import path from 'node:path'; @@ -11,233 +11,295 @@ import { WebSocketServer, type WebSocket } from 'ws'; import { checkPrerequisites, resolveBinaryPath } from './utils/broker-harness.js'; -test( - 'targeted Claude spawn confirms real broker readiness inside the default budget', - { timeout: 130_000 }, - async (t: TestContext) => { - // Skip rather than fail when the broker has not been built, matching every - // sibling suite here. - const missing = checkPrerequisites(); - if (missing) { - t.skip(missing); - return; - } - const binary = resolveBinaryPath(); - const directory = await mkdtemp(path.join(tmpdir(), 'fleet-ready-')); - const bin = path.join(directory, 'bin'); - await mkdir(bin); - await writeFile(path.join(bin, 'claude'), "#!/bin/sh\nprintf '%s\\n' '->pty:ready'\nsleep 120\n", { - mode: 0o755, - }); - const receipts = new Map>(); - let nodeSocket: WebSocket | undefined; - let registrations = 0; - const server = http.createServer(async (request, response) => { - const chunks: Buffer[] = []; - for await (const chunk of request) chunks.push(Buffer.from(chunk)); - const body = chunks.length ? JSON.parse(Buffer.concat(chunks).toString()) : {}; - response.setHeader('content-type', 'application/json'); - if (request.method === 'GET' && request.url?.startsWith('/v1/agents/')) { - response.end( - JSON.stringify({ - ok: true, - data: { - id: 'worker-1', - name: 'Probe', - workspace_id: 'fixture-workspace', - status: 'online', - channels: [{ name: 'general' }, { name: 'engineering' }], - metadata: {}, - }, - }) - ); - return; - } - if (request.url?.endsWith('/members')) { - response.end( - JSON.stringify({ - ok: true, - data: [ - { - agent_id: 'worker-1', - agent_name: 'Probe', - role: 'member', - joined_at: '2026-01-01T00:00:00Z', - }, - ], - }) - ); +for (const scenario of [ + { cli: 'claude', size: 0, explicit: false }, + { cli: 'claude', size: 400, explicit: false }, + { cli: 'claude', size: 8192, explicit: false }, + { cli: 'codex', size: 400, explicit: false }, + { cli: 'codex', size: 8192, explicit: false }, + { cli: 'claude', size: 8192, explicit: true }, + { cli: 'codex', size: 8192, explicit: true }, +]) { + const { cli, size, explicit } = scenario; + test( + `targeted ${cli} spawn preserves a ${size}-byte initial turn and confirms readiness${explicit ? ' with explicit PTY config' : ''}`, + { timeout: 130_000 }, + async (t: TestContext) => { + // Skip rather than fail when the broker has not been built, matching every + // sibling suite here. + const missing = checkPrerequisites(); + if (missing) { + t.skip(missing); return; } - response.end( - JSON.stringify({ - ok: true, - data: - request.url === '/v1/agents' - ? { - id: 'fixture-broker', - workspace_id: 'fixture-workspace', - name: body.name, - token: 'at_fixture', - status: 'online', - created_at: '2026-01-01T00:00:00Z', - } - : { - id: 'channel-fixture', - name: body.name ?? 'general', - workspace_id: 'fixture-workspace', - created_at: '2026-01-01T00:00:00Z', - created_by: 'fixture-broker', - is_archived: false, - topic: null, - members: [], - member_count: 1, - }, - }) + const binary = resolveBinaryPath(); + const directory = await mkdtemp(path.join(tmpdir(), 'fleet-ready-')); + const bin = path.join(directory, 'bin'); + await mkdir(bin); + const turnFile = path.join(directory, 'turn.json'); + const prefix = '--Implement issue #34 exactly; preserve spaces and punctuation.\n'; + const suffix = size === 400 ? '\nEND\t ' : '\nEND é🙂\t '; + const task = size ? prefix + 'x'.repeat(size - Buffer.byteLength(prefix + suffix)) + suffix : undefined; + if (task) assert.equal(Buffer.byteLength(task), size); + await writeFile( + path.join(bin, cli), + `#!${process.execPath} +const fs = require('node:fs'); +const args = process.argv.slice(2); +if (args.includes('app-server')) process.exit(1); +if (args.includes('--version')) { console.log('${cli} 1.0.0'); process.exit(0); } +const separator = args.lastIndexOf('--'); +const turns = separator >= 0 ? args.slice(separator + 1) : []; +let input = ''; +if (process.stdin.isTTY) process.stdin.setRawMode(true); +process.stdin.on('data', chunk => { + input += chunk.toString(); + fs.writeFileSync(${JSON.stringify(turnFile)}, JSON.stringify({ turns, input })); +}); +// The TUI drops input while loading. The native startup prompt is retained. +setTimeout(() => { + fs.writeFileSync(${JSON.stringify(turnFile)}, JSON.stringify({ turns, input })); + const screen = turns.length + ? (${JSON.stringify(cli)} === 'codex' ? 'Working (1s • esc to interrupt)' : '✻ Thinking… (esc to interrupt)') + : '->pty:ready'; + process.stdout.write('\\x1b[2J\\x1b[H' + screen); +}, 300); +setInterval(() => {}, 1000); +`, + { mode: 0o755 } ); - }); - const wss = new WebSocketServer({ server }); - wss.on('connection', (socket, request) => { - const isNode = request.url?.startsWith('/v1/node/ws'); - socket.on('message', (raw) => { - const frame = JSON.parse(raw.toString()); - if (isNode && frame.type === 'inventory.sync') nodeSocket = socket; - if (frame.type === 'action.result') - receipts.set(frame.invocation_id, { - invocation_id: frame.invocation_id, - status: frame.error ? 'failed' : 'completed', - output: frame.output, - error: frame.error, - }); - if (['node.register', 'inventory.sync', 'agent.register', 'agent.deregister'].includes(frame.type)) { - if (frame.type === 'agent.register') registrations++; - socket.send( + const receipts = new Map>(); + let nodeSocket: WebSocket | undefined; + let registrations = 0; + const server = http.createServer(async (request, response) => { + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(Buffer.from(chunk)); + const body = chunks.length ? JSON.parse(Buffer.concat(chunks).toString()) : {}; + response.setHeader('content-type', 'application/json'); + if ( + request.method === 'GET' && + (request.url === '/v1/agent' || request.url?.startsWith('/v1/agents/')) + ) { + response.end( + JSON.stringify({ + ok: true, + data: { + id: 'worker-1', + name: 'Probe', + type: 'agent', + workspace_id: 'fixture-workspace', + status: 'online', + channels: ['general', 'engineering'].map((name) => ({ + id: `channel-${name}`, + name, + role: 'member', + joined_at: '2026-01-01T00:00:00Z', + })), + metadata: {}, + }, + }) + ); + return; + } + if (request.url?.endsWith('/members')) { + response.end( JSON.stringify({ - v: 1, - type: 'reply', - id: frame.id, ok: true, - data: - frame.type === 'agent.register' - ? { - agent_id: `worker-${registrations}`, - token: 'at_worker_fixture', - name: frame.name, - delivery_ack_seq: 0, - } - : {}, + data: [ + { + agent_id: 'worker-1', + agent_name: 'Probe', + role: 'member', + joined_at: '2026-01-01T00:00:00Z', + }, + ], }) ); + return; } + response.end( + JSON.stringify({ + ok: true, + data: + request.url === '/v1/agents' + ? { + id: 'fixture-broker', + workspace_id: 'fixture-workspace', + name: body.name, + token: 'at_fixture', + status: 'online', + created_at: '2026-01-01T00:00:00Z', + } + : { + id: 'channel-fixture', + name: body.name ?? 'general', + workspace_id: 'fixture-workspace', + created_at: '2026-01-01T00:00:00Z', + created_by: 'fixture-broker', + is_archived: false, + topic: null, + members: [], + member_count: 1, + }, + }) + ); }); - }); - await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); - const address = server.address(); - assert(address && typeof address !== 'string'); - const baseUrl = `http://127.0.0.1:${address.port}`; - const broker = spawn( - binary, - [ - 'init', - '--instance-name', - 'ready-node', - '--workspace-key', - 'rk_fixture', - '--state-dir', - directory, - '--api-port', - '0', - '--channels', - '', - ], - { - cwd: directory, - env: { - PATH: `${bin}:${process.env.PATH}`, - HOME: directory, - TMPDIR: directory, - RELAYCAST_BASE_URL: baseUrl, - RELAY_BASE_URL: baseUrl, - RELAY_BROKER_API_KEY: 'br_fixture', - RELAY_NODE_ID: 'node-ready', - RELAY_NODE_TOKEN: 'nt_fixture', - AGENT_RELAY_BROKER_LOG: 'stderr', - AGENT_RELAY_TELEMETRY_DISABLED: '1', - AGENT_RELAY_NO_DEBUG_FILES: '1', - }, - stdio: ['ignore', 'pipe', 'pipe'], - } - ); - let logs = ''; - let spawnError: Error | undefined; - broker.on('error', (error) => { - spawnError = error; - }); - broker.stdout.on('data', (chunk) => { - logs = (logs + chunk).slice(-16000); - if (process.env.FLEET_TEST_DEBUG) process.stderr.write(chunk); - }); - broker.stderr.on('data', (chunk) => { - logs = (logs + chunk).slice(-16000); - if (process.env.FLEET_TEST_DEBUG) process.stderr.write(chunk); - }); - try { - const deadline = Date.now() + 10_000; - while (!nodeSocket && Date.now() < deadline && !spawnError && broker.exitCode === null) { - await new Promise((resolve) => setTimeout(resolve, 20)); - } - assert.ifError(spawnError); - assert(nodeSocket, `broker did not connect: ${logs}`); - const node = { - id: 'node-ready', - name: 'ready-node', - status: 'online', - live: true, - handlers_live: true, - capabilities: [{ name: 'spawn:claude', kind: 'spawn' }], - }; - let invocations = 0; - const client = new RelaycastMessagingClient({ - relaycast: { nodes: { list: async () => [node], get: async () => node } } as never, - agentClient: { - actions: { - invoke: async (_name: string, input: Record) => { - const id = `inv-${++invocations}`; - nodeSocket!.send( - JSON.stringify({ v: 1, type: 'action.invoke', invocation_id: id, action: 'spawn', input }) - ); - return { invocation_id: id, status: 'invoked', handler_node_id: node.id }; - }, - getInvocation: async (_name: string, id: string) => - receipts.get(id) ?? { invocation_id: id, status: 'invoked' }, + const wss = new WebSocketServer({ server }); + wss.on('connection', (socket, request) => { + const isNode = request.url?.startsWith('/v1/node/ws'); + socket.on('message', (raw) => { + const frame = JSON.parse(raw.toString()); + if (isNode && frame.type === 'inventory.sync') nodeSocket = socket; + if (frame.type === 'action.result') + receipts.set(frame.invocation_id, { + invocation_id: frame.invocation_id, + status: frame.error ? 'failed' : 'completed', + output: frame.output, + error: frame.error, + }); + if ( + ['node.register', 'inventory.sync', 'agent.register', 'agent.deregister'].includes(frame.type) + ) { + if (frame.type === 'agent.register') registrations++; + socket.send( + JSON.stringify({ + v: 1, + type: 'reply', + id: frame.id, + ok: true, + data: + frame.type === 'agent.register' + ? { + agent_id: `worker-${registrations}`, + token: 'at_worker_fixture', + name: frame.name, + delivery_ack_seq: 0, + } + : {}, + }) + ); + } + }); + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const address = server.address(); + assert(address && typeof address !== 'string'); + const baseUrl = `http://127.0.0.1:${address.port}`; + const broker = spawn( + binary, + [ + 'init', + '--instance-name', + 'ready-node', + '--workspace-key', + 'rk_fixture', + '--state-dir', + directory, + '--api-port', + '0', + '--channels', + '', + ], + { + cwd: directory, + env: { + PATH: `${bin}:${process.env.PATH}`, + HOME: directory, + TMPDIR: directory, + RELAYCAST_BASE_URL: baseUrl, + RELAY_BASE_URL: baseUrl, + RELAY_BROKER_API_KEY: 'br_fixture', + RELAY_NODE_ID: 'node-ready', + RELAY_NODE_TOKEN: 'nt_fixture', + AGENT_RELAY_BROKER_LOG: 'stderr', + AGENT_RELAY_TELEMETRY_DISABLED: '1', + AGENT_RELAY_NO_DEBUG_FILES: '1', }, - } as never, + stdio: ['ignore', 'pipe', 'pipe'], + } + ); + let logs = ''; + let spawnError: Error | undefined; + broker.on('error', (error) => { + spawnError = error; + }); + broker.stdout.on('data', (chunk) => { + logs = (logs + chunk).slice(-16000); + if (process.env.FLEET_TEST_DEBUG) process.stderr.write(chunk); }); - const result = await client.placement.spawn({ - capability: 'spawn:claude', - node: node.name, - confirm: true, - input: { name: 'Probe' }, + broker.stderr.on('data', (chunk) => { + logs = (logs + chunk).slice(-16000); + if (process.env.FLEET_TEST_DEBUG) process.stderr.write(chunk); }); - assert.equal(result.placement.state, 'ready', logs); - assert.equal(invocations, 1); - assert.equal(registrations, 1); - assert.deepEqual(result.confirmation?.output, { spawned: true, ready: true, name: 'Probe' }); - } catch (error) { - throw new Error(`${String(error)}\nBroker output:\n${logs}`, { cause: error }); - } finally { - if (broker.exitCode === null && !spawnError) { - const exited = new Promise((resolve) => broker.once('exit', resolve)); - broker.kill('SIGTERM'); - const timer = setTimeout(() => broker.kill('SIGKILL'), 2000); - await exited; - clearTimeout(timer); + try { + const deadline = Date.now() + 10_000; + while (!nodeSocket && Date.now() < deadline && !spawnError && broker.exitCode === null) { + await new Promise((resolve) => setTimeout(resolve, 20)); + } + assert.ifError(spawnError); + assert(nodeSocket, `broker did not connect: ${logs}`); + const node = { + id: 'node-ready', + name: 'ready-node', + status: 'online', + live: true, + handlers_live: true, + capabilities: [{ name: `spawn:${cli}`, kind: 'spawn' }], + }; + let invocations = 0; + const client = new RelaycastMessagingClient({ + relaycast: { nodes: { list: async () => [node], get: async () => node } } as never, + agentClient: { + actions: { + invoke: async (_name: string, input: Record) => { + const id = `inv-${++invocations}`; + nodeSocket!.send( + JSON.stringify({ v: 1, type: 'action.invoke', invocation_id: id, action: 'spawn', input }) + ); + return { invocation_id: id, status: 'invoked', handler_node_id: node.id }; + }, + getInvocation: async (_name: string, id: string) => + receipts.get(id) ?? { invocation_id: id, status: 'invoked' }, + }, + } as never, + }); + const result = await client.placement.spawn({ + capability: `spawn:${cli}`, + node: node.name, + confirm: true, + input: { + name: 'Probe', + cli, + ...(task ? { task } : {}), + ...(explicit ? { harnessConfig: { runtime: 'pty', command: path.join(bin, cli) } } : {}), + }, + }); + assert.equal(result.placement.state, 'ready', logs); + // Readiness must complete while the first task is still working, and + // no duplicate initial turn may arrive over the PTY after that ACK. + await new Promise((resolve) => setTimeout(resolve, 500)); + const submitted = JSON.parse(await readFile(turnFile, 'utf8')); + assert.deepEqual(submitted.turns, task ? [task] : []); + assert.equal(submitted.input, '', 'startup task must never be typed through the PTY'); + assert.equal(invocations, 1); + assert.equal(registrations, 1); + assert.deepEqual(result.confirmation?.output, { spawned: true, ready: true, name: 'Probe' }); + } catch (error) { + throw new Error(`${String(error)}\nBroker output:\n${logs}`, { cause: error }); + } finally { + if (broker.exitCode === null && !spawnError) { + const exited = new Promise((resolve) => broker.once('exit', resolve)); + broker.kill('SIGTERM'); + const timer = setTimeout(() => broker.kill('SIGKILL'), 2000); + await exited; + clearTimeout(timer); + } + for (const socket of wss.clients) socket.terminate(); + await new Promise((resolve) => wss.close(() => resolve())); + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + await rm(directory, { recursive: true, force: true }); } - for (const socket of wss.clients) socket.terminate(); - await new Promise((resolve) => wss.close(() => resolve())); - server.closeAllConnections(); - await new Promise((resolve) => server.close(() => resolve())); - await rm(directory, { recursive: true, force: true }); } - } -); + ); +}