From 2b5ee346d6a49b783033256d4e3a8ae04949fb20 Mon Sep 17 00:00:00 2001 From: Yash Date: Sat, 5 Sep 2026 12:12:00 +0000 Subject: [PATCH] feat(harness): persist bootstrap claims [Agent Map 05/15] --- .changeset/bootstrap-storage.md | 5 + .../src/core/project-bootstrap-outbox.test.ts | 181 +++ .../src/core/project-bootstrap-outbox.ts | 182 +++ .../src/core/project-bootstrap-store.test.ts | 567 ++++++++ .../src/core/project-bootstrap-store.ts | 1294 +++++++++++++++++ .../src/core/studio-project-catalog.ts | 39 +- packages/harness/src/index.ts | 1 + packages/harness/src/shared/agent-map.ts | 81 ++ 8 files changed, 2349 insertions(+), 1 deletion(-) create mode 100644 .changeset/bootstrap-storage.md create mode 100644 packages/harness/src/core/project-bootstrap-outbox.test.ts create mode 100644 packages/harness/src/core/project-bootstrap-outbox.ts create mode 100644 packages/harness/src/core/project-bootstrap-store.test.ts create mode 100644 packages/harness/src/core/project-bootstrap-store.ts diff --git a/.changeset/bootstrap-storage.md b/.changeset/bootstrap-storage.md new file mode 100644 index 00000000..8ecf9f68 --- /dev/null +++ b/.changeset/bootstrap-storage.md @@ -0,0 +1,5 @@ +--- +"@sapiom/harness": patch +--- + +Internal storage groundwork for automatic Agent Map bootstrap. Clean up temporary state after failed writes and ignore unrelated files when reading durable project intents. No user-facing behavior changes in this release. diff --git a/packages/harness/src/core/project-bootstrap-outbox.test.ts b/packages/harness/src/core/project-bootstrap-outbox.test.ts new file mode 100644 index 00000000..8acd55e4 --- /dev/null +++ b/packages/harness/src/core/project-bootstrap-outbox.test.ts @@ -0,0 +1,181 @@ +import { randomUUID } from "node:crypto"; +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import * as path from "node:path"; + +import { afterEach, describe, expect, it } from "vitest"; + +import { + ProjectBootstrapOutbox, + ProjectBootstrapOutboxError, +} from "./project-bootstrap-outbox.js"; +import { StudioProjectCatalog } from "./studio-project-catalog.js"; + +describe("ProjectBootstrapOutbox", () => { + const roots: string[] = []; + + afterEach(async () => { + await Promise.all( + roots + .splice(0) + .map((root) => fs.rm(root, { recursive: true, force: true })), + ); + }); + + async function fixture() { + const root = await fs.mkdtemp( + path.join(os.tmpdir(), "project-bootstrap-outbox-"), + ); + roots.push(root); + const outbox = new ProjectBootstrapOutbox(path.join(root, "outbox")); + const lifecycle = { + beforeProjectsCreatedCommit: ( + projects: Parameters[0], + ) => outbox.stage(projects), + }; + return { + root, + outbox, + lifecycle, + catalogPath: path.join(root, "studio-projects.json"), + }; + } + + it("stages an explicit project before its catalog commit survives a restart", async () => { + const { catalogPath, lifecycle, outbox } = await fixture(); + const catalog = new StudioProjectCatalog( + catalogPath, + undefined, + undefined, + lifecycle, + ); + + const project = await catalog.create("Explicit project"); + const restarted = new ProjectBootstrapOutbox( + path.join(path.dirname(catalogPath), "outbox"), + ); + + expect(await restarted.pending()).toEqual([ + { + projectId: project.projectId, + projectCreatedAt: project.createdAt, + }, + ]); + await outbox.complete(project.projectId); + expect(await restarted.pending()).toEqual([]); + }); + + it("ignores a strict stale writer temporary without blocking a valid marker", async () => { + const { catalogPath, lifecycle } = await fixture(); + const catalog = new StudioProjectCatalog( + catalogPath, + undefined, + undefined, + lifecycle, + ); + const project = await catalog.create("Interrupted writer"); + const outboxRoot = path.join(path.dirname(catalogPath), "outbox"); + const staleTemporary = path.join( + outboxRoot, + `${project.projectId}.json.tmp-123-${randomUUID()}`, + ); + await fs.writeFile(staleTemporary, "partial marker", { mode: 0o600 }); + + const restarted = new ProjectBootstrapOutbox(outboxRoot); + await expect(restarted.pending()).resolves.toEqual([ + { + projectId: project.projectId, + projectCreatedAt: project.createdAt, + }, + ]); + await expect(fs.stat(staleTemporary)).resolves.toBeDefined(); + }); + + it("ignores unrelated directory entries without blocking a valid marker or deleting them", async () => { + const { root, outbox, catalogPath, lifecycle } = await fixture(); + const catalog = new StudioProjectCatalog(catalogPath, undefined, undefined, lifecycle); + const project = await catalog.create("Pending bootstrap"); + const outboxRoot = path.join(root, "outbox"); + const unrelated = [".DS_Store", "notes.txt", "project-marker.tmp-unknown"]; + for (const name of unrelated) { + await fs.writeFile(path.join(outboxRoot, name), "unrelated", { mode: 0o600 }); + } + await fs.mkdir(path.join(outboxRoot, "backups")); + + await expect(outbox.pending()).resolves.toEqual([{ + projectId: project.projectId, + projectCreatedAt: project.createdAt, + }]); + for (const name of [...unrelated, "backups"]) { + await expect(fs.stat(path.join(outboxRoot, name))).resolves.toBeDefined(); + } + }); + + it.each(["project_invalid.json", `project_${randomUUID()}.json`])( + "fails closed on malformed reserved project marker %s", + async (name) => { + const { root, outbox } = await fixture(); + const outboxRoot = path.join(root, "outbox"); + await fs.mkdir(outboxRoot, { recursive: true }); + const file = path.join(outboxRoot, name); + await fs.writeFile(file, "malformed reserved state", { mode: 0o600 }); + + await expect(outbox.pending()).rejects.toBeInstanceOf(ProjectBootstrapOutboxError); + await expect(fs.stat(file)).resolves.toBeDefined(); + }, + ); + + it("stages only reconcile-created projects and never enrolls a legacy catalog project", async () => { + const { root, catalogPath, lifecycle, outbox } = await fixture(); + const legacyRoot = path.join(root, "legacy-project"); + const newRoot = path.join(root, "new-project"); + await Promise.all([fs.mkdir(legacyRoot), fs.mkdir(newRoot)]); + const legacy = await new StudioProjectCatalog(catalogPath).reconcile([ + { workspaceKey: "legacy-root", cwd: legacyRoot }, + ]); + const legacyProjectId = legacy.projects[0]!.projectId; + + const catalog = new StudioProjectCatalog( + catalogPath, + undefined, + undefined, + lifecycle, + ); + const reconciled = await catalog.reconcile([ + { workspaceKey: "legacy-root", cwd: legacyRoot }, + { workspaceKey: "new-root", cwd: newRoot }, + ]); + const newProjectId = reconciled.workspaceScopes.find( + (scope) => scope.cwd === newRoot, + )!.projectId!; + + expect(newProjectId).not.toBe(legacyProjectId); + expect(await outbox.pending()).toEqual([ + expect.objectContaining({ projectId: newProjectId }), + ]); + expect( + (await outbox.pending()).some( + (entry) => entry.projectId === legacyProjectId, + ), + ).toBe(false); + }); + + it("aborts catalog creation when the write-ahead marker cannot commit", async () => { + const { catalogPath } = await fixture(); + const catalog = new StudioProjectCatalog( + catalogPath, + undefined, + undefined, + { + beforeProjectsCreatedCommit: async () => { + throw new Error("simulated outbox outage"); + }, + }, + ); + + await expect(catalog.create("Must remain absent")).rejects.toThrow( + "simulated outbox outage", + ); + expect(await new StudioProjectCatalog(catalogPath).list()).toEqual([]); + }); +}); diff --git a/packages/harness/src/core/project-bootstrap-outbox.ts b/packages/harness/src/core/project-bootstrap-outbox.ts new file mode 100644 index 00000000..d65dd64d --- /dev/null +++ b/packages/harness/src/core/project-bootstrap-outbox.ts @@ -0,0 +1,182 @@ +import { randomUUID } from "node:crypto"; +import * as fs from "node:fs/promises"; +import * as path from "node:path"; + +import type { + StudioProjectId, + StudioProjectSummary, +} from "../shared/agent-map.js"; +import { isStudioProjectId } from "./studio-project-catalog.js"; + +interface PersistedProjectBootstrapOutboxEntry { + schemaVersion: 1; + projectId: StudioProjectId; + projectCreatedAt: string; +} + +const OUTBOX_TEMP_FILE_RE = + /^(project_[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})\.json\.tmp-[1-9][0-9]*-([0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})$/; + +export interface ProjectBootstrapOutboxEntry { + projectId: StudioProjectId; + projectCreatedAt: string; +} + +export class ProjectBootstrapOutboxError extends Error { + readonly code = "project_bootstrap_outbox_unavailable"; + + constructor() { + super("project bootstrap outbox is unavailable"); + this.name = "ProjectBootstrapOutboxError"; + } +} + +function isTimestamp(value: unknown): value is string { + if (typeof value !== "string") return false; + try { + return new Date(value).toISOString() === value; + } catch { + return false; + } +} + +function parseEntry( + value: unknown, + expectedProjectId: StudioProjectId, +): ProjectBootstrapOutboxEntry | null { + if ( + typeof value !== "object" || + value === null || + Array.isArray(value) || + Object.keys(value).sort().join(",") !== + "projectCreatedAt,projectId,schemaVersion" || + !("schemaVersion" in value) || + value.schemaVersion !== 1 || + !("projectId" in value) || + value.projectId !== expectedProjectId || + !("projectCreatedAt" in value) || + !isTimestamp(value.projectCreatedAt) + ) { + return null; + } + return { + projectId: expectedProjectId, + projectCreatedAt: value.projectCreatedAt, + }; +} + +/** + * Write-ahead marker for the catalog -> bootstrap-intent boundary. + * + * A marker is committed before a new Studio project enters the catalog. The + * marker is removed only after ProjectBootstrapCoordinator has durably + * scheduled that project. Therefore either side of a process crash is safe: + * an orphan marker has no catalog project and can be discarded, while a + * committed project with a marker is recovered without guessing that older + * catalog projects should be enrolled. + */ +export class ProjectBootstrapOutbox { + private readonly root: string; + + constructor(root: string) { + this.root = path.resolve(root); + } + + private file(projectId: StudioProjectId): string { + if (!isStudioProjectId(projectId)) throw new ProjectBootstrapOutboxError(); + const file = path.resolve(this.root, `${projectId}.json`); + if (!file.startsWith(`${this.root}${path.sep}`)) { + throw new ProjectBootstrapOutboxError(); + } + return file; + } + + async stage( + projects: readonly Pick[], + ): Promise { + try { + await fs.mkdir(this.root, { recursive: true, mode: 0o700 }); + for (const project of projects) { + const file = this.file(project.projectId); + try { + const existing = parseEntry( + JSON.parse(await fs.readFile(file, "utf8")) as unknown, + project.projectId, + ); + if (!existing || existing.projectCreatedAt !== project.createdAt) { + throw new ProjectBootstrapOutboxError(); + } + continue; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + } + const entry: PersistedProjectBootstrapOutboxEntry = { + schemaVersion: 1, + projectId: project.projectId, + projectCreatedAt: project.createdAt, + }; + const temporary = `${file}.tmp-${process.pid}-${randomUUID()}`; + try { + await fs.writeFile(temporary, `${JSON.stringify(entry, null, 2)}\n`, { + encoding: "utf8", + mode: 0o600, + }); + await fs.rename(temporary, file); + } finally { + await fs.rm(temporary, { force: true }).catch(() => {}); + } + } + } catch (error) { + if (error instanceof ProjectBootstrapOutboxError) throw error; + throw new ProjectBootstrapOutboxError(); + } + } + + async pending(): Promise { + try { + const names = await fs.readdir(this.root); + const entries: ProjectBootstrapOutboxEntry[] = []; + for (const name of names.sort()) { + const temporary = OUTBOX_TEMP_FILE_RE.exec(name); + if (temporary && isStudioProjectId(temporary[1])) { + // A process may die after writing a private temporary marker but + // before its atomic rename. The corresponding catalog transaction + // cannot have committed yet. Ignore this exact writer-owned shape; + // deleting it could race another process that still owns the active + // catalog transaction. + continue; + } + // Desktop metadata and unrelated files do not describe project work. + // Only the reserved committed-marker namespace can block recovery. + if (!name.startsWith("project_") || !name.endsWith(".json")) continue; + const match = /^(project_[0-9a-f-]+)\.json$/.exec(name); + if (!match || !isStudioProjectId(match[1])) { + throw new ProjectBootstrapOutboxError(); + } + const projectId = match[1]; + const entry = parseEntry( + JSON.parse( + await fs.readFile(this.file(projectId), "utf8"), + ) as unknown, + projectId, + ); + if (!entry) throw new ProjectBootstrapOutboxError(); + entries.push(entry); + } + return entries; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return []; + if (error instanceof ProjectBootstrapOutboxError) throw error; + throw new ProjectBootstrapOutboxError(); + } + } + + async complete(projectId: StudioProjectId): Promise { + try { + await fs.rm(this.file(projectId), { force: true }); + } catch (error) { + if (error instanceof ProjectBootstrapOutboxError) throw error; + throw new ProjectBootstrapOutboxError(); + } + } +} diff --git a/packages/harness/src/core/project-bootstrap-store.test.ts b/packages/harness/src/core/project-bootstrap-store.test.ts new file mode 100644 index 00000000..8a6807ae --- /dev/null +++ b/packages/harness/src/core/project-bootstrap-store.test.ts @@ -0,0 +1,567 @@ +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import * as path from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { + ProjectAgentSession, + ProjectBootstrapMetadata, + ProjectBootstrapState, +} from "../shared/agent-map.js"; +import type { HarnessSession } from "../shared/types.js"; +import { + ProjectBootstrapStore, + ProjectBootstrapDispatchForbiddenError, + type ProjectBootstrapStoreOptions, +} from "./project-bootstrap-store.js"; + +vi.mock("node:fs/promises", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + writeFile: vi.fn(actual.writeFile), + rename: vi.fn(actual.rename), + }; +}); + +class TestBootstrapStore extends ProjectBootstrapStore { + read(session: HarnessSession, emptyProject = true) { + return this.load(session, emptyProject); + } + + save(session: HarnessSession) { + return this.writeState(this.file(session.id), this.newState(session, true)); + } +} + +const PROJECT_ID = "project_00000000-0000-7000-8000-000000000001"; + +const USER_ID = "user-1"; + +const NOW = "2026-09-01T00:00:00.000Z"; + +interface DurableBootstrapState { + schemaVersion: number; + metadata: ProjectBootstrapMetadata; + inputs: Array<{ + id: string; + sessionId: string; + text: string; + acceptedAt: string; + }>; + dispatchingInputId: string | null; + retryCount: number; + emptyProject: boolean; + attempts: Array<{ + attemptId: string; + retryOrdinal: number; + status: "active" | "retired" | "completed"; + phase?: "claimed" | "dispatching" | "not-submitted" | "submitted"; + }>; + uncertainInputIds?: string[]; + uncertainInputs?: Array<{ + id: string; + sessionId: string; + text: string; + acceptedAt: string; + }>; + receipts?: Array<{ + requestId: string | null; + inputId: string; + status: "queued" | "submitted" | "uncertain" | "completed"; + acceptedAt: string; + payloadDigest: string; + }>; +} + +function projectSession( + id = "session-1", + bootstrap: ProjectBootstrapState = { status: "pending" }, +): HarnessSession { + const identity: ProjectAgentSession = { + projectId: PROJECT_ID, + sessionId: id, + userId: USER_ID, + }; + return { + id, + agentSessionId: "provider-conversation-1", + harness: "codex", + cwd: "/private/project", + title: "Plan Agents", + status: "running", + createdAt: NOW, + lastActiveAt: NOW, + exitCode: null, + boundWorkflowPath: null, + ready: true, + agentMapIdentity: identity, + projectBootstrap: { + projectId: identity.projectId, + userId: identity.userId, + targetSessionId: identity.sessionId, + bootstrap: structuredClone(bootstrap), + queuedInputIds: [], + }, + }; +} + +function stateFile(root: string, sessionId: string): string { + return path.join(root, sessionId, "input-queue.json"); +} + +async function readState( + root: string, + sessionId: string, +): Promise { + return JSON.parse( + await fs.readFile(stateFile(root, sessionId), "utf8"), + ) as DurableBootstrapState; +} + +async function writeState( + root: string, + sessionId: string, + state: DurableBootstrapState, +): Promise { + await fs.mkdir(path.dirname(stateFile(root, sessionId)), { recursive: true }); + await fs.writeFile(stateFile(root, sessionId), `${JSON.stringify(state)}\n`); +} +describe("ProjectBootstrapStore", () => { + let root: string; + let legacyRoot: string; + let session: HarnessSession; + let sessions: Map; + let manager: ProjectBootstrapStoreOptions["sessionManager"]; + beforeEach(async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "project-bootstrap-store-")); + legacyRoot = await fs.mkdtemp( + path.join(os.tmpdir(), "project-bootstrap-store-legacy-"), + ); + session = projectSession(); + sessions = new Map([[session.id, session]]); + manager = { + get: (id: string) => sessions.get(id), + setProjectBootstrapMetadata: async ( + id: string, + metadata: ProjectBootstrapMetadata, + ) => { + const target = sessions.get(id); + if (!target) throw new Error("session missing"); + target.projectBootstrap = structuredClone(metadata); + }, + }; + }); + afterEach(async () => { + await fs.rm(root, { recursive: true, force: true }); + await fs.rm(legacyRoot, { recursive: true, force: true }); + }); + + it.each(["write", "rename"] as const)( + "removes private temporary state after a failed %s without replacing durable state", + async (phase) => { + const file = stateFile(root, session.id); + await fs.mkdir(path.dirname(file), { recursive: true }); + await fs.writeFile(file, "last durable state"); + const failure = new Error(`injected ${phase} failure`); + if (phase === "write") { + const actual = await vi.importActual("node:fs/promises"); + vi.mocked(fs.writeFile).mockImplementationOnce(async (...args) => { + await actual.writeFile(...args); + throw failure; + }); + } else { + vi.mocked(fs.rename).mockRejectedValueOnce(failure); + } + const store = new TestBootstrapStore({ root, sessionManager: manager }); + + await expect(store.save(session)).rejects.toBe(failure); + + expect(await fs.readdir(path.dirname(file))).toEqual(["input-queue.json"]); + expect(await fs.readFile(file, "utf8")).toBe("last durable state"); + }, + ); + + it("durably schedules one project lifecycle and atomically claims its first ordinary session", async () => { + const first = new TestBootstrapStore({ + root, + sessionManager: manager, + }); + + await expect(first.scheduleProject(PROJECT_ID, USER_ID)).resolves.toBe( + true, + ); + await expect(first.scheduleProject(PROJECT_ID, USER_ID)).resolves.toBe( + false, + ); + + const restarted = new TestBootstrapStore({ + root, + sessionManager: manager, + }); + await expect( + restarted.needsProjectSession(PROJECT_ID, USER_ID), + ).resolves.toBe(true); + + const claimed = await restarted.claimProject(session.agentMapIdentity!); + expect(claimed).toEqual({ + projectId: PROJECT_ID, + userId: USER_ID, + targetSessionId: session.id, + bootstrap: { status: "pending" }, + queuedInputIds: [], + }); + session.projectBootstrap = claimed!; + await expect( + restarted.needsProjectSession(PROJECT_ID, USER_ID), + ).resolves.toBe(false); + + const second = projectSession("session-2"); + sessions.set(second.id, second); + await expect( + restarted.claimProject(second.agentMapIdentity!), + ).resolves.toBeNull(); + + const intent = JSON.parse( + await fs.readFile( + path.join(root, "projects", `${PROJECT_ID}.json`), + "utf8", + ), + ) as Record; + expect(intent).toMatchObject({ + schemaVersion: 1, + projectId: PROJECT_ID, + userId: USER_ID, + targetSessionId: session.id, + status: "claimed", + }); + expect(JSON.stringify(intent)).not.toContain(session.cwd); + }); + + it("does not let a concurrent create steal a claim before SessionManager publishes its session", async () => { + sessions.delete(session.id); + const coordinator = new TestBootstrapStore({ + root, + sessionManager: manager, + }); + await coordinator.scheduleProject(PROJECT_ID, USER_ID); + + const first = await coordinator.claimProject(session.agentMapIdentity!); + expect(first?.targetSessionId).toBe(session.id); + + const racing = projectSession("session-racing-create"); + expect(await coordinator.claimProject(racing.agentMapIdentity!)).toBeNull(); + await expect( + coordinator.needsProjectSession(PROJECT_ID, USER_ID), + ).resolves.toBe(false); + + // A proven pre-spawn failure releases only the volatile claim. The durable + // project intent remains available for a replacement ordinary session. + await coordinator.releaseSessionClaim(session.id); + expect( + await coordinator.claimProject(racing.agentMapIdentity!), + ).toMatchObject({ targetSessionId: racing.id }); + }); + + it("rejects a foreign project-intent claimant and can recover a missing claimed target", async () => { + const coordinator = new TestBootstrapStore({ + root, + sessionManager: manager, + }); + await coordinator.scheduleProject(PROJECT_ID, USER_ID); + await expect( + coordinator.claimProject({ + ...session.agentMapIdentity!, + userId: "foreign-user", + }), + ).rejects.toBeInstanceOf(ProjectBootstrapDispatchForbiddenError); + + const first = await coordinator.claimProject(session.agentMapIdentity!); + session.projectBootstrap = first!; + await coordinator.releaseSessionClaim(session.id); + sessions.delete(session.id); + const replacement = projectSession("session-replacement"); + sessions.set(replacement.id, replacement); + + const recovered = await coordinator.claimProject( + replacement.agentMapIdentity!, + ); + expect(recovered?.targetSessionId).toBe(replacement.id); + expect(recovered?.bootstrap).toEqual({ status: "pending" }); + }); + + it("keeps a published pre-provider exit as the bounded project failure tombstone", async () => { + const coordinator = new TestBootstrapStore({ + root, + sessionManager: manager, + }); + await coordinator.scheduleProject(PROJECT_ID, USER_ID); + const first = await coordinator.claimProject(session.agentMapIdentity!); + session.projectBootstrap = first!; + session.status = "exited"; + session.agentSessionId = null; + + await expect( + coordinator.needsProjectSession(PROJECT_ID, USER_ID), + ).resolves.toBe(false); + + const replacement = projectSession("session-replacement"); + sessions.set(replacement.id, replacement); + await expect( + coordinator.claimProject(replacement.agentMapIdentity!), + ).resolves.toBeNull(); + + const restarted = new TestBootstrapStore({ + root, + sessionManager: manager, + }); + for (let read = 0; read < 10; read += 1) { + await expect( + restarted.needsProjectSession(PROJECT_ID, USER_ID), + ).resolves.toBe(false); + } + + expect(sessions.get(session.id)).toBe(session); + await expect( + fs + .readFile(path.join(root, "projects", `${PROJECT_ID}.json`), "utf8") + .then(JSON.parse), + ).resolves.toMatchObject({ + schemaVersion: 1, + projectId: PROJECT_ID, + targetSessionId: session.id, + status: "claimed", + }); + }); + + it("refuses replacement when an abandoned target still owns durable input", async () => { + const coordinator = new TestBootstrapStore({ + root, + sessionManager: manager, + }); + await coordinator.scheduleProject(PROJECT_ID, USER_ID); + const first = await coordinator.claimProject(session.agentMapIdentity!); + session.projectBootstrap = first!; + await writeState(root, session.id, { + schemaVersion: 2, + metadata: { + ...structuredClone(first!), + bootstrap: { status: "skipped", reason: "user-proceeded" }, + queuedInputIds: ["durable-user-input"], + }, + inputs: [ + { + id: "durable-user-input", + sessionId: session.id, + text: "preserve this exact request", + acceptedAt: NOW, + }, + ], + dispatchingInputId: null, + retryCount: 0, + emptyProject: true, + attempts: [], + }); + session.status = "exited"; + session.agentSessionId = null; + + await expect( + coordinator.needsProjectSession(PROJECT_ID, USER_ID), + ).resolves.toBe(false); + const replacement = projectSession("session-replacement-refused"); + sessions.set(replacement.id, replacement); + await expect( + coordinator.claimProject(replacement.agentMapIdentity!), + ).resolves.toBeNull(); + + expect((await readState(root, session.id)).inputs).toEqual([ + expect.objectContaining({ + id: "durable-user-input", + sessionId: session.id, + text: "preserve this exact request", + }), + ]); + const intent = JSON.parse( + await fs.readFile( + path.join(root, "projects", `${PROJECT_ID}.json`), + "utf8", + ), + ) as { targetSessionId: string }; + expect(intent.targetSessionId).toBe(session.id); + }); + + it("records real input already pending at claim time as higher priority", async () => { + const coordinator = new TestBootstrapStore({ + root, + sessionManager: manager, + }); + await coordinator.scheduleProject(PROJECT_ID, USER_ID); + + const claimed = await coordinator.claimProject( + session.agentMapIdentity!, + true, + ); + + expect(claimed?.bootstrap).toEqual({ + status: "skipped", + reason: "user-proceeded", + }); + }); + + it("fails closed on malformed primary state without deleting, replacing, or quarantining it", async () => { + const directory = path.join(root, session.id); + await fs.mkdir(directory, { recursive: true }); + await fs.writeFile( + stateFile(root, session.id), + "{private-undelivered-input", + ); + session.projectBootstrap!.queuedInputIds = ["unknown-undelivered-input"]; + const coordinator = new TestBootstrapStore({ + root, + sessionManager: manager, + }); + + await expect(coordinator.read(session, true)).rejects.toThrow( + "project bootstrap state is unavailable", + ); + expect(await fs.readFile(stateFile(root, session.id), "utf8")).toBe( + "{private-undelivered-input", + ); + expect(await fs.readdir(directory)).toEqual(["input-queue.json"]); + expect(session.projectBootstrap?.queuedInputIds).toEqual([ + "unknown-undelivered-input", + ]); + }); + + it("rejects a persisted receipt request ID beyond the public 200-character bound", async () => { + session.projectBootstrap!.bootstrap = { + status: "delivered", + messageId: "bootstrap-complete", + }; + await writeState(root, session.id, { + schemaVersion: 3, + metadata: structuredClone(session.projectBootstrap!), + inputs: [], + dispatchingInputId: null, + retryCount: 0, + emptyProject: false, + attempts: [], + uncertainInputIds: [], + uncertainInputs: [], + receipts: [ + { + requestId: "r".repeat(201), + inputId: "completed-input", + status: "completed", + acceptedAt: NOW, + payloadDigest: "a".repeat(64), + }, + ], + }); + const original = await fs.readFile(stateFile(root, session.id), "utf8"); + const coordinator = new TestBootstrapStore({ + root, + sessionManager: manager, + }); + + await expect(coordinator.read(session, false)).rejects.toThrow( + "project bootstrap state is unavailable", + ); + expect(await fs.readFile(stateFile(root, session.id), "utf8")).toBe( + original, + ); + }); + + it("rejects a session identity that could escape the bootstrap root", async () => { + session = projectSession("../escape"); + sessions = new Map([[session.id, session]]); + const coordinator = new TestBootstrapStore({ + root, + sessionManager: manager, + }); + + await expect(coordinator.read(session, true)).rejects.toThrow( + "project bootstrap state is unavailable", + ); + await expect(fs.readdir(root)).resolves.toEqual([]); + }); + + it("migrates a planner-era schema-1 FIFO in place without quarantine or input loss", async () => { + session.ready = false; + const legacyDirectory = path.join(legacyRoot, session.id); + await fs.mkdir(legacyDirectory, { recursive: true }); + await fs.writeFile( + path.join(legacyDirectory, "input-queue.json"), + `${JSON.stringify({ + schemaVersion: 1, + metadata: { + identity: { + projectId: PROJECT_ID, + userId: USER_ID, + sessionId: session.id, + role: "map-planner", + }, + greeting: { status: "delivered", messageId: "legacy-greeting" }, + queuedInputIds: ["legacy-input-1", "legacy-input-2"], + }, + inputs: [ + { + id: "legacy-input-1", + sessionId: session.id, + text: "first durable user request", + acceptedAt: NOW, + }, + { + id: "legacy-input-2", + sessionId: session.id, + text: "second durable user request", + acceptedAt: NOW, + }, + ], + dispatchingInputId: null, + retryCount: 0, + emptyProject: true, + // Schema 1 never defined this field. Migration must ignore it rather + // than accepting forged keyed receipt authority. + receipts: [ + { + requestId: "forged-legacy-key", + inputId: "legacy-input-1", + status: "queued", + acceptedAt: NOW, + payloadDigest: "f".repeat(64), + }, + ], + })}\n`, + ); + + const coordinator = new TestBootstrapStore({ + root, + legacyStateRoot: legacyRoot, + sessionManager: manager, + }); + await coordinator.read(session); + + const migrated = await readState(root, session.id); + expect(migrated).toMatchObject({ + schemaVersion: 3, + metadata: { + projectId: PROJECT_ID, + userId: USER_ID, + targetSessionId: session.id, + bootstrap: { status: "delivered", messageId: "legacy-greeting" }, + queuedInputIds: ["legacy-input-1", "legacy-input-2"], + }, + }); + expect(migrated.metadata).not.toHaveProperty("identity"); + expect(migrated.metadata).not.toHaveProperty("greeting"); + expect(migrated.receipts).toHaveLength(2); + expect(migrated.receipts?.map((receipt) => receipt.requestId)).toEqual([ + null, + null, + ]); + expect( + new Set(migrated.receipts?.map((receipt) => receipt.inputId)).size, + ).toBe(2); + expect(await fs.readdir(legacyDirectory)).toEqual(["input-queue.json"]); + }); +}); diff --git a/packages/harness/src/core/project-bootstrap-store.ts b/packages/harness/src/core/project-bootstrap-store.ts new file mode 100644 index 00000000..5e6fc03b --- /dev/null +++ b/packages/harness/src/core/project-bootstrap-store.ts @@ -0,0 +1,1294 @@ +import { createHash, randomUUID } from "node:crypto"; +import * as fs from "node:fs/promises"; +import * as path from "node:path"; + +import type { + ProjectBootstrapLifecycleEvent, + ProjectBootstrapInputReceipt, + ProjectBootstrapQueuedInput, + ProjectBootstrapMetadata, + ProjectAgentSession, +} from "../shared/agent-map.js"; +import type { HarnessSession } from "../shared/types.js"; +import type { SessionManager } from "./session-manager.js"; + +export type ProjectBootstrapAttemptPhase = + | "claimed" + | "dispatching" + | "not-submitted" + | "submitted"; + +export interface PersistedProjectBootstrapInputReceipt extends ProjectBootstrapInputReceipt { + payloadDigest: string; +} + +export interface PersistedProjectBootstrapState { + schemaVersion: 3; + metadata: ProjectBootstrapMetadata; + inputs: ProjectBootstrapQueuedInput[]; + /** + * Durable write-ahead intent for the one FIFO head that may be crossing the + * PTY boundary. An unresolved intent is never replayed automatically after a + * restart because the process cannot prove whether the PTY accepted it. + */ + dispatchingInputId: string | null; + retryCount: number; + emptyProject: boolean; + attempts: Array<{ + attemptId: string; + retryOrdinal: number; + status: "active" | "retired" | "completed"; + /** `dispatching` is written before the first PTY byte and is therefore + * conservatively uncertain after process loss. `not-submitted` is written + * only when SessionManager positively proves Enter was never attempted. */ + phase: ProjectBootstrapAttemptPhase; + }>; + /** IDs retained for schema-2 compatibility and bounded inspection. */ + uncertainInputIds: string[]; + /** + * Durable, content-bearing tombstones for FIFO entries whose PTY acceptance + * could not be proven. They are removed from the dispatchable FIFO so later + * user input can progress, but are never replayed or discarded. + */ + uncertainInputs: ProjectBootstrapQueuedInput[]; + /** Session-scoped idempotency receipts for the bounded bootstrap FIFO. */ + receipts: PersistedProjectBootstrapInputReceipt[]; +} + +export interface AcceptedInputLedger { + schemaVersion: 1; + inputIds: string[]; +} + +export interface PersistedProjectBootstrapIntent { + schemaVersion: 1; + projectId: string; + userId: string; + targetSessionId: string | null; + status: "scheduled" | "claimed"; + createdAt: string; + updatedAt: string; +} + +export class ProjectBootstrapDispatchForbiddenError extends Error { + readonly code = "project_bootstrap_dispatch_forbidden"; + + constructor() { + super("project bootstrap is no longer authorized for this session"); + this.name = "ProjectBootstrapDispatchForbiddenError"; + } +} + +export class ProjectBootstrapCoordinatorClosedError extends Error { + readonly code = "project_bootstrap_coordinator_closed"; + + constructor() { + super("project bootstrap coordinator is closed"); + this.name = "ProjectBootstrapCoordinatorClosedError"; + } +} + +export class ProjectBootstrapInputCapacityError extends Error { + readonly code = "project_bootstrap_input_capacity"; + + constructor() { + super("project bootstrap input receipt capacity is temporarily full"); + this.name = "ProjectBootstrapInputCapacityError"; + } +} + +export const MAX_RETRIES = 2; + +export const MAX_INPUT_RECEIPTS = 128; + +/** + * Keep a bounded recent idempotency window. Entries that still own queued or + * submitted work are never evicted. Only completed unkeyed bookkeeping is + * retired; keyed receipts remain stable until the bounded store reaches + * capacity, at which point a new logical request fails before mutation. + */ +export function compactInputReceipts( + receipts: readonly PersistedProjectBootstrapInputReceipt[], + reserveSlots = 0, +): PersistedProjectBootstrapInputReceipt[] { + const limit = Math.max(0, MAX_INPUT_RECEIPTS - reserveSlots); + const compacted: PersistedProjectBootstrapInputReceipt[] = receipts.map( + (receipt) => structuredClone(receipt), + ); + while (compacted.length > limit) { + const index = compacted.findIndex( + (receipt) => receipt.status === "completed" && receipt.requestId === null, + ); + if (index < 0) break; + compacted.splice(index, 1); + } + if (compacted.length > limit) { + throw new ProjectBootstrapInputCapacityError(); + } + return compacted; +} + +export function projectBootstrapInputDigest(text: string): string { + return createHash("sha256") + .update(JSON.stringify({ schemaVersion: 1, submit: true, text })) + .digest("hex"); +} + +export function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +export function isTerminal(metadata: ProjectBootstrapMetadata): boolean { + return ( + metadata.bootstrap.status === "delivered" || + metadata.bootstrap.status === "skipped" + ); +} + +export function validBootstrapState(value: unknown): boolean { + if (!isRecord(value) || typeof value.status !== "string") return false; + switch (value.status) { + case "pending": + return true; + case "generating": + return typeof value.attemptId === "string" && value.attemptId !== ""; + case "delivered": + return typeof value.messageId === "string" && value.messageId !== ""; + case "failed": + return ( + typeof value.retryable === "boolean" && + typeof value.errorCode === "string" && + [ + "session_not_ready", + "session_exited", + "injection_failed", + "model_turn_failed", + "delivery_timeout", + "persistence_failed", + "scope_unavailable", + ].includes(value.errorCode) + ); + case "skipped": + return ( + value.reason === "user-proceeded" || value.reason === "map-not-empty" + ); + default: + return false; + } +} + +export function parsePersistedProjectBootstrapState( + value: unknown, + session: HarnessSession, +): PersistedProjectBootstrapState | null { + if (!isRecord(value) || !session.projectBootstrap) return null; + const metadata = value.metadata; + if (!isRecord(metadata)) return null; + const expected = session.projectBootstrap; + const legacyIdentity = isRecord(metadata.identity) ? metadata.identity : null; + const projectId = legacyIdentity?.projectId ?? metadata.projectId; + const userId = legacyIdentity?.userId ?? metadata.userId; + const targetSessionId = legacyIdentity?.sessionId ?? metadata.targetSessionId; + const bootstrap = metadata.bootstrap ?? metadata.greeting; + if ( + (value.schemaVersion !== 1 && + value.schemaVersion !== 2 && + value.schemaVersion !== 3) || + projectId !== expected.projectId || + userId !== expected.userId || + targetSessionId !== expected.targetSessionId || + !Array.isArray(metadata.queuedInputIds) || + !metadata.queuedInputIds.every((id) => typeof id === "string") || + !validBootstrapState(bootstrap) || + !Array.isArray(value.inputs) || + (value.dispatchingInputId !== undefined && + value.dispatchingInputId !== null && + typeof value.dispatchingInputId !== "string") || + !Number.isSafeInteger(value.retryCount) || + (value.retryCount as number) < 0 || + (value.retryCount as number) > MAX_RETRIES || + typeof value.emptyProject !== "boolean" + ) { + return null; + } + const inputs = value.inputs; + const storedUncertainInputs = Array.isArray(value.uncertainInputs) + ? value.uncertainInputs + : []; + const validInput = (input: unknown): input is ProjectBootstrapQueuedInput => + isRecord(input) && + typeof input.id === "string" && + input.id !== "" && + input.sessionId === session.id && + typeof input.text === "string" && + input.text.length <= 100_000 && + typeof input.acceptedAt === "string"; + if ( + !inputs.every(validInput) || + !storedUncertainInputs.every(validInput) || + new Set( + [...inputs, ...storedUncertainInputs].map( + (input) => (input as ProjectBootstrapQueuedInput).id, + ), + ).size !== + inputs.length + storedUncertainInputs.length || + metadata.queuedInputIds.length !== inputs.length || + metadata.queuedInputIds.some((id, index) => id !== inputs[index]?.id) || + (typeof value.dispatchingInputId === "string" && + value.dispatchingInputId !== inputs[0]?.id) + ) { + return null; + } + const sourceSchemaVersion = Number(value.schemaVersion); + const storedAttempts = Array.isArray(value.attempts) ? value.attempts : []; + const validAttempt = ( + attempt: unknown, + ): attempt is PersistedProjectBootstrapState["attempts"][number] => + isRecord(attempt) && + typeof attempt.attemptId === "string" && + attempt.attemptId !== "" && + Number.isSafeInteger(attempt.retryOrdinal) && + Number(attempt.retryOrdinal) >= 0 && + Number(attempt.retryOrdinal) <= MAX_RETRIES && + ["active", "retired", "completed"].includes(String(attempt.status)) && + (sourceSchemaVersion < 3 || + ["claimed", "dispatching", "not-submitted", "submitted"].includes( + String(attempt.phase), + )); + if ( + sourceSchemaVersion >= 3 && + (!Array.isArray(value.attempts) || + storedAttempts.length > 8 || + !storedAttempts.every(validAttempt) || + new Set( + storedAttempts.filter(validAttempt).map((attempt) => attempt.attemptId), + ).size !== storedAttempts.length || + new Set( + storedAttempts + .filter(validAttempt) + .map((attempt) => attempt.retryOrdinal), + ).size !== storedAttempts.length) + ) { + // Current-schema attempt evidence is retry authority. Never turn a missing, + // malformed, or duplicate entry into an apparently safe empty history. + return null; + } + const attempts = Array.isArray(value.attempts) + ? value.attempts.filter(validAttempt).map((attempt) => ({ + attemptId: attempt.attemptId, + retryOrdinal: attempt.retryOrdinal, + status: attempt.status, + // Schema 1/2 could already have crossed Enter and therefore migrates + // conservatively. Only schema 3 can prove a pre-PTY claim. + phase: + sourceSchemaVersion >= 3 + ? attempt.phase + : attempt.status === "completed" + ? "submitted" + : "dispatching", + })) + : []; + const persistedUncertainIds = Array.isArray(value.uncertainInputIds) + ? value.uncertainInputIds.filter( + (inputId): inputId is string => typeof inputId === "string", + ) + : []; + const legacyUncertainIds = new Set( + persistedUncertainIds.filter((inputId) => + inputs.some((input) => isRecord(input) && input.id === inputId), + ), + ); + const normalizedInputs = ( + structuredClone(value.inputs) as ProjectBootstrapQueuedInput[] + ).filter((input) => !legacyUncertainIds.has(input.id)); + const normalizedUncertainInputs = [ + ...(structuredClone( + storedUncertainInputs, + ) as ProjectBootstrapQueuedInput[]), + ...(structuredClone(value.inputs) as ProjectBootstrapQueuedInput[]).filter( + (input) => legacyUncertainIds.has(input.id), + ), + ]; + const uncertainInputIds = normalizedUncertainInputs.map((input) => input.id); + let normalizedBootstrap = structuredClone( + bootstrap, + ) as ProjectBootstrapMetadata["bootstrap"]; + if ( + sourceSchemaVersion < 3 && + normalizedBootstrap.status === "failed" && + normalizedBootstrap.retryable && + normalizedBootstrap.errorCode !== "session_not_ready" + ) { + // Schema 1/2 had no durable phase evidence. A legacy retryable flag cannot + // prove that an injection/persistence failure preceded Enter. + normalizedBootstrap = { ...normalizedBootstrap, retryable: false }; + } + if ( + sourceSchemaVersion < 3 && + attempts.length === 0 && + normalizedBootstrap.status === "generating" + ) { + attempts.push({ + attemptId: normalizedBootstrap.attemptId, + retryOrdinal: Math.max(0, Number(value.retryCount) || 0), + status: "active", + phase: "dispatching", + }); + } + if ( + sourceSchemaVersion >= 3 && + normalizedBootstrap.status === "generating" && + !attempts.some((attempt) => { + const activeAttemptId = + normalizedBootstrap.status === "generating" + ? normalizedBootstrap.attemptId + : null; + return ( + attempt.attemptId === activeAttemptId && attempt.status === "active" + ); + }) + ) { + return null; + } + if ( + normalizedInputs.length > 0 && + !isTerminal({ + projectId: expected.projectId, + userId: expected.userId, + targetSessionId: expected.targetSessionId, + bootstrap: normalizedBootstrap, + queuedInputIds: normalizedInputs.map((input) => input.id), + }) + ) { + if (sourceSchemaVersion >= 3) return null; + // Legacy planner queues could persist the FIFO before their greeting skip. + // The user input is authoritative, so migration completes that transition + // without changing IDs or message bodies. + normalizedBootstrap = { status: "skipped", reason: "user-proceeded" }; + } + if (sourceSchemaVersion >= 3 && !Array.isArray(value.receipts)) return null; + // Schema 1/2 never owned receipt authority. Ignore any injected property and + // derive canonical unkeyed receipts solely from the migrated FIFO/tombstone + // state so legacy data cannot mint an idempotency key. + const storedReceipts = + sourceSchemaVersion >= 3 && Array.isArray(value.receipts) + ? value.receipts + : []; + const validReceipts = storedReceipts.filter( + (receipt): receipt is Record => + isRecord(receipt) && + (receipt.requestId === null || + (typeof receipt.requestId === "string" && + receipt.requestId !== "" && + receipt.requestId.length <= 200)) && + typeof receipt.inputId === "string" && + receipt.inputId !== "" && + ["queued", "submitted", "uncertain", "completed"].includes( + String(receipt.status), + ) && + typeof receipt.acceptedAt === "string" && + typeof receipt.payloadDigest === "string" && + /^[0-9a-f]{64}$/.test(receipt.payloadDigest), + ); + const receipts: PersistedProjectBootstrapInputReceipt[] = validReceipts.map( + (receipt) => ({ + requestId: receipt.requestId as string | null, + inputId: receipt.inputId as string, + status: receipt.status as ProjectBootstrapInputReceipt["status"], + acceptedAt: receipt.acceptedAt as string, + payloadDigest: receipt.payloadDigest as string, + }), + ); + if (sourceSchemaVersion < 3) { + for (const input of normalizedInputs) { + receipts.push({ + requestId: null, + inputId: input.id, + status: "queued", + acceptedAt: input.acceptedAt, + payloadDigest: projectBootstrapInputDigest(input.text), + }); + } + for (const input of normalizedUncertainInputs) { + receipts.push({ + requestId: null, + inputId: input.id, + status: "uncertain", + acceptedAt: input.acceptedAt, + payloadDigest: projectBootstrapInputDigest(input.text), + }); + } + } + if ( + sourceSchemaVersion >= 3 && + (validReceipts.length !== storedReceipts.length || + new Set(receipts.map((receipt) => receipt.inputId)).size !== + receipts.length || + new Set( + receipts + .filter((receipt) => receipt.requestId !== null) + .map((receipt) => receipt.requestId), + ).size !== + receipts.filter((receipt) => receipt.requestId !== null).length) + ) { + return null; + } + if ( + sourceSchemaVersion >= 3 && + (normalizedInputs.some((input) => { + const receipt = receipts.find( + (candidate) => candidate.inputId === input.id, + ); + return ( + !receipt || + receipt.acceptedAt !== input.acceptedAt || + receipt.payloadDigest !== projectBootstrapInputDigest(input.text) || + receipt.status === "completed" || + receipt.status === "uncertain" + ); + }) || + normalizedUncertainInputs.some((input) => { + const receipt = receipts.find( + (candidate) => candidate.inputId === input.id, + ); + return ( + !receipt || + receipt.acceptedAt !== input.acceptedAt || + receipt.payloadDigest !== projectBootstrapInputDigest(input.text) || + receipt.status !== "uncertain" + ); + }) || + receipts.some( + (receipt) => + receipt.status === "queued" && + !normalizedInputs.some((input) => input.id === receipt.inputId), + ) || + // Receipt order is the durable logical-arrival order. Live FIFO rows may + // have terminal receipt-only predecessors after a crash, but the rows + // themselves must remain a monotonic subsequence so boot recovery can + // conservatively terminalize the causal prefix without guessing. + normalizedInputs.some((input, index) => { + if (index === 0) return false; + const priorIndex = receipts.findIndex( + (receipt) => receipt.inputId === normalizedInputs[index - 1]?.id, + ); + const currentIndex = receipts.findIndex( + (receipt) => receipt.inputId === input.id, + ); + return priorIndex < 0 || currentIndex <= priorIndex; + }) || + normalizedInputs.some( + (input, index) => + receipts.at(index - normalizedInputs.length)?.inputId !== input.id, + )) + ) { + return null; + } + let compactedReceipts: PersistedProjectBootstrapInputReceipt[]; + try { + compactedReceipts = compactInputReceipts(receipts); + } catch { + return null; + } + return { + schemaVersion: 3, + metadata: { + projectId: expected.projectId, + userId: expected.userId, + targetSessionId: expected.targetSessionId, + bootstrap: normalizedBootstrap, + queuedInputIds: normalizedInputs.map((input) => input.id), + }, + inputs: normalizedInputs, + dispatchingInputId: + typeof value.dispatchingInputId === "string" && + legacyUncertainIds.has(value.dispatchingInputId) + ? null + : (value.dispatchingInputId ?? null), + retryCount: Number(value.retryCount), + emptyProject: Boolean(value.emptyProject), + attempts: attempts.slice(-8), + uncertainInputIds, + uncertainInputs: normalizedUncertainInputs, + receipts: compactedReceipts, + }; +} +export interface ProjectBootstrapStoreOptions { + root: string; + legacyStateRoot?: string; + now?: () => string; + sessionManager: Pick; + writeState?: (file: string, state: unknown) => Promise; + writeAcceptedLedger?: (file: string, state: unknown) => Promise; + onEvent?: (event: ProjectBootstrapLifecycleEvent) => Promise | void; +} + +/** Durable project enrollment and input state shared by the bootstrap lifecycle. + * This storage boundary does not start sessions, timers, or model turns. */ +export class ProjectBootstrapStore { + constructor(protected readonly storageOptions: ProjectBootstrapStoreOptions) { + this.root = path.resolve(storageOptions.root); + this.legacyStateRoot = storageOptions.legacyStateRoot + ? path.resolve(storageOptions.legacyStateRoot) + : null; + this.now = storageOptions.now ?? (() => new Date().toISOString()); + } + + protected readonly root: string; + + protected readonly legacyStateRoot: string | null; + + protected readonly now: () => string; + + protected readonly states = new Map(); + + protected readonly writes = new Map>(); + + /** Project claims made before SessionManager publishes the new session. */ + protected readonly provisionalProjectClaims = new Map(); + + protected readonly provisionalSessionClaims = new Map(); + + protected closed = false; + + protected sessionDirectory(sessionId: string): string { + const directory = path.resolve(this.root, sessionId); + const rootPrefix = `${this.root}${path.sep}`; + if (!directory.startsWith(rootPrefix)) { + throw new Error("invalid project bootstrap storage identity"); + } + return directory; + } + + protected file(sessionId: string): string { + return path.join(this.sessionDirectory(sessionId), "input-queue.json"); + } + + protected legacyFile(sessionId: string, name: string): string | null { + if (!this.legacyStateRoot) return null; + const directory = path.resolve(this.legacyStateRoot, sessionId); + if (!directory.startsWith(`${this.legacyStateRoot}${path.sep}`)) { + throw new Error("invalid legacy project bootstrap storage identity"); + } + return path.join(directory, name); + } + + protected acceptedFile(sessionId: string): string { + return path.join(this.sessionDirectory(sessionId), "accepted-inputs.json"); + } + + protected projectIntentFile(projectId: string): string { + if (!/^project_[0-9a-f-]+$/.test(projectId)) { + throw new Error("invalid project bootstrap identity"); + } + const directory = path.resolve(this.root, "projects"); + const file = path.resolve(directory, `${projectId}.json`); + if (!file.startsWith(`${directory}${path.sep}`)) { + throw new Error("invalid project bootstrap identity"); + } + return file; + } + + protected emit(event: ProjectBootstrapLifecycleEvent): void { + try { + void Promise.resolve(this.storageOptions.onEvent?.(event)).catch( + () => {}, + ); + } catch { + // Telemetry is best effort and must never change bootstrap semantics. + } + } + + protected assertOpen(): void { + if (this.closed) throw new ProjectBootstrapCoordinatorClosedError(); + } + + protected receiptForInput( + state: PersistedProjectBootstrapState, + inputId: string, + ): PersistedProjectBootstrapInputReceipt | undefined { + return state.receipts.find((receipt) => receipt.inputId === inputId); + } + + protected updateReceiptStatus( + state: PersistedProjectBootstrapState, + inputId: string, + status: ProjectBootstrapInputReceipt["status"], + ): void { + const receipt = this.receiptForInput(state, inputId); + if (!receipt) return; + // `completed` and `uncertain` are distinct terminal evidence. Neither may + // be weakened or rewritten by later boot reconciliation or a late hook. + if (receipt.status === "completed" || receipt.status === "uncertain") + return; + if (receipt.status === "submitted" && status === "queued") return; + receipt.status = status; + } + + protected serialize( + sessionId: string, + operation: () => Promise, + ): Promise { + const prior = this.writes.get(sessionId) ?? Promise.resolve(); + const next = prior.catch(() => {}).then(operation); + this.writes.set(sessionId, next); + void next.then( + () => { + if (this.writes.get(sessionId) === next) this.writes.delete(sessionId); + }, + () => { + if (this.writes.get(sessionId) === next) this.writes.delete(sessionId); + }, + ); + return next; + } + + protected newState( + session: HarnessSession, + emptyProject: boolean, + ): PersistedProjectBootstrapState { + if (!session.projectBootstrap) { + throw new Error("project bootstrap metadata missing"); + } + return { + schemaVersion: 3, + metadata: { + ...structuredClone(session.projectBootstrap), + // The queue file owns FIFO membership. If that file is missing or was + // quarantined, stale registry IDs cannot resurrect content we no + // longer possess or make the replacement state invalid on next boot. + queuedInputIds: [], + }, + inputs: [], + dispatchingInputId: null, + retryCount: 0, + emptyProject, + attempts: [], + uncertainInputIds: [], + uncertainInputs: [], + receipts: [], + }; + } + + protected async load( + session: HarnessSession, + emptyProject = true, + ): Promise { + const cached = this.states.get(session.id); + // Every transition works on an isolated snapshot. Nothing may mutate the + // authoritative cache until persist() commits the primary queue file. + if (cached) return structuredClone(cached); + let state: PersistedProjectBootstrapState; + try { + const parsed: unknown = JSON.parse( + await fs.readFile(this.file(session.id), "utf8"), + ); + const normalized = parsePersistedProjectBootstrapState(parsed, session); + if (!normalized) throw new Error("invalid project bootstrap state"); + state = normalized; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") { + // Keep malformed legacy files in place for explicit recovery. Never + // rename away or overwrite a file that may contain undelivered input. + throw new Error("project bootstrap state is unavailable"); + } + const legacyFile = this.legacyFile(session.id, "input-queue.json"); + if (legacyFile) { + try { + const legacy: unknown = JSON.parse( + await fs.readFile(legacyFile, "utf8"), + ); + const normalized = parsePersistedProjectBootstrapState( + legacy, + session, + ); + if (!normalized) { + throw new Error("invalid legacy project bootstrap state"); + } + const legacyAccepted = this.legacyFile( + session.id, + "accepted-inputs.json", + ); + if (legacyAccepted) { + try { + const accepted = await fs.readFile(legacyAccepted, "utf8"); + const decoded: unknown = JSON.parse(accepted); + if ( + !isRecord(decoded) || + decoded.schemaVersion !== 1 || + !Array.isArray(decoded.inputIds) || + !decoded.inputIds.every( + (inputId) => typeof inputId === "string" && inputId !== "", + ) + ) { + throw new Error("invalid legacy accepted-input ledger"); + } + await this.writeAcceptedInputIds( + session.id, + decoded.inputIds as string[], + ); + } catch (legacyAcceptedError) { + if ( + (legacyAcceptedError as NodeJS.ErrnoException).code !== "ENOENT" + ) { + throw legacyAcceptedError; + } + } + } + await this.writeState(this.file(session.id), normalized); + state = normalized; + } catch (legacyError) { + if ((legacyError as NodeJS.ErrnoException).code !== "ENOENT") { + // Preserve the only copy and fail closed. Never quarantine or + // overwrite a planner-era FIFO that may contain user input. + throw new Error("legacy project bootstrap state is unavailable"); + } + state = this.newState(session, emptyProject); + } + } else { + state = this.newState(session, emptyProject); + } + } + this.states.set(session.id, structuredClone(state)); + return state; + } + + protected async writeState( + file: string, + state: PersistedProjectBootstrapState, + ): Promise { + if (this.storageOptions.writeState) { + await this.storageOptions.writeState(file, structuredClone(state)); + return; + } + await fs.mkdir(path.dirname(file), { recursive: true, mode: 0o700 }); + const tmp = `${file}.tmp-${process.pid}-${randomUUID()}`; + try { + await fs.writeFile(tmp, JSON.stringify(state, null, 2) + "\n", { + encoding: "utf8", + mode: 0o600, + }); + await fs.rename(tmp, file); + } finally { + await fs.rm(tmp, { force: true }).catch(() => {}); + } + } + + protected async writeIntent( + file: string, + intent: PersistedProjectBootstrapIntent, + ): Promise { + await fs.mkdir(path.dirname(file), { recursive: true, mode: 0o700 }); + const temporary = `${file}.tmp-${process.pid}-${randomUUID()}`; + try { + await fs.writeFile(temporary, `${JSON.stringify(intent, null, 2)}\n`, { + encoding: "utf8", + mode: 0o600, + }); + await fs.rename(temporary, file); + } finally { + await fs.rm(temporary, { force: true }).catch(() => {}); + } + } + + protected async readIntent( + projectId: string, + ): Promise { + let decoded: unknown; + try { + decoded = JSON.parse( + await fs.readFile(this.projectIntentFile(projectId), "utf8"), + ); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; + throw new Error("project bootstrap intent is unavailable"); + } + if ( + !isRecord(decoded) || + decoded.schemaVersion !== 1 || + decoded.projectId !== projectId || + typeof decoded.userId !== "string" || + decoded.userId === "" || + (decoded.targetSessionId !== null && + (typeof decoded.targetSessionId !== "string" || + decoded.targetSessionId === "")) || + (decoded.status !== "scheduled" && decoded.status !== "claimed") || + (decoded.status === "scheduled" && decoded.targetSessionId !== null) || + (decoded.status === "claimed" && decoded.targetSessionId === null) || + typeof decoded.createdAt !== "string" || + typeof decoded.updatedAt !== "string" + ) { + throw new Error("project bootstrap intent is malformed"); + } + return structuredClone( + decoded, + ) as unknown as PersistedProjectBootstrapIntent; + } + + /** Durably schedules the lifecycle before a project has a launchable root. */ + scheduleProject(projectId: string, userId: string): Promise { + return this.serialize(`project:${projectId}`, async () => { + this.assertOpen(); + const existing = await this.readIntent(projectId); + if (existing) { + if (existing.userId !== userId) { + throw new ProjectBootstrapDispatchForbiddenError(); + } + return false; + } + const timestamp = this.now(); + await this.writeIntent(this.projectIntentFile(projectId), { + schemaVersion: 1, + projectId, + userId, + targetSessionId: null, + status: "scheduled", + createdAt: timestamp, + updatedAt: timestamp, + }); + return true; + }); + } + + /** A replacement session may only claim an abandoned pre-provider target + * when doing so cannot strand content in that target's durable FIFO. Missing + * or malformed state is treated conservatively whenever a queue file exists. */ + protected async targetHasUnresolvedInput( + sessionId: string, + ): Promise { + const target = this.storageOptions.sessionManager.get(sessionId); + if ((target?.projectBootstrap?.queuedInputIds.length ?? 0) > 0) return true; + + const files = [ + this.file(sessionId), + this.legacyFile(sessionId, "input-queue.json"), + ].filter((file): file is string => file !== null); + for (const file of files) { + let decoded: unknown; + try { + decoded = JSON.parse(await fs.readFile(file, "utf8")); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") continue; + return true; + } + if (!isRecord(decoded) || !Array.isArray(decoded.inputs)) return true; + const metadata = isRecord(decoded.metadata) ? decoded.metadata : null; + if (!metadata || !Array.isArray(metadata.queuedInputIds)) return true; + if (metadata.queuedInputIds.length > 0) return true; + if (decoded.inputs.length > 0) return true; + if ( + decoded.dispatchingInputId !== null && + decoded.dispatchingInputId !== undefined + ) { + return true; + } + if ( + (decoded.uncertainInputs !== undefined && + !Array.isArray(decoded.uncertainInputs)) || + (decoded.uncertainInputIds !== undefined && + !Array.isArray(decoded.uncertainInputIds)) || + (Array.isArray(decoded.uncertainInputs) && + decoded.uncertainInputs.length > 0) || + (Array.isArray(decoded.uncertainInputIds) && + decoded.uncertainInputIds.length > 0) + ) { + return true; + } + } + return false; + } + + /** Whether a scheduled project still needs its one ordinary first session. */ + needsProjectSession(projectId: string, userId: string): Promise { + return this.serialize(`project:${projectId}`, async () => { + this.assertOpen(); + const intent = await this.readIntent(projectId); + if (!intent) return false; + if (intent.userId !== userId) { + throw new ProjectBootstrapDispatchForbiddenError(); + } + if (intent.status === "scheduled") return true; + const target = intent.targetSessionId + ? this.storageOptions.sessionManager.get(intent.targetSessionId) + : undefined; + if ( + intent.targetSessionId && + this.provisionalProjectClaims.get(projectId) === intent.targetSessionId + ) { + // A published row is already the durable first-session outcome, even + // when its process exited before a provider session ID was observed. + // A missing row can only be replaced after create() releases this + // provisional fence, proving publication never committed. + return false; + } + return ( + !target && + !(await this.targetHasUnresolvedInput(intent.targetSessionId!)) + ); + }); + } + + /** Atomically binds a scheduled project lifecycle to its first real session. */ + claimProject( + identity: ProjectAgentSession, + initialUserInputPending = false, + ): Promise { + return this.serialize(`project:${identity.projectId}`, async () => { + this.assertOpen(); + const intent = await this.readIntent(identity.projectId); + if (!intent) return null; + if (intent.userId !== identity.userId) { + throw new ProjectBootstrapDispatchForbiddenError(); + } + if ( + intent.status === "claimed" && + intent.targetSessionId !== identity.sessionId + ) { + const target = this.storageOptions.sessionManager.get( + intent.targetSessionId!, + ); + if ( + this.provisionalProjectClaims.get(identity.projectId) === + intent.targetSessionId + ) { + return null; + } + if (target) return null; + if (await this.targetHasUnresolvedInput(intent.targetSessionId!)) { + return null; + } + } + const claimed: PersistedProjectBootstrapIntent = { + ...intent, + targetSessionId: identity.sessionId, + status: "claimed", + updatedAt: this.now(), + }; + await this.writeIntent( + this.projectIntentFile(identity.projectId), + claimed, + ); + this.provisionalProjectClaims.set(identity.projectId, identity.sessionId); + this.provisionalSessionClaims.set(identity.sessionId, identity.projectId); + return { + projectId: identity.projectId, + userId: identity.userId, + targetSessionId: identity.sessionId, + bootstrap: initialUserInputPending + ? { status: "skipped", reason: "user-proceeded" } + : { status: "pending" }, + queuedInputIds: [], + }; + }); + } + + /** Release only an unpublished/failed create claim. Durable intent remains + * available for the next proven session and is never deleted. */ + releaseSessionClaim(sessionId: string): Promise { + const projectId = this.provisionalSessionClaims.get(sessionId); + if (!projectId) return Promise.resolve(); + return this.serialize(`project:${projectId}`, async () => { + if (this.provisionalProjectClaims.get(projectId) === sessionId) { + this.provisionalProjectClaims.delete(projectId); + } + this.provisionalSessionClaims.delete(sessionId); + }); + } + + protected async acceptedInputIds( + sessionId: string, + ): Promise | null> { + const file = this.acceptedFile(sessionId); + try { + const decoded: unknown = JSON.parse(await fs.readFile(file, "utf8")); + if ( + !isRecord(decoded) || + decoded.schemaVersion !== 1 || + !Array.isArray(decoded.inputIds) || + decoded.inputIds.length > MAX_INPUT_RECEIPTS || + !decoded.inputIds.every( + (inputId) => typeof inputId === "string" && inputId !== "", + ) || + new Set(decoded.inputIds).size !== decoded.inputIds.length + ) { + throw new Error("invalid accepted-input ledger"); + } + return new Set(decoded.inputIds); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return new Set(); + // An unreadable acknowledgement is safety-significant. Keep the queue's + // write-ahead intent unresolved instead of guessing and replaying it. + // The artifact stays in place until its FIFO is durably terminalized; + // otherwise a crash after quarantine would turn unknown proof into + // ENOENT and authorize replay on the next process. + return null; + } + } + + protected async quarantineAcceptedLedger(sessionId: string): Promise { + const file = this.acceptedFile(sessionId); + const quarantine = path.join( + path.dirname(file), + `accepted-inputs.corrupt-${this.now().replace(/[^0-9A-Za-z]/g, "-")}-${randomUUID()}.json`, + ); + await fs.rename(file, quarantine).catch(() => {}); + } + + protected async terminalizeUnreadableAcceptedLedger( + state: PersistedProjectBootstrapState, + ): Promise { + const sessionId = state.metadata.targetSessionId; + const terminal = structuredClone(state); + const uncertainById = new Map( + terminal.uncertainInputs.map((input) => [input.id, input]), + ); + for (const input of terminal.inputs) { + uncertainById.set(input.id, structuredClone(input)); + const receipt = this.receiptForInput(terminal, input.id); + if (receipt && receipt.status !== "completed") + receipt.status = "uncertain"; + } + for (const receipt of terminal.receipts) { + if (receipt.status === "submitted") receipt.status = "uncertain"; + } + terminal.inputs = []; + terminal.metadata.queuedInputIds = []; + terminal.dispatchingInputId = null; + terminal.uncertainInputs = [...uncertainById.values()]; + terminal.uncertainInputIds = terminal.uncertainInputs.map( + (input) => input.id, + ); + await this.persist(sessionId, terminal); + // Safe state is authoritative before the unreadable artifact moves. A + // failed rename merely causes the same idempotent normalization next time. + await this.quarantineAcceptedLedger(sessionId); + return terminal; + } + + protected async writeAcceptedInputIds( + sessionId: string, + inputIds: readonly string[], + ): Promise { + const file = this.acceptedFile(sessionId); + const temporary = `${file}.tmp-${process.pid}-${randomUUID()}`; + const ledger: AcceptedInputLedger = { + schemaVersion: 1, + inputIds: [...inputIds], + }; + if (this.storageOptions.writeAcceptedLedger) { + await this.storageOptions.writeAcceptedLedger( + file, + structuredClone(ledger), + ); + return; + } + try { + await fs.mkdir(path.dirname(file), { recursive: true, mode: 0o700 }); + await fs.writeFile(temporary, `${JSON.stringify(ledger, null, 2)}\n`, { + encoding: "utf8", + mode: 0o600, + }); + await fs.rename(temporary, file); + } finally { + await fs.rm(temporary, { force: true }).catch(() => {}); + } + } + + protected async recordAcceptedInput( + state: PersistedProjectBootstrapState, + inputId: string, + ): Promise { + const sessionId = state.metadata.targetSessionId; + const accepted = await this.acceptedInputIds(sessionId); + if (accepted === null) { + throw new Error("project bootstrap input acceptance ledger unavailable"); + } + // IDs whose queue entries were already durably removed are stale cleanup + // residue and can be compacted. The active FIFO is bounded by the request + // body limit, and only its IDs are retained here (never input content). + const queuedIds = new Set(state.inputs.map((input) => input.id)); + const retained = [...accepted].filter((id) => queuedIds.has(id)); + if (!retained.includes(inputId)) retained.push(inputId); + await this.writeAcceptedInputIds(sessionId, retained); + } + + protected async reconcileAcceptedInputs( + state: PersistedProjectBootstrapState, + ): Promise { + const sessionId = state.metadata.targetSessionId; + const accepted = await this.acceptedInputIds(sessionId); + if (accepted === null) + return this.terminalizeUnreadableAcceptedLedger(state); + if (accepted.size === 0) return state; + const remaining = state.inputs.filter((input) => !accepted.has(input.id)); + const uncertainById = new Map( + state.uncertainInputs.map((input) => [input.id, structuredClone(input)]), + ); + const receipts = structuredClone(state.receipts); + for (const inputId of accepted) { + const receipt = receipts.find( + (candidate) => candidate.inputId === inputId, + ); + // The side ledger is written only after SessionManager has positively + // acknowledged the PTY submission. In the live coordinator it is + // stronger evidence than the stale FIFO/dequeue marker and removes that + // row without writing Enter again while the active completion barrier + // remains authoritative. Boot normalization is deliberately separate. + if ( + receipt && + receipt.status !== "completed" && + receipt.status !== "uncertain" + ) { + receipt.status = "submitted"; + } + const queued = state.inputs.find((input) => input.id === inputId); + if (queued && receipt?.status === "uncertain") { + uncertainById.set(inputId, structuredClone(queued)); + } + } + const remainingUncertain = [...uncertainById.values()]; + const remainingUncertainIds = remainingUncertain.map((input) => input.id); + const changed = + remaining.length !== state.inputs.length || + remainingUncertain.length !== state.uncertainInputs.length || + remainingUncertain.some( + (input, index) => input.id !== state.uncertainInputs[index]?.id, + ) || + receipts.some( + (receipt, index) => receipt.status !== state.receipts[index]?.status, + ) || + remainingUncertainIds.length !== state.uncertainInputIds.length || + remainingUncertainIds.some( + (inputId, index) => inputId !== state.uncertainInputIds[index], + ) || + (state.dispatchingInputId !== null && + accepted.has(state.dispatchingInputId)); + const reconciled: PersistedProjectBootstrapState = { + ...structuredClone(state), + inputs: remaining, + dispatchingInputId: + state.dispatchingInputId && accepted.has(state.dispatchingInputId) + ? null + : state.dispatchingInputId, + metadata: { + ...structuredClone(state.metadata), + queuedInputIds: remaining.map((input) => input.id), + }, + uncertainInputIds: remainingUncertainIds, + uncertainInputs: remainingUncertain, + receipts, + }; + if (changed) await this.persist(sessionId, reconciled); + // Only after the authoritative queue/receipt transition is durable may its + // acknowledgement be removed. A cleanup failure leaves harmless positive + // proof that the next boot can reconcile again; it never makes Enter + // replayable. + await this.writeAcceptedInputIds( + sessionId, + [...accepted].filter((id) => remaining.some((input) => input.id === id)), + ).catch(() => {}); + return reconciled; + } + + protected async persist( + sessionId: string, + state: PersistedProjectBootstrapState, + ): Promise { + try { + await this.writeState(this.file(sessionId), state); + } catch { + if (!isTerminal(state.metadata)) { + const previousBootstrap = state.metadata.bootstrap; + const attemptId = + previousBootstrap.status === "generating" + ? previousBootstrap.attemptId + : undefined; + const latestAttempt = state.attempts.at(-1); + const positivelyPreSubmit = + !latestAttempt || + latestAttempt.phase === "claimed" || + latestAttempt.phase === "not-submitted"; + const retryable = + positivelyPreSubmit && + (previousBootstrap.status !== "failed" || + previousBootstrap.retryable); + if (retryable && latestAttempt) latestAttempt.status = "retired"; + state.metadata.bootstrap = state.inputs.length + ? { status: "skipped", reason: "user-proceeded" } + : { + status: "failed", + retryable, + errorCode: "persistence_failed", + }; + // At least one of the two stores may still be available. Keep the + // bounded classification wherever possible; never persist raw errors. + let fallbackCommitted = false; + try { + await this.storageOptions.sessionManager.setProjectBootstrapMetadata( + sessionId, + state.metadata, + ); + fallbackCommitted = true; + } catch { + // The primary queue fallback below may still retain the bounded + // terminal classification. + } + try { + await this.writeState(this.file(sessionId), state); + fallbackCommitted = true; + } catch { + // If sessions.json committed, mergeRegistration treats that terminal + // projection as authoritative on restart. Only a total two-store + // outage retains the last committed cache. + } + if (fallbackCommitted) + this.states.set(sessionId, structuredClone(state)); + this.emit({ + name: "project_bootstrap.failed", + projectId: state.metadata.projectId, + sessionId, + ...(attemptId ? { attemptId } : {}), + errorCode: "persistence_failed", + retryable, + queueDepth: state.inputs.length, + }); + } + throw new Error("project bootstrap state persistence failed"); + } + + // The queue file contains the full coordinator state and is authoritative. + // Publish its clone only after that primary write commits: a transient + // failure before this point must not leave a phantom dispatch intent in + // memory. SessionManager's sessions.json metadata is a UI/list projection, + // not a second commit prerequisite. If that projection write fails after + // the queue commit, aborting here would strand a durable pre-PTY intent + // that restart must conservatively drop even though submitInput was never + // called. Keep dispatch moving and retry the projection on every later + // transition/registration instead. + this.states.set(sessionId, structuredClone(state)); + await this.storageOptions.sessionManager + .setProjectBootstrapMetadata(sessionId, state.metadata) + .catch(() => {}); + } + + /** + * Merge the two durable stores under a single serialized registration CAS. + * Queue-file inputs are authoritative. A terminal manager greeting is newer + * than a non-terminal queue greeting (resume suppression), while a terminal + * queue greeting is newer than a stale non-terminal manager snapshot. + */ + protected mergeRegistration( + state: PersistedProjectBootstrapState, + session: HarnessSession, + ): void { + if (!session.projectBootstrap) return; + const managerTerminal = isTerminal(session.projectBootstrap); + const queueTerminal = isTerminal(state.metadata); + if (managerTerminal && !queueTerminal) { + state.metadata.bootstrap = structuredClone( + session.projectBootstrap.bootstrap, + ); + } + state.metadata.queuedInputIds = state.inputs.map((input) => input.id); + } +} diff --git a/packages/harness/src/core/studio-project-catalog.ts b/packages/harness/src/core/studio-project-catalog.ts index a0b90be9..b7816ba5 100644 --- a/packages/harness/src/core/studio-project-catalog.ts +++ b/packages/harness/src/core/studio-project-catalog.ts @@ -327,6 +327,25 @@ function storageError(): StudioProjectCatalogError { /** Internal deterministic seams used only by file-lock race regressions. */ export type StudioProjectCatalogLockTestHooks = DurableFileLockTestHooks; +export interface StudioProjectCatalogLifecycleHooks { + /** + * Runs under the catalog lock before newly allocated project identities are + * committed. A durable write-ahead consumer can make the subsequent catalog + * commit recoverable without changing the public project schema. + */ + beforeProjectsCreatedCommit?: ( + projects: readonly StudioProjectSummary[], + ) => void | Promise; + /** + * Runs after both the catalog file and this instance reflect the committed + * identities. Delivery may be retried, so consumers must be project-keyed + * and idempotent. + */ + afterProjectsCreatedCommit?: ( + projects: readonly StudioProjectSummary[], + ) => void | Promise; +} + /** * Durable, serialized owner of Studio project identity. Catalog reads never * run package inventory or source discovery; callers provide the already @@ -342,6 +361,7 @@ export class StudioProjectCatalog { private readonly catalogPath: string, private readonly now: () => Date = () => new Date(), private readonly lockTestHooks: StudioProjectCatalogLockTestHooks = {}, + private readonly lifecycleHooks: StudioProjectCatalogLifecycleHooks = {}, ) {} private enqueue(operation: () => Promise): Promise { @@ -506,9 +526,14 @@ export class StudioProjectCatalog { updatedAt: timestamp, }; const next = [...cloneProjects(this.projects!), project]; + await this.lifecycleHooks.beforeProjectsCreatedCommit?.([ + publicSummary(project), + ]); await this.persist(next); this.projects = next; - return publicSummary(project); + const summary = publicSummary(project); + await this.lifecycleHooks.afterProjectsCreatedCommit?.([summary]); + return summary; }); } @@ -610,6 +635,7 @@ export class StudioProjectCatalog { } const reconciledScopes: WorkspaceScopeSummary[] = []; + const createdProjects: StudioProjectIdentity[] = []; for (const { canonical, scope } of dedupedScopes.values()) { const matchingProjects = next.filter( (candidate) => @@ -644,6 +670,7 @@ export class StudioProjectCatalog { updatedAt: timestamp, }; next.push(project); + createdProjects.push(project); changed = true; } else { let projectChanged = false; @@ -679,8 +706,18 @@ export class StudioProjectCatalog { } if (changed) { + if (createdProjects.length > 0) { + await this.lifecycleHooks.beforeProjectsCreatedCommit?.( + createdProjects.map(publicSummary), + ); + } await this.persist(next); this.projects = next; + if (createdProjects.length > 0) { + await this.lifecycleHooks.afterProjectsCreatedCommit?.( + createdProjects.map(publicSummary), + ); + } } return { projects: (changed ? next : this.projects!).map(publicSummary), diff --git a/packages/harness/src/index.ts b/packages/harness/src/index.ts index f1241f95..a3164fe0 100644 --- a/packages/harness/src/index.ts +++ b/packages/harness/src/index.ts @@ -23,6 +23,7 @@ export type { PlanRelationshipId, ProposalActor, ProjectAgentSession, + ProjectBootstrapInputReceipt, ProjectBootstrapMetadata, ProjectBootstrapState, ProposalOperationId, diff --git a/packages/harness/src/shared/agent-map.ts b/packages/harness/src/shared/agent-map.ts index 900fe348..303b673f 100644 --- a/packages/harness/src/shared/agent-map.ts +++ b/packages/harness/src/shared/agent-map.ts @@ -474,3 +474,84 @@ export interface ProjectBootstrapMetadata { bootstrap: ProjectBootstrapState; queuedInputIds: string[]; } + + + +export interface ProjectBootstrapQueuedInput { + id: string; + sessionId: string; + text: string; + acceptedAt: string; +} + + + +/** + * Content-free receipt for input accepted by the durable bootstrap FIFO. + * `uncertain` is terminal: Studio cannot prove whether that logical turn ran, + * so it will never replay it automatically. + */ +export interface ProjectBootstrapInputReceipt { + requestId: string | null; + inputId: string; + status: "queued" | "submitted" | "uncertain" | "completed"; + acceptedAt: string; +} + + + +export type ProjectBootstrapRegistrationMode = + | "boot" + | "created" + | "live" + | "resumed"; + + + +/** Content-free lifecycle telemetry for project bootstrap reliability. */ +export type ProjectBootstrapLifecycleEvent = + | { + name: "project_bootstrap.scheduled" | "project_bootstrap.recovered"; + projectId: StudioProjectId; + sessionId: string; + } + | { + name: "project_bootstrap.attempted" | "project_bootstrap.retried"; + projectId: StudioProjectId; + sessionId: string; + attemptId: string; + retryOrdinal: number; + queueDepth: number; + } + | { + name: "project_bootstrap.delivered"; + projectId: StudioProjectId; + sessionId: string; + attemptId: string; + queueDepth: number; + } + | { + name: "project_bootstrap.failed"; + projectId: StudioProjectId; + sessionId: string; + attemptId?: string; + errorCode: ProjectBootstrapErrorCode; + retryable: boolean; + queueDepth: number; + } + | { + name: "project_bootstrap.preempted" | "project_bootstrap.skipped"; + projectId: StudioProjectId; + sessionId: string; + attemptId?: string; + reason: "user-proceeded" | "map-not-empty"; + queueDepth: number; + } + | { + name: "project_bootstrap.input_delivery_uncertain"; + projectId: StudioProjectId; + sessionId: string; + inputId: string; + errorCode: "delivery_uncertain"; + queueDepth: number; + };