diff --git a/packages/typed-storage/__tests__/index.test.ts b/packages/typed-storage/__tests__/index.test.ts index 0d0a1b36d..78d816dd3 100644 --- a/packages/typed-storage/__tests__/index.test.ts +++ b/packages/typed-storage/__tests__/index.test.ts @@ -185,6 +185,32 @@ let EVE: User = { groups: [] }; +// Schemas shared by the non-unique-index suites: users indexed by level, and the same collection +// with no index declared (for simulating writes from before the index existed). +const INDEXED_SCHEMA = { + collections: { + users: collection()({ + primaryKey: "name", + nonUniqueIndexes: { + byLevel: (user: User) => user.level == 0 ? null : user.level + } + }) + } +}; +const PLAIN_SCHEMA = { + collections: { users: collection()({ primaryKey: "name" }) } +}; +const GROUP_SCHEMA = { + collections: { + users: collection()({ + primaryKey: "name", + nonUniqueIndexes: { + byGroup: (user: User) => user.groups + } + }) + } +}; + describe("basic collections with string primary key", () => { let mockStorage = makeMockStorage(); let storage = createTypedStorage(mockStorage, { @@ -215,7 +241,8 @@ describe("basic collections with string primary key", () => { testByName(storage.users); }); -function testByName(index: UniqueIndex) { +// Accepts both a UniqueIndex and a Collection (which has no rebuild()). +function testByName(index: Pick, "get" | "list" | "delete">) { it("supports get", () => { expect(index.get("alice")).toStrictEqual(ALICE); expect(index.get("bob")).toStrictEqual(BOB); @@ -276,7 +303,8 @@ describe("basic collections with integer primary key", () => { testByNumber(storage.users); }); -function testByNumber(index: UniqueIndex) { +// Accepts both a UniqueIndex and a Collection (which has no rebuild()). +function testByNumber(index: Pick, "get" | "list" | "delete">) { it("supports get", () => { expect(index.get(ALICE.uid)).toStrictEqual(ALICE); expect(index.get(BOB.uid)).toStrictEqual(BOB); @@ -509,16 +537,7 @@ describe("unique index by array", () => { describe("non-unique index by number", () => { let mockStorage = makeMockStorage(); - let storage = createTypedStorage(mockStorage, { - collections: { - users: collection()({ - primaryKey: "name", - nonUniqueIndexes: { - byLevel: (user: User) => user.level == 0 ? null : user.level - } - }) - } - }); + let storage = createTypedStorage(mockStorage, INDEXED_SCHEMA); expect([...storage.users.byLevel.list()]).toStrictEqual([]); @@ -611,3 +630,181 @@ describe("non-unique index by array", () => { expect([...index.list()]).toStrictEqual([BOB, DAVE, DAVE]); }); }); + +describe("non-unique index rebuild", () => { + it("backfills an index declared after records were written", () => { + let mockStorage = makeMockStorage(); + let legacy = createTypedStorage(mockStorage, PLAIN_SCHEMA); + legacy.users.put(BOB); + legacy.users.put(ALICE); + legacy.users.put(CAROL); + legacy.users.put(EVE); + + let storage = createTypedStorage(mockStorage, INDEXED_SCHEMA); + + // Declared over pre-existing records, the index starts empty -- and resolving one of those + // records would throw on the index's remove. + expect([...storage.users.byLevel.list()]).toStrictEqual([]); + expect(() => storage.users.put({...ALICE, level: 0})).toThrow("inconsistent"); + + storage.users.byLevel.rebuild(); + expect([...storage.users.byLevel.list()]).toStrictEqual([BOB, ALICE, CAROL]); + expect([...storage.users.byLevel.get(8)]).toStrictEqual([ALICE, CAROL]); + + // Writes after the rebuild keep the index consistent, including key removal. + storage.users.put({...ALICE, level: 0}); + expect([...storage.users.byLevel.list()]).toStrictEqual([BOB, CAROL]); + }); + + it("backfills a multi-key index (array index function)", () => { + let mockStorage = makeMockStorage(); + let legacy = createTypedStorage(mockStorage, PLAIN_SCHEMA); + legacy.users.put(ALICE); + legacy.users.put(BOB); + legacy.users.put(DAVE); + legacy.users.put(EVE); // no groups: unindexed + + let storage = createTypedStorage(mockStorage, GROUP_SCHEMA); + storage.users.byGroup.rebuild(); + + expect([...storage.users.byGroup.get("everyone")]).toStrictEqual([ALICE, BOB, DAVE]); + expect([...storage.users.byGroup.get("admin")]).toStrictEqual([ALICE]); + expect([...storage.users.byGroup.get("interns")]).toStrictEqual([DAVE]); + expect([...storage.users.byGroup.list({dedupe: true})]).toStrictEqual([ALICE, BOB, DAVE]); + }); + + it("reclaims orphaned child rows and never reuses child-group ids", () => { + let mockStorage = makeMockStorage(); + let storage = createTypedStorage(mockStorage, INDEXED_SCHEMA); + storage.users.put(ALICE); // level 8 + storage.users.put(BOB); // level 4 + + // A child row whose parent key vanished (e.g. a partially-applied wipe): unreachable from + // the parent keys, but still swept by rebuild's raw-range deleteAll. + mockStorage.kv.put("users.byLevel.999:zombie", {}); + let preIds = new Set([...mockStorage.kv.list({prefix: "users.byLevel:"})].map(([, id]) => id)); + let counterBefore = mockStorage.kv.get("users.byLevel#")!; + + storage.users.byLevel.rebuild(); + + expect(mockStorage.kv.get("users.byLevel.999:zombie")).toStrictEqual(undefined); + // The unique-id counter survives the wipe, so the rebuilt groups get fresh ids. + expect(mockStorage.kv.get("users.byLevel#")).toBeGreaterThanOrEqual(counterBefore); + for (let [, id] of mockStorage.kv.list({prefix: "users.byLevel:"})) { + expect(preIds.has(id)).toStrictEqual(false); + } + expect([...storage.users.byLevel.list()]).toStrictEqual([BOB, ALICE]); + }); + + it("discards stale entries for records changed behind the index's back", () => { + let mockStorage = makeMockStorage(); + let storage = createTypedStorage(mockStorage, INDEXED_SCHEMA); + storage.users.put(ALICE); + storage.users.put(DAVE); + + // Mutate through a view without the index, leaving it stale: DAVE gone, BOB unindexed. + let legacy = createTypedStorage(mockStorage, PLAIN_SCHEMA); + legacy.users.delete("dave"); + legacy.users.put(BOB); + + storage.users.byLevel.rebuild(); + expect([...storage.users.byLevel.list()]).toStrictEqual([BOB, ALICE]); + expect([...storage.users.byLevel.get(1)]).toStrictEqual([]); + }); +}); + +describe("unique index rebuild", () => { + const UNIQUE_SCHEMA = { + collections: { + users: collection()({ + primaryKey: "name", + uniqueIndexes: { + byUid: (user: User) => user.uid === 404 ? null : user.uid + } + }) + } + }; + + it("backfills an index declared after records were written", () => { + let mockStorage = makeMockStorage(); + let legacy = createTypedStorage(mockStorage, PLAIN_SCHEMA); + legacy.users.put(ALICE); + legacy.users.put(BOB); + legacy.users.put(EVE); + + let storage = createTypedStorage(mockStorage, UNIQUE_SCHEMA); + + // Declared over pre-existing records, the index starts empty -- and changing one of those + // records' keys would throw on the index's remove. + expect(storage.users.byUid.get(ALICE.uid)).toStrictEqual(undefined); + expect(() => storage.users.put({...ALICE, uid: 46})).toThrow("inconsistent"); + + storage.users.byUid.rebuild(); + expect(storage.users.byUid.get(ALICE.uid)).toStrictEqual(ALICE); + expect(storage.users.byUid.get(BOB.uid)).toStrictEqual(BOB); + expect(storage.users.byUid.get(EVE.uid)).toStrictEqual(undefined); // null-keyed, unindexed + + // Writes after the rebuild keep the index consistent, including key changes. + storage.users.put({...ALICE, uid: 46}); + expect(storage.users.byUid.get(ALICE.uid)).toStrictEqual(undefined); + expect(storage.users.byUid.get(46)).toStrictEqual({...ALICE, uid: 46}); + }); + + it("throws when two records derive the same key, leaving no partial index", () => { + let mockStorage = makeMockStorage(); + let legacy = createTypedStorage(mockStorage, PLAIN_SCHEMA); + legacy.users.put(ALICE); + legacy.users.put({...BOB, uid: ALICE.uid}); + + let storage = createTypedStorage(mockStorage, UNIQUE_SCHEMA); + expect(() => storage.users.byUid.rebuild()).toThrow("conflicts"); + + // The rebuild runs in one transaction: the entries added before the conflict rolled back. + expect(storage.users.byUid.get(ALICE.uid)).toStrictEqual(undefined); + expect([...storage.users.byUid.list()]).toStrictEqual([]); + }); +}); + +describe("non-unique index ranged get", () => { + it("ranges and pages within one group by primary key", () => { + let storage = createTypedStorage(makeMockStorage(), GROUP_SCHEMA); + storage.users.put(BOB); + storage.users.put(DAVE); + storage.users.put(CAROL); + storage.users.put(ALICE); + + let index = storage.users.byGroup; + expect([...index.get("everyone", {start: "bob"})]).toStrictEqual([BOB, CAROL, DAVE]); + expect([...index.get("everyone", {startAfter: "bob"})]).toStrictEqual([CAROL, DAVE]); + expect([...index.get("everyone", {end: "carol"})]).toStrictEqual([ALICE, BOB]); + expect([...index.get("everyone", {limit: 2})]).toStrictEqual([ALICE, BOB]); + expect([...index.get("everyone", {reverse: true})]).toStrictEqual([DAVE, CAROL, BOB, ALICE]); + expect([...index.get("admin", {reverse: true, limit: 1})]).toStrictEqual([CAROL]); + expect([...index.get("nobody", {limit: 2})]).toStrictEqual([]); + }); + + it("pages a numeric-pk group descending with an exclusive end", () => { + type Row = {id: number, group: string}; + let storage = createTypedStorage(makeMockStorage(), { + collections: { + rows: collection()({ + primaryKey: "id", + nonUniqueIndexes: { + byGroup: (row: Row) => row.group + } + }) + } + }); + for (let id = 0; id < 7; id++) { + storage.rows.put({id, group: id % 2 === 0 ? "even" : "odd"}); + } + + // The cursored-history shape: a newest-first page strictly below the cursor. + let index = storage.rows.byGroup; + expect([...index.get("even", {end: 6, reverse: true, limit: 2})].map(r => r.id)) + .toStrictEqual([4, 2]); + expect([...index.get("even", {reverse: true, limit: 2})].map(r => r.id)) + .toStrictEqual([6, 4]); + expect([...index.get("even", {end: 2})].map(r => r.id)).toStrictEqual([0]); + }); +}); diff --git a/packages/typed-storage/src/index.ts b/packages/typed-storage/src/index.ts index 2a1aa8911..757a73d2c 100644 --- a/packages/typed-storage/src/index.ts +++ b/packages/typed-storage/src/index.ts @@ -61,13 +61,34 @@ export interface UniqueIndex { get(key: Key): T | undefined; list(options?: ListOptions): Iterable; delete(key: Key): boolean; + + /** + * Discard the index's contents and re-derive them from the collection's records. Indexes are + * only maintained at write time, so an index declared after records already exist starts empty + * (and updates to those records would corrupt it, or throw); a migration must rebuild() such an + * index before the records are touched. Throws if two records derive the same key. + */ + rebuild(): void; } /** An index where each key may match multiple records. */ -export interface NonUniqueIndex { - get(key: Key): Iterable; +export interface NonUniqueIndex { + /** + * List the records matching `key`, ordered by primary key. `options` ranges and pages over the + * matching records' primary keys. Note that `limit` here counts records, unlike in a top-level + * list(), where it counts index keys. + */ + get(key: Key, options?: ListOptions): Iterable; list(options?: ListOptions): Iterable; delete(key: Key): number; + + /** + * Discard the index's contents and re-derive them from the collection's records. Indexes are + * only maintained at write time, so an index declared after records already exist starts empty + * (and updates to those records would corrupt it, or throw); a migration must rebuild() such an + * index before the records are touched. + */ + rebuild(): void; } type Key = string | number; @@ -86,8 +107,8 @@ type UniqueIndexed = { [K in keyof Indexes]: UniqueIndex>> } -type NonUniqueIndexed = { - [K in keyof Indexes]: NonUniqueIndex>> +type NonUniqueIndexed = { + [K in keyof Indexes]: NonUniqueIndex>, PK> } export interface Subscriber { @@ -96,7 +117,14 @@ export interface Subscriber { remove(record: T): void; } -export interface Collection extends UniqueIndex { +/** + * A collection of records addressed by primary key. + */ +export interface Collection { + get(key: PrimaryKey): T | undefined; + list(options?: ListOptions): Iterable; + delete(key: PrimaryKey): boolean; + put(value: T): void; subscribe(subscriber: Subscriber): void; @@ -168,7 +196,7 @@ type CollectionImpl = & Collection> & UniqueIndexed - & NonUniqueIndexed; + & NonUniqueIndexed & Key>; type TypedStorageImpl = TypedStorage & { @@ -281,6 +309,21 @@ class KvPrefixedView { return new KvPrefixedView(this.#kv, `${this.#name}.${name}`); } + /** + * Delete every child row (`name.` prefix) and record (`name:` prefix) under this view, + * unbuffered -- point deletes are permitted under an open list() cursor. Sweeping the raw key + * ranges also reclaims child rows orphaned by earlier inconsistencies, which a walk of the + * parent keys would never reach. The `name#` unique-id counter is intentionally kept: ids must + * never be reused. + */ + deleteAll(): void { + for (let prefix of [`${this.#name}.`, `${this.#name}:`]) { + for (let [key, _] of this.#kv.list({prefix})) { + this.#kv.delete(key); + } + } + } + getUnidqueId(): number { let key = `${this.#name}#`; let id = this.#kv.get(key) || 0; @@ -375,14 +418,15 @@ function createCollection< // Add a subscriber subscribing on behalf of an index based on the given IndexFunction. This // code is shared for unique and non-unique indexes. This code in particular takes care of the - // case where the index function returns an array. + // case where the index function returns an array. Returns the subscriber's add(), so callers + // can also feed pre-existing records into the index (see rebuild()). function addIndexSubscriber( idx: IndexFunction, ops: { add(idxKey: Key, pk: Key, type: "Insertion" | "Update"): void; remove(idxKey: Key, pk: Key): void; - }) { - subscribers.add({ + }): (record: T) => void { + let subscriber: Subscriber = { add(record: T) { let pk = pkForT(record); let idxKeys = idx(record); @@ -451,7 +495,9 @@ function createCollection< ops.remove(idxKeys, pk); } } - }); + }; + subscribers.add(subscriber); + return subscriber.add; } // --------------------------------------------------------------------------- @@ -460,6 +506,22 @@ function createCollection< for (let [idxName, idx] of Object.entries(schema.uniqueIndexes || {})) { let idxKv = new KvPrefixedView(storage.kv, `${name}.${idxName}`); + let addToIndex = addIndexSubscriber(idx as IndexFunction, { + add(idxKey: Key, pk: Key, type: "Insertion" | "Update") { + let oldValue = idxKv.get(idxKey); + if (oldValue !== undefined) { + throw new Error(`${type} conflicts with record '${oldValue}' in '${name}.${idxName}'.`); + } + idxKv.put(idxKey, pk); + }, + remove(idxKey: Key, pk: Key) { + if (!idxKv.delete(idxKey)) { + throw new Error( + `Index '${name}.${idxName}' is inconsistent: removed record is not present.`); + } + } + }); + let index: UniqueIndex = { get(key: Key): T | undefined { let pk = idxKv.get(key); @@ -484,38 +546,59 @@ function createCollection< let pk = idxKv.get(key); return pk === undefined ? false : collection.delete(pk); }, + rebuild(): void { + // One transaction, so a mid-scan throw (e.g. a key conflict) can't leave the index + // partially built after the wipe. The adds are point reads/writes, permitted under the + // record scan's open cursor. + storage.transactionSync(() => { + idxKv.deleteAll(); + for (let record of collection.list()) { + addToIndex(record); + } + }); + }, }; result[idxName] = index; + } - addIndexSubscriber(idx as IndexFunction, { + // --------------------------------------------------------------------------- + // Non-unique indexes + + for (let [idxName, idx] of Object.entries(schema.nonUniqueIndexes || {})) { + let idxKv = new KvPrefixedView(storage.kv, `${name}.${idxName}`); + + let addToIndex = addIndexSubscriber(idx as IndexFunction, { add(idxKey: Key, pk: Key, type: "Insertion" | "Update") { - let oldValue = idxKv.get(idxKey); - if (oldValue !== undefined) { - throw new Error(`${type} conflicts with record '${oldValue}' in '${name}.${idxName}'.`); + let id = idxKv.get(idxKey); + if (id === undefined) { + id = idxKv.getUnidqueId(); + idxKv.put(idxKey, id); } - idxKv.put(idxKey, pk); + + let child = idxKv.getChild(id.toString()); + child.put(pk, {}); }, remove(idxKey: Key, pk: Key) { - if (!idxKv.delete(idxKey)) { + let id = idxKv.get(idxKey); + if (id === undefined) { throw new Error( `Index '${name}.${idxName}' is inconsistent: removed record is not present.`); } + + let child = idxKv.getChild(id.toString()); + child.delete(pk); + if (Array.from(child.list({limit: 1})).length == 0) { + idxKv.delete(idxKey); + } } }); - } - - // --------------------------------------------------------------------------- - // Non-unique indexes - - for (let [idxName, idx] of Object.entries(schema.nonUniqueIndexes || {})) { - let idxKv = new KvPrefixedView(storage.kv, `${name}.${idxName}`); let index: NonUniqueIndex = { - *get(key: Key): Generator { + *get(key: Key, options?: ListOptions): Generator { let id = idxKv.get(key) if (id === undefined) return; let child = idxKv.getChild(id.toString()); - for (let pk of child.listKeys()) { + for (let pk of child.listKeys(options)) { yield collection.get(pk)!; } }, @@ -562,34 +645,19 @@ function createCollection< return count; } }, + rebuild(): void { + // One transaction, so a mid-scan throw (e.g. a key conflict) can't leave the index + // partially built after the wipe. The adds are point reads/writes, permitted under the + // record scan's open cursor. + storage.transactionSync(() => { + idxKv.deleteAll(); + for (let record of collection.list()) { + addToIndex(record); + } + }); + }, }; result[idxName] = index; - - addIndexSubscriber(idx as IndexFunction, { - add(idxKey: Key, pk: Key, type: "Insertion" | "Update") { - let id = idxKv.get(idxKey); - if (id === undefined) { - id = idxKv.getUnidqueId(); - idxKv.put(idxKey, id); - } - - let child = idxKv.getChild(id.toString()); - child.put(pk, {}); - }, - remove(idxKey: Key, pk: Key) { - let id = idxKv.get(idxKey); - if (id === undefined) { - throw new Error( - `Index '${name}.${idxName}' is inconsistent: removed record is not present.`); - } - - let child = idxKv.getChild(id.toString()); - child.delete(pk); - if (Array.from(child.list({limit: 1})).length == 0) { - idxKv.delete(idxKey); - } - } - }); } // --------------------------------------------------------------------------- diff --git a/packages/workshop-backend/__integration__/open-gadget-rpc.test.ts b/packages/workshop-backend/__integration__/open-gadget-rpc.test.ts index 8aa29ecea..9df844e32 100644 --- a/packages/workshop-backend/__integration__/open-gadget-rpc.test.ts +++ b/packages/workshop-backend/__integration__/open-gadget-rpc.test.ts @@ -220,3 +220,20 @@ describe("workspace session across a user-DO-only reset", () => { expect(await gadget.getTitle()).toBe("post-reset gadget"); }); }); + +// Smoke the paged action-log read against a real workspace DO: proves the @validateRpc wiring +// accepts the option shape (the semantics live in __tests__/action-log-pagination.test.ts). +// Runs after the reset tests so this session's DOs aren't torn down by abortAllDurableObjects(). +describe("paged action-log reads", () => { + it("answers listActions on a fresh workspace", async () => { + using publicApi = await connect(); + const account = await createAccount(publicApi, "actionlog"); + using authenticated = await publicApi.authenticate(account.token); + using workspace = await authenticated.newGadget(); + + expect(await workspace.listActions({ filter: "action" })).toEqual({ entries: [] }); + // The pending filter is a distinct union member; this proves the regenerated validator + // accepts it end to end. + expect(await workspace.listActions({ filter: "pending" })).toEqual({ entries: [] }); + }); +}); diff --git a/packages/workshop-backend/__tests__/action-log-pagination.test.ts b/packages/workshop-backend/__tests__/action-log-pagination.test.ts new file mode 100644 index 000000000..cc66ace96 --- /dev/null +++ b/packages/workshop-backend/__tests__/action-log-pagination.test.ts @@ -0,0 +1,408 @@ +import { describe, expect, it, vi } from "vitest"; +import type { RpcStub } from "capnweb"; +import type { ActionLogEntry, ActionsSubscriber } from "@gadgets/workshop-shared/api"; +import { + ACTION_HISTORY_PAGE_DEFAULT_LIMIT, ACTION_REPLAY_PAGE_SIZE, +} from "../src/overseer.js"; +import { makeMockStorage } from "./mock-storage.js"; +import { + FIXTURE_EPOCH, makeActionStorage, makePreIndexActionStorage, openFakeOverseer, putAction, +} from "./fixtures.js"; + +vi.mock("capnweb-validate", () => ({ validateRpc: () => () => undefined })); + +// Hand-rolled ActionsSubscriber stub. `events` interleaves entry ids with "ready", so tests can +// assert both content and ordering of the delivered stream. +function makeSubscriber(entry?: (record: ActionLogEntry) => Promise) { + let events: Array = []; + let subscriber = { + entry: entry ?? (async (record: ActionLogEntry) => { events.push(record.id); }), + ready: async () => { events.push("ready"); }, + dup: () => subscriber, + onRpcBroken: () => {}, + [Symbol.dispose]: () => {}, + }; + return { subscriber: subscriber as unknown as RpcStub, events }; +} + +describe("subscribeToActions", () => { + it("delivers no pre-existing records: ready fires immediately", async () => { + // Live deltas only — the current pending set is queried via listActions({filter: "pending"}). + let storage = makeActionStorage(); + putAction(storage, 0); // pending action + putAction(storage, 1, { state: "approved" }); + putAction(storage, 2, { type: "bindHook", state: "pending" }); // pending, non-action type + let client = await openFakeOverseer(storage); + let { subscriber, events } = makeSubscriber(); + + using _sub = await client.subscribeToActions(subscriber); + expect(events).toEqual(["ready"]); + }); + + it("delivers adds and resolutions live, in stream order", async () => { + let storage = makeActionStorage(); + let client = await openFakeOverseer(storage); + let { subscriber, events } = makeSubscriber(); + + using _sub = await client.subscribeToActions(subscriber); + putAction(storage, 0); + let record = storage.actions.get(0)!; + record.state = "approved"; + storage.actions.put(record); + + expect(events).toEqual(["ready", 0, 0]); // the add, then the resolving update + }); + + it("replays every record, resolved included, for an epoch startAfter", async () => { + let storage = makeActionStorage(); + putAction(storage, 0); + putAction(storage, 1, { state: "approved" }); + putAction(storage, 2, { type: "observation", state: "rejected" }); + putAction(storage, 3, { type: "bindHook", state: "pending" }); + let client = await openFakeOverseer(storage); + let { subscriber, events } = makeSubscriber(); + + using _sub = await client.subscribeToActions(subscriber, new Date(0)); + expect(events).toEqual([0, 1, 2, 3, "ready"]); + }); + + it("replays only records whose last state change is at or past startAfter", async () => { + // putAction stamps createdAt = FIXTURE_EPOCH + id, so the cutoff falls mid-log. The bound is + // inclusive: the record last changed exactly at the cutoff is re-delivered. + let storage = makeActionStorage(); + putAction(storage, 0); + putAction(storage, 1, { state: "approved" }); + putAction(storage, 2, { state: "rejected" }); + putAction(storage, 3); + let client = await openFakeOverseer(storage); + let { subscriber, events } = makeSubscriber(); + + using _sub = await client.subscribeToActions(subscriber, new Date(FIXTURE_EPOCH + 1)); + expect(events).toEqual([1, 2, 3, "ready"]); + }); + + it("replays only the changed records, in change-time order, however large the log", async () => { + let storage = makeActionStorage(); + for (let id = 0; id < 550; id++) putAction(storage, id, { state: "approved" }); + // Two records resolved after the cutoff, in the opposite of id order. + putAction(storage, 7, { state: "approved", appliedAt: new Date(FIXTURE_EPOCH + 2000) }); + putAction(storage, 3, { state: "approved", appliedAt: new Date(FIXTURE_EPOCH + 3000) }); + let client = await openFakeOverseer(storage); + let { subscriber, events } = makeSubscriber(); + + using _sub = await client.subscribeToActions(subscriber, new Date(FIXTURE_EPOCH + 1000)); + expect(events).toEqual([7, 3, "ready"]); + }); + + it("replays a whole batch tied at the cutoff instant, each record once", async () => { + // The frozen clock stamps whole batches with one instant; an exclusive bound would lose the + // siblings of the last-seen record. + let instant = new Date(FIXTURE_EPOCH + 100); + let storage = makeActionStorage(); + putAction(storage, 0, { state: "approved" }); // predates the instant + putAction(storage, 1, { createdAt: instant }); + putAction(storage, 2, { createdAt: instant }); + // Changed twice within the instant: one index key, so one delivery of the final state. + putAction(storage, 3, { createdAt: instant, appliedAt: instant }); + putAction(storage, 3, { createdAt: instant, appliedAt: instant, state: "approved" }); + let client = await openFakeOverseer(storage); + let entries: ActionLogEntry[] = []; + let { subscriber } = makeSubscriber(async record => { entries.push(record); }); + + using _sub = await client.subscribeToActions(subscriber, instant); + expect(entries.map(e => e.id)).toEqual([1, 2, 3]); + expect(entries[2].state).toBe("approved"); + }); + + it("hands records changing mid-replay to the live stream and still terminates", async () => { + let storage = makeActionStorage(); + // Two pages, so the sweep is parked mid-replay while the gate holds the first page open. + for (let id = 0; id <= ACTION_REPLAY_PAGE_SIZE; id++) putAction(storage, id); + let client = await openFakeOverseer(storage); + let release!: () => void; + let gate = new Promise(resolve => { release = resolve; }); + let { subscriber, events } = makeSubscriber(async record => { + events.push(record.id); + await gate; + }); + + let pending = client.subscribeToActions(subscriber, new Date(0)); + let newId = ACTION_REPLAY_PAGE_SIZE + 100; + putAction(storage, newId); // changes past the sweep's fixed end key + release(); + using _sub = await pending; + + // Delivered once, by the live subscription; the replay ends at its end key. + expect(events.filter(id => id === newId)).toEqual([newId]); + expect(events.at(-1)).toBe("ready"); + expect(events.length).toBe(ACTION_REPLAY_PAGE_SIZE + 3); // replayed pages + live add + ready + }); + + it("replays a record created before the cutoff but resolved after it", async () => { + let storage = makeActionStorage(); + putAction(storage, 0, { state: "approved", appliedAt: new Date(FIXTURE_EPOCH + 500) }); + putAction(storage, 1, { state: "approved" }); // both created and resolved before the cutoff + let client = await openFakeOverseer(storage); + let { subscriber, events } = makeSubscriber(); + + using _sub = await client.subscribeToActions(subscriber, new Date(FIXTURE_EPOCH + 100)); + expect(events).toEqual([0, "ready"]); + }); + + it("replays a hook toggled after the cutoff, carrying the toggled state", async () => { + let storage = makeActionStorage(); + putAction(storage, 0, { type: "bindHook", state: "pending" }); + // The mock kv structuredClones every write, which a live controller stub wouldn't survive, + // so serve disableHook()'s boundHooks reads from a hand-rolled view instead. + let hook = { + id: 7, actionId: 0, gatekeeperId: 1, enabled: true, + controller: { disable: async () => {} }, + }; + let client = await openFakeOverseer({ + ...storage, + boundHooks: { + get: (id: number) => (id === hook.id ? hook : undefined), + put: (record: typeof hook) => Object.assign(hook, record), + }, + }); + await client.disableHook(hook.id); // stamps appliedAt = now on the bindHook action record + + let entries: ActionLogEntry[] = []; + let { subscriber } = makeSubscriber(async record => { entries.push(record); }); + // Cutoff after creation (fixture epoch) but before the toggle (wall clock). + using _sub = await client.subscribeToActions(subscriber, new Date(FIXTURE_EPOCH + 100)); + expect(entries.map(e => e.id)).toEqual([0]); + expect(entries[0]).toMatchObject({ type: "bindHook", enabled: false }); + }); + + it("rejects the resume replay when the subscriber fails mid-sweep", async () => { + let storage = makeActionStorage(); + // More than one page, so the failure must also stop the sweep from advancing. + for (let id = 0; id <= ACTION_REPLAY_PAGE_SIZE; id++) putAction(storage, id); + let client = await openFakeOverseer(storage); + let entries = 0; + let { subscriber, events } = makeSubscriber(async () => { + ++entries; + throw new Error("entry failed"); + }); + + // Each page's delivery is awaited, so the rejection surfaces directly, before ready(). + await expect(client.subscribeToActions(subscriber, new Date(0))) + .rejects.toThrow("entry failed"); + expect(events).not.toContain("ready"); + expect(entries).toBeLessThanOrEqual(ACTION_REPLAY_PAGE_SIZE); + }); + + it("stops delivering after the subscription is disposed", async () => { + let storage = makeActionStorage(); + putAction(storage, 0); + let client = await openFakeOverseer(storage); + let { subscriber, events } = makeSubscriber(); + + let sub = await client.subscribeToActions(subscriber); + sub[Symbol.dispose](); + await scheduler.wait(0); // let the stub's disposer run + putAction(storage, 1); + + expect(events).toEqual(["ready"]); + }); +}); + +describe("listActions", () => { + it("returns records newest-first, pending included", async () => { + let storage = makeActionStorage(); + putAction(storage, 0, { state: "approved" }); + putAction(storage, 1); // pending + putAction(storage, 2, { state: "rejected" }); + putAction(storage, 3, { type: "observation", state: "approved" }); + let client = await openFakeOverseer(storage); + + let page = await client.listActions(); + expect(page.entries.map(e => e.id)).toEqual([3, 2, 1, 0]); + expect(page.nextBeforeId).toBeUndefined(); + }); + + it("filters by record type, pending included", async () => { + let storage = makeActionStorage(); + putAction(storage, 0, { state: "approved" }); + putAction(storage, 1, { type: "observation", state: "approved" }); + putAction(storage, 2, { type: "bindHook", state: "approved" }); + putAction(storage, 3, { type: "observation", state: "pending" }); + let client = await openFakeOverseer(storage); + + let page = await client.listActions({ filter: "observation" }); + expect(page.entries.map(e => e.id)).toEqual([3, 1]); + }); + + it("applies the default limit and reports more history", async () => { + let storage = makeActionStorage(); + let total = ACTION_HISTORY_PAGE_DEFAULT_LIMIT + 10; + for (let id = 0; id < total; id++) putAction(storage, id, { state: "approved" }); + let client = await openFakeOverseer(storage); + + let first = await client.listActions(); + expect(first.entries.length).toBe(ACTION_HISTORY_PAGE_DEFAULT_LIMIT); + expect(first.nextBeforeId).toBe(total - ACTION_HISTORY_PAGE_DEFAULT_LIMIT); + + let second = await client.listActions({ beforeId: first.nextBeforeId }); + expect(second.entries.length).toBe(10); + expect(second.nextBeforeId).toBeUndefined(); + }); + + it("returns sparse matches in one full page, however much history buries them", async () => { + let storage = makeActionStorage(); + // A few observations buried under far more history than the old design's per-call scan cap: + // the index-backed read must surface them in ONE call, with no cursor dance. + putAction(storage, 0, { type: "observation", state: "approved" }); + putAction(storage, 1, { type: "observation", state: "rejected" }); + for (let id = 2; id < 550; id++) putAction(storage, id, { state: "approved" }); + let client = await openFakeOverseer(storage); + + let page = await client.listActions({ filter: "observation" }); + expect(page.entries.map(e => e.id)).toEqual([1, 0]); + expect(page.nextBeforeId).toBeUndefined(); + }); + + it("pages without overlap or gaps", async () => { + let storage = makeActionStorage(); + let expected: number[] = []; + for (let id = 0; id < 130; id++) { + // Mixed states, so the "all" pages span records with differing byHistoryFilter keys. + putAction(storage, id, { state: id % 4 === 0 ? "pending" : "approved" }); + expected.unshift(id); + } + let client = await openFakeOverseer(storage); + + let ids: number[] = []; + let beforeId: number | undefined; + do { + let page = await client.listActions({ beforeId }); + ids.push(...page.entries.map(e => e.id)); + beforeId = page.nextBeforeId; + } while (beforeId !== undefined); + + expect(ids).toEqual(expected); + }); + + it("rejects an invalid beforeId", async () => { + let client = await openFakeOverseer(makeActionStorage()); + + await expect(client.listActions({ beforeId: -1 })).rejects.toThrow("Invalid beforeId"); + }); +}); + +describe("listActions with the pending filter", () => { + it("returns pendings of any type newest-first across gatekeepers, excluding resolved", + async () => { + let storage = makeActionStorage(); + putAction(storage, 0, { gatekeeperId: 2 }); + putAction(storage, 1, { state: "approved" }); + putAction(storage, 2, { type: "bindHook", state: "pending", gatekeeperId: 1 }); + putAction(storage, 3, { state: "rejected" }); + putAction(storage, 4, { gatekeeperId: 3 }); + let client = await openFakeOverseer(storage); + + // The index groups by gatekeeper; the page must still be one id-ordered (descending) stream. + let page = await client.listActions({ filter: "pending" }); + expect(page.entries.map(e => e.id)).toEqual([4, 2, 0]); + expect(page.nextBeforeId).toBeUndefined(); + }); + + it("pages to exhaustion without overlap or gaps", async () => { + let storage = makeActionStorage(); + let expected: number[] = []; + for (let id = 0; id < ACTION_HISTORY_PAGE_DEFAULT_LIMIT * 2 + 30; id++) { + let pending = id % 3 !== 0; + putAction(storage, id, { state: pending ? "pending" : "approved", gatekeeperId: id % 4 }); + if (pending) expected.unshift(id); + } + let client = await openFakeOverseer(storage); + + let ids: number[] = []; + let beforeId: number | undefined; + do { + let page = await client.listActions({ filter: "pending", beforeId }); + expect(page.entries.length).toBeLessThanOrEqual(ACTION_HISTORY_PAGE_DEFAULT_LIMIT); + ids.push(...page.entries.map(e => e.id)); + beforeId = page.nextBeforeId; + } while (beforeId !== undefined); + + expect(ids).toEqual(expected); + }); + + it("reflects a resolution between pages: the record stops appearing", async () => { + let storage = makeActionStorage(); + let total = ACTION_HISTORY_PAGE_DEFAULT_LIMIT + 10; + for (let id = 0; id < total; id++) putAction(storage, id); + let client = await openFakeOverseer(storage); + + let first = await client.listActions({ filter: "pending" }); + expect(first.entries.length).toBe(ACTION_HISTORY_PAGE_DEFAULT_LIMIT); + expect(first.nextBeforeId).toBe(10); + + // Resolve a record that would have been on the second page. + let record = storage.actions.get(5)!; + record.state = "approved"; + storage.actions.put(record); + + let second = await client.listActions({ filter: "pending", beforeId: first.nextBeforeId }); + expect(second.entries.map(e => e.id)).toEqual([9, 8, 7, 6, 4, 3, 2, 1, 0]); + expect(second.nextBeforeId).toBeUndefined(); + }); + + it("sees records written before the indexes existed once a rebuild backfills them", async () => { + // Mirrors the version-3 migration: records predate the index declarations, so each index + // starts empty until the migration's rebuild() runs. + let mock = makeMockStorage(); + let legacy = makePreIndexActionStorage(mock); + putAction(legacy, 0); + putAction(legacy, 1, { state: "approved" }); + putAction(legacy, 2); + putAction(legacy, 3, { type: "observation", state: "rejected" }); + + let storage = makeActionStorage(mock); + storage.actions.pendingByGatekeeper.rebuild(); + storage.actions.byHistoryFilter.rebuild(); + storage.actions.byLastChanged.rebuild(); + let client = await openFakeOverseer(storage); + + // Every filter serves the legacy records. + expect((await client.listActions({ filter: "pending" })).entries.map(e => e.id)) + .toEqual([2, 0]); + expect((await client.listActions()).entries.map(e => e.id)).toEqual([3, 2, 1, 0]); + expect((await client.listActions({ filter: "action" })).entries.map(e => e.id)) + .toEqual([2, 1, 0]); + expect((await client.listActions({ filter: "observation" })).entries.map(e => e.id)) + .toEqual([3]); + + // Resolving a backfilled record must not throw on any index's update. + let record = storage.actions.get(2)!; + record.state = "approved"; + record.appliedAt = new Date(FIXTURE_EPOCH + 100); + storage.actions.put(record); + expect((await client.listActions({ filter: "pending" })).entries.map(e => e.id)).toEqual([0]); + expect((await client.listActions()).entries.map(e => e.id)).toEqual([3, 2, 1, 0]); + + // The resume replay serves the backfilled records too. + let { subscriber, events } = makeSubscriber(); + using _sub = await client.subscribeToActions(subscriber, new Date(FIXTURE_EPOCH + 3)); + expect(events).toEqual([3, 2, "ready"]); + }); +}); + +describe("UseOverseerInterface", () => { + it("answers listActions with an empty terminal page and the subscription inertly", async () => { + let storage = makeActionStorage(); + putAction(storage, 0); + putAction(storage, 1, { state: "approved" }); + let client = await openFakeOverseer(storage, { role: "use" }); + let { subscriber, events } = makeSubscriber(); + + expect(await client.listActions()).toEqual({ entries: [] }); + expect(await client.listActions({ filter: "pending" })).toEqual({ entries: [] }); + + using _sub = await client.subscribeToActions(subscriber); + putAction(storage, 2); + expect(events).toEqual(["ready"]); // settled empty; nothing replayed or delivered + }); +}); diff --git a/packages/workshop-backend/__tests__/auto-approval.test.ts b/packages/workshop-backend/__tests__/auto-approval.test.ts index 0b4ce7389..a2e0a42f2 100644 --- a/packages/workshop-backend/__tests__/auto-approval.test.ts +++ b/packages/workshop-backend/__tests__/auto-approval.test.ts @@ -1,20 +1,12 @@ import { describe, it, expect } from "vitest"; -import { createTypedStorage, collection } from "@gadgets/typed-storage"; -import { AutoApprovalDrainer, AutoApprovalStorage, ApplyPendingActionFn } from "../src/auto-approval.js"; -import type { ActionRecord, AutoApproveTagRecord } from "../src/overseer.js"; +import { AutoApprovalDrainer, AutoApprovalStorage, ApplyPendingActionFn } + from "../src/auto-approval.js"; +import type { ActionRecord } from "../src/overseer.js"; import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; import { makeMockStorage } from "./mock-storage.js"; +import { makeActionStorage, makePreIndexActionStorage, putAction } from "./fixtures.js"; -function makeStorage(): AutoApprovalStorage { - return createTypedStorage(makeMockStorage(), { - collections: { - actions: collection()({ primaryKey: "id" }), - autoApproveTags: collection()({ - primaryKey: (r: AutoApproveTagRecord) => `${r.gatekeeperId}:${r.actionKind.tag}`, - }), - }, - }); -} +const makeStorage = makeActionStorage; const GK = 1; const ENABLER: AiChatAuthorInfo = { type: "user", id: "enabler@example.com", name: "Enabler" }; @@ -24,28 +16,6 @@ function enableRule(storage: AutoApprovalStorage, actionTag = "edit", gatekeeper gatekeeperId, actionKind: { tag: actionTag, label: "Edits" }, enabledBy: ENABLER }); } -function putAction( - storage: AutoApprovalStorage, id: number, - opts: { gatekeeperId?: number; actionTag?: string; autoApprovable?: boolean; - state?: ActionRecord["state"] } = {}) { - storage.actions.put({ - id, - gatekeeperId: opts.gatekeeperId ?? GK, - caller: { from: "agent", chatId: 1 }, - createdAt: new Date(), - state: opts.state ?? "pending", - type: "action", - action: id, - description: { - title: `Action ${id}`, - description: `Action ${id} description`, - implementsRevert: true, - actionKind: { tag: opts.actionTag ?? "edit", label: "Edits" }, - autoApprovable: opts.autoApprovable ?? true, - }, - }); -} - function getAction(storage: AutoApprovalStorage, id: number): ActionRecord & {type: "action"} { let record = storage.actions.get(id); if (!record || record.type !== "action") throw new Error(`No action ${id}`); @@ -211,4 +181,103 @@ describe("AutoApprovalDrainer.drain", () => { expect(getAction(storage, 1).state).toBe("approved"); expect(getAction(storage, 2).state).toBe("approved"); }); + + it("drains a large log, applying eligible actions in ascending order", async () => { + let storage = makeStorage(); + enableRule(storage); + let eligible: number[] = []; + for (let id = 0; id < 230; id++) { + if (id % 5 === 0) { + putAction(storage, id, { gatekeeperId: GK + 1 }); // other gatekeeper: skipped, not a gate + } else if (id % 5 === 1) { + putAction(storage, id, { state: "approved" }); // already resolved + } else { + putAction(storage, id); + eligible.push(id); + } + } + + let { applyFn, calls } = makeImmediateApply(storage); + await new AutoApprovalDrainer(storage, applyFn).drain(GK); + + expect(calls).toEqual(eligible); + }); + + it("halts at a manual gate deep in the log", async () => { + let storage = makeStorage(); + enableRule(storage); + let gateId = 105; + for (let id = 0; id < 120; id++) { + putAction(storage, id, { autoApprovable: id !== gateId }); + } + + let { applyFn, calls } = makeImmediateApply(storage); + await new AutoApprovalDrainer(storage, applyFn).drain(GK); + + expect(calls).toEqual(Array.from({ length: gateId }, (_, i) => i)); + expect(getAction(storage, gateId).state).toBe("pending"); + expect(getAction(storage, gateId + 1).state).toBe("pending"); + }); + + it("drains pendings written before the index existed once a rebuild backfills it", async () => { + // Mirrors the version-3 migration: the records predate the action-index declarations. + let mock = makeMockStorage(); + let legacy = makePreIndexActionStorage(mock); + putAction(legacy, 1); + putAction(legacy, 2, { state: "approved" }); + putAction(legacy, 3); + + let storage = makeStorage(mock); + storage.actions.pendingByGatekeeper.rebuild(); + storage.actions.byHistoryFilter.rebuild(); + storage.actions.byLastChanged.rebuild(); + enableRule(storage); + + // The apply persists a resolved state, which must not throw on the backfilled index. + let { applyFn, calls } = makeImmediateApply(storage); + await new AutoApprovalDrainer(storage, applyFn).drain(GK); + + expect(calls).toEqual([1, 3]); + expect(getAction(storage, 1).state).toBe("approved"); + expect(getAction(storage, 3).state).toBe("approved"); + }); + + it("halts when an apply fails, leaving it and everything after pending", async () => { + let storage = makeStorage(); + enableRule(storage); + putAction(storage, 1); + putAction(storage, 2); + putAction(storage, 3); + + let inner = makeImmediateApply(storage); + let applyFn: ApplyPendingActionFn = (record, resolvedBy, autoApproved) => { + if (record.id === 2) throw new Error("apply failed"); + return inner.applyFn(record, resolvedBy, autoApproved); + }; + await new AutoApprovalDrainer(storage, applyFn).drain(GK); + + expect(inner.calls).toEqual([1]); + expect(getAction(storage, 2).state).toBe("pending"); + expect(getAction(storage, 3).state).toBe("pending"); + }); + + // An action created after a drain snapshotted the pending index is out of that drain's scope; + // the creation path is responsible for its own drain() call (which the rerun flag folds in -- + // see the parked-mid-apply test above). + it("leaves actions created after the drain's snapshot for their own drain call", async () => { + let storage = makeStorage(); + enableRule(storage); + putAction(storage, 1); + + let apply = makeControlledApply(storage); + let drainer = new AutoApprovalDrainer(storage, apply.applyFn); + let first = drainer.drain(GK); // snapshots pending = [1] + + putAction(storage, 2); // arrives mid-drain, with no accompanying drain() call + apply.releaseNext(); + await first; + + expect(apply.calls).toEqual([1]); + expect(getAction(storage, 2).state).toBe("pending"); + }); }); diff --git a/packages/workshop-backend/__tests__/fixtures.ts b/packages/workshop-backend/__tests__/fixtures.ts new file mode 100644 index 000000000..9afac58d0 --- /dev/null +++ b/packages/workshop-backend/__tests__/fixtures.ts @@ -0,0 +1,105 @@ +// Shared fixtures for the action-log test suites: the production overseer storage over a mock, +// a putAction record factory, and a fake overseer client forged over +// OverseerDurableObject.prototype.open. + +import { RpcStub as NativeRpcStub } from "cloudflare:workers"; +import { createTypedStorage, collection } from "@gadgets/typed-storage"; +import type { Collection, Singleton } from "@gadgets/typed-storage"; +import type { Overseer } from "@gadgets/workshop-shared/api"; +import { OverseerDurableObject, makeOverseerStorage } from "../src/overseer.js"; +import type { ActionRecord } from "../src/overseer.js"; +import { makeMockStorage } from "./mock-storage.js"; + +/** + * The production schema over mock storage, so the action suites (auto-approval drain, pending + * history query) exercise the shipped actions collection and pendingByGatekeeper index rather + * than a copy. + */ +export function makeActionStorage(mockStorage = makeMockStorage()) { + return makeOverseerStorage(mockStorage); +} + +export type ActionTestStorage = ReturnType; + +/** + * Storage over the same records but without the index declared, for simulating a workspace + * written before the index existed (see the migration-backfill tests). + */ +export function makePreIndexActionStorage(mockStorage: DurableObjectStorage) { + return createTypedStorage(mockStorage, { + singletons: { nextActionId: 0 }, + collections: { actions: collection()({ primaryKey: "id" }) }, + }); +} + +/** Base timestamp for fixture records: putAction stamps createdAt = FIXTURE_EPOCH + id. */ +export const FIXTURE_EPOCH = 1700000000000; + +/** Puts a record and keeps nextActionId ahead of it, as the real allocator does. */ +export function putAction( + storage: { actions: Collection, nextActionId: Singleton }, + id: number, + opts: { state?: ActionRecord["state"], type?: ActionRecord["type"], gatekeeperId?: number, + actionTag?: string, autoApprovable?: boolean, createdAt?: Date, + appliedAt?: Date } = {}) { + let base = { + id, + gatekeeperId: opts.gatekeeperId ?? 1, + caller: { from: "agent", chatId: 1 } as const, + resourceTitle: `Resource ${id}`, + createdAt: opts.createdAt ?? new Date(FIXTURE_EPOCH + id), + ...(opts.appliedAt !== undefined ? { appliedAt: opts.appliedAt } : {}), + state: opts.state ?? "pending", + }; + let description = { title: `Action ${id}`, description: `Action ${id} description` }; + let type = opts.type ?? "action"; + storage.actions.put( + type === "action" ? { ...base, type, action: id, description: { + ...description, + implementsRevert: true, + actionKind: { tag: opts.actionTag ?? "edit", label: "Edits" }, + autoApprovable: opts.autoApprovable ?? true, + } } + : type === "observation" ? { ...base, type, description } + : { ...base, type, description, enabled: true }); + if (id >= storage.nextActionId.get()) storage.nextActionId.put(id + 1); +} + +/** + * Forges a client interface over the given storage via open(). `role` picks the returned + * interface class ("build" opens as the owner); `exports` supplies any ctx.exports entries the + * exercised paths dereference. + */ +export async function openFakeOverseer( + storage: object, + opts: { role?: "build" | "use", exports?: object } = {}): Promise { + let role = opts.role ?? "build"; + let ownerId = "owner-id"; + let userId = role === "build" ? ownerId : "viewer-id"; + let overseer = { + open: OverseerDurableObject.prototype.open, + impl: { + ownerId, + ensureAmbientCapsules: async () => {}, + markOutputsDirty: () => {}, + joinPresence: () => () => {}, + joinOutputsFanout: () => () => {}, + ensureObserver: async () => {}, + syncOutputsTo: async () => {}, + getSharingManager: async () => ({ getEffectiveRole: () => role }), + ctx: { id: { toString: () => "workspace-id" }, exports: opts.exports ?? {} }, + users: { + idFromString: (id: string) => id, + get: () => ({ + whoami: async () => ({ id: "profile-id", name: "Test User" }), + recordSharedGadgetOpen: async () => {}, + }), + }, + storage: Object.assign(storage, { + prohibitAllSharing: { get: () => false }, + title: { get: () => "Test Workspace" }, + }), + }, + } satisfies Pick & { impl: object }; + return overseer.open(userId, `${userId}-profile`, new NativeRpcStub<() => void>(() => {})); +} diff --git a/packages/workshop-backend/__tests__/git-migration-do.test.ts b/packages/workshop-backend/__tests__/git-migration-do.test.ts index 0df536374..326f214ab 100644 --- a/packages/workshop-backend/__tests__/git-migration-do.test.ts +++ b/packages/workshop-backend/__tests__/git-migration-do.test.ts @@ -7,10 +7,13 @@ // // This lives in __tests__/ (the unit workerd config), not __integration__/: the TEST_OVERSEER // DO binding exists only in vitest.config.ts, and no public API path can create a legacy -// workspace anymore (new workspaces are born at version 2), so seeding must reach into +// workspace anymore (new workspaces are born at version 3), so seeding must reach into // impl.storage -- the same pattern as chat-changes.test.ts. The public DO surface (open() etc.) -// is deliberately never called: #initializeNewWorkspace would stamp version 2 and shadow the +// is deliberately never called: #initializeNewWorkspace would stamp version 3 and shadow the // scenario. +// +// The version-3 action-index backfill rides the same constructor trigger, so its tests live +// here too. import { describe, expect, it } from "vitest"; import { env } from "cloudflare:workers"; @@ -20,6 +23,7 @@ import { HISTORY_COMMIT_GAP_MS } from "../src/git-migration"; import { LegacyWorkspace, MINUTE, T0, USER, expectHeadsMatchDoc, readDocFiles, setFile, } from "./legacy-workspace"; +import { makePreIndexActionStorage, putAction } from "./fixtures.js"; declare module "cloudflare:workers" { interface ProvidedEnv { @@ -80,13 +84,21 @@ describe("git-storage migration via the Overseer constructor", () => { setFile(doc, "", "app.js", "hello\nworld\n"); setFile(doc, "", "util.js", "util two\n"); }); // v6 + + // A pending action written through an index-less view of the same storage, simulating a + // record that predates the pendingByGatekeeper declaration. The version-3 step (chained + // after the git migration in the same blockConcurrencyWhile) must backfill it. + putAction(makePreIndexActionStorage(impl.ctx.storage), 1); }); await abortAllDurableObjects(); await inOverseer("git-migration-single", async impl => { - // The constructor's blockConcurrencyWhile completed before this event was delivered. - expect(impl.storage.version.get()).toBe(2); + // The constructor's blockConcurrencyWhile completed before this event was delivered, + // running the whole migration ladder: git storage (2), then the action indexes (3). + expect(impl.storage.version.get()).toBe(3); + expect([...impl.storage.actions.pendingByGatekeeper.list()].map((r: any) => r.id)) + .toEqual([1]); await expectHeadsMatchDoc(impl.storage, impl.gitStore, ws.docAt("current"), 1); @@ -137,7 +149,7 @@ describe("git-storage migration via the Overseer constructor", () => { await abortAllDurableObjects(); await inOverseer("git-migration-multi", async impl => { - expect(impl.storage.version.get()).toBe(2); + expect(impl.storage.version.get()).toBe(3); // Every gadget's head equals its own root's content in an independent replay of the log. await expectHeadsMatchDoc(impl.storage, impl.gitStore, ws.docAt("current"), 1); @@ -163,3 +175,50 @@ describe("git-storage migration via the Overseer constructor", () => { }); }); }); + +describe("action-index backfills via the Overseer constructor", () => { + it("backfills a version-2 workspace's indexes and stamps version 3", async () => { + await inOverseer("pending-index-v2", async impl => { + expect(impl.storage.version.get()).toBe(0); + // Seed through an index-less view of the same real storage, simulating records written + // before the action indexes were declared (their entries only exist for writes made after + // the declarations). + let legacy = makePreIndexActionStorage(impl.ctx.storage); + putAction(legacy, 1); + putAction(legacy, 2, { state: "approved" }); + putAction(legacy, 3, { gatekeeperId: 2 }); + // Last write: arm the constructor's version-2 migration. + impl.storage.version.put(2); + }); + + await abortAllDurableObjects(); + + await inOverseer("pending-index-v2", async impl => { + expect(impl.storage.version.get()).toBe(3); + // The pending index sees exactly the pendings (grouped by gatekeeper, so 1 before 3 here). + expect([...impl.storage.actions.pendingByGatekeeper.list()].map((r: any) => r.id)) + .toEqual([1, 3]); + // The history-filter index serves every key over the seeded records. + expect([...impl.storage.actions.byHistoryFilter.get("action")].map((r: any) => r.id)) + .toEqual([1, 2, 3]); + expect([...impl.storage.actions.byHistoryFilter.get("pending")].map((r: any) => r.id)) + .toEqual([1, 3]); + // The last-changed index covers the whole log, in change-time order. + expect([...impl.storage.actions.byLastChanged.list()].map((r: any) => r.id)) + .toEqual([1, 2, 3]); + + // Resolving a backfilled record must not throw on the index updates -- the failure mode + // that makes these backfills mandatory rather than an optimization. + let record = impl.storage.actions.get(1)!; + record.state = "approved"; + record.appliedAt = new Date(); + impl.storage.actions.put(record); + expect([...impl.storage.actions.pendingByGatekeeper.list()].map((r: any) => r.id)) + .toEqual([3]); + expect([...impl.storage.actions.byHistoryFilter.get("pending")].map((r: any) => r.id)) + .toEqual([3]); + expect([...impl.storage.actions.byLastChanged.list()].map((r: any) => r.id)) + .toEqual([2, 3, 1]); + }); + }); +}); diff --git a/packages/workshop-backend/__tests__/overseer-hooks.test.ts b/packages/workshop-backend/__tests__/overseer-hooks.test.ts index 96bf7a5ec..263bfcd0f 100644 --- a/packages/workshop-backend/__tests__/overseer-hooks.test.ts +++ b/packages/workshop-backend/__tests__/overseer-hooks.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it, vi } from "vitest"; -import { RpcStub as NativeRpcStub } from "cloudflare:workers"; import { DEFAULT_ADMIN_CONFIG, serializeAdminConfig } from "../src/admin-config.js"; import { OverseerDurableObject } from "../src/overseer.js"; +import { openFakeOverseer } from "./fixtures.js"; vi.mock("capnweb-validate", () => ({ validateRpc: () => () => undefined })); @@ -108,33 +108,10 @@ async function makeTargetOverseer(gadgetId?: number) { description: {title: "Incoming email", description: "Receives email"}, enabled: false, }; - let overseer = { - open: OverseerDurableObject.prototype.open, - impl: { - ownerId: "user-id", - ensureAmbientCapsules: async () => {}, - markOutputsDirty: () => {}, - joinPresence: () => () => {}, - joinOutputsFanout: () => () => {}, - users: { - idFromString: (id: string) => id, - get: () => ({ - whoami: async () => ({id: "profile-id", name: "Test User"}), - }), - }, - ctx: { - id: {toString: () => "workspace-id"}, - exports: {GatekeeperHookLoopback: ({props}: {props: object}) => props}, - }, - storage: { - prohibitAllSharing: {get: () => false}, - boundHooks: {get: () => record, put: vi.fn()}, - actions: {get: () => undefined, put: vi.fn()}, - }, - }, - } satisfies Pick & {impl: object}; - let notifyClosed = new NativeRpcStub<() => void>(() => {}); - let client = await overseer.open("user-id", "profile-id", notifyClosed); + let client = await openFakeOverseer({ + boundHooks: {get: () => record, put: vi.fn()}, + actions: {get: () => undefined, put: vi.fn()}, + }, {exports: {GatekeeperHookLoopback: ({props}: {props: object}) => props}}); return {client, controllerEnable}; } diff --git a/packages/workshop-backend/src/auto-approval.ts b/packages/workshop-backend/src/auto-approval.ts index 0c1fa032f..ba3e1078d 100644 --- a/packages/workshop-backend/src/auto-approval.ts +++ b/packages/workshop-backend/src/auto-approval.ts @@ -1,9 +1,9 @@ -// Auto-approval drain core: applies eligible pending actions in id order, with a per-gatekeeper -// single-flight guard so two concurrent drains (the DO's input gate is open across the apply await) -// can't double-apply the same action. The apply is injected, keeping this constructible over a -// mock storage in tests. +// Auto-approval drain core: applies the gatekeeper's eligible pending actions (read off the sparse +// pendingByGatekeeper index) in id order, with a per-gatekeeper single-flight guard so two +// concurrent drains (the DO's input gate is open across the apply await) can't double-apply the +// same action. The apply is injected, keeping this constructible over a mock storage in tests. -import type { Collection } from "@gadgets/typed-storage"; +import type { Collection, NonUniqueIndex } from "@gadgets/typed-storage"; import type { AiChatAuthorInfo } from "@gadgets/workshop-shared/api"; import { createWorkshopLogger } from "./observability"; import type { ActionRecord, AutoApproveTagRecord } from "./overseer.js"; @@ -11,7 +11,8 @@ import type { ActionRecord, AutoApproveTagRecord } from "./overseer.js"; const logger = createWorkshopLogger("workshop.auto.approval"); export interface AutoApprovalStorage { - actions: Collection; + actions: Collection + & { pendingByGatekeeper: NonUniqueIndex }; autoApproveTags: Collection; } @@ -50,28 +51,29 @@ export class AutoApprovalDrainer { } } - // Apply all currently-eligible pending actions of the gatekeeper, in ascending id order. Stops at - // the first pending action that is NOT auto-eligible (a manual gate) or that throws while applying - // -- it is never skipped ahead of. This preserves in-order application and the invariant that - // nothing is silently applied past a human gate. + // Apply all currently-eligible pending actions of the gatekeeper, in ascending id order. Stops + // at the first pending action that is NOT auto-eligible (a manual gate) or that throws while + // applying -- it is never skipped ahead of. This preserves in-order application and the + // invariant that nothing is silently applied past a human gate. // // Eligibility requires BOTH signals: the author's `autoApprovable` verdict on the action AND a // user-enabled rule for the action's type on this gatekeeper. async #drainOnce(gatekeeperId: number): Promise { - // Materialize a snapshot first: list() is a lazy generator over storage, and we mutate the - // actions collection (via applyPendingAction) as we go. - let pending = [...this.storage.actions.list()].filter( - (rec): rec is ActionRecord & {type: "action"} => - rec.gatekeeperId === gatekeeperId && rec.type === "action" && rec.state === "pending"); + // Materialize before applying: the index yields lazily in ascending id order, and applying + // mutates it mid-iteration. Actions created after this snapshot trigger their own drain(), + // which drain()'s rerun flag folds into this run if it's still in flight. + let pending = [...this.storage.actions.pendingByGatekeeper.get(gatekeeperId)]; for (let record of pending) { + if (record.type !== "action") continue; + let tag = record.description.actionKind?.tag; let rule = tag !== undefined ? this.storage.autoApproveTags.get(`${gatekeeperId}:${tag}`) : undefined; if (record.description.autoApprovable !== true || rule === undefined) { // A manual gate. Stop rather than skipping ahead to any later auto-eligible action. - break; + return; } // Re-check immediately before applying, to guard against a concurrent drain having already @@ -90,7 +92,7 @@ export class AutoApprovalDrainer { logger.error("auto-approval failed", { event: "auto.approval.failed", actionId: fresh.id, error: err, }); - break; + return; } } } diff --git a/packages/workshop-backend/src/overseer.ts b/packages/workshop-backend/src/overseer.ts index 1f2c450b9..8fded60d7 100644 --- a/packages/workshop-backend/src/overseer.ts +++ b/packages/workshop-backend/src/overseer.ts @@ -1,6 +1,6 @@ import { RpcCompatible, RpcStub, RpcTarget } from "capnweb"; import { validateRpc } from "capnweb-validate"; -import { Overseer, GadgetMetadata, UiBundle, WorkpieceId, WorkpieceSummary, WorkpiecesSubscriber, GadgetClient, GadgetBindingInfo, GatekeeperClient, ActionState, ActionLogEntry, ActionsSubscriber, ChatGadgetPin, ChatCodeBase, ChatGadgetPinState, CodeChangeSubmission, CommitIdentity, CommitInfo, MergeChangesResult, AiChatMetadata, AiChatMessage, AiChatHistoryPage, AiChatSubscriber, AiChatAuthorInfo, AiModelConfig, AiChatMessageBody, AgentSpawnerConfig, ConsoleLogSubscriber, ConsoleLogEvent, CapsuleSpecifier, CollaboratorInfo, CollaboratorRole, AffectedCollaborator, ShareLinkInfo, GatekeeperCreationSpec, ObserverConfigCallback, ObserverBindingNeed, ObserverBindingFailure, BlueprintBindingAnnotation, BlueprintBinding, BlueprintMetadata, BlueprintOutput, MessageFormatRef, isOutputIcon, SpawnerEnvTarget, BlueprintGadgetSummary, AiChatStreamEvent, BlueprintScreenshotUpload, BLUEPRINT_SCREENSHOT_R2_PREFIX, blueprintScreenshotUrl, ChatAttachmentUpload, ChatAttachmentHandle, ChatAttachmentRef, BoundHookInfo, PreApprovableAction, PresenceParticipant, PresenceSubscriber, SlashCommandChoice, SlashCommandRequest, validateBindingName, createOpenGadgetError, OPEN_GADGET_ERROR_CODES, resolveSiteName } from '@gadgets/workshop-shared/api'; +import { Overseer, GadgetMetadata, UiBundle, WorkpieceId, WorkpieceSummary, WorkpiecesSubscriber, GadgetClient, GadgetBindingInfo, GatekeeperClient, ActionState, ActionLogEntry, ActionsSubscriber, ActionHistoryFilter, ActionHistoryPage, ChatGadgetPin, ChatCodeBase, ChatGadgetPinState, CodeChangeSubmission, CommitIdentity, CommitInfo, MergeChangesResult, AiChatMetadata, AiChatMessage, AiChatHistoryPage, AiChatSubscriber, AiChatAuthorInfo, AiModelConfig, AiChatMessageBody, AgentSpawnerConfig, ConsoleLogSubscriber, ConsoleLogEvent, CapsuleSpecifier, CollaboratorInfo, CollaboratorRole, AffectedCollaborator, ShareLinkInfo, GatekeeperCreationSpec, ObserverConfigCallback, ObserverBindingNeed, ObserverBindingFailure, BlueprintBindingAnnotation, BlueprintBinding, BlueprintMetadata, BlueprintOutput, MessageFormatRef, isOutputIcon, SpawnerEnvTarget, BlueprintGadgetSummary, AiChatStreamEvent, BlueprintScreenshotUpload, BLUEPRINT_SCREENSHOT_R2_PREFIX, blueprintScreenshotUrl, ChatAttachmentUpload, ChatAttachmentHandle, ChatAttachmentRef, BoundHookInfo, PreApprovableAction, PresenceParticipant, PresenceSubscriber, SlashCommandChoice, SlashCommandRequest, validateBindingName, createOpenGadgetError, OPEN_GADGET_ERROR_CODES, resolveSiteName } from '@gadgets/workshop-shared/api'; import { applyCodeChange, changedGadgets, composeCodeChange, diffFiles, transformCodeChange, validateCodeChangeContent, validateCodeChangeSchema, type CodeContent, type CodeChange } from "@gadgets/workshop-shared/code-change"; @@ -10,6 +10,7 @@ import { RpcTarget as NativeRpcTarget, restore, } from "cloudflare:workers"; import { createTypedStorage, collection, keyString } from "@gadgets/typed-storage"; +import type { ListOptions } from "@gadgets/typed-storage"; import { GitStore, commitIdentityForAuthor, filesEqual, gitObjectsCollection, threeWayMerge } from "./git-store"; import { migrateCodeLogToGit } from "./git-migration"; @@ -547,6 +548,14 @@ export type ActionRecord = { resourceTitle?: string; // denormalized to avoid gatekeeper query resourceUrl?: string; // denormalized to avoid gatekeeper query createdAt: Date; + + /** + * When the record last changed state: an action's approval/rejection, a hook's enable/disable + * toggle or deletion. Absent while nothing has happened since creation (and on legacy records + * from before it was tracked). + */ + appliedAt?: Date; + state: ActionState; /** @@ -556,7 +565,6 @@ export type ActionRecord = { bindingName?: string; } & ({ type: "action"; - appliedAt?: Date; action: number; // action key assigned by the gatekeeper, passed back on apply/reject/revert description: ActionDescription; resolvedBy?: AiChatAuthorInfo; // set when resolved (approved/rejected); absent while pending (or legacy) @@ -865,10 +873,9 @@ async function computeSessionAffinity(gadgetId: string, chatId: number): Promise } function actionRecordToLog(record: ActionRecord): ActionLogEntry { - // TODO: ActionRecord and ActionLogEntry are almost identical. The main differences are: - // - ActionRecord includes `appliedAt` only when type == "action". ActionLogEntry could match. - // - ActionRecord includes `action`, which should NOT be provided to the client. - // We could make the two match more -- just `action` needs to be different. + // TODO: ActionRecord and ActionLogEntry are almost identical. The main difference is that + // ActionRecord includes `action`, which should NOT be provided to the client. We could make + // the two match more -- just `action` needs to be different. // ActionLogEntry omits the gatekeeperId for records that didn't come from a real gatekeeper // (built-in agent tools use the BUILTIN_TOOL_GATEKEEPER_ID sentinel). @@ -907,6 +914,7 @@ function actionRecordToLog(record: ActionRecord): ActionLogEntry { resourceTitle: record.resourceTitle || "(title unavailable)", resourceUrl: record.resourceUrl, createdAt: record.createdAt, + appliedAt: record.appliedAt, state: record.state, type: "bindHook", hookId: record.hookId, @@ -919,6 +927,25 @@ function actionRecordToLog(record: ActionRecord): ActionLogEntry { } } +// Reflect a hook toggle (or deletion, which also severs the hookId reference) onto the hook's +// bindHook action record, stamping the state-change time the byLastChanged index keys on. +function stampBindHookAction(storage: OverseerStorage, actionId: number, enabled: boolean, + opts?: {clearHookId?: boolean}): void { + let actionRecord = storage.actions.get(actionId); + if (actionRecord?.type !== "bindHook") return; + actionRecord.enabled = enabled; + if (opts?.clearHookId) delete actionRecord.hookId; + actionRecord.appliedAt = new Date(); + storage.actions.put(actionRecord); +} + +// Key of the actions `byLastChanged` index: last state-change time, id-disambiguated because the +// frozen clock makes same-instant records routine. Every mutation path stamps appliedAt (apply, +// reject, stampBindHookAction); one that doesn't would be missed by the resume replay. +function actionLastChangedKey(record: ActionRecord): string { + return `${keyString((record.appliedAt ?? record.createdAt).valueOf())}.${keyString(record.id)}`; +} + /** * One incremental update in the workspace-wide Yjs code log (the `code` and `snapshots` * collections). Formerly the public `CodeUpdate` wire type; the git-storage transition removed it @@ -967,6 +994,8 @@ export function makeOverseerStorage(storage: DurableObjectStorage) { // and every live chat was converted to the commit-pinned change stream (a // `conversionBoundary` changes message plus a `codeBase`). The `code`/`snapshots` // collections are dead stored data from this version on. + // 3 = the actions collection's indexes (pendingByGatekeeper, byHistoryFilter, + // byLastChanged) exist and are backfilled. version: 0, // The workspace title. (Each chat, gatekeeper, and gadget has its own title, elsewhere.) @@ -1076,7 +1105,29 @@ export function makeOverseerStorage(storage: DurableObjectStorage) { }), actions: collection()({ - primaryKey: "id" + primaryKey: "id", + + // All three indexes are backfilled by the version-3 migration. + uniqueIndexes: { + // Resume-replay index (see subscribeToActions): keyed by last state-change time so a + // reconnect replays only the records changed during the gap. + byLastChanged: actionLastChangedKey, + }, + + nonUniqueIndexes: { + // Sparse index over just the pending records, keyed by gatekeeper, so the auto-approval + // drain is O(pending on that gatekeeper) rather than a full-log scan. + pendingByGatekeeper(record: ActionRecord) { + return record.state === "pending" ? record.gatekeeperId : null; + }, + + // Keyed by the wire ActionHistoryFilter values, in lockstep with + // matchesActionHistoryFilter (api.ts), so every listActions() filter is one ranged + // read. The "all" filter has no key: it reads the collection itself. + byHistoryFilter(record: ActionRecord) { + return record.state === "pending" ? ["pending", record.type] : record.type; + }, + } }), boundHooks: collection()({ @@ -1298,6 +1349,15 @@ const LISTING_REFRESH_BATCH = 16; // Longest noun accepted on a format reference. Denormalized display data. const MAX_FORMAT_REF_NOUN = 128; +/** + * Raw records examined per page of subscribeToActions()'s startAfter resume replay. Exported + * for tests. + */ +export const ACTION_REPLAY_PAGE_SIZE = 256; + +/** listActions() entries returned per page. Exported for tests. */ +export const ACTION_HISTORY_PAGE_DEFAULT_LIMIT = 50; + /** * Keeps `commandPosition` only if it's a real index into `args`. Anything else becomes undefined, * and the command renders at the front. Display-only, so a bad value isn't worth an error. @@ -1695,9 +1755,12 @@ class OverseerImpl implements AgentHooks { // blocked event (including the alarm handler) is delivered. On failure there is nothing // to do -- blockConcurrencyWhile has already aborted the DO -- but the rejection must be // consumed so it doesn't also surface as an unhandled rejection. - this.ctx.blockConcurrencyWhile(() => this.#migrateToGitStorage()) - .then(() => this.#resumeInterruptedAgents(), () => {}); + this.ctx.blockConcurrencyWhile(async () => { + await this.#migrateToGitStorage(); + this.#migrateToActionIndexes(); + }).then(() => this.#resumeInterruptedAgents(), () => {}); } else { + this.#migrateToActionIndexes(); this.#resumeInterruptedAgents(); } } @@ -1758,6 +1821,26 @@ class OverseerImpl implements AgentHooks { }); } + // Version 2 -> 3: backfill the actions indexes. Indexes are only maintained at write time, so + // over records that predate their declaration they start empty -- and updating a pre-existing + // action would then throw on the index update. Runs synchronously in the constructor (chained + // after the git-storage migration when that one is still pending), so nothing can observe + // pre-migration state; transactionSync makes rebuilds-plus-stamp atomic, so a crash + // mid-rebuild retries whole. The `!== 2` guard keeps never-initialized DOs write-free (they + // stamp the current version at first initialization). + #migrateToActionIndexes(): void { + if (this.storage.version.get() !== 2) return; + this.ctx.storage.transactionSync(() => { + this.storage.actions.pendingByGatekeeper.rebuild(); + this.storage.actions.byHistoryFilter.rebuild(); + this.storage.actions.byLastChanged.rebuild(); + this.storage.version.put(3); + }); + this.logger.info("backfilled the action-log indexes", { + event: "storage.migration.action-indexes.completed", + }); + } + // The workspace owner's commit identity, for commits synthesized by the git-storage migration. // A transient user-DO reset is retried once (pure read on a fresh-stub helper); anything past // that degrades to a placeholder rather than failing: identity on synthesized history is @@ -2273,12 +2356,7 @@ class OverseerImpl implements AgentHooks { } this.storage.boundHooks.delete(record.id); - let actionRecord = this.storage.actions.get(record.actionId); - if (actionRecord?.type === "bindHook") { - actionRecord.enabled = false; - delete actionRecord.hookId; - this.storage.actions.put(actionRecord); - } + stampBindHookAction(this.storage, record.actionId, false, {clearHookId: true}); } // Subscribe to the workspace's workpiece list. In v1 only gadget-type workpieces are published. @@ -8146,7 +8224,7 @@ export class OverseerDurableObject extends DurableObject { // A workspace initialized by this version of the code is born at the current schema version; // there is nothing to migrate. - this.impl.storage.version.put(2); + this.impl.storage.version.put(3); } /** @@ -9364,13 +9442,26 @@ class OverseerClientInterface extends RpcTarget implements Overseer { return result; } - async listActions(): Promise { - let result: ActionLogEntry[] = []; - for (let record of this.impl.storage.actions.list()) { - result.push(actionRecordToLog(record)); + async listActions(options?: {beforeId?: number, filter?: ActionHistoryFilter}) + : Promise { + let {beforeId, filter = "all"} = options ?? {}; + if (beforeId !== undefined && (!Number.isSafeInteger(beforeId) || beforeId < 0)) { + throw new TypeError(`Invalid beforeId: ${beforeId}`); } - return result; + // One ranged read -- off the collection itself for "all" (already id-ordered), off + // byHistoryFilter otherwise -- so the work is O(page) however sparse the matches. Pages are + // full until the last; the +1 record probes whether an older page exists. + let actions = this.impl.storage.actions; + let range = {end: beforeId, reverse: true, limit: ACTION_HISTORY_PAGE_DEFAULT_LIMIT + 1}; + let page = [...(filter === "all" + ? actions.list(range) : actions.byHistoryFilter.get(filter, range))]; + let more = page.length > ACTION_HISTORY_PAGE_DEFAULT_LIMIT; + if (more) page.pop(); + return { + entries: page.map(actionRecordToLog), + nextBeforeId: more ? page.at(-1)!.id : undefined, + }; } async approveAction(id: number): Promise { @@ -9451,12 +9542,7 @@ class OverseerClientInterface extends RpcTarget implements Overseer { record.enabled = true; this.impl.storage.boundHooks.put(record); - - let actionRecord = this.impl.storage.actions.get(record.actionId); - if (actionRecord?.type === "bindHook") { - actionRecord.enabled = true; - this.impl.storage.actions.put(actionRecord); - } + stampBindHookAction(this.impl.storage, record.actionId, true); } } @@ -9469,12 +9555,7 @@ class OverseerClientInterface extends RpcTarget implements Overseer { record.enabled = false; this.impl.storage.boundHooks.put(record); - - let actionRecord = this.impl.storage.actions.get(record.actionId); - if (actionRecord?.type === "bindHook") { - actionRecord.enabled = false; - this.impl.storage.actions.put(actionRecord); - } + stampBindHookAction(this.impl.storage, record.actionId, false); } } @@ -9759,17 +9840,35 @@ class OverseerClientInterface extends RpcTarget implements Overseer { actions.subscribe(dbSubscriber); subscribed = true; - // Replay actions changed since `startAfter`; resolved actions use `appliedAt`, - // pending actions use `createdAt`. + // The subscription delivers live deltas only; clients query current pending state via + // listActions({filter: "pending"}) after initiating the subscribe (see api.ts). if (startAfter !== undefined) { - let startAfterTimestamp = startAfter.valueOf(); - for (let record of actions.list()) { - if (disposed) break; - let appliedAt = record.type === "action" ? record.appliedAt : undefined; - let recordTimestamp = (appliedAt ?? record.createdAt).valueOf(); - if (recordTimestamp > startAfterTimestamp) { - subscriber.entry(actionRecordToLog(record)).catch(unsubscribe); + // Resubscribe after a disconnect: sweep byLastChanged for everything changed since the + // client's last-seen time -- O(changed during the gap), not O(log). The bound is + // inclusive: the frozen clock stamps whole batches with one instant, so an exclusive bound + // would drop the last-seen record's siblings, while re-delivery is just a harmless upsert. + // The end key is fixed up front; a record changing mid-replay re-sorts past it and arrives + // via the live subscription instead. Each page's delivery is awaited, so a failure rejects + // the subscribe call before ready() and a huge gap can't queue unbounded callbacks. + try { + let newest = [...actions.byLastChanged.list({reverse: true, limit: 1})].at(0); + if (newest !== undefined) { + let end = actionLastChangedKey({...newest, id: newest.id + 1}); + // keyString(t) is a prefix of every key with that timestamp, so `start` is inclusive + // of the whole cutoff instant. + let from: ListOptions = {start: keyString(startAfter.valueOf())}; + for (;;) { + if (disposed) throw new Error("Action subscriber failed during replay"); + let page = [...actions.byLastChanged.list( + {...from, end, limit: ACTION_REPLAY_PAGE_SIZE})]; + await Promise.all(page.map(record => subscriber.entry(actionRecordToLog(record)))); + if (page.length < ACTION_REPLAY_PAGE_SIZE) break; + from = {startAfter: actionLastChangedKey(page.at(-1)!)}; + } } + } catch (err) { + unsubscribe(); + throw err; // rejecting the subscribe call is the client's error signal } } @@ -9858,7 +9957,7 @@ class OverseerClientInterface extends RpcTarget implements Overseer { end: beforeSequence === undefined ? undefined : compactionKey(chatId, beforeSequence), })]; return { - messages: await Promise.all(result.map((msg) => this.#getChatMessageForClient(msg))), + messages: result.map((msg) => this.#getChatMessageForClient(msg)), compacted: checkpoint && { to: checkpoint.compactedTo, summary: checkpoint.summary, @@ -9872,7 +9971,7 @@ class OverseerClientInterface extends RpcTarget implements Overseer { return msg && this.#getChatMessageForClient(msg); } - async #getChatMessageForClient(msg: AiChatMessage): Promise { + #getChatMessageForClient(msg: AiChatMessage): AiChatMessage { if (msg.type === "action") { let record = this.impl.storage.actions.get(msg.actionId); if (record) { @@ -9911,18 +10010,11 @@ class OverseerClientInterface extends RpcTarget implements Overseer { let self = this; function deliverMessage(record: AiChatMessage) { - subscriber.message(self.impl.hydrateChatMessageForClient(record)).catch(unsubscribe); + subscriber.message(self.#getChatMessageForClient(record)).catch(unsubscribe); } let msgSubscriber = { add(record: AiChatMessage) { - if (record.type == "action") { - let actionRecord = self.impl.storage.actions.get(record.actionId); - if (actionRecord) { - record.actionLog = actionRecordToLog(actionRecord); - } - } - deliverMessage(record); }, update(oldRecord: AiChatMessage, newRecord: AiChatMessage): void { @@ -10437,10 +10529,11 @@ class OverseerClientInterface extends RpcTarget implements Overseer { // subscribeToMetadata(), subscribeToPresence(), subscribeToWorkpieces(), and getGadget() // (returning a restricted, mainline-only UseGadgetClientInterface). Presence includes active // viewers' names, profile IDs, and roles. Every other -// method throws "Unauthorized", with two exceptions: subscribeToConsoleLogs() and -// subscribeToActions() return inert subscriptions (they never deliver data) rather than denying. -// The editor subscribes to both speculatively from its top-level hooks, before it has switched to -// the use-only view; an inert subscription lets those calls resolve quietly instead of surfacing +// method throws "Unauthorized", with a few exceptions: subscribeToConsoleLogs() and +// subscribeToActions() return inert subscriptions (they never deliver data), and +// listActions() returns an empty terminal page, rather than denying. +// The editor calls all of these speculatively from its top-level hooks, before it has switched to +// the use-only view; an inert result lets those calls resolve quietly instead of surfacing // as spurious client-side errors, while still revealing nothing to the "use" collaborator. // // Default-deny is enforced at compile time: because this class `implements Overseer`, adding any @@ -10587,7 +10680,12 @@ class UseOverseerInterface extends RpcTarget implements Overseer { async newAgentSpawnerGatekeeper(_config: AgentSpawnerConfig): Promise> { this.#deny(); } - async listActions(): Promise { this.#deny(); } + // Pending actions are queried eagerly for the badge; resolved history is demand-loaded. Return + // an empty terminal page so this speculative read does not fail for "use" collaborators. + async listActions(_options?: {beforeId?: number, filter?: ActionHistoryFilter}) + : Promise { + return {entries: []}; + } async approveAction(_id: number): Promise { this.#deny(); } async rejectAction(_id: number): Promise { this.#deny(); } async listHooks(): Promise { this.#deny(); } diff --git a/packages/workshop-frontend/src/Activity.tsx b/packages/workshop-frontend/src/Activity.tsx index f4ab21e53..31cf4ee24 100644 --- a/packages/workshop-frontend/src/Activity.tsx +++ b/packages/workshop-frontend/src/Activity.tsx @@ -9,6 +9,8 @@ import { HookToggle } from './components/HookToggle' import { AlwaysApproveButton, ResolveButton } from './components/ResolveButton' import { WorkshopButton } from './components/WorkshopControls' import { useActions } from './useActions' +import { useActionHistory } from './useActionHistory' +import type { HistoryViewFilter } from './useActionHistory' import { useAutoApproval, autoApprovalKey, type AutoApprovalEntry } from './useAutoApproval' import { useAlwaysApproveTag } from './useAlwaysApproveTag' import { useAuthenticatedApi } from './AuthContext' @@ -20,8 +22,6 @@ import AutoApproveConfirmDialog from './components/AutoApproveConfirmDialog' export type ActivityView = 'review' | 'history' | 'auto' -type HistoryFilter = 'all' | ActionLogEntry['type'] - const PANE_BAR = 'flex h-9 flex-shrink-0 items-center border-b border-kumo-line' interface ActivityProps { @@ -34,17 +34,18 @@ interface ActivityProps { autoApproveReloadTrigger?: number } -const HISTORY_FILTERS: { value: HistoryFilter; label: string }[] = [ +/** Pending-status copy while the pending set is still being gathered (also in the popover). */ +export const PENDING_CHECKING_COPY = 'Checking for requests…' +/** Pending-status copy when gathering the pending set failed (also in the popover). */ +export const PENDING_ERROR_COPY = 'Could not check for requests — reload the page to try again.' + +const HISTORY_FILTERS: { value: HistoryViewFilter; label: string }[] = [ { value: 'all', label: 'All' }, { value: 'action', label: 'Actions' }, { value: 'observation', label: 'Observations' }, { value: 'bindHook', label: 'Hooks' }, ] -function timeValue(date: Date | undefined): number { - return date ? new Date(date).getTime() : 0 -} - function formatClockTime(date: Date): string { return new Date(date).toLocaleTimeString([], { hour: 'numeric', minute: '2-digit' }) } @@ -94,7 +95,7 @@ function activityStatus( : { label: 'Disabled', dotClass: 'bg-kumo-inactive', textClass: 'text-kumo-subtle' } } if (record.state === 'pending') { - return { label: 'Waiting', dotClass: 'bg-kumo-brand', textClass: 'text-kumo-strong' } + return { label: 'Pending', dotClass: 'bg-kumo-brand', textClass: 'text-kumo-strong' } } if (record.state === 'rejected') { return { label: 'Denied', dotClass: 'bg-kumo-danger', textClass: 'text-kumo-danger' } @@ -109,6 +110,46 @@ function TypeIcon({ record, className }: { record: ActionLogEntry; className?: s return } +function LoadOlderButton({ history, className, label = 'Load older' }: { + history: { loadMore: () => void; isLoadingMore: boolean } + className?: string + label?: string +}) { + return ( + + {history.isLoadingMore ? 'Loading…' : label} + + ) +} + +/** Centered full-pane notice: an empty, error, or call-to-action state. */ +function ActivityNotice({ icon, title, description, children }: { + icon?: ReactNode + title: string + description?: string + children?: ReactNode +}) { + return ( +
+ {icon && ( + + {icon} + + )} +

