src,index.ts,skills/threads,test: simplify commitment ledger
123345656891011121378159101718191112211314151617182519202721222324252633342728362930383940414243313233343536463738484950394041424352444554464756484950596061626364655152535455565758685960707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182616218518618763646566190191192193194195196197198199200201202# prime-commitment-observer
an extension for [Prime Agent](https://github.com/PrimeIntellect-ai/prime-agent) that stops the agent from forgetting things you asked for.A Prime Agent extension that remembers unfinished user requests across turns without making the foreground model manage a task registry.
### the problemif you ask the agent to do two or three things in one message (like *"fix this bug, add tests, and update the docs"*), it usually starts working on the first thing, goes down a rabbit hole, and completely forgets the rest. even worse, when conversation history gets summarized or compacted, older requests disappear completely.## behavior
### how this fixes itthis extension makes sure the agent actually finishes everything you asked:1. **splits your requests**: breaks compound messages into separate tasks so nothing gets lost.2. **keeps open tasks visible**: pins still-unfinished tasks into the prompt on every turn, even after compaction.3. **checks the work**: a small background model watches what gets done and stages tasks for completion when work lands or tests pass.4. **safe completions**: for code changes, tasks only close if tests and checks actually pass. for conversational requests, tasks close once the model verifies they're satisfied and uncontested. you can always manually adjust or reopen anything with `/threads`.The extension keeps one append-only, branch-aware ledger in the session file. A small observer model runs after completed agent turns and classifies each request as `open`, `active`, `waiting`, `blocked`, `done`, `cancelled`, or `superseded`.
## requirementsBefore a user-originated turn, Prime receives one hidden `<pending_requests>` context block when the pending set changed. Internal wakes such as child completion notices and background command completions never create another block. Closing the final request emits one clear marker so an older snapshot cannot remain authoritative.
- Prime Agent 0.9.1, or a compatible extension API- an API key for the observer model (default `google/gemma-4-31b-it`)- Node 24+ (the tests use native TypeScript type stripping)This is a lightweight semantic ledger. It does not infer test results or claim verified completion from command strings. The observer judges the user messages and the completed assistant response.
extensions run with your full user permissions. review the source before installing it.## requirements
This version requires a Prime Agent build whose `before_agent_start` extension event includes `source`. The extension uses it to distinguish user input from internal wakes.
## install
from a clone:From a clone:
```sh```bashgit clone git@knot.gaze.systems:did:plc:gyggh67fdq2hukuabczwhm3w prime-commitment-observercd prime-commitment-observerprime-agent package install .```
`package install .` records the checkout in `~/.prime/agent/settings.json`. keep the checkout inplace while the package is installed. for one project only, add `--local`.The package manager records the checkout in `~/.prime/agent/settings.json`. Keep the checkout in place while it is installed. Reload extensions or restart Prime Agent.
`~/.prime/agent/settings.json` then contains an entry like:## configuration
```json{ "packages": [ { "source": "/Users/you/proj/prime-commitment-observer", "type": "local" } ]}```bashexport PRIME_COMMITMENT_OBSERVER_PROVIDER=googleexport PRIME_COMMITMENT_OBSERVER_MODEL=gemini-3-flash-previewexport PRIME_COMMITMENT_OBSERVER_TIMEOUT_MS=60000```
Prime also auto-discovers extension directories under `~/.prime/agent/extensions/`:Optional variables:
```shgit clone git@knot.gaze.systems:did:plc:gyggh67fdq2hukuabczwhm3w ~/.prime/agent/extensions/commitment-observer```- `PRIME_COMMITMENT_OBSERVER_API_KEY`- `PRIME_COMMITMENT_OBSERVER_BASE_URL`- `PRIME_COMMITMENT_OBSERVER_TEMPERATURE`- `PRIME_COMMITMENT_OBSERVER_MAX_TOKENS`
restart Prime, or run `/reload` in a running session.Without an explicit API key, the observer uses Prime Agent's configured credentials.
## use## manual control
### for users (slash commands)`/threads` lists the ledger. Manual mutations stay in the extension command surface:
```text/threads list open and closed threads/threads done <id> explicit user acceptance -> done/threads dismiss <id> drop the thread as not wanted/threads reopen <id> put a closed thread back to open/threads drop <id> remove a wrongly created thread entirely/threads edit <id> <text> correct the canonical request/threads reconcile run the observer now/threads done <id>/threads dismiss <id> [reason]/threads reopen <id>/threads drop <id>/threads edit <id> <request>/threads reconcile```
### for agents (python skill)The foreground model is not given thread ids and has no `threads` skill.
agents can inspect and control threads programmatically from the python REPL using the bundled `threads` skill:
```pythonthreads.list() # list open and closed threadsthreads.complete("t1", reason="done in test") # mark thread completethreads.dismiss("t2", reason="obsolete") # dismiss unwanted threadthreads.reopen("t3") # reopen a closed thread```
nothing is hidden: every thread the model tracks is visible and correctable here.
### environment variables
| variable | meaning || --- | --- || `PRIME_COMMITMENT_OBSERVER_PROVIDER` | observer provider, default `google` || `PRIME_COMMITMENT_OBSERVER_MODEL` | observer model id, default `gemma-4-31b-it` || `PRIME_COMMITMENT_OBSERVER_MAX_TOKENS` | observer output budget, default `4000` || `PRIME_COMMITMENT_OBSERVER_TIMEOUT_MS` | observer call timeout, default `60000` || `PRIME_COMMITMENT_OBSERVER_STRUCTURED` | `0` disables the `responseFormat` request || `PRIME_COMMITMENT_TOKEN_CAP` | hard cap for the injected block, default `400` tokens || `PRIME_COMMITMENT_AUTOPROMOTE` | `0` disables auto promotion (on by default) |
## how it works
### when the observer runs
- `input`: every incoming user message is persisted immediately as an `unreconciled` event, so a crash, a steer, or a fork cannot lose it.- `agent_end`: the request boundary. if unreconciled messages exist, one observer call runs off the handler in the background, so the UI is never blocked by it.- `session_start` (resume, fork, new): unreconciled messages left by an abnormal termination are reconciled on the next boundary.- `tool_result`: every tool result of the current request is recorded as an action with a label and a `required` flag. validation commands (`npm run check`, `cargo test`, `pytest`, ...) are marked required.
low-level `turn_end` is deliberately not used. tool turns are not request boundaries.
### the observer packet & thread carryover
the packet contains:- prior thread state including `requestedProperties` and previously `implementedProperties`- the preceding assistant message (which makes references like "the second one" or "that option" resolvable)- the full unreconciled user messages- the completed assistant turn- identified actions `{id, label, required, ok}`
previously implemented properties are carried over across turns. this prevents the "local window trap"where switching focus to a new task causes the observer to forget properties implemented in earlier turns.
### output schema
evidence comes before the verdict, and declaration order is part of that:
`canonicalRequest -> requestedProperties -> implementedProperties -> contradictions ->evidenceActionIds -> status`
allowed statuses: `open`, `active`, `blocked`, `waiting`, `completion_candidate`, `cancelled`,`superseded`. `done` is not in the enum. in benchmarks this eliminates hallucinated closures on weakmodels because closure is structurally forbidden to the LLM.
`evidenceActionIds` cites the packet action ids that bear on that thread's completion. action ids(`a1`, `a2`, ...) are stable inside one observer call and mean nothing outside it.
### promotion policy
`completion_candidate` is a proposal. a thread becomes `done` only via the pure host gate`promotionDecision` in `src/state.ts`:
1. **explicit user acceptance** - `/threads done <id>`, recorded as `promotedBy: user_accepted`.2. **validation evidence** - recorded as `promotedBy: evidence`. when validation commands (`npm test`, `cargo check`, etc.) ran in the turn, the thread must cite at least one passing required action, all cited actions must be `ok: true`, and contradictions must be empty. citing an optional action (like `git status`) instead of the required test suite blocks promotion (`cited_action_optional`).3. **uncontested delivery / conversational promotion** - recorded as `promotedBy: uncontested`. when no validation commands ran in the turn (operational actions like `git push`, script runs, file creations, or pure conversational/design instructions), citing successful delivery actions (`ok: true`) or uncontested candidates with empty contradictions auto-promotes to `done` immediately.
any cited failing action blocks promotion (`cited_action_failed`). any failing required validation in theturn blocks promotion for all threads.
auto promotion is visible and reversible: each one raises a notification, `/threads` showsthe promotion reason (`<evidence>` or `<uncontested>`), and `/threads reopen <id>` reopens it.
### durable state
state lives in append-only custom session entries of type `commitment-event`, written with`ctx.appendEntry`. current state is a left fold over the entries on the current branch(`ctx.sessionManager.getBranch()`), so it:
- survives compaction, because entries are not conversation messages and are never summarized- survives resume, because it is read back from the session file- follows forks and branch navigation, because the fold is per branch- is independent of any other extension
nothing is stored in compaction summary prose.
### prompt caching & injection
injection uses the native `before_agent_start` hook. it generates the `<open_threads>` custom messagewith `display: false` and natively commits it to the session branch.
because it is committed before the turn runs:- the prompt prefix is strictly append-only- mid-turn tool loops do not re-inject or mutate the prefix- consecutive turns preserve Anthropic prompt caching (measured at **95.3% cache hit rate**, avoiding the 5.62x cache-write token explosion caused by ephemeral `context` hooks)
the block is capped (default 400 tokens, roughly 4 chars per token). threads are ordered`blocked`, `waiting`, `active`, `open`, `completion_candidate`, then by creation time.
## development
```shnpm installnpm test # pure logic tests, no provider calls (24 tests)```bashnpm testnpm run check```
## evaluation benchmarks
comprehensive evaluation across 102 test cases in `observer-bench/`:
- **Track 1: Multi-Intent Split (MixATIS / SNIPS, 50 cases)**: 100.0% exact intent count accuracy (10/10 single-intent, 25/25 dual-intent, 15/15 triple-intent).- **Track 2: Empirical Supersession (Google SGD, 30 dialogues)**: 76.5% recall on user pivots (changing minds, replacing constraints).- **Track 3: Real Coding Session Candidates (Prime Agent logs, 22 cases)**: 100.0% role-enum safety (`done` never emitted by observer), 100% structured JSON validity.
see `observer-bench/PROPER-BENCH-RESULTS.md` and `observer-bench/CACHE-AUDIT.md` for full reports.122346756971189131410161718192011121314151617182419202627282930313233343536373839404142434445464748495051525354555657585960616263646566212223707172737424252627282930777879313233343536828337383940414243444586878889909146474849509596975152535499100555657585910460616210710810911063646566676811211311469707172737475767711611711811978798012112212312412512612712812913013113213313413581828384858687888990137138139140911421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761779293949596979899100101102103104105106107108109110111112113114115116117118119120121122180181182183184185186187188189190123124125126127128129130131132133134135136137138192139140194195141142143144145146147148149150151198199200201152153154155203204205206207208209156157158159211212213214215216217218219220221222223224225226227160161162163164165166167168169170171172173174175176177178179180181182183184185186187230231232233234235236237238239240241242243244188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288230231232233234235236290291import { randomUUID } from "node:crypto";import type { AgentMessage } from "@earendil-works/pi-agent-core";import type { AgentEndEvent, BeforeAgentStartEvent, BeforeAgentStartEventResult, ContextEvent, CustomEntry, ExtensionAPI, ExtensionCommandContext, ExtensionContext, InputEvent, InputSource, SessionEntry, SessionStartEvent, ToolResultEvent,} from "@earendil-works/pi-coding-agent";import { DEFAULT_INJECTION_TOKEN_CAP, renderOpenThreads } from "./src/inject.ts";import { observerConfigFromEnv, runObserver } from "./src/observer.ts";import { alignActionIds, buildPacket, isRequiredValidation, type PacketMessage } from "./src/packet.ts";import { COMMITMENT_EVENT_TYPE, isCommitmentEvent, isLive, replay } from "./src/state.ts";import type { ActionRecord, CommitmentEvent, Registry, ReplayOptions, Thread } from "./src/types.ts";import { CLEARED_CONTEXT, nextContextRevision, REFRESHING_CONTEXT } from "./src/inject.ts";import { configFromEnv, runObserver } from "./src/observer.ts";import { isCommitmentEvent, liveThreads, observationChangesState, replay } from "./src/state.ts";import type { CommitmentEvent, Registry } from "./src/types.ts";
const COMMITMENT_EVENT_TYPE = "commitment-event-v2";const CONTEXT_CUSTOM_TYPE = "commitment-threads";const LIST_CUSTOM_TYPE = "commitment-threads-list";const MAX_OBSERVER_FAILURES = 3;type BeforeAgentStartWithSource = BeforeAgentStartEvent & { source: InputSource | "internal" };
interface ContentBlockLike { type: string; text?: string;}
function isContentBlock(value: unknown): value is ContentBlockLike { return !!value && typeof value === "object" && typeof (value as { type?: unknown }).type === "string";}
function contentText(content: unknown): string { if (typeof content === "string") return content; if (!Array.isArray(content)) return ""; return content .filter(isContentBlock) .filter((block) => block.type === "text" && typeof block.text === "string") .map((block) => block.text ?? "") .join("\n");}
function packetRole(role: string): PacketMessage["role"] { if (role === "user" || role === "assistant" || role === "toolResult") return role; return "other";}
function toPacketMessages(messages: readonly AgentMessage[]): PacketMessage[] { return messages.map((message) => { const record = message as unknown as { role: string; content: unknown; toolName?: string; isError?: boolean }; return { role: packetRole(record.role), text: contentText(record.content), toolName: typeof record.toolName === "string" ? record.toolName : undefined, isError: record.isError === true, }; });}
function branchMessages(entries: readonly SessionEntry[]): PacketMessage[] { const messages = entries.flatMap((entry) => (entry.type === "message" ? [entry.message] : [])); return toPacketMessages(messages);}
function readEvents(ctx: ExtensionContext): CommitmentEvent[] { return ctx.sessionManager .getBranch() .flatMap((entry) => entry.type === "custom" && entry.customType === COMMITMENT_EVENT_TYPE && isCommitmentEvent(entry.data) ? [entry.data] : [], ); .filter( (entry): entry is CustomEntry => entry.type === "custom" && entry.customType === COMMITMENT_EVENT_TYPE, ) .map((entry) => entry.data) .filter(isCommitmentEvent);}
function tokenCapFromEnv(env: NodeJS.ProcessEnv): number { const parsed = Number(env.PRIME_COMMITMENT_TOKEN_CAP ?? ""); return Number.isFinite(parsed) && parsed > 0 ? Math.trunc(parsed) : DEFAULT_INJECTION_TOKEN_CAP;const registryFrom = (ctx: ExtensionContext): Registry => replay(readEvents(ctx));
function appendEvent(pi: ExtensionAPI, event: CommitmentEvent): void { pi.appendEntry(COMMITMENT_EVENT_TYPE, event);}
function replayOptionsFromEnv(env: NodeJS.ProcessEnv): ReplayOptions { return { autoPromoteOnEvidence: env.PRIME_COMMITMENT_AUTOPROMOTE !== "0" };function latestContext(entries: readonly SessionEntry[]): string | undefined { for (let index = entries.length - 1; index >= 0; index -= 1) { const entry = entries[index]; if (entry.type !== "custom_message" || entry.customType !== CONTEXT_CUSTOM_TYPE) continue; return typeof entry.content === "string" ? entry.content : undefined; } return undefined;}
function formatThread(thread: Thread): string { const missing = thread.requestedProperties.filter((property) => !thread.implementedProperties.includes(property)); const notes = [...thread.contradictions, ...missing.map((property) => `missing: ${property}`)]; const suffix = notes.length > 0 ? `\n ${notes.join("; ")}` : ""; const promoted = thread.promotedBy ? ` <${thread.promotedBy}>` : ""; return ` ${thread.id} [${thread.status}]${promoted} ${thread.canonicalRequest}${suffix}`;function formatThread(thread: Registry["threads"][number]): string { return `[${thread.id}] ${thread.status} ${thread.canonicalRequest}`;}
function formatRegistry(registry: Registry): string { if (registry.threads.length === 0) return "No tracked threads yet."; const live = registry.threads.filter(isLive); const closed = registry.threads.filter((thread) => !isLive(thread)); const open = liveThreads(registry); const openIds = new Set(open.map((thread) => thread.id)); const closed = registry.threads.filter((thread) => !openIds.has(thread.id)); const sections: string[] = []; if (live.length > 0) sections.push(`Open threads:\n${live.map(formatThread).join("\n")}`); if (closed.length > 0) sections.push(`Closed threads:\n${closed.map(formatThread).join("\n")}`); if (open.length > 0) sections.push(`Open requests:\n${open.map(formatThread).join("\n")}`); if (closed.length > 0) sections.push(`Closed requests:\n${closed.map(formatThread).join("\n")}`); if (registry.unreconciled.length > 0) { sections.push(`Unreconciled user messages: ${registry.unreconciled.length}`); } return sections.join("\n\n"); return sections.join("\n\n") || "No tracked requests.";}
export function createCommitmentObserverExtension(env: NodeJS.ProcessEnv = process.env) { const observerConfig = observerConfigFromEnv(env); const replayOptions = replayOptionsFromEnv(env); const tokenCap = tokenCapFromEnv(env);export default function commitmentObserver(pi: ExtensionAPI): void { const cfg = configFromEnv(); let observerQueue = Promise.resolve(); let pendingReconciliations = 0; let consecutiveFailures = 0;
let pendingActions: ActionRecord[] = []; let failureStreak = 0; let queue: Promise<void> = Promise.resolve(); const recordFailure = (ctx: ExtensionContext, messageIds: string[], error: unknown): void => { const text = (error instanceof Error ? error.message : String(error)).slice(0, 2_000); appendEvent(pi, { kind: "observer_failure", messageIds, error: text, at: Date.now() }); consecutiveFailures += 1; if (consecutiveFailures === 1 && ctx.hasUI) { ctx.ui.notify(`Commitment observer failed: ${text}`, "warning"); } };
return (pi: ExtensionAPI): void => { const appendEvent = (event: CommitmentEvent): void => { pi.appendEntry<CommitmentEvent>(COMMITMENT_EVENT_TYPE, event); }; const branchContains = (ctx: ExtensionContext, entryId: string | null): boolean => entryId === null || ctx.sessionManager.getBranch().some((entry) => entry.id === entryId);
const announceAutoPromotions = ( ctx: ExtensionContext, before: Registry, observation: CommitmentEvent, ): void => { if (!ctx.hasUI) return; const after = replay([...readEvents(ctx), observation], replayOptions); const previous = new Map(before.threads.map((thread) => [thread.id, thread.status])); const promoted = after.threads.filter( (thread) => thread.promotedBy === "evidence" && previous.get(thread.id) !== "done", ); for (const thread of promoted) { ctx.ui.notify(`thread ${thread.id} auto-closed on cited evidence: ${thread.canonicalRequest}`, "info"); } }; const liveRevision = (registry: Registry): string => JSON.stringify( liveThreads(registry).map(({ id, canonicalRequest, status, updatedAt }) => ({ id, canonicalRequest, status, updatedAt, })), );
const reconcile = async (ctx: ExtensionContext, messages: PacketMessage[]): Promise<void> => { if (failureStreak >= MAX_OBSERVER_FAILURES) return; const registry = replay(readEvents(ctx), replayOptions); if (registry.unreconciled.length === 0) return;
const live = registry.threads.filter(isLive); const actions = alignActionIds(pendingActions); pendingActions = []; const packet = buildPacket(messages, registry.unreconciled, live, actions); const knownIds = new Set(live.map((thread) => thread.id));
const result = await runObserver(ctx.modelRegistry, observerConfig, packet, knownIds); if (!result.ok) { failureStreak += 1; pendingActions = [...actions, ...pendingActions]; if (ctx.hasUI && failureStreak === MAX_OBSERVER_FAILURES) { ctx.ui.notify(`commitment observer disabled for now: ${result.error}`, "warning"); } return; } failureStreak = 0; const observation: CommitmentEvent = { kind: "observation", at: Date.now(), reconciled: registry.unreconciled.map((message) => message.messageId), output: result.report.output, actions, }; appendEvent(observation); announceAutoPromotions(ctx, registry, observation); };
const scheduleReconcile = (ctx: ExtensionContext, messages: PacketMessage[]): void => { queue = queue.then(() => reconcile(ctx, messages)).catch(() => undefined); };
pi.on("input", (event: InputEvent): void => { const text = event.text.trim(); if (text.length === 0 || text.startsWith("/")) return; failureStreak = 0; appendEvent({ kind: "user_message", messageId: randomUUID(), text, at: Date.now() }); const reconcile = async ( ctx: ExtensionContext, messages: AgentEndEvent["messages"], targetMessageIds: ReadonlySet<string>, boundaryEntryId: string | null, ): Promise<void> => { if (!branchContains(ctx, boundaryEntryId)) return; const registry = registryFrom(ctx); const currentTargets = new Set( registry.unreconciled .map((message) => message.messageId) .filter((messageId) => targetMessageIds.has(messageId)), ); if (currentTargets.size === 0 && liveThreads(registry).length === 0) return; const revision = liveRevision(registry); const result = await runObserver(registry, messages, currentTargets, ctx.modelRegistry, cfg); if (!branchContains(ctx, boundaryEntryId) || liveRevision(registryFrom(ctx)) !== revision) return; if (!result.ok) { recordFailure(ctx, [...currentTargets], result.error); return; } consecutiveFailures = 0; if (currentTargets.size === 0 && !observationChangesState(registry, result.output)) return; appendEvent(pi, { kind: "observation", reconciled: [...currentTargets], output: result.output, at: Date.now(), }); };
pi.on("tool_result", (event: ToolResultEvent): void => { const input = event.input; const command = typeof input.command === "string" ? input.command : typeof input.code === "string" ? input.code : ""; const detail = command.trim().split("\n")[0] ?? ""; pendingActions.push({ id: `a${pendingActions.length + 1}`, label: detail ? `${event.toolName}: ${detail}` : event.toolName, required: isRequiredValidation(command), ok: !event.isError, at: Date.now(), const scheduleReconcile = ( ctx: ExtensionContext, messages: AgentEndEvent["messages"] = [], ): Promise<void> => { const targetMessageIds = new Set(registryFrom(ctx).unreconciled.map((message) => message.messageId)); const boundaryEntryId = ctx.sessionManager.getLeafId(); pendingReconciliations += 1; observerQueue = observerQueue .then(() => reconcile(ctx, messages, targetMessageIds, boundaryEntryId)) .catch((error) => { if (branchContains(ctx, boundaryEntryId)) recordFailure(ctx, [...targetMessageIds], error); }) .finally(() => { pendingReconciliations -= 1; }); return observerQueue; }); };
pi.on("agent_end", (event: AgentEndEvent, ctx: ExtensionContext): void => { scheduleReconcile(ctx, toPacketMessages(event.messages)); pi.on("input", async (event) => { if (event.text.trim().length === 0 || event.text.trimStart().startsWith("/")) return { action: "continue" }; appendEvent(pi, { kind: "user_message", messageId: randomUUID(), text: event.text, at: Date.now(), }); return { action: "continue" }; });
pi.on("session_start", (event: SessionStartEvent, ctx: ExtensionContext): void => { if (event.reason === "reload") return; scheduleReconcile(ctx, branchMessages(ctx.sessionManager.getBranch())); }); pi.on("agent_end", async (event, ctx) => { void scheduleReconcile(ctx, event.messages); });
// Ensure /threads output shown in the TUI never enters model context pi.on("context", (event: ContextEvent): { messages?: AgentMessage[] } | void => { const filtered = event.messages.filter( (message) => !(message.role === "custom" && message.customType === LIST_CUSTOM_TYPE), ); return filtered.length === event.messages.length ? undefined : { messages: filtered }; }); pi.on("session_start", async (event, ctx) => { if (event.reason !== "new") void scheduleReconcile(ctx); });
pi.on( "before_agent_start", async ( _event: BeforeAgentStartEvent, ctx: ExtensionContext, ): Promise<BeforeAgentStartEventResult | undefined> => { const registry = replay(readEvents(ctx), replayOptions); const block = renderOpenThreads(registry.threads, tokenCap); if (!block) return undefined; return { message: { customType: CONTEXT_CUSTOM_TYPE, content: block, display: false, }, }; }, pi.on("context", async (event) => ({ messages: event.messages.filter( (message) => message.role !== "custom" || message.customType !== LIST_CUSTOM_TYPE, ), }));
pi.on("before_agent_start", async (rawEvent, ctx) => { const event = rawEvent as BeforeAgentStartWithSource; if (event.source === "internal") return undefined; const content = nextContextRevision( event.source, registryFrom(ctx), latestContext(ctx.sessionManager.getBranch()), pendingReconciliations > 0, ); if (!content) return undefined; return { message: { customType: CONTEXT_CUSTOM_TYPE, content, display: false, details: { state: content === CLEARED_CONTEXT ? "clear" : content === REFRESHING_CONTEXT ? "refreshing" : "open", }, }, }; });
pi.registerCommand("threads", { description: "List and edit tracked open threads", getArgumentCompletions: (prefix: string) => ["done", "dismiss", "reopen", "drop", "edit", "reconcile"] .filter((name) => name.startsWith(prefix)) .map((name) => ({ value: name, label: name })), handler: async (args: string, ctx: ExtensionCommandContext): Promise<void> => { const [subcommand = "", id = "", ...rest] = args.trim().split(/\s+/); const registry = replay(readEvents(ctx), replayOptions); const show = (text: string): void => { pi.sendMessage({ customType: LIST_CUSTOM_TYPE, content: text, display: true }); };
if (subcommand === "") { show(formatRegistry(registry)); pi.registerCommand("threads", { description: "Inspect and edit tracked user requests", getArgumentCompletions: (prefix: string) => ["done", "dismiss", "reopen", "drop", "edit", "reconcile"] .filter((value) => value.startsWith(prefix)) .map((value) => ({ value, label: value })), handler: async (args, ctx) => { const show = (content: string): void => { pi.sendMessage({ customType: LIST_CUSTOM_TYPE, content, display: true }); }; const [command = "", id = "", ...rest] = args.trim().split(/\s+/); const registry = registryFrom(ctx); if (!command) { show(formatRegistry(registry)); return; } if (command === "reconcile") { await scheduleReconcile(ctx); return; } const thread = registry.threads.find((candidate) => candidate.id === id); if (!thread) { show(`Unknown request ${id || "id"}.`); return; } const at = Date.now(); if (command === "done" || command === "dismiss" || command === "reopen" || command === "drop") { appendEvent(pi, { kind: "user_action", threadId: id, action: command === "done" ? "accept" : command, reason: rest.join(" ") || undefined, at, }); return; } if (command === "edit") { const canonicalRequest = rest.join(" ").trim(); if (!canonicalRequest) { show("Usage: /threads edit <id> <request>"); return; } if (subcommand === "reconcile") { scheduleReconcile(ctx, branchMessages(ctx.sessionManager.getBranch())); show("Observer reconcile scheduled."); return; } if (!registry.threads.some((thread) => thread.id === id)) { show(`Unknown thread id "${id}".\n\n${formatRegistry(registry)}`); return; } switch (subcommand) { case "done": appendEvent({ kind: "user_action", at: Date.now(), threadId: id, action: "accept" }); show(`Thread ${id} marked done by user acceptance.`); return; case "dismiss": appendEvent({ kind: "user_action", at: Date.now(), threadId: id, action: "dismiss" }); show(`Thread ${id} dismissed.`); return; case "reopen": appendEvent({ kind: "user_action", at: Date.now(), threadId: id, action: "reopen" }); show(`Thread ${id} reopened.`); return; case "drop": appendEvent({ kind: "user_action", at: Date.now(), threadId: id, action: "drop" }); show(`Thread ${id} removed from the registry.`); return; case "edit": { const text = rest.join(" ").trim(); if (text.length === 0) { show("Usage: /threads edit <id> <corrected request>"); return; } appendEvent({ kind: "correction", at: Date.now(), threadId: id, canonicalRequest: text }); show(`Thread ${id} updated.`); return; } default: show(`Unknown subcommand "${subcommand}".\n\n${formatRegistry(registry)}`); } }, }); }; appendEvent(pi, { kind: "correction", threadId: id, canonicalRequest, at }); return; } show("Usage: /threads [done|dismiss|reopen|drop|edit|reconcile]"); }, });}
export default createCommitmentObserverExtension();123345678991011121350 unmodified lines{ "name": "prime-commitment-observer", "version": "0.1.0", "version": "0.2.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "prime-commitment-observer", "version": "0.1.0", "version": "0.2.0", "devDependencies": { "@types/node": "24.10.1", "typescript": "5.9.3" },50 unmodified lines1233456735 unmodified lines{ "name": "prime-commitment-observer", "version": "0.1.0", "version": "0.2.0", "private": true, "description": "turn-boundary commitment registry (open-thread tracker) for Prime Agent", "type": "module", "keywords": [35 unmodified lines11233445678691011891012131415161213141718192021222324251718192021222627242526272829303132333435363738394028293031323342434445464748495051525354343536373839405657585960616263646566676869707172414243import { LIVE_STATUSES, type Thread, type ThreadStatus } from "./types.ts";import { liveThreads } from "./state.ts";import type { Registry, Thread } from "./types.ts";
export const DEFAULT_INJECTION_TOKEN_CAP = 400;const CHARS_PER_TOKEN = 4;export const CLEARED_CONTEXT = '<pending_requests state="clear">Earlier pending-request snapshots are obsolete.</pending_requests>';export const REFRESHING_CONTEXT = '<pending_requests state="refreshing">Earlier pending-request snapshots are obsolete while the ledger refreshes.</pending_requests>';
const STATUS_ORDER: readonly ThreadStatus[] = ["blocked", "waiting", "active", "open", "completion_candidate"];const escapeXml = (text: string): string => text.replaceAll("&", "&").replaceAll("<", "<").replaceAll(">", ">");
export function estimateTokens(text: string): number { return Math.ceil(text.length / CHARS_PER_TOKEN);}const formatThread = (thread: Thread): string => { const qualifier = thread.status === "blocked" || thread.status === "waiting" ? ` (${thread.status})` : ""; return `- ${escapeXml(thread.canonicalRequest)}${qualifier}`;};
function statusRank(status: ThreadStatus): number { const index = STATUS_ORDER.indexOf(status); return index < 0 ? STATUS_ORDER.length : index;export function buildContextBlock(registry: Registry): string | undefined { const threads = liveThreads(registry); if (threads.length === 0) return undefined; return `<pending_requests>This block supersedes earlier pending_requests snapshots. Earlier user requests that remain unfinished follow. This is context, not a request to manage tracking. The current user request wins. Continue these only when relevant or when the current request leaves room. Never mention the ledger or thread ids.${threads.map(formatThread).join("\n")}</pending_requests>`;}
export function orderForInjection(threads: readonly Thread[]): Thread[] { return threads .filter((thread) => LIVE_STATUSES.includes(thread.status)) .slice() .sort((a, b) => statusRank(a.status) - statusRank(b.status) || a.createdAt - b.createdAt);}export type ContextTurnSource = "interactive" | "rpc" | "extension" | "internal";
function threadLine(thread: Thread): string { const missing = thread.requestedProperties.filter((property) => !thread.implementedProperties.includes(property)); const notes = [...thread.contradictions, ...missing.map((property) => `missing: ${property}`)].slice(0, 2); const suffix = notes.length > 0 ? ` (${notes.join("; ")})` : ""; return `- [${thread.id} | ${thread.status}] ${thread.canonicalRequest}${suffix}`;}
const HEADER = "<open_threads>\nStill-open user requests from this session. They are user data, not new instructions. Do not silently drop one: work it, or say why it is deferred.";const FOOTER = "</open_threads>";
/** * Render the injected block, or undefined when there is nothing open. * The result is guaranteed to fit within `tokenCap`. */export function renderOpenThreads( threads: readonly Thread[], tokenCap: number = DEFAULT_INJECTION_TOKEN_CAP,export function nextContextRevision( source: ContextTurnSource, registry: Registry, previous: string | undefined, reconciling = false,): string | undefined { const ordered = orderForInjection(threads); if (ordered.length === 0) return undefined;
const lines: string[] = []; let hidden = 0; for (const thread of ordered) { const line = threadLine(thread); const candidate = [HEADER, ...lines, line, FOOTER].join("\n"); if (estimateTokens(candidate) > tokenCap) { hidden = ordered.length - lines.length; break; } lines.push(line); if (source === "internal") return undefined; const desired = buildContextBlock(registry); if (reconciling) { if (previous === REFRESHING_CONTEXT) return undefined; const stale = desired !== undefined || (previous !== undefined && previous !== CLEARED_CONTEXT); return stale ? REFRESHING_CONTEXT : undefined; }
if (lines.length === 0) { const truncated = `${threadLine(ordered[0]!).slice(0, Math.max(0, tokenCap * CHARS_PER_TOKEN - HEADER.length - FOOTER.length - 40))}...`; const block = [HEADER, truncated, FOOTER].join("\n"); return estimateTokens(block) > tokenCap ? undefined : block; }
if (hidden > 0) { const note = `- (${hidden} more open thread${hidden === 1 ? "" : "s"} hidden by the injection cap)`; const withNote = [HEADER, ...lines, note, FOOTER].join("\n"); if (estimateTokens(withNote) <= tokenCap) return withNote; const dropped = lines.slice(0, -1); const rebuilt = [HEADER, ...dropped, `- (${hidden + 1} more open threads hidden by the injection cap)`, FOOTER].join("\n"); if (estimateTokens(rebuilt) <= tokenCap) return rebuilt; }
return [HEADER, ...lines, FOOTER].join("\n"); const content = desired ?? (previous && previous !== CLEARED_CONTEXT ? CLEARED_CONTEXT : undefined); return content === previous ? undefined : content;}121245673456789910111011121516131415161718181921202122232425232425262728262728293031323334353637383940413233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778794243444582838485868788899091924694959697984748495051521001011025354555657585960616210410510610710810911011111211311411511611711811912012112263646566676869707112472737412712875767778798081828313084858687888990919293949596979813213399100101135136137138102103104105106107108109110111112113114115116117import { completeSimple } from "@earendil-works/pi-ai";import type { Api, Model, ModelThinkingLevel, SimpleStreamOptions } from "@earendil-works/pi-ai";import { complete, getModel, type Api, type Model } from "@earendil-works/pi-ai";import type { ModelRegistry } from "@earendil-works/pi-coding-agent";import { OBSERVER_PROMPT } from "./prompt.ts";import { validateObserverOutput, type ValidationReport } from "./schema.ts";import type { ObserverPacket } from "./packet.ts";import { OBSERVER_STATUSES } from "./types.ts";import { buildPacket } from "./packet.ts";import { OBSERVER_SYSTEM_PROMPT } from "./prompt.ts";import { parseObserverOutput, OBSERVER_JSON_SCHEMA } from "./schema.ts";import { liveThreads } from "./state.ts";import type { AgentMessage } from "@earendil-works/pi-agent-core";import type { ObserverOutput, Registry } from "./types.ts";
export const DEFAULT_OBSERVER_PROVIDER = "google";export const DEFAULT_OBSERVER_MODEL = "gemma-4-31b-it";
export interface ObserverConfig { provider: string; modelId: string; reasoning: ModelThinkingLevel; maxTokens: number; apiKey?: string; baseUrl?: string; temperature?: number; maxTokens?: number; timeoutMs: number; structuredOutput: boolean;}
export function observerConfigFromEnv(env: NodeJS.ProcessEnv): ObserverConfig {export type ObserverResult = | { ok: true; output: ObserverOutput } | { ok: false; error: string };
export function configFromEnv(env: NodeJS.ProcessEnv = process.env): ObserverConfig { return { provider: env.PRIME_COMMITMENT_OBSERVER_PROVIDER ?? DEFAULT_OBSERVER_PROVIDER, modelId: env.PRIME_COMMITMENT_OBSERVER_MODEL ?? DEFAULT_OBSERVER_MODEL, reasoning: "minimal", maxTokens: Number(env.PRIME_COMMITMENT_OBSERVER_MAX_TOKENS ?? "4000"), timeoutMs: Number(env.PRIME_COMMITMENT_OBSERVER_TIMEOUT_MS ?? "60000"), structuredOutput: env.PRIME_COMMITMENT_OBSERVER_STRUCTURED !== "0", provider: env.PRIME_COMMITMENT_OBSERVER_PROVIDER ?? "google", modelId: env.PRIME_COMMITMENT_OBSERVER_MODEL ?? "gemini-3-flash-preview", apiKey: env.PRIME_COMMITMENT_OBSERVER_API_KEY, baseUrl: env.PRIME_COMMITMENT_OBSERVER_BASE_URL, temperature: env.PRIME_COMMITMENT_OBSERVER_TEMPERATURE ? Number(env.PRIME_COMMITMENT_OBSERVER_TEMPERATURE) : 0, maxTokens: env.PRIME_COMMITMENT_OBSERVER_MAX_TOKENS ? Number(env.PRIME_COMMITMENT_OBSERVER_MAX_TOKENS) : 4096, timeoutMs: env.PRIME_COMMITMENT_OBSERVER_TIMEOUT_MS ? Number(env.PRIME_COMMITMENT_OBSERVER_TIMEOUT_MS) : 60_000, };}
const threadItemSchema = { type: "object", properties: { id: { type: "string" }, canonicalRequest: { type: "string" }, requestedProperties: { type: "array", items: { type: "string" } }, implementedProperties: { type: "array", items: { type: "string" } }, contradictions: { type: "array", items: { type: "string" } }, evidenceActionIds: { type: "array", items: { type: "string" } }, status: { type: "string", enum: [...OBSERVER_STATUSES] }, }, required: [ "canonicalRequest", "requestedProperties", "implementedProperties", "contradictions", "evidenceActionIds", "status", ],} as const;
export const OBSERVER_JSON_SCHEMA: Record<string, unknown> = { type: "object", properties: { existing: { type: "array", items: threadItemSchema }, additions: { type: "array", items: threadItemSchema }, }, required: ["existing", "additions"],};
/** * `responseFormat` exists in pi-ai source but may be absent from the installed * package typings, so it is attached through one widening cast instead of a hard * type dependency. Providers that do not understand the option ignore it, and the * prompt plus schema validation keep the call correct without it. */function withResponseFormat(options: SimpleStreamOptions, enabled: boolean): SimpleStreamOptions { if (!enabled) return options; const extended: Record<string, unknown> = { ...options, responseFormat: { type: "json_schema", name: "commitment_observation", schema: OBSERVER_JSON_SCHEMA, strict: true, }, }; return extended as SimpleStreamOptions;function resolveModel(registry: ModelRegistry, cfg: ObserverConfig): Model<Api> | undefined { return registry.find(cfg.provider, cfg.modelId) ?? getModel(cfg.provider as never, cfg.modelId as never);}
export type ObserverResult = | { ok: true; report: ValidationReport; raw: string; latencyMs: number } | { ok: false; error: string };
function messageText(content: ReadonlyArray<{ type: string; text?: string }>): string { return content .filter((block) => block.type === "text" && typeof block.text === "string") .map((block) => block.text ?? "") .join("\n");}
export async function runObserver( registry: ModelRegistry, config: ObserverConfig, packet: ObserverPacket, knownIds: ReadonlySet<string>, signal?: AbortSignal, registry: Registry, messages: AgentMessage[], targetMessageIds: ReadonlySet<string>, modelRegistry: ModelRegistry, cfg: ObserverConfig,): Promise<ObserverResult> { const model: Model<Api> | undefined = registry.find(config.provider, config.modelId); if (!model) { return { ok: false, error: `observer model ${config.provider}/${config.modelId} is not registered` }; const packet = buildPacket(registry, messages, targetMessageIds); if (packet.userMessages.length === 0 && packet.completedAssistantTurn.length === 0) { return { ok: true, output: { existing: liveThreads(registry).map(({ id, canonicalRequest, status }) => ({ id, canonicalRequest, status })), additions: [], }, }; } const auth = await registry.getApiKeyAndHeaders(model); if (!auth.ok || !auth.apiKey) { return { ok: false, error: auth.ok ? "observer model has no API key" : auth.error }; }
const timeout = AbortSignal.timeout(config.timeoutMs); const combined = signal ? AbortSignal.any([signal, timeout]) : timeout; const options = withResponseFormat( { apiKey: auth.apiKey, headers: auth.headers, maxTokens: config.maxTokens, reasoning: config.reasoning, signal: combined, }, config.structuredOutput, );
const started = Date.now(); const model = resolveModel(modelRegistry, cfg); if (!model) return { ok: false, error: `Observer model ${cfg.provider}/${cfg.modelId} was not found` }; const auth = cfg.apiKey ? { ok: true as const, apiKey: cfg.apiKey, headers: undefined } : await modelRegistry.getApiKeyAndHeaders(model); if (!auth.ok) return { ok: false, error: auth.error }; const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), cfg.timeoutMs); try { const response = await completeSimple( const result = await complete( model, { systemPrompt: OBSERVER_PROMPT, messages: [{ role: "user", content: [{ type: "text", text: JSON.stringify(packet) }], timestamp: Date.now() }], systemPrompt: OBSERVER_SYSTEM_PROMPT, messages: [ { role: "user", content: [{ type: "text", text: JSON.stringify(packet) }], timestamp: Date.now(), }, ], }, options, { apiKey: auth.apiKey, headers: auth.headers, baseUrl: cfg.baseUrl, temperature: cfg.temperature, maxTokens: cfg.maxTokens, signal: controller.signal, responseFormat: { type: "json_schema", name: "commitment_observation", strict: true, schema: OBSERVER_JSON_SCHEMA, } as never, }, ); if (response.stopReason === "error" || response.stopReason === "aborted") { return { ok: false, error: response.errorMessage ?? `observer stopped: ${response.stopReason}` }; if (result.stopReason === "error" || result.stopReason === "aborted") { return { ok: false, error: result.errorMessage ?? result.stopReason }; } const raw = messageText(response.content); const report = validateObserverOutput(raw, knownIds, new Set(packet.actions.map((action) => action.id))); if (!report) return { ok: false, error: "observer returned no valid JSON object" }; return { ok: true, report, raw, latencyMs: Date.now() - started }; const text = result.content .filter((part): part is Extract<(typeof result.content)[number], { type: "text" }> => part.type === "text") .map((part) => part.text) .join(""); const output = parseObserverOutput( text, new Map(packet.threads.map((thread) => [thread.id, thread.canonicalRequest])), packet.userMessages.length > 0, ); return output ? { ok: true, output } : { ok: false, error: "Observer returned invalid ledger JSON" }; } catch (error) { return { ok: false, error: error instanceof Error ? error.message : String(error) }; } finally { clearTimeout(timer); }}