diff --git a/cli/CLAUDE.md b/cli/CLAUDE.md index ac55d30..7717976 100644 --- a/cli/CLAUDE.md +++ b/cli/CLAUDE.md @@ -75,6 +75,7 @@ cli/src/ | `init --url ` | 初期設定 + hooks + MCP インストール | | `init --url --proxy ` | プロキシ経由で接続 | | `init --url --dev` | 開発モード（ローカルCLIパス使用） | +| `init --url --async` | 非同期送信モードを有効化（`send_mode: async`） | | `init --url --local` | プロジェクト単位で hooks/MCP を設定 | | `init --url --local --separate-local-config` | プロジェクト単位で config も作成 | | `login` | Webログイン URL 発行 | @@ -82,6 +83,7 @@ cli/src/ | `send --claude-session-id ` | 既存セッションを手動送信（差分のみ） | | `mcp-server` | MCPサーバー起動（stdio通信） | | `on` / `off` | hooks + MCP 有効化/無効化 | +| `on --async` | `send_mode` を async に切替（保存） | | `on --local` / `off --local` | プロジェクト単位で hooks + MCP 有効化/無効化 | | `uninstall` | hooks/MCP/config 削除 | | `uninstall --local` | プロジェクト単位の hooks/MCP/config 削除 | @@ -108,12 +110,22 @@ cli/src/ { "server_url": "http://localhost:8080", "api_key": "agtr_xxxxxxxxxxxxxxxxxxxxxxxx", - "proxy_url": "http://proxy.example.com:8080" + "proxy_url": "http://proxy.example.com:8080", + "send_mode": "async" } ``` **proxy_url** はオプション。設定しない場合は環境変数 `HTTPS_PROXY` / `HTTP_PROXY` にフォールバックする。 +**send_mode** はオプション（`"sync"` | `"async"`、未設定は `"sync"`）。 + +| モード | 挙動 | +|--------|------| +| `sync`（既定） | hook が送信（HTTPS 往復）の完了を待つ。従来どおり。 | +| `async` | hook は detached worker を spawn して即 return し、送信は背後で行う。worker は per-session ロックで同一セッションの送信を直列化する。 | + +`async` への切替は `init --async` / `on --async`、確認は `doctor` の `Send mode` 行で行う。手動送信（`--claude-session-id`）は常に同期。 + ### ローカル設定（--local オプション使用時） `--local` オプションを使うと、プロジェクト単位で AgenTrace を有効/無効にできる。 diff --git a/cli/src/commands/doctor.ts b/cli/src/commands/doctor.ts index 32b2a7e..11995c3 100644 --- a/cli/src/commands/doctor.ts +++ b/cli/src/commands/doctor.ts @@ -3,6 +3,7 @@ import { loadConfigWithFallback, getConfigPath, findAndLoadLocalConfig, + getSendMode, } from "../config/manager.js"; import { createDispatcher } from "../utils/proxy.js"; import { fetch } from "undici"; @@ -48,6 +49,7 @@ export async function doctorCommand(): Promise { console.log(` Active config: ${configSource} (${configPath})`); console.log(` Server URL: ${effectiveConfig.server_url}`); console.log(` API Key: ${maskApiKey(effectiveConfig.api_key)}`); + console.log(` Send mode: ${getSendMode(effectiveConfig)}`); if (effectiveConfig.proxy_url) { console.log(` Proxy URL: ${effectiveConfig.proxy_url}`); } diff --git a/cli/src/commands/init.ts b/cli/src/commands/init.ts index 528b019..2b599fb 100644 --- a/cli/src/commands/init.ts +++ b/cli/src/commands/init.ts @@ -21,6 +21,7 @@ export interface InitOptions { dev?: boolean; local?: boolean; separateLocalConfig?: boolean; + async?: boolean; } export async function initCommand(options: InitOptions = {}): Promise { @@ -110,6 +111,7 @@ export async function initCommand(options: InitOptions = {}): Promise { server_url: serverUrlStr, api_key: result.apiKey, ...(options.proxy && { proxy_url: options.proxy }), + ...(options.async && { send_mode: "async" as const }), }; if (options.local && options.separateLocalConfig && projectDir) { @@ -125,6 +127,9 @@ export async function initCommand(options: InitOptions = {}): Promise { if (options.proxy) { console.log(` Proxy: ${options.proxy}`); } + if (options.async) { + console.log(` Send mode: async`); + } // Determine hook command let hookCommand: string | undefined; diff --git a/cli/src/commands/on.ts b/cli/src/commands/on.ts index 156be29..ed50303 100644 --- a/cli/src/commands/on.ts +++ b/cli/src/commands/on.ts @@ -1,7 +1,7 @@ import * as path from "node:path"; import { fileURLToPath } from "node:url"; import { installHooks, installMcpServer, installPreToolUseHook } from "../hooks/installer.js"; -import { loadConfigWithFallback } from "../config/manager.js"; +import { loadConfigWithFallback, persistSendMode } from "../config/manager.js"; const __filename = fileURLToPath(import.meta.url); const __dirname = path.dirname(__filename); @@ -9,6 +9,7 @@ const __dirname = path.dirname(__filename); export interface OnOptions { dev?: boolean; local?: boolean; + async?: boolean; } export async function onCommand(options: OnOptions = {}): Promise { @@ -25,6 +26,15 @@ export async function onCommand(options: OnOptions = {}): Promise { console.log("[Local Mode] Enabling hooks/MCP for this project only\n"); } + if (options.async) { + const result = persistSendMode("async", { cwd: process.cwd() }); + if (result.ok) { + console.log(`✓ Send mode set to async (${result.path})`); + } else { + console.error("✗ Failed to update send_mode: config not found"); + } + } + // Determine hook command let hookCommand: string | undefined; if (options.dev) { diff --git a/cli/src/commands/send.ts b/cli/src/commands/send.ts index 3b0bf0b..ab43f17 100644 --- a/cli/src/commands/send.ts +++ b/cli/src/commands/send.ts @@ -1,7 +1,8 @@ -import { execSync } from "child_process"; -import { loadConfigWithFallback } from "../config/manager.js"; +import { execSync, spawn } from "child_process"; +import { loadConfigWithFallback, getSendMode } from "../config/manager.js"; import { getNewLines, saveCursor, hasCursor } from "../config/cursor.js"; import { sendIngest } from "../utils/http.js"; +import { WORKER_ENV } from "../send/worker.js"; import { findSessionFile, extractCwdFromTranscript, @@ -21,19 +22,38 @@ interface SendTranscriptParams { isHook: boolean; } +export interface RunSendParams { + sessionId: string; + transcriptPath: string; + cwd?: string; +} + +export type SendOutcome = + | { status: "no-config" } + | { status: "no-lines" } + | { status: "no-valid-lines" } + | { status: "sent"; lineCount: number } + | { status: "error"; error: string }; + // Event types that should not be sent to the server (high-volume, not needed for display) const SKIPPED_EVENT_TYPES = ["progress", "file-history-snapshot"]; +// Cap git lookups so a hung git (e.g. a stuck .git/index.lock) cannot block the +// send. In async mode this also keeps the session lock from being held past its +// stale timeout, which would let a later fire take over as a second holder. +const GIT_EXEC_TIMEOUT_MS = 5_000; + function getGitRemoteUrl(cwd: string): string | null { try { const url = execSync("git remote get-url origin", { cwd, encoding: "utf-8", stdio: ["pipe", "pipe", "pipe"], + timeout: GIT_EXEC_TIMEOUT_MS, }).trim(); return url || null; } catch { - return null; // Not a git repo or no remote + return null; // Not a git repo, no remote, or git timed out } } @@ -43,6 +63,7 @@ function getGitBranch(cwd: string): string | null { cwd, encoding: "utf-8", stdio: ["pipe", "pipe", "pipe"], + timeout: GIT_EXEC_TIMEOUT_MS, }).trim(); return branch || null; } catch { @@ -51,34 +72,21 @@ function getGitBranch(cwd: string): string | null { } /** - * Core logic for sending transcript data to the server. - * Shared between hook-based and manual invocations. + * Send the cursor diff to the server and advance the cursor on success. + * Returns an outcome instead of exiting so callers control reporting and, + * for the async worker, lock release. */ -async function sendTranscript(params: SendTranscriptParams): Promise { - const { sessionId, transcriptPath, cwd, isHook } = params; - - const exitWithError = (message: string) => { - console.error(message); - process.exit(isHook ? 0 : 1); - }; +export async function runSend(params: RunSendParams): Promise { + const { sessionId, transcriptPath, cwd } = params; - // Check if config exists (local config takes precedence over global) const config = loadConfigWithFallback(cwd); if (!config) { - exitWithError( - "[agentrace] Warning: Config not found. Run 'npx agentrace init' first." - ); - return; + return { status: "no-config" }; } - // Get new lines from transcript const { lines, totalLineCount } = getNewLines(transcriptPath, sessionId); - if (lines.length === 0) { - if (!isHook) { - console.log("[agentrace] No new lines to send."); - } - process.exit(0); + return { status: "no-lines" }; } // Parse JSONL lines and filter out skipped event types @@ -86,7 +94,6 @@ async function sendTranscript(params: SendTranscriptParams): Promise { for (const line of lines) { try { const parsed = JSON.parse(line) as Record; - // Skip high-volume event types that are not needed for display if (typeof parsed.type === "string" && SKIPPED_EVENT_TYPES.includes(parsed.type)) { continue; } @@ -97,10 +104,7 @@ async function sendTranscript(params: SendTranscriptParams): Promise { } if (transcriptLines.length === 0) { - if (!isHook) { - console.log("[agentrace] No valid transcript lines to send."); - } - process.exit(0); + return { status: "no-valid-lines" }; } // Detect subagent (Task tool) sessions from first transcript line @@ -135,7 +139,6 @@ async function sendTranscript(params: SendTranscriptParams): Promise { gitBranch = getGitBranch(cwd) ?? undefined; } - // Send to server const result = await sendIngest( { session_id: sessionId, @@ -143,7 +146,6 @@ async function sendTranscript(params: SendTranscriptParams): Promise { cwd: cwd, git_remote_url: gitRemoteUrl, git_branch: gitBranch, - // Subagent fields parent_session_id: parentSessionId, agent_id: agentId, is_sidechain: isSidechain || undefined, @@ -152,20 +154,56 @@ async function sendTranscript(params: SendTranscriptParams): Promise { cwd ); - if (result.ok) { - // Update cursor on success - saveCursor(sessionId, totalLineCount); - if (!isHook) { - console.log( - `[agentrace] Sent ${transcriptLines.length} lines for session ${sessionId}` - ); - } - } else { - exitWithError(`[agentrace] Warning: ${result.error}`); - return; + if (!result.ok) { + return { status: "error", error: result.error ?? "unknown error" }; } - process.exit(0); + // Update cursor only on success so a failed send is retried by the next fire. + saveCursor(sessionId, totalLineCount); + return { status: "sent", lineCount: transcriptLines.length }; +} + +/** + * Send wrapper for the synchronous hook path and manual invocation. + * Maps the outcome to logging and the existing exit-code contract + * (hook: always exit 0; manual: exit 1 on error). + */ +async function sendTranscript(params: SendTranscriptParams): Promise { + const { sessionId, transcriptPath, cwd, isHook } = params; + + const outcome = await runSend({ sessionId, transcriptPath, cwd }); + + let exitCode = 0; + switch (outcome.status) { + case "no-config": + console.error( + "[agentrace] Warning: Config not found. Run 'npx agentrace init' first." + ); + exitCode = isHook ? 0 : 1; + break; + case "no-lines": + if (!isHook) { + console.log("[agentrace] No new lines to send."); + } + break; + case "no-valid-lines": + if (!isHook) { + console.log("[agentrace] No valid transcript lines to send."); + } + break; + case "sent": + if (!isHook) { + console.log( + `[agentrace] Sent ${outcome.lineCount} lines for session ${sessionId}` + ); + } + break; + case "error": + console.error(`[agentrace] Warning: ${outcome.error}`); + exitCode = isHook ? 0 : 1; + break; + } + process.exit(exitCode); } /** @@ -204,15 +242,30 @@ export async function sendCommand(): Promise { process.exit(0); } + // Use CLAUDE_PROJECT_DIR (stable project root) instead of cwd (can change during builds) + const projectDir = process.env.CLAUDE_PROJECT_DIR || data.cwd; + + // Async mode: hand off to a detached worker and return immediately, keeping + // the HTTPS send off the hook's critical path (the 10s UserPromptSubmit wait + // is sync-only). spawn() reports launch failures asynchronously, not as a + // throw, so the catch below only covers a synchronous spawn() error; a worker + // that fails to launch is harmless because the cursor only advances on HTTP + // 200, so the batch is retried on the next fire. + if (getSendMode(loadConfigWithFallback(projectDir)) === "async") { + try { + spawnWorker({ sessionId, transcriptPath, projectDir }); + process.exit(0); + } catch { + // fall through to the synchronous send below + } + } + // For UserPromptSubmit, wait for transcript to be written // (Claude hasn't started processing yet, so transcript may not be updated) if (data.hook_event_name === "UserPromptSubmit") { await sleep(10000); } - // Use CLAUDE_PROJECT_DIR (stable project root) instead of cwd (can change during builds) - const projectDir = process.env.CLAUDE_PROJECT_DIR || data.cwd; - await sendTranscript({ sessionId, transcriptPath, @@ -221,6 +274,29 @@ export async function sendCommand(): Promise { }); } +function spawnWorker(payload: { + sessionId: string; + transcriptPath: string; + projectDir?: string; +}): void { + const child = spawn( + process.execPath, + [...process.execArgv, process.argv[1], "__send-worker"], + { + detached: true, + stdio: "ignore", + env: { + ...process.env, + [WORKER_ENV.sessionId]: payload.sessionId, + [WORKER_ENV.transcriptPath]: payload.transcriptPath, + [WORKER_ENV.projectDir]: payload.projectDir ?? "", + }, + } + ); + child.on("error", () => {}); + child.unref(); +} + /** * Manual send command. * Finds session file by ID and sends to server. diff --git a/cli/src/config/manager.test.ts b/cli/src/config/manager.test.ts index 8e6f226..7ea3245 100644 --- a/cli/src/config/manager.test.ts +++ b/cli/src/config/manager.test.ts @@ -11,6 +11,8 @@ import { loadConfigWithFallback, findLocalConfigPath, findAndLoadLocalConfig, + getSendMode, + persistSendMode, type AgentraceConfig, } from "./manager.js"; @@ -20,6 +22,74 @@ describe("config/manager", () => { api_key: "agtr_test_key", }; + describe("getSendMode", () => { + it("returns 'sync' when config is null", () => { + expect(getSendMode(null)).toBe("sync"); + }); + + it("returns 'sync' when send_mode is not set", () => { + expect(getSendMode(testConfig)).toBe("sync"); + }); + + it("returns 'async' when send_mode is 'async'", () => { + expect(getSendMode({ ...testConfig, send_mode: "async" })).toBe("async"); + }); + + it("returns 'sync' when send_mode is 'sync'", () => { + expect(getSendMode({ ...testConfig, send_mode: "sync" })).toBe("sync"); + }); + + it("falls back to 'sync' for an unrecognized send_mode value", () => { + expect( + getSendMode({ ...testConfig, send_mode: "bogus" as unknown as "sync" }) + ).toBe("sync"); + }); + }); + + describe("persistSendMode (local config)", () => { + let tempProjectDir: string; + + beforeEach(() => { + tempProjectDir = fs.mkdtempSync(path.join(os.tmpdir(), "agentrace-sendmode-")); + }); + + afterEach(() => { + if (fs.existsSync(tempProjectDir)) { + fs.rmSync(tempProjectDir, { recursive: true }); + } + }); + + it("sets send_mode in an existing local config", () => { + saveLocalConfig(tempProjectDir, testConfig); + + const result = persistSendMode("async", { cwd: tempProjectDir }); + + expect(result.ok).toBe(true); + expect(loadLocalConfig(tempProjectDir)?.send_mode).toBe("async"); + }); + + it("preserves other config fields when updating send_mode", () => { + saveLocalConfig(tempProjectDir, testConfig); + + persistSendMode("async", { cwd: tempProjectDir }); + + const updated = loadLocalConfig(tempProjectDir); + expect(updated?.server_url).toBe(testConfig.server_url); + expect(updated?.api_key).toBe(testConfig.api_key); + }); + + it("updates a local config found in a parent directory", () => { + saveLocalConfig(tempProjectDir, testConfig); + const subDir = path.join(tempProjectDir, "sub"); + fs.mkdirSync(subDir); + + const result = persistSendMode("async", { cwd: subDir }); + + expect(result.ok).toBe(true); + expect(loadLocalConfig(tempProjectDir)?.send_mode).toBe("async"); + }); + }); + describe("global config", () => { it("getConfigPath returns expected path", () => { expect(getConfigPath()).toBe( diff --git a/cli/src/config/manager.ts b/cli/src/config/manager.ts index e7b3072..68258de 100644 --- a/cli/src/config/manager.ts +++ b/cli/src/config/manager.ts @@ -2,10 +2,41 @@ import * as fs from "node:fs"; import * as path from "node:path"; import * as os from "node:os"; +export type SendMode = "sync" | "async"; + export interface AgentraceConfig { server_url: string; api_key: string; proxy_url?: string; + send_mode?: SendMode; +} + +export function getSendMode(config: AgentraceConfig | null | undefined): SendMode { + return config?.send_mode === "async" ? "async" : "sync"; +} + +/** + * Write send_mode into the effective config file — the local config found in the + * tree (which takes precedence when sending), otherwise the global config. This + * targets the same file that loadConfigWithFallback reads. Returns the updated path. + */ +export function persistSendMode( + mode: SendMode, + opts: { cwd: string } +): { ok: boolean; path?: string } { + const found = findAndLoadLocalConfig(opts.cwd); + if (found) { + const projectDir = path.dirname(path.dirname(found.path)); + saveLocalConfig(projectDir, { ...found.config, send_mode: mode }); + return { ok: true, path: found.path }; + } + + const global = loadConfig(); + if (!global) { + return { ok: false }; + } + saveConfig({ ...global, send_mode: mode }); + return { ok: true, path: getConfigPath() }; } const CONFIG_DIR = path.join(os.homedir(), ".agentrace"); diff --git a/cli/src/index.ts b/cli/src/index.ts index 00d1e73..098eecf 100644 --- a/cli/src/index.ts +++ b/cli/src/index.ts @@ -9,6 +9,7 @@ import { onCommand } from "./commands/on.js"; import { offCommand } from "./commands/off.js"; import { mcpServerCommand } from "./commands/mcp-server.js"; import { doctorCommand } from "./commands/doctor.js"; +import { workerMain } from "./send/worker.js"; const program = new Command(); @@ -22,13 +23,15 @@ program .option("--dev", "Use local CLI path for development") .option("--local", "Install hooks/MCP for current project only (project-local scope)") .option("--separate-local-config", "Store config in project directory (requires --local)") - .action(async (options: { url: string; proxy?: string; dev?: boolean; local?: boolean; separateLocalConfig?: boolean }) => { + .option("--async", "Send transcripts asynchronously (off the hook critical path)") + .action(async (options: { url: string; proxy?: string; dev?: boolean; local?: boolean; separateLocalConfig?: boolean; async?: boolean }) => { await initCommand({ url: options.url, proxy: options.proxy, dev: options.dev, local: options.local, separateLocalConfig: options.separateLocalConfig, + async: options.async, }); }); @@ -64,8 +67,9 @@ program .description("Enable agentrace hooks (credentials preserved)") .option("--dev", "Use local CLI path for development") .option("--local", "Enable hooks/MCP for current project only") - .action(async (options: { dev?: boolean; local?: boolean }) => { - await onCommand({ dev: options.dev, local: options.local }); + .option("--async", "Switch send mode to asynchronous") + .action(async (options: { dev?: boolean; local?: boolean; async?: boolean }) => { + await onCommand({ dev: options.dev, local: options.local, async: options.async }); }); program @@ -90,4 +94,11 @@ program await doctorCommand(); }); +program + .command("__send-worker", { hidden: true }) + .action(async () => { + await workerMain(); + process.exit(0); + }); + program.parse(); diff --git a/cli/src/send/lock.test.ts b/cli/src/send/lock.test.ts new file mode 100644 index 0000000..6a9fafa --- /dev/null +++ b/cli/src/send/lock.test.ts @@ -0,0 +1,296 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import * as fs from "node:fs"; +import * as path from "node:path"; +import * as os from "node:os"; +import { spawn } from "node:child_process"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { + acquireHolder, + acquireSessionLock, + releaseSessionLock, + acquireWaiting, + releaseWaiting, + evictStale, + readMeta, + locksDir, + holderDir, + waitingDir, + MAX_LOCK_MS, +} from "./lock.js"; +import { SEND_TIMEOUT_MS } from "../utils/http.js"; + +const SID = "session-abc"; + +function writeMeta(dir: string, meta: { pid?: number; startedAt: number }): void { + fs.mkdirSync(dir, { recursive: true }); + fs.writeFileSync(path.join(dir, "meta.json"), JSON.stringify(meta)); +} + +// A pid that is essentially guaranteed not to exist. +const DEAD_PID = 2147483646; + +describe("send/lock", () => { + let tmpHome: string; + let prevHome: string | undefined; + let prevUserProfile: string | undefined; + + beforeEach(() => { + tmpHome = fs.mkdtempSync(path.join(os.tmpdir(), "agentrace-lockhome-")); + prevHome = process.env.HOME; + prevUserProfile = process.env.USERPROFILE; + process.env.HOME = tmpHome; + process.env.USERPROFILE = tmpHome; + }); + + afterEach(() => { + if (prevHome === undefined) delete process.env.HOME; + else process.env.HOME = prevHome; + if (prevUserProfile === undefined) delete process.env.USERPROFILE; + else process.env.USERPROFILE = prevUserProfile; + fs.rmSync(tmpHome, { recursive: true, force: true }); + }); + + it("MAX_LOCK_MS is greater than SEND_TIMEOUT_MS (do not evict a slow but live holder)", () => { + expect(MAX_LOCK_MS).toBeGreaterThan(SEND_TIMEOUT_MS); + }); + + it("locksDir is under ~/.agentrace/locks", () => { + expect(locksDir()).toBe(path.join(tmpHome, ".agentrace", "locks")); + }); + + describe("mutual exclusion", () => { + it("first acquire succeeds, second acquire on same sid fails", () => { + expect(acquireHolder(SID)).toBe(true); + expect(acquireHolder(SID)).toBe(false); + }); + + it("writes meta.json with pid and startedAt on acquire", () => { + acquireHolder(SID); + const metaPath = path.join(holderDir(SID), "meta.json"); + expect(fs.existsSync(metaPath)).toBe(true); + const meta = JSON.parse(fs.readFileSync(metaPath, "utf-8")); + expect(meta.pid).toBe(process.pid); + expect(typeof meta.startedAt).toBe("number"); + }); + + it("can re-acquire after release", () => { + expect(acquireHolder(SID)).toBe(true); + releaseSessionLock(SID); + expect(fs.existsSync(holderDir(SID))).toBe(false); + expect(acquireHolder(SID)).toBe(true); + }); + }); + + describe("waiting slot (holder 1 + waiting 1, max 2)", () => { + it("waiting slot can be taken while holder is held, third is rejected", () => { + expect(acquireHolder(SID)).toBe(true); + expect(acquireWaiting(SID)).toBe(true); + // holder + waiting both occupied -> third caller (another waiting) fails + expect(acquireWaiting(SID)).toBe(false); + }); + + it("can re-take waiting slot after release", () => { + acquireHolder(SID); + expect(acquireWaiting(SID)).toBe(true); + releaseWaiting(SID); + expect(acquireWaiting(SID)).toBe(true); + }); + }); + + describe("staleness / self-recovery", () => { + it("evicts a holder whose pid is dead (immediate)", () => { + writeMeta(holderDir(SID), { pid: DEAD_PID, startedAt: Date.now() }); + expect(acquireHolder(SID)).toBe(true); + // new holder meta belongs to us + const meta = JSON.parse( + fs.readFileSync(path.join(holderDir(SID), "meta.json"), "utf-8") + ); + expect(meta.pid).toBe(process.pid); + }); + + it("evicts a holder whose age exceeds MAX_LOCK_MS even if pid is alive", () => { + writeMeta(holderDir(SID), { + pid: process.pid, // alive + startedAt: Date.now() - (MAX_LOCK_MS + 5_000), + }); + expect(acquireHolder(SID)).toBe(true); + }); + + it("does NOT evict a live, recent holder (no meta yet = just-acquired, not stale)", () => { + // holder dir exists but meta not written yet, and dir is fresh + fs.mkdirSync(holderDir(SID), { recursive: true }); + expect(acquireHolder(SID)).toBe(false); + }); + + it("evicts a meta-less holder dir once it is older than MAX_LOCK_MS (crash backstop)", () => { + fs.mkdirSync(holderDir(SID), { recursive: true }); + // backdate the directory's mtime beyond MAX_LOCK_MS + const old = new Date(Date.now() - (MAX_LOCK_MS + 5_000)); + fs.utimesSync(holderDir(SID), old, old); + expect(acquireHolder(SID)).toBe(true); + }); + + it("reclaims a stale waiting slot (crashed waiter) on next acquireWaiting", () => { + acquireHolder(SID); + writeMeta(waitingDir(SID), { pid: DEAD_PID, startedAt: Date.now() }); + expect(acquireWaiting(SID)).toBe(true); + const meta = JSON.parse( + fs.readFileSync(path.join(waitingDir(SID), "meta.json"), "utf-8") + ); + expect(meta.pid).toBe(process.pid); + }); + }); + + describe("release only removes a lock we own", () => { + it("does not remove a holder owned by another process", () => { + writeMeta(holderDir(SID), { pid: DEAD_PID, startedAt: Date.now() }); + releaseSessionLock(SID); + expect(fs.existsSync(holderDir(SID))).toBe(true); + }); + + it("removes a holder we acquired ourselves", () => { + expect(acquireHolder(SID)).toBe(true); + releaseSessionLock(SID); + expect(fs.existsSync(holderDir(SID))).toBe(false); + }); + }); + + describe("acquireSessionLock (high-level)", () => { + it("returns 'acquired' when the holder is free", async () => { + await expect(acquireSessionLock(SID)).resolves.toBe("acquired"); + expect(fs.existsSync(holderDir(SID))).toBe(true); + }); + + it("returns 'dropped' when holder and waiting are both occupied", async () => { + acquireHolder(SID); // holder taken by a live (this) process + acquireWaiting(SID); // waiting taken too + await expect( + acquireSessionLock(SID, { pollIntervalMs: 5, maxWaitMs: 50 }) + ).resolves.toBe("dropped"); + }); + + it("promotes the waiter to holder once the holder releases, freeing the waiting slot", async () => { + acquireHolder(SID); // someone else holds it + const waiterPromise = acquireSessionLock(SID, { + pollIntervalMs: 5, + maxWaitMs: 2_000, + }); + // release the holder shortly after so the waiter can promote + setTimeout(() => releaseSessionLock(SID), 30); + await expect(waiterPromise).resolves.toBe("acquired"); + expect(fs.existsSync(holderDir(SID))).toBe(true); + expect(fs.existsSync(waitingDir(SID))).toBe(false); + }); + }); + + describe("eviction never clobbers a refreshed holder", () => { + it("aborts eviction when the holder was replaced since it was observed as stale", () => { + // A stale holder is observed (dead pid). + writeMeta(holderDir(SID), { pid: DEAD_PID, startedAt: Date.now() }); + const observed = readMeta(holderDir(SID))!; + + // Concurrently, that stale holder is reclaimed and replaced by a fresh, + // live holder (us) with a different instanceId. + fs.rmSync(holderDir(SID), { recursive: true, force: true }); + expect(acquireHolder(SID)).toBe(true); + const freshMeta = readMeta(holderDir(SID))!; + expect(freshMeta.instanceId).not.toBe(observed.instanceId); + + // An evictor still holding the STALE observation must not remove the fresh + // holder's dir. + expect(evictStale(holderDir(SID), observed)).toBe(false); + expect(readMeta(holderDir(SID))).toEqual(freshMeta); + expect(fs.existsSync(holderDir(SID))).toBe(true); + }); + + it("evicts when the observed stale instance is still the current one", () => { + writeMeta(holderDir(SID), { pid: DEAD_PID, startedAt: Date.now() }); + const observed = readMeta(holderDir(SID))!; + expect(evictStale(holderDir(SID), observed)).toBe(true); + expect(fs.existsSync(holderDir(SID))).toBe(false); + }); + + it("treats a meta-less dir with a different mtime as a different instance", () => { + // Observe a meta-less, aged-out holder (crash-before-meta backstop). + fs.mkdirSync(holderDir(SID), { recursive: true }); + const old1 = new Date(Date.now() - (MAX_LOCK_MS + 10_000)); + fs.utimesSync(holderDir(SID), old1, old1); + const observed = readMeta(holderDir(SID))!; + expect(observed.instanceId).toBeUndefined(); + + // Replace it with a different meta-less dir (different mtime, still stale). + fs.rmSync(holderDir(SID), { recursive: true, force: true }); + fs.mkdirSync(holderDir(SID), { recursive: true }); + const old2 = new Date(Date.now() - (MAX_LOCK_MS + 3_000)); + fs.utimesSync(holderDir(SID), old2, old2); + + // The mtime-based identity differs → eviction must abort. + expect(evictStale(holderDir(SID), observed)).toBe(false); + expect(fs.existsSync(holderDir(SID))).toBe(true); + }); + }); + + describe("concurrent stale takeover yields a unique holder", () => { + it("only one of N processes becomes the holder of a stale lock", async () => { + // Pre-create a stale holder (dead pid). + writeMeta(holderDir(SID), { pid: DEAD_PID, startedAt: Date.now() }); + + const here = path.dirname(fileURLToPath(import.meta.url)); + // Import lock.ts by absolute file URL so the racer can live under tmpHome + // (cleaned in afterEach) instead of being written into cli/src/send. + const lockUrl = pathToFileURL(path.join(here, "lock.ts")).href; + const racerPath = path.join(tmpHome, "__lock_racer__.ts"); + const sentinel = path.join(tmpHome, "release-racers"); + // Each racer reports its result, then stays alive until the sentinel file + // appears. The test writes the sentinel only after all racers have reported, + // so every racer is alive while the others attempt acquisition — a winner's + // pid never dies mid-race to let a late starter reclaim it as stale. The + // unique-holder guarantee then rests purely on atomic mkdir, independent of + // process scheduling. + fs.writeFileSync( + racerPath, + `import { acquireHolder } from ${JSON.stringify(lockUrl)};\n` + + `import * as fs from "node:fs";\n` + + `const got = acquireHolder(process.env.SID!);\n` + + `process.stdout.write(got ? "ACQUIRED" : "NOPE");\n` + + `const t = setInterval(() => {\n` + + ` if (fs.existsSync(process.env.SENTINEL!)) { clearInterval(t); process.exit(0); }\n` + + `}, 20);\n` + ); + + try { + const N = 5; + let reported = 0; + const runs = Array.from({ length: N }, () => + new Promise((resolve) => { + // Launch each racer as a separate OS process under tsx so the + // mkdir/rename race is genuinely cross-process, not cooperative. + const child = spawn( + process.execPath, + ["--import", "tsx", racerPath], + { + env: { ...process.env, SID, SENTINEL: sentinel }, + stdio: ["ignore", "pipe", "ignore"], + } + ); + let out = ""; + let counted = false; + child.stdout.on("data", (c) => { + out += c.toString(); + if (!counted) { + counted = true; + if (++reported === N) fs.writeFileSync(sentinel, "go"); + } + }); + child.on("close", () => resolve(out.trim())); + }) + ); + const results = await Promise.all(runs); + const winners = results.filter((r) => r === "ACQUIRED").length; + expect(winners).toBe(1); + } finally { + fs.rmSync(racerPath, { force: true }); + } + }, 30_000); + }); +}); diff --git a/cli/src/send/lock.ts b/cli/src/send/lock.ts new file mode 100644 index 0000000..31feae4 --- /dev/null +++ b/cli/src/send/lock.ts @@ -0,0 +1,181 @@ +import * as fs from "node:fs"; +import * as path from "node:path"; +import * as os from "node:os"; + +export const MAX_LOCK_MS = 60_000; +export const LOCK_POLL_INTERVAL_MS = 100; +export const MAX_WAITER_MS = MAX_LOCK_MS; + +export interface LockMeta { + pid?: number; + startedAt: number; + instanceId?: string; +} + +let staleCounter = 0; +let instanceCounter = 0; + +export function locksDir(): string { + return path.join(os.homedir(), ".agentrace", "locks"); +} + +export function holderDir(sid: string): string { + return path.join(locksDir(), sid); +} + +export function waitingDir(sid: string): string { + return path.join(locksDir(), `${sid}.waiting`); +} + +function isPidAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (err) { + return (err as NodeJS.ErrnoException).code === "EPERM"; + } +} + +export function readMeta(dir: string): LockMeta | null { + let dirStat: fs.Stats; + try { + dirStat = fs.statSync(dir); + } catch { + return null; + } + try { + const raw = fs.readFileSync(path.join(dir, "meta.json"), "utf-8"); + const parsed = JSON.parse(raw) as LockMeta; + if (typeof parsed.startedAt === "number") { + return parsed; + } + } catch { + // meta.json not written yet; identify the dir by its mtime instead. + } + return { startedAt: dirStat.mtimeMs }; +} + +function isStale(meta: LockMeta): boolean { + if (typeof meta.pid === "number" && !isPidAlive(meta.pid)) { + return true; + } + return Date.now() - meta.startedAt > MAX_LOCK_MS; +} + +function sameInstance(a: LockMeta, b: LockMeta): boolean { + if (a.instanceId !== undefined || b.instanceId !== undefined) { + return a.instanceId === b.instanceId; + } + return a.pid === b.pid && a.startedAt === b.startedAt; +} + +// Remove a stale dir only if it is still the same instance `observed` earlier, +// re-checking right before the rename so a freshly re-acquired holder is not +// removed. Returns true only when the observed instance was removed. +export function evictStale(dir: string, observed: LockMeta): boolean { + const current = readMeta(dir); + if (!current || !sameInstance(current, observed) || !isStale(current)) { + return false; + } + const target = `${dir}.stale.${process.pid}.${staleCounter++}`; + try { + fs.renameSync(dir, target); + } catch { + return false; + } + fs.rmSync(target, { recursive: true, force: true }); + return true; +} + +function tryMkdir(dir: string): boolean { + fs.mkdirSync(locksDir(), { recursive: true }); + try { + fs.mkdirSync(dir); + } catch (err) { + if ((err as NodeJS.ErrnoException).code === "EEXIST") { + return false; + } + throw err; + } + const meta: LockMeta = { + pid: process.pid, + startedAt: Date.now(), + instanceId: `${process.pid}-${Date.now()}-${instanceCounter++}`, + }; + fs.writeFileSync(path.join(dir, "meta.json"), JSON.stringify(meta)); + return true; +} + +function takeSlot(dir: string): boolean { + if (tryMkdir(dir)) { + return true; + } + const observed = readMeta(dir); + if (observed && isStale(observed) && evictStale(dir, observed)) { + return tryMkdir(dir); + } + return false; +} + +export function acquireHolder(sid: string): boolean { + return takeSlot(holderDir(sid)); +} + +export function acquireWaiting(sid: string): boolean { + return takeSlot(waitingDir(sid)); +} + +// Remove a slot only if we still own it, so a worker whose lock was stale-evicted +// and re-acquired by another process never deletes that new holder's lock. +function removeIfOwned(dir: string): void { + const meta = readMeta(dir); + if (meta && meta.pid === process.pid) { + fs.rmSync(dir, { recursive: true, force: true }); + } +} + +export function releaseSessionLock(sid: string): void { + removeIfOwned(holderDir(sid)); +} + +export function releaseWaiting(sid: string): void { + removeIfOwned(waitingDir(sid)); +} + +export type AcquireOutcome = "acquired" | "dropped"; + +export interface AcquireOptions { + pollIntervalMs?: number; + maxWaitMs?: number; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +export async function acquireSessionLock( + sid: string, + opts: AcquireOptions = {} +): Promise { + if (acquireHolder(sid)) { + return "acquired"; + } + if (!acquireWaiting(sid)) { + return "dropped"; + } + + const pollIntervalMs = opts.pollIntervalMs ?? LOCK_POLL_INTERVAL_MS; + const maxWaitMs = opts.maxWaitMs ?? MAX_WAITER_MS; + const deadline = Date.now() + maxWaitMs; + try { + while (Date.now() < deadline) { + await sleep(pollIntervalMs); + if (acquireHolder(sid)) { + return "acquired"; + } + } + return "dropped"; + } finally { + releaseWaiting(sid); + } +} diff --git a/cli/src/send/send-async-e2e.test.ts b/cli/src/send/send-async-e2e.test.ts new file mode 100644 index 0000000..7105d74 --- /dev/null +++ b/cli/src/send/send-async-e2e.test.ts @@ -0,0 +1,94 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import * as fs from "node:fs"; +import * as path from "node:path"; +import * as os from "node:os"; +import * as http from "node:http"; +import { spawn } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import type { AddressInfo } from "node:net"; + +const indexPath = path.resolve( + path.dirname(fileURLToPath(import.meta.url)), + "../index.ts" +); + +describe("async send handoff (sendCommand → detached worker)", () => { + let tmpHome: string; + let projectDir: string; + let server: http.Server; + let baseUrl: string; + let received: string[]; + const sockets = new Set(); + + beforeEach(async () => { + tmpHome = fs.mkdtempSync(path.join(os.tmpdir(), "agentrace-asynchome-")); + projectDir = fs.mkdtempSync(path.join(os.tmpdir(), "agentrace-asyncproj-")); + fs.mkdirSync(path.join(tmpHome, ".agentrace"), { recursive: true }); + + received = []; + server = http.createServer((req, res) => { + let body = ""; + req.on("data", (c) => (body += c)); + req.on("end", () => { + received.push(body); + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, events_created: 1 })); + }); + }); + server.on("connection", (s) => { + sockets.add(s); + s.on("close", () => sockets.delete(s)); + }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + baseUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + + fs.writeFileSync( + path.join(tmpHome, ".agentrace", "config.json"), + JSON.stringify({ + server_url: baseUrl, + api_key: "agtr_test", + send_mode: "async", + }) + ); + }); + + afterEach(async () => { + for (const s of sockets) s.destroy(); + sockets.clear(); + await new Promise((r) => server.close(() => r())); + fs.rmSync(tmpHome, { recursive: true, force: true }); + fs.rmSync(projectDir, { recursive: true, force: true }); + }); + + it("the hook send exits immediately and a detached worker delivers", async () => { + const transcriptPath = path.join(projectDir, "t.jsonl"); + fs.writeFileSync(transcriptPath, '{"type":"user","uuid":"a1"}\n'); + + const exitCode = await new Promise((resolve) => { + const child = spawn(process.execPath, ["--import", "tsx", indexPath, "send"], { + env: { ...process.env, HOME: tmpHome, USERPROFILE: tmpHome }, + stdio: ["pipe", "ignore", "ignore"], + }); + child.stdin!.end( + JSON.stringify({ + session_id: "async-handoff", + transcript_path: transcriptPath, + cwd: projectDir, + }) + ); + child.on("close", (c) => resolve(c ?? -1)); + }); + + // The parent (hook) returns success without waiting for the HTTPS send. + expect(exitCode).toBe(0); + + // The detached worker delivers shortly after, off the critical path. + const deadline = Date.now() + 8000; + while (received.length === 0 && Date.now() < deadline) { + await new Promise((r) => setTimeout(r, 50)); + } + + expect(received).toHaveLength(1); + expect(JSON.parse(received[0]).session_id).toBe("async-handoff"); + }, 30_000); +}); diff --git a/cli/src/send/worker-entry.test.ts b/cli/src/send/worker-entry.test.ts new file mode 100644 index 0000000..53ecce0 --- /dev/null +++ b/cli/src/send/worker-entry.test.ts @@ -0,0 +1,89 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import * as fs from "node:fs"; +import * as path from "node:path"; +import * as os from "node:os"; +import * as http from "node:http"; +import { spawn } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import type { AddressInfo } from "node:net"; + +const indexPath = path.resolve( + path.dirname(fileURLToPath(import.meta.url)), + "../index.ts" +); + +describe("__send-worker entry", () => { + let tmpHome: string; + let projectDir: string; + let server: http.Server; + let baseUrl: string; + let received: string[]; + const sockets = new Set(); + + beforeEach(async () => { + tmpHome = fs.mkdtempSync(path.join(os.tmpdir(), "agentrace-entryhome-")); + projectDir = fs.mkdtempSync(path.join(os.tmpdir(), "agentrace-entryproj-")); + fs.mkdirSync(path.join(tmpHome, ".agentrace"), { recursive: true }); + + received = []; + server = http.createServer((req, res) => { + let body = ""; + req.on("data", (c) => (body += c)); + req.on("end", () => { + received.push(body); + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, events_created: 1 })); + }); + }); + server.on("connection", (s) => { + sockets.add(s); + s.on("close", () => sockets.delete(s)); + }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + baseUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + + fs.writeFileSync( + path.join(tmpHome, ".agentrace", "config.json"), + JSON.stringify({ server_url: baseUrl, api_key: "agtr_test" }) + ); + }); + + afterEach(async () => { + for (const s of sockets) s.destroy(); + sockets.clear(); + await new Promise((r) => server.close(() => r())); + fs.rmSync(tmpHome, { recursive: true, force: true }); + fs.rmSync(projectDir, { recursive: true, force: true }); + }); + + it("delivers the transcript via the worker env contract and exits 0", async () => { + const transcriptPath = path.join(projectDir, "t.jsonl"); + fs.writeFileSync( + transcriptPath, + '{"type":"user","uuid":"e1"}\n{"type":"assistant","uuid":"e2"}\n' + ); + + const code = await new Promise((resolve) => { + const child = spawn( + process.execPath, + ["--import", "tsx", indexPath, "__send-worker"], + { + env: { + ...process.env, + HOME: tmpHome, + USERPROFILE: tmpHome, + AGENTRACE_WORKER_SESSION_ID: "entry-sess", + AGENTRACE_WORKER_TRANSCRIPT_PATH: transcriptPath, + AGENTRACE_WORKER_PROJECT_DIR: projectDir, + }, + stdio: "ignore", + } + ); + child.on("close", (c) => resolve(c ?? -1)); + }); + + expect(code).toBe(0); + expect(received).toHaveLength(1); + expect(JSON.parse(received[0]).transcript_lines).toHaveLength(2); + }, 30_000); +}); diff --git a/cli/src/send/worker.test.ts b/cli/src/send/worker.test.ts new file mode 100644 index 0000000..7e30741 --- /dev/null +++ b/cli/src/send/worker.test.ts @@ -0,0 +1,164 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import * as fs from "node:fs"; +import * as path from "node:path"; +import * as os from "node:os"; +import * as http from "node:http"; +import type { AddressInfo } from "node:net"; + +interface Recorded { + session_id: string; + transcript_lines: Array>; +} + +describe("send/worker", () => { + let tmpHome: string; + let projectDir: string; + let transcriptPath: string; + let prevHome: string | undefined; + let prevUserProfile: string | undefined; + let server: http.Server; + let baseUrl: string; + let received: Recorded[]; + let respondStatus = 200; + const sockets = new Set(); + + // Imported fresh per test so cursor/lock modules bind to the temp HOME. + let runWorker: ( + p: { sessionId: string; transcriptPath: string; projectDir?: string }, + lockOptions?: { pollIntervalMs?: number; maxWaitMs?: number } + ) => Promise; + let getCursor: (sid: string) => number; + let holderDir: (sid: string) => string; + let acquireHolder: (sid: string) => boolean; + let saveCursor: (sid: string, n: number) => void; + let releaseSessionLock: (sid: string) => void; + + function writeLines(lines: Array>): void { + fs.writeFileSync( + transcriptPath, + lines.map((l) => JSON.stringify(l)).join("\n") + "\n" + ); + } + + function lineRange(start: number, end: number): Array> { + const out: Array> = []; + for (let i = start; i < end; i++) { + out.push({ type: "user", uuid: `u${i}`, n: i }); + } + return out; + } + + beforeEach(async () => { + tmpHome = fs.mkdtempSync(path.join(os.tmpdir(), "agentrace-wkhome-")); + projectDir = fs.mkdtempSync(path.join(os.tmpdir(), "agentrace-wkproj-")); + transcriptPath = path.join(projectDir, "transcript.jsonl"); + prevHome = process.env.HOME; + prevUserProfile = process.env.USERPROFILE; + process.env.HOME = tmpHome; + process.env.USERPROFILE = tmpHome; + + received = []; + respondStatus = 200; + server = http.createServer((req, res) => { + let body = ""; + req.on("data", (c) => (body += c)); + req.on("end", () => { + if (respondStatus === 200) { + received.push(JSON.parse(body)); + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, events_created: 1 })); + } else { + res.writeHead(respondStatus); + res.end("error"); + } + }); + }); + server.on("connection", (s) => { + sockets.add(s); + s.on("close", () => sockets.delete(s)); + }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + baseUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + + vi.resetModules(); + const manager = await import("../config/manager.js"); + manager.saveLocalConfig(projectDir, { + server_url: baseUrl, + api_key: "agtr_test", + }); + ({ runWorker } = await import("./worker.js")); + ({ getCursor, saveCursor } = await import("../config/cursor.js")); + ({ holderDir, acquireHolder, releaseSessionLock } = await import("./lock.js")); + }); + + afterEach(async () => { + if (prevHome === undefined) delete process.env.HOME; + else process.env.HOME = prevHome; + if (prevUserProfile === undefined) delete process.env.USERPROFILE; + else process.env.USERPROFILE = prevUserProfile; + for (const s of sockets) s.destroy(); + sockets.clear(); + await new Promise((r) => server.close(() => r())); + fs.rmSync(tmpHome, { recursive: true, force: true }); + fs.rmSync(projectDir, { recursive: true, force: true }); + }); + + it("advances the cursor and releases the lock on a successful send", async () => { + writeLines(lineRange(0, 3)); + + await runWorker({ sessionId: "s1", transcriptPath, projectDir }); + + expect(getCursor("s1")).toBe(3); + expect(received).toHaveLength(1); + expect(received[0].transcript_lines).toHaveLength(3); + expect(fs.existsSync(holderDir("s1"))).toBe(false); + }); + + it("does not advance the cursor on a failed send (next worker can pick it up)", async () => { + respondStatus = 500; + writeLines(lineRange(0, 3)); + + await runWorker({ sessionId: "s1", transcriptPath, projectDir }); + + expect(getCursor("s1")).toBe(0); + expect(fs.existsSync(holderDir("s1"))).toBe(false); + }); + + it("exits without sending when the lock is dropped (holder + waiter both taken)", async () => { + // Occupy holder and waiting slot so the worker is dropped. + acquireHolder("s1"); + const { acquireWaiting } = await import("./lock.js"); + acquireWaiting("s1"); + writeLines(lineRange(0, 3)); + + await runWorker({ sessionId: "s1", transcriptPath, projectDir }); + + expect(received).toHaveLength(0); + expect(getCursor("s1")).toBe(0); + }); + + it("a waiter reads cursor→tail after the holder releases, covering lines added while waiting", async () => { + // Holder is busy; the worker must wait, then read the latest tail. + acquireHolder("s1"); + writeLines(lineRange(0, 2)); + + const waiter = runWorker( + { sessionId: "s1", transcriptPath, projectDir }, + { pollIntervalMs: 10, maxWaitMs: 2000 } + ); + + // Let the worker reach the waiting state, then simulate the holder finishing + // its own send (cursor → 2) and appending two more lines before releasing. + await new Promise((r) => setTimeout(r, 50)); + writeLines(lineRange(0, 4)); + saveCursor("s1", 2); + releaseSessionLock("s1"); + + await waiter; + + expect(getCursor("s1")).toBe(4); + expect(received).toHaveLength(1); + expect(received[0].transcript_lines).toHaveLength(2); // lines 2..4 only + expect((received[0].transcript_lines[0] as { n: number }).n).toBe(2); + }); +}); diff --git a/cli/src/send/worker.ts b/cli/src/send/worker.ts new file mode 100644 index 0000000..d618976 --- /dev/null +++ b/cli/src/send/worker.ts @@ -0,0 +1,47 @@ +import { acquireSessionLock, releaseSessionLock, type AcquireOptions } from "./lock.js"; +import { runSend } from "../commands/send.js"; + +export const WORKER_ENV = { + sessionId: "AGENTRACE_WORKER_SESSION_ID", + transcriptPath: "AGENTRACE_WORKER_TRANSCRIPT_PATH", + projectDir: "AGENTRACE_WORKER_PROJECT_DIR", +} as const; + +export interface WorkerPayload { + sessionId: string; + transcriptPath: string; + projectDir?: string; +} + +export async function runWorker( + payload: WorkerPayload, + lockOptions?: AcquireOptions +): Promise { + const { sessionId, transcriptPath, projectDir } = payload; + + const outcome = await acquireSessionLock(sessionId, lockOptions); + if (outcome === "dropped") { + return; + } + + const release = () => releaseSessionLock(sessionId); + process.once("exit", release); + try { + await runSend({ sessionId, transcriptPath, cwd: projectDir }); + } finally { + process.removeListener("exit", release); + release(); + } +} + +export async function workerMain(): Promise { + const sessionId = process.env[WORKER_ENV.sessionId]; + const transcriptPath = process.env[WORKER_ENV.transcriptPath]; + const projectDir = process.env[WORKER_ENV.projectDir] || undefined; + + if (!sessionId || !transcriptPath) { + return; + } + + await runWorker({ sessionId, transcriptPath, projectDir }); +} diff --git a/cli/src/utils/http.test.ts b/cli/src/utils/http.test.ts new file mode 100644 index 0000000..50ac83b --- /dev/null +++ b/cli/src/utils/http.test.ts @@ -0,0 +1,80 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import * as fs from "node:fs"; +import * as path from "node:path"; +import * as os from "node:os"; +import * as http from "node:http"; +import { sendIngest, SEND_TIMEOUT_MS } from "./http.js"; +import { saveLocalConfig } from "../config/manager.js"; + +describe("utils/http sendIngest", () => { + let tempProjectDir: string; + let server: http.Server; + let baseUrl: string; + const sockets = new Set(); + // Controls whether the test server ever responds. + let hang = false; + + beforeEach(async () => { + tempProjectDir = fs.mkdtempSync(path.join(os.tmpdir(), "agentrace-http-")); + + server = http.createServer((_req, res) => { + if (hang) { + // Never respond — forces the client-side timeout to fire. + return; + } + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ ok: true, events_created: 1 })); + }); + server.on("connection", (socket) => { + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); + }); + + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const addr = server.address(); + if (addr && typeof addr === "object") { + baseUrl = `http://127.0.0.1:${addr.port}`; + } + + saveLocalConfig(tempProjectDir, { + server_url: baseUrl, + api_key: "agtr_test", + }); + }); + + afterEach(async () => { + delete process.env.AGENTRACE_SEND_TIMEOUT_MS; + hang = false; + for (const socket of sockets) socket.destroy(); + sockets.clear(); + await new Promise((resolve) => server.close(() => resolve())); + if (fs.existsSync(tempProjectDir)) { + fs.rmSync(tempProjectDir, { recursive: true }); + } + }); + + it("exposes a tunable default send timeout longer than a request", () => { + expect(SEND_TIMEOUT_MS).toBeGreaterThanOrEqual(30_000); + }); + + it("returns ok on a successful response", async () => { + const result = await sendIngest( + { session_id: "s1", transcript_lines: [{ type: "user" }] }, + tempProjectDir + ); + expect(result.ok).toBe(true); + }); + + it("returns { ok: false } when the request exceeds the send timeout", async () => { + hang = true; + process.env.AGENTRACE_SEND_TIMEOUT_MS = "150"; + + const result = await sendIngest( + { session_id: "s1", transcript_lines: [{ type: "user" }] }, + tempProjectDir + ); + + expect(result.ok).toBe(false); + expect(result.error).toBeTruthy(); + }); +}); diff --git a/cli/src/utils/http.ts b/cli/src/utils/http.ts index 3a3f3ef..de98fa0 100644 --- a/cli/src/utils/http.ts +++ b/cli/src/utils/http.ts @@ -21,6 +21,13 @@ export interface IngestResponse { error?: string; } +export const SEND_TIMEOUT_MS = 30_000; + +function getSendTimeoutMs(): number { + const override = Number(process.env.AGENTRACE_SEND_TIMEOUT_MS); + return Number.isFinite(override) && override > 0 ? override : SEND_TIMEOUT_MS; +} + export interface WebSessionResponse { url: string; expires_at: string; @@ -50,6 +57,7 @@ export async function sendIngest( }, body: JSON.stringify(payload), dispatcher: createDispatcher(projectDir), + signal: AbortSignal.timeout(getSendTimeoutMs()), }); if (!response.ok) {