limits: start the board's best-ranked held thread first

FORK.md

@@ -381,7 +381,11 @@ Users read about it in the Memory and CPU limits section of `docs/user/remote-ac
   queue, held, with a `system_notice`. Her own messages always start, and so does a queue holding one of them.
   Work launched from a thread whose latest message is her "launch anyway" starts too. Idle sessions close on
   every pass over the limits, before anything waits. Below the critical line, the newest running turn is
-  interrupted every 30 s, and its thread queues a resume note. Held threads start oldest first, 90 s apart,
+  interrupted every 30 s, and its thread queues a resume note. A held thread the Task Priorities board ranks
+  starts first, best rank first. A thread without a rank, such as a task's helper or a fork of its thread, takes
+  the rank of the nearest ranked thread above it (`lineage.parentThreadId`, up to 10 levels; a deleted thread
+  ends the walk). A thread another thread launched at the top level doesn't inherit, since the waiting list
+  doesn't record who launched it. The rest start oldest first. Either way they start 90 s apart,
   or one every pass while the reading would stay at its release lines once each session released in the
   last two minutes, and the next one, grows by 2 GiB (`releaseDue`). Without a reading or a release line,
   the 90 s spacing holds. On Oct 29, 18 threads waited on the PC with 30 to 50 GB free, because new work
@@ -410,11 +414,14 @@ Users read about it in the Memory and CPU limits section of `docs/user/remote-ac
   - `LoadBalancingSettings.tsx` shows each machine's limits under its row, and hands a lone machine to
     `MachineLimitsSection`. `settingsSearch.ts` adds search terms.
   - `packages/client-runtime/src/load-balancing.ts`: Auto balance skips a machine over its limits.
-- The memory gate's hook reads the governor's files under `<stateDir>/resource-limits/`:
+- teenylilconfig's memory gate and task wake share the governor's files under `<stateDir>/resource-limits/`:
   - `state.json`: state, readings, waiting threads, recent release times, and each running session's thread and
     "launch anyway". It's rewritten every pass, so a stale file means T3 isn't guarding.
   - `off.json`: `memgate off` writes `{ at, until }` in epoch ms. It switches the limits off for at most a day.
   - `inbox/`: one record per subagent launch the hook refused. Its thread resumes with a note listing them.
+  - `priority.json`: teenylilconfig's task wake writes `{ version: 1, at, ranks }` from the board every 10 minutes,
+    each ranked task's thread ID with its rank (1 is the top). One more than a day old, more than a minute
+    ahead, over 1 MiB, or malformed counts as no ranks (`priorityRanks`).
 - Stock compatibility: holds, notices and resume notes reuse `run.updated`, `turn-item.updated` and system
   notifications. Stock T3 drops `resourceLimits` the next time it saves settings.
 - Known gaps:

apps/server/src/resourceLimits/Governor.fork.test.ts

@@ -15,7 +15,9 @@ import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAda
 import * as HostProcess from "@t3tools/shared/HostProcess";
 import * as Clock from "effect/Clock";
 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 * as Ref from "effect/Ref";
 import * as Stream from "effect/Stream";
 import * as TestClock from "effect/testing/TestClock";
@@ -402,6 +404,149 @@ it.effect("R3: a restart keeps counting the sessions still growing", () =>
   }).pipe(Effect.provide(layerTest)),
 );
 
