Repository navigation
feat(spike): durable multiview host, gateway and remote client (P0) - #426
Conversation
A headless Node host owns the Durable Harness, SQLite store, and lock, and serves conversation views, the task graph, and the conversation list over a 127.0.0.1 WebSocket gateway as snapshots plus chord ops. RemoteDurable implements the upstream DurableViewSource/DurableController over it and reconnects from fresh snapshots after a host restart. Includes a line-mode observer and tests for two-client streaming and kill -9 recovery.
bubbleptr
changed the title
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| // Unsubscribed, or the socket went away, while the watch was being acquired. | ||
| if (this.#disposed || this.#subscriptions.get(stream) !== "pending") { | ||
| await subscription.stop(); | ||
| return; | ||
| } | ||
| this.#subscriptions.set(stream, subscription); |
There was a problem hiding this comment.
🔴 Overlapping subscriptions corrupt stream state
When a stream is resubscribed before its first watch opens, #subscribe can install the older watch. Both opens send snapshots, so the older watch's ops can apply to the newer snapshot and corrupt the view.
Learn more
The gateway handles incoming messages concurrently because the socket message listener does not await earlier calls to #receive (line 183). Each #subscribe uses the same string, "pending", as its identity. A later subscription clears the earlier pending value and installs its own; the earlier opener can then claim that later subscription's slot. Both #open calls send snapshots before the ownership check, and only one watch remains active. A delta from that watch can be applied to the other snapshot.
Example: A client sends two subscribe frames for conversation:0 in quick succession. Request A pauses acquiring its watch. Request B marks the stream pending. A finishes, sends snapshot A, and claims B's slot; B sends snapshot B but stops its watch. The client applies A's later ops to snapshot B.
Recommended fix: Give each subscription attempt a unique token and verify it before sending any frames or starting the watch. Serialize subscription changes per stream, or arrange for only the current attempt to publish a snapshot and ops.
Was this helpful? React with 👍 or 👎 to provide feedback.
| #lost(socket: WebSocket, code: number): void { | ||
| if (this.#socket !== socket) return; | ||
| this.#socket = undefined; | ||
| for (const pending of this.#pending.values()) pending.reject(new Error("Disconnected from host")); | ||
| this.#pending.clear(); |
There was a problem hiding this comment.
🟡 Lost snapshots stall controller commands
If the connection drops while toggleTasks or switchConversation awaits a snapshot, #lost leaves that waiter pending. The serialized command queue then blocks later commands until that stream receives another snapshot, potentially forever.
Learn more
A task-panel toggle and a conversation switch each add a waiter through #snapshot and await it inside the serialized #command chain. The disconnect handler rejects pending RPC calls but never rejects snapshot waiters. Until a snapshot arrives for that exact stream, the operation and every later queued command stay pending. This is permanent when the connection is closed or the host does not return.
Example: The client sends a task-graph subscription, then the host exits before returning its snapshot. The panel toggle promise never settles. Clicking another model or submitting text queues behind the toggle and never sends a call.
Recommended fix: Reject and clear all snapshot waiters on disconnect and on explicit close. Let the command chain handle these rejections, and register new waiters only when resubscribing on the new socket.
Was this helpful? React with 👍 or 👎 to provide feedback.
| if (this.#state !== undefined) this.#update({ session: frame.session, models: frame.models, connection: "connected" }); | ||
| else this.#awaitFirstView(frame); |
There was a problem hiding this comment.
🟡 Reconnect exposes stale conversation as connected
On reconnect, hello sets connection to connected before fresh snapshots arrive. Consumers can treat the previous host's conversation as recovered and act on stale state.
Learn more
The previous conversation, conversation list and task graph remain in #values during a disconnect. The new host sends hello before the subscriptions return their snapshots. Marking the view connected on hello therefore announces recovery while the visible conversation still comes from the previous host, including stale streaming and busy state.
Example: The host dies during an answer; the client retains an unfinished partial. The restarted host sends hello and the view becomes connected while still showing that old partial, before its recovery snapshot replaces it.
Recommended fix: Keep connection as reconnecting until fresh snapshots for the selected conversation and required streams have arrived. Clear old stream values or track a connection generation to prevent old snapshots and ops from satisfying the new connection's readiness.
Was this helpful? React with 👍 or 👎 to provide feedback.
| for (const notice of view.notices.slice(shown === undefined ? 0 : shown.notices.length)) { | ||
| process.stdout.write(`\n[${notice.level}] ${notice.message}\n`); | ||
| } |
There was a problem hiding this comment.
🟡 Observer hides notices after twenty
After twenty notices, render slices each new notice list at index twenty. The client retains only twenty notices, so subsequent warnings and errors never print.
Learn more
The client truncates its notice array to the most recent twenty entries in #notice. The observer compares the new array against the old array's length rather than tracking notice IDs. Once both arrays contain twenty items, its slice is empty even though old notices have been evicted and new ones appended.
Example: After twenty informational notices, the host emits an error. The client's notice array now contains entries 2–21 instead of 1–20, but the observer slices from index 20 and prints nothing.
Recommended fix: Track the last printed notice ID and filter for notices with a higher ID; preserve a sensible indication if enough notices were skipped to overflow the cap.
Was this helpful? React with 👍 or 👎 to provide feedback.
| const partial = streamingText(view); | ||
| if (partial !== undefined && partial !== "") { | ||
| if (streamed === undefined) output += `assistant: ${partial}`; | ||
| else if (partial.startsWith(streamed)) output += partial.slice(streamed.length); | ||
| else output += `\nassistant: ${partial}`; | ||
| streamed = partial; | ||
| } |
There was a problem hiding this comment.
🟡 Abandoned responses merge with later messages
If an in-flight response disappears without a committed assistant line, printIncrement keeps streamed and leaves its terminal line open. The next user message prints on that same line.
Learn more
The printer stores the visible in-flight response in streamed and writes it without a newline. It clears that value only while iterating over new transcript lines. When the in-flight response disappears without producing a matching assistant line, no branch closes the terminal line; a later user entry is appended to the open line.
Example: The terminal has printed assistant: working; the next view contains no live generation and no assistant entry. The next view adds a user entry retry. The output becomes assistant: workinguser: retry instead of separate lines.
Recommended fix: When streamingText(view) becomes absent, close any outstanding partial and clear streamed, unless a new committed assistant line already consumed it. Ensure any new non-assistant transcript entry starts on its own line.
Was this helpful? React with 👍 or 👎 to provide feedback.
| const options = values.faux === undefined ? await piOptions() : fauxOptions(values.faux); | ||
| const host = await openHost(options); | ||
| const model = options.initialModel === undefined ? null : `${options.initialModel.provider}/${options.initialModel.modelId}`; | ||
| console.log(JSON.stringify({ event: "ready", url: host.url, token: host.token, dataDir, model })); |
There was a problem hiding this comment.
🟥 Host logs expose the bearer token
On startup, main prints the reusable gateway token to stdout. Anyone with access to captured logs can connect and control the host.
Was this helpful? React with 👍 or 👎 to provide feedback.
| server.on("connection", (socket: WebSocket, request: IncomingMessage) => { | ||
| const token = new URL(request.url ?? "/", "ws://127.0.0.1").searchParams.get("token"); | ||
| if (token !== options.token) { |
There was a problem hiding this comment.
🟥 Gateway accepts untrusted browser origins
A malicious website can open the localhost WebSocket endpoint because the gateway never checks Origin. If the token is exposed, that page can read conversations and issue host commands.
Was this helpful? React with 👍 or 👎 to provide feedback.
| this.#socket = socket; | ||
| this.#options = options; | ||
| this.#conversations = conversations; | ||
| socket.on("message", (data) => void this.#receive(JSON.parse(String(data)) as ClientFrame)); |
There was a problem hiding this comment.
🟨 Malformed WebSocket frames can crash host
A connected client can send invalid JSON or malformed command fields to #receive. The uncaught error escapes the async message callback and can terminate the host.
Was this helpful? React with 👍 or 👎 to provide feedback.
No linked issue: first stage (P0) of the Pi Durable multi-client spike planned in
.scratch/durable-multiview/HANDOFF.md. The spike lives underspikes/and is deleted when it ends; no Pace app code changes.Summary
spikes/durable-multiview(@pace/durable-multiview-spike), with exact pins on@earendil-works/pi-durable,chord,pi-aiandpi-coding-agent1.0.0. The root workspaces addspikes/*and the root Vitest excludesspikes/**; the spike has its own Node-environmentvitest.config.tsandtsconfig.json, and runs from source on Node's type stripping.host/: a headless Node process that owns the Durable Harness, its SQLite store and aproper-lockfilelock (Durable has no cross-process lock). It serves a WebSocket gateway on127.0.0.1with a token that is persisted in the data directory, so clients that were connected before a restart are still accepted. Real mode uses Pi'sModelRuntime,~/.pi/agentcredentials and default model;--fauxuses pi-ai's scripted provider for tests. Pi's HTTP dispatcher is copied in with attribution, because it is outsidepi-coding-agent's package exports.Conversation.watch()), the task graph (watchTaskGraph()) and a commit-driven conversation list, each sent as a snapshot followed by chord ops. Each client's watch waits for the socket write before taking the next frame, so a slow client falls back on Durable's own 100-frame folding.protocol/:connectRemoteDurableimplements upstream'sDurableViewSource/DurableControllercontract over the gateway, with no Harness on the client. It adds one field,connection, and on reconnect it resubscribes and starts from fresh snapshots instead of replaying.submitcarries arequestIdso that a resend after a lost reply cannot admit the input twice.cli/observe.ts: a line-mode client that both watches the host and drives it.stopReason: "aborted"assistant entry (spec §8.1/§8.3) and then resends the request, so the transcript reads user → aborted partial → full answer. Every UI needs to render aborted entries differently.Test plan
bun run typecheckbun run lintbun run test: 167 files, 1970 tests (spike excluded)bun run buildbun run test(inspikes/durable-multiview): 7 tests; two clients see the same streamed answer, the wrong token is rejected, and the host iskill -9'd mid-stream and restarted with both clients resuming and identical transcripts (5 consecutive green runs)radius/deepseek-v4.1-flash)Checklist
bun run typecheck,bun run test, andbun run buildare greendocs/research/in P5)CONTEXT.mdupdated if vocabulary changed (none changed)shared/ui/components are registered on/design(none changed)feat/; this is not a push tomain