summaryrefslogtreecommitdiffhomepage
path: root/packages/session-orchestrator/src
diff options
context:
space:
mode:
authorAdam Malczewski <[email protected]>2026-06-10 08:29:59 +0900
committerAdam Malczewski <[email protected]>2026-06-10 08:29:59 +0900
commit6db12ff70acb3333d05a5020ab66da4172a5225a (patch)
treede5cc6314a3a6dd966d7c4fdb9b20adb04ae8307 /packages/session-orchestrator/src
parent4248cd1d546a4c1fb4e68940c11b5e309c2c2736 (diff)
downloaddispatch-6db12ff70acb3333d05a5020ab66da4172a5225a.tar.gz
dispatch-6db12ff70acb3333d05a5020ab66da4172a5225a.zip
feat(metrics): durable per-turn/step token+timing metrics (observability spans + persisted replay)
Two-part token-data improvement: #2 Observability spans (kernel run-turn): turn & step span-close now stamp ALL four Usage fields — added usage.cacheReadTokens/cacheWriteTokens (were silently dropped) and normalized usage_* -> usage.* to match the provider.request span (consistent D9 GROUP BY). No contract change. #3 Persisted replay metrics (conversation-store + read endpoint): new StepMetrics/TurnMetrics wire types; conversation-store persists per-turn metrics in a separate key space (appendMetrics/loadMetrics, turn-append order); session-orchestrator accumulates per-step+turn metrics from the event stream (pure metrics.ts) and persists after seal; transport-http serves GET /conversations/:id/metrics -> ConversationMetricsResponse. Contracts: @dispatch/wire + @dispatch/transport-contract bumped 0.3.0->0.4.0 (additive). GLOSSARY: turn metrics / step metrics. typecheck EXIT 0, biome clean, 546 vitest + 89 bun = 635 tests.
Diffstat (limited to 'packages/session-orchestrator/src')
-rw-r--r--packages/session-orchestrator/src/metrics.test.ts264
-rw-r--r--packages/session-orchestrator/src/metrics.ts124
-rw-r--r--packages/session-orchestrator/src/orchestrator.test.ts319
-rw-r--r--packages/session-orchestrator/src/orchestrator.ts12
4 files changed, 717 insertions, 2 deletions
diff --git a/packages/session-orchestrator/src/metrics.test.ts b/packages/session-orchestrator/src/metrics.test.ts
new file mode 100644
index 0000000..c123dba
--- /dev/null
+++ b/packages/session-orchestrator/src/metrics.test.ts
@@ -0,0 +1,264 @@
+import type { AgentEvent, StepId } from "@dispatch/kernel";
+import { describe, expect, it } from "vitest";
+import { createMetricsAccumulator } from "./metrics.js";
+
+function stepId(id: string): StepId {
+ return id as StepId;
+}
+
+describe("createMetricsAccumulator", () => {
+ it("builds a TurnMetrics from a single-step turn", () => {
+ const acc = createMetricsAccumulator();
+
+ const usageEvent: AgentEvent = {
+ type: "usage",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#0"),
+ usage: { inputTokens: 10, outputTokens: 5 },
+ };
+ const stepCompleteEvent: AgentEvent = {
+ type: "step-complete",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#0"),
+ ttftMs: 50,
+ decodeMs: 150,
+ genTotalMs: 200,
+ };
+ const doneEvent: AgentEvent = {
+ type: "done",
+ conversationId: "c1",
+ turnId: "t1",
+ reason: "stop",
+ durationMs: 500,
+ usage: { inputTokens: 10, outputTokens: 5 },
+ };
+
+ acc.ingest(usageEvent);
+ acc.ingest(stepCompleteEvent);
+ acc.ingest(doneEvent);
+
+ const tm = acc.build("t1");
+ expect(tm.turnId).toBe("t1");
+ expect(tm.usage.inputTokens).toBe(10);
+ expect(tm.usage.outputTokens).toBe(5);
+ expect(tm.durationMs).toBe(500);
+ expect(tm.steps).toHaveLength(1);
+ expect(tm.steps[0]?.stepId).toBe(stepId("t1#0"));
+ expect(tm.steps[0]?.usage.inputTokens).toBe(10);
+ expect(tm.steps[0]?.usage.outputTokens).toBe(5);
+ expect(tm.steps[0]?.ttftMs).toBe(50);
+ expect(tm.steps[0]?.decodeMs).toBe(150);
+ expect(tm.steps[0]?.genTotalMs).toBe(200);
+ });
+
+ it("aggregates multi-step usage and preserves step order", () => {
+ const acc = createMetricsAccumulator();
+
+ acc.ingest({
+ type: "usage",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#0"),
+ usage: { inputTokens: 10, outputTokens: 5 },
+ });
+ acc.ingest({
+ type: "step-complete",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#0"),
+ genTotalMs: 100,
+ });
+ acc.ingest({
+ type: "usage",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#1"),
+ usage: { inputTokens: 20, outputTokens: 10 },
+ });
+ acc.ingest({
+ type: "step-complete",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#1"),
+ genTotalMs: 150,
+ });
+ acc.ingest({
+ type: "done",
+ conversationId: "c1",
+ turnId: "t1",
+ reason: "stop",
+ usage: { inputTokens: 30, outputTokens: 15 },
+ });
+
+ const tm = acc.build("t1");
+ expect(tm.steps).toHaveLength(2);
+ expect(tm.steps[0]?.stepId).toBe(stepId("t1#0"));
+ expect(tm.steps[0]?.usage.inputTokens).toBe(10);
+ expect(tm.steps[1]?.stepId).toBe(stepId("t1#1"));
+ expect(tm.steps[1]?.usage.inputTokens).toBe(20);
+ expect(tm.usage.inputTokens).toBe(30);
+ expect(tm.usage.outputTokens).toBe(15);
+ });
+
+ it("joins per-step timing and usage by stepId", () => {
+ const acc = createMetricsAccumulator();
+
+ acc.ingest({
+ type: "step-complete",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#0"),
+ ttftMs: 50,
+ decodeMs: 100,
+ genTotalMs: 150,
+ });
+ acc.ingest({
+ type: "usage",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#0"),
+ usage: { inputTokens: 10, outputTokens: 5 },
+ });
+
+ const tm = acc.build("t1");
+ expect(tm.steps).toHaveLength(1);
+ expect(tm.steps[0]?.usage.inputTokens).toBe(10);
+ expect(tm.steps[0]?.ttftMs).toBe(50);
+ expect(tm.steps[0]?.decodeMs).toBe(100);
+ expect(tm.steps[0]?.genTotalMs).toBe(150);
+ });
+
+ it("turn-level usage comes from done event, not step sum", () => {
+ const acc = createMetricsAccumulator();
+
+ acc.ingest({
+ type: "usage",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#0"),
+ usage: { inputTokens: 10, outputTokens: 5 },
+ });
+ acc.ingest({
+ type: "step-complete",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#0"),
+ });
+ acc.ingest({
+ type: "done",
+ conversationId: "c1",
+ turnId: "t1",
+ reason: "stop",
+ usage: { inputTokens: 100, outputTokens: 50 },
+ });
+
+ const tm = acc.build("t1");
+ expect(tm.usage.inputTokens).toBe(100);
+ expect(tm.usage.outputTokens).toBe(50);
+ });
+
+ it("falls back to summing step usage when done.usage is absent", () => {
+ const acc = createMetricsAccumulator();
+
+ acc.ingest({
+ type: "usage",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#0"),
+ usage: { inputTokens: 10, outputTokens: 5 },
+ });
+ acc.ingest({
+ type: "step-complete",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#0"),
+ });
+ acc.ingest({
+ type: "done",
+ conversationId: "c1",
+ turnId: "t1",
+ reason: "stop",
+ });
+
+ const tm = acc.build("t1");
+ expect(tm.usage.inputTokens).toBe(10);
+ expect(tm.usage.outputTokens).toBe(5);
+ });
+
+ it("tolerates missing usage on a step (timing only)", () => {
+ const acc = createMetricsAccumulator();
+
+ acc.ingest({
+ type: "step-complete",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#0"),
+ genTotalMs: 200,
+ });
+ acc.ingest({
+ type: "done",
+ conversationId: "c1",
+ turnId: "t1",
+ reason: "stop",
+ });
+
+ const tm = acc.build("t1");
+ expect(tm.steps).toHaveLength(1);
+ expect(tm.steps[0]?.usage.inputTokens).toBe(0);
+ expect(tm.steps[0]?.usage.outputTokens).toBe(0);
+ expect(tm.steps[0]?.genTotalMs).toBe(200);
+ });
+
+ it("tolerates missing timing on a step (usage only)", () => {
+ const acc = createMetricsAccumulator();
+
+ acc.ingest({
+ type: "usage",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#0"),
+ usage: { inputTokens: 10, outputTokens: 5 },
+ });
+ acc.ingest({
+ type: "done",
+ conversationId: "c1",
+ turnId: "t1",
+ reason: "stop",
+ });
+
+ const tm = acc.build("t1");
+ expect(tm.steps).toHaveLength(1);
+ expect(tm.steps[0]?.usage.inputTokens).toBe(10);
+ expect(tm.steps[0]?.ttftMs).toBeUndefined();
+ expect(tm.steps[0]?.decodeMs).toBeUndefined();
+ expect(tm.steps[0]?.genTotalMs).toBeUndefined();
+ });
+
+ it("reset clears all accumulated state", () => {
+ const acc = createMetricsAccumulator();
+
+ acc.ingest({
+ type: "usage",
+ conversationId: "c1",
+ turnId: "t1",
+ stepId: stepId("t1#0"),
+ usage: { inputTokens: 10, outputTokens: 5 },
+ });
+ acc.ingest({
+ type: "done",
+ conversationId: "c1",
+ turnId: "t1",
+ reason: "stop",
+ usage: { inputTokens: 10, outputTokens: 5 },
+ });
+
+ acc.reset();
+
+ const tm = acc.build("t2");
+ expect(tm.steps).toHaveLength(0);
+ expect(tm.usage.inputTokens).toBe(0);
+ expect(tm.usage.outputTokens).toBe(0);
+ });
+});
diff --git a/packages/session-orchestrator/src/metrics.ts b/packages/session-orchestrator/src/metrics.ts
new file mode 100644
index 0000000..e953bd9
--- /dev/null
+++ b/packages/session-orchestrator/src/metrics.ts
@@ -0,0 +1,124 @@
+import type {
+ AgentEvent,
+ StepId,
+ StepMetrics,
+ TurnDoneEvent,
+ TurnMetrics,
+ TurnStepCompleteEvent,
+ TurnUsageEvent,
+ Usage,
+} from "@dispatch/kernel";
+
+const zeroUsage: Usage = { inputTokens: 0, outputTokens: 0 };
+
+interface StepAccumulator {
+ readonly stepId: StepId;
+ usage: Usage | undefined;
+ ttftMs: number | undefined;
+ decodeMs: number | undefined;
+ genTotalMs: number | undefined;
+}
+
+export interface MetricsAccumulator {
+ readonly ingest: (event: AgentEvent) => void;
+ readonly build: (turnId: string) => TurnMetrics;
+ readonly reset: () => void;
+}
+
+export function createMetricsAccumulator(): MetricsAccumulator {
+ const steps = new Map<StepId, StepAccumulator>();
+ const stepOrder: StepId[] = [];
+ let doneUsage: Usage | undefined;
+ let doneDurationMs: number | undefined;
+
+ function getOrCreateStep(stepId: StepId): StepAccumulator {
+ let acc = steps.get(stepId);
+ if (acc === undefined) {
+ acc = {
+ stepId,
+ usage: undefined,
+ ttftMs: undefined,
+ decodeMs: undefined,
+ genTotalMs: undefined,
+ };
+ steps.set(stepId, acc);
+ stepOrder.push(stepId);
+ }
+ return acc;
+ }
+
+ function ingest(event: AgentEvent): void {
+ switch (event.type) {
+ case "usage": {
+ const e = event as TurnUsageEvent;
+ if (e.stepId !== undefined) {
+ const acc = getOrCreateStep(e.stepId);
+ acc.usage = e.usage;
+ }
+ break;
+ }
+ case "step-complete": {
+ const e = event as TurnStepCompleteEvent;
+ const acc = getOrCreateStep(e.stepId);
+ acc.ttftMs = e.ttftMs;
+ acc.decodeMs = e.decodeMs;
+ acc.genTotalMs = e.genTotalMs;
+ break;
+ }
+ case "done": {
+ const e = event as TurnDoneEvent;
+ doneUsage = e.usage;
+ doneDurationMs = e.durationMs;
+ break;
+ }
+ }
+ }
+
+ function build(turnId: string): TurnMetrics {
+ const stepMetrics: StepMetrics[] = stepOrder.map((stepId) => {
+ const acc = steps.get(stepId);
+ if (acc === undefined) {
+ return { stepId, usage: zeroUsage };
+ }
+ const usage = acc.usage ?? zeroUsage;
+ const sm: StepMetrics = { stepId, usage };
+ if (acc.ttftMs !== undefined) {
+ (sm as { ttftMs?: number }).ttftMs = acc.ttftMs;
+ }
+ if (acc.decodeMs !== undefined) {
+ (sm as { decodeMs?: number }).decodeMs = acc.decodeMs;
+ }
+ if (acc.genTotalMs !== undefined) {
+ (sm as { genTotalMs?: number }).genTotalMs = acc.genTotalMs;
+ }
+ return sm;
+ });
+
+ const aggregateUsage = doneUsage ?? sumStepUsage(stepMetrics);
+
+ const tm: TurnMetrics = { turnId, usage: aggregateUsage, steps: stepMetrics };
+ if (doneDurationMs !== undefined) {
+ (tm as { durationMs?: number }).durationMs = doneDurationMs;
+ }
+ return tm;
+ }
+
+ function reset(): void {
+ steps.clear();
+ stepOrder.length = 0;
+ doneUsage = undefined;
+ doneDurationMs = undefined;
+ }
+
+ return { ingest, build, reset };
+}
+
+function sumStepUsage(steps: readonly StepMetrics[]): Usage {
+ let inputTokens = 0;
+ let outputTokens = 0;
+ for (const s of steps) {
+ inputTokens += s.usage.inputTokens;
+ outputTokens += s.usage.outputTokens;
+ }
+ return { inputTokens, outputTokens };
+}
diff --git a/packages/session-orchestrator/src/orchestrator.test.ts b/packages/session-orchestrator/src/orchestrator.test.ts
index 3954ffe..ea564c5 100644
--- a/packages/session-orchestrator/src/orchestrator.test.ts
+++ b/packages/session-orchestrator/src/orchestrator.test.ts
@@ -7,6 +7,8 @@ import type {
RunTurnInput,
RunTurnResult,
StoredChunk,
+ ToolContract,
+ TurnMetrics,
} from "@dispatch/kernel";
import { runTurn } from "@dispatch/kernel";
import { describe, expect, it } from "vitest";
@@ -14,10 +16,13 @@ import { createSessionOrchestrator } from "./orchestrator.js";
function createInMemoryStore(): ConversationStore & {
readonly data: Map<string, ChatMessage[]>;
+ readonly metricsData: Map<string, TurnMetrics[]>;
} {
const data = new Map<string, ChatMessage[]>();
+ const metricsData = new Map<string, TurnMetrics[]>();
return {
data,
+ metricsData,
async append(conversationId, messages) {
const existing = data.get(conversationId) ?? [];
data.set(conversationId, [...existing, ...messages]);
@@ -39,6 +44,13 @@ function createInMemoryStore(): ConversationStore & {
}
return result;
},
+ async appendMetrics(conversationId, metrics) {
+ const existing = metricsData.get(conversationId) ?? [];
+ metricsData.set(conversationId, [...existing, metrics]);
+ },
+ async loadMetrics(conversationId) {
+ return [...(metricsData.get(conversationId) ?? [])];
+ },
};
}
@@ -63,6 +75,18 @@ function collectEvents(): { events: AgentEvent[]; onEvent: (event: AgentEvent) =
return { events, onEvent: (event) => events.push(event) };
}
+function createFakeTool(
+ name: string,
+ handler: (input: unknown) => Promise<{ content: string }>,
+): ToolContract {
+ return {
+ name,
+ description: `Fake tool: ${name}`,
+ parameters: { type: "object" },
+ execute: async (input) => handler(input),
+ };
+}
+
describe("handleMessage integration", () => {
it("loads history, runs turn, emits events, and persists result", async () => {
const store = createInMemoryStore();
@@ -465,6 +489,13 @@ describe("turn-sealed event", () => {
async loadSince(conversationId, sinceSeq) {
return store.loadSince(conversationId, sinceSeq);
},
+ async appendMetrics(conversationId, metrics) {
+ await store.appendMetrics(conversationId, metrics);
+ ordering.push("appendMetrics");
+ },
+ async loadMetrics(conversationId) {
+ return store.loadMetrics(conversationId);
+ },
};
const orchestrator = createSessionOrchestrator({
@@ -484,7 +515,7 @@ describe("turn-sealed event", () => {
},
});
- expect(ordering).toEqual(["append", "turn-sealed"]);
+ expect(ordering).toEqual(["append", "appendMetrics", "turn-sealed"]);
});
it("does not emit turn-sealed when append throws", async () => {
@@ -505,6 +536,12 @@ describe("turn-sealed event", () => {
async loadSince() {
return [];
},
+ async appendMetrics() {
+ return undefined;
+ },
+ async loadMetrics() {
+ return [];
+ },
};
const orchestrator = createSessionOrchestrator({
@@ -528,3 +565,283 @@ describe("turn-sealed event", () => {
expect(sealedEvents).toHaveLength(0);
});
});
+
+describe("turn metrics persistence", () => {
+ it("persists a TurnMetrics after a single-step turn seals", async () => {
+ const store = createInMemoryStore();
+ const provider = createFakeProvider([
+ [
+ { type: "text-delta", delta: "Hello" },
+ { type: "usage", usage: { inputTokens: 10, outputTokens: 5 } },
+ { type: "finish", reason: "stop" },
+ ],
+ ]);
+
+ const orchestrator = createSessionOrchestrator({
+ conversationStore: store,
+ resolveProvider: () => provider,
+ resolveTools: () => [],
+ runTurn,
+ now: () => 1000,
+ });
+
+ await orchestrator.handleMessage({
+ conversationId: "conv-metrics-1",
+ text: "test",
+ onEvent: () => {},
+ });
+
+ const metrics = store.metricsData.get("conv-metrics-1");
+ expect(metrics).toBeDefined();
+ expect(metrics).toHaveLength(1);
+ expect(metrics?.[0]?.turnId).toMatch(/^turn-/);
+ expect(metrics?.[0]?.usage.inputTokens).toBe(10);
+ expect(metrics?.[0]?.usage.outputTokens).toBe(5);
+ expect(metrics?.[0]?.steps).toHaveLength(1);
+ expect(metrics?.[0]?.steps[0]?.usage.inputTokens).toBe(10);
+ expect(metrics?.[0]?.steps[0]?.usage.outputTokens).toBe(5);
+ });
+
+ it("TurnMetrics aggregates multi-step usage and carries each step's StepMetrics in order", async () => {
+ const store = createInMemoryStore();
+ const tool = createFakeTool("echo", async () => ({ content: "echoed" }));
+
+ let callIndex = 0;
+ const provider: ProviderContract = {
+ id: "fake",
+ stream() {
+ const idx = callIndex++;
+ return (async function* () {
+ if (idx === 0) {
+ yield {
+ type: "tool-call",
+ toolCallId: "tc1",
+ toolName: "echo",
+ input: {},
+ } as ProviderEvent;
+ yield {
+ type: "usage",
+ usage: { inputTokens: 10, outputTokens: 5 },
+ } as ProviderEvent;
+ yield { type: "finish", reason: "tool-calls" } as ProviderEvent;
+ } else {
+ yield { type: "text-delta", delta: "Step2" } as ProviderEvent;
+ yield {
+ type: "usage",
+ usage: { inputTokens: 20, outputTokens: 10 },
+ } as ProviderEvent;
+ yield { type: "finish", reason: "stop" } as ProviderEvent;
+ }
+ })();
+ },
+ };
+
+ const orchestrator = createSessionOrchestrator({
+ conversationStore: store,
+ resolveProvider: () => provider,
+ resolveTools: () => [tool],
+ runTurn,
+ now: () => 1000,
+ });
+
+ await orchestrator.handleMessage({
+ conversationId: "conv-metrics-multi",
+ text: "test",
+ onEvent: () => {},
+ });
+
+ const metrics = store.metricsData.get("conv-metrics-multi");
+ expect(metrics).toBeDefined();
+ expect(metrics).toHaveLength(1);
+
+ const tm = metrics?.[0];
+ if (tm === undefined) throw new Error("expected metrics");
+
+ expect(tm.steps.length).toBeGreaterThanOrEqual(2);
+
+ expect(tm.steps[0]?.usage.inputTokens).toBe(10);
+ expect(tm.steps[0]?.usage.outputTokens).toBe(5);
+ expect(tm.steps[1]?.usage.inputTokens).toBe(20);
+ expect(tm.steps[1]?.usage.outputTokens).toBe(10);
+
+ expect(tm.usage.inputTokens).toBe(30);
+ expect(tm.usage.outputTokens).toBe(15);
+ });
+
+ it("per-step timing and usage are joined by stepId into one StepMetrics", async () => {
+ const store = createInMemoryStore();
+ const clock = createCounterNow();
+ clock.tick(100);
+
+ let callIndex = 0;
+ const provider: ProviderContract = {
+ id: "fake",
+ stream() {
+ const idx = callIndex++;
+ return (async function* () {
+ if (idx === 0) {
+ clock.tick(50);
+ yield { type: "text-delta", delta: "Hello" } as ProviderEvent;
+ clock.tick(100);
+ yield {
+ type: "usage",
+ usage: { inputTokens: 10, outputTokens: 5 },
+ } as ProviderEvent;
+ clock.tick(50);
+ yield { type: "finish", reason: "stop" } as ProviderEvent;
+ }
+ })();
+ },
+ };
+
+ const orchestrator = createSessionOrchestrator({
+ conversationStore: store,
+ resolveProvider: () => provider,
+ resolveTools: () => [],
+ runTurn,
+ now: clock.now,
+ });
+
+ await orchestrator.handleMessage({
+ conversationId: "conv-metrics-join",
+ text: "test",
+ onEvent: () => {},
+ });
+
+ const metrics = store.metricsData.get("conv-metrics-join");
+ expect(metrics).toBeDefined();
+ expect(metrics).toHaveLength(1);
+
+ const tm = metrics?.[0];
+ if (tm === undefined) throw new Error("expected metrics");
+
+ expect(tm.steps).toHaveLength(1);
+ const step = tm.steps[0];
+ if (step === undefined) throw new Error("expected step");
+
+ expect(step.usage.inputTokens).toBe(10);
+ expect(step.usage.outputTokens).toBe(5);
+ expect(step.genTotalMs).toBe(200);
+ expect(step.ttftMs).toBe(50);
+ expect(step.decodeMs).toBe(150);
+ });
+
+ it("turn-level usage comes from the done event aggregate", async () => {
+ const store = createInMemoryStore();
+ const tool = createFakeTool("echo", async () => ({ content: "echoed" }));
+
+ let callIndex = 0;
+ const provider: ProviderContract = {
+ id: "fake",
+ stream() {
+ const idx = callIndex++;
+ return (async function* () {
+ if (idx === 0) {
+ yield {
+ type: "tool-call",
+ toolCallId: "tc1",
+ toolName: "echo",
+ input: {},
+ } as ProviderEvent;
+ yield {
+ type: "usage",
+ usage: { inputTokens: 10, outputTokens: 5 },
+ } as ProviderEvent;
+ yield { type: "finish", reason: "tool-calls" } as ProviderEvent;
+ } else {
+ yield { type: "text-delta", delta: "Step2" } as ProviderEvent;
+ yield {
+ type: "usage",
+ usage: { inputTokens: 20, outputTokens: 10 },
+ } as ProviderEvent;
+ yield { type: "finish", reason: "stop" } as ProviderEvent;
+ }
+ })();
+ },
+ };
+
+ const orchestrator = createSessionOrchestrator({
+ conversationStore: store,
+ resolveProvider: () => provider,
+ resolveTools: () => [tool],
+ runTurn,
+ now: () => 1000,
+ });
+
+ await orchestrator.handleMessage({
+ conversationId: "conv-metrics-done",
+ text: "test",
+ onEvent: () => {},
+ });
+
+ const metrics = store.metricsData.get("conv-metrics-done");
+ expect(metrics).toBeDefined();
+ expect(metrics).toHaveLength(1);
+
+ const tm = metrics?.[0];
+ if (tm === undefined) throw new Error("expected metrics");
+
+ expect(tm.usage.inputTokens).toBe(30);
+ expect(tm.usage.outputTokens).toBe(15);
+ });
+
+ it("does not persist metrics nor emit turn-sealed when chunk append fails", async () => {
+ const provider = createFakeProvider([
+ [
+ { type: "text-delta", delta: "ok" },
+ { type: "usage", usage: { inputTokens: 5, outputTokens: 3 } },
+ { type: "finish", reason: "stop" },
+ ],
+ ]);
+
+ let metricsAppended = false;
+ const failingMetricsStore: ConversationStore = {
+ async append() {
+ throw new Error("storage failure");
+ },
+ async load() {
+ return [];
+ },
+ async loadSince() {
+ return [];
+ },
+ async appendMetrics() {
+ metricsAppended = true;
+ },
+ async loadMetrics() {
+ return [];
+ },
+ };
+
+ const orchestrator = createSessionOrchestrator({
+ conversationStore: failingMetricsStore,
+ resolveProvider: () => provider,
+ resolveTools: () => [],
+ runTurn,
+ });
+
+ const { events, onEvent } = collectEvents();
+
+ await expect(
+ orchestrator.handleMessage({
+ conversationId: "conv-fail-metrics",
+ text: "test",
+ onEvent,
+ }),
+ ).rejects.toThrow("storage failure");
+
+ const sealedEvents = events.filter((e) => e.type === "turn-sealed");
+ expect(sealedEvents).toHaveLength(0);
+ expect(metricsAppended).toBe(false);
+ });
+});
+
+function createCounterNow(): { now: () => number; tick: (ms: number) => void } {
+ let t = 0;
+ return {
+ now: () => t,
+ tick(ms: number) {
+ t += ms;
+ },
+ };
+}
diff --git a/packages/session-orchestrator/src/orchestrator.ts b/packages/session-orchestrator/src/orchestrator.ts
index 04f6ad2..d84b805 100644
--- a/packages/session-orchestrator/src/orchestrator.ts
+++ b/packages/session-orchestrator/src/orchestrator.ts
@@ -11,6 +11,7 @@ import type {
ToolDispatchPolicy,
} from "@dispatch/kernel";
import { defineService } from "@dispatch/kernel";
+import { createMetricsAccumulator } from "./metrics.js";
import { buildUserMessage, defaultDispatchPolicy, generateTurnId } from "./pure.js";
export interface SessionOrchestrator {
@@ -73,13 +74,19 @@ export function createSessionOrchestrator(deps: SessionOrchestratorDeps): Sessio
const tools = deps.resolveTools();
const dispatch = deps.resolveDispatch?.() ?? defaultDispatchPolicy();
const turnLogger = deps.logger?.child({ conversationId, turnId });
+ const metrics = createMetricsAccumulator();
+
+ const emitAndAccumulate = (event: AgentEvent): void => {
+ metrics.ingest(event);
+ onEvent(event);
+ };
const opts: RunTurnInput = {
provider,
messages: [...history, userMsg],
tools,
dispatch,
- emit: onEvent,
+ emit: emitAndAccumulate,
conversationId,
turnId,
...(modelOverride !== undefined
@@ -96,6 +103,9 @@ export function createSessionOrchestrator(deps: SessionOrchestratorDeps): Sessio
const toPersist: ChatMessage[] = [userMsg, ...result.messages];
await deps.conversationStore.append(conversationId, toPersist);
+ const turnMetrics = metrics.build(turnId);
+ await deps.conversationStore.appendMetrics(conversationId, turnMetrics);
+
onEvent({ type: "turn-sealed", conversationId, turnId });
},
};