| 1 | import assert from "node:assert/strict"; |
| 2 | import { acceptSessionRuntimeSnapshot, createRuntimeStateStore, selectRuntime, selectRuntimeSession, type RuntimeProjection, type RuntimeState } from "../lib/runtimeStateStore"; |
| 3 | import { startRuntimeStateSync } from "../lib/runtimeStateSync"; |
| 4 | import { acceptRuntimeState } from "../lib/runtimeStateReducer"; |
| 5 | |
| 6 | const state: RuntimeState = { schemaVersion: 1, runtimeEpoch: "controller-a", activityRevision: 1, revision: 1, phase: "executing", running: true, |
| 7 | turnId: "turn-a", turnStatus: "in_progress", turnEventSeq: 1, pendingPrompt: false, cancelRequested: false, cancellable: true, backgroundJobs: 0, activity: "thinking" }; |
| 8 | const projection = (revision: number, changes: Partial<RuntimeState> = {}): RuntimeProjection => ({ epoch: "app-a", revision, topics: [], |
| 9 | sessions: [{ tabId: "a", scope: "project", workspaceRoot: "/fixture", topicId: "topic", sessionPath: "/fixture/session", sessionGeneration: 1, |
| 10 | open: true, remote: false, freshness: "synced", state: { ...state, revision, ...changes } }] }); |
| 11 | |
| 12 | { |
| 13 | const versioned = (projectionEpoch: string, revision: number, todos: RuntimeState["todos"]): RuntimeState => ({ |
| 14 | ...state, projectionEpoch, revision, todos, |
| 15 | }); |
| 16 | const newest = versioned("producer-a", 20, [{ content: "new", status: "in_progress" }]); |
| 17 | assert.equal(acceptSessionRuntimeSnapshot(newest, versioned("producer-a", 10, [])), newest, |
| 18 | "an old empty snapshot cannot clear a newer list"); |
| 19 | const cleared = acceptSessionRuntimeSnapshot(newest, versioned("producer-a", 21, [])); |
| 20 | assert.deepEqual(cleared.todos, [], "a newer empty snapshot is an authoritative clear"); |
| 21 | const replacement = versioned("producer-b", 1, [{ content: "replacement", status: "pending" }]); |
| 22 | assert.equal(acceptSessionRuntimeSnapshot(cleared, replacement), cleared, |
| 23 | "an ordinary frame cannot replace the current producer"); |
| 24 | assert.equal(acceptSessionRuntimeSnapshot(cleared, replacement, true), replacement, |
| 25 | "a binding-validated baseline can establish a replacement producer"); |
| 26 | assert.equal(acceptSessionRuntimeSnapshot(replacement, versioned("producer-a", 99, [])), replacement, |
| 27 | "the retired producer cannot overwrite its replacement"); |
| 28 | } |
| 29 | |
| 30 | { |
| 31 | const frame = projection(1, { todos: [{ content: "A's task", status: "in_progress" }] }); |
| 32 | const session = frame.sessions[0]; |
| 33 | assert.equal(selectRuntimeSession(frame, "a", session.sessionPath), session); |
| 34 | assert.equal(selectRuntimeSession(frame, "b", session.sessionPath), undefined, "a different tab cannot borrow the runtime"); |
| 35 | assert.equal(selectRuntimeSession(frame, "a", "/fixture/other"), undefined, "reusing a tab cannot borrow the previous session's runtime"); |
| 36 | for (const identity of [undefined, "", {}, { sessionPath: "" }]) { |
| 37 | assert.equal(selectRuntimeSession(frame, "a", identity), undefined, "unbound and empty sessions never act as wildcards"); |
| 38 | } |
| 39 | assert.equal(selectRuntimeSession(frame, "a", { sessionPath: session.sessionPath, sessionGeneration: 2 }), undefined, "ABA navigation rejects the old binding generation"); |
| 40 | assert.equal(selectRuntimeSession(frame, "a", { sessionPath: session.sessionPath, sessionGeneration: 1 }), session); |
| 41 | session.open = false; |
| 42 | assert.equal(selectRuntimeSession(frame, "a", session.sessionPath), undefined, "detached runtimes are not visible sessions"); |
| 43 | session.open = true; |
| 44 | session.sessionId = "canonical-a"; |
| 45 | const canonical = { session: { hostId: "local", sessionId: "canonical-a" }, sessionPath: "session-id:canonical-a", sessionGeneration: 1 }; |
| 46 | assert.equal(selectRuntimeSession(frame, "a", canonical), session, "canonical references match even when route and physical path differ"); |
| 47 | assert.equal(selectRuntimeSession(frame, "a", { ...canonical, session: { hostId: "local", sessionId: "other" }, sessionPath: session.sessionPath }), undefined, "matching legacy paths cannot override distinct canonical identities"); |
| 48 | session.remote = true; |
| 49 | session.hostId = "remote-host"; |
| 50 | assert.equal(selectRuntimeSession(frame, "a", canonical), undefined, "the same session ID on another host is not the same session"); |
| 51 | assert.equal(selectRuntimeSession(frame, "a", { ...canonical, session: { ...canonical.session, hostId: "remote-host" } }), session); |
| 52 | assert.equal(selectRuntimeSession(frame, "a", session.sessionPath), session, "remote path callers keep their scoped selection"); |
| 53 | } |
| 54 | const rawStore = createRuntimeStateStore(); |
| 55 | const store = { ...rawStore, accept: (next: RuntimeProjection, authoritative = false) => acceptRuntimeState(rawStore, next, authoritative) }; |
| 56 | let updates = 0; |
| 57 | store.subscribe(() => updates++); |
| 58 | assert.equal(store.accept(projection(1)), "accepted"); |
| 59 | const first = store.getSnapshot(); |
| 60 | assert.equal(store.accept(projection(1)), "duplicate"); |
| 61 | assert.equal(store.getSnapshot(), first); |
| 62 | assert.equal(updates, 1); |
| 63 | assert.equal(store.accept(projection(1, { running: false })), "conflict"); |
| 64 | assert.equal(store.getSnapshot(), first); |
| 65 | assert.equal(store.accept(projection(0)), "stale"); |
| 66 | assert.equal(store.accept(projection(2, { phase: "finishing", activity: "", cancellable: false })), "accepted"); |
| 67 | let view = selectRuntime(store.getSnapshot()!.sessions[0]); |
| 68 | assert.equal(view.kind, "finishing"); assert.equal(view.spinning, false); assert.equal(view.cancellable, false); assert.equal(view.running, true); |
| 69 | store.accept(projection(3, { phase: "idle", running: false, activity: "", cancellable: false, backgroundJobs: 2 })); |
| 70 | view = selectRuntime(store.getSnapshot()!.sessions[0]); |
| 71 | assert.equal(view.kind, "background_job"); assert.equal(view.running, false); |
| 72 | store.fail(); |
| 73 | assert.equal(selectRuntime(store.getSnapshot()!.sessions[0], store.getFailed()).kind, "unknown"); |
| 74 | assert.equal(store.getSnapshot()!.sessions[0].state.backgroundJobs, 2); |
| 75 | assert.equal(store.accept({ ...projection(1), epoch: "other" }), "conflict"); |
| 76 | assert.equal(store.accept({ ...projection(1), epoch: "other" }, true), "accepted"); |
| 77 | const mutable = projection(10); |
| 78 | assert.equal(store.accept(mutable, true), "accepted"); |
| 79 | mutable.sessions[0].state.running = false; |
| 80 | assert.equal(store.getSnapshot()!.sessions[0].state.running, true, "caller mutation cannot change a committed revision"); |
| 81 | const malformed = projection(11); |
| 82 | delete (malformed.sessions[0].state as Partial<RuntimeState>).cancellable; |
| 83 | assert.equal(store.accept(malformed), "conflict", "partial new-schema booleans cannot imply idle or uncancellable"); |
| 84 | |
| 85 | function deferred<T>() { let resolve!: (value: T) => void; let reject!: (err: unknown) => void; |
| 86 | const promise = new Promise<T>((yes, no) => { resolve = yes; reject = no; }); return { promise, resolve, reject }; } |
| 87 | const synced = createRuntimeStateStore(); |
| 88 | let receive!: (snapshot: RuntimeProjection) => void, focus!: () => void; |
| 89 | let pending = deferred<RuntimeProjection>(); |
| 90 | let reads = 0, subscribed = false, unsubscribed = false; |
| 91 | const timers = new Map<number, { callback: () => void; delay: number }>(); |
| 92 | let timerID = 0; |
| 93 | const stop = startRuntimeStateSync({ |
| 94 | subscribe: callback => { subscribed = true; receive = callback; return () => { unsubscribed = true; }; }, |
| 95 | read: () => { assert.equal(subscribed, true, "subscribe before initial GET"); reads++; return pending.promise; }, |
| 96 | timer: (callback, delay) => { const id = ++timerID; timers.set(id, { callback, delay }); return id; }, |
| 97 | clearTimer: id => { timers.delete(id as number); }, |
| 98 | focus: callback => { focus = callback; return () => {}; }, |
| 99 | }, synced); |
| 100 | focus(); focus(); assert.equal(reads, 1, "focus shares in-flight GET"); |
| 101 | receive(projection(4)); |
| 102 | pending.resolve(projection(2)); |
| 103 | await pending.promise; await Promise.resolve(); |
| 104 | assert.equal(synced.getSnapshot()!.revision, 4, "old GET cannot overwrite SSE"); |
| 105 | assert.equal([...timers.values()][0].delay, 30000); |
| 106 | for (const delay of [5000, 10000, 20000, 30000, 30000]) { |
| 107 | pending = deferred<RuntimeProjection>(); |
| 108 | [...timers.values()][0].callback(); |
| 109 | pending.reject(new Error("offline")); |
| 110 | await pending.promise.catch(() => {}); await Promise.resolve(); |
| 111 | assert.equal([...timers.values()][0].delay, delay); |
| 112 | assert.equal(synced.getSnapshot()!.revision, 4, "failure preserves known state"); |
| 113 | } |
| 114 | pending = deferred<RuntimeProjection>(); focus(); pending.resolve(projection(5)); |
| 115 | await pending.promise; await Promise.resolve(); |
| 116 | assert.equal([...timers.values()][0].delay, 30000); |
| 117 | assert.equal(synced.getFailed(), false); |
| 118 | const final = synced.getSnapshot(); |
| 119 | stop(); receive(projection(6)); focus(); |
| 120 | assert.equal(synced.getSnapshot(), final); assert.equal(unsubscribed, true); assert.equal(timers.size, 0); |
| 121 | console.log("runtime state: immutable revisions, selectors, GET/SSE ordering, recovery, singleflight and disposal passed"); |
| 122 | |
| 123 | { |
| 124 | const restored = createRuntimeStateStore(); |
| 125 | let recover!: () => void; |
| 126 | const replies: Array<(value: RuntimeProjection) => void> = []; |
| 127 | const dispose = startRuntimeStateSync({ |
| 128 | subscribe: () => () => {}, |
| 129 | recover: callback => { recover = callback; return () => {}; }, |
| 130 | read: () => new Promise(resolve => replies.push(resolve)), |
| 131 | timer: () => 1, |
| 132 | clearTimer: () => {}, |
| 133 | focus: () => () => {}, |
| 134 | }, restored); |
| 135 | recover(); recover(); |
| 136 | assert.equal(restored.getFailed(), true, "a gap marks runtime state unknown until an authoritative read"); |
| 137 | replies.shift()!(projection(99)); await Promise.resolve(); await Promise.resolve(); |
| 138 | assert.equal(restored.getSnapshot(), undefined, "a pre-gap async reply cannot repair the new generation"); |
| 139 | assert.equal(replies.length, 1, "recovery queued during an in-flight read is not lost"); |
| 140 | replies.shift()!({ ...projection(1), epoch: "new-app" }); await Promise.resolve(); await Promise.resolve(); |
| 141 | assert.equal(restored.getSnapshot()?.epoch, "new-app"); |
| 142 | assert.equal(restored.getFailed(), false); |
| 143 | dispose(); |
| 144 | } |
| 145 |