Repository navigation
feat(spike): durable multiview host, gateway and remote client (P0) #426
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Large diffs are not rendered by default.
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,58 @@ | ||
| #!/usr/bin/env node | ||
| // A line-mode client of a running host: prints the shown conversation as it | ||
| // streams, and sends each typed line. Run several to watch one host from many terminals. | ||
| // | ||
| // node cli/observe.ts [--data-dir DIR] [--url ws://127.0.0.1:7420] | ||
| // typed lines: text (prompt; follow-up while busy), /steer text, /abort, /tasks, /quit | ||
| import { readFile } from "node:fs/promises"; | ||
| import { join, resolve } from "node:path"; | ||
| import { createInterface } from "node:readline"; | ||
| import { parseArgs } from "node:util"; | ||
| import { getAgentDir } from "@earendil-works/pi-coding-agent"; | ||
| import { connectRemoteDurable } from "../protocol/remote-durable.ts"; | ||
| import type { DurableView } from "../protocol/view.ts"; | ||
| import { initialPrinter, printIncrement } from "./printer.ts"; | ||
|
|
||
| const { values } = parseArgs({ | ||
| options: { | ||
| "data-dir": { type: "string" }, | ||
| url: { type: "string", default: "ws://127.0.0.1:7420" }, | ||
| token: { type: "string" }, | ||
| }, | ||
| }); | ||
| const dataDir = resolve(values["data-dir"] ?? join(getAgentDir(), "experimental", "durable-multiview", "default")); | ||
| const token = values.token ?? (await readFile(join(dataDir, "token"), "utf8")).trim(); | ||
|
|
||
| const remote = await connectRemoteDurable({ url: values.url, token }); | ||
| let printer = initialPrinter; | ||
| let shown: DurableView | undefined; | ||
| const render = (): void => { | ||
| const view = remote.view.current(); | ||
| if (shown !== undefined && view.connection !== shown.connection) process.stdout.write(`\n[${view.connection}]\n`); | ||
| for (const notice of view.notices.slice(shown === undefined ? 0 : shown.notices.length)) { | ||
| process.stdout.write(`\n[${notice.level}] ${notice.message}\n`); | ||
| } | ||
| if (view.tasks !== shown?.tasks && view.tasks !== undefined) { | ||
| const nodes = Object.values(view.tasks.tasks).map((task) => `${task.id}:${task.kind}(${task.state.status})`); | ||
| process.stdout.write(`\n[tasks] ${nodes.length === 0 ? "none" : nodes.join(" ")}\n`); | ||
| } | ||
| shown = view; | ||
| const step = printIncrement(printer, view.conversation); | ||
| printer = step.state; | ||
| process.stdout.write(step.output); | ||
| }; | ||
| remote.view.subscribe(render); | ||
| process.stdout.write(`[connected to ${values.url}, session ${remote.view.current().session.id}]\n`); | ||
| render(); | ||
|
|
||
| const input = createInterface({ input: process.stdin }); | ||
| for await (const line of input) { | ||
| const text = line.trim(); | ||
| if (text === "") continue; | ||
| if (text === "/quit") break; | ||
| if (text === "/abort") await remote.controller.abort(); | ||
| else if (text === "/tasks") await remote.controller.toggleTasks(); | ||
| else if (text.startsWith("/steer ")) await remote.controller.submit(text.slice("/steer ".length), "steer"); | ||
| else await remote.controller.submit(text, "followUp"); | ||
| } | ||
| remote.close(); | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| import type { ConversationView } from "@earendil-works/pi-durable"; | ||
| import { streamingText, transcript, type TranscriptLine } from "../protocol/transcript.ts"; | ||
|
|
||
| /** What is already on the terminal: finished lines, and a partial left open on the last line. */ | ||
| export interface PrinterState { | ||
| readonly printed: readonly TranscriptLine[]; | ||
| readonly streamed: string | undefined; | ||
| } | ||
|
|
||
| export const initialPrinter: PrinterState = { printed: [], streamed: undefined }; | ||
|
|
||
| const sameLine = (a: TranscriptLine, b: TranscriptLine): boolean => JSON.stringify(a) === JSON.stringify(b); | ||
|
|
||
| const suffixOf = (line: TranscriptLine): string => | ||
| line.role === "assistant" && line.stopReason !== "stop" ? ` [${line.stopReason}]` : ""; | ||
|
|
||
| const format = (line: TranscriptLine): string => `${line.role}: ${line.text}${suffixOf(line)}`; | ||
|
|
||
| /** Append-only terminal output for the next view; a view that rewrites printed history is printed afresh. */ | ||
| export function printIncrement(state: PrinterState, view: ConversationView): { state: PrinterState; output: string } { | ||
| const lines = transcript(view); | ||
| let { printed, streamed } = state; | ||
| let output = ""; | ||
|
|
||
| if (printed.length > lines.length || printed.some((line, i) => !sameLine(line, lines[i]!))) { | ||
| output += `${streamed === undefined ? "" : "\n"}--- view replaced ---\n`; | ||
| printed = []; | ||
| streamed = undefined; | ||
| } | ||
|
|
||
| for (const line of lines.slice(printed.length)) { | ||
| if (streamed === undefined) output += `${format(line)}\n`; | ||
| else if (line.role === "assistant" && line.text.startsWith(streamed)) output += `${line.text.slice(streamed.length)}${suffixOf(line)}\n`; | ||
| else output += `\n${format(line)}\n`; | ||
| streamed = undefined; | ||
| } | ||
| printed = lines; | ||
|
|
||
| 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; | ||
| } | ||
|
Comment on lines
+39
to
+45
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🟡 Abandoned responses merge with later messages If an in-flight response disappears without a committed assistant line, Learn moreThe printer stores the visible in-flight response in Example: The terminal has printed Recommended fix: When Was this helpful? React with 👍 or 👎 to provide feedback. |
||
| return { state: { printed, streamed }, output }; | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🟡 Observer hides notices after twenty
After twenty notices,
renderslices 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.