diff --git a/docs/machines.md b/docs/machines.md index a04affad0..b347a802e 100644 --- a/docs/machines.md +++ b/docs/machines.md @@ -82,17 +82,30 @@ handshake does not finish within twelve seconds can be retried from Machines. Once the machine has said hello, loading its lists is waited for however long it takes on a slow link; the connection is never dropped for it. -The client detects silent remote sockets (`packages/ui/src/lib/client.ts`). -On a remote host, 25 seconds without any -frame while no call waits for its answer sends a `hello`, which the core +The client detects silent remote sockets (`packages/ui/src/lib/ws-client.ts`). +On a remote host, 25 seconds without any frame sends a `hello`, which the core answers on an open connection with its info and nothing else. That `hello` carries no token and no device signature, since the core reads none of its -params once the socket is authenticated. If no frame of -any kind arrives within 15 seconds, the socket is replaced and the calls it -carried fail with "connection lost; check the conversation before resending". -While a call waits, silence may be its answer still downloading, one frame of -up to 16 MB behind which the probe's answer would queue, so only that call's -own 120 second timeout starts the check, on any host. Returning to the page, an `online` event or +params once the socket is authenticated. A call that hears no frame at all for +3 seconds sends the same `hello`, so a click on a conversation of a machine +whose link died shows that machine offline within 13 seconds. The call counts +frames rather than reading the clock, since a frame and a call can share a +millisecond. If no frame of any kind arrives within 10 seconds of a `hello`, +the socket is replaced and the calls it carried fail as dropped. A remote +client asks the core to slice every answer (`progress`, +`features.chunkedAnswers`): each 64 KiB slice is a frame, so a long answer on a +slow link keeps the socket alive while a dead link stays silent. Requests are +not sliced, and the `hello` queues behind what the client still uploads: past +the first 64 KiB sent since the last frame arrived, the deadline grows by the +time those bytes take at 32 KiB/s, so a prompt with 2 MiB of pictures gets 62 +seconds more. A core without sliced answers may still be sending one frame of up to +16 MB: there, while a call waits, only that call's own 120 second timeout +starts the check. Until 2026-10-22 that rule held on every core, and a VPN +that dropped under an open conversation left the machine reading as connected +for over two minutes while every call made there waited. A failure that only +says the machine went away is not shown in red: the machine's status, its +greyed rows and the composer already say it, and a conversation opened as the +link went shows offline until the reconnect opens it again. Returning to the page, an `online` event or a page restored from the back/forward cache also send the `hello` first, and replace the socket only if it stays silent for 4 seconds, so a tab switch no longer drops a healthy connection or reloads the lists. An `offline` event asks @@ -314,10 +327,47 @@ its reason and holds the entries written after it; Send again retries it under the same id, and the cross removes it. A tap takes it back into the box, and sending it from there draws a new id. A thread that moved to another model meanwhile is put back on the written one first; a choice for another agent or -account is ignored. A new thread needs the core, so a -draft keeps its text until the machine returns. A machine offline since the app +account is ignored. A machine offline since the app started has no cards to open: the client keeps no copy of the thread list. +A new conversation waits the same way. Send on a draft of a machine that is +offline puts the prompt in the draft's own queue with a request id and the +chips' choice, and the box says the conversation starts when the machine +reconnects. The draft's row in the sidebar shows a clock and the prompt, and +its tooltip names the machine it waits for. Its menu has **Cancel the start**, +which puts the words back in the box, where they wait as an ordinary draft; +the cross on the queued prompt removes it. The queue is kept on the device +like an outbox, so it survives closing the app. When the machine is back, the +client starts each queued draft in the order it was sent: `threads.create` +under the prompt's request id (`clientRequestId`), then the draft's queue +becomes the new conversation's outbox, which sends it under the same ids. The +core answers a creation it already made under that id with the conversation +it made (`packages/core/src/threads/creation-requests.ts`, kept for an hour in +memory), so a creation whose answer the link lost is not made twice. The draft +goes unless its box holds new text, and a draft on screen gives way to its +conversation. A refusal stays on the queued prompt with its reason and Send +again. A start on its way can be neither deleted nor taken back, since the +core may already hold the conversation. The thread's outbox is written to the +device before the draft leaves it, so a crash between the two leaves the prompt +in both, which the core takes once under its ids, never in neither. The first +start writes down the project and the worktree and incognito switches it asked +for, and a retry asks for exactly that, since the core answers a request id +that may already hold its conversation only when the parameters match. Once +asked, a start the core did not refuse is retried until it lands and is no +longer taken back or removed from the queue, only deleted with its draft: the +same words sent again under a new id could make a second conversation. A +refusal made nothing, so a refused start can be taken back, edited or sent +again, and its next ask reads the draft afresh. The exception is a refusal +about the request id itself (`field: 'clientRequestId'`): the core holds a +conversation under that id, so the start keeps what it asked and stays locked. +When that conversation was deleted since (`reason: 'deleted'`), nothing holds +the id any more: the start unlocks, and Send again goes under a fresh id. A start +that gets no answer on a live link, or finds no drafts project, is tried again +15 seconds later rather than at once. A start taken back before it was ever +asked, then sent again, goes at once under its new id. An incognito draft is never kept on the +device, so it still waits for its machine before it can be sent +(`packages/ui/src/lib/store/draft-launch.svelte.ts`). + Archiving or restoring a project of a machine that is offline still works: the project moves at once and the choice is kept on this device, per core, in localStorage (`boite.pendingArchive.v1`). The machine's next project list is diff --git a/docs/phone.md b/docs/phone.md index 6940c49b5..0177f5f9e 100644 --- a/docs/phone.md +++ b/docs/phone.md @@ -757,7 +757,8 @@ devices you intend to give that control. journal is on the core. Prompts written in a thread that was open before the connection went wait in the device's outbox (`docs/machines.md`), and go out once it is back, even after - the PWA was closed in between; a new thread still needs the core. + the PWA was closed in between. A new conversation sent from a draft waits + there too, and starts once the core is back. - A phone is added with a QR code or a link somebody shows it, within ten minutes. There is no discovery on the network. Adding it once is enough for every machine of its [group](groups.md). diff --git a/packages/contracts/src/messages.ts b/packages/contracts/src/messages.ts index 27cfd3df9..b51ae5c08 100644 --- a/packages/contracts/src/messages.ts +++ b/packages/contracts/src/messages.ts @@ -326,7 +326,7 @@ export interface MessagesRpcMethods { 'push.test': { params: Record; result: { ok: true } }; /** Reserve the final workspace and initialize a native session without publishing a conversation or sending text. */ 'drafts.prepare': { - params: Omit & { preparationId: string }; + params: Omit & { preparationId: string }; result: { preparationId: string; threadId: ThreadId }; }; /** Release only the named reservation owned by this connection, including preparation still in flight. */ diff --git a/packages/contracts/src/threads.ts b/packages/contracts/src/threads.ts index 880f5f68a..11009aac3 100644 --- a/packages/contracts/src/threads.ts +++ b/packages/contracts/src/threads.ts @@ -589,6 +589,17 @@ export interface ThreadsRpcMethods { preparationId?: string; /** Subscribe this connection before returning the newly created conversation. */ subscribe?: boolean; + /** + * The client's id for this creation, 1 to 128 characters. Asked again + * under the same id within an hour, with the same parameters (project, + * provider, account, title, folder, model, effort, speed, mode, + * worktree, incognito, context), the core answers with the conversation + * the first call made instead of making another; with other ones it + * refuses. A + * conversation queued while its machine was away is started this way. + * Kept in memory: a core restart forgets it. An older core ignores it. + */ + clientRequestId?: string; }; result: ThreadSummary; }; diff --git a/packages/core/src/threads/creation-requests.ts b/packages/core/src/threads/creation-requests.ts new file mode 100644 index 000000000..9ea05264d --- /dev/null +++ b/packages/core/src/threads/creation-requests.ts @@ -0,0 +1,50 @@ +import type { RpcParams, ThreadId, ThreadSummary } from '@boite/contracts'; +import { invalidParams, refused } from '../errors.ts'; + +/** How long a creation's request id answers with the thread it made: well past a reconnect. */ +const RECEIPT_MS = 60 * 60_000; +/** Receipts kept at most; the oldest leaves first. */ +const RECEIPT_LIMIT = 512; +const REQUEST_ID_MAX = 128; + +type Receipt = { expiresAt: number; key: string; created: Promise }; + +/** + * `threads.create` under a `clientRequestId`. A client that lost the answer + * with its connection asks again under the same id and gets the conversation + * the first call made, not a second one: a conversation queued on a device + * while its machine was away starts once however often the link drops on the + * way. Kept in memory, so a core restart forgets them. + */ +export class CreationRequests { + private readonly receipts = new Map(); + + async once(params: RpcParams<'threads.create'>, create: () => Promise, lookup: (threadId: ThreadId) => ThreadSummary | null): Promise { + const id = params.clientRequestId; + if (id === undefined) return create(); + if (typeof id !== 'string' || id.length === 0 || id.length > REQUEST_ID_MAX) { + throw invalidParams(`clientRequestId must be a string of 1 to ${REQUEST_ID_MAX} characters`, { field: 'clientRequestId', expected: `1 to ${REQUEST_ID_MAX} characters` }); + } + // Everything that shapes the conversation: a repeat that differs is another request, refused. + const key = JSON.stringify([params.projectId, params.providerId, params.accountId, params.title ?? null, params.cwd ?? null, + params.model ?? null, params.effort ?? null, params.speed ?? null, params.permissionMode ?? 'default', params.worktree ?? null, + params.incognito === true, params.context ?? null]); + const now = Date.now(); + for (const [stale, receipt] of this.receipts) if (receipt.expiresAt <= now) this.receipts.delete(stale); + const previous = this.receipts.get(id); + if (previous) { + if (previous.key !== key) throw refused('clientRequestId was already used for another conversation', { field: 'clientRequestId', expected: 'a new id for a different conversation' }); + const threadId = await previous.created; + const thread = lookup(threadId); + if (thread === null) throw refused('the conversation this request started was deleted since', { field: 'clientRequestId', threadId, reason: 'deleted' }); + return thread; + } + const made = create(); + const receipt: Receipt = { expiresAt: now + RECEIPT_MS, key, created: made.then(thread => thread.id) }; + this.receipts.set(id, receipt); + while (this.receipts.size > RECEIPT_LIMIT) this.receipts.delete(this.receipts.keys().next().value!); + // A creation that failed made nothing: the same id may try again. + receipt.created.catch(() => { if (this.receipts.get(id) === receipt) this.receipts.delete(id); }); + return made; + } +} diff --git a/packages/core/src/threads/rpc.ts b/packages/core/src/threads/rpc.ts index 3ddff6f9c..ccd37f50c 100644 --- a/packages/core/src/threads/rpc.ts +++ b/packages/core/src/threads/rpc.ts @@ -11,6 +11,7 @@ import { agentShape } from './agent-shape.ts'; import { steerUser } from './user-steering.ts'; import { readToolOutput, readToolPart } from './records.ts'; import { readMessageAttachment } from './attachment-read.ts'; +import { CreationRequests } from './creation-requests.ts'; /** * A turn in a folder that is gone is refused by that folder. An archived @@ -64,6 +65,7 @@ async function watchTarget(linked: LinkedPullRequests, threadId: ThreadId, input } export function registerThreadMethods(core: Core, probe: ProviderProbe): void { + const creations = new CreationRequests(); async function discover(providerId: ProviderId, accountId: AccountId, model: string | null, effort: string | null, speed: string | null): Promise { const provider = core.providers.require(providerId); const account = core.accounts.require(accountId); @@ -130,13 +132,15 @@ export function registerThreadMethods(core: Core, probe: ProviderProbe): void { return { ok: true }; }); core.router.register('threads.create', async (params, ctx) => { - // A thread made in a folder that is gone could never run a turn: refused here, by the folder. - const project = core.projects.require(params.projectId); - await core.projects.requireFolder(project.id); - const provider = core.providers.require(params.providerId); - await discover(provider.id, params.accountId, params.model ?? defaultModel(provider), params.effort ?? null, params.speed ?? null); - const adopted = await core.drafts.claim(ctx.connection.id, params); - const thread = adopted ?? (params.worktree === undefined ? core.threads.create(params) : await core.threads.createInWorktree(params)); + const thread = await creations.once(params, async () => { + // A thread made in a folder that is gone could never run a turn: refused here, by the folder. + const project = core.projects.require(params.projectId); + await core.projects.requireFolder(project.id); + const provider = core.providers.require(params.providerId); + await discover(provider.id, params.accountId, params.model ?? defaultModel(provider), params.effort ?? null, params.speed ?? null); + const adopted = await core.drafts.claim(ctx.connection.id, params); + return adopted ?? (params.worktree === undefined ? core.threads.create(params) : await core.threads.createInWorktree(params)); + }, threadId => core.journal.getThread(threadId)); if (params.subscribe === true) ctx.connection.subscriptions.add(thread.id); return thread; }); diff --git a/packages/core/test/creation-requests.test.ts b/packages/core/test/creation-requests.test.ts new file mode 100644 index 000000000..e6fcfeb33 --- /dev/null +++ b/packages/core/test/creation-requests.test.ts @@ -0,0 +1,69 @@ +import { afterEach, beforeEach, expect, test } from 'bun:test'; +import { RpcErrorCode, type RpcParams } from '@boite/contracts'; +import type { CoreClient } from '../src/client.ts'; +import { startTestCore, type TestCore } from './harness.ts'; +import { rejection } from './settle.ts'; + +let h: TestCore; +let choice: RpcParams<'threads.create'>; +beforeEach(async () => { + h = await startTestCore({ settings: { warmProcessMinutes: 0 } }); + const client = await h.connect(); + const project = await client.call('projects.drafts', {}); + const account = (await client.call('accounts.list', {})).find(account => account.providerId === 'echo')!; + choice = { projectId: project.id, providerId: 'echo', accountId: account.id, title: 'Queued while away' }; + client.close(); +}); +afterEach(async () => { await h.stop(); }); + +const titled = async (client: CoreClient, title: string) => (await client.call('threads.list', {})).filter(thread => thread.title === title); + +test('a creation asked again under its request id, on a new connection, answers with the conversation it made', async () => { + const first = await h.connect(); + const made = await first.call('threads.create', { ...choice, clientRequestId: 'req_queued_1' }); + // The answer was lost with the link: the device reconnects and asks again. + first.close(); + const second = await h.connect(); + const again = await second.call('threads.create', { ...choice, clientRequestId: 'req_queued_1' }); + expect(again.id).toBe(made.id); + expect(await titled(second, 'Queued while away')).toHaveLength(1); + // No mode is the default mode, as the core stores it. + const explicit = await second.call('threads.create', { ...choice, permissionMode: 'default', clientRequestId: 'req_queued_1' }); + expect(explicit.id).toBe(made.id); + + // Asked twice at once, it is still one conversation. + const [a, b] = await Promise.all([ + second.call('threads.create', { ...choice, title: 'Twice at once', clientRequestId: 'req_queued_2' }), + second.call('threads.create', { ...choice, title: 'Twice at once', clientRequestId: 'req_queued_2' }) + ]); + expect(a.id).toBe(b.id); + expect(await titled(second, 'Twice at once')).toHaveLength(1); + + // Without an id, or under a new one, every call makes its own. + await second.call('threads.create', { ...choice, title: 'Plain' }); + await second.call('threads.create', { ...choice, title: 'Plain' }); + expect(await titled(second, 'Plain')).toHaveLength(2); + second.close(); +}); + +test('a request id names one conversation: reused for another it is refused, and a refused creation may try again', async () => { + const client = await h.connect(); + await client.call('threads.create', { ...choice, clientRequestId: 'req_once' }); + const reused = await rejection(client.call('threads.create', { ...choice, title: 'Something else', clientRequestId: 'req_once' })); + expect((reused.reason as { rpc: unknown }).rpc).toMatchObject({ code: RpcErrorCode.Refused, data: { field: 'clientRequestId' } }); + // Every parameter names the conversation, the mode and the effort included. + const otherMode = await rejection(client.call('threads.create', { ...choice, permissionMode: 'bypassPermissions', clientRequestId: 'req_once' })); + expect((otherMode.reason as { rpc: unknown }).rpc).toMatchObject({ code: RpcErrorCode.Refused, data: { field: 'clientRequestId' } }); + const otherEffort = await rejection(client.call('threads.create', { ...choice, effort: 'low', clientRequestId: 'req_once' })); + expect((otherEffort.reason as { rpc: unknown }).rpc).toMatchObject({ code: RpcErrorCode.Refused, data: { field: 'clientRequestId' } }); + + const missing = await rejection(client.call('threads.create', { ...choice, accountId: 'acc_missing', clientRequestId: 'req_retry' })); + expect(missing.reason).toBeDefined(); + // Nothing was made under that id: the corrected call takes it. + const retried = await client.call('threads.create', { ...choice, title: 'After a refusal', clientRequestId: 'req_retry' }); + expect(retried.title).toBe('After a refusal'); + + const bad = await rejection(client.call('threads.create', { ...choice, clientRequestId: 'x'.repeat(129) })); + expect((bad.reason as { rpc: unknown }).rpc).toMatchObject({ code: RpcErrorCode.InvalidParams, data: { field: 'clientRequestId' } }); + client.close(); +}); diff --git a/packages/ui/src/components/Composer.svelte b/packages/ui/src/components/Composer.svelte index fc308c896..d5c87c1e1 100644 --- a/packages/ui/src/components/Composer.svelte +++ b/packages/ui/src/components/Composer.svelte @@ -116,11 +116,11 @@ } }); /** - * A thread whose machine dropped still takes prompts: they wait in its queue, - * which the store keeps on the device and sends once the machine is back. - * A new thread needs the core to exist, so a draft waits for the machine. + * A thread whose machine dropped still takes prompts: they wait in its queue on the device and go out + * once the machine is back. A draft's queue starts its conversation then (`launchQueuedDrafts`); an + * incognito draft is never kept on the device, so it waits for the machine. */ - let offline = $derived(store.connection !== 'ready' && thread !== null && !store.draft); + let offline = $derived(!target && store.connection !== 'ready' && (thread !== null ? !store.draft : !!store.draft && !store.draft.incognito)); let machineLabel = $derived(workspace.machines.find((machine) => machine.store === store)?.label ?? strings.machines.local); /** * A question waiting for a written answer makes the box its free field: Send @@ -148,7 +148,7 @@ : reply ? strings.composer.replyPlaceholder : offline - ? fill(strings.composer.placeholderOffline, { machine: machineLabel }) + ? fill(thread ? strings.composer.placeholderOffline : strings.composer.placeholderOfflineDraft, { machine: machineLabel }) : !thread && store.draft ? strings.composer.placeholderNew : store.openProject && provider @@ -718,7 +718,7 @@ {#if composer && composer.queued.length > 0} 0 || attachments.length > 0 || previewReferences.length > 0} - paused={composer.paused} sending={composer.sending} outgoing={composer.outgoing ?? 0} connected={store.connection === 'ready'} machine={machineLabel} + paused={composer.paused} sending={composer.sending} outgoing={composer.outgoing ?? 0} connected={store.connection === 'ready'} machine={machineLabel} starts={!thread} {sendNow} onrestore={restoreQueued} onremove={(at) => store.removeQueued(key, at)} onretry={(at) => store.retryQueued(key, at)} onsendnow={sendQueuedNow} /> diff --git a/packages/ui/src/components/Composer.test.ts b/packages/ui/src/components/Composer.test.ts index 18ae10e3f..be7f58ac7 100644 --- a/packages/ui/src/components/Composer.test.ts +++ b/packages/ui/src/components/Composer.test.ts @@ -676,7 +676,7 @@ test('a refused turn keeps the prompt for Enter and Ctrl+Enter', async ({ app: _ } }); -test('reconnecting queues a send instead of sending it, and a new thread waits for the machine', async ({ app: _app }) => { +test('reconnecting queues a send instead of sending it, and a new thread queues to start with the machine', async ({ app: _app }) => { await store.open('t-trace'); const submit = vi.spyOn(store, 'submit'); store.connection = 'connecting'; @@ -685,17 +685,18 @@ test('reconnecting queues a send instead of sending it, and a new thread waits f press('Enter'); await waitFor(() => input().value === '' && store.composerStates['t-trace']?.queued.length === 1); expect(submit).not.toHaveBeenCalled(); - // A draft has no thread to queue behind: its text stays and nothing goes out. + // A draft queues too: its conversation starts once the machine is back, under the prompt's request id. store.connection = 'ready'; await openDraft(); store.connection = 'connecting'; await type('a new thread'); - expect(query('[data-testid=composer-send]').disabled).toBe(true); + expect(input().placeholder).toContain('the conversation starts when it reconnects'); + expect(query('[data-testid=composer-send]').disabled).toBe(false); press('Enter'); - press('Enter', { ctrlKey: true }); - await new Promise((resolve) => setTimeout(resolve, 0)); + await waitFor(() => input().value === '' && store.composerStates.draft?.queued.length === 1); + expect(store.composerStates.draft!.queued[0]).toMatchObject({ text: 'a new thread', request: { id: expect.any(String), choice: expect.objectContaining({ providerId: expect.any(String) }) } }); expect(submit).not.toHaveBeenCalled(); - expect(input().value).toBe('a new thread'); + expect(store.draftEntries.find(entry => entry.active)).toMatchObject({ text: 'a new thread', queued: 'waiting' }); }); test('ArrowUp removes the latest queued prompt and restores its images for editing', async ({ app: _app }) => { diff --git a/packages/ui/src/components/ComposerQueue.svelte b/packages/ui/src/components/ComposerQueue.svelte index 949420fd3..3afeee7da 100644 --- a/packages/ui/src/components/ComposerQueue.svelte +++ b/packages/ui/src/components/ComposerQueue.svelte @@ -19,6 +19,7 @@ outgoing = 0, connected = true, machine = '', + starts = false, sendNow, onrestore, onremove, @@ -37,6 +38,8 @@ /** The machine answers: an outbox prompt waits for the thread, not for it. */ connected?: boolean; machine?: string; + /** A draft's queue: its first prompt starts the conversation once the machine is back. */ + starts?: boolean; /** What Send now would do: steer the turn, resume the held queue, or nothing. */ sendNow: 'steer' | 'resume' | null; onrestore: (at: number) => void; @@ -53,7 +56,7 @@
+ disabled={at < outgoing || !!entry.request?.start} onclick={() => onremove(at)}> {/if}
{/if} {/each}
- {outbox ? strings.composer.outboxKept : paused ? strings.composer.queuedPaused : strings.composer.queued} + {outbox && starts ? fill(strings.composer.outboxStarts, { machine }) : outbox ? strings.composer.outboxKept : paused ? strings.composer.queuedPaused : strings.composer.queued} {#if sendNow} @@ -42,6 +56,9 @@ .draft { width: 100%; height: auto; min-height: var(--row); justify-content: flex-start; gap: 8px; padding: 8px 10px; font-weight: 400; -webkit-touch-callout: none; } .draft.active { background: var(--color-active); } .draft :global(svg) { flex: none; color: var(--color-accent); } + .draft[data-queued='waiting'] :global(svg) { color: var(--color-muted-foreground); } + .draft[data-queued='starting'] :global(svg) { color: var(--color-accent); } + .draft.failed :global(svg) { color: var(--color-danger); } .title { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; text-align: left; } .mark { width: 8px; height: 8px; flex: none; border-radius: 50%; border: 1.5px dashed var(--color-muted-foreground); } diff --git a/packages/ui/src/lib/client.test.ts b/packages/ui/src/lib/client.test.ts index a9e751d88..81939acd2 100644 --- a/packages/ui/src/lib/client.test.ts +++ b/packages/ui/src/lib/client.test.ts @@ -14,6 +14,9 @@ const CORE: CoreInfo = { trace: { os: 'windows', mode: 'events', note: 'exact' } }; +/** A core that answers a call asking for `progress` in slices, as every core since 2026-10-05 does. */ +const SLICING: CoreInfo = { ...CORE, features: { chunkedAnswers: true } }; + interface Frame { id?: number; method?: string; @@ -302,7 +305,160 @@ describe('WsClient', () => { } finally { online.mockRestore(); client.close(); vi.useRealTimers(); } }); - test('a remote call whose one large answer takes 45 s to arrive keeps its socket and resolves', async () => { + test('a remote call asks for its answer in slices, and slices arriving for 45 s keep the socket', async () => { + vi.useFakeTimers(); + const sockets: FakeSocket[] = []; + const client = new WsClient({ url: 'https://core.test', token: 'session', backoff: () => 1_000, socketFactory: () => { + const socket = new FakeSocket(); sockets.push(socket); return socket; + } }); + try { + void client.connect(); + const socket = take(sockets, 0); + socket.open(); + socket.receive({ id: socket.frame(0).id, result: { core: SLICING, principal: 'session' } }); + await client.connect(); + let outcome = ''; + void client.call('threads.list', {}).then(() => { outcome = 'answered'; }, (error: Error) => { outcome = error.message; }); + const request = JSON.parse(socket.sent[1]!) as { id: number; progress?: boolean }; + expect(request.progress).toBe(true); + + // A slow link: one slice every 9 s, the answer whole after 45 s. + const body = JSON.stringify({ jsonrpc: '2.0', id: request.id, result: [] }); + const parts = [body.slice(0, 10), body.slice(10, 20), body.slice(20, 30), body.slice(30, 40), body.slice(40)]; + for (const part of parts) { + await vi.advanceTimersByTimeAsync(9_000); + socket.onmessage?.({ data: `${RPC_CHUNK_MARK}${JSON.stringify({ id: request.id, bytes: new TextEncoder().encode(part).length, total: new TextEncoder().encode(body).length })}\n${part}` }); + } + await vi.advanceTimersByTimeAsync(0); + expect(outcome).toBe('answered'); + expect(socket.closed).toBe(false); + expect(sockets).toHaveLength(1); + // Slices are frames: the probe the call took with it, if any, was answered by them. + expect(client.state).toBe('ready'); + } finally { client.close(); vi.useRealTimers(); } + }); + + test('a half-open remote socket carrying a call is replaced 35 s after its last frame, not after the call times out', async () => { + vi.useFakeTimers(); + const sockets: FakeSocket[] = []; + const client = new WsClient({ url: 'https://core.test', token: 'session', backoff: () => 1_000, socketFactory: () => { + const socket = new FakeSocket(); sockets.push(socket); return socket; + } }); + try { + void client.connect(); + const first = take(sockets, 0); + first.open(); + first.receive({ id: first.frame(0).id, result: { core: SLICING, principal: 'session' } }); + await client.connect(); + let failure: unknown = null; + void client.call('threads.list', {}).catch((error: unknown) => { failure = error; }); + // An event a second later: the socket was heard after the call, which waits on. + await vi.advanceTimersByTimeAsync(1_000); + first.receive({ method: 'core.log', params: { level: 'info', message: 'tick', at: 1 } }); + + await vi.advanceTimersByTimeAsync(24_999); + expect(first.sent.map(raw => JSON.parse(raw).method)).toEqual(['hello', 'threads.list']); + await vi.advanceTimersByTimeAsync(1); + expect(first.sent.map(raw => JSON.parse(raw).method)).toEqual(['hello', 'threads.list', 'hello']); + expect(first.closed).toBe(false); + await vi.advanceTimersByTimeAsync(10_000); + expect(first.closed).toBe(true); + expect(client.state).toBe('connecting'); + // Dropped, not timed out: the store resends a prompt it carried once the machine is back. + expect(wasDropped(failure)).toBe(true); + await vi.advanceTimersByTimeAsync(1_000); + expect(sockets).toHaveLength(2); + } finally { client.close(); vi.useRealTimers(); } + }); + + test('a call sent in the millisecond of the last frame still probes a socket that then stays silent', async () => { + vi.useFakeTimers(); + const sockets: FakeSocket[] = []; + const client = new WsClient({ url: 'https://core.test', token: 'session', backoff: () => 1_000, socketFactory: () => { + const socket = new FakeSocket(); sockets.push(socket); return socket; + } }); + try { + void client.connect(); + const first = take(sockets, 0); + first.open(); + first.receive({ id: first.frame(0).id, result: { core: SLICING, principal: 'session' } }); + await client.connect(); + // Same instant as the hello's answer: the clock cannot tell them apart, the frame count can. + void client.call('threads.list', {}).catch(() => undefined); + await vi.advanceTimersByTimeAsync(3_000); + expect(first.sent.map(raw => JSON.parse(raw).method)).toEqual(['hello', 'threads.list', 'hello']); + await vi.advanceTimersByTimeAsync(10_000); + expect(first.closed).toBe(true); + } finally { client.close(); vi.useRealTimers(); } + }); + + test('a probe waits behind a large upload at 32 KiB/s, and an answer that lands meanwhile keeps the socket', async () => { + vi.useFakeTimers(); + const sockets: FakeSocket[] = []; + const client = new WsClient({ url: 'https://core.test', token: 'session', backoff: () => 1_000, socketFactory: () => { + const socket = new FakeSocket(); sockets.push(socket); return socket; + } }); + try { + void client.connect(); + const first = take(sockets, 0); + first.open(); + first.receive({ id: first.frame(0).id, result: { core: SLICING, principal: 'session' } }); + await client.connect(); + await vi.advanceTimersByTimeAsync(1_000); + // A prompt with 2 MiB of picture on a slow uplink: the core reads its hello only after it. + let outcome = ''; + void client.call('turns.start', { threadId: 'thr_far', prompt: 'x'.repeat(2 * 1024 * 1024) }).then(() => { outcome = 'answered'; }, (error: Error) => { outcome = error.message; }); + await vi.advanceTimersByTimeAsync(3_000); + expect(first.sent.map(raw => JSON.parse(raw).method)).toEqual(['hello', 'turns.start', 'hello']); + // 10 s and 64 s for the upload: still waiting at 70 s. + await vi.advanceTimersByTimeAsync(70_000); + expect(first.closed).toBe(false); + first.receive({ id: first.frame(1).id, result: { id: 'trn_far' } }); + await vi.advanceTimersByTimeAsync(0); + expect(outcome).toBe('answered'); + await vi.advanceTimersByTimeAsync(20_000); + expect(first.closed).toBe(false); + expect(sockets).toHaveLength(1); + } finally { client.close(); vi.useRealTimers(); } + }); + + test('a call that hears nothing for 3 s on a remote socket probes it, so a dead link shows within 13 s', async () => { + vi.useFakeTimers(); + const sockets: FakeSocket[] = []; + const client = new WsClient({ url: 'https://core.test', token: 'session', backoff: () => 1_000, socketFactory: () => { + const socket = new FakeSocket(); sockets.push(socket); return socket; + } }); + try { + void client.connect(); + const first = take(sockets, 0); + first.open(); + first.receive({ id: first.frame(0).id, result: { core: SLICING, principal: 'session' } }); + await client.connect(); + // Busy a moment ago: a call goes alone. + const quick = client.call('threads.list', {}); + first.receive({ id: first.frame(1).id, result: [] }); + await quick; + expect(first.sent.map(raw => JSON.parse(raw).method)).toEqual(['hello', 'threads.list']); + + // The link dies a second later, and the user opens a conversation there. + await vi.advanceTimersByTimeAsync(1_000); + let failure: unknown = null; + void client.call('threads.get', { threadId: 'thr_far' }).catch((error: unknown) => { failure = error; }); + void client.call('permissions.list', {}).catch(() => undefined); + await vi.advanceTimersByTimeAsync(2_999); + expect(first.sent.map(raw => JSON.parse(raw).method)).toEqual(['hello', 'threads.list', 'threads.get', 'permissions.list']); + await vi.advanceTimersByTimeAsync(1); + expect(first.sent.map(raw => JSON.parse(raw).method)).toEqual(['hello', 'threads.list', 'threads.get', 'permissions.list', 'hello']); + await vi.advanceTimersByTimeAsync(9_999); + expect(first.closed).toBe(false); + await vi.advanceTimersByTimeAsync(1); + expect(first.closed).toBe(true); + expect(client.state).toBe('connecting'); + expect(wasDropped(failure)).toBe(true); + } finally { client.close(); vi.useRealTimers(); } + }); + + test('a core that answers in one frame: a call whose answer takes 45 s to arrive keeps its socket and resolves', async () => { vi.useFakeTimers(); const sockets: FakeSocket[] = []; const client = new WsClient({ url: 'https://core.test', token: 'session', backoff: () => 1_000, socketFactory: () => { @@ -333,7 +489,7 @@ describe('WsClient', () => { } finally { client.close(); vi.useRealTimers(); } }); - test('a half-open remote socket carrying a call is replaced once that call times out', async () => { + test('a core that answers in one frame: a half-open socket carrying a call is replaced once that call times out', async () => { vi.useFakeTimers(); const sockets: FakeSocket[] = []; const client = new WsClient({ url: 'https://core.test', token: 'session', backoff: () => 1_000, socketFactory: () => { diff --git a/packages/ui/src/lib/drafts.test.ts b/packages/ui/src/lib/drafts.test.ts index cf634c854..aaa3a8ac7 100644 --- a/packages/ui/src/lib/drafts.test.ts +++ b/packages/ui/src/lib/drafts.test.ts @@ -556,8 +556,8 @@ test('a project holds several drafts: each reopens by id and a new Store restore const two = store.draft!.id!; expect(two).not.toBe(one); expect(store.draftEntries).toEqual([ - { id: two, projectId: a!.id, text: 'Two', active: true }, - { id: one, projectId: a!.id, text: 'One', active: false } + { id: two, projectId: a!.id, text: 'Two', active: true, queued: null }, + { id: one, projectId: a!.id, text: 'One', active: false, queued: null } ]); store.startDraft(undefined, { draftId: one }); expect(store.draft?.id).toBe(one); @@ -634,7 +634,7 @@ test('a journal entry keyed by its project restores with that key as its id and '"p-boite"': { draft: { projectId: 'p-boite', worktree: false }, input: { text: 'Legacy draft', attachments: [], queued: [] } } } }); const store = await ready('legacy-key'); - expect(store.draftEntries).toEqual([{ id: '"p-boite"', projectId: 'p-boite', text: 'Legacy draft', active: expect.any(Boolean) }]); + expect(store.draftEntries).toEqual([{ id: '"p-boite"', projectId: 'p-boite', text: 'Legacy draft', active: expect.any(Boolean), queued: null }]); store.startDraft('p-boite'); expect(store.draft?.id).toBe('"p-boite"'); store.editComposerText('draft', 'Legacy draft, edited'); diff --git a/packages/ui/src/lib/fake-client/threads.ts b/packages/ui/src/lib/fake-client/threads.ts index 49e87e137..ef2256e6f 100644 --- a/packages/ui/src/lib/fake-client/threads.ts +++ b/packages/ui/src/lib/fake-client/threads.ts @@ -277,6 +277,7 @@ export async function deleteExpiredDoneThreads(ctx: FakeContext, remove: (id: st } export function threadMethods(ctx: FakeContext) { + const creations = new Map }>(); const removing = new Set(); return { ...sideQuestionMethods(ctx, (source, messages) => writeFakeFork(ctx, source, messages)), @@ -286,7 +287,32 @@ export function threadMethods(ctx: FakeContext) { .filter((t) => (params.includeArchived ? true : !t.archived)) .map((t) => structuredClone(toSummary(t))); }, - 'threads.create': async (params) => publishFakeThread(ctx, await createFakeThread(ctx, params), params.subscribe), + 'threads.create': async (params) => { + // The core's creation receipts: asked again under its request id, a creation answers with the conversation it made. + const id = params.clientRequestId; + if (id !== undefined && (typeof id !== 'string' || id.length === 0 || id.length > 128)) { + throw new RpcFailure({ code: RpcErrorCode.InvalidParams, message: 'clientRequestId must be a string of 1 to 128 characters', data: { field: 'clientRequestId', expected: '1 to 128 characters' } }); + } + const key = JSON.stringify([params.projectId, params.providerId, params.accountId, params.title ?? null, params.cwd ?? null, + params.model ?? null, params.effort ?? null, params.speed ?? null, params.permissionMode ?? 'default', params.worktree ?? null, + params.incognito === true, params.context ?? null]); + const previous = id === undefined ? undefined : creations.get(id); + if (previous) { + if (previous.key !== key) throw refusal('clientRequestId was already used for another conversation', { field: 'clientRequestId', expected: 'a new id for a different conversation' }); + const threadId = await previous.created; + const made = ctx.threads.get(threadId); + if (!made) throw refusal('the conversation this request started was deleted since', { field: 'clientRequestId', threadId, reason: 'deleted' }); + if (params.subscribe) ctx.bus.subscribed.add(made.id); + return structuredClone(toSummary(made)); + } + const creating = createFakeThread(ctx, params).then(thread => publishFakeThread(ctx, thread, params.subscribe)); + if (id !== undefined) { + const receipt = { key, created: creating.then(thread => thread.id) }; + creations.set(id, receipt); + receipt.created.catch(() => { if (creations.get(id) === receipt) creations.delete(id); }); + } + return creating; + }, 'threads.get': async (params) => { const thread = ctx.thread(params.threadId); const problem = snapshotOptionsProblem(params) ?? pageBytesProblem(params.maxBytes); diff --git a/packages/ui/src/lib/queued-drafts.test.ts b/packages/ui/src/lib/queued-drafts.test.ts new file mode 100644 index 000000000..001d33100 --- /dev/null +++ b/packages/ui/src/lib/queued-drafts.test.ts @@ -0,0 +1,325 @@ +import { afterEach, beforeEach, expect, test, vi } from 'vitest'; +import { RpcErrorCode } from '@boite/contracts'; +import { RpcFailure, droppedFailure, unansweredFailure } from './client'; +import { FakeClient } from './fake-client'; +import { Store } from './store.svelte'; +import { readDraftJournal, writeDraftJournal } from './draft-journal'; + +// jsdom has no IndexedDB: the durable journal is a recorder here, and the +// synchronous backup in localStorage is what a new Store reads back. +vi.mock('./draft-journal', () => ({ readDraftJournal: vi.fn(async () => null), writeDraftJournal: vi.fn(async () => {}) })); + +const stores: Store[] = []; +const clients: FakeClient[] = []; + +/** A Store on `machine` whose client records every creation and start. */ +async function machine(id: string) { + const store = new Store(); + store.machineId = id; + const client = new FakeClient({ delayMs: 0 }); + clients.push(client); + const original = client.call.bind(client); + const creates: { title?: string; clientRequestId?: string }[] = []; + const starts: { threadId: string; prompt: string; clientRequestId: string }[] = []; + vi.spyOn(client, 'call').mockImplementation(async (method, params) => { + if (method === 'threads.create') creates.push(params as { title?: string; clientRequestId?: string }); + if (method === 'turns.start') starts.push(params as { threadId: string; prompt: string; clientRequestId: string }); + return original(method, params); + }); + store.attach(client); + stores.push(store); + return { store, client, creates, starts, original }; +} + +beforeEach(() => localStorage.clear()); +afterEach(() => { + for (const store of stores.splice(0)) { store.client?.close(); store.detach(); } + for (const client of clients.splice(0)) client.close(); + vi.restoreAllMocks(); + vi.mocked(readDraftJournal).mockResolvedValue(null); + vi.mocked(writeDraftJournal).mockResolvedValue(); +}); + +/** What the composer does with Send on a draft whose machine is away. */ +function sendAway(store: Store, text: string) { + store.queuePrompt('draft', { text, attachments: [] }, { choice: store.defaultChoice() }); +} + +async function titled(client: FakeClient, title: string) { + return (await client.call('threads.list', {})).filter(thread => thread.title === title); +} + +test('a draft sent while its machine is away waits on its row, survives closing the app and starts once, on screen', async () => { + const first = await machine('one'); + await first.store.connect(); + const project = first.store.projects.find(entry => entry.kind !== 'drafts')!; + first.store.startDraft(project.id, { fresh: true }); + first.client.drop(); + sendAway(first.store, 'Start the migration'); + // The prompt waits in the draft's own queue, under a request id; the row says the conversation waits for the machine. + expect(first.store.composerStates.draft!.queued).toMatchObject([{ text: 'Start the migration', request: { id: expect.any(String) } }]); + const [row] = first.store.draftEntries; + expect(row).toMatchObject({ projectId: project.id, text: 'Start the migration', queued: 'waiting', active: true }); + expect(first.creates).toHaveLength(0); + first.store.client?.close(); first.store.detach(); + + const next = await machine('one'); + await next.store.connect(); + const draftId = row!.id; + // Back on the draft it was written in, as the next start opens where it was left. + next.store.startDraft(project.id, { draftId }); + await vi.waitFor(() => expect(next.starts).toHaveLength(1), { timeout: 5000 }); + expect(next.creates).toHaveLength(1); + const created = next.creates[0]!; + expect(created.title).toBe('Start the migration'); + expect(next.starts[0]).toMatchObject({ prompt: 'Start the migration', clientRequestId: created.clientRequestId }); + // The draft became that conversation, which is on screen now. + await vi.waitFor(() => expect(next.store.openThread?.title).toBe('Start the migration')); + expect(next.store.draftEntries).toHaveLength(0); + expect(await titled(next.client, 'Start the migration')).toHaveLength(1); +}); + +test('a start whose answer the link lost is asked again under its id and makes one conversation', async () => { + const { store, client, creates, starts, original } = await machine('one'); + await store.connect(); + const project = store.projects.find(entry => entry.kind !== 'drafts')!; + store.startDraft(project.id, { fresh: true }); + client.drop(); + sendAway(store, 'Only once please'); + let lose = true; + vi.mocked(client.call).mockImplementation(async (method, params) => { + if (method === 'threads.create') creates.push(params as { title?: string; clientRequestId?: string }); + if (method === 'turns.start') starts.push(params as { threadId: string; prompt: string; clientRequestId: string }); + const result = await original(method, params); + // The core made it; the link went before its answer arrived. + if (method === 'threads.create' && lose) { lose = false; client.drop(); throw droppedFailure('connection lost; check the conversation before resending'); } + return result; + }); + await client.restore(); + await vi.waitFor(() => expect(creates).toHaveLength(1), { timeout: 5000 }); + expect(starts).toHaveLength(0); + await vi.waitFor(() => expect(store.connection).toBe('connecting')); + expect(store.draftEntries[0]).toMatchObject({ queued: 'waiting' }); + // Moved to another project meanwhile: the retry still asks for what the core may already hold. + const other = store.projects.find(entry => entry.kind !== 'drafts' && entry.id !== project.id); + store.draft!.projectId = other!.id; + await client.restore(); + await vi.waitFor(() => expect(starts).toHaveLength(1), { timeout: 5000 }); + expect(creates).toHaveLength(2); + expect(creates[1]!.clientRequestId).toBe(creates[0]!.clientRequestId); + expect(creates[1]).toEqual(creates[0]); + expect(await titled(client, 'Only once please')).toHaveLength(1); +}); + +test('cancelling a queued start brings its words back into the draft, which no longer starts by itself', async () => { + const { store, client, creates } = await machine('one'); + await store.connect(); + const project = store.projects.find(entry => entry.kind !== 'drafts')!; + store.startDraft(project.id, { fresh: true }); + client.drop(); + sendAway(store, 'Never mind'); + const [row] = store.draftEntries; + store.cancelQueuedDraft(row!.id); + expect(store.composerStates.draft).toMatchObject({ text: 'Never mind', queued: [] }); + expect(store.draftEntries[0]).toMatchObject({ text: 'Never mind', queued: null }); + await client.restore(); + await vi.waitFor(() => expect(store.connection).toBe('ready')); + // Long enough for a start to have gone out: a queued draft starts within a tick of the machine's lists. + await new Promise(resolve => setTimeout(resolve, 500)); + expect(creates).toHaveLength(0); +}); + +test('a refused start keeps its reason on the draft, and Send again starts it', async () => { + const { store, client, creates, original } = await machine('one'); + await store.connect(); + const project = store.projects.find(entry => entry.kind !== 'drafts')!; + store.startDraft(project.id, { fresh: true }); + client.drop(); + sendAway(store, 'Signed out meanwhile'); + let refuse = true; + vi.mocked(client.call).mockImplementation(async (method, params) => { + if (method === 'threads.create') { + creates.push(params as { title?: string; clientRequestId?: string }); + if (refuse) { refuse = false; throw new Error('The account is signed out'); } + } + return original(method, params); + }); + await client.restore(); + await vi.waitFor(() => expect(store.draftEntries[0]?.queued).toEqual({ failed: expect.stringContaining('signed out') }), { timeout: 5000 }); + expect(creates).toHaveLength(1); + // The core made nothing: the start is no longer held as asked, so it could be taken back or edited. + expect(store.draftEntries[0]).not.toHaveProperty('asked'); + expect(store.composerStates.draft!.queued[0]!.request).not.toHaveProperty('start'); + store.retryQueued('draft', 0); + await vi.waitFor(() => expect(store.openThread?.title).toBe('Signed out meanwhile'), { timeout: 5000 }); + expect(creates).toHaveLength(2); +}); + +test('a start the core never answered, on a link still up, is not asked again in a loop', async () => { + const { store, client, creates, original } = await machine('one'); + await store.connect(); + const project = store.projects.find(entry => entry.kind !== 'drafts')!; + store.startDraft(project.id, { fresh: true }); + client.drop(); + sendAway(store, 'Slow core'); + vi.mocked(client.call).mockImplementation(async (method, params) => { + if (method === 'threads.create') { creates.push(params as { title?: string; clientRequestId?: string }); throw unansweredFailure('threads.create'); } + return original(method, params); + }); + await client.restore(); + await vi.waitFor(() => expect(creates).toHaveLength(1), { timeout: 5000 }); + await new Promise(resolve => setTimeout(resolve, 300)); + // Asked once, then left alone for a while under the same id; the draft still waits. + expect(creates).toHaveLength(1); + expect(store.draftEntries[0]).toMatchObject({ queued: 'waiting' }); +}); + +test('at every write on its way to the conversation the queued prompt is kept somewhere on the device', async () => { + const { store, client, starts } = await machine('one'); + await store.connect(); + const project = store.projects.find(entry => entry.kind !== 'drafts')!; + store.startDraft(project.id, { fresh: true }); + client.drop(); + sendAway(store, 'Never lost'); + const id = store.composerStates.draft!.queued[0]!.request!.id; + const kept = () => Object.keys(localStorage).some(name => name.includes(':entry:') && localStorage.getItem(name)!.includes(id)); + expect(kept()).toBe(true); + // A crash may come after any one write: until the core has the prompt, one of them must still hold it. + const lost: string[] = []; + const watch = (name: string) => { if (starts.length === 0 && !kept()) lost.push(name); }; + const set = Storage.prototype.setItem, remove = Storage.prototype.removeItem; + vi.spyOn(Storage.prototype, 'setItem').mockImplementation(function (this: Storage, name: string, value: string) { set.call(this, name, value); watch(name); }); + vi.spyOn(Storage.prototype, 'removeItem').mockImplementation(function (this: Storage, name: string) { remove.call(this, name); watch(name); }); + await client.restore(); + await vi.waitFor(() => expect(starts).toHaveLength(1), { timeout: 5000 }); + expect(lost).toEqual([]); +}); + +test('a draft whose start is on its way is not deleted: its prompt reaches the conversation the core made', async () => { + const { store, client, creates, starts, original } = await machine('one'); + await store.connect(); + const project = store.projects.find(entry => entry.kind !== 'drafts')!; + store.startDraft(project.id, { fresh: true }); + client.drop(); + sendAway(store, 'Already on its way'); + let release: () => void = () => undefined; + const held = new Promise(resolve => { release = resolve; }); + vi.mocked(client.call).mockImplementation(async (method, params) => { + if (method === 'threads.create') { creates.push(params as { title?: string; clientRequestId?: string }); await held; } + if (method === 'turns.start') starts.push(params as { threadId: string; prompt: string; clientRequestId: string }); + return original(method, params); + }); + await client.restore(); + await vi.waitFor(() => expect(creates).toHaveLength(1), { timeout: 5000 }); + const [row] = store.draftEntries; + expect(row).toMatchObject({ queued: 'starting' }); + store.discardDraft(row!.id); + store.cancelQueuedDraft(row!.id); + expect(store.draftEntries[0]).toMatchObject({ id: row!.id, queued: 'starting' }); + release(); + await vi.waitFor(() => expect(starts).toHaveLength(1), { timeout: 5000 }); + expect(starts[0]!.prompt).toBe('Already on its way'); + expect(store.draftEntries).toHaveLength(0); +}); + +test('a start asked once is retried as it was asked, across a restart, and is not taken back meanwhile', async () => { + const first = await machine('one'); + await first.store.connect(); + const project = first.store.projects.find(entry => entry.kind !== 'drafts')!; + first.store.startDraft(project.id, { fresh: true }); + first.client.drop(); + sendAway(first.store, 'Asked once'); + vi.mocked(first.client.call).mockImplementation(async (method, params) => { + if (method === 'threads.create') { first.creates.push(params as { title?: string; clientRequestId?: string }); throw unansweredFailure('threads.create'); } + return first.original(method, params); + }); + await first.client.restore(); + await vi.waitFor(() => expect(first.creates).toHaveLength(1), { timeout: 5000 }); + const [row] = first.store.draftEntries; + expect(row).toMatchObject({ queued: 'waiting', asked: true }); + // The core may hold it: neither the row's menu, the cross nor a tap takes it back. + first.store.cancelQueuedDraft(row!.id); + first.store.removeQueued('draft', 0); + expect(first.store.restoreQueued('draft', 0)).toBeUndefined(); + expect(first.store.composerStates.draft!.queued).toHaveLength(1); + // Moved to another project before the app closes. + first.store.draft!.projectId = first.store.projects.find(entry => entry.kind !== 'drafts' && entry.id !== project.id)!.id; + first.store.client?.close(); first.store.detach(); + + const next = await machine('one'); + await next.store.connect(); + next.store.startDraft(undefined, { draftId: row!.id }); + await vi.waitFor(() => expect(next.creates).toHaveLength(1), { timeout: 5000 }); + expect(next.creates[0]).toEqual(first.creates[0]); + expect((next.creates[0] as { projectId?: string }).projectId).toBe(project.id); +}); + +test('a start asked once and then refused may be taken back: the core made nothing', async () => { + const first = await machine('one'); + await first.store.connect(); + const project = first.store.projects.find(entry => entry.kind !== 'drafts')!; + first.store.startDraft(project.id, { fresh: true }); + first.client.drop(); + sendAway(first.store, 'Asked, then refused'); + vi.mocked(first.client.call).mockImplementation(async (method, params) => { + if (method === 'threads.create') { first.creates.push(params as { title?: string; clientRequestId?: string }); throw unansweredFailure('threads.create'); } + return first.original(method, params); + }); + await first.client.restore(); + await vi.waitFor(() => expect(first.creates).toHaveLength(1), { timeout: 5000 }); + const [row] = first.store.draftEntries; + expect(row).toMatchObject({ asked: true }); + first.store.client?.close(); first.store.detach(); + + const next = await machine('one'); + vi.mocked(next.client.call).mockImplementation(async (method, params) => { + if (method === 'threads.create') { next.creates.push(params as { title?: string; clientRequestId?: string }); throw new Error('The account is signed out'); } + return next.original(method, params); + }); + await next.store.connect(); + next.store.startDraft(undefined, { draftId: row!.id }); + await vi.waitFor(() => expect(next.store.draftEntries[0]?.queued).toEqual({ failed: expect.stringContaining('signed out') }), { timeout: 5000 }); + expect(next.store.draftEntries[0]).not.toHaveProperty('asked'); + next.store.cancelQueuedDraft(row!.id); + expect(next.store.composerStates.draft).toMatchObject({ text: 'Asked, then refused', queued: [] }); +}); + +test('a start whose conversation was deleted since unlocks, and Send again goes under a fresh id', async () => { + const first = await machine('one'); + await first.store.connect(); + const project = first.store.projects.find(entry => entry.kind !== 'drafts')!; + first.store.startDraft(project.id, { fresh: true }); + first.client.drop(); + sendAway(first.store, 'Deleted meanwhile'); + vi.mocked(first.client.call).mockImplementation(async (method, params) => { + if (method === 'threads.create') { first.creates.push(params as { title?: string; clientRequestId?: string }); throw unansweredFailure('threads.create'); } + return first.original(method, params); + }); + await first.client.restore(); + await vi.waitFor(() => expect(first.creates).toHaveLength(1), { timeout: 5000 }); + const [row] = first.store.draftEntries; + expect(row).toMatchObject({ asked: true }); + first.store.client?.close(); first.store.detach(); + + const next = await machine('one'); + let deleted = true; + vi.mocked(next.client.call).mockImplementation(async (method, params) => { + if (method === 'threads.create') { + next.creates.push(params as { title?: string; clientRequestId?: string }); + if (deleted) { + deleted = false; + throw new RpcFailure({ code: RpcErrorCode.Refused, message: 'the conversation this request started was deleted since', data: { field: 'clientRequestId', reason: 'deleted' } }); + } + } + return next.original(method, params); + }); + await next.store.connect(); + next.store.startDraft(undefined, { draftId: row!.id }); + await vi.waitFor(() => expect(next.store.draftEntries[0]?.queued).toEqual({ failed: expect.stringContaining('deleted since') }), { timeout: 5000 }); + expect(next.store.draftEntries[0]).not.toHaveProperty('asked'); + // Send again goes under a fresh id: the deleted conversation burned the first one. + next.store.retryQueued('draft', 0); + await vi.waitFor(() => expect(next.creates).toHaveLength(2), { timeout: 5000 }); + expect(next.creates[1]!.clientRequestId).not.toBe(next.creates[0]!.clientRequestId); + await vi.waitFor(() => expect(next.store.openThread?.title).toBe('Deleted meanwhile'), { timeout: 5000 }); +}); diff --git a/packages/ui/src/lib/store.svelte.ts b/packages/ui/src/lib/store.svelte.ts index 07af47567..d6db5d7e1 100644 --- a/packages/ui/src/lib/store.svelte.ts +++ b/packages/ui/src/lib/store.svelte.ts @@ -1,7 +1,7 @@ import { startDiagnosticsReport } from './diagnostics-report'; import { RpcErrorCode, type AgentShape, type Message, type MessageId, type PermissionMode, type ProjectId, type ProviderId, type ThreadId } from '@boite/contracts'; import type { Client } from './client'; -import { RpcFailure, wasClosed } from './rpc-error'; +import { RpcFailure, wasClosed, wasDropped } from './rpc-error'; import { readModelDefaults } from './model-defaults'; import { modelFavorites } from './model-favorites.svelte'; import { readLayout, readPrefs } from './prefs'; @@ -115,9 +115,14 @@ export class Store { this.errorThreadId = null; } - /** Only explicitly minor action failures expire; unknown failures remain visible. */ + /** + * Only explicitly minor action failures expire; unknown failures remain + * visible. A call that failed because its machine is gone says only what the + * machine's offline state already shows, so it is minor too. + */ reportError(error: unknown, severity: 'minor' | 'major' = 'major'): void { - this.#error = { message: this.#ctx.reason(error), severity, offline: wasClosed(error) }; + const offline = wasClosed(error) || (wasDropped(error) && this.connection !== 'ready'); + this.#error = { message: this.#ctx.reason(error), severity: offline ? 'minor' : severity, offline }; this.errorThreadId = null; } @@ -600,6 +605,7 @@ export class Store { get composerStates() { return this.#ctx.composer.composerStates; } get draftEntries() { return this.#ctx.drafts.entries; } discardDraft(id: string) { return this.#ctx.drafts.discard(id); } + cancelQueuedDraft(id: string) { return this.#ctx.drafts.unqueue(id); } get draftsReadable() { return this.#ctx.drafts.readable; } flushDrafts() { return this.#ctx.drafts.flush(); } set composerStates(value) { this.#ctx.composer.composerStates = value; } diff --git a/packages/ui/src/lib/store/composer.svelte.ts b/packages/ui/src/lib/store/composer.svelte.ts index 95aa1c459..4526dd4fe 100644 --- a/packages/ui/src/lib/store/composer.svelte.ts +++ b/packages/ui/src/lib/store/composer.svelte.ts @@ -16,6 +16,7 @@ import { secureId } from '../secure-id'; import type { Choice, Draft } from '../store.svelte'; import type { StoreContext } from './context'; import { draftKey } from './drafts.svelte'; +import { DraftLaunch } from './draft-launch.svelte'; /** * A prompt written while its machine could not be reached: the outbox. It goes @@ -26,7 +27,11 @@ import { draftKey } from './drafts.svelte'; * edits it or removes it. `sent` marks one that went out at least once: only * such a prompt may already have been taken by the core. */ -export type OutboxRequest = { id: string; choice: Choice | null; queuedAt: number; failed?: string; sent?: true }; +export type OutboxRequest = { + id: string; choice: Choice | null; queuedAt: number; failed?: string; sent?: true; + /** A draft's start as first asked of the core, so a retry under the same id asks for the same conversation. */ + start?: { projectId: string; worktree: boolean; incognito: boolean }; +}; /** A prompt waiting behind a running turn or for its machine, with what it was written with. */ export type QueuedPrompt = { text: string; attachments: Attachment[]; previewReferences?: PreviewReference[]; afterBoundary?: string; request?: OutboxRequest }; @@ -87,6 +92,7 @@ export class Composer { private steerRequests = new Map(); /** Outbox prompts already asked about while their thread ran, on this connection. */ private probed = new Set(); + private readonly launcher: DraftLaunch; previewUndo = new Map(); composerInsertions = new Map void>(); @@ -100,7 +106,7 @@ export class Composer { /** The local row a landed message took the place of: the timeline keeps its height and does not replay its rise. */ readonly landings = new Map(); - constructor(private readonly ctx: StoreContext) {} + constructor(private readonly ctx: StoreContext) { this.launcher = new DraftLaunch(ctx); } /** Queued input follows its machine and thread even when their composer is off screen. */ watchQueues(): () => void { @@ -131,10 +137,13 @@ export class Composer { } else untrack(() => void drainQueue(s, threadId, state)); } }); + // A draft written while its machine was away becomes its conversation once the machine is back. + this.launcher.watch(); }); } + private blocked(threadId: string): boolean { const s = this.ctx.store; return s.pendingPermissions.some(request => request.threadId === threadId) || @@ -533,7 +542,8 @@ export class Composer { */ restoreQueued(threadId: string, at: number): QueuedPrompt | undefined { const state = this.composerStates[threadId]; - if (!state || state.sending) return undefined; + // A draft's start asked once is retried, not taken back (`Drafts.unqueue`). + if (!state || state.sending || state.queued[at]?.request?.start) return undefined; const [entry] = state.queued.splice(at, 1); if (entry?.request) { const s = this.ctx.store; @@ -547,7 +557,7 @@ export class Composer { /** Takes a pending prompt out of the queue, unless it is the one going out right now. */ removeQueued(threadId: string, at: number): void { const state = this.composerStates[threadId]; - if (!state || at < (state.outgoing ?? 0)) return; + if (!state || at < (state.outgoing ?? 0) || state.queued[at]?.request?.start) return; state.queued.splice(at, 1); } diff --git a/packages/ui/src/lib/store/draft-launch.svelte.ts b/packages/ui/src/lib/store/draft-launch.svelte.ts new file mode 100644 index 000000000..8ad9efcef --- /dev/null +++ b/packages/ui/src/lib/store/draft-launch.svelte.ts @@ -0,0 +1,196 @@ +import { untrack } from 'svelte'; +import type { ThreadId } from '@boite/contracts'; +import { RpcFailure, wasDropped, wasUnanswered } from '../client'; +import { titleFrom } from '../format'; +import { secureId } from '../secure-id'; +import { DRAFT_STASH_KEY } from '../prefs'; +import { wasClosed } from '../rpc-error'; +import type { Choice, Draft } from '../store.svelte'; +import type { OutboxRequest } from './composer.svelte'; +import type { StoreContext } from './context'; +import { draftKey, type Input } from './drafts.svelte'; + +/** How long a queued draft whose start got nowhere waits before it is tried again. */ +const DRAFT_RETRY_MS = 15_000; + +/** + * The drafts sent while their machine was away: each becomes its conversation + * once the machine is back, in the order they were sent (`docs/machines.md`). + */ +export class DraftLaunch { + /** The queued drafts are being started, one after the other; `again` asks for one more pass. */ + private launches: { again: boolean } | null = null; + /** Drafts whose start got nowhere for a reason that may pass (no answer, no drafts project): left alone for a while, by the request that stalled. */ + private stalled = new Map(); + /** Bumped when a stalled draft may be tried again. */ + private tick = $state(0); + + constructor(private readonly ctx: StoreContext) {} + + /** Runs inside the composer's effect root: a ready machine starts what its drafts queued. */ + watch(): void { + $effect(() => { + void this.tick; + if (this.ctx.store.connection !== 'ready' || this.queuedDrafts().length === 0) return; + untrack(() => void this.launchQueuedDrafts()); + }); + } + + /** The open draft, else a saved one, under its key, with what it was written with. */ + private draftAt(key: string): { draft: Draft; input: Input; choice: Choice | null } | null { + const s = this.ctx.store; + const input = this.ctx.composer.composerStates[DRAFT_STASH_KEY]; + if (s.draft && draftKey(s.draft) === key) return input ? { draft: s.draft, input, choice: s.draftChoice } : null; + const saved = this.ctx.drafts.saved[key]; + return saved ? { draft: saved.draft, input: saved.input, choice: saved.choice } : null; + } + + /** The drafts whose first prompt waits to start their conversation, oldest queue first. */ + private queuedDrafts(): string[] { + const s = this.ctx.store; + const waiting = (input: Input | undefined): input is Input => !!input && !input.sending && !!input.queued[0]?.request && input.queued[0].request.failed === undefined; + // Only the request that stalled waits: one sent again after a cancel goes at once. + const ready = (key: string, input: Input) => this.stalled.get(key) !== input.queued[0]?.request?.id; + const keys: { key: string; at: number }[] = []; + const open = s.draft ? draftKey(s.draft) : null; + const input = this.ctx.composer.composerStates[DRAFT_STASH_KEY]; + if (open !== null && waiting(input) && ready(open, input)) keys.push({ key: open, at: input.queued[0]!.request!.queuedAt }); + for (const [key, saved] of Object.entries(this.ctx.drafts.saved)) { + if (key !== open && waiting(saved.input) && ready(key, saved.input)) keys.push({ key, at: saved.input.queued[0]!.request!.queuedAt }); + } + return keys.sort((a, b) => a.at - b.at).map(entry => entry.key); + } + + /** One pass at a time, in the order the drafts were queued, so their conversations start in that order. */ + private async launchQueuedDrafts(): Promise { + if (this.launches) { this.launches.again = true; return; } + const pass = { again: true }; + this.launches = pass; + try { + while (pass.again) { + pass.again = false; + for (const key of this.queuedDrafts()) { + if (this.ctx.store.connection !== 'ready') return; + await this.launchDraft(key); + } + } + } finally { this.launches = null; } + } + + /** + * Starts the conversation of a draft whose first prompt was written while its + * machine was away. The core makes it under that prompt's request id, so a + * creation whose answer the link lost is not made twice, and the draft's queue + * moves to the new thread's outbox, which sends it in order under the same ids. + * A refusal stays on the prompt with its reason and Send again. + */ + private async launchDraft(key: string): Promise { + const s = this.ctx.store; + const client = this.ctx.client; + const generation = this.ctx.clientGeneration; + const source = this.draftAt(key); + const head = source?.input.queued[0]; + if (!client || !source || !head?.request || head.request.failed !== undefined || source.input.sending) return; + const { draft, input } = source; + const request = head.request; + // The composer queues with the chips' choice; a draft saved without one starts on the machine's default. + const choice = request.choice ?? source.choice ?? s.defaultChoice(); + if (!choice) { this.stall(key, request.id); return; } + input.sending = true; + input.outgoing = 1; + try { + await this.ctx.connection.reloading?.essential; + if (!this.ctx.currentClient(client, generation) || s.connection !== 'ready' || input.queued[0] !== head) return; + // A start asked once is asked again the same way, whatever the draft became since: the core holds the id to it. + const start = request.start ?? await this.startOf(draft); + if (!this.ctx.currentClient(client, generation)) return; + if (start === null) { this.stall(key, request.id); return; } + if (!request.start && input.queued[0] === head) { + head.request = { ...request, start }; + this.ctx.drafts.persist(); void this.ctx.drafts.flush(); + } + const created = await client.call('threads.create', { + projectId: start.projectId, + providerId: choice.providerId, + accountId: choice.accountId, + permissionMode: choice.permissionMode, + title: titleFrom(head.text) || undefined, + effort: choice.effort, + speed: choice.speed ?? null, + ...(choice.model ? { model: choice.model } : {}), + ...(start.worktree ? { worktree: {} } : {}), + ...(start.incognito ? { incognito: true } : {}), + clientRequestId: request.id + }); + if (!this.ctx.currentClient(client, generation)) return; + this.ctx.threads.upsertThread(created); + this.adoptLaunched(key, input, created.id); + } catch (error) { + if (!this.ctx.currentClient(client, generation)) return; + // Lost with the link: the next connection asks again under the same id. Unanswered on a live one: a little later. + if (wasDropped(error) || wasClosed(error) || wasUnanswered(error) || s.connection !== 'ready') { + if (s.connection === 'ready') this.stall(key, request.id); + return; + } + if (input.queued[0] === head) { + // A refusal made nothing, so the start may be taken back or asked again differently; one about its + // request id says the core holds a conversation under it, and keeps what was asked, unless that + // conversation was deleted since: then nothing holds it, the start unlocks and Send again goes under a fresh id. + const { start: asked, ...rest } = request; + const data = error instanceof RpcFailure ? error.data as { field?: unknown; reason?: unknown } | undefined : undefined; + const held = data?.field === 'clientRequestId' && data.reason !== 'deleted'; + // A deleted conversation burned its id: Send again goes under a fresh one. + const burned = data?.field === 'clientRequestId' && data.reason === 'deleted'; + head.request = { ...rest, ...(burned ? { id: secureId() } : {}), ...(held && asked ? { start: asked } : {}), failed: this.ctx.reason(error) }; + } + this.ctx.drafts.persist(); void this.ctx.drafts.flush(); + } finally { + input.sending = false; + input.outgoing = 0; + } + } + + /** A start that got nowhere is not tried again at once, which would loop on the same answer. */ + private stall(key: string, requestId: string): void { + this.stalled.set(key, requestId); + setTimeout(() => { if (this.stalled.get(key) === requestId) this.stalled.delete(key); this.tick++; }, DRAFT_RETRY_MS); + } + + /** + * Where the draft starts: its project, or the drafts project, which a + * background start looks up without a notification, and the worktree and + * incognito switches as the draft's own send would read them. + */ + private async startOf(draft: Draft): Promise | null> { + const s = this.ctx.store; + const projectId = draft.projectId ?? await this.ctx.projects.ensureDrafts({ quiet: true }); + if (projectId === null) return null; + return { projectId, worktree: draft.worktree, incognito: !!draft.incognito && projectId === s.draftsProject?.id }; + } + + /** + * The new conversation takes the draft's queue as its outbox. The draft goes + * unless text waits in its box; the one on screen gives way to the conversation. + */ + private adoptLaunched(key: string, input: Input, threadId: ThreadId): void { + const s = this.ctx.store; + const state = (this.ctx.composer.composerStates[threadId] ??= { text: '', attachments: [], queued: [], sending: false, paused: false }); + // A draft started again after a crash finds its prompts there already: the same ids, once. + const moved = input.queued.splice(0) + .filter(entry => !state.queued.some(queued => queued.request && queued.request.id === entry.request?.id)) + .map(entry => { + if (!entry.request?.start) return entry; + const { start: _start, ...request } = entry.request; + return { ...entry, request }; + }); + state.queued.push(...moved); + // The thread's outbox is written before the draft goes: a crash between the two leaves the + // prompt in both, which the core takes once under its ids, never in neither. + this.ctx.drafts.persist(threadId); + const viewing = !!s.draft && draftKey(s.draft) === key && !s.openThread; + const kept = !!(input.text || input.attachments.length || input.previewReferences?.length); + if (!kept) this.ctx.drafts.discard(key, { started: true }); + this.ctx.drafts.persist(); void this.ctx.drafts.flush(); + if (viewing && !kept) void s.open(threadId); + } +} diff --git a/packages/ui/src/lib/store/drafts.svelte.ts b/packages/ui/src/lib/store/drafts.svelte.ts index b98b65761..8ed9a39e4 100644 --- a/packages/ui/src/lib/store/drafts.svelte.ts +++ b/packages/ui/src/lib/store/drafts.svelte.ts @@ -8,7 +8,7 @@ import { readDraftJournal, writeDraftJournal } from '../draft-journal'; import { unresolvedAssetId, type DraftAttachment } from '../draft-attachments'; import { secureId } from '../secure-id'; -type Input = Composer['composerStates'][string]; +export type Input = Composer['composerStates'][string]; /** * `parked` orders the drafts a project resumes: the most recently set aside * wins. It is a time from `created`, so the order survives a reload. @@ -18,6 +18,12 @@ type Journal = { updatedAt?: number; inputs?: Record; drafts?: type Source = { section: 'inputs' | 'drafts'; id: string; input: Input; draft?: Draft; parked?: number }; type CachedEntry = { section: Source['section']; id: string; value: unknown; serialized: string; updatedAt: number; backedUp: boolean }; const entryKey = (section: Source['section'], id: string) => JSON.stringify([section, id]); +/** A draft's row: its words, and whether it was sent while its machine was away and waits for it, or was refused. */ +export type DraftEntry = { + id: string; projectId: string | null; text: string; active: boolean; queued: 'waiting' | 'starting' | { failed: string } | null; + /** Its start reached the core once: it may hold the conversation, so the start is retried, never taken back. */ + asked?: true; +}; const empty = (): Input => ({ text: '', attachments: [], queued: [], sending: false, paused: false }); const hasText = (input: Input | undefined) => !!(input?.text || input?.queued.length); const hasContent = (input: Input | undefined) => hasText(input) || !!(input?.attachments.length || input?.previewReferences?.length); @@ -297,9 +303,35 @@ export class Drafts { for (const [id, entry] of Object.entries(this.saved)) if (entry.draft.projectId === projectId) delete this.saved[id]; } - /** Drops a draft. The open one empties and stays open, in the same project. */ - discard(id: string): void { + /** + * Takes back the prompts a draft queued while its machine was away: the + * conversation will not start by itself. Their words return to the box, + * after what it holds, so nothing typed is lost. A start already on its way, + * or asked once (`start`), is not taken back: sent again under a new id, the + * same words could make a second conversation beside one the core made. + */ + unqueue(id: string): void { const s = this.ctx.store; + const input = s.draft && draftKey(s.draft) === id ? this.ctx.composer.composerStates.draft : this.saved[id]?.input; + if (!input || input.sending || !input.queued.length || input.queued.some(entry => entry.request?.start)) return; + const taken = input.queued.splice(0); + input.text = [input.text, ...taken.map(entry => entry.text)].filter(text => text.length > 0).join('\n\n'); + input.attachments = [...input.attachments, ...taken.flatMap(entry => entry.attachments)]; + const references = [...(input.previewReferences ?? []), ...taken.flatMap(entry => entry.previewReferences ?? [])]; + if (references.length) input.previewReferences = references; + input.paused = false; + this.persist(); + } + + /** + * Drops a draft. The open one empties and stays open, in the same project. + * One whose conversation is being started stays: the core may already hold + * it, and its prompt goes there (`started` is that start handing it over). + */ + discard(id: string, options: { started?: boolean } = {}): void { + const s = this.ctx.store; + const input = s.draft && draftKey(s.draft) === id ? this.ctx.composer.composerStates.draft : this.saved[id]?.input; + if (!options.started && input?.sending && input.queued[0]?.request) return; delete this.saved[id]; if (s.draft && draftKey(s.draft) === id) { this.ctx.composer.composerStates.draft = empty(); @@ -319,20 +351,28 @@ export class Drafts { /** * Every draft with something in it, newest first by creation, so typing * never reorders the list. A draft with no project names the drafts project - * once it exists; before that its project is null. + * once it exists; before that its project is null. A draft sent while its + * machine was away says so (`queued`) and shows that prompt until its box + * holds text of its own. */ - get entries(): { id: string; projectId: string | null; text: string; active: boolean }[] { + get entries(): DraftEntry[] { const s = this.ctx.store; - const rows = new Map(); + const rows = new Map(); + const row = (id: string, projectId: string | null, input: Input, active: boolean, createdAt: number) => { + const head = input.queued[0]?.request ? input.queued[0] : undefined; + return { id, projectId, text: input.text || head?.text || '', active, createdAt, + queued: head?.request?.failed !== undefined ? { failed: head.request.failed } : head && input.sending ? 'starting' as const : head ? 'waiting' as const : null, + ...(head?.request?.start ? { asked: true as const } : {}) }; + }; for (const [id, entry] of Object.entries(this.saved)) { - if (hasContent(entry.input)) rows.set(id, { id, projectId: entry.draft.projectId, text: entry.input.text, active: false, createdAt: entry.draft.createdAt ?? 0 }); + if (hasContent(entry.input)) rows.set(id, row(id, entry.draft.projectId, entry.input, false, entry.draft.createdAt ?? 0)); } if (s.draft) { // The open draft gets a row only once it holds something: an empty one is just the composer. const id = draftKey(s.draft); const input = this.ctx.composer.composerStates.draft; - if (!hasContent(input)) rows.delete(id); - else rows.set(id, { id, projectId: s.draft.projectId, text: input?.text ?? '', active: true, createdAt: s.draft.createdAt ?? 0 }); + if (!input || !hasContent(input)) rows.delete(id); + else rows.set(id, row(id, s.draft.projectId, input, true, s.draft.createdAt ?? 0)); } // A draft with no project is sent to the drafts, so it is listed there once they exist. const drafts = s.draftsProject?.id ?? null; @@ -539,7 +579,11 @@ function outboxRequest(value: unknown): OutboxRequest | undefined { const raw = value as Partial; if (typeof raw.id !== 'string' || !/^[A-Za-z0-9_-]{8,128}$/.test(raw.id)) return undefined; const choice = raw.choice && typeof raw.choice === 'object' && typeof raw.choice.providerId === 'string' && typeof raw.choice.accountId === 'string' ? raw.choice : null; - return { id: raw.id, choice, queuedAt: typeof raw.queuedAt === 'number' ? raw.queuedAt : 0, ...(typeof raw.failed === 'string' ? { failed: raw.failed } : {}), ...(raw.sent === true ? { sent: true as const } : {}) }; + const start = raw.start && typeof raw.start === 'object' && typeof raw.start.projectId === 'string' && raw.start.projectId.length > 0 + && typeof raw.start.worktree === 'boolean' && typeof raw.start.incognito === 'boolean' + ? { projectId: raw.start.projectId, worktree: raw.start.worktree, incognito: raw.start.incognito } : null; + return { id: raw.id, choice, queuedAt: typeof raw.queuedAt === 'number' ? raw.queuedAt : 0, ...(typeof raw.failed === 'string' ? { failed: raw.failed } : {}), + ...(raw.sent === true ? { sent: true as const } : {}), ...(start ? { start } : {}) }; } function references(value: unknown): PreviewReference[] { diff --git a/packages/ui/src/lib/store/projects.svelte.ts b/packages/ui/src/lib/store/projects.svelte.ts index dd790d312..11752e778 100644 --- a/packages/ui/src/lib/store/projects.svelte.ts +++ b/packages/ui/src/lib/store/projects.svelte.ts @@ -184,7 +184,8 @@ export class Projects { * The drafts project, asked of the core on the first send into it. The core * makes the folder then, and the thread the send creates goes in it. */ - async ensureDrafts(): Promise { + /** `quiet`: a background start that tries again later, so a failure is not the user's to read. */ + async ensureDrafts(options: { quiet?: boolean } = {}): Promise { const known = this.ctx.store.draftsProject; if (known) return known.id; const client = this.ctx.client; @@ -197,7 +198,7 @@ export class Projects { return project.id; } catch (error) { // A switched machine closed the old client: its rejection is not this machine's error. - if (this.ctx.currentClient(client, clientGeneration)) this.ctx.fail(error); + if (this.ctx.currentClient(client, clientGeneration) && !options.quiet) this.ctx.fail(error); return null; } } diff --git a/packages/ui/src/lib/store/threads.svelte.ts b/packages/ui/src/lib/store/threads.svelte.ts index 7963354ce..5d7cd4c93 100644 --- a/packages/ui/src/lib/store/threads.svelte.ts +++ b/packages/ui/src/lib/store/threads.svelte.ts @@ -23,7 +23,8 @@ import { rightPanel } from '../right-panel.svelte'; import { workingChildren, type WorkingChildren, lastIndexById, mergeResumed, patchFields, patchRow, reconcileThread, threadsByProject } from '../thread-rows'; import type { StoreContext } from './context'; import { RpcErrorCode, threadActive } from '@boite/contracts'; -import { RpcFailure } from '../client'; +import { RpcFailure, wasDropped } from '../client'; +import { wasClosed } from '../rpc-error'; import { strings } from '../strings'; import { archiveWhile, keepArchived, pinWhile } from './optimistic'; @@ -419,7 +420,8 @@ export class Threads { this.readingThreads.delete(threadId); this.readingPositions.delete(threadId); } - this.ctx.fail(error); + // The machine went while the thread opened: it shows offline, and the reconnect opens it again. + if (!((wasDropped(error) || wasClosed(error)) && s.connection !== 'ready')) this.ctx.fail(error); } await retire(); } finally { diff --git a/packages/ui/src/lib/strings/en/composer.ts b/packages/ui/src/lib/strings/en/composer.ts index fcff6725e..22bc0f640 100644 --- a/packages/ui/src/lib/strings/en/composer.ts +++ b/packages/ui/src/lib/strings/en/composer.ts @@ -27,6 +27,7 @@ export const strings = { /** A new conversation, before anything is sent. */ placeholderNew: 'What do you want to do?', placeholderOffline: '{machine} is offline: messages wait here and go out when it reconnects', + placeholderOfflineDraft: '{machine} is offline: the conversation starts when it reconnects', send: 'Send', stop: 'Stop', queued: 'Sent after the next tool, or when the turn ends', @@ -40,6 +41,7 @@ export const strings = { retryQueuedHint: 'Sends all pending messages together again. Enter in the empty composer does the same.', /** The outbox: messages written while the machine was away, kept on this device. */ outboxKept: 'Kept on this device and sent in order, once each', + outboxStarts: 'Kept on this device: the conversation starts once {machine} is back', outboxWaiting: 'Waiting for {machine}', outboxNext: 'Goes out when the thread is free', outboxSending: 'Sending…', diff --git a/packages/ui/src/lib/strings/en/sidebar.ts b/packages/ui/src/lib/strings/en/sidebar.ts index da9603bda..6d7f9ccef 100644 --- a/packages/ui/src/lib/strings/en/sidebar.ts +++ b/packages/ui/src/lib/strings/en/sidebar.ts @@ -161,6 +161,12 @@ export const strings = { undoHint: 'Undo (Ctrl+Z)', draft: 'New thread', /** A draft row's menu: what is typed there and not sent goes. */ - deleteDraft: 'Delete draft' + deleteDraft: 'Delete draft', + /** A draft sent while its machine was away: its conversation starts once the machine is back. */ + draftQueued: 'Starts when {machine} is back', + draftStarting: 'Starting on {machine}', + draftQueueFailed: 'Not started: {reason}', + /** Its menu takes the queued prompt back into the draft, which then waits to be sent again. */ + cancelQueuedDraft: 'Cancel the start' } } as const; diff --git a/packages/ui/src/lib/strings/fr/composer.ts b/packages/ui/src/lib/strings/fr/composer.ts index 7e4602779..34594df4c 100644 --- a/packages/ui/src/lib/strings/fr/composer.ts +++ b/packages/ui/src/lib/strings/fr/composer.ts @@ -26,6 +26,7 @@ export const fr: Translation = { placeholderNoProject: "Message à l'agent", placeholderNew: 'Que voulez-vous faire ?', placeholderOffline: '{machine} est hors ligne : les messages attendent ici et partent à sa reconnexion', + placeholderOfflineDraft: '{machine} est hors ligne : la conversation démarre à sa reconnexion', send: 'Envoyer', stop: 'Arrêter', queued: 'Envoyé après le prochain outil, ou à la fin du tour', @@ -37,6 +38,7 @@ export const fr: Translation = { sendNowHint: "Transmet les messages à l'agent en cours sans l'arrêter. Entrée dans le champ vide fait de même.", retryQueuedHint: 'Renvoie tous les messages en attente ensemble. Entrée dans le champ vide fait de même.', outboxKept: "Gardés sur cet appareil, envoyés dans l'ordre, une seule fois chacun", + outboxStarts: 'Gardé sur cet appareil : la conversation démarre dès le retour de {machine}', outboxWaiting: 'En attente de {machine}', outboxNext: 'Part dès que le fil est libre', outboxSending: 'Envoi…', diff --git a/packages/ui/src/lib/strings/fr/sidebar.ts b/packages/ui/src/lib/strings/fr/sidebar.ts index d2f13a559..a67db7229 100644 --- a/packages/ui/src/lib/strings/fr/sidebar.ts +++ b/packages/ui/src/lib/strings/fr/sidebar.ts @@ -137,6 +137,10 @@ export const fr: Translation = { undo: 'Annuler', undoHint: 'Annuler (Ctrl+Z)', draft: 'Nouvelle conversation', - deleteDraft: 'Supprimer le brouillon' + deleteDraft: 'Supprimer le brouillon', + draftQueued: 'Démarre au retour de {machine}', + draftStarting: 'Démarrage sur {machine}', + draftQueueFailed: 'Non démarrée : {reason}', + cancelQueuedDraft: 'Annuler le démarrage' } }; diff --git a/packages/ui/src/lib/ws-client-utils.ts b/packages/ui/src/lib/ws-client-utils.ts index 66d1dca43..34ab545af 100644 --- a/packages/ui/src/lib/ws-client-utils.ts +++ b/packages/ui/src/lib/ws-client-utils.ts @@ -35,13 +35,32 @@ export function rpcUrl(coreUrl: string): string { */ export const IDLE_PROBE_MS = 25_000; /** - * How long a probe waits for any frame at all. The idle probe only goes out - * while no call waits for its answer: frames arrive in order, so behind a large - * answer still downloading on a slow link, the probe's answer would come too - * late and a healthy socket would read as dead. A call's own 120 s timeout - * decides for a socket that carries one. + * How long a probe waits for any frame at all, the same 10 s a new socket gets + * to say hello. Frames arrive in order, so a probe's answer queues behind a long + * answer still downloading; a remote client asks the core to slice every answer + * (`progress`), and each 64 KiB slice counts as a frame, so a slow link keeps + * answering while a dead one stays silent. Until 2026-10-22 a call waiting kept + * the probe away and a link that dropped under an open conversation read as + * connected for the two minutes of that call's timeout. */ -export const PROBE_DEADLINE_MS = 15_000; +export const PROBE_DEADLINE_MS = 10_000; +/** + * A call on a remote socket to a core that slices its answers + * (`chunkedAnswers`) that hears no frame at all for this long sends a probe: + * the click on a conversation of a machine whose link died shows the machine + * offline within this and the probe's deadline. A socket that answers, or + * carries any event meanwhile, is not asked. Older cores keep the call's own + * 120 s timeout. + */ +export const QUIET_CALL_MS = 3_000; +/** + * The slowest upload a probe still waits for: its `hello` queues behind the + * bytes this client sent since the last frame came back, past the first 64 KiB, + * so 2 MiB of pictures give it 62 s more. Answers need no such allowance: they come in 64 KiB + * slices, so a download faster than about 6.4 KiB/s lands one within the 10 s + * deadline. + */ +export const UPLOAD_FLOOR_BYTES_PER_S = 32 * 1024; /** How long `resume()` gives the socket it already has before replacing it. */ export const RESUME_PROBE_MS = 4_000; diff --git a/packages/ui/src/lib/ws-client.ts b/packages/ui/src/lib/ws-client.ts index 53a79a37a..5907fca20 100644 --- a/packages/ui/src/lib/ws-client.ts +++ b/packages/ui/src/lib/ws-client.ts @@ -3,7 +3,7 @@ import { secureId } from './secure-id'; import { RELEASE_VERSION } from './release'; import { signedUngrouped, type DeviceKey } from './device-key'; import { - PROTOCOL_VERSION, RPC_MAX_FRAME_BYTES, RpcCloseCode, RpcErrorCode, + PROTOCOL_VERSION, RPC_CHUNK_BYTES, RPC_MAX_FRAME_BYTES, RpcCloseCode, RpcErrorCode, type ClientName, type CoreInfo, type DeviceHello, type Principal, type UngroupedData, type RpcEventName, type RpcMethodName, type RpcParams, type RpcResult, type ThreadId } from '@boite/contracts'; @@ -11,7 +11,7 @@ import type { CallOptions, ClientState, EventHandler, ObservableClient, Session, import { RpcFailure, closedFailure, droppedFailure, transportFailure, unansweredFailure } from './rpc-error'; import { receiveFrame, type Pending } from './ws-client-frames'; import { - IDLE_PROBE_MS, PROBE_DEADLINE_MS, RESUME_PROBE_MS, MESSAGE_TOO_BIG, + IDLE_PROBE_MS, PROBE_DEADLINE_MS, QUIET_CALL_MS, RESUME_PROBE_MS, MESSAGE_TOO_BIG, UPLOAD_FLOOR_BYTES_PER_S, browserSocket, defaultBackoff, frameBytes, isLoopbackUrl, megabytes, openDeadline, pageHidden, rpcUrl } from './ws-client-utils'; @@ -57,6 +57,10 @@ export class WsClient implements ObservableClient { /** Remote sockets are probed when silent; a loopback one never goes half-open. */ #remote: boolean; #lastFrameAt = 0; + /** Frames received on this client: a call compares it with the count it was sent at, two of which may share a millisecond. */ + #framesReceived = 0; + /** Bytes sent since the last frame arrived: the core answers nothing before it has read them. */ + #sentSinceFrame = 0; #liveness: ReturnType | null = null; /** The one probe in flight: settled true by any frame, false by its deadline or a teardown. */ #probe: { socket: SocketLike; result: Promise; settle: (alive: boolean) => void } | null = null; @@ -222,8 +226,11 @@ export class WsClient implements ObservableClient { params: RpcParams, onProgress?: (received: number, total: number) => void ): Promise> { - // Asking for progress is what makes the core slice a long answer. - const frame = JSON.stringify({ jsonrpc: '2.0', id: this.#nextId, method, params, ...(onProgress ? { progress: true } : {}) }); + // Asking for progress is what makes the core slice a long answer. A remote socket asks it of + // every call: slices keep frames arriving while a long answer downloads, so silence there means + // a dead link and the liveness probe may ask whatever waits. + const sliced = onProgress !== undefined || (this.#remote && method !== 'hello'); + const frame = JSON.stringify({ jsonrpc: '2.0', id: this.#nextId, method, params, ...(sliced ? { progress: true } : {}) }); const size = frameBytes(frame); if (size > RPC_MAX_FRAME_BYTES) { // The core would close the socket on it before reading a byte, and every @@ -235,7 +242,14 @@ export class WsClient implements ObservableClient { })); } const id = this.#nextId++; + const heard = this.#framesReceived; return new Promise>((resolve, reject) => { + // A call that hears nothing at all on a remote socket asks whether the machine is still there, + // so a click on a conversation of a machine whose link died gets its offline state within the + // probe's deadline rather than once the call's two minutes are spent. + const watch = this.#remote && method !== 'hello' && this.#sliced() + ? setTimeout(() => { if (this.#framesReceived === heard) this.#suspect(socket); }, QUIET_CALL_MS) + : null; const expire = () => { this.#pending.delete(id); reject(unansweredFailure(method)); @@ -247,13 +261,13 @@ export class WsClient implements ObservableClient { let timer = setTimeout(expire, wait); const pending: Pending = { method, - resolve: (value) => { clearTimeout(timer); resolve(value as RpcResult); }, - reject: (error) => { clearTimeout(timer); reject(error); }, + resolve: (value) => { clearTimeout(timer); if (watch !== null) clearTimeout(watch); resolve(value as RpcResult); }, + reject: (error) => { clearTimeout(timer); if (watch !== null) clearTimeout(watch); reject(error); }, touch: () => { clearTimeout(timer); timer = setTimeout(expire, wait); }, ...(onProgress ? { onProgress } : {}) }; this.#pending.set(id, pending); - try { socket.send(frame); } + try { socket.send(frame); this.#sentSinceFrame += size; } catch (error) { this.#pending.delete(id); pending.reject(error instanceof Error ? error : transportFailure(String(error))); @@ -263,6 +277,7 @@ export class WsClient implements ObservableClient { #open(): Promise { this.#setState('connecting'); + this.#sentSinceFrame = 0; const socket = this.#options.socketFactory(rpcUrl(this.#options.url)); this.#socket = socket; @@ -442,8 +457,11 @@ export class WsClient implements ObservableClient { #probeSocket(socket: SocketLike, deadlineMs: number): Promise { if (this.#probe?.socket === socket) return this.#probe.result; let settle: (alive: boolean) => void = () => undefined; + // The hello waits behind what this client still uploads, a picture in a prompt on a slow uplink: + // the deadline grows by the time those bytes take at the slowest rate a link is still trusted with. + const behind = Math.ceil((Math.max(0, this.#sentSinceFrame - RPC_CHUNK_BYTES) / UPLOAD_FLOOR_BYTES_PER_S) * 1000); const result = new Promise((resolve) => { - const timer = setTimeout(() => settle(false), deadlineMs); + const timer = setTimeout(() => settle(false), deadlineMs + behind); settle = (alive) => { clearTimeout(timer); if (this.#probe?.socket === socket) this.#probe = null; @@ -486,9 +504,9 @@ export class WsClient implements ObservableClient { if (pageHidden()) { this.#armLiveness(IDLE_PROBE_MS); return; } const quiet = Date.now() - this.#lastFrameAt; if (quiet < IDLE_PROBE_MS) { this.#armLiveness(IDLE_PROBE_MS - quiet); return; } - // Silence while a call waits may be its answer still arriving: that - // call's timeout asks the socket instead. - if (this.#awaitingAnswer()) { this.#armLiveness(IDLE_PROBE_MS); return; } + // A core that answers in one frame may still be downloading the answer a call waits for: + // that call's own timeout asks the socket instead. + if (!this.#sliced() && this.#awaitingAnswer()) { this.#armLiveness(IDLE_PROBE_MS); return; } void this.#probeSocket(socket, PROBE_DEADLINE_MS).then((alive) => { if (alive) this.#armLiveness(IDLE_PROBE_MS); else this.#lost(socket); @@ -502,6 +520,11 @@ export class WsClient implements ObservableClient { return false; } + /** The core slices every long answer this client asks for (`features.chunkedAnswers`): silence is a dead link, never a frame still downloading. */ + #sliced(): boolean { + return this.#core?.features?.chunkedAnswers === true; + } + #stopLiveness(): void { if (this.#liveness !== null) clearTimeout(this.#liveness); this.#liveness = null; @@ -554,6 +577,8 @@ export class WsClient implements ObservableClient { #receive(raw: unknown): void { this.#lastFrameAt = Date.now(); + this.#framesReceived += 1; + this.#sentSinceFrame = 0; this.#probe?.settle(true); receiveFrame(raw, this.#pending, this.#handlers, (whole) => this.#receive(whole)); } diff --git a/tests/e2e/machine-outage.test.ts b/tests/e2e/machine-outage.test.ts new file mode 100644 index 000000000..49fdcbe22 --- /dev/null +++ b/tests/e2e/machine-outage.test.ts @@ -0,0 +1,169 @@ +import { afterAll, beforeAll, expect, test } from 'bun:test'; +import { createServer, connect as tcpConnect, type Server, type Socket } from 'node:net'; +import { join } from 'node:path'; +import { BrowserPage, freePort } from './lib/cdp.ts'; +import { startUi } from './lib/ui.ts'; +import { startCore, type RunningCore } from './lib/core.ts'; +import { connect } from '../../packages/core/src/client.ts'; + +let server: { close(): Promise }; +let page: BrowserPage; +let url: string; +const cores: RunningCore[] = []; +const id = (name: string) => `[data-testid="${name}"]`; +async function capture(name: string) { + await page.settleAnimations(); + await page.screenshot(join(import.meta.dir, '.artifacts', name)); +} + +/** + * Another loopback address: the page holds only 127.0.0.1, localhost and ::1 for + * this computer's own core, so a core behind this one is a remote machine to it. + */ +const REMOTE_HOST = '127.0.0.2'; + +/** + * A TCP relay in front of a core that can go silent the way a dropped VPN does: + * frozen, it keeps every socket open and delivers nothing either way, and a new + * connection is accepted and never answered. Thawed, the stale sockets are reset. + */ +async function blackhole(target: number) { + const pairs = new Set<{ client: Socket; upstream: Socket | null }>(); + let frozen = false; + const relay: Server = createServer(client => { + const pair: { client: Socket; upstream: Socket | null } = { client, upstream: null }; + pairs.add(pair); + client.on('error', () => undefined); + client.on('close', () => { pair.upstream?.destroy(); pairs.delete(pair); }); + if (frozen) return; + const upstream = tcpConnect(target, '127.0.0.1'); + pair.upstream = upstream; + upstream.on('error', () => undefined); + upstream.on('close', () => client.destroy()); + client.on('data', chunk => { if (!frozen) upstream.write(chunk); }); + upstream.on('data', chunk => { if (!frozen) client.write(chunk); }); + }); + const host = REMOTE_HOST; + await new Promise(resolve => relay.listen(0, host, resolve)); + const port = (relay.address() as { port: number }).port; + return { + url: `http://${host}:${port}`, + freeze() { frozen = true; }, + thaw() { + frozen = false; + for (const pair of [...pairs]) { pair.client.destroy(); pair.upstream?.destroy(); } + }, + close: () => new Promise(resolve => { for (const pair of pairs) { pair.client.destroy(); pair.upstream?.destroy(); } relay.close(() => resolve()); }) + }; +} + +beforeAll(async () => { + const port = await freePort(); + server = await startUi(port); + url = `http://127.0.0.1:${port}`; + page = await BrowserPage.launch({ url: `${url}/?fake=1`, args: [`--host-resolver-rules=MAP * ~NOTFOUND, EXCLUDE 127.0.0.1, EXCLUDE localhost, EXCLUDE ${REMOTE_HOST}`] }); +}, 30_000); +afterAll(async () => { + await page?.close(); + await server?.close(); + for (const core of cores) await core.stop(); +}, 30_000); + +/** + * The desktop of 2026-10-22: the VPN to the server dropped under an open + * conversation. Every call the window made there waited its two minutes while + * the machine still read as connected, a red "connection lost" stayed on + * screen, and a new conversation on that machine could not be sent at all. + */ +test('a machine whose link goes silent reads offline within seconds, and a conversation started there waits for it', async () => { + const first = await startCore(), second = await startCore(); + cores.push(first, second); + const hole = await blackhole(second.port); + const a = await connect(first.url, first.token), b = await connect(second.url, second.token); + try { + await Promise.all([a.call('settings.set', { browserOrigins: [url] }), b.call('settings.set', { browserOrigins: [url] })]); + const create = async (client: typeof a, path: string, name: string) => { + const project = await client.call('projects.add', { path, name }); + const account = (await client.call('accounts.list', {})).find(entry => entry.providerId === 'echo')!; + return client.call('threads.create', { projectId: project.id, providerId: 'echo', accountId: account.id, title: name }); + }; + const ta = await create(a, first.dataDir, 'Desk project'); + const tb = await create(b, second.dataDir, 'Server project'); + await page.navigate(`${url}/?core=${encodeURIComponent(first.url)}&token=${encodeURIComponent(first.token)}`); + await page.waitFor(`document.querySelector('${id('confirm-ok')}')`); + await page.click(id('confirm-ok')); + await page.waitFor(`document.querySelector('[data-thread-id="${ta.id}"]')`); + await page.evaluate(`globalThis.__boiteTest.workspace.add({ url: ${JSON.stringify(hole.url)}, token: ${JSON.stringify(second.token)} }, 'Server')`); + await page.waitFor(`document.querySelector('[data-thread-id="${tb.id}"]')`); + await page.click(`[data-thread-id="${ta.id}"]`); + await page.waitFor(`document.querySelector('${id('thread-title')}')?.textContent === 'Desk project'`); + const server = `globalThis.__boiteTest.workspace.machines.find(machine => machine.id === ${JSON.stringify(hole.url)}).store`; + + // The link goes silent the moment before the user opens a conversation there. + hole.freeze(); + const clicked = Date.now(); + await page.click(`[data-thread-id="${tb.id}"]`); + await page.waitFor(`${server}.connection !== 'ready'`, 16_000); + expect(Date.now() - clicked).toBeLessThan(16_000); + await page.waitFor(`document.querySelector('${id('composer-input')}')?.getAttribute('placeholder')?.startsWith('Server is offline')`); + // Its status says it: no error shows over the window, now or in the seconds after. + await page.evaluate(`(() => { + globalThis.__toasts = []; + const seen = () => { const toast = document.querySelector('${id('error-toast')}'); if (toast) globalThis.__toasts.push(toast.textContent); }; + seen(); + globalThis.__toastWatch = new MutationObserver(seen); + globalThis.__toastWatch.observe(document.body, { childList: true, subtree: true }); + })()`); + await Bun.sleep(6_000); + expect(await page.evaluate(`(() => { globalThis.__toastWatch.disconnect(); return globalThis.__toasts; })()`)).toEqual([]); + await capture('outage-offline-thread-desktop.png'); + + // The other machine is untouched. + await page.click(`[data-thread-id="${ta.id}"]`); + await page.waitFor(`document.querySelector('${id('thread-title')}')?.textContent === 'Desk project'`); + await page.type(id('composer-input'), 'Local while the server is away'); + await page.click(id('composer-send')); + await page.waitFor(`document.querySelector('${id('chat')}')?.textContent.includes('Local while the server is away')`); + + // A new conversation on the server: Send queues it as a draft that waits for the machine. + await page.evaluate(`globalThis.__boiteTest.workspace.select(${server}, undefined, ${JSON.stringify(tb.projectId)}, { fresh: true })`); + await page.waitFor(`document.querySelector('${id('composer-input')}')?.getAttribute('placeholder') === 'Server is offline: the conversation starts when it reconnects'`); + await page.type(id('composer-input'), 'Start this when the server is back'); + await page.waitFor(`document.querySelector('${id('composer-send')}')?.disabled === false`); + await page.click(id('composer-send')); + const row = `[data-testid="draft-row"][data-queued="waiting"]`; + await page.waitFor(`document.querySelector('${row}')?.textContent.includes('Start this when the server is back')`); + await page.waitFor(`document.querySelector('${id('composer-outbox-pending')}')?.textContent.includes('Waiting for Server')`); + expect(await page.evaluate(`document.querySelector('${id('composer-input')}').value`)).toBe(''); + await capture('outage-queued-draft-desktop.png'); + await page.send('Emulation.setDeviceMetricsOverride', { width: 390, height: 844, deviceScaleFactor: 1, mobile: true }); + await capture('outage-queued-draft-phone.png'); + expect(await page.evaluate('document.documentElement.scrollWidth <= innerWidth')).toBe(true); + await page.send('Emulation.clearDeviceMetricsOverride', {}); + + // Its row's menu takes the start back: the words return to the box and nothing waits any more. + await page.evaluate(`document.querySelector('${row}').dispatchEvent(new MouseEvent('contextmenu', { bubbles: true, clientX: 100, clientY: 180 }))`); + const cancel = `${id('context-menu')} [data-value=unqueue]`; + await page.waitFor(`document.querySelector('${cancel}')`); + await capture('outage-queued-draft-menu.png'); + await page.click(cancel); + await page.waitFor(`!document.querySelector('${row}') && document.querySelector('[data-testid="draft-row"]')?.textContent.includes('Start this when the server is back')`); + expect(await page.evaluate(`document.querySelector('${id('composer-input')}').value`)).toBe('Start this when the server is back'); + await page.click(id('composer-send')); + await page.waitFor(`document.querySelector('${row}')`); + + // The machine returns: the conversation starts there once, with its prompt, and takes the draft's place. + hole.thaw(); + // The echo agent may retitle it from the start of the prompt ("Echo: Start this when the server"). + await page.waitFor(`document.querySelector('${id('thread-title')}')?.textContent.includes('Start this when the server')`, 30_000); + await page.waitFor(`document.querySelector('${id('chat')}')?.textContent.includes('Start this when the server is back') && !document.querySelector('${id('composer-queued')}')`); + expect(await page.evaluate(`document.querySelector('[data-testid="draft-row"]')`)).toBe(null); + const started = (await b.call('threads.list', {})).filter(thread => thread.projectId === tb.projectId && thread.id !== tb.id); + expect(started).toHaveLength(1); + const thread = await b.call('threads.get', { threadId: started[0]!.id }); + expect(thread.messages.filter(message => message.role === 'user')).toHaveLength(1); + await capture('outage-started-desktop.png'); + } finally { + a.close(); b.close(); await hole.close(); + } +}, 120_000);