Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 16 additions & 2 deletions src/Capacitor.Cli.Core/Models.cs
Original file line number Diff line number Diff line change
Expand Up @@ -1012,6 +1012,7 @@ public sealed record CurationApplyResponse {
[JsonSerializable(typeof(LiveAgentInfo))]
[JsonSerializable(typeof(QuarantinedAgentInfo))]
[JsonSerializable(typeof(DaemonStatusReport))]
[JsonSerializable(typeof(StatusReportRequest))]
// Surface 3 (new-harness detection): per-machine coding-agent inventory carried on the status report.
[JsonSerializable(typeof(Capacitor.Cli.Core.Setup.HarnessInventory))]
[JsonSerializable(typeof(Capacitor.Cli.Core.Setup.HarnessInventoryEntry))]
Expand Down Expand Up @@ -1700,9 +1701,18 @@ public readonly record struct DaemonStatusReport(
long? HighestResolutionGeneration = null,
// Surface 3 (new-harness detection): this machine's coding-agent inventory. Additive/optional
// — recomputed by the daemon on its own 6h in-memory cadence, attached to every report.
Capacitor.Cli.Core.Setup.HarnessInventory? HarnessInventory = null
Capacitor.Cli.Core.Setup.HarnessInventory? HarnessInventory = null,
// Echo of the nonce from a server-sent StatusReportRequest (RequestStatusReport2), null on
// every unsolicited report. The server's idle-marker coordinator confirms a claim only
// against the report carrying ITS nonce — an unechoed report can never confirm one.
string? EchoNonce = null
);

/// <summary>The server's correlated status-report request (hub method <c>RequestStatusReport2</c>,
/// sent only to a daemon that advertised <see cref="DaemonConnect.SupportsCorrelatedStatusReports"/>).
/// The daemon answers with an ordinary <see cref="DaemonStatusReport"/> echoing <see cref="Nonce"/>.</summary>
public readonly record struct StatusReportRequest(string Nonce);

// ── Phase B2-b (sequenced-settlement design): startup-completeness / heal-barrier report DTOs ──

/// <summary>Phase B2-b (sequenced-settlement design): positive per-id death evidence for a
Expand Down Expand Up @@ -1998,7 +2008,11 @@ public readonly record struct DaemonConnect(
// hostable vendors that route permissions through the ACP bridge, computed INDEPENDENTLY of
// unattended certification (a preset is an interactive-launch feature, not a reviewer one).
// Null from a daemon predating this field. Trailing so the wire stays compatible.
string[]? AcpPresetVendors = null
string[]? AcpPresetVendors = null,
// Advertises the RequestStatusReport2 handler (nonce-echoed reports). False from a daemon
// predating it — the server then never sends the correlated request. Trailing name-bound
// field so the wire stays compatible.
bool SupportsCorrelatedStatusReports = false
);

public sealed record UnattendedVendorCapability(
Expand Down
10 changes: 10 additions & 0 deletions src/Capacitor.Cli.Daemon/Services/AgentActivityClock.cs
Original file line number Diff line number Diff line change
Expand Up @@ -108,13 +108,23 @@ public void Advance() {
lock (_gate) AdvanceLocked();
}

/// <summary>Fired on the turn's FALLING edge only (in-flight true → false) — the moment the
/// server needs an out-of-cycle report, since idleness starts here and the next periodic tick is
/// up to 60s away. A turn start needs no report of its own: the delivered input already fires one.</summary>
public Action? OnTurnEnded { get; set; }

/// <summary>Turn start/end — also counts as activity, independent of accompanying envelope
/// traffic.</summary>
public void SetTurnInFlight(bool value) {
bool ended;
lock (_gate) {
ended = _turnInFlight && !value;
_turnInFlight = value;
AdvanceLocked();
}
// Outside the lock, same rule as OnLaunchStageChanged: the callback's send reads this
// clock's own (independently guarded) properties.
if (ended) OnTurnEnded?.Invoke();
}

/// <summary>A handshake stage transition (spawned → initialized → session_created → model_set) —
Expand Down
19 changes: 12 additions & 7 deletions src/Capacitor.Cli.Daemon/Services/AgentOrchestrator.cs
Original file line number Diff line number Diff line change
Expand Up @@ -712,7 +712,8 @@ public AgentOrchestrator(
// RequestStatusReport is answered by an immediate out-of-band DaemonStatusReport.
_server.OnStopAgentV2 += HandleStopAgentV2;
_server.OnAckProcessedPrefix += ack => _processor?.AckPrefix(ack);
_server.OnRequestStatusReport += SendDaemonStatusReportOnceAsync;
_server.OnRequestStatusReport += () => SendDaemonStatusReportOnceAsync();
_server.OnRequestStatusReport2 += nonce => SendDaemonStatusReportOnceAsync(nonce);
_server.GetHighestAcceptedSeq = () => _processor?.HighestAcceptedSeq;
_server.GetLastProcessedSeq = () => _processor?.LastProcessedSeq;
_server.GetQuarantined = () => [.. QuarantineSnapshot()];
Expand Down Expand Up @@ -1061,7 +1062,7 @@ void RefreshHarnessInventoryIfStale() {
/// <summary>Phase B (D2): the daemon's self-report snapshot — its authoritative
/// <see cref="ActiveCount"/> plus the live-agent metadata (and, once D4/Task 8 lands, the
/// kill-quarantine). Pure; the send loop + tests share it.</summary>
internal DaemonStatusReport BuildStatusReport() =>
internal DaemonStatusReport BuildStatusReport(string? echoNonce = null) =>
new(ActiveCount, [.. BuildLiveAgents()], [.. QuarantineSnapshot()],
// Phase B2-b (sequenced-settlement design §4.2.4): re-advertise the durable resolved-
// candidates ledger on every self-report until the server prunes it per-entry via
Expand All @@ -1086,7 +1087,8 @@ internal DaemonStatusReport BuildStatusReport() =>
HighestResolutionGeneration: _resolvedLedger?.HighestResolutionGeneration,
// Surface 3 (new-harness detection): the last cached machine inventory (null until the first
// send refreshes it); the server raises the "installed but not configured" notification from it.
HarnessInventory: CurrentHarnessInventory());
HarnessInventory: CurrentHarnessInventory(),
EchoNonce: echoNonce);

/// <summary>Phase B2-b (sequenced-settlement design): the per-platform startup-reap-complete
/// roll-up. A blocked known-id candidate (pending_marker / legacy_unresolvable /
Expand Down Expand Up @@ -1434,7 +1436,7 @@ async Task RetryQuarantineOnceAsync(CancellationToken ct) {
/// waiting further: invocation order, not completion order, is what the invariant needs, so
/// dropping the wait here is safe while dropping the invocation itself out from under the gate
/// would not be.</summary>
internal async Task SendDaemonStatusReportOnceAsync() {
internal async Task SendDaemonStatusReportOnceAsync(string? echoNonce = null) {
// Surface 3: refresh the machine inventory before building the report. Cheap and self-throttled
// (recomputes at most once per 6h); the first send is the effective startup evaluation.
RefreshHarnessInventoryIfStale();
Expand All @@ -1453,7 +1455,7 @@ internal async Task SendDaemonStatusReportOnceAsync() {
// the same as an awaited failure.
Task? send = null;
try {
send = _server.DaemonStatusReportAsync(BuildStatusReport());
send = _server.DaemonStatusReportAsync(BuildStatusReport(echoNonce));
await send.WaitAsync(StatusReportSendTimeout, _shutdownCts.Token);
} catch (TimeoutException) {
// The invocation already happened under the gate (see the gate's own doc), so wire
Expand Down Expand Up @@ -1592,7 +1594,10 @@ sealed class PendingLaunchScope(AgentOrchestrator owner, string agentId) : IDisp
/// builds its own clock silently loses the stage-report wiring. Shared with
/// <see cref="SeedAgentForTest"/> so tests exercise the same wiring, never a test-only hookup.</summary>
AgentActivityClock CreateActivityClock() =>
new(TimeProvider.System) { OnLaunchStageChanged = () => _ = SendStatusReportNowAsync() };
new(TimeProvider.System) {
OnLaunchStageChanged = () => _ = SendStatusReportNowAsync(),
OnTurnEnded = () => _ = SendStatusReportNowAsync(),
};

/// <summary>Phase B: test-only seam — insert a minimal <see cref="AgentInstance"/> (Noop
/// PTY runtime, no real process/worktree) so unit tests can exercise <see cref="BuildLiveAgents"/>
Expand Down Expand Up @@ -3591,7 +3596,7 @@ async Task HandleSendInput(SendInputCommand cmd) {
// Task.Run, not a bare discard — WaitAsync can complete synchronously on an uncontended
// gate, which would otherwise run BuildStatusReport() (disk I/O) inline on this receive
// loop while still holding BorrowedSnapshotGate.
_ = Task.Run(SendDaemonStatusReportOnceAsync);
_ = Task.Run(() => SendDaemonStatusReportOnceAsync());

LogSendInputDelivered(agentId, agent.Runtime.Vendor, message.Length);
} finally {
Expand Down
16 changes: 15 additions & 1 deletion src/Capacitor.Cli.Daemon/Services/ServerConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,14 @@ internal partial class ServerConnection : IAsyncDisposable, IDaemonHeartbeatPort
public event Func<StopAgentV2, Task>? OnStopAgentV2;
public event Action<AckProcessedPrefix>? OnAckProcessedPrefix;
public event Func<Task>? OnRequestStatusReport;
/// <summary>The correlated variant: the server's <c>RequestStatusReport2</c> hands over a nonce
/// the answering report must echo (<c>DaemonStatusReport.EchoNonce</c>) — sent only to a daemon
/// whose connect advertised <c>SupportsCorrelatedStatusReports</c>.</summary>
public event Func<string, Task>? OnRequestStatusReport2;

/// <summary>What the connect payload claims for RequestStatusReport2 — the live handler itself, so
/// the claim and the routing cannot drift apart.</summary>
internal bool AdvertisesCorrelatedStatusReports => OnRequestStatusReport2 is not null;

/// <summary>Phase B2-b (sequenced-settlement design §4.2.4): snapshot of the un-acked resolved-
/// candidates ledger, re-advertised on <c>DaemonConnect</c> (mirrors <see cref="GetLiveAgents"/>).
Expand Down Expand Up @@ -255,6 +263,8 @@ public ServerConnection(
// OnRequestStatusReport ends in the same gated SendDaemonStatusReportOnceAsync, and awaiting it
// inline here would park this receive loop behind another emission's whole hub send.
_hub.On("RequestStatusReport", () => { _ = Task.Run(() => SafeInvoke("RequestStatusReport", () => OnRequestStatusReport?.Invoke())); return Task.CompletedTask; });
// Same offload shape as RequestStatusReport above, for the same reason.
_hub.On<StatusReportRequest>("RequestStatusReport2", req => { _ = Task.Run(() => SafeInvoke("RequestStatusReport2", () => OnRequestStatusReport2?.Invoke(req.Nonce))); return Task.CompletedTask; });

// Client-result invocations for per-phase eval dispatch.
_hub.On<PrepareEvalCommand, PrepareResult>("PrepareEval",
Expand Down Expand Up @@ -677,7 +687,11 @@ await _hub.InvokeAsync(
// Launch-time ACP permission-preset advertisement: the supported vendors that route
// permissions through the ACP bridge. Null on an unwired/early-startup config;
// wire-compatible with old servers (ignored).
AcpPresetVendors: _config.AcpPresetVendors
AcpPresetVendors: _config.AcpPresetVendors,
// Read off the handler, never asserted: an unwired connection (early startup,
// a test, a second ServerConnection) would otherwise invite RequestStatusReport2
// frames that its null-conditional invoke answers with silence.
SupportsCorrelatedStatusReports: AdvertisesCorrelatedStatusReports
),
cancellationToken: _ct
);
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,125 @@
using System.Text.Json;
using Capacitor.Cli.Core;
using Capacitor.Cli.Daemon.Services;
using Capacitor.Cli.Daemon.Tests.Unit.Pty;
using Microsoft.Extensions.Logging.Abstractions;

namespace Capacitor.Cli.Daemon.Tests.Unit.Services;

/// <summary>
/// The daemon half of the correlated status report: a server-requested report echoes the request's
/// nonce (<see cref="DaemonStatusReport.EchoNonce"/>), which is what lets the server's idle-marker
/// coordinator confirm a claim against THIS report rather than any report — without it every
/// durable idle marker dies un-confirmed. Plus the falling-edge turn report: a turn ending fires an
/// out-of-cycle report so the server learns about idleness at turn end, not at the next 60s tick.
/// </summary>
public class CorrelatedStatusReportTests {
[Test]
public async Task Sent_report_echoes_the_requested_nonce() {
var capture = new CaptureServerConnection();
await using var orch = AgentOrchestratorHarness.BuildOrchestrator(
capture, new SpyPtyProcessFactory(), new Dictionary<string, IHostedAgentLauncher>());

await orch.SendDaemonStatusReportOnceAsync(echoNonce: "nonce-1");

await Assert.That(capture.StatusReports[^1].EchoNonce).IsEqualTo("nonce-1");
}

[Test]
public async Task Unsolicited_reports_carry_no_echo_nonce() {
var capture = new CaptureServerConnection();
await using var orch = AgentOrchestratorHarness.BuildOrchestrator(
capture, new SpyPtyProcessFactory(), new Dictionary<string, IHostedAgentLauncher>());

await orch.SendDaemonStatusReportOnceAsync();

await Assert.That(capture.StatusReports[^1].EchoNonce).IsNull();
}

/// <summary>The wire tokens the server binds by name — a drift here reads as a legacy daemon
/// and the whole confirmation handshake silently goes inert.</summary>
[Test]
public async Task Correlation_fields_serialize_as_the_pinned_wire_tokens() {
var report = new DaemonStatusReport(0, [], [], EchoNonce: "n-1");
var reportJson = JsonSerializer.Serialize(report, CapacitorJsonContext.Default.DaemonStatusReport);
await Assert.That(reportJson).Contains("\"echo_nonce\":\"n-1\"");

var connect = new DaemonConnect("d", "mac", [], 1, [], SupportsCorrelatedStatusReports: true);
var connectJson = JsonSerializer.Serialize(connect, CapacitorJsonContext.Default.DaemonConnect);
await Assert.That(connectJson).Contains("\"supports_correlated_status_reports\":true");

var request = JsonSerializer.Deserialize(
"""{"nonce":"n-2"}""", CapacitorJsonContext.Default.StatusReportRequest);
await Assert.That(request.Nonce).IsEqualTo("n-2");
}

/// <summary>The advertisement is the server's licence to send RequestStatusReport2, and the
/// receive seam answers through a null-conditional invoke — so a connection nothing subscribed to
/// must not claim it, or the request is met with silence rather than a report.</summary>
[Test]
public async Task An_unwired_connection_does_not_advertise_correlated_status_reports() {
var unwired = new ServerConnection(
new DaemonConfig { Name = "test", ServerUrl = "http://127.0.0.1:1" },
NullLoggerFactory.Instance,
NullLogger<ServerConnection>.Instance);
await Assert.That(unwired.AdvertisesCorrelatedStatusReports).IsFalse();

unwired.OnRequestStatusReport2 += _ => Task.CompletedTask;
await Assert.That(unwired.AdvertisesCorrelatedStatusReports).IsTrue();
}

[Test]
public async Task Turn_end_triggers_exactly_one_out_of_cycle_send() {
var server = new StatusReportCountingConnection();
await using var orch = AgentOrchestratorHarness.BuildOrchestrator(
server, new SpyPtyProcessFactory(), new Dictionary<string, IHostedAgentLauncher>());
var agent = orch.SeedAgentForTest("turn-edge-1", status: "Running");

agent.ActivityClock.SetTurnInFlight(true);
await Task.Delay(50);
await Assert.That(server.SendCount).IsEqualTo(0);

agent.ActivityClock.SetTurnInFlight(false);
await WaitHarness.PollUntilAsync(() => server.SendCount >= 1);
await Assert.That(server.SendCount).IsEqualTo(1);

// The repeat is the point: the trigger is the falling edge, not the level.
agent.ActivityClock.SetTurnInFlight(false);
await Task.Delay(50);
await Assert.That(server.SendCount).IsEqualTo(1);
}

[Test]
public async Task Clock_fires_turn_ended_only_on_the_falling_edge() {
var clock = new AgentActivityClock(TimeProvider.System);
var fired = 0;
clock.OnTurnEnded = () => fired++;

clock.SetTurnInFlight(true);
await Assert.That(fired).IsEqualTo(0);

clock.SetTurnInFlight(false);
await Assert.That(fired).IsEqualTo(1);

clock.SetTurnInFlight(false);
await Assert.That(fired).IsEqualTo(1);

clock.SetTurnInFlight(true);
clock.SetTurnInFlight(false);
await Assert.That(fired).IsEqualTo(2);
}

sealed class StatusReportCountingConnection() : ServerConnection(
new() { Name = "test", ServerUrl = "http://127.0.0.1:1" },
NullLoggerFactory.Instance,
NullLogger<ServerConnection>.Instance
) {
int _sendCount;
public int SendCount => Volatile.Read(ref _sendCount);

public override Task DaemonStatusReportAsync(DaemonStatusReport report) {
Interlocked.Increment(ref _sendCount);
return Task.CompletedTask;
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -189,21 +189,21 @@ await Assert.That(probe.CompletedInOrder[0].LiveAgents.Single(a => a.Id == agent
await Assert.That(seqs[1]).IsEqualTo(1UL);
}

/// <summary>Contract pin for the "no correlation nonce" half of the design: the unsolicited
/// delivery-triggered report reuses the ONE report shape, and that shape carries no nonce/echo
/// member at all — so it is structurally incapable of masquerading as the answer to a correlated
/// server-side status request. Should a correlated request/echo ever be added to this DTO, this
/// test fails and forces the unsolicited emission to be re-examined (it must leave the field
/// unset) rather than silently inheriting it.</summary>
/// <summary>The unsolicited delivery-triggered report reuses the ONE report shape but must leave
/// <see cref="DaemonStatusReport.EchoNonce"/> unset: a nonce belongs only to the report answering
/// the server's own correlated request, and an unsolicited report carrying one could masquerade
/// as that answer and confirm an idle-marker claim the daemon never attested for.</summary>
[Test]
public async Task The_report_shape_carries_no_correlation_nonce() {
var reportMembers = typeof(DaemonStatusReport).GetProperties().Select(p => p.Name);
var agentMembers = typeof(LiveAgentInfo).GetProperties().Select(p => p.Name);
public async Task The_delivery_triggered_report_carries_no_echo_nonce() {
var server = new CaptureServerConnection();

foreach (var name in reportMembers.Concat(agentMembers)) {
await Assert.That(name.Contains("nonce", StringComparison.OrdinalIgnoreCase)).IsFalse();
await Assert.That(name.Contains("echo", StringComparison.OrdinalIgnoreCase)).IsFalse();
}
await using var orch = AgentOrchestratorHarness.BuildOrchestrator(server, new SpyPtyProcessFactory(), new Dictionary<string, IHostedAgentLauncher>());
var agent = orch.SeedAgentForTest("dispatch-report-nonce", pty: new RecordingPtyProcess());

await orch.HandleSendInputForTest(new SendInputCommand(agent.Id, "hello", null));

await WaitHarness.PollUntilAsync(() => server.StatusReportCount >= 1);
await Assert.That(server.StatusReports[0].EchoNonce).IsNull();
}

/// <summary><see cref="ServerConnection"/> double that can PARK its first
Expand Down