feat: integrate PR 5450 review specs and implementation

.ai/specs/2026-08-21-background-work-04-solution.md

@@ -196,6 +196,23 @@ What the pattern implies elsewhere in the repo, as **named follow-up specs**:
 
 None of these is needed for the leased tier or for `data_sync`; all of them reuse the same two ideas, which is why they are listed here rather than designed.
 
+## 📝 Complexity model and benchmark gate
+
+The leased tier must keep the unit of work addressable by key; it must not move the failure from "one long job" to "one repairer that scans every job". The model is:
+
+| Path | Required shape | Bound |
+|---|---|---|
+| Enqueue / re-enqueue | deterministic `queueJobId = pj-<jobId>-<seq>-<redrives>`; broker dedup is a keyed lookup | O(log queue) or broker-native O(1) |
+| Claim | primary-key lookup plus `(seq, redrives)` and lease predicate | O(log jobs), no backlog scan |
+| Single-runner check | partial unique index on `(lock_key, tenant, org)` for live leased rows | O(log live_keys) |
+| Heartbeat | one HOT update on the row; no indexed heartbeat/lease columns | O(1) row update, no index churn |
+| Q2/Q5 repair | indexed pending/mirror cells (`pending_since`, `finished_at`) | O(due + log jobs) |
+| Q1 orphan detection | bounded scan of live leased rows because `lease_expires_at` is intentionally unindexed | O(live_leased) until the measured ceiling below |
+
+Q1 is the only intentional scan. It is allowed for phase 1 only with a measured ceiling: the implementation PR must show the reconciler tick stays below **100 ms p95 at 10k live leased rows with 100 expired rows** on the integration Postgres, and below **1 s p95 at 100k live leased rows** or else add a narrow repair-cell side table before merge. Do **not** index `lease_expires_at` on `progress_jobs` to satisfy this; that would turn every heartbeat into index churn and violate invariant 6. The side-table fallback is a separate cell index owned by the repairer, not another authoritative clock.
+
+The acceptance report must include tick p50/p95, database queries per tick, rows scanned, rows repaired, and `pg_stat_user_tables.n_tup_hot_upd` for `progress_jobs` during the heartbeat phase. A speed number without the repaired-row count is not a result.
+
 ## 📋 Implementation specs (replaces the single phasing table of v3)
 
 | Part | Spec | Owner surface | Depends on | Ships alone? | Closes | Approval state |

.ai/specs/2026-08-21-background-work-06-leased-jobs-in-progress.md

@@ -244,7 +244,7 @@ flowchart TB
 
 Every reconciler transition that ends a row (Q1 park, Q3 cancel) is a §7 terminal transaction: the CAS and the domain mirror commit together, and a throwing mirror rolls the CAS back so the row is selected again next tick. Q5 is the only path in which the mirror runs *after* a terminal CAS committed, and it exists for two cases only: kinds that declared `mirror: 'deferred'`, and rows whose terminal CAS committed before the column existed (migration backfill sets `domain_mirrored_at = finished_at` for rows without `subject_type`, and leaves it null where a subject exists so Q5 re-mirrors them once).
 
-Constraint: `pendingTtlMs` must exceed the worst-case queue backlog for that kind, otherwise Q2 re-drives healthy queued deliveries (they are refused on `redrives` and cost nothing but a wasted delivery, yet each one counts toward `never_started`); the default 15 min is per kind and the reconciler logs a warning when a Q2 re-drive finds the previous delivery still queued (via `Queue.getJobState` where the capability exists). Cost note: Q1 scans the `running ∧ leased` partial index and filters `lease_expires_at` in the heap — a bounded scan of live leased rows every 30 s, which stays cheap to ~10k live rows; beyond that the follow-up is a narrow `progress_job_leases` side table, not an index on `lease_expires_at` (invariant 6).
+Constraint: `pendingTtlMs` must exceed the worst-case queue backlog for that kind, otherwise Q2 re-drives healthy queued deliveries (they are refused on `redrives` and cost nothing but a wasted delivery, yet each one counts toward `never_started`); the default 15 min is per kind and the reconciler logs a warning when a Q2 re-drive finds the previous delivery still queued (via `Queue.getJobState` where the capability exists). Cost note: Q1 scans the `running ∧ leased` partial index and filters `lease_expires_at` in the heap — a bounded scan of live leased rows every 30 s. This is the only intentional O(live) path in the design; all other repair cells are indexed by their due class. It is acceptable for phase 1 only if the part 4 benchmark gate is met (100 ms p95 at 10k live rows / 100 expired, and 1 s p95 at 100k live rows). If it misses, the implementation must add the narrow `progress_job_leases` repair-cell side table before shipping, rather than indexing `lease_expires_at` on `progress_jobs` and breaking HOT heartbeats (invariant 6).
 
 **Connection ceiling** (`packages/queue/AGENTS.md` → Connection Budget): a leased slice uses its request-scoped EM plus one *transient* pooled connection during each heartbeat/lease statement (≤ 1 query every 20 s per slice), and the reconciler worker uses one connection per batch query. The worst case per worker process is therefore `Σconcurrency + 1` concurrent connections instead of `Σconcurrency`; the existing DB-budget clamp in `mercato worker --all` is updated to reserve that one extra connection and the integration test asserts `pg_stat_activity` stays under the clamp during the soak.
 
