feat: Plan 7 PR B — sync_cursors, lag endpoint, SSE integration

[messagesgoel-blip]

Summary

• Persist monotonic sync_cursors from authenticated edge nodes (last_written_hw only), with periodic + disconnect writes
• Add staff/admin:read GET /v1/admin/sync/lag for per-edge lag
• Add sse-sync integration coverage (bootstrap/heartbeat, 503, 429 backlog, cursor persist + lag, scope forbid); score cases when TRUST_ENGINE_ADDR is set

Test plan

• cd control-plane && npm run test:unit
• cd control-plane && npm run test:integration (Postgres verilink_test on 15432)
• CI: control-plane integration + trust-engine sidecar for score cases
• npx tsc --noEmit in control-plane

@coderabbitai review

[gemini-code-assist]

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

[cursor]

Bugbot is not enabled for your account, so this pull request was not reviewed.

Enable Bugbot in the Cursor dashboard to get automatic reviews on future PRs.

[coderabbitai]

Warning

Review limit reached

You’ve reached a temporary PR review limit under our Fair Usage Limits Policy.

Your recent review volume is higher than typical usage, so adaptive limits are currently applied.

Next review available in: 8 minutes

Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available.
You're only billed for reviews past your plan's rate limits ($0.25/file).

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews.

How do review limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please refer docs for additional details.

Review details

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: ASSERTIVE

Plan: Pro

Run ID: 0dc873b7-22c3-4f93-8b97-d00941c1caef

📥 Commits

Reviewing files that changed from the base of the PR and between 0c3ca95 and a9e51b3.

📒 Files selected for processing (10)

• control-plane/migrations/013_edge_nodes_api_key_unique/migration.sql
• control-plane/src/__tests__/integration/sse-sync.test.ts
• control-plane/src/app.ts
• control-plane/src/config.ts
• control-plane/src/domains/sync/sseSession.ts
• control-plane/src/domains/sync/syncCursorRepository.ts
• control-plane/src/middleware/requireStaff.ts
• control-plane/src/routes/admin.ts
• control-plane/src/routes/sync.ts
• docs/superpowers/plans/HANDOVER.md

---

_{Comment @coderabbitai help to get the list of available commands.}

[qodo-code-review]

PR Summary by Qodo

Persist SSE sync cursors, add admin lag endpoint, and expand integration coverage

✨ Enhancement 🧪 Tests 📝 Documentation ⚙️ Configuration changes 🕐 40+ Minutes

AI Description

• Persist monotonic per-edge sync cursors during SSE sessions (interval, event-count, disconnect).
• Add staff/admin:read endpoint to report per-edge sync lag from stored cursors.
• Add SSE integration tests covering validation, limits, cursor persistence, and live score
 streaming.

Diagram

graph TD
  E{{"Edge node"}} --> S["GET /v1/sync/events (SSE)"] --> SS["SSE session"] --> Q["SseWriteQueue"]
  SS --> SC[("sync_cursors")] --> DB[("Postgres")]
  SS --> SE[("sync_events")]
  A{{"Staff/Admin client"}} --> L["GET /v1/admin/sync/lag"] --> SCR["SyncCursorRepository"] --> SC
  SCR --> SE

  subgraph Legend
    direction LR
    _ext{{"External"}} ~~~ _svc["Service/Handler"] ~~~ _db[("DB table")]
  end

Loading

High-Level Assessment

The following are alternative approaches to this PR:

1. Persist cursors only on disconnect (no periodic writes)

• ➕ No steady-state write load during long-lived streams
• ➕ Less risk of cursor writes competing with event streaming
• ➖ Disconnects/crashes lose progress for long-lived clients
• ➖ Lag endpoint becomes stale until disconnect

2. Debounced async persistence worker (per-session)

• ➕ Avoids awaiting DB writes in the hot poll loop (reduced tail latency)
• ➕ Allows batching/coalescing rapid cursor updates
• ➖ More moving parts (worker lifecycle, shutdown coordination)
• ➖ Must ensure flush-on-close semantics remain correct

3. Maintain per-tenant high-water and lag tenant-scoped

• ➕ Avoids global high-water coupling if sync_events is logically tenant-scoped
• ➕ More accurate per-tenant lag reporting
• ➖ Requires schema/query changes (and likely more indexes)
• ➖ Broader impact across sync event ingestion/reporting

