diff --git a/src/Capacitor.Cli.Core/AgentRegisteredResponse.cs b/src/Capacitor.Cli.Core/AgentRegisteredResponse.cs new file mode 100644 index 0000000000..c91da0f7db --- /dev/null +++ b/src/Capacitor.Cli.Core/AgentRegisteredResponse.cs @@ -0,0 +1,8 @@ +namespace Capacitor.Cli.Core; + +/// The server's answer to AgentRegisteredAcked. Not registered means the server holds no live +/// entry under the agent id. is an open set; a value this daemon does not know +/// is still a refusal. +public sealed record AgentRegisteredResponse(bool Registered, string? RefusedReason = null) { + public static readonly AgentRegisteredResponse Accepted = new(true); +} diff --git a/src/Capacitor.Cli.Core/Models.cs b/src/Capacitor.Cli.Core/Models.cs index 5bf08079ee..9af3029fb2 100644 --- a/src/Capacitor.Cli.Core/Models.cs +++ b/src/Capacitor.Cli.Core/Models.cs @@ -1211,6 +1211,7 @@ public sealed record CurationApplyResponse { [JsonSerializable(typeof(AgentPidRecord))] [JsonSerializable(typeof(PidIdentityKind))] [JsonSerializable(typeof(AgentRegistered))] +[JsonSerializable(typeof(AgentRegisteredResponse))] [JsonSerializable(typeof(AgentStatusChanged))] [JsonSerializable(typeof(AgentUnregistered))] [JsonSerializable(typeof(LaunchFailed))] diff --git a/src/Capacitor.Cli.Daemon/Services/AgentOrchestrator.Resume.cs b/src/Capacitor.Cli.Daemon/Services/AgentOrchestrator.Resume.cs index 05ceb8aabe..4b6f838fca 100644 --- a/src/Capacitor.Cli.Daemon/Services/AgentOrchestrator.Resume.cs +++ b/src/Capacitor.Cli.Daemon/Services/AgentOrchestrator.Resume.cs @@ -474,14 +474,18 @@ internal sealed record ResumeLaunch(ResumeLedgerEntry Entry, WorktreeInfo Worktr /// Ends the runtime start with the attempt: a stop during the handshake must not wait for the vendor. public CancellationToken Token { get; init; } + /// The server's reason, empty when it gave none, once it refused the registration. + public string? RefusedRegistration { get; set; } + public WorkLocation Work => Entry.Launch.Work; public bool IsPrivate => Entry.Launch.IsPrivate; public string Origin => Entry.Launch.Origin; } - const string NativeResumeUnsupported = "native_resume_unsupported"; - const string StoppedDuringResume = "stopped"; - const string TranscriptRefusedReason = "acp_binding_unavailable"; + const string NativeResumeUnsupported = "native_resume_unsupported"; + const string StoppedDuringResume = "stopped"; + const string TranscriptRefusedReason = "acp_binding_unavailable"; + const string RegistrationRefusedReason = "registration_refused"; static bool TranscriptRefused(AgentInstance agent) => agent.ReopenedTranscript == ReopenedTranscript.Refused; @@ -843,7 +847,7 @@ internal async Task ResumeEntryAsync(ResumeLedgerEntry entry, Can if (IsWithdrawn(entry.AgentId)) return ResumeOutcome.Stopped; InterruptEntry(entry.AgentId, LastErrorOf(reason, exitCode)); - if (!entry.Launch.IsPrivate) await ReportResumeFailedAsync(entry, reason, exitCode, ct); + if (!entry.Launch.IsPrivate && reason != RegistrationRefusedReason) await ReportResumeFailedAsync(entry, reason, exitCode, ct); return ResumeOutcome.Interrupted; } @@ -871,6 +875,11 @@ static string LastErrorOf(string reason, int? exitCode) => if (resume.Agent is null && IsWithdrawn(entry.AgentId)) return (StoppedDuringResume, null); + // The server holds no live entry under this id, so the resume goes on under a new one, unless a stop + // came first: it gave up the worktree, and the teardown has removed it. + if (resume.RefusedRegistration is not null) + return resume.Agent?.StopRequested == true ? (StoppedDuringResume, null) : (RegistrationRefusedReason, null); + if (outcome.Kind == CommandOutcomeKind.LaunchRejected) return (outcome.RejectReason == CommandRejectedReason.DaemonCapacity ? "daemon_capacity" : "launch_rejected", null); @@ -947,7 +956,7 @@ internal async Task ContinueEntryAsync(ResumeLedgerEntry entry, s } if (survived && TryHandOverContinuation(entry, launch!.Entry, agent!)) { - await ReportSupersededAsync(entry, ct); + await ReportSupersededAsync(entry, reason, ct); if (entry.SessionId is { } previous) StartContinuationTakeover(agent!, previous); return ResumeOutcome.Continued; @@ -967,7 +976,7 @@ internal async Task ContinueEntryAsync(ResumeLedgerEntry entry, s // A stop of the successor itself ends the agent it continued too. if (agent.StopRequested) { DeleteLedger(entry.AgentId); - await ReportSupersededAsync(entry, ct); + await ReportSupersededAsync(entry, reason, ct); return ResumeOutcome.Stopped; } @@ -990,9 +999,10 @@ bool TryHandOverContinuation(ResumeLedgerEntry entry, ResumeLedgerEntry successo } } - /// A private agent was never registered, so the server holds no agent to end. - Task ReportSupersededAsync(ResumeLedgerEntry entry, CancellationToken ct) => - entry.Launch.IsPrivate + /// A private agent was never registered, and a refused one has no live entry left on the server: + /// neither has an agent to end, and a status for the refused id could overwrite the failure it recorded. + Task ReportSupersededAsync(ResumeLedgerEntry entry, string reason, CancellationToken ct) => + entry.Launch.IsPrivate || reason == RegistrationRefusedReason ? Task.CompletedTask : _server.ReportAgentEndedForShutdownAsync(entry.AgentId, entry.SessionId, "Completed", SupersededReason, null, ct); diff --git a/src/Capacitor.Cli.Daemon/Services/AgentOrchestrator.cs b/src/Capacitor.Cli.Daemon/Services/AgentOrchestrator.cs index 0c4e5f454b..9a666c881d 100644 --- a/src/Capacitor.Cli.Daemon/Services/AgentOrchestrator.cs +++ b/src/Capacitor.Cli.Daemon/Services/AgentOrchestrator.cs @@ -320,6 +320,13 @@ public string? Title { /// public bool IsPrivate { get; init; } + /// Set when the server refused this agent's registration: it holds no live entry under the id, and a + /// status sent for it could overwrite the failure it recorded. The agent keeps running with nothing sent. + public bool RegistrationRefused { get; set; } + + /// No per-agent server call applies: a private agent, or one whose registration was refused. + public bool IsServerSilent => IsPrivate || RegistrationRefused; + /// Owned worktree (daemon-created — safe to remove on cleanup) vs borrowed cwd /// (the user's own checkout — never removed). public WorkLocation Work { get; init; } = WorkLocation.OwnedWorktree; @@ -732,7 +739,7 @@ internal partial class AgentOrchestrator : IAsyncDisposable { /// the sink starts its pump, so admission under the lock leaves no untracked pump and no /// tracked sink that has not started. internal CloudTerminalSink? TryStartCloudSink(AgentInstance agent) { - if (agent.IsPrivate) return null; + if (agent.IsServerSilent) return null; lock (_cloudSinksLock) { if (_cloudSinksClosed) return null; @@ -933,7 +940,7 @@ public AgentOrchestrator( _server.GetLiveAgentIds = () => [ .. _agents - .Where(kvp => (kvp.Value.Status is "Starting" or "Running") && !kvp.Value.IsPrivate) + .Where(kvp => (kvp.Value.Status is "Starting" or "Running") && !kvp.Value.IsServerSilent) .Select(kvp => kvp.Key) ]; @@ -1981,7 +1988,7 @@ async Task RunDaemonStatusReportLoopAsync(CancellationToken ct) { /// a DUPLICATE id, which would double-count against the server's capacity tally. internal IReadOnlyList BuildLiveAgents() { var live = _agents.Values - .Where(a => a.Status is "Starting" or "Running" && !a.IsPrivate) + .Where(a => a.Status is "Starting" or "Running" && !a.IsServerSilent) .Select(a => new LiveAgentInfo( a.Id, a.Kind.ToString(), a.CreatedAt, a.FlowRunId, a.FlowRole, ActivitySeq: a.ActivityClock.ActivitySeq, @@ -2890,7 +2897,23 @@ await LaunchFailedUnlessPrivateAsync(isPrivate, agentId, // routes it through the single-flight cleanup (the agent is already in _agents). PersistPidRecordOrThrow(agent, runtime.Pid, runtime.StartIdentity); - await RegisterAgentAsync(agent); + var registration = await RegisterAgentAsync(agent); + + // A resume the server refused must not run under an id it holds no entry for; the resume + // lane takes it from here, and the pending resume keeps the worktree and ledger entry. + if (!registration.Registered && resume is not null) { + resume.RefusedRegistration = registration.RefusedReason ?? ""; + + try { + await runtime.TerminateAsync(ShutdownTerminateGrace); + } catch (Exception ex) { + LogCleanupStepFailed(ex, "terminating a resume whose registration was refused", agentId); + } + + await CleanupAgentAsync(agentId); + + return new CommandOutcome(CommandOutcomeKind.LaunchFailedCleaned, agentId); + } // Slash commands are reported out of band, and the server drops a report for an agent it // has not yet registered — so attach the callback (which synchronously flushes any list @@ -2899,9 +2922,9 @@ await LaunchFailedUnlessPrivateAsync(isPrivate, agentId, // same way. PTY runtimes never raise the callback; Claude has no live stream, so it is // probed once, off the launch path, and only for an interactive launch (a review-flow // reviewer has no composer). - if (!isPrivate) runtime.OnCommandsAvailable = commands => ReportCommands(agentId, commands); + if (!agent.IsServerSilent) runtime.OnCommandsAvailable = commands => ReportCommands(agentId, commands); - if (!isPrivate && string.Equals(cmd.Vendor, "claude", StringComparison.OrdinalIgnoreCase) && cmd.Kind == LaunchKind.Default) + if (!agent.IsServerSilent && string.Equals(cmd.Vendor, "claude", StringComparison.OrdinalIgnoreCase) && cmd.Kind == LaunchKind.Default) _ = ProbeClaudeCommandsAsync(agentId, worktree.Path); // A runtime with no terminal output (ACP/cursor) has no output-chunk signal to flip @@ -2915,7 +2938,7 @@ await LaunchFailedUnlessPrivateAsync(isPrivate, agentId, if (!runtime.EmitsTerminalOutput) { SetAgentStatus(agent, "Running"); agent.HasReceivedOutput = true; - if (!agent.IsPrivate) TrySendAgentStatus(agent, "Running", null, out _); + if (!agent.IsServerSilent) TrySendAgentStatus(agent, "Running", null, out _); // LaunchStage is Starting-only; cleared at the exact instant this agent leaves it. agent.ActivityClock.ClearLaunchStage(); @@ -2933,7 +2956,10 @@ await LaunchFailedUnlessPrivateAsync(isPrivate, agentId, // A private agent records nothing, so there is no acceptance to wait for. if (isPrivate && start.Transcript is { Reopened: true } privateTranscript) privateTranscript.ReleaseFirstTurn(); - if (!isPrivate && start.Transcript is { } transcript) { + // A refused agent keeps running with nothing sent, so a held first turn goes out without the claim. + if (agent.RegistrationRefused && start.Runtime.RequiresSourceClaimBeforeFirstTurn) _ = BeginRefusedFirstTurnAsync(agent, start.Runtime); + + if (!agent.IsServerSilent && start.Transcript is { } transcript) { // Create + store the per-agent CTS BEFORE firing the setup task, so it exists for // FinalizeAgentRunAsync to cancel even if the agent finalizes before the bind below // ever resolves (see AgentInstance.AcpCts). @@ -2996,12 +3022,12 @@ await LaunchFailedUnlessPrivateAsync(isPrivate, agentId, // Report the resolved model so the server can display / validate the real model the agent // is running. Best-effort: never let a report failure break the launch. - if (!isPrivate && cmd.ExplicitReviewerModel is { } explicitReviewerModel) { + if (!agent.IsServerSilent && cmd.ExplicitReviewerModel is { } explicitReviewerModel) { // Task 8: a protocol-v3 explicit-model reviewer launch reports the CONCRETE // resolved model via the dedicated ReportExplicitReviewerModelResolved channel, keyed by // the durable LaunchAttemptId, so the server's one-shot waiter can validate + re-price it. ReportExplicitReviewerModel(agentId, cmd.Vendor, explicitReviewerModel, runtimeFactory); - } else if (!isPrivate && string.Equals(cmd.Vendor, "codex", StringComparison.OrdinalIgnoreCase)) { + } else if (!agent.IsServerSilent && string.Equals(cmd.Vendor, "codex", StringComparison.OrdinalIgnoreCase)) { // Legacy path (name/arity/behavior UNCHANGED): the dispatched `model` may be the "default" // no-override sentinel, in which case Codex resolves the model from ~/.codex/config.toml. // Codex-only — Claude/other agents never call the ReportAgentResolvedModel hub. @@ -3167,8 +3193,10 @@ void ReportResolvedModel(string agentId, string vendor, string model, bool updat /// void ReportCommands(string agentId, IReadOnlyList commands) { try { - if (_agents.TryGetValue(agentId, out var agent)) + if (_agents.TryGetValue(agentId, out var agent)) { SetCommands(agent, commands); + if (agent.IsServerSilent) return; + } _ = _server.ReportAgentCommandsAsync(agentId, commands); } catch (Exception ex) { @@ -3298,7 +3326,7 @@ async Task ReadAgentOutputAsync(AgentInstance agent) { if (agent.Status == "Starting") { SetAgentStatus(agent, "Running"); - if (!agent.IsPrivate) TrySendAgentStatus(agent, "Running", null, out _); + if (!agent.IsServerSilent) TrySendAgentStatus(agent, "Running", null, out _); RescheduleQuietTurn(); } @@ -3385,7 +3413,7 @@ async Task FailWedgedLaunchAsync(AgentInstance agent, byte[] triggeringChunk, st SetAgentStatus(agent, "Failed"); agent.PendingEndReason = "consent_dialog_wedge"; - if (!agent.IsPrivate) { + if (!agent.IsServerSilent) { _ = _server.LaunchFailedAsync(agent.Id, reason); TrySendAgentStatus(agent, "Failed", null, out _); _ = _server.AppendAgentRunEventAsync(agent.Id, new AgentRunStopped("failed", null)); @@ -3453,7 +3481,7 @@ async Task FinalizeAgentRunAsync(AgentInstance agent) { // The CAS above (already committed) and this status are both visible before we await. SetAgentStatus(agent, "Failed"); - if (!agent.IsPrivate) { + if (!agent.IsServerSilent) { // Own try/catch, deliberately separate from this method's own outer one below — // a report fault must never skip CleanupAgentAsync or the normal unregister // (capacity decrement + the durable death record ride AgentUnregistered and must @@ -3529,14 +3557,14 @@ await _server.LaunchFailedAsync( // worktree, so a startup failure is diagnosable post-mortem (see FailedLaunchLog). PersistFailedLaunchLog(agent, reason); - if (!agent.IsPrivate) _ = _server.LaunchFailedAsync(agent.Id, reason); + if (!agent.IsServerSilent) _ = _server.LaunchFailedAsync(agent.Id, reason); } SetAgentStatus(agent, status); // PrivateLocal agents make no per-agent server calls (deny-all). A refused report must // not skip the local teardown below: that leaks the process and keeps the forwarder alive. - if (!agent.IsPrivate) { + if (!agent.IsServerSilent) { try { if (TrySendAgentStatus(agent, status, null, out var statusSend)) await statusSend; } catch (Exception ex) { @@ -3566,7 +3594,7 @@ await _server.LaunchFailedAsync( // FinalDrainAcpTranscriptAsync's remarks); it always returns within AcpFinalDrainBudget // regardless of outcome. PTY agents have no AcpForwarder and take none of this path — the // runtime is disposed exactly where it always was, inside CleanupAgentAsync below. - if (agent.AcpForwarder is { } acpForwarder) { + if (!agent.IsServerSilent && agent.AcpForwarder is { } acpForwarder) { await FinalDrainAcpTranscriptAsync(agent, acpForwarder); } @@ -3605,7 +3633,7 @@ await _server.LaunchFailedAsync( // server's B5 close authority owns the eventual close. Every other teardown step (the ACP // final-drain above, UnregisterAcpBinding + CleanupAgentAsync below) still runs, so the slot // is freed exactly as a reap frees it. - if (!agent.IsPrivate && agent.PendingEndReason != ReviewerParkedResumableReason) { + if (!agent.IsServerSilent && agent.PendingEndReason != ReviewerParkedResumableReason) { var endTask = _server.EndAgentSessionAsync(agent.Id, agent.PendingEndReason); try { @@ -4115,7 +4143,7 @@ async Task StopAgentCoreAsync(AgentInstance agent) { // The server hears of the stop only once termination is confirmed or refuted, below. An // unregistered agent has no server-side row. - if (!agent.IsPrivate) ClaimRunReport(agent, RunReportOwedByStop); + if (!agent.IsServerSilent) ClaimRunReport(agent, RunReportOwedByStop); // Set status BEFORE cancelling ReadCts so the read loop's finally // block sees "Completed" and skips its own status change / event append. @@ -4904,12 +4932,19 @@ ReviewerModelResolveResponseV1 HandleResolveReviewerModel(ReviewerModelResolveRe /// terminal dims + AgentRunStarted, then persists/announces the repo path. No-ops for a /// PrivateLocal agent. Shared by the hosted launch and the registered local launch so the /// two cannot drift. Dims come from /CurrentRows - /// (hosted = HostedPtyCols/Rows; local = the client's terminal size). + /// (hosted = HostedPtyCols/Rows; local = the client's terminal size). A refusal sends nothing + /// further: the server holds no entry for the dims or the run to attach to. /// - async Task RegisterAgentAsync(AgentInstance agent) { - if (agent.IsPrivate) return; + async Task RegisterAgentAsync(AgentInstance agent) { + if (agent.IsPrivate) return AgentRegisteredResponse.Accepted; - await _server.AgentRegisteredAsync(agent.Id, agent.Prompt, agent.Model, agent.Effort, agent.RepoPath, agent.SandboxPolicy, agent.ApprovalPolicy, agent.PermissionPreset, agent.RuntimeTransport, agent.StartTitle); + var registration = await _server.AgentRegisteredAsync(agent.Id, agent.Prompt, agent.Model, agent.Effort, agent.RepoPath, agent.SandboxPolicy, agent.ApprovalPolicy, agent.PermissionPreset, agent.RuntimeTransport, agent.StartTitle); + + if (!registration.Registered) { + agent.RegistrationRefused = true; + LogRegistrationRefused(agent.Id, registration.RefusedReason); + return registration; + } // Report the PTY size so read-only viewers lock their xterm to it. Best-effort. try { @@ -4933,12 +4968,20 @@ async Task RegisterAgentAsync(AgentInstance agent) { } } ); + + return registration; } /// The only caller of AgentStatusChangedAsync. A failure status is always sent. Any /// other status for a runtime that can publish a termination verdict is initiated under that runtime's - /// gate, so it cannot follow — and clear — a launch failure. + /// gate, so it cannot follow — and clear — a launch failure. Nothing is sent for an agent whose + /// registration was refused. internal bool TrySendAgentStatus(AgentInstance agent, string status, string? sessionId, out Task send) { + if (agent.RegistrationRefused) { + send = Task.CompletedTask; + return false; + } + Task Send() => _server.AgentStatusChangedAsync(agent.Id, status, sessionId ?? agent.SessionId); if (string.Equals(status, "Failed", StringComparison.Ordinal) || agent.Runtime is not ITerminationVerdictSource source) { @@ -5103,6 +5146,16 @@ void StartForwarderAfterBind(AgentInstance agent, IAcpTranscriptSource transcrip agent.AcpForwarder = new AcpForwarderHandle(forwarder, runTask); } + /// The held first turn of an agent the server refused: no claim can be made for it, and holding the + /// turn would leave the agent idle for good. + async Task BeginRefusedFirstTurnAsync(AgentInstance agent, IHostedAgentRuntime runtime) { + try { + await runtime.BeginFirstTurnAsync(agent.ReadCts.Token); + } catch (Exception ex) { + LogCleanupStepFailed(ex, "starting the first turn of an agent whose registration was refused", agent.Id); + } + } + /// /// §2.5 deferred-first-turn SOURCE-CLAIM sequence, fired fire-and-forget after /// RegisterAgentAsync for a runtime that holds its first turn @@ -5363,7 +5416,7 @@ Task HandleResizeTerminal(ResizeTerminalCommand cmd) { /// internal async Task ReRegisterAgentsAsync() { // PrivateLocal agents are never registered with the server, so never re-register them. - foreach (var agent in _agents.Values.Where(a => (a.Status is "Starting" or "Running") && !a.IsPrivate)) { + foreach (var agent in _agents.Values.Where(a => (a.Status is "Starting" or "Running") && !a.IsServerSilent)) { // A published launch-window verdict (or its reported flag) means this agent is terminally // Failed and about to be unregistered by the finalizer — re-sending its (possibly stale // "Running") non-failure Status would clear the FailureReason the verdict's LaunchFailed @@ -5375,7 +5428,31 @@ internal async Task ReRegisterAgentsAsync() { for (var attempt = 1; ; attempt++) { try { - await _server.AgentRegisteredAsync(agent.Id, agent.Prompt, agent.Model, agent.Effort, agent.RepoPath, agent.SandboxPolicy, agent.ApprovalPolicy, agent.PermissionPreset, agent.RuntimeTransport, agent.StartTitle); + var registration = await _server.AgentRegisteredAsync(agent.Id, agent.Prompt, agent.Model, agent.Effort, agent.RepoPath, agent.SandboxPolicy, agent.ApprovalPolicy, agent.PermissionPreset, agent.RuntimeTransport, agent.StartTitle); + + // A refusal is the server's answer, not a transient failure: it is not retried, and the agent is + // left out of later re-registrations too. A status sent for the refused id could clear the + // failure the server recorded for it. + if (!registration.Registered) { + agent.RegistrationRefused = true; + LogRegistrationRefused(agent.Id, registration.RefusedReason); + + // The forwarder would go on sending the transcript under the refused id. + if (agent.AcpCts is { } acpCts) { + try { await acpCts.CancelAsync(); } catch { /* best-effort */ } + } + + // Readiness returns once this pass ends, and a live sink would then mirror under the refused id. + if (agent.CloudSink is { } cloud) { + try { + await cloud.StopAsync(TimeSpan.Zero); + } catch (Exception ex) { + LogCloudSinkStopFailed(ex, agent.Id); + } + } + + break; + } // The acknowledgement is what re-registered means: whatever was written to // the old connection may not have arrived, and that holds even if the sends @@ -5542,7 +5619,7 @@ await discovery.RunAsync(locate, async winner => { _statusNotifier.Pulse(); LogSessionIdDetected(agent.Id, winner.SessionId); - if (agent.IsPrivate || cts.IsCancellationRequested || agent.Status is "Completed" or "Failed") return; + if (agent.IsServerSilent || cts.IsCancellationRequested || agent.Status is "Completed" or "Failed") return; await _server.AppendAgentRunEventAsync(agent.Id, new AgentRunHeartbeat(winner.SessionId)); if (TrySendAgentStatus(agent, agent.Status, winner.SessionId, out var detected)) await detected; }, cts.Token); @@ -5607,6 +5684,8 @@ async Task RunHeartbeatLoopAsync(CancellationToken ct) { continue; } + if (agent.RegistrationRefused) continue; + _ = _server.AppendAgentRunEventAsync( agent.Id, new AgentRunHeartbeat(agent.SessionId) @@ -5708,7 +5787,7 @@ async Task RunTitleResolveLoopAsync(CancellationToken ct) { IReadOnlyList SnapshotAgentsForTitles() => [.. _agents.Values.Select(a => new TitleAgentView( a.Id, a.Vendor, a.Prompt, - a.IsPrivate ? null : a.SessionId ?? (a.Runtime as IAcpTranscriptSource)?.AcpSessionId, + a.IsServerSilent ? null : a.SessionId ?? (a.Runtime as IAcpTranscriptSource)?.AcpSessionId, a.TranscriptPath, a.CreatedAt, a.StartTitle))]; internal static HarnessTitlePost? NativeTitleFor(TitleAgentView agent) => @@ -5811,7 +5890,7 @@ async Task CleanupAgentAsync(string agentId) { // disconnection through SignalR's transport-level signals. Filtered // catch covers the residual race where shutdown fires mid-call. // PrivateLocal agents were never registered, so never unregister them (deny-all). - if (!agent.IsPrivate && !_shutdownCts.IsCancellationRequested) { + if (!agent.IsServerSilent && !_shutdownCts.IsCancellationRequested) { try { await _server.AgentUnregisteredAsync(agentId, StopReasonCode(agent)); } catch (OperationCanceledException) when (_shutdownCts.IsCancellationRequested) { } catch (Exception ex) { LogCleanupStepFailed(ex, "unregistering", agentId); } @@ -5846,7 +5925,7 @@ async Task ReportAgentsEndedForShutdownAsync() { // Only agents this teardown still owns. One whose own finalizer already ran has reported its // real ending and is waiting on cleanup; reporting it again would append a second stop event // over the top of a truer one and race that finalizer's unregister. - var live = _agents.Values.Where(a => !a.IsPrivate && !a.ResumePending && a.Status is "Starting" or "Running").ToList(); + var live = _agents.Values.Where(a => !a.IsServerSilent && !a.ResumePending && a.Status is "Starting" or "Running").ToList(); if (live.Count == 0) return; @@ -6263,6 +6342,9 @@ static string ExtractTerminalText(TerminalOutputBuffer buffer) { [LoggerMessage(Level = LogLevel.Warning, Message = "Failed to re-register agent {AgentId}")] partial void LogReRegisterFailed(Exception ex, string agentId); + [LoggerMessage(Level = LogLevel.Warning, Message = "Server refused the registration of agent {AgentId} ({Reason})")] + partial void LogRegistrationRefused(string agentId, string? reason); + [LoggerMessage(Level = LogLevel.Warning, Message = "Failed to send terminal dimensions for agent {AgentId} (read-only viewers may render garbled until the next reconnect)")] partial void LogTerminalDimsSendFailed(Exception ex, string agentId); @@ -6516,8 +6598,9 @@ internal ReviewerModelResolveResponseV1 HandleResolveReviewerModelForTest(Review string agentId, string vendor, ExplicitReviewerModelLaunch block, IReviewerModelResolver? resolver) => BuildExplicitReviewerModelReport(agentId, vendor, block, resolver); - internal Task RegisterAgentForTestAsync(AgentInstance agent) => RegisterAgentAsync(agent); + internal Task RegisterAgentForTestAsync(AgentInstance agent) => RegisterAgentAsync(agent); internal Task ReRegisterAgentsForTestAsync() => ReRegisterAgentsAsync(); + internal void ReportCommandsForTest(string agentId, IReadOnlyList commands) => ReportCommands(agentId, commands); internal void HandleResizeTerminalForTest(ResizeTerminalCommand cmd) => _ = HandleResizeTerminal(cmd); internal LocalPermissionBridge PermissionBridgeForTest => _permissionBridge; } diff --git a/src/Capacitor.Cli.Daemon/Services/ServerConnection.cs b/src/Capacitor.Cli.Daemon/Services/ServerConnection.cs index a63826f940..967f308d88 100644 --- a/src/Capacitor.Cli.Daemon/Services/ServerConnection.cs +++ b/src/Capacitor.Cli.Daemon/Services/ServerConnection.cs @@ -905,16 +905,40 @@ internal virtual async Task UpdateTitleAsync(string sessionId, string titl } // Outgoing messages to server - public virtual Task AgentRegisteredAsync( + /// Registers the agent and returns the server's answer. A server without AgentRegisteredAcked + /// gets the unanswered AgentRegistered instead, read as accepted; that server is remembered for the + /// rest of the connection, and a reconnect, which may reach an upgraded server, asks again. + public virtual async Task AgentRegisteredAsync( string agentId, string? prompt, string? model, string? effort, string? repoPath, string? sandboxPolicy = null, string? approvalPolicy = null, string? permissionPreset = null, - string? runtimeTransport = null, AgentStartTitle? title = null) - => _hub.InvokeAsync( - "AgentRegistered", - new AgentRegistered( - agentId, prompt, model, effort, repoPath, sandboxPolicy, approvalPolicy, permissionPreset, runtimeTransport, - title?.Text, title?.Derived ?? false), - cancellationToken: _ct); + string? runtimeTransport = null, AgentStartTitle? title = null) { + var registration = new AgentRegistered( + agentId, prompt, model, effort, repoPath, sandboxPolicy, approvalPolicy, permissionPreset, runtimeTransport, + title?.Text, title?.Derived ?? false); + var connection = CurrentConnectionId ?? ""; + + if (!string.Equals(Volatile.Read(ref _registrationAckMissingOn), connection, StringComparison.Ordinal)) { + try { + return await InvokeAgentRegisteredAckedRawAsync(registration) + ?? throw new InvalidOperationException($"The server answered the registration of agent {agentId} with no outcome"); + } catch (Exception ex) when (IsUnknownHubMethod(ex)) { + Volatile.Write(ref _registrationAckMissingOn, connection); + } + } + + await InvokeAgentRegisteredRawAsync(registration); + + return AgentRegisteredResponse.Accepted; + } + + /// The connection id on which the server answered AgentRegisteredAcked as an unknown method. + string? _registrationAckMissingOn; + + internal virtual Task InvokeAgentRegisteredAckedRawAsync(AgentRegistered registration) + => _hub.InvokeAsync("AgentRegisteredAcked", registration, cancellationToken: _ct); + + internal virtual Task InvokeAgentRegisteredRawAsync(AgentRegistered registration) + => _hub.InvokeAsync("AgentRegistered", registration, cancellationToken: _ct); /// /// Reports the hosted agent's fixed PTY dimensions to the server, which stores diff --git a/test/Capacitor.Cli.Core.Tests.Unit/AgentRegisteredResponseWireTests.cs b/test/Capacitor.Cli.Core.Tests.Unit/AgentRegisteredResponseWireTests.cs new file mode 100644 index 0000000000..840e9b4fd6 --- /dev/null +++ b/test/Capacitor.Cli.Core.Tests.Unit/AgentRegisteredResponseWireTests.cs @@ -0,0 +1,24 @@ +using System.Text.Json; + +namespace Capacitor.Cli.Core.Tests.Unit; + +/// The hub client deserializes the AgentRegisteredAcked result through the AOT context, so the +/// type must be registered there and read the server's snake_case shape. +public class AgentRegisteredResponseWireTests { + [Test] + public async Task A_refusal_round_trips_through_the_aot_context_in_snake_case() { + var refused = new AgentRegisteredResponse(false, "ownership_fence"); + + var json = JsonSerializer.Serialize(refused, CapacitorJsonContext.Default.AgentRegisteredResponse); + + await Assert.That(json).IsEqualTo("""{"registered":false,"refused_reason":"ownership_fence"}"""); + await Assert.That(JsonSerializer.Deserialize(json, CapacitorJsonContext.Default.AgentRegisteredResponse)).IsEqualTo(refused); + } + + [Test] + public async Task An_acceptance_without_a_reason_reads_as_registered() { + var accepted = JsonSerializer.Deserialize("""{"registered":true}""", CapacitorJsonContext.Default.AgentRegisteredResponse); + + await Assert.That(accepted).IsEqualTo(AgentRegisteredResponse.Accepted); + } +} diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorCloudSinkTests.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorCloudSinkTests.cs index 42f5727973..5de448a9f1 100644 --- a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorCloudSinkTests.cs +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorCloudSinkTests.cs @@ -1,4 +1,5 @@ using System.Text; +using Capacitor.Cli.Core; using Capacitor.Cli.Daemon.Services; using Capacitor.Cli.Daemon.Tests.Unit.Pty; @@ -135,6 +136,34 @@ public async Task A_private_agent_never_sends_terminal_output() { await Assert.That(agent.CloudSink).IsNull(); } + /// A reconnect the server answers with a refusal ends the agent's mirror: output after it stays local. + [Test] + public async Task A_reconnect_refusal_stops_the_agents_mirror() { + var server = new CaptureServerConnection(); + + await using var orch = Build(server); + + var pty = new ScriptedPtyProcess(); + var agent = orch.SeedAgentForTest("refused", status: "Running", pty: pty); + var local = AttachLocal(agent); + var loop = orch.ReadAgentOutputForTest(agent); + + pty.Emit("before;"); + await WaitHarness.PollUntilAsync(() => server.TerminalSends.Any(s => s.AgentId == "refused")); + + server.RegistrationResponse = _ => new AgentRegisteredResponse(false, "superseded"); + await server.ReRegisterAgentsHook!(); + var sendsAtRefusal = server.TerminalSendStarts; + + pty.Emit("after;"); + await WaitHarness.PollUntilAsync(() => local.Chunks.Count == 2); + pty.Exit(); + await loop.WaitAsync(WaitHarness.Bounded); + + await Assert.That(server.TerminalSendStarts).IsEqualTo(sendsAtRefusal); + await Assert.That(string.Concat(MirrorOf(server, "refused"))).DoesNotContain("after;"); + } + [Test] public async Task No_terminal_send_starts_after_the_agent_unregisters() { // OnAgentUnregistered is init-only, so the callback reaches the connection and the agent diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorContinueTests.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorContinueTests.cs index 96c7676545..98d762276c 100644 --- a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorContinueTests.cs +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorContinueTests.cs @@ -1,5 +1,6 @@ using System.Collections.Concurrent; using System.Text.Json.Nodes; +using Capacitor.Cli.Core; using Capacitor.Cli.Core.LocalIpc; using Capacitor.Cli.Core.Resume; using Capacitor.Cli.Daemon.Services; @@ -95,7 +96,7 @@ Fixture Build( return new Fixture(orch, server, factory, captured!, repo); } - ResumeLedgerEntry Entry(Fixture f, bool isPrivate = false, bool standalone = false) { + ResumeLedgerEntry Entry(Fixture f, bool isPrivate = false, bool standalone = false, string? vendorSession = null) { var stamp = TimeProvider.System.GetUtcNow(); var worktree = Tmp.CreateDir("wt-old"); if (standalone) Directory.CreateDirectory(Path.Combine(worktree, ".git")); @@ -109,7 +110,7 @@ ResumeLedgerEntry Entry(Fixture f, bool isPrivate = false, bool standalone = fal WorktreeBranch: "capacitor/custom-old", WorktreeIsStandalone: standalone, WorktreeSourceRepo: f.Repo); var entry = new ResumeLedgerEntry( ResumeLedgerEntry.CurrentSchemaVersion, OldAgent, "daemon-1", stamp, stamp, "cursor", "acp", - OldSession, VendorSession: null, launch, ResumeState.Resuming, Attempts: 0, LastError: null, + OldSession, vendorSession, launch, ResumeState.Resuming, Attempts: 0, LastError: null, PredecessorInstanceId: "instance-old", ResumingAt: stamp, DaemonInstanceId: null); f.Orch.ResumeLedger!.Write(entry); @@ -223,6 +224,70 @@ await Assert.That(f.Server.ShutdownReports.Select(r => (r.AgentId, r.Status, r.R await Assert.That(f.Orch.ResumeLedger!.TryRead(OldAgent)).IsNull(); } + /// A refused same-id registration leaves the server no live entry for the resumed agent: the + /// resume goes on under a new agent id, and nothing further is sent under the refused one. + [Test] + public async Task A_native_resume_whose_registration_is_refused_continues_under_a_new_agent_id() { + var f = Build(resumeSupported: true); + await using var _ = f.Orch; + f.Server.RegistrationResponse = id => id == OldAgent ? new AgentRegisteredResponse(false, "recorded_failure") : null; + var entry = Entry(f, vendorSession: "vendor-session-1"); + + f.Orch.LoadResumeLedger(); + await f.Orch.ResumeOnBootAsync(CancellationToken.None); + + var starts = f.Factory.Starts.Select(s => s.Ctx).ToList(); + await Assert.That(starts.Count).IsEqualTo(2); + await Assert.That(starts[0].AgentId).IsEqualTo(OldAgent); + await Assert.That(starts[0].ResumeSessionId).IsEqualTo("vendor-session-1"); + + var successor = starts[1].AgentId; + await Assert.That(successor).IsNotEqualTo(OldAgent); + await Assert.That(starts[1].Prompt).Contains("(registration_refused)"); + await Assert.That(starts[1].Worktree.Path).IsEqualTo(entry.Launch.WorktreePath); + + await Assert.That(f.Server.AgentRegisteredCalls.Select(c => c.AgentId)).IsEquivalentTo([OldAgent, successor]); + await Assert.That(f.Server.StatusChangedCalls.Any(c => c.AgentId == OldAgent)).IsFalse(); + await Assert.That(f.Server.AgentUnregisteredCalls).DoesNotContain(OldAgent); + await Assert.That(f.Server.RunEvents.Any(e => e.AgentId == OldAgent)).IsFalse(); + await Assert.That(f.Server.AcpSessionStartedCalls.Any(c => c.AgentId == OldAgent)).IsFalse(); + await Assert.That(f.Server.ShutdownReports).IsEmpty(); + + await Assert.That(f.Orch.GetAgentForTest(OldAgent)).IsNull(); + await Assert.That(f.Orch.GetAgentForTest(successor)).IsNotNull(); + await Assert.That(f.Orch.ResumeLedger!.TryRead(OldAgent)).IsNull(); + await Assert.That(f.Orch.ResumeLedger!.TryRead(successor)!.State).IsEqualTo(ResumeState.Running); + } + + /// A stop that lands while the registration is still unanswered gives up the worktree, so the + /// refusal that follows must not start a successor in it. + [Test] + public async Task A_stop_while_a_refused_registration_is_pending_starts_no_successor() { + var f = Build(resumeSupported: true); + await using var _ = f.Orch; + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + f.Server.AgentRegisteredEntered = entered; + f.Server.AgentRegisteredGate = gate; + f.Server.RegistrationResponse = _ => new AgentRegisteredResponse(false, "recorded_failure"); + Entry(f, vendorSession: "vendor-session-1"); + + f.Orch.LoadResumeLedger(); + var lane = f.Orch.ResumeOnBootAsync(CancellationToken.None); + await WaitHarness.WaitBoundedAsync(entered.Task, "the resumed registration"); + + var stop = f.Orch.HandleStopAgent(OldAgent); + await WaitHarness.PollUntilAsync(() => f.Orch.GetAgentForTest(OldAgent) is not { StopRequested: false }); + gate.SetResult(); + + await WaitHarness.WaitBoundedAsync(lane, "the resume lane"); + await WaitHarness.WaitBoundedAsync(stop, "the stop"); + + await Assert.That(f.Factory.Starts.Count).IsEqualTo(1); + await Assert.That(f.Orch.GetAgentForTest(OldAgent)).IsNull(); + await Assert.That(f.Server.ShutdownReports.Any(r => r.Reason.StartsWith("resume_failed", StringComparison.Ordinal))).IsFalse(); + } + [Test] public async Task A_private_entry_skips_the_superseded_report_but_still_runs_the_takeover() { ServeOldSession(); diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorLocalAttachTests.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorLocalAttachTests.cs index da6e161833..d785076f7b 100644 --- a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorLocalAttachTests.cs +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorLocalAttachTests.cs @@ -1123,7 +1123,7 @@ sealed class TripwireServerConnection() : ServerConnection( public override Task SendTerminalDimensionsAsync(string agentId, int cols, int rows) { LastDims = (cols, rows); Calls.Add(nameof(SendTerminalDimensionsAsync)); return Task.CompletedTask; } public override Task LaunchFailedAsync(string agentId, string reason) { Calls.Add(nameof(LaunchFailedAsync)); return Task.CompletedTask; } public ConcurrentBag RegisteredTitles { get; } = []; - public override Task AgentRegisteredAsync(string agentId, string? prompt, string? model, string? effort, string? repoPath, string? sandboxPolicy = null, string? approvalPolicy = null, string? permissionPreset = null, string? runtimeTransport = null, AgentStartTitle? title = null) { Calls.Add(nameof(AgentRegisteredAsync)); RegisteredTitles.Add(title); return Task.CompletedTask; } + public override Task AgentRegisteredAsync(string agentId, string? prompt, string? model, string? effort, string? repoPath, string? sandboxPolicy = null, string? approvalPolicy = null, string? permissionPreset = null, string? runtimeTransport = null, AgentStartTitle? title = null) { Calls.Add(nameof(AgentRegisteredAsync)); RegisteredTitles.Add(title); return Task.FromResult(AgentRegisteredResponse.Accepted); } public override Task AgentStatusChangedAsync(string agentId, string status, string? sessionId) { Calls.Add(nameof(AgentStatusChangedAsync)); return Task.CompletedTask; } public override Task AgentUnregisteredAsync(string agentId, string? stopReason = null) { Calls.Add(nameof(AgentUnregisteredAsync)); return Task.CompletedTask; } public override Task UpdateRepoPathsAsync() { Calls.Add(nameof(UpdateRepoPathsAsync)); return Task.CompletedTask; } diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorVendorTests.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorVendorTests.cs index 4eb293d996..0d6af38718 100644 --- a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorVendorTests.cs +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/AgentOrchestratorVendorTests.cs @@ -85,6 +85,63 @@ public async Task ReRegister_retries_a_transient_per_agent_failure_then_succeeds await Assert.That(server.AgentRegisteredCallCount).IsEqualTo(2); } + /// A refusal is the server's answer rather than a transient failure: no retry, and no status that + /// could overwrite what the server recorded for the refused id. + [Test] + public async Task ReRegister_neither_retries_a_refusal_nor_sends_a_status_after_it() { + using var worktree = new TempDir(); + string worktreePath = worktree.CreateDir("worktree"); + var server = new CaptureServerConnection { RegistrationResponse = _ => new AgentRegisteredResponse(false, "recorded_failure") }; + + await using var orch = AgentOrchestratorHarness.BuildOrchestrator(server, new SpyPtyProcessFactory(), new Dictionary()); + + orch.RegisterAgentForTest(new AgentInstance( + "agent-refused", null, "", null, worktreePath, "claude", + new PtyHostedAgentRuntime("claude", new StubPtyProcess(), TimeProvider.System), new WorktreeInfo(worktreePath, "", worktreePath, IsStandalone: true), new CancellationTokenSource() + ) { + ActivityClock = new AgentActivityClock(TimeProvider.System), + CreatedAt = DateTime.UtcNow, + LastOutputAt = DateTime.UtcNow + }); + + await server.ReRegisterAgentsHook!(); + await server.ReRegisterAgentsHook!(); + orch.ReportCommandsForTest("agent-refused", [new HostedAgentCommand("review", null, null)]); + var commands = orch.GetAgentForTest("agent-refused")!.Commands; + await orch.CleanupAgentForTest("agent-refused"); + + await Assert.That(server.AgentRegisteredCallCount).IsEqualTo(1); + await Assert.That(server.StatusChangedCalls).IsEmpty(); + await Assert.That(server.AgentUnregisteredCalls).IsEmpty(); + await Assert.That(server.CommandReports).IsEmpty(); + await Assert.That(commands!.Single().Name).IsEqualTo("review"); + } + + /// A refused fresh launch keeps running locally, and nothing more reaches the server under its id. + [Test] + public async Task A_refused_launch_keeps_running_and_sends_nothing_more_for_its_id() { + using var repoPath = GitRepo.CreateWithCommit(); + var server = new CaptureServerConnection { RegistrationResponse = _ => new AgentRegisteredResponse(false, "launch_refused") }; + + await using var orch = AgentOrchestratorHarness.BuildOrchestrator(server, new SpyPtyProcessFactory(), AgentOrchestratorHarness.Launcher("claude"), allowedRepoPath: repoPath); + + await orch.HandleLaunchAgentForTest(new LaunchAgentCommand( + AgentId: "refused-launch", Prompt: "hi", Model: "opus", Effort: null, + RepoPath: repoPath, Tools: null, AttachmentIds: null, Vendor: "claude")); + + var agent = orch.GetAgentForTest("refused-launch"); + await Assert.That(agent).IsNotNull(); + await Assert.That(agent!.IsServerSilent).IsTrue(); + + await orch.CleanupAgentForTest("refused-launch"); + + await Assert.That(server.AgentRegisteredCallCount).IsEqualTo(1); + await Assert.That(server.StatusChangedCalls).IsEmpty(); + await Assert.That(server.RunEvents.Any(e => e.AgentId == "refused-launch")).IsFalse(); + await Assert.That(server.LaunchFailedCalls).IsEmpty(); + await Assert.That(server.AgentUnregisteredCalls).IsEmpty(); + } + // A PTY codex runtime reports "pty" on AgentRegistered — the transport rides the runtime TYPE, // not the vendor (a codex agent is not automatically app-server). [Test] diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/CaptureServerConnection.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/CaptureServerConnection.cs index df24f36462..92fa27679e 100644 --- a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/CaptureServerConnection.cs +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/CaptureServerConnection.cs @@ -90,11 +90,11 @@ public override Task DaemonStatusReportAsync(DaemonStatusReport report) { /// Completed the instant AgentRegisteredAsync is entered — lets a test hold /// re-registration inside that await and publish a verdict there (finding 1 refinement). - public TaskCompletionSource? AgentRegisteredEntered { get; init; } + public TaskCompletionSource? AgentRegisteredEntered { get; set; } /// When set, AgentRegisteredAsync awaits this before returning — holding the /// re-registration open across the AgentRegistered await. - public TaskCompletionSource? AgentRegisteredGate { get; init; } + public TaskCompletionSource? AgentRegisteredGate { get; set; } /// When set, every non-failure AgentStatusChanged send checks this runtime's verdict /// AT SEND TIME (via the lock-synchronised ReadVerdict); if a launch-window verdict is already @@ -119,7 +119,10 @@ public override Task DaemonStatusReportAsync(DaemonStatusReport report) { /// The start title echoed on each registration, in call order. public List<(string AgentId, AgentStartTitle? Title)> AgentRegisteredTitles { get; } = []; - public override async Task AgentRegisteredAsync( + /// The server's answer per agent id; accepted when unset or null. + public Func? RegistrationResponse { get; set; } + + public override async Task AgentRegisteredAsync( string agentId, string? prompt, string? model, string? effort, string? repoPath, string? sandboxPolicy = null, string? approvalPolicy = null, string? permissionPreset = null, string? runtimeTransport = null, AgentStartTitle? title = null) { @@ -135,6 +138,8 @@ public override async Task AgentRegisteredAsync( if (AgentRegisteredCallCount <= AgentRegisteredFailTimes) throw new InvalidOperationException("transient re-register failure"); + + return RegistrationResponse?.Invoke(agentId) ?? AgentRegisteredResponse.Accepted; } // ── Option B task 4: ACP bind/forward capture ──────────────────────────────────── @@ -428,6 +433,13 @@ public override async Task ReportAgentEndedForShutdownAsync( /// await the fire-and-forget report deterministically instead of racing Task.Delay. public Channel ExplicitReviewerModelReportSignal { get; } = Channel.CreateUnbounded(); + public List CommandReports { get; } = []; + + public override Task ReportAgentCommandsAsync(string agentId, IReadOnlyList commands) { + lock (CommandReports) CommandReports.Add(agentId); + return Task.CompletedTask; + } + public override Task ReportAgentResolvedModelAsync(string agentId, string model) { lock (ReportAgentResolvedModelCalls) ReportAgentResolvedModelCalls.Add((agentId, model)); diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/SeqCaptureServerConnection.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/SeqCaptureServerConnection.cs index 885f513409..401902bfcc 100644 --- a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/SeqCaptureServerConnection.cs +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/SeqCaptureServerConnection.cs @@ -29,8 +29,8 @@ internal sealed class SeqCaptureServerConnection() : ServerConnection( public override Task CommandRejectedAsync(CommandRejected rej) { lock (Rejects) Rejects.Add(rej); return Task.CompletedTask; } public override Task CommandAckAsync(CommandAck ack) { lock (Acks) Acks.Add(ack); return Task.CompletedTask; } - public override Task AgentRegisteredAsync( + public override Task AgentRegisteredAsync( string agentId, string? prompt, string? model, string? effort, string? repoPath, string? sandboxPolicy = null, string? approvalPolicy = null, string? permissionPreset = null, - string? runtimeTransport = null, AgentStartTitle? title = null) => Task.CompletedTask; + string? runtimeTransport = null, AgentStartTitle? title = null) => Task.FromResult(AgentRegisteredResponse.Accepted); } diff --git a/test/Capacitor.Cli.Daemon.Tests.Unit/Services/ServerConnectionAgentRegisteredTests.cs b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/ServerConnectionAgentRegisteredTests.cs new file mode 100644 index 0000000000..421fbb88f9 --- /dev/null +++ b/test/Capacitor.Cli.Daemon.Tests.Unit/Services/ServerConnectionAgentRegisteredTests.cs @@ -0,0 +1,118 @@ +using Capacitor.Cli.Core; +using Capacitor.Cli.Daemon.Services; +using Microsoft.AspNetCore.SignalR; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Capacitor.Cli.Daemon.Tests.Unit.Services; + +/// Registration asks the server for its answer through AgentRegisteredAcked, and falls back to +/// the unanswered AgentRegistered against a server that has no such method. +public class ServerConnectionAgentRegisteredTests { + sealed class TestServerConnection() : ServerConnection( + new DaemonConfig { Name = "test", ServerUrl = "http://127.0.0.1:1" }, + new FixedCredentialSource(), + NullLoggerFactory.Instance, + NullLogger.Instance, + TimeProvider.System) { + public string? ConnectionId { get; set; } = "connection-1"; + + internal override string? CurrentConnectionId => ConnectionId; + + public Func> Acked { get; set; } = _ => Task.FromResult(AgentRegisteredResponse.Accepted); + + public List AckedCalls { get; } = []; + public List LegacyCalls { get; } = []; + + internal override Task InvokeAgentRegisteredAckedRawAsync(AgentRegistered registration) { + AckedCalls.Add(registration); + return Acked(registration); + } + + internal override Task InvokeAgentRegisteredRawAsync(AgentRegistered registration) { + LegacyCalls.Add(registration); + return Task.CompletedTask; + } + } + + static Task Register(ServerConnection conn, string agentId = "agent-1") => + conn.AgentRegisteredAsync(agentId, "prompt", "model", null, "/repo", runtimeTransport: "pty"); + + static Task UnknownMethod(AgentRegistered _) => + Task.FromException(new HubException("Unknown hub method 'AgentRegisteredAcked'")); + + [Test] + public async Task An_accepted_registration_is_returned_and_sends_only_the_acked_call() { + var conn = new TestServerConnection(); + + var response = await Register(conn); + + await Assert.That(response.Registered).IsTrue(); + await Assert.That(conn.AckedCalls.Single().AgentId).IsEqualTo("agent-1"); + await Assert.That(conn.AckedCalls.Single().RuntimeTransport).IsEqualTo("pty"); + await Assert.That(conn.LegacyCalls).IsEmpty(); + } + + [Test] + public async Task A_refusal_is_returned_with_its_reason() { + var conn = new TestServerConnection { + Acked = _ => Task.FromResult(new AgentRegisteredResponse(false, "recorded_failure")) + }; + + var response = await Register(conn); + + await Assert.That(response).IsEqualTo(new AgentRegisteredResponse(false, "recorded_failure")); + await Assert.That(conn.LegacyCalls).IsEmpty(); + } + + [Test] + public async Task A_server_without_the_acked_method_gets_the_legacy_call_and_reads_as_accepted() { + var conn = new TestServerConnection { Acked = UnknownMethod }; + + var response = await Register(conn); + + await Assert.That(response.Registered).IsTrue(); + await Assert.That(conn.LegacyCalls.Single().AgentId).IsEqualTo("agent-1"); + await Assert.That(conn.LegacyCalls.Single().RuntimeTransport).IsEqualTo("pty"); + } + + /// The missing method is remembered per connection: a reconnect may reach an upgraded server. + [Test] + public async Task The_missing_method_is_not_probed_again_until_the_connection_changes() { + var conn = new TestServerConnection { Acked = UnknownMethod }; + + await Register(conn, "agent-1"); + await Register(conn, "agent-2"); + + await Assert.That(conn.AckedCalls.Count).IsEqualTo(1); + await Assert.That(conn.LegacyCalls.Select(c => c.AgentId)).IsEquivalentTo(["agent-1", "agent-2"]); + + conn.ConnectionId = "connection-2"; + conn.Acked = _ => Task.FromResult(new AgentRegisteredResponse(false, "no_capacity")); + + var response = await Register(conn, "agent-3"); + + await Assert.That(conn.AckedCalls.Count).IsEqualTo(2); + await Assert.That(response.Registered).IsFalse(); + await Assert.That(conn.LegacyCalls.Count).IsEqualTo(2); + } + + /// A reply with no outcome proves nothing either way, so it is a failure rather than an acceptance. + [Test] + public async Task A_reply_without_an_outcome_fails_rather_than_reading_as_accepted() { + var conn = new TestServerConnection { Acked = _ => Task.FromResult(null) }; + + await Assert.That(async () => await Register(conn)).Throws(); + await Assert.That(conn.LegacyCalls).IsEmpty(); + } + + /// Any other hub failure is no proof the method is missing, so it surfaces to the caller's retry. + [Test] + public async Task Another_hub_failure_propagates_without_the_legacy_call() { + var conn = new TestServerConnection { + Acked = _ => Task.FromException(new HubException("Caller is not a registered daemon")) + }; + + await Assert.That(async () => await Register(conn)).Throws(); + await Assert.That(conn.LegacyCalls).IsEmpty(); + } +}