+ {title} +

+ {description && ( +

+ {description} +

+ )} + {children} +
+ ) +} + export default function Activity({ overseer, view, @@ -116,8 +157,8 @@ export default function Activity({ onAutoApproveChange, autoApproveReloadTrigger, }: ActivityProps) { - const { actionsById, isReady } = useActions(overseer) - const [historyFilter, setHistoryFilter] = useState('all') + const { status: pendingStatus, pending: pendingActions } = useActions(overseer) + const [historyFilter, setHistoryFilter] = useState('all') const [processingActions, setProcessingActions] = useState>(new Set()) const [togglingHooks, setTogglingHooks] = useState>(new Set()) const [expandedActionId, setExpandedActionId] = useState(null) @@ -130,30 +171,20 @@ export default function Activity({ } | null>(null) const toasts = useKumoToastManager() - const { pendingActions, historyGroups, historyTotal, historyShown } = useMemo(() => { - const records = [...actionsById.values()] - const pending = records - .filter(record => record.state === 'pending') - .toSorted((a, b) => timeValue(a.createdAt) - timeValue(b.createdAt) || a.id - b.id) - const resolved = records.filter(record => record.state !== 'pending') - const filtered = resolved - .filter(record => historyFilter === 'all' || record.type === historyFilter) - .toSorted((a, b) => - timeValue(b.appliedAt ?? b.createdAt) - timeValue(a.appliedAt ?? a.createdAt) || b.id - a.id) + const history = useActionHistory(overseer, historyFilter, view === 'history') + + // Grouped by day in id order (newest first). A day label can repeat when resolution order + // differs from creation order — accepted for a paged, creation-ordered log. + const historyGroups = useMemo(() => { const groups: { label: string; records: ActionLogEntry[] }[] = [] - for (const record of filtered) { + for (const record of history.entries) { const label = dayLabel(record.appliedAt ?? record.createdAt) const last = groups.at(-1) if (last?.label === label) last.records.push(record) else groups.push({ label, records: [record] }) } - return { - pendingActions: pending, - historyGroups: groups, - historyTotal: resolved.length, - historyShown: filtered.length, - } - }, [actionsById, historyFilter]) + return groups + }, [history.entries]) const resolveAction = useResolveAction(overseer, setProcessingActions) @@ -181,151 +212,224 @@ export default function Activity({ setExpandedActionId(previous => (previous === id ? null : id)) } - if (!isReady) { + function renderReviewContent(): ReactNode { + if (pendingActions.length > 0) { + return ( + <> +
+ + {pendingActions.length} {pendingActions.length === 1 ? 'request' : 'requests'} waiting + + Oldest first +
+
+ {pendingActions.map(record => { + const autoApproveTarget = + record.type === 'action' && record.gatekeeperId !== undefined && + record.description.actionKind !== undefined && + record.description.autoApprovable === true + ? { + actionId: record.id, + gatekeeperId: record.gatekeeperId, + resourceTitle: record.resourceTitle, + actionKind: record.description.actionKind, + actionLabel: record.description.title, + } + : undefined + return ( + toggleExpanded(record.id)} + onApprove={() => void resolveAction(record.id, 'approve')} + onReject={() => void resolveAction(record.id, 'deny')} + onAlwaysApprove={ + autoApproveTarget && + !isTagAutoApproved(autoApproveTarget.gatekeeperId, autoApproveTarget.actionKind.tag) + ? () => setConfirmAutoApprove(autoApproveTarget) + : undefined + } + /> + ) + })} + {pendingStatus === 'checking' && ( +

+ Still checking older activity… +

+ )} + {pendingStatus === 'error' && ( +

+ Could not finish checking for requests — reload the page to try again. +

+ )} +
+ + ) + } + + if (pendingStatus === 'checking') { + return ( +
+ {PENDING_CHECKING_COPY} +
+ ) + } + + if (pendingStatus === 'error') { + return ( + + ) + } + return ( -
- Loading activity… -
+ } + title="Nothing to review" + description="Requests that need your approval show up here and in the workspace header." + > + onViewChange('history')}> + View history + + ) } - return ( -
- {view === 'review' ? ( - pendingActions.length === 0 ? ( -
- - - -

- Nothing to review -

-

- Requests that need your approval show up here and in the workspace header. -

- onViewChange('history')}> - View history - + function renderHistoryBody(): ReactNode { + if (history.entries.length > 0) { + return ( +
+
+ Time + Event + Status +
- ) : ( - <> -
- - {pendingActions.length} {pendingActions.length === 1 ? 'request' : 'requests'} waiting + {historyGroups.map(group => ( + // Keyed by the group's oldest record: live inserts land at the front of a group, so + // keying by the first would remount the section (dropping focus) on every insert. + // Day labels can repeat (see historyGroups), so the label alone can't be the key. +
+

+ {group.label} +

+ {group.records.map(record => ( + toggleExpanded(record.id)} + togglingHook={record.type === 'bindHook' && record.hookId !== undefined + ? togglingHooks.has(record.hookId) + : false} + onToggleHook={handleToggleHook} + /> + ))} +
+ ))} + {history.loadMoreFailed ? ( +
+ + Couldn't load older activity - Oldest first +
-
- {pendingActions.map(record => { - const autoApproveTarget = - record.type === 'action' && record.gatekeeperId !== undefined && - record.description.actionKind !== undefined && - record.description.autoApprovable === true - ? { - actionId: record.id, - gatekeeperId: record.gatekeeperId, - resourceTitle: record.resourceTitle, - actionKind: record.description.actionKind, - actionLabel: record.description.title, - } - : undefined - return ( - toggleExpanded(record.id)} - onApprove={() => void resolveAction(record.id, 'approve')} - onReject={() => void resolveAction(record.id, 'deny')} - onAlwaysApprove={ - autoApproveTarget && - !isTagAutoApproved(autoApproveTarget.gatekeeperId, autoApproveTarget.actionKind.tag) - ? () => setConfirmAutoApprove(autoApproveTarget) - : undefined - } - /> - ) - })} + ) : history.hasMore && ( +
+
- - ) - ) : view === 'history' ? ( - <> -
- {HISTORY_FILTERS.map(filter => ( - - ))} - - {historyShown} {historyShown === 1 ? 'event' : 'events'} - + )} +
+ ) + } -
+ if (history.status === 'error') { + return ( + + + + ) + } - {historyTotal === 0 ? ( -
-

- No activity yet -

-

- Every resource an agent reads or changes is recorded here. -

-
- ) : historyShown === 0 ? ( -
-

No matching events

- -
- ) : ( -
-
- Time - Event - Status - -
- {historyGroups.map(group => ( -
-

- {group.label} -

- {group.records.map(record => ( - toggleExpanded(record.id)} - togglingHook={record.type === 'bindHook' && record.hookId !== undefined - ? togglingHooks.has(record.hookId) - : false} - onToggleHook={handleToggleHook} - /> - ))} -
+ if (history.status === 'loading') { + return ( +
+ Loading activity… +
+ ) + } + + if (history.hasMore) { + return ( + + + + ) + } + + if (historyFilter === 'all') { + return ( + + ) + } + + return ( + + + + ) + } + + function renderActivityContent(): ReactNode { + switch (view) { + case 'review': + return renderReviewContent() + case 'history': + return ( + <> +
+ {HISTORY_FILTERS.map(filter => ( + ))} + + {history.entries.length} loaded +
- )} - - ) : ( - - )} + {renderHistoryBody()} + + ) + case 'auto': + return + } + } + + return ( +
+ {renderActivityContent()} {confirmAutoApprove && ( - pendingActions: ActionLogEntry[] onViewActivity: (view: ActivityView) => void } @@ -18,16 +23,12 @@ const PREVIEW_LIMIT = 3 export default function ActivityNotifications({ overseer, - pendingActions, onViewActivity, }: ActivityNotificationsProps) { const [open, setOpen] = useState(false) const [processing, setProcessing] = useState>(new Set()) const resolveAction = useResolveAction(overseer, setProcessing) - - const pending = useMemo(() => pendingActions.toSorted((a, b) => - new Date(a.createdAt).getTime() - new Date(b.createdAt).getTime() || a.id - b.id), - [pendingActions]) + const { status, pending } = useActions(overseer) const openFullView = (view: ActivityView) => { setOpen(false) @@ -69,7 +70,9 @@ export default function ActivityNotifications({ {pending.length === 0 ? (

- Nothing is waiting on you. + {status === 'error' ? PENDING_ERROR_COPY + : status === 'checking' ? PENDING_CHECKING_COPY + : 'Nothing is waiting on you.'}

) : (
diff --git a/packages/workshop-frontend/src/ChatInterface.tsx b/packages/workshop-frontend/src/ChatInterface.tsx index c5ae6faff..31e49fda3 100644 --- a/packages/workshop-frontend/src/ChatInterface.tsx +++ b/packages/workshop-frontend/src/ChatInterface.tsx @@ -5766,6 +5766,47 @@ function ChatInterface({ if (applyActionLogUpdateToCachedMessages(record)) scheduleUpdate(); }); + // On (re)connect, re-fetch cached action cards whose log can still change: blank or pending + // cards (a resolution may have landed while we were away), and bindHook cards, which stay + // mutable after resolution (`enabled` toggles). The action subscription carries live deltas + // only, so changes from the gap never reach us through it. + // TODO: resubscribing with startAfter would replay the gap through the subscription instead. + useEffect(() => { + let cancelled = false; + const targets = [...cacheRef.current.actionMessages.values()].flatMap((locations) => { + const location = locations.values().next().value; + const msg = location && getCachedActionMessage(location)?.msg; + return msg && (!msg.actionLog || msg.actionLog.state === "pending" || + msg.actionLog.type === "bindHook") ? [location] : []; + }); + + const refresh = async (location: { chatId: number; sequence: number }) => { + try { + const fetched = await overseer.getChatMessage(location.chatId, location.sequence); + if (cancelled || fetched?.type !== "action" || !fetched.actionLog) return; + // Resolution is monotonic: never regress a card another channel already resolved. + const current = getCachedActionMessage(location)?.msg; + if (fetched.actionLog.state === "pending" && + current?.actionLog && current.actionLog.state !== "pending") return; + if (applyActionLogUpdateToCachedMessages(fetched.actionLog)) scheduleUpdate(); + } catch (err) { + console.error("Failed to refresh action card:", err); + } + }; + + // A few at a time: a large cache refreshing all at once would flood the workspace DO. + let next = 0; + for (let i = Math.min(4, targets.length); i > 0; i--) { + void (async () => { + while (next < targets.length) { + if (cancelled) return; + await refresh(targets[next++]); + } + })(); + } + return () => { cancelled = true; }; + }, [overseer]); + // Reset per-chat UI state when selectedChatId changes useEffect(() => { setExpandedToolCalls(new Set()); @@ -6144,21 +6185,29 @@ function ChatInterface({ } }; + // Resolves an actionMessages location to the cached action message it points at (with its + // containing message array, for copy-on-write patches). Undefined if the cache no longer holds + // an action message there. + const getCachedActionMessage = (location: { chatId: number; sequence: number }) => { + const messages = cacheRef.current.messages.get(location.chatId); + const msg = messages?.[location.sequence]; + return msg?.type === "action" ? { messages: messages!, msg } : undefined; + }; + const applyActionLogUpdateToCachedMessages = (record: ActionLogEntry): boolean => { let changed = false; const locations = cacheRef.current.actionMessages.get(record.id); if (!locations) return false; for (const [key, location] of locations) { - const messages = cacheRef.current.messages.get(location.chatId); - const msg = messages?.[location.sequence]; - if (msg?.type !== "action" || msg.actionId !== record.id) { + const cached = getCachedActionMessage(location); + if (!cached || cached.msg.actionId !== record.id) { locations.delete(key); continue; } - const nextMessages = [...messages!]; - nextMessages[location.sequence] = { ...msg, actionLog: record }; + const nextMessages = [...cached.messages]; + nextMessages[location.sequence] = { ...cached.msg, actionLog: record }; cacheRef.current.messages.set(location.chatId, nextMessages); changed = true; } @@ -6173,17 +6222,16 @@ function ChatInterface({ if (!locations) return false; for (const [key, location] of locations) { - const messages = cacheRef.current.messages.get(location.chatId); - const msg = messages?.[location.sequence]; - if (msg?.type !== "action" || msg.actionId !== actionId || !msg.actionLog) { + const cached = getCachedActionMessage(location); + if (!cached || cached.msg.actionId !== actionId || !cached.msg.actionLog) { locations.delete(key); continue; } - const nextMessages = [...messages!]; + const nextMessages = [...cached.messages]; nextMessages[location.sequence] = { - ...msg, - actionLog: { ...msg.actionLog, state, appliedAt: new Date() }, + ...cached.msg, + actionLog: { ...cached.msg.actionLog, state, appliedAt: new Date() }, }; cacheRef.current.messages.set(location.chatId, nextMessages); changed = true; @@ -6199,17 +6247,16 @@ function ChatInterface({ if (!locations) return false; for (const [key, location] of locations) { - const messages = cacheRef.current.messages.get(location.chatId); - const msg = messages?.[location.sequence]; - if (msg?.type !== "action" || msg.actionId !== actionId || msg.actionLog?.type !== "bindHook") { + const cached = getCachedActionMessage(location); + if (!cached || cached.msg.actionId !== actionId || cached.msg.actionLog?.type !== "bindHook") { locations.delete(key); continue; } - const nextMessages = [...messages!]; + const nextMessages = [...cached.messages]; nextMessages[location.sequence] = { - ...msg, - actionLog: { ...msg.actionLog, enabled }, + ...cached.msg, + actionLog: { ...cached.msg.actionLog, enabled }, }; cacheRef.current.messages.set(location.chatId, nextMessages); changed = true; diff --git a/packages/workshop-frontend/src/GadgetEditor.tsx b/packages/workshop-frontend/src/GadgetEditor.tsx index 5339e0ef0..4226b835a 100644 --- a/packages/workshop-frontend/src/GadgetEditor.tsx +++ b/packages/workshop-frontend/src/GadgetEditor.tsx @@ -25,7 +25,6 @@ import { AiChatAuthorInfo, ConsoleLogSubscriber, ConsoleLogEvent, - ActionLogEntry, WorkpieceId, WorkpieceSummary, BlueprintOutput, @@ -56,7 +55,7 @@ import { GadgetPresence } from './components/GadgetPresence' import BlueprintModal from './BlueprintModal' import TopBarNotice from './TopBarNotice' import { WorkshopButton, WorkshopIconButton, WorkshopInput } from './components/WorkshopControls' -import { useActions } from './useActions' +import { useActionEntries, useActions } from './useActions' import DeleteConfirmationDialog from './components/DeleteConfirmationDialog' import ReconnectingChip from './components/ReconnectingChip' import WorkspaceOpenErrorPage from './components/WorkspaceOpenErrorPage' @@ -750,16 +749,24 @@ export default function GadgetEditor() { ? streamingActiveFile.filename : undefined - const { actionsById } = useActions(overseer?.stub ?? null) - // Hook bindings change once in a while, but `actionsById` is a fresh Map on every action-log - // frame. Track just the bindHook enable states so the refetch isn't driven at animation rate. - const hookSignature = useMemo(() => { - const parts: string[] = [] - for (const record of actionsById.values()) { - if (record.type === 'bindHook') parts.push(`${record.hookId}:${record.enabled}`) - } - return parts.join() - }, [actionsById]) + const overseerStub = overseer?.stub ?? null + const { pending: pendingActions } = useActions(overseerStub) + // Hook bindings change once in a while; fold the entry stream into a signature over just the + // bindHook enable states, in state only when it changes, so the refetch below isn't driven at + // animation rate. listHooks() is the authoritative initial source; entries only trigger + // refetches. useActionEntries replays already-received entries on mount, repopulating the ref + // after the reset when the stub changes. + const hookStatesRef = useRef(new Map()) + const [hookSignature, setHookSignature] = useState('') + useEffect(() => { + hookStatesRef.current = new Map() + setHookSignature('') + }, [overseerStub]) + useActionEntries(overseerStub, record => { + if (record.type !== 'bindHook') return + hookStatesRef.current.set(record.id, `${record.hookId}:${record.enabled}`) + setHookSignature([...hookStatesRef.current.values()].join()) + }) const [hookedGadgetIds, setHookedGadgetIds] = useState>(NO_GADGETS) useEffect(() => { if (!overseer || metadata === null || isUseOnly) return @@ -772,14 +779,7 @@ export default function GadgetEditor() { // Clear on teardown so a workspace switch never shows the previous workspace's indicators. return () => { cancelled = true; setHookedGadgetIds(NO_GADGETS) } }, [overseer, hookSignature, metadata !== null, isUseOnly]) - const pendingActions = useMemo(() => { - const pending: ActionLogEntry[] = [] - for (const record of actionsById.values()) { - if (record.state === 'pending') pending.push(record) - } - return pending - }, [actionsById]) - const pendingActionsCount = pendingActions.length + const pendingActionCount = pendingActions.length // Whether the *selected* gadget has code. When no gadget is selected, the code interface is // unmounted and raw `hasCode` can't update, but a gadget-less workspace has no code to show. @@ -1470,11 +1470,7 @@ export default function GadgetEditor() { )} - + {showReconnecting && } @@ -1545,14 +1541,14 @@ export default function GadgetEditor() { @@ -1760,7 +1756,7 @@ export default function GadgetEditor() { key={tab.value} active={activityView === tab.value} label={tab.label} - count={tab.value === 'review' ? pendingActionsCount : undefined} + count={tab.value === 'review' ? pendingActionCount : undefined} onClick={() => setActivityView(tab.value)} /> )) @@ -1813,7 +1809,7 @@ export default function GadgetEditor() { key={tab.value} active={activityView === tab.value} label={tab.label} - count={tab.value === 'review' ? pendingActionsCount : undefined} + count={tab.value === 'review' ? pendingActionCount : undefined} onClick={() => setActivityView(tab.value)} /> ))} @@ -1927,8 +1923,8 @@ export default function GadgetEditor() { onExpandedChange={handleWorkpieceRailExpandedChange} onSelect={handleSelectWorkpiece} onRename={handleRenameWorkpiece} - pendingActivityCount={pendingActionsCount} - onOpenActivity={() => openActivity(pendingActionsCount > 0 ? 'review' : 'history')} + pendingActivityCount={pendingActionCount} + onOpenActivity={() => openActivity(pendingActionCount > 0 ? 'review' : 'history')} />
)} diff --git a/packages/workshop-frontend/src/action-test-harness.ts b/packages/workshop-frontend/src/action-test-harness.ts new file mode 100644 index 000000000..5e5c796b4 --- /dev/null +++ b/packages/workshop-frontend/src/action-test-harness.ts @@ -0,0 +1,152 @@ +// Shared harness for the action-store hook tests (useActions / useActionHistory): act-enabled +// root management, manually-pumped rAF frames (the store coalesces entry commits through rAF), +// an ActionLogEntry factory, and a fake overseer covering the subscription and history-paging +// surfaces. Not a test file itself -- vitest only collects *.test.* -- so importing it is what +// installs the act environment and the rAF stub. + +import { act } from 'react' +import { createRoot, type Root } from 'react-dom/client' +import { vi } from 'vitest' +import type { RpcStub } from 'capnweb' +import type { + ActionHistoryPage, + ActionLogEntry, + ActionsSubscriber, + Overseer, +} from '@gadgets/workshop-shared/api' + +;(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true + +const rafQueue: FrameRequestCallback[] = [] +vi.stubGlobal('requestAnimationFrame', (cb: FrameRequestCallback) => rafQueue.push(cb)) + +/** Run every queued rAF callback inside act(). */ +export function flushFrames() { + act(() => { + while (rafQueue.length) rafQueue.shift()!(0) + }) +} + +export function entry(id: number, over: Partial> = {}): ActionLogEntry { + return { + id, + resourceTitle: `Resource ${id}`, + createdAt: new Date(1700000000000 + id * 60_000), + state: 'pending', + type: 'action', + description: { title: `Action ${id}`, description: '', implementsRevert: false }, + ...over, + } as ActionLogEntry +} + +type ListOptions = Parameters[0] + +type Parked = { resolve: (value: T) => void, reject: (err: unknown) => void } + +/** + * A queue of parked listActions calls: park() records the call and returns a promise that stays + * unsettled until resolveNext()/rejectNext() settles the oldest one inside act(). + */ +function parkedQueue() { + const calls: ListOptions[] = [] + const parked: Array> = [] + return { + calls, + park(call: ListOptions) { + calls.push(call) + return new Promise((resolve, reject) => parked.push({ resolve, reject })) + }, + async resolveNext(page: ActionHistoryPage) { + await act(async () => { parked.shift()!.resolve(page) }) + }, + async rejectNext(err: unknown) { + await act(async () => { parked.shift()!.reject(err) }) + }, + } +} + +/** + * Mocks the server side of the action APIs. subscribeToActions: live records are pushed through + * the captured subscriber's entry() via emit(); the call itself parks until + * resolveSubscription()/rejectSubscription(). listActions: each call parks — the shared store's + * pending queries ({filter: 'pending'}) on their own queue drained by + * resolvePendingQuery()/rejectPendingQuery(), every other filter on the history queue drained by + * resolvePage()/rejectPage(). `ops` records the initiation order across all three surfaces. + */ +export function makeOverseer() { + const ops: Array<'subscribe' | 'list' | 'listPending'> = [] + const subscribeCalls: unknown[][] = [] + const pendingSubscribes: Array>> = [] + const historyQueue = parkedQueue() + const pendingQueue = parkedQueue() + const subscriptionDispose = vi.fn<() => void>() + let subscriber: ActionsSubscriber | undefined + const overseer = { + subscribeToActions: (...args: unknown[]) => { + ops.push('subscribe') + subscribeCalls.push(args) + subscriber = args[0] as ActionsSubscriber + return new Promise>((resolve, reject) => + pendingSubscribes.push({ resolve, reject })) + }, + listActions: (options?: ListOptions) => { + if (options?.filter === 'pending') { + ops.push('listPending') + return pendingQueue.park(options) + } + ops.push('list') + return historyQueue.park(options) + }, + [Symbol.dispose]: () => {}, + } as unknown as RpcStub + return { + overseer, + ops, + subscribeCalls, + listCalls: historyQueue.calls, + pendingQueryCalls: pendingQueue.calls, + subscriptionDispose, + async resolveSubscription() { + await act(async () => { + pendingSubscribes.shift()!.resolve( + { [Symbol.dispose]: subscriptionDispose } as unknown as RpcStub<{}>) + }) + }, + async rejectSubscription(err: unknown) { + await act(async () => { pendingSubscribes.shift()!.reject(err) }) + }, + resolvePage: historyQueue.resolveNext, + rejectPage: historyQueue.rejectNext, + resolvePendingQuery: pendingQueue.resolveNext, + rejectPendingQuery: pendingQueue.rejectNext, + async emit(record: ActionLogEntry) { + await act(async () => { subscriber!.entry(record) }) + }, + } +} + +/** A DOM root with act()-wrapped render/unmount. cleanup() resets it for the next test. */ +export function makeTestRoot() { + let root: Root | undefined + let container: HTMLDivElement | undefined + return { + async render(node: React.ReactNode) { + if (!root) { + container = document.createElement('div') + document.body.append(container) + root = createRoot(container) + } + await act(async () => root!.render(node)) + }, + unmount() { + act(() => root?.unmount()) + root = undefined + }, + cleanup() { + this.unmount() + container?.remove() + container = undefined + rafQueue.length = 0 + }, + } +} diff --git a/packages/workshop-frontend/src/useActionHistory.test.tsx b/packages/workshop-frontend/src/useActionHistory.test.tsx new file mode 100644 index 000000000..af8d45305 --- /dev/null +++ b/packages/workshop-frontend/src/useActionHistory.test.tsx @@ -0,0 +1,233 @@ +// @vitest-environment jsdom +/* eslint-disable react/react-in-jsx-scope */ + +import { act } from 'react' +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { RpcStub } from 'capnweb' +import type { ActionLogEntry, Overseer } from '@gadgets/workshop-shared/api' +import { entry as pendingEntry, makeOverseer, makeTestRoot } from './action-test-harness' +import { useActionHistory } from './useActionHistory' +import type { HistoryViewFilter } from './useActionHistory' + +// Most history fixtures are resolved records; use the harness `pendingEntry` for pending ones. +function entry(id: number, over: Partial> = {}): ActionLogEntry { + return pendingEntry(id, { + appliedAt: new Date(1700000000000 + id * 60_000), + state: 'approved', + ...over, + }) +} + +type HookResult = ReturnType + +describe('useActionHistory', () => { + const view = makeTestRoot() + let latest: HookResult + + function Probe({ overseer, filter, active }: { + overseer: RpcStub | null + filter: HistoryViewFilter + active: boolean + }) { + latest = useActionHistory(overseer, filter, active) + return null + } + + async function render(overseer: RpcStub | null, filter: HistoryViewFilter, + active: boolean) { + await view.render() + } + + afterEach(() => view.cleanup()) + + it('fetches nothing until activated, then loads the first page', async () => { + const server = makeOverseer() + await render(server.overseer, 'all', false) + expect(server.listCalls).toEqual([]) + expect(latest.status).toBe('loading') + expect(latest.hasMore).toBe(false) + + await render(server.overseer, 'all', true) + expect(server.listCalls).toEqual([{ beforeId: undefined, filter: 'all' }]) + expect(latest.status).toBe('loading') + + await server.resolvePage({ entries: [entry(30), entry(20)], nextBeforeId: 10 }) + expect(latest.status).toBe('ready') + expect(latest.entries.map(e => e.id)).toEqual([30, 20]) + expect(latest.hasMore).toBe(true) + + await render(server.overseer, 'all', false) + await render(server.overseer, 'all', true) + expect(server.listCalls).toHaveLength(1) + }) + + it('continues from the cursor, dedupes by id, and ignores blocked loads', async () => { + const server = makeOverseer() + await render(server.overseer, 'all', true) + + act(() => latest.loadMore()) + expect(server.listCalls).toHaveLength(1) + + await server.resolvePage({ entries: [entry(30), entry(20)], nextBeforeId: 10 }) + act(() => latest.loadMore()) + act(() => latest.loadMore()) + expect(latest.isLoadingMore).toBe(true) + expect(server.listCalls).toHaveLength(2) + expect(server.listCalls[1]).toEqual({ beforeId: 10, filter: 'all' }) + + await server.resolvePage({ entries: [entry(20), entry(5)] }) + expect(latest.entries.map(e => e.id)).toEqual([30, 20, 5]) + expect(latest.hasMore).toBe(false) + expect(latest.isLoadingMore).toBe(false) + + act(() => latest.loadMore()) + expect(server.listCalls).toHaveLength(2) + }) + + it('keeps hasMore after an empty page that carries a cursor', async () => { + const server = makeOverseer() + await render(server.overseer, 'all', true) + await server.resolvePage({ entries: [], nextBeforeId: 400 }) + + expect(latest.status).toBe('ready') + expect(latest.entries).toEqual([]) + expect(latest.hasMore).toBe(true) + + act(() => latest.loadMore()) + expect(server.listCalls[1]).toEqual({ beforeId: 400, filter: 'all' }) + }) + + it('resets on filter change and ignores the stale in-flight page', async () => { + const server = makeOverseer() + await render(server.overseer, 'all', true) + + await render(server.overseer, 'action', true) + expect(latest.entries).toEqual([]) + expect(latest.status).toBe('loading') + expect(server.listCalls[1]).toEqual({ beforeId: undefined, filter: 'action' }) + + await server.resolvePage({ entries: [entry(30)] }) + expect(latest.entries).toEqual([]) + expect(latest.status).toBe('loading') + + await server.resolvePage({ entries: [entry(20)] }) + expect(latest.entries.map(e => e.id)).toEqual([20]) + expect(latest.status).toBe('ready') + }) + + it('resets and refetches when the stub changes', async () => { + const first = makeOverseer() + await render(first.overseer, 'all', true) + await first.resolvePage({ entries: [entry(30)], nextBeforeId: 10 }) + act(() => latest.loadMore()) // leave a second fetch in flight on the old stub + + const second = makeOverseer() + await render(second.overseer, 'all', true) + expect(latest.entries).toEqual([]) + expect(second.listCalls).toEqual([{ beforeId: undefined, filter: 'all' }]) + + // The abandoned stub's in-flight page must not leak into the new session. + await first.resolvePage({ entries: [entry(9)] }) + expect(latest.entries).toEqual([]) + expect(latest.status).toBe('loading') + + await second.resolvePage({ entries: [entry(40)] }) + expect(latest.entries.map(e => e.id)).toEqual([40]) + }) + + it('recovers from a failed first load on retry', async () => { + vi.spyOn(console, 'error').mockImplementation(() => {}) + const server = makeOverseer() + await render(server.overseer, 'all', true) + await server.rejectPage(new Error('nope')) + expect(latest.status).toBe('error') + expect(latest.loadMoreFailed).toBe(false) // first-load failure is owned by status + + act(() => latest.loadMore()) + expect(latest.status).toBe('loading') + expect(server.listCalls[1]).toEqual({ beforeId: undefined, filter: 'all' }) + await server.resolvePage({ entries: [entry(30)] }) + expect(latest.status).toBe('ready') + expect(latest.entries.map(e => e.id)).toEqual([30]) + vi.restoreAllMocks() + }) + + it('preserves a later page cursor and entries when retrying after failure', async () => { + vi.spyOn(console, 'error').mockImplementation(() => {}) + const server = makeOverseer() + await render(server.overseer, 'all', true) + await server.resolvePage({ entries: [entry(30)], nextBeforeId: 10 }) + + act(() => latest.loadMore()) + await server.rejectPage(new Error('nope')) + expect(latest.status).toBe('ready') + expect(latest.loadMoreFailed).toBe(true) + expect(latest.entries.map(e => e.id)).toEqual([30]) + expect(latest.hasMore).toBe(true) + expect(latest.isLoadingMore).toBe(false) + + act(() => latest.loadMore()) + expect(latest.loadMoreFailed).toBe(false) // cleared as soon as the retry starts + expect(server.listCalls[2]).toEqual({ beforeId: 10, filter: 'all' }) + await server.resolvePage({ entries: [entry(5)] }) + expect(latest.loadMoreFailed).toBe(false) + expect(latest.entries.map(e => e.id)).toEqual([30, 5]) + expect(latest.hasMore).toBe(false) + vi.restoreAllMocks() + }) + + describe('live updates', () => { + it('patches a loaded record in place', async () => { + const server = makeOverseer() + await render(server.overseer, 'all', true) + await server.resolvePage({ entries: [entry(30)], nextBeforeId: 10 }) + + await server.emit(entry(30, { state: 'rejected' })) + expect(latest.entries.map(e => [e.id, e.state])).toEqual([[30, 'rejected']]) + }) + + it('inserts new and in-window resolutions, ordered by id', async () => { + const server = makeOverseer() + await render(server.overseer, 'all', true) + await server.resolvePage({ entries: [entry(30), entry(20)], nextBeforeId: 10 }) + + await server.emit(entry(40)) // newly resolved, above the window + await server.emit(entry(25)) // old pending resolved inside the window + expect(latest.entries.map(e => e.id)).toEqual([40, 30, 25, 20]) + }) + + it('inserts a new pending record and patches its resolution in place', async () => { + const server = makeOverseer() + await render(server.overseer, 'all', true) + await server.resolvePage({ entries: [entry(30)], nextBeforeId: 10 }) + + await server.emit(pendingEntry(40)) + expect(latest.entries.map(e => [e.id, e.state])) + .toEqual([[40, 'pending'], [30, 'approved']]) + + await server.emit(entry(40, { state: 'rejected' })) + expect(latest.entries.map(e => [e.id, e.state])) + .toEqual([[40, 'rejected'], [30, 'approved']]) + }) + + it('drops records below the loaded window and filter mismatches', async () => { + const server = makeOverseer() + await render(server.overseer, 'action', true) + await server.resolvePage({ entries: [entry(30)], nextBeforeId: 10 }) + + await server.emit(entry(5)) // below the window + await server.emit(pendingEntry(35, { type: 'observation' })) // filter mismatch + await server.emit(entry(36, { type: 'observation' })) // filter mismatch + expect(latest.entries.map(e => e.id)).toEqual([30]) + }) + + it('drops updates until a page has loaded', async () => { + const server = makeOverseer() + await render(server.overseer, 'all', true) + await server.emit(entry(50)) // arrives while the first page is in flight + await server.resolvePage({ entries: [entry(30)] }) + + expect(latest.entries.map(e => e.id)).toEqual([30]) + }) + }) +}) diff --git a/packages/workshop-frontend/src/useActionHistory.ts b/packages/workshop-frontend/src/useActionHistory.ts new file mode 100644 index 000000000..617bb967d --- /dev/null +++ b/packages/workshop-frontend/src/useActionHistory.ts @@ -0,0 +1,129 @@ +import { useCallback, useEffect, useMemo, useRef, useState } from 'react' +import type { RpcStub } from 'capnweb' +import { matchesActionHistoryFilter } from '@gadgets/workshop-shared/api' +import type { ActionHistoryFilter, ActionLogEntry, Overseer } from '@gadgets/workshop-shared/api' +import { useActionEntries } from './useActions' + +export type ActionHistoryStatus = 'loading' | 'ready' | 'error' + +/** + * The filters this hook pages: everything but "pending", whose live set is useActions' job. The + * live-merge below leans on the exclusion — against the remaining filters a record never stops + * matching (type never changes, and resolving only moves a record out of "pending"), so updates + * failing the filter can be ignored rather than needing removal. + */ +export type HistoryViewFilter = Exclude + +type HistoryState = { + byId: ReadonlyMap + // 'initial' = the first page failed (surfaces as status 'error'); 'more' = a loadMore failed. + error: 'initial' | 'more' | null +} + +type HistorySession = { + frontier: number | undefined + inFlight: boolean + hasLoadedPage: boolean +} + +function createHistorySession(): HistorySession { + return { frontier: undefined, inFlight: false, hasLoadedPage: false } +} + +const INITIAL: HistoryState = { byId: new Map(), error: null } + +/** + * Demand-loads action history (pending records included), one page at a time, newest first by id + * (creation order). Nothing is fetched until `active` first becomes true; `loadMore()` continues + * from the server cursor, and `hasMore` is the termination signal. The overseer stub or filter + * changing resets everything (a reconnect hands out a fresh stub). + * + * Live updates from the shared action subscription are merged in: a filter-matching record — + * a fresh pending one or a resolution — patches in place or inserts if it falls inside the + * loaded id window. Records below the window are dropped — they surface, read fresh, when their + * page loads. + */ +export function useActionHistory( + overseer: RpcStub | null, + filter: HistoryViewFilter, + active: boolean, +) { + const [state, setState] = useState(INITIAL) + // The request-identity token, mutated in place. `frontier` is undefined before the first page + // and after the terminal page; `hasLoadedPage` distinguishes those states. The returned + // status/hasMore/isLoadingMore are derived from it at the return site, which is safe because + // every mutation of it is paired with a setState (so a render always follows). + const sessionRef = useRef(createHistorySession()) + + useEffect(() => { + sessionRef.current = createHistorySession() + setState(INITIAL) + }, [overseer, filter]) + + const loadMore = useCallback(() => { + const session = sessionRef.current + if (!overseer || session.inFlight || + (session.hasLoadedPage && session.frontier === undefined)) return + const first = !session.hasLoadedPage + session.inFlight = true + setState(prev => ({ ...prev, error: null })) + + overseer.listActions({ beforeId: session.frontier, filter }).then(page => { + if (sessionRef.current !== session) return + session.inFlight = false + session.hasLoadedPage = true + session.frontier = page.nextBeforeId + setState(prev => { + const byId = new Map(prev.byId) + for (const record of page.entries) byId.set(record.id, record) + return { byId, error: null } + }) + }, (err: unknown) => { + if (sessionRef.current !== session) return + session.inFlight = false + console.error('Failed to load action history:', err) + setState(prev => ({ ...prev, error: first ? 'initial' : 'more' })) + }) + }, [overseer, filter]) + + // Registered before the initial-load effect below: mounting the shared store initiates its + // subscribeToActions, and effects run in declaration order, so capnweb e-order registers the + // subscriber server-side before the first page's listActions reads (the subscribe-before-query + // contract, see api.ts). Otherwise a record resolved between the two would be missed by both. + useActionEntries(overseer, record => { + const session = sessionRef.current + // Dropping records here can't lose an update: listActions snapshots and responds in one DO + // turn, so on the ordered RPC session an entry reflecting a post-snapshot change always + // arrives after the page it would race with. Anything dropped pre-first-page or below the + // frontier is state a loaded page already supersedes, or is read fresh when its page loads. + if (!session.hasLoadedPage) return + if (!matchesActionHistoryFilter(record, filter)) return + if (session.frontier !== undefined && record.id < session.frontier) return + setState(prev => { + const byId = new Map(prev.byId) + byId.set(record.id, record) + return { ...prev, byId } + }) + }) + + useEffect(() => { + if (active && !sessionRef.current.hasLoadedPage) loadMore() + }, [active, loadMore]) + + const entries = useMemo( + () => Array.from(state.byId.values()).sort((a, b) => b.id - a.id), + [state.byId]) + + const session = sessionRef.current + return { + entries, + // 'loading' from the start: nothing is fetched until `active`, but the panel isn't visible + // (and the status unread) until then either, and the first fetch follows immediately. + status: (state.error === 'initial' ? 'error' + : session.hasLoadedPage ? 'ready' : 'loading') satisfies ActionHistoryStatus, + hasMore: session.frontier !== undefined, + isLoadingMore: session.inFlight && session.hasLoadedPage, + loadMoreFailed: state.error === 'more', + loadMore, + } +} diff --git a/packages/workshop-frontend/src/useActions.test.tsx b/packages/workshop-frontend/src/useActions.test.tsx new file mode 100644 index 000000000..96cbdf706 --- /dev/null +++ b/packages/workshop-frontend/src/useActions.test.tsx @@ -0,0 +1,312 @@ +// @vitest-environment jsdom +/* eslint-disable react/react-in-jsx-scope */ + +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { RpcStub } from 'capnweb' +import type { ActionLogEntry, Overseer } from '@gadgets/workshop-shared/api' +import { entry, flushFrames, makeOverseer, makeTestRoot } from './action-test-harness' +import { useActionEntries, useActions, type ActionsState } from './useActions' + +describe('useActions', () => { + const view = makeTestRoot() + let latest: ActionsState + + function Probe({ overseer }: { overseer: RpcStub | null }) { + latest = useActions(overseer) + return null + } + + afterEach(() => { + view.cleanup() + vi.restoreAllMocks() + }) + + it('initiates the subscription before the first pending page, with no startAfter', async () => { + const server = makeOverseer() + await view.render() + + // Subscribe-first is what lets the pages be a complete snapshot: capnweb e-order registers + // the subscriber server-side before the first page reads. + expect(server.ops).toEqual(['subscribe', 'listPending']) + // Single argument: the legacy full replay (startAfter) must not be requested. + expect(server.subscribeCalls).toEqual([[expect.anything()]]) + expect(server.pendingQueryCalls).toEqual([{ filter: 'pending', beforeId: undefined }]) + }) + + it('stays checking until the last pending page loads', async () => { + const server = makeOverseer() + await view.render() + + await server.resolveSubscription() + expect(latest.status).toBe('checking') // the subscription alone doesn't settle + + await server.resolvePendingQuery({ entries: [entry(1)], nextBeforeId: 1 }) + expect(latest.status).toBe('checking') + flushFrames() + expect(latest.pending.map(e => e.id)).toEqual([1]) // counts accumulate mid-paging + expect(server.pendingQueryCalls[1]).toEqual({ filter: 'pending', beforeId: 1 }) + + await server.resolvePendingQuery({ entries: [entry(0)] }) + expect(latest.status).toBe('ready') // settles synchronously, no frame needed + expect(latest.pending.map(e => e.id)).toEqual([0, 1]) + }) + + it('sorts pendings oldest-first by createdAt, breaking ties by id', async () => { + const server = makeOverseer() + await view.render() + await server.resolveSubscription() + + const at = new Date(1700000000000) + await server.emit(entry(4, { createdAt: at })) + await server.emit(entry(6, { createdAt: new Date(1600000000000) })) + await server.emit(entry(5, { createdAt: at })) + flushFrames() + expect(latest.pending.map(e => e.id)).toEqual([6, 4, 5]) + }) + + it('removes a pending record when it resolves live', async () => { + const server = makeOverseer() + await view.render() + await server.resolveSubscription() + + await server.emit(entry(9)) + flushFrames() + expect(latest.pending.map(e => e.id)).toEqual([9]) + + await server.emit(entry(9, { state: 'rejected' })) + flushFrames() + expect(latest.pending).toEqual([]) + }) + + it('never re-marks a record pending after a live resolution beat its stale page copy', + async () => { + const server = makeOverseer() + await view.render() + await server.resolveSubscription() + + // The record resolves on the live stream while its page is still in flight; the page's + // pending copy is a stale call-time snapshot and must lose. + await server.emit(entry(5)) + await server.emit(entry(5, { state: 'approved' })) + await server.resolvePendingQuery({ entries: [entry(5)] }) + + expect(latest.status).toBe('ready') + expect(latest.pending).toEqual([]) + }) + + it('fences a released store and disposes its late-arriving subscription', async () => { + const server = makeOverseer() + await view.render() + await server.emit(entry(1)) + + view.unmount() + + await server.resolveSubscription() + expect(server.subscriptionDispose).toHaveBeenCalledOnce() + // A late entry on the fenced generation must not throw or resurrect state. + await server.emit(entry(2)) + }) + + it('fences the page loop: a late page neither folds nor requests a successor', async () => { + const server = makeOverseer() + await view.render() + await server.resolveSubscription() + + view.unmount() + + await server.resolvePendingQuery({ entries: [entry(1)], nextBeforeId: 1 }) + expect(server.pendingQueryCalls).toHaveLength(1) + }) + + it('starts a fresh subscription when the stub changes', async () => { + const first = makeOverseer() + await view.render() + await first.emit(entry(1)) + await first.resolveSubscription() + await first.resolvePendingQuery({ entries: [] }) + expect(latest.status).toBe('ready') + + const second = makeOverseer() + await view.render() + expect(second.subscribeCalls).toHaveLength(1) + expect(second.pendingQueryCalls).toHaveLength(1) + expect(latest.status).toBe('checking') + expect(latest.pending).toEqual([]) + }) + + it('reports error but keeps gathered pendings when the subscribe call fails', async () => { + vi.spyOn(console, 'error').mockImplementation(() => {}) + const server = makeOverseer() + await view.render() + + await server.emit(entry(1)) + await server.rejectSubscription(new Error('DO overloaded')) + + expect(latest.status).toBe('error') + expect(latest.pending.map(e => e.id)).toEqual([1]) + flushFrames() + expect(latest.status).toBe('error') + }) + + it('stays error when the pages drain after the subscribe call already failed', async () => { + vi.spyOn(console, 'error').mockImplementation(() => {}) + const server = makeOverseer() + await view.render() + + await server.rejectSubscription(new Error('DO overloaded')) + await server.resolvePendingQuery({ entries: [entry(1)] }) + + // The successful pages fold, but a dead live stream must not present as settled. + expect(latest.status).toBe('error') + expect(latest.pending.map(e => e.id)).toEqual([1]) + }) + + it('reports error but keeps gathered pendings when a later page fails', async () => { + vi.spyOn(console, 'error').mockImplementation(() => {}) + const server = makeOverseer() + await view.render() + await server.resolveSubscription() + + await server.emit(entry(1)) + await server.resolvePendingQuery({ entries: [entry(3)], nextBeforeId: 3 }) + await server.rejectPendingQuery(new Error('DO overloaded')) + + expect(latest.status).toBe('error') + expect(latest.pending.map(e => e.id)).toEqual([1, 3]) + flushFrames() + expect(latest.status).toBe('error') + }) + + it('downgrades ready to error when the subscribe call fails after the pages drained', + async () => { + vi.spyOn(console, 'error').mockImplementation(() => {}) + const server = makeOverseer() + await view.render() + + // The pages can drain before the subscribe call's return trip fails; a store with a dead + // live stream must not present as settled. + await server.resolvePendingQuery({ entries: [entry(2)] }) + expect(latest.status).toBe('ready') + + await server.rejectSubscription(new Error('broken tube')) + expect(latest.status).toBe('error') + expect(latest.pending.map(e => e.id)).toEqual([2]) + }) + + it('fans out live entries only — paged pendings are not entries', async () => { + const server = makeOverseer() + const received: number[] = [] + const late: number[] = [] + + function EntriesProbe({ sink }: { sink: number[] }) { + useActionEntries(server.overseer, record => sink.push(record.id)) + return null + } + + await view.render( + <> + + + , + ) + await server.resolveSubscription() + await server.resolvePendingQuery({ entries: [entry(1)] }) + await server.emit(entry(2)) + expect(received).toEqual([2]) + + // A late consumer's mount-time replay is live-only too. + await view.render( + <> + + + + , + ) + expect(late).toEqual([2]) + }) + + it('isolates a throwing entry listener from other consumers', async () => { + const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {}) + const server = makeOverseer() + const received: number[] = [] + + function EntriesProbe({ onEntry }: { onEntry: (record: ActionLogEntry) => void }) { + useActionEntries(server.overseer, onEntry) + return null + } + + await view.render( + <> + + { throw new Error('listener broke') }} /> + received.push(record.id)} /> + , + ) + await server.resolveSubscription() + + await server.emit(entry(3)) + flushFrames() + expect(received).toEqual([3]) + expect(latest.pending.map(e => e.id)).toEqual([3]) + expect(consoleError).toHaveBeenCalledWith('Action entry listener failed:', expect.any(Error)) + }) + + it('updates a late listener without replaying and disposes after the last consumer', async () => { + const server = makeOverseer() + const firstReceived: number[] = [] + const secondReceived: number[] = [] + const firstCallback = (record: ActionLogEntry) => firstReceived.push(record.id) + const secondCallback = (record: ActionLogEntry) => secondReceived.push(record.id) + + function EntriesProbe({ onEntry }: { onEntry: (record: ActionLogEntry) => void }) { + useActionEntries(server.overseer, onEntry) + return null + } + + function Harness({ + onEntry, + showActions, + showEntries, + }: { + onEntry: (record: ActionLogEntry) => void + showActions: boolean + showEntries: boolean + }) { + return ( + <> + {showActions && } + {showEntries && } + + ) + } + + await view.render( + ) + await server.resolveSubscription() + await server.emit(entry(1)) + await server.emit(entry(2, { state: 'approved' })) + + await view.render( + ) + expect(firstReceived).toEqual([1, 2]) + expect(server.subscribeCalls).toHaveLength(1) + + await view.render( + ) + expect(firstReceived).toEqual([1, 2]) + expect(secondReceived).toEqual([]) + + await server.emit(entry(6)) + expect(firstReceived).toEqual([1, 2]) + expect(secondReceived).toEqual([6]) + expect(server.subscribeCalls).toHaveLength(1) + + await view.render( + ) + expect(server.subscriptionDispose).not.toHaveBeenCalled() + + await view.render( + ) + expect(server.subscriptionDispose).toHaveBeenCalledOnce() + }) +}) diff --git a/packages/workshop-frontend/src/useActions.ts b/packages/workshop-frontend/src/useActions.ts index 81d8acaad..5ed9b4a58 100644 --- a/packages/workshop-frontend/src/useActions.ts +++ b/packages/workshop-frontend/src/useActions.ts @@ -1,15 +1,33 @@ -import { useCallback, useEffect, useRef, useState, useSyncExternalStore } from 'react' +import { useCallback, useEffect, useRef, useSyncExternalStore } from 'react' import { RpcStub, RpcTarget } from 'capnweb' import { ActionLogEntry, ActionsSubscriber, Overseer } from '@gadgets/workshop-shared/api' -// One ref-counted subscription per Overseer stub, shared across consumers. -// `startAfter: epoch 0` asks the backend to replay full history through the -// subscriber, avoiding a stale-state race against a separate `listActions()`. +// One ref-counted store per Overseer stub, shared across consumers. On open the store initiates +// the live subscription first, then pages the currently-pending set via +// listActions({filter: 'pending'}): capnweb e-order registers the subscriber server-side before +// the first page reads, so every record is covered — pages snapshot call-time state, and +// everything that changes after arrives on the subscription. Pages fold with live-wins +// semantics; the last page loading is the "settled" signal. Resolved history is demand-paged +// separately (see useActionHistory). + +export type ActionsState = { + /** + * 'checking' until the subscription is registered and the last pending page has loaded; + * 'error' if either failed (`pending` keeps whatever was gathered). + */ + status: 'checking' | 'ready' | 'error' + + /** Pending-review records found so far, oldest first (createdAt, then id). */ + pending: readonly ActionLogEntry[] +} type Store = { - actionsById: Map - pendingActionsById: Map - isReady: boolean + // Immutable object handed to useSyncExternalStore; rebuilt from the staged fields on commit. + snapshot: ActionsState + stagedPending: Map + // Records delivered on the live subscription (only — paged records are not entries), retained + // for useActionEntries' mount-time replay. + stagedEntries: Map refCount: number listeners: Set<() => void> entryListeners: Set<(record: ActionLogEntry) => void> @@ -18,15 +36,20 @@ type Store = { notifyScheduled: boolean } +const EMPTY_STATE: ActionsState = { + status: 'checking', + pending: [], +} + const stores = new WeakMap, Store>() function getStore(overseer: RpcStub): Store { let store = stores.get(overseer) if (!store) { store = { - actionsById: new Map(), - pendingActionsById: new Map(), - isReady: false, + snapshot: EMPTY_STATE, + stagedPending: new Map(), + stagedEntries: new Map(), refCount: 0, listeners: new Set(), entryListeners: new Set(), @@ -39,71 +62,115 @@ function getStore(overseer: RpcStub): Store { return store } -function notify(store: Store) { +function commit(store: Store, status: ActionsState['status'] = store.snapshot.status): void { + // Entries arrive in ascending id order, so this sort is near-free at pending-count scale. + const pending = [...store.stagedPending.values()].toSorted((a, b) => + a.createdAt.getTime() - b.createdAt.getTime() || a.id - b.id) + store.snapshot = { status, pending } for (const listener of store.listeners) listener() } +// Coalesce bursts of entries into one snapshot per frame. Status transitions commit synchronously +// instead (see openSubscription) so a throttled background tab still settles. function scheduleNotify(store: Store) { if (store.notifyScheduled) return store.notifyScheduled = true window.requestAnimationFrame(() => { store.notifyScheduled = false - store.actionsById = new Map(store.pendingActionsById) - notify(store) + commit(store) }) } +// Bumps the generation (orphaning any in-flight callbacks) and clears the staged session state. +function resetSession(store: Store): number { + store.generation++ + store.stagedPending = new Map() + store.stagedEntries = new Map() + store.snapshot = EMPTY_STATE + return store.generation +} + function openSubscription(overseer: RpcStub, store: Store) { - const generation = ++store.generation - store.actionsById = new Map() - store.pendingActionsById = new Map() - store.isReady = false + const generation = resetSession(store) class ActionsSubscriberImpl extends RpcTarget implements ActionsSubscriber { entry(record: ActionLogEntry): void { if (store.generation !== generation) return - store.pendingActionsById.set(record.id, record) - scheduleNotify(store) - for (const listener of store.entryListeners) listener(record) + store.stagedEntries.set(record.id, record) + let pendingChanged: boolean + if (record.state === 'pending') { + store.stagedPending.set(record.id, record) + pendingChanged = true + } else { + pendingChanged = store.stagedPending.delete(record.id) + } + // Entries that never touch the pending set (observations, hook events) don't need a + // re-sorted snapshot or a consumer re-render. + if (pendingChanged) scheduleNotify(store) + for (const listener of store.entryListeners) { + try { + listener(record) + } catch (err) { + console.error('Action entry listener failed:', err) + } + } } - ready(): void { - if (store.generation !== generation) return - store.actionsById = new Map(store.pendingActionsById) - store.isReady = true - notify(store) - } + // Settledness is signalled by the pending page loop draining, not by the subscription. + ready(): void {} } + let failed = false + const fail = (error: unknown) => { + if (store.generation !== generation) return + console.error('Failed to load pending actions:', error) + // Deliberately also downgrades an already-'ready' store: the pages can drain before the + // subscribe call's return trip fails, and a store with a dead live stream must not present + // as settled. + failed = true + commit(store, 'error') + } + + // Initiated first — the page loop below relies on capnweb e-order having registered the + // subscriber server-side before the first page reads. + overseer.subscribeToActions( + new ActionsSubscriberImpl() as unknown as RpcStub, + ).then(sub => { + if (store.generation !== generation) { + sub[Symbol.dispose]() + return + } + store.subscription = sub + }, fail) + + // One page in flight at a time. Records created mid-paging are live-only (their ids are above + // page 1's snapshot bound), so the fold only has to resolve one conflict class: a page's stale + // copy of a record the subscription already delivered — the subscription's copy (in any state) + // is newer by definition, so a record resolved live is never re-marked pending by a page that + // predates the resolution. ;(async () => { - try { - const sub = await overseer.subscribeToActions( - new ActionsSubscriberImpl() as unknown as RpcStub, - new Date(0), - ) - if (store.generation !== generation) { - sub[Symbol.dispose]() - return - } - store.subscription = sub - } catch (err) { - if (store.generation === generation) { - store.isReady = true - notify(store) - console.error('Failed to subscribe to actions:', err) + let beforeId: number | undefined + do { + const page = await overseer.listActions({ filter: 'pending', beforeId }) + if (store.generation !== generation) return + for (const record of page.entries) { + if (!store.stagedEntries.has(record.id)) store.stagedPending.set(record.id, record) } - } - })() + beforeId = page.nextBeforeId + scheduleNotify(store) + } while (beforeId !== undefined) + // Synchronous commit (not scheduleNotify) so a throttled background tab still settles. A + // subscribe failure is sticky: pages draining afterwards must not upgrade a store whose live + // stream is dead back to 'ready'. + commit(store, failed ? 'error' : 'ready') + })().catch(fail) } function closeSubscription(store: Store) { - store.generation++ + resetSession(store) store.subscription?.[Symbol.dispose]() store.subscription = null - store.actionsById = new Map() - store.pendingActionsById = new Map() - store.isReady = false } function acquire(overseer: RpcStub): Store { @@ -125,15 +192,8 @@ function release(overseer: RpcStub) { } } -export type UseActionsResult = { - actionsById: Map - isReady: boolean -} - -const EMPTY_MAP: Map = new Map() - -/** Subscribe to the gadget's action log. Pass `null` to no-op. */ -export function useActions(overseer: RpcStub | null): UseActionsResult { +/** Subscribe to the gadget's action log. Pass `null` to no-op ('checking', empty maps). */ +export function useActions(overseer: RpcStub | null): ActionsState { useEffect(() => { if (!overseer) return acquire(overseer) @@ -147,51 +207,46 @@ export function useActions(overseer: RpcStub | null): UseActionsResult return () => { store.listeners.delete(cb) } }, [overseer]) const getSnapshot = useCallback(() => { - if (!overseer) return EMPTY_MAP - return stores.get(overseer)?.actionsById ?? EMPTY_MAP + if (!overseer) return EMPTY_STATE + return stores.get(overseer)?.snapshot ?? EMPTY_STATE }, [overseer]) - const getIsReady = useCallback(() => { - if (!overseer) return false - return stores.get(overseer)?.isReady ?? false - }, [overseer]) - - const actionsById = useSyncExternalStore(subscribe, getSnapshot, getSnapshot) - const isReady = useSyncExternalStore(subscribe, getIsReady, getIsReady) - return { actionsById, isReady } + return useSyncExternalStore(subscribe, getSnapshot, getSnapshot) } /** - * Per-entry callback variant. Fires once for every existing action on mount - * (replay) and once per live update — no separate seeding needed. + * Per-entry callback variant. Fires once per entry delivered on the live subscription (paged + * pending records are not entries); on mount it replays the records already received this + * session. That covers patching content fetched over the same stub: anything fetched reflects + * the log at fetch time, and everything that changed since arrived through the stream. */ export function useActionEntries( overseer: RpcStub | null, onEntry: (record: ActionLogEntry) => void, ): void { - // Stable listener identity; latest callback read via ref so updating - // `onEntry` doesn't resubscribe. const callbackRef = useRef(onEntry) callbackRef.current = onEntry - const [stableListener] = useState(() => (record: ActionLogEntry) => { - callbackRef.current(record) - }) useEffect(() => { if (!overseer) return + + function listener(record: ActionLogEntry): void { + callbackRef.current(record) + } + const store = acquire(overseer) - store.entryListeners.add(stableListener) + store.entryListeners.add(listener) // Retained until `release()` drops refCount to 0 and deletes the store, so late - // consumers can replay entries while a shared subscription is still alive. - for (const record of store.pendingActionsById.values()) { - stableListener(record) + // consumers can replay already-received entries while a shared subscription is still alive. + for (const record of store.stagedEntries.values()) { + listener(record) } return () => { const s = stores.get(overseer) - s?.entryListeners.delete(stableListener) + s?.entryListeners.delete(listener) release(overseer) } - }, [overseer, stableListener]) + }, [overseer]) } diff --git a/packages/workshop-shared/src/api.ts b/packages/workshop-shared/src/api.ts index c7d34b9e0..0d86e0bc7 100644 --- a/packages/workshop-shared/src/api.ts +++ b/packages/workshop-shared/src/api.ts @@ -1763,10 +1763,17 @@ export interface Overseer extends RpcTarget { newAgentSpawnerGatekeeper(config: AgentSpawnerConfig): Promise>; /** - * List history of actions. - * TODO: This should be paginated. + * Fetch one page of action history, newest first by id (creation order). "all" (the default) + * pages every record and a record type pages that type — pending records included, each at its + * creation position; `filter: "pending"` pages only the currently-pending records — the query + * half of the query-for-state/subscribe-for-deltas contract (see subscribeToActions()). + * + * Page size is a server constant. Pages are full until the last: absence of `nextBeforeId` + * means the history is exhausted; otherwise it is the id of the last returned entry, to pass + * as `beforeId` for the next-older page. */ - listActions(): Promise; + listActions(options?: {beforeId?: number, filter?: ActionHistoryFilter}) + : Promise; /** * Approve an action that is currently in the "pending" state. The action will be performed on @@ -1841,7 +1848,23 @@ export interface Overseer extends RpcTarget { /** * Subscribe to action adds/updates. Dispose the returned stub to unsubscribe. - * If `startAfter` is set, replay actions changed after that timestamp. + * + * The subscription delivers live deltas only — nothing pre-existing is replayed. Query for + * state, subscribe for deltas: fetch the current pending set via + * listActions({filter: "pending"}) and resolved history via the other filters. As with + * subscribeToChat(), initiate the subscribe call before those reads — there is no need to + * await its return, only to start it first — so nothing can slip between the snapshot the + * pages reflect and the stream. + * + * The `startAfter` parameter is intended to be used when resubscribing after a disconnect: + * specify the time of the last action seen, in order to ensure no actions were missed during + * the disconnect. The bound is inclusive -- records last changed at exactly that time are + * re-delivered (entries are upserts) -- and the replay arrives in change-time order, not + * creation order. If not specified, the subscription starts from the current time. + * + * Do NOT use `startAfter` as a way to enumerate historical data. Use `listActions()` instead. + * To ensure no holes between a subscription and historical data, call `subscribeToActions()` + * immediately before `listActions()`, similar to `subscribeToChat()`. */ subscribeToActions(subscriber: RpcStub, startAfter?: Date): Promise>; @@ -2460,6 +2483,38 @@ export type AiChatHistoryPage = { }; }; +/** + * Filter for listActions(): "all" for every record, one specific record type, or "pending" for + * only the currently-pending records (of any type). Pending records appear in the type views and + * "all" too, so history shows everything the agent has attempted. + */ +export type ActionHistoryFilter = "all" | "pending" | ActionLogEntry["type"]; + +/** + * Whether a record passes an ActionHistoryFilter. Used by the client's live-merge; the server's + * listActions() answers the same question from its byHistoryFilter index, whose key derivation + * must stay in lockstep with this function so the two ends of the wire can't drift. + */ +export function matchesActionHistoryFilter( + record: {type: ActionLogEntry["type"], state: ActionState}, + filter: ActionHistoryFilter): boolean { + return filter === "pending" + ? record.state === "pending" + : filter === "all" || record.type === filter; +} + +/** One page of action history from listActions(). */ +export type ActionHistoryPage = { + /** Matching records, descending id (creation order, newest first). */ + entries: ActionLogEntry[]; + + /** + * Id of the last returned entry; pass as `beforeId` for the next-older page. Absent when the + * page reached the start of the history. + */ + nextBeforeId?: number; +}; + export type AiChatAuthorInfo = { /** * Is the author a human, AI, or Gadget? @@ -3301,6 +3356,13 @@ export type AiChatStreamEvent = { /** Interface implemented by the client to receive action-log upserts. */ export interface ActionsSubscriber { entry(record: ActionLogEntry): void; + + /** + * @deprecated Fires after the subscription has caught up to the current time. However, this is + * only a useful signal when a subscription is being used to enumerate past actions using a + * distant-past `startAfter`. This is not the correct way to use `subscribeToActions()`; use + * `listActions()` instead. + */ ready(): void; }