Merge branch 'feat/edge-glyph'

This commit is contained in:
amosblomqvist
2026-09-19 22:39:03 +03:00
15 changed files with 740 additions and 68 deletions
+44 -15
View File
@@ -19,7 +19,8 @@ import { z } from "zod";
import { lhTool, toPiTools } from "../lh-tool.ts";
import type { PiEngine } from "../models.ts";
import { runSubagent } from "./pi-run.ts";
import type { GraphNode } from "../../core/contracts.ts";
import type { CallUsage, GraphNode, NamingRunUsage } from "../../core/contracts.ts";
import { callUsage } from "./extraction.ts";
import { getGraph, clusterContentId } from "../../core/graph.ts";
import { matchClusterLabelings, MIN_CLUSTER_SIZE, type Community } from "../../core/clusters.ts";
import type { GraphStore } from "../../core/ports.ts";
@@ -68,6 +69,7 @@ async function nameClusters(
engine: PiEngine | undefined,
signal: AbortSignal | undefined,
unnamed: UnnamedCluster[],
onStep?: (usage: CallUsage, model: string) => void,
): Promise<{ key: string; label: string }[]> {
let submitted: { key: string; label: string }[] | undefined;
await runSubagent(
@@ -98,6 +100,7 @@ async function nameClusters(
),
maxTurns: 4,
abortSignal: signal,
onAssistantMessage: (message) => onStep?.(callUsage(message.usage), message.model),
shouldStop: (_message, toolResults) =>
toolResults.some((r) => r.toolName === "submit_labels" && !r.isError),
},
@@ -156,23 +159,49 @@ async function doRefresh(deps: {
if (unnamed.length === 0) return { detected: communities.length, claimed, named: 0, appended: 0 };
const labels = await nameClusters(engine, signal, unnamed);
// Money channel (2026-09 graph cost metrics): the naming call is the
// pipeline's third billable stage — capture per-step usage + model as
// they bill, then persist ONE record for the run, success or failure
// (money spent either way, same discipline as extraction runs).
const startedAt = Date.now();
const t0 = performance.now();
const steps: NamingRunUsage["steps"] = [];
let model = "unknown";
let appended = 0;
const seen = new Set<string>();
for (const { key, members } of unnamed) {
if (seen.has(key)) continue;
const label = labels.find((l) => l.key === key)?.label?.trim();
if (!label) continue;
seen.add(key);
graphStore.appendClusterLabel({
id: clusterContentId(members),
label,
core: members,
createdAt: new Date().toISOString(),
try {
const labels = await nameClusters(engine, signal, unnamed, (usage, modelId) => {
steps.push(usage);
model = modelId;
});
const seen = new Set<string>();
for (const { key, members } of unnamed) {
if (seen.has(key)) continue;
const label = labels.find((l) => l.key === key)?.label?.trim();
if (!label) continue;
seen.add(key);
graphStore.appendClusterLabel({
id: clusterContentId(members),
label,
core: members,
createdAt: new Date().toISOString(),
});
appended++;
}
return { detected: communities.length, claimed, named: unnamed.length, appended };
} finally {
// The finally (not just success) keeps failed/aborted runs billed
// too — zero appended, steps as far as they got. Runs whose model
// never billed a step (aborted before the first message) still
// record: the idempotency key is this run's startedAt.
graphStore.appendNamingUsage({
startedAt,
at: new Date().toISOString(),
durationMs: Math.round(performance.now() - t0),
model,
steps,
stats: { detected: communities.length, claimed, named: unnamed.length, appended },
});
appended++;
}
return { detected: communities.length, claimed, named: unnamed.length, appended };
}
let namingTail: Promise<unknown> = Promise.resolve();
+16 -13
View File
@@ -588,6 +588,22 @@ export interface ExtractionPipeline {
importNotes(input: NotesImportInput): Promise<EmissionStats>;
}
/** pi's per-step usage, flattened to the persisted CallUsage fields
* (the money channel's capture shape, contracts/chat.ts). Exported for
* the cluster namer — the same capture shape (2026-09 graph cost
* metrics wave). Module scope: no closure deps, pure projection. */
export function callUsage(usage: Usage): CallUsage {
return {
inputTokens: usage.input,
outputTokens: usage.output,
cacheReadTokens: usage.cacheRead,
cacheWriteTokens: usage.cacheWrite,
// Optional in pi's Usage (providers that never emit reasoning
// tokens leave it unset) — persisted as 0, never dropped.
reasoningTokens: usage.reasoning ?? 0,
};
}
export function createExtractionPipeline(deps: ExtractionPipelineDeps): ExtractionPipeline {
const { chatStore, graphStore, embedder: embed } = deps;
// THE SHARED REGISTRIES (runs() above — globalThis): every pipeline
@@ -619,19 +635,6 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
run.usage[stage]!.steps.push(usage);
}
/** pi's per-step usage, flattened to the persisted CallUsage fields
* (the money channel's capture shape, contracts/chat.ts). */
function callUsage(usage: Usage): CallUsage {
return {
inputTokens: usage.input,
outputTokens: usage.output,
cacheReadTokens: usage.cacheRead,
cacheWriteTokens: usage.cacheWrite,
// Optional in pi's Usage (providers that never emit reasoning
// tokens leave it unset) — persisted as 0, never dropped.
reasoningTokens: usage.reasoning ?? 0,
};
}
async function extract(chatId: string, view: string): Promise<SessionFindings> {
const signal = runningExtractions.get(chatId)?.controller.signal;
+36 -1
View File
@@ -20,7 +20,7 @@
/** Two relations only (DECIDED 2026-09-02): follows = b is derivable from a;
* supports = a grounds b without entailing it. No scalars, no encoded
* strength — strength is derived at read time from topology. */
import type { ExtractionRunUsage } from "./chat.ts";
import type { CallUsage, ExtractionRunUsage } from "./chat.ts";
export type EdgeType = "follows" | "supports";
@@ -65,6 +65,41 @@ export interface NotesSource {
extractions?: ExtractionRunUsage[];
}
/** What one cluster-naming refresh did — the namer's ClusterRefreshStats,
* contract-shaped so the agent layer needs no contracts import.
* Structural twins stay interchangeable. */
export interface NamingRunStats {
/** Communities the current graph divides into (all sizes). */
detected: number;
/** Communities that kept a label via core-survival. */
claimed: number;
/** Communities sent to the namer (unnamed, size >= MIN_CLUSTER_SIZE). */
named: number;
/** Labeling records actually appended (idempotent dedupe may shrink). */
appended: number;
}
/** One cluster-naming run's usage record — the money channel's THIRD stage
* (graph cost metrics, 2026-09). The namer is graph-global (tied to no
* chat or source), so runs live in their own append-only ledger
* (config.LEDGER_FILES.namingRuns). Appended when a run actually called
* the model; skipped refreshes (nothing to name) never bill and never
* append. Idempotent per startedAt — same discipline as every usage
* channel. Cost is INFERRED at read time from steps + model
* (core/prices.ts) — dollars are never stored. */
export interface NamingRunUsage {
/** ms epoch — the idempotency key. */
startedAt: number;
/** ISO — wall-clock time of the run. */
at: string;
durationMs: number;
/** The labeler role's model id. */
model: string;
/** Every billed call (usually ONE — the namer is single-submit). */
steps: CallUsage[];
stats: NamingRunStats;
}
export interface GraphNode {
/** Opaque stable id, store-assigned: "n_" + nanoid-like. */
id: string;
+185
View File
@@ -0,0 +1,185 @@
/**
* Graph-pipeline cost summary (2026-09 graph cost metrics wave) — the pure
* fold behind the graph view's cost pill. Three usage channels in: chat
* extraction runs (ChatRecord.extractions), notes-source runs
* (NotesSource.extractions), and cluster-naming runs (naming-runs.jsonl).
* Out: per-stage totals (extraction / audit / cluster-naming) + a capped
* recent-runs list.
*
* MONEY CHANNEL LAW: dollars are INFERRED here at read time via
* core/prices.ts — never stored anywhere. An unpriced model contributes
* ZERO dollars but is counted in `unpricedRuns` so the UI can render
* "— unpriced" honestly, never a silent zero. Tokens always sum.
*
* Client-safe: pure types + math, no node imports; exposed via the core
* barrel so the pill/panel components carry no cost math.
*/
import type { CallUsage, EmissionStats, ExtractionRunUsage, NamingRunUsage } from "./contracts.ts";
import { callCost, sumSteps } from "./prices.ts";
/** One pipeline stage's cumulative cost across every run. */
export interface StageCost {
/** Inferred USD summed over PRICED runs (partial when unpricedRuns > 0). */
dollars: number;
/** Stage-runs whose model is missing from PRICES (render "—", never zero). */
unpricedRuns: number;
/** Runs that billed this stage. */
runs: number;
/** Token totals across every run. */
tokens: CallUsage;
}
/** One run in the recent-runs list (extraction or naming). */
export interface RecentRun {
startedAt: number;
/** ISO wall-clock. */
at: string;
durationMs: number;
/** "extraction" covers both front doors (chat + notes). */
kind: "extraction" | "naming";
/** Chat title / source file path / fixed label for naming runs. */
title: string;
/** Whole-run inferred USD; null only when EVERY stage is unpriced. */
dollars: number | null;
tokens: CallUsage;
/** What the run merged into the graph; absent on failed runs. */
merged?: EmissionStats;
/** Naming runs: labeling records appended (extraction runs omit it). */
labels?: number;
}
/** The cost pill's whole payload: stage totals + recent runs. */
export interface GraphCostSummary {
/** Inferred USD across all priced stages of all runs. */
totalDollars: number;
/** Runs where EVERY stage was unpriced (totalDollars excludes them). */
unpricedRuns: number;
/** The three billable pipeline stages (merger/embedder are free). */
stages: {
extraction: StageCost;
audit: StageCost;
naming: StageCost;
};
/** Newest first, capped at `recentCap` (default 10). */
runs: RecentRun[];
}
/** What the fold reads: already-collected run records, front-door-tagged. */
export interface CostSummaryInput {
/** Chat-front-door runs: chat title + the record's usage log. */
chats: { title: string; extractions: ExtractionRunUsage[] }[];
/** Notes-front-door runs: imported file path + the source's usage log. */
sources: { filePath: string; extractions: ExtractionRunUsage[] }[];
/** Cluster-naming runs (graph-global ledger). */
naming: NamingRunUsage[];
recentCap?: number;
}
const ZERO_TOKENS: CallUsage = {
inputTokens: undefined,
outputTokens: undefined,
cacheReadTokens: undefined,
cacheWriteTokens: undefined,
reasoningTokens: undefined,
};
const emptyStage = (): StageCost => ({ dollars: 0, unpricedRuns: 0, runs: 0, tokens: { ...ZERO_TOKENS } });
/** One stage's dollars over its steps; null when the model is unpriced
* (all-or-nothing: a stage runs one model). */
function stageDollars(model: string, steps: readonly CallUsage[]): number | null {
let total = 0;
for (const step of steps) {
const c = callCost(step, model);
if (c === null) return null;
total += c;
}
return total;
}
/** Merge one step-set's tokens into an accumulating CallUsage. */
function mergeTokens(into: CallUsage, steps: readonly CallUsage[]): void {
const summed = sumSteps(steps);
into.inputTokens = (into.inputTokens ?? 0) + (summed.inputTokens ?? 0);
into.outputTokens = (into.outputTokens ?? 0) + (summed.outputTokens ?? 0);
into.cacheReadTokens = (into.cacheReadTokens ?? 0) + (summed.cacheReadTokens ?? 0);
into.cacheWriteTokens = (into.cacheWriteTokens ?? 0) + (summed.cacheWriteTokens ?? 0);
into.reasoningTokens = (into.reasoningTokens ?? 0) + (summed.reasoningTokens ?? 0);
}
/** Fold every usage channel into the pill's payload. Pure: same input,
* same output — no clock, no fs, no store reads. */
export function graphCostSummary(input: CostSummaryInput): GraphCostSummary {
const extraction = emptyStage();
const audit = emptyStage();
const naming = emptyStage();
const runs: RecentRun[] = [];
const foldExtractionRun = (run: ExtractionRunUsage, title: string) => {
const stages = run.stages ?? { extractor: { model: "unknown", steps: [] }, auditor: { model: "unknown", steps: [] } };
// Per-stage: tokens always count; dollars only when priced.
extraction.runs++;
mergeTokens(extraction.tokens, stages.extractor.steps);
const extDollars = stageDollars(stages.extractor.model, stages.extractor.steps);
if (extDollars === null) extraction.unpricedRuns++;
else extraction.dollars += extDollars;
audit.runs++;
mergeTokens(audit.tokens, stages.auditor.steps);
const audDollars = stageDollars(stages.auditor.model, stages.auditor.steps);
if (audDollars === null) audit.unpricedRuns++;
else audit.dollars += audDollars;
// Run-level: null only when BOTH stages are unpriced.
const priced = extDollars ?? 0;
const runDollars = extDollars === null && audDollars === null ? null : priced + (audDollars ?? 0);
const allSteps = [...stages.extractor.steps, ...stages.auditor.steps];
runs.push({
startedAt: run.startedAt,
at: run.at,
durationMs: run.durationMs,
kind: "extraction",
title,
dollars: runDollars,
tokens: sumSteps(allSteps),
...(run.stats ? { merged: run.stats } : {}),
});
};
for (const chat of input.chats) {
for (const run of chat.extractions) foldExtractionRun(run, chat.title);
}
for (const source of input.sources) {
for (const run of source.extractions) foldExtractionRun(run, source.filePath);
}
for (const run of input.naming) {
naming.runs++;
mergeTokens(naming.tokens, run.steps);
const dollars = stageDollars(run.model, run.steps);
if (dollars === null) naming.unpricedRuns++;
else naming.dollars += dollars;
runs.push({
startedAt: run.startedAt,
at: run.at,
durationMs: run.durationMs,
kind: "naming",
title: "Cluster naming",
dollars,
tokens: sumSteps(run.steps),
labels: run.stats.appended,
});
}
runs.sort((a, b) => b.startedAt - a.startedAt);
const capped = runs.slice(0, input.recentCap ?? 10);
// A run is "unpriced" when it billed nothing countable at all.
const unpricedRuns = runs.filter((r) => r.dollars === null).length;
return {
totalDollars: extraction.dollars + audit.dollars + naming.dollars,
unpricedRuns,
stages: { extraction, audit, naming },
runs: capped,
};
}
+1
View File
@@ -34,6 +34,7 @@ export * from "./math-delims.ts";
export * from "./ui-message.ts";
export * from "./tool-parts.ts";
export * from "./prices.ts";
export * from "./cost-summary.ts";
export * from "./fuzzy.ts";
// settings.ts + settings-resolution.ts (settings-storage wave, 2026-09-19):
// the user-settings record's shape/sanitizer (pure) and its fully-resolved
+6
View File
@@ -28,6 +28,7 @@ import type {
TurnUsage,
NotesSource,
ClusterLabeling,
NamingRunUsage,
EmbeddingsLedger,
} from "./contracts.ts";
import type { UserSettings } from "./settings.ts";
@@ -101,6 +102,11 @@ export interface GraphStore {
appendClusterLabel(labeling: ClusterLabeling): void;
readClusterLabels(): ClusterLabeling[];
/** Append one cluster-naming run's usage (graph-global money channel) —
* idempotent per startedAt. */
appendNamingUsage(entry: NamingRunUsage): void;
readNamingRuns(): NamingRunUsage[];
/** Upsert one imported-notes source record (config.SOURCES_FILE — a
* JSON map, not a ledger; an upsert keeps the existing usage log). */
upsertSource(source: NotesSource): void;
+24
View File
@@ -1,4 +1,6 @@
import type { GraphSnapshot } from "../core/contracts.ts";
import type { GraphCostSummary } from "../core/cost-summary.ts";
import { graphCostSummary } from "../core/cost-summary.ts";
import { graphSnapshot } from "../core/graph.ts";
import { stores } from "./stores.ts";
@@ -11,3 +13,25 @@ import { stores } from "./stores.ts";
export function graphView(): GraphSnapshot {
return graphSnapshot(stores);
}
/** The graph view's cost surface (GET /api/graph/costs — the 2026-09 cost
* pill): fold every usage channel into stage totals + recent runs. A pure
* read over the chat records (chat-front-door runs), the notes sources
* (notes-front-door runs), and the naming ledger (graph-global runs);
* dollars are inferred by the pure core fold (core/cost-summary.ts +
* prices.ts) — nothing here stores or caches money. */
export function graphCosts(): GraphCostSummary {
const chats = stores.chatStore
.listChats()
.flatMap((summary) => {
const record = stores.chatStore.getChat(summary.id);
return record?.extractions?.length
? [{ title: record.title ?? summary.id, extractions: record.extractions }]
: [];
});
const sources = stores.graphStore
.readSources()
.filter((s) => s.extractions?.length)
.map((s) => ({ filePath: s.filePath, extractions: s.extractions! }));
return graphCostSummary({ chats, sources, naming: stores.graphStore.readNamingRuns() });
}
+26
View File
@@ -44,6 +44,7 @@ import type {
ProvenancePoint,
CoverageEntry,
SessionCheckpoint,
NamingRunUsage,
} from "../core/contracts.ts";
import type { GraphStore } from "../core/ports.ts";
@@ -90,6 +91,19 @@ function readValid<T>(file: string, isValid: (v: unknown) => v is T): T[] {
const isProvenancePoint = (v: unknown): v is ProvenancePoint =>
isRecord(v) && isString(v.chatId) && isString(v.messageId);
const isNamingStats = (v: unknown): v is NamingRunUsage["stats"] =>
isRecord(v) &&
["detected", "claimed", "named", "appended"].every((k) => typeof v[k] === "number");
const isNamingRun = (v: unknown): v is NamingRunUsage =>
isRecord(v) &&
typeof v.startedAt === "number" &&
isString(v.at) &&
typeof v.durationMs === "number" &&
isString(v.model) &&
Array.isArray(v.steps) &&
isNamingStats(v.stats);
const isClusterLabeling = (v: unknown): v is ClusterLabeling =>
isRecord(v) &&
isString(v.id) &&
@@ -198,6 +212,18 @@ export const graphStore: GraphStore = {
return readValid(LEDGER_FILES.clusterLabels, isClusterLabeling);
},
// ── cluster-naming run usage (config.LEDGER_FILES.namingRuns) ──
/** Idempotent per startedAt — the same discipline as every usage
* channel (chats.appendExtractionUsage, appendSourceUsage). */
appendNamingUsage(entry: NamingRunUsage): void {
if (this.readNamingRuns().some((r) => r.startedAt === entry.startedAt)) return;
appendLine(LEDGER_FILES.namingRuns, entry);
},
readNamingRuns(): NamingRunUsage[] {
return readValid(LEDGER_FILES.namingRuns, isNamingRun);
},
// ── imported-notes bookkeeping (config.SOURCES_FILE — a JSON map) ──
/** Upsert: keeps the existing record's usage log when re-importing. */