sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix(durable): bound the scheduler's ownership indexes by live work

Browse source 
The scheduler kept every conversation edge it saw and, for conversation owners, chain loads, and owners of known
tasks, the ended tasks above them, for the life of the process. Now an owner edge or an ended task's ownership is kept
only while something is listed below it in #below: only the chains above live work are known, whole. Unneeded nodes
are queued and swept at the end of each publication, or by a sweep job on the Session line after operations that
publish nothing (rejected commits, reads), never during an operation, so a commit that creates work below a chain its
own callback loaded keeps it. Later work in an old conversation loads its chain from storage again.

Removes #conversationOwners, #pinned, #knownChildren, and #forget. A queued input triggers the withdrawal scan only
while some task has cancellation intent, and chains are loaded only for queued inputs.

10,000 ended subagent cycles under a live parent: 10,000 settled tasks and 10,001 edges before, 0 and 1 now
(bench:scheduler-memory); a later message into an old subagent conversation reads 7 records instead of 5.
This commit is contained in:
Mario Zechner 2026-10-10 19:29:35 +00:00
parent e37cdc4206
commit ea448f454a
6 changed files with 807 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

1
packages/durable/CHANGELOG.md
View file

@ -26,6 +26,7 @@
- A throwing `onReport` no longer replaces the error being reported or becomes an unhandled rejection, a throwing `now()` is reported once and `Date.now` used, and a watch listener's error is reported as well as ending the watch with `listener_error`.
- A task whose fault write failed no longer runs again in the same process.
- The scheduler no longer visits every live task on each pass: abort cascades start only from tasks with cancellation intent, owned live work and idle scopes are walks down an in-memory ownership index, and reservation, finalization, and failFast checks keep their own candidate sets. A chain of 500 owned tasks that took 12 s to settle takes 50 ms, 20,000 take 1.5 s; a tool making 8000 parallel nested calls finishes in 2.7 s instead of 18.8 s, with one document per nested call instead of one growing index (`npm run bench:scheduler-fanout`).
- The scheduler's ownership indexes are bounded by live work: it keeps a conversation's owner and an ended task's ownership only while live work is below them, instead of every conversation and subagent chain it ever saw. Work that later appears in an old conversation loads its chain from storage again. After 10,000 ended subagent calls under a live parent it kept 10,000 ended tasks and 10,001 conversation edges; now none of either beyond the parent's conversation (`npm run bench:scheduler-memory`). A queued input no longer makes the scheduler scan every queued submission when nothing is being cancelled.
## [1.1.0] - 2026-10-07

1
packages/durable/package.json
View file

