Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions ENVIRONMENT.md
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,8 @@ Manage shared web env var additions and rotations with `pnpm web:env set <VARIAB
- `STYTCH_PROJECT_SECRET` - Stytch project secret. `[SECRET]`
- `STYTCH_PUBLIC_TOKEN` - Stytch legacy public token alias used in some test fixtures. [PUBLIC]
- `INTERNAL_API_SECRET` - Shared secret for internal API calls between services; used in `apps/web/src/lib/kiloclaw/cli-runs.test.ts`, `kiloclaw-router.test.ts`, dev seed scripts, and other service routers. `[SECRET]`
- `USAGE_INGEST_URL` - Optional complete usage-ingest HTTP endpoint (for example, ending in `/usage`) for the unused gateway usage publisher helper. No default; missing URL or publisher secret disables the helper. It makes one attempt with a two-second timeout and accepts only HTTP 202. For isolated local tests, use `http://127.0.0.1:<fixture-port>/usage` with the real usage-ingest Worker and a local queue. Configuring it does not activate gateway publication. [SERVER]
- `USAGE_INGEST_PUBLISH_SECRET` - Optional dedicated Bearer secret for the unused gateway usage publisher helper; must match the usage-ingest Worker's setting of the same name. Separate from `INTERNAL_API_SECRET` and user credentials. Use only a synthetic matching value in local fixtures. `[SECRET]`
- `SUPPORT_API_SECRET` - Shared bearer token for Customer Support Automation (CSA) internal API calls. Cloud uses it to authorize CSA → Cloud `apps/web/src/app/api/internal/support/` and Cloud → CSA `POST /api/internal/cloud/users/gdpr-scrub`. A CSA compromise can also call Cloud deletion and Cloud can scrub CSA-local PII. Leak can look up any email and enqueue deletion for non-admin, non-bot, non-live-subscription customers; access disable is deferred to worker preflight and pending requests can be cancelled. Keep production values off preview deployments; rotate Cloud and CSA together. `[SECRET]`
- `BOUNDED_INTERNAL_SERVICE_TOKENS_ENABLED` - Set to exact `true` to enable modern, purpose-labelled internal assertions at the Phase 5.1 bounded Git broker, export, deletion, and Session Ingest callsites. Unset or any other value retains their existing legacy token formats. Enable only after compatible readers, including the dedicated GitHub disconnect audience, are deployed; generic human/control/runtime signers are not affected. [SERVER]
- `NATIVE_RESOURCE_TOKENS_ENABLED` - Set to exact `true` to permit fresh native adoption of separate one-hour API/gateway access tokens only for clients explicitly requesting `api-gateway-v1`, provided `SHARED_RESOURCE_TOKENS_ENABLED` is also exact `true`. Default-off. Unsupported clients keep legacy responses. Turning this flag off affects subsequent native issuance/refreshes only; active modern device credentials can continue receiving bounded control tokens while their owned device session and current user pepper remain valid. [SERVER]
Expand Down
228 changes: 228 additions & 0 deletions packages/web-shared/src/lib/ai-gateway/usage-publisher.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,228 @@
jest.mock('@kilocode/web-shared/lib/config.server', () => ({
USAGE_INGEST_URL: 'https://usage.example.test/usage',
USAGE_INGEST_PUBLISH_SECRET: 'synthetic-publisher-secret',
}));

import { afterEach, beforeEach, describe, expect, test } from '@jest/globals';
import { createServer } from 'node:http';
import { enqueueUsage, USAGE_ENQUEUE_TIMEOUT_MS } from './usage-publisher';
import type { UsageRecordRequest } from '@kilocode/usage-contracts';

const payload: UsageRecordRequest = {
core: {
id: '3f2504e0-4f89-11d3-9a0c-0305e82c3301',
kilo_user_id: 'oauth/synthetic',
cost: 1234,
input_tokens: 10,
output_tokens: 20,
cache_write_tokens: 0,
cache_hit_tokens: 0,
created_at: '2026-08-05T10:11:12.945Z',
provider: 'openrouter',
model: 'synthetic-model',
requested_model: 'synthetic-model',
cache_discount: null,
has_error: false,
abuse_classification: 0,
organization_id: null,
inference_provider: null,
project_id: null,
},
metadata: {
id: '3f2504e0-4f89-11d3-9a0c-0305e82c3301',
message_id: 'msg-1',
created_at: '2026-08-05T10:11:12.945Z',
http_x_forwarded_for: null,
http_x_vercel_ip_city: null,
http_x_vercel_ip_country: null,
http_x_vercel_ip_latitude: null,
http_x_vercel_ip_longitude: null,
http_x_vercel_ja4_digest: null,
user_prompt_prefix: '🙂',
system_prompt_prefix: null,
system_prompt_length: null,
http_user_agent: null,
max_tokens: null,
has_middle_out_transform: null,
status_code: 200,
upstream_id: null,
finish_reason: 'stop',
latency: null,
moderation_latency: null,
generation_time: null,
is_byok: null,
is_user_byok: false,
streamed: null,
cancelled: null,
editor_name: null,
api_kind: 'chat_completions',
has_tools: null,
machine_id: null,
feature: null,
session_id: null,
mode: null,
auto_model: null,
reasoning_setting: 'medium',
market_cost: null,
is_free: null,
abuse_delay: null,
abuse_downgraded_from: null,
},
prior_microdollar_usage: 0,
posthog_distinct_id: null,
};

