fix(server-beta): re-pin payload.generation_job_id on idempotent re-create

A generation job whose payload points at a non-existent job id is dropped
silently: `ProviderObservationGenerator` resolves the row through
`payload.generation_job_id`, so `lockOutbox` finds nothing, logs at info level
and returns `{ status: "completed" }` to BullMQ without ever touching Postgres.
The queue reports success, the row stays `queued` forever, and nothing is
recorded as failed.

That pointer goes stale on the upsert path. Every re-create mints a fresh id
for `$1` and puts the same id in the payload it hands over. On
`ON CONFLICT (idempotency_key)` the existing row wins, so the minted id is
discarded for the `id` column -- but `payload || excluded.payload` still lets
it overwrite `generation_job_id`, leaving the row pointing at an id that was
never inserted.

This only bites `session_summary`. Its idempotency key is derived from
`sourceId`, which for summaries is the session, so every re-run of a session
summary collides. Event jobs key off the individual event id and never conflict.

Re-pin `generation_job_id` to the surviving row inside the same merge.

Notes for reviewers:
- An operator re-queue flow that "reuses the persisted payload" (#2459 /
  discussion #2460) does not recover these rows: replaying the stored payload
  replays the stale pointer, and the job is dropped again just as silently.
- Observed on a production deployment running server-beta since May: 190 summary
  jobs stranded in `queued` across four months, 190/190 with
  `payload->>generation_job_id` absent from the table. Event jobs on the same
  instance: 0 divergences in ~221k rows. After re-pointing the payload the same
  rows started draining immediately.

Co-authored-by: Alessandro <alessandropcostabr@users.noreply.github.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/storage/postgres/generation-jobs.ts

@@ -125,7 +125,16 @@ export class PostgresObservationGenerationJobRepository {
         )
         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13::jsonb)
         ON CONFLICT (idempotency_key) DO UPDATE SET
-          payload = observation_generation_jobs.payload || excluded.payload,
+          -- The conflicting INSERT carries a freshly generated job id in both $1
+          -- and payload.generation_job_id. ON CONFLICT keeps the existing row, so
+          -- $1 is discarded -- but the merge below would otherwise let the new
+          -- payload overwrite generation_job_id with that discarded id, leaving
+          -- the row pointing at a job that never existed. The worker resolves the
+          -- job via payload.generation_job_id, so such a row can never be locked
+          -- and is silently dropped. Re-pin the pointer to the surviving row.
+          payload = observation_generation_jobs.payload
+                    || excluded.payload
+                    || jsonb_build_object('generation_job_id', observation_generation_jobs.id),
           updated_at = now()
         RETURNING *
       `,

tests/storage/postgres/postgres-storage.test.ts

@@ -504,6 +504,39 @@ describe('server beta postgres observation storage', () => {
     })).resolves.toBeNull();
   });
 
+  it('re-pins payload.generation_job_id to the surviving row on idempotent re-create', async () => {
+    const { project, event } = await createFixtureScopeWithEventJob(storage);
+
+    const first = await storage.observationGenerationJobs.create({
+      projectId: project.id,
+      teamId: project.teamId,
+      sourceType: 'agent_event',
+      sourceId: event.id,
+      agentEventId: event.id,
+      jobType: 'payload_id_drift',
+      payload: { generation_job_id: 'id-minted-for-the-first-insert' }
+    });
+
+    // The second create collides on idempotency_key. ON CONFLICT keeps the
+    // existing row, so the id minted for this attempt is discarded -- but the
+    // caller still put it in the payload it is handing over.
+    const second = await storage.observationGenerationJobs.create({
+      projectId: project.id,
+      teamId: project.teamId,
+      sourceType: 'agent_event',
+      sourceId: event.id,
+      agentEventId: event.id,
+      jobType: 'payload_id_drift',
+      payload: { generation_job_id: 'id-discarded-by-on-conflict' }
+    });
+
+    expect(second.id).toBe(first.id);
+    // The worker resolves the job through payload.generation_job_id. If the
+    // discarded id survived here, the row could never be locked and the job
+    // would be dropped silently while reporting success to the queue.
+    expect((second.payload as { generation_job_id?: string }).generation_job_id).toBe(first.id);
+  });
+
   it('rejects illegal generation job lifecycle transitions and max-attempt retries', async () => {
     const { project, event } = await createFixtureScopeWithEventJob(storage);
     const job = await storage.observationGenerationJobs.create({