diff --git a/docs/goals-system-plan.md b/docs/goals-system-plan.md index 0ea82455..24975b9f 100644 --- a/docs/goals-system-plan.md +++ b/docs/goals-system-plan.md @@ -175,6 +175,19 @@ rotation) we're ahead of both. Imported the rest as persona edits (`SKILL/SPEC/I --- +## Live-org fixes (2026-07-24) + +Two failures found driving a real org (canora-sync, goals `gp6`/`gp7`) — both in the plumbing +around the org, not the org itself: + +| Symptom | Root cause | Fix | +|---|---|---| +| Reports reach the orchestrator's inbox but it never acts on them | The bus is **pull-only**: `POST /panes/:id/messages` stores + pings WS clients, nothing writes to the pane. An interactive `claude` orchestrator blocks on stdin the moment its turn ends, so mail sat unread (88 messages deep) until a human typed. | `control/nudge.rs` + `arm_inbox_nudge` in `routes.rs`: when mail lands for an opted-in pane (`meta.role` in `goals-orch`/`spec`, or `hp.nudge=on`), wait for the pane to leave `Busy`, then type ONE coalesced, rate-limited (`60s`) line naming the read cursor. `HYPERPANES_MSG_NUDGE=0` disables. | +| Replies to a spec agent 404 (`no such pane`) | **Two pane-id spellings**: app-created panes are `pane-`, control-created ones bare ``. An agent handed one and reconstructing the other addressed a queue nobody reads. | `ReadModel::resolve_pane_id` canonicalizes either spelling (and a session uid); `find_pane_scoped` returns the canonical id, so post and read land on the same inbox. | +| Impl-agent work is invisible: no chat, and the pane is gone afterwards | `spawn_workers` put **N agents in one pane**; `claude -p` prints nothing until it exits; the runner exits on drain and the pane auto-closes with its scrollback. | `hyperpanes worker` gains `--stream` (render Claude `stream-json` events as progress), `--log-dir` (per-task raw transcript that outlives the pane) and `--linger`; `spawn_workers` defaults to `layout:"pane-per-worker"` (one pane per agent) and exposes `stream`/`logDir`/`lingerSecs`. Personas pass `--output-format stream-json --verbose`. | + +--- + ## Build order 1. **Queue plumbing (B)** — `work_db()` path, disk-backed `Shared::default`, boot recovery, reaper diff --git a/resources/claude/goal-orchestrator/IMPL.md b/resources/claude/goal-orchestrator/IMPL.md index 0471798a..4cfa02e8 100644 --- a/resources/claude/goal-orchestrator/IMPL.md +++ b/resources/claude/goal-orchestrator/IMPL.md @@ -14,7 +14,10 @@ done. Fix until it passes. 3. **Commit** your work on your worktree's branch with a clear message. Leave the branch for the spec agent to integrate — do not push, do not merge to main. -4. **Exit 0 on success**, non-zero on genuine failure. The runner acks on 0 (subtask `done`, which +4. **Narrate as you go.** Your pane renders your turn live (the runner is started with `--stream` + against `--output-format stream-json`), so a short line before each phase — what you're about + to do and why — is what the human sees. Silence for ten minutes reads as a wedged pane. +5. **Exit 0 on success**, non-zero on genuine failure. The runner acks on 0 (subtask `done`, which unblocks any dependents) and nacks on non-zero (requeue with backoff, or dead-letter after retries). Your printed last line is recorded — make it a one-line summary (what changed + the branch/commit). @@ -29,7 +32,8 @@ the spec is a higher-tier (opus/fable) model and is live. This is the "plan big, trade — one cheap round-trip buys an opus-grade decision without taking the build off sonnet. - Your advisor's pane id is in the payload (the spec agent stamped `advisor=`); your own - pane id is `$HYPERPANES_PANE_ID`. + pane id is `$HYPERPANES_PANE_ID`. Pass both **verbatim** — don't add or strip a `pane-` prefix + (the API tolerates either spelling now, but only the exact id is guaranteed to be your queue). - Ask one tight, decidable question — propose your answer, don't write an essay: `send_message {to:"", from:"$HYPERPANES_PANE_ID", body:": "}`. diff --git a/resources/claude/goal-orchestrator/SKILL.md b/resources/claude/goal-orchestrator/SKILL.md index e7847ac3..31b6540f 100644 --- a/resources/claude/goal-orchestrator/SKILL.md +++ b/resources/claude/goal-orchestrator/SKILL.md @@ -65,7 +65,12 @@ For each goal you're given (free text): Your own pane already carries this identity (the app set it); keep the scheme for everything you spawn so a glance at the workspace reads project → task. 3. **Ingest reports.** Read spec-agent messages (`read_messages` on your pane; spec agents - `send_to_parent`). A report is one of: `progress` (incl. `spec:`/`respec:`), `blocked `, + `send_to_parent`). The bus is pull-only, so the app helps: when mail lands for your pane while + you're idle it types a one-line `[hyperpanes] inbox: N new message(s) … read_messages {paneId, + after:}` nudge into you. **Treat that line as a work order** — read from the given cursor + and act before anything else. It is coalesced (one line per burst) and rate-limited, so still + poll `read_messages` yourself on every loop pass; never assume the nudge is your only signal. + A report is one of: `progress` (incl. `spec:`/`respec:`), `blocked `, `needs-decision `, `done `, `failed `. Act: - `progress` — update ledger, continue. - `needs-decision` — answer from the goal intent if you can; otherwise surface to the human @@ -130,7 +135,9 @@ plan doc), so you don't babysit individual tasks — you watch goals and health. **Impl-agent pane budget:** fan-out is soft-capped at **16 worker panes** — the spec agent sets `count <= 16` and the queue multiplexes any overflow (competing-consumers), so more subtasks than 16 -drain through the 16 panes rather than opening more. It's persona-enforced (see `SPEC.md` section 2), +drain through the 16 panes rather than opening more. `spawn_workers` now gives each worker its own +pane (`layout:"pane-per-worker"`, the default), so `count` IS the pane count — one readable agent +per pane instead of N interleaved into one. It's persona-enforced (see `SPEC.md` section 2), not a code limit; hold the line so concurrent goals don't explode the pane count. ### MCP config on every spawned claude @@ -143,7 +150,7 @@ the default `~/.claude.json` once `CLAUDE_CONFIG_DIR` is set — without the fla every `mcp__hyperpanes__*` tool. The app already appends it on your own spawn; pass it down the same way when you spawn a spec agent, and tell the spec agent to do the same in its `spawn_workers` command, e.g.: -`spawn_workers {queue, count:N, isolation:"worktree", command:"sh -c 'claude --dangerously-skip-permissions --mcp-config /goals-mcp.json -p \"$HP_TASK_PAYLOAD\" --append-system-prompt-file $HP_GOAL_PERSONA_DIR/IMPL.md ${HP_GOAL_SETTINGS:+--settings $HP_GOAL_SETTINGS} --model ${HP_GOAL_IMPL_MODEL:-claude-sonnet-5[1m]}'"}` +`spawn_workers {queue, count:N, isolation:"worktree", stream:true, lingerSecs:120, command:"sh -c 'claude --dangerously-skip-permissions --mcp-config /goals-mcp.json -p \"$HP_TASK_PAYLOAD\" --output-format stream-json --verbose --append-system-prompt-file $HP_GOAL_PERSONA_DIR/IMPL.md ${HP_GOAL_SETTINGS:+--settings $HP_GOAL_SETTINGS} --model ${HP_GOAL_IMPL_MODEL:-claude-sonnet-5[1m]}'"}` ### If the `mcp__hyperpanes__*` tools won't load — drop to the Control API, don't reverse-engineer diff --git a/resources/claude/goal-orchestrator/SPEC.md b/resources/claude/goal-orchestrator/SPEC.md index 229b1d4e..dffc5b6b 100644 --- a/resources/claude/goal-orchestrator/SPEC.md +++ b/resources/claude/goal-orchestrator/SPEC.md @@ -53,8 +53,15 @@ prompt wedges the pane). the advisor:** include `advisor=` in every payload so an impl agent that hits a strategic fork can consult you mid-build instead of guessing or bouncing the whole subtask (see IMPL.md "Consult your advisor"). -- `spawn_workers {queue, count:N, isolation:"worktree", command:"sh -c 'claude --dangerously-skip-permissions --mcp-config /goals-mcp.json -p \"$HP_TASK_PAYLOAD\" --append-system-prompt-file $HP_GOAL_PERSONA_DIR/IMPL.md ${HP_GOAL_SETTINGS:+--settings $HP_GOAL_SETTINGS} --model ${HP_GOAL_IMPL_MODEL:-claude-sonnet-5[1m]}'"}` - — the `--mcp-config` flag is required (see `SKILL.md` "MCP config on every spawned claude"); +- `spawn_workers {queue, count:N, isolation:"worktree", stream:true, lingerSecs:120, command:"sh -c 'claude --dangerously-skip-permissions --mcp-config /goals-mcp.json -p \"$HP_TASK_PAYLOAD\" --output-format stream-json --verbose --append-system-prompt-file $HP_GOAL_PERSONA_DIR/IMPL.md ${HP_GOAL_SETTINGS:+--settings $HP_GOAL_SETTINGS} --model ${HP_GOAL_IMPL_MODEL:-claude-sonnet-5[1m]}'"}` + — **keep the visibility trio**: `stream:true` + `--output-format stream-json --verbose` makes the + impl agent's turn readable in its pane (a bare `claude -p` prints nothing until it exits, so the + pane looks dead for the whole build), and `lingerSecs` holds the pane open after the queue drains + (the pane auto-closes when the runner exits, taking the scrollback with it). Add + `logDir:"/worker-logs"` when you want the raw transcript to outlive the pane. + `spawn_workers` gives each worker **its own pane** by default (`layout:"pane-per-worker"`), so + `count:N` = N readable panes; `layout:"single-pane"` multiplexes them into one if you'd rather. + The `--mcp-config` flag is required (see `SKILL.md` "MCP config on every spawned claude"); without it, account rotation hides `mcp__hyperpanes__*` tools from the impl agent. `${HP_GOAL_SETTINGS:+--settings $HP_GOAL_SETTINGS}` likewise carries the user's statusline (see `SKILL.md` "Statusline on every spawned claude") — harmless when the var is unset. @@ -80,7 +87,9 @@ prompt wedges the pane). ## 3. Integrate & verify - **Be the impl agents' advisor while the wave runs.** You're the higher-tier model that wrote the - spec, so you're on call: watch your inbox (`read_messages {paneId:}`) + spec, so you're on call. The app types a one-line `[hyperpanes] inbox: N new message(s)…` nudge + into your pane when mail lands while you're idle — when you see it, read and answer immediately. + Don't rely on it alone: also watch your inbox (`read_messages {paneId:}`) for `:` consults and answer fast (`send_message {to:, from:"$HYPERPANES_PANE_ID", body:}`). A 20-second answer here saves a thrown-away subtask and a whole re-spec round-trip — this is the point of pairing your intelligence with their diff --git a/rs/crates/app/src/worker.rs b/rs/crates/app/src/worker.rs index 1ca85504..fc55da6c 100644 --- a/rs/crates/app/src/worker.rs +++ b/rs/crates/app/src/worker.rs @@ -3,7 +3,8 @@ //! Usage: //! ```text //! hyperpanes worker --queue [--worker ] [--count N] [--worktree] \ -//! [--retry-window ] [--nack-delay ] -- [args...] +//! [--retry-window ] [--nack-delay ] \ +//! [--stream] [--log-dir ] [--linger ] -- [args...] //! ``` //! //! Discovers the running app's control API from `control.json` (or `HYPERPANES_CONTROL_FILE`), @@ -16,7 +17,17 @@ //! task in a throwaway git worktree that auto-removes (#14); `--retry-window ` keeps //! polling after the queue empties so backoff retries get reclaimed, and `--nack-delay ` //! overrides the retry backoff (#13). A lease heartbeat renews the lease while a task runs so a -//! long task isn't reclaimed mid-flight (#12). The child reads its task from the environment, so +//! long task isn't reclaimed mid-flight (#12). +//! +//! Visibility flags (a headless `claude -p` child prints nothing until it exits, and the pane +//! auto-closes the moment the queue drains, so a run could finish leaving no trace): +//! `--stream` renders the child's Claude `--output-format stream-json` events as readable +//! progress lines (non-JSON output passes through); `--log-dir ` tees every child's raw +//! output to `/-.log`, which outlives the pane; `--linger ` holds the +//! process open after the drain so the pane stays readable. All three default off = the original +//! inherit-stdio, exit-on-drain behaviour. +//! +//! The child reads its task from the environment, so //! shell expansion like `$HP_TASK_PAYLOAD` needs an explicit inner shell: //! `-- sh -c 'claude -p "$HP_TASK_PAYLOAD"'`. @@ -43,6 +54,15 @@ pub struct WorkerArgs { pub nack_delay_ms: Option, /// Run each task in a throwaway git worktree, auto-removed on exit (#14). pub worktree: bool, + /// Render the child's Claude `--output-format stream-json` lines as readable progress + /// instead of raw JSON, so a worker pane shows what the agent is doing while it runs. + pub stream: bool, + /// Tee every child's raw output to `/-.log`, so the transcript survives + /// the pane closing when the queue drains. `None` = no logs (frozen behaviour). + pub log_dir: Option, + /// Stay alive this many seconds after the queue drains, so the pane (which auto-closes when + /// this process exits) stays readable. 0 = exit immediately, as before. + pub linger_secs: u64, /// Everything after `--`: program + args, executed directly (no shell). pub child: Vec, } @@ -93,6 +113,9 @@ pub fn parse_args(argv: &[String]) -> Result { let mut count_arg: Option = None; let mut retry_window_arg: Option = None; let mut nack_delay_arg: Option = None; + let mut linger_arg: Option = None; + let mut log_dir: Option = None; + let mut stream = false; let mut worktree = false; let mut child: Vec = Vec::new(); let mut i = 2; @@ -112,8 +135,11 @@ pub fn parse_args(argv: &[String]) -> Result { i += 2; } "--retry-window" => { - retry_window_arg = - Some(argv.get(i + 1).ok_or("--retry-window needs a value")?.clone()); + retry_window_arg = Some( + argv.get(i + 1) + .ok_or("--retry-window needs a value")? + .clone(), + ); i += 2; } "--nack-delay" => { @@ -124,6 +150,20 @@ pub fn parse_args(argv: &[String]) -> Result { worktree = true; i += 1; } + "--stream" => { + stream = true; + i += 1; + } + "--log-dir" => { + log_dir = Some(PathBuf::from( + argv.get(i + 1).ok_or("--log-dir needs a value")?, + )); + i += 2; + } + "--linger" => { + linger_arg = Some(argv.get(i + 1).ok_or("--linger needs a value")?.clone()); + i += 2; + } "--" => { child = argv[i + 1..].to_vec(); break; @@ -139,6 +179,10 @@ pub fn parse_args(argv: &[String]) -> Result { retry_window_arg = Some(v.to_string()); } else if let Some(v) = other.strip_prefix("--nack-delay=") { nack_delay_arg = Some(v.to_string()); + } else if let Some(v) = other.strip_prefix("--log-dir=") { + log_dir = Some(PathBuf::from(v)); + } else if let Some(v) = other.strip_prefix("--linger=") { + linger_arg = Some(v.to_string()); } else { return Err(format!("unexpected argument: {other}")); } @@ -176,6 +220,12 @@ pub fn parse_args(argv: &[String]) -> Result { ), None => None, }; + let linger_secs = match linger_arg { + Some(s) => s + .parse() + .map_err(|_| format!("--linger must be a non-negative integer (secs), got '{s}'"))?, + None => 0, + }; Ok(WorkerArgs { queue, worker, @@ -183,6 +233,9 @@ pub fn parse_args(argv: &[String]) -> Result { retry_window_secs, nack_delay_ms, worktree, + stream, + log_dir, + linger_secs, child, }) } @@ -232,6 +285,19 @@ pub fn run(argv: &[String]) -> Result<(), Box> { let retry_window = Duration::from_secs(args.retry_window_secs); let nack_delay = args.nack_delay_ms; let worktree = args.worktree; + // Output policy shared by every worker thread: render Claude's stream-json readably and/or + // tee raw child output to a per-task log that outlives the pane. + let out = Arc::new(ChildOutput { + stream: args.stream, + log_dir: args.log_dir.clone(), + }); + if let Some(dir) = &out.log_dir { + if let Err(e) = std::fs::create_dir_all(dir) { + eprintln!("worker: cannot create --log-dir {}: {e}", dir.display()); + } else { + eprintln!("worker: per-task logs in {}", dir.display()); + } + } // One worker drains in this thread; `--count N` spawns N competing workers (#11), each // with its own id, and the process exits once they have all seen the queue empty. @@ -246,11 +312,13 @@ pub fn run(argv: &[String]) -> Result<(), Box> { retry_window, nack_delay, worktree, + &out, )?; eprintln!( "[{}] queue drained — {done} task(s) acked, exiting", args.worker ); + linger(args.linger_secs); return Ok(()); } @@ -266,9 +334,19 @@ pub fn run(argv: &[String]) -> Result<(), Box> { let queue = args.queue.clone(); let child = args.child.clone(); let worker = format!("{}-{i}", args.worker); + let out = Arc::clone(&out); handles.push(std::thread::spawn(move || { match drain( - &client, &base, &token, &queue, &worker, &child, retry_window, nack_delay, worktree, + &client, + &base, + &token, + &queue, + &worker, + &child, + retry_window, + nack_delay, + worktree, + &out, ) { Ok(n) => { eprintln!("[{worker}] drained {n} task(s)"); @@ -283,9 +361,20 @@ pub fn run(argv: &[String]) -> Result<(), Box> { } let total: u64 = handles.into_iter().filter_map(|h| h.join().ok()).sum(); eprintln!("all {} workers exited — {total} task(s) total", args.count); + linger(args.linger_secs); Ok(()) } +/// Hold the process (and therefore its pane, which auto-closes on exit) open after the drain, so +/// the run stays readable. No-op at 0. +fn linger(secs: u64) { + if secs == 0 { + return; + } + eprintln!("worker: holding this pane open for {secs}s (--linger)"); + std::thread::sleep(Duration::from_secs(secs)); +} + /// One worker's claim → run → ack/nack loop. Returns the number of tasks acked; stops when a /// claim comes back empty. Shared by the single-worker and `--count` paths. #[allow(clippy::too_many_arguments)] @@ -299,6 +388,7 @@ fn drain( retry_window: Duration, nack_delay_ms: Option, worktree: bool, + out: &ChildOutput, ) -> Result> { eprintln!("[{worker}] online — draining '{queue}'"); let mut done: u64 = 0; @@ -344,7 +434,7 @@ fn drain( None }; let cwd = wt.as_ref().map(|w| w.path.as_path()); - let outcome = run_child(child, &task, queue, client, base, token, cwd); + let outcome = run_child(child, &task, queue, client, base, token, cwd, worker, out); if let Some(w) = &wt { w.remove(); } @@ -400,8 +490,30 @@ fn claim_one( Ok(out.tasks.into_iter().next()) } +/// How a child's output reaches the operator. Default (`stream: false`, `log_dir: None`) inherits +/// the worker's stdio — byte-for-byte the original behaviour. Either option switches to piped +/// stdio so the runner can render and/or persist what the child says. +#[derive(Debug, Default)] +pub struct ChildOutput { + pub stream: bool, + pub log_dir: Option, +} + +impl ChildOutput { + fn piped(&self) -> bool { + self.stream || self.log_dir.is_some() + } + /// Where this task's raw transcript is kept (so it outlives the pane). + fn log_path(&self, queue: &str, task_id: &str) -> Option { + self.log_dir + .as_ref() + .map(|d| d.join(format!("{queue}-{}.log", short(task_id)))) + } +} + /// Run the child command with the task in its environment, while a background heartbeat renews /// the lease (#12) so a long-running task is not reclaimed mid-flight. Returns Ok(true) on exit 0. +#[allow(clippy::too_many_arguments)] fn run_child( child: &[String], task: &Task, @@ -410,6 +522,8 @@ fn run_child( base: &str, token: &str, cwd: Option<&Path>, + worker: &str, + out: &ChildOutput, ) -> Result> { // Heartbeat: while the child runs, `extend` the lease at ~half the remaining lease interval. let stop = Arc::new(AtomicBool::new(false)); @@ -443,16 +557,176 @@ fn run_child( if let Some(dir) = cwd { cmd.current_dir(dir).env("HP_WORKTREE", dir); } - let result = cmd - .status() - .map_err(|e| format!("failed to spawn '{}': {e}", child[0])); + let result = if out.piped() { + run_child_piped(cmd, task, queue, worker, out) + } else { + cmd.status() + .map(|s| s.success()) + .map_err(|e| format!("failed to spawn '{}': {e}", child[0])) + }; // Stop the heartbeat before ack/nack so we never extend a finished task. stop.store(true, Ordering::Relaxed); if let Some(h) = heartbeat { let _ = h.join(); } - Ok(result?.success()) + Ok(result?) +} + +/// Piped variant of the child run: stdout+stderr are captured line-by-line so each line can be +/// (a) appended raw to this task's log — the transcript that survives the pane closing — and +/// (b) rendered into the pane. With `--stream`, Claude `--output-format stream-json` lines become +/// readable progress; anything else passes through unchanged, so a plain shell task still looks +/// the same. Every line is prefixed with the worker + task so `--count N` interleaving is legible. +fn run_child_piped( + mut cmd: Command, + task: &Task, + queue: &str, + worker: &str, + out: &ChildOutput, +) -> Result { + use std::io::{BufRead, BufReader, Write}; + use std::process::Stdio; + + let log = out.log_path(queue, &task.id).and_then(|p| { + std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&p) + .map_err(|e| eprintln!("[{worker}] cannot open log {}: {e}", p.display())) + .ok() + .map(|f| (p, Arc::new(std::sync::Mutex::new(f)))) + }); + if let Some((p, _)) = &log { + eprintln!("[{worker}] log: {}", p.display()); + } + + let mut c = cmd + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .map_err(|e| format!("failed to spawn child: {e}"))?; + + let tag = format!("[{worker}/{}]", short(&task.id)); + let stream = out.stream; + let mut pumps = Vec::new(); + for (reader, is_stderr) in [ + ( + c.stdout + .take() + .map(|s| Box::new(s) as Box), + false, + ), + ( + c.stderr + .take() + .map(|s| Box::new(s) as Box), + true, + ), + ] { + let Some(reader) = reader else { continue }; + let tag = tag.clone(); + let log = log.as_ref().map(|(_, f)| Arc::clone(f)); + pumps.push(std::thread::spawn(move || { + for line in BufReader::new(reader).lines().map_while(Result::ok) { + if let Some((_, f)) = log.as_ref().map(|f| ((), f)) { + if let Ok(mut f) = f.lock() { + let _ = writeln!(f, "{line}"); + } + } + // stderr is already human text; only stdout carries the JSON stream. + match if stream && !is_stderr { + render_stream_line(&line) + } else { + Some(line) + } { + Some(text) if !text.is_empty() => eprintln!("{tag} {text}"), + _ => {} + } + } + })); + } + let status = c.wait().map_err(|e| format!("child wait failed: {e}"))?; + for p in pumps { + let _ = p.join(); + } + Ok(status.success()) +} + +/// Render one line of Claude's `--output-format stream-json` into a short human line, or `None` +/// to drop it. A line that isn't such an event (plain program output) is returned unchanged, so +/// this is safe to run over any child's stdout. +fn render_stream_line(line: &str) -> Option { + let trimmed = line.trim(); + if !trimmed.starts_with('{') { + return Some(line.to_string()); + } + let v: serde_json::Value = match serde_json::from_str(trimmed) { + Ok(v) => v, + Err(_) => return Some(line.to_string()), + }; + match v.get("type").and_then(serde_json::Value::as_str) { + Some("system") => { + let model = v.get("model").and_then(serde_json::Value::as_str)?; + Some(format!("▶ started ({model})")) + } + Some("assistant") => { + let content = v.get("message")?.get("content")?.as_array()?; + let mut parts: Vec = Vec::new(); + for block in content { + match block.get("type").and_then(serde_json::Value::as_str) { + Some("text") => { + let t = block.get("text").and_then(serde_json::Value::as_str)?; + let t = t.trim(); + if !t.is_empty() { + parts.push(one_line(t, 300)); + } + } + Some("tool_use") => { + let name = block + .get("name") + .and_then(serde_json::Value::as_str) + .unwrap_or("tool"); + parts.push(format!("⚙ {name}")); + } + _ => {} + } + } + if parts.is_empty() { + None + } else { + Some(parts.join(" ")) + } + } + Some("result") => { + let ok = v + .get("is_error") + .and_then(serde_json::Value::as_bool) + .map(|e| !e) + .unwrap_or(true); + let text = v + .get("result") + .and_then(serde_json::Value::as_str) + .unwrap_or(""); + Some(format!( + "{} {}", + if ok { "✓ done:" } else { "✗ failed:" }, + one_line(text, 300) + )) + } + // user turns are tool results — noise in a progress view. + _ => None, + } +} + +/// Collapse whitespace and clip to `max` chars, so one rendered event stays one readable line. +fn one_line(s: &str, max: usize) -> String { + let flat = s.split_whitespace().collect::>().join(" "); + if flat.chars().count() <= max { + return flat; + } + let clipped: String = flat.chars().take(max).collect(); + format!("{clipped}…") } /// Sleep up to `ms`, waking early and returning `true` if `stop` gets set; `false` on timeout. @@ -558,8 +832,7 @@ impl Worktree { let id8 = &task_id[..8.min(task_id.len())]; let safe = queue.replace(['/', ' '], "-"); let branch = format!("worker/{safe}/{id8}"); - let path = - std::env::temp_dir().join(format!("hp-wt-{safe}-{id8}-{}", std::process::id())); + let path = std::env::temp_dir().join(format!("hp-wt-{safe}-{id8}-{}", std::process::id())); // The branch name is deterministic (per task id), so a stale `worker//` left by a // prior run makes `git worktree add -b` fail. Prune dead worktree admin entries first, then // decide by whether the stale branch carries uncollected commits: if it's an ancestor of @@ -642,15 +915,33 @@ mod tests { #[test] fn detects_worker_mode() { - assert!(wants_worker(&argv(&["hyperpanes", "worker", "--queue", "q"]))); + assert!(wants_worker(&argv(&[ + "hyperpanes", + "worker", + "--queue", + "q" + ]))); assert!(!wants_worker(&argv(&["hyperpanes"]))); - assert!(!wants_worker(&argv(&["hyperpanes", "--session-daemon", "x"]))); + assert!(!wants_worker(&argv(&[ + "hyperpanes", + "--session-daemon", + "x" + ]))); } #[test] fn parses_queue_worker_and_child() { let a = parse_args(&argv(&[ - "hp", "worker", "--queue", "hp-issues", "--worker", "w1", "--", "claude", "-p", "hi", + "hp", + "worker", + "--queue", + "hp-issues", + "--worker", + "w1", + "--", + "claude", + "-p", + "hi", ])) .unwrap(); assert_eq!(a.queue, "hp-issues"); @@ -699,9 +990,16 @@ mod tests { 4 ); assert_eq!( - parse_args(&argv(&["hp", "worker", "--queue=q", "--count=2", "--", "true"])) - .unwrap() - .count, + parse_args(&argv(&[ + "hp", + "worker", + "--queue=q", + "--count=2", + "--", + "true" + ])) + .unwrap() + .count, 2 ); assert!(parse_args(&argv(&[ @@ -737,17 +1035,123 @@ mod tests { assert_eq!(d.nack_delay_ms, None); assert!(parse_args(&argv(&[ - "hp", "worker", "--queue", "q", "--retry-window", "x", "--", "true" + "hp", + "worker", + "--queue", + "q", + "--retry-window", + "x", + "--", + "true" ])) .is_err()); } + #[test] + fn parses_visibility_flags_with_frozen_defaults() { + let a = parse_args(&argv(&[ + "hp", + "worker", + "--queue", + "q", + "--stream", + "--log-dir", + "/tmp/hp-logs", + "--linger", + "30", + "--", + "true", + ])) + .unwrap(); + assert!(a.stream); + assert_eq!(a.log_dir.as_deref(), Some(Path::new("/tmp/hp-logs"))); + assert_eq!(a.linger_secs, 30); + + let eq = parse_args(&argv(&[ + "hp", + "worker", + "--queue=q", + "--log-dir=/tmp/l", + "--linger=5", + "--", + "true", + ])) + .unwrap(); + assert_eq!(eq.log_dir.as_deref(), Some(Path::new("/tmp/l"))); + assert_eq!(eq.linger_secs, 5); + + // Defaults keep the original inherit-stdio, exit-on-drain behaviour. + let d = parse_args(&argv(&["hp", "worker", "--queue", "q", "--", "true"])).unwrap(); + assert!(!d.stream); + assert_eq!(d.log_dir, None); + assert_eq!(d.linger_secs, 0); + assert!(!ChildOutput::default().piped()); + assert!(parse_args(&argv(&[ + "hp", "worker", "--queue", "q", "--linger", "x", "--", "true" + ])) + .is_err()); + } + + #[test] + fn renders_claude_stream_json_events_and_passes_other_output_through() { + // Plain program output is untouched — a non-claude task still reads normally. + assert_eq!( + render_stream_line("cargo test: 12 passed").as_deref(), + Some("cargo test: 12 passed") + ); + // Malformed JSON is output, not an error. + assert_eq!( + render_stream_line("{not json").as_deref(), + Some("{not json") + ); + // Assistant text + tool calls collapse to one progress line. + let a = render_stream_line( + r#"{"type":"assistant","message":{"content":[{"type":"text","text":"Fixing the\n installer"},{"type":"tool_use","name":"Edit"}]}}"#, + ) + .unwrap(); + assert_eq!(a, "Fixing the installer ⚙ Edit"); + // Tool results (user turns) are dropped as noise. + assert_eq!( + render_stream_line(r#"{"type":"user","message":{"content":[]}}"#), + None + ); + // The final result line reports success/failure. + let ok = render_stream_line(r#"{"type":"result","is_error":false,"result":"branch x"}"#) + .unwrap(); + assert_eq!(ok, "✓ done: branch x"); + let bad = render_stream_line(r#"{"type":"result","is_error":true,"result":"build broke"}"#) + .unwrap(); + assert_eq!(bad, "✗ failed: build broke"); + } + + #[test] + fn per_task_log_path_is_queue_and_task_scoped() { + let out = ChildOutput { + stream: false, + log_dir: Some(PathBuf::from("/tmp/logs")), + }; + assert!(out.piped()); + assert_eq!( + out.log_path("g7", "0123456789abcdef"), + Some(PathBuf::from("/tmp/logs/g7-01234567.log")) + ); + assert_eq!(ChildOutput::default().log_path("g7", "abc"), None); + } + #[test] fn parses_worktree_flag() { assert!( - parse_args(&argv(&["hp", "worker", "--queue", "q", "--worktree", "--", "true"])) - .unwrap() - .worktree + parse_args(&argv(&[ + "hp", + "worker", + "--queue", + "q", + "--worktree", + "--", + "true" + ])) + .unwrap() + .worktree ); assert!( !parse_args(&argv(&["hp", "worker", "--queue", "q", "--", "true"])) diff --git a/rs/crates/core/src/control/mod.rs b/rs/crates/core/src/control/mod.rs index 0d4126f6..1f247256 100644 --- a/rs/crates/core/src/control/mod.rs +++ b/rs/crates/core/src/control/mod.rs @@ -4,6 +4,7 @@ pub mod inbox; pub mod input; pub mod lock; +pub mod nudge; pub mod output; pub mod scope; pub mod work; diff --git a/rs/crates/core/src/control/nudge.rs b/rs/crates/core/src/control/nudge.rs new file mode 100644 index 00000000..5267abf1 --- /dev/null +++ b/rs/crates/core/src/control/nudge.rs @@ -0,0 +1,239 @@ +//! Inbox nudges — waking an agent pane that has mail. +//! +//! The message bus (`control::inbox`) is PULL-only: a post lands in the target pane's durable +//! queue and pings live WS clients, but nothing writes to the pane's pty. That is fine for a +//! client that polls, and fatal for the goals org: an orchestrator/spec agent is an interactive +//! `claude` TUI, so the moment it finishes a turn it blocks on stdin forever. Reports from its +//! children pile up unread until a human types into the pane. +//! +//! A nudge closes that loop: when a message lands for a pane that has opted in (org roles — +//! `goals-orch` / `spec`, or an explicit `hp.nudge=on` meta), we wait for the pane to go quiet +//! and then type ONE short line into it telling it to read its inbox. Rules that keep this from +//! becoming a footgun: +//! • **Never mid-turn** — only when the pane's activity is not `Busy`, so we never inject into +//! a running command or a half-typed prompt. +//! • **Coalesced** — a burst of messages produces one nudge naming the whole seq range; only +//! one waiter task per pane is ever in flight. +//! • **Rate-limited** — at most one nudge per pane per [`MIN_INTERVAL_MS`], so a chatty wave of +//! impl agents can't turn into a typing storm. +//! • **Opt-out** — `hp.nudge=off` meta on the pane, or `HYPERPANES_MSG_NUDGE=0` in the app's +//! environment, disables it everywhere. +//! +//! This module is pure + in-memory so the policy is unit-testable without a server; the routes +//! layer owns the timer and the pty write. + +use std::collections::{BTreeMap, HashMap}; + +/// Pane `meta.role` values that get nudged by default — the goals-system org tiers that run as +/// interactive TUIs and are expected to act on their inbox. +pub const NUDGED_ROLES: &[&str] = &["goals-orch", "spec"]; + +/// Meta key an embedder/agent can set to force nudging on (`on`) or off (`off`) regardless of role. +pub const NUDGE_META_KEY: &str = "hp.nudge"; + +/// Never nudge the same pane more often than this, however much mail arrives. +pub const MIN_INTERVAL_MS: i64 = 60_000; + +/// How long a waiter re-checks the pane's activity before it gives up (the pane stayed busy the +/// whole time). The pending count survives — the next message re-arms a fresh waiter. +pub const MAX_WAIT_MS: i64 = 30 * 60_000; + +/// Settle window before the first delivery attempt: lets a burst of sibling reports coalesce into +/// one nudge, and keeps us from typing the instant a pane's own turn ends. +pub const SETTLE_MS: u64 = 3_000; + +/// How often a waiter re-checks whether the pane went quiet. +pub const POLL_MS: u64 = 2_000; + +/// Whether a pane with this meta wants inbox nudges. Explicit `hp.nudge` wins; otherwise the +/// org roles in [`NUDGED_ROLES`] opt in and everything else stays pull-only (frozen behaviour). +pub fn wants_nudge(meta: Option<&BTreeMap>) -> bool { + let Some(meta) = meta else { return false }; + match meta.get(NUDGE_META_KEY).map(String::as_str) { + Some("off") | Some("0") | Some("false") => return false, + Some("on") | Some("1") | Some("true") => return true, + _ => {} + } + meta.get("role") + .is_some_and(|r| NUDGED_ROLES.contains(&r.as_str())) +} + +/// The line typed into a nudged pane. One line, no shell metacharacters, and it names the exact +/// cursor to read from so the agent doesn't re-ingest its whole backlog. +pub fn nudge_text(pane_id: &str, pending: usize, first_seq: u64) -> String { + let plural = if pending == 1 { "" } else { "s" }; + format!( + "[hyperpanes] inbox: {pending} new message{plural}. Read with read_messages \ + {{paneId:\"{pane_id}\", after:{}}}, act on {} (report/consult/decision), then continue \ + your loop.", + first_seq.saturating_sub(1), + if pending == 1 { "it" } else { "them" } + ) +} + +/// One pane's nudge bookkeeping. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +struct PaneNudge { + /// Unread-since-last-nudge count (what the nudge line reports). + pending: usize, + /// Seq of the first message in the pending batch — the read cursor we hand the agent. + first_seq: u64, + /// A waiter task is in flight for this pane (so a burst spawns exactly one). + waiting: bool, + /// When the current waiter armed, for the [`MAX_WAIT_MS`] give-up. + armed_at_ms: i64, + /// When we last typed into this pane, for [`MIN_INTERVAL_MS`]. + last_sent_ms: Option, +} + +/// What a waiter should do on this poll. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Step { + /// Pane is busy / rate-limited — check again later. + Wait, + /// Type this line into the pane; the waiter is done afterwards. + Send(String), + /// Nothing pending, or the waiter hit [`MAX_WAIT_MS`]; the waiter is done. + Stop, +} + +/// Per-pane nudge state for the whole server. Pure — the caller supplies `now` and the pane's +/// current activity, and performs the actual write. +#[derive(Debug, Default)] +pub struct NudgeLedger { + panes: HashMap, +} + +impl NudgeLedger { + pub fn new() -> Self { + Self::default() + } + + /// Record a message for `pane_id`. Returns `true` if the caller should spawn a waiter task + /// (no other waiter is in flight for this pane). + pub fn arm(&mut self, pane_id: &str, seq: u64, now_ms: i64) -> bool { + let e = self.panes.entry(pane_id.to_string()).or_default(); + if e.pending == 0 { + e.first_seq = seq; + } + e.pending += 1; + if e.waiting { + return false; + } + e.waiting = true; + e.armed_at_ms = now_ms; + true + } + + /// One waiter poll. `busy` is the pane's live activity (true ⇒ mid-turn, never type). + /// On [`Step::Send`] the pending batch is consumed and the waiter retires. + pub fn poll(&mut self, pane_id: &str, busy: bool, now_ms: i64) -> Step { + let Some(e) = self.panes.get_mut(pane_id) else { + return Step::Stop; + }; + if e.pending == 0 { + e.waiting = false; + return Step::Stop; + } + let rate_limited = e.last_sent_ms.is_some_and(|t| now_ms - t < MIN_INTERVAL_MS); + if busy || rate_limited { + if now_ms - e.armed_at_ms >= MAX_WAIT_MS { + // Give up this round; the pending batch stays, so the next message re-arms. + e.waiting = false; + return Step::Stop; + } + return Step::Wait; + } + let text = nudge_text(pane_id, e.pending, e.first_seq); + e.pending = 0; + e.first_seq = 0; + e.waiting = false; + e.last_sent_ms = Some(now_ms); + Step::Send(text) + } + + /// Forget a pane (on close), so a closed pane's counters don't leak. + pub fn drop_pane(&mut self, pane_id: &str) { + self.panes.remove(pane_id); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn meta(pairs: &[(&str, &str)]) -> BTreeMap { + pairs + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect() + } + + #[test] + fn org_roles_opt_in_and_explicit_meta_wins_both_ways() { + assert!(wants_nudge(Some(&meta(&[("role", "goals-orch")])))); + assert!(wants_nudge(Some(&meta(&[("role", "spec")])))); + // Impl/worker panes and plain panes stay pull-only. + assert!(!wants_nudge(Some(&meta(&[("role", "impl")])))); + assert!(!wants_nudge(Some(&meta(&[])))); + assert!(!wants_nudge(None)); + // Explicit override in both directions. + assert!(!wants_nudge(Some(&meta(&[ + ("role", "spec"), + (NUDGE_META_KEY, "off") + ])))); + assert!(wants_nudge(Some(&meta(&[(NUDGE_META_KEY, "on")])))); + } + + #[test] + fn a_burst_arms_one_waiter_and_coalesces_into_one_nudge() { + let mut l = NudgeLedger::new(); + assert!(l.arm("p1", 10, 0)); // first message spawns the waiter + assert!(!l.arm("p1", 11, 100)); // burst rides along + assert!(!l.arm("p1", 12, 200)); + let text = match l.poll("p1", false, 5_000) { + Step::Send(t) => t, + other => panic!("expected Send, got {other:?}"), + }; + assert!(text.contains("3 new messages"), "{text}"); + // Cursor is first_seq - 1, so read_messages returns exactly the batch. + assert!(text.contains("after:9"), "{text}"); + // Batch consumed; the waiter retires. + assert_eq!(l.poll("p1", false, 6_000), Step::Stop); + } + + #[test] + fn never_types_into_a_busy_pane_and_gives_up_after_the_max_wait() { + let mut l = NudgeLedger::new(); + l.arm("p1", 1, 0); + assert_eq!(l.poll("p1", true, 1_000), Step::Wait); + assert_eq!(l.poll("p1", true, MAX_WAIT_MS), Step::Stop); + // The message is still pending — a later arrival re-arms a fresh waiter. + assert!(l.arm("p1", 2, MAX_WAIT_MS + 1)); + match l.poll("p1", false, MAX_WAIT_MS + 2) { + Step::Send(t) => assert!(t.contains("2 new messages"), "{t}"), + other => panic!("expected Send, got {other:?}"), + } + } + + #[test] + fn rate_limits_a_chatty_wave_to_one_nudge_per_interval() { + let mut l = NudgeLedger::new(); + l.arm("p1", 1, 0); + assert!(matches!(l.poll("p1", false, 0), Step::Send(_))); + l.arm("p1", 2, 10); + assert_eq!(l.poll("p1", false, 10), Step::Wait); // too soon + assert!(matches!( + l.poll("p1", false, MIN_INTERVAL_MS + 10), + Step::Send(_) + )); + } + + #[test] + fn drop_pane_forgets_a_closed_panes_counters() { + let mut l = NudgeLedger::new(); + l.arm("p1", 1, 0); + l.drop_pane("p1"); + assert_eq!(l.poll("p1", false, 0), Step::Stop); + } +} diff --git a/rs/crates/core/src/control/readmodel.rs b/rs/crates/core/src/control/readmodel.rs index 05e8705f..2461506c 100644 --- a/rs/crates/core/src/control/readmodel.rs +++ b/rs/crates/core/src/control/readmodel.rs @@ -184,6 +184,26 @@ impl ReadModel { }) } + /// Canonicalize a caller-supplied pane id, tolerating the two id spellings that coexist in + /// the wild: app-created panes are `pane-`, control-created ones are a bare ``. + /// An agent that was handed one form and reconstructs the other (goals-system spec agents do + /// exactly this) would otherwise get `404 no such pane` on every message it sends or reads. + /// Also accepts a session uid, which is what a pane's own tooling sometimes has to hand. + /// Returns the id as the read-model knows it, or `None` when nothing matches. + pub fn resolve_pane_id(&self, id: &str) -> Option { + if self.pane_loc.contains_key(id) { + return Some(id.to_string()); + } + let alias = match id.strip_prefix("pane-") { + Some(bare) => bare.to_string(), + None => format!("pane-{id}"), + }; + if self.pane_loc.contains_key(&alias) { + return Some(alias); + } + self.uid_to_pane(id) + } + pub fn tab_window(&self, tab_id: &str) -> Option { self.tab_to_window.get(tab_id).copied() } @@ -697,6 +717,22 @@ mod tests { assert_eq!(out.windows[0].tabs[0].panes[0].id, "p2"); } + #[test] + fn resolve_pane_id_accepts_both_spellings_and_the_session_uid() { + let mut m = seeded(); + m.insert_pane(1, pane("pane-abc", "u-abc")); + m.insert_pane(1, pane("bare", "u-bare")); + // Exact match wins. + assert_eq!(m.resolve_pane_id("pane-abc").as_deref(), Some("pane-abc")); + assert_eq!(m.resolve_pane_id("bare").as_deref(), Some("bare")); + // Alias in both directions: add the prefix / strip it. + assert_eq!(m.resolve_pane_id("abc").as_deref(), Some("pane-abc")); + assert_eq!(m.resolve_pane_id("pane-bare").as_deref(), Some("bare")); + // A session uid resolves to its pane. + assert_eq!(m.resolve_pane_id("u-abc").as_deref(), Some("pane-abc")); + assert_eq!(m.resolve_pane_id("ghost"), None); + } + #[test] fn insert_and_remove_pane_maintains_indexes() { let mut m = seeded(); diff --git a/rs/crates/core/src/control/routes.rs b/rs/crates/core/src/control/routes.rs index 1837ec3b..b07b3f3a 100644 --- a/rs/crates/core/src/control/routes.rs +++ b/rs/crates/core/src/control/routes.rs @@ -33,11 +33,12 @@ use crate::ansi_strip::strip_ansi; use crate::control::dispatch; use crate::control::events::ControlEvent; use crate::control::input::{keys_to_bytes, submit_newlines, KeysResult, SUBMIT_DELAY_MS}; +use crate::control::nudge; use crate::control::output::{ detect_awaiting_input, next_poll_delay, slice_since, wait_decision, WaitVerdict, DEFAULT_SETTLE_MS, DEFAULT_WAIT_TIMEOUT_MS, }; -use crate::control::readmodel::PaneStatus; +use crate::control::readmodel::{Activity, PaneStatus}; use crate::control::scope::{check_mintable, coerce_scope, pane_in_scope, queue_in_scope, Scope}; use crate::control::server::{events_url, notify_state, now_ms, Shared}; use crate::control::tokens::TokenInfo; @@ -122,9 +123,12 @@ fn authorize(shared: &Arc, headers: &HeaderMap) -> Result Result { let m = shared.model.lock().unwrap(); - match m.coords_of(pane_id) { + let canonical = match m.resolve_pane_id(pane_id) { + None => { + return Err(jstatus( + 404, + json!({ "error": "no such pane", "paneId": pane_id }), + )) + } + Some(c) => c, + }; + match m.coords_of(&canonical) { None => Err(jstatus( 404, json!({ "error": "no such pane", "paneId": pane_id }), @@ -148,10 +161,13 @@ fn find_pane_scoped( )); } let uid = m - .pane(pane_id) + .pane(&canonical) .map(|p| p.session_uid.clone()) .unwrap_or_default(); - Ok(FoundPane { uid }) + Ok(FoundPane { + uid, + pane_id: canonical, + }) } } } @@ -780,9 +796,12 @@ async fn messages_get( Ok(i) => i, Err(e) => return e, }; - if let Err(e) = find_pane_scoped(&shared, info.scope.as_ref(), &id) { - return e; - } + // Canonical id: an inbox is keyed by the read-model's pane id, so an alias-addressed read + // (`pane-` vs bare ``) must land on the same queue the writer posted to. + let id = match find_pane_scoped(&shared, info.scope.as_ref(), &id) { + Ok(f) => f.pane_id, + Err(e) => return e, + }; let after = non_neg_num(q.get("after")) .filter(|&a| a > 0) .map(|a| a as u64) @@ -807,9 +826,12 @@ async fn messages_post( Ok(i) => i, Err(e) => return e, }; - if let Err(e) = find_pane_scoped(&shared, info.scope.as_ref(), &id) { - return e; - } + // Canonical id: the inbox is keyed by the read-model's pane id, so an alias-addressed post + // (`pane-` vs bare ``) lands on the queue the target actually reads. + let id = match find_pane_scoped(&shared, info.scope.as_ref(), &id) { + Ok(f) => f.pane_id, + Err(e) => return e, + }; let b: Value = serde_json::from_slice(&body).unwrap_or(Value::Null); let from = b .get("from") @@ -825,6 +847,9 @@ async fn messages_post( .lock() .unwrap() .post(&id, &from, &msg_body, now_ms()); + // Wake an opted-in agent pane (goals org): the durable read stays the source of truth, but a + // TUI agent parked at its prompt would never perform that read on its own. + arm_inbox_nudge(&shared, &id, msg.seq); // Nudge live, in-scope clients (the durable read remains the source of truth). let coords = shared.model.lock().unwrap().coords_of(&id); shared.events.broadcast_for_pane( @@ -839,6 +864,75 @@ async fn messages_post( jstatus(200, json!({ "ok": true, "seq": msg.seq })) } +/// Whether inbox nudges are enabled at all — `HYPERPANES_MSG_NUDGE=0` (or `false`/`off`) in the +/// app's environment turns the whole mechanism off, leaving the bus pull-only as before. +fn nudges_enabled() -> bool { + !matches!( + std::env::var("HYPERPANES_MSG_NUDGE").as_deref(), + Ok("0") | Ok("false") | Ok("off") + ) +} + +/// Arm (and, if needed, spawn) the waiter that types a one-line "you have mail" into `pane_id` +/// once it goes quiet. See `control::nudge` for the policy: role opt-in, coalescing, +/// never-mid-turn, rate limit. +fn arm_inbox_nudge(shared: &Arc, pane_id: &str, seq: u64) { + if !nudges_enabled() { + return; + } + let (wants, uid, status) = { + let m = shared.model.lock().unwrap(); + match m.pane(pane_id) { + None => return, + Some(p) => ( + nudge::wants_nudge(p.meta.as_ref()), + p.session_uid.clone(), + p.status, + ), + } + }; + if !wants || status == PaneStatus::Exited { + return; + } + if !shared.nudges.lock().unwrap().arm(pane_id, seq, now_ms()) { + return; // a waiter is already in flight for this pane; it will pick the batch up + } + let shared = Arc::clone(shared); + let pane_id = pane_id.to_string(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(nudge::SETTLE_MS)).await; + loop { + // Busy = mid-turn (a running command, or output within the idle threshold): typing + // now would land in a live prompt, so wait for the pane to come to rest. + let busy = { + let m = shared.model.lock().unwrap(); + match m.pane(&pane_id) { + None => return, // pane closed while we waited + Some(p) => shared.compute_activity(p) == Activity::Busy, + } + }; + let step = shared.nudges.lock().unwrap().poll(&pane_id, busy, now_ms()); + match step { + nudge::Step::Stop => return, + nudge::Step::Wait => { + tokio::time::sleep(Duration::from_millis(nudge::POLL_MS)).await; + } + nudge::Step::Send(text) => { + // Same cadence the goal/resume delivery uses: text, gap, CR, insurance CR — + // a bracketed-paste TUI reads text+CR in one read as a paste otherwise. + let sessions = Arc::clone(&shared.sessions); + sessions.write(&uid, &text); + tokio::time::sleep(Duration::from_millis(250)).await; + sessions.write(&uid, "\r"); + tokio::time::sleep(Duration::from_millis(600)).await; + sessions.write(&uid, "\r"); + return; + } + } + } + }); +} + // ---- /panes/{id}/lock --------------------------------------------------------------------- async fn lock_post( @@ -2051,6 +2145,53 @@ mod golden { assert!(m["ts"].is_number()); } + /// The two pane-id spellings in the wild (`pane-` from the app, bare `` from the + /// control API) must address the SAME inbox: an agent handed one form and reconstructing the + /// other used to get `404 no such pane`, silently breaking the reply direction of the bus. + #[tokio::test] + async fn messages_are_addressable_by_either_pane_id_spelling() { + let s = boot(true).await; + s.shared + .model + .lock() + .unwrap() + .insert_pane(1, pane("pane-abc", "u1")); + // Post to the bare-uuid alias… + let post = client() + .post(format!("{}/panes/abc/messages", s.base)) + .header("authorization", format!("Bearer {}", s.token)) + .header("content-type", "application/json") + .body(r#"{"from":"impl","body":"done"}"#) + .send() + .await + .unwrap(); + assert_eq!(post.status().as_u16(), 200); + // …and read it back under the canonical id. + let get: serde_json::Value = client() + .get(format!("{}/panes/pane-abc/messages", s.base)) + .header("authorization", format!("Bearer {}", s.token)) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(get["paneId"], serde_json::json!("pane-abc")); + assert_eq!(get["messages"][0]["body"], serde_json::json!("done")); + // The reverse spelling reads the same queue and reports the canonical id. + let alias: serde_json::Value = client() + .get(format!("{}/panes/abc/messages", s.base)) + .header("authorization", format!("Bearer {}", s.token)) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(alias["paneId"], serde_json::json!("pane-abc")); + assert_eq!(alias["messages"][0]["body"], serde_json::json!("done")); + } + #[tokio::test] async fn lock_acquire_then_nonowner_input_is_423() { let s = boot(true).await; diff --git a/rs/crates/core/src/control/server.rs b/rs/crates/core/src/control/server.rs index c92ed2b4..795fea76 100644 --- a/rs/crates/core/src/control/server.rs +++ b/rs/crates/core/src/control/server.rs @@ -25,6 +25,7 @@ use tokio::sync::mpsc::UnboundedReceiver; use crate::control::events::{ControlEvent, EventHub}; use crate::control::inbox::MessageInbox; use crate::control::lock::PaneLocks; +use crate::control::nudge::NudgeLedger; use crate::control::readmodel::{Activity, PaneInfo, PaneRef, PaneStatus, ReadModel}; use crate::control::routes; use crate::control::supervisor::{Decision, Supervisor}; @@ -52,6 +53,9 @@ pub struct Shared { pub model: Mutex, pub tokens: Mutex, pub inbox: Mutex, + /// Inbox-nudge bookkeeping (`control::nudge`): which agent panes have unread mail and when + /// they were last woken. The inbox itself stays pull-only; this is the opt-in wake-up. + pub nudges: Mutex, pub locks: Mutex, /// Durable, claimable work queue backing the `/queues` + `/tasks` routes /// (worker-pool phase-2). `rusqlite::Connection` is `Send` but not `Sync`, so — @@ -108,6 +112,7 @@ impl Shared { model: Mutex::new(ReadModel::new()), tokens: Mutex::new(TokenStore::new()), inbox: Mutex::new(MessageInbox::new()), + nudges: Mutex::new(NudgeLedger::new()), locks: Mutex::new(PaneLocks::new()), work: Mutex::new(WorkQueue::open_in_memory().expect("open in-memory work queue")), events: EventHub::new(),