From 62cdc849819878c0106aa09f3667d351b3dd51c5 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 23 Sep 2026 16:14:29 +0200 Subject: [PATCH 1/2] Find a sessionless driver's flow from a local run ledger (#1123) The ledger entry is written before the start's poll lane, so it survives a harness abort: the server ignores MCP cancellations and still holds the id. Co-Authored-By: Claude Opus 5.5 --- README.md | 4 +- docs/CHANGES.md | 13 ++ kcap/skills/agent-flows/SKILL.md | 6 +- kcap/skills/review-flows/SKILL.md | 6 +- src/Capacitor.Cli/Commands/FlowRunLedger.cs | 81 ++++++++++ src/Capacitor.Cli/Commands/McpFlowsServer.cs | 143 ++++++++++++----- src/Capacitor.Cli/Commands/McpSessionId.cs | 9 +- .../Commands/FlowRunLedgerTests.cs | 83 ++++++++++ .../Commands/StatusWithoutFlowRunIdTests.cs | 150 +++++++++++++++++- 9 files changed, 440 insertions(+), 55 deletions(-) create mode 100644 src/Capacitor.Cli/Commands/FlowRunLedger.cs create mode 100644 test/Capacitor.Cli.Tests.Unit/Commands/FlowRunLedgerTests.cs diff --git a/README.md b/README.md index f865b2b94..fc1d2b9f3 100644 --- a/README.md +++ b/README.md @@ -574,7 +574,7 @@ It provides four generic tools: - **`start_flow`** — start a new flow from either `definition_id` (the server's flow-definition catalog, e.g. `spec-review`, `code-review`, or a custom definition) **or** `definition_yaml` (an inline dynamic flow definition — the full YAML, same schema as catalog definitions); provide exactly one of the two. Catalog starts use the guarded v2 protocol. A `definition_yaml` flow has extra constraints: every participant must declare `workspace: none` and a concrete, priced model (no `default`), the server clamps `limits`/`mcp` to its own caps rather than trusting the definition, may reject the whole thing with a coded error, and requires a server with dynamic flows enabled. Also provide `target_kind`, `target_ref`, `target_title`, and `context`. Requester context (session ID, cwd, repo root, owner, name) is resolved automatically from the environment. Returns a `flow_run_id`. A single-participant definition starts its first round eagerly; a multi-participant definition starts **round-less** — the response carries no round, and each declared role's agent launches lazily on its first `send_to_participant`. Optional `mode`: by default, when the daemon runs on the same machine and the selected vendor declares a borrowed-context containment strategy, the participant sees current tracked, dirty, and non-ignored untracked checkout content; the safety boundary is vendor-specific. Cursor runs in a daemon-owned snapshot refreshed before follow-up rounds, while direct borrowing is reserved for runtimes with a native read-only tool clamp. The capability is advertised for whatever build of the vendor CLI is installed — it is not gated on a per-version validation record, since a vendor auto-update would then silently drop the participant back onto a stale committed base. It IS gated per platform where the containment boundary has not been measured: **GitHub Copilot advertises borrowed review only on macOS/ARM64, and only on a daemon that can both enforce an OS sandbox (`sandbox-exec`) and broker a token** (`COPILOT_GITHUB_TOKEN`/`GH_TOKEN`/`GITHUB_TOKEN`) — widening its tool surface enough to read a snapshot also widens what a read tool can be pointed at, so the boundary is an OS sandbox rather than the vendor's own permission prompts, and that sandbox grants neither your keychain nor your Copilot state (see [Borrowed-context Copilot reviews](#borrowed-context-copilot-reviews)). Where any of the three is missing, a Copilot borrowed request returns `vendor_containment_unreadable` naming the daemon and the remedies; `mode="context-only"` works and is the remedy. Pass `mode="context-only"` to opt out. Claude review flows currently use owned worktrees because it declares no borrowed-review containment strategy. Optional `vendor`: for reserved `spec-review`/`code-review`, explicitly selects the reviewer independently of the driver; when omitted, resolution falls through to the definition's authored vendor when it declares one, then to your saved `flows.reviewer_vendor` preference (applied via one automatic retry, with the response saying so), and finally a coded `reviewer_vendor_required` if neither is set — ask the user which reviewer vendor to use, pass it explicitly, and offer to save it with `kcap config set flows.reviewer_vendor `. Custom single-participant catalog definitions retain their authored vendor unless explicitly overridden. The server records requested/applied vendor plus selection source, and rejects an unavailable or uncertified vendor without silent fallback. Dynamic (`definition_yaml`) flows reject a top-level vendor override because each participant declares its own vendor. Optional `model`: a per-run reviewer **model** override for a single-participant catalog review — REQUIRES `vendor` (the model is interpreted against that vendor; there is no vendor→model table anywhere in the CLI, so `model` without `vendor` is rejected locally before any request is sent) and is rejected on a `definition_yaml` (dynamic) or multi-participant start (each participant already pins its own model). Pass the vendor's own model id or alias verbatim, case-sensitive — the CLI never translates, canonicalizes, or guesses it. - **`send_to_participant`** — send a follow-up message to a participant role declared by the flow definition (single-participant definitions use `"reviewer"`; the server rejects an unknown role, naming the valid ones). One round runs at a time per role — a second send to a busy role is rejected naming the busy round, while other roles remain addressable. Returns the new round's findings. -- **`get_flow_status`** — get the current status (running, waiting, completed, failed) and last result of a flow run. Optional `wait` (boolean, default `false`): when `true`, blocks — via repeated bounded internal checks, never a raw long-poll — until the round is terminal or roughly 3.5 minutes pass, instead of returning the current snapshot immediately. A long-running round is expected, not an error; on that cap this returns the same benign "still running" text an unset/false `wait` already returns on the round-submission path, so simply call it again with `wait: true`. `flow_run_id` is optional on Claude Code and Codex, the harnesses whose session the flows server can identify (a start from any other harness carries no session, so the id must be passed there): omitted, the newest open flow this session started is read — with several open, the reply lists them to choose from; with none open, the newest settled one, so a failure you were not watching for is still readable — and `session_id` selects another session's flows. The lookup needs a server with the by-session flow route; an older server answers with an error that says to pass the id. +- **`get_flow_status`** — get the current status (running, waiting, completed, failed) and last result of a flow run. Optional `wait` (boolean, default `false`): when `true`, blocks — via repeated bounded internal checks, never a raw long-poll — until the round is terminal or roughly 3.5 minutes pass, instead of returning the current snapshot immediately. A long-running round is expected, not an error; on that cap this returns the same benign "still running" text an unset/false `wait` already returns on the round-submission path, so simply call it again with `wait: true`. `flow_run_id` is optional: omitted, the newest open flow this session started is read (Cursor, Copilot, Gemini, Kiro, OpenCode and Antigravity give the flows server no session, so there it reads the flows started from the same repository — or working directory — on this machine, which the flows server records in `flow-runs-v1.json` under the config directory) — with several open, the reply lists them to choose from; with none open, the newest settled one, so a failure you were not watching for is still readable — and `session_id` selects another session's flows. The lookup needs a server with the by-session flow route; an older server answers with an error that says to pass the id. - **`close_flow`** — mark a completed flow run as closed. Every flow response reports which **workspace** the reviewer actually used, so you can tell whether it saw your uncommitted work: @@ -616,7 +616,7 @@ The four review tools — **`start_review_flow`**, **`submit_review_round`**, ** #### Harness tool timeouts -A start or round call holds the tool call open while the round runs. The CLI bounds it to end under 5 minutes — the shortest per-tool timeout among the harnesses that drive flows — with the round result or the benign `Flow still running for flow_run_id …` text, which is also the only place the driver is handed its `flow_run_id`. If a harness aborts the call anyway (a shorter timeout in its config, or a user-edited one), the flow is still running server-side: the driver should not start it again and should not go looking for it, but call `get_review_flow_status` / `get_flow_status` with `wait: true`, omitting `flow_run_id` when the aborted start never delivered it (Claude Code and Codex; the other harnesses give the flows server no session to look it up by). The tool descriptions and both flow skills say so up front, because the harness words its own timeout error and kcap cannot. +A start or round call holds the tool call open while the round runs. The CLI bounds it to end under 5 minutes — the shortest per-tool timeout among the harnesses that drive flows — with the round result or the benign `Flow still running for flow_run_id …` text, which is also the only place the driver is handed its `flow_run_id`. If a harness aborts the call anyway (a shorter timeout in its config, or a user-edited one), the flow is still running server-side: the driver should not start it again and should not go looking for it, but call `get_review_flow_status` / `get_flow_status` with `wait: true`, omitting `flow_run_id` when the aborted start never delivered it. The tool descriptions and both flow skills say so up front, because the harness words its own timeout error and kcap cannot. **Reviewer model override — capability gating, protocol, and semantics:** diff --git a/docs/CHANGES.md b/docs/CHANGES.md index 833d3b151..53ce2493f 100644 --- a/docs/CHANGES.md +++ b/docs/CHANGES.md @@ -6,6 +6,19 @@ diff. `CLAUDE.md` holds the invariants; `docs/superpowers/specs/` holds the full Not release notes. Each entry is written as of the change that produced it and is not revised as the code moves on; where an entry disagrees with the code, the code wins. +## A driver without a session finds its flow from a local run ledger + +The status tools find a flow without its id by the calling session, and Cursor, Copilot, Gemini, +Kiro, OpenCode and Antigravity export no session into the long-lived MCP child, so a driver there +had no way back to a run whose start the harness aborted. The flows server now records each run it +starts without a session in `flow-runs-v1.json` under the config root, keyed on the repository root +(or working directory), and a bare status call without a session reads that workspace's newest +entries and confirms each with `GET /api/flows/{id}`. The record is written before the start's poll +lane begins, which is what makes it survive an abort: the server ignores MCP cancellations, so the +lane holds the id after the harness has given up on the call. Session-bearing harnesses record +nothing and keep the server-side lookup. Several chats in one workspace share the ledger; the same +"several open flows are listed" rule applies, so the worst case is a choice, never a wrong run. + ## The ledger servers are pre-approved, and every tool advertises annotations Codex decides whether an MCP call needs approval from the tool's annotations, and with its automatic diff --git a/kcap/skills/agent-flows/SKILL.md b/kcap/skills/agent-flows/SKILL.md index 4471aaee5..942aa55ae 100644 --- a/kcap/skills/agent-flows/SKILL.md +++ b/kcap/skills/agent-flows/SKILL.md @@ -24,7 +24,7 @@ Use the `kcap mcp flows` MCP tools (`start_flow`, `send_to_participant`, `get_fl `round_timeout` is an **inactivity** bound, not a wall-clock cap: a round only fails for taking too long if the participant goes genuinely quiet for that whole stretch — an actively-working participant can legitimately run a round for a long time. If a status check (or `start_flow`/`send_to_participant`) returns the benign "Flow still running" text, that is an expected outcome on a long round, not a problem — re-enter the wait with `get_flow_status(flow_run_id, wait: true)`, which blocks (via bounded, internally-retried checks — never a raw long-poll) until the round finishes or roughly 3.5 minutes pass, then call it again if it's still running. A round result of **`unclear` now genuinely means the participant went dead or silent** (no activity for the whole inactivity bound, or it crashed/was stopped) — it is no longer a symptom of a merely slow participant; see the `participant_unreachable` and `participant_died`/`participant_stopped` guardrail entries below. -**A harness tool timeout is not a flow failure.** `start_flow` and `send_to_participant` block for minutes while the participant works. If your harness aborts the call with its own tool timeout (on Codex the error reads `timed out awaiting tools/call`), the flow is still running server-side. Do not start a new flow and do not investigate — call `get_flow_status(wait: true)`. Without a `flow_run_id` it reads the newest open flow this session started (pass `session_id` for another session's), so it works even when the aborted start never handed you the id or context compaction dropped it; with several open flows it lists them for you to pick from. That lookup needs a session identity, which the flows server has on Claude Code and Codex only — on any other harness the `flow_run_id` must be passed. +**A harness tool timeout is not a flow failure.** `start_flow` and `send_to_participant` block for minutes while the participant works. If your harness aborts the call with its own tool timeout (on Codex the error reads `timed out awaiting tools/call`), the flow is still running server-side. Do not start a new flow and do not investigate — call `get_flow_status(wait: true)`. Without a `flow_run_id` it reads the newest open flow this session started (pass `session_id` for another session's), so it works even when the aborted start never handed you the id or context compaction dropped it; with several open flows it lists them for you to pick from. On a harness that gives the flows server no session identity, it reads the flows started from this workspace on this machine instead. ## Role-surface safety gate @@ -139,7 +139,7 @@ After applying the role-surface safety gate, if `start_flow` / `send_to_particip ## Core rules -1. **Start exactly one flow per user task.** Call `start_flow` once and hold the returned `flow_run_id`. Do NOT start a new flow for follow-up rounds — reuse the same ID. If the id is gone (a harness timeout aborted the start, or compaction dropped it), get it back with `get_flow_status(wait: true)` on Claude Code and Codex — never by starting another flow. +1. **Start exactly one flow per user task.** Call `start_flow` once and hold the returned `flow_run_id`. Do NOT start a new flow for follow-up rounds — reuse the same ID. If the id is gone (a harness timeout aborted the start, or compaction dropped it), get it back with `get_flow_status(wait: true)` — never by starting another flow. 2. **After receiving a non-clean result**, address it, then call `send_to_participant` with the same `flow_run_id` and the updated message. 3. **Do NOT finish the user task while the flow has unresolved results.** Keep iterating until the definition's clean/complete signal. 4. **Only call `close_flow` after the clean signal.** The run stays open until you explicitly close it — don't rely on it closing itself. Then report completion to the user. @@ -218,7 +218,7 @@ report completion to user |---|---|---|---| | `start_flow` | Exactly one of `definition_id` (catalog id, e.g. `spec-review`, `code-review`, or a custom catalog id) or `definition_yaml` (inline dynamic definition — see "Composing a dynamic flow"); plus `target_kind` (what is being worked on: `spec`, `code`, `pr`, `branch`, `file`, etc.), `target_ref` (a path, branch name, or PR URL/number that identifies the target), `target_title` (short human-readable title), `context` (background context: what to focus on, constraints, definition of done) | `vendor` (reserved aliases only — explicit reviewer vendor; omit to use the definition's authored vendor, or your saved `flows.reviewer_vendor` preference if it declares none), `model` (reserved aliases only — explicit reviewer model override; REQUIRES `vendor`, rejected on dynamic/multi-participant starts), `instructions`, `mode` (`context-only` — optional; by default the participant's worktree is mirrored from THIS SESSION's project directory, not from the directory you are working in. Pass `context-only` to opt out and treat the submitted context as authoritative) | Once, at the start of a flow task. | | `send_to_participant` | `flow_run_id`, `participant` (role name declared in the flow definition's `participants` map; single-participant definitions use `reviewer` — an unknown role is rejected, naming the valid ones), `message` | `instructions`, `async` (defaults to `true`) | After addressing a non-clean result for that role, or to launch a role for the first time. Pass the same `flow_run_id`, the role's name, and the updated message. | -| `get_flow_status` | — | `flow_run_id` (on Claude Code and Codex, omit to read the newest open flow this session started; several open flows are listed instead — other harnesses must pass it), `session_id` (look up another session's flows; defaults to this session), `wait` (`true`/`false`, defaults to `false`) — when `true`, blocks until the round is terminal or roughly 3.5 minutes pass, instead of returning the current snapshot immediately | Poll or check the current status of a flow run (running, waiting, completed, failed). Use `wait: true` to ride out a long round instead of polling repeatedly yourself, and omit `flow_run_id` to recover a flow whose id you never received or lost. | +| `get_flow_status` | — | `flow_run_id` (omit to read the newest open flow this session started, or on a harness without a session identity the newest one started from this workspace; several open flows are listed instead), `session_id` (look up another session's flows; defaults to this session), `wait` (`true`/`false`, defaults to `false`) — when `true`, blocks until the round is terminal or roughly 3.5 minutes pass, instead of returning the current snapshot immediately | Poll or check the current status of a flow run (running, waiting, completed, failed). Use `wait: true` to ride out a long round instead of polling repeatedly yourself, and omit `flow_run_id` to recover a flow whose id you never received or lost. | | `close_flow` | `flow_run_id` | — | Only after the definition's clean signal — or when abandoning the task early; the run otherwise stays open until closed. | ## Example (custom definition) diff --git a/kcap/skills/review-flows/SKILL.md b/kcap/skills/review-flows/SKILL.md index 8b9b64c1a..15b3fe19b 100644 --- a/kcap/skills/review-flows/SKILL.md +++ b/kcap/skills/review-flows/SKILL.md @@ -21,7 +21,7 @@ These four tools are aliases of the generic flow tools (`start_flow`, `send_to_p `round_timeout` is an **inactivity** bound, not a wall-clock cap: a round only fails for taking too long if the reviewer goes genuinely quiet for that whole stretch — an actively-working reviewer can legitimately run a round for a long time. If a status check (or a `submit_review_round`/`start_review_flow` call) returns the benign "Flow still running" text, that is an expected outcome on a long round, not a problem — re-enter the wait with `get_review_flow_status(flow_run_id, wait: true)`, which blocks (via bounded, internally-retried checks — never a raw long-poll) until the round finishes or roughly 3.5 minutes pass, then call it again if it's still running. A round result of **`unclear` now genuinely means the reviewer went dead or silent** (no activity for the whole inactivity bound, or it crashed/was stopped) — it is no longer a symptom of a merely slow reviewer, so treat it as a liveness problem (see the `participant_unreachable` and `participant_died`/`participant_stopped` entries below), not as "the review is taking a while." -**A harness tool timeout is not a flow failure.** `start_review_flow` and `submit_review_round` block for minutes while the reviewer works. If your harness aborts the call with its own tool timeout (on Codex the error reads `timed out awaiting tools/call`), the flow is still running server-side. Do not start a new flow and do not investigate — call `get_review_flow_status(wait: true)`. Without a `flow_run_id` it reads the newest open flow this session started (pass `session_id` for another session's), so it works even when the aborted start never handed you the id or context compaction dropped it; with several open flows it lists them for you to pick from. That lookup needs a session identity, which the flows server has on Claude Code and Codex only — on any other harness the `flow_run_id` must be passed. +**A harness tool timeout is not a flow failure.** `start_review_flow` and `submit_review_round` block for minutes while the reviewer works. If your harness aborts the call with its own tool timeout (on Codex the error reads `timed out awaiting tools/call`), the flow is still running server-side. Do not start a new flow and do not investigate — call `get_review_flow_status(wait: true)`. Without a `flow_run_id` it reads the newest open flow this session started (pass `session_id` for another session's), so it works even when the aborted start never handed you the id or context compaction dropped it; with several open flows it lists them for you to pick from. On a harness that gives the flows server no session identity, it reads the flows started from this workspace on this machine instead. ## Role-surface safety gate @@ -132,7 +132,7 @@ After applying the role-surface safety gate, if `start_review_flow` / `submit_re ## Core rules -1. **Start exactly one flow per user task.** Call `start_review_flow` once and hold the returned `flow_run_id`. Do NOT start a new flow for follow-up rounds — reuse the same ID. If the id is gone (a harness timeout aborted the start, or compaction dropped it), get it back with `get_review_flow_status(wait: true)` on Claude Code and Codex — never by starting another flow. +1. **Start exactly one flow per user task.** Call `start_review_flow` once and hold the returned `flow_run_id`. Do NOT start a new flow for follow-up rounds — reuse the same ID. If the id is gone (a harness timeout aborted the start, or compaction dropped it), get it back with `get_review_flow_status(wait: true)` — never by starting another flow. 2. **After receiving a `findings` result**, address every finding, then call `submit_review_round` with the updated context and the same `flow_run_id`. 3. **Do NOT finish the user task while the flow has unresolved findings.** Keep iterating until the reviewer returns `clean`. 4. **Only call `close_review_flow` after a `clean` result.** Then report completion to the user. @@ -177,7 +177,7 @@ if findings: |---|---|---|---| | `start_review_flow` | `kind` (`spec-review`\|`code-review`), `target_kind` (what is being reviewed: `spec`, `code`, `pr`, `branch`, `file`, etc.), `target_ref` (a path, branch name, or PR URL/number that identifies the target), `target_title` (short human-readable title, e.g. spec name or PR title), `context` (background context: what to focus on, constraints, definition of done) | `vendor` (explicit reviewer vendor; omit to use the definition's authored vendor, or your saved `flows.reviewer_vendor` preference if it declares none), `model` (explicit reviewer model override — REQUIRES `vendor`; only pass it when the user named a model), `instructions`, `mode` (`context-only` — optional) | Once, at the start of a review task. | | `submit_review_round` | `flow_run_id`, `context` | `instructions` | After addressing findings. Pass the same `flow_run_id` and the updated context. | -| `get_review_flow_status` | — | `flow_run_id` (on Claude Code and Codex, omit to read the newest open flow this session started; several open flows are listed instead — other harnesses must pass it), `session_id` (look up another session's flows; defaults to this session), `wait` (`true`/`false`, defaults to `false`) — when `true`, blocks until the round is terminal or roughly 3.5 minutes pass, instead of returning the current snapshot immediately | Poll or check the current status of a flow (running, waiting, completed, failed). Use `wait: true` to ride out a long round instead of polling repeatedly yourself, and omit `flow_run_id` to recover a flow whose id you never received or lost. | +| `get_review_flow_status` | — | `flow_run_id` (omit to read the newest open flow this session started, or on a harness without a session identity the newest one started from this workspace; several open flows are listed instead), `session_id` (look up another session's flows; defaults to this session), `wait` (`true`/`false`, defaults to `false`) — when `true`, blocks until the round is terminal or roughly 3.5 minutes pass, instead of returning the current snapshot immediately | Poll or check the current status of a flow (running, waiting, completed, failed). Use `wait: true` to ride out a long round instead of polling repeatedly yourself, and omit `flow_run_id` to recover a flow whose id you never received or lost. | | `close_review_flow` | `flow_run_id` | — | Only after the reviewer returns `clean`. | ## Example (code review) diff --git a/src/Capacitor.Cli/Commands/FlowRunLedger.cs b/src/Capacitor.Cli/Commands/FlowRunLedger.cs new file mode 100644 index 000000000..a0263ab76 --- /dev/null +++ b/src/Capacitor.Cli/Commands/FlowRunLedger.cs @@ -0,0 +1,81 @@ +using System.Text.Json; +using System.Text.Json.Nodes; +using Capacitor.Cli.Core; +using Capacitor.Cli.Core.Skills; + +namespace Capacitor.Cli.Commands; + +/// +/// The flows this machine started, by workspace, for a harness that gives the flows server no +/// session identity: a status call without a flow_run_id reads its candidates here instead of +/// asking the server by session. Best-effort on both sides — a lost write costs only that lookup, +/// and a missing or corrupt file reads as empty. +/// +public sealed class FlowRunLedger(ConfigRoot config, TimeProvider time) { + public const string FileName = "flow-runs-v1.json"; + public const int MaxEntries = 50; + public static readonly TimeSpan Retention = TimeSpan.FromDays(14); + + readonly string _path = config.Path(FileName); + + /// False when the run could not be persisted; never throws. + public bool Record(string flowRunId, string workspace) { + IDisposable lease; + try { lease = config.AcquireLock(FileName, TimeSpan.FromSeconds(2)); } catch { return false; } + + using (lease) { + try { + var now = time.GetUtcNow(); + var entries = Load(now).Where(e => e.FlowRunId != flowRunId).ToList(); + entries.Insert(0, new(flowRunId, workspace, now)); + + var rows = new JsonArray(); + foreach (var e in entries.Take(MaxEntries)) + rows.Add((JsonNode)new JsonObject { + ["flow_run_id"] = e.FlowRunId, + ["workspace"] = e.Workspace, + ["started_at"] = e.StartedAt.ToString("O"), + }); + + AtomicFile.Replace(_path, new JsonObject { ["runs"] = rows }.ToJsonString()); + return true; + } catch { + return false; + } + } + } + + /// This workspace's runs, newest first. + public IReadOnlyList Recent(string workspace, int limit) => + Load(time.GetUtcNow()) + .Where(e => e.Workspace == workspace) + .Take(limit) + .Select(e => e.FlowRunId) + .ToList(); + + List Load(DateTimeOffset now) { + try { + if (!File.Exists(_path) || JsonNode.Parse(File.ReadAllText(_path)) is not JsonObject { } root + || root["runs"] is not JsonArray rows) + return []; + + var entries = new List(); + foreach (var row in rows.OfType()) { + if (Text(row, "flow_run_id") is not { } flowRunId || Text(row, "workspace") is not { } workspace + || !DateTimeOffset.TryParse(Text(row, "started_at"), out var startedAt) + || now - startedAt > Retention) + continue; + entries.Add(new(flowRunId, workspace, startedAt)); + } + + return entries; + } catch (Exception e) when (e is IOException or UnauthorizedAccessException or JsonException) { + return []; + } + } + + static string? Text(JsonObject row, string key) => + row[key] is JsonValue value && value.TryGetValue(out var text) ? text : null; + + readonly record struct Entry(string FlowRunId, string Workspace, DateTimeOffset StartedAt); +} diff --git a/src/Capacitor.Cli/Commands/McpFlowsServer.cs b/src/Capacitor.Cli/Commands/McpFlowsServer.cs index 8ff90466a..c03de2b30 100644 --- a/src/Capacitor.Cli/Commands/McpFlowsServer.cs +++ b/src/Capacitor.Cli/Commands/McpFlowsServer.cs @@ -10,6 +10,7 @@ using Capacitor.Cli.Core.Config; using Capacitor.Cli.Core.Http; using Capacitor.Cli.Core.Telemetry; +using Capacitor.Cli.Core.WorkItems; using Capacitor.Cli.PrDetection; namespace Capacitor.Cli.Commands; @@ -367,6 +368,8 @@ internal async Task HandleToolCallAsync( isError: true); } + await RecordStartedRunAsync(retryBody, requestingSessionId, cwd, repoRoot); + var (retryPayload, retryIsError) = await ResolveRoundResultAsync(client, apiRoot, retryBody, toolName, wasDynamicStart, clock, backoff, settlementStartedAt); return BuildToolResult(id, $"{PreferenceAppliedPrefix(preference)}\n{retryPayload}", retryIsError); @@ -375,6 +378,9 @@ internal async Task HandleToolCallAsync( if (!postResponse.IsSuccessStatusCode) return BuildToolResult(id, FormatFlowStartError((int)postResponse.StatusCode, postBody, wasDynamicStart), isError: true); + if (toolName is "start_review_flow" or "start_flow") + await RecordStartedRunAsync(postBody, requestingSessionId, cwd, repoRoot); + var (payload, isError) = await ResolveRoundResultAsync(client, apiRoot, postBody, toolName, wasDynamicStart, clock, backoff, settlementStartedAt); return BuildToolResult(id, payload, isError); } @@ -383,7 +389,7 @@ internal async Task HandleToolCallAsync( // false never reaches this branch — that untouched single-GET path IS the backwards-compat // contract for every existing caller. if (toolName is "get_review_flow_status" or "get_flow_status") { - var resolution = await ResolveStatusFlowRunIdAsync(client, apiRoot, arguments, requestingSessionId, clock); + var resolution = await ResolveStatusFlowRunIdAsync(client, apiRoot, arguments, requestingSessionId, cwd, repoRoot, clock); if (resolution.Error is { } unresolved) return BuildToolResult(id, unresolved, isError: true); arguments = WithFlowRunId(arguments, resolution.FlowRunId!); } @@ -1248,28 +1254,25 @@ record SessionFlow(string FlowRunId, string Status, string? DefinitionId, string const string PassTheFlowRunId = "Pass the flow_run_id returned by start_review_flow or start_flow."; /// The run a status call reads. Without a flow_run_id it is the newest open flow the - /// calling session started — or, when none is open, the session's newest flow of any state, so a - /// run that failed or closed while the driver was away is still readable. Several open flows are - /// listed for the driver to choose from, never guessed between. + /// calling session started — or, when none is open, the newest flow of any state, so a run that + /// failed or closed while the driver was away is still readable. A harness that gives this server + /// no session is answered from the runs this machine recorded for the workspace instead. Several + /// open flows are listed for the driver to choose from, never guessed between. async Task ResolveStatusFlowRunIdAsync( - HttpClient client, string apiRoot, JsonObject? arguments, string? requestingSessionId, FlowRetryClock clock) { + HttpClient client, string apiRoot, JsonObject? arguments, string? requestingSessionId, + string cwd, string? repoRoot, FlowRetryClock clock) { if (arguments?["flow_run_id"] is { } node) { if (node is not JsonValue value || !value.TryGetValue(out var given)) throw new ArgumentException("Invalid argument: flow_run_id must be a string"); if (!string.IsNullOrWhiteSpace(given)) return new(given, null); } - var sessionId = McpSessionId.ResolveWithin(arguments, requestingSessionId); - var url = $"{apiRoot}/api/flows?requesting_session_id={Uri.EscapeDataString(sessionId)}&state=all"; - using var getCts = clock.CreateTimeoutSource(PerGetTimeout); - HttpResponseMessage sent; - try { - sent = await client.GetAsync(url, getCts.Token); - } catch (Exception ex) when (ex is HttpRequestException or OperationCanceledException) { - // No caller token reaches this lane, so a cancellation here is the lookup's own timeout. - var how = ex is OperationCanceledException ? $"timed out after {(int)PerGetTimeout.TotalSeconds} s" : $"failed: {ex.Message}"; - return new(null, $"Error: the flow lookup (GET /api/flows) {how}; the flow itself is unaffected — retry the call. {PassTheFlowRunId}"); - } + if (McpSessionId.TryResolveWithin(arguments, requestingSessionId) is not { } sessionId) + return await ResolveFromRunLedgerAsync(client, apiRoot, RunLedgerWorkspace(cwd, repoRoot), clock); + + var url = $"{apiRoot}/api/flows?requesting_session_id={Uri.EscapeDataString(sessionId)}&state=all"; + var (sent, failure) = await GetForLookupAsync(client, url, "/api/flows", clock); + if (sent is null) return new(null, failure); using var resp = sent; var body = await resp.Content.ReadAsStringAsync(); @@ -1284,13 +1287,76 @@ async Task ResolveStatusFlowRunIdAsync( try { flows = ParseSessionFlows(body); } catch (JsonException) { flows = null; } if (flows is null) return new(null, $"Error: unreadable flow list from GET /api/flows. {PassTheFlowRunId}"); + return ChooseFlow(flows, $"session {sessionId}", $"Error: no flow was started by session {sessionId}. {PassTheFlowRunId}"); + } + + const int RunLedgerCandidates = 5; + + static string RunLedgerWorkspace(string cwd, string? repoRoot) => repoRoot ?? cwd; + + /// Recorded before the start's poll lane, so a start the harness aborts is still found. + /// A session-bearing harness is looked up by session instead and records nothing. + async Task RecordStartedRunAsync(string postBody, string? requestingSessionId, string cwd, string? repoRoot) { + if (WorkContextIds.CanonicalSessionId(requestingSessionId) is not null) return; + + string? flowRunId; + try { flowRunId = JsonNode.Parse(postBody) is JsonObject root ? TryGetString(root, "flow_run_id") : null; } + catch (JsonException) { return; } + if (string.IsNullOrWhiteSpace(flowRunId)) return; + + if (!new FlowRunLedger(config, time).Record(flowRunId, RunLedgerWorkspace(cwd, repoRoot))) + await Console.Error.WriteLineAsync($"kcap mcp flows: could not record flow {flowRunId} in {FlowRunLedger.FileName}; a status call without its flow_run_id will not find it"); + } + + /// Confirms each recorded run against the server, newest first; a run the server no + /// longer knows is skipped rather than read. + async Task ResolveFromRunLedgerAsync(HttpClient client, string apiRoot, string workspace, FlowRetryClock clock) { + var recorded = new FlowRunLedger(config, time).Recent(workspace, RunLedgerCandidates); + if (recorded.Count == 0) + return new(null, $"Error: this harness gives kcap no session id, and no flow started from {workspace} is recorded on this machine. {PassTheFlowRunId}"); + + var flows = new List(); + foreach (var flowRunId in recorded) { + var route = $"/api/flows/{Uri.EscapeDataString(flowRunId)}"; + var (sent, failure) = await GetForLookupAsync(client, apiRoot + route, route, clock); + if (sent is null) return new(null, failure); + using var resp = sent; + var body = await resp.Content.ReadAsStringAsync(); + + if (resp.StatusCode == HttpStatusCode.Unauthorized) + return new(null, await AuthRejectionNotice.ForPersistentUnauthorizedAsync(store, profiles.Name, apiRoot, time)); + if (resp.StatusCode == HttpStatusCode.NotFound) continue; + if (!resp.IsSuccessStatusCode) + return new(null, FormatFlowStartError((int)resp.StatusCode, body, wasDynamicStart: false)); + + SessionFlow? flow; + try { flow = JsonNode.Parse(body) is JsonObject row ? ParseFlowRow(row) : null; } catch (JsonException) { flow = null; } + if (flow is null) return new(null, $"Error: unreadable flow from GET {route}. {PassTheFlowRunId}"); + flows.Add(flow); + } + + return ChooseFlow(flows, $"workspace {workspace}", $"Error: no flow recorded for {workspace} is known to this server. {PassTheFlowRunId}"); + } + + /// A null response comes with the actionable error text in its place. + static async Task<(HttpResponseMessage? Response, string Failure)> GetForLookupAsync( + HttpClient client, string url, string route, FlowRetryClock clock) { + using var getCts = clock.CreateTimeoutSource(PerGetTimeout); + try { + return (await client.GetAsync(url, getCts.Token), ""); + } catch (Exception ex) when (ex is HttpRequestException or OperationCanceledException) { + // No caller token reaches this lane, so a cancellation here is the lookup's own timeout. + var how = ex is OperationCanceledException ? $"timed out after {(int)PerGetTimeout.TotalSeconds} s" : $"failed: {ex.Message}"; + return (null, $"Error: the flow lookup (GET {route}) {how}; the flow itself is unaffected — retry the call. {PassTheFlowRunId}"); + } + } + + static FlowRunIdResolution ChooseFlow(List flows, string owner, string noneError) { var open = flows.Where(f => f.Status is not ("closed" or "failed")).ToList(); - if (open.Count > 1) return new(null, DescribeOpenFlows(sessionId, open)); + if (open.Count > 1) return new(null, DescribeOpenFlows(owner, open)); var chosen = open.Count == 1 ? open[0] : flows.FirstOrDefault(); - return chosen is null - ? new(null, $"Error: no flow was started by session {sessionId}. {PassTheFlowRunId}") - : new(chosen.FlowRunId, null); + return chosen is null ? new(null, noneError) : new(chosen.FlowRunId, null); } /// Null for a body that is not the route's shape — an object with a flows array — @@ -1298,24 +1364,27 @@ async Task ResolveStatusFlowRunIdAsync( static List? ParseSessionFlows(string body) { if (JsonNode.Parse(body) is not JsonObject root || root["flows"] is not JsonArray rows) return null; var flows = new List(); - foreach (var row in rows.OfType()) { - var flowRunId = TryGetString(row, "flow_run_id"); - if (string.IsNullOrWhiteSpace(flowRunId)) continue; - flows.Add(new( - flowRunId, - TryGetString(row, "status") ?? "", - TryGetString(row, "definition_id"), - TryGetString(row, "target_title"), - row["round_number"] is JsonValue round && round.TryGetValue(out var roundNumber) ? roundNumber : null, - TryGetString(row, "round_status"), - TryGetString(row, "started_at"))); - } + foreach (var row in rows.OfType()) + if (ParseFlowRow(row) is { } flow) flows.Add(flow); return flows; } - static string DescribeOpenFlows(string sessionId, IReadOnlyList open) { + static SessionFlow? ParseFlowRow(JsonObject row) { + var flowRunId = TryGetString(row, "flow_run_id"); + if (string.IsNullOrWhiteSpace(flowRunId)) return null; + return new( + flowRunId, + TryGetString(row, "status") ?? "", + TryGetString(row, "definition_id"), + TryGetString(row, "target_title"), + row["round_number"] is JsonValue round && round.TryGetValue(out var roundNumber) ? roundNumber : null, + TryGetString(row, "round_status"), + TryGetString(row, "started_at")); + } + + static string DescribeOpenFlows(string owner, IReadOnlyList open) { var sb = new StringBuilder(); - sb.Append("Error: session ").Append(sessionId).Append(" has ").Append(open.Count).Append(" open flows. Pass flow_run_id for the one you mean:"); + sb.Append("Error: ").Append(owner).Append(" has ").Append(open.Count).Append(" open flows. Pass flow_run_id for the one you mean:"); foreach (var flow in open) { sb.AppendLine().Append(" ").Append(flow.FlowRunId); if (flow.DefinitionId is { } definitionId) sb.Append(" ").Append(definitionId); @@ -2203,7 +2272,7 @@ static string ToResponse(JsonNode id, T result, JsonTypeInfo typeInfo) { /// Carried by every tool that holds the call open while a round runs. A harness that /// aborts the call words the error itself, so this is the only guidance the driver has read. static string HarnessTimeoutGuidance(string statusTool) => - $"This call blocks for minutes while the round runs. If your harness aborts it with a tool timeout, the flow is still running server-side: do not start it again and do not investigate — call {statusTool} with wait: true (on Claude Code and Codex flow_run_id may be omitted and resolves this session's open flow; other harnesses give this server no session identity, so pass the id there). "; + $"This call blocks for minutes while the round runs. If your harness aborts it with a tool timeout, the flow is still running server-side: do not start it again and do not investigate — call {statusTool} with wait: true (flow_run_id may be omitted: it resolves the open flow this session started, or on a harness without a session identity the open flow started from this workspace). "; internal static McpTool[] BuildToolsList() => [ new( @@ -2257,7 +2326,7 @@ internal static McpTool[] BuildToolsList() => [ new( "object", new() { - ["flow_run_id"] = new("string", "Flow run ID returned by start_review_flow. Optional on Claude Code and Codex, the harnesses whose session this server can identify: when omitted, the newest open flow this session started is read — the way back to a run whose start the harness aborted, or whose id was lost to context compaction. With several open flows the reply lists them instead."), + ["flow_run_id"] = new("string", "Flow run ID returned by start_review_flow. Optional: when omitted, the newest open flow this session started is read (on a harness that gives this server no session, the newest one started from this workspace on this machine) — the way back to a run whose start the harness aborted, or whose id was lost to context compaction. With several open flows the reply lists them instead."), ["session_id"] = new("string", "Session whose flows to look up when flow_run_id is omitted. Defaults to the session this server runs in when omitted."), ["wait"] = new("boolean", "Optional, defaults to false. When true, block until the round is terminal or roughly 3.5 minutes elapse, instead of returning immediately.") }, @@ -2332,7 +2401,7 @@ internal static McpTool[] BuildToolsList() => [ new( "object", new() { - ["flow_run_id"] = new("string", "Flow run ID returned by start_flow. Optional on Claude Code and Codex, the harnesses whose session this server can identify: when omitted, the newest open flow this session started is read — the way back to a run whose start the harness aborted, or whose id was lost to context compaction. With several open flows the reply lists them instead."), + ["flow_run_id"] = new("string", "Flow run ID returned by start_flow. Optional: when omitted, the newest open flow this session started is read (on a harness that gives this server no session, the newest one started from this workspace on this machine) — the way back to a run whose start the harness aborted, or whose id was lost to context compaction. With several open flows the reply lists them instead."), ["session_id"] = new("string", "Session whose flows to look up when flow_run_id is omitted. Defaults to the session this server runs in when omitted."), ["wait"] = new("boolean", "Optional, defaults to false. When true, block until the round is terminal or roughly 3.5 minutes elapse, instead of returning immediately.") }, diff --git a/src/Capacitor.Cli/Commands/McpSessionId.cs b/src/Capacitor.Cli/Commands/McpSessionId.cs index a6f770de5..cb95ce812 100644 --- a/src/Capacitor.Cli/Commands/McpSessionId.cs +++ b/src/Capacitor.Cli/Commands/McpSessionId.cs @@ -21,9 +21,12 @@ internal static string Resolve(JsonObject? args, Func getEnv) = /// For a server that resolved its harness session once at startup and hands it down: /// an explicit argument still wins, and the environment is never consulted here. internal static string ResolveWithin(JsonObject? args, string? ambientSessionId) => - Explicit(args) - ?? WorkContextIds.CanonicalSessionId(ambientSessionId) - ?? throw new ArgumentException(NoSessionIdMessage); + TryResolveWithin(args, ambientSessionId) ?? throw new ArgumentException(NoSessionIdMessage); + + /// for a caller with a fallback: null when neither source + /// names a session. + internal static string? TryResolveWithin(JsonObject? args, string? ambientSessionId) => + Explicit(args) ?? WorkContextIds.CanonicalSessionId(ambientSessionId); static string? Explicit(JsonObject? args) { if (args?["session_id"] is not { } node) return null; diff --git a/test/Capacitor.Cli.Tests.Unit/Commands/FlowRunLedgerTests.cs b/test/Capacitor.Cli.Tests.Unit/Commands/FlowRunLedgerTests.cs new file mode 100644 index 000000000..e772761d3 --- /dev/null +++ b/test/Capacitor.Cli.Tests.Unit/Commands/FlowRunLedgerTests.cs @@ -0,0 +1,83 @@ +using Capacitor.Cli.Commands; +using Microsoft.Extensions.Time.Testing; +using TUnit.Assertions.Enums; + +namespace Capacitor.Cli.Tests.Unit.Commands; + +public class FlowRunLedgerTests { + [TempConfigRoot] public required TempConfigRoot Config { get; init; } + + readonly FakeTimeProvider _time = new(new DateTimeOffset(2026, 9, 23, 10, 0, 0, TimeSpan.Zero)); + + FlowRunLedger Ledger() => new(Config.Root, _time); + + [Test] + public async Task Recent_lists_this_workspaces_runs_newest_first() { + var ledger = Ledger(); + ledger.Record("run-1", "/repo/a"); + _time.Advance(TimeSpan.FromMinutes(1)); + ledger.Record("run-2", "/repo/b"); + _time.Advance(TimeSpan.FromMinutes(1)); + ledger.Record("run-3", "/repo/a"); + + await Assert.That(Ledger().Recent("/repo/a", 5)).IsEquivalentTo(["run-3", "run-1"], CollectionOrdering.Matching); + } + + [Test] + public async Task Recent_honours_the_limit() { + var ledger = Ledger(); + for (var i = 0; i < 4; i++) { + ledger.Record($"run-{i}", "/repo/a"); + _time.Advance(TimeSpan.FromSeconds(1)); + } + + await Assert.That(ledger.Recent("/repo/a", 2)).IsEquivalentTo(["run-3", "run-2"], CollectionOrdering.Matching); + } + + [Test] + public async Task Recording_a_run_again_moves_it_to_the_front_without_a_duplicate() { + var ledger = Ledger(); + ledger.Record("run-1", "/repo/a"); + _time.Advance(TimeSpan.FromSeconds(1)); + ledger.Record("run-2", "/repo/a"); + _time.Advance(TimeSpan.FromSeconds(1)); + ledger.Record("run-1", "/repo/a"); + + await Assert.That(ledger.Recent("/repo/a", 5)).IsEquivalentTo(["run-1", "run-2"], CollectionOrdering.Matching); + } + + [Test] + public async Task Runs_past_the_retention_are_not_returned() { + var ledger = Ledger(); + ledger.Record("old", "/repo/a"); + _time.Advance(FlowRunLedger.Retention + TimeSpan.FromMinutes(1)); + ledger.Record("new", "/repo/a"); + + await Assert.That(ledger.Recent("/repo/a", 5)).IsEquivalentTo(["new"], CollectionOrdering.Matching); + } + + [Test] + public async Task The_ledger_keeps_only_the_newest_entries() { + var ledger = Ledger(); + for (var i = 0; i < FlowRunLedger.MaxEntries + 3; i++) { + ledger.Record($"run-{i}", "/repo/a"); + _time.Advance(TimeSpan.FromSeconds(1)); + } + + var all = ledger.Recent("/repo/a", int.MaxValue); + await Assert.That(all.Count).IsEqualTo(FlowRunLedger.MaxEntries); + await Assert.That(all[0]).IsEqualTo($"run-{FlowRunLedger.MaxEntries + 2}"); + await Assert.That(all).DoesNotContain("run-2"); + } + + [Test] + public async Task A_corrupt_file_reads_as_empty_and_is_replaced_on_the_next_record() { + File.WriteAllText(Config.Root.Path(FlowRunLedger.FileName), "{not json"); + var ledger = Ledger(); + + await Assert.That(ledger.Recent("/repo/a", 5)).IsEmpty(); + + ledger.Record("run-1", "/repo/a"); + await Assert.That(ledger.Recent("/repo/a", 5)).IsEquivalentTo(["run-1"], CollectionOrdering.Matching); + } +} diff --git a/test/Capacitor.Cli.Tests.Unit/Commands/StatusWithoutFlowRunIdTests.cs b/test/Capacitor.Cli.Tests.Unit/Commands/StatusWithoutFlowRunIdTests.cs index e1fd70ecc..11c01ed46 100644 --- a/test/Capacitor.Cli.Tests.Unit/Commands/StatusWithoutFlowRunIdTests.cs +++ b/test/Capacitor.Cli.Tests.Unit/Commands/StatusWithoutFlowRunIdTests.cs @@ -2,6 +2,7 @@ using Capacitor.Cli.Commands; using Capacitor.Cli.Core; using Capacitor.Cli.PrDetection; +using Microsoft.Extensions.Time.Testing; using WireMock.RequestBuilders; using WireMock.ResponseBuilders; using WireMock.Server; @@ -9,7 +10,8 @@ namespace Capacitor.Cli.Tests.Unit.Commands; /// -/// A status call without a flow_run_id resolves the flow from the calling session: the driver +/// A status call without a flow_run_id resolves the flow from the calling session, or on a +/// harness with no session from the runs this machine recorded for the workspace: the driver /// that never received the id (its harness aborted the start) or lost it (context compaction) has /// no other way back to a run that is still working for it. /// @@ -292,21 +294,155 @@ public async Task A_server_without_the_lookup_route_says_to_pass_the_id() { await Assert.That(text).Contains("Pass the flow_run_id"); } + const string Workspace = "/repo/a"; + + void GivenRecordedRuns(params string[] newestLast) { + var time = new FakeTimeProvider(DateTimeOffset.UtcNow); + var ledger = new FlowRunLedger(Config.Root, time); + foreach (var flowRunId in newestLast) { + ledger.Record(flowRunId, Workspace); + time.Advance(TimeSpan.FromSeconds(1)); + } + } + + Task SessionlessStatusAsync(WireMockServer server, HttpClient client, JsonObject arguments, string? repoRoot = Workspace, FlowRetryClock? clock = null) => + Server().HandleToolCallAsync( + JsonNode.Parse("1")!, ToolCallRequest("get_review_flow_status", arguments), + client, server.Url!, cwd: "/tmp/cwd", repoRoot: repoRoot, repoInfo: null, clock: clock, requestingSessionId: null); + [Test] - public async Task No_session_id_anywhere_is_a_clean_error_before_any_request() { + public async Task No_session_and_no_recorded_run_is_a_clean_error_before_any_request() { using var server = WireMockServer.Start(); using var client = new HttpClient(); - var response = await Server().HandleToolCallAsync( - JsonNode.Parse("1")!, ToolCallRequest("get_review_flow_status", new JsonObject()), - client, server.Url!, cwd: "/tmp/cwd", repoRoot: null, repoInfo: null, requestingSessionId: null); + var (text, isError) = Unwrap(await SessionlessStatusAsync(server, client, new JsonObject())); - var (text, isError) = Unwrap(response); await Assert.That(isError).IsTrue(); - await Assert.That(text).Contains("pass session_id explicitly"); + await Assert.That(text).Contains("no session id"); + await Assert.That(text).Contains(Workspace); + await Assert.That(text).Contains("Pass the flow_run_id"); await Assert.That(server.LogEntries.Count).IsEqualTo(0); } + [Test] + public async Task Without_a_session_the_workspaces_recorded_open_run_is_read() { + using var server = WireMockServer.Start(); + GivenRecordedRuns("flow-old", "flow-new"); + GivenFlow(server, "flow-old", "closed"); + GivenFlow(server, "flow-new", "waiting"); + using var client = new HttpClient(); + + var (text, isError) = Unwrap(await SessionlessStatusAsync(server, client, new JsonObject())); + + await Assert.That(isError).IsFalse(); + await Assert.That(text).Contains("flow_run_id: flow-new"); + await Assert.That(text).Contains("fix line 42"); + await Assert.That(Gets(server, "/api/flows")).IsEqualTo(0); + } + + [Test] + public async Task Without_a_session_wait_polls_the_recorded_run() { + using var server = WireMockServer.Start(); + GivenRecordedRuns("flow-1"); + server.Given(Request.Create().WithPath("/api/flows/flow-1").UsingGet()) + .InScenario("ledger-wait").WillSetStateTo("second") + .RespondWith(Response.Create().WithStatusCode(200).WithHeader("Content-Type", "application/json") + .WithBody("""{"flow_run_id":"flow-1","status":"running","definition_id":"code-review","target_title":"t","round_count":1,"round_number":1,"round_status":"running"}""")); + server.Given(Request.Create().WithPath("/api/flows/flow-1").UsingGet()) + .InScenario("ledger-wait").WhenStateIs("second") + .RespondWith(Response.Create().WithStatusCode(200).WithHeader("Content-Type", "application/json") + .WithBody("""{"flow_run_id":"flow-1","status":"waiting","definition_id":"code-review","target_title":"t","round_count":1,"round_number":1,"round_status":"findings","last_result_kind":"findings","last_result_text":"fix line 42"}""")); + using var client = new HttpClient(); + + var (text, isError) = Unwrap(await SessionlessStatusAsync(server, client, new JsonObject { ["wait"] = true }, clock: new VirtualFlowRetryClock())); + + await Assert.That(isError).IsFalse(); + await Assert.That(text).Contains("fix line 42"); + } + + [Test] + public async Task Another_workspaces_runs_are_not_candidates() { + using var server = WireMockServer.Start(); + GivenRecordedRuns("flow-1"); + using var client = new HttpClient(); + + var (text, isError) = Unwrap(await SessionlessStatusAsync(server, client, new JsonObject(), repoRoot: "/repo/b")); + + await Assert.That(isError).IsTrue(); + await Assert.That(text).Contains("/repo/b"); + await Assert.That(server.LogEntries.Count).IsEqualTo(0); + } + + [Test] + public async Task Without_a_session_several_open_recorded_runs_are_listed() { + using var server = WireMockServer.Start(); + GivenRecordedRuns("flow-1", "flow-2"); + GivenFlow(server, "flow-1", "running"); + GivenFlow(server, "flow-2", "waiting"); + using var client = new HttpClient(); + + var (text, isError) = Unwrap(await SessionlessStatusAsync(server, client, new JsonObject())); + + await Assert.That(isError).IsTrue(); + await Assert.That(text).Contains($"workspace {Workspace} has 2 open flows"); + await Assert.That(text).Contains("flow-1"); + await Assert.That(text).Contains("flow-2"); + } + + [Test] + public async Task A_recorded_run_the_server_no_longer_knows_is_skipped() { + using var server = WireMockServer.Start(); + GivenRecordedRuns("flow-1", "flow-gone"); + server.Given(Request.Create().WithPath("/api/flows/flow-gone").UsingGet()) + .RespondWith(Response.Create().WithStatusCode(404)); + GivenFlow(server, "flow-1", "closed"); + using var client = new HttpClient(); + + var (text, isError) = Unwrap(await SessionlessStatusAsync(server, client, new JsonObject())); + + await Assert.That(isError).IsFalse(); + await Assert.That(text).Contains("flow_run_id: flow-1"); + } + + [Test] + public async Task A_start_without_a_session_records_its_run_for_the_workspace() { + using var server = WireMockServer.Start(); + server.Given(Request.Create().WithPath("/api/flows/review/start/v2").UsingPost()) + .RespondWith(Response.Create().WithStatusCode(200).WithBody( + """{"flow_run_id":"flow-started","status":"running","round_id":null,"round_number":null}""")); + using var client = new HttpClient(); + + await Server().HandleToolCallAsync( + JsonNode.Parse("1")!, ToolCallRequest("start_review_flow", StartArguments()), + client, server.Url!, cwd: "/tmp/cwd", repoRoot: Workspace, repoInfo: null, requestingSessionId: null); + + await Assert.That(new FlowRunLedger(Config.Root, TimeProvider.System).Recent(Workspace, 5)) + .IsEquivalentTo(["flow-started"]); + } + + [Test] + public async Task A_start_with_a_session_records_nothing() { + using var server = WireMockServer.Start(); + server.Given(Request.Create().WithPath("/api/flows/review/start/v2").UsingPost()) + .RespondWith(Response.Create().WithStatusCode(200).WithBody( + """{"flow_run_id":"flow-started","status":"running","round_id":null,"round_number":null}""")); + using var client = new HttpClient(); + + await Server().HandleToolCallAsync( + JsonNode.Parse("1")!, ToolCallRequest("start_review_flow", StartArguments()), + client, server.Url!, cwd: "/tmp/cwd", repoRoot: Workspace, repoInfo: null, requestingSessionId: SessionId); + + await Assert.That(new FlowRunLedger(Config.Root, TimeProvider.System).Recent(Workspace, 5)).IsEmpty(); + } + + static JsonObject StartArguments() => new() { + ["kind"] = "code-review", + ["target_kind"] = "pr", + ["target_ref"] = "123", + ["target_title"] = "some PR", + ["context"] = "some context" + }; + [Test] public async Task A_non_string_flow_run_id_is_a_clean_error_before_any_request() { using var server = WireMockServer.Start(); From 747b74cedfa7b29c92c6904c5ea70cadc54c4915 Mon Sep 17 00:00:00 2001 From: Alexey Zimarev Date: Wed, 23 Sep 2026 17:47:36 +0200 Subject: [PATCH 2/2] Check every retained run and keep workspace paths from the model (#1123) A fixed probe count let newer settled runs hide an open one, and a 404 inside the start's not-found grace may only mean the server has not caught up. Co-Authored-By: Claude Opus 5.5 --- docs/CHANGES.md | 6 ++- src/Capacitor.Cli/Commands/FlowRunLedger.cs | 12 ++--- src/Capacitor.Cli/Commands/McpFlowsServer.cs | 22 ++++---- .../Commands/FlowRunLedgerTests.cs | 30 ++++++----- .../Commands/StatusWithoutFlowRunIdTests.cs | 52 ++++++++++++++++--- 5 files changed, 82 insertions(+), 40 deletions(-) diff --git a/docs/CHANGES.md b/docs/CHANGES.md index 53ce2493f..1b7982243 100644 --- a/docs/CHANGES.md +++ b/docs/CHANGES.md @@ -12,8 +12,10 @@ The status tools find a flow without its id by the calling session, and Cursor, Kiro, OpenCode and Antigravity export no session into the long-lived MCP child, so a driver there had no way back to a run whose start the harness aborted. The flows server now records each run it starts without a session in `flow-runs-v1.json` under the config root, keyed on the repository root -(or working directory), and a bare status call without a session reads that workspace's newest -entries and confirms each with `GET /api/flows/{id}`. The record is written before the start's poll +(or working directory), and a bare status call without a session confirms every entry that +workspace retains with `GET /api/flows/{id}` — all of them, because a fixed probe count would let +newer settled runs hide an open one. A 404 on a run recorded within the start poll's not-found grace +answers "retry" rather than skipping it, since the server may not have caught up with the start. The record is written before the start's poll lane begins, which is what makes it survive an abort: the server ignores MCP cancellations, so the lane holds the id after the harness has given up on the call. Session-bearing harnesses record nothing and keep the server-side lookup. Several chats in one workspace share the ledger; the same diff --git a/src/Capacitor.Cli/Commands/FlowRunLedger.cs b/src/Capacitor.Cli/Commands/FlowRunLedger.cs index a0263ab76..59a0696af 100644 --- a/src/Capacitor.Cli/Commands/FlowRunLedger.cs +++ b/src/Capacitor.Cli/Commands/FlowRunLedger.cs @@ -45,13 +45,9 @@ public bool Record(string flowRunId, string workspace) { } } - /// This workspace's runs, newest first. - public IReadOnlyList Recent(string workspace, int limit) => - Load(time.GetUtcNow()) - .Where(e => e.Workspace == workspace) - .Take(limit) - .Select(e => e.FlowRunId) - .ToList(); + /// Every run this workspace still retains, newest first. + public IReadOnlyList Retained(string workspace) => + Load(time.GetUtcNow()).Where(e => e.Workspace == workspace).ToList(); List Load(DateTimeOffset now) { try { @@ -77,5 +73,5 @@ List Load(DateTimeOffset now) { static string? Text(JsonObject row, string key) => row[key] is JsonValue value && value.TryGetValue(out var text) ? text : null; - readonly record struct Entry(string FlowRunId, string Workspace, DateTimeOffset StartedAt); + public readonly record struct Entry(string FlowRunId, string Workspace, DateTimeOffset StartedAt); } diff --git a/src/Capacitor.Cli/Commands/McpFlowsServer.cs b/src/Capacitor.Cli/Commands/McpFlowsServer.cs index c03de2b30..b04157f27 100644 --- a/src/Capacitor.Cli/Commands/McpFlowsServer.cs +++ b/src/Capacitor.Cli/Commands/McpFlowsServer.cs @@ -1290,8 +1290,6 @@ async Task ResolveStatusFlowRunIdAsync( return ChooseFlow(flows, $"session {sessionId}", $"Error: no flow was started by session {sessionId}. {PassTheFlowRunId}"); } - const int RunLedgerCandidates = 5; - static string RunLedgerWorkspace(string cwd, string? repoRoot) => repoRoot ?? cwd; /// Recorded before the start's poll lane, so a start the harness aborts is still found. @@ -1308,15 +1306,17 @@ async Task RecordStartedRunAsync(string postBody, string? requestingSessionId, s await Console.Error.WriteLineAsync($"kcap mcp flows: could not record flow {flowRunId} in {FlowRunLedger.FileName}; a status call without its flow_run_id will not find it"); } - /// Confirms each recorded run against the server, newest first; a run the server no - /// longer knows is skipped rather than read. + /// Confirms every retained run against the server, newest first, so an open run behind + /// newer settled ones is still found. A run the server no longer knows is skipped — unless it + /// was recorded within , when the 404 may only mean the server has + /// not caught up with the start. The workspace path is never shown to the model. async Task ResolveFromRunLedgerAsync(HttpClient client, string apiRoot, string workspace, FlowRetryClock clock) { - var recorded = new FlowRunLedger(config, time).Recent(workspace, RunLedgerCandidates); + var recorded = new FlowRunLedger(config, time).Retained(workspace); if (recorded.Count == 0) - return new(null, $"Error: this harness gives kcap no session id, and no flow started from {workspace} is recorded on this machine. {PassTheFlowRunId}"); + return new(null, $"Error: this harness gives kcap no session id, and no flow started from this workspace is recorded on this machine. {PassTheFlowRunId}"); var flows = new List(); - foreach (var flowRunId in recorded) { + foreach (var (flowRunId, _, startedAt) in recorded) { var route = $"/api/flows/{Uri.EscapeDataString(flowRunId)}"; var (sent, failure) = await GetForLookupAsync(client, apiRoot + route, route, clock); if (sent is null) return new(null, failure); @@ -1325,7 +1325,11 @@ async Task ResolveFromRunLedgerAsync(HttpClient client, str if (resp.StatusCode == HttpStatusCode.Unauthorized) return new(null, await AuthRejectionNotice.ForPersistentUnauthorizedAsync(store, profiles.Name, apiRoot, time)); - if (resp.StatusCode == HttpStatusCode.NotFound) continue; + if (resp.StatusCode == HttpStatusCode.NotFound) { + if (time.GetUtcNow() - startedAt < NotFoundGrace) + return new(null, $"Error: flow {flowRunId} was started moments ago and the server cannot read it yet — retry the call."); + continue; + } if (!resp.IsSuccessStatusCode) return new(null, FormatFlowStartError((int)resp.StatusCode, body, wasDynamicStart: false)); @@ -1335,7 +1339,7 @@ async Task ResolveFromRunLedgerAsync(HttpClient client, str flows.Add(flow); } - return ChooseFlow(flows, $"workspace {workspace}", $"Error: no flow recorded for {workspace} is known to this server. {PassTheFlowRunId}"); + return ChooseFlow(flows, "this workspace", $"Error: no flow recorded for this workspace is known to this server. {PassTheFlowRunId}"); } /// A null response comes with the actionable error text in its place. diff --git a/test/Capacitor.Cli.Tests.Unit/Commands/FlowRunLedgerTests.cs b/test/Capacitor.Cli.Tests.Unit/Commands/FlowRunLedgerTests.cs index e772761d3..6c2053b90 100644 --- a/test/Capacitor.Cli.Tests.Unit/Commands/FlowRunLedgerTests.cs +++ b/test/Capacitor.Cli.Tests.Unit/Commands/FlowRunLedgerTests.cs @@ -7,12 +7,17 @@ namespace Capacitor.Cli.Tests.Unit.Commands; public class FlowRunLedgerTests { [TempConfigRoot] public required TempConfigRoot Config { get; init; } - readonly FakeTimeProvider _time = new(new DateTimeOffset(2026, 9, 23, 10, 0, 0, TimeSpan.Zero)); + static readonly DateTimeOffset Start = new(2026, 9, 23, 10, 0, 0, TimeSpan.Zero); + + readonly FakeTimeProvider _time = new(Start); FlowRunLedger Ledger() => new(Config.Root, _time); + static List Ids(FlowRunLedger ledger, string workspace) => + ledger.Retained(workspace).Select(e => e.FlowRunId).ToList(); + [Test] - public async Task Recent_lists_this_workspaces_runs_newest_first() { + public async Task Retained_lists_this_workspaces_runs_newest_first() { var ledger = Ledger(); ledger.Record("run-1", "/repo/a"); _time.Advance(TimeSpan.FromMinutes(1)); @@ -20,18 +25,15 @@ public async Task Recent_lists_this_workspaces_runs_newest_first() { _time.Advance(TimeSpan.FromMinutes(1)); ledger.Record("run-3", "/repo/a"); - await Assert.That(Ledger().Recent("/repo/a", 5)).IsEquivalentTo(["run-3", "run-1"], CollectionOrdering.Matching); + await Assert.That(Ids(Ledger(), "/repo/a")).IsEquivalentTo(["run-3", "run-1"], CollectionOrdering.Matching); } [Test] - public async Task Recent_honours_the_limit() { + public async Task Retained_carries_when_each_run_was_recorded() { var ledger = Ledger(); - for (var i = 0; i < 4; i++) { - ledger.Record($"run-{i}", "/repo/a"); - _time.Advance(TimeSpan.FromSeconds(1)); - } + ledger.Record("run-1", "/repo/a"); - await Assert.That(ledger.Recent("/repo/a", 2)).IsEquivalentTo(["run-3", "run-2"], CollectionOrdering.Matching); + await Assert.That(ledger.Retained("/repo/a").Single().StartedAt).IsEqualTo(Start); } [Test] @@ -43,7 +45,7 @@ public async Task Recording_a_run_again_moves_it_to_the_front_without_a_duplicat _time.Advance(TimeSpan.FromSeconds(1)); ledger.Record("run-1", "/repo/a"); - await Assert.That(ledger.Recent("/repo/a", 5)).IsEquivalentTo(["run-1", "run-2"], CollectionOrdering.Matching); + await Assert.That(Ids(ledger, "/repo/a")).IsEquivalentTo(["run-1", "run-2"], CollectionOrdering.Matching); } [Test] @@ -53,7 +55,7 @@ public async Task Runs_past_the_retention_are_not_returned() { _time.Advance(FlowRunLedger.Retention + TimeSpan.FromMinutes(1)); ledger.Record("new", "/repo/a"); - await Assert.That(ledger.Recent("/repo/a", 5)).IsEquivalentTo(["new"], CollectionOrdering.Matching); + await Assert.That(Ids(ledger, "/repo/a")).IsEquivalentTo(["new"], CollectionOrdering.Matching); } [Test] @@ -64,7 +66,7 @@ public async Task The_ledger_keeps_only_the_newest_entries() { _time.Advance(TimeSpan.FromSeconds(1)); } - var all = ledger.Recent("/repo/a", int.MaxValue); + var all = Ids(ledger, "/repo/a"); await Assert.That(all.Count).IsEqualTo(FlowRunLedger.MaxEntries); await Assert.That(all[0]).IsEqualTo($"run-{FlowRunLedger.MaxEntries + 2}"); await Assert.That(all).DoesNotContain("run-2"); @@ -75,9 +77,9 @@ public async Task A_corrupt_file_reads_as_empty_and_is_replaced_on_the_next_reco File.WriteAllText(Config.Root.Path(FlowRunLedger.FileName), "{not json"); var ledger = Ledger(); - await Assert.That(ledger.Recent("/repo/a", 5)).IsEmpty(); + await Assert.That(Ids(ledger, "/repo/a")).IsEmpty(); ledger.Record("run-1", "/repo/a"); - await Assert.That(ledger.Recent("/repo/a", 5)).IsEquivalentTo(["run-1"], CollectionOrdering.Matching); + await Assert.That(Ids(ledger, "/repo/a")).IsEquivalentTo(["run-1"], CollectionOrdering.Matching); } } diff --git a/test/Capacitor.Cli.Tests.Unit/Commands/StatusWithoutFlowRunIdTests.cs b/test/Capacitor.Cli.Tests.Unit/Commands/StatusWithoutFlowRunIdTests.cs index 11c01ed46..59d3259be 100644 --- a/test/Capacitor.Cli.Tests.Unit/Commands/StatusWithoutFlowRunIdTests.cs +++ b/test/Capacitor.Cli.Tests.Unit/Commands/StatusWithoutFlowRunIdTests.cs @@ -296,8 +296,11 @@ public async Task A_server_without_the_lookup_route_says_to_pass_the_id() { const string Workspace = "/repo/a"; - void GivenRecordedRuns(params string[] newestLast) { - var time = new FakeTimeProvider(DateTimeOffset.UtcNow); + /// Recorded an hour ago, past the grace a just-started run gets for a 404. + void GivenRecordedRuns(params string[] newestLast) => GivenRecordedRunsAt(DateTimeOffset.UtcNow.AddHours(-1), newestLast); + + void GivenRecordedRunsAt(DateTimeOffset firstAt, params string[] newestLast) { + var time = new FakeTimeProvider(firstAt); var ledger = new FlowRunLedger(Config.Root, time); foreach (var flowRunId in newestLast) { ledger.Record(flowRunId, Workspace); @@ -319,7 +322,8 @@ public async Task No_session_and_no_recorded_run_is_a_clean_error_before_any_req await Assert.That(isError).IsTrue(); await Assert.That(text).Contains("no session id"); - await Assert.That(text).Contains(Workspace); + await Assert.That(text).Contains("this workspace"); + await Assert.That(text).DoesNotContain(Workspace); await Assert.That(text).Contains("Pass the flow_run_id"); await Assert.That(server.LogEntries.Count).IsEqualTo(0); } @@ -369,7 +373,7 @@ public async Task Another_workspaces_runs_are_not_candidates() { var (text, isError) = Unwrap(await SessionlessStatusAsync(server, client, new JsonObject(), repoRoot: "/repo/b")); await Assert.That(isError).IsTrue(); - await Assert.That(text).Contains("/repo/b"); + await Assert.That(text).Contains("no flow started from this workspace"); await Assert.That(server.LogEntries.Count).IsEqualTo(0); } @@ -384,7 +388,8 @@ public async Task Without_a_session_several_open_recorded_runs_are_listed() { var (text, isError) = Unwrap(await SessionlessStatusAsync(server, client, new JsonObject())); await Assert.That(isError).IsTrue(); - await Assert.That(text).Contains($"workspace {Workspace} has 2 open flows"); + await Assert.That(text).Contains("this workspace has 2 open flows"); + await Assert.That(text).DoesNotContain(Workspace); await Assert.That(text).Contains("flow-1"); await Assert.That(text).Contains("flow-2"); } @@ -404,6 +409,39 @@ public async Task A_recorded_run_the_server_no_longer_knows_is_skipped() { await Assert.That(text).Contains("flow_run_id: flow-1"); } + [Test] + public async Task An_older_open_run_behind_many_newer_settled_ones_is_still_found() { + using var server = WireMockServer.Start(); + var settled = Enumerable.Range(1, 8).Select(i => $"flow-settled-{i}").ToArray(); + GivenRecordedRuns(["flow-open", ..settled]); + GivenFlow(server, "flow-open", "waiting"); + foreach (var flowRunId in settled) GivenFlow(server, flowRunId, "closed"); + using var client = new HttpClient(); + + var (text, isError) = Unwrap(await SessionlessStatusAsync(server, client, new JsonObject())); + + await Assert.That(isError).IsFalse(); + await Assert.That(text).Contains("flow_run_id: flow-open"); + } + + [Test] + public async Task A_just_recorded_run_the_server_cannot_read_yet_is_a_retry_not_a_skip() { + using var server = WireMockServer.Start(); + GivenRecordedRunsAt(DateTimeOffset.UtcNow.AddHours(-1), "flow-older"); + GivenRecordedRunsAt(DateTimeOffset.UtcNow, "flow-just-started"); + GivenFlow(server, "flow-older", "waiting"); + server.Given(Request.Create().WithPath("/api/flows/flow-just-started").UsingGet()) + .RespondWith(Response.Create().WithStatusCode(404)); + using var client = new HttpClient(); + + var (text, isError) = Unwrap(await SessionlessStatusAsync(server, client, new JsonObject())); + + await Assert.That(isError).IsTrue(); + await Assert.That(text).Contains("flow-just-started"); + await Assert.That(text).Contains("retry"); + await Assert.That(text).DoesNotContain("flow-older"); + } + [Test] public async Task A_start_without_a_session_records_its_run_for_the_workspace() { using var server = WireMockServer.Start(); @@ -416,7 +454,7 @@ await Server().HandleToolCallAsync( JsonNode.Parse("1")!, ToolCallRequest("start_review_flow", StartArguments()), client, server.Url!, cwd: "/tmp/cwd", repoRoot: Workspace, repoInfo: null, requestingSessionId: null); - await Assert.That(new FlowRunLedger(Config.Root, TimeProvider.System).Recent(Workspace, 5)) + await Assert.That(new FlowRunLedger(Config.Root, TimeProvider.System).Retained(Workspace).Select(e => e.FlowRunId)) .IsEquivalentTo(["flow-started"]); } @@ -432,7 +470,7 @@ await Server().HandleToolCallAsync( JsonNode.Parse("1")!, ToolCallRequest("start_review_flow", StartArguments()), client, server.Url!, cwd: "/tmp/cwd", repoRoot: Workspace, repoInfo: null, requestingSessionId: SessionId); - await Assert.That(new FlowRunLedger(Config.Root, TimeProvider.System).Recent(Workspace, 5)).IsEmpty(); + await Assert.That(new FlowRunLedger(Config.Root, TimeProvider.System).Retained(Workspace)).IsEmpty(); } static JsonObject StartArguments() => new() {