From 771cb4783cc9fc591b46aec7d29f9a6cf07c14d4 Mon Sep 17 00:00:00 2001 From: y-oga Date: Sat, 6 Jun 2026 19:30:37 +0900 Subject: [PATCH 01/10] =?UTF-8?q?feat:=20config=20=E3=81=AB=20send=5Fmode?= =?UTF-8?q?=20=E3=81=A8=E9=80=81=E4=BF=A1=E3=82=BF=E3=82=A4=E3=83=A0?= =?UTF-8?q?=E3=82=A2=E3=82=A6=E3=83=88=E3=82=92=E8=BF=BD=E5=8A=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 非同期送信モード（send_mode: async）の土台として、config に send_mode フィールドと getSendMode() を追加する。未設定・不正値は "sync" にフォールバックし、既存ユーザーの同期挙動を後方互換で維持する（HC-1）。 sendIngest の fetch に AbortSignal.timeout(SEND_TIMEOUT_MS≈30s) を付与し、 ハングした送信が後続 Phase の per-session ロックを長時間保持するのを防ぐ （HC-6 の前提）。MAX_LOCK_MS(60s) > SEND_TIMEOUT_MS の不等式の土台となる。 タイムアウト/Abort は既存の catch で {ok:false} に正規化される。 Co-Authored-By: Claude Opus 4.8 (1M context) --- cli/src/config/manager.test.ts | 25 +++++++++++ cli/src/config/manager.ts | 12 +++++ cli/src/utils/http.test.ts | 80 ++++++++++++++++++++++++++++++++++ cli/src/utils/http.ts | 16 +++++++ 4 files changed, 133 insertions(+) create mode 100644 cli/src/utils/http.test.ts diff --git a/cli/src/config/manager.test.ts b/cli/src/config/manager.test.ts index 8e6f226..9649411 100644 --- a/cli/src/config/manager.test.ts +++ b/cli/src/config/manager.test.ts @@ -11,6 +11,7 @@ import { loadConfigWithFallback, findLocalConfigPath, findAndLoadLocalConfig, + getSendMode, type AgentraceConfig, } from "./manager.js"; @@ -20,6 +21,30 @@ 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("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..a43468c 100644 --- a/cli/src/config/manager.ts +++ b/cli/src/config/manager.ts @@ -2,10 +2,22 @@ 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; +} + +/** + * Resolve the effective send mode from a config. + * Defaults to "sync" when unset, null, or set to any unrecognized value + * (opt-in / backward-compatible default — HC-1). + */ +export function getSendMode(config: AgentraceConfig | null | undefined): SendMode { + return config?.send_mode === "async" ? "async" : "sync"; } const CONFIG_DIR = path.join(os.homedir(), ".agentrace"); 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..8b89d44 100644 --- a/cli/src/utils/http.ts +++ b/cli/src/utils/http.ts @@ -21,6 +21,19 @@ export interface IngestResponse { error?: string; } +/** + * Default send timeout for ingest requests (~30s, tunable). + * Bounds how long a worker can hold a per-session lock while a send hangs, + * which is a precondition for self-recovering liveness (HC-6). + * Must stay below MAX_LOCK_MS in send/lock.ts. + */ +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 +63,9 @@ export async function sendIngest( }, body: JSON.stringify(payload), dispatcher: createDispatcher(projectDir), + // Bound the request so a hung send cannot hold a session lock forever. + // Timeout/abort throws and is normalized to { ok: false } by the catch below. + signal: AbortSignal.timeout(getSendTimeoutMs()), }); if (!response.ok) { From 012f69f4c3d30df825a567e2ec68f91235eca8b0 Mon Sep 17 00:00:00 2001 From: y-oga Date: Sat, 6 Jun 2026 20:14:42 +0900 Subject: [PATCH 02/10] =?UTF-8?q?style:=20send=5Fmode/=E9=80=81=E4=BF=A1?= =?UTF-8?q?=E3=82=BF=E3=82=A4=E3=83=A0=E3=82=A2=E3=82=A6=E3=83=88=E5=91=A8?= =?UTF-8?q?=E3=82=8A=E3=81=AE=E9=81=8E=E5=89=B0=E3=81=AA=E3=82=B3=E3=83=A1?= =?UTF-8?q?=E3=83=B3=E3=83=88=E3=82=92=E5=89=8A=E6=B8=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 既存実装（cursor.ts 等）のコメント密度に合わせ、設計書を読まないと わからない情報やコードから自明な説明をコメントから外す。 Co-Authored-By: Claude Opus 4.8 (1M context) --- cli/src/config/manager.ts | 5 ----- cli/src/utils/http.ts | 8 -------- 2 files changed, 13 deletions(-) diff --git a/cli/src/config/manager.ts b/cli/src/config/manager.ts index a43468c..c553672 100644 --- a/cli/src/config/manager.ts +++ b/cli/src/config/manager.ts @@ -11,11 +11,6 @@ export interface AgentraceConfig { send_mode?: SendMode; } -/** - * Resolve the effective send mode from a config. - * Defaults to "sync" when unset, null, or set to any unrecognized value - * (opt-in / backward-compatible default — HC-1). - */ export function getSendMode(config: AgentraceConfig | null | undefined): SendMode { return config?.send_mode === "async" ? "async" : "sync"; } diff --git a/cli/src/utils/http.ts b/cli/src/utils/http.ts index 8b89d44..de98fa0 100644 --- a/cli/src/utils/http.ts +++ b/cli/src/utils/http.ts @@ -21,12 +21,6 @@ export interface IngestResponse { error?: string; } -/** - * Default send timeout for ingest requests (~30s, tunable). - * Bounds how long a worker can hold a per-session lock while a send hangs, - * which is a precondition for self-recovering liveness (HC-6). - * Must stay below MAX_LOCK_MS in send/lock.ts. - */ export const SEND_TIMEOUT_MS = 30_000; function getSendTimeoutMs(): number { @@ -63,8 +57,6 @@ export async function sendIngest( }, body: JSON.stringify(payload), dispatcher: createDispatcher(projectDir), - // Bound the request so a hung send cannot hold a session lock forever. - // Timeout/abort throws and is normalized to { ok: false } by the catch below. signal: AbortSignal.timeout(getSendTimeoutMs()), }); From a34ba28423dc35cd6f0df44cd632af029d95f1eb Mon Sep 17 00:00:00 2001 From: y-oga Date: Sat, 6 Jun 2026 20:14:43 +0900 Subject: [PATCH 03/10] =?UTF-8?q?feat:=20per-session=20=E9=80=81=E4=BF=A1?= =?UTF-8?q?=E3=83=AD=E3=83=83=E3=82=AF=E3=82=92=E8=BF=BD=E5=8A=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 非同期送信で同一セッションの送信を直列化するため、ディレクトリ存在を ロック実体とする per-session ロックを実装する。保持1＋待機1 の上限で worker の積み上がりを防ぎ、超過分は待機者の cursor→末尾読込が回収する。 ロックは pid 死亡（即時）または経過時間（MAX_LOCK_MS=60s）で stale と 判定して自己回復する。MAX_LOCK_MS は送信タイムアウト(30s)より大きく取り、 送信中の生存 holder を誤って奪わない。 stale 奪取は取得ごとに一意な instanceId を持たせ、退避直前に同一インスタンス かを再検証することで、別 worker が再取得した新 holder を誤って奪う二重保持を 防ぐ。再検証と rename の間に残る極小窓で万一二重保持が起きても、欠落は カーソル＋サーバ冪等が防ぐ（送信2本に留まりデッドロックしない）。 Co-Authored-By: Claude Opus 4.8 (1M context) --- cli/src/send/lock.test.ts | 261 ++++++++++++++++++++++++++++++++++++++ cli/src/send/lock.ts | 172 +++++++++++++++++++++++++ 2 files changed, 433 insertions(+) create mode 100644 cli/src/send/lock.test.ts create mode 100644 cli/src/send/lock.ts diff --git a/cli/src/send/lock.test.ts b/cli/src/send/lock.test.ts new file mode 100644 index 0000000..a8873c2 --- /dev/null +++ b/cli/src/send/lock.test.ts @@ -0,0 +1,261 @@ +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 } 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("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)); + const racerPath = path.join(here, "__lock_racer__.ts"); + // Each racer reports its result, then stays alive holding the lock so a + // winner's pid does not die and let the next racer reclaim it as stale. + // The unique-holder guarantee then rests purely on atomic mkdir. + fs.writeFileSync( + racerPath, + `import { acquireHolder } from "./lock.js";\n` + + `const got = acquireHolder(process.env.SID!);\n` + + `process.stdout.write(got ? "ACQUIRED" : "NOPE");\n` + + `setTimeout(() => process.exit(0), 1500);\n` + ); + + try { + const N = 5; + 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 }, stdio: ["ignore", "pipe", "ignore"] } + ); + let out = ""; + child.stdout.on("data", (c) => (out += c.toString())); + 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..6234622 --- /dev/null +++ b/cli/src/send/lock.ts @@ -0,0 +1,172 @@ +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)); +} + +export function releaseSessionLock(sid: string): void { + fs.rmSync(holderDir(sid), { recursive: true, force: true }); +} + +export function releaseWaiting(sid: string): void { + fs.rmSync(waitingDir(sid), { recursive: true, force: true }); +} + +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); + } +} From 733bd2d9f482669cd33b5d1ec73473435d52d7d6 Mon Sep 17 00:00:00 2001 From: y-oga Date: Sat, 6 Jun 2026 20:27:04 +0900 Subject: [PATCH 04/10] =?UTF-8?q?feat:=20=E3=83=AD=E3=83=83=E3=82=AF?= =?UTF-8?q?=E4=BF=9D=E6=8C=81=E5=86=85=E3=81=A7=E9=80=81=E4=BF=A1=E3=81=99?= =?UTF-8?q?=E3=82=8B=20worker=20=E3=82=92=E8=BF=BD=E5=8A=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 非同期送信の本体を追加する。send.ts の送信ロジックを process.exit に 依存しない純送信部 runSend に分離し、sync(hook)/manual は wrapper が 従来どおりの exit code とログを担う（後方互換を維持）。 worker はロック取得後に read→送信→saveCursor を行い、解放は saveCursor 後の finally で実施する（read はロック取得後・解放は cursor 前進後、の 順序不変条件を担保）。ロックが取れなければ送信せず終了し、待機者の cursor→末尾読込が取りこぼし分を回収する。 ロック解放は自分が保持している場合のみ削除するよう変更した。送信が長時間 ハングして別プロセスにロックを奪取・再取得された後でも、新しい保持者の ロックを誤って削除しない。 Co-Authored-By: Claude Opus 4.8 (1M context) --- cli/src/commands/send.ts | 108 +++++++++++++++--------- cli/src/send/lock.test.ts | 14 +++ cli/src/send/lock.ts | 13 ++- cli/src/send/worker.test.ts | 164 ++++++++++++++++++++++++++++++++++++ cli/src/send/worker.ts | 41 +++++++++ 5 files changed, 299 insertions(+), 41 deletions(-) create mode 100644 cli/src/send/worker.test.ts create mode 100644 cli/src/send/worker.ts diff --git a/cli/src/commands/send.ts b/cli/src/commands/send.ts index 3b0bf0b..a95821e 100644 --- a/cli/src/commands/send.ts +++ b/cli/src/commands/send.ts @@ -21,6 +21,19 @@ 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"]; @@ -51,34 +64,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; +export async function runSend(params: RunSendParams): Promise { + const { sessionId, transcriptPath, cwd } = params; - const exitWithError = (message: string) => { - console.error(message); - process.exit(isHook ? 0 : 1); - }; - - // 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 +86,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 +96,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 +131,6 @@ async function sendTranscript(params: SendTranscriptParams): Promise { gitBranch = getGitBranch(cwd) ?? undefined; } - // Send to server const result = await sendIngest( { session_id: sessionId, @@ -143,7 +138,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 +146,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); } /** diff --git a/cli/src/send/lock.test.ts b/cli/src/send/lock.test.ts index a8873c2..f7ac29d 100644 --- a/cli/src/send/lock.test.ts +++ b/cli/src/send/lock.test.ts @@ -141,6 +141,20 @@ describe("send/lock", () => { }); }); + 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"); diff --git a/cli/src/send/lock.ts b/cli/src/send/lock.ts index 6234622..31feae4 100644 --- a/cli/src/send/lock.ts +++ b/cli/src/send/lock.ts @@ -125,12 +125,21 @@ 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 { - fs.rmSync(holderDir(sid), { recursive: true, force: true }); + removeIfOwned(holderDir(sid)); } export function releaseWaiting(sid: string): void { - fs.rmSync(waitingDir(sid), { recursive: true, force: true }); + removeIfOwned(waitingDir(sid)); } export type AcquireOutcome = "acquired" | "dropped"; 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..66de22d --- /dev/null +++ b/cli/src/send/worker.ts @@ -0,0 +1,41 @@ +import { acquireSessionLock, releaseSessionLock, type AcquireOptions } from "./lock.js"; +import { runSend } from "../commands/send.js"; + +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.AGENTRACE_WORKER_SESSION_ID; + const transcriptPath = process.env.AGENTRACE_WORKER_TRANSCRIPT_PATH; + const projectDir = process.env.AGENTRACE_WORKER_PROJECT_DIR || undefined; + + if (!sessionId || !transcriptPath) { + return; + } + + await runWorker({ sessionId, transcriptPath, projectDir }); +} From 201cf08eb275e2924f02111cb68c0d72f4551a47 Mon Sep 17 00:00:00 2001 From: y-oga Date: Sat, 6 Jun 2026 20:37:54 +0900 Subject: [PATCH 05/10] =?UTF-8?q?feat:=20async=20=E3=83=A2=E3=83=BC?= =?UTF-8?q?=E3=83=89=E3=81=A7=20hook=20=E3=82=92=20detached=20worker=20?= =?UTF-8?q?=E3=81=AB=20handoff=20=E3=81=99=E3=82=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit send_mode が async のとき、hook 経由の send は payload を env に載せた detached worker（隠し __send-worker サブコマンド）を spawn して即 exit する。 HTTPS 送信を hook の critical path から外し、Claude Code の応答性を 外部サーバのレイテンシから切り離す。 worker 起動は process.execPath + process.execArgv + process.argv[1] で 構成し、dev（tsx loader を execArgv が保持）/ 本番（dist の素の node）を 分岐なく解決する。env キーは WORKER_ENV 定数で spawn 側と read 側を共有する。 async では UserPromptSubmit の 10 秒待機を行わない（未書き込み分は次の 発火がカーソルから拾う）。sync・手動送信は従来どおり同期送信のまま。 Co-Authored-By: Claude Opus 4.8 (1M context) --- cli/src/commands/send.ts | 41 +++++++++++-- cli/src/index.ts | 8 +++ cli/src/send/send-async-e2e.test.ts | 94 +++++++++++++++++++++++++++++ cli/src/send/worker-entry.test.ts | 89 +++++++++++++++++++++++++++ cli/src/send/worker.ts | 12 +++- 5 files changed, 236 insertions(+), 8 deletions(-) create mode 100644 cli/src/send/send-async-e2e.test.ts create mode 100644 cli/src/send/worker-entry.test.ts diff --git a/cli/src/commands/send.ts b/cli/src/commands/send.ts index a95821e..510a39e 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, @@ -234,15 +235,23 @@ 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. + if (getSendMode(loadConfigWithFallback(projectDir)) === "async") { + spawnWorker({ sessionId, transcriptPath, projectDir }); + process.exit(0); + } + // 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, @@ -251,6 +260,28 @@ 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.unref(); +} + /** * Manual send command. * Finds session file by ID and sends to server. diff --git a/cli/src/index.ts b/cli/src/index.ts index 00d1e73..1ea0d1b 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(); @@ -90,4 +91,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/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.ts b/cli/src/send/worker.ts index 66de22d..d618976 100644 --- a/cli/src/send/worker.ts +++ b/cli/src/send/worker.ts @@ -1,6 +1,12 @@ 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; @@ -29,9 +35,9 @@ export async function runWorker( } export async function workerMain(): Promise { - const sessionId = process.env.AGENTRACE_WORKER_SESSION_ID; - const transcriptPath = process.env.AGENTRACE_WORKER_TRANSCRIPT_PATH; - const projectDir = process.env.AGENTRACE_WORKER_PROJECT_DIR || undefined; + 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; From f02382500f04d8e4951e9c1fae4d6ae723ce2ff3 Mon Sep 17 00:00:00 2001 From: y-oga Date: Sat, 6 Jun 2026 21:40:56 +0900 Subject: [PATCH 06/10] =?UTF-8?q?fix:=20worker=20spawn=20=E5=A4=B1?= =?UTF-8?q?=E6=95=97=E3=81=A7=20hook=20=E3=82=92=E8=90=BD=E3=81=A8?= =?UTF-8?q?=E3=81=95=E3=81=9A=E3=80=81=E4=B8=A6=E8=A1=8C=E3=83=AD=E3=83=83?= =?UTF-8?q?=E3=82=AF=E3=83=86=E3=82=B9=E3=83=88=E3=82=92=E6=B1=BA=E5=AE=9A?= =?UTF-8?q?=E8=AB=96=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit async 送信で spawn が失敗（同期 throw / 非同期 error）しても hook が クラッシュしないよう、try/catch で sync 送信へフォールバックし、子プロセスに error ハンドラを付ける。fork 資源枯渇などの異常時でも hook 契約（落とさない・ バッチを捨てない）を守る。 並行 stale 奪取テストは固定 sleep で待機者の生存を仮定していたため、高負荷で spawn 間隔が広がると勝者が先に exit して別プロセスが stale 再取得し稀に複数 勝者になっていた。sentinel ファイルで全 racer の取得試行が出揃うまで全員を 生存させ、holder 一意性を atomic mkdir のみに依存させて決定論化した。 Co-Authored-By: Claude Opus 4.8 (1M context) --- cli/src/commands/send.ts | 12 +++++++++--- cli/src/send/lock.test.ts | 30 ++++++++++++++++++++++++------ 2 files changed, 33 insertions(+), 9 deletions(-) diff --git a/cli/src/commands/send.ts b/cli/src/commands/send.ts index 510a39e..fefd017 100644 --- a/cli/src/commands/send.ts +++ b/cli/src/commands/send.ts @@ -240,10 +240,15 @@ export async function sendCommand(): Promise { // 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. + // is sync-only. If the worker cannot be spawned, fall back to a sync send so + // the hook neither crashes nor drops the batch. if (getSendMode(loadConfigWithFallback(projectDir)) === "async") { - spawnWorker({ sessionId, transcriptPath, projectDir }); - process.exit(0); + try { + spawnWorker({ sessionId, transcriptPath, projectDir }); + process.exit(0); + } catch { + // fall through to the synchronous send below + } } // For UserPromptSubmit, wait for transcript to be written @@ -279,6 +284,7 @@ function spawnWorker(payload: { }, } ); + child.on("error", () => {}); child.unref(); } diff --git a/cli/src/send/lock.test.ts b/cli/src/send/lock.test.ts index f7ac29d..6a1e332 100644 --- a/cli/src/send/lock.test.ts +++ b/cli/src/send/lock.test.ts @@ -237,19 +237,27 @@ describe("send/lock", () => { const here = path.dirname(fileURLToPath(import.meta.url)); const racerPath = path.join(here, "__lock_racer__.ts"); - // Each racer reports its result, then stays alive holding the lock so a - // winner's pid does not die and let the next racer reclaim it as stale. - // The unique-holder guarantee then rests purely on atomic mkdir. + 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 "./lock.js";\n` + + `import * as fs from "node:fs";\n` + `const got = acquireHolder(process.env.SID!);\n` + `process.stdout.write(got ? "ACQUIRED" : "NOPE");\n` + - `setTimeout(() => process.exit(0), 1500);\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 @@ -257,10 +265,20 @@ describe("send/lock", () => { const child = spawn( process.execPath, ["--import", "tsx", racerPath], - { env: { ...process.env, SID }, stdio: ["ignore", "pipe", "ignore"] } + { + env: { ...process.env, SID, SENTINEL: sentinel }, + stdio: ["ignore", "pipe", "ignore"], + } ); let out = ""; - child.stdout.on("data", (c) => (out += c.toString())); + 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())); }) ); From 08b05c7f889a2533b4686ad83055d616a39fddf1 Mon Sep 17 00:00:00 2001 From: y-oga Date: Sat, 6 Jun 2026 21:40:56 +0900 Subject: [PATCH 07/10] =?UTF-8?q?feat:=20--async=20/=20doctor=20=E3=81=A7?= =?UTF-8?q?=20send=5Fmode=20=E3=81=AE=20opt-in=20=E5=B0=8E=E7=B7=9A?= =?UTF-8?q?=E3=82=92=E8=BF=BD=E5=8A=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit init / on に --async を追加し send_mode を config に保存、doctor で現在の send_mode を表示する。送信モードの切替を CLI から行えるようにする。 send_mode の保存は persistSendMode で実効 config（tree 内の local を優先、 無ければ global）に書き込む。これは送信時に loadConfigWithFallback が読む ファイルと一致するため、project-local 設定のみの環境でも切替が確実に効く。 cli/CLAUDE.md に send_mode（sync/async）と --async 導線の説明を追記。 Co-Authored-By: Claude Opus 4.8 (1M context) --- cli/CLAUDE.md | 14 ++++++++++- cli/src/commands/doctor.ts | 2 ++ cli/src/commands/init.ts | 5 ++++ cli/src/commands/on.ts | 12 ++++++++- cli/src/config/manager.test.ts | 45 ++++++++++++++++++++++++++++++++++ cli/src/config/manager.ts | 24 ++++++++++++++++++ cli/src/index.ts | 9 ++++--- 7 files changed, 106 insertions(+), 5 deletions(-) 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/config/manager.test.ts b/cli/src/config/manager.test.ts index 9649411..7ea3245 100644 --- a/cli/src/config/manager.test.ts +++ b/cli/src/config/manager.test.ts @@ -12,6 +12,7 @@ import { findLocalConfigPath, findAndLoadLocalConfig, getSendMode, + persistSendMode, type AgentraceConfig, } from "./manager.js"; @@ -45,6 +46,50 @@ describe("config/manager", () => { }); }); + 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 c553672..68258de 100644 --- a/cli/src/config/manager.ts +++ b/cli/src/config/manager.ts @@ -15,6 +15,30 @@ export function getSendMode(config: AgentraceConfig | null | undefined): SendMod 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"); const CONFIG_FILE = path.join(CONFIG_DIR, "config.json"); diff --git a/cli/src/index.ts b/cli/src/index.ts index 1ea0d1b..098eecf 100644 --- a/cli/src/index.ts +++ b/cli/src/index.ts @@ -23,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, }); }); @@ -65,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 From 3072674b3eef2fea7066b4c9bfe6f9150b7397f7 Mon Sep 17 00:00:00 2001 From: y-oga Date: Mon, 15 Jun 2026 12:10:53 +0900 Subject: [PATCH 08/10] =?UTF-8?q?docs(cli):=20async=E9=80=81=E4=BF=A1?= =?UTF-8?q?=E3=81=AEspawn=E5=A4=B1=E6=95=97=E3=82=B3=E3=83=A1=E3=83=B3?= =?UTF-8?q?=E3=83=88=E3=82=92=E5=AE=9F=E6=8C=99=E5=8B=95=E3=81=AB=E4=BF=AE?= =?UTF-8?q?=E6=AD=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit spawn() の launch 失敗は throw ではなく非同期の 'error' イベントで 報告されるため、try/catch では捕捉されず「同期フォールバックする」 というコメントは誤り。実際は cursor が HTTP 200 でのみ進むため、 worker が起動失敗しても次回発火でリトライされ無害である旨に修正。 Co-Authored-By: Claude Opus 4.8 (1M context) --- cli/src/commands/send.ts | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/cli/src/commands/send.ts b/cli/src/commands/send.ts index fefd017..92a16c0 100644 --- a/cli/src/commands/send.ts +++ b/cli/src/commands/send.ts @@ -239,9 +239,11 @@ export async function sendCommand(): Promise { 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. If the worker cannot be spawned, fall back to a sync send so - // the hook neither crashes nor drops the batch. + // 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 }); From a470b1d2aa5203821c4ade38f78101b8d9129035 Mon Sep 17 00:00:00 2001 From: y-oga Date: Mon, 15 Jun 2026 12:11:27 +0900 Subject: [PATCH 09/10] =?UTF-8?q?fix(cli):=20git=E5=8F=82=E7=85=A7?= =?UTF-8?q?=E3=81=AEexecSync=E3=81=ABtimeout=E3=82=92=E8=BF=BD=E5=8A=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 初回送信時の `git remote get-url` / `git branch --show-current` は timeout なしで実行しており、git がハング（stuck .git/index.lock 等） すると holder ロックを MAX_LOCK_MS 超で保持し、後続発火が stale 判定で 二重 holder 化する恐れがあった。5s の timeout を付与。超過時は既存の catch→null に流れ、git 情報なし（非 git リポジトリと同等）として送信する。 Co-Authored-By: Claude Opus 4.8 (1M context) --- cli/src/commands/send.ts | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/cli/src/commands/send.ts b/cli/src/commands/send.ts index 92a16c0..ab43f17 100644 --- a/cli/src/commands/send.ts +++ b/cli/src/commands/send.ts @@ -38,16 +38,22 @@ export type SendOutcome = // 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 } } @@ -57,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 { From 74cf0650ce39500d2ff3a69f80a60fa3f91098b4 Mon Sep 17 00:00:00 2001 From: y-oga Date: Mon, 15 Jun 2026 12:12:34 +0900 Subject: [PATCH 10/10] =?UTF-8?q?test(cli):=20racer=E3=83=95=E3=82=A1?= =?UTF-8?q?=E3=82=A4=E3=83=AB=E3=82=92src=E5=A4=96=E3=81=AEtmpdir=E3=81=AB?= =?UTF-8?q?=E7=94=9F=E6=88=90=E3=81=99=E3=82=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 並行ロックテストが生成する __lock_racer__.ts を cli/src/send/ 配下に 書き込んでいた（gitignore 対象外で、SIGKILL 時に残留・read-only チェックアウトで失敗しうる）。スイート他箇所と同様 tmpHome に出力し、 lock.ts を絶対 file URL で import するよう変更。 Co-Authored-By: Claude Opus 4.8 (1M context) --- cli/src/send/lock.test.ts | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/cli/src/send/lock.test.ts b/cli/src/send/lock.test.ts index 6a1e332..6a9fafa 100644 --- a/cli/src/send/lock.test.ts +++ b/cli/src/send/lock.test.ts @@ -3,7 +3,7 @@ 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 } from "node:url"; +import { fileURLToPath, pathToFileURL } from "node:url"; import { acquireHolder, acquireSessionLock, @@ -236,7 +236,10 @@ describe("send/lock", () => { writeMeta(holderDir(SID), { pid: DEAD_PID, startedAt: Date.now() }); const here = path.dirname(fileURLToPath(import.meta.url)); - const racerPath = path.join(here, "__lock_racer__.ts"); + // 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, @@ -246,7 +249,7 @@ describe("send/lock", () => { // process scheduling. fs.writeFileSync( racerPath, - `import { acquireHolder } from "./lock.js";\n` + + `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` +