diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 8ab9c0b284..3f3e02c5bd 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -153,7 +153,7 @@ import { } from "./workspace-registry.js"; import { createWorkspaceRuntimeService, - type BoundWorkspaceRuntime, + DEFAULT_DOCKER_WORKSPACE_IMAGE, type WorkspaceRuntimeConfig, type WorkspaceRuntimeRecordStore, type WorkspaceRuntimeService, @@ -867,17 +867,12 @@ export async function createPaseoDaemon( path.join(config.paseoHome, "projects", "workspaces.json"), logger, ); - const providerWorkspaceBindings = new Map< - string, - { runtime: BoundWorkspaceRuntime; workspace: ReturnType } - >(); - // Record removal evicts bindings below. Until Phase 4 adds probe invalidation/TTL destruction, - // live probe entries remain bounded to one deterministic workspace per project/runtime pair. + const providerWorkspaceBindings = new Map>(); const bindWorkspaceProviderCapability = async (workspaceId: string, runtimeId: string) => { if (!workspaceRuntime) throw new Error("Workspace runtime is not available"); - const runtime = await workspaceRuntime.bind(workspaceId); const cached = providerWorkspaceBindings.get(workspaceId); - if (cached?.runtime === runtime) return cached.workspace; + if (cached) return cached; + const runtime = await workspaceRuntime.bind(workspaceId); const workspace = bindProviderWorkspace({ runtime, cwd: ".", @@ -887,7 +882,7 @@ export async function createPaseoDaemon( isolatedEnvironment: config.workspaceRuntimes?.[runtimeId]?.providerEnvironment, }), }); - providerWorkspaceBindings.set(workspaceId, { runtime, workspace }); + providerWorkspaceBindings.set(workspaceId, workspace); return workspace; }; providerProbe = createProviderProbeService({ @@ -907,8 +902,20 @@ export async function createPaseoDaemon( if (!workspaceRuntime) throw new Error("Workspace runtime is not available"); return workspaceRuntime.resume(workspaceId); }, + pause: (workspaceId) => { + if (!workspaceRuntime) throw new Error("Workspace runtime is not available"); + return workspaceRuntime.pause(workspaceId); + }, + destroy: (workspaceId) => { + if (!workspaceRuntime) throw new Error("Workspace runtime is not available"); + return workspaceRuntime.destroy(workspaceId); + }, + listRuntimes: () => { + if (!workspaceRuntime) throw new Error("Workspace runtime is not available"); + return workspaceRuntime.listRuntimes(); + }, }, - runtimeConfiguration: config.workspaceRuntimes, + runtimeConfiguration: resolveProviderProbeRuntimeConfiguration(config.workspaceRuntimes), bindWorkspaceProviderCapability, }); const workspaceRecords: WorkspaceRuntimeRecordStore = { @@ -952,6 +959,11 @@ export async function createPaseoDaemon( externalRuntimes: config.workspaceRuntimes, ...workspaceRecords, }); + const detachProviderProbeInvalidation = projectRegistry.subscribeToMutations(async (mutation) => { + if (mutation.kind === "archive" || mutation.kind === "remove") { + await providerProbe?.invalidateProject(mutation.projectId); + } + }); async function resolveRegistryRuntimeId(workspaceId: string): Promise { const workspace = await workspaceRegistry?.get(workspaceId); @@ -1076,6 +1088,12 @@ export async function createPaseoDaemon( await agentStorage.initialize(); logger.info({ elapsed: elapsed() }, "Agent storage initialized"); await Promise.all([projectRegistry.initialize(), workspaceRegistry.initialize()]); + try { + await providerProbe.reconcile(); + logger.info({ elapsed: elapsed() }, "Provider probes reconciled"); + } catch (error) { + logger.warn({ err: error }, "Provider probe reconciliation failed"); + } try { await workspaceRuntime.reconcile(); logger.info({ elapsed: elapsed() }, "Workspace runtimes reconciled"); @@ -1879,6 +1897,7 @@ export async function createPaseoDaemon( const stop = async () => { await hubRelationships.stop(); workspaceReconciliation.dispose(); + detachProviderProbeInvalidation(); scriptHealthMonitor.stop(); // Freeze both ingress and registration before taking the agent closure snapshot. wsServer?.prepareForShutdown(); @@ -1888,6 +1907,9 @@ export async function createPaseoDaemon( detachAgentStoragePersistence(); await agentStorage.flush().catch(() => undefined); await providerSnapshotManager.shutdown(); + await providerProbe?.close().catch((error) => { + logger.warn({ err: error }, "Failed to pause provider probes during shutdown"); + }); await terminalManager.killAll(); speechService.stop(); await scheduleService.stop().catch(() => undefined); @@ -1927,6 +1949,23 @@ export async function createPaseoDaemon( }; } +function resolveProviderProbeRuntimeConfiguration( + configured: Readonly> | undefined, +): Readonly> { + const docker = configured?.docker; + return { + ...configured, + docker: { + type: "docker", + ...(docker?.type === "docker" ? docker : {}), + image: + docker?.type === "docker" + ? (docker.image ?? DEFAULT_DOCKER_WORKSPACE_IMAGE) + : DEFAULT_DOCKER_WORKSPACE_IMAGE, + }, + }; +} + async function closeAllAgents(logger: Logger, agentManager: AgentManager): Promise { const agents = agentManager.listAgents(); await Promise.all( diff --git a/packages/server/src/server/provider-probe/index.ts b/packages/server/src/server/provider-probe/index.ts index c1d38b2afb..1225a55e3e 100644 --- a/packages/server/src/server/provider-probe/index.ts +++ b/packages/server/src/server/provider-probe/index.ts @@ -13,9 +13,22 @@ import { createService } from "./internal/service.js"; export interface ProviderProbeService { ensure(input: { projectId: string; runtimeId: string }): Promise<{ workspaceId: string }>; resolveProviderWorkspace(workspaceId: string): Promise; + invalidateProject(projectId: string): Promise; + reconcile(): Promise; + close(): Promise; readonly records: WorkspaceRuntimeRecordStore; } +export interface ProviderProbeTimer { + unref?(): void; +} + +export interface ProviderProbeClock { + now(): Date; + setTimeout(callback: () => void | Promise, delayMs: number): ProviderProbeTimer; + clearTimeout(timer: ProviderProbeTimer): void; +} + export interface ProviderProbeServiceOptions { filePath: string; logger: pino.Logger; @@ -30,8 +43,12 @@ export interface ProviderProbeServiceOptions { runtime: { create(input: CreateWorkspaceInput): Promise; inspect(workspaceId: string): Promise; + pause(workspaceId: string): Promise; resume(workspaceId: string): Promise; + destroy(workspaceId: string): Promise; + listRuntimes(): readonly { runtimeId: string }[]; }; + clock?: ProviderProbeClock; runtimeConfiguration?: Readonly>; bindWorkspaceProviderCapability( workspaceId: string, diff --git a/packages/server/src/server/provider-probe/internal/probe-store.ts b/packages/server/src/server/provider-probe/internal/probe-store.ts index cd39eab76d..6796022223 100644 --- a/packages/server/src/server/provider-probe/internal/probe-store.ts +++ b/packages/server/src/server/provider-probe/internal/probe-store.ts @@ -12,9 +12,9 @@ const ProbeRecordSchema = z projectId: z.string(), runtimeId: z.string(), fingerprint: z.string(), - status: z.enum(["materializing", "ready", "paused"]), - lastUsedAt: z.string(), - createdAt: z.string(), + status: z.enum(["materializing", "ready", "paused", "destroying"]), + lastUsedAt: z.iso.datetime(), + createdAt: z.iso.datetime(), }) .strict(); @@ -23,6 +23,7 @@ export type ProviderProbeRecord = z.infer; export interface ProbeStore { readonly records: WorkspaceRuntimeRecordStore; get(workspaceId: string): Promise; + list(): Promise; put(record: ProviderProbeRecord): Promise; update( workspaceId: string, @@ -45,9 +46,12 @@ export function createProbeStore(filePath: string, logger: pino.Logger): ProbeSt .parse(JSON.parse(await fs.readFile(filePath, "utf8"))); for (const record of records) cache.set(record.workspaceId, record); } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "ENOENT") { - log.error({ err: error, filePath }, "Failed to load provider probes"); + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + loaded = true; + return; } + log.error({ err: error, filePath }, "Failed to load provider probes"); + throw error; } loaded = true; } @@ -84,7 +88,6 @@ export function createProbeStore(filePath: string, logger: pino.Logger): ProbeSt ...record, runtimeId, status: "ready", - lastUsedAt: new Date().toISOString(), })); if (!updated) throw new Error(`Provider probe record not found: ${workspaceId}`); }, @@ -94,7 +97,9 @@ export function createProbeStore(filePath: string, logger: pino.Logger): ProbeSt async restoreWorkspaceRecord(workspaceId) { await store.update(workspaceId, (record) => ({ ...record, status: "ready" })); }, - async beginWorkspaceDeletion() {}, + async beginWorkspaceDeletion(workspaceId) { + await store.update(workspaceId, (record) => ({ ...record, status: "destroying" })); + }, async removeWorkspaceRecord(workspaceId) { await store.remove(workspaceId); }, @@ -104,6 +109,7 @@ export function createProbeStore(filePath: string, logger: pino.Logger): ProbeSt workspaceId: record.workspaceId, runtimeId: record.runtimeId, archived: record.status === "paused", + deleting: record.status === "destroying", })); }, }, @@ -111,6 +117,10 @@ export function createProbeStore(filePath: string, logger: pino.Logger): ProbeSt await load(); return cache.get(workspaceId) ?? null; }, + async list() { + await load(); + return [...cache.values()]; + }, async put(record) { const parsed = ProbeRecordSchema.parse(record); await mutate((draft) => draft.set(parsed.workspaceId, parsed)); diff --git a/packages/server/src/server/provider-probe/internal/service.ts b/packages/server/src/server/provider-probe/internal/service.ts index 7fcea88c32..b4f13bf860 100644 --- a/packages/server/src/server/provider-probe/internal/service.ts +++ b/packages/server/src/server/provider-probe/internal/service.ts @@ -1,25 +1,55 @@ import { createHash } from "node:crypto"; -import type { ProviderProbeService, ProviderProbeServiceOptions } from "../index.js"; +import { runGitCommand } from "../../../utils/run-git-command.js"; +import type { + ProviderProbeClock, + ProviderProbeService, + ProviderProbeServiceOptions, + ProviderProbeTimer, +} from "../index.js"; import type { ProbeStore, ProviderProbeRecord } from "./probe-store.js"; +const idlePauseMilliseconds = 30 * 60_000; +const startupDestroyMilliseconds = 14 * 24 * 60 * 60_000; + interface ServiceOptions extends ProviderProbeServiceOptions { store: ProbeStore; } +const systemClock: ProviderProbeClock = { + now: () => new Date(), + setTimeout(callback, delayMs) { + return setTimeout(() => void callback(), delayMs); + }, + clearTimeout(timer) { + clearTimeout(timer as ReturnType); + }, +}; + export function createService(options: ServiceOptions): ProviderProbeService { - const inFlight = new Map>(); + const clock = options.clock ?? systemClock; + const log = options.logger.child({ module: "provider-probe" }); + const inFlight = new Map< + string, + { projectId: string; promise: Promise<{ workspaceId: string }> } + >(); + const workspaceTails = new Map>(); + const idleTimers = new Map(); + let closed = false; return { records: options.store.records, ensure(input) { + if (closed) return Promise.reject(new Error("Provider probe service is closed")); const workspaceId = probeWorkspaceId(input.projectId, input.runtimeId); const pending = inFlight.get(workspaceId); - if (pending) return pending; - const ensurePromise = materialize(input, workspaceId).finally(() => { - if (inFlight.get(workspaceId) === ensurePromise) inFlight.delete(workspaceId); - }); - inFlight.set(workspaceId, ensurePromise); + if (pending) return pending.promise; + const ensurePromise = sequence(workspaceId, () => materialize(input, workspaceId)).finally( + () => { + if (inFlight.get(workspaceId)?.promise === ensurePromise) inFlight.delete(workspaceId); + }, + ); + inFlight.set(workspaceId, { projectId: input.projectId, promise: ensurePromise }); return ensurePromise; }, async resolveProviderWorkspace(workspaceId) { @@ -27,6 +57,57 @@ export function createService(options: ServiceOptions): ProviderProbeService { if (!record || record.status !== "ready") return null; return options.bindWorkspaceProviderCapability(workspaceId, record.runtimeId); }, + async invalidateProject(projectId) { + await Promise.allSettled( + [...inFlight.values()] + .filter((pending) => pending.projectId === projectId) + .map((pending) => pending.promise), + ); + const records = (await options.store.list()).filter( + (record) => record.projectId === projectId, + ); + await Promise.all( + records.map((record) => + sequence(record.workspaceId, async () => { + const current = await options.store.get(record.workspaceId); + if (current?.projectId === projectId) await destroy(current); + }), + ), + ); + }, + async reconcile() { + const failures: unknown[] = []; + for (const record of await options.store.list()) { + try { + await sequence(record.workspaceId, () => reconcileRecord(record.workspaceId)); + } catch (error) { + failures.push(error); + } + } + if (failures.length > 0) { + throw new AggregateError(failures, "Provider probe reconciliation failed"); + } + }, + async close() { + if (closed) return; + closed = true; + clearAllTimers(); + await Promise.allSettled([...inFlight.values()].map((pending) => pending.promise)); + const failures: unknown[] = []; + for (const record of await options.store.list()) { + try { + await sequence(record.workspaceId, async () => { + const current = await options.store.get(record.workspaceId); + if (current?.status === "ready") await pause(current); + }); + } catch (error) { + failures.push(error); + } + } + if (failures.length > 0) { + throw new AggregateError(failures, "Provider probe shutdown failed"); + } + }, }; async function materialize( @@ -35,32 +116,42 @@ export function createService(options: ServiceOptions): ProviderProbeService { ): Promise<{ workspaceId: string }> { const project = await options.projects.get(input.projectId); if (!project || project.archivedAt) throw new Error(`Project not found: ${input.projectId}`); - const timestamp = new Date().toISOString(); - const fingerprint = probeFingerprint({ + if (!isRuntimeRegistered(input.runtimeId)) { + throw new Error(`Workspace runtime is not registered: ${input.runtimeId}`); + } + const timestamp = clock.now().toISOString(); + const fingerprint = await probeFingerprint({ projectId: project.projectId, rootPath: project.rootPath, runtimeId: input.runtimeId, runtimeConfiguration: options.runtimeConfiguration?.[input.runtimeId], }); - const existing = await options.store.get(workspaceId); - if (existing?.status === "ready") { + let existing = await options.store.get(workspaceId); + if ( + existing && + (existing.projectId !== project.projectId || + existing.runtimeId !== input.runtimeId || + existing.fingerprint !== fingerprint || + existing.status === "destroying") + ) { + await destroy(existing); + existing = null; + } + if (existing) { const inspection = await options.runtime.inspect(workspaceId); if (inspection.status === "ready") { - await touch(existing, timestamp); + await markReady(existing, timestamp); return { workspaceId }; } if (inspection.status === "paused") { await options.runtime.resume(workspaceId); - await options.store.update(workspaceId, (record) => ({ - ...record, - status: "ready", - lastUsedAt: timestamp, - })); + await markReady(existing, timestamp); return { workspaceId }; } + await destroy(existing); } - const record: ProviderProbeRecord = existing ?? { + await options.store.put({ workspaceId, projectId: project.projectId, runtimeId: input.runtimeId, @@ -68,12 +159,6 @@ export function createService(options: ServiceOptions): ProviderProbeService { status: "materializing", lastUsedAt: timestamp, createdAt: timestamp, - }; - await options.store.put({ - ...record, - fingerprint, - status: "materializing", - lastUsedAt: timestamp, }); try { await options.runtime.create({ @@ -87,15 +172,152 @@ export function createService(options: ServiceOptions): ProviderProbeService { purpose: "provider-probe", setupFromPaseoConfig: false, }); + const created = await options.store.get(workspaceId); + if (!created) throw new Error(`Provider probe record not found after create: ${workspaceId}`); + await markReady(created, timestamp); return { workspaceId }; } catch (error) { - await options.store.remove(workspaceId); + const created = await options.store.get(workspaceId); + if (created) { + try { + await destroy(created); + } catch (cleanupError) { + throw new Error(`Provider probe creation failed before cleanup: ${String(error)}`, { + cause: cleanupError, + }); + } + } throw error; } } - async function touch(record: ProviderProbeRecord, timestamp: string): Promise { - await options.store.put({ ...record, lastUsedAt: timestamp }); + async function reconcileRecord(workspaceId: string): Promise { + const record = await options.store.get(workspaceId); + if (!record) return; + if (!isRuntimeRegistered(record.runtimeId)) { + clearIdleTimer(workspaceId); + await options.store.remove(workspaceId); + return; + } + if (record.status === "destroying") { + await destroy(record); + return; + } + const project = await options.projects.get(record.projectId); + if (!project || project.archivedAt || isExpired(record)) { + await destroy(record); + return; + } + const fingerprint = await probeFingerprint({ + projectId: project.projectId, + rootPath: project.rootPath, + runtimeId: record.runtimeId, + runtimeConfiguration: options.runtimeConfiguration?.[record.runtimeId], + }); + if (fingerprint !== record.fingerprint) { + await destroy(record); + return; + } + const inspection = await options.runtime.inspect(workspaceId); + if (inspection.status === "missing" || inspection.status === "error") { + await destroy(record); + return; + } + if (inspection.status === "ready") { + await pause(record); + return; + } + await options.store.update(workspaceId, (current) => ({ ...current, status: "paused" })); + } + + async function markReady(record: ProviderProbeRecord, timestamp: string): Promise { + await options.store.update(record.workspaceId, (current) => ({ + ...current, + status: "ready", + lastUsedAt: timestamp, + })); + scheduleIdlePause(record.workspaceId, timestamp); + } + + async function pause(record: ProviderProbeRecord): Promise { + clearIdleTimer(record.workspaceId); + const inspection = await options.runtime.inspect(record.workspaceId); + if (inspection.status === "missing" || inspection.status === "error") { + await destroy(record); + return; + } + if (inspection.status === "ready") await options.runtime.pause(record.workspaceId); + await options.store.update(record.workspaceId, (current) => ({ ...current, status: "paused" })); + } + + async function destroy(record: ProviderProbeRecord): Promise { + clearIdleTimer(record.workspaceId); + await options.runtime.destroy(record.workspaceId); + } + + function scheduleIdlePause(workspaceId: string, lastUsedAt: string): void { + clearIdleTimer(workspaceId); + if (closed) return; + const remaining = Math.max( + 0, + idlePauseMilliseconds - (clock.now().getTime() - Date.parse(lastUsedAt)), + ); + const timer = clock.setTimeout(async () => { + idleTimers.delete(workspaceId); + try { + await sequence(workspaceId, async () => { + const record = await options.store.get(workspaceId); + if (!record || record.status !== "ready") return; + const elapsed = clock.now().getTime() - Date.parse(record.lastUsedAt); + if (elapsed < idlePauseMilliseconds) { + scheduleIdlePause(workspaceId, record.lastUsedAt); + return; + } + await pause(record); + }); + } catch (error) { + log.warn({ err: error, workspaceId }, "Failed to pause idle provider probe"); + } + }, remaining); + timer.unref?.(); + idleTimers.set(workspaceId, timer); + } + + function isExpired(record: ProviderProbeRecord): boolean { + return clock.now().getTime() - Date.parse(record.lastUsedAt) >= startupDestroyMilliseconds; + } + + function isRuntimeRegistered(runtimeId: string): boolean { + return options.runtime.listRuntimes().some((runtime) => runtime.runtimeId === runtimeId); + } + + function clearIdleTimer(workspaceId: string): void { + const timer = idleTimers.get(workspaceId); + if (!timer) return; + clock.clearTimeout(timer); + idleTimers.delete(workspaceId); + } + + function clearAllTimers(): void { + for (const timer of idleTimers.values()) clock.clearTimeout(timer); + idleTimers.clear(); + } + + async function sequence(workspaceId: string, operation: () => Promise): Promise { + const previous = workspaceTails.get(workspaceId) ?? Promise.resolve(); + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const tail = previous.then(() => gate); + workspaceTails.set(workspaceId, tail); + await previous; + try { + return await operation(); + } finally { + release(); + if (workspaceTails.get(workspaceId) === tail) workspaceTails.delete(workspaceId); + } } } @@ -103,15 +325,37 @@ function probeWorkspaceId(projectId: string, runtimeId: string): string { return `wksprobe_${hash({ projectId, runtimeId }).slice(0, 32)}`; } -function probeFingerprint(input: { +async function probeFingerprint(input: { projectId: string; rootPath: string; runtimeId: string; runtimeConfiguration: unknown; -}): string { - return hash(input); +}): Promise { + return hash({ ...input, sourceRevision: await resolveSourceRevision(input.rootPath) }); +} + +async function resolveSourceRevision(rootPath: string): Promise { + try { + const { stdout } = await runGitCommand(["rev-parse", "HEAD"], { cwd: rootPath }); + const revision = stdout.trim(); + return revision || null; + } catch { + return null; + } } function hash(value: unknown): string { - return createHash("sha256").update(JSON.stringify(value)).digest("hex"); + return createHash("sha256") + .update(JSON.stringify(canonicalize(value))) + .digest("hex"); +} + +function canonicalize(value: unknown): unknown { + if (Array.isArray(value)) return value.map(canonicalize); + if (!value || typeof value !== "object") return value; + return Object.fromEntries( + Object.entries(value as Record) + .sort(([left], [right]) => left.localeCompare(right)) + .map(([key, entry]) => [key, canonicalize(entry)]), + ); } diff --git a/packages/server/src/server/provider-probe/provider-probe.posix.test.ts b/packages/server/src/server/provider-probe/provider-probe.posix.test.ts index 4414e64b00..5d4ae982c7 100644 --- a/packages/server/src/server/provider-probe/provider-probe.posix.test.ts +++ b/packages/server/src/server/provider-probe/provider-probe.posix.test.ts @@ -1,4 +1,5 @@ -import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { execFileSync } from "node:child_process"; +import { mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; @@ -64,6 +65,7 @@ posixDescribe("provider probe service", () => { test("reuses persisted ready identity and resolves only probe provider workspaces", async () => { const first = await createFixture(); const ensured = await first.service.ensure({ projectId: "project-1", runtimeId: "fixture" }); + const providerWorkspace = await first.service.resolveProviderWorkspace(ensured.workspaceId); const reopened = createService(first); first.inspection = { status: "ready", cwd: first.projectRoot }; @@ -72,7 +74,7 @@ posixDescribe("provider probe service", () => { ).resolves.toEqual(ensured); expect(first.creates).toHaveLength(1); await expect(reopened.resolveProviderWorkspace(ensured.workspaceId)).resolves.toBe( - first.providerWorkspace, + providerWorkspace, ); await expect(reopened.resolveProviderWorkspace("ordinary-workspace")).resolves.toBeNull(); }); @@ -111,6 +113,7 @@ posixDescribe("provider probe service", () => { ); fixture.inspection = { status: "missing" }; const reopened = createService(fixture); + fixture.service = reopened; await reopened.ensure({ projectId: "project-1", runtimeId: "fixture" }); @@ -121,32 +124,264 @@ posixDescribe("provider probe service", () => { expect(recreated.workspaceId).toBe(ensured.workspaceId); expect(recreated.fingerprint).not.toBe("stale"); }); + + test("pauses after the idle TTL and resumes on the next ensure without rotating the provider binding", async () => { + const fixture = await createFixture(); + const ensured = await fixture.service.ensure({ projectId: "project-1", runtimeId: "fixture" }); + const firstBinding = await fixture.service.resolveProviderWorkspace(ensured.workspaceId); + + await fixture.clock.advanceBy(30 * 60_000); + + expect(fixture.pauses).toEqual([ensured.workspaceId]); + expect(fixture.inspection.status).toBe("paused"); + await expect(fixture.service.resolveProviderWorkspace(ensured.workspaceId)).resolves.toBeNull(); + + await expect( + fixture.service.ensure({ projectId: "project-1", runtimeId: "fixture" }), + ).resolves.toEqual(ensured); + expect(fixture.resumes).toEqual([ensured.workspaceId]); + await expect(fixture.service.resolveProviderWorkspace(ensured.workspaceId)).resolves.toBe( + firstBinding, + ); + }); + + test("destroys and recreates when the committed project source changes", async () => { + const fixture = await createFixture({ gitProject: true }); + const first = await fixture.service.ensure({ projectId: "project-1", runtimeId: "fixture" }); + const firstBinding = await fixture.service.resolveProviderWorkspace(first.workspaceId); + + await writeFile(path.join(fixture.projectRoot, "source.txt"), "v2\n"); + execFileSync("git", ["add", "source.txt"], { cwd: fixture.projectRoot }); + execFileSync("git", ["-c", "commit.gpgsign=false", "commit", "-m", "v2"], { + cwd: fixture.projectRoot, + }); + + await expect( + fixture.service.ensure({ projectId: "project-1", runtimeId: "fixture" }), + ).resolves.toEqual(first); + expect(fixture.destroys).toEqual([first.workspaceId]); + expect(fixture.creates).toHaveLength(2); + await expect(fixture.service.resolveProviderWorkspace(first.workspaceId)).resolves.not.toBe( + firstBinding, + ); + }); + + test("destroys and recreates when runtime configuration changes", async () => { + const fixture = await createFixture(); + const first = await fixture.service.ensure({ projectId: "project-1", runtimeId: "fixture" }); + const reconfigured = createService(fixture, { + fixture: { type: "command", options: { sentinel: false } }, + }); + fixture.service = reconfigured; + + await expect( + reconfigured.ensure({ projectId: "project-1", runtimeId: "fixture" }), + ).resolves.toEqual(first); + expect(fixture.destroys).toEqual([first.workspaceId]); + expect(fixture.creates).toHaveLength(2); + }); + + test("startup reconcile pauses valid probes and destroys stale, interrupted, and orphaned probes", async () => { + const fixture = await createFixture(); + const valid = await fixture.service.ensure({ projectId: "project-1", runtimeId: "fixture" }); + const recordsPath = path.join(fixture.root, "provider-probes.json"); + const [validRecord] = JSON.parse(await readFile(recordsPath, "utf8")) as Array< + Record + >; + const oldTimestamp = "2026-07-01T00:00:00.000Z"; + await writeFile( + recordsPath, + `${JSON.stringify( + [ + validRecord, + { + ...validRecord, + workspaceId: "wksprobe_old", + projectId: "project-1", + lastUsedAt: oldTimestamp, + createdAt: oldTimestamp, + }, + { + ...validRecord, + workspaceId: "wksprobe_interrupted", + status: "destroying", + }, + { + ...validRecord, + workspaceId: "wksprobe_materializing_ready", + status: "materializing", + }, + { + ...validRecord, + workspaceId: "wksprobe_materializing_missing", + status: "materializing", + }, + { + ...validRecord, + workspaceId: "wksprobe_orphan", + projectId: "missing-project", + }, + ], + null, + 2, + )}\n`, + ); + fixture.inspections.set(valid.workspaceId, { status: "ready", cwd: fixture.projectRoot }); + fixture.inspections.set("wksprobe_old", { status: "paused", cwd: fixture.projectRoot }); + fixture.inspections.set("wksprobe_interrupted", { status: "missing" }); + fixture.inspections.set("wksprobe_materializing_ready", { + status: "ready", + cwd: fixture.projectRoot, + }); + fixture.inspections.set("wksprobe_materializing_missing", { status: "missing" }); + fixture.inspections.set("wksprobe_orphan", { status: "ready", cwd: fixture.projectRoot }); + const restarted = createService(fixture); + fixture.service = restarted; + + await restarted.reconcile(); + + expect(fixture.pauses).toContain(valid.workspaceId); + expect(fixture.destroys).toEqual( + expect.arrayContaining([ + "wksprobe_old", + "wksprobe_interrupted", + "wksprobe_materializing_missing", + "wksprobe_orphan", + ]), + ); + const remaining = JSON.parse(await readFile(recordsPath, "utf8")) as Array<{ + workspaceId: string; + status: string; + }>; + expect(remaining).toEqual( + expect.arrayContaining([ + expect.objectContaining({ workspaceId: valid.workspaceId, status: "paused" }), + expect.objectContaining({ + workspaceId: "wksprobe_materializing_ready", + status: "paused", + }), + ]), + ); + expect(remaining).toHaveLength(2); + }); + + test("project invalidation converges with concurrent ensure and removes the runtime record", async () => { + const fixture = await createFixture(); + let releaseCreate!: () => void; + const createBarrier = new Promise((resolve) => { + releaseCreate = resolve; + }); + const originalCreate = fixture.runtime.create; + fixture.runtime.create = async (input) => { + await createBarrier; + return originalCreate(input); + }; + + const ensure = fixture.service.ensure({ projectId: "project-1", runtimeId: "fixture" }); + const invalidations = Promise.all([ + fixture.service.invalidateProject("project-1"), + fixture.service.invalidateProject("project-1"), + ]); + releaseCreate(); + const ensured = await ensure; + await invalidations; + + expect(fixture.destroys).toEqual([ensured.workspaceId]); + await expect(fixture.service.records.resolveRuntimeId(ensured.workspaceId)).resolves.toBeNull(); + }); + + test("rejects invalid persisted lifecycle timestamps at the store boundary", async () => { + const fixture = await createFixture(); + await writeFile( + path.join(fixture.root, "invalid-provider-probes.json"), + `${JSON.stringify([ + { + workspaceId: "wksprobe_invalid", + projectId: "project-1", + runtimeId: "fixture", + fingerprint: "fingerprint", + status: "paused", + lastUsedAt: "yesterday-ish", + createdAt: "2026-08-12T00:00:00.000Z", + }, + ])}\n`, + ); + const invalid = createProviderProbeService({ + filePath: path.join(fixture.root, "invalid-provider-probes.json"), + logger: pino({ enabled: false }), + projects: { get: async () => null }, + runtime: fixture.runtime, + clock: fixture.clock, + bindWorkspaceProviderCapability: async () => ({ cwd: "." }) as ProviderWorkspace, + }); + + await expect(invalid.reconcile()).rejects.toThrow(); + }); }); -async function createFixture() { +async function createFixture(options: { gitProject?: boolean } = {}) { const root = await mkdtemp(path.join(tmpdir(), "paseo-provider-probe-")); roots.push(root); const projectRoot = path.join(root, "project"); + await mkdir(projectRoot); + if (options.gitProject) { + execFileSync("git", ["init", "-b", "main"], { cwd: projectRoot }); + execFileSync("git", ["config", "user.email", "test@getpaseo.local"], { cwd: projectRoot }); + execFileSync("git", ["config", "user.name", "Paseo Test"], { cwd: projectRoot }); + await writeFile(path.join(projectRoot, "source.txt"), "v1\n"); + execFileSync("git", ["add", "source.txt"], { cwd: projectRoot }); + execFileSync("git", ["commit", "-m", "v1"], { cwd: projectRoot }); + } let inspection: WorkspaceRuntimeInspection = { status: "missing" }; + const inspections = new Map(); const creates: CreateWorkspaceInput[] = []; - const providerWorkspace = { cwd: "." } as ProviderWorkspace; + const pauses: string[] = []; + const resumes: string[] = []; + const destroys: string[] = []; + const providerWorkspaces = new Map(); + const clock = new TestClock(Date.parse("2026-08-12T00:00:00.000Z")); + let service: ReturnType; const runtime = { create: async (input: CreateWorkspaceInput) => { creates.push(input); inspection = { status: "ready", cwd: projectRoot }; + inspections.set(input.workspaceId, inspection); await service.records.persistRuntimeId(input.workspaceId, input.runtimeId, { cwd: projectRoot, }); return placement(input, projectRoot); }, - inspect: async () => inspection, - resume: async () => {}, + inspect: async (workspaceId: string) => inspections.get(workspaceId) ?? inspection, + pause: async (workspaceId: string) => { + pauses.push(workspaceId); + inspection = { status: "paused", cwd: projectRoot }; + inspections.set(workspaceId, inspection); + }, + resume: async (workspaceId: string) => { + resumes.push(workspaceId); + inspection = { status: "ready", cwd: projectRoot }; + inspections.set(workspaceId, inspection); + }, + destroy: async (workspaceId: string) => { + destroys.push(workspaceId); + inspection = { status: "missing" }; + inspections.set(workspaceId, inspection); + await service.records.beginWorkspaceDeletion?.(workspaceId); + await service.records.removeWorkspaceRecord?.(workspaceId); + providerWorkspaces.delete(workspaceId); + }, + listRuntimes: () => [{ runtimeId: "fixture", builtin: false, requiresGitProject: false }], }; const fixture = { root, projectRoot, creates, - providerWorkspace, + pauses, + resumes, + destroys, + inspections, + providerWorkspaces, + clock, runtime, get inspection() { return inspection; @@ -154,23 +389,41 @@ async function createFixture() { set inspection(value: WorkspaceRuntimeInspection) { inspection = value; }, - service: undefined as unknown as ReturnType, + get service() { + return service; + }, + set service(value: ReturnType) { + service = value; + }, }; - const service = createService(fixture); + service = createService(fixture); fixture.service = service; return fixture; } -function createService(fixture: { - root: string; - projectRoot: string; - providerWorkspace: ProviderWorkspace; - runtime: { - create(input: CreateWorkspaceInput): Promise; - inspect(workspaceId: string): Promise; - resume(workspaceId: string): Promise; - }; -}) { +function createService( + fixture: { + root: string; + projectRoot: string; + providerWorkspaces: Map; + clock: TestClock; + runtime: { + create(input: CreateWorkspaceInput): Promise; + inspect(workspaceId: string): Promise; + pause(workspaceId: string): Promise; + resume(workspaceId: string): Promise; + destroy(workspaceId: string): Promise; + listRuntimes(): readonly { + runtimeId: string; + builtin: boolean; + requiresGitProject: boolean; + }[]; + }; + }, + runtimeConfiguration: Readonly> = { + fixture: { type: "command", options: { sentinel: true } }, + }, +) { return createProviderProbeService({ filePath: path.join(fixture.root, "provider-probes.json"), logger: pino({ enabled: false }), @@ -181,11 +434,57 @@ function createService(fixture: { : null, }, runtime: fixture.runtime, - runtimeConfiguration: { fixture: { type: "command", options: { sentinel: true } } }, - bindWorkspaceProviderCapability: async () => fixture.providerWorkspace, + clock: fixture.clock, + runtimeConfiguration, + bindWorkspaceProviderCapability: async (workspaceId) => { + let workspace = fixture.providerWorkspaces.get(workspaceId); + if (!workspace) { + workspace = { cwd: "." } as ProviderWorkspace; + fixture.providerWorkspaces.set(workspaceId, workspace); + } + return workspace; + }, }); } function placement(input: CreateWorkspaceInput, cwd: string): WorkspaceRuntimePlacement { return { workspaceId: input.workspaceId, runtimeId: input.runtimeId, cwd }; } + +interface TestTimer { + callback: () => void | Promise; + at: number; + cleared: boolean; +} + +class TestClock { + private readonly timers: TestTimer[] = []; + + constructor(private time: number) {} + + now = (): Date => new Date(this.time); + + setTimeout = (callback: () => void | Promise, delayMs: number): TestTimer => { + const timer = { callback, at: this.time + delayMs, cleared: false }; + this.timers.push(timer); + return timer; + }; + + clearTimeout = (timer: TestTimer): void => { + timer.cleared = true; + }; + + async advanceBy(milliseconds: number): Promise { + const end = this.time + milliseconds; + while (true) { + const timer = this.timers + .filter((candidate) => !candidate.cleared && candidate.at <= end) + .sort((left, right) => left.at - right.at)[0]; + if (!timer) break; + timer.cleared = true; + this.time = timer.at; + await timer.callback(); + } + this.time = end; + } +} diff --git a/packages/server/src/server/workspace-runtime/workspace-runtime.docker.e2e.test.ts b/packages/server/src/server/workspace-runtime/workspace-runtime.docker.e2e.test.ts index 05434eb19c..1b292904d7 100644 --- a/packages/server/src/server/workspace-runtime/workspace-runtime.docker.e2e.test.ts +++ b/packages/server/src/server/workspace-runtime/workspace-runtime.docker.e2e.test.ts @@ -756,7 +756,120 @@ test("public daemon and client keep Docker provider discovery, launch, files, Gi } }, 180_000); -test.todo("Phase 4: Docker destroy/recreate rotates the provider snapshot to fixture-model-v2"); +test("Docker probe pause survives restart and source invalidation rotates the provider snapshot", async () => { + const fixture = await dockerTestFixture("paseo-runtime-docker-probe-lifecycle-"); + const root = fixture.root; + const repo = await createRepository(root); + await writeFile(path.join(repo, "provider-model.txt"), "fixture-model-v1\n"); + execFileSync("git", ["add", "provider-model.txt"], { cwd: repo }); + execFileSync("git", ["-c", "commit.gpgsign=false", "commit", "-m", "provider v1"], { + cwd: repo, + }); + const paseoHomeRoot = path.join(root, "daemon-home"); + const paseoHome = path.join(paseoHomeRoot, ".paseo"); + await mkdir(paseoHome, { recursive: true }); + const projectId = "docker-probe-lifecycle-project"; + const projectRegistry = new FileBackedProjectRegistry( + path.join(paseoHome, "projects", "projects.json"), + createTestLogger(), + ); + await projectRegistry.initialize(); + const createdAt = new Date().toISOString(); + await projectRegistry.upsert( + createPersistedProjectRecord({ + projectId, + rootPath: repo, + kind: "git", + displayName: projectId, + createdAt, + updatedAt: createdAt, + }), + ); + const daemonOptions = { + paseoHomeRoot, + cleanup: false, + mcpEnabled: false, + agentClients: {}, + workspaceRuntimes: { docker: { type: "docker" as const, image } }, + providerOverrides: { + "docker-fixture": { + extends: "acp" as const, + label: "Docker Fixture", + command: ["node", "./fixture-agent.mjs"], + params: { supportsMcpServers: false }, + enabled: true, + }, + }, + }; + let daemon = await createTestPaseoDaemon(daemonOptions); + let client = new DaemonClient({ + url: `ws://127.0.0.1:${daemon.port}/ws`, + appVersion: "0.3.0-beta.2", + reconnect: { enabled: false }, + }); + let workspaceId = ""; + + try { + await client.connect(); + const firstProbe = await client.ensureWorkspaceRuntimeProbe({ projectId, runtimeId: "docker" }); + expect(firstProbe).toMatchObject({ status: "ready", error: null }); + workspaceId = firstProbe.workspaceId!; + fixture.workspace(workspaceId); + const firstSnapshot = await client.getProvidersSnapshot({ cwd: "/workspace", workspaceId }); + expect(firstSnapshot.entries).toContainEqual( + expect.objectContaining({ + provider: "docker-fixture", + status: "ready", + models: [expect.objectContaining({ id: "fixture-model-v1" })], + }), + ); + + await client.close(); + await daemon.close(); + expect(dockerRunningContainerCount(workspaceId)).toBe(0); + expect(dockerResourceCount("ps", workspaceId)).toBe(1); + expect(dockerResourceCount("volume", workspaceId)).toBe(1); + + await writeFile(path.join(repo, "provider-model.txt"), "fixture-model-v2\n"); + execFileSync("git", ["add", "provider-model.txt"], { cwd: repo }); + execFileSync("git", ["-c", "commit.gpgsign=false", "commit", "-m", "provider v2"], { + cwd: repo, + }); + + daemon = await createTestPaseoDaemon(daemonOptions); + client = new DaemonClient({ + url: `ws://127.0.0.1:${daemon.port}/ws`, + appVersion: "0.3.0-beta.2", + reconnect: { enabled: false }, + }); + await client.connect(); + const recreatedProbe = await client.ensureWorkspaceRuntimeProbe({ + projectId, + runtimeId: "docker", + }); + expect(recreatedProbe).toEqual({ + requestId: recreatedProbe.requestId, + workspaceId, + status: "ready", + error: null, + }); + const rotatedSnapshot = await client.getProvidersSnapshot({ cwd: "/workspace", workspaceId }); + expect(rotatedSnapshot.entries).toContainEqual( + expect.objectContaining({ + provider: "docker-fixture", + status: "ready", + models: [expect.objectContaining({ id: "fixture-model-v2" })], + }), + ); + + await expect(client.removeProject(projectId)).resolves.toEqual({ removedWorkspaceIds: [] }); + expect(dockerResourceCount("ps", workspaceId)).toBe(0); + expect(dockerResourceCount("volume", workspaceId)).toBe(0); + } finally { + await client.close().catch(() => undefined); + await daemon.close().catch(() => undefined); + } +}, 180_000); test("Docker creation validates roots and rolls back when runtime selection cannot persist", async () => { const fixture = await dockerTestFixture("paseo-runtime-docker-rollback-"); @@ -1634,6 +1747,16 @@ function dockerResourceCount(kind: "ps" | "volume", workspaceId: string): number return execFileSync("docker", args, { encoding: "utf8" }).trim() ? 1 : 0; } +function dockerRunningContainerCount(workspaceId: string): number { + return execFileSync( + "docker", + ["ps", "-q", "--filter", `label=sh.paseo.workspace-id=${workspaceId}`], + { encoding: "utf8" }, + ).trim() + ? 1 + : 0; +} + function nextFile(directory: string, expectedName: string): Promise { return new Promise((resolve, reject) => { const expected = path.join(directory, expectedName); diff --git a/packages/server/src/server/workspace-runtime/workspace-runtime.posix.test.ts b/packages/server/src/server/workspace-runtime/workspace-runtime.posix.test.ts index b8d9621998..6e9bcdf4a9 100644 --- a/packages/server/src/server/workspace-runtime/workspace-runtime.posix.test.ts +++ b/packages/server/src/server/workspace-runtime/workspace-runtime.posix.test.ts @@ -149,7 +149,12 @@ posixDescribe.each(runtimeContractIds)("%s runtime public contract", (runtimeId) const runtime = await fixture.service.bind(fixture.workspaceId); - expect(Object.keys(runtime).sort()).toEqual(["files", "resolveCommand", "run"]); + expect(Object.keys(runtime).sort()).toEqual([ + "files", + "resolveCommand", + "run", + "scriptTerminal", + ]); await expect(runtime.resolveCommand("git")).resolves.toMatch(/^\//u); await expect(runtime.resolveCommand("paseo-command-that-does-not-exist")).resolves.toBeNull(); await fixture.service.destroy(fixture.workspaceId);