Repository navigation
feat(spike): on-call demo extensions on the durable host (P4a) #431
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| .tmp/ |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,92 @@ | ||
| import type { Context } from "@earendil-works/chord"; | ||
| import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; | ||
| import type { ConversationId, Harness } from "@earendil-works/pi-durable"; | ||
| import type { ApprovalDecision, PendingApproval } from "../../protocol/demo.ts"; | ||
| import { ApprovalsDoc } from "./state.ts"; | ||
|
|
||
| type Waiter = { readonly request: PendingApproval; resolve(decision: ApprovalDecision): void }; | ||
|
|
||
| /** | ||
| * Who is waiting for a decision. A hook can neither commit nor sleep durably, so the waiting | ||
| * itself lives here, in memory; a restarted host's hook runs again and asks again. Decisions | ||
| * are committed to `ApprovalsDoc`, which is what makes them stand. | ||
| */ | ||
| export class ApprovalBoard { | ||
| readonly #waiters = new Map<string, Waiter[]>(); | ||
| readonly #decided = new Map<string, ApprovalDecision>(); | ||
| readonly #listeners = new Set<(pending: readonly PendingApproval[]) => void>(); | ||
| #harness: Harness | undefined; | ||
|
|
||
| attach(harness: Harness): void { | ||
| this.#harness = harness; | ||
| } | ||
|
|
||
| get value(): readonly PendingApproval[] { | ||
| return [...this.#waiters.values()].map((waiters) => waiters[0]!.request); | ||
| } | ||
|
|
||
| subscribe(listener: (pending: readonly PendingApproval[]) => void): () => void { | ||
| this.#listeners.add(listener); | ||
| return () => this.#listeners.delete(listener); | ||
| } | ||
|
|
||
| /** Resolves with the decision; rejects when `context` is cancelled, for example by Esc. */ | ||
| wait(request: PendingApproval, context: Context): Promise<ApprovalDecision> { | ||
| const decided = this.#decided.get(request.id); | ||
| if (decided !== undefined) return Promise.resolve(decided); | ||
| return new Promise((resolve, reject) => { | ||
| const signal = context.abortSignal; | ||
| const waiter: Waiter = { | ||
| request, | ||
| resolve: (decision) => { | ||
| signal?.removeEventListener("abort", cancel); | ||
| resolve(decision); | ||
| }, | ||
| }; | ||
| const cancel = (): void => { | ||
| this.#remove(waiter); | ||
| reject(signal?.reason ?? new Error("Approval wait cancelled")); | ||
| }; | ||
| if (signal?.aborted) return cancel(); | ||
| signal?.addEventListener("abort", cancel, { once: true }); | ||
| this.#waiters.set(request.id, [...(this.#waiters.get(request.id) ?? []), waiter]); | ||
| this.#changed(); | ||
| }); | ||
| } | ||
|
|
||
| /** Commit a decision unless one exists; returns the one that stands and whether it is this one. */ | ||
| async decide( | ||
| conversationId: ConversationId, | ||
| id: string, | ||
| approved: boolean, | ||
| by: string, | ||
| ): Promise<{ decision: ApprovalDecision; first: boolean }> { | ||
| const harness = this.#harness; | ||
| if (harness === undefined) throw new Error("Approvals are not attached to a Harness"); | ||
| const candidate: ApprovalDecision = { approved, by, at: Date.now() }; | ||
| const decision = await harness.commit(async (tx) => { | ||
| const doc = await tx.doc(ApprovalsDoc, conversationId); | ||
| if (doc.decisions[id] === undefined && !this.#waiters.has(id)) throw new Error(`No pending approval ${id}`); | ||
| doc.decisions[id] ??= { ...candidate }; | ||
| return { ...doc.decisions[id] } as ApprovalDecision; | ||
| }, BACKGROUND_CONTEXT); | ||
| this.#decided.set(id, decision); | ||
| const waiters = this.#waiters.get(id) ?? []; | ||
| this.#waiters.delete(id); | ||
| for (const waiter of waiters) waiter.resolve(decision); | ||
| if (waiters.length > 0) this.#changed(); | ||
| return { decision, first: decision.by === candidate.by && decision.at === candidate.at }; | ||
|
Contributor
There was a problem hiding this comment. 🟡 Losing approval reports a winning decision When the same client name votes twice within one millisecond, Learn moreThe transaction persists only the first decision, but the returned Example: Two controls named Recommended fix: Compute a boolean inside the transaction when the decision is inserted, and return that boolean alongside the stored decision. Was this helpful? React with 👍 or 👎 to provide feedback. |
||
| } | ||
|
|
||
| #remove(waiter: Waiter): void { | ||
| const rest = (this.#waiters.get(waiter.request.id) ?? []).filter((candidate) => candidate !== waiter); | ||
| if (rest.length > 0) this.#waiters.set(waiter.request.id, rest); | ||
| else this.#waiters.delete(waiter.request.id); | ||
| this.#changed(); | ||
| } | ||
|
|
||
| #changed(): void { | ||
| const pending = this.value; | ||
| for (const listener of this.#listeners) listener(pending); | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| // A scripted stand-in for a real model that walks the on-call playbook. Requests of the main | ||
| // conversation and its parallel subagents arrive interleaved, so every answer is decided from | ||
| // the request itself: the tools it offers and the messages since the newest user input. | ||
| import type { Message } from "@earendil-works/pi-ai"; | ||
| import { fauxAssistantMessage, type FauxResponseFactory, fauxText, fauxToolCall } from "@earendil-works/pi-ai/providers/faux"; | ||
| import { AREAS } from "./oncall.ts"; | ||
|
|
||
| const INVESTIGATION: readonly string[] = Object.values(AREAS).map((area) => area.tool); | ||
|
|
||
| function textOf(message: Message | undefined): string { | ||
| if (message === undefined || message.role === "system") return ""; | ||
| if (typeof message.content === "string") return message.content; | ||
| return message.content.flatMap((block) => (block.type === "text" ? [block.text] : [])).join(""); | ||
| } | ||
|
|
||
| /** The tools a request offers, from its system messages' additions and removals. */ | ||
| function offeredTools(messages: readonly Message[]): string[] { | ||
| const tools = new Set<string>(); | ||
| for (const message of messages) { | ||
| if (message.role !== "system") continue; | ||
| for (const tool of message.toolsAdded ?? []) tools.add(tool.name); | ||
| for (const tool of message.toolsRemoved ?? []) tools.delete(tool.name); | ||
| } | ||
| return [...tools]; | ||
| } | ||
|
|
||
| const PLAN = ["Search the deploy logs", "Compare error metrics", "Review the commits in v2.3"]; | ||
|
|
||
| export function oncallScript({ reminderSeconds = 60 }: { reminderSeconds?: number } = {}): FauxResponseFactory { | ||
| return (context, _options, state) => { | ||
| const { messages } = context; | ||
| const call = (name: string, args: Parameters<typeof fauxToolCall>[1], index = 0) => | ||
| fauxToolCall(name, args, { id: `call-${state.callCount}-${index}` }); | ||
| const calls = (...blocks: ReturnType<typeof call>[]) => fauxAssistantMessage(blocks, { stopReason: "toolUse" }); | ||
| const tools = offeredTools(messages); | ||
| const userAt = messages.findLastIndex((message) => message.role === "user"); | ||
| const input = textOf(messages[userAt]); | ||
| // A changed section, such as the plan after update_plan, lands as a system message after the results. | ||
| const last = messages.slice(userAt + 1).findLast((message) => message.role !== "system"); | ||
|
|
||
| // A subagent: one investigation tool, called once, then reported on. | ||
| const own = tools.length === 1 && INVESTIGATION.includes(tools[0]!) ? tools[0]! : undefined; | ||
| if (own !== undefined) { | ||
| if (last?.role !== "toolResult") return calls(call(own, { query: input })); | ||
| const first = textOf(last).split("\n").find((line) => /ERROR|->|pg driver/.test(line)) ?? textOf(last).split("\n")[0]; | ||
| return fauxAssistantMessage(`Found in ${own}: ${first}`); | ||
| } | ||
|
|
||
| if (last?.role === "toolResult") { | ||
| if (last.isError) return fauxAssistantMessage(`${last.toolName} did not run: ${textOf(last)} Standing by.`); | ||
| switch (last.toolName) { | ||
| case "update_plan": | ||
| if (messages.some((message) => message.role === "toolResult" && message.toolName === "subagent")) { | ||
| return fauxAssistantMessage( | ||
| "The pg driver bump in a1b2c3d cut the pool from 100 to 20; eu-west exhausted it at 09:14 and errors went to 14.8%. I recommend rolling back to v2.2.", | ||
| ); | ||
| } | ||
| return calls( | ||
| call("subagent", { area: "logs", task: "find the errors around the v2.3 deploy in the logs" }, 0), | ||
| call("subagent", { area: "metrics", task: "compare error metrics before and after v2.3" }, 1), | ||
| call("subagent", { area: "commits", task: "review the commits in v2.3 for risky changes" }, 2), | ||
| ); | ||
| case "subagent": | ||
| return calls(call("update_plan", { items: PLAN.map((text) => ({ text, status: "done" })) })); | ||
| case "rollback": | ||
| return calls(call("schedule_check", { seconds: reminderSeconds, note: "verify the error rate after the rollback" })); | ||
| case "schedule_check": | ||
| return fauxAssistantMessage("Check scheduled; I will re-check when the reminder arrives."); | ||
| default: | ||
| return fauxAssistantMessage("Done."); | ||
| } | ||
| } | ||
|
|
||
| if (input.startsWith("Reminder:")) return fauxAssistantMessage("Re-checked: the error rate is back to baseline (0.2%) in every region."); | ||
| if (/postmortem/i.test(input)) { | ||
| return fauxAssistantMessage([fauxText("Postmortem: v2.3 shipped a pg driver bump that shrank the connection pool; eu-west saturated first.")]); | ||
| } | ||
| if (/roll ?back/i.test(input)) return calls(call("rollback", { version: "v2.2", reason: "pg driver pool regression in v2.3" })); | ||
| const delay = /check again in (\d+) seconds?/i.exec(input); | ||
| if (delay !== null) return calls(call("schedule_check", { seconds: Number(delay[1]), note: "re-check the error rate" })); | ||
| if (tools.includes("update_plan") && !messages.some((message) => message.role === "toolResult" && message.toolName === "update_plan")) { | ||
| return calls(call("update_plan", { items: PLAN.map((text) => ({ text, status: "todo" })) })); | ||
|
Comment on lines
+81
to
+82
Contributor
There was a problem hiding this comment. 🟡 Later investigations skip the demo playbook After one investigation, Learn moreThe scripted model selects the latest user message at Example: The user asks to investigate v2.3, receives a recommendation, then asks to investigate the release again. The second request contains the old Recommended fix: Scope the Was this helpful? React with 👍 or 👎 to provide feedback. |
||
| } | ||
| return fauxAssistantMessage(`Noted: ${input}`); | ||
| }; | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| import { fileURLToPath, pathToFileURL } from "node:url"; | ||
| import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; | ||
| import { createRegistry, type Extension, type Registry } from "@earendil-works/pi-durable"; | ||
| import type { OpenHostOptions } from "../host.ts"; | ||
| import { ApprovalBoard } from "./approvals.ts"; | ||
| import { AREAS, createOncall } from "./oncall.ts"; | ||
| import { PlanDoc, RolloutDoc } from "./state.ts"; | ||
| import { DemoTasks } from "./tasks.ts"; | ||
|
|
||
| export interface DemoOptions { | ||
| /** Delay between streamed lines of an investigation tool's output. */ | ||
| readonly paceMs?: number; | ||
| /** Length of one rollback step in a region. */ | ||
| readonly stepMs?: number; | ||
| /** The investigation tools' module; `reload()` imports it again. */ | ||
| readonly investigationModule?: string; | ||
| } | ||
|
|
||
| export interface Demo { | ||
| readonly registry: Registry; | ||
| readonly investigationModule: string; | ||
| /** What `openHost` needs to run the demo: its registry, approvals, documents, and root setup. */ | ||
| readonly hostOptions: Pick<OpenHostOptions, "registry" | "approvals" | "docs" | "onOpen">; | ||
| /** Reinstall the investigation tools from their module; returns the module's load count. */ | ||
| reload(): Promise<number>; | ||
| } | ||
|
|
||
| const defaultModule = fileURLToPath(new URL("./investigation.ts", import.meta.url)); | ||
|
|
||
| export async function createDemo(options: DemoOptions = {}): Promise<Demo> { | ||
| const { paceMs = 400, stepMs = 2000, investigationModule = defaultModule } = options; | ||
| const registry = createRegistry(); | ||
| const approvals = new ApprovalBoard(); | ||
| let loads = 0; | ||
| const load = async (): Promise<Extension> => { | ||
| // A fresh URL is a fresh module instance; the registry replaces the extension of the same name in place. | ||
| const module = (await import(`${pathToFileURL(investigationModule).href}?v=${++loads}`)) as typeof import("./investigation.ts"); | ||
| return module.createInvestigation({ paceMs }); | ||
| }; | ||
| registry.install(DemoTasks); | ||
| registry.install(await load()); | ||
| registry.install(createOncall({ approvals, stepMs })); | ||
| return { | ||
| registry, | ||
| investigationModule, | ||
| hostOptions: { | ||
| registry, | ||
| approvals, | ||
| docs: [PlanDoc, RolloutDoc], | ||
| async onOpen(harness, root) { | ||
| approvals.attach(harness); | ||
| // The main agent delegates instead of investigating itself; tools are stored by name, so this survives reloads. | ||
| const investigation = registry.snapshot().extension("investigation")!.tools ?? []; | ||
| const own = investigation.filter((tool) => Object.values(AREAS).some((area) => area.tool === tool.name)); | ||
| await root.configure({ tools: { remove: own } }, BACKGROUND_CONTEXT); | ||
| }, | ||
| }, | ||
| async reload() { | ||
| registry.install(await load()); | ||
| return loads; | ||
| }, | ||
| }; | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.
🟥 Approval decision bypasses conversation binding
A client can send an approval ID with a different conversation ID.
ApprovalBoardchecks pending IDs globally, then records the decision in the supplied conversation while resolving the original waiter. The decision does not survive replay of the original conversation.Was this helpful? React with 👍 or 👎 to provide feedback.