From d0cfcee4d0bd09441c893468298d5fc3b48d463c Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 11 Sep 2026 17:53:31 -0500 Subject: [PATCH] fix: revalidate retained stop evidence on every admission Co-Authored-By: Paperclip --- .../explicit-native-continuation.test.ts | 16 +++++++++++++++- .../src/services/native-local-process-stop.ts | 17 ++++++++++++----- 2 files changed, 27 insertions(+), 6 deletions(-) diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 86062b6176..3c8e6fa365 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -63,9 +63,23 @@ const support = await getEmbeddedPostgresTestSupport(); expect(await admit(f, true)).toMatchObject({ previousRunId: f.sourceRunId }); expect(await hasNativeLocalProcessStop(db, f.companyId, f.sourceRunId)).toBe(false); expect(await admit(f)).toMatchObject({ previousRunId: f.sourceRunId }); - expect(await hasNativeLocalProcessStop(db, f.companyId, f.sourceRunId)).toBe(true); + expect(await hasNativeLocalProcessStop(db, f.companyId, f.sourceRunId)).toBe(false); expect(await admit(f)).toBeNull(); + const held = await seed(); + await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, held.sourceRunId)); + await db.insert(environmentLeases).values({ companyId: held.companyId, heartbeatRunId: held.sourceRunId, + environmentId: environment.id, provider: "local", status: "failed", leasePolicy: "ephemeral", releasedAt: new Date() }); + const [active] = await db.insert(heartbeatRuns).values({ companyId: held.companyId, agentId: held.agentId, + nativeIssueId: held.issueId, status: "queued" }).returning(); + expect(await admit(held)).toBeNull(); // Records the observation but keeps the active-run gate. + await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, active.id)); + verify.mockReturnValueOnce({ fingerprint: "changed", providerProcessIds: [999999998], controllerPid: 999999999 }); + expect(await admit(held)).toBeNull(); // An old receipt cannot authorize changed state. + verify.mockReturnValueOnce(null); + expect(await admit(held)).toBeNull(); // Nor can it authorize a now-live provider. + expect(await admit(held)).toMatchObject({ previousRunId: held.sourceRunId }); + const next = await seed(); await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, next.sourceRunId)); await appendHeartbeatRunEvent(db, { companyId: next.companyId, runId: next.sourceRunId, diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index 2d0353efbb..7df3a0aa67 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -46,7 +46,7 @@ export async function recordNativeLocalProcessStop(db: Db, run: typeof heartbeat /** Only server-authored evidence counts. A later launch invalidates the receipt. */ export async function hasNativeLocalProcessStop(db: Db, companyId: string, runId: string) { - const [event] = await db.select({ eventType: heartbeatRunEvents.eventType }) + const [event] = await db.select({ eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload }) .from(heartbeatRunEvents) .where(and( eq(heartbeatRunEvents.companyId, companyId), @@ -56,7 +56,7 @@ export async function hasNativeLocalProcessStop(db: Db, companyId: string, runId )) .orderBy(desc(heartbeatRunEvents.seq)) .limit(1); - return event?.eventType === LOCAL_PROCESS_STOPPED; + return event?.eventType === LOCAL_PROCESS_STOPPED && event.payload?.source !== "retained_local_session_v1"; } /** Upgrade old cleared identities only from an exact, closed retained session. @@ -68,12 +68,14 @@ export async function reconcileLegacyNativeLocalStop( dryRun: boolean, ): Promise { if (!coordinator) return false; - const [newFormat] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + const [latest] = await db.select({ eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), isNull(heartbeatRunEvents.sourceEventId), inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, LOCAL_PROCESS_STOPPED]), - )).limit(1); - if (newFormat) return false; + )).orderBy(desc(heartbeatRunEvents.seq)).limit(1); + const previousProof = latest?.eventType === LOCAL_PROCESS_STOPPED && + latest.payload?.source === "retained_local_session_v1" ? latest.payload : null; + if (latest && !previousProof) return false; const leases = await db.select().from(environmentLeases).where(and( eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), )); @@ -81,6 +83,11 @@ export async function reconcileLegacyNativeLocalStop( const { verifyRetainedLocalProcessStop } = await import("./native-runtime/native-session-executor.js"); const proof = verifyRetainedLocalProcessStop(run, coordinator); if (!proof) return false; + // A retained-state receipt is an audit observation, not permanent authority. + // Every later use must still prove the same files and absent process inventory. + if (previousProof) return previousProof.fingerprint === proof.fingerprint && + previousProof.controllerPid === proof.controllerPid && + JSON.stringify(previousProof.providerProcessIds) === JSON.stringify(proof.providerProcessIds); if (!dryRun) await appendHeartbeatRunEvent(db, { companyId: run.companyId, runId: run.id, agentId: run.agentId, eventType: LOCAL_PROCESS_STOPPED, stream: "system", level: "info",