From 170cceb42f4a4a55066a9c262cac4a9e8584cdf1 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Thu, 10 Sep 2026 22:39:29 +0300 Subject: [PATCH] src,index.ts,skills/threads,test: simplify commitment ledger --- README.md | 202 ++------- index.ts | 435 ++++++++---------- package-lock.json | 4 +- package.json | 2 +- skills/threads/SKILL.md | 23 - skills/threads/pyproject.toml | 13 - skills/threads/src/threads/__init__.py | 151 ------- src/inject.ts | 104 ++--- src/observer.ts | 205 ++++----- src/packet.ts | 110 +---- src/prompt.ts | 45 +- src/schema.ts | 187 ++++---- src/state.ts | 351 +++++---------- src/types.ts | 84 +--- test/extension.test.ts | 50 ++ test/live-observer.ts | 61 ++- test/logic.test.ts | 601 ++++++------------------- 17 files changed, 821 insertions(+), 1807 deletions(-) delete mode 100644 skills/threads/SKILL.md delete mode 100644 skills/threads/pyproject.toml delete mode 100644 skills/threads/src/threads/__init__.py create mode 100644 test/extension.test.ts diff --git a/README.md b/README.md index ec990db..4b961b8 100644 --- a/README.md +++ b/README.md @@ -1,202 +1,66 @@ # 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 problem -if 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 it -this 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`. -## requirements +Before a user-originated turn, Prime receives one hidden `` 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. + +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. -- 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) +## requirements -extensions run with your full user permissions. review the source before installing it. +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 +```bash git clone git@knot.gaze.systems:did:plc:gyggh67fdq2hukuabczwhm3w prime-commitment-observer cd prime-commitment-observer prime-agent package install . ``` -`package install .` records the checkout in `~/.prime/agent/settings.json`. keep the checkout in -place 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" } - ] -} +```bash +export PRIME_COMMITMENT_OBSERVER_PROVIDER=google +export PRIME_COMMITMENT_OBSERVER_MODEL=gemini-3-flash-preview +export PRIME_COMMITMENT_OBSERVER_TIMEOUT_MS=60000 ``` -Prime also auto-discovers extension directories under `~/.prime/agent/extensions/`: +Optional variables: -```sh -git 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 explicit user acceptance -> done -/threads dismiss drop the thread as not wanted -/threads reopen put a closed thread back to open -/threads drop remove a wrongly created thread entirely -/threads edit correct the canonical request -/threads reconcile run the observer now -``` - -### for agents (python skill) - -agents can inspect and control threads programmatically from the python REPL using the bundled `threads` skill: - -```python -threads.list() # list open and closed threads -threads.complete("t1", reason="done in test") # mark thread complete -threads.dismiss("t2", reason="obsolete") # dismiss unwanted thread -threads.reopen("t3") # reopen a closed thread +/threads done +/threads dismiss [reason] +/threads reopen +/threads drop +/threads edit +/threads reconcile ``` -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 weak -models 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 `, 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 the -turn blocks promotion for all threads. - -auto promotion is visible and reversible: each one raises a notification, `/threads` shows -the promotion reason (`` or ``), and `/threads reopen ` 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 `` custom message -with `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. +The foreground model is not given thread ids and has no `threads` skill. ## development -```sh -npm install -npm test # pure logic tests, no provider calls (24 tests) +```bash +npm test npm 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. diff --git a/index.ts b/index.ts index 30c2d39..e698984 100644 --- a/index.ts +++ b/index.ts @@ -1,291 +1,236 @@ import { 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; - -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); -} +type BeforeAgentStartWithSource = BeforeAgentStartEvent & { source: InputSource | "internal" }; 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); - - let pendingActions: ActionRecord[] = []; - let failureStreak = 0; - let queue: Promise = Promise.resolve(); - - return (pi: ExtensionAPI): void => { - const appendEvent = (event: CommitmentEvent): void => { - pi.appendEntry(COMMITMENT_EVENT_TYPE, event); - }; - - 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 reconcile = async (ctx: ExtensionContext, messages: PacketMessage[]): Promise => { - 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)); +export default function commitmentObserver(pi: ExtensionAPI): void { + const cfg = configFromEnv(); + let observerQueue = Promise.resolve(); + let pendingReconciliations = 0; + let consecutiveFailures = 0; + + 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"); + } + }; - 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 branchContains = (ctx: ExtensionContext, entryId: string | null): boolean => + entryId === null || ctx.sessionManager.getBranch().some((entry) => entry.id === entryId); + + const liveRevision = (registry: Registry): string => + JSON.stringify( + liveThreads(registry).map(({ id, canonicalRequest, status, updatedAt }) => ({ + id, + canonicalRequest, + status, + updatedAt, + })), + ); - 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, + boundaryEntryId: string | null, + ): Promise => { + 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 => { + 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 => { - 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 => { - 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)); - 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)}`); + 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 "); 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 "); - 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(); diff --git a/package-lock.json b/package-lock.json index b13e118..c5ceeaa 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "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" diff --git a/package.json b/package.json index fffe2ca..e8c700d 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "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", diff --git a/skills/threads/SKILL.md b/skills/threads/SKILL.md deleted file mode 100644 index 244a960..0000000 --- a/skills/threads/SKILL.md +++ /dev/null @@ -1,23 +0,0 @@ ---- -name: threads -description: Inspect and update tracked open user requests (threads) from the Python REPL. Use to inspect open tasks, mark a finished task complete, or dismiss an obsolete task. ---- - -# Threads - -The commitment observer tracks open user requests across turns so multi-part requests are never forgotten. -Use this skill to inspect and update tracked threads directly from Python: - -```python -threads.list() -threads.complete("t1") -threads.dismiss("t2") -threads.reopen("t3") -``` - -## API - -- `threads.list()` — list all open and closed threads as structured dicts (`id`, `status`, `canonicalRequest`, `missing`, etc.). -- `threads.complete(thread_id, reason=None)` — mark a thread as completed (`status: done`). -- `threads.dismiss(thread_id, reason=None)` — dismiss an obsolete or unwanted thread (`status: dismissed`). -- `threads.reopen(thread_id)` — reopen a closed thread (`status: open`). diff --git a/skills/threads/pyproject.toml b/skills/threads/pyproject.toml deleted file mode 100644 index f800f70..0000000 --- a/skills/threads/pyproject.toml +++ /dev/null @@ -1,13 +0,0 @@ -[project] -name = "threads" -version = "0.1.0" -description = "Prime Agent threads skill: inspect and update commitment threads from Python" -requires-python = ">=3.10" -dependencies = [] - -[build-system] -requires = ["hatchling"] -build-backend = "hatchling.build" - -[tool.hatch.build.targets.wheel] -packages = ["src/threads"] diff --git a/skills/threads/src/threads/__init__.py b/skills/threads/src/threads/__init__.py deleted file mode 100644 index a4bb139..0000000 --- a/skills/threads/src/threads/__init__.py +++ /dev/null @@ -1,151 +0,0 @@ -"""Prime Agent threads skill: manage tracked commitment threads from Python.""" - -from __future__ import annotations - -import json -import os -import pathlib -import time -import uuid -from typing import Any - - -def _get_session_path() -> pathlib.Path | None: - sess_dir = os.environ.get("RLM_SESSION_DIR") - if sess_dir: - sid = pathlib.Path(sess_dir).name - sp = pathlib.Path.home() / f".prime/agent/sessions/{sid}.jsonl" - if sp.exists(): - return sp - sessions = sorted((pathlib.Path.home() / ".prime/agent/sessions").glob("*.jsonl"), key=lambda p: p.stat().st_mtime) - return sessions[-1] if sessions else None - - -def _append_event(event_data: dict[str, Any]) -> None: - sp = _get_session_path() - if not sp or not sp.exists(): - raise RuntimeError("No active session file found") - - entry = { - "type": "custom", - "customType": "commitment-event", - "data": event_data, - "id": str(uuid.uuid4())[:8], - "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), - } - with open(sp, "a", encoding="utf-8") as f: - f.write(json.dumps(entry) + "\n") - - -def _read_threads() -> list[dict[str, Any]]: - sp = _get_session_path() - if not sp or not sp.exists(): - return [] - - events = [] - with open(sp, "r", encoding="utf-8", errors="ignore") as f: - for line in f: - try: - row = json.loads(line) - if row.get("type") == "custom" and row.get("customType") == "commitment-event": - events.append(row["data"]) - except Exception: - pass - - threads = [] - by_id = {} - for ev in events: - kind = ev.get("kind") - if kind == "observation": - out = ev.get("output", {}) - for existing in out.get("existing", []): - tid = existing.get("id") - if tid in by_id: - th = by_id[tid] - for k in ["canonicalRequest", "requestedProperties", "implementedProperties", "contradictions", "evidenceActionIds", "status"]: - if k in existing and existing[k] is not None: - th[k] = existing[k] - th["updatedAt"] = ev.get("at", int(time.time() * 1000)) - for add in out.get("additions", []): - tid = f"t{len(threads) + 1}" - new_th = { - "id": tid, - "canonicalRequest": add.get("canonicalRequest", ""), - "requestedProperties": add.get("requestedProperties", []), - "implementedProperties": add.get("implementedProperties", []), - "contradictions": add.get("contradictions", []), - "evidenceActionIds": add.get("evidenceActionIds", []), - "status": add.get("status", "open"), - "createdAt": ev.get("at", int(time.time() * 1000)), - "updatedAt": ev.get("at", int(time.time() * 1000)), - } - threads.append(new_th) - by_id[tid] = new_th - elif kind == "user_action": - tid = ev.get("threadId") - act = ev.get("action") - if tid in by_id: - th = by_id[tid] - if act == "accept": - th["status"] = "done" - th["promotedBy"] = "user_accepted" - elif act == "dismiss": - th["status"] = "dismissed" - elif act == "reopen": - th["status"] = "open" - elif act == "drop": - threads = [t for t in threads if t["id"] != tid] - del by_id[tid] - if tid in by_id: - by_id[tid]["updatedAt"] = ev.get("at", int(time.time() * 1000)) - - return threads - - -def list(status: str | None = None) -> list[dict[str, Any]]: - """List tracked commitment threads. Optional filter by status ('open', 'active', 'done', etc.).""" - all_threads = _read_threads() - if status is not None: - return [t for t in all_threads if t["status"] == status] - return all_threads - - -def complete(thread_id: str, reason: str | None = None) -> dict[str, Any]: - """Mark a thread complete (status: done).""" - if not isinstance(thread_id, str): - raise TypeError(f"thread_id must be str, got {type(thread_id).__name__}") - _append_event({ - "kind": "user_action", - "action": "accept", - "threadId": thread_id, - "reason": reason, - "at": int(time.time() * 1000), - }) - return {"status": "ok", "threadId": thread_id, "action": "accept"} - - -def dismiss(thread_id: str, reason: str | None = None) -> dict[str, Any]: - """Dismiss an obsolete or unwanted thread (status: dismissed).""" - if not isinstance(thread_id, str): - raise TypeError(f"thread_id must be str, got {type(thread_id).__name__}") - _append_event({ - "kind": "user_action", - "action": "dismiss", - "threadId": thread_id, - "reason": reason, - "at": int(time.time() * 1000), - }) - return {"status": "ok", "threadId": thread_id, "action": "dismiss"} - - -def reopen(thread_id: str) -> dict[str, Any]: - """Reopen a closed thread (status: open).""" - if not isinstance(thread_id, str): - raise TypeError(f"thread_id must be str, got {type(thread_id).__name__}") - _append_event({ - "kind": "user_action", - "action": "reopen", - "threadId": thread_id, - "at": int(time.time() * 1000), - }) - return {"status": "ok", "threadId": thread_id, "action": "reopen"} diff --git a/src/inject.ts b/src/inject.ts index 6cfced1..2de6d37 100644 --- a/src/inject.ts +++ b/src/inject.ts @@ -1,73 +1,43 @@ -import { LIVE_STATUSES, type Thread, type ThreadStatus } from "./types.ts"; - -export const DEFAULT_INJECTION_TOKEN_CAP = 400; -const CHARS_PER_TOKEN = 4; - -const STATUS_ORDER: readonly ThreadStatus[] = ["blocked", "waiting", "active", "open", "completion_candidate"]; - -export function estimateTokens(text: string): number { - return Math.ceil(text.length / CHARS_PER_TOKEN); -} - -function statusRank(status: ThreadStatus): number { - const index = STATUS_ORDER.indexOf(status); - return index < 0 ? STATUS_ORDER.length : index; -} - -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); +import { liveThreads } from "./state.ts"; +import type { Registry, Thread } from "./types.ts"; + +export const CLEARED_CONTEXT = + 'Earlier pending-request snapshots are obsolete.'; +export const REFRESHING_CONTEXT = + 'Earlier pending-request snapshots are obsolete while the ledger refreshes.'; + +const escapeXml = (text: string): string => + text.replaceAll("&", "&").replaceAll("<", "<").replaceAll(">", ">"); + +const formatThread = (thread: Thread): string => { + const qualifier = thread.status === "blocked" || thread.status === "waiting" ? ` (${thread.status})` : ""; + return `- ${escapeXml(thread.canonicalRequest)}${qualifier}`; +}; + +export function buildContextBlock(registry: Registry): string | undefined { + const threads = liveThreads(registry); + if (threads.length === 0) return undefined; + return ` +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")} +`; } -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}`; -} +export type ContextTurnSource = "interactive" | "rpc" | "extension" | "internal"; -const HEADER = "\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 = ""; - -/** - * 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 (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 (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 (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; } diff --git a/src/observer.ts b/src/observer.ts index 554a61c..6776be6 100644 --- a/src/observer.ts +++ b/src/observer.ts @@ -1,142 +1,117 @@ -import { 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"; - -export const DEFAULT_OBSERVER_PROVIDER = "google"; -export const DEFAULT_OBSERVER_MODEL = "gemma-4-31b-it"; +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 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 = { - 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 = { - ...options, - responseFormat: { - type: "json_schema", - name: "commitment_observation", - schema: OBSERVER_JSON_SCHEMA, - strict: true, - }, - }; - return extended as SimpleStreamOptions; -} - -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"); +function resolveModel(registry: ModelRegistry, cfg: ObserverConfig): Model | undefined { + return registry.find(cfg.provider, cfg.modelId) ?? getModel(cfg.provider as never, cfg.modelId as never); } export async function runObserver( - registry: ModelRegistry, - config: ObserverConfig, - packet: ObserverPacket, - knownIds: ReadonlySet, - signal?: AbortSignal, + registry: Registry, + messages: AgentMessage[], + targetMessageIds: ReadonlySet, + modelRegistry: ModelRegistry, + cfg: ObserverConfig, ): Promise { - const model: Model | undefined = registry.find(config.provider, config.modelId); - if (!model) { - return { ok: false, error: `observer model ${config.provider}/${config.modelId} is not registered` }; - } - 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 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 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(), + }, + ], + }, + { + 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, }, - options, ); - 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); } } diff --git a/src/packet.ts b/src/packet.ts index af29597..8b150d5 100644 --- a/src/packet.ts +++ b/src/packet.ts @@ -1,98 +1,24 @@ -import type { ActionRecord, Thread, UserMessageRecord } from "./types.ts"; +import type { AgentMessage } from "@earendil-works/pi-agent-core"; +import { liveThreads } from "./state.ts"; +import type { ObserverPacket, Registry } from "./types.ts"; + +const assistantText = (messages: AgentMessage[]): string => + messages + .flatMap((message) => + message.role === "assistant" + ? message.content.filter((part) => part.type === "text").map((part) => part.text) + : [], + ) + .join("\n\n"); -/** Minimal structural view of a session message, so packet building stays host-independent. */ -export interface PacketMessage { - role: "user" | "assistant" | "toolResult" | "other"; - text: string; - toolName?: string; - isError?: boolean; -} - -/** Action metadata as the observer sees it. `id` is the only handle for citing evidence. */ -export interface PacketAction { - id: string; - label: string; - required: boolean; - ok: boolean; -} - -export interface ObserverPacket { - threads: Array<{ - id: string; - canonicalRequest: string; - status: string; - requestedProperties?: readonly string[]; - implementedProperties?: readonly string[]; - }>; - precedingAssistant: string; - userMessages: string[]; - completedAssistantTurn: string; - actions: PacketAction[]; -} - -const MAX_ASSISTANT_CHARS = 6000; -const MAX_USER_CHARS = 6000; -const MAX_ACTIONS = 25; -const MAX_LABEL_CHARS = 120; - -function clip(text: string, max: number): string { - return text.length <= max ? text : `${text.slice(0, max)}...[truncated]`; -} - -/** Mint packet-stable action ids. Ids mean nothing outside the call they were built for. */ -export function toPacketActions(actions: readonly ActionRecord[]): PacketAction[] { - return actions.slice(-MAX_ACTIONS).map((action, index) => ({ - id: `a${index + 1}`, - label: action.label.slice(0, MAX_LABEL_CHARS), - required: action.required, - ok: action.ok, - })); -} - -/** Re-key recorded actions with the same ids the observer saw, for the promotion gate. */ -export function alignActionIds(actions: readonly ActionRecord[]): ActionRecord[] { - return actions.slice(-MAX_ACTIONS).map((action, index) => ({ ...action, id: `a${index + 1}` })); -} - -/** - * Build the observer packet: prior thread state, the preceding assistant message, - * the unreconciled user messages, the completed assistant turn and identified actions. - */ export function buildPacket( - messages: readonly PacketMessage[], - unreconciled: readonly UserMessageRecord[], - threads: readonly Thread[], - actions: readonly ActionRecord[], + registry: Registry, + messages: AgentMessage[], + targetMessageIds: ReadonlySet, ): ObserverPacket { - const lastAssistantIndex = messages.map((message) => message.role).lastIndexOf("assistant"); - const completedAssistantTurn = lastAssistantIndex >= 0 ? messages[lastAssistantIndex]!.text : ""; - - const lastUserIndex = messages.map((message) => message.role).lastIndexOf("user"); - const precedingAssistant = - messages - .slice(0, Math.max(0, lastUserIndex)) - .filter((message) => message.role === "assistant") - .at(-1)?.text ?? ""; - return { - threads: threads.map((thread) => ({ - id: thread.id, - canonicalRequest: thread.canonicalRequest, - status: thread.status, - requestedProperties: thread.requestedProperties, - implementedProperties: thread.implementedProperties, - })), - precedingAssistant: clip(precedingAssistant, MAX_ASSISTANT_CHARS), - userMessages: unreconciled.map((message) => clip(message.text, MAX_USER_CHARS)), - completedAssistantTurn: clip(completedAssistantTurn, MAX_ASSISTANT_CHARS), - actions: toPacketActions(actions), + threads: liveThreads(registry), + userMessages: registry.unreconciled.filter((message) => targetMessageIds.has(message.messageId)), + completedAssistantTurn: assistantText(messages), }; } - -const VALIDATION_PATTERN = - /\b(npm (?:run )?(?:check|test)|yarn (?:check|test)|pnpm (?:check|test)|cargo (?:test|check|clippy)|pytest|vitest|jest|tsc|go test|biome check|eslint)\b/; - -/** Whether a shell/ipython command counts as a required validation for promotion. */ -export function isRequiredValidation(command: string): boolean { - return VALIDATION_PATTERN.test(command); -} diff --git a/src/prompt.ts b/src/prompt.ts index 29beecc..cedd611 100644 --- a/src/prompt.ts +++ b/src/prompt.ts @@ -1,31 +1,20 @@ -export const OBSERVER_PROMPT = `You are a turn-boundary state observer for a coding agent. Update thread state after exactly one completed agent turn. +export const OBSERVER_SYSTEM_PROMPT = `You maintain a lightweight ledger of unfinished user requests after one completed agent turn. -Perform these operations explicitly: -1. Resolve user references such as yes, that, second, other, it, and this against the preceding assistant and existing threads. -2. Split the user message into atomic requests. An atomic request has its own independently verifiable completion condition. Clauses joined by and, also, then, bullets, or separate sentences must be separate whenever one can finish while another remains. -3. Map each user instruction to either one existing thread or one new thread, never both. Preserve an existing id for refinements, corrections, resumption, or reopening. -4. Write an explicit canonical request and its atomic requested properties. -5. Extract implemented properties from the completed assistant and action metadata. List contradictions and missing requested properties. -6. Cite evidence. For every thread set evidenceActionIds to the ids of packet actions that bear on that thread's completion. -7. Assign status: open, active, blocked, waiting, completion_candidate, cancelled, or superseded. Never emit done. When the request appears satisfied, emit completion_candidate and let the host verify and promote it. +Return JSON matching the supplied schema. Rules: -- Additions come only from user instructions. Never create a thread from tool activity, an assistant suggestion, or a failed optional check. -- Do not create work from incidental facts or hypotheticals explicitly deferred. -- Use cancelled when the user abandoned the request, superseded when the user replaced it with a different request, and done only when the original request itself was satisfied. -- Use blocked when a required validation failed or an external dependency stops progress; use open only when no work has started. -- Keep unresolved sibling requests separate and open. -- Assistant claims such as done or implemented never override a semantic mismatch or failed relevant validation. -- A wrong option is not done even if tools pass. -- A failed test or check prevents done unless the packet explicitly proves it was unrelated. -- waiting means user input is required; blocked means work cannot currently proceed. -- Return every existing id exactly once. -- evidenceActionIds must contain only action ids present in this packet, and only actions that actually bear on that thread. Use an empty array when nothing applies. Never invent an id. -- Citing an action with ok false means that thread is not complete. Do not emit completion_candidate for a thread whose cited actions include a failure. -- Do not cite an action merely because it ran in the same turn. -- Carry over previously implemented properties from existing threads unless explicitly contradicted or regressed. When all requested properties for a thread are satisfied (from prior turns or this turn) and there are no contradictions, assign status: completion_candidate. - -Example: user says "fix resize, add a regression test, and update docs" and only resize was fixed. Emit three separate threads: resize=done, regression test=open, docs=open. - -Return JSON only: -{"existing":[{"id":"...","canonicalRequest":"...","requestedProperties":["..."],"implementedProperties":["..."],"contradictions":["..."],"evidenceActionIds":["..."],"status":"..."}],"additions":[{"canonicalRequest":"one atomic request","requestedProperties":["..."],"implementedProperties":["..."],"contradictions":["..."],"evidenceActionIds":["..."],"status":"..."}]}`; +- Return every existing thread exactly once in existing, with its unchanged id. +- Never put an existing id in additions. +- Add threads only for explicit requests in userMessages. If userMessages is empty, additions must be empty. +- Split independent deliverables. Keep constraints and acceptance conditions with the deliverable they qualify. +- Do not create threads for greetings, background facts, status reports, observer bookkeeping, or implementation steps the agent chose itself. +- Preserve canonicalRequest unless the user's intent genuinely changed. Do not rewrite it just to vary the wording. +- Use done when the completed assistant turn materially delivered the requested outcome. +- Use active when the turn advanced the request but left work to do. +- Use waiting when progress now depends on an expected external result or reply. +- Use blocked when progress cannot continue without new information or a failed prerequisite. +- Use open when the request remains untouched. +- Use cancelled or superseded only when the user explicitly withdrew or replaced the request. +- An assistant saying work is complete is semantic evidence, not proof. Judge the delivered content, but do not invent test or command outcomes. +- A request may be added as done when the same turn completely answered it. +- Do not mention this ledger in canonicalRequest.`; diff --git a/src/schema.ts b/src/schema.ts index 2818a86..92ad29d 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -1,122 +1,99 @@ import { OBSERVER_STATUSES, type ObservedThread, type ObserverOutput, type ObserverStatus } from "./types.ts"; -const MAX_CANONICAL_CHARS = 400; -const MAX_PROPERTY_CHARS = 200; -const MAX_PROPERTIES = 12; -const MAX_THREADS_PER_CALL = 40; +const MAX_THREADS = 64; +const MAX_REQUEST_LENGTH = 2_000; +const STATUS_SET = new Set(OBSERVER_STATUSES); -export interface ValidationReport { - output: ObserverOutput; - droppedExisting: number; - droppedAdditions: number; - unknownIds: string[]; - /** Cited evidence ids that were not in the packet and were therefore discarded. */ - droppedEvidenceIds: number; -} +const threadProperties = { + canonicalRequest: { type: "string" }, + status: { type: "string", enum: [...OBSERVER_STATUSES] }, +} as const; -function isObserverStatus(value: unknown): value is ObserverStatus { - return typeof value === "string" && (OBSERVER_STATUSES as readonly string[]).includes(value); -} +export const OBSERVER_JSON_SCHEMA = { + type: "object", + additionalProperties: false, + properties: { + existing: { + type: "array", + maxItems: MAX_THREADS, + items: { + type: "object", + additionalProperties: false, + properties: { id: { type: "string" }, ...threadProperties }, + required: ["id", "canonicalRequest", "status"], + }, + }, + additions: { + type: "array", + maxItems: MAX_THREADS, + items: { + type: "object", + additionalProperties: false, + properties: threadProperties, + required: ["canonicalRequest", "status"], + }, + }, + }, + required: ["existing", "additions"], +} as const; -function stringList(value: unknown): string[] { - if (!Array.isArray(value)) return []; - return value - .filter((item): item is string => typeof item === "string") - .map((item) => item.trim().slice(0, MAX_PROPERTY_CHARS)) - .filter((item) => item.length > 0) - .slice(0, MAX_PROPERTIES); -} +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); -function parseThread( - value: unknown, - knownIds: ReadonlySet, - knownActionIds: ReadonlySet, - requireId: boolean, -): { thread: ObservedThread; droppedEvidenceIds: number } | undefined { - if (!value || typeof value !== "object") return undefined; - const record = value as Record; - if (!isObserverStatus(record.status)) return undefined; - const canonicalRequest = typeof record.canonicalRequest === "string" ? record.canonicalRequest.trim() : ""; - if (canonicalRequest.length === 0) return undefined; - const id = typeof record.id === "string" ? record.id.trim() : undefined; - if (requireId && (id === undefined || !knownIds.has(id))) return undefined; - const citedRaw = Array.isArray(record.evidenceActionIds) - ? record.evidenceActionIds.filter((item): item is string => typeof item === "string").map((item) => item.trim()) - : []; - const cited = [...new Set(citedRaw.filter((item) => knownActionIds.has(item)))].slice(0, MAX_PROPERTIES); +function parseThread(value: unknown, requireId: boolean): ObservedThread | undefined { + if (!isRecord(value)) return undefined; + const keys = Object.keys(value); + const allowed = requireId ? new Set(["id", "canonicalRequest", "status"]) : new Set(["canonicalRequest", "status"]); + if (keys.some((key) => !allowed.has(key))) return undefined; + if ( + typeof value.canonicalRequest !== "string" || + value.canonicalRequest.trim().length === 0 || + value.canonicalRequest.length > MAX_REQUEST_LENGTH || + typeof value.status !== "string" || + !STATUS_SET.has(value.status) || + (requireId && typeof value.id !== "string") + ) { + return undefined; + } return { - thread: { - id: requireId ? id : undefined, - canonicalRequest: canonicalRequest.slice(0, MAX_CANONICAL_CHARS), - requestedProperties: stringList(record.requestedProperties), - implementedProperties: stringList(record.implementedProperties), - contradictions: stringList(record.contradictions), - evidenceActionIds: cited, - status: record.status, - }, - droppedEvidenceIds: citedRaw.length - cited.length, + ...(requireId ? { id: value.id as string } : {}), + canonicalRequest: value.canonicalRequest.trim(), + status: value.status as ObserverStatus, }; } -/** Strip markdown fences and locate the outermost JSON object. */ -export function extractJsonObject(raw: string): unknown { - const text = raw.trim().replace(/^```(?:json)?/i, "").replace(/```$/, "").trim(); - const start = text.indexOf("{"); - const end = text.lastIndexOf("}"); - if (start < 0 || end <= start) return undefined; +export function parseObserverOutput( + raw: string, + knownThreads: ReadonlyMap, + allowUserChanges: boolean, +): ObserverOutput | undefined { + let parsed: unknown; try { - return JSON.parse(text.slice(start, end + 1)) as unknown; + parsed = JSON.parse(raw); } catch { return undefined; } -} - -/** - * Validate raw observer output. Invalid entries are dropped rather than trusted; - * an unparseable payload degrades to "no update" (undefined). - */ -export function validateObserverOutput( - raw: string, - knownIds: ReadonlySet, - knownActionIds: ReadonlySet = new Set(), -): ValidationReport | undefined { - const parsed = extractJsonObject(raw); - if (!parsed || typeof parsed !== "object") return undefined; - const record = parsed as Record; - const rawExisting = Array.isArray(record.existing) ? record.existing.slice(0, MAX_THREADS_PER_CALL) : []; - const rawAdditions = Array.isArray(record.additions) ? record.additions.slice(0, MAX_THREADS_PER_CALL) : []; - if (!Array.isArray(record.existing) && !Array.isArray(record.additions)) return undefined; - - const seen = new Set(); - const existing: ObservedThread[] = []; - const unknownIds: string[] = []; - let droppedEvidenceIds = 0; - let droppedExisting = 0; - for (const candidate of rawExisting) { - const parsedThread = parseThread(candidate, knownIds, knownActionIds, true); - const thread = parsedThread?.thread; - if (!thread || thread.id === undefined || seen.has(thread.id)) { - droppedExisting += 1; - const id = (candidate as Record | null)?.id; - if (typeof id === "string" && !knownIds.has(id)) unknownIds.push(id); - continue; - } - seen.add(thread.id); - droppedEvidenceIds += parsedThread?.droppedEvidenceIds ?? 0; - existing.push(thread); + if (!isRecord(parsed) || !Array.isArray(parsed.existing) || !Array.isArray(parsed.additions)) return undefined; + if (parsed.existing.length > MAX_THREADS || parsed.additions.length > MAX_THREADS) return undefined; + if (!allowUserChanges && parsed.additions.length > 0) return undefined; + const existing = parsed.existing.map((value) => parseThread(value, true)); + const additions = parsed.additions.map((value) => parseThread(value, false)); + if (existing.some((thread) => thread === undefined) || additions.some((thread) => thread === undefined)) { + return undefined; } - - const additions: ObservedThread[] = []; - let droppedAdditions = 0; - for (const candidate of rawAdditions) { - const parsedThread = parseThread(candidate, knownIds, knownActionIds, false); - if (!parsedThread) { - droppedAdditions += 1; - continue; - } - droppedEvidenceIds += parsedThread.droppedEvidenceIds; - additions.push(parsedThread.thread); + const ids = existing.map((thread) => thread!.id!); + if (new Set(ids).size !== ids.length || ids.some((id) => !knownThreads.has(id))) return undefined; + if (ids.length !== knownThreads.size || [...knownThreads.keys()].some((id) => !ids.includes(id))) return undefined; + if ( + !allowUserChanges && + (existing as ObservedThread[]).some( + (thread) => thread.canonicalRequest !== knownThreads.get(thread.id!), + ) + ) { + return undefined; } - - return { output: { existing, additions }, droppedExisting, droppedAdditions, unknownIds, droppedEvidenceIds }; + return { + existing: existing as ObservedThread[], + additions: additions as ObservedThread[], + }; } diff --git a/src/state.ts b/src/state.ts index a2205c2..a30eea2 100644 --- a/src/state.ts +++ b/src/state.ts @@ -1,263 +1,160 @@ -import { - HOST_FINAL_STATUSES, - LIVE_STATUSES, - type CommitmentEvent, - type ObservedThread, - type PromotionReason, - type ActionRecord, - type Registry, - type ReplayOptions, - type Thread, - type UserMessageRecord, -} from "./types.ts"; +import { OBSERVER_STATUSES, type CommitmentEvent, type ObservedThread, type ObserverOutput, type Registry, type Thread, type ThreadStatus } from "./types.ts"; -export const COMMITMENT_EVENT_TYPE = "commitment-event"; +export const emptyRegistry = (): Registry => ({ threads: [], unreconciled: [], created: 0 }); -export function emptyRegistry(): Registry { - return { threads: [], unreconciled: [], created: 0 }; -} - -export function isLive(thread: Thread): boolean { - return LIVE_STATUSES.includes(thread.status); -} - -function isHostFinal(thread: Thread): boolean { - return HOST_FINAL_STATUSES.includes(thread.status); -} - -function threadId(index: number): string { - return `t${index + 1}`; -} - -function newThread(id: string, observed: ObservedThread, at: number): Thread { - return { - id, - canonicalRequest: observed.canonicalRequest, - requestedProperties: [...observed.requestedProperties], - implementedProperties: [...observed.implementedProperties], - contradictions: [...observed.contradictions], - evidenceActionIds: [...observed.evidenceActionIds], - status: observed.status, - createdAt: at, - updatedAt: at, - }; -} +export const isFinal = (thread: Pick): boolean => + thread.status === "done" || + thread.status === "cancelled" || + thread.status === "superseded" || + thread.status === "dismissed"; -function patchThread(thread: Thread, observed: ObservedThread, at: number): Thread { - return { - ...thread, - canonicalRequest: observed.canonicalRequest || thread.canonicalRequest, - requestedProperties: [...observed.requestedProperties], - implementedProperties: [...observed.implementedProperties], - contradictions: [...observed.contradictions], - evidenceActionIds: [...observed.evidenceActionIds], - status: observed.status, - updatedAt: at, - }; -} +type LiveThread = Thread & { status: Exclude }; -export type PromotionBlocker = - | "not_candidate" - | "disabled" - | "no_cited_evidence" - | "cited_action_failed" - | "cited_action_optional" - | "open_contradictions"; +export const liveThreads = (registry: Registry): LiveThread[] => + registry.threads.filter((thread): thread is LiveThread => !isFinal(thread)); -export interface PromotionOutcome { - promote: boolean; - reason?: PromotionReason; - blockedBy?: PromotionBlocker; +export function observationChangesState(registry: Registry, output: ObserverOutput): boolean { + if (output.additions.length > 0) return true; + const existing = new Map(output.existing.map((thread) => [thread.id, thread])); + const live = liveThreads(registry); + return ( + existing.size !== live.length || + live.some((thread) => { + const observed = existing.get(thread.id); + return ( + observed === undefined || + observed.canonicalRequest !== thread.canonicalRequest || + observed.status !== thread.status + ); + }) + ); } -/** - * Thread-scoped promotion gate. The observer can only propose candidacy and cite - * evidence; `done` is written here or by explicit user acceptance. This is the one - * place the closure policy lives - keep it pure and host-free. - * - * Attribution eval (gemma-4-31b-it, minimal, schema-constrained, 10 cases x 3 reps): - * gate decision accuracy 30/30 with 0 false and 0 missed promotions, while raw citation - * precision was only ~70%. Every condition below absorbs a measured model failure: - * - `cited_action_optional` caught a cited `git status` (3/3) and is the only thing that - * stopped a false closure there. - * - candidacy-first ordering caught a passing regression suite cited against an untouched - * sibling thread (3/3). - * The model is a good candidacy and binding proposer but a sloppy evidence selector. - * Removing any single condition is a safety regression, not a simplification. - */ -export function promotionDecision( - thread: Thread, - actions: readonly ActionRecord[], - options: ReplayOptions = {}, -): PromotionOutcome { - if (thread.status !== "completion_candidate") { - return { promote: false, blockedBy: "not_candidate" }; - } - if (options.autoPromoteOnEvidence === false) { - return { promote: false, blockedBy: "disabled" }; - } - if (thread.contradictions.length > 0) { - return { promote: false, blockedBy: "open_contradictions" }; - } - const byId = new Map(actions.map((action) => [action.id, action])); - const cited = thread.evidenceActionIds.flatMap((id) => { - const action = byId.get(id); - return action ? [action] : []; - }); - - const hasRequiredValidation = actions.some((action) => action.required); - - if (cited.length > 0) { - if (cited.some((action) => !action.ok)) { - return { promote: false, blockedBy: "cited_action_failed" }; - } - const hasRequiredCitation = cited.some((action) => action.required); - if (hasRequiredCitation) { - return { promote: true, reason: "evidence" }; - } - // If required validations ran in this turn, citing only optional actions is blocked - if (hasRequiredValidation) { - return { promote: false, blockedBy: "cited_action_optional" }; - } - // When no validation suite ran in the turn, citing successful delivery actions promotes - return { promote: true, reason: "uncontested" }; - } - - if (hasRequiredValidation) { - return { promote: false, blockedBy: "no_cited_evidence" }; - } - // For conversational, operational, or design tasks with no validation tools in the turn, - // an uncontested candidate auto-promotes when contradictions are empty. - return { promote: true, reason: "uncontested" }; -} +const patchThread = (thread: Thread, observed: ObservedThread, at: number): Thread => ({ + ...thread, + canonicalRequest: observed.canonicalRequest, + status: observed.status, + updatedAt: at, +}); -function applyObservation( +const applyObservation = ( registry: Registry, event: Extract, - options: ReplayOptions, -): Registry { - const byId = new Map(registry.threads.map((thread) => [thread.id, thread])); - for (const observed of event.output.existing) { - const id = observed.id; - if (id === undefined) continue; - const existing = byId.get(id); - if (!existing || isHostFinal(existing)) continue; - byId.set(id, patchThread(existing, observed, event.at)); - } - - let created = registry.created; - const additions: Thread[] = []; - for (const observed of event.output.additions) { - const id = threadId(created); - created += 1; - additions.push(newThread(id, observed, event.at)); - } - - const merged = [...registry.threads.map((thread) => byId.get(thread.id) ?? thread), ...additions]; - const promoted = merged.map((thread) => { - if (thread.updatedAt !== event.at) return thread; - const decision = promotionDecision(thread, event.actions, options); - if (!decision.promote || decision.reason === undefined) return thread; - return { ...thread, status: "done" as const, promotedBy: decision.reason, updatedAt: event.at }; +): Registry => { + const byId = new Map(event.output.existing.map((thread) => [thread.id, thread])); + const threads = registry.threads.map((thread) => { + const observed = byId.get(thread.id); + return observed && !isFinal(thread) ? patchThread(thread, observed, event.at) : thread; }); - + const additions = event.output.additions.map((observed, index): Thread => ({ + id: `t${registry.created + index + 1}`, + canonicalRequest: observed.canonicalRequest, + status: observed.status, + createdAt: event.at, + updatedAt: event.at, + })); const reconciled = new Set(event.reconciled); return { - threads: promoted, + threads: [...threads, ...additions], unreconciled: registry.unreconciled.filter((message) => !reconciled.has(message.messageId)), - created, + created: registry.created + additions.length, }; -} +}; -function applyUserAction( +const applyUserAction = ( registry: Registry, event: Extract, -): Registry { - const action = event.action; - if (action === "drop") { +): Registry => { + if (event.action === "drop") { return { ...registry, threads: registry.threads.filter((thread) => thread.id !== event.threadId) }; } - const threads = registry.threads.map((thread) => { - if (thread.id !== event.threadId) return thread; - switch (action) { - case "accept": - return { ...thread, status: "done" as const, promotedBy: "user_accepted" as const, updatedAt: event.at }; - case "dismiss": - return { ...thread, status: "dismissed" as const, promotedBy: undefined, updatedAt: event.at }; - case "reopen": + const status: ThreadStatus = + event.action === "accept" ? "done" : event.action === "dismiss" ? "dismissed" : "open"; + return { + ...registry, + threads: registry.threads.map((thread) => + thread.id === event.threadId ? { ...thread, status, updatedAt: event.at } : thread, + ), + }; +}; + +export function replay(events: CommitmentEvent[]): Registry { + return events.reduce((registry, event) => { + switch (event.kind) { + case "user_message": return { - ...thread, - status: "open" as const, - promotedBy: undefined, - evidenceActionIds: [], - updatedAt: event.at, + ...registry, + unreconciled: [...registry.unreconciled, event], }; - default: { - const exhaustive: never = action; - return exhaustive; - } - } - }); - return { ...registry, threads }; -} - -export function applyEvent(registry: Registry, event: CommitmentEvent, options: ReplayOptions = {}): Registry { - switch (event.kind) { - case "user_message": { - if (registry.unreconciled.some((message) => message.messageId === event.messageId)) return registry; - const record: UserMessageRecord = { messageId: event.messageId, text: event.text, at: event.at }; - return { ...registry, unreconciled: [...registry.unreconciled, record] }; - } - case "observation": - return applyObservation(registry, event, options); - case "user_action": - return applyUserAction(registry, event); - case "correction": - return { - ...registry, - threads: registry.threads.map((thread) => - thread.id === event.threadId - ? { ...thread, canonicalRequest: event.canonicalRequest, updatedAt: event.at } - : thread, - ), - }; - default: { - const exhaustive: never = event; - return exhaustive; + case "observation": + return applyObservation(registry, event); + case "user_action": + return applyUserAction(registry, event); + case "correction": + return { + ...registry, + threads: registry.threads.map((thread) => + thread.id === event.threadId + ? { + ...thread, + ...(event.canonicalRequest === undefined + ? {} + : { canonicalRequest: event.canonicalRequest }), + ...(event.status === undefined ? {} : { status: event.status }), + updatedAt: event.at, + } + : thread, + ), + }; + case "observer_failure": + return registry; } - } + }, emptyRegistry()); } -export function replay(events: readonly CommitmentEvent[], options: ReplayOptions = {}): Registry { - return events.reduce((registry, event) => applyEvent(registry, event, options), emptyRegistry()); -} - -export function liveThreads(registry: Registry): Thread[] { - return registry.threads.filter(isLive); -} - -const EVENT_KINDS = new Set(["user_message", "observation", "user_action", "correction"]); +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + +const statuses = new Set([...OBSERVER_STATUSES, "dismissed"]); +const observerStatuses = new Set(OBSERVER_STATUSES); +const userActions = new Set(["accept", "dismiss", "reopen", "drop"]); +const strings = (value: unknown): value is string[] => + Array.isArray(value) && value.every((item) => typeof item === "string"); +const observed = (value: unknown, requireId: boolean): boolean => + isRecord(value) && + typeof value.canonicalRequest === "string" && + typeof value.status === "string" && + observerStatuses.has(value.status) && + (requireId ? typeof value.id === "string" : value.id === undefined); -/** Structural guard for custom entry payloads read back from the session file. */ export function isCommitmentEvent(value: unknown): value is CommitmentEvent { - if (!value || typeof value !== "object") return false; - const record = value as Record; - if (typeof record.kind !== "string" || !EVENT_KINDS.has(record.kind)) return false; - if (typeof record.at !== "number" && record.kind !== "user_message") return false; - switch (record.kind) { + if (!isRecord(value) || typeof value.kind !== "string" || typeof value.at !== "number") return false; + switch (value.kind) { case "user_message": - return typeof record.messageId === "string" && typeof record.text === "string" && typeof record.at === "number"; + return typeof value.messageId === "string" && typeof value.text === "string"; case "observation": return ( - Array.isArray(record.reconciled) && - Array.isArray(record.actions ?? []) && - !!record.output && - typeof record.output === "object" + strings(value.reconciled) && + isRecord(value.output) && + Array.isArray(value.output.existing) && + value.output.existing.every((thread) => observed(thread, true)) && + Array.isArray(value.output.additions) && + value.output.additions.every((thread) => observed(thread, false)) ); case "user_action": - return typeof record.threadId === "string" && typeof record.action === "string"; + return ( + typeof value.threadId === "string" && + typeof value.action === "string" && + userActions.has(value.action) + ); case "correction": - return typeof record.threadId === "string" && typeof record.canonicalRequest === "string"; + return ( + typeof value.threadId === "string" && + (value.canonicalRequest === undefined || typeof value.canonicalRequest === "string") && + (value.status === undefined || (typeof value.status === "string" && statuses.has(value.status))) + ); + case "observer_failure": + return strings(value.messageIds) && typeof value.error === "string"; default: return false; } diff --git a/src/types.ts b/src/types.ts index 0bcd19e..8a90413 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1,54 +1,19 @@ -/** Pure domain types for the commitment registry. No host imports. */ - -/** - * Statuses the observer model is allowed to emit. `done` is deliberately absent: - * closure is a host decision, which makes false closure structurally impossible. - */ -export const OBSERVER_STATUSES = [ - "open", - "active", - "blocked", - "waiting", - "completion_candidate", - "cancelled", - "superseded", -] as const; +export const OBSERVER_STATUSES = ["open", "active", "waiting", "blocked", "done", "cancelled", "superseded"] as const; export type ObserverStatus = (typeof OBSERVER_STATUSES)[number]; - -/** Host-owned terminal states added on top of the observer enum. */ -export type HostStatus = "done" | "dismissed"; - -export type ThreadStatus = ObserverStatus | HostStatus; - -export const LIVE_STATUSES: readonly ThreadStatus[] = ["open", "active", "blocked", "waiting", "completion_candidate"]; - -export const HOST_FINAL_STATUSES: readonly ThreadStatus[] = ["done", "dismissed"]; - -export type PromotionReason = "user_accepted" | "evidence" | "uncontested"; +export type ThreadStatus = ObserverStatus | "dismissed"; export interface Thread { id: string; canonicalRequest: string; - requestedProperties: string[]; - implementedProperties: string[]; - contradictions: string[]; - /** Ids of packet actions the observer cited as bearing on this thread. */ - evidenceActionIds: string[]; status: ThreadStatus; createdAt: number; updatedAt: number; - promotedBy?: PromotionReason; } -/** One validated observer thread record (existing update or addition). */ export interface ObservedThread { id?: string; canonicalRequest: string; - requestedProperties: string[]; - implementedProperties: string[]; - contradictions: string[]; - evidenceActionIds: string[]; status: ObserverStatus; } @@ -57,55 +22,34 @@ export interface ObserverOutput { additions: ObservedThread[]; } -/** A user message recorded on arrival so a crash cannot lose it. */ export interface UserMessageRecord { messageId: string; text: string; at: number; } -/** - * One action performed during a request. Ids are stable inside a single observer - * call and are the only way an observer can attach evidence to a thread. - */ -export interface ActionRecord { - id: string; - label: string; - required: boolean; - ok: boolean; +export interface ObservationFailure { + kind: "observer_failure"; + messageIds: string[]; + error: string; at: number; } -export type UserAction = "accept" | "dismiss" | "reopen" | "drop"; - -/** - * Append-only session events. Current state is a left fold over these, - * so the registry survives compaction, resume, fork and branch navigation. - */ export type CommitmentEvent = | { kind: "user_message"; messageId: string; text: string; at: number } - | { - kind: "observation"; - at: number; - reconciled: string[]; - output: ObserverOutput; - actions: ActionRecord[]; - } - | { kind: "user_action"; at: number; threadId: string; action: UserAction } - | { kind: "correction"; at: number; threadId: string; canonicalRequest: string }; + | { kind: "observation"; reconciled: string[]; output: ObserverOutput; at: number } + | { kind: "user_action"; threadId: string; action: "accept" | "dismiss" | "reopen" | "drop"; reason?: string; at: number } + | { kind: "correction"; threadId: string; canonicalRequest?: string; status?: ThreadStatus; at: number } + | ObservationFailure; export interface Registry { threads: Thread[]; unreconciled: UserMessageRecord[]; - /** Threads created so far, used to mint deterministic ids during replay. */ created: number; } -export interface ReplayOptions { - /** - * Promote a completion_candidate to done from thread-scoped action evidence. - * On unless explicitly disabled: evidence is cited per thread, so a passing - * check can no longer close an unrelated thread. - */ - autoPromoteOnEvidence?: boolean; +export interface ObserverPacket { + threads: Thread[]; + userMessages: UserMessageRecord[]; + completedAssistantTurn: string; } diff --git a/test/extension.test.ts b/test/extension.test.ts new file mode 100644 index 0000000..c621634 --- /dev/null +++ b/test/extension.test.ts @@ -0,0 +1,50 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { + buildContextBlock, + CLEARED_CONTEXT, + nextContextRevision, + REFRESHING_CONTEXT, +} from "../src/inject.ts"; +import { replay } from "../src/state.ts"; + +const open = replay([ + { + kind: "observation", + reconciled: [], + output: { existing: [], additions: [{ canonicalRequest: "finish the report", status: "active" }] }, + at: 1, + }, +]); +const done = replay([ + { + kind: "observation", + reconciled: [], + output: { existing: [], additions: [{ canonicalRequest: "finish the report", status: "active" }] }, + at: 1, + }, + { kind: "user_action", threadId: "t1", action: "accept", at: 2 }, +]); + +test("internal wakes never inject pending-request context", () => { + assert.equal(nextContextRevision("internal", open, undefined), undefined); +}); + +test("user turns inject only changed pending-request state", () => { + const first = nextContextRevision("interactive", open, undefined)!; + assert.match(first, /finish the report/); + assert.equal(first, buildContextBlock(open)); + assert.equal(nextContextRevision("rpc", open, first), undefined); +}); + +test("a pending reconciliation invalidates stale context without blocking", () => { + const previous = buildContextBlock(open)!; + assert.equal(nextContextRevision("interactive", open, previous, true), REFRESHING_CONTEXT); + assert.equal(nextContextRevision("interactive", open, REFRESHING_CONTEXT, true), undefined); +}); + +test("closing the final request emits one clear revision", () => { + const previous = buildContextBlock(open)!; + assert.equal(nextContextRevision("interactive", done, previous), CLEARED_CONTEXT); + assert.equal(nextContextRevision("interactive", done, CLEARED_CONTEXT), undefined); +}); diff --git a/test/live-observer.ts b/test/live-observer.ts index 2f46235..c985bbd 100644 --- a/test/live-observer.ts +++ b/test/live-observer.ts @@ -1,42 +1,33 @@ - import { readFileSync } from "node:fs"; import { homedir } from "node:os"; import { join } from "node:path"; -import { getModel } from "@earendil-works/pi-ai"; -import type { Api, Model } from "@earendil-works/pi-ai"; -import { runObserver, observerConfigFromEnv, OBSERVER_JSON_SCHEMA } from "../src/observer.ts"; -import type { ObserverPacket } from "../src/packet.ts"; +import type { AgentMessage } from "@earendil-works/pi-agent-core"; +import { getModel, type Api, type Model } from "@earendil-works/pi-ai"; +import { configFromEnv, runObserver } from "../src/observer.ts"; +import { replay } from "../src/state.ts"; const key = JSON.parse(readFileSync(join(homedir(), ".prime/agent/auth.json"), "utf8")).google.key as string; - -const packet: ObserverPacket = { - threads: [ - { id: "t1", canonicalRequest: "add a retry with exponential backoff", status: "active" }, - { id: "t2", canonicalRequest: "update the terminal docs", status: "open" }, - ], - precedingAssistant: "I will start with the retry logic.", - userMessages: ["also fix the resize flicker while you are in there"], - completedAssistantTurn: "Done, retry added.", - actions: [{ id: "a1", label: "bash: npm test", required: true, ok: false }], +const cfg = configFromEnv(process.env); +const lookup = getModel as unknown as (provider: string, modelId: string) => Model | undefined; +const model = lookup(cfg.provider, cfg.modelId); +const modelRegistry = { + find: () => model, + getApiKeyAndHeaders: async () => ({ ok: true, apiKey: key, headers: {} }), }; +const registry = replay([ + { kind: "user_message", messageId: "m1", text: "add retry logic and update the docs", at: 1 }, +]); +const messages = [ + { + role: "assistant", + content: [{ type: "text", text: "The retry logic is complete. The docs still need work." }], + stopReason: "stop", + timestamp: Date.now(), + }, +] as unknown as AgentMessage[]; -const statusNode = (OBSERVER_JSON_SCHEMA as { properties: { existing: { items: { properties: { status: unknown } } } } }) - .properties.existing.items.properties.status; - -async function main() { - const cfg = observerConfigFromEnv(process.env); - console.log("status schema node:", JSON.stringify(statusNode)); - console.log("config:", JSON.stringify(cfg)); - // getModel's model-id parameter is keyed off the provider literal, so a config-driven - // provider string cannot satisfy it. Resolve through a string-typed view of the same function. - const lookup = getModel as unknown as (provider: string, modelId: string) => Model | undefined; - const model = lookup(cfg.provider, cfg.modelId); - console.log("model:", model ? `${model.provider}/${model.id} api=${model.api}` : "NOT FOUND"); - const registry = { - find: () => model, - getApiKeyAndHeaders: async () => ({ ok: true, apiKey: key, headers: {} }), - }; - const res = await runObserver(registry as never, cfg, packet, new Set(["t1", "t2"])); - console.log("RESULT:", JSON.stringify(res, null, 1).slice(0, 3000)); -} -main().catch((e) => { console.error("FAILED:", e); process.exit(1); }); +const result = await runObserver(registry, messages, new Set(["m1"]), modelRegistry as never, { + ...cfg, + apiKey: key, +}); +console.log(JSON.stringify(result, null, 2)); diff --git a/test/logic.test.ts b/test/logic.test.ts index f6ac6dc..315c89e 100644 --- a/test/logic.test.ts +++ b/test/logic.test.ts @@ -1,514 +1,187 @@ import assert from "node:assert/strict"; import test from "node:test"; -import { renderOpenThreads, estimateTokens, orderForInjection } from "../src/inject.ts"; -import { alignActionIds, buildPacket, isRequiredValidation, type PacketMessage } from "../src/packet.ts"; -import { validateObserverOutput, extractJsonObject } from "../src/schema.ts"; -import { isCommitmentEvent, isLive, promotionDecision, replay } from "../src/state.ts"; -import type { ActionRecord, CommitmentEvent, ObservedThread, Thread } from "../src/types.ts"; - -function observed(overrides: Partial): ObservedThread { - return { - canonicalRequest: "do the thing", - requestedProperties: ["thing done"], - implementedProperties: [], - contradictions: [], - evidenceActionIds: [], - status: "open", - ...overrides, - }; -} - -function thread(overrides: Partial): Thread { - return { - id: "t1", - canonicalRequest: "do the thing", - requestedProperties: [], - implementedProperties: [], - contradictions: [], - evidenceActionIds: [], - status: "open", - createdAt: 1, - updatedAt: 1, - ...overrides, - }; -} - -function action(overrides: Partial): ActionRecord { - return { id: "a1", label: "bash: npm run check", required: true, ok: true, at: 1, ...overrides }; -} +import type { AgentMessage } from "@earendil-works/pi-agent-core"; +import { buildContextBlock, CLEARED_CONTEXT } from "../src/inject.ts"; +import { buildPacket } from "../src/packet.ts"; +import { OBSERVER_SYSTEM_PROMPT } from "../src/prompt.ts"; +import { parseObserverOutput } from "../src/schema.ts"; +import { isCommitmentEvent, liveThreads, observationChangesState, replay } from "../src/state.ts"; +import type { CommitmentEvent, Registry } from "../src/types.ts"; + +const observation = (overrides: Partial> = {}) => ({ + kind: "observation" as const, + reconciled: ["m1"], + output: { + existing: [], + additions: [{ canonicalRequest: "ship the patch", status: "active" as const }], + }, + at: 2, + ...overrides, +}); -test("user messages are unreconciled until an observation reconciles them", () => { - const events: CommitmentEvent[] = [ - { kind: "user_message", messageId: "m1", text: "fix resize and add docs", at: 1 }, - ]; - const before = replay(events); - assert.equal(before.unreconciled.length, 1); - assert.equal(before.threads.length, 0); +const registry = (...events: CommitmentEvent[]): Registry => + replay([{ kind: "user_message", messageId: "m1", text: "ship the patch", at: 1 }, ...events]); - const after = replay([ - ...events, +const assistant = (text: string): AgentMessage[] => + [ { - kind: "observation", - at: 2, - reconciled: ["m1"], - actions: [], - output: { - existing: [], - additions: [observed({ canonicalRequest: "fix resize" }), observed({ canonicalRequest: "add docs" })], - }, + role: "assistant", + content: [{ type: "text", text }], + stopReason: "stop", + timestamp: 3, }, - ]); - assert.equal(after.unreconciled.length, 0); - assert.deepEqual( - after.threads.map((item) => [item.id, item.status]), - [ - ["t1", "open"], - ["t2", "open"], - ], - ); -}); + ] as unknown as AgentMessage[]; -test("replay is deterministic and ids are stable across repeated folds", () => { - const events: CommitmentEvent[] = [ - { kind: "user_message", messageId: "m1", text: "a", at: 1 }, - { - kind: "observation", - at: 2, - reconciled: ["m1"], - actions: [], - output: { existing: [], additions: [observed({ canonicalRequest: "a" })] }, - }, - { kind: "user_message", messageId: "m2", text: "b", at: 3 }, - { - kind: "observation", - at: 4, - reconciled: ["m2"], - actions: [], - output: { - existing: [observed({ id: "t1", canonicalRequest: "a", status: "completion_candidate" })], - additions: [observed({ canonicalRequest: "b" })], - }, - }, - ]; - assert.deepEqual(replay(events), replay(events)); - const registry = replay(events); - assert.deepEqual( - registry.threads.map((item) => [item.id, item.status]), - [ - ["t1", "done"], - ["t2", "open"], - ], +test("observation creates stable threads and reconciles only named messages", () => { + const result = registry( + { kind: "user_message", messageId: "m2", text: "write docs", at: 2 }, + observation(), ); -}); - -test("observer can never close a thread and never overwrites host-final state", () => { - const registry = replay([ - { kind: "user_message", messageId: "m1", text: "a", at: 1 }, - { - kind: "observation", - at: 2, - reconciled: ["m1"], - actions: [], - output: { existing: [], additions: [observed({ canonicalRequest: "a" })] }, - }, - { kind: "user_action", at: 3, threadId: "t1", action: "accept" }, - { - kind: "observation", - at: 4, - reconciled: [], - actions: [], - output: { existing: [observed({ id: "t1", canonicalRequest: "a", status: "open" })], additions: [] }, - }, + assert.deepEqual(result.threads, [ + { id: "t1", canonicalRequest: "ship the patch", status: "active", createdAt: 2, updatedAt: 2 }, ]); - assert.equal(registry.threads[0]?.status, "done"); - assert.equal(registry.threads[0]?.promotedBy, "user_accepted"); -}); - -test("promotion gate: cited passing required action promotes", () => { - const candidate = thread({ status: "completion_candidate", evidenceActionIds: ["a1"] }); - assert.deepEqual(promotionDecision(candidate, [action({ id: "a1" })]), { - promote: true, - reason: "evidence", - }); -}); - -test("promotion gate: only candidates are promotable and the kill switch disables it", () => { - assert.deepEqual(promotionDecision(thread({ evidenceActionIds: ["a1"] }), [action({})]), { - promote: false, - blockedBy: "not_candidate", - }); - const candidate = thread({ status: "completion_candidate", evidenceActionIds: ["a1"] }); - assert.deepEqual(promotionDecision(candidate, [action({})], { autoPromoteOnEvidence: false }), { - promote: false, - blockedBy: "disabled", - }); -}); - -test("promotion gate: a cited failing action blocks promotion", () => { - const candidate = thread({ status: "completion_candidate", evidenceActionIds: ["a1", "a2"] }); - const actions = [action({ id: "a1", ok: true }), action({ id: "a2", label: "bash: npm test", ok: false })]; - assert.deepEqual(promotionDecision(candidate, actions), { - promote: false, - blockedBy: "cited_action_failed", - }); -}); - -test("promotion gate: citing only an optional action does not promote when required validation ran", () => { - const candidate = thread({ status: "completion_candidate", evidenceActionIds: ["a1"] }); - const actions = [ - action({ id: "a1", required: false, label: "bash: ls" }), - action({ id: "a2", required: true, label: "bash: npm test", ok: true }), - ]; - assert.deepEqual(promotionDecision(candidate, actions), { - promote: false, - blockedBy: "cited_action_optional", - }); -}); - -test("promotion gate: citing successful delivery action promotes when no validation ran", () => { - const candidate = thread({ status: "completion_candidate", evidenceActionIds: ["a1"] }); - const actions = [action({ id: "a1", required: false, label: "bash: git push", ok: true })]; - assert.deepEqual(promotionDecision(candidate, actions), { - promote: true, - reason: "uncontested", - }); -}); - -test("promotion gate: no surviving cited ids means no promotion", () => { - const candidate = thread({ status: "completion_candidate", evidenceActionIds: [] }); - assert.deepEqual(promotionDecision(candidate, [action({})]), { - promote: false, - blockedBy: "no_cited_evidence", - }); - const ghost = thread({ status: "completion_candidate", evidenceActionIds: ["a9"] }); - assert.deepEqual(promotionDecision(ghost, [action({ id: "a1" })]), { - promote: false, - blockedBy: "no_cited_evidence", - }); -}); - -test("promotion gate: contradictions block even with a passing cited action", () => { - const candidate = thread({ - status: "completion_candidate", - evidenceActionIds: ["a1"], - contradictions: ["asked for option two, implemented option one"], - }); - assert.deepEqual(promotionDecision(candidate, [action({ id: "a1" })]), { - promote: false, - blockedBy: "open_contradictions", - }); + assert.deepEqual(result.unreconciled.map(({ messageId }) => messageId), ["m2"]); }); -test("attribution: only the thread that cites the passing action is promoted", () => { - const registry = replay([ - { kind: "user_message", messageId: "m1", text: "add retries and update docs", at: 1 }, +test("observer transitions are semantic and need no action evidence", () => { + const initial = registry(observation()); + const result = replay([ + ...([observation()] as CommitmentEvent[]), { kind: "observation", - at: 2, - reconciled: ["m1"], - actions: [ - { id: "a1", label: "bash: npm run check", required: true, ok: true, at: 2 }, - { id: "a2", label: "bash: ls", required: false, ok: true, at: 2 }, - ], - output: { - existing: [], - additions: [ - observed({ canonicalRequest: "add retries", status: "completion_candidate", evidenceActionIds: ["a1"] }), - observed({ canonicalRequest: "update docs", status: "completion_candidate", evidenceActionIds: [] }), - ], - }, + reconciled: [], + output: { existing: [{ id: "t1", canonicalRequest: "ship the patch", status: "done" }], additions: [] }, + at: 4, }, ]); - assert.deepEqual( - registry.threads.map((item) => [item.id, item.status, item.promotedBy ?? null]), - [ - ["t1", "done", "evidence"], - ["t2", "completion_candidate", null], - ], - ); + assert.equal(initial.threads[0].status, "active"); + assert.equal(result.threads[0].status, "done"); }); -test("unknown cited ids are dropped by validation, so replay cannot promote on them", () => { - const report = validateObserverOutput( - JSON.stringify({ - existing: [], - additions: [ - { - canonicalRequest: "add retries", - requestedProperties: [], - implementedProperties: [], - contradictions: [], - evidenceActionIds: ["a1", "a7", "a7"], - status: "completion_candidate", - }, - ], +test("final threads stay final until an explicit reopen", () => { + const done = registry( + observation({ + output: { existing: [], additions: [{ canonicalRequest: "ship the patch", status: "done" }] }, }), - new Set(), - new Set(["a1"]), ); - assert.ok(report); - assert.deepEqual(report.output.additions[0]?.evidenceActionIds, ["a1"]); - assert.equal(report.droppedEvidenceIds, 2); - - const ghostOnly = validateObserverOutput( - JSON.stringify({ - existing: [], - additions: [ - { - canonicalRequest: "add retries", - requestedProperties: [], - implementedProperties: [], - contradictions: [], - evidenceActionIds: ["a7"], - status: "completion_candidate", - }, - ], - }), - new Set(), - new Set(["a1"]), - ); - assert.ok(ghostOnly); - const registry = replay([ - { kind: "user_message", messageId: "m1", text: "add retries", at: 1 }, + const observedAgain = replay([ + observation({ output: { existing: [], additions: [{ canonicalRequest: "ship the patch", status: "done" }] } }), { kind: "observation", - at: 2, - reconciled: ["m1"], - actions: [{ id: "a1", label: "bash: npm run check", required: true, ok: true, at: 2 }], - output: ghostOnly.output, + reconciled: [], + output: { existing: [{ id: "t1", canonicalRequest: "changed", status: "active" }], additions: [] }, + at: 3, }, ]); - assert.equal(registry.threads[0]?.status, "completion_candidate"); -}); - -test("auto promotion is on by default and reversible", () => { - const events: CommitmentEvent[] = [ - { kind: "user_message", messageId: "m1", text: "a", at: 1 }, - { - kind: "observation", - at: 2, - reconciled: ["m1"], - actions: [{ id: "a1", label: "bash: npm run check", required: true, ok: true, at: 2 }], - output: { - existing: [], - additions: [observed({ canonicalRequest: "a", status: "completion_candidate", evidenceActionIds: ["a1"] })], - }, - }, - ]; - assert.equal(replay(events).threads[0]?.status, "done"); - assert.equal(replay(events).threads[0]?.promotedBy, "evidence"); - assert.equal(replay(events, { autoPromoteOnEvidence: false }).threads[0]?.status, "completion_candidate"); - - const reopened = replay([...events, { kind: "user_action", at: 3, threadId: "t1", action: "reopen" }]); - assert.equal(reopened.threads[0]?.status, "open"); - assert.equal(reopened.threads[0]?.promotedBy, undefined); - assert.deepEqual(reopened.threads[0]?.evidenceActionIds, []); -}); - -test("user actions: dismiss, reopen, drop, correction", () => { - const seed: CommitmentEvent[] = [ - { kind: "user_message", messageId: "m1", text: "a", at: 1 }, - { - kind: "observation", - at: 2, - reconciled: ["m1"], - actions: [], - output: { existing: [], additions: [observed({ canonicalRequest: "a" }), observed({ canonicalRequest: "b" })] }, - }, - ]; - const dismissed = replay([...seed, { kind: "user_action", at: 3, threadId: "t1", action: "dismiss" }]); - assert.equal(dismissed.threads[0]?.status, "dismissed"); - assert.equal(isLive(dismissed.threads[0]!), false); - - const dropped = replay([...seed, { kind: "user_action", at: 3, threadId: "t1", action: "drop" }]); - assert.deepEqual(dropped.threads.map((item) => item.id), ["t2"]); - - const corrected = replay([...seed, { kind: "correction", at: 3, threadId: "t2", canonicalRequest: "b, precisely" }]); - assert.equal(corrected.threads[1]?.canonicalRequest, "b, precisely"); + assert.equal(observedAgain.threads[0].canonicalRequest, done.threads[0].canonicalRequest); + const reopened = replay([ + observation({ output: { existing: [], additions: [{ canonicalRequest: "ship the patch", status: "done" }] } }), + { kind: "user_action", threadId: "t1", action: "reopen", at: 4 }, + ]); + assert.equal(reopened.threads[0].status, "open"); }); -test("schema validation drops invalid entries instead of trusting them", () => { - const known = new Set(["t1"]); - const raw = JSON.stringify({ +test("schema requires an exact existing-thread census", () => { + const valid = JSON.stringify({ existing: [ - { id: "t1", canonicalRequest: "a", requestedProperties: ["x"], implementedProperties: [], contradictions: [], evidenceActionIds: [], status: "blocked" }, - { id: "t9", canonicalRequest: "ghost", requestedProperties: [], implementedProperties: [], contradictions: [], evidenceActionIds: [], status: "open" }, - { id: "t1", canonicalRequest: "dup", requestedProperties: [], implementedProperties: [], contradictions: [], evidenceActionIds: [], status: "open" }, - { id: "t1", canonicalRequest: "bad status", requestedProperties: [], implementedProperties: [], contradictions: [], evidenceActionIds: [], status: "done" }, - ], - additions: [ - { canonicalRequest: "new work", requestedProperties: ["y"], implementedProperties: [], contradictions: [], evidenceActionIds: [], status: "open" }, - { canonicalRequest: "", requestedProperties: [], implementedProperties: [], contradictions: [], evidenceActionIds: [], status: "open" }, - { canonicalRequest: "no status" }, + { id: "t1", canonicalRequest: "one", status: "active" }, + { id: "t2", canonicalRequest: "two", status: "open" }, ], + additions: [], }); - const report = validateObserverOutput(raw, known); - assert.ok(report); - assert.deepEqual(report.output.existing.map((item) => item.id), ["t1"]); - assert.equal(report.output.existing[0]?.status, "blocked"); - assert.equal(report.droppedExisting, 3); - assert.deepEqual(report.unknownIds, ["t9"]); - assert.deepEqual(report.output.additions.map((item) => item.canonicalRequest), ["new work"]); - assert.equal(report.droppedAdditions, 2); - assert.equal(report.droppedEvidenceIds, 0); + assert.ok(parseObserverOutput(valid, new Map([["t1", "one"], ["t2", "two"]]), false)); + assert.equal( + parseObserverOutput( + JSON.stringify({ existing: [{ id: "t1", canonicalRequest: "one", status: "active" }], additions: [] }), + new Map([["t1", "one"], ["t2", "two"]]), + false, + ), + undefined, + ); + assert.equal( + parseObserverOutput( + JSON.stringify({ + existing: [ + { id: "t1", canonicalRequest: "one", status: "active" }, + { id: "t1", canonicalRequest: "one", status: "active" }, + ], + additions: [], + }), + new Map([["t1", "one"]]), + false, + ), + undefined, + ); }); -test("observer output that is not usable JSON degrades to no update", () => { - assert.equal(validateObserverOutput("sorry, I cannot", new Set()), undefined); - assert.equal(validateObserverOutput("{ not json", new Set()), undefined); - assert.equal(validateObserverOutput(JSON.stringify({ threads: [] }), new Set()), undefined); - assert.deepEqual(extractJsonObject('```json\n{"existing":[],"additions":[]}\n```'), { - existing: [], +test("schema forbids additions on a turn without new user messages", () => { + const raw = JSON.stringify({ existing: [], additions: [{ canonicalRequest: "invented", status: "open" }] }); + assert.equal(parseObserverOutput(raw, new Map(), false), undefined); + assert.ok(parseObserverOutput(raw, new Map(), true)); +}); + +test("schema forbids request drift without a new user message", () => { + const raw = JSON.stringify({ + existing: [{ id: "t1", canonicalRequest: "a rewritten request", status: "active" }], additions: [], }); - const fenced = validateObserverOutput('```json\n{"existing":[],"additions":[]}\n```', new Set()); - assert.deepEqual(fenced?.output, { existing: [], additions: [] }); + assert.equal(parseObserverOutput(raw, new Map([["t1", "the original request"]]), false), undefined); + assert.ok(parseObserverOutput(raw, new Map([["t1", "the original request"]]), true)); }); -test("injection renders only live threads and stays under the token cap", () => { - assert.equal(renderOpenThreads([]), undefined); - assert.equal(renderOpenThreads([thread({ status: "done" }), thread({ id: "t2", status: "dismissed" })]), undefined); - - const many = Array.from({ length: 40 }, (_, index) => - thread({ - id: `t${index + 1}`, - canonicalRequest: `request number ${index + 1} with a reasonably long description of the work`, - createdAt: index, - status: index % 2 === 0 ? "open" : "blocked", - }), - ); - const cap = 120; - const block = renderOpenThreads(many, cap); - assert.ok(block); - assert.ok(estimateTokens(block) <= cap, `block estimated ${estimateTokens(block)} tokens`); - assert.match(block, /hidden by the injection cap/); - assert.match(block, //); - assert.match(block, /<\/open_threads>$/); +test("packet snapshots only the user-message ids owned by its boundary", () => { + const current = registry({ kind: "user_message", messageId: "m2", text: "write docs", at: 2 }); + const packet = buildPacket(current, assistant("implemented the patch"), new Set(["m1"])); + assert.deepEqual(packet.userMessages.map(({ messageId }) => messageId), ["m1"]); + assert.equal(packet.completedAssistantTurn, "implemented the patch"); }); -test("injection orders blocked and waiting threads first", () => { - const ordered = orderForInjection([ - thread({ id: "t1", status: "open", createdAt: 1 }), - thread({ id: "t2", status: "waiting", createdAt: 2 }), - thread({ id: "t3", status: "blocked", createdAt: 3 }), - thread({ id: "t4", status: "done", createdAt: 4 }), - ]); - assert.deepEqual(ordered.map((item) => item.id), ["t3", "t2", "t1"]); +test("pending context contains no ids or model bookkeeping", () => { + const current = registry(observation()); + const block = buildContextBlock(current)!; + assert.match(block, /ship the patch/); + assert.doesNotMatch(block, /t1|missing|threads\./); + assert.equal(buildContextBlock({ ...current, threads: [{ ...current.threads[0], status: "done" }] }), undefined); + assert.match(CLEARED_CONTEXT, /snapshots are obsolete/); }); -test("packet carries prior state, adjacent turns and identified actions", () => { - const messages: PacketMessage[] = [ - { role: "user", text: "first request" }, - { role: "assistant", text: "here are two options" }, - { role: "user", text: "the second one" }, - { role: "assistant", text: "picked option two" }, - { role: "toolResult", text: "ok", toolName: "bash", isError: false }, - { role: "assistant", text: "done, tests fail" }, - ]; - const recorded: ActionRecord[] = [ - { id: "x", label: "bash: npm run check", required: true, ok: true, at: 1 }, - { id: "y", label: "bash: npm test", required: true, ok: false, at: 2 }, - ]; - const packet = buildPacket( - messages, - [{ messageId: "m1", text: "the second one", at: 1 }], - [thread({ id: "t1", canonicalRequest: "choose an option" })], - recorded, - ); - assert.equal(packet.precedingAssistant, "here are two options"); - assert.equal(packet.completedAssistantTurn, "done, tests fail"); - assert.deepEqual(packet.userMessages, ["the second one"]); - assert.deepEqual(packet.actions, [ - { id: "a1", label: "bash: npm run check", required: true, ok: true }, - { id: "a2", label: "bash: npm test", required: true, ok: false }, - ]); - assert.deepEqual(packet.threads, [ - { - id: "t1", - canonicalRequest: "choose an option", - status: "open", - requestedProperties: [], - implementedProperties: [], - }, - ]); - assert.deepEqual( - alignActionIds(recorded).map((item) => item.id), - packet.actions.map((item) => item.id), +test("pending context escapes user-authored markup", () => { + const current = registry( + observation({ output: { existing: [], additions: [{ canonicalRequest: "fix & docs", status: "open" }] } }), ); + assert.match(buildContextBlock(current)!, /fix <xml> & docs/); }); -test("required validation detection", () => { - assert.equal(isRequiredValidation("npm run check"), true); - assert.equal(isRequiredValidation("cargo test --all"), true); - assert.equal(isRequiredValidation("ls -la"), false); - assert.equal(isRequiredValidation("echo npm"), false); +test("prompt has one coherent completion contract", () => { + assert.match(OBSERVER_SYSTEM_PROMPT, /Use done when/); + assert.doesNotMatch(OBSERVER_SYSTEM_PROMPT, /Never emit done|evidenceActionIds|required validation/); }); -test("persisted event guard rejects foreign payloads", () => { - assert.equal(isCommitmentEvent({ kind: "user_message", messageId: "m", text: "t", at: 1 }), true); - assert.equal(isCommitmentEvent({ kind: "user_message", messageId: "m" }), false); - assert.equal(isCommitmentEvent({ kind: "nope", at: 1 }), false); - assert.equal(isCommitmentEvent(undefined), false); +test("event parser rejects malformed persisted values", () => { + assert.equal(isCommitmentEvent({ kind: "observation", output: {}, reconciled: [], at: 1 }), false); + assert.equal(isCommitmentEvent({ kind: "user_action", threadId: "t1", action: "invented", at: 1 }), false); + assert.equal(isCommitmentEvent({ kind: "observer_failure", messageIds: [1], error: "bad", at: 1 }), false); }); -test("end to end pure pipeline: raw output -> events -> replay -> injected block", () => { - const events: CommitmentEvent[] = [ - { kind: "user_message", messageId: "m1", text: "fix resize, add a regression test, update docs", at: 1 }, - ]; - const actions: ActionRecord[] = [{ id: "a1", label: "bash: npm run check", required: true, ok: true, at: 2 }]; - const report = validateObserverOutput( - JSON.stringify({ - existing: [], - additions: [ - { canonicalRequest: "fix resize", requestedProperties: ["resize fixed"], implementedProperties: ["resize fixed"], contradictions: [], evidenceActionIds: ["a1"], status: "completion_candidate" }, - { canonicalRequest: "add a regression test", requestedProperties: ["test exists"], implementedProperties: [], contradictions: [], evidenceActionIds: [], status: "open" }, - { canonicalRequest: "update docs", requestedProperties: ["docs updated"], implementedProperties: [], contradictions: [], evidenceActionIds: [], status: "open" }, - ], +test("unchanged observations do not create empty ledger revisions", () => { + const current = registry(observation()); + assert.equal( + observationChangesState(current, { + existing: [{ id: "t1", canonicalRequest: "ship the patch", status: "active" }], + additions: [], }), - new Set(), - new Set(actions.map((item) => item.id)), + false, + ); + assert.equal( + observationChangesState(current, { + existing: [{ id: "t1", canonicalRequest: "ship the patch", status: "done" }], + additions: [], + }), + true, ); - assert.ok(report); - const registry = replay([ - ...events, - { kind: "observation", at: 2, reconciled: ["m1"], actions, output: report.output }, - ]); - assert.equal(registry.unreconciled.length, 0); - assert.equal(registry.threads.length, 3); - assert.equal(registry.threads[0]?.status, "done"); - const block = renderOpenThreads(registry.threads, 400); - assert.ok(block); - assert.match(block, /add a regression test/); - assert.match(block, /update docs/); - assert.equal(/fix resize/.test(block), false); -}); - -test("promotion gate: uncontested candidate auto-promotes when no validation commands ran in the turn", () => { - const candidate = thread({ status: "completion_candidate", evidenceActionIds: [] }); - // When actions are empty or contain only non-validation tools, uncontested candidate promotes - assert.deepEqual(promotionDecision(candidate, []), { - promote: true, - reason: "uncontested", - }); - assert.deepEqual(promotionDecision(candidate, [action({ required: false, label: "bash: ls" })]), { - promote: true, - reason: "uncontested", - }); - // But if a required validation command DID run in the turn and wasn't cited, it does not promote - assert.deepEqual(promotionDecision(candidate, [action({ required: true, label: "bash: npm test" })]), { - promote: false, - blockedBy: "no_cited_evidence", - }); }); -test("packet construction carries over requested and implemented properties", () => { - const t = thread({ - id: "t1", - canonicalRequest: "build feature", - requestedProperties: ["feature works"], - implementedProperties: ["feature works"], - }); - const packet = buildPacket([], [], [t], []); - assert.deepEqual(packet.threads[0].requestedProperties, ["feature works"]); - assert.deepEqual(packet.threads[0].implementedProperties, ["feature works"]); +test("liveThreads derives the pending view from status", () => { + const current = registry(observation()); + assert.deepEqual(liveThreads(current).map(({ id }) => id), ["t1"]); }); -- 2.51.2