diff --git a/packages/coding-agent/.changes/eng-5838-traces-outbox.md b/packages/coding-agent/.changes/eng-5838-traces-outbox.md new file mode 100644 index 0000000000..1911ba4011 --- /dev/null +++ b/packages/coding-agent/.changes/eng-5838-traces-outbox.md @@ -0,0 +1 @@ +- Reworked agent-trace upload scheduling as a disk-cursor outbox: upload intent and per-session uploaded-content cursors persist as one small entry file per session under `agent-traces-outbox/` in the agent dir, a startup catch-up uploads anything a previous process never finished (pruning cursors of deleted session files), scheduled and catch-up uploads never re-send unchanged sessions (the explicit `/traces upload` command still force-uploads), and rate-limited uploads reschedule (honoring an advertised Retry-After) instead of sleeping. Session disposal and process exit no longer wait on trace uploads at all, and upload timers never keep the process alive; the exit drain barrier is gone (the startup catch-up replaces it). diff --git a/packages/coding-agent/src/core/agent-session-runtime.ts b/packages/coding-agent/src/core/agent-session-runtime.ts index fd5465833e..21dd2c9ff0 100644 --- a/packages/coding-agent/src/core/agent-session-runtime.ts +++ b/packages/coding-agent/src/core/agent-session-runtime.ts @@ -7,7 +7,6 @@ import type { AgentSessionRuntimeDiagnostic, AgentSessionServices, } from "./agent-session-services.js"; -import { flushAgentTraceUpload, logDetachedAgentTraceFlushFailure } from "./agent-traces.js"; import { isNoModelsAvailableMessage } from "./auth-guidance.js"; import type { ReplacedSessionContext, SessionShutdownEvent, SessionStartEvent } from "./extensions/index.js"; import { emitSessionShutdownEvent } from "./extensions/runner.js"; @@ -204,7 +203,6 @@ export class AgentSessionRuntime implements SubagentRuntimeHost { reason, targetSessionFile, }); - this.detachTraceFlush(); this.beforeSessionInvalidate?.(); // Await the kernel's final snapshot flush before invalidating the session. await this.session.disposeAsync(); @@ -686,13 +684,6 @@ export class AgentSessionRuntime implements SubagentRuntimeHost { return { cancelled: false }; } - private detachTraceFlush(): void { - const sessionManager = this.session.sessionManager; - void flushAgentTraceUpload(sessionManager).catch((error) => - logDetachedAgentTraceFlushFailure(sessionManager.getSessionFile(), error), - ); - } - private async disposeOnce(options: AgentSessionRuntimeDisposeOptions): Promise { let disposeError: unknown; try { @@ -703,7 +694,6 @@ export class AgentSessionRuntime implements SubagentRuntimeHost { } catch (error) { disposeError ??= error; } - this.detachTraceFlush(); try { this.beforeSessionInvalidate?.(); } catch (error) { diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 39dab65732..15a2e9e2c6 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -71,7 +71,6 @@ import { normalizeObserveMaxChars, ORCHESTRATION_HEARTBEAT_SKILL_NAME, } from "./agent-observe.js"; -import { flushAgentTraceUpload } from "./agent-traces.js"; import { addLoginGuidanceToAuthError, formatAuthenticationFailedMessage, @@ -10591,7 +10590,6 @@ export class AgentSession { } const text = compactRlmText(readAssistantText(assistant)); if (text) run.answerPreview = text; - void flushAgentTraceUpload(child.sessionManager).catch(() => undefined); emitChildUpdate(); } else if (event.type === "message_start" || event.type === "message_update") { if (event.message.role === "assistant") { diff --git a/packages/coding-agent/src/core/agent-traces.ts b/packages/coding-agent/src/core/agent-traces.ts index 77b165b8d2..c1c6f5648d 100644 --- a/packages/coding-agent/src/core/agent-traces.ts +++ b/packages/coding-agent/src/core/agent-traces.ts @@ -1,8 +1,9 @@ import { Buffer } from "node:buffer"; -import type { Dirent } from "node:fs"; -import { readdir, readFile, stat } from "node:fs/promises"; +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, getAgentTracesLogPath, getSessionsDir, VERSION } from "../config.js"; +import { appendRotatingLog, getAgentDir, getAgentTracesLogPath, getSessionsDir, VERSION } from "../config.js"; import { readFirstLineSync } from "../utils/file-lines.js"; import type { AuthStorage } from "./auth-storage.js"; import { @@ -27,6 +28,7 @@ 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; @@ -47,12 +49,13 @@ export type AgentTraceUploadResult = 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 }; + | { status: "failed"; statusCode?: number; message: string; retryAfterMs?: number }; export interface AgentTraceUploadOptions { sessionFile: string | undefined; @@ -176,7 +179,8 @@ const RETRIABLE_NETWORK_CODES = new Set([ "UND_ERR_CONNECT_TIMEOUT", ]); -const RETRIABLE_HTTP_STATUSES = new Set([408, 425, 429, 500, 502, 503, 504]); +// 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) { @@ -341,6 +345,7 @@ async function fetchWithTimeout( timedOut = true; controller.abort(timeoutError); }, timeoutMs); + timeout.unref(); const onAbort = () => controller.abort(signal?.reason); if (signal?.aborted) { onAbort(); @@ -368,6 +373,7 @@ function delay(ms: number, signal?: AbortSignal): Promise { return; } const timeout = setTimeout(finish, ms); + timeout.unref(); const onAbort = () => finish(); function finish() { clearTimeout(timeout); @@ -387,19 +393,17 @@ function traceUploadRetryDelay(retryIndex: number): number { return Math.max(0, Math.round(exponentialDelay * jitterMultiplier)); } -function retryAfterDelay(response: Response): number | undefined { +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), TRACE_UPLOAD_RATE_LIMIT_WINDOW_MS); + 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()), TRACE_UPLOAD_RATE_LIMIT_WINDOW_MS) - : undefined; + return Number.isFinite(retryAt) ? Math.min(Math.max(0, retryAt - Date.now()), capMs) : undefined; } type BeforeTraceUploadRequest = () => Promise; @@ -423,9 +427,7 @@ async function fetchWithRetry( if (attempt >= TRACE_UPLOAD_MAX_RETRIES || !RETRIABLE_HTTP_STATUSES.has(response.status)) { return response; } - if (response.status === 429) { - retryDelayMs = retryAfterDelay(response) ?? TRACE_UPLOAD_RATE_LIMIT_WINDOW_MS; - } else if (response.status === 503) { + if (response.status === 503) { retryDelayMs = retryAfterDelay(response); } await response.body?.cancel().catch(() => undefined); @@ -632,6 +634,158 @@ export async function uploadAllAgentTraces(options: AgentTraceUploadAllOptions): }; } +interface AgentTraceUploadedSignature { + size: number; + mtimeMs: number; +} + +export interface AgentTraceCatchUpResult { + pruned: 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; uploaded: AgentTraceUploadedSignature | null } | 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, uploaded }; +} + +/** `undefined` = no usable cursor; `null` = scheduled but never uploaded. */ +async function readAgentTraceOutboxEntry(sessionFile: string): Promise { + 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(); + +/** Best-effort and synchronous: upload intent must be on disk the moment the transcript persist returns. */ +function markAgentTraceOutboxPendingSync(sessionFile: 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({ sessionFile })}\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 { + 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, +): Promise { + const catchUp: AgentTraceCatchUpResult = { pruned: 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 (locallyManagedSessionFiles.has(entry.sessionFile)) { + continue; + } + let stats: Awaited>; + 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; configPath?: string } = {}, @@ -730,22 +884,27 @@ async function performAgentTraceUpload( return { status: "no_session_file" }; } - let fileSize: number; + let signature: AgentTraceUploadedSignature; try { const stats = await stat(options.sessionFile); if (!stats.isFile()) { return { status: "no_session_file" }; } - fileSize = stats.size; + 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) { @@ -826,11 +985,17 @@ async function performAgentTraceUpload( 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, @@ -850,10 +1015,9 @@ export function uploadAgentTraceSession(options: AgentTraceSessionUploadOptions) class AgentTraceUploadController { private timeout: NodeJS.Timeout | undefined; private pending = false; - private inFlight: Promise | undefined; - private flushPromise: Promise | undefined; + private inFlight: Promise | undefined; private lastUploadStartedAt: number | undefined; - private lastUploadedSignature: string | undefined; + private notBeforeAt = 0; constructor( private readonly sessionManager: SessionManager, @@ -866,96 +1030,70 @@ class AgentTraceUploadController { schedule = (): void => { this.pending = true; - pendingUploadControllers.add(this); + const sessionFile = this.sessionManager.getSessionFile(); + if (sessionFile && !locallyManagedSessionFiles.has(sessionFile) && markAgentTraceOutboxPendingSync(sessionFile)) { + locallyManagedSessionFiles.add(sessionFile); + } + this.arm(); + }; + + private arm(): void { if (this.timeout) { clearTimeout(this.timeout); } - const elapsed = this.lastUploadStartedAt === undefined ? 0 : Date.now() - this.lastUploadStartedAt; - const throttleDelay = - this.lastUploadStartedAt === undefined ? 0 : Math.max(0, TRACE_UPLOAD_MIN_INTERVAL_MS - elapsed); + 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()); this.timeout = setTimeout( () => { this.timeout = undefined; - void this.flush().catch(() => undefined); + void this.runScheduledUpload(); }, - Math.max(TRACE_UPLOAD_DEBOUNCE_MS, throttleDelay), + Math.max(TRACE_UPLOAD_DEBOUNCE_MS, throttleDelay, notBeforeDelay), ); - }; - - private async getCurrentFileSignature(): Promise { - const sessionFile = this.sessionManager.getSessionFile(); - if (!sessionFile) { - return undefined; - } - try { - const stats = await stat(sessionFile); - return `${sessionFile}:${stats.size}:${stats.mtimeMs}`; - } catch { - return undefined; - } - } - - async flush(): Promise { - if (this.flushPromise) { - const result = await this.flushPromise; - if (!this.pending) { - return result; - } - return (await this.flush()) ?? result; - } - - this.flushPromise = this.runFlush(); - try { - return await this.flushPromise; - } finally { - this.flushPromise = undefined; - if (!this.pending) { - pendingUploadControllers.delete(this); - } - } + this.timeout.unref(); } - private async runFlush(): Promise { - if (this.timeout) { - clearTimeout(this.timeout); - this.timeout = undefined; - } + private async runScheduledUpload(): Promise { if (this.inFlight) { - await this.inFlight.catch(() => undefined); - } - if (!this.pending) { - return undefined; - } - - const signature = await this.getCurrentFileSignature(); - if (signature && signature === this.lastUploadedSignature) { - this.pending = false; - return undefined; + return; } - this.pending = false; this.lastUploadStartedAt = Date.now(); this.inFlight = uploadAgentTraceSession({ ...this.options, sessionManager: this.sessionManager, - }); - try { - const result = await this.inFlight; - if (result.status === "uploaded" && signature) { - this.lastUploadedSignature = signature; - } - return result; - } finally { - this.inFlight = undefined; + }).then( + (result) => { + if (result.status === "failed" && isRescheduledUploadFailure(result.statusCode)) { + this.pending = true; + if (result.retryAfterMs !== undefined) { + this.notBeforeAt = Date.now() + result.retryAfterMs; + } + } + }, + () => undefined, + ); + await this.inFlight; + this.inFlight = undefined; + if (this.pending) { + this.arm(); } } } +function isRescheduledUploadFailure(statusCode: number | undefined): boolean { + return statusCode === undefined || statusCode === 429 || RETRIABLE_HTTP_STATUSES.has(statusCode); +} + const traceUploadControllers = new WeakMap(); -// Disposal never awaits uploads; exit paths drain this set instead. -const pendingUploadControllers = new Set(); +let catchUpTriggered = false; export function installAgentTraceUpload(sessionManager: SessionManager, options: AgentTraceUploadInstallOptions): void { + if (!catchUpTriggered) { + catchUpTriggered = true; + void catchUpAgentTraceUploads(options).catch(() => undefined); + } let controller = traceUploadControllers.get(sessionManager); if (controller) { controller.update(options); @@ -966,23 +1104,3 @@ export function installAgentTraceUpload(sessionManager: SessionManager, options: traceUploadControllers.set(sessionManager, controller); sessionManager.onPersist(controller.schedule); } - -export async function flushAgentTraceUpload( - sessionManager: SessionManager, -): Promise { - return traceUploadControllers.get(sessionManager)?.flush(); -} - -/** Exit barrier: drains every scheduled or in-flight trace upload. */ -export async function flushAllPendingAgentTraceUploads(): Promise { - await Promise.allSettled([...pendingUploadControllers].map((controller) => controller.flush())); -} - -/** Log a detached (fire-and-forget) flush rejection; upload outcomes themselves are already logged per attempt. */ -export function logDetachedAgentTraceFlushFailure(sessionFile: string | undefined, error: unknown): void { - const suffix = sessionFile ? ` [${sessionFile}]` : ""; - appendRotatingLog( - getAgentTracesLogPath(), - `[${new Date().toISOString()}] detached flush failed: ${describeError(error)}${suffix}`, - ); -} diff --git a/packages/coding-agent/src/modes/agent-connection/in-process-agent-connection.ts b/packages/coding-agent/src/modes/agent-connection/in-process-agent-connection.ts index 9fc5851d1e..b3d4986e28 100644 --- a/packages/coding-agent/src/modes/agent-connection/in-process-agent-connection.ts +++ b/packages/coding-agent/src/modes/agent-connection/in-process-agent-connection.ts @@ -3,7 +3,6 @@ import type { AgentMessage, ThinkingLevel } from "@earendil-works/pi-agent-core" import type { ImageContent, ServiceTier, Transport } from "@earendil-works/pi-ai"; import type { AgentSessionMessageReceipt, AgentSessionMessageSafetyStatus } from "../../core/agent-messages.js"; import type { AgentSessionRuntime } from "../../core/agent-session-runtime.js"; -import { flushAllPendingAgentTraceUploads } from "../../core/agent-traces.js"; import type { AgentAutonomousStatus } from "../../core/autonomous.js"; import type { BashResult } from "../../core/bash-executor.js"; import type { CompactionResult } from "../../core/compaction/index.js"; @@ -634,11 +633,7 @@ export class InProcessAgentConnection implements AgentConnection { this.runtimeHost.setBeforeSessionInvalidate(undefined); } this.runtimeHost.setRebindSession(undefined); - try { - await this.runtimeHost.dispose(); - } finally { - await flushAllPendingAgentTraceUploads(); - } + await this.runtimeHost.dispose(); } private get session() { diff --git a/packages/coding-agent/src/modes/daemon/daemon-mode.ts b/packages/coding-agent/src/modes/daemon/daemon-mode.ts index b99c047ccf..1cfe6f0117 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-mode.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-mode.ts @@ -73,7 +73,6 @@ import { type CreateAgentSessionRuntimeFactory, createAgentSessionRuntime, } from "../../core/agent-session-runtime.js"; -import { flushAllPendingAgentTraceUploads } from "../../core/agent-traces.js"; import { type AgentCronJob, AgentCronJobStore, @@ -3672,7 +3671,6 @@ export class AgentDaemon { for (const state of [...this.sessions.values()]) { await this.closeSession(state, "killed"); } - await flushAllPendingAgentTraceUploads(); this.fencePeerTransports(); this.writeWorkerSuccess(client, command); setImmediate(() => void this.shutdown(0)); @@ -6256,7 +6254,6 @@ export class AgentDaemon { } } for (const state of [...this.sessions.values()]) await this.closeSession(state, "killed"); - await flushAllPendingAgentTraceUploads(); return manifest; } @@ -7304,12 +7301,8 @@ export class AgentDaemon { cleanup(); } this.cronScheduler.stop(); - try { - for (const state of [...this.sessions.values()]) { - await this.closeSession(state, closingReason); - } - } finally { - await flushAllPendingAgentTraceUploads(); + for (const state of [...this.sessions.values()]) { + await this.closeSession(state, closingReason); } for (const client of this.clients) { client.detachInput(); diff --git a/packages/coding-agent/src/modes/interactive/interactive-mode.ts b/packages/coding-agent/src/modes/interactive/interactive-mode.ts index 53cc14670c..358cbb17cb 100644 --- a/packages/coding-agent/src/modes/interactive/interactive-mode.ts +++ b/packages/coding-agent/src/modes/interactive/interactive-mode.ts @@ -9347,6 +9347,8 @@ export class InteractiveMode { return `Trace uploaded (${result.bytesStored.toLocaleString()} bytes).`; case "disabled": return "Trace sharing is disabled."; + case "unchanged": + return "Trace is already uploaded; no new content since the last upload."; case "missing_credentials": return "Trace sharing needs a Prime API key. Run /traces login."; case "no_session_file": diff --git a/packages/coding-agent/test/agent-connection-in-process.test.ts b/packages/coding-agent/test/agent-connection-in-process.test.ts index aa36e58bc4..beb868cb0c 100644 --- a/packages/coding-agent/test/agent-connection-in-process.test.ts +++ b/packages/coding-agent/test/agent-connection-in-process.test.ts @@ -1,17 +1,9 @@ -import { mkdtempSync, rmSync } from "node:fs"; -import { tmpdir } from "node:os"; -import { join } from "node:path"; import type { AgentMessage } from "@earendil-works/pi-agent-core"; import { getModel } from "@earendil-works/pi-ai"; import { describe, expect, it, vi } from "vitest"; import type { AgentSessionEvent, AgentSessionEventListener, PromptOptions } from "../src/core/agent-session.js"; import type { AgentSessionRuntime } from "../src/core/agent-session-runtime.js"; -import { installAgentTraceUpload } from "../src/core/agent-traces.js"; -import { AuthStorage } from "../src/core/auth-storage.js"; import { emptyGoalState } from "../src/core/goals.js"; -import { PRIME_AGENT_TRACES_PROVIDER_ID } from "../src/core/prime-inference-auth.js"; -import { SessionManager } from "../src/core/session-manager.js"; -import { SettingsManager } from "../src/core/settings-manager.js"; import { InProcessAgentConnection } from "../src/modes/agent-connection/in-process-agent-connection.js"; import type { AgentConnectionEvent, AgentConnectionState } from "../src/modes/agent-connection/types.js"; @@ -151,56 +143,6 @@ function createFakeSession(id: string, messages: AgentMessage[]): FakeSessionCon } describe("InProcessAgentConnection", () => { - it("drains pending trace uploads even when runtime disposal throws", async () => { - const tempDir = mkdtempSync(join(tmpdir(), "prime-agent-in-process-dispose-")); - try { - const sessionManager = SessionManager.create(tempDir, join(tempDir, "sessions")); - sessionManager.newSession(); - const calls: string[] = []; - let releaseFetch: () => void = () => {}; - const gate = new Promise((resolve) => { - releaseFetch = resolve; - }); - installAgentTraceUpload(sessionManager, { - authStorage: AuthStorage.inMemory({ - [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, - }), - settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), - baseUrl: "https://api.example.test", - fetchFn: (async (input: unknown) => { - calls.push(String(input)); - await gate; - return new Response(JSON.stringify({ bytes_stored: 1 }), { - status: 200, - headers: { "content-type": "application/json" }, - }); - }) as typeof fetch, - }); - sessionManager.appendMessage({ role: "user", content: "pending trace data", timestamp: 1 }); - sessionManager.flushNow(); - - const control = createFakeSession("dispose-throws", []); - const runtime = new FakeRuntime(control.session); - runtime.dispose = async () => { - throw new Error("teardown failed"); - }; - const connection = new InProcessAgentConnection(asRuntime(runtime)); - - let settledError: Error | undefined; - const disposal = connection.dispose().catch((error: Error) => { - settledError = error; - return error; - }); - await vi.waitFor(() => expect(calls).toHaveLength(1)); - expect(settledError).toBeUndefined(); - releaseFetch(); - await expect(disposal).resolves.toMatchObject({ message: "teardown failed" }); - expect(calls).toHaveLength(1); - } finally { - rmSync(tempDir, { recursive: true, force: true }); - } - }); - it.each([ { accepted: true, promptResult: "pending", expectedError: undefined }, { accepted: false, promptResult: "resolve", expectedError: "Prompt was not accepted by the session." }, diff --git a/packages/coding-agent/test/agent-traces.test.ts b/packages/coding-agent/test/agent-traces.test.ts index 7694cd8e30..e2c2ff48a8 100644 --- a/packages/coding-agent/test/agent-traces.test.ts +++ b/packages/coding-agent/test/agent-traces.test.ts @@ -1,3 +1,5 @@ +import { Buffer } from "node:buffer"; +import { createHash } from "node:crypto"; import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { stat } from "node:fs/promises"; import { tmpdir } from "node:os"; @@ -6,9 +8,8 @@ import type { AssistantMessage } from "@earendil-works/pi-ai"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { ENV_AGENT_DIR, getAgentTracesLogPath } from "../src/config.js"; import { + catchUpAgentTraceUploads, findAgentTraceFiles, - flushAgentTraceUpload, - flushAllPendingAgentTraceUploads, installAgentTraceUpload, previewAgentTraceFile, uploadAgentTraceFile, @@ -75,8 +76,33 @@ function writeSession(cwd: string, sessionDir: string, id: string, parentSession return sessionManager; } +/** Mirrors the outbox on-disk format: one JSON entry per session file, named by path hash. */ +function outboxEntryPath(agentDir: string, sessionFile: string): string { + const key = createHash("sha256").update(sessionFile).digest("hex").slice(0, 32); + return join(agentDir, "agent-traces-outbox", `${key}.json`); +} + +function writeOutboxEntry(agentDir: string, sessionFile: string, signature?: { size: number; mtimeMs: number }): void { + mkdirSync(join(agentDir, "agent-traces-outbox"), { recursive: true }); + writeFileSync(outboxEntryPath(agentDir, sessionFile), JSON.stringify({ sessionFile, ...signature })); +} + +function readOutboxEntry( + agentDir: string, + sessionFile: string, +): { sessionFile: string; size?: number; mtimeMs?: number } | undefined { + if (!existsSync(outboxEntryPath(agentDir, sessionFile))) { + return undefined; + } + return JSON.parse(readFileSync(outboxEntryPath(agentDir, sessionFile), "utf8")) as { + sessionFile: string; + size?: number; + mtimeMs?: number; + }; +} + async function advanceTimersUntil(condition: () => boolean): Promise { - for (let step = 0; step < 20 && !condition(); step += 1) { + for (let step = 0; step < 200 && !condition(); step += 1) { await stat(new URL(import.meta.url)); if (!condition() && vi.getTimerCount() > 0) { await vi.advanceTimersToNextTimerAsync(); @@ -312,7 +338,42 @@ describe("agent trace upload", () => { expect(calls[0].url).toBe("https://trace-api.example/api/v1/agent-traces/sessions/override-session"); }); + it("runs a startup catch-up on the first trace-upload install", async () => { + const cwd = join(tempDir, "project"); + const sessionDir = join(tempDir, "sessions"); + mkdirSync(cwd, { recursive: true }); + const missed = writeSession(cwd, sessionDir, "missed-session"); + const missedFile = missed.getSessionFile(); + expect(missedFile).toBeDefined(); + writeOutboxEntry(tempDir, missedFile as string); + + const live = SessionManager.create(cwd, sessionDir); + live.newSession({ id: "live-session" }); + const calls: FetchCall[] = []; + installAgentTraceUpload(live, { + authStorage: AuthStorage.inMemory({ + [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, + }), + settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), + baseUrl: "https://api.example.test", + fetchFn: createFetchRecorder(calls), + }); + + await vi.waitFor(() => expect(calls).toHaveLength(1)); + expect(calls[0].url).toBe("https://api.example.test/api/v1/agent-traces/sessions/missed-session"); + expect(calls[0].init.body).toBe(readFileSync(missedFile as string, "utf8")); + const stats = await stat(missedFile as string); + await vi.waitFor(() => + expect(readOutboxEntry(tempDir, missedFile as string)).toEqual({ + sessionFile: missedFile, + size: stats.size, + mtimeMs: stats.mtimeMs, + }), + ); + }); + it("schedules upload only after the session file is persisted", async () => { + vi.useFakeTimers(); const cwd = join(tempDir, "project"); const sessionDir = join(tempDir, "sessions"); mkdirSync(cwd, { recursive: true }); @@ -330,26 +391,21 @@ describe("agent trace upload", () => { }); sessionManager.appendMessage(createUserMessage("hello")); - await flushAgentTraceUpload(sessionManager); - expect(calls).toHaveLength(0); + expect(vi.getTimerCount()).toBe(0); sessionManager.appendMessage(createAssistantMessage("hi")); - await flushAgentTraceUpload(sessionManager); - expect(calls).toHaveLength(1); + await advanceTimersUntil(() => calls.length === 1); expect(calls[0].url).toBe("https://api.example.test/api/v1/agent-traces/sessions/listener-session"); }); - it("serializes concurrent flushes for the same pending upload", async () => { + it("coalesces new content that persists during an in-flight upload into one follow-up upload", async () => { + vi.useFakeTimers(); const cwd = join(tempDir, "project"); const sessionDir = join(tempDir, "sessions"); mkdirSync(cwd, { recursive: true }); const sessionManager = SessionManager.create(cwd, sessionDir); - sessionManager.newSession({ id: "concurrent-flush-session" }); + sessionManager.newSession({ id: "concurrent-upload-session" }); - let markFetchStarted: () => void = () => {}; - const fetchStarted = new Promise((resolve) => { - markFetchStarted = resolve; - }); let releaseFetch: () => void = () => {}; const fetchReleased = new Promise((resolve) => { releaseFetch = resolve; @@ -357,8 +413,9 @@ describe("agent trace upload", () => { const calls: FetchCall[] = []; const fetchFn: typeof fetch = async (input, init) => { calls.push({ url: String(input), init: init ?? {} }); - markFetchStarted(); - await fetchReleased; + if (calls.length === 1) { + await fetchReleased; + } return new Response(JSON.stringify({ bytes_stored: 123 }), { status: 200, headers: { "content-type": "application/json" }, @@ -376,65 +433,24 @@ describe("agent trace upload", () => { sessionManager.appendMessage(createUserMessage("hello")); sessionManager.appendMessage(createAssistantMessage("hi")); - const firstFlush = flushAgentTraceUpload(sessionManager); - const secondFlush = flushAgentTraceUpload(sessionManager); + await advanceTimersUntil(() => calls.length === 1); - await fetchStarted; + // New content lands while the first upload is still in flight. + sessionManager.appendMessage(createUserMessage("more")); + sessionManager.appendMessage(createAssistantMessage("content")); + await advanceTimersUntil(() => vi.getTimerCount() > 0); + await vi.advanceTimersToNextTimerAsync(); expect(calls).toHaveLength(1); - releaseFetch(); - const results = await Promise.all([firstFlush, secondFlush]); - expect(results.map((result) => result?.status)).toEqual(["uploaded", "uploaded"]); - expect(calls).toHaveLength(1); - }); - - it("drains scheduled and in-flight uploads through the exit barrier", async () => { - const cwd = join(tempDir, "project"); - const sessionDir = join(tempDir, "sessions"); - mkdirSync(cwd, { recursive: true }); - let releaseFetch: () => void = () => {}; - const gate = new Promise((resolve) => { - releaseFetch = resolve; - }); - const calls: string[] = []; - const fetchFn: typeof fetch = async (input) => { - calls.push(String(input)); - await gate; - return new Response(JSON.stringify({ bytes_stored: 1 }), { - status: 200, - headers: { "content-type": "application/json" }, - }); - }; - const install = (sessionManager: SessionManager) => { - installAgentTraceUpload(sessionManager, { - authStorage: AuthStorage.inMemory({ - [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, - }), - settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), - baseUrl: "https://api.example.test", - fetchFn, - }); - sessionManager.appendMessage(createUserMessage("hello")); - sessionManager.appendMessage(createAssistantMessage("hi")); - }; - const scheduled = SessionManager.create(cwd, sessionDir); - scheduled.newSession({ id: "barrier-scheduled" }); - install(scheduled); - const inFlight = SessionManager.create(cwd, sessionDir); - inFlight.newSession({ id: "barrier-in-flight" }); - install(inFlight); - const detached = flushAgentTraceUpload(inFlight); - - let drained = false; - const barrier = flushAllPendingAgentTraceUploads().then(() => { - drained = true; - }); - await vi.waitFor(() => expect(calls).toHaveLength(2)); - expect(drained).toBe(false); releaseFetch(); - await barrier; - await detached; - expect(calls).toHaveLength(2); + await advanceTimersUntil(() => calls.length === 2); + const finalBody = readFileSync(sessionManager.getSessionFile() as string, "utf8"); + expect(calls[1].init.body).toBe(finalBody); + // Drain the follow-up upload's completion so its chain cannot leak into later fake-timer tests. + await advanceTimersUntil( + () => + readOutboxEntry(tempDir, sessionManager.getSessionFile() as string)?.size === Buffer.byteLength(finalBody), + ); }); it("schedules automatic uploads at most once per minute and only after new entries persist", async () => { @@ -459,11 +475,7 @@ describe("agent trace upload", () => { sessionManager.appendMessage(createUserMessage("hello")); sessionManager.appendMessage(createAssistantMessage("hi")); expect(Number(setTimeoutSpy.mock.calls.at(-1)?.[1])).toBe(1_000); - await flushAgentTraceUpload(sessionManager); - expect(calls).toHaveLength(1); - - await flushAgentTraceUpload(sessionManager); - expect(calls).toHaveLength(1); + await advanceTimersUntil(() => calls.length === 1); setTimeoutSpy.mockClear(); sessionManager.appendMessage(createUserMessage("next")); @@ -626,44 +638,67 @@ describe("agent trace upload", () => { randomSpy.mockRestore(); }); - it("waits a full platform window before retrying a rate limit without Retry-After", async () => { - vi.useFakeTimers(); + it("returns a rate-limited upload immediately instead of sleeping in-request", async () => { const timeoutSpy = vi.spyOn(globalThis, "setTimeout"); - const session = writeSession(tempDir, join(tempDir, "sessions"), "rate-limit-retry-session"); - let markFirstAttemptStarted: () => void = () => {}; - const firstAttemptStarted = new Promise((resolve) => { - markFirstAttemptStarted = resolve; - }); + const session = writeSession(tempDir, join(tempDir, "sessions"), "rate-limit-session"); let attempts = 0; const fetchFn: typeof fetch = async () => { + attempts += 1; + return new Response(JSON.stringify({ detail: "Too Many Requests" }), { status: 429 }); + }; + + const result = await uploadAgentTraceFile({ + sessionFile: session.getSessionFile(), + authStorage: AuthStorage.inMemory({ + [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, + }), + settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), + baseUrl: "https://api.example.test", + fetchFn, + reloadConfig: false, + }); + + expect(attempts).toBe(1); + expect(result).toMatchObject({ status: "failed", statusCode: 429 }); + expect(timeoutSpy.mock.calls.map((call) => Number(call[1]))).not.toContain(60_000); + }); + + it("reschedules a rate-limited automatic upload without blocking and retries on the next cycle", async () => { + vi.useFakeTimers(); + const cwd = join(tempDir, "project"); + const sessionDir = join(tempDir, "sessions"); + mkdirSync(cwd, { recursive: true }); + const sessionManager = SessionManager.create(cwd, sessionDir); + sessionManager.newSession({ id: "rate-limit-reschedule-session" }); + + let attempts = 0; + const calls: FetchCall[] = []; + const fetchFn: typeof fetch = async (input, init) => { attempts += 1; if (attempts === 1) { - markFirstAttemptStarted(); return new Response(JSON.stringify({ detail: "Too Many Requests" }), { status: 429 }); } - return new Response(JSON.stringify({ bytes_stored: 42 }), { status: 200 }); + return createFetchRecorder(calls)(input, init); }; - const upload = uploadAgentTraceFile({ - sessionFile: session.getSessionFile(), + installAgentTraceUpload(sessionManager, { authStorage: AuthStorage.inMemory({ [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, }), settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), baseUrl: "https://api.example.test", fetchFn, - reloadConfig: false, }); - await firstAttemptStarted; - await vi.runAllTimersAsync(); - const result = await upload; + sessionManager.appendMessage(createUserMessage("hello")); + sessionManager.appendMessage(createAssistantMessage("hi")); + await advanceTimersUntil(() => attempts === 1); + // The rate-limited cycle re-arms itself; the retry succeeds without any caller waiting. + await advanceTimersUntil(() => calls.length === 1); expect(attempts).toBe(2); - expect(result.status).toBe("uploaded"); - expect(timeoutSpy.mock.calls.map((call) => Number(call[1]))).toContain(60_000); }); - it.each([429, 503])("honors Retry-After when retrying HTTP %i", async (status) => { + it.each([503])("honors Retry-After when retrying HTTP %i", async (status) => { vi.useFakeTimers(); const timeoutSpy = vi.spyOn(globalThis, "setTimeout"); const session = writeSession(tempDir, join(tempDir, "sessions"), `retry-after-session-${status}`); @@ -713,7 +748,7 @@ describe("agent trace upload", () => { attempts += 1; if (attempts === 1) { markFirstAttemptStarted(); - return new Response(null, { status: 429, headers: { "retry-after": "3600" } }); + return new Response(null, { status: 503, headers: { "retry-after": "3600" } }); } return new Response(JSON.stringify({ bytes_stored: 42 }), { status: 200 }); }; @@ -1078,6 +1113,281 @@ describe("agent trace upload", () => { expect(result.results).toHaveLength(0); }); + it("durably records upload intent on disk before any upload happens", async () => { + vi.useFakeTimers(); + const cwd = join(tempDir, "project"); + const sessionDir = join(tempDir, "sessions"); + mkdirSync(cwd, { recursive: true }); + const sessionManager = SessionManager.create(cwd, sessionDir); + sessionManager.newSession({ id: "unflushed-session" }); + + const calls: FetchCall[] = []; + installAgentTraceUpload(sessionManager, { + authStorage: AuthStorage.inMemory({ + [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, + }), + settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), + baseUrl: "https://api.example.test", + fetchFn: createFetchRecorder(calls), + }); + + sessionManager.appendMessage(createUserMessage("hello")); + sessionManager.appendMessage(createAssistantMessage("hi")); + const sessionFile = sessionManager.getSessionFile(); + expect(sessionFile).toBeDefined(); + + // Synchronously durable: the intent marker is on disk the moment the persist returns. + expect(readOutboxEntry(tempDir, sessionFile as string)).toEqual({ sessionFile }); + expect(calls).toHaveLength(0); + }); + + it("catch-up uploads exactly the content a previous process never uploaded, then goes quiet", async () => { + const cwd = join(tempDir, "project"); + const sessionDir = join(tempDir, "sessions"); + mkdirSync(cwd, { recursive: true }); + const missed = writeSession(cwd, sessionDir, "crash-lost-session"); + const missedFile = missed.getSessionFile() as string; + writeOutboxEntry(tempDir, missedFile); + + const calls: FetchCall[] = []; + const options = { + authStorage: AuthStorage.inMemory({ + [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, + }), + settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), + baseUrl: "https://api.example.test", + fetchFn: createFetchRecorder(calls), + reloadConfig: false, + }; + + const first = await catchUpAgentTraceUploads(options); + expect(first.results.map(({ result }) => result.status)).toEqual(["uploaded"]); + expect(calls).toHaveLength(1); + expect(calls[0].init.body).toBe(readFileSync(missedFile, "utf8")); + const stats = await stat(missedFile); + expect(readOutboxEntry(tempDir, missedFile)).toEqual({ + sessionFile: missedFile, + size: stats.size, + mtimeMs: stats.mtimeMs, + }); + + // Unchanged content: subsequent cycles and restarts never re-POST. + const second = await catchUpAgentTraceUploads(options); + expect(second.results).toEqual([]); + const automatic = await uploadAgentTraceFile({ ...options, sessionFile: missedFile }); + expect(automatic).toEqual({ status: "unchanged" }); + expect(calls).toHaveLength(1); + }); + + it("prunes cursor entries whose session file was deleted", async () => { + const cwd = join(tempDir, "project"); + const sessionDir = join(tempDir, "sessions"); + mkdirSync(cwd, { recursive: true }); + const kept = writeSession(cwd, sessionDir, "kept-session"); + const keptFile = kept.getSessionFile() as string; + const keptStats = await stat(keptFile); + const keptSignature = { size: keptStats.size, mtimeMs: keptStats.mtimeMs }; + const deletedFile = join(sessionDir, "deleted-session.jsonl"); + writeOutboxEntry(tempDir, deletedFile); + writeOutboxEntry(tempDir, keptFile, keptSignature); + + const calls: FetchCall[] = []; + const result = await catchUpAgentTraceUploads({ + authStorage: AuthStorage.inMemory({ + [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, + }), + settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), + baseUrl: "https://api.example.test", + fetchFn: createFetchRecorder(calls), + reloadConfig: false, + }); + + expect(result).toEqual({ pruned: 1, results: [] }); + expect(calls).toHaveLength(0); + expect(existsSync(outboxEntryPath(tempDir, deletedFile))).toBe(false); + expect(readOutboxEntry(tempDir, keptFile)).toEqual({ sessionFile: keptFile, ...keptSignature }); + }); + + it("arms upload timers that never hold the process open", async () => { + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const cwd = join(tempDir, "project"); + const sessionDir = join(tempDir, "sessions"); + mkdirSync(cwd, { recursive: true }); + const sessionManager = SessionManager.create(cwd, sessionDir); + sessionManager.newSession({ id: "unref-session" }); + + const calls: FetchCall[] = []; + installAgentTraceUpload(sessionManager, { + authStorage: AuthStorage.inMemory({ + [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, + }), + settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), + baseUrl: "https://api.example.test", + fetchFn: createFetchRecorder(calls), + }); + + sessionManager.appendMessage(createUserMessage("hello")); + sessionManager.appendMessage(createAssistantMessage("hi")); + const timer = setTimeoutSpy.mock.results.at(-1)?.value as NodeJS.Timeout; + expect(timer.hasRef()).toBe(false); + }); + + it("honors an advertised Retry-After on the next scheduled cycle without blocking", async () => { + vi.useFakeTimers(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const cwd = join(tempDir, "project"); + const sessionDir = join(tempDir, "sessions"); + mkdirSync(cwd, { recursive: true }); + const sessionManager = SessionManager.create(cwd, sessionDir); + sessionManager.newSession({ id: "retry-after-reschedule-session" }); + + let attempts = 0; + const calls: FetchCall[] = []; + const fetchFn: typeof fetch = async (input, init) => { + attempts += 1; + if (attempts === 1) { + return new Response(null, { status: 429, headers: { "retry-after": "300" } }); + } + return createFetchRecorder(calls)(input, init); + }; + + installAgentTraceUpload(sessionManager, { + authStorage: AuthStorage.inMemory({ + [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, + }), + settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), + baseUrl: "https://api.example.test", + fetchFn, + }); + + sessionManager.appendMessage(createUserMessage("hello")); + sessionManager.appendMessage(createAssistantMessage("hi")); + await advanceTimersUntil(() => attempts === 1); + await advanceTimersUntil(() => vi.getTimerCount() > 0); + expect(Number(setTimeoutSpy.mock.calls.at(-1)?.[1])).toBe(300_000); + + // A fresh persist must not re-arm inside the advertised window. + sessionManager.appendMessage(createUserMessage("more")); + expect(Number(setTimeoutSpy.mock.calls.at(-1)?.[1])).toBeGreaterThanOrEqual(299_000); + + await advanceTimersUntil(() => calls.length === 1); + expect(attempts).toBe(2); + }); + + it("retries the intent marker on the next persist after a failed write", async () => { + vi.useFakeTimers(); + const cwd = join(tempDir, "project"); + const sessionDir = join(tempDir, "sessions"); + mkdirSync(cwd, { recursive: true }); + const sessionManager = SessionManager.create(cwd, sessionDir); + sessionManager.newSession({ id: "marker-retry-session" }); + const blocker = join(tempDir, "agent-traces-outbox"); + writeFileSync(blocker, "not a directory"); + + const calls: FetchCall[] = []; + installAgentTraceUpload(sessionManager, { + authStorage: AuthStorage.inMemory({ + [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, + }), + settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), + baseUrl: "https://api.example.test", + fetchFn: createFetchRecorder(calls), + }); + + sessionManager.appendMessage(createUserMessage("hello")); + sessionManager.appendMessage(createAssistantMessage("hi")); + const sessionFile = sessionManager.getSessionFile() as string; + expect(readOutboxEntry(tempDir, sessionFile)).toBeUndefined(); + + rmSync(blocker); + sessionManager.appendMessage(createUserMessage("again")); + expect(readOutboxEntry(tempDir, sessionFile)).toEqual({ sessionFile }); + }); + + it("caps an absurd Retry-After at the max timer delay instead of retrying immediately", async () => { + vi.useFakeTimers(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const cwd = join(tempDir, "project"); + const sessionDir = join(tempDir, "sessions"); + mkdirSync(cwd, { recursive: true }); + const sessionManager = SessionManager.create(cwd, sessionDir); + sessionManager.newSession({ id: "absurd-retry-after-session" }); + + let attempts = 0; + const fetchFn: typeof fetch = async () => { + attempts += 1; + // 30 days, far beyond Node's ~24.8-day timer maximum. + return new Response(null, { status: 429, headers: { "retry-after": "2592000" } }); + }; + + installAgentTraceUpload(sessionManager, { + authStorage: AuthStorage.inMemory({ + [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, + }), + settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), + baseUrl: "https://api.example.test", + fetchFn, + }); + + sessionManager.appendMessage(createUserMessage("hello")); + sessionManager.appendMessage(createAssistantMessage("hi")); + await advanceTimersUntil(() => attempts === 1); + await advanceTimersUntil(() => vi.getTimerCount() > 0); + expect(Number(setTimeoutSpy.mock.calls.at(-1)?.[1])).toBe(2_147_483_647); + }); + + it("returns a retryable failure when the upload cursor cannot be persisted", async () => { + const session = writeSession(tempDir, join(tempDir, "sessions"), "cursor-persist-failure"); + writeFileSync(join(tempDir, "agent-traces-outbox"), "not a directory"); + + const calls: FetchCall[] = []; + const result = await uploadAgentTraceFile({ + sessionFile: session.getSessionFile(), + authStorage: AuthStorage.inMemory({ + [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, + }), + settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), + baseUrl: "https://api.example.test", + fetchFn: createFetchRecorder(calls), + reloadConfig: false, + }); + + expect(calls).toHaveLength(1); + expect(result.status).toBe("failed"); + if (result.status === "failed") { + expect(result.message).toContain("cursor"); + } + }); + + it("a corrupt outbox entry costs only itself; other cursors survive", async () => { + const cwd = join(tempDir, "project"); + const sessionDir = join(tempDir, "sessions"); + mkdirSync(cwd, { recursive: true }); + const kept = writeSession(cwd, sessionDir, "kept-cursor-session"); + const keptFile = kept.getSessionFile() as string; + const keptStats = await stat(keptFile); + writeOutboxEntry(tempDir, keptFile, { size: keptStats.size, mtimeMs: keptStats.mtimeMs }); + writeFileSync(join(tempDir, "agent-traces-outbox", "deadbeef.json"), "not json"); + + const calls: FetchCall[] = []; + const options = { + authStorage: AuthStorage.inMemory({ + [PRIME_AGENT_TRACES_PROVIDER_ID]: { type: "api_key", key: "trace-key" }, + }), + settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), + baseUrl: "https://api.example.test", + fetchFn: createFetchRecorder(calls), + reloadConfig: false, + }; + + const result = await catchUpAgentTraceUploads(options); + expect(result).toEqual({ pruned: 1, results: [] }); + expect(calls).toHaveLength(0); + expect(existsSync(join(tempDir, "agent-traces-outbox", "deadbeef.json"))).toBe(false); + expect(await uploadAgentTraceFile({ ...options, sessionFile: keptFile })).toEqual({ status: "unchanged" }); + expect(calls).toHaveLength(0); + }); + it("prefers the prime-inference credential over the prime-cli config key", async () => { const session = writeSession(tempDir, join(tempDir, "sessions"), "credential-order-session"); const calls: FetchCall[] = []; diff --git a/packages/coding-agent/test/daemon-mode.test.ts b/packages/coding-agent/test/daemon-mode.test.ts index 2fe4077d06..5f0f7fabce 100644 --- a/packages/coding-agent/test/daemon-mode.test.ts +++ b/packages/coding-agent/test/daemon-mode.test.ts @@ -1,3 +1,4 @@ +import { createHash } from "node:crypto"; import { EventEmitter } from "node:events"; import { chmodSync, @@ -15,6 +16,7 @@ import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; import type { Api, Model } from "@earendil-works/pi-ai"; import { describe, expect, it, vi } from "vitest"; +import { ENV_AGENT_DIR } from "../src/config.js"; import { AGENT_FAMILY_REACH_ERROR, type AgentSessionMessageController, @@ -24,7 +26,7 @@ import { } from "../src/core/agent-messages.js"; import type { AgentObserveController } from "../src/core/agent-observe.js"; import type { CreateAgentSessionRuntimeFactory } from "../src/core/agent-session-runtime.js"; -import { flushAgentTraceUpload, installAgentTraceUpload } from "../src/core/agent-traces.js"; +import { installAgentTraceUpload } from "../src/core/agent-traces.js"; import { AuthStorage } from "../src/core/auth-storage.js"; import type { AgentCronJob, AgentCronJobStore } from "../src/core/cron-jobs.js"; import { PRIME_AGENT_TRACES_PROVIDER_ID } from "../src/core/prime-inference-auth.js"; @@ -6515,8 +6517,10 @@ describe("daemon mode helpers", () => { } }); - it("deletes a resident child without awaiting its pending trace upload and skips the doomed kernel snapshot", async () => { - const tempDir = mkdtempSync(join(tmpdir(), "prime-agent-daemon-delete-trace-flush-")); + it("resolves delete_subagent while the child's trace upload is still in flight, then the transcript upload completes", async () => { + const tempDir = mkdtempSync(join(tmpdir(), "prime-agent-daemon-delete-trace-outbox-")); + const originalAgentDir = process.env[ENV_AGENT_DIR]; + process.env[ENV_AGENT_DIR] = tempDir; try { const fixture = makePersistedRlmDaemonFixture(tempDir); const internals = fixture.daemon as unknown as { @@ -6529,24 +6533,40 @@ describe("daemon mode helpers", () => { const { calls, releaseFetch } = installGatedTraceUpload(childManager); childManager.appendMessage({ role: "user", content: "pending trace data", timestamp: 3 }); childManager.flushNow(); + await vi.waitFor(() => expect(calls).toHaveLength(1), { timeout: 5_000 }); + const transcriptAtUpload = readFileSync(fixture.childSessionFile, "utf8"); // The fetch gate is still held: the delete must not await the upload. await internals .createSubagentRuntimeHost(parentState) .deleteRlmSubagentRuntime(fixture.childId, childState.runtime.session); - - await vi.waitFor(() => expect(calls).toHaveLength(1)); + expect(calls).toHaveLength(1); expect(childState.runtime.session.disposeAsync).toHaveBeenCalledWith({ kernelSnapshot: false }); + + // The transcript survives deletion and its upload completes independently. releaseFetch(); - await flushAgentTraceUpload(childManager); - expect(calls).toHaveLength(1); + const entryKey = createHash("sha256").update(fixture.childSessionFile).digest("hex").slice(0, 32); + await vi.waitFor(() => { + const entry = JSON.parse( + readFileSync(join(tempDir, "agent-traces-outbox", `${entryKey}.json`), "utf8"), + ) as { sessionFile: string; size?: number }; + expect(entry.size).toBeGreaterThan(0); + }); + expect(calls[0]?.body).toBe(transcriptAtUpload); } finally { + if (originalAgentDir === undefined) { + delete process.env[ENV_AGENT_DIR]; + } else { + process.env[ENV_AGENT_DIR] = originalAgentDir; + } rmSync(tempDir, { recursive: true, force: true }); } }); it("resolves a delete that joins an in-flight passivation close without awaiting the trace upload", async () => { const tempDir = mkdtempSync(join(tmpdir(), "prime-agent-daemon-delete-passivation-flush-")); + const originalAgentDir = process.env[ENV_AGENT_DIR]; + process.env[ENV_AGENT_DIR] = tempDir; let releaseDispose!: () => void; const disposeGate = new Promise((resolve) => { releaseDispose = resolve; @@ -6574,6 +6594,7 @@ describe("daemon mode helpers", () => { const { calls, releaseFetch } = installGatedTraceUpload(childManager); childManager.appendMessage({ role: "user", content: "pending trace data", timestamp: 3 }); childManager.flushNow(); + await vi.waitFor(() => expect(calls).toHaveLength(1), { timeout: 5_000 }); const passivation = internals.passivateIdleChildren(90, Date.parse("2036-08-01T12:00:00Z"), 1); await disposeStarted; @@ -6586,9 +6607,13 @@ describe("daemon mode helpers", () => { expect(calls).toHaveLength(1); expect(childState.runtime.session.disposeAsync).toHaveBeenCalledWith({ kernelSnapshot: true }); releaseFetch(); - await flushAgentTraceUpload(childManager); } finally { releaseDispose(); + if (originalAgentDir === undefined) { + delete process.env[ENV_AGENT_DIR]; + } else { + process.env[ENV_AGENT_DIR] = originalAgentDir; + } rmSync(tempDir, { recursive: true, force: true }); } }); @@ -9015,10 +9040,10 @@ function makeCronJob(input: { /** Gated fetch stub on a session's trace-upload controller: observes whether a close awaits the upload. */ function installGatedTraceUpload(sessionManager: SessionManager): { - calls: string[]; + calls: Array<{ url: string; body: string }>; releaseFetch: () => void; } { - const calls: string[] = []; + const calls: Array<{ url: string; body: string }> = []; let releaseFetch: () => void = () => {}; const gate = new Promise((resolveGate) => { releaseFetch = resolveGate; @@ -9029,8 +9054,8 @@ function installGatedTraceUpload(sessionManager: SessionManager): { }), settingsManager: SettingsManager.inMemory({ agentTraces: { enabled: true } }), baseUrl: "https://api.example.test", - fetchFn: (async (input: unknown) => { - calls.push(String(input)); + fetchFn: (async (input: unknown, init?: RequestInit) => { + calls.push({ url: String(input), body: String(init?.body ?? "") }); await gate; return new Response(JSON.stringify({ bytes_stored: 1 }), { status: 200,