# Task 1926 Portal Push Concurrency Guard Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Make two concurrent pushes of one account's portal index safe, so a reader never sees an empty or partial tree and no DELETE ever names the generation the pointer is on.

**Architecture:** Three changes to `pushAccount`, no lock and no schema change. A push claims a generation above every generation the account already has, the pointer flip only moves forward, and the sweep only removes strictly older generations. The Task 1923 pre-stage clear is removed because the new claim makes it unreachable.

**Tech Stack:** Node ESM (`.mjs`), vitest 4, Cloudflare D1 over HTTP (SQLite).

## Global Constraints

- Design spec: `platform/docs/superpowers/specs/2026-07-26-task-1926-portal-push-concurrency-guard-design.md`.
- Only two files change: `platform/plugins/cloudflare/bin/portal-index-push.mjs` and `platform/plugins/cloudflare/mcp/__tests__/portal-index-push.test.ts`. Schema comments in `platform/plugins/cloudflare/skills/data-portal/schema.sql` are updated to match.
- Tests run from `platform/plugins/cloudflare/mcp` with `npx vitest run __tests__/portal-index-push.test.ts`.
- No change to `runTargets`, `recordSuccess`, `walkExposed`, or the portal reader in `files.ts`.
- Every comment added states why, not what. No comment may describe behaviour the code does not have.
- Commit trailer on every commit: `Session: 7117b00c-9cf0-4045-99e8-d8405e34a7c6`.

---

### Task 1: Two-writer harness and the task file's interleave

**Files:**
- Modify: `platform/plugins/cloudflare/mcp/__tests__/portal-index-push.test.ts` (append)
- Modify: `platform/plugins/cloudflare/bin/portal-index-push.mjs:194-199` (generation claim), `:234-239` (flip), `:245-248` (sweep)

**Interfaces:**
- Produces: `sharedStore()` returning `{ exec, visible, pointerValue, pointerHistory, snapshots, deletesAtPointer, rows, seed }`; `gate()` returning `{ reached, opened, hit, open }`; `pausingClient(store, gate, {before, after})` returning `{ query, calls }`. Task 2 consumes all three.
- Consumes: the existing `push(client, overrides)` helper and the 7-row fixture already in this file.

- [ ] **Step 1: Write the failing test**

Append to `__tests__/portal-index-push.test.ts`:

```ts
// Task 1926. Two pushes of ONE account are reachable: the admin server's 60s
// loop (pushAllAccounts) and a hand-run `--account <id>` are separate OS
// processes calling the same pushAccount. These two tests drive both writers
// against ONE store and snapshot the reader's view after every statement, so
// "never a half-built tree" stays a measured property rather than a claim.
//
// Statement-level atomicity matches SQLite: each exec() applies whole.
function sharedStore() {
  const rows: { accountId: string; relPath: string; generation: number }[] = []
  let pointer: number | null = null
  const pointerHistory: number[] = []
  const snapshots: string[][] = []
  const deletesAtPointer: number[] = []

  const visible = () =>
    pointer === null
      ? []
      : rows.filter((r) => r.generation === pointer).map((r) => r.relPath).sort()

  // The three DELETE shapes this file can issue. Order matters: `!=` and `<`
  // must be tested before `=`, or `generation != ?` would read as an equality.
  const deleteMatcher = (sql: string, g: number) => {
    if (/generation\s*<\s*\?/i.test(sql)) return (n: number) => n < g
    if (/generation\s*!=\s*\?/i.test(sql)) return (n: number) => n !== g
    return (n: number) => n === g
  }

  return {
    rows,
    snapshots,
    deletesAtPointer,
    pointerHistory,
    visible,
    pointerValue: () => pointer,
    seed(generation: number, relPaths: string[]) {
      for (const relPath of relPaths) rows.push({ accountId: 'acc-1', relPath, generation })
      pointer = generation
      pointerHistory.push(generation)
    },
    async exec(sql: string, params: unknown[] = []) {
      const out = await (async () => {
        if (/SELECT\s+currentGeneration/i.test(sql)) {
          return pointer === null ? [] : [{ currentGeneration: pointer }]
        }
        if (/SELECT\s+MAX\(generation\)/i.test(sql)) {
          const mine = rows.filter((r) => r.accountId === String(params[0]))
          return [{ g: mine.length ? Math.max(...mine.map((r) => r.generation)) : null }]
        }
        if (/INSERT INTO directory_state/i.test(sql)) {
          const g = Number(params[1])
          // The upsert's WHERE, modelled: with the guard present the pointer
          // moves forward only. Without it, every flip lands.
          const forwardOnly = /excluded\.currentGeneration\s*>\s*directory_state\.currentGeneration/i.test(sql)
          if (pointer === null || !forwardOnly || g > pointer) {
            pointer = g
            pointerHistory.push(g)
          }
          return []
        }
        if (/INSERT INTO directory \(/i.test(sql)) {
          const accountId = String(params[0])
          const relPath = String(params[1])
          const generation = Number(params[6])
          if (rows.some((r) => r.accountId === accountId && r.relPath === relPath && r.generation === generation)) {
            throw new Error(
              'd1 query failed: HTTP 500 — UNIQUE constraint failed: directory.accountId, directory.relPath, directory.generation',
            )
          }
          rows.push({ accountId, relPath, generation })
          return []
        }
        if (/DELETE FROM directory\b/i.test(sql)) {
          const g = Number(params[1])
          const matches = deleteMatcher(sql, g)
          // The bug's signature: a DELETE that removes rows at the generation
          // the pointer names is a reader-visible deletion of the live tree.
          if (pointer !== null && matches(pointer) && rows.some((r) => r.generation === pointer)) {
            deletesAtPointer.push(pointer)
          }
          for (let i = rows.length - 1; i >= 0; i--) if (matches(rows[i].generation)) rows.splice(i, 1)
          return []
        }
        if (/SELECT COUNT/i.test(sql)) {
          if (/generation\s*=\s*\?/i.test(sql)) {
            const g = Number(params[1])
            return [{ n: rows.filter((r) => r.generation === g).length }]
          }
          return [{ n: rows.filter((r) => r.generation === pointer).length }]
        }
        return []
      })()
      snapshots.push(visible())
      return out
    },
  }
}

/** A one-shot rendezvous: `reached` settles when the writer parks, `open()` frees it. */
function gate() {
  let hit!: () => void
  let open!: () => void
  const reached = new Promise<void>((r) => (hit = r))
  const opened = new Promise<void>((r) => (open = r))
  return { reached, opened, hit: () => hit(), open: () => open() }
}

/** A client that parks once, either before or after the matching statement. */
function pausingClient(
  store: ReturnType<typeof sharedStore>,
  g: ReturnType<typeof gate>,
  at: { before?: (sql: string) => boolean; after?: (sql: string) => boolean },
) {
  let armed = true
  const calls: { sql: string; params: unknown[] }[] = []
  return {
    calls,
    async query(sql: string, params: unknown[] = []) {
      calls.push({ sql, params })
      if (armed && at.before?.(sql)) {
        armed = false
        g.hit()
        await g.opened
      }
      const out = await store.exec(sql, params)
      if (armed && at.after?.(sql)) {
        armed = false
        g.hit()
        await g.opened
      }
      return out
    },
  }
}

const plainClient = (store: ReturnType<typeof sharedStore>) => ({
  query: (sql: string, params: unknown[] = []) => store.exec(sql, params),
})

describe('two concurrent pushes of one account (Task 1926)', () => {
  // The task file's own interleave. B reads the pointer, then stalls. A stages,
  // flips and sweeps a whole publish. B resumes holding a stale `current`.
  //
  // Before the guard: B recomputed next = current + 1 = the generation the
  // pointer now names, its pre-stage COUNT returned A's live rows, and its
  // DELETE emptied the tree a client was reading.
  it('a peer holding a stale pointer never empties or halves the live tree', async () => {
    const store = sharedStore()
    store.seed(5, ['output', 'output/report.pdf', 'quotes', 'quotes/CR2969'])
    const g = gate()
    const b = pausingClient(store, g, { after: (sql) => /SELECT\s+currentGeneration/i.test(sql) })

    const bRun = push(b, { log: () => {} }).catch((e: unknown) => e)
    await g.reached
    await push(plainClient(store), { log: () => {} })
    g.open()
    await bRun

    // 4 is the seeded tree, 7 is a complete walk of the fixture. Any other
    // count is a tree a client could see half-built.
    expect(store.snapshots.every((s) => s.length === 4 || s.length === 7)).toBe(true)
    expect(store.deletesAtPointer).toEqual([])
    expect(store.visible().length).toBe(7)
  })
})
```

- [ ] **Step 2: Run test to verify it fails**

