From 8b4750b880acb904be1c3be42b14c04ddcd2ee27 Mon Sep 17 00:00:00 2001 From: Paul Logan Date: Fri, 17 Jul 2026 23:25:55 -0700 Subject: [PATCH 1/6] docs: design bounded session lifecycle --- SESSION_LOG_2026_07_17.md | 77 +++++++++ ...26-07-17-multi-session-daemon-lifecycle.md | 163 ++++++++++++++++++ ...7-multi-session-daemon-lifecycle-design.md | 130 ++++++++++++++ 3 files changed, 370 insertions(+) create mode 100644 SESSION_LOG_2026_07_17.md create mode 100644 docs/superpowers/plans/2026-07-17-multi-session-daemon-lifecycle.md create mode 100644 docs/superpowers/specs/2026-07-17-multi-session-daemon-lifecycle-design.md diff --git a/SESSION_LOG_2026_07_17.md b/SESSION_LOG_2026_07_17.md new file mode 100644 index 0000000..7cfac2f --- /dev/null +++ b/SESSION_LOG_2026_07_17.md @@ -0,0 +1,77 @@ +# Session log — 2026-07-17 + +## Objective + +Fix Wire multi-session daemon and identity failure modes while preserving the +live launchd-managed installation. Work occurs on +`fix/multi-session-daemon-lifecycle` in an isolated worktree from `main`. + +## Safety boundary + +No service install/reinstall, `wire upgrade`, daemon hand-start, launchd +restart, wildcard process kill, dotfile edit, or live-home rewrite. Live checks +are read-only and redact session keys, relay tokens, private keys, and raw +environment values. Tests use temporary forced homes. + +## Baseline evidence + +Captured before source edits: + +- `wire service status`: daemon launchd unit loaded from + `~/Library/LaunchAgents/sh.slancha.wire.daemon.plist`. +- `launchctl print`: `/Users/laul_pogan/.cargo/bin/wire daemon + --all-sessions --interval 5`; supervisor PID owned 495 children. +- Process snapshot: 559 `wire daemon` processes, aggregate 4,538,096 KiB RSS; + 556 launched from `.cargo/bin`, 3 from `.local/bin`; 24 children already + reparented to PID 1. +- Both on-disk binaries report Wire 0.17.0, so same-version duplicates exist. +- Session root: 677 by-key homes, all 677 containing private keys; 78 retired; + 676 with sync timestamps in seven days and 592 in one day. +- Pidfiles: 565 live daemon PIDs; recorded versions include 84 at 0.16.0 and + 593 at 0.17.0. Recorded binary paths include 673 `.cargo/bin` and 4 + `.local/bin`. Version-skewed and same-version processes therefore coexist. +- MCP pidfiles: 167 total, 41 live. +- Local relay TCP `127.0.0.1:8771`: connection refused. launchd local-relay + label absent. +- Daemon log: 1,540,032 lines, including 451,421 timeout matches and 330,475 + connection-refused matches. +- Current identity reports schema v3.2, valid suffixed DID, and session source + `override`; no raw override value was printed. +- Baseline isolated `cargo test --lib`: 645 passed, 0 failed, 1 ignored. + +## Root causes + +1. `supervisor_eligible` gives every private-key home permanent eligibility. + Since every historical home is initialized, the idle filter is bypassed. +2. `fs_last_active` reads daemon-written sync files. Spawned workers refresh + their own activity signal, making historical homes self-perpetuating. +3. Supervisor has backoff but no hard worker cap or queue. Restart adopts live + pidfiles without retiring now-ineligible legacy workers. +4. MCP unconditionally calls per-home daemon startup even when an all-session + supervisor exists, bypassing global orchestration. +5. Service renderers hardcode interval 5 and reinstall from scratch. +6. Doctor treats large all-session fan-out as legitimate, lacks lease/home, + override collision, PATH/service shadow, stale MCP, and local-relay service + checks, and includes a mutating endpoint-heal check. +7. Local-sister dial writes caller trust/relay state before its first relay POST + and returns after `pair_drop`, leaving verification dependent on later daemon + pull/ack timing. + +## Decision + +Use persisted live-owner leases and a capped fork supervisor. Default cap 16; +MCP lease TTL 90 seconds with 30-second heartbeat. Keep process isolation for +now; a single multiplexed process requires converting process-global +`WIRE_HOME` APIs and belongs in a follow-up architecture change. Preserve +identity-bearing homes; prune only safe husks. Classify operator configuration +separately in doctor. Do not edit dotfiles; emit a precise handoff for the +literal Codex override. + +## Artifacts + +- `docs/superpowers/specs/2026-07-17-multi-session-daemon-lifecycle-design.md` + — selected architecture and invariants. +- `docs/superpowers/plans/2026-07-17-multi-session-daemon-lifecycle.md` — TDD + implementation and verification plan. +- `SESSION_LOG_2026_07_17.md` — durable evidence, decisions, commands, and final + results. diff --git a/docs/superpowers/plans/2026-07-17-multi-session-daemon-lifecycle.md b/docs/superpowers/plans/2026-07-17-multi-session-daemon-lifecycle.md new file mode 100644 index 0000000..a033e5d --- /dev/null +++ b/docs/superpowers/plans/2026-07-17-multi-session-daemon-lifecycle.md @@ -0,0 +1,163 @@ +# Multi-session daemon lifecycle implementation plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Bound multi-session service work using persisted leases, diagnose the +observed machine failures, preserve service tuning, and make local pairing +converge without manual pulls. + +**Architecture:** A new lifecycle module owns lease persistence and pure +eligibility classification. The existing fork supervisor consumes that model, +selects at most a configured cap, and exposes its queue. Existing relay and +pair state machines remain authoritative; pairing adds preflight and drives both +pull halves synchronously. + +**Tech Stack:** Rust 2024, clap, serde/serde_json, fs2 locks, existing blocking +relay client, launchd/systemd/Task Scheduler renderers, Cargo tests and shell CI. + +## Global constraints + +- All subprocess tests using temporary Wire homes set both `WIRE_HOME=` + and `WIRE_HOME_FORCE=1` and remove inherited session IDs. +- No test or probe writes the live Wire root. +- No launchd restart, service install, daemon hand-start, upgrade, wildcard + kill, or live-home rewrite. +- Automatic pruning never deletes a private key or identity-bearing home. +- Default worker cap: 16. Default MCP lease TTL: 90 seconds. Heartbeat: 30 + seconds. +- Branch is pushed but never merged. + +--- + +### Task 1: Persisted lifecycle leases + +**Files:** +- Create: `src/session_lifecycle.rs` +- Modify: `src/lib.rs` +- Modify: `src/mcp.rs` +- Test: `src/session_lifecycle.rs` + +**Interfaces:** +- Produces `LeaseGuard::acquire(role)`, `LeaseGuard::heartbeat()`, + `active_leases_at(home, now)`, `prune_expired_leases_at(home, now)`, and + `classify_home(home, session, now)`. +- Lease JSON stores schema, role, PID, heartbeat/expiry RFC3339 timestamps, + version, executable path, and `session_source`; no raw key. + +- [ ] Add tests for acquisition, heartbeat, clean removal, crash/expiry, + restart reads, dead PID, malformed records, and no secret field; run the + lifecycle test filter and observe missing-symbol failures. +- [ ] Implement atomic per-PID lease writes and conservative reads/pruning. +- [ ] Acquire before MCP daemon orchestration; heartbeat on the existing watcher + cadence and drop on clean shutdown. +- [ ] Run lifecycle and MCP unit filters; commit `Add restart-safe session leases`. + +### Task 2: Bounded supervisor plan and recovery + +**Files:** +- Modify: `src/daemon_supervisor.rs` +- Modify: `src/cli/mod.rs` +- Modify: `src/cli/relay.rs` +- Modify: `src/ensure_up.rs` +- Test: `src/daemon_supervisor.rs` + +**Interfaces:** +- `run_supervisor(interval_secs, max_workers, as_json)`. +- `SupervisorPlan { selected, queued, inactive, retired }` from a pure planner. +- `SupervisorState` exposes `max_workers`, counts, queued sessions, worker + versions, and binary paths. + +- [ ] Add failing table tests for 1, 10, and 655 homes; 650 stale; cap 16; + deterministic queue; retired/missing/expired lease; pending outbox; restart; + crashed child backoff; externally running workers counting toward cap. +- [ ] Remove private-key and daemon-sync eligibility. Add planner and cap. +- [ ] Prevent MCP bypass spawn when supervisor is alive. +- [ ] Add precise validation before targeted stop of stale/overflow daemon + pidfiles; never wildcard-kill. +- [ ] Run supervisor/ensure-up filters and commit `Bound all-session daemon workers`. + +### Task 3: Preserve service limits + +**Files:** +- Modify: `src/service.rs` +- Modify: `src/cli/mod.rs` +- Modify: `src/cli/upgrade.rs` +- Test: `src/service.rs` + +**Interfaces:** +- `DaemonServiceOptions { interval_secs, max_workers }`. +- `wire service install [--interval N] [--max-workers N]`. + +- [ ] Add failing render/parse tests for defaults, explicit values, and + preservation from existing launchd, systemd, and Task Scheduler text. +- [ ] Thread optional CLI overrides into installer and generate both flags on + every platform. +- [ ] Report effective options without changing local-relay installation. +- [ ] Run service and CLI parsing tests; commit `Preserve daemon service limits`. + +### Task 4: Read-only doctor failure classification + +**Files:** +- Modify: `src/cli/status.rs` +- Modify: `src/daemon_supervisor.rs` +- Modify: `src/session.rs` +- Modify: `src/service.rs` +- Modify: `src/platform.rs` +- Test: module tests plus `tests/cli.rs` + +**Interfaces:** +- `DoctorCheck.classification` is one of `wire_defect`, `operator_config`, + `runtime_health`. +- Pure verdict helpers accept process/session/service snapshots. + +- [ ] Add failing fixtures for 655-home fan-out, duplicate override home, + distinct homes, stale MCP pid, PATH/service shadow, same-version duplicates, + version skew, local relay down/absent, and healthy state. +- [ ] Reuse home collision detection; never inspect or render raw session IDs. +- [ ] Add safe executable/PATH/service snapshot helpers and supervisor lifecycle + counts. +- [ ] Replace doctor endpoint healing with a read-only verdict. +- [ ] Run doctor/CLI tests and commit `Diagnose multi-session failure modes`. + +### Task 5: Relay preflight and bilateral local pairing + +**Files:** +- Modify: `src/cli/pairing.rs` +- Modify: `src/relay_client.rs` +- Modify: `src/send.rs` or its direct-delivery caller +- Test: `tests/stress_within_system.rs` +- Test: `tests/cli.rs` + +**Interfaces:** +- A relay preflight returns a classified actionable error for an unavailable + local endpoint. +- `add_local_sister_core` returns only after both effective tiers verify. + +- [ ] Add failing relay-unavailable test proving trust/relay files unchanged. +- [ ] Add failing bilateral test: one dial, both tiers `VERIFIED`, direct send + delivered without manual pull. +- [ ] Preflight before mutation; preserve normal multi-endpoint failover. +- [ ] Run sister pull child with `WIRE_HOME_FORCE=1`, then caller pull; verify + signed ack through existing state machine. +- [ ] Run pairing/send focused tests; commit `Converge local sister pairing`. + +### Task 6: Isolation audit and complete verification + +**Files:** +- Modify only test helpers missing `WIRE_HOME_FORCE=1` +- Update: `SESSION_LOG_2026_07_17.md` + +- [ ] Audit every `Command::new`/`assert_cmd` Wire spawn and add forced-home + isolation where missing; run affected tests serially. +- [ ] Run `cargo fmt --all -- --check`, focused suites, and + `cargo clippy --all-targets -- -D warnings`. +- [ ] Build a 655-home isolated fixture, run the supervisor briefly, and record + supervisor/worker process count and RSS; assert worker count never exceeds 16. +- [ ] Run `test-env/run.sh`. +- [ ] Run GitNexus `detect-changes --scope compare --base-ref main`; inspect all + affected flows and actual diff. +- [ ] Run the branch binary's read-only doctor against the live host and record + detected actual classes without exposing secrets. +- [ ] Run one independent read-only semantic review; fix BLOCKER/MAJOR findings + and rerun deterministic checks. +- [ ] Commit final log/test audit, push branch, and do not merge. diff --git a/docs/superpowers/specs/2026-07-17-multi-session-daemon-lifecycle-design.md b/docs/superpowers/specs/2026-07-17-multi-session-daemon-lifecycle-design.md new file mode 100644 index 0000000..c344720 --- /dev/null +++ b/docs/superpowers/specs/2026-07-17-multi-session-daemon-lifecycle-design.md @@ -0,0 +1,130 @@ +# Multi-session daemon lifecycle design + +## Goal + +Keep Wire responsive across hundreds of historical session homes without +reactivating stale identities, and make identity, binary, relay, and pairing +failures explicit and recoverable. + +## Constraints + +- Never use a private key, daemon-written sync timestamp, or historical home + existence as proof that a session is live. +- Keep launchd-managed production state untouched during development and test + only under forced, temporary `WIRE_HOME` roots. +- Preserve keys, inboxes, outboxes, and trust state. Automatic pruning may + delete only identity-less, unbound, inactive husks. +- Bound worker count even when every discovered home is eligible. +- Keep service configuration stable across reinstall unless the operator passes + an explicit replacement value. +- Diagnose raw session-key collisions without printing session-key values, + relay tokens, private keys, or full process environments. + +## Considered approaches + +1. **Persisted leases plus capped worker processes (selected).** Keep current + process isolation, replace identity-based eligibility with leases and + deliberate registry bindings, cap workers, and expose overflow as a queue. + This is compatible with existing global `WIRE_HOME` code and can ship with + surgical changes. +2. **One multiplexed daemon process.** Pass an explicit home/context through all + config, relay, trust, and cursor APIs and service every session from one + runtime. Best long-term RSS, but the present code relies on process-global + `WIRE_HOME`; converting it safely is a separate architectural project. +3. **One launchd job per session.** Let launchd own every worker. This moves but + does not solve stale eligibility, creates hundreds of service definitions, + and makes cross-platform parity worse. + +## Lifecycle model + +Each live long-running session owner writes a versioned lease under its own +`state/wire/leases/` directory. A lease contains role, PID, heartbeat and expiry +times, Wire version, executable path, and session-source label. It never stores +the raw session identifier. MCP acquires a lease after bootstrap, renews it from +its existing watcher thread, and removes it on clean shutdown. Crashed owners +leave a bounded stale record which expires and is pruned. + +Supervisor eligibility becomes: + +- never retired; +- initialized; and +- registry-bound, carrying an unexpired live-owner lease, or holding pending + outbound work. + +Missing lifecycle state means inactive during migration. A private key and +daemon-generated `last_sync` do not make a home eligible. Retired and expired +states survive supervisor restart because markers and lease expiries live on +disk. Identity-less unbound husks retain the existing conservative age-based +pruner; identity-bearing homes are never auto-deleted. + +## Supervisor bounds and recovery + +Default worker cap is 16. `--max-workers` configures it. Selection is stable: +registry-bound sessions first, then live leases ordered by newest expiry, then +pending-outbox sessions. Remaining eligible sessions appear in an observable +queue. Existing verified per-session daemons count toward the cap. + +On restart, the supervisor adopts eligible live workers by pidfile. It +terminates only a precisely validated daemon belonging to an ineligible or +overflow session: live PID, daemon role, and matching home. It never kills a +process family or wildcard. Rapid child failures retain exponential backoff; +queue admission never exceeds the cap. State snapshots expose discovered, +eligible, running, queued, stale, retired, cap, binary version, and executable +path counts. + +MCP writes its lease before daemon startup. When a live all-session supervisor +exists, MCP relies on it instead of spawning a bypass worker. Without a +supervisor, the existing single-home fallback remains. + +## Doctor + +`wire doctor` remains read-only and adds checks with an explicit classification: +`wire_defect`, `operator_config`, or `runtime_health`. + +- Supervisor fan-out, cap overflow, queue, stale/retired homes, and legacy homes + lacking lifecycle state. +- Concurrent inbox owners sharing the current home. When the session source is + the universal override, report a likely literal `WIRE_SESSION_ID` collision + and prescribe distinct launcher session IDs. Never print the value. +- PATH candidates, installed service executable, live daemon/MCP executable + paths, pidfile versions, and stale/dead MCP pidfiles. +- Local relay TCP/health state plus installed-service state. +- Same-version duplicates separately from version-skewed processes. + +Existing endpoint-userinfo repair is removed from doctor; diagnostics do not +mutate configuration. + +## Service configuration + +Daemon service install accepts `--interval` and `--max-workers`. Omitted values +are parsed from the existing launchd plist, systemd unit, or Windows task; only +a first install uses defaults (5 seconds, 16 workers). Generated units always +carry both values. Install output reports effective settings. + +## Relay preflight and local pairing + +Local-sister pairing checks the selected local relay before any trust or relay +state mutation. Failure names the endpoint, states that no pairing state +changed, and points to `wire service status --local-relay` / install recovery. +Send preserves endpoint failover; when all attempted endpoints are local and +down, it returns the same precise local-relay classification. + +After a successful local pair-drop, Wire synchronously runs the existing pull +state machine for the sister home and then the caller home. Both sides therefore +consume signed pair/drop-ack events and reach effective `VERIFIED` before dial +returns. Every spawned Wire child pins `WIRE_HOME` and `WIRE_HOME_FORCE=1` and +removes inherited session identifiers. + +## Verification + +- Unit fixtures for 1, 10, and 655 homes; mostly stale; restart persistence; + expired/crashed leases; worker cap and queue; relay failure; service option + preservation; collision/path/version classification. +- Bilateral isolated local-relay integration: one dial, both effective tiers + `VERIFIED`, direct message delivery, no manual pull. +- Audit every test helper that spawns `wire`; temporary homes must also set + `WIRE_HOME_FORCE=1`. +- Focused serial tests, `cargo fmt`, `cargo clippy`, `test-env/run.sh`, + GitNexus change detection against `main`, isolated 655-home runtime process + measurement, read-only branch-binary doctor against the live machine, diff + review, and independent semantic review. From 39a81511514ab22ae9781414aaa92f7d60093d4e Mon Sep 17 00:00:00 2001 From: Paul Logan Date: Fri, 17 Jul 2026 23:32:55 -0700 Subject: [PATCH 2/6] feat: add restart-safe session leases --- SESSION_LOG_2026_07_17.md | 13 ++ src/lib.rs | 1 + src/mcp.rs | 16 ++ src/session_lifecycle.rs | 317 +++++++++++++++++++++++++++++++++++++ tests/session_lifecycle.rs | 42 +++++ 5 files changed, 389 insertions(+) create mode 100644 src/session_lifecycle.rs create mode 100644 tests/session_lifecycle.rs diff --git a/SESSION_LOG_2026_07_17.md b/SESSION_LOG_2026_07_17.md index 7cfac2f..938e57d 100644 --- a/SESSION_LOG_2026_07_17.md +++ b/SESSION_LOG_2026_07_17.md @@ -75,3 +75,16 @@ literal Codex override. implementation and verification plan. - `SESSION_LOG_2026_07_17.md` — durable evidence, decisions, commands, and final results. + +## Iteration 1 — lifecycle leases + +- RED: `cargo test session_lifecycle::tests --lib -- --test-threads=1` failed + with eight missing lifecycle symbols. +- GREEN: five lease unit tests pass: persistence, expiry/dead owner, heartbeat + restart read, conservative pruning, and clean guard removal. +- RED: `cargo test --test session_lifecycle -- --test-threads=1` failed because + MCP published no lease. +- GREEN: MCP process integration passes with forced temporary home; lease exists + for process lifetime and disappears on clean stdin EOF. +- MCP now writes the lease before daemon orchestration and renews it from the + existing watcher thread every 30 seconds. No raw session key is persisted. diff --git a/src/lib.rs b/src/lib.rs index 38da840..f4b5850 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -55,6 +55,7 @@ pub mod same_machine; pub mod send; pub mod service; pub mod session; +pub mod session_lifecycle; pub mod signing; pub mod sso_provider; pub mod tls; diff --git a/src/mcp.rs b/src/mcp.rs index 0fa59e7..ff4a235 100644 --- a/src/mcp.rs +++ b/src/mcp.rs @@ -105,6 +105,12 @@ pub fn run() -> Result<()> { // WIRE_MCP_SKIP_AUTO_UP (tests + manual-identity operators). ensure_session_bootstrapped(); + // A live MCP host is the authoritative activity signal for this identity. + // Publish the lease before daemon orchestration so the all-session + // supervisor can admit this home without treating private-key existence or + // daemon-generated sync timestamps as permanent liveness. + let session_lease = Arc::new(crate::session_lifecycle::LeaseGuard::acquire("mcp")?); + // v0.15.x: minting an identity isn't enough — without a running sync loop // the session is "born deaf" (never pulls inbound, never pushes outbound), // the #1 MCP first-run failure. `ensure_session_bootstrapped` only creates @@ -171,6 +177,7 @@ pub fn run() -> Result<()> { let subs_w = state.subscribed.clone(); let tx_w = tx.clone(); let shutdown_w = shutdown.clone(); + let lease_w = session_lease.clone(); let watcher_handle = std::thread::spawn(move || { let mut watcher = match crate::inbox_watch::InboxWatcher::from_head() { Ok(w) => w, @@ -178,11 +185,20 @@ pub fn run() -> Result<()> { }; let poll_interval = Duration::from_secs(2); let mut next_poll = Instant::now() + poll_interval; + let mut next_lease_heartbeat = + Instant::now() + crate::session_lifecycle::LEASE_HEARTBEAT_INTERVAL; loop { if shutdown_w.load(Ordering::SeqCst) { return; } std::thread::sleep(Duration::from_millis(100)); + if Instant::now() >= next_lease_heartbeat { + if let Err(e) = lease_w.heartbeat() { + eprintln!("wire mcp: session lease heartbeat failed: {e:#}"); + } + next_lease_heartbeat = + Instant::now() + crate::session_lifecycle::LEASE_HEARTBEAT_INTERVAL; + } if Instant::now() < next_poll { continue; } diff --git a/src/session_lifecycle.rs b/src/session_lifecycle.rs new file mode 100644 index 0000000..6a25724 --- /dev/null +++ b/src/session_lifecycle.rs @@ -0,0 +1,317 @@ +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use anyhow::{Context, Result, anyhow}; +use serde::{Deserialize, Serialize}; +use time::OffsetDateTime; + +pub const LEASE_SCHEMA: &str = "wire-session-lease-v1"; +pub const DEFAULT_LEASE_TTL: Duration = Duration::from_secs(90); +pub const LEASE_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30); + +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct LeaseRecord { + pub schema: String, + pub role: String, + pub pid: u32, + pub heartbeat_at: String, + pub expires_at: String, + pub wire_version: String, + pub bin_path: String, + pub session_source: String, +} + +pub fn lease_dir(home: &Path) -> PathBuf { + home.join("state").join("wire").join("leases") +} + +fn format_time(ts: OffsetDateTime) -> Result { + ts.format(&time::format_description::well_known::Rfc3339) + .map_err(|e| anyhow!("formatting lease timestamp: {e}")) +} + +fn parse_time(raw: &str) -> Option { + OffsetDateTime::parse(raw, &time::format_description::well_known::Rfc3339).ok() +} + +fn persist_record(path: &Path, record: &LeaseRecord) -> Result<()> { + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("creating lease directory {parent:?}"))?; + } + let tmp = path.with_extension(format!("json.tmp-{}", std::process::id())); + std::fs::write(&tmp, serde_json::to_vec_pretty(record)?) + .with_context(|| format!("writing session lease {tmp:?}"))?; + if cfg!(windows) && path.exists() { + std::fs::remove_file(path).with_context(|| format!("replacing session lease {path:?}"))?; + } + std::fs::rename(&tmp, path).with_context(|| format!("committing session lease {path:?}"))?; + Ok(()) +} + +#[allow(clippy::too_many_arguments)] +pub fn write_lease_at( + home: &Path, + role: &str, + pid: u32, + now: OffsetDateTime, + ttl: Duration, + wire_version: &str, + bin_path: &Path, + session_source: &str, +) -> Result { + if role.is_empty() + || !role + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || matches!(b, b'-' | b'_')) + { + return Err(anyhow!("invalid lease role `{role}`")); + } + let path = lease_dir(home).join(format!("{role}-{pid}.json")); + let expires = now + time::Duration::seconds(ttl.as_secs() as i64); + let record = LeaseRecord { + schema: LEASE_SCHEMA.to_string(), + role: role.to_string(), + pid, + heartbeat_at: format_time(now)?, + expires_at: format_time(expires)?, + wire_version: wire_version.to_string(), + bin_path: bin_path.to_string_lossy().into_owned(), + session_source: session_source.to_string(), + }; + persist_record(&path, &record)?; + Ok(path) +} + +pub fn heartbeat_lease_at(path: &Path, now: OffsetDateTime, ttl: Duration) -> Result<()> { + let body = std::fs::read(path).with_context(|| format!("reading session lease {path:?}"))?; + let mut record: LeaseRecord = + serde_json::from_slice(&body).with_context(|| format!("parsing session lease {path:?}"))?; + record.heartbeat_at = format_time(now)?; + record.expires_at = format_time(now + time::Duration::seconds(ttl.as_secs() as i64))?; + persist_record(path, &record) +} + +fn read_lease(path: &Path) -> Option { + let body = std::fs::read(path).ok()?; + let record: LeaseRecord = serde_json::from_slice(&body).ok()?; + (record.schema == LEASE_SCHEMA).then_some(record) +} + +pub fn active_leases_at( + home: &Path, + now: OffsetDateTime, + is_alive: impl Fn(u32) -> bool, +) -> Vec { + let Ok(entries) = std::fs::read_dir(lease_dir(home)) else { + return Vec::new(); + }; + let mut active: Vec = entries + .flatten() + .filter_map(|entry| read_lease(&entry.path())) + .filter(|record| { + parse_time(&record.expires_at).is_some_and(|expires| expires > now) + && is_alive(record.pid) + }) + .collect(); + active.sort_by(|a, b| a.role.cmp(&b.role).then(a.pid.cmp(&b.pid))); + active +} + +pub fn prune_stale_leases_at( + home: &Path, + now: OffsetDateTime, + is_alive: impl Fn(u32) -> bool, +) -> usize { + let Ok(entries) = std::fs::read_dir(lease_dir(home)) else { + return 0; + }; + entries + .flatten() + .filter(|entry| { + let stale = read_lease(&entry.path()).is_none_or(|record| { + parse_time(&record.expires_at).is_none_or(|expires| expires <= now) + || !is_alive(record.pid) + }); + stale && std::fs::remove_file(entry.path()).is_ok() + }) + .count() +} + +pub struct LeaseGuard { + path: PathBuf, + ttl: Duration, +} + +impl LeaseGuard { + pub fn acquire(role: &str) -> Result { + let state_dir = crate::config::state_dir()?; + let home = state_dir + .parent() + .and_then(Path::parent) + .ok_or_else(|| anyhow!("state directory has no session-home parent"))?; + let bin = crate::platform::current_exe_resolved()?; + Self::acquire_at( + home, + role, + std::process::id(), + OffsetDateTime::now_utc(), + DEFAULT_LEASE_TTL, + env!("CARGO_PKG_VERSION"), + &bin, + crate::session::session_source(), + ) + } + + pub fn heartbeat(&self) -> Result<()> { + self.heartbeat_at(OffsetDateTime::now_utc()) + } + + #[allow(clippy::too_many_arguments)] + fn acquire_at( + home: &Path, + role: &str, + pid: u32, + now: OffsetDateTime, + ttl: Duration, + wire_version: &str, + bin_path: &Path, + session_source: &str, + ) -> Result { + let path = write_lease_at( + home, + role, + pid, + now, + ttl, + wire_version, + bin_path, + session_source, + )?; + Ok(Self { path, ttl }) + } + + fn heartbeat_at(&self, now: OffsetDateTime) -> Result<()> { + heartbeat_lease_at(&self.path, now, self.ttl) + } +} + +impl Drop for LeaseGuard { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::path::Path; + use std::time::Duration; + use tempfile::tempdir; + use time::OffsetDateTime; + + fn write_test_lease( + home: &Path, + pid: u32, + now: OffsetDateTime, + ttl: Duration, + ) -> std::path::PathBuf { + write_lease_at( + home, + "mcp", + pid, + now, + ttl, + "0.17.0", + Path::new("/opt/wire"), + "override", + ) + .unwrap() + } + + #[test] + fn active_lease_round_trips_without_raw_session_key() { + let tmp = tempdir().unwrap(); + let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap(); + let path = write_test_lease(tmp.path(), 42, now, Duration::from_secs(90)); + + let body = std::fs::read_to_string(path).unwrap(); + assert!(!body.contains("WIRE_SESSION_ID")); + assert!(!body.contains("codex-skill-router")); + let leases = active_leases_at(tmp.path(), now, |pid| pid == 42); + assert_eq!(leases.len(), 1); + assert_eq!(leases[0].role, "mcp"); + assert_eq!(leases[0].pid, 42); + assert_eq!(leases[0].session_source, "override"); + } + + #[test] + fn expired_or_dead_owner_is_not_active() { + let tmp = tempdir().unwrap(); + let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap(); + write_test_lease(tmp.path(), 42, now, Duration::from_secs(90)); + + assert!(active_leases_at(tmp.path(), now, |_| false).is_empty()); + assert!( + active_leases_at(tmp.path(), now + time::Duration::seconds(91), |_| true).is_empty() + ); + } + + #[test] + fn heartbeat_extends_persisted_expiry_across_restart_read() { + let tmp = tempdir().unwrap(); + let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap(); + let path = write_test_lease(tmp.path(), 42, now, Duration::from_secs(90)); + heartbeat_lease_at( + &path, + now + time::Duration::seconds(60), + Duration::from_secs(90), + ) + .unwrap(); + + let restarted_at = now + time::Duration::seconds(120); + let leases = active_leases_at(tmp.path(), restarted_at, |pid| pid == 42); + assert_eq!(leases.len(), 1); + assert_eq!(leases[0].pid, 42); + } + + #[test] + fn pruning_removes_expired_dead_and_malformed_records_only() { + let tmp = tempdir().unwrap(); + let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap(); + let live = write_test_lease(tmp.path(), 1, now, Duration::from_secs(90)); + let dead = write_test_lease(tmp.path(), 2, now, Duration::from_secs(90)); + let malformed = lease_dir(tmp.path()).join("bad.json"); + std::fs::write(&malformed, "not-json").unwrap(); + + let removed = prune_stale_leases_at(tmp.path(), now, |pid| pid == 1); + assert_eq!(removed, 2); + assert!(live.exists()); + assert!(!dead.exists()); + assert!(!malformed.exists()); + } + + #[test] + fn guard_heartbeats_and_removes_lease_on_clean_drop() { + let tmp = tempdir().unwrap(); + let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap(); + let guard = LeaseGuard::acquire_at( + tmp.path(), + "mcp", + 42, + now, + Duration::from_secs(90), + "0.17.0", + Path::new("/opt/wire"), + "codex-cli", + ) + .unwrap(); + let path = guard.path.clone(); + guard + .heartbeat_at(now + time::Duration::seconds(30)) + .unwrap(); + assert_eq!(active_leases_at(tmp.path(), now, |_| true).len(), 1); + drop(guard); + assert!(!path.exists()); + } +} diff --git a/tests/session_lifecycle.rs b/tests/session_lifecycle.rs new file mode 100644 index 0000000..530b0b1 --- /dev/null +++ b/tests/session_lifecycle.rs @@ -0,0 +1,42 @@ +use std::process::{Command, Stdio}; +use std::time::{Duration, Instant}; + +use tempfile::tempdir; + +fn wire_bin() -> &'static str { + env!("CARGO_BIN_EXE_wire") +} + +#[test] +fn mcp_process_owns_lease_for_its_lifetime() { + let home = tempdir().unwrap(); + let mut child = Command::new(wire_bin()) + .arg("mcp") + .env("WIRE_HOME", home.path()) + .env("WIRE_HOME_FORCE", "1") + .env("WIRE_AUTO_INIT", "0") + .env("WIRE_MCP_SKIP_AUTO_UP", "1") + .env_remove("WIRE_SESSION_ID") + .env_remove("CLAUDE_CODE_SESSION_ID") + .env_remove("CODEX_SESSION_ID") + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + + let lease = home + .path() + .join("state/wire/leases") + .join(format!("mcp-{}.json", child.id())); + let deadline = Instant::now() + Duration::from_secs(5); + while !lease.exists() && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(20)); + } + assert!(lease.exists(), "MCP did not publish its lifecycle lease"); + + drop(child.stdin.take()); + let status = child.wait().unwrap(); + assert!(status.success()); + assert!(!lease.exists(), "clean MCP shutdown left a stale lease"); +} From 83927962393c505cf0c9c8a3ac1e87f456baf397 Mon Sep 17 00:00:00 2001 From: Paul Logan Date: Sat, 18 Jul 2026 00:18:15 -0700 Subject: [PATCH 3/6] feat: bound multi-session daemon workers --- src/cli/mod.rs | 12 +- src/cli/relay.rs | 6 +- src/cli/status.rs | 373 +++++++++++++++++++- src/cli/upgrade.rs | 12 +- src/daemon_supervisor.rs | 672 ++++++++++++++++++------------------ src/mcp.rs | 24 +- src/service.rs | 260 ++++++++++++-- tests/supervisor_bounded.rs | 94 +++++ 8 files changed, 1085 insertions(+), 368 deletions(-) create mode 100644 tests/supervisor_bounded.rs diff --git a/src/cli/mod.rs b/src/cli/mod.rs index f1d99d9..dbbcd9f 100644 --- a/src/cli/mod.rs +++ b/src/cli/mod.rs @@ -578,6 +578,9 @@ pub enum Command { /// at login without the operator running N tmux panes. #[arg(long)] all_sessions: bool, + /// Maximum session workers in `--all-sessions` supervisor mode. + #[arg(long, default_value_t = crate::daemon_supervisor::DEFAULT_MAX_WORKERS)] + max_workers: usize, /// v0.14.2 (#162): run the daemon loop pinned to a specific /// named session by setting WIRE_HOME for the process. The /// supervisor (`--all-sessions`) spawns children with this @@ -1792,6 +1795,12 @@ pub enum ServiceAction { /// Install the local-relay service instead of the daemon. #[arg(long)] local_relay: bool, + /// Daemon sync interval. Omitted preserves an existing service value. + #[arg(long)] + interval: Option, + /// Supervisor worker cap. Omitted preserves an existing service value. + #[arg(long)] + max_workers: Option, #[arg(long)] json: bool, }, @@ -2061,9 +2070,10 @@ pub fn run() -> Result<()> { interval, once, all_sessions, + max_workers, session, json, - } => relay::cmd_daemon(interval, once, all_sessions, session, json), + } => relay::cmd_daemon(interval, once, all_sessions, max_workers, session, json), Command::Session(cmd) => cmd_session(cmd), Command::Identity { cmd } => identity::cmd_identity(cmd), Command::Mesh(cmd) => cmd_mesh(cmd), diff --git a/src/cli/relay.rs b/src/cli/relay.rs index 165b521..6b82534 100644 --- a/src/cli/relay.rs +++ b/src/cli/relay.rs @@ -1058,6 +1058,7 @@ pub(super) fn cmd_daemon( interval_secs: u64, once: bool, all_sessions: bool, + max_workers: usize, session: Option, as_json: bool, ) -> Result<()> { @@ -1074,7 +1075,10 @@ pub(super) fn cmd_daemon( "--all-sessions and --session are mutually exclusive (supervisor manages every session, not a single named one)" ); } - return crate::daemon_supervisor::run_supervisor(interval_secs, as_json); + if max_workers == 0 { + bail!("--max-workers must be at least 1"); + } + return crate::daemon_supervisor::run_supervisor(interval_secs, max_workers, as_json); } // v0.14.2 (#162): pin this process's WIRE_HOME to the named session's // home dir BEFORE any config read. Used by the supervisor when it diff --git a/src/cli/status.rs b/src/cli/status.rs index d13b6e8..e6789ad 100644 --- a/src/cli/status.rs +++ b/src/cli/status.rs @@ -970,6 +970,8 @@ pub struct DoctorCheck { pub id: String, /// PASS / WARN / FAIL. pub status: String, + /// `runtime_health`, `operator_config`, or `wire_defect`. + pub classification: String, /// One-line human summary. pub detail: String, /// Optional remediation hint shown after the failing line. @@ -982,6 +984,7 @@ impl DoctorCheck { Self { id: id.into(), status: "PASS".into(), + classification: "runtime_health".into(), detail: detail.into(), fix: None, } @@ -990,6 +993,7 @@ impl DoctorCheck { Self { id: id.into(), status: "WARN".into(), + classification: "runtime_health".into(), detail: detail.into(), fix: Some(fix.into()), } @@ -998,10 +1002,284 @@ impl DoctorCheck { Self { id: id.into(), status: "FAIL".into(), + classification: "runtime_health".into(), detail: detail.into(), fix: Some(fix.into()), } } + + fn classified(mut self, classification: &str) -> Self { + self.classification = classification.into(); + self + } +} + +fn supervisor_fanout_verdict( + workers: usize, + cap: usize, + homes: usize, + inactive_homes: usize, +) -> DoctorCheck { + if workers > cap { + DoctorCheck::fail( + "supervisor_fanout", + format!( + "{workers} live session workers exceed cap {cap} across {homes} homes; {inactive_homes} homes have no active lifecycle signal" + ), + "install this Wire build, then restart through the existing service manager; do not hand-start daemons or wildcard-kill processes", + ) + .classified("wire_defect") + } else { + DoctorCheck::pass( + "supervisor_fanout", + format!( + "{workers} live session workers within cap {cap}; {inactive_homes}/{homes} homes inactive" + ), + ) + } +} + +fn session_override_collision_verdict(processes: usize, distinct_homes: usize) -> DoctorCheck { + if processes > 1 && distinct_homes < processes { + DoctorCheck::warn( + "session_id_collision", + format!( + "{processes} concurrent MCP processes share {distinct_homes} session identity from a literal WIRE_SESSION_ID override" + ), + "remove the fixed WIRE_SESSION_ID from the launcher and pass one stable, unique session id per Codex session; Wire configuration is behaving as requested", + ) + .classified("operator_config") + } else { + DoctorCheck::pass( + "session_id_collision", + format!( + "{processes} concurrent MCP process(es) map to {distinct_homes} distinct session home(s)" + ), + ) + } +} + +fn stale_mcp_verdict(live: usize, skewed: usize) -> DoctorCheck { + if skewed > 0 { + DoctorCheck::warn( + "stale_mcp", + format!("{skewed}/{live} live MCP process(es) record a different Wire version"), + "reconnect affected MCP hosts after deploying the new binary; do not kill all MCP processes as a first step", + ) + } else { + DoctorCheck::pass( + "stale_mcp", + format!("{live} live MCP process(es), no version skew"), + ) + } +} + +fn executable_path_verdict(path_first_is_current: bool, service_is_current: bool) -> DoctorCheck { + if !path_first_is_current { + DoctorCheck::warn( + "path_shadow", + "the first `wire` on PATH is not this running executable", + "remove or reorder the older PATH entry; confirm with `type -a wire` and `wire --version`", + ) + .classified("operator_config") + } else if !service_is_current { + DoctorCheck::warn( + "path_shadow", + "the installed daemon service points at a different Wire executable", + "re-run `wire service install` from the intended binary; existing interval and worker cap will be preserved", + ) + } else { + DoctorCheck::pass( + "path_shadow", + "PATH and installed service resolve to this executable", + ) + } +} + +fn local_relay_verdict(installed: bool, reachable: bool) -> DoctorCheck { + if reachable { + DoctorCheck::pass("local_relay", "127.0.0.1:8771 accepts connections") + } else { + DoctorCheck::fail( + "local_relay", + format!( + "127.0.0.1:8771 refuses connections; local-relay service installed={installed}" + ), + if installed { + "inspect `wire service status --local-relay` and launchd/systemd logs; recover through the service manager" + } else { + "install with `wire service install --local-relay`" + }, + ) + } +} + +fn check_supervisor_fanout() -> DoctorCheck { + let sessions = crate::session::list_sessions().unwrap_or_default(); + let workers = sessions + .iter() + .filter_map(|session| crate::session::session_daemon_pid(&session.home_dir)) + .filter(|pid| crate::platform::process_alive(*pid)) + .count(); + let inactive = sessions + .iter() + .filter(|session| { + session.cwd.is_none() + && crate::session_lifecycle::active_leases_at( + &session.home_dir, + time::OffsetDateTime::now_utc(), + crate::platform::process_alive, + ) + .is_empty() + }) + .count(); + let cap = crate::service::installed_daemon_options().max_workers; + supervisor_fanout_verdict(workers, cap, sessions.len(), inactive) +} + +fn check_session_id_collisions() -> DoctorCheck { + let processes = actual_wire_role_pids("mcp").len(); + let (live_homes, _) = live_mcp_inventory(); + if codex_has_fixed_session_override() && processes > 1 { + DoctorCheck::warn( + "session_id_collision", + format!( + "Codex launcher config sets one literal WIRE_SESSION_ID while {processes} Wire MCP processes run; every Codex session inheriting it collapses to one identity" + ), + "remove the fixed WIRE_SESSION_ID from ~/.codex/config.toml and pass one stable, unique session id per Codex session; Wire configuration is behaving as requested", + ) + .classified("operator_config") + } else if crate::session::session_source() == "override" { + session_override_collision_verdict(processes, live_homes) + } else { + DoctorCheck::pass( + "session_id_collision", + format!( + "current session source is {}; {processes} concurrent MCP process(es), {live_homes} independently addressable live home(s)", + crate::session::session_source() + ), + ) + } +} + +fn codex_has_fixed_session_override() -> bool { + let Some(home) = std::env::var_os("HOME") else { + return false; + }; + let Ok(body) = std::fs::read_to_string( + std::path::PathBuf::from(home) + .join(".codex") + .join("config.toml"), + ) else { + return false; + }; + codex_config_has_fixed_session_override(&body) +} + +fn codex_config_has_fixed_session_override(body: &str) -> bool { + let mut in_set_section = false; + for line in body.lines() { + let line = line.trim(); + if line.starts_with('[') && line.ends_with(']') { + in_set_section = line == "[shell_environment_policy.set]"; + continue; + } + if in_set_section + && line.split_once('=').is_some_and(|(key, value)| { + key.trim() == "WIRE_SESSION_ID" && !value.trim().is_empty() + }) + { + return true; + } + } + false +} + +fn actual_wire_role_pids(role: &str) -> Vec { + crate::platform::find_processes_by_cmdline(&format!("wire {role}")) + .into_iter() + .filter(|pid| { + crate::platform::pid_cmdline(*pid).is_some_and(|cmdline| { + let mut tokens = cmdline.split_whitespace(); + let executable = tokens.next().and_then(|token| { + std::path::Path::new(token) + .file_name() + .and_then(|name| name.to_str()) + }); + executable.is_some_and(|name| name == "wire" || name == "wire.exe") + && tokens.next() == Some(role) + }) + }) + .collect() +} + +fn live_mcp_inventory() -> (usize, usize) { + let mut live_homes = 0usize; + let mut skewed = 0usize; + for session in crate::session::list_sessions().unwrap_or_default() { + let pidfile = session.home_dir.join("state").join("wire").join("mcp.pid"); + let Ok(body) = std::fs::read_to_string(pidfile) else { + continue; + }; + let Ok(record) = serde_json::from_str::(&body) else { + continue; + }; + if crate::platform::process_alive(record.pid) { + live_homes += 1; + skewed += usize::from(record.version != env!("CARGO_PKG_VERSION")); + } + } + (live_homes, skewed) +} + +fn check_stale_mcp_processes() -> DoctorCheck { + let (live, skewed) = live_mcp_inventory(); + stale_mcp_verdict(live, skewed) +} + +fn paths_match(a: &std::path::Path, b: &std::path::Path) -> bool { + let resolved = + |path: &std::path::Path| std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()); + resolved(a) == resolved(b) +} + +fn check_executable_paths() -> DoctorCheck { + let Ok(current) = crate::platform::current_exe_resolved() else { + return DoctorCheck::warn( + "path_shadow", + "could not resolve current Wire executable", + "run `type -a wire` and inspect the installed service executable", + ); + }; + let path_first = std::process::Command::new("which") + .args(["-a", "wire"]) + .output() + .ok() + .filter(|output| output.status.success()) + .and_then(|output| { + String::from_utf8_lossy(&output.stdout) + .lines() + .next() + .map(std::path::PathBuf::from) + }); + let path_matches = path_first + .as_deref() + .is_some_and(|path| paths_match(path, ¤t)); + let service_matches = crate::service::installed_daemon_executable() + .as_deref() + .is_none_or(|path| paths_match(path, ¤t)); + executable_path_verdict(path_matches, service_matches) +} + +fn check_local_relay_health() -> DoctorCheck { + let address = std::net::SocketAddr::from(([127, 0, 0, 1], 8771)); + let reachable = + std::net::TcpStream::connect_timeout(&address, std::time::Duration::from_millis(300)) + .is_ok(); + let installed = crate::service::status_kind(crate::service::ServiceKind::LocalRelay) + .map(|report| report.status != "absent") + .unwrap_or(false); + local_relay_verdict(installed, reachable) } /// `wire doctor` — single-command diagnostic for the silent-fail classes @@ -1010,6 +1288,11 @@ impl DoctorCheck { /// so operators don't have to know where each lives. pub(super) fn cmd_doctor(as_json: bool, recent_rejections: usize) -> Result<()> { let checks: Vec = vec![ + check_supervisor_fanout(), + check_session_id_collisions(), + check_executable_paths(), + check_stale_mcp_processes(), + check_local_relay_health(), check_daemon_health(), check_daemon_pid_consistency(), check_relay_reachable(), @@ -1017,7 +1300,7 @@ pub(super) fn cmd_doctor(as_json: bool, recent_rejections: usize) -> Result<()> check_sync_freshness(), check_cursor_progress(), check_peer_staleness(7), - check_and_heal_self_userinfo_endpoints(), + check_self_userinfo_endpoints(), check_stale_inbound_pairs(), ]; @@ -1049,7 +1332,10 @@ pub(super) fn cmd_doctor(as_json: bool, recent_rejections: usize) -> Result<()> "FAIL" => "✗", _ => "?", }; - println!(" {bullet} [{}] {}: {}", c.status, c.id, c.detail); + println!( + " {bullet} [{}] {} {}: {}", + c.status, c.classification, c.id, c.detail + ); if let Some(fix) = &c.fix { println!(" fix: {fix}"); } @@ -1415,6 +1701,55 @@ fn check_pair_rejections(recent_n: usize) -> DoctorCheck { ) } +/// Read-only check for malformed self relay endpoints. Doctor never rewrites +/// relay state; it reports the count and delegates repair to explicit bind and +/// claim commands. +fn check_self_userinfo_endpoints() -> DoctorCheck { + let Ok(state) = config::read_relay_state() else { + return DoctorCheck::pass( + "self-userinfo-endpoints", + "no relay state yet — nothing published", + ); + }; + let Some(self_block) = state.get("self") else { + return DoctorCheck::pass( + "self-userinfo-endpoints", + "no self block in relay state — nothing published", + ); + }; + let endpoint_count = self_block + .get("endpoints") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter(|endpoint| { + endpoint + .get("relay_url") + .and_then(Value::as_str) + .is_some_and(|url| super::setup::assert_relay_url_clean_for_publish(url).is_err()) + }) + .count(); + let legacy_bad = self_block + .get("relay_url") + .and_then(Value::as_str) + .is_some_and(|url| super::setup::assert_relay_url_clean_for_publish(url).is_err()); + let malformed = endpoint_count + usize::from(legacy_bad); + if malformed == 0 { + DoctorCheck::pass( + "self-userinfo-endpoints", + "no malformed endpoints in self-state", + ) + } else { + DoctorCheck::warn( + "self-userinfo-endpoints", + format!( + "{malformed} malformed userinfo-bearing self endpoint(s); doctor made no changes" + ), + "bind a clean relay explicitly, then republish with `wire claim `; inspect state before removing legacy entries", + ) + } +} + /// Check: cursor isn't stuck. We can't tell without polling — but we can /// report the current cursor position so operators see if it changes. /// Real "stuck" detection needs two pulls separated in time; defer that @@ -1478,6 +1813,7 @@ fn check_pair_rejections(recent_n: usize) -> DoctorCheck { /// round-trip + persona arg. Operators get the explicit next step in /// the WARN fix text. Two-step is the right friction: heal silently, /// claim explicitly. +#[cfg(test)] fn check_and_heal_self_userinfo_endpoints() -> DoctorCheck { let mut state = match config::read_relay_state() { Ok(s) => s, @@ -1806,6 +2142,39 @@ fn check_cursor_progress() -> DoctorCheck { mod doctor_tests { use super::*; + #[test] + fn supervisor_fanout_verdict_catches_655_home_failure() { + let c = supervisor_fanout_verdict(566, 16, 655, 650); + assert_eq!(c.status, "FAIL"); + assert_eq!(c.classification, "wire_defect"); + assert!(c.detail.contains("566") && c.detail.contains("16")); + } + + #[test] + fn duplicate_literal_override_is_operator_configuration() { + let c = session_override_collision_verdict(2, 1); + assert_eq!(c.status, "WARN"); + assert_eq!(c.classification, "operator_config"); + assert!(c.detail.contains("2")); + assert_eq!(session_override_collision_verdict(2, 2).status, "PASS"); + assert!(codex_config_has_fixed_session_override( + "[shell_environment_policy.set]\nWIRE_SESSION_ID = \"fixed\"" + )); + assert!(!codex_config_has_fixed_session_override( + "[shell_environment_policy]\ninherit = \"core\"" + )); + } + + #[test] + fn stale_mcp_path_skew_and_local_relay_have_precise_verdicts() { + assert_eq!(stale_mcp_verdict(41, 7).status, "WARN"); + assert_eq!( + executable_path_verdict(false, true).classification, + "operator_config" + ); + assert_eq!(local_relay_verdict(false, false).status, "FAIL"); + } + #[test] fn sync_freshness_fails_when_stale_with_queued_events() { // The silent-send class doctor exists to catch: the daemon process diff --git a/src/cli/upgrade.rs b/src/cli/upgrade.rs index 58ba988..1d660d3 100644 --- a/src/cli/upgrade.rs +++ b/src/cli/upgrade.rs @@ -14,9 +14,15 @@ pub(crate) fn cmd_service(action: ServiceAction) -> Result<()> { } }; let (report, as_json) = match action { - ServiceAction::Install { local_relay, json } => { - (crate::service::install_kind(kind(local_relay))?, json) - } + ServiceAction::Install { + local_relay, + interval, + max_workers, + json, + } => ( + crate::service::install_kind_with_options(kind(local_relay), interval, max_workers)?, + json, + ), ServiceAction::Uninstall { local_relay, json } => { (crate::service::uninstall_kind(kind(local_relay))?, json) } diff --git a/src/daemon_supervisor.rs b/src/daemon_supervisor.rs index 3f37700..2236470 100644 --- a/src/daemon_supervisor.rs +++ b/src/daemon_supervisor.rs @@ -15,7 +15,7 @@ //! from the project cwd. That works for one session but doesn't scale //! to N. The architectural fix is a supervisor that owns the //! multi-session orchestration: one supervisor process per launchd -//! unit, N child `wire daemon --session ` processes — each with +//! unit, a bounded set of child `wire daemon` processes — each with //! its own pinned `WIRE_HOME` and its own pidfile under that session's //! state dir. `wire status` from any cwd then sees its session's child //! pid and reports truthfully. @@ -39,11 +39,12 @@ //! session (corrupt key, missing relay) from fork-bombing. //! - **Don't exit on zero sessions.** Sleep and re-poll the registry — //! new sessions get picked up without supervisor restart. -//! - **Adopt orphaned children on supervisor restart.** When launchd -//! relaunches the supervisor, the previous supervisor's children -//! keep running (correct: they're still syncing). New supervisor -//! sees their pidfiles, skips re-spawning, and lets them keep going -//! until their next natural exit (then it spawns a fresh child). +//! - **Lease-backed lifecycle.** A registry binding, live process lease, +//! or pending outbox makes a session eligible. A private key and sync +//! timestamps do not. Retired and inactive homes stay inactive across +//! supervisor restarts. +//! - **Hard worker cap.** Excess eligible sessions remain in an +//! observable queue instead of expanding process count without bound. //! //! ## Invariants //! @@ -77,43 +78,76 @@ const REGISTRY_POLL_SECS: u64 = 10; const INITIAL_BACKOFF: Duration = Duration::from_secs(1); const MAX_BACKOFF: Duration = Duration::from_secs(60); const RAPID_FAIL_WINDOW: Duration = Duration::from_secs(10); +pub const DEFAULT_MAX_WORKERS: usize = 16; -/// Default idle cutoff for registry-unbound sessions. `list_sessions()` -/// enumerates *every* session home ever minted on the machine — and -/// because each Claude tab / `wire session new` mints a fresh persona -/// home, a long-lived box accumulates hundreds (honey-pine's had 147). -/// Spawning one daemon per home turns `--all-sessions` into a fork -/// storm. A session is kept regardless of age if it has a registry cwd -/// binding (operator deliberately bound it) OR holds a real identity -/// (`config/wire/private.key` — not a husk); an unbound, identity-less -/// session is only kept if it has been active within this window. Override -/// via `WIRE_ALL_SESSIONS_MAX_IDLE_DAYS` (0 disables the filter → legacy -/// spawn-for-all behavior). -const DEFAULT_MAX_IDLE_DAYS: u64 = 7; - -/// Parse the idle cutoff. `None` raw → default; a `0` value → `None` -/// (no filter, spawn for every session); any other integer → that many -/// days; unparseable → default. Pure, so it's unit-testable without -/// mutating process env. -fn parse_max_idle(raw: Option<&str>) -> Option { - match raw { - Some(v) => { - let days: u64 = v.trim().parse().unwrap_or(DEFAULT_MAX_IDLE_DAYS); - (days != 0).then(|| Duration::from_secs(days * 86_400)) - } - None => Some(Duration::from_secs(DEFAULT_MAX_IDLE_DAYS * 86_400)), +fn next_worker_backoff(previous: Option, rapid_failure: bool) -> Duration { + if rapid_failure { + (previous.unwrap_or(INITIAL_BACKOFF) * 2).min(MAX_BACKOFF) + } else { + INITIAL_BACKOFF } } -/// Read the idle cutoff from the environment. `None` means "no idle -/// filter" (spawn a daemon for every session — pre-fix behavior), -/// selected by setting `WIRE_ALL_SESSIONS_MAX_IDLE_DAYS=0`. -fn max_idle_from_env() -> Option { - parse_max_idle( - std::env::var("WIRE_ALL_SESSIONS_MAX_IDLE_DAYS") - .ok() - .as_deref(), - ) +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum WorkerVersion { + Current, + Skewed, +} + +fn classify_session_worker( + record: &crate::ensure_up::DaemonPid, + alive: bool, + cmdline: Option<&str>, +) -> Option { + if !alive + || record.schema != crate::ensure_up::DAEMON_PID_SCHEMA + || record.pid == std::process::id() + { + return None; + } + let cmdline = cmdline?; + let args: Vec<&str> = cmdline.split_whitespace().collect(); + if !args.contains(&"daemon") || args.contains(&"--all-sessions") { + return None; + } + Some(if record.version == env!("CARGO_PKG_VERSION") { + WorkerVersion::Current + } else { + WorkerVersion::Skewed + }) +} + +/// Stop one pidfile-owned worker for a session that lifecycle planning did +/// not select. Validation is deliberately narrow: JSON pidfile, live PID, +/// daemon command line, and never the all-session supervisor. No process +/// family signal and no home mutation. +fn retire_inactive_worker(session: &crate::session::SessionInfo) { + let pidfile = session + .home_dir + .join("state") + .join("wire") + .join("daemon.pid"); + let Ok(body) = std::fs::read_to_string(pidfile) else { + return; + }; + let Ok(record) = serde_json::from_str::(&body) else { + return; + }; + let alive = crate::platform::process_alive(record.pid); + let cmdline = crate::platform::pid_cmdline(record.pid); + let Some(version) = classify_session_worker(&record, alive, cmdline.as_deref()) else { + return; + }; + eprintln!( + "supervisor: retiring unselected session worker '{}' pid={} version={version:?}", + session.name, record.pid + ); + if !crate::platform::kill_process(record.pid, false) { + eprintln!( + "supervisor: failed to signal unselected session worker '{}' pid={}", + session.name, record.pid + ); + } } /// Newest mtime among a session home's activity files — the @@ -137,26 +171,6 @@ fn fs_last_active(home: &Path) -> Option { .max() } -/// True iff the session home holds a real wire identity (an initialized -/// `config/wire/private.key`). Such a home ran `wire up`/`init` — it is NOT a -/// husk, so it must keep a daemon to push its outbox AND pull inbound mail even -/// when idle + registry-unbound. Otherwise a real-but-idle (or never-synced) -/// session is starved forever: ineligible → no daemon → never syncs → never -/// eligible (the wildflower-gleam catch-22 a live audit surfaced — for outbound -/// mail and, equally, inbound never-pulled mail). Husks (ephemeral read-only -/// home mints) have no private.key and stay excluded, so the idle filter still -/// stops their fork-storm. This generalizes #340's pending-outbox keep: a queued -/// outbox implies an identity, so identity subsumes it. Injected into -/// `supervisor_eligible` so the filter stays unit-testable. -fn fs_has_identity(home: &Path) -> bool { - // A real wire identity = an initialized `config/wire/private.key`. Same - // signal the husk-reaper uses to decide "not a husk". - home.join("config") - .join("wire") - .join("private.key") - .exists() -} - /// True iff the session home has been retired (`state/wire/retired.json`). /// A retired home is ineligible for a daemon regardless of cwd/identity/idle — /// the supervisor kills any running child and never respawns. Pure existence @@ -165,60 +179,100 @@ fn fs_is_retired(home: &Path) -> bool { crate::retire::is_retired(home) } -/// Filter `list_sessions()` down to the sessions the supervisor should -/// own a daemon for. A session is eligible iff it has a registry cwd -/// binding OR it was active within `max_idle`. `max_idle == None` -/// disables the filter (every session eligible). Pure: the activity -/// probe is injected so this is unit-testable without touching disk. -fn supervisor_eligible( +fn fs_has_live_lease(home: &Path) -> bool { + !crate::session_lifecycle::active_leases_at( + home, + time::OffsetDateTime::now_utc(), + crate::platform::process_alive, + ) + .is_empty() +} + +fn fs_has_pending_outbox(home: &Path) -> bool { + let outbox = home.join("state").join("wire").join("outbox"); + let Ok(entries) = std::fs::read_dir(outbox) else { + return false; + }; + entries.flatten().any(|entry| { + let path = entry.path(); + path.extension().and_then(|s| s.to_str()) == Some("jsonl") + && !path + .file_name() + .and_then(|s| s.to_str()) + .is_some_and(|name| name.ends_with(".pushed.jsonl")) + && std::fs::read_to_string(path).is_ok_and(|body| !body.trim().is_empty()) + }) +} + +#[derive(Debug, Clone)] +struct SupervisorPlan { + selected: Vec, + queued: Vec, + inactive: Vec, + retired: Vec, +} + +#[cfg(test)] +impl SupervisorPlan { + fn selected_names(&self) -> Vec<&str> { + self.selected + .iter() + .map(|session| session.name.as_str()) + .collect() + } +} + +fn plan_supervisor_sessions( sessions: Vec, - max_idle: Option, - now: SystemTime, - last_active: F, - has_identity: G, + max_workers: usize, + has_live_lease: F, + has_pending_outbox: G, is_retired: H, -) -> Vec +) -> SupervisorPlan where - F: Fn(&Path) -> Option, + F: Fn(&Path) -> bool, G: Fn(&Path) -> bool, H: Fn(&Path) -> bool, { - // Retired homes are ineligible in EVERY configuration — filtered FIRST, - // ahead of the `max_idle == None` early-return and the cwd/identity keeps - // below. A retired identity's daemon must stop and never respawn; an - // `is_retired` check placed after either branch would let a cwd-bound or - // identity-ful retired home stay eligible and get respawned. - let sessions: Vec = sessions - .into_iter() - .filter(|s| !is_retired(&s.home_dir)) + let mut eligible = Vec::new(); + let mut inactive = Vec::new(); + let mut retired = Vec::new(); + for session in sessions { + if is_retired(&session.home_dir) { + retired.push(session.name); + continue; + } + if session.did.is_none() { + inactive.push(session.name); + continue; + } + if session.cwd.is_some() + || has_live_lease(&session.home_dir) + || has_pending_outbox(&session.home_dir) + { + eligible.push(session); + } else { + inactive.push(session.name); + } + } + eligible.sort_by(|a, b| { + b.cwd + .is_some() + .cmp(&a.cwd.is_some()) + .then_with(|| a.name.cmp(&b.name)) + }); + let split = eligible.len().min(max_workers); + let queued = eligible[split..] + .iter() + .map(|session| session.name.clone()) .collect(); - let Some(max_idle) = max_idle else { - return sessions; - }; - sessions - .into_iter() - .filter(|s| { - if s.cwd.is_some() { - return true; - } - // A real wire identity (private.key) is not a husk → keep a daemon so - // it can push its outbox AND pull inbound mail, regardless of idle. - // Closes the wildflower-gleam catch-22 (ineligible → no daemon → - // never syncs → never eligible) for BOTH outbound-queued and - // never-sent/inbound-waiting sessions. Generalizes #340's - // pending-outbox keep (an outbox implies an identity). Husks have no - // identity → still excluded, so the idle fork-storm guard stands. - if has_identity(&s.home_dir) { - return true; - } - match last_active(&s.home_dir) { - // `duration_since` errors when the file mtime is in the - // future (clock skew) — treat that as "active now". - Some(t) => now.duration_since(t).map(|d| d <= max_idle).unwrap_or(true), - None => false, - } - }) - .collect() + let selected = eligible.into_iter().take(split).collect(); + SupervisorPlan { + selected, + queued, + inactive, + retired, + } } // ---- husk reaper (the 175-dir by-key accumulation fix) ---- @@ -237,7 +291,7 @@ const HUSK_REAP_INTERVAL: Duration = Duration::from_secs(3600); /// Parse the husk reap cutoff. `None` raw → default; a `0` value → /// `None` (reaper disabled); any other integer → that many hours; -/// unparseable → default. Pure, mirrors `parse_max_idle`. +/// unparseable → default. fn parse_husk_reap_max_age(raw: Option<&str>) -> Option { match raw { Some(v) => { @@ -264,9 +318,9 @@ fn husk_reap_max_age_from_env() -> Option { /// Every wire invocation inside an agent terminal mints a /// `sessions/by-key//` home via session adoption (RFC-008), even /// for read-only commands, and nothing ever deleted them — a dev box -/// accumulated 175 empty dirs in two weeks. The idle filter -/// (`supervisor_eligible`) stops the daemon fork-storm but leaves the -/// dirs. This is the complement: the filter hides, the reaper removes. +/// accumulated 175 empty dirs in two weeks. Lifecycle planning prevents +/// those homes from starting workers; this reaper removes old identityless +/// husks without touching initialized identities. /// /// A dir is reaped only if ALL of these hold: /// - its name has the by-key shape (exactly 16 lowercase hex chars, @@ -349,7 +403,7 @@ struct ChildState { /// Entrypoint for `wire daemon --all-sessions`. Loops forever; only /// returns Err on a setup error (e.g. cannot resolve sessions_root). -pub fn run_supervisor(interval_secs: u64, as_json: bool) -> Result<()> { +pub fn run_supervisor(interval_secs: u64, max_workers: usize, as_json: bool) -> Result<()> { // Supervisor singleton — one per machine. Separate pidfile from the // per-session daemon pidfile so the two layers can't collide. let pid_path = supervisor_pid_path()?; @@ -375,7 +429,7 @@ pub fn run_supervisor(interval_secs: u64, as_json: bool) -> Result<()> { if !as_json { eprintln!( - "wire daemon --all-sessions: supervisor up. interval={interval_secs}s, registry-poll={REGISTRY_POLL_SECS}s. SIGINT to stop." + "wire daemon --all-sessions: supervisor up. interval={interval_secs}s, registry-poll={REGISTRY_POLL_SECS}s, max-workers={max_workers}. SIGINT to stop." ); } else { println!( @@ -384,21 +438,11 @@ pub fn run_supervisor(interval_secs: u64, as_json: bool) -> Result<()> { "status": "supervisor_started", "interval_secs": interval_secs, "registry_poll_secs": REGISTRY_POLL_SECS, + "max_workers": max_workers, }) ); } - // Idle cutoff for registry-unbound sessions — read once at startup - // (env doesn't change under a running supervisor). - let max_idle = max_idle_from_env(); - eprintln!( - "supervisor: idle cutoff for unbound sessions = {}", - match max_idle { - Some(d) => format!("{} days", d.as_secs() / 86_400), - None => "disabled (spawn-for-all)".to_string(), - } - ); - // Husk reap cutoff — also read once at startup. let husk_max_age = husk_reap_max_age_from_env(); eprintln!( @@ -429,15 +473,7 @@ pub fn run_supervisor(interval_secs: u64, as_json: bool) -> Result<()> { "supervisor: child '{name}' exited (status={status:?}, lived={}s, rapid={rapid})", lived.as_secs() ); - let next_backoff = if rapid { - let prev = session_backoff - .get(name) - .copied() - .unwrap_or(INITIAL_BACKOFF); - (prev * 2).min(MAX_BACKOFF) - } else { - INITIAL_BACKOFF - }; + let next_backoff = next_worker_backoff(session_backoff.get(name).copied(), rapid); session_backoff.insert(name.clone(), next_backoff); session_last_exit.insert(name.clone(), Instant::now()); exited.push(name.clone()); @@ -447,26 +483,35 @@ pub fn run_supervisor(interval_secs: u64, as_json: bool) -> Result<()> { children.remove(&n); } - // 2. Read registry, identify wanted sessions. Filter out - // registry-unbound sessions that have been idle past the - // cutoff so the supervisor doesn't fan out a daemon per - // every ephemeral persona home (the 147-home fork storm). + // 2. Read registry and select only explicitly live sessions. + // Private keys and sync timestamps are historical state, not + // liveness signals. let all_sessions = crate::session::list_sessions().unwrap_or_default(); let total_sessions = all_sessions.len(); - let wanted: Vec = supervisor_eligible( - all_sessions, - max_idle, - SystemTime::now(), - fs_last_active, - fs_has_identity, + for session in &all_sessions { + crate::session_lifecycle::prune_stale_leases_at( + &session.home_dir, + time::OffsetDateTime::now_utc(), + crate::platform::process_alive, + ); + } + let plan = plan_supervisor_sessions( + all_sessions.clone(), + max_workers, + fs_has_live_lease, + fs_has_pending_outbox, fs_is_retired, ); - if wanted.len() != total_sessions { + let wanted = plan.selected; + if wanted.len() != total_sessions || !plan.queued.is_empty() { eprintln!( - "supervisor: {} of {} sessions eligible (skipped {} registry-unbound + idle > cutoff)", + "supervisor: {} running target(s), {} queued, {} inactive, {} retired from {} discovered (cap {})", wanted.len(), + plan.queued.len(), + plan.inactive.len(), + plan.retired.len(), total_sessions, - total_sessions - wanted.len() + max_workers, ); } @@ -525,6 +570,11 @@ pub fn run_supervisor(interval_secs: u64, as_json: bool) -> Result<()> { let _ = state.child.wait(); } } + for session in &all_sessions { + if !wanted_names.contains(&session.name) { + retire_inactive_worker(session); + } + } // 4. Spawn missing children, respecting backoff + existing // pidfiles (operator-spawned daemons coexist). @@ -574,11 +624,8 @@ pub fn run_supervisor(interval_secs: u64, as_json: bool) -> Result<()> { ); // Treat spawn failure as a rapid failure so the // backoff curve kicks in. - let prev = session_backoff - .get(&info.name) - .copied() - .unwrap_or(INITIAL_BACKOFF); - session_backoff.insert(info.name.clone(), (prev * 2).min(MAX_BACKOFF)); + let next = next_worker_backoff(session_backoff.get(&info.name).copied(), true); + session_backoff.insert(info.name.clone(), next); session_last_exit.insert(info.name.clone(), Instant::now()); } } @@ -695,15 +742,10 @@ pub struct SupervisorState { /// future `wire upgrade --refresh-stale-children` to force the /// supervisor to respawn them on the current binary. pub stale_binary_sessions: Vec, - /// v0.16.x (#275): the subset of `stale_binary_sessions` the - /// `--all-sessions` supervisor would NOT respawn — i.e. sessions - /// the supervisor's eligibility filter (`supervisor_eligible`: - /// registry-bound OR active within the idle cutoff) drops. Killing - /// one of these (which `wire upgrade --refresh-stale-children` used - /// to do indiscriminately) orphans it: the supervisor never brings - /// it back, so the identity silently stops syncing. `wire upgrade` - /// must NOT kill these — it surfaces them as "relaunch manually" - /// instead. + /// Subset of `stale_binary_sessions` that lacks a registry binding, + /// live lease, or pending outbox. The supervisor will not respawn + /// these inactive sessions, so upgrade must not kill them under the + /// assumption that a replacement worker will appear. pub stale_unmanaged_sessions: Vec, } @@ -747,19 +789,17 @@ pub fn read_supervisor_state() -> Result { // pidfile + last_sync. let infos = crate::session::list_sessions().unwrap_or_default(); - // #275: the names the `--all-sessions` supervisor would actually own a - // daemon for, computed with the SAME predicate the supervisor loop uses - // (`supervisor_eligible`: registry-bound OR active within the idle cutoff). - // Used below to flag stale sessions the supervisor will NOT respawn, so - // `wire upgrade --refresh-stale-children` doesn't kill-and-orphan them. - let eligible_names: std::collections::HashSet = supervisor_eligible( + // Use the same lifecycle predicate as the supervisor. No cap here: this + // set answers whether a worker is respawnable eventually, not whether it + // occupies a slot in this poll. + let eligible_names: std::collections::HashSet = plan_supervisor_sessions( infos.clone(), - max_idle_from_env(), - SystemTime::now(), - fs_last_active, - fs_has_identity, + usize::MAX, + fs_has_live_lease, + fs_has_pending_outbox, fs_is_retired, ) + .selected .into_iter() .map(|s| s.name) .collect(); @@ -929,6 +969,17 @@ fn supervisor_pid_path() -> Result { Ok(root.join("supervisor.pid")) } +/// True when the machine-wide all-session supervisor owns daemon lifecycle. +/// Read-only and fail-closed: a missing, corrupt, or dead pidfile does not +/// suppress normal single-session recovery. +pub fn supervisor_is_alive() -> bool { + supervisor_pid_path() + .and_then(|path| read_alive_supervisor_pid(&path)) + .ok() + .flatten() + .is_some() +} + fn read_alive_supervisor_pid(path: &std::path::Path) -> Result> { if !path.exists() { return Ok(None); @@ -1064,53 +1115,50 @@ mod tests { assert!(existing_daemon_for_session(tmp.path()).unwrap()); } - // ---- supervisor eligibility filter (the 147-home fork-storm fix) ---- - - fn mk_session(name: &str, cwd: Option<&str>) -> crate::session::SessionInfo { - crate::session::SessionInfo { - name: name.to_string(), - cwd: cwd.map(String::from), - home_dir: PathBuf::from(format!("/sessions/{name}")), - did: None, - handle: None, - daemon_running: false, - character: None, - } - } - #[test] - fn parse_max_idle_default_when_unset() { + fn inactive_worker_classification_distinguishes_version_skew() { + let mut record = crate::ensure_up::DaemonPid { + schema: crate::ensure_up::DAEMON_PID_SCHEMA.to_string(), + pid: 4242, + bin_path: "/opt/wire".to_string(), + version: env!("CARGO_PKG_VERSION").to_string(), + started_at: "2026-07-17T00:00:00Z".to_string(), + did: None, + relay_url: None, + }; assert_eq!( - parse_max_idle(None), - Some(Duration::from_secs(DEFAULT_MAX_IDLE_DAYS * 86_400)) + classify_session_worker(&record, true, Some("/opt/wire daemon --interval 5")), + Some(WorkerVersion::Current) ); - } - - #[test] - fn parse_max_idle_zero_disables_filter() { - assert_eq!(parse_max_idle(Some("0")), None); - } - - #[test] - fn parse_max_idle_explicit_days() { + record.version = "0.16.0".to_string(); assert_eq!( - parse_max_idle(Some("3")), - Some(Duration::from_secs(3 * 86_400)) + classify_session_worker(&record, true, Some("/opt/wire daemon --interval 5")), + Some(WorkerVersion::Skewed) ); assert_eq!( - parse_max_idle(Some(" 14 ")), - Some(Duration::from_secs(14 * 86_400)) + classify_session_worker(&record, true, Some("/opt/wire daemon --all-sessions")), + None, + "never classify the supervisor itself as a session worker" ); + assert_eq!(classify_session_worker(&record, false, None), None); } #[test] - fn parse_max_idle_garbage_falls_back_to_default() { + fn crashed_worker_backoff_is_bounded_and_healthy_run_resets() { + assert_eq!(next_worker_backoff(None, true), Duration::from_secs(2)); assert_eq!( - parse_max_idle(Some("not-a-number")), - Some(Duration::from_secs(DEFAULT_MAX_IDLE_DAYS * 86_400)) + next_worker_backoff(Some(Duration::from_secs(32)), true), + MAX_BACKOFF + ); + assert_eq!(next_worker_backoff(Some(MAX_BACKOFF), true), MAX_BACKOFF); + assert_eq!( + next_worker_backoff(Some(MAX_BACKOFF), false), + INITIAL_BACKOFF ); } + // ---- stale binary lifecycle partitioning ---- + #[test] fn partition_stale_splits_respawnable_from_unmanaged() { // #275: stale sessions the supervisor would respawn (eligible) vs ones @@ -1141,141 +1189,6 @@ mod tests { assert_eq!(unmanaged, stale); } - #[test] - fn eligible_keeps_cwd_bound_even_when_ancient() { - // A registry-bound session is kept no matter how idle — the - // operator deliberately attached it to a project dir. (This is - // the real-world case: the cwd-bound `wire`/`slancha-*` sessions - // were the *oldest* on the box, yet must survive.) - let now = SystemTime::now(); - let ancient = now - Duration::from_secs(365 * 86_400); - let sessions = vec![mk_session("wire", Some("/Users/p/Source/wire"))]; - let out = supervisor_eligible( - sessions, - Some(Duration::from_secs(7 * 86_400)), - now, - |_| Some(ancient), - |_| false, - |_| false, - ); - assert_eq!(out.len(), 1); - assert_eq!(out[0].name, "wire"); - } - - #[test] - fn eligible_keeps_unbound_recent_drops_unbound_idle() { - // The live-but-unbound persona sessions (each Claude tab) are - // recent → kept. The abandoned ones are idle → dropped. - let now = SystemTime::now(); - let recent = now - Duration::from_secs(2 * 86_400); - let stale = now - Duration::from_secs(30 * 86_400); - let sessions = vec![ - mk_session("rosy-rook", None), // live tab - mk_session("agate-nimbus", None), // abandoned - ]; - let out = supervisor_eligible( - sessions, - Some(Duration::from_secs(7 * 86_400)), - now, - |home| { - if home.ends_with("rosy-rook") { - Some(recent) - } else { - Some(stale) - } - }, - |_| false, - |_| false, - ); - let names: Vec<_> = out.iter().map(|s| s.name.as_str()).collect(); - assert_eq!(names, vec!["rosy-rook"]); - } - - #[test] - fn eligible_drops_unbound_with_no_activity_signal() { - // A never-synced husk (no activity files at all) and no cwd → - // dropped: nothing says it's a session anyone is using. - let now = SystemTime::now(); - let sessions = vec![mk_session("husk", None)]; - let out = supervisor_eligible( - sessions, - Some(Duration::from_secs(7 * 86_400)), - now, - |_| None, - |_| false, - |_| false, - ); - assert!(out.is_empty()); - } - - #[test] - fn eligible_keeps_unbound_idle_session_with_identity() { - // The wildflower-gleam catch-22, generalized: no cwd binding, never - // synced (last_active None → normally dropped). A home with a real - // identity (private.key) MUST stay eligible so a daemon spawns to push - // its outbox AND pull inbound mail — otherwise it strands forever - // (ineligible → no daemon → never syncs → never eligible). The - // identity-less husk sibling is still correctly dropped. - let now = SystemTime::now(); - let sessions = vec![mk_session("real", None), mk_session("husk", None)]; - let out = supervisor_eligible( - sessions, - Some(Duration::from_secs(7 * 86_400)), - now, - |_| None, // neither has ever synced - |home| home.ends_with("real"), // only this one has a private.key - |_| false, - ); - let names: Vec<_> = out.iter().map(|s| s.name.as_str()).collect(); - assert_eq!(names, vec!["real"]); - } - - #[test] - fn eligible_drops_retired_even_with_cwd_identity_or_no_cutoff() { - // The retire invariant: a `.retired` home is ineligible in EVERY - // configuration — with a cwd binding, with an identity, and even under - // `max_idle = None` (the spawn-for-all override). All three would - // otherwise keep it eligible; the retired filter must beat them all. - let now = SystemTime::now(); - let sessions = vec![ - mk_session("retired-proj", Some("/Users/p/proj")), // cwd-bound - mk_session("retired-id", None), // identity-ful - mk_session("live", Some("/Users/p/live")), - ]; - let retired = |home: &Path| home.to_string_lossy().contains("retired-"); - // With the 7-day cutoff. - let out = supervisor_eligible( - sessions.clone(), - Some(Duration::from_secs(7 * 86_400)), - now, - |_| Some(now), - |home| home.ends_with("retired-id"), // retired-id has identity - retired, - ); - assert_eq!( - out.iter().map(|s| s.name.as_str()).collect::>(), - vec!["live"], - "both retired homes dropped despite cwd/identity" - ); - // And under the max_idle=None spawn-for-all override. - let out2 = supervisor_eligible(sessions, None, now, |_| Some(now), |_| true, retired); - assert_eq!( - out2.iter().map(|s| s.name.as_str()).collect::>(), - vec!["live"], - "retired dropped even with max_idle=None" - ); - } - - #[test] - fn eligible_none_cutoff_keeps_everything() { - // Override = 0 (max_idle None) restores legacy spawn-for-all. - let now = SystemTime::now(); - let ancient = now - Duration::from_secs(999 * 86_400); - let sessions = vec![mk_session("husk", None), mk_session("agate-nimbus", None)]; - let out = supervisor_eligible(sessions, None, now, |_| Some(ancient), |_| false, |_| false); - assert_eq!(out.len(), 2); - } - // ---- husk reaper ---- use std::collections::HashSet; @@ -1439,4 +1352,89 @@ mod tests { Some(Duration::from_secs(48 * 3600)) ); } + + fn initialized_session(name: &str, bound: bool) -> crate::session::SessionInfo { + crate::session::SessionInfo { + name: name.to_string(), + cwd: bound.then(|| format!("/projects/{name}")), + home_dir: PathBuf::from(format!("/sessions/{name}")), + did: Some(format!("did:wire:{name}-1234abcd")), + handle: Some(format!("{name}-1234abcd")), + daemon_running: false, + character: None, + } + } + + #[test] + fn planner_covers_one_ten_and_655_homes_with_hard_cap() { + for total in [1usize, 10, 655] { + let sessions: Vec<_> = (0..total) + .map(|i| initialized_session(&format!("s{i:03}"), false)) + .collect(); + let plan = plan_supervisor_sessions(sessions, 16, |_| true, |_| false, |_| false); + assert_eq!(plan.selected.len(), total.min(16)); + assert_eq!(plan.queued.len(), total.saturating_sub(16)); + assert!(plan.inactive.is_empty()); + } + } + + #[test] + fn planner_keeps_five_live_and_leaves_650_stale_inactive() { + let sessions: Vec<_> = (0..655) + .map(|i| initialized_session(&format!("s{i:03}"), false)) + .collect(); + let plan = plan_supervisor_sessions( + sessions, + 16, + |home| { + home.file_name() + .and_then(|n| n.to_str()) + .and_then(|n| n.strip_prefix('s')) + .and_then(|n| n.parse::().ok()) + .is_some_and(|i| i < 5) + }, + |_| false, + |_| false, + ); + assert_eq!(plan.selected.len(), 5); + assert!(plan.queued.is_empty()); + assert_eq!(plan.inactive.len(), 650); + } + + #[test] + fn private_key_without_bound_lease_or_outbox_is_inactive() { + let tmp = tempdir().unwrap(); + let config = tmp.path().join("config/wire"); + std::fs::create_dir_all(&config).unwrap(); + std::fs::write(config.join("private.key"), [7u8; 32]).unwrap(); + let mut session = initialized_session("historical", false); + session.home_dir = tmp.path().to_path_buf(); + + let plan = plan_supervisor_sessions(vec![session], 16, |_| false, |_| false, |_| false); + assert!(plan.selected.is_empty()); + assert_eq!(plan.inactive, vec!["historical"]); + } + + #[test] + fn planner_state_is_restart_stable_and_retirement_wins() { + let sessions = vec![ + initialized_session("bound", true), + initialized_session("leased", false), + initialized_session("retired", true), + ]; + let build = || { + plan_supervisor_sessions( + sessions.clone(), + 16, + |home| home.ends_with("leased"), + |_| false, + |home| home.ends_with("retired"), + ) + }; + let before = build(); + let after_restart = build(); + assert_eq!(before.selected_names(), after_restart.selected_names()); + assert_eq!(before.retired, vec!["retired"]); + assert_eq!(before.selected_names(), vec!["bound", "leased"]); + } } diff --git a/src/mcp.rs b/src/mcp.rs index ff4a235..5ba8ebf 100644 --- a/src/mcp.rs +++ b/src/mcp.rs @@ -117,9 +117,11 @@ pub fn run() -> Result<()> { // identity (and early-returns for already-initialized homes), so arm the // daemon unconditionally here. Idempotent (singleton-guarded) and gated on // an existing identity + the same skip env bootstrap honors. - if std::env::var("WIRE_MCP_SKIP_AUTO_UP").is_err() - && crate::config::is_initialized().unwrap_or(false) - { + if should_start_embedded_daemon( + std::env::var("WIRE_MCP_SKIP_AUTO_UP").is_err(), + crate::config::is_initialized().unwrap_or(false), + crate::daemon_supervisor::supervisor_is_alive(), + ) { let _ = crate::ensure_up::ensure_daemon_running(); } @@ -1607,6 +1609,14 @@ fn ensure_session_bootstrapped() { } } +fn should_start_embedded_daemon( + auto_up_enabled: bool, + initialized: bool, + supervisor_alive: bool, +) -> bool { + auto_up_enabled && initialized && !supervisor_alive +} + fn tool_init(args: &Value) -> Result { let handle = args .get("handle") @@ -2152,6 +2162,14 @@ fn error_response(id: &Value, code: i32, message: &str) -> Value { mod tests { use super::*; + #[test] + fn supervisor_ownership_suppresses_per_mcp_daemon() { + assert!(should_start_embedded_daemon(true, true, false)); + assert!(!should_start_embedded_daemon(true, true, true)); + assert!(!should_start_embedded_daemon(false, true, false)); + assert!(!should_start_embedded_daemon(true, false, false)); + } + #[test] fn mcp_stale_binary_note_flags_only_real_mismatch() { // No daemon version to compare → no note. diff --git a/src/service.rs b/src/service.rs index 8bc265d..202a0b7 100644 --- a/src/service.rs +++ b/src/service.rs @@ -47,6 +47,59 @@ pub enum ServiceKind { LocalRelay, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DaemonServiceOptions { + pub interval_secs: u64, + pub max_workers: usize, +} + +impl Default for DaemonServiceOptions { + fn default() -> Self { + Self { + interval_secs: 5, + max_workers: crate::daemon_supervisor::DEFAULT_MAX_WORKERS, + } + } +} + +fn value_after_flag(text: &str, flag: &str) -> Option { + let tail = text.split_once(flag)?.1; + let digits = tail + .trim_start_matches(|c: char| !c.is_ascii_digit()) + .chars() + .take_while(char::is_ascii_digit) + .collect::(); + digits.parse().ok() +} + +fn parse_daemon_service_options(text: &str) -> Option { + let interval_secs = value_after_flag(text, "--interval")?; + let max_workers = value_after_flag(text, "--max-workers")? as usize; + (interval_secs > 0 && max_workers > 0).then_some(DaemonServiceOptions { + interval_secs, + max_workers, + }) +} + +fn resolve_daemon_service_options( + existing: Option, + interval_secs: Option, + max_workers: Option, +) -> Result { + let baseline = existing.unwrap_or_default(); + let resolved = DaemonServiceOptions { + interval_secs: interval_secs.unwrap_or(baseline.interval_secs), + max_workers: max_workers.unwrap_or(baseline.max_workers), + }; + if resolved.interval_secs == 0 { + bail!("service daemon --interval must be at least 1"); + } + if resolved.max_workers == 0 { + bail!("service daemon --max-workers must be at least 1"); + } + Ok(resolved) +} + impl ServiceKind { /// launchd Label / systemd unit base name (without `.service`). fn label(self) -> &'static str { @@ -83,12 +136,22 @@ impl ServiceKind { /// isolation gap. Operators upgrading from a pre-0.14.2 install /// must re-run `wire service install` (or `wire upgrade /// --restart-service`) to pick up the new ProgramArguments line. - fn binary_args(self) -> &'static [&'static str] { + fn binary_args(self, daemon_options: DaemonServiceOptions) -> Vec { match self { - ServiceKind::Daemon => &["daemon", "--all-sessions", "--interval", "5"], - ServiceKind::LocalRelay => { - &["relay-server", "--bind", "127.0.0.1:8771", "--local-only"] - } + ServiceKind::Daemon => vec![ + "daemon".into(), + "--all-sessions".into(), + "--interval".into(), + daemon_options.interval_secs.to_string(), + "--max-workers".into(), + daemon_options.max_workers.to_string(), + ], + ServiceKind::LocalRelay => vec![ + "relay-server".into(), + "--bind".into(), + "127.0.0.1:8771".into(), + "--local-only".into(), + ], } } @@ -153,6 +216,25 @@ pub fn status() -> Result { /// Install a user-scope service unit for the given kind. pub fn install_kind(kind: ServiceKind) -> Result { + install_kind_with_options(kind, None, None) +} + +pub fn install_kind_with_options( + kind: ServiceKind, + interval_secs: Option, + max_workers: Option, +) -> Result { + let daemon_options = if kind == ServiceKind::Daemon { + let existing = existing_service_text(kind) + .as_deref() + .and_then(parse_daemon_service_options); + resolve_daemon_service_options(existing, interval_secs, max_workers)? + } else { + if interval_secs.is_some() || max_workers.is_some() { + bail!("--interval and --max-workers apply only to the daemon service"); + } + DaemonServiceOptions::default() + }; // Robust to a `cargo install` in-place replace mid-`wire upgrade`: the // kernel marks `/proc/self/exe` with a trailing ` (deleted)` that, written // verbatim into ExecStart=, corrupts the unit (issue #274). @@ -175,7 +257,7 @@ pub fn install_kind(kind: ServiceKind) -> Result { if let Some(parent) = plist_path.parent() { std::fs::create_dir_all(parent).with_context(|| format!("creating {parent:?}"))?; } - let plist = launchd_plist_xml(kind, &exe_str, &log_str); + let plist = launchd_plist_xml(kind, &exe_str, &log_str, daemon_options); std::fs::write(&plist_path, plist).with_context(|| format!("writing {plist_path:?}"))?; // launchctl bootstrap is idempotent if we bootout first. @@ -201,7 +283,10 @@ pub fn install_kind(kind: ServiceKind) -> Result { "written".into() }, detail: if loaded { - format!("plist written + bootstrapped; logs at {log_str}") + format!( + "plist written + bootstrapped; interval={}s, max-workers={}; logs at {log_str}", + daemon_options.interval_secs, daemon_options.max_workers + ) } else { format!( "plist written; `launchctl bootstrap` failed — try `launchctl bootstrap {} {}` manually", @@ -217,7 +302,7 @@ pub fn install_kind(kind: ServiceKind) -> Result { if let Some(parent) = unit_path.parent() { std::fs::create_dir_all(parent).with_context(|| format!("creating {parent:?}"))?; } - let unit = systemd_unit_text(kind, &exe_str); + let unit = systemd_unit_text(kind, &exe_str, daemon_options); std::fs::write(&unit_path, unit).with_context(|| format!("writing {unit_path:?}"))?; // Reload + enable + start. Each is idempotent on linux. @@ -273,7 +358,7 @@ pub fn install_kind(kind: ServiceKind) -> Result { } if cfg!(target_os = "windows") { let task_name = kind.windows_task_name(); - let xml = windows_task_xml(kind, &exe_str); + let xml = windows_task_xml(kind, &exe_str, daemon_options); // schtasks /Create /XML reads the file at the given path. UTF-8 // without BOM is accepted on Win10+; older builds expected // UTF-16LE-BOM. We write UTF-8 — if a user hits a parse error @@ -579,9 +664,71 @@ fn ensure_macos_log_path(_kind: ServiceKind) -> Result { Ok(PathBuf::new()) } -fn launchd_plist_xml(kind: ServiceKind, exe: &str, log_path: &str) -> String { +fn existing_service_text(kind: ServiceKind) -> Option { + if cfg!(target_os = "macos") { + return launchd_plist_path(kind) + .ok() + .and_then(|path| std::fs::read_to_string(path).ok()); + } + if cfg!(target_os = "linux") { + return systemd_unit_path(kind) + .ok() + .and_then(|path| std::fs::read_to_string(path).ok()); + } + if cfg!(target_os = "windows") { + let output = Command::new("schtasks.exe") + .args(["/Query", "/TN", kind.windows_task_name(), "/XML"]) + .output() + .ok()?; + return output + .status + .success() + .then(|| String::from_utf8_lossy(&output.stdout).into_owned()); + } + None +} + +fn parse_service_executable(text: &str) -> Option { + if let Some(program_args) = text.split_once("ProgramArguments") { + let after = program_args.1.split_once("")?.1; + return after + .split_once("") + .map(|(exe, _)| exe.to_string()); + } + if let Some(exec_start) = text + .lines() + .find_map(|line| line.strip_prefix("ExecStart=")) + { + return exec_start.split_whitespace().next().map(str::to_string); + } + let after = text.split_once("")?.1; + after + .split_once("") + .map(|(exe, _)| exe.to_string()) +} + +pub fn installed_daemon_options() -> DaemonServiceOptions { + existing_service_text(ServiceKind::Daemon) + .as_deref() + .and_then(parse_daemon_service_options) + .unwrap_or_default() +} + +pub fn installed_daemon_executable() -> Option { + existing_service_text(ServiceKind::Daemon) + .as_deref() + .and_then(parse_service_executable) + .map(PathBuf::from) +} + +fn launchd_plist_xml( + kind: ServiceKind, + exe: &str, + log_path: &str, + daemon_options: DaemonServiceOptions, +) -> String { let args_xml = kind - .binary_args() + .binary_args(daemon_options) .iter() .map(|a| format!(" {a}")) .collect::>() @@ -624,8 +771,8 @@ fn systemd_unit_path(kind: ServiceKind) -> Result { .join(kind.systemd_unit_name())) } -fn systemd_unit_text(kind: ServiceKind, exe: &str) -> String { - let args = kind.binary_args().join(" "); +fn systemd_unit_text(kind: ServiceKind, exe: &str, daemon_options: DaemonServiceOptions) -> String { + let args = kind.binary_args(daemon_options).join(" "); let desc = kind.description(); format!( r#"[Unit] @@ -655,9 +802,9 @@ WantedBy=default.target /// /// Returned as a String for `cfg!(test)` cross-target compilation; the /// caller writes it to disk via `std::fs::write` which handles encoding. -fn windows_task_xml(kind: ServiceKind, exe: &str) -> String { +fn windows_task_xml(kind: ServiceKind, exe: &str, daemon_options: DaemonServiceOptions) -> String { let desc = kind.description(); - let args = kind.binary_args().join(" "); + let args = kind.binary_args(daemon_options).join(" "); // Escape XML special chars in fields that take operator-influenced // strings. exe is `std::env::current_exe()` (trusted) but args may // grow operator-passed values later. @@ -726,12 +873,59 @@ fn xml_escape(s: &str) -> String { mod tests { use super::*; + #[test] + fn daemon_service_options_preserve_then_override_existing_values() { + let launchd = "--interval17--max-workers9"; + let systemd = "ExecStart=/opt/wire daemon --all-sessions --interval 23 --max-workers 11"; + let task = "daemon --all-sessions --interval 31 --max-workers 13"; + for (text, expected) in [(launchd, (17, 9)), (systemd, (23, 11)), (task, (31, 13))] { + let parsed = parse_daemon_service_options(text).unwrap(); + assert_eq!((parsed.interval_secs, parsed.max_workers), expected); + let resolved = resolve_daemon_service_options(Some(parsed), Some(41), None).unwrap(); + assert_eq!(resolved.interval_secs, 41); + assert_eq!(resolved.max_workers, expected.1); + } + } + + #[test] + fn daemon_service_options_first_install_defaults_and_rejects_zero() { + let options = resolve_daemon_service_options(None, None, None).unwrap(); + assert_eq!(options.interval_secs, 5); + assert_eq!( + options.max_workers, + crate::daemon_supervisor::DEFAULT_MAX_WORKERS + ); + assert!(resolve_daemon_service_options(None, Some(0), None).is_err()); + assert!(resolve_daemon_service_options(None, None, Some(0)).is_err()); + } + + #[test] + fn installed_executable_parser_handles_all_service_formats() { + assert_eq!( + parse_service_executable( + "ProgramArguments/opt/wiredaemon" + ) + .as_deref(), + Some("/opt/wire") + ); + assert_eq!( + parse_service_executable("ExecStart=/usr/local/bin/wire daemon --all-sessions") + .as_deref(), + Some("/usr/local/bin/wire") + ); + assert_eq!( + parse_service_executable("C:\\wire\\wire.exe").as_deref(), + Some("C:\\wire\\wire.exe") + ); + } + #[test] fn launchd_plist_xml_for_daemon_contains_required_keys() { let xml = launchd_plist_xml( ServiceKind::Daemon, "/usr/local/bin/wire", "/tmp/wire-daemon.log", + DaemonServiceOptions::default(), ); assert!(xml.contains("Label")); assert!(xml.contains(ServiceKind::Daemon.label())); @@ -739,6 +933,7 @@ mod tests { assert!(xml.contains("daemon")); assert!(xml.contains("--all-sessions")); assert!(xml.contains("--interval")); + assert!(xml.contains("--max-workers")); assert!(xml.contains("KeepAlive")); assert!(xml.contains("RunAtLoad")); assert!(xml.contains("")); @@ -753,6 +948,7 @@ mod tests { ServiceKind::LocalRelay, "/usr/local/bin/wire", "/tmp/wire-local-relay.log", + DaemonServiceOptions::default(), ); assert!(xml.contains(ServiceKind::LocalRelay.label())); assert!(xml.contains("relay-server")); @@ -765,18 +961,30 @@ mod tests { #[test] fn systemd_unit_text_for_daemon_contains_required_directives() { - let unit = systemd_unit_text(ServiceKind::Daemon, "/usr/local/bin/wire"); + let unit = systemd_unit_text( + ServiceKind::Daemon, + "/usr/local/bin/wire", + DaemonServiceOptions::default(), + ); assert!(unit.contains("[Unit]")); assert!(unit.contains("[Service]")); assert!(unit.contains("[Install]")); - assert!(unit.contains("/usr/local/bin/wire daemon --all-sessions --interval 5")); + assert!( + unit.contains( + "/usr/local/bin/wire daemon --all-sessions --interval 5 --max-workers 16" + ) + ); assert!(unit.contains("Restart=on-failure")); assert!(unit.contains("WantedBy=default.target")); } #[test] fn systemd_unit_text_for_local_relay_uses_correct_exec() { - let unit = systemd_unit_text(ServiceKind::LocalRelay, "/usr/local/bin/wire"); + let unit = systemd_unit_text( + ServiceKind::LocalRelay, + "/usr/local/bin/wire", + DaemonServiceOptions::default(), + ); assert!( unit.contains("/usr/local/bin/wire relay-server --bind 127.0.0.1:8771 --local-only") ); @@ -804,7 +1012,11 @@ mod tests { #[test] fn windows_task_xml_for_daemon_contains_required_elements_v0_7_2() { - let xml = windows_task_xml(ServiceKind::Daemon, r"C:\Program Files\wire\wire.exe"); + let xml = windows_task_xml( + ServiceKind::Daemon, + r"C:\Program Files\wire\wire.exe", + DaemonServiceOptions::default(), + ); // Schema declaration + 1.2 task version (Win 7+ / matches what // schtasks /XML expects). assert!(xml.contains(r#""#)); @@ -827,12 +1039,18 @@ mod tests { // Actual exec line uses XML-escaped exe path + correct daemon // args. assert!(xml.contains(r"C:\Program Files\wire\wire.exe")); - assert!(xml.contains("daemon --all-sessions --interval 5")); + assert!(xml.contains( + "daemon --all-sessions --interval 5 --max-workers 16" + )); } #[test] fn windows_task_xml_for_local_relay_uses_correct_args_v0_7_2() { - let xml = windows_task_xml(ServiceKind::LocalRelay, r"C:\wire\wire.exe"); + let xml = windows_task_xml( + ServiceKind::LocalRelay, + r"C:\wire\wire.exe", + DaemonServiceOptions::default(), + ); assert!(xml.contains(r"C:\wire\wire.exe")); assert!( xml.contains("relay-server --bind 127.0.0.1:8771 --local-only") diff --git a/tests/supervisor_bounded.rs b/tests/supervisor_bounded.rs new file mode 100644 index 0000000..4d478de --- /dev/null +++ b/tests/supervisor_bounded.rs @@ -0,0 +1,94 @@ +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::time::Duration; + +fn wire_bin() -> PathBuf { + PathBuf::from(env!("CARGO_BIN_EXE_wire")) +} + +fn make_stale_homes(root: &Path, count: usize) { + for index in 0..count { + let home = root + .join("sessions") + .join("by-key") + .join(format!("{index:016x}")); + let config = home.join("config").join("wire"); + std::fs::create_dir_all(&config).unwrap(); + std::fs::write( + config.join("agent-card.json"), + serde_json::to_vec(&serde_json::json!({ + "did": format!("did:wire:stale-{index:04}-00000000"), + "handle": format!("stale-{index:04}"), + })) + .unwrap(), + ) + .unwrap(); + } +} + +fn spawn_supervisor(root: &Path) -> Child { + Command::new(wire_bin()) + .args([ + "daemon", + "--all-sessions", + "--interval", + "60", + "--max-workers", + "4", + ]) + .env("WIRE_HOME", root) + .env("WIRE_HOME_FORCE", "1") + .env_remove("WIRE_SESSION_ID") + .env_remove("CLAUDE_CODE_SESSION_ID") + .env_remove("CODEX_SESSION_ID") + .env_remove("COPILOT_AGENT_SESSION_ID") + .env_remove("VSCODE_GIT_REPOSITORY_ROOT") + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn isolated supervisor") +} + +fn direct_children(pid: u32) -> usize { + let output = Command::new("ps") + .args(["-axo", "ppid="]) + .output() + .expect("read process parents"); + String::from_utf8_lossy(&output.stdout) + .lines() + .filter_map(|line| line.trim().parse::().ok()) + .filter(|parent| *parent == pid) + .count() +} + +fn rss_kib(pid: u32) -> u64 { + let output = Command::new("ps") + .args(["-o", "rss=", "-p", &pid.to_string()]) + .output() + .expect("read supervisor RSS"); + String::from_utf8_lossy(&output.stdout) + .trim() + .parse() + .unwrap_or(u64::MAX) +} + +#[test] +fn stale_655_home_supervisor_is_bounded_across_restart() { + let temp = tempfile::tempdir().unwrap(); + make_stale_homes(temp.path(), 655); + + for restart in 0..2 { + let mut supervisor = spawn_supervisor(temp.path()); + std::thread::sleep(Duration::from_millis(900)); + assert!(supervisor.try_wait().unwrap().is_none()); + let children = direct_children(supervisor.id()); + let rss = rss_kib(supervisor.id()); + eprintln!("restart={restart} children={children} rss_kib={rss}"); + assert!(children <= 4, "worker count exceeded configured cap"); + assert!(rss < 256 * 1024, "supervisor RSS exceeded 256 MiB"); + supervisor.kill().unwrap(); + supervisor.wait().unwrap(); + std::thread::sleep(Duration::from_millis(100)); + } +} From 2915c00c196661b3599888ef3c1dc4993509acb9 Mon Sep 17 00:00:00 2001 From: Paul Logan Date: Sat, 18 Jul 2026 00:19:32 -0700 Subject: [PATCH 4/6] fix: converge local sister pairing --- SESSION_LOG_2026_07_17.md | 65 ++++++++++++++++ src/cli/pairing.rs | 85 ++++++++++++++++++++- src/cli/session.rs | 65 ++++------------ src/mcp.rs | 1 + src/send.rs | 10 ++- tests/stress_within_system.rs | 136 ++++++++++++++++++++++++++++++++++ 6 files changed, 306 insertions(+), 56 deletions(-) diff --git a/SESSION_LOG_2026_07_17.md b/SESSION_LOG_2026_07_17.md index 938e57d..cff9f0a 100644 --- a/SESSION_LOG_2026_07_17.md +++ b/SESSION_LOG_2026_07_17.md @@ -88,3 +88,68 @@ literal Codex override. for process lifetime and disappears on clean stdin EOF. - MCP now writes the lease before daemon orchestration and renews it from the existing watcher thread every 30 seconds. No raw session key is persisted. + +## Iteration 2 — bounded supervisor + +- RED: planner tests failed before `plan_supervisor_sessions` existed. +- GREEN: 1-, 10-, and 655-home fixtures pass. A 655-home fixture selects at + most 16 workers; 650 stale homes remain inactive; retirement wins over a + live lease; planning is stable across restart. +- Deleted private-key and daemon-sync-time eligibility. Registry binding, live + lease, or pending outbox now establishes eligibility. +- MCP processes publish leases. Daemon workers deliberately do not: worker- + generated activity would recreate the self-perpetuating eligibility bug. + MCP skips direct daemon startup when a live all-session supervisor owns + lifecycle. +- Supervisor reports selected/queued/inactive/retired counts, uses bounded + exponential crash backoff, and validates exact JSON pidfile-owned workers + before targeted SIGTERM. Logs distinguish current-version and skewed workers. + No process-family signal is used. + +## Iteration 3 — durable service options + +- RED/GREEN parser tests cover existing launchd plist, systemd unit, and Task + Scheduler XML plus first-install defaults and zero rejection. +- `wire service install --interval N --max-workers N` now configures the daemon + service. Omitted flags preserve installed values; only first install uses + 5 seconds and 16 workers. Every platform renders both flags. +- Local-relay service arguments remain unchanged. No service command ran on the + live machine. + +## Iteration 4 — read-only doctor + +- Added classified verdicts: `wire_defect`, `operator_config`, and + `runtime_health`. +- Added supervisor fan-out/stale-home, fixed launcher override, PATH/service + executable shadow, stale MCP version, and local-relay health checks. +- Production doctor no longer calls the endpoint mutation helper; endpoint + repair is reported, never applied. +- Read-only branch doctor against the live machine reported: 563 live session + workers over cap 16 across 681 homes; 681 homes with no lease; one fixed + Codex `WIRE_SESSION_ID` launcher override while 43 Wire MCP processes run; + PATH/service executable mismatch relative to the worktree build; 39 live MCP + pidfiles with no current skew; local relay absent and connection refused. + It made no service or home changes. + +## Iteration 5 — local relay preflight and bilateral convergence + +- RED: direct local-sister dial returned `drop_sent`; relay-down dial mutated + initiator state before connection refusal; send returned a generic transport + error. +- GREEN: relay health is checked before trust/relay mutation. Failure says no + pairing state changed and points to local-relay service status. +- A local-sister dial now performs a guarded reverse one-way dial, pulls both + pair events/acks synchronously in isolated forced homes, and verifies + `VERIFIED` on both sides before returning `status: verified`. +- Local-scope send failures identify the local relay and state that send state + did not change. +- Focused end-to-end tests pass for bilateral convergence without manual pull, + direct delivery after pairing, unavailable relay with no half-pair state, + three-session local-only mesh, and pair-all-local idempotence. + +## Isolated runtime measurement + +`cargo test --test supervisor_bounded -- --test-threads=1 --nocapture` created +655 initialized but stale homes and started the real all-session supervisor +twice with a cap of four. Restart 0: 0 children, 12,336 KiB RSS. Restart 1: +0 children, 12,608 KiB RSS. The live installation remained unchanged. diff --git a/src/cli/pairing.rs b/src/cli/pairing.rs index fd7d74a..4f2d96d 100644 --- a/src/cli/pairing.rs +++ b/src/cli/pairing.rs @@ -821,6 +821,49 @@ pub(crate) struct LocalSisterDrop { pub event_id: String, pub delivered_via: String, pub delivery_relay_url: String, + pub bilateral_verified: bool, +} + +fn run_wire_for_session(home: &std::path::Path, args: &[&str], one_way: bool) -> Result<()> { + let exe = std::env::current_exe().context("resolving Wire executable for sister handshake")?; + let mut command = std::process::Command::new(exe); + command + .args(args) + .env("WIRE_HOME", home) + .env("WIRE_HOME_FORCE", "1") + .env("WIRE_QUIET_AUTOSESSION", "1") + .env_remove("WIRE_SESSION_ID") + .env_remove("CLAUDE_CODE_SESSION_ID") + .env_remove("CODEX_SESSION_ID") + .env_remove("COPILOT_AGENT_SESSION_ID") + .env_remove("VSCODE_GIT_REPOSITORY_ROOT"); + if one_way { + command.env("WIRE_LOCAL_PAIR_ONE_WAY", "1"); + } else { + command.env_remove("WIRE_LOCAL_PAIR_ONE_WAY"); + } + let output = command + .output() + .with_context(|| format!("running isolated `wire {}`", args.join(" ")))?; + if !output.status.success() { + bail!( + "isolated `wire {}` failed: {}", + args.join(" "), + String::from_utf8_lossy(&output.stderr).trim() + ); + } + Ok(()) +} + +fn home_effective_tier(home: &std::path::Path, peer: &str) -> Option { + let read = |name: &str| { + std::fs::read(home.join("config").join("wire").join(name)) + .ok() + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) + }; + let trust = read("trust.json")?; + let relay = read("relay.json")?; + Some(crate::trust::effective_tier(&trust, &relay, peer)) } /// Core of the local-sister pair: resolve the sister, pin them VERIFIED, @@ -833,7 +876,7 @@ pub(crate) fn add_local_sister_core(sister_name: &str) -> Result s, + Ok(s) => s.clone(), Err(ResolveError::NotFound) => bail!( "no sister session named `{sister_name}` (matched by session name or character nickname). \ Run `wire session list` to see what's available." @@ -907,6 +950,14 @@ pub(crate) fn add_local_sister_core(sister_name: &str) -> Result sister_endpoints[0].clone(), }; + let client = crate::relay_client::RelayClient::new(&delivery_endpoint.relay_url); + client.check_healthz().map_err(|error| { + anyhow!( + "local relay preflight failed for {}: {error:#}; no pairing state changed. Check `wire service status --local-relay` and recover through the service manager", + delivery_endpoint.relay_url + ) + })?; + // 4. Ensure WE have a slot to advertise back. For local-only sessions // this is the local slot; for dual-slot sessions, federation is fine. // `ensure_self_with_relay(None)` defaults to wireup.net which is wrong @@ -973,7 +1024,6 @@ pub(crate) fn add_local_sister_core(sister_name: &str) -> Result Result "local", crate::endpoints::EndpointScope::Lan => "lan", @@ -996,6 +1072,7 @@ pub(crate) fn add_local_sister_core(sister_name: &str) -> Result Result<( "peer_handle": drop.peer_handle, "event_id": drop.event_id, "delivered_via": drop.delivered_via, - "status": "drop_sent", + "status": if drop.bilateral_verified { "verified" } else { "drop_sent" }, }))? ); } else { println!( - "→ found sister `{sister_name}` (did={})\n→ pinned peer locally\n→ pair_drop delivered to {} slot on {}\nawaiting pair_drop_ack from {} to complete bilateral pin.", + "→ found sister `{sister_name}` (did={})\n→ pair_drop delivered to {} slot on {}\n→ bilateral local-sister pair VERIFIED with {}.", drop.paired_with_did, drop.delivered_via, drop.delivery_relay_url, drop.peer_handle ); } diff --git a/src/cli/session.rs b/src/cli/session.rs index 42b4e76..f94a150 100644 --- a/src/cli/session.rs +++ b/src/cli/session.rs @@ -1,4 +1,4 @@ -use anyhow::{Context, Result, anyhow, bail}; +use anyhow::{Context, Result, bail}; use serde_json::{Value, json}; fn resolve_session_name(name: Option<&str>) -> Result { @@ -1191,10 +1191,10 @@ pub(super) fn cmd_session_pair_all_local( Ok(()) } -/// Drive one bilateral pair handshake between two sister sessions -/// using their session home dirs as `WIRE_HOME`. Sequential 8-step -/// flow so failures bubble up at the offending step, not buried in -/// a parallel race. See `cmd_session_pair_all_local` docstring. +/// Drive one bilateral pair handshake between two sister sessions. +/// `wire add --local-sister` now performs relay preflight, reverse dial, +/// both pulls, and VERIFIED convergence synchronously, so orchestration +/// no longer repeats the old manual pull/accept choreography. /// /// v0.6.6: step 1 (the `wire add`) uses `--local-sister` instead of /// federation `.well-known/wire/agent` resolution. Reads B's card + @@ -1208,18 +1208,23 @@ pub(super) fn cmd_session_pair_all_local( fn drive_bilateral_pair( a_home: &std::path::Path, a_name: &str, - b_home: &std::path::Path, + _b_home: &std::path::Path, b_name: &str, _fed_host: &str, _federation_relay: &str, - settle_secs: u64, + _settle_secs: u64, ) -> Result<()> { - use std::time::Duration; let bin = std::env::current_exe().context("locating self exe")?; let run = |home: &std::path::Path, args: &[&str]| -> Result<()> { let out = std::process::Command::new(&bin) .env("WIRE_HOME", home) + .env("WIRE_HOME_FORCE", "1") + .env_remove("WIRE_SESSION_ID") + .env_remove("CLAUDE_CODE_SESSION_ID") + .env_remove("CODEX_SESSION_ID") + .env_remove("COPILOT_AGENT_SESSION_ID") + .env_remove("VSCODE_GIT_REPOSITORY_ROOT") .env_remove("RUST_LOG") .args(args) .output() @@ -1234,50 +1239,8 @@ fn drive_bilateral_pair( Ok(()) }; - // v0.11: each session's agent-card.handle is the DID-derived - // character, not the session name. wire-accept lookups key on the - // CARD HANDLE, so we discover each side's canonical handle from - // its agent-card on disk before driving the pair flow. - let read_card_handle = |home: &std::path::Path| -> Result { - let card_path = home.join("config").join("wire").join("agent-card.json"); - let bytes = std::fs::read(&card_path) - .with_context(|| format!("reading agent-card at {card_path:?}"))?; - let card: Value = serde_json::from_slice(&bytes)?; - card.get("handle") - .and_then(Value::as_str) - .map(str::to_string) - .ok_or_else(|| anyhow!("agent-card at {card_path:?} missing `handle` field")) - }; - let a_handle = read_card_handle(a_home) - .with_context(|| format!("session {a_name} (a): read agent-card.handle"))?; - let b_handle = read_card_handle(b_home) - .with_context(|| format!("session {b_name} (b): read agent-card.handle"))?; - - // 1. A initiates via --local-sister (uses the session NAME for - // the registry lookup; cmd_add_local_sister auto-resolves - // session→handle internally). run(a_home, &["add", b_name, "--local-sister", "--json"]) - .with_context(|| format!("step 1/8: {a_name} `wire add {b_name} --local-sister`"))?; - - // 3. settle so pair_drop reaches B's slot - std::thread::sleep(Duration::from_secs(settle_secs)); - - // 4. B pulls pair_drop → 5. B accept (pins A by CARD HANDLE, - // not by session name — under v0.11 these differ) → 6. B push ack - run(b_home, &["pull", "--json"]).with_context(|| format!("step 4/8: {b_name} `wire pull`"))?; - run(b_home, &["accept", &a_handle, "--json"]).with_context(|| { - format!("step 5/8: {b_name} `wire accept {a_handle}` (a session={a_name})") - })?; - run(b_home, &["push", "--json"]).with_context(|| format!("step 6/8: {b_name} `wire push`"))?; - - // 7. settle so ack reaches A's slot - std::thread::sleep(Duration::from_secs(settle_secs)); - - // 8. A pulls ack (pins B by CARD HANDLE) - run(a_home, &["pull", "--json"]).with_context(|| format!("step 8/8: {a_name} `wire pull`"))?; - // suppress unused warning when both handles are consumed - let _ = &b_handle; - + .with_context(|| format!("{a_name} `wire add {b_name} --local-sister`"))?; Ok(()) } diff --git a/src/mcp.rs b/src/mcp.rs index 5ba8ebf..5c600bc 100644 --- a/src/mcp.rs +++ b/src/mcp.rs @@ -1733,6 +1733,7 @@ fn tool_dial(args: &Value) -> Result { "paired_with": drop.paired_with_did, "event_id": drop.event_id, "delivered_via": drop.delivered_via, + "tier": if drop.bilateral_verified { "VERIFIED" } else { "PENDING_ACK" }, })) } // Unresolvable: surface the resolver's own did-you-mean message diff --git a/src/send.rs b/src/send.rs index f9de0cc..2f8c72a 100644 --- a/src/send.rs +++ b/src/send.rs @@ -207,7 +207,15 @@ pub fn attempt_deliver(peer_handle: &str, signed_event: &Value) -> Result { - let detail = crate::relay_client::format_transport_error(&e); + let transport = crate::relay_client::format_transport_error(&e); + let detail = if ep.scope == crate::endpoints::EndpointScope::Local { + format!( + "local relay unavailable at {}: {transport}. No send state changed; check `wire service status --local-relay`", + ep.relay_url + ) + } else { + transport + }; // Classify 4xx/410 (stale slot) distinctly from transport // errors; reuse the relay's error-text classifier so both // paths agree. Keep as last_failure and try the next endpoint. diff --git a/tests/stress_within_system.rs b/tests/stress_within_system.rs index 1999d7d..eb7dfdb 100644 --- a/tests/stress_within_system.rs +++ b/tests/stress_within_system.rs @@ -546,6 +546,142 @@ async fn local_only_sessions_pair_without_federation_v0_6_6() { assert_eq!(bs["failed"].as_u64().unwrap_or(99), 0); } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn local_sister_dial_converges_bilaterally_without_manual_pull() { + let local_url = spawn_local_only_relay().await; + let root = fresh_dir("bilateral-local-dial"); + for name in ["alpha", "beta"] { + let out = wire( + &root, + &[ + "session", + "new", + name, + "--local-only", + "--local-relay", + &local_url, + "--no-daemon", + "--json", + ], + ); + assert!(out.status.success(), "session new failed: {out:?}"); + } + let alpha = session_home_for(&root, "alpha"); + let beta = session_home_for(&root, "beta"); + let alpha_handle = handle_for_session(&root, "alpha"); + let beta_handle = handle_for_session(&root, "beta"); + + let dial = wire(&alpha, &["add", &beta_handle, "--local-sister", "--json"]); + assert!( + dial.status.success(), + "local sister dial failed: {}", + String::from_utf8_lossy(&dial.stderr) + ); + let dial_json: Value = serde_json::from_slice(&dial.stdout).unwrap(); + assert_eq!(dial_json["status"], "verified"); + + for (home, peer) in [(&alpha, &beta_handle), (&beta, &alpha_handle)] { + let peers = wire(home, &["peers", "--json"]); + let peers: Value = serde_json::from_slice(&peers.stdout).unwrap(); + assert!( + peers + .as_array() + .unwrap() + .iter() + .any(|entry| { entry["handle"] == *peer && entry["tier"] == "VERIFIED" }), + "{home:?} did not converge to VERIFIED for {peer}: {peers}" + ); + } + + let send = wire(&alpha, &["send", &beta_handle, "claim", "bilateral-ready"]); + assert!( + send.status.success(), + "direct send after dial failed: {}", + String::from_utf8_lossy(&send.stderr) + ); + assert!(wire(&beta, &["pull", "--json"]).status.success()); + let tail = wire(&beta, &["tail", &alpha_handle, "--json"]); + assert!(String::from_utf8_lossy(&tail.stdout).contains("bilateral-ready")); + + let alpha_relay_path = alpha.join("config").join("wire").join("relay.json"); + let mut alpha_relay: Value = + serde_json::from_slice(&std::fs::read(&alpha_relay_path).unwrap()).unwrap(); + for endpoint in alpha_relay["peers"][&beta_handle]["endpoints"] + .as_array_mut() + .unwrap() + { + endpoint["relay_url"] = Value::String("http://127.0.0.1:9".into()); + } + for endpoint in alpha_relay["self"]["endpoints"].as_array_mut().unwrap() { + endpoint["relay_url"] = Value::String("http://127.0.0.1:9".into()); + } + std::fs::write( + &alpha_relay_path, + serde_json::to_vec_pretty(&alpha_relay).unwrap(), + ) + .unwrap(); + let unavailable = wire(&alpha, &["send", &beta_handle, "claim", "relay-down"]); + assert!(!unavailable.status.success()); + assert!( + String::from_utf8_lossy(&unavailable.stdout).contains("local relay unavailable"), + "{}", + String::from_utf8_lossy(&unavailable.stdout) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn unavailable_local_relay_does_not_leave_half_pair_state() { + let local_url = spawn_local_only_relay().await; + let root = fresh_dir("local-relay-preflight"); + for name in ["alpha", "beta"] { + assert!( + wire( + &root, + &[ + "session", + "new", + name, + "--local-only", + "--local-relay", + &local_url, + "--no-daemon", + "--json", + ], + ) + .status + .success() + ); + } + let alpha = session_home_for(&root, "alpha"); + let beta_handle = handle_for_session(&root, "beta"); + let beta_relay = session_home_for(&root, "beta") + .join("config") + .join("wire") + .join("relay.json"); + let mut relay: Value = serde_json::from_slice(&std::fs::read(&beta_relay).unwrap()).unwrap(); + relay["self"]["relay_url"] = Value::String("http://127.0.0.1:9".into()); + for endpoint in relay["self"]["endpoints"].as_array_mut().unwrap() { + endpoint["relay_url"] = Value::String("http://127.0.0.1:9".into()); + } + std::fs::write(&beta_relay, serde_json::to_vec_pretty(&relay).unwrap()).unwrap(); + + let dial = wire(&alpha, &["add", &beta_handle, "--local-sister", "--json"]); + assert!(!dial.status.success()); + let stderr = String::from_utf8_lossy(&dial.stderr); + assert!(stderr.contains("local relay preflight failed"), "{stderr}"); + assert!(stderr.contains("no pairing state changed"), "{stderr}"); + let alpha_relay: Value = serde_json::from_slice( + &std::fs::read(alpha.join("config").join("wire").join("relay.json")).unwrap(), + ) + .unwrap(); + assert!( + alpha_relay + .get("peers") + .and_then(Value::as_object) + .is_none_or(serde_json::Map::is_empty) + ); +} + // ---------- TEST 8: mesh route picks one sister by role + strategy (v0.6.5 / #21) ---------- /// v0.6.5 (issue #21): `wire mesh route ` filters sister sessions by From c5d8208f73dbe1bcc26e99ce554dbb1f265965c0 Mon Sep 17 00:00:00 2001 From: Paul Logan Date: Sat, 18 Jul 2026 00:44:16 -0700 Subject: [PATCH 5/6] test: harden reproducible CI gate --- SESSION_LOG_2026_07_17.md | 48 +++++++++++++++++++++++++++++++++++++++ test-env/Dockerfile | 5 ++++ test-env/README.md | 8 +++++++ 3 files changed, 61 insertions(+) diff --git a/SESSION_LOG_2026_07_17.md b/SESSION_LOG_2026_07_17.md index cff9f0a..2ebda3f 100644 --- a/SESSION_LOG_2026_07_17.md +++ b/SESSION_LOG_2026_07_17.md @@ -153,3 +153,51 @@ literal Codex override. 655 initialized but stale homes and started the real all-session supervisor twice with a cap of four. Restart 0: 0 children, 12,336 KiB RSS. Restart 1: 0 children, 12,608 KiB RSS. The live installation remained unchanged. + +## Canonical gate repair + +- The first complete `test-env/run.sh` attempt passed formatting, clippy, and + every Rust test, then Cargo 1.88 lost a release bytecode path in the shared + Docker target volume. A release-only build succeeded on a fresh task-scoped + volume, while fresh full-gate builds reproduced missing incremental and + fingerprint paths on both volume names. +- `CARGO_INCREMENTAL=0` made the same failed clippy cache pass immediately. + The test image now disables incremental compilation, matching CI's + disposable job workspaces and avoiding Docker named-volume rename races. +- The full gate then exposed a pre-existing fixture gap in + `100-fleet-link.sh`: the minimal Debian image had neither Linux machine-id + path, so Wire correctly failed closed. The image now supplies a deterministic + non-secret machine-id fixture. Focused fleet-link then passed all assertions. +- One generated, unmounted, corrupt `wire-testenv-target` cache volume was + removed and recreated. No source, service state, or identity state lived in + that volume. + +## Final verification + +- `test-env/run.sh`: exit 0. Formatting and clippy passed; Rust unit and + integration targets passed serially; the 655-home restart test passed; + release build and both demos passed; shell integration suite reported + 11 passed, 0 failed. +- Focused lifecycle, supervisor, relay-down, bilateral local-sister, mesh, and + fleet-link tests passed before the full gate. +- Test subprocess audit confirmed temporary `WIRE_HOME` helpers set + `WIRE_HOME_FORCE=1`; `tests/it/lib.sh` exports it for shell fixtures. + +## Final live read-only snapshot + +Captured after verification, before push: 575 `wire daemon` processes, +4,596,528 KiB aggregate RSS, 52 `wire mcp` processes, 681 by-key homes, and no +new-format live lease files because the deployed binary predates this branch. +The daemon launchd unit remains loaded. Local relay `127.0.0.1:8771` still +refuses connections. Counts grew during the session under the unchanged legacy +service, confirming that no live remediation or restart occurred. + +## Caller-side handoff + +Do not change this repo to compensate for the launcher collision. In +`/Users/laul_pogan/.codex/config.toml`, caller Codex configuration currently +sets one literal `WIRE_SESSION_ID` under `[shell_environment_policy.set]`. +Remove that fixed override and have the Codex launcher/session adapter pass a +stable unique `CODEX_SESSION_ID` (or unique `WIRE_SESSION_ID`) per concurrent +session. Wire doctor now classifies this as `operator_config` and warns when +concurrent Wire MCP processes share the fixed launcher identity. diff --git a/test-env/Dockerfile b/test-env/Dockerfile index 66d3733..80a5801 100644 --- a/test-env/Dockerfile +++ b/test-env/Dockerfile @@ -21,10 +21,15 @@ FROM rust:1.88-bookworm RUN rustup component add clippy rustfmt \ && apt-get update \ && apt-get install -y --no-install-recommends jq \ + && printf '776972652d74657374656e762d303031\n' > /etc/machine-id \ && rm -rf /var/lib/apt/lists/* WORKDIR /wire ENV CARGO_TERM_COLOR=always +# CI jobs use disposable workspaces. Disable incremental compilation here too: +# Cargo 1.88 can lose incremental/fingerprint paths on Docker named volumes +# when `clippy --all-targets` compiles the lib and lib-test targets together. +ENV CARGO_INCREMENTAL=0 # Default command = the full CI gate. Overridden by any args passed to # `docker run` (see run.sh). diff --git a/test-env/README.md b/test-env/README.md index 1264214..e3ba673 100644 --- a/test-env/README.md +++ b/test-env/README.md @@ -17,6 +17,14 @@ First run builds the image (`wire-testenv`) and warms the build cache; later runs reuse the cached `target/` and cargo registry (named Docker volumes), so they're fast and never touch your host `target/`. +Incremental compilation is disabled inside the container. CI jobs use +disposable workspaces, and Cargo 1.88 can race incremental-directory renames +when `clippy --all-targets` writes to a Docker named volume. + +The image also carries a deterministic, non-secret machine-id fixture. Minimal +Debian images omit both Linux machine-id paths, but the fleet-link integration +test needs a stable same-machine identity to exercise its fail-closed path. + ## What it mirrors The default command is exactly the CI gate from `.github/workflows/ci.yml`: From b8f74ba796a619a314746e4cf49e58ffe1880890 Mon Sep 17 00:00:00 2001 From: Paul Logan Date: Sat, 18 Jul 2026 01:04:46 -0700 Subject: [PATCH 6/6] fix: keep MCP session lease alive --- SESSION_LOG_2026_07_17.md | 16 ++++++++++++++++ src/mcp.rs | 33 +++++++++++++++++++-------------- 2 files changed, 35 insertions(+), 14 deletions(-) diff --git a/SESSION_LOG_2026_07_17.md b/SESSION_LOG_2026_07_17.md index 2ebda3f..2c14c2e 100644 --- a/SESSION_LOG_2026_07_17.md +++ b/SESSION_LOG_2026_07_17.md @@ -201,3 +201,19 @@ Remove that fixed override and have the Codex launcher/session adapter pass a stable unique `CODEX_SESSION_ID` (or unique `WIRE_SESSION_ID`) per concurrent session. Wire doctor now classifies this as `operator_config` and warns when concurrent Wire MCP processes share the fixed launcher identity. + +## Final review + +- Build-loop cross-provider review was attempted twice through the read-only + Claude subscription reviewer: once with the complete 137 KiB diff (300s), + then with the 84 KiB executable-only diff (420s). Both timed out without a + verdict or findings. No reviewer approval is claimed; no third retry was + made. +- Manual review covered every critical diff named by GitNexus. It found one + lifecycle edge: MCP heartbeat shared the inbox watcher thread's early-return + path, so a watcher initialization failure could let a live MCP lease expire. + The watcher is now best-effort and retries while the same thread continues + the authoritative heartbeat. Focused MCP tests passed after the correction. +- GitNexus impact for `mcp::run` was LOW with no upstream dependants. Final + staged/compare change detection and the full canonical gate were rerun after + the correction before commit. diff --git a/src/mcp.rs b/src/mcp.rs index 5c600bc..78c848c 100644 --- a/src/mcp.rs +++ b/src/mcp.rs @@ -181,10 +181,10 @@ pub fn run() -> Result<()> { let shutdown_w = shutdown.clone(); let lease_w = session_lease.clone(); let watcher_handle = std::thread::spawn(move || { - let mut watcher = match crate::inbox_watch::InboxWatcher::from_head() { - Ok(w) => w, - Err(_) => return, - }; + // Inbox notification setup is best-effort. Lease heartbeat is not: + // a transient watcher failure must not make a live MCP identity age + // out of supervisor eligibility. Retry watcher creation on poll. + let mut watcher = crate::inbox_watch::InboxWatcher::from_head().ok(); let poll_interval = Duration::from_secs(2); let mut next_poll = Instant::now() + poll_interval; let mut next_lease_heartbeat = @@ -213,16 +213,21 @@ pub fn run() -> Result<()> { let mut affected: HashSet = HashSet::new(); // ---- inbox events ---- - if !subs_snapshot.is_empty() - && let Ok(events) = watcher.poll() - { - for ev in &events { - if subs_snapshot.contains("wire://inbox/all") { - affected.insert("wire://inbox/all".to_string()); - } - let peer_uri = format!("wire://inbox/{}", ev.peer); - if subs_snapshot.contains(&peer_uri) { - affected.insert(peer_uri); + if !subs_snapshot.is_empty() { + if watcher.is_none() { + watcher = crate::inbox_watch::InboxWatcher::from_head().ok(); + } + if let Some(watcher) = watcher.as_mut() + && let Ok(events) = watcher.poll() + { + for ev in &events { + if subs_snapshot.contains("wire://inbox/all") { + affected.insert("wire://inbox/all".to_string()); + } + let peer_uri = format!("wire://inbox/{}", ev.peer); + if subs_snapshot.contains(&peer_uri) { + affected.insert(peer_uri); + } } } }