Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions packages/cli/src/acp-integration/acpAgent.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6153,6 +6153,16 @@ class QwenAgent implements Agent {
sendUpdate: async (update) => {
updates.push(update);
},
// Fresh accumulator for this replay: MessageEmitter advances it from
// replayed usage metadata (tokens only — no per-turn durations) and
// PlanEmitter snapshots it onto each todo update, so resumed sessions
// recover per-task token spend (API time stays live-only).
cumulativeUsage: {
promptTokens: 0,
cachedTokens: 0,
candidateTokens: 0,
apiTimeMs: 0,
},
};
let replayError: string | undefined;
try {
Expand Down
11 changes: 11 additions & 0 deletions packages/cli/src/acp-integration/session/Session.ts
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,7 @@ import { getPersistScopeForModelSelection } from '../../config/modelProvidersSco
// Import modular session components
import type {
ApprovalModeValue,
CumulativeUsage,
SessionContext,
ToolCallStartParams,
} from './types.js';
Expand Down Expand Up @@ -417,6 +418,16 @@ export class Session implements SessionContext {
private followupAbort: AbortController | null = null;
private turn: number = 0;
private readonly createdAt: number = Date.now();
/**
* Running cumulative usage for this session, snapshotted onto each todo/plan
* update by PlanEmitter so the web-shell can show per-task token/API spend.
*/
readonly cumulativeUsage: CumulativeUsage = {
promptTokens: 0,
cachedTokens: 0,
candidateTokens: 0,
apiTimeMs: 0,
};
private readonly runtimeBaseDir: string;

// Cron scheduling state
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -278,5 +278,73 @@ describe('MessageEmitter', () => {
},
});
});

it('accumulates token counts and API time into the context cumulative usage', async () => {
const cumulativeUsage = {
promptTokens: 0,
cachedTokens: 0,
candidateTokens: 0,
apiTimeMs: 0,
};
const ctx: SessionContext = {
sessionId: 'test-session-id',
config: {} as Config,
sendUpdate: sendUpdateSpy,
cumulativeUsage,
};
const e = new MessageEmitter(ctx);
await e.emitUsageMetadata(
{
promptTokenCount: 100,
candidatesTokenCount: 50,
cachedContentTokenCount: 10,
},
'',
800,
);
await e.emitUsageMetadata(
{
promptTokenCount: 30,
candidatesTokenCount: 20,
cachedContentTokenCount: 5,
},
'',
200,
);

expect(cumulativeUsage).toEqual({
promptTokens: 130,
cachedTokens: 15,
candidateTokens: 70,
apiTimeMs: 1000,
});
});

it('accumulates tokens but not API time when no duration is provided (replay)', async () => {
const cumulativeUsage = {
promptTokens: 0,
cachedTokens: 0,
candidateTokens: 0,
apiTimeMs: 0,
};
const ctx: SessionContext = {
sessionId: 'test-session-id',
config: {} as Config,
sendUpdate: sendUpdateSpy,
cumulativeUsage,
};
await new MessageEmitter(ctx).emitUsageMetadata({
promptTokenCount: 100,
candidatesTokenCount: 50,
cachedContentTokenCount: 10,
});

expect(cumulativeUsage).toEqual({
promptTokens: 100,
cachedTokens: 10,
candidateTokens: 50,
apiTimeMs: 0,
});
});
});
});
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,19 @@ export class MessageEmitter extends BaseEmitter {
cachedReadTokens: usageMetadata.cachedContentTokenCount,
};

// Advance the running cumulative usage so PlanEmitter can snapshot it onto
// the next todo update. apiTimeMs only moves when a per-turn duration is
// present (live), keeping API time live-only on replay by design.
const cumulative = this.ctx.cumulativeUsage;
if (cumulative) {
cumulative.promptTokens += usage.inputTokens;
cumulative.candidateTokens += usage.outputTokens;
cumulative.cachedTokens += usage.cachedReadTokens ?? 0;
if (typeof durationMs === 'number') {
Comment thread
wenshao marked this conversation as resolved.
Resolved
cumulative.apiTimeMs += durationMs;
}
}

const meta =
typeof durationMs === 'number'
? { usage, durationMs, ...subagentMeta }
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,37 @@ describe('PlanEmitter', () => {
});
});

it('omits _meta.stats when the context has no cumulative usage', async () => {
await emitter.emitPlan([{ id: '1', content: 'Task', status: 'pending' }]);

const update = sendUpdateSpy.mock.calls[0][0];
expect(update['_meta']).toBeUndefined();
});

it('stamps a copy of the cumulative usage on _meta.stats when present', async () => {
const cumulativeUsage = {
promptTokens: 100,
cachedTokens: 10,
candidateTokens: 20,
apiTimeMs: 500,
};
const ctx: SessionContext = {
sessionId: 'test-session-id',
config: {} as Config,
sendUpdate: sendUpdateSpy,
cumulativeUsage,
};
await new PlanEmitter(ctx).emitPlan([
{ id: '1', content: 'Task', status: 'completed' },
]);

const update = sendUpdateSpy.mock.calls[0][0];
expect(update['_meta']).toEqual({ stats: { ...cumulativeUsage } });
// Snapshot is a copy: later accumulation must not mutate what was sent.
cumulativeUsage.promptTokens = 999;
expect(update['_meta'].stats.promptTokens).toBe(100);
});