const config = jest.requireMock<{
USAGE_INGEST_URL: string;
USAGE_INGEST_PUBLISH_SECRET: string;
}>('@kilocode/web-shared/lib/config.server');
const mockFetch = jest.fn<ReturnType<typeof fetch>, Parameters<typeof fetch>>();
const nativeFetch = globalThis.fetch;
let warn: jest.SpiedFunction<typeof console.warn>;

beforeEach(() => {
config.USAGE_INGEST_URL = 'https://usage.example.test/usage';
config.USAGE_INGEST_PUBLISH_SECRET = 'synthetic-publisher-secret';
mockFetch.mockReset();
jest.spyOn(globalThis, 'fetch').mockImplementation(mockFetch);
warn = jest.spyOn(console, 'warn').mockImplementation(() => {});
});

afterEach(() => {
jest.restoreAllMocks();
});

function expectUnavailableWarning(reason: string) {
expect(warn.mock.calls).toEqual([
['usage enqueue unavailable', { usageId: payload.core.id, kind: 'unavailable', reason }],
]);
const logged = JSON.stringify(warn.mock.calls);
expect(logged).not.toContain(config.USAGE_INGEST_URL);
expect(logged).not.toContain(config.USAGE_INGEST_PUBLISH_SECRET);
expect(logged).not.toMatch(/Bearer|sensitive|prompt|metadata/);
}

describe('enqueueUsage', () => {
test('preserves the complete payload and ID with dedicated publisher authentication', async () => {
const event = {
...payload,
bouncer_usage_event: { request_id: 'synthetic-request', payload: { cost: 1234 } },
} satisfies UsageRecordRequest;
const response = new Response(null, { status: 202 });
const read = jest.spyOn(response, 'text');
mockFetch.mockResolvedValue(response);
const timeout = jest.spyOn(AbortSignal, 'timeout');

expect(await enqueueUsage(event)).toEqual({ kind: 'accepted' });
expect(mockFetch).toHaveBeenCalledTimes(1);
expect(mockFetch).toHaveBeenCalledWith('https://usage.example.test/usage', {
method: 'POST',
headers: {
'content-type': 'application/json',
Authorization: 'Bearer synthetic-publisher-secret',
},
body: JSON.stringify(event),
signal: expect.any(AbortSignal),
cache: 'no-store',
redirect: 'error',
});
expect(timeout).toHaveBeenCalledWith(USAGE_ENQUEUE_TIMEOUT_MS);
expect(read).not.toHaveBeenCalled();
expect(warn).not.toHaveBeenCalled();
});

test.each(['USAGE_INGEST_URL', 'USAGE_INGEST_PUBLISH_SECRET'] as const)(
'is disabled without %s and makes no network call',
async key => {
config[key] = '';
expect(await enqueueUsage(payload)).toEqual({ kind: 'disabled' });
expect(mockFetch).not.toHaveBeenCalled();
expect(warn).not.toHaveBeenCalled();
}
);

test('contains a malformed configured URL through the request failure path', async () => {
config.USAGE_INGEST_URL = 'malformed-sensitive-url';
expect(await enqueueUsage(payload)).toEqual({ kind: 'unavailable', reason: 'request_failed' });
expect(mockFetch).not.toHaveBeenCalled();
expectUnavailableWarning('request_failed');
});

test('uses the configured URL without an additional path, query or fragment policy', async () => {
config.USAGE_INGEST_URL = 'https://usage.example.test/proxy/usage?fixture=synthetic#fragment';
mockFetch.mockResolvedValue(new Response(null, { status: 202 }));
expect(await enqueueUsage(payload)).toEqual({ kind: 'accepted' });
expect(mockFetch).toHaveBeenCalledWith(config.USAGE_INGEST_URL, expect.any(Object));
});

test('preserves confirmed acceptance when unused response-body cleanup fails', async () => {
const cancel = jest.fn(() => {
throw new Error('sensitive cleanup error');
});
const body = new ReadableStream({ cancel });
mockFetch.mockResolvedValue(new Response(body, { status: 202 }));

expect(await enqueueUsage(payload)).toEqual({ kind: 'accepted' });
// Allow a rejected cancellation promise to surface as an unhandled rejection.
await new Promise<void>(resolve => setImmediate(resolve));
expect(cancel).toHaveBeenCalledTimes(1);
expect(mockFetch).toHaveBeenCalledTimes(1);
expect(warn).not.toHaveBeenCalled();
});

test.each([200, 204, 401, 503])(
'reports HTTP %i without retrying or reading its body',
async status => {
const response = new Response(status === 204 ? null : 'sensitive-upstream-body', { status });
const read = jest.spyOn(response, 'text');
const cancel = response.body ? jest.spyOn(response.body, 'cancel') : null;
mockFetch.mockResolvedValue(response);
const outcome = { kind: 'unavailable', reason: `http_${status}` };
expect(await enqueueUsage(payload)).toEqual(outcome);
expect(mockFetch).toHaveBeenCalledTimes(1);
expect(read).not.toHaveBeenCalled();
if (cancel) expect(cancel).toHaveBeenCalledTimes(1);
expectUnavailableWarning(outcome.reason);
}
);

test('bounds a real HTTP request that never responds to two seconds', async () => {
let requests = 0;
const server = createServer(() => {
requests++;
});
try {
await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve));
const address = server.address();
if (!address || typeof address === 'string') throw new Error('Missing fixture port');
config.USAGE_INGEST_URL = `http://127.0.0.1:${address.port}/usage`;
mockFetch.mockImplementation(nativeFetch);
const start = performance.now();
expect(await enqueueUsage(payload)).toEqual({ kind: 'unavailable', reason: 'timeout' });
const elapsed = performance.now() - start;
expect(elapsed).toBeGreaterThanOrEqual(USAGE_ENQUEUE_TIMEOUT_MS - 100);
expect(elapsed).toBeLessThan(USAGE_ENQUEUE_TIMEOUT_MS + 1_000);
expect(requests).toBe(1);
expect(mockFetch).toHaveBeenCalledTimes(1);
expectUnavailableWarning('timeout');
} finally {
server.closeAllConnections();
await new Promise<void>((resolve, reject) =>
server.close(error => (error ? reject(error) : resolve()))
);
}
});