Recommendation: Current approach (monotonic upsert + periodic/disconnect persistence) is a solid baseline for correctness and observability with minimal schema impact. If reviewers see SSE poll-loop latency spikes under load, consider the debounced async persistence worker next; it preserves correctness while reducing blocking awaits in the streaming loop.

Files changed (9) +456 / -21

Enhancement (6) +169 / -15

app.tsMount new admin router under /v1/admin +2/-0

Mount new admin router under /v1/admin

• Registers the admin router so privileged endpoints (e.g., sync lag) are exposed under /v1/admin alongside existing v1 routes.

control-plane/src/app.ts

sseSession.tsPersist monotonic sync cursor for SSE sessions +52/-15

Persist monotonic sync cursor for SSE sessions

• Extends SSE sessions to resolve an edge_node identity from the authenticated API key and persist the last written cursor periodically, every N durable events, and on disconnect. Refactors shutdown to a single finish() path that unregisters, stops timers, closes the queue, and persists the cursor before teardown.

control-plane/src/domains/sync/sseSession.ts

syncCursorRepository.tsAdd cursor persistence + lag reporting repository +74/-0

Add cursor persistence + lag reporting repository

• Adds repository helpers to resolve/create edge_nodes for API keys, upsert sync_cursors monotonically using GREATEST on conflicts, and list lag rows by joining sync_cursors to edge_nodes and comparing to the sync_events high-water mark.

control-plane/src/domains/sync/syncCursorRepository.ts

requireStaff.tsAdd requireStaff middleware for privileged routes +18/-0

Add requireStaff middleware for privileged routes

• Implements staff gating allowing either OIDC staff users or API keys with admin:read scope; otherwise returns FORBIDDEN via AppError.

control-plane/src/middleware/requireStaff.ts

admin.tsAdd staff-only GET /v1/admin/sync/lag endpoint +22/-0

Add staff-only GET /v1/admin/sync/lag endpoint

• Creates an admin router protected by auth + requireStaff and exposes /sync/lag which returns sync lag rows from SyncCursorRepository.

control-plane/src/routes/admin.ts

sync.tsPass apiKeyId into SSE session for edge resolution +1/-0

Pass apiKeyId into SSE session for edge resolution

• Wires req.user.apiKeyId into runSseSession options so the session can map authenticated API keys to edge_nodes and persist cursors server-side.

control-plane/src/routes/sync.ts

Tests (1) +278 / -0

sse-sync.test.tsAdd SSE sync integration tests (limits, cursor, lag, live score events) +278/-0

Add SSE sync integration tests (limits, cursor, lag, live score events)

• Introduces an integration suite validating SSE request parsing, empty bootstrap behavior (cursor=0 + heartbeat), max-connection 503 handling, backlog-cap 429 handling, cursor persistence on disconnect, and admin lag access control. Adds a second describe block for live trust-engine score.upsert streaming gated by TRUST_ENGINE_ADDR/CI.

control-plane/src/tests/integration/sse-sync.test.ts

Documentation (1) +7 / -6

HANDOVER.mdUpdate Plan 7 handover status for PR B +7/-6

Update Plan 7 handover status for PR B

• Updates the handover note date, references Plan 7 PR A as merged, and documents PR B scope/next steps.

docs/superpowers/plans/HANDOVER.md

Other (1) +2 / -0

config.tsAdd SSE cursor persistence configuration knobs +2/-0

Add SSE cursor persistence configuration knobs

• Adds SYNC_SSE_CURSOR_PERSIST_INTERVAL_MS and SYNC_SSE_CURSOR_PERSIST_EVERY_EVENTS to control periodic and event-count-based cursor writes.

control-plane/src/config.ts

[qodo-code-review]

Code Review by Qodo

🐞 Bugs (0) 📘 Rule violations (1) 📜 Skill insights (0)

Context used

✅ Compliance rules (platform): 18 rules

1. Lag endpoint leaks tenants ✓ Resolved 🐞 Bug ⛨ Security

Description

/v1/admin/sync/lag is authorized for any API key with admin:read but returns lag rows across all
tenants, allowing a tenant-scoped key to enumerate other tenants’ edge metadata. This breaks tenant
isolation for operational data (tenant_id, edge IDs/names, cursors, last_sync_at).

Code

control-plane/src/routes/admin.ts[R8-18]