@ -108,6 +108,7 @@
"bench:nested-tools": "node --conditions=source --experimental-strip-types --expose-gc test/nested-tools-bench.ts",
"bench:nested-images": "node --conditions=source --experimental-strip-types --expose-gc test/nested-images-bench.ts",
"bench:scheduler-fanout": "node --conditions=source --experimental-strip-types test/scheduler-fanout-bench.ts",
"bench:scheduler-memory": "node --conditions=source --experimental-strip-types --expose-gc test/scheduler-memory-bench.ts",
"prepublishOnly": "npm run clean && npm run build"
},
"keywords": [

8
packages/durable/src/harness/harness.ts
View file

@ -80,6 +80,9 @@ type ConversationHost<Tool extends ToolRegistration> = {
create(target: CreateTarget, options: CreateOptions, context: Context): Promise<Conversation>;
};
/** Key of the Harness's internal index-size getter, for tests. */
export const SCHEDULER_INDEX_SIZES = Symbol.for("pi-durable.schedulerIndexSizes");
class ConversationImpl<Tool extends ToolRegistration> implements Conversation {
readonly id: ConversationId;
readonly #host: ConversationHost<Tool>;
@ -246,6 +249,11 @@ class HarnessImpl<Tool extends ToolRegistration> extends SessionImpl implements
>;
}
/** Internal, for tests: sizes of the scheduler's ownership indexes. */
get [SCHEDULER_INDEX_SIZES](): TaskScheduler["indexSizes"] {
return this.#tasks.indexSizes;
}
inspect(context: Context): Promise<HarnessInspection> {
return this.readOnLine(async () => {
const { scheduling, tasks } = await this.#tasks.inspect(this.#options.registry.snapshot());

130
packages/durable/src/harness/scheduler.ts
View file

@ -187,6 +187,10 @@ export type TaskSchedulerOptions = {
* it: a task's live ordinary owned work, which holds its outcome as `completing` and delays its abort handler, a
* scope's tasks, a cascade's reach. No pass visits every live task; the sets below hold each pass's candidates.
*
* The ownership indexes are bounded by live work: an owner edge and an ended task's ownership fields are kept only while
* something is listed below them in `#below`, so only the chains above live tasks are known, whole. Work that appears
* later below a dropped node finds it unknown and loads its chain from storage again.
*
* Invariant: every task transition is decided and written by one callback serialized on the Session line. That covers
* reservation, marks, runtime commits, finalization, and the synchronous step before each phase, which applies the
* precedence rules and writes a fault or handover. Handlers and joins run off the line. An invocation ends inside the
@ -221,18 +225,15 @@ export class TaskScheduler {
readonly #idleWaiters = new Waiters<ConversationId | undefined, void>();
/** Definition whose migration failed per task; retried only once the registry resolves another definition. */
readonly #failedMigrations = new Map<TaskId, { readonly task: AnyTask; readonly error: unknown }>();
/** Owner task of each loaded conversation, `null` when ownerless. */
/** Owner task of each known conversation, `null` when ownerless. */
readonly #edges = new Map<ConversationId, TaskId | null>();
/**
* Tasks that own a loaded conversation; their nodes stay in `#settled` once terminal, and with them, through
* `#knownChildren`, every task above them. Never cleared: about one entry per task on a subagent's chain.
*/
readonly #conversationOwners = new Set<TaskId>();
/**
* Ownership fields of terminal tasks that walks pass through: conversation owners, chain loads, owners of known
* tasks, and those with nodes listed below them. `#forget` drops the rest.
*/
/** Ownership fields of the known terminal tasks, which walks pass through. */
readonly #settled = new Map<TaskId, TaskNode>();
/**
* Ended tasks and conversations found unneeded, dropped by `#sweep()` at the end of the commit's publication unless
* needed again by then: a chain load installs nodes before the work below them is listed.
*/
readonly #dropQueue = new Set<Node>();
/**
* The tree downward around live work: each node's child nodes that are live tasks or have nodes listed below them.
* A conversation's parent is its owner task; ownerless ones with work are in `#roots`.
@ -249,13 +250,6 @@ export class TaskScheduler {
readonly #finalizeChecks = new Set<TaskId>();
/** Live waiting tasks by each task in their `on`. */
readonly #waiters = new Map<TaskId, Set<TaskId>>();
/** Terminal tasks a chain load read; kept in `#settled`, so loaded chains stay loaded. */
readonly #pinned = new Set<TaskId>();
/**
* Per task, how many known tasks (live, or in `#settled`) it owns directly. An ended task stays known while any does,
* so the chain above every known task stays loaded.
*/
readonly #knownChildren = new Map<TaskId, number>();
/**
* `failFast` waiters the next reconcile checks for a failed task in `on`: at open, when they start waiting, and when
* one of their tasks fails.
@ -267,6 +261,7 @@ export class TaskScheduler {
*/
readonly #abandoned = new Set<TaskId>();
#reconcileScheduled = false;
#sweepScheduled = false;
#cascadePending = false;
#unsubscribeRegistry: () => void = () => {};
#enabled = false;
@ -478,6 +473,8 @@ export class TaskScheduler {
// A queued input below a cancelled owner is withdrawn, even after its cascade.
if (change.type !== "submission" || change.value.status !== "queued" || change.value.type !== "input")
continue;
// Only an owner with cancellation intent withdraws it; a later intent's mark runs the cascade then.
if (this.#intent.size === 0) continue;
const conversation = change.value.conversationId;
if (!this.#known(conversationNode(conversation)) || this.#belowCancelled({ conversation })) {
this.#cascadePending = true;
@ -492,6 +489,7 @@ export class TaskScheduler {
}
// A cascade that a reservation or step found pending runs with the next commit of any kind.
if (this.#cascadePending) this.#scheduleReconcile();
this.#sweep();
if (!changed) return;
this.#settleIdle();
this.#kick();
@ -676,7 +674,7 @@ export class TaskScheduler {
}
/**
* Load the owner chains of the tasks in `#unloaded` and, with `queued`, of every conversation with queued submissions,
* Load the owner chains of the tasks in `#unloaded` and, with `queued`, of every conversation with queued inputs,
* on the Session line; returns the latter. Reads committed Storage directly, so it may run inside a commit callback.
*/
async #loadScopes(queued: boolean): Promise<ConversationId[]> {
@ -688,7 +686,8 @@ export class TaskScheduler {
const submissions = await scanAll((cursor) =>
this.#storage.scanSubmissions({ status: "queued" }, SCAN_PAGE_SIZE, cursor, this.#context),
);
const conversations = [...new Set(submissions.map((submission) => submission.conversationId))];
const inputs = submissions.filter((submission) => submission.type === "input");
const conversations = [...new Set(inputs.map((submission) => submission.conversationId))];
for (const id of conversations) await this.#loadChain({ conversation: id });
return conversations;
}
@ -735,43 +734,61 @@ export class TaskScheduler {
#setEdge(conversationId: ConversationId, owner: TaskId | null): void {
this.#edges.set(conversationId, owner);
if (owner !== null) this.#conversationOwners.add(owner);
// The conversation now hangs from its owner, or is a root.
// The conversation now hangs from its owner, or is a root; with nothing below it, it waits for the sweep.
this.#relist(conversationNode(conversationId));
this.#dropIfUnneeded(conversationNode(conversationId));
}
/** Keep a terminal task a chain load read, so walks pass through it; its own chain is not loaded yet. */
/** Know a terminal task a chain load read, so walks pass through it; its own chain is not loaded yet. */
#settle(id: TaskId, node: TaskNode): void {
this.#settled.set(id, node);
this.#pinned.add(id);
this.#unloaded.add(id);
this.#countChild(node, 1);
this.#relist(id);
this.#dropIfUnneeded(id);
}
/** Count a task that became known in its owner task's `#knownChildren`, or one that is forgotten out of it. */
#countChild(node: TaskNode, delta: 1 | -1): void {
if (node.owner === undefined) return;
const count = (this.#knownChildren.get(node.owner) ?? 0) + delta;
if (count === 0) this.#knownChildren.delete(node.owner);
else this.#knownChildren.set(node.owner, count);
/** Queue an ended task or a conversation for the sweep when nothing is listed below it. */
#dropIfUnneeded(node: Node): void {
if (typeof node !== "string" && this.#live.has(node)) return;
if (this.#below.has(node)) return;
this.#dropQueue.add(node);
this.#scheduleSweep();
}
/**
* Forget ended task `id` unless walks up still need it: it owns a loaded conversation, a chain load read it, or it
* owns a known task; then its owner, as far up as that changes. A loop: chains can be deeper than the call stack.
* Sweep in a job of its own on the Session line, after the operation underway: a rejected commit, one with nothing to
* write, and a read publish nothing, so the sweep at the end of `#observe` would not run for what they loaded.
*/
#forget(id: TaskId): void {
for (let at: TaskId | undefined = id; at !== undefined; ) {
const node = this.#settled.get(at);
if (node === undefined || this.#live.has(at)) return;
if (this.#conversationOwners.has(at) || this.#pinned.has(at) || this.#knownChildren.has(at)) return;
if (this.#below.has(at)) return;
this.#settled.delete(at);
this.#unloaded.delete(at);
this.#countChild(node, -1);
at = node.owner;
#scheduleSweep(): void {
if (this.#sweepScheduled || this.#closing) return;
this.#sweepScheduled = true;
this.#session
.readOnLine(async () => {
this.#sweepScheduled = false;
this.#sweep();
})
// Rejected only once the Session is closing or failed, when nothing is kept anyway.
.catch(() => {});
}
/**
* Drop the queued nodes that are still unneeded: not live, nothing listed below. Runs once a publication is fully
* tracked, and as a job of its own on the line, never inside an operation: work a commit creates below a chain its
* own callback loaded keeps that chain. A flat loop: unlisting already queued every node on the way up that lost its
* last listed node.
*/
#sweep(): void {
for (const node of this.#dropQueue) {
if (this.#below.has(node)) continue;
if (typeof node === "string") {
this.#edges.delete(conversationOf(node));
this.#roots.delete(conversationOf(node));
} else if (!this.#live.has(node)) {
this.#settled.delete(node);
this.#unloaded.delete(node);
}
}
this.#dropQueue.clear();
}
/**
@ -793,13 +810,12 @@ export class TaskScheduler {
this.#release(node);
for (const waiter of this.#waiters.get(id) ?? []) this.#runnable.add(waiter);
this.#relist(id);
this.#forget(id);
this.#dropIfUnneeded(id);
return;
}
this.#live.set(id, record);
if (previous === undefined) {
if (!this.#known(parentNode(record))) this.#unloaded.add(id);
this.#countChild(record, 1);
this.#relist(id);
}
if (cancellationIntent(record)) this.#intent.add(id);
@ -823,9 +839,9 @@ export class TaskScheduler {
/**
* List `start` under its parent node in `#below` while it is a live task or has nodes listed below it, and unlist it
* otherwise; then its parent, as far up as that changes. So `#below` holds exactly the tree around live work; an
* ended parent that loses its last listed node may be forgotten (`#forget`). A loop, as ownership chains can be
* deeper than the call stack.
* otherwise; then its parent, as far up as that changes. So `#below` holds exactly the tree around live work; a
* parent that loses its last listed node is queued for the sweep. A loop, as ownership chains can be deeper than the
* call stack.
*/
#relist(start: Node): void {
let node: Node = start;
@ -854,8 +870,7 @@ export class TaskScheduler {
} else {
deleteFrom(this.#below, parent, node);
if (this.#below.has(parent)) return;
// An ended task with nothing listed below it any more may be forgotten.
if (typeof parent !== "string") this.#forget(parent);
this.#dropIfUnneeded(parent);
}
node = parent;
}
@ -1220,6 +1235,23 @@ export class TaskScheduler {
return { task, migrates: true };
}
/** Sizes of the ownership indexes, for tests that bound the scheduler's memory. */
get indexSizes(): Readonly<
Record<"live" | "settled" | "edges" | "below" | "roots" | "unloaded" | "dropQueue", number>
> {
let below = 0;
for (const children of this.#below.values()) below += children.size;
return {
live: this.#live.size,
settled: this.#settled.size,
edges: this.#edges.size,
below,
roots: this.#roots.size,
unloaded: this.#unloaded.size,
dropQueue: this.#dropQueue.size,
};
}
/**
* Scheduling state and every live task with its derived state, read on the Session line. Runs no task code: a
* pending migration shows as `ready` with `migrates`, and only a migration the scheduler already tried, or one that

582
packages/durable/test/harness-scheduler-retention.test.ts Normal file
View file

@ -0,0 +1,582 @@
// Scheduler memory is bounded by live work: an ended task or a conversation edge is kept only while live work is
// below it, and a chain dropped that way is loaded again from storage when work appears below it.
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import {
type Conversation,
defineTask,
type Harness,
LiveDoc,
type Storage,
type TaskId,
} from "@earendil-works/pi-durable";
import { afterEach, describe, expect, it } from "vitest";
import { SCHEDULER_INDEX_SIZES } from "../src/harness/harness.ts";
import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts";
import { context } from "./session-support.ts";
import { aborted, type Deferred, deferred, eventually, openTasks, settled } from "./task-support.ts";
const gates = new Map<string, Deferred<void>>();
function gate(name: string): Deferred<void> {
let found = gates.get(name);
if (found === undefined) {
found = deferred<void>();
gates.set(name, found);
}
return found;
}
/** How often each named task started its phase. */
const runs = new Map<string, number>();
/** Holds until its gate opens, then completes; its abort handler ends it `aborted`. */
const Hold = defineTask<{ name: string }, { phase: "hold" }, null>({
name: "test.retention.hold",
version: 1,
initial: () => ({ phase: "hold" }),
phases: {
hold: async (task, runtime, ctx) => {
runs.set(task.input.name, (runs.get(task.input.name) ?? 0) + 1);
const opened = await Promise.race([gate(task.input.name).promise.then(() => true), aborted(runtime.signal)]);
if (opened !== true) return;
await runtime.commit(() => ({ status: "terminal", outcome: { status: "completed", result: null } }), ctx);
},
},
abort: async (_task, runtime, ctx) =>
runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx),
});
/** Holds until its gate opens, then creates a held task `inside.<name>` in conversation `input.into` and ends. */
const Spawner = defineTask<{ name: string; into: number }, { phase: "hold" }, null>({
name: "test.retention.spawner",
version: 1,
initial: () => ({ phase: "hold" }),
phases: {
hold: async (task, runtime, ctx) => {
const opened = await Promise.race([gate(task.input.name).promise.then(() => true), aborted(runtime.signal)]);
if (opened !== true) return;
await runtime.commit(async (tx) => {
await tx.createTask(
Hold,
{ name: `inside.${task.input.name}` },
{ ownership: { kind: "conversation" }, conversationId: task.input.into as never },
);
return { status: "terminal", outcome: { status: "completed", result: null } };
}, ctx);
},
},
abort: async (_task, runtime, ctx) =>
runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx),
});
/** Rejected commits it caught. */
const rejections: unknown[] = [];
/**
* Holds until `rejected` opens, then tries one terminal commit that creates a task in conversation `into` and a child of
* its own, which the Transaction rejects as the task is ending. Catches the rejection and holds.
*/
const Rejected = defineTask<{ into: number }, { phase: "hold" }, null>({
name: "test.retention.rejected",
version: 1,
initial: () => ({ phase: "hold" }),
phases: {
hold: async (task, runtime, ctx) => {
const opened = await Promise.race([gate("rejected").promise.then(() => true), aborted(runtime.signal)]);
if (opened !== true) return;
try {
await runtime.commit(async (tx) => {
await tx.createTask(
Hold,
{ name: "staged" },
{ ownership: { kind: "conversation" }, conversationId: task.input.into as never },
);
await tx.createTask(Hold, { name: "child" }, { ownership: { kind: "task", taskId: task.id } });
return { status: "terminal", outcome: { status: "completed", result: null } };
}, ctx);
} catch (error) {
rejections.push(error);
}
await aborted(runtime.signal);
},
},
abort: async (_task, runtime, ctx) =>
runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx),
});
type Sizes = Record<"live" | "settled" | "edges" | "below" | "roots" | "unloaded" | "dropQueue", number>;
function sizes(harness: Harness): Sizes {
return (harness as unknown as Record<symbol, Sizes>)[SCHEDULER_INDEX_SIZES]!;
}
const directories: string[] = [];
afterEach(async () => {
gates.clear();
runs.clear();
rejections.length = 0;
for (const directory of directories.splice(0)) await rm(directory, { recursive: true, force: true });
});
async function sqlitePath(): Promise<string> {
const directory = await mkdtemp(join(tmpdir(), "pi-durable-retention-"));
directories.push(directory);
return join(directory, "session.sqlite");
}
/** Count the task and conversation reads a storage serves. */
function countReads(storage: Storage): { tasks: number; conversations: number } {
const reads = { tasks: 0, conversations: 0 };
const task = storage.task.bind(storage);
const conversation = storage.conversation.bind(storage);
storage.task = (id, readContext) => {
reads.tasks++;
return task(id, readContext);
};
storage.conversation = (id, readContext) => {
reads.conversations++;
return conversation(id, readContext);
};
return reads;
}
async function open(storage?: Storage) {
const opened = await openTasks(storage ?? (await openNodeSqliteStorage(await sqlitePath())), [
Hold,
Spawner,
Rejected,
]);
const root = await opened.harness.root(context);
opened.harness.resume();
return { ...opened, root };
}
async function status(harness: Harness, id: TaskId) {
return (await harness.getTask(id, context))!.state;
}
/** Under live `parent`: tool task T owning conversation C, a task in C; all of them end. Returns C. */
async function subagentCycle(harness: Harness, root: Conversation, parent: TaskId, n: number) {
const { tool, inner, child } = await root.commit(async (tx) => {
const tool = await tx.createTask(Hold, { name: `tool${n}` }, { ownership: { kind: "task", taskId: parent } });
const child = await tx.createConversation({ ownership: { kind: "task", taskId: tool } });
const inner = await tx.createTask(
Hold,
{ name: `inner${n}` },
{ ownership: { kind: "conversation" }, conversationId: child.id },
);
return { tool, inner, child: child.id };
}, context);
gate(`inner${n}`).resolve();
gate(`tool${n}`).resolve();
await harness.waitForTask(inner, context);
await harness.waitForTask(tool, context);
return child;
}
describe("scheduler retention", () => {
it("forgets ended subagent chains once their work ends", async () => {
const { harness, root } = await open();
const parent = await root.commit(
(tx) => tx.createTask(Hold, { name: "parent" }, { ownership: { kind: "conversation" } }),
context,
);
await subagentCycle(harness, root, parent, 0);
const baseline = sizes(harness);
for (let n = 1; n <= 200; n++) await subagentCycle(harness, root, parent, n);
expect(sizes(harness)).toEqual(baseline);
gate("parent").resolve();
await harness.waitForTask(parent, context);
await harness.close(context);
});
it("holds a live owner for work later created in a forgotten conversation, loading its chain again", async () => {
const storage = await openNodeSqliteStorage(await sqlitePath());
const reads = countReads(storage);
const { harness, root } = await open(storage);
// Live P -> ended B -> ended A -> conversation C, empty: forgotten.
const tree = await root.commit(async (tx) => {
const p = await tx.createTask(Hold, { name: "p" }, { ownership: { kind: "conversation" } });
const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "task", taskId: p } });
const a = await tx.createTask(Hold, { name: "a" }, { ownership: { kind: "task", taskId: b } });
const c = await tx.createConversation({ ownership: { kind: "task", taskId: a } });
return { p, b, a, c: c.id };
}, context);
gate("a").resolve();
gate("b").resolve();
await harness.waitForTask(tree.b, context);
const before = sizes(harness);
const loaded = reads.tasks + reads.conversations;
const c = (await harness.conversation(tree.c, context))!;
const d = await c.commit(
(tx) => tx.createTask(Hold, { name: "d" }, { ownership: { kind: "conversation" } }),
context,
);
gate("p").resolve();
await eventually(async () => (await status(harness, tree.p)).status === "completing");
expect(reads.tasks + reads.conversations).toBeGreaterThan(loaded);
gate("d").resolve();
expect((await harness.waitForTask(d, context)).state.outcome.status).toBe("completed");
expect((await harness.waitForTask(tree.p, context)).state.outcome.status).toBe("completed");
expect(sizes(harness).settled).toBeLessThanOrEqual(before.settled);
await harness.close(context);
});
it("holds an owner for work its own terminal commit creates below a forgotten chain", async () => {
const { harness, root } = await open();
// Live P (a spawner) -> ended B -> ended A -> conversation C, empty: forgotten. P's terminal commit creates D in C.
const tree = await root.commit(async (tx) => {
const p = await tx.createTask(Spawner, { name: "p", into: 0 }, { ownership: { kind: "conversation" } });
const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "task", taskId: p } });
const a = await tx.createTask(Hold, { name: "a" }, { ownership: { kind: "task", taskId: b } });
const c = await tx.createConversation({ ownership: { kind: "task", taskId: a } });
return { p, b, c: c.id };
}, context);
// Point the spawner at C, now that its ID is known.
await root.commit(async (tx) => {
const record = (await tx.task(tree.p))!;
(tx as unknown as { setTask(value: unknown): void }).setTask({
...record,
input: { name: "p", into: tree.c },
});
}, context);
gate("a").resolve();
gate("b").resolve();
await harness.waitForTask(tree.b, context);
gate("p").resolve();
await eventually(async () => (await status(harness, tree.p)).status !== "running");
expect((await status(harness, tree.p)).status).toBe("completing");
gate("inside.p").resolve();
expect((await harness.waitForTask(tree.p, context)).state.outcome.status).toBe("completed");
await harness.close(context);
});
it("withdraws an input queued in a forgotten conversation below an abort-marked owner", async () => {
const { harness, root } = await open();
// Live P -> ended B -> conversation C with no task; C gets a queued input while P is cancelling.
const tree = await root.commit(async (tx) => {
const p = await tx.createTask(Hold, { name: "p" }, { ownership: { kind: "conversation" } });
const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "task", taskId: p } });
const c = await tx.createConversation({ ownership: { kind: "task", taskId: b } });
// A live child keeps P holding its abort until the input is withdrawn and the child ends.
const keep = await tx.createTask(Hold, { name: "keep" }, { ownership: { kind: "task", taskId: p } });
return { p, b, c: c.id, keep };
}, context);
gate("b").resolve();
await harness.waitForTask(tree.b, context);
const c = (await harness.conversation(tree.c, context))!;
// Busy, so the input queues instead of starting a run.
await c.commit(async (tx) => {
(await tx.doc(LiveDoc, tree.c)).run = { taskId: tree.keep, inputs: [] };
}, context);
const queued = (await c.submit({ type: "input", content: "later" }, context)).id;
await harness.abortTask(tree.p, context);
await eventually(async () => {
const settledState = await (await harness.submission(queued, context))!.status(context);
return settledState.status === "unanswered";
});
expect(await (await harness.submission(queued, context))!.status(context)).toMatchObject({ reason: "aborted" });
await harness.waitForTask(tree.p, context);
await harness.close(context);
});
it("aborts work created in a forgotten conversation below an abort-marked owner before it runs", async () => {
const { harness, root } = await open();
const tree = await root.commit(async (tx) => {
const p = await tx.createTask(Hold, { name: "p" }, { ownership: { kind: "conversation" } });
const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "task", taskId: p } });
const c = await tx.createConversation({ ownership: { kind: "task", taskId: b } });
const keep = await tx.createTask(Hold, { name: "keep" }, { ownership: { kind: "task", taskId: p } });
return { p, b, c: c.id, keep };
}, context);
gate("b").resolve();
await harness.waitForTask(tree.b, context);
// P cancels; its live child keeps it holding.
await harness.commit(async (tx) => {
const record = (await tx.task(tree.p))!;
(tx as unknown as { setTask(value: unknown): void }).setTask({ ...record, abortRequested: true });
}, context);
await eventually(async () => (await harness.getTask(tree.keep, context))!.abortRequested);
const c = (await harness.conversation(tree.c, context))!;
const late = await c.commit(
(tx) => tx.createTask(Hold, { name: "late" }, { ownership: { kind: "conversation" } }),
context,
);
expect((await harness.waitForTask(late, context)).state.outcome.status).toBe("aborted");
expect(runs.get("late") ?? 0).toBe(0);
await harness.close(context);
});
it("keeps a scope busy while the chain of new work below a forgotten conversation loads", async () => {
const path = await sqlitePath();
const storage = await openNodeSqliteStorage(path);
const { harness, root } = await open(storage);
const tree = await root.commit(async (tx) => {
const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "conversation" } });
const c = await tx.createConversation({ ownership: { kind: "task", taskId: b } });
return { b, c: c.id };
}, context);
gate("b").resolve();
await harness.waitForTask(tree.b, context);
// Hold the chain load's read of C: the second read once holding, after the creating commit's own.
const read = storage.conversation.bind(storage);
const held = deferred<void>();
const reached = deferred<void>();
let holding = false;
let readsOfC = 0;
storage.conversation = async (id, readContext) => {
if (holding && id === tree.c && ++readsOfC === 2) {
reached.resolve();
await held.promise;
}
return read(id, readContext);
};
const c = (await harness.conversation(tree.c, context))!;
holding = true;
await c.commit((tx) => tx.createTask(Hold, { name: "d" }, { ownership: { kind: "conversation" } }), context);
await reached.promise;
expect(await settled(harness.waitForIdle(context))).toBe(false);
held.resolve();
gate("d").resolve();
await harness.waitForIdle(context);
await harness.close(context);
});
it("keeps a chain when one commit ends its last task and creates new work below it", async () => {
const { harness, root } = await open();
// Live P -> ended B -> conversation C with background task `old`, which keeps the chain known without holding B.
const tree = await root.commit(async (tx) => {
const p = await tx.createTask(Hold, { name: "p" }, { ownership: { kind: "conversation" } });
const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "task", taskId: p } });
const c = await tx.createConversation({ ownership: { kind: "task", taskId: b } });
const old = await tx.createTask(
Hold,
{ name: "old" },
{ ownership: { kind: "conversation" }, conversationId: c.id, background: true },
);
return { p, b, c: c.id, old };
}, context);
gate("b").resolve();
await harness.waitForTask(tree.b, context);
// One commit: the old task ends, which leaves nothing live below B, and a replacement starts in C.
const replacement = await harness.commit(async (tx) => {
const old = (await tx.task(tree.old))!;
const id = await tx.createTask(
Hold,
{ name: "new" },
{ ownership: { kind: "conversation" }, conversationId: tree.c },
);
(tx as unknown as { setTask(value: unknown): void }).setTask({
...old,
state: { status: "terminal", outcome: { status: "completed", result: null } },
});
return id;
}, context);
// The chain stayed known through the publication, so P holds for the replacement.
gate("p").resolve();
await eventually(async () => (await status(harness, tree.p)).status === "completing");
gate("new").resolve();
await harness.waitForTask(replacement, context);
expect((await harness.waitForTask(tree.p, context)).state.outcome.status).toBe("completed");
await harness.close(context);
});
it("drops what a rejected commit's chain load read, without a later publication", async () => {
const { harness, root } = await open();
// Live P owns ended B, which owns conversation C: forgotten. P's terminal commit stages work in C, which loads C's
// chain, and a child of P, which the Transaction rejects, as P is ending.
const tree = await root.commit(async (tx) => {
const p = await tx.createTask(Rejected, { into: 0 }, { ownership: { kind: "conversation" } });
const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "task", taskId: p } });
const c = await tx.createConversation({ ownership: { kind: "task", taskId: b } });
return { p, b, c: c.id };
}, context);
await root.commit(async (tx) => {
const record = (await tx.task(tree.p))!;
(tx as unknown as { setTask(value: unknown): void }).setTask({ ...record, input: { into: tree.c } });
}, context);
gate("b").resolve();
await harness.waitForTask(tree.b, context);
const baseline = sizes(harness);
gate("rejected").resolve();
await eventually(() => rejections.length > 0);
// The rejected commit published nothing; the sweep job still cleans up what its chain load read.
await eventually(() => sizes(harness).dropQueue === 0);
expect(sizes(harness)).toEqual(baseline);
await harness.close(context);
});
it("drops what a read loaded after reopen, without a later publication", async () => {
const path = await sqlitePath();
let opened = await open(await openNodeSqliteStorage(path));
// A live task D in conversation C, owned by ended B: after reopen D's chain is not loaded.
const tree = await opened.root.commit(async (tx) => {
const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "conversation" } });
const c = await tx.createConversation({ ownership: { kind: "task", taskId: b } });
const d = await tx.createTask(
Hold,
{ name: "d" },
{ ownership: { kind: "conversation" }, conversationId: c.id },
);
return { b, d };
}, context);
gate("b").resolve();
await eventually(async () => (await status(opened.harness, tree.b)).status === "completing");
await opened.harness.close(context);
gates.clear();
opened = await open(await openNodeSqliteStorage(path));
await opened.harness.inspect(context);
// D is still live, so its chain stays; once D ends, everything below the root goes, published or not.
expect(sizes(opened.harness).unloaded).toBe(0);
gate("d").resolve();
await opened.harness.waitForTask(tree.d, context);
await opened.harness.waitForTask(tree.b, context);
await eventually(() => sizes(opened.harness).dropQueue === 0);
expect(sizes(opened.harness)).toMatchObject({ settled: 0, below: 0, unloaded: 0 });
await opened.harness.close(context);
});
it("withdraws an input queued two ended levels below an owner whose intent appears later", async () => {
const { harness, root } = await open();
// Live P -> ended B -> ended A -> conversation C with no task, and a live child keeping P holding.
const tree = await root.commit(async (tx) => {
const p = await tx.createTask(Hold, { name: "p" }, { ownership: { kind: "conversation" } });
const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "task", taskId: p } });
const a = await tx.createTask(Hold, { name: "a" }, { ownership: { kind: "task", taskId: b } });
const c = await tx.createConversation({ ownership: { kind: "task", taskId: a } });
const keep = await tx.createTask(Hold, { name: "keep" }, { ownership: { kind: "task", taskId: p } });
return { p, b, c: c.id, keep };
}, context);
gate("a").resolve();
gate("b").resolve();
await harness.waitForTask(tree.b, context);
const c = (await harness.conversation(tree.c, context))!;
await c.commit(async (tx) => {
(await tx.doc(LiveDoc, tree.c)).run = { taskId: tree.keep, inputs: [] };
}, context);
// Queued while nothing cancels: no withdrawal scan.
const queued = (await c.submit({ type: "input", content: "later" }, context)).id;
await eventually(() => sizes(harness).dropQueue === 0);
// P fails while its child runs: it holds the failed outcome, which is cancellation intent.
await harness.commit(async (tx) => {
const record = (await tx.task(tree.p))!;
(tx as unknown as { setTask(value: unknown): void }).setTask({
...record,
state: { status: "completing", outcome: { status: "failed", error: { message: "x" } } },
});
}, context);
await eventually(
async () => (await (await harness.submission(queued, context))!.status(context)).status === "unanswered",
);
expect(await (await harness.submission(queued, context))!.status(context)).toMatchObject({ reason: "aborted" });
await harness.close(context);
});
it("forgets a 20,000-deep chain of ended tasks without recursion", async () => {
const { harness, root } = await open();
const depth = 20_000;
const baseline = sizes(harness);
const { links, leaf } = await root.commit(async (tx) => {
const links: TaskId[] = [];
let owner = await tx.createTask(Hold, { name: "link0" }, { ownership: { kind: "conversation" } });
links.push(owner);
for (let n = 1; n < depth; n++) {
owner = await tx.createTask(Hold, { name: `link${n}` }, { ownership: { kind: "task", taskId: owner } });
links.push(owner);
}
const leaf = await tx.createTask(Hold, { name: "leaf" }, { ownership: { kind: "task", taskId: owner } });
return { links, leaf };
}, context);
for (let n = 0; n < depth; n++) gate(`link${n}`).resolve();
gate("leaf").resolve();
await harness.waitForTask(leaf, context);
await harness.waitForTask(links[0]!, context);
await harness.waitForIdle(context);
expect(sizes(harness)).toEqual(baseline);
await harness.close(context);
}, 60_000);
it("keeps no more after all work ended than a fresh reopen", async () => {
const path = await sqlitePath();
let opened = await open(await openNodeSqliteStorage(path));
const parent = await opened.root.commit(
(tx) => tx.createTask(Hold, { name: "parent" }, { ownership: { kind: "conversation" } }),
context,
);
for (let n = 0; n < 20; n++) await subagentCycle(opened.harness, opened.root, parent, n);
gate("parent").resolve();
await opened.harness.waitForTask(parent, context);
await opened.harness.waitForIdle(context);
const ended = sizes(opened.harness);
await opened.harness.close(context);
opened = await open(await openNodeSqliteStorage(path));
await opened.harness.waitForIdle(context);
expect(ended).toEqual(sizes(opened.harness));
await opened.harness.close(context);
});
it("does not scan queued submissions for an input when nothing cancels", async () => {
const storage = await openNodeSqliteStorage(await sqlitePath());
let scans = 0;
const scan = storage.scanSubmissions.bind(storage);
storage.scanSubmissions = (query, limit, cursor, scanContext) => {
if (query.status === "queued") scans++;
return scan(query, limit, cursor, scanContext);
};
const { harness, root } = await open(storage);
const busy = await root.commit(
(tx) => tx.createTask(Hold, { name: "busy" }, { ownership: { kind: "conversation" } }),
context,
);
await root.commit(async (tx) => {
(await tx.doc(LiveDoc, root.id)).run = { taskId: busy, inputs: [] };
}, context);
const before = scans;
await root.submit({ type: "input", content: "queued" }, context);
// Let any reconcile it would cause run.
await new Promise((resolve) => setTimeout(resolve, 20));
expect(scans).toBe(before);
gate("busy").resolve();
await harness.close(context);
});
it("still rejects waiting on a forgotten ended owner", async () => {
const opened = await openTasks(await openNodeSqliteStorage(await sqlitePath()), [Hold, Waiter]);
const root = await opened.harness.root(context);
opened.harness.resume();
const tree = await root.commit(async (tx) => {
const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "conversation" } });
const c = await tx.createConversation({ ownership: { kind: "task", taskId: b } });
return { b, c: c.id };
}, context);
gate("b").resolve();
await opened.harness.waitForTask(tree.b, context);
const c = (await opened.harness.conversation(tree.c, context))!;
const waiter = await c.commit(
(tx) => tx.createTask(Waiter, { on: [tree.b] }, { ownership: { kind: "conversation" } }),
context,
);
const outcome = (await opened.harness.waitForTask(waiter, context)).state.outcome;
expect(outcome).toMatchObject({ status: "faulted" });
expect(JSON.stringify(outcome)).toMatch(/cannot wait on itself or its owner/);
await opened.harness.close(context);
});
});
/** Waits on the tasks in its input, then completes. */
const Waiter = defineTask<{ on: TaskId[] }, { phase: "wait" } | { phase: "done" }, null>({
name: "test.retention.waiter",
version: 1,
initial: () => ({ phase: "wait" }),
phases: {
wait: async (task, runtime, ctx) =>
runtime.commit(
() => ({ status: "waiting", checkpoint: { phase: "done" }, on: task.input.on, policy: "allSettled" }),
ctx,
),
done: async (_task, runtime, ctx) =>
runtime.commit(() => ({ status: "terminal", outcome: { status: "completed", result: null } }), ctx),
},
abort: async (_task, runtime, ctx) =>
runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx),
});

