diff --git a/backend/src/CodeSpace.Core/Services/Agents/Exceptions/AgentRunEndedException.cs b/backend/src/CodeSpace.Core/Services/Agents/Exceptions/AgentRunEndedException.cs new file mode 100644 index 000000000..c6a23a318 --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Agents/Exceptions/AgentRunEndedException.cs @@ -0,0 +1,16 @@ +using CodeSpace.Messages.Failures; + +namespace CodeSpace.Core.Services.Agents.Exceptions; + +/// +/// A call reached an agent run that has already ended — succeeded, failed, been held for review, stopped or recovered — +/// as a NEW tool call to claim, or as a decision claimed before the end that was about to be parked after it. Nothing +/// reads its result any more, so it is refused before anything is recorded or shown to a human; the MCP handler answers +/// it as final, never as something to retry. +/// +public sealed class AgentRunEndedException : Exception, IFailure +{ + public FailureKind Kind => FailureKind.Conflict; + public string Code => FailureCodes.AgentRunEnded; + public AgentRunEndedException(Guid runId) : base($"Agent run {runId} has already ended; it takes no new tool call.") { } +} diff --git a/backend/src/CodeSpace.Core/Services/Agents/Mcp/IToolCallLedgerService.cs b/backend/src/CodeSpace.Core/Services/Agents/Mcp/IToolCallLedgerService.cs index 07f8fdf58..ffafebb1f 100644 --- a/backend/src/CodeSpace.Core/Services/Agents/Mcp/IToolCallLedgerService.cs +++ b/backend/src/CodeSpace.Core/Services/Agents/Mcp/IToolCallLedgerService.cs @@ -8,6 +8,7 @@ using CodeSpace.Messages.Decisions; using CodeSpace.Messages.Enums; using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Storage; using Microsoft.Extensions.Logging; namespace CodeSpace.Core.Services.Agents.Mcp; @@ -48,13 +49,19 @@ public interface IToolCallLedgerService /// leaks the refusal as an ungoverned side effect. ( only pre-checks /// Guard.CheckAsync before delegating to the inner handler — it has no catch of its own.) /// + /// + /// + /// Under the same lock it reads whether the run has ENDED, because a run's own end keeps its epoch: a NEW claim on an + /// ended run throws , which the dispatch answers as final, never as a + /// retry. An existing key still replays as above — the refusal is only for a call the run never made. + /// /// Task TryClaimAsync(Guid agentRunId, Guid teamId, string toolKind, string idempotencyKey, string inputHash, long fenceEpoch, CancellationToken cancellationToken); /// Status-guarded CAS Pending → terminal (mirrors completion), team-scoped (defense-in-depth — the design mandates all reads team-scoped). Stores the ALREADY-REDACTED result/error. Throws when the transition is illegal or lost the CAS. Task RecordTerminalAsync(Guid ledgerId, Guid teamId, ToolCallLedgerStatus status, string? resultJson, string? error, CancellationToken cancellationToken); - /// Status-guarded CAS Pending → AwaitingApproval (durable mid-turn HITL, item D2 — mirrors 's discipline), team-scoped. Stamps ApprovalToken + ApprovalDeadlineAt so the row is resolvable BEFORE the card posts (the token is the authority; the card's message id is a best-effort follow-up), and in the same write the preview the card is built from and the target a rejection sticks to — so a row awaiting approval never lacks the pins its card showed. Returns false when the CAS is lost (the row already moved — e.g. a re-claim of an already-parked call), so the handler re-reads + re-blocks rather than posting a second card. + /// Status-guarded CAS Pending → AwaitingApproval (durable mid-turn HITL, item D2 — mirrors 's discipline), team-scoped. Stamps ApprovalToken + ApprovalDeadlineAt so the row is resolvable BEFORE the card posts (the token is the authority; the card's message id is a best-effort follow-up), and in the same write the preview the card is built from and the target a rejection sticks to — so a row awaiting approval never lacks the pins its card showed. Returns false when the CAS is lost (the row already moved — e.g. a re-claim of an already-parked call, or the run's end closed it), so the handler re-reads + re-blocks rather than posting a second card. Never parks a row whose run has ended: one still Pending throws . Task TryBeginApprovalAsync(Guid ledgerId, Guid teamId, ToolCallApprovalPark park, CancellationToken cancellationToken); /// Whether a reviewer rejected a call on earlier in — a row of that run and target that a rejection failed (). Team-scoped. A target expired unanswered was not rejected. @@ -208,15 +215,17 @@ public async Task TryClaimAsync(Guid agentRunId, Guid teamId, str // caller, whose epoch is always >= 1 once Running). await using var transaction = await _db.Database.BeginTransactionAsync(cancellationToken).ConfigureAwait(false); - var lockedEpoch = (await _db.Database.SqlQuery($"SELECT fence_epoch AS \"Value\" FROM agent_run WHERE id = {agentRunId} FOR SHARE") - .ToListAsync(cancellationToken).ConfigureAwait(false)).FirstOrDefault(); + var run = await LockRunForClaimAsync(agentRunId, cancellationToken).ConfigureAwait(false); - if (!AgentRunFence.StillOwns(lockedEpoch, fenceEpoch)) + if (run.HasEnded) return await ReplayOnEndedRunAsync(transaction, agentRunId, toolKind, idempotencyKey, fenceEpoch, cancellationToken).ConfigureAwait(false); + + // An ended run answers "ended" whatever ended it: a stop, abandon or recovery bumps the epoch too, so the fence + // checked first would report it as lost ownership, which the handler answers as a retry. + if (!AgentRunFence.StillOwns(run.FenceEpoch, fenceEpoch)) { - await transaction.RollbackAsync(cancellationToken).ConfigureAwait(false); - _db.ChangeTracker.Clear(); + await AbandonClaimAsync(transaction, cancellationToken).ConfigureAwait(false); - _logger.LogWarning("Agent run {RunId}: {Note}", agentRunId, AgentRunFence.RefusalNote($"tool-call claim ({toolKind})", lockedEpoch, fenceEpoch)); + _logger.LogWarning("Agent run {RunId}: {Note}", agentRunId, AgentRunFence.RefusalNote($"tool-call claim ({toolKind})", run.FenceEpoch, fenceEpoch)); throw new AgentRunOwnershipLostException(agentRunId); } @@ -242,6 +251,43 @@ public async Task TryClaimAsync(Guid agentRunId, Guid teamId, str } } + /// The owning run as a claim sees it: its fence epoch, and whether it has ended. + private sealed record ClaimedRun(long FenceEpoch, bool HasEnded); + + /// + /// Read the run's fence epoch and whether it has ended, together, under the claim's FOR SHARE. A run's own + /// completion keeps its epoch, so the fence alone let a call already past the authority guard claim on a run that had + /// just finished; a stop, abandon or recovery bumps it, so the end is checked first — the run's end is the answer. Every terminal write takes this row FOR UPDATE, so it either committed first (this + /// read sees the run ended) or waits for this claim to commit and then closes the row the claim inserted along with + /// the run's other open questions. A run with no row reads epoch 0 and not ended. + /// + private async Task LockRunForClaimAsync(Guid agentRunId, CancellationToken cancellationToken) => + (await _db.Database.SqlQuery($"SELECT fence_epoch, status = ANY({TerminalAgentRunStatusNames}) AS has_ended FROM agent_run WHERE id = {agentRunId} FOR SHARE").ToListAsync(cancellationToken).ConfigureAwait(false)).FirstOrDefault() ?? new ClaimedRun(0, false); + + /// + /// A run that has ended takes no NEW call: nothing reads its result any more, and a question claimed now would wait in + /// the queue of a finished run, answerable into nothing. A call it already made still replays exactly as on a live run + /// — its recorded result, or the close its question got — so the answer an identical re-call gets never changes. + /// + private async Task ReplayOnEndedRunAsync(IDbContextTransaction transaction, Guid agentRunId, string toolKind, string idempotencyKey, long fenceEpoch, CancellationToken cancellationToken) + { + await AbandonClaimAsync(transaction, cancellationToken).ConfigureAwait(false); + + if (await _db.ToolCallLedger.AsNoTracking().AnyAsync(l => l.AgentRunId == agentRunId && l.IdempotencyKey == idempotencyKey, cancellationToken).ConfigureAwait(false)) + return await ReadExistingClaimAsync(agentRunId, idempotencyKey, fenceEpoch, cancellationToken).ConfigureAwait(false); + + _logger.LogWarning("Agent run {RunId} has ended; refused a new {ToolKind} tool call", agentRunId, toolKind); + + throw new AgentRunEndedException(agentRunId); + } + + /// Roll back a claim that will not insert, and drop the Pending row it staged. + private async Task AbandonClaimAsync(IDbContextTransaction transaction, CancellationToken cancellationToken) + { + await transaction.RollbackAsync(cancellationToken).ConfigureAwait(false); + _db.ChangeTracker.Clear(); + } + public async Task RecordTerminalAsync(Guid ledgerId, Guid teamId, ToolCallLedgerStatus status, string? resultJson, string? error, CancellationToken cancellationToken) { if (!ToolCallLedgerStateMachine.IsTerminal(status)) @@ -278,16 +324,44 @@ public async Task RecordTerminalAsync(Guid ledgerId, Guid teamId, ToolCallLedger } public async Task TryBeginApprovalAsync(Guid ledgerId, Guid teamId, ToolCallApprovalPark park, CancellationToken cancellationToken) + { + if (await ParkUnlessRunEndedAsync(ledgerId, teamId, park, cancellationToken).ConfigureAwait(false)) + { + _logger.LogInformation("Tool call ledger awaiting approval. LedgerId={LedgerId}", ledgerId); + + return true; + } + + // Still Pending after a lost CAS can only mean its run has ended: nothing moves a row back to Pending, so the + // status guard did not stop it, and the run predicate did — and an ended run stays ended. + if (await PendingRowRunAsync(ledgerId, teamId, cancellationToken).ConfigureAwait(false) is { } runId) + throw new AgentRunEndedException(runId); + + return false; + } + + /// + /// Status-guarded CAS Pending → AwaitingApproval (mirrors RecordTerminalAsync's ExecuteUpdate discipline), unless the + /// row's run has ended. The Status == Pending guard is the single-winner: a concurrent transition (a re-claim that + /// already parked, a racing terminal) leaves the row not-Pending so this update affects 0 rows, and the caller re-reads + /// + re-blocks instead of posting a second card. The preview and target ride the same write: a parked row always + /// carries the pins its card showed. + /// + /// The run predicate READS the run and never locks it. Every terminal write locks the run's questions and then + /// the run (), so a park that locked the run and then waited on its question would + /// hold what the end waits for while waiting on what the end holds — a deadlock. This UPDATE holds only its own row, + /// for its own statement: a park that queues behind an end holding that row re-checks it once the end commits and + /// finds it Expired (the end closes Pending questions too); one that runs after the end sees the run ended. Only a + /// park whose read began before a stop's flip committed can still land. For a decision.request row the stop's own + /// close, which runs after its flip, then expires it; a write-approval row parked in that window waits for the + /// deadline reaper. + /// + private async Task ParkUnlessRunEndedAsync(Guid ledgerId, Guid teamId, ToolCallApprovalPark park, CancellationToken cancellationToken) { var now = DateTimeOffset.UtcNow; - // Status-guarded CAS Pending → AwaitingApproval (mirrors RecordTerminalAsync's ExecuteUpdate discipline). The - // Status == Pending guard is the single-winner: a concurrent transition (a re-claim that already parked, a - // racing terminal) leaves the row not-Pending so this update affects 0 rows → false, and the caller re-reads - // + re-blocks instead of posting a second card. The preview and target ride the same write: a parked row always - // carries the pins its card showed. var flipped = await _db.ToolCallLedger - .Where(l => l.Id == ledgerId && l.TeamId == teamId && l.Status == ToolCallLedgerStatus.Pending) + .Where(l => l.Id == ledgerId && l.TeamId == teamId && l.Status == ToolCallLedgerStatus.Pending && !_db.AgentRun.Any(r => r.Id == l.AgentRunId && TerminalAgentRunStatuses.Contains(r.Status))) .ExecuteUpdateAsync(s => s .SetProperty(l => l.Status, ToolCallLedgerStatus.AwaitingApproval) .SetProperty(l => l.ApprovalToken, park.Token) @@ -297,11 +371,16 @@ public async Task TryBeginApprovalAsync(Guid ledgerId, Guid teamId, ToolCa .SetProperty(l => l.LastModifiedDate, now), cancellationToken) .ConfigureAwait(false); - if (flipped > 0) _logger.LogInformation("Tool call ledger awaiting approval. LedgerId={LedgerId}", ledgerId); - return flipped > 0; } + /// The run of while that row is still Pending (team-scoped), else null. + private async Task PendingRowRunAsync(Guid ledgerId, Guid teamId, CancellationToken cancellationToken) => + await _db.ToolCallLedger.AsNoTracking() + .Where(l => l.Id == ledgerId && l.TeamId == teamId && l.Status == ToolCallLedgerStatus.Pending) + .Select(l => (Guid?)l.AgentRunId) + .SingleOrDefaultAsync(cancellationToken).ConfigureAwait(false); + public async Task SetApprovalMessageAsync(Guid ledgerId, Guid teamId, Guid messageId, CancellationToken cancellationToken) => // Best-effort, idempotent (the approval_message_id IS NULL guard → exactly one card recorded per row). The // token + deadline stamped by TryBeginApprovalAsync already make the row resolvable, so a lost CAS is harmless. @@ -452,6 +531,9 @@ private async Task TryExpireOneAsync(Guid ledgerId, DateTimeOffset now, Ca /// The terminal set, derived ONCE from the state machine so the stranded-reaper predicate can't drift from (mirrors the StuckRunReconciler's NonTerminalDecisionStatuses discipline). private static readonly AgentRunStatus[] TerminalAgentRunStatuses = Enum.GetValues().Where(AgentRunStateMachine.IsTerminal).ToArray(); + /// The same set as stored, for the raw SQL of the claim's status read. + private static readonly string[] TerminalAgentRunStatusNames = TerminalAgentRunStatuses.Select(status => status.ToString()).ToArray(); + public async Task ExpireStaleToolCallsAsync(DateTimeOffset now, CancellationToken cancellationToken) { // Candidate set (bounded): a side-effecting execution row left non-terminal whose owning run is BOTH already diff --git a/backend/src/CodeSpace.Core/Services/Agents/Mcp/McpRequestHandler.cs b/backend/src/CodeSpace.Core/Services/Agents/Mcp/McpRequestHandler.cs index af24b3caf..718de92b9 100644 --- a/backend/src/CodeSpace.Core/Services/Agents/Mcp/McpRequestHandler.cs +++ b/backend/src/CodeSpace.Core/Services/Agents/Mcp/McpRequestHandler.cs @@ -82,6 +82,13 @@ public sealed class McpRequestHandler : IMcpRequestHandler /// public const string InterruptedToolCallError = "This tool call was interrupted before it completed (the tool timed out, the run was cancelled, or the worker running it was lost), so whether its effect was applied is unknown; it may have been. It is recorded as failed: re-issuing it with identical arguments returns this same result without running it again. Check whether it took effect (for example, read back the PR or comment it would have created or changed) before re-issuing it with changed arguments."; + /// + /// The answer to a call that reached a run which has already ended (): a new call + /// it would claim, or a decision it claimed before the end and would park after it. Nothing was recorded or run, and + /// nothing this run asks can take effect any more, so unlike the governed-fault answer it must never invite a retry. + /// + public const string RunEndedToolCallError = "The agent run making this call has already ended, so the call was refused and nothing ran. Calling it again cannot help: nothing this run asks can take effect any more."; + /// /// The answer — recorded as the row's Denied reason — to a call whose target a reviewer already rejected in this run: /// the same tool on the same target (a merge: the same repository and pull request at the same head and base), however @@ -241,6 +248,14 @@ private async Task DispatchToolCallAsync(JsonElement id, JsonEl { return await HandleToolCallAsync(id, request, cancellationToken).ConfigureAwait(false); } + catch (AgentRunEndedException ended) + { + // Every entry point meets the run's end here — a claim of a decision, of an allowed write or of a write parked + // for approval, or the park of a decision claimed before the end. Nothing ran and nothing can: final, not "retry". + _logger.LogWarning(ended, "Agent run {RunId}: tools/call for tool {ToolName} refused — the run has already ended", _runId, ReadToolNameOrNull(request) ?? "(unknown)"); + + return JsonRpcResponse.Ok(id, ToolResult(isError: true, RunEndedToolCallError)); + } catch (Exception ex) when (ex is not OperationCanceledException) { _logger.LogWarning(ex, "Agent run {RunId}: tools/call dispatch threw for tool {ToolName}; degrading to a retryable governed-error result", _runId, ReadToolNameOrNull(request) ?? "(unknown)"); @@ -730,7 +745,7 @@ private async Task RunDecisionFlowAsync(IAgentTool tool, JsonElemen return claim.Outcome switch { ToolCallClaimOutcome.Duplicate => WrapDecisionResult(claim.PriorStatus, claim.PriorResultJson, claim.PriorError), // already answered/expired — replay (AC1) - ToolCallClaimOutcome.InFlight => await ResumeDecisionOrTicketAsync(teamId, claim.LedgerId, cancellationToken).ConfigureAwait(false), // a re-call of a still-parked decision — never a 2nd card + ToolCallClaimOutcome.InFlight => await ResumeDecisionOrTicketAsync(arguments, key, teamId, claim.LedgerId, cancellationToken).ConfigureAwait(false), // a re-call of a still-open decision — never a 2nd card _ => await ParkDecisionAsync(arguments, key, teamId, claim.LedgerId, cancellationToken).ConfigureAwait(false), // fresh claim — park + post + block }; } @@ -746,7 +761,8 @@ private bool HasDecisionCardSurface() => /// /// A FRESH decision claim's park: CAS Pending → AwaitingApproval (stamping token + the bounded deadline), post the /// typed-options card (stamping its message id), then BLOCK until answered. If the park CAS is lost (a concurrent - /// path already parked the row), DON'T post — re-bind to whatever it became (the no-second-card guard). + /// path already parked the row, or the run's end closed it), DON'T post — re-bind to whatever it became (the + /// no-second-card guard). A row its ended run has not closed yet is refused by the park itself. /// private async Task ParkDecisionAsync(JsonElement arguments, string key, Guid teamId, Guid ledgerId, CancellationToken cancellationToken) { @@ -755,7 +771,7 @@ private async Task ParkDecisionAsync(JsonElement arguments, string var parked = await _ledger!.TryBeginApprovalAsync(ledgerId, teamId, new ToolCallApprovalPark { Token = token, DeadlineAt = deadlineAt }, cancellationToken).ConfigureAwait(false); - if (!parked) return await ResumeDecisionOrTicketAsync(teamId, ledgerId, cancellationToken).ConfigureAwait(false); + if (!parked) return await RebindDecisionAsync(teamId, ledgerId, cancellationToken).ConfigureAwait(false); var request = BuildDecisionRequest(arguments, ledgerId, key, deadlineAt); @@ -777,8 +793,23 @@ private async Task ParkDecisionAsync(JsonElement arguments, string return await BlockForDecisionAnswerAsync(teamId, ledgerId, cancellationToken).ConfigureAwait(false); } + /// + /// A re-call of a decision still open on its row. One still Pending was claimed and never parked — its park faulted and + /// its call was told to retry — and nothing else parks it, nor does any queue show it: blocking on it waited for an + /// answer nobody could give, and came back with a ticket on every re-call. The re-call parks it instead, posting its one + /// card; the park's CAS lets exactly one racing re-call win. A parked one is re-bound without a second card. + /// + private async Task ResumeDecisionOrTicketAsync(JsonElement arguments, string key, Guid teamId, Guid ledgerId, CancellationToken cancellationToken) + { + var state = await _ledger!.ReadApprovalStateAsync(ledgerId, teamId, cancellationToken).ConfigureAwait(false); + + if (state is { Status: ToolCallLedgerStatus.Pending }) return await ParkDecisionAsync(arguments, key, teamId, ledgerId, cancellationToken).ConfigureAwait(false); + + return await RebindDecisionAsync(teamId, ledgerId, cancellationToken).ConfigureAwait(false); + } + /// Re-bind to an already-parked (or just-resolved) decision WITHOUT posting a second card: a terminal row replays its answer; a still-AwaitingApproval one blocks again on a freshly-registered waiter (the card was posted on the first park). - private async Task ResumeDecisionOrTicketAsync(Guid teamId, Guid ledgerId, CancellationToken cancellationToken) + private async Task RebindDecisionAsync(Guid teamId, Guid ledgerId, CancellationToken cancellationToken) { var state = await _ledger!.ReadApprovalStateAsync(ledgerId, teamId, cancellationToken).ConfigureAwait(false); diff --git a/backend/src/CodeSpace.Core/Services/Agents/StoppedRunDecisions.cs b/backend/src/CodeSpace.Core/Services/Agents/StoppedRunDecisions.cs index b9cf47917..c6e9ff0a1 100644 --- a/backend/src/CodeSpace.Core/Services/Agents/StoppedRunDecisions.cs +++ b/backend/src/CodeSpace.Core/Services/Agents/StoppedRunDecisions.cs @@ -8,8 +8,8 @@ namespace CodeSpace.Core.Services.Agents; /// /// Closes the decisions a stopped agent run left unanswered. An agent raises a decision.request mid-run and -/// blocks on it as an AwaitingApproval ledger row; a cancel or the reconciler's abandon ends the run, but nothing ended -/// that row, so the question stayed in the team queue and on the Room — answerable into a run that no longer exists, +/// blocks on it as an AwaitingApproval ledger row — Pending between its claim and its park; a cancel or the reconciler's +/// abandon ends the run, but nothing ended that row, so the question stayed in the team queue and on the Room — answerable into a run that no longer exists, /// and for a human-required decision forever, since the decision reaper only ever defers those. The stop closes them /// here instead, as : the no-decision terminal every pending-decision reader /// already excludes. A run that ends on its own — its completion, or its recovery from the spool — closes them the same @@ -80,9 +80,11 @@ public static async Task LockAsync(CodeSpaceDbContext db, Guid agentRunId, Cance /// /// The CAS itself: every still-unanswered decision.request row raised → Expired, - /// stamped with . It locks the rows in id order before it writes them, in the one statement — - /// the order takes them in; a plain UPDATE locks them in whatever order its scan meets them. + /// stamped with — the parked ones, and the ones claimed but not yet parked, which would + /// otherwise survive the end Pending and then park on the finished run. It locks the rows in id order before it writes + /// them, in the one statement — the order takes them in; a plain UPDATE locks them in whatever + /// order its scan meets them. /// public static async Task ExpireAsync(CodeSpaceDbContext db, Guid agentRunId, string reason, CancellationToken cancellationToken) => - await db.Database.ExecuteSqlInterpolatedAsync($"WITH locked AS MATERIALIZED (SELECT id FROM tool_call_ledger WHERE agent_run_id = {agentRunId} AND tool_kind = {DecisionToolKinds.DecisionRequest} AND status = {nameof(ToolCallLedgerStatus.AwaitingApproval)} AND approved_at IS NULL ORDER BY id FOR UPDATE) UPDATE tool_call_ledger AS target SET status = {nameof(ToolCallLedgerStatus.Expired)}, error = {reason}, last_modified_date = {DateTimeOffset.UtcNow} FROM locked WHERE target.id = locked.id AND target.status = {nameof(ToolCallLedgerStatus.AwaitingApproval)} AND target.approved_at IS NULL", cancellationToken).ConfigureAwait(false); + await db.Database.ExecuteSqlInterpolatedAsync($"WITH locked AS MATERIALIZED (SELECT id FROM tool_call_ledger WHERE agent_run_id = {agentRunId} AND tool_kind = {DecisionToolKinds.DecisionRequest} AND status IN ({nameof(ToolCallLedgerStatus.Pending)}, {nameof(ToolCallLedgerStatus.AwaitingApproval)}) AND approved_at IS NULL ORDER BY id FOR UPDATE) UPDATE tool_call_ledger AS target SET status = {nameof(ToolCallLedgerStatus.Expired)}, error = {reason}, last_modified_date = {DateTimeOffset.UtcNow} FROM locked WHERE target.id = locked.id AND target.status IN ({nameof(ToolCallLedgerStatus.Pending)}, {nameof(ToolCallLedgerStatus.AwaitingApproval)}) AND target.approved_at IS NULL", cancellationToken).ConfigureAwait(false); } diff --git a/backend/src/CodeSpace.Core/Services/Agents/ToolCallLedgerStateMachine.cs b/backend/src/CodeSpace.Core/Services/Agents/ToolCallLedgerStateMachine.cs index 6135b75c5..c81d6808e 100644 --- a/backend/src/CodeSpace.Core/Services/Agents/ToolCallLedgerStateMachine.cs +++ b/backend/src/CodeSpace.Core/Services/Agents/ToolCallLedgerStateMachine.cs @@ -7,7 +7,8 @@ namespace CodeSpace.Core.Services.Agents; /// (Succeeded | Failed | Denied) for a synchronously-resolved call, or /// Pending → AwaitingApproval → Running → (Succeeded | Failed) for a /// durable mid-turn approval that is approved then executed, or AwaitingApproval → Expired when the -/// reaper expires an undecided approval (item D). The Running hop is the single-winner execution claim: out of +/// reaper expires an undecided approval (item D) — or Pending | AwaitingApproval → Expired when a +/// run ends with a decision it raised still unanswered, claimed or parked. The Running hop is the single-winner execution claim: out of /// AwaitingApproval exactly one executor flips the row to Running BEFORE running the side effect, so a /// concurrent racer that lost the CAS never re-runs it. The single source of truth for which transitions are legal, so /// the service's status-guarded CAS can REJECT an illegal/lost transition rather than silently apply it. Mirrors @@ -21,7 +22,7 @@ public static bool IsTerminal(ToolCallLedgerStatus status) => /// /// True when moving → is allowed: a call may succeed / fail / - /// be denied / pause for approval out of Pending; out of AwaitingApproval it may be claimed for + /// be denied / pause for approval / expire with its ended run out of Pending; out of AwaitingApproval it may be claimed for /// execution (→ Running) or expire (the reaper); out of Running it succeeds / fails. A synchronous /// (Pending) call still resolves straight to its terminal without the Running hop. A denial is a /// synchronous-only verdict, an approval can never re-pend. Nothing transitions out of a terminal, and @@ -40,7 +41,8 @@ public static bool IsTerminal(ToolCallLedgerStatus status) => ToolCallLedgerStatus.Denied => from == ToolCallLedgerStatus.Pending, ToolCallLedgerStatus.AwaitingApproval => from == ToolCallLedgerStatus.Pending, ToolCallLedgerStatus.Running => from == ToolCallLedgerStatus.AwaitingApproval, - ToolCallLedgerStatus.Expired => from == ToolCallLedgerStatus.AwaitingApproval, + // A run's end closes a decision it claimed but had not parked yet as well as a parked one (StoppedRunDecisions). + ToolCallLedgerStatus.Expired => from is ToolCallLedgerStatus.Pending or ToolCallLedgerStatus.AwaitingApproval, _ => false, }; } diff --git a/backend/src/CodeSpace.Messages/Failures/FailureCodes.cs b/backend/src/CodeSpace.Messages/Failures/FailureCodes.cs index f62cd4ddc..dc6601954 100644 --- a/backend/src/CodeSpace.Messages/Failures/FailureCodes.cs +++ b/backend/src/CodeSpace.Messages/Failures/FailureCodes.cs @@ -19,6 +19,7 @@ namespace CodeSpace.Messages.Failures; public static class FailureCodes { public const string AgentRunOwnershipLost = "agent-run-ownership-lost"; + public const string AgentRunEnded = "agent-run-ended"; public const string AgentRunLaunchAcknowledgementLost = "agent-run-launch-acknowledgement-lost"; public const string AgentAuthorityDenied = "agent.authority_denied"; // ── Identity and access ──────────────────────────────────────────────────────── diff --git a/backend/tests/CodeSpace.IntegrationTests/Agents/AgentMcpEndpointFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Agents/AgentMcpEndpointFlowTests.cs index b4efe62e9..57ef98206 100644 --- a/backend/tests/CodeSpace.IntegrationTests/Agents/AgentMcpEndpointFlowTests.cs +++ b/backend/tests/CodeSpace.IntegrationTests/Agents/AgentMcpEndpointFlowTests.cs @@ -1,3 +1,4 @@ +using System.Data.Common; using System.Diagnostics; using System.Net.Sockets; using System.Text; @@ -19,6 +20,7 @@ using CodeSpace.Messages.Dtos.Decisions; using CodeSpace.Messages.Enums; using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging.Abstractions; using Shouldly; @@ -1189,6 +1191,155 @@ public async Task A_real_agent_raising_a_decision_parks_blocks_is_answered_then_ } } + // ── A question asked as the run ends: answered that the run ended, never left in the queue ── + + /// Where the endpoint holds an agent's question while its run ends: past the authority guard but before the claim reads the run, or after the claim but before the park. + public enum HeldBefore + { + Claim, + Park, + } + + [Theory] + [InlineData(HeldBefore.Claim)] + [InlineData(HeldBefore.Park)] + public async Task A_question_asked_as_the_run_ends_is_answered_that_the_run_ended_and_never_reaches_the_queue(HeldBefore held) + { + // 🟢 High fidelity: the real executor, its real per-run UDS endpoint and the real handler over real Postgres. The + // endpoint holds the agent's decision.request while the agent exits and the executor lands the run's own end — + // its CompleteAsync under the owner token it claimed — and is then held just past that commit, the endpoint still + // open. A run's end keeps its fence, so the question used to claim, park and sit in the queue of a finished run. + if (OperatingSystem.IsWindows()) return; + if (!Socket.OSSupportsUnixDomainSockets) return; + + var teamId = await SeedTeamAsync(); + var runId = await CreateRunAsync(teamId, AgentAutonomyLevel.Standard); + using var gate = new TempDir(); + var agentExits = Path.Combine(gate.Path, "exit"); + var question = new HoldQuestion(held); + var end = new HoldPastTheEnd(); + using var connects = ConnectRegistryFromGovernanceContainer(); + + var run = Task.Run(() => ExecuteAsync(runId, new ScriptedHarness($"while [ ! -e '{agentExits}' ]; do sleep 0.05; done"), useGovernanceContainer: true, interceptors: [question, end])); + + try + { + var connect = await WaitForConnectAsync(connects, runId, run); + await using var client = await McpClient.ConnectAsync(connect); + + var ask = client.CallToolAsync(1, DecisionRequestTool.ToolKind, new { question = "Ship it?", decisionType = "choose_one", options = new[] { new { id = "a", label = "Ship" }, new { id = "b", label = "Hold" } } }, TimeSpan.FromSeconds(60)); + await question.Reached.WaitAsync(TimeSpan.FromSeconds(20)); + + await File.WriteAllTextAsync(agentExits, ""); + await end.Reached.WaitAsync(TimeSpan.FromSeconds(30)); + question.Release(); + + var answer = await ask; + + answer.GetProperty("isError").GetBoolean().ShouldBeTrue(); + Text(answer).ShouldBe(held == HeldBefore.Claim ? McpRequestHandler.RunEndedToolCallError : StoppedRunDecisions.EndedError, "the agent is told its run ended — not to retry, and not that its question awaits anyone"); + (await ListPendingDecisionsAsync(teamId)).ShouldBeEmpty("no question of a finished run reaches the queue"); + + var rows = await ReadDecisionRowsAsync(runId); + if (held == HeldBefore.Claim) rows.ShouldBeEmpty("the claim found the run ended and recorded nothing"); + else rows.ShouldHaveSingleItem().Status.ShouldBe(ToolCallLedgerStatus.Expired, "the end closed the claimed question before its park ran"); + + end.Release(); + await run.WaitAsync(TimeSpan.FromSeconds(60)); + (await ReadAgentRunStatusAsync(runId)).ShouldBe(held == HeldBefore.Claim ? AgentRunStatus.Succeeded : AgentRunStatus.NeedsReview, "a question claimed before the end holds the run for review; one refused at its claim does not"); + } + finally + { + question.Release(); + end.Release(); + if (!File.Exists(agentExits)) await File.WriteAllTextAsync(agentExits, ""); + await Record.ExceptionAsync(() => run.WaitAsync(TimeSpan.FromSeconds(60))); + } + } + + private async Task> ListPendingDecisionsAsync(Guid teamId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().ListPendingAsync(teamId, CancellationToken.None); + } + + private async Task> ReadDecisionRowsAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().ToolCallLedger.AsNoTracking().Where(l => l.AgentRunId == runId && l.ToolKind == DecisionRequestTool.ToolKind).ToListAsync(); + } + + /// Holds the endpoint's database work on the question at — the claim's locked read of the run, or the park's write — until released. + private sealed class HoldQuestion : DbCommandInterceptor + { + private readonly HeldBefore _held; + private readonly TaskCompletionSource _reached = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _released = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _fired; + + public HoldQuestion(HeldBefore held) { _held = held; } + + public Task Reached => _reached.Task; + + public void Release() => _released.TrySetResult(); + + public override async ValueTask> ReaderExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) + { + if (_held == HeldBefore.Claim && IsClaimRead(command)) await HoldAsync(); + + return result; + } + + public override async ValueTask> NonQueryExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) + { + if (_held == HeldBefore.Park && IsPark(command)) await HoldAsync(); + + return result; + } + + private static bool IsClaimRead(DbCommand command) => command.CommandText.Contains("FROM agent_run", StringComparison.Ordinal) && command.CommandText.Contains("FOR SHARE", StringComparison.Ordinal); + + private static bool IsPark(DbCommand command) => command.CommandText.Contains("UPDATE tool_call_ledger", StringComparison.Ordinal) && command.CommandText.Contains("approval_token", StringComparison.Ordinal); + + private async Task HoldAsync() + { + if (Interlocked.Exchange(ref _fired, 1) == 1) return; + + _reached.TrySetResult(); + + await _released.Task.WaitAsync(TimeSpan.FromSeconds(30)); + } + } + + /// Holds the executor just past the commit of the run's own terminal write — the run has ended, its endpoint still open — until released. + private sealed class HoldPastTheEnd : IDbCommandInterceptor, IDbTransactionInterceptor + { + private readonly TaskCompletionSource _reached = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _released = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _terminalWritten; + private int _fired; + + public Task Reached => _reached.Task; + + public void Release() => _released.TrySetResult(); + + public ValueTask NonQueryExecutedAsync(DbCommand command, CommandExecutedEventData eventData, int result, CancellationToken cancellationToken = default) + { + if (command.CommandText.Contains("UPDATE agent_run", StringComparison.Ordinal) && command.CommandText.Contains("completed_at", StringComparison.Ordinal)) Interlocked.Exchange(ref _terminalWritten, 1); + + return ValueTask.FromResult(result); + } + + public async Task TransactionCommittedAsync(DbTransaction transaction, TransactionEndEventData eventData, CancellationToken cancellationToken = default) + { + if (Volatile.Read(ref _terminalWritten) == 0 || Interlocked.Exchange(ref _fired, 1) == 1) return; + + _reached.TrySetResult(); + + await _released.Task.WaitAsync(TimeSpan.FromSeconds(30)); + } + } + /// The team's owner user id — the human actor that answers the agent's decision via the queue. private async Task TeamOwnerAsync(Guid teamId) { @@ -1260,7 +1411,7 @@ private async Task ReadAgentRunStatusAsync(Guid runId) /// at a real existing stand-in (the test only File.Exists-checks it; the scripted harness runs /bin/sh, not the proxy); /// when false we point it at a missing path to exercise the fail-closed "no declaration" branch. /// - private async Task ExecuteAsync(Guid runId, IAgentHarness harness, bool proxyPresent = true, bool useGovernanceContainer = false, string? proxyPath = null, CancellationToken cancellationToken = default, ISandboxRunner? commandRunner = null) + private async Task ExecuteAsync(Guid runId, IAgentHarness harness, bool proxyPresent = true, bool useGovernanceContainer = false, string? proxyPath = null, CancellationToken cancellationToken = default, ISandboxRunner? commandRunner = null, IInterceptor[]? interceptors = null) { // The catalog choice rides the RUN now (CreateRunAsync's enableMcp) and governance is a committed constant, so // the only environment this still drives is the proxy path — a genuine filesystem seam, not a feature flag. @@ -1277,7 +1428,10 @@ private async Task ExecuteAsync(Guid runId, IAgentHarness harness, bool proxyPre // A command runner, when given, is what the scopes the executor makes for itself — the MCP endpoint's among // them, and so every tool call's — run commands on. The harness keeps the real runner the executor is handed. using var toolScope = commandRunner is null ? null : scope.BeginLifetimeScope(b => b.RegisterInstance(new SandboxRunnerRegistry([commandRunner])).As()); - await NewExecutor(scope, harness, toolScope).ExecuteAsync(runId, cancellationToken); + + // Interceptors, when given, see every database command and transaction of the executor and of the scopes it makes. + using var intercepted = interceptors is null ? null : scope.BeginLifetimeScope(b => Intercept(b, scope, interceptors)); + await NewExecutor(intercepted ?? scope, harness, intercepted ?? toolScope).ExecuteAsync(runId, cancellationToken); } finally { @@ -1285,6 +1439,15 @@ private async Task ExecuteAsync(Guid runId, IAgentHarness harness, bool proxyPre } } + private static void Intercept(Autofac.ContainerBuilder builder, ILifetimeScope scope, IInterceptor[] interceptors) + { + builder.RegisterInstance(new DbContextOptionsBuilder(scope.Resolve>()).AddInterceptors(interceptors).Options).As>().SingleInstance(); + + // The endpoint mints each connection's scope through IServiceScopeFactory, which the container answers from its + // root; nesting it here instead keeps the interceptors on the connection's own database commands. + builder.Register(c => new ScopedServiceScopeFactory(c.Resolve())).As(); + } + /// Drive a FRESH executor's ReattachAsync with the endpoint flag (+ proxy path) ON — the cross-worker-death re-open path. private async Task ReattachAsync(AgentRunReattachReservation reservation, IAgentHarness harness, bool proxyPresent = true, CancellationToken cancellationToken = default) { @@ -1446,9 +1609,9 @@ public async Task ExchangeAsync(int id, string method) return JsonDocument.Parse(line).RootElement.Clone(); } - public async Task CallToolAsync(int id, string name, object arguments) + public async Task CallToolAsync(int id, string name, object arguments, TimeSpan? readTimeout = null) { - var line = await SendThenReadAsync(JsonSerializer.Serialize(new { jsonrpc = "2.0", id, method = "tools/call", @params = new { name, arguments } })) + var line = await SendThenReadAsync(JsonSerializer.Serialize(new { jsonrpc = "2.0", id, method = "tools/call", @params = new { name, arguments } }), readTimeout) ?? throw new InvalidOperationException("endpoint closed before a response"); return JsonDocument.Parse(line).RootElement.Clone().GetProperty("result"); } @@ -1465,12 +1628,12 @@ public async Task CallToolAsync(int id, string name, object argumen catch (SocketException) { return null; } } - private async Task SendThenReadAsync(string request) + private async Task SendThenReadAsync(string request, TimeSpan? readTimeout = null) { await _writer.WriteLineAsync(request); await _writer.FlushAsync(); - return await _reader.ReadLineAsync().WaitAsync(TimeSpan.FromSeconds(10)); + return await _reader.ReadLineAsync().WaitAsync(readTimeout ?? TimeSpan.FromSeconds(10)); } public async ValueTask DisposeAsync() diff --git a/backend/tests/CodeSpace.IntegrationTests/Agents/McpDecisionFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Agents/McpDecisionFlowTests.cs index fc782813b..b1290c034 100644 --- a/backend/tests/CodeSpace.IntegrationTests/Agents/McpDecisionFlowTests.cs +++ b/backend/tests/CodeSpace.IntegrationTests/Agents/McpDecisionFlowTests.cs @@ -1,3 +1,4 @@ +using System.Data.Common; using System.Text.Json; using Autofac; using CodeSpace.Core.Persistence.Db; @@ -14,6 +15,7 @@ using CodeSpace.Messages.Agents; using CodeSpace.Messages.Enums; using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; using Shouldly; namespace CodeSpace.IntegrationTests.Agents; @@ -632,6 +634,111 @@ public async Task Decision_governance_disabled_refuses_to_raise() Text(result).ShouldContain("governance", Case.Insensitive); } + [Fact] + public async Task A_decision_whose_park_faulted_is_parked_by_the_retry_and_reaches_the_queue() + { + // The park faulted after the claim, so the call was told to retry — rightly, the run is live. But the retry found + // the claimed row Pending and only blocked on it: no queue shows a Pending row and nothing else parks it, so every + // retry came back with a ticket, and the run's end then held it for review over a question nobody was ever asked. + var (teamId, ownerId, channelId) = await SeedTeamChannelAsync(); + var runId = await SeedLiveRunAsync(teamId); + var parkFault = new FirstParkFaults(); + + using var scope = BeginScopeThrough(parkFault); + var handler = new McpRequestHandler(new SingleToolRegistry(new DecisionRequestTool()), AgentAutonomyLevel.Standard, teamId, null, runId, + scope.Resolve(), 1, governanceEnabled: true, approvalConversationId: channelId, + scope.Resolve(), scope.Resolve(), scope.Resolve()); + var arguments = new { question = "Which path?", decisionType = "choose_one", options = new[] { new { id = "a", label = "A" }, new { id = "b", label = "B" } } }; + + var faulted = await CallToolAsync(handler, arguments); + + parkFault.Fired.ShouldBeTrue("precondition: the first park faulted"); + Text(faulted).ShouldContain("retry shortly", customMessage: "a fault on a live run is the governed fault, and retrying it is right"); + (await ReadRunRowsAsync(teamId, runId)).ShouldHaveSingleItem().Status.ShouldBe(ToolCallLedgerStatus.Pending, "precondition: the claim landed and the park did not"); + + var retry = Task.Run(() => CallToolAsync(handler, arguments)); + var (ledgerId, _) = await WaitForPostedCardAsync(teamId, runId); + + (await ListPendingAsync(teamId)).ShouldContain(d => d.Id == ledgerId, "the retry parked the question, so the queue shows it"); + (await ReadRunCardCountAsync(teamId, channelId)).ShouldBe(1, "and posted its one card"); + + (await AnswerViaQueueAsync(ledgerId, new[] { "b" }, null, teamId, ownerId)).Outcome.ShouldBe(DecisionAnswerOutcome.Answered); + + var answered = await retry.WaitAsync(TimeSpan.FromSeconds(20)); + answered.GetProperty("isError").GetBoolean().ShouldBeFalse("the answer reaches the retried call"); + answered.GetProperty("structuredContent").GetProperty("selectedOptions")[0].GetString().ShouldBe("b"); + } + + [Fact] + public async Task A_decision_raised_on_a_run_that_was_stopped_is_told_the_run_ended_not_to_retry() + { + // A stop bumps the fence epoch too, so the handler's claim (at the epoch the agent held) is stale as well as the run + // ended; the agent must get the final answer, never "retry shortly". + var (teamId, _, channelId) = await SeedTeamChannelAsync(); + var runId = await SeedLiveRunAsync(teamId); + + using (var stop = _fixture.BeginScope()) + (await stop.Resolve().CancelRunningAsync(runId, "stopped by the operator", AgentRunAbandonCause.OperatorCancelled, CancellationToken.None)).ShouldBeTrue("precondition: the stop won the run"); // the production writer of a stop + + using var scope = _fixture.BeginScope(); + var handler = new McpRequestHandler(new SingleToolRegistry(new DecisionRequestTool()), AgentAutonomyLevel.Standard, teamId, null, runId, + scope.Resolve(), 1, governanceEnabled: true, approvalConversationId: channelId, + scope.Resolve(), scope.Resolve(), scope.Resolve()); + + var result = await CallToolAsync(handler, new { question = "Which path?", decisionType = "choose_one", options = new[] { new { id = "a", label = "A" }, new { id = "b", label = "B" } } }); + + result.GetProperty("isError").GetBoolean().ShouldBeTrue(); + Text(result).ShouldBe(McpRequestHandler.RunEndedToolCallError); + (await ReadRunRowsAsync(teamId, runId)).ShouldBeEmpty(); + } + + /// A live agent run its worker holds — what ClaimOwnershipAsync leaves: Running, owned, at fence epoch 1, its lease current. + private async Task SeedLiveRunAsync(Guid teamId) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + + var runId = Guid.NewGuid(); + var now = DateTimeOffset.UtcNow; + db.AgentRun.Add(new AgentRun { Id = runId, TeamId = teamId, Harness = "codex-cli", Status = AgentRunStatus.Running, StartedAt = now, HeartbeatAt = now, LeaseExpiresAt = now + AgentRunLiveness.Window, OwnerId = Guid.NewGuid(), FenceEpoch = 1 }); + + await db.SaveChangesAsync(); + return runId; + } + + private async Task> ListPendingAsync(Guid teamId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().ListPendingAsync(teamId, CancellationToken.None); + } + + /// A scope whose database commands pass . + private ILifetimeScope BeginScopeThrough(IInterceptor interceptor) + { + DbContextOptions production; + using (var probe = _fixture.BeginScope()) + production = probe.Resolve>(); + + var options = new DbContextOptionsBuilder(production).AddInterceptors(interceptor).Options; + + return _fixture.BeginScope(b => b.RegisterInstance(options).As>().SingleInstance()); + } + + /// Faults the scope's first park — the write that stamps a claimed row's approval token — as a lost connection would. + private sealed class FirstParkFaults : DbCommandInterceptor + { + private int _fired; + + public bool Fired => _fired == 1; + + public override ValueTask> NonQueryExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) + { + if (!command.CommandText.Contains("UPDATE tool_call_ledger", StringComparison.Ordinal) || !command.CommandText.Contains("approval_token", StringComparison.Ordinal) || Interlocked.Exchange(ref _fired, 1) == 1) return ValueTask.FromResult(result); + + throw new TimeoutException("the connection was lost as the claimed decision was parked"); + } + } + private McpRequestHandler DecisionHandlerCfg(ILifetimeScope scope, Guid teamId, Guid runId, Guid? conversationId, bool governanceEnabled = true) => new(new SingleToolRegistry(new DecisionRequestTool()), AgentAutonomyLevel.Standard, teamId, null, runId, scope.Resolve(), 0, governanceEnabled: governanceEnabled, approvalConversationId: conversationId, diff --git a/backend/tests/CodeSpace.IntegrationTests/Agents/ToolCallLedgerServiceTests.cs b/backend/tests/CodeSpace.IntegrationTests/Agents/ToolCallLedgerServiceTests.cs index 79139d24a..f7fcc3473 100644 --- a/backend/tests/CodeSpace.IntegrationTests/Agents/ToolCallLedgerServiceTests.cs +++ b/backend/tests/CodeSpace.IntegrationTests/Agents/ToolCallLedgerServiceTests.cs @@ -447,6 +447,26 @@ public async Task A_stale_epoch_claim_after_a_reclaim_is_refused_and_the_current .ShouldBe(ToolCallClaimOutcome.Proceed, "the CURRENT owner's claim, at its own live fence, must still proceed"); } + // The claim reads the run's status beside its fence. A run with no row has neither: it reads epoch 0, so only a caller + // at epoch 0 owns it, and it has no status, so it is never refused as one that has ended. + [Theory] + [InlineData(0L, true)] // the epoch an unclaimed run carries: the claim proceeds + [InlineData(1L, false)] // any later epoch: no live owner of a run with no row + public async Task A_claim_on_a_run_with_no_row_is_fenced_at_epoch_zero_and_never_refused_as_ended(long fenceEpoch, bool proceeds) + { + var teamId = await SeedTeamAsync(); + var runId = Guid.NewGuid(); + + using (var scope = _fixture.BeginScope()) + { + if (proceeds) (await Svc(scope).TryClaimAsync(runId, teamId, "git.open_pr", Key, InputHash, fenceEpoch, CancellationToken.None)).Outcome.ShouldBe(ToolCallClaimOutcome.Proceed); + else await Should.ThrowAsync(() => Svc(scope).TryClaimAsync(runId, teamId, "git.open_pr", Key, InputHash, fenceEpoch, CancellationToken.None)); + } + + using var verify = _fixture.BeginScope(); + (await verify.Resolve().ToolCallLedger.AsNoTracking().CountAsync(l => l.AgentRunId == runId)).ShouldBe(proceeds ? 1 : 0); + } + // A worker hard-killed mid-call leaves its row non-terminal (its recovery write never ran) while the detached agent // survives and the run is re-attached at a new epoch. The run is still live, so ExpireStaleToolCallsAsync (terminal runs // only) never settles the row, and every identical re-call used to read InFlight for the rest of the run. diff --git a/backend/tests/CodeSpace.IntegrationTests/Decisions/StoppedRunDecisionFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Decisions/StoppedRunDecisionFlowTests.cs index bb1a547d2..40bc7f2cb 100644 --- a/backend/tests/CodeSpace.IntegrationTests/Decisions/StoppedRunDecisionFlowTests.cs +++ b/backend/tests/CodeSpace.IntegrationTests/Decisions/StoppedRunDecisionFlowTests.cs @@ -6,6 +6,7 @@ using CodeSpace.Core.Persistence.Entities; using CodeSpace.Core.Services.Agents; using CodeSpace.Core.Services.Agents.Exceptions; +using CodeSpace.Core.Services.Agents.Mcp; using CodeSpace.Core.Services.Decisions; using CodeSpace.Core.Services.Workflows; using CodeSpace.IntegrationTests.Infrastructure; @@ -502,6 +503,217 @@ public async Task An_answer_issued_while_a_recovered_run_is_landing_waits_for_it (await LandedAsync(agentId)).ShouldBe(Landing(AgentRunStatus.Succeeded, decisionId), "recovered from a clean exit with its question open: held for review naming it"); } + // ─── A run that has ended takes no new question, and no question parks on it ─── + + [Theory] + [InlineData(AgentRunStatus.Succeeded, true)] + [InlineData(AgentRunStatus.Succeeded, false)] + [InlineData(AgentRunStatus.Failed, true)] + [InlineData(AgentRunStatus.Failed, false)] + [InlineData(AgentRunStatus.NeedsReview, true)] + [InlineData(AgentRunStatus.NeedsReview, false)] + public async Task A_new_claim_on_a_run_that_has_ended_is_refused_and_records_nothing(AgentRunStatus ending, bool owned) + { + // A run's own end keeps its fence epoch, so a decision.request already past the authority guard still owned the run + // by epoch: it claimed a row, parked it, and left a finished run with an open question in the queue. + var (teamId, _) = await SeedTeamAsync(); + var runId = await SeedWorkflowRunAsync(teamId); + var agent = await SeedLiveAgentAsync(teamId, runId, owned); + + await EndAgentAsync(agent, ending); + (await AgentStatusAsync(agent.RunId)).ShouldBe(ending, "precondition: the run landed where its writer put it, at the epoch the claim below holds"); + + await Should.ThrowAsync(() => ClaimDecisionAsync(teamId, agent.RunId)); + + (await DecisionRowsAsync(agent.RunId)).ShouldBeEmpty("a run that has ended records no new question — so none reaches the queue"); + } + + [Theory] + [InlineData(TerminalPath.Cancel)] + [InlineData(TerminalPath.Abandon)] + [InlineData(TerminalPath.Recovery)] + public async Task A_new_claim_on_a_run_a_stop_ended_is_refused_as_ended_not_as_lost_ownership(TerminalPath path) + { + // A stop, an abandon and a recovery bump the fence epoch as they end the run, so the caller's epoch is stale as well + // as the run ended — and the run's end is the answer: the fence checked first reported it as lost ownership, which + // the handler tells the agent to retry. + if (path == TerminalPath.Recovery && OperatingSystem.IsWindows()) return; + + var (teamId, _) = await SeedTeamAsync(); + var runId = await SeedWorkflowRunAsync(teamId); + var agent = await SeedAgentToEndAsync(path, teamId, runId); + + await EndThroughAsync(path, agent); // CancelRunningAsync / the reconciler's abandon and recovery: the production writers of each ending + (await AgentStatusAsync(agent.RunId)).ShouldNotBe(AgentRunStatus.Running, "precondition: the run ended"); + + await Should.ThrowAsync(() => ClaimDecisionAsync(teamId, agent.RunId)); + + (await DecisionRowsAsync(agent.RunId)).ShouldBeEmpty("and nothing was recorded"); + } + + [Fact] + public async Task A_claim_waiting_behind_the_runs_end_is_refused_once_the_end_commits() + { + // The claim reads the run under a share lock, so one issued while the end holds the run waits for the end — and + // must then read the run the end left, not the live one it waited behind. + var (teamId, _) = await SeedTeamAsync(); + var runId = await SeedWorkflowRunAsync(teamId); + var agent = await SeedLiveAgentAsync(teamId, runId, owned: true); + + var claimDuringEnd = new IssueWhileHeld(_fixture, command => IsLedgerWriteOf(command, agent.RunId), afterExecution: false, () => Record.ExceptionAsync(() => ClaimDecisionAsync(teamId, agent.RunId))); + await EndAgentThroughAsync(agent, AgentRunStatus.Succeeded, claimDuringEnd); + + claimDuringEnd.Issued.ShouldBeTrue("precondition: the end had flipped the run and was closing its decisions when the claim was issued"); + claimDuringEnd.WaitingBeforeTheStep.ShouldBeFalse("precondition: nothing was waiting on the end before the claim was issued"); + claimDuringEnd.WaitedBehindTheHeld.ShouldBeTrue("the claim's read of the run waits on the end's hold on it"); + + (await claimDuringEnd.Step!.WaitAsync(Bound)).ShouldBeOfType("once the end committed, the claim read the run as ended"); + (await DecisionRowsAsync(agent.RunId)).ShouldBeEmpty(); + (await LandedAsync(agent.RunId)).ShouldBe(Landing(AgentRunStatus.Succeeded, openDecisionId: null), "the end landed its own result, untouched by the refused claim"); + } + + /// Each way an agent run ends: its worker's completion, an operator's stop, the reconciler's abandon of a run whose worker is gone, and its recovery of one that exited while nobody watched. + public enum TerminalPath + { + Completion, + Cancel, + Abandon, + Recovery, + } + + [Theory] + [InlineData(TerminalPath.Completion)] + [InlineData(TerminalPath.Cancel)] + [InlineData(TerminalPath.Abandon)] + [InlineData(TerminalPath.Recovery)] + public async Task A_decision_claimed_before_its_run_ends_closes_with_the_run_and_never_parks_after_it(TerminalPath path) + { + // Claimed before the end, parked after it: every end closed only the parked questions, so the claimed one survived + // Pending — and its park then landed on the finished run. + if (path == TerminalPath.Recovery && OperatingSystem.IsWindows()) return; + + var (teamId, _) = await SeedTeamAsync(); + var runId = await SeedWorkflowRunAsync(teamId); + var agent = await SeedAgentToEndAsync(path, teamId, runId); + + var claim = await ClaimDecisionAsync(teamId, agent.RunId); + claim.Outcome.ShouldBe(ToolCallClaimOutcome.Proceed, "precondition: the question was claimed while the run lived"); + + await EndThroughAsync(path, agent); + + (await ParkAsync(claim.LedgerId, teamId)).ShouldBeFalse("a park after the end finds nothing left to park"); + + var row = await LedgerRowAsync(claim.LedgerId); + row.Status.ShouldBe(ToolCallLedgerStatus.Expired, "the run's end closed the claimed question along with the parked ones"); + row.Error.ShouldBe(path is TerminalPath.Cancel or TerminalPath.Abandon ? StoppedRunDecisions.ExpiredError : StoppedRunDecisions.EndedError); + row.ApprovalToken.ShouldBeNull("and the park never stamped it"); + (await ListPendingAsync(teamId)).ShouldNotContain(d => d.Id == claim.LedgerId); + } + + [Fact] + public async Task A_decision_whose_close_the_stop_lost_is_still_never_parked_on_the_stopped_run() + { + // A stop closes its run's questions in a statement of its own after the run is stopped, and a failure there is + // swallowed — the run is stopped either way. A question claimed before the stop must not then park on it. + var (teamId, _) = await SeedTeamAsync(); + var runId = await SeedWorkflowRunAsync(teamId); + var agent = await SeedLiveAgentAsync(teamId, runId, owned: true); + var claim = await ClaimDecisionAsync(teamId, agent.RunId); + + var lostClose = new LostDecisionClose(agent.RunId); + (await CancelAgentThroughAsync(agent.RunId, lostClose)).ShouldBeTrue("precondition: the stop won the run"); + lostClose.Fired.ShouldBeTrue("precondition: the stop's close of the run's questions was lost"); + (await LedgerRowAsync(claim.LedgerId)).Status.ShouldBe(ToolCallLedgerStatus.Pending, "precondition: so the claimed question is still Pending"); + + await Should.ThrowAsync(() => ParkAsync(claim.LedgerId, teamId)); + + var row = await LedgerRowAsync(claim.LedgerId); + row.Status.ShouldBe(ToolCallLedgerStatus.Pending, "the park reads the run itself, so it refuses a stopped run even when the close never ran"); + row.ApprovalToken.ShouldBeNull(); + (await ListPendingAsync(teamId)).ShouldNotContain(d => d.Id == claim.LedgerId); + } + + [Theory] + [InlineData(TerminalPath.Completion)] + [InlineData(TerminalPath.Cancel)] + public async Task A_decision_asked_as_its_run_ends_never_stays_open_on_the_ended_run(TerminalPath path) + { + var (teamId, _) = await SeedTeamAsync(); + var runId = await SeedWorkflowRunAsync(teamId); + + for (var i = 0; i < 10; i++) + { + var agent = await SeedLiveAgentAsync(teamId, runId, owned: true); + + var ask = AskAsync(teamId, agent.RunId); + var end = EndThroughAsync(path, agent); + var askFault = await Record.ExceptionAsync(() => ask); + var endFault = await Record.ExceptionAsync(() => end); + + endFault.ShouldBeNull($"round {i}: the end lands — a deadlock (40P01) here means the question and the end took their locks in different orders"); + if (askFault is not null) askFault.ShouldBeOfType($"round {i}: the question was either asked before the end or refused as ended — never a deadlock"); + (await DecisionRowsAsync(agent.RunId)).ShouldNotContain(d => d.Status == ToolCallLedgerStatus.Pending || d.Status == ToolCallLedgerStatus.AwaitingApproval, $"round {i}: no question stays open on a run that has ended"); + } + } + + [Fact] + public async Task A_park_racing_the_runs_end_waits_for_the_end_and_never_deadlocks() + { + // The end locks the run's questions, then the run. A park that took the run first and then its question would hold + // what the end waits for while waiting on what the end holds: 40P01. The park reads the run without locking it, so it + // only ever waits on its own question — behind the end — and finds it closed. + var (teamId, _) = await SeedTeamAsync(); + var runId = await SeedWorkflowRunAsync(teamId); + var agent = await SeedLiveAgentAsync(teamId, runId, owned: true); + var claim = await ClaimDecisionAsync(teamId, agent.RunId); + + var parkDuringEnd = new IssueWhileHeld(_fixture, command => IsRunFlipOf(command, agent.RunId), afterExecution: false, () => ParkAsync(claim.LedgerId, teamId)); + var transactions = new TransactionCounter(); + await EndAgentThroughAsync(agent, AgentRunStatus.Succeeded, parkDuringEnd, transactions); + + parkDuringEnd.Issued.ShouldBeTrue("precondition: the end held the run's questions and was about to write the run itself when the park was issued"); + parkDuringEnd.WaitingBeforeTheStep.ShouldBeFalse("precondition: nothing was waiting on the end before the park was issued"); + parkDuringEnd.WaitedBehindTheHeld.ShouldBeTrue("the park waits on the end's hold on its question"); + + (await parkDuringEnd.Step!.WaitAsync(Bound)).ShouldBeFalse("once the end committed, the park found its question closed — no deadlock surfaced on its side"); + transactions.Started.ShouldBe(1, "the end landed on its first attempt: a second one means it lost a deadlock to the park and was written again"); + (await LandedAsync(agent.RunId)).ShouldBe(Landing(AgentRunStatus.Succeeded, claim.LedgerId), "the end saw the claimed question open and held the run for it"); + (await LedgerRowAsync(claim.LedgerId)).Status.ShouldBe(ToolCallLedgerStatus.Expired); + } + + /// A call the run made before it ended: a write whose result was recorded, or a question it claimed and never parked. + public enum MadeBeforeTheEnd + { + RecordedWrite, + ClaimedQuestion, + } + + [Theory] + [InlineData(MadeBeforeTheEnd.RecordedWrite)] + [InlineData(MadeBeforeTheEnd.ClaimedQuestion)] + public async Task A_call_made_before_the_run_ended_still_replays_after_it(MadeBeforeTheEnd made) + { + // A run that has ended takes no NEW call — but an identical re-call of one it already made replays that call's + // terminal instead of being refused, so the answer a call got never changes under it. + var (teamId, _) = await SeedTeamAsync(); + var runId = await SeedWorkflowRunAsync(teamId); + var agent = await SeedLiveAgentAsync(teamId, runId, owned: true); + var toolKind = made == MadeBeforeTheEnd.RecordedWrite ? "git.open_pr" : DecisionToolKinds.DecisionRequest; + var key = $"{toolKind}:{Guid.NewGuid():N}"; + + var first = await ClaimAsync(teamId, agent.RunId, toolKind, key); + if (made == MadeBeforeTheEnd.RecordedWrite) await RecordSucceededAsync(first.LedgerId, teamId); + + await EndAgentAsync(agent, AgentRunStatus.Succeeded); + + var replay = await ClaimAsync(teamId, agent.RunId, toolKind, key); + + replay.Outcome.ShouldBe(ToolCallClaimOutcome.Duplicate, "the existing call replays — the run's end refuses only a new one"); + replay.LedgerId.ShouldBe(first.LedgerId); + replay.PriorStatus.ShouldBe(made == MadeBeforeTheEnd.RecordedWrite ? ToolCallLedgerStatus.Succeeded : ToolCallLedgerStatus.Expired); + replay.PriorError.ShouldBe(made == MadeBeforeTheEnd.RecordedWrite ? null : StoppedRunDecisions.EndedError, "a question the end closed replays that the run ended"); + } + // ─── Drive the real services ────────────────────────────────────────────────── private async Task CancelAgentAsync(Guid agentId) @@ -551,6 +763,106 @@ private async Task ReconcileThroughAsync(IInterceptor interceptor) await scope.Resolve().ReconcileAsync(CancellationToken.None); } + /// Stop the run through a scope whose database commands pass . + private async Task CancelAgentThroughAsync(Guid agentId, IInterceptor interceptor) + { + using var scope = BeginScopeThrough(interceptor); + return await scope.Resolve().CancelRunningAsync(agentId, "stopped by the operator", AgentRunAbandonCause.OperatorCancelled, CancellationToken.None); + } + + /// The ledger claim the MCP handler makes for a call, at the attempt's own fence epoch (1, its claim's). + private async Task ClaimAsync(Guid teamId, Guid agentId, string toolKind, string key) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().TryClaimAsync(agentId, teamId, toolKind, key, new string('0', 64), 1, CancellationToken.None); + } + + /// A fresh decision.request's claim: its row stays Pending until the handler parks it. + private Task ClaimDecisionAsync(Guid teamId, Guid agentId) => ClaimAsync(teamId, agentId, DecisionToolKinds.DecisionRequest, $"{DecisionToolKinds.DecisionRequest}:{Guid.NewGuid():N}"); + + /// The handler's park of a claimed question: the real CAS Pending → AwaitingApproval. + private async Task ParkAsync(Guid ledgerId, Guid teamId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().TryBeginApprovalAsync(ledgerId, teamId, new ToolCallApprovalPark { Token = Guid.NewGuid().ToString("N"), DeadlineAt = DateTimeOffset.UtcNow.AddHours(1) }, CancellationToken.None); + } + + /// Ask a question as the handler does: claim it, then park what was claimed. + private async Task AskAsync(Guid teamId, Guid agentId) + { + var claim = await ClaimDecisionAsync(teamId, agentId); + + if (claim.Outcome == ToolCallClaimOutcome.Proceed) await ParkAsync(claim.LedgerId, teamId); + } + + /// A side-effecting call's own recorded result. + private async Task RecordSucceededAsync(Guid ledgerId, Guid teamId) + { + using var scope = _fixture.BeginScope(); + await scope.Resolve().RecordTerminalAsync(ledgerId, teamId, ToolCallLedgerStatus.Succeeded, """{"content":[{"type":"text","text":"opened"}],"isError":false}""", null, CancellationToken.None); + } + + /// The card a parked question's handler posted, recorded on its row. + private async Task RecordCardAsync(Guid ledgerId, Guid teamId, Guid messageId) + { + using var scope = _fixture.BeginScope(); + await scope.Resolve().SetApprovalMessageAsync(ledgerId, teamId, messageId, CancellationToken.None); + } + + private async Task> DecisionRowsAsync(Guid agentId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().ToolCallLedger.AsNoTracking().Where(l => l.AgentRunId == agentId && l.ToolKind == DecisionToolKinds.DecisionRequest).ToListAsync(); + } + + /// A live agent run for to end, at fence epoch 1: one its worker holds, for a completion or a stop; one whose worker is gone, for an abandon; one that exited cleanly while nobody watched, for a recovery. + private async Task SeedAgentToEndAsync(TerminalPath path, Guid teamId, Guid workflowRunId) => path switch + { + TerminalPath.Abandon => new LiveAgent(await SeedRunningAgentAsync(teamId, livenessAgo: TimeSpan.FromDays(3650), fenceEpoch: 1), null), // legacy claim, then its worker died: the sweep reaches its lapsed lease first + TerminalPath.Recovery => new LiveAgent(await SeedRecoverableAgentAsync(teamId, workflowRunId, fenceEpoch: 1), null), // legacy claim, then its agent exited unwatched + _ => await SeedLiveAgentAsync(teamId, workflowRunId, owned: true), // ClaimOwnershipAsync's live run + }; + + /// End the run along , through the production writer of that ending. + private async Task EndThroughAsync(TerminalPath path, LiveAgent agent) + { + if (path == TerminalPath.Completion) await EndAgentAsync(agent, AgentRunStatus.Succeeded); + else if (path == TerminalPath.Cancel) (await CancelAgentAsync(agent.RunId)).ShouldBeTrue("precondition: the stop won the run"); + else await ReconcileUntilEndedAsync(agent.RunId); + } + + /// Reconcile until the run this test owns has ended — a pass is bounded and deployment-wide, so it may reach other tests' stale runs first. + private async Task ReconcileUntilEndedAsync(Guid agentId) + { + for (var pass = 0; pass < 10 && await AgentStatusAsync(agentId) == AgentRunStatus.Running; pass++) + await ReconcileThroughAsync(new TransactionCounter()); + + (await AgentStatusAsync(agentId)).ShouldNotBe(AgentRunStatus.Running, $"the reconciler did not end run {agentId} within 10 passes — look in agent_run for its lease and runner handle"); + } + + /// Fails the scope's close of agentRunId's questions, the step a stop swallows once its run is stopped. + private sealed class LostDecisionClose : DbCommandInterceptor + { + private readonly Guid _agentRunId; + + public LostDecisionClose(Guid agentRunId) { _agentRunId = agentRunId; } + + public bool Fired { get; private set; } + + public override ValueTask> ReaderExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) => Lose(command, result); + + public override ValueTask> NonQueryExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) => Lose(command, result); + + private ValueTask Lose(DbCommand command, T result) + { + if (!IsLedgerWriteOf(command, _agentRunId)) return ValueTask.FromResult(result); + + Fired = true; + + throw new InvalidOperationException("the connection was lost as the stop closed the run's questions"); + } + } + /// The end queues on the held decision first: its lock waits on the hold; then the stop wins the run and its close queues too. private async Task<(Task End, Task Stop)> EndThenStopAsync(LiveAgent agent) { @@ -645,15 +957,21 @@ private async Task WaitForBlockedSessionsAsync(int count, string signal) /// The end's lock on the run's decisions, taken before it reads which of them are open. private static bool IsDecisionLockOf(DbCommand command, Guid agentRunId) => command.CommandText.Contains("FROM tool_call_ledger", StringComparison.Ordinal) && command.CommandText.Contains("FOR UPDATE", StringComparison.Ordinal) && CarriesParameter(command, agentRunId); + /// The end's terminal write of the run's own row — taken after it has locked the run's decisions. + private static bool IsRunFlipOf(DbCommand command, Guid agentRunId) => command.CommandText.Contains("UPDATE agent_run", StringComparison.Ordinal) && CarriesParameter(command, agentRunId); + private static bool CarriesParameter(DbCommand command, Guid value) => command.Parameters.Cast().Any(parameter => parameter.Value is Guid id && id == value); private static Task EndAsync(IAgentRunService runs, LiveAgent agent, AgentRunStatus ending) => agent.Owner is { } owner ? runs.CompleteAsync(owner, Ending(ending), CancellationToken.None) : runs.CompleteAsync(agent.RunId, Ending(ending), CancellationToken.None); - /// The terminal result a worker hands in: the harness's own success, or the executor's generic failure. - private static AgentRunResult Ending(AgentRunStatus ending) => ending == AgentRunStatus.Succeeded - ? new AgentRunResult { Status = AgentRunStatus.Succeeded, ExitReason = "completed", Summary = "did the work" } - : new AgentRunResult { Status = AgentRunStatus.Failed, ExitReason = AgentRunExecutor.GenericExecutorExitReason, Error = "the executor faulted" }; + /// The terminal result a worker hands in: the harness's own success, the executor's stalled-agent hold for review, or its generic failure. + private static AgentRunResult Ending(AgentRunStatus ending) => ending switch + { + AgentRunStatus.Succeeded => new AgentRunResult { Status = AgentRunStatus.Succeeded, ExitReason = "completed", Summary = "did the work" }, + AgentRunStatus.NeedsReview => new AgentRunResult { Status = AgentRunStatus.NeedsReview, CompletionDisposition = CompletionDisposition.Blocked, ExitReason = AgentAcceptanceContract.StalledExitReason, Error = "the agent produced no output for the idle window" }, + _ => new AgentRunResult { Status = AgentRunStatus.Failed, ExitReason = AgentRunExecutor.GenericExecutorExitReason, Error = "the executor faulted" }, + }; /// A scope whose database commands and transactions pass . private ILifetimeScope BeginScopeThrough(params IInterceptor[] interceptors) @@ -690,7 +1008,8 @@ public override async ValueTask> NonQueryExecutingAsync( /// Issues a racing step once, at the command of the intercepted scope that at picks — just before it runs, or just /// after — and holds that scope there until the step is either seen WAITING behind the scope's own database session or /// has finished without waiting. Which of the two happened is the verdict: a step that finished first got past whatever - /// the held session had locked, or found nothing locked at all. + /// the held session had locked, or found nothing locked at all. Both kinds of command are watched: the close of a run's + /// decisions returns the rows it closed, so it runs as a reader. /// private sealed class IssueWhileHeld : DbCommandInterceptor { @@ -730,6 +1049,20 @@ public override async ValueTask NonQueryExecutedAsync(DbCommand command, Co return result; } + public override async ValueTask> ReaderExecutingAsync(DbCommand command, CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default) + { + if (!_afterExecution) await HoldAsync(command).ConfigureAwait(false); + + return result; + } + + public override async ValueTask ReaderExecutedAsync(DbCommand command, CommandExecutedEventData eventData, DbDataReader result, CancellationToken cancellationToken = default) + { + if (_afterExecution) await HoldAsync(command).ConfigureAwait(false); + + return result; + } + private async Task HoldAsync(DbCommand command) { if (!_at(command) || Interlocked.Exchange(ref _fired, 1) == 1) return; @@ -885,8 +1218,8 @@ private async Task LedgerRowAsync(Guid ledgerId) // ─── Seeding ────────────────────────────────────────────────────────────────── - /// A Running agent run; past the window makes it one the reconciler abandons (the heartbeat and lease lapse together). - private async Task SeedRunningAgentAsync(Guid teamId, TimeSpan livenessAgo) + /// A Running agent run; past the window makes it one the reconciler abandons (the heartbeat and lease lapse together). 1 is what the legacy claim (MarkRunningAsync) stamps on a Queued run. + private async Task SeedRunningAgentAsync(Guid teamId, TimeSpan livenessAgo, long fenceEpoch = 0) { using var scope = _fixture.BeginScope(); var db = scope.Resolve(); @@ -894,7 +1227,7 @@ private async Task SeedRunningAgentAsync(Guid teamId, TimeSpan livenessAgo var agentId = Guid.NewGuid(); var stamp = DateTimeOffset.UtcNow - livenessAgo; - db.AgentRun.Add(new AgentRun { Id = agentId, TeamId = teamId, Harness = "codex-cli", Status = AgentRunStatus.Running, StartedAt = stamp, HeartbeatAt = stamp, LeaseExpiresAt = stamp + AgentRunLiveness.Window }); + db.AgentRun.Add(new AgentRun { Id = agentId, TeamId = teamId, Harness = "codex-cli", Status = AgentRunStatus.Running, StartedAt = stamp, HeartbeatAt = stamp, LeaseExpiresAt = stamp + AgentRunLiveness.Window, FenceEpoch = fenceEpoch }); await db.SaveChangesAsync(); return agentId; @@ -981,7 +1314,7 @@ private static (Guid Lower, Guid Higher) OrderedIds() } /// A Running agent run of whose worker is gone and whose process exited cleanly while nobody watched — a lapsed lease, a dead pid, a clean exit marker in its spool: the run the reconciler recovers rather than abandons. - private async Task SeedRecoverableAgentAsync(Guid teamId, Guid workflowRunId) + private async Task SeedRecoverableAgentAsync(Guid teamId, Guid workflowRunId, long fenceEpoch = 0) { var spoolDir = Path.Combine(Path.GetTempPath(), "cs-decision-recover-" + Guid.NewGuid().ToString("N")); Directory.CreateDirectory(spoolDir); @@ -995,7 +1328,7 @@ private async Task SeedRecoverableAgentAsync(Guid teamId, Guid workflowRun using var scope = _fixture.BeginScope(); var db = scope.Resolve(); - db.AgentRun.Add(new AgentRun { Id = agentId, TeamId = teamId, WorkflowRunId = workflowRunId, Harness = "codex-cli", Status = AgentRunStatus.Running, StartedAt = stamp, HeartbeatAt = stamp, LeaseExpiresAt = stamp + AgentRunLiveness.Window, RunnerHandleJson = JsonSerializer.Serialize(handle, AgentJson.Options) }); + db.AgentRun.Add(new AgentRun { Id = agentId, TeamId = teamId, WorkflowRunId = workflowRunId, Harness = "codex-cli", Status = AgentRunStatus.Running, StartedAt = stamp, HeartbeatAt = stamp, LeaseExpiresAt = stamp + AgentRunLiveness.Window, RunnerHandleJson = JsonSerializer.Serialize(handle, AgentJson.Options), FenceEpoch = fenceEpoch }); await db.SaveChangesAsync(); return agentId; diff --git a/backend/tests/CodeSpace.UnitTests/Agents/McpRequestHandlerTests.cs b/backend/tests/CodeSpace.UnitTests/Agents/McpRequestHandlerTests.cs index a8ec8c05c..013d03930 100644 --- a/backend/tests/CodeSpace.UnitTests/Agents/McpRequestHandlerTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Agents/McpRequestHandlerTests.cs @@ -1,5 +1,6 @@ using System.Text.Json; using CodeSpace.Core.Services.Agents; +using CodeSpace.Core.Services.Agents.Exceptions; using CodeSpace.Core.Services.Agents.Mcp; using CodeSpace.Core.Services.Agents.Tools; using CodeSpace.Core.Services.Providers.Resilience; @@ -2049,4 +2050,109 @@ public async Task A_re_posted_card_shows_the_preview_its_row_was_parked_with() card.ShouldContain("- commitTitle: as parked", customMessage: "the card a reviewer approves is the one the row's pins were stamped from"); card.ShouldNotContain("as read now"); } + + // ── a call that reaches a run which has already ended ── + + private const string DecisionQuestion = """{"question":"ok?","decisionType":"choose_one","options":[{"id":"a","label":"A"},{"id":"b","label":"B"}]}"""; + + /// Where a call meets its run's end: the claim of a decision, of a write the tier allows, or of a write parked for approval — or the park of a decision claimed before the end. + public enum EndedAt + { + DecisionClaim, + AllowedWriteClaim, + ApprovalClaim, + DecisionPark, + } + + [Theory] + [InlineData(EndedAt.DecisionClaim)] + [InlineData(EndedAt.AllowedWriteClaim)] + [InlineData(EndedAt.ApprovalClaim)] + [InlineData(EndedAt.DecisionPark)] + public async Task A_call_refused_because_its_run_ended_is_answered_as_final_never_as_a_retry(EndedAt at) + { + // The run is over, so nothing it asks can take effect. "Retry shortly" sent the model round again and again; the + // answer must say the run ended, and that calling again cannot help. + var runId = Guid.NewGuid(); + var ledger = new SpyLedger + { + PendingDecisionCount = () => 0, + ClaimResult = () => ToolCallClaim.Proceed(Guid.NewGuid()), + OnClaimThrow = at == EndedAt.DecisionPark ? null : () => new AgentRunEndedException(runId), + OnBeginApprovalThrow = at == EndedAt.DecisionPark ? () => new AgentRunEndedException(runId) : null, + }; + var bot = new StubBot { ConversationInTeam = true }; + var tool = new FakeTool { Kind = "git.merge_pr", IsDestructiveOverride = true }; + + var result = (await Respond(EndedRunHandler(at, runId, ledger, bot, tool), at is EndedAt.DecisionClaim or EndedAt.DecisionPark ? Call(DecisionRequestTool.ToolKind, DecisionQuestion) : Call("git.merge_pr", "{}"))).GetProperty("result"); + + result.GetProperty("isError").GetBoolean().ShouldBeTrue(); + var text = result.GetProperty("content")[0].GetProperty("text").GetString(); + text.ShouldBe(McpRequestHandler.RunEndedToolCallError, "the model is told its run ended — never the governed-fault answer that invites a retry"); + text.ShouldNotContain("retry", Case.Insensitive); + tool.CallCount.ShouldBe(0, "nothing ran"); + bot.PostCount.ShouldBe(0, "and nobody was asked"); + } + + /// The handler a call meets its run's end through: a decision's, a write the Unleashed tier allows, or a write the Standard tier parks for approval. + private static McpRequestHandler EndedRunHandler(EndedAt at, Guid runId, SpyLedger ledger, StubBot bot, FakeTool tool) => at switch + { + EndedAt.AllowedWriteClaim => new McpRequestHandler(new FakeRegistry(tool), AgentAutonomyLevel.Unleashed, Guid.NewGuid(), null, runId, ledger, fenceEpoch: 1, governanceEnabled: true), + EndedAt.ApprovalClaim => new McpRequestHandler(new FakeRegistry(tool), AgentAutonomyLevel.Standard, Guid.NewGuid(), null, runId, ledger, fenceEpoch: 1, governanceEnabled: true, approvalConversationId: Guid.NewGuid(), bot, new StubWaiters(), new StubComponents()), + _ => DecisionHandlerOver(runId, ledger, bot), + }; + + /// A decision.request handler with a card surface, over a waiter that arms but is never signalled — the durable row is its only way out of a block. + private static McpRequestHandler DecisionHandlerOver(Guid runId, SpyLedger ledger, StubBot bot) => + new(new FakeRegistry(new DecisionRequestTool()), AgentAutonomyLevel.Standard, Guid.NewGuid(), null, runId, ledger, fenceEpoch: 1, governanceEnabled: true, approvalConversationId: Guid.NewGuid(), bot, new ArmedButNeverSignalledWaiters(), new StubComponents()); + + /// How a re-call learns its decision was closed by its run's end: the claim replays the closed row, or the park finds the row closed under it. + public enum ClosedBy + { + ClaimReplay, + LostPark, + } + + [Theory] + [InlineData(ClosedBy.ClaimReplay)] + [InlineData(ClosedBy.LostPark)] + public async Task A_decision_its_runs_end_closed_is_answered_that_the_run_ended(ClosedBy closedBy) + { + var ledgerId = Guid.NewGuid(); + var ledger = new SpyLedger + { + PendingDecisionCount = () => 0, + ClaimResult = () => closedBy == ClosedBy.ClaimReplay ? ToolCallClaim.Duplicate(ledgerId, ToolCallLedgerStatus.Expired, null, StoppedRunDecisions.EndedError) : ToolCallClaim.Proceed(ledgerId), + BeginApprovalResult = () => false, // the end closed the row before the park's CAS reached it + ApprovalState = () => new ToolCallApprovalState { Status = ToolCallLedgerStatus.Expired, Error = StoppedRunDecisions.EndedError }, + }; + var bot = new StubBot { ConversationInTeam = true }; + + var result = (await WithinArmRaceBudgetAsync(DecisionHandlerOver(Guid.NewGuid(), ledger, bot).HandleAsync(Parse(Call(DecisionRequestTool.ToolKind, DecisionQuestion)), CancellationToken.None), "a decision its run's end closed")).GetProperty("result"); + + result.GetProperty("isError").GetBoolean().ShouldBeTrue(); + result.GetProperty("content")[0].GetProperty("text").GetString().ShouldBe(StoppedRunDecisions.EndedError, "the closed row's own reason: the run that asked ended before it was answered"); + bot.PostCount.ShouldBe(0, "a closed question is never put to anyone"); + } + + [Fact] + public async Task A_re_call_of_a_decision_whose_park_never_landed_parks_it_and_posts_its_one_card() + { + // The first call claimed the row and its park faulted, so it was told to retry. The row is Pending: no queue shows + // it and nothing else parks it, so a re-call that only blocked on it waited for an answer nobody could give, and + // came back with a ticket every time. The re-call parks it instead, posting the card the first call never did. + var ledgerId = Guid.NewGuid(); + var ledger = new SpyLedger { PendingDecisionCount = () => 0, ClaimResult = () => ToolCallClaim.InFlight(ledgerId), BeginApprovalResult = () => true }; + ledger.ApprovalState = () => ledger.BegunPark is null + ? new ToolCallApprovalState { Status = ToolCallLedgerStatus.Pending } // claimed by the first call, never parked + : new ToolCallApprovalState { Status = ToolCallLedgerStatus.Succeeded, ResultJson = """{"decisionId":"d","selectedOptions":["a"]}""" }; // parked by this re-call, then answered + var bot = new StubBot { ConversationInTeam = true }; + + var result = (await WithinArmRaceBudgetAsync(DecisionHandlerOver(Guid.NewGuid(), ledger, bot).HandleAsync(Parse(Call(DecisionRequestTool.ToolKind, DecisionQuestion)), CancellationToken.None), "a re-call of a decision whose park never landed")).GetProperty("result"); + + ledger.BegunPark.ShouldNotBeNull("the re-call parks the row the first call claimed"); + bot.PostCount.ShouldBe(1, "and posts its one card"); + ledger.ApprovalMessages.ShouldHaveSingleItem().LedgerId.ShouldBe(ledgerId); + result.GetProperty("isError").GetBoolean().ShouldBeFalse("the answer to the parked question comes back"); + } } diff --git a/backend/tests/CodeSpace.UnitTests/Agents/ToolCallLedgerStateMachineTests.cs b/backend/tests/CodeSpace.UnitTests/Agents/ToolCallLedgerStateMachineTests.cs index 74d6ebb11..c5ac19d28 100644 --- a/backend/tests/CodeSpace.UnitTests/Agents/ToolCallLedgerStateMachineTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Agents/ToolCallLedgerStateMachineTests.cs @@ -22,6 +22,8 @@ public class ToolCallLedgerStateMachineTests [InlineData(ToolCallLedgerStatus.AwaitingApproval, ToolCallLedgerStatus.Running, true)] [InlineData(ToolCallLedgerStatus.AwaitingApproval, ToolCallLedgerStatus.Failed, true)] [InlineData(ToolCallLedgerStatus.AwaitingApproval, ToolCallLedgerStatus.Expired, true)] + // legal — a decision its run ended without parking: claimed, then closed by the run's end before its park landed + [InlineData(ToolCallLedgerStatus.Pending, ToolCallLedgerStatus.Expired, true)] // legal — an executing (claimed) row resolves to a terminal [InlineData(ToolCallLedgerStatus.Running, ToolCallLedgerStatus.Succeeded, true)] [InlineData(ToolCallLedgerStatus.Running, ToolCallLedgerStatus.Failed, true)] @@ -30,9 +32,8 @@ public class ToolCallLedgerStateMachineTests // ONLY by TryAnswerDecisionAsync, which guards on tool_kind == 'decision.request', so a real side-effecting approval // row can never reach Succeeded without its Running claim. [InlineData(ToolCallLedgerStatus.AwaitingApproval, ToolCallLedgerStatus.Succeeded, true)] - // illegal — a denial is synchronous-only; an approval can never re-pend or expire from Pending; Running is reached ONLY from AwaitingApproval + // illegal — a denial is synchronous-only; an approval can never re-pend; Running is reached ONLY from AwaitingApproval [InlineData(ToolCallLedgerStatus.AwaitingApproval, ToolCallLedgerStatus.Denied, false)] - [InlineData(ToolCallLedgerStatus.Pending, ToolCallLedgerStatus.Expired, false)] [InlineData(ToolCallLedgerStatus.Pending, ToolCallLedgerStatus.Running, false)] [InlineData(ToolCallLedgerStatus.Running, ToolCallLedgerStatus.Expired, false)] [InlineData(ToolCallLedgerStatus.Running, ToolCallLedgerStatus.Denied, false)] diff --git a/backend/tests/CodeSpace.UnitTests/Architecture/FailureTaxonomyTests.cs b/backend/tests/CodeSpace.UnitTests/Architecture/FailureTaxonomyTests.cs index 16caa1791..fde8af4bb 100644 --- a/backend/tests/CodeSpace.UnitTests/Architecture/FailureTaxonomyTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Architecture/FailureTaxonomyTests.cs @@ -59,6 +59,7 @@ public void The_wire_codes_are_pinned() // A client routes a signed-in user to the password-rotation form on one of these strings, and // opens the identity-link modal on another. They are API, not identifiers. FailureCodes.AgentAuthorityDenied.ShouldBe("agent.authority_denied"); + FailureCodes.AgentRunEnded.ShouldBe("agent-run-ended"); FailureCodes.Forbidden.ShouldBe("forbidden"); FailureCodes.Unauthorized.ShouldBe("unauthorized"); FailureCodes.InvalidCredentials.ShouldBe("invalid_credentials");