+const router = Router();
+router.use(authMiddleware);
+router.use(requireStaff);
+
+router.get(
+  '/sync/lag',
+  defineHandler({
+    async handler(_req, res) {
+      const rows = await syncCursorRepo.listSyncLag();
+      ok(res, { edges: rows });
+    },

Relevance

●●● Strong

Tenant-isolation/authz bypass patterns (tenant-scoped admin treated as staff) were previously
accepted to fix.

PR-#6
PR-#9

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The admin route is protected by requireStaff, which explicitly allows API keys with admin:read;
API keys are tenant-scoped in auth. The lag handler then calls listSyncLag() which returns
sc.tenant_id rows without a tenant filter, enabling cross-tenant access.

control-plane/src/routes/admin.ts[8-18]
control-plane/src/middleware/requireStaff.ts[5-17]
control-plane/src/middleware/auth.ts[61-92]
control-plane/src/domains/sync/syncCursorRepository.ts[59-72]
PR-#6

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`GET /v1/admin/sync/lag` currently returns lag rows for **all tenants** while `requireStaff` allows **tenant-scoped API keys** with `admin:read`. This enables cross-tenant metadata disclosure.

### Issue Context
- API keys are associated with a single `tenant_id` in auth.
- `requireStaff` treats `admin:read` API keys as staff-equivalent.
- `listSyncLag()` performs an unscoped query returning `sc.tenant_id` for every row.

### Fix Focus Areas
- control-plane/src/routes/admin.ts[8-18]
- control-plane/src/middleware/requireStaff.ts[8-17]
- control-plane/src/domains/sync/syncCursorRepository.ts[59-72]

### Suggested fix
1. Make lag listing tenant-aware:
  - Add `listSyncLagForTenant(tenantId: string)` (or add an optional `tenantId` parameter) and include `WHERE sc.tenant_id = $1`.
2. In the route:
  - If `req.user.type === 'apikey'`, call the tenant-scoped query using `req.user.tenantId`.
  - If `req.user.isStaff === true` (OIDC platform staff), allow the unscoped/global query.
3. Add an integration test with two tenants proving a tenant API key only sees its own rows.

(Alternative: disallow API keys entirely for `/v1/admin/*` by removing the `admin:read` API-key path from `requireStaff`, but that would change the intended auth model.)

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

2. Duplicate edge_nodes possible ✓ Resolved 🐞 Bug ☼ Reliability

Description

resolveEdgeNodeForApiKey() uses a non-atomic SELECT-then-INSERT and the schema does not enforce
uniqueness on (tenant_id, api_key_id), so concurrent SSE sessions can create multiple edge_nodes for
the same API key. This can fragment cursor persistence across multiple edge_node_ids and make lag
reporting inconsistent/noisy.

Code

control-plane/src/domains/sync/syncCursorRepository.ts[R9-26]

+export async function resolveEdgeNodeForApiKey(
+  tenantId: string,
+  apiKeyId: string
+): Promise<string> {
+  const existing = await pool.query(
+    `SELECT id FROM edge_nodes WHERE tenant_id = $1 AND api_key_id = $2 LIMIT 1`,
+    [tenantId, apiKeyId]
+  );
+  if (existing.rows[0]) {
+    return existing.rows[0].id as string;
+  }
+  const inserted = await pool.query(
+    `INSERT INTO edge_nodes (tenant_id, name, api_key_id, status, last_seen_at)
+     VALUES ($1, $2, $3, 'online', now())
+     RETURNING id`,
+    [tenantId, `edge-${apiKeyId.slice(0, 8)}`, apiKeyId]
+  );
+  return inserted.rows[0].id as string;

Relevance

●●● Strong

Team has accepted fixing TOCTOU races by enforcing uniqueness and handling conflicts instead of
SELECT-then-INSERT.

PR-#4

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The resolver is a check-then-insert sequence, and the schema shows no uniqueness constraint on
(tenant_id, api_key_id), enabling duplicates under concurrency.

control-plane/src/domains/sync/syncCursorRepository.ts[9-27]
control-plane/migrations/005_policy/migration.sql[26-37]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`resolveEdgeNodeForApiKey()` does a `SELECT ... LIMIT 1` and then `INSERT` if missing. Without a uniqueness guarantee for `(tenant_id, api_key_id)`, two concurrent calls can both insert, producing duplicate edge_nodes for the same API key.

### Issue Context
- The existing migration defines `edge_nodes` but does not add a unique constraint on `(tenant_id, api_key_id)`.
- The resolver query also has no `ORDER BY`, so if duplicates exist, which row you get is arbitrary.

### Fix Focus Areas
- control-plane/src/domains/sync/syncCursorRepository.ts[9-27]
- control-plane/migrations/005_policy/migration.sql[26-37]

### Suggested fix
1. Add a new migration to enforce uniqueness:
  - `CREATE UNIQUE INDEX edge_nodes_tenant_api_key_unique ON edge_nodes(tenant_id, api_key_id) WHERE api_key_id IS NOT NULL;`
2. Switch to an atomic upsert-style insert:
  - `INSERT ... ON CONFLICT (tenant_id, api_key_id) DO UPDATE SET last_seen_at = now(), status='online' RETURNING id`
  - (Using the unique index as the conflict target.)
3. Consider updating `last_seen_at` on every resolve, not only on insert, since the edge is actively connected.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

3. Lag ordering is lexicographic ✓ Resolved 🐞 Bug ≡ Correctness

Description

listSyncLag() casts the computed lag to text and then orders by the lag alias, so PostgreSQL sorts
lexicographically (e.g., '9' ahead of '100') and mis-ranks edges by backlog. This makes the
endpoint’s primary diagnostic signal incorrect.

Code

control-plane/src/domains/sync/syncCursorRepository.ts[R66-72]

+       COALESCE((SELECT MAX(sync_version) FROM sync_events), 0)::text AS high_water_version,
+       (COALESCE((SELECT MAX(sync_version) FROM sync_events), 0) - sc.last_cursor)::text AS lag,
+       sc.last_sync_at
+     FROM sync_cursors sc
+     JOIN edge_nodes en ON en.id = sc.edge_node_id AND en.tenant_id = sc.tenant_id
+     ORDER BY lag DESC, sc.tenant_id, sc.edge_node_id`
+  );

Relevance

●●● Strong

Deterministic correctness bug: ordering numeric lag as text misranks results; likely to be fixed
when noticed.

PR-#15

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

The SQL explicitly casts lag to text and orders by the lag alias, so sorting uses string
collation rather than numeric comparison.

control-plane/src/domains/sync/syncCursorRepository.ts[59-72]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`listSyncLag()` computes `lag` as `(... - sc.last_cursor)::text AS lag` and then does `ORDER BY lag DESC`, which sorts **text**, not the numeric lag value.

### Issue Context
The endpoint is intended to highlight the most-behind edges first; lexicographic ordering can produce incorrect rankings.

### Fix Focus Areas
- control-plane/src/domains/sync/syncCursorRepository.ts[59-72]

### Suggested fix
- Keep text casting for JSON friendliness if desired, but order by a numeric expression:
 - Option A: `ORDER BY (COALESCE((SELECT MAX(sync_version) FROM sync_events), 0) - sc.last_cursor) DESC, ...`
 - Option B: select both `lag_num` (BIGINT) and `lag` (text) and `ORDER BY lag_num DESC`.
- (Optional) Add a small unit/integration assertion that rows with lag 100 sort ahead of lag 9.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

4. Hardcoded DATABASE_URL and HMAC secret 📘 Rule violation ⛨ Security

Description

The integration test hard-codes a Postgres connection string (including credentials) and a fixed
API_KEY_HMAC_SECRET, which can be detected as secrets/credentials by gitleaks and risks leaking
sensitive values if copied or reused beyond local tests.

Code

control-plane/src/tests/integration/sse-sync.test.ts[R7-10]

+process.env.DATABASE_URL ||=
+  'postgresql://verilink:********@127.0.0.1:15432/verilink_test';
+process.env.API_KEY_HMAC_SECRET ||= 'test-hmac-secret-for-integration';
+process.env.SYNC_SSE_POLL_INTERVAL_MS ||= '100';

Relevance

● Weak

Very similar “remove hardcoded test DB/HMAC secrets” suggestion was explicitly rejected for
integration tests.

PR-#7

ⓘ Recommendations generated based on similar findings in past PRs

Evidence

PR Compliance ID 2373972 requires that source code contain no secrets/credentials as detected by
gitleaks. The added test code includes a hard-coded Postgres connection string with
username/password and a fixed HMAC secret value.

Rule 2373972: Source code must contain no secrets or credentials as detected by gitleaks
control-plane/src/tests/integration/sse-sync.test.ts[7-10]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
A Postgres DSN with embedded credentials and a fixed `API_KEY_HMAC_SECRET` are committed in an integration test. This can trigger gitleaks and creates a pattern of hardcoding secrets in source.

## Issue Context
These values are set via `process.env.* ||= ...` at module load time, so they are always present when the test runs locally (even if CI provides env vars).

## Fix Focus Areas
- control-plane/src/__tests__/integration/sse-sync.test.ts[7-15]

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

_{To customize comments, go to the Qodo configuration screen, or learn more in the docs.}

[messagesgoel-blip]

Review: PR #16 — Plan 7 PR B

Well-structured follow-up to PR #15. Here is my review:

What it adds

• syncCursorRepository.ts — resolves edge nodes from API keys, monotonic GREATEST() cursor upserts, listSyncLag() with per-edge lag
• sseSession.ts — periodic + event-count-based cursor persistence, finish() function that persists cursor on disconnect, apiKeyId passthrough
• requireStaff.ts — middleware for isStaff or admin:read scope
• routes/admin.ts — GET /v1/admin/sync/lag
• Integration tests — 278 lines: 400 parsing, bootstrap cursor id 0 + ping, 503 cap, 429 backlog, cursor persist + lag, forbid without admin:read, score.upsert streaming

What looks good

Area	Notes
Cursor persistence	Monotonic GREATEST(), errors caught and logged. Correct.
Edge node resolution	Auto-creates row on first connect from API key (informational per plan).
finish() consolidation	Eliminates scattered cleanup, fires persist on disconnect. Cleaner.
Integration tests	Comprehensive. Uses ControlPlaneHarness. Score tests skip gracefully without trust-engine.
Lag endpoint	Properly auth-gated via requireStaff. Bigint-safe with ::text.

Minor notes (non-blocking)

• listSyncLag runs MAX(sync_version) twice — negligible at this scale
• edge_nodes not cleaned up on API key revocation — acceptable for v1
• durableEventsWritten is cumulative across initial batch + poll — threshold fires slightly early on first poll, functionally fine

Verdict

Clean, well-tested, integrates correctly with PR #15. Ready to merge.

[messagesgoel-blip]

Addressed the three Qodo findings in a9e51b3:

1. Tenant-scoped lag for admin:read API keys (staff keep global)
2. Numeric lag ordering
3. Unique edge node per API key + atomic upsert

@coderabbitai review

[coderabbitai]

@messagesgoel-blip: Thanks—I'll review the updated changes, including the three addressed findings.

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[messagesgoel-blip]

Review of latest commit (a9e51b3)

This is a cleanup/fix commit that addresses several real issues from the initial PR B implementation.

Changes

1. Migration 013 — unique edge_nodes per API key ✅
Adds a unique index on (tenant_id, api_key_id) with deduplication of existing rows first (cleans up related sync_cursors, decision_aggregates, decision_samples, decision_batches before deleting duplicate edge_nodes). This eliminates the race where concurrent SSE connections from the same API key created duplicate rows.

2. resolveEdgeNodeForApiKey — now upsert instead of select-then-insert ✅
Uses INSERT ... ON CONFLICT ... DO UPDATE — eliminates the race between the SELECT and INSERT. Much cleaner.

3. listSyncLag — tenant-scoped + CTE optimization ✅

• Accepts optional tenantId parameter for tenant-scoped queries
• Uses CTE to compute MAX(sync_version) once instead of two subqueries
• ORDER BY (hw.high_water_version - sc.last_cursor) DESC — numeric sorting, fixing lexicographic ordering (where "9" > "100" as text)

4. routes/admin.ts — proper tenant isolation ✅
Staff sees global lag; API key callers see only their own tenant. The scopeTenantId logic correctly gates this.

5. Integration tests — 2 new tests ✅

• tenant admin:read lag is scoped to own tenant only — multi-tenant isolation verified
• orders lag rows by numeric lag descending — edge_low (lag=9) after edge_high (lag=100), confirming numeric sort

Any issues?

None. Every change is correct and improves the original code.

Verdict

Clean, well-motivated fix commit. Ready to merge.