From 240d46e2ec0f87196210316f7e826e6d4d195845 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 5 Nov 2026 21:51:47 -0600 Subject: [PATCH] fix(native-runtime): join the deferred observer seen when a duplicate append arrives (#19397) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The native runtime writes runner events through `PaperclipControlPlanePort.appendEvent`. > - A terminal event can defer its post-commit observer. A duplicate append of the same event must join that observer. It must not run `onDuplicateEvent`. > - The duplicate looked up the observer only after its own event transaction finished. If the original observer settled during that transaction, the observer left the map. Then the duplicate ran `onDuplicateEvent`. > - This made the test "joins the exact deferred observer on duplicate append..." fail on slow CI (job 127633060467, and once on Windows). > - This pull request captures the pending observer when the duplicate arrives. The join decision no longer depends on database latency. > - The benefit is a deterministic join and a stable test. ## Linked Issues or Issue Description **What happened?** The server test `paperclip-control-plane-port.test.ts > joins the exact deferred observer on duplicate append and replays it after a new port opens` failed intermittently with `Error: must join the original owned observation`. The test gives the duplicate append 30 ms and then releases the original observer. On a slow runner, the duplicate's event transaction took longer than 30 ms. The original observer settled and its `.then` handler removed the map entry. The duplicate then found no observer after its transaction and ran `onDuplicateEvent`. (The CI failure took 90 ms.) **Expected behavior** A duplicate append that arrives while the original deferred observer is pending joins that observer. Database latency does not change this result. **Steps to reproduce** 1. Add a 60 ms sleep inside the `appendEvent` transaction when the receipt disposition is `duplicate`. 2. Run `pnpm vitest run src/services/native-runtime/paperclip-control-plane-port.test.ts -t "joins the exact deferred"` from `server/`. 3. The test fails with `must join the original owned observation`. This is the same error and stack as CI. **Paperclip version or commit** `master` at d8b0eae687. ## What Changed - `appendEvent` now reads the pending deferred observer before it opens the event transaction. After the transaction, it uses that observer. If the duplicate did not see an observer on arrival, it reads the map again after the transaction, as before. - A new regression test holds a `FOR UPDATE` lock on the run row from a second transaction. This forces the duplicate append to wait in persistence while the original observer settles. The test then checks that the duplicate joined and that `onDuplicateEvent` did not run. No sleeps or product hooks are necessary. ## Verification - Red: without the product change, the new test fails (`expected 1 to be +0`: `onDuplicateEvent` ran once). - Green: with the change, `cd server && npx vitest run src/services/native-runtime/paperclip-control-plane-port.test.ts` passes 95/95. The two "joins the" tests pass on 5 consecutive runs. - Injected 60 ms delay on the duplicate transaction: the original test fails without the fix and passes with the fix. The delay was local only and is not in this PR. - `pnpm --filter @paperclipai/server typecheck` passes. ## Risks - Low risk. The change affects only a duplicate append that arrives while a deferred observer for the same event is pending. Such a duplicate now always joins that observer. Before this change, it joined only when its transaction finished first. If the original observer rejects, the joined duplicate gets the original rejection, as a fast duplicate did before. Committed appends and duplicates with no pending observer behave as before. ## Model Used - Claude Opus 5.5 (Anthropic), model ID `claude-opus-5-5`, used through Claude Code with tool use (shell, file edits, test execution). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 --- .../paperclip-control-plane-port.test.ts | 26 +++++++++++++++++++ .../paperclip-control-plane-port.ts | 5 ++-- 2 files changed, 29 insertions(+), 2 deletions(-) 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") {