Run: `cd platform/plugins/cloudflare/mcp && npx vitest run __tests__/portal-index-push.test.ts -t 'never empties or halves'`
Expected: FAIL. `store.snapshots.every(...)` is `false` (a zero-length snapshot appears after B's pre-stage DELETE) and `deletesAtPointer` is `[6]`.

- [ ] **Step 3: Write minimal implementation**

In `bin/portal-index-push.mjs`, replace the `const next = current + 1` line and the two statements around it:

```js
  const state = await client.query(
    'SELECT currentGeneration FROM directory_state WHERE accountId = ?',
    [accountId],
  )
  const current = Number(state[0]?.currentGeneration ?? 0)

  // Claim a generation above EVERY generation this account already holds, not
  // merely above the pointer (Task 1926). Two pushes of one account are
  // reachable: the admin server's 60s loop and a hand-run `--account` are
  // separate processes, so no in-process lock can see both. `current + 1` let
  // the second writer claim the generation the first had just flipped onto and
  // then delete it. A peer that has inserted even one row is visible in this
  // MAX, so the two runs claim different generations.
  const staged = await client.query(
    'SELECT MAX(generation) AS g FROM directory WHERE accountId = ?',
    [accountId],
  )
  const next = Math.max(current, Number(staged[0]?.g ?? 0)) + 1
```

Replace the flip with a forward-only upsert plus a readback:

```js
  // THE atomic point. Upsert rather than insert-or-update, so an account
  // publishing for the first time and one republishing take the same path.
  //
  // The WHERE moves the pointer FORWARD ONLY (Task 1926). Without it a writer
  // that was overtaken while staging drags every reader back onto a generation
  // the winner has already swept, which is an empty tree. SQLite has carried a
  // WHERE on upsert's DO UPDATE since 3.24, and D1 is SQLite.
  await client.query(
    'INSERT INTO directory_state (accountId, currentGeneration) VALUES (?, ?) ' +
      'ON CONFLICT (accountId) DO UPDATE SET currentGeneration = excluded.currentGeneration ' +
      'WHERE excluded.currentGeneration > directory_state.currentGeneration',
    [accountId, next],
  )
  log(`${TAG} op=flip account=${accountId} generation=${next}`)
```

Replace the sweep predicate:

```js
    // Strictly older only (Task 1926): `!= next` would delete a peer's HIGHER
    // in-flight stage. Rows above `next` left by an abandoned stage still
    // converge, because the next cycle claims a generation above them and
    // sweeps them once it flips.
    await client.query('DELETE FROM directory WHERE accountId = ? AND generation < ?', [
      accountId,
      next,
    ])
```

- [ ] **Step 4: Run test to verify it passes**

Run: `cd platform/plugins/cloudflare/mcp && npx vitest run __tests__/portal-index-push.test.ts`
Expected: PASS, 28 tests. The two Task 1923 stage-clear tests still pass at this point (the clear is unreachable but harmless).

- [ ] **Step 5: Commit**

```bash
git add platform/plugins/cloudflare/bin/portal-index-push.mjs platform/plugins/cloudflare/mcp/__tests__/portal-index-push.test.ts
git commit -m "fix(portal): claim a generation above every staged row, flip forward only"
```

---

### Task 2: The superseded writer

**Files:**
- Modify: `platform/plugins/cloudflare/mcp/__tests__/portal-index-push.test.ts` (append inside the Task 1926 describe)
- Modify: `platform/plugins/cloudflare/bin/portal-index-push.mjs` (flip readback branch)

**Interfaces:**
- Consumes: `sharedStore`, `gate`, `pausingClient`, `plainClient` from Task 1.
- Produces: `pushAccount` returns `superseded: true` on the losing path.

- [ ] **Step 1: Write the failing test**

Append inside the `two concurrent pushes of one account (Task 1926)` describe block:

```ts
  // The other order: A stages a whole generation, then stalls before its flip.
  // B starts fresh, sees A's staged rows in the MAX, claims the generation
  // above, and publishes. A wakes holding a generation that is now behind.
  it('a writer overtaken while staging leaves the winner live and says so', async () => {
    const store = sharedStore()
    store.seed(5, ['output', 'output/report.pdf', 'quotes', 'quotes/CR2969'])
    const g = gate()
    const a = pausingClient(store, g, { before: (sql) => /INSERT INTO directory_state/i.test(sql) })
    const lines: string[] = []

    const aRun = push(a, { log: (l: string) => lines.push(l) })
    await g.reached
    await push(plainClient(store), { log: () => {} })
    g.open()
    const aResult = await aRun

    expect(store.snapshots.every((s) => s.length === 4 || s.length === 7)).toBe(true)
    expect(store.deletesAtPointer).toEqual([])
    // The pointer never goes backward, which is what keeps the loser harmless.
    expect(store.pointerHistory).toEqual([...store.pointerHistory].sort((x, y) => x - y))
    expect(store.pointerValue()).toBe(7)

    const contended = lines.find((l) => l.includes('op=stage-contended'))!
    expect(contended).toContain('account=acc-1')
    expect(contended).toContain('generation=6')
    expect(contended).toContain('pointer=7')
    expect(contended).toContain('action=superseded')
    expect(lines.some((l) => l.includes('op=flip'))).toBe(false)
    // No sweep and no verify from the loser: both would measure the WINNER's
    // tree against this run's expected row count and read as a failure.
    expect(a.calls.some((c) => /DELETE FROM directory/i.test(c.sql))).toBe(false)
    expect(lines.some((l) => l.includes('op=verify'))).toBe(false)
    expect(aResult.superseded).toBe(true)
  })
```

- [ ] **Step 2: Run test to verify it fails**

Run: `cd platform/plugins/cloudflare/mcp && npx vitest run __tests__/portal-index-push.test.ts -t 'overtaken while staging'`
Expected: FAIL on `lines.find(...)` returning undefined, because nothing emits `op=stage-contended` yet.

- [ ] **Step 3: Write minimal implementation**

In `bin/portal-index-push.mjs`, put the readback between the flip statement and the `op=flip` log line, and move the log into the winning branch:

```js
  // Read back rather than assume: the D1 client returns rows, not affected-row
  // meta, so a suppressed flip is otherwise indistinguishable from an applied
  // one.
  const flipped = await client.query(
    'SELECT currentGeneration FROM directory_state WHERE accountId = ?',
    [accountId],
  )
  const pointer = Number(flipped[0]?.currentGeneration ?? 0)
  if (pointer !== next) {
    // A peer published while this run was staging. Its tree is live and
    // complete, ours is invisible, and its rows are collected by the peer's
    // next sweep. Sweeping or verifying from here would act on, and measure,
    // the PEER's tree — reporting a failure where nothing failed. Logged
    // rather than silent: a suppressed run is otherwise indistinguishable from
    // one that never started.
    log(
      `${TAG} op=stage-contended account=${accountId} generation=${next} ` +
        `pointer=${pointer} action=superseded`,
    )
    return {
      exposed: r.exposed,
      rows: rows.length,
      schemaPresent: true,
      generation: next,
      superseded: true,
    }
  }
  log(`${TAG} op=flip account=${accountId} generation=${next}`)
```

Update the `pushAccount` JSDoc return type to
`Promise<{ exposed: string[], rows: number, schemaPresent: boolean, generation?: number, superseded?: boolean }>`.

- [ ] **Step 4: Run test to verify it passes**

Run: `cd platform/plugins/cloudflare/mcp && npx vitest run __tests__/portal-index-push.test.ts`
Expected: PASS, 29 tests.

- [ ] **Step 5: Commit**

```bash
git add platform/plugins/cloudflare/bin/portal-index-push.mjs platform/plugins/cloudflare/mcp/__tests__/portal-index-push.test.ts
git commit -m "fix(portal): a superseded push logs stage-contended instead of sweeping"
```

---

### Task 3: Retire the unreachable Task 1923 pre-stage clear

**Files:**
- Modify: `platform/plugins/cloudflare/bin/portal-index-push.mjs` (delete the orphan COUNT, the DELETE, the `op=stage-clear` log)
- Modify: `platform/plugins/cloudflare/mcp/__tests__/portal-index-push.test.ts` (delete the `a partial stage self-heals (Task 1923)` describe, replace with one test of the heal's new mechanism)
- Modify: `platform/plugins/cloudflare/skills/data-portal/schema.sql` (the `generation` column comment)

**Interfaces:**
- Consumes: `wedgeableClient` already in the test file, kept because it is the only client that models the UNIQUE constraint.

- [ ] **Step 1: Write the failing test**

Replace the whole `describe('a partial stage self-heals (Task 1923)')` block with:

```ts
describe('a partial stage self-heals (Task 1923, by Task 1926 generation claim)', () => {
  // The wedge: a stage that died mid-batch left rows at current + 1 while the
  // pointer stayed at current, so every later run recomputed the SAME target
  // and its first INSERT hit UNIQUE(accountId, relPath, generation). The index
  // froze at the stale generation until an operator hand-cleared the orphans,
  // which is what held gls-data for ~6 h on 2026-07-22.
  //
  // Claiming above the highest EXISTING generation heals it with no DELETE
  // before the flip at all, which is what removed Task 1926's hazard rather
  // than narrowing it.
  it('claims above the orphans instead of colliding with them, and sweeps them after the flip', async () => {
    const c = wedgeableClient({
      pointer: 5,
      current: ['output', 'output/report.pdf', 'quotes', 'quotes/CR2969'],
      orphans: ['output', 'output/report.pdf', 'quotes'],
    })
    const lines: string[] = []
    const r = await push(c, { log: (l: string) => lines.push(l) })

    expect(r.rows).toBe(7)
    // 7, not 6: the orphans sat at 6, so this run claims the one above them.
    expect(c.pointerValue()).toBe(7)
    expect(c.rows.every((row) => row.generation === 7)).toBe(true)
    // No DELETE runs before the flip any more, so the window that let a peer
    // empty the live tree does not exist.
    expect(lines.some((l) => l.includes('op=stage-clear'))).toBe(false)
  })
})
```

- [ ] **Step 2: Run test to verify it fails**

Run: `cd platform/plugins/cloudflare/mcp && npx vitest run __tests__/portal-index-push.test.ts -t 'claims above the orphans'`
Expected: FAIL on `lines.some(op=stage-clear)` being `true`, because the pre-stage clear still emits its line (its COUNT now returns 0, so this assertion is the only one that fails).

- [ ] **Step 3: Write minimal implementation**

Delete from `bin/portal-index-push.mjs` the whole Task 1923 block: its comment, the `orphan` query, `orphanCount`, the `if (orphanCount > 0)` DELETE and the `op=stage-clear` log line. Add to the comment above the generation claim:

```js
  // This also retires the Task 1923 pre-stage clear. That clear existed because
  // a later run recomputed the SAME generation as a dead partial stage and
  // collided on UNIQUE(accountId, relPath, generation); claiming above the
  // orphans means there is nothing to collide with, and the post-flip sweep
  // collects them. The clear was itself the DELETE that a concurrent peer could
  // aim at the live tree, so removing it removes the hazard rather than
  // narrowing it.
```

In `skills/data-portal/schema.sql`, replace the `generation` paragraph's second sentence so it reads:

```
-- `generation` is which staging round a row belongs to (Task 1910). The push
-- claims a generation above every generation the account already holds, writes
-- the whole tree there, moves the pointer below in ONE forward-only statement,
-- then deletes strictly older generations (Task 1926). Uniqueness is scoped by
-- generation because two of them hold the same relPath between the write and
-- the flip; UNIQUE (accountId, relPath) would reject the staged write outright.
```

- [ ] **Step 4: Run test to verify it passes**

Run: `cd platform/plugins/cloudflare/mcp && npx vitest run __tests__/portal-index-push.test.ts`
Expected: PASS, 28 tests (29 minus the two removed stage-clear tests, plus this one).

- [ ] **Step 5: Commit**

```bash
git add platform/plugins/cloudflare/bin/portal-index-push.mjs platform/plugins/cloudflare/mcp/__tests__/portal-index-push.test.ts platform/plugins/cloudflare/skills/data-portal/schema.sql
git commit -m "refactor(portal): retire the pre-stage clear the generation claim makes unreachable"
```

---

## Self-review

**Spec coverage.** Generation claim (Task 1 step 3), forward-only flip (Task 1 step 3), backward-only sweep (Task 1 step 3), superseded branch and `op=stage-contended` (Task 2), removal of the pre-stage clear and its tests (Task 3), the two-writer regression tests (Tasks 1 and 2). The spec's residual case (a schema change between the two generation reads) is documented, not coded, and needs no task.

**Placeholders.** None. Every code step carries the code.

**Type consistency.** `sharedStore`, `gate`, `pausingClient`, `plainClient` are defined once in Task 1 and used by the same names in Tasks 2 and 3. `next`, `current`, `pointer`, `staged` are the only new locals in the module and none shadow an existing name.