diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index 1de7c9a6b37..9c6506cac88 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts @@ -587,6 +587,32 @@ describe("PaperclipControlPlanePort conformance", () => { } finally { release(); await duplicate; } }); + it("joins the deferred observer it saw on arrival even when that observer settles during the duplicate's persistence", async () => { + let release!: () => void, unlock!: () => void, locked!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const unlocked = new Promise(resolve => { unlock = resolve; }); + const lockHeld = new Promise(resolve => { locked = resolve; }); + let observed = 0, duplicates = 0; + const f = await terminalObserverFixture({ async onCommittedEvent() { observed++; await gate; }, + async onDuplicateEvent() { duplicates++; } }); + const signal = new AbortController().signal; + await f.port.appendEvent(f.event, { signal, deferPostCommitObservers: true }); + const lock = db.transaction(async tx => { + await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.id, f.identity.runId)).for("update"); + locked(); await unlocked; + }); + let duplicate: Promise | undefined; + try { + await lockHeld; + duplicate = f.port.appendEvent(f.event, { signal }); + release(); await new Promise(resolve => setTimeout(resolve, 0)); + unlock(); await lock; + expect(await duplicate).toMatchObject({ disposition: "duplicate" }); + expect(observed).toBe(1); + expect(duplicates).toBe(0); + } finally { release(); unlock(); await lock; await duplicate?.catch(() => {}); } + }); + it.each(["company", "agent", "run", "session", "source", "failed", "unproven", "nonterminal", "unscoped"] as const)("does not grant deferred terminal authority to %s evidence", async fault => { const f = await terminalObserverFixture(); const event = { ...f.event, diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.ts b/server/src/services/native-runtime/paperclip-control-plane-port.ts index ac0f582c445..649eeedcb97 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -406,6 +406,8 @@ export class PaperclipControlPlanePort implements ControlPlanePort { } } event = redactRunnerApiEvent(event); + const observerKey = JSON.stringify([event.sourceInstanceId, event.sourceEventId]); + const pendingObservation = this.#deferredEventObservers.get(observerKey); const persisted = await this.#db.transaction(async tx => { this.#assertActive(options); const receipt = await appendHeartbeatRunEvent(tx as unknown as Db, { @@ -431,8 +433,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { }); this.#assertActive(options); const persistenceEndedAt = persistenceStartedAt + performance.now() - persistenceClock; - const observerKey = JSON.stringify([event.sourceInstanceId, event.sourceEventId]); - let observation = this.#deferredEventObservers.get(observerKey); + let observation = pendingObservation ?? this.#deferredEventObservers.get(observerKey); if (!observation) observation = (async () => { try { await this.#onEventPersistence?.(event, persistenceStartedAt, persistenceEndedAt); } catch {} if (persisted.disposition === "committed") {