+/** Writes the board's ranks where task wake does, stamped now unless `at` says otherwise. */
+const writeRanks = (contents: Record<string, unknown> | string, at?: number) =>
+  Effect.gen(function* () {
+    const config = yield* ServerConfig.ServerConfig;
+    const fs = yield* FileSystem.FileSystem;
+    const path = yield* Path.Path;
+    const now = yield* Clock.currentTimeMillis;
+    const dir = path.join(config.stateDir, "resource-limits");
+    const text =
+      typeof contents === "string"
+        ? contents
+        : JSON.stringify({ version: 1, at: at ?? now, source: "task-wake", ranks: contents });
+    yield* fs.makeDirectory(dir, { recursive: true });
+    yield* fs.writeFileString(path.join(dir, "priority.json"), text);
+  });
+
+/**
+ * Agent work held in each of `held`, a second apart, by a governor that sees `parents` as each thread's parent.
+ * `others` are threads that exist without held work.
+ */
+const holdThreads = (
+  name: string,
+  held: ReadonlyArray<string>,
+  parents: Readonly<Record<string, string>> = {},
+  others: ReadonlyArray<string> = [],
+) =>
+  Effect.gen(function* () {
+    const projections = yield* ProjectionStore.ProjectionStoreV2;
+    const id = (thread: string) => ThreadId.make(`thread:${name}:${thread}`);
+    const parentOf = new Map(
+      Object.entries(parents).map(([child, parent]) => [id(child), id(parent)]),
+    );
+    const governor = yield* makeGovernor.pipe(
+      Effect.provideService(ProjectionStore.ProjectionStoreV2, {
+        ...projections,
+        getThreadShell: (threadId) =>
+          projections.getThreadShell(threadId).pipe(
+            Effect.map((shell) => {
+              const parent = parentOf.get(threadId);
+              return shell !== null && parent !== undefined
+                ? { ...shell, lineage: { ...shell.lineage, parentThreadId: parent } }
+                : shell;
+            }),
+          ),
+      }),
+    );
+    const threads = held.map(id);
+    yield* Effect.forEach([...threads, ...others.map(id)], createThread);
+    yield* governor.pass(yield* tight);
+    for (const threadId of threads) {
+      yield* send(threadId, "agent work", agent);
+      yield* TestClock.adjust("1 second");
+    }
+    const started = Effect.forEach(threads, (threadId) =>
+      runs(threadId).pipe(Effect.map((all) => all.join() === "starting")),
+    );
+    // Past the 90 s spacing and the growth allowance, so the next pass releases one more.
+    const nextRelease = Effect.gen(function* () {
+      yield* TestClock.adjust(TIMING.releaseSpacingMs + 15_000);
+      yield* governor.pass(yield* recovered);
+    });
+    return { governor, id, started, nextRelease };
+  });
+
+// The board's tasks a, b and c, and a helper c launched.
+const withHelper = ["a", "b", "c", "helper"];
+
+it.effect(
+  "R6: the board's best-ranked task starts first, its helpers next, the rest oldest first",
+  () =>
+    Effect.gen(function* () {
+      const { governor, id, started, nextRelease } = yield* holdThreads("r6", withHelper, {
+        helper: "c",
+      });
+      yield* writeRanks({ [id("b")]: 5, [id("c")]: 2, [id("not-waiting")]: 1 });
+
+      yield* governor.pass(yield* recovered);
+      assert.deepEqual(yield* started, [false, false, true, false]);
+      yield* nextRelease;
+      assert.deepEqual(yield* started, [false, false, true, true]);
+      yield* nextRelease;
+      assert.deepEqual(yield* started, [false, true, true, true]);
+    }).pipe(Effect.provide(layerTest)),
+);
+
+it.effect("R6: each release reads the latest ranks, and an oversized file counts as none", () =>
+  Effect.gen(function* () {
+    const { governor, id, started, nextRelease } = yield* holdThreads("r6-change", withHelper, {
+      helper: "c",
+    });
+    yield* writeRanks({ [id("b")]: 5 });
+    yield* governor.pass(yield* recovered);
+    assert.deepEqual(yield* started, [false, true, false, false]);
+
+    yield* writeRanks({ [id("a")]: 1 });
+    yield* nextRelease;
+    assert.deepEqual(yield* started, [true, true, false, false]);
+
+    const at = (yield* Clock.currentTimeMillis) + TIMING.releaseSpacingMs + 15_000;
+    const ranks = { [id("helper")]: 1 };
+    yield* writeRanks(
+      JSON.stringify({ version: 1, at, source: "task-wake", ranks, pad: "x".repeat(1024 * 1024) }),
+    );
+    yield* nextRelease;
+    assert.deepEqual(yield* started, [true, true, true, false]);
+  }).pipe(Effect.provide(layerTest)),
+);
+
+it.effect("R6: a rank passes down at most 10 levels of helpers", () =>
+  Effect.gen(function* () {
+    // x0's parent is x1, and so on up to x10, which has the rank; y0 sits one level below x0.
+    const chain = Array.from({ length: 10 }, (_, level) => [`x${level}`, `x${level + 1}`]);
+    const { governor, id, started, nextRelease } = yield* holdThreads(
+      "r6-deep",
+      ["old", "x0", "y0"],
+      { ...Object.fromEntries(chain), y0: "x0" },
+      chain.map(([, parent]) => parent!),
+    );
+    yield* writeRanks({ [id("x10")]: 1 });
+
+    yield* governor.pass(yield* recovered);
+    assert.deepEqual(yield* started, [false, true, false]);
+    yield* nextRelease;
+    assert.deepEqual(yield* started, [true, true, false]);
+  }).pipe(Effect.provide(layerTest)),
+);
+
+it.effect("R6: an unreadable or day-old ranks file leaves the line oldest first", () =>
+  Effect.gen(function* () {
+    const { governor, id, started, nextRelease } = yield* holdThreads("r6-stale", withHelper, {
+      helper: "c",
+    });
+    yield* writeRanks("{ not json");
+    yield* governor.pass(yield* recovered);
+    assert.deepEqual(yield* started, [true, false, false, false]);
+
+    const day = TIMING.priorityStaleMs + TIMING.releaseSpacingMs + 15_001;
+    yield* writeRanks({ [id("c")]: 1 }, (yield* Clock.currentTimeMillis) - day);
+    yield* nextRelease;
+    assert.deepEqual(yield* started, [true, true, false, false]);
+  }).pipe(Effect.provide(layerTest)),
+);
+
 it.effect("R4: idle sessions close on every pass over the limits, before any work waits", () =>
   Effect.gen(function* () {
     const governor = yield* makeGovernor;

apps/server/src/resourceLimits/Governor.ts

@@ -41,7 +41,9 @@ import {
   newestInterruptible,
   nextState,
   offUntil,
+  priorityRanks,
   releaseDue,
+  releaseOrder,
   type ResumeNote,
   type RunningTurn,
   TIMING,
@@ -89,6 +91,10 @@ const Json = Schema.fromJsonString(Schema.Unknown, { space: 2 });
 const decodeJson = Schema.decodeUnknownOption(Json);
 const encodeJson = Schema.encodeEffect(Json);
 const MAX_INTERRUPTED_RUN_IDS = 50;
+// A helper thread takes the rank of the nearest thread above it that has one, this many levels up at most.
+const MAX_RANK_HOPS = 10;
+// Any local process can write the ranks file, so a larger one is skipped unread.
+const MAX_PRIORITY_BYTES = 1024 * 1024;
 
 const waitsOnAlice = (shell: OrchestrationV2ThreadShell) =>
   shell.pendingRuntimeRequest !== null && shell.pendingRuntimeRequest.blocking !== false;
@@ -115,6 +121,8 @@ export const make = Effect.fn("makeResourceLimitsGovernor")(function* () {
     state: path.join(dir, "state.json"),
     off: path.join(dir, "off.json"),
     inbox: path.join(dir, "inbox"),
+    // The Task Priorities board's ranks, which teenylilconfig's task wake writes; see FORK.md.
+    priority: path.join(dir, "priority.json"),
   };
 
   const memory = yield* Ref.make<Memory>({
@@ -271,6 +279,51 @@ export const make = Effect.fn("makeResourceLimitsGovernor")(function* () {
         })
       : Effect.succeed(true);
 
+  const readPriority = (now: number) =>
+    fs.stat(files.priority).pipe(
+      Effect.flatMap((info) =>
+        Number(info.size) > MAX_PRIORITY_BYTES ? Effect.succeed(null) : readJson(files.priority),
+      ),
+      Effect.orElseSucceed((): unknown => null),
+      Effect.map((file) => priorityRanks(file, now)),
+    );
+
+  // The board's best-ranked task goes first. A thread without a rank of its own, such as a task's helper or a
+  // fork of its thread, takes the rank of the nearest thread above it that has one. A deleted thread ends the walk.
+  const byRank = Effect.fn("ResourceLimitsGovernor.byRank")(function* (
+    ready: ReadonlyArray<Inspected>,
+    now: number,
+  ) {
+    if (ready.length < 2) return ready;
+    const ranks = yield* readPriority(now);
+    if (ranks.size === 0) return ready;
+    // Each thread above is read once a pass, however many helpers share it.
+    const shells = new Map<ThreadId, OrchestrationV2ThreadShell | null>();
+    const shellOf = (threadId: ThreadId) =>
+      shells.has(threadId)
+        ? Effect.succeed(shells.get(threadId) ?? null)
+        : projections.getThreadShell(threadId).pipe(
+            Effect.orElseSucceed(() => null),
+            Effect.tap((shell) => Effect.sync(() => shells.set(threadId, shell))),
+          );
+    const rankOf = new Map<ThreadId, number | null>();
+    for (const item of ready) {
+      let shell = item.shell;
+      let rank: number | null = null;
+      const seen = new Set<ThreadId>();
+      for (let hop = 0; hop <= MAX_RANK_HOPS; hop++) {
+        if (shell === null || shell.deletedAt !== null || seen.has(shell.id)) break;
+        seen.add(shell.id);
+        rank = ranks.get(shell.id) ?? null;
+        const parent = shell.lineage.parentThreadId;
+        if (rank !== null || parent == null) break;
+        shell = yield* shellOf(parent);
+      }
+      rankOf.set(item.entry.threadId, rank);
+    }
+    return releaseOrder(ready, (item) => rankOf.get(item.entry.threadId) ?? null);
+  });
+
   const sessionsFor = Effect.fn("ResourceLimitsGovernor.sessionsFor")(function* (
     shells: ReadonlyArray<OrchestrationV2ThreadShell>,
   ) {
@@ -481,7 +534,8 @@ export const make = Effect.fn("makeResourceLimitsGovernor")(function* () {
       }
     }
 
-    // Release oldest first, spaced out, once the reading this pass took is back under the limits.
+    // Release the best-ranked board task first and the rest oldest first, spaced out, once the reading this pass
+    // took is back under the limits.
     let lastReleaseAt = remembered.lastReleaseAt;
     const growingSince = remembered.growingSince.filter((at) => now - at < TIMING.sessionGrowthMs);
     const released = new Set<ThreadId>();
@@ -493,7 +547,7 @@ export const make = Effect.fn("makeResourceLimitsGovernor")(function* () {
         (item) => !dropped.has(item.entry.threadId) && item.held && !item.blocked,
       );
       // Switched off, everything waiting goes at once.
-      for (const item of off ? ready : ready.slice(0, 1)) {
+      for (const item of off ? ready : (yield* byRank(ready, now)).slice(0, 1)) {
         yield* unsnooze({ ...item, entry: changed.get(item.entry.threadId) ?? item.entry }, now);
         yield* dispatch({
           type: "queue.resume",

apps/server/src/resourceLimits/ResourceLimits.ts

@@ -126,7 +126,7 @@ const make = Effect.fn("makeResourceLimits")(function* () {
     const now = yield* Clock.currentTimeMillis;
     const current = yield* Ref.get(snapshot);
     const state = liveState(current, now);
-    // While anything waits, new work queues behind it, so release order stays oldest first.
+    // While anything waits, new work queues behind it, so the governor picks what starts next.
     if (current.limits === null || isOff(current, now)) return ADMITTED;
     if (state === "ok" && current.waiting.length === 0) return ADMITTED;
     const launcher = input.message.senderThreadId ?? input.thread.lineage.parentThreadId;

apps/server/src/resourceLimits/policy.fork.test.ts

@@ -15,7 +15,9 @@ import {
   newestInterruptible,
   nextState,
   offUntil,
+  priorityRanks,
   releaseDue,
+  releaseOrder,
   SESSION_GIB,
   TIMING,
 } from "./policy.ts";
@@ -325,3 +327,66 @@ describe("heldLaunchesNote", () => {
     expect(note.text.length).toBeLessThan(5 * 200 + 300);
   });
 });
+
+describe("priorityRanks", () => {
+  const at = 10 * 3_600_000;
+
+  it("reads the board's positive ranks from a fresh file", () => {
+    const ranks = priorityRanks(
+      { version: 1, at, source: "task-wake", ranks: { "thread:a": 2, "mcp:b": 1, "thread:c": 0 } },
+      at + 60_000,
+    );
+    expect([...ranks]).toEqual([
+      ["thread:a", 2],
+      ["mcp:b", 1],
+    ]);
+  });
+
+  it("treats a day-old, future, malformed or missing file as no ranks", () => {
+    const ranks = { "thread:a": 1 };
+    expect(priorityRanks({ version: 1, at, ranks }, at + TIMING.priorityStaleMs + 1).size).toBe(0);
+    expect(priorityRanks({ version: 1, at, ranks }, at - TIMING.priorityAheadMs - 1).size).toBe(0);
+    expect(priorityRanks({ version: 2, at, ranks }, at).size).toBe(0);
+    expect(priorityRanks({ version: 1, at: String(at), ranks }, at).size).toBe(0);
+    expect(priorityRanks({ version: 1, at, ranks: [1, 2] }, at).size).toBe(0);
+    expect(priorityRanks(null, at).size).toBe(0);
+  });
+
+  it("drops ranks that aren't positive finite numbers", () => {
+    const ranks = priorityRanks(
+      {
+        version: 1,
+        at,
+        ranks: { a: "1", b: -3, c: Number.POSITIVE_INFINITY, d: null, e: 1.5 },
+      },
+      at,
+    );
+    expect([...ranks]).toEqual([["e", 1.5]]);
+  });
+
+  it("skips thread IDs longer than any T3 makes", () => {
+    const ranks = priorityRanks({ version: 1, at, ranks: { ["x".repeat(513)]: 1, ok: 2 } }, at);
+    expect([...ranks]).toEqual([["ok", 2]]);
+  });
+});
+
+describe("releaseOrder", () => {
+  it("puts ranked threads first, best rank first, and keeps arrival order otherwise", () => {
+    const ranks = new Map([
+      ["d", 3],
+      ["b", 1],
+      ["e", 3],
+    ]);
+    expect(releaseOrder(["a", "b", "c", "d", "e"], (id) => ranks.get(id) ?? null)).toEqual([
+      "b",
+      "d",
+      "e",
+      "a",
+      "c",
+    ]);
+  });
+
+  it("leaves the order alone when nothing is ranked", () => {
+    expect(releaseOrder(["a", "b", "c"], () => null)).toEqual(["a", "b", "c"]);
+  });
+});