fix: a fallback whose first identity is being saved is still admitted

mutations.toml

@@ -372,6 +372,38 @@ broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core
 broad_report = "tmp/mutations/core.xml"
 broad_id = "{name}"
 
+[[control]]
+id = "serve-adopts-concurrent-first-identity"
+guards = "A serving caller captured before first identity publication must use the same row's validated credential without bypassing admission."
+file = "packages/core/src/native-refresh-coordinator.ts"
+old = "const learned = { ...request.binding, identity: row.identity }"
+new = "const learned = request.binding // NON-VACUITY BREAK: retain the pre-publication identity-absent binding"
+test_file = "packages/core/src/tests/native-refresh-coordinator.test.ts"
+runner = "command"
+command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/native-refresh-coordinator.test.ts", "{test}"]
+test_count_pattern = "CKDEV executed {count} tests"
+expect_red = ["serve follows first identity published after binding capture without a second exchange"]
+expect_message = "error: expect(received).toMatchObject(expected)"
+broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"]
+broad_report = "tmp/mutations/core.xml"
+broad_id = "{name}"
+
+[[control]]
+id = "serve-joins-concurrent-first-identity-write"
+guards = "A partial first-identity pool write seen after the host's snapshot must be joined rather than silently remove a fallback."
+file = "packages/core/src/native-refresh-coordinator.ts"
+old = "row.credentialEpoch === request.binding.credentialEpoch &&"
+new = "row.credentialEpoch === -1 && // NON-VACUITY BREAK: refuse a partially saved first identity"
+test_file = "packages/core/src/tests/native-refresh-coordinator.test.ts"
+runner = "command"
+command = ["bun", "scripts/run-mutation-bun.ts", "packages/core/src/tests/native-refresh-coordinator.test.ts", "{test}"]
+test_count_pattern = "CKDEV executed {count} tests"
+expect_red = ["serve joins a first identity write encountered after binding capture"]
+expect_message = "error: expect(received).toMatchObject(expected)"
+broad_command = ["bun", "scripts/run-mutation-bun.ts", "--broad", "packages/core", "tmp/mutations/core.xml"]
+broad_report = "tmp/mutations/core.xml"
+broad_id = "{name}"
+
 [[control]]
 id = "refresh-join-checks-cause"
 guards = "A proactive refresh must not delay a proven unexpired access token."

packages/core/src/native-refresh-coordinator.ts

@@ -26,14 +26,21 @@ import {
   CustodyTombstoneRefreshError,
   isCustodyTombstoneValue,
 } from './claustrum.ts'
-import { nativeAccountCanServe } from './native-account-view.ts'
+import {
+  nativeAccountCanServe,
+  nativeLocalRowAwaitsTornWrite,
+} from './native-account-view.ts'
 import {
   type NativeLocalCredentialValidation,
   type NativeLocalCredentialVersion,
   nativeLocalCredentialValidationMatches,
 } from './native-credential-validation.ts'
 import { nativeLockObserver } from './native-lock-observer.ts'
