Skip to content

Commit 28bc56c

Browse files
ruby-dleeruby-dlee
authored
fix(pi): preserve direct reply continuity across compaction (#365)
* fix(pi): preserve captain exchanges across compaction * no-mistakes(review): Replace source assertions with behavioral coverage * no-mistakes(test): Update session-start assertion for Pi continuity mode * fix(pi): retain queued images across compaction * no-mistakes(review): Captain, admit Pi input and prove queue ordering * no-mistakes(review): Require exact input-specific Pi admission * no-mistakes(review): Preserve duplicate Pi input obligations * no-mistakes(review): Separate Pi observations from admitted exchanges * no-mistakes(review): Scope Pi answers to active runs * no-mistakes(review): Emit stable Pi continuity proof * no-mistakes(review): Preserve exact Pi delivery association, captain * no-mistakes(review): Prove exact Pi delivery continuity * no-mistakes(review): Captain, wire targeted Pi consumer proof * no-mistakes(review): Prove exact Pi delivery association * no-mistakes(review): Preserve Pi retry reply cohorts, captain * no-mistakes(document): Align Pi continuity documentation with retry lifecycle * no-mistakes(lint): Fix ShellCheck-safe Pi runtime discovery loops * no-mistakes: apply CI fixes
1 parent c3ce453 commit 28bc56c

15 files changed

Lines changed: 1366 additions & 29 deletions

‎.pi/extensions/fm-primary-pi-watch.ts‎

Lines changed: 297 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -4,16 +4,171 @@ import { createHash } from "node:crypto";
44
import { mkdirSync, readFileSync, writeFileSync } from "node:fs";
55
import { dirname, resolve } from "node:path";
66
import { fileURLToPath } from "node:url";
7-
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
7+
import type { ExtensionAPI, SessionEntry } from "@earendil-works/pi-coding-agent";
88
import { Type } from "typebox";
99

1010
type ArmResult = {
1111
ok: boolean;
1212
message: string;
1313
};
1414

15+
type DirectExchangeEvent = {
16+
version: 1;
17+
event: "submitted" | "admitted" | "delivered" | "answered";
18+
exchangeId: string;
19+
at: number;
20+
inputText?: string;
21+
imageCount?: number;
22+
inputContent?: unknown;
23+
delivery?: "immediate" | "steer" | "followUp";
24+
content?: unknown;
25+
runKey?: string;
26+
};
27+
28+
type DirectExchangeState = {
29+
exchangeId: string;
30+
submittedAt: number;
31+
inputText: string;
32+
imageCount: number;
33+
inputContent?: unknown;
34+
delivery: "immediate" | "steer" | "followUp";
35+
admittedAt?: number;
36+
deliveredAt?: number;
37+
deliveredContent?: unknown;
38+
deliveredRecordIndex?: number;
39+
deliveredRunKey?: string;
40+
answeredAt?: number;
41+
answerContent?: unknown;
42+
};
43+
1544
type LockOwnership = "owned" | "missing" | "other";
1645

46+
const directExchangeEntryType = "firstmate-direct-exchange";
47+
const directInputObservationEntryType = "firstmate-direct-input-observation";
48+
const continuityMessageType = "firstmate-direct-exchange-continuity";
49+
50+
function cloneJson(value: unknown): unknown {
51+
try {
52+
return JSON.parse(JSON.stringify(value));
53+
} catch {
54+
return String(value);
55+
}
56+
}
57+
58+
function textContent(content: unknown): string {
59+
if (typeof content === "string") return content;
60+
if (!Array.isArray(content)) return "";
61+
return content
62+
.filter((part): part is { type: "text"; text: string } => {
63+
if (!part || typeof part !== "object") return false;
64+
const candidate = part as { type?: unknown; text?: unknown };
65+
return candidate.type === "text" && typeof candidate.text === "string";
66+
})
67+
.map((part) => part.text)
68+
.join("\n");
69+
}
70+
71+
function parseDirectExchangeEvent(entry: SessionEntry): DirectExchangeEvent | undefined {
72+
if (entry.type !== "custom" || entry.customType !== directExchangeEntryType) return undefined;
73+
const data = entry.data;
74+
if (!data || typeof data !== "object") return undefined;
75+
const candidate = data as Partial<DirectExchangeEvent>;
76+
if (
77+
candidate.version !== 1 ||
78+
!["submitted", "admitted", "delivered", "answered"].includes(candidate.event ?? "") ||
79+
typeof candidate.exchangeId !== "string" ||
80+
!candidate.exchangeId ||
81+
typeof candidate.at !== "number"
82+
) {
83+
return undefined;
84+
}
85+
return candidate as DirectExchangeEvent;
86+
}
87+
88+
function foldDirectExchanges(entries: SessionEntry[]): DirectExchangeState[] {
89+
const ordered: DirectExchangeState[] = [];
90+
const byId = new Map<string, DirectExchangeState>();
91+
entries.forEach((entry, index) => {
92+
const event = parseDirectExchangeEvent(entry);
93+
if (!event) return;
94+
if (event.event === "submitted") {
95+
if (byId.has(event.exchangeId) || typeof event.inputText !== "string") return;
96+
const state: DirectExchangeState = {
97+
exchangeId: event.exchangeId,
98+
submittedAt: event.at,
99+
inputText: event.inputText,
100+
imageCount: event.imageCount ?? 0,
101+
delivery: event.delivery ?? "immediate",
102+
};
103+
if (event.inputContent !== undefined) state.inputContent = event.inputContent;
104+
byId.set(event.exchangeId, state);
105+
ordered.push(state);
106+
return;
107+
}
108+
const state = byId.get(event.exchangeId);
109+
if (!state) return;
110+
if (event.event === "admitted") {
111+
state.admittedAt = event.at;
112+
} else if (event.event === "delivered") {
113+
state.admittedAt ??= event.at;
114+
state.deliveredAt = event.at;
115+
state.deliveredContent = event.content;
116+
state.deliveredRecordIndex = index;
117+
state.deliveredRunKey = event.runKey;
118+
} else if (event.event === "answered") {
119+
state.answeredAt = event.at;
120+
state.answerContent = event.content;
121+
}
122+
});
123+
return ordered;
124+
}
125+
126+
function contentPresent(messages: Array<{ role: string; content?: unknown }>, role: string, content: unknown): boolean {
127+
const expected = JSON.stringify(content);
128+
return messages.some((message) => message.role === role && JSON.stringify(message.content) === expected);
129+
}
130+
131+
function renderDirectExchangeContinuity(
132+
entries: SessionEntry[],
133+
messages: Array<{ role: string; content?: unknown }>,
134+
): string | undefined {
135+
const exchanges = foldDirectExchanges(entries);
136+
const sections: string[] = [];
137+
const latestAnswered = [...exchanges].reverse().find((exchange) => exchange.answerContent !== undefined);
138+
if (
139+
latestAnswered &&
140+
(!contentPresent(messages, "user", latestAnswered.deliveredContent) ||
141+
!contentPresent(messages, "assistant", latestAnswered.answerContent))
142+
) {
143+
sections.push(
144+
[
145+
`Exchange ${latestAnswered.exchangeId}: ANSWERED`,
146+
`Human input, exact JSON: ${JSON.stringify(latestAnswered.deliveredContent)}`,
147+
`Assistant answer, exact JSON: ${JSON.stringify(latestAnswered.answerContent)}`,
148+
].join("\n"),
149+
);
150+
}
151+
for (const exchange of exchanges) {
152+
if (exchange.answerContent !== undefined) continue;
153+
if (exchange.deliveredContent !== undefined) {
154+
sections.push(
155+
[
156+
`Exchange ${exchange.exchangeId}: OPEN_REPLY_OBLIGATION`,
157+
`Human input, exact JSON: ${JSON.stringify(exchange.deliveredContent)}`,
158+
"No completed assistant answer was observed before compaction.",
159+
].join("\n"),
160+
);
161+
}
162+
}
163+
if (sections.length === 0) return undefined;
164+
return [
165+
"FIRSTMATE DIRECT EXCHANGE CONTINUITY",
166+
"This is extension-generated context metadata, not human-authored input.",
167+
"Watcher and turn-end supervision prompts are custom messages, not captain-authored requests.",
168+
...sections,
169+
].join("\n\n");
170+
}
171+
17172
const extensionFile = fileURLToPath(import.meta.url);
18173
const extensionDir = dirname(extensionFile);
19174
const root = resolve(extensionDir, "../..");
@@ -90,6 +245,16 @@ function failureLine(stdout: string, stderr: string, code: number | null): strin
90245
}
91246

