diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index c224b46cb..00dfa1d37 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -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)). diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 01247fec6..1b3e9bcde 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -4312,15 +4312,20 @@ type Page = { type Cursor = Readonly>; +/** 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 diff --git a/packages/durable/src/harness/harness.ts b/packages/durable/src/harness/harness.ts index fc598202a..320c712bf 100644 --- a/packages/durable/src/harness/harness.ts +++ b/packages/durable/src/harness/harness.ts @@ -134,6 +134,7 @@ class ConversationImpl 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)); } diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index 9f60c7053..6c012ddc9 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -536,7 +536,7 @@ export interface Conversation { * `fork(at)` would start with, without creating a conversation. */ context(context: Context, options?: { readonly at?: EntryId }): Promise; - /** Newest-first fork-aware history of this conversation. */ + /** Fork-aware history of this conversation, newest first unless `query.order` is `ascending`. */ entries( query: Omit, limit: number, diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index d3d93babb..2b1dc3037 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -149,6 +149,7 @@ export type { RewindableConversationDocToken, RewindableConversationSemantics, RunningTask, + ScanOrder, Seq, Session, SessionDocFamilyToken, diff --git a/packages/durable/src/storage/memory.ts b/packages/durable/src/storage/memory.ts index 003b5a926..7333d37f9 100644 --- a/packages/durable/src/storage/memory.ts +++ b/packages/durable/src/storage/memory.ts @@ -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; type TaskStatus = StoredTask["state"]["status"]; @@ -196,12 +198,27 @@ const isCurrentOnly = (record: DocumentRecord): boolean => const tableContaining = (state: State, id: Id): TableName | undefined => state.recordTypes.get(id); -const page = }>(values: readonly T[], limit: number): Page => { +const page = }>( + values: readonly T[], + limit: number, + order: ScanOrder = "ascending", +): Page => { 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[], start: ScanStart): Generator { + 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> { 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 { @@ -527,19 +551,19 @@ export class MemoryStorage implements Storage { _context: Context, ): Promise> { 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 { @@ -555,17 +579,17 @@ export class MemoryStorage implements Storage { _context: Context, ): Promise> { 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 { + 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, TableName>(); for (const write of writes) { diff --git a/packages/durable/src/storage/scan.ts b/packages/durable/src/storage/scan.ts new file mode 100644 index 000000000..8cbc1938c --- /dev/null +++ b/packages/durable/src/storage/scan.ts @@ -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> | 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, order: ScanOrder): { readonly after: number; readonly order: ScanOrder } { + return { after: id, order }; +} diff --git a/packages/durable/src/storage/sqlite/storage.ts b/packages/durable/src/storage/sqlite/storage.ts index 55cf8ea6d..90666efc2 100644 --- a/packages/durable/src/storage/sqlite/storage.ts +++ b/packages/durable/src/storage/sqlite/storage.ts @@ -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 = >(cursor: Cursor | undefined): I | undefin return idFromNumber(after); }; -const page = }>(values: readonly T[], limit: number): Page => { +const page = }>( + values: readonly T[], + limit: number, + order: ScanOrder = "ascending", +): Page => { 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> { 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( - `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(row.record)), limit, + start.order, ); } @@ -319,9 +334,10 @@ export class SqliteStorage implements Storage { limit: number, cursor: Cursor | undefined, ): Promise> { + 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> { + 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( + `SELECT record FROM entries WHERE ${clauses.join(" AND ")} ORDER BY id ASC LIMIT ?`, + ...params, + ); + values.push(...rows.map((row) => parseJson(row.record))); + if (values.length > limit) break; + } + return page(values, limit, "ascending"); } async task(id: TaskId, _context: Context): Promise { @@ -363,8 +421,10 @@ export class SqliteStorage implements Storage { _context: Context, ): Promise> { 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( - `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(row.record)), limit, + start.order, ); } @@ -409,8 +470,10 @@ export class SqliteStorage implements Storage { _context: Context, ): Promise> { 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( - `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(row.record)), limit, + start.order, ); } diff --git a/packages/durable/src/testing/storage-conformance.ts b/packages/durable/src/testing/storage-conformance.ts index 0d6e7f065..753bf2395 100644 --- a/packages/durable/src/testing/storage-conformance.ts +++ b/packages/durable/src/testing/storage-conformance.ts @@ -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; +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[] = []; + const submissionIds: SubmissionId[] = []; + for (let index = 0; index < 3; index++) { + const conversationId = await storage.mintId(); + const taskId = await storage.mintId>(); + const submissionId = await storage.mintId(); + 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>; + 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 => { + 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>(); diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index 7b6447fa6..b2924e82d 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -611,22 +611,32 @@ export type Page = { 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>; +/** 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["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; - /** 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 | 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; - /** 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, diff --git a/packages/durable/test/harness-conversations.test.ts b/packages/durable/test/harness-conversations.test.ts index 88be495be..b92e46a17 100644 --- a/packages/durable/test/harness-conversations.test.ts +++ b/packages/durable/test/harness-conversations.test.ts @@ -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 { +async function allEntries(conversation: Conversation, order?: ScanOrder): Promise { const texts: string[] = []; let cursor: Parameters[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, diff --git a/packages/durable/test/harness-tasks.test.ts b/packages/durable/test/harness-tasks.test.ts index af4ac1d17..f88b25567 100644 --- a/packages/durable/test/harness-tasks.test.ts +++ b/packages/durable/test/harness-tasks.test.ts @@ -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);