fix(observability): harden phase usage exporter plumbing

j5ik2o · j5ik2o · commit b6a55324378f · 2026-06-06T17:30:02.000+09:00
diff --git a/src/__tests__/logging-contracts.test.ts b/src/__tests__/logging-contracts.test.ts
@@ -72,6 +72,8 @@ describe('logging contracts', () => {
     );
 
     expect(record.usage.cached_input_tokens).toBe(4);
+    expect(record.usage.cache_creation_input_tokens).toBe(2);
+    expect(record.usage.cache_read_input_tokens).toBe(2);
   });
 
   it('should reject usage_missing records with unknown reason values', () => {
diff --git a/src/__tests__/otelFoundation.test.ts b/src/__tests__/otelFoundation.test.ts
@@ -339,66 +339,38 @@ describe('otel foundation', () => {
     }
   });
 
-  it('should release the shared SDK when run exporter registration fails', async () => {
+  it('should reject mismatched run-scoped exporter ids before starting the SDK', async () => {
     const foundation = await loadFoundationWithMockedSdk();
     const tempDir = mkdtempSync(join(tmpdir(), 'takt-otel-registration-failure-'));
-    const firstPhaseUsageLogPath = join(tempDir, 'first-usage-events.phase.jsonl');
-    const secondShadowLogPath = join(tempDir, 'second-otel-session-shadow.jsonl');
-    const secondPhaseUsageLogPath = join(tempDir, 'second-usage-events.phase.jsonl');
-    let first: { shutdown(): Promise<void> } | undefined;
+    const shadowLogPath = join(tempDir, 'session-otel-session-shadow.jsonl');
+    const phaseUsageLogPath = join(tempDir, 'session-usage-events.phase.jsonl');
+    const monitorPath = join(tempDir, 'monitor.json');
 
     try {
-      first = await foundation.initializeOtelFoundation(
-        enabledUsageEventsPhaseObservability,
-        {
-          usageEventsExporter: {
-            runId: 'run-1',
-            sessionId: 'session-1',
-            phaseUsageLogPath: firstPhaseUsageLogPath,
-          },
-        },
-      );
-
       await expect(foundation.initializeOtelFoundation(
         enabledAllObservability,
         {
           sessionLogExporter: {
             runId: 'run-2',
-            shadowLogPath: secondShadowLogPath,
-            sanitizedTask: 'second task',
+            shadowLogPath,
+            sanitizedTask: 'task',
             workflowName: 'default',
           },
           usageEventsExporter: {
             runId: 'run-1',
-            sessionId: 'session-duplicate',
-            phaseUsageLogPath: secondPhaseUsageLogPath,
+            sessionId: 'session-1',
+            phaseUsageLogPath,
+          },
+          monitorJsonExporter: {
+            runId: 'run-2',
+            monitorPath,
           },
         },
-      )).rejects.toThrow('Phase usage event exporter is already registered for runId: run-1');
-
-      const sessionLogProcessor = foundation.constructedOptions[0]?.spanProcessors?.[0] as {
-        onStart(span: unknown, parentContext: unknown): void;
-      };
-      sessionLogProcessor.onStart({
-        name: 'step.implement',
-        startTime: [1_778_777_205, 0],
-        attributes: {
-          'takt.run.id': 'run-2',
-          'takt.step.name': 'implement',
-          'takt.step.persona': 'coder',
-          'takt.step.iteration': 1,
-        },
-      }, {});
-      const secondShadowRecords = readFileSync(secondShadowLogPath, 'utf-8').trim().split('\n');
-      expect(secondShadowRecords).toHaveLength(1);
+      )).rejects.toThrow('Run-scoped OpenTelemetry exporters must share the same runId');
 
-      await first.shutdown();
-
-      expect(foundation.shutdownMock).toHaveBeenCalledOnce();
+      expect(foundation.startMock).not.toHaveBeenCalled();
+      expect(foundation.constructedOptions).toEqual([]);
     } finally {
-      if (first) {
-        await first.shutdown();
-      }
       rmSync(tempDir, { recursive: true, force: true });
     }
   });
diff --git a/src/__tests__/usageEventsSpanProcessor.test.ts b/src/__tests__/usageEventsSpanProcessor.test.ts
@@ -63,7 +63,7 @@ describe('UsageEventsSpanProcessor', () => {
     ]);
   });
 
-  it('rejects duplicate runId registrations', () => {
+  it('ignores duplicate runId registrations without redirecting existing output', () => {
     const firstLogPath = createTempLogPath('first-usage-events.phase.jsonl');
     const secondLogPath = createTempLogPath('second-usage-events.phase.jsonl');
     const processor = new UsageEventsSpanProcessor();
@@ -77,7 +77,17 @@ describe('UsageEventsSpanProcessor', () => {
       runId: 'run-1',
       sessionId: 'session-duplicate',
       phaseUsageLogPath: secondLogPath,
-    })).toThrow('Phase usage event exporter is already registered for runId: run-1');
+    })).not.toThrow();
+
+    processor.onEnd(makePhaseSpan('run-1') as unknown as ReadableSpan);
+
+    expect(readRecords(firstLogPath)).toEqual([
+      expect.objectContaining({
+        run_id: 'run-1',
+        session_id: 'session-1',
+        phase: 'phase1_execute',
+      }),
+    ]);
     expect(existsSync(secondLogPath)).toBe(false);
   });
 
diff --git a/src/agents/judge-status-usecase.ts b/src/agents/judge-status-usecase.ts
@@ -16,20 +16,24 @@ export interface JudgeStatusOptions {
   language?: Language;
   interactive?: boolean;
   onStream?: StreamCallback;
-  onJudgeStage?: (entry: {
-    stage: 1 | 2 | 3;
-    method: 'structured_output' | 'phase3_tag' | 'ai_judge';
-    status: 'done' | 'error' | 'skipped';
-    instruction: string;
-    response: string;
-    providerUsage?: ProviderUsageSnapshot;
-  }) => void;
+  onJudgeStage?: (entry: JudgeStageLogEntry) => void;
   onStructuredPromptResolved?: (promptParts: {
     systemPrompt: string;
     userInstruction: string;
   }) => void;
 }
 
+export interface JudgeStageLogEntry {
+  stage: 1 | 2 | 3;
+  method: 'structured_output' | 'phase3_tag' | 'ai_judge';
+  status: 'done' | 'error' | 'skipped';
+  instruction: string;
+  response: string;
+  providerUsage?: ProviderUsageSnapshot;
+}
+
+type JudgeResponseEntry = Pick<JudgeStageLogEntry, 'instruction' | 'status' | 'response' | 'providerUsage'>;
+
 export interface TagJudgeRunOptions {
   cwd: string;
   provider?: ProviderType;
@@ -133,6 +137,28 @@ export async function evaluateCondition(
   return detectJudgeIndex(response.content);
 }
 
+export function createJudgeStageRecorder(): {
+  capture(entry: JudgeResponseEntry): void;
+  stage(entry: Pick<JudgeStageLogEntry, 'stage' | 'method'>): JudgeStageLogEntry;
+} {
+  let latest: JudgeResponseEntry = {
+    status: 'skipped',
+    instruction: '',
+    response: '',
+  };
+  return {
+    capture(entry): void {
+      latest = entry;
+    },
+    stage(entry): JudgeStageLogEntry {
+      return {
+        ...entry,
+        ...latest,
+      };
+    },
+  };
+}
+
 export async function judgeStatus(
   structuredInstruction: string,
   tagInstruction: string,
@@ -207,32 +233,20 @@ export async function judgeStatus(
   const conditions = buildJudgeConditions(rules, interactiveEnabled);
 
   if (conditions.length > 0) {
-    let stage3Status: 'done' | 'error' | 'skipped' = 'skipped';
-    let stage3Instruction = '';
-    let stage3Response = '';
-    let stage3ProviderUsage: ProviderUsageSnapshot | undefined;
+    const stage3 = createJudgeStageRecorder();
     const normalizedConditions = conditions.map((c, pos) => ({ index: pos, text: c.text }));
     const fallbackPosition = await evaluateCondition(structuredInstruction, normalizedConditions, {
       cwd: options.cwd,
       provider: options.provider,
       resolvedProvider: options.resolvedProvider,
       resolvedModel: options.resolvedModel,
-      onJudgeResponse: (entry) => {
-        stage3Status = entry.status;
-        stage3Instruction = entry.instruction;
-        stage3Response = entry.response;
-        stage3ProviderUsage = entry.providerUsage;
-      },
+      onJudgeResponse: stage3.capture,
     });
 
-    options.onJudgeStage?.({
+    options.onJudgeStage?.(stage3.stage({
       stage: 3,
       method: 'ai_judge',
-      status: stage3Status,
-      instruction: stage3Instruction,
-      response: stage3Response,
-      providerUsage: stage3ProviderUsage,
-    });
+    }));
 
     if (fallbackPosition >= 0 && fallbackPosition < conditions.length) {
       const originalIndex = conditions[fallbackPosition]?.index;
diff --git a/src/agents/structured-caller/prompt-based-structured-caller.ts b/src/agents/structured-caller/prompt-based-structured-caller.ts
@@ -1,5 +1,4 @@
 import type { WorkflowRule, PartDefinition } from '../../core/models/types.js';
-import type { ProviderUsageSnapshot } from '../../core/models/response.js';
 import {
   buildPromptBasedDecomposePrompt,
   buildPromptBasedMorePartsPrompt,
@@ -8,6 +7,7 @@ import {
 import { buildJudgePrompt, detectJudgeIndex, isValidRuleIndex, buildJudgeConditions } from '../judge-utils.js';
 import { runAgent } from '../runner.js';
 import {
+  createJudgeStageRecorder,
   runTagJudgeStage,
   type EvaluateConditionOptions,
   type JudgeStatusOptions,
@@ -102,31 +102,19 @@ export class PromptBasedStructuredCaller implements StructuredCaller {
     const conditions = buildJudgeConditions(rules, interactiveEnabled);
 
     if (conditions.length > 0) {
-      let stage3Status: 'done' | 'error' | 'skipped' = 'skipped';
-      let stage3Instruction = '';
-      let stage3Response = '';
-      let stage3ProviderUsage: ProviderUsageSnapshot | undefined;
+      const stage3 = createJudgeStageRecorder();
       const fallbackIndex = await this.evaluateCondition(structuredInstruction, conditions, {
         cwd: options.cwd,
         provider: options.provider,
         resolvedProvider: options.resolvedProvider,
         resolvedModel: options.resolvedModel,
-        onJudgeResponse: (entry) => {
-          stage3Status = entry.status;
-          stage3Instruction = entry.instruction;
-          stage3Response = entry.response;
-          stage3ProviderUsage = entry.providerUsage;
-        },
+        onJudgeResponse: stage3.capture,
       });
 
-      options.onJudgeStage?.({
+      options.onJudgeStage?.(stage3.stage({
         stage: 3,
         method: 'ai_judge',
-        status: stage3Status,
-        instruction: stage3Instruction,
-        response: stage3Response,
-        providerUsage: stage3ProviderUsage,
-      });
+      }));
 
       if (isValidRuleIndex(fallbackIndex, rules, interactiveEnabled)) {
         return { ruleIndex: fallbackIndex, method: 'ai_judge' };
diff --git a/src/core/logging/phaseUsageEvent.ts b/src/core/logging/phaseUsageEvent.ts
@@ -3,6 +3,8 @@ import {
   USAGE_MISSING_REASONS,
   type UsageMissingReason,
 } from './contracts.js';
+import { buildUsageEventPayload } from './providerEvent.js';
+import type { ProviderUsageSnapshot } from '../models/response.js';
 import { isProviderType, type ProviderType } from '../../shared/types/provider.js';
 
 export type PhaseUsageType =
@@ -171,12 +173,19 @@ function buildRecord(
 function extractUsage(attributes: Record<string, unknown>): Pick<PhaseUsageEventLogRecord, 'usage_missing' | 'reason' | 'usage'> & {
   missing: boolean;
 } {
+  const snapshot = usageSnapshotFromAttributes(attributes);
+  const payload = buildUsageEventPayload(snapshot);
+  return {
+    missing: payload.usage_missing,
+    ...payload,
+  };
+}
+
+function usageSnapshotFromAttributes(attributes: Record<string, unknown>): ProviderUsageSnapshot {
   if (attributes['takt.usage.missing'] === true) {
     return {
-      missing: true,
-      usage_missing: true,
+      usageMissing: true,
       reason: getUsageMissingReason(attributes['takt.usage.missing_reason']),
-      usage: {},
     };
   }
 
@@ -187,26 +196,21 @@ function extractUsage(attributes: Record<string, unknown>): Pick<PhaseUsageEvent
 
   if (inputTokens === undefined || outputTokens === undefined || totalTokens === undefined) {
     return {
-      missing: true,
-      usage_missing: true,
+      usageMissing: true,
       reason: hasAnyUsageAttribute(attributes)
         ? USAGE_MISSING_REASONS.TOKENS_MISSING
         : USAGE_MISSING_REASONS.NOT_AVAILABLE,
-      usage: {},
     };
   }
 
   return {
-    missing: false,
-    usage_missing: false,
-    usage: {
-      input_tokens: inputTokens,
-      output_tokens: outputTokens,
-      total_tokens: totalTokens,
-      ...optionalNumber('cached_input_tokens', getNumber(attributes, 'gen_ai.usage.cached_input_tokens')),
-      ...optionalNumber('cache_creation_input_tokens', getNumber(attributes, 'gen_ai.usage.cache_creation_input_tokens')),
-      ...optionalNumber('cache_read_input_tokens', getNumber(attributes, 'gen_ai.usage.cache_read_input_tokens')),
-    },
+    usageMissing: false,
+    inputTokens,
+    outputTokens,
+    totalTokens,
+    cachedInputTokens: getNumber(attributes, 'gen_ai.usage.cached_input_tokens'),
+    cacheCreationInputTokens: getNumber(attributes, 'gen_ai.usage.cache_creation_input_tokens'),
+    cacheReadInputTokens: getNumber(attributes, 'gen_ai.usage.cache_read_input_tokens'),
   };
 }
 
@@ -287,10 +291,6 @@ function getUsageMissingReason(value: unknown): UsageMissingReason {
     : USAGE_MISSING_REASONS.NOT_AVAILABLE;
 }
 
-function optionalNumber<K extends string>(key: K, value: number | undefined): Partial<Record<K, number>> {
-  return value !== undefined ? { [key]: value } as Partial<Record<K, number>> : {};
-}
-
 function hrTimeToIso(time: readonly [number, number] | undefined): string {
   if (!time) {
     return new Date().toISOString();
diff --git a/src/core/logging/providerEvent.ts b/src/core/logging/providerEvent.ts
diff --git a/src/infra/observability/otelFoundation.ts b/src/infra/observability/otelFoundation.ts
diff --git a/src/infra/observability/usageEventsSpanProcessor.ts b/src/infra/observability/usageEventsSpanProcessor.ts