-import { NativeTransientAccountError } from './native-transient-account.ts'
+import {
+  isNativeTransientAccountError,
+  joinNativePoolWrite,
+  NativeTransientAccountError,
+} from './native-transient-account.ts'
 import { isTransientNetworkError } from './network-errors.ts'
 import {
   isNativeLocalPoolBinding,
@@ -1325,7 +1332,7 @@ export function createNativeRefreshCoordinator(
     request = { ...request, binding: Object.freeze({ ...request.binding }) }
     if (request.signal?.aborted)
       return { status: 'refused', reason: 'cancelled', persisted: false }
-    const key = nativeLocalRefreshJobKey(request.binding)
+    let key = nativeLocalRefreshJobKey(request.binding)
     let job = jobs.get(key)
     const recordRejection = () => {
       // A 401 can arrive after proactive work started. Record its rejected token
@@ -1338,7 +1345,38 @@ export function createNativeRefreshCoordinator(
     recordRejection()
     if (request.intent === 'serve') {
       try {
-        const row = await rowFor(request.binding)
+        let row = await rowFor(request.binding)
+        if (
+          row &&
+          row.credentialEpoch === request.binding.credentialEpoch &&
+          nativeLocalRowAwaitsTornWrite(options.paths, row)
+        ) {
+          // The host's earlier read can precede a partial state/config write.
+          // Wait for that writer once before deciding this account is ineligible.
+          const joined = await joinNativePoolWrite(
+            options.paths,
+            request.signal ? [request.signal] : [],
+            () => request.signal?.throwIfAborted(),
+            onLockEvent,
+          )
+          if (joined === 'deadline') throw new NativeTransientAccountError()
+          row = await rowFor(request.binding)
+          if (row && nativeLocalRowAwaitsTornWrite(options.paths, row))
+            throw new NativeTransientAccountError()
+        }
+        if (
+          request.binding.identity === undefined &&
+          row?.identity !== undefined
+        ) {
+          const learned = { ...request.binding, identity: row.identity }
+          // First-identity publication can finish after the caller captures its
+          // binding. Use the saved identity only if the row, store and credential
+          // epoch still match; admission still requires permission and exact proof.
+          if (nativeLocalPoolBindingMatches(learned, options.paths, row)) {
+            request = { ...request, binding: Object.freeze(learned) }
+            key = nativeLocalRefreshJobKey(request.binding)
+          }
+        }
         job = jobs.get(key) ?? job
         recordRejection()
         // A pending recovery cannot override custody, quota or refresh-error
@@ -1367,6 +1405,7 @@ export function createNativeRefreshCoordinator(
             return current
         }
       } catch (caught) {
+        if (isNativeTransientAccountError(caught)) throw caught
         return {
           status: 'failed',
           failure: failureOf(asError(caught)),

packages/core/src/tests/native-refresh-coordinator.test.ts

@@ -61,6 +61,10 @@ import {
   type NativeRefreshSubject,
   nativeAccountProviderLock,
 } from '../native-refresh-coordinator.ts'
+import {
+  isNativeTransientAccountError,
+  NativeTransientAccountError,
+} from '../native-transient-account.ts'
 import {
   captureNativeLocalPoolBinding,
   nativeLocalPoolBindingMatches,
@@ -1233,6 +1237,228 @@ test('callback-end identity alias joins during reconciliation and handoff enclos
   expect(active.size).toBe(0)
 })
 
+test('serve follows first identity published after binding capture without a second exchange', async () => {
+  const f = await fixture()
+  await f.add('a')
+  const before = await f.binding('a')
+  const publishing = barrier()
+  const finish = barrier()
+  let calls = 0
+  const c = f.coordinator({
+    refreshToken: async () => {
+      calls++
+      return { ...credential('new'), expiresIn: 3600 }
+    },
+    reconcile: async (event) => {
+      if (event.status === 'persisted') {
+        publishing.release()
+        await finish.promise
+      }
+    },
+  })
+  const first = c.authorize({ intent: 'serve', mode: 'local', binding: before })
+  await publishing.promise
+  const learned = await f.binding('a')
+  expect(learned).toEqual({ ...before, identity: 'A' })
+  // This binding was captured before the row had an identity. Saving its first
+  // identity must not make a serving caller treat the account as replaced.
+  const joined = c.authorize({
+    intent: 'serve',
+    mode: 'local',
+    binding: before,
+  })
+  finish.release()
+  expect(await joined).toMatchObject({
+    status: 'usable',
+    binding: learned,
+    access: 'access-new',
+  })
+  expect(await first).toMatchObject({ status: 'usable', binding: learned })
+  expect(
+    await c.authorize({ intent: 'serve', mode: 'local', binding: before }),
+  ).toMatchObject({ status: 'usable', binding: learned, access: 'access-new' })
+  expect(calls).toBe(1)
+})
+
+test('serve joins a first identity write encountered after binding capture', async () => {
+  const stateWritten = barrier()
+  const finishWrite = barrier()
+  const joining = barrier()
+  let held = false
+  let waitForJoin = false
+  const f = await fixture({
+    onStep: async (step, info) => {
+      if (
+        held ||
+        step !== 'after-state-write' ||
+        info.operation !== 'refresh' ||
+        info.rowId !== 'a'
+      )
+        return
+      held = true
+      stateWritten.release()
+      await finishWrite.promise
+    },
+  })
+  await f.add('a')
+  const before = await f.binding('a')
+  let calls = 0
+  const c = f.coordinator({
+    refreshToken: async () => {
+      calls++
+      return { ...credential('new'), expiresIn: 3600 }
+    },
+    onLockEvent: (event) => {
+      if (
+        waitForJoin &&
+        event.type === 'acquiring' &&
+        event.name === 'pool-config'
+      )
+        joining.release()
+    },
+  })
+  const first = c.authorize({ intent: 'serve', mode: 'local', binding: before })
+  await stateWritten.promise
+  expect(await f.row('a')).toMatchObject({
+    torn: true,
+    candidate: false,
+    identity: 'A',
+  })
+  waitForJoin = true
+  const joined = c.authorize({
+    intent: 'serve',
+    mode: 'local',
+    binding: before,
+  })
+  // Release when the reader waits on the lock or returns early; never leave the
+  // writer held after a refusal.
+  await Promise.race([joining.promise, joined.then(() => {})])
+  finishWrite.release()
+  expect(await joined).toMatchObject({ status: 'usable', access: 'access-new' })
+  expect(await first).toMatchObject({ status: 'usable', access: 'access-new' })
+  expect(calls).toBe(1)
+})
+
+test('a first identity write deadline keeps the transient account error brand', async () => {
+  const stateWritten = barrier()
+  const finishWrite = barrier()
+  let held = false
+  const f = await fixture({
+    onStep: async (step, info) => {
+      if (
+        held ||
+        step !== 'after-state-write' ||
+        info.operation !== 'refresh' ||
+        info.rowId !== 'a'
+      )
+        return
+      held = true
+      stateWritten.release()
+      await finishWrite.promise
+    },
+  })
+  await f.add('a')
+  const before = await f.binding('a')
+  const c = f.coordinator()
+  const first = c.authorize({ intent: 'serve', mode: 'local', binding: before })
+  await stateWritten.promise
+  const refused = await c
+    .authorize({ intent: 'serve', mode: 'local', binding: before })
+    .then(
+      (value) => ({ value }),
+      (error: unknown) => ({ error }),
+    )
+  finishWrite.release()
+  expect(refused).toHaveProperty('error')
+  if (!('error' in refused))
+    throw new Error('Expected a transient account error')
+  expect(isNativeTransientAccountError(refused.error)).toBe(true)
+  expect(refused.error).toBeInstanceOf(NativeTransientAccountError)
+  expect(await first).toMatchObject({ status: 'usable', access: 'access-new' })
+})
+
+for (const reason of [
+  'quota-ineligible',
+  'account-disabled',
+  'local-mode-unavailable',
+  'refresh-backoff',
+  'invalid-grant',
+  'validation-backoff',
+] as const) {
+  test(`first identity publication preserves the ${reason} admission exclusion`, async () => {
+    const f = await fixture()
+    await f.add('a')
+    const before = await f.binding('a')
+    const publishing = barrier()
+    const finish = barrier()
+    let calls = 0
+    const c = f.coordinator({
+      refreshToken: async () => {
+        calls++
+        return { ...credential('new'), expiresIn: 3600 }
+      },
+      readAdmission: async () => ({ status: 'blocked', reason }),
+      reconcile: async (event) => {
+        if (event.status === 'persisted') {
+          publishing.release()
+          await finish.promise
+        }
+      },
+    })
+    const first = c.authorize({
+      intent: 'serve',
+      mode: 'local',
+      binding: before,
+    })
+    await publishing.promise
+    const refused = await c.authorize({
+      intent: 'serve',
+      mode: 'local',
+      binding: before,
+    })
+    finish.release()
+    expect(refused).toMatchObject({ status: 'refused', reason })
+    expect(await first).toMatchObject({ status: 'refused', reason })
+    expect(calls).toBe(1)
+  })
+}
+
+test('first identity adoption never follows a replaced epoch or a known different identity', async () => {
+  const f = await fixture()
+  await f.add('a')
+  const before = await f.binding('a')
+  let calls = 0
+  const c = f.coordinator({
+    refreshToken: async () => {
+      calls++
+      return { ...credential('new'), expiresIn: 3600 }
+    },
+  })
+  expect(
+    await c.authorize({ intent: 'serve', mode: 'local', binding: before }),
+  ).toMatchObject({ status: 'usable', access: 'access-new' })
+  expect(
+    await c.authorize({
+      intent: 'serve',
+      mode: 'local',
+      binding: { ...before, identity: 'B' },
+    }),
+  ).toMatchObject({ status: 'refused', reason: 'row-ineligible' })
+  await f.store.replace('a', credential('replacement'), { identity: 'A' })
+  await f.seedProof('a')
+  expect(
+    await c.authorize({ intent: 'serve', mode: 'local', binding: before }),
+  ).toMatchObject({ status: 'refused', reason: 'row-ineligible' })
+  expect(
+    await c.authorize({
+      intent: 'serve',
+      mode: 'local',
+      binding: await f.binding('a'),
+    }),
+  ).toMatchObject({ status: 'usable', access: 'access-replacement' })
+  expect(calls).toBe(1)
+})
+
 test('existing known job retains its alias while unknown handoff waits and both successors persist', async () => {
   const knownReconciling = barrier()
   const unknownBootstrapped = barrier()