[compass] Fix: a request queued behind a long seat operation loses its reply and detaches the seat

CHANGELOG.md

@@ -7,6 +7,12 @@
   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 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

docs/PROTOCOL.md

@@ -21,7 +21,12 @@ 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 (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`

src/Anode/Cli/AgentChecks.cs

@@ -271,6 +271,72 @@ async Task<JsonObject> 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<string> 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<string, int?>();
+        async Task<JsonObject> 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<JsonObject> active = lease.HandleAsync(Input("A", token, "first"), Dispatch);
+        Task<JsonObject> 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");
+
+        // 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<JsonObject> 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"] = 2000;
+        clock.Restart();
+        await lease.HandleAsync(held, Hold).WaitAsync(TimeSpan.FromSeconds(5));
+        long ms = clock.ElapsedMilliseconds;
+        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";
+    }
+
     public static async Task<string> Line()
     {
         long now = 0;

src/Anode/Cli/DisplayChecks.cs

@@ -272,9 +272,12 @@ public static async Task<string> 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");

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());

src/Anode/Cli/TransportChecks.cs

@@ -54,6 +54,65 @@ public static async Task<string> QueueDeadline()
         return "queued requests time out without interrupting or replaying other commands";
     }
 
+    public static async Task<string> ForwardTimeLeft()
+    {
+        var entered = Signal();
+        var release = Signal();
+        var given = new System.Collections.Concurrent.ConcurrentDictionary<string, int?>();
+        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");
+        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<JsonObject> first = seat.RequestAsync("slow", timeoutMs: 10_000);
+        Task<JsonObject> 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");
+
+        // 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<JsonObject> 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";
+    }
+
     public static async Task<string> LateReply()
     {
         string name = PipeName();

src/Anode/Core/Agents/AgentAccess.cs

@@ -61,6 +61,17 @@ public static JsonObject Arguments(JsonObject request)
         return null;
     }
 
+    /// <summary>
+    /// Takes time a request spent waiting off its <c>timeoutMs</c>. 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.
+    /// </summary>
+    public static void Spend(JsonObject request, long 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)
     {
         var args = arguments is null ? new JsonObject() : (JsonObject)arguments.DeepClone();

src/Anode/Core/Agents/DesktopLease.cs

@@ -198,9 +198,16 @@ private JsonObject Manage(JsonObject request)
         }
     }
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    internal static int StopAfterMs(int timeoutMs) => Math.Max(1, timeoutMs - Math.Min(1000, timeoutMs / 8));
+
     public async Task<JsonObject> HandleAsync(JsonObject request,
         Func<JsonObject, CancellationToken, Task<JsonObject>> 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);
@@ -210,7 +217,7 @@ public async Task<JsonObject> 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
@@ -225,6 +232,8 @@ public async Task<JsonObject> 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; }

src/Anode/Core/Android/SeatEmulators.cs

@@ -127,7 +127,8 @@ public async Task<JsonObject> StatusAsync(CancellationToken cancel)
     public async Task<JsonObject> 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);

src/Anode/Core/Bridge/Pipes.cs

@@ -268,7 +268,12 @@ private JsonPipeClient(NamedPipeClientStream pipe)
         }
     }
 
-    public async Task<JsonObject> RequestAsync(string op, JsonObject? args = null, int timeoutMs = 60_000, CancellationToken cancel = default)
+    /// <param name="sendTimeLeft">
+    /// Sets the request's <c>timeoutMs</c> 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.
+    /// </param>
+    public async Task<JsonObject> 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<JsonObject> 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,14 @@ public async Task<JsonObject> 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)
+            {
+                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);
             string? line = await _reader.ReadLineAsync(deadline.Token).ConfigureAwait(false);
@@ -319,6 +333,14 @@ public async Task<JsonObject> RequestAsync(string op, JsonObject? args = null, i
         }
     }
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    internal static int TimeLeft(int timeoutMs, long waitedMs) =>
+        waitedMs < Math.Min(100, timeoutMs / 16) ? timeoutMs : (int)Math.Max(1, timeoutMs - waitedMs);
+
     private void Close()
     {
         if (Interlocked.Exchange(ref _closed, 1) == 0) _pipe.Dispose();