fix(waits): monotonic judge deadlines, early messages, hidden-hold count, pin with the first write, settings follow-ups

apps/server/src/scheduling/DueHint.test.ts

@@ -0,0 +1,26 @@
+import { assert, it } from "@effect/vitest";
+import * as Effect from "effect/Effect";
+
+import * as DueHint from "./DueHint.ts";
+
+const NOW = Date.parse("2026-11-14T12:00:00.000Z");
+const HOUR = 3_600_000;
+
+it.effect("a recorded hint keeps the poller idle until its work is due", () =>
+  Effect.gen(function* () {
+    const due = DueHint.make();
+    yield* due.record(yield* due.version, NOW + 30_000, NOW);
+    assert.isTrue(yield* due.isIdle(NOW + 5_000));
+    assert.isFalse(yield* due.isIdle(NOW + 30_000));
+  }),
+);
+
+it.effect("a hint recorded before the wall clock was set back does not keep the poller idle", () =>
+  Effect.gen(function* () {
+    const due = DueHint.make();
+    // The next poll has work in 30 s.
+    yield* due.record(yield* due.version, NOW + 30_000, NOW);
+    // The wall clock is set back an hour; a scheduler tick 5 s of real time later.
+    assert.isFalse(yield* due.isIdle(NOW - HOUR + 5_000));
+  }),
+);

apps/server/src/scheduling/DueHint.ts

@@ -17,10 +17,14 @@ export const make = () => {
       hint = null;
     }),
     version: Effect.sync(() => version),
