diff --git a/packages/coding-agent/src/modes/components/status-line/component.ts b/packages/coding-agent/src/modes/components/status-line/component.ts index ae4a649c2..a0c8687bf 100644 --- a/packages/coding-agent/src/modes/components/status-line/component.ts +++ b/packages/coding-agent/src/modes/components/status-line/component.ts @@ -28,6 +28,7 @@ import type { } from "./types"; const JJ_REFRESH_TTL_MS = 5000; +const WATCHER_FAILURE_POLL_TTL_MS = 5000; // ═══════════════════════════════════════════════════════════════════════════ // Context-usage memo @@ -180,6 +181,11 @@ interface BranchResolveRequest { controller: AbortController; } +interface JjResolveRequest { + id: number; + controller: AbortController; +} + interface WorktreeContext { /** Primary-checkout (project) name shown by the path segment. */ projectName: string; @@ -267,8 +273,19 @@ export class StatusLineComponent implements Component { // it (a newer resolve superseded it), so its result is stale and must be // dropped rather than overwrite the value the newer resolve committed. // Mirrors #jjCacheGeneration / #getJjBranch in this file. + // Timestamp of the latest branch read; only bounds cache freshness when the + // HEAD watcher could not be installed. + #branchLastFetch: number | undefined = undefined; + // Bumped on every branch-cache reset (invalidateGitCaches — a HEAD move or + // repo-context change). An in-flight reftable resolve captures this at + // launch; a mismatch on resolve means the cache was invalidated underneath + // it (a newer resolve superseded it), so its result is stale and must be + // dropped rather than overwrite the value the newer resolve committed. + // Mirrors #jjCacheGeneration / #getJjBranch in this file. #branchCacheGeneration = 0; #gitWatcher: fs.FSWatcher | null = null; + #gitWatcherErrorListener: (() => void) | undefined = undefined; + #gitWatcherUnavailable = false; #onBranchChange: (() => void) | null = null; #disposed = false; #autoCompactEnabled: boolean = true; @@ -317,10 +334,11 @@ export class StatusLineComponent implements Component { #jjRootCwd: string | undefined = undefined; #cachedJjBranch: string | null = null; #jjBranchLastFetch = 0; - #jjBranchInFlight = false; + #jjResolveSeq = 0; + #jjBranchActive: JjResolveRequest | undefined = undefined; #cachedJjStatus: { staged: number; unstaged: number; untracked: number } | null = null; #jjStatusLastFetch = 0; - #jjStatusInFlight = false; + #jjStatusActive: JjResolveRequest | undefined = undefined; // Bumped on every jj-cache reset — a cwd switch (#jjRootFor) or a HEAD / // bookmark move (#invalidateGitCaches). An in-flight jj query captures this // at launch; a mismatch on resolve means the caches were reset underneath it @@ -580,47 +598,68 @@ export class StatusLineComponent implements Component { } #setupGitWatcher(): void { - if (this.#gitWatcher) { - this.#gitWatcher.close(); - this.#gitWatcher = null; - } + this.#retireGitWatcher(); + this.#gitWatcherUnavailable = false; if (!this.#gitEnabled() || !this.#hasGitBackedSegment()) { - this.#invalidateGitCaches(); + this.invalidateGitCaches(); return; } const { effectiveGitCwd } = this.#resolveActiveRepoCache(); const repository = git.repo.resolveSync(effectiveGitCwd); - if (!repository) return; + if (!repository) { + // There is no path to watch yet. Cache the negative result only for the + // fallback poll interval so a later `git init` becomes visible without + // generic invalidations or a render-path probe on every paint. + this.#gitWatcherUnavailable = true; + return; + } const watchPath = git.repo.isReftableSync(repository) ? path.join(repository.gitDir, "reftable") : repository.headPath; try { - this.#gitWatcher = fs.watch(watchPath, () => { - if (this.#disposed) return; - this.#invalidateGitCaches(); - if (this.#onBranchChange) { - this.#onBranchChange(); - } + const watcher = fs.watch(watchPath, () => { + if (this.#disposed || this.#gitWatcher !== watcher) return; + this.invalidateGitCaches(); + this.#onBranchChange?.(); }); + const onError = () => { + if (this.#gitWatcher !== watcher) return; + this.#retireGitWatcher(); + this.#gitWatcherUnavailable = true; + if (this.#disposed) return; + this.invalidateGitCaches(); + this.#onBranchChange?.(); + }; + this.#gitWatcher = watcher; + this.#gitWatcherErrorListener = onError; + watcher.on("error", onError); } catch { - this.#invalidateGitCaches(); + this.#gitWatcherUnavailable = true; } } + #retireGitWatcher(): void { + const watcher = this.#gitWatcher; + const onError = this.#gitWatcherErrorListener; + this.#gitWatcher = null; + this.#gitWatcherErrorListener = undefined; + if (!watcher) return; + if (onError) watcher.off("error", onError); + watcher.close(); + } + dispose(): void { this.#disposed = true; this.#branchResolveActive?.controller.abort(); this.#branchResolveActive = undefined; + this.#resetJjRequests(); this.#onBranchChange = null; this.#clearUsageStartTimer(); - if (this.#gitWatcher) { - this.#gitWatcher.close(); - this.#gitWatcher = null; - } + this.#retireGitWatcher(); } #clearUsageStartTimer(): void { @@ -630,7 +669,14 @@ export class StatusLineComponent implements Component { } invalidate(): void { - this.#invalidateGitCaches(); + // Generic repaint invalidation (theme change, message event, model + // switch, …). Must NOT abort or restart a live reftable HEAD/PR resolve: + // the render path self-invalidates via cwd/context cache-miss checks, so + // a generic paint only needs to re-render — not tear down in-flight VCS + // work. Aborting here would fan out a new git subprocess on every agent + // event, re-introducing the render-path spawn churn the async resolve + // was designed to avoid. Explicit Git/repository invalidation (watcher + // HEAD-move, cwd/repo switch) goes through {@link invalidateGitCaches}. } #invalidateSessionCaches(): void { this.#clearUsageStartTimer(); @@ -642,7 +688,15 @@ export class StatusLineComponent implements Component { this.#lastTokensPerSecondTimestamp = null; } - #invalidateGitCaches(): void { + /** + * Explicit Git/repository cache invalidation. Aborts any in-flight + * reftable HEAD/PR resolve, bumps the stale-result generation, and drops + * the branch/PR/jj caches so the next render refetches from disk. Called + * by the git watcher on a HEAD move and by {@link applyCwdChange} on a + * repo/cwd switch. Generic repaints use {@link invalidate} instead and + * must never reach this path. + */ + invalidateGitCaches(): void { this.#cachedBranch = undefined; this.#cachedBranchRepoId = undefined; this.#cachedBranchCwd = undefined; @@ -650,11 +704,13 @@ export class StatusLineComponent implements Component { // repeated invalidations to fan out still-running git subprocesses. this.#branchResolveActive?.controller.abort(); this.#branchResolveActive = undefined; + this.#branchLastFetch = undefined; this.#branchCacheGeneration++; this.#cachedPrContext = undefined; // jj label/status share the git segment's lifecycle: a HEAD move (e.g. a // colocated `jj new`/bookmark move) must drop the throttled jj caches too, // mirroring #jjRootFor's per-cwd reset so the next render refetches. + this.#resetJjRequests(); this.#jjRoot = undefined; this.#jjRootCwd = undefined; this.#cachedJjBranch = null; @@ -663,11 +719,38 @@ export class StatusLineComponent implements Component { this.#jjStatusLastFetch = 0; this.#jjCacheGeneration++; } + + /** + * Re-point the status line's VCS watcher and caches at a new cwd/repository. + * Atomically retires the old watcher/listeners, invalidates VCS caches and + * in-flight controllers, then runs watcher setup for the new cwd and requests + * a repaint. Called by {@link InteractiveMode.applyCwdChange} after the + * SessionManager's cwd has moved — the watcher ownership always follows the + * effective cwd/repo, so a stale watcher for the previous repo can never + * invalidate the new one. Generic repaints use {@link invalidate} and must + * never retire the watcher or abort a live resolve. + */ + applyCwdChange(): void { + this.#retireGitWatcher(); + this.invalidateGitCaches(); + this.#setupGitWatcher(); + this.#onBranchChange?.(); + } + + #resetJjRequests(): void { + this.#jjBranchActive?.controller.abort(); + this.#jjBranchActive = undefined; + this.#jjStatusActive?.controller.abort(); + this.#jjStatusActive = undefined; + } #getCurrentBranch(effectiveGitCwd?: string): string | null { if (!this.#gitEnabled()) return null; const gitCwd = effectiveGitCwd ?? this.#resolveActiveRepoCache().effectiveGitCwd; - if (this.#cachedBranch !== undefined && this.#cachedBranchCwd === gitCwd) { + const fallbackCacheExpired = + this.#gitWatcherUnavailable && + (this.#branchLastFetch === undefined || Date.now() - this.#branchLastFetch >= WATCHER_FAILURE_POLL_TTL_MS); + if (this.#cachedBranch !== undefined && this.#cachedBranchCwd === gitCwd && !fallbackCacheExpired) { return this.#cachedBranch; } @@ -689,7 +772,7 @@ export class StatusLineComponent implements Component { controller: new AbortController(), }; this.#branchResolveActive = request; - // Capture the cache generation at launch. #invalidateGitCaches bumps it + // Capture the cache generation at launch. invalidateGitCaches bumps it // on a HEAD move and clears the in-flight slot, so a fresher resolve can // start while this one is still pending. Without a generation check the // older resolve would finish later, install its stale HEAD, and clear the @@ -723,6 +806,7 @@ export class StatusLineComponent implements Component { this.#cachedBranchCwd = gitCwd; this.#cachedBranchRepoId = repoId; this.#cachedBranch = next; + this.#branchLastFetch = Date.now(); if (prev !== next && this.#onBranchChange) this.#onBranchChange(); })(); return this.#cachedBranchCwd === gitCwd ? (this.#cachedBranch ?? null) : null; @@ -733,6 +817,7 @@ export class StatusLineComponent implements Component { const gitHeadPath = head?.headPath ?? null; this.#cachedBranchCwd = gitCwd; this.#cachedBranchRepoId = gitHeadPath; + this.#branchLastFetch = Date.now(); if (!head) { this.#cachedBranch = null; return null; @@ -822,17 +907,24 @@ export class StatusLineComponent implements Component { const cwd = effectiveGitCwd ?? this.#resolveActiveRepoCache().effectiveGitCwd; const root = this.#jjRootFor(cwd); if (!root) return null; - if (this.#jjBranchInFlight || Date.now() - this.#jjBranchLastFetch < JJ_REFRESH_TTL_MS) { + if (this.#jjBranchActive || Date.now() - this.#jjBranchLastFetch < JJ_REFRESH_TTL_MS) { return this.#cachedJjBranch; } - this.#jjBranchInFlight = true; + const request: JjResolveRequest = { + id: ++this.#jjResolveSeq, + controller: new AbortController(), + }; + this.#jjBranchActive = request; const generation = this.#jjCacheGeneration; (async () => { let next: string | null = null; try { - next = await jj.workingCopy.label(root); + next = await jj.workingCopy.label(root, { + signal: request.controller.signal, + timeoutMs: jj.JJ_COMMAND_TIMEOUT_MS, + }); } finally { - this.#jjBranchInFlight = false; + if (this.#jjBranchActive?.id === request.id) this.#jjBranchActive = undefined; // Advance the throttle only if no reset raced this query; a reset // leaves LastFetch at 0 so the current root refetches instead of // being throttled on a superseded result. @@ -844,7 +936,7 @@ export class StatusLineComponent implements Component { if (this.#jjCacheGeneration !== generation || this.#disposed) return; const changed = next !== this.#cachedJjBranch; this.#cachedJjBranch = next; - if (changed && this.#onBranchChange) this.#onBranchChange(); + if (changed) this.#onBranchChange?.(); })(); return this.#cachedJjBranch; } @@ -856,23 +948,30 @@ export class StatusLineComponent implements Component { const cwd = effectiveGitCwd ?? this.#resolveActiveRepoCache().effectiveGitCwd; const root = this.#jjRootFor(cwd); if (!root) return null; - if (this.#jjStatusInFlight || Date.now() - this.#jjStatusLastFetch < JJ_REFRESH_TTL_MS) { + if (this.#jjStatusActive || Date.now() - this.#jjStatusLastFetch < JJ_REFRESH_TTL_MS) { return this.#cachedJjStatus; } - this.#jjStatusInFlight = true; + const request: JjResolveRequest = { + id: ++this.#jjResolveSeq, + controller: new AbortController(), + }; + this.#jjStatusActive = request; const generation = this.#jjCacheGeneration; (async () => { let next: { staged: number; unstaged: number; untracked: number } | null = null; try { - next = await jj.status.summary(root); + next = await jj.status.summary(root, { + signal: request.controller.signal, + timeoutMs: jj.JJ_COMMAND_TIMEOUT_MS, + }); } finally { - this.#jjStatusInFlight = false; + if (this.#jjStatusActive?.id === request.id) this.#jjStatusActive = undefined; if (this.#jjCacheGeneration === generation) this.#jjStatusLastFetch = Date.now(); } if (this.#jjCacheGeneration !== generation || this.#disposed) return; const prev = this.#cachedJjStatus; this.#cachedJjStatus = next; - if (this.#onBranchChange && JSON.stringify(prev) !== JSON.stringify(next)) this.#onBranchChange(); + if (JSON.stringify(prev) !== JSON.stringify(next)) this.#onBranchChange?.(); })(); return this.#cachedJjStatus; } diff --git a/packages/coding-agent/src/modes/interactive-mode.ts b/packages/coding-agent/src/modes/interactive-mode.ts index 713ab4801..762997a32 100644 --- a/packages/coding-agent/src/modes/interactive-mode.ts +++ b/packages/coding-agent/src/modes/interactive-mode.ts @@ -1273,8 +1273,7 @@ export class InteractiveMode implements InteractiveModeContext { await this.refreshSkillState(); await this.refreshSlashCommandState(newCwd); setSessionTerminalTitle(this.sessionManager.getSessionName(), this.sessionManager.getCwd()); - this.statusLine.invalidate(); - this.ui.requestRender(); + this.statusLine.applyCwdChange(); } async getUserInput(): Promise { diff --git a/packages/coding-agent/src/utils/jj.ts b/packages/coding-agent/src/utils/jj.ts index 16ec4a513..624839256 100644 --- a/packages/coding-agent/src/utils/jj.ts +++ b/packages/coding-agent/src/utils/jj.ts @@ -2,6 +2,7 @@ import * as fs from "node:fs"; import * as path from "node:path"; import { $which } from "@oh-my-pi/pi-utils"; import { LRUCache } from "lru-cache/raw"; +import { withTimeoutSignal } from "./fetch-timeout"; import * as git from "./git"; // ════════════════════════════════════════════════════════════════════════════ @@ -27,19 +28,24 @@ export interface JjRepository { } /** Options for `jj diff` invocations. */ -export interface DiffOptions { +export interface DiffOptions extends JjCommandOptions { /** Optional file paths to restrict the diff with `-- `. */ readonly files?: readonly string[]; /** Return only changed file names instead of Git-format diff text. */ readonly nameOnly?: boolean; - /** Optional abort signal passed to the spawned `jj` process. */ - readonly signal?: AbortSignal; } -interface CommandOptions { +/** Options for a bounded `jj` subprocess query. */ +export interface JjCommandOptions { + /** Optional cancellation signal for the subprocess. */ readonly signal?: AbortSignal; + /** Deadline in milliseconds. Defaults to {@link JJ_COMMAND_TIMEOUT_MS}. */ + readonly timeoutMs?: number; } +/** Default finite deadline for local jj subprocesses. */ +export const JJ_COMMAND_TIMEOUT_MS = 5_000; + // ════════════════════════════════════════════════════════════════════════════ // Error // ════════════════════════════════════════════════════════════════════════════ @@ -83,10 +89,10 @@ function formatCommandFailure( return `jj ${args.join(" ")} failed with exit code ${result.exitCode}`; } -async function jj(cwd: string, args: readonly string[], options: CommandOptions = {}): Promise { +async function jj(cwd: string, args: readonly string[], options: JjCommandOptions = {}): Promise { const child = Bun.spawn(["jj", "--no-pager", "--color=never", ...args], { cwd, - signal: options.signal, + signal: withTimeoutSignal(options.timeoutMs ?? JJ_COMMAND_TIMEOUT_MS, options.signal), stdin: "ignore", stdout: "pipe", stderr: "pipe", @@ -109,7 +115,7 @@ async function jj(cwd: string, args: readonly string[], options: CommandOptions async function runChecked( cwd: string, args: readonly string[], - options: CommandOptions = {}, + options: JjCommandOptions = {}, ): Promise { ensureAvailable(); const result = await jj(cwd, args, options); @@ -119,14 +125,14 @@ async function runChecked( return result; } -async function runText(cwd: string, args: readonly string[], options: CommandOptions = {}): Promise { +async function runText(cwd: string, args: readonly string[], options: JjCommandOptions = {}): Promise { return (await runChecked(cwd, args, options)).stdout; } async function runOptionalText( cwd: string, args: readonly string[], - options: CommandOptions = {}, + options: JjCommandOptions = {}, ): Promise { try { const result = await jj(cwd, args, options); @@ -294,7 +300,7 @@ export const workingCopy = { * Label `@` with its nearest bookmark, falling back to its short change ID. * Returns `null` when `jj` is unavailable or the query fails. */ - async label(cwd: string, signal?: AbortSignal): Promise { + async label(cwd: string, options?: JjCommandOptions): Promise { const raw = await runOptionalText( cwd, [ @@ -306,7 +312,7 @@ export const workingCopy = { "-T", WORKING_COPY_LABEL_TEMPLATE, ], - { signal }, + options, ); return raw === null ? null : parseWorkingCopyLabel(raw); }, @@ -325,8 +331,8 @@ export const status = { * Count changes in `@` relative to its parent using the Git status shape. * Jujutsu has no index, so `staged` is always zero. */ - async summary(cwd: string, signal?: AbortSignal): Promise { - const raw = await runOptionalText(cwd, ["diff", "-r", "@", "--summary", "--ignore-working-copy"], { signal }); + async summary(cwd: string, options?: JjCommandOptions): Promise { + const raw = await runOptionalText(cwd, ["diff", "-r", "@", "--summary", "--ignore-working-copy"], options); return raw === null ? null : parseStatusSummary(raw); }, diff --git a/packages/coding-agent/test/git-reftable.test.ts b/packages/coding-agent/test/git-reftable.test.ts index 5b05c4c15..5a09821f5 100644 --- a/packages/coding-agent/test/git-reftable.test.ts +++ b/packages/coding-agent/test/git-reftable.test.ts @@ -130,12 +130,12 @@ describe.skipIf(!supportsReftable)("git reftable support", () => { exitCode: 0, exitedDueToTimeout: true, stdout: Buffer.from("refs/heads/feature-branch\n"), - } satisfies ReturnType; + } satisfies Bun.ReadableSyncSubprocess; const successfulRevParse = { ...baseResult, exitCode: 0, stdout: Buffer.from(`${headSha}\n`), - } satisfies ReturnType; + } satisfies Bun.ReadableSyncSubprocess; vi.spyOn(Bun, "spawnSync").mockReturnValueOnce(timedOutSymbolicRef).mockReturnValueOnce(successfulRevParse); const headState = git.head.resolveSync(sharedRepoDir); diff --git a/packages/coding-agent/test/modes/components/status-line/component.jj-cache.test.ts b/packages/coding-agent/test/modes/components/status-line/component.jj-cache.test.ts index d4341c2fd..47bfb158d 100644 --- a/packages/coding-agent/test/modes/components/status-line/component.jj-cache.test.ts +++ b/packages/coding-agent/test/modes/components/status-line/component.jj-cache.test.ts @@ -101,7 +101,7 @@ afterEach(() => { }); describe("StatusLineComponent jj cache coherence", () => { - it("invalidate() drops the throttled jj branch cache within its TTL and refetches", async () => { + it("invalidateGitCaches() drops the throttled jj branch cache within its TTL and refetches", async () => { // A live jj bookmark label; a second query for the SAME root returns a new // label, simulating a colocated bookmark/HEAD move mid-TTL. const branchSpy = spyOn(jj.workingCopy, "label").mockResolvedValue("bookmark-v1"); @@ -118,17 +118,16 @@ describe("StatusLineComponent jj cache coherence", () => { // Move the bookmark: a plain render within the 5s TTL must keep serving the // cached label without a refetch (guards that the TTL is real, so the next - // assertion proves invalidate() — not TTL expiry — forces the refresh). + // assertion proves invalidateGitCaches() — not TTL expiry — forces the refresh). branchSpy.mockResolvedValue("bookmark-v2"); const throttled = visible(statusLine.getTopBorder(WIDTH).content); await flushMicrotasks(); expect(throttled).toContain("bookmark-v1"); expect(branchSpy).toHaveBeenCalledTimes(1); - // invalidate() (public watcher trigger) must reset the jj caches so the - // next render refetches despite being inside the TTL, and paint the new - // label — the finding-1 contract. - statusLine.invalidate(); + // A HEAD/bookmark change must explicitly reset the jj caches so the next + // render refetches despite being inside the TTL and paints the new label. + statusLine.invalidateGitCaches(); statusLine.getTopBorder(WIDTH); await flushMicrotasks(); expect(branchSpy).toHaveBeenCalledTimes(2); @@ -154,18 +153,19 @@ describe("StatusLineComponent jj cache coherence", () => { expect(visible(statusLine.getTopBorder(WIDTH).content)).not.toContain("branch-A-STALE"); expect(branchSpy).toHaveBeenCalledTimes(1); - // Switch to repo B mid-flight. #jjRootFor(tmpB) re-points #jjRoot to ROOT_B - // and resets the jj caches; ROOT_A's lookup is now stale. The render can't - // start B's lookup yet — the single in-flight flag is still held by A. + // Switch to repo B mid-flight. The cwd-change caller explicitly resets the + // VCS caches, aborting ROOT_A's in-flight query and clearing the slot so + // B's lookup can start immediately on the next render. setProjectDir(tmpB); + statusLine.invalidateGitCaches(); statusLine.getTopBorder(WIDTH); await flushMicrotasks(); - expect(branchSpy).toHaveBeenCalledTimes(1); + expect(branchSpy).toHaveBeenCalledTimes(2); // Let ROOT_A's slow query finish, then drain its continuation. The - // root-keyed guard must DROP it: #jjRoot is ROOT_B, so A's label must never - // become B's cached branch, and A's completion must not advance B's - // throttle (leaving B free to refetch) — the finding-2/4 contract. + // generation guard must DROP it, so A's label never becomes B's cached + // branch and its completion cannot advance B's throttle (leaving B free + // to refetch) — the finding-2/4 contract. deferredA.resolve("branch-A-STALE"); await flushMicrotasks(); @@ -206,9 +206,9 @@ describe("StatusLineComponent jj cache coherence", () => { await flushMicrotasks(); expect(branchSpy).toHaveBeenCalledTimes(1); - // A HEAD/bookmark move fires the watcher → invalidate(). The cwd is + // A HEAD/bookmark move fires the watcher → invalidateGitCaches(). The cwd is // unchanged, so the next #jjRootFor re-resolves #jjRoot to the SAME ROOT_A. - statusLine.invalidate(); + statusLine.invalidateGitCaches(); statusLine.getTopBorder(WIDTH); await flushMicrotasks(); diff --git a/packages/coding-agent/test/status-line-vcs-refresh.test.ts b/packages/coding-agent/test/status-line-vcs-refresh.test.ts index 4594bc4ce..aa64153c8 100644 --- a/packages/coding-agent/test/status-line-vcs-refresh.test.ts +++ b/packages/coding-agent/test/status-line-vcs-refresh.test.ts @@ -12,6 +12,8 @@ * same callback is covered by status-line-dispose-async-leak.test.ts.) */ import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "bun:test"; +import { EventEmitter } from "node:events"; +import * as nodeFs from "node:fs"; import * as fs from "node:fs/promises"; import * as os from "node:os"; import * as path from "node:path"; @@ -207,9 +209,9 @@ describe("StatusLineComponent reftable branch resolve honors mid-flight invalida component.getTopBorder(80); expect(git.head.resolve).toHaveBeenCalledTimes(1); - // A HEAD move fires the watcher: invalidate bumps the generation and - // releases the in-flight slot. - component.invalidate(); + // A HEAD move fires the watcher: invalidateGitCaches bumps the + // generation and releases the in-flight slot. + component.invalidateGitCaches(); // The repaint starts a fresh resolve (R2) for the same cwd. component.getTopBorder(80); @@ -258,9 +260,9 @@ describe("StatusLineComponent reftable branch resolve honors mid-flight invalida vi.spyOn(git.head, "resolve").mockImplementation((_cwd, signal) => { if (!signal) throw new Error("reftable resolve must receive an abort signal"); signals.push(signal); - return new Promise((_resolve, reject) => { - signal.addEventListener("abort", () => reject(signal.reason), { once: true }); - }); + const { promise, reject } = Promise.withResolvers(); + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + return promise; }); const component = new StatusLineComponent(makeSession()); @@ -268,9 +270,9 @@ describe("StatusLineComponent reftable branch resolve honors mid-flight invalida component.getTopBorder(80); expect(git.head.resolve).toHaveBeenCalledTimes(1); - component.invalidate(); + component.invalidateGitCaches(); expect(signals[0]?.aborted).toBe(true); - component.invalidate(); + component.invalidateGitCaches(); component.getTopBorder(80); component.getTopBorder(80); expect(git.head.resolve).toHaveBeenCalledTimes(2); @@ -280,6 +282,96 @@ describe("StatusLineComponent reftable branch resolve honors mid-flight invalida await Promise.resolve(); }); + it("generic invalidate does not abort or restart a live reftable HEAD resolve", async () => { + const fakeRepo = { + commonDir: "/fake/.git", + gitDir: "/fake/.git", + gitEntryPath: "/fake/.git", + headPath: "/fake/.git/HEAD", + repoRoot: "/fake", + } satisfies GitRepository; + vi.spyOn(git.repo, "resolveSync").mockReturnValue(fakeRepo); + vi.spyOn(git.repo, "isReftableSync").mockReturnValue(true); + vi.spyOn(git.status, "summary").mockReturnValue(Promise.withResolvers().promise); + vi.spyOn(jj.repo, "rootSync").mockReturnValue(null); + vi.spyOn(nodeFs, "watch").mockImplementation(() => { + throw new Error("watch unavailable"); + }); + + const signals: AbortSignal[] = []; + const { promise, reject } = Promise.withResolvers(); + vi.spyOn(git.head, "resolve").mockImplementation((_cwd, signal) => { + if (!signal) throw new Error("reftable resolve must receive an abort signal"); + signals.push(signal); + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + return promise; + }); + + const component = new StatusLineComponent(makeSession()); + component.updateSettings(gitSegment); + component.watchBranch(vi.fn()); + component.getTopBorder(80); + expect(git.head.resolve).toHaveBeenCalledTimes(1); + + // Many generic invalidations (message events, model switches, theme + // changes, …) must not abort the live resolve or fan out replacement + // git subprocesses — the render path self-invalidates via cwd/context + // cache-miss checks, so a generic paint only re-renders. + for (let i = 0; i < 10; i++) { + component.invalidate(); + } + component.getTopBorder(80); + component.getTopBorder(80); + + expect(signals[0]?.aborted).toBe(false); + expect(git.head.resolve).toHaveBeenCalledTimes(1); + + // Disposal still aborts the in-flight resolve. + component.dispose(); + expect(signals[0]?.aborted).toBe(true); + await Promise.resolve(); + }); + + it("polls a reftable branch after HEAD watcher installation fails", async () => { + const fakeRepo = { + commonDir: "/fake/.git", + gitDir: "/fake/.git", + gitEntryPath: "/fake/.git", + headPath: "/fake/.git/HEAD", + repoRoot: "/fake", + } satisfies GitRepository; + vi.spyOn(git.repo, "resolveSync").mockReturnValue(fakeRepo); + vi.spyOn(git.repo, "isReftableSync").mockReturnValue(true); + vi.spyOn(git.status, "summary").mockReturnValue(Promise.withResolvers().promise); + vi.spyOn(jj.repo, "rootSync").mockReturnValue(null); + vi.spyOn(nodeFs, "watch").mockImplementation(() => { + throw new Error("watch unavailable"); + }); + let now = 1_000_000; + vi.spyOn(Date, "now").mockImplementation(() => now); + vi.spyOn(git.head, "resolve") + .mockResolvedValueOnce({ ...fakeRefHead, branchName: "before-change", ref: "refs/heads/before-change" }) + .mockResolvedValueOnce({ ...fakeRefHead, branchName: "after-change", ref: "refs/heads/after-change" }); + + const component = new StatusLineComponent(makeSession()); + component.updateSettings(gitSegment); + component.watchBranch(vi.fn()); + component.getTopBorder(80); + await Promise.resolve(); + await Promise.resolve(); + expect(component.getTopBorder(80).content).toContain("before-change"); + expect(git.head.resolve).toHaveBeenCalledTimes(1); + + // No filesystem event arrives, but the next bounded poll observes the new HEAD. + now += 5_001; + component.getTopBorder(80); + expect(git.head.resolve).toHaveBeenCalledTimes(2); + await Promise.resolve(); + await Promise.resolve(); + expect(component.getTopBorder(80).content).toContain("after-change"); + component.dispose(); + }); + it("does not query an ancestor jj workspace while nested Git HEAD resolution is pending", async () => { const jjRootDir = await fs.mkdtemp(path.join(os.tmpdir(), "status-line-jj-root-")); const nestedGitCwd = path.join(jjRootDir, "nested-ordinary-git"); @@ -317,3 +409,308 @@ describe("StatusLineComponent reftable branch resolve honors mid-flight invalida } }); }); + +describe("StatusLineComponent VCS watcher and jj request lifecycle", () => { + const fakeRepo = { + commonDir: "/fake/.git", + gitDir: "/fake/.git", + gitEntryPath: "/fake/.git", + headPath: "/fake/.git/HEAD", + repoRoot: "/fake", + } satisfies GitRepository; + + it("retires an asynchronously failed watcher without an unhandled EventEmitter error", () => { + const firstWatcher = Object.assign(new EventEmitter(), { close: vi.fn() }) as unknown as nodeFs.FSWatcher; + const failedWatcher = Object.assign(new EventEmitter(), { close: vi.fn() }) as unknown as nodeFs.FSWatcher; + const disposedWatcher = Object.assign(new EventEmitter(), { close: vi.fn() }) as unknown as nodeFs.FSWatcher; + vi.spyOn(git.repo, "resolveSync").mockReturnValue(fakeRepo); + vi.spyOn(git.repo, "isReftableSync").mockReturnValue(false); + vi.spyOn(git.head, "resolveSync") + .mockReturnValueOnce({ ...fakeRefHead, branchName: "before-error", ref: "refs/heads/before-error" }) + .mockReturnValueOnce({ ...fakeRefHead, branchName: "after-error", ref: "refs/heads/after-error" }); + vi.spyOn(git.branch, "default").mockReturnValue(Promise.withResolvers().promise); + vi.spyOn(git.status, "summary").mockReturnValue(Promise.withResolvers().promise); + vi.spyOn(jj.repo, "rootSync").mockReturnValue(null); + vi.spyOn(nodeFs, "watch") + .mockReturnValueOnce(firstWatcher) + .mockReturnValueOnce(failedWatcher) + .mockReturnValueOnce(disposedWatcher); + + const onBranchChange = vi.fn(); + const component = new StatusLineComponent(makeSession()); + component.updateSettings(gitSegment); + component.watchBranch(onBranchChange); + component.getTopBorder(80); + expect(firstWatcher.listenerCount("error")).toBe(1); + + // Replacement detaches the first listener before closing that watcher. + component.updateSettings(gitSegment); + expect(firstWatcher.listenerCount("error")).toBe(0); + expect(firstWatcher.close).toHaveBeenCalledTimes(1); + expect(failedWatcher.listenerCount("error")).toBe(1); + + // `error` without a listener throws synchronously. The component must own + // the event, retire the watcher, and request the repaint that observes the + // invalidated VCS cache. + expect(() => failedWatcher.emit("error", new Error("watch failed"))).not.toThrow(); + expect(failedWatcher.listenerCount("error")).toBe(0); + expect(failedWatcher.close).toHaveBeenCalledTimes(1); + expect(onBranchChange).toHaveBeenCalledTimes(1); + expect(component.getTopBorder(80).content).toContain("after-error"); + + component.updateSettings(gitSegment); + expect(disposedWatcher.listenerCount("error")).toBe(1); + component.dispose(); + expect(disposedWatcher.listenerCount("error")).toBe(0); + expect(disposedWatcher.close).toHaveBeenCalledTimes(1); + }); + + it("discovers a repository created after setup with bounded single-flight polling", async () => { + let now = 1_000_000; + const repositoryCreatedAt = now + 5_000; + vi.spyOn(Date, "now").mockImplementation(() => now); + vi.spyOn(git.repo, "resolveSync").mockImplementation(() => (now >= repositoryCreatedAt ? fakeRepo : null)); + vi.spyOn(git.repo, "isReftableSync").mockReturnValue(true); + vi.spyOn(git.status, "summary").mockReturnValue(Promise.withResolvers().promise); + vi.spyOn(jj.repo, "rootSync").mockReturnValue(null); + const head = Promise.withResolvers(); + vi.spyOn(git.head, "resolve").mockReturnValue(head.promise); + + const component = new StatusLineComponent(makeSession()); + component.updateSettings(gitSegment); + component.watchBranch(vi.fn()); + component.getTopBorder(80); + expect(git.head.resolve).not.toHaveBeenCalled(); + + now += 1_000; + component.getTopBorder(80); + expect(git.head.resolve).not.toHaveBeenCalled(); + + // The bounded discovery interval reaches the new repository. Repeated + // paints while its reftable resolve is hung must reuse the one request. + now += 4_001; + component.getTopBorder(80); + component.getTopBorder(80); + expect(git.head.resolve).toHaveBeenCalledTimes(1); + + head.resolve({ ...fakeRefHead, branchName: "created-later", ref: "refs/heads/created-later" }); + await Promise.resolve(); + await Promise.resolve(); + expect(component.getTopBorder(80).content).toContain("created-later"); + component.dispose(); + }); + + it("aborts superseded jj branch and status queries without blocking their replacements", async () => { + vi.spyOn(git.head, "resolveSync").mockReturnValue(null); + vi.spyOn(git.branch, "default").mockReturnValue(Promise.withResolvers().promise); + vi.spyOn(git.status, "summary").mockReturnValue(Promise.withResolvers().promise); + vi.spyOn(jj.repo, "rootSync").mockReturnValue("/fake/jj/root"); + + const labelRequests: Array<{ signal: AbortSignal; resolve: (value: string | null) => void }> = []; + vi.spyOn(jj.workingCopy, "label").mockImplementation((_root, options) => { + if (!options?.signal || options.timeoutMs !== jj.JJ_COMMAND_TIMEOUT_MS) { + throw new Error("jj label requires the central bounded options"); + } + const request = Promise.withResolvers(); + options.signal.addEventListener("abort", () => request.resolve(null), { once: true }); + labelRequests.push({ signal: options.signal, resolve: request.resolve }); + return request.promise; + }); + const statusRequests: Array<{ signal: AbortSignal; resolve: (value: GitStatus | null) => void }> = []; + vi.spyOn(jj.status, "summary").mockImplementation((_root, options) => { + if (!options?.signal || options.timeoutMs !== jj.JJ_COMMAND_TIMEOUT_MS) { + throw new Error("jj status requires the central bounded options"); + } + const request = Promise.withResolvers(); + options.signal.addEventListener("abort", () => request.resolve(null), { once: true }); + statusRequests.push({ signal: options.signal, resolve: request.resolve }); + return request.promise; + }); + + const component = new StatusLineComponent(makeSession()); + component.updateSettings(gitSegment); + component.getTopBorder(80); + expect(labelRequests).toHaveLength(1); + expect(statusRequests).toHaveLength(1); + + component.invalidateGitCaches(); + expect(labelRequests[0]?.signal.aborted).toBe(true); + expect(statusRequests[0]?.signal.aborted).toBe(true); + component.getTopBorder(80); + expect(labelRequests).toHaveLength(2); + expect(statusRequests).toHaveLength(2); + + labelRequests[1]?.resolve("fresh-bookmark"); + statusRequests[1]?.resolve({ staged: 0, unstaged: 1, untracked: 0 }); + await Promise.resolve(); + await Promise.resolve(); + expect(component.getTopBorder(80).content).toContain("fresh-bookmark"); + + component.invalidateGitCaches(); + component.getTopBorder(80); + expect(labelRequests).toHaveLength(3); + expect(statusRequests).toHaveLength(3); + component.dispose(); + expect(labelRequests[2]?.signal.aborted).toBe(true); + expect(statusRequests[2]?.signal.aborted).toBe(true); + await Promise.resolve(); + }); +}); + +describe("StatusLineComponent applyCwdChange re-points watcher ownership", () => { + let dirA: string; + let dirB: string; + let dirNoRepo: string; + let repoA: GitRepository; + let repoB: GitRepository; + + beforeAll(async () => { + dirA = await fs.mkdtemp(path.join(os.tmpdir(), "status-line-repoA-")); + dirB = await fs.mkdtemp(path.join(os.tmpdir(), "status-line-repoB-")); + dirNoRepo = await fs.mkdtemp(path.join(os.tmpdir(), "status-line-norepo-")); + repoA = { + commonDir: path.join(dirA, ".git"), + gitDir: path.join(dirA, ".git"), + gitEntryPath: path.join(dirA, ".git"), + headPath: path.join(dirA, ".git", "HEAD"), + repoRoot: dirA, + }; + repoB = { + commonDir: path.join(dirB, ".git"), + gitDir: path.join(dirB, ".git"), + gitEntryPath: path.join(dirB, ".git"), + headPath: path.join(dirB, ".git", "HEAD"), + repoRoot: dirB, + }; + }); + + afterAll(async () => { + setProjectDir(originalProjectDir); + await Promise.all([ + fs.rm(dirA, { recursive: true, force: true }), + fs.rm(dirB, { recursive: true, force: true }), + fs.rm(dirNoRepo, { recursive: true, force: true }), + ]); + }); + + // Test double for node:fs.FSWatcher — extends EventEmitter with just the + // `close` method the component calls. FSWatcher has dozens of members we + // never exercise, so a structural implementation would be pure ceremony. + function createFakeWatcher(): nodeFs.FSWatcher { + return Object.assign(new EventEmitter(), { close: vi.fn() }) as unknown as nodeFs.FSWatcher; + } + + it("retires the old watcher and re-points at the new repo on cwd change", () => { + const watcherA = createFakeWatcher(); + const watcherB = createFakeWatcher(); + + vi.spyOn(git.repo, "isReftableSync").mockReturnValue(false); + vi.spyOn(git.repo, "linkedWorktreeSync").mockReturnValue(null); + vi.spyOn(git.repo, "resolveSync").mockImplementation((cwd: string) => { + if (cwd === dirA) return repoA; + if (cwd === dirB) return repoB; + return null; + }); + vi.spyOn(git.head, "resolveSync").mockImplementation((cwd: string) => { + if (cwd === dirA) return { ...fakeRefHead, branchName: "branch-a", ref: "refs/heads/branch-a" }; + if (cwd === dirB) return { ...fakeRefHead, branchName: "branch-b", ref: "refs/heads/branch-b" }; + return null; + }); + vi.spyOn(git.branch, "default").mockReturnValue(Promise.withResolvers().promise); + vi.spyOn(git.status, "summary").mockReturnValue(Promise.withResolvers().promise); + vi.spyOn(jj.repo, "rootSync").mockReturnValue(null); + const watchSpy = vi.spyOn(nodeFs, "watch").mockReturnValueOnce(watcherA).mockReturnValueOnce(watcherB); + + const onBranchChange = vi.fn(); + setProjectDir(dirA); + const component = new StatusLineComponent(makeSession()); + component.updateSettings(gitSegment); + component.watchBranch(onBranchChange); + expect(component.getTopBorder(80).content).toContain("branch-a"); + + // Move cwd to repo B — the SessionManager's cwd has already moved. + setProjectDir(dirB); + component.applyCwdChange(); + // applyCwdChange itself requests one repaint; clear it so subsequent + // calls are attributable solely to watcher events. + onBranchChange.mockClear(); + + // Old watcher is retired: closed and its error listener detached. + expect(watcherA.close).toHaveBeenCalledTimes(1); + expect(watcherA.listenerCount("error")).toBe(0); + // New watcher is live with an error listener attached. + expect(watcherB.listenerCount("error")).toBe(1); + + // fs.watch is mocked to return EventEmitters without registering the + // change callback, so extract it from the call args and attach it. + const aCallArgs = watchSpy.mock.calls[0]; + const bCallArgs = watchSpy.mock.calls[1]; + const aListener = aCallArgs?.[1]; + const bListener = bCallArgs?.[1]; + expect(typeof aListener).toBe("function"); + expect(typeof bListener).toBe("function"); + if (typeof aListener === "function") watcherA.on("change", aListener as () => void); + if (typeof bListener === "function") watcherB.on("change", bListener as () => void); + + // Stale change event from repo A's retired watcher must not invalidate + // B's caches or request a repaint — the ownership guard rejects it. + watcherA.emit("change"); + expect(onBranchChange).not.toHaveBeenCalled(); + + // Fresh change event from repo B's watcher refreshes B. + onBranchChange.mockClear(); + watcherB.emit("change"); + expect(onBranchChange).toHaveBeenCalledTimes(1); + expect(component.getTopBorder(80).content).toContain("branch-b"); + + // No watcher leak: dispose closes B, not A (A was already closed). + component.dispose(); + expect(watcherB.close).toHaveBeenCalledTimes(1); + expect(watcherB.listenerCount("error")).toBe(0); + expect(watcherA.close).toHaveBeenCalledTimes(1); + }); + + it("falls back to bounded polling when the new cwd has no repository", () => { + const watcherA = createFakeWatcher(); + + vi.spyOn(git.repo, "isReftableSync").mockReturnValue(false); + vi.spyOn(git.repo, "linkedWorktreeSync").mockReturnValue(null); + vi.spyOn(git.repo, "resolveSync").mockImplementation((cwd: string) => { + if (cwd === dirA) return repoA; + return null; + }); + vi.spyOn(git.head, "resolveSync").mockImplementation((cwd: string) => { + if (cwd === dirA) return { ...fakeRefHead, branchName: "branch-a", ref: "refs/heads/branch-a" }; + return null; + }); + vi.spyOn(git.branch, "default").mockReturnValue(Promise.withResolvers().promise); + vi.spyOn(git.status, "summary").mockReturnValue(Promise.withResolvers().promise); + vi.spyOn(jj.repo, "rootSync").mockReturnValue(null); + vi.spyOn(nodeFs, "watch").mockReturnValueOnce(watcherA); + + const onBranchChange = vi.fn(); + setProjectDir(dirA); + const component = new StatusLineComponent(makeSession()); + component.updateSettings(gitSegment); + component.watchBranch(onBranchChange); + component.getTopBorder(80); + + // Move to a directory with no git repo — watcher unavailable fallback. + setProjectDir(dirNoRepo); + onBranchChange.mockClear(); + component.applyCwdChange(); + + // Old watcher retired; no new watcher created (fs.watch not called again). + expect(watcherA.close).toHaveBeenCalledTimes(1); + expect(nodeFs.watch).toHaveBeenCalledTimes(1); + // applyCwdChange still requests a repaint so the stale segment clears. + expect(onBranchChange).toHaveBeenCalledTimes(1); + + // Rendering does not crash and the git segment is blank for no-repo. + const border = component.getTopBorder(80); + expect(border).toBeDefined(); + expect(border.content).not.toContain("branch-a"); + + component.dispose(); + }); +}); diff --git a/packages/coding-agent/test/utils/jj.test.ts b/packages/coding-agent/test/utils/jj.test.ts index 9f59bc495..37d750cc6 100644 --- a/packages/coding-agent/test/utils/jj.test.ts +++ b/packages/coding-agent/test/utils/jj.test.ts @@ -1,14 +1,16 @@ -import { afterEach, describe, expect, it } from "bun:test"; +import { afterEach, describe, expect, it, vi } from "bun:test"; import * as fs from "node:fs/promises"; import * as os from "node:os"; import * as path from "node:path"; import * as jj from "@oh-my-pi/pi-coding-agent/utils/jj"; import { removeWithRetries } from "@oh-my-pi/pi-utils"; +import type { Subprocess } from "bun"; describe("jj workspace detection", () => { let tmpDir: string | undefined; afterEach(async () => { + vi.restoreAllMocks(); jj.repo.clearRootCache(); if (tmpDir) { await removeWithRetries(tmpDir); @@ -184,3 +186,59 @@ describe("jj status", () => { expect(jj.status.parse("")).toEqual({ staged: 0, unstaged: 0, untracked: 0 }); }); }); + +describe("jj subprocess deadlines", () => { + type SpawnOptions = Bun.SpawnOptions.SpawnOptions< + Bun.SpawnOptions.Writable, + Bun.SpawnOptions.Readable, + Bun.SpawnOptions.Readable + >; + const calls: SpawnOptions[] = []; + + afterEach(() => { + calls.length = 0; + vi.restoreAllMocks(); + }); + + function textStream(text = ""): ReadableStream { + const body = new Response(text).body; + if (!body) throw new Error("missing response body"); + return body; + } + + function mockSpawn(options: SpawnOptions & { cmd: string[] }): Subprocess; + function mockSpawn(cmd: string[], options?: SpawnOptions): Subprocess; + function mockSpawn(first: string[] | (SpawnOptions & { cmd: string[] }), second?: SpawnOptions): Subprocess { + calls.push(Array.isArray(first) ? (second ?? ({} as SpawnOptions)) : first); + return { + pid: 12345, + stdout: textStream(), + stderr: textStream(), + exited: Promise.resolve(0), + } as Subprocess; + } + + it("combines caller cancellation with a finite subprocess deadline", async () => { + vi.spyOn(Bun, "spawn").mockImplementation(mockSpawn); + const controller = new AbortController(); + + await jj.workingCopy.label("/fake", { signal: controller.signal, timeoutMs: 1 }); + const signal = calls[0]?.signal; + expect(signal).toBeDefined(); + expect(signal).not.toBe(controller.signal); + expect(signal?.aborted).toBe(false); + + controller.abort(); + expect(signal?.aborted).toBe(true); + }); + + it("aborts the spawned process signal at its explicit deadline", async () => { + vi.spyOn(Bun, "spawn").mockImplementation(mockSpawn); + + await jj.status.summary("/fake", { timeoutMs: 1 }); + const signal = calls[0]?.signal; + expect(signal?.aborted).toBe(false); + await Bun.sleep(10); + expect(signal?.aborted).toBe(true); + }); +});