@@ -335,6 +335,7 @@ Real Postgres + Redis (`__integration__`, docker-compose runner), with a test ki
 9. Reconciler vs live driver → take fences the driver at its next heartbeat; at most one in-flight page applied twice.
 10. Connection ceiling: `pg_stat_activity` stays under the `mercato worker --all` clamp during a 3-replica soak.
 11. Retention never deletes a terminal row with `domain_mirrored_at is null` and a subject.
+12. Reconciler complexity benchmark: a synthetic leased-job fixture with 10k live rows / 100 expired rows and 100k live rows / 100 expired rows runs through the repo-native benchmark target and, when `spbench` is available, through the part 4 `spbench run` commands. The report includes p50/p95 tick time, rows scanned, rows repaired, query count, and `n_tup_hot_upd` growth during heartbeats.
 
 ## 📋 Implementation Plan
 
@@ -344,8 +345,9 @@ Real Postgres + Redis (`__integration__`, docker-compose runner), with a test ki
 4. Kind registry (`mirror` option, `maxMirrorAttempts`) + `runSlice` factory + generic worker binding. Tests: refused claim does no work; budget → yield → rewritten delivery claims; throw → `failSlice` + rethrow → retry claims; unrecoverable → terminal via §7; shutdown → interrupted yield with counters untouched; mirror throw on `drained` → slice fails, retry completes.
 5. Reconciler worker (`progress-reconcile` repeatable every 30 s via `upsertRepeatable`, registered at worker boot) + `progress-retention`; Q1–Q5; `SKIP LOCKED` batches; per-row isolation; stale sweep moved here; `OM_PROGRESS_SWEEP_ON_READ` (default on). Tests: orphan take bumps epoch and re-drives with backoff; Q1 waits for `next_run_at`; poison park on `redrives_since_commit`; lost hand-back re-drive; Q3 cancels a dead slice with the mirror in the same commit; Q5 mirrors a deferred row and a backfilled legacy row; a throwing park mirror leaves the row for the next tick and counts; two reconcilers on one set; tracked-tier sweep parity; repeatable survives a failed run and a Redis flush + boot; retention skips unmirrored rows.
 6. Leased cancel semantics + minimal cascade (§8); `POST /api/progress/jobs/[id]/redrive` (ACL `progress.update` + kind features); top bar "cancelling"/"parked"; OpenAPI.
-7. Integration coverage above.
-8. AGENTS.md (`progress` incl. the repairer carve-out and the `onTransition` idempotency rule, root Task Router row), docs page "leased jobs", `BACKWARD_COMPATIBILITY.md` and UPGRADE_NOTES entries, `.ai/lessons.md`.
+7. Repo-native `progress bench-leases` command (or equivalent test-only CLI) that seeds synthetic leased rows and prints the part 4 benchmark columns; run it directly and through `spbench` where available.
+8. Integration coverage above.
+9. AGENTS.md (`progress` incl. the repairer carve-out and the `onTransition` idempotency rule, root Task Router row), docs page "leased jobs", `BACKWARD_COMPATIBILITY.md` and UPGRADE_NOTES entries, `.ai/lessons.md`.
 
 ## Open items for review
 

.ai/specs/2026-08-21-background-work-07-data-sync-adoption.md

@@ -91,6 +91,7 @@ Adapter contract: **signature unchanged**. Three `data_sync/AGENTS.md` Ask-First
 4. Cancel during adapter I/O → adapter receives `signal` → run `cancelled` and mirrored in one commit.
 5. Existing `data_sync` integration specs green.
 6. **Soak**: docker-compose with 3 worker replicas, a fake adapter producing 500 batches, a supervisor SIGKILLing one replica per minute and flushing Redis once; assert every `(run_id, batch_no)` appears exactly once in a test ledger the adapter writes inside the fenced commit, `sync_runs.batches_completed = 500`, `sync_runs.status = 'completed'` with `finished_at` set, and no row is left `running`/`pending`/unmirrored after the reconciler's next tick.
