From 6ac4743618d647f584685057937af943fc9d0ee1 Mon Sep 17 00:00:00 2001 From: Jasper Date: Fri, 2 Oct 2026 01:21:16 -0400 Subject: [PATCH 01/13] fix(usage): account for tool-ended inference turns Tool results with endTurn bypass the natural complete event, leaving usage totals and completion observers unaware of completed work. Register a stream-scoped, idempotent accounting callback that both terminal paths use. Finalize tool-ended turns after resetting the agent and before asynchronous cancellation can dispose an ephemeral or hand control to another stream. Retain the latest cumulative Membrane usage snapshot rather than summing round snapshots. Preserve cache counts and provider-priced estimated cost. Natural completion continues to use response.details.usage as its authoritative total. Cancellation and MCPL lifecycle behavior stay unchanged. Added five integration checks for one and multiple tool rounds, natural completion, attribution and persistence through reopen, and missing usage. On the original 03c31d9 source these yielded 1 pass / 4 fail; the fix passes all five. Built compiled output and ran the new suite plus ephemeral-settle and deferred-wake-integrity: 18 pass / 0 fail. Bounded full-suite comparison is running separately. Co-Authored-By: GPT-6 Astra --- CHANGELOG.md | 4 + src/framework.ts | 58 +++++++---- test/end-turn-usage.test.ts | 196 ++++++++++++++++++++++++++++++++++++ 3 files changed, 241 insertions(+), 17 deletions(-) create mode 100644 test/end-turn-usage.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b9a30b4c..14488fad5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,10 @@ Releases up to and including 0.7.3 predate this file; for their contents see ## Unreleased +### Fixed + +- Tool-ended turns now emit `inference:completed` and record cumulative token, cache, and estimated-cost usage exactly once, including turns ended by `skip_reply` or other `endTurn` tools. + ## 0.19.0 — 2026-09-28 ### Added diff --git a/src/framework.ts b/src/framework.ts index 510b9183a..b2c66c876 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -3,7 +3,7 @@ import { INLINE_WITHHELD_TEXT, classifyBlock, isInlineContradiction, referenceRe import { ReferenceFetcher, DEFAULT_FETCH_MAX_BYTES, EAGER_FETCH_TIMEOUT_MS } from './mcpl/reference-fetcher.js'; import { appendFileSync, existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from 'node:fs'; import { JsStore } from '@animalabs/chronicle'; -import type { Membrane, ContentBlock, NormalizedRequest, YieldingStream, ToolResult as MembraneToolResult, ToolResultContentBlock } from '@animalabs/membrane'; +import type { Membrane, ContentBlock, NormalizedRequest, YieldingStream, DetailedUsage, ToolResult as MembraneToolResult, ToolResultContentBlock } from '@animalabs/membrane'; import { MembraneError } from '@animalabs/membrane'; import { ContextManager, PassthroughStrategy, WindowedPassthroughStrategy, OverBudgetError, UncoveredDropError } from '@animalabs/context-manager'; import type { CacheWireReceipt } from './kv-unified-wire.js'; @@ -1002,6 +1002,10 @@ export class AgentFramework { private processLoggingPersist: boolean; private processLoggingBroadcast: boolean; private activeStreams: Map> = new Map(); + // Tool results settle outside driveStream. Key accounting by the physical + // stream so an endTurn result can finalize it before cancellation teardown + // or ephemeral disposal, without touching a successor's usage. + private streamCompletionAccounting = new WeakMap void>(); /** Per-agent output locus FROZEN for the CURRENT logical turn. Resolved * eagerly in startAgentStream (home → addressed trigger → global default) @@ -6635,6 +6639,9 @@ export class AgentFramework { // driveStream's finally is the backstop.) this.eventGate?.onInferenceEnded(agent.name); this.settleAgent(agent.name, { stopReason: 'turn_ended', speech: '' }); + if (currentState.stream) { + this.streamCompletionAccounting.get(currentState.stream)?.(); + } this.emitTrace({ type: 'inference:turn_ended', agentName: agent.name }); } else if (overBudget || overPhysical) { // Context budget exceeded: break the stream, let compile() compress. @@ -9072,6 +9079,31 @@ export class AgentFramework { const startTime = Date.now(); const requestId = `${agent.name}-${startTime}-${Math.random().toString(36).slice(2, 8)}`; const myStreamId = agent.streamId; + let latestUsage: DetailedUsage | undefined; + let completionAccounted = false; + const completeAccounting = (usage: DetailedUsage | undefined): void => { + if (completionAccounted) return; + completionAccounted = true; + this.streamCompletionAccounting.delete(stream); + this.emitTrace({ + type: 'inference:completed', + agentName: agent.name, + durationMs: Date.now() - startTime, + tokenUsage: usage ? { + input: usage.inputTokens, + output: usage.outputTokens, + cacheCreation: usage.cacheCreationTokens, + cacheRead: usage.cacheReadTokens, + } : undefined, + }); + if (usage) { + this.usageTracker.onInferenceCompleted(agent.name, usage, usage.estimatedCost + ? { total: usage.estimatedCost.total, currency: usage.estimatedCost.currency } + : undefined); + this.persistUsageState(); + } + }; + this.streamCompletionAccounting.set(stream, () => completeAccounting(latestUsage)); // Membrane usage events are cumulative across the native/XML tool loop. // Keep the previous cumulative sample so consumers that operate at the // physical provider-call boundary (estimator calibration and kv receipt @@ -9609,22 +9641,7 @@ export class AgentFramework { .join('\n'), }); - this.emitTrace({ - type: 'inference:completed', - agentName: agent.name, - durationMs, - tokenUsage, - }); - - if (du) { - this.usageTracker.onInferenceCompleted(agent.name, { - inputTokens: du.inputTokens, - outputTokens: du.outputTokens, - cacheCreationTokens: du.cacheCreationTokens, - cacheReadTokens: du.cacheReadTokens, - }, du.estimatedCost ? { total: du.estimatedCost.total, currency: du.estimatedCost.currency } : undefined); - this.persistUsageState(); - } + completeAccounting(du); // Log inference this.logInference({ @@ -10144,6 +10161,12 @@ export class AgentFramework { } case 'usage': { + // Already cumulative across rounds: replace, rather than sum, and + // retain Membrane's cost at the rates that served those rounds. + latestUsage = { + ...event.usage, + estimatedCost: event.usage.estimatedCost ? { ...event.usage.estimatedCost } : undefined, + }; agent.lastStreamInputTokens = event.usage.inputTokens; agent.lastStreamRealInputTokens = (event.usage.inputTokens ?? 0) + @@ -10361,6 +10384,7 @@ export class AgentFramework { } } this.frameworkCancelledStreams.delete(`${agent.name}:${myStreamId}`); + this.streamCompletionAccounting.delete(stream); if (ownsPhysicalStream) { this.activeStreams.delete(agent.name); this.pendingAssistantBlocks.delete(agent.name); diff --git a/test/end-turn-usage.test.ts b/test/end-turn-usage.test.ts new file mode 100644 index 000000000..eb4e21451 --- /dev/null +++ b/test/end-turn-usage.test.ts @@ -0,0 +1,196 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import type { DetailedUsage, NormalizedResponse } from '@animalabs/membrane'; +import { AgentFramework } from '../src/index.js'; +import type { Module, ToolCall, TraceEvent } from '../src/index.js'; +import { createMockResponse, MockMembrane } from './helpers/mock-membrane.js'; + +class TurnTools implements Module { + readonly name = 'test'; + async start() {} + async stop() {} + async onProcess() { return {}; } + getTools() { + return ['echo', 'finish'].map((name) => ({ + name, + description: name, + inputSchema: { type: 'object' as const, properties: {} }, + })); + } + async handleToolCall(call: ToolCall) { + return { success: true, endTurn: call.name.endsWith('finish'), data: {} }; + } +} + +const firstUsage: DetailedUsage = { + inputTokens: 100, outputTokens: 20, cacheCreationTokens: 30, cacheReadTokens: 40, + estimatedCost: { input: 0.01, output: 0.02, total: 0.03, currency: 'USD' }, +}; +const totalUsage: DetailedUsage = { + inputTokens: 250, outputTokens: 55, cacheCreationTokens: 45, cacheReadTokens: 90, + estimatedCost: { input: 0.03, output: 0.05, total: 0.08, currency: 'USD' }, +}; + +function toolResponse(name: string, id: string, usage?: DetailedUsage): NormalizedResponse { + const response = createMockResponse([{ type: 'tool_use', name: 'test--' + name, id, input: {} }], 'tool_use'); + // Membrane emits cumulative usage across all rounds of a yielding stream. + response.usage = usage!; + return response; +} + +function expectedTokens(usage: DetailedUsage) { + return { + input: usage.inputTokens, output: usage.outputTokens, + cacheCreation: usage.cacheCreationTokens, cacheRead: usage.cacheReadTokens, + }; +} + +function expectedTotals(usage: DetailedUsage) { + return { + inputTokens: usage.inputTokens, outputTokens: usage.outputTokens, + cacheCreationTokens: usage.cacheCreationTokens ?? 0, + cacheReadTokens: usage.cacheReadTokens ?? 0, + estimatedCost: usage.estimatedCost + ? { total: usage.estimatedCost.total, currency: usage.estimatedCost.currency } + : undefined, + }; +} + +async function fixture() { + const dir = mkdtempSync(join(tmpdir(), 'af-end-turn-usage-')); + const storePath = join(dir, 'store.chronicle'); + const membrane = new MockMembrane(); + const framework = await AgentFramework.create({ + storePath, membrane: membrane.asMembrane(), agents: [], + modules: [new TurnTools()], syncIntervalMs: 0, + }); + const traces: TraceEvent[] = []; + const completionStates: Array = []; + framework.onTrace((event) => { + traces.push(event); + if (event.type === 'inference:completed') { + completionStates.push(framework.getAgent(event.agentName)?.state.status); + } + }); + framework.start(); + return { dir, storePath, membrane, framework, traces, completionStates }; +} + +async function run(framework: AgentFramework, name: string) { + const { agent, contextManager } = await framework.createEphemeralAgent({ + name, model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', + proseRouting: 'disabled', + }); + contextManager.addMessage('user', [{ type: 'text', text: 'Run once.' }]); + const result = await framework.runEphemeralToCompletion(agent, contextManager); + // Also observe asynchronous cancellation teardown before checking counts. + await new Promise((resolve) => setImmediate(resolve)); + return result; +} + +describe('tool-ended turn usage accounting', () => { + for (const rounds of [1, 2]) { + it('accounts once for ' + rounds + ' tool round(s), with cumulative cache usage and cost', async () => { + const f = await fixture(); + try { + if (rounds === 2) f.membrane.pushResponse(toolResponse('echo', 'echo-1', firstUsage)); + f.membrane.pushResponse(toolResponse('finish', 'finish-1', totalUsage)); + const result = await run(f.framework, 'worker'); + assert.equal(result.toolCallsCount, rounds); + assert.equal(f.membrane.calls.length, 1, 'endTurn must not request another inference'); + assert.equal(f.membrane.lastStream!.receivedToolResults.length, rounds - 1); + assert.deepEqual(f.completionStates, ['idle']); + const completed = f.traces.filter((e) => e.type === 'inference:completed'); + assert.equal(completed.length, 1); + assert.equal(completed[0].agentName, 'worker'); + assert.deepEqual(completed[0].tokenUsage, expectedTokens(totalUsage)); + assert.ok(completed[0].durationMs >= 0); + assert.equal(f.traces.filter((e) => e.type === 'inference:turn_ended').length, 1); + assert.equal(f.traces.filter((e) => e.type === 'usage:updated').length, 1); + assert.equal(f.traces.filter((e) => e.type === 'inference:exhausted').length, 0); + const snapshot = f.framework.getSessionUsage(); + assert.equal(snapshot.inferenceCount, 1); + assert.deepEqual(snapshot.totals, expectedTotals(totalUsage)); + assert.deepEqual(snapshot.byAgent, [{ + agentName: 'worker', usage: expectedTotals(totalUsage), inferenceCount: 1, + }]); + } finally { + await f.framework.stop(); + rmSync(f.dir, { recursive: true, force: true }); + } + }); + } + + it('keeps natural completion authoritative and counts neither cumulative samples nor final usage twice', async () => { + const f = await fixture(); + try { + f.membrane.pushResponse(toolResponse('echo', 'echo-1', firstUsage)); + const response = createMockResponse([{ type: 'text', text: 'Done.' }]); + response.usage = { ...totalUsage, outputTokens: 50 }; // final details remain authoritative + response.details!.usage = totalUsage; + f.membrane.pushResponse(response); + await run(f.framework, 'ordinary'); + assert.deepEqual(f.completionStates, ['idle']); + const completed = f.traces.filter((e) => e.type === 'inference:completed'); + assert.equal(completed.length, 1); + assert.deepEqual(completed[0].tokenUsage, expectedTokens(totalUsage)); + assert.equal(f.traces.filter((e) => e.type === 'usage:updated').length, 1); + assert.equal(f.traces.filter((e) => e.type === 'inference:turn_ended').length, 0); + assert.equal(f.framework.getSessionUsage().inferenceCount, 1); + assert.deepEqual(f.framework.getSessionUsage().totals, expectedTotals(totalUsage)); + } finally { + await f.framework.stop(); + rmSync(f.dir, { recursive: true, force: true }); + } + }); + + it('attributes successive streams separately and persists their totals through store reopen', async () => { + const f = await fixture(); + let stopped = false; + try { + f.membrane.pushResponse(toolResponse('finish', 'finish-a', firstUsage)); + await run(f.framework, 'alice'); + f.membrane.pushResponse(toolResponse('finish', 'finish-b', totalUsage)); + await run(f.framework, 'bob'); + const snapshot = f.framework.getSessionUsage(); + assert.equal(snapshot.inferenceCount, 2); + assert.deepEqual(snapshot.byAgent, [ + { agentName: 'alice', usage: expectedTotals(firstUsage), inferenceCount: 1 }, + { agentName: 'bob', usage: expectedTotals(totalUsage), inferenceCount: 1 }, + ]); + assert.deepEqual(f.traces.filter((e) => e.type === 'usage:updated').map((e) => e.agentName), ['alice', 'bob']); + await f.framework.stop(); + stopped = true; + const reopened = await AgentFramework.create({ + storePath: f.storePath, membrane: f.membrane.asMembrane(), agents: [], modules: [], syncIntervalMs: 0, + }); + try { + assert.deepEqual(reopened.getSessionUsage(), snapshot); + } finally { + await reopened.stop(); + } + } finally { + if (!stopped) await f.framework.stop(); + rmSync(f.dir, { recursive: true, force: true }); + } + }); + + it('reports completion without inventing usage when no sample is available', async () => { + const f = await fixture(); + try { + f.membrane.pushResponse(toolResponse('finish', 'finish-1')); + await run(f.framework, 'unknown-usage'); + const completed = f.traces.filter((e) => e.type === 'inference:completed'); + assert.equal(completed.length, 1); + assert.equal(completed[0].tokenUsage, undefined); + assert.equal(f.traces.filter((e) => e.type === 'usage:updated').length, 0); + assert.equal(f.framework.getSessionUsage().inferenceCount, 0); + } finally { + await f.framework.stop(); + rmSync(f.dir, { recursive: true, force: true }); + } + }); +}); From b71587726b9ba024b40d4f7aa50afb58d268193a Mon Sep 17 00:00:00 2001 From: Jasper Date: Fri, 2 Oct 2026 02:04:05 -0400 Subject: [PATCH 02/13] fix(usage): log tool-ended completions for health accounting Address the source-confirmed Greptile finding on PR #207: restored usage/completion telemetry still bypassed the successful inference log, which HealthModule uses for inference and token totals. Write the terminal log inside the same per-stream idempotent completion callback. Natural completion retains its raw provider response, stop reason, and duration; tool-ended completion records turn_ended and an explicit note that the stream ended by a tool result rather than inventing a final provider response. Missing usage remains absent. Extend regression coverage for one log per one/multiple-tool-round stream, normal completion without duplication, agent attribution, unique request IDs, store reopen, missing usage, and the real log-backed HealthModule snapshot. On 6ac4743 the six-case suite gives 1 pass / 5 fail; the fix passes 6/6. Expanded compiled focused tests pass 32/32 without force-exit. The revised full compiled suite is running separately before push. Co-Authored-By: GPT-6 Astra --- CHANGELOG.md | 2 +- src/framework.ts | 52 +++++++++++++++++++++---------------- src/types/framework.ts | 2 +- test/end-turn-usage.test.ts | 49 ++++++++++++++++++++++++++++++++++ 4 files changed, 80 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 14488fad5..66f48f1a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,7 @@ Releases up to and including 0.7.3 predate this file; for their contents see ### Fixed -- Tool-ended turns now emit `inference:completed` and record cumulative token, cache, and estimated-cost usage exactly once, including turns ended by `skip_reply` or other `endTurn` tools. +- Tool-ended turns now emit `inference:completed`, record a successful inference-log entry, and record cumulative token, cache, and estimated-cost usage exactly once, including turns ended by `skip_reply` or other `endTurn` tools. Log-backed health totals now include these turns. ## 0.19.0 — 2026-09-28 diff --git a/src/framework.ts b/src/framework.ts index b2c66c876..4a9cc7757 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -3,7 +3,7 @@ import { INLINE_WITHHELD_TEXT, classifyBlock, isInlineContradiction, referenceRe import { ReferenceFetcher, DEFAULT_FETCH_MAX_BYTES, EAGER_FETCH_TIMEOUT_MS } from './mcpl/reference-fetcher.js'; import { appendFileSync, existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from 'node:fs'; import { JsStore } from '@animalabs/chronicle'; -import type { Membrane, ContentBlock, NormalizedRequest, YieldingStream, DetailedUsage, ToolResult as MembraneToolResult, ToolResultContentBlock } from '@animalabs/membrane'; +import type { Membrane, ContentBlock, NormalizedRequest, NormalizedResponse, YieldingStream, DetailedUsage, ToolResult as MembraneToolResult, ToolResultContentBlock } from '@animalabs/membrane'; import { MembraneError } from '@animalabs/membrane'; import { ContextManager, PassthroughStrategy, WindowedPassthroughStrategy, OverBudgetError, UncoveredDropError } from '@animalabs/context-manager'; import type { CacheWireReceipt } from './kv-unified-wire.js'; @@ -9081,20 +9081,25 @@ export class AgentFramework { const myStreamId = agent.streamId; let latestUsage: DetailedUsage | undefined; let completionAccounted = false; - const completeAccounting = (usage: DetailedUsage | undefined): void => { + const completeAccounting = ( + usage: DetailedUsage | undefined, + response?: NormalizedResponse, + durationMs = Date.now() - startTime, + ): void => { if (completionAccounted) return; completionAccounted = true; this.streamCompletionAccounting.delete(stream); + const tokenUsage = usage ? { + input: usage.inputTokens, + output: usage.outputTokens, + cacheCreation: usage.cacheCreationTokens, + cacheRead: usage.cacheReadTokens, + } : undefined; this.emitTrace({ type: 'inference:completed', agentName: agent.name, - durationMs: Date.now() - startTime, - tokenUsage: usage ? { - input: usage.inputTokens, - output: usage.outputTokens, - cacheCreation: usage.cacheCreationTokens, - cacheRead: usage.cacheReadTokens, - } : undefined, + durationMs, + tokenUsage, }); if (usage) { this.usageTracker.onInferenceCompleted(agent.name, usage, usage.estimatedCost @@ -9102,6 +9107,20 @@ export class AgentFramework { : undefined); this.persistUsageState(); } + // Health snapshots and log queries read this same terminal accounting. + // A tool-ended stream has no final provider response; record that fact + // rather than fabricating response content or waiting for cancellation. + this.logInference({ + timestamp: startTime, + agentName: agent.name, + requestId, + success: true, + request: compiledRequest ?? { note: 'streaming request' }, + response: response?.raw ?? { note: response ? 'streaming response' : 'stream ended by tool result' }, + durationMs, + tokenUsage, + stopReason: response?.stopReason ?? 'turn_ended', + }); }; this.streamCompletionAccounting.set(stream, () => completeAccounting(latestUsage)); // Membrane usage events are cumulative across the native/XML tool loop. @@ -9641,20 +9660,7 @@ export class AgentFramework { .join('\n'), }); - completeAccounting(du); - - // Log inference - this.logInference({ - timestamp: startTime, - agentName: agent.name, - requestId, - success: true, - request: compiledRequest ?? { note: 'streaming request' }, - response: response.raw ?? { note: 'streaming response' }, - durationMs, - tokenUsage, - stopReason: response.stopReason, - }); + completeAccounting(du, response, durationMs); // Surface refusals instead of going silently mute: stderr line // (headless inference failures are otherwise under-logged) + an diff --git a/src/types/framework.ts b/src/types/framework.ts index 3ff4ba9dc..8d4ee3636 100644 --- a/src/types/framework.ts +++ b/src/types/framework.ts @@ -396,7 +396,7 @@ export interface InferenceLogEntry { cacheCreation?: number; cacheRead?: number; }; - /** Stop reason from the model */ + /** Stop reason from the model, or 'turn_ended' for a tool-ended stream. */ stopReason?: string; } diff --git a/test/end-turn-usage.test.ts b/test/end-turn-usage.test.ts index eb4e21451..276c6dd63 100644 --- a/test/end-turn-usage.test.ts +++ b/test/end-turn-usage.test.ts @@ -5,6 +5,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import type { DetailedUsage, NormalizedResponse } from '@animalabs/membrane'; import { AgentFramework } from '../src/index.js'; +import { HealthModule } from '../src/modules/health/index.js'; import type { Module, ToolCall, TraceEvent } from '../src/index.js'; import { createMockResponse, MockMembrane } from './helpers/mock-membrane.js'; @@ -108,6 +109,13 @@ describe('tool-ended turn usage accounting', () => { assert.equal(completed[0].agentName, 'worker'); assert.deepEqual(completed[0].tokenUsage, expectedTokens(totalUsage)); assert.ok(completed[0].durationMs >= 0); + const logs = f.framework.queryInferenceLogs({ agentName: 'worker' }); + assert.equal(logs.total, 1); + assert.equal(logs.entries[0].entry.success, true); + assert.equal(logs.entries[0].entry.stopReason, 'turn_ended'); + assert.deepEqual(logs.entries[0].entry.tokenUsage, expectedTokens(totalUsage)); + assert.deepEqual(logs.entries[0].entry.response, { note: 'stream ended by tool result' }); + assert.equal(logs.entries[0].entry.durationMs, completed[0].durationMs); assert.equal(f.traces.filter((e) => e.type === 'inference:turn_ended').length, 1); assert.equal(f.traces.filter((e) => e.type === 'usage:updated').length, 1); assert.equal(f.traces.filter((e) => e.type === 'inference:exhausted').length, 0); @@ -141,6 +149,10 @@ describe('tool-ended turn usage accounting', () => { assert.equal(f.traces.filter((e) => e.type === 'inference:turn_ended').length, 0); assert.equal(f.framework.getSessionUsage().inferenceCount, 1); assert.deepEqual(f.framework.getSessionUsage().totals, expectedTotals(totalUsage)); + const logs = f.framework.queryInferenceLogs({ agentName: 'ordinary' }); + assert.equal(logs.total, 1, 'natural completion logs exactly once'); + assert.equal(logs.entries[0].entry.stopReason, 'end_turn'); + assert.deepEqual(logs.entries[0].entry.tokenUsage, expectedTokens(totalUsage)); } finally { await f.framework.stop(); rmSync(f.dir, { recursive: true, force: true }); @@ -162,6 +174,10 @@ describe('tool-ended turn usage accounting', () => { { agentName: 'bob', usage: expectedTotals(totalUsage), inferenceCount: 1 }, ]); assert.deepEqual(f.traces.filter((e) => e.type === 'usage:updated').map((e) => e.agentName), ['alice', 'bob']); + const logs = f.framework.queryInferenceLogs(); + assert.equal(logs.total, 2); + assert.equal(new Set(logs.entries.map((e) => e.entry.requestId)).size, 2); + assert.deepEqual(logs.entries.map((e) => e.entry.agentName).sort(), ['alice', 'bob']); await f.framework.stop(); stopped = true; const reopened = await AgentFramework.create({ @@ -169,6 +185,7 @@ describe('tool-ended turn usage accounting', () => { }); try { assert.deepEqual(reopened.getSessionUsage(), snapshot); + assert.deepEqual(reopened.queryInferenceLogs(), logs); } finally { await reopened.stop(); } @@ -178,6 +195,34 @@ describe('tool-ended turn usage accounting', () => { } }); + it('includes tool-ended turns in the log-backed health snapshot', async () => { + const f = await fixture(); + try { + f.membrane.pushResponse(toolResponse('echo', 'echo-1', firstUsage)); + f.membrane.pushResponse(toolResponse('finish', 'finish-1', totalUsage)); + await run(f.framework, 'worker'); + const health = new HealthModule({ timeZone: 'UTC' }); + health.bind(f.framework, f.framework.getStore()); + const result = await health.handleToolCall({ + id: 'health-snapshot', name: 'snapshot', input: { includeSubagents: false }, + }); + assert.equal(result.success, true); + const data = result.data as { + inferences: { successCount: number; errorCount: number }; + tokenTotalsByAgent: Record; + }; + assert.equal(data.inferences.successCount, 1); + assert.equal(data.inferences.errorCount, 0); + assert.deepEqual(data.tokenTotalsByAgent.worker, { + input: totalUsage.inputTokens, output: totalUsage.outputTokens, + cacheRead: totalUsage.cacheReadTokens, inferences: 1, + }); + } finally { + await f.framework.stop(); + rmSync(f.dir, { recursive: true, force: true }); + } + }); + it('reports completion without inventing usage when no sample is available', async () => { const f = await fixture(); try { @@ -188,6 +233,10 @@ describe('tool-ended turn usage accounting', () => { assert.equal(completed[0].tokenUsage, undefined); assert.equal(f.traces.filter((e) => e.type === 'usage:updated').length, 0); assert.equal(f.framework.getSessionUsage().inferenceCount, 0); + const logs = f.framework.queryInferenceLogs({ agentName: 'unknown-usage' }); + assert.equal(logs.total, 1); + assert.equal(logs.entries[0].entry.success, true); + assert.equal(logs.entries[0].entry.tokenUsage, undefined); } finally { await f.framework.stop(); rmSync(f.dir, { recursive: true, force: true }); From 099efc9d4aba13655b456c98b970cd12aef4b636 Mon Sep 17 00:00:00 2001 From: Ada-1017 Date: Fri, 9 Oct 2026 22:34:48 -0400 Subject: [PATCH 03/13] chore(changelog): move #67's entry into a changelog.d fragment The branch edited CHANGELOG.md's Unreleased section directly, which conflicts with main (02adf27) now that main's own entries live in changelog.d/ (CONTRIBUTING.md#changelog). Same text, moved verbatim into changelog.d/fix-67-end-turn-accounting.fixed.md; CHANGELOG.md is back to the merge base's version, so the branch no longer touches it. --- CHANGELOG.md | 4 ---- changelog.d/fix-67-end-turn-accounting.fixed.md | 1 + 2 files changed, 1 insertion(+), 4 deletions(-) create mode 100644 changelog.d/fix-67-end-turn-accounting.fixed.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 66f48f1a4..6b9a30b4c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,10 +12,6 @@ Releases up to and including 0.7.3 predate this file; for their contents see ## Unreleased -### Fixed - -- Tool-ended turns now emit `inference:completed`, record a successful inference-log entry, and record cumulative token, cache, and estimated-cost usage exactly once, including turns ended by `skip_reply` or other `endTurn` tools. Log-backed health totals now include these turns. - ## 0.19.0 — 2026-09-28 ### Added diff --git a/changelog.d/fix-67-end-turn-accounting.fixed.md b/changelog.d/fix-67-end-turn-accounting.fixed.md new file mode 100644 index 000000000..58a9b2d58 --- /dev/null +++ b/changelog.d/fix-67-end-turn-accounting.fixed.md @@ -0,0 +1 @@ +- Tool-ended turns now emit `inference:completed`, record a successful inference-log entry, and record cumulative token, cache, and estimated-cost usage exactly once, including turns ended by `skip_reply` or other `endTurn` tools. Log-backed health totals now include these turns. From 344770608442cd4e7c12d602786ac85a3554a4f8 Mon Sep 17 00:00:00 2001 From: Ada-1017 Date: Fri, 9 Oct 2026 23:59:37 -0400 Subject: [PATCH 04/13] fix(usage): count a stream's usage once however it ends; a tool-ended turn ends as a completion does Skill pass on #207 (house review of Jasper-314's 6ac4743/b715877, now on current main via 4484548). The rule #207 set at one site (a tool-ended stream's usage reaches the session tracker) holds at its siblings now: - A context-budget or physical-window restart cancels its stream at the same tool boundary as endTurn, after rounds that were billed. It is recorded where the restart is decided: counted once and logged as a success with the restart's reason (stopReason 'context_budget' or 'physical_window'), with no inference:completed (the turn goes on; a completed would end typing, TTS activations and piped waits mid-turn). - Every other end (a stream error, an abort, a throw, a provider cooldown, a superseded generation) counts the rounds it finished in driveStream's finally, once (countUsage's flag; the natural completion's and the guard's authoritative usage set it first). Every inference-log entry the stream writes carries that usage unless it names its own (logStream), so log-backed health totals agree with the tracker. - main's #159 guard path counts its abandoned stream through the same countUsage, so the backstop can't count it twice. A tool-ended turn is a successful turn end, so it ends what a natural completion ends: besides inference:completed (whose emitTrace intercept already resets the hard-down streak and the poison-history rewind budget), the provider-cooldown recovery receipt and the refusal-rewind episode (a forced /unstick reports "the model responded"). endRefusalEpisode() is the natural path's else-branch, moved into a helper both call. Growth: logInference stores a compiled request over 10 KB as an uncompressed, fsync'd, content-addressed chronicle blob, one per entry, with no retention. Jasper's tool-ended entry kept the request body, adding one whole-context blob per tool-ended turn for turns that wrote nothing on main. Boundary-ended entries (tool-ended, restarts) record a note instead (BOUNDARY_END_REQUEST); natural completions and failures keep theirs. Cut as not paying rent: the copy of each usage event (membrane >=0.5.78 emits a fresh snapshot each round, and AF never mutates it); the completionAccounted flag and both WeakMap deletes (no path calls a stream's boundary end twice or after its driveStream ends: every end resets the agent or leaves its state to a successor; the map is weak). Docs: InferenceLogEntry's request/tokenUsage/stopReason say what a boundary end records; the inference:completed trace says a tool-ended turn emits it before inference:turn_ended. Tests: end-turn-usage gains restarts (both reasons: counted when the restart is decided, logged, not completed), failure ends after a finished round (error event, abort, throw), and a tool-ended turn ending the failure streak, the refusal episode with a forced /unstick, a provider cooldown (recovery receipt) and a cooldown that cuts a turn after a round; the first test checks the request note and that usage is counted by turn_ended. tool-result-guard's abandoned-round test asserts 1007 exactly. --- src/framework.ts | 212 ++++++++++++--------- src/types/framework.ts | 10 +- src/types/trace.ts | 6 + test/end-turn-usage.test.ts | 337 ++++++++++++++++++++++++++++++++- test/tool-result-guard.test.ts | 2 +- 5 files changed, 475 insertions(+), 92 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index b15d0d988..e1770e483 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -490,6 +490,16 @@ const PROCESS_LOG_ID = 'framework/process-log'; const TURN_CHECKPOINTS_ID = 'framework/turn-checkpoints'; // legacy single-map layout, read-only fallback const TURN_CHECKPOINTS_TREE_ID = 'framework/turn-checkpoints/tree'; +/** Why the framework ended a stream at a tool boundary: an endTurn tool + * result, or a context-budget restart (the `inference:stream_restarted` + * reasons). It is the stream's inference-log stop reason. */ +type StreamBoundaryEnd = 'turn_ended' | 'context_budget' | 'physical_window'; + +/** The request a boundary-ended stream's inference-log entry records in + * place of its body. A body is the whole compiled context, one blob per + * entry; these ends happen every tool-ended turn and budget restart. */ +const BOUNDARY_END_REQUEST = { note: 'request body not kept for a stream ended at a tool boundary' }; + /** Maximum number of turn checkpoints to keep per agent. */ const MAX_TURN_CHECKPOINTS = 20; const DEFAULT_DISCORD_AWARENESS_DEADLINE_MS = 10_000; @@ -1015,10 +1025,14 @@ export class AgentFramework { private processLoggingPersist: boolean; private processLoggingBroadcast: boolean; private activeStreams: Map> = new Map(); - // Tool results settle outside driveStream. Key accounting by the physical - // stream so an endTurn result can finalize it before cancellation teardown - // or ephemeral disposal, without touching a successor's usage. - private streamCompletionAccounting = new WeakMap void>(); + // The framework ends a stream at a tool boundary from handleProcessEvent: + // an endTurn result, or a context-budget restart. Each live stream + // registers here how to record that end (its usage, its log entry and, for + // endTurn, the completed turn), so the record is made where the decision + // is taken, before the cancellation's asynchronous teardown or an + // ephemeral's disposal. Keyed by the physical stream, so a successor's + // record is never touched. + private streamBoundaryEnds = new WeakMap void>(); /** Per-agent output locus FROZEN for the CURRENT logical turn. Resolved * eagerly in startAgentStream (home → addressed trigger → global default) @@ -4968,6 +4982,23 @@ export class AgentFramework { } } + /** + * A turn that ended WITHOUT a refusal (a natural completion that wasn't + * one, or a tool-ended turn) ends the rewind episode: the model responded. + * Leave the consolidated marker in place as the durable record; just + * clear the per-episode counters, and report a forced `/unstick` done. + */ + private endRefusalEpisode(agentName: string): void { + if (this.forcedRewind.has(agentName)) { + this.finishUnstick(agentName, true); + } + if (this.refusalRewinds.get(agentName)) { + this.refusalRewinds.set(agentName, 0); + } + this.rewindEpisode.delete(agentName); + this.refusalStreak.delete(agentName); + } + /** * Conclude an active `/unstick` session: post the outcome (what was shed and * whether the model stopped refusing) to the channel the command came from, @@ -6871,8 +6902,10 @@ export class AgentFramework { // driveStream's finally is the backstop.) this.eventGate?.onInferenceEnded(agent.name); this.settleAgent(agent.name, { stopReason: 'turn_ended', speech: '' }); + // A successful turn end, recorded as a natural completion's is: + // inference:completed with the rounds' usage, then turn_ended. if (currentState.stream) { - this.streamCompletionAccounting.get(currentState.stream)?.(); + this.streamBoundaryEnds.get(currentState.stream)?.('turn_ended'); } this.emitTrace({ type: 'inference:turn_ended', agentName: agent.name }); } else if (overBudget || overPhysical) { @@ -6895,6 +6928,9 @@ export class AgentFramework { // overwrote it). Harmless by accident before turn-alive // tracking; a genuine teardown leak after it. currentState.stream.cancel(); + // Its rounds were billed; the restart opens a fresh stream + // with its own usage. + this.streamBoundaryEnds.get(currentState.stream)?.(overBudget ? 'context_budget' : 'physical_window'); } agent.cancelStream(); this.emitTrace({ @@ -9488,50 +9524,89 @@ export class AgentFramework { const startTime = Date.now(); const requestId = `${agent.name}-${startTime}-${Math.random().toString(36).slice(2, 8)}`; const myStreamId = agent.streamId; + // Membrane's usage events are cumulative across the stream's tool loop, + // so the latest is everything the stream has been billed for so far: + // the measure of a stream that ends without a final response. let latestUsage: DetailedUsage | undefined; - let completionAccounted = false; - const completeAccounting = ( - usage: DetailedUsage | undefined, - response?: NormalizedResponse, - durationMs = Date.now() - startTime, - ): void => { - if (completionAccounted) return; - completionAccounted = true; - this.streamCompletionAccounting.delete(stream); - const tokenUsage = usage ? { - input: usage.inputTokens, - output: usage.outputTokens, - cacheCreation: usage.cacheCreationTokens, - cacheRead: usage.cacheReadTokens, - } : undefined; + const tokenUsageOf = (usage: DetailedUsage | undefined): InferenceLogEntry['tokenUsage'] => usage + ? { + input: usage.inputTokens, + output: usage.outputTokens, + cacheCreation: usage.cacheCreationTokens, + cacheRead: usage.cacheReadTokens, + } + : undefined; + // The stream's billed usage reaches the session totals once, however the + // stream ends: a final response's usage where there is one (completion, + // or the guard's abandoned round), else the rounds it finished (the + // boundary ends below, or `finally` for every other end). + let usageCounted = false; + const countUsage = (usage: DetailedUsage | undefined): void => { + if (usageCounted) return; + usageCounted = true; + if (!usage) return; + this.usageTracker.onInferenceCompleted(agent.name, usage, usage.estimatedCost + ? { total: usage.estimatedCost.total, currency: usage.estimatedCost.currency } + : undefined); + this.persistUsageState(); + }; + // Every inference-log entry of this stream carries the usage of the + // rounds it finished, unless the entry names its own. + const logStream = (entry: Omit): void => { + this.logInference({ + timestamp: startTime, + agentName: agent.name, + requestId, + tokenUsage: tokenUsageOf(latestUsage), + ...entry, + }); + }; + // A turn that ended successfully: with its final response (natural + // completion), or without one, when a tool result with endTurn ended it. + const completeTurn = (usage: DetailedUsage | undefined, durationMs: number, response?: NormalizedResponse): void => { + const tokenUsage = tokenUsageOf(usage); this.emitTrace({ type: 'inference:completed', agentName: agent.name, durationMs, tokenUsage, }); - if (usage) { - this.usageTracker.onInferenceCompleted(agent.name, usage, usage.estimatedCost - ? { total: usage.estimatedCost.total, currency: usage.estimatedCost.currency } - : undefined); - this.persistUsageState(); - } - // Health snapshots and log queries read this same terminal accounting. - // A tool-ended stream has no final provider response; record that fact - // rather than fabricating response content or waiting for cancellation. - this.logInference({ - timestamp: startTime, - agentName: agent.name, - requestId, + countUsage(usage); + logStream(response + ? { + success: true, + request: compiledRequest ?? { note: 'streaming request' }, + response: response.raw ?? { note: 'streaming response' }, + durationMs, + tokenUsage, + stopReason: response.stopReason, + } + : { + success: true, + request: BOUNDARY_END_REQUEST, + response: { note: 'stream ended by tool result' }, + durationMs, + tokenUsage, + stopReason: 'turn_ended', + }); + }; + this.streamBoundaryEnds.set(stream, (end) => { + const durationMs = Date.now() - startTime; + if (end === 'turn_ended') { + this.recordProviderAccelerationRecovery(agent, compiledRequest, end); + completeTurn(latestUsage, durationMs); + this.endRefusalEpisode(agent.name); + return; + } + countUsage(latestUsage); + logStream({ success: true, - request: compiledRequest ?? { note: 'streaming request' }, - response: response?.raw ?? { note: response ? 'streaming response' : 'stream ended by tool result' }, + request: BOUNDARY_END_REQUEST, + response: { note: 'stream restarted to compress its context' }, durationMs, - tokenUsage, - stopReason: response?.stopReason ?? 'turn_ended', + stopReason: end, }); - }; - this.streamCompletionAccounting.set(stream, () => completeAccounting(latestUsage)); + }); // Membrane usage events are cumulative across the native/XML tool loop. // Keep the previous cumulative sample so consumers that operate at the // physical provider-call boundary (estimator calibration and kv receipt @@ -9979,20 +10054,8 @@ export class AgentFramework { // The abandoned stream was billed (membrane usage here is // cumulative across this physical tool loop): count it before // the retry opens a fresh stream with its own usage. - const abandonedUsage = response.details?.usage; - if (abandonedUsage) { - this.usageTracker.onInferenceCompleted(agent.name, { - inputTokens: abandonedUsage.inputTokens, - outputTokens: abandonedUsage.outputTokens, - cacheCreationTokens: abandonedUsage.cacheCreationTokens, - cacheReadTokens: abandonedUsage.cacheReadTokens, - }, abandonedUsage.estimatedCost - ? { total: abandonedUsage.estimatedCost.total, currency: abandonedUsage.estimatedCost.currency } - : undefined); - this.persistUsageState(); - } - this.logInference({ - timestamp: startTime, agentName: agent.name, requestId, + countUsage(response.details?.usage); + logStream({ success: false, error: 'Tool output withheld by the guard', request: compiledRequest ?? {}, response, durationMs, tokenUsage, stopReason: 'refusal', }); @@ -10185,7 +10248,7 @@ export class AgentFramework { .join('\n'), }); - completeAccounting(du, response, durationMs); + completeTurn(du, durationMs, response); // Surface refusals instead of going silently mute: stderr line // (headless inference failures are otherwise under-logged) + an @@ -10293,17 +10356,7 @@ export class AgentFramework { void this.reactToRefusal(agent.name, category); } } else { - // A turn that completed WITHOUT a refusal ends the rewind episode: - // the model responded. Leave the consolidated marker in place as - // the durable record; just clear the per-episode counters. - if (this.forcedRewind.has(agent.name)) { - this.finishUnstick(agent.name, true); - } - if (this.refusalRewinds.get(agent.name)) { - this.refusalRewinds.set(agent.name, 0); - } - this.rewindEpisode.delete(agent.name); - this.refusalStreak.delete(agent.name); + this.endRefusalEpisode(agent.name); } // Dispatch speech (and thoughts if any) @@ -10517,10 +10570,7 @@ export class AgentFramework { stack: err.stack, }); - this.logInference({ - timestamp: startTime, - agentName: agent.name, - requestId, + logStream({ success: false, error: err.message, request: compiledRequest ?? { note: 'streaming request failed' }, @@ -10684,10 +10734,7 @@ export class AgentFramework { // state. Without this, abort-terminated inferences are // invisible to forensic queries (only request-side telemetry // via llm-calls.jsonl shows them, and only by absence). - this.logInference({ - timestamp: startTime, - agentName: agent.name, - requestId, + logStream({ success: false, error: `Stream aborted: ${reason}`, request: compiledRequest ?? { note: 'streaming request aborted' }, @@ -10699,12 +10746,8 @@ export class AgentFramework { } case 'usage': { - // Already cumulative across rounds: replace, rather than sum, and - // retain Membrane's cost at the rates that served those rounds. - latestUsage = { - ...event.usage, - estimatedCost: event.usage.estimatedCost ? { ...event.usage.estimatedCost } : undefined, - }; + // Cumulative across the stream's rounds: replace, never sum. + latestUsage = event.usage; agent.lastStreamInputTokens = event.usage.inputTokens; agent.lastStreamRealInputTokens = (event.usage.inputTokens ?? 0) + @@ -10807,7 +10850,7 @@ export class AgentFramework { const durationMs = Date.now() - startTime; if (ownsProviderGate && this.holdProviderAcceleration(agent, err, trigger)) { this.emitTrace({ type: 'inference:failed', agentName: agent.name, error: err.message, stack: err.stack }); - this.logInference({ timestamp: startTime, agentName: agent.name, requestId, success: false, + logStream({ success: false, error: `Provider acceleration cooldown: ${err.message}`, request: compiledRequest ?? { note: 'streaming request rate-limited' }, durationMs }); this.abortAgentScript(agent.name, 'provider acceleration cooldown'); @@ -10834,10 +10877,7 @@ export class AgentFramework { // Postmortem 2026-05-28 P2 #7: catch-path failures (stream itself // threw) were previously only visible via in-memory trace listeners. // Persist to inference log for forensic attribution. - this.logInference({ - timestamp: startTime, - agentName: agent.name, - requestId, + logStream({ success: false, error: `Stream threw: ${err.message}`, request: compiledRequest ?? { note: 'streaming request threw' }, @@ -10848,6 +10888,9 @@ export class AgentFramework { agent.reset(); if (this.agents.get(agent.name) === agent && agent.streamId === myStreamId) this.eventGate?.onInferenceEnded(agent.name); } finally { + // Every end the cases above didn't count (a stream error or abort, a + // throw, a superseded generation) still billed the rounds it finished. + countUsage(latestUsage); const unsettledKvSubmissions = drainKvSubmissionIds?.() ?? []; if (unsettledKvSubmissions.length > 0) { try { @@ -10938,7 +10981,6 @@ export class AgentFramework { } } this.frameworkCancelledStreams.delete(`${agent.name}:${myStreamId}`); - this.streamCompletionAccounting.delete(stream); if (ownsPhysicalStream) { this.activeStreams.delete(agent.name); this.pendingAssistantBlocks.delete(agent.name); diff --git a/src/types/framework.ts b/src/types/framework.ts index a127866c2..87fb67491 100644 --- a/src/types/framework.ts +++ b/src/types/framework.ts @@ -403,18 +403,24 @@ export interface InferenceLogEntry { success: boolean; /** Error message if failed */ error?: string; - /** Request data or blob ID if large */ + /** Request data or blob ID if large. A stream the framework ended at a + * tool boundary (stopReason 'turn_ended', 'context_budget' or + * 'physical_window') records a note in place of the body. */ request: unknown | { blobId: string }; /** Response data or blob ID if large (only if successful) */ response?: unknown | { blobId: string }; durationMs: number; + /** The stream's usage: its final response's, or, for a stream that ended + * without one, the provider rounds it finished before it ended. */ tokenUsage?: { input: number; output: number; cacheCreation?: number; cacheRead?: number; }; - /** Stop reason from the model, or 'turn_ended' for a tool-ended stream. */ + /** Stop reason from the model, or why the framework ended the stream at a + * tool boundary: 'turn_ended' (a tool result with endTurn), or + * 'context_budget' / 'physical_window' (a context-budget restart). */ stopReason?: string; } diff --git a/src/types/trace.ts b/src/types/trace.ts index de8f41ea8..0adc00d68 100644 --- a/src/types/trace.ts +++ b/src/types/trace.ts @@ -52,6 +52,12 @@ export type TraceEvent = silent?: boolean; }) | (TraceEventBase & { + /** + * The model answered and the turn ended: with the stream's final + * response, or with a tool result carrying `endTurn`, which + * `inference:turn_ended` then follows. `tokenUsage` is the stream's + * usage, cumulative across its rounds. + */ type: 'inference:completed'; agentName: string; durationMs: number; diff --git a/test/end-turn-usage.test.ts b/test/end-turn-usage.test.ts index 276c6dd63..84946b564 100644 --- a/test/end-turn-usage.test.ts +++ b/test/end-turn-usage.test.ts @@ -3,11 +3,12 @@ import assert from 'node:assert/strict'; import { mkdtempSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import type { DetailedUsage, NormalizedResponse } from '@animalabs/membrane'; +import { MembraneError } from '@animalabs/membrane'; +import type { DetailedUsage, NormalizedRequest, NormalizedResponse, StreamEvent, YieldingStream } from '@animalabs/membrane'; import { AgentFramework } from '../src/index.js'; import { HealthModule } from '../src/modules/health/index.js'; -import type { Module, ToolCall, TraceEvent } from '../src/index.js'; -import { createMockResponse, MockMembrane } from './helpers/mock-membrane.js'; +import type { EventResponse, Module, ProcessEvent, ToolCall, TraceEvent } from '../src/index.js'; +import { createMockResponse, MockMembrane, MockYieldingStream } from './helpers/mock-membrane.js'; class TurnTools implements Module { readonly name = 'test'; @@ -70,14 +71,19 @@ async function fixture() { }); const traces: TraceEvent[] = []; const completionStates: Array = []; + // What the session totals already hold when the turn's terminal traces fire. + const countedAt: Array<{ type: string; inferenceCount: number }> = []; framework.onTrace((event) => { traces.push(event); if (event.type === 'inference:completed') { completionStates.push(framework.getAgent(event.agentName)?.state.status); } + if (event.type === 'inference:turn_ended' || event.type === 'inference:stream_restarted') { + countedAt.push({ type: event.type, inferenceCount: framework.getSessionUsage().inferenceCount }); + } }); framework.start(); - return { dir, storePath, membrane, framework, traces, completionStates }; + return { dir, storePath, membrane, framework, traces, completionStates, countedAt }; } async function run(framework: AgentFramework, name: string) { @@ -115,6 +121,10 @@ describe('tool-ended turn usage accounting', () => { assert.equal(logs.entries[0].entry.stopReason, 'turn_ended'); assert.deepEqual(logs.entries[0].entry.tokenUsage, expectedTokens(totalUsage)); assert.deepEqual(logs.entries[0].entry.response, { note: 'stream ended by tool result' }); + assert.deepEqual(logs.entries[0].entry.request, { note: 'request body not kept for a stream ended at a tool boundary' }, + 'no compiled-context blob per tool-ended turn'); + assert.deepEqual(f.countedAt, [{ type: 'inference:turn_ended', inferenceCount: 1 }], + 'counted before turn_ended fires, as a completion is before its caller resumes'); assert.equal(logs.entries[0].entry.durationMs, completed[0].durationMs); assert.equal(f.traces.filter((e) => e.type === 'inference:turn_ended').length, 1); assert.equal(f.traces.filter((e) => e.type === 'usage:updated').length, 1); @@ -243,3 +253,322 @@ describe('tool-ended turn usage accounting', () => { } }); }); + +/** One queued response per stream: a restart opens a fresh stream. */ +class SequentialStreamMembrane extends MockMembrane { + private next = 0; + override streamYielding(request: NormalizedRequest): YieldingStream { + this.calls.push(request); + const stream = new MockYieldingStream(this.responses.slice(this.next, ++this.next)); + this.lastStream = stream; + return stream; + } +} + +/** A stream that finishes one tool round with `usage`, then ends as `ending` says. */ +class RoundThenEnd implements YieldingStream { + isWaitingForTools = false; + pendingToolCallIds: string[] = []; + toolDepth = 0; + isCancelled = false; + private resumed = false; + private resume: (() => void) | null = null; + constructor(private readonly usage: DetailedUsage, private readonly ending: StreamEvent | Error) {} + provideToolResults(): void { + this.isWaitingForTools = false; + this.resumed = true; + this.resume?.(); + } + cancel(): void { + this.isCancelled = true; + this.resumed = true; + this.resume?.(); + } + async *[Symbol.asyncIterator](): AsyncIterator { + yield { type: 'usage', usage: this.usage } as StreamEvent; + const call = { type: 'tool_use' as const, id: 'echo-1', name: 'test--echo', input: {} }; + this.isWaitingForTools = true; + this.pendingToolCallIds = [call.id]; + yield { + type: 'tool-calls', + calls: [{ id: call.id, name: call.name, input: {} }], + context: { rawText: '', preamble: '', depth: 0, previousResults: [], accumulated: '', roundContent: [call] }, + } as StreamEvent; + if (!this.resumed) await new Promise((r) => { this.resume = r; }); + if (this.isCancelled) { + yield { type: 'aborted', reason: 'user' } as StreamEvent; + return; + } + if (this.ending instanceof Error) throw this.ending; + yield this.ending; + } +} + +class StreamsMembrane extends MockMembrane { + constructor(private readonly streams: Array<() => YieldingStream>) { super(); } + override streamYielding(request: NormalizedRequest): YieldingStream { + this.calls.push(request); + const make = this.streams.shift(); + return make ? make() : new MockYieldingStream([createMockResponse([{ type: 'text', text: 'done' }])]); + } +} + +async function pollUntil(cond: () => boolean, timeoutMs = 5_000): Promise { + const deadline = Date.now() + timeoutMs; + while (!cond() && Date.now() < deadline) await new Promise((r) => setTimeout(r, 10)); + return cond(); +} + +function finalResponse(text: string, usage: DetailedUsage): NormalizedResponse { + const response = createMockResponse([{ type: 'text', text }]); + response.usage = usage; + response.details!.usage = usage; + return response; +} + +const secondUsage: DetailedUsage = { + inputTokens: 300, outputTokens: 7, cacheCreationTokens: 0, cacheReadTokens: 200, + estimatedCost: { input: 0.02, output: 0.01, total: 0.03, currency: 'USD' }, +}; + +function summed(a: DetailedUsage, b: DetailedUsage) { + return { + inputTokens: a.inputTokens + b.inputTokens, outputTokens: a.outputTokens + b.outputTokens, + cacheCreationTokens: (a.cacheCreationTokens ?? 0) + (b.cacheCreationTokens ?? 0), + cacheReadTokens: (a.cacheReadTokens ?? 0) + (b.cacheReadTokens ?? 0), + estimatedCost: { total: a.estimatedCost!.total + b.estimatedCost!.total, currency: 'USD' }, + }; +} + +describe('a stream counts its usage once, however it ends', () => { + const nearCap: DetailedUsage = { + inputTokens: 5_000, outputTokens: 5, cacheCreationTokens: 0, cacheReadTokens: 195_000, + estimatedCost: { input: 0.05, output: 0.01, total: 0.06, currency: 'USD' }, + }; + for (const [reason, limits, usage] of [ + ['context_budget', { maxStreamTokens: 50 }, firstUsage], + ['physical_window', { physicalWindowTokens: 200_000, maxTokens: 4_000 }, nearCap], + ] as const) { + it(`a ${reason} restart counts and logs the stream it cancels, without completing the turn`, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-end-turn-usage-')); + const membrane = new SequentialStreamMembrane(); + membrane.pushResponse(toolResponse('echo', 'echo-1', usage)); + membrane.pushResponse(finalResponse('Recovered.', secondUsage)); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store.chronicle'), membrane: membrane.asMembrane(), agents: [], + modules: [new TurnTools()], syncIntervalMs: 0, + }); + const traces: TraceEvent[] = []; + const countedAtRestart: number[] = []; + framework.onTrace((event) => { + traces.push(event); + if (event.type === 'inference:stream_restarted') countedAtRestart.push(framework.getSessionUsage().inferenceCount); + }); + framework.start(); + try { + const { agent, contextManager } = await framework.createEphemeralAgent({ + name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', + proseRouting: 'disabled', ...limits, + }); + contextManager.addMessage('user', [{ type: 'text', text: 'Run once.' }]); + await framework.runEphemeralToCompletion(agent, contextManager); + await new Promise((resolve) => setImmediate(resolve)); + + assert.equal(membrane.calls.length, 2, 'the restart opened a second stream'); + assert.equal((traces.find((e) => e.type === 'inference:stream_restarted') as { reason?: string })?.reason, reason); + assert.deepEqual(countedAtRestart, [1], 'the cancelled stream is counted where the restart is decided'); + assert.equal(traces.filter((e) => e.type === 'inference:completed').length, 1, 'only the turn\'s end completes'); + assert.equal(traces.filter((e) => e.type === 'usage:updated').length, 2); + const snapshot = framework.getSessionUsage(); + assert.equal(snapshot.inferenceCount, 2); + assert.deepEqual(snapshot.totals, summed(usage, secondUsage)); + const logs = framework.queryInferenceLogs({ agentName: 'worker' }).entries.map((e) => e.entry).reverse(); + assert.equal(logs.length, 2); + assert.deepEqual( + { success: logs[0].success, stopReason: logs[0].stopReason, tokenUsage: logs[0].tokenUsage, request: logs[0].request, response: logs[0].response }, + { + success: true, stopReason: reason, tokenUsage: expectedTokens(usage), + request: { note: 'request body not kept for a stream ended at a tool boundary' }, + response: { note: 'stream restarted to compress its context' }, + }, + ); + assert.equal(logs[1].stopReason, 'end_turn'); + assert.deepEqual(logs[1].tokenUsage, expectedTokens(secondUsage)); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + } + + for (const [name, ending] of [ + ['a stream error', { type: 'error', error: new MembraneError({ type: 'auth', retryable: false, message: 'key revoked', rawError: { status: 401 } }) } as StreamEvent], + ['an abort', { type: 'aborted', reason: 'timeout' } as StreamEvent], + ['a thrown stream', new Error('socket hang up')], + ] as const) { + it(`${name} after a finished round counts that round once and logs it with the failure`, async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-end-turn-usage-')); + const membrane = new StreamsMembrane([() => new RoundThenEnd(firstUsage, ending)]); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store.chronicle'), membrane: membrane.asMembrane(), agents: [], + modules: [new TurnTools()], syncIntervalMs: 0, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + framework.start(); + try { + const { agent, contextManager } = await framework.createEphemeralAgent({ + name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', proseRouting: 'disabled', + }); + contextManager.addMessage('user', [{ type: 'text', text: 'Run once.' }]); + await framework.runEphemeralToCompletion(agent, contextManager).catch(() => undefined); + assert.ok(await pollUntil(() => traces.some((e) => e.type === 'usage:updated')), 'the finished round was counted'); + await new Promise((r) => setTimeout(r, 50)); + + assert.equal(membrane.calls.length, 1, 'no retry'); + assert.equal(traces.filter((e) => e.type === 'inference:completed').length, 0); + assert.equal(traces.filter((e) => e.type === 'usage:updated').length, 1); + assert.deepEqual(framework.getSessionUsage().totals, expectedTotals(firstUsage)); + const logs = framework.queryInferenceLogs({ agentName: 'worker' }).entries.map((e) => e.entry); + assert.equal(logs.length, 1); + assert.equal(logs[0].success, false); + assert.deepEqual(logs[0].tokenUsage, expectedTokens(firstUsage), 'the failure entry carries the finished round'); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + } +}); + +class InputModule implements Module { + readonly name = 'input'; + async start() {} + async stop() {} + getTools() { return []; } + async handleToolCall() { return { success: false, isError: true, error: 'none' }; } + async onProcess(event: ProcessEvent): Promise { + if (event.type !== 'external-message') return {}; + return { addMessages: [{ participant: 'User', content: [{ type: 'text', text: String(event.content) }] }], requestInference: true }; + } +} + +function accelerationLimit(request: unknown): MembraneError { + return new MembraneError({ + type: 'rate_limit', retryable: true, httpStatus: 429, + message: "This request would exceed your organization's maximum usage increase rate for input tokens per minute", + rawError: { status: 429 }, rawRequest: request, + }); +} + +class ThrowingStream implements YieldingStream { + isWaitingForTools = false; + pendingToolCallIds: string[] = []; + toolDepth = 0; + isCancelled = false; + constructor(private readonly error: Error) {} + provideToolResults(): void { throw new Error('not waiting'); } + cancel(): void { this.isCancelled = true; } + async *[Symbol.asyncIterator](): AsyncIterator { + throw this.error; + } +} + +/** A resident whose streams come from `streams`, one per inference, with a 1 s provider cooldown. */ +async function resident(streams: Array<(request: NormalizedRequest) => YieldingStream>) { + const dir = mkdtempSync(join(tmpdir(), 'af-end-turn-usage-')); + const membrane = new MockMembrane(); + membrane.streamYielding = (request: NormalizedRequest): YieldingStream => { + membrane.calls.push(request); + const make = streams.shift(); + return make ? make(request) : new MockYieldingStream([createMockResponse([{ type: 'text', text: 'done' }])]); + }; + const framework = await AgentFramework.create({ + storePath: join(dir, 'store.chronicle'), membrane: membrane.asMembrane(), + agents: [{ name: 'resident', model: 'test-model', systemPrompt: 'system', allowedTools: 'all' }], + modules: [new InputModule(), new TurnTools()], syncIntervalMs: 0, maintenanceIntervalMs: 0, + }); + const internal = framework as unknown as { + providerAccelerationDefaultCooldownMs: number; + providerAccelerationJitterMs: number; + providerAccelerationLastRecovery: Map; + }; + internal.providerAccelerationDefaultCooldownMs = 1_000; + internal.providerAccelerationJitterMs = 0; + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + return { dir, membrane, framework, internal, traces }; +} + +describe('a tool-ended turn ends what a successful turn ends', () => { + for (const [ending, response] of [ + ['a tool-ended turn', () => toolResponse('finish', 'finish-1', totalUsage)], + ['a natural completion', () => createMockResponse([{ type: 'text', text: 'Done.' }])], + ] as const) { + it(`the failure streak and the refusal-rewind episode, a forced /unstick included: ${ending}`, async () => { + const f = await fixture(); + try { + const internal = f.framework as unknown as Record< + 'consecutiveInferenceFailures' | 'exhaustionRewinds' | 'refusalRewinds' | 'refusalStreak' | 'rewindEpisode' | 'forcedRewind', + Map + >; + for (const key of ['consecutiveInferenceFailures', 'exhaustionRewinds', 'refusalRewinds', 'refusalStreak'] as const) { + internal[key].set('worker', 2); + } + internal.rewindEpisode.set('worker', { markerId: 'marker', count: 1, category: 'cyber' }); + internal.forcedRewind.set('worker', { remaining: 1, removed: [], serverId: 'none', channelId: '' }); + f.membrane.pushResponse(response()); + await run(f.framework, 'worker'); + assert.equal(internal.consecutiveInferenceFailures.get('worker'), 0, 'the hard-down streak'); + assert.equal(internal.exhaustionRewinds.get('worker'), 0, 'the poison-history rewind budget'); + assert.equal(internal.refusalRewinds.get('worker'), 0); + assert.equal(internal.refusalStreak.has('worker'), false); + assert.equal(internal.rewindEpisode.has('worker'), false); + assert.equal(internal.forcedRewind.has('worker'), false, 'the /unstick session reports the model responded'); + } finally { + await f.framework.stop(); + rmSync(f.dir, { recursive: true, force: true }); + } + }); + } + + it('a provider cooldown, which records its recovery', async () => { + const r = await resident([ + (request) => new ThrowingStream(accelerationLimit(request)), + () => new MockYieldingStream([toolResponse('finish', 'finish-1', totalUsage)]), + ]); + r.framework.start(); + try { + r.framework.pushEvent({ type: 'external-message', source: 'test', content: 'first', metadata: {} } as unknown as ProcessEvent); + await r.framework.runUntilIdle(); + assert.equal(r.membrane.calls.length, 1, 'held for the cooldown'); + await new Promise((resolve) => setTimeout(resolve, 1_100)); + await r.framework.runUntilIdle(); + assert.ok(await pollUntil(() => r.traces.some((e) => e.type === 'inference:turn_ended')), 'the retried turn ended by tool'); + assert.equal(r.membrane.calls.length, 2); + assert.equal(r.internal.providerAccelerationLastRecovery.get('resident')?.stopReason, 'turn_ended'); + } finally { + await r.framework.stop(); + rmSync(r.dir, { recursive: true, force: true }); + } + }); + + it('a cooldown that cuts a turn after a finished round logs and counts that round', async () => { + const r = await resident([(request) => new RoundThenEnd(firstUsage, accelerationLimit(request))]); + r.framework.start(); + try { + r.framework.pushEvent({ type: 'external-message', source: 'test', content: 'first', metadata: {} } as unknown as ProcessEvent); + await r.framework.runUntilIdle(); + assert.ok(await pollUntil(() => r.traces.some((e) => e.type === 'usage:updated')), 'the finished round was counted'); + const logs = r.framework.queryInferenceLogs({ agentName: 'resident' }).entries.map((e) => e.entry); + assert.equal(logs.length, 1); + assert.match(String(logs[0].error), /^Provider acceleration cooldown: /); + assert.deepEqual(logs[0].tokenUsage, expectedTokens(firstUsage)); + assert.deepEqual(r.framework.getSessionUsage().totals, expectedTotals(firstUsage)); + assert.equal(r.traces.filter((e) => e.type === 'usage:updated').length, 1); + } finally { + await r.framework.stop(); + rmSync(r.dir, { recursive: true, force: true }); + } + }); +}); diff --git a/test/tool-result-guard.test.ts b/test/tool-result-guard.test.ts index 71ad0787a..dd3474649 100644 --- a/test/tool-result-guard.test.ts +++ b/test/tool-result-guard.test.ts @@ -523,7 +523,7 @@ test('usage of an abandoned guarded round is counted in session totals', async ( try { await h.run(); const totals = h.framework.getSessionUsage().totals as unknown as Record; - assert.ok(totals.inputTokens >= 1_007, `refused round usage missing: ${JSON.stringify(totals)}`); + assert.equal(totals.inputTokens, 1_007, `the refused round is counted, once: ${JSON.stringify(totals)}`); } finally { await h.framework.stop(); } }); From dc2158931f8af80ef3c8c96a4411ebffb75ca2e0 Mon Sep 17 00:00:00 2001 From: Ada-1017 Date: Sat, 10 Oct 2026 00:13:48 -0400 Subject: [PATCH 05/13] docs(changelog): #67's fragment says what callers will notice The fragment named only the tool-ended turn's new trace, log entry and usage. It now says, in callers' terms: a tool-ended turn emits inference:completed then inference:turn_ended and ends what a completion ends (failure streak, poison-history rewind budget, refusal episode and a forced /unstick, provider-cooldown recovery), so consumers that end a turn's display on inference:completed end it on these turns too; every stream's billed rounds reach the session totals once however it ends (restarts counted and logged without inference:completed; failures carry their finished rounds on their log entries; inferenceCount and health totals include them); and boundary-ended entries record a note instead of the compiled request. --- .../fix-67-end-turn-accounting.fixed.md | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/changelog.d/fix-67-end-turn-accounting.fixed.md b/changelog.d/fix-67-end-turn-accounting.fixed.md index 58a9b2d58..41c6895fd 100644 --- a/changelog.d/fix-67-end-turn-accounting.fixed.md +++ b/changelog.d/fix-67-end-turn-accounting.fixed.md @@ -1 +1,17 @@ -- Tool-ended turns now emit `inference:completed`, record a successful inference-log entry, and record cumulative token, cache, and estimated-cost usage exactly once, including turns ended by `skip_reply` or other `endTurn` tools. Log-backed health totals now include these turns. +- **A turn ended by a tool result (`endTurn`, such as `skip_reply`) is recorded as a completed + turn** (#67). It emits `inference:completed` with the stream's cumulative usage, then + `inference:turn_ended`, writes a successful inference-log entry (`stopReason: 'turn_ended'`), + and adds its usage once to the session totals (`usage:updated`, `getSessionUsage()`). Like a + completion, it resets the consecutive-failure streak and the poison-history rewind budget, ends + a refusal-rewind episode (a forced `/unstick` reports that the model responded), and records a + provider-cooldown recovery. Consumers that end a turn's display on `inference:completed`, such as + typing indicators, voice relays and alerts that close on success, now end it on these turns too. +- **Every stream's billed rounds reach the session totals once, however the stream ends.** A + context-budget or physical-window restart is counted and logged where it is decided + (`stopReason: 'context_budget'` or `'physical_window'`), without `inference:completed`. A stream + that errors, aborts or throws after finishing rounds counts them, and its failure entry in the + inference log carries them. `inferenceCount` counts these streams too, and log-backed health + totals include them. +- An inference-log entry for a stream the framework ended at a tool boundary (a tool-ended turn or + a restart) records a note in place of the compiled request, so these frequent entries don't each + add a whole-context blob to the store. From d7ea2944870740e26ddefb50badf907359412a8f Mon Sep 17 00:00:00 2001 From: Ada-1017 Date: Sat, 10 Oct 2026 00:19:19 -0400 Subject: [PATCH 06/13] fix(usage): a final response without usage counts the rounds' samples Basil-1007's review of 3447706: countUsage(undefined) took the once-flag without counting, so a natural completion whose response carries no details.usage, or #159's guard abandoning such a response, lost the rounds already finished, and the finally backstop couldn't recover them. countUsage now falls back to the latest cumulative sample, and completeTurn uses the same value for inference:completed's tokenUsage and the log entry, so the trace, the totals and the log agree. Identical to main whenever the final response carries usage. Tests: a natural completion whose response has no details.usage counts the last sample in all three places; an abandoned guarded round whose response has none counts each stream once from its samples. --- src/framework.ts | 20 ++++++++++++-------- test/end-turn-usage.test.ts | 21 +++++++++++++++++++++ test/tool-result-guard.test.ts | 11 +++++++++++ 3 files changed, 44 insertions(+), 8 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index e1770e483..7bb11917b 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -9537,16 +9537,18 @@ export class AgentFramework { } : undefined; // The stream's billed usage reaches the session totals once, however the - // stream ends: a final response's usage where there is one (completion, - // or the guard's abandoned round), else the rounds it finished (the - // boundary ends below, or `finally` for every other end). + // stream ends: a final response's usage where it carries one (completion, + // or the guard's abandoned round), else the rounds it finished (a final + // response without usage, the boundary ends below, or `finally` for + // every other end). let usageCounted = false; const countUsage = (usage: DetailedUsage | undefined): void => { if (usageCounted) return; usageCounted = true; - if (!usage) return; - this.usageTracker.onInferenceCompleted(agent.name, usage, usage.estimatedCost - ? { total: usage.estimatedCost.total, currency: usage.estimatedCost.currency } + const counted = usage ?? latestUsage; + if (!counted) return; + this.usageTracker.onInferenceCompleted(agent.name, counted, counted.estimatedCost + ? { total: counted.estimatedCost.total, currency: counted.estimatedCost.currency } : undefined); this.persistUsageState(); }; @@ -9564,14 +9566,16 @@ export class AgentFramework { // A turn that ended successfully: with its final response (natural // completion), or without one, when a tool result with endTurn ended it. const completeTurn = (usage: DetailedUsage | undefined, durationMs: number, response?: NormalizedResponse): void => { - const tokenUsage = tokenUsageOf(usage); + // The trace, the session totals and the log entry all say the same. + const used = usage ?? latestUsage; + const tokenUsage = tokenUsageOf(used); this.emitTrace({ type: 'inference:completed', agentName: agent.name, durationMs, tokenUsage, }); - countUsage(usage); + countUsage(used); logStream(response ? { success: true, diff --git a/test/end-turn-usage.test.ts b/test/end-turn-usage.test.ts index 84946b564..4636625e7 100644 --- a/test/end-turn-usage.test.ts +++ b/test/end-turn-usage.test.ts @@ -169,6 +169,27 @@ describe('tool-ended turn usage accounting', () => { } }); + it('counts the rounds\' samples when a final response carries no usage', async () => { + const f = await fixture(); + try { + f.membrane.pushResponse(toolResponse('echo', 'echo-1', firstUsage)); + const response = createMockResponse([{ type: 'text', text: 'Done.' }]); + response.usage = totalUsage; // the last round's cumulative sample; no details.usage + f.membrane.pushResponse(response); + await run(f.framework, 'ordinary'); + const completed = f.traces.filter((e) => e.type === 'inference:completed'); + assert.equal(completed.length, 1); + assert.deepEqual(completed[0].tokenUsage, expectedTokens(totalUsage)); + assert.equal(f.traces.filter((e) => e.type === 'usage:updated').length, 1); + assert.deepEqual(f.framework.getSessionUsage().totals, expectedTotals(totalUsage)); + const logs = f.framework.queryInferenceLogs({ agentName: 'ordinary' }); + assert.deepEqual(logs.entries[0].entry.tokenUsage, expectedTokens(totalUsage), 'the trace, the totals and the log agree'); + } finally { + await f.framework.stop(); + rmSync(f.dir, { recursive: true, force: true }); + } + }); + it('attributes successive streams separately and persists their totals through store reopen', async () => { const f = await fixture(); let stopped = false; diff --git a/test/tool-result-guard.test.ts b/test/tool-result-guard.test.ts index dd3474649..2dd9e2768 100644 --- a/test/tool-result-guard.test.ts +++ b/test/tool-result-guard.test.ts @@ -527,6 +527,17 @@ test('usage of an abandoned guarded round is counted in session totals', async ( } finally { await h.framework.stop(); } }); +test('an abandoned guarded round whose response carries no usage still counts its rounds', async () => { + // Each scripted response emits a usage sample of 10 input tokens; neither + // final response carries details.usage, so both streams count their samples. + const h = await harness([[calls('one'), refused()], [answer()]], { toolResultGuard: true }); + try { + await h.run(); + const totals = h.framework.getSessionUsage().totals as unknown as Record; + assert.equal(totals.inputTokens, 20, `each stream is counted once, from its samples: ${JSON.stringify(totals)}`); + } finally { await h.framework.stop(); } +}); + // --------------------------------------------------------------------------- // Greptile review regressions (PR #159, head 647f081). // --------------------------------------------------------------------------- From f7d45e5616c48bf27e8bcffb9c57746c1e737ce2 Mon Sep 17 00:00:00 2001 From: Ada-1017 Date: Sat, 10 Oct 2026 00:24:36 -0400 Subject: [PATCH 07/13] fix(usage): the guard's abandoned stream logs and notes what it counts Basil-1007's review of d7ea294: the guard path logged `response.details?.usage ?? response.usage` while countUsage counted `details.usage ?? latestUsage`, so with details.usage missing the log entry said response.usage while the totals had the samples, and with both missing logStream's default was overwritten by an explicit undefined. - The guard path computes one value, `details.usage ?? latestUsage`, for countUsage, its log entry and noteRefusal. - logStream defaults tokenUsage when an entry's is undefined, not only when the key is absent, so no caller can erase it. - countUsage no longer falls back itself (d7ea294 added that): every caller now passes the one value it also reports (completeTurn's `used`, the guard's `abandoned`, latestUsage at a restart and in finally), so the fallback there was redundant. Test: the guard test without final usage also checks that the withheld stream's and the retry's log entries carry what the totals counted. --- src/framework.ts | 25 +++++++++++-------------- test/tool-result-guard.test.ts | 4 ++++ 2 files changed, 15 insertions(+), 14 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index 7bb11917b..50e26ca6d 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -9540,15 +9540,14 @@ export class AgentFramework { // stream ends: a final response's usage where it carries one (completion, // or the guard's abandoned round), else the rounds it finished (a final // response without usage, the boundary ends below, or `finally` for - // every other end). + // every other end). Each caller passes the one value it also reports. let usageCounted = false; const countUsage = (usage: DetailedUsage | undefined): void => { if (usageCounted) return; usageCounted = true; - const counted = usage ?? latestUsage; - if (!counted) return; - this.usageTracker.onInferenceCompleted(agent.name, counted, counted.estimatedCost - ? { total: counted.estimatedCost.total, currency: counted.estimatedCost.currency } + if (!usage) return; + this.usageTracker.onInferenceCompleted(agent.name, usage, usage.estimatedCost + ? { total: usage.estimatedCost.total, currency: usage.estimatedCost.currency } : undefined); this.persistUsageState(); }; @@ -9559,8 +9558,8 @@ export class AgentFramework { timestamp: startTime, agentName: agent.name, requestId, - tokenUsage: tokenUsageOf(latestUsage), ...entry, + tokenUsage: entry.tokenUsage ?? tokenUsageOf(latestUsage), }); }; // A turn that ended successfully: with its final response (natural @@ -10049,16 +10048,14 @@ export class AgentFramework { // after the single recovery retry. proseStream?.reset(); if (withheld) { - const usage = response.details?.usage ?? response.usage; - const tokenUsage = usage ? { - input: usage.inputTokens, output: usage.outputTokens, - cacheCreation: usage.cacheCreationTokens, cacheRead: usage.cacheReadTokens, - } : undefined; - this.noteRefusal(agent.name, category, tokenUsage); // The abandoned stream was billed (membrane usage here is // cumulative across this physical tool loop): count it before - // the retry opens a fresh stream with its own usage. - countUsage(response.details?.usage); + // the retry opens a fresh stream with its own usage. One value + // for the count, the log entry and the refusal note. + const abandoned = response.details?.usage ?? latestUsage; + const tokenUsage = tokenUsageOf(abandoned); + this.noteRefusal(agent.name, category, tokenUsage); + countUsage(abandoned); logStream({ success: false, error: 'Tool output withheld by the guard', request: compiledRequest ?? {}, response, durationMs, tokenUsage, stopReason: 'refusal', diff --git a/test/tool-result-guard.test.ts b/test/tool-result-guard.test.ts index 2dd9e2768..510ca08b9 100644 --- a/test/tool-result-guard.test.ts +++ b/test/tool-result-guard.test.ts @@ -535,6 +535,10 @@ test('an abandoned guarded round whose response carries no usage still counts it await h.run(); const totals = h.framework.getSessionUsage().totals as unknown as Record; assert.equal(totals.inputTokens, 20, `each stream is counted once, from its samples: ${JSON.stringify(totals)}`); + // The log says what the totals count: the withheld stream, then the retry. + const logs = h.framework.queryInferenceLogs({ agentName: 'assistant' }).entries.map((e) => e.entry).reverse(); + assert.deepEqual(logs.map((e) => [e.success, e.tokenUsage]), + [[false, { input: 10, output: 5 }], [true, { input: 10, output: 5 }]]); } finally { await h.framework.stop(); } }); From d096f6dbea2348ca34cb0718bddbca4c7faac6c3 Mon Sep 17 00:00:00 2001 From: Olive-1011 Date: Sun, 11 Oct 2026 16:42:13 -0400 Subject: [PATCH 08/13] fix(usage): a stream's usage is logged once, however it ends; the guard keeps its response-usage fallback A review of this pull request found that the count and the log disagreed for some ends: driveStream's finally counts every stream's usage once, but a turn abandoned by an operator quiesce returned before any log write, a superseded stream (its natural completion, its abort, the generation check at the top of the loop, a throw) returned or broke without one, and an entry written after another entry of the same stream copied the stream's latest usage sample again. So session totals and log-backed health totals disagreed, and the text contradicted itself. Counted once and logged once: - Every entry of a stream goes through logStream, which now carries the stream's usage on the first entry that has any (the value the entry names, else the latest cumulative sample) and on no later entry of that stream. Before, a failure entry written after a completion's carried the rounds again, and log-backed totals counted them twice. - countUsage records the value it counted. In finally, after counting, a stream whose counted usage no entry carried writes one entry, success false, naming what ended it: "Turn abandoned by operator quiesce" (set where the quiesce-abandoned abort returns), "Stream superseded: a newer stream of the agent took over" when a newer stream of the same agent took over, or "Stream abandoned: its agent was disposed or replaced" when the agent itself is gone (a disposed ephemeral, a replaced conversation). The entry carries exactly what was counted. A stream with no usage writes nothing there, as before. - InferenceLogEntry.tokenUsage's comment says that exactly one entry of a stream carries its usage. #159's guard counts and notes one value again with main's fallback: response.details?.usage ?? latestUsage ?? response.usage. Before this commit the fallback stopped at the running sample, so a guard refusal whose stream emitted no usage event counted and noted nothing, where main noted the response's own usage. Comments that list the ends taken at a tool boundary (StreamBoundaryEnd, streamBoundaryEnds, BOUNDARY_END_REQUEST, InferenceLogEntry.stopReason) now name the physical-window restart beside the context-budget one, as the code handles both. Changelog: consumers that end a turn's display on inference:completed are a host's, such as connectome-host's; quiesce-abandoned and superseded streams, and those of a disposed agent, now write a failure entry carrying their rounds, and exactly one entry of each stream carries its usage; the new boundary entries still grow the log, which keeps no retention, by one small entry per tool-ended turn and restart. Tests: - end-turn-usage: a turn abandoned by an operator quiesce after a finished round, and a stream superseded by a newer one after a finished round, each count that round once and write one failure entry carrying it, naming what ended it; a failure after a completion (on a resident, whose stream's catch logs it) is logged without carrying the usage again. - end-turn-usage: the stream error, abort and thrown-stream tests wait for the run to settle, bounded, and check it failed with the stream and that no retry had been made by then, instead of swallowing the rejection and sleeping 50 ms; the provider-cooldown test waits until the retried call happens, with time only as the outer bound, instead of sleeping 1.1 s; the forced /unstick tests read the posted outcome ("the model responded"), not only that the session was cleared. - tool-result-guard: the abandoned-round test scripts membrane's cumulative samples (10, then 20 input tokens) in the test, so a count taken from the first sample fails it; a new test strips the usage events, so the guard counts and logs the refused response's own usage. Each of these fixes, undone alone, fails end-turn-usage or tool-result-guard: the finally entry; the quiesce-abandoned name; the once-only entry; the guard's response-usage fallback; keeping the latest sample; the forced /unstick's success report. --- .../fix-67-end-turn-accounting.fixed.md | 14 +- src/framework.ts | 59 +++++-- src/types/framework.ts | 8 +- test/end-turn-usage.test.ts | 148 +++++++++++++++++- test/tool-result-guard.test.ts | 38 ++++- 5 files changed, 236 insertions(+), 31 deletions(-) diff --git a/changelog.d/fix-67-end-turn-accounting.fixed.md b/changelog.d/fix-67-end-turn-accounting.fixed.md index 41c6895fd..8eaffbd16 100644 --- a/changelog.d/fix-67-end-turn-accounting.fixed.md +++ b/changelog.d/fix-67-end-turn-accounting.fixed.md @@ -4,14 +4,18 @@ and adds its usage once to the session totals (`usage:updated`, `getSessionUsage()`). Like a completion, it resets the consecutive-failure streak and the poison-history rewind budget, ends a refusal-rewind episode (a forced `/unstick` reports that the model responded), and records a - provider-cooldown recovery. Consumers that end a turn's display on `inference:completed`, such as - typing indicators, voice relays and alerts that close on success, now end it on these turns too. + provider-cooldown recovery. In hosts such as connectome-host, consumers that end a turn's display + on `inference:completed` (a typing indicator, notices that close on success, a voice relay) now + end it on these turns too. - **Every stream's billed rounds reach the session totals once, however the stream ends.** A context-budget or physical-window restart is counted and logged where it is decided (`stopReason: 'context_budget'` or `'physical_window'`), without `inference:completed`. A stream that errors, aborts or throws after finishing rounds counts them, and its failure entry in the - inference log carries them. `inferenceCount` counts these streams too, and log-backed health - totals include them. + inference log carries them. A turn abandoned by an operator quiesce, a stream superseded by a + newer one, or one whose agent was disposed, which wrote no entry before, now writes a failure + entry naming what ended it and carrying its rounds. `inferenceCount` counts these streams too. Exactly one entry of each stream + carries its usage, so log-backed health totals agree with the session totals. - An inference-log entry for a stream the framework ended at a tool boundary (a tool-ended turn or a restart) records a note in place of the compiled request, so these frequent entries don't each - add a whole-context blob to the store. + add a whole-context blob to the store. They are still new entries: the inference log keeps no + retention, so it grows by one small entry for every tool-ended turn and restart. diff --git a/src/framework.ts b/src/framework.ts index 50e26ca6d..2ea0f4b2f 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -491,13 +491,15 @@ const TURN_CHECKPOINTS_ID = 'framework/turn-checkpoints'; // legacy single-map l const TURN_CHECKPOINTS_TREE_ID = 'framework/turn-checkpoints/tree'; /** Why the framework ended a stream at a tool boundary: an endTurn tool - * result, or a context-budget restart (the `inference:stream_restarted` - * reasons). It is the stream's inference-log stop reason. */ + * result, or a context-budget or physical-window restart (the + * `inference:stream_restarted` reasons). It is the stream's inference-log + * stop reason. */ type StreamBoundaryEnd = 'turn_ended' | 'context_budget' | 'physical_window'; /** The request a boundary-ended stream's inference-log entry records in * place of its body. A body is the whole compiled context, one blob per - * entry; these ends happen every tool-ended turn and budget restart. */ + * entry; these ends happen every tool-ended turn and every context-budget + * or physical-window restart. */ const BOUNDARY_END_REQUEST = { note: 'request body not kept for a stream ended at a tool boundary' }; /** Maximum number of turn checkpoints to keep per agent. */ @@ -1026,12 +1028,12 @@ export class AgentFramework { private processLoggingBroadcast: boolean; private activeStreams: Map> = new Map(); // The framework ends a stream at a tool boundary from handleProcessEvent: - // an endTurn result, or a context-budget restart. Each live stream - // registers here how to record that end (its usage, its log entry and, for - // endTurn, the completed turn), so the record is made where the decision - // is taken, before the cancellation's asynchronous teardown or an - // ephemeral's disposal. Keyed by the physical stream, so a successor's - // record is never touched. + // an endTurn result, or a context-budget or physical-window restart. Each + // live stream registers here how to record that end (its usage, its log + // entry and, for endTurn, the completed turn), so the record is made where + // the decision is taken, before the cancellation's asynchronous teardown + // or an ephemeral's disposal. Keyed by the physical stream, so a + // successor's record is never touched. private streamBoundaryEnds = new WeakMap void>(); /** Per-agent output locus FROZEN for the CURRENT logical turn. Resolved @@ -9542,26 +9544,37 @@ export class AgentFramework { // response without usage, the boundary ends below, or `finally` for // every other end). Each caller passes the one value it also reports. let usageCounted = false; + let countedUsage: DetailedUsage | undefined; const countUsage = (usage: DetailedUsage | undefined): void => { if (usageCounted) return; usageCounted = true; + countedUsage = usage; if (!usage) return; this.usageTracker.onInferenceCompleted(agent.name, usage, usage.estimatedCost ? { total: usage.estimatedCost.total, currency: usage.estimatedCost.currency } : undefined); this.persistUsageState(); }; - // Every inference-log entry of this stream carries the usage of the - // rounds it finished, unless the entry names its own. + // ...and is carried by exactly one of its inference-log entries, so + // log-backed totals agree with the session's. The first entry carries + // it (the value the entry names, else the rounds finished so far); no + // later entry of this stream carries it again. An end that writes no + // entry of its own is logged in `finally`. + let usageLogged = false; const logStream = (entry: Omit): void => { + const tokenUsage = usageLogged ? undefined : (entry.tokenUsage ?? tokenUsageOf(latestUsage)); + if (tokenUsage) usageLogged = true; this.logInference({ timestamp: startTime, agentName: agent.name, requestId, ...entry, - tokenUsage: entry.tokenUsage ?? tokenUsageOf(latestUsage), + tokenUsage, }); }; + // What ended a stream whose end writes no log entry of its own, for the + // entry `finally` writes for it. + let unloggedEnd: string | undefined; // A turn that ended successfully: with its final response (natural // completion), or without one, when a tool result with endTurn ended it. const completeTurn = (usage: DetailedUsage | undefined, durationMs: number, response?: NormalizedResponse): void => { @@ -10052,7 +10065,7 @@ export class AgentFramework { // cumulative across this physical tool loop): count it before // the retry opens a fresh stream with its own usage. One value // for the count, the log entry and the refusal note. - const abandoned = response.details?.usage ?? latestUsage; + const abandoned: DetailedUsage | undefined = response.details?.usage ?? latestUsage ?? response.usage; const tokenUsage = tokenUsageOf(abandoned); this.noteRefusal(agent.name, category, tokenUsage); countUsage(abandoned); @@ -10675,6 +10688,7 @@ export class AgentFramework { this.eventGate?.onInferenceEnded(agent.name); } if (cancelKind === 'quiesce_abandoned') { + unloggedEnd = 'Turn abandoned by operator quiesce'; // Operator-initiated cancel during a quiesce drain. Settle // the turn honestly — deliberately NO inference:exhausted: // that trace centrally drives the consecutive-failure @@ -10892,6 +10906,25 @@ export class AgentFramework { // Every end the cases above didn't count (a stream error or abort, a // throw, a superseded generation) still billed the rounds it finished. countUsage(latestUsage); + // An end that wrote no entry carrying them (a quiesce-abandoned turn, + // a stream superseded by a newer one, a disposed agent's stream) writes + // one now, as a failure naming what ended it, so the log carries what + // the totals count. + if (countedUsage && !usageLogged) { + const replaced = this.agents.get(agent.name) !== agent; + const superseded = generationLost || agent.streamId !== myStreamId; + logStream({ + success: false, + error: unloggedEnd ?? (replaced + ? 'Stream abandoned: its agent was disposed or replaced' + : superseded + ? 'Stream superseded: a newer stream of the agent took over' + : 'Stream ended without a final response'), + request: compiledRequest ?? { note: 'streaming request' }, + durationMs: Date.now() - startTime, + tokenUsage: tokenUsageOf(countedUsage), + }); + } const unsettledKvSubmissions = drainKvSubmissionIds?.() ?? []; if (unsettledKvSubmissions.length > 0) { try { diff --git a/src/types/framework.ts b/src/types/framework.ts index 87fb67491..8daf6e6b2 100644 --- a/src/types/framework.ts +++ b/src/types/framework.ts @@ -411,7 +411,10 @@ export interface InferenceLogEntry { response?: unknown | { blobId: string }; durationMs: number; /** The stream's usage: its final response's, or, for a stream that ended - * without one, the provider rounds it finished before it ended. */ + * without one, the provider rounds it finished before it ended. Exactly + * one entry of a stream carries it, the one its end wrote; a stream whose + * end writes none (a quiesce-abandoned turn, a superseded stream, a + * disposed agent's stream) gets a failure entry that carries it. */ tokenUsage?: { input: number; output: number; @@ -420,7 +423,8 @@ export interface InferenceLogEntry { }; /** Stop reason from the model, or why the framework ended the stream at a * tool boundary: 'turn_ended' (a tool result with endTurn), or - * 'context_budget' / 'physical_window' (a context-budget restart). */ + * 'context_budget' / 'physical_window' (a context-budget or + * physical-window restart). */ stopReason?: string; } diff --git a/test/end-turn-usage.test.ts b/test/end-turn-usage.test.ts index 4636625e7..2e026490f 100644 --- a/test/end-turn-usage.test.ts +++ b/test/end-turn-usage.test.ts @@ -442,11 +442,17 @@ describe('a stream counts its usage once, however it ends', () => { name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', proseRouting: 'disabled', }); contextManager.addMessage('user', [{ type: 'text', text: 'Run once.' }]); - await framework.runEphemeralToCompletion(agent, contextManager).catch(() => undefined); + // The run settles once the failure path has decided: it rejects with + // the stream's failure, or, had a retry been scheduled, it would wait + // for it and resolve on the default response. + const outcome = await Promise.race([ + framework.runEphemeralToCompletion(agent, contextManager).then(() => 'resolved', (error: unknown) => error), + new Promise((resolve) => setTimeout(() => resolve('timed out'), 10_000)), + ]); + assert.ok(outcome instanceof Error, `the run failed with the stream: ${String(outcome)}`); + assert.equal(membrane.calls.length, 1, 'no retry was scheduled by the time the run failed'); assert.ok(await pollUntil(() => traces.some((e) => e.type === 'usage:updated')), 'the finished round was counted'); - await new Promise((r) => setTimeout(r, 50)); - assert.equal(membrane.calls.length, 1, 'no retry'); assert.equal(traces.filter((e) => e.type === 'inference:completed').length, 0); assert.equal(traces.filter((e) => e.type === 'usage:updated').length, 1); assert.deepEqual(framework.getSessionUsage().totals, expectedTotals(firstUsage)); @@ -537,9 +543,15 @@ describe('a tool-ended turn ends what a successful turn ends', () => { internal[key].set('worker', 2); } internal.rewindEpisode.set('worker', { markerId: 'marker', count: 1, category: 'cyber' }); - internal.forcedRewind.set('worker', { remaining: 1, removed: [], serverId: 'none', channelId: '' }); + internal.forcedRewind.set('worker', { remaining: 1, removed: [], serverId: 'ops', channelId: 'ops-channel' }); + const posted: Array<{ channelId: string; content: string }> = []; + (f.framework as unknown as { postToChannel: (s: string, c: string, content: string) => Promise }) + .postToChannel = async (_server, channelId, content) => { posted.push({ channelId, content }); }; f.membrane.pushResponse(response()); await run(f.framework, 'worker'); + assert.equal(posted.length, 1, 'the /unstick outcome was posted'); + assert.equal(posted[0].channelId, 'ops-channel'); + assert.match(posted[0].content, /the model responded/); assert.equal(internal.consecutiveInferenceFailures.get('worker'), 0, 'the hard-down streak'); assert.equal(internal.exhaustionRewinds.get('worker'), 0, 'the poison-history rewind budget'); assert.equal(internal.refusalRewinds.get('worker'), 0); @@ -563,8 +575,7 @@ describe('a tool-ended turn ends what a successful turn ends', () => { r.framework.pushEvent({ type: 'external-message', source: 'test', content: 'first', metadata: {} } as unknown as ProcessEvent); await r.framework.runUntilIdle(); assert.equal(r.membrane.calls.length, 1, 'held for the cooldown'); - await new Promise((resolve) => setTimeout(resolve, 1_100)); - await r.framework.runUntilIdle(); + assert.ok(await pollUntil(() => r.membrane.calls.length === 2, 10_000), 'the cooldown ended and the call was retried'); assert.ok(await pollUntil(() => r.traces.some((e) => e.type === 'inference:turn_ended')), 'the retried turn ended by tool'); assert.equal(r.membrane.calls.length, 2); assert.equal(r.internal.providerAccelerationLastRecovery.get('resident')?.stopReason, 'turn_ended'); @@ -593,3 +604,128 @@ describe('a tool-ended turn ends what a successful turn ends', () => { } }); }); + +/** A completion the stream would yield next; these tests end it before it's handled. */ +const unseenCompletion = (): StreamEvent => + ({ type: 'complete', response: createMockResponse([{ type: 'text', text: 'unseen' }]) }) as StreamEvent; + +/** A tool that returns only once released, so the stream waits on it. */ +class HeldTools implements Module { + readonly name = 'test'; + release: () => void = () => {}; + private readonly held = new Promise((resolve) => { this.release = resolve; }); + calledWith: string[] = []; + async start() {} + async stop() {} + async onProcess() { return {}; } + getTools() { + return [{ name: 'echo', description: 'echo', inputSchema: { type: 'object' as const, properties: {} } }]; + } + async handleToolCall(call: ToolCall) { + this.calledWith.push(call.id); + await this.held; + return { success: true, data: {} }; + } +} + +describe('a stream that ends without writing its own entry is logged once, in finally', () => { + async function residentWith(stream: YieldingStream, tools: Module) { + const dir = mkdtempSync(join(tmpdir(), 'af-end-turn-usage-')); + const membrane = new MockMembrane(); + let made = false; + membrane.streamYielding = (request: NormalizedRequest): YieldingStream => { + membrane.calls.push(request); + if (made) return new MockYieldingStream([createMockResponse([{ type: 'text', text: 'done' }])]); + made = true; + return stream; + }; + const framework = await AgentFramework.create({ + storePath: join(dir, 'store.chronicle'), membrane: membrane.asMembrane(), + agents: [{ name: 'resident', model: 'test-model', systemPrompt: 'system', allowedTools: 'all' }], + modules: [new InputModule(), tools], syncIntervalMs: 0, maintenanceIntervalMs: 0, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + framework.start(); + framework.pushEvent({ type: 'external-message', source: 'test', content: 'go', metadata: {} } as unknown as ProcessEvent); + return { dir, membrane, framework, traces }; + } + + function loggedOnce(framework: AgentFramework, traces: TraceEvent[], error: RegExp) { + assert.equal(traces.filter((e) => e.type === 'usage:updated').length, 1, 'counted once'); + assert.deepEqual(framework.getSessionUsage().totals, expectedTotals(firstUsage)); + const logs = framework.queryInferenceLogs({ agentName: 'resident' }).entries.map((e) => e.entry); + assert.equal(logs.length, 1, 'one entry'); + assert.equal(logs[0].success, false); + assert.match(String(logs[0].error), error); + assert.deepEqual(logs[0].tokenUsage, expectedTokens(firstUsage), 'the entry carries what was counted'); + } + + it('a turn abandoned by an operator quiesce after a finished round', async () => { + const tools = new HeldTools(); + const r = await residentWith(new RoundThenEnd(firstUsage, unseenCompletion()), tools); + try { + assert.ok(await pollUntil(() => tools.calledWith.length === 1), 'the round finished and its tool is running'); + await r.framework.quiesce({ timeoutMs: 50, abandon: true }); + assert.ok(await pollUntil(() => r.framework.queryInferenceLogs({ agentName: 'resident' }).total > 0), 'logged'); + loggedOnce(r.framework, r.traces, /^Turn abandoned by operator quiesce$/); + } finally { + // Let the held tool finish while the framework's queue is still open. + tools.release(); + await pollUntil(() => r.traces.some((e) => + (e.type === 'tool:completed' || e.type === 'tool:failed') && (e as { callId?: string }).callId === 'echo-1')); + await r.framework.stop(); + rmSync(r.dir, { recursive: true, force: true }); + } + }); + + it('a failure after a completion is logged without carrying the usage again', async () => { + const r = await resident([() => new MockYieldingStream([finalResponse('Done.', totalUsage)])]); + // Something after the completion's entry throws: the stream's catch + // writes a failure entry too. (A resident: an ephemeral is deregistered + // once its completion settles it, so its catch treats the throw as a + // superseded stream's.) + const internal = r.framework as unknown as { endRefusalEpisode: (name: string) => void }; + const end = internal.endRefusalEpisode.bind(r.framework); + internal.endRefusalEpisode = () => { + internal.endRefusalEpisode = end; + throw new Error('after the completion'); + }; + r.framework.start(); + try { + r.framework.pushEvent({ type: 'external-message', source: 'test', content: 'go', metadata: {} } as unknown as ProcessEvent); + assert.ok(await pollUntil(() => r.framework.queryInferenceLogs({ agentName: 'resident' }).total === 2), 'both entries'); + // Both entries carry the stream's start time, so find each by outcome. + const logs = r.framework.queryInferenceLogs({ agentName: 'resident' }).entries.map((e) => e.entry); + const completed = logs.find((e) => e.success); + const failed = logs.find((e) => !e.success); + assert.deepEqual(completed?.tokenUsage, expectedTokens(totalUsage), 'the completion carries the usage'); + assert.equal(failed?.error, 'Stream threw: after the completion'); + assert.equal(failed?.tokenUsage, undefined, 'the later failure does not carry it again'); + assert.deepEqual(r.framework.getSessionUsage().totals, expectedTotals(totalUsage)); + } finally { + await r.framework.stop(); + rmSync(r.dir, { recursive: true, force: true }); + } + }); + + it('a stream superseded by a newer one after a finished round', async () => { + const stream = new RoundThenEnd(firstUsage, unseenCompletion()); + let framework: AgentFramework | undefined; + const provide = stream.provideToolResults.bind(stream); + // A newer stream of the agent takes over while this one waits on its tool. + stream.provideToolResults = () => { + (framework!.getAgent('resident') as unknown as { _streamId: number })._streamId++; + provide(); + }; + const r = await residentWith(stream, new TurnTools()); + framework = r.framework; + try { + assert.ok(await pollUntil(() => r.framework.queryInferenceLogs({ agentName: 'resident' }).total > 0), 'logged'); + loggedOnce(r.framework, r.traces, /^Stream superseded/); + } finally { + await r.framework.stop(); + rmSync(r.dir, { recursive: true, force: true }); + } + }); +}); diff --git a/test/tool-result-guard.test.ts b/test/tool-result-guard.test.ts index 510ca08b9..9fc1f5001 100644 --- a/test/tool-result-guard.test.ts +++ b/test/tool-result-guard.test.ts @@ -528,17 +528,45 @@ test('usage of an abandoned guarded round is counted in session totals', async ( }); test('an abandoned guarded round whose response carries no usage still counts its rounds', async () => { - // Each scripted response emits a usage sample of 10 input tokens; neither - // final response carries details.usage, so both streams count their samples. - const h = await harness([[calls('one'), refused()], [answer()]], { toolResultGuard: true }); + // Membrane's usage samples are cumulative across a stream's rounds: the + // withheld stream's two rounds emit 10, then 20 input tokens, and the + // retry's one round 10. No final response carries details.usage, so each + // stream counts its latest sample. + const sampled = (response: NormalizedResponse, inputTokens: number, outputTokens: number) => + ({ ...response, usage: { inputTokens, outputTokens } }) as NormalizedResponse; + const h = await harness([[sampled(calls('one'), 10, 5), sampled(refused(), 20, 10)], [sampled(answer(), 10, 5)]], { toolResultGuard: true }); try { await h.run(); const totals = h.framework.getSessionUsage().totals as unknown as Record; - assert.equal(totals.inputTokens, 20, `each stream is counted once, from its samples: ${JSON.stringify(totals)}`); + assert.equal(totals.inputTokens, 30, `each stream is counted once, from its latest sample: ${JSON.stringify(totals)}`); // The log says what the totals count: the withheld stream, then the retry. const logs = h.framework.queryInferenceLogs({ agentName: 'assistant' }).entries.map((e) => e.entry).reverse(); assert.deepEqual(logs.map((e) => [e.success, e.tokenUsage]), - [[false, { input: 10, output: 5 }], [true, { input: 10, output: 5 }]]); + [[false, { input: 20, output: 10 }], [true, { input: 10, output: 5 }]]); + } finally { await h.framework.stop(); } +}); + +test("an abandoned guarded round with no usage sample counts and notes its response's own usage", async () => { + const h = await harness([[calls('one'), refused()], [answer()]], { toolResultGuard: true }); + // A stream that emits no usage events: only the refused response's own + // usage (10 input tokens, from the shared mock) says what it was billed. + const make = h.membrane.streamYielding.bind(h.membrane); + h.membrane.streamYielding = (request, options) => { + const stream = make(request, options); + const events = stream[Symbol.asyncIterator].bind(stream); + stream[Symbol.asyncIterator] = async function* () { + const it = events(); + for (let r = await it.next(); !r.done; r = await it.next()) if (r.value.type !== 'usage') yield r.value; + }; + return stream; + }; + try { + await h.run(); + const totals = h.framework.getSessionUsage().totals as unknown as Record; + assert.equal(totals.inputTokens, 10, `the refused round is counted from its response: ${JSON.stringify(totals)}`); + const withheld = h.framework.queryInferenceLogs({ agentName: 'assistant' }).entries.map((e) => e.entry) + .find((e) => e.stopReason === 'refusal'); + assert.deepEqual(withheld?.tokenUsage, { input: 10, output: 5 }, "the withheld stream's entry carries the response's usage"); } finally { await h.framework.stop(); } }); From bcc8d00807051cd3c3f52aaab400b7ed67b1963a Mon Sep 17 00:00:00 2001 From: Milo-2653 Date: Fri, 16 Oct 2026 15:44:54 -0400 Subject: [PATCH 09/13] fix(usage): a conversation agent's provider wait logs its stream's one entry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit main's 5a45eb6e (Oct 11) added a failure site in driveStream's catch: a turn that doesn't own provider admission, whose stated wait holds it (nonOwningProviderWait answers 'held'), writes a "Provider wait: …" entry. It wrote it with logInference directly, past this branch's once-only log, so a stream that had finished rounds got two entries: the wait's, without usage, and the one `finally` writes for an end that wrote none ("Stream ended without a final response"), with it. It now writes through logStream, as the cooldown entry beside it does, so the wait's entry is the stream's one entry and carries the rounds it finished. That was the last direct logInference in driveStream. Test: a conversation agent whose stream finishes a round and then throws a stated 429 logs one entry, the wait's, carrying that round, and counts it once. Co-Authored-By: Claude Opus 5.5 --- src/framework.ts | 2 +- test/end-turn-usage.test.ts | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/src/framework.ts b/src/framework.ts index fdea5aac0..a44f7ad17 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -13651,7 +13651,7 @@ export class AgentFramework { const statedWait = ownsProviderGate ? undefined : this.nonOwningProviderWait(agent, err, trigger); if (statedWait === 'held') { this.emitTrace({ type: 'inference:failed', agentName: agent.name, error: err.message, stack: err.stack }); - this.logInference({ timestamp: startTime, agentName: agent.name, requestId, success: false, + logStream({ success: false, error: `Provider wait: ${err.message}`, request: compiledRequest ?? { note: 'streaming request rate-limited' }, durationMs }); this.abortAgentScript(agent.name, 'provider wait'); diff --git a/test/end-turn-usage.test.ts b/test/end-turn-usage.test.ts index 2e026490f..39bc220db 100644 --- a/test/end-turn-usage.test.ts +++ b/test/end-turn-usage.test.ts @@ -603,6 +603,40 @@ describe('a tool-ended turn ends what a successful turn ends', () => { rmSync(r.dir, { recursive: true, force: true }); } }); + + it("a conversation agent's stated provider wait that cuts a turn after a finished round logs and counts that round once", async () => { + const statedWait = (request: unknown) => new MembraneError({ + type: 'rate_limit', retryable: true, httpStatus: 429, retryAfterMs: 60_000, + message: 'zz rate limit reached for requests', rawError: { status: 429 }, rawRequest: request, + }); + const r = await resident([(request) => new RoundThenEnd(firstUsage, statedWait(request))]); + // A conversation agent doesn't own provider admission, so a stated wait + // holds its turn (the "Provider wait" end) instead of the organization's + // cooldown. Its turns aren't an external message's default target, so + // the turn is started directly. + const internal = r.framework as unknown as { + conversationAgentHomes: Map; + startAgentStream(agent: unknown, trigger: unknown): Promise; + }; + r.framework.start(); + try { + internal.conversationAgentHomes.set('resident', 'world:zz'); + r.framework.getAgent('resident')!.getContextManager().addMessage('User', [{ type: 'text', text: 'first' }]); + await internal.startAgentStream(r.framework.getAgent('resident'), + { agentName: 'resident', reason: 'conversation', source: 'test', timestamp: Date.now() }); + assert.ok(await pollUntil(() => r.traces.some((e) => e.type === 'usage:updated') + && r.framework.queryInferenceLogs({ agentName: 'resident' }).total > 0), 'the finished round was counted and logged'); + const logs = r.framework.queryInferenceLogs({ agentName: 'resident' }).entries.map((e) => e.entry); + assert.equal(logs.length, 1, 'the wait writes the stream\'s one entry'); + assert.match(String(logs[0].error), /^Provider wait: /); + assert.deepEqual(logs[0].tokenUsage, expectedTokens(firstUsage), 'and it carries the finished round'); + assert.deepEqual(r.framework.getSessionUsage().totals, expectedTotals(firstUsage)); + assert.equal(r.traces.filter((e) => e.type === 'usage:updated').length, 1); + } finally { + await r.framework.stop(); + rmSync(r.dir, { recursive: true, force: true }); + } + }); }); /** A completion the stream would yield next; these tests end it before it's handled. */ From 6cd06892409a5e08ce0da6fd7f744458a64d8b66 Mon Sep 17 00:00:00 2001 From: Milo-2653 Date: Fri, 16 Oct 2026 15:46:48 -0400 Subject: [PATCH 10/13] docs(usage): the failure streak resets when a turn completes, not when a call answers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit emitTrace clears the consecutive-failure streak and the poison-history breaker's rewind budget at inference:completed, which this branch sends for a tool-ended turn too. Its comment said "a successful response — even mid-turn between tool calls — proves the agent isn't hard-down", which the code didn't do. A reset at each answered call (inference:usage, which membrane sends once a round's provider stream has returned) would make it true, and would hide what the streak is for: noteInferenceExhausted raises the hard-down alert ("it cannot complete a turn") and runs the breaker only at a streak of 3, so an agent whose every turn has a first call that answers and a later one that fails would stay at 1, with neither. The comment now says what resets it and why an answered call doesn't, and quarantinePoisonedHistory's doc says the budget resets when a turn completes. Test: three turns that each fail with a 400 after a call that answered count 1, 2, 3, raise one hard-down alert and run the breaker's shed. Co-Authored-By: Claude Opus 5.5 --- src/framework.ts | 13 ++++++---- test/end-turn-usage.test.ts | 47 +++++++++++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 4 deletions(-) diff --git a/src/framework.ts b/src/framework.ts index a44f7ad17..4f9285aaa 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -15727,9 +15727,14 @@ export class AgentFramework { }, ); } else if (event.type === 'inference:completed') { - // A successful response — even mid-turn between tool calls — proves the - // agent isn't hard-down; clear its consecutive-failure streak (and the - // poison-history breaker's rewind budget). + // A completed turn, a tool-ended one included, shows the agent can + // complete turns: clear its consecutive-failure streak and the + // poison-history breaker's rewind budget. A provider call answered + // mid-turn (inference:usage) doesn't. The streak exists to catch an + // agent that can't complete a turn, and an agent whose every turn has a + // first call that answers and a later one that fails is exactly that: + // a reset per answered call would keep it below the hard-down + // threshold, so neither the alert nor the breaker would come. const name = event.agentName as string | undefined; if (name && this.consecutiveInferenceFailures.get(name)) { this.consecutiveInferenceFailures.set(name, 0); @@ -16232,7 +16237,7 @@ export class AgentFramework { * push event to wake the agent onto the same context. * * Bounded: at most `refusalHandling.maxRewinds` (default 3, hard cap 10) - * sheds per failure episode; the budget resets on any successful inference. + * sheds per failure episode; the budget resets when a turn completes. * At the cap (or with nothing left to shed) it stops — the agent stays up * and hard-down logging continues, but no further history is consumed. */ diff --git a/test/end-turn-usage.test.ts b/test/end-turn-usage.test.ts index 39bc220db..819928d97 100644 --- a/test/end-turn-usage.test.ts +++ b/test/end-turn-usage.test.ts @@ -637,6 +637,53 @@ describe('a tool-ended turn ends what a successful turn ends', () => { rmSync(r.dir, { recursive: true, force: true }); } }); + + it('turns that each fail after an answered call reach the hard-down threshold and run the breaker: an answered call is not a completed turn', async () => { + // Each turn's first call answers (its round's usage arrives) and the + // call after its tool result is refused as the provider can't take it, + // which shedMayHelp matches. + const refused = (): StreamEvent => ({ + type: 'error', + error: new MembraneError({ + type: 'invalid_request', retryable: false, httpStatus: 400, + message: 'zz the provider could not process the tool result', rawError: { status: 400 }, + }), + }) as StreamEvent; + const r = await resident([1, 2, 3].map(() => () => new RoundThenEnd(firstUsage, refused()))); + const internal = r.framework as unknown as { consecutiveInferenceFailures: Map }; + const streaks: number[] = []; + const answeredBefore: number[] = []; + let answered = 0; + // emitTrace counts the failure before listeners hear it. + r.framework.onTrace((event) => { + if (event.type === 'inference:usage') answered++; + if (event.type === 'inference:exhausted') { + streaks.push(internal.consecutiveInferenceFailures.get('resident') ?? 0); + answeredBefore.push(answered); + } + }); + const lines: string[] = []; + const original = console.error; + console.error = (...args: unknown[]) => { lines.push(args.map(String).join(' ')); }; + r.framework.start(); + try { + for (const [turn, text] of ['first', 'second', 'third'].entries()) { + r.framework.pushEvent({ type: 'external-message', source: 'test', content: text, metadata: {} } as unknown as ProcessEvent); + await r.framework.runUntilIdle(); + assert.ok(await pollUntil(() => streaks.length === turn + 1), `turn ${turn + 1} failed`); + } + assert.deepEqual(answeredBefore, [1, 2, 3], 'each failed turn had a call that answered'); + assert.deepEqual(streaks, [1, 2, 3], 'an answered call leaves the streak standing'); + const alerts = r.traces.filter((e) => e.type === 'ops:alert' && (e as { kind?: string }).kind === 'hard-down'); + assert.equal(alerts.length, 1, 'the third failed turn raises the hard-down alert'); + assert.ok(lines.some((line) => /^\[inference-rewind\] agent=resident auto-quarantined /.test(line)), + 'and the poison-history breaker sheds the newest exchange'); + } finally { + console.error = original; + await r.framework.stop(); + rmSync(r.dir, { recursive: true, force: true }); + } + }); }); /** A completion the stream would yield next; these tests end it before it's handled. */ From bc125bb2756b380497f796dd0399359c75b8d424 Mon Sep 17 00:00:00 2001 From: Milo-2653 Date: Fri, 16 Oct 2026 21:41:50 -0400 Subject: [PATCH 11/13] fix(usage): a failed write can't skip a stream's teardown; texts and tests for what each end logs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A review of this branch merged with main found these, each checked against the code: - driveStream's finally now writes to the store before its teardown: it counts the stream's usage and, for an end that wrote no entry, logs one. logInference has no try/catch, so a store error there skipped the lifecycle terminal, the event gate's release, the turn token and the KV receipts, and rejected driveStream's promise. The accounting is now guarded: a failed write is said on stderr ("[usage] agent=… could not record a stream's usage at its end") and the teardown runs. - A refusal completion's note took its tokens from the final response's detailed usage alone, while the count falls back to the stream's latest sample. The note now carries what was counted, as the guard's refusal note already did. - Texts said more than the code does: - an end that writes no entry of its own (a quiesce abandon, a superseded stream, a disposed agent's stream) gets one, and is counted, only when it had finished a round; with none there is nothing to carry, and nothing is written, as before. The changelog, the InferenceLogEntry doc and the finally's comment say so; - totals summed from the inference log count each stream once, as the session's do, rather than "agreeing" with them: a health snapshot's token totals read the log's newest 1,000 entries, which these entries now fill with more turns. The changelog says so; - the changelog's caller list says a consumer that totals inference:completed's tokenUsage now counts tool-ended turns, and one that also adds every usage sample counts them twice; its growth note names the failure entry, with its request, of an abandoned, superseded or disposed stream; - the module contract's onTrace doc says a tool-ended turn emits inference:completed and then inference:turn_ended; - the field docs of the failure streak and the poison-history rewind budget say they reset when a turn completes, and the comments on the stream's count and its one log entry say which value each end counts and which entry carries it. Tests: the replaced-agent and closed-stream texts of the finally's entry; a quiesce abandon before the first round returns writes no entry and counts nothing; a write that fails in finally leaves the stream and its turn released; a refusal completion's note carries the counted tokens. The tool-ended and restart tests wait for every driven stream's teardown instead of one setImmediate, and the failure-end test clears its 10 s bound. Co-Authored-By: Claude Opus 5.5 --- .../fix-67-end-turn-accounting.fixed.md | 14 +- src/framework.ts | 85 +++++---- src/types/framework.ts | 4 +- src/types/module.ts | 3 +- test/end-turn-usage.test.ts | 173 +++++++++++++++++- 5 files changed, 227 insertions(+), 52 deletions(-) diff --git a/changelog.d/fix-67-end-turn-accounting.fixed.md b/changelog.d/fix-67-end-turn-accounting.fixed.md index 8eaffbd16..f660cf7bd 100644 --- a/changelog.d/fix-67-end-turn-accounting.fixed.md +++ b/changelog.d/fix-67-end-turn-accounting.fixed.md @@ -6,16 +6,22 @@ a refusal-rewind episode (a forced `/unstick` reports that the model responded), and records a provider-cooldown recovery. In hosts such as connectome-host, consumers that end a turn's display on `inference:completed` (a typing indicator, notices that close on success, a voice relay) now - end it on these turns too. + end it on these turns too, and a consumer that totals `inference:completed`'s `tokenUsage` now + counts them too; one that also adds each `inference:usage` sample counts them twice. - **Every stream's billed rounds reach the session totals once, however the stream ends.** A context-budget or physical-window restart is counted and logged where it is decided (`stopReason: 'context_budget'` or `'physical_window'`), without `inference:completed`. A stream that errors, aborts or throws after finishing rounds counts them, and its failure entry in the inference log carries them. A turn abandoned by an operator quiesce, a stream superseded by a newer one, or one whose agent was disposed, which wrote no entry before, now writes a failure - entry naming what ended it and carrying its rounds. `inferenceCount` counts these streams too. Exactly one entry of each stream - carries its usage, so log-backed health totals agree with the session totals. + entry naming what ended it and carrying its rounds, and `inferenceCount` counts it, when it had + finished a round; one that hadn't has nothing to carry and writes none, as before. Exactly one + entry of each stream carries its usage, so totals summed from the inference log count each + stream once, as the session's do. A health snapshot's token totals read the log's newest 1,000 + entries, so with these new entries that window spans fewer turns. - An inference-log entry for a stream the framework ended at a tool boundary (a tool-ended turn or a restart) records a note in place of the compiled request, so these frequent entries don't each add a whole-context blob to the store. They are still new entries: the inference log keeps no - retention, so it grows by one small entry for every tool-ended turn and restart. + retention, so it grows by one small entry for every tool-ended turn and restart, and by one + entry, with its request, for each abandoned, superseded or disposed stream that had finished a + round. diff --git a/src/framework.ts b/src/framework.ts index 4f9285aaa..c23a4879b 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -1760,7 +1760,8 @@ export class AgentFramework { * The name is single-generation, so these frames still own their terminal * typing/outgoing close even after runEphemeralToCompletion deregisters them. */ private disposedEphemeralFrames: WeakSet = new WeakSet(); - /** Per-agent count of consecutive exhausted inferences (reset on any success). + /** Per-agent count of consecutive exhausted inferences, reset when a turn + * completes (inference:completed), not when a call answers mid-turn. * Drives hard-down escalation — see noteInferenceExhausted. */ private consecutiveInferenceFailures: Map = new Map(); /** N consecutive failed inferences ⇒ the agent is treated as hard-down and @@ -1784,7 +1785,8 @@ export class AgentFramework { * rewind loop — see refusalHandling + rewindTriggeringTurn. */ private refusalRewinds: Map = new Map(); /** Per-agent count of poison-history rewinds performed by the hard-down - * breaker (noteInferenceExhausted). Reset on any successful inference. + * breaker (noteInferenceExhausted). Reset when a turn completes + * (inference:completed). * Bounds the automatic quarantine loop the same way refusalRewinds bounds * the refusal loop, so the breaker can never shed the whole history. */ private exhaustionRewinds: Map = new Map(); @@ -12195,10 +12197,12 @@ export class AgentFramework { } : undefined; // The stream's billed usage reaches the session totals once, however the - // stream ends: a final response's usage where it carries one (completion, - // or the guard's abandoned round), else the rounds it finished (a final - // response without usage, the boundary ends below, or `finally` for - // every other end). Each caller passes the one value it also reports. + // stream ends. A completion counts its final response's detailed usage, + // and the guard's abandoned round its detailed usage, else the latest + // sample, else the response's own. Every other end counts the rounds it + // finished: a final response without detailed usage, the boundary ends + // below, and `finally` for the rest. Each caller passes the one value it + // also reports. let usageCounted = false; let countedUsage: DetailedUsage | undefined; const countUsage = (usage: DetailedUsage | undefined): void => { @@ -12212,10 +12216,12 @@ export class AgentFramework { this.persistUsageState(); }; // ...and is carried by exactly one of its inference-log entries, so - // log-backed totals agree with the session's. The first entry carries - // it (the value the entry names, else the rounds finished so far); no - // later entry of this stream carries it again. An end that writes no - // entry of its own is logged in `finally`. + // totals summed from the log count each stream once, as the session's + // do. The first entry that has usage to carry carries it (the value the + // entry names, else the rounds finished so far): an entry before it with + // none leaves it to the next, and no entry after it carries it again. An + // end that writes no entry of its own is logged in `finally`, when it has + // rounds to carry. let usageLogged = false; const logStream = (entry: Omit): void => { const tokenUsage = usageLogged ? undefined : (entry.tokenUsage ?? tokenUsageOf(latestUsage)); @@ -12964,14 +12970,8 @@ export class AgentFramework { const admittedSegments = admitFinal ? finalSegments.map((seg) => this.admitProse(seg)) : []; const du = response.details?.usage; - const tokenUsage = du - ? { - input: du.inputTokens, - output: du.outputTokens, - cacheCreation: du.cacheCreationTokens, - cacheRead: du.cacheReadTokens, - } - : undefined; + // A refusal's note carries what completeTurn counts. + const tokenUsage = tokenUsageOf(du ?? latestUsage); // Async completion work may overlap a replacement stream. Recheck // physical ownership before any terminal state/settlement side effect. @@ -13689,27 +13689,34 @@ export class AgentFramework { agent.reset(); if (this.agents.get(agent.name) === agent && agent.streamId === myStreamId) this.eventGate?.onInferenceEnded(agent.name); } finally { - // Every end the cases above didn't count (a stream error or abort, a - // throw, a superseded generation) still billed the rounds it finished. - countUsage(latestUsage); - // An end that wrote no entry carrying them (a quiesce-abandoned turn, - // a stream superseded by a newer one, a disposed agent's stream) writes - // one now, as a failure naming what ended it, so the log carries what - // the totals count. - if (countedUsage && !usageLogged) { - const replaced = this.agents.get(agent.name) !== agent; - const superseded = generationLost || agent.streamId !== myStreamId; - logStream({ - success: false, - error: unloggedEnd ?? (replaced - ? 'Stream abandoned: its agent was disposed or replaced' - : superseded - ? 'Stream superseded: a newer stream of the agent took over' - : 'Stream ended without a final response'), - request: compiledRequest ?? { note: 'streaming request' }, - durationMs: Date.now() - startTime, - tokenUsage: tokenUsageOf(countedUsage), - }); + // The accounting writes to the store, and a write that fails must not + // skip the teardown after it. + try { + // Every end the cases above didn't count (a stream error or abort, a + // throw, a superseded generation) still billed the rounds it finished. + countUsage(latestUsage); + // An end that wrote no entry carrying them (a quiesce-abandoned turn, + // a stream superseded by a newer one, a disposed agent's stream) writes + // one now, as a failure naming what ended it, so the log carries what + // the totals count. With no finished round there is nothing to carry, + // and no entry is written, as before. + if (countedUsage && !usageLogged) { + const replaced = this.agents.get(agent.name) !== agent; + const superseded = generationLost || agent.streamId !== myStreamId; + logStream({ + success: false, + error: unloggedEnd ?? (replaced + ? 'Stream abandoned: its agent was disposed or replaced' + : superseded + ? 'Stream superseded: a newer stream of the agent took over' + : 'Stream ended without a final response'), + request: compiledRequest ?? { note: 'streaming request' }, + durationMs: Date.now() - startTime, + tokenUsage: tokenUsageOf(countedUsage), + }); + } + } catch (error) { + console.error(`[usage] agent=${agent.name} could not record a stream's usage at its end: ${describeFailureWithFrames(error)}`); } const unsettledKvSubmissions = drainKvSubmissionIds?.() ?? []; if (unsettledKvSubmissions.length > 0) { diff --git a/src/types/framework.ts b/src/types/framework.ts index 35221d27d..003ed8d92 100644 --- a/src/types/framework.ts +++ b/src/types/framework.ts @@ -424,7 +424,9 @@ export interface InferenceLogEntry { * without one, the provider rounds it finished before it ended. Exactly * one entry of a stream carries it, the one its end wrote; a stream whose * end writes none (a quiesce-abandoned turn, a superseded stream, a - * disposed agent's stream) gets a failure entry that carries it. */ + * disposed agent's stream) gets a failure entry that carries it, when it + * had finished a round. One that hadn't has nothing to carry and gets no + * entry. */ tokenUsage?: { input: number; output: number; diff --git a/src/types/module.ts b/src/types/module.ts index a6fda9ed2..605fcccbb 100644 --- a/src/types/module.ts +++ b/src/types/module.ts @@ -322,7 +322,8 @@ export interface ModuleContext { * Modules use this to react to agent lifecycle without polling — e.g. an * adapter that wants to show a "typing" indicator can bracket it between * `inference:started` and the terminal events: `inference:completed` (the - * normal end), `inference:turn_ended` (a tool result ended the turn), + * model answered and the turn ended, with a final response or with a tool + * result that ended it, which `inference:turn_ended` then follows), * `inference:aborted`, `inference:failed` and `inference:exhausted`, all of * which carry `agentName`. A turn's tool-call cycles stay inside its * stream, so one bracket spans them. A context-budget restart emits diff --git a/test/end-turn-usage.test.ts b/test/end-turn-usage.test.ts index 819928d97..9fcbe8ea0 100644 --- a/test/end-turn-usage.test.ts +++ b/test/end-turn-usage.test.ts @@ -86,15 +86,34 @@ async function fixture() { return { dir, storePath, membrane, framework, traces, completionStates, countedAt }; } +/** Every stream the framework has driven so far, by the promise driveStream + * returns, so a test can wait until each one's teardown has run. */ +const driven = new WeakMap[]>(); +function streamsDriven(framework: AgentFramework): () => Promise { + let handles = driven.get(framework); + if (!handles) { + const all: Promise[] = []; + handles = all; + driven.set(framework, all); + const streams = (framework as unknown as { activeStreams: Map> }).activeStreams; + const set = streams.set.bind(streams); + streams.set = (name, handle) => { all.push(handle); return set(name, handle); }; + } + const watched = handles; + return async () => { await Promise.allSettled(watched); }; +} + async function run(framework: AgentFramework, name: string) { + const tornDown = streamsDriven(framework); const { agent, contextManager } = await framework.createEphemeralAgent({ name, model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', proseRouting: 'disabled', }); contextManager.addMessage('user', [{ type: 'text', text: 'Run once.' }]); const result = await framework.runEphemeralToCompletion(agent, contextManager); - // Also observe asynchronous cancellation teardown before checking counts. - await new Promise((resolve) => setImmediate(resolve)); + // A cancelled stream's teardown runs after the run settles: wait for it + // before checking counts. + await tornDown(); return result; } @@ -286,7 +305,8 @@ class SequentialStreamMembrane extends MockMembrane { } } -/** A stream that finishes one tool round with `usage`, then ends as `ending` says. */ +/** A stream that finishes one tool round with `usage`, then ends as `ending` + * says: with that event, by throwing it, or, for 'close', by just closing. */ class RoundThenEnd implements YieldingStream { isWaitingForTools = false; pendingToolCallIds: string[] = []; @@ -294,7 +314,7 @@ class RoundThenEnd implements YieldingStream { isCancelled = false; private resumed = false; private resume: (() => void) | null = null; - constructor(private readonly usage: DetailedUsage, private readonly ending: StreamEvent | Error) {} + constructor(private readonly usage: DetailedUsage, private readonly ending: StreamEvent | Error | 'close') {} provideToolResults(): void { this.isWaitingForTools = false; this.resumed = true; @@ -320,6 +340,7 @@ class RoundThenEnd implements YieldingStream { yield { type: 'aborted', reason: 'user' } as StreamEvent; return; } + if (this.ending === 'close') return; if (this.ending instanceof Error) throw this.ending; yield this.ending; } @@ -386,6 +407,7 @@ describe('a stream counts its usage once, however it ends', () => { if (event.type === 'inference:stream_restarted') countedAtRestart.push(framework.getSessionUsage().inferenceCount); }); framework.start(); + const tornDown = streamsDriven(framework); try { const { agent, contextManager } = await framework.createEphemeralAgent({ name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', @@ -393,7 +415,8 @@ describe('a stream counts its usage once, however it ends', () => { }); contextManager.addMessage('user', [{ type: 'text', text: 'Run once.' }]); await framework.runEphemeralToCompletion(agent, contextManager); - await new Promise((resolve) => setImmediate(resolve)); + // The cancelled stream's teardown has run, so whatever it would log is in. + await tornDown(); assert.equal(membrane.calls.length, 2, 'the restart opened a second stream'); assert.equal((traces.find((e) => e.type === 'inference:stream_restarted') as { reason?: string })?.reason, reason); @@ -445,10 +468,11 @@ describe('a stream counts its usage once, however it ends', () => { // The run settles once the failure path has decided: it rejects with // the stream's failure, or, had a retry been scheduled, it would wait // for it and resolve on the default response. + let bound: ReturnType | undefined; const outcome = await Promise.race([ framework.runEphemeralToCompletion(agent, contextManager).then(() => 'resolved', (error: unknown) => error), - new Promise((resolve) => setTimeout(() => resolve('timed out'), 10_000)), - ]); + new Promise((resolve) => { bound = setTimeout(() => resolve('timed out'), 10_000); }), + ]).finally(() => clearTimeout(bound)); assert.ok(outcome instanceof Error, `the run failed with the stream: ${String(outcome)}`); assert.equal(membrane.calls.length, 1, 'no retry was scheduled by the time the run failed'); assert.ok(await pollUntil(() => traces.some((e) => e.type === 'usage:updated')), 'the finished round was counted'); @@ -638,6 +662,33 @@ describe('a tool-ended turn ends what a successful turn ends', () => { } }); + it("a refusal completion's note carries what its count counted", async () => { + // The refused response carries no detailed usage, so the stream is + // counted from its latest usage sample (10 input, 5 output). + const r = await resident([() => new MockYieldingStream([createMockResponse([{ type: 'text', text: 'zz-refused' }], 'refusal')])]); + const notes: Array<{ tokens: unknown; totals: unknown }> = []; + const internal = r.framework as unknown as { logFailure: (entry: Record) => void }; + const logFailure = internal.logFailure.bind(r.framework); + // What the session had counted when the refusal was noted: this stream + // alone, before any retry a rewind might queue. + internal.logFailure = (entry) => { + if (entry.kind === 'refusal') notes.push({ tokens: entry.tokens, totals: structuredClone(r.framework.getSessionUsage().totals) }); + logFailure(entry); + }; + r.framework.start(); + try { + r.framework.pushEvent({ type: 'external-message', source: 'test', content: 'first', metadata: {} } as unknown as ProcessEvent); + await r.framework.runUntilIdle(); + assert.ok(await pollUntil(() => notes.length > 0), 'the refusal was noted'); + const counted: DetailedUsage = { inputTokens: 10, outputTokens: 5 }; + assert.deepEqual(notes[0].totals, expectedTotals(counted), 'counted from the latest sample'); + assert.deepEqual(notes[0].tokens, expectedTokens(counted), 'and the note carries the same'); + } finally { + await r.framework.stop(); + rmSync(r.dir, { recursive: true, force: true }); + } + }); + it('turns that each fail after an answered call reach the hard-down threshold and run the breaker: an answered call is not a completed turn', async () => { // Each turn's first call answers (its round's usage arrives) and the // call after its tool result is refused as the provider can't take it, @@ -690,6 +741,26 @@ describe('a tool-ended turn ends what a successful turn ends', () => { const unseenCompletion = (): StreamEvent => ({ type: 'complete', response: createMockResponse([{ type: 'text', text: 'unseen' }]) }) as StreamEvent; +/** A stream whose first round never returns: it waits until it's cancelled. */ +class FirstRoundHeld implements YieldingStream { + isWaitingForTools = false; + pendingToolCallIds: string[] = []; + toolDepth = 0; + isCancelled = false; + started = false; + private wake: (() => void) | null = null; + provideToolResults(): void { throw new Error('not waiting'); } + cancel(): void { + this.isCancelled = true; + this.wake?.(); + } + async *[Symbol.asyncIterator](): AsyncIterator { + this.started = true; + if (!this.isCancelled) await new Promise((resolve) => { this.wake = resolve; }); + yield { type: 'aborted', reason: 'user' } as StreamEvent; + } +} + /** A tool that returns only once released, so the stream waits on it. */ class HeldTools implements Module { readonly name = 'test'; @@ -790,6 +861,94 @@ describe('a stream that ends without writing its own entry is logged once, in fi } }); + it('a stream whose agent was replaced under its name after a finished round', async () => { + const stream = new RoundThenEnd(firstUsage, unseenCompletion()); + let framework: AgentFramework | undefined; + let original: object | undefined; + const agentsOf = () => (framework as unknown as { agents: Map }).agents; + const provide = stream.provideToolResults.bind(stream); + // The agent object is replaced under its name while this stream waits on its tool. + stream.provideToolResults = () => { + original = agentsOf().get('resident'); + agentsOf().set('resident', Object.create(original!)); + provide(); + }; + const r = await residentWith(stream, new TurnTools()); + framework = r.framework; + try { + assert.ok(await pollUntil(() => r.framework.queryInferenceLogs({ agentName: 'resident' }).total > 0), 'logged'); + loggedOnce(r.framework, r.traces, /^Stream abandoned: its agent was disposed or replaced$/); + } finally { + if (original) agentsOf().set('resident', original); + await r.framework.stop(); + rmSync(r.dir, { recursive: true, force: true }); + } + }); + + it('a stream that just closes after a finished round', async () => { + const r = await residentWith(new RoundThenEnd(firstUsage, 'close'), new TurnTools()); + try { + assert.ok(await pollUntil(() => r.framework.queryInferenceLogs({ agentName: 'resident' }).total > 0), 'logged'); + loggedOnce(r.framework, r.traces, /^Stream ended without a final response$/); + } finally { + await r.framework.stop(); + rmSync(r.dir, { recursive: true, force: true }); + } + }); + + it('a turn abandoned by an operator quiesce before its first round returns has nothing to carry, and writes no entry', async () => { + const stream = new FirstRoundHeld(); + const r = await residentWith(stream, new TurnTools()); + const internal = r.framework as unknown as { activeTurnTokens: Map; activeStreams: Map }; + try { + assert.ok(await pollUntil(() => stream.started), 'the stream is running'); + await r.framework.quiesce({ timeoutMs: 50, abandon: true }); + assert.ok(await pollUntil(() => stream.isCancelled), 'the abandon cancelled it'); + // The stream and its turn are released at the end of finally, after its accounting. + assert.ok(await pollUntil(() => !internal.activeTurnTokens.has('resident') && !internal.activeStreams.has('resident')), + 'its teardown ran'); + assert.equal(r.framework.queryInferenceLogs({ agentName: 'resident' }).total, 0, 'no entry'); + assert.equal(r.traces.filter((e) => e.type === 'usage:updated').length, 0, 'nothing counted'); + assert.equal(r.framework.getSessionUsage().inferenceCount, 0); + } finally { + await r.framework.stop(); + rmSync(r.dir, { recursive: true, force: true }); + } + }); + + it("a write that fails while finally logs the stream doesn't skip the teardown after it", async () => { + const tools = new HeldTools(); + const r = await residentWith(new RoundThenEnd(firstUsage, unseenCompletion()), tools); + const internal = r.framework as unknown as { + logInference: (entry: { error?: string }) => void; + activeTurnTokens: Map; + activeStreams: Map; + }; + const write = internal.logInference.bind(r.framework); + internal.logInference = (entry) => { + if (entry.error === 'Turn abandoned by operator quiesce') throw new Error('zz store refused the write'); + write(entry); + }; + const lines: string[] = []; + const original = console.error; + console.error = (...args: unknown[]) => { lines.push(args.map(String).join(' ')); }; + try { + assert.ok(await pollUntil(() => tools.calledWith.length === 1), 'the round finished and its tool is running'); + await r.framework.quiesce({ timeoutMs: 50, abandon: true }); + assert.ok(await pollUntil(() => lines.some((line) => line.startsWith('[usage] agent=resident could not record'))), + 'the failed write is said'); + assert.ok(await pollUntil(() => !internal.activeTurnTokens.has('resident') && !internal.activeStreams.has('resident')), + 'and the stream and its turn are still released'); + } finally { + console.error = original; + tools.release(); + await pollUntil(() => r.traces.some((e) => + (e.type === 'tool:completed' || e.type === 'tool:failed') && (e as { callId?: string }).callId === 'echo-1')); + await r.framework.stop(); + rmSync(r.dir, { recursive: true, force: true }); + } + }); + it('a stream superseded by a newer one after a finished round', async () => { const stream = new RoundThenEnd(firstUsage, unseenCompletion()); let framework: AgentFramework | undefined; From 57ad6586d0c06f0be5d8c0400d9f7206c915d2b2 Mon Sep 17 00:00:00 2001 From: Milo-2653 Date: Sat, 17 Oct 2026 02:52:24 -0400 Subject: [PATCH 12/13] docs(usage): say at logInference that a stream's entries go through its logStream; name two more ends in the changelog logStream, inside driveStream, gives a stream's usage to one inference-log entry only. An end that called logInference directly for a stream would write a second entry for it, and an entry carrying the usage would count it twice in the log. Saying so where a new end would reach for logInference points the next change at the rule. The changelog fragment now names two ends it left out: a stream that ended with no final response and no other end, which writes its failure entry as the abandoned, superseded and disposed ones do, and a store error while a stream's end is recorded, which is reported on stderr while the stream's teardown still runs. Both have tests ("a stream that just closes after a finished round", "a write that fails while finally logs the stream doesn't skip the teardown after it"). Comments and the changelog only. Co-Authored-By: Claude Opus 5.5 --- changelog.d/fix-67-end-turn-accounting.fixed.md | 8 +++++--- src/framework.ts | 4 ++++ 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/changelog.d/fix-67-end-turn-accounting.fixed.md b/changelog.d/fix-67-end-turn-accounting.fixed.md index f660cf7bd..a023508d7 100644 --- a/changelog.d/fix-67-end-turn-accounting.fixed.md +++ b/changelog.d/fix-67-end-turn-accounting.fixed.md @@ -13,9 +13,11 @@ (`stopReason: 'context_budget'` or `'physical_window'`), without `inference:completed`. A stream that errors, aborts or throws after finishing rounds counts them, and its failure entry in the inference log carries them. A turn abandoned by an operator quiesce, a stream superseded by a - newer one, or one whose agent was disposed, which wrote no entry before, now writes a failure - entry naming what ended it and carrying its rounds, and `inferenceCount` counts it, when it had - finished a round; one that hadn't has nothing to carry and writes none, as before. Exactly one + newer one, one whose agent was disposed, or one that ended with no final response and no other + end, which wrote no entry before, now writes a failure entry naming what ended it and carrying + its rounds, and `inferenceCount` counts it, when it had finished a round; one that hadn't has + nothing to carry and writes none, as before. If the store fails while a stream's end is + recorded, the error is reported on stderr and the stream's teardown still runs. Exactly one entry of each stream carries its usage, so totals summed from the inference log count each stream once, as the session's do. A health snapshot's token totals read the log's newest 1,000 entries, so with these new entries that window spans fewer turns. diff --git a/src/framework.ts b/src/framework.ts index c23a4879b..682dd3f3c 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -15132,6 +15132,10 @@ export class AgentFramework { return clone; } + /** Writes one inference-log entry. A stream's entries go through its + * driveStream's `logStream`, which gives the stream's usage to one entry + * only; an end that calls this directly for a stream writes a second + * entry for it, and its usage can be counted twice in the log. */ private logInference(entry: InferenceLogEntry): void { // Store large request/response as blobs const entryToStore = { ...entry }; From bb11216a094e3ed19b97e82407cfb334b6c1af24 Mon Sep 17 00:00:00 2001 From: Milo-2653 Date: Sat, 17 Oct 2026 14:12:59 -0400 Subject: [PATCH 13/13] fix(usage): a write that fails at a tool boundary no longer skips the boundary's work The boundary's ends call driveStream's accounting through streamBoundaryEnds: a tool-ended turn's completion, and a context-budget or physical-window restart's entry for the stream it cancels. That accounting writes the stream's inference-log entry, and logInference has no try/catch, so a store that refused the write threw out of the tool-result handler, into runLoop's catch. On a restart that skipped agent.cancelStream(), the inference:stream_restarted trace and the context_budget_restart request. The cancelled stream's abort keeps the event gate for its successor (preserveEventGateForSuccessor), and none was queued, so the agent stayed in `ready` with the gate held. On a tool-ended turn it skipped endRefusalEpisode, the inference:turn_ended trace and, on both, the tool-result event's dispatch to modules. Greptile found the restart case on the pull request; the turn's end is the same call one branch up. The closure now catches what its accounting throws and reports it on stderr with the message finally's accounting already uses, so what the boundary does after the call still runs. The provider-cooldown receipt and the refusal episode's end stay outside the try, in the order they had, since neither writes to the store. A failed entry is not written again later: logStream has already marked the stream's usage as logged, as with finally's failed write. Tests, in end-turn-usage: a restart whose entry's write throws still opens its next stream, announces the restart and ends the turn on it; a tool-ended turn whose completion entry's write throws still ends its refusal episode, emits inference:turn_ended and dispatches the tool's result event to modules. Both report the failure on stderr. The changelog's sentence on store errors now names the boundary's ends beside the teardown. --- .../fix-67-end-turn-accounting.fixed.md | 3 +- src/framework.ts | 35 +++++--- test/end-turn-usage.test.ts | 79 +++++++++++++++++++ 3 files changed, 103 insertions(+), 14 deletions(-) diff --git a/changelog.d/fix-67-end-turn-accounting.fixed.md b/changelog.d/fix-67-end-turn-accounting.fixed.md index a023508d7..dcf24db7e 100644 --- a/changelog.d/fix-67-end-turn-accounting.fixed.md +++ b/changelog.d/fix-67-end-turn-accounting.fixed.md @@ -17,7 +17,8 @@ end, which wrote no entry before, now writes a failure entry naming what ended it and carrying its rounds, and `inferenceCount` counts it, when it had finished a round; one that hadn't has nothing to carry and writes none, as before. If the store fails while a stream's end is - recorded, the error is reported on stderr and the stream's teardown still runs. Exactly one + recorded, the error is reported on stderr and what comes after it still runs: the rest of a + tool-ended turn's end, a restart's next stream, and the stream's teardown. Exactly one entry of each stream carries its usage, so totals summed from the inference log count each stream once, as the session's do. A health snapshot's token totals read the log's newest 1,000 entries, so with these new entries that window spans fewer turns. diff --git a/src/framework.ts b/src/framework.ts index 682dd3f3c..7c2bd23ee 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -12270,20 +12270,29 @@ export class AgentFramework { }; this.streamBoundaryEnds.set(stream, (end) => { const durationMs = Date.now() - startTime; - if (end === 'turn_ended') { - this.recordProviderAccelerationRecovery(agent, compiledRequest, end); - completeTurn(latestUsage, durationMs); - this.endRefusalEpisode(agent.name); - return; + const turnEnded = end === 'turn_ended'; + if (turnEnded) this.recordProviderAccelerationRecovery(agent, compiledRequest, end); + // The accounting writes to the store, and a write that fails must not + // skip what the boundary does after this call: the rest of a turn's + // end, or a restart's cancel and its request, and then the event's + // dispatch to modules. + try { + if (turnEnded) { + completeTurn(latestUsage, durationMs); + } else { + countUsage(latestUsage); + logStream({ + success: true, + request: BOUNDARY_END_REQUEST, + response: { note: 'stream restarted to compress its context' }, + durationMs, + stopReason: end, + }); + } + } catch (error) { + console.error(`[usage] agent=${agent.name} could not record a stream's usage at its end: ${describeFailureWithFrames(error)}`); } - countUsage(latestUsage); - logStream({ - success: true, - request: BOUNDARY_END_REQUEST, - response: { note: 'stream restarted to compress its context' }, - durationMs, - stopReason: end, - }); + if (turnEnded) this.endRefusalEpisode(agent.name); }); // Membrane usage events are cumulative across the native/XML tool loop. // Keep the previous cumulative sample so consumers that operate at the diff --git a/test/end-turn-usage.test.ts b/test/end-turn-usage.test.ts index 9fcbe8ea0..9efbb5c9e 100644 --- a/test/end-turn-usage.test.ts +++ b/test/end-turn-usage.test.ts @@ -969,3 +969,82 @@ describe('a stream that ends without writing its own entry is logged once, in fi } }); }); + +describe("a write that fails at a tool boundary doesn't skip what the boundary does after it", () => { + /** Makes the framework's inference-log write throw for the entries + * `refuse` picks, as a store that refuses the write would, and collects + * stderr so a test can see the failure reported. */ + function refuseWrites(framework: AgentFramework, refuse: (entry: { stopReason?: string }) => boolean) { + const internal = framework as unknown as { logInference: (entry: { stopReason?: string }) => void }; + const write = internal.logInference.bind(framework); + internal.logInference = (entry) => { + if (refuse(entry)) throw new Error('zz store refused the write'); + write(entry); + }; + const lines: string[] = []; + const original = console.error; + console.error = (...args: unknown[]) => { lines.push(args.map(String).join(' ')); }; + return { + reported: (agent: string) => lines.some((line) => + line.startsWith(`[usage] agent=${agent} could not record a stream's usage at its end`) + && line.includes('zz store refused the write')), + restore: () => { console.error = original; }, + }; + } + + it('a restart still opens its next stream', async () => { + const dir = mkdtempSync(join(tmpdir(), 'af-end-turn-usage-')); + const membrane = new SequentialStreamMembrane(); + membrane.pushResponse(toolResponse('echo', 'echo-1', firstUsage)); + membrane.pushResponse(finalResponse('Recovered.', secondUsage)); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store.chronicle'), membrane: membrane.asMembrane(), agents: [], + modules: [new TurnTools()], syncIntervalMs: 0, + }); + const traces: TraceEvent[] = []; + framework.onTrace((event) => traces.push(event)); + const stderr = refuseWrites(framework, (entry) => entry.stopReason === 'context_budget'); + framework.start(); + try { + const { agent, contextManager } = await framework.createEphemeralAgent({ + name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', + proseRouting: 'disabled', maxStreamTokens: 50, + }); + contextManager.addMessage('user', [{ type: 'text', text: 'Run once.' }]); + const run = framework.runEphemeralToCompletion(agent, contextManager); + // Awaited below once the restart is in. Without the restart it never + // settles, and stop() mustn't turn it into an unhandled rejection. + run.catch(() => {}); + assert.ok(await pollUntil(() => membrane.calls.length === 2), 'the restart opened its next stream'); + await run; + assert.equal(traces.filter((e) => e.type === 'inference:stream_restarted').length, 1, 'the restart is announced'); + assert.equal(traces.filter((e) => e.type === 'inference:completed').length, 1, 'the turn ends on its next stream'); + assert.ok(stderr.reported('worker'), 'the failed write is said'); + } finally { + stderr.restore(); + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }); + + it("a tool-ended turn still ends its refusal episode, says it ended, and hands the tool's result to the modules", async () => { + const f = await fixture(); + const stderr = refuseWrites(f.framework, (entry) => entry.stopReason === 'turn_ended'); + try { + const refusalStreak = (f.framework as unknown as { refusalStreak: Map }).refusalStreak; + refusalStreak.set('worker', 2); + f.membrane.pushResponse(toolResponse('finish', 'finish-1', totalUsage)); + await run(f.framework, 'worker'); + assert.equal(refusalStreak.has('worker'), false, 'the refusal episode ends'); + assert.equal(f.traces.filter((e) => e.type === 'inference:turn_ended').length, 1, 'inference:turn_ended'); + assert.ok(await pollUntil(() => f.traces.some((e) => e.type === 'process:completed' + && (e as { processEvent?: { type?: string } }).processEvent?.type === 'tool-result')), + "the tool result's event reaches the modules"); + assert.ok(stderr.reported('worker'), 'the failed write is said'); + } finally { + stderr.restore(); + await f.framework.stop(); + rmSync(f.dir, { recursive: true, force: true }); + } + }); +});