Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
40 kB · 1252 lines
TypeScript
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253import { Buffer } from "node:buffer";import { createHash, randomUUID } from "node:crypto";import { type Dirent, existsSync, mkdirSync, renameSync, writeFileSync } from "node:fs";import { mkdir, readdir, readFile, rename, stat, unlink, writeFile } from "node:fs/promises";import { dirname, isAbsolute, join, resolve } from "node:path";import { appendRotatingLog, getAgentDir, getAgentTracesLogPath, getSessionsDir, VERSION } from "../config.js";import { readFirstLineSync } from "../utils/file-lines.js";import type { AuthStorage } from "./auth-storage.js";import { PRIME_AGENT_TRACES_PROVIDER_ID, PRIME_INFERENCE_PROVIDER_ID, resolvePrimeAgentTracesBaseUrl,} from "./prime-inference-auth.js";import { getSessionArtifactsRoot, type SessionHeader, type SessionManager } from "./session-manager.js";import type { SettingsManager } from "./settings-manager.js";
const MAX_TRACE_BYTES = 20 * 1024 * 1024;const DEFAULT_REQUEST_TIMEOUT_MS = 15_000;const TRACE_UPLOAD_DEBOUNCE_MS = 1_000;const TRACE_UPLOAD_MIN_INTERVAL_MS = 60_000;const TRACE_UPLOAD_RETRY_BASE_DELAY_MS = 500;const TRACE_UPLOAD_RETRY_MAX_DELAY_MS = 10_000;const TRACE_UPLOAD_MAX_RETRIES = 3;const TRACE_UPLOAD_RETRY_JITTER = 0.2;const TRACE_PREVIEW_MAX_CHARS = 8_000;const TRACE_UPLOAD_ALL_CONCURRENCY = 4;const TRACE_UPLOAD_RATE_LIMIT_REQUESTS = 5;const TRACE_UPLOAD_RATE_LIMIT_WINDOW_MS = 60_000;const TRACE_UPLOAD_RATE_LIMIT_SAFETY_MS = 100;const MAX_TIMER_DELAY_MS = 2 ** 31 - 1;const TRACE_UPLOAD_ALL_MIN_REQUEST_INTERVAL_MS = Math.ceil(TRACE_UPLOAD_RATE_LIMIT_WINDOW_MS / TRACE_UPLOAD_RATE_LIMIT_REQUESTS) + TRACE_UPLOAD_RATE_LIMIT_SAFETY_MS;
export type AgentTraceCredentialSource = "environment" | "stored" | "prime-inference" | "prime-cli";
export interface AgentTraceCredential { apiKey: string; source: AgentTraceCredentialSource; label: string;}
export type AgentTraceUploadResult = | { status: "uploaded"; sessionId: string; traceId: string; bytesStored: number; key?: string; } | { status: "disabled" } | { status: "unchanged" } | { status: "missing_credentials" } | { status: "no_session_file" } | { status: "empty_session" } | { status: "invalid_session"; message: string } | { status: "too_large"; size: number; maxBytes: number } | { status: "failed"; statusCode?: number; message: string; retryAfterMs?: number };
export interface AgentTraceUploadDelay { /** Retry backoff after a failed attempt, or the batch gate holding the platform rate limit. */ reason: "retry-backoff" | "rate-limit"; delayMs: number;}
export interface AgentTraceUploadOptions { sessionFile: string | undefined; authStorage: AuthStorage; settingsManager: SettingsManager; /** Require the global automatic-sharing opt-in. Set false only for an explicit one-shot upload command. */ requireEnabled?: boolean; baseUrl?: string; fetchFn?: typeof fetch; reloadConfig?: boolean; requestTimeoutMs?: number; signal?: AbortSignal; /** Reports every wait the upload arms before its next request. */ onUploadDelay?: (delay: AgentTraceUploadDelay) => void;}
export interface AgentTraceSessionUploadOptions extends Omit<AgentTraceUploadOptions, "sessionFile"> { sessionManager: SessionManager;}
export interface AgentTraceUploadSchedule { delayMs: number;}
export interface AgentTraceUploadInstallOptions { authStorage: AuthStorage; settingsManager: SettingsManager; baseUrl?: string; fetchFn?: typeof fetch; requestTimeoutMs?: number; /** The session's semantic-edge ledger; registered with the outbox as its own delivery kind. */ semanticEdgesLedgerPath?: string; /** Reports the timer the scheduler arms for its next automatic upload. */ onUploadScheduled?: (schedule: AgentTraceUploadSchedule) => void; /** Reports each scheduled cycle once it settles, after any follow-up cycle is armed. */ onUploadSettled?: (outcome: AgentTraceUploadCycleOutcome) => void;}
/** A scheduled cycle either finishes an upload or folds into the upload already in flight. */export type AgentTraceUploadCycleOutcome = AgentTraceUploadResult | { status: "coalesced" };
export interface AgentTraceUploadInstallation { /** Resolves once the startup catch-up and any in-flight automatic upload have finished writing. */ whenIdle: () => Promise<void>;}
export type AgentTracePreviewResult = | { status: "ready"; sessionFile: string; sessionId: string; traceId: string; parentSessionId?: string; cwd: string; size: number; maxBytes: number; uploadable: boolean; endpoint: string; gitRepo?: string; gitCommit?: string; contentPreview: string; truncated: boolean; } | { status: "no_session_file" } | { status: "empty_session" } | { status: "invalid_session"; message: string } | { status: "failed"; message: string };
export interface AgentTracePreviewOptions { sessionFile: string | undefined; baseUrl?: string; maxContentChars?: number;}
export interface AgentTraceUploadAllProgress { completed: number; total: number; sessionFile?: string; result?: AgentTraceUploadResult;}
export interface AgentTraceUploadAllOptions extends Omit<AgentTraceUploadOptions, "sessionFile"> { sessionDir?: string; concurrency?: number; onProgress?: (progress: AgentTraceUploadAllProgress) => void;}
export interface AgentTraceUploadAllResult { total: number; uploaded: number; failed: number; skipped: number; bytesStored: number; results: Array<{ sessionFile: string; result: AgentTraceUploadResult }>;}
function stringEnv(name: string): string | undefined { const value = process.env[name]; return typeof value === "string" && value.trim() ? value.trim() : undefined;}
function isRecord(value: unknown): value is Record<string, unknown> { return typeof value === "object" && value !== null && !Array.isArray(value);}
function describeError(error: unknown): string { if (!(error instanceof Error)) { return String(error); } const cause = (error as { cause?: unknown }).cause; if (isRecord(cause)) { const code = typeof cause.code === "string" ? cause.code : undefined; const causeMessage = typeof cause.message === "string" ? cause.message : undefined; const detail = code ?? causeMessage; if (detail && detail !== error.message) { return `${error.message} (${detail})`; } } return error.message;}
class TraceUploadTimeoutError extends Error { constructor(timeoutMs: number) { super(`Trace upload timed out after ${timeoutMs}ms`); this.name = "TraceUploadTimeoutError"; }}
const RETRIABLE_NETWORK_CODES = new Set([ "ECONNRESET", "ECONNREFUSED", "ECONNABORTED", "EPIPE", "ETIMEDOUT", "ENETUNREACH", "ENETDOWN", "EAI_AGAIN", "UND_ERR_SOCKET", "UND_ERR_CONNECT_TIMEOUT",]);
// 429 is deliberately absent: a rate-limited upload is rescheduled by its caller instead of sleeping in-request.const RETRIABLE_HTTP_STATUSES = new Set([408, 425, 500, 502, 503, 504]);
function isRetriableNetworkError(error: unknown): boolean { if (error instanceof TraceUploadTimeoutError) { return true; } if (!(error instanceof Error) || error.name === "AbortError") { return false; } const cause = (error as { cause?: unknown }).cause; return isRecord(cause) && typeof cause.code === "string" && RETRIABLE_NETWORK_CODES.has(cause.code);}
function stringField(data: Record<string, unknown>, key: string): string | undefined { const value = data[key]; return typeof value === "string" && value.trim() ? value.trim() : undefined;}
function numberField(data: Record<string, unknown>, key: string): number | undefined { const value = data[key]; return typeof value === "number" && Number.isFinite(value) ? value : undefined;}
function isSessionHeader(value: unknown): value is SessionHeader { return ( isRecord(value) && value.type === "session" && typeof value.id === "string" && typeof value.timestamp === "string" && typeof value.cwd === "string" && (value.parentSession === undefined || typeof value.parentSession === "string") );}
function readSessionHeader(sessionFile: string): SessionHeader | undefined { try { const firstLine = readFirstLineSync(sessionFile); if (!firstLine?.trim()) { return undefined; } const parsed = JSON.parse(firstLine) as unknown; return isSessionHeader(parsed) ? parsed : undefined; } catch { return undefined; }}
/** Active-branch git for the indexing headers: walk leaf to root, not the last git_state in * file order (which may belong to a sibling branch). */export function activeGitContext( body: string, header: SessionHeader,): { repoUrl?: string; commit?: string } | undefined { const byId = new Map<string, { parentId: string | null; type: string; git?: unknown }>(); let leafId: string | null = null; for (const line of body.split("\n")) { if (!line.trim()) continue; let parsed: unknown; try { parsed = JSON.parse(line); } catch { continue; } if (!isRecord(parsed) || parsed.type === "session" || typeof parsed.id !== "string") continue; byId.set(parsed.id, { parentId: typeof parsed.parentId === "string" ? parsed.parentId : null, type: typeof parsed.type === "string" ? parsed.type : "", git: parsed.git, }); leafId = parsed.id; }
let current = leafId ? byId.get(leafId) : undefined; for (let depth = 0; current && depth < byId.size + 1; depth += 1) { if (current.type === "git_state" && isRecord(current.git)) { return current.git as { repoUrl?: string; commit?: string }; } current = current.parentId ? byId.get(current.parentId) : undefined; } return header.git;}
function resolveParentSessionPath(sessionFile: string, parentSession: string): string { return isAbsolute(parentSession) ? parentSession : resolve(dirname(sessionFile), parentSession);}
function resolveTraceContext( sessionFile: string, header: SessionHeader,): { traceId: string; parentSessionId?: string } { let traceId = header.id; let parentSessionId: string | undefined; let currentFile = sessionFile; let currentHeader = header;
for (let depth = 0; depth < 32; depth += 1) { if (!currentHeader.parentSession) { break; }
const parentPath = resolveParentSessionPath(currentFile, currentHeader.parentSession); const parentHeader = readSessionHeader(parentPath); if (!parentHeader) { break; }
if (depth === 0) { parentSessionId = parentHeader.id; } traceId = parentHeader.id; currentFile = parentPath; currentHeader = parentHeader; }
return { traceId, parentSessionId };}
function parseResponseObject(text: string): Record<string, unknown> | undefined { if (!text.trim()) { return undefined; } try { const parsed = JSON.parse(text) as unknown; return isRecord(parsed) ? parsed : undefined; } catch { return undefined; }}
async function readResponseMessage(response: Response): Promise<string> { const text = await response.text().catch(() => ""); if (!text.trim()) { return response.statusText || "Unknown error"; }
const parsed = parseResponseObject(text); if (parsed) { const error = parsed.error; if (isRecord(error)) { const message = stringField(error, "message"); if (message) return message; } const detail = stringField(parsed, "detail"); if (detail) return detail; const message = stringField(parsed, "message"); if (message) return message; }
return text.trim();}
async function fetchWithTimeout( fetchFn: typeof fetch, url: string, init: RequestInit, timeoutMs: number, signal?: AbortSignal,): Promise<Response> { const controller = new AbortController(); const timeoutError = new TraceUploadTimeoutError(timeoutMs); let timedOut = false; const timeout = setTimeout(() => { timedOut = true; controller.abort(timeoutError); }, timeoutMs); timeout.unref(); const onAbort = () => controller.abort(signal?.reason); if (signal?.aborted) { onAbort(); } else { signal?.addEventListener("abort", onAbort, { once: true }); }
try { return await fetchFn(url, { ...init, signal: controller.signal }); } catch (error) { if (timedOut) { throw timeoutError; } throw error; } finally { clearTimeout(timeout); signal?.removeEventListener("abort", onAbort); }}
function delay(ms: number, signal?: AbortSignal): Promise<void> { return new Promise((resolve) => { if (signal?.aborted) { resolve(); return; } const timeout = setTimeout(finish, ms); timeout.unref(); const onAbort = () => finish(); function finish() { clearTimeout(timeout); signal?.removeEventListener("abort", onAbort); resolve(); } signal?.addEventListener("abort", onAbort, { once: true }); });}
function traceUploadRetryDelay(retryIndex: number): number { const exponentialDelay = Math.min( TRACE_UPLOAD_RETRY_MAX_DELAY_MS, TRACE_UPLOAD_RETRY_BASE_DELAY_MS * 2 ** retryIndex, ); const jitterMultiplier = 1 - TRACE_UPLOAD_RETRY_JITTER + Math.random() * TRACE_UPLOAD_RETRY_JITTER * 2; return Math.max(0, Math.round(exponentialDelay * jitterMultiplier));}
function retryAfterDelay(response: Response, capMs: number = TRACE_UPLOAD_RATE_LIMIT_WINDOW_MS): number | undefined { const value = response.headers.get("retry-after")?.trim(); if (!value) { return undefined; } const seconds = Number(value); if (Number.isFinite(seconds) && seconds >= 0) { return Math.min(Math.ceil(seconds * 1_000), capMs); } const retryAt = Date.parse(value); return Number.isFinite(retryAt) ? Math.min(Math.max(0, retryAt - Date.now()), capMs) : undefined;}
type BeforeTraceUploadRequest = () => Promise<void>;
interface TraceUploadRequest { fetchFn: typeof fetch; url: string; init: RequestInit; timeoutMs: number; signal?: AbortSignal; beforeRequest?: BeforeTraceUploadRequest; onUploadDelay?: (delay: AgentTraceUploadDelay) => void;}
async function fetchWithRetry(request: TraceUploadRequest): Promise<Response> { const { fetchFn, url, init, timeoutMs, signal, beforeRequest, onUploadDelay } = request; for (let attempt = 0; ; attempt += 1) { let retryDelayMs: number | undefined; try { await beforeRequest?.(); if (signal?.aborted) { throw signal.reason ?? new Error("Trace upload cancelled"); } const response = await fetchWithTimeout(fetchFn, url, init, timeoutMs, signal); if (attempt >= TRACE_UPLOAD_MAX_RETRIES || !RETRIABLE_HTTP_STATUSES.has(response.status)) { return response; } if (response.status === 503) { retryDelayMs = retryAfterDelay(response); } await response.body?.cancel().catch(() => undefined); } catch (error) { if (signal?.aborted) { throw signal.reason ?? error; } if (attempt >= TRACE_UPLOAD_MAX_RETRIES || !isRetriableNetworkError(error)) { throw error; } }
const backoffMs = retryDelayMs ?? traceUploadRetryDelay(attempt); if (!signal?.aborted) { onUploadDelay?.({ reason: "retry-backoff", delayMs: backoffMs }); } await delay(backoffMs, signal); if (signal?.aborted) { throw signal.reason ?? new Error("Trace upload cancelled"); } }}
function traceContentPreview(body: string, maxChars: number): { content: string; truncated: boolean } { if (body.length <= maxChars) { return { content: body.trimEnd(), truncated: false }; } const marker = "\n... middle of trace omitted ...\n"; const available = Math.max(0, maxChars - marker.length); const headChars = Math.ceil(available / 2); const tailChars = Math.floor(available / 2); return { content: `${body.slice(0, headChars).trimEnd()}${marker}${body.slice(body.length - tailChars).trimStart()}`, truncated: true, };}
export async function previewAgentTraceFile(options: AgentTracePreviewOptions): Promise<AgentTracePreviewResult> { if (!options.sessionFile) { return { status: "no_session_file" }; }
let fileSize: number; try { const stats = await stat(options.sessionFile); if (!stats.isFile()) { return { status: "no_session_file" }; } fileSize = stats.size; } catch { return { status: "no_session_file" }; } if (fileSize === 0) { return { status: "empty_session" }; }
const header = readSessionHeader(options.sessionFile); if (!header) { return { status: "invalid_session", message: "Session file is missing a valid session header" }; }
let body = ""; if (fileSize <= MAX_TRACE_BYTES) { try { body = await readFile(options.sessionFile, "utf8"); } catch (error) { return { status: "failed", message: describeError(error) }; } if (!body.trim()) { return { status: "empty_session" }; } }
const traceContext = resolveTraceContext(options.sessionFile, header); const baseUrl = resolvePrimeAgentTracesBaseUrl(options.baseUrl); const git = body ? activeGitContext(body, header) : header.git; const preview = body ? traceContentPreview(body, Math.max(256, options.maxContentChars ?? TRACE_PREVIEW_MAX_CHARS)) : { content: "", truncated: true }; return { status: "ready", sessionFile: options.sessionFile, sessionId: header.id, traceId: traceContext.traceId, parentSessionId: traceContext.parentSessionId, cwd: header.cwd, size: fileSize, maxBytes: MAX_TRACE_BYTES, uploadable: fileSize <= MAX_TRACE_BYTES, endpoint: `${baseUrl}/api/v1/agent-traces/sessions/${encodeURIComponent(header.id)}`, gitRepo: git?.repoUrl, gitCommit: git?.commit, contentPreview: preview.content, truncated: preview.truncated, };}
async function findSessionFilesUnder(root: string, files: Set<string>): Promise<void> { let entries: Dirent[]; try { entries = await readdir(root, { withFileTypes: true }); } catch { return; }
for (const entry of entries) { const entryPath = join(root, entry.name); if (entry.isDirectory()) { await findSessionFilesUnder(entryPath, files); continue; } if (entry.isFile() && entry.name.endsWith(".jsonl") && readSessionHeader(entryPath)) { files.add(entryPath); } }}
export async function findAgentTraceFiles(sessionDir: string = getSessionsDir()): Promise<string[]> { const files = new Set<string>(); const roots = new Set([resolve(sessionDir), resolve(getSessionArtifactsRoot(sessionDir))]); await Promise.all([...roots].map((root) => findSessionFilesUnder(root, files))); return [...files].sort();}
function createTraceUploadAllRequestGate( signal?: AbortSignal, onUploadDelay?: (delay: AgentTraceUploadDelay) => void,): BeforeTraceUploadRequest { let nextRequestAt = 0; let queue = Promise.resolve(); return () => { const slot = queue.then(async () => { const waitMs = Math.max(0, nextRequestAt - Date.now()); if (waitMs > 0) { onUploadDelay?.({ reason: "rate-limit", delayMs: waitMs }); await delay(waitMs, signal); } if (!signal?.aborted) { nextRequestAt = Date.now() + TRACE_UPLOAD_ALL_MIN_REQUEST_INTERVAL_MS; } }); queue = slot.catch(() => undefined); return slot; };}
export async function uploadAllAgentTraces(options: AgentTraceUploadAllOptions): Promise<AgentTraceUploadAllResult> { const { sessionDir, concurrency, onProgress, ...uploadOptions } = options; const sessionFiles = await findAgentTraceFiles(sessionDir); type UploadResultItem = AgentTraceUploadAllResult["results"][number]; const results: Array<UploadResultItem | undefined> = new Array(sessionFiles.length); let cursor = 0; let completed = 0; const beforeRequest = createTraceUploadAllRequestGate(uploadOptions.signal, uploadOptions.onUploadDelay); onProgress?.({ completed, total: sessionFiles.length });
const worker = async () => { while (true) { if (uploadOptions.signal?.aborted) { return; } const index = cursor; cursor += 1; const sessionFile = sessionFiles[index]; if (!sessionFile) { return; } const result = await uploadAgentTraceFileWithRequestGate( { ...uploadOptions, sessionFile, reloadConfig: false, }, beforeRequest, ); if (uploadOptions.signal?.aborted && result.status === "failed") { return; } results[index] = { sessionFile, result }; completed += 1; onProgress?.({ completed, total: sessionFiles.length, sessionFile, result }); } };
const requestedConcurrency = concurrency ?? TRACE_UPLOAD_ALL_CONCURRENCY; const normalizedConcurrency = Number.isFinite(requestedConcurrency) && requestedConcurrency > 0 ? Math.max(1, Math.floor(requestedConcurrency)) : TRACE_UPLOAD_ALL_CONCURRENCY; const workerCount = Math.min(sessionFiles.length, normalizedConcurrency); await Promise.all(Array.from({ length: workerCount }, worker));
const completedResults = results.filter((item): item is UploadResultItem => item !== undefined); let uploaded = 0; let failed = 0; let bytesStored = 0; for (const item of completedResults) { if (item.result.status === "uploaded") { uploaded += 1; bytesStored += item.result.bytesStored; } else if (item.result.status === "failed") { failed += 1; } } return { total: sessionFiles.length, uploaded, failed, skipped: sessionFiles.length - uploaded - failed, bytesStored, results: completedResults, };}
interface AgentTraceUploadedSignature { size: number; mtimeMs: number;}
export const SEMANTIC_EDGES_OUTBOX_KIND = "semantic-edges";
export interface AgentTraceCatchUpResult { pruned: number; /** Registered semantic-edge ledgers with bytes beyond their cursor; no delivery endpoint exists yet. */ semanticEdgeLedgersPending: number; results: Array<{ sessionFile: string; result: AgentTraceUploadResult }>;}
function getAgentTraceOutboxDir(): string { return join(getAgentDir(), "agent-traces-outbox");}
// One entry file per session file (keyed by path hash): concurrent writers cannot lose each other's cursors, and a bad read costs only its own entry.function agentTraceOutboxEntryPath(sessionFile: string): string { const key = createHash("sha256").update(sessionFile).digest("hex").slice(0, 32); return join(getAgentTraceOutboxDir(), `${key}.json`);}
function parseOutboxEntry(raw: string): | { sessionFile: string; kind?: string; uploaded: AgentTraceUploadedSignature | null; uploadedBytes?: number; } | undefined { const parsed = parseResponseObject(raw); if (!parsed || typeof parsed.sessionFile !== "string") { return undefined; } const uploaded = typeof parsed.size === "number" && typeof parsed.mtimeMs === "number" ? { size: parsed.size, mtimeMs: parsed.mtimeMs } : null; return { sessionFile: parsed.sessionFile, kind: typeof parsed.kind === "string" ? parsed.kind : undefined, uploaded, uploadedBytes: typeof parsed.uploadedBytes === "number" ? parsed.uploadedBytes : undefined, };}
/** `undefined` = no usable cursor; `null` = scheduled but never uploaded. */async function readAgentTraceOutboxEntry(sessionFile: string): Promise<AgentTraceUploadedSignature | null | undefined> { let raw: string; try { raw = await readFile(agentTraceOutboxEntryPath(sessionFile), "utf8"); } catch { return undefined; } const entry = parseOutboxEntry(raw); return entry && entry.sessionFile === sessionFile ? entry.uploaded : undefined;}
function signatureEquals(a: AgentTraceUploadedSignature | null | undefined, b: AgentTraceUploadedSignature): boolean { return a != null && a.size === b.size && a.mtimeMs === b.mtimeMs;}
/** Session files with a live upload controller in this process; catch-up leaves them to their controller. */const locallyManagedSessionFiles = new Set<string>();
/** Best-effort and synchronous: upload intent must be on disk the moment the transcript persist returns. */function markAgentTraceOutboxPendingSync(sessionFile: string, kind?: string): boolean { try { const entryPath = agentTraceOutboxEntryPath(sessionFile); if (existsSync(entryPath)) { return true; } mkdirSync(getAgentTraceOutboxDir(), { recursive: true }); const tempPath = `${entryPath}.${process.pid}.${randomUUID()}.tmp`; writeFileSync( tempPath, `${JSON.stringify(kind === undefined ? { sessionFile } : { sessionFile, kind })}\n`, "utf8", ); renameSync(tempPath, entryPath); return true; } catch { // A broken agent dir must not break session persists. return false; }}
async function recordAgentTraceOutboxUpload( sessionFile: string, signature: AgentTraceUploadedSignature,): Promise<void> { const entryPath = agentTraceOutboxEntryPath(sessionFile); await mkdir(getAgentTraceOutboxDir(), { recursive: true }); const tempPath = `${entryPath}.${process.pid}.${randomUUID()}.tmp`; await writeFile(tempPath, `${JSON.stringify({ sessionFile, ...signature })}\n`, "utf8"); await rename(tempPath, entryPath);}
/** * Startup catch-up: upload every outbox entry whose file content is ahead of its * cursor, and prune entries whose file no longer exists. Runs once per process, * in whichever process hosts sessions (the only place trace upload is installed). */export async function catchUpAgentTraceUploads( options: Omit<AgentTraceUploadOptions, "sessionFile">,): Promise<AgentTraceCatchUpResult> { const catchUp: AgentTraceCatchUpResult = { pruned: 0, semanticEdgeLedgersPending: 0, results: [] }; if (options.requireEnabled !== false && !(await getAgentTracesEnabled(options))) { return catchUp; } let entryNames: string[]; try { entryNames = await readdir(getAgentTraceOutboxDir()); } catch { return catchUp; } const beforeRequest = createTraceUploadAllRequestGate(options.signal); for (const entryName of entryNames) { if (options.signal?.aborted) { break; } if (!entryName.endsWith(".json")) { continue; } const entryPath = join(getAgentTraceOutboxDir(), entryName); let raw: string; try { raw = await readFile(entryPath, "utf8"); } catch { // Transient read error: keep the entry and retry at the next startup. continue; } const entry = parseOutboxEntry(raw); if (!entry) { await unlink(entryPath).catch(() => undefined); catchUp.pruned += 1; continue; } if (entry.kind === SEMANTIC_EDGES_OUTBOX_KIND) { let ledgerStats: Awaited<ReturnType<typeof stat>>; try { ledgerStats = await stat(entry.sessionFile); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") { await unlink(entryPath).catch(() => undefined); catchUp.pruned += 1; } continue; } if (!ledgerStats.isFile()) { await unlink(entryPath).catch(() => undefined); catchUp.pruned += 1; continue; } // Append-only byte cursor: a ledger whose size equals its delivered offset has nothing new. if (ledgerStats.size === entry.uploadedBytes) { continue; } // No delivery endpoint exists yet (verifiers#2449 consumes edges in-band over ACP // metadata; the trace server has no semantic-edges route). The delta and cursor stay // untouched so the first real sender delivers the whole backlog. catchUp.semanticEdgeLedgersPending += 1; continue; } if (entry.kind !== undefined) { // A newer build may register kinds this one cannot deliver; leave their cursors alone. continue; } if (locallyManagedSessionFiles.has(entry.sessionFile)) { continue; } let stats: Awaited<ReturnType<typeof stat>>; try { stats = await stat(entry.sessionFile); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") { await unlink(entryPath).catch(() => undefined); catchUp.pruned += 1; } continue; } if (!stats.isFile()) { await unlink(entryPath).catch(() => undefined); catchUp.pruned += 1; continue; } if (signatureEquals(entry.uploaded, { size: stats.size, mtimeMs: stats.mtimeMs })) { continue; } const result = await uploadAgentTraceFileWithRequestGate( { ...options, sessionFile: entry.sessionFile, reloadConfig: false }, beforeRequest, ); catchUp.results.push({ sessionFile: entry.sessionFile, result }); } return catchUp;}
export async function getPrimeAgentTraceCredential( authStorage: AuthStorage, options: { reloadAuth?: boolean } = {},): Promise<AgentTraceCredential | undefined> { const traceEnvKey = stringEnv("PRIME_AGENT_TRACES_API_KEY"); if (traceEnvKey) { return { apiKey: traceEnvKey, source: "environment", label: "PRIME_AGENT_TRACES_API_KEY" }; }
if (options.reloadAuth !== false) { authStorage.reload(); }
const traceKey = await authStorage.getApiKey(PRIME_AGENT_TRACES_PROVIDER_ID, { includeFallback: false }); if (traceKey) { return { apiKey: traceKey, source: "stored", label: "Prime Agent Traces credential" }; }
const primeEnvKey = stringEnv("PRIME_API_KEY"); if (primeEnvKey) { return { apiKey: primeEnvKey, source: "environment", label: "PRIME_API_KEY" }; }
const primeCredential = authStorage.get(PRIME_INFERENCE_PROVIDER_ID); if (primeCredential) { const primeKey = await authStorage.getApiKey(PRIME_INFERENCE_PROVIDER_ID, { includeFallback: false }); if (primeKey) { return { apiKey: primeKey, source: "prime-inference", label: "Prime Inference credential" }; } }
return undefined;}
async function getAgentTracesEnabled( options: Pick<AgentTraceUploadOptions, "reloadConfig" | "settingsManager">,): Promise<boolean> { if (options.reloadConfig !== false) { await options.settingsManager.reload().catch(() => undefined); } return options.settingsManager.getAgentTracesEnabled();}
async function uploadAgentTraceFileWithRequestGate( options: AgentTraceUploadOptions, beforeRequest?: BeforeTraceUploadRequest,): Promise<AgentTraceUploadResult> { const result = await performAgentTraceUpload(options, beforeRequest); logAgentTraceOutcome(options.sessionFile, result); return result;}
export function uploadAgentTraceFile(options: AgentTraceUploadOptions): Promise<AgentTraceUploadResult> { return uploadAgentTraceFileWithRequestGate(options);}
function logAgentTraceOutcome(sessionFile: string | undefined, result: AgentTraceUploadResult): void { let line: string | undefined; switch (result.status) { case "uploaded": line = `uploaded session ${result.sessionId} (${result.bytesStored} bytes)`; break; case "failed": line = `upload failed${result.statusCode ? ` (HTTP ${result.statusCode})` : ""}: ${result.message}`; break; case "too_large": line = `upload skipped: session is ${result.size} bytes (limit ${result.maxBytes})`; break; case "invalid_session": line = `upload skipped: ${result.message}`; break; case "missing_credentials": line = "upload skipped: no Prime credential configured (run /traces login)"; break; default: return; } const suffix = sessionFile ? ` [${sessionFile}]` : ""; appendRotatingLog(getAgentTracesLogPath(), `[${new Date().toISOString()}] ${line}${suffix}`);}
async function performAgentTraceUpload( options: AgentTraceUploadOptions, beforeRequest?: BeforeTraceUploadRequest,): Promise<AgentTraceUploadResult> { const requireEnabled = options.requireEnabled !== false; if (requireEnabled && !(await getAgentTracesEnabled(options))) { return { status: "disabled" }; } if (!options.sessionFile) { return { status: "no_session_file" }; }
let signature: AgentTraceUploadedSignature; try { const stats = await stat(options.sessionFile); if (!stats.isFile()) { return { status: "no_session_file" }; } signature = { size: stats.size, mtimeMs: stats.mtimeMs }; } catch { return { status: "no_session_file" }; } const fileSize = signature.size; if (fileSize === 0) { return { status: "empty_session" }; } if (fileSize > MAX_TRACE_BYTES) { return { status: "too_large", size: fileSize, maxBytes: MAX_TRACE_BYTES }; } // Cursor invariant: an automatic upload never re-sends a file whose content already matches its uploaded cursor. if (requireEnabled && signatureEquals(await readAgentTraceOutboxEntry(options.sessionFile), signature)) { return { status: "unchanged" }; }
const header = readSessionHeader(options.sessionFile); if (!header) { return { status: "invalid_session", message: "Session file is missing a valid session header" }; }
const credential = await getPrimeAgentTraceCredential(options.authStorage, { reloadAuth: options.reloadConfig !== false, }); if (!credential) { return { status: "missing_credentials" }; }
if (requireEnabled && !(await getAgentTracesEnabled(options))) { return { status: "disabled" }; }
let body: string; try { body = await readFile(options.sessionFile, "utf8"); } catch (error) { return { status: "failed", message: describeError(error) }; } if (!body.trim()) { return { status: "empty_session" }; }
const traceContext = resolveTraceContext(options.sessionFile, header); const bodyBytes = Buffer.byteLength(body, "utf8"); const headers: Record<string, string> = { Authorization: `Bearer ${credential.apiKey}`, "Content-Type": "application/x-ndjson", Accept: "application/json", "X-Trace-Id": traceContext.traceId, "X-Cwd": header.cwd, "X-Agent-Version": VERSION, }; if (traceContext.parentSessionId) { headers["X-Parent-Session"] = traceContext.parentSessionId; } const git = activeGitContext(body, header); if (git?.repoUrl) { headers["X-Git-Repo"] = git.repoUrl; } if (git?.commit) { headers["X-Git-Commit"] = git.commit; }
if (requireEnabled && !(await getAgentTracesEnabled(options))) { return { status: "disabled" }; }
const baseUrl = resolvePrimeAgentTracesBaseUrl(options.baseUrl); const url = `${baseUrl}/api/v1/agent-traces/sessions/${encodeURIComponent(header.id)}`; const fetchFn = options.fetchFn ?? fetch;
let response: Response; try { response = await fetchWithRetry({ fetchFn, url, init: { method: "PUT", headers, body }, timeoutMs: options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS, signal: options.signal, beforeRequest, onUploadDelay: options.onUploadDelay, }); } catch (error) { return { status: "failed", message: describeError(error) }; }
if (!response.ok) { return { status: "failed", statusCode: response.status, message: await readResponseMessage(response), retryAfterMs: retryAfterDelay(response, MAX_TIMER_DELAY_MS), }; }
const responseText = await response.text().catch(() => ""); const responseData = parseResponseObject(responseText); try { await recordAgentTraceOutboxUpload(options.sessionFile, signature); } catch (error) { return { status: "failed", message: `stored, but recording the upload cursor failed: ${describeError(error)}` }; } return { status: "uploaded", sessionId: responseData ? (stringField(responseData, "session_id") ?? header.id) : header.id, traceId: responseData ? (stringField(responseData, "trace_id") ?? traceContext.traceId) : traceContext.traceId, bytesStored: responseData ? (numberField(responseData, "bytes_stored") ?? bodyBytes) : bodyBytes, key: responseData ? stringField(responseData, "key") : undefined, };}
export function uploadAgentTraceSession(options: AgentTraceSessionUploadOptions): Promise<AgentTraceUploadResult> { return uploadAgentTraceFile({ ...options, sessionFile: options.sessionManager.getSessionFile(), });}
/** Bounds whenIdle() drain loops so a retry that re-arms cannot spin forever. */const MAX_IDLE_DRAIN_CYCLES = 4;
class AgentTraceUploadController { private timeout: NodeJS.Timeout | undefined; private pending = false; private inFlight: Promise<AgentTraceUploadResult> | undefined; private lastUploadStartedAt: number | undefined; private notBeforeAt = 0;
constructor( private readonly sessionManager: SessionManager, private options: AgentTraceUploadInstallOptions, ) {}
update(options: AgentTraceUploadInstallOptions): void { this.options = options; }
schedule = (): void => { this.pending = true; // Intent is consent-gated at persist time: an entry created while sharing // is off would turn a later enable into retroactive collection of // opted-out sessions. Marking re-runs every persist (existsSync-cheap), // so an entry pruned by a racing catch-up is re-registered. if (this.options.settingsManager.getAgentTracesEnabled()) { const sessionFile = this.sessionManager.getSessionFile(); if ( sessionFile && !locallyManagedSessionFiles.has(sessionFile) && markAgentTraceOutboxPendingSync(sessionFile) ) { locallyManagedSessionFiles.add(sessionFile); } const ledgerPath = this.options.semanticEdgesLedgerPath; if (ledgerPath) { markAgentTraceOutboxPendingSync(ledgerPath, SEMANTIC_EDGES_OUTBOX_KIND); } } this.arm(); };
async whenIdle(): Promise<void> { // A debounced upload is still work in progress: waiting only on `inFlight` // lets a caller tear down during the delay and drop the scheduled upload. // Flush the armed timer instead of waiting for it, so draining never // depends on wall-clock time. Bounded because a retry re-arms. for (let drain = 0; drain < MAX_IDLE_DRAIN_CYCLES; drain += 1) { // An active writer is always awaited first, so a drain never returns while a // request is still in flight. if (this.inFlight !== undefined) { await this.inFlight; continue; } if (this.timeout !== undefined) { if (!this.isUploadDue()) { // Backoff outlives this drain; the durable outbox replays it on the // next catch-up rather than overrunning the server's Retry-After. return; } clearTimeout(this.timeout); this.timeout = undefined; await this.runScheduledUpload(); continue; } return; } }
private arm(): void { if (this.timeout) { clearTimeout(this.timeout); } const elapsed = this.lastUploadStartedAt === undefined ? undefined : Date.now() - this.lastUploadStartedAt; const throttleDelay = elapsed === undefined ? 0 : Math.max(0, TRACE_UPLOAD_MIN_INTERVAL_MS - elapsed); const notBeforeDelay = Math.max(0, this.notBeforeAt - Date.now()); const delayMs = Math.max(TRACE_UPLOAD_DEBOUNCE_MS, throttleDelay, notBeforeDelay); this.timeout = setTimeout(() => { this.timeout = undefined; void this.runScheduledUpload(); }, delayMs); this.timeout.unref(); this.options.onUploadScheduled?.({ delayMs }); }
/** True when neither the server's Retry-After nor the throttle window blocks an upload. */ private isUploadDue(): boolean { const now = Date.now(); if (now < this.notBeforeAt) { return false; } return this.lastUploadStartedAt === undefined || now - this.lastUploadStartedAt >= TRACE_UPLOAD_MIN_INTERVAL_MS; }
private async runScheduledUpload(): Promise<void> { if (!this.isUploadDue()) { // A flush must not overrun Retry-After or the throttle window; re-arm instead. this.pending = true; this.arm(); return; } if (this.inFlight) { // Keep the work pending so the in-flight upload's settle path re-arms: // the coalesce notice always has a guaranteed follow-up behind it. this.pending = true; this.options.onUploadSettled?.({ status: "coalesced" }); return; } this.pending = false; this.lastUploadStartedAt = Date.now(); const upload = uploadAgentTraceSession({ ...this.options, sessionManager: this.sessionManager, }).catch((error: unknown): AgentTraceUploadResult => ({ status: "failed", message: describeError(error) })); this.inFlight = upload; const result = await upload; if (result.status === "failed" && isRescheduledUploadFailure(result.statusCode)) { this.pending = true; if (result.retryAfterMs !== undefined) { this.notBeforeAt = Date.now() + result.retryAfterMs; } } this.inFlight = undefined; if (this.pending) { this.arm(); } this.options.onUploadSettled?.(result); }}
function isRescheduledUploadFailure(statusCode: number | undefined): boolean { return statusCode === undefined || statusCode === 429 || RETRIABLE_HTTP_STATUSES.has(statusCode);}
const traceUploadControllers = new WeakMap<SessionManager, AgentTraceUploadController>();let startupCatchUp: Promise<void> | undefined;
export function installAgentTraceUpload( sessionManager: SessionManager, options: AgentTraceUploadInstallOptions,): AgentTraceUploadInstallation { if (!startupCatchUp) { startupCatchUp = catchUpAgentTraceUploads(options).then( () => undefined, () => undefined, ); } const pendingCatchUp = startupCatchUp; let controller = traceUploadControllers.get(sessionManager); if (controller) { controller.update(options); } else { controller = new AgentTraceUploadController(sessionManager, options); traceUploadControllers.set(sessionManager, controller); sessionManager.onPersist(controller.schedule); } const installed = controller; return { whenIdle: async () => { await pendingCatchUp; await installed.whenIdle(); }, };}