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

feat(durable): scan conversations, entries, tasks, and submissions in either order

Browse source 
ScanOrder is an optional order on each scan query; cursors carry it.

closes #10546
This commit is contained in:
Mario Zechner 2026-10-06 20:02:59 +02:00
parent 43d3763991
commit 4dd2af42c8
12 changed files with 329 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/durable/CHANGELOG.md
View file

@ -4,12 +4,14 @@
### Breaking Changes
- `Storage` implementations must honor the new `order` field of `ConversationQuery`, `EntryQuery`, `TaskQuery`, and `SubmissionQuery`, and continue a cursor in the order it was returned with. A storage that ignores it returns pages in the wrong direction; the conformance suite covers both orders ([#10546](https://github.com/earendil-works/pi/issues/10546)).
- `TaskRuntime.context()` takes the cutoff as an options object, matching `Conversation.context()`: `runtime.context(conversationId, context, { at })` replaces `runtime.context(conversationId, context, at)`.
### Added
- `ToolExecutionApi.models` and `HookApi.models`: the Harness's `models`, so tools and hooks can resolve models and make requests with the same catalog, credentials, and request transforms as generation ([#10395](https://github.com/earendil-works/pi/issues/10395)).
- `Conversation.context(context, { at })` returns the model context as of a visible earlier entry, the same view `fork(at)` would start with, without creating a conversation ([#10512](https://github.com/earendil-works/pi/issues/10512)).
- `ScanOrder` (`ascending` or `descending`) as `order` on conversation, entry, task, and submission scans, including `Tx.scanTasks()` and `Conversation.entries()`, so hosts can page the newest tasks first. Defaults are unchanged: entries newest first, everything else oldest first. A cursor continues in its scan's order, so the query may omit `order`; asking for the other order with it throws ([#10546](https://github.com/earendil-works/pi/issues/10546)).
- Task records carry `startedAt` and `endedAt`, wall-clock times the Session stamps at the first change to `running` and at the change to `terminal`, with `HarnessOptions.now` or the new `createSession(storage, { now })` option. `startedAt` survives waits and reopen. `inspect()` shows them on each live task's record. Records written by earlier versions have neither ([#10549](https://github.com/earendil-works/pi/issues/10549)).
- `pi.tool-result` messages carry `durationMs`: how long `execute()` took in that attempt, measured with a monotonic clock. Calls that did not execute, and interrupted or aborted calls, have none ([#10549](https://github.com/earendil-works/pi/issues/10549)).
- Assistant messages carry the `durationMs` pi-ai measures for each response ([#10549](https://github.com/earendil-works/pi/issues/10549)).

20
packages/durable/docs/spec.md
View file

@ -4312,15 +4312,20 @@ type Page<T, C> = {
type Cursor = Readonly<Record<string, JsonValue>>;
/** ID order: `ascending` is oldest first, `descending` newest first. */
type ScanOrder = "ascending" | "descending";
type ConversationQuery = {
readonly ownerConversationId?: ConversationId;
readonly ownerTaskId?: TaskId;
readonly order?: ScanOrder; // default ascending
};
type EntryQuery = {
readonly conversationId: ConversationId;
readonly minEntryId?: EntryId; // inclusive
readonly maxEntryId?: EntryId; // inclusive
readonly order?: ScanOrder; // default descending
};
type TaskQuery = {
@ -4329,11 +4334,13 @@ type TaskQuery = {
readonly status?: "pending" | "running" | "waiting" | "completing" | "terminal";
readonly abortRequested?: boolean;
readonly background?: boolean;
readonly order?: ScanOrder; // default ascending
};
type SubmissionQuery = {
readonly conversationId?: ConversationId;
readonly status?: SubmissionRecord["status"];
readonly order?: ScanOrder; // default ascending
};
type DocumentPoint = Seq | "current";
@ -4433,15 +4440,22 @@ not create, change, or retire a selected source. Later source changes,
reclamation, retirement, or backend reopen cannot affect the child.
Cursors are backend-owned JSON objects. Callers only round-trip them to the same
scan on the same storage; cross-storage or cross-query use is unsupported. The
scan on the same storage; cross-storage or cross-query use is unsupported.
Conversation, entry, task, and submission scans run in their query's `order`,
by ID. A cursor carries the order of the scan that returned it: a scan given a
cursor continues in that order whether its query repeats `order` or omits it,
and rejects a query that asks for the other order. Storage implementations
must honor `order`; one that ignored it would return pages in the wrong
direction without an error. The
Session owns the mutation line, so storage implementations do not add a second
caller-facing commit mutex. Each backend still makes one admitted batch atomic.
`findLatestHeadMarker()` returns the newest visible entry carrying `head` at or
below its optional inclusive cutoff. The returned entry is the marker; its
`head` value is the actual lower bound for context. `scanEntries()` pages the
inclusive ID range in newest-first order while applying every conversation
ancestry cap. With no bounds it pages complete visible history. To read context
inclusive ID range, newest first by default, while applying every conversation
ancestry cap; oldest first, it reads the root's segment first and then each
fork's. With no bounds it pages complete visible history. To read context
through entry `E`, find the marker at or before `E`, then scan from
`marker?.head` through `E`. For current context the upper bound is omitted.
Conversation owner filters are indexed and conjunctive. They support ownership

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

@ -134,6 +134,7 @@ class ConversationImpl<Tool extends ToolRegistration> implements Conversation {
conversationId: this.id,
...(query.minEntryId === undefined ? {} : { minEntryId: query.minEntryId }),
...(query.maxEntryId === undefined ? {} : { maxEntryId: query.maxEntryId }),
...(query.order === undefined ? {} : { order: query.order }),
};
return this.#host.harness.readOnLine(() => this.#host.storage.scanEntries(bounded, limit, cursor, context));
}

2
packages/durable/src/harness/types.ts
View file

@ -536,7 +536,7 @@ export interface Conversation {
* `fork(at)` would start with, without creating a conversation.
*/
context(context: Context, options?: { readonly at?: EntryId }): Promise<ContextView>;
/** Newest-first fork-aware history of this conversation. */
/** Fork-aware history of this conversation, newest first unless `query.order` is `ascending`. */
entries(
query: Omit<EntryQuery, "conversationId">,
limit: number,

1
packages/durable/src/index.ts
View file

@ -149,6 +149,7 @@ export type {
RewindableConversationDocToken,
RewindableConversationSemantics,
RunningTask,
ScanOrder,
Seq,
Session,
SessionDocFamilyToken,

96
packages/durable/src/storage/memory.ts
View file

@ -20,6 +20,7 @@ import type {
Id,
JsonObject,
Page,
ScanOrder,
Seq,
Storage,
StorageWrite,
@ -31,6 +32,7 @@ import type {
TaskQuery,
TaskRecord,
} from "../types.ts";
import { nextCursor, type ScanStart, scanStart } from "./scan.ts";
type StoredTask = TaskRecord<JsonValue, JsonValue, JsonValue>;
type TaskStatus = StoredTask["state"]["status"];
@ -196,12 +198,27 @@ const isCurrentOnly = (record: DocumentRecord): boolean =>
const tableContaining = (state: State, id: Id<string>): TableName | undefined => state.recordTypes.get(id);
const page = <T extends { readonly id: Id<string> }>(values: readonly T[], limit: number): Page<T, Cursor> => {
const page = <T extends { readonly id: Id<string> }>(
values: readonly T[],
limit: number,
order: ScanOrder = "ascending",
): Page<T, Cursor> => {
const items = values.slice(0, limit);
if (values.length <= limit) return { items: clone(items) };
return { items: clone(items), next: { after: items.at(-1)!.id } };
return { items: clone(items), next: nextCursor(items.at(-1)!.id, order) };
};
/** Indexes of sorted `ids` in scan order, after the cursor's ID when there is one. */
function* scanIndexes(ids: readonly Id<string>[], start: ScanStart): Generator<number> {
if (start.order === "ascending") {
const first = start.after === undefined ? 0 : upperBound(ids, start.after);
for (let index = first; index < ids.length; index++) yield index;
} else {
const end = start.after === undefined ? ids.length : lowerBound(ids, start.after);
for (let index = end - 1; index >= 0; index--) yield index;
}
}
/** A fully validated, detached state mutation whose application performs no fallible preparation. */
export interface PreparedMemoryCommit {
readonly seq: Seq;
@ -432,17 +449,17 @@ export class MemoryStorage implements Storage {
: query.ownerConversationId !== undefined
? (this.state.conversationIdsByOwnerConversation.get(query.ownerConversationId) ?? [])
: this.state.conversationIds;
const after = cursorId(cursor);
const start = after === undefined ? 0 : upperBound(ids, after);
const start = scanStart(query.order, cursor, "ascending");
const values: ConversationRecord[] = [];
for (let index = start; index < ids.length && values.length <= limit; index++) {
for (const index of scanIndexes(ids, start)) {
if (values.length > limit) break;
const value = this.state.conversations.get(ids[index]!)!;
if (query.ownerConversationId !== undefined && value.owner?.conversationId !== query.ownerConversationId) {
continue;
}
values.push(value);
}
return page(values, limit);
return page(values, limit, start.order);
}
entry(id: EntryId, context: Context): Promise<{ readonly entry: EntryRecord; readonly commitSeq: Seq } | undefined>;
@ -503,15 +520,22 @@ export class MemoryStorage implements Storage {
_context: Context,
): Promise<Page<EntryRecord, Cursor>> {
this.assertOpen();
const after = cursorId(cursor);
const maxEntryId =
after === undefined ? query.maxEntryId : Math.min(query.maxEntryId ?? Number.POSITIVE_INFINITY, after - 1);
const { order, after } = scanStart(query.order, cursor, "descending");
// The cursor narrows the bound on the side the scan moves away from.
let minEntryId: number | undefined = query.minEntryId;
let maxEntryId: number | undefined = query.maxEntryId;
if (after !== undefined && order === "descending") maxEntryId = Math.min(maxEntryId ?? after, after - 1);
if (after !== undefined && order === "ascending") minEntryId = Math.max(minEntryId ?? after, after + 1);
const visible: EntryRecord[] = [];
for (const entry of this.visibleEntries(query.conversationId, query.minEntryId, maxEntryId)) {
const entries =
order === "descending"
? this.visibleEntries(query.conversationId, minEntryId, maxEntryId)
: this.visibleEntriesAscending(query.conversationId, minEntryId, maxEntryId);
for (const entry of entries) {
visible.push(entry);
if (visible.length > limit) break;
}
return page(visible, limit);
return page(visible, limit, order);
}
async task(id: TaskId, _context: Context): Promise<StoredTask | undefined> {
@ -527,19 +551,19 @@ export class MemoryStorage implements Storage {
_context: Context,
): Promise<Page<StoredTask, Cursor>> {
this.assertOpen();
const after = cursorId(cursor);
const start = scanStart(query.order, cursor, "ascending");
const ids = query.status === undefined ? this.state.taskIds : this.state.taskIdsByStatus[query.status];
const start = after === undefined ? 0 : upperBound(ids, after);
const values: StoredTask[] = [];
for (let index = start; index < ids.length && values.length <= limit; index++) {
const value = this.state.tasks.get(ids[index])!;
for (const index of scanIndexes(ids, start)) {
if (values.length > limit) break;
const value = this.state.tasks.get(ids[index]!)!;
if (query.conversationId !== undefined && value.conversationId !== query.conversationId) continue;
if (query.kind !== undefined && value.kind !== query.kind) continue;
if (query.abortRequested !== undefined && value.abortRequested !== query.abortRequested) continue;
if (query.background !== undefined && value.background !== query.background) continue;
values.push(value);
}
return page(values, limit);
return page(values, limit, start.order);
}
async submission(id: SubmissionId, _context: Context): Promise<SubmissionRecord | undefined> {
@ -555,17 +579,17 @@ export class MemoryStorage implements Storage {
_context: Context,
): Promise<Page<SubmissionRecord, Cursor>> {
this.assertOpen();
const after = cursorId(cursor);
const start = scanStart(query.order, cursor, "ascending");
const ids =
query.status === undefined ? this.state.submissionIds : this.state.submissionIdsByStatus[query.status];
const start = after === undefined ? 0 : upperBound(ids, after);
const values: SubmissionRecord[] = [];
for (let index = start; index < ids.length && values.length <= limit; index++) {
const value = this.state.submissions.get(ids[index])!;
for (const index of scanIndexes(ids, start)) {
if (values.length > limit) break;
const value = this.state.submissions.get(ids[index]!)!;
if (query.conversationId !== undefined && value.conversationId !== query.conversationId) continue;
values.push(value);
}
return page(values, limit);
return page(values, limit, start.order);
}
async submissionByRequest(
@ -676,6 +700,36 @@ export class MemoryStorage implements Storage {
}
}
/** Visible entries oldest first: the fork chain's segments from the root conversation forward. */
private *visibleEntriesAscending(
conversationId: ConversationId,
minEntryId: number = Number.NEGATIVE_INFINITY,
maxEntryId: number = Number.POSITIVE_INFINITY,
): Generator<EntryRecord> {
if (!this.state.conversations.has(conversationId)) {
throw new Error(`Unknown conversation: ${conversationId}`);
}
const segments: { readonly conversationId: ConversationId; readonly upper: number }[] = [];
let currentId = conversationId;
let upperEntryId = maxEntryId;
while (true) {
segments.push({ conversationId: currentId, upper: upperEntryId });
const conversation = this.state.conversations.get(currentId)!;
if (conversation.parent === undefined) break;
upperEntryId = Math.min(upperEntryId, conversation.parent.at);
if (upperEntryId < minEntryId) break;
currentId = conversation.parent.conversationId;
}
for (const segment of segments.reverse()) {
const ids = this.state.entryIds.get(segment.conversationId) ?? [];
for (let index = lowerBound(ids, minEntryId); index < ids.length; index++) {
const id = ids[index]!;
if (id > segment.upper) break;
yield this.state.entries.get(id)!;
}
}
}
private checkGlobalIds(writes: readonly StorageWrite[]): void {
const claimed = new Map<Id<string>, TableName>();
for (const write of writes) {

38
packages/durable/src/storage/scan.ts Normal file
View file

@ -0,0 +1,38 @@
import type { Id, ScanOrder } from "../types.ts";
/** Where a built-in storage scan starts: its order and the last ID a previous page returned. */
export type ScanStart = {
readonly order: ScanOrder;
readonly after: number | undefined;
};
/**
* Resolve a scan's order and position. A cursor continues in the order it was created with, whether the query repeats
* that order or omits it; a different `order` throws. A cursor without an order, written before scans had one,
* continues in the scan's default order.
*/
export function scanStart(
requested: ScanOrder | undefined,
cursor: Readonly<Record<string, unknown>> | undefined,
fallback: ScanOrder,
): ScanStart {
if (requested !== undefined && requested !== "ascending" && requested !== "descending") {
throw new TypeError(`Invalid scan order: ${String(requested)}`);
}
if (cursor === undefined) return { order: requested ?? fallback, after: undefined };
const { after, order: stored } = cursor;
if (typeof after !== "number" || !Number.isSafeInteger(after)) throw new TypeError("Invalid storage cursor");
if (stored !== undefined && stored !== "ascending" && stored !== "descending") {
throw new TypeError("Invalid storage cursor");
}
const order = stored ?? fallback;
if (requested !== undefined && requested !== order) {
throw new TypeError(`The cursor continues a ${order} scan; the query asks for ${requested}`);
}
return { order, after };
}
/** The continuation of a scan whose last returned item has `id`. */
export function nextCursor(id: Id<string>, order: ScanOrder): { readonly after: number; readonly order: ScanOrder } {
return { after: id, order };
}

90
packages/durable/src/storage/sqlite/storage.ts
View file

@ -20,6 +20,7 @@ import type {
Id,
JsonObject,
Page,
ScanOrder,
Seq,
Storage,
StorageWrite,
@ -31,6 +32,7 @@ import type {
TaskQuery,
TaskRecord,
} from "../../types.ts";
import { nextCursor, type ScanStart, scanStart } from "../scan.ts";
import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "./database.ts";
import { applySqliteMigrations } from "./migrations.ts";
@ -70,12 +72,22 @@ const cursorId = <I extends Id<string>>(cursor: Cursor | undefined): I | undefin
return idFromNumber<I>(after);
};
const page = <T extends { readonly id: Id<string> }>(values: readonly T[], limit: number): Page<T, Cursor> => {
const page = <T extends { readonly id: Id<string> }>(
values: readonly T[],
limit: number,
order: ScanOrder = "ascending",
): Page<T, Cursor> => {
const items = values.slice(0, limit);
if (values.length <= limit) return { items };
return { items, next: { after: items.at(-1)!.id } };
return { items, next: nextCursor(items.at(-1)!.id, order) };
};
/** The ID condition, its parameter, and the `ORDER BY` direction of a table scan. */
const scanSql = (start: ScanStart): { readonly clause: string; readonly param: number; readonly direction: string } =>
start.order === "ascending"
? { clause: "id > ?", param: start.after ?? -1, direction: "ASC" }
: { clause: "id < ?", param: start.after ?? Number.MAX_SAFE_INTEGER, direction: "DESC" };
const scopeColumns = (scope: DocumentRecord["scope"]): ScopeColumns => {
switch (scope.kind) {
case "session":
@ -203,8 +215,10 @@ export class SqliteStorage implements Storage {
_context: Context,
): Promise<Page<ConversationRecord, Cursor>> {
this.assertOpen();
const clauses = ["id > ?"];
const params: SqliteValue[] = [cursorId(cursor) ?? -1];
const start = scanStart(query.order, cursor, "ascending");
const scan = scanSql(start);
const clauses = [scan.clause];
const params: SqliteValue[] = [scan.param];
if (query.ownerConversationId !== undefined) {
clauses.push("owner_conversation_id = ?");
params.push(query.ownerConversationId);
@ -215,12 +229,13 @@ export class SqliteStorage implements Storage {
}
params.push(limit + 1);
const rows = await this.db.all<JsonRow>(
`SELECT record FROM conversations WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`,
`SELECT record FROM conversations WHERE ${clauses.join(" AND ")} ORDER BY id ${scan.direction} LIMIT ?`,
...params,
);
return page(
rows.map((row) => parseJson<ConversationRecord>(row.record)),
limit,
start.order,
);
}
@ -319,9 +334,10 @@ export class SqliteStorage implements Storage {
limit: number,
cursor: Cursor | undefined,
): Promise<Page<EntryRecord, Cursor>> {
const { order, after } = scanStart(query.order, cursor, "descending");
if (order === "ascending") return this.readEntriesAscending(query, limit, after);
let conversation = await this.readConversation(query.conversationId);
if (conversation === undefined) throw new Error(`Unknown conversation: ${query.conversationId}`);
const after = cursorId(cursor);
let upper: number | undefined = query.maxEntryId;
if (after !== undefined) upper = Math.min(upper ?? Number.MAX_SAFE_INTEGER, after - 1);
const values: EntryRecord[] = [];
@ -347,7 +363,49 @@ export class SqliteStorage implements Storage {
if (query.minEntryId !== undefined && upper < query.minEntryId) break;
conversation = (await this.readConversation(conversation.parent.conversationId))!;
}
return page(values, limit);
return page(values, limit, "descending");
}
/** Oldest first: the fork chain's segments from the root conversation forward, each up to its fork point. */
private async readEntriesAscending(
query: EntryQuery,
limit: number,
after: number | undefined,
): Promise<Page<EntryRecord, Cursor>> {
const segments: { readonly conversationId: ConversationId; readonly upper: number | undefined }[] = [];
let conversation = await this.readConversation(query.conversationId);
if (conversation === undefined) throw new Error(`Unknown conversation: ${query.conversationId}`);
let upper: number | undefined = query.maxEntryId;
while (true) {
segments.push({ conversationId: conversation.id, upper });
if (conversation.parent === undefined) break;
upper = upper === undefined ? conversation.parent.at : Math.min(upper, conversation.parent.at);
if (query.minEntryId !== undefined && upper < query.minEntryId) break;
conversation = (await this.readConversation(conversation.parent.conversationId))!;
}
let lower: number | undefined = query.minEntryId;
if (after !== undefined) lower = Math.max(lower ?? after + 1, after + 1);
const values: EntryRecord[] = [];
for (const segment of segments.reverse()) {
const clauses = ["conversation_id = ?"];
const params: SqliteValue[] = [segment.conversationId];
if (lower !== undefined) {
clauses.push("id >= ?");
params.push(lower);
}
if (segment.upper !== undefined) {
clauses.push("id <= ?");
params.push(segment.upper);
}
params.push(limit + 1 - values.length);
const rows = await this.db.all<JsonRow>(
`SELECT record FROM entries WHERE ${clauses.join(" AND ")} ORDER BY id ASC LIMIT ?`,
...params,
);
values.push(...rows.map((row) => parseJson<EntryRecord>(row.record)));
if (values.length > limit) break;
}
return page(values, limit, "ascending");
}
async task(id: TaskId, _context: Context): Promise<StoredTask | undefined> {
@ -363,8 +421,10 @@ export class SqliteStorage implements Storage {
_context: Context,
): Promise<Page<StoredTask, Cursor>> {
this.assertOpen();
const clauses = ["id > ?"];
const params: SqliteValue[] = [cursorId(cursor) ?? -1];
const start = scanStart(query.order, cursor, "ascending");
const scan = scanSql(start);
const clauses = [scan.clause];
const params: SqliteValue[] = [scan.param];
if (query.conversationId !== undefined) {
clauses.push("conversation_id = ?");
params.push(query.conversationId);
@ -387,12 +447,13 @@ export class SqliteStorage implements Storage {
}
params.push(limit + 1);
const rows = await this.db.all<JsonRow>(
`SELECT record FROM tasks WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`,
`SELECT record FROM tasks WHERE ${clauses.join(" AND ")} ORDER BY id ${scan.direction} LIMIT ?`,
...params,
);
return page(
rows.map((row) => parseJson<StoredTask>(row.record)),
limit,
start.order,
);
}
@ -409,8 +470,10 @@ export class SqliteStorage implements Storage {
_context: Context,
): Promise<Page<SubmissionRecord, Cursor>> {
this.assertOpen();
const clauses = ["id > ?"];
const params: SqliteValue[] = [cursorId(cursor) ?? -1];
const start = scanStart(query.order, cursor, "ascending");
const scan = scanSql(start);
const clauses = [scan.clause];
const params: SqliteValue[] = [scan.param];
if (query.conversationId !== undefined) {
clauses.push("conversation_id = ?");
params.push(query.conversationId);
@ -421,12 +484,13 @@ export class SqliteStorage implements Storage {
}
params.push(limit + 1);
const rows = await this.db.all<JsonRow>(
`SELECT record FROM submissions WHERE ${clauses.join(" AND ")} ORDER BY id LIMIT ?`,
`SELECT record FROM submissions WHERE ${clauses.join(" AND ")} ORDER BY id ${scan.direction} LIMIT ?`,
...params,
);
return page(
rows.map((row) => parseJson<SubmissionRecord>(row.record)),
limit,
start.order,
);
}

79
packages/durable/src/testing/storage-conformance.ts
View file

@ -10,7 +10,9 @@ import {
type EntryId,
type EntryRecord,
type JsonObject,
type Page,
ROOT_CONVERSATION_ID,
type ScanOrder,
type Storage,
type StorageWrite,
type SubmissionId,
@ -53,6 +55,8 @@ function entry(
type ConformanceTest = (storage: Storage) => Promise<void>;
type Identified = { readonly id: number };
type AssertionResult = {
toBe(expected: unknown): void;
toBeDefined(): void;
@ -430,6 +434,29 @@ export function createStorageConformance(options: StorageConformanceOptions): re
expect(third.items.map(({ id }) => id)).toEqual([rootFirst]);
expect(third.next).toBeUndefined();
// Oldest first: the root's segment, then each fork's, each capped at the next fork point.
const ascending = { conversationId: grandchildId, order: "ascending" } as const;
const firstUp = await storage.scanEntries(ascending, 2, undefined, context);
expect(firstUp.items.map(({ id }) => id)).toEqual([rootFirst, rootForkPoint]);
const secondUp = await storage.scanEntries(ascending, 2, firstUp.next, context);
expect(secondUp.items.map(({ id }) => id)).toEqual([childForkPoint, grandchildHead]);
const thirdUp = await storage.scanEntries({ conversationId: grandchildId }, 2, secondUp.next, context);
expect(thirdUp.items.map(({ id }) => id)).toEqual([grandchildTail]);
expect(thirdUp.next).toBeUndefined();
expect(
(
await storage.scanEntries(
{ ...ascending, minEntryId: rootForkPoint, maxEntryId: childForkPoint },
10,
undefined,
context,
)
).items.map(({ id }) => id),
).toEqual([rootForkPoint, childForkPoint]);
await expect(
storage.scanEntries({ conversationId: grandchildId, order: "descending" }, 2, firstUp.next, context),
).rejects.toThrow("cursor");
const currentMarker = await storage.findLatestHeadMarker(grandchildId, undefined, context);
expect(currentMarker?.id).toBe(grandchildHead);
expect(currentMarker?.head).toBe(grandchildHead);
@ -499,6 +526,58 @@ export function createStorageConformance(options: StorageConformanceOptions): re
).rejects.toThrow("Unknown conversation");
}),
createCase(options, "scans tables in either ID order and continues a cursor in its order", async (storage) => {
const rootId = await createRoot(storage);
const conversationIds = [rootId];
const taskIds: TaskId<JsonValue>[] = [];
const submissionIds: SubmissionId[] = [];
for (let index = 0; index < 3; index++) {
const conversationId = await storage.mintId<ConversationId>();
const taskId = await storage.mintId<TaskId<JsonValue>>();
const submissionId = await storage.mintId<SubmissionId>();
await storage.commit(
[
{ type: "conversation", value: { id: conversationId } },
{ type: "task", value: pendingTask(taskId, rootId) },
{
type: "submission",
value: { id: submissionId, conversationId: rootId, type: "input", status: "queued" },
},
],
context,
);
conversationIds.push(conversationId);
taskIds.push(taskId);
submissionIds.push(submissionId);
}
type Scan = (order: ScanOrder | undefined, cursor: Cursor | undefined) => Promise<Page<Identified, Cursor>>;
const scans: readonly (readonly [Scan, readonly number[]])[] = [
[(order, cursor) => storage.scanConversations({ order }, 2, cursor, context), conversationIds],
[(order, cursor) => storage.scanTasks({ order }, 2, cursor, context), taskIds],
[(order, cursor) => storage.scanSubmissions({ order }, 2, cursor, context), submissionIds],
];
const all = async (scan: Scan, order: ScanOrder | undefined): Promise<number[]> => {
const found: number[] = [];
let page = await scan(order, undefined);
found.push(...page.items.map(({ id }) => id));
while (page.next !== undefined) {
// A cursor carries its order; the query may omit it.
page = await scan(undefined, JSON.parse(JSON.stringify(page.next)) as Cursor);
found.push(...page.items.map(({ id }) => id));
}
return found;
};
for (const [scan, ids] of scans) {
const reversed = [...ids].reverse();
expect(await all(scan, undefined)).toEqual(ids);
expect(await all(scan, "ascending")).toEqual(ids);
expect(await all(scan, "descending")).toEqual(reversed);
const descending = await scan("descending", undefined);
expect((await scan("descending", descending.next)).items.map(({ id }) => id)).toEqual(reversed.slice(2, 4));
await expect(scan("ascending", descending.next)).rejects.toThrow("cursor");
}
}),
createCase(options, "replaces complete task records and pages filtered task scans", async (storage) => {
const rootId = await createRoot(storage);
const firstId = await storage.mintId<TaskId<JsonValue>>();

26
packages/durable/src/types.ts
View file

@ -611,22 +611,32 @@ export type Page<T, C> = {
readonly next?: C;
};
/** Backend-owned JSON continuation state that callers only round-trip to the same scan. */
/**
* Backend-owned JSON continuation state that callers only round-trip to the same scan. It carries the scan's order: a
* scan given a cursor continues in that order, and rejects a different `order` in its query.
*/
export type Cursor = Readonly<Record<string, JsonValue>>;
/** ID order of a scan: `ascending` is oldest first, `descending` newest first. */
export type ScanOrder = "ascending" | "descending";
/** Optional filters for an ordered conversation scan. */
export type ConversationQuery = {
readonly ownerConversationId?: ConversationId;
readonly ownerTaskId?: TaskId;
/** Default `ascending`; with a cursor, the cursor's order. */
readonly order?: ScanOrder;
};
/** Inclusive ID bounds for a newest-first scan of one conversation's fork-aware history. */
/** Inclusive ID bounds for a scan of one conversation's fork-aware history. */
export type EntryQuery = {
readonly conversationId: ConversationId;
/** Oldest entry ID that may be returned. */
readonly minEntryId?: EntryId;
/** Newest entry ID that may be returned. */
readonly maxEntryId?: EntryId;
/** Default `descending`; with a cursor, the cursor's order. */
readonly order?: ScanOrder;
};
/** Optional filters for an ordered scan of durable task records. */
@ -636,12 +646,16 @@ export type TaskQuery = {
readonly status?: TaskState<JsonValue, JsonValue>["status"];
readonly abortRequested?: boolean;
readonly background?: boolean;
/** Default `ascending`; with a cursor, the cursor's order. */
readonly order?: ScanOrder;
};
/** Optional filters for an ordered scan of submission records. */
export type SubmissionQuery = {
readonly conversationId?: ConversationId;
readonly status?: SubmissionRecord["status"];
/** Default `ascending`; with a cursor, the cursor's order. */
readonly order?: ScanOrder;
};
/** Current state or one historical commit sequence used for document membership and content reads. */
@ -1010,7 +1024,7 @@ export interface Storage {
/** Look up one conversation by exact ID. */
conversation(id: ConversationId, context: Context): Promise<ConversationRecord | undefined>;
/** Scan conversations in ascending ID order. */
/** Scan conversations in `query.order` (default ascending) by ID. */
scanConversations(
query: ConversationQuery,
limit: number,
@ -1037,7 +1051,7 @@ export interface Storage {
context: Context,
): Promise<(EntryRecord & { readonly head: EntryId }) | undefined>;
/** Scan the inclusive visible range newest-first, returning at most `limit` entries. */
/** Scan the inclusive visible range in `query.order` (default descending), returning at most `limit` entries. */
scanEntries(
query: EntryQuery,
limit: number,
@ -1048,7 +1062,7 @@ export interface Storage {
/** Look up the latest complete record for one task. */
task(id: TaskId, context: Context): Promise<TaskRecord<JsonValue, JsonValue, JsonValue> | undefined>;
/** Scan task records matching every supplied filter. */
/** Scan task records matching every supplied filter in `query.order` (default ascending) by ID. */
scanTasks(
query: TaskQuery,
limit: number,
@ -1059,7 +1073,7 @@ export interface Storage {
/** Look up the latest complete record for one admitted submission. */
submission(id: SubmissionId, context: Context): Promise<SubmissionRecord | undefined>;
/** Scan submissions matching every supplied filter in ascending ID order. */
/** Scan submissions matching every supplied filter in `query.order` (default ascending) by ID. */
scanSubmissions(
query: SubmissionQuery,
limit: number,

7
packages/durable/test/harness-conversations.test.ts
View file

@ -19,6 +19,7 @@ import {
MemoryStorage,
ProviderDoc,
ROOT_CONVERSATION_ID,
type ScanOrder,
} from "@earendil-works/pi-durable";
import { afterEach, describe, expect, it } from "vitest";
import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts";
@ -57,11 +58,11 @@ async function append(conversation: Conversation, text: string): Promise<EntryRe
);
}
async function allEntries(conversation: Conversation): Promise<string[]> {
async function allEntries(conversation: Conversation, order?: ScanOrder): Promise<string[]> {
const texts: string[] = [];
let cursor: Parameters<Conversation["entries"]>[2];
do {
const page = await conversation.entries({}, 2, cursor, context);
const page = await conversation.entries(order === undefined ? {} : { order }, 2, cursor, context);
for (const entry of page.items) texts.push((entry.model?.[0] as { content: string }).content);
cursor = page.next;
} while (cursor !== undefined);
@ -250,6 +251,8 @@ describe("Harness root and conversations", () => {
expect(await allEntries(root)).toEqual(["r3", "r2", "r1"]);
expect(await allEntries(child)).toEqual(["c2", "c1", "r2", "r1"]);
expect(await allEntries(grandchild)).toEqual(["g1", "c1", "r2", "r1"]);
// #10546
expect(await allEntries(grandchild, "ascending")).toEqual(["r1", "r2", "c1", "g1"]);
const bounded = await grandchild.entries(
{ minEntryId: r2!.id, maxEntryId: c1.id, conversationId: root.id } as never,
10,

13
packages/durable/test/harness-tasks.test.ts
View file

@ -852,6 +852,19 @@ describe("task scheduling", () => {
await expect(root.waitForIdle(context)).rejects.toThrow("closed");
});
// #10546
it("pages the newest tasks first without scanning older ones", async () => {
const Idle = oneStep("test.idle", async () => {});
const { harness, root } = await openRoot([Idle]);
const ids: TaskId[] = [];
for (let index = 0; index < 5; index++) ids.push(await start(root, Idle));
const first = await harness.commit((tx) => tx.scanTasks({ order: "descending" }, 2), context);
expect(first.items.map(({ id }) => id)).toEqual([ids[4], ids[3]]);
const second = await harness.commit((tx) => tx.scanTasks({}, 2, first.next), context);
expect(second.items.map(({ id }) => id)).toEqual([ids[2], ids[1]]);
await harness.close(context);
});
it("rejects unknown tasks and reports terminal tasks", async () => {
const Done = oneStep("test.quick", async (_task, runtime, ctx) => {
await runtime.commit(() => completed(null), ctx);