mirror of
https://github.com/nexu-io/open-design.git
synced 2026-10-27 09:51:12 +01:00
feat(daemon): populate run_devloop_iterations.tokens_used from run-level usage (#4644)
The devloop audit column tokens_used was always NULL: the column and write path exist, but StageRegistryOutcome carried no token field and the registry runStage closure dropped it. Reuse the existing scanRunEventsForUsageAnalytics reducer over the run's usage events to record the run-level token total on the iteration audit rows (run-cumulative, last-write-wins via a run-end reconciliation pass; not per-iteration attribution). Runtimes that emit no usage leave tokens_used NULL.
This commit is contained in:
@@ -27,6 +27,10 @@ import type {
|
||||
PipelineStage,
|
||||
} from '@open-design/contracts';
|
||||
import type { UntilSignals } from '../until.js';
|
||||
import {
|
||||
scanRunEventsForUsageAnalytics,
|
||||
type RunEventForAnalyticsObservability,
|
||||
} from '../../run-analytics-observability.js';
|
||||
|
||||
type SqliteDb = Database.Database;
|
||||
|
||||
@@ -38,6 +42,7 @@ export interface AtomWorkerContext {
|
||||
stage: PipelineStage;
|
||||
iteration: number;
|
||||
snapshot: AppliedPluginSnapshot;
|
||||
runEvents?: RunEventForAnalyticsObservability[];
|
||||
}
|
||||
|
||||
export interface AtomOutcome {
|
||||
@@ -93,6 +98,7 @@ export interface StageRegistryOutcome {
|
||||
critiqueSummary: string | null;
|
||||
notes: string[];
|
||||
observedAtoms: string[];
|
||||
tokensUsed?: number | null;
|
||||
}
|
||||
|
||||
// Walk every atom in the stage, invoke its registered worker (if
|
||||
@@ -131,9 +137,18 @@ export async function runStageWithRegistry(
|
||||
critiqueSummary: notes.length > 0 ? notes.join('\n') : null,
|
||||
notes,
|
||||
observedAtoms: observed,
|
||||
tokensUsed: tokensUsedFromRunEvents(ctx.runEvents),
|
||||
};
|
||||
}
|
||||
|
||||
function tokensUsedFromRunEvents(
|
||||
events: RunEventForAnalyticsObservability[] | undefined,
|
||||
): number | null {
|
||||
if (!events?.length) return null;
|
||||
const usage = scanRunEventsForUsageAnalytics(events, null, 0);
|
||||
return usage.total_tokens ?? null;
|
||||
}
|
||||
|
||||
// Pessimistic merge between multiple workers contributing to the
|
||||
// same signal key. Cross-worker false-wins / lowest-number-wins
|
||||
// so a single failing gate still surfaces as a failed convergence.
|
||||
|
||||
@@ -3898,14 +3898,16 @@ export async function startServer({
|
||||
stage,
|
||||
iteration,
|
||||
snapshot: stageSnapshot,
|
||||
runEvents: run.events,
|
||||
});
|
||||
return {
|
||||
signals: outcome.signals,
|
||||
critiqueSummary: outcome.critiqueSummary,
|
||||
tokensUsed: outcome.tokensUsed,
|
||||
};
|
||||
};
|
||||
}
|
||||
void runPipelineForRun({
|
||||
const pipelineDone = runPipelineForRun({
|
||||
db: dbHandle,
|
||||
runId: run.id,
|
||||
projectId: projectIdForRun,
|
||||
@@ -3925,6 +3927,17 @@ export async function startServer({
|
||||
});
|
||||
} catch { /* ignore */ }
|
||||
});
|
||||
void Promise.all([runs.wait(run), pipelineDone])
|
||||
.then(() => {
|
||||
const tokensUsed = scanRunEventsForUsageAnalytics(run.events, null, 0).total_tokens ?? null;
|
||||
if (tokensUsed === null) return;
|
||||
dbHandle.prepare(
|
||||
'UPDATE run_devloop_iterations SET tokens_used = ? WHERE run_id = ?',
|
||||
).run(tokensUsed, run.id);
|
||||
})
|
||||
.catch((err) => {
|
||||
console.warn('[plugins] devloop tokens_used reconciliation failed', err);
|
||||
});
|
||||
};
|
||||
|
||||
const startChatRun = async (chatBody, run) => {
|
||||
|
||||
@@ -28,6 +28,7 @@ import { runPipelineForRun } from '../src/plugins/pipeline-runner.js';
|
||||
import { listIterationsForRun } from '../src/plugins/pipeline.js';
|
||||
import { respondSurface } from '../src/genui/registry.js';
|
||||
import { findPendingByRunAndSurfaceId } from '../src/genui/store.js';
|
||||
import type { RunEventForAnalyticsObservability } from '../src/run-analytics-observability.js';
|
||||
import {
|
||||
clearAtomWorkers,
|
||||
registerAtomWorker,
|
||||
@@ -256,6 +257,66 @@ describe('pipeline-runner: Stage D registry runner integration', () => {
|
||||
expect(outcomes[0]?.iterations).toBe(1);
|
||||
expect(outcomes[0]?.termination).toBe('no-loop');
|
||||
});
|
||||
|
||||
it('persists registry run-level usage totals and leaves no-usage runs null', async () => {
|
||||
const snap = snapshotWith({
|
||||
pipeline: {
|
||||
stages: [{ id: 'compose', atoms: ['unknown-atom'] }],
|
||||
},
|
||||
});
|
||||
const usageEvents: RunEventForAnalyticsObservability[] = [{
|
||||
id: 1,
|
||||
event: 'agent',
|
||||
data: {
|
||||
type: 'usage',
|
||||
usage: { input_tokens: 17, output_tokens: 23, total_tokens: 45 },
|
||||
},
|
||||
timestamp: Date.now(),
|
||||
}];
|
||||
|
||||
await runPipelineForRun({
|
||||
db,
|
||||
runId: 'run-with-usage',
|
||||
projectId: 'project-1',
|
||||
conversationId: 'conv-A',
|
||||
snapshot: snap,
|
||||
pipeline: snap.pipeline!,
|
||||
env: { maxIterations: 1 },
|
||||
runStage: async ({ stage, iteration, snapshot: snap2 }) => runStageWithRegistry({
|
||||
db,
|
||||
runId: 'run-with-usage',
|
||||
projectId: 'project-1',
|
||||
conversationId: 'conv-A',
|
||||
stage,
|
||||
iteration,
|
||||
snapshot: snap2,
|
||||
runEvents: usageEvents,
|
||||
}),
|
||||
});
|
||||
|
||||
await runPipelineForRun({
|
||||
db,
|
||||
runId: 'run-without-usage',
|
||||
projectId: 'project-1',
|
||||
conversationId: 'conv-A',
|
||||
snapshot: snap,
|
||||
pipeline: snap.pipeline!,
|
||||
env: { maxIterations: 1 },
|
||||
runStage: async ({ stage, iteration, snapshot: snap2 }) => runStageWithRegistry({
|
||||
db,
|
||||
runId: 'run-without-usage',
|
||||
projectId: 'project-1',
|
||||
conversationId: 'conv-A',
|
||||
stage,
|
||||
iteration,
|
||||
snapshot: snap2,
|
||||
runEvents: [],
|
||||
}),
|
||||
});
|
||||
|
||||
expect(listIterationsForRun(db, 'run-with-usage')[0]?.tokensUsed).toBe(45);
|
||||
expect(listIterationsForRun(db, 'run-without-usage')[0]?.tokensUsed).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe('pipeline-runner: GenUI cross-conversation cache (e2e-5)', () => {
|
||||
|
||||
Reference in New Issue
Block a user