Something went wrong. Try again.
watches turns and automatically creates and resolves tasks that get pinned into main agent context
Something went wrong. Try again.
7.3 kB · 193 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194import assert from "node:assert/strict";import test from "node:test";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, OBSERVER_JSON_SCHEMA } 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<Extract<CommitmentEvent, { kind: "observation" }>> = {}) => ({ kind: "observation" as const, reconciled: ["m1"], output: { existing: [], additions: [{ canonicalRequest: "ship the patch", status: "active" as const }], }, at: 2, ...overrides,});
const registry = (...events: CommitmentEvent[]): Registry => replay([{ kind: "user_message", messageId: "m1", text: "ship the patch", at: 1 }, ...events]);
const assistant = (text: string): AgentMessage[] => [ { role: "assistant", content: [{ type: "text", text }], stopReason: "stop", timestamp: 3, }, ] as unknown as AgentMessage[];
test("observation creates stable threads and reconciles only named messages", () => { const result = registry( { kind: "user_message", messageId: "m2", text: "write docs", at: 2 }, observation(), ); assert.deepEqual(result.threads, [ { id: "t1", canonicalRequest: "ship the patch", status: "active", createdAt: 2, updatedAt: 2 }, ]); assert.deepEqual(result.unreconciled.map(({ messageId }) => messageId), ["m2"]);});
test("observer transitions are semantic and need no action evidence", () => { const initial = registry(observation()); const result = replay([ ...([observation()] as CommitmentEvent[]), { kind: "observation", reconciled: [], output: { existing: [{ id: "t1", canonicalRequest: "ship the patch", status: "done" }], additions: [] }, at: 4, }, ]); assert.equal(initial.threads[0].status, "active"); assert.equal(result.threads[0].status, "done");});
test("final threads stay final until an explicit reopen", () => { const done = registry( observation({ output: { existing: [], additions: [{ canonicalRequest: "ship the patch", status: "done" }] }, }), ); const observedAgain = replay([ observation({ output: { existing: [], additions: [{ canonicalRequest: "ship the patch", status: "done" }] } }), { kind: "observation", reconciled: [], output: { existing: [{ id: "t1", canonicalRequest: "changed", status: "active" }], additions: [] }, at: 3, }, ]); 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 leaves array bounds to local validation", () => { assert.equal("maxItems" in OBSERVER_JSON_SCHEMA.properties.existing, false); assert.equal("maxItems" in OBSERVER_JSON_SCHEMA.properties.additions, false);});
test("schema requires an exact existing-thread census", () => { const valid = JSON.stringify({ existing: [ { id: "t1", canonicalRequest: "one", status: "active" }, { id: "t2", canonicalRequest: "two", status: "open" }, ], additions: [], }); 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("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: [], }); assert.equal(parseObserverOutput(raw, new Map([["t1", "the original request"]]), false), undefined); assert.ok(parseObserverOutput(raw, new Map([["t1", "the original request"]]), true));});
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("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("pending context escapes user-authored markup", () => { const current = registry( observation({ output: { existing: [], additions: [{ canonicalRequest: "fix <xml> & docs", status: "open" }] } }), ); assert.match(buildContextBlock(current)!, /fix <xml> & docs/);});
test("prompt has one coherent completion contract", () => { assert.match(OBSERVER_SYSTEM_PROMPT, /Use done when/); assert.match(OBSERVER_SYSTEM_PROMPT, /Addition entries never include id/); assert.doesNotMatch(OBSERVER_SYSTEM_PROMPT, /Never emit done|evidenceActionIds|required validation/);});
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("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: [], }), false, ); assert.equal( observationChangesState(current, { existing: [{ id: "t1", canonicalRequest: "ship the patch", status: "done" }], additions: [], }), true, );});
test("liveThreads derives the pending view from status", () => { const current = registry(observation()); assert.deepEqual(liveThreads(current).map(({ id }) => id), ["t1"]);});