test('contains network failures and emits only the ID and categorical outcome', async () => {
const error = Object.assign(new Error('synthetic-publisher-secret sensitive prompt'), {
name: 'sensitive-error-name',
});
mockFetch.mockRejectedValue(error);
expect(await enqueueUsage(payload)).toEqual({
kind: 'unavailable',
reason: 'request_failed',
});
expect(mockFetch).toHaveBeenCalledTimes(1);
expectUnavailableWarning('request_failed');
});
});
46 changes: 46 additions & 0 deletions packages/web-shared/src/lib/ai-gateway/usage-publisher.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
import 'server-only';

import {
USAGE_INGEST_URL,
USAGE_INGEST_PUBLISH_SECRET,
} from '@kilocode/web-shared/lib/config.server';
import type { UsageRecordRequest } from '@kilocode/usage-contracts';

export const USAGE_ENQUEUE_TIMEOUT_MS = 2_000;

export type UsageEnqueueOutcome =
| { kind: 'disabled' }
| { kind: 'accepted' }
| { kind: 'unavailable'; reason: string };

/** Enqueues usage through the authenticated endpoint; currently has no gateway callers. */
export async function enqueueUsage(payload: UsageRecordRequest): Promise<UsageEnqueueOutcome> {
if (!USAGE_INGEST_URL || !USAGE_INGEST_PUBLISH_SECRET) return { kind: 'disabled' };

const signal = AbortSignal.timeout(USAGE_ENQUEUE_TIMEOUT_MS);
let outcome: UsageEnqueueOutcome;
try {
const url = new URL(USAGE_INGEST_URL).toString();
const response = await fetch(url, {
method: 'POST',
headers: {
'content-type': 'application/json',
Authorization: `Bearer ${USAGE_INGEST_PUBLISH_SECRET}`,
},
body: JSON.stringify(payload),
signal,
cache: 'no-store',
redirect: 'error',
});
// Only the status is needed. Cancel the unused body without letting cleanup failures
// change a confirmed acceptance or become unhandled rejections.
void response.body?.cancel().catch(() => {});
if (response.status === 202) return { kind: 'accepted' };
outcome = { kind: 'unavailable', reason: `http_${response.status}` };
} catch {
outcome = { kind: 'unavailable', reason: signal.aborted ? 'timeout' : 'request_failed' };
}

console.warn('usage enqueue unavailable', { usageId: payload.core.id, ...outcome });
return outcome;
}
2 changes: 2 additions & 0 deletions packages/web-shared/src/lib/config.server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,8 @@ export const MISTRAL_API_KEY = getEnvVariable('MISTRAL_API_KEY');
export const INCEPTION_API_KEY = getEnvVariable('INCEPTION_API_KEY');
export const EXA_API_KEY = getEnvVariable('EXA_API_KEY');
export const INTERNAL_API_SECRET = getEnvVariable('INTERNAL_API_SECRET');
export const USAGE_INGEST_URL = getEnvVariable('USAGE_INGEST_URL');
export const USAGE_INGEST_PUBLISH_SECRET = getEnvVariable('USAGE_INGEST_PUBLISH_SECRET');
// Shared secret with the MCP catalog CI job
// (.github/workflows/kilo-mcp-catalog.yml). It authenticates only the mint in
// app/api/internal/mcp-catalog/token; it is never accepted as a Kilo credential.
Expand Down
Loading