From dc7cc072aa005421183c9d81d176c09787a663d0 Mon Sep 17 00:00:00 2001 From: skulitom Date: Tue, 6 Oct 2026 23:30:04 +0100 Subject: [PATCH 1/4] Take queue time off a forwarded request's deadline MCP and the CLI give a seat request 60 s, and the daemon forwards it on its one connection to the seat host with the same timeoutMs. The seat host started a fresh clock when the request arrived, so time spent queued behind another long operation, such as typing, was counted twice. A queued request could still be running when the daemon gave up on it: JsonPipeClient then closed the connection, the daemon marked the seat detached, and the reply was lost. JsonPipeClient.RequestAsync can now send the time left as timeoutMs when the request goes out (sendTimeLeft), and the daemon's forwards use it through AnodeDaemon.SendOnAsync. In the seat host, DesktopLease takes its own wait for an earlier action off before dispatch, and HandleOwnedAsync takes off a wait for the previous lease's display restore. Every operation that bounds its work by timeoutMs (typing, emulator start, Steam launch) then answers in time. No new envelope field: timeoutMs already exists on both hops, so an older peer is unaffected. Reported in Compass's Anode inbox as B-017. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 5 ++++ docs/PROTOCOL.md | 5 +++- src/Anode/Cli/AgentChecks.cs | 38 +++++++++++++++++++++++++ src/Anode/Cli/SelfTest.cs | 2 ++ src/Anode/Cli/TransportChecks.cs | 41 +++++++++++++++++++++++++++ src/Anode/Core/Agents/AgentAccess.cs | 9 ++++++ src/Anode/Core/Agents/DesktopLease.cs | 3 ++ src/Anode/Core/Bridge/Pipes.cs | 9 +++++- src/Anode/Daemon/AnodeDaemon.cs | 12 ++++++-- src/Anode/Seat/SeatHost.cs | 9 ++++-- 10 files changed, 127 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 91dfedd..88643bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ reuse window. - **Quicker startup feedback.** Readiness, failure and Stop wake waiting startup calls immediately. MCP connects as soon as it launches the daemon, and failed host connections retry sooner. +- **Queued requests keep their deadline.** A request that waited behind another long operation, such as + typing, got a full 60 seconds of its own once it reached the seat. So it could still be running when + the daemon gave up on it: the reply was lost and the seat showed as `detached` until it reconnected. + The daemon now forwards the time a request has left, and the seat takes its own waits off as well, so + an operation queued behind another ends, or is refused, in time. - **Typing keeps pace with the program.** `seat_type` and `anode type` sent characters as fast as Windows took them, so a busy program, such as Chrome editing a long field, could drop and reorder them while the call still reported success. Characters now go out 15 ms apart by default, timed diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 7565b52..752ed85 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -21,7 +21,10 @@ remains usable for the next request. flat on the same object. `timeoutMs` on a request bound for the seat overrides the daemon's default 60 second forwarding timeout. Client deadlines include queueing, writing and reading. A timeout after sending closes that connection, because a late reply must not become the next command's -result. Reconnect for subsequent requests; a timed-out command may already have executed. +result. Reconnect for subsequent requests; a timed-out command may already have executed. The +daemon forwards a request with `timeoutMs` set to the time it has left after waiting behind other +operations on its one connection to the seat host, and the seat host takes its own waits off as well, +so an operation queued behind another, such as a long `input.text`, still answers in time. Desktop operations additionally require an `agentId` and live `leaseToken`, acquired through `lease` with `action: "acquire"`, and may carry an `agentName` for status and the viewer. `lease` diff --git a/src/Anode/Cli/AgentChecks.cs b/src/Anode/Cli/AgentChecks.cs index 3944886..1c24db8 100644 --- a/src/Anode/Cli/AgentChecks.cs +++ b/src/Anode/Cli/AgentChecks.cs @@ -271,6 +271,44 @@ async Task Dispatch(JsonObject request, CancellationToken cancel) return "private clients contend, expired queued input is fenced, disconnect recovers by expiry, and Stop bypasses ownership"; } + public static async Task QueueTime() + { + long now = 0; + var entered = Signal(); var finish = Signal(); + var lease = new DesktopLease(milliseconds: () => Interlocked.Read(ref now)); + var given = new System.Collections.Concurrent.ConcurrentDictionary(); + async Task Dispatch(JsonObject request, CancellationToken cancel) + { + given[request.Str("text")!] = request.Int("timeoutMs"); + if (request.Str("text") == "first") + { + entered.TrySetResult(); + await finish.Task.WaitAsync(cancel); + } + return JsonLine.Ok(); + } + string token = (await lease.HandleAsync(Lease("A", "acquire"), Dispatch)).Obj("result")!.Str("leaseToken")!; + Task active = lease.HandleAsync(Input("A", token, "first"), Dispatch); + Task plain, own; + try + { + await entered.Task.WaitAsync(TimeSpan.FromSeconds(2)); + // HandleAsync runs synchronously up to its occupied gate, so these two are known to be queued. + plain = lease.HandleAsync(Input("A", token, "plain"), Dispatch); + var deadline = Input("A", token, "own deadline"); + deadline["timeoutMs"] = 20_000; + own = lease.HandleAsync(deadline, Dispatch); + Interlocked.Exchange(ref now, 7_000); + } + finally { finish.TrySetResult(); } + foreach (var call in new[] { active, plain, own }) + Require((await call.WaitAsync(TimeSpan.FromSeconds(2))).Bool("ok") == true, "a queued action failed"); + Require(given["first"] is null && given["plain"] == 53_000 && given["own deadline"] == 13_000, + $"queued actions were not given the time left: {given["first"]}, {given["plain"]} and {given["own deadline"]} ms, " + + "expected none, 53000 and 13000"); + return "an action queued behind another of its agent's runs in the time its caller has left"; + } + public static async Task Line() { long now = 0; diff --git a/src/Anode/Cli/SelfTest.cs b/src/Anode/Cli/SelfTest.cs index b4978db..5d16451 100644 --- a/src/Anode/Cli/SelfTest.cs +++ b/src/Anode/Cli/SelfTest.cs @@ -103,6 +103,7 @@ public static int Run(string[] args) }); Check(results, "pipe queue deadline", () => TransportChecks.QueueDeadline().GetAwaiter().GetResult()); + Check(results, "pipe forwarded time left", () => TransportChecks.ForwardTimeLeft().GetAwaiter().GetResult()); Check(results, "pipe late reply", () => TransportChecks.LateReply().GetAwaiter().GetResult()); Check(results, "pipe reply identity", () => TransportChecks.ReplyId().GetAwaiter().GetResult()); Check(results, "pipe close during request", () => TransportChecks.CloseDuringRequest().GetAwaiter().GetResult()); @@ -139,6 +140,7 @@ public static int Run(string[] args) Check(results, "agent client validation", AgentChecks.ClientValidation); Check(results, "MCP lease client process", () => McpChecks.LeaseClientProcess().GetAwaiter().GetResult()); Check(results, "agent queue and disconnect", () => AgentChecks.QueueAndDisconnect().GetAwaiter().GetResult()); + Check(results, "agent queue time", () => AgentChecks.QueueTime().GetAwaiter().GetResult()); Check(results, "agent line", () => AgentChecks.Line().GetAwaiter().GetResult()); Check(results, "agent names", () => AgentChecks.Names().GetAwaiter().GetResult()); Check(results, "MCP ownership", () => AgentChecks.McpOwnership().GetAwaiter().GetResult()); diff --git a/src/Anode/Cli/TransportChecks.cs b/src/Anode/Cli/TransportChecks.cs index 84f6635..5435c9a 100644 --- a/src/Anode/Cli/TransportChecks.cs +++ b/src/Anode/Cli/TransportChecks.cs @@ -54,6 +54,47 @@ public static async Task QueueDeadline() return "queued requests time out without interrupting or replaying other commands"; } + public static async Task ForwardTimeLeft() + { + var entered = Signal(); + var release = Signal(); + var given = new System.Collections.Concurrent.ConcurrentDictionary(); + string name = PipeName(); + using var server = new JsonPipeServer(name, async request => + { + given[request.Str("op")!] = request.Int("timeoutMs"); + if (request.Str("op") == "slow") + { + entered.TrySetResult(); + await release.Task; + } + return JsonLine.Ok(); + }); + server.Start(); + using var seat = await JsonPipeClient.TryConnectAsync(name, 4000) + ?? throw new InvalidOperationException("could not connect to the test pipe"); + Require((await seat.RequestAsync("plain", timeoutMs: 5000)).Bool("ok") == true && given["plain"] is null, + "an ordinary request gained a timeoutMs it was not given"); + Require((await Daemon.AnodeDaemon.SendOnAsync(seat, "fresh", new JsonObject(), 5000)).Bool("ok") == true + && given["fresh"] is > 4000 and <= 5000, $"a forwarded request that waited for nothing was given {given["fresh"]} ms of 5000"); + + // The daemon has one connection to the seat host; a request queued behind a long one has only what is left. + Task first = seat.RequestAsync("slow", timeoutMs: 10_000); + Task queued; + try + { + await entered.Task.WaitAsync(TimeSpan.FromSeconds(3)); + queued = Daemon.AnodeDaemon.SendOnAsync(seat, "queued", new JsonObject { ["timeoutMs"] = 8000 }, 8000); + await Task.Delay(1500); + } + finally { release.TrySetResult(); } + Require((await first.WaitAsync(TimeSpan.FromSeconds(3))).Bool("ok") == true + && (await queued.WaitAsync(TimeSpan.FromSeconds(3))).Bool("ok") == true, "a request lost its reply"); + Require(given["queued"] is int left && left <= 8000 - 1400 && left > 8000 - 4500, + $"a request queued about 1.5 s behind another was forwarded with {given["queued"]} ms of 8000, not the time it had left"); + return "the daemon forwards each request with its caller's time left, so time queued behind another isn't counted twice"; + } + public static async Task LateReply() { string name = PipeName(); diff --git a/src/Anode/Core/Agents/AgentAccess.cs b/src/Anode/Core/Agents/AgentAccess.cs index cc86c9a..c561b0e 100644 --- a/src/Anode/Core/Agents/AgentAccess.cs +++ b/src/Anode/Core/Agents/AgentAccess.cs @@ -61,6 +61,15 @@ public static JsonObject Arguments(JsonObject request) return null; } + /// + /// Takes time a request spent waiting off its timeoutMs. Operations bound their work by that value, so they + /// then still answer before the deadline their caller counts from when it sent the request. + /// + public static void Spend(JsonObject request, long waitedMs) + { + if (waitedMs > 0) request["timeoutMs"] = (int)Math.Max(1, (request.Int("timeoutMs") ?? 60_000) - waitedMs); + } + public static JsonObject Attach(JsonObject? arguments, string? agentId, string? leaseToken) { var args = arguments is null ? new JsonObject() : (JsonObject)arguments.DeepClone(); diff --git a/src/Anode/Core/Agents/DesktopLease.cs b/src/Anode/Core/Agents/DesktopLease.cs index 49d19ab..1cf7fe7 100644 --- a/src/Anode/Core/Agents/DesktopLease.cs +++ b/src/Anode/Core/Agents/DesktopLease.cs @@ -201,6 +201,7 @@ private JsonObject Manage(JsonObject request) public async Task HandleAsync(JsonObject request, Func> dispatch, CancellationToken cancel = default) { + long arrived = _milliseconds(); if (AgentAccess.Validate(request) is { } invalid) return JsonLine.Fail(invalid); string op = request.Str("op") ?? ""; if (Mcp.Tools.ValidateOperation(op, AgentAccess.Arguments(request)) is { } badArgs) return JsonLine.Fail(badArgs); @@ -225,6 +226,8 @@ public async Task HandleAsync(JsonObject request, // start, so only an idle owner needs renew. It never revives an expired lease. _expires = Math.Max(_expires, _milliseconds() + _ttl); } + // Waiting for an earlier action of this agent comes off the time this one has. + AgentAccess.Spend(request, _milliseconds() - arrived); var response = await dispatch(request, deadline.Token).ConfigureAwait(false); // Tell the owner someone is waiting, so it can hand the desktop on when it finishes. lock (_state) { if (_line.Count > 0) response["waitingAgents"] = _line.Count; } diff --git a/src/Anode/Core/Bridge/Pipes.cs b/src/Anode/Core/Bridge/Pipes.cs index 6b66d22..b5a7f61 100644 --- a/src/Anode/Core/Bridge/Pipes.cs +++ b/src/Anode/Core/Bridge/Pipes.cs @@ -268,7 +268,12 @@ private JsonPipeClient(NamedPipeClientStream pipe) } } - public async Task RequestAsync(string op, JsonObject? args = null, int timeoutMs = 60_000, CancellationToken cancel = default) + /// + /// Sets the request's timeoutMs to the time left when it goes out. The receiver then bounds its work by + /// what this deadline still allows after the queue here, instead of starting a full clock of its own. + /// + public async Task RequestAsync(string op, JsonObject? args = null, int timeoutMs = 60_000, CancellationToken cancel = default, + bool sendTimeLeft = false) { var request = args is null ? new JsonObject() : (JsonObject)args.DeepClone(); request["op"] = op; @@ -277,6 +282,7 @@ public async Task RequestAsync(string op, JsonObject? args = null, i // Include queueing and writes in the deadline. In particular, a stop request // must not wait indefinitely behind a long-running operation. + long started = Environment.TickCount64; using var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancel); deadline.CancelAfter(timeoutMs); bool entered = false; @@ -286,6 +292,7 @@ public async Task RequestAsync(string op, JsonObject? args = null, i await _oneAtATime.WaitAsync(deadline.Token).ConfigureAwait(false); entered = true; if (!IsConnected) return JsonLine.Fail("connection is closed; reconnect before sending another request"); + if (sendTimeLeft) request["timeoutMs"] = (int)Math.Max(1, timeoutMs - (Environment.TickCount64 - started)); sent = true; await _writer.WriteLineAsync(JsonLine.Serialize(request).AsMemory(), deadline.Token).ConfigureAwait(false); string? line = await _reader.ReadLineAsync(deadline.Token).ConfigureAwait(false); diff --git a/src/Anode/Daemon/AnodeDaemon.cs b/src/Anode/Daemon/AnodeDaemon.cs index c900c4d..a3905c8 100644 --- a/src/Anode/Daemon/AnodeDaemon.cs +++ b/src/Anode/Daemon/AnodeDaemon.cs @@ -791,9 +791,9 @@ private async Task ForwardToSeatAsync(string op, JsonObject request) using var independent = await JsonPipeClient.TryConnectAsync(Env.SeatPipe, Math.Min(1000, timeout)).ConfigureAwait(false); int remaining = timeout - (int)clock.ElapsedMilliseconds; if (independent is null || remaining <= 0) return JsonLine.Fail("Could not reach the seat host within the request deadline."); - return await independent.RequestAsync(op, forwarded, remaining).ConfigureAwait(false); + return await SendOnAsync(independent, op, forwarded, remaining).ConfigureAwait(false); } - var response = await seat.RequestAsync(op, forwarded, timeout).ConfigureAwait(false); + var response = await SendOnAsync(seat, op, forwarded, timeout).ConfigureAwait(false); if (response.Bool("ok") != true && !seat.IsConnected && ReferenceEquals(_seat, seat) && !_stopRequested) { @@ -804,6 +804,14 @@ private async Task ForwardToSeatAsync(string op, JsonObject request) return response; } + /// + /// Sends a request on to the seat host with the time its caller has left. Time spent queued here, behind another + /// agent's long operation such as typing, comes off what the seat host gives it, so it still answers before this + /// deadline instead of losing its reply and the connection. + /// + internal static Task SendOnAsync(JsonPipeClient seat, string op, JsonObject forwarded, int timeoutMs) => + seat.RequestAsync(op, forwarded, timeoutMs, sendTimeLeft: true); + // ----------------------------------------------------------------- display /// The display a seat starts with: as measured once it is ready, or as Anode was asked for. diff --git a/src/Anode/Seat/SeatHost.cs b/src/Anode/Seat/SeatHost.cs index d751d01..03dff55 100644 --- a/src/Anode/Seat/SeatHost.cs +++ b/src/Anode/Seat/SeatHost.cs @@ -158,8 +158,13 @@ private static Task HandleAsync(JsonObject request) private static async Task HandleOwnedAsync(JsonObject request, CancellationToken cancel) { string op = request.Str("op") ?? string.Empty; - // The display a previous lease changed is put back before the next owner looks or acts. - if (AgentAccess.RequiresLease(op) && await Display.RestoredAsync(cancel).ConfigureAwait(false) is { } restoring) return restoring; + // The display a previous lease changed is put back before the next owner looks or acts, in the action's time. + if (AgentAccess.RequiresLease(op)) + { + long waiting = Environment.TickCount64; + if (await Display.RestoredAsync(cancel).ConfigureAwait(false) is { } restoring) return restoring; + AgentAccess.Spend(request, Environment.TickCount64 - waiting); + } if (op == "display.set") return await ChangeDisplayAsync(request, cancel).ConfigureAwait(false); if (op == "android.status") return JsonLine.Ok(await Emulators.Value.StatusAsync(cancel).ConfigureAwait(false)); if (op == "android.emulator") return await EmulatorAsync(request, cancel).ConfigureAwait(false); From 88df5aa41a329376dd80f0e2c69965d84e5a63fb Mon Sep 17 00:00:00 2001 From: skulitom Date: Tue, 6 Oct 2026 23:36:40 +0100 Subject: [PATCH 2/4] Take the display-restore wait off inside SeatDisplay, where a check covers it SeatDisplay.RestoredAsync takes the request and spends the time it waited for the previous lease's restore, instead of SeatHost measuring around the call. "display restored after a lease" now checks that an action which waited for the restore keeps less than its 10,000 ms. Co-Authored-By: Claude Opus 5.5 --- src/Anode/Cli/DisplayChecks.cs | 5 ++++- src/Anode/Core/Display/SeatDisplay.cs | 8 ++++++-- src/Anode/Seat/SeatHost.cs | 7 +------ 3 files changed, 11 insertions(+), 9 deletions(-) diff --git a/src/Anode/Cli/DisplayChecks.cs b/src/Anode/Cli/DisplayChecks.cs index d83d392..32268f4 100644 --- a/src/Anode/Cli/DisplayChecks.cs +++ b/src/Anode/Cli/DisplayChecks.cs @@ -272,9 +272,12 @@ public static async Task Restore() var busy = await display.RestoredAsync(impatient.Token); Require(busy?.Str("errorCode") == "display_restoring" && busy.Str("error")!.Contains("Nothing was done"), "a pending restore did not hold the next action"); } - Require(await display.RestoredAsync(CancellationToken.None) is null && seat.Screen == Hd + // The wait comes off the action's own time, so it still answers before its caller's deadline. + var action = new JsonObject { ["op"] = "input.click", ["timeoutMs"] = 10_000 }; + Require(await display.RestoredAsync(CancellationToken.None, action) is null && seat.Screen == Hd && seat.Changes().SequenceEqual(new[] { "live 1920x1080 at 150%", "live 1280x720 at 100%" }) && record.Any(line => line.Contains("restored the startup display 1280x720 at 100%")), "the lease's end did not restore the startup display"); + Require(action.Int("timeoutMs") is < 10_000 and > 5_000, $"an action that waited for the restore kept {action.Int("timeoutMs")} ms of 10000"); display.EndLease(); await display.RestoredAsync(CancellationToken.None); Require(seat.Changes().Length == 2, "a restored display was restored again"); diff --git a/src/Anode/Core/Display/SeatDisplay.cs b/src/Anode/Core/Display/SeatDisplay.cs index 28243f9..af3f37b 100644 --- a/src/Anode/Core/Display/SeatDisplay.cs +++ b/src/Anode/Core/Display/SeatDisplay.cs @@ -1,6 +1,7 @@ using System.Diagnostics; using System.Runtime.InteropServices; using System.Text.Json.Nodes; +using Anode.Core.Agents; using Anode.Core.Bridge; namespace Anode.Core.Display; @@ -117,13 +118,15 @@ public void EndLease() /// /// Null once no restore is pending; otherwise waits for it, so the next owner acts on the startup display. - /// A failure explains why the action was not attempted. + /// A failure explains why the action was not attempted. The wait comes off 's + /// timeoutMs, so the action that follows still answers in its caller's time. /// - public async Task RestoredAsync(CancellationToken cancel) + public async Task RestoredAsync(CancellationToken cancel, JsonObject? request = null) { Task pending; lock (_gate) pending = _restoring; if (pending.IsCompleted) return null; + long waiting = Environment.TickCount64; try { await pending.WaitAsync(cancel).ConfigureAwait(false); } catch (OperationCanceledException) when (cancel.IsCancellationRequested) { @@ -131,6 +134,7 @@ public void EndLease() busy["errorCode"] = "display_restoring"; return busy; } + if (request is not null) AgentAccess.Spend(request, Environment.TickCount64 - waiting); return null; } diff --git a/src/Anode/Seat/SeatHost.cs b/src/Anode/Seat/SeatHost.cs index 03dff55..fab172e 100644 --- a/src/Anode/Seat/SeatHost.cs +++ b/src/Anode/Seat/SeatHost.cs @@ -159,12 +159,7 @@ private static async Task HandleOwnedAsync(JsonObject request, Cance { string op = request.Str("op") ?? string.Empty; // The display a previous lease changed is put back before the next owner looks or acts, in the action's time. - if (AgentAccess.RequiresLease(op)) - { - long waiting = Environment.TickCount64; - if (await Display.RestoredAsync(cancel).ConfigureAwait(false) is { } restoring) return restoring; - AgentAccess.Spend(request, Environment.TickCount64 - waiting); - } + if (AgentAccess.RequiresLease(op) && await Display.RestoredAsync(cancel, request).ConfigureAwait(false) is { } restoring) return restoring; if (op == "display.set") return await ChangeDisplayAsync(request, cancel).ConfigureAwait(false); if (op == "android.status") return JsonLine.Ok(await Emulators.Value.StatusAsync(cancel).ConfigureAwait(false)); if (op == "android.emulator") return await EmulatorAsync(request, cancel).ConfigureAwait(false); From 9d1ccdedbe1ad8f9749eab1fe0e29bd9e0af6cb9 Mon Sep 17 00:00:00 2001 From: skulitom Date: Tue, 6 Oct 2026 23:56:35 +0100 Subject: [PATCH 3/4] Stop a desktop action before its caller gives up, and ignore tick-sized waits Review follow-up. - The lease's deadline now falls a second (or an eighth of a shorter time) before the request's timeoutMs, so an action cut short by it, such as a queued seat_wait, still answers before the daemon gives up, and the seat connection survives. Typing's own stop (7/8) comes before it, unchanged. - Waits under 100 ms aren't counted (JsonPipeClient.TimeLeft), so the time forwarded to the seat, and limits derived from it such as the 3,001 characters one call may type, no longer depend on clock ticks. - A request whose time ran out while it queued isn't sent. - A refusal of text too long for a call that waited says the call had less than the usual 60 s. - PROTOCOL.md, the CHANGELOG and comments say exactly which operations size their work by timeoutMs and that the rest are stopped before it. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 7 +++--- docs/PROTOCOL.md | 6 +++-- src/Anode/Cli/AgentChecks.cs | 30 ++++++++++++++++++++++++- src/Anode/Cli/TransportChecks.cs | 2 ++ src/Anode/Core/Agents/AgentAccess.cs | 8 ++++--- src/Anode/Core/Agents/DesktopLease.cs | 8 ++++++- src/Anode/Core/Android/SeatEmulators.cs | 3 ++- src/Anode/Core/Bridge/Pipes.cs | 16 ++++++++++++- src/Anode/Core/Input/TextTyping.cs | 9 +++++--- 9 files changed, 74 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 88643bc..0b82ae2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,9 +9,10 @@ MCP connects as soon as it launches the daemon, and failed host connections retry sooner. - **Queued requests keep their deadline.** A request that waited behind another long operation, such as typing, got a full 60 seconds of its own once it reached the seat. So it could still be running when - the daemon gave up on it: the reply was lost and the seat showed as `detached` until it reconnected. - The daemon now forwards the time a request has left, and the seat takes its own waits off as well, so - an operation queued behind another ends, or is refused, in time. + the daemon gave up on it: the reply was lost and the seat showed as `detached` until it was + reconnected. The daemon now forwards the time a request has left, the seat takes its own waits off as + well, and a desktop operation is stopped a second before its time runs out, so one queued behind + another answers, or fails, in time. - **Typing keeps pace with the program.** `seat_type` and `anode type` sent characters as fast as Windows took them, so a busy program, such as Chrome editing a long field, could drop and reorder them while the call still reported success. Characters now go out 15 ms apart by default, timed diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 752ed85..f5b883f 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -23,8 +23,10 @@ flat on the same object. `timeoutMs` on a request bound for the seat overrides t after sending closes that connection, because a late reply must not become the next command's result. Reconnect for subsequent requests; a timed-out command may already have executed. The daemon forwards a request with `timeoutMs` set to the time it has left after waiting behind other -operations on its one connection to the seat host, and the seat host takes its own waits off as well, -so an operation queued behind another, such as a long `input.text`, still answers in time. +operations on its one connection to the seat host (waits under 100 ms aren't counted), and the seat +host takes its own waits off as well. A desktop operation is stopped a second (or an eighth of a +shorter time) before that time runs out, so even one cut short answers before its caller gives up. +Limits derived from the time, such as how much one `input.text` may type, shrink with it. Desktop operations additionally require an `agentId` and live `leaseToken`, acquired through `lease` with `action: "acquire"`, and may carry an `agentName` for status and the viewer. `lease` diff --git a/src/Anode/Cli/AgentChecks.cs b/src/Anode/Cli/AgentChecks.cs index 1c24db8..6856561 100644 --- a/src/Anode/Cli/AgentChecks.cs +++ b/src/Anode/Cli/AgentChecks.cs @@ -306,7 +306,35 @@ async Task Dispatch(JsonObject request, CancellationToken cancel) Require(given["first"] is null && given["plain"] == 53_000 && given["own deadline"] == 13_000, $"queued actions were not given the time left: {given["first"]}, {given["plain"]} and {given["own deadline"]} ms, " + "expected none, 53000 and 13000"); - return "an action queued behind another of its agent's runs in the time its caller has left"; + + // Waits under 100 ms are the cost of passing a request on, and don't shrink limits such as typing's. + var tick = new JsonObject { ["timeoutMs"] = 5000 }; + AgentAccess.Spend(tick, 99); + var counted = new JsonObject(); + AgentAccess.Spend(counted, 100); + var over = new JsonObject { ["timeoutMs"] = 5000 }; + AgentAccess.Spend(over, 9000); + Require(tick.Int("timeoutMs") == 5000 && counted.Int("timeoutMs") == 59_900 && over.Int("timeoutMs") == 1, + $"waits were counted wrongly: 99 ms left {tick.Int("timeoutMs")}, 100 ms left {counted.Int("timeoutMs")}, too long left {over.Int("timeoutMs")}"); + + // An action is stopped before its caller gives up, so its reply, even a cancellation, arrives in time. + Require(DesktopLease.StopAfterMs(60_000) == 59_000 && DesktopLease.StopAfterMs(8000) == 7000 + && DesktopLease.StopAfterMs(1000) == 875 && DesktopLease.StopAfterMs(1) == 1, "the stop before the caller's deadline is miscounted"); + var stopped = Signal(); + var clock = Stopwatch.StartNew(); + async Task Hold(JsonObject request, CancellationToken cancel) + { + try { await Task.Delay(Timeout.Infinite, cancel); } + catch (OperationCanceledException) { stopped.TrySetResult(); } + return JsonLine.Ok(); + } + var held = Input("A", token, "held"); + held["timeoutMs"] = 1000; + clock.Restart(); + await lease.HandleAsync(held, Hold).WaitAsync(TimeSpan.FromSeconds(3)); + long ms = clock.ElapsedMilliseconds; + Require(stopped.Task.IsCompleted && ms >= 750 && ms < 990, $"an action with 1000 ms was stopped after {ms} ms, not before its caller gave up"); + return "an action queued behind another of its agent's runs in the time its caller has left, and stops before it"; } public static async Task Line() diff --git a/src/Anode/Cli/TransportChecks.cs b/src/Anode/Cli/TransportChecks.cs index 5435c9a..eb511a6 100644 --- a/src/Anode/Cli/TransportChecks.cs +++ b/src/Anode/Cli/TransportChecks.cs @@ -77,6 +77,8 @@ public static async Task ForwardTimeLeft() "an ordinary request gained a timeoutMs it was not given"); Require((await Daemon.AnodeDaemon.SendOnAsync(seat, "fresh", new JsonObject(), 5000)).Bool("ok") == true && given["fresh"] is > 4000 and <= 5000, $"a forwarded request that waited for nothing was given {given["fresh"]} ms of 5000"); + Require(JsonPipeClient.TimeLeft(5000, 99) == 5000 && JsonPipeClient.TimeLeft(5000, 100) == 4900 && JsonPipeClient.TimeLeft(5000, 6000) == 1, + "time left after a wait is miscounted"); // The daemon has one connection to the seat host; a request queued behind a long one has only what is left. Task first = seat.RequestAsync("slow", timeoutMs: 10_000); diff --git a/src/Anode/Core/Agents/AgentAccess.cs b/src/Anode/Core/Agents/AgentAccess.cs index c561b0e..e65a594 100644 --- a/src/Anode/Core/Agents/AgentAccess.cs +++ b/src/Anode/Core/Agents/AgentAccess.cs @@ -62,12 +62,14 @@ public static JsonObject Arguments(JsonObject request) } /// - /// Takes time a request spent waiting off its timeoutMs. Operations bound their work by that value, so they - /// then still answer before the deadline their caller counts from when it sent the request. + /// Takes time a request spent waiting off its timeoutMs. Typing, emulator start and Steam launch size their + /// work by it, and the lease's deadline, which falls a little before it, stops the rest, so the reply still comes + /// before the deadline its caller counts from when it sent the request. /// public static void Spend(JsonObject request, long waitedMs) { - if (waitedMs > 0) request["timeoutMs"] = (int)Math.Max(1, (request.Int("timeoutMs") ?? 60_000) - waitedMs); + int timeout = request.Int("timeoutMs") ?? 60_000, left = JsonPipeClient.TimeLeft(timeout, waitedMs); + if (left != timeout) request["timeoutMs"] = left; } public static JsonObject Attach(JsonObject? arguments, string? agentId, string? leaseToken) diff --git a/src/Anode/Core/Agents/DesktopLease.cs b/src/Anode/Core/Agents/DesktopLease.cs index 1cf7fe7..963764f 100644 --- a/src/Anode/Core/Agents/DesktopLease.cs +++ b/src/Anode/Core/Agents/DesktopLease.cs @@ -198,6 +198,12 @@ private JsonObject Manage(JsonObject request) } } + /// + /// When an action is stopped: a second, or an eighth of a shorter time, before its caller gives up, so even an + /// action cut short answers in time and the caller's connection survives. + /// + internal static int StopAfterMs(int timeoutMs) => Math.Max(1, timeoutMs - Math.Min(1000, timeoutMs / 8)); + public async Task HandleAsync(JsonObject request, Func> dispatch, CancellationToken cancel = default) { @@ -211,7 +217,7 @@ public async Task HandleAsync(JsonObject request, lock (_state) { if (Check(request) is { } error) return error; } using var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancel); - deadline.CancelAfter(request.Int("timeoutMs") ?? 60000); + deadline.CancelAfter(StopAfterMs(request.Int("timeoutMs") ?? 60000)); await _interaction.WaitAsync(deadline.Token).ConfigureAwait(false); bool admitted = false; try diff --git a/src/Anode/Core/Android/SeatEmulators.cs b/src/Anode/Core/Android/SeatEmulators.cs index bb022e0..309cf1b 100644 --- a/src/Anode/Core/Android/SeatEmulators.cs +++ b/src/Anode/Core/Android/SeatEmulators.cs @@ -127,7 +127,8 @@ public async Task StatusAsync(CancellationToken cancel) public async Task StartAsync(JsonObject request, CancellationToken cancel) { long entry = _machine.Milliseconds(); - // Leave time to build and forward the reply; the pipe deadline includes queue time and adb startup. + // Leave time to build and forward the reply. timeoutMs is what the caller has left after any queue; adb may + // still have to start. const int ReplyReserveMs = 10_000; long replyBy = entry + Math.Max(0, (request.Int("timeoutMs") ?? 60_000) - ReplyReserveMs); var sdk = AndroidSdk.Find(_machine.Environment); diff --git a/src/Anode/Core/Bridge/Pipes.cs b/src/Anode/Core/Bridge/Pipes.cs index b5a7f61..4746c50 100644 --- a/src/Anode/Core/Bridge/Pipes.cs +++ b/src/Anode/Core/Bridge/Pipes.cs @@ -292,7 +292,13 @@ public async Task RequestAsync(string op, JsonObject? args = null, i await _oneAtATime.WaitAsync(deadline.Token).ConfigureAwait(false); entered = true; if (!IsConnected) return JsonLine.Fail("connection is closed; reconnect before sending another request"); - if (sendTimeLeft) request["timeoutMs"] = (int)Math.Max(1, timeoutMs - (Environment.TickCount64 - started)); + if (sendTimeLeft) + { + long waited = Environment.TickCount64 - started; + // With no time left, any reply would come too late, and a late reply costs the connection. + if (waited >= timeoutMs) return JsonLine.Fail($"'{op}' timed out after {timeoutMs} ms"); + request["timeoutMs"] = TimeLeft(timeoutMs, waited); + } sent = true; await _writer.WriteLineAsync(JsonLine.Serialize(request).AsMemory(), deadline.Token).ConfigureAwait(false); string? line = await _reader.ReadLineAsync(deadline.Token).ConfigureAwait(false); @@ -326,6 +332,14 @@ public async Task RequestAsync(string op, JsonObject? args = null, i } } + /// + /// What is left of a request's time after a wait. Waits under 100 ms are the ordinary cost of passing a request + /// on and aren't counted, so limits derived from the time, such as how much one call may type, don't depend on + /// clock ticks. + /// + internal static int TimeLeft(int timeoutMs, long waitedMs) => + waitedMs < 100 ? timeoutMs : (int)Math.Max(1, timeoutMs - waitedMs); + private void Close() { if (Interlocked.Exchange(ref _closed, 1) == 0) _pipe.Dispose(); diff --git a/src/Anode/Core/Input/TextTyping.cs b/src/Anode/Core/Input/TextTyping.cs index 2301561..38a7a0a 100644 --- a/src/Anode/Core/Input/TextTyping.cs +++ b/src/Anode/Core/Input/TextTyping.cs @@ -26,13 +26,14 @@ internal static class TextTyping /// /// Typing time a call may plan for; longer text is refused before anything is typed. MCP, the CLI and the daemon - /// wait 60 seconds for the reply. A request that sets a shorter timeoutMs plans for three quarters of it. + /// wait 60 seconds for the reply. A request with a shorter timeoutMs, set by its caller or shortened by + /// waiting behind another operation, plans for three quarters of it. /// public const int BudgetMs = 45_000; /// /// When a call stops, however far it got: timer lateness and a busy program stretch the plan, and the reply must - /// still arrive in time. A request that sets a shorter timeoutMs stops at seven eighths of it. + /// still arrive in time. A request with a shorter timeoutMs stops at seven eighths of it. /// public const int StopMs = 52_000; @@ -114,7 +115,9 @@ internal static long MinimumMs(IReadOnlyList units, int perCharMs) public static JsonObject Type(string text, int? perCharMs, Typist typist, int timeoutMs = 60_000) { int plan = (int)Math.Min(BudgetMs, timeoutMs * 3L / 4), stop = (int)Math.Min(StopMs, timeoutMs * 7L / 8); - if (Problem(text, perCharMs, plan) is { } problem) return JsonLine.Fail(problem); + if (Problem(text, perCharMs, plan) is { } problem) + return JsonLine.Fail(timeoutMs >= 60_000 ? problem : problem + string.Create(CultureInfo.InvariantCulture, + $" This call had {timeoutMs / 1000.0:0.#} s, less than the usual 60, because it waited behind another operation or asked for less.")); int pace = perCharMs ?? DefaultPerCharMs; var units = Units(text); double start = typist.Now(), previous = start; From 7158c1e5faaba6e6794acea48b50bf4bc1cc1784 Mon Sep 17 00:00:00 2001 From: skulitom Date: Wed, 7 Oct 2026 00:12:06 +0100 Subject: [PATCH 4/4] Don't send a request with almost no time left, and widen a timing check Second review follow-up (P3s): a forwarded request left with less than 100 ms (an eighth of a shorter time), or whose deadline already fired, isn't sent, so its late reply can't cost the connection; waits under a sixteenth of a short time aren't counted either. "pipe forwarded time left" now checks the unsent path, and "agent queue time" gives its live stop 240 ms of slack. Co-Authored-By: Claude Opus 5.5 --- src/Anode/Cli/AgentChecks.cs | 6 +++--- src/Anode/Cli/TransportChecks.cs | 16 ++++++++++++++++ src/Anode/Core/Bridge/Pipes.cs | 17 +++++++++-------- 3 files changed, 28 insertions(+), 11 deletions(-) diff --git a/src/Anode/Cli/AgentChecks.cs b/src/Anode/Cli/AgentChecks.cs index 6856561..0850607 100644 --- a/src/Anode/Cli/AgentChecks.cs +++ b/src/Anode/Cli/AgentChecks.cs @@ -329,11 +329,11 @@ async Task Hold(JsonObject request, CancellationToken cancel) return JsonLine.Ok(); } var held = Input("A", token, "held"); - held["timeoutMs"] = 1000; + held["timeoutMs"] = 2000; clock.Restart(); - await lease.HandleAsync(held, Hold).WaitAsync(TimeSpan.FromSeconds(3)); + await lease.HandleAsync(held, Hold).WaitAsync(TimeSpan.FromSeconds(5)); long ms = clock.ElapsedMilliseconds; - Require(stopped.Task.IsCompleted && ms >= 750 && ms < 990, $"an action with 1000 ms was stopped after {ms} ms, not before its caller gave up"); + Require(stopped.Task.IsCompleted && ms >= 1500 && ms < 1990, $"an action with 2000 ms was stopped after {ms} ms, not before its caller gave up"); return "an action queued behind another of its agent's runs in the time its caller has left, and stops before it"; } diff --git a/src/Anode/Cli/TransportChecks.cs b/src/Anode/Cli/TransportChecks.cs index eb511a6..d138e22 100644 --- a/src/Anode/Cli/TransportChecks.cs +++ b/src/Anode/Cli/TransportChecks.cs @@ -94,6 +94,22 @@ public static async Task ForwardTimeLeft() && (await queued.WaitAsync(TimeSpan.FromSeconds(3))).Bool("ok") == true, "a request lost its reply"); Require(given["queued"] is int left && left <= 8000 - 1400 && left > 8000 - 4500, $"a request queued about 1.5 s behind another was forwarded with {given["queued"]} ms of 8000, not the time it had left"); + + // A request left with almost no time isn't sent: its reply would come too late and cost the connection. + entered = Signal(); + release = Signal(); + first = seat.RequestAsync("slow", timeoutMs: 10_000); + Task starved; + try + { + await entered.Task.WaitAsync(TimeSpan.FromSeconds(3)); + starved = Daemon.AnodeDaemon.SendOnAsync(seat, "starved", new JsonObject(), 1000); + await Task.Delay(960); + } + finally { release.TrySetResult(); } + var unsent = await starved.WaitAsync(TimeSpan.FromSeconds(3)); + Require((await first.WaitAsync(TimeSpan.FromSeconds(3))).Bool("ok") == true && unsent.Bool("ok") == false + && !given.ContainsKey("starved") && seat.IsConnected, "a request with almost no time left was sent, or cost the connection: " + unsent.ToJsonString()); return "the daemon forwards each request with its caller's time left, so time queued behind another isn't counted twice"; } diff --git a/src/Anode/Core/Bridge/Pipes.cs b/src/Anode/Core/Bridge/Pipes.cs index 4746c50..f40e7e7 100644 --- a/src/Anode/Core/Bridge/Pipes.cs +++ b/src/Anode/Core/Bridge/Pipes.cs @@ -294,10 +294,11 @@ public async Task RequestAsync(string op, JsonObject? args = null, i if (!IsConnected) return JsonLine.Fail("connection is closed; reconnect before sending another request"); if (sendTimeLeft) { - long waited = Environment.TickCount64 - started; - // With no time left, any reply would come too late, and a late reply costs the connection. - if (waited >= timeoutMs) return JsonLine.Fail($"'{op}' timed out after {timeoutMs} ms"); - request["timeoutMs"] = TimeLeft(timeoutMs, waited); + long left = timeoutMs - (Environment.TickCount64 - started); + // With almost no time left, any reply would come too late, and a late reply costs the connection. + if (deadline.IsCancellationRequested || left < Math.Min(100, timeoutMs / 8)) + return JsonLine.Fail($"'{op}' timed out after {timeoutMs} ms while it waited; nothing was sent"); + request["timeoutMs"] = TimeLeft(timeoutMs, timeoutMs - left); } sent = true; await _writer.WriteLineAsync(JsonLine.Serialize(request).AsMemory(), deadline.Token).ConfigureAwait(false); @@ -333,12 +334,12 @@ public async Task RequestAsync(string op, JsonObject? args = null, i } /// - /// What is left of a request's time after a wait. Waits under 100 ms are the ordinary cost of passing a request - /// on and aren't counted, so limits derived from the time, such as how much one call may type, don't depend on - /// clock ticks. + /// What is left of a request's time after a wait. Waits under 100 ms (a sixteenth of a shorter time) are the + /// ordinary cost of passing a request on and aren't counted, so limits derived from the time, such as how much + /// one call may type, don't depend on clock ticks. /// internal static int TimeLeft(int timeoutMs, long waitedMs) => - waitedMs < 100 ? timeoutMs : (int)Math.Max(1, timeoutMs - waitedMs); + waitedMs < Math.Min(100, timeoutMs / 16) ? timeoutMs : (int)Math.Max(1, timeoutMs - waitedMs); private void Close() {