Wave D: native AgentEvent wire + client session rewrite (SDK purged)

Server: ui-stream.ts is verbatim pi AgentEvent passthrough (SSE data
lines + [DONE]); only app-level events are the loop-level error and
turn-settled terminal; onError carries raw provider text (owner ruling).

Client: useChat + the AI-SDK transport replaced by components/session-core.ts
(SessionModel: 409 identity-retry, optimistic c_ appends, absorbed()
adoption, needsContinuation, shared consume for POST + reattach) and the
session.ts hook; the golden-master grouping canon moved client-side into
core/view.ts (projectView); scanners read the native record (phaseAt3Lh,
danglingQuestionOf, graph toolParts, plan-blocks, runner, extraction).

Coverage: the native message is the unit (fork 3) with the call+result
inseparability bond rule; view-line-less messages (markers, error
results) are invisible to coverage and never checkpointed.

Deletes: chat-transport, projection, ui-message, their tests (ported to
tests/session-core.test.ts), and the ai + @ai-sdk/react dependencies —
the Vercel SDK is fully out of the tree. Wire pins in chat-flow rewritten
onto the AgentEvent dialect (eventsOf parser).

Gates: tsc clean, 994/994 tests, eslint clean, build green; live smoke
on PORT=3004 (verbatim wire → native record).
This commit is contained in:
amosblomqvist
2026-09-26 01:52:47 +03:00
parent 7c0e8d9a50
commit f3003d357a
44 changed files with 2299 additions and 2205 deletions
+13 -16
View File
@@ -1,5 +1,4 @@
import * as crypto from "node:crypto";
import type { UIMessage } from "ai";
import type { Usage } from "@earendil-works/pi-ai";
import type { AgentTool } from "@earendil-works/pi-agent-core";
import { Type, type Static, type TSchema } from "typebox";
@@ -28,7 +27,6 @@ import { getGraph, mergeFindings, type EmissionStats, type GraphDeps } from "../
import type { GraphStore } from "../../core/ports.ts";
import { settleGraphLayout } from "../../core/layout-settle.ts";
import { batchRuns, batchTag, coverageOf, messageViewLines } from "../../core/coverage.ts";
import { projectNative } from "../../core/projection.ts";
/**
* The extraction pipeline (vision §1 construction, 2026-09-02 redesign) — the
@@ -93,12 +91,11 @@ import { projectNative } from "../../core/projection.ts";
export function transcriptView(record: ChatRecord): string {
// The shared seam (core/coverage.ts messageViewLines): the LLM's prompt
// view and the coverage fingerprint use ONE line builder, so "covered"
// means — exactly — what the extractor saw. Wave B: the record is native
// history — the extraction view reads through the one-way projection.
// means — exactly — what the extractor saw. Wave D: the record is native
// history — the view reads the native messages directly.
const lines = [`SESSION: ${record.title}`];
const projected = projectNative(record.messages);
projected.forEach((message, index) => {
lines.push(...messageViewLines(message, message.id || `m${index}`));
record.messages.forEach((message, index) => {
lines.push(...messageViewLines(message, (message as { id?: string }).id || `m${index}`));
});
return lines.join("\n");
}
@@ -871,14 +868,16 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
// engines AND their providers' keys; a refusal lands BEFORE any
// stage runs (typed — the trigger skips, routes surface 503).
refuseIfStagesUnready();
const projected = projectNative(record.messages);
const covered = coverageOf(projected, graphStore.readCoverage(), chatId);
const covered = coverageOf(record.messages, graphStore.readCoverage(), chatId);
if (covered.uncovered.length === 0) {
console.info(`[extraction ${chatId}] nothing uncovered — no-op`);
return { nodes: 0, edges: 0, events: 0 };
}
const chunkTokens = deps.getChunkTokens?.() ?? FACTORY_SETTINGS.graph.extractionChunkTokens;
const batches = batchRuns(covered.uncovered, chunkTokens);
// The batcher's one rule (owner, Wave D): a tool-call message and
// its result message(s) are INSEPARABLE — one chunk, covered
// together, read together (core/coverage.ts batchRuns).
const batches = batchRuns(covered.slots, covered.uncovered, chunkTokens);
console.info(
`[extraction ${chatId}] pipeline start — ${covered.uncoveredTokens} uncovered tok, ` +
`${covered.uncoveredCount} messages, ${batches.length} batch(es) of ≤${chunkTokens} tok`,
@@ -887,13 +886,9 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
// The batch's prompt view: SESSION header + the batch's lines,
// same provenance [id] tags the fingerprints used (coverage
// and prompt share ONE line builder — core/coverage.ts).
const mids = projected.map((m, i) => (m.id || `m${i}`) as string);
const view = [
`SESSION: ${record.title}`,
...batch.flatMap((slot) => {
const idx = projected.indexOf(slot.message);
return messageViewLines(slot.message, mids[idx] ?? slot.message.id);
}),
...batch.flatMap((slot) => messageViewLines(slot.message, slot.message.id)),
].join("\n");
setPhase(chatId, "extracting", "chat");
const findings = await extract(chatId, view);
@@ -918,7 +913,9 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
const stats = mergeFindings(chatId, auditResult, { chatStore, graphStore }, {
messages: batch.map((slot) => slot.message),
foldedAt,
coverage: batch.map((slot) => ({ chatId, fp: slot.fp, tokens: slot.tokens, foldedAt })),
coverage: batch
.filter((slot) => messageViewLines(slot.message, slot.message.id).length > 0)
.map((slot) => ({ chatId, fp: slot.fp, tokens: slot.tokens, foldedAt })),
batchTag: batchTag(batch),
});
total.nodes += stats.nodes;
+10 -17
View File
@@ -1,4 +1,3 @@
import { isToolUIPart } from "ai";
import type { PiEngine } from "../models.ts";
import type { SearchProvider } from "../../core/search.ts";
import {
@@ -23,8 +22,6 @@ import { registerJobAbort, unregisterJobAbort } from "./aborts.ts";
import type { ModelRole } from "../../../config.ts";
import { ROLE_SLOT, modelsNotConfigured, type ModelSlotName, type ProviderKeyMissing } from "../../core/settings.ts";
import type { ChatStore, JobStore } from "../../core/ports.ts";
import { projectNative } from "../../core/projection.ts";
import type { UIMessage } from "ai";
import { runDemoBuilder } from "./demo-builder.ts";
import { runPdfReaderAgent } from "./pdf-reader.ts";
import { runResearcherAgent } from "./researcher.ts";
@@ -276,22 +273,18 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
if (input) return input;
const chat = chatStore.getChat(job.chatId);
if (!chat) return undefined;
// Wave B: the record is native history — the transcript scan reads
// through the projection (same shape the old record held).
return (findToolPart(projectNative(chat.messages), job.toolCallId)?.input ?? undefined) as
| JobToolInput
| undefined;
}
// ─── shared plumbing ────────────────────────────────────────────────────
function findToolPart(messages: UIMessage[], toolCallId: string) {
for (const m of messages) {
for (const part of m.parts) {
if (isToolUIPart(part) && part.toolCallId === toolCallId) return part;
// Wave D: the record is native history — the scan reads the native
// assistant blocks directly (a toolCall's arguments are the input).
for (const m of chat.messages) {
if ((m as { role?: string }).role !== "assistant") continue;
const content = (m as { content?: unknown }).content;
if (!Array.isArray(content)) continue;
for (const block of content as Array<{ type?: string; id?: string; arguments?: unknown }>) {
if (block.type !== "toolCall" || block.id !== job.toolCallId) continue;
return block.arguments as JobToolInput | undefined;
}
}
return null;
return undefined;
}
/**
+3 -5
View File
@@ -3,10 +3,8 @@
*
* Part of the domain-split contracts (src/core/contracts/): each domain file
* is owned outright by its stream — cross-domain schema changes go through
* master. Pure types + protocol docs + tiny pure helpers; type-only imports
* of "ai" (safe everywhere: server, client, tests). See docs/vision.md.
* master. Pure types + protocol docs + tiny pure helpers. See docs/vision.md.
*/
import type { UIMessage } from "ai";
import type { ThinkingLevel } from "@earendil-works/pi-agent-core";
import type { LhMessage } from "../native-messages.ts";
import type { Phase3 } from "../phases.ts";
@@ -20,8 +18,8 @@ import type { EmissionStats, UnderstandingLevel } from "./graph.ts";
* agent loop's own shapes, provider/model/usage/thinkingSignature stamped
* by pi) — the canonical session record IS the agent state; replay is
* `state.messages = record.messages` + continue(). The client-shaped view
* is the one-way projection (core/projection.ts) — a pending quiz is an
* unanswered question tool call, which is exactly what makes
* is the client's own grouping fold (core/view.ts, Wave D) — a pending quiz
* is an unanswered question tool call, which is exactly what makes
* resume-mid-question free. */
export interface ChatRecord {
id: string;
+25
View File
@@ -115,3 +115,28 @@ export interface PlanToolResult {
tree: PlanTree;
}
// ─── the native fold (Wave D: derivePlanStatuses over LhMessage[]) ───────────
import type { LhMessage } from "../native-messages.ts";
/** The same derivation over the NATIVE record: quiz CALLS only (any state —
* a dangling call still moves position; results feed nothing). */
export function derivePlanStatusesLh(tree: PlanTree, messages: readonly LhMessage[]): DerivedPlanStatuses {
const ids = new Set(tree.nodes.map((n) => n.id));
const runs: string[] = [];
for (const m of messages) {
if ((m as { role?: string }).role !== "assistant") continue;
const content = (m as { content?: unknown }).content;
if (!Array.isArray(content)) continue;
for (const block of content as Array<{ type?: string; name?: string; arguments?: unknown }>) {
if (block.type !== "toolCall" || block.name !== "quiz") continue;
const nodeId = (block.arguments as { nodeId?: unknown } | undefined)?.nodeId;
if (typeof nodeId !== "string" || !ids.has(nodeId)) continue;
if (runs.at(-1) !== nodeId) runs.push(nodeId);
}
}
const current = runs.at(-1) ?? null;
const visited = new Set(runs.slice(0, -1));
return { current, visited };
}
+185 -73
View File
@@ -1,63 +1,83 @@
/**
* Extraction coverage (vision-extraction-coverage.md, settled 2026-09-15) —
* the pure core. THE MESSAGE IS THE UNIT OF COVERAGE: a per-chat append-only
* ledger records which messages have been extracted, by CONTENT FINGERPRINT
* over exactly what the extraction view sees; "uncovered" is DERIVED at
* read time (fingerprint the current transcript, diff against the ledger) —
* never dual-tracked. Runs are token-budgeted batches of the oldest
* uncovered messages, so a tool call and its result (which share a message)
* can never be separated.
* the pure core. THE MESSAGE IS THE UNIT OF COVERAGE (Wave D, owner ruling
* 2026-09-26: the native message, not the old projected group): a per-chat
* append-only ledger records which messages have been extracted, by CONTENT
* FINGERPRINT over exactly what the extraction view sees; "uncovered" is
* DERIVED at read time (fingerprint the current transcript, diff against the
* ledger) — never dual-tracked. Runs are token-budgeted batches of the oldest
* uncovered messages, and THE ONE BATCH RULE (owner): a tool-call message and
* its tool-result message(s) are INSEPARABLE — they always appear in the same
* chunk (a covered partner rides along as context; checkpoints are
* idempotent, so re-marking it is a no-op).
*
* The fingerprint and the LLM prompt share ONE line builder —
* messageViewLines — so "covered" and "what the model saw" cannot drift
* apart. The line logic is tree-blind, identical to the extractor's view:
* user/teacher text verbatim, quiz outcomes as one-line summaries, other
* tool results as one generic line each, data parts invisible (plan-block
* attach does not re-open a covered message).
* clean tool results as one generic line each, markers invisible (a
* plan-block marker does not re-open a covered message).
*/
import { createHash } from "node:crypto";
import type { UIMessage } from "ai";
import { isToolUIPart } from "ai";
import type { CoverageEntry } from "./contracts/graph.ts";
import type { LhMessage } from "./native-messages.ts";
import type { ToolResultMessage } from "@earendil-works/pi-ai";
export type { CoverageEntry };
/** The quiz outcome line — both the fingerprint and the LLM view see the
* quiz answer, so an answered quiz re-opens the message that asked it. */
function quizLine(part: { toolCallId: string; input: unknown; output: unknown }): string {
const input = part.input as { question?: unknown } | undefined;
const out = part.output as Partial<{ questionText: string; correct: boolean; dontKnow: boolean; note: string }> | undefined;
const question = typeof input?.question === "string" ? input.question : out?.questionText ?? "(no question)";
* quiz answer, so an answered quiz re-opens (the answer is its own native
* message now; its fingerprint is new the moment it lands). */
function quizLine(result: ToolResultMessage): string {
const out = result.details as
| Partial<{ questionText: string; correct: boolean; dontKnow: boolean; note: string }>
| undefined;
const question = typeof out?.questionText === "string" ? out.questionText : "(no question)";
const outcome = out?.dontKnow === true ? "did not know" : out?.correct === true ? "answered correctly" : "answered incorrectly";
const note = typeof out?.note === "string" && out.note ? ` (note: ${out.note})` : "";
return `QUIZ [${part.toolCallId}] ${question} — the learner ${outcome}${note}`;
return `QUIZ [${result.toolCallId}] ${question} — the learner ${outcome}${note}`;
}
/** The user-facing text of a native message (string or text blocks). */
function textOf(content: unknown): string[] {
if (typeof content === "string") return content.trim() ? [content] : [];
if (!Array.isArray(content)) return [];
return content
.filter((c): c is { type: "text"; text: string } => (c as { type?: string }).type === "text")
.map((c) => c.text)
.filter((t) => t.trim().length > 0);
}
/**
* The extraction-relevant lines of ONE message — the shared seam between
* transcriptView (the LLM's prompt) and messageFingerprint (coverage). Same
* logic, same order, so a covered message is by construction exactly what
* the extractor saw. `mid` is the provenance tag (the message id; the
* transcript view passes an index fallback for legacy id-less messages,
* coverage passes the id — persisted messages always have ids).
* The extraction-relevant lines of ONE native message — the shared seam
* between transcriptView (the LLM's prompt) and messageFingerprint
* (coverage). Same logic, same order, so a covered message is by
* construction exactly what the extractor saw. `mid` is the provenance tag
* (the message id; the transcript view passes an index fallback for legacy
* id-less messages, coverage passes the id — persisted messages always have
* ids).
*/
export function messageViewLines(message: UIMessage, mid: string): string[] {
export function messageViewLines(message: LhMessage, mid: string): string[] {
const lines: string[] = [];
const who = message.role === "user" ? "LEARNER" : "TEACHER";
for (const part of message.parts) {
if (part.type === "text" && part.text.trim()) {
lines.push(`${who} [${mid}] ${part.text.trim()}`);
} else if (isToolUIPart(part)) {
const name = part.type.slice("tool-".length);
if (name === "quiz" && part.state === "output-available") {
lines.push(quizLine(part as unknown as { toolCallId: string; input: unknown; output: unknown }));
} else if (part.state === "output-available") {
lines.push(`(${name} tool result delivered)`);
}
const role = (message as { role?: string }).role;
if (role === "user") {
for (const text of textOf((message as { content: unknown }).content)) {
lines.push(`LEARNER [${mid}] ${text.trim()}`);
}
} else if (role === "assistant") {
for (const text of textOf((message as { content: unknown }).content)) {
lines.push(`TEACHER [${mid}] ${text.trim()}`);
}
} else if (role === "toolResult") {
const result = message as ToolResultMessage;
if (result.isError) return lines; // a failed call is retry context, not learning evidence
lines.push(
result.toolName === "quiz" ? quizLine(result) : `(${result.toolName} tool result delivered)`,
);
}
// custom markers are invisible to the extraction view.
return lines;
}
@@ -69,33 +89,28 @@ export function viewTokens(lines: string[]): number {
/**
* A message's coverage fingerprint: sha256 of role + its view lines (with
* the message's real id as the provenance tag). Changes exactly when the
* extraction view of the message changes: an answered quiz merges into the
* asking message → new line → the message re-opens itself, and only itself.
* Plan-block part attach changes data parts (invisible to the view) → the
* fingerprint is stable → still covered.
* extraction view of the message changes. (Wave D: the unit is the native
* message — an answered quiz is the answer's OWN toolResult message, so
* answering opens exactly that new message; the asking message is pulled
* into the batch by the inseparability rule, not re-opened.)
*/
export function messageFingerprint(message: UIMessage): string {
export function messageFingerprint(message: LhMessage): string {
const h = createHash("sha256");
h.update(JSON.stringify([message.role, messageViewLines(message, message.id)]));
h.update(JSON.stringify([(message as { role?: string }).role, messageViewLines(message, message.id)]));
return h.digest("hex");
}
/** One message's coverage-relevant projection. */
export interface CoverageSlot {
message: UIMessage;
message: LhMessage;
fp: string;
tokens: number;
}
/** Fingerprint a whole transcript, in order (used by the trigger, the run,
* and the coverage view — one derivation everywhere). */
export function coverageSlots(messages: ReadonlyArray<UIMessage>): CoverageSlot[] {
return messages.map((message) => {
const lines = messageViewLines(message, message.id);
const h = createHash("sha256");
h.update(JSON.stringify([message.role, lines]));
return { message, fp: h.digest("hex"), tokens: viewTokens(lines) };
});
export function coverageSlots(messages: ReadonlyArray<LhMessage>): CoverageSlot[] {
return messages.map((message) => ({ message, fp: messageFingerprint(message), tokens: viewTokens(messageViewLines(message, message.id)) }));
}
/** The covered-fingerprint set for one chat, from the ledger. */
@@ -106,14 +121,19 @@ export function coveredSet(entries: ReadonlyArray<CoverageEntry>, chatId: string
}
/** The derived state: which slots are uncovered, plus the token totals.
* "Uncovered" needs no ledger of its own — this diff IS the coverage view. */
* "Uncovered" needs no ledger of its own — this diff IS the coverage view.
* `uncovered` is INDEXES into `slots` (the batcher needs positions for the
* inseparability rule). Messages with NO view lines (custom markers, an
* error result) are invisible to coverage entirely: the extractor can
* never see them, so they are neither covered nor uncovered — a marker
* attach never opens (or joins) an extraction run. */
export function coverageOf(
messages: ReadonlyArray<UIMessage>,
messages: ReadonlyArray<LhMessage>,
entries: ReadonlyArray<CoverageEntry>,
chatId: string,
): {
slots: CoverageSlot[];
uncovered: CoverageSlot[];
uncovered: number[];
coveredTokens: number;
uncoveredTokens: number;
coveredCount: number;
@@ -121,40 +141,132 @@ export function coverageOf(
} {
const covered = coveredSet(entries, chatId);
const slots = coverageSlots(messages);
const uncovered = slots.filter((s) => !covered.has(s.fp));
const coveredCount = slots.length - uncovered.length;
const uncovered: number[] = [];
let coveredTokens = 0;
let uncoveredTokens = 0;
for (const s of slots) {
if (covered.has(s.fp)) coveredTokens += s.tokens;
else uncoveredTokens += s.tokens;
for (let i = 0; i < slots.length; i++) {
const slot = slots[i]!;
if (messageViewLines(slot.message, slot.message.id).length === 0) continue; // invisible
if (covered.has(slot.fp)) coveredTokens += slot.tokens;
else {
uncovered.push(i);
uncoveredTokens += slot.tokens;
}
}
return { slots, uncovered, coveredTokens, uncoveredTokens, coveredCount, uncoveredCount: uncovered.length };
const visibleCount = slots.length - slots.filter(
(s) => messageViewLines(s.message, s.message.id).length === 0,
).length;
return {
slots,
uncovered,
coveredTokens,
uncoveredTokens,
coveredCount: visibleCount - uncovered.length,
uncoveredCount: uncovered.length,
};
}
// ─── the inseparability rule (owner, Wave D fork 3) ─────────────────────────
//
// A tool-call message and its tool-result message(s) always appear in the
// same chunk. Bonding is adjacency-honest: results follow their call's
// assistant message contiguously in the native record (the agent loop
// appends the results batch; the question gate guarantees nothing can land
// between a quiz call and its learner answer). A result whose call's
// assistant message has NO other bond partners still pulls that message in
// (the extractor needs the question with its answer).
/** The call ids an assistant message carries. */
function callIdsOf(message: LhMessage): Set<string> {
const out = new Set<string>();
const content = (message as { content?: unknown }).content;
if (!Array.isArray(content)) return out;
for (const block of content as Array<{ type?: string; id?: string }>) {
if (block.type === "toolCall" && typeof block.id === "string") out.add(block.id);
}
return out;
}
/** The contiguous bond-range containing slot i: an assistant-with-calls
* message plus its immediately-following results; a toolResult message
* plus the assistant message that made its call (and that call's sibling
* results). Anything else bonds to itself alone. */
function bondRange(messages: ReadonlyArray<LhMessage>, i: number): [number, number] {
const roleOf = (m: LhMessage | undefined) => (m ? (m as { role?: string }).role : undefined);
const m = messages[i];
if (!m) return [i, i];
if (roleOf(m) === "assistant") {
const calls = callIdsOf(m);
if (calls.size === 0) return [i, i];
let end = i;
while (
roleOf(messages[end + 1]) === "toolResult" &&
calls.has((messages[end + 1] as ToolResultMessage).toolCallId)
) {
end++;
}
return [i, end];
}
if (roleOf(m) === "toolResult") {
let start = i;
while (roleOf(messages[start - 1]) === "toolResult") start--;
const owner = messages[start - 1];
if (
owner &&
roleOf(owner) === "assistant" &&
callIdsOf(owner).has((m as ToolResultMessage).toolCallId)
) {
const [s, e] = bondRange(messages, start - 1);
return [s, Math.max(e, i)];
}
return [start, i]; // orphaned results (no asking message): keep the contiguous result run together
}
return [i, i];
}
/**
* Batch the uncovered messages into token-budgeted RUNS: consecutive slots,
* oldest first, greedy fill — a message is never split (it is the unit), so
* a tool call and its result always travel together. A single message larger
* than the budget becomes its own (over-budget) run rather than being
* dropped. Runs are ephemeral — the durable record is per-message coverage,
* checkpointed after each run merges.
* oldest first, greedy fill — a message is never split, and a bond-range
* (call + results) never splits either (owner fork 3): a selected uncovered
* slot EXPANDS to its whole bond-range, covered partners riding as context
* (checkpoint idempotency makes re-marking them a no-op). A single
* bond-range larger than the budget becomes its own (over-budget) run
* rather than being dropped. Runs are ephemeral — the durable record is
* per-message coverage, checkpointed after each run merges.
*/
export function batchRuns(uncovered: ReadonlyArray<CoverageSlot>, budget: number): CoverageSlot[][] {
export function batchRuns(slots: ReadonlyArray<CoverageSlot>, uncovered: ReadonlyArray<number>, budget: number): CoverageSlot[][] {
if (budget <= 0 || uncovered.length === 0) return [];
const messages = slots.map((s) => s.message);
// Expand: every uncovered index marks its whole bond-range as needed.
const needed = new Set<number>();
for (const i of uncovered) {
const [start, end] = bondRange(messages, i);
for (let j = start; j <= end; j++) needed.add(j);
}
// Group the needed slots in order, greedy by budget, never splitting a
// bond-range: a range that would overflow starts a new run (and an
// over-budget range rides alone).
const runs: CoverageSlot[][] = [];
let current: CoverageSlot[] = [];
let currentTokens = 0;
for (const slot of uncovered) {
if (current.length > 0 && currentTokens + slot.tokens > budget) {
runs.push(current);
current = [];
currentTokens = 0;
const flush = () => {
if (current.length > 0) runs.push(current);
current = [];
currentTokens = 0;
};
for (let i = 0; i < slots.length; i++) {
if (!needed.has(i)) continue;
const [start, end] = bondRange(messages, i);
const rangeTokens = slots.slice(start, end + 1).reduce((t, s) => t + s.tokens, 0);
if (current.length > 0 && currentTokens + rangeTokens > budget) flush();
// The whole range enters together — skip past its end in the outer loop.
for (let j = start; j <= end; j++) {
current.push(slots[j]!);
currentTokens += slots[j]!.tokens;
}
current.push(slot);
currentTokens += slot.tokens;
i = end;
}
if (current.length > 0) runs.push(current);
flush();
return runs;
}
@@ -171,4 +283,4 @@ export function batchTag(batch: ReadonlyArray<CoverageSlot>): string {
const h = createHash("sha256");
for (const slot of batch) h.update(slot.fp);
return h.digest("hex").slice(0, 12);
}
}
+27 -12
View File
@@ -39,7 +39,8 @@
*/
import { randomBytes } from "node:crypto";
import { isToolUIPart, type UIMessage } from "ai";
import type { LhMessage } from "./native-messages.ts";
import type { ToolResultMessage } from "@earendil-works/pi-ai";
import type { CoverageEntry } from "./contracts/graph.ts";
import type {
AuditResult,
@@ -85,7 +86,7 @@ function newGraphNodeId(): string {
* fresh understanding observations land as new events; the notes path
* passes neither and keeps the whole-file checkpoint. */
export interface MergeSource {
messages: UIMessage[] | null;
messages: LhMessage[] | null;
foldedAt: string;
/** The batch being merged (chat path): one entry per message in
* `messages`, same order — undefined (notes path / coverage-less
@@ -97,7 +98,11 @@ export interface MergeSource {
// ─── transcript scanning (shared by the merger + trigger) ─────────────────────
/** A flattened, order-preserving view of every tool part in a transcript. */
/** A flattened, order-preserving view of every tool call (+ its result
* state) in a native transcript. Wave D: the pairing is native — calls
* come from assistant content blocks; a result map built from the
* toolResult messages carries the outcome (a dangling call is
* input-available, an error result is output-error). */
interface ToolPartView {
name: string;
toolCallId: string;
@@ -106,17 +111,27 @@ interface ToolPartView {
output: unknown;
}
function toolParts(messages: UIMessage[]): ToolPartView[] {
function toolParts(messages: LhMessage[]): ToolPartView[] {
const results = new Map<string, ToolResultMessage>();
for (const m of messages) {
if ((m as { role?: string }).role === "toolResult") {
results.set((m as ToolResultMessage).toolCallId, m as ToolResultMessage);
}
}
const out: ToolPartView[] = [];
for (const message of messages) {
for (const part of message.parts) {
if (!isToolUIPart(part)) continue;
for (const m of messages) {
if ((m as { role?: string }).role !== "assistant") continue;
const content = (m as { content?: unknown }).content;
if (!Array.isArray(content)) continue;
for (const block of content as Array<{ type?: string; id?: string; name?: string; arguments?: unknown }>) {
if (block.type !== "toolCall" || typeof block.id !== "string") continue;
const result = results.get(block.id);
out.push({
name: part.type.slice("tool-".length),
toolCallId: part.toolCallId,
state: part.state,
input: part.input,
output: part.output,
name: block.name ?? "",
toolCallId: block.id,
state: result ? (result.isError ? "output-error" : "output-available") : "input-available",
input: block.arguments,
output: result && !result.isError ? result.details : undefined,
});
}
}
+5 -3
View File
@@ -31,13 +31,15 @@ export * from "./phases.ts";
export * from "./pending.ts";
export * from "./prompt-segments.ts";
export * from "./math-delims.ts";
export * from "./ui-message.ts";
// Wave B (owner-approved 2026-09-26): the native record (LhMessage), the
// one-way client-view projection, and the input-events contract — all
// and the input-events contract — all
// pure logic, browser-safe.
export * from "./native-messages.ts";
export * from "./projection.ts";
export * from "./input-events.ts";
// Wave D (owner-approved 2026-09-26): the client view model — the grouping
// canon (one continuous assistant document, step separators) ported from
// the deleted server projection to the client side. Pure, browser-safe.
export * from "./view.ts";
export * from "./tool-parts.ts";
export * from "./prices.ts";
export * from "./cost-summary.ts";
+45
View File
@@ -282,3 +282,48 @@ export function hasProbeDone(messages: readonly LhMessage[]): boolean {
export function answeredCallIds(messages: readonly LhMessage[]): Set<string> {
return resultIdsFrom(messages);
}
// ─── the continuation fold (Wave D: the SDK's
// lastAssistantMessageIsCompleteWithToolCalls, natively) ──────────────────
/**
* Does the record END on un-consumed tool results — a turn that stopped
* after its tools ran (a job dispatch's ack, an aborted answer turn),
* leaving the agent owing a continuation? This is the mount/sync
* continuation seam's gate (chat.tsx): when true and nothing runs, the
* client sends {type:"continue"} and the teacher picks up from the record.
*
* The pre-Wave-D client asked the SDK's
* lastAssistantMessageIsCompleteWithToolCalls (last view message is
* assistant; its last step's tool calls all output-available or
* output-error; at least one). The native fold over the record: the last
* message is a toolResult, and the results' OWNING assistant message has
* at least one call, ALL of them answered (a dangling question — the
* dock's domain — fails the all-answered rule and never fires a turn,
* exactly as the SDK's every(output-available) did).
*/
export function needsContinuation(messages: readonly LhMessage[]): boolean {
const last = messages[messages.length - 1];
if (!last || (last as { role?: string }).role !== "toolResult") return false;
// The trailing results batch (contiguous by construction — the agent
// appends its batch; the question gate keeps a quiz call and its
// learner answer adjacent).
let start = messages.length - 1;
while (
start > 0 &&
(messages[start - 1] as { role?: string } | undefined)?.role === "toolResult"
) {
start--;
}
const owner = messages[start - 1];
if (!owner || (owner as { role?: string }).role !== "assistant") return false;
const content = (owner as { content?: unknown }).content;
if (!Array.isArray(content)) return false;
const answered = new Set(
messages.slice(start).map((m) => (m as ToolResultMessage).toolCallId),
);
const calls = (content as Array<{ type?: string; id?: string }>).filter(
(b) => b.type === "toolCall" && typeof b.id === "string",
);
return calls.length > 0 && calls.every((b) => answered.has(b.id!));
}
+16 -2
View File
@@ -1,4 +1,18 @@
import { isToolUIPart, type UIMessage } from "ai";
/**
* Wave D: these scans read VIEW messages (core/view.ts LhViewMessage — the
* client's grouped transcript). The part shapes are the same vocabulary the
* pre-Wave-D client dialect used; a tool part is structural: `tool-` prefix +
* toolCallId. The server-side record scans (danglingQuestionOf over native
* LhMessage[]) live in native-messages.ts.
*/
interface UIMessage {
role?: string;
parts: Array<Record<string, unknown> & { type: string }>;
}
function isToolUIPart(part: UIMessage["parts"][number]): boolean {
return part.type.startsWith("tool-") && typeof part.toolCallId === "string";
}
/**
* The pending-question gate: an unanswered QUESTION tool part (quiz/ask)
@@ -76,7 +90,7 @@ function findPending(messages: UIMessage[], names: readonly string[]): PendingTo
if (isToolUIPart(part) && part.state === "input-available") {
const tool = part.type.slice("tool-".length);
if (names.includes(tool)) {
return { toolCallId: part.toolCallId, tool, input: part.input };
return { toolCallId: part.toolCallId as string, tool, input: part.input };
}
}
}
+51 -7
View File
@@ -1,15 +1,14 @@
/**
* Phase derivation (docs/probe-teach-split.md §4) — EXPLICIT EVENTS, never
* occupancy. The probe→teach agent split (2026-09-11) gave every phase
* boundary a transcript marker:
* boundary a transcript event:
*
* probe — until the first message containing a `tool-probe_done` part
* (the prober concludes; the assessment rides the tool part)
* probe — until the prober's `probe_done` call CONCLUDES (a clean
* toolResult for it; the assessment rides the call's arguments)
* plan — probe concluded, until the learner ACCEPTS the presented
* plan (a `data-verdict` part with accepted:true on the verdict
* button's own user message — the explicit verdict gate;
* plan_init itself is only the PRESENTATION and moves no
* phase; a declined verdict stays in plan)
* plan (the verdict marker with accepted:true — the explicit
* verdict gate; plan_init itself is only the PRESENTATION and
* moves no phase; a declined verdict stays in plan)
* teach — teaching is sanctioned
*
* The former derived-occupancy seam (phase-of-tree via `currentNodeId`)
@@ -18,6 +17,14 @@
* progress fields at all. No backwards compatibility: legacy sessions
* without probe_done simply read as probe (they are wiped data).
*
* TWO FOLDS over the same events (Wave D): `phaseAt3`/`phaseTimeline` fold
* VIEW messages (the client's dividers and dock — part shapes, pure and
* structural); `phaseAt3Lh` folds the NATIVE record (server summaries,
* tests) — probe→plan advances at the probe_done RESULT message, plan→teach
* at the verdict marker. Both are strictly sequential; the two derivations
* agree on every settled record (a result/marker message projects into the
* same view boundary).
*
* Pure: no imports, no runtime deps — safe on server, client, and tests.
*/
@@ -83,3 +90,40 @@ export function phaseTimeline(messages: readonly PhaseMessage[]): Phase3[] {
}
return out;
}
// ─── the native fold (Wave D: record-side phase over LhMessage[]) ───────────
import type { LhMessage } from "./native-messages.ts";
import type { ToolResultMessage } from "@earendil-works/pi-ai";
/** The phase at `upToIndex` (inclusive; default: the last message) of a
* NATIVE record. probe→plan advances at the probe_done call's clean
* toolResult; plan→teach at the verdict marker (details.accepted true). */
export function phaseAt3Lh(messages: readonly LhMessage[], upToIndex?: number): Phase3 {
let phase: Phase3 = "probe";
const last = Math.min(upToIndex ?? messages.length - 1, messages.length - 1);
for (let i = 0; i <= last; i++) {
const m = messages[i]!;
const role = (m as { role?: string }).role;
if (role === "toolResult") {
const result = m as ToolResultMessage;
if (
phase === "probe" &&
result.toolName === "probe_done" &&
!result.isError
) {
phase = "plan";
}
} else if (role === "custom") {
const marker = m as { customType?: string; details?: { accepted?: unknown } };
if (
phase === "plan" &&
marker.customType === "verdict" &&
marker.details?.accepted === true
) {
phase = "teach";
}
}
}
return phase;
}
+9 -8
View File
@@ -16,8 +16,7 @@
*/
import type { PlanBlockData, PlanTree } from "./contracts.ts";
import { derivePlanStatuses } from "./contracts.ts";
import type { UIMessage } from "ai";
import { derivePlanStatusesLh } from "./contracts.ts";
import type { LhMessage } from "./native-messages.ts";
import { newPlanBlockMessage } from "./native-messages.ts";
import type { ToolResultMessage } from "@earendil-works/pi-ai";
@@ -32,9 +31,11 @@ export interface PlanBlockPart {
}
/** The snapshot payload (unchanged since wave 4): statuses derived AT
* EMISSION time from the transcript — the block never re-derives. */
function snapshot(tree: PlanTree, messages: readonly UIMessage[]): PlanBlockData {
const statuses = derivePlanStatuses(tree, messages);
* EMISSION time from the transcript — the block never re-derives. Wave D:
* the derivation reads the NATIVE transcript (quiz CALLS only — results
* feed nothing, contracts/plan.ts derivePlanStatusesLh). */
function snapshot(tree: PlanTree, messages: readonly LhMessage[]): PlanBlockData {
const statuses = derivePlanStatusesLh(tree, messages);
return {
kind: "full",
at: new Date().toISOString(),
@@ -68,14 +69,14 @@ export function emittedPlanBlockCalls(messages: readonly LhMessage[]): Set<strin
* once; a later turn's rewrite gets its own block — the golden-master
* rule the UIMessage-era attach had) that the record hasn't already
* emitted. `tree` is the record's current tree (the plan tool saved it
* during the turn); `transcript` is the full PROJECTED transcript the
* status derivation reads.
* during the turn); `transcript` is the full NATIVE transcript the status
* derivation reads.
*/
export function planBlocksFor(
record: readonly LhMessage[],
newMessages: readonly LhMessage[],
tree: PlanTree | null | undefined,
transcript: readonly UIMessage[],
transcript: readonly LhMessage[],
): LhMessage[] {
if (!tree || tree.nodes.length === 0) return [];
const emitted = emittedPlanBlockCalls(record);
-199
View File
@@ -1,199 +0,0 @@
/**
* THE ONE-WAY PROJECTION (Wave B, owner-approved 2026-09-26): native record
* messages → AI-SDK UIMessages. The record is pi-native history; every
* existing consumer — GET /api/chats/[id], the client's renderer, phase/
* pending/coverage/extraction scanners — reads THIS projection until Wave
* D replaces the client dialect. One direction only: nothing ever converts
* UIMessages back into native messages (the lossy record-adapter direction
* is deleted wholesale).
*
* GROUPING RULE (the golden master): a turn's UIMessage shape is one
* assistant message per GROUP — consecutive assistant + toolResult
* messages, bounded by user/marker messages; toolResults attach to the
* tool part of their call (searching backward — an answered quiz's result
* lands on the part that asked it, even turns later); a group's id is its
* FIRST assistant message's id. This reproduces the client's live view
* exactly: the continuation merge (start-chunk id = the group's id) and
* the settled record converge to the same shape, so the settle sync never
* re-renders what streamed.
*
* Plan blocks (plan-blocks.ts's splice, moved here): a plan-block marker
* has no message of its own — its data part splices into the preceding
* assistant group, IN FLOW right after the last successful plan_init call
* (the presentation moment).
*/
import type { UIMessage } from "ai";
import type { AssistantMessage, ToolResultMessage } from "@earendil-works/pi-ai";
import type { LhMarkerMessage, LhMessage } from "./native-messages.ts";
import type { SubagentArtifactData, SubagentResultMeta } from "./contracts/jobs.ts";
import type { PlanBlockData, PlanVerdictData } from "./contracts/chat.ts";
/** Loose part shapes the projection emits (the client's part shapes). */
type Part = Record<string, unknown> & { type: string };
type LooseMessage = { id: string; role: "user" | "assistant"; parts: Part[]; metadata?: unknown };
function textPart(text: string): Part {
return { type: "text", text, state: "done" };
}
/** The result a tool part should show, from a native toolResult. */
function applyResult(part: Part, result: ToolResultMessage): void {
if (result.isError) {
part.state = "output-error";
part.rawInput = part.input;
const text = result.content.find((c) => c.type === "text");
part.errorText = text && text.type === "text" ? text.text : "The tool call failed.";
} else {
part.state = "output-available";
part.output = result.details;
}
}
/** Does this assistant group carry a successful plan_init (the splice target)? */
function lastSuccessfulPlanInit(parts: Part[]): number {
let idx = -1;
parts.forEach((p, i) => {
if (p.type !== "tool-plan_init") return;
const ok = (p.output as { ok?: unknown } | undefined)?.ok;
if (ok === true) idx = i;
});
return idx;
}
/** Project the native record into client-shaped UIMessages. Pure and
* deterministic: the same record always projects to the same list. */
export function projectNative(messages: readonly LhMessage[]): UIMessage[] {
const out: LooseMessage[] = [];
const lastAssistant = (): LooseMessage | undefined => {
const last = out[out.length - 1];
return last && last.role === "assistant" ? last : undefined;
};
const findToolPart = (toolCallId: string): Part | undefined => {
for (let i = out.length - 1; i >= 0; i--) {
for (const part of out[i]!.parts) {
if (part.type.startsWith("tool-") && part.toolCallId === toolCallId) return part;
}
}
return undefined;
};
for (const m of messages) {
const role = (m as { role?: string }).role;
if (role === "user") {
const content = m.content as string | Array<{ type: string; text?: string }>;
const texts = typeof content === "string" ? [content] : content.filter((c) => c.type === "text").map((c) => (c as { text: string }).text);
out.push({ id: m.id, role: "user", parts: texts.map(textPart) });
continue;
}
if (role === "custom") {
const marker = m as LhMarkerMessage;
projectMarker(marker, out);
continue;
}
if (role === "toolResult") {
const result = m as ToolResultMessage;
const part = findToolPart(result.toolCallId);
if (part) applyResult(part, result);
continue;
}
if (role === "assistant") {
const assistant = m as AssistantMessage;
const parts: Part[] = [];
let textN = 0;
let reasoningN = 0;
for (const block of assistant.content) {
if (block.type === "text") {
parts.push({ type: "text", id: `${m.id}-t${textN++}`, text: block.text, state: "done" });
} else if (block.type === "thinking") {
parts.push({ type: "reasoning", id: `${m.id}-r${reasoningN++}`, text: block.thinking, state: "done" });
} else if (block.type === "toolCall") {
parts.push({
type: `tool-${block.name}`,
toolCallId: block.id,
state: "input-available",
input: block.arguments,
});
}
}
if (parts.length === 0) continue; // presentation-only (nothing renderable)
const group = lastAssistant();
if (group) {
// Same group: the step separator renders between the steps.
group.parts.push({ type: "step-start" }, ...parts);
} else {
out.push({ id: m.id, role: "assistant", parts: [{ type: "step-start" }, ...parts] });
}
continue;
}
}
return out as unknown as UIMessage[];
}
/** A marker's projected message (verdict/handoff/steers) or splice (plan block). */
function projectMarker(marker: LhMarkerMessage, out: LooseMessage[]): void {
switch (marker.customType) {
case "verdict": {
const data = marker.details as PlanVerdictData;
out.push({
id: marker.id,
role: "user",
parts: [{ type: "data-verdict", id: `vd_${marker.id}`, data }],
});
return;
}
case "probe-handoff": {
out.push({
id: marker.id,
role: "user",
parts: [{ type: "data-probe-handoff", id: `ph_${marker.id}`, data: marker.details }],
});
return;
}
case "subagent-steer": {
const meta = marker.details as SubagentResultMeta;
const artifact = marker.details as { kind: SubagentArtifactData["kind"]; result: SubagentArtifactData["result"] };
out.push({
id: marker.id,
role: "user",
metadata: { subagentResult: meta },
parts: [
textPart(marker.content),
{ type: "data-subagent-artifact", id: `sa_${marker.id}`, data: { kind: artifact.kind, result: artifact.result } },
],
});
return;
}
case "subagent-abort": {
const meta = marker.details as SubagentResultMeta & { aborted: true };
out.push({
id: marker.id,
role: "user",
metadata: { subagentResult: meta },
parts: [
textPart(marker.content),
{ type: "data-subagent-aborted", id: `sa_${marker.id}`, data: meta },
],
});
return;
}
case "plan-block": {
// No message of its own: splice into the preceding assistant
// group, IN FLOW after the last successful plan_init call.
const group = out[out.length - 1];
if (!group || group.role !== "assistant") return;
const idx = lastSuccessfulPlanInit(group.parts);
if (idx === -1) return;
const { block } = marker.details as { planCallId: string; block: PlanBlockData };
group.parts.splice(idx + 1, 0, { type: "data-plan-block", id: marker.id, data: block });
return;
}
}
}
-120
View File
@@ -1,120 +0,0 @@
import type { UIMessage } from "ai";
import type {
DemoToolInput,
PdfToolInput,
PlanVerdictData,
PlanBlockData,
ProbeDoneInput,
ProbeDoneResult,
ResearchToolInput,
SubagentArtifactData,
SubagentResultMeta,
VisualToolInput,
} from "./contracts.ts";
import type { AskToolInput, QuestionAnswerData, QuizToolInput } from "./question-core.ts";
/**
* The typed UI message for the teacher chat — gives the frontend typed tool
* parts ("tool-quiz" / "tool-ask_user_question" / "tool-research" / …) without
* importing the server-side agent into the client bundle.
*
* Question tools are CLIENT-SIDE (no execute): their OUTPUT type is the
* structured answer record the client builds and sends via addToolOutput.
* Job tools are SERVER-EXECUTED: their output is the plain-string dispatch
* acknowledgement (the result later arrives as a steer message — see
* SubagentSteerMessage in contracts.ts, whose artifact rides the
* data-subagent-artifact part typed here as DATA_PARTS).
*/
export type TeacherUITools = {
quiz: { input: QuizToolInput; output: QuestionAnswerData };
ask_user_question: { input: AskToolInput; output: QuestionAnswerData };
/** The prober's conclusion tool (docs/probe-teach-split.md §2): the
* input IS the learner assessment; the output confirms the handoff. */
probe_done: { input: ProbeDoneInput; output: ProbeDoneResult };
research: { input: ResearchToolInput; output: string };
demo: { input: DemoToolInput; output: string };
visual: { input: VisualToolInput; output: string };
/** The pdf dispatch tool (sources v0): fire-and-forget ack output like
* research — the brief arrives later as a steer message. */
pdf: { input: PdfToolInput; output: string };
/** The learner-filesystem tools (sources v1, both agents, always-on):
* read may return an image-bearing result; the rest are plain text. */
read: { input: { path: string; offset?: number; limit?: number }; output: unknown };
ls: { input: { path?: string; limit?: number }; output: unknown };
find: { input: { pattern: string; path?: string; limit?: number }; output: unknown };
grep: {
input: {
pattern: string;
path?: string;
glob?: string;
ignoreCase?: boolean;
literal?: boolean;
context?: number;
limit?: number;
};
output: unknown;
};
};
export type TeacherUIDataParts = {
"subagent-artifact": SubagentArtifactData;
/** Plan blocks (wave 4): frozen plan snapshots rendered in the chat flow. */
"plan-block": PlanBlockData;
/** The learner's plan verdict (accept/decline): its own user message,
* carrying ONLY this part — the plan→teach phase seam (docs/probe-
* teach-split.md §4, the custom-message scheme 2026-09-12). */
verdict: PlanVerdictData;
/** The probe→teach handoff wake: the wake pipeline's steer message,
* carrying ONLY this part (the model-facing text lives in the adapter). */
"probe-handoff": { chatId: string };
/** The abort steer's message-type marker (the abort badge renders off it). */
"subagent-aborted": SubagentResultMeta & { aborted: true };
};
export type TeacherUIMessage = UIMessage<unknown, TeacherUIDataParts, TeacherUITools>;
// ─── Append-only record adoption (background-process ground rules) ──────────
/**
* Merge server-added messages into the client's live list WITHOUT the full
* snapshot swap (owner ground rules, 2026-09-15: content delivered into the
* record — context blocks, sub-agent artifacts, abort notes — must become
* visible the moment it persists, mid-turn included; it sits between messages
* and nothing about a running turn should hide it).
*
* Safe against the stale-snapshot lesson (mid-turn full syncs break the SDK's
* in-place continuation merge): this NEVER rewrites or removes an existing
* message — it returns the RECORD's message order when that order is exactly
* the client's persisted core plus new server-added ids (the server appends
* only: steers are inserted right after the last persisted user message),
* with the client's in-flight streaming assistant message — its last entry,
* not yet persisted — kept at the tail. Any other shape (a missing local id
* = the record was rewritten/reordered) returns null: the caller defers to
* the full sync at idle, the pre-existing safety net.
*/
export function mergeAdopted(
local: ReadonlyArray<TeacherUIMessage>,
record: ReadonlyArray<TeacherUIMessage>,
): TeacherUIMessage[] | null {
const recordIds = new Set(record.map((m) => m.id));
// The in-flight tail: the client's last message may legitimately be absent
// from the record (the SDK's streaming assistant message persists only at
// onFinish). Exactly one such tail is allowed.
let core = local;
let tail: TeacherUIMessage | undefined;
const last = local.at(-1);
if (last && !recordIds.has(last.id)) {
tail = last;
core = local.slice(0, -1);
}
// No rewrite happened if every core message is still in the record —
// otherwise this is not a pure append; bail to the full-sync-at-idle.
if (core.some((m) => !recordIds.has(m.id))) return null;
// Nothing new → nothing to do.
const localIds = new Set(local.map((m) => m.id));
if (!record.some((m) => !localIds.has(m.id))) return null;
// The record's order is canonical for everything persisted (the server
// appends only; plan-block part attachment changes parts, not order) —
// merged = record + the in-flight tail.
return tail ? [...record, tail] : [...record];
}
+289
View File
@@ -0,0 +1,289 @@
/**
* THE VIEW MODEL (Wave D, owner-approved 2026-09-26): the client's transcript
* view — native record messages (+ the live turn's working messages) grouped
* into today's exact render shape. The golden master is the pre-Wave-D UI
* (owner fork 1): everything the agent produces is ONE continuous document
* per run — step separators between the loop's steps, artifact and answered
* cards inline, the teacher's reaction continuing below an answered quiz.
*
* This is the grouping canon of the deleted server-side projection
* (core/projection.ts, Wave B), moved client-side, app-owned, with zero AI-SDK
* types — plus the LIVE-TURN layer the projection never had: mid-turn tool
* execution states (input-streaming → input-available/output-error/output),
* so the dock mounts only after a call clears the validation boundary
* (tool_execution_end) and the shimmer names the running activity.
*
* GROUPING RULE: one assistant view message per GROUP — consecutive assistant
* + toolResult messages, bounded by user/marker messages; toolResults attach
* to the tool part of their call (searching backward — an answered quiz's
* result lands on the part that asked it, even turns later); a group's id is
* its FIRST assistant message's id.
*
* Plan blocks: a plan-block marker has no message of its own — its data part
* splices into the preceding assistant group, IN FLOW right after the last
* successful plan_init call (the presentation moment).
*
* Question tools on the wire (owner fork 2): the agent's closing
* tool_execution_end carries the in-memory "Question posed" placeholder —
* projectView IGNORES clean question executions (the part stays
* input-available; the learner's answer arrives as a real toolResult). The
* placeholder never persists, so settled records never see it.
*
* Pure and deterministic: client-safe (no node deps), same input → same view.
*/
import type { AssistantMessage, ToolResultMessage } from "@earendil-works/pi-ai";
import type { LhMarkerMessage, LhMessage } from "./native-messages.ts";
import { QUESTION_TOOLS } from "./pending.ts";
import type { SubagentArtifactData, SubagentResultMeta } from "./contracts/jobs.ts";
import type { PlanBlockData, PlanVerdictData } from "./contracts/chat.ts";
/** A view part — the loose part vocabulary the renderers consume (the same
* shape names the pre-Wave-D UIMessage dialect used, so the renderer
* switch and the phase/pending scanners transplant unchanged). The common
* fields are optional-typed so the pre-Wave-D renderer code compiles
* against it with minimal casts; anything else stays `unknown` through
* the index signature. */
export interface LhViewPart extends Record<string, unknown> {
type: string;
id?: string;
text?: string;
state?: string;
toolCallId?: string;
input?: unknown;
output?: unknown;
rawInput?: unknown;
errorText?: string;
data?: unknown;
}
/** A view message: user (bubble) or assistant (one continuous run). */
export interface LhViewMessage {
id: string;
role: "user" | "assistant";
parts: LhViewPart[];
/** The subagentResult meta the steer cards read (marker details). */
metadata?: unknown;
}
/** A tool call's live execution outcome (from tool_execution_end). */
export interface LiveExecution {
isError: boolean;
/** The clean result's details payload (the card's output). */
output?: unknown;
/** The failure text (output-error parts). */
errorText?: string;
}
/** The live turn's execution layer, maintained by the session hook. */
export interface LiveTurnState {
/** toolCallId → its execution outcome seen so far. */
executions: ReadonlyMap<string, LiveExecution>;
/** True while the turn runs — calls with no execution yet render
* input-streaming (the shimmer); settled dangling calls render
* input-available (the dock mounts from the record). */
inFlight: boolean;
}
function textPart(text: string): LhViewPart {
return { type: "text", text, state: "done" };
}
/** The result a tool part should show, from a native toolResult. */
function applyResult(part: LhViewPart, result: ToolResultMessage): void {
if (result.isError) {
part.state = "output-error";
part.rawInput = part.input;
const text = result.content.find((c) => c.type === "text");
part.errorText = text && text.type === "text" ? text.text : "The tool call failed.";
} else {
part.state = "output-available";
part.output = result.details;
}
}
/** Does this assistant group carry a successful plan_init (the splice target)? */
function lastSuccessfulPlanInit(parts: LhViewPart[]): number {
let idx = -1;
parts.forEach((p, i) => {
if (p.type !== "tool-plan_init") return;
const ok = (p.output as { ok?: unknown } | undefined)?.ok;
if (ok === true) idx = i;
});
return idx;
}
/** Project the native record (+ live turn working list) into the client's
* view messages. Pure: the same input always projects to the same list. */
export function projectView(messages: readonly LhMessage[], live?: LiveTurnState): LhViewMessage[] {
const out: LhViewMessage[] = [];
const lastAssistant = (): LhViewMessage | undefined => {
const last = out[out.length - 1];
return last && last.role === "assistant" ? last : undefined;
};
const findToolPart = (toolCallId: string): LhViewPart | undefined => {
for (let i = out.length - 1; i >= 0; i--) {
for (const part of out[i]!.parts) {
if (part.type.startsWith("tool-") && part.toolCallId === toolCallId) return part;
}
}
return undefined;
};
for (const m of messages) {
const role = (m as { role?: string }).role;
if (role === "user") {
const content = m.content as string | Array<{ type: string; text?: string }>;
const texts =
typeof content === "string"
? [content]
: content.filter((c) => c.type === "text").map((c) => (c as { text: string }).text);
out.push({ id: m.id, role: "user", parts: texts.map(textPart) });
continue;
}
if (role === "custom") {
projectMarker(m as LhMarkerMessage, out);
continue;
}
if (role === "toolResult") {
applyResultTo(findToolPart((m as ToolResultMessage).toolCallId), m as ToolResultMessage);
continue;
}
if (role === "assistant") {
const assistant = m as AssistantMessage;
const parts: LhViewPart[] = [];
let textN = 0;
let reasoningN = 0;
for (const block of assistant.content) {
if (block.type === "text") {
parts.push({ type: "text", id: `${m.id}-t${textN++}`, text: block.text, state: "done" });
} else if (block.type === "thinking") {
parts.push({ type: "reasoning", id: `${m.id}-r${reasoningN++}`, text: block.thinking, state: "done" });
} else if (block.type === "toolCall") {
parts.push(toolPartOf(block, live));
}
}
if (parts.length === 0) continue; // presentation-only (nothing renderable)
const group = lastAssistant();
if (group) {
// Same group: the step separator renders between the steps.
group.parts.push({ type: "step-start" }, ...parts);
} else {
out.push({ id: m.id, role: "assistant", parts: [{ type: "step-start" }, ...parts] });
}
continue;
}
}
return out;
}
/** A tool call block → its view part, with the live execution layer. */
function toolPartOf(
block: { id: string; name: string; arguments?: unknown },
live: LiveTurnState | undefined,
): LhViewPart {
const part: LhViewPart = {
type: `tool-${block.name}`,
toolCallId: block.id,
input: block.arguments,
};
const exec = live?.executions.get(block.id);
const isQuestion = (QUESTION_TOOLS as readonly string[]).includes(block.name);
if (exec) {
if (exec.isError) {
// An invalid call NEVER becomes input-available — the dock's mount
// gate keys on the clean execution end (the live modal-realism
// crash: a malformed quiz must never mount).
part.state = "output-error";
part.rawInput = block.arguments;
part.errorText = exec.errorText ?? "The tool call failed.";
} else if (isQuestion) {
// Fork 2: the clean "Question posed" placeholder is IGNORED —
// question tools are client-owned; the dock mounts on the clean
// end, the answer arrives as a real toolResult.
part.state = "input-available";
} else {
part.state = "output-available";
part.output = exec.output;
}
} else {
// No execution yet: mid-turn it is input-streaming (the shimmer
// activity); in a settled record a dangling call is the dock's
// pending question (input-available).
part.state = live?.inFlight ? "input-streaming" : "input-available";
}
return part;
}
function applyResultTo(part: LhViewPart | undefined, result: ToolResultMessage): void {
if (part) applyResult(part, result);
// A result with no visible part (e.g. a call whose assistant message
// rendered nothing) is dropped from the view — the record keeps it.
}
/** A marker's projected message (verdict/handoff/steers) or splice (plan block). */
function projectMarker(marker: LhMarkerMessage, out: LhViewMessage[]): void {
switch (marker.customType) {
case "verdict": {
const data = marker.details as PlanVerdictData;
out.push({
id: marker.id,
role: "user",
parts: [{ type: "data-verdict", id: `vd_${marker.id}`, data }],
});
return;
}
case "probe-handoff": {
out.push({
id: marker.id,
role: "user",
parts: [{ type: "data-probe-handoff", id: `ph_${marker.id}`, data: marker.details }],
});
return;
}
case "subagent-steer": {
const meta = marker.details as SubagentResultMeta;
const artifact = marker.details as {
kind: SubagentArtifactData["kind"];
result: SubagentArtifactData["result"];
};
out.push({
id: marker.id,
role: "user",
metadata: { subagentResult: meta },
parts: [
textPart(marker.content),
{ type: "data-subagent-artifact", id: `sa_${marker.id}`, data: { kind: artifact.kind, result: artifact.result } },
],
});
return;
}
case "subagent-abort": {
const meta = marker.details as SubagentResultMeta & { aborted: true };
out.push({
id: marker.id,
role: "user",
metadata: { subagentResult: meta },
parts: [textPart(marker.content), { type: "data-subagent-aborted", id: `sa_${marker.id}`, data: meta }],
});
return;
}
case "plan-block": {
// No message of its own: splice into the preceding assistant
// group, IN FLOW after the last successful plan_init call.
const group = out[out.length - 1];
if (!group || group.role !== "assistant") return;
const idx = lastSuccessfulPlanInit(group.parts);
if (idx === -1) return;
const { block } = marker.details as { planCallId: string; block: PlanBlockData };
group.parts.splice(idx + 1, 0, { type: "data-plan-block", id: marker.id, data: block });
return;
}
}
}
-1
View File
@@ -322,7 +322,6 @@ export async function handleChatInput(
// late views reattach via GET /api/chats/[id]/stream.
const handle = runPiTurn({
agent: constructed,
base: merged,
// Stream-level failures are provider/transport internals — masked,
// like the SDK defaults. Tool-input validation messages never
// route through here: they ride the error-result path and reach the
+1 -3
View File
@@ -1,9 +1,7 @@
import { abortExtraction, extractionStatuses } from "../agent/subagents/extraction.ts";
import { coverageOf } from "../core/coverage.ts";
import { projectNative } from "../core/projection.ts";
import { deleteChatTraces, type ChatDeleteMode, type ChatDeletionStats } from "../core/graph-deletion.ts";
import { titleFromPrompt } from "../core/contracts.ts";
import type { UIMessage } from "ai";
import { graphRevision } from "./graph.ts";
import { stores } from "./stores.ts";
import { isTurnRunning } from "./turn-liveness.ts";
@@ -144,7 +142,7 @@ export function sessionCoverage(chatId: string): SessionCoverage | null {
const entries = stores.graphStore.readCoverage();
// Wave B: the record is native history — coverage reads through the
// projection (same view the extraction pipeline folds).
const cov = coverageOf(projectNative(record.messages), entries, chatId);
const cov = coverageOf(record.messages, entries, chatId);
let lastFoldedAt: string | null = null;
for (const entry of entries) {
if (entry.chatId !== chatId) continue;
+78 -302
View File
@@ -1,67 +1,66 @@
import type { Agent, AgentEvent, AgentMessage } from "@earendil-works/pi-agent-core";
import type { UIMessage } from "ai";
import { QUESTION_TOOLS } from "../../core/pending.ts";
import type { LhMessage } from "../../core/native-messages.ts";
import { projectNative } from "../../core/projection.ts";
/**
* The UI-stream bridge (docs/pi-engine-swap.md §5): runs one agent turn and
* re-emits it as an AI-SDK-shaped UIMessageStream SSE response — the wire
* protocol `useChat` (Band D, unchanged) parses:
* THE NATIVE WIRE (Wave D, owner-approved 2026-09-26): one agent turn, streamed
* to every view as VERBATIM pi AgentEvents — no translation, no remapping, no
* part assembly. The SSE body is the agent's own event stream:
*
* data: {"type":"text-delta", …}\n\n … data: [DONE]
* data: {"type":"agent_start"}\n\n
* data: {"type":"message_update","message":{…},"assistantMessageEvent":{…}}\n\n
* …
* data: {"type":"agent_end","messages":[…]}\n\n
* data: [DONE]\n\n
*
* Load-bearing protocol rules (verified against ai@7's client):
* - The `start` chunk's messageId MUST be the transcript's last message id
* when it is an assistant message (the continuation id rule) — a fresh id
* there duplicates every prior part client-side.
* - Question tools (quiz / ask_user_question): emit tool-input-available and
* NEVER tool-output-available (the client throws on an output for a call it
* owns). The synthetic "question posed" result lives in memory only; the
* record's part stays output-less until the learner answers.
* - tool-input-available is emitted ONLY after the call cleared validation
* at the boundary (clean tool_execution_end) — the SDK's
* InvalidToolInputError gating. An invalid call surfaces as
* tool-input-error instead and NEVER becomes input-available, so a
* malformed question can never mount the dock (the live modal-realism
* crash: GLM double-encoded `options` as a JSON string).
* - The WIRE assembly (below) is the client view only (Wave B): the
* record write is the agent's own native final state — the projection
* (core/projection.ts) reproduces the client's part shapes for every
* consumer (phases, plan blocks, pending, extraction) until Wave D.
* Load-bearing protocol facts (verified against pi-agent-core):
* - Every `message_update` carries the FULL partial assistant message (each
* AssistantMessageEvent ships `partial`) — a view REPLACES its streaming
* assistant with `event.message`; no delta assembly exists anywhere.
* - Question tools (quiz / ask_user_question): NO suppression (owner ruling,
* Wave D fork 2) — the closing `tool_execution_end` with the in-memory
* "Question posed" placeholder flows to the client, which ignores it
* (question tools are client-owned; the answer arrives as the input event).
* The placeholder still never persists and never reaches the model.
* - Tool-call validation gating survives by construction: `tool_execution_end`
* only fires after the call cleared the boundary (an invalid call arrives
* as isError — never a clean end), so the dock's mount rule keys on it.
* - App-level events beyond pi's taxonomy: the reattach terminal
* {"type":"turn-settled"} (served by turn-runs when no turn is running) and
* the loop-level {"type":"error","errorText"} (agent.continue() threw
* outside the event stream — mapped through the handler's masking policy).
* - Stream-level provider errors ride their message verbatim
* (message_end with stopReason "error" + errorMessage).
*
* BACKGROUND-TURN RUNNER (owner ruling, 2026-09-25): a turn executes as a
* server-side background object decoupled from any single HTTP response —
* clients are VIEWS. runPiTurn returns the handle (event log + subscriber
* set); piTurnResponse wraps a subscriber stream as the SSE response, and
* the POST /api/chat response and GET /api/chats/[id]/stream are BOTH
* exactly that. Cancelling a view detaches it ONLY — never aborts the
* turn; the handle's abort() is the one remote kill.
* exactly that. Cancelling a view detaches it ONLY — never aborts the turn;
* the handle's abort() is the one remote kill.
*/
/** SSE chunk — the flat JSON line shape the client's zod schema accepts. */
type Chunk = Record<string, unknown> & { type: string };
/** The wire event: a verbatim pi AgentEvent, or one of the two app events. */
export type TurnWireEvent =
| AgentEvent
| { type: "error"; errorText: string };
export interface PiTurnSpec {
/** The per-turn agent (stateless engine — built from the record). */
agent: Agent;
/** The turn's base: the native record + this turn's appends
* (write-ahead persisted already). Wave B: the agent's context and
* the record share one shape; onFinish receives the full agent state
* and the caller persists the slice past base.length. */
base: LhMessage[];
/** The handler's error-unmasking policy (validation messages stay readable). */
/** The handler's error-masking policy for LOOP-level failures (the
* event stream itself is verbatim — provider errors ride their
* message's stopReason/errorMessage through untouched). */
onError?: (error: unknown) => string;
/** The canonical end-of-turn seam: persist + usage + extraction + endTurn.
* Receives the agent's FULL final state (native messages) — the turn's
* new messages are the slice past base.length. */
* Receives the agent's FULL final state (native messages) — the caller
* persists the slice past its own base count. */
onFinish: (messages: AgentMessage[], isAborted: boolean) => void;
}
/** One view onto a running turn: push() replays/streams a chunk, done()
/** One view onto a running turn: push() replays/streams an event, done()
* terminates the view ([DONE] + close). Cheap to be a Set member. */
export interface PiTurnSubscriber {
push(chunk: Chunk): void;
push(event: TurnWireEvent): void;
done(): void;
}
@@ -75,109 +74,35 @@ export interface PiTurnHandle {
* view only — the turn runs on. */
subscribe(): ReadableStream<Uint8Array>;
/** The remote kill — the ONLY way a running turn dies early: sawAbort +
* agent.abort(); the loop's agent_end does the rest (finish chunk to
* every subscriber, then onFinish persists the partial transcript). */
* agent.abort(); the loop's agent_end does the rest (the settle seam
* persists the partial transcript to every listener). */
abort(): void;
}
/** Map a pi stop reason to the finish chunk's enum. */
function finishReasonOf(stopReason: string | undefined): string {
switch (stopReason) {
case "stop":
return "stop";
case "length":
return "length";
case "toolUse":
return "tool-calls";
case "error":
return "error";
default:
return "other";
}
}
function resultErrorText(result: unknown): string {
const content = (result as { content?: Array<{ type: string; text?: string }> } | undefined)?.content;
const text = content?.find((c) => c.type === "text")?.text;
return text ?? "The tool call failed.";
}
/** Loose part shapes the bridge assembles (the client's part shapes). */
interface TextPart {
type: "text";
text: string;
state: string;
}
interface ReasoningPart {
type: "reasoning";
id: string;
text: string;
state: string;
}
interface ToolPart {
type: string;
toolCallId: string;
state: string;
input?: unknown;
rawInput?: unknown;
output?: unknown;
errorText?: string;
}
export function runPiTurn(spec: PiTurnSpec): PiTurnHandle {
const { agent, base, onFinish } = spec;
// The streaming target: continuation (same id, parts appended to a copy
// of the existing message) or a fresh assistant message. The continuation
// rule is the CLIENT's: it keys on the last UIMessage's id — read through
// the projection (the record is native; a turn after an answer continues
// the group that carries the answered quiz, which is the projection's
// last message).
const projected = projectNative(base);
const last = projected[projected.length - 1] as
| { id?: string; role?: string; parts?: UIMessage["parts"] }
| undefined;
const continuing = last?.role === "assistant" && typeof last.id === "string";
const messageId =
(continuing && last.id) || `msg_${Date.now().toString(36)}_${crypto.randomUUID().slice(0, 8)}`;
const baseParts: UIMessage["parts"] = continuing && last.parts ? [...last.parts] : [];
const parts: UIMessage["parts"] = [...baseParts];
// Stable, collision-free part ids across continuation runs (reasoning
// parts persist their id; text ids only need per-run uniqueness).
let textN = baseParts.filter((p) => p.type === "text").length;
let reasoningN = baseParts.filter((p) => p.type === "reasoning").length;
const textIds = new Map<number, string>(); // contentIndex → chunk id
const reasoningIds = new Map<number, string>();
const toolInputs = new Map<string, { toolName: string; input: unknown }>();
/** The streaming text/reasoning parts of THIS run, keyed by chunk id. */
const activeParts = new Map<string, TextPart | ReasoningPart>();
/** Tool parts created this run, keyed by toolCallId. */
const newToolParts = new Map<string, ToolPart>();
const { agent, onError, onFinish } = spec;
let settled = false;
let sawAbort = false;
let finishReason: string | undefined;
let errorMessage: string | undefined;
// THE PER-TURN EVENT LOG (memory-only — the settle-only disk writes
// are unchanged): every emitted chunk, in order, so any late subscriber
// gets a FULL replay. Unbounded BY HONEST DESIGN (owner ruling
// 2026-09-25): a turn's chunk log is finite by construction — the agent
// loop ends (or the 30-minute liveness TTL bounds the pathological
// never-settling case), and concurrency is human-scale (one turn per
// chat, a few chats). No cap, no truncation: replay must never
// silently drop history. The log dies with the turn at settle.
const log: Chunk[] = [];
// are unchanged): every forwarded event, in order, so any late
// subscriber gets a FULL replay. Unbounded BY HONEST DESIGN (owner
// ruling 2026-09-25): a turn's event log is finite by construction —
// the agent loop ends (or the 30-minute liveness TTL bounds the
// pathological never-settling case), and concurrency is human-scale
// (one turn per chat, a few chats). No cap, no truncation: replay must
// never silently drop history. The log dies with the turn at settle.
const log: TurnWireEvent[] = [];
const subscribers = new Set<PiTurnSubscriber>();
// Send = append to the log + fan out to every live view. A view's own
// enqueue failure (a cancelled reader) is swallowed by the view — the
// turn must still settle; onFinish persists regardless of who watches.
const send = (chunk: Chunk) => {
const send = (event: TurnWireEvent) => {
if (settled) return;
log.push(chunk);
for (const sub of subscribers) sub.push(chunk);
log.push(event);
for (const sub of subscribers) sub.push(event);
};
const closeAll = () => {
if (settled) return;
@@ -185,175 +110,23 @@ export function runPiTurn(spec: PiTurnSpec): PiTurnHandle {
for (const sub of subscribers) sub.done();
subscribers.clear();
};
send({ type: "start", messageId });
// The settle seam: the record write is the agent's OWN final state
// (native history — signatures, usage, provider stamps all real).
// Copy the array: the agent may outlive this call's read.
const settleWith = () => {
// Wave B: the record write is the agent's OWN final state (native
// history — signatures, usage, provider stamps all real); the wire
// assembly above served the client view only. Copy the array: the
// agent may outlive this call's read.
onFinish([...agent.state.messages], sawAbort);
};
const handle = (event: AgentEvent) => {
switch (event.type) {
case "turn_start":
send({ type: "start-step" });
parts.push({ type: "step-start" });
break;
case "message_update": {
const streamEvent = event.assistantMessageEvent;
switch (streamEvent.type) {
case "text_start": {
const id = `${messageId}-t${textN++}`;
textIds.set(streamEvent.contentIndex, id);
const part: TextPart = { type: "text", text: "", state: "streaming" };
activeParts.set(id, part);
parts.push(part as UIMessage["parts"][number]);
send({ type: "text-start", id });
break;
}
case "text_delta": {
const id = textIds.get(streamEvent.contentIndex);
const part = id !== undefined ? (activeParts.get(id) as TextPart | undefined) : undefined;
if (part) part.text += streamEvent.delta;
if (id !== undefined) send({ type: "text-delta", id, delta: streamEvent.delta });
break;
}
case "text_end": {
const id = textIds.get(streamEvent.contentIndex);
const part = id !== undefined ? (activeParts.get(id) as TextPart | undefined) : undefined;
if (part) part.state = "done";
activeParts.delete(id ?? "");
if (id !== undefined) send({ type: "text-end", id });
break;
}
case "thinking_start": {
const id = `${messageId}-r${reasoningN++}`;
reasoningIds.set(streamEvent.contentIndex, id);
const part: ReasoningPart = { type: "reasoning", id, text: "", state: "streaming" };
activeParts.set(id, part);
parts.push(part as UIMessage["parts"][number]);
send({ type: "reasoning-start", id });
break;
}
case "thinking_delta": {
const id = reasoningIds.get(streamEvent.contentIndex);
const part = id !== undefined ? (activeParts.get(id) as ReasoningPart | undefined) : undefined;
if (part) part.text += streamEvent.delta;
if (id !== undefined) send({ type: "reasoning-delta", id, delta: streamEvent.delta });
break;
}
case "thinking_end": {
const id = reasoningIds.get(streamEvent.contentIndex);
const part = id !== undefined ? (activeParts.get(id) as ReasoningPart | undefined) : undefined;
if (part) part.state = "done";
activeParts.delete(id ?? "");
if (id !== undefined) send({ type: "reasoning-end", id });
break;
}
case "toolcall_start": {
const block = (event.message as { content: any[] }).content[streamEvent.contentIndex];
if (block?.type !== "toolCall") break;
toolInputs.set(block.id, { toolName: block.name, input: undefined });
const part: ToolPart = {
type: `tool-${block.name}`,
toolCallId: block.id,
state: "input-streaming",
input: undefined,
};
newToolParts.set(block.id, part);
parts.push(part as unknown as UIMessage["parts"][number]);
send({ type: "tool-input-start", toolCallId: block.id, toolName: block.name });
break;
}
case "toolcall_delta": {
const block = (event.message as { content: any[] }).content[streamEvent.contentIndex];
if (block?.type !== "toolCall") break;
send({
type: "tool-input-delta",
toolCallId: block.id,
inputTextDelta: streamEvent.delta,
});
break;
}
case "toolcall_end": {
const toolCall = streamEvent.toolCall;
toolInputs.set(toolCall.id, { toolName: toolCall.name, input: toolCall.arguments });
// Input stays input-streaming here: it only becomes available
// once the call clears validation at the execution boundary
// (tool_execution_end) — an invalid input must never reach
// input-available, where the dock would mount on it.
const part = newToolParts.get(toolCall.id);
if (part) part.input = toolCall.arguments;
break;
}
}
break;
}
case "tool_execution_end": {
const known = toolInputs.get(event.toolCallId);
const toolName = event.toolName || known?.toolName || "";
const input = known?.input;
const part = newToolParts.get(event.toolCallId);
if (event.isError) {
const errorText = resultErrorText(event.result);
if (part) {
part.state = "output-error";
part.rawInput = input;
part.errorText = errorText;
}
send({
type: "tool-input-error",
toolCallId: event.toolCallId,
toolName,
input,
errorText,
});
} else {
// Validation cleared: only NOW does the input become available
// to the client (the InvalidToolInputError gating the SDK owned).
if (part) {
part.state = "input-available";
part.input = input;
}
send({ type: "tool-input-available", toolCallId: event.toolCallId, toolName, input });
if ((QUESTION_TOOLS as readonly string[]).includes(toolName)) {
// Client-owned: the answer arrives via addToolOutput;
// the synthetic placeholder result never reaches wire
// or record.
break;
}
const output = (event.result as { details?: unknown } | undefined)?.details;
if (part) {
part.state = "output-available";
part.output = output;
}
send({ type: "tool-output-available", toolCallId: event.toolCallId, output });
}
break;
}
case "turn_end":
send({ type: "finish-step" });
break;
case "message_end":
if (event.message && (event.message as { role?: string }).role === "assistant") {
const m = event.message as { stopReason?: string; errorMessage?: string };
finishReason = m.stopReason;
if (m.stopReason === "aborted") sawAbort = true;
if (m.stopReason === "error") errorMessage = m.errorMessage;
}
break;
case "agent_end": {
if (errorMessage !== undefined) {
send({ type: "error", errorText: spec.onError?.(errorMessage) ?? errorMessage });
}
send({ type: "finish", finishReason: finishReasonOf(finishReason) });
closeAll();
settleWith();
break;
}
// VERBATIM: every pi AgentEvent forwards as-is — no remapping.
send(event);
if (event.type === "message_end") {
const m = event.message as { role?: string; stopReason?: string } | undefined;
if (m?.role === "assistant" && m.stopReason === "aborted") sawAbort = true;
}
if (event.type === "agent_end") {
closeAll();
settleWith();
}
};
agent.subscribe(handle);
@@ -363,9 +136,12 @@ export function runPiTurn(spec: PiTurnSpec): PiTurnHandle {
await agent.continue();
} catch (err) {
// A loop-level failure (not a provider error — those ride the
// stream as stopReason "error"): surface, persist, close.
send({ type: "error", errorText: spec.onError?.(err) ?? "An error occurred." });
send({ type: "finish", finishReason: "error" });
// stream as stopReason "error"): surface, close, persist. If the
// agent already settled (agent_end reached before the throw),
// the seam has run — never settle twice.
if (settled) return;
console.error("[turn] agent loop failed:", err);
send({ type: "error", errorText: onError?.(err) ?? "An error occurred." });
closeAll();
settleWith();
}
@@ -379,10 +155,10 @@ export function runPiTurn(spec: PiTurnSpec): PiTurnHandle {
const encoder = new TextEncoder();
let closed = false;
sub = {
push(chunk) {
push(event) {
if (closed) return;
try {
controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`));
controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`));
} catch {
/* reader cancelled — stop emitting, keep running */
closed = true;
@@ -399,11 +175,11 @@ export function runPiTurn(spec: PiTurnSpec): PiTurnHandle {
}
},
};
// Attach THEN replay — synchronous, so no chunk can fall
// Attach THEN replay — synchronous, so no event can fall
// between replay and live (single-threaded: the agent's
// handle() only runs after this stack returns).
subscribers.add(sub);
for (const chunk of log) sub.push(chunk);
for (const event of log) sub.push(event);
// A subscriber that attached after settle (the unregister
// race) still gets the full history + [DONE].
if (settled) sub.done();
@@ -426,8 +202,8 @@ export function runPiTurn(spec: PiTurnSpec): PiTurnHandle {
}
/** The subscriber-shaped SSE response — the POST /api/chat response and
* GET /api/chats/[id]/stream are BOTH this: same headers, same wire
* dialect, one protocol for the client to consume. */
* GET /api/chats/[id]/stream are BOTH this: same headers, same wire
* dialect, one protocol for the client to consume. */
export function piTurnResponse(handle: PiTurnHandle): Response {
return new Response(handle.subscribe(), {
headers: {
@@ -436,4 +212,4 @@ export function piTurnResponse(handle: PiTurnHandle): Response {
"X-Accel-Buffering": "no",
},
});
}
}
+5 -8
View File
@@ -21,7 +21,6 @@ import type { LhMessage } from "../core/native-messages.ts";
import { newMessageId } from "../core/native-messages.ts";
import { QUESTION_TOOLS } from "../core/pending.ts";
import { planBlocksFor } from "../core/plan-blocks.ts";
import { projectNative } from "../core/projection.ts";
import { coverageOf } from "../core/coverage.ts";
import type { ChatStore, GraphStore } from "../core/ports.ts";
import type { ExtractionPipeline } from "../agent/subagents/extraction.ts";
@@ -72,12 +71,10 @@ export function persistTurn(
stamped.push(message.id ? message : ({ ...message, id: newMessageId() } as LhMessage));
}
const tree = record.tree;
const blocks = planBlocksFor(
record.messages,
stamped,
tree,
projectNative([...record.messages, ...stamped]),
);
const blocks = planBlocksFor(record.messages, stamped, tree, [
...record.messages,
...stamped,
]);
chatStore.saveMessages(chatId, [...record.messages, ...stamped, ...blocks]);
return stamped;
} catch (err) {
@@ -110,7 +107,7 @@ export function triggerExtraction(
try {
const record = deps.chatStore.getChat(chatId);
if (!record) return;
const covered = coverageOf(projectNative(record.messages), deps.graphStore.readCoverage(), chatId);
const covered = coverageOf(record.messages, deps.graphStore.readCoverage(), chatId);
if (covered.uncoveredTokens < chunkTokens) return;
void extraction.runExtraction(chatId).catch((err) => {
// BYOK (owner, 2026-09-22) + key presence (provider-key interlude,
+10 -11
View File
@@ -13,9 +13,8 @@ import { PATHS } from "../../config.ts";
import type { ChatStore } from "../core/ports.ts";
import type { ChatRecord, ChatSettings, ChatSummary, ExtractionRunUsage, PlanTree, TurnUsage } from "../core/contracts.ts";
import type { LhMessage } from "../core/native-messages.ts";
import { findPendingToolCall } from "../core/pending.ts";
import { phaseAt3 } from "../core/phases.ts";
import { projectNative } from "../core/projection.ts";
import { danglingQuestionOf } from "../core/native-messages.ts";
import { phaseAt3Lh } from "../core/phases.ts";
function chatFile(id: string): string {
if (!/^[A-Za-z0-9_-]+$/.test(id)) throw new Error(`Invalid chat id: ${id}`);
@@ -58,7 +57,6 @@ function listChats(): ChatSummary[] {
// zero-back-compat wipe makes these "corrupt") must not blank the
// whole sidebar — same isolation getChat's per-file catch gives.
try {
const projected = projectNative(messages);
return {
// turns and extractions stay out of the list payload (list =
// summaries; usage aggregates ride the full record, the picker
@@ -66,13 +64,14 @@ function listChats(): ChatSummary[] {
...summary,
messageCount: messages.length,
// Record-derived status for the session picker: phase from the
// transcript's explicit markers (probe_done / plan_init — the
// event-driven phases, docs/probe-teach-split.md §4), pending
// question from the last unanswered quiz/ask part — both read
// through the one-way projection. Liveness (busy) is NOT
// record-derived — the API route merges it (ChatListItem).
phase: phaseAt3(projected),
pendingQuestion: findPendingToolCall(projected) !== null,
// transcript's explicit events (probe_done concluded / verdict
// accepted — docs/probe-teach-split.md §4), pending question
// from the native dangling-call scan. Wave D: both folds read
// the NATIVE record (core/phases.ts phaseAt3Lh +
// native-messages.ts danglingQuestionOf). Liveness (busy) is
// NOT record-derived — the API route merges it (ChatListItem).
phase: phaseAt3Lh(messages),
pendingQuestion: danglingQuestionOf(messages) !== null,
};
} catch (err) {
console.error(`[chat ${summary.id}] summary projection failed:`, err);