diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 27a1da2ca6b9..2644c713295e 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 7241eca51c53..d57957a27ff2 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,481 @@ 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, + }); + }), + ); + + // 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 it, never adds or extends 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, 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"]); + } + 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"); + // 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("1 second"); + 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, + }); + }), + ); + }), + ); + 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 4fe8535cac9e..65fafadbc1de 100644 --- a/apps/server/src/slots/SlotGate.ts +++ b/apps/server/src/slots/SlotGate.ts @@ -1,17 +1,24 @@ /** * 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, + onePerThread, 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 +42,46 @@ 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). 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 }; + +/** 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 +101,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 +127,323 @@ 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), ); - const acquire = Effect.fn("SlotGate.acquire")( + /** + * 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); + 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); + + /** + * 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. + */ + 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: [] as ReadonlyArray, + }; + } + 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: onePerThread( + 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: [] as ReadonlyArray, + }; } 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: 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), ); + 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). + yield* clearStored(used); + return outcome; + }); + + 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 { grants: requests.map((): SlotGrant => ({ type: "off" })), replaced: [] }; + } + 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 => + 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; + 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." }); + continue; + } + const decision = slotDecision({ + limits: own, + running: (read?.running ?? []).filter((agent) => agent.threadId !== request.threadId), + starting: kept, + // 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; + } + // 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, + ...(request.account === undefined ? {} : { account: request.account }), + untilMs, + }; + // 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, 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; + 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 = 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; + }; + 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 000000000000..f80f4062cb91 --- /dev/null +++ b/apps/server/src/waits/WaitService.slots.test.ts @@ -0,0 +1,788 @@ +/** + * 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)))); + }), + ); + + 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 d6fb090ae8aa..1a097482e96c 100644 --- a/apps/server/src/waits/WaitService.ts +++ b/apps/server/src/waits/WaitService.ts @@ -14,14 +14,21 @@ 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 { + boundHeldSlot, capEndedWaits, deliveryCommandId, type DeliveryGroup, @@ -56,6 +63,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 +221,71 @@ 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 "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`: 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 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 [ + { + key: wait.id, + threadId: wait.threadId, + untilMs, + ...(account === undefined ? {} : { account }), + }, + ]; +}; + const waitingResult = (wait: WaitRecord, deduped: boolean): WaitForResult => ({ state: "waiting", waitId: wait.id, @@ -327,6 +397,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; @@ -354,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. */ @@ -508,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; }); @@ -551,6 +637,67 @@ 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 settings = yield* readSettingsOrNull; + if (settings === null) return []; + const now = yield* Clock.currentTimeMillis; + return loaded.contents.waits.flatMap((wait) => + storedReservation(wait, now, settings.agentSlots.reserveSeconds), + ); + }), + ); + + /** 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; + // 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; + 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 +723,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 +776,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 +793,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 +893,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 +901,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 +933,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 +1543,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 +1634,109 @@ 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(); + 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); + } + 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 @@ -1729,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, @@ -1747,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); @@ -1933,7 +2223,16 @@ 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, 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( @@ -1995,7 +2294,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 +2335,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 7ace48d28e24..bdf976b9d34a 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 845939d70044..4b0fe1c1b58c 100644 --- a/packages/shared/src/agentSlots.test.ts +++ b/packages/shared/src/agentSlots.test.ts @@ -6,12 +6,15 @@ import { DISPATCH_SAFETY_MS, effectiveSlotLimits, limitsCount, + liveReservations, liveStarting, + onePerThread, runningAgents, slotDecision, slotRefusalText, type SlotDecision, type SlotLimits, + type SlotReservation, type SlotThread, type StartingEntry, visibleRunThreadIds, @@ -243,6 +246,74 @@ 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 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 } }), + 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 cc36e43d80d6..9ec5d0213b8f 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,55 @@ 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)); +} + +/** + * 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 = @@ -139,6 +188,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 +198,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 +206,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 +214,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 +232,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 +252,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 68c3f0b5757b..dc4115ab661d 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 2c47a1e28e85..59bb68b49c4e 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 { @@ -802,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,