+7. **Performance/complexity gate**: the same fake adapter runs under a backlog fixture with 10k and 100k unrelated live leased jobs. The data_sync run must still complete with exactly-once ledger rows, and the progress reconciler must meet the part 4 p95 targets. Report quality (`500/500` unique committed batches), speed (slice throughput and reconciler p95), and baseline (today's one-job data_sync path or the previous implementation PR run on the same fixture).
 
 ## 📋 Implementation Plan
 

.ai/specs/2026-08-23-local-queue-filesystem-watch-recovery.md

@@ -0,0 +1,70 @@
+# Local Queue Filesystem Watch Recovery
+
+## Status
+
+Proposed implementation specification for the local queue strategy. This
+specification makes the existing behavior explicit before changing it.
+
+## Contract
+
+Filesystem notifications are an acceleration path, never the delivery
+guarantee. A continuous worker MUST process a committed queue entry when:
+
+1. the queue file is atomically replaced;
+2. the queue directory is moved away and recreated;
+3. `fs.watch` cannot be installed or later emits an error; or
+4. the host exhausts its inotify watch quota (`ENOSPC`).
+
+The worker MUST preserve at-least-once processing. A missed notification may
+increase latency, but MUST NOT lose a committed entry. Idle workers MUST NOT
+read the queue file more frequently than the configured idle safety interval.
+
+## Failure model
+
+On Linux, `fs.watch(file)` is backed by an inotify watch attached to the inode
+identified when the watch is installed. The local queue writer replaces
+`queue.json` during an atomic write. The replacement has a new inode, so a
+watch on the old file is not a durable subscription to future queue files.
+Moving the containing directory is stronger: the watched pathname disappears
+and a later directory with the same pathname is a different filesystem object.
+
+`ENOSPC` from `fs.watch` means the process has exhausted the kernel's inotify
+watch quota; it does not mean the disk is full. Retrying continuously increases
+log volume and CPU usage without changing the quota. Retries therefore MUST be
+bounded and the normal safety poll MUST remain authoritative.
+
+## Required design
+
+The implementation MUST combine:
+
+- a file watcher for low-latency wakeups while the current queue inode exists;
+- identity comparison (`device`, `inode`) before reusing a watcher;
+- re-arm after a file `rename` event;
+- a parent-directory recovery signal for queue-directory replacement, when
+  available;
+- bounded watcher re-arm attempts after setup/error failure; and
+- the existing idle safety poll as the final correctness mechanism.
+
+The implementation MUST NOT add an unconditional high-frequency queue-file
+read loop. Recovery probes may be enabled only while the watched directory is
+absent or the watcher is unavailable, and MUST stop once the watcher is
+re-established or the bounded recovery window expires.
+
+## Verification matrix
+
+| Case | Expected result |
+|---|---|
+| New enqueue with watcher | handler runs before safety interval |
+| Atomic queue-file replacement | handler runs and watcher is re-armed |
+| Queue directory moved/recreated | first entry is recovered by polling; second entry is delivered without waiting for the long idle interval |
+| Watch setup throws | no job loss; bounded fallback polling remains active |
+| Watch emits `ENOSPC` | no retry storm; no job loss; bounded retries then safety polling |
+| Idle worker | no queue-file reads before the configured idle interval |
+| Worker restart | previous watcher is closed exactly once |
+
+## Acceptance criteria
+
+The local queue strategy test file MUST pass in full. The full queue package
+suite MUST pass without an open-handle timeout or unbounded watcher retry log.
+The implementation change MUST report the observed event latency and fallback
+latency separately; an event-driven result does not prove recovery behavior.

packages/core/src/modules/data_sync/lib/__tests__/repair-cell.test.ts

@@ -0,0 +1,19 @@
+import { clearDataSyncRepairCell, recordDataSyncRepairCell } from '../repair-cell'
+
+describe('data-sync repair-cell integration', () => {
+  it('records a scoped repair cell for a failed delivery', async () => {
+    const em = { upsert: jest.fn().mockResolvedValue(undefined) }
+    await recordDataSyncRepairCell(em as never, 'progress-1', { tenantId: 'tenant-1', organizationId: 'org-1' }, 'import', 'upstream timeout')
+    expect(em.upsert).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({
+      jobId: 'progress-1', tenantId: 'tenant-1', organizationId: 'org-1', cell: 'data_sync.import', reason: 'upstream timeout',
+    }), expect.anything())
+  })
+
+  it('clears the same scoped repair cell after successful delivery', async () => {
+    const em = { nativeDelete: jest.fn().mockResolvedValue(1) }
+    await clearDataSyncRepairCell(em as never, 'progress-1', { tenantId: 'tenant-1', organizationId: 'org-1' })
+    expect(em.nativeDelete).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({
+      jobId: 'progress-1', tenantId: 'tenant-1', organizationId: 'org-1',
+    }))
+  })
+})

packages/core/src/modules/data_sync/lib/repair-cell.ts

