diff --git a/.changeset/smoothflow-engine-e2e.md b/.changeset/smoothflow-engine-e2e.md new file mode 100644 index 000000000..fc6e0db79 --- /dev/null +++ b/.changeset/smoothflow-engine-e2e.md @@ -0,0 +1,7 @@ +--- +'@smooai/smooth': patch +--- + +SmoothFlow engine e2e suite (th-8e3087): `crates/smooth-daemon/tests/flow_e2e` boots a real `smooth-daemon` per test — isolated HOME, ephemeral port, private tmux server, no lock / tailscale / relay / credentials — and drives it over the flow WS, the HTTP siblings, the `th` binary and `POST /api/flow/hooks`, with `fake-agent` installed through harness manifests in four state-source flavours (hooks, learned id, native, scrape). 24 tests (~70 s): shell lifecycle, every agent transition, the permission long-poll, resume-on-death with `--resume`, the three-resume give-up, the duplicate-resume guard, the hooks contract per event, `th flow` / `th harness` JSON, the harness matrix + sort/hide prefs, `th harness add`, and proof the suite never touches the real `~/.smooth`. Runs on every PR (tmux installed, `SMOOTH_E2E_STRICT=1`); docs in `docs/Engineering/SmoothFlow-Testing.md`. + +Engine bugs the suite found and this release fixes: tmux under a non-UTF-8 locale (any daemon not started from a shell) rewrote the tab in the pane-dead / pane-size queries, so a dead pane was never detected and supervision was inert; `{daemon_url}` was rendered from a port-0 request address; a resumed or prompt-less harness stayed `starting` forever (`SessionStart` now makes a starting row idle); a `held` row was flapped by the supervisor. Rule 4's `held` was also racy end to end: `kill`, the supervision tick and `POST /api/flow/hooks` all read a row and then wrote it, so a hold written between one reader's read and its write was silently overwritten (~1 in 3 under load) — a dying agent's last `Stop` un-held the row and the next tick read its dead tmux session as a crash. All three now serialise on a per-session lock and re-read under it, and a hook naming a held row is ignored. The same read-before-write ordering also stranded a resumed row: `hook` read the row before `relaunch` wrote `starting`, saw the pre-kill `idle`, and `SessionStart`'s promote-a-starting-row rule declined, so a `kill --resume` left the row `starting` for good. diff --git a/.github/workflows/pr-checks.yml b/.github/workflows/pr-checks.yml index ac12c6f8b..fc5887b43 100644 --- a/.github/workflows/pr-checks.yml +++ b/.github/workflows/pr-checks.yml @@ -94,6 +94,10 @@ jobs: shell: bash run: echo "run=${{ github.event_name != 'pull_request' || steps.filter.outputs.rust == 'true' }}" >>"$GITHUB_OUTPUT" + # tmux: the SmoothFlow engine e2e suite (crates/smooth-daemon/tests/ + # flow_e2e, th-8e3087) boots a real daemon per test and runs + # fake-agent under a private tmux server. Without it the suite + # skips — and SMOOTH_E2E_STRICT below turns that skip into a failure. - name: Install system dependencies if: steps.gate.outputs.run == 'true' && runner.os == 'Linux' # ubuntu-latest ships a google-chrome apt source that intermittently @@ -103,7 +107,7 @@ jobs: # a mirror we don't use (pearl th-79cef7). run: | sudo rm -f /etc/apt/sources.list.d/google-chrome* - sudo apt-get update && sudo apt-get install -y libdbus-1-dev libcap-ng-dev pkg-config protobuf-compiler + sudo apt-get update && sudo apt-get install -y libdbus-1-dev libcap-ng-dev pkg-config protobuf-compiler tmux # No apt on Windows — protoc is the only native build-time tool the # workspace needs there (tonic-build in smooth-scribe et al.). @@ -165,6 +169,11 @@ jobs: run: cargo nextest run --profile ci ${{ matrix.test_args }} env: NEXTEST_EXPERIMENTAL_LIBTEST_JSON: 1 + # The flow e2e suite must RUN on Linux, not skip: a missing + # tmux/bash/curl/`th` fails the job instead of passing it + # having proved nothing. Windows compiles the suite empty + # (`#![cfg(unix)]`) — no tmux there. + SMOOTH_E2E_STRICT: ${{ runner.os == 'Linux' && '1' || '0' }} # Windows test threads get ~1MB of stack; parsing/dropping the # full clap derive tree in a debug build overflows it as the # `th` Command graph grows (th-6c4ddf, then th-9483e8). libtest diff --git a/crates/smooth-daemon/examples/flow_e2e_server.rs b/crates/smooth-daemon/examples/flow_e2e_server.rs index 9ca1f7cae..84a0b425f 100644 --- a/crates/smooth-daemon/examples/flow_e2e_server.rs +++ b/crates/smooth-daemon/examples/flow_e2e_server.rs @@ -22,6 +22,12 @@ fn arg(args: &[String], name: &str) -> Option { #[tokio::main] async fn main() -> anyhow::Result<()> { + // `RUST_LOG` (default `warn`) — the supervisor's trace lines are how a + // silent state machine gets debugged. + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("warn"))) + .with_writer(std::io::stderr) + .init(); let args: Vec = std::env::args().collect(); let addr = arg(&args, "--addr").unwrap_or_else(|| "127.0.0.1:0".into()); let workspace = PathBuf::from(arg(&args, "--workspace").unwrap_or_else(|| ".".into())); diff --git a/crates/smooth-daemon/src/operator.rs b/crates/smooth-daemon/src/operator.rs index d1371168c..2d581faa9 100644 --- a/crates/smooth-daemon/src/operator.rs +++ b/crates/smooth-daemon/src/operator.rs @@ -1071,6 +1071,13 @@ pub async fn serve_local_flavor(addr: SocketAddr) -> Result<()> { // the app bundle) funnels through here, so this is the choke point. Held // to shutdown; the OS releases it if we die (pearl th-c71e6f). let _instance = crate::single_instance::acquire_default().await?; + // Port 0 (an ephemeral port — what the e2e suites bind) has to be + // resolved BEFORE the server is built: `{daemon_url}` (the address a + // `th code` / fake-agent pane posts its hooks back to) is rendered from + // this value when the flow router is installed below, and the bound port + // is only known after `spawn()`. th-8e3087 caught every hook of a port-0 + // daemon going to `http://127.0.0.1:0`. + let addr = resolve_ephemeral_port(addr)?; let token = provision_local_token()?; // The local flavor's tools: the workspace-confined fs/grep set + an // OS-sandboxed `bash` whose egress is routed through the goalie proxy (when @@ -1363,6 +1370,19 @@ pub async fn serve_local_flavor(addr: SocketAddr) -> Result<()> { Ok(()) } +/// `addr` with a port of 0 replaced by a port the OS just handed out (bound +/// and released — the same tiny race every "pick a free port" helper has). +/// Any other port is returned unchanged. +fn resolve_ephemeral_port(addr: SocketAddr) -> Result { + if addr.port() != 0 { + return Ok(addr); + } + let probe = std::net::TcpListener::bind(addr).with_context(|| format!("probing an ephemeral port on {addr}"))?; + let bound = probe.local_addr().context("reading the probed ephemeral port")?; + drop(probe); + Ok(bound) +} + /// The URL a process on this host reaches the daemon at (th-0f6126: the /// `{daemon_url}` a `th code` pane connects back to). An unspecified bind /// address is reachable on loopback. @@ -2395,6 +2415,20 @@ mod tests { assert_eq!(std::fs::read_to_string(&path).unwrap(), "127.0.0.1:9999"); } + /// th-8e3087: a port-0 bind is resolved to a real port BEFORE the flow + /// router renders `{daemon_url}` from it; fixed ports pass through. + #[test] + fn ephemeral_port_is_resolved_and_fixed_ports_pass_through() { + let fixed: SocketAddr = "127.0.0.1:8787".parse().unwrap(); + assert_eq!(resolve_ephemeral_port(fixed).unwrap(), fixed); + let zero: SocketAddr = "127.0.0.1:0".parse().unwrap(); + let resolved = resolve_ephemeral_port(zero).unwrap(); + assert_ne!(resolved.port(), 0); + assert_eq!(resolved.ip(), zero.ip()); + assert!(loopback_url(resolved).ends_with(&format!(":{}", resolved.port()))); + assert_eq!(loopback_url("0.0.0.0:8787".parse().unwrap()), "http://127.0.0.1:8787"); + } + #[test] fn token_path_is_under_the_home_dot_smooth() { let p = token_path(); diff --git a/crates/smooth-daemon/tests/fixtures/fake-agent b/crates/smooth-daemon/tests/fixtures/fake-agent new file mode 100755 index 000000000..4392577f7 --- /dev/null +++ b/crates/smooth-daemon/tests/fixtures/fake-agent @@ -0,0 +1,221 @@ +#!/usr/bin/env bash +# fake-agent — a scripted coding-agent CLI for SmoothFlow e2e tests +# (pearl th-8e3087). The engine launches it through a HARNESS MANIFEST +# (tests/fixtures/harnesses/*.toml → ~/.smooth/harnesses/ in the test HOME), +# exactly the way it launches claude / codex / opencode, and it behaves like a +# real harness everywhere the engine can tell: argv, hook posts, the +# permission long-poll, the usage-limit banner, exit codes. No LLM, no network +# beyond the flow daemon. A superset of `fake-claude` (the macOS lane's stub). +# +# Contract (docs/Engineering/SmoothFlow-Testing.md): +# argv `--session-id ` (preassigned), `--resume ` (resume=1), +# `--model `; the first positional is a PROMPT: commands +# separated by `;`, run after the start-up script. +# daemon `$SMOOTH_URL` (the manifest's `{daemon_url}`), else +# `./.flow-e2e-addr` in the cwd, else `$HOME/.smooth/daemon.addr`. +# mode `$FAKE_AGENT_MODE`: hooks (default) — Claude Code's event names; +# native — th code's `turn_start`/`turn_end`/`ask`/`bye` (the +# manifest's event_map); scrape — posts NOTHING, paints the pane +# so the engine's scraper drives state. +# id learned mode (no --session-id): `./.fake-agent-session-id` in +# the cwd, else `$FAKE_AGENT_SESSION_ID`, else a fresh uuid — the +# first hook binds it to the row by cwd, like opencode/codex. +# script `./.fake-agent-script` — one command per line, run on EVERY +# start (fresh and resumed) before the prompt. How a test makes a +# relaunch crash again, or a resumed agent do work. +# stdin one command per line (what `flow.send` pastes): +# /work a turn: working → idle +# /perm ask permission; prints `decision: ` +# /ask a question notification (needs_you · question) +# /limit [time] the usage-limit banner (default `11:59pm`) +# /exit [code] SessionEnd / bye, then exit (default 0) +# /crash [code] exit with no hook at all (default 3) +# /sleep +# anything else echoed back as `echo: ` +# log `$FAKE_AGENT_LOG` (default `./.fake-agent.log`): one line per +# launch (`argv …`), hook (`hook → `) and command. +# screen `fake-agent ready sid= resume=<0|1> mode=` on boot +# (proves a relaunch used `--resume`), then `> ` after each line. +set -u + +sid="" +resume=0 +prompt="" +while [ $# -gt 0 ]; do + case "$1" in + --session-id) sid="$2"; shift 2 ;; + --resume) sid="$2"; resume=1; shift 2 ;; + --model) shift 2 ;; + --*) shift ;; + *) if [ -z "$prompt" ]; then prompt="$1"; fi; shift ;; + esac +done + +mode="${FAKE_AGENT_MODE:-hooks}" +harness="${FAKE_AGENT_HARNESS:-fake-agent}" +log="${FAKE_AGENT_LOG:-./.fake-agent.log}" + +if [ -z "$sid" ]; then + if [ -r ./.fake-agent-session-id ]; then + sid="$(tr -d '[:space:]' <./.fake-agent-session-id)" + elif [ -n "${FAKE_AGENT_SESSION_ID:-}" ]; then + sid="$FAKE_AGENT_SESSION_ID" + else + sid="learned-$(date +%s)-$$" + fi +fi + +addr="${SMOOTH_URL:-}" +[ -n "$addr" ] || addr="$(cat ./.flow-e2e-addr 2>/dev/null || cat "$HOME/.smooth/daemon.addr" 2>/dev/null || true)" +addr="$(printf '%s' "$addr" | tr -d '[:space:]')" +case "$addr" in + http://* | https://*) ;; + *) addr="http://$addr" ;; +esac +addr="${addr%/}" + +logline() { printf '%s\n' "$*" >>"$log" 2>/dev/null || true; } +logline "argv sid=$sid resume=$resume mode=$mode model_prompt=[$prompt]" + +# hook [timeout-s] — prints the daemon's reply body. +# Scrape mode never posts: the engine must learn everything from the pane. +hook() { + if [ "$mode" = scrape ]; then + return 0 + fi + local reply + reply="$(curl -sS -m "${3:-5}" -X POST -H 'Content-Type: application/json' \ + --data-binary "{\"harness\":\"$harness\",\"event\":\"$1\",\"session_id\":\"$sid\",\"cwd\":\"$PWD\",\"payload\":$2}" \ + "$addr/api/flow/hooks" 2>/dev/null)" + logline "hook $1 → ${reply:-}" + printf '%s' "$reply" +} + +json_str() { + # Minimal JSON string escaping for the text we generate. + printf '%s' "$1" | sed -e 's/\\/\\\\/g' -e 's/"/\\"/g' +} + +# The scraper's "working" marker is painted on its own line and ERASED when +# the turn ends, so the engine's last-12-lines rule sees an idle pane again. +paint_working() { printf 'Thinking... (esc to interrupt)\n'; } +erase_working() { printf '\033[1A\r\033[2K'; } + +run_cmd() { + local line="$1" + logline "cmd $line" + case "$line" in + /work*) + local text="${line#/work}" + text="${text# }" + case "$mode" in + native) + hook turn_start '{}' >/dev/null + sleep "${FAKE_AGENT_WORK_SECS:-0}" + hook turn_end "{\"message\":\"done: $(json_str "$text")\"}" >/dev/null + ;; + scrape) + paint_working + sleep "${FAKE_AGENT_WORK_SECS:-3}" + erase_working + ;; + *) + hook UserPromptSubmit "{\"prompt\":\"$(json_str "$text")\"}" >/dev/null + hook PreToolUse '{"tool_name":"Bash","tool_input":{"command":"echo hi"}}' >/dev/null + sleep "${FAKE_AGENT_WORK_SECS:-0}" + hook PostToolUse '{"tool_name":"Bash","tool_response":{"stdout":"hi"}}' >/dev/null + hook Stop "{\"last_assistant_message\":\"done: $(json_str "$text")\"}" >/dev/null + ;; + esac + printf 'worked: %s\n' "$text" + ;; + /perm*) + case "$mode" in + native) + hook ask '{"reason":"permission","message":"run git push?"}' >/dev/null + printf 'asked\n' + ;; + scrape) + printf 'Bash(git push)\nDo you want to proceed?\n > 1. Yes\n 2. Yes, and don'"'"'t ask again\n 3. No (esc)\n' + local key + IFS= read -r -s -n1 key || key="" + case "$key" in + 1) printf 'decision: allow\n' ;; + 2) printf 'decision: allow_session\n' ;; + *) printf 'decision: deny\n' ;; + esac + ;; + *) + local reply + reply="$(hook PermissionRequest '{"tool_name":"Bash","tool_input":{"command":"git push"}}' 130)" + printf 'decision: %s\n' "$reply" + ;; + esac + ;; + /ask*) + local text="${line#/ask}" + text="${text# }" + if [ "$mode" = native ]; then + hook ask "{\"reason\":\"question\",\"message\":\"$(json_str "$text")\"}" >/dev/null + else + hook Notification "{\"notification_type\":\"idle_prompt\",\"message\":\"$(json_str "$text")\"}" >/dev/null + fi + printf 'asked: %s\n' "$text" + ;; + /limit*) + local at="${line#/limit}" + at="${at# }" + printf "You've hit your usage limit. Your limit will reset at %s\n" "${at:-11:59pm}" + ;; + /exit*) + local code="${line#/exit}" + code="${code// /}" + if [ "$mode" = native ]; then + hook bye '{}' >/dev/null + else + hook SessionEnd '{"reason":"exit"}' >/dev/null + fi + logline "exit ${code:-0}" + exit "${code:-0}" + ;; + /crash*) + local code="${line#/crash}" + code="${code// /}" + logline "crash ${code:-3}" + exit "${code:-3}" + ;; + /sleep*) + local secs="${line#/sleep}" + sleep "${secs// /}" + ;; + *) + printf 'echo: %s\n' "$line" + ;; + esac +} + +if [ "$mode" = native ]; then + : # th code has no SessionStart; its first report is turn_start. +else + hook SessionStart "{\"source\":\"$([ "$resume" = 1 ] && echo resume || echo startup)\"}" >/dev/null +fi +printf 'fake-agent ready sid=%s resume=%s mode=%s\n' "$sid" "$resume" "$mode" + +if [ -r ./.fake-agent-script ]; then + while IFS= read -r line || [ -n "$line" ]; do + run_cmd "$line" + done <./.fake-agent-script +fi +if [ -n "$prompt" ]; then + IFS=';' read -r -a parts <<<"$prompt" + for part in "${parts[@]}"; do + run_cmd "${part# }" + done +fi + +printf '> ' +while IFS= read -r line; do + run_cmd "$line" + printf '> ' +done +logline "stdin closed" diff --git a/crates/smooth-daemon/tests/fixtures/harnesses/fake-agent-learned.toml b/crates/smooth-daemon/tests/fixtures/harnesses/fake-agent-learned.toml new file mode 100644 index 000000000..23f2b5ee3 --- /dev/null +++ b/crates/smooth-daemon/tests/fixtures/harnesses/fake-agent-learned.toml @@ -0,0 +1,43 @@ +# fake-agent-learned — hooks state, session id LEARNED from the first hook +# (the opencode / codex shape): no `--session-id`; the agent reports the id it +# picked (./.fake-agent-session-id in the cwd, else a fresh one) and the engine +# binds it to the newest id-less row in that worktree. +name = "fake-agent-learned" +display_name = "Fake Agent (learned id)" + +[binary] +names = ["fake-agent"] +prefer_paths = [".local/bin/fake-agent"] + +[launch] +argv = ["--model", "{model}", "{prompt}"] +prompt_as = "argv" +session_id = "learned" + +[launch.env] +SMOOTH_URL = "{daemon_url}" +FAKE_AGENT_MODE = "hooks" + +[resume] +argv = ["--resume", "{session_id}"] +mode = "resume_session" + +[state] +source = "hooks" + +[state.hooks] +install = "none — fake-agent posts to /api/flow/hooks itself" + +[state.scrape] +working = ["esc to interrupt"] +idle = ["(?m)^> ?$"] +needs_you = ["do you want to proceed"] +usage_limit = ["usage limit", "limit will reset"] + +[steer] +method = "bracketed_paste" +submit_key = "Enter" + +[kill] +signal = "TERM" +grace_ms = 1000 diff --git a/crates/smooth-daemon/tests/fixtures/harnesses/fake-agent-native.toml b/crates/smooth-daemon/tests/fixtures/harnesses/fake-agent-native.toml new file mode 100644 index 000000000..a04d2a940 --- /dev/null +++ b/crates/smooth-daemon/tests/fixtures/harnesses/fake-agent-native.toml @@ -0,0 +1,46 @@ +# fake-agent-native — NATIVE state (the th code shape): the agent reports its +# own turns with its own event names, mapped by `event_map`; no SessionStart, +# no permission long-poll (an `ask` is a plain needs_you the scraper's +# keystroke path answers). +name = "fake-agent-native" +display_name = "Fake Agent (native)" + +[binary] +names = ["fake-agent"] +prefer_paths = [".local/bin/fake-agent"] + +[launch] +argv = ["--session-id", "{session_id}", "--model", "{model}", "{prompt}"] +prompt_as = "argv" +session_id = "preassigned" + +[launch.env] +SMOOTH_URL = "{daemon_url}" +FAKE_AGENT_MODE = "native" + +[resume] +argv = ["--resume", "{session_id}"] +mode = "resume_session" + +[state] +source = "native" + +[state.hooks] +install = "none — fake-agent reports turn_start / turn_end itself" + +[state.hooks.event_map] +turn_start = "working" +turn_end = "idle" +ask = "needs_you" +bye = "ended" + +[state.scrape] +usage_limit = ["usage limit", "limit will reset"] + +[steer] +method = "bracketed_paste" +submit_key = "Enter" + +[kill] +signal = "TERM" +grace_ms = 1000 diff --git a/crates/smooth-daemon/tests/fixtures/harnesses/fake-agent-scrape.toml b/crates/smooth-daemon/tests/fixtures/harnesses/fake-agent-scrape.toml new file mode 100644 index 000000000..5874fb5a7 --- /dev/null +++ b/crates/smooth-daemon/tests/fixtures/harnesses/fake-agent-scrape.toml @@ -0,0 +1,39 @@ +# fake-agent-scrape — SCRAPE state: the agent posts nothing; working / idle / +# approval / usage-limit all come from what it paints on the pane. Resume is +# `relaunch_command` (the original argv again), the other resume mode. +name = "fake-agent-scrape" +display_name = "Fake Agent (scrape)" + +[binary] +names = ["fake-agent"] +prefer_paths = [".local/bin/fake-agent"] + +[launch] +argv = ["--session-id", "{session_id}", "--model", "{model}", "{prompt}"] +prompt_as = "argv" +session_id = "preassigned" + +[launch.env] +FAKE_AGENT_MODE = "scrape" +FAKE_AGENT_WORK_SECS = "5" + +[resume] +argv = [] +mode = "relaunch_command" + +[state] +source = "scrape" + +[state.scrape] +working = ["esc to interrupt"] +idle = ["(?m)^> ?$"] +needs_you = ["do you want to proceed"] +usage_limit = ["usage limit", "limit will reset"] + +[steer] +method = "bracketed_paste" +submit_key = "Enter" + +[kill] +signal = "TERM" +grace_ms = 1000 diff --git a/crates/smooth-daemon/tests/fixtures/harnesses/fake-agent.toml b/crates/smooth-daemon/tests/fixtures/harnesses/fake-agent.toml new file mode 100644 index 000000000..66976b1c0 --- /dev/null +++ b/crates/smooth-daemon/tests/fixtures/harnesses/fake-agent.toml @@ -0,0 +1,43 @@ +# fake-agent — hooks state, pre-assigned session id (the Claude Code shape). +# Installed into ~/.smooth/harnesses/ of a test HOME by the e2e harness +# (crates/smooth-daemon/tests/flow_e2e). The binary is tests/fixtures/fake-agent, +# copied to ~/.local/bin/fake-agent so `prefer_paths` resolves it under any PATH. +name = "fake-agent" +display_name = "Fake Agent (hooks)" + +[binary] +names = ["fake-agent"] +prefer_paths = [".local/bin/fake-agent"] + +[launch] +argv = ["--session-id", "{session_id}", "--model", "{model}", "{prompt}"] +prompt_as = "argv" +session_id = "preassigned" + +[launch.env] +SMOOTH_URL = "{daemon_url}" +FAKE_AGENT_MODE = "hooks" + +[resume] +argv = ["--resume", "{session_id}"] +mode = "resume_session" + +[state] +source = "hooks" + +[state.hooks] +install = "none — fake-agent posts to /api/flow/hooks itself" + +[state.scrape] +working = ["esc to interrupt"] +idle = ["(?m)^> ?$"] +needs_you = ["do you want to proceed"] +usage_limit = ["usage limit", "limit will reset"] + +[steer] +method = "bracketed_paste" +submit_key = "Enter" + +[kill] +signal = "TERM" +grace_ms = 1000 diff --git a/crates/smooth-daemon/tests/flow_e2e/agent.rs b/crates/smooth-daemon/tests/flow_e2e/agent.rs new file mode 100644 index 000000000..74fe3bd4d --- /dev/null +++ b/crates/smooth-daemon/tests/flow_e2e/agent.rs @@ -0,0 +1,452 @@ +//! Agent sessions end to end: the state machine (working / idle / +//! needs_you / limited / done / dead), steering, the permission long-poll, +//! resume-on-death with `--resume`, the give-up after three resumes, the +//! learned-id binding, the duplicate-resume guard, and the native + scrape +//! state sources. + +use std::time::Duration; + +use serde_json::json; + +use crate::support::{local_clock_in, prereqs, state, Daemon, TICK, WAIT}; + +/// Rule 2's first backoff (`RESUME_BACKOFF_BASE · 2^0`) plus a tick. +const FIRST_RESUME: Duration = Duration::from_secs(5 + 2 + 2); + +#[tokio::test] +async fn agent_transitions_working_idle_needs_you_done() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + let mut ws = d.ws().await; + + let s = d.new_session("fake-agent", Some("/work first")).await; + let id = s["id"].as_str().unwrap().to_string(); + assert_eq!(s["kind"], "fake-agent"); + assert_eq!(s["state"], "starting"); + assert_eq!(s["state_source"], "inferred", "nothing has reported yet"); + let agent_id = s["agent_session_id"].as_str().unwrap().to_string(); + assert_eq!(agent_id.len(), 36, "a pre-assigned uuid: {s}"); + let argv = s["argv"].as_array().unwrap(); + assert!( + argv[0].as_str().unwrap().ends_with("/.local/bin/fake-agent"), + "resolved through prefer_paths: {argv:?}" + ); + assert_eq!( + argv[1..], + [json!("--session-id"), json!(agent_id), json!("/work first")], + "no --model when none was given" + ); + assert_eq!(s["title"], "/work first"); + + // The prompt runs a turn: hooks drive working → idle, unread, `hooks`. + // (SessionStart already made it idle-and-read; the turn's Stop is the + // unread idle.) + let idle = d + .wait_until(&id, "unread idle via hooks", WAIT, |s| { + state(s) == "idle" && s["state_source"] == "hooks" && s["unread"] == true + }) + .await; + assert!(idle["attention"].is_null()); + let screen = d.wait_screen(&id, "worked: first", WAIT).await; + assert!(screen.contains(&format!("fake-agent ready sid={agent_id} resume=0 mode=hooks")), "{screen}"); + let log = d.agent_log(); + assert!( + log.contains("hook UserPromptSubmit → {}") && log.contains("hook Stop → {}"), + "every hook got a 200 body:\n{log}" + ); + + // The WS saw the whole story as flow.event lines. + let mut kinds = Vec::new(); + let deadline = std::time::Instant::now() + WAIT; + while std::time::Instant::now() < deadline && !kinds.iter().any(|(k, t): &(String, String)| k == "agent" && t == "done: first") { + let Some(f) = ws.next(Duration::from_secs(5)).await else { break }; + if f["type"] == "flow.event" && f["id"] == id { + kinds.push((f["kind"].as_str().unwrap().to_string(), f["text"].as_str().unwrap().to_string())); + } + } + assert!(kinds.contains(&("user".into(), "first".into())), "{kinds:?}"); + assert!(kinds.iter().any(|(k, t)| k == "tool" && t.contains("Bash(echo hi)")), "{kinds:?}"); + assert!(kinds.contains(&("system".into(), "working".into())), "{kinds:?}"); + assert!(kinds.contains(&("agent".into(), "done: first".into())), "{kinds:?}"); + + // mark_read clears the flag. + ws.send(json!({"type":"flow.mark_read","id":id})).await; + d.wait_until(&id, "read", WAIT, |s| s["unread"] == false).await; + + // Steer: /perm → a hook-reported permission with a request_id; the hook + // POST is held open until flow.approve answers it. + d.send(&id, "/perm").await; + let ask = d.wait_state(&id, "needs_you", WAIT).await; + assert_eq!(ask["attention"]["reason"], "permission"); + assert_eq!(ask["attention"]["detail"], "Bash: git push"); + let request_id = ask["attention"]["request_id"].as_str().unwrap().to_string(); + assert!(!request_id.starts_with("scrape-"), "reported by the hook, not scraped: {ask}"); + let att = ws + .wait_for("flow.attention", WAIT, |v| { + v["type"] == "flow.attention" && v["id"] == id && v["attention"]["reason"] == "permission" + }) + .await; + assert_eq!(att["attention"]["request_id"], request_id); + // The daemon's inbox view (needs_you) is what `th flow inbox` filters on. + tokio::time::sleep(Duration::from_secs(1)).await; + assert!( + !d.agent_log().contains("hook PermissionRequest →"), + "the long-poll is still open:\n{}", + d.agent_log() + ); + + ws.send(json!({"type":"flow.approve","id":id,"request_id":request_id,"decision":"allow_session"})) + .await; + d.wait_state(&id, "working", WAIT).await; + let screen = d.wait_screen(&id, "decision:", WAIT).await; + assert!(screen.contains(r#""behavior":"allow""#), "the agent printed the long-polled reply: {screen}"); + assert!(screen.contains(r#""destination":"session""#), "allow_session adds a session rule: {screen}"); + // Approving is a user event; the state line follows. + ws.wait_for("approve event", WAIT, |v| { + v["type"] == "flow.event" && v["id"] == id && v["text"] == "approve: allow_session" + }) + .await; + + // Another turn, then a clean exit → done with the code. + d.send(&id, "/work second").await; + d.wait_until(&id, "idle again", WAIT, |s| state(s) == "idle" && s["unread"] == true).await; + d.wait_screen(&id, "worked: second", WAIT).await; + d.send(&id, "/exit 0").await; + let done = d.wait_state(&id, "done", WAIT + TICK).await; + assert_eq!(done["exit_code"], 0, "{done}"); + assert!(done["attention"].is_null()); + let log = d.agent_log(); + assert!(log.contains("hook SessionEnd → {}"), "{log}"); + // A done agent is not resumed by the supervisor. + tokio::time::sleep(TICK * 2).await; + assert_eq!(state(&d.session(&id).await), "done"); +} + +#[tokio::test] +async fn agent_question_notification_needs_you_and_steer_answers_it() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + let s = d.new_session("fake-agent", Some("/ask which branch?")).await; + let id = s["id"].as_str().unwrap().to_string(); + let q = d.wait_state(&id, "needs_you", WAIT).await; + assert_eq!(q["attention"]["reason"], "question"); + assert_eq!(q["attention"]["detail"], "which branch?"); + assert!(q["attention"]["request_id"].is_null(), "a question has nothing to long-poll: {q}"); + // Steering text is the answer; the next turn is working → idle. + d.send(&id, "/work main").await; + d.wait_until(&id, "idle", WAIT, |s| state(s) == "idle" && s["state_source"] == "hooks").await; + d.wait_screen(&id, "worked: main", WAIT).await; + d.kill(&id, false).await; + d.wait_state(&id, "done", WAIT).await; +} + +#[tokio::test] +async fn agent_usage_limit_is_scheduled_from_the_banner() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + let s = d.new_session("fake-agent", Some("/work warm; /limit 11:59pm")).await; + let id = s["id"].as_str().unwrap().to_string(); + let limited = d.wait_state(&id, "limited", WAIT).await; + assert_eq!(limited["attention"]["reason"], "usage_limit"); + assert_eq!( + limited["state_source"], "hooks", + "limits are scraped even when hooks own working/idle: {limited}" + ); + let at = chrono::DateTime::parse_from_rfc3339(limited["attention"]["resume_at"].as_str().unwrap()).unwrap(); + let wait = at.signed_duration_since(chrono::Utc::now()); + assert!( + wait > chrono::Duration::seconds(30) && wait <= chrono::Duration::hours(24), + "resumes at the next 11:59pm: {limited}" + ); + assert!(limited["attention"]["detail"].as_str().unwrap().starts_with("resumes at "), "{limited}"); + // The window is in the future — nothing fires, the state holds. + tokio::time::sleep(TICK * 2).await; + assert_eq!(state(&d.session(&id).await), "limited"); + let killed = d.kill(&id, false).await; + assert_eq!(state(&killed), "done"); +} + +/// Slow (~90 s): the banner names a time ~1 min out; the supervisor presses +/// Enter when it passes and the session is working again. +#[tokio::test] +async fn agent_usage_limit_resume_fires_when_the_window_passes() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + // Minute resolution + "at least one minute out": ~70–130 s from now. + let at = local_clock_in(75); + let s = d.new_session("fake-agent", Some(&format!("/work warm; /limit {at}"))).await; + let id = s["id"].as_str().unwrap().to_string(); + let limited = d.wait_state(&id, "limited", WAIT).await; + let resume_at = chrono::DateTime::parse_from_rfc3339(limited["attention"]["resume_at"].as_str().unwrap()).unwrap(); + let wait = resume_at.signed_duration_since(chrono::Utc::now()); + assert!( + wait > chrono::Duration::seconds(30) && wait < chrono::Duration::seconds(150), + "parsed `{at}` → {limited}" + ); + let budget = Duration::from_secs(wait.num_seconds().max(0) as u64) + TICK * 3; + let working = d.wait_state(&id, "working", budget).await; + assert!(working["attention"].is_null(), "{working}"); + // The Enter reached the agent: an empty line is echoed (`echo:` — tmux + // strips the trailing space from a captured pane). + d.wait_screen(&id, "echo:", WAIT).await; + d.kill(&id, false).await; +} + +#[tokio::test] +async fn agent_that_dies_is_resumed_with_its_session_id() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + let mut ws = d.ws().await; + let s = d.new_session("fake-agent", Some("/work once; /crash 2")).await; + let id = s["id"].as_str().unwrap().to_string(); + let agent_id = s["agent_session_id"].as_str().unwrap().to_string(); + + // Death → starting + `crashed` attention with the schedule (rule 2). + let crashed = d.wait_until(&id, "crashed attention", WAIT, |s| s["attention"]["reason"] == "crashed").await; + assert_eq!(state(&crashed), "starting", "{crashed}"); + assert_eq!(crashed["exit_code"], 2); + let detail = crashed["attention"]["detail"].as_str().unwrap(); + assert!(detail.contains("exit 2") && detail.contains("resuming in 5s (attempt 1/3)"), "{detail}"); + assert!(crashed["attention"]["resume_at"].is_string()); + + // Relaunched with `--resume `; the pane proves it; the row's argv + // is the resume argv; state comes from the scraper again until hooks + // speak (a relaunch resets state_source to inferred). + let resumed = d + .wait_until(&id, "relaunched", FIRST_RESUME + WAIT, |s| { + s["attention"].is_null() && s["argv"][1] == "--resume" + }) + .await; + assert_eq!(resumed["argv"][2], agent_id, "{resumed}"); + assert_eq!(resumed["agent_session_id"], agent_id, "the harness session survives the relaunch"); + let screen = d.wait_screen(&id, "resume=1", WAIT).await; + assert!(screen.contains(&format!("sid={agent_id} resume=1")), "{screen}"); + let idle = d.wait_state(&id, "idle", WAIT).await; + assert!(idle["pid"].as_u64().unwrap() != s["pid"].as_u64().unwrap(), "a new process: {idle}"); + // The story is in the event stream: crashed, then starting, then idle. + let mut texts = Vec::new(); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while std::time::Instant::now() < deadline { + let Some(f) = ws.next(Duration::from_secs(1)).await else { break }; + if f["type"] == "flow.event" && f["id"] == id && f["kind"] == "system" { + texts.push(f["text"].as_str().unwrap().to_string()); + } + } + assert!(texts.iter().any(|t| t.starts_with("starting · crashed: exit 2")), "{texts:?}"); + // A relaunch is starting → starting (no line); the resumed harness's + // SessionStart is what makes it idle again. + assert!( + texts.iter().filter(|t| *t == "idle").count() >= 2, + "idle before the crash and after the resume: {texts:?}" + ); + // It works again after the resume (hooks re-attach to the same id). + d.send(&id, "/work after").await; + d.wait_until(&id, "idle via hooks after resume", WAIT, |s| state(s) == "idle" && s["state_source"] == "hooks") + .await; + let log = d.agent_log(); + assert!(log.contains("argv sid=") && log.matches("argv sid=").count() == 2, "two launches:\n{log}"); + d.kill(&id, false).await; +} + +/// Slow (~45 s): three failed resumes (5 s + 10 s + 20 s backoff) → dead. +#[tokio::test] +async fn agent_that_keeps_crashing_is_dead_after_three_resumes() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + // The start-up script runs on every launch — fresh and resumed — so each + // relaunch dies the same way. + d.write_script(&d.ws.clone(), &["/crash 3"]); + let s = d.new_session("fake-agent", None).await; + let id = s["id"].as_str().unwrap().to_string(); + let budget = Duration::from_secs(5 + 10 + 20) + TICK * 6 + WAIT; + let dead = d.wait_state(&id, "dead", budget).await; + assert_eq!(dead["attention"]["reason"], "crashed"); + let detail = dead["attention"]["detail"].as_str().unwrap(); + assert!(detail.contains("exit 3") && detail.contains("gave up after 3 resumes"), "{detail}"); + assert_eq!(dead["exit_code"], 3); + let log = d.agent_log(); + assert_eq!(log.matches("argv sid=").count(), 4, "one launch + three resumes:\n{log}"); + assert_eq!(log.matches("resume=1").count(), 3, "{log}"); + // Dead stays dead: no fourth relaunch. + tokio::time::sleep(Duration::from_secs(8)).await; + assert_eq!(d.agent_log().matches("argv sid=").count(), 4); + assert_eq!(state(&d.session(&id).await), "dead"); +} + +#[tokio::test] +async fn learned_session_id_binds_from_the_first_hook_and_kill_resume_relaunches() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + std::fs::write(d.ws.join(".fake-agent-session-id"), "learned-abc-123\n").unwrap(); + let s = d.new_session("fake-agent-learned", Some("/work bind")).await; + let id = s["id"].as_str().unwrap().to_string(); + assert!(s["agent_session_id"].is_null(), "learned: no id until the harness reports one: {s}"); + assert_eq!(s["argv"][1], "/work bind", "no --session-id in a learned launch: {s}"); + + // The first hook from the worktree binds the id to this row. + let bound = d + .wait_until(&id, "bound", WAIT, |s| s["agent_session_id"] == "learned-abc-123" && state(s) == "idle") + .await; + assert_eq!(bound["state_source"], "hooks"); + + // kill --resume relaunches with `--resume `. + let relaunched = d.kill(&id, true).await; + assert_eq!(state(&relaunched), "starting", "{relaunched}"); + assert_eq!(relaunched["argv"][1], "--resume"); + assert_eq!(relaunched["argv"][2], "learned-abc-123"); + d.wait_screen(&id, "sid=learned-abc-123 resume=1", WAIT).await; + d.wait_state(&id, "idle", WAIT).await; + d.kill(&id, false).await; +} + +/// Rule 4: a resume is refused while another live process owns the same +/// harness session. Two rows can only share an id through store state (the +/// engine never binds an id twice), so the second row is real — a live +/// `fake-agent-learned` process — and the shared id is written to its row +/// the way a stale or foreign db row would carry it. +#[tokio::test] +async fn duplicate_resume_guard_holds_when_a_live_pid_owns_the_session() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + std::fs::write(d.ws.join(".fake-agent-session-id"), "shared-s1\n").unwrap(); + let a = d.new_session("fake-agent-learned", Some("/work a")).await; + let a_id = a["id"].as_str().unwrap().to_string(); + d.wait_until(&a_id, "A bound", WAIT, |s| s["agent_session_id"] == "shared-s1" && state(s) == "idle") + .await; + + let other = d.extra_repo("ws2"); + let b = d.new_session_in("fake-agent-learned", Some("/work b"), &other).await; + let b_id = b["id"].as_str().unwrap().to_string(); + let b = d.wait_state(&b_id, "idle", WAIT).await; + let b_pid = b["pid"].as_u64().unwrap() as u32; + assert!(Daemon::pid_alive(b_pid)); + d.store().set_agent_session(&b_id, "shared-s1").unwrap(); + + // A's resume is held, naming B's pid; A is not relaunched. + let held = d.kill(&a_id, true).await; + assert_eq!(state(&held), "needs_you", "{held}"); + assert_eq!(held["attention"]["reason"], "held"); + let detail = held["attention"]["detail"].as_str().unwrap(); + assert!(detail.contains("shared-s1") && detail.contains(&format!("live pid {b_pid}")), "{detail}"); + tokio::time::sleep(TICK * 2).await; + let a_now = d.session(&a_id).await; + assert_eq!(state(&a_now), "needs_you", "the supervisor leaves a held row alone: {a_now}"); + assert_eq!(d.agent_log().matches("argv sid=").count(), 1, "A was not relaunched:\n{}", d.agent_log()); + + // Once B is gone the claim lapses and the resume goes through. + d.kill(&b_id, false).await; + d.wait_state(&b_id, "done", WAIT).await; + let relaunched = d.kill(&a_id, true).await; + assert_eq!(state(&relaunched), "starting", "{relaunched}"); + d.wait_screen(&a_id, "sid=shared-s1 resume=1", WAIT).await; + d.kill(&a_id, false).await; +} + +#[tokio::test] +async fn native_harness_reports_its_own_turns() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + let s = d.new_session("fake-agent-native", Some("/work n1")).await; + let id = s["id"].as_str().unwrap().to_string(); + let idle = d + .wait_until(&id, "idle via native", WAIT, |s| state(s) == "idle" && s["state_source"] == "native") + .await; + assert_eq!(idle["unread"], true); + d.wait_screen(&id, "worked: n1", WAIT).await; + let log = d.agent_log(); + assert!(log.contains("hook turn_start → {}") && log.contains("hook turn_end → {}"), "{log}"); + assert!(!log.contains("SessionStart"), "a native harness has no SessionStart: {log}"); + + // `ask` maps to needs_you through event_map, with the payload's reason; + // there is no long-poll, so approve presses the key on the pane. + d.send(&id, "/perm").await; + let ask = d.wait_state(&id, "needs_you", WAIT).await; + assert_eq!(ask["attention"]["reason"], "permission"); + assert_eq!(ask["attention"]["detail"], "run git push?"); + assert!(ask["attention"]["request_id"].is_null(), "{ask}"); + d.approve(&id, "none", "allow").await; + d.wait_state(&id, "working", WAIT).await; + // The keystroke path typed `1`; the next steer line shows it (tmux can't + // paste an empty buffer, so the line is `1x`). + d.send(&id, "x").await; + d.wait_screen(&id, "echo: 1x", WAIT).await; + d.send(&id, "/ask ready?").await; + let q = d.wait_until(&id, "question", WAIT, |s| s["attention"]["reason"] == "question").await; + assert_eq!(q["attention"]["detail"], "ready?"); + d.send(&id, "/exit 0").await; + let done = d.wait_state(&id, "done", WAIT + TICK).await; + assert_eq!(done["exit_code"], 0); + assert!(d.agent_log().contains("hook bye → {}")); +} + +#[tokio::test] +async fn scrape_harness_state_is_inferred_from_the_pane() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + let s = d.new_session("fake-agent-scrape", Some("/work s1")).await; + let id = s["id"].as_str().unwrap().to_string(); + // The "working" marker is on screen for FAKE_AGENT_WORK_SECS (5 s). + let working = d.wait_state(&id, "working", WAIT).await; + assert_eq!(working["state_source"], "inferred"); + let idle = d.wait_until(&id, "idle scraped", WAIT, |s| state(s) == "idle").await; + assert_eq!(idle["state_source"], "inferred", "scraping never claims `hooks`: {idle}"); + assert_eq!(idle["unread"], true, "working → idle by scrape is unread too: {idle}"); + assert!( + d.agent_log().contains("cmd /work s1") && !d.agent_log().contains("hook "), + "scrape mode posts nothing:\n{}", + d.agent_log() + ); + + // An approval menu on the pane → needs_you with a scrape- request id; + // approve presses the menu key (2 = allow_session). + d.send(&id, "/perm").await; + let ask = d.wait_state(&id, "needs_you", WAIT).await; + assert!(ask["attention"]["request_id"].as_str().unwrap().starts_with("scrape-"), "{ask}"); + assert_eq!(ask["attention"]["detail"], "approval prompt on screen"); + let request_id = ask["attention"]["request_id"].as_str().unwrap().to_string(); + d.approve(&id, &request_id, "allow_session").await; + d.wait_screen(&id, "decision: allow_session", WAIT).await; + d.wait_state(&id, "working", WAIT).await; + + // The banner → limited, scraped like any harness. + d.send(&id, "/limit 11:59pm").await; + let limited = d.wait_state(&id, "limited", WAIT).await; + assert_eq!(limited["attention"]["reason"], "usage_limit"); + + // relaunch_command: a crash relaunches the ORIGINAL argv (resume=0, the + // prompt runs again). + let (status, _) = d.post(&format!("/api/flow/sessions/{id}/send"), json!({"text":"/crash 4"})).await; + assert_eq!(status, 200); + d.wait_until(&id, "crashed", WAIT, |s| s["attention"]["reason"] == "crashed").await; + let relaunched = d + .wait_until(&id, "relaunched", FIRST_RESUME + WAIT, |s| s["attention"].is_null() && state(s) != "limited") + .await; + assert_eq!(relaunched["argv"], s["argv"], "relaunch_command reuses the launch argv: {relaunched}"); + let screen = d.wait_screen(&id, "resume=0 mode=scrape", WAIT).await; + assert!( + d.agent_log().matches("cmd /work s1").count() >= 2, + "the prompt ran again: {screen}\n{}", + d.agent_log() + ); + d.kill(&id, false).await; +} diff --git a/crates/smooth-daemon/tests/flow_e2e/cli.rs b/crates/smooth-daemon/tests/flow_e2e/cli.rs new file mode 100644 index 000000000..90d427b1d --- /dev/null +++ b/crates/smooth-daemon/tests/flow_e2e/cli.rs @@ -0,0 +1,140 @@ +//! `th flow` and `th harness` against the live daemon — the `--json` +//! contracts scripts depend on, and the human lines a person reads. + +use serde_json::json; + +use crate::support::{prereqs_with_th, state, Daemon, WAIT}; + +#[tokio::test] +async fn th_flow_json_against_the_live_daemon() { + if !prereqs_with_th() { + return; + } + let d = Daemon::boot().await; + let ws = d.ws.to_string_lossy().into_owned(); + + // ls on an empty engine: a confirmed empty read, in both renderings. + assert_eq!(d.th_json(&["flow", "ls", "--json"]), json!({"sessions": []})); + let (code, out, _) = d.th(&["flow", "ls"]); + assert_eq!(code, 0); + assert!(out.contains("No flow sessions"), "{out}"); + assert_eq!(d.th_json(&["flow", "inbox", "--json"]), json!({"sessions": []})); + + // new --json → the row; --kind is any manifest name. + let v = d.th_json(&[ + "flow", + "new", + "--kind", + "fake-agent", + "--worktree", + &ws, + "--prompt", + "/work c1", + "--title", + "cli one", + "--json", + ]); + let id = v["session"]["id"].as_str().unwrap().to_string(); + assert_eq!(v["session"]["kind"], "fake-agent"); + assert_eq!(v["session"]["title"], "cli one"); + d.wait_until(&id, "unread idle via hooks", WAIT, |s| { + state(s) == "idle" && s["state_source"] == "hooks" && s["unread"] == true + }) + .await; + + // ls --json carries the engine's row verbatim; the table shows the id, + // the state glyph text and the VIA column. + let ls = d.th_json(&["flow", "ls", "--json"]); + assert_eq!(ls["sessions"][0]["id"], id); + assert_eq!(ls["sessions"][0]["state"], "idle"); + assert_eq!(ls["sessions"][0]["state_source"], "hooks"); + let (_, table, _) = d.th(&["flow", "ls"]); + assert!( + table.contains(&id) && table.contains("idle") && table.contains("hooks") && table.contains("cli one"), + "{table}" + ); + + // snapshot --json is the flow.screen frame; the human form is the text. + let snap = d.th_json(&["flow", "snapshot", &id, "--json"]); + assert_eq!(snap["type"], "flow.screen"); + assert!(snap["text"].as_str().unwrap().contains("worked: c1"), "{snap}"); + let (_, text, _) = d.th(&["flow", "snapshot", &id]); + assert!(text.contains("worked: c1"), "{text}"); + + // send → inbox shows the permission; approve --json answers it. + let (code, out, err) = d.th(&["flow", "send", &id, "/perm"]); + assert_eq!(code, 0, "{err}"); + assert!(out.contains(&format!("sent to {id}")), "{out}"); + let ask = d.wait_state(&id, "needs_you", WAIT).await; + let request_id = ask["attention"]["request_id"].as_str().unwrap().to_string(); + let inbox = d.th_json(&["flow", "inbox", "--json"]); + assert_eq!(inbox["sessions"][0]["id"], id); + assert_eq!(inbox["sessions"][0]["attention"]["request_id"], request_id); + let (_, inbox_text, _) = d.th(&["flow", "inbox"]); + assert!(inbox_text.contains("needs you") && inbox_text.contains("[permission]"), "{inbox_text}"); + // approve without --request picks the session's current request. + let approved = d.th_json(&["flow", "approve", &id, "--decision", "allow", "--json"]); + assert_eq!(approved["session"]["state"], "working", "{approved}"); + d.wait_screen(&id, r#""behavior":"allow""#, WAIT).await; + let (code, _, err) = d.th(&["flow", "approve", &id, "--decision", "maybe"]); + assert_ne!(code, 0, "an unknown decision is refused before any request"); + assert!(err.contains("allow | deny | allow_session"), "{err}"); + let (code, _, err) = d.th(&["flow", "approve", "fs-nope", "--decision", "allow"]); + assert_ne!(code, 0); + assert!(err.contains("no pending permission request") || err.contains("th flow ls"), "{err}"); + + // handoff: git facts from the engine (no pearl → null pearl). + let (code, out, err) = d.th(&["flow", "handoff", &id]); + assert_eq!(code, 0, "{err}"); + let handoff: serde_json::Value = serde_json::from_str(&out).unwrap(); + assert_eq!(handoff["handoff"]["worktree"], ws); + assert_eq!(handoff["handoff"]["branch"], "main"); + assert!(handoff["handoff"]["head"].as_str().unwrap().len() >= 7, "{handoff}"); + assert!(handoff["pearl"].is_null() && handoff["checkpoints"].is_array(), "{handoff}"); + + // kill --json → done; the two-line error contract for an unknown id. + let killed = d.th_json(&["flow", "kill", &id, "--json"]); + assert_eq!(killed["session"]["state"], "done", "{killed}"); + let (code, _, err) = d.th(&["flow", "kill", "fs-nope"]); + assert_ne!(code, 0); + assert!(err.contains("no such session") && err.contains("→ check the id with `th flow ls`"), "{err}"); + + // new with explicit argv after `--` runs that, in the worktree. + let v = d.th_json(&[ + "flow", + "new", + "--kind", + "shell", + "--worktree", + &ws, + "--json", + "--", + "sh", + "-c", + "echo ARGV-OK; exec sh -l", + ]); + let sh = v["session"]["id"].as_str().unwrap().to_string(); + assert_eq!(v["session"]["argv"], json!(["sh", "-c", "echo ARGV-OK; exec sh -l"])); + d.wait_screen(&sh, "ARGV-OK", WAIT).await; + d.th_json(&["flow", "kill", &sh, "--json"]); +} + +#[tokio::test] +async fn th_without_a_daemon_says_so_in_two_lines() { + if !prereqs_with_th() { + return; + } + // A daemon whose HOME has no daemon.addr yet: point `th` at a HOME of + // its own by booting a rig and deleting the advertisement. + let d = Daemon::boot().await; + std::fs::remove_file(d.home.join(".smooth").join("daemon.addr")).unwrap(); + let (code, _, err) = d.th(&["flow", "ls"]); + assert_ne!(code, 0); + assert!(err.contains("no daemon advertised") && err.contains("th up"), "{err}"); + // `th harness list` degrades to the on-disk registry with a note. + let (code, out, _) = d.th(&["harness", "list"]); + assert_eq!(code, 0, "{out}"); + assert!(out.contains("daemon not reachable") && out.contains("fake-agent"), "{out}"); + let v = d.th_json(&["harness", "list", "--json"]); + assert_eq!(v["source"], "local"); +} diff --git a/crates/smooth-daemon/tests/flow_e2e/harnesses.rs b/crates/smooth-daemon/tests/flow_e2e/harnesses.rs new file mode 100644 index 000000000..a3c9607f1 --- /dev/null +++ b/crates/smooth-daemon/tests/flow_e2e/harnesses.rs @@ -0,0 +1,296 @@ +//! Harness manifests end to end: the per-harness state-source matrix, the +//! sort/hide prefs (PUT → `th harness list` order, `flow.hello` omits +//! hidden), `th harness add ` of a custom manifest and a session on it, +//! and — opt-in — the real coding CLIs installed on this machine. + +use std::time::Duration; + +use serde_json::{json, Value}; + +use crate::support::{prereqs, prereqs_with_th, skip, state, Daemon, WAIT}; + +fn names(v: &Value) -> Vec { + v.as_array().unwrap().iter().map(|h| h["name"].as_str().unwrap().to_string()).collect() +} + +fn by_name<'a>(v: &'a Value, name: &str) -> &'a Value { + v.as_array() + .unwrap() + .iter() + .find(|h| h["name"] == name) + .unwrap_or_else(|| panic!("no harness {name} in {v}")) +} + +/// Every state source the engine supports, one fake-agent flavour each: +/// what `Session.state_source` reads once the harness has done a turn. +#[tokio::test] +async fn harness_matrix_state_source_per_manifest() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + let (status, v) = d.get("/api/flow/harnesses").await; + assert_eq!(status, 200); + let rows = &v["harnesses"]; + // The built-ins are always listed, with their manifest's source, whether + // or not the binary resolves on this runner. + for (name, source) in [("claude", "hooks"), ("opencode", "hooks"), ("codex", "hooks"), ("th-code", "native")] { + let h = by_name(rows, name); + assert_eq!(h["state_source"], source, "{h}"); + assert_eq!(h["origin"], "builtin"); + assert_eq!( + h["installed"].is_boolean() && (h["installed"] == true) == h["binary_path"].is_string(), + true, + "{h}" + ); + if h["installed"] == false { + assert!(h["reason"].as_str().unwrap().contains("not found on PATH"), "{h}"); + } + } + let mut matrix = Vec::new(); + for (kind, manifest_source, session_source) in [ + ("fake-agent", "hooks", "hooks"), + ("fake-agent-learned", "hooks", "hooks"), + ("fake-agent-native", "native", "native"), + ("fake-agent-scrape", "scrape", "inferred"), + ] { + let h = by_name(rows, kind); + assert_eq!(h["state_source"], manifest_source, "{h}"); + assert_eq!(h["origin"], "user", "installed from ~/.smooth/harnesses: {h}"); + assert_eq!(h["installed"], true, "{h}"); + assert!(h["binary_path"].as_str().unwrap().ends_with("/.local/bin/fake-agent"), "{h}"); + let s = d.new_session(kind, Some("/work m")).await; + let id = s["id"].as_str().unwrap().to_string(); + assert_eq!(s["state_source"], "inferred", "before any report every harness is inferred: {s}"); + let idle = d + .wait_until(&id, "idle", WAIT, |s| state(s) == "idle" && s["state_source"] == session_source) + .await; + matrix.push((kind, manifest_source, idle["state_source"].as_str().unwrap().to_string())); + d.wait_screen(&id, "worked: m", WAIT).await; + d.kill(&id, false).await; + } + eprintln!("state-source matrix (kind, manifest, session): {matrix:?}"); + assert_eq!( + matrix.iter().map(|(_, _, s)| s.as_str()).collect::>(), + ["hooks", "hooks", "native", "inferred"] + ); + // An unknown kind is refused with the pointer to `th harness list`. + let (status, v) = d.post("/api/flow/sessions", json!({"kind":"aider","worktree":d.ws})).await; + assert_eq!(status, 400, "{v}"); + assert!(v["error"].as_str().unwrap().contains("th harness list"), "{v}"); +} + +#[tokio::test] +async fn harness_prefs_sort_and_hide_reach_every_picker() { + if !prereqs_with_th() { + return; + } + let d = Daemon::boot().await; + let ws0 = d.ws().await; + let before = names(&ws0.hello["harnesses"]); + assert_eq!( + &before[..4], + ["claude", "opencode", "codex", "th-code"], + "built-ins first, in registry order: {before:?}" + ); + assert!(before.contains(&"fake-agent".to_string())); + + // PUT prefs → the reply is the full list (hidden flagged), the WS gets + // flow.harnesses with the visible list, `th harness list` follows. + let mut ws = d.ws().await; + let (status, v) = d + .put("/api/flow/harnesses/prefs", json!({"order":["fake-agent","th-code"],"hidden":["codex"]})) + .await; + assert_eq!(status, 200, "{v}"); + let all = names(&v["harnesses"]); + assert_eq!(&all[..3], ["fake-agent", "th-code", "claude"], "{all:?}"); + assert_eq!(by_name(&v["harnesses"], "codex")["hidden"], true); + let bc = ws.wait_for("flow.harnesses", WAIT, |f| f["type"] == "flow.harnesses").await; + let visible = names(&bc["harnesses"]); + assert!(!visible.contains(&"codex".to_string()), "{visible:?}"); + assert_eq!(&visible[..2], ["fake-agent", "th-code"]); + // A fresh hello omits hidden, keeps the order. + let ws2 = d.ws().await; + let hello = names(&ws2.hello["harnesses"]); + assert_eq!(hello, visible); + + let list = d.th_json(&["harness", "list", "--json"]); + assert_eq!(list["source"], "daemon"); + let shown = names(&list["harnesses"]); + assert_eq!(&shown[..2], ["fake-agent", "th-code"]); + assert!(!shown.contains(&"codex".to_string()), "hidden by default: {shown:?}"); + let all = d.th_json(&["harness", "list", "--all", "--json"]); + assert_eq!(by_name(&all["harnesses"], "codex")["hidden"], true); + let (_, table, _) = d.th(&["harness", "list", "--all"]); + assert!(table.contains("codex") && table.contains("hidden"), "{table}"); + + // The CLI verbs: unhide / hide / order. + let (code, _, err) = d.th(&["harness", "unhide", "codex"]); + assert_eq!(code, 0, "{err}"); + assert!(names(&d.th_json(&["harness", "list", "--json"])["harnesses"]).contains(&"codex".to_string())); + let (code, _, err) = d.th(&["harness", "hide", "opencode"]); + assert_eq!(code, 0, "{err}"); + assert!(!names(&d.th_json(&["harness", "list", "--json"])["harnesses"]).contains(&"opencode".to_string())); + let (code, out, err) = d.th(&["harness", "order", "th-code", "claude"]); + assert_eq!(code, 0, "{err}"); + assert!(out.contains("th-code"), "{out}"); + let shown = names(&d.th_json(&["harness", "list", "--json"])["harnesses"]); + assert_eq!(&shown[..2], ["th-code", "claude"], "{shown:?}"); + // Unknown names are refused, by the daemon and by the CLI. + let (status, v) = d.put("/api/flow/harnesses/prefs", json!({"hidden":["cursor"]})).await; + assert_eq!(status, 400, "{v}"); + assert!(v["error"].as_str().unwrap().contains("cursor"), "{v}"); + let (code, _, err) = d.th(&["harness", "hide", "cursor"]); + assert_ne!(code, 0); + assert!(err.contains("cursor"), "{err}"); + // Prefs survive in flow.db: a fresh registry read applies them. + let (_, v) = d.get("/api/flow/harnesses").await; + assert_eq!(&names(&v["harnesses"])[..2], ["th-code", "claude"]); + assert_eq!(by_name(&v["harnesses"], "opencode")["hidden"], true); + let _ = ws0; +} + +#[tokio::test] +async fn th_harness_add_installs_a_custom_manifest_the_engine_launches() { + if !prereqs_with_th() { + return; + } + let d = Daemon::boot().await; + // A custom manifest: the hooks fake-agent under a new name, from a file + // outside ~/.smooth (what a user would `th harness add`). + let base = std::fs::read_to_string(d.home.join(".smooth/harnesses/fake-agent.toml")).unwrap(); + let custom = base + .replace("name = \"fake-agent\"", "name = \"custom-agent\"") + .replace("display_name = \"Fake Agent (hooks)\"", "display_name = \"Custom Agent\""); + let src = d.home.join("custom-agent.toml"); + std::fs::write(&src, &custom).unwrap(); + + let (code, out, err) = d.th(&["harness", "add", src.to_str().unwrap()]); + assert_eq!(code, 0, "{err}"); + assert!(out.contains("custom-agent") && out.contains(".smooth/harnesses/custom-agent.toml"), "{out}"); + assert!(out.contains("/.local/bin/fake-agent"), "the resolved binary is shown: {out}"); + assert!(d.home.join(".smooth/harnesses/custom-agent.toml").is_file()); + // Adding it again refuses without --force. + let (code, _, err) = d.th(&["harness", "add", src.to_str().unwrap()]); + assert_ne!(code, 0); + assert!(err.contains("--force"), "{err}"); + assert_eq!(d.th(&["harness", "add", src.to_str().unwrap(), "--force"]).0, 0); + // A manifest that fails validation is refused with the field named. + let bad = d.home.join("bad.toml"); + std::fs::write(&bad, custom.replace("prompt_as = \"argv\"", "prompt_as = \"paste\"")).unwrap(); + let (code, _, err) = d.th(&["harness", "add", bad.to_str().unwrap()]); + assert_ne!(code, 0); + assert!(err.contains("prompt") || err.contains("launch.argv"), "{err}"); + assert!(!d.home.join(".smooth/harnesses/bad.toml").exists()); + + // No daemon restart: the daemon lists it, `th harness show` reads it, + // pickers get it on their next hello, and a session launches on it. + let (_, v) = d.get("/api/flow/harnesses").await; + let h = by_name(&v["harnesses"], "custom-agent"); + assert_eq!(h["display_name"], "Custom Agent"); + assert_eq!(h["origin"], "user"); + assert_eq!(h["installed"], true, "{h}"); + let show = d.th_json(&["harness", "show", "custom-agent", "--json"]); + assert_eq!(show["manifest"]["name"], "custom-agent", "{show}"); + assert_eq!(show["origin"], "user"); + assert_eq!(show["installed"], true, "{show}"); + assert!(show["path"].as_str().unwrap().ends_with(".smooth/harnesses/custom-agent.toml"), "{show}"); + let ws = d.ws().await; + assert!(names(&ws.hello["harnesses"]).contains(&"custom-agent".to_string())); + let v = d.th_json(&[ + "flow", + "new", + "--kind", + "custom-agent", + "--worktree", + d.ws.to_str().unwrap(), + "--prompt", + "/work custom", + "--json", + ]); + let id = v["session"]["id"].as_str().unwrap().to_string(); + assert_eq!(v["session"]["kind"], "custom-agent"); + d.wait_until(&id, "idle via hooks", WAIT, |s| state(s) == "idle" && s["state_source"] == "hooks") + .await; + d.wait_screen(&id, "worked: custom", WAIT).await; + d.kill(&id, false).await; +} + +/// Opt-in (`SMOOTH_E2E_REAL_HARNESSES=1`): launch every REAL coding CLI this +/// machine has — claude / opencode / codex where installed, and `th code` — +/// through its built-in manifest, assert the launch shape and the state +/// source the engine reads, then kill it. No credentials are needed: the +/// engine's side of the contract holds before the CLI ever talks to a model. +#[tokio::test] +async fn real_harnesses_launch_through_their_manifests() { + if !prereqs() { + return; + } + if !std::env::var("SMOOTH_E2E_REAL_HARNESSES").is_ok_and(|v| !v.is_empty() && v != "0") { + skip("SMOOTH_E2E_REAL_HARNESSES is not set — real claude/opencode/codex/th-code launches are opt-in"); + return; + } + let d = Daemon::boot().await; + // th code refuses to boot without LLM providers (`th model login`); the + // real ones on this machine, copied into the rig's HOME, let it come up + // and report its first turn natively. Absent, th code is skipped. + let providers = dirs_next::home_dir().map(|h| h.join(".smooth/providers.json")).filter(|p| p.is_file()); + if let Some(p) = &providers { + std::fs::copy(p, d.home.join(".smooth/providers.json")).unwrap(); + } + let (_, v) = d.get("/api/flow/harnesses").await; + let mut matrix = Vec::new(); + for name in ["claude", "opencode", "codex", "th-code"] { + let h = by_name(&v["harnesses"], name); + if h["installed"] != true { + eprintln!("[skip] {name}: {}", h["reason"]); + continue; + } + if name == "th-code" && providers.is_none() { + eprintln!("[skip] th-code: no ~/.smooth/providers.json to boot it with"); + continue; + } + let s = d.new_session(name, Some("say ok and stop")).await; + let id = s["id"].as_str().unwrap().to_string(); + assert_eq!(s["argv"][0], h["binary_path"], "{s}"); + assert_eq!(s["state_source"], "inferred"); + match name { + "claude" => { + assert_eq!(s["argv"][1], "--session-id"); + assert_eq!(s["argv"][2], s["agent_session_id"]); + } + "opencode" => assert!(s["argv"].as_array().unwrap().iter().any(|a| a == "--prompt"), "{s}"), + "codex" => assert!(s["agent_session_id"].is_null(), "learned: {s}"), + "th-code" => { + assert_eq!(s["argv"][1], "code"); + assert!(s["agent_session_id"].is_string(), "{s}"); + } + _ => unreachable!(), + } + // Without credentials a real CLI paints onboarding / sign-in in the + // fresh HOME — a pane no scrape pattern matches, so `starting` is the + // engine's honest reading. What is provable creds-free: the process + // came up and painted, its pid is live, the row's shape is right; + // th code additionally reports natively (its first turn_start + // happens before any model call). + let painted = d.wait_screen_nonblank(&id, Duration::from_secs(60)).await; + assert!(!painted.trim().is_empty()); + let live = d.session(&id).await; + assert!(Daemon::pid_alive(live["pid"].as_u64().unwrap() as u32), "{live}"); + assert_ne!(state(&live), "dead", "{live}"); + if name == "th-code" { + d.wait_until(&id, "native report", Duration::from_secs(60), |s| s["state_source"] == "native") + .await; + } + let live = d.session(&id).await; + matrix.push(( + name, + live["state"].as_str().unwrap().to_string(), + live["state_source"].as_str().unwrap().to_string(), + )); + let killed = d.kill(&id, false).await; + assert_eq!(state(&killed), "done", "{killed}"); + } + eprintln!("real harness matrix (kind, state, source): {matrix:?}"); + assert!(!matrix.is_empty(), "SMOOTH_E2E_REAL_HARNESSES is set but no real harness is installed"); +} diff --git a/crates/smooth-daemon/tests/flow_e2e/hooks.rs b/crates/smooth-daemon/tests/flow_e2e/hooks.rs new file mode 100644 index 000000000..bd4ff0f1c --- /dev/null +++ b/crates/smooth-daemon/tests/flow_e2e/hooks.rs @@ -0,0 +1,228 @@ +//! The `POST /api/flow/hooks` contract, event by event, against a live +//! session — what `flow-hook.sh` (the smooth-agent plugin) and the harness +//! plugins post, and what the engine does with each. + +use std::time::Duration; + +use serde_json::{json, Value}; + +use crate::support::{prereqs, state, Daemon, Ws, WAIT}; + +/// The next `flow.event` for `id` whose text contains `needle`. +async fn expect_event(ws: &mut Ws, id: &str, kind: &str, needle: &str) -> Value { + let what = format!("flow.event {kind} …{needle}…"); + ws.wait_for(&what, WAIT, |v| { + v["type"] == "flow.event" && v["id"] == id && v["kind"] == kind && v["text"].as_str().is_some_and(|t| t.contains(needle)) + }) + .await +} + +#[tokio::test] +async fn hooks_contract_per_event() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + let mut ws = d.ws().await; + // A live agent that just sits at its prompt — the hooks below are what + // ITS hook script would post, with its pre-assigned id. + let s = d.new_session("fake-agent", None).await; + let id = s["id"].as_str().unwrap().to_string(); + let agent = s["agent_session_id"].as_str().unwrap().to_string(); + let cwd = d.ws.to_string_lossy().into_owned(); + let idle = d.wait_state(&id, "idle", WAIT).await; + assert_eq!(idle["state_source"], "hooks", "fake-agent's SessionStart already flipped the source: {idle}"); + let hook = |event: &'static str, payload: Value| { + let d = &d; + let agent = agent.clone(); + let cwd = cwd.clone(); + async move { + let (status, body) = d.hook("claude-code", event, &agent, Some(&cwd), payload).await; + assert_eq!(status, 200, "{event}: {body}"); + body + } + }; + + // SessionStart / PreCompact / SubagentStop / anything unknown: no state change. + for ev in ["SessionStart", "PreCompact", "SubagentStop", "SomethingNew"] { + assert_eq!(hook(ev, json!({"source":"startup"})).await, json!({})); + } + assert_eq!(state(&d.session(&id).await), "idle"); + + // UserPromptSubmit → working + the user line. + assert_eq!(hook("UserPromptSubmit", json!({"prompt":" fix it "})).await, json!({})); + d.wait_state(&id, "working", WAIT).await; + expect_event(&mut ws, &id, "user", "fix it").await; + expect_event(&mut ws, &id, "system", "working").await; + + // PreToolUse → working + the tool line; PostToolUse → working, no line. + hook("PreToolUse", json!({"tool_name":"Bash","tool_input":{"command":"ls -la"}})).await; + expect_event(&mut ws, &id, "tool", "Bash(ls -la)").await; + hook("PostToolUse", json!({"tool_name":"Bash","tool_response":{"stdout":"x"}})).await; + assert_eq!(state(&d.session(&id).await), "working"); + + // Stop → idle, unread, the agent's last message. + hook("Stop", json!({"last_assistant_message":"all done"})).await; + let s = d.wait_until(&id, "idle+unread", WAIT, |s| state(s) == "idle" && s["unread"] == true).await; + assert!(s["attention"].is_null()); + expect_event(&mut ws, &id, "agent", "all done").await; + expect_event(&mut ws, &id, "system", "idle").await; + + // Notification(permission) → needs_you · permission, detail = message, + // no request_id (nothing to long-poll); the message is a system line. + hook( + "Notification", + json!({"notification_type":"permission_prompt","message":"Claude needs your permission to use Bash"}), + ) + .await; + let p = d.wait_state(&id, "needs_you", WAIT).await; + assert_eq!(p["attention"]["reason"], "permission"); + assert_eq!(p["attention"]["detail"], "Claude needs your permission to use Bash"); + assert!(p["attention"]["request_id"].is_null(), "{p}"); + expect_event(&mut ws, &id, "system", "needs your permission").await; + // approve with no pending request presses the key on the pane → working. + // fake-agent reads whole lines, so an empty steer (just Enter) shows the + // `1` the keystroke path typed. + d.approve(&id, "whatever", "allow").await; + d.wait_state(&id, "working", WAIT).await; + d.send(&id, "x").await; + d.wait_screen(&id, "echo: 1x", WAIT).await; + + // Notification(question) → needs_you · question. + hook( + "Notification", + json!({"notification_type":"idle_prompt","message":"Claude is waiting for your input"}), + ) + .await; + let q = d.wait_state(&id, "needs_you", WAIT).await; + assert_eq!(q["attention"]["reason"], "question"); + d.approve(&id, "whatever", "deny").await; + d.wait_state(&id, "working", WAIT).await; + + // Any other notification: a line, no state change. + hook("Notification", json!({"notification_type":"other","message":"fyi only"})).await; + expect_event(&mut ws, &id, "system", "fyi only").await; + assert_eq!(state(&d.session(&id).await), "working"); + + // SessionEnd: the exit is decided by the PTY, not the hook — state holds, + // the line lands. + hook("SessionEnd", json!({"reason":"exit"})).await; + expect_event(&mut ws, &id, "system", "session ended (exit)").await; + assert_eq!(state(&d.session(&id).await), "working"); + + // Unknown session: quiet 200 {} (the hook script must never block). + let (status, body) = d.hook("claude-code", "Stop", "nobody-here", Some(&cwd), json!({})).await; + assert_eq!((status, body), (200, json!({}))); + // Missing session_id: also never a 5xx. + let r = reqwest::Client::new() + .post(d.url("/api/flow/hooks")) + .header("content-type", "application/json") + .body("not json at all") + .send() + .await + .unwrap(); + assert!(r.status().is_client_error(), "malformed body: {}", r.status()); + + d.kill(&id, false).await; +} + +#[tokio::test] +async fn permission_request_long_polls_until_approved_each_decision_shape() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + let s = d.new_session("fake-agent", None).await; + let id = s["id"].as_str().unwrap().to_string(); + let agent = s["agent_session_id"].as_str().unwrap().to_string(); + d.wait_state(&id, "idle", WAIT).await; + + for (decision, want_behavior) in [("deny", "deny"), ("allow", "allow"), ("allow_session", "allow")] { + let (dd, aa) = (d.url("/api/flow/hooks"), agent.clone()); + let post = tokio::spawn(async move { + reqwest::Client::new() + .post(dd) + .json(&json!({"harness":"claude-code","event":"PermissionRequest","session_id":aa, + "payload":{"tool_name":"Bash","tool_input":{"command":"git push"}}})) + .send() + .await + .unwrap() + .json::() + .await + .unwrap() + }); + let ask = d + .wait_until(&id, "needs_you with request_id", WAIT, |s| { + state(s) == "needs_you" && s["attention"]["request_id"].is_string() + }) + .await; + let request_id = ask["attention"]["request_id"].as_str().unwrap().to_string(); + assert_eq!(ask["attention"]["detail"], "Bash: git push"); + tokio::time::sleep(Duration::from_millis(500)).await; + assert!(!post.is_finished(), "held open until a decision"); + + // A Notification for the same prompt keeps the request_id. + d.hook( + "claude-code", + "Notification", + &agent, + None, + json!({"notification_type":"permission_prompt","message":"Claude needs your permission to use Bash"}), + ) + .await; + assert_eq!(d.session(&id).await["attention"]["request_id"], request_id); + + // A wrong request_id falls through to the keystroke path (the row is + // live, so it succeeds) but does NOT answer the long-poll. + d.approve(&id, "not-this-one", "allow").await; + tokio::time::sleep(Duration::from_millis(300)).await; + assert!(!post.is_finished(), "a mismatched request_id must not resolve the hook"); + + d.approve(&id, &request_id, decision).await; + let body = tokio::time::timeout(WAIT, post).await.unwrap().unwrap(); + assert_eq!(body["hookSpecificOutput"]["hookEventName"], "PermissionRequest", "{body}"); + assert_eq!(body["hookSpecificOutput"]["decision"]["behavior"], want_behavior, "{body}"); + match decision { + "deny" => assert!(body["hookSpecificOutput"]["decision"]["message"].is_string()), + "allow_session" => { + let rule = &body["hookSpecificOutput"]["decision"]["updatedPermissions"][0]; + assert_eq!(rule["rules"][0]["toolName"], "Bash"); + assert_eq!(rule["destination"], "session"); + } + _ => assert!(body["hookSpecificOutput"]["decision"]["updatedPermissions"].is_null()), + } + d.wait_state(&id, "working", WAIT).await; + } + d.kill(&id, false).await; +} + +#[tokio::test] +async fn hooks_are_unauthenticated_and_everything_else_is_gated() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + assert_eq!(d.get_unauthed("/api/flow/sessions").await, 401); + assert_eq!(d.get_unauthed("/api/flow/harnesses").await, 401); + assert_eq!(d.get_unauthed("/api/flow/sessions/fs-nope/snapshot").await, 401); + let (status, _) = d.get("/api/flow/sessions").await; + assert_eq!(status, 200); + // Bearer + ?token= are the other two spellings. + let http = reqwest::Client::new(); + let r = http.get(d.url("/api/flow/sessions")).bearer_auth(&d.token).send().await.unwrap(); + assert_eq!(r.status(), 200); + let r = http.get(format!("{}?token={}", d.url("/api/flow/sessions"), d.token)).send().await.unwrap(); + assert_eq!(r.status(), 200); + // Hooks: no token, still 200. + let (status, body) = d.hook("claude-code", "Stop", "nobody", None, json!({})).await; + assert_eq!((status, body), (200, json!({}))); + // The WS handshake is refused without the token. + assert!(tokio_tungstenite::connect_async(format!("ws://{}/api/flow/ws", d.addr)).await.is_err()); + assert!(tokio_tungstenite::connect_async(format!("ws://{}/api/flow/ws?token=nope", d.addr)) + .await + .is_err()); + // Unknown session on a gated route → 404 with an error object. + let (status, v) = d.get("/api/flow/sessions/fs-nope/snapshot").await; + assert_eq!(status, 404, "{v}"); + assert!(v["error"].is_string()); +} diff --git a/crates/smooth-daemon/tests/flow_e2e/isolation.rs b/crates/smooth-daemon/tests/flow_e2e/isolation.rs new file mode 100644 index 000000000..83352aa17 --- /dev/null +++ b/crates/smooth-daemon/tests/flow_e2e/isolation.rs @@ -0,0 +1,177 @@ +//! Proof that the suite never touches the developer's Big Smooth, and that +//! the macOS lane's lighter host (`flow_e2e_server`) is the same engine. + +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; + +use serde_json::json; + +use crate::support::{e2e_server_bin, prereqs, real_daemon_addr, skip, state, Daemon, WAIT}; + +/// The real `~/.smooth/{daemon.addr,daemon.lock,operator-token,flow.db}` +/// — bytes + mtime of each that exists. +fn real_smooth_files() -> Vec<(String, Option<(Vec, std::time::SystemTime)>)> { + let home = dirs_next::home_dir().unwrap(); + ["daemon.addr", "daemon.lock", "operator-token", "flow.db"] + .into_iter() + .map(|f| { + let p = home.join(".smooth").join(f); + let snap = std::fs::read(&p).ok().and_then(|b| std::fs::metadata(&p).ok()?.modified().ok().map(|m| (b, m))); + (f.to_string(), snap) + }) + .collect() +} + +#[tokio::test] +async fn suite_never_writes_the_real_daemon_addr_or_token() { + if !prereqs() { + return; + } + let before = real_smooth_files(); + let real_addr = real_daemon_addr(); + { + let d = Daemon::boot().await; + // The rig's advertisement lives in ITS home, and is not the user's daemon. + let test_addr = std::fs::read_to_string(d.home.join(".smooth/daemon.addr")).unwrap(); + assert_eq!(test_addr.trim(), d.addr); + if let Some((bytes, _)) = &real_addr { + assert_ne!( + String::from_utf8_lossy(bytes).trim(), + d.addr, + "the rig must never land on the user's daemon port" + ); + } + assert!(d.home.join(".smooth/operator-token").is_file()); + assert!(d.home.join(".smooth/flow.db").is_file(), "flow.db is under the test HOME"); + // The rig's tmux server is private too: nothing on `smooth-flow`. + let s = d.new_session("shell", None).await; + let id = s["id"].as_str().unwrap().to_string(); + d.wait_state(&id, "idle", WAIT).await; + assert!(d.socket.starts_with("flow-e2e-")); + let on_default = Command::new("tmux") + .args(["-L", "smooth-flow", "has-session", "-t", &id]) + .stderr(Stdio::null()) + .status() + .unwrap(); + assert!(!on_default.success(), "the session must not exist on the default flow server"); + let on_private = Command::new("tmux").args(["-L", &d.socket, "has-session", "-t", &id]).status().unwrap(); + assert!(on_private.success()); + d.kill(&id, false).await; + } + // Drop killed the daemon and its tmux server; the real files are untouched. + assert_eq!(real_smooth_files(), before, "a real ~/.smooth file changed during the test"); + assert_eq!(real_daemon_addr(), real_addr); +} + +#[tokio::test] +async fn daemon_teardown_leaves_no_tmux_server_or_process() { + if !prereqs() { + return; + } + let (socket, pid) = { + let d = Daemon::boot().await; + let s = d.new_session("fake-agent", Some("/work bye")).await; + let id = s["id"].as_str().unwrap().to_string(); + d.wait_until(&id, "idle", WAIT, |s| state(s) == "idle").await; + (d.socket.clone(), s["pid"].as_u64().unwrap() as u32) + }; + let gone = Command::new("tmux") + .args(["-L", &socket, "list-sessions"]) + .stderr(Stdio::null()) + .stdout(Stdio::null()) + .status() + .unwrap(); + assert!(!gone.success(), "the private tmux server is killed with the rig"); + let start = Instant::now(); + while Daemon::pid_alive(pid) && start.elapsed() < Duration::from_secs(5) { + std::thread::sleep(Duration::from_millis(100)); + } + assert!(!Daemon::pid_alive(pid), "the agent pane's process died with its tmux server"); +} + +/// The macOS XCUITest lane hosts `flow_e2e_server` (the daemon's router + +/// supervisor, nothing else). Same manifests, same fake-agent, same states — +/// so a green mac lane means the same engine the daemon ships. +#[tokio::test] +async fn flow_e2e_server_example_hosts_the_same_engine() { + if !prereqs() { + return; + } + let Some(server) = e2e_server_bin() else { + skip("flow_e2e_server not built: cargo build -p smooai-smooth-daemon --example flow_e2e_server"); + return; + }; + // Borrow the rig for its HOME (fixtures installed) and workspace, but + // point the example at them instead of the daemon. + let d = Daemon::boot().await; + let socket = format!("{}-ex", d.socket); + let mut path = d.home.join(".local/bin").into_os_string(); + path.push(":"); + path.push(std::env::var_os("PATH").unwrap_or_default()); + let mut child = Command::new(&server) + .args(["--addr", "127.0.0.1:0", "--workspace"]) + .arg(&d.ws) + .arg("--db") + .arg(d.home.join("example-flow.db")) + .args(["--token", "ex-tok", "--tmux-socket", &socket]) + .env_clear() + .env("PATH", path) + .env("HOME", &d.home) + .env("RUST_LOG", "warn") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let addr_file = d.ws.join(".flow-e2e-addr"); + let start = Instant::now(); + let addr = loop { + let a = std::fs::read_to_string(&addr_file).unwrap_or_default().trim().to_string(); + if !a.is_empty() { + break a; + } + assert!(start.elapsed() < WAIT, "flow_e2e_server never advertised"); + tokio::time::sleep(Duration::from_millis(100)).await; + }; + let http = reqwest::Client::new(); + let post = |path: String, body: serde_json::Value| { + let http = http.clone(); + let addr = addr.clone(); + async move { + http.post(format!("http://{addr}{path}")) + .header("x-smooth-token", "ex-tok") + .json(&body) + .send() + .await + .unwrap() + .json::() + .await + .unwrap() + } + }; + // No `{daemon_url}` here — fake-agent falls back to ./.flow-e2e-addr, + // exactly what the mac lane relies on. + let v = post("/api/flow/sessions".into(), json!({"kind":"fake-agent","worktree":d.ws,"prompt":"/work ex"})).await; + let id = v["session"]["id"].as_str().unwrap().to_string(); + let start = Instant::now(); + loop { + let list = http + .get(format!("http://{addr}/api/flow/sessions")) + .header("x-smooth-token", "ex-tok") + .send() + .await + .unwrap() + .json::() + .await + .unwrap(); + let s = &list["sessions"][0]; + if s["id"] == id && s["state"] == "idle" && s["state_source"] == "hooks" { + break; + } + assert!(start.elapsed() < WAIT, "never idle via hooks on the example server: {list}"); + tokio::time::sleep(Duration::from_millis(250)).await; + } + post(format!("/api/flow/sessions/{id}/kill"), json!({})).await; + let _ = child.kill(); + let _ = child.wait(); + let _ = Command::new("tmux").args(["-L", &socket, "kill-server"]).output(); +} diff --git a/crates/smooth-daemon/tests/flow_e2e/main.rs b/crates/smooth-daemon/tests/flow_e2e/main.rs new file mode 100644 index 000000000..cea33109f --- /dev/null +++ b/crates/smooth-daemon/tests/flow_e2e/main.rs @@ -0,0 +1,33 @@ +//! SmoothFlow engine e2e — the cmux/orca-style suite (pearl th-8e3087). +//! +//! Every test boots a REAL `smooth-daemon` (its own HOME, port, tmux server; +//! see `support`) and drives it exactly the way the apps, the phones, `th +//! flow` and a harness's hook script do: the flow WS, the HTTP siblings, the +//! `th` binary, `POST /api/flow/hooks`. The agent under test is `fake-agent` +//! (`tests/fixtures/fake-agent`), installed through a harness manifest like +//! any other coding CLI, in four flavours: hooks / learned-id / native / +//! scrape — one per state source the engine supports. +//! +//! Skips (or fails, with `SMOOTH_E2E_STRICT=1`) when tmux, bash, curl or the +//! `th` binary is missing. Run it alone with +//! `cargo nextest run -p smooai-smooth-daemon --test flow_e2e`; the full +//! contract, the runtime budget and the CI split are in +//! docs/Engineering/SmoothFlow-Testing.md. + +#![cfg(unix)] +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::too_many_lines, + clippy::cast_possible_truncation, + reason = "unwrap/expect are the idiom for test assertions; a scenario is one long test on purpose" +)] + +mod support; + +mod agent; +mod cli; +mod harnesses; +mod hooks; +mod isolation; +mod shell; diff --git a/crates/smooth-daemon/tests/flow_e2e/shell.rs b/crates/smooth-daemon/tests/flow_e2e/shell.rs new file mode 100644 index 000000000..c1b8787f1 --- /dev/null +++ b/crates/smooth-daemon/tests/flow_e2e/shell.rs @@ -0,0 +1,161 @@ +//! Shell sessions over the flow WS: new → attach → input echo → resize → +//! snapshot → kill, plus the two ways a shell ends on its own (exit 0 is +//! `done`, a non-zero exit is `dead` — rule 5: the PTY's own report). + +use std::time::Duration; + +use serde_json::json; + +use crate::support::{prereqs, state, unb64, Daemon, TICK, WAIT}; + +#[tokio::test] +async fn shell_lifecycle_new_attach_input_resize_snapshot_kill() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + let mut ws = d.ws().await; + assert_eq!(ws.hello["sessions"], json!([]), "a fresh daemon has no sessions"); + assert!( + ws.hello["harnesses"].as_array().unwrap().iter().any(|h| h["name"] == "fake-agent"), + "{}", + ws.hello + ); + + // new — the direct reply is the row; a shell is `idle` at once. + ws.send(json!({"type":"flow.new","kind":"shell","worktree":d.ws,"title":"e2e shell"})).await; + let created = ws + .wait_for("flow.session for the new shell", WAIT, |v| { + v["type"] == "flow.session" && v["session"]["kind"] == "shell" + }) + .await; + let id = created["session"]["id"].as_str().unwrap().to_string(); + assert!(id.starts_with("fs-"), "{created}"); + let s = d.wait_state(&id, "idle", WAIT).await; + assert_eq!(s["title"], "e2e shell"); + assert_eq!(s["worktree"], d.ws.to_string_lossy().as_ref()); + assert_eq!(s["branch"], "main"); + assert_eq!(s["state_source"], "inferred"); + assert!(s["pid"].as_u64().is_some(), "{s}"); + assert_eq!(s["tmux_socket"], d.socket, "the row records the private tmux server"); + + // attach → the PTY bridge streams output only to attached clients. The + // first frame (the redrawn pane) proves the bridge is up before typing. + ws.attach(&id, 100, 30).await; + ws.wait_for("first flow.output", WAIT, |v| v["type"] == "flow.output" && v["id"] == id).await; + let marker = format!("FLOW-E2E-{}", std::process::id()); + ws.input(&id, &format!("echo {marker}\r")).await; + let mut out = ws.wait_output(&id, &marker, WAIT).await; + assert!(out.contains(&marker), "{out}"); + // …twice: the typed command echo and the command's output. + let deadline = std::time::Instant::now() + WAIT; + while out.matches(&marker).count() < 2 && std::time::Instant::now() < deadline { + if let Some(f) = ws.next(Duration::from_secs(2)).await { + if f["type"] == "flow.output" && f["id"] == id { + out.push_str(&unb64(f["data_b64"].as_str().unwrap_or(""))); + } + } + } + assert!(out.matches(&marker).count() >= 2, "echo + output both stream: {out}"); + + // snapshot — the plain-text pane, sized as attached. + let snap = d.snapshot(&id).await; + assert_eq!(snap["type"], "flow.screen"); + assert!(snap["text"].as_str().unwrap().contains(&marker), "{snap}"); + assert_eq!( + (snap["cols"].as_u64(), snap["rows"].as_u64()), + (Some(100), Some(30)), + "attach sized the pane: {snap}" + ); + + // resize → the pane follows the client. + ws.send(json!({"type":"flow.resize","id":id,"cols":90,"rows":28})).await; + let start = std::time::Instant::now(); + loop { + let snap = d.snapshot(&id).await; + if snap["cols"] == 90 && snap["rows"] == 28 { + break; + } + assert!(start.elapsed() < WAIT, "pane never resized to 90x28: {snap}"); + tokio::time::sleep(Duration::from_millis(200)).await; + } + + // A second client attaching sees the same bytes; detaching one keeps the + // other streaming. + let mut ws2 = d.ws().await; + ws2.attach(&id, 90, 28).await; + ws.send(json!({"type":"flow.detach","id":id})).await; + ws2.input(&id, "echo SECOND-CLIENT\r").await; + ws2.wait_output(&id, "SECOND-CLIENT", WAIT).await; + + // kill → done, exit code recorded, the tmux session gone. + ws2.send(json!({"type":"flow.kill","id":id})).await; + let killed = ws2 + .wait_for("flow.session done", WAIT, |v| { + v["type"] == "flow.session" && v["session"]["id"] == id && state(&v["session"]) == "done" + }) + .await; + assert!(killed["session"]["ended_at"].is_string(), "{killed}"); + let alive = std::process::Command::new("tmux") + .args(["-L", &d.socket, "has-session", "-t", &id]) + .output() + .unwrap(); + assert!(!alive.status.success(), "tmux session should be gone after kill"); + // Input to a dead session is an error object, not a hang. + ws2.send(json!({"type":"flow.input","id":id,"data_b64":"aGk=","seq":42})).await; + let err = ws2.wait_for("flow.error", WAIT, |v| v["type"] == "flow.error").await; + assert_eq!(err["ref"], 42); + assert_eq!(err["code"], "failed", "{err}"); + // A terminal row can be removed; the broadcast tells every client. + let (status, _) = d.post(&format!("/api/flow/sessions/{id}/kill"), json!({})).await; + assert_eq!(status, 200, "killing a done session is idempotent"); +} + +#[tokio::test] +async fn shell_that_exits_is_done_and_a_failing_command_is_dead() { + if !prereqs() { + return; + } + let d = Daemon::boot().await; + let mut ws = d.ws().await; + + // exit 0 by itself → done with exit_code 0 (rule 5: the PTY reported it). + let s = d.new_session("shell", None).await; + let id = s["id"].as_str().unwrap().to_string(); + d.wait_state(&id, "idle", WAIT).await; + ws.attach(&id, 80, 24).await; + ws.wait_for("first flow.output", WAIT, |v| v["type"] == "flow.output" && v["id"] == id).await; + // `exit 0`, explicitly: a bare `exit` in a fresh login sh returns the + // status of the last profile test, which is 1 on macOS. + ws.input(&id, "exit 0\r").await; + let done = d.wait_state(&id, "done", WAIT + TICK).await; + assert_eq!(done["exit_code"], 0, "{done}"); + assert!(done["attention"].is_null(), "a clean exit needs nobody: {done}"); + let ev = ws + .wait_for("flow.session done broadcast", WAIT, |v| { + v["type"] == "flow.session" && v["session"]["id"] == id && state(&v["session"]) == "done" + }) + .await; + assert_eq!(ev["session"]["exit_code"], 0); + + // A non-zero exit → dead, attention `crashed` with the code; shells are + // never resumed (rule 2 is for agents). + let (status, v) = d + .post( + "/api/flow/sessions", + json!({"kind":"shell","worktree":d.ws,"argv":["sh","-c","echo boom; exit 7"]}), + ) + .await; + assert_eq!(status, 200, "{v}"); + let id2 = v["session"]["id"].as_str().unwrap().to_string(); + let dead = d.wait_state(&id2, "dead", WAIT + TICK).await; + assert_eq!(dead["exit_code"], 7, "{dead}"); + assert_eq!(dead["attention"]["reason"], "crashed"); + assert!(dead["attention"]["detail"].as_str().unwrap().contains("exit 7"), "{dead}"); + // The list is newest-first and carries both. + let ids: Vec = d.sessions().await.iter().map(|s| s["id"].as_str().unwrap().to_string()).collect(); + assert_eq!(ids, vec![id2.clone(), id.clone()]); + // Nothing streamed for a session nobody attached to. + let frames = ws.collect(Duration::from_secs(1)).await; + assert!(!frames.iter().any(|f| f["type"] == "flow.output" && f["id"] == id2), "{frames:?}"); +} diff --git a/crates/smooth-daemon/tests/flow_e2e/support.rs b/crates/smooth-daemon/tests/flow_e2e/support.rs new file mode 100644 index 000000000..834ea37cf --- /dev/null +++ b/crates/smooth-daemon/tests/flow_e2e/support.rs @@ -0,0 +1,719 @@ +//! The e2e rig: a REAL `smooth-daemon` per test on an ephemeral port, fully +//! isolated from the developer's Big Smooth — its own `$HOME` (so +//! `~/.smooth/{daemon.addr,operator-token,flow.db}` and `~/.smooth/harnesses/` +//! are throwaway), its own tmux server (`tmux -L flow-e2e--`), no +//! single-instance lock, no `tailscale serve`, no relay, no gateway +//! credentials. `fake-agent` and its four manifests are installed into that +//! HOME so the engine launches it the way it launches any harness. +//! +//! Everything here is driven the way real clients drive it: HTTP + the flow +//! WS for the apps, the `th` binary for the CLI, `POST /api/flow/hooks` for +//! what a harness's hook script posts. + +#![allow(dead_code, reason = "each test file uses a different slice of the rig")] + +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::time::{Duration, Instant, SystemTime}; + +use futures_util::stream::{SplitSink, SplitStream}; +use futures_util::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use tokio::net::TcpStream; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::{MaybeTlsStream, WebSocketStream}; + +/// How long a daemon may take to advertise its address. +const BOOT_TIMEOUT: Duration = Duration::from_secs(60); +/// The default wait for a state / snapshot / frame. The machine running this +/// suite is shared with other agents (load 10–20 is normal here), so waits +/// are generous and every one of them polls. +pub const WAIT: Duration = Duration::from_secs(30); +/// Supervision cadence in the daemon — a state driven by the tick lands +/// within one of these after its cause. +pub const TICK: Duration = Duration::from_secs(2); + +static SEQ: AtomicU32 = AtomicU32::new(0); + +fn fixtures() -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")).join("tests").join("fixtures") +} + +/// The `smooth-daemon` this test crate was built against. +pub fn daemon_bin() -> PathBuf { + PathBuf::from(env!("CARGO_BIN_EXE_smooth-daemon")) +} + +/// `$SMOOTH_TH_BIN`, else the `th` cargo put next to the daemon (a workspace +/// `cargo test`/`nextest run` builds both), else none. +pub fn th_bin() -> Option { + if let Some(p) = std::env::var_os("SMOOTH_TH_BIN").filter(|p| !p.is_empty()) { + return Some(PathBuf::from(p)); + } + let sibling = daemon_bin().with_file_name("th"); + sibling.is_file().then_some(sibling) +} + +/// The `flow_e2e_server` example next to the daemon (the macOS lane's host), +/// when it was built. +pub fn e2e_server_bin() -> Option { + let p = daemon_bin().parent()?.join("examples").join("flow_e2e_server"); + p.is_file().then_some(p) +} + +/// `SMOOTH_E2E_STRICT=1` (CI) turns every skip into a failure, so a runner +/// missing tmux/bash/curl/`th` can never report green having run nothing. +pub fn strict() -> bool { + std::env::var("SMOOTH_E2E_STRICT").is_ok_and(|v| !v.is_empty() && v != "0") +} + +fn have(bin: &str, arg: &str) -> bool { + Command::new(bin) + .arg(arg) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .is_ok_and(|s| s.success()) +} + +/// Skip (or, strict, fail) with `why`. Returns `false` when the test should +/// return early. +pub fn skip(why: &str) -> bool { + assert!(!strict(), "SMOOTH_E2E_STRICT is set and a prerequisite is missing: {why}"); + eprintln!("[skip] {why}"); + false +} + +/// tmux + bash + curl — what the rig and `fake-agent` need. +pub fn prereqs() -> bool { + if !have("tmux", "-V") { + return skip("tmux is not installed"); + } + if !have("bash", "--version") { + return skip("bash is not installed"); + } + if !have("curl", "--version") { + return skip("curl is not installed"); + } + true +} + +/// Prereqs plus a `th` binary. +pub fn prereqs_with_th() -> bool { + prereqs() && (th_bin().is_some() || skip("no `th` binary: build smooai-smooth-cli or set SMOOTH_TH_BIN")) +} + +/// The real `~/.smooth/daemon.addr` of the user running this suite — +/// bytes + mtime, or `None` — for the never-touch-the-real-daemon proof. +pub fn real_daemon_addr() -> Option<(Vec, SystemTime)> { + let p = dirs_next::home_dir()?.join(".smooth").join("daemon.addr"); + let bytes = std::fs::read(&p).ok()?; + let mtime = std::fs::metadata(&p).ok()?.modified().ok()?; + Some((bytes, mtime)) +} + +/// Base64 of `s` (the `flow.input` payload). +pub fn b64(s: &str) -> String { + use base64::Engine as _; + base64::engine::general_purpose::STANDARD.encode(s.as_bytes()) +} + +/// Decode a `flow.output` payload. +pub fn unb64(s: &str) -> String { + use base64::Engine as _; + String::from_utf8_lossy(&base64::engine::general_purpose::STANDARD.decode(s).unwrap_or_default()).into_owned() +} + +/// Sessions are `starting` until the first supervision tick or hook. +pub fn state(v: &Value) -> &str { + v.get("state").and_then(Value::as_str).unwrap_or("") +} + +pub fn sid(v: &Value) -> String { + v.get("id").and_then(Value::as_str).unwrap_or("").to_string() +} + +/// One booted daemon and the throwaway world it lives in. +pub struct Daemon { + root: tempfile::TempDir, + pub home: PathBuf, + /// A git repo (branch `main`, one commit) sessions run in. + pub ws: PathBuf, + pub addr: String, + pub token: String, + pub socket: String, + child: Child, + log_path: PathBuf, + http: reqwest::Client, +} + +impl Daemon { + /// Boot. Panics (with the daemon log) when it doesn't come up. + pub async fn boot() -> Self { + let n = SEQ.fetch_add(1, Ordering::Relaxed); + let root = tempfile::Builder::new().prefix("flow-e2e-").tempdir().expect("tempdir"); + let home = root.path().join("home"); + let ws = root.path().join("ws"); + let socket = format!("flow-e2e-{}-{n}", std::process::id()); + install_fixtures(&home); + git_init(&ws); + + let log_path = root.path().join("daemon.log"); + let log = std::fs::File::create(&log_path).expect("daemon log"); + let err = log.try_clone().expect("daemon log"); + let mut path = home.join(".local").join("bin").into_os_string(); + if let Some(p) = std::env::var_os("PATH") { + path.push(":"); + path.push(p); + } + let mut cmd = Command::new(daemon_bin()); + cmd.args(["operator", "--addr", "127.0.0.1:0", "--tmux-socket", &socket]) + .env_clear() + .env("PATH", &path) + .env("HOME", &home) + .env("TMPDIR", root.path()) + .env("SMOOTH_ALLOW_SECOND_DAEMON", "1") + .env("SMOOTH_RELAY", "0") + .env("SMOOTH_TAILSCALE_SERVE", "0") + .env("SMOOTH_WORKSPACE", &ws) + .env("RUST_LOG", "info,smooth_flow=debug,smooth_daemon::flow_route=debug") + .env("TERM", "xterm-256color") + .current_dir(&ws) + .stdin(Stdio::null()) + .stdout(Stdio::from(log)) + .stderr(Stdio::from(err)); + if let Some(th) = th_bin() { + cmd.env("SMOOTH_TH_BIN", th); + } + let mut child = cmd.spawn().expect("spawn smooth-daemon"); + + // A second instance (SMOOTH_ALLOW_SECOND_DAEMON) deliberately does NOT + // advertise itself in ~/.smooth/daemon.addr (#546: SmoothFlow's + // daemon must never repoint `th` at itself). The bound port is on the + // daemon's own "listening" log line; the rig then writes daemon.addr + // in ITS home so `th flow` / `th harness` find this daemon. + let addr_file = home.join(".smooth").join("daemon.addr"); + let token_file = home.join(".smooth").join("operator-token"); + let start = Instant::now(); + let addr = loop { + if let Some(status) = child.try_wait().expect("try_wait") { + panic!( + "smooth-daemon exited during boot ({status}):\n{}", + std::fs::read_to_string(&log_path).unwrap_or_default() + ); + } + let log = std::fs::read_to_string(&log_path).unwrap_or_default(); + if let Some(addr) = listening_addr(&log) { + break addr; + } + assert!( + start.elapsed() < BOOT_TIMEOUT, + "smooth-daemon did not report a listening address within {BOOT_TIMEOUT:?}:\n{log}" + ); + tokio::time::sleep(Duration::from_millis(100)).await; + }; + std::fs::write(&addr_file, format!("{addr}\n")).expect("write the rig's daemon.addr"); + let token = std::fs::read_to_string(&token_file).expect("operator-token").trim().to_string(); + assert!(!token.is_empty(), "empty operator-token"); + let http = reqwest::Client::builder().timeout(Duration::from_secs(150)).build().expect("reqwest"); + let d = Self { + root, + home, + ws, + addr, + token, + socket, + child, + log_path, + http, + }; + // The router is merged into the operator server; once the address is + // advertised it is listening, but poll the flow route anyway. + loop { + if let Ok(r) = d.http.get(d.url("/api/flow/sessions")).header("x-smooth-token", &d.token).send().await { + if r.status().is_success() { + break; + } + } + assert!(start.elapsed() < BOOT_TIMEOUT, "flow routes never answered:\n{}", d.log()); + tokio::time::sleep(Duration::from_millis(100)).await; + } + d + } + + pub fn url(&self, path: &str) -> String { + format!("http://{}{path}", self.addr) + } + + pub fn ws_url(&self) -> String { + format!("ws://{}/api/flow/ws?token={}", self.addr, self.token) + } + + /// The daemon's log so far. + pub fn log(&self) -> String { + std::fs::read_to_string(&self.log_path).unwrap_or_default() + } + + /// `fake-agent`'s own log in the workspace (argv, every hook + reply). + pub fn agent_log(&self) -> String { + std::fs::read_to_string(self.ws.join(".fake-agent.log")).unwrap_or_default() + } + + /// `./.fake-agent-script` — commands fake-agent runs on EVERY start. + pub fn write_script(&self, dir: &Path, lines: &[&str]) { + std::fs::write(dir.join(".fake-agent-script"), format!("{}\n", lines.join("\n"))).expect("script"); + } + + /// A second git repo beside `ws` (another worktree for a session). + pub fn extra_repo(&self, name: &str) -> PathBuf { + let p = self.root.path().join(name); + git_init(&p); + p + } + + // ── HTTP ────────────────────────────────────────────────────────────── + + pub async fn get(&self, path: &str) -> (u16, Value) { + let r = self.http.get(self.url(path)).header("x-smooth-token", &self.token).send().await.expect("GET"); + let status = r.status().as_u16(); + (status, r.json().await.unwrap_or(Value::Null)) + } + + pub async fn post(&self, path: &str, body: Value) -> (u16, Value) { + let r = self + .http + .post(self.url(path)) + .header("x-smooth-token", &self.token) + .json(&body) + .send() + .await + .expect("POST"); + let status = r.status().as_u16(); + (status, r.json().await.unwrap_or(Value::Null)) + } + + pub async fn put(&self, path: &str, body: Value) -> (u16, Value) { + let r = self + .http + .put(self.url(path)) + .header("x-smooth-token", &self.token) + .json(&body) + .send() + .await + .expect("PUT"); + let status = r.status().as_u16(); + (status, r.json().await.unwrap_or(Value::Null)) + } + + /// A raw request with no token (auth tests). + pub async fn get_unauthed(&self, path: &str) -> u16 { + self.http.get(self.url(path)).send().await.expect("GET").status().as_u16() + } + + /// `POST /api/flow/hooks` — what a hook script posts. Never sends the + /// token (hooks are unauthenticated by contract). Waits up to 150 s so a + /// `PermissionRequest` long-poll can be awaited. + pub async fn hook(&self, harness: &str, event: &str, session_id: &str, cwd: Option<&str>, payload: Value) -> (u16, Value) { + let mut body = json!({ "harness": harness, "event": event, "session_id": session_id, "payload": payload }); + if let Some(c) = cwd { + body["cwd"] = json!(c); + } + let r = self.http.post(self.url("/api/flow/hooks")).json(&body).send().await.expect("POST hooks"); + let status = r.status().as_u16(); + (status, r.json().await.unwrap_or(Value::Null)) + } + + pub async fn sessions(&self) -> Vec { + let (status, v) = self.get("/api/flow/sessions").await; + assert_eq!(status, 200, "{v}"); + v["sessions"].as_array().cloned().unwrap_or_default() + } + + pub async fn session(&self, id: &str) -> Value { + self.sessions() + .await + .into_iter() + .find(|s| s["id"] == id) + .unwrap_or_else(|| panic!("no session {id}")) + } + + /// `POST /api/flow/sessions` for `kind` in the workspace. + pub async fn new_session(&self, kind: &str, prompt: Option<&str>) -> Value { + self.new_session_in(kind, prompt, &self.ws.clone()).await + } + + pub async fn new_session_in(&self, kind: &str, prompt: Option<&str>, worktree: &Path) -> Value { + let (status, v) = self + .post("/api/flow/sessions", json!({ "kind": kind, "worktree": worktree, "prompt": prompt })) + .await; + assert_eq!(status, 200, "new {kind}: {v}\n{}", self.log()); + v["session"].clone() + } + + pub async fn send(&self, id: &str, text: &str) { + let (status, v) = self.post(&format!("/api/flow/sessions/{id}/send"), json!({ "text": text })).await; + assert_eq!(status, 200, "send: {v}"); + } + + pub async fn approve(&self, id: &str, request_id: &str, decision: &str) -> Value { + let (status, v) = self + .post( + &format!("/api/flow/sessions/{id}/approve"), + json!({ "request_id": request_id, "decision": decision }), + ) + .await; + assert_eq!(status, 200, "approve: {v}"); + v["session"].clone() + } + + pub async fn kill(&self, id: &str, resume: bool) -> Value { + let (status, v) = self.post(&format!("/api/flow/sessions/{id}/kill"), json!({ "resume": resume })).await; + assert_eq!(status, 200, "kill: {v}"); + v["session"].clone() + } + + pub async fn snapshot(&self, id: &str) -> Value { + let (status, v) = self.get(&format!("/api/flow/sessions/{id}/snapshot")).await; + assert_eq!(status, 200, "snapshot: {v}"); + v + } + + pub async fn screen(&self, id: &str) -> String { + self.snapshot(id).await["text"].as_str().unwrap_or("").to_string() + } + + /// The pane, or the engine's error for a dead one — for failure messages. + pub async fn screen_lossy(&self, id: &str) -> String { + let (_, v) = self.get(&format!("/api/flow/sessions/{id}/snapshot")).await; + v["text"].as_str().map_or_else(|| format!(""), str::to_string) + } + + /// Poll the pane until it shows anything at all (a real CLI painting + /// its first screen). + pub async fn wait_screen_nonblank(&self, id: &str, timeout: Duration) -> String { + let start = Instant::now(); + while start.elapsed() < timeout { + let s = self.screen(id).await; + if !s.trim().is_empty() { + return s; + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + panic!("pane of {id} stayed blank for {timeout:?}\nagent log:\n{}", self.agent_log()); + } + + // ── waits (always polled, never slept-and-hoped) ───────────────────── + + /// Poll the session until `pred` holds; panics with the session, the + /// pane and the daemon log otherwise. + pub async fn wait_until(&self, id: &str, what: &str, timeout: Duration, pred: impl Fn(&Value) -> bool) -> Value { + let start = Instant::now(); + let mut last = Value::Null; + while start.elapsed() < timeout { + last = self.session(id).await; + if pred(&last) { + return last; + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + let screen = self.screen_lossy(id).await; + panic!( + "session {id} never reached `{what}` within {timeout:?}\nlast: {last}\npane:\n{screen}\nagent log:\n{}\ndaemon log tail:\n{}", + self.agent_log(), + tail(&self.log(), 40) + ); + } + + pub async fn wait_state(&self, id: &str, want: &str, timeout: Duration) -> Value { + self.wait_until(id, want, timeout, |s| state(s) == want).await + } + + /// Poll the pane until it contains `needle`. + pub async fn wait_screen(&self, id: &str, needle: &str, timeout: Duration) -> String { + let start = Instant::now(); + let mut last = String::new(); + while start.elapsed() < timeout { + last = self.screen(id).await; + if last.contains(needle) { + return last; + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + panic!( + "pane of {id} never showed `{needle}` within {timeout:?}\npane:\n{last}\nagent log:\n{}", + self.agent_log() + ); + } + + // ── th ──────────────────────────────────────────────────────────────── + + /// Run `th ` against THIS daemon (its HOME carries daemon.addr + + /// operator-token). Returns (exit code, stdout, stderr). + pub fn th(&self, args: &[&str]) -> (i32, String, String) { + let th = th_bin().expect("th binary (checked by prereqs_with_th)"); + let mut path = self.home.join(".local").join("bin").into_os_string(); + if let Some(p) = std::env::var_os("PATH") { + path.push(":"); + path.push(p); + } + let out = Command::new(&th) + .args(args) + .env_clear() + .env("PATH", path) + .env("HOME", &self.home) + .env("TMPDIR", self.root.path()) + .env("NO_COLOR", "1") + .env("TERM", "dumb") + .current_dir(&self.ws) + .output() + .unwrap_or_else(|e| panic!("run {}: {e}", th.display())); + ( + out.status.code().unwrap_or(-1), + String::from_utf8_lossy(&out.stdout).into_owned(), + String::from_utf8_lossy(&out.stderr).into_owned(), + ) + } + + /// `th ` that must succeed and print JSON. + pub fn th_json(&self, args: &[&str]) -> Value { + let (code, out, err) = self.th(args); + assert_eq!(code, 0, "th {} failed:\nstdout: {out}\nstderr: {err}", args.join(" ")); + serde_json::from_str(&out).unwrap_or_else(|e| panic!("th {}: not JSON ({e}):\n{out}", args.join(" "))) + } + + // ── flow WS ─────────────────────────────────────────────────────────── + + pub async fn ws(&self) -> Ws { + Ws::connect(&self.ws_url()).await + } + + /// Open the flow.db this daemon writes (WAL — a second writer is fine). + pub fn store(&self) -> smooth_flow::FlowStore { + smooth_flow::FlowStore::open(&self.home.join(".smooth").join("flow.db")).expect("open flow.db") + } + + /// Is `pid` alive? + pub fn pid_alive(pid: u32) -> bool { + Command::new("kill") + .args(["-0", &pid.to_string()]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .is_ok_and(|s| s.success()) + } +} + +impl Drop for Daemon { + fn drop(&mut self) { + let _ = Command::new("kill").args(["-TERM", &self.child.id().to_string()]).status(); + let start = Instant::now(); + while start.elapsed() < Duration::from_secs(5) { + if self.child.try_wait().ok().flatten().is_some() { + break; + } + std::thread::sleep(Duration::from_millis(50)); + } + let _ = self.child.kill(); + let _ = self.child.wait(); + let _ = Command::new("tmux").args(["-L", &self.socket, "kill-server"]).output(); + if std::thread::panicking() { + eprintln!("--- daemon log ({}) ---\n{}", self.log_path.display(), tail(&self.log(), 60)); + } + } +} + +/// A flow WS client: `hello` consumed, frames read with a deadline. +pub struct Ws { + sink: SplitSink>, Message>, + source: SplitStream>>, + pub hello: Value, +} + +impl Ws { + pub async fn connect(url: &str) -> Self { + let (ws, _) = tokio_tungstenite::connect_async(url).await.expect("flow ws connect"); + let (sink, mut source) = ws.split(); + let hello = next_text(&mut source, Duration::from_secs(10)).await.expect("hello"); + assert_eq!(hello["type"], "flow.hello", "{hello}"); + Self { sink, source, hello } + } + + pub async fn send(&mut self, mut frame: Value) { + if frame.get("channel").is_none() { + frame["channel"] = json!("flow"); + } + self.sink.send(Message::Text(frame.to_string().into())).await.expect("ws send"); + } + + /// The next frame, or `None` at the deadline. + pub async fn next(&mut self, timeout: Duration) -> Option { + next_text(&mut self.source, timeout).await + } + + /// Drain frames until `pred` matches one; panics at the deadline listing + /// what was seen. + pub async fn wait_for(&mut self, what: &str, timeout: Duration, pred: impl Fn(&Value) -> bool) -> Value { + let deadline = Instant::now() + timeout; + let mut seen = Vec::new(); + while Instant::now() < deadline { + let left = deadline.saturating_duration_since(Instant::now()); + match self.next(left).await { + Some(v) if pred(&v) => return v, + Some(v) => seen.push(brief(&v)), + None => break, + } + } + panic!("no `{what}` frame within {timeout:?}; saw:\n {}", seen.join("\n ")); + } + + /// Collect frames for `dur` (everything that arrives). + pub async fn collect(&mut self, dur: Duration) -> Vec { + let deadline = Instant::now() + dur; + let mut out = Vec::new(); + while Instant::now() < deadline { + let left = deadline.saturating_duration_since(Instant::now()); + match self.next(left).await { + Some(v) => out.push(v), + None => break, + } + } + out + } + + pub async fn attach(&mut self, id: &str, cols: u16, rows: u16) { + self.send(json!({"type":"flow.attach","id":id,"cols":cols,"rows":rows})).await; + } + + pub async fn input(&mut self, id: &str, text: &str) { + self.send(json!({"type":"flow.input","id":id,"data_b64":b64(text)})).await; + } + + /// Wait until the concatenated `flow.output` for `id` contains `needle`. + pub async fn wait_output(&mut self, id: &str, needle: &str, timeout: Duration) -> String { + let deadline = Instant::now() + timeout; + let mut acc = String::new(); + while Instant::now() < deadline { + let left = deadline.saturating_duration_since(Instant::now()); + let Some(v) = self.next(left).await else { break }; + if v["type"] == "flow.output" && v["id"] == id { + acc.push_str(&unb64(v["data_b64"].as_str().unwrap_or(""))); + if acc.contains(needle) { + return acc; + } + } + } + panic!("output of {id} never contained `{needle}` within {timeout:?}; got:\n{acc}"); + } +} + +async fn next_text(source: &mut SplitStream>>, timeout: Duration) -> Option { + loop { + let msg = tokio::time::timeout(timeout, source.next()).await.ok()??; + match msg { + Ok(Message::Text(t)) => return serde_json::from_str(&t).ok(), + Ok(Message::Close(_)) | Err(_) => return None, + Ok(_) => {} + } + } +} + +/// A one-line rendering of a frame for failure messages. +pub fn brief(v: &Value) -> String { + let ty = v["type"].as_str().unwrap_or("?"); + match ty { + "flow.session" => format!( + "flow.session {} {} src={}", + v["session"]["id"], v["session"]["state"], v["session"]["state_source"] + ), + "flow.output" => format!("flow.output {} seq={}", v["id"], v["seq"]), + "flow.event" => format!("flow.event {} {} {:?}", v["id"], v["kind"], v["text"]), + "flow.attention" => format!("flow.attention {} {}", v["id"], v["attention"]), + _ => { + let s = v.to_string(); + s.chars().take(160).collect() + } + } +} + +/// The `host:port` from the daemon's `… operator listening … addr= …` +/// log line, once it is there. +pub fn listening_addr(log: &str) -> Option { + // tracing colours the log even into a file; strip `ESC[...m` first. + let line = strip_ansi(log.lines().find(|l| l.contains("operator listening"))?); + let rest = line.split("addr=").nth(1)?; + let addr: String = rest.chars().take_while(|c| !c.is_whitespace()).collect(); + (addr.contains(':') && !addr.ends_with(":0")).then_some(addr) +} + +fn strip_ansi(s: &str) -> String { + let mut out = String::with_capacity(s.len()); + let mut chars = s.chars().peekable(); + while let Some(c) = chars.next() { + if c == '\u{1b}' && chars.peek() == Some(&'[') { + for d in chars.by_ref() { + if d.is_ascii_alphabetic() { + break; + } + } + } else { + out.push(c); + } + } + out +} + +pub fn tail(s: &str, n: usize) -> String { + let lines: Vec<&str> = s.lines().collect(); + let start = lines.len().saturating_sub(n); + lines[start..].join("\n") +} + +/// The local clock time `secs` from now as `h:mm(am|pm)` — what a usage-limit +/// banner says, at the minute resolution the parser reads. +pub fn local_clock_in(secs: i64) -> String { + let t = chrono::Local::now() + chrono::Duration::seconds(secs); + let (h, m) = (t.format("%I").to_string(), t.format("%M").to_string()); + let h = h.trim_start_matches('0'); + let h = if h.is_empty() { "12" } else { h }; + format!("{h}:{m}{}", t.format("%p").to_string().to_lowercase()) +} + +fn install_fixtures(home: &Path) { + let bin = home.join(".local").join("bin"); + let manifests = home.join(".smooth").join("harnesses"); + std::fs::create_dir_all(&bin).expect("mkdir bin"); + std::fs::create_dir_all(&manifests).expect("mkdir harnesses"); + let agent = fixtures().join("fake-agent"); + let dest = bin.join("fake-agent"); + std::fs::copy(&agent, &dest).expect("copy fake-agent"); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&dest, std::fs::Permissions::from_mode(0o755)).expect("chmod fake-agent"); + } + for entry in std::fs::read_dir(fixtures().join("harnesses")).expect("fixtures/harnesses") { + let p = entry.expect("entry").path(); + if p.extension().is_some_and(|e| e == "toml") { + std::fs::copy(&p, manifests.join(p.file_name().expect("name"))).expect("copy manifest"); + } + } +} + +fn git_init(dir: &Path) { + std::fs::create_dir_all(dir).expect("mkdir ws"); + let run = |args: &[&str]| { + let out = Command::new("git") + .args(args) + .current_dir(dir) + .env("GIT_AUTHOR_NAME", "e2e") + .env("GIT_AUTHOR_EMAIL", "e2e@test") + .env("GIT_COMMITTER_NAME", "e2e") + .env("GIT_COMMITTER_EMAIL", "e2e@test") + .output() + .expect("git"); + assert!(out.status.success(), "git {}: {}", args.join(" "), String::from_utf8_lossy(&out.stderr)); + }; + run(&["init", "-q", "-b", "main"]); + run(&["commit", "-q", "--allow-empty", "-m", "init"]); +} diff --git a/crates/smooth-flow/src/engine.rs b/crates/smooth-flow/src/engine.rs index d13ca5245..b76dc8965 100644 --- a/crates/smooth-flow/src/engine.rs +++ b/crates/smooth-flow/src/engine.rs @@ -138,6 +138,8 @@ struct Inner { ptys: Mutex>>, pending: Mutex>, rt: Mutex, + /// Per-session serialisation between `kill` and the supervision tick. + session_locks: Mutex>>>, info: DaemonInfo, default_project: PathBuf, home: PathBuf, @@ -359,6 +361,7 @@ impl Engine { ptys: Mutex::new(HashMap::new()), pending: Mutex::new(HashMap::new()), rt: Mutex::new(Runtime::default()), + session_locks: Mutex::new(HashMap::new()), info: DaemonInfo { version: cfg.version, machine_label: cfg.machine_label, @@ -882,6 +885,8 @@ impl Engine { /// # Errors /// When the session is unknown or the relaunch fails. pub fn kill(&self, id: &str, resume: bool) -> Result { + let lock = self.session_lock(id); + let _held = lock.lock().unwrap_or_else(std::sync::PoisonError::into_inner); let s = self.require(id)?; let sock = socket_of(&s); let tmux_name = s.tmux_session.clone(); @@ -903,6 +908,17 @@ impl Engine { self.set_state(id, SessionState::Done, None)?.ok_or_else(|| anyhow!("no such session: {id}")) } + /// The mutex serialising `kill` and supervision for one session. + fn session_lock(&self, id: &str) -> Arc> { + self.inner + .session_locks + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .entry(id.to_string()) + .or_default() + .clone() + } + fn drop_pty(&self, id: &str) { let removed = self.inner.ptys.lock().unwrap_or_else(std::sync::PoisonError::into_inner).remove(id); if let Some(p) = removed { @@ -923,6 +939,7 @@ impl Engine { tmux::kill_session(&socket_of(&s), t); } self.with_store(|st| st.remove(id))?; + self.inner.session_locks.lock().unwrap_or_else(std::sync::PoisonError::into_inner).remove(id); self.emit(ServerFrame::SessionRemoved { id: id.to_string() }); Ok(()) } @@ -1007,6 +1024,9 @@ impl Engine { .find_map(|o| o.pid.filter(|p| proc::is_alive(*p, o.pid_start))) }); if let Some(pid) = holder { + // A backoff scheduled by an earlier death would fire into this + // same refusal on the next tick; drop it with the hold. + self.rt().relaunch_at.remove(&s.id); let att = Attention::new("held").with_detail(format!("session {agent} is owned by live pid {pid}")); return self .set_state(&s.id, SessionState::NeedsYou, Some(att))? @@ -1046,6 +1066,23 @@ impl Engine { tracing::debug!(session = %ev.session_id, event = %ev.event, "flow hook for an unknown session"); return Ok(HookReply::Immediate(json!({}))); }; + // Serialise this row against `kill` and supervision, then re-read it: + // `s` was resolved before the lock, and a `kill --resume` landing in + // that window wrote rule 4's `held` under us. + let lock = self.session_lock(&s.id); + let _guard = lock.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(s) = self.with_store(|st| st.get(&s.id))? else { + return Ok(HookReply::Immediate(json!({}))); + }; + // Rule 4 parked this row (`held`): another live process owns this + // harness session id, so a hook carrying it is not evidence about + // THIS row. Letting one through un-held the row (a dying agent's + // last `Stop` read as "idle"), and the next supervision pass then + // took its dead tmux session for a crash (th-8e3087). + if s.attention.as_ref().is_some_and(|a| a.reason == "held") { + tracing::debug!(session = %s.id, event = %ev.event, "flow hook for a held session — ignored"); + return Ok(HookReply::Immediate(json!({}))); + } // th-0f6126: the manifest says how this harness's events read // (`state.hooks.event_map`), and whether they count as `hooks` or // `native` state; an empty map is the Claude Code table. @@ -1100,6 +1137,11 @@ impl Engine { self.set_state(&s.id, SessionState::NeedsYou, Some(att))?; } } + HookOutcome::Started => { + if s.state == SessionState::Starting { + self.set_state(&s.id, SessionState::Idle, None)?; + } + } HookOutcome::Ended | HookOutcome::None => {} } Ok(HookReply::Immediate(json!({}))) @@ -1157,6 +1199,13 @@ impl Engine { } fn supervise_one(&self, s: &Session, now: DateTime) -> Result<()> { + // A `kill` is mid-flight on this row — it owns the outcome. Without + // this, `kill --resume` tore the tmux session down between this pass's + // liveness check and its write, so the tick read the kill as an + // unexpected death and overwrote rule 4's `held` with a crash backoff + // (th-8e3087, ~1-in-3 under load). + let lock = self.session_lock(&s.id); + let Ok(_held) = lock.try_lock() else { return Ok(()) }; // A relaunch is scheduled (rule 2 backoff) — fire it when due. let due = self.rt().relaunch_at.get(&s.id).copied(); if let Some(at) = due { @@ -1166,12 +1215,31 @@ impl Engine { } return Ok(()); } + // Rule 4 parked this row (`held`: its harness session is owned by a + // live pid). Its tmux session is gone, which is NOT an unexpected + // death — a human unholds it (`kill --resume` once the holder is + // gone). Without this the tick re-scheduled a resume every backoff, + // the guard refused it again, and the row flapped starting ↔ held + // forever (th-8e3087). + // + // Re-read the row: `s` is a snapshot taken at the top of the tick, so + // a hold written by a concurrent `kill --resume` (the common case — + // holding is what kills the tmux session this pass is reacting to) is + // not in it, and the stale copy sent the row to `on_death` instead. + let fresh = self.with_store(|st| st.get(&s.id))?; + let Some(s) = fresh.as_ref() else { return Ok(()) }; + if s.attention.as_ref().is_some_and(|a| a.reason == "held") { + return Ok(()); + } let Some(t) = s.tmux_session.as_deref() else { return Ok(()) }; let sock = socket_of(s); - if !tmux::session_alive(&sock, t) { + let alive = tmux::session_alive(&sock, t); + let exit = if alive { tmux::pane_exit_status(&sock, t) } else { Ok(None) }; + tracing::trace!(session = %s.id, state = %s.state, tmux = %t, socket = %sock, alive, exit = ?exit, "flow: supervise"); + if !alive { return self.on_death(s, None); } - if let Some(code) = tmux::pane_exit_status(&sock, t)? { + if let Some(code) = exit? { self.with_store(|st| st.set_exit_code(&s.id, Some(code)))?; tmux::kill_session(&sock, t); self.drop_pty(&s.id); @@ -1826,6 +1894,65 @@ mod tests { assert!(root == d || root.join(".git").exists()); } + /// th-8e3087: a `SessionStart` that arrives while a kill+resume is still + /// deciding the row's state must still promote it. + /// + /// The original bug was pure ordering: `hook` read the row BEFORE + /// `relaunch` wrote `Starting`, saw the pre-kill `idle`, and + /// [`HookOutcome::Started`]'s `state == Starting` test declined — the + /// resumed row then sat at `starting` until the test timed out. The kill's + /// slow half is simulated by holding the session lock (no `KILL_GRACE` + /// sleep) so this pins the ordering, not the timing. + #[test] + fn session_start_during_a_kill_resume_still_promotes_the_row() { + let tmp = tempfile::tempdir().unwrap(); + let e = engine(tmp.path()); + let s = e + .with_store(|st| { + st.create(NewSession { + kind: Some(SessionKind::Claude), + agent_session_id: Some("uuid-resume".into()), + project: tmp.path().to_string_lossy().into(), + worktree: tmp.path().to_string_lossy().into(), + ..Default::default() + }) + }) + .unwrap(); + // The row as a finished turn leaves it: idle, not starting. + e.set_state(&s.id, SessionState::Idle, None).unwrap(); + + // Stand in for `kill`'s locked tail, which ends by writing `Starting`. + let lock = e.session_lock(&s.id); + let held = lock.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + + let hooking = { + let e = e.clone(); + std::thread::spawn(move || { + e.hook(HookEvent { + harness: "claude-code".into(), + event: "SessionStart".into(), + session_id: "uuid-resume".into(), + cwd: None, + payload: json!({ "source": "resume" }), + }) + .unwrap(); + }) + }; + // Give the hook thread every chance to read the row early — which is + // exactly what it used to do. + std::thread::sleep(Duration::from_millis(150)); + assert_eq!(e.get(&s.id).unwrap().unwrap().state, SessionState::Idle); + e.set_state(&s.id, SessionState::Starting, None).unwrap(); + drop(held); + + hooking.join().unwrap(); + assert_eq!( + e.get(&s.id).unwrap().unwrap().state, + SessionState::Idle, + "SessionStart must promote the row relaunch just marked `starting`" + ); + } + #[test] fn hook_for_unknown_session_is_a_quiet_ok() { let tmp = tempfile::tempdir().unwrap(); @@ -1865,9 +1992,23 @@ mod tests { cwd: None, payload, }; + // th-8e3087: SessionStart on a starting row ⇒ idle, not unread — a + // resumed / prompt-less harness is otherwise `starting` for good. + assert_eq!(s.state, SessionState::Starting); + e.hook(ev("SessionStart", json!({"source":"resume"}))).unwrap(); + let up = e.get(&s.id).unwrap().unwrap(); + assert_eq!(up.state, SessionState::Idle); + assert!(!up.unread, "nothing happened yet"); + assert_eq!(up.state_source, "hooks"); + while rx.try_recv().is_ok() {} + e.hook(ev("UserPromptSubmit", json!({}))).unwrap(); assert_eq!(e.get(&s.id).unwrap().unwrap().state, SessionState::Working); assert!(matches!(rx.try_recv().unwrap(), ServerFrame::Session { .. })); + // …and a SessionStart mid-turn changes nothing. + e.hook(ev("SessionStart", json!({}))).unwrap(); + assert_eq!(e.get(&s.id).unwrap().unwrap().state, SessionState::Working); + while rx.try_recv().is_ok() {} e.hook(ev("Stop", json!({}))).unwrap(); let after = e.get(&s.id).unwrap().unwrap(); @@ -2190,11 +2331,21 @@ mod tests { e.with_store(|st| st.set_process(&holder.id, Some("x"), Some(me), proc::start_time(me), &["claude".into()])) .unwrap(); let victim = mk(); + // The victim's own pane is gone (that is why it is being resumed). + e.with_store(|st| st.set_process(&victim.id, Some("gone-pane"), Some(4_000_000), None, &["claude".into()])) + .unwrap(); let out = e.relaunch(&victim).unwrap(); assert_eq!(out.state, SessionState::NeedsYou); - let att = out.attention.unwrap(); + let att = out.attention.clone().unwrap(); assert_eq!(att.reason, "held"); assert!(att.detail.unwrap().contains(&me.to_string())); + // th-8e3087: the supervisor leaves a held row alone — its dead tmux + // session is not an unexpected death to schedule a resume for. + e.supervise_tick().unwrap(); + let still = e.get(&victim.id).unwrap().unwrap(); + assert_eq!(still.state, SessionState::NeedsYou, "{still:?}"); + assert_eq!(still.attention.as_ref().map(|a| a.reason.as_str()), Some("held")); + assert!(!e.rt().relaunch_at.contains_key(&victim.id), "no resume scheduled for a held row"); } #[test] @@ -2535,6 +2686,8 @@ mod tests { // A user manifest under the engine's home shows up with its origin. let dir = tmp.path().join("home/.smooth/harnesses"); std::fs::create_dir_all(&dir).unwrap(); + // A binary name nothing installs, so the assertion holds on a + // machine that happens to have the real tool (th-8e3087 hit `aider`). std::fs::write( dir.join("nosuchtool.toml"), "name=\"nosuchtool\"\n[binary]\nnames=[\"nosuchtool-xyzzy\"]\n[launch]\nargv=[\"{prompt}\"]\n", diff --git a/crates/smooth-flow/src/protocol.rs b/crates/smooth-flow/src/protocol.rs index 9a1615a50..95bbfc166 100644 --- a/crates/smooth-flow/src/protocol.rs +++ b/crates/smooth-flow/src/protocol.rs @@ -355,7 +355,13 @@ pub enum HookOutcome { NeedsYou(Attention), /// The harness session ended (`SessionEnd`); exit is decided by the PTY. Ended, - /// Nothing to do (SessionStart, PreCompact, SubagentStop, unknown). + /// The harness is up (`SessionStart`): a `starting` row becomes `idle` + /// (not unread — nothing happened yet); any other state is untouched. + /// Without this a resumed or prompt-less harness sat in `starting` for + /// good — its first hook switched the source to `hooks`, which stops + /// the pane scraper, and nothing else ever said idle (th-8e3087). + Started, + /// Nothing to do (PreCompact, SubagentStop, unknown). None, } @@ -402,6 +408,7 @@ pub fn map_hook_event(event: &str, payload: &Value) -> HookOutcome { } } "SessionEnd" => HookOutcome::Ended, + "SessionStart" => HookOutcome::Started, _ => HookOutcome::None, } } @@ -447,7 +454,7 @@ pub const fn outcome_state(outcome: &HookOutcome) -> Option { HookOutcome::Working => Some(SessionState::Working), HookOutcome::Idle => Some(SessionState::Idle), HookOutcome::NeedsYou(_) => Some(SessionState::NeedsYou), - HookOutcome::Ended | HookOutcome::None => None, + HookOutcome::Ended | HookOutcome::Started | HookOutcome::None => None, } } @@ -825,7 +832,12 @@ mod tests { assert_eq!(map_hook_event("PostToolUse", &empty), HookOutcome::Working); assert_eq!(map_hook_event("Stop", &empty), HookOutcome::Idle); assert_eq!(map_hook_event("SessionEnd", &empty), HookOutcome::Ended); - assert_eq!(map_hook_event("SessionStart", &empty), HookOutcome::None); + assert_eq!( + map_hook_event("SessionStart", &empty), + HookOutcome::Started, + "th-8e3087: up ⇒ idle when starting" + ); + assert_eq!(outcome_state(&HookOutcome::Started), None, "the engine decides from the current state"); assert_eq!(map_hook_event("PreCompact", &empty), HookOutcome::None); assert_eq!(map_hook_event("SubagentStop", &empty), HookOutcome::None); assert_eq!(map_hook_event("Whatever", &empty), HookOutcome::None); diff --git a/crates/smooth-flow/src/tmux.rs b/crates/smooth-flow/src/tmux.rs index 3d56473d8..ea50e02b3 100644 --- a/crates/smooth-flow/src/tmux.rs +++ b/crates/smooth-flow/src/tmux.rs @@ -214,13 +214,26 @@ pub fn pane_pid(socket: &str, session: &str) -> Result { /// # Errors /// When the session is gone or tmux fails. pub fn pane_exit_status(socket: &str, session: &str) -> Result> { - let s = tmux_ok(socket, &["display-message", "-p", "-t", session, "#{pane_dead}\t#{pane_dead_status}"])?; - let mut parts = s.split('\t'); + let s = tmux_ok(socket, &["display-message", "-p", "-t", session, "#{pane_dead}|#{pane_dead_status}"])?; + tracing::trace!(socket, session, raw = ?s, "tmux: pane_dead query"); + Ok(parse_pane_dead(&s)) +} + +/// Parse `#{pane_dead}|#{pane_dead_status}`: `None` while the pane runs, +/// `Some(status)` once it died (`-1` when tmux has no status for it). +/// +/// The separator is `|`, NOT a tab: under a non-UTF-8 locale (no `LANG` — +/// a launchd-started daemon, a CI runner, an `env -i`) tmux rewrites every +/// control character in `display-message -p` output to `_`, so a tab-joined +/// format read as `1_2` and the engine never saw a pane die (th-8e3087). +#[must_use] +pub fn parse_pane_dead(raw: &str) -> Option { + let mut parts = raw.split('|'); let dead = parts.next().unwrap_or("0").trim() == "1"; if !dead { - return Ok(None); + return None; } - Ok(Some(parts.next().unwrap_or("").trim().parse::().unwrap_or(-1))) + Some(parts.next().unwrap_or("").trim().parse::().unwrap_or(-1)) } /// `(cols, rows)` of the pane. @@ -228,11 +241,18 @@ pub fn pane_exit_status(socket: &str, session: &str) -> Result> { /// # Errors /// When the session is gone or tmux fails. pub fn pane_size(socket: &str, session: &str) -> Result<(u16, u16)> { - let s = tmux_ok(socket, &["display-message", "-p", "-t", session, "#{pane_width}\t#{pane_height}"])?; - let mut parts = s.split('\t'); + let s = tmux_ok(socket, &["display-message", "-p", "-t", session, "#{pane_width}|#{pane_height}"])?; + Ok(parse_pane_size(&s)) +} + +/// Parse `#{pane_width}|#{pane_height}` (80×24 when a half is unreadable). +/// `|`-joined for the same locale reason as [`parse_pane_dead`]. +#[must_use] +pub fn parse_pane_size(raw: &str) -> (u16, u16) { + let mut parts = raw.split('|'); let cols = parts.next().unwrap_or("80").trim().parse().unwrap_or(80); let rows = parts.next().unwrap_or("24").trim().parse().unwrap_or(24); - Ok((cols, rows)) + (cols, rows) } /// Plain-text capture of the visible pane. @@ -336,6 +356,23 @@ mod tests { ); } + /// th-8e3087: tmux under a C locale turns a tab into `_` — the joined + /// format must survive that, and the parsers must read what tmux prints. + #[test] + fn pane_queries_parse_without_a_tab_separator() { + assert_eq!(parse_pane_dead("0|"), None); + assert_eq!(parse_pane_dead("0|0"), None); + assert_eq!(parse_pane_dead("1|2"), Some(2)); + assert_eq!(parse_pane_dead("1|0"), Some(0)); + assert_eq!(parse_pane_dead("1|"), Some(-1)); + assert_eq!(parse_pane_dead(""), None); + // What a tab-joined format came back as under `LANG` unset. + assert_eq!(parse_pane_dead("1_2"), None, "the old format read as alive — the bug"); + assert_eq!(parse_pane_size("120|40"), (120, 40)); + assert_eq!(parse_pane_size("garbage"), (80, 24)); + assert_eq!(parse_pane_size("100|"), (100, 24)); + } + #[test] fn attach_argv_targets_the_flow_socket() { let a = attach_argv("smoothflow", "fs-1"); diff --git a/docs/Architecture/SmoothFlow.md b/docs/Architecture/SmoothFlow.md index 9caa0ab46..9f4680dc0 100644 --- a/docs/Architecture/SmoothFlow.md +++ b/docs/Architecture/SmoothFlow.md @@ -436,3 +436,11 @@ nonce layout, tamper + replay rejection) and regenerate/verify the shared fixture (`SMOOTH_E2E_WRITE_FIXTURE=1` rewrites it). All live tests name a private tmux socket per call (`tmux_socket` on the request) so they never touch a running daemon's sessions. + +End to end (th-8e3087): `crates/smooth-daemon/tests/flow_e2e` boots a REAL +`smooth-daemon` per test — its own HOME, port, tmux server — and drives it the +way the apps, `th flow` and a harness's hook script do, with `fake-agent` +installed through a manifest in the four state-source flavours. The full +strategy (what runs where, the fake-agent contract, runtimes, the CI split) is +[SmoothFlow-Testing.md](../Engineering/SmoothFlow-Testing.md); the macOS UI +lane is [SmoothFlow-Testing-macOS.md](../Engineering/SmoothFlow-Testing-macOS.md). diff --git a/docs/Engineering/SmoothFlow-Testing.md b/docs/Engineering/SmoothFlow-Testing.md new file mode 100644 index 000000000..a391910b4 --- /dev/null +++ b/docs/Engineering/SmoothFlow-Testing.md @@ -0,0 +1,195 @@ +# SmoothFlow — testing strategy + +Pearl th-8e3087 (epic th-6ac036). Brent: "can we write similar tests to +orca / cmux". This page is the map: what each layer proves, where it runs, +how long it takes, and how to add to it. The engine itself is +[Architecture/SmoothFlow.md](../Architecture/SmoothFlow.md); the macOS UI +lane has its own page, [SmoothFlow-Testing-macOS.md](SmoothFlow-Testing-macOS.md). + +## What cmux and orca do, and what we took + +| Project | Their e2e | Ours | +| -------- | ------------------------------------------------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| **orca** | a "golden stub agent" — a scripted binary on PATH that behaves like the real agent where the orchestrator can tell | `tests/fixtures/fake-agent`: launched through a **harness manifest** like any coding CLI; honours `--session-id` / `--resume`, posts hook events, long-polls a permission, prints the usage-limit banner, exits with a code — in four state-source flavours | +| **orca** | the orchestrator booted for real, driven over its API, state asserted by polling | a REAL `smooth-daemon` per test (isolated HOME / port / tmux server), driven over the flow WS, the HTTP siblings and the `th` binary; every wait polls with a timeout, never sleeps-and-hopes | +| **cmux** | XCUITests over the built app against a mock server and against a real backend | `apps/smoothflow` XCUITests (th-a58a97) against `mock/server.mjs` and `flow_e2e_server` + `fake-claude` — see the macOS page | +| **cmux** | terminal-level assertions: what the pane shows | `flow.snapshot` / `th flow snapshot` on the pane after every step; `flow.output` bytes on attach | + +## The layers + +| Layer | Where | Boots | Proves | Runtime | +| -------------- | ------------------------------------------------- | ------------------------------------------- | -------------------------------------------------------------------------------- | ------------------------- | +| engine unit | `crates/smooth-flow/src/*` (`#[cfg(test)]`) | nothing (a live shell when tmux is present) | store, frames, hook table, reset parser, guard, backoff, manifests, scrape rules | ~3 s | +| route unit | `crates/smooth-daemon/src/flow_route.rs` | the axum router in-process | auth gate, WS hello/errors, hook long-poll over a socket, harness prefs routes | ~2 s | +| **engine e2e** | `crates/smooth-daemon/tests/flow_e2e/` | **a real `smooth-daemon` per test** | everything below | **~70 s wall** (24 tests) | +| macOS UI | `apps/smoothflow/UITests` | the built app + `flow_e2e_server` | the shell renders and drives the engine's states | ~10 min (mac runner) | +| phones | `apps/smoothflow-mobile/{ios,android}` unit tests | nothing | frame codec + reducer | seconds | + +## The engine e2e suite + +`cargo nextest run -p smooai-smooth-daemon --test flow_e2e` (nextest runs +each test in its own process, so each has its own daemon). Skips — or fails, +with `SMOOTH_E2E_STRICT=1`, which CI sets — when `tmux`, `bash`, `curl` or the +`th` binary is missing. `#![cfg(unix)]`: Windows compiles it empty. + +| File | Test | Asserts | +| -------------- | ------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `shell.rs` | `shell_lifecycle_new_attach_input_resize_snapshot_kill` | `flow.new` → idle row (branch, socket, pid); attach streams only to attached clients; typed bytes echo twice; snapshot text + size follow attach and `flow.resize`; a second client; `flow.kill` → done, tmux session gone; input to a dead session is a `flow.error` with the client's `seq` | +| | `shell_that_exits_is_done_and_a_failing_command_is_dead` | `exit 0` → `done` with exit_code 0 (rule 5); `sh -c 'exit 7'` → `dead · crashed: exit 7`, never resumed; the list is newest-first; nothing streams for an unattached session | +| `agent.rs` | `agent_transitions_working_idle_needs_you_done` | pre-assigned uuid in argv, binary resolved through `prefer_paths`; the prompt's turn → unread idle via hooks with user/tool/system/agent `flow.event` lines; `mark_read`; `/perm` → `needs_you · permission` with a request_id, the hook held open; `flow.approve allow_session` → the agent prints the long-polled reply with the session rule; `/exit 0` → `done`, not resumed | +| | `agent_question_notification_needs_you_and_steer_answers_it` | a question notification → `needs_you · question` (no request_id); a steer answers it | +| | `agent_usage_limit_is_scheduled_from_the_banner` | the banner → `limited` with `resume_at` at the next 11:59pm, source still `hooks`; the state holds | +| | `agent_usage_limit_resume_fires_when_the_window_passes` (slow) | a banner ~1 min out → the supervisor presses Enter when it passes → `working`; the Enter reached the agent | +| | `agent_that_dies_is_resumed_with_its_session_id` | exit 2 → `starting · crashed: exit 2; resuming in 5s (attempt 1/3)`; relaunched with `--resume ` (argv, pane, same `agent_session_id`, new pid); the story in the event stream; hooks re-attach after the resume | +| | `agent_that_keeps_crashing_is_dead_after_three_resumes` (slow) | 5 + 10 + 20 s backoff, four launches (three with `--resume`), then `dead · gave up after 3 resumes`; no fourth | +| | `learned_session_id_binds_from_the_first_hook_and_kill_resume…` | opencode/codex shape: no `--session-id`, the first hook from the worktree binds the id; `kill --resume` relaunches with `--resume ` | +| | `duplicate_resume_guard_holds_when_a_live_pid_owns_the_session` | rule 4: with a live process owning the same harness session, a resume is `needs_you · held` naming the pid, the supervisor leaves it alone, and it resumes once the holder is gone | +| | `native_harness_reports_its_own_turns` | th code shape: `turn_start` / `turn_end` through `event_map` → `native`; `ask` → needs_you answered by the keystroke path; `bye` + exit | +| | `scrape_harness_state_is_inferred_from_the_pane` | no hooks at all: working/idle/approval/limit scraped, source stays `inferred`; a scraped approval gets a `scrape-` request id and the menu key; `relaunch_command` reruns the original argv | +| `hooks.rs` | `hooks_contract_per_event` | every event Claude Code posts — SessionStart, UserPromptSubmit, PreToolUse, PostToolUse, Stop, Notification (permission / question / other), PreCompact, SubagentStop, SessionEnd, unknown — with the state and the `flow.event` line each produces; unknown session → quiet `{}` | +| | `permission_request_long_polls_until_approved_each_decision_shape` | deny / allow / allow_session reply bodies; a Notification keeps the request_id; a mismatched request_id does not resolve the poll | +| | `hooks_are_unauthenticated_and_everything_else_is_gated` | 401 without the token on every route but hooks; `?token=`, Bearer, `X-Smooth-Token`; the WS handshake refuses a bad token | +| `cli.rs` | `th_flow_json_against_the_live_daemon` | `ls` / `new` / `snapshot` / `send` / `inbox` / `approve` / `handoff` / `kill` `--json` shapes and the human lines; the two-line error contract; explicit argv after `--` | +| | `th_without_a_daemon_says_so_in_two_lines` | no `daemon.addr` → the hint; `th harness list` degrades to the files with a note | +| `harnesses.rs` | `harness_matrix_state_source_per_manifest` | the built-ins are listed with their source + install reason; each fake-agent flavour ends a turn with the expected `state_source` (hooks / hooks / native / inferred); an unknown kind is refused with the pointer | +| | `harness_prefs_sort_and_hide_reach_every_picker` | `PUT /api/flow/harnesses/prefs` → `flow.harnesses`, a fresh hello omits hidden, `th harness list` follows; `hide` / `unhide` / `order` verbs; unknown names refused; prefs persist | +| | `th_harness_add_installs_a_custom_manifest_the_engine_launches` | `th harness add ` validates and copies; refuses to clobber; an invalid manifest is refused; no daemon restart — the daemon lists it, `show` reads it, a session launches on it | +| | `real_harnesses_launch_through_their_manifests` (opt-in) | `SMOOTH_E2E_REAL_HARNESSES=1`: claude / opencode / codex where installed + `th code`, launched through their built-in manifests; argv[0] is the resolved binary, the launch shape per manifest, th code reports `native` | +| `isolation.rs` | `suite_never_writes_the_real_daemon_addr_or_token` | the user's `~/.smooth/{daemon.addr,daemon.lock,operator-token,flow.db}` are byte- and mtime-identical after a full boot + session; the rig's session is not on the default `smooth-flow` server | +| | `daemon_teardown_leaves_no_tmux_server_or_process` | dropping the rig kills the daemon, its tmux server and the agent's process | +| | `flow_e2e_server_example_hosts_the_same_engine` | the macOS lane's host runs the same manifests + fake-agent to the same states | + +Measured on a loaded dev machine (load 25–40, other agents building): +**~70 s wall, 24 tests**, two of them "slow" (60 s and 66 s — the backoff and +usage-limit windows are real time, the engine has no test clock). CI runs +the whole suite on every PR; if it ever passes ~15 min on the runner, split +the two slow ones to a nightly `schedule:` — they are the only tests over +25 s. + +### The rig (`support.rs`) + +`Daemon::boot()` spawns `smooth-daemon operator --addr 127.0.0.1:0 +--tmux-socket flow-e2e--` with a scrubbed environment: + +| env | why | +| ------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------- | +| `HOME=/home` | `~/.smooth/{daemon.addr,operator-token,flow.db}` and `~/.smooth/harnesses/` are throwaway; `th` run with the same HOME finds THIS daemon | +| `SMOOTH_ALLOW_SECOND_DAEMON=1` | skip the machine-wide single-instance lock | +| `SMOOTH_TAILSCALE_SERVE=0` | never re-point the tailnet `:443` at a test daemon (it happened once) | +| `SMOOTH_RELAY=0` | no relay dial-out | +| `SMOOTH_WORKSPACE=/ws` | a git repo (`main`, one commit) sessions run in | +| `PATH=/.local/bin:$PATH` | `fake-agent` lives there; the manifests' `prefer_paths` resolve it under any PATH | +| no `LANG`, no `SHELL` | on purpose — a launchd-started daemon has neither. This is how th-8e3087 found the tmux tab/locale bug below | + +Port 0 is real: the daemon resolves the ephemeral port BEFORE the router +renders `{daemon_url}` (a bug this suite caught — hooks went to +`http://127.0.0.1:0`). A second instance deliberately does NOT advertise +itself (#546 — SmoothFlow's own daemon must never repoint `th`), so the rig +reads the bound port off the daemon's `operator listening … addr=` log line +and writes `daemon.addr` under the test HOME itself; that is what `th flow` +and `th harness` read. `operator-token` is the `X-Smooth-Token`. + +The opt-in real-harness test copies your `~/.smooth/providers.json` into the +rig's HOME (th code refuses to boot without providers) and skips th code when +there is none; claude / opencode / codex sit on onboarding in a fresh HOME, so +only "painted + pid live + the launch shape" is asserted for them. + +Waits: `wait_state` / `wait_until` / `wait_screen` poll every 250 ms with a +30 s cap (`WAIT`) and fail with the row, the pane, fake-agent's log and the +daemon log tail. `Ws::wait_for` drains frames until a predicate matches. +Dropping the rig SIGTERMs the daemon, kills its tmux server and prints the +daemon log when the test is panicking. + +### fake-agent + +`crates/smooth-daemon/tests/fixtures/fake-agent` (bash; needs `curl`). The +contract is in its header; the short form: + +| Input | Behaviour | +| ---------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `--session-id ` / `--resume ` / `--model ` | as Claude Code; the first positional is a PROMPT of `;`-separated commands | +| `$SMOOTH_URL` (the manifest's `{daemon_url}`) | where hooks go; else `./.flow-e2e-addr`, else `$HOME/.smooth/daemon.addr` | +| `$FAKE_AGENT_MODE=hooks\|native\|scrape` | Claude Code events / th code events (`turn_start`, `turn_end`, `ask`, `bye`) / nothing posted, the pane is painted for the scraper | +| `./.fake-agent-session-id` | learned mode: the id to report (else `$FAKE_AGENT_SESSION_ID`, else fresh) | +| `./.fake-agent-script` | commands run on EVERY start, fresh and resumed — how a test makes each relaunch crash | +| `/work ` `/perm` `/ask ` `/limit [time]` `/exit [code]` `/crash [code]` `/sleep n` | a turn; a permission (long-polled, decision printed); a question; the banner; SessionEnd + exit; exit with no hook; wait. Anything else is `echo: ` | +| `./.fake-agent.log` | argv, every hook and its reply, every command — the assertions read it | + +The four manifests in `tests/fixtures/harnesses/` are what a real manifest +looks like for each `[state] source` and `[launch] session_id` mode +(`fake-agent` = Claude shape, `-learned` = opencode/codex shape, `-native` = +th code shape, `-scrape` = a tool with no hooks). Copy one when you add a +harness — and run its flavour of the matrix test against it. + +`fake-claude` (the macOS lane's stub, same directory) predates fake-agent and +is a strict subset of it; the mac lane keeps its own until that lane is +re-pointed (pearl filed). + +## Bugs this suite found on day one + +Recorded here because each is a class, not a one-off: + +1. **tmux rewrites control characters under a non-UTF-8 locale.** The engine + joined `#{pane_dead}\t#{pane_dead_status}` with a tab; with no `LANG` + (launchd, CI, `env -i`) tmux printed `1_2`, the parser read "alive", and + **no dead pane was ever detected** — the whole supervision story (resume, + backoff, done/dead) was inert for any daemon not started from a shell. + Fixed: `|`-joined formats + pure parsers (`tmux::parse_pane_dead`, + `parse_pane_size`). +2. **`{daemon_url}` rendered from the requested address.** `--addr +127.0.0.1:0` gave every pane `SMOOTH_URL=http://127.0.0.1:0`. Fixed: + `resolve_ephemeral_port` before the server is built. +3. **A resumed or prompt-less harness sat in `starting` for good.** Its + `SessionStart` flipped the source to `hooks` (which stops the scraper) but + mapped to nothing. Fixed: `SessionStart` on a `starting` row ⇒ `idle`. +4. **A `held` row flapped.** After the duplicate-resume guard parked a row, + the next tick saw its dead tmux session and scheduled a resume; the guard + refused it again; forever. Fixed: the supervisor skips held rows. + +## Running locally + +```sh +brew install tmux # bash + curl ship with macOS +CARGO_TARGET_DIR=$HOME/.cargo/target-e2e \ + cargo nextest run -p smooai-smooth-daemon --test flow_e2e # ~70 s +# one test, with the daemon log on failure: +CARGO_TARGET_DIR=$HOME/.cargo/target-e2e \ + cargo nextest run -p smooai-smooth-daemon --test flow_e2e agent_that_dies --no-capture +# the real CLIs installed here, through their manifests: +SMOOTH_E2E_REAL_HARNESSES=1 cargo nextest run -p smooai-smooth-daemon --test flow_e2e real_harnesses +``` + +`th` is found next to the daemon binary (a workspace build puts both in +`target/debug/`), or via `SMOOTH_TH_BIN`. Never `pnpm install:th` for this — +the suite must not depend on, or replace, the `th` on your PATH. The rig +never reads or writes your `~/.smooth`; `isolation.rs` proves it every run. + +## CI + +`pr-checks.yml`, the `rust` job, Linux leg: `tmux` is installed with the +other system packages and `SMOOTH_E2E_STRICT=1` makes a skip a failure. The +suite runs inside the normal `cargo nextest run --profile ci`; nextest builds +`th` and `smooth-daemon` first, and the `cargo build --examples` step builds +`flow_e2e_server` for the example test. Windows compiles the crate empty. + +## Adding a scenario + +1. Boot a rig: `let d = Daemon::boot().await;` after `if !prereqs() { return; }` + (`prereqs_with_th()` when the test runs `th`). +2. Drive it the way a client would — `d.ws()` for frames, `d.post` / + `d.get` for the HTTP siblings, `d.th(&[…])` for the CLI, `d.hook(…)` for + what a hook script posts. Steer fake-agent with `d.send(id, "/work x")`. +3. Assert with `wait_state` / `wait_until` / `wait_screen`; read + `d.agent_log()` for what the agent saw. Never `sleep` and check. +4. A new harness shape = a new manifest in `tests/fixtures/harnesses/` and a + row in `harness_matrix_state_source_per_manifest`. + +## Gaps (pearls) + +- The two slow tests wait real backoff/limit windows; an engine test clock + would cut ~2 min of CI to seconds. +- Fan-out (`flow.fanout.new` / `pick`) is unit-tested only; an e2e needs + `th pearls` in the rig's HOME. +- The real-harness launches are opt-in and creds-free: a nightly on a + runner with Claude Code / opencode / codex installed would make them a gate. +- `fake-claude` → `fake-agent` in the macOS lane.