134
packages/durable/test/scheduler-memory-bench.ts Normal file
View file

@ -0,0 +1,134 @@
/**
* Scheduler memory benchmark. Under one live parent task, N subagent cycles run and end: a tool task owned by the
* parent, a conversation owned by the tool task, and a task in that conversation. Reports the scheduler's ownership
* index sizes and the heap after GC once all cycles ended, then the storage reads a later message into an old subagent
* conversation costs. On SQLite, so durable history stays out of the heap.
*
* node --conditions=source --experimental-strip-types --expose-gc test/scheduler-memory-bench.ts [cycles...]
*/
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { BACKGROUND_CONTEXT as context } from "@earendil-works/chord/context";
import { createModels } from "@earendil-works/pi-ai/models";
import { SCHEDULER_INDEX_SIZES } from "../src/harness/harness.ts";
import {
type ConversationId,
createRegistry,
defineExtension,
defineTask,
Harness,
type TaskId,
} from "../src/index.ts";
import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts";
const gc = (globalThis as { gc?: () => void }).gc;
if (gc === undefined) throw new Error("Run with --expose-gc");
const waiting = new Map<string, () => void>();
/** Holds until released by name, then completes. */
const Hold = defineTask<{ name: string }, { phase: "hold" }, null>({
name: "bench.hold",
version: 1,
initial: () => ({ phase: "hold" }),
phases: {
hold: async (task, runtime, ctx) => {
await new Promise<void>((resolve) => waiting.set(task.input.name, resolve));
await runtime.commit(() => ({ status: "terminal", outcome: { status: "completed", result: null } }), ctx);
},
},
abort: async (_task, runtime, ctx) =>
runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx),
});
async function release(name: string): Promise<void> {
while (!waiting.has(name)) await new Promise((resolve) => setTimeout(resolve, 0));
waiting.get(name)!();
waiting.delete(name);
}
type Sizes = Record<"live" | "settled" | "edges" | "below" | "roots" | "unloaded" | "dropQueue", number>;
async function run(cycles: number) {
const directory = mkdtempSync(join(tmpdir(), "pi-durable-memory-bench-"));
try {
const storage = await openNodeSqliteStorage(join(directory, "session.sqlite"));
const reads = { count: 0 };
const task = storage.task.bind(storage);
const conversation = storage.conversation.bind(storage);
storage.task = (id, readContext) => {
reads.count++;
return task(id, readContext);
};
storage.conversation = (id, readContext) => {
reads.count++;
return conversation(id, readContext);
};
const registry = createRegistry();
registry.install(defineExtension({ name: "bench", tasks: [Hold] }));
const harness = await Harness.open(storage, { models: createModels(), registry }, context);
const root = await harness.root(context);
harness.resume();
const parent = await root.commit(
(tx) => tx.createTask(Hold, { name: "parent" }, { ownership: { kind: "conversation" } }),
context,
);
gc!();
const heapBefore = process.memoryUsage().heapUsed;
let first: ConversationId | undefined;
const started = performance.now();
for (let n = 0; n < cycles; n++) {
const { tool, inner, child } = await root.commit(async (tx) => {
const tool = await tx.createTask(
Hold,
{ name: `tool${n}` },
{ ownership: { kind: "task", taskId: parent } },
);
const child = await tx.createConversation({ ownership: { kind: "task", taskId: tool } });
const inner = await tx.createTask(
Hold,
{ name: `inner${n}` },
{ ownership: { kind: "conversation" }, conversationId: child.id },
);
return { tool, inner, child: child.id };
}, context);
await release(`inner${n}`);
await release(`tool${n}`);
await harness.waitForTask(inner as TaskId, context);
await harness.waitForTask(tool as TaskId, context);
first ??= child;
}
const ms = Math.round(performance.now() - started);
gc!();
const heapAfter = process.memoryUsage().heapUsed;
const sizes = (harness as unknown as Record<symbol, Sizes>)[SCHEDULER_INDEX_SIZES]!;
// A later message into an old subagent conversation: a task created there, run, and ended.
const readsBefore = reads.count;
const old = (await harness.conversation(first!, context))!;
const late = await old.commit(
(tx) => tx.createTask(Hold, { name: "late" }, { ownership: { kind: "conversation" } }),
context,
);
await release("late");
await harness.waitForTask(late, context);
const lateReads = reads.count - readsBefore;
await release("parent");
await harness.waitForTask(parent, context);
await harness.close(context);
return { ms, sizes, heap: heapAfter - heapBefore, lateReads };
} finally {
rmSync(directory, { recursive: true, force: true });
}
}
const cycles = process.argv.slice(2).map(Number);
console.log("cycles | ms | settled | edges | below | heap growth | reads for a later message");
for (const n of cycles.length > 0 ? cycles : [1000, 10000]) {
const { ms, sizes, heap, lateReads } = await run(n);
console.log(
`${String(n).padEnd(6)} | ${String(ms).padEnd(6)} | ${String(sizes.settled).padEnd(7)} | ${String(sizes.edges).padEnd(6)} | ${String(sizes.below).padEnd(5)} | ${`${(heap / 1024 / 1024).toFixed(1)} MiB, ${Math.round(heap / n)} B/cycle`.padEnd(23)} | ${lateReads}`,
);
// The scheduler's ownership indexes must not grow with ended history; the heap still includes the Session's caches.
if (sizes.settled > 0 || sizes.edges > 1)
throw new Error(`ownership indexes grew with history: ${JSON.stringify(sizes)}`);
}