92247
export default function (pi: ExtensionAPI) {
248+
let exchangeSequence = 0;
249+
let agentRunSequence = 0;
250+
let activeAgentRunKey: string | undefined;
251+
let replyCohort = new Set<string>();
252+
let agentStartAwaitingHumanDelivery = false;
253+
254+
function appendDirectExchange(event: DirectExchangeEvent): void {
255+
pi.appendEntry(directExchangeEntryType, event);
256+
}
257+
93258
function stopArm(): void {
94259
if (child) child.kill("SIGTERM");
95260
child = null;
@@ -100,10 +265,15 @@ export default function (pi: ExtensionAPI) {
100265
};
101266
process.once("exit", cleanupOnProcessExit);
102267

103-
async function sendWake(message: string) {
104-
await pi.sendUserMessage(
105-
`FIRSTMATE WATCHER WAKE: ${message}\n\nRun bin/fm-wake-drain.sh first, handle the queued wake, then resume Pi supervision.`,
106-
{ deliverAs: "followUp" },
268+
function sendWake(message: string): void {
269+
pi.sendMessage(
270+
{
271+
customType: "firstmate-watcher-wake",
272+
content: `FIRSTMATE WATCHER WAKE: ${message}\n\nRun bin/fm-wake-drain.sh first, handle the queued wake, then resume Pi supervision.`,
273+
display: true,
274+
details: { version: 1, source: "firstmate-extension", kind: "watcher-wake" },
275+
},
276+
{ deliverAs: "followUp", triggerTurn: true },
107277
);
108278
}
109279

@@ -138,15 +308,15 @@ export default function (pi: ExtensionAPI) {
138308
const failure = reason ? "" : failureLine(stdout, stderr, code);
139309
if (!reason && !failure) return;
140310
try {
141-
await sendWake(reason || failure);
311+
sendWake(reason || failure);
142312
} catch {
143313
// Pi owns delivery errors; fail open so the extension never wedges the session.
144314
}
145315
});
146316
child.on("error", async (error: Error) => {
147317
child = null;
148318
try {
149-
await sendWake(`watcher: FAILED - Pi extension arm child ${id} failed: ${error.message}`);
319+
sendWake(`watcher: FAILED - Pi extension arm child ${id} failed: ${error.message}`);
150320
} catch {
151321
// Fail open.
152322
}
@@ -157,6 +327,126 @@ export default function (pi: ExtensionAPI) {
157327
pi.on?.("session_start", () => {
158328
markLoaded();
159329
});
330+
331+
pi.on("agent_start", () => {
332+
activeAgentRunKey = `${process.pid}:${++agentRunSequence}`;
333+
agentStartAwaitingHumanDelivery = true;
334+
});
335+
336+
function newExchangeId(at: number, content: unknown): string {
337+
return createHash("sha256")
338+
.update(`${at}\0${++exchangeSequence}\0${JSON.stringify(content)}`)
339+
.digest("hex")
340+
.slice(0, 16);
341+
}
342+
343+
function admitExchange(exchangeId: string, at: number): void {
344+
appendDirectExchange({ version: 1, event: "admitted", exchangeId, at });
345+
}
346+
347+
function createAdmittedExchange(
348+
content: unknown,
349+
delivery: "immediate" | "steer" | "followUp",
350+
at: number,
351+
): string {
352+
const exchangeId = newExchangeId(at, content);
353+
appendDirectExchange({
354+
version: 1,
355+
event: "submitted",
356+
exchangeId,
357+
at,
358+
inputText: textContent(content),
359+
imageCount: Array.isArray(content)
360+
? content.filter((part) => part && typeof part === "object" && (part as { type?: unknown }).type === "image").length
361+
: 0,
362+
inputContent: content,
363+
delivery,
364+
});
365+
admitExchange(exchangeId, at);
366+
return exchangeId;
367+
}
368+
369+
pi.on("input", (event) => {
370+
if (event.source === "extension") return;
371+
const at = Date.now();
372+
const inputContent = cloneJson([{ type: "text", text: event.text }, ...(event.images ?? [])]);
373+
const delivery = event.streamingBehavior ?? "immediate";
374+
pi.appendEntry(directInputObservationEntryType, {
375+
version: 1,
376+
observationId: newExchangeId(at, inputContent),
377+
at,
378+
inputText: event.text,
379+
imageCount: event.images?.length ?? 0,
380+
inputContent,
381+
delivery,
382+
});
383+
});
384+
385+
pi.on("message_end", (event, ctx) => {
386+
if (event.message.role === "user") {
387+
const content = cloneJson(event.message.content);
388+
const exchangeId = createAdmittedExchange(content, "steer", event.message.timestamp);
389+
appendDirectExchange({
390+
version: 1,
391+
event: "delivered",
392+
exchangeId,
393+
at: event.message.timestamp,
394+
content,
395+
runKey: activeAgentRunKey,
396+
});
397+
if (agentStartAwaitingHumanDelivery) {
398+
replyCohort = new Set<string>();
399+
agentStartAwaitingHumanDelivery = false;
400+
}
401+
replyCohort.add(exchangeId);
402+
return;
403+
}
404+
if (event.message.role !== "assistant" || event.message.stopReason !== "stop") return;
405+
const answerText = textContent(event.message.content);
406+
if (!answerText) return;
407+
const branch = ctx.sessionManager.getBranch();
408+
const open = foldDirectExchanges(branch).filter((exchange) => {
409+
if (exchange.deliveredContent === undefined || exchange.answerContent !== undefined) return false;
410+
if (!replyCohort.has(exchange.exchangeId)) return false;
411+
const deliveredIndex = exchange.deliveredRecordIndex;
412+
if (deliveredIndex === undefined) return false;
413+
return !branch.slice(deliveredIndex + 1).some((entry) => entry.type === "custom_message");
414+
});
415+
for (const exchange of open) {
416+
appendDirectExchange({
417+
version: 1,
418+
event: "answered",
419+
exchangeId: exchange.exchangeId,
420+
at: event.message.timestamp,
421+
content: cloneJson(event.message.content),
422+
});
423+
}
424+
if (open.length > 0) replyCohort.clear();
425+
});
426+
427+
pi.on("context", (event, ctx) => {
428+
const continuity = renderDirectExchangeContinuity(
429+
ctx.sessionManager.getBranch(),
430+
event.messages,
431+
);
432+
if (!continuity) return;
433+
const message = {
434+
role: "custom" as const,
435+
customType: continuityMessageType,
436+
content: continuity,
437+
display: false,
438+
details: { version: 1, source: "firstmate-extension", kind: "direct-exchange-continuity" },
439+
timestamp: Date.now(),
440+
};
441+
let insertAt = 0;
442+
for (let i = event.messages.length - 1; i >= 0; i -= 1) {
443+
if (event.messages[i]?.role === "user") {
444+
insertAt = i;
445+
break;
446+
}
447+
}
448+
return { messages: [...event.messages.slice(0, insertAt), message, ...event.messages.slice(insertAt)] };
449+
});
160450
pi.on?.("session_shutdown", () => {
161451
stopArm();
162452
process.off("exit", cleanupOnProcessExit);

‎.pi/extensions/fm-primary-turnend-guard.ts‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -132,11 +132,17 @@ export default function (pi: ExtensionAPI) {
132132

133133
guardFollowupActive = true;
134134
try {
135-
await pi.sendUserMessage(
136-
"TURN WOULD END BLIND - supervision is off. " +
137-
"Resume supervision according to the session-start operating block before ending the turn.\n\n" +
138-
result.stderr,
139-
{ deliverAs: "followUp" },
135+
pi.sendMessage(
136+
{
137+
customType: "firstmate-turnend-guard",
138+
content:
139+
"TURN WOULD END BLIND - supervision is off. " +
140+
"Resume supervision according to the session-start operating block before ending the turn.\n\n" +
141+
result.stderr,
142+
display: true,
143+
details: { version: 1, source: "firstmate-extension", kind: "turnend-guard" },
144+
},
145+
{ deliverAs: "followUp", triggerTurn: true },
140146
);
141147
} catch {
142148
guardFollowupActive = false;

‎docs/architecture.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@ This Firstmate tooling change requires no web-app, API, or realtime deployment.
5858

5959
At session start, `bin/fm-session-start.sh` emits exactly one primary-harness supervision block rendered by `bin/fm-supervision-instructions.sh` from `docs/supervision-protocols/`.
6060
That block owns the live wait shape for the running primary harness: Claude and Grok use background-notify cycles, Codex uses bounded foreground checkpoints, Pi uses its two tracked primary extensions, and OpenCode uses its TUI plugin.
61+
Pi's supervision prompts remain custom-message context while exact direct exchanges and open reply obligations survive compaction through the contract in [`supervision-protocols/pi.md`](supervision-protocols/pi.md#input-provenance-and-compaction-continuity).
6162
`bin/fm-watch-arm.sh` remains the verified arm wrapper for protocols that call it; it forks the watcher as a tracked child, verifies it is genuinely alive with a fresh liveness beacon, and prints an honest initial status (`started` / `attached` / restart-only `healthy` / `FAILED`, the last exiting non-zero).
6263
On `attached` it stays live until that existing cycle ends so background-notify harnesses do not get an empty false wake from a healthy no-op exit.
6364
Because the initial status stays in the task's buffer long after the instant it describes, an attach is always closed by a terminal `FAILED` line - either `attached cycle ended` when the holder stops passing the liveness proof, or `attach interrupted` when the arm itself is signalled away - with the beacon age re-measured at exit and a non-zero status.

0 commit comments

Comments
 (0)