fix(native-runtime): join the deferred observer seen when a duplicate append arrives (#19397)

## 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 <noreply@anthropic.com>

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<void>(resolve => { release = resolve; });
+    const unlocked = new Promise<void>(resolve => { unlock = resolve; });
+    const lockHeld = new Promise<void>(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<unknown> | 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,

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") {