| 1 | import type { |
| 2 | HistoryWindowPage, |
| 3 | HistoryWindowRequest, |
| 4 | MessageFieldPage, |
| 5 | MessageHistoryPage, |
| 6 | MessageLocation, |
| 7 | PersistentMessage, |
| 8 | Ref as SessionContentRef, |
| 9 | SearchHistoryPage, |
| 10 | SessionHistoryContentChunk, |
| 11 | SessionOpenView, |
| 12 | FollowRequest, TranscriptFollowResponse, Change, |
| 13 | } from "../generated/desktopContract.generated"; |
| 14 | import type { HistoryContentChunk, HistoryContentRef, HistoryMessage, HistorySlice, HistorySliceRequest, WireEvent } from "./types"; |
| 15 | |
| 16 | export interface SessionReaderBindings { |
| 17 | TranscriptFollowForTab(tabID: string, request: FollowRequest): Promise<TranscriptFollowResponse>; |
| 18 | RemoteTranscriptFollowForTab(tabID: string, request: FollowRequest): Promise<TranscriptFollowResponse>; |
| 19 | SessionHistoryPageForTab(tabID: string, cursor: string, limit: number): Promise<MessageHistoryPage>; |
| 20 | SessionOpenForTab(tabID: string): Promise<SessionOpenView>; |
| 21 | SessionHistoryContentForTab(tabID: string, ref: SessionContentRef, offset: number): Promise<SessionHistoryContentChunk>; |
| 22 | RemoteSessionOpenForTab(tabID: string): Promise<SessionOpenView>; |
| 23 | RemoteSessionHistoryPageForTab(tabID: string, cursor: string, limit: number): Promise<MessageHistoryPage>; |
| 24 | RemoteSessionHistoryContentForTab(tabID: string, ref: SessionContentRef, offset: number): Promise<SessionHistoryContentChunk>; |
| 25 | SearchSessionHistoryForTab(tabID: string, textQuery: string, cursor: string, limit: number): Promise<SearchHistoryPage>; |
| 26 | RemoteSearchSessionHistoryForTab(tabID: string, textQuery: string, cursor: string, limit: number): Promise<SearchHistoryPage>; |
| 27 | LocateSessionMessageForTab(tabID: string, messageID: string, snapshot: number): Promise<MessageLocation>; |
| 28 | RemoteLocateSessionMessageForTab(tabID: string, messageID: string, snapshot: number): Promise<MessageLocation>; |
| 29 | // history-window-v1. A binding that does not implement these (or a remote |
| 30 | // service that does not advertise the capability) is read through the |
| 31 | // protocol-7 adapter in canonicalTranscriptBackend instead. |
| 32 | SessionHistoryWindowForTab(tabID: string, req: HistoryWindowRequest): Promise<HistoryWindowPage>; |
| 33 | SessionMessageFieldForTab(tabID: string, messageID: string, version: number, field: string, offset: number, length: number): Promise<MessageFieldPage>; |
| 34 | RemoteSessionHistoryWindowForTab(tabID: string, req: HistoryWindowRequest): Promise<HistoryWindowPage>; |
| 35 | RemoteSessionMessageFieldForTab(tabID: string, messageID: string, version: number, field: string, offset: number, length: number): Promise<MessageFieldPage>; |
| 36 | } |
| 37 | |
| 38 | interface MockSessionReaderHost { |
| 39 | MetaForTab?(tabID: string): Promise<{ running?: boolean; runtime?: { epoch?: string } }>; |
| 40 | HistoryForTab?(tabID: string): Promise<HistoryMessage[]>; |
| 41 | HistorySliceForTab(tabID: string, req: HistorySliceRequest): Promise<HistorySlice>; |
| 42 | HistoryContentForTab(tabID: string, ref: HistoryContentRef, chunkIndex: number): Promise<HistoryContentChunk>; |
| 43 | SessionHistoryPageForTab(tabID: string, cursor: string, limit: number): Promise<MessageHistoryPage>; |
| 44 | SessionOpenForTab(tabID: string): Promise<SessionOpenView>; |
| 45 | SessionHistoryContentForTab(tabID: string, ref: SessionContentRef, offset: number): Promise<SessionHistoryContentChunk>; |
| 46 | SessionHistoryWindowForTab(tabID: string, req: HistoryWindowRequest): Promise<HistoryWindowPage>; |
| 47 | } |
| 48 | |
| 49 | async function mockReaderSlice(host: MockSessionReaderHost, tab: string, request: HistorySliceRequest): Promise<HistorySlice> { |
| 50 | if (typeof host.HistorySliceForTab === "function") return host.HistorySliceForTab(tab, request); |
| 51 | const messages = await host.HistoryForTab?.(tab) ?? []; |
| 52 | let turn = 0; |
| 53 | const entries = messages.map((message, order) => { |
| 54 | if (message.role === "user") turn++; |
| 55 | return { entryId: message.messageId ?? `mock:${tab}:${order}`, order, turn, message, refs: [] }; |
| 56 | }).slice(-(request.entries ?? 32)); |
| 57 | return { entries, totalTurns: turn, startTurn: entries[0]?.turn ?? 0, endTurn: turn, nextCursor: "", |
| 58 | hasOlder: false, hasNewer: false, revision: 1, revisionKnown: true, digest: `mock:${tab}`, stale: false }; |
| 59 | } |
| 60 | |
| 61 | // Test/browser host implementation of the mandatory protocol. Production |
| 62 | // transports never route an unsupported Follow request through this adapter. |
| 63 | const mockFollowers = new Map<string, { tab: string; epoch?: string; revision: number; coverage: number; queue: Change[]; wake?: () => void }>(); |
| 64 | const mockPrompts = new Map<string, Map<string, WireEvent>>(); |
| 65 | const mockActive = new Map<string, { id: string; text: string; reasoning: string }>(); |
| 66 | const mockMetadata = new Map<string, { turnId?: string; running?: boolean; runtime?: { epoch?: string } }>(); |
| 67 | export function setMockTranscriptMetadata(tab: string, meta: { turnId?: string; running?: boolean; runtime?: { epoch?: string } }): void { |
| 68 | const prior = mockMetadata.get(tab)?.runtime?.epoch; |
| 69 | if (prior && meta.runtime?.epoch && prior !== meta.runtime.epoch) { mockActive.delete(tab); mockPrompts.delete(tab); } |
| 70 | mockMetadata.set(tab, { ...mockMetadata.get(tab), ...meta }); |
| 71 | } |
| 72 | let mockSubscription = 0; |
| 73 | let mockMessage = 0; |
| 74 | export function publishMockTranscriptEvent(event: WireEvent): void { |
| 75 | if (event.tabId) { |
| 76 | const tab = event.tabId; |
| 77 | const epoch = mockMetadata.get(tab)?.runtime?.epoch; |
| 78 | if (event.runtimeEpoch && epoch && event.runtimeEpoch !== epoch) return; |
| 79 | if (event.kind === "turn_started") setMockTranscriptMetadata(tab, { running: true }); |
| 80 | if (event.kind === "stream_attempt" && event.streamAttempt?.action === "begin" && event.messageId) { |
| 81 | mockActive.set(tab, { id: event.messageId, text: "", reasoning: "" }); |
| 82 | } |
| 83 | if (event.kind === "text" || event.kind === "reasoning") { |
| 84 | const active = mockActive.get(tab) ?? { id: event.messageId ?? `mock-active:${tab}:${++mockMessage}`, text: "", reasoning: "" }; |
| 85 | if (event.kind === "text") active.text += event.text ?? ""; else active.reasoning += event.text ?? ""; |
| 86 | mockActive.set(tab, active); |
| 87 | event = { ...event, messageId: active.id, attemptId: active.id }; |
| 88 | setMockTranscriptMetadata(tab, { running: true }); |
| 89 | } |
| 90 | if (event.kind === "turn_done") { mockActive.delete(tab); setMockTranscriptMetadata(tab, { running: false }); } |
| 91 | const prompts = mockPrompts.get(tab) ?? new Map<string, WireEvent>(); |
| 92 | const id = event.approval?.id ?? event.ask?.id; |
| 93 | if ((event.kind === "approval_request" || event.kind === "ask_request") && id) prompts.set(id, event); |
| 94 | if (event.kind === "prompt_answered" && event.itemId) prompts.delete(event.itemId); |
| 95 | if (event.kind === "turn_done") prompts.clear(); |
| 96 | mockPrompts.set(tab, prompts); |
| 97 | } |
| 98 | for (const follower of mockFollowers.values()) { |
| 99 | if (event.tabId && event.tabId !== follower.tab) continue; |
| 100 | if (event.runtimeEpoch && follower.epoch && event.runtimeEpoch !== follower.epoch) continue; |
| 101 | follower.queue.push({ revision: ++follower.revision, commitSeq: follower.coverage, durableSeq: follower.coverage, |
| 102 | index: 0, event: { ...event, seq: undefined } as unknown as Change["event"] }); |
| 103 | follower.wake?.(); |
| 104 | } |
| 105 | } |
| 106 | |
| 107 | function messageIndex(entryId: string): number { |
| 108 | return Number(/:m(\d+):o\d+$/.exec(entryId)?.[1] ?? -1); |
| 109 | } |
| 110 | |
| 111 | function canonicalBody(message: HistoryMessage | undefined, messageId: string): Uint8Array { |
| 112 | if (!message) return new Uint8Array(); |
| 113 | const body = { |
| 114 | ...message, |
| 115 | id: message.messageId ?? messageId, |
| 116 | reasoning_content: message.reasoning, |
| 117 | tool_calls: message.toolCalls?.map(call => ({ |
| 118 | ...call, |
| 119 | resolved_name: call.resolvedName, |
| 120 | capability_id: call.capabilityId, |
| 121 | resolved_read_only: call.resolvedReadOnly, |
| 122 | })), |
| 123 | tool_call_id: message.toolCallId, |
| 124 | name: message.toolName, |
| 125 | tool_execution: message.execution, |
| 126 | presented_files: message.presentedFiles ? { files: message.presentedFiles } : undefined, |
| 127 | server_search: message.serverSearch, |
| 128 | }; |
| 129 | return new TextEncoder().encode(JSON.stringify(body)); |
| 130 | } |
| 131 | |
| 132 | function persistentMessages(slice: HistorySlice, history: HistoryMessage[]): PersistentMessage[] { |
| 133 | return slice.entries.map(entry => { |
| 134 | const index = messageIndex(entry.entryId); |
| 135 | const messageId = entry.message.messageId ?? entry.entryId; |
| 136 | const body = canonicalBody(history[index], messageId); |
| 137 | return { |
| 138 | messageId, position: entry.order, |
| 139 | version: 1, role: entry.message.role, preview: entry.message.content ?? "", |
| 140 | eventSequence: 0, visibleTurn: entry.turn, inline: JSON.parse(new TextDecoder().decode(canonicalBody(entry.message, messageId))), |
| 141 | contentRef: (entry.refs ?? []).length > 0 ? { digest: `mock-canonical:${index}`, bytes: body.length, mediaType: "application/json" } : undefined, |
| 142 | }; |
| 143 | }); |
| 144 | } |
| 145 | |
| 146 | function encodedChunk(bytes: Uint8Array): string { |
| 147 | let binary = ""; |
| 148 | for (let start = 0; start < bytes.length; start += 0x8000) binary += String.fromCharCode(...bytes.subarray(start, start + 0x8000)); |
| 149 | return btoa(binary); |
| 150 | } |
| 151 | |
| 152 | export function makeMockSessionReaderBindings(): SessionReaderBindings { |
| 153 | const search = (): SearchHistoryPage => ({ hits: [], snapshotSequence: 0, coverageSequence: 0, status: "preparing", hasMore: false }); |
| 154 | async function follow(this: MockSessionReaderHost, tab: string, request: FollowRequest): Promise<TranscriptFollowResponse> { |
| 155 | const response: TranscriptFollowResponse = { protocolVersion: 2, subscription: request.subscription ?? "", changes: [], resetRequired: false }; |
| 156 | if (request.close) { mockFollowers.get(response.subscription)?.wake?.(); mockFollowers.delete(response.subscription); return response; } |
| 157 | if (!request.subscription) { |
| 158 | const id = `mock-follow-${++mockSubscription}`; |
| 159 | const follower = { tab, epoch: undefined as string | undefined, revision: 1, coverage: 0, queue: [] as Change[] }; |
| 160 | mockFollowers.set(id, follower); |
| 161 | const meta = mockMetadata.get(tab); |
| 162 | const active = mockActive.get(tab) ? { ...mockActive.get(tab)! } : undefined; |
| 163 | const pendingEvents = [...(mockPrompts.get(tab)?.values() ?? [])]; |
| 164 | follower.epoch = meta?.runtime?.epoch; |
| 165 | const page = await (this.SessionHistoryWindowForTab ?? bindings.SessionHistoryWindowForTab).call(this, tab, { anchor: "newest", limit: 32 }); |
| 166 | follower.coverage = page.snapshotSequence; |
| 167 | follower.queue = follower.queue.map(change => ({ ...change, commitSeq: page.snapshotSequence, durableSeq: page.snapshotSequence })); |
| 168 | return { ...response, subscription: id, history: page, snapshot: { |
| 169 | protocolVersion: 2, snapshotId: id, identity: { sessionId: tab, runtimeEpoch: meta?.runtime?.epoch ?? "mock-runtime", headId: "", rewriteEpoch: 0 }, |
| 170 | projectionRevision: 1, coveredThroughSeq: page.snapshotSequence, durableSeq: page.snapshotSequence, |
| 171 | records: [], activeRecords: active ? [{ id: `m:${active.id}`, order: page.messages.length, message: { role: "assistant", messageId: active.id, content: active.text, reasoning: active.reasoning }, refs: [] }] : [], |
| 172 | activeAttempts: active ? [{ id: active.id, messageId: active.id, turnId: "mock-turn", nextIndex: 0 }] : [], totalRecords: page.messages.length + (active ? 1 : 0), totalTurns: page.totalTurns, |
| 173 | before: 0, hasOlder: page.hasOlder, stale: false, |
| 174 | runtime: { turnId: meta?.turnId, status: pendingEvents.length ? "waiting_user" : meta?.running ? "in_progress" : "completed", pendingEvents: pendingEvents as unknown as NonNullable<TranscriptFollowResponse["snapshot"]>["runtime"]["pendingEvents"], samplingCount: 0, toolCount: 0 }, |
| 175 | } }; |
| 176 | } |
| 177 | const follower = mockFollowers.get(request.subscription); |
| 178 | if (!follower) return { ...response, resetRequired: true }; |
| 179 | follower.queue = follower.queue.filter(change => change.revision > (request.afterRevision ?? 0)); |
| 180 | if (!follower.queue.length) await new Promise<void>(resolve => { follower.wake = resolve; }); |
| 181 | follower.wake = undefined; |
| 182 | return { ...response, changes: [...follower.queue] }; |
| 183 | } |
| 184 | const bindings: SessionReaderBindings = { |
| 185 | TranscriptFollowForTab: follow, |
| 186 | RemoteTranscriptFollowForTab: follow, |
| 187 | async SessionHistoryPageForTab(this: MockSessionReaderHost, tabID, cursor, limit) { |
| 188 | const slice = await mockReaderSlice(this, tabID, { cursor, entries: limit, turns: limit }); |
| 189 | const history = slice.entries.some(entry => entry.refs?.length) ? await this.HistoryForTab?.(tabID) ?? [] : []; |
| 190 | return { messages: persistentMessages(slice, history), snapshotSequence: slice.revision, coverageSequence: slice.revision, status: "ready", totalTurns: slice.totalTurns, generation: slice.digest ?? "", nextCursor: slice.nextCursor, hasMore: slice.hasOlder }; |
| 191 | }, |
| 192 | async SessionOpenForTab(this: MockSessionReaderHost, tabID) { |
| 193 | const slice = await mockReaderSlice(this, tabID, { cursor: "", entries: 100, turns: 100 }); |
| 194 | const history = slice.entries.some(entry => entry.refs?.length) ? await this.HistoryForTab?.(tabID) ?? [] : []; |
| 195 | const entries = persistentMessages(slice, history); |
| 196 | return { session: { hostId: "local", sessionId: tabID }, storageGeneration: slice.digest, snapshotSequence: slice.revision, acceptedSequence: slice.revision, durableSequence: slice.revision, recent: { version: 1, sessionId: tabID, storageGeneration: slice.digest ?? "", durableSequence: slice.revision, totalTurns: slice.totalTurns, entries }, recovery: "ready", history: "ready", search: "preparing", canExecute: true }; |
| 197 | }, |
| 198 | async SessionHistoryContentForTab(this: MockSessionReaderHost, tabID, ref, offset) { |
| 199 | const index = Number(ref.digest.replace("mock-canonical:", "")); |
| 200 | const history = await this.HistoryForTab?.(tabID) ?? []; |
| 201 | const message = history[index]; |
| 202 | const entryId = `smock-${tabID}:r0:m${index}:o0`; |
| 203 | await this.HistoryContentForTab(tabID, { entryId, field: "content", size: message?.content?.length ?? 0, chunks: 1, revision: 0, digest: "mock" }, 0); |
| 204 | const body = canonicalBody(message, entryId); |
| 205 | const nextOffset = Math.min(body.length, offset + (1 << 20)); |
| 206 | return { data: encodedChunk(body.subarray(offset, nextOffset)), nextOffset, done: nextOffset >= body.length }; |
| 207 | }, |
| 208 | async RemoteSessionOpenForTab(this: MockSessionReaderHost, tabID) { return this.SessionOpenForTab(tabID); }, |
| 209 | async RemoteSessionHistoryPageForTab(this: MockSessionReaderHost, tabID, cursor, limit) { return this.SessionHistoryPageForTab(tabID, cursor, limit); }, |
| 210 | async RemoteSessionHistoryContentForTab(this: MockSessionReaderHost, tabID, ref, offset) { return this.SessionHistoryContentForTab(tabID, ref, offset); }, |
| 211 | async SearchSessionHistoryForTab() { return search(); }, |
| 212 | async RemoteSearchSessionHistoryForTab() { return search(); }, |
| 213 | async LocateSessionMessageForTab(_tabID, messageID) { return { status: "not_found", messageId: messageID, snapshotSequence: 0, coverageSequence: 0 }; }, |
| 214 | async RemoteLocateSessionMessageForTab(_tabID, messageID) { return { status: "not_found", messageId: messageID, snapshotSequence: 0, coverageSequence: 0 }; }, |
| 215 | // The in-memory mock has one direction of history: a newest page and its |
| 216 | // older cursors. It answers a window request without inventing a newer |
| 217 | // cursor, which is exactly how a protocol-7 service behaves. |
| 218 | async SessionHistoryWindowForTab(this: MockSessionReaderHost, tabID, req) { |
| 219 | const slice = await mockReaderSlice(this, tabID, { |
| 220 | cursor: req.anchor === "cursor" ? req.cursor ?? "" : "", |
| 221 | entries: req.limit, |
| 222 | turns: req.limit, |
| 223 | }); |
| 224 | const history = slice.entries.some(entry => entry.refs?.length) ? await this.HistoryForTab?.(tabID) ?? [] : []; |
| 225 | return { |
| 226 | messages: persistentMessages(slice, history), |
| 227 | status: slice.stale ? "stale_cursor" : "ready", |
| 228 | snapshotSequence: slice.revision, |
| 229 | coverageSequence: slice.revision, |
| 230 | generation: slice.digest, |
| 231 | totalTurns: slice.totalTurns, |
| 232 | hasOlder: slice.hasOlder, |
| 233 | // The in-memory mock has one direction of history, exactly like a |
| 234 | // protocol-7 service: it never invents a newer cursor. |
| 235 | hasNewer: false, |
| 236 | olderCursor: slice.nextCursor, |
| 237 | newerCursor: "", |
| 238 | }; |
| 239 | }, |
| 240 | async SessionMessageFieldForTab() { return { status: "not_found", messageId: "", version: 0, field: "", totalBytes: 0, offset: 0, encoding: "utf-8" }; }, |
| 241 | async RemoteSessionHistoryWindowForTab(this: MockSessionReaderHost, tabID, req) { return this.SessionHistoryWindowForTab(tabID, req); }, |
| 242 | async RemoteSessionMessageFieldForTab() { return { status: "not_found", messageId: "", version: 0, field: "", totalBytes: 0, offset: 0, encoding: "utf-8" }; }, |
| 243 | }; |
| 244 | return bindings; |
| 245 | } |
| 246 |