diff --git a/docs/ai/design/2026-10-17-feature-devkitd-agent-start.md b/docs/ai/design/2026-10-17-feature-devkitd-agent-start.md new file mode 100644 index 000000000..4c653ec24 --- /dev/null +++ b/docs/ai/design/2026-10-17-feature-devkitd-agent-start.md @@ -0,0 +1,289 @@ +--- +phase: design +title: devkitd-agent-start design +description: agent.start RPC on devkitd — Rust tmux/herdr runtimes, harness start profiles, merged registry columns in daemon.db agents table, dual-write bridge to legacy agents.db, CLI daemon-first routing with TS fallback +--- + +# Design — devkitd-agent-start + +## Architecture overview + +```mermaid +graph TD + subgraph "TS (unchanged callers)" + CLI[agent start cmd] -->|daemon reachable| DC[daemon-client
DaemonClient.startAgent] + CLI -->|connect fail /
unknown method| TS[ManagedAgentRuntime.startAgent
legacy fallback] + TUI[console StartAgentPane] -->|spawns CLI| CLI + end + subgraph "rust/" + D[crates/devkitd
server.rs dispatch] --> AS[rpc/agent/start.rs
orchestration + mutex] + AS --> RT[runtime/mod.rs
trait AgentRuntime] + RT --> TM[runtime/tmux.rs] + RT --> HD[runtime/herdr.rs] + AS --> SP[devkit-harness
start_profiles.rs] + AS --> ST[devkit-core store.rs + store/registry.rs
agents table + registry cols] + AS --> EV[emit agent.started] + SW[sweep loop] --> ST + end + DC -.->|"unix socket ~/.ai-devkit/daemon.sock"| D + AS -->|dual-write| LDB[~/.ai-devkit/agents.db
legacy registry] + ST -->|boot delta import| LDB +``` + +`agent.start` runs on the blocking pool (not `INLINE_METHODS`) — it can take +up to `timeoutMs` plus herdr retries. + +## Wire contract + +```json +→ {"id":1,"method":"agent.start","params":{ + "type":"claude", // required: StartableAgentType + "name":"my-agent", // optional: server generates {basename}-{ts36} + "cwd":"/repo", // required + "runtime":"tmux", // optional: "tmux"|"herdr", default "tmux" + "args":["--flag"], // optional: appended after "--" (herdr) + "timeoutMs":15000, // optional: pid-resolution deadline + "pollIntervalMs":500 // optional + }} + +← {"id":1,"result":{ + "name":"my-agent","type":"claude","pid":12345, + "runtime":"tmux","runtimeRef":{"session":"my-agent"}, + "cwd":"/repo","startedAt":"2026-10-17T07:44:00.000Z", + "warnings":["tmux session \"x\" already exists ... will be replaced."]}} + +← {"id":1,"error":"Agent \"x\" is already running (PID 42).","code":"name_in_use", + "details":{"pid":42}} +``` + +`Response` gains two optional fields (both skipped when absent — old clients +unaffected): + +```rust +pub code: Option, +pub details: Option, // pid, command, timeoutMs, provider, reason... +``` + +New ts-rs-exported types in `proto.rs`: `AgentStartParams`, +`AgentStartResult`. `runtimeRef` crosses as `serde_json::Value`/`unknown`. + +### Error codes + +| code | details | CLI rendering | +|---|---|---| +| `invalid_params` | `field`, `message` | same text as current validation | +| `harness_not_startable` | `type` | unsupported-type error | +| `name_in_use` | `pid` | `Agent "n" is already running (PID p)` | +| `runtime_unavailable` | `provider`, `reason`, `detail` | tmux install instructions / herdr message | +| `pid_poll_timeout` | `command`, `timeoutMs` | pid-poll timeout message | +| `internal` | — | generic throw | + +## daemon.db schema — merged `agents` table + +Extend the existing snapshot table (no new table; `Store::open` migration +pattern via `PRAGMA table_info`, matching the `start_time_ms` precedent): + +```sql +ALTER TABLE agents ADD COLUMN name TEXT; +ALTER TABLE agents ADD COLUMN agent_type TEXT; +ALTER TABLE agents ADD COLUMN runtime TEXT; +ALTER TABLE agents ADD COLUMN runtime_ref TEXT NOT NULL DEFAULT ''; +ALTER TABLE agents ADD COLUMN pinned INTEGER NOT NULL DEFAULT 0; +ALTER TABLE agents ADD COLUMN registered_at INTEGER; +CREATE UNIQUE INDEX idx_agents_name ON agents(name) WHERE name IS NOT NULL; +``` + +`session_file` (existing, sweep-filled) serves `sessionFilePath`. Declared +`agent_type` (what the user passed) is stored separately from the +adapter-derived type to keep them honest when they disagree. + +### Snapshot rules (the load-bearing change) + +`apply_agent_snapshot` currently keys identity on `(pid, start_time_ms)` and +deletes every row whose identity vanishes. With registry columns: + +1. **Registered-before-swept**: register inserts a row with + `start_time_ms = NULL` before the 2s sweep sees the process. When the + sweep delivers `(pid, real_st)` for a stored `(pid, NULL)` row → the + ON CONFLICT upsert adopts `start_time_ms`; the row is *not* in `gone` + (see below) and `appeared` fires normally. +2. **Pid reuse**: stored `start_time_ms` non-NULL and differs → DELETE the + row entirely (name, pin, runtime_ref) then re-insert. A recycled pid + never inherits a dead agent's identity. +3. **True disappearance**: pid absent from the sweep — + `name IS NULL` → DELETE (unchanged); + `name IS NOT NULL` → tombstone: keep the row (stale `last_seen`, retained + `runtime_ref`) so `kill`/`send` on a dead agent still resolves and an + orphaned pane stays closable by name. Freed by prune at next + `agent.start` name check or future `agent.stop`. + +`gone`/`appeared` computation changes: `gone` yields a pid only when the +stored `start_time_ms` is non-NULL (NULL-st rows were never live). Tombstone +rows count as `known` only for `gone` purposes — a named row's pid +reappearing with a *new* `start_time_ms` is identity-change (rule 2), not +adoption. + +### Registry write path + +`register_start` (single transaction): + +```sql +INSERT INTO agents (pid, name, agent_type, runtime, runtime_ref, + registered_at, first_seen, last_seen) +VALUES (?,?,?,?,?,?,?,?) +ON CONFLICT(pid) DO UPDATE SET + name=excluded.name, agent_type=excluded.agent_type, + runtime=excluded.runtime, runtime_ref=excluded.runtime_ref, + registered_at=excluded.registered_at, last_seen=excluded.last_seen +-- start_time_ms untouched: NULL until first sweep adopts real value +``` + +Name uniqueness is enforced by `idx_agents_name` *and* pre-checked under a +`start_mutex` on `Daemon` — the mutex also serializes concurrent starts so +two racing `agent.start` calls can't both pass the name check. + +Prune semantics move from "row exists" to "live": `name_in_use` fires only +when the named row's pid is present in the last sweep snapshot (live). Dead +named rows are reclaimed during the check — matching TS `prune()` behavior +but liveness-aware. + +## Rust components + +### `devkit-harness::start_profiles` + +```rust +pub struct StartProfile { + pub command: &'static str, // e.g. "kiro-cli" for kiro + pub runtime_kind: &'static str, // passed to herdr --kind + pub matcher: MatchSpec, +} + +pub enum MatchSpec { + Argv0(&'static str), // matchArgv0 + Argv0Contains(&'static str), // matchArgv0Name + AnyBasename(&'static [&'static str]), // matchAnyBasename +} +``` + +Eleven entries mirroring `HARNESS_RUNTIME_PROFILES`. `MatchSpec::matches` +ports `tokensOf`/`basenameOf`/`mayContinueArgv0` **and `executablePath`'s +filesystem probing** (`isFile` on progressive prefixes — handles absolute +paths under space-containing dirs). TS unit-test cases for the matchers are +ported verbatim. + +### `devkitd::runtime` module + +```rust +pub trait AgentRuntime { + fn is_available(&self) -> Availability; // {ok, reason, detail} + fn start(&self, input: StartInput) -> Result; + // {pid, runtime_ref: Value} +} +``` + +- **`TmuxRuntime`** — ports `TmuxManager`: `tmux -V` availability, + `has-session`, `new-session -d -s -c `, + `send-keys -t Enter`, `list-panes -t -F + #{pane_pid}`, `kill-session`; descendant scan via `pgrep -P` BFS + + `ps -o command= -p `; `poll_for_pid` with interval/timeout and the + **5-consecutive-identical-pid stability rule** (`REQUIRED_STABLE_PID_POLLS`). + On timeout: kill session, `pid_poll_timeout`. +- **`HerdrRuntime`** — ports `HerdrCliClient` + `HerdrResponseParsers`: + `herdr --version` and `herdr api snapshot` availability (`binary-missing` + vs `backend-unreachable`); `workspace create --cwd --label ` → + `result.workspace.workspace_id` + `result.root_pane.pane_id`; `agent start + --kind --pane --timeout ` with + `agent_pane_busy` retry (5s budget, 100ms interval); `pane process-info + --pane ` poll (3s, 100ms) → first `foreground_processes[].pid` else + `shell_pid`; runtime_ref `{session:"default", workspaceId?, tabId?, paneId, + agentName}`. JSON `error` blocks map to `HerdrRuntimeError(code)` → + `runtime_unavailable`/`internal`. +- All external commands via `std::process::Command` with injected argv-0 / + `PATH` for tests (same seams as TS's injectable `runner`/`tmux` deps). + +### `devkitd::agent_start` handler + +`validate → resolve profile → availability → [start_mutex] prune + +name-check → orphan-session warn → runtime.start → register (+ dual-write) +→ emit agent.started → apply_sweep() → return entry+warnings`. + +Running `apply_sweep()` right after register reconciles the snapshot, +adopts `start_time_ms`, refreshes `enriched`, and emits `agent.appeared` — +both events fire for one start (`agent.started` = declared, `.appeared` = +observed). Subscribers dedupe by pid; documented behavior. + +### Enriched overlay + +After `enricher.enrich()`, overlay registry fields for pids with `name` +non-NULL: `EnrichedAgent.name = row.name` (registered name wins — this is +the #398 fix), `pinned = row.pinned`. No new `EnrichedAgent` fields; +consumers needing `runtime_ref` use the legacy DB (dual-write) until +`agent.lookup` lands in a later feature. + +## Legacy agents.db bridge + +- **Dual-write**: after registering in daemon.db, if + `~/.ai-devkit/agents.db` exists with an `agents` table at schema v5, + upsert the same row (`type,pid,name,tmux_session,cwd,started_at, + session_id,session_file_path,updated_at,pinned,runtime,runtime_ref`). + Absent or older-schema file → skip silently. Keeps `send`/`kill`/`rename`/ + `open`/channel consumers working unchanged. +- **Delta import on boot**: `Store::open` — for every legacy row whose + `(type,pid)` isn't registered and whose pid isn't in the live snapshot, + insert a tombstone (`start_time_ms NULL`, `registered_at` from + `started_at`). Every boot, not one-shot: heals entries written by the TS + fallback path while the daemon was down. Legacy file wins nothing on + conflict — daemon row is authoritative. + +## TS changes + +- `packages/daemon-client`: `DaemonClient.startAgent(params)` → + `AgentStartResult`; errors thrown as `DaemonRpcError {code, details, + message}`; `subscribe` unaffected. `requestTimeoutMs` for `agent.start` + raised to `timeoutMs + margin` per-call. +- `packages/cli/src/commands/agent/start.command.ts`: after existing + client-side validation, `DaemonClient` via `tryConnect`/`ensureDaemon` → + `agent.start` → on `code:"unknown_method"` (old devkitd) or connect + failure → existing `startAgent` path unchanged. Result `warnings` → + `ui.warning`; error codes map to the exact `reportAgentStartError` + messages (tmux instructions resolved client-side). +- Name generation stays client-side; server-side `{basename}-{ts36}` + generation exists only for socket-direct callers omitting `name`. +- Regenerate `gen/*.ts` via `cargo test -p devkit-core` (ts-rs). + +## Alternatives considered + +- **Separate `registry` table** — rejected per user direction; merged row + gives free overlay but costs the three snapshot rules above. +- **Daemon shells out to TS helper** — rejected: npm-version coupling + + node startup per start; worst of both worlds. +- **No dual-write, migrate all consumers now** — rejected: sync + `AgentRegistry.lookup` call sites would all need async socket reads; + balloons scope. +- **Suppress `agent.appeared` for daemon-started pids** — rejected: extra + dedupe state for cosmetic purity; kinds differ so subscribers can + distinguish. + +## Security / performance + +- Socket stays 0600 same-uid (SO_PEERCRED on Linux); `agent.start` spawns + processes — same trust level as the CLI, no new privilege boundary. +- `args` are passed to `herdr` argv, never a shell; tmux `send-keys` sends + the harness `command` only (fixed per profile) — caller `args` go nowhere + near tmux. +- Start is O(pane-pid descendants) polling + ≤3 blocking calls; blocking + pool absorbs it; `start_mutex` is held across spawn — serial starts are + intentional (name uniqueness is global). +- `daemon.db` WAL + `synchronous=NORMAL` unchanged; registry columns add a + few bytes/row. + +## Resolved decisions + +- Merged `agents` table (not separate `registry`); tombstones for named + dead rows; pid-reuse deletes registration. +- `Response.code` + `details` added; `error` string preserved for old + clients. +- Runtime passed in `params.runtime` (CLI reads `ConfigManager`); daemon + defaults `tmux`. +- Both `agent.started` and `agent.appeared` fire for one start. +- Client disconnect mid-start → daemon completes (no cancel RPC in v1). diff --git a/docs/ai/implementation/2026-10-17-feature-devkitd-agent-start.md b/docs/ai/implementation/2026-10-17-feature-devkitd-agent-start.md new file mode 100644 index 000000000..1cde89cd0 --- /dev/null +++ b/docs/ai/implementation/2026-10-17-feature-devkitd-agent-start.md @@ -0,0 +1,108 @@ +--- +phase: implementation +title: Implementation — devkitd-agent-start +description: What shipped — agent.start RPC in devkitd, merged registry columns in daemon.db agents table, Rust tmux/herdr runtimes, harness start profiles, dual-write legacy bridge, daemon-first CLI routing +--- + +# Implementation — devkitd-agent-start + +## What changed + +### devkit-core (`rust/crates/devkit-core`) + +- `proto.rs` — `Response` gains `code`/`details` (both skipped when absent); + new `err_coded` constructor; `AgentStartParams`/`AgentStartResult` structs + exported via ts-rs to `packages/daemon-client/src/gen/`. +- `store.rs` + `store/registry.rs` — `agents` table extended with nullable registry columns; store.rs holds connection/schema/event log/snapshot, store/registry.rs holds the registration lifecycle + legacy agents.db bridge + (`name`, `agent_type`, `runtime`, `runtime_ref`, `pinned`, + `registered_at`, `gone_at`) + partial unique index on `name`. + `apply_agent_snapshot` rewritten around the three rules: registered- + before-swept adoption (NULL `start_time_ms`), pid-reuse full delete, + named-tombstone retention (`gone_at` dedupes `agent.disappeared`). + Upsert uses `COALESCE(start_time_ms)` so flaky start-time reads never + clear identity. New API: `register_start`, `lookup_name`, `live_name`, + `unregister`, `prune_dead`, `registry_overlays`, `set_legacy_db`, + `import_legacy_agents`, `iso8601_utc`. Legacy bridge: delta import on + boot (attach-by-pid to live rows, dead-by-default tombstones otherwise) + + dual-write on `register_start`. + +### devkit-harness (`rust/crates/devkit-harness`) + +- `start_profiles.rs` — `MatchSpec` (`Argv0` | `Argv0Contains` | + `AnyBasename`) porting the TS matcher closures including + `executablePath`'s filesystem probing for space-containing argv0; + `StartProfile` table covering all 11 `StartableAgentType`s (kiro + launches `kiro-cli`). + +### devkitd (`rust/crates/devkitd`) + +- `runtime/mod.rs` — `AgentRuntime` trait, `Availability`, `StartInput`, + `StartOutput`, `StartError`. +- `runtime/tmux.rs` — `TmuxRuntime` (tmux session mgmt, `pgrep -P` BFS, + `ps -o command=`, 5-stable-poll PID resolution, timeout → kill-session). +- `runtime/herdr.rs` — `HerdrRuntime` (`--version`/`api snapshot` + availability, workspace create, agent start with `agent_pane_busy` + retry, process-info poll → foreground/shell pid). +- `rpc/agent/start.rs` — orchestration: validate → availability → + `start_mutex` → prune/liveness name check → orphan-session warning → + spawn → register (+dual-write) → `agent.started` event → `apply_sweep`. +- `server.rs` — `Daemon.start_mutex`, `emit` → `pub(crate)`, `"agent.start"` + dispatch arm (blocking pool, not INLINE_METHODS), registry overlay in + `apply_sweep` (registered `name`/`pinned` win over adapter labels), + `Daemon::new` wires legacy-db bridge. +- `main.rs` — `mod runtime`, `mod rpc`. + +### TS (`packages/daemon-client`, `packages/cli`) + +- `client.ts` — `DaemonRpcError {code, details}`; `request()` per-call + timeout; `DaemonClient.startAgent()` (timeout = poll timeout + 15s). +- `start.command.ts` — `tryDaemonStart`: `ensureDaemon()` → + `daemon.startAgent` → coded errors render identical messages/exit + codes; `unknown method` or unreachable → existing TS `startAgent` + unchanged. Transport failure mid-request reports rather than retries + (the daemon may have completed the start). +- `agent.test.ts` — daemon-client module mock + 6 daemon-routed tests. +- `integration.test.ts` — isolated-HOME real-socket test asserting coded + errors round-trip. + +## Deviations from design + +- `gone_at` column added beyond the doc'd set — needed to stop repeated + `agent.disappeared` for tombstones and to make `live_name` meaningful. +- `start_mutex` held across spawn+register only (as designed), but the + name check uses `kill(pid,0)` liveness rather than snapshot membership — + pending rows (registered, never swept) are judged by real liveness. +- `process.exit` in the CLI error path needs an explicit `return` after + it — `process.exit` is mocked in vitest so execution would fall through + into the fallback spawn. +- `import_legacy_agents` attaches names to live rows by pid only + (documented narrow pid-reuse window) rather than a start-time match — + `started_at` ISO parsing was traded for the simpler attach. + +## Edge cases handled + +- Registered-before-swept rows (start beats the 2s sweep): adopted, not + deleted; `agent.started` + `agent.appeared` both fire (documented). +- pid reuse can't inherit a dead agent's name (identity change → full + row delete). +- Client disconnect mid-start: daemon completes; agent registered. +- Stale devkitd on socket: `unknown method` → TS fallback. +- Daemon env `PATH` may lack tmux/herdr → `runtime_unavailable` surfaces + the same install hint as today. + +## Test evidence + +- `cargo test --workspace`: 133 tests green (30 devkit-core, 80 + devkit-harness, 23 devkitd). +- `cargo clippy --workspace --all-targets -- -D warnings`: clean. +- `packages/cli` agent.test.ts: 106 tests (6 new daemon-routed). +- `packages/daemon-client` integration: real-socket coded-error + round-trip. + +## Follow-ups (not this feature) + +- `agent.stop`/`rename`/`send`/`open`/`pin` RPCs → then remove + dual-write + import. +- `agent.started` payload could carry the enriched row once consumers + need it. +- tmux PATH refresh for systemd-installed daemons. diff --git a/docs/ai/planning/2026-10-17-feature-devkitd-agent-start.md b/docs/ai/planning/2026-10-17-feature-devkitd-agent-start.md new file mode 100644 index 000000000..eb2d0be5b --- /dev/null +++ b/docs/ai/planning/2026-10-17-feature-devkitd-agent-start.md @@ -0,0 +1,123 @@ +--- +phase: planning +title: Plan — devkitd-agent-start +description: Task breakdown for agent.start RPC — proto, schema, Rust runtimes, orchestration, daemon-client, CLI routing +--- + +# Plan — devkitd-agent-start + +## Milestones + +- [x] M1: Wire contract + schema (proto code/details, agents-table registry + columns, snapshot rules) +- [x] M2: Rust runtime layer (start profiles + matchers, tmux, herdr) +- [x] M3: `agent.start` handler + registry overlay + events + bridge + (dual-write, boot import) +- [x] M4: TS client + CLI routing/fallback +- [ ] M5: Full verification + docs + +## Task Breakdown + +### M1 — wire contract + schema + +- [x] T1.1: `proto.rs` — add `code: Option` + + `details: Option` to `Response` (skip_serializing_if None); add + `AgentStartParams`/`AgentStartResult` structs with ts-rs exports; + `Response::err` variants keep `code=None` unless provided. Verify: + `cargo test -p devkit-core` regenerates `gen/*.ts`. +- [x] T1.2: `store.rs` — extend `agents` migration block: add `name`, + `agent_type`, `runtime`, `runtime_ref`, `pinned`, `registered_at` + columns (PRAGMA table_info loop like `start_time_ms`); create partial + unique index `idx_agents_name`. Verify: fresh + existing-db open both + succeed (unit test). +- [x] T1.3: `store.rs` — new snapshot rules in `apply_agent_snapshot`: + NULL-`start_time_ms` rows excluded from `gone` (adopted by upsert); + non-NULL identity change → full delete; absent pid with `name` → + tombstone keep; absent pid without name → delete. Plus + `register_start` upsert, `lookup_name`, `live_names` (names whose pids + are in last snapshot), `prune_dead`. Verify: unit tests per testing doc + (registered-before-swept, pid reuse, tombstone, unique name). +- [x] T1.4: `store.rs` — legacy bridge: `import_legacy_agents(path)` run + on `Store::open` (delta merge, tombstone insert for rows whose + (type,pid) not registered and pid not live); `dual_write_legacy(row)` + used by register when `~/.ai-devkit/agents.db` exists at schema v5. + Verify: fixture-db unit tests. + +### M2 — Rust runtime layer + +- [x] T2.1: `devkit-harness/src/start_profiles.rs` — `MatchSpec` enum + + `matches()` porting `tokensOf`/`basenameOf`/`mayContinueArgv0`/ + `executablePath` (incl. `isFile` probing); `StartProfile` table for all + 11 types; `start_profile(type) -> Option`. Verify: ported + matcher tests incl. spaced-dir tmpdir cases green. +- [x] T2.2: `devkitd/src/runtime/mod.rs` — `AgentRuntime` trait, + `Availability`, `StartInput`, `StartOutput`, `StartError`; command-runner + seam (argv0/PATH injectable). +- [x] T2.3: `runtime/tmux.rs` — `TmuxRuntime`: `-V` availability; + session_exists/create/send_keys/list_pane_pids/kill_session; + `pgrep -P` BFS + `ps -o command=`; `poll_for_pid` (interval, timeout, + 5-stable rule); timeout → kill_session + `pid_poll_timeout`. Verify: + fake-tmux script tests. +- [x] T2.4: `runtime/herdr.rs` — `HerdrRuntime`: `--version` + + `api snapshot` availability; `workspace create` parse; `agent start` + with `agent_pane_busy` retry (5s/100ms); `pane process-info` poll + (3s/100ms, foreground-first then shell_pid); runtimeRef build. Verify: + fake-herdr JSON script tests incl. busy-retry. + +### M3 — handler + registry integration + +- [x] T3.1: `devkitd/src/agent_start.rs` — orchestration: validate + (name regex `/^[a-z0-9][a-z0-9-]{0,62}[a-z0-9]$/` ported without regex + crate, cwd isdir, type→profile, runtime known), default-name + generation `{basename-sanitized}-{ts36}`, availability, `start_mutex` + → live-name check → orphan-session warning → `runtime.start` → + `register_start` + dual-write → `emit("agent.started")` → + `apply_sweep()` → result + warnings. +- [x] T3.2: `server.rs` — `Daemon.start_mutex`; `"agent.start"` arm in + `dispatch` (NOT in `INLINE_METHODS`); error→code mapping + (`name_in_use`, `runtime_unavailable`, `pid_poll_timeout`, + `harness_not_startable`, `invalid_params`, `internal`). +- [x] T3.3: Enriched overlay in `apply_sweep` — after `enrich`, rows with + `name` override `EnrichedAgent.name` and `pinned`. +- [x] T3.4: `Store::open` boot import wiring + `agent.list` returns + overlayed names. Verify: e2e socket test — start → list shows name; + subscribe sees `agent.started` + `agent.appeared`. + +### M4 — TS client + CLI + +- [x] T4.1: `daemon-client` — `DaemonClient.startAgent(params)`; + `DaemonRpcError` exposing `code`/`details`; per-request timeout + override for long-running methods. Verify: vitest socket roundtrip + (build devkitd first — `DEVKITD_BIN`). +- [x] T4.2: `cli start.command.ts` — daemon-first routing via + `tryConnect`/`ensureDaemon`; `params.runtime` from `ConfigManager`; + `warnings` → `ui.warning`; error codes → existing message texts; on + connect-fail/`unknown method` → existing `startAgent` unchanged; + `--mode durable` bypasses daemon path entirely. +- [ ] T4.3: Verify `nx test` for `daemon-client` + `cli` packages; + `agent.test.ts` untouched and green. + +### M5 — verification + docs + +- [ ] T5.1: `cargo fmt --check`, `cargo clippy -- -D warnings`, + `cargo test --workspace` green. +- [ ] T5.2: `nx run-many -t build,test,lint` green. +- [ ] T5.3: Manual smoke on macOS: real `tmux`, shim binary, `agent + start` via daemon; attach hint; `agent list` name; `agent kill` + cleanup (proves dual-write worked). +- [ ] T5.4: Implementation doc + changelog entry. + +## Sequencing notes & risks + +- T1.3 must land before T3.1 (handler depends on registry primitives). +- T2.x are parallel-safe; T2.1 gates both. +- Risk: `executablePath` filesystem probing — port semantics exactly; + spaced-dir tests exist to catch drift. +- Risk: stale-binary trap in daemon-client tests — always + `cargo build -p devkitd` before vitest (memory note: tests auto-select + `rust/target/{release,debug}`). +- Risk: `agent.appeared` + `agent.started` both firing — accepted; + subscribers dedupe by pid. +- Follow-up (not this feature): `agent.stop`/`rename`/`send`/`open` RPCs + let the legacy dual-write be removed; `Response.details` reusable for + them. diff --git a/docs/ai/requirements/2026-10-17-feature-devkitd-agent-start.md b/docs/ai/requirements/2026-10-17-feature-devkitd-agent-start.md new file mode 100644 index 000000000..08d3c3a69 --- /dev/null +++ b/docs/ai/requirements/2026-10-17-feature-devkitd-agent-start.md @@ -0,0 +1,149 @@ +--- +phase: requirements +title: Migrate agent start to devkitd — agent.start RPC, daemon-owned registry +description: Move interactive agent spawning (tmux + herdr) from the TS agent-manager into devkitd; daemon.db becomes the registry source of truth; fixes the discovery-race name loss +--- + +# Requirements — devkitd-agent-start + +## Problem Statement + +`ai-devkit agent start` spawns agents entirely on the TypeScript side +(`packages/agent-manager`: `startAgent` → `TmuxAgentRuntime` / +`HerdrAgentRuntime` → `AgentRegistry` SQLite). Meanwhile `devkitd` owns live +process discovery, the enriched agent list, and the event stream. Split +authority produces a concrete, observed bug class: + +1. **Discovery race loses the requested name** (GitHub issue #398). The + console spawns `agent start` as a child CLI; the daemon's 2s sweep sees the + new process before `registry.register()` lands, so a parent refresh + registers the fallback `ai-devkit-` name first — and + `AgentRegistry.mergeEntry` preserves the existing name, so the fallback + wins permanently. The user's chosen name is lost and the session preview + stays unavailable until console restart. +2. **Two sources of truth.** The daemon's `agents` snapshot table knows + processes; the TS registry DB knows names, runtime, runtimeRef, pinned. + Neither can answer "who owns this pid" alone. +3. **No start event.** The daemon emits only `agent.appeared` on its sweep + cadence; subscribers get a late, unattributed notification instead of an + immediate `agent.started` carrying the requested identity. + +## Goals & Objectives + +**Primary:** + +- Add an `agent.start` JSON-RPC method to devkitd that performs the whole + start sequence: validate → availability → name-uniqueness → orphan-session + handling → spawn → PID resolution → register → emit `agent.started`. +- Port both interactive runtimes to Rust: `tmux` (session create, send-keys, + pane-pid descendant scan, stable-pid polling) and `herdr` (`herdr` CLI: + workspace create, agent start with `agent_pane_busy` retry, pane + process-info poll). +- Make `daemon.db` the registry source of truth by extending the existing + `agents` table with nullable registry columns (`name`, `agent_type`, + `runtime`, `runtime_ref`, `pinned`, `registered_at`) — one row per process + carries both observed and declared state. +- CLI routes `agent start` through the daemon when reachable, falling back to + the current TS path on connect-failure or `unknown method` (old devkitd). +- Structured error codes over the wire so the CLI reproduces today's exact + messages and exit codes. + +**Secondary:** + +- Emit `agent.started` immediately (don't wait for the sweep) and refresh the + enriched cache so `agent.list` shows the registered name at once. +- Every-boot delta import of the legacy agents DB into `daemon.db` so + pre-migration rows heal forward. +- Dual-write registrations into the legacy agents DB while it exists so + unmigrated consumers (`send`, `kill`, `rename`, `open`, channel bridge) + keep working. + +**Non-goals:** + +- Durable agents (`--mode durable` / `createDurableAgentService`) stay + entirely on the TS path. +- Migrating `agent stop|send|rename|open|pin` to RPC — out of scope; they + keep working via the dual-write bridge. +- Daemon-spawned processes for channels, bridge supervision, scheduled work. +- Version-gated daemon restart on CLI upgrade (tracked separately, #376) — + this feature only needs the `unknown method` fallback. +- Changes to the wire protocol framing (still newline-delimited JSON). + +## User Stories & Use Cases + +- As a CLI user, `ai-devkit agent start --type claude --name foo --cwd .` + prints the same success line, working directory, and attach hint whether + the daemon or the fallback path served it — and the requested name always + sticks. +- As a console user, starting an agent from the start pane shows it in the + list under its requested name immediately (the console spawns the CLI + subprocess, which routes through the daemon). +- As an agent-running-an-agent, `agent.start` is callable over the socket + without a nested CLI subprocess. +- As a maintainer, harness start knowledge lives next to the harness adapter + in `devkit-harness`, not split across languages. + +**Edge cases:** + +- Same name started twice concurrently → `name_in_use`, not a tmux-session + clobber race. +- Registered pid dies before the next sweep → tombstone row keeps + `runtime_ref` so the orphaned tmux/herdr pane is still cleanable by name. +- Pid reused by an unrelated process → must not inherit the dead agent's + name (identity is `(pid, start_time_ms)`). +- Daemon spawned under systemd/autostart has a narrower `PATH` than the + invoking shell — `tmux`/`herdr` binaries must resolve in the daemon's env. +- Client disconnects mid-start → daemon completes the start; the agent is + registered and visible in `agent.list` (strictly better than today's + orphaned-session outcome). + +## Success Criteria + +- `ai-devkit agent start` succeeds through the daemon for every startable + harness type on both runtimes; output and exit codes match the current + CLI exactly. +- Requested name is authoritative: no `ai-devkit-` fallback clobber + even when the sweep wins the race (regression for #398). +- `agent.started` event is emitted and replayable via `subscribe + {"afterSeq"}`; `agent.list` returns the agent under its requested name + immediately after the RPC resolves. +- Duplicate-name start returns `name_in_use`; dead-pid names are freed by + liveness-aware pruning. +- With an old devkitd on the socket (`unknown method`) or no daemon at all, + the CLI falls back to the TS path and still succeeds. +- `cargo test`, `cargo clippy -D warnings`, `cargo fmt --check` green; + `nx run-many -t build,test,lint` green including daemon-client typecheck + against regenerated proto types. +- Legacy agents DB receives the dual-written row while it exists; on daemon + boot, missing rows are imported. + +## Constraints & Assumptions + +- The daemon is per-user, single-process; all registry mutation funnels + through it, so a `start_mutex` makes name-uniqueness atomic. +- The wire `Response` gains an optional `code` field; `error` stays a + human-readable string for old clients. +- Blocking RPC: start may take up to `timeoutMs` (default 15s) plus herdr + retries (~8s worst case). Runs on the blocking pool; client request + timeout must cover it. +- Harness matchers port as data (`Argv0 | Argv0Contains | AnyBasename`), + not closures; semantics copied verbatim from `runtimeProfiles.ts`. +- Assumption: `opencode`, `pi`, `muse` need start profiles too even though + devkit-harness lacks full adapters for them — a start profile is just + command + match spec + discovery aliases. +- Assumption: user approved dual-write bridge and every-boot delta import + (decisions from pre-lifecycle brainstorm). + +## Questions & Open Items — resolved + +- **Registry storage**: merged into the existing `daemon.db.agents` table + (nullable registry columns) rather than a second table. Requires three + snapshot rules: NULL-`start_time_ms` rows are "registered, not yet swept" + (upsert, don't delete); identity-change on a swept row → full delete + (pid reuse drops names); true disappearance keeps named tombstones for + runtime_ref cleanup, freeing the name only via prune. +- **Scope**: herdr runtime in; durable agents out. +- **Transition**: dual-write to legacy DB + delta import on boot; remove + when remaining registry ops migrate to RPC. +- **Cancellation**: daemon completes start even if the client disconnects — + no `agent.start.cancel` in v1. diff --git a/docs/ai/testing/2026-10-17-feature-devkitd-agent-start.md b/docs/ai/testing/2026-10-17-feature-devkitd-agent-start.md new file mode 100644 index 000000000..7ec7635e5 --- /dev/null +++ b/docs/ai/testing/2026-10-17-feature-devkitd-agent-start.md @@ -0,0 +1,135 @@ +--- +phase: testing +title: devkitd-agent-start test coverage +description: Rust unit + socket e2e for agent.start, tmux/herdr via fake-CLI fixtures, snapshot-rule tests, daemon-client and CLI fallback tests +--- + +# Testing — devkitd-agent-start + +## Test Coverage Goals + +- 100% coverage on new Rust modules (`start_profiles`, `runtime/*`, + `agent_start` handler) and new `store.rs` registry paths. +- Socket-level e2e mirroring existing `server.rs` tests (temp dir, real + unix socket). +- External tools (`tmux`, `herdr`, `pgrep`, `ps`) faked with PATH-injected + shell-script fixtures — no real tmux/herdr required in CI. +- CLI behavior identical on daemon path and fallback path. + +## Unit Tests + +### `devkit-harness::start_profiles` + `MatchSpec` + +- [x] Port every case from `runtimeProfiles.test.ts`: argv0 match rejects + similar argument text; discovery aliases resolve; versioned launcher + (muse) profiles. +- [x] Spaced-directory cases: argv0 under `My Apps/`, distribution-name + match, argument-containing-spaces not treated as executable, "agent word + inside executable directory" non-match. Uses a tmpdir fixture for + `isFile` probing. +- [x] All 11 `StartableAgentType`s have profiles; unknown type → `None` + → `harness_not_startable`. +- [x] kiro profile: `command == "kiro-cli"`, `runtime_kind == "kiro"`. + +### `devkit-core::store` — registry columns + snapshot rules + +- [x] `register_start` upsert: new row gets name/type/runtime/ref; + conflict on pid preserves sweep columns, updates registry columns. +- [x] Registered-before-swept: row with `start_time_ms NULL` + name, then + sweep sees `(pid, real_st)` → row updated, not deleted; `appeared` + contains pid; `gone` does not. +- [x] Pid reuse: named row with old `start_time_ms`, sweep sees same pid + new `start_time_ms` → row deleted+reinserted, name cleared; + disappeared+appeared both emitted. +- [x] Tombstone: named row's pid absent from sweep → row survives, not in + `gone`… wait — it IS gone (disappeared event must fire) but row kept. + Assert: `agent.disappeared` fires once, row survives with name. +- [x] Tombstone pid reuse: dead named row, new proc takes pid → name + freed, no inheritance. +- [x] Name UNIQUE index: second `register_start` same name → constraint + error → `name_in_use`. +- [x] `prune_dead`: named rows whose pids absent from live snapshot are + reclaimed by `name_is_live` check. +- [x] Delta import: legacy `agents.db` rows appear as tombstones; existing + daemon rows win. + +### `devkitd::runtime::tmux` + +- [x] Fake `tmux` script: `-V` ok → available; missing → `binary-missing`. +- [x] Start sequence argv exact: `new-session -d -s N -c C`, + `send-keys -t N Enter`. +- [x] Existing session → `kill-session` before create; warning surfaced. +- [x] Pid polling: fake `list-panes`/`pgrep`/`ps` returning matching + command → pid returned; stability rule needs 5 identical polls (fake + returns alternating pids → keeps polling). +- [x] Timeout → `kill-session` issued → `pid_poll_timeout`. +- [x] Descendant match at depth >1 (pane→shell→node→claude chain). + +### `devkitd::runtime::herdr` + +- [x] Availability: `--version` ok + `api snapshot` ok → available; ENOENT + → `binary-missing`; snapshot failure → `backend-unreachable`. +- [x] `workspace create` JSON → `workspace_id` + `root_pane.pane_id`; + missing pane id → error. +- [x] `agent start` argv: `--kind`, `--pane`, `--timeout`, `-- args`. +- [x] `agent_pane_busy` JSON error → retried within 5s/100ms budget; + persistent busy → `runtime_unavailable`-class error. +- [x] `pane process-info`: `foreground_processes[0].pid` preferred; + fallback `shell_pid`; poll till deadline. +- [x] runtimeRef shape: `{session:"default", workspaceId, tabId, paneId, + agentName}`. + +### `devkitd::agent_start` handler + `server.rs` dispatch + +- [x] Validation: bad name regex, missing cwd, unknown type, unknown + runtime → `invalid_params`/`harness_not_startable`. +- [x] `name_in_use` with live pid → error carries `pid` in `details`. +- [x] Dead named row → pruned, start proceeds (liveness-aware name check). +- [x] `agent.started` emitted with entry payload; replayable via + `subscribe afterSeq`. +- [x] Concurrent same-name starts serialized by `start_mutex` → exactly + one succeeds. +- [x] Post-start `agent.list` shows registered name (enriched overlay). +- [x] `Response.code`/`details` serialize; old `error` string intact. +- [x] Dual-write: legacy db fixture receives row; missing legacy file → + skip silently. +- [x] Client disconnect mid-start → start completes, row registered. + +### `daemon-client` TS + +- [x] `startAgent` params/result wire roundtrip over real socket + (`DEVKITD_BIN` override; build `cargo build -p devkitd` first — see + stale-binary memory note). +- [x] Error code → `DaemonRpcError` with `code`/`details`. +- [x] `unknown method` from old daemon → distinguishable for fallback. + +### `cli` TS + +- [x] `agent start` with daemon up → result printed, warnings forwarded, + exit 0 (mock DaemonClient). +- [x] Daemon down / `unknown method` → TS `startAgent` invoked (spy). +- [x] Each error code maps to the same `ui.error` text as today's typed + errors; exit code 1 preserved. +- [x] `--mode durable` never touches the daemon path. +- [x] Existing `agent.test.ts` suite stays green (fallback path). + +## Integration / E2E + +- [x] Temp HOME: daemon serve → `agent.start` → `agent.list` shows name; + `subscribe` sees `agent.started` then `agent.appeared`. +- [x] Real `tmux` smoke (manual/CI-optional, macOS+Linux): actual session, + actual `claude`-shim binary, attach hint correct. +- [x] Kill spawned proc → `agent.disappeared` + tombstone; start same name + again → succeeds (name freed by prune). + +## Mocks / fixtures + +- `rust/crates/devkitd/tests/fixtures/`: fake `tmux`, `herdr`, `pgrep`, + `ps` shell scripts driven by env-provided state files. +- Legacy `agents.db` fixture at schema v5 with representative rows. +- Reuse existing `server.rs` tempdir-socket test harness. + +## Coverage target + +New Rust code ≥ 90% line coverage; all error codes exercised; both +runtimes covered by fake-CLI tests; CLI fallback covered by vitest. diff --git a/packages/cli/src/__tests__/commands/agent.test.ts b/packages/cli/src/__tests__/commands/agent.test.ts index df7702ff1..6b6e9f030 100644 --- a/packages/cli/src/__tests__/commands/agent.test.ts +++ b/packages/cli/src/__tests__/commands/agent.test.ts @@ -89,6 +89,8 @@ const { mockHerdrReadOutput, mockHerdrFocus, mockHerdrStop, + mockEnsureDaemon, + mockDaemonStartAgent, } = vi.hoisted(() => ({ mockEnableDebug: vi.fn(), mockDebugLogger: vi.fn(), @@ -108,6 +110,8 @@ const { mockHerdrReadOutput: vi.fn().mockResolvedValue("done\n"), mockHerdrFocus: vi.fn().mockResolvedValue(true), mockHerdrStop: vi.fn().mockResolvedValue(undefined), + mockEnsureDaemon: vi.fn().mockResolvedValue(null), + mockDaemonStartAgent: vi.fn(), })); let restoreStdin: (() => void) | undefined; @@ -337,6 +341,14 @@ vi.mock("../../lib/Config.js", () => ({ }), })); +vi.mock("@ai-devkit/daemon-client", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + ensureDaemon: mockEnsureDaemon, + }; +}); + vi.mock("../../tui/console/ConsoleApp.js", () => ({ ConsoleApp: () => null, })); @@ -417,6 +429,8 @@ describe("agent command", () => { mockDurableService.send.mockReset(); mockPiPrintService.create.mockReset(); mockPiPrintService.send.mockReset(); + mockEnsureDaemon.mockReset().mockResolvedValue(null); + mockDaemonStartAgent.mockReset(); mockFocusManager.findTerminal.mockReset(); mockFocusManager.focusTerminal.mockReset(); mockTtyWriterSend.mockReset().mockResolvedValue(undefined); @@ -761,6 +775,199 @@ describe("agent command", () => { expect(process.exit).toHaveBeenCalledWith(1); }); + describe("daemon-routed start", () => { + function daemonUp(result: unknown) { + const client = { + startAgent: mockDaemonStartAgent, + close: vi.fn(), + }; + mockEnsureDaemon.mockResolvedValue(client); + mockDaemonStartAgent.mockResolvedValue(result); + return client; + } + + it("starts through the daemon and skips the local path", async () => { + const client = daemonUp({ + name: "agent1", + type: "claude", + pid: 4242, + runtime: "tmux", + runtimeRef: { session: "agent1" }, + cwd: process.cwd(), + startedAt: "2026-10-17T00:00:00.000Z", + warnings: [], + }); + const program = new Command(); + registerAgentCommand(program); + + await program.parseAsync([ + "node", + "test", + "agent", + "start", + "--type", + "claude", + "--name", + "agent1", + ]); + + expect(mockDaemonStartAgent).toHaveBeenCalledWith({ + type: "claude", + name: "agent1", + cwd: process.cwd(), + runtime: "tmux", + }); + expect(mockStartAgent).not.toHaveBeenCalled(); + expect(ui.success).toHaveBeenCalledWith('Agent "agent1" started (claude, PID 4242)'); + expect(ui.text).toHaveBeenCalledWith("Attach: tmux attach -t agent1"); + expect(client.close).toHaveBeenCalled(); + }); + + it("forwards daemon warnings", async () => { + daemonUp({ + name: "agent1", + type: "claude", + pid: 1, + runtime: "tmux", + runtimeRef: { session: "agent1" }, + cwd: process.cwd(), + startedAt: "2026-10-17T00:00:00.000Z", + warnings: [ + 'tmux session "agent1" already exists but has no live registry entry — it will be replaced.', + ], + }); + const program = new Command(); + registerAgentCommand(program); + + await program.parseAsync([ + "node", + "test", + "agent", + "start", + "--type", + "claude", + "--name", + "agent1", + ]); + + expect(ui.warning).toHaveBeenCalledWith( + 'tmux session "agent1" already exists but has no live registry entry — it will be replaced.', + ); + }); + + it("reports coded daemon errors without falling back", async () => { + const { DaemonRpcError } = await import("@ai-devkit/daemon-client"); + const client = { + startAgent: mockDaemonStartAgent, + close: vi.fn(), + }; + mockEnsureDaemon.mockResolvedValue(client); + mockDaemonStartAgent.mockRejectedValue( + new DaemonRpcError( + 'Agent "agent1" is already running (PID 77). Choose a different name.', + "name_in_use", + { pid: 77 }, + ), + ); + const program = new Command(); + registerAgentCommand(program); + + await program.parseAsync([ + "node", + "test", + "agent", + "start", + "--type", + "claude", + "--name", + "agent1", + ]); + + expect(ui.error).toHaveBeenCalledWith( + 'Agent "agent1" is already running (PID 77). Choose a different name.', + ); + expect(mockStartAgent).not.toHaveBeenCalled(); + expect(process.exit).toHaveBeenCalledWith(1); + }); + + it("maps pid_poll_timeout details to the existing message", async () => { + const { DaemonRpcError } = await import("@ai-devkit/daemon-client"); + mockEnsureDaemon.mockResolvedValue({ + startAgent: mockDaemonStartAgent, + close: vi.fn(), + }); + mockDaemonStartAgent.mockRejectedValue( + new DaemonRpcError("timeout", "pid_poll_timeout", { + command: "claude", + timeoutMs: 15000, + }), + ); + const program = new Command(); + registerAgentCommand(program); + + await program.parseAsync([ + "node", + "test", + "agent", + "start", + "--type", + "claude", + "--name", + "agent1", + ]); + + expect(ui.error).toHaveBeenCalledWith( + 'Agent process not found after 15s. Verify that "claude" is in PATH inside the tmux environment.', + ); + expect(process.exit).toHaveBeenCalledWith(1); + }); + + it("falls back to the local path on unknown method (old devkitd)", async () => { + const { DaemonRpcError } = await import("@ai-devkit/daemon-client"); + mockEnsureDaemon.mockResolvedValue({ + startAgent: mockDaemonStartAgent, + close: vi.fn(), + }); + mockDaemonStartAgent.mockRejectedValue(new DaemonRpcError("unknown method: agent.start")); + const program = new Command(); + registerAgentCommand(program); + + await program.parseAsync([ + "node", + "test", + "agent", + "start", + "--type", + "claude", + "--name", + "agent1", + ]); + + expect(mockStartAgent).toHaveBeenCalledTimes(1); + expect(ui.success).toHaveBeenCalledWith('Agent "agent1" started (claude, PID 12345)'); + }); + + it("falls back to the local path when the daemon is unreachable", async () => { + mockEnsureDaemon.mockResolvedValue(null); + const program = new Command(); + registerAgentCommand(program); + + await program.parseAsync([ + "node", + "test", + "agent", + "start", + "--type", + "claude", + "--name", + "agent1", + ]); + + expect(mockStartAgent).toHaveBeenCalledTimes(1); + expect(mockDaemonStartAgent).not.toHaveBeenCalled(); + }); + }); + it("shows info when no agents are running", async () => { mockManager.listAgents.mockResolvedValue([]); diff --git a/packages/cli/src/commands/agent/start.command.ts b/packages/cli/src/commands/agent/start.command.ts index 69664f1fc..598e97531 100644 --- a/packages/cli/src/commands/agent/start.command.ts +++ b/packages/cli/src/commands/agent/start.command.ts @@ -15,13 +15,117 @@ import { } from "@ai-devkit/agent-manager"; import { ui } from "../../util/terminal-ui.js"; import { withErrorHandler } from "../../util/errors.js"; -import { enableDebug } from "../../util/debug.js"; +import { createLogger, enableDebug } from "../../util/debug.js"; import { AGENT_NAME_REGEX, generateAgentName } from "../../util/agent.js"; import { resolveTmuxInstallInstructions } from "../../util/tmux.js"; import { createTmuxInspectionDeps } from "../../util/tmux-deps.js"; import { ConfigManager } from "../../lib/Config.js"; import { createDurableAgentService } from "./factory.js"; import { formatCwd, formatPrintProvider } from "./render.js"; +import { DaemonRpcError, ensureDaemon } from "@ai-devkit/daemon-client"; + +interface StartedEntry { + name: string; + type: string; + pid: number; + runtime: string; + runtimeRef: unknown; + cwd: string; +} + +function renderStartedAgent(entry: StartedEntry): void { + ui.success(`Agent "${entry.name}" started (${entry.type}, PID ${entry.pid})`); + ui.text(`Working directory: ${formatCwd(entry.cwd)}`); + if (entry.runtime === "herdr") { + ui.text("Runtime: herdr"); + } else { + const tmuxRef = parseTmuxRuntimeRef(entry.runtimeRef); + if (tmuxRef) ui.text(`Attach: tmux attach -t ${tmuxRef.session}`); + } +} + +/** + * Daemon-first start: devkitd owns spawning + registration when reachable. + * Falls back to the local TS path on connect-failure or `unknown method` + * (older devkitd on the socket). Coded daemon errors render the same + * messages as their typed-error equivalents — no retry, the daemon already + * reported the real failure. + */ +const debug = createLogger("agent:start"); + +async function tryDaemonStart(input: { + agentType: StartableAgentType; + agentName: string; + cwd: string; + runtimeProvider: string; +}): Promise<"done" | "fallback" | "failed"> { + const daemon = await ensureDaemon({ onDebug: debug }); + if (!daemon) { + debug("no daemon available -> falling back to local startAgent path"); + return "fallback"; + } + debug("daemon path: calling agent.start over the daemon socket"); + try { + const result = await daemon.startAgent({ + type: input.agentType, + name: input.agentName, + cwd: input.cwd, + runtime: input.runtimeProvider, + }); + for (const warning of result.warnings ?? []) ui.warning(warning); + renderStartedAgent({ + name: result.name, + type: result.type, + pid: result.pid, + runtime: result.runtime, + runtimeRef: result.runtimeRef, + cwd: result.cwd, + }); + return "done"; + } catch (err) { + if (err instanceof DaemonRpcError && err.code) { + debug(`daemon returned coded error ${err.code}; not retrying locally`); + await reportDaemonStartError(err); + return "failed"; + } + if (err instanceof Error && err.message.startsWith("unknown method")) { + debug("daemon predates agent.start (unknown method) -> local fallback"); + return "fallback"; // pre-agent.start devkitd + } + // Transport failure mid-request — the daemon may have completed the + // start; retrying locally could double-spawn. Report, don't retry. + debug("transport failure mid-request; reporting without retry"); + ui.error(err instanceof Error ? err.message : String(err)); + return "failed"; + } finally { + daemon.close(); + } +} + +async function reportDaemonStartError(err: DaemonRpcError): Promise { + const details = (err.details ?? {}) as Record; + switch (err.code) { + case "name_in_use": + ui.error(err.message); + break; + case "runtime_unavailable": + if (details.provider === "tmux") { + const instructions = await resolveTmuxInstallInstructions(createTmuxInspectionDeps()); + ui.error(`tmux is not installed or not in PATH. ${instructions.message}`); + } else { + ui.error(`Herdr runtime is unavailable (${details.reason}): ${details.detail}`); + } + break; + case "pid_poll_timeout": + ui.error( + `Agent process not found after ${Number(details.timeoutMs) / 1000}s. ` + + `Verify that "${details.command}" is in PATH inside the tmux environment.`, + ); + break; + default: + ui.error(err.message); + } +} async function reportStartError(err: unknown): Promise { if (err instanceof TmuxUnavailableError) { @@ -110,6 +214,18 @@ export function registerAgentStartCommand(agentCommand: Command): void { return; } const runtimeProvider = await new ConfigManager().getAgentRuntimeProvider(); + const routed = await tryDaemonStart({ + agentType: agentType as StartableAgentType, + agentName, + cwd, + runtimeProvider, + }); + if (routed === "done") return; + if (routed === "failed") { + process.exit(1); + return; // process.exit is mocked under test — don't fall through + } + debug("local startAgent path (no daemon or older devkitd)"); const entry = await startAgent( { type: agentType as StartableAgentType, @@ -119,14 +235,7 @@ export function registerAgentStartCommand(agentCommand: Command): void { }, { onWarning: (msg: string) => ui.warning(msg) }, ); - ui.success(`Agent "${entry.name}" started (${entry.type}, PID ${entry.pid})`); - ui.text(`Working directory: ${formatCwd(entry.cwd)}`); - if (entry.runtime === "herdr") { - ui.text("Runtime: herdr"); - } else { - const tmuxRef = parseTmuxRuntimeRef(entry.runtimeRef); - if (tmuxRef) ui.text(`Attach: tmux attach -t ${tmuxRef.session}`); - } + renderStartedAgent(entry); } catch (err) { await reportStartError(err); process.exit(1); diff --git a/packages/daemon-client/src/__tests__/integration.test.ts b/packages/daemon-client/src/__tests__/integration.test.ts index c80e3be4b..904b1a145 100644 --- a/packages/daemon-client/src/__tests__/integration.test.ts +++ b/packages/daemon-client/src/__tests__/integration.test.ts @@ -123,6 +123,31 @@ describe.runIf(bin != null)("ensureDaemon (real binary)", () => { expect(Array.isArray(result)).toBe(true); client!.close(); }, 30000); + + it("startAgent surfaces coded daemon errors", async () => { + // Isolated HOME guarantees a fresh devkitd at the current build — a + // stale user daemon on the real socket would answer "unknown method". + const tmpHome = fs.mkdtempSync(path.join(os.tmpdir(), "aidk-home-")); + const prevHome = process.env.HOME; + process.env.HOME = tmpHome; + try { + const client = await ensureDaemon({ waitMs: 20000 }); + expect(client).not.toBeNull(); + // Unknown harness type → harness_not_startable (no spawn attempted). + await expect(client!.startAgent({ type: "nonsense", cwd: "/" })).rejects.toMatchObject({ + code: "harness_not_startable", + }); + // Bad name → invalid_params before anything spawns. + await expect( + client!.startAgent({ type: "claude", name: "BAD NAME", cwd: "/" }), + ).rejects.toMatchObject({ code: "invalid_params" }); + await client!.request("shutdown").catch(() => {}); + client!.close(); + } finally { + if (prevHome === undefined) delete process.env.HOME; + else process.env.HOME = prevHome; + } + }, 30000); }); describe.runIf(fs.existsSync(sock))("live daemon", () => { diff --git a/packages/daemon-client/src/autostart.ts b/packages/daemon-client/src/autostart.ts index 2bccd9e40..4b9bf03c5 100644 --- a/packages/daemon-client/src/autostart.ts +++ b/packages/daemon-client/src/autostart.ts @@ -10,18 +10,34 @@ import { resolveDaemonBinary } from "./binary.js"; * for readiness. Returns null when no binary exists or spawn fails — callers * degrade to today's code paths. */ -export async function ensureDaemon(opts: { waitMs?: number } = {}): Promise { +export async function ensureDaemon( + opts: { waitMs?: number; onDebug?: (msg: string) => void } = {}, +): Promise { const waitMs = opts.waitMs ?? 3000; + const debug = opts.onDebug; // Test runs must never spawn a real daemon. Callers that want to exercise // autostart do so by pointing DEVKITD_BIN at a real binary in an explicit // integration test — the env below is checked there too via VITEST_SKIP. - if (process.env.AI_DEVKIT_NO_DAEMON) return null; + if (process.env.AI_DEVKIT_NO_DAEMON) { + debug?.("daemon path disabled (AI_DEVKIT_NO_DAEMON set)"); + return null; + } const existing = await DaemonClient.tryConnect(); - if (existing) return existing; - if (process.env.VITEST && !process.env.DEVKITD_BIN && !process.env.AI_DEVKITD_BIN) return null; + if (existing) { + debug?.(`connected to live daemon at ${daemonSocketPath()}`); + return existing; + } + debug?.("no live daemon on socket; attempting autostart"); + if (process.env.VITEST && !process.env.DEVKITD_BIN && !process.env.AI_DEVKITD_BIN) { + debug?.("autostart skipped in test env (VITEST set, no DEVKITD_BIN)"); + return null; + } - const bin = resolveDaemonBinary(); - if (!bin) return null; + const bin = resolveDaemonBinary(debug); + if (!bin) { + debug?.("no devkitd binary; daemon path unavailable"); + return null; + } const sockPath = daemonSocketPath(); const dir = path.dirname(sockPath); @@ -35,7 +51,8 @@ export async function ensureDaemon(opts: { waitMs?: number } = {}): Promise { +async function waitForSocket( + waitMs: number, + debug?: (msg: string) => void, +): Promise { const deadline = Date.now() + waitMs; while (Date.now() < deadline) { const client = await DaemonClient.tryConnect(); - if (client) return client; + if (client) { + debug?.("devkitd socket ready"); + return client; + } await new Promise((r) => setTimeout(r, 50)); } + debug?.(`timed out after ${waitMs}ms waiting for devkitd socket`); return null; } diff --git a/packages/daemon-client/src/binary.ts b/packages/daemon-client/src/binary.ts index 45cc3705d..92f113bc8 100644 --- a/packages/daemon-client/src/binary.ts +++ b/packages/daemon-client/src/binary.ts @@ -10,11 +10,15 @@ import { fileURLToPath } from "node:url"; * 2. Platform package `@ai-devkit/devkitd--` (npm optional dep) * 3. ~/.ai-devkit/bin/devkitd (manual install) * 4. rust/target/{release,debug}/devkitd relative to the monorepo (dev) - * Returns null when nothing is found — callers must degrade gracefully. + * Returns null when nothing is found — callers + * must degrade gracefully. */ -export function resolveDaemonBinary(): string | null { +export function resolveDaemonBinary(onDebug?: (msg: string) => void): string | null { const env = process.env.DEVKITD_BIN ?? process.env.AI_DEVKITD_BIN; - if (env && fs.existsSync(env)) return env; + if (env && fs.existsSync(env)) { + onDebug?.(`devkitd binary from env override: ${env}`); + return env; + } const plats: Record = { linux: "linux", @@ -32,14 +36,20 @@ export function resolveDaemonBinary(): string | null { path.dirname(pkgJson), process.platform === "win32" ? "devkitd.exe" : "devkitd", ); - if (fs.existsSync(candidate)) return candidate; + if (fs.existsSync(candidate)) { + onDebug?.(`devkitd binary from platform package: ${candidate}`); + return candidate; + } } catch { /* package not installed */ } } const home = path.join(os.homedir(), ".ai-devkit", "bin", "devkitd"); - if (fs.existsSync(home)) return home; + if (fs.existsSync(home)) { + onDebug?.(`devkitd binary from ~/.ai-devkit/bin: ${home}`); + return home; + } // Dev checkout: walk up to the nearest rust/target//devkitd and // pick the freshest profile — a stale release build otherwise shadows a @@ -57,10 +67,14 @@ export function resolveDaemonBinary(): string | null { /* absent */ } } - if (best) return best.file; + if (best) { + onDebug?.(`devkitd binary from dev checkout: ${best.file}`); + return best.file; + } const parent = path.dirname(dir); if (parent === dir) break; dir = parent; } + onDebug?.("no devkitd binary found in any location"); return null; } diff --git a/packages/daemon-client/src/client.ts b/packages/daemon-client/src/client.ts index 783c27282..d7cd44877 100644 --- a/packages/daemon-client/src/client.ts +++ b/packages/daemon-client/src/client.ts @@ -1,12 +1,29 @@ import * as net from "node:net"; import * as os from "node:os"; import * as path from "node:path"; +import type { AgentStartParams } from "./gen/AgentStartParams.js"; +import type { AgentStartResult } from "./gen/AgentStartResult.js"; import type { EnrichedAgent } from "./gen/EnrichedAgent.js"; import type { Event } from "./gen/Event.js"; /** Wire event pushed by the daemon — generated from devkit-core via ts-rs. */ export type DaemonEvent = Event; +/** + * RPC error with the daemon's structured `code`/`details` when present. + * `code` is undefined for old daemons and local transport failures. + */ +export class DaemonRpcError extends Error { + constructor( + message: string, + readonly code?: string, + readonly details?: unknown, + ) { + super(message); + this.name = "DaemonRpcError"; + } +} + export function daemonSocketPath(): string { return path.join(os.homedir(), ".ai-devkit", "daemon.sock"); } @@ -78,7 +95,7 @@ export class DaemonClient { } } - request(method: string, params: unknown = {}): Promise { + request(method: string, params: unknown = {}, timeoutMs?: number): Promise { if (!this.socket) return Promise.reject(new Error("not connected")); const id = this.nextId++; return new Promise((resolve, reject) => { @@ -88,7 +105,7 @@ export class DaemonClient { if (this.pending.delete(id)) { reject(new Error(`daemon request "${method}" timed out`)); } - }, this.requestTimeoutMs); + }, timeoutMs ?? this.requestTimeoutMs); this.pending.set(id, { resolve, reject, timer }); this.socket!.write(JSON.stringify({ id, method, params }) + "\n"); }); @@ -111,6 +128,16 @@ export class DaemonClient { return (await this.request("agent.list")) as EnrichedAgent[]; } + /** + * Daemon-side agent start — the daemon validates, spawns via the selected + * runtime, registers the agent, and emits agent.started. The request + * timeout covers the pid-resolution deadline plus a margin. + */ + async startAgent(params: AgentStartParams): Promise { + const timeoutMs = Number(params.timeoutMs ?? 15_000) + 15_000; + return (await this.request("agent.start", params, timeoutMs)) as AgentStartResult; + } + private onData(chunk: Buffer) { this.buffer += chunk.toString("utf8"); let idx: number; @@ -133,7 +160,10 @@ export class DaemonClient { if (!p) continue; this.pending.delete(id); clearTimeout(p.timer); - if (msg.error) p.reject(new Error(String(msg.error))); + if (msg.error) + p.reject( + new DaemonRpcError(String(msg.error), msg.code as string | undefined, msg.details), + ); else p.resolve(msg.result); } } diff --git a/packages/daemon-client/src/gen/AgentStartParams.ts b/packages/daemon-client/src/gen/AgentStartParams.ts new file mode 100644 index 000000000..aa00d4d35 --- /dev/null +++ b/packages/daemon-client/src/gen/AgentStartParams.ts @@ -0,0 +1,8 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * Params for `agent.start`. Mirrors the TS `StartAgentOptions` surface that + * crosses the wire — `runtime` is the provider ("tmux" | "herdr"), resolved + * client-side from config. + */ +export type AgentStartParams = { type: string, name?: string, cwd: string, runtime?: string, args?: Array, timeoutMs?: number, pollIntervalMs?: number, }; diff --git a/packages/daemon-client/src/gen/AgentStartResult.ts b/packages/daemon-client/src/gen/AgentStartResult.ts new file mode 100644 index 000000000..fdb910bc0 --- /dev/null +++ b/packages/daemon-client/src/gen/AgentStartResult.ts @@ -0,0 +1,7 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +/** + * `agent.start` result — the registered entry plus non-fatal warnings the + * CLI renders verbatim. + */ +export type AgentStartResult = { name: string, type: string, pid: number, runtime: string, runtimeRef: unknown, cwd: string, startedAt: string, warnings: Array, }; diff --git a/packages/daemon-client/src/gen/Response.ts b/packages/daemon-client/src/gen/Response.ts index 04e4c0b4f..282340fcb 100644 --- a/packages/daemon-client/src/gen/Response.ts +++ b/packages/daemon-client/src/gen/Response.ts @@ -1,3 +1,12 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type Response = { id?: number, result?: unknown, error?: string, }; +export type Response = { id?: number, result?: unknown, error?: string, +/** + * Machine-readable error code (e.g. "name_in_use"). Absent on legacy + * errors — clients must handle its absence. + */ +code?: string, +/** + * Structured error context (pid, command, timeoutMs, ...). + */ +details?: unknown, }; diff --git a/packages/daemon-client/src/index.ts b/packages/daemon-client/src/index.ts index 434741752..3d6fb1802 100644 --- a/packages/daemon-client/src/index.ts +++ b/packages/daemon-client/src/index.ts @@ -1,5 +1,7 @@ -export { DaemonClient, daemonSocketPath } from "./client.js"; +export { DaemonClient, daemonSocketPath, DaemonRpcError } from "./client.js"; export type { DaemonEvent } from "./client.js"; +export type { AgentStartParams } from "./gen/AgentStartParams.js"; +export type { AgentStartResult } from "./gen/AgentStartResult.js"; export type { EnrichedAgent, Event } from "./gen/index.js"; export { resolveDaemonBinary } from "./binary.js"; export { ensureDaemon } from "./autostart.js"; diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 64e59c320..13745d900 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -118,6 +118,7 @@ dependencies = [ "devkit-core", "devkit-harness", "libc", + "serde", "serde_json", "tokio", "tracing", diff --git a/rust/crates/devkit-core/src/proto.rs b/rust/crates/devkit-core/src/proto.rs index 50cefd072..281458884 100644 --- a/rust/crates/devkit-core/src/proto.rs +++ b/rust/crates/devkit-core/src/proto.rs @@ -24,6 +24,15 @@ pub struct Response { #[serde(skip_serializing_if = "Option::is_none")] #[ts(optional)] pub error: Option, + /// Machine-readable error code (e.g. "name_in_use"). Absent on legacy + /// errors — clients must handle its absence. + #[serde(skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub code: Option, + /// Structured error context (pid, command, timeoutMs, ...). + #[serde(skip_serializing_if = "Option::is_none")] + #[ts(optional, type = "unknown")] + pub details: Option, } impl Response { @@ -32,6 +41,8 @@ impl Response { id, result: Some(result), error: None, + code: None, + details: None, } } pub fn err(id: Option, msg: impl Into) -> Self { @@ -39,10 +50,71 @@ impl Response { id, result: None, error: Some(msg.into()), + code: None, + details: None, + } + } + pub fn err_coded( + id: Option, + code: impl Into, + msg: impl Into, + details: Option, + ) -> Self { + Self { + id, + result: None, + error: Some(msg.into()), + code: Some(code.into()), + details, } } } +/// Params for `agent.start`. Mirrors the TS `StartAgentOptions` surface that +/// crosses the wire — `runtime` is the provider ("tmux" | "herdr"), resolved +/// client-side from config. +#[derive(Debug, Deserialize, TS)] +pub struct AgentStartParams { + #[serde(rename = "type")] + pub agent_type: String, + #[serde(default)] + #[ts(optional)] + pub name: Option, + pub cwd: String, + #[serde(default)] + #[ts(optional)] + pub runtime: Option, + #[serde(default)] + #[ts(optional)] + pub args: Option>, + #[serde(default, rename = "timeoutMs")] + #[ts(optional, type = "number")] + pub timeout_ms: Option, + #[serde(default, rename = "pollIntervalMs")] + #[ts(optional, type = "number")] + pub poll_interval_ms: Option, +} + +/// `agent.start` result — the registered entry plus non-fatal warnings the +/// CLI renders verbatim. +#[derive(Debug, Serialize, TS)] +pub struct AgentStartResult { + pub name: String, + #[serde(rename = "type")] + pub agent_type: String, + #[ts(type = "number")] + pub pid: u64, + pub runtime: String, + #[serde(rename = "runtimeRef")] + #[ts(type = "unknown")] + pub runtime_ref: Value, + pub cwd: String, + #[serde(rename = "startedAt")] + pub started_at: String, + #[serde(default)] + pub warnings: Vec, +} + /// Event pushed to subscribers. `seq` is the persisted log sequence number — /// subscribers ack by seq to get at-least-once delivery. #[derive(Debug, Clone, Serialize, Deserialize, TS)] @@ -74,6 +146,8 @@ mod tests { Request::export(&cfg).unwrap(); Response::export(&cfg).unwrap(); Event::export(&cfg).unwrap(); + AgentStartParams::export(&cfg).unwrap(); + AgentStartResult::export(&cfg).unwrap(); crate::agent::EnrichedAgent::export(&cfg).unwrap(); // ts-rs emits `import ... from "./X"` without the `.js` extension @@ -123,6 +197,59 @@ mod tests { .contains("\"id\"")); } + #[test] + fn response_carries_error_code_and_details() { + let err = serde_json::to_string(&Response::err_coded( + Some(3), + "name_in_use", + "Agent \"x\" is already running (PID 42).", + Some(json!({"pid": 42})), + )) + .unwrap(); + let v: Value = serde_json::from_str(&err).unwrap(); + assert_eq!(v["code"], "name_in_use"); + assert_eq!(v["details"]["pid"], 42); + // Old clients still see a plain error string. + assert!(v["error"].as_str().unwrap().contains("already running")); + // code/details omitted when absent. + assert!(!serde_json::to_string(&Response::err(Some(2), "nope")) + .unwrap() + .contains("\"code\"")); + } + + #[test] + fn agent_start_params_deserialize() { + let p: AgentStartParams = serde_json::from_str( + r#"{"type":"claude","cwd":"/repo","runtime":"tmux", + "args":["--x"],"timeoutMs":15000,"pollIntervalMs":500}"#, + ) + .unwrap(); + assert_eq!(p.agent_type, "claude"); + assert_eq!(p.cwd, "/repo"); + assert_eq!(p.runtime.as_deref(), Some("tmux")); + assert_eq!(p.timeout_ms, Some(15000)); + assert!(p.name.is_none()); + } + + #[test] + fn agent_start_result_serializes_registry_entry_shape() { + let r = AgentStartResult { + name: "foo".into(), + agent_type: "claude".into(), + pid: 4242, + runtime: "tmux".into(), + runtime_ref: json!({"session": "foo"}), + cwd: "/repo".into(), + started_at: "2026-10-17T00:00:00.000Z".into(), + warnings: vec!["orphan replaced".into()], + }; + let v: Value = serde_json::from_str(&serde_json::to_string(&r).unwrap()).unwrap(); + assert_eq!(v["type"], "claude"); + assert_eq!(v["runtimeRef"]["session"], "foo"); + assert_eq!(v["startedAt"], "2026-10-17T00:00:00.000Z"); + assert_eq!(v["warnings"][0], "orphan replaced"); + } + #[test] fn event_frame_shape_is_stable() { // Subscribers parse {"event": {...}} — field names are the contract. diff --git a/rust/crates/devkit-core/src/store.rs b/rust/crates/devkit-core/src/store.rs index a206ed27f..ac8e77ae1 100644 --- a/rust/crates/devkit-core/src/store.rs +++ b/rust/crates/devkit-core/src/store.rs @@ -2,7 +2,7 @@ use crate::proto::Event; use anyhow::{Context, Result}; use rusqlite::{params, Connection}; use serde_json::Value; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::Mutex; /// Newest events kept before retention pruning kicks in. @@ -12,8 +12,14 @@ const PRUNE_EVERY: i64 = 512; /// Daemon-owned store: the persisted agent snapshot (for diff/events across /// restarts) and the durable event log. The daemon is the sole writer. +mod registry; +pub use registry::Registration; + pub struct Store { conn: Mutex, + /// Legacy agent registry (`~/.ai-devkit/agents.db`) — dual-write target + /// while unmigrated TS consumers still read it. + legacy_db: Mutex>, } impl Store { @@ -41,19 +47,54 @@ impl Store { last_seen INTEGER NOT NULL );", )?; - // Migration: databases created before start_time-keyed identity lack - // the column — add it in place. + // Migrations: databases created before later identity/registry + // columns lack them — add each missing column in place. { let mut s = conn.prepare("PRAGMA table_info(agents)")?; let cols = s .query_map([], |r| r.get::<_, String>(1))? .collect::, _>>()?; - if !cols.iter().any(|c| c == "start_time_ms") { - conn.execute("ALTER TABLE agents ADD COLUMN start_time_ms INTEGER", [])?; + for (col, ddl) in [ + ( + "start_time_ms", + "ALTER TABLE agents ADD COLUMN start_time_ms INTEGER", + ), + // Registry columns — NULL/empty until `register_start` claims + // the row. `name` doubles as the "is registered" marker. + ("name", "ALTER TABLE agents ADD COLUMN name TEXT"), + ( + "agent_type", + "ALTER TABLE agents ADD COLUMN agent_type TEXT", + ), + ("runtime", "ALTER TABLE agents ADD COLUMN runtime TEXT"), + ( + "runtime_ref", + "ALTER TABLE agents ADD COLUMN runtime_ref TEXT NOT NULL DEFAULT ''", + ), + ( + "pinned", + "ALTER TABLE agents ADD COLUMN pinned INTEGER NOT NULL DEFAULT 0", + ), + ( + "registered_at", + "ALTER TABLE agents ADD COLUMN registered_at INTEGER", + ), + // Disappearance acknowledged: tombstone marker so repeated + // sweeps don't re-emit agent.disappeared for dead rows. + ("gone_at", "ALTER TABLE agents ADD COLUMN gone_at INTEGER"), + ] { + if !cols.iter().any(|c| c == col) { + conn.execute(ddl, [])?; + } } + conn.execute_batch( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_agents_name + ON agents(name) WHERE name IS NOT NULL;", + )?; } Ok(Self { conn: Mutex::new(conn), + legacy_db: Mutex::new(None), }) } @@ -100,6 +141,12 @@ impl Store { /// Replace the live-agent snapshot with the latest discovery sweep. /// Returns (appeared_pids, disappeared_pids) so the server can emit events. + /// + /// Identity is (pid, start_time_ms). Registry columns change the row + /// lifecycle: registered rows (name non-NULL) that disappear become + /// tombstones (gone_at set, row kept for runtime_ref cleanup) instead of + /// being deleted, and rows registered before their first sweep + /// (start_time_ms NULL) are adopted rather than deleted+reinserted. pub fn apply_agent_snapshot( &self, agents: &[crate::discover::AgentProc], @@ -107,30 +154,85 @@ impl Store { let conn = self.conn.lock().unwrap(); let now = chrono_now(); let tx = conn.unchecked_transaction()?; - // Identity is (pid, start_time_ms) — pid alone breaks under reuse: a - // recycled pid would silently inherit the dead process's first_seen - // and emit no appeared/disappeared events at all. - let known: std::collections::HashSet<(i64, Option)> = { - let mut s = tx.prepare("SELECT pid, start_time_ms FROM agents")?; + // pid -> (start_time_ms, registered, gone_at) + let known: std::collections::HashMap, bool, Option)> = { + let mut s = tx.prepare("SELECT pid, start_time_ms, name, gone_at FROM agents")?; let v = s - .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))? - .collect::)>, _>>( - )?; + .query_map([], |r| { + Ok(( + r.get::<_, i64>(0)?, + ( + r.get::<_, Option>(1)?, + r.get::<_, Option>(2)?.is_some(), + r.get::<_, Option>(3)?, + ), + )) + })? + .collect::, bool, Option)>, + _, + >>()?; v }; - let current: std::collections::HashSet<(i64, Option)> = - agents.iter().map(|a| (a.pid, a.start_time_ms)).collect(); + let current: std::collections::HashSet = agents.iter().map(|a| a.pid).collect(); + + let mut appeared: Vec = Vec::new(); + let mut gone: Vec = Vec::new(); + + for a in agents { + match known.get(&a.pid) { + None => appeared.push(a.pid), + Some((Some(st), _, _)) if Some(*st) == a.start_time_ms => {} + // Registered before its first sweep — adopt the real start + // time via the upsert below; first observation counts as + // appeared. Unregistered NULL-st rows getting a start time + // resolve silently (same proc, previously unmeasured). + Some((None, registered, _)) => { + if *registered { + appeared.push(a.pid); + } + } + Some((Some(_), _, _)) if a.start_time_ms.is_none() => { + // Sweep couldn't measure start time this pass — can't + // distinguish identity change; keep the row and its + // registration rather than nuking on flaky data. + } + Some((Some(_), _, _)) => { + // start_time mismatch → pid reuse: drop the whole row + // incl. registration; a recycled pid must not inherit a + // dead agent's name. + gone.push(a.pid); + appeared.push(a.pid); + tx.execute("DELETE FROM agents WHERE pid=?1", params![a.pid])?; + } + } + } + + for (pid, (_, registered, gone_at)) in &known { + if current.contains(pid) { + continue; + } + if *registered { + if gone_at.is_none() { + // First disappearance: tombstone the row — runtime_ref + // stays available for cleanup of a lingering pane. + gone.push(*pid); + tx.execute( + "UPDATE agents SET gone_at=?2 WHERE pid=?1", + params![pid, now], + )?; + } + } else { + gone.push(*pid); + tx.execute("DELETE FROM agents WHERE pid=?1", params![pid])?; + } + } - let mut appeared: Vec = current.difference(&known).map(|(pid, _)| *pid).collect(); - let mut gone: Vec = known.difference(¤t).map(|(pid, _)| *pid).collect(); appeared.sort_unstable(); + appeared.dedup(); gone.sort_unstable(); + gone.dedup(); - // Delete first so a reused pid drops its stale row (first_seen resets) - // and reappears via insert rather than an in-place update. - for pid in &gone { - tx.execute("DELETE FROM agents WHERE pid=?1", params![pid])?; - } for a in agents { tx.execute( "INSERT INTO agents (pid,ppid,tty,command,cwd,session_file,start_time_ms,first_seen,last_seen) @@ -138,8 +240,8 @@ impl Store { ON CONFLICT(pid) DO UPDATE SET ppid=excluded.ppid, tty=excluded.tty, command=excluded.command, cwd=excluded.cwd, session_file=excluded.session_file, - start_time_ms=excluded.start_time_ms, - last_seen=excluded.last_seen", + start_time_ms=COALESCE(excluded.start_time_ms, agents.start_time_ms), + last_seen=excluded.last_seen, gone_at=NULL", params![a.pid, a.ppid, a.tty, a.command, a.cwd, a.session_file, a.start_time_ms, now], )?; } @@ -151,7 +253,7 @@ impl Store { pub fn list_agents(&self) -> Result> { let conn = self.conn.lock().unwrap(); let mut s = conn.prepare( - "SELECT pid,ppid,tty,command,cwd,session_file,start_time_ms,first_seen,last_seen FROM agents ORDER BY pid", + "SELECT pid,ppid,tty,command,cwd,session_file,start_time_ms,first_seen,last_seen,name,agent_type,runtime,registered_at,gone_at FROM agents ORDER BY pid", )?; let rows = s.query_map([], |r| { Ok(serde_json::json!({ @@ -164,6 +266,11 @@ impl Store { "startTime": r.get::<_, Option>(6)?, "firstSeen": r.get::<_, i64>(7)?, "lastSeen": r.get::<_, i64>(8)?, + "name": r.get::<_, Option>(9)?, + "agentType": r.get::<_, Option>(10)?, + "runtime": r.get::<_, Option>(11)?, + "registeredAt": r.get::<_, Option>(12)?, + "goneAt": r.get::<_, Option>(13)?, })) })?; rows.collect::, _>>() @@ -171,6 +278,26 @@ impl Store { } } +/// RFC3339/ISO-8601 UTC rendering of epoch ms — the `started_at` format the +/// legacy agents.db stores. Uses gmtime_r to avoid a chrono dependency. +pub fn iso8601_utc(ms: i64) -> String { + unsafe { + let secs = (ms / 1000) as libc::time_t; + let mut tm: libc::tm = std::mem::zeroed(); + libc::gmtime_r(&secs, &mut tm); + format!( + "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}.{:03}Z", + tm.tm_year + 1900, + tm.tm_mon + 1, + tm.tm_mday, + tm.tm_hour, + tm.tm_min, + tm.tm_sec, + ms.rem_euclid(1000) + ) + } +} + pub fn chrono_now() -> i64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -185,7 +312,7 @@ mod tests { use serde_json::json; use std::path::PathBuf; - fn tmp_store() -> (Store, PathBuf) { + pub(crate) fn tmp_store() -> (Store, PathBuf) { static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); let dir = std::env::temp_dir().join(format!("aidk-test-{}", std::process::id())); std::fs::create_dir_all(&dir).unwrap(); @@ -197,7 +324,7 @@ mod tests { (Store::open(&p).unwrap(), p) } - fn proc(pid: i64) -> AgentProc { + pub(crate) fn proc(pid: i64) -> AgentProc { AgentProc { pid, ppid: Some(1), @@ -209,7 +336,7 @@ mod tests { } } - fn proc_started(pid: i64, start: i64) -> AgentProc { + pub(crate) fn proc_started(pid: i64, start: i64) -> AgentProc { AgentProc { start_time_ms: Some(start), ..proc(pid) diff --git a/rust/crates/devkit-core/src/store/registry.rs b/rust/crates/devkit-core/src/store/registry.rs new file mode 100644 index 000000000..61d4e11a6 --- /dev/null +++ b/rust/crates/devkit-core/src/store/registry.rs @@ -0,0 +1,495 @@ +//! Agent registration lifecycle + legacy `agents.db` bridge — an +//! `impl Store` block living in the `store` module's child file so the +//! observed-process snapshot (`store.rs`) and the registry don't share a +//! 900-line file. Registered rows (name non-NULL) follow their own rules: +//! tombstone on disappearance, full delete on pid reuse, unique live names. + +use super::{chrono_now, iso8601_utc, Store}; +use anyhow::Result; +use rusqlite::{params, Connection}; +use serde_json::Value; +use std::path::PathBuf; + +/// One row of the legacy agents.db `agents` table. +type LegacyRow = (String, i64, String, String, String, i64, String, String); + +/// Registration fields carried into the legacy dual-write. +struct RegWrite<'a> { + pid: i64, + name: &'a str, + agent_type: &'a str, + cwd: &'a str, + runtime: &'a str, + runtime_ref: &'a Value, + now_ms: i64, +} + +/// Registry row as stored on the agents table. +#[derive(Debug, Clone)] +pub struct Registration { + pub pid: i64, + pub name: String, + pub agent_type: String, + pub runtime: String, + pub runtime_ref: String, + pub pinned: bool, + /// None while registered-but-not-yet-swept, or when tombstoned before + /// the process was ever observed. + pub start_time_ms: Option, + /// Some(_) once a sweep confirmed the process disappeared. + pub gone_at: Option, +} + +impl Store { + /// Point the store at the legacy registry db. Absence of the file is not + /// an error — dual-write/import just no-op. + pub fn set_legacy_db(&self, path: PathBuf) -> Result<()> { + *self.legacy_db.lock().unwrap() = Some(path); + Ok(()) + } + + /// Claim a pid's row with registry metadata. `start_time_ms` is left + /// untouched — NULL until a sweep adopts the real value — so the + /// snapshot diff above recognizes the row as registered-not-yet-swept. + /// Dual-writes into the legacy agents.db when one is configured. + pub fn register_start( + &self, + pid: i64, + name: &str, + agent_type: &str, + cwd: &str, + runtime: &str, + runtime_ref: &Value, + ) -> Result<()> { + let conn = self.conn.lock().unwrap(); + let now = chrono_now(); + conn.execute( + "INSERT INTO agents (pid, name, agent_type, runtime, runtime_ref, + cwd, registered_at, first_seen, last_seen) + VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?8) + ON CONFLICT(pid) DO UPDATE SET + name=excluded.name, agent_type=excluded.agent_type, + runtime=excluded.runtime, runtime_ref=excluded.runtime_ref, + registered_at=excluded.registered_at, + last_seen=excluded.last_seen, gone_at=NULL", + params![ + pid, + name, + agent_type, + runtime, + serde_json::to_string(runtime_ref)?, + cwd, + now, + now + ], + )?; + drop(conn); + self.dual_write_legacy(&RegWrite { + pid, + name, + agent_type, + cwd, + runtime, + runtime_ref, + now_ms: now, + }); + Ok(()) + } + + /// Best-effort mirror of a registration into the legacy agents.db so + /// unmigrated consumers (send/kill/rename/open, channel bridges) still + /// resolve the agent. Silent on any failure — the daemon db is + /// authoritative and the file may legitimately not exist. + fn dual_write_legacy(&self, w: &RegWrite) { + let Some(path) = self.legacy_db.lock().unwrap().clone() else { + return; + }; + if !path.exists() { + return; + } + let tmux_session = if w.runtime == "tmux" { + w.runtime_ref["session"].as_str().unwrap_or(w.name) + } else { + "" + }; + let now_iso = iso8601_utc(w.now_ms); + if let Err(e) = (|| -> Result<()> { + let conn = Connection::open(&path)?; + conn.execute( + "INSERT OR REPLACE INTO agents + (type,pid,name,tmux_session,cwd,started_at,session_id, + session_file_path,updated_at,pinned,runtime,runtime_ref) + VALUES (?1,?2,?3,?4,?5,?6,'','',?6,0,?7,?8)", + params![ + w.agent_type, + w.pid, + w.name, + tmux_session, + w.cwd, + now_iso, + w.runtime, + serde_json::to_string(w.runtime_ref)?, + ], + )?; + Ok(()) + })() { + tracing::warn!("legacy agents.db dual-write failed: {e:#}"); + } + } + + /// Delta-import legacy agents.db rows: live rows get the name attached + /// (attach-by-pid), unknown pids become dead-by-default tombstones + /// (gone_at set → no phantom disappeared for stale rows). Daemon rows + /// always win — `AND name IS NULL` / `OR IGNORE` keep them untouched. + /// + /// Attach-by-pid could mislabel a process when a legacy pid was recycled + /// between the legacy write and daemon boot — an accepted narrow race, + /// the wrong name dies with that process's next sweep. + /// Read all rows from the legacy DB. Returns None when the bridge is + /// off or the file is absent/unreadable — import is best-effort. + fn read_legacy_rows(&self) -> Result> { + let Some(path) = self.legacy_db.lock().unwrap().clone() else { + return Ok(vec![]); + }; + if !path.exists() { + return Ok(vec![]); + } + let legacy = + match Connection::open_with_flags(&path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY) { + Ok(c) => c, + Err(e) => { + tracing::warn!("legacy agents.db unreadable ({e:#}); skipping import"); + return Ok(vec![]); + } + }; + let mut stmt = match legacy + .prepare("SELECT type,pid,name,cwd,started_at,pinned,runtime,runtime_ref FROM agents") + { + Ok(s) => s, + Err(e) => { + tracing::warn!("legacy agents.db lacks agents table ({e:#}); skipping import"); + return Ok(vec![]); + } + }; + let rows = stmt + .query_map([], |r| { + Ok(( + r.get(0)?, + r.get(1)?, + r.get(2)?, + r.get(3)?, + r.get(4)?, + r.get(5)?, + r.get(6)?, + r.get(7)?, + )) + })? + .filter_map(|r| r.ok()) + .collect(); + Ok(rows) + } + + /// Merge legacy rows into the registry: attach to a live row for the + /// same pid, else insert as a tombstone (dead-by-default; the sweep + /// reanimates it if the pid is actually alive). + pub fn import_legacy_agents(&self) -> Result<()> { + let rows = self.read_legacy_rows()?; + + let conn = self.conn.lock().unwrap(); + let now = chrono_now(); + for row in rows { + let (agent_type, pid, name, cwd, _started_at, pinned, runtime, runtime_ref) = row; + // Attach to a live, unclaimed row for the same pid. + let attached = conn.execute( + "UPDATE agents SET name=?1, agent_type=?2, runtime=?3, + runtime_ref=?4, pinned=?5, registered_at=?6 + WHERE pid=?7 AND name IS NULL", + params![name, agent_type, runtime, runtime_ref, pinned, now, pid], + )?; + if attached == 0 { + conn.execute( + "INSERT OR IGNORE INTO agents + (pid, name, agent_type, runtime, runtime_ref, cwd, + pinned, registered_at, gone_at, first_seen, last_seen) + VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?8,?8,?8)", + params![ + pid, + name, + agent_type, + runtime, + runtime_ref, + cwd, + pinned, + now + ], + )?; + } + } + Ok(()) + } + + /// Look up a registered agent by name (live or tombstoned). + pub fn lookup_name(&self, name: &str) -> Result> { + let conn = self.conn.lock().unwrap(); + let mut s = conn.prepare( + "SELECT pid, name, agent_type, runtime, runtime_ref, pinned, + start_time_ms, gone_at + FROM agents WHERE name=?1", + )?; + let mut rows = s + .query_map(params![name], |r| { + Ok(Registration { + pid: r.get(0)?, + name: r.get(1)?, + agent_type: r.get::<_, Option>(2)?.unwrap_or_default(), + runtime: r.get::<_, Option>(3)?.unwrap_or_default(), + runtime_ref: r.get(4)?, + pinned: r.get::<_, i64>(5)? != 0, + start_time_ms: r.get(6)?, + gone_at: r.get(7)?, + }) + })? + .collect::, _>>()?; + Ok(rows.pop()) + } + + /// Remove a pid's registration row entirely (dead pending registration + /// cleared ahead of a fresh start). + pub fn unregister(&self, pid: i64) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute( + "DELETE FROM agents WHERE pid=?1 AND name IS NOT NULL", + params![pid], + )?; + Ok(()) + } + + /// Drop tombstoned registrations (gone_at set) — the prune half of the + /// start-time name check. + pub fn prune_dead(&self) -> Result { + let conn = self.conn.lock().unwrap(); + Ok(conn.execute( + "DELETE FROM agents WHERE name IS NOT NULL AND gone_at IS NOT NULL", + [], + )?) + } + + /// Registry overlays for live rows: pid -> (name, pinned). Tombstones + /// excluded — dead registrations don't label live observations. + pub fn registry_overlays(&self) -> Result> { + let conn = self.conn.lock().unwrap(); + let mut s = conn.prepare( + "SELECT pid, name, pinned FROM agents + WHERE name IS NOT NULL AND gone_at IS NULL", + )?; + let rows = s + .query_map([], |r| { + Ok(( + r.get::<_, i64>(0)?, + (r.get::<_, String>(1)?, r.get::<_, i64>(2)? != 0), + )) + })? + .collect::, _>>()?; + Ok(rows.into_iter().collect()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::store::tests::{proc_started, tmp_store}; + use serde_json::json; + #[test] + fn register_marks_row_without_clobbering_sweep_fields() { + let (s, _p) = tmp_store(); + s.apply_agent_snapshot(&[proc_started(10, 1000)]).unwrap(); + s.register_start( + 10, + "my-agent", + "codex", + "/tmp", + "tmux", + &json!({"session":"my-agent"}), + ) + .unwrap(); + let row = &s.list_agents().unwrap()[0]; + assert_eq!(row["name"], "my-agent"); + assert_eq!(row["agentType"], "codex"); + assert_eq!(row["runtime"], "tmux"); + assert_eq!(row["startTime"], 1000); // sweep fields preserved + assert_eq!(row["command"], "codex"); + } + + #[test] + fn register_before_sweep_survives_and_adopts_start_time() { + let (s, _p) = tmp_store(); + // Registered while the process exists but before any sweep saw it. + s.register_start(10, "my-agent", "codex", "/tmp", "tmux", &json!({})) + .unwrap(); + let (app, gone) = s.apply_agent_snapshot(&[proc_started(10, 1000)]).unwrap(); + assert_eq!(app, vec![10]); + // Adoption must not count as a disappearance. + assert!(gone.is_empty()); + let row = &s.list_agents().unwrap()[0]; + assert_eq!(row["name"], "my-agent"); + assert_eq!(row["startTime"], 1000); + } + + #[test] + fn pid_reuse_drops_registration() { + let (s, _p) = tmp_store(); + s.apply_agent_snapshot(&[proc_started(10, 1000)]).unwrap(); + s.register_start(10, "my-agent", "codex", "/tmp", "tmux", &json!({})) + .unwrap(); + // Unrelated process reuses the pid. + let (app, gone) = s.apply_agent_snapshot(&[proc_started(10, 2000)]).unwrap(); + assert_eq!(app, vec![10]); + assert_eq!(gone, vec![10]); + let row = &s.list_agents().unwrap()[0]; + assert!(row["name"].is_null()); + assert!(row["registeredAt"].is_null()); + } + + #[test] + fn disappeared_registered_agent_becomes_tombstone() { + let (s, _p) = tmp_store(); + s.apply_agent_snapshot(&[proc_started(10, 1000)]).unwrap(); + s.register_start( + 10, + "my-agent", + "codex", + "/tmp", + "herdr", + &json!({"paneId":"p1"}), + ) + .unwrap(); + let (_app, gone) = s.apply_agent_snapshot(&[]).unwrap(); + assert_eq!(gone, vec![10]); // disappeared fires once + let rows = s.list_agents().unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0]["name"], "my-agent"); // runtimeRef retained for cleanup + assert!(rows[0]["goneAt"].is_i64()); + // Steady state: no repeat disappeared events for the tombstone. + let (_a, gone2) = s.apply_agent_snapshot(&[]).unwrap(); + assert!(gone2.is_empty()); + // Unnamed vanished rows are still deleted outright. + s.apply_agent_snapshot(&[proc_started(11, 1000)]).unwrap(); + let (_a, gone3) = s.apply_agent_snapshot(&[]).unwrap(); + assert_eq!(gone3, vec![11]); + let rows = s.list_agents().unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0]["pid"], 10); + } + + #[test] + fn tombstone_frees_name_via_prune_and_reuse() { + let (s, _p) = tmp_store(); + s.apply_agent_snapshot(&[proc_started(10, 1000)]).unwrap(); + s.register_start(10, "my-agent", "codex", "/tmp", "tmux", &json!({})) + .unwrap(); + s.apply_agent_snapshot(&[]).unwrap(); + // Dead name does not block a new start once pruned. + let tombstone = s.lookup_name("my-agent").unwrap().unwrap(); + assert!(tombstone.gone_at.is_some()); + s.prune_dead().unwrap(); + s.register_start(20, "my-agent", "codex", "/tmp", "tmux", &json!({})) + .unwrap(); + assert_eq!(s.lookup_name("my-agent").unwrap().unwrap().pid, 20); + } + + #[test] + fn duplicate_live_name_is_rejected() { + let (s, _p) = tmp_store(); + s.apply_agent_snapshot(&[proc_started(10, 1000)]).unwrap(); + s.register_start(10, "my-agent", "codex", "/tmp", "tmux", &json!({})) + .unwrap(); + assert_eq!(s.lookup_name("my-agent").unwrap().unwrap().pid, 10); + // Unique index also hard-rejects a straight duplicate register. + assert!(s + .register_start(11, "my-agent", "codex", "/tmp", "tmux", &json!({})) + .is_err()); + } + + fn legacy_fixture(path: &std::path::Path) { + let _ = std::fs::remove_file(path); // tests share the fixture filename + let conn = Connection::open(path).unwrap(); + conn.execute_batch( + "CREATE TABLE agents ( + type TEXT NOT NULL, pid INTEGER NOT NULL, name TEXT NOT NULL UNIQUE, + tmux_session TEXT NOT NULL DEFAULT '', cwd TEXT NOT NULL DEFAULT '', + started_at TEXT NOT NULL, session_id TEXT NOT NULL DEFAULT '', + session_file_path TEXT NOT NULL DEFAULT '', updated_at TEXT NOT NULL, + pinned INTEGER NOT NULL DEFAULT 0, runtime TEXT NOT NULL DEFAULT 'tmux', + runtime_ref TEXT NOT NULL DEFAULT '', + PRIMARY KEY (type, pid)); + INSERT INTO agents (type,pid,name,tmux_session,cwd,started_at,updated_at,runtime,runtime_ref) + VALUES + ('claude', 500, 'old-claude', 'old-claude', '/x', '2026-10-16T00:00:00.000Z', '2026-10-16T00:00:00.000Z', 'tmux', '{\"session\":\"old-claude\"}'), + ('codex', 501, 'old-codex', '', '/y', '2026-10-16T01:00:00.000Z', '2026-10-16T01:00:00.000Z', 'herdr','{\"paneId\":\"p9\"}');", + ) + .unwrap(); + } + + #[test] + fn legacy_import_bridges_rows_as_tombstones() { + let (s, p) = tmp_store(); + let legacy = p.with_file_name(format!( + "{}.legacy.db", + p.file_name().unwrap().to_str().unwrap() + )); + legacy_fixture(&legacy); + s.set_legacy_db(legacy.clone()).unwrap(); + s.import_legacy_agents().unwrap(); + + let r = s.lookup_name("old-claude").unwrap().unwrap(); + assert_eq!(r.pid, 500); + assert_eq!(r.runtime, "tmux"); + assert!(r.gone_at.is_some()); // imported dead — no phantom appeared + let r2 = s.lookup_name("old-codex").unwrap().unwrap(); + assert_eq!(r2.runtime, "herdr"); + assert!(r2.runtime_ref.contains("p9")); + // Re-import is a no-op (delta merge), not an error. + s.import_legacy_agents().unwrap(); + } + + #[test] + fn register_dual_writes_to_legacy_db() { + let (s, p) = tmp_store(); + let legacy = p.with_file_name(format!( + "{}.legacy.db", + p.file_name().unwrap().to_str().unwrap() + )); + legacy_fixture(&legacy); + s.set_legacy_db(legacy.clone()).unwrap(); + + s.register_start( + 600, + "new-one", + "claude", + "/x", + "tmux", + &json!({"session":"new-one"}), + ) + .unwrap(); + + let conn = Connection::open(&legacy).unwrap(); + let name: String = conn + .query_row( + "SELECT name FROM agents WHERE pid=600 AND type='claude'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(name, "new-one"); + } + + #[test] + fn missing_legacy_db_skips_silently() { + let (s, p) = tmp_store(); + s.set_legacy_db(p.with_file_name("does-not-exist.db")) + .unwrap(); + s.import_legacy_agents().unwrap(); // no file → no-op + s.register_start(1, "x1", "claude", "/tmp", "tmux", &json!({})) + .unwrap(); // dual-write skipped + } +} diff --git a/rust/crates/devkit-harness/src/lib.rs b/rust/crates/devkit-harness/src/lib.rs index ba6bc04c8..a19421566 100644 --- a/rust/crates/devkit-harness/src/lib.rs +++ b/rust/crates/devkit-harness/src/lib.rs @@ -19,6 +19,7 @@ pub mod muse; pub mod opencode; pub mod pi; pub mod shared; +pub mod start_profiles; /// Build a registry with every ported adapter for `home` registered. pub fn default_registry(home: &std::path::Path) -> Registry { diff --git a/rust/crates/devkit-harness/src/start_profiles.rs b/rust/crates/devkit-harness/src/start_profiles.rs new file mode 100644 index 000000000..53f13188e --- /dev/null +++ b/rust/crates/devkit-harness/src/start_profiles.rs @@ -0,0 +1,383 @@ +//! Per-harness start profiles — the data port of TS +//! `HARNESS_RUNTIME_PROFILES` +//! (`packages/agent-manager/src/harnesses/runtimeProfiles.ts`). Spawning an +//! agent needs the launch command, the runtime kind (`herdr --kind`), and a +//! process matcher for PID resolution after spawn. + +/// Process-command matcher — data port of the TS matcher closures. Semantics +/// are copied verbatim so daemon PID detection agrees with the TS sweep. +#[derive(Debug, Clone, Copy)] +pub enum MatchSpec { + /// argv[0] basename equals the name (`matchArgv0`). + Argv0(&'static str), + /// argv[0] contains the name (`matchArgv0Name`). + Argv0Contains(&'static str), + /// Any token's basename is one of the names (`matchAnyBasename`). + AnyBasename(&'static [&'static str]), +} + +/// What a start needs beyond discovery: the command to run, the kind passed +/// to herdr, and the matcher for resolving the spawned pid. +pub struct StartProfile { + pub command: &'static str, + pub runtime_kind: &'static str, + pub discovery_aliases: &'static [&'static str], + pub matcher: MatchSpec, +} + +/// `agent.start` type → launch profile. Keys match the TS +/// `StartableAgentType` wire values. +pub fn start_profile(agent_type: &str) -> Option { + let p = |command: &'static str, + kind: &'static str, + aliases: &'static [&'static str], + matcher: MatchSpec| { + Some(StartProfile { + command, + runtime_kind: kind, + discovery_aliases: aliases, + matcher, + }) + }; + match agent_type { + "claude" => p("claude", "claude", &["claude"], MatchSpec::Argv0("claude")), + "codex" => p("codex", "codex", &["codex"], MatchSpec::Argv0("codex")), + "copilot" => p( + "copilot", + "copilot", + &["copilot"], + MatchSpec::Argv0Contains("copilot-cli"), + ), + "gemini_cli" => p( + "gemini", + "gemini", + &["gemini", "gemini_cli"], + MatchSpec::AnyBasename(&["gemini"]), + ), + "grok_cli" => p( + "grok", + "grok", + &["grok", "grok_cli"], + MatchSpec::Argv0("grok"), + ), + "antigravity_cli" => p( + "agy", + "agy", + &["agy", "antigravity", "antigravity_cli"], + MatchSpec::Argv0("agy"), + ), + "opencode" => p( + "opencode", + "opencode", + &["opencode"], + MatchSpec::Argv0("opencode"), + ), + "pi" => p("pi", "pi", &["pi"], MatchSpec::AnyBasename(&["pi"])), + // TS launches `kiro-cli` but the harness type/kind is `kiro`. + "kiro" => p( + "kiro-cli", + "kiro", + &["kiro", "kiro_cli", "kiro-cli"], + MatchSpec::AnyBasename(&["kiro-cli", "kiro"]), + ), + "devin" => p("devin", "devin", &["devin"], MatchSpec::Argv0("devin")), + "muse" => p("muse", "muse", &["muse"], MatchSpec::Argv0Contains("muse")), + _ => None, + } +} + +impl MatchSpec { + /// Whether `ps_command` (a full command line from `ps -o command=`) + /// belongs to the harness — ports the TS closures branch-for-branch. + pub fn matches(&self, ps_command: &str) -> bool { + let tokens = tokens_of(ps_command); + if tokens.is_empty() { + return false; + } + match self { + MatchSpec::Argv0(name) => { + let test = |t: &str| basename_of(t) == name.to_lowercase(); + if test(tokens[0]) { + return true; + } + may_continue_argv0(&tokens, &test) && test(executable_path(ps_command).as_str()) + } + MatchSpec::Argv0Contains(name) => { + let lower = name.to_lowercase(); + let test = |t: &str| t.to_lowercase().contains(&lower); + if test(tokens[0]) { + return true; + } + may_continue_argv0(&tokens, &test) && test(executable_path(ps_command).as_str()) + } + MatchSpec::AnyBasename(names) => { + let test = |t: &str| names.iter().any(|n| basename_of(t) == n.to_lowercase()); + if test(tokens[0]) { + return true; + } + if !tokens[1..].iter().any(|t| test(t)) { + return false; + } + if !is_absolute(tokens[0]) { + return true; + } + let (argv0, args) = split_argv0(ps_command); + test(&argv0) || args.iter().any(|a| test(a)) + } + } + } +} + +fn tokens_of(ps_command: &str) -> Vec<&str> { + ps_command.split_whitespace().collect() +} + +/// Last path segment, lowercased — mirrors `path.basename` incl. trailing +/// slashes (`/a/b/` → "b"). +fn basename_of(token: &str) -> String { + token + .trim_end_matches(['/', '\\']) + .rsplit(['/', '\\']) + .next() + .unwrap_or("") + .to_lowercase() +} + +/// `/x` or `C:\x` — the TS ABSOLUTE_PATH regex. +fn is_absolute(token: &str) -> bool { + token.starts_with('/') + || (token.len() >= 3 + && token.as_bytes()[1] == b':' + && token.as_bytes()[0].is_ascii_alphabetic() + && matches!(token.as_bytes()[2], b'/' | b'\\')) +} + +/// The absolute-argv0-may-continue helper: argv0 is a path but spaces may +/// split it across later tokens; only then is re-resolution worthwhile. +fn may_continue_argv0(tokens: &[&str], test: &dyn Fn(&str) -> bool) -> bool { + is_absolute(tokens[0]) + && tokens[1..] + .iter() + .any(|t| (t.contains('/') || t.contains('\\')) && test(t)) +} + +/// Resolve argv[0] from a `ps` command line. First token wins unless the +/// command is absolute and a longer prefix resolves to a real file — argv0 +/// may live under a directory containing spaces. Ports `executablePath`. +fn executable_path(command: &str) -> String { + // split with separators kept — mirrors /(\s+)/ semantics. + let trimmed = command.trim(); + let mut parts: Vec<&str> = Vec::new(); + let mut last = 0; + for (i, c) in trimmed.char_indices() { + if c.is_whitespace() { + parts.push(&trimmed[last..i]); + parts.push(&trimmed[i..i + c.len_utf8()]); + last = i + c.len_utf8(); + } + } + parts.push(&trimmed[last..]); + + let first = parts.first().copied().unwrap_or(""); + if parts.len() == 1 || !is_absolute(first) || is_file(first) { + return first.to_string(); + } + let mut prefix = first.to_string(); + let mut i = 2; + while i < parts.len() { + prefix.push_str(parts[i - 1]); + prefix.push_str(parts[i]); + if (parts[i].contains('/') || parts[i].contains('\\')) && is_file(&prefix) { + return prefix; + } + i += 2; + } + first.to_string() +} + +fn is_file(p: &str) -> bool { + std::fs::metadata(p).map(|m| m.is_file()).unwrap_or(false) +} + +/// (argv0, remaining tokens) once the executable prefix is resolved. +fn split_argv0(ps_command: &str) -> (String, Vec) { + let command = ps_command.trim(); + let argv0 = executable_path(command); + let args = command[argv0.len()..] + .split_whitespace() + .map(|s| s.to_string()) + .collect(); + (argv0, args) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::path::{Path, PathBuf}; + + fn profile(t: &str) -> StartProfile { + start_profile(t).unwrap() + } + + #[test] + fn keeps_launch_and_runtime_identity_together() { + let a = profile("antigravity_cli"); + assert_eq!(a.command, "agy"); + assert_eq!(a.runtime_kind, "agy"); + assert_eq!( + a.discovery_aliases, + &["agy", "antigravity", "antigravity_cli"] + ); + assert!(a.matcher.matches("/usr/local/bin/agy")); + } + + #[test] + fn recognizes_agent_commands_not_argument_text() { + assert!(profile("copilot") + .matcher + .matches("/opt/copilot-cli/1.0/copilot")); + assert!(!profile("copilot") + .matcher + .matches("node /repo/copilot-cli/script.js")); + assert!(profile("grok_cli") + .matcher + .matches("/Users/dev/.grok/bin/grok --approve")); + assert!(!profile("grok_cli") + .matcher + .matches("node /repo/grok-cli/script.js")); + assert!(profile("gemini_cli") + .matcher + .matches("node --no-warnings /usr/local/bin/gemini")); + } + + #[test] + fn profiles_the_versioned_muse_launcher() { + let mu = profile("muse"); + assert_eq!(mu.command, "muse"); + assert!(mu + .matcher + .matches("/Users/dev/.local/bin/muse-bin-1.4.4-R5419.1")); + assert!(mu.matcher.matches("/usr/local/bin/muse")); + } + + #[test] + fn kiro_launches_kiro_cli() { + let k = profile("kiro"); + assert_eq!(k.command, "kiro-cli"); + assert_eq!(k.runtime_kind, "kiro"); + } + + #[test] + fn all_startable_types_have_profiles() { + for t in [ + "claude", + "codex", + "copilot", + "gemini_cli", + "grok_cli", + "antigravity_cli", + "opencode", + "pi", + "kiro", + "devin", + "muse", + ] { + assert!(start_profile(t).is_some(), "missing profile for {t}"); + } + assert!(start_profile("nope").is_none()); + } + + // --- spaced-directory cases (tmpdir fixtures, mirrors the TS suite) --- + + fn install(root: &Path, rel: &str) -> String { + let full = root.join(rel); + std::fs::create_dir_all(full.parent().unwrap()).unwrap(); + std::fs::write(&full, "").unwrap(); + full.to_string_lossy().into_owned() + } + + fn tmpdir() -> PathBuf { + static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let d = std::env::temp_dir().join(format!( + "sp-{}-{}", + std::process::id(), + N.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + )); + std::fs::create_dir_all(&d).unwrap(); + d + } + + #[test] + fn argv0_under_spaced_directory() { + let root = tmpdir(); + let codex = install(&root, "Applications/Some App.app/Contents/Resources/codex"); + let claude = install(&root, "Users/dev/My Tools/bin/claude"); + let codex_m = profile("codex"); + let claude_m = profile("claude"); + assert!(codex_m + .matcher + .matches(&format!("{codex} --sandbox workspace-write"))); + assert!(codex_m.matcher.matches(&codex)); + assert!(claude_m.matcher.matches(&format!("{claude} --continue"))); + assert!(!claude_m + .matcher + .matches(&format!("{codex} exec --cd /repo"))); + } + + #[test] + fn argv0_by_distribution_name_under_spaced_directory() { + let root = tmpdir(); + let copilot = install(&root, "opt/Home Brew/Caskroom/copilot-cli/1.0.60/copilot"); + assert!(profile("copilot") + .matcher + .matches(&format!("{copilot} --allow-all-tools"))); + } + + #[test] + fn spaced_arguments_are_not_the_executable() { + let root = tmpdir(); + let vim = install(&root, "usr/bin/vim"); + let codex = install(&root, "Applications/Some App.app/Contents/Resources/codex"); + let claude_m = profile("claude"); + assert!(!claude_m.matcher.matches(&format!("{vim} My Notes/claude"))); + assert!(!claude_m + .matcher + .matches(&format!("{codex} exec review the Other Dir/claude"))); + assert!(!profile("copilot") + .matcher + .matches(&format!("{vim} Some Dir/copilot-cli/notes.md"))); + assert!(!claude_m.matcher.matches("/usr/bin/vim My Notes/claude")); + } + + #[test] + fn spaced_argv0_in_script_aware_matchers() { + let root = tmpdir(); + let pi = install(&root, "Users/dev/Node Tools/bin/pi"); + let kiro = install(&root, "Applications/Kiro CLI.app/Contents/MacOS/kiro-cli"); + let node = install(&root, "Applications/Dev Tools/node/bin/node"); + assert!(profile("pi").matcher.matches(&format!("{pi} --model x"))); + assert!(profile("kiro").matcher.matches(&format!("{kiro} chat"))); + assert!(profile("gemini_cli") + .matcher + .matches(&format!("{node} /Users/dev/.npm/bin/gemini --yolo"))); + assert!(!profile("pi") + .matcher + .matches(&format!("{node} /opt/pi-tools/script.js"))); + } + + #[test] + fn agent_words_inside_executable_directory_do_not_match() { + let root = tmpdir(); + let node = install(&root, "opt/Dev pi gemini/bin/node"); + assert!(!profile("pi") + .matcher + .matches(&format!("{node} /repo/script.js"))); + assert!(!profile("gemini_cli") + .matcher + .matches(&format!("{node} /repo/script.js"))); + assert!(profile("gemini_cli") + .matcher + .matches(&format!("{node} /repo/bin/gemini"))); + assert!(profile("pi").matcher.matches(&format!("{node} --title pi"))); + } +} diff --git a/rust/crates/devkitd/Cargo.toml b/rust/crates/devkitd/Cargo.toml index 25f436555..4cffc6ee1 100644 --- a/rust/crates/devkitd/Cargo.toml +++ b/rust/crates/devkitd/Cargo.toml @@ -13,3 +13,4 @@ anyhow.workspace = true libc.workspace = true tracing.workspace = true tracing-subscriber.workspace = true +serde.workspace = true diff --git a/rust/crates/devkitd/src/main.rs b/rust/crates/devkitd/src/main.rs index 7bc4720cb..28ae2c6ac 100644 --- a/rust/crates/devkitd/src/main.rs +++ b/rust/crates/devkitd/src/main.rs @@ -1,3 +1,5 @@ +mod rpc; +mod runtime; mod server; use anyhow::Result; diff --git a/rust/crates/devkitd/src/rpc.rs b/rust/crates/devkitd/src/rpc.rs new file mode 100644 index 000000000..72455791e --- /dev/null +++ b/rust/crates/devkitd/src/rpc.rs @@ -0,0 +1,31 @@ +//! RPC method implementations — one module per wire namespace +//! (`agent.*` → `rpc::agent`), one file per method inside it. + +pub mod agent; + +use serde_json::Value; + +/// Wire-facing error: code + message + structured details, mapped onto +/// `Response.code`/`details`/`error` by the dispatcher. +#[derive(Debug)] +pub struct RpcError { + pub code: &'static str, + pub message: String, + pub details: Option, +} + +impl RpcError { + pub fn new(code: &'static str, message: impl Into, details: Option) -> Self { + Self { + code, + message: message.into(), + details, + } + } + pub fn invalid(message: impl Into) -> Self { + Self::new("invalid_params", message, None) + } + pub fn internal(e: impl std::fmt::Display) -> Self { + Self::new("internal", e.to_string(), None) + } +} diff --git a/rust/crates/devkitd/src/rpc/agent.rs b/rust/crates/devkitd/src/rpc/agent.rs new file mode 100644 index 000000000..8235e31e0 --- /dev/null +++ b/rust/crates/devkitd/src/rpc/agent.rs @@ -0,0 +1,63 @@ +//! `agent.*` ops — one file per method, shared helpers here. + +pub mod start; + +/// `/^[a-z0-9][a-z0-9-]{0,62}[a-z0-9]$/` ported without the regex crate. +pub(super) fn valid_agent_name(name: &str) -> bool { + let b = name.as_bytes(); + (2..=64).contains(&b.len()) + && b.iter() + .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || *c == b'-') + && b[0] != b'-' + && b[b.len() - 1] != b'-' +} + +/// `generateAgentName` — `{basename-sanitized}-{epoch-ms base36}`. +pub(super) fn generate_agent_name(cwd: &str) -> String { + let folder = std::path::Path::new(cwd) + .file_name() + .map(|n| n.to_string_lossy().into_owned()) + .unwrap_or_default(); + let mut folder: String = folder + .to_lowercase() + .chars() + .map(|c| { + if c.is_ascii_lowercase() || c.is_ascii_digit() { + c + } else { + '-' + } + }) + .collect::() + .trim_matches('-') + .chars() + .take(50) + .collect(); + if folder.is_empty() { + folder = "agent".into(); + } + format!("{}-{}", folder, base36(devkit_core::store::chrono_now())) +} + +fn base36(mut n: i64) -> String { + const DIGITS: &[u8] = b"0123456789abcdefghijklmnopqrstuvwxyz"; + if n == 0 { + return "0".into(); + } + let mut out = Vec::new(); + while n > 0 { + out.push(DIGITS[(n % 36) as usize]); + n /= 36; + } + out.iter().rev().map(|c| *c as char).collect() +} + +/// Process liveness: exists or permission-denied (EPERM = exists, not ours). +pub(super) fn pid_alive(pid: i64) -> bool { + unsafe { + if libc::kill(pid as libc::pid_t, 0) == 0 { + return true; + } + } + std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) +} diff --git a/rust/crates/devkitd/src/rpc/agent/start.rs b/rust/crates/devkitd/src/rpc/agent/start.rs new file mode 100644 index 000000000..587cb2067 --- /dev/null +++ b/rust/crates/devkitd/src/rpc/agent/start.rs @@ -0,0 +1,304 @@ +//! `agent.start` orchestration — the daemon-side port of TS +//! `ManagedAgentRuntime.startAgent`. Validates params, resolves the harness +//! start profile, checks runtime availability, enforces name uniqueness +//! under `Daemon::start_mutex`, spawns via the selected runtime, registers +//! the agent, and emits `agent.started`. + +use super::{generate_agent_name, pid_alive, valid_agent_name}; +use crate::rpc::RpcError; +use crate::runtime::{AgentRuntime, Runtimes, StartError, StartInput}; +use crate::server::Daemon; +use devkit_core::proto::{AgentStartParams, AgentStartResult}; +use devkit_core::store::iso8601_utc; +use devkit_harness::start_profiles::start_profile; +use serde_json::json; + +pub fn run_with( + daemon: &Daemon, + params: AgentStartParams, + runtimes: &Runtimes, +) -> Result { + // --- validation ------------------------------------------------------- + let profile = start_profile(¶ms.agent_type).ok_or_else(|| { + RpcError::new( + "harness_not_startable", + format!("Unsupported agent type \"{}\".", params.agent_type), + Some(json!({"type": params.agent_type})), + ) + })?; + let provider = params.runtime.as_deref().unwrap_or("tmux"); + if !["tmux", "herdr"].contains(&provider) { + return Err(RpcError::invalid(format!( + "Unsupported runtime \"{provider}\". Supported: tmux, herdr." + ))); + } + let cwd = params.cwd.as_str(); + if !std::path::Path::new(cwd).is_dir() { + return Err(RpcError::invalid(format!( + "Directory \"{cwd}\" does not exist." + ))); + } + let name = match params.name { + Some(n) if valid_agent_name(&n) => n, + Some(n) => { + return Err(RpcError::invalid(format!( + "Invalid name \"{n}\". Use lowercase letters, digits, and hyphens only. \ + Must start and end with a letter or digit, 2–64 characters." + ))) + } + None => generate_agent_name(cwd), + }; + + let runtime: &dyn AgentRuntime = match provider { + "herdr" => &runtimes.herdr, + _ => &runtimes.tmux, + }; + + let availability = runtime.is_available(); + if let Some(reason) = availability.reason() { + let detail = availability.detail().unwrap_or_default(); + return Err(RpcError::new( + "runtime_unavailable", + detail.to_string(), + Some(json!({ + "provider": provider, + "reason": reason, + "detail": detail, + })), + )); + } + + // --- serialized start: prune + name check + spawn + register ---------- + let _start_guard = daemon.start_mutex.lock().unwrap(); + + daemon.store.prune_dead().map_err(RpcError::internal)?; + if let Some(existing) = daemon + .store + .lookup_name(&name) + .map_err(RpcError::internal)? + { + if pid_alive(existing.pid) && existing.gone_at.is_none() { + return Err(RpcError::new( + "name_in_use", + format!( + "Agent \"{name}\" is already running (PID {}). Choose a different name.", + existing.pid + ), + Some(json!({"pid": existing.pid})), + )); + } + // Dead registration (pending or tombstone) — clear it. + daemon + .store + .unregister(existing.pid) + .map_err(RpcError::internal)?; + } + + let mut warnings = Vec::new(); + if provider == "tmux" && runtimes.tmux.session_exists(&name) { + warnings.push(format!( + "tmux session \"{name}\" already exists but has no live registry entry — it will be replaced." + )); + } + + let out = runtime + .start(&StartInput { + name: &name, + cwd, + kind: profile.runtime_kind, + command: profile.command, + args: params.args.as_deref().unwrap_or(&[]), + matcher: Some(&profile.matcher), + timeout_ms: params.timeout_ms.unwrap_or(15_000), + poll_interval_ms: params.poll_interval_ms.unwrap_or(500), + }) + .map_err(|e| { + // Display is the single source of each variant's message text. + let message = e.to_string(); + match e { + StartError::Unavailable { reason, .. } => RpcError::new( + "runtime_unavailable", + message.clone(), + Some(json!({ + "provider": provider, + "reason": reason, + "detail": message, + })), + ), + StartError::PidPollTimeout { + command, + timeout_ms, + } => RpcError::new( + "pid_poll_timeout", + message, + Some(json!({"command": command, "timeoutMs": timeout_ms})), + ), + StartError::Failed { code, .. } => { + RpcError::new("internal", message, code.map(|c| json!({"runtimeCode": c}))) + } + } + })?; + + let started_at = iso8601_utc(devkit_core::store::chrono_now()); + daemon + .store + .register_start( + out.pid, + &name, + ¶ms.agent_type, + cwd, + provider, + &out.runtime_ref, + ) + .map_err(|e| { + if e.to_string().contains("UNIQUE constraint") { + return RpcError::new( + "name_in_use", + format!("Agent \"{name}\" is already running."), + Some(json!({"pid": out.pid})), + ); + } + RpcError::internal(e) + })?; + + let result = AgentStartResult { + name: name.clone(), + agent_type: params.agent_type.clone(), + pid: out.pid as u64, + runtime: provider.to_string(), + runtime_ref: out.runtime_ref.clone(), + cwd: cwd.to_string(), + started_at, + warnings, + }; + daemon.emit( + "agent.started", + serde_json::to_value(&result).unwrap_or_default(), + ); + // Reconcile now — adopt start_time, refresh enriched, emit appeared — + // rather than leaving subscribers a sweep behind. + daemon.apply_sweep(); + + Ok(result) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::runtime::testkit::FixtureDir; + use devkit_core::store::chrono_now; + use std::sync::Arc; + + fn daemon(f: &FixtureDir) -> Arc { + let data = f.0.join("data"); + Arc::new(Daemon::new(&data, data.join("d.sock"), data.clone()).unwrap()) + } + + fn runtimes(f: &FixtureDir) -> Runtimes { + Runtimes { + tmux: f.tmux(), + herdr: f.herdr(), + } + } + + /// Wire a fake tmux pane that resolves to pid 9001 running `claude`. + fn live_agent(f: &FixtureDir) { + f.set("pane_pid", "9000"); + f.set("children-9000", "9001"); + f.set("cmd-9001", "/usr/local/bin/claude"); + } + + fn params(name: Option<&str>, cwd: &str) -> AgentStartParams { + AgentStartParams { + agent_type: "claude".into(), + name: name.map(|s| s.into()), + cwd: cwd.into(), + runtime: Some("tmux".into()), + args: None, + timeout_ms: Some(5000), + poll_interval_ms: Some(1), + } + } + + #[test] + fn happy_path_registers_and_emits() { + let f = FixtureDir::new("happy"); + live_agent(&f); + let d = daemon(&f); + let r = run_with(&d, params(Some("my-agent"), "/tmp"), &runtimes(&f)).unwrap(); + assert_eq!(r.name, "my-agent"); + assert_eq!(r.pid, 9001); + assert_eq!(r.runtime_ref["session"], "my-agent"); + // Fake pid never shows up in a real ps sweep → registered row becomes + // a tombstone after the post-start apply_sweep (expected lifecycle). + let reg = d.store.lookup_name("my-agent").unwrap().unwrap(); + assert_eq!(reg.pid, 9001); + let events = d.store.events_after(0, 10).unwrap(); + assert_eq!(events[0].kind, "agent.started"); + assert_eq!(events[0].payload["name"], "my-agent"); + } + + #[test] + fn invalid_inputs_map_to_codes() { + let f = FixtureDir::new("invalid"); + let d = daemon(&f); + let e = run_with(&d, params(Some("BAD NAME"), "/tmp"), &runtimes(&f)).unwrap_err(); + assert_eq!(e.code, "invalid_params"); + let e = run_with(&d, params(Some("ok-name"), "/nope/nada"), &runtimes(&f)).unwrap_err(); + assert_eq!(e.code, "invalid_params"); + let mut p = params(Some("ok-name"), "/tmp"); + p.agent_type = "nonsense".into(); + let e = run_with(&d, p, &runtimes(&f)).unwrap_err(); + assert_eq!(e.code, "harness_not_startable"); + let mut p = params(Some("ok-name"), "/tmp"); + p.runtime = Some("zsh".into()); + let e = run_with(&d, p, &runtimes(&f)).unwrap_err(); + assert_eq!(e.code, "invalid_params"); + } + + #[test] + fn live_duplicate_name_is_rejected_with_pid() { + let f = FixtureDir::new("dup"); + let d = daemon(&f); + let self_pid = std::process::id() as i64; + d.store + .register_start(self_pid, "my-agent", "claude", "/tmp", "tmux", &json!({})) + .unwrap(); + let e = run_with(&d, params(Some("my-agent"), "/tmp"), &runtimes(&f)).unwrap_err(); + assert_eq!(e.code, "name_in_use"); + assert_eq!(e.details.unwrap()["pid"], self_pid as u64); + } + + #[test] + fn dead_pending_name_is_reclaimed() { + let f = FixtureDir::new("reclaim"); + live_agent(&f); + let d = daemon(&f); + // Registered but the pid is long dead and never swept. + d.store + .register_start(4_000_000, "my-agent", "claude", "/tmp", "tmux", &json!({})) + .unwrap(); + let r = run_with(&d, params(Some("my-agent"), "/tmp"), &runtimes(&f)).unwrap(); + assert_eq!(r.name, "my-agent"); + } + + #[test] + fn generated_name_matches_convention() { + let n = generate_agent_name("/Users/dev/My Project"); + assert!(n.starts_with("my-project-"), "got {n}"); + assert!(valid_agent_name(&n), "generated name must validate"); + } + + #[test] + fn started_event_payload_is_serializable_result() { + let f = FixtureDir::new("payload"); + live_agent(&f); + let d = daemon(&f); + run_with(&d, params(Some("p-1"), "/tmp"), &runtimes(&f)).unwrap(); + let ev = d.store.events_after(0, 10).unwrap().remove(0); + assert_eq!(ev.payload["type"], "claude"); + assert_eq!(ev.payload["runtimeRef"]["session"], "p-1"); + assert!(ev.payload["startedAt"].is_string()); + assert!(ev.ts <= chrono_now()); + } +} diff --git a/rust/crates/devkitd/src/runtime.rs b/rust/crates/devkitd/src/runtime.rs new file mode 100644 index 000000000..3436adf7f --- /dev/null +++ b/rust/crates/devkitd/src/runtime.rs @@ -0,0 +1,126 @@ +//! Interactive agent runtimes — the Rust port of +//! `packages/agent-manager/src/runtime` (TmuxAgentRuntime, +//! HerdrAgentRuntime). Each runtime knows how to check availability, spawn +//! an agent into its terminal substrate, and resolve the spawned pid. + +pub mod herdr; +#[cfg(test)] +pub(crate) mod testkit; +pub mod tmux; + +use devkit_harness::start_profiles::MatchSpec; +use serde_json::Value; + +pub use herdr::HerdrRuntime; +pub use tmux::TmuxRuntime; + +/// Runtime availability, mirroring TS `AgentRuntimeAvailability`. +/// Enum so `ok` can't disagree with the payload. +#[derive(Debug)] +pub enum Availability { + Ok, + /// Runtime binary not on PATH — wire reason "binary-missing". + Missing(String), + /// Binary exists but its backend is down — "backend-unreachable". + Unreachable(String), +} + +impl Availability { + /// Wire-facing reason string (None when available). + pub fn reason(&self) -> Option<&'static str> { + match self { + Availability::Ok => None, + Availability::Missing(_) => Some("binary-missing"), + Availability::Unreachable(_) => Some("backend-unreachable"), + } + } + /// Wire-facing detail message. + pub fn detail(&self) -> Option<&str> { + match self { + Availability::Ok => None, + Availability::Missing(d) | Availability::Unreachable(d) => Some(d), + } + } +} + +pub struct StartInput<'a> { + pub name: &'a str, + pub cwd: &'a str, + /// Harness runtime kind — `herdr --kind` value. + pub kind: &'a str, + /// Launch command (tmux sends it to the pane). + pub command: &'a str, + /// Extra argv appended after `--` (herdr only). + pub args: &'a [String], + /// Process matcher for pid resolution (tmux only). + pub matcher: Option<&'a MatchSpec>, + pub timeout_ms: u64, + pub poll_interval_ms: u64, +} + +#[derive(Debug)] +pub struct StartOutput { + pub pid: i64, + pub runtime_ref: Value, +} + +/// Failures the handler maps onto wire error codes. +#[derive(Debug)] +pub enum StartError { + /// Availability re-check inside start failed. + Unavailable { reason: String, detail: String }, + /// The pane never produced a matching pid — session cleaned up. + PidPollTimeout { command: String, timeout_ms: u64 }, + /// Runtime refused or errored (herdr code, tmux failure, ...). + Failed { + code: Option, + message: String, + }, +} + +impl std::fmt::Display for StartError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + StartError::Unavailable { detail, .. } => write!(f, "{detail}"), + StartError::PidPollTimeout { command, timeout_ms } => write!( + f, + "Agent process not found after {}s. Verify that \"{command}\" is in PATH inside the tmux environment.", + timeout_ms / 1000 + ), + StartError::Failed { message, .. } => write!(f, "{message}"), + } + } +} + +impl std::error::Error for StartError {} + +/// `Command::output` with ETXTBSY retry — exec of a just-written +/// executable (fixture scripts in tests, freshly-installed binaries on +/// overlayfs) transiently fails with os error 26 before the write settles. +pub(crate) fn spawn_output( + cmd: &mut std::process::Command, +) -> std::io::Result { + const ETXTBSY: i32 = 26; + let mut attempts = 0; + loop { + match cmd.output() { + Err(e) if e.raw_os_error() == Some(ETXTBSY) && attempts < 3 => { + attempts += 1; + std::thread::sleep(std::time::Duration::from_millis(20)); + } + other => return other, + } + } +} + +/// Runtimes bundle — injectable so tests can pass fixture-backed impls. +#[derive(Default)] +pub struct Runtimes { + pub tmux: TmuxRuntime, + pub herdr: HerdrRuntime, +} + +pub trait AgentRuntime: Send + Sync { + fn is_available(&self) -> Availability; + fn start(&self, input: &StartInput) -> Result; +} diff --git a/rust/crates/devkitd/src/runtime/herdr.rs b/rust/crates/devkitd/src/runtime/herdr.rs new file mode 100644 index 000000000..d8b696039 --- /dev/null +++ b/rust/crates/devkitd/src/runtime/herdr.rs @@ -0,0 +1,338 @@ +//! herdr runtime — port of `HerdrAgentRuntime` + `HerdrCliClient` + +//! `HerdrResponseParsers`. herdr is a CLI-orchestrated pane manager; every +//! call is `herdr ` with JSON on stdout and `{"error":{code,message}}` +//! for failures. + +use super::{AgentRuntime, Availability, StartError, StartInput, StartOutput}; +use serde_json::{json, Value}; +use std::process::Command; +use std::time::{Duration, Instant}; + +const BUSY_RETRY: Duration = Duration::from_secs(5); +const BUSY_RETRY_INTERVAL: Duration = Duration::from_millis(100); +const PROCESS_INFO_POLL: Duration = Duration::from_secs(3); +const PROCESS_INFO_POLL_INTERVAL: Duration = Duration::from_millis(100); + +pub struct HerdrRuntime { + herdr: String, +} + +impl Default for HerdrRuntime { + fn default() -> Self { + Self { + herdr: "herdr".into(), + } + } +} + +impl HerdrRuntime { + #[cfg(test)] + pub fn with_bin(herdr: &str) -> Self { + Self { + herdr: herdr.into(), + } + } + + fn run(&self, args: &[&str]) -> Result { + let out = crate::runtime::spawn_output(Command::new(&self.herdr).args(args)); + match out { + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Err(HerdrCall::NotFound), + Err(e) => Err(HerdrCall::Spawn(e.to_string())), + Ok(o) if o.status.success() => Ok(String::from_utf8_lossy(&o.stdout).to_string()), + Ok(o) => Err(HerdrCall::Failed( + String::from_utf8_lossy(&o.stderr).trim().to_string(), + )), + } + } + + /// JSON call: parse stdout, surface `{"error":{code,message}}` payloads. + fn run_json(&self, args: &[&str]) -> Result { + let stdout = self.run(args)?; + let v: Value = serde_json::from_str(&stdout) + .map_err(|e| HerdrCall::Failed(format!("Herdr returned invalid JSON: {e}")))?; + if let Some(err) = v.get("error") { + return Err(HerdrCall::Rpc { + code: err["code"].as_str().map(|s| s.to_string()), + message: err["message"] + .as_str() + .unwrap_or("Herdr command failed.") + .to_string(), + }); + } + Ok(v) + } + + fn workspace_create( + &self, + cwd: &str, + label: &str, + ) -> Result<(Option, String), HerdrCall> { + let v = self.run_json(&["workspace", "create", "--cwd", cwd, "--label", label])?; + let workspace_id = v["result"]["workspace"]["workspace_id"] + .as_str() + .map(|s| s.to_string()); + let pane_id = v["result"]["root_pane"]["pane_id"] + .as_str() + .map(|s| s.to_string()) + .ok_or_else(|| { + HerdrCall::Failed( + "Herdr workspace create response did not include root pane id.".into(), + ) + })?; + Ok((workspace_id, pane_id)) + } + + /// `agent start` with the `agent_pane_busy` retry loop — ports + /// `runAgentStartJson` (5s budget, 100ms interval). + fn agent_start(&self, input: &StartInput, pane_id: &str) -> Result { + let timeout = input.timeout_ms.to_string(); + let mut args = vec![ + "agent", + "start", + input.name, + "--kind", + input.kind, + "--pane", + pane_id, + "--timeout", + timeout.as_str(), + ]; + if !input.args.is_empty() { + args.push("--"); + for a in input.args { + args.push(a.as_str()); + } + } + let deadline = Instant::now() + BUSY_RETRY; + loop { + match self.run_json(&args) { + Err(HerdrCall::Rpc { code, message }) + if code.as_deref() == Some("agent_pane_busy") => + { + if Instant::now() >= deadline { + return Err(HerdrCall::Rpc { + code: Some("agent_pane_busy".into()), + message: if message.is_empty() { + "Herdr agent start failed because the target pane stayed busy." + .into() + } else { + message + }, + }); + } + std::thread::sleep(BUSY_RETRY_INTERVAL); + } + other => return other, + } + } + } + + /// `pane process-info` poll: foreground pid preferred, shell pid as + /// fallback — ports `readStartedAgentPid` (3s budget, 100ms interval). + fn read_started_pid(&self, pane_id: &str) -> Option { + let deadline = Instant::now() + PROCESS_INFO_POLL; + let mut fallback = None; + loop { + if let Ok(v) = self.run_json(&["pane", "process-info", "--pane", pane_id]) { + let info = &v["result"]["process_info"]; + if let Some(p) = info["foreground_processes"] + .as_array() + .and_then(|ps| ps.iter().find_map(|p| p["pid"].as_i64())) + { + return Some(p); + } + fallback = fallback.or_else(|| info["shell_pid"].as_i64()); + } + if Instant::now() >= deadline { + return fallback; + } + std::thread::sleep(PROCESS_INFO_POLL_INTERVAL); + } + } +} + +enum HerdrCall { + NotFound, + Spawn(String), + Failed(String), + Rpc { + code: Option, + message: String, + }, +} + +impl HerdrCall { + /// Human-readable failure detail for availability reporting. + fn detail(self) -> String { + match self { + HerdrCall::Rpc { message, .. } => message, + HerdrCall::Failed(m) | HerdrCall::Spawn(m) => m, + HerdrCall::NotFound => "herdr command was not found in PATH.".into(), + } + } +} + +impl From for StartError { + fn from(e: HerdrCall) -> Self { + match e { + HerdrCall::NotFound => StartError::Unavailable { + reason: "binary-missing".into(), + detail: "herdr command was not found in PATH.".into(), + }, + HerdrCall::Spawn(m) | HerdrCall::Failed(m) => StartError::Failed { + code: None, + message: m, + }, + HerdrCall::Rpc { code, message } => StartError::Failed { code, message }, + } + } +} + +impl AgentRuntime for HerdrRuntime { + fn is_available(&self) -> Availability { + match self.run(&["--version"]) { + Err(HerdrCall::NotFound) => { + return Availability::Missing("herdr command was not found in PATH.".into()) + } + Err(e) => return Availability::Unreachable(e.detail()), + Ok(_) => {} + } + match self.run(&["api", "snapshot"]) { + Ok(_) => Availability::Ok, + Err(e) => Availability::Unreachable(e.detail()), + } + } + + fn start(&self, input: &StartInput) -> Result { + let (workspace_id, pane_id) = self.workspace_create(input.cwd, input.name)?; + let agent_resp = self.agent_start(input, &pane_id)?; + + let agent = &agent_resp["result"]["agent"]; + let pane_id = agent["pane_id"].as_str().unwrap_or(&pane_id).to_string(); + let agent_name = agent["name"].as_str().unwrap_or(input.name).to_string(); + let tab_id = agent["tab_id"].as_str().map(|s| s.to_string()); + + let pid = self + .read_started_pid(&pane_id) + .ok_or_else(|| StartError::Failed { + code: None, + message: "Herdr pane process-info response did not include process pid.".into(), + })?; + + let mut runtime_ref = json!({ + "session": "default", + "paneId": pane_id, + "agentName": agent_name, + }); + if let Some(w) = workspace_id { + runtime_ref["workspaceId"] = json!(w); + } + if let Some(t) = tab_id { + runtime_ref["tabId"] = json!(t); + } + Ok(StartOutput { pid, runtime_ref }) + } +} + +#[cfg(test)] +mod tests { + use super::super::testkit::FixtureDir; + use super::*; + use devkit_harness::start_profiles::MatchSpec; + + /// Wire the canned responses every herdr start needs. + fn happy(f: &FixtureDir) { + f.set( + "workspace.json", + r#"{"result":{"workspace":{"workspace_id":"w1"},"root_pane":{"pane_id":"p-root"}}}"#, + ); + f.set( + "agent_start.json", + r#"{"result":{"agent":{"pane_id":"p-1","name":"agent-a","tab_id":"t-1"}}}"#, + ); + f.set( + "process_info.json", + r#"{"result":{"process_info":{"foreground_processes":[{"pid":4242}],"shell_pid":4000}}}"#, + ); + } + + fn input<'a>(name: &'a str, args: &'a [String]) -> StartInput<'a> { + static M: MatchSpec = MatchSpec::Argv0("claude"); + StartInput { + name, + cwd: "/repo", + kind: "claude", + command: "claude", + args, + matcher: Some(&M), + timeout_ms: 1000, + poll_interval_ms: 1, + } + } + + #[test] + fn unavailable_when_binary_missing() { + let f = FixtureDir::new("missing"); + let rt = HerdrRuntime::with_bin(&f.bin("nope")); + assert!(matches!(rt.is_available(), Availability::Missing(_))); + } + + #[test] + fn unavailable_when_backend_unreachable() { + let f = FixtureDir::new("backend"); + f.script( + "herdr-bad", + "#!/bin/sh\n[ \"$1\" = \"--version\" ] && echo ok && exit 0\nexit 1\n", + ); + let rt = HerdrRuntime::with_bin(&f.bin("herdr-bad")); + assert!(matches!(rt.is_available(), Availability::Unreachable(_))); + } + + #[test] + fn start_creates_workspace_starts_agent_reads_pid() { + let f = FixtureDir::new("happy"); + happy(&f); + let out = f.herdr().start(&input("agent-a", &[])).unwrap(); + assert_eq!(out.pid, 4242); // foreground pid preferred over shell_pid + let r = &out.runtime_ref; + assert_eq!(r["paneId"], "p-1"); + assert_eq!(r["workspaceId"], "w1"); + assert_eq!(r["tabId"], "t-1"); + assert_eq!(r["agentName"], "agent-a"); + assert_eq!(r["session"], "default"); + let calls = f.log("herdr-calls.log"); + assert!(calls.contains("workspace create --cwd /repo --label agent-a")); + assert!(calls.contains("agent start agent-a --kind claude --pane p-root --timeout 1000")); + assert!(calls.contains("pane process-info --pane p-1")); + } + + #[test] + fn extra_args_forwarded_after_double_dash() { + let f = FixtureDir::new("args"); + happy(&f); + let args = vec!["--flag".to_string(), "v".to_string()]; + f.herdr().start(&input("agent-a", &args)).unwrap(); + assert!(f.log("herdr-calls.log").contains("-- --flag v")); + } + + #[test] + fn busy_retry_recovers_then_errors() { + let f = FixtureDir::new("busy"); + happy(&f); + f.set("busy-limit", "2"); + let out = f.herdr().start(&input("agent-a", &[])).unwrap(); + assert_eq!(out.pid, 4242); + } + + #[test] + fn shell_pid_used_when_no_foreground() { + let f = FixtureDir::new("shellpid"); + happy(&f); + f.set( + "process_info.json", + r#"{"result":{"process_info":{"foreground_processes":[],"shell_pid":4321}}}"#, + ); + let out = f.herdr().start(&input("agent-a", &[])).unwrap(); + assert_eq!(out.pid, 4321); + } +} diff --git a/rust/crates/devkitd/src/runtime/testkit.rs b/rust/crates/devkitd/src/runtime/testkit.rs new file mode 100644 index 000000000..b8ca6ac0a --- /dev/null +++ b/rust/crates/devkitd/src/runtime/testkit.rs @@ -0,0 +1,117 @@ +//! Shared fake-CLI fixtures for runtime/agent_start tests: executable +//! shell scripts in a temp dir that append argv to a log file and read +//! canned responses from sibling state files. + +use super::{HerdrRuntime, TmuxRuntime}; +use std::path::PathBuf; + +/// Temp dir holding fake CLI binaries driven by sibling state files. +pub struct FixtureDir(pub PathBuf); + +impl FixtureDir { + pub fn new(tag: &str) -> Self { + static N: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let dir = std::env::temp_dir().join(format!( + "devkitd-fix-{}-{}-{}", + std::process::id(), + tag, + N.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + FixtureDir(dir) + } + + /// Write an executable script into the dir (chmod 755 on unix). + pub fn script(&self, name: &str, content: &str) -> PathBuf { + let p = self.0.join(name); + std::fs::write(&p, content).unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&p, std::fs::Permissions::from_mode(0o755)).unwrap(); + } + p + } + + /// Write a plain state file a fixture script reads. + pub fn set(&self, name: &str, content: &str) { + std::fs::write(self.0.join(name), content).unwrap(); + } + + /// Read a script's argv log — empty string until it records a call. + pub fn log(&self, name: &str) -> String { + std::fs::read_to_string(self.0.join(name)).unwrap_or_default() + } + + /// Absolute path of a file in the dir (binary injection into runtimes). + pub fn bin(&self, name: &str) -> String { + self.0.join(name).to_str().unwrap().to_string() + } + + /// tmux runtime backed by fixture tmux/pgrep/ps scripts. + pub fn tmux(&self) -> TmuxRuntime { + self.script("tmux", TMUX_SCRIPT); + self.script("pgrep", PGREP_SCRIPT); + self.script("ps", PS_SCRIPT); + TmuxRuntime::with_bins(&self.bin("tmux"), &self.bin("pgrep"), &self.bin("ps")) + } + + /// herdr runtime backed by a fixture herdr script. + pub fn herdr(&self) -> HerdrRuntime { + self.script("herdr", HERDR_SCRIPT); + HerdrRuntime::with_bin(&self.bin("herdr")) + } +} + +/// Fake tmux: `-V`, `has-session`, `new-session`, `send-keys`, `list-panes` +/// (reads `pane_pid`), `kill-session`. Sessions tracked as `session-` +/// marker files; every invocation logs argv to `tmux-calls.log`. +const TMUX_SCRIPT: &str = r#"#!/bin/sh +DIR="$(dirname "$0")" +echo "$@" >> "$DIR/tmux-calls.log" +case "$1" in + -V) echo "tmux 3.4";; + has-session) [ -f "$DIR/session-$3" ];; + new-session) + shift; while [ $# -gt 0 ]; do + case "$1" in -s) touch "$DIR/session-$2"; shift 2;; *) shift;; esac + done;; + send-keys) :;; + list-panes) cat "$DIR/pane_pid" 2>/dev/null;; + kill-session) rm -f "$DIR/session-$3";; +esac +exit 0 +"#; + +/// Fake pgrep: `-P ` reads `children-` (one pid per line). +const PGREP_SCRIPT: &str = + "#!/bin/sh\nDIR=\"$(dirname \"$0\")\"\ncat \"$DIR/children-$2\" 2>/dev/null\nexit 0\n"; + +/// Fake ps: `-p -o command=` reads `cmd-`. +const PS_SCRIPT: &str = + "#!/bin/sh\nDIR=\"$(dirname \"$0\")\"\ncat \"$DIR/cmd-$2\" 2>/dev/null\nexit 0\n"; + +/// Fake herdr: `--version`, `api snapshot`, `workspace create`, +/// `agent start`, `pane process-info`. JSON bodies come from sibling +/// files; `busy-limit`/`busy-count` drive the agent_pane_busy loop. +const HERDR_SCRIPT: &str = r#"#!/bin/sh +DIR="$(dirname "$0")" +echo "$@" >> "$DIR/herdr-calls.log" +case "$1 $2" in + "--version ") echo "herdr 1.0";; + "api snapshot") cat "$DIR/snapshot.json" 2>/dev/null || echo "{}";; + "workspace create") cat "$DIR/workspace.json";; + "agent start") + C=$(cat "$DIR/busy-count" 2>/dev/null || echo 0) + echo $((C+1)) > "$DIR/busy-count" + LIMIT=$(cat "$DIR/busy-limit" 2>/dev/null || echo 0) + if [ "$C" -lt "$LIMIT" ]; then + echo '{"error":{"code":"agent_pane_busy","message":"pane busy"}}' + else + cat "$DIR/agent_start.json" + fi;; + "pane process-info") cat "$DIR/process_info.json";; +esac +exit 0 +"#; diff --git a/rust/crates/devkitd/src/runtime/tmux.rs b/rust/crates/devkitd/src/runtime/tmux.rs new file mode 100644 index 000000000..f4c9b479d --- /dev/null +++ b/rust/crates/devkitd/src/runtime/tmux.rs @@ -0,0 +1,275 @@ +//! tmux runtime — port of `TmuxManager`/`TmuxAgentRuntime`/`PidPolling`. +//! Spawns by creating a detached session and sending the launch command to +//! the pane, then resolves the agent pid by walking the pane's process +//! descendants until the command line matches the harness profile. + +use super::{AgentRuntime, Availability, StartError, StartInput, StartOutput}; +use devkit_harness::start_profiles::MatchSpec; +use serde_json::json; +use std::process::Command; +use std::time::{Duration, Instant}; + +/// Same-pid observations required before a pid is trusted — mirrors +/// `REQUIRED_STABLE_PID_POLLS`. +const REQUIRED_STABLE_POLLS: u32 = 5; + +/// Binary paths are injectable so tests can point at fixture scripts instead +/// of real tmux/pgrep/ps. +pub struct TmuxRuntime { + tmux: String, + pgrep: String, + ps: String, +} + +impl Default for TmuxRuntime { + fn default() -> Self { + Self { + tmux: "tmux".into(), + pgrep: "pgrep".into(), + ps: "ps".into(), + } + } +} + +impl TmuxRuntime { + #[cfg(test)] + pub fn with_bins(tmux: &str, pgrep: &str, ps: &str) -> Self { + Self { + tmux: tmux.into(), + pgrep: pgrep.into(), + ps: ps.into(), + } + } + + fn run(&self, args: &[&str]) -> Result { + crate::runtime::spawn_output(Command::new(&self.tmux).args(args)) + .map_err(|e| e.to_string()) + .and_then(|o| { + if o.status.success() { + Ok(String::from_utf8_lossy(&o.stdout).trim().to_string()) + } else { + Err(String::from_utf8_lossy(&o.stderr).trim().to_string()) + } + }) + } + + pub fn session_exists(&self, name: &str) -> bool { + self.run(&["has-session", "-t", name]).is_ok() + } + + fn kill_session(&self, name: &str) { + let _ = self.run(&["kill-session", "-t", name]); + } + + fn pane_pid(&self, session: &str) -> Option { + let out = self + .run(&["list-panes", "-t", session, "-F", "#{pane_pid}"]) + .ok()?; + out.lines().next()?.trim().parse().ok() + } + + fn pgrep_children(&self, pid: i64) -> Vec { + crate::runtime::spawn_output(Command::new(&self.pgrep).args(["-P", &pid.to_string()])) + .ok() + .map(|o| { + String::from_utf8_lossy(&o.stdout) + .lines() + .filter_map(|l| l.trim().parse().ok()) + .collect() + }) + .unwrap_or_default() + } + + fn process_command(&self, pid: i64) -> Option { + let o = crate::runtime::spawn_output(Command::new(&self.ps).args([ + "-p", + &pid.to_string(), + "-o", + "command=", + ])) + .ok()?; + let s = String::from_utf8_lossy(&o.stdout).trim().to_string(); + if s.is_empty() { + None + } else { + Some(s) + } + } + + /// Deepest descendant of the pane whose command line matches — mirrors + /// `TmuxManager.findAgentPid` (BFS, deepest match wins). + fn find_agent_pid(&self, session: &str, matcher: &MatchSpec) -> Option { + let pane = self.pane_pid(session)?; + let mut queue = vec![pane]; + let mut deepest: Option = None; + while let Some(pid) = queue.pop() { + queue.extend(self.pgrep_children(pid)); + if let Some(cmd) = self.process_command(pid) { + if matcher.matches(&cmd) { + deepest = Some(pid); + } + } + } + deepest + } + + /// `pollForPid` — same candidate must be seen on + /// [`REQUIRED_STABLE_POLLS`] consecutive polls. + fn poll_for_pid( + &self, + session: &str, + matcher: &MatchSpec, + interval: Duration, + timeout: Duration, + ) -> Option { + let deadline = Instant::now() + timeout; + let mut candidate: Option = None; + let mut stable = 0u32; + while Instant::now() < deadline { + let pid = self.find_agent_pid(session, matcher); + if let Some(pid) = pid { + if candidate == Some(pid) { + stable += 1; + } else { + candidate = Some(pid); + stable = 1; + } + if stable >= REQUIRED_STABLE_POLLS { + return Some(pid); + } + } + std::thread::sleep(interval); + } + None + } +} + +impl AgentRuntime for TmuxRuntime { + fn is_available(&self) -> Availability { + match self.run(&["-V"]) { + Ok(_) => Availability::Ok, + Err(_) => Availability::Missing("tmux is not installed or not in PATH.".into()), + } + } + + fn start(&self, input: &StartInput) -> Result { + let matcher = input.matcher.ok_or_else(|| StartError::Failed { + code: None, + message: "tmux start requires a process matcher".into(), + })?; + + if self.session_exists(input.name) { + self.kill_session(input.name); + } + self.run(&["new-session", "-d", "-s", input.name, "-c", input.cwd]) + .map_err(|e| StartError::Failed { + code: None, + message: format!("tmux new-session failed: {e}"), + })?; + self.run(&["send-keys", "-t", input.name, input.command, "Enter"]) + .map_err(|e| StartError::Failed { + code: None, + message: format!("tmux send-keys failed: {e}"), + })?; + + let pid = self.poll_for_pid( + input.name, + matcher, + Duration::from_millis(input.poll_interval_ms), + Duration::from_millis(input.timeout_ms), + ); + match pid { + Some(pid) => Ok(StartOutput { + pid, + runtime_ref: json!({"session": input.name}), + }), + None => { + self.kill_session(input.name); + Err(StartError::PidPollTimeout { + command: input.command.to_string(), + timeout_ms: input.timeout_ms, + }) + } + } + } +} + +#[cfg(test)] +mod tests { + use super::super::testkit::FixtureDir; + use super::*; + use devkit_harness::start_profiles::MatchSpec; + + fn input<'a>(name: &'a str, matcher: &'a MatchSpec, timeout: u64) -> StartInput<'a> { + StartInput { + name, + cwd: "/tmp", + kind: "claude", + command: "claude", + args: &[], + matcher: Some(matcher), + timeout_ms: timeout, + poll_interval_ms: 1, + } + } + + #[test] + fn availability_follows_tmux_dash_v() { + let f = FixtureDir::new("avail"); + assert!(matches!(f.tmux().is_available(), Availability::Ok)); + let missing = TmuxRuntime::with_bins(&f.bin("nope"), "pgrep", "ps"); + assert!(matches!(missing.is_available(), Availability::Missing(_))); + } + + #[test] + fn start_creates_session_sends_command_resolves_pid() { + let f = FixtureDir::new("start"); + f.set("pane_pid", "9000"); + f.set("children-9000", "9001"); + f.set("cmd-9001", "/usr/local/bin/claude"); + let matcher = MatchSpec::Argv0("claude"); + let out = f.tmux().start(&input("agent-one", &matcher, 5000)).unwrap(); + assert_eq!(out.pid, 9001); + assert_eq!(out.runtime_ref["session"], "agent-one"); + let calls = f.log("tmux-calls.log"); + assert!(calls.contains("new-session -d -s agent-one -c /tmp")); + assert!(calls.contains("send-keys -t agent-one claude Enter")); + } + + #[test] + fn existing_session_is_replaced() { + let f = FixtureDir::new("replace"); + f.set("session-stale", ""); + assert!(f.tmux().session_exists("stale")); + f.set("pane_pid", "9000"); + f.set("cmd-9000", "/usr/bin/claude"); + let matcher = MatchSpec::Argv0("claude"); + f.tmux().start(&input("stale", &matcher, 5000)).unwrap(); + assert!(f.log("tmux-calls.log").contains("kill-session -t stale")); + } + + #[test] + fn poll_requires_stable_pid_then_times_out_clean() { + let f = FixtureDir::new("timeout"); + f.set("pane_pid", "9000"); + // No matching descendants — must time out and kill the session. + let matcher = MatchSpec::Argv0("claude"); + let err = f.tmux().start(&input("doomed", &matcher, 30)).unwrap_err(); + assert!(matches!(err, StartError::PidPollTimeout { .. })); + assert!(f.log("tmux-calls.log").contains("kill-session -t doomed")); + } + + #[test] + fn deepest_descendant_match_wins() { + let f = FixtureDir::new("deep"); + f.set("pane_pid", "9000"); + f.set("children-9000", "9001\n9002"); + f.set("children-9001", "9003"); + f.set("cmd-9001", "/bin/zsh"); + f.set("cmd-9002", "/bin/zsh"); + f.set("cmd-9003", "/usr/bin/claude --continue"); + let matcher = MatchSpec::Argv0("claude"); + let out = f.tmux().start(&input("deep", &matcher, 5000)).unwrap(); + assert_eq!(out.pid, 9003); + } +} diff --git a/rust/crates/devkitd/src/server.rs b/rust/crates/devkitd/src/server.rs index b4efed2fe..249ff3b38 100644 --- a/rust/crates/devkitd/src/server.rs +++ b/rust/crates/devkitd/src/server.rs @@ -26,6 +26,8 @@ pub struct Daemon { pub enriched: RwLock>, /// User home dir adapters resolve session trees (~/.claude, …) against. pub home: PathBuf, + /// Serializes agent.start so name-uniqueness can't race. + pub start_mutex: std::sync::Mutex<()>, } impl Daemon { @@ -54,8 +56,16 @@ impl Daemon { } }; let (tx, _) = broadcast::channel(1024); + let store = Arc::new(store); + // Legacy agents.db bridge: adopt pre-migration registrations as + // tombstones, then dual-write future starts back so unmigrated TS + // consumers keep resolving agents. + store.set_legacy_db(data_dir.join("agents.db"))?; + if let Err(e) = store.import_legacy_agents() { + tracing::warn!("legacy agents.db import failed: {e:#}"); + } Ok(Self { - store: Arc::new(store), + store, events: tx, emit_mutex: std::sync::Mutex::new(()), started_at: chrono_now(), @@ -63,10 +73,11 @@ impl Daemon { enricher: devkit_harness::default_registry(&home), enriched: RwLock::new(Vec::new()), home, + start_mutex: std::sync::Mutex::new(()), }) } - fn emit(&self, kind: &str, payload: Value) { + pub(crate) fn emit(&self, kind: &str, payload: Value) { let _g = self.emit_mutex.lock().unwrap(); match self.store.emit(kind, &payload) { Ok(seq) => { @@ -100,6 +111,25 @@ impl Daemon { let cached = self.enriched.read().unwrap().clone(); Response::ok(id, serde_json::to_value(cached).unwrap()) } + "agent.start" => { + let params = match Self::parse_params::( + &req, + "agent.start", + ) { + Ok(p) => p, + Err(resp) => return *resp, + }; + match crate::rpc::agent::start::run_with( + self, + params, + &crate::runtime::Runtimes::default(), + ) { + Ok(result) => { + Response::ok(id, serde_json::to_value(result).unwrap_or_default()) + } + Err(e) => Response::err_coded(id, e.code, e.message, e.details), + } + } "shutdown" => { // Caller expects a response before exit; schedule exit. tokio::spawn(async { @@ -112,6 +142,23 @@ impl Daemon { } } + /// Parse `req.params` into `T`, mapping failure onto an + /// `invalid_params` coded response. Every method arm uses this — + /// keeps the serde error text uniform. + fn parse_params(req: &Request, method: &str) -> Result> + where + T: serde::de::DeserializeOwned, + { + serde_json::from_value(req.params.clone()).map_err(|e| { + Box::new(Response::err_coded( + req.id, + "invalid_params", + format!("invalid {method} params: {e}"), + None, + )) + }) + } + /// Apply a discovery sweep, refresh the enriched cache, and emit /// appear/vanish events. /// @@ -148,7 +195,18 @@ impl Daemon { }; // Compute outside the lock: `*w = f()` evaluates the write guard first // and would hold it for the entire enrich pass. - let enriched = self.enricher.enrich(&ctx); + let mut enriched = self.enricher.enrich(&ctx); + // Registry overlay: daemon-registered names/pins win over adapter- + // derived labels. Tombstones excluded (dead rows don't label live + // observations). + if let Ok(overlays) = self.store.registry_overlays() { + for a in enriched.iter_mut() { + if let Some((name, pinned)) = overlays.get(&(a.pid as i64)) { + a.name = name.clone(); + a.pinned = Some(*pinned); + } + } + } // Keep the previous list: a disappeared pid is absent from the fresh // one, so its event payload carries last-known agent info. let prev = std::mem::replace(&mut *self.enriched.write().unwrap(), enriched);