diff --git a/app/api/chats/[id]/route.ts b/app/api/chats/[id]/route.ts index 265828a..029f520 100644 --- a/app/api/chats/[id]/route.ts +++ b/app/api/chats/[id]/route.ts @@ -1,4 +1,4 @@ -import { getChat, isTurnRunning, stores } from "@/src/server"; +import { deleteChatSession, getChat, isTurnRunning, stores, type ChatDeleteMode } from "@/src/server"; export async function GET(_req: Request, { params }: { params: Promise<{ id: string }> }) { const { id } = await params; @@ -28,3 +28,55 @@ export async function PUT(req: Request, { params }: { params: Promise<{ id: stri return Response.json({ title: record.title }); } +/** + * DELETE /api/chats/[id]?mode=preserve-graph|delete-nodes — session deletion + * (wave 6, owner rulings 2026-09-22/23). The mode is REQUIRED and explicit: + * no default, because the two differ in destructiveness — + * + * preserve-graph the session record, its jobs, and its extraction + * bookkeeping (coverage/checkpoint rows) die; the graph + * keeps every contribution the session merged. + * delete-nodes "this session never happened": every ledger trace of + * the session goes too — provenance, link lines, + * events, mappings — and orphaned nodes die entirely + * (shared nodes survive automatically, their other + * sessions' content intact). + * + * Gates are the server orchestration's (server/chats.ts deleteChatSession): + * 409 while a turn or background job is running, 503 while the session's + * extraction is winding down after its abort — both retryable. The + * response carries what was removed plus the new graphRevision (the + * version stamp the UI polls to refetch clusters/layout). + */ +export async function DELETE(req: Request, { params }: { params: Promise<{ id: string }> }) { + const { id } = await params; + const mode = new URL(req.url).searchParams.get("mode"); + if (mode !== "preserve-graph" && mode !== "delete-nodes") { + return Response.json( + { + error: 'mode is required and must be "preserve-graph" or "delete-nodes"', + code: "bad_mode", + }, + { status: 400 }, + ); + } + const outcome = await deleteChatSession(id, mode as ChatDeleteMode); + if (!outcome.ok) { + return Response.json( + { + error: outcome.error, + code: outcome.code, + ...("jobIds" in outcome ? { jobIds: outcome.jobIds } : {}), + }, + { status: outcome.status }, + ); + } + return Response.json({ + ok: true, + mode: outcome.mode, + stats: outcome.stats, + jobsRemoved: outcome.jobsRemoved, + graphRevision: outcome.graphRevision, + }); +} + diff --git a/src/agent/subagents/extraction.ts b/src/agent/subagents/extraction.ts index 1606823..9beea12 100644 --- a/src/agent/subagents/extraction.ts +++ b/src/agent/subagents/extraction.ts @@ -725,7 +725,12 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti * Idempotent per startedAt (the store dedupes). */ function persistRunUsage(chatId: string, durationMs: number, stats: EmissionStats | undefined): void { const run = runningExtractions.get(chatId); - if (!run) return; + if (!run) return; // not running (tests driving stages directly) + // SESSION-DELETION GUARD (wave 6): a chat deleted mid-run has no + // usage log to append to — the record's death is the user's choice + // and its partial spend logs with it (appendSourceUsage below + // already no-ops for a vanished source; the chat path guards here). + if (run.kind === "chat" && !chatStore.getChat(chatId)) return; const entry: ExtractionRunUsage = { startedAt: run.startedAt, at: new Date().toISOString(), @@ -787,6 +792,16 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti setPhase(chatId, "auditing", "chat"); const auditResult = await audit(chatId, findings, view); setPhase(chatId, "merging", "chat"); + // SESSION-DELETION GUARD (wave 6): a chat deleted mid-run must never + // merge. The deletion route aborts this run first and runs its purge + // as one sync block; this closes the last window — a run that slipped + // past the wind-down finds the chat gone HERE and rides the existing + // abort path (info, not error): nothing merged, nothing checkpointed. + if (!chatStore.getChat(chatId)) { + const gone = new Error(`chat deleted mid-run: ${chatId}`); + gone.name = "AbortError"; + throw gone; + } // The batch's messages, snapshotted when the run computed // coverage — if a message mutates mid-run (a quiz answer // merges in), the stored fingerprint won't match its new diff --git a/src/core/graph-deletion.ts b/src/core/graph-deletion.ts new file mode 100644 index 0000000..d2ece62 --- /dev/null +++ b/src/core/graph-deletion.ts @@ -0,0 +1,277 @@ +/** + * Session deletion — the FIRST destructive operation on the append-only + * graph ledgers (owner ruling 2026-09-22, wave 6 chunk A). + * + * Until now mergeFindings has been the sole writer; deletion is the second, + * and it is held to the same discipline: PURE code over the storage ports + * (core/ports.ts), no LLM judgment anywhere, deterministic and idempotent. + * The caller (the DELETE /api/chats/[id] route through the server + * orchestration) owns the cross-store choreography (chat record, jobs, + * extraction wind-down); this module owns the LEDGERS. + * + * Two modes (owner-locked semantics): + * + * preserve-graph the chat's extraction bookkeeping dies — its coverage + * rows and its legacy checkpoint lines — and NOTHING else: + * nodes, edges, events, mappings stay byte-identical. + * The graph keeps the session's contributions; the + * snapshot simply no longer resolves the dead chatId to + * a title (nodeSessions/nodeObservations filter unknown + * ids at read time — no UI ghosting). + * + * delete-nodes "this session never happened": every trace of the + * chat leaves the ledgers — its provenance points on + * every node and edge line, its link lines (their + * phrasings die with them), its quiz events and + * observations, its mapping lines (the audit trail), its + * coverage/checkpoint rows — and orphans die: a node + * whose references all came from this chat disappears + * entirely, pruned from the positions and embedding + * ledgers too (dead members would break cluster hulls). + * + * Shared nodes survive automatically (the ruling): a node lives iff ANY + * surviving reference remains — a provenance point from another chat or + * notes source, or a surviving event (another session's quiz outcome / + * observation) naming it. The create line of a surviving node is special: + * name/statement/createdAt are the node's IDENTITY — the shared anchor + * other sessions' link verdicts recognized — so they stay even when the + * deleted chat created the node; but that session's phrasings (the line's + * aliases) are stripped and its provenance points removed (owner ruling + * 2026-09-23: all traces, identity survives). + * + * Not touched, by design: + * - cluster-labels.jsonl — immutable history by contract, and the + * namer is graph-global (tied to no session); core-survival renames + * communities whose core shrank below half. + * - imported-notes sources (src_…) — a different front door with its + * own lifecycle (re-import dedupes by path hash); this wave deletes + * CHATS. + * + * Resurrection safety: the deletion runs as ONE synchronous block (no + * awaits — atomic vs the event loop, so no extraction append can + * interleave), and the extraction pipeline re-checks the chat record + * before each batch's merge (a deleted chat never merges). Both seams + * together close every corrupt/resurrect path. + */ + +import type { + CoverageEntry, + GraphEdge, + GraphNode, + LedgerEvent, + NodeMapping, + SessionCheckpoint, +} from "./contracts.ts"; +import type { GraphStore } from "./ports.ts"; + +/** The two deletion modes (owner-locked 2026-09-22). Machine-checkable on + * the API as the required `mode` query param — no default, so no + * accidental destructive choice. */ +export type ChatDeleteMode = "preserve-graph" | "delete-nodes"; + +/** What one deletion removed from the ledgers (the DELETE response's + * stats; the UI can show "3 topics, 2 links removed" from this). */ +export interface ChatDeletionStats { + /** Distinct nodes that died entirely (all their lines). */ + nodesRemoved: number; + /** Edge assertion LINES removed (empty-provenance or dead endpoint). */ + edgesRemoved: number; + /** Events removed (quiz outcomes + observations). */ + eventsRemoved: number; + /** Mapping lines removed (the chat's audit trail). */ + mappingsRemoved: number; + /** Coverage rows removed. */ + coverageRemoved: number; + /** Checkpoint lines removed. */ + checkpointsRemoved: number; +} + +const EMPTY_STATS: ChatDeletionStats = { + nodesRemoved: 0, + edgesRemoved: 0, + eventsRemoved: 0, + mappingsRemoved: 0, + coverageRemoved: 0, + checkpointsRemoved: 0, +}; + +/** + * Remove one chat's traces from the graph ledgers. Synchronous by design + * (the atomicity argument above); idempotent (a second run finds nothing + * and returns zero stats). + */ +export function deleteChatTraces(graphStore: GraphStore, chatId: string, mode: ChatDeleteMode): ChatDeletionStats { + if (mode !== "preserve-graph" && mode !== "delete-nodes") { + throw new Error(`Unknown deletion mode: ${mode as string}`); + } + + // The bookkeeping rows die in BOTH modes — coverage and checkpoints + // are extraction state, not graph content (owner semantics: even + // preserve-graph deletes "the chat's coverage rows"). + const coverage = graphStore.readCoverage().filter((row) => row.chatId !== chatId); + const checkpoints = graphStore.readCheckpoints().filter((row) => row.chatId !== chatId); + const stats: ChatDeletionStats = { + ...EMPTY_STATS, + coverageRemoved: graphStore.readCoverage().length - coverage.length, + checkpointsRemoved: graphStore.readCheckpoints().length - checkpoints.length, + }; + + if (mode === "preserve-graph") { + if (stats.coverageRemoved > 0 || stats.checkpointsRemoved > 0) { + graphStore.rewriteLedgers({ coverage, checkpoints }); + } + return stats; + } + + // ── delete-nodes: the full purge ────────────────────────────────────── + const nodeLines = graphStore.readNodes(); + const edgeLines = graphStore.readEdges(); + const events = graphStore.readEvents(); + const mappings = graphStore.readMappings(); + + // Does the chat touch the graph at all? (A never-extracted chat still + // needs its coverage rows dropped above, but no ledger rewrite — keeps + // mtimes stable so the graph revision only moves when something really + // changed.) + const touched = + nodeLines.some((line) => line.provenance.some((pt) => pt.chatId === chatId)) || + edgeLines.some((line) => line.provenance.some((pt) => pt.chatId === chatId)) || + events.some((event) => event.chatId === chatId) || + mappings.some((m) => m.chatId === chatId); + if (!touched) { + if (stats.coverageRemoved > 0 || stats.checkpointsRemoved > 0) { + graphStore.rewriteLedgers({ coverage, checkpoints }); + } + return stats; + } + + // 1. Survival: a node lives iff a reference from ANY surviving source + // remains — another chat/src's provenance point on any of its lines, + // a surviving mapping line (notes-front-door nodes carry EMPTY + // provenance: the file IS the anchor, recorded through the mapping + // line), or a surviving event naming it. (Edge endpoints are deliberately + // NOT survival references: an edge's asserting session always left + // provenance on its endpoints, so endpoint-only liveness cannot + // occur in a consistent ledger — and counting it would keep nodes + // alive whose lines we would then drop, dangling the edge.) + const survives = (points: ReadonlyArray<{ chatId: string }>): boolean => + points.some((pt) => pt.chatId !== chatId); + const live = new Set(); + for (const line of nodeLines) { + if (survives(line.provenance)) live.add(line.id); + } + for (const mapping of mappings) { + if (mapping.chatId !== chatId) live.add(mapping.graphNodeId); + } + for (const event of events) { + if (event.chatId !== chatId && event.nodeId !== null) live.add(event.nodeId); + } + + // 2. Node lines. Per-line (each line is ONE extraction run's append, so + // per-session alias provenance exists at line granularity): + // dead node every line dies + // untouched line kept verbatim + // partially-deleted kept with the chat's points stripped + // (defensive — single-run lines never mix chats) + // all points deleted CREATE line: identity survives, aliases + // stripped (owner ruling 2026-09-23), points + // empty / LINK line: dies with its phrasings + const outNodes: GraphNode[] = []; + const seenIds = new Set(); + for (const line of nodeLines) { + if (!live.has(line.id)) continue; // the whole node dies + const first = !seenIds.has(line.id); + seenIds.add(line.id); + const points = line.provenance.filter((pt) => pt.chatId !== chatId); + if (points.length === line.provenance.length) { + outNodes.push(line); // untouched by this chat + continue; + } + if (points.length > 0) { + outNodes.push({ ...line, provenance: points }); + continue; + } + if (first) { + // The create line of a surviving node the deleted chat minted: + // name/statement/createdAt are the shared identity other + // sessions recognized — they stay; the session's phrasings and + // provenance die (all traces, owner 2026-09-23). + outNodes.push({ ...line, provenance: [], aliases: [] }); + } + // else: the deleted chat's link line — dies with its phrasings. + } + + // 3. Edges: strip the chat's points; an assertion dies when its + // provenance empties or either endpoint died IN THIS DELETION. + // (Only nodes that existed and died count — an edge whose endpoints + // never resolved as nodes is foreign data the deletion leaves alone: + // destructive rewriting may not exceed the deleted chat's blast + // radius.) + const deadNodes = new Set([...new Set(nodeLines.map((line) => line.id))].filter((id) => !live.has(id))); + const outEdges: GraphEdge[] = []; + for (const line of edgeLines) { + if (deadNodes.has(line.from) || deadNodes.has(line.to)) continue; + const points = line.provenance.filter((pt) => pt.chatId !== chatId); + // An assertion dies when THIS deletion empties its provenance — a + // line that already had no points is foreign/legacy data, not this + // chat's trace; it rides untouched. + if (line.provenance.length > 0 && points.length === 0) continue; + outEdges.push(points.length === line.provenance.length ? line : { ...line, provenance: points }); + } + + // 4. Events, mappings: the chat's rows die outright. + const outEvents = events.filter((event) => event.chatId !== chatId); + const outMappings = mappings.filter((m) => m.chatId !== chatId); + + stats.nodesRemoved = [...new Set(nodeLines.map((line) => line.id))].filter((id) => !live.has(id)).length; + stats.edgesRemoved = edgeLines.length - outEdges.length; + stats.eventsRemoved = events.length - outEvents.length; + stats.mappingsRemoved = mappings.length - outMappings.length; + + graphStore.rewriteLedgers({ + nodes: outNodes, + edges: outEdges, + events: outEvents, + mappings: outMappings, + coverage, + checkpoints, + }); + + // 5. Positions: prune dead nodes (stale resting spots would pin + // nothing — but pruning keeps the ledger honest and the revision + // moving for exactly the right reason). + const positions = graphStore.readPositions(); + const knownNodeIds = new Set(nodeLines.map((line) => line.id)); + const prunedPositions = Object.fromEntries( + Object.entries(positions).filter(([id]) => live.has(id) || !knownNodeIds.has(id)), + ); + if (Object.keys(prunedPositions).length !== Object.keys(positions).length) { + graphStore.writePositions(prunedPositions); + } + + // 6. Embeddings: prune dead vectors and dead cluster members — a dead + // member in a hull list would break rendering (its position is + // gone). Clusters emptied by pruning drop out; sub-minimum ones + // vanish at the read-side filter. The model/dim headers stay. + const embeddings = graphStore.readEmbeddings(); + const prunedVectors = Object.fromEntries( + Object.entries(embeddings.vectors).filter(([id]) => live.has(id) || !knownNodeIds.has(id)), + ); + const prunedClusters = embeddings.clusters + .map((cluster) => ({ members: cluster.members.filter((id) => live.has(id) || !knownNodeIds.has(id)) })) + .filter((cluster) => cluster.members.length > 0); + if ( + Object.keys(prunedVectors).length !== Object.keys(embeddings.vectors).length || + prunedClusters.length !== embeddings.clusters.length || + prunedClusters.some((c, i) => c.members.length !== embeddings.clusters[i]!.members.length) + ) { + graphStore.writeEmbeddings({ + ...embeddings, + vectors: prunedVectors, + clusters: prunedClusters, + updatedAt: new Date().toISOString(), + }); + } + + return stats; +} \ No newline at end of file diff --git a/src/core/ports.ts b/src/core/ports.ts index 97acbef..683a9e1 100644 --- a/src/core/ports.ts +++ b/src/core/ports.ts @@ -60,6 +60,11 @@ export interface ChatStore { * Like saveSettings: a rename is NOT conversation activity — updatedAt * stays put, so sidebar order never churns on a rename. */ renameChat(id: string, title: string): ChatRecord; + /** Delete the session record entirely (session-deletion wave 6, owner + * 2026-09-22). False when the id is unknown. The graph-ledger purge and + * job cleanup are the caller's choreography (core/graph-deletion.ts + + * the DELETE route), not the store's. */ + deleteChat(id: string): boolean; } // ─── GraphStore — the append-only evidence ledgers (data/graph) ────────────── @@ -129,6 +134,23 @@ export interface GraphStore { readEmbeddings(): EmbeddingsLedger; /** Replace the whole ledger (embedding is a full-batch computation). */ writeEmbeddings(ledger: EmbeddingsLedger): void; + + /** THE DESTRUCTIVE SEAM (session-deletion wave 6, owner 2026-09-22 — + * the first non-append operation on these ledgers): atomically + * replace the full contents of any subset of the six rewritable + * ledgers. Each provided array is written to a temp file and renamed + * over the ledger (crash-safe: a torn write never replaces the + * original); absent keys leave their ledger untouched. The pure + * computation is core/graph-deletion.ts — the store only makes the + * replacement atomic. */ + rewriteLedgers(next: { + nodes?: GraphNode[]; + edges?: GraphEdge[]; + events?: LedgerEvent[]; + mappings?: NodeMapping[]; + coverage?: CoverageEntry[]; + checkpoints?: SessionCheckpoint[]; + }): void; } // ─── JobStore — background sub-agent jobs (data/jobs) ───────────────────────── @@ -154,6 +176,11 @@ export interface JobStore { listJobs(chatId: string): JobRecord[]; /** The job fulfilling a given tool part, if any (spawn-hook dedup). */ findJobByToolCallId(chatId: string, toolCallId: string): JobRecord | null; + /** Delete every job record for a chat — running or finished (session- + * deletion wave 6). Returns how many files were removed. The deletion + * route refuses while jobs are still running, so by the time this + * fires every record is terminal. */ + deleteJobsForChat(chatId: string): number; } // ─── SettingsStore — THE user settings record (settings-storage wave, 2026-09-19) ── diff --git a/src/server/chats.ts b/src/server/chats.ts index d654a47..68f433f 100644 --- a/src/server/chats.ts +++ b/src/server/chats.ts @@ -1,11 +1,112 @@ -import { stores } from "./stores.ts"; +import { abortExtraction, extractionStatuses } from "../agent/subagents/extraction.ts"; +import { deleteChatTraces, type ChatDeleteMode, type ChatDeletionStats } from "../core/graph-deletion.ts"; import { titleFromPrompt } from "../core/contracts.ts"; +import { graphRevision } from "./graph.ts"; +import { stores } from "./stores.ts"; +import { isTurnRunning } from "./turn-liveness.ts"; /** * Server-side chat entries for the transport layer (app/api/chats*, * app/chat/[id]). Today these delegate straight to the ChatStore port; the * services stage carves the real orchestration behind this same surface. + * + * deleteChatSession (wave 6, session deletion — the FIRST destructive + * operation on the graph ledgers) is the one real orchestration here: + * gates → extraction wind-down → one synchronous deletion block. See + * core/graph-deletion.ts for the ledger semantics and the atomicity + * argument. */ export const { createChat, getChat, listChats } = stores.chatStore; export { titleFromPrompt }; +export type { ChatDeleteMode, ChatDeletionStats } from "../core/graph-deletion.ts"; + +/** How long the wind-down wait may take before the caller gets a retryable + * 503 (an aborted extraction ends at its next model-call boundary — + * provider cancellation is normally sub-second, but a wedged provider + * must never wedge deletion). */ +const WIND_DOWN_TIMEOUT_MS = 10_000; +const WIND_DOWN_POLL_MS = 50; + +/** What deleteChatSession did — the route maps it straight to HTTP. */ +export type ChatDeleteOutcome = + | { ok: false; status: 404; code: "not_found"; error: string } + | { ok: false; status: 409; code: "turn_in_flight" | "jobs_running"; error: string; jobIds: string[] } + | { ok: false; status: 503; code: "extraction_winding_down"; error: string } + | { + ok: true; + mode: ChatDeleteMode; + stats: ChatDeletionStats; + jobsRemoved: number; + graphRevision: number; + }; + +/** + * Delete a session (owner rulings 2026-09-22/23): gates first, then the + * purge. The owner-locked choreography: + * + * 1. GATES — a turn in flight or a running background job refuses with + * 409 (they write the chat record and steer into it; deleting + * mid-flight would corrupt the write-ahead flow). The UI retries once + * work drains. + * 2. EXTRACTION WIND-DOWN — the chat's own pipeline run is TRUE- + * CANCELLED (abortExtraction: spend stops, nothing merges), then + * awaited off the registry. A timeout is a retryable 503, never a + * forced race. + * 3. THE SYNC BLOCK — everything below runs without a single await: + * no extraction append can interleave (the event loop never yields + * mid-block), so the read-filter-write of the ledger rewrites is + * atomic against the pipeline. Ledgers first, jobs next, the chat + * record LAST (a crash mid-block leaves a deletable chat with a + * purged graph — re-running DELETE fixes it — never a deleted chat + * with live traces). The pipeline's per-batch liveness guard + * (extraction.ts) closes the last window: a run that slipped past + * the wind-down finds the chat gone at its merge and aborts with + * nothing merged. + */ +export async function deleteChatSession(chatId: string, mode: ChatDeleteMode): Promise { + if (!stores.chatStore.getChat(chatId)) { + return { ok: false, status: 404, code: "not_found", error: "No such session" }; + } + if (isTurnRunning(chatId)) { + return { + ok: false, + status: 409, + code: "turn_in_flight", + error: "A turn is running in this session — try again when it finishes", + jobIds: [], + }; + } + const runningJobs = stores.jobStore.listJobs(chatId).filter((job) => job.status === "running"); + if (runningJobs.length > 0) { + return { + ok: false, + status: 409, + code: "jobs_running", + error: "Background jobs are still running in this session — try again when they finish", + jobIds: runningJobs.map((job) => job.id), + }; + } + if (extractionStatuses().some((run) => run.chatId === chatId)) { + abortExtraction(chatId); // true cancellation — spend stops, nothing merges + const deadline = Date.now() + WIND_DOWN_TIMEOUT_MS; + while (extractionStatuses().some((run) => run.chatId === chatId)) { + if (Date.now() > deadline) { + return { + ok: false, + status: 503, + code: "extraction_winding_down", + error: "The session's extraction is still winding down — try again in a moment", + }; + } + await new Promise((resolve) => setTimeout(resolve, WIND_DOWN_POLL_MS)); + } + } + + // THE SYNC BLOCK — no awaits below (see the doc comment). Order: + // ledgers → jobs → the chat record. + const stats = deleteChatTraces(stores.graphStore, chatId, mode); + const jobsRemoved = stores.jobStore.deleteJobsForChat(chatId); + stores.chatStore.deleteChat(chatId); + return { ok: true, mode, stats, jobsRemoved, graphRevision: graphRevision() }; +} diff --git a/src/server/graph.ts b/src/server/graph.ts index 79e3d29..e51f596 100644 --- a/src/server/graph.ts +++ b/src/server/graph.ts @@ -1,3 +1,5 @@ +import * as fs from "node:fs"; +import { EMBEDDINGS_FILE, LEDGER_FILES, POSITIONS_FILE } from "../../config.ts"; import type { GraphSnapshot } from "../core/contracts.ts"; import type { GraphCostSummary } from "../core/cost-summary.ts"; import { graphCostSummary } from "../core/cost-summary.ts"; @@ -35,3 +37,33 @@ export function graphCosts(): GraphCostSummary { .map((s) => ({ filePath: s.filePath, extractions: s.extractions! })); return graphCostSummary({ chats, sources, naming: stores.graphStore.readNamingRuns() }); } + +/** The graph's version stamp (session-deletion wave 6): the max mtime + * across every graph data file — ledgers, positions, embeddings. Moves + * on ANY graph write (merge append, settle, embed, or a deletion's + * atomic rewrite), never moves on reads, and needs no bookkeeping of + * its own: the fs IS the counter. Consumers poll it (the activity feed + * carries it; the DELETE response returns it) to know clusters/layout + * must refetch — the "graph changed underneath me" signal. */ +export function graphRevision(): number { + let revision = 0; + for (const file of [ + LEDGER_FILES.nodes, + LEDGER_FILES.edges, + LEDGER_FILES.events, + LEDGER_FILES.mappings, + LEDGER_FILES.sessions, + LEDGER_FILES.coverage, + LEDGER_FILES.clusterLabels, + LEDGER_FILES.namingRuns, + POSITIONS_FILE, + EMBEDDINGS_FILE, + ]) { + try { + revision = Math.max(revision, fs.statSync(file).mtimeMs); + } catch { + /* absent — nothing has ever written it */ + } + } + return revision; +} diff --git a/src/server/observability.ts b/src/server/observability.ts index 3bd21bf..bd0b84a 100644 --- a/src/server/observability.ts +++ b/src/server/observability.ts @@ -21,6 +21,7 @@ import { extractionStatuses, type ExtractionPhase } from "../agent/subagents/extraction.ts"; import { chatStore } from "../store/chats.ts"; +import { graphRevision } from "./graph.ts"; import { listJobs } from "./jobs.ts"; import type { JobRecord } from "../core/contracts.ts"; @@ -45,7 +46,13 @@ export interface ActivityExtraction { * the blob disappears when the process ends. Cost math is NOT done here: * raw usage rides on the records and the client folds it with the pure * helpers in core/prices.ts (transport stays thin). */ -export function activityFeed(): { jobs: ActivityJob[]; extractions: ActivityExtraction[] } { +/** The dock's whole world + the graph's version stamp (wave 6 session + * deletion): the payload carries graphRevision (max mtime across the + * graph data files — see server/graph.ts) so any view can detect "the + * graph changed underneath me" (a merge landed, a deletion purged) and + * refetch clusters/layout. Reads never move it; the DELETE response + * returns the same stamp. */ +export function activityFeed(): { jobs: ActivityJob[]; extractions: ActivityExtraction[]; graphRevision: number } { const chats = chatStore.listChats(); const jobs: ActivityJob[] = []; for (const chat of chats) { @@ -63,5 +70,9 @@ export function activityFeed(): { jobs: ActivityJob[]; extractions: ActivityExtr })); // oldest-first on last activity: the stable blob order (ISO strings sort chronologically) jobs.sort((a, b) => a.updatedAt.localeCompare(b.updatedAt)); - return { jobs, extractions: extractions.sort((a, b) => a.startedAt - b.startedAt) }; + return { + jobs, + extractions: extractions.sort((a, b) => a.startedAt - b.startedAt), + graphRevision: graphRevision(), + }; } diff --git a/src/store/chats.ts b/src/store/chats.ts index 663e38c..0326188 100644 --- a/src/store/chats.ts +++ b/src/store/chats.ts @@ -129,6 +129,19 @@ function renameChat(id: string, title: string): ChatRecord { return record; } +/** Delete the session record file entirely (session-deletion wave 6, + * owner 2026-09-22). False when the id is unknown — the route turns that + * into a 404. The record's graph traces and jobs are the route's + * choreography (core/graph-deletion.ts), not the store's. */ +function deleteChat(id: string): boolean { + try { + fs.unlinkSync(chatFile(id)); + return true; + } catch { + return false; // never existed, already deleted, or unreadable — same answer + } +} + /** Persist the session plan tree (plan tools call this in execute). */ function saveTree(id: string, tree: PlanTree): void { const record = getChat(id); @@ -147,6 +160,7 @@ export const chatStore: ChatStore = { saveTree, saveSettings, renameChat, + deleteChat, appendTurnUsage, appendExtractionUsage, }; diff --git a/src/store/graph.ts b/src/store/graph.ts index c5da267..f45c441 100644 --- a/src/store/graph.ts +++ b/src/store/graph.ts @@ -260,6 +260,49 @@ export const graphStore: GraphStore = { writeEmbeddings(ledger: EmbeddingsLedger): void { writeEmbeddingsLedger(ledger); }, + + /** THE DESTRUCTIVE SEAM (wave 6 session deletion): atomic full-content + * replacement per provided ledger — write a uniquely-named temp file + * beside the ledger, then rename over it (POSIX-atomic: a crash mid- + * write leaves the original ledger and an orphan .tmp-*, never a torn + * ledger). Empty arrays write an empty file — a purged ledger reads + * as "nothing yet", same as never-emitted. */ + rewriteLedgers(next: { + nodes?: GraphNode[]; + edges?: GraphEdge[]; + events?: LedgerEvent[]; + mappings?: NodeMapping[]; + coverage?: CoverageEntry[]; + checkpoints?: SessionCheckpoint[]; + }): void { + const files = { + nodes: LEDGER_FILES.nodes, + edges: LEDGER_FILES.edges, + events: LEDGER_FILES.events, + mappings: LEDGER_FILES.mappings, + coverage: LEDGER_FILES.coverage, + checkpoints: LEDGER_FILES.sessions, + } as const; + for (const [key, records] of Object.entries(next) as [keyof typeof files, unknown[]][]) { + const file = files[key]; + fs.mkdirSync(path.dirname(file), { recursive: true }); + const tmp = `${file}.tmp-${crypto.randomUUID().slice(0, 8)}`; + try { + // JSONL: one record per line, trailing newline — a reader + // must see exactly what appendLine would have written. + const body = records.map((record) => JSON.stringify(record)).join("\n") + (records.length > 0 ? "\n" : ""); + fs.writeFileSync(tmp, body, "utf-8"); + fs.renameSync(tmp, file); + } catch (err) { + try { + fs.unlinkSync(tmp); + } catch { + /* the rename either already happened or the tmp never existed */ + } + throw err; + } + } + }, }; /** positions.json — { positions, updatedAt }. Read-modify-write, NOT diff --git a/src/store/jobs.ts b/src/store/jobs.ts index d32bd6c..32b55bc 100644 --- a/src/store/jobs.ts +++ b/src/store/jobs.ts @@ -79,10 +79,28 @@ function findJobByToolCallId(chatId: string, toolCallId: string): JobRecord | nu return listJobs(chatId).find((j) => j.toolCallId === toolCallId) ?? null; } +/** Delete every job record file for a chat (session-deletion wave 6) — + * running or finished, they are the chat's records and die with it. + * Returns how many files were removed. The deletion route refuses while + * jobs are still running, so every record here is terminal. */ +function deleteJobsForChat(chatId: string): number { + let removed = 0; + for (const job of listJobs(chatId)) { + try { + fs.unlinkSync(jobFile(job.id)); + removed++; + } catch { + /* already gone — the count is best-effort truth */ + } + } + return removed; +} + export const jobStore: JobStore = { createJob, getJob, updateJob, listJobs, findJobByToolCallId, + deleteJobsForChat, }; diff --git a/tests/extraction.test.ts b/tests/extraction.test.ts index fd0e4f2..1fe3192 100644 --- a/tests/extraction.test.ts +++ b/tests/extraction.test.ts @@ -135,6 +135,16 @@ function memoryLedgers() { for (const [id, v] of Object.entries(next.vectors)) embeddings.vectors.set(id, [...v]); embeddings.clusters = next.clusters.map((c) => [...c.members]); }, + // The destructive seam (wave 6 session deletion) — mirrors the real + // store: replace the provided ledgers' full contents. + rewriteLedgers: (next) => { + if (next.nodes) nodes.splice(0, nodes.length, ...next.nodes.map((n) => ({ ...n, provenance: [...n.provenance], aliases: [...n.aliases] }))); + if (next.edges) edges.splice(0, edges.length, ...next.edges.map((e) => ({ ...e, provenance: [...e.provenance] }))); + if (next.events) events.splice(0, events.length, ...next.events.map((e) => ({ ...e }))); + if (next.mappings) mappings.splice(0, mappings.length, ...next.mappings.map((m) => ({ ...m }))); + if (next.coverage) coverage.splice(0, coverage.length, ...next.coverage.map((c) => ({ ...c }))); + if (next.checkpoints) checkpoints.splice(0, checkpoints.length, ...next.checkpoints.map((c) => ({ ...c }))); + }, } satisfies GraphStore; } @@ -1253,3 +1263,45 @@ describe("the chunked pipeline (extraction coverage)", () => { assert.equal(coverageOf(messages(), ledgers.readCoverage(), chatId).uncoveredCount, 0, "re-covered"); }); }); + +// ─── session deletion mid-run (wave 6) ───────────────────────────────────── + +describe("the session-deletion guard (chat deleted mid-run)", () => { + it("a run whose chat dies mid-extraction aborts at the merge with nothing merged, checkpointed, or billed", async () => { + const { chatId } = fixtureChat(); + const ledgers = memoryLedgers(); + + // The chat record is deleted DURING the audit model call — the worst + // window: findings already spent, the merge next. + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + embedder: fakeEmbedder, + extractor: submitModel("submit_findings", FINDINGS), + auditor: submitModel("submit_review", { propositions: FINDINGS.propositions, edges: [], attribution: { quiz: {} }, dropped: [] }, () => chatStore.deleteChat(chatId)), + labeler: submitModel("submit_labels", { labels: [] }), + }); + const stats = await pipeline.runExtraction(chatId); + + // The abort rides the existing abort path: info, not error; zero + // emissions (stats present but empty — nothing merged). + assert.ok(stats); + assert.equal(stats.nodes, 0); + assert.equal(stats.edges, 0); + assert.equal(stats.events, 0); + + // Nothing merged, nothing checkpointed — the deleted chat leaves no + // trace for a later snapshot to resurrect. + assert.equal(ledgers.readNodes().length, 0); + assert.equal(ledgers.readEdges().length, 0); + assert.equal(ledgers.readEvents().length, 0); + assert.equal(ledgers.readMappings().length, 0); + assert.equal(ledgers.readCoverage().length, 0); + assert.equal(ledgers.readCheckpoints().length, 0); + + // The run is off the registry (the finally always cleans up). + assert.equal(extractionStatuses().some((run) => run.chatId === chatId), false); + // And its partial spend died with the chat — no usage row on a ghost. + assert.equal(chatStore.getChat(chatId), null); + }); +}); diff --git a/tests/graph.test.ts b/tests/graph.test.ts index c75073a..2105254 100644 --- a/tests/graph.test.ts +++ b/tests/graph.test.ts @@ -634,3 +634,211 @@ describe("graph ledger store", () => { assert.deepEqual(merged.provenance.map((p) => p.messageId), ["m1", "m2"]); }); }); + +// ─── session deletion (wave 6 — the first destructive ledger operation) ───── + +import { deleteChatTraces, type ChatDeletionStats } from "../src/core/graph-deletion.ts"; +import { deleteChatSession } from "../src/server/chats.ts"; +import { beginTurn, endTurn } from "../src/server/turn-liveness.ts"; +import { jobStore } from "../src/store/jobs.ts"; + +describe("session deletion (deleteChatTraces + deleteChatSession, wave 6)", () => { + /** Two-session scenario: dA creates the shared node X + the orphan Y and + * asserts X→Y; dB links X (its own phrasings), creates Z, asserts X→Z. + * Deleting dA must: keep X (dB's link line survives, dA's create-line + * aliases stripped — the identity stays), kill Y (orphan), keep Z and + * dB's edge, kill dA's events/mappings/coverage/checkpoints. */ + function buildTwoSessionScenario() { + const dA = fixtureChat("Deletion session A", [ + user("du1", "teach me the deletion topic"), + assistant("da1", [ + quizPart( + "d-call-1", + { question: "DQ1", options: [{ label: "a" }, { label: "b" }], correctAnswer: "a", explanation: "e" }, + answer({ questionId: "d-call-1", questionText: "DQ1 text", correct: true, answeredAt: "2026-09-23T00:00:00.000Z" }), + ), + ]), + ]); + const auditA: AuditResult = { + propositions: [ + { id: "p1", name: "Shared core", statement: "Shared core statement", aliases: ["shared feel"], messages: ["du1"], evidence: ["e"], understanding: "solid" }, + { id: "p2", name: "Solo fact", statement: "Solo fact statement", aliases: ["solo feel"], messages: ["du1"], evidence: ["e"] }, + ], + edges: [{ from: "p1", to: "p2", type: "follows", messages: ["du1"], evidence: ["e"] }], + attribution: { quiz: { "d-call-1": "p1" } }, + dropped: [], + }; + mergeFindings(dA.id, auditA, graphDeps, sessionSource(dA.id)); + // Resolve through THIS session's mappings — the shared ledger holds + // lookalike statements from earlier tests, so statement-lookup would + // link dB to the wrong node. + const mappedId = (sourceId: string): string => + graphStore.readMappings().find((m) => m.chatId === dA.id && m.sourceId === sourceId)!.graphNodeId; + const xId = mappedId("p1"); + const yId = mappedId("p2"); + + const dB = fixtureChat("Deletion session B", [ + user("eu1", "more on the shared core"), + assistant("eb1", [ + quizPart( + "e-call-1", + { question: "EQ1", options: [{ label: "a" }, { label: "b" }], correctAnswer: "a", explanation: "e" }, + answer({ questionId: "e-call-1", questionText: "EQ1 text", correct: false, answeredAt: "2026-09-23T00:01:00.000Z" }), + ), + ]), + ]); + const auditB: AuditResult = { + propositions: [ + { id: "q1", name: "Shared again", statement: "Shared core, rephrased", aliases: ["B phrasing"], messages: ["eu1"], evidence: ["e"], linkTo: xId, understanding: "shaky" }, + { id: "q2", name: "B extra", statement: "B extra statement", aliases: ["b extra feel"], messages: ["eu1"], evidence: ["e"] }, + ], + edges: [{ from: "q1", to: "q2", type: "follows", messages: ["eu1"], evidence: ["e"] }], + attribution: { quiz: { "e-call-1": "q1" } }, + dropped: [], + }; + mergeFindings(dB.id, auditB, graphDeps, sessionSource(dB.id)); + const zId = graphStore.readMappings().find((m) => m.chatId === dB.id && m.sourceId === "q2")!.graphNodeId; + + // Coverage rows (both chats) + positions + embeddings garnish — the + // deletion prunes all three ledgers. + for (const chat of [dA, dB]) { + for (let i = 0; i < 2; i++) { + graphStore.appendCoverage({ chatId: chat.id, fp: `${chat.id}-${i}`, tokens: 10, foldedAt: "2026-09-23T00:02:00.000Z" }); + } + } + graphStore.writePositions({ ...graphStore.readPositions(), [xId]: { x: 1, y: 1 }, [yId]: { x: 2, y: 2 }, [zId]: { x: 3, y: 3 } }); + const emb = graphStore.readEmbeddings(); + graphStore.writeEmbeddings({ + ...emb, + vectors: { ...emb.vectors, [xId]: [1, 0], [yId]: [0, 1], [zId]: [1, 1] }, + clusters: [...emb.clusters, { members: [xId, yId, zId] }], + updatedAt: "2026-09-23T00:03:00.000Z", + }); + return { dA, dB, xId, yId, zId }; + } + + it("delete-nodes: the shared node survives with the other session's content; the orphan dies; the creator's aliases are stripped (owner ruling 2026-09-23)", () => { + const { dA, dB, xId, yId, zId } = buildTwoSessionScenario(); + const zLineBefore = JSON.stringify(graphStore.readNodes().find((n) => n.id === zId)); + + const stats = deleteChatTraces(graphStore, dA.id, "delete-nodes"); + assert.deepEqual(stats, { + nodesRemoved: 1, + edgesRemoved: 1, + eventsRemoved: 2, + mappingsRemoved: 2, + coverageRemoved: 2, + checkpointsRemoved: 1, + } satisfies ChatDeletionStats); + + const graph = getGraph(graphStore); + // X survives: identity (name/statement/createdAt from the create + // line) intact, dA's aliases + provenance gone, dB's phrasings in. + const x = graph.nodes.find((n) => n.id === xId)!; + assert.equal(x.statement, "Shared core statement", "the shared identity survives"); + assert.equal(x.name, "Shared core"); + assert.ok(!x.aliases.includes("shared feel"), "the deleted creator's phrasings are stripped (owner ruling)"); + assert.ok(x.aliases.includes("B phrasing") && x.aliases.includes("Shared core, rephrased")); + assert.deepEqual(x.provenance, [{ chatId: dB.id, messageId: "eu1" }], "only dB's provenance remains"); + // Y (the orphan) is gone entirely. + assert.equal(graph.nodes.some((n) => n.id === yId), false); + assert.equal(graphStore.readNodes().some((n) => n.id === yId), false); + // Z (dB-only) is untouched, byte-identical. + assert.equal(JSON.stringify(graphStore.readNodes().find((n) => n.id === zId)), zLineBefore); + // Edges: dA's assertion X→Y died; dB's X→Z survived. + const scoped = scopedGraph([dA.id, dB.id]); + assert.deepEqual(scoped.edges, [{ from: xId, to: zId, type: "follows", provenance: [{ chatId: dB.id, messageId: "eu1" }] }]); + // Events: only dB's quiz + observation remain. + assert.deepEqual(scoped.events.map((e) => [e.kind, e.chatId === dB.id]), [ + ["miss", true], + ["observation", true], + ]); + // Mappings: dA's audit trail died; dB's stays. + const mappings = graphStore.readMappings().filter((m) => [dA.id, dB.id].includes(m.chatId)); + assert.deepEqual(mappings, [{ chatId: dB.id, sourceId: "q1", graphNodeId: xId }, { chatId: dB.id, sourceId: "q2", graphNodeId: zId }]); + // Positions/embeddings: the orphan pruned everywhere. + const positions = graphStore.readPositions(); + assert.equal(positions[yId], undefined); + assert.deepEqual(positions[xId], { x: 1, y: 1 }); + const emb = graphStore.readEmbeddings(); + assert.equal(emb.vectors[yId], undefined); + assert.ok(emb.vectors[xId] && emb.vectors[zId]); + assert.deepEqual(emb.clusters.at(-1), { members: [xId, zId] }, "dead members pruned from the hull"); + // The snapshot resolves X to dB only — the deleted chat left no ghost. + const snapshot = graphSnapshot(graphDeps); + assert.deepEqual(snapshot.nodeSessions[xId], [dB.id]); + assert.deepEqual(snapshot.nodeObservations[xId], { [dB.id]: "shaky" }); + assert.ok(snapshot.sessions[dB.id] && !snapshot.sessions[dA.id]); + // Atomic rewrites left no temp files, and appends still work. + assert.equal(fs.readdirSync(PATHS.graph).filter((f) => f.includes(".tmp-")).length, 0); + const eventsBefore = graphStore.readEvents().length; + graphStore.appendEvent({ nodeId: null, kind: "correct", chatId: "chat_del_append_probe", at: "2026-09-23T00:04:00.000Z" }); + assert.equal(graphStore.readEvents().length, eventsBefore + 1); + // Idempotent: a second run finds nothing. + assert.deepEqual(deleteChatTraces(graphStore, dA.id, "delete-nodes"), { ...stats, nodesRemoved: 0, edgesRemoved: 0, eventsRemoved: 0, mappingsRemoved: 0, coverageRemoved: 0, checkpointsRemoved: 0 } satisfies ChatDeletionStats); + }); + + it("preserve-graph: only the extraction bookkeeping dies — the graph ledgers are byte-identical", () => { + const { dA, dB, xId, yId, zId } = buildTwoSessionScenario(); + const before = { + nodes: graphStore.readNodes(), + edges: graphStore.readEdges(), + events: graphStore.readEvents(), + mappings: graphStore.readMappings(), + positions: graphStore.readPositions(), + embeddings: graphStore.readEmbeddings(), + }; + const stats = deleteChatTraces(graphStore, dA.id, "preserve-graph"); + assert.deepEqual(stats, { + nodesRemoved: 0, + edgesRemoved: 0, + eventsRemoved: 0, + mappingsRemoved: 0, + coverageRemoved: 2, + checkpointsRemoved: 1, + } satisfies ChatDeletionStats); + assert.deepEqual(graphStore.readNodes(), before.nodes); + assert.deepEqual(graphStore.readEdges(), before.edges); + assert.deepEqual(graphStore.readEvents(), before.events); + assert.deepEqual(graphStore.readMappings(), before.mappings); + assert.deepEqual(graphStore.readPositions(), before.positions); + assert.deepEqual(graphStore.readEmbeddings(), before.embeddings); + assert.equal(graphStore.readCoverage().some((row) => row.chatId === dA.id), false); + assert.equal(graphStore.readCoverage().some((row) => row.chatId === dB.id), true); + void xId; void yId; void zId; + }); + + it("deleteChatSession: the orchestration's gates and the sync block", async () => { + // 404 — unknown chat. + assert.deepEqual(await deleteChatSession("chat_missing_delete", "delete-nodes"), { ok: false, status: 404, code: "not_found", error: "No such session" }); + + const { dA } = buildTwoSessionScenario(); + // 409 — a turn in flight. + beginTurn(dA.id); + const turnedAway = await deleteChatSession(dA.id, "delete-nodes"); + assert.deepEqual(turnedAway, { ok: false, status: 409, code: "turn_in_flight", error: "A turn is running in this session — try again when it finishes", jobIds: [] }); + endTurn(dA.id); + // 409 — running background jobs. + const job = jobStore.createJob({ kind: "research", chatId: dA.id, toolCallId: "del-call-1" }); + const jobsBlocked = await deleteChatSession(dA.id, "delete-nodes"); + assert.ok(!jobsBlocked.ok && jobsBlocked.status === 409, "refused with 409"); + if (jobsBlocked.code === "jobs_running") assert.deepEqual(jobsBlocked.jobIds, [job.id]); + assert.equal(jobStore.deleteJobsForChat(dA.id), 1, "test cleanup — the job file is gone"); + + // Success — a terminal job rides along; the chat record dies last. + const doneJob = jobStore.createJob({ kind: "research", chatId: dA.id, toolCallId: "del-call-2" }); + jobStore.updateJob(doneJob.id, { status: "done", activity: "delivered" }); + const outcome = await deleteChatSession(dA.id, "delete-nodes"); + assert.equal(outcome.ok, true); + if (outcome.ok) { + assert.equal(outcome.mode, "delete-nodes"); + assert.equal(outcome.stats.nodesRemoved, 1); + assert.equal(outcome.jobsRemoved, 1); + assert.ok(outcome.graphRevision > 0, "the version stamp moved"); + } + assert.equal(chatStore.getChat(dA.id), null, "the chat record is gone"); + assert.equal(jobStore.getJob(doneJob.id), null, "the chat's job records are gone"); + // And a second call is an honest 404. + assert.deepEqual(await deleteChatSession(dA.id, "delete-nodes"), { ok: false, status: 404, code: "not_found", error: "No such session" }); + }); +}); diff --git a/tests/web-search.test.ts b/tests/web-search.test.ts index 4dfd373..1b062ac 100644 --- a/tests/web-search.test.ts +++ b/tests/web-search.test.ts @@ -108,6 +108,7 @@ function fakeJobStore(): JobStore { }, listJobs: () => [...jobs.values()], findJobByToolCallId: () => null, + deleteJobsForChat: () => 0, // the drop tests never delete }; }