import 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"]);
});