diff --git a/packages/core/src/base.ts b/packages/core/src/base.ts index e8ad49ce0..903671fa3 100644 --- a/packages/core/src/base.ts +++ b/packages/core/src/base.ts @@ -35,14 +35,14 @@ async function repositoryBase(folder: string, root: string): Promise { * repository Tesota's snapshot of it. Verification reads this record and * never rebuilds the base after the agent changed the files. */ -export async function captureBase(cwd: string): Promise { +export async function captureBase(cwd: string, signal?: AbortSignal): Promise { const units = workspaceUnits(cwd); const workspace = units.some(({ folder }) => folder !== ""); const found: BaseUnit[] = []; for (const { folder, repository } of units) { const root = join(cwd, folder); if (repository) { found.push(await repositoryBase(folder, root)); continue; } - const snapshot = await takeSnapshot(root, workspace && folder === "", withLockWhile); + const snapshot = await takeSnapshot(root, workspace && folder === "", withLockWhile, undefined, signal); found.push("refused" in snapshot ? { folder, kind: "refused", reason: snapshot.refused } : { folder, kind: "snapshot", commit: snapshot.base, tooLarge: snapshot.tooLarge }); } diff --git a/packages/core/src/loop.ts b/packages/core/src/loop.ts index 8ff91a161..8e744e4c5 100644 --- a/packages/core/src/loop.ts +++ b/packages/core/src/loop.ts @@ -33,7 +33,7 @@ export interface LoopOptions { /** What a resumed session already holds; a new session starts empty. */ readonly resumed?: LoopState; /** The base of a request; the gate's units (`captureBase`) when not given. */ - readonly captureBase?: (cwd: string) => Promise; + readonly captureBase?: (cwd: string, signal: AbortSignal) => Promise; /** The session's extensions; they settle before the verifier and cannot replace it. */ readonly extensions?: readonly ExtensionFactory[]; /** What verifies each run before it may complete (#439); a run whose verifier has no settle handler never completes. */ @@ -745,7 +745,7 @@ export function createLoop(options: LoopOptions): LoopSession { async function start(run: number, text: string, origin: Origin, signal: AbortSignal): Promise { const at = new Date().toISOString(); const written = (async () => { - writer.append({ type: "request", run, text, origin, at, base: await base(options.cwd) }); + writer.append({ type: "request", run, text, origin, at, base: await base(options.cwd, signal) }); writer.sync(); })(); recorded = written.catch(() => undefined); diff --git a/packages/core/src/session-lock.ts b/packages/core/src/session-lock.ts index c3305360a..e8f75cffc 100644 --- a/packages/core/src/session-lock.ts +++ b/packages/core/src/session-lock.ts @@ -110,11 +110,15 @@ export function withLock(path: string, work: () => T): T { /** * Runs `work`, which may wait on the network, holding the lock at `path`, * waiting up to `waitMs` for another holder in this process or another - * (#459): an MCP token refresh, which a second at once would spend. + * (#459): an MCP token refresh, which a second at once would spend. A + * `signal` that stops ends the wait, refused as the lock being held; work + * that already holds the lock runs to its end, so the lock is never released + * under it. */ -export async function withLockWhile(path: string, work: () => Promise, waitMs: number): Promise { +export async function withLockWhile(path: string, work: () => Promise, waitMs: number, signal?: AbortSignal): Promise { const until = Date.now() + waitMs; - const waiting = Schedule.spaced(50).pipe(Schedule.while(({ input }) => input instanceof SessionLockedError && Date.now() < until)); + const waiting = Schedule.spaced(50).pipe(Schedule.while(({ input }) => + input instanceof SessionLockedError && Date.now() < until && signal?.aborted !== true)); const acquire = Effect.retry(Effect.try({ try: () => lockSession(path), catch: (error) => error }), waiting); return Effect.runPromise(Effect.acquireUseRelease(acquire, () => Effect.tryPromise({ try: work, catch: (error) => error }), diff --git a/packages/core/tests/snapshot.test.ts b/packages/core/tests/snapshot.test.ts index cb7a7ea49..7109ddcd3 100644 --- a/packages/core/tests/snapshot.test.ts +++ b/packages/core/tests/snapshot.test.ts @@ -1,5 +1,5 @@ import { expect, test } from "bun:test"; -import { mkdtempSync, readdirSync, rmSync, writeFileSync } from "node:fs"; +import { mkdirSync, mkdtempSync, readdirSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { git } from "../../../src/git.ts"; @@ -72,3 +72,36 @@ test("two processes snapshot one fresh folder together", async () => { rmSync(root, { recursive: true, force: true }); } }); + +test("a stop while the snapshot store is held by another ends the request at once, with no snapshot taken", async () => { + const { createModels, fauxProvider } = await import("@earendil-works/pi-ai"); + const { createSession } = await import("../src/session.ts"); + const { lockSession } = await import("../src/session-lock.ts"); + const { tesotaHome } = await import("../../../src/home.ts"); + const { scriptedReply, accepts } = await import("./scripted.ts"); + const root = mkdtempSync(join(tmpdir(), "tesota-core-snapshot-")); + writeFileSync(join(root, "notes.txt"), "hello\n"); + mkdirSync(tesotaHome(), { recursive: true }); + // Another session's snapshot holds the store, which a request waits for up to 30 s. + const held = lockSession(join(tesotaHome(), "snapshots.lock")); + const faux = fauxProvider(); + faux.setResponses(["done"].map(scriptedReply)); + const models = createModels(); + models.setProvider(faux.provider); + const session = createSession({ cwd: root, folder: mkdtempSync(join(tmpdir(), "tesota-core-sessions-")), models, model: faux.getModel(), + verifier: accepts }); + try { + const run = session.prompt("go"); + await new Promise((resolve) => setTimeout(resolve, 300)); + const stopped = performance.now(); + await session.abort(); + await run; + expect(performance.now() - stopped).toBeLessThan(5_000); + expect(session.outcome()).toBe("aborted"); + expect(session.ledger()[0]?.base.units).toEqual([{ folder: "", kind: "refused", reason: "the run was stopped before the snapshot was taken" }]); + } finally { + held.release(); + await session.dispose(); + rmSync(root, { recursive: true, force: true }); + } +}, 60_000); diff --git a/src/host.ts b/src/host.ts index fa0a864ed..5af376105 100644 --- a/src/host.ts +++ b/src/host.ts @@ -11,7 +11,7 @@ export interface HostParts { /** The POSIX shell the project's checks run in: its path and arguments, and whether it reads the command from its input instead. */ readonly shellConfig: () => { shell: string; args: string[]; commandTransport?: "argv" | "stdin" }; /** Runs asynchronous work under an OS lock, waiting up to `waitMs` for another writer in this process or another. */ - readonly withLockWhile: (path: string, work: () => Promise, waitMs: number) => Promise; + readonly withLockWhile: (path: string, work: () => Promise, waitMs: number, signal?: AbortSignal) => Promise; } let given: HostParts | undefined; diff --git a/src/snapshot.ts b/src/snapshot.ts index 9ba72b8f0..ae533e0ff 100644 --- a/src/snapshot.ts +++ b/src/snapshot.ts @@ -106,7 +106,7 @@ export function homeOrDriveRoot(folder: string): boolean { * `ownFiles`. A home folder or a drive's root is never snapshotted. */ export async function takeSnapshot(root: string, ownFiles: boolean, withLockWhile: HostParts["withLockWhile"], - limits: SnapshotLimits = SNAPSHOT_LIMITS): Promise { + limits: SnapshotLimits = SNAPSHOT_LIMITS, signal?: AbortSignal): Promise { const at = resolve(root); if (homeOrDriveRoot(at)) return { refused: "Tesota does not snapshot a home folder or a drive's root" }; const tooLarge = scan(at, ownFiles, limits); @@ -115,7 +115,12 @@ export async function takeSnapshot(root: string, ownFiles: boolean, withLockWhil const gitDir = join(snapshots, createHash("sha256").update(`${at}\0${ownFiles}`).digest("hex").slice(0, 16)); mkdirSync(snapshots, { recursive: true }); // One writer for the store (#461): pruning also removes other folders. The lock stays outside what it prunes. + // A stop ends the wait for the store, and no Git step starts after it; one under way is short and runs to its end. + const stopped = { refused: "the run was stopped before the snapshot was taken" }; + const isStopped = (): boolean => signal?.aborted === true; + if (isStopped()) return stopped; return withLockWhile(join(tesotaHome(), "snapshots.lock"), async () => { + if (isStopped()) return stopped; useSnapshotGit(at, gitDir); if (!existsSync(join(gitDir, "HEAD"))) { mkdirSync(gitDir, { recursive: true }); @@ -126,11 +131,15 @@ export async function takeSnapshot(root: string, ownFiles: boolean, withLockWhil writeFileSync(join(gitDir, "info", "exclude"), [...excludedFolders.map((name) => `${name}/`), ".env", ".env.*", ...ownFiles ? ["/*/"] : [], ...tooLarge.map(exactly)].join("\n")); writeFileSync(join(gitDir, "tesota-used"), at); + if (isStopped()) return stopped; const committed = await git(at, ["-c", "advice.addEmbeddedRepo=false", "add", "-A"]) !== undefined && await git(at, ["-c", "user.name=Tesota", "-c", "user.email=tesota@localhost", "commit", "-q", "--allow-empty", "--no-verify", "-m", "Tesota snapshot"]) !== undefined; const base = committed ? (await git(at, ["rev-parse", "HEAD"]))?.trim() : undefined; prune(snapshots); return base === undefined || base === "" ? { refused: "Git could not make a snapshot of it" } : { base, tooLarge }; - }, 30_000); + }, 30_000, signal).catch((error: unknown) => { + if (isStopped()) return stopped; + throw error; + }); }