Skip to content
32 changes: 30 additions & 2 deletions src/adapters/kiro.ts
Original file line number Diff line number Diff line change
Expand Up @@ -465,6 +465,7 @@ interface KiroAttemptResult {
interface KiroFallbackAttempt {
response: Response;
inputTokens: number;
contextInputEstimate: number;
nameMap: Map<string, string>;
conversationId: string;
}
Expand All @@ -491,6 +492,9 @@ function mergeKiroUsage(first: OcxUsage | undefined, second: OcxUsage | undefine
return {
inputTokens: first.inputTokens + second.inputTokens,
outputTokens: first.outputTokens + second.outputTokens,
...(typeof first.contextTotalTokens === "number" || typeof second.contextTotalTokens === "number"
? { contextTotalTokens: Math.max(first.contextTotalTokens ?? 0, second.contextTotalTokens ?? 0) }
Comment thread
coseung2 marked this conversation as resolved.
Outdated
Comment thread
coseung2 marked this conversation as resolved.
Outdated
: {}),
Comment thread
coseung2 marked this conversation as resolved.
...(totalTokens !== undefined ? { totalTokens } : {}),
...(sumOptional("cachedInputTokens") !== undefined ? { cachedInputTokens: sumOptional("cachedInputTokens") } : {}),
...(sumOptional("cacheReadInputTokens") !== undefined ? { cacheReadInputTokens: sumOptional("cacheReadInputTokens") } : {}),
Expand Down Expand Up @@ -534,6 +538,7 @@ async function* parseKiroAttempt(
nameMap: Map<string, string> | undefined,
conversationId: string | undefined,
previousAssistantText?: string,
contextInputEstimate?: number,
): AsyncGenerator<AdapterEvent, KiroAttemptResult> {
const emptyResult = (): KiroAttemptResult => ({ assistantText: "", sawReasoning: false });
if (!response.body) {
Expand All @@ -560,11 +565,23 @@ async function* parseKiroAttempt(
const providerState = (): { kiro: { conversationId: string } } | undefined =>
returnedConversationId ? { kiro: { conversationId: returnedConversationId } } : undefined;

const usage = (): OcxUsage => authoritativeUsage ?? ({
const contextUsageTotalFloor = (): number | undefined => {
if (contextUsagePercentage === undefined || !contextWindow) return undefined;
const floor = Math.ceil(contextWindow * Math.min(contextUsagePercentage, 100) / 100);
Comment thread
coseung2 marked this conversation as resolved.
Outdated
Comment thread
coseung2 marked this conversation as resolved.
Outdated
return Number.isFinite(floor) && floor > 0 ? floor : undefined;
};
const usage = (): OcxUsage => {
const base = authoritativeUsage ?? {
inputTokens,
outputTokens: estimateTokens(outputChars, modelId),
estimated: true,
});
};
const estimatedContextTotal = contextInputEstimate !== undefined
? contextInputEstimate + base.outputTokens
: undefined;
const contextTotal = Math.max(estimatedContextTotal ?? 0, contextUsageTotalFloor() ?? 0);
Comment thread
coseung2 marked this conversation as resolved.
Outdated
return contextTotal > 0 ? { ...base, contextTotalTokens: contextTotal } : base;
};

const classifiedTerminal = (failure: KiroErrorClassification): AdapterEvent => ({
type: "error",
Expand Down Expand Up @@ -970,6 +987,7 @@ export async function* parseKiroStream(
conversationId?: string,
completionMode: KiroCompletionMode = "disabled",
fallbackFactory?: KiroFallbackFactory,
contextInputEstimate?: number,
): AsyncGenerator<AdapterEvent> {
const first = parseKiroAttempt(
response,
Expand All @@ -979,6 +997,8 @@ export async function* parseKiroStream(
contextWindow,
nameMap,
conversationId,
undefined,
contextInputEstimate,
);
let firstNext = await first.next();
while (!firstNext.done) {
Expand Down Expand Up @@ -1043,6 +1063,7 @@ export async function* parseKiroStream(
fallback.nameMap,
fallback.conversationId,
firstResult.assistantText,
fallback.contextInputEstimate,
);
let secondNext = await second.next();
while (!secondNext.done) {
Expand Down Expand Up @@ -1080,6 +1101,7 @@ export function createKiroAdapter(provider: OcxProviderConfig): ProviderAdapter
// Per-request closure (resolveAdapter builds a fresh adapter per request — server.ts:440 — so this
// is race-free) carrying the heuristic input-token estimate from buildRequest into the stream.
let inputTokens = 0;
let contextInputEstimate = 0;
let modelId: string | undefined;
let contextWindow: number | undefined;
let toolNameMap: Map<string, string> | undefined;
Expand All @@ -1097,6 +1119,7 @@ export function createKiroAdapter(provider: OcxProviderConfig): ProviderAdapter
conversationId: string;
completionMode: KiroCompletionMode;
inputTokens: number;
contextInputEstimate: number;
}> => {
if (typeof provider.apiKey !== "string" || provider.apiKey.trim() === "") {
throw new Error("kiro token missing — run ocx login kiro");
Expand Down Expand Up @@ -1141,6 +1164,7 @@ export function createKiroAdapter(provider: OcxProviderConfig): ProviderAdapter
conversationId: built.conversationId,
completionMode: built.completionMode,
inputTokens: estimateKiroInputTokens(parsed),
contextInputEstimate: estimateKiroLogInputTokens(parsed),
Comment thread
coseung2 marked this conversation as resolved.
Outdated
};
};

Expand Down Expand Up @@ -1179,6 +1203,7 @@ export function createKiroAdapter(provider: OcxProviderConfig): ProviderAdapter
return {
response,
inputTokens: retry.inputTokens,
contextInputEstimate: retry.contextInputEstimate,
nameMap: retry.nameMap,
conversationId: retry.conversationId,
};
Expand All @@ -1191,6 +1216,7 @@ export function createKiroAdapter(provider: OcxProviderConfig): ProviderAdapter
modelId = parsed.modelId;
contextWindow = configuredKiroContextWindow(provider, parsed.modelId);
inputTokens = built.inputTokens;
contextInputEstimate = built.contextInputEstimate;
toolNameMap = built.nameMap;
conversationId = built.conversationId;
completionMode = built.completionMode;
Expand All @@ -1209,6 +1235,7 @@ export function createKiroAdapter(provider: OcxProviderConfig): ProviderAdapter
conversationId,
completionMode,
completionMode === "required" ? fallbackFactory : undefined,
contextInputEstimate,
);
},

Expand Down Expand Up @@ -1239,6 +1266,7 @@ export function createKiroAdapter(provider: OcxProviderConfig): ProviderAdapter
conversationId,
completionMode,
completionMode === "required" ? fallbackFactory : undefined,
contextInputEstimate,
)) events.push(e);
return events;
},
Expand Down
11 changes: 8 additions & 3 deletions src/bridge.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,12 +15,17 @@ function sseEvent(name: string, data: Record<string, unknown>): string {

function responsesUsage(usage: OcxUsage | undefined): Record<string, unknown> {
if (!usage) return { input_tokens: 0, output_tokens: 0, total_tokens: 0 };
// inputTokens is already inclusive of cache read/write (types.ts convention).
const inputTokens = usage.inputTokens;
// Stateful providers may report an absolute active-context checkpoint separately from their
// per-attempt usage. Split that checkpoint into input + output without adding output twice.
const inputTokens = usage.contextTotalTokens !== undefined
? Math.max(0, usage.contextTotalTokens - usage.outputTokens)
: usage.inputTokens;
Comment thread
coseung2 marked this conversation as resolved.
const out: Record<string, unknown> = {
input_tokens: inputTokens,
Comment thread
coseung2 marked this conversation as resolved.
output_tokens: usage.outputTokens,
total_tokens: usageDisplayTotalTokens(usage) ?? inputTokens + usage.outputTokens,
total_tokens: usage.contextTotalTokens !== undefined
? usage.contextTotalTokens
: usageDisplayTotalTokens(usage) ?? inputTokens + usage.outputTokens,
};
const inputDetails: Record<string, number> = {};
if (usage.cachedInputTokens !== undefined) {
Expand Down
6 changes: 6 additions & 0 deletions src/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -308,6 +308,12 @@ export interface OcxUrlCitation {
export interface OcxUsage {
inputTokens: number;
outputTokens: number;
/**
* Absolute active-context size after the response. Stateful providers can expose this separately
* from their per-attempt usage. Responses serialization derives the input side from
* `contextTotalTokens - outputTokens` so output is never added to an absolute checkpoint twice.
*/
contextTotalTokens?: number;
totalTokens?: number;
cachedInputTokens?: number;
cacheReadInputTokens?: number;
Expand Down
37 changes: 37 additions & 0 deletions tests/bridge.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,43 @@ describe("Responses bridge reasoning and usage parity", () => {
});
});

test("absolute context total drives Responses compaction without double-counting output", async () => {
const frames = await collectSse(bridgeToResponsesSSE(replay([
{
type: "done",
usage: {
inputTokens: 58,
contextTotalTokens: 226_000,
outputTokens: 12,
estimated: true,
},
},
]), "kiro/claude-opus-5"));

const completed = frames.find(f => f.event === "response.completed")?.data.response as Record<string, unknown>;
expect(completed.usage).toEqual({
input_tokens: 225_988,
output_tokens: 12,
total_tokens: 226_000,
});
});

test("consecutive context checkpoints remain absolute instead of accumulating in the bridge", async () => {
const totals: number[] = [];
for (const [contextTotalTokens, outputTokens] of [[10_000, 42], [10_300, 20]] as const) {
const frames = await collectSse(bridgeToResponsesSSE(replay([{
type: "done",
usage: { inputTokens: 1, contextTotalTokens, outputTokens, estimated: true },
}]), "kiro/claude-opus-5"));
const completed = frames.find(f => f.event === "response.completed")?.data.response as Record<string, unknown>;
const usage = completed.usage as Record<string, number>;
expect(usage.input_tokens).toBe(contextTotalTokens - outputTokens);
expect(usage.total_tokens).toBe(contextTotalTokens);
totals.push(usage.total_tokens);
}
expect(totals).toEqual([10_000, 10_300]);
});

test("Anthropic cache read and write tokens pass through Responses usage without re-adding", async () => {
const frames = await collectSse(bridgeToResponsesSSE(replay([
{
Expand Down
30 changes: 29 additions & 1 deletion tests/kiro-stream.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -312,6 +312,29 @@ describe("kiro adapter — parseStream", () => {
});
});

test("bounded fallback uses its rebuilt context estimate for the final absolute checkpoint", async () => {
const firstText = "p".repeat(7000);
const finalText = "f".repeat(3500);
globalThis.fetch = (async () => new Response(streamOf(eventFrame({ content: finalText })))) as typeof fetch;
const adapter = createKiroAdapter(provider);
const request = await adapter.buildRequest(parsedWith([{ role: "user", content: "do it" }], [bashTool]));
const initialContextEstimate = request.usageLog?.inputTokens ?? 0;

const events = await collectAdapterEvents(adapter.parseStream(new Response(streamOf(
eventFrame({ content: firstText }),
))));
const done = events.at(-1);
expect(done?.type).toBe("done");
const usage = done?.type === "done" ? done.usage : undefined;
expect(usage?.outputTokens).toBe(estimateTokens(firstText, "claude-sonnet-4.5") + estimateTokens(finalText, "claude-sonnet-4.5"));
expect(usage?.contextTotalTokens).toBeGreaterThan(
initialContextEstimate + Math.max(
estimateTokens(firstText, "claude-sonnet-4.5"),
estimateTokens(finalText, "claude-sonnet-4.5"),
),
);
});

test("keeps a private-completion fallback after reasoning-only output as the final answer", async () => {
globalThis.fetch = (async () => new Response(streamOf(...completionFrames("Done.")))) as typeof fetch;
const adapter = createKiroAdapter(provider);
Expand Down Expand Up @@ -777,6 +800,7 @@ describe("kiro adapter — parseStream", () => {
);
expect(done).toEqual({
inputTokens: 15,
contextTotalTokens: 204,
cachedInputTokens: 3,
cacheReadInputTokens: 3,
cacheCreationInputTokens: 2,
Expand Down Expand Up @@ -823,7 +847,7 @@ describe("kiro adapter — parseStream", () => {
expect((events[0] as { message: string }).message).toContain("Compact or reduce the history");
});

test("Kiro contextUsagePercentage remains diagnostic and does not override totals", async () => {
test("Kiro contextUsagePercentage drives context pressure without overriding turn totals", async () => {
const adapter = createKiroAdapter(provider);
await adapter.buildRequest(parsedWith([{ role: "user", content: "x".repeat(700) }]));
const done = await doneUsage(
Expand All @@ -836,6 +860,7 @@ describe("kiro adapter — parseStream", () => {
expect(done.outputTokens).toBe(100);
expect(done.totalTokens).toBeUndefined();
expect(done.estimated).toBe(true);
expect(done.contextTotalTokens).toBe(50_000);
});

test("Kiro auto ignores provider-level context window and falls back to heuristic totals", async () => {
Expand All @@ -850,6 +875,7 @@ describe("kiro adapter — parseStream", () => {
expect(done.inputTokens).toBe(200);
expect(done.outputTokens).toBe(100);
expect(done.totalTokens).toBeUndefined();
expect(done.contextTotalTokens).toBe(300);
});

test("fresh payload includes history while usage counts only the current turn", async () => {
Expand All @@ -875,6 +901,7 @@ describe("kiro adapter — parseStream", () => {
expect(longBody.length).toBeGreaterThan(shortBody.length + 10_000);
expect(longUsage.inputTokens).toBe(shortUsage.inputTokens);
expect(longUsage.inputTokens).toBe(estimateTokens(latest, "claude-sonnet-4.5"));
expect(longUsage.contextTotalTokens).toBeGreaterThan(shortUsage.contextTotalTokens ?? 0);
});

test("request log usage estimates the full Codex context while SSE usage stays current-turn", async () => {
Expand All @@ -891,6 +918,7 @@ describe("kiro adapter — parseStream", () => {
expect(usage.inputTokens).toBe(estimateTokens(latest, "claude-sonnet-4.5"));
expect(request.usageLog?.estimated).toBe(true);
expect(request.usageLog?.inputTokens).toBeGreaterThan(usage.inputTokens + 4000);
expect(usage.contextTotalTokens).toBe((request.usageLog?.inputTokens ?? 0) + usage.outputTokens);
});

test("resumed payload preserves the complete locally expanded history", async () => {
Expand Down
36 changes: 36 additions & 0 deletions tests/request-log.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,14 @@ import {
sealRequestAttemptIdentity,
type RequestLogContext,
} from "../src/server/request-log";
import { bridgeToResponsesSSE } from "../src/bridge";
import type { AdapterEvent } from "../src/types";
import type { PersistedUsageEntry } from "../src/usage/log";

async function* replayAdapterEvents(events: AdapterEvent[]): AsyncGenerator<AdapterEvent> {
for (const event of events) yield event;
}

function log(overrides: Partial<RequestLogEntry>): RequestLogEntry {
return {
requestId: "ocx-test",
Expand Down Expand Up @@ -746,6 +752,36 @@ describe("request log metadata", () => {
});
});

test("deferred logging preserves a bridged Kiro absolute context checkpoint", async () => {
const entries: RequestLogEntry[] = [];
const body = bridgeToResponsesSSE(replayAdapterEvents([{
type: "done",
usage: {
inputTokens: 58,
outputTokens: 100,
contextTotalTokens: 50_000,
estimated: true,
},
}]), "kiro/claude-opus-5");
const response = responseWithDeferredRequestLog(
new Response(body, { status: 200, headers: { "content-type": "text/event-stream" } }),
"ocx-test-kiro-context-checkpoint",
Date.now(),
{ model: "kiro/claude-opus-5", provider: "kiro-p9d8524", usageLogInputTokens: 200 },
entry => entries.push(entry),
);

const text = await response.text();
expect(text).toContain('"input_tokens":49900');
expect(text).toContain('"total_tokens":50000');
expect(entries).toHaveLength(1);
expect(entries[0]).toMatchObject({
usageStatus: "estimated",
totalTokens: 50_000,
usage: { inputTokens: 49_900, outputTokens: 100, totalTokens: 50_000, estimated: true },
});
Comment thread
coseung2 marked this conversation as resolved.
});

test("final logging shows numeric Kiro estimates even when SSE usage is absent", async () => {
const entries: RequestLogEntry[] = [];
const response = responseWithDeferredRequestLog(
Expand Down
Loading