+    // A wall clock set back (now before the check) is not idle: the hint's times are stale.
     isIdle: (nowMs: number) =>
       Effect.sync(
         () =>
-          hint !== null && hint.dueAtMs > nowMs && nowMs - hint.checkedAtMs < DUE_HINT_MAX_AGE_MS,
+          hint !== null &&
+          hint.dueAtMs > nowMs &&
+          nowMs >= hint.checkedAtMs &&
+          nowMs - hint.checkedAtMs < DUE_HINT_MAX_AGE_MS,
       ),
     record: (readVersion: number, dueAtMs: number, nowMs: number) =>
       Effect.sync(() => {

apps/server/src/slots/SlotGate.test.ts

@@ -999,6 +999,88 @@ describe("SlotGate", () => {
         }),
     );
 
+    // While a re-ask on another account (or none) is not kept, the old hold still counts
+    // against its own account, so a dropped re-ask never leaves that account booked twice.
+    it.effect(
+      "a re-ask not kept yet still counts the old hold against its account, once on the machine",
+      () =>
+        Effect.gen(function* () {
+          const world = newWorld();
+          const third = ThreadId.make("thread:slot-third");
+          yield* run(
+            world,
+            waitsOn({ maxPerMachine: 2, maxPerAccount: { [SECOND]: 1 }, reserveSeconds: 60 }),
+            Effect.gen(function* () {
+              const gate = yield* SlotGate.SlotGate;
+              assert.deepStrictEqual(
+                types(yield* grant({ key: "h0", threadId: holder, account: SECOND })),
+                ["granted"],
+              );
+              yield* gate.keep(["h0"]);
+              // One sweep: the holder re-asks with no account, then `other` asks for SECOND.
+              // The holder's SECOND hold is hidden but still counts there: `other` waits.
+              assert.deepStrictEqual(
+                types(
+                  yield* grant(
+                    { key: "w1", threadId: holder },
+                    { key: "w2", threadId: other, account: SECOND },
+                  ),
+                ),
+                ["granted", "full"],
+              );
+              // The settings screen shows the hidden hold on SECOND too (1 of 1, not 0 of 1).
+              assert.deepStrictEqual(
+                (yield* gate.usage)?.accounts.find((account) => account.id === SECOND),
+                { id: SECOND, used: 1, max: 1 },
+              );
+              assert.strictEqual((yield* acquireAs(other, SECOND)).type, "refuse");
+              // Not twice against the machine: the holder counts once (its new hold), so a
+              // third thread still gets the second machine slot.
+              assert.deepStrictEqual(types(yield* grant({ key: "w3", threadId: third })), [
+                "granted",
+              ]);
+              // The re-ask is cancelled before it is written: the old hold is back, and SECOND
+              // holds it once (not over its limit), for the holder only.
+              yield* gate.drop(["w1"]);
+              assert.deepStrictEqual(
+                types(yield* grant({ key: "w2", threadId: other, account: SECOND })),
+                ["full"],
+              );
+              const usage = yield* gate.usage;
+              assert.deepStrictEqual(
+                usage?.accounts.find((account) => account.id === SECOND),
+                { id: SECOND, used: 1, max: 1 },
+              );
+              assert.strictEqual((yield* acquireAs(holder, SECOND)).type, "allow");
+            }),
+          );
+          // Kept this time: the old SECOND hold is used up, and `other` gets SECOND next sweep.
+          yield* run(
+            newWorld(),
+            waitsOn({ maxPerMachine: 3, maxPerAccount: { [SECOND]: 1 }, reserveSeconds: 60 }),
+            Effect.gen(function* () {
+              const gate = yield* SlotGate.SlotGate;
+              yield* grant({ key: "h0", threadId: holder, account: SECOND });
+              yield* gate.keep(["h0"]);
+              assert.deepStrictEqual(
+                types(
+                  yield* grant(
+                    { key: "w1", threadId: holder },
+                    { key: "w2", threadId: other, account: SECOND },
+                  ),
+                ),
+                ["granted", "full"],
+              );
+              yield* gate.keep(["w1"]);
+              assert.deepStrictEqual(
+                types(yield* grant({ key: "w2", threadId: other, account: SECOND })),
+                ["granted"],
+              );
+            }),
+          );
+        }),
+    );
+
     it.effect("a thread already holding a slot that is refused keeps the slot it holds", () =>
       Effect.gen(function* () {
         const world = newWorld();

apps/server/src/slots/SlotGate.ts

@@ -6,6 +6,7 @@
  */
 import type { ProviderInstanceId, ThreadId } from "@t3tools/contracts";
 import {
+  accountOnlyThreads,
   effectiveSlotLimits,
   limitsCount,
   liveReservations,
@@ -140,8 +141,9 @@ const make = Effect.gen(function* () {
   /** Keys used up, until their end: the store's copy stops counting even if clearing it failed. */
   const usedKeys = new Map<string, number>();
   /**
-   * A re-ask's grant not kept yet, by its key: the thread's earlier holds, hidden (they do not
-   * count) but not used up, so a grant that is dropped gives them back.
+   * A re-ask's grant not kept yet, by its key: the thread's earlier holds, hidden (they count only
+   * against their own account, see `hiddenHeld`) but not used up, so a grant that is dropped gives
+   * them back.
    */
   const replacing = new Map<string, ReadonlyArray<SlotReservation>>();
   const hiddenKeys = () =>
@@ -209,6 +211,18 @@ const make = Effect.gen(function* () {
     );
   });
 
+  /**
+   * The live holds that re-asks not kept yet hide, of threads other than `threadId`. They still
+   * count against their own account (never the machine, where the new hold counts), so a slot
+   * that a dropped re-ask gives back is never booked twice. None while waits are off.
+   */
+  const hiddenHeld = (waitsOn: boolean, nowMs: number, threadId: string) =>
+    waitsOn
+      ? liveReservations([...replacing.values()].flat(), nowMs).filter(
+          (old) => old.threadId !== threadId && !usedKeys.has(old.key),
+        )
+      : [];
+
   const freeMemoryBytes = (limits: SlotLimits) =>
     limits.minFreeMemoryBytes === null
       ? Effect.succeed(null)
@@ -294,6 +308,7 @@ const make = Effect.gen(function* () {
         reservations: onePerThread(
           reservations.filter((reservation) => reservation.threadId !== input.callerThreadId),
         ),
+        accountOnly: hiddenHeld(waitsOn, nowMs, input.callerThreadId),
         targetInstanceId: input.targetInstanceId,
         freeMemoryBytes: yield* freeMemoryBytes(limits),
       });
@@ -374,6 +389,8 @@ const make = Effect.gen(function* () {
           reservations: onePerThread(
             reservations.filter((reservation) => reservation.threadId !== request.threadId),
           ),
+          // Read again for each request: a grant earlier in this call may have hidden holds.
+          accountOnly: hiddenHeld(settings.value.waitFor.enabled, nowMs, request.threadId),
           targetInstanceId: request.account ?? "",
           freeMemoryBytes: memoryBytes,
         });
@@ -446,6 +463,8 @@ const make = Effect.gen(function* () {
         ? []
         : liveStarting(yield* Ref.get(entries), read.visible, nowMs, limits.startGraceMs);
     const reservations = onePerThread(yield* liveHeld(settings.value.waitFor.enabled, nowMs));
+    // Holds hidden by a re-ask not kept yet: on their own account only, as `decide` counts them.
+    const hidden = hiddenHeld(settings.value.waitFor.enabled, nowMs, "");
     const accountMax = (id: string) => {
       const max = agentSlots.maxPerAccount[id as ProviderInstanceId];
       return max === undefined || max <= 0 ? null : max;
@@ -454,7 +473,7 @@ const make = Effect.gen(function* () {
       ...Object.keys(agentSlots.maxPerAccount).filter((id) => accountMax(id) !== null),
       ...read.running.map((agent) => agent.instanceId),
       ...starting.map((entry) => entry.instanceId),
-      ...reservations.flatMap((reservation) =>
+      ...[...reservations, ...hidden].flatMap((reservation) =>
         reservation.account === undefined ? [] : [reservation.account],
       ),
     ]);
@@ -469,7 +488,8 @@ const make = Effect.gen(function* () {
         used:
           read.running.filter((agent) => agent.instanceId === id).length +
           starting.filter((entry) => entry.instanceId === id).length +
-          reservations.filter((reservation) => reservation.account === id).length,
+          reservations.filter((reservation) => reservation.account === id).length +
+          accountOnlyThreads(reservations, hidden, id),
         max: accountMax(id),
       })),
       freeMemoryMb: Math.floor(memory.availableMemoryBytes / (1024 * 1024)),

apps/server/src/waits/WaitPrJudge.test.ts

@@ -15,7 +15,9 @@ import {
 } from "@t3tools/contracts";
 import type { WaitForInput, WaitRecord } from "@t3tools/contracts/waitFor";
 import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider";
+import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
 import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
@@ -32,6 +34,7 @@ import type { WaitDelivery } from "./WaitDelivery.ts";
 import type { WaitJudge } from "./WaitJudges.ts";
 import { makePrJudge } from "./WaitPrJudge.ts";
 import * as WaitService from "./WaitService.ts";
+import { memoryFileSystem } from "./WaitService.testkit.ts";
 import { WAIT_STORE_FILE } from "./WaitStore.ts";
 
 const NOW = Date.parse("2026-11-15T12:00:00.000Z");
@@ -771,6 +774,87 @@ it.layer(NodeServices.layer)("WaitPrJudge", (it) => {
     },
   );
 
+  it.effect(
+    "a pin made while the new wait is being written is saved with it, so a changed remote is never read (#118)",
+    () => {
+      const host = fakeHost();
+      return Effect.gen(function* () {
+        yield* TestClock.setTime(NOW);
+        host.prs.set("acme/widgets#7", { state: "open", headSha: HEAD_A });
+        // In memory, so every write answers at once while the test clock moves.
+        const filePath = `/waits/${WAIT_STORE_FILE}`;
+        const memoryFs = yield* FileSystem.FileSystem;
+        const writing = yield* Deferred.make<void>();
+        const pinned = yield* Deferred.make<void>();
+        let armed = true;
+        // The wait's first write ends in a rename onto the store; it is held there until the pin.
+        const fs: FileSystem.FileSystem = {
+          ...memoryFs,
+          rename: (from, to) =>
+            armed && to === filePath
+              ? Effect.gen(function* () {
+                  armed = false;
+                  yield* Deferred.succeed(writing, undefined);
+                  yield* Deferred.await(pinned);
+                  return yield* memoryFs.rename(from, to);
+                })
+              : memoryFs.rename(from, to),
+        };
+        const letFibersRun = Effect.gen(function* () {
+          for (let turn = 0; turn < 20; turn += 1) yield* Effect.yieldNow;
+        });
+        const prJudge = yield* makePrJudge();
+        // At registration the read answers only after the 3 s limit, while the wait is written.
+        const lateRead: WaitJudge<"pr"> = (target, context) =>
+          context.waitId !== null
+            ? prJudge(target, context)
+            : Deferred.await(writing).pipe(
+                Effect.andThen(
+                  prJudge(target, {
+                    ...context,
+                    pin: (next) =>
+                      context.pin!(next).pipe(Effect.andThen(Deferred.succeed(pinned, undefined))),
+                  }),
+                ),
+              );
+        const service = yield* makeService(filePath, lateRead).pipe(
+          Effect.provideService(FileSystem.FileSystem, fs),
+        );
+        const registering = yield* Effect.forkChild(
+          register(service, prWait({ number: 7, until: "merged" })),
+        );
+        // One second at a time, so each forked step has set its timer before the clock moves.
+        for (let second = 0; second < 5; second += 1) {
+          yield* letFibersRun;
+          yield* TestClock.adjust(1_000);
+        }
+        assert.strictEqual((yield* Fiber.join(registering)).state, "waiting");
+        const pinnedSpec = {
+          number: 7,
+          repository: "acme/widgets",
+          host: "github.com",
+          until: "merged" as const,
+        };
+        assert.deepStrictEqual((yield* storedWaits(filePath))[0]?.spec, pinnedSpec);
+
+        // The project's remote flips before the next check: the wait still reads acme/widgets.
+        host.projects.set(PROJECT, shell(PROJECT, "upstream/widgets"));
+        host.remotes.set(`/work/${PROJECT}`, identityOf("upstream/widgets"));
+        host.prs.set("upstream/widgets#7", { state: "merged", headSha: HEAD_B });
+        for (let pass = 0; pass < 2; pass += 1) {
+          yield* TestClock.adjust(SWEEP);
+          yield* service.judgeDue();
+        }
+        assert.isFalse(host.reads.some((read) => read.includes("upstream/")));
+        assert.deepStrictEqual((yield* storedWaits(filePath))[0]?.spec, pinnedSpec);
+        assert.strictEqual((yield* storedWaits(filePath))[0]?.state, "waiting");
+      }).pipe(
+        Effect.provideService(FileSystem.FileSystem, memoryFileSystem().fs),
+        Effect.provide(Layer.merge(host.layer, ServerSettingsService.layerTest(ON))),
+      );
+    },
+  );
+
   it.effect("a pin never overwrites a wait cancelled while it was read", () => {
     const host = fakeHost();
     return Effect.gen(function* () {