@@ -0,0 +1,34 @@
+import type { EntityManager } from '@mikro-orm/postgresql'
+import { removeRepairCell, upsertRepairCell } from '../../progress/lib/repair-cells'
+
+type RepairScope = {
+  tenantId: string
+  organizationId?: string | null
+}
+
+export async function clearDataSyncRepairCell(
+  em: EntityManager,
+  progressJobId: string | null | undefined,
+  scope: RepairScope,
+): Promise<void> {
+  if (!progressJobId) return
+  await removeRepairCell(em, progressJobId, scope)
+}
+
+export async function recordDataSyncRepairCell(
+  em: EntityManager,
+  progressJobId: string | null | undefined,
+  scope: RepairScope,
+  direction: 'import' | 'export',
+  reason: string,
+): Promise<void> {
+  if (!progressJobId) return
+  await upsertRepairCell(em, {
+    jobId: progressJobId,
+    tenantId: scope.tenantId,
+    organizationId: scope.organizationId,
+    cell: `data_sync.${direction}`,
+    dueAt: new Date(),
+    reason,
+  })
+}

packages/core/src/modules/data_sync/workers/sync-export.ts

@@ -9,6 +9,7 @@ import {
 } from '../lib/queue-policy'
 import { failAbandonedRun } from '../lib/abandoned-run'
 import { createLogger } from '@open-mercato/shared/lib/logger'
+import { clearDataSyncRepairCell, recordDataSyncRepairCell } from '../lib/repair-cell'
 
 const logger = createLogger('data_sync').child({ component: 'sync-export' })
 
@@ -41,14 +42,20 @@ export default async function handle(job: QueuedJob<SyncJobPayload>, ctx: Handle
   try {
     const engine = ctx.resolve<SyncEngine>('dataSyncEngine')
     await engine.runExport(job.payload.runId, job.payload.batchSize, job.payload.scope)
+    const em = ctx.resolve<import('@mikro-orm/postgresql').EntityManager>('em')
+    const syncRunService = ctx.resolve<SyncRunService>('dataSyncRunService')
+    const run = await syncRunService.getRun(job.payload.runId, job.payload.scope)
+    await clearDataSyncRepairCell(em, run?.progressJobId, job.payload.scope)
   } catch (error) {
     const message = error instanceof Error ? error.message : 'Data sync export worker failed'
     const errorStack = error instanceof Error ? error.stack : undefined
 
     try {
       const syncRunService = ctx.resolve<SyncRunService>('dataSyncRunService')
+      const em = ctx.resolve<import('@mikro-orm/postgresql').EntityManager>('em')
       const progressService = ctx.resolve<ProgressService>('progressService')
       const run = await syncRunService.getRun(job.payload.runId, job.payload.scope)
+      await recordDataSyncRepairCell(em, run?.progressJobId, job.payload.scope, 'export', message)
 
       if (run && run.status !== 'completed' && run.status !== 'failed' && run.status !== 'cancelled') {
         await syncRunService.markStatus(run.id, 'failed', job.payload.scope, message)

packages/core/src/modules/data_sync/workers/sync-import.ts

@@ -9,6 +9,7 @@ import {
 } from '../lib/queue-policy'
 import { failAbandonedRun } from '../lib/abandoned-run'
 import { createLogger } from '@open-mercato/shared/lib/logger'
+import { clearDataSyncRepairCell, recordDataSyncRepairCell } from '../lib/repair-cell'
 
 const logger = createLogger('data_sync').child({ component: 'sync-import' })
 
@@ -41,14 +42,20 @@ export default async function handle(job: QueuedJob<SyncJobPayload>, ctx: Handle
   try {
     const engine = ctx.resolve<SyncEngine>('dataSyncEngine')
     await engine.runImport(job.payload.runId, job.payload.batchSize, job.payload.scope)
+    const em = ctx.resolve<import('@mikro-orm/postgresql').EntityManager>('em')
+    const syncRunService = ctx.resolve<SyncRunService>('dataSyncRunService')
+    const run = await syncRunService.getRun(job.payload.runId, job.payload.scope)
+    await clearDataSyncRepairCell(em, run?.progressJobId, job.payload.scope)
   } catch (error) {
     const message = error instanceof Error ? error.message : 'Data sync import worker failed'
     const errorStack = error instanceof Error ? error.stack : undefined
 
     try {
       const syncRunService = ctx.resolve<SyncRunService>('dataSyncRunService')
+      const em = ctx.resolve<import('@mikro-orm/postgresql').EntityManager>('em')
       const progressService = ctx.resolve<ProgressService>('progressService')
       const run = await syncRunService.getRun(job.payload.runId, job.payload.scope)
+      await recordDataSyncRepairCell(em, run?.progressJobId, job.payload.scope, 'import', message)
 
       if (run && run.status !== 'completed' && run.status !== 'failed' && run.status !== 'cancelled') {
         await syncRunService.markStatus(run.id, 'failed', job.payload.scope, message)