it('should set default priority to medium for all entries', async () => {
const todos: TodoItem[] = [
{ id: '1', content: 'Task', status: 'pending' },
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,9 +28,15 @@ export class PlanEmitter extends BaseEmitter {
status: todo.status,
}));

// Snapshot the running cumulative usage as a per-snapshot baseline. The
// web-shell diffs consecutive snapshots to attribute tokens/API time to the
// task that ran between two todo updates. Copied so later accumulation
// doesn't mutate this snapshot.
const cumulative = this.ctx.cumulativeUsage;
Comment thread
wenshao marked this conversation as resolved.
Resolved
await this.sendUpdate({
sessionUpdate: 'plan',
entries,
...(cumulative ? { _meta: { stats: { ...cumulative } } } : {}),
});
}

Expand Down
25 changes: 25 additions & 0 deletions packages/cli/src/acp-integration/session/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,24 @@ export interface SessionUpdateSender {
sendUpdate(update: SessionUpdate): Promise<void>;
}

/**
* Running cumulative usage for the conversation, mutated in place as usage
* metadata is emitted (MessageEmitter) and snapshotted onto each plan/todo
* update (PlanEmitter). The web-shell diffs consecutive snapshots to show a
* finished task's token/time spend.
*
* `apiTimeMs` only advances on the live path: history replay re-emits usage
* metadata without per-turn durations, so on `/resume` it stays 0 — the
* intended "API time is live-only" behaviour. Tokens accumulate on both paths
* because replayed usage metadata carries the counts.
*/
export interface CumulativeUsage {
promptTokens: number;
cachedTokens: number;
candidateTokens: number;
apiTimeMs: number;
}

/**
* Session context shared across all emitters.
* Provides access to session state and configuration.
Expand All @@ -38,6 +56,13 @@ export interface SessionContext extends SessionUpdateSender {
/** Optional message rewrite middleware for ACP message transformation.
* Installed after history replay to avoid rewriting historical messages. */
messageRewriter?: MessageRewriteMiddleware;
/**
* Running cumulative usage, when the context wants per-todo resource detail.
* Mutated by MessageEmitter as usage is emitted and read by PlanEmitter to
* stamp plan updates. Optional so contexts that don't need it (export, etc.)
* can omit it.
*/
readonly cumulativeUsage?: CumulativeUsage;
}

/**
Expand Down
7 changes: 6 additions & 1 deletion packages/sdk-typescript/src/daemon/ui/normalizer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -654,6 +654,11 @@ function normalizePlanUpdate(
base.eventId !== undefined
? `${DAEMON_PLAN_TOOL_CALL_ID}-${base.eventId}`
: DAEMON_PLAN_TOOL_CALL_ID;
// Carry the cumulative-usage snapshot the agent stamps on each plan update
// (PlanEmitter) through to rawOutput, so the web-shell can diff consecutive
// todo snapshots into per-task token/time detail.
const meta = isRecord(update['_meta']) ? update['_meta'] : undefined;
const stats = meta && isRecord(meta['stats']) ? meta['stats'] : undefined;
return {
...base,
type: 'tool.update',
Expand All @@ -668,7 +673,7 @@ function normalizePlanUpdate(
content: { type: 'text', text: contentText },
},
],
rawOutput: { entries },
rawOutput: stats ? { entries, stats } : { entries },
};
}

Expand Down
52 changes: 52 additions & 0 deletions packages/sdk-typescript/test/unit/daemonUi.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,58 @@ describe('daemon UI normalizer and transcript reducer', () => {
]);
});

it('passes the agent-stamped plan stats snapshot through to rawOutput', () => {
const events = normalizeDaemonEvent({
id: 5,
v: 1,
type: 'session_update',
data: {
update: {
sessionUpdate: 'plan',
entries: [
{ content: 'Task', status: 'completed', priority: 'medium' },
],
_meta: {
stats: {
promptTokens: 100,
cachedTokens: 10,
candidateTokens: 20,
apiTimeMs: 500,
},
},
},
},
});

expect(events).toHaveLength(1);
expect(events[0]).toMatchObject({
type: 'tool.update',
toolName: 'TodoWrite',
rawOutput: {
entries: [{ content: 'Task', status: 'completed', priority: 'medium' }],
stats: {
promptTokens: 100,
cachedTokens: 10,
candidateTokens: 20,
apiTimeMs: 500,
},
},
});
});

it('omits stats from a plan rawOutput when the update carries none', () => {
const events = normalizeDaemonEvent({
id: 6,
v: 1,
type: 'session_update',
data: { update: { sessionUpdate: 'plan', entries: [] } },
});

const rawOutput = (events[0] as { rawOutput: Record<string, unknown> })
.rawOutput;
expect(rawOutput).toEqual({ entries: [] });
});

it('keeps optimistic local user blocks before daemon replies when sorting', () => {
let state = createDaemonTranscriptState({ now: 1 });
state = appendLocalUserTranscriptMessage(state, 'hello', { now: 10 });
Expand Down
Loading
Loading