From 527b252497471c6ea9cad37499d698ed7f3938cd Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 12 Aug 2026 19:35:26 +0200 Subject: [PATCH] feat(workspaces): reconcile provider probe lifecycle Keep lifecycle policy in provider-probe while WorkspaceRuntimeService remains the sole lifecycle executor. Idle probes pause without discarding runtime state; source and runtime fingerprints recreate probes and rotate provider bindings, while restart and project removal converge persisted records. Close provider snapshot and helper bindings before lifecycle transitions to avoid leaks. Keep scriptTerminal in the exact bound-runtime surface because reusable local script terminals were intentionally introduced in the preceding accepted runtime slice. --- packages/server/src/server/bootstrap.ts | 61 +++- .../server/src/server/provider-probe/index.ts | 17 + .../provider-probe/internal/probe-store.ts | 24 +- .../server/provider-probe/internal/service.ts | 306 ++++++++++++++-- .../provider-probe.posix.test.ts | 341 ++++++++++++++++-- .../workspace-runtime.docker.e2e.test.ts | 125 ++++++- .../workspace-runtime.posix.test.ts | 7 +- 7 files changed, 809 insertions(+), 72 deletions(-) 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);