W4: graph store + emission + home graph view

This commit is contained in:
Amos
2026-08-22 10:30:00 +03:00
parent 9a99b01159
commit 8728573253
8 changed files with 1297 additions and 11 deletions
+410
View File
@@ -0,0 +1,410 @@
/**
* Knowledge-graph emission + derivation (W4; vision §1 "One substrate",
* contracts.ts "v1 emission").
*
* emitChat(chatId) folds ONE chat into the global graph — naive union,
* deterministic, no LLM:
* plan-tree nodes → GraphNodes (statement = title, aliases empty)
* plan-tree edges → `requires` GraphEdges (from = prerequisite)
* answered quizzes → ledger events (probe-* before the first plan_init
* output, check / dont-know after; ref = toolCallId)
* plan_mark_done → "taught" events (ref = taught:<treeNodeId>)
*
* Idempotent + incremental: mappings dedupe nodes, (from,to,type) dedupes
* edge assertions, chat-scoped refs dedupe events, and sessions.jsonl
* checkpoints which chats have been folded. Re-emitting a chat appends
* nothing. The store (src/store/graph.ts) holds the append-only ledgers;
* everything here derives from them.
*/
import { isToolUIPart, type UIMessage } from "ai";
import type { GraphEdge, GraphNode, LedgerEvent, LedgerEventKind, PlanTree, PlanToolResult } from "./contracts.ts";
import type { QuestionAnswerData } from "./question-core.ts";
import { getChat, listChats } from "../store/chats.ts";
import { graphLedgers, newGraphNodeId } from "../store/graph.ts";
// ─── transcript scanning ─────────────────────────────────────────────────────
/** A flattened, order-preserving view of every tool part in a transcript. */
interface ToolPartView {
name: string;
toolCallId: string;
state: string;
input: unknown;
output: unknown;
}
function toolParts(messages: UIMessage[]): ToolPartView[] {
const out: ToolPartView[] = [];
for (const message of messages) {
for (const part of message.parts) {
if (!isToolUIPart(part)) continue;
out.push({
name: part.type.slice("tool-".length),
toolCallId: part.toolCallId,
state: part.state,
input: part.input,
output: part.output,
});
}
}
return out;
}
/** A plan tool part's output, if it is a well-formed PlanToolResult. */
function planResult(part: ToolPartView): PlanToolResult | null {
if (part.state !== "output-available") return null;
const out = part.output;
if (typeof out !== "object" || out === null) return null;
if (!("tree" in out) || typeof (out as PlanToolResult).ok !== "boolean") return null;
return out as PlanToolResult;
}
/**
* The FINAL live tree, replayed from plan tool outputs in message order.
* Fallback for records whose stored `tree` is missing (it is the same state
* saveTree persists, so the two normally agree); also proves the probe/check
* boundary scan below sees the same transcript the teacher wrote.
*/
function replayFinalTree(messages: UIMessage[]): PlanTree {
let tree: PlanTree = { nodes: [], edges: [], currentNodeId: null, doneNodeIds: [], version: 0 };
for (const part of toolParts(messages)) {
if (!part.name.startsWith("plan_")) continue;
const result = planResult(part);
if (result && result.ok && result.tree) tree = result.tree;
}
return tree;
}
// ─── emission ────────────────────────────────────────────────────────────────
export interface EmissionStats {
/** New graph nodes appended (and mappings with them). */
nodes: number;
/** New edge assertions appended. */
edges: number;
/** New ledger events appended. */
events: number;
}
/**
* Fold one chat into the graph ledgers. Idempotent — safe to call after every
* turn (chat-handler onFinish) and again on catch-up (GET /api/graph). Sync
* and fast: a handful of small-file reads plus one append per new record.
*/
export function emitChat(chatId: string): EmissionStats {
const record = getChat(chatId);
if (!record) return { nodes: 0, edges: 0, events: 0 };
const stats: EmissionStats = { nodes: 0, edges: 0, events: 0 };
// 1. This chat's existing mappings — the identity map is per-chat (v1 is
// naive union: two sessions' tree nodes never merge into one graph node).
const mapping = new Map<string, string>();
for (const m of graphLedgers.readMappings()) {
if (m.chatId === chatId) mapping.set(m.treeNodeId, m.graphNodeId);
}
// 2. The tree state to fold. record.tree is the truth (it is the exact
// state every plan tool persisted); the transcript replay covers records
// that predate tree persistence. Nodes REMOVED from the tree are simply
// absent here — if they were emitted by an earlier fold, their mappings
// and ledger lines stay (the ledgers are append-only evidence).
const tree =
record.tree && record.tree.nodes.length > 0 ? record.tree : replayFinalTree(record.messages);
// 3. Tree nodes without a mapping → GraphNode + mapping.
const now = new Date().toISOString();
for (const node of tree.nodes) {
if (mapping.has(node.id)) continue;
const graphNode: GraphNode = {
id: newGraphNodeId(),
// statement is the title; the summary stays out (contracts).
statement: node.title,
aliases: [],
kind: "truth",
createdAt: now,
};
graphLedgers.appendNode(graphNode);
graphLedgers.appendMapping({ chatId, treeNodeId: node.id, graphNodeId: graphNode.id });
mapping.set(node.id, graphNode.id);
stats.nodes++;
}
// 4. Tree edges → `requires` edges. Dedupe by (from,to,type) across the
// whole ledger: an existing assertion just gains this chatId as
// provenance (appended as a new line; reads merge).
const asserted = new Map<string, Set<string>>();
for (const line of graphLedgers.readEdges()) {
const key = `${line.from}\u0000${line.to}\u0000${line.type}`;
let provenance = asserted.get(key);
if (!provenance) {
provenance = new Set();
asserted.set(key, provenance);
}
for (const c of line.provenance) provenance.add(c);
}
for (const edge of tree.edges) {
const from = mapping.get(edge.from);
const to = mapping.get(edge.to);
if (!from || !to || from === to) continue; // unmapped endpoint — skip
const key = `${from}\u0000${to}\u0000requires`;
const provenance = asserted.get(key);
if (provenance?.has(chatId)) continue; // already asserted by this chat
graphLedgers.appendEdge({ from, to, type: "requires", provenance: [chatId] });
if (provenance) provenance.add(chatId);
else asserted.set(key, new Set([chatId]));
stats.edges++;
}
// 5. Events, replayed in transcript order so the probe/check boundary is
// exact: quizzes BEFORE the first plan_init output are probes. Refs are
// chat-scoped (a quiz toolCallId or taught:<treeNodeId>) — a ref already
// in the ledger means this turn was folded before.
const seenRefs = new Set<string>();
for (const event of graphLedgers.readEvents()) {
if (event.chatId === chatId && event.ref) seenRefs.add(event.ref);
}
let planSeen = false;
for (const part of toolParts(record.messages)) {
// The boundary flips on the first plan_init OUTPUT, whatever it says
// (an ok:false "tree already initialized" still means a tree exists).
if (part.name === "plan_init" && part.state === "output-available") planSeen = true;
if (part.name === "quiz" && part.state === "output-available") {
const event = quizEvent(part, chatId, mapping, planSeen, now);
if (event?.ref !== undefined && !seenRefs.has(event.ref)) {
graphLedgers.appendEvent(event);
seenRefs.add(event.ref);
stats.events++;
}
}
if (part.name === "plan_mark_done" && part.state === "output-available") {
const result = planResult(part);
const input = part.input as { id?: unknown } | undefined;
if (result?.ok && typeof input?.id === "string") {
const ref = `taught:${input.id}`;
if (!seenRefs.has(ref)) {
graphLedgers.appendEvent({
nodeId: mapping.get(input.id) ?? null,
kind: "taught",
chatId,
at: now,
ref,
});
seenRefs.add(ref);
stats.events++;
}
}
}
}
// 6. Checkpoint: this chat is folded through its current updatedAt.
graphLedgers.appendCheckpoint({ chatId, foldedAt: record.updatedAt });
return stats;
}
/** Build the ledger event for one answered quiz tool part (or null to skip). */
function quizEvent(
part: ToolPartView,
chatId: string,
mapping: Map<string, string>,
planSeen: boolean,
now: string,
): LedgerEvent | null {
const out = part.output;
if (typeof out !== "object" || out === null) return null;
const answer = out as Partial<QuestionAnswerData>;
if (answer.kind !== "quiz") return null; // ask questions are not graded
if (answer.dismissed === true) return null; // no answer — no evidence
let kind: LedgerEventKind;
if (answer.dontKnow === true) kind = "dont-know";
else if (answer.correct === true) kind = planSeen ? "check-correct" : "probe-correct";
else kind = planSeen ? "check-incorrect" : "probe-miss";
const input = part.input as { nodeId?: unknown } | undefined;
const treeNodeId = typeof input?.nodeId === "string" ? input.nodeId : null;
const nodeId = treeNodeId !== null ? (mapping.get(treeNodeId) ?? null) : null;
const event: LedgerEvent = {
nodeId,
kind,
chatId,
at: typeof answer.answeredAt === "string" ? answer.answeredAt : now,
ref: part.toolCallId,
};
if (typeof answer.note === "string" && answer.note.length > 0) event.note = answer.note;
// The question text is the attribution input for untagged/unmapped probes.
if (nodeId === null && typeof answer.questionText === "string") event.question = answer.questionText;
return event;
}
// ─── derivation (reads only — everything below is recomputable) ─────────────
export interface DerivedGraph {
nodes: GraphNode[];
edges: GraphEdge[];
events: LedgerEvent[];
}
/**
* The whole graph from the ledgers. Edge LINES are assertions; the returned
* edges are merged by (from,to,type) with unioned provenance, first-seen
* order. `store events, derive state` — this is the only reader that matters.
*/
export function getGraph(): DerivedGraph {
const nodes = graphLedgers.readNodes();
const merged: GraphEdge[] = [];
const byKey = new Map<string, GraphEdge>();
for (const line of graphLedgers.readEdges()) {
const key = `${line.from}\u0000${line.to}\u0000${line.type}`;
const existing = byKey.get(key);
if (existing) {
for (const c of line.provenance) {
if (!existing.provenance.includes(c)) existing.provenance.push(c);
}
} else {
const edge: GraphEdge = { ...line, provenance: [...line.provenance] };
byKey.set(key, edge);
merged.push(edge);
}
}
return { nodes, edges: merged, events: graphLedgers.readEvents() };
}
/** Per-node state classes for the home view (vision §1 "Home page"). */
export type NodeStateClass = "taught" | "known" | "missed" | "dont-know" | "candidate";
const QUIZ_EVENT_KINDS: ReadonlySet<string> = new Set([
"probe-correct",
"check-correct",
"probe-miss",
"check-incorrect",
"dont-know",
]);
/**
* Derive node states from events (never stored): taught if ANY taught event;
* else the LATEST quiz event decides (known / missed / dont-know); a node
* with no events is a candidate.
*/
export function deriveNodeStates(nodes: GraphNode[], events: LedgerEvent[]): Record<string, NodeStateClass> {
const states: Record<string, NodeStateClass> = {};
for (const node of nodes) states[node.id] = "candidate";
const taught = new Set<string>();
const latest = new Map<string, { at: string; kind: LedgerEventKind; index: number }>();
events.forEach((event, index) => {
if (event.nodeId === null) return;
if (event.kind === "taught") {
taught.add(event.nodeId);
return;
}
if (!QUIZ_EVENT_KINDS.has(event.kind)) return;
const current = latest.get(event.nodeId);
if (!current || event.at > current.at || (event.at === current.at && index > current.index)) {
latest.set(event.nodeId, { at: event.at, kind: event.kind, index });
}
});
for (const node of nodes) {
if (taught.has(node.id)) {
states[node.id] = "taught";
continue;
}
const last = latest.get(node.id);
if (!last) continue; // candidate
if (last.kind === "probe-correct" || last.kind === "check-correct") states[node.id] = "known";
else if (last.kind === "dont-know") states[node.id] = "dont-know";
else states[node.id] = "missed";
}
return states;
}
// ─── the /api/graph payload ──────────────────────────────────────────────────
export interface GraphSnapshot {
nodes: GraphNode[];
edges: GraphEdge[];
nodeStates: Record<string, NodeStateClass>;
/** chatId → title, for every chat that touched the graph. */
sessions: Record<string, string>;
/** nodeId → chatIds that touched the node (click-through provenance). */
nodeSessions: Record<string, string[]>;
}
/** Everything the home graph view needs, in one fetch. */
export function graphSnapshot(): GraphSnapshot {
const { nodes, edges, events } = getGraph();
const nodeStates = deriveNodeStates(nodes, events);
const nodeChats = new Map<string, Set<string>>();
const touch = (nodeId: string, chatId: string): void => {
let chats = nodeChats.get(nodeId);
if (!chats) {
chats = new Set();
nodeChats.set(nodeId, chats);
}
chats.add(chatId);
};
for (const m of graphLedgers.readMappings()) touch(m.graphNodeId, m.chatId);
for (const edge of edges) {
for (const chatId of edge.provenance) {
touch(edge.from, chatId);
touch(edge.to, chatId);
}
}
for (const event of events) {
if (event.nodeId !== null) touch(event.nodeId, event.chatId);
}
// Titles for every chat the graph can point back at.
const chatIds = new Set<string>();
for (const chats of nodeChats.values()) {
for (const id of chats) chatIds.add(id);
}
const sessions: Record<string, string> = {};
for (const id of chatIds) {
const chat = getChat(id);
if (chat) sessions[id] = chat.title;
}
const nodeSessions: Record<string, string[]> = {};
for (const [nodeId, chats] of nodeChats) {
const ids = [...chats].filter((id) => sessions[id] !== undefined);
if (ids.length > 0) nodeSessions[nodeId] = ids;
}
return { nodes, edges, nodeStates, sessions, nodeSessions };
}
// ─── catch-up ────────────────────────────────────────────────────────────────
/**
* Fold every chat whose updatedAt is newer than its checkpoint (partial
* sessions deposit value — vision §1 "frequent updates"). Called at the top
* of GET /api/graph; the chat-handler onFinish hook covers the live path.
*/
export function foldStaleChats(): number {
const checkpoints = new Map<string, string>();
for (const checkpoint of graphLedgers.readCheckpoints()) {
const previous = checkpoints.get(checkpoint.chatId);
if (previous === undefined || checkpoint.foldedAt > previous) {
checkpoints.set(checkpoint.chatId, checkpoint.foldedAt);
}
}
let folded = 0;
for (const chat of listChats()) {
const foldedAt = checkpoints.get(chat.id);
if (foldedAt !== undefined && chat.updatedAt <= foldedAt) continue;
try {
emitChat(chat.id);
folded++;
} catch (err) {
console.error(`[graph] fold failed for chat ${chat.id}:`, err);
}
}
return folded;
}
+9
View File
@@ -2,6 +2,7 @@ import { createAgentUIStreamResponse, type UIMessage } from "ai";
import { createTeacherAgent, type TeacherAgent } from "../agent/teacher-agent.ts";
import { createTeacherTools } from "../agent/tools.ts";
import { spawnPendingJobs } from "../agent/subagents/runner.ts";
import { emitChat } from "../core/graph.ts";
import { getChat, saveMessages } from "../store/chats.ts";
/**
@@ -47,6 +48,14 @@ export async function handleChatRequest(req: Request, agent?: TeacherAgent): Pro
} catch (err) {
console.error(`[chat ${chatId}] job spawn failed:`, err);
}
// W4: fold this chat into the knowledge-graph ledgers (tree nodes /
// edges, quiz outcomes, taught events) — synchronous but tiny
// (append-only JSONL), and errors never touch the response.
try {
emitChat(chatId);
} catch (err) {
console.error(`[chat ${chatId}] graph emission failed:`, err);
}
},
});
}
+166
View File
@@ -0,0 +1,166 @@
/**
* Graph ledger store (W4; vision §1, contracts.ts "v1 emission").
*
* Five append-only JSONL ledgers at config.LEDGER_FILES:
* nodes.jsonl GraphNode — one line per graph node
* edges.jsonl GraphEdge — one line per edge ASSERTION (a second
* session asserting the same (from,to,type)
* appends another line; reads merge)
* events.jsonl LedgerEvent — quiz outcomes + taught events
* mappings.jsonl NodeMapping — the naive-union identity map:
* (chatId, treeNodeId) → graphNodeId
* sessions.jsonl SessionCheckpoint— emission checkpoints (foldedAt)
*
* data/graph/* is exclusively this store's (the legacy V0 scaffold lives in
* data/legacy-graph). Reads are strict against the contracts.ts shapes:
* corrupt lines are skipped, never guessed at — the ledgers are the source of
* truth and everything else is derived (src/core/graph.ts).
*
* Same idioms as src/store/chats.ts: sync fs, tiny records, no framework.
*/
import * as fs from "node:fs";
import * as path from "node:path";
import { randomBytes } from "node:crypto";
import { LEDGER_FILES } from "../../config.ts";
import type { GraphEdge, GraphNode, LedgerEvent, LedgerEventKind } from "../core/contracts.ts";
/** The naive-union identity map line: a chat's tree node → graph node. */
export interface NodeMapping {
chatId: string;
treeNodeId: string;
graphNodeId: string;
}
/** sessions.jsonl line — the emission checkpoint for one chat. */
export interface SessionCheckpoint {
chatId: string;
/** The chat record's `updatedAt` at the time of the fold. */
foldedAt: string;
}
/** "n_" + nanoid-like (contracts.ts GraphNode.id). */
export function newGraphNodeId(): string {
return `n_${randomBytes(6).toString("hex")}`;
}
// ─── JSONL primitives ────────────────────────────────────────────────────────
function appendLine<T>(file: string, record: T): void {
fs.mkdirSync(path.dirname(file), { recursive: true });
fs.appendFileSync(file, `${JSON.stringify(record)}\n`, "utf-8");
}
function readRawLines(file: string): unknown[] {
let raw: string;
try {
raw = fs.readFileSync(file, "utf-8");
} catch {
return []; // no ledger yet — nothing has ever been emitted
}
const out: unknown[] = [];
for (const line of raw.split("\n")) {
const trimmed = line.trim();
if (trimmed.length === 0) continue;
try {
out.push(JSON.parse(trimmed));
} catch {
// Corrupt line: skip rather than fail the whole read — a torn
// append must never take the home graph down with it.
}
}
return out;
}
function isRecord(v: unknown): v is Record<string, unknown> {
return typeof v === "object" && v !== null;
}
const isString = (v: unknown): v is string => typeof v === "string";
/** Strict shape guards — anything not matching a W4 ledger shape is skipped. */
function readValid<T>(file: string, isValid: (v: unknown) => v is T): T[] {
return readRawLines(file).filter(isValid);
}
const isGraphNode = (v: unknown): v is GraphNode =>
isRecord(v) &&
isString(v.id) &&
isString(v.statement) &&
Array.isArray(v.aliases) &&
(v.kind === "truth" || v.kind === "misconception") &&
isString(v.createdAt);
const EDGE_TYPES = ["requires", "supports"] as const;
const isGraphEdge = (v: unknown): v is GraphEdge =>
isRecord(v) &&
isString(v.from) &&
isString(v.to) &&
(EDGE_TYPES as readonly string[]).includes(v.type as string) &&
Array.isArray(v.provenance) &&
v.provenance.every(isString);
const EVENT_KINDS: readonly LedgerEventKind[] = [
"probe-correct",
"probe-miss",
"dont-know",
"misconception-signaled",
"taught",
"check-correct",
"check-incorrect",
];
const isLedgerEvent = (v: unknown): v is LedgerEvent =>
isRecord(v) &&
(v.nodeId === null || isString(v.nodeId)) &&
(EVENT_KINDS as readonly string[]).includes(v.kind as string) &&
isString(v.chatId) &&
isString(v.at);
const isNodeMapping = (v: unknown): v is NodeMapping =>
isRecord(v) && isString(v.chatId) && isString(v.treeNodeId) && isString(v.graphNodeId);
const isCheckpoint = (v: unknown): v is SessionCheckpoint =>
isRecord(v) && isString(v.chatId) && isString(v.foldedAt);
// ─── typed ledger access ─────────────────────────────────────────────────────
export const graphLedgers = {
appendNode(node: GraphNode): void {
appendLine(LEDGER_FILES.nodes, node);
},
/** Raw node lines in append order (ids are unique — no merging needed). */
readNodes(): GraphNode[] {
return readValid(LEDGER_FILES.nodes, isGraphNode);
},
/** Append one edge ASSERTION; readers merge lines by (from,to,type). */
appendEdge(edge: GraphEdge): void {
appendLine(LEDGER_FILES.edges, edge);
},
readEdges(): GraphEdge[] {
return readValid(LEDGER_FILES.edges, isGraphEdge);
},
appendEvent(event: LedgerEvent): void {
appendLine(LEDGER_FILES.events, event);
},
readEvents(): LedgerEvent[] {
return readValid(LEDGER_FILES.events, isLedgerEvent);
},
appendMapping(mapping: NodeMapping): void {
appendLine(LEDGER_FILES.mappings, mapping);
},
readMappings(): NodeMapping[] {
return readValid(LEDGER_FILES.mappings, isNodeMapping);
},
appendCheckpoint(checkpoint: SessionCheckpoint): void {
appendLine(LEDGER_FILES.sessions, checkpoint);
},
readCheckpoints(): SessionCheckpoint[] {
return readValid(LEDGER_FILES.sessions, isCheckpoint);
},
};