From 9804190b07bf9963913760fd962b1473608c8e3b Mon Sep 17 00:00:00 2001 From: scuttletoby <123843277+scuttletoby@users.noreply.github.com> Date: Sun, 15 Nov 2026 01:02:33 +0000 Subject: [PATCH 1/3] Fork (waits PR 8): slot waits, woken when an agent slot frees A slot wait (wait_for kind "slot") asks the slot gate for a free agent slot. Free at once: met, and the slot is held for the thread for reserveSeconds. Otherwise the sweep grants slots oldest first, holds each one for its thread, and wakes it; the thread's next allowed delegate_task uses the slot up. Held slots count as running agents for every other thread, only while waits are on. Counts that cannot be read are an error, never "free". With slots off, a slot wait is met at once and holds nothing. Waits off: nothing of the store is read or written, and no held slot counts. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/runtimeLayer.ts | 4 + apps/server/src/slots/SlotGate.test.ts | 316 +++++++++- apps/server/src/slots/SlotGate.ts | 327 ++++++++-- .../src/waits/WaitService.slots.test.ts | 567 ++++++++++++++++++ apps/server/src/waits/WaitService.ts | 292 ++++++++- apps/server/src/waits/WaitWorker.ts | 3 + packages/shared/src/agentSlots.test.ts | 52 ++ packages/shared/src/agentSlots.ts | 62 +- packages/shared/src/waitFor.test.ts | 44 ++ packages/shared/src/waitFor.ts | 12 +- 10 files changed, 1602 insertions(+), 77 deletions(-) create mode 100644 apps/server/src/waits/WaitService.slots.test.ts diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 27a1da2ca6b97..2644c713295ec 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -60,6 +60,7 @@ import * as ThreadForkService from "./ThreadForkService.ts"; import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; import * as SecretRequests from "../secrets/SecretRequests.ts"; +import * as SlotGate from "../slots/SlotGate.ts"; export const layerEventInfrastructure = Layer.mergeAll( OrchestrationEventStore.layer, @@ -410,6 +411,9 @@ export const layerProduction = Layer.mergeAll( ), // Fork (waits): off by default; the sweep reads and writes nothing until it is switched on. WaitWorker.layer.pipe( + // The server's slot gate: the very layer server.ts gives /mcp (layers memoize by reference), + // so slot waits and delegate_task share one gate. Its needs are provided below and by server.ts. + Layer.provide(SlotGate.layer), Layer.provide( Layer.mergeAll( layerThreadManagementProvided, diff --git a/apps/server/src/slots/SlotGate.test.ts b/apps/server/src/slots/SlotGate.test.ts index 7241eca51c539..25c5386d9eb9c 100644 --- a/apps/server/src/slots/SlotGate.test.ts +++ b/apps/server/src/slots/SlotGate.test.ts @@ -14,7 +14,7 @@ import { } from "@t3tools/contracts"; import type { AgentSlotsSettings } from "@t3tools/contracts/waitFor"; import type { DeepPartial } from "@t3tools/shared/Struct"; -import { DISPATCH_SAFETY_MS } from "@t3tools/shared/agentSlots"; +import { DISPATCH_SAFETY_MS, type SlotReservation } from "@t3tools/shared/agentSlots"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; @@ -487,4 +487,318 @@ describe("SlotGate", () => { assert.strictEqual((yield* run(world, settings, acquire())).type, "allow"); }), ); + + describe("slots held for slot waits (4.6)", () => { + const holder = ThreadId.make("thread:slot-holder"); + const other = ThreadId.make("thread:slot-other"); + const waitsOn = (patch: Partial) => ({ + ...on(patch), + waitFor: { enabled: true }, + }); + const acquireAs = (callerThreadId: ThreadId, targetInstanceId = MAIN) => + SlotGate.SlotGate.pipe( + Effect.flatMap((gate) => gate.acquire({ callerThreadId, targetInstanceId })), + ); + const grant = (...requests: ReadonlyArray) => + SlotGate.SlotGate.pipe(Effect.flatMap((gate) => gate.grant(requests))); + const types = (grants: ReadonlyArray) => grants.map((g) => g.type); + + it.effect("off, or no limit: every request is 'off', nothing read, nothing held", () => + Effect.gen(function* () { + for (const settings of [{}, { waitFor: { enabled: true } }, waitsOn({})]) { + const world = newWorld(); + world.threads = [{ id: "a", instanceId: MAIN }]; + const grants = yield* run( + world, + settings, + grant({ key: "w1", threadId: holder }, { key: "w2", threadId: other }), + ); + assert.deepStrictEqual(types(grants), ["off", "off"]); + assert.strictEqual(world.snapshotReads, 0); + assert.strictEqual(world.memoryReads, 0); + } + }), + ); + + it.effect( + "two waits for one free slot: the first is held, the second is not; no double start", + () => + Effect.gen(function* () { + const world = newWorld(); + world.threads = [{ id: "a", instanceId: MAIN }]; + yield* run( + world, + waitsOn({ maxPerMachine: 2, reserveSeconds: 60 }), + Effect.gen(function* () { + const grants = yield* grant( + { key: "w1", threadId: holder }, + { key: "w2", threadId: other }, + ); + assert.deepStrictEqual(grants, [{ type: "granted", untilMs: 60_000 }, grants[1]!]); + assert.strictEqual(grants[1]!.type, "full"); + // The slot is held: the other thread is refused, and told to wait for a slot. + const refused = yield* acquireAs(other); + assert.strictEqual(refused.type, "refuse"); + if (refused.type === "refuse") assert.include(refused.message, "wait_for"); + // The holder starts its agent; its held slot is used up, not counted twice. + assert.strictEqual((yield* acquireAs(holder)).type, "allow"); + assert.strictEqual((yield* acquireAs(holder)).type, "refuse"); + assert.strictEqual((yield* acquireAs(other)).type, "refuse"); + }), + ); + }), + ); + + it.effect("a held slot ends after reserveSeconds, and a dropped grant holds nothing", () => + Effect.gen(function* () { + const world = newWorld(); + yield* run( + world, + waitsOn({ maxPerMachine: 1, reserveSeconds: 30 }), + Effect.gen(function* () { + assert.deepStrictEqual(types(yield* grant({ key: "w1", threadId: holder })), [ + "granted", + ]); + assert.strictEqual((yield* acquireAs(other)).type, "refuse"); + yield* TestClock.adjust("30 seconds"); + assert.deepStrictEqual(types(yield* grant({ key: "w2", threadId: holder })), [ + "granted", + ]); + const gate = yield* SlotGate.SlotGate; + yield* gate.drop(["w2"]); + yield* gate.drop([]); + assert.strictEqual((yield* acquireAs(other)).type, "allow"); + }), + ); + }), + ); + + it.effect("reserveSeconds 0: the wait is woken but no slot is held", () => + Effect.gen(function* () { + const world = newWorld(); + yield* run( + world, + waitsOn({ maxPerMachine: 1, reserveSeconds: 0 }), + Effect.gen(function* () { + assert.deepStrictEqual(yield* grant({ key: "w1", threadId: holder }), [ + { type: "granted", untilMs: 0 }, + ]); + assert.strictEqual((yield* acquireAs(other)).type, "allow"); + }), + ); + }), + ); + + it.effect("held slots count only while waits are on (turned off while held)", () => + Effect.gen(function* () { + const world = newWorld(); + yield* run( + world, + waitsOn({ maxPerMachine: 1 }), + Effect.gen(function* () { + yield* grant({ key: "w1", threadId: holder }); + assert.strictEqual((yield* acquireAs(other)).type, "refuse"); + const settings = yield* ServerSettings.ServerSettingsService; + yield* settings.updateSettings({ waitFor: { enabled: false } }); + const allowed = yield* acquireAs(other); + assert.strictEqual(allowed.type, "allow"); + if (allowed.type === "allow") { + const gate = yield* SlotGate.SlotGate; + yield* gate.release(allowed.token); + } + // Slots turned off: everything is allowed and slot waits are 'off'. + yield* settings.updateSettings({ waitFor: { enabled: true } }); + yield* settings.updateSettings({ agentSlots: { enabled: false } }); + assert.strictEqual((yield* acquireAs(other)).type, "allow"); + assert.deepStrictEqual(types(yield* grant({ key: "w2", threadId: other })), ["off"]); + }), + ); + }), + ); + + it.effect("the limit raised while held: the next grant sees the new limit", () => + Effect.gen(function* () { + const world = newWorld(); + world.threads = [{ id: "a", instanceId: MAIN }]; + yield* run( + world, + waitsOn({ maxPerMachine: 1 }), + Effect.gen(function* () { + assert.deepStrictEqual(types(yield* grant({ key: "w1", threadId: holder })), ["full"]); + const settings = yield* ServerSettings.ServerSettingsService; + yield* settings.updateSettings({ agentSlots: { maxPerMachine: 3 } }); + assert.deepStrictEqual( + types( + yield* grant( + { key: "w1", threadId: holder }, + { key: "w2", threadId: other }, + { key: "w3", threadId: caller }, + ), + ), + ["granted", "granted", "full"], + ); + }), + ); + }), + ); + + it.effect("an account wait counts its account; a wait with no account only the machine", () => + Effect.gen(function* () { + const world = newWorld(); + world.threads = [{ id: "a", instanceId: SECOND }]; + yield* run( + world, + waitsOn({ maxPerAccount: { [SECOND]: 1 } }), + Effect.gen(function* () { + assert.deepStrictEqual( + types( + yield* grant( + { key: "w1", threadId: holder, account: SECOND }, + { key: "w2", threadId: other, account: MAIN }, + ), + ), + ["full", "granted"], + ); + // No account and no machine limit: nothing to count, so nothing is read for it. + const reads = world.snapshotReads; + assert.deepStrictEqual(types(yield* grant({ key: "w3", threadId: caller })), [ + "granted", + ]); + assert.strictEqual(world.snapshotReads, reads); + }), + ); + }), + ); + + it.effect("running agents that cannot be counted: 'unread', never granted, nothing held", () => + Effect.gen(function* () { + const world = newWorld(); + world.snapshotFails = true; + yield* run( + world, + waitsOn({ maxPerMachine: 5 }), + Effect.gen(function* () { + assert.deepStrictEqual(types(yield* grant({ key: "w1", threadId: holder })), [ + "unread", + ]); + assert.isNull(yield* (yield* SlotGate.SlotGate).usage); + world.snapshotFails = false; + world.snapshotDies = true; + assert.deepStrictEqual(types(yield* grant({ key: "w1", threadId: holder })), [ + "unread", + ]); + }), + ); + }), + ); + + it.effect("slots held in the wait store: read only while waits are on; used up once", () => + Effect.gen(function* () { + const world = newWorld(); + let reads = 0; + const used: Array = []; + let inStore: ReadonlyArray = [ + { key: "stored", threadId: holder, untilMs: 90_000 }, + ]; + yield* run( + world, + on({ maxPerMachine: 1 }), + Effect.gen(function* () { + const gate = yield* SlotGate.SlotGate; + yield* gate.attachReservations({ + read: Effect.sync(() => { + reads += 1; + return inStore; + }), + used: (key) => + Effect.sync(() => { + used.push(key); + inStore = inStore.filter((entry) => entry.key !== key); + }), + }); + // Waits off: the store is never read and its slot does not count. + assert.strictEqual((yield* acquireAs(other)).type, "allow"); + assert.strictEqual(reads, 0); + assert.deepStrictEqual(used, []); + }), + ); + // Restart with waits on (stored slots outlive the gate's memory). + yield* run( + world, + waitsOn({ maxPerMachine: 1 }), + Effect.gen(function* () { + const gate = yield* SlotGate.SlotGate; + yield* gate.attachReservations({ + read: Effect.sync(() => { + reads += 1; + return inStore; + }), + used: (key) => + Effect.sync(() => { + used.push(key); + inStore = inStore.filter((entry) => entry.key !== key); + }), + }); + assert.strictEqual((yield* acquireAs(other)).type, "refuse"); + assert.isAbove(reads, 0); + const usage = yield* gate.usage; + assert.deepStrictEqual(usage?.machine, { used: 1, max: 1 }); + assert.strictEqual((yield* acquireAs(holder)).type, "allow"); + assert.deepStrictEqual(used, ["stored"]); + }), + ); + }), + ); + + it.effect( + "a used slot whose store write fails no longer counts; the acquire still allows", + () => + Effect.gen(function* () { + const world = newWorld(); + yield* run( + world, + waitsOn({ maxPerMachine: 2 }), + Effect.gen(function* () { + const gate = yield* SlotGate.SlotGate; + yield* gate.attachReservations({ + read: Effect.succeed([{ key: "stored", threadId: holder, untilMs: 90_000 }]), + used: () => Effect.die(new Error("disk full")), + }); + assert.strictEqual((yield* acquireAs(holder)).type, "allow"); + // One starting agent, the store's copy (not cleared) not counted again: one slot free. + assert.strictEqual((yield* acquireAs(other)).type, "allow"); + assert.strictEqual((yield* acquireAs(other)).type, "refuse"); + }), + ); + }), + ); + + it.effect("usage: null while slots are off; counts running, starting and held", () => + Effect.gen(function* () { + const world = newWorld(); + world.threads = [{ id: "a", instanceId: SECOND }]; + assert.isNull( + yield* run(world, {}, SlotGate.SlotGate.pipe(Effect.flatMap((gate) => gate.usage))), + ); + assert.strictEqual(world.snapshotReads, 0); + const usage = yield* run( + world, + waitsOn({ maxPerMachine: 4, maxPerAccount: { [MAIN]: 2 } }), + Effect.gen(function* () { + yield* grant({ key: "w1", threadId: holder, account: MAIN }); + yield* acquireAs(other, MAIN); + return yield* (yield* SlotGate.SlotGate).usage; + }), + ); + assert.deepStrictEqual(usage, { + machine: { used: 3, max: 4 }, + accounts: [ + { id: SECOND, used: 1, max: null }, + { id: MAIN, used: 2, max: 2 }, + ].sort((a, b) => (a.id < b.id ? -1 : 1)), + freeMemoryMb: 16_000, + }); + }), + ); + }); }); diff --git a/apps/server/src/slots/SlotGate.ts b/apps/server/src/slots/SlotGate.ts index 4fe8535cac9e4..722d5e10f99f8 100644 --- a/apps/server/src/slots/SlotGate.ts +++ b/apps/server/src/slots/SlotGate.ts @@ -1,17 +1,23 @@ /** * Fork: the agent slot gate on `delegate_task` (`agentSlots`, off: reads only the settings). Counts - * running agents from the projection; its own state (starting agents) is memory only, never disk. + * running agents from the projection; its own state (starting agents, slots held for met slot + * waits) is memory only, never disk. Slots held by delivered slot waits come from the wait store, + * which the wait service attaches (`attachReservations`); while waits are off none count. */ import type { ProviderInstanceId, ThreadId } from "@t3tools/contracts"; import { effectiveSlotLimits, limitsCount, + liveReservations, liveStarting, runningAgents, slotDecision, slotRefusalText, visibleRunThreadIds, type RunningAgent, + type SlotDecision, + type SlotLimits, + type SlotReservation, type StartingEntry, } from "@t3tools/shared/agentSlots"; import * as Clock from "effect/Clock"; @@ -35,6 +41,41 @@ export type SlotAcquire = | { readonly type: "allow"; readonly token: SlotToken | null } | { readonly type: "refuse"; readonly message: string }; +/** Slots held by delivered slot waits, as the wait store keeps them (`reservedUntil`). */ +export interface StoredReservations { + /** The live ones; read only while waits are on. */ + readonly read: Effect.Effect>; + /** The gate let the holder start an agent: clear the wait's `reservedUntil` (a write). */ + readonly used: (key: string) => Effect.Effect; +} + +/** One slot wait asking for a slot; `key` is its wait id. */ +export interface SlotRequest { + readonly key: string; + readonly threadId: ThreadId; + readonly account?: ProviderInstanceId | undefined; +} + +export type SlotGrant = + /** Slots are off or have no limit: nothing to wait for, nothing held. */ + | { readonly type: "off" } + /** Free: held for the thread until `untilMs` (memory only). */ + | { readonly type: "granted"; readonly untilMs: number } + | { readonly type: "full"; readonly decision: Extract } + /** The running agents cannot be counted now: never "free". */ + | { readonly type: "unread"; readonly message: string }; + +/** What `wait_list` shows while slots are on (agents running, starting or held). */ +export interface SlotUsage { + readonly machine: { readonly used: number; readonly max: number | null }; + readonly accounts: ReadonlyArray<{ + readonly id: ProviderInstanceId; + readonly used: number; + readonly max: number | null; + }>; + readonly freeMemoryMb: number | null; +} + export class SlotGate extends Context.Service< SlotGate, { @@ -54,6 +95,19 @@ export class SlotGate extends Context.Service< * start grace. Calling it again does nothing more. */ readonly release: (token: SlotToken | null) => Effect.Effect; + /** The wait service gives the gate the slots its store holds (once, when it starts). */ + readonly attachReservations: (stored: StoredReservations) => Effect.Effect; + /** + * Slot waits ask for slots, oldest first: each one granted is held for its thread at once, so + * the next request (and every other thread's delegation) counts it. + */ + readonly grant: ( + requests: ReadonlyArray, + ) => Effect.Effect>; + /** Grants that were not kept (the wait was cancelled, or its write failed) hold nothing. */ + readonly drop: (keys: ReadonlyArray) => Effect.Effect; + /** For `wait_list`: null while slots are off, or when the running agents cannot be read. */ + readonly usage: Effect.Effect; } >()("t3/slots/SlotGate") {} @@ -67,83 +121,272 @@ const make = Effect.gen(function* () { const semaphore = yield* Semaphore.make(1); const entries = yield* Ref.make>([]); const nextToken = yield* Ref.make(1); + /** Slots granted to slot waits (memory only; lost on a restart, which costs one more wait). */ + const held = yield* Ref.make>([]); + /** Keys used up, until their end: the store's copy stops counting even if clearing it failed. */ + const usedKeys = new Map(); + let stored: StoredReservations | null = null; /** - * Threads with an active run, settled or not (settling a thread does not stop its run). `undefined` - * when they cannot be read in time: counts never stop a delegation; the memory floor still applies. - * Interruptible on its own, so the timeout works inside the caller's uninterruptible acquire. + * Threads with an active run, settled or not (settling a thread does not stop its run), nobody + * left out. `undefined` when they cannot be read in time: counts never stop a delegation (the + * memory floor still applies) and never make a slot wait "free". Interruptible on its own, so + * the timeout works inside the caller's uninterruptible acquire. */ - const readRunning = (callerThreadId: ThreadId) => - Effect.interruptible(threadManagement.getShellSnapshot({ activeRunOnly: true })).pipe( - // A timeout fails with a TimeoutError, which the catch below logs like any other read failure. - Effect.timeout(COUNT_READ_TIMEOUT), - Effect.map((snapshot) => { - const threads = [...snapshot.threads, ...snapshot.archivedThreads]; - return { - running: runningAgents(threads, callerThreadId), - visible: visibleRunThreadIds(threads), - } satisfies { running: ReadonlyArray; visible: ReadonlySet }; - }), - Effect.catch((cause) => - Effect.logWarning("agent slots: running agents could not be counted", { cause }).pipe( - Effect.as(undefined), - ), + const readRunning = Effect.interruptible( + threadManagement.getShellSnapshot({ activeRunOnly: true }), + ).pipe( + // A timeout fails with a TimeoutError, which the catch below logs like any other read failure. + Effect.timeout(COUNT_READ_TIMEOUT), + Effect.map((snapshot) => { + const threads = [...snapshot.threads, ...snapshot.archivedThreads]; + return { + running: runningAgents(threads, null), + visible: visibleRunThreadIds(threads), + } satisfies { running: ReadonlyArray; visible: ReadonlySet }; + }), + Effect.catch((cause) => + Effect.logWarning("agent slots: running agents could not be counted", { cause }).pipe( + Effect.as(undefined), ), - // A bug in the count read (a defect, not a failure) must not refuse every delegation either. - Effect.catchDefect((defect) => - Effect.logWarning("agent slots: counting running agents crashed", { defect }).pipe( - Effect.as(undefined), - ), + ), + // A bug in the count read (a defect, not a failure) must not refuse every delegation either. + Effect.catchDefect((defect) => + Effect.logWarning("agent slots: counting running agents crashed", { defect }).pipe( + Effect.as(undefined), ), + ), + ); + + /** Ended entries go for good: a child seen with a run counts as running from now on. */ + const keepStarting = ( + visible: ReadonlySet | undefined, + nowMs: number, + limits: SlotLimits, + ) => + Ref.updateAndGet(entries, (current) => + liveStarting(current, visible ?? new Set(), nowMs, limits.startGraceMs), + ); + + /** + * Every live held slot: granted in memory and kept in the wait store, one per wait. None while + * waits are off (nothing of the store is read then). A store that cannot be read holds none. + */ + const liveHeld = Effect.fn("SlotGate.liveHeld")(function* (waitsOn: boolean, nowMs: number) { + for (const [key, untilMs] of usedKeys) if (untilMs <= nowMs) usedKeys.delete(key); + const memory = yield* Ref.updateAndGet(held, (current) => liveReservations(current, nowMs)); + if (!waitsOn) return []; + const fromStore = stored === null ? [] : yield* stored.read; + return liveReservations([...memory, ...fromStore], nowMs).filter( + (reservation) => !usedKeys.has(reservation.key), ); + }); + + const freeMemoryBytes = (limits: SlotLimits) => + limits.minFreeMemoryBytes === null + ? Effect.succeed(null) + : hostResources.read.pipe(Effect.map((sample) => sample.availableMemoryBytes)); + + /** Settings that cannot be read count as off, as for the usage pacing gate. */ + const readSettings = serverSettings.getSettings.pipe(Effect.option); - const acquire = Effect.fn("SlotGate.acquire")( + /** + * The decision, under the semaphore, and the caller's held slot it used up (if any). The + * caller's own held slots never count against it; they are not a pass past a limit either. + */ + const decide = Effect.fn("SlotGate.decide")( function* (input: { readonly callerThreadId: ThreadId; readonly targetInstanceId: ProviderInstanceId; }) { - // Settings that cannot be read count as off, as for the usage pacing gate. - const settings = yield* serverSettings.getSettings.pipe(Effect.option); + const settings = yield* readSettings; const limits = Option.isNone(settings) ? null : effectiveSlotLimits(settings.value.agentSlots); - if (limits === null) return { type: "allow", token: null } satisfies SlotAcquire; + if (Option.isNone(settings) || limits === null) { + return { outcome: { type: "allow", token: null } satisfies SlotAcquire, used: null }; + } + const waitsOn = settings.value.waitFor.enabled; const nowMs = yield* Clock.currentTimeMillis; const counted = limitsCount(limits, input.targetInstanceId); - const read = counted ? yield* readRunning(input.callerThreadId) : undefined; - const running = read?.running; - // Ended entries go for good: a child seen with a run counts as running from now on. - const kept = yield* Ref.updateAndGet(entries, (current) => - liveStarting(current, read?.visible ?? new Set(), nowMs, limits.startGraceMs), - ); - const freeMemoryBytes = - limits.minFreeMemoryBytes === null - ? null - : (yield* hostResources.read).availableMemoryBytes; + const read = counted ? yield* readRunning : undefined; + const running = read?.running.filter((agent) => agent.threadId !== input.callerThreadId); + const kept = yield* keepStarting(read?.visible, nowMs, limits); + const reservations = yield* liveHeld(waitsOn, nowMs); const decision = slotDecision({ limits: running === undefined ? { ...limits, machine: null, account: () => null } : limits, running: running ?? [], starting: kept, + reservations: reservations.filter( + (reservation) => reservation.threadId !== input.callerThreadId, + ), targetInstanceId: input.targetInstanceId, - freeMemoryBytes, + freeMemoryBytes: yield* freeMemoryBytes(limits), }); if (decision.type === "refuse") { - // Slot waits (the `slot` kind of wait_for) come with the wait tools; until then: later. - return { type: "refuse", message: slotRefusalText(decision, false) } satisfies SlotAcquire; + return { + outcome: { + type: "refuse", + message: slotRefusalText(decision, waitsOn), + } satisfies SlotAcquire, + used: null, + }; } const token = yield* Ref.getAndUpdate(nextToken, (value) => value + 1); yield* Ref.update(entries, (current) => [ ...current, { token, instanceId: input.targetInstanceId, atMs: nowMs }, ]); - return { type: "allow", token } satisfies SlotAcquire; + // The agent it held a slot for starts now: the slot ending first is used up. + const own = reservations.find((reservation) => reservation.threadId === input.callerThreadId); + if (own !== undefined) { + usedKeys.set(own.key, own.untilMs); + yield* Ref.update(held, (current) => current.filter((entry) => entry.key !== own.key)); + } + return { outcome: { type: "allow", token } satisfies SlotAcquire, used: own?.key ?? null }; }, // One at a time: the next caller sees the entry the last one added. semaphore.withPermits(1), ); + const acquire = Effect.fn("SlotGate.acquire")(function* (input: { + readonly callerThreadId: ThreadId; + readonly targetInstanceId: ProviderInstanceId; + }) { + const { outcome, used } = yield* decide(input); + // Cleared in the store after the decision, outside the semaphore; a failed write only means + // the stored copy ends by itself (it no longer counts here either way). + if (used !== null && stored !== null) { + yield* stored.used(used).pipe( + Effect.catchCause((cause) => + Effect.logWarning("agent slots: a used slot could not be cleared in the wait store", { + cause, + }), + ), + ); + } + return outcome; + }); + + const grant = Effect.fn("SlotGate.grant")( + function* (requests: ReadonlyArray) { + const settings = yield* readSettings; + const limits = Option.isNone(settings) + ? null + : effectiveSlotLimits(settings.value.agentSlots); + if (Option.isNone(settings) || limits === null) { + return requests.map((): SlotGrant => ({ type: "off" })); + } + if (requests.length === 0) return []; + const nowMs = yield* Clock.currentTimeMillis; + /** A wait without an account waits for the machine limit and the memory floor only. */ + const limitsFor = (request: SlotRequest): SlotLimits => + request.account === undefined ? { ...limits, account: () => null } : limits; + const counted = requests.some((request) => + limitsCount(limitsFor(request), request.account ?? ""), + ); + // Read only when a count limit applies to one of them (3.6). + const read = counted ? yield* readRunning : undefined; + const kept = yield* keepStarting(read?.visible, nowMs, limits); + const memoryBytes = yield* freeMemoryBytes(limits); + const reserveMs = settings.value.agentSlots.reserveSeconds * 1000; + // Slots held for the requesting thread count too: it never holds two for one free slot. + const reservations = [...(yield* liveHeld(settings.value.waitFor.enabled, nowMs))]; + const grants: Array = []; + for (const request of requests) { + const own = limitsFor(request); + if (read === undefined && limitsCount(own, request.account ?? "")) { + grants.push({ type: "unread", message: "The running agents cannot be counted now." }); + continue; + } + const decision = slotDecision({ + limits: own, + running: (read?.running ?? []).filter((agent) => agent.threadId !== request.threadId), + starting: kept, + reservations, + targetInstanceId: request.account ?? "", + freeMemoryBytes: memoryBytes, + }); + if (decision.type === "refuse") { + grants.push({ type: "full", decision }); + continue; + } + const untilMs = nowMs + reserveMs; + const reservation: SlotReservation = { + key: request.key, + threadId: request.threadId, + ...(request.account === undefined ? {} : { account: request.account }), + untilMs, + }; + // A slot held for 0 s is not held at all; the wait is still woken. + if (reserveMs > 0) { + reservations.push(reservation); + yield* Ref.update(held, (current) => [...current, reservation]); + } + grants.push({ type: "granted", untilMs }); + } + return grants; + }, + // Under the same semaphore as `acquire`: a delegation never takes a slot being granted. + semaphore.withPermits(1), + ); + + const usage = Effect.gen(function* () { + const settings = yield* readSettings; + if (Option.isNone(settings) || !settings.value.agentSlots.enabled) return null; + const agentSlots = settings.value.agentSlots; + const read = yield* readRunning; + if (read === undefined) return null; + const nowMs = yield* Clock.currentTimeMillis; + const limits = effectiveSlotLimits(agentSlots); + const starting = + limits === null + ? [] + : liveStarting(yield* Ref.get(entries), read.visible, nowMs, limits.startGraceMs); + const reservations = yield* liveHeld(settings.value.waitFor.enabled, nowMs); + const accountMax = (id: string) => { + const max = agentSlots.maxPerAccount[id as ProviderInstanceId]; + return max === undefined || max <= 0 ? null : max; + }; + const ids = new Set([ + ...Object.keys(agentSlots.maxPerAccount).filter((id) => accountMax(id) !== null), + ...read.running.map((agent) => agent.instanceId), + ...starting.map((entry) => entry.instanceId), + ...reservations.flatMap((reservation) => + reservation.account === undefined ? [] : [reservation.account], + ), + ]); + const memory = yield* hostResources.read; + return { + machine: { + used: read.running.length + starting.length + reservations.length, + max: agentSlots.maxPerMachine > 0 ? agentSlots.maxPerMachine : null, + }, + accounts: [...ids].sort().map((id) => ({ + id: id as ProviderInstanceId, + used: + read.running.filter((agent) => agent.instanceId === id).length + + starting.filter((entry) => entry.instanceId === id).length + + reservations.filter((reservation) => reservation.account === id).length, + max: accountMax(id), + })), + freeMemoryMb: Math.floor(memory.availableMemoryBytes / (1024 * 1024)), + } satisfies SlotUsage; + }).pipe(Effect.withSpan("SlotGate.usage")); + return SlotGate.of({ acquire, + grant, + usage, + attachReservations: Effect.fn("SlotGate.attachReservations")((source: StoredReservations) => + Effect.sync(() => { + stored = source; + }), + ), + drop: Effect.fn("SlotGate.drop")(function* (keys: ReadonlyArray) { + if (keys.length === 0) return; + yield* Ref.update(held, (current) => current.filter((entry) => !keys.includes(entry.key))); + }), started: Effect.fn("SlotGate.started")((token: SlotToken | null, childThreadId: ThreadId) => token === null ? Effect.void diff --git a/apps/server/src/waits/WaitService.slots.test.ts b/apps/server/src/waits/WaitService.slots.test.ts new file mode 100644 index 0000000000000..12499c5707355 --- /dev/null +++ b/apps/server/src/waits/WaitService.slots.test.ts @@ -0,0 +1,567 @@ +/** + * Fork (waits, PR 8): slot waits with the real slot gate. Who is woken when slots free, that a + * woken thread's slot is held for it and used once, and the state question: settings changed while + * waiting, a stopped or archived waiter, a restart with held slots in the store, counts that cannot + * be read, and waits off (nothing read or written). + */ +import { assert, describe, it } from "@effect/vitest"; +import { + type OrchestrationV2ThreadShell, + type OrchestrationV2ThreadShellSnapshot, + ProjectId, + ProviderInstanceId, + type ServerSettings as ServerSettingsValue, + ThreadId, +} from "@t3tools/contracts"; +import type { WaitForInput, WaitRecord } from "@t3tools/contracts/waitFor"; +import type { DeepPartial } from "@t3tools/shared/Struct"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import { TestClock } from "effect/testing"; + +import * as OrchestratorV2 from "../orchestration-v2/Orchestrator.ts"; +import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; +import * as HostResources from "../resourceTelemetry/HostResources.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as SlotGate from "../slots/SlotGate.ts"; +import type { WaitDelivery, WaitReceiver } from "./WaitDelivery.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-14T12:00:00.000Z"); +const PROJECT = ProjectId.make("project-a"); +const PARENT = ThreadId.make("thread-parent"); +const [T1, T2, T3, OUTSIDER] = ["thread-1", "thread-2", "thread-3", "thread-4"].map((id) => + ThreadId.make(id), +) as [ThreadId, ThreadId, ThreadId, ThreadId]; +const MAIN = ProviderInstanceId.make("codex"); +const FILE = `/waits/state/${WAIT_STORE_FILE}`; +const SLOT: WaitForInput = { kind: "slot" }; + +/** Slots and waits on, `max` agents on this machine, a slot held 60 s. */ +const slotsOn = (max: number, extra: DeepPartial = {}) => + ({ + ...extra, + waitFor: { enabled: true, ...extra.waitFor }, + agentSlots: { enabled: true, maxPerMachine: max, reserveSeconds: 60, ...extra.agentSlots }, + }) satisfies DeepPartial; + +interface World { + /** Agents the projection shows running (thread id per agent). */ + running: Array; + snapshotReads: number; + snapshotFails: boolean; + /** Runs once, right after the next count read (a slot frees while a wait is being made). */ + afterRead: (() => void) | null; +} + +const newWorld = (): World => ({ + running: [], + snapshotReads: 0, + snapshotFails: false, + afterRead: null, +}); + +const snapshotOf = (world: World): OrchestrationV2ThreadShellSnapshot => ({ + schemaVersion: 1, + snapshotSequence: 1, + threads: world.running.map( + (id) => + ({ + id: ThreadId.make(id), + providerInstanceId: MAIN, + activeRunId: `run:${id}`, + }) as unknown as OrchestrationV2ThreadShell, + ), + archivedThreads: [], +}); + +/** One server: settings, the gate (built fresh on each provide, as on a restart) and its inputs. */ +const serverLayer = (world: World, settings: DeepPartial) => + SlotGate.layer.pipe( + Layer.provideMerge( + Layer.mergeAll( + ServerSettings.layerTest(settings), + Layer.mock(ThreadManagementService.ThreadManagementService)({ + getShellSnapshot: () => + Effect.gen(function* () { + world.snapshotReads += 1; + if (world.snapshotFails) { + return yield* new OrchestratorV2.OrchestratorProjectionError({ threadId: T1 }); + } + const snapshot = snapshotOf(world); + const after = world.afterRead; + world.afterRead = null; + after?.(); + return snapshot; + }), + }), + Layer.mock(HostResources.HostResources)({ + read: Effect.succeed({ + sampledAt: 0, + cpuUtilization: null, + cpuCount: 4, + availableMemoryBytes: 16_000 * 1024 * 1024, + totalMemoryBytes: 32_000 * 1024 * 1024, + }), + }), + ), + ), + ); + +/** Delivers at once unless the receiver is stopped or gone; records every message sent. */ +const fakeDelivery = () => { + const sent: Array = []; + const receivers = new Map(); + const delivery: WaitDelivery = { + receiver: (threadId) => Effect.sync(() => receivers.get(threadId) ?? { type: "ok" }), + deliver: (message) => + Effect.sync(() => { + const receiver = receivers.get(message.threadId); + if (receiver !== undefined && receiver.type !== "ok") return receiver; + if (!sent.some((earlier) => earlier.messageId === message.messageId)) sent.push(message); + return { type: "delivered" } as const; + }), + }; + return { sent, receivers, delivery }; +}; + +/** The store on the in-memory disk, with a count of its reads. */ +const disk = () => { + const memory = memoryFileSystem(); + const counts = { reads: 0, writes: 0 }; + const fs: FileSystem.FileSystem = { + ...memory.fs, + readFileString: (path, encoding) => { + counts.reads += 1; + return memory.fs.readFileString(path, encoding); + }, + writeFileString: (path, contents, options) => { + counts.writes += 1; + return memory.fs.writeFileString(path, contents, options); + }, + }; + return { fs, files: memory.files, counts }; +}; + +const stored = (files: Map) => + (JSON.parse(files.get(FILE) ?? '{"waits":[]}') as { waits: ReadonlyArray }).waits; + +const startService = (fs: FileSystem.FileSystem, delivery: WaitDelivery) => + Effect.gen(function* () { + const gate = yield* SlotGate.SlotGate; + const service = yield* WaitService.make({ + filePath: FILE, + delivery, + fileChildJudge: null, + slots: gate, + }).pipe(Effect.provideService(FileSystem.FileSystem, fs), Effect.provide(Path.layer)); + const register = (threadId: ThreadId, input: WaitForInput = SLOT) => + service.register({ threadId, projectId: PROJECT, parentThreadId: PARENT, input }); + const delegate = (callerThreadId: ThreadId) => + gate + .acquire({ callerThreadId, targetInstanceId: MAIN }) + .pipe(Effect.map((outcome) => outcome.type)); + return { gate, service, register, delegate }; + }); + +const waitingId = (result: { readonly state: string; readonly waitId?: string }) => { + assert.strictEqual(result.state, "waiting"); + return result.waitId!; +}; + +describe("WaitService slot waits", () => { + it.effect("T17: three waits, two slots free: the two oldest are woken and held for", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.running = ["busy-1", "busy-2"]; + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, delegate, service } = yield* startService(store.fs, send.delivery); + const woken = () => send.sent.map((message) => message.threadId).sort(); + const ids: Array = []; + for (const thread of [T1, T2, T3]) { + ids.push(waitingId(yield* register(thread))); + yield* TestClock.adjust(1_000); + } + // Both agents end: one sweep wakes the two oldest, the third keeps waiting. + world.running = []; + yield* service.runDue(); + // Woken together; their messages may go out in either order. + assert.deepStrictEqual(woken(), [T1, T2]); + assert.include(send.sent[0]!.text, "free"); + assert.include(send.sent[0]!.text, "delegate_task"); + const after = stored(store.files); + assert.deepStrictEqual( + after.map((wait) => [wait.threadId, wait.state, wait.reservedUntil !== undefined]), + [ + [T1, "delivered", true], + [T2, "delivered", true], + [T3, "waiting", false], + ], + ); + // Another thread finds no slot; the woken ones each start exactly one agent. + assert.strictEqual(yield* delegate(OUTSIDER), "refuse"); + assert.strictEqual(yield* delegate(T1), "allow"); + assert.strictEqual(yield* delegate(T1), "refuse"); + assert.isUndefined(stored(store.files).find((wait) => wait.id === ids[0])?.reservedUntil); + // T2 never delegates: its slot ends after 60 s and goes to T3. + yield* TestClock.adjust(59_000); + yield* service.runDue(); + assert.strictEqual(send.sent.length, 2); + yield* TestClock.adjust(2_000); + yield* service.runDue(); + assert.deepStrictEqual(woken(), [T1, T2, T3]); + }).pipe(Effect.provide(serverLayer(world, slotsOn(2)))); + }), + ); + + it.effect("two waits for one free slot: only the oldest is woken; no double start", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.running = ["busy-1"]; + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, delegate, service } = yield* startService(store.fs, send.delivery); + waitingId(yield* register(T2)); + yield* TestClock.adjust(1_000); + waitingId(yield* register(T1)); + world.running = []; + // Two sweeps: the second must not wake T1 on the slot held for T2. + yield* service.runDue(); + yield* service.runDue(); + assert.deepStrictEqual( + send.sent.map((message) => message.threadId), + [T2], + ); + assert.strictEqual(yield* delegate(T1), "refuse"); + assert.strictEqual(yield* delegate(T2), "allow"); + assert.strictEqual(yield* delegate(T2), "refuse"); + // Its agent counts as running: the slot is still taken. + yield* service.runDue(); + assert.strictEqual(send.sent.length, 1); + }).pipe(Effect.provide(serverLayer(world, slotsOn(1)))); + }), + ); + + it.effect("a free slot at registration is met at once, held, and nothing is stored", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, delegate } = yield* startService(store.fs, send.delivery); + const result = yield* register(T1); + assert.strictEqual(result.state, "met"); + if (result.state === "met") { + assert.strictEqual(result.outcome, "free"); + assert.include(result.detail, "held for this thread for 60 s"); + } + assert.deepStrictEqual(stored(store.files), []); + assert.strictEqual(yield* delegate(OUTSIDER), "refuse"); + assert.strictEqual(yield* delegate(T1), "allow"); + // A newer slot wait behind it waits (the slot is now running T1's agent). + waitingId(yield* register(T2)); + }).pipe(Effect.provide(serverLayer(world, slotsOn(1)))); + }), + ); + + it.effect("a slot that frees while the wait is being made: stored, then woken next sweep", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.running = ["busy-1"]; + world.afterRead = () => { + world.running = []; + }; + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, service } = yield* startService(store.fs, send.delivery); + const id = waitingId(yield* register(T1)); + assert.deepStrictEqual( + stored(store.files).map((wait) => [wait.id, wait.state]), + [[id, "waiting"]], + ); + // Due at once, not after a sweep interval. + const { nextDueAt } = yield* service.runDue(); + assert.deepStrictEqual( + send.sent.map((message) => message.threadId), + [T1], + ); + assert.isAbove(nextDueAt, NOW); + }).pipe(Effect.provide(serverLayer(world, slotsOn(1)))); + }), + ); + + it.effect("a newer wait never jumps the queue, even when it sees a free slot", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.running = ["busy-1"]; + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, service } = yield* startService(store.fs, send.delivery); + waitingId(yield* register(T1)); + world.running = []; + yield* TestClock.adjust(1_000); + // Free now, but T1 is older: T2 waits for the sweep, which wakes T1. + waitingId(yield* register(T2)); + yield* service.runDue(); + assert.deepStrictEqual( + send.sent.map((message) => message.threadId), + [T1], + ); + }).pipe(Effect.provide(serverLayer(world, slotsOn(1)))); + }), + ); + + it.effect("slots turned off while waiting: woken with 'slots are off'; limit raised: woken", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.running = ["busy-1", "busy-2"]; + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, service } = yield* startService(store.fs, send.delivery); + const settings = yield* ServerSettings.ServerSettingsService; + waitingId(yield* register(T1)); + yield* TestClock.adjust(1_000); + waitingId(yield* register(T2)); + yield* service.runDue(); + assert.deepStrictEqual(send.sent, []); + yield* settings.updateSettings({ agentSlots: { maxPerMachine: 3 } }); + yield* service.runDue(); + assert.deepStrictEqual( + send.sent.map((message) => message.threadId), + [T1], + ); + yield* settings.updateSettings({ agentSlots: { enabled: false } }); + yield* service.runDue(); + assert.deepStrictEqual( + send.sent.map((message) => message.threadId), + [T1, T2], + ); + assert.include(send.sent[1]!.text, "slots are off"); + // Off: no slot is held for it. + assert.isUndefined(stored(store.files).find((wait) => wait.threadId === T2)?.reservedUntil); + }).pipe(Effect.provide(serverLayer(world, slotsOn(2)))); + }), + ); + + it.effect("an archived waiter: its wait goes to the parent and never holds a slot", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.running = ["busy-1"]; + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, delegate, service } = yield* startService(store.fs, send.delivery); + waitingId(yield* register(T1)); + send.receivers.set(T1, { type: "gone", parentThreadId: PARENT }); + yield* service.receiverChanged(T1); + yield* service.runDue(); + assert.deepStrictEqual( + send.sent.map((message) => message.threadId), + [PARENT], + ); + world.running = []; + yield* service.runDue(); + assert.strictEqual(send.sent.length, 1); + assert.isUndefined(stored(store.files)[0]?.reservedUntil); + assert.strictEqual(yield* delegate(OUTSIDER), "allow"); + }).pipe(Effect.provide(serverLayer(world, slotsOn(1)))); + }), + ); + + it.effect("a stopped waiter: granted, its message held, its slot freed after 60 s", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.running = ["busy-1"]; + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, delegate, service } = yield* startService(store.fs, send.delivery); + waitingId(yield* register(T1)); + send.receivers.set(T1, { type: "stopped", parentThreadId: PARENT }); + world.running = []; + yield* service.runDue(); + assert.deepStrictEqual(send.sent, []); + assert.strictEqual(stored(store.files)[0]?.state, "delivering"); + assert.strictEqual(yield* delegate(OUTSIDER), "refuse"); + yield* TestClock.adjust(61_000); + yield* service.runDue(); + const allowed = yield* SlotGate.SlotGate.pipe( + Effect.flatMap((gate) => + gate.acquire({ callerThreadId: OUTSIDER, targetInstanceId: MAIN }), + ), + ); + assert.strictEqual(allowed.type, "allow"); + }).pipe(Effect.provide(serverLayer(world, slotsOn(1)))); + }), + ); + + it.effect("server restart: open waits are woken again, and stored held slots still count", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.running = ["busy-1", "busy-2"]; + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, service } = yield* startService(store.fs, send.delivery); + waitingId(yield* register(T1)); + yield* TestClock.adjust(1_000); + waitingId(yield* register(T2)); + world.running = ["busy-1"]; + yield* service.runDue(); + assert.deepStrictEqual( + send.sent.map((message) => message.threadId), + [T1], + ); + }).pipe(Effect.provide(serverLayer(world, slotsOn(2)))); + // A new server: the gate's memory is gone, the store keeps T1's held slot and T2's wait. + yield* Effect.gen(function* () { + const { delegate, service } = yield* startService(store.fs, send.delivery); + yield* service.runDue(); + assert.strictEqual(send.sent.length, 1); + assert.strictEqual(yield* delegate(OUTSIDER), "refuse"); + assert.strictEqual(yield* delegate(T1), "allow"); + assert.isUndefined(stored(store.files).find((wait) => wait.threadId === T1)?.reservedUntil); + // T1's agent is now starting; busy-1 ends and T2 is woken. + world.running = []; + yield* service.runDue(); + assert.deepStrictEqual( + send.sent.map((message) => message.threadId), + [T1, T2], + ); + }).pipe(Effect.provide(serverLayer(world, slotsOn(2)))); + }), + ); + + it.effect("counts that cannot be read: an error, then check_error; never 'free'", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.snapshotFails = true; + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, service } = yield* startService(store.fs, send.delivery); + waitingId(yield* register(T1)); + yield* service.runDue(); + assert.deepStrictEqual(send.sent, []); + assert.include(stored(store.files)[0]?.lastError ?? "", "cannot be counted"); + // The read works again: the error clears and the free slot is granted. + world.snapshotFails = false; + yield* TestClock.adjust(15_000); + yield* service.runDue(); + assert.deepStrictEqual( + send.sent.map((message) => message.threadId), + [T1], + ); + // Unreadable for longer than the grace: the wait ends as an error. + world.snapshotFails = true; + waitingId(yield* register(T2)); + for (let step = 0; step < 6; step += 1) { + yield* TestClock.adjust(15_000); + yield* service.runDue(); + } + const ended = stored(store.files).find((wait) => wait.threadId === T2); + assert.strictEqual(ended?.result?.outcome, "check_error"); + assert.isUndefined(ended?.reservedUntil); + }).pipe( + Effect.provide(serverLayer(world, slotsOn(1, { waitFor: { errorGraceSeconds: 60 } }))), + ); + }), + ); + + it.effect("waits off (the default): no store read or written, its held slots do not count", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + const store = disk(); + store.files.set( + FILE, + JSON.stringify({ + version: 1, + waits: [ + { + id: "w_aaaaaaaaaa", + threadId: T1, + projectId: PROJECT, + key: "0123456789abcdef", // gitleaks:allow (a fake test value, not a key) + createdAt: "2026-11-14T11:00:00.000Z", + deadlineAt: "2026-11-14T13:00:00.000Z", + state: "delivered", + endedAt: "2026-11-14T11:59:30.000Z", + attempt: 0, + kind: "slot", + spec: {}, + result: { outcome: "free", detail: "", at: "2026-11-14T11:59:30.000Z" }, + reservedUntil: "2026-11-14T12:00:30.000Z", + }, + ], + }), + ); + const before = store.files.get(FILE); + const send = fakeDelivery(); + const settings = { agentSlots: { enabled: true, maxPerMachine: 1 } }; + yield* Effect.gen(function* () { + const { delegate, service } = yield* startService(store.fs, send.delivery); + assert.strictEqual(yield* delegate(OUTSIDER), "allow"); + yield* service.runDue(); + assert.deepStrictEqual(store.counts, { reads: 0, writes: 0 }); + assert.strictEqual(store.files.get(FILE), before); + // Slots off too: nothing is counted at all. + }).pipe(Effect.provide(serverLayer(world, settings))); + world.snapshotReads = 0; + yield* Effect.gen(function* () { + const { delegate, service } = yield* startService(store.fs, send.delivery); + assert.strictEqual(yield* delegate(OUTSIDER), "allow"); + yield* service.runDue(); + assert.strictEqual(world.snapshotReads, 0); + assert.deepStrictEqual(store.counts, { reads: 0, writes: 0 }); + }).pipe(Effect.provide(serverLayer(world, {}))); + // Waits on: the same stored slot counts, so another thread is refused. + yield* Effect.gen(function* () { + const { delegate } = yield* startService(store.fs, send.delivery); + assert.strictEqual(yield* delegate(OUTSIDER), "refuse"); + }).pipe(Effect.provide(serverLayer(world, { ...settings, waitFor: { enabled: true } }))); + }), + ); + + it.effect("wait_list shows slot usage only while slots are on", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.running = ["busy-1"]; + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, service } = yield* startService(store.fs, send.delivery); + waitingId(yield* register(T1)); + const listed = yield* service.list({ threadId: T1, projectId: PROJECT }); + assert.deepStrictEqual(listed.slots?.machine, { used: 1, max: 1 }); + const settings = yield* ServerSettings.ServerSettingsService; + yield* settings.updateSettings({ agentSlots: { enabled: false } }); + const reads = world.snapshotReads; + const off = yield* service.list({ threadId: T1, projectId: PROJECT }); + assert.isUndefined(off.slots); + assert.strictEqual(world.snapshotReads, reads); + }).pipe(Effect.provide(serverLayer(world, slotsOn(1)))); + }), + ); +}); diff --git a/apps/server/src/waits/WaitService.ts b/apps/server/src/waits/WaitService.ts index d6fb090ae8aa8..ad43918e1779e 100644 --- a/apps/server/src/waits/WaitService.ts +++ b/apps/server/src/waits/WaitService.ts @@ -14,12 +14,18 @@ import { type WaitKind, type WaitListResult, type WaitRecord, + type WaitSlotSpec, type WaitStoreNotice, type WaitTarget, WaitToolFailure, type WaitToolFailureCode, } from "@t3tools/contracts/waitFor"; import * as HostProcess from "@t3tools/shared/HostProcess"; +import { + effectiveSlotLimits, + slotRefusalReason, + type SlotReservation, +} from "@t3tools/shared/agentSlots"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { capEndedWaits, @@ -56,6 +62,7 @@ import * as Option from "effect/Option"; import * as Semaphore from "effect/Semaphore"; import { ServerSettingsService } from "../serverSettings.ts"; +import type { SlotGate, SlotGrant } from "../slots/SlotGate.ts"; import type { WaitDelivery, WaitDeliveryResult } from "./WaitDelivery.ts"; import { type CheckLookup, @@ -213,9 +220,61 @@ const stateSettings = (settings: ServerSettings): WaitStateSettings => ({ errorGraceSeconds: settings.waitFor.errorGraceSeconds, keepEndedHours: settings.waitFor.keepEndedHours, stoppedHandOverHours: settings.waitFor.stoppedHandOverHours, - reserveSeconds: settings.agentSlots.reserveSeconds, + // Slots off or without a limit: a woken slot wait holds nothing. + reserveSeconds: + effectiveSlotLimits(settings.agentSlots) === null ? 0 : settings.agentSlots.reserveSeconds, }); +/** A slot wait's answer when the gate gave or refused a slot (`atRegister`: the call's own answer). */ +const slotJudgement = ( + grant: SlotGrant, + settings: ServerSettings, + atRegister: boolean, +): Judgement => { + const seconds = settings.agentSlots.reserveSeconds; + switch (grant.type) { + case "off": + return { + type: "met", + outcome: "free", + detail: "Agent slots are off, so delegate_task is not held back.", + }; + case "granted": + return { + type: "met", + outcome: "free", + detail: + seconds === 0 + ? "Call delegate_task now." + : atRegister + ? `A slot is free and held for this thread for ${seconds} s: call delegate_task now.` + : `Call delegate_task within ${seconds} s; the slot is held for this thread.`, + }; + case "full": + return { type: "notYet", detail: `No free slot yet: ${slotRefusalReason(grant.decision)}.` }; + case "unread": + return { type: "failed", message: grant.message, permanent: false }; + } +}; + +/** The slot a delivered slot wait still holds, from its `reservedUntil`. */ +const storedReservation = (wait: WaitRecord, now: number): ReadonlyArray => { + if (wait.kind !== "slot" || wait.state !== "delivered" || wait.reservedUntil === undefined) { + return []; + } + const untilMs = Date.parse(wait.reservedUntil); + if (!(untilMs > now)) return []; + const account = wait.spec.account; + return [ + { + key: wait.id, + threadId: wait.threadId, + untilMs, + ...(account === undefined ? {} : { account }), + }, + ]; +}; + const waitingResult = (wait: WaitRecord, deduped: boolean): WaitForResult => ({ state: "waiting", waitId: wait.id, @@ -327,6 +386,11 @@ export const make = Effect.fn("WaitService.make")(function* (input: { * the check judge works it out itself, and runs are timed by the stored project. */ readonly checkPlaceOf?: (context: WaitJudgeContext) => Effect.Effect; + /** + * The server's slot gate: it grants slot waits their slots, and counts the slots delivered slot + * waits hold. Without it, slot waits are refused as not supported. + */ + readonly slots?: SlotGate["Service"]; }) { const onChange = input.onChange ?? Effect.void; const settingsService = yield* ServerSettingsService; @@ -551,6 +615,58 @@ export const make = Effect.fn("WaitService.make")(function* (input: { Effect.mapError((error) => failure("store_unavailable", error.message)), ); + /** + * A slot wait's answer at registration: the gate grants a free slot and holds it at once (under + * the wait's id, which is never stored when it is met). Behind older slot waits it waits for the + * sweep, which grants oldest first; with slots off it is met at once. + */ + const slotAtRegister = Effect.fn("WaitService.slotAtRegister")(function* ( + gate: SlotGate["Service"], + waitId: string, + threadId: ThreadId, + spec: typeof WaitSlotSpec.Type, + queued: boolean, + ) { + const settings = yield* readSettings; + if (queued && effectiveSlotLimits(settings.agentSlots) !== null) { + return { type: "notYet", detail: "Older slot waits go first." } satisfies Judgement; + } + const [grant] = yield* gate.grant([{ key: waitId, threadId, account: spec.account }]); + return grant === undefined + ? ({ type: "notYet" } satisfies Judgement) + : slotJudgement(grant, settings, true); + }); + + /** The slots delivered slot waits still hold, for the gate (it reads them only while waits are on). */ + const heldInStore = lock.withPermits(1)( + Effect.gen(function* () { + const loaded = yield* ensureLoaded(true); + if (!loaded.ok) return []; + const now = yield* Clock.currentTimeMillis; + return loaded.contents.waits.flatMap((wait) => storedReservation(wait, now)); + }), + ); + + /** The gate let a holder start its agent: its wait's `reservedUntil` is cleared (row 15). */ + const useStoredSlot = (key: string) => + lock.withPermits(1)( + Effect.gen(function* () { + const settings = yield* readSettingsOrNull; + if (settings === null || !settings.waitFor.enabled) return; + const loaded = yield* ensureLoaded(true); + if (!loaded.ok) return; + const wait = loaded.contents.waits.find((candidate) => candidate.id === key); + if (wait === undefined) return; + const now = yield* Clock.currentTimeMillis; + const transition = nextWaitState(wait, { type: "slotUsed" }, now, stateSettings(settings)); + if (transition === null) return; + const waits = loaded.contents.waits.map((candidate) => + candidate.id === key ? transition.record : candidate, + ); + yield* commitWaits(loaded.contents, waits, "the used slot ends by itself.", settings); + }), + ); + /** `wait_for`: an open wait for the same condition, a met answer, or a new stored wait. */ const register = Effect.fn("WaitService.register")(function* (request: { readonly threadId: ThreadId; @@ -576,7 +692,8 @@ export const make = Effect.fn("WaitService.make")(function* (input: { if (!valid.ok) return yield* failure(valid.code, valid.message); const target = valid.target; const judge = judgeOf(target.kind); - if (judge === undefined) { + const slotGate = target.kind === "slot" ? input.slots : undefined; + if (judge === undefined && slotGate === undefined) { return yield* failure("not_supported", `This server cannot wait for "${target.kind}" yet.`); } if (target.kind === "check") { @@ -628,7 +745,11 @@ export const make = Effect.fn("WaitService.make")(function* (input: { if (Option.isSome(found)) return { type: "existing", wait: found.value } as const; const id = newWaitId(new Set([...contents.waits.map((wait) => wait.id), ...reserved])); reserved.add(id); - return { type: "new", id } as const; + // Older slot waits go first: a new one waits behind them for the sweep (R17). + const queued = contents.waits.some( + (wait) => wait.kind === "slot" && wait.state === "waiting", + ); + return { type: "new", id, queued } as const; }), ); if (first.type === "existing") return waitingResult(first.wait, true); @@ -641,20 +762,31 @@ export const make = Effect.fn("WaitService.make")(function* (input: { // Waits for a free place within its limit; with none, the next sweep judges it first. const until = now + REGISTER_JUDGE_LIMIT_MS; const context: WaitJudgeContext = { waitId: null, now, threadId, projectId }; - started = yield* startJudgeWhenFree( - target.kind, - { - waitId, - threadId, - target: waitTargetText(target), - ...(target.kind === "check" ? { check: { spec: target.spec, projectId } } : {}), - }, - until, - () => judge(target, context), - childJudgeOf(target, context), - judgeLimitMs(target, projectId, settings), - ); - const answer = started === null ? Option.none() : yield* awaitJudge(started, until); + // A slot wait asks the gate instead, which holds a slot it grants at once. + started = + slotGate !== undefined + ? null + : yield* startJudgeWhenFree( + target.kind, + { + waitId, + threadId, + target: waitTargetText(target), + ...(target.kind === "check" ? { check: { spec: target.spec, projectId } } : {}), + }, + until, + () => judge!(target, context), + childJudgeOf(target, context), + judgeLimitMs(target, projectId, settings), + ); + const answer = + slotGate !== undefined && target.kind === "slot" + ? Option.some( + yield* slotAtRegister(slotGate, waitId, threadId, target.spec, first.queued), + ) + : started === null + ? Option.none() + : yield* awaitJudge(started, until); const judgement: Judgement = Option.getOrElse(answer, (): Judgement => ({ type: "notYet" })); // A wait that could only ever end in this error is refused, and nothing is stored. if (judgement.type === "failed" && judgement.permanent) { @@ -730,7 +862,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { readonly projectId: ProjectId; readonly scope?: "thread" | "project" | undefined; readonly includeEnded?: boolean | undefined; - }) { + }): Effect.fn.Return { const settings = yield* readSettings; const checks = Object.entries( resolveProjectSettings(settings, request.projectId).settings.waitChecks, @@ -738,12 +870,17 @@ export const make = Effect.fn("WaitService.make")(function* (input: { .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)) .map(([name, check]) => ({ name, description: check.description, maxArgs: check.maxArgs })); const loaded = yield* lock.withPermits(1)(ensureLoaded(settings.waitFor.enabled)); + // Outside the lock (the gate reads the store's held slots under it); only while slots are on. + const usage = + input.slots === undefined || !settings.agentSlots.enabled ? null : yield* input.slots.usage; + const slots = usage === null ? {} : { slots: usage }; if (!loaded.ok) { return { enabled: settings.waitFor.enabled, storeError: loaded.reason, waits: [], checks, + ...slots, } satisfies WaitListResult; } const waits = loaded.contents.waits @@ -765,7 +902,7 @@ export const make = Effect.fn("WaitService.make")(function* (input: { ...(wait.lastError === undefined ? {} : { lastError: wait.lastError }), ...(wait.via === undefined ? {} : { via: wait.via }), })); - return { enabled: settings.waitFor.enabled, waits, checks } satisfies WaitListResult; + return { enabled: settings.waitFor.enabled, waits, checks, ...slots } satisfies WaitListResult; }); /** @@ -1375,7 +1512,8 @@ export const make = Effect.fn("WaitService.make")(function* (input: { for (const wait of loaded.contents.waits) { if (wait.state !== "waiting") continue; waiting.add(wait.id); - if (isDue(wait, startedAt, settings)) due.push(wait); + // Slot waits are granted by their own step, oldest first. + if (wait.kind !== "slot" && isDue(wait, startedAt, settings)) due.push(wait); } for (const id of nextCheckAt.keys()) if (!waiting.has(id)) nextCheckAt.delete(id); yield* stopEndedRuns; @@ -1465,6 +1603,106 @@ export const make = Effect.fn("WaitService.make")(function* (input: { return { judged, ended }; }); + /** + * Sweep step 3 (4.6): every `waiting` slot wait asks the gate for a slot, oldest first, so a + * sweep wakes as many as there are free slots, each slot held for its thread. A full machine is + * "not yet" (the deadline still ends it); counts that cannot be read are an error, never "free". + * Reads nothing of the projection when no slot wait is waiting. Returns the waits it ended. + */ + const grantSweep = Effect.fnUntraced(function* (settings: ServerSettings) { + const gate = input.slots; + if (gate === undefined) return 0; + const waiting = yield* lock.withPermits(1)( + Effect.gen(function* () { + const loaded = yield* ensureLoaded(true); + if (!loaded.ok) return []; + return loaded.contents.waits + .flatMap((wait) => (wait.kind === "slot" && wait.state === "waiting" ? [wait] : [])) + .sort((a, b) => + a.createdAt !== b.createdAt + ? a.createdAt < b.createdAt + ? -1 + : 1 + : a.id < b.id + ? -1 + : 1, + ); + }), + ); + if (waiting.length === 0) return 0; + // Outside the lock: the gate reads the store's held slots under it. + const grants = yield* gate.grant( + waiting.map((wait) => ({ + key: wait.id, + threadId: wait.threadId, + account: wait.spec.account, + })), + ); + const answers = waiting.map((wait, index) => { + const grant = grants[index]; + return [ + wait.id, + grant === undefined + ? ({ type: "notYet" } satisfies Judgement) + : slotJudgement(grant, settings, false), + ] as const; + }); + const freed = answers.flatMap(([id, judgement]) => + judgement.type === "met" ? [[id, judgement.detail] as const] : [], + ); + // Switched off meanwhile: nothing is written, and nothing stays held. + if (!(yield* stillOn)) { + yield* gate.drop(freed.map(([id]) => id)); + return 0; + } + const granted = yield* applyGrants(new Map(freed), settings); + return ( + granted + + (yield* applyJudgements( + answers.filter(([, judgement]) => judgement.type !== "met"), + settings, + )) + ); + }); + + /** Writes the slots granted; a grant not kept (cancelled meanwhile, or not written) holds nothing. */ + const applyGrants = (granted: ReadonlyMap, settings: ServerSettings) => + lock.withPermits(1)( + Effect.gen(function* () { + if (granted.size === 0) return 0; + const loaded = yield* ensureLoaded(true); + const now = yield* Clock.currentTimeMillis; + const changes = new Map(); + for (const wait of loaded.ok ? loaded.contents.waits : []) { + const detail = granted.get(wait.id); + if (detail === undefined) continue; + const transition = nextWaitState( + wait, + { type: "slotGranted", detail }, + now, + stateSettings(settings), + ); + if (transition !== null) changes.set(wait.id, transition); + } + const written = + loaded.ok && + changes.size > 0 && + (yield* commitWaits( + loaded.contents, + loaded.contents.waits.map((wait) => changes.get(wait.id)?.record ?? wait), + "the slots are granted again next time.", + settings, + )); + const kept = written ? changes : new Map(); + yield* input.slots!.drop([...granted.keys()].filter((id) => !kept.has(id))); + for (const [id, transition] of kept) { + nextCheckAt.delete(id); + yield* logTransition(transition); + } + return kept.size; + }), + ); + /** * Judges every `waiting` wait that is due, at most `JUDGE_CONCURRENCY` at a time (and * `MAX_RUNNING_JUDGES_PER_KIND` per kind), each within its kind's `judgeLimit`. Answers that end @@ -1933,7 +2171,12 @@ export const make = Effect.fn("WaitService.make")(function* (input: { }; if (current.notices.length > 0) at(Math.max(noticesRetryAt, now)); if (receiversToCheck.size > 0) at(now); + // A held slot that ends may go to a slot wait still waiting. + const slotsWaiting = current.waits.some( + (wait) => wait.kind === "slot" && wait.state === "waiting", + ); for (const wait of current.waits) { + if (slotsWaiting) for (const held of storedReservation(wait, now)) at(held.untilMs); if (wait.state === "waiting") at(nextCheckAt.get(wait.id) ?? now); else if (wait.state !== "delivering" && wait.endedAt !== undefined) { at( @@ -1995,7 +2238,9 @@ export const make = Effect.fn("WaitService.make")(function* (input: { if ((yield* phase("deliver", deliverSweep, none)) === null) return stopped; const judged = yield* phase("judge", judgeSweep, { judged: 0, ended: 0 }); if (judged === null) return stopped; - if (judged.ended > 0 && (yield* phase("deliver", deliverSweep, none)) === null) { + const granted = yield* phase("grant", grantSweep, 0); + if (granted === null) return stopped; + if (judged.ended + granted > 0 && (yield* phase("deliver", deliverSweep, none)) === null) { return stopped; } if ((yield* phase("prune", pruneSweep, 0)) === null) return stopped; @@ -2034,6 +2279,11 @@ export const make = Effect.fn("WaitService.make")(function* (input: { if (current !== null) receiversToCheck.add(threadId); }); + // The gate counts the slots the store holds (only while waits are on) and clears a used one. + if (input.slots !== undefined) { + yield* input.slots.attachReservations({ read: heldInStore, used: useStoredSlot }); + } + return { register, list, diff --git a/apps/server/src/waits/WaitWorker.ts b/apps/server/src/waits/WaitWorker.ts index 7ace48d28e244..bdf976b9d34a6 100644 --- a/apps/server/src/waits/WaitWorker.ts +++ b/apps/server/src/waits/WaitWorker.ts @@ -21,6 +21,7 @@ import * as DueHint from "../scheduling/DueHint.ts"; import * as Scheduler from "../scheduling/Scheduler.ts"; import { forkParked } from "../serverActivation.ts"; import { ServerSettingsService } from "../serverSettings.ts"; +import * as SlotGate from "../slots/SlotGate.ts"; import * as WaitDelivery from "./WaitDelivery.ts"; import { type CheckPlace, @@ -88,6 +89,8 @@ export const layer = Layer.effect( onChange: due.invalidate, // Each run is timed by the project its thread is in now. checkPlaceOf, + // The server's one gate (layers memoize by reference): slot waits and delegate_task share it. + slots: yield* SlotGate.SlotGate, }); yield* forkParked( Stream.runForEach(orchestrator.streamDomainEvents, (event) => diff --git a/packages/shared/src/agentSlots.test.ts b/packages/shared/src/agentSlots.test.ts index 845939d70044d..e43fe95535323 100644 --- a/packages/shared/src/agentSlots.test.ts +++ b/packages/shared/src/agentSlots.test.ts @@ -6,12 +6,14 @@ import { DISPATCH_SAFETY_MS, effectiveSlotLimits, limitsCount, + liveReservations, liveStarting, runningAgents, slotDecision, slotRefusalText, type SlotDecision, type SlotLimits, + type SlotReservation, type SlotThread, type StartingEntry, visibleRunThreadIds, @@ -243,6 +245,56 @@ describe("slotDecision (T13)", () => { }); }); +describe("slot reservations (4.6)", () => { + const held = (key: string, untilMs: number, account?: string): SlotReservation => ({ + key, + threadId: `holder-${key}`, + untilMs, + ...(account === undefined ? {} : { account }), + }); + + it("keeps live ones, one per key with the latest end, oldest end first", () => { + expect( + liveReservations( + [held("w_b", 300), held("w_a", 200), held("w_a", 400), held("w_old", 100)], + 100, + ).map((reservation) => [reservation.key, reservation.untilMs]), + ).toEqual([ + ["w_b", 300], + ["w_a", 400], + ]); + expect(liveReservations([], 0)).toEqual([]); + }); + + it("count on the machine, and on an account only when they name it", () => { + const base = { + limits: limitsOf({ maxPerMachine: 3, maxPerAccount: { [MAIN]: 1 } }), + running: [{ threadId: "t", instanceId: SECOND }], + starting: [], + freeMemoryBytes: null, + }; + // One running and two held: the machine is full. + expect( + refused( + slotDecision({ + ...base, + reservations: [held("w_1", 1), held("w_2", 1)], + targetInstanceId: SECOND, + }), + ), + ).toMatchObject({ rule: "machine", used: 3 }); + // A slot held without an account does not count on MAIN's own limit; one held on MAIN does. + expect( + slotDecision({ ...base, reservations: [held("w_1", 1)], targetInstanceId: MAIN }), + ).toEqual({ type: "allow" }); + expect( + refused( + slotDecision({ ...base, reservations: [held("w_1", 1, MAIN)], targetInstanceId: MAIN }), + ), + ).toMatchObject({ rule: "account", used: 1, max: 1 }); + }); +}); + describe("slotRefusalText", () => { const refusal = (rule: "machine" | "account" | "memory", used: number, max: number) => ({ type: "refuse", rule, used, max, instanceId: MAIN }) as const; diff --git a/packages/shared/src/agentSlots.ts b/packages/shared/src/agentSlots.ts index cc36e43d80d67..e9ec96dc67263 100644 --- a/packages/shared/src/agentSlots.ts +++ b/packages/shared/src/agentSlots.ts @@ -56,12 +56,12 @@ export interface RunningAgent { } /** - * Threads with an active run, less the caller and less threads blocked in a thread wait (a - * parent inside `t3_thread_wait` holds no model turn of its own while it waits). + * Threads with an active run, less the caller (`null`: nobody) and less threads blocked in a + * thread wait (a parent inside `t3_thread_wait` holds no model turn of its own while it waits). */ export function runningAgents( threads: ReadonlyArray, - callerThreadId: string, + callerThreadId: string | null, ): ReadonlyArray { const seen = new Set(); const running: Array = []; @@ -120,6 +120,34 @@ export function liveStarting( }); } +/** + * A slot held for a thread whose slot wait was met (gate memory, or a delivered wait's + * `reservedUntil`; `key` is the wait id, so one wait counts once). Counts as a running agent on the + * machine and its `account`; the thread's next allowed delegation uses it up. + */ +export interface SlotReservation { + readonly key: string; + readonly threadId: string; + readonly account?: string; + readonly untilMs: number; +} + +/** Reservations still live at `nowMs`, one per key (the latest end wins), oldest end first. */ +export function liveReservations( + reservations: ReadonlyArray, + nowMs: number, +): ReadonlyArray { + const byKey = new Map(); + for (const reservation of reservations) { + if (reservation.untilMs <= nowMs) continue; + const seen = byKey.get(reservation.key); + if (seen === undefined || seen.untilMs < reservation.untilMs) { + byKey.set(reservation.key, reservation); + } + } + return [...byKey.values()].sort((a, b) => a.untilMs - b.untilMs || (a.key < b.key ? -1 : 1)); +} + export type SlotRule = "machine" | "account" | "memory"; export type SlotDecision = @@ -139,6 +167,8 @@ export interface SlotDecisionInput { readonly running: ReadonlyArray; /** Live entries only (`liveStarting`). */ readonly starting: ReadonlyArray; + /** Live reservations that count (`liveReservations`), without the caller's own. */ + readonly reservations?: ReadonlyArray; readonly targetInstanceId: string; /** `null` when the memory floor is off. */ readonly freeMemoryBytes: number | null; @@ -147,6 +177,7 @@ export interface SlotDecisionInput { /** Machine limit first, then the target's account, then the memory floor. */ export function slotDecision(input: SlotDecisionInput): SlotDecision { const { limits, running, starting, targetInstanceId } = input; + const reservations = input.reservations ?? []; const refuse = (rule: SlotRule, used: number, max: number): SlotDecision => ({ type: "refuse", rule, @@ -154,7 +185,7 @@ export function slotDecision(input: SlotDecisionInput): SlotDecision { max, instanceId: targetInstanceId, }); - const machineUsed = running.length + starting.length; + const machineUsed = running.length + starting.length + reservations.length; if (limits.machine !== null && machineUsed >= limits.machine) { return refuse("machine", machineUsed, limits.machine); } @@ -162,7 +193,8 @@ export function slotDecision(input: SlotDecisionInput): SlotDecision { if (accountMax !== null) { const accountUsed = running.filter((agent) => agent.instanceId === targetInstanceId).length + - starting.filter((entry) => entry.instanceId === targetInstanceId).length; + starting.filter((entry) => entry.instanceId === targetInstanceId).length + + reservations.filter((reservation) => reservation.account === targetInstanceId).length; if (accountUsed >= accountMax) return refuse("account", accountUsed, accountMax); } if ( @@ -179,6 +211,18 @@ export function slotDecision(input: SlotDecisionInput): SlotDecision { return { type: "allow" }; } +/** Why a slot is not free, without a full stop (held slots count as running agents). */ +export function slotRefusalReason( + decision: Extract, +): string { + const agents = (count: number) => (count === 1 ? "1 agent is" : `${count} agents are`); + return decision.rule === "machine" + ? `${agents(decision.used)} running on this machine, and the limit is ${decision.max}` + : decision.rule === "account" + ? `${agents(decision.used)} running on account ${decision.instanceId}, and the limit is ${decision.max}` + : `only ${decision.used} MB of memory is free, and a new agent needs at least ${decision.max} MB`; +} + /** * The `slots_full` message. `waitsEnabled`: a slot wait can be registered (the `slot` kind of * `wait_for`), so the text names the call; else it says to try again later. @@ -187,13 +231,7 @@ export function slotRefusalText( decision: Extract, waitsEnabled: boolean, ): string { - const agents = (count: number) => (count === 1 ? "1 agent is" : `${count} agents are`); - const reason = - decision.rule === "machine" - ? `${agents(decision.used)} running on this machine, and the limit is ${decision.max}` - : decision.rule === "account" - ? `${agents(decision.used)} running on account ${decision.instanceId}, and the limit is ${decision.max}` - : `only ${decision.used} MB of memory is free, and a new agent needs at least ${decision.max} MB`; + const reason = slotRefusalReason(decision); const slot = decision.rule === "account" ? `{"account":${JSON.stringify(decision.instanceId)}}` : "{}"; const next = waitsEnabled diff --git a/packages/shared/src/waitFor.test.ts b/packages/shared/src/waitFor.test.ts index 68c3f0b5757ba..dc4115ab661d4 100644 --- a/packages/shared/src/waitFor.test.ts +++ b/packages/shared/src/waitFor.test.ts @@ -709,6 +709,50 @@ describe("nextWaitState: judging", () => { shouldPruneWait({ ...sent, reservedUntil: iso(T0 + 30 * HOUR) }, T0 + 25 * HOUR, settings), ).toBe(false); }); + + it("holds a slot only for a granted slot wait, and drops it once used (rows 9, 15)", () => { + const slot = makeWait({ kind: "slot", spec: {} }); + // Timed out, or handed over before a grant: delivered, but nothing is held. + const timedOut = nextWaitState( + { ...slot, deadlineAt: iso(T0) }, + { type: "judged", judgement: { type: "notYet" } }, + T0, + stateSettings, + )!.record; + expect(timedOut.result?.outcome).toBe("timeout"); + const timedOutSent = nextWaitState(timedOut, { type: "delivered" }, T0, stateSettings)!.record; + expect(timedOutSent.state).toBe("delivered"); + expect(timedOutSent.reservedUntil).toBeUndefined(); + // `reserveSeconds: 0` holds nothing either. + const granted = nextWaitState( + slot, + { type: "slotGranted", detail: "" }, + T0, + stateSettings, + )!.record; + expect( + nextWaitState(granted, { type: "delivered" }, T0, { ...stateSettings, reserveSeconds: 0 })! + .record.reservedUntil, + ).toBeUndefined(); + const sent = nextWaitState(granted, { type: "delivered" }, T0, stateSettings)!.record; + expect(nextWaitState(granted, { type: "slotUsed" }, T0, stateSettings)).toBeNull(); + const used = nextWaitState(sent, { type: "slotUsed" }, T0 + SECOND, stateSettings)!.record; + expect(used).toMatchObject({ state: "delivered", endedAt: sent.endedAt }); + expect(used.reservedUntil).toBeUndefined(); + // Used again (a second delegation): nothing changes and nothing is written. + expect(nextWaitState(used, { type: "slotUsed" }, T0 + SECOND, stateSettings)).toBeNull(); + // Granted, then its thread was archived: the parent gets the message, but no slot is held. + const handed = nextWaitState( + { ...granted, parentThreadId: ThreadId.make("thread:parent") }, + { type: "threadGone", parentThreadId: null }, + T0, + stateSettings, + )!.record; + expect(handed).toMatchObject({ state: "delivering", threadId: "thread:parent" }); + const handedSent = nextWaitState(handed, { type: "delivered" }, T0, stateSettings)!.record; + expect(handedSent.state).toBe("delivered"); + expect(handedSent.reservedUntil).toBeUndefined(); + }); }); describe("nextWaitState: delivery", () => { diff --git a/packages/shared/src/waitFor.ts b/packages/shared/src/waitFor.ts index 2c47a1e28e85c..00834732c2071 100644 --- a/packages/shared/src/waitFor.ts +++ b/packages/shared/src/waitFor.ts @@ -469,6 +469,8 @@ export type WaitEvent = | { readonly type: "judged"; readonly judgement: Judgement } /** The sweep gave a `waiting` slot wait a free slot. */ | { readonly type: "slotGranted"; readonly detail: string } + /** The slot gate let the wait's thread start an agent: its reservation is used up. */ + | { readonly type: "slotUsed" } | { readonly type: "cancel" } /** * Written before the first send to the current receiver. `replaces` names an id the wait has @@ -654,6 +656,10 @@ export function nextWaitState( return wait.state === "waiting" && wait.kind === "slot" ? { record: freeze(wait, "free", event.detail, now) } : null; + case "slotUsed": + return wait.state === "delivered" && wait.reservedUntil !== undefined + ? { record: update(wait, {}, ["reservedUntil"]) } + : null; case "cancel": return wait.state === "waiting" ? { record: update(wait, { state: "cancelled", endedAt: iso(now) }) } @@ -666,8 +672,12 @@ export function nextWaitState( : null; case "delivered": { if (wait.state !== "delivering") return null; + // Only a granted slot is held; a slot wait that timed out or was handed over holds none. const reservation = - wait.kind === "slot" + wait.kind === "slot" && + wait.result?.outcome === "free" && + wait.handedOverBecause === undefined && + settings.reserveSeconds > 0 ? { reservedUntil: iso(now + settings.reserveSeconds * MS_PER_SECOND) } : {}; return { From 997e16aba6bc719d5ca8e8e059b73c1b25d3db74 Mon Sep 17 00:00:00 2001 From: scuttletoby <123843277+scuttletoby@users.noreply.github.com> Date: Sun, 15 Nov 2026 01:31:26 +0000 Subject: [PATCH 2/3] Slot waits: one held slot per thread, wakes within free slots, bounded stored holds - A thread holds at most one slot. A repeat ask (met at register or granted in a sweep) replaces its earlier hold instead of adding one; counts and the usage view count one hold per thread (onePerThread). - Grants in one sweep count against the free slots even with reserveSeconds 0, so one free slot wakes one waiter; a second wait of a granted thread asks again next time. - A stored reservedUntil is capped at endedAt/now + reserveSeconds on load and on read (clock ahead or hand-edited file), so an expired record is pruned. - A granted wait handed over to the parent, or ended without a receiver, frees its slot at once; a slot used before its wake is delivered is not held again by the delivery. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/slots/SlotGate.test.ts | 136 +++++++++++ apps/server/src/slots/SlotGate.ts | 119 +++++++--- .../src/waits/WaitService.slots.test.ts | 221 ++++++++++++++++++ apps/server/src/waits/WaitService.ts | 72 +++++- packages/shared/src/agentSlots.test.ts | 19 ++ packages/shared/src/agentSlots.ts | 21 ++ packages/shared/src/waitFor.ts | 17 ++ 7 files changed, 561 insertions(+), 44 deletions(-) diff --git a/apps/server/src/slots/SlotGate.test.ts b/apps/server/src/slots/SlotGate.test.ts index 25c5386d9eb9c..bfee646682b0f 100644 --- a/apps/server/src/slots/SlotGate.test.ts +++ b/apps/server/src/slots/SlotGate.test.ts @@ -800,5 +800,141 @@ describe("SlotGate", () => { }); }), ); + + // R17: "as many wakes as free slots". reserveSeconds 0 is a valid setting. + it.effect("reserveSeconds 0: one free slot wakes one wait, not every wait", () => + Effect.gen(function* () { + const world = newWorld(); + const grants = yield* run( + world, + waitsOn({ maxPerMachine: 1, reserveSeconds: 0 }), + grant( + { key: "w1", threadId: holder }, + { key: "w2", threadId: other }, + { key: "w3", threadId: caller }, + ), + ); + assert.deepStrictEqual(types(grants), ["granted", "full", "full"]); + }), + ); + + it.effect("one held slot per thread: asking again moves and renews it, never adds one", () => + Effect.gen(function* () { + const world = newWorld(); + yield* run( + world, + waitsOn({ maxPerMachine: 3, reserveSeconds: 60 }), + Effect.gen(function* () { + const gate = yield* SlotGate.SlotGate; + // Asked three times (three slots free): still one slot held. + for (const key of ["w1", "w2", "w3"]) { + assert.deepStrictEqual(types(yield* grant({ key, threadId: holder })), ["granted"]); + yield* TestClock.adjust("20 seconds"); + } + assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 1, max: 3 }); + assert.strictEqual((yield* acquireAs(other)).type, "allow"); + assert.strictEqual((yield* acquireAs(other)).type, "allow"); + assert.strictEqual((yield* acquireAs(other)).type, "refuse"); + // Renewed by the last ask (at 40 s): still held at 90 s, gone after 100 s. + yield* TestClock.adjust("30 seconds"); + assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 3, max: 3 }); + yield* TestClock.adjust("15 seconds"); + assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 2, max: 3 }); + }), + ); + }), + ); + + it.effect("two slot waits of one thread in one sweep: one wake, the other asks next time", () => + Effect.gen(function* () { + const world = newWorld(); + yield* run( + world, + waitsOn({ maxPerMachine: 3 }), + Effect.gen(function* () { + assert.deepStrictEqual( + types( + yield* grant( + { key: "w1", threadId: holder, account: MAIN }, + { key: "w2", threadId: holder, account: SECOND }, + { key: "w3", threadId: other }, + ), + ), + ["granted", "sameThread", "granted"], + ); + assert.deepStrictEqual((yield* (yield* SlotGate.SlotGate).usage)?.machine, { + used: 2, + max: 3, + }); + }), + ); + }), + ); + + it.effect( + "two stored held slots of one thread (a restart): counted once, both replaced by a new ask", + () => + Effect.gen(function* () { + const world = newWorld(); + const used: Array = []; + let inStore: ReadonlyArray = [ + { key: "stored-1", threadId: holder, untilMs: 50_000 }, + // A restart lost the gate's note that this older copy was replaced. + { key: "stored-0", threadId: holder, untilMs: 40_000 }, + ]; + yield* run( + world, + waitsOn({ maxPerMachine: 2, reserveSeconds: 60 }), + Effect.gen(function* () { + const gate = yield* SlotGate.SlotGate; + yield* gate.attachReservations({ + read: Effect.sync(() => inStore), + used: (key) => + Effect.sync(() => { + used.push(key); + inStore = inStore.filter((entry) => entry.key !== key); + }), + }); + assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 1, max: 2 }); + // Asked again: granted, and both stored copies are cleared (replaced). + assert.deepStrictEqual(types(yield* grant({ key: "w1", threadId: holder })), [ + "granted", + ]); + assert.deepStrictEqual([...used].sort(), ["stored-0", "stored-1"]); + assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 1, max: 2 }); + assert.strictEqual((yield* acquireAs(other)).type, "allow"); + // Its agent starts: the one held slot is used up and the store asked to clear it. + assert.strictEqual((yield* acquireAs(holder)).type, "allow"); + assert.deepStrictEqual(used.slice(2), ["w1"]); + assert.strictEqual((yield* acquireAs(caller)).type, "refuse"); + }), + ); + }), + ); + + it.effect("a thread already holding a slot that is refused keeps the slot it holds", () => + Effect.gen(function* () { + const world = newWorld(); + yield* run( + world, + waitsOn({ maxPerMachine: 2, maxPerAccount: { [SECOND]: 1 } }), + Effect.gen(function* () { + yield* grant({ key: "w1", threadId: other, account: SECOND }); + assert.deepStrictEqual(types(yield* grant({ key: "w2", threadId: holder })), [ + "granted", + ]); + // The account is full: refused, and its held machine slot stays. + assert.deepStrictEqual( + types(yield* grant({ key: "w3", threadId: holder, account: SECOND })), + ["full"], + ); + assert.deepStrictEqual((yield* (yield* SlotGate.SlotGate).usage)?.machine, { + used: 2, + max: 2, + }); + }), + ); + }), + ); }); }); diff --git a/apps/server/src/slots/SlotGate.ts b/apps/server/src/slots/SlotGate.ts index 722d5e10f99f8..42de654f9cbc0 100644 --- a/apps/server/src/slots/SlotGate.ts +++ b/apps/server/src/slots/SlotGate.ts @@ -10,6 +10,7 @@ import { limitsCount, liveReservations, liveStarting, + onePerThread, runningAgents, slotDecision, slotRefusalText, @@ -59,8 +60,13 @@ export interface SlotRequest { export type SlotGrant = /** Slots are off or have no limit: nothing to wait for, nothing held. */ | { readonly type: "off" } - /** Free: held for the thread until `untilMs` (memory only). */ + /** + * Free: held for the thread until `untilMs` (memory only). A thread holds at most one slot: a + * slot it already held moves to this request (its end renewed), it is never added to. + */ | { readonly type: "granted"; readonly untilMs: number } + /** An earlier request of the same thread was granted in this call: asked again next time. */ + | { readonly type: "sameThread" } | { readonly type: "full"; readonly decision: Extract } /** The running agents cannot be counted now: never "free". */ | { readonly type: "unread"; readonly message: string }; @@ -169,8 +175,9 @@ const make = Effect.gen(function* () { ); /** - * Every live held slot: granted in memory and kept in the wait store, one per wait. None while - * waits are off (nothing of the store is read then). A store that cannot be read holds none. + * Every live held slot: granted in memory and kept in the wait store, one per wait (counted one + * per thread, `onePerThread`). None while waits are off (nothing of the store is read then). A + * store that cannot be read holds none. */ const liveHeld = Effect.fn("SlotGate.liveHeld")(function* (waitsOn: boolean, nowMs: number) { for (const [key, untilMs] of usedKeys) if (untilMs <= nowMs) usedKeys.delete(key); @@ -190,6 +197,33 @@ const make = Effect.gen(function* () { /** Settings that cannot be read count as off, as for the usage pacing gate. */ const readSettings = serverSettings.getSettings.pipe(Effect.option); + /** + * Every slot held for `threadId` stops counting (memory ones go; stored ones are skipped until + * their end). Returns their keys, for the store to clear. Under the semaphore. + */ + const useUp = Effect.fnUntraced(function* ( + reservations: ReadonlyArray, + threadId: string, + ) { + const own = reservations.filter((reservation) => reservation.threadId === threadId); + if (own.length === 0) return []; + for (const reservation of own) usedKeys.set(reservation.key, reservation.untilMs); + yield* Ref.update(held, (current) => current.filter((entry) => entry.threadId !== threadId)); + return own.map((reservation) => reservation.key); + }); + + /** Clears used or replaced held slots in the store, outside the semaphore (see `acquire`). */ + const clearStored = (keys: ReadonlyArray) => + stored === null || keys.length === 0 + ? Effect.void + : Effect.forEach(keys, stored.used, { discard: true }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("agent slots: a used slot could not be cleared in the wait store", { + cause, + }), + ), + ); + /** * The decision, under the semaphore, and the caller's held slot it used up (if any). The * caller's own held slots never count against it; they are not a pass past a limit either. @@ -204,7 +238,10 @@ const make = Effect.gen(function* () { ? null : effectiveSlotLimits(settings.value.agentSlots); if (Option.isNone(settings) || limits === null) { - return { outcome: { type: "allow", token: null } satisfies SlotAcquire, used: null }; + return { + outcome: { type: "allow", token: null } satisfies SlotAcquire, + used: [] as ReadonlyArray, + }; } const waitsOn = settings.value.waitFor.enabled; const nowMs = yield* Clock.currentTimeMillis; @@ -217,8 +254,8 @@ const make = Effect.gen(function* () { limits: running === undefined ? { ...limits, machine: null, account: () => null } : limits, running: running ?? [], starting: kept, - reservations: reservations.filter( - (reservation) => reservation.threadId !== input.callerThreadId, + reservations: onePerThread( + reservations.filter((reservation) => reservation.threadId !== input.callerThreadId), ), targetInstanceId: input.targetInstanceId, freeMemoryBytes: yield* freeMemoryBytes(limits), @@ -229,7 +266,7 @@ const make = Effect.gen(function* () { type: "refuse", message: slotRefusalText(decision, waitsOn), } satisfies SlotAcquire, - used: null, + used: [] as ReadonlyArray, }; } const token = yield* Ref.getAndUpdate(nextToken, (value) => value + 1); @@ -237,13 +274,9 @@ const make = Effect.gen(function* () { ...current, { token, instanceId: input.targetInstanceId, atMs: nowMs }, ]); - // The agent it held a slot for starts now: the slot ending first is used up. - const own = reservations.find((reservation) => reservation.threadId === input.callerThreadId); - if (own !== undefined) { - usedKeys.set(own.key, own.untilMs); - yield* Ref.update(held, (current) => current.filter((entry) => entry.key !== own.key)); - } - return { outcome: { type: "allow", token } satisfies SlotAcquire, used: own?.key ?? null }; + // The agent it held a slot for starts now: its held slot is used up, with every older copy. + const used = yield* useUp(reservations, input.callerThreadId); + return { outcome: { type: "allow", token } satisfies SlotAcquire, used }; }, // One at a time: the next caller sees the entry the last one added. semaphore.withPermits(1), @@ -256,28 +289,20 @@ const make = Effect.gen(function* () { const { outcome, used } = yield* decide(input); // Cleared in the store after the decision, outside the semaphore; a failed write only means // the stored copy ends by itself (it no longer counts here either way). - if (used !== null && stored !== null) { - yield* stored.used(used).pipe( - Effect.catchCause((cause) => - Effect.logWarning("agent slots: a used slot could not be cleared in the wait store", { - cause, - }), - ), - ); - } + yield* clearStored(used); return outcome; }); - const grant = Effect.fn("SlotGate.grant")( + const grantLocked = Effect.fn("SlotGate.grantLocked")( function* (requests: ReadonlyArray) { const settings = yield* readSettings; const limits = Option.isNone(settings) ? null : effectiveSlotLimits(settings.value.agentSlots); if (Option.isNone(settings) || limits === null) { - return requests.map((): SlotGrant => ({ type: "off" })); + return { grants: requests.map((): SlotGrant => ({ type: "off" })), replaced: [] }; } - if (requests.length === 0) return []; + if (requests.length === 0) return { grants: [], replaced: [] }; const nowMs = yield* Clock.currentTimeMillis; /** A wait without an account waits for the machine limit and the memory floor only. */ const limitsFor = (request: SlotRequest): SlotLimits => @@ -290,10 +315,16 @@ const make = Effect.gen(function* () { const kept = yield* keepStarting(read?.visible, nowMs, limits); const memoryBytes = yield* freeMemoryBytes(limits); const reserveMs = settings.value.agentSlots.reserveSeconds * 1000; - // Slots held for the requesting thread count too: it never holds two for one free slot. - const reservations = [...(yield* liveHeld(settings.value.waitFor.enabled, nowMs))]; + let reservations = yield* liveHeld(settings.value.waitFor.enabled, nowMs); const grants: Array = []; + const replaced: Array = []; + const grantedThreads = new Set(); for (const request of requests) { + // One wake per thread in a call: its other slot waits ask again next time. + if (grantedThreads.has(request.threadId)) { + grants.push({ type: "sameThread" }); + continue; + } const own = limitsFor(request); if (read === undefined && limitsCount(own, request.account ?? "")) { grants.push({ type: "unread", message: "The running agents cannot be counted now." }); @@ -303,11 +334,15 @@ const make = Effect.gen(function* () { limits: own, running: (read?.running ?? []).filter((agent) => agent.threadId !== request.threadId), starting: kept, - reservations, + // Its own held slot is not counted: a grant moves it here, so the thread keeps one. + reservations: onePerThread( + reservations.filter((reservation) => reservation.threadId !== request.threadId), + ), targetInstanceId: request.account ?? "", freeMemoryBytes: memoryBytes, }); if (decision.type === "refuse") { + // Refused: a slot it already holds stays as it is. grants.push({ type: "full", decision }); continue; } @@ -318,19 +353,31 @@ const make = Effect.gen(function* () { ...(request.account === undefined ? {} : { account: request.account }), untilMs, }; - // A slot held for 0 s is not held at all; the wait is still woken. - if (reserveMs > 0) { - reservations.push(reservation); - yield* Ref.update(held, (current) => [...current, reservation]); - } + // The slot it held before (memory or store) is replaced by this one. + replaced.push(...(yield* useUp(reservations, request.threadId))); + // Counted for the rest of this call even when held 0 s: one free slot, one wake (R17). + reservations = [ + ...reservations.filter((entry) => entry.threadId !== request.threadId), + reservation, + ]; + // A slot held for 0 s is not held after this call; the wait is still woken. + if (reserveMs > 0) yield* Ref.update(held, (current) => [...current, reservation]); + grantedThreads.add(request.threadId); grants.push({ type: "granted", untilMs }); } - return grants; + return { grants, replaced }; }, // Under the same semaphore as `acquire`: a delegation never takes a slot being granted. semaphore.withPermits(1), ); + const grant = Effect.fn("SlotGate.grant")(function* (requests: ReadonlyArray) { + const { grants, replaced } = yield* grantLocked(requests); + // A replaced slot in the store is cleared like a used one, outside the semaphore. + yield* clearStored(replaced); + return grants; + }); + const usage = Effect.gen(function* () { const settings = yield* readSettings; if (Option.isNone(settings) || !settings.value.agentSlots.enabled) return null; @@ -343,7 +390,7 @@ const make = Effect.gen(function* () { limits === null ? [] : liveStarting(yield* Ref.get(entries), read.visible, nowMs, limits.startGraceMs); - const reservations = yield* liveHeld(settings.value.waitFor.enabled, nowMs); + const reservations = onePerThread(yield* liveHeld(settings.value.waitFor.enabled, nowMs)); const accountMax = (id: string) => { const max = agentSlots.maxPerAccount[id as ProviderInstanceId]; return max === undefined || max <= 0 ? null : max; diff --git a/apps/server/src/waits/WaitService.slots.test.ts b/apps/server/src/waits/WaitService.slots.test.ts index 12499c5707355..f80f4062cb911 100644 --- a/apps/server/src/waits/WaitService.slots.test.ts +++ b/apps/server/src/waits/WaitService.slots.test.ts @@ -564,4 +564,225 @@ describe("WaitService slot waits", () => { }).pipe(Effect.provide(serverLayer(world, slotsOn(1)))); }), ); + + it.effect( + "one thread holds at most one slot: repeated met slot waits do not hoard every free slot", + () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, delegate, service } = yield* startService(store.fs, send.delivery); + // Nothing runs; three slots free. T1 asks three times and starts nothing. + for (let i = 0; i < 3; i += 1) { + const result = yield* register(T1); + assert.strictEqual(result.state, "met"); + } + // Nothing is stored (so maxPerThread never applied), and one slot is held. + assert.deepStrictEqual(stored(store.files), []); + const listed = yield* service.list({ threadId: T1, projectId: PROJECT }); + assert.deepStrictEqual(listed.slots?.machine, { used: 1, max: 3 }); + // Another thread, with no agent running anywhere, still finds a slot. + assert.strictEqual(yield* delegate(OUTSIDER), "allow"); + assert.strictEqual(yield* delegate(OUTSIDER), "allow"); + assert.strictEqual(yield* delegate(OUTSIDER), "refuse"); + // T1 still has its one slot. + assert.strictEqual(yield* delegate(T1), "allow"); + }).pipe(Effect.provide(serverLayer(world, slotsOn(3)))); + }), + ); + + it.effect("a woken thread that asks again keeps one slot: the stored one is replaced", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.running = ["busy-1", "busy-2", "busy-3"]; + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, delegate, service } = yield* startService(store.fs, send.delivery); + const id = waitingId(yield* register(T1)); + world.running = ["busy-1"]; + yield* service.runDue(); + assert.isDefined(stored(store.files).find((wait) => wait.id === id)?.reservedUntil); + // Asked again before using it: met, and the slot it held moves to this answer. + assert.strictEqual((yield* register(T1)).state, "met"); + assert.isUndefined(stored(store.files).find((wait) => wait.id === id)?.reservedUntil); + const listed = yield* service.list({ threadId: T1, projectId: PROJECT }); + assert.deepStrictEqual(listed.slots?.machine, { used: 2, max: 3 }); + assert.strictEqual(yield* delegate(OUTSIDER), "allow"); + assert.strictEqual(yield* delegate(T1), "allow"); + }).pipe(Effect.provide(serverLayer(world, slotsOn(3)))); + }), + ); + + it.effect( + "reserveSeconds 0: one free slot wakes one waiter per sweep, not all of them (R17)", + () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.running = ["busy-1"]; + const store = disk(); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { register, service } = yield* startService(store.fs, send.delivery); + for (const thread of [T1, T2, T3]) { + waitingId(yield* register(thread)); + yield* TestClock.adjust(1_000); + } + world.running = []; + yield* service.runDue(); + assert.deepStrictEqual( + send.sent.map((message) => message.threadId), + [T1], + ); + }).pipe( + Effect.provide(serverLayer(world, slotsOn(1, { agentSlots: { reserveSeconds: 0 } }))), + ); + }), + ); + + it.effect( + "a stored held slot far ahead (clock ahead, hand-edited file) ends reserveSeconds after it is read, and is pruned", + () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + const store = disk(); + const heldWait = ( + id: string, + threadId: ThreadId, + endedAt: string, + reservedUntil: string, + ) => ({ + id, + threadId, + projectId: PROJECT, + key: id === "w_aaaaaaaaaa" ? "000000000000000a" : "000000000000000b", // gitleaks:allow (fake test values) + createdAt: "2026-11-14T11:00:00.000Z", + deadlineAt: "2026-11-14T13:00:00.000Z", + state: "delivered", + endedAt, + attempt: 0, + kind: "slot", + spec: {}, + result: { outcome: "free", detail: "", at: endedAt }, + reservedUntil, + }); + store.files.set( + FILE, + JSON.stringify({ + version: 1, + waits: [ + // Delivered 10 s ago, but held until the year 9999. + heldWait("w_aaaaaaaaaa", T1, "2026-11-14T11:59:50.000Z", "9999-01-01T00:00:00.000Z"), + // Written while the clock was a year ahead. + heldWait("w_bbbbbbbbbb", T2, "2027-11-14T12:00:00.000Z", "2027-11-14T12:01:00.000Z"), + ], + }), + ); + const send = fakeDelivery(); + yield* Effect.gen(function* () { + const { service } = yield* startService(store.fs, send.delivery); + const used = Effect.map( + service.list({ threadId: OUTSIDER, projectId: PROJECT }), + (listed) => listed.slots?.machine.used, + ); + assert.strictEqual(yield* used, 2); + // 60 s after its delivery the first one ends; 60 s after the read the second. + yield* TestClock.adjust(51_000); + assert.strictEqual(yield* used, 1); + yield* TestClock.adjust(10_000); + assert.strictEqual(yield* used, 0); + // Ended waits are pruned after keepEndedHours; the held slot no longer keeps them. + yield* TestClock.adjust(3_600_000); + yield* service.runDue(); + assert.deepStrictEqual( + stored(store.files).map((wait) => wait.id), + ["w_bbbbbbbbbb"], + ); + assert.isAtMost( + Date.parse(stored(store.files)[0]!.reservedUntil ?? "1970-01-01T00:00:00.000Z"), + NOW + 60_000, + ); + }).pipe(Effect.provide(serverLayer(world, slotsOn(2, { waitFor: { keepEndedHours: 1 } })))); + }), + ); + + it.effect("a granted wait handed over to the parent frees its slot at once", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.running = ["busy-1"]; + const store = disk(); + const send = fakeDelivery(); + // Archived after the slot is granted: it looks alive until its wake is sent. + let archived = false; + const delivery: WaitDelivery = { + receiver: (threadId) => + threadId === T1 && archived + ? Effect.succeed({ type: "gone", parentThreadId: PARENT }) + : send.delivery.receiver(threadId), + deliver: (message) => { + if (message.threadId !== T1) return send.delivery.deliver(message); + archived = true; + return Effect.succeed({ type: "gone", parentThreadId: PARENT }); + }, + }; + yield* Effect.gen(function* () { + const { register, delegate, service } = yield* startService(store.fs, delivery); + waitingId(yield* register(T1)); + world.running = []; + yield* service.runDue(); + yield* service.runDue(); + assert.isTrue(archived); + assert.deepStrictEqual( + send.sent.map((message) => message.threadId), + [PARENT], + ); + assert.isUndefined(stored(store.files)[0]?.reservedUntil); + assert.strictEqual(yield* delegate(OUTSIDER), "allow"); + }).pipe(Effect.provide(serverLayer(world, slotsOn(1)))); + }), + ); + + it.effect("a slot used before its wake is delivered is not held again by the delivery", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const world = newWorld(); + world.running = ["busy-1", "busy-2", "busy-3"]; + const store = disk(); + const send = fakeDelivery(); + let gate: SlotGate.SlotGate["Service"] | null = null; + const allowed: Array = []; + const delivery: WaitDelivery = { + receiver: send.delivery.receiver, + // The thread delegates (on its held slot) while its wake is on the way. + deliver: (message) => + gate!.acquire({ callerThreadId: message.threadId, targetInstanceId: MAIN }).pipe( + Effect.tap((outcome) => Effect.sync(() => allowed.push(outcome.type))), + Effect.andThen(send.delivery.deliver(message)), + ), + }; + yield* Effect.gen(function* () { + const started = yield* startService(store.fs, delivery); + gate = started.gate; + waitingId(yield* started.register(T1)); + world.running = ["busy-1"]; + yield* started.service.runDue(); + assert.deepStrictEqual(allowed, ["allow"]); + assert.deepStrictEqual( + send.sent.map((message) => message.threadId), + [T1], + ); + assert.strictEqual(stored(store.files)[0]?.state, "delivered"); + assert.isUndefined(stored(store.files)[0]?.reservedUntil); + // busy-1 and T1's starting agent (2 of 3): the used slot does not count again. + assert.strictEqual(yield* started.delegate(OUTSIDER), "allow"); + }).pipe(Effect.provide(serverLayer(world, slotsOn(3)))); + }), + ); }); diff --git a/apps/server/src/waits/WaitService.ts b/apps/server/src/waits/WaitService.ts index ad43918e1779e..1a097482e96c8 100644 --- a/apps/server/src/waits/WaitService.ts +++ b/apps/server/src/waits/WaitService.ts @@ -28,6 +28,7 @@ import { } from "@t3tools/shared/agentSlots"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { + boundHeldSlot, capEndedWaits, deliveryCommandId, type DeliveryGroup, @@ -252,17 +253,27 @@ const slotJudgement = ( }; case "full": return { type: "notYet", detail: `No free slot yet: ${slotRefusalReason(grant.decision)}.` }; + case "sameThread": + return { type: "notYet", detail: "This thread's other slot wait was just given a slot." }; case "unread": return { type: "failed", message: grant.message, permanent: false }; } }; -/** The slot a delivered slot wait still holds, from its `reservedUntil`. */ -const storedReservation = (wait: WaitRecord, now: number): ReadonlyArray => { +/** + * The slot a delivered slot wait still holds, from its `reservedUntil`: never longer than + * `reserveSeconds` (as set now) after its delivery, so a lowered setting shortens it too. + */ +const storedReservation = ( + wait: WaitRecord, + now: number, + reserveSeconds: number, +): ReadonlyArray => { if (wait.kind !== "slot" || wait.state !== "delivered" || wait.reservedUntil === undefined) { return []; } - const untilMs = Date.parse(wait.reservedUntil); + const endedMs = wait.endedAt === undefined ? Number.NaN : Date.parse(wait.endedAt); + const untilMs = Math.min(Date.parse(wait.reservedUntil), endedMs + reserveSeconds * 1000); if (!(untilMs > now)) return []; const account = wait.spec.account; return [ @@ -418,6 +429,11 @@ export const make = Effect.fn("WaitService.make")(function* (input: { const runProject = new Map(); /** Ids that registrations in progress will store their waits under. */ const reserved = new Set(); + /** + * Granted slot waits whose thread started its agent before the wake was delivered (memory only; + * after a restart such a slot counts until it ends, the safe side). + */ + const usedBeforeDelivery = new Set(); /** Checks that have not returned, per place; late once past their own `lateAt`. */ const running = new Map>(); /** Places whose checks are all stuck, already logged; cleared when one of them returns. */ @@ -572,7 +588,13 @@ export const make = Effect.fn("WaitService.make")(function* (input: { case "Loaded": break; } - current = { waits: load.waits, notices: load.notices }; + // Held slots are bounded as they are read (a clock that was ahead, a hand-edited file). + const settings = yield* readSettingsOrNull; + const reserveSeconds = settings?.agentSlots.reserveSeconds ?? 0; + current = { + waits: load.waits.map((wait) => boundHeldSlot(wait, nowMs, reserveSeconds)), + notices: load.notices, + }; lastProblem = null; return { ok: true, contents: current } satisfies Loaded; }); @@ -642,8 +664,12 @@ export const make = Effect.fn("WaitService.make")(function* (input: { Effect.gen(function* () { const loaded = yield* ensureLoaded(true); if (!loaded.ok) return []; + const settings = yield* readSettingsOrNull; + if (settings === null) return []; const now = yield* Clock.currentTimeMillis; - return loaded.contents.waits.flatMap((wait) => storedReservation(wait, now)); + return loaded.contents.waits.flatMap((wait) => + storedReservation(wait, now, settings.agentSlots.reserveSeconds), + ); }), ); @@ -657,6 +683,11 @@ export const make = Effect.fn("WaitService.make")(function* (input: { if (!loaded.ok) return; const wait = loaded.contents.waits.find((candidate) => candidate.id === key); if (wait === undefined) return; + // Granted but not delivered yet: its slot is not held when the delivery is written. + if (wait.kind === "slot" && (wait.state === "waiting" || wait.state === "delivering")) { + usedBeforeDelivery.add(key); + return; + } const now = yield* Clock.currentTimeMillis; const transition = nextWaitState(wait, { type: "slotUsed" }, now, stateSettings(settings)); if (transition === null) return; @@ -1694,7 +1725,10 @@ export const make = Effect.fn("WaitService.make")(function* (input: { settings, )); const kept = written ? changes : new Map(); - yield* input.slots!.drop([...granted.keys()].filter((id) => !kept.has(id))); + const notKept = [...granted.keys()].filter((id) => !kept.has(id)); + yield* input.slots!.drop(notKept); + // A cancelled wait is never delivered; one not written is granted again from the start. + for (const id of notKept) usedBeforeDelivery.delete(id); for (const [id, transition] of kept) { nextCheckAt.delete(id); yield* logTransition(transition); @@ -1967,9 +2001,11 @@ export const make = Effect.fn("WaitService.make")(function* (input: { if (result === undefined || wait.state !== "delivering") return wait; const group = planned.find((candidate) => candidate.waits.some((w) => w.id === wait.id)); if (group === undefined || wait.messageId !== group.messageId) return wait; + const events = eventsFor(result, wait, now, settings); + // Its slot was used before this delivery: the delivered wait holds none. return applyEvents( wait, - eventsFor(result, wait, now, settings), + usedBeforeDelivery.has(wait.id) ? [...events, { type: "slotUsed" }] : events, now, settings, transitions, @@ -1985,6 +2021,22 @@ export const make = Effect.fn("WaitService.make")(function* (input: { // Not written: the next sweep finds the message in the thread, or replays its receipt. if (!written) return { sent: 0, ended: 0 }; yield* Effect.forEach(transitions, logTransition, { discard: true }); + // A granted slot wait handed over to the parent, or ended without a receiver: the slot + // held for its thread is freed at once (the parent cannot use it). + const noReceiver = transitions.flatMap(({ record }) => + record.kind === "slot" && + (record.handedOverBecause !== undefined || + ["thread_gone", "thread_stopped", "dropped"].includes(record.state)) + ? [record.id] + : [], + ); + if (input.slots !== undefined) yield* input.slots.drop(noReceiver); + for (const id of usedBeforeDelivery) { + const wait = waits.find((candidate) => candidate.id === id); + if (wait === undefined || (wait.state !== "waiting" && wait.state !== "delivering")) { + usedBeforeDelivery.delete(id); + } + } } for (const [group, result] of results) { const at = nextTry(group, result, waits, now, settings); @@ -2176,7 +2228,11 @@ export const make = Effect.fn("WaitService.make")(function* (input: { (wait) => wait.kind === "slot" && wait.state === "waiting", ); for (const wait of current.waits) { - if (slotsWaiting) for (const held of storedReservation(wait, now)) at(held.untilMs); + if (slotsWaiting) { + for (const held of storedReservation(wait, now, settings.agentSlots.reserveSeconds)) { + at(held.untilMs); + } + } if (wait.state === "waiting") at(nextCheckAt.get(wait.id) ?? now); else if (wait.state !== "delivering" && wait.endedAt !== undefined) { at( diff --git a/packages/shared/src/agentSlots.test.ts b/packages/shared/src/agentSlots.test.ts index e43fe95535323..4b0fe1c1b58cd 100644 --- a/packages/shared/src/agentSlots.test.ts +++ b/packages/shared/src/agentSlots.test.ts @@ -8,6 +8,7 @@ import { limitsCount, liveReservations, liveStarting, + onePerThread, runningAgents, slotDecision, slotRefusalText, @@ -266,6 +267,24 @@ describe("slot reservations (4.6)", () => { expect(liveReservations([], 0)).toEqual([]); }); + it("count at most one per thread: the latest end, then the last key", () => { + const of = (key: string, threadId: string, untilMs: number): SlotReservation => ({ + key, + threadId, + untilMs, + }); + expect( + onePerThread([ + of("w_a", "t1", 200), + of("w_b", "t2", 300), + of("w_c", "t1", 400), + of("w_d", "t1", 400), + of("w_e", "t3", 100), + ]).map((reservation) => reservation.key), + ).toEqual(["w_b", "w_d", "w_e"]); + expect(onePerThread([])).toEqual([]); + }); + it("count on the machine, and on an account only when they name it", () => { const base = { limits: limitsOf({ maxPerMachine: 3, maxPerAccount: { [MAIN]: 1 } }), diff --git a/packages/shared/src/agentSlots.ts b/packages/shared/src/agentSlots.ts index e9ec96dc67263..9ec5d0213b8fe 100644 --- a/packages/shared/src/agentSlots.ts +++ b/packages/shared/src/agentSlots.ts @@ -148,6 +148,27 @@ export function liveReservations( return [...byKey.values()].sort((a, b) => a.untilMs - b.untilMs || (a.key < b.key ? -1 : 1)); } +/** + * The held slots that count: at most one per thread (the latest end wins, then the last key), in + * the order given. A thread that asks again moves its slot; it never holds two. + */ +export function onePerThread( + reservations: ReadonlyArray, +): ReadonlyArray { + const byThread = new Map(); + for (const reservation of reservations) { + const seen = byThread.get(reservation.threadId); + if ( + seen === undefined || + seen.untilMs < reservation.untilMs || + (seen.untilMs === reservation.untilMs && seen.key < reservation.key) + ) { + byThread.set(reservation.threadId, reservation); + } + } + return reservations.filter((reservation) => byThread.get(reservation.threadId) === reservation); +} + export type SlotRule = "machine" | "account" | "memory"; export type SlotDecision = diff --git a/packages/shared/src/waitFor.ts b/packages/shared/src/waitFor.ts index 00834732c2071..59bb68b49c4e5 100644 --- a/packages/shared/src/waitFor.ts +++ b/packages/shared/src/waitFor.ts @@ -812,6 +812,23 @@ export function limitErrorWakes(input: { return { changes, heldThreads: held.size, joined }; } +/** + * A stored held slot ends at most `reserveSeconds` after its delivery (`endedAt`) and after `nowMs`, + * the time the store is read: a clock that was ahead or a hand-edited or damaged file never holds + * a slot longer, and the record can be pruned after it. Ended or unreadable: no hold at all. + */ +export function boundHeldSlot(wait: WaitRecord, nowMs: number, reserveSeconds: number): WaitRecord { + if (wait.reservedUntil === undefined) return wait; + const endedMs = wait.endedAt === undefined ? Number.NaN : Date.parse(wait.endedAt); + const limitMs = + Math.min(nowMs, Number.isFinite(endedMs) ? endedMs : nowMs) + reserveSeconds * MS_PER_SECOND; + const untilMs = Date.parse(wait.reservedUntil); + if (untilMs <= limitMs) return wait; + return limitMs > nowMs + ? update(wait, { reservedUntil: iso(limitMs) }) + : update(wait, {}, ["reservedUntil"]); +} + /** Ended waits leave the file after `keepEndedHours`, a slot wait not before its reservation. */ export function shouldPruneWait( wait: WaitRecord, From 920926155964e8944b54ab4b3754dbb768678f54 Mon Sep 17 00:00:00 2001 From: scuttletoby <123843277+scuttletoby@users.noreply.github.com> Date: Sun, 15 Nov 2026 01:47:08 +0000 Subject: [PATCH 3/3] fix(slots): asking again moves a held slot but never extends it A thread that kept asking without starting an agent renewed its hold on every ask, so an older wait of another thread never got the slot. The moved hold now keeps the earlier end. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/slots/SlotGate.test.ts | 41 +++++++++++++++++++++----- apps/server/src/slots/SlotGate.ts | 12 +++++++- 2 files changed, 45 insertions(+), 8 deletions(-) diff --git a/apps/server/src/slots/SlotGate.test.ts b/apps/server/src/slots/SlotGate.test.ts index bfee646682b0f..d57957a27ff26 100644 --- a/apps/server/src/slots/SlotGate.test.ts +++ b/apps/server/src/slots/SlotGate.test.ts @@ -818,7 +818,7 @@ describe("SlotGate", () => { }), ); - it.effect("one held slot per thread: asking again moves and renews it, never adds one", () => + it.effect("one held slot per thread: asking again moves it, never adds or extends one", () => Effect.gen(function* () { const world = newWorld(); yield* run( @@ -826,19 +826,19 @@ describe("SlotGate", () => { waitsOn({ maxPerMachine: 3, reserveSeconds: 60 }), Effect.gen(function* () { const gate = yield* SlotGate.SlotGate; - // Asked three times (three slots free): still one slot held. - for (const key of ["w1", "w2", "w3"]) { + // Asked three times (three slots free, at 0, 20 and 40 s): still one slot held. + for (const [index, key] of ["w1", "w2", "w3"].entries()) { + if (index > 0) yield* TestClock.adjust("20 seconds"); assert.deepStrictEqual(types(yield* grant({ key, threadId: holder })), ["granted"]); - yield* TestClock.adjust("20 seconds"); } assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 1, max: 3 }); assert.strictEqual((yield* acquireAs(other)).type, "allow"); assert.strictEqual((yield* acquireAs(other)).type, "allow"); assert.strictEqual((yield* acquireAs(other)).type, "refuse"); - // Renewed by the last ask (at 40 s): still held at 90 s, gone after 100 s. - yield* TestClock.adjust("30 seconds"); + // Not renewed by the later asks: held until the first ask's end (60 s), then gone. + yield* TestClock.adjust("19 seconds"); assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 3, max: 3 }); - yield* TestClock.adjust("15 seconds"); + yield* TestClock.adjust("1 second"); assert.deepStrictEqual((yield* gate.usage)?.machine, { used: 2, max: 3 }); }), ); @@ -936,5 +936,32 @@ describe("SlotGate", () => { ); }), ); + it.effect("asking again moves a held slot but never extends it past the first end", () => + Effect.gen(function* () { + const world = newWorld(); + yield* run( + world, + waitsOn({ maxPerMachine: 1, reserveSeconds: 60 }), + Effect.gen(function* () { + assert.deepStrictEqual(types(yield* grant({ key: "h0", threadId: holder })), [ + "granted", + ]); + // An older wait of another thread comes first in every sweep; the holder asks + // again every 50 s without starting an agent. + const otherGot: Array = []; + for (let round = 1; round <= 10; round += 1) { + yield* TestClock.adjust("50 seconds"); + const grants = yield* grant( + { key: "o1", threadId: other }, + { key: `h${round}`, threadId: holder }, + ); + otherGot.push(grants[0]!.type); + } + // The first hold ended at 60 s: the other thread got the slot at the 100 s sweep. + assert.deepStrictEqual(otherGot.slice(0, 2), ["full", "granted"]); + }), + ); + }), + ); }); }); diff --git a/apps/server/src/slots/SlotGate.ts b/apps/server/src/slots/SlotGate.ts index 42de654f9cbc0..65fafadbc1de7 100644 --- a/apps/server/src/slots/SlotGate.ts +++ b/apps/server/src/slots/SlotGate.ts @@ -346,7 +346,17 @@ const make = Effect.gen(function* () { grants.push({ type: "full", decision }); continue; } - const untilMs = nowMs + reserveMs; + // Asking again moves its hold but never extends it: a thread that keeps asking without + // starting an agent loses the slot at the old end, so an older waiter gets it (R17). + const heldUntil = reservations + .filter((reservation) => reservation.threadId === request.threadId) + .reduce( + (latest, reservation) => + latest === null ? reservation.untilMs : Math.max(latest, reservation.untilMs), + null, + ); + const untilMs = + heldUntil === null ? nowMs + reserveMs : Math.min(nowMs + reserveMs, heldUntil); const reservation: SlotReservation = { key: request.key, threadId: request.threadId,