Skip to content

Commit 716c2f2

Browse files
yinjialuyinjialu
authored
Fix Langfuse completion telemetry fallback (#4073)
* Fix Langfuse completion telemetry fallback * Arm Langfuse fallback for headerless runs Generated-By: looper 0.9.6 (runner=fixer, agent=codex) * Scope Langfuse fallback to failed terminal runs Generated-By: looper 0.9.6 (runner=fixer, agent=codex) * Prevent Langfuse fallback from owning late finalization Generated-By: looper 0.9.6 (runner=fixer, agent=codex) * Report buffered Langfuse fallback traces Generated-By: looper 0.9.6 (runner=fixer, agent=codex) * Drop buffered payload state from Langfuse fallback guard Generated-By: looper 0.9.6 (runner=fixer, agent=codex) * Disambiguate Langfuse report capture IDs Generated-By: looper 0.9.6 (runner=fixer, agent=codex)
1 parent 7f35dc0 commit 716c2f2

8 files changed

Lines changed: 317 additions & 19 deletions

File tree

‎apps/daemon/src/db.ts‎

Lines changed: 37 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,7 @@ function migrate(db: SqliteDb): void {
111111
session_mode TEXT,
112112
run_context_json TEXT,
113113
applied_plugin_snapshot_json TEXT,
114+
telemetry_finalized_at INTEGER,
114115
started_at INTEGER,
115116
ended_at INTEGER,
116117
position INTEGER NOT NULL,
@@ -285,6 +286,9 @@ function migrate(db: SqliteDb): void {
285286
if (!messageCols.some((c: DbRow) => c.name === 'applied_plugin_snapshot_json')) {
286287
db.exec(`ALTER TABLE messages ADD COLUMN applied_plugin_snapshot_json TEXT`);
287288
}
289+
if (!messageCols.some((c: DbRow) => c.name === 'telemetry_finalized_at')) {
290+
db.exec(`ALTER TABLE messages ADD COLUMN telemetry_finalized_at INTEGER`);
291+
}
288292
const routineRunCols = db.prepare(`PRAGMA table_info(routine_runs)`).all() as DbRow[];
289293
if (!routineRunCols.some((c: DbRow) => c.name === 'error_code')) {
290294
db.exec(`ALTER TABLE routine_runs ADD COLUMN error_code TEXT`);
@@ -1207,6 +1211,10 @@ export function upsertMessage(db: SqliteDb, conversationId: string, m: DbRow) {
12071211
produced_files_json = ?, feedback_json = ?,
12081212
pre_turn_file_names_json = ?,
12091213
session_mode = ?, run_context_json = ?, applied_plugin_snapshot_json = ?,
1214+
telemetry_finalized_at = CASE
1215+
WHEN ? THEN COALESCE(telemetry_finalized_at, ?)
1216+
ELSE telemetry_finalized_at
1217+
END,
12101218
started_at = ?, ended_at = ?
12111219
WHERE id = ?`,
12121220
).run(
@@ -1226,6 +1234,8 @@ export function upsertMessage(db: SqliteDb, conversationId: string, m: DbRow) {
12261234
normalizeMessageSessionModeForStorage(m.sessionMode),
12271235
m.runContext ? JSON.stringify(m.runContext) : null,
12281236
m.appliedPluginSnapshot ? JSON.stringify(m.appliedPluginSnapshot) : null,
1237+
m.telemetryFinalized === true ? 1 : 0,
1238+
now,
12291239
m.startedAt ?? null,
12301240
m.endedAt ?? null,
12311241
m.id,
@@ -1237,20 +1247,21 @@ export function upsertMessage(db: SqliteDb, conversationId: string, m: DbRow) {
12371247
)
12381248
.get(conversationId) as DbRow | undefined;
12391249
const position = (max?.m ?? -1) + 1;
1240-
// 22 values: id, conversation_id, role, content, agent_id, agent_name,
1250+
// 23 values: id, conversation_id, role, content, agent_id, agent_name,
12411251
// run_id, run_status, last_run_event_id, events_json, attachments_json,
12421252
// comment_attachments_json, produced_files_json, feedback_json,
12431253
// pre_turn_file_names_json, session_mode, run_context_json,
1244-
// applied_plugin_snapshot_json, started_at, ended_at, position, created_at.
1254+
// applied_plugin_snapshot_json, telemetry_finalized_at, started_at,
1255+
// ended_at, position, created_at.
12451256
db.prepare(
12461257
`INSERT INTO messages
12471258
(id, conversation_id, role, content, agent_id, agent_name,
12481259
run_id, run_status, last_run_event_id, events_json,
12491260
attachments_json, comment_attachments_json, produced_files_json,
12501261
feedback_json, pre_turn_file_names_json,
12511262
session_mode, run_context_json, applied_plugin_snapshot_json,
1252-
started_at, ended_at, position, created_at)
1253-
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
1263+
telemetry_finalized_at, started_at, ended_at, position, created_at)
1264+
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
12541265
).run(
12551266
m.id,
12561267
conversationId,
@@ -1270,6 +1281,7 @@ export function upsertMessage(db: SqliteDb, conversationId: string, m: DbRow) {
12701281
normalizeMessageSessionModeForStorage(m.sessionMode),
12711282
m.runContext ? JSON.stringify(m.runContext) : null,
12721283
m.appliedPluginSnapshot ? JSON.stringify(m.appliedPluginSnapshot) : null,
1284+
m.telemetryFinalized === true ? now : null,
12731285
m.startedAt ?? null,
12741286
m.endedAt ?? null,
12751287
position,
@@ -1303,6 +1315,27 @@ export function upsertMessage(db: SqliteDb, conversationId: string, m: DbRow) {
13031315
return row ? normalizeMessage(row) : null;
13041316
}
13051317

1318+
export function getMessageTelemetryFinalizationState(db: SqliteDb, messageId: string) {
1319+
const row = db
1320+
.prepare(
1321+
`SELECT telemetry_finalized_at AS telemetryFinalizedAt
1322+
FROM messages
1323+
WHERE id = ?`,
1324+
)
1325+
.get(messageId) as DbRow | undefined;
1326+
if (!row) {
1327+
return {
1328+
exists: false,
1329+
finalizedAt: null,
1330+
};
1331+
}
1332+
return {
1333+
exists: true,
1334+
finalizedAt:
1335+
typeof row.telemetryFinalizedAt === 'number' ? row.telemetryFinalizedAt : null,
1336+
};
1337+
}
1338+
13061339
export function appendMessageStatusEvent(db: SqliteDb, messageId: string, event: DbRow) {
13071340
const label = typeof event?.label === 'string' ? event.label.trim() : '';
13081341
const detail = typeof event?.detail === 'string' ? event.detail.trim() : '';

‎apps/daemon/src/project-routes.ts‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import {
1919
} from './plugins/index.js';
2020
import { connectorService } from './connectors/service.js';
2121
import type { RouteDeps } from './server-context.js';
22+
import { readAnalyticsContext } from './analytics.js';
2223
import { listSkills } from './skills.js';
2324
import { isSafeId } from './projects.js';
2425
import {
@@ -1626,7 +1627,11 @@ export function registerProjectRoutes(app: Express, ctx: RegisterProjectRoutesDe
16261627
});
16271628
// Bump the parent project's updatedAt so the project list re-orders.
16281629
updateProject(db, req.params.id, {});
1629-
ctx.telemetry?.reportFinalizedMessage(saved, m);
1630+
ctx.telemetry?.reportFinalizedMessage(saved, m, {
1631+
analyticsContext: readAnalyticsContext(req),
1632+
projectId: req.params.id,
1633+
conversationId: req.params.cid,
1634+
});
16301635
res.json({ message: saved });
16311636
});
16321637

‎apps/daemon/src/server-context.ts‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,16 @@ export interface ProjectPreviewScopeDeps {
5959
}
6060

6161
export interface TelemetryDeps {
62-
reportFinalizedMessage: (saved: any, body?: any) => void;
62+
reportFinalizedMessage: (
63+
saved: any,
64+
body?: any,
65+
options?: {
66+
analyticsContext?: any;
67+
projectId?: string;
68+
conversationId?: string;
69+
reportTrigger?: 'final_message' | 'terminal_fallback';
70+
},
71+
) => void;
6372
/**
6473
* Best-effort Langfuse score emission for assistant-turn user ratings.
6574
* Returns the categorical outcome so the API surface in chat-routes can

‎apps/daemon/src/server.ts‎

Lines changed: 65 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -407,6 +407,7 @@ import {
407407
getConversation,
408408
getDeployment,
409409
getDeploymentById,
410+
getMessageTelemetryFinalizationState,
410411
getProject,
411412
getTemplate,
412413
insertConversation,
@@ -2566,6 +2567,7 @@ async function ensureGhReady() {
25662567
}
25672568

25682569
const TERMINAL_RUN_STATUSES = new Set(['succeeded', 'failed', 'canceled']);
2570+
const LANGFUSE_TERMINAL_FALLBACK_DELAY_MS = 15_000;
25692571

25702572
function reconcileAssistantMessageOnRunEnd(db, runs, run) {
25712573
if (!run.assistantMessageId) return;
@@ -3032,6 +3034,7 @@ export function createFinalizedMessageTelemetryReporter({
30323034
durationMs,
30333035
projectId,
30343036
reportResult,
3037+
reportTrigger = 'final_message',
30353038
run,
30363039
runId,
30373040
skipReason,
@@ -3056,15 +3059,15 @@ export function createFinalizedMessageTelemetryReporter({
30563059
? { langfuse_drop_reason: delivery.langfuse_drop_reason }
30573060
: {}),
30583061
langfuse_report_result: reportResult,
3059-
langfuse_report_trigger: 'final_message',
3062+
langfuse_report_trigger: reportTrigger,
30603063
...(skipReason ? { langfuse_report_skip_reason: skipReason } : {}),
30613064
...(durationMs !== undefined ? { report_duration_ms: durationMs } : {}),
30623065
...(terminalResult ? { result: terminalResult } : {}),
30633066
...(run?.errorCode ? { error_code: run.errorCode } : {}),
30643067
...(run?.agentId ? { agent_provider_id: agentIdToTracking(run.agentId) } : {}),
30653068
...(run?.model !== undefined ? { model_id: modelIdForTracking(run.model) } : {}),
30663069
},
3067-
insertId: `${runId}-langfuse-report-${reportResult}${skipReason ? `-${skipReason}` : ''}`,
3070+
insertId: `${runId}-langfuse-report-${reportTrigger}-${reportResult}${skipReason ? `-${skipReason}` : ''}`,
30683071
});
30693072
};
30703073
return (saved, body = {}, options = {}) => {
@@ -3081,13 +3084,15 @@ export function createFinalizedMessageTelemetryReporter({
30813084
langfuse_drop_reason: 'network_error',
30823085
},
30833086
projectId: options.projectId,
3087+
reportTrigger: options.reportTrigger,
30843088
reportResult: 'skipped',
30853089
runId,
30863090
skipReason: 'run_not_found',
30873091
status: saved.runStatus,
30883092
});
30893093
return;
30903094
}
3095+
const reportTrigger = options.reportTrigger ?? 'final_message';
30913096
if (reportedRuns.has(run.id)) {
30923097
captureResult({
30933098
analyticsContext: options.analyticsContext,
@@ -3098,6 +3103,7 @@ export function createFinalizedMessageTelemetryReporter({
30983103
langfuse_drop_reason: 'network_error',
30993104
},
31003105
projectId: options.projectId,
3106+
reportTrigger: options.reportTrigger,
31013107
reportResult: 'skipped',
31023108
run,
31033109
runId: run.id,
@@ -3106,7 +3112,9 @@ export function createFinalizedMessageTelemetryReporter({
31063112
});
31073113
return;
31083114
}
3109-
reportedRuns.add(run.id);
3115+
if (reportTrigger !== 'terminal_fallback') {
3116+
reportedRuns.add(run.id);
3117+
}
31103118
void (async () => {
31113119
const start = Date.now();
31123120
const delivery = await report({
@@ -3127,6 +3135,7 @@ export function createFinalizedMessageTelemetryReporter({
31273135
delivery: state,
31283136
durationMs: Date.now() - start,
31293137
projectId: options.projectId,
3138+
reportTrigger,
31303139
reportResult: state.langfuse_expected === false
31313140
? 'skipped'
31323141
: state.langfuse_delivery_status === 'accepted'
@@ -3143,6 +3152,10 @@ export function createFinalizedMessageTelemetryReporter({
31433152
};
31443153
}
31453154

3155+
export function shouldReportRunCompletionTelemetryFallbackStatus(status: unknown): boolean {
3156+
return status === 'failed' || status === 'canceled';
3157+
}
3158+
31463159
const CLOUDFLARE_PAGES_PROJECT_METADATA_KEY = 'cloudflarePagesProjectName';
31473160

31483161
function cloudflarePagesDeploymentMetadata(projectName) {
@@ -5780,8 +5793,10 @@ export async function startServer({
57805793
});
57815794
});
57825795

5783-
// Tracks runs whose completion has already been forwarded to Langfuse so
5784-
// repeated message updates only emit one trace per run.
5796+
// Tracks runs whose finalized assistant message has already been forwarded
5797+
// to Langfuse so repeated message updates only emit one final trace per run.
5798+
// Terminal fallback reports intentionally do not claim this set; a delayed
5799+
// telemetry-finalized message can still replace the synthetic fallback.
57855800
const reportedRuns = new Set();
57865801

57875802
// App-version snapshot read once at server start for Langfuse trace metadata.
@@ -5810,6 +5825,42 @@ export async function startServer({
58105825
reportedRuns,
58115826
getAppVersion: () => cachedAppVersion,
58125827
});
5828+
const reportRunCompletionTelemetryFallback = ({
5829+
analyticsContext,
5830+
run,
5831+
status,
5832+
}: {
5833+
analyticsContext: any;
5834+
run: any;
5835+
status: string;
5836+
}) => {
5837+
if (!shouldReportRunCompletionTelemetryFallbackStatus(status)) return;
5838+
const timer = setTimeout(() => {
5839+
if (reportedRuns.has(run.id)) return;
5840+
if (run.assistantMessageId) {
5841+
const messageTelemetry = getMessageTelemetryFinalizationState(db, run.assistantMessageId);
5842+
if (messageTelemetry.finalizedAt !== null) return;
5843+
}
5844+
reportFinalizedMessage(
5845+
{
5846+
id: run.assistantMessageId ?? `${run.id}-terminal`,
5847+
conversationId: run.conversationId,
5848+
endedAt: run.updatedAt,
5849+
role: 'assistant',
5850+
runId: run.id,
5851+
runStatus: status,
5852+
},
5853+
{ telemetryFinalized: true },
5854+
{
5855+
analyticsContext,
5856+
conversationId: run.conversationId,
5857+
projectId: run.projectId,
5858+
reportTrigger: 'terminal_fallback',
5859+
},
5860+
);
5861+
}, LANGFUSE_TERMINAL_FALLBACK_DELAY_MS);
5862+
timer.unref?.();
5863+
};
58135864

58145865
const reportFeedback = (req: {
58155866
runId: string;
@@ -14391,6 +14442,15 @@ export async function startServer({
1439114442
// here so PostHog actually receives the event. Both fire under the
1439214443
// same insert_id prefix so any web-side mirror dedupes by $insert_id.
1439314444
const analyticsContext = readAnalyticsContext(req);
14445+
design.runs.wait(run).then((status: { status: string }) => {
14446+
reportRunCompletionTelemetryFallback({
14447+
analyticsContext: analyticsContext ?? null,
14448+
run,
14449+
status: status.status,
14450+
});
14451+
}).catch(() => {
14452+
// wait() can't reject in current runs.ts impl, but guard anyway.
14453+
});
1439414454
if (analyticsContext) {
1439514455
const reqBody = (req.body || {}) as Record<string, unknown>;
1439614456
const runInsertId = newInsertId();

0 commit comments

Comments
 (0)