sorokin.oleg.edu/learnharness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

Graph pipeline cost metrics: capture naming spend, fold all channels, cost pill + panel

Browse Source 
- Cluster namer bills the money channel now: per-step usage + model
  captured via onStep, persisted as NamingRunUsage records to
  data/graph/naming-runs.jsonl (idempotent per startedAt; failed/aborted
  runs persist too — money spent either way). callUsage promoted to
  extraction.ts module scope for reuse.
- core/cost-summary.ts: the pure fold behind the pill — chat runs,
  notes-source runs, and naming runs into stage totals (extraction /
  audit / naming) + newest-first recent runs. Dollars INFERRED read-time
  via prices.ts, never stored; unpriced models excluded-but-counted
  (render '—', never a silent zero); tokens always sum.
- GET /api/graph/costs (src/server graphCosts composition, thin route).
- Graph view: bottom-left cost pill (cumulative dollars headline) +
  expandable breakdown panel (stage rows with tokens secondary, recent
  runs with time/title/merged/cost); fetched on mount, refetched when a
  pipeline run finishes; merged into one bottom-left stack with the
  running-pipeline pills.
- Tests: pure-fold suite, naming capture wiring (full importNotes ->
  naming path), store idempotency + strict reads.
This commit is contained in:
amosblomqvist
2026-09-19 20:55:35 +03:00
parent c32915b232
commit 3e263a18cc
14 changed files with 666 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agent/subagents/cluster-namer.ts
+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();
src/agent/subagents/extraction.ts
+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;