diff --git a/.changeset/brown-loops-race.md b/.changeset/brown-loops-race.md new file mode 100644 index 0000000..294c473 --- /dev/null +++ b/.changeset/brown-loops-race.md @@ -0,0 +1,11 @@ +--- +"warpforge": patch +--- + +A workflow no longer ends for good when one of its agents is lost. If an agent +process dies part-way through a stage — killed by something outside the run, +not by anything wrong with the work — the pipeline now pauses at that stage +instead of finishing as failed. Press Resume and it runs the stage again, +warned that the working copy may already hold partial changes. Previously the +run was over: resume was refused and the only way forward was a new task, even +when the work was already done. diff --git a/.changeset/eighty-pugs-smile.md b/.changeset/eighty-pugs-smile.md new file mode 100644 index 0000000..0252a9b --- /dev/null +++ b/.changeset/eighty-pugs-smile.md @@ -0,0 +1,10 @@ +--- +"warpforge": patch +--- + +Starting a task no longer pauses while its name is written. Naming a task runs +a short agent in the background, and the app used to wait on it before handling +anything else — so the first message, tool approvals, and other tasks all sat +still until the name came back. Naming now happens alongside your work, as do +installing an agent or a language server, which had the same problem and could +hold things up for much longer. diff --git a/.changeset/great-moons-tap.md b/.changeset/great-moons-tap.md new file mode 100644 index 0000000..6b0f554 --- /dev/null +++ b/.changeset/great-moons-tap.md @@ -0,0 +1,10 @@ +--- +"warpforge": patch +--- + +The app now handles several requests at once instead of one at a time. A single +slow action — listing a large project, loading a diff, scanning for agents — +used to hold up everything else you did, so a tool approval could sit waiting +until the slow one finished. Requests that only read now run alongside each +other, and replies are sent without waiting on the network's send delay, which +takes tens of milliseconds off routine actions. diff --git a/.changeset/lazy-jars-cheer.md b/.changeset/lazy-jars-cheer.md new file mode 100644 index 0000000..02ee533 --- /dev/null +++ b/.changeset/lazy-jars-cheer.md @@ -0,0 +1,9 @@ +--- +"warpforge": patch +--- + +Starting a task in its own workspace copy no longer holds up everything else. +Setting that copy up takes a moment, and until now the whole app waited on it — +your other tasks' replies and approvals paused until the new task's workspace +was ready. The task now shows up on the board immediately and begins work as +soon as its workspace lands, while the rest of the app keeps moving. diff --git a/.changeset/olive-books-melt.md b/.changeset/olive-books-melt.md new file mode 100644 index 0000000..449855f --- /dev/null +++ b/.changeset/olive-books-melt.md @@ -0,0 +1,9 @@ +--- +"warpforge": patch +--- + +Branching a conversation now carries your uncommitted work across, including +when the original task runs in the project folder itself rather than its own +workspace copy. The branch used to start from the last commit in that case, so +edits you had not committed were missing from the conversation meant to +continue them. diff --git a/.changeset/olive-moons-race.md b/.changeset/olive-moons-race.md new file mode 100644 index 0000000..b74e239 --- /dev/null +++ b/.changeset/olive-moons-race.md @@ -0,0 +1,8 @@ +--- +"warpforge": patch +--- + +Searching for files no longer freezes the rest of the app. On a large project +the search reads through every file, and until now everything else — agent +replies, approvals, service controls — stopped until it finished. Search now +runs out of the way, so the app keeps responding while it works. diff --git a/.changeset/plain-days-tell.md b/.changeset/plain-days-tell.md new file mode 100644 index 0000000..072a5eb --- /dev/null +++ b/.changeset/plain-days-tell.md @@ -0,0 +1,6 @@ +--- +"warpforge": patch +--- + +Merging a task's workspace copy back into your project no longer pauses the +rest of the app while git works. diff --git a/.changeset/quick-pans-shake.md b/.changeset/quick-pans-shake.md new file mode 100644 index 0000000..75ab8e8 --- /dev/null +++ b/.changeset/quick-pans-shake.md @@ -0,0 +1,10 @@ +--- +"warpforge": patch +--- + +Approving a tool call, sending a message, or starting a task no longer waits on +whatever else is happening. Previously, while an agent was streaming its answer, +the app saved every fragment as it arrived and everything else queued up behind +that — so an approval prompt could sit unresponsive for as long as the agent +kept typing, even in a different task. Saving now happens out of the way, and +the interface stays responsive while agents work. diff --git a/.changeset/small-horses-jump.md b/.changeset/small-horses-jump.md new file mode 100644 index 0000000..078a39b --- /dev/null +++ b/.changeset/small-horses-jump.md @@ -0,0 +1,11 @@ +--- +"warpforge": patch +--- + +Long conversations no longer grow memory without limit. The app used to keep +every line of everything your agents had said in memory and reload it all on +start, so the more work agents did, the more memory the app held onto even when +it was only showing the latest exchange. It now keeps just what the current +view needs — the latest message and the most recent exchange — and loads the +rest only when you resume a session or open a project. Resuming a session +still shows each reply once, and nothing in the chat history is lost. diff --git a/.changeset/tall-melons-wave.md b/.changeset/tall-melons-wave.md new file mode 100644 index 0000000..ef6b6eb --- /dev/null +++ b/.changeset/tall-melons-wave.md @@ -0,0 +1,9 @@ +--- +"warpforge": patch +--- + +Warpforge no longer stops processes it did not start. When shutting down it used +to clear everything listening on the project's port range, which could take down +a server you were running yourself — or, when running warpforge's own tests, the +agents of the warpforge you were running them from. It now only stops the +services it started. diff --git a/.changeset/tidy-crabs-argue.md b/.changeset/tidy-crabs-argue.md new file mode 100644 index 0000000..fc25c76 --- /dev/null +++ b/.changeset/tidy-crabs-argue.md @@ -0,0 +1,10 @@ +--- +"warpforge": patch +--- + +Viewing changes no longer slows the rest of the app down. The changes panel +refreshes on a timer, and each refresh used to hold everything else up while it +inspected the repository — with a task open, that was a steady drip of pauses +affecting agent replies and approvals. Reading diffs, file contents, file lists +and branches now happens alongside the rest of the app instead of in front of +it. diff --git a/.changeset/wise-cups-look.md b/.changeset/wise-cups-look.md new file mode 100644 index 0000000..26ebd33 --- /dev/null +++ b/.changeset/wise-cups-look.md @@ -0,0 +1,9 @@ +--- +"warpforge": patch +--- + +Committing, pushing, merging, switching branches, saving a file and opening a +pull request no longer pause the rest of the app while they run. Each of these +waits on git, and until now everything else — agent replies, approvals, your +other tasks — waited with it. They now run alongside your work, so a slow push +costs you the push and nothing else. diff --git a/docs/adr/0002-daemon-concurrency.md b/docs/adr/0002-daemon-concurrency.md new file mode 100644 index 0000000..6813ed7 --- /dev/null +++ b/docs/adr/0002-daemon-concurrency.md @@ -0,0 +1,144 @@ +# 0002 — Daemon concurrency: non-blocking mailboxes, sharded per task + +**Status:** accepted (2026-08-15) + +## Context + +The daemon started as one actor: a single `tokio::select!` loop in +`daemon/actor.rs`, owning all state and draining one `mpsc` mailbox. For the +scale it was written at — a couple of projects, one agent session — that was the +right call, and it made every state transition trivially race-free. + +It stopped holding. `actor.rs` is past 7000 lines, `handle_command` awaits each +command inline, and the work behind those commands grew: worktree creation, +subprocess `git`, filesystem walks, SQLite writes. What users hit: starting a +task stalls the conversation, approving a tool call waits behind unrelated work +in a different task, the title generator delays the first turn. One queue for +everything. + +Every source of head-of-line blocking below was confirmed in the code, not +inferred: + +- **Blocking `rusqlite` on the actor's thread.** `emit_session` persists on + *every* streamed chunk from an agent, and `emit_session_unless_last_duplicate` + reads the last row back from disk before each one. While an agent streams, the + actor hammers the disk and nothing else in the daemon advances. +- **A second queue above the actor.** The WebSocket read loop in + `daemon/server.rs` awaits `dispatch` inline, so one connection serves one + request at a time. A tool approval is not even read off the socket until the + file search ahead of it returns. +- **Inline I/O in `handle_command`** — worktree create, diff, branch operations. + `diff::search_files` is a fully synchronous tree walk that reads every file. +- **A latent self-deadlock.** The actor sends to its own bounded mailbox + (`Command::ProbeAgent`). If that mailbox ever fills, the actor blocks on its + own send with nobody left to drain it. + +ADR 0001 invariant 9 — "waits on a child's exit are bounded … the daemon actor +is single-threaded and awaits handlers inline" — is a workaround for this +architecture, not a property worth keeping. + +## Decisions + +**The actor loop never blocks and never awaits I/O.** A handler may read and +mutate in-memory state, then it either replies or hands the work to a task. +Results come back as ordinary messages. *Rejected:* case-by-case `tokio::spawn` +where a handler looks slow — that is what produced today's state, where four +handlers spawn and thirty do not, and no reader can tell which rule applies. + +**Persistence is a write-behind actor on its own blocking thread.** It owns the +`rusqlite` connection, coalesces streamed session updates in memory, and flushes +batched transactions. Callers get fire-and-forget. *Rejected:* `spawn_blocking` +per write — it keeps one disk round-trip per streamed chunk, which is the actual +cost; the fix is batching, not moving the same work sideways. *Rejected:* an +async SQLite wrapper — same round-trip count, plus a dependency. + +**Reads do not enter a mailbox.** Diffs, file listings, search, file contents and +snapshots are served from an `ArcSwap` state snapshot plus I/O on a worker. They +need a consistent *view*, not exclusive access. This removes roughly half the +`Command` variants from the write path. *Rejected:* keeping reads in the mailbox +for strict read-your-writes ordering — the UI already tolerates eventual +refresh, and it is what makes polled reads (the diff panel) cost the whole +daemon. + +**Requests are concurrent per connection.** `server.rs` spawns each dispatch, +bounded by a semaphore, rather than awaiting it in the read loop. + +**State is split by ownership, then sharded per task.** Global state (projects, +accounts, configured agents, services, port forwards) stays in one actor. Per-task +state (agent session, pending permissions, workflow run, worktree) moves to a +task actor with its own mailbox, supervised so a wedged task cannot take the +daemon with it. The global actor routes. *Rejected:* one actor with finer-grained +locks — it trades a queue for a lock graph and loses the property that makes the +actor model worth having. + +**Control-plane messages never share a queue with data-plane.** Permission +answers, cancels and stops ride a separate channel, drained first in a `biased` +select. A user answering a permission prompt must not wait behind a stream of +agent output. + +**Every message that mutates task state carries the task's epoch.** Handing work +to a task means results arrive after the world may have moved on; a result whose +epoch does not match is dropped. *Rejected:* checking only that the task still +exists — an id is reused across cancel-and-restart, and the stale write lands on +the new run. + +**New machinery grows beside the old, and ownership moves in one step per +piece.** The new runtime lives under `daemon/runtime/`; the existing actor +delegates into it as each piece lands. *Rejected:* a parallel implementation kept +running alongside the old one behind a flag — two owners of the same mutable +state diverge, and the resulting bug reports are unreadable. Alongside means +*not yet wired*, never *wired twice*. + +## Invariants + +Named by module, because each one fails quietly. + +1. **`daemon/actor.rs` handlers hold no `.await` on I/O.** git, filesystem, + subprocess and store calls are handed off. A handler that awaits I/O + reintroduces the whole class of bug this record exists for, and it will look + local and harmless in review. +2. **Nothing calls `store::*` from an actor loop.** The store is reachable only + through the persistence actor's channel. A direct call compiles, runs, and + silently puts a blocking disk write back on the hot path. +3. **An actor never `.await`s a send to its own mailbox.** Use `try_send` and + handle the full case, or a dedicated unbounded self-channel. This is a hard + deadlock, not a slowdown. +4. **Every reply path stays total.** Handing work to a task adds paths where a + `oneshot` sender can be dropped — a task that panics, an epoch mismatch, a + shard that was torn down. A dropped reply is a client promise that never + settles: a spinner that spins forever. Every early return sends something. +5. **Epoch is checked before mutation, not before dispatch.** The gap between + accepting a result and applying it is where the stale write lands. +6. **Snapshot publication is atomic per command.** Readers must never observe a + half-applied transition — publish once, after the handler completes, not on + each field it touches. +7. **Ordering guarantees are per task, not global.** Two commands for the same + task keep their order; commands for different tasks do not, and nothing may + assume they do. Workflow stage transitions are the place this will be + assumed by accident. +8. **A task shard's death is contained and observable.** Supervision restarts it + from persisted state and marks the task; a silent restart that loses queued + commands is worse than the freeze it replaced. + +## Consequences + +- ADR 0001 invariant 9 (bounded waits on child exit) loses its original + justification once handlers stop blocking the daemon. Bounded waits stay — + they are good hygiene — but they are no longer load-bearing for liveness. +- Read-your-writes is no longer automatic. A mutation followed immediately by a + read may observe the prior snapshot; flows that depend on it must await the + mutation's reply, not re-read. +- `actor.rs` stops being one file. Splitting by ownership is what makes the + 400–500 line rule in `CLAUDE.md` reachable here; the file size is a symptom of + the undivided state, not a formatting problem. +- Debugging changes shape: a stall is no longer "the actor is busy" but "which + shard, which channel". Shard identity belongs in log lines from the start. +- More moving parts. This is only worth it because the single queue is now the + user-visible bottleneck; it would have been premature a year ago. + +## Out of scope, deliberately + +Event sourcing (replay, audit, undo) — a different investment with a different +payoff, and not a latency fix; the write-behind persistence actor here is a +prerequisite for it, not a competitor. Multi-process daemons. Distributing work +across machines. Replacing the broadcast event bus, which is not a bottleneck. diff --git a/docs/adr/0003-workflow-agent-loss.md b/docs/adr/0003-workflow-agent-loss.md new file mode 100644 index 0000000..bca39fa --- /dev/null +++ b/docs/adr/0003-workflow-agent-loss.md @@ -0,0 +1,77 @@ +# 0003 — Losing a stage's agent pauses a pipeline, it does not fail it + +**Status:** accepted (2026-08-15) + +## Context + +A workflow stage whose child session ended called `workflow_finalize` with an +error. `RunState::Failed` is terminal: `is_active()` is false, and +`workflow_resume` answers "the pipeline is not paused". The run was over. + +That is the wrong response to the failure that actually happens. The reported +case: an agent process was killed mid-stage by something unrelated to the work +(warpforge's own test suite, which kills every listener in the project's port +range — see the *Consequences* below). The user re-prompted the stage's task, +the session reconnected, and the agent finished the implementation. The work was +done and reviewable. The pipeline was still dead, and the only way forward was +to start a new task. + +Losing the agent process is an infrastructure failure. The stage never got to +say whether the work was good, so the pipeline has no verdict — but it also has +no reason to conclude the run failed. + +## Decisions + +**A stage that loses its agent parks the run at the pause barrier** +(`RunState::Paused { next: }`) instead of finalizing it. Resume +re-runs the stage. This applies to a stage child whose session ended, and to a +review round where every reviewer's agent died before producing a verdict — +both are the absence of a verdict, not a rejection. + +*Rejected:* a new `WorkflowWaitKind` with retry/accept/stop controls. It is the +better long-term shape — in the reported case the work was already finished, so +"accept this stage and move to review" would have been the right answer — but it +needs a protocol addition and desktop work, and the pipeline being unrecoverable +at all is the part that hurts. The pause barrier already exists, the desktop +already renders it, and resuming already works. + +*Rejected:* automatically retrying the stage. A dead agent is often dead for a +reason that will repeat, and a pipeline that silently re-runs stages burns +tokens without telling anyone. + +**The re-run is told the working copy may already contain partial work**, via +`pending_guidance` — the same warning `restore_workflow_runs` gives a stage +interrupted by a daemon restart. A stage that assumes a clean tree will redo +work that is already there, or worse, conflict with it. + +## Invariants + +1. **This path is for a lost agent, not a bad outcome.** A stage that finishes + and produces a poor verdict goes through `workflow_stage_finished`. Only + `workflow_child_failed` — reached when a child's session ends — parks. Widen + it and a genuinely failing pipeline becomes an infinite pause loop. +2. **Parking must clear the stage's bookkeeping.** `active_children`, and for a + review round `review_pending` / `review_collected` / `reasked`. A resumed run + that still lists the dead child waits for a turn that will never end. +3. **The review round counter is given back when a review round parks**, for + the same reason `restore_workflow_runs` gives it back: spawning re-increments + it, and without the decrement a re-run reports "round 3/2" and lands straight + on the limit decision. +4. **ADR 0001 invariant 6 still holds and is narrower than it looks.** A stage + whose session *fails to start* must fail the run — no handle is inserted, so + no `TurnEnded` will ever arrive and nothing else would notice. A stage whose + session started and *then* died is this record's case. + +## Consequences + +- A pipeline can now sit paused indefinitely after an agent dies. That is + visible on the board (the parent goes to `Waiting` with pause controls) and is + the intended trade against silently burning tokens on retries. +- Resuming re-runs the whole stage. When the lost agent had already finished the + work — the reported case — the re-run is redundant. The rejected barrier + design is what fixes that; this record does not. +- Worth fixing separately: `cargo test` in this repo kills every process + listening on the project's port range (`kill_listeners_in_ranges`, reached + from the daemon teardown that tests exercise). That is what killed the agent + in the reported case, and it kills unrelated processes on the developer's + machine too. diff --git a/docs/adr/README.md b/docs/adr/README.md index c4387e6..045d40d 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -16,3 +16,5 @@ stale and then misleads. | ADR | Subject | | --- | --- | | [0001](0001-workflow-pipelines.md) | Workflow pipelines: deterministic engine, project-configured | +| [0002](0002-daemon-concurrency.md) | Daemon concurrency: non-blocking mailboxes, sharded per task | +| [0003](0003-workflow-agent-loss.md) | Losing a stage's agent pauses a pipeline, it does not fail it | diff --git a/src/daemon/actor.rs b/src/daemon/actor.rs index 4991ac7..42fe8e6 100644 --- a/src/daemon/actor.rs +++ b/src/daemon/actor.rs @@ -32,6 +32,7 @@ use crate::registry::ProjectEntry; use crate::service::{kill_listeners_in_ranges, ServiceEvent, ServiceManager, ServiceStatus}; use super::acp::{spawn_acp_session, AcpHandle, AcpUpdate, PolicyCheck}; +use super::runtime::{Ask as PersistAsk, Write as PersistWrite}; use super::store::Store; use super::task::{Task, TaskStatus}; use super::wire as wireconv; @@ -166,6 +167,64 @@ fn is_acp_replay_update(update: &wire::SessionUpdate) -> bool { } } +/// The replayable subset of a task's persisted history, in order — the seed +/// for the resume replay guard. Built from the store on demand, because the +/// actor holds no full transcript in memory. +fn replayable_history(updates: &[wire::SessionUpdate]) -> VecDeque { + updates + .iter() + .filter(|update| is_acp_replay_update(update)) + .cloned() + .collect() +} + +/// Fold a turn's updates into the two shapes the pipeline needs: the agent's +/// closing message and the whole turn. A new user message starts a fresh turn. +/// The input is the current-turn buffer, which is bounded by a turn, not by the +/// session's length. +fn stage_text_from_updates(updates: &[wire::SessionUpdate]) -> StageText { + let mut full: Vec = Vec::new(); + let mut closing: Vec = Vec::new(); + for update in updates { + match update { + // A new user message starts a fresh turn. + wire::SessionUpdate::UserMessage { .. } => { + full.clear(); + closing.clear(); + } + wire::SessionUpdate::AgentText { text } => { + full.push(text.clone()); + closing.push(text.clone()); + } + // Any work the agent does ends whatever it was narrating, so the + // closing message restarts after it. + wire::SessionUpdate::ToolCall { .. } + | wire::SessionUpdate::FileEdit { .. } + | wire::SessionUpdate::AgentThought { .. } + | wire::SessionUpdate::Plan { .. } => closing.clear(), + _ => {} + } + } + StageText { + closing: closing.join(""), + full: full.join(""), + } +} + +/// Concatenate every `AgentText` update in a task's history — a task's full +/// text output, used as the orchestrator node's result. Computed off the actor +/// loop from the store (the actor holds no full transcript). +fn agent_text_from_updates(updates: &[wire::SessionUpdate]) -> String { + updates + .iter() + .filter_map(|update| match update { + wire::SessionUpdate::AgentText { text } => Some(text.as_str()), + _ => None, + }) + .collect::>() + .join("") +} + /// System preamble prepended to an orchestrator-chat session's first prompt. const ORCHESTRATOR_SYSTEM: &str = "\ You are an orchestrator agent in warpforge. You coordinate work by delegating to \ @@ -583,6 +642,45 @@ pub enum Command { KillAgent { id: String, }, + /// A task's worktree checkout finished (or failed); start its session. + WorktreeReady { + task_id: String, + created: Result<(String, super::worktree::Worktree), String>, + }, + /// Test-only: report whether a finished turn's output has a consumer. + #[cfg(test)] + TurnOutputConsumerProbe { + task_id: String, + workflow_child: bool, + reply: oneshot::Sender, + }, + /// A worktree merge finished and its checkout is gone; drop it from the + /// manager and clear the task's worktree. + WorktreeMerged { + task_id: String, + project: String, + }, + /// A git operation that ran off the loop finished; apply what it changed + /// to the task's state. + GitOpFinished { + task_id: String, + effect: GitEffect, + }, + /// A task's resume replay guard was read from the store; start its session + /// now that replayed history can be de-duplicated. Loaded off the loop + /// (write-behind flush + store read), mirroring [`Command::WorktreeReady`]. + ResumeReplayReady { + task_id: String, + replay: std::collections::VecDeque, + }, + /// A finished turn's full text output was assembled from the store; deliver + /// it to the orchestrator / parent inbox off the actor loop. + TaskOutputReady { + task_id: String, + success: bool, + workflow_child: bool, + output: String, + }, CreateTask { project: String, prompt: String, @@ -1848,6 +1946,113 @@ impl DaemonHandle { } } +/// What a finished git operation changed about a task. +/// +/// These describe something that already happened on disk, so applying one late +/// is still correct — a commit that landed is a commit that landed. The only +/// guard needed is that the task still exists, which the handlers check. +#[derive(Debug, Clone, Copy)] +pub enum GitEffect { + /// HEAD or the working tree moved: nudge clients to refetch. + Bump, + /// A commit landed, so the task has no uncommitted changes left. + Committed, + /// A hunk was rejected, so one fewer file differs. + HunkRejected, +} + +/// A session that cannot start until its worktree exists. +struct PendingSessionStart { + project: String, + agent: String, + prompt: String, + include_runtime_context: bool, + attachments: Vec, + default_model: Option, + config_overrides: std::collections::HashMap, +} + +/// A session that cannot start until its resume replay guard has been loaded +/// from the store. Carries everything `start_session` needs to resume. +struct PendingResume { + project: String, + agent: String, + text: String, + session_id: String, + attachments: Vec, +} + +/// De-duplicates the ACP updates an agent replays on `session/load` against the +/// daemon's persisted history. While the replay matches history in order it is +/// dropped; the first mismatch is new live output and disables the guard. +struct ResumeReplayGuard { + history: VecDeque, +} + +impl ResumeReplayGuard { + /// The replayable subset of `updates`, in order. `None` when there is + /// nothing to de-duplicate (a session with no persisted history). + fn from_updates(updates: &[wire::SessionUpdate]) -> Option { + let history = replayable_history(updates); + (!history.is_empty()).then_some(Self { history }) + } + + fn is_empty(&self) -> bool { + self.history.is_empty() + } + + /// True when `update` is the next replayed update and should be dropped. + /// False when it is live output (and the caller must disable the guard). + fn consume(&mut self, update: &wire::SessionUpdate) -> bool { + if self.history.front() == Some(update) { + self.history.pop_front(); + true + } else { + false + } + } +} + +/// A worktree checkout resolved against actor state, ready to run elsewhere. +struct WorktreeRequest { + project: String, + base_repo: PathBuf, + task_id: String, + /// What a conversation branch inherits from. + source: Option, +} + +/// Where a branched conversation picks up from. +struct BranchSource { + /// The source task's own branch, when it has a worktree. `None` means the + /// source works in the project checkout, so the branch starts from HEAD. + base_branch: Option, + /// The working tree whose uncommitted changes carry over. + path: PathBuf, +} + +impl WorktreeRequest { + async fn run(self) -> Result<(String, super::worktree::Worktree), String> { + let created = match self.source { + Some(ref source) => { + super::worktree::create_branched_detached( + &self.base_repo, + &self.task_id, + source.base_branch.as_deref(), + &source.path, + ) + .await + } + None => super::worktree::create_detached(&self.base_repo, &self.task_id, None).await, + }; + created + .map(|wt| (self.project, wt)) + // `{:#}` keeps the context chain: the top line alone says only + // "failed to copy working state", never which git step failed. + .map_err(|e| format!("{e:#}")) + } +} + pub struct Daemon { projects: Vec, config_observer: ConfigObserver, @@ -1870,13 +2075,24 @@ pub struct Daemon { /// (e.g. the ACP probe) can deliver results without needing a borrow of the /// actor. Held alongside `store` etc. as a primary mutator handle. cmd_tx: mpsc::Sender, - store: Option, + /// Queued writes, applied off the actor thread. Every mutation goes here — + /// calling `store` directly from a handler puts a blocking disk write back + /// on the hot path (ADR 0002). + persist: super::runtime::Persist, + /// Shared with the persistence thread, for the reads that still run on the + /// actor. Those move to an in-memory projection next; until then, use + /// [`Daemon::with_store`] rather than locking at the call site. + store: Option>>, /// `session/load` may replay already persisted ACP updates. While the /// replay matches local history in order, drop it; the first mismatch is /// new live output and disables the guard. - resume_replay: HashMap>, + resume_replay: HashMap, /// Per-project git worktree managers, lazily created on first worktree use. worktrees: HashMap, + /// Sessions waiting on a worktree checkout, keyed by task id. Presence is + /// the token that lets a finished checkout start its session: cancel and + /// delete remove the entry, so a late checkout cannot resurrect the task. + pending_session_starts: HashMap, /// Policy engine: gates agent actions through configurable policies. policies: PolicyRegistry, /// Channel for ACP reader tasks to request policy checks before file ops. @@ -1896,6 +2112,20 @@ pub struct Daemon { pending_wake: std::collections::HashSet, /// Stable first-seen timestamps for streamed frames of the same tool call. tool_call_starts: HashMap<(String, String), u64>, + /// The last session update emitted per task — all `emit_session_unless_last_duplicate` + /// needs to catch a reconnect retry or a repeated usage frame. O(1) per task, + /// never the whole transcript. + last_session_update: HashMap, + /// Updates emitted since the task's last user message (its current turn). + /// Reset on each new user message, so this is bounded by a turn, not by the + /// session's length. Serves the workflow engine's stage-text reads, which + /// used to fold the entire in-memory transcript. + turn_updates: HashMap>, + /// A session that cannot start until its resume replay guard has been read + /// from the store (off the loop). Presence is the token that lets the loaded + /// guard start the session: cancel and delete remove the entry, so a late + /// load cannot resurrect the task (ADR 0002 invariant 5). + pending_resume: HashMap, /// Deterministic workflow pipelines keyed by parent task id. Finished runs /// stay in the map so their state remains visible on the board. workflow_runs: HashMap, @@ -1960,28 +2190,24 @@ impl Daemon { .flatten() .unwrap_or_default(); + // Only the stable tool-call timestamps survive startup. The full + // transcripts are NOT loaded or held in memory — resume replay guards, + // snapshots and finished-turn output read them from the store on demand. let tool_call_starts = store .as_ref() - .and_then(|s| s.load_all_session_updates().ok()) - .unwrap_or_default() - .into_iter() - .flat_map(|(task_id, updates)| { - updates.into_iter().filter_map(move |update| match update { - wire::SessionUpdate::ToolCall { - tool_call_id, - started_at: Some(started_at), - .. - } => Some(((task_id.clone(), tool_call_id), started_at)), - _ => None, - }) - }) - .collect(); + .and_then(|s| s.load_tool_call_starts().ok()) + .unwrap_or_default(); let accounts = store .as_ref() .and_then(|s| s.load_accounts().ok()) .unwrap_or_default(); + // Everything above read from the store directly — it is startup, the + // actor is not running yet. From here the connection belongs to the + // persistence thread and writes go through the queue. + let (persist, store) = super::runtime::Persist::spawn(store); + let config_observer = ConfigObserver::new(&projects); let daemon = Daemon { projects, @@ -1997,9 +2223,11 @@ impl Daemon { event_tx: event_tx.clone(), acp_tx, cmd_tx: cmd_tx.clone(), + persist, store, resume_replay: HashMap::new(), worktrees: HashMap::new(), + pending_session_starts: HashMap::new(), policies: default_policies(), policy_tx, orch_tx: None, @@ -2008,6 +2236,9 @@ impl Daemon { orchestrator_inbox: HashMap::new(), pending_wake: std::collections::HashSet::new(), tool_call_starts, + last_session_update: HashMap::new(), + turn_updates: HashMap::new(), + pending_resume: HashMap::new(), workflow_runs: HashMap::new(), accounts, }; @@ -2070,9 +2301,23 @@ impl Daemon { } fn persist(&self, task: &Task) { - if let Some(store) = &self.store { - let _ = store.upsert_task(task); - } + self.persist.task(task); + } + + /// A blocking store read, for startup only. + /// + /// Its one caller — `restore_workflow_runs` — runs inside [`Daemon::spawn`] + /// before the actor loop starts, so blocking here blocks nothing. A handler + /// must never use it: that is a blocking disk read on the actor's thread, + /// which is the whole subject of ADR 0002. Reads from a handler go through + /// `runtime::store_read` on the blocking pool, and writes through + /// `self.persist`. + fn with_store(&self, read: impl FnOnce(&Store) -> T) -> Option { + let store = self.store.as_ref()?; + // Recover a poisoned lock instead of taking the daemon down with the + // persistence thread. + let guard = store.lock().unwrap_or_else(|e| e.into_inner()); + Some(read(&guard)) } fn build_project_config_state( @@ -2201,7 +2446,10 @@ impl Daemon { } /// Build the serializable snapshot handed to a client on subscribe. - fn build_snapshot(&self) -> wire::Snapshot { + /// + /// `session_history` is filled in by the caller (from the store, off the + /// loop) — the actor holds no in-memory transcript to fold here. + fn build_snapshot_core(&self) -> wire::Snapshot { let mut projects = Vec::new(); let mut services = Vec::new(); let mut portforwards = Vec::new(); @@ -2232,19 +2480,15 @@ impl Daemon { }) .collect(); - let session_history = self - .store - .as_ref() - .and_then(|s| s.load_all_session_updates().ok()) - .unwrap_or_default(); - + // History is read from the store by the caller, then folded; the actor + // holds no transcript in memory to fold here. wire::Snapshot { projects, services, portforwards, tasks, terminals, - session_history, + session_history: HashMap::new(), agents: self.configured_agents.clone(), accounts: self.account_infos(), } @@ -2328,8 +2572,19 @@ impl Daemon { // Teardown — stop everything we started. self.services.stop_all().await.ok(); self.portforwards.stop_all().await.ok(); - kill_listeners_in_ranges(&self.project_port_ranges()).await; + // Only ports this daemon handed out. Sweeping the whole range kills + // whatever else happens to listen there — a developer's own server, or + // an agent process — and this runs on every shutdown, including the + // ones the test suite performs on the developer's machine. + crate::service::kill_listeners_on_ports(&crate::ports::allocated_in_ranges( + &self.project_port_ranges(), + )) + .await; self.agents.kill_all(); + // Writes are applied on another thread, so exiting without draining the + // queue drops the tail of every transcript written since the last + // batch. Everything above can still enqueue, so flush last. + self.persist.flush().await; match shutdown_reply { Some(ShutdownReply::Requested(reply)) => { let _ = reply.send(()); @@ -2581,7 +2836,22 @@ impl Daemon { let _ = reply.send(tasks); } Command::Snapshot(reply) => { - let _ = reply.send(self.build_snapshot()); + // The snapshot's history must not be read on the loop: a disk + // read here would miss whatever write-behind persistence still + // has queued, and it would block the actor (ADR 0002). Flush + + // read + fold happen on a worker; only the reply crosses back. + let mut snapshot = self.build_snapshot_core(); + let persist = self.persist.clone(); + let store = self.store.clone(); + tokio::spawn(async move { + persist.flush().await; + snapshot.session_history = super::runtime::store_read(store, |store| { + store.load_all_session_updates().unwrap_or_default() + }) + .await + .unwrap_or_default(); + let _ = reply.send(snapshot); + }); } Command::OpenProject { name } => self.open_project(&name).await, Command::StartService { project, service } => { @@ -2761,27 +3031,6 @@ impl Daemon { .map(str::to_string); let mut task = Task::new(&project, &prompt, &agent, tags); task.parent_task_id = parent_task_id; - // Create worktree if requested and project has a git repo. - if use_worktree { - if let Some(path) = self.project_path(&project) { - let wt_mgr = self.worktrees.entry(project.clone()).or_insert_with(|| { - WorktreeManager::new(std::path::PathBuf::from(&path)) - }); - let created = match branched_from { - Some(ref src) => wt_mgr.create_branched(&task.id, src).await, - None => wt_mgr.create(&task.id, None).await, - }; - match created { - Ok(wt) => { - task.worktree = Some(wt.path.to_string_lossy().to_string()); - } - Err(e) => { - eprintln!("[daemon] worktree creation failed: {e}"); - // Fall back to non-isolated run. - } - } - } - } // Resolve the model the session should start with: an explicit // UI pick wins; otherwise fall back to the user's last choice // for this agent (so orchestrator-spawned sub-agents inherit it @@ -2799,13 +3048,11 @@ impl Daemon { { if agent_cfg.last_model.as_deref() != Some(m.as_str()) { agent_cfg.last_model = Some(m.clone()); - if let Some(ref store) = self.store { - let _ = store.update_agent_models( - &agent_cfg.id, - &agent_cfg.models, - agent_cfg.last_model.as_deref(), - ); - } + self.persist.write(PersistWrite::AgentModels { + id: agent_cfg.id.clone(), + models: agent_cfg.models.clone(), + last_model: agent_cfg.last_model.clone(), + }); let agents = self.configured_agents.clone(); self.emit(Event::AgentsUpdated { agents }); } @@ -2816,17 +3063,133 @@ impl Daemon { self.persist(&task); self.emit(Event::TaskCreated(task)); let _ = reply.send(id.clone()); - self.start_session( - &id, - &project, - &agent, - &prompt, + + let start = PendingSessionStart { + project: project.clone(), + agent: agent.clone(), + prompt: prompt.clone(), include_runtime_context, - None, attachments, - resolved_model, + default_model: resolved_model, config_overrides, - ); + }; + // A worktree checkout is git work, so it runs off the loop and + // the session starts when it lands. The task is on the board + // before then, which is also why a new task no longer delays + // every other task's messages (ADR 0002). + match use_worktree + .then(|| self.worktree_request(&id, &project, branched_from.as_deref())) + .flatten() + { + Some(request) => { + self.pending_session_starts.insert(id.clone(), start); + let cmd_tx = self.cmd_tx.clone(); + tokio::spawn(async move { + let created = request.run().await; + let _ = cmd_tx + .send(Command::WorktreeReady { + task_id: id, + created, + }) + .await; + }); + } + None => self.start_pending_session(&id, start), + } + } + Command::WorktreeReady { task_id, created } => { + // Record the checkout even if nobody is waiting for it any + // more: the directory exists on disk either way, and a + // worktree the manager does not know about is one nothing can + // clean up later. + match created { + Ok((project, wt)) => { + if let Some(task) = self.tasks.get_mut(&task_id) { + task.worktree = Some(wt.path.to_string_lossy().to_string()); + let updated = task.clone(); + self.persist(&updated); + self.emit(Event::TaskUpdated(updated)); + } + if let Some(mgr) = self.worktrees.get_mut(&project) { + mgr.adopt(wt); + } + } + // Fall back to a non-isolated run, as before. + Err(error) => eprintln!("[daemon] worktree creation failed: {error}"), + } + // The pending entry is the token: cancelling or deleting the + // task removes it, so a checkout that lands afterwards must not + // start a session for it (ADR 0002 invariant 5). + if let Some(start) = self.pending_session_starts.remove(&task_id) { + self.start_pending_session(&task_id, start); + } + } + #[cfg(test)] + Command::TurnOutputConsumerProbe { + task_id, + workflow_child, + reply, + } => { + let _ = reply.send(self.turn_output_has_consumer(&task_id, workflow_child)); + } + Command::GitOpFinished { task_id, effect } => match effect { + GitEffect::Bump => self.bump_task(&task_id), + GitEffect::Committed => { + if let Some(task) = self.tasks.get_mut(&task_id) { + task.updated_at = super::task::now_secs(); + task.files_changed = 0; + let updated = task.clone(); + self.persist(&updated); + self.emit(Event::TaskUpdated(updated)); + } + } + GitEffect::HunkRejected => { + if let Some(task) = self.tasks.get_mut(&task_id) { + task.updated_at = super::task::now_secs(); + task.files_changed = task.files_changed.saturating_sub(1); + let updated = task.clone(); + self.persist(&updated); + self.emit(Event::TaskUpdated(updated)); + } + } + }, + Command::ResumeReplayReady { + task_id, + mut replay, + } => { + // The pending entry is the token: cancelling or deleting the + // task removes it, so a guard that lands afterwards must not + // resurrect a cancelled task's session (ADR 0002 invariant 5). + if let Some(pending) = self.pending_resume.remove(&task_id) { + if let Some(guard) = ResumeReplayGuard::from_updates(replay.make_contiguous()) { + self.resume_replay.insert(task_id.clone(), guard); + } + self.start_session( + &task_id, + &pending.project, + &pending.agent, + &pending.text, + false, + Some(pending.session_id), + pending.attachments, + None, + std::collections::HashMap::new(), + ); + } + } + Command::TaskOutputReady { + task_id, + success, + workflow_child, + output, + } => { + // A finished turn's full text was assembled off the loop; now + // deliver it the way TurnEnded used to. notify_orch_finished is + // a no-op unless the task is an orchestrator child. + self.notify_orch_finished(&task_id, success, output.clone()); + if !workflow_child { + self.deliver_child_result(&task_id, success, output); + } } Command::CreateWorkflowTask { project, @@ -2893,23 +3256,27 @@ impl Daemon { let _ = reply.send(results); } Command::GetDiff { task_id, reply } => { - // Resolve the repo path (sync) before awaiting git, so no shared - // borrow of self is held across the await. + // Resolve the repo path from actor state, then run git off the + // loop. The diff panel polls this, so awaiting it here put a + // pair of git processes between every poll and the next + // command — a tool approval included (ADR 0002). let repo = self .tasks .get(&task_id) .and_then(|_| self.task_repo_path(&task_id)); - let (files, branch) = match repo { - Some(path) => ( - super::diff::working_diff(&path).await.unwrap_or_default(), - super::diff::current_branch(&path).await, - ), - None => (Vec::new(), None), - }; - let _ = reply.send(wire::TaskDiff { - task_id, - files, - branch, + tokio::spawn(async move { + let (files, branch) = match repo { + Some(path) => ( + super::diff::working_diff(&path).await.unwrap_or_default(), + super::diff::current_branch(&path).await, + ), + None => (Vec::new(), None), + }; + let _ = reply.send(wire::TaskDiff { + task_id, + files, + branch, + }); }); } Command::GetFileContents { @@ -2921,11 +3288,13 @@ impl Daemon { .tasks .get(&task_id) .and_then(|_| self.task_repo_path(&task_id)); - let doc = match repo { - Some(p) => super::diff::file_doc(&p, &path).await.ok(), - None => None, - }; - let _ = reply.send(doc); + tokio::spawn(async move { + let doc = match repo { + Some(p) => super::diff::file_doc(&p, &path).await.ok(), + None => None, + }; + let _ = reply.send(doc); + }); } Command::ListFiles { task_id, @@ -2938,13 +3307,15 @@ impl Daemon { .get(&task_id) .and_then(|_| self.task_repo_path(&task_id)) .or_else(|| project.as_deref().and_then(|name| self.project_path(name))); - let files = match repo { - Some(p) => super::diff::list_files(&p, include_ignored) - .await - .unwrap_or_default(), - None => Vec::new(), - }; - let _ = reply.send(files); + tokio::spawn(async move { + let files = match repo { + Some(p) => super::diff::list_files(&p, include_ignored) + .await + .unwrap_or_default(), + None => Vec::new(), + }; + let _ = reply.send(files); + }); } Command::SearchFiles { task_id, @@ -2956,11 +3327,21 @@ impl Daemon { .tasks .get(&task_id) .and_then(|t| self.project_path(&t.project)); - let matches = match repo { - Some(p) => super::diff::search_files(&p, &query, limit).unwrap_or_default(), - None => Vec::new(), - }; - let _ = reply.send(matches); + match repo { + // A synchronous walk that reads every file in the project. + // Run inline it freezes the whole daemon for the length of + // the search — on a large repo, seconds (ADR 0002). + Some(p) => { + tokio::task::spawn_blocking(move || { + let matches = + super::diff::search_files(&p, &query, limit).unwrap_or_default(); + let _ = reply.send(matches); + }); + } + None => { + let _ = reply.send(Vec::new()); + } + } } Command::SaveFile { task_id, @@ -2971,17 +3352,17 @@ impl Daemon { .tasks .get(&task_id) .and_then(|_| self.task_repo_path(&task_id)); - if let Some(p) = repo { + let cmd_tx = self.cmd_tx.clone(); + tokio::task::spawn_blocking(move || { + let Some(p) = repo else { return }; if super::diff::save_file(&p, &path, &content).is_ok() { // Nudge clients so the diff/file list refetches. - if let Some(task) = self.tasks.get_mut(&task_id) { - task.updated_at = super::task::now_secs(); - let updated = task.clone(); - self.persist(&updated); - self.emit(Event::TaskUpdated(updated)); - } + let _ = cmd_tx.blocking_send(Command::GitOpFinished { + task_id, + effect: GitEffect::Bump, + }); } - } + }); } Command::CreateFile { task_id, @@ -2989,15 +3370,21 @@ impl Daemon { directory, reply, } => { - let result = self + // Filesystem work: resolve the path here, touch the disk on the + // blocking pool (ADR 0002 invariant 1). + let repo = self .tasks .get(&task_id) - .and_then(|_| self.task_repo_path(&task_id)) - .ok_or_else(|| format!("no repo for task {task_id}")) - .and_then(|repo| { - super::diff::create_file(&repo, &path, directory).map_err(|e| e.to_string()) - }); - let _ = reply.send(result); + .and_then(|_| self.task_repo_path(&task_id)); + tokio::task::spawn_blocking(move || { + let result = repo + .ok_or_else(|| format!("no repo for task {task_id}")) + .and_then(|repo| { + super::diff::create_file(&repo, &path, directory) + .map_err(|e| e.to_string()) + }); + let _ = reply.send(result); + }); } Command::RenameFile { task_id, @@ -3005,30 +3392,37 @@ impl Daemon { new_path, reply, } => { - let result = self + let repo = self .tasks .get(&task_id) - .and_then(|_| self.task_repo_path(&task_id)) - .ok_or_else(|| format!("no repo for task {task_id}")) - .and_then(|repo| { - super::diff::rename_file(&repo, &path, &new_path).map_err(|e| e.to_string()) - }); - let _ = reply.send(result); + .and_then(|_| self.task_repo_path(&task_id)); + tokio::task::spawn_blocking(move || { + let result = repo + .ok_or_else(|| format!("no repo for task {task_id}")) + .and_then(|repo| { + super::diff::rename_file(&repo, &path, &new_path) + .map_err(|e| e.to_string()) + }); + let _ = reply.send(result); + }); } Command::DeleteFile { task_id, path, reply, } => { - let result = self + let repo = self .tasks .get(&task_id) - .and_then(|_| self.task_repo_path(&task_id)) - .ok_or_else(|| format!("no repo for task {task_id}")) - .and_then(|repo| { - super::diff::delete_file(&repo, &path).map_err(|e| e.to_string()) - }); - let _ = reply.send(result); + .and_then(|_| self.task_repo_path(&task_id)); + tokio::task::spawn_blocking(move || { + let result = repo + .ok_or_else(|| format!("no repo for task {task_id}")) + .and_then(|repo| { + super::diff::delete_file(&repo, &path).map_err(|e| e.to_string()) + }); + let _ = reply.send(result); + }); } Command::ResolveHunk { task_id, @@ -3039,22 +3433,21 @@ impl Daemon { // accept keeps the change (no-op); only reject touches the tree. if resolution == wire::HunkResolution::Reject { let repo = self.task_repo_path(&task_id); - if let Some(path) = repo { + let cmd_tx = self.cmd_tx.clone(); + tokio::spawn(async move { + let Some(path) = repo else { return }; if super::diff::reject_hunk(&path, &file, hunk_index) .await .is_ok() { - if let Some(task) = self.tasks.get_mut(&task_id) { - task.updated_at = super::task::now_secs(); - if task.files_changed > 0 { - task.files_changed -= 1; - } - let updated = task.clone(); - self.persist(&updated); - self.emit(Event::TaskUpdated(updated)); - } + let _ = cmd_tx + .send(Command::GitOpFinished { + task_id, + effect: GitEffect::HunkRejected, + }) + .await; } - } + }); } } Command::GitCommit { @@ -3064,47 +3457,62 @@ impl Daemon { amend, reply, } => { + // git shells out; resolve the repo here and run it off the loop, + // reporting what changed back as GitOpFinished (ADR 0002). let repo = self.task_repo_path(&task_id); - let result = match repo { - Some(p) => super::diff::commit(&p, &message, files.as_deref(), amend) - .await - .map_err(|e| e.to_string()), - None => Err(format!("no repo for task {task_id}")), - }; - if result.is_ok() { - if let Some(task) = self.tasks.get_mut(&task_id) { - task.updated_at = super::task::now_secs(); - task.files_changed = 0; - let updated = task.clone(); - self.persist(&updated); - self.emit(Event::TaskUpdated(updated)); + let cmd_tx = self.cmd_tx.clone(); + tokio::spawn(async move { + let result = match repo { + Some(p) => super::diff::commit(&p, &message, files.as_deref(), amend) + .await + .map_err(|e| e.to_string()), + None => Err(format!("no repo for task {task_id}")), + }; + if result.is_ok() { + let _ = cmd_tx + .send(Command::GitOpFinished { + task_id, + effect: GitEffect::Committed, + }) + .await; } - } - let _ = reply.send(result); + let _ = reply.send(result); + }); } Command::GitUpdate { task_id, reply } => { + // git shells out; resolve the repo here and run it off + // the loop, reporting what changed back as + // GitOpFinished (ADR 0002). let repo = self.task_repo_path(&task_id); - let result = match repo { - Some(p) => super::diff::update_project(&p).await.unwrap_or_else(|e| { - wire::GitOpResult { + let cmd_tx = self.cmd_tx.clone(); + tokio::spawn(async move { + let result = match repo { + Some(p) => super::diff::update_project(&p).await.unwrap_or_else(|e| { + wire::GitOpResult { + status: wire::GitOpStatus::Error, + message: e.to_string(), + conflicts: Vec::new(), + branch: None, + } + }), + None => wire::GitOpResult { status: wire::GitOpStatus::Error, - message: e.to_string(), + message: format!("no repo for task {task_id}"), conflicts: Vec::new(), branch: None, - } - }), - None => wire::GitOpResult { - status: wire::GitOpStatus::Error, - message: format!("no repo for task {task_id}"), - conflicts: Vec::new(), - branch: None, - }, - }; - // A clean update changed HEAD/tree — nudge clients to refetch. - if result.status == wire::GitOpStatus::Ok { - self.bump_task(&task_id); - } - let _ = reply.send(result); + }, + }; + // A clean update changed HEAD/tree — nudge clients to refetch. + if result.status == wire::GitOpStatus::Ok { + let _ = cmd_tx + .send(Command::GitOpFinished { + task_id, + effect: GitEffect::Bump, + }) + .await; + } + let _ = reply.send(result); + }); } Command::GitBranches { task_id, @@ -3117,39 +3525,52 @@ impl Daemon { Some(id) => self.task_repo_path(&id), None => project.as_deref().and_then(|p| self.project_path(p)), }; - let list = match repo { - Some(p) => super::diff::list_branches(&p).await.unwrap_or_default(), - None => wire::GitBranchList::default(), - }; - let _ = reply.send(list); + tokio::spawn(async move { + let list = match repo { + Some(p) => super::diff::list_branches(&p).await.unwrap_or_default(), + None => wire::GitBranchList::default(), + }; + let _ = reply.send(list); + }); } Command::GitSwitchBranch { task_id, branch, reply, } => { + // git shells out; resolve the repo here and run it off + // the loop, reporting what changed back as + // GitOpFinished (ADR 0002). let repo = self.task_repo_path(&task_id); - let result = match repo { - Some(p) => super::diff::switch_branch(&p, &branch) - .await - .unwrap_or_else(|e| wire::GitOpResult { + let cmd_tx = self.cmd_tx.clone(); + tokio::spawn(async move { + let result = match repo { + Some(p) => super::diff::switch_branch(&p, &branch) + .await + .unwrap_or_else(|e| wire::GitOpResult { + status: wire::GitOpStatus::Error, + message: e.to_string(), + conflicts: Vec::new(), + branch: None, + }), + None => wire::GitOpResult { status: wire::GitOpStatus::Error, - message: e.to_string(), + message: format!("no repo for task {task_id}"), conflicts: Vec::new(), branch: None, - }), - None => wire::GitOpResult { - status: wire::GitOpStatus::Error, - message: format!("no repo for task {task_id}"), - conflicts: Vec::new(), - branch: None, - }, - }; - // Switching branches changes the whole working tree — refetch. - if result.status == wire::GitOpStatus::Ok { - self.bump_task(&task_id); - } - let _ = reply.send(result); + }, + }; + // Switching branches changes the whole working tree — refetch. + if result.status == wire::GitOpStatus::Ok { + let _ = cmd_tx + .send(Command::GitOpFinished { + task_id, + effect: GitEffect::Bump, + }) + .await; + } + let _ = reply.send(result); + }); } Command::GitBranchRename { task_id, @@ -3157,27 +3578,38 @@ impl Daemon { new_name, reply, } => { + // git shells out; resolve the repo here and run it off + // the loop, reporting what changed back as + // GitOpFinished (ADR 0002). let repo = self.task_repo_path(&task_id); - let result = match repo { - Some(p) => super::diff::rename_branch(&p, &branch, &new_name) - .await - .unwrap_or_else(|e| wire::GitOpResult { + let cmd_tx = self.cmd_tx.clone(); + tokio::spawn(async move { + let result = match repo { + Some(p) => super::diff::rename_branch(&p, &branch, &new_name) + .await + .unwrap_or_else(|e| wire::GitOpResult { + status: wire::GitOpStatus::Error, + message: e.to_string(), + conflicts: Vec::new(), + branch: None, + }), + None => wire::GitOpResult { status: wire::GitOpStatus::Error, - message: e.to_string(), + message: format!("no repo for task {task_id}"), conflicts: Vec::new(), branch: None, - }), - None => wire::GitOpResult { - status: wire::GitOpStatus::Error, - message: format!("no repo for task {task_id}"), - conflicts: Vec::new(), - branch: None, - }, - }; - if result.status == wire::GitOpStatus::Ok { - self.bump_task(&task_id); - } - let _ = reply.send(result); + }, + }; + if result.status == wire::GitOpStatus::Ok { + let _ = cmd_tx + .send(Command::GitOpFinished { + task_id, + effect: GitEffect::Bump, + }) + .await; + } + let _ = reply.send(result); + }); } Command::GitBranchDelete { task_id, @@ -3185,27 +3617,38 @@ impl Daemon { force, reply, } => { + // git shells out; resolve the repo here and run it off + // the loop, reporting what changed back as + // GitOpFinished (ADR 0002). let repo = self.task_repo_path(&task_id); - let result = match repo { - Some(p) => super::diff::delete_branch(&p, &branch, force) - .await - .unwrap_or_else(|e| wire::GitOpResult { + let cmd_tx = self.cmd_tx.clone(); + tokio::spawn(async move { + let result = match repo { + Some(p) => super::diff::delete_branch(&p, &branch, force) + .await + .unwrap_or_else(|e| wire::GitOpResult { + status: wire::GitOpStatus::Error, + message: e.to_string(), + conflicts: Vec::new(), + branch: None, + }), + None => wire::GitOpResult { status: wire::GitOpStatus::Error, - message: e.to_string(), + message: format!("no repo for task {task_id}"), conflicts: Vec::new(), branch: None, - }), - None => wire::GitOpResult { - status: wire::GitOpStatus::Error, - message: format!("no repo for task {task_id}"), - conflicts: Vec::new(), - branch: None, - }, - }; - if result.status == wire::GitOpStatus::Ok { - self.bump_task(&task_id); - } - let _ = reply.send(result); + }, + }; + if result.status == wire::GitOpStatus::Ok { + let _ = cmd_tx + .send(Command::GitOpFinished { + task_id, + effect: GitEffect::Bump, + }) + .await; + } + let _ = reply.send(result); + }); } Command::GitBranchCreate { task_id, @@ -3215,29 +3658,44 @@ impl Daemon { overwrite, reply, } => { + // git shells out; resolve the repo here and run it off + // the loop, reporting what changed back as + // GitOpFinished (ADR 0002). let repo = self.task_repo_path(&task_id); - let result = match repo { - Some(p) => { - super::diff::branch_create(&p, &name, from.as_deref(), checkout, overwrite) - .await - .unwrap_or_else(|e| wire::GitOpResult { - status: wire::GitOpStatus::Error, - message: e.to_string(), - conflicts: Vec::new(), - branch: None, + let cmd_tx = self.cmd_tx.clone(); + tokio::spawn(async move { + let result = match repo { + Some(p) => super::diff::branch_create( + &p, + &name, + from.as_deref(), + checkout, + overwrite, + ) + .await + .unwrap_or_else(|e| wire::GitOpResult { + status: wire::GitOpStatus::Error, + message: e.to_string(), + conflicts: Vec::new(), + branch: None, + }), + None => wire::GitOpResult { + status: wire::GitOpStatus::Error, + message: format!("no repo for task {task_id}"), + conflicts: Vec::new(), + branch: None, + }, + }; + if result.status == wire::GitOpStatus::Ok { + let _ = cmd_tx + .send(Command::GitOpFinished { + task_id, + effect: GitEffect::Bump, }) + .await; } - None => wire::GitOpResult { - status: wire::GitOpStatus::Error, - message: format!("no repo for task {task_id}"), - conflicts: Vec::new(), - branch: None, - }, - }; - if result.status == wire::GitOpStatus::Ok { - self.bump_task(&task_id); - } - let _ = reply.send(result); + let _ = reply.send(result); + }); } Command::GitRebase { task_id, @@ -3245,54 +3703,76 @@ impl Daemon { target, reply, } => { + // git shells out; resolve the repo here and run it off + // the loop, reporting what changed back as + // GitOpFinished (ADR 0002). let repo = self.task_repo_path(&task_id); - let result = match repo { - Some(p) => super::diff::rebase(&p, &branch, &target) - .await - .unwrap_or_else(|e| wire::GitOpResult { + let cmd_tx = self.cmd_tx.clone(); + tokio::spawn(async move { + let result = match repo { + Some(p) => super::diff::rebase(&p, &branch, &target) + .await + .unwrap_or_else(|e| wire::GitOpResult { + status: wire::GitOpStatus::Error, + message: e.to_string(), + conflicts: Vec::new(), + branch: None, + }), + None => wire::GitOpResult { status: wire::GitOpStatus::Error, - message: e.to_string(), + message: format!("no repo for task {task_id}"), conflicts: Vec::new(), branch: None, - }), - None => wire::GitOpResult { - status: wire::GitOpStatus::Error, - message: format!("no repo for task {task_id}"), - conflicts: Vec::new(), - branch: None, - }, - }; - if result.status == wire::GitOpStatus::Ok { - self.bump_task(&task_id); - } - let _ = reply.send(result); + }, + }; + if result.status == wire::GitOpStatus::Ok { + let _ = cmd_tx + .send(Command::GitOpFinished { + task_id, + effect: GitEffect::Bump, + }) + .await; + } + let _ = reply.send(result); + }); } Command::GitMerge { task_id, target, reply, } => { + // git shells out; resolve the repo here and run it off + // the loop, reporting what changed back as + // GitOpFinished (ADR 0002). let repo = self.task_repo_path(&task_id); - let result = match repo { - Some(p) => super::diff::merge(&p, &target).await.unwrap_or_else(|e| { - wire::GitOpResult { + let cmd_tx = self.cmd_tx.clone(); + tokio::spawn(async move { + let result = match repo { + Some(p) => super::diff::merge(&p, &target).await.unwrap_or_else(|e| { + wire::GitOpResult { + status: wire::GitOpStatus::Error, + message: e.to_string(), + conflicts: Vec::new(), + branch: None, + } + }), + None => wire::GitOpResult { status: wire::GitOpStatus::Error, - message: e.to_string(), + message: format!("no repo for task {task_id}"), conflicts: Vec::new(), branch: None, - } - }), - None => wire::GitOpResult { - status: wire::GitOpStatus::Error, - message: format!("no repo for task {task_id}"), - conflicts: Vec::new(), - branch: None, - }, - }; - if result.status == wire::GitOpStatus::Ok { - self.bump_task(&task_id); - } - let _ = reply.send(result); + }, + }; + if result.status == wire::GitOpStatus::Ok { + let _ = cmd_tx + .send(Command::GitOpFinished { + task_id, + effect: GitEffect::Bump, + }) + .await; + } + let _ = reply.send(result); + }); } Command::GitPushInfo { task_id, reply } => { let repo = self.tasks.get(&task_id).and_then(|task| { @@ -3300,44 +3780,57 @@ impl Daemon { .clone() .or_else(|| self.project_path(&task.project)) }); - let result = match repo { - Some(path) => super::diff::push_info(&path) - .await - .map_err(|e| e.to_string()), - None => Err(format!("no repo for task {task_id}")), - }; - let _ = reply.send(result); + tokio::spawn(async move { + let result = match repo { + Some(path) => super::diff::push_info(&path) + .await + .map_err(|e| e.to_string()), + None => Err(format!("no repo for task {task_id}")), + }; + let _ = reply.send(result); + }); } Command::GitPush { task_id, force, reply, } => { + // git shells out; resolve the repo here and run it off + // the loop, reporting what changed back as + // GitOpFinished (ADR 0002). let repo = self.tasks.get(&task_id).and_then(|task| { task.worktree .clone() .or_else(|| self.project_path(&task.project)) }); - let result = match repo { - Some(path) => super::diff::push(&path, force).await.unwrap_or_else(|e| { - wire::GitOpResult { + let cmd_tx = self.cmd_tx.clone(); + tokio::spawn(async move { + let result = match repo { + Some(path) => super::diff::push(&path, force).await.unwrap_or_else(|e| { + wire::GitOpResult { + status: wire::GitOpStatus::Error, + message: e.to_string(), + conflicts: Vec::new(), + branch: None, + } + }), + None => wire::GitOpResult { status: wire::GitOpStatus::Error, - message: e.to_string(), + message: format!("no repo for task {task_id}"), conflicts: Vec::new(), branch: None, - } - }), - None => wire::GitOpResult { - status: wire::GitOpStatus::Error, - message: format!("no repo for task {task_id}"), - conflicts: Vec::new(), - branch: None, - }, - }; - if result.status == wire::GitOpStatus::Ok { - self.bump_task(&task_id); - } - let _ = reply.send(result); + }, + }; + if result.status == wire::GitOpStatus::Ok { + let _ = cmd_tx + .send(Command::GitOpFinished { + task_id, + effect: GitEffect::Bump, + }) + .await; + } + let _ = reply.send(result); + }); } Command::GitCreatePr { task_id, @@ -3351,13 +3844,18 @@ impl Daemon { .clone() .or_else(|| self.project_path(&task.project)) }); - let result = match repo { - Some(path) => super::diff::create_pr(&path, &title, &body, base.as_deref()) - .await - .map_err(|e| e.to_string()), - None => Err(format!("no repo for task {task_id}")), - }; - let _ = reply.send(result); + // Creating a PR shells out to the forge's CLI over the network; + // it changes nothing the actor holds, so it just answers from a + // task of its own (ADR 0002). + tokio::spawn(async move { + let result = match repo { + Some(path) => super::diff::create_pr(&path, &title, &body, base.as_deref()) + .await + .map_err(|e| e.to_string()), + None => Err(format!("no repo for task {task_id}")), + }; + let _ = reply.send(result); + }); } Command::GenerateText { task_id, @@ -3411,6 +3909,13 @@ impl Daemon { Some(handle) => handle.cancel_and_wait().await, None => Ok(()), }; + // A worktree checkout may still be running for this task; + // dropping its token stops it from starting a session the + // user just cancelled. Same for a pending resume: its + // guard must not start a session the user cancelled. + self.pending_session_starts.remove(&id); + self.pending_resume.remove(&id); + self.resume_replay.remove(&id); self.pending_permissions.cleanup_task(&id); // A finished pipeline's parent keeps its terminal status: // cancelling it must not rewrite that back to Waiting. @@ -3481,9 +3986,8 @@ impl Daemon { }; let mut delete_result = stop_result; if delete_result.is_ok() && self.workflow_runs.remove(&id).is_some() { - if let Some(store) = &self.store { - let _ = store.delete_workflow_run(&id); - } + self.persist + .write(PersistWrite::DeleteWorkflowRun(id.clone())); } if delete_result.is_ok() { self.pending_permissions.cleanup_task(&id); @@ -3501,10 +4005,16 @@ impl Daemon { if delete_result.is_ok() && self.tasks.remove(&id).is_some() { self.tool_call_starts .retain(|(task_id, _), _| task_id != &id); - if let Some(store) = &self.store { - if let Err(error) = store.delete_task(&id) { - delete_result = Err(error.to_string()); - } + self.last_session_update.remove(&id); + self.turn_updates.remove(&id); + self.resume_replay.remove(&id); + self.pending_resume.remove(&id); + self.pending_session_starts.remove(&id); + // Awaited, not queued: a failed delete is reported to the + // user, and dropping the error would leave a task that + // reappears on the next start with no explanation. + if let Err(error) = self.persist.ask(PersistAsk::DeleteTask(id.clone())).await { + delete_result = Err(error); } self.emit(Event::TaskRemoved { id: id.clone() }); // Deleting a stage child mid-run fails that stage. @@ -3522,37 +4032,61 @@ impl Daemon { } } Command::MergeWorktree { task_id, reply } => { - let result = if let Some(task) = self.tasks.get(&task_id) { - if let Some(wt_mgr) = self.worktrees.get(&task.project) { - match wt_mgr.merge(&task_id).await { - Ok(super::worktree::MergeResult::Ok { branch }) => { - // Clean up after merge. - if let Some(wt_mgr) = self.worktrees.get_mut(&task.project) { - let _ = wt_mgr.remove(&task_id).await; - } - // Clear the worktree field on the task. - if let Some(task) = self.tasks.get_mut(&task_id) { - task.worktree = None; - task.updated_at = super::task::now_secs(); - let updated = task.clone(); - self.persist(&updated); - self.emit(Event::TaskUpdated(updated)); - } - Ok(branch) - } - Ok(super::worktree::MergeResult::Conflict { message, branch }) => { - Err(format!("merge conflict on {branch}: {message}")) - } - Ok(super::worktree::MergeResult::Error(msg)) => Err(msg), - Err(e) => Err(e.to_string()), - } - } else { - Err("no worktree manager for this project".into()) - } - } else { - Err(format!("unknown task {task_id}")) + // Merging runs two git commands and removes a checkout. Resolve + // what they need here, run them off the loop, and record the + // outcome through Command::WorktreeMerged (ADR 0002). + let resolved = self + .tasks + .get(&task_id) + .map(|t| t.project.clone()) + .and_then(|project| { + let mgr = self.worktrees.get(&project)?; + let wt = mgr.get(&task_id)?; + Some(( + project, + mgr.base_repo().to_path_buf(), + wt.path.clone(), + wt.branch.clone(), + wt.base_branch.clone(), + )) + }); + let Some((project, base_repo, path, branch, base_branch)) = resolved else { + let _ = reply.send(Err(format!("no worktree for task {task_id}"))); + return; }; - let _ = reply.send(result); + let cmd_tx = self.cmd_tx.clone(); + tokio::spawn(async move { + let merged = + super::worktree::merge_detached(&base_repo, &branch, &base_branch).await; + let result = match merged { + Ok(super::worktree::MergeResult::Ok { branch }) => { + let _ = + super::worktree::remove_detached(&base_repo, &path, &branch).await; + let _ = cmd_tx + .send(Command::WorktreeMerged { task_id, project }) + .await; + Ok(branch) + } + Ok(super::worktree::MergeResult::Conflict { message, branch }) => { + Err(format!("merge conflict on {branch}: {message}")) + } + Ok(super::worktree::MergeResult::Error(msg)) => Err(msg), + Err(e) => Err(format!("{e:#}")), + }; + let _ = reply.send(result); + }); + } + Command::WorktreeMerged { task_id, project } => { + if let Some(mgr) = self.worktrees.get_mut(&project) { + mgr.forget(&task_id); + } + if let Some(task) = self.tasks.get_mut(&task_id) { + task.worktree = None; + task.updated_at = super::task::now_secs(); + let updated = task.clone(); + self.persist(&updated); + self.emit(Event::TaskUpdated(updated)); + } } Command::ListWorktrees { project, reply } => { let wts = if let Some(wt_mgr) = self.worktrees.get(&project) { @@ -3779,24 +4313,30 @@ impl Daemon { if let Some((project, agent, session_id)) = resume { self.mark_task_running(&task_id); - self.prepare_resume_replay_guard(&task_id); self.emit_session( &task_id, wire::SessionUpdate::AgentText { text: "Reconnecting to the saved agent session…".into(), }, ); - self.start_session( - &task_id, - &project, - &agent, - &text, - false, - Some(session_id), - attachments, - None, - std::collections::HashMap::new(), + // The replay guard is built from the persisted + // transcript, which must be read off the loop + // (write-behind flush + store read). Start the + // session only once the guard has landed, mirroring + // WorktreeReady: starting before it would let the + // agent's replayed history through unfiltered and + // double the output. + self.pending_resume.insert( + task_id.clone(), + PendingResume { + project, + agent, + text: text.clone(), + session_id, + attachments, + }, ); + self.request_resume_replay_guard(&task_id); let _ = reply.send(Ok(())); } else { // Reject without echoing a user message that was never delivered. @@ -3840,9 +4380,7 @@ impl Daemon { }); } Command::UpdateAgents { agents } => { - if let Some(store) = &self.store { - let _ = store.save_agents(&agents); - } + self.persist.write(PersistWrite::Agents(agents.clone())); self.configured_agents = agents.clone(); self.emit(Event::AgentsUpdated { agents: self.configured_agents.clone(), @@ -3866,7 +4404,8 @@ impl Daemon { label, reply, } => { - let _ = reply.send(self.import_account(&agent_id, &label)); + let result = self.import_account(&agent_id, &label).await; + let _ = reply.send(result); } Command::RenameAccount { account_id, @@ -3877,9 +4416,7 @@ impl Daemon { Some(account) => { account.label = label; let updated = account.clone(); - if let Some(store) = &self.store { - let _ = store.upsert_account(&updated); - } + self.persist.write(PersistWrite::Account(Box::new(updated))); Ok(()) } None => Err(format!("no account {account_id}")), @@ -3887,14 +4424,15 @@ impl Daemon { let _ = reply.send(result.map(|()| self.emit_accounts())); } Command::RemoveAccount { account_id, reply } => { - let _ = reply.send(self.remove_account(&account_id)); + let result = self.remove_account(&account_id).await; + let _ = reply.send(result); } Command::SetActiveAccount { agent_id, account_id, reply, } => { - let result = self.set_active_account(&agent_id, &account_id); + let result = self.set_active_account(&agent_id, &account_id).await; let _ = reply.send(result); } Command::ProbeAgent { id } => { @@ -3936,9 +4474,11 @@ impl Daemon { if let Some(agent) = self.configured_agents.iter_mut().find(|a| a.id == id) { agent.models = models.clone(); agent.last_model = last_model.clone(); - if let Some(store) = &self.store { - let _ = store.update_agent_models(&id, &models, last_model.as_deref()); - } + self.persist.write(PersistWrite::AgentModels { + id: id.clone(), + models: models.clone(), + last_model: last_model.clone(), + }); } self.emit(Event::AgentsUpdated { agents: self.configured_agents.clone(), @@ -3987,10 +4527,10 @@ impl Daemon { } Command::SaveOrchestratorConfig { config, reply } => { self.orch_config = config.into(); - // Persist to store if available. - if let Some(ref store) = self.store { - let _ = store.save_orchestrator_config(&self.orch_config); - } + self.persist + .write(PersistWrite::OrchestratorConfig(Box::new( + self.orch_config.clone(), + ))); let _ = reply.send(true); } Command::SetTaskStatus { id, status } => { @@ -4420,7 +4960,7 @@ impl Daemon { /// Register the agent's currently-authenticated login as a new account by /// copying its credentials into a fresh vault. The agent's own home is only /// ever read. - fn import_account( + async fn import_account( &mut self, agent_id: &str, label: &str, @@ -4458,11 +4998,19 @@ impl Daemon { // wait to be picked, so importing never moves live sessions. active: !self.accounts.iter().any(|a| a.agent_id == agent_id), }; - if let Some(store) = &self.store { - store.upsert_account(&account).map_err(|e| e.to_string())?; - if account.active { - let _ = store.set_active_account(agent_id, &account.id); - } + let active = account.active; + let account_id = account.id.clone(); + self.persist + .ask(PersistAsk::Account(Box::new(account.clone()))) + .await?; + if active { + let _ = self + .persist + .ask(PersistAsk::SetActiveAccount { + agent_id: agent_id.to_string(), + account_id, + }) + .await; } self.accounts.push(account); Ok(self.emit_accounts()) @@ -4475,7 +5023,7 @@ impl Daemon { /// the selection is only recorded if that succeeded: a stored "active" /// account the CLI is not actually using would misreport which login every /// session runs under. - fn set_active_account( + async fn set_active_account( &mut self, agent_id: &str, account_id: &str, @@ -4501,11 +5049,12 @@ impl Daemon { ) .map_err(|e| e.to_string())?; } - if let Some(store) = &self.store { - store - .set_active_account(agent_id, account_id) - .map_err(|e| e.to_string())?; - } + self.persist + .ask(PersistAsk::SetActiveAccount { + agent_id: agent_id.to_string(), + account_id: account_id.to_string(), + }) + .await?; for account in &mut self.accounts { if account.agent_id == agent_id { account.active = account.id == account_id; @@ -4554,18 +5103,16 @@ impl Daemon { .unwrap_or(agent) } - fn remove_account(&mut self, account_id: &str) -> Result, String> { + async fn remove_account(&mut self, account_id: &str) -> Result, String> { let Some(index) = self.accounts.iter().position(|a| a.id == account_id) else { return Err(format!("no account {account_id}")); }; let account = self.accounts[index].clone(); super::accounts::remove_vault(std::path::Path::new(&account.home_dir), &account.id) .map_err(|e| e.to_string())?; - if let Some(store) = &self.store { - store - .delete_account(account_id) - .map_err(|e| e.to_string())?; - } + self.persist + .ask(PersistAsk::DeleteAccount(account_id.to_string())) + .await?; self.accounts.remove(index); if account.agent_id == "claude" { let _ = self @@ -4583,7 +5130,7 @@ impl Daemon { .find(|a| a.agent_id == account.agent_id) .map(|a| a.id.clone()) { - self.set_active_account(&account.agent_id, &next_id)?; + self.set_active_account(&account.agent_id, &next_id).await?; } } Ok(self.emit_accounts()) @@ -4622,6 +5169,51 @@ impl Daemon { /// /// If the task has a worktree, the agent runs in the worktree directory /// instead of the project root — so its edits are isolated. + /// Start a session whose worktree checkout has finished. + fn start_pending_session(&mut self, task_id: &str, start: PendingSessionStart) { + self.start_session( + task_id, + &start.project, + &start.agent, + &start.prompt, + start.include_runtime_context, + None, + start.attachments, + start.default_model, + start.config_overrides, + ); + } + + /// Resolve everything a worktree checkout needs from actor state, so the + /// git work itself can run without borrowing the actor. + fn worktree_request( + &mut self, + task_id: &str, + project: &str, + branched_from: Option<&str>, + ) -> Option { + let path = self.project_path(project)?; + // Resolve the source's working directory before borrowing the manager. + // A source task without a worktree runs in the project checkout, and + // that is still the tree its branch must inherit from. + let source_path = branched_from.and_then(|src| self.task_repo_path(src)); + let mgr = self + .worktrees + .entry(project.to_string()) + .or_insert_with(|| WorktreeManager::new(std::path::PathBuf::from(&path))); + let base_branch = branched_from.and_then(|src| mgr.source_state(src).map(|(b, _)| b)); + let source = source_path.map(|path| BranchSource { + base_branch, + path: PathBuf::from(path), + }); + Some(WorktreeRequest { + project: project.to_string(), + base_repo: mgr.base_repo().to_path_buf(), + task_id: task_id.to_string(), + source, + }) + } + #[allow(clippy::too_many_arguments)] fn start_session( &mut self, @@ -4907,25 +5499,30 @@ impl Daemon { } } } - let output = self.collect_agent_text(&task_id); - self.notify_orch_finished(&task_id, success, output.clone()); if workflow_child { // A workflow stage finished — advance the pipeline. Parse - // only the latest turn's text: answered questions and - // superseded verdicts from earlier turns must not count. - // The legacy orchestrator inbox path does not apply here. + // only the latest turn's text from the in-memory turn buffer + // (bounded by a turn): answered questions and superseded + // verdicts from earlier turns must not count. The legacy + // orchestrator inbox path does not apply here. let text = self.collect_stage_text(&task_id); self.workflow_stage_finished(&task_id, success, text).await; - } else { - // Deliver to a parent if this was a sub-agent; and drain our - // own inbox if we are a parent that just went idle. - self.deliver_child_result(&task_id, success, output); } // If we are an orchestrator whose sub-agents finished mid-turn, // process them now that the turn is over. if self.pending_wake.remove(&task_id) { self.wake_parent(&task_id); } + // The finished task's full text output is assembled off the loop + // (write-behind flush + store read) and delivered back as + // Command::TaskOutputReady, which notifies the orchestrator and + // the parent inbox. Only ask for it when somebody consumes it: + // both consumers are no-ops for an ordinary task, and reading + // its whole transcript per turn would trade the memory this + // change saves for disk it never needed to touch. + if self.turn_output_has_consumer(&task_id, workflow_child) { + self.request_task_output(&task_id, success, workflow_child); + } } AcpUpdate::Error { run_id, message } => { if self @@ -4988,31 +5585,99 @@ impl Daemon { } } - fn emit_session_unless_last_duplicate(&self, task_id: &str, update: wire::SessionUpdate) { - if let Some(store) = &self.store { - if let Ok(Some(last)) = store.load_last_session_update(task_id) { - if last == update { - return; - } - } + /// Emit unless this is byte-for-byte the update that went out last for the + /// task — a reconnect retry re-sending a prompt, or a repeated usage frame. + /// + /// The comparison is against what this daemon last emitted, held in memory. + /// It used to `SELECT` the last persisted row, which write-behind + /// persistence makes wrong as well as slow: the row it needs is usually + /// still in the queue, so every duplicate would slip through. + fn emit_session_unless_last_duplicate(&mut self, task_id: &str, update: wire::SessionUpdate) { + if self.last_session_update.get(task_id) == Some(&update) { + return; } self.emit_session(task_id, update); } - fn prepare_resume_replay_guard(&mut self, task_id: &str) { - let Some(store) = &self.store else { - return; - }; - let Ok(updates) = store.load_session_updates(task_id) else { - return; + /// Ask a worker to read this task's persisted history off the loop and send + /// the replay guard back as [`Command::ResumeReplayReady`]. The session does + /// not start until then (see the SessionPrompt resume path). + fn request_resume_replay_guard(&self, task_id: &str) { + let persist = self.persist.clone(); + let store = self.store.clone(); + let cmd_tx = self.cmd_tx.clone(); + let task_id = task_id.to_string(); + tokio::spawn(async move { + persist.flush().await; + let lookup = task_id.clone(); + let replay = super::runtime::store_read(store, move |store| { + store + .load_session_updates(&lookup) + .map(|updates| replayable_history(&updates)) + .unwrap_or_default() + }) + .await + .unwrap_or_default(); + let _ = cmd_tx + .send(Command::ResumeReplayReady { task_id, replay }) + .await; + }); + } + + /// Whether anything reads a finished turn's full text output. + /// + /// `notify_orch_finished` is a no-op unless the task is an orchestrator + /// node, and `deliver_child_result` returns early without a parent — and it + /// is skipped entirely for a workflow stage, which reads its own turn + /// buffer instead. For everything else the assembled output is discarded, + /// so it should never be assembled. + fn turn_output_has_consumer(&self, task_id: &str, workflow_child: bool) -> bool { + let Some(task) = self.tasks.get(task_id) else { + return false; }; - let replayable = updates - .into_iter() - .filter(is_acp_replay_update) - .collect::>(); - if !replayable.is_empty() { - self.resume_replay.insert(task_id.to_string(), replayable); - } + let orchestrator_node = + self.orch_tx.is_some() && task.tags.iter().any(|tag| tag == "orchestrator"); + let feeds_parent = !workflow_child && task.parent_task_id.is_some(); + orchestrator_node || feeds_parent + } + + /// Ask a worker to assemble a finished task's full text output off the loop + /// and send it back as [`Command::TaskOutputReady`], so the orchestrator / + /// parent-inbox delivery never blocks the actor on a disk read. + fn request_task_output(&self, task_id: &str, success: bool, workflow_child: bool) { + let persist = self.persist.clone(); + let store = self.store.clone(); + let cmd_tx = self.cmd_tx.clone(); + let task_id = task_id.to_string(); + // Without a database the actor's turn buffer is the only history there + // is; hand it back rather than an empty result. + let fallback = agent_text_from_updates( + self.turn_updates + .get(&task_id) + .map(Vec::as_slice) + .unwrap_or_default(), + ); + tokio::spawn(async move { + persist.flush().await; + let lookup = task_id.clone(); + let output = super::runtime::store_read(store, move |store| { + store + .load_session_updates(&lookup) + .map(|updates| agent_text_from_updates(&updates)) + .unwrap_or_default() + }) + .await + // Without a database the actor's turn buffer is the only history. + .unwrap_or(fallback); + let _ = cmd_tx + .send(Command::TaskOutputReady { + task_id, + success, + workflow_child, + output, + }) + .await; + }); } fn should_skip_resume_replay(&mut self, task_id: &str, update: &wire::SessionUpdate) -> bool { @@ -5020,13 +5685,12 @@ impl Daemon { return false; } - let Some(history) = self.resume_replay.get_mut(task_id) else { + let Some(guard) = self.resume_replay.get_mut(task_id) else { return false; }; - if history.front() == Some(update) { - history.pop_front(); - if history.is_empty() { + if guard.consume(update) { + if guard.is_empty() { self.resume_replay.remove(task_id); } return true; @@ -5038,30 +5702,10 @@ impl Daemon { false } - /// Concatenate the agent's text output for a task (its persisted - /// `AgentText` updates) — used as the orchestrator node's result, e.g. the - /// planner's task-graph JSON. - fn collect_agent_text(&self, task_id: &str) -> String { - let Some(store) = &self.store else { - return String::new(); - }; - let Ok(updates) = store.load_session_updates(task_id) else { - return String::new(); - }; - updates - .into_iter() - .filter_map(|u| match u { - wire::SessionUpdate::AgentText { text } => Some(text), - _ => None, - }) - .collect::>() - .join("") - } - - /// Like [`collect_agent_text`], but only the text streamed since the last - /// user message — i.e. the output of the task's latest turn. The workflow - /// engine parses this: a `need_user_input` block answered two turns ago - /// must not be mistaken for a fresh question. + /// Like [`agent_text_from_updates`], but only the text streamed since the + /// last user message — i.e. the output of the task's latest turn. The + /// workflow engine parses this: a `need_user_input` block answered two turns + /// ago must not be mistaken for a fresh question. fn collect_last_turn_text(&self, task_id: &str) -> String { self.collect_stage_text(task_id).full } @@ -5074,40 +5718,16 @@ impl Daemon { /// reads as the result, so it is what reviewers and fixers are handed. /// `full` is every chunk of the turn, kept as a parsing fallback for an /// agent that emits its protocol block before a trailing tool call. + /// + /// Reads the in-memory current-turn buffer (bounded by a turn, reset on + /// each user message), not the whole session transcript. fn collect_stage_text(&self, task_id: &str) -> StageText { - let Some(updates) = self - .store - .as_ref() - .and_then(|s| s.load_session_updates(task_id).ok()) - else { - return StageText::default(); - }; - let mut full: Vec = Vec::new(); - let mut closing: Vec = Vec::new(); - for update in updates { - match update { - // A new user message starts a fresh turn. - wire::SessionUpdate::UserMessage { .. } => { - full.clear(); - closing.clear(); - } - wire::SessionUpdate::AgentText { text } => { - full.push(text.clone()); - closing.push(text); - } - // Any work the agent does ends whatever it was narrating, so - // the closing message restarts after it. - wire::SessionUpdate::ToolCall { .. } - | wire::SessionUpdate::FileEdit { .. } - | wire::SessionUpdate::AgentThought { .. } - | wire::SessionUpdate::Plan { .. } => closing.clear(), - _ => {} - } - } - StageText { - closing: closing.join(""), - full: full.join(""), - } + let updates = self + .turn_updates + .get(task_id) + .map(Vec::as_slice) + .unwrap_or_default(); + stage_text_from_updates(updates) } /// Tell the orchestrator a dispatched task finished. No-op unless the task @@ -5195,10 +5815,19 @@ impl Daemon { }); } - fn emit_session(&self, task_id: &str, update: wire::SessionUpdate) { - if let Some(store) = &self.store { - let _ = store.save_session_update(task_id, &update); + fn emit_session(&mut self, task_id: &str, update: wire::SessionUpdate) { + self.persist.session_update(task_id, &update); + // A new user message begins a fresh turn: drop the previous turn's + // buffer so stage-text reads stay bounded by a turn, not the session. + if matches!(update, wire::SessionUpdate::UserMessage { .. }) { + self.turn_updates.remove(task_id); } + self.turn_updates + .entry(task_id.to_string()) + .or_default() + .push(update.clone()); + self.last_session_update + .insert(task_id.to_string(), update.clone()); self.emit(Event::SessionUpdate { task_id: task_id.to_string(), update, @@ -5397,7 +6026,7 @@ impl Daemon { /// but would glue unrelated workflow transitions into one Markdown blob. #[allow(clippy::too_many_arguments)] fn workflow_event( - &self, + &mut self, parent_id: &str, event: wire::WorkflowEventKind, title: impl Into, @@ -5422,7 +6051,7 @@ impl Daemon { /// Convenience wrapper for transitions that do not reference a particular /// agent. Split the first paragraph into the card title and keep the rest /// as Markdown detail. - fn workflow_timeline(&self, parent_id: &str, text: impl Into) { + fn workflow_timeline(&mut self, parent_id: &str, text: impl Into) { let text = text.into(); let text = text.trim(); let (heading, detail) = text @@ -5478,10 +6107,8 @@ impl Daemon { self.persist(&updated); self.emit(Event::TaskUpdated(updated)); } - if let Some(store) = &self.store { - if let Ok(json) = serde_json::to_string(run) { - let _ = store.save_workflow_run(&run.parent_id, &json); - } + if let Ok(json) = serde_json::to_string(run) { + self.persist.workflow_run(&run.parent_id, json); } } @@ -6019,6 +6646,51 @@ impl Daemon { /// A non-review stage child failed, or a reviewer died. Reviewers are /// excluded from the verdict; any other stage failure fails the pipeline. + /// Park a run whose stage lost its agent, instead of failing it outright. + /// + /// Losing the agent process is an infrastructure failure, not a verdict: + /// the stage never got to say whether the work was good. Failing the run + /// made that unrecoverable — the pipeline was finished, and a user whose + /// agent died (or who revived the session by hand and watched it finish the + /// work) had no way to continue and had to start over. Parking at the + /// existing pause barrier leaves it resumable, and resume re-runs the stage. + /// + /// Mirrors the daemon-restart recovery in `restore_workflow_runs`, down to + /// warning the re-run that the working copy may already hold partial work. + fn workflow_park_after_failure( + &mut self, + parent_id: &str, + mut run: WorkflowRun, + stage: StageKind, + reason: &str, + ) { + run.active_children.clear(); + if stage == StageKind::Review { + run.review_pending.clear(); + run.review_collected.clear(); + run.reasked.clear(); + // Re-running a review re-increments `round` on spawn; give the + // abandoned round back or the re-run reports "round 3/2". + run.round = run.round.saturating_sub(1); + } + run.pause_requested = false; + run.state = RunState::Paused { next: stage }; + run.pending_guidance = Some(format!( + "The previous attempt of this stage ended before it finished: {reason}. The working \ + copy may already contain its partial changes — inspect the current diff before \ + assuming you are starting from scratch." + )); + self.workflow_sync(&run); + self.workflow_runs.insert(parent_id.to_string(), run); + self.workflow_timeline( + parent_id, + format!( + "Stage **{}** lost its agent: {reason}. Paused — resume to run it again.", + stage.label() + ), + ); + } + async fn workflow_child_failed( &mut self, parent_id: &str, @@ -6056,13 +6728,15 @@ impl Daemon { ); if run.review_pending.is_empty() { if run.review_collected.is_empty() { - self.workflow_runs.insert(parent_id.to_string(), run); - let _ = self - .workflow_finalize( - parent_id, - WorkflowOutcome::Error("all reviewers failed".to_string()), - ) - .await; + // Every reviewer lost its agent, so the round produced no + // verdict at all. That is the same infrastructure failure + // as a dead implement stage, not a rejection of the work. + self.workflow_park_after_failure( + parent_id, + run, + stage, + "every reviewer's agent ended before producing a verdict", + ); } else { self.workflow_merge_reviews(parent_id, run).await; } @@ -6097,13 +6771,7 @@ impl Daemon { event_agent.into_iter().collect(), wire::WorkflowEventTone::Error, ); - self.workflow_runs.insert(parent_id.to_string(), run); - let _ = self - .workflow_finalize( - parent_id, - WorkflowOutcome::Error(format!("stage {} failed: {reason}", stage.label())), - ) - .await; + self.workflow_park_after_failure(parent_id, run, stage, &reason); } /// One reviewer's turn ended: parse its verdict, re-ask once on garbage, @@ -6679,9 +7347,8 @@ impl Daemon { /// (resume re-runs the interrupted stage from scratch). fn restore_workflow_runs(&mut self) { let rows = self - .store - .as_ref() - .and_then(|s| s.load_workflow_runs().ok()) + .with_store(|store| store.load_workflow_runs().ok()) + .flatten() .unwrap_or_default(); for (task_id, json) in rows { let Ok(mut run) = serde_json::from_str::(&json) else { @@ -6689,9 +7356,8 @@ impl Daemon { // the parent sits with no pipeline state and therefore no // pause/resume/stop controls. Say so, once, and move on. eprintln!("[daemon] dropping unreadable workflow run for task {task_id}"); - if let Some(store) = &self.store { - let _ = store.delete_workflow_run(&task_id); - } + self.persist + .write(PersistWrite::DeleteWorkflowRun(task_id.clone())); if let Some(task) = self.tasks.get_mut(&task_id) { task.blocked_reason = Some("workflow state could not be restored after an upgrade".to_string()); @@ -6761,10 +7427,8 @@ impl Daemon { let updated = task.clone(); self.persist(&updated); } - if let Some(store) = &self.store { - if let Ok(json) = serde_json::to_string(&run) { - let _ = store.save_workflow_run(&task_id, &json); - } + if let Ok(json) = serde_json::to_string(&run) { + self.persist.workflow_run(&task_id, json); } self.workflow_runs.insert(task_id, run); } @@ -7245,3 +7909,423 @@ mod lifecycle_action_tests { assert_eq!(task.snoozed_at, None); } } + +#[cfg(test)] +mod worktree_start_tests { + use super::*; + use crate::registry::ProjectEntry; + use std::time::Duration; + + const MOCK_AGENT: &str = concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/fixtures/mock-acp-inspect.mjs" + ); + + /// A git repo with one commit, so `git worktree add` has something to + /// branch from. + async fn repo_with_commit() -> tempfile::TempDir { + let dir = tempfile::tempdir().unwrap(); + let git = |args: &[&str]| { + tokio::process::Command::new("git") + .args(args) + .current_dir(dir.path()) + .env("GIT_AUTHOR_NAME", "test") + .env("GIT_AUTHOR_EMAIL", "t@t") + .env("GIT_COMMITTER_NAME", "test") + .env("GIT_COMMITTER_EMAIL", "t@t") + .status() + }; + git(&["init"]).await.unwrap(); + std::fs::write(dir.path().join("README.md"), "init\n").unwrap(); + git(&["add", "."]).await.unwrap(); + git(&["commit", "-m", "init"]).await.unwrap(); + dir + } + + async fn spawn_with_repo(dir: &tempfile::TempDir) -> DaemonHandle { + Daemon::spawn( + vec![ProjectEntry { + name: "demo".into(), + path: dir.path().to_string_lossy().into_owned(), + added_at: "0".into(), + }], + None, + ) + } + + async fn create_worktree_task(handle: &DaemonHandle) -> String { + handle + .create_task( + "demo", + "do the thing", + &format!("node {MOCK_AGENT}"), + Vec::new(), + false, + true, + None, + Vec::new(), + None, + Default::default(), + ) + .await + } + + async fn task_now(handle: &DaemonHandle, id: &str) -> Task { + handle + .tasks() + .await + .into_iter() + .find(|t| t.id == id) + .expect("task on the board") + } + + /// The task must reach the board before its checkout finishes. It used to + /// be created only after `git worktree add` returned, so starting a task + /// held up every other task's messages and approvals (ADR 0002). + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn task_appears_before_its_worktree_is_ready() { + let dir = repo_with_commit().await; + let handle = spawn_with_repo(&dir).await; + + let id = create_worktree_task(&handle).await; + assert!(!id.is_empty()); + assert_eq!( + task_now(&handle, &id).await.worktree, + None, + "create must return before the checkout, not after it" + ); + + // The worktree is attached once the checkout lands. + let mut path = None; + for _ in 0..100 { + if let Some(p) = task_now(&handle, &id).await.worktree { + path = Some(p); + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + let path = path.expect("checkout should attach a worktree"); + assert!(std::path::Path::new(&path).exists(), "worktree on disk"); + + handle.shutdown().await; + } + + /// Branching a conversation whose source runs in the project checkout — + /// no worktree of its own — must still carry the uncommitted work over. + /// + /// This regressed once: the lookup only knew how to find a source + /// *worktree*, so a source without one silently produced a branch on a + /// clean HEAD, and the change the user was continuing from was gone. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn branching_from_the_project_checkout_carries_its_changes() { + let dir = repo_with_commit().await; + let handle = spawn_with_repo(&dir).await; + + // The source task has no worktree, and its checkout has uncommitted + // work: one tracked edit and one new file. + let source = handle + .create_task( + "demo", + "source", + &format!("node {MOCK_AGENT}"), + Vec::new(), + false, + false, + None, + Vec::new(), + None, + Default::default(), + ) + .await; + std::fs::write(dir.path().join("README.md"), "edited\n").unwrap(); + std::fs::write(dir.path().join("NEW.md"), "new file\n").unwrap(); + + let branch = handle + .create_task( + "demo", + "branch", + &format!("node {MOCK_AGENT}"), + vec![format!("branched-from:{source}")], + false, + true, + None, + Vec::new(), + None, + Default::default(), + ) + .await; + + let mut path = None; + for _ in 0..100 { + if let Some(p) = task_now(&handle, &branch).await.worktree { + path = Some(p); + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + let path = std::path::PathBuf::from(path.expect("branch should get a worktree")); + + assert_eq!( + std::fs::read_to_string(path.join("README.md")).unwrap(), + "edited\n", + "the tracked edit must carry over" + ); + assert_eq!( + std::fs::read_to_string(path.join("NEW.md")).unwrap(), + "new file\n", + "the new untracked file must carry over" + ); + + handle.shutdown().await; + } + + /// Cancelling while the checkout is still running must not start a session + /// when it lands — but the worktree still gets recorded, because it exists + /// on disk and something has to be able to clean it up. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancelling_during_checkout_does_not_start_a_session() { + let dir = repo_with_commit().await; + let handle = spawn_with_repo(&dir).await; + + let id = create_worktree_task(&handle).await; + handle.cancel_task(&id).await.ok(); + + // Wait for the checkout to land, then give a session every chance to + // start before concluding that none did. + for _ in 0..100 { + if task_now(&handle, &id).await.worktree.is_some() { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + tokio::time::sleep(Duration::from_millis(300)).await; + + let task = task_now(&handle, &id).await; + assert!( + task.worktree.is_some(), + "the checkout must still be recorded so it can be cleaned up" + ); + assert_eq!( + task.session_id, None, + "a cancelled task must not be started by its own checkout" + ); + + handle.shutdown().await; + } +} + +/// Transcript memory: the daemon holds no full session transcript in memory, +/// only bounded projections. These tests pin the behavior of the projections +/// that replaced it. +#[cfg(test)] +mod transcript_projection_tests { + use super::*; + + /// A plain task's finished turn feeds nothing: the orchestrator hook is a + /// no-op without the tag, and there is no parent inbox. Assembling its + /// output would read the whole transcript back per turn — trading the + /// memory this projection saves for disk it never needed. + #[tokio::test] + async fn a_plain_task_does_not_assemble_turn_output() { + let handle = Daemon::spawn(Vec::new(), None); + let id = handle + .create_task( + "demo", + "prompt", + "agent", + Vec::new(), + false, + false, + None, + Vec::new(), + None, + Default::default(), + ) + .await; + + let (tx, rx) = oneshot::channel(); + handle + .send(Command::TurnOutputConsumerProbe { + task_id: id.clone(), + workflow_child: false, + reply: tx, + }) + .await; + assert!(!rx.await.unwrap(), "nothing consumes a plain task's output"); + + // A sub-agent's parent does consume it. + let child = handle + .create_task( + "demo", + "child", + "agent", + Vec::new(), + false, + false, + Some(id.clone()), + Vec::new(), + None, + Default::default(), + ) + .await; + let (tx, rx) = oneshot::channel(); + handle + .send(Command::TurnOutputConsumerProbe { + task_id: child, + workflow_child: false, + reply: tx, + }) + .await; + assert!(rx.await.unwrap(), "a sub-agent's result feeds its parent"); + + handle.shutdown().await; + } + + /// A long, realistic history: alternating tool calls and streamed text + /// chunks, as a long agent turn produces. + fn long_history(turns: usize, chunks_per_turn: usize) -> Vec { + let mut history = Vec::new(); + for turn in 0..turns { + history.push(wire::SessionUpdate::UserMessage { + text: format!("prompt {turn}"), + attachments: vec![], + }); + for chunk in 0..chunks_per_turn { + history.push(wire::SessionUpdate::ToolCall { + tool_call_id: format!("turn-{turn}-call-{chunk}"), + title: "tool".into(), + status: wire::ToolCallStatus::Completed, + started_at: Some(1000 + (turn * chunks_per_turn + chunk) as u64), + tool_kind: "read".into(), + content: None, + }); + history.push(wire::SessionUpdate::AgentText { + text: format!("turn-{turn} chunk {chunk} "), + }); + } + } + history + } + + /// A resumed session replays its whole persisted history, update for + /// update, before producing live output. The guard must drop every replayed + /// update — a long history must not surface as duplicated output — and then + /// let live output through. + #[test] + fn resume_replay_guard_drops_long_replayed_history_whole() { + let history = long_history(10, 20); // 410 updates, 400 of them replayable + let mut guard = ResumeReplayGuard::from_updates(&history).expect("replayable history"); + + // The guard only covers the replayable subset; user prompts are not + // part of the agent's replay. + let replayable = replayable_history(&history); + let mut dropped = 0; + for update in &replayable { + if guard.consume(update) { + dropped += 1; + } + } + assert_eq!( + dropped, + replayable.len(), + "every replayed update must be dropped — none may reach the UI twice" + ); + assert!(guard.is_empty(), "guard exhausted after the replay"); + + // Live output after the replay is never dropped. + let live = wire::SessionUpdate::AgentText { + text: "fresh output".into(), + }; + assert!(!guard.consume(&live)); + } + + /// The guard reports non-matches so the caller (should_skip_resume_replay) + /// can disable it on the first divergence — otherwise a live update that + /// happens to equal a later entry of the old history would be eaten. + #[test] + fn resume_replay_guard_reports_divergence() { + let history = long_history(1, 3); + // history[0] is the user prompt (not replayable); the first replayed + // update is history[1]. + let mut guard = ResumeReplayGuard::from_updates(&history).unwrap(); + assert!(guard.consume(&history[1])); + assert!( + !guard.consume(&wire::SessionUpdate::AgentText { + text: "diverged".into() + }), + "a divergent update must be reported so the guard can be disabled" + ); + } + + /// Stage text must reflect only the latest turn: after several turns, the + /// closing message and the full-turn text must not leak earlier turns' text. + #[test] + fn stage_text_is_scoped_to_the_latest_turn() { + let history = vec![ + wire::SessionUpdate::UserMessage { + text: "first".into(), + attachments: vec![], + }, + wire::SessionUpdate::AgentText { + text: "old turn text ".into(), + }, + wire::SessionUpdate::UserMessage { + text: "second".into(), + attachments: vec![], + }, + wire::SessionUpdate::AgentText { + text: "work ".into(), + }, + wire::SessionUpdate::ToolCall { + tool_call_id: "c1".into(), + title: "tool".into(), + status: wire::ToolCallStatus::Completed, + started_at: Some(2000), + tool_kind: "read".into(), + content: None, + }, + wire::SessionUpdate::AgentText { + text: "final message".into(), + }, + ]; + let text = stage_text_from_updates(&history); + assert_eq!( + text.full, "work final message", + "earlier turns must not leak into full" + ); + assert_eq!( + text.closing, "final message", + "tool call restarts the closing message" + ); + } + + /// The orchestrator's node result is the task's whole text output, + /// including text from every turn. + #[test] + fn agent_text_spans_every_turn() { + let history = long_history(3, 2); + let text = agent_text_from_updates(&history); + assert_eq!(text, "turn-0 chunk 0 turn-0 chunk 1 turn-1 chunk 0 turn-1 chunk 1 turn-2 chunk 0 turn-2 chunk 1 "); + } + + /// The replayable subset skips the "Reconnecting…" placeholder the daemon + /// emits before a resume, so it is never replayed back as agent output. + #[test] + fn replayable_history_excludes_reconnect_placeholder() { + let history = vec![ + wire::SessionUpdate::AgentText { + text: "Reconnecting to the saved agent session…".into(), + }, + wire::SessionUpdate::AgentText { + text: "real text".into(), + }, + wire::SessionUpdate::UserMessage { + text: "prompt".into(), + attachments: vec![], + }, + ]; + let replayable = replayable_history(&history); + assert_eq!(replayable.len(), 1); + assert_eq!(replayable[0], history[1]); + } +} diff --git a/src/daemon/diff.rs b/src/daemon/diff.rs index 8c73b67..95b2e15 100644 --- a/src/daemon/diff.rs +++ b/src/daemon/diff.rs @@ -40,36 +40,50 @@ pub async fn list_files(repo: &str, include_ignored: bool) -> Result>(); - let mut files = String::from_utf8_lossy(&out.stdout) - .lines() - .filter_map(|line| { - let path = line.trim(); - let exists = std::path::Path::new(repo).join(path).exists(); - (!path.is_empty() && exists && !is_ignored_path(path)).then(|| wire::ProjectFile { - path: path.to_string(), - changed: changed.contains(path), + return tokio::task::spawn_blocking(move || { + let mut files = String::from_utf8_lossy(&out.stdout) + .lines() + .filter_map(|line| { + let path = line.trim(); + let exists = std::path::Path::new(&repo).join(path).exists(); + (!path.is_empty() && exists && !is_ignored_path(path)).then(|| { + wire::ProjectFile { + path: path.to_string(), + changed: changed.contains(path), + } + }) }) - }) - .collect::>(); - files.sort_by(|a, b| a.path.cmp(&b.path)); - return Ok(files); + .collect::>(); + files.sort_by(|a, b| a.path.cmp(&b.path)); + files + }) + .await + .map_err(anyhow::Error::from); } - let mut files = Vec::new(); - walk_files( - std::path::Path::new(repo), - std::path::Path::new(repo), - &mut files, - )?; - files.sort_by(|a, b| a.path.cmp(&b.path)); - Ok(files) + tokio::task::spawn_blocking(move || { + let mut files = Vec::new(); + walk_files( + std::path::Path::new(&repo), + std::path::Path::new(&repo), + &mut files, + )?; + files.sort_by(|a, b| a.path.cmp(&b.path)); + Ok(files) + }) + .await? } /// Case-insensitive substring search across the project working tree. Reuses diff --git a/src/daemon/mod.rs b/src/daemon/mod.rs index c600ec2..a9189a8 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -19,6 +19,7 @@ pub mod diff; pub mod lsp; pub mod lsp_servers; pub mod prompt; +pub mod runtime; pub mod server; pub mod sessions; pub mod store; @@ -125,6 +126,55 @@ mod tests { ); } + /// Shutting a daemon down must not kill processes it never started. + /// + /// Teardown used to sweep the project's whole port range with `lsof` and + /// kill every listener in it. Every test that shuts a daemon down did that + /// too — so `cargo test` in this repo killed whatever the developer had + /// listening on 4000-4099, including the agent processes of the warpforge + /// running the tests. + #[tokio::test] + async fn shutdown_does_not_kill_listeners_it_did_not_start() { + // A stranger's server on a port inside the project's range. Spawned as + // a child process so the sweep would kill it, not the test runner. + let (start, end) = crate::ports::port_range(0); + let port = (start..=end) + .find(|p| std::net::TcpListener::bind(("127.0.0.1", *p)).is_ok()) + .expect("a free port in the project range"); + let mut stranger = tokio::process::Command::new("python3") + .args([ + "-c", + &format!( + "import socket,time\ns=socket.socket()\ns.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)\ns.bind(('127.0.0.1',{port}))\ns.listen()\ntime.sleep(30)" + ), + ]) + .spawn() + .expect("spawn the stranger"); + + // Wait until it is actually listening. + let mut listening = false; + for _ in 0..100 { + if std::net::TcpStream::connect(("127.0.0.1", port)).is_ok() { + listening = true; + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert!( + listening, + "the stranger should be listening before we start" + ); + + let daemon = Daemon::spawn(test_projects(), None); + daemon.shutdown().await; + + assert!( + stranger.try_wait().expect("poll the stranger").is_none(), + "daemon shutdown killed a process it never started" + ); + stranger.kill().await.ok(); + } + #[tokio::test] async fn session_id_stays_separate_from_task_id_when_attached() { // A task can attach a session without the two ids ever being unified — @@ -1172,6 +1222,67 @@ mod tests { ); } + /// Losing a stage's agent must not finish the pipeline. It used to call + /// workflow_finalize, which is terminal — the run was over, resume refused + /// with "the pipeline is not paused", and a user whose agent died had to + /// start again from a new task even when the work was already done. The run + /// now parks at the pause barrier, and resuming re-runs the stage. + #[tokio::test] + async fn workflow_lost_stage_agent_pauses_instead_of_failing() { + use warpforge_protocol as wire; + let (dir, projects) = workflow_project("name: placeholder\n"); + let reviewer = wf_agent(&dir, "rev.state", "approve"); + std::fs::write( + dir.path().join(".warpforge/workflows/test.yaml"), + format!("name: Lost agent\nreview:\n reviewers:\n - agent: {reviewer}\n"), + ) + .unwrap(); + // Implement dies mid-turn; the re-run after resume implements normally. + let lead = wf_agent(&dir, "impl.state", "die impl"); + + let store = Store::open_at(std::path::Path::new(":memory:")).ok(); + let daemon = Daemon::spawn(projects, store); + let mut events = daemon.subscribe(); + let parent_id = create_workflow_task(&daemon, &lead).await; + + let paused = wait_for_parent(&mut events, &parent_id, "paused after lost agent", |t| { + t.workflow_run + .as_ref() + .and_then(|w| w.waiting.as_ref()) + .is_some_and(|w| w.kind == wire::WorkflowWaitKind::Paused) + }) + .await; + assert_eq!(paused.status, TaskStatus::Waiting); + assert_eq!( + paused.workflow_run.as_ref().unwrap().stage, + wire::WorkflowStage::Implement, + "it parks at the stage that lost its agent, ready to re-run it" + ); + + // The run is genuinely resumable — the whole point. + let (tx, rx) = tokio::sync::oneshot::channel(); + daemon + .send(Command::WorkflowResume { + task: parent_id.clone(), + note: None, + reply: tx, + }) + .await; + rx.await.unwrap().expect("a parked run must accept resume"); + + let done = wait_for_parent(&mut events, &parent_id, "pipeline done", |t| { + t.workflow_run + .as_ref() + .is_some_and(|w| w.stage == wire::WorkflowStage::Done) + }) + .await; + assert_eq!( + done.workflow_run.unwrap().verdict, + Some(wire::WorkflowVerdict::Approve), + "resuming re-runs the stage and the pipeline completes" + ); + } + #[tokio::test] async fn workflow_plan_question_reply_flow() { use warpforge_protocol as wire; diff --git a/src/daemon/runtime/mod.rs b/src/daemon/runtime/mod.rs new file mode 100644 index 0000000..46c279e --- /dev/null +++ b/src/daemon/runtime/mod.rs @@ -0,0 +1,10 @@ +//! Daemon runtime: the concurrency machinery the actor delegates to. +//! +//! `actor.rs` owns state and decides; everything here owns work that must not +//! run on the actor's thread. Pieces land one at a time and take ownership when +//! they do — see `docs/adr/0002` for the target shape and the invariants that +//! keep blocking work from creeping back in. + +pub mod persist; + +pub use persist::{read as store_read, Ask, Persist, Write}; diff --git a/src/daemon/runtime/persist.rs b/src/daemon/runtime/persist.rs new file mode 100644 index 0000000..cd1a7f4 --- /dev/null +++ b/src/daemon/runtime/persist.rs @@ -0,0 +1,371 @@ +//! Write-behind persistence for the daemon actor. +//! +//! Every state change used to write to SQLite from inside the actor loop — +//! including one INSERT per streamed chunk of agent output. Those are blocking +//! disk writes on a tokio worker, so while an agent streamed, the actor could +//! not accept a tool approval queued behind it. See `docs/adr/0002`. +//! +//! Writes are queued here and applied by a dedicated OS thread that drains +//! whatever is pending into a single transaction. +//! +//! **Batching, never coalescing.** Rows keep their exact identity and order. +//! `actor.rs`'s resume replay guard compares persisted history against the +//! agent's replay chunk for chunk, so merging two `AgentText` rows into one +//! would silently break de-duplication and double the output on resume. + +use std::sync::{Arc, Mutex}; + +use tokio::sync::{mpsc, oneshot}; +use warpforge_protocol as wire; + +use crate::daemon::store::{Store, StoredAccount}; +use crate::daemon::task::Task; + +/// Upper bound on writes folded into one transaction. Large enough that a +/// stream burst commits once, small enough that a reader waiting on the store +/// mutex is never held off for long. +const MAX_BATCH: usize = 512; + +/// A queued write. Fire-and-forget: the actor does not learn whether it landed. +pub enum Write { + Task(Box), + DeleteTask(String), + SessionUpdate { + task_id: String, + update: Box, + }, + Agents(Vec), + AgentModels { + id: String, + models: Vec, + last_model: Option, + }, + Account(Box), + OrchestratorConfig(Box), + WorkflowRun { + task_id: String, + json: String, + }, + DeleteWorkflowRun(String), +} + +impl Write { + fn apply(self, store: &Store) -> anyhow::Result<()> { + match self { + Write::Task(task) => store.upsert_task(&task), + Write::DeleteTask(id) => store.delete_task(&id), + Write::SessionUpdate { task_id, update } => { + store.save_session_update(&task_id, &update) + } + Write::Agents(agents) => store.save_agents(&agents), + Write::AgentModels { + id, + models, + last_model, + } => store.update_agent_models(&id, &models, last_model.as_deref()), + Write::Account(account) => store.upsert_account(&account), + Write::OrchestratorConfig(config) => store.save_orchestrator_config(&config), + Write::WorkflowRun { task_id, json } => store.save_workflow_run(&task_id, &json), + Write::DeleteWorkflowRun(task_id) => store.delete_workflow_run(&task_id), + } + } +} + +/// A write whose outcome the caller needs, because it already reports failure +/// through the UI and dropping the error would change what the user sees. +/// +/// Every variant is user-initiated and rare — an account edit, a task deletion. +/// Nothing on the streaming path belongs here: awaiting a write from the actor +/// stalls the mailbox, which is the whole problem ADR 0002 exists to fix. +pub enum Ask { + Account(Box), + DeleteAccount(String), + SetActiveAccount { + agent_id: String, + account_id: String, + }, + DeleteTask(String), +} + +impl Ask { + fn apply(self, store: &Store) -> anyhow::Result<()> { + match self { + Ask::Account(account) => store.upsert_account(&account), + Ask::DeleteAccount(id) => store.delete_account(&id), + Ask::SetActiveAccount { + agent_id, + account_id, + } => store.set_active_account(&agent_id, &account_id), + Ask::DeleteTask(id) => store.delete_task(&id), + } + } +} + +/// Reply channel for a write whose outcome the caller waits on. +type AskReply = oneshot::Sender>; + +enum Msg { + Write(Write), + Ask(Ask, AskReply), + /// Applied after everything queued ahead of it. Used by shutdown and by + /// tests that read the database back. + Flush(oneshot::Sender<()>), +} + +/// Handle to the persistence thread. Cloneable and cheap. +#[derive(Clone)] +pub struct Persist { + tx: Option>, +} + +impl Persist { + /// Start the persistence thread over `store`, returning the handle and the + /// shared store for the reads that still run on the actor (ADR 0002 moves + /// those to an in-memory projection next). + /// + /// Without a store — the database failed to open — every write is dropped + /// and the daemon runs in memory, which is what it did before. + pub fn spawn(store: Option) -> (Self, Option>>) { + let Some(store) = store else { + return (Self { tx: None }, None); + }; + let store = Arc::new(Mutex::new(store)); + let (tx, rx) = mpsc::unbounded_channel(); + let worker_store = Arc::clone(&store); + // A dedicated OS thread, not spawn_blocking: this runs for the life of + // the daemon and must never occupy a pooled blocking slot. + std::thread::Builder::new() + .name("warpforge-persist".into()) + .spawn(move || run(rx, &worker_store)) + .expect("spawning the persistence thread"); + (Self { tx: Some(tx) }, Some(store)) + } + + /// Queue a write. Dropped silently when there is no database, matching the + /// previous `if let Some(store)` behaviour at every call site. + pub fn write(&self, write: Write) { + if let Some(tx) = &self.tx { + let _ = tx.send(Msg::Write(write)); + } + } + + pub fn task(&self, task: &Task) { + self.write(Write::Task(Box::new(task.clone()))); + } + + pub fn session_update(&self, task_id: &str, update: &wire::SessionUpdate) { + self.write(Write::SessionUpdate { + task_id: task_id.to_string(), + update: Box::new(update.clone()), + }); + } + + pub fn workflow_run(&self, task_id: &str, json: String) { + self.write(Write::WorkflowRun { + task_id: task_id.to_string(), + json, + }); + } + + /// Apply a write and wait for its outcome. Waits behind whatever is already + /// queued, which is bounded by the drain loop below. + pub async fn ask(&self, ask: Ask) -> Result<(), String> { + let Some(tx) = &self.tx else { + return Ok(()); + }; + let (reply_tx, reply_rx) = oneshot::channel(); + if tx.send(Msg::Ask(ask, reply_tx)).is_err() { + return Err("persistence thread is gone".into()); + } + reply_rx + .await + .unwrap_or_else(|_| Err("persistence thread dropped the reply".into())) + } + + /// Wait until everything queued so far has been committed. + /// + /// Shutdown must await this: with writes in flight on another thread, a + /// daemon that exits without flushing loses the tail of every transcript. + pub async fn flush(&self) { + let Some(tx) = &self.tx else { + return; + }; + let (reply_tx, reply_rx) = oneshot::channel(); + if tx.send(Msg::Flush(reply_tx)).is_ok() { + let _ = reply_rx.await; + } + } +} + +/// Run a read against the store on the blocking pool. +/// +/// SQLite is blocking and the store is behind a plain mutex, so doing this +/// inline in a spawned task occupies a runtime worker for the length of the +/// query — and blocks outright whenever the persistence thread happens to hold +/// the lock for a batch commit. `None` when there is no database. +pub async fn read( + store: Option>>, + read: impl FnOnce(&Store) -> T + Send + 'static, +) -> Option { + let store = store?; + tokio::task::spawn_blocking(move || { + // Recover a poisoned lock rather than cascade the panic, as the + // persistence thread does. + let guard = store.lock().unwrap_or_else(|e| e.into_inner()); + read(&guard) + }) + .await + .ok() +} + +/// Drain the queue into transactions until every sender is gone. +fn run(mut rx: mpsc::UnboundedReceiver, store: &Arc>) { + // Replies are sent after the batch commits, never inside it, so a caller + // that hears "ok" knows the row is durable. + let mut replies: Vec> = Vec::new(); + let mut asked: Vec<(AskReply, Result<(), String>)> = Vec::new(); + + while let Some(first) = rx.blocking_recv() { + let mut batch = vec![first]; + while batch.len() < MAX_BATCH { + match rx.try_recv() { + Ok(msg) => batch.push(msg), + Err(_) => break, + } + } + + { + // Recover a poisoned mutex rather than cascade the panic: losing + // persistence is bad, taking the daemon down with it is worse. + let store = store.lock().unwrap_or_else(|e| e.into_inner()); + let result = store.write_batch(|store| { + for msg in batch.drain(..) { + match msg { + Msg::Write(write) => { + if let Err(error) = write.apply(store) { + eprintln!("[persist] write failed: {error}"); + } + } + Msg::Ask(ask, reply) => { + let outcome = ask.apply(store).map_err(|e| e.to_string()); + asked.push((reply, outcome)); + } + Msg::Flush(reply) => replies.push(reply), + } + } + Ok(()) + }); + if let Err(error) = result { + eprintln!("[persist] batch failed to commit: {error}"); + } + } + + for (reply, outcome) in asked.drain(..) { + let _ = reply.send(outcome); + } + for reply in replies.drain(..) { + let _ = reply.send(()); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::daemon::task::Task; + + fn temp_store() -> (tempfile::TempDir, std::path::PathBuf) { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("test.db"); + (dir, path) + } + + #[tokio::test] + async fn flush_makes_queued_writes_durable() { + let (_dir, path) = temp_store(); + let (persist, _store) = Persist::spawn(Store::open_at(&path).ok()); + + let task = Task::new("demo", "prompt", "agent", vec![]); + persist.task(&task); + for i in 0..50 { + persist.session_update( + &task.id, + &wire::SessionUpdate::AgentText { + text: format!("chunk {i}"), + }, + ); + } + persist.flush().await; + + let reader = Store::open_at(&path).unwrap(); + assert_eq!(reader.load_tasks().unwrap().len(), 1); + assert_eq!(reader.load_session_updates(&task.id).unwrap().len(), 50); + } + + /// The resume replay guard compares persisted updates one by one against + /// the agent's replay, so a batch must not merge adjacent text chunks. + #[tokio::test] + async fn batching_preserves_row_identity_and_order() { + let (_dir, path) = temp_store(); + let (persist, _store) = Persist::spawn(Store::open_at(&path).ok()); + + let task = Task::new("demo", "prompt", "agent", vec![]); + persist.task(&task); + let chunks = ["one ", "two ", "three"]; + for text in chunks { + persist.session_update( + &task.id, + &wire::SessionUpdate::AgentText { + text: text.to_string(), + }, + ); + } + persist.flush().await; + + let reader = Store::open_at(&path).unwrap(); + let stored = reader.load_session_updates(&task.id).unwrap(); + let texts: Vec = stored + .into_iter() + .filter_map(|u| match u { + wire::SessionUpdate::AgentText { text } => Some(text), + _ => None, + }) + .collect(); + assert_eq!(texts, chunks, "chunks must stay separate rows, in order"); + } + + #[tokio::test] + async fn ask_reports_its_outcome() { + let (_dir, path) = temp_store(); + let (persist, _store) = Persist::spawn(Store::open_at(&path).ok()); + + let account = StoredAccount { + id: "acct-1".into(), + agent_id: "claude".into(), + label: "work".into(), + email: None, + plan: None, + home_dir: "/tmp/vault".into(), + created_at: 0, + active: true, + }; + persist + .ask(Ask::Account(Box::new(account))) + .await + .expect("account write should report success"); + + let reader = Store::open_at(&path).unwrap(); + assert_eq!(reader.load_accounts().unwrap().len(), 1); + } + + /// Without a database the daemon still runs; writes are dropped rather than + /// failing, which is what the actor's old `if let Some(store)` guards did. + #[tokio::test] + async fn no_store_drops_writes_without_blocking() { + let (persist, store) = Persist::spawn(None); + assert!(store.is_none()); + persist.task(&Task::new("demo", "prompt", "agent", vec![])); + persist.flush().await; + assert!(persist.ask(Ask::DeleteAccount("nope".into())).await.is_ok()); + } +} diff --git a/src/daemon/server.rs b/src/daemon/server.rs index 645f4b6..1d8c18d 100644 --- a/src/daemon/server.rs +++ b/src/daemon/server.rs @@ -20,7 +20,7 @@ use serde_json::json; use tokio::net::{TcpListener, TcpStream}; use tokio::sync::broadcast; use tokio::sync::oneshot; -use tokio::sync::{Notify, RwLock}; +use tokio::sync::{mpsc, Notify, RwLock, Semaphore}; use tokio_tungstenite::tungstenite::Message; use uuid::Uuid; use warpforge_protocol as wire; @@ -28,6 +28,15 @@ use warpforge_protocol as wire; use super::actor::{Command, DaemonHandle}; use super::wire as wireconv; +/// Outgoing frames buffered per connection before the read loop slows down. +const OUTGOING_QUEUE: usize = 256; + +/// Requests one connection may have in flight at once. Concurrent requests +/// answer off the read loop, so without a cap a client could fan out unbounded +/// git, filesystem and subprocess work by sending faster than the daemon +/// completes it. +const MAX_CONCURRENT_REQUESTS: usize = 8; + fn daemon_json_path() -> PathBuf { dirs::home_dir() .unwrap_or_else(|| PathBuf::from(".")) @@ -162,6 +171,10 @@ async fn run_controlled( accepted = listener.accept() => accepted?, _ = lifecycle.shutdown.notified() => return Ok(()), }; + // Replies are small frames. Left to Nagle they wait on an ACK for the + // previous one, which pairs with the peer's delayed ACK to add tens of + // milliseconds to an otherwise instant answer. + let _ = stream.set_nodelay(true); let handle = handle.clone(); let token = token.clone(); let lifecycle = Arc::clone(&lifecycle); @@ -180,15 +193,29 @@ async fn handle_conn( lifecycle: Arc, ) -> Result<()> { let ws = tokio_tungstenite::accept_async(stream).await?; - let (mut tx, mut rx) = ws.split(); + let (mut sink, mut rx) = ws.split(); let mut events = handle.subscribe(); let mut authed = token.is_empty(); let mut subscribed = false; + // One writer owns the socket so read requests answered off the read loop + // have somewhere to reply to. Bounded: a client that stops draining slows + // its own connection rather than growing the queue without limit. + let (out_tx, mut out_rx) = mpsc::channel::(OUTGOING_QUEUE); + tokio::spawn(async move { + while let Some(msg) = out_rx.recv().await { + if sink.send(msg).await.is_err() { + break; + } + } + }); + // Caps the work one client can have in flight at once. + let request_slots = Arc::new(Semaphore::new(MAX_CONCURRENT_REQUESTS)); + macro_rules! send { ($msg:expr) => {{ let text = serde_json::to_string(&$msg)?; - if tx.send(Message::Text(text)).await.is_err() { + if out_tx.send(Message::Text(text)).await.is_err() { break; } }}; @@ -203,7 +230,7 @@ async fn handle_conn( }; let text = match msg { Message::Text(t) => t.as_str().to_string(), - Message::Ping(p) => { let _ = tx.send(Message::Pong(p)).await; continue; } + Message::Ping(p) => { let _ = out_tx.send(Message::Pong(p)).await; continue; } Message::Close(_) => break, _ => continue, }; @@ -217,7 +244,7 @@ async fn handle_conn( if ok { authed = true; } else { - let _ = tx.send(Message::Close(None)).await; + let _ = out_tx.send(Message::Close(None)).await; break; } continue; @@ -256,6 +283,45 @@ async fn handle_conn( continue; } + // Independent requests answer without holding up the next one. + // Until now a connection served one request at a time, so a + // tool approval was not even read off the socket while a title + // was being generated ahead of it (ADR 0002). + if method_runs_concurrently(&req.method) { + let gated = method_is_mutation(&req.method); + let handle = handle.clone(); + let lifecycle = Arc::clone(&lifecycle); + let out = out_tx.clone(); + let slots = Arc::clone(&request_slots); + tokio::spawn(async move { + let _permit = slots.acquire_owned().await; + // The same update gate the serial path applies, kept + // here so moving a method between the two lists cannot + // quietly let it run during a daemon handover. + let result = if gated { + let _guard = lifecycle.mutations.read().await; + if lifecycle.quiescing.load(Ordering::Acquire) { + Err(wire::RpcError { + code: wire::ErrorCode::Updating, + message: "daemon is quiescing for an application update".into(), + }) + } else { + dispatch(&handle, req.method, &lifecycle).await + } + } else { + dispatch(&handle, req.method, &lifecycle).await + }; + let message = match result { + Ok(result) => wire::ServerMessage::Response { id, result }, + Err(error) => wire::ServerMessage::Error { id, error }, + }; + if let Ok(text) = serde_json::to_string(&message) { + let _ = out.send(Message::Text(text)).await; + } + }); + continue; + } + let is_handoff = matches!(&req.method, wire::Method::UpdatePrepareShutdown { .. }); let result = if method_is_mutation(&req.method) && !is_handoff { let _guard = lifecycle.mutations.read().await; @@ -278,7 +344,7 @@ async fn handle_conn( Err(error) => wire::ServerMessage::Error { id, error }, }; let text = serde_json::to_string(&message)?; - let sent = tx.send(Message::Text(text)).await.is_ok(); + let sent = out_tx.send(Message::Text(text)).await.is_ok(); if handoff_ready { // Queue the acknowledgement on the socket before stopping @@ -1373,6 +1439,50 @@ fn accounts_result( } } +/// Requests answered off the connection's read loop instead of ahead of +/// everything behind them. +/// +/// The question is not whether a request writes — it is whether it has to be +/// ordered against the others on the connection. So this is its own list rather +/// than the negation of [`method_is_mutation`], which exists to decide what the +/// update gate holds back: +/// +/// - Reads are independent by definition. +/// - So are one-shot jobs whose result depends on nothing else in flight: +/// generating a title, installing an agent or a language server. These are +/// the slowest things the daemon does — a title spawns an agent process with +/// a two-minute ceiling, an install shells out to a package manager — and +/// they are what made starting a task feel like it stalled everything else. +/// - Ordered work stays serial. LSP is a streaming protocol, so dispatching +/// `LspSend` concurrently would reorder a language server's inbox, and git +/// writes mean what they mean only in sequence: commit, then push. +fn method_runs_concurrently(method: &wire::Method) -> bool { + use wire::Method::*; + matches!( + method, + TextGenerate { .. } + | AgentsInstall { .. } + | LanguageServersInstall { .. } + | DiffGet { .. } + | FileContents { .. } + | FileList { .. } + | FileSearch { .. } + | GitBranches { .. } + | GitPushInfo { .. } + | ServiceLogs { .. } + | PortForwardLogs { .. } + | TaskListWorktrees { .. } + | SessionsList { .. } + | OrchestratorListAgents { .. } + | AgentsDetect {} + | AccountsList {} + | OrchestrateList {} + | OrchestrateGetConfig {} + | WorkflowList { .. } + | LanguageServersDetect {} + ) +} + fn method_is_mutation(method: &wire::Method) -> bool { use wire::Method::*; !matches!( @@ -1455,6 +1565,131 @@ mod tests { } } + /// Generating a title spawns an agent process and can run for minutes. It + /// used to be dispatched on the read loop, so the daemon read nothing else + /// from that client meanwhile — which is what made starting a task appear + /// to stall the conversation it was starting. + #[test] + fn the_slowest_requests_do_not_block_the_connection() { + use wire::Method::*; + for method in [ + TextGenerate { + task_id: "t".into(), + agent_id: "claude".into(), + kind: wire::TextGenKind::TaskTitle, + model: None, + }, + AgentsInstall { + id: "claude".into(), + }, + LanguageServersInstall { id: "rust".into() }, + ] { + assert!( + method_runs_concurrently(&method), + "{method:?} shells out for seconds to minutes and must not hold the read loop" + ); + } + } + + /// Ordered work must stay on the serial path. LSP is a streaming protocol + /// and git writes only mean what they mean in sequence. + #[test] + fn ordered_requests_stay_serial() { + use wire::Method::*; + for method in [ + LspSend { + server_id: "s".into(), + payload: serde_json::Value::Null, + }, + GitCommit { + task_id: "t".into(), + message: "m".into(), + files: None, + amend: false, + }, + ] { + assert!( + !method_runs_concurrently(&method), + "{method:?} depends on arriving in order" + ); + } + } + + /// A read must not hold up whatever is queued behind it. One connection + /// used to serve one request at a time, so a slow read delayed everything + /// after it — a tool approval was not even read off the socket until the + /// read ahead of it finished. The cheap request sent second must come back + /// first. + // Multi-threaded on purpose: the daemon runs on a multi-thread runtime, and + // on the single-threaded test default a synchronous filesystem walk inside + // a spawned task blocks the very socket read this is measuring. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn a_slow_read_does_not_delay_the_request_behind_it() { + // A project big enough that listing it takes real time, and sized here + // rather than inherited from the checkout so the margin is the same + // everywhere this runs. + let dir = tempfile::tempdir().unwrap(); + for i in 0..4000 { + std::fs::write(dir.path().join(format!("file{i}.txt")), "x").unwrap(); + } + let projects = vec![ProjectEntry { + name: "demo".into(), + path: dir.path().to_string_lossy().into_owned(), + added_at: "0".into(), + }]; + let store = Store::open_at(std::path::Path::new(":memory:")).ok(); + let handle = Daemon::spawn(projects, store); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(run(listener, handle.clone(), String::new())); + // Without this the second request sits in the client's send buffer + // waiting on an ACK for the first, and the test measures Nagle rather + // than the daemon. + let tcp = TcpStream::connect(addr).await.unwrap(); + tcp.set_nodelay(true).unwrap(); + let (mut ws, _) = tokio_tungstenite::client_async(format!("ws://{addr}"), tcp) + .await + .unwrap(); + + // Listing walks and stats every file; the accounts list is answered + // from memory. + ws.send(Message::Text( + json!({ + "id": 1, + "method": "file.list", + "params": { "project": "demo", "include_ignored": true } + }) + .to_string(), + )) + .await + .unwrap(); + ws.send(Message::Text( + json!({ "id": 2, "method": "accounts.list", "params": {} }).to_string(), + )) + .await + .unwrap(); + + // Both are answered; the order is the point. + let mut ids = Vec::new(); + for _ in 0..2 { + let msg = timeout(Duration::from_secs(10), ws.next()) + .await + .expect("a reply, not silence") + .expect("some") + .expect("ok"); + let Message::Text(text) = msg else { + panic!("expected a text frame") + }; + let reply: serde_json::Value = serde_json::from_str(text.as_str()).unwrap(); + ids.push(reply["id"].as_u64()); + } + assert_eq!( + ids, + vec![Some(2), Some(1)], + "the cheap request must not wait behind the project listing" + ); + } + #[tokio::test] async fn subscribe_then_create_task_over_websocket() { // Daemon with one project, in-memory store. diff --git a/src/daemon/store.rs b/src/daemon/store.rs index f35ce86..490212a 100644 --- a/src/daemon/store.rs +++ b/src/daemon/store.rs @@ -264,6 +264,26 @@ impl Store { Ok(Self { conn }) } + /// Run `writes` as one transaction. The persistence actor (see + /// `daemon/runtime/persist.rs`) drains its queue through here so a burst of + /// streamed session updates costs one commit instead of one per row. + /// + /// On any error the whole batch is rolled back — a half-applied batch would + /// leave a task row without the session updates that explain it. + pub fn write_batch(&self, writes: impl FnOnce(&Self) -> Result<()>) -> Result<()> { + self.conn.execute_batch("BEGIN")?; + match writes(self) { + Ok(()) => { + self.conn.execute_batch("COMMIT")?; + Ok(()) + } + Err(error) => { + let _ = self.conn.execute_batch("ROLLBACK"); + Err(error) + } + } + } + pub fn upsert_task(&self, task: &Task) -> Result<()> { let tags = serde_json::to_string(&task.tags)?; let config_options = serde_json::to_string(&task.config_options)?; @@ -603,10 +623,54 @@ impl Store { } } + /// First-seen timestamps of every streamed tool call, keyed by + /// `(task_id, tool_call_id)`. Read once at daemon startup: the actor keeps + /// only this small map rather than the full transcripts it was derived from. + pub fn load_tool_call_starts(&self) -> Result> { + let mut stmt = self + .conn + .prepare("SELECT task_id, update_json FROM session_updates ORDER BY id")?; + let mut map = HashMap::new(); + let rows = stmt.query_map([], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + })?; + for row in rows.filter_map(|r| r.ok()) { + if let Ok(wire::SessionUpdate::ToolCall { + tool_call_id, + started_at: Some(started_at), + .. + }) = serde_json::from_str::(&row.1) + { + // First-seen wins. Later frames of the same tool call repeat the + // timestamp the daemon assigned, but a daemon restart mid-call + // can assign a new one — and this map exists precisely so a + // call's start time does not move under the user. + map.entry((row.0, tool_call_id)).or_insert(started_at); + } + } + Ok(map) + } + /// Load persisted histories as semantic rows. Raw ACP text chunks and /// repeated tool lifecycle frames remain in SQLite for replay fidelity but /// are folded before building the desktop snapshot. pub fn load_all_session_updates(&self) -> Result>> { + let mut map = self.load_all_session_updates_raw()?; + for updates in map.values_mut() { + *updates = fold_for_snapshot(updates); + } + Ok(map) + } + + /// Every persisted update, per task, exactly as written. + /// + /// Unlike [`Store::load_all_session_updates`] nothing is folded or trimmed: + /// the resume replay guard matches the agent's replay against this history + /// chunk for chunk, so a folded `AgentText` would never compare equal and + /// the whole turn would be emitted twice. + pub fn load_all_session_updates_raw( + &self, + ) -> Result>> { let mut stmt = self .conn .prepare("SELECT task_id, update_json FROM session_updates ORDER BY id")?; @@ -616,24 +680,32 @@ impl Store { })?; for row in rows.filter_map(|r| r.ok()) { if let Ok(update) = serde_json::from_str::(&row.1) { - let output = map.entry(row.0).or_default(); - append_snapshot_update(output, update); - if output.len() > MAX_SESSION_SNAPSHOT_UPDATES + SNAPSHOT_TRIM_HEADROOM { - let overflow = output.len() - MAX_SESSION_SNAPSHOT_UPDATES; - output.drain(..overflow); - } - } - } - for updates in map.values_mut() { - if updates.len() > MAX_SESSION_SNAPSHOT_UPDATES { - let overflow = updates.len() - MAX_SESSION_SNAPSHOT_UPDATES; - updates.drain(..overflow); + map.entry(row.0).or_default().push(update); } } Ok(map) } } +/// Fold a raw history into the shape the desktop snapshot wants: streamed text +/// concatenated, repeated tool frames collapsed, oldest entries dropped past +/// the cap. +pub fn fold_for_snapshot(updates: &[wire::SessionUpdate]) -> Vec { + let mut output: Vec = Vec::new(); + for update in updates { + append_snapshot_update(&mut output, update.clone()); + if output.len() > MAX_SESSION_SNAPSHOT_UPDATES + SNAPSHOT_TRIM_HEADROOM { + let overflow = output.len() - MAX_SESSION_SNAPSHOT_UPDATES; + output.drain(..overflow); + } + } + if output.len() > MAX_SESSION_SNAPSHOT_UPDATES { + let overflow = output.len() - MAX_SESSION_SNAPSHOT_UPDATES; + output.drain(..overflow); + } + output +} + /// `"idle"` and `"needs_review"` are the pre-merge spellings of `Waiting`. Rows /// written by older daemons are still on disk in every existing install, so both /// must keep loading — this arm is load-bearing, not tidy-up. diff --git a/src/daemon/worktree.rs b/src/daemon/worktree.rs index 4baca7f..bf757a1 100644 --- a/src/daemon/worktree.rs +++ b/src/daemon/worktree.rs @@ -37,50 +37,38 @@ impl WorktreeManager { /// Create a worktree for `task_id`. If `base_branch` is provided, branch /// from that; otherwise branch from the current HEAD. pub async fn create(&mut self, task_id: &str, base_branch: Option<&str>) -> Result { - let wt_dir = self.base_repo.join(".worktrees").join(task_id); - let branch = format!("warpforge/task/{task_id}"); - - // Resolve the base branch. - let base = match base_branch { - Some(b) => b.to_string(), - None => { - let output = tokio::process::Command::new("git") - .args(["rev-parse", "--abbrev-ref", "HEAD"]) - .current_dir(&self.base_repo) - .output() - .await - .context("failed to run git rev-parse")?; - String::from_utf8_lossy(&output.stdout).trim().to_string() - } - }; + let wt = create_detached(&self.base_repo, task_id, base_branch).await?; + self.adopt(wt.clone()); + Ok(wt) + } - // Create the worktree + branch. - let status = tokio::process::Command::new("git") - .args([ - "worktree", - "add", - "-b", - &branch, - wt_dir.to_str().unwrap_or(".worktrees/task"), - &base, - ]) - .current_dir(&self.base_repo) - .status() - .await - .context("failed to run git worktree add")?; + /// The repo this manager tracks worktrees for. + pub fn base_repo(&self) -> &Path { + &self.base_repo + } - if !status.success() { - anyhow::bail!("git worktree add failed (exit {status})"); - } + /// Drop a worktree from the map without touching git — for a removal that + /// already ran off the actor (see [`remove_detached`]). + pub fn forget(&mut self, task_id: &str) -> Option { + self.worktrees.remove(task_id) + } - let wt = Worktree { - task_id: task_id.to_string(), - path: wt_dir, - branch, - base_branch: base, - }; - self.worktrees.insert(task_id.to_string(), wt.clone()); - Ok(wt) + /// This task's worktree metadata, if the manager tracks one. + pub fn get(&self, task_id: &str) -> Option<&Worktree> { + self.worktrees.get(task_id) + } + + /// Record a worktree created outside the manager (see [`create_detached`]). + pub fn adopt(&mut self, wt: Worktree) { + self.worktrees.insert(wt.task_id.clone(), wt); + } + + /// The branch and path a branched worktree would inherit from, if the + /// source task has a worktree here. + pub fn source_state(&self, source_task_id: &str) -> Option<(String, PathBuf)> { + self.worktrees + .get(source_task_id) + .map(|wt| (wt.branch.clone(), wt.path.clone())) } /// Create a worktree for `task_id` that inherits the state of a source @@ -267,6 +255,147 @@ impl WorktreeManager { } } +/// Create a worktree without a manager, so the git work can run off the daemon +/// actor. The caller records the result with [`WorktreeManager::adopt`]. +/// +/// `git worktree add` takes long enough to be felt: run inside a command +/// handler it delayed every other task's messages and approvals until the new +/// task's checkout finished (ADR 0002). +pub async fn create_detached( + base_repo: &Path, + task_id: &str, + base_branch: Option<&str>, +) -> Result { + let wt_dir = base_repo.join(".worktrees").join(task_id); + let branch = format!("warpforge/task/{task_id}"); + + let base = match base_branch { + Some(b) => b.to_string(), + None => { + let output = tokio::process::Command::new("git") + .args(["rev-parse", "--abbrev-ref", "HEAD"]) + .current_dir(base_repo) + .output() + .await + .context("failed to run git rev-parse")?; + String::from_utf8_lossy(&output.stdout).trim().to_string() + } + }; + + let status = tokio::process::Command::new("git") + .args([ + "worktree", + "add", + "-b", + &branch, + wt_dir.to_str().unwrap_or(".worktrees/task"), + &base, + ]) + .current_dir(base_repo) + .status() + .await + .context("failed to run git worktree add")?; + + if !status.success() { + anyhow::bail!("git worktree add failed (exit {status})"); + } + + Ok(Worktree { + task_id: task_id.to_string(), + path: wt_dir, + branch, + base_branch: base, + }) +} + +/// [`create_detached`] for a conversation branch: branch from `base_branch` +/// (the current HEAD when `None`) and carry over the uncommitted changes in +/// `source_path`. +/// +/// `source_path` is wherever the source task actually works, which is its own +/// worktree only when it has one — a task running in the project checkout +/// branches from there. Getting this wrong is silent: the branch comes up on a +/// clean HEAD and the work it was meant to continue is simply absent. +pub async fn create_branched_detached( + base_repo: &Path, + task_id: &str, + base_branch: Option<&str>, + source_path: &Path, +) -> Result { + let wt = create_detached(base_repo, task_id, base_branch).await?; + copy_working_state(source_path, &wt.path) + .await + .with_context(|| { + format!("failed to copy working state into branched worktree {task_id}") + })?; + Ok(wt) +} + +/// Merge `branch` into `base_branch` without a manager, so the git work can run +/// off the daemon actor. The caller records the outcome. +pub async fn merge_detached( + base_repo: &Path, + branch: &str, + base_branch: &str, +) -> Result { + let status = tokio::process::Command::new("git") + .args(["checkout", base_branch]) + .current_dir(base_repo) + .status() + .await + .context("failed to checkout base branch")?; + if !status.success() { + return Ok(MergeResult::Error("failed to checkout base branch".into())); + } + + let output = tokio::process::Command::new("git") + .args(["merge", branch, "--no-edit"]) + .current_dir(base_repo) + .output() + .await + .context("failed to run git merge")?; + + if output.status.success() { + return Ok(MergeResult::Ok { + branch: branch.to_string(), + }); + } + let stderr = String::from_utf8_lossy(&output.stderr).to_string(); + if stderr.contains("CONFLICT") || stderr.contains("conflict") { + // Abort the failed merge. + let _ = tokio::process::Command::new("git") + .args(["merge", "--abort"]) + .current_dir(base_repo) + .status() + .await; + return Ok(MergeResult::Conflict { + message: stderr, + branch: branch.to_string(), + }); + } + Ok(MergeResult::Error(stderr)) +} + +/// Remove a worktree and delete its branch, without a manager. The caller drops +/// it from the map with [`WorktreeManager::forget`]. +pub async fn remove_detached(base_repo: &Path, path: &Path, branch: &str) -> Result<()> { + let status = tokio::process::Command::new("git") + .args(["worktree", "remove", "--force", path.to_str().unwrap_or("")]) + .current_dir(base_repo) + .status() + .await + .context("failed to run git worktree remove")?; + if !status.success() { + anyhow::bail!("git worktree remove failed (exit {status})"); + } + let _ = tokio::process::Command::new("git") + .args(["branch", "-D", branch]) + .current_dir(base_repo) + .status() + .await; + Ok(()) +} + /// Copy the uncommitted working-tree state of `source` into `target` so a /// branched worktree starts from the exact files the source left behind. /// Handles tracked modifications/deletions (via a binary diff applied with @@ -324,6 +453,15 @@ async fn copy_working_state(source: &Path, target: &Path) -> Result<()> { continue; } let src = source.join(line); + // `git ls-files --others` reports a nested checkout as one directory + // entry rather than its contents, and every worktree lives inside the + // project at `.worktrees/`. So when the source is the project + // checkout itself, its own worktrees show up here — copying one would + // fail outright, and copying it successfully would be worse. Nothing + // that is not a plain file belongs in a branch's starting state. + if !src.is_file() { + continue; + } let dst = target.join(line); if let Some(parent) = dst.parent() { tokio::fs::create_dir_all(parent) diff --git a/src/ports.rs b/src/ports.rs index 4ac26c3..fcb1963 100644 --- a/src/ports.rs +++ b/src/ports.rs @@ -48,6 +48,26 @@ pub fn allocate( )) } +/// Ports this process handed out that fall inside `ranges`. +/// +/// Teardown sweeps use this instead of the whole range: a range holds whatever +/// happens to be listening, including a developer's unrelated server, and +/// warpforge has no business killing a process it did not start. +pub fn allocated_in_ranges(ranges: &[(u16, u16)]) -> Vec { + let map = alloc_map().lock().unwrap(); + let mut ports: Vec = map + .keys() + .copied() + .filter(|port| { + ranges + .iter() + .any(|&(start, end)| *port >= start && *port <= end) + }) + .collect(); + ports.sort_unstable(); + ports +} + /// Release the port allocated for a service. pub fn release(project_name: &str, service_name: &str) { let key = format!("{project_name}/{service_name}"); diff --git a/src/service.rs b/src/service.rs index ef82858..df0ca2d 100644 --- a/src/service.rs +++ b/src/service.rs @@ -149,6 +149,30 @@ pub async fn kill_listeners_in_ranges(ranges: &[(u16, u16)]) { let _ = ranges; } +/// Kill whatever listens on exactly these ports. +/// +/// Unlike [`kill_listeners_in_ranges`] this touches only ports the caller knows +/// warpforge allocated, so it can never reach a process warpforge did not +/// start. +pub async fn kill_listeners_on_ports(ports: &[u16]) { + #[cfg(unix)] + { + if ports.is_empty() { + return; + } + for &port in ports { + kill_listeners_in_range(port, port, "TERM").await; + } + sleep(Duration::from_millis(600)).await; + for &port in ports { + kill_listeners_in_range(port, port, "KILL").await; + } + } + + #[cfg(not(unix))] + let _ = ports; +} + #[cfg(unix)] async fn kill_listeners_in_range(start: u16, end: u16, signal: &str) { let spec = format!("-iTCP:{start}-{end}"); diff --git a/tests/fixtures/mock-acp-workflow.mjs b/tests/fixtures/mock-acp-workflow.mjs index a5f7645..9d9f65e 100644 --- a/tests/fixtures/mock-acp-workflow.mjs +++ b/tests/fixtures/mock-acp-workflow.mjs @@ -98,6 +98,11 @@ function handle(msg) { endTurn(msg.id); setTimeout(() => process.exit(0), 100); break; + case "die": + // Exit mid-turn without ever ending it: the agent process is lost + // before the stage produces any verdict at all. + process.exit(1); + break; case "slow-fix": text("FIX-DONE: addressed the findings (slowly)."); setTimeout(() => endTurn(msg.id), 600);