Improve Langfuse observability attribution (#4003)

This commit is contained in:
yinjialu
2026-06-09 22:34:10 +00:00
committed by GitHub
parent fe49408313
commit 0ba68c8f23
11 changed files with 812 additions and 44 deletions
+30 -4
View File
@@ -18,6 +18,7 @@ import { readAppConfig } from './app-config.js';
import type { AppVersionInfo } from './app-version.js';
import { listMessages } from './db.js';
import {
deriveLangfuseDeliveryState,
readTelemetrySinkConfig,
reportRunCompleted,
reportRunFeedback,
@@ -27,6 +28,7 @@ import {
type AttachmentManifestEntry,
type EventsSummary,
type FeedbackReportContext,
type LangfuseDeliveryState,
type InputTextSnapshotManifestEntry,
type ObjectManifestCompleteness,
type MessageSummary,
@@ -45,7 +47,11 @@ import {
type RunTelemetryTimestamps,
type RunUsageAnalytics,
} from './run-analytics-observability.js';
import { collectStderrTailSummary } from './run-diagnostics.js';
import {
collectStderrTailSummary,
collectStdoutTailSummary,
summarizeRunDiagnosticsForAnalytics,
} from './run-diagnostics.js';
import { classifyRunFailure } from './run-failure-classification.js';
import { deriveRunErrorCode, runResultFromStatus } from './run-result.js';
import { buildTraceObjectManifests } from './trace-object-manifest.js';
@@ -821,12 +827,14 @@ function normalizeStatus(s: string): ReportContext['run']['status'] {
export async function reportRunCompletedFromDaemon(
opts: ReportRunCompletedFromDaemonOpts,
): Promise<void> {
): Promise<LangfuseDeliveryState> {
try {
const { db, dataDir, run } = opts;
const cfg = await readAppConfig(dataDir);
const prefs = cfg.telemetry ?? {};
if (prefs.metrics !== true) return;
if (prefs.metrics !== true) {
return deriveLangfuseDeliveryState(prefs, null);
}
const installationId = cfg.installationId ?? null;
let messageContent = '';
@@ -903,12 +911,23 @@ export async function reportRunCompletedFromDaemon(
const stderr = status === 'succeeded'
? undefined
: collectStderrTailSummary(run.events);
const stdout = status === 'succeeded'
? undefined
: collectStdoutTailSummary(run.events);
const turn = turnInfoFromRun(run, usageAnalytics.agent_reported_model);
const runtime: RuntimeInfo = {
...getRuntimeInfo(opts.appVersion ?? null),
...(run.clientType ? { clientType: run.clientType } : {}),
};
const artifacts = summarizeProducedFiles(producedFilesRaw);
const diagnostics = summarizeRunDiagnosticsForAnalytics({
events: run.events,
exitCode: run.exitCode ?? null,
signal: run.signal ?? null,
cancelRequested: run.status === 'canceled',
firstTokenSeen: Boolean(run.analyticsTelemetry?.firstTokenAt),
artifactWriteSeen: artifacts.length > 0,
});
const manifests = buildTraceSafeManifests({
projectId: run.projectId,
runId: run.id,
@@ -943,6 +962,8 @@ export async function reportRunCompletedFromDaemon(
timings,
...(run.analyticsTelemetry ? { timingMarks: run.analyticsTelemetry } : {}),
...(stderr ? { stderr } : {}),
...(stdout ? { stdout } : {}),
diagnostics,
},
message: {
messageId: run.assistantMessageId ?? '',
@@ -986,12 +1007,17 @@ export async function reportRunCompletedFromDaemon(
const uploadedManifests = await buildTraceObjectManifests(objectManifestOptions);
const finalManifests = mergeTraceSafeManifests(manifests, uploadedManifests);
await reportRunCompleted(
return await reportRunCompleted(
buildContext(finalManifests),
opts.fetchImpl ? { fetchImpl: opts.fetchImpl } : {},
);
} catch (err) {
console.warn('[langfuse-bridge] report failed:', String(err));
return {
langfuse_expected: true,
langfuse_delivery_status: 'failed',
langfuse_drop_reason: 'network_error',
};
}
}
+8
View File
@@ -106,6 +106,12 @@ export interface RunSummary {
lineCount: number;
truncated: boolean;
};
stdout?: {
tail: string;
lineCount: number;
truncated: boolean;
};
diagnostics?: unknown;
}
export interface MessageSummary {
@@ -1113,6 +1119,8 @@ export function buildTracePayload(ctx: ReportContext): unknown[] {
...(ctx.run.failure ?? {}),
...(ctx.run.timings ?? {}),
stderr: ctx.run.stderr,
stdout: ctx.run.stdout,
diagnostics: ctx.run.diagnostics,
eventsSummary: ctx.eventsSummary,
tokens,
cost_usd: costBreakdown.cost_usd,
+129 -12
View File
@@ -19,18 +19,39 @@ export type StderrLineCountBucket =
| '21_100'
| 'gt_100';
export type RunCloseReason =
| 'exit_0'
| 'exit_nonzero'
| 'signal'
| 'cancel_requested'
| 'stream_error'
| 'fatal_rpc_error'
| 'empty_output'
| 'unknown';
export interface RunDiagnosticsAnalytics {
diagnostic_source: RunDiagnosticSource;
stderr_present: boolean;
stderr_line_count_bucket: StderrLineCountBucket;
stdout_present: boolean;
stdout_line_count_bucket: StderrLineCountBucket;
rpc_close_reason: RunCloseReason;
first_token_seen: boolean;
user_visible_output_seen: boolean;
tool_call_seen: boolean;
artifact_write_seen: boolean;
live_artifact_seen: boolean;
}
export interface StderrTailSummary {
export interface StreamTailSummary {
tail: string;
lineCount: number;
truncated: boolean;
}
export type StderrTailSummary = StreamTailSummary;
export type StdoutTailSummary = StreamTailSummary;
const STDERR_TAIL_MAX_LINES = 20;
const STDERR_TAIL_MAX_BYTES = 4 * 1024;
@@ -43,6 +64,15 @@ function readStderrChunk(data: unknown): string | null {
return null;
}
function readStdoutChunk(data: unknown): string | null {
if (typeof data === 'string') return data;
if (!data || typeof data !== 'object' || Array.isArray(data)) return null;
const obj = data as Record<string, unknown>;
if (typeof obj.chunk === 'string') return obj.chunk;
if (typeof obj.text === 'string') return obj.text;
return null;
}
function countLines(text: string): number {
if (!text) return 0;
return text.split(/\r?\n/).filter((line) => line.length > 0).length;
@@ -69,19 +99,21 @@ function truncateUtf8(value: string, maxBytes: number): {
return { value: value.slice(0, end), truncated: true };
}
export function collectStderrTailSummary(
function collectStreamTailSummary(
events: RunEventForDiagnostics[] = [],
): StderrTailSummary | undefined {
let stderr = '';
eventName: string,
readChunk: (data: unknown) => string | null,
): StreamTailSummary | undefined {
let streamText = '';
for (const event of events) {
if (event.event !== 'stderr') continue;
const chunk = readStderrChunk(event.data);
if (chunk) stderr += chunk;
if (event.event !== eventName) continue;
const chunk = readChunk(event.data);
if (chunk) streamText += chunk;
}
const lineCount = countLines(stderr);
const lineCount = countLines(streamText);
if (lineCount <= 0) return undefined;
const lines = stderr.trimEnd().split(/\r?\n/);
const lines = streamText.trimEnd().split(/\r?\n/);
const tailLines = lines.slice(-STDERR_TAIL_MAX_LINES);
const lineTruncated = lines.length > tailLines.length;
const redacted = redactSecrets(tailLines.join('\n'));
@@ -94,21 +126,87 @@ export function collectStderrTailSummary(
};
}
export function collectStderrTailSummary(
events: RunEventForDiagnostics[] = [],
): StderrTailSummary | undefined {
return collectStreamTailSummary(events, 'stderr', readStderrChunk);
}
export function collectStdoutTailSummary(
events: RunEventForDiagnostics[] = [],
): StdoutTailSummary | undefined {
return collectStreamTailSummary(events, 'stdout', readStdoutChunk);
}
export function summarizeRunDiagnosticsForAnalytics(args: {
events?: RunEventForDiagnostics[];
exitCode?: number | null;
signal?: string | null;
cancelRequested?: boolean;
streamErrorSeen?: boolean;
fatalRpcErrorSeen?: boolean;
emptyOutputFailure?: boolean;
firstTokenSeen?: boolean;
artifactWriteSeen?: boolean;
liveArtifactSeen?: boolean;
}): RunDiagnosticsAnalytics {
const events = args.events ?? [];
let stderr = '';
let stdout = '';
let userVisibleOutputSeen = false;
let toolCallSeen = false;
let artifactWriteSeen = args.artifactWriteSeen === true;
let liveArtifactSeen = args.liveArtifactSeen === true;
let recordedCloseReason: RunCloseReason | null = null;
for (const event of events) {
if (event.event !== 'stderr') continue;
const chunk = readStderrChunk(event.data);
if (chunk) stderr += chunk;
if (event.event === 'stderr') {
const chunk = readStderrChunk(event.data);
if (chunk) stderr += chunk;
}
if (event.event === 'stdout') {
const chunk = readStdoutChunk(event.data);
if (chunk) {
stdout += chunk;
userVisibleOutputSeen = true;
}
}
const data = event.data && typeof event.data === 'object'
? event.data as Record<string, unknown>
: {};
if (data.type === 'text_delta' || data.type === 'thinking_delta') {
const delta = typeof data.delta === 'string' ? data.delta : '';
if (delta.length > 0) userVisibleOutputSeen = true;
}
if (data.type === 'tool_use') toolCallSeen = true;
if (data.type === 'artifact') artifactWriteSeen = true;
if (data.type === 'live_artifact' || event.event === 'live_artifact') {
liveArtifactSeen = true;
}
if (
event.event === 'diagnostic' &&
data.type === 'runtime_close' &&
typeof data.rpc_close_reason === 'string'
) {
const reason = data.rpc_close_reason;
if (
reason === 'exit_0' ||
reason === 'exit_nonzero' ||
reason === 'signal' ||
reason === 'cancel_requested' ||
reason === 'stream_error' ||
reason === 'fatal_rpc_error' ||
reason === 'empty_output' ||
reason === 'unknown'
) {
recordedCloseReason = reason;
}
}
}
const stderrLineCount = countLines(stderr);
const stdoutLineCount = countLines(stdout);
const hasErrorEvent = events.some((event) => event.event === 'error');
const stderrPresent = stderrLineCount > 0;
const stdoutPresent = stdoutLineCount > 0;
let diagnosticSource: RunDiagnosticSource = 'unknown';
if (hasErrorEvent) diagnosticSource = 'error_event';
@@ -116,9 +214,28 @@ export function summarizeRunDiagnosticsForAnalytics(args: {
else if (args.signal) diagnosticSource = 'signal';
else if (typeof args.exitCode === 'number') diagnosticSource = 'exit_code';
let rpcCloseReason: RunCloseReason = 'unknown';
if (recordedCloseReason) rpcCloseReason = recordedCloseReason;
else if (args.cancelRequested === true) rpcCloseReason = 'cancel_requested';
else if (args.fatalRpcErrorSeen === true) rpcCloseReason = 'fatal_rpc_error';
else if (args.streamErrorSeen === true) rpcCloseReason = 'stream_error';
else if (args.emptyOutputFailure === true) rpcCloseReason = 'empty_output';
else if (args.signal) rpcCloseReason = 'signal';
else if (typeof args.exitCode === 'number') {
rpcCloseReason = args.exitCode === 0 ? 'exit_0' : 'exit_nonzero';
}
return {
diagnostic_source: diagnosticSource,
stderr_present: stderrPresent,
stderr_line_count_bucket: stderrLineCountBucket(stderrLineCount),
stdout_present: stdoutPresent,
stdout_line_count_bucket: stderrLineCountBucket(stdoutLineCount),
rpc_close_reason: rpcCloseReason,
first_token_seen: args.firstTokenSeen === true,
user_visible_output_seen: userVisibleOutputSeen,
tool_call_seen: toolCallSeen,
artifact_write_seen: artifactWriteSeen,
live_artifact_seen: liveArtifactSeen,
};
}
+72 -10
View File
@@ -51,11 +51,16 @@ function eventErrorText(data: unknown): string[] {
const nested = payload.error && typeof payload.error === 'object'
? payload.error as Record<string, unknown>
: {};
const nestedData = nested.data && typeof nested.data === 'object'
? nested.data as Record<string, unknown>
: {};
return [
readString(payload.message),
readString(payload.code),
readString(nested.message),
readString(nested.code),
readString(nestedData.message),
typeof nestedData.statusCode === 'number' ? `statusCode:${nestedData.statusCode}` : undefined,
].filter((value): value is string => Boolean(value));
}
@@ -81,7 +86,13 @@ function latestRetryable(
const nested = payload.error && typeof payload.error === 'object'
? payload.error as Record<string, unknown>
: {};
const retryable = readBool(payload.retryable) ?? readBool(nested.retryable);
const nestedData = nested.data && typeof nested.data === 'object'
? nested.data as Record<string, unknown>
: {};
const retryable =
readBool(payload.retryable) ??
readBool(nested.retryable) ??
readBool(nestedData.isRetryable);
if (retryable !== undefined) return retryable;
}
return undefined;
@@ -121,10 +132,25 @@ function isEmptyOutputText(text: string): boolean {
}
function isToolErrorText(text: string): boolean {
return /\b(tool|mcp|connector)\b/i.test(text) &&
if (isPluginArtifactMissingText(text)) return true;
return /\b(tool|mcp|connector|plugin)\b/i.test(text) &&
/\b(error|failed|failure)\b/i.test(text);
}
function isPluginArtifactMissingText(text: string): boolean {
return /\bPlugin authoring ended before generating the required generated-plugin artifacts\b/i
.test(text);
}
function isAgentConfigInvalidText(text: string): boolean {
return /\bError loading config\.toml: unknown variant\b/i.test(text) ||
/\bunknown variant [`'"][^`'"]+[`'"], expected\b[\s\S]*\bin `service_tier`/i.test(text);
}
function isFabricatedRoleMarkerText(text: string): boolean {
return /\bmodel emitted fabricated role marker\b/i.test(text);
}
function isPermissionRequestNotFoundText(text: string): boolean {
return /\b(PermissionNotFoundError|Permission request not found|permissions\/per_[A-Za-z0-9_-]+\s+returned\s+HTTP\s+404)\b/i
.test(text);
@@ -141,7 +167,12 @@ function isPromptTooLargeText(text: string): boolean {
}
function isUpstreamDetailText(text: string): boolean {
return /\b(stream disconnected before completion|response\.completed|Transport error: network error|Upstream request failed|websocket closed|socket connection was closed unexpectedly|tls handshake eof|Connection reset by peer|TLS close_notify|Broken pipe|remote host|远程主机强迫关闭|No route to host|Connection refused|error sending request|Provider returned error|high demand|upstream_error|http2: response body closed)\b/i
return /\b(stream disconnected before completion|response\.completed|Transport error: network error|Upstream request failed|websocket closed|socket connection was closed unexpectedly|tls handshake eof|Connection reset by (?:peer|server)|TLS close_notify|Broken pipe|remote host|远程主机强迫关闭|No route to host|Connection refused|error sending request|Provider returned error|high demand|upstream_error|http2: response body closed|AMR model catalog is unavailable|statusCode[\"']?\s*:\s*(?:400|404)|400 Bad Request|404 Not Found)\b/i
.test(text);
}
function isUpstreamClientErrorText(text: string): boolean {
return /\b(statusCode[\"']?\s*:\s*(?:400|404)|400 Bad Request|404 Not Found)\b/i
.test(text);
}
@@ -179,7 +210,7 @@ function authDetail(text: string): TrackingRunFailureDetail {
}
function upstreamDetail(text: string): TrackingRunFailureDetail {
if (/\b(no endpoints found that support tool use|provider routing)\b/i.test(text)) {
if (/\b(AMR model catalog is unavailable|no endpoints found that support tool use|provider routing)\b/i.test(text)) {
return 'provider_routing_error';
}
if (/\bhigh demand|temporary errors\b/i.test(text)) return 'provider_high_demand';
@@ -191,6 +222,7 @@ function upstreamDetail(text: string): TrackingRunFailureDetail {
.test(text)) {
return 'upstream_5xx';
}
if (isUpstreamClientErrorText(text)) return 'upstream_client_error';
return 'network_error';
}
@@ -252,6 +284,11 @@ function signalInterruptClassification(
return classification('process_exit', 'terminated_unknown', 'child_close', false, 'none');
}
function toolErrorDetail(text: string): TrackingRunFailureDetail {
if (isPluginArtifactMissingText(text)) return 'plugin_artifact_missing';
return 'tool_error';
}
function processExitDetail(
errorCode: string,
text: string,
@@ -264,6 +301,8 @@ function processExitDetail(
if (/\bspawn failed: spawn EPERM\b/i.test(text)) return 'spawn_eperm';
if (/\bspawn failed: spawn\b/i.test(text)) return 'spawn_failed';
if (/\bstdin: write EOF\b/i.test(text)) return 'stdin_write_eof';
if (isAgentConfigInvalidText(text)) return 'agent_config_invalid';
if (isFabricatedRoleMarkerText(text)) return 'fabricated_role_marker';
if (/\bjson-rpc id \d+: Internal error\b/i.test(text)) {
return 'agent_protocol_error';
}
@@ -390,6 +429,16 @@ export function classifyRunFailure(
);
}
if (isAgentConfigInvalidText(text)) {
return classification(
'process_exit',
'agent_config_invalid',
'session_init',
false,
'fix_config',
);
}
const serviceFailure = classifyAgentServiceFailure(text);
if (serviceFailure === 'AGENT_AUTH_REQUIRED' || isAuthDetailText(text)) {
return classification(
@@ -417,12 +466,13 @@ export function classifyRunFailure(
serviceFailure === 'UPSTREAM_UNAVAILABLE' ||
isUpstreamDetailText(text)
) {
const retryable = retryableHint ?? !isUpstreamClientErrorText(text);
return classification(
'upstream_unavailable',
upstreamDetail(text),
'first_token_wait',
retryableHint ?? true,
'retry',
retryable,
retryable ? 'retry' : 'none',
);
}
@@ -450,12 +500,24 @@ export function classifyRunFailure(
}
if (isToolErrorText(text)) {
const retryable = retryableHint ?? !isPluginArtifactMissingText(text);
return classification(
'tool_error',
'tool_error',
'tool_execution',
retryableHint ?? false,
retryableHint ? 'retry' : 'none',
toolErrorDetail(text),
isPluginArtifactMissingText(text) ? 'artifact_write' : 'tool_execution',
retryable,
retryable ? 'retry' : 'none',
);
}
if (isFabricatedRoleMarkerText(text)) {
const retryable = retryableHint ?? true;
return classification(
'process_exit',
'fabricated_role_marker',
'child_close',
retryable,
retryable ? 'retry' : 'none',
);
}
+159 -16
View File
@@ -3030,20 +3030,129 @@ export function createFinalizedMessageTelemetryReporter({
getAppVersion?: () => any;
report?: typeof reportRunCompletedFromDaemon;
}) {
return (saved, body = {}) => {
if (!shouldReportRunCompletedFromMessage(saved, body)) return;
const run = design.runs.get(saved.runId);
if (!run || reportedRuns.has(run.id)) return;
reportedRuns.add(run.id);
void report({
db,
dataDir,
run,
persistedRunStatus: saved.runStatus,
persistedEndedAt: saved.endedAt,
appVersion: getAppVersion(),
const appVersionForCapture = () => {
const appVersion = getAppVersion();
if (typeof appVersion === 'string') return appVersion;
if (appVersion && typeof appVersion.version === 'string') return appVersion.version;
if (typeof design?.getAppVersion === 'function') return design.getAppVersion();
return 'unknown';
};
const captureResult = ({
analyticsContext,
conversationId,
delivery,
durationMs,
projectId,
reportResult,
run,
runId,
skipReason,
status,
}) => {
if (!analyticsContext || !design?.analytics?.capture || !runId || !delivery) return;
const terminalResult = status ? runResultFromStatus(status) : undefined;
design.analytics.capture({
eventName: 'langfuse_report_result',
context: analyticsContext,
appVersion: appVersionForCapture(),
properties: {
page_name: 'chat_panel',
area: 'chat_panel',
project_id: run?.projectId ?? projectId ?? null,
conversation_id: run?.conversationId ?? conversationId ?? null,
run_id: runId,
langfuse_trace_id: runId,
langfuse_expected: delivery.langfuse_expected,
langfuse_delivery_status: delivery.langfuse_delivery_status,
...(delivery.langfuse_drop_reason
? { langfuse_drop_reason: delivery.langfuse_drop_reason }
: {}),
langfuse_report_result: reportResult,
langfuse_report_trigger: 'final_message',
...(skipReason ? { langfuse_report_skip_reason: skipReason } : {}),
...(durationMs !== undefined ? { report_duration_ms: durationMs } : {}),
...(terminalResult ? { result: terminalResult } : {}),
...(run?.errorCode ? { error_code: run.errorCode } : {}),
...(run?.agentId ? { agent_provider_id: agentIdToTracking(run.agentId) } : {}),
...(run?.model !== undefined ? { model_id: modelIdForTracking(run.model) } : {}),
},
insertId: `${runId}-langfuse-report-${reportResult}${skipReason ? `-${skipReason}` : ''}`,
});
};
return (saved, body = {}, options = {}) => {
if (!shouldReportRunCompletedFromMessage(saved, body)) return;
const runId = saved.runId;
const run = design.runs.get(runId);
if (!run) {
captureResult({
analyticsContext: options.analyticsContext,
conversationId: options.conversationId ?? saved.conversationId,
delivery: {
langfuse_expected: true,
langfuse_delivery_status: 'failed',
langfuse_drop_reason: 'network_error',
},
projectId: options.projectId,
reportResult: 'skipped',
runId,
skipReason: 'run_not_found',
status: saved.runStatus,
});
return;
}
if (reportedRuns.has(run.id)) {
captureResult({
analyticsContext: options.analyticsContext,
conversationId: options.conversationId ?? saved.conversationId,
delivery: {
langfuse_expected: true,
langfuse_delivery_status: 'failed',
langfuse_drop_reason: 'network_error',
},
projectId: options.projectId,
reportResult: 'skipped',
run,
runId: run.id,
skipReason: 'duplicate_run',
status: saved.runStatus,
});
return;
}
reportedRuns.add(run.id);
void (async () => {
const start = Date.now();
const delivery = await report({
db,
dataDir,
run,
persistedRunStatus: saved.runStatus,
persistedEndedAt: saved.endedAt,
appVersion: getAppVersion(),
});
const state = delivery ?? {
langfuse_expected: true,
langfuse_delivery_status: 'accepted',
};
captureResult({
analyticsContext: options.analyticsContext,
conversationId: options.conversationId ?? saved.conversationId,
delivery: state,
durationMs: Date.now() - start,
projectId: options.projectId,
reportResult: state.langfuse_expected === false
? 'skipped'
: state.langfuse_delivery_status === 'accepted'
? 'accepted'
: state.langfuse_delivery_status === 'failed'
? 'failed'
: 'skipped',
run,
runId: run.id,
skipReason: state.langfuse_expected === false ? 'not_expected' : undefined,
status: saved.runStatus,
});
})();
};
}
const CLOUDFLARE_PAGES_PROJECT_METADATA_KEY = 'cloudflarePagesProjectName';
@@ -6296,7 +6405,11 @@ export async function startServer({
});
// Bump the parent project's updatedAt so the project list re-orders.
updateProject(db, req.params.id, {});
reportFinalizedMessage(saved, m);
reportFinalizedMessage(saved, m, {
analyticsContext: readAnalyticsContext(req),
projectId: req.params.id,
conversationId: req.params.cid,
});
res.json({ message: saved });
});
@@ -11808,6 +11921,17 @@ export async function startServer({
: {}),
});
};
let pendingRpcCloseReason = null;
const markRpcCloseReason = (reason) => {
pendingRpcCloseReason = reason;
};
const deriveRpcCloseReason = (status, code, signal) => {
if (pendingRpcCloseReason) return pendingRpcCloseReason;
if (run.cancelRequested || status === 'canceled') return 'cancel_requested';
if (signal) return 'signal';
if (typeof code === 'number') return code === 0 ? 'exit_0' : 'exit_nonzero';
return 'unknown';
};
const finishWithRetryDecision = (status, code = null, signal = null) => {
run.analyticsTelemetry = {
...(run.analyticsTelemetry ?? {}),
@@ -11904,6 +12028,15 @@ export async function startServer({
});
}
finalizeRetryTelemetry(status, decision, failure, errorCode);
const rpcCloseReason = deriveRpcCloseReason(status, code, signal);
design.runs.emit(run, 'diagnostic', {
type: 'runtime_close',
rpc_close_reason: rpcCloseReason,
status,
...(typeof code === 'number' ? { exit_code: code } : {}),
...(signal ? { signal } : {}),
});
pendingRpcCloseReason = null;
design.runs.finish(run, status, code, signal);
return false;
};
@@ -13417,9 +13550,11 @@ export async function startServer({
revokeToolToken('child_exit');
unregisterChatAgentEventSink();
if (acpSession?.hasFatalError()) {
markRpcCloseReason('fatal_rpc_error');
return finishWithRetryDecision('failed', code ?? 1, signal ?? null);
}
if (agentStreamError) {
markRpcCloseReason('stream_error');
return finishWithRetryDecision('failed', code ?? 1, signal ?? null);
}
if (
@@ -13477,6 +13612,7 @@ export async function startServer({
trackingSubstantiveOutput &&
!agentProducedOutput
) {
markRpcCloseReason('empty_output');
send('error', createSseErrorPayload(
'AGENT_EXECUTION_FAILED',
'Agent completed without producing any output. The model or provider may have returned an empty response — check the agent logs for upstream errors.',
@@ -13542,6 +13678,7 @@ export async function startServer({
!trackingSubstantiveOutput &&
!childStdoutSeen
) {
markRpcCloseReason('empty_output');
let combinedDetail = `${agentStderrTail}\n${agentStdoutTail}`;
if (def.id === 'antigravity' && agentLogFilePath) {
try {
@@ -14538,10 +14675,16 @@ export async function startServer({
telemetry: run.analyticsTelemetry,
events: run.events,
});
const artifactCount = countNewHtmlArtifacts(run.events);
const designSystemCreated = didRunCreateDesignSystemFile(run.events);
const previewModuleCount = countDesignSystemPreviewModules(run.events);
const diagnosticsAnalytics = summarizeRunDiagnosticsForAnalytics({
events: run.events,
exitCode: status.exitCode ?? null,
signal: status.signal ?? null,
cancelRequested: !!run.cancelRequested,
firstTokenSeen: Boolean(run.analyticsTelemetry?.firstTokenAt),
artifactWriteSeen: artifactCount > 0 || designSystemCreated || previewModuleCount > 0,
});
const finishedModelId = hasExplicitRequestedModelForAnalytics(reqBody.model)
? modelIdForTracking(reqBody.model)
@@ -14576,7 +14719,7 @@ export async function startServer({
// artifact?" funnel on PostHog. See `run-artifacts.ts`
// for the dedup semantics; tested in
// `tests/run-artifacts.test.ts`.
artifact_count: countNewHtmlArtifacts(run.events),
artifact_count: artifactCount,
// True when the run raised an AskUserQuestion clarification
// card. Clarification turns inherently produce no artifact, so
// the dashboard excludes them from the "run finished -> has
@@ -14593,8 +14736,8 @@ export async function startServer({
// succeeded; the run-artifacts inspector reuses the
// same Write/Edit pairing it already does for HTML
// artifact counts, just keyed on `DESIGN.md`.
design_system_created: didRunCreateDesignSystemFile(run.events),
preview_module_count: countDesignSystemPreviewModules(run.events),
design_system_created: designSystemCreated,
preview_module_count: previewModuleCount,
// `missing_font_count` defaults to 0 — the agent flow
// doesn't emit a structured "missing fonts" signal yet.
// Kept on the wire so the dashboard has the column from
@@ -95,7 +95,10 @@ describe('langfuse-bridge non-blocking behavior', () => {
run: makeRun() as any,
fetchImpl: vi.fn() as any,
}),
).resolves.toBeUndefined();
).resolves.toEqual({
langfuse_expected: true,
langfuse_delivery_status: 'accepted',
});
expect(reportRunCompletedMock).toHaveBeenCalledTimes(1);
expect(warnSpy).toHaveBeenCalledWith(
@@ -118,7 +121,11 @@ describe('langfuse-bridge non-blocking behavior', () => {
run: makeRun() as any,
fetchImpl: vi.fn() as any,
}),
).resolves.toBeUndefined();
).resolves.toEqual({
langfuse_expected: true,
langfuse_delivery_status: 'failed',
langfuse_drop_reason: 'network_error',
});
expect(warnSpy).toHaveBeenCalledWith(
'[langfuse-bridge] report failed:',
+63
View File
@@ -2,6 +2,7 @@ import { describe, expect, it } from 'vitest';
import {
collectStderrTailSummary,
collectStdoutTailSummary,
summarizeRunDiagnosticsForAnalytics,
} from '../src/run-diagnostics.js';
@@ -37,6 +38,14 @@ describe('run diagnostics', () => {
diagnostic_source: 'stderr',
stderr_present: true,
stderr_line_count_bucket: '1_5',
stdout_present: false,
stdout_line_count_bucket: 'none',
rpc_close_reason: 'exit_nonzero',
first_token_seen: false,
user_visible_output_seen: false,
tool_call_seen: false,
artifact_write_seen: false,
live_artifact_seen: false,
});
});
@@ -57,6 +66,14 @@ describe('run diagnostics', () => {
diagnostic_source: 'stderr',
stderr_present: true,
stderr_line_count_bucket: '1_5',
stdout_present: false,
stdout_line_count_bucket: 'none',
rpc_close_reason: 'exit_nonzero',
first_token_seen: false,
user_visible_output_seen: false,
tool_call_seen: false,
artifact_write_seen: false,
live_artifact_seen: false,
});
});
@@ -74,6 +91,52 @@ describe('run diagnostics', () => {
diagnostic_source: 'error_event',
stderr_present: true,
stderr_line_count_bucket: '1_5',
stdout_present: false,
stdout_line_count_bucket: 'none',
rpc_close_reason: 'exit_nonzero',
first_token_seen: false,
user_visible_output_seen: false,
tool_call_seen: false,
artifact_write_seen: false,
live_artifact_seen: false,
});
});
it('summarizes stdout into redacted bounded tails for Langfuse', () => {
const summary = collectStdoutTailSummary([
{ event: 'stdout', data: { chunk: 'visible line\nOPENAI_API_KEY=sk-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n' } },
]);
expect(summary).toBeTruthy();
expect(summary?.lineCount).toBe(2);
expect(summary?.tail).toContain('visible line');
expect(summary?.tail).toContain('[REDACTED:sk_key]');
expect(summary?.tail).not.toContain('sk-');
});
it('captures close reason and side-effect flags for aggregation', () => {
const result = summarizeRunDiagnosticsForAnalytics({
events: [
{ event: 'stdout', data: { chunk: 'hello\n' } },
{ event: 'agent', data: { type: 'tool_use', name: 'Read' } },
{ event: 'agent', data: { type: 'artifact' } },
],
exitCode: null,
signal: 'SIGTERM',
firstTokenSeen: true,
liveArtifactSeen: true,
});
expect(result).toMatchObject({
diagnostic_source: 'signal',
stdout_present: true,
stdout_line_count_bucket: '1_5',
rpc_close_reason: 'signal',
first_token_seen: true,
user_visible_output_seen: true,
tool_call_seen: true,
artifact_write_seen: true,
live_artifact_seen: true,
});
});
});
@@ -289,6 +289,101 @@ describe('classifyRunFailure', () => {
});
});
it('promotes AMR exit 130 connection resets into upstream stream disconnects', () => {
expect(
classify(
'AGENT_EXIT_130',
'json-rpc id 4: Connection reset by server',
),
).toMatchObject({
failure_category: 'upstream_unavailable',
failure_detail: 'stream_disconnected',
failure_stage: 'first_token_wait',
retryable: true,
user_action: 'retry',
});
});
it('promotes opencode API 4xx session errors out of process-exit fallback', () => {
expect(
classify(
'AGENT_EXIT_130',
'json-rpc id 4: opencode event stream: opencode session error: {"sessionID":"ses_16a081173ffeQy9mUJTmYowj5p","error":{"name":"APIError","data":{"message":"Not Found","statusCode":404,"isRetryable":false,"responseBody":"<html><head><title>404 Not Found</title></head>"}}}',
),
).toMatchObject({
failure_category: 'upstream_unavailable',
failure_detail: 'upstream_client_error',
failure_stage: 'first_token_wait',
retryable: false,
user_action: 'none',
});
expect(
classify(
'AGENT_EXIT_130',
'json-rpc id 4: opencode event stream: opencode session error: {"error":{"name":"APIError","data":{"message":"Bad Request","statusCode":400,"isRetryable":false,"responseBody":"<html><head><title>400 Bad Request</title></head>"}}}',
),
).toMatchObject({
failure_category: 'upstream_unavailable',
failure_detail: 'upstream_client_error',
retryable: false,
user_action: 'none',
});
});
it('uses structured opencode API error data when raw error text is sparse', () => {
expect(
classifyRunFailure({
result: 'failed',
status: {
status: 'failed',
error: null,
errorCode: 'AGENT_EXIT_130',
exitCode: 130,
signal: null,
},
errorCode: 'AGENT_EXIT_130',
agentId: 'opencode',
events: [
{
event: 'error',
data: {
error: {
name: 'APIError',
data: {
message: 'Not Found',
statusCode: 404,
isRetryable: false,
},
},
},
},
],
}),
).toMatchObject({
failure_category: 'upstream_unavailable',
failure_detail: 'upstream_client_error',
failure_stage: 'first_token_wait',
retryable: false,
user_action: 'none',
});
});
it('maps AMR model catalog outages to provider routing failures', () => {
expect(
classify(
'AGENT_EXIT_130',
'json-rpc id 2: AMR model catalog is unavailable.',
),
).toMatchObject({
failure_category: 'upstream_unavailable',
failure_detail: 'provider_routing_error',
failure_stage: 'first_token_wait',
retryable: true,
user_action: 'retry',
});
});
it('maps AMR insufficient balance to recharge guidance', () => {
expect(
classify('AMR_INSUFFICIENT_BALANCE', 'insufficient wallet balance'),
@@ -421,6 +516,51 @@ describe('classifyRunFailure', () => {
});
});
it('maps invalid agent config to fix-config guidance', () => {
expect(
classify(
'AGENT_EXECUTION_FAILED',
'Error loading config.toml: unknown variant `priority`, expected `fast` or `flex`\nin `service_tier`',
),
).toMatchObject({
failure_category: 'process_exit',
failure_detail: 'agent_config_invalid',
failure_stage: 'session_init',
retryable: false,
user_action: 'fix_config',
});
});
it('maps fabricated role marker termination to a retryable protocol guard detail', () => {
expect(
classify(
'AGENT_EXECUTION_FAILED',
'Run terminated: model emitted fabricated role marker (`## user`). No further tokens or tool calls accepted from this turn.',
),
).toMatchObject({
failure_category: 'process_exit',
failure_detail: 'fabricated_role_marker',
failure_stage: 'child_close',
retryable: true,
user_action: 'retry',
});
});
it('maps missing generated plugin artifacts to artifact-write tool failures', () => {
expect(
classify(
'AGENT_EXECUTION_FAILED',
'Plugin authoring ended before generating the required generated-plugin artifacts.',
),
).toMatchObject({
failure_category: 'tool_error',
failure_detail: 'plugin_artifact_missing',
failure_stage: 'artifact_write',
retryable: false,
user_action: 'none',
});
});
it('keeps process exits as an explicit fallback category', () => {
expect(classify('AGENT_EXIT_1', 'process exited with code 1')).toMatchObject({
failure_category: 'process_exit',
@@ -263,4 +263,114 @@ describe('Langfuse message finalization gate', () => {
appVersion: { version: '0.7.0', channel: 'beta', packaged: true },
});
});
it('captures Langfuse report acceptance after final message reporting resolves', async () => {
const run = {
id: 'run-accepted',
projectId: 'project-1',
conversationId: 'conv-1',
assistantMessageId: 'assistant-1',
agentId: 'codex',
model: 'default',
status: 'succeeded',
createdAt: 1,
updatedAt: 2,
events: [],
};
const capture = vi.fn();
const report = vi.fn(async () => ({
langfuse_expected: true,
langfuse_delivery_status: 'accepted' as const,
}));
const reporter = createFinalizedMessageTelemetryReporter({
design: {
analytics: { capture },
getAppVersion: () => '0.7.0',
runs: { get: vi.fn(() => run) },
},
db: 'db',
dataDir: '/tmp/od-data',
reportedRuns: new Set<string>(),
getAppVersion: () => ({ version: '0.7.0', channel: 'beta', packaged: true }),
report,
});
reporter(
{ ...terminalMessage, runId: run.id, endedAt: 1234 },
{ telemetryFinalized: true },
{
analyticsContext: {
deviceId: 'device-1',
sessionId: 'session-1',
clientType: 'desktop',
locale: 'zh-CN',
requestId: 'request-1',
},
projectId: 'project-1',
conversationId: 'conv-1',
},
);
await Promise.resolve();
expect(capture).toHaveBeenCalledWith(
expect.objectContaining({
eventName: 'langfuse_report_result',
insertId: 'run-accepted-langfuse-report-accepted',
properties: expect.objectContaining({
run_id: 'run-accepted',
langfuse_trace_id: 'run-accepted',
langfuse_expected: true,
langfuse_delivery_status: 'accepted',
langfuse_report_result: 'accepted',
langfuse_report_trigger: 'final_message',
}),
}),
);
});
it('captures skipped Langfuse reporting when a finalized message references a missing run', () => {
const capture = vi.fn();
const reporter = createFinalizedMessageTelemetryReporter({
design: {
analytics: { capture },
getAppVersion: () => '0.7.0',
runs: { get: vi.fn(() => undefined) },
},
db: 'db',
dataDir: '/tmp/od-data',
reportedRuns: new Set<string>(),
getAppVersion: () => ({ version: '0.7.0', channel: 'beta', packaged: true }),
report: vi.fn(),
});
reporter(
{ ...terminalMessage, runId: 'run-missing', endedAt: 1234 },
{ telemetryFinalized: true },
{
analyticsContext: {
deviceId: 'device-1',
sessionId: 'session-1',
clientType: 'desktop',
locale: 'zh-CN',
requestId: 'request-1',
},
projectId: 'project-1',
conversationId: 'conv-1',
},
);
expect(capture).toHaveBeenCalledWith(
expect.objectContaining({
eventName: 'langfuse_report_result',
insertId: 'run-missing-langfuse-report-skipped-run_not_found',
properties: expect.objectContaining({
project_id: 'project-1',
conversation_id: 'conv-1',
run_id: 'run-missing',
langfuse_report_result: 'skipped',
langfuse_report_skip_reason: 'run_not_found',
}),
}),
);
});
});
@@ -25,6 +25,7 @@ export type AnalyticsEventName =
// Run lifecycle (daemon authoritative)
| 'run_created'
| 'run_finished'
| 'langfuse_report_result'
| 'run_retry_attempted'
| 'run_retry_finished'
// Packaged updater lifecycle
@@ -247,6 +248,7 @@ export type TrackingRunFailureDetail =
| 'cli_version_incompatible'
| 'prompt_too_large'
| 'upstream_5xx'
| 'upstream_client_error'
| 'stream_disconnected'
| 'network_error'
| 'provider_high_demand'
@@ -255,13 +257,16 @@ export type TrackingRunFailureDetail =
| 'timeout'
| 'empty_output'
| 'tool_error'
| 'plugin_artifact_missing'
| 'cli_not_installed'
| 'agent_config_invalid'
| 'spawn_failed'
| 'spawn_enoexec'
| 'spawn_ebadf'
| 'spawn_eperm'
| 'stdin_write_eof'
| 'agent_protocol_error'
| 'fabricated_role_marker'
| 'permission_request_not_found'
| 'qoder_stop_sequence'
| 'signal_killed'
@@ -290,6 +295,7 @@ export type TrackingRunFailureUserAction =
| 'switch_model'
| 'reduce_context'
| 'install_cli'
| 'fix_config'
| 'none';
export type TrackingRunRetryStrategy = 'same_run_transient';
export type TrackingRunRetryFinalResult =
@@ -320,6 +326,15 @@ export type TrackingStderrLineCountBucket =
| '6_20'
| '21_100'
| 'gt_100';
export type TrackingRunCloseReason =
| 'exit_0'
| 'exit_nonzero'
| 'signal'
| 'cancel_requested'
| 'stream_error'
| 'fatal_rpc_error'
| 'empty_output'
| 'unknown';
export type TrackingLangfuseDeliveryStatus =
| 'not_expected'
| 'queued'
@@ -336,6 +351,14 @@ export type TrackingLangfuseDropReason =
| 'langfuse_4xx'
| 'langfuse_5xx'
| 'network_error';
export type TrackingLangfuseReportResult =
| 'accepted'
| 'failed'
| 'skipped';
export type TrackingLangfuseReportSkipReason =
| 'run_not_found'
| 'duplicate_run'
| 'not_expected';
export type TrackingFeedbackRating = 'positive' | 'negative';
// Click events emit `none` when the user clears a previously-set rating, so
@@ -2249,6 +2272,14 @@ export interface RunFinishedProps extends Omit<RunCreatedProps, 'area'> {
diagnostic_source?: TrackingRunDiagnosticSource;
stderr_present?: boolean;
stderr_line_count_bucket?: TrackingStderrLineCountBucket;
stdout_present?: boolean;
stdout_line_count_bucket?: TrackingStderrLineCountBucket;
rpc_close_reason?: TrackingRunCloseReason;
first_token_seen?: boolean;
user_visible_output_seen?: boolean;
tool_call_seen?: boolean;
artifact_write_seen?: boolean;
live_artifact_seen?: boolean;
artifact_count: number;
// True when the run raised an AskUserQuestion clarification card. Such runs
// are intent-clarification turns (the agent stops to ask the user a question)
@@ -2289,6 +2320,26 @@ export interface RunFinishedProps extends Omit<RunCreatedProps, 'area'> {
retry_suppressed_reason?: TrackingRunRetrySuppressedReason;
}
export interface LangfuseReportResultProps {
page_name: 'chat_panel' | 'design_system_project';
area: 'chat_panel' | 'design_system_generation';
project_id: string | null;
conversation_id: string | null;
run_id: string;
langfuse_trace_id: string;
langfuse_expected: boolean;
langfuse_delivery_status: TrackingLangfuseDeliveryStatus;
langfuse_drop_reason?: TrackingLangfuseDropReason;
langfuse_report_result: TrackingLangfuseReportResult;
langfuse_report_trigger: 'final_message';
langfuse_report_skip_reason?: TrackingLangfuseReportSkipReason;
report_duration_ms?: number;
result?: TrackingRunResult;
error_code?: string;
agent_provider_id?: TrackingCliProviderId;
model_id?: string;
}
export interface RunRetryBaseProps {
page_name: 'chat_panel' | 'design_system_project';
area: 'chat_panel' | 'design_system_generation';
@@ -2556,6 +2607,7 @@ export type AnalyticsEventPayload =
| { event: 'plugin_replacement_result'; props: PluginReplacementResultProps }
| { event: 'run_created'; props: RunCreatedProps }
| { event: 'run_finished'; props: RunFinishedProps }
| { event: 'langfuse_report_result'; props: LangfuseReportResultProps }
| { event: 'run_retry_attempted'; props: RunRetryAttemptedProps }
| { event: 'run_retry_finished'; props: RunRetryFinishedProps }
| { event: 'update_install_result'; props: UpdateInstallResultProps }
@@ -76,6 +76,17 @@ describe('analytics run_finished contract', () => {
tool_call_count: 2,
tool_duration_ms: 350,
finalize_duration_ms: 30,
diagnostic_source: 'error_event',
stderr_present: true,
stderr_line_count_bucket: '1_5',
stdout_present: true,
stdout_line_count_bucket: '1_5',
rpc_close_reason: 'stream_error',
first_token_seen: true,
user_visible_output_seen: true,
tool_call_seen: true,
artifact_write_seen: false,
live_artifact_seen: false,
retry_attempt_count: 1,
retry_final_result: 'success',
},
@@ -89,6 +100,8 @@ describe('analytics run_finished contract', () => {
expect(payload.props.cache_token_source).toBe('anthropic');
expect(payload.props.total_duration_ms).toBe(1234);
expect(payload.props.tool_call_count).toBe(2);
expect(payload.props.rpc_close_reason).toBe('stream_error');
expect(payload.props.first_token_seen).toBe(true);
expect(payload.props.retry_attempt_count).toBe(1);
expect(payload.props.retry_final_result).toBe('success');
});
@@ -128,4 +141,31 @@ describe('analytics run_finished contract', () => {
expect(attempted.props.retry_strategy).toBe('same_run_transient');
expect(finished.props.retry_suppressed_reason).toBe('tool_call_seen');
});
it('accepts Langfuse report result events for actual delivery monitoring', () => {
const payload = {
event: 'langfuse_report_result',
props: {
page_name: 'chat_panel',
area: 'chat_panel',
project_id: 'proj-1',
conversation_id: 'conv-1',
run_id: 'run-1',
langfuse_trace_id: 'run-1',
langfuse_expected: true,
langfuse_delivery_status: 'failed',
langfuse_drop_reason: 'network_error',
langfuse_report_result: 'failed',
langfuse_report_trigger: 'final_message',
report_duration_ms: 123,
result: 'failed',
error_code: 'AGENT_EXECUTION_FAILED',
agent_provider_id: 'codex_cli',
model_id: 'default',
},
} satisfies Extract<AnalyticsEventPayload, { event: 'langfuse_report_result' }>;
expect(payload.props.langfuse_report_result).toBe('failed');
expect(payload.props.langfuse_delivery_status).toBe('failed');
});
});