// The Phase 7 failure drills (design §12, plan §Phase 7 F1–F5), plus the clock-skew drills F6–F8.
//
// Two independently-configured daemon stacks — two record stores, two turn ledgers, two claim
// ledgers, two agent identities on two PDSes, and two CLOCKS — over one shared fake network.
// Everything is offline and driven by an explicit fake clock and explicit ticks: no Docker, no
// wall-clock waits, no timers (plan §6 R8). Each drill asserts a claim the exit criterion makes, and
// the one-ingestion-cycle bound is asserted NUMERICALLY rather than "eventually".
//
// The two clocks are load-bearing. F1–F5 were written against a single shared clock, which made
// every question about skew unaskable: two operators that agree about the time cannot disagree
// about a lease. F6–F8 give each operator its own offset from the drill's true time and ask what
// actually breaks — a reader running fast (F6), a writer that jumps forward mid-lease (F7, and its
// unretractable record in F7b), and a writer running slow that wins every race (F8, which asserts
// the exposure that is deliberately NOT fixed; see operators.md §8).

import assert from 'node:assert/strict'
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { it } from 'node:test'
import { createSession, CredentialClient } from '../../atproto/dist/index.js'
import {
  COLLECTIONS,
  MemoryRecordStore,
  indexDigest,
  digestDifferences,
  materialize,
} from '../../core/dist/index.js'
import { MemorySyncStateStore, RepoPoller, SpaceIngestor } from '../../ingest/dist/index.js'
import { LocalPds } from '../../atproto/test/local-pds.mjs'
import {
  ClaimLedger,
  ClaimManager,
  TurnDispatcher,
  TurnLedger,
  claimRkey,
  selectDispatchable,
} from '../dist/index.js'

const ROOT = 'did:plc:root'
const AGENT_A = 'did:plc:agenta'
const AGENT_B = 'did:plc:agentb'
const OUTSIDER = 'did:plc:outsider'
const SPACE_URI = `at://${ROOT}/${COLLECTIONS.space}/space`

const TIMING = { leaseMs: 600_000, renewIntervalMs: 200_000, confirmCycles: 1, maxOutstanding: 2 }
/** `run.concurrency`'s default: one turn at a time, which is also one claim's worth of capacity. */
const CONCURRENCY = 1
const T0 = Date.parse('2026-01-01T00:00:00.000Z')

/** Writes a record straight into a repo, as though its owner had already created it. */
function seed(pds, collection, rkey, cid, value) {
  const uri = `at://${pds.did}/${collection}/${rkey}`
  pds.seq += 1
  pds.records.set(uri, { uri, cid, value })
  return { uri, cid, value }
}

const ref = (record) => ({ uri: record.uri, cid: record.cid })

/**
 * The shared network: one repo per DID, and a `RepoReadTransport` over all of them. `hidden` is how
 * a drill gives one operator a STALE view — the whole point of a multi-operator test is that the
 * two do not see the same records at the same moment.
 */
function network(repos) {
  return (hidden = new Set()) => ({
    async resolvePds(did) {
      return new URL(repos.get(did)?.service ?? 'https://missing.pds.test')
    },
    async getLatestCommit(did) {
      const pds = repos.get(did)
      if (!pds) return { rev: '0000000000000', commitCid: 'empty' }
      // The head has to move on every write — including a lease RENEWAL, which rewrites a record
      // rather than adding one — and also when a hidden record is revealed to this operator, or the
      // poller short-circuits on an unchanged head and the change is never ingested. `seq` counts
      // writes; `visible` counts what this operator is allowed to see. Both only ever grow.
      const visible = [...pds.records.values()].filter((record) => !hidden.has(record.uri)).length
      const head = pds.seq * 100 + visible
      return { rev: String(head).padStart(13, '0'), commitCid: `head-${did}-${head}` }
    },
    async listRecords({ did, collection }) {
      const pds = repos.get(did)
      if (!pds) return { records: [] }
      return {
        records: [...pds.records.values()].filter(
          (record) => record.value.$type === collection && !hidden.has(record.uri),
        ),
      }
    },
  })
}

/** The space every drill runs in: one human admin, two agent members, one OPEN plan request. */
function seedSpace(repos) {
  const root = repos.get(ROOT)
  const space = seed(root, COLLECTIONS.space, 'space', 'cid-space', {
    $type: COLLECTIONS.space,
    name: 'Radial',
    description: 'multi-operator drills',
    createdAt: '2026-01-01T00:00:00Z',
  })
  const memberships = new Map()
  for (const [index, did] of [AGENT_A, AGENT_B].entries()) {
    memberships.set(
      did,
      seed(root, COLLECTIONS.addMember, `member-${index}`, `cid-member-${index}`, {
        $type: COLLECTIONS.addMember,
        space: ref(space),
        did,
        kind: 'agent',
        role: 'agent',
        createdAt: `2026-01-01T00:00:0${index + 1}Z`,
      }),
    )
  }
  seed(root, COLLECTIONS.artifactType, 'type-plan', 'cid-type-plan', {
    $type: COLLECTIONS.artifactType,
    space: ref(space),
    name: 'plan',
    brief: 'Write a concrete plan.',
    outputSpec: { format: 'markdown', description: 'A plan' },
    scope: 'goal',
    createdAt: '2026-01-01T00:00:10Z',
  })
  const project = seed(root, COLLECTIONS.project, 'project', 'cid-project', {
    $type: COLLECTIONS.project,
    space: ref(space),
    name: 'radial-ng',
    gitUrl: 'https://example.test/radial-ng.git',
    defaultBranch: 'main',
    checks: [],
    autoReview: {},
    createdAt: '2026-01-01T00:01:00Z',
  })
  const goal = seed(root, COLLECTIONS.goal, 'goal', 'cid-goal', {
    $type: COLLECTIONS.goal,
    space: ref(space),
    project: ref(project),
    title: 'Ship the thing',
    body: 'Get it out the door.',
    createdAt: '2026-01-01T00:02:00Z',
  })
  // No assignee: the whole point. Only a claim can make this dispatchable.
  const request = seed(root, COLLECTIONS.artifactRequest, 'request-1', 'cid-request-1', {
    $type: COLLECTIONS.artifactRequest,
    goal: ref(goal),
    type: 'plan',
    basedOn: [],
    createdAt: '2026-01-01T00:03:00Z',
  })
  return { space, project, goal, request, memberships }
}

/**
 * One operator's whole stack, independent of the other's in every durable way — INCLUDING its
 * clock.
 *
 * `clock` is the drill's single true time, the only thing a drill advances. `offsetMs` is how far
 * this operator's own clock runs ahead of (or behind) it, and every place this stack reads a clock
 * goes through `local()`: the claim manager's lease arithmetic, the `asOf` it folds at, the
 * dispatch scan, the first-sighting stamp its poller writes, and the records it authors. Two
 * operators sharing one clock is the shape the first five drills were written in, and it is exactly
 * what made every skew question below unaskable.
 */
async function operator(did, repos, transportFor, clock, offsetMs = 0) {
  let offset = offsetMs
  const local = () => clock.now + offset
  const localIso = () => new Date(local()).toISOString()
  const pds = repos.get(did)
  const session = await createSession(pds.service, pds.handle, 'pw', pds.fetch.bind(pds))
  const client = new CredentialClient(session, pds.fetch.bind(pds))
  const records = new MemoryRecordStore()
  const hidden = new Set()
  const poller = new RepoPoller(transportFor(hidden), records, new MemorySyncStateStore(), {
    now: localIso,
  })
  const ingestor = new SpaceIngestor(SPACE_URI, poller, records)
  const claims = new ClaimLedger()
  const turns = new TurnLedger()
  const actor = {
    did,
    profile: 'planner',
    artifactTypes: ['plan'],
    harness: 'claude',
    models: [],
    session,
    client,
  }
  const registry = {
    all: [actor],
    byDid: new Map([[did, [actor]]]),
    select: (candidate, type) => (candidate === did && type === 'plan' ? actor : undefined),
  }
  const abandoned = []
  const logs = []
  const manager = new ClaimManager({
    ledger: claims,
    turns,
    timing: TIMING,
    now: local,
    // The production wiring, at the default `run.concurrency: 1` — `concurrency - inFlight`, with a
    // running turn standing in for a turn the dispatcher is holding a slot for. Without this dep the
    // drills claim against infinite capacity, which is not what any daemon does and hides the case
    // where the slot in question belongs to the very turn the claim is meant to cover.
    capacity: () => CONCURRENCY - turns.running().length,
    abandon: (uri, reason) => {
      abandoned.push({ uri, reason })
      // The production wiring (cli.ts) hands this straight to `TurnDispatcher.abandon`.
      self.onAbandon?.(uri, reason)
    },
    log: (message) => logs.push(message),
  })
  const self = {
    did,
    pds,
    client,
    records,
    hidden,
    claims,
    turns,
    registry,
    manager,
    abandoned,
    logs,
    syncs: 0,
    /** Move this operator's clock relative to the drill's true time — an NTP step, a resumed VM. */
    skew(ms) {
      offset = ms
    },
    /** Set by a drill that runs a real dispatcher, to complete the production wiring. */
    onAbandon: undefined,
    index() {
      return materialize(records, { spaceUri: SPACE_URI, asOf: localIso() })
    },
    /** One daemon tick: ingest, then pump claims, exactly as `runDaemon` orders them. */
    async tick() {
      const index = await self.syncOnly()
      manager.pump(index, registry)
      await manager.drain()
      return index
    },
    /** Ingest without pumping — for asserting what a fold says at an exact moment. */
    async syncOnly() {
      self.syncs += 1
      await ingestor.sync()
      return self.index()
    },
    dispatchable(index = self.index()) {
      return selectDispatchable(
        index,
        registry,
        turns,
        localIso(),
        undefined,
        claims.heldRequests(),
      )
    },
    /** What a turn does, reduced to its only protocol-visible effect. */
    async fulfil(request, goal, body = 'A plan.') {
      const written = await client.create(COLLECTIONS.artifact, {
        $type: COLLECTIONS.artifact,
        request: { uri: request.uri, cid: request.cid },
        goal: { uri: goal.uri, cid: goal.cid },
        type: 'plan',
        body,
        links: {},
        criteria: [],
        createdAt: localIso(),
      })
      turns.markFulfilled(request.uri, written)
      return written
    },
    close() {
      claims.close()
      turns.close()
    },
  }
  return self
}

/** Claim, wait out the confirm window, and end up `held` — three ticks with `confirmCycles: 1`. */
async function claimAndConfirm(op) {
  await op.tick() // writes the claim
  await op.tick() // sees it win; starts the confirm window
  return op.tick() // confirm window elapsed → held
}

async function setup({ offsetA = 0, offsetB = 0 } = {}) {
  const repos = new Map(
    [ROOT, AGENT_A, AGENT_B].map((did) => [did, new LocalPds(did)]),
  )
  const clock = { now: T0 }
  const transportFor = network(repos)
  const space = seedSpace(repos)
  const a = await operator(AGENT_A, repos, transportFor, clock, offsetA)
  const b = await operator(AGENT_B, repos, transportFor, clock, offsetB)
  return { repos, clock, space, a, b, close: () => [a, b].forEach((op) => op.close()) }
}

const artifactsIn = (index) => index.goals.flatMap((view) => view.artifacts)

// --- F1: crash mid-turn ------------------------------------------------------

it('drill 1: a crashed daemon lease lapses, the other claims and fulfils, and exactly one artifact exists', async () => {
  const { clock, space, a, b, close } = await setup()

  await claimAndConfirm(a)
  assert.deepEqual([...a.claims.heldRequests()], [space.request.uri])
  assert.deepEqual(
    a.dispatchable().map((item) => item.request.uri),
    [space.request.uri],
  )

  // Operator A dies here: no more ticks, no more renewals, and — per design §11 — no way to retract
  // the claim it holds. B must wait the lease out; nothing else can free the work.
  await b.tick()
  assert.deepEqual(b.dispatchable(), [], 'B must not touch a request A holds a live claim on')

  clock.now = T0 + TIMING.leaseMs + 1
  const lapsed = await b.tick()
  assert.equal(
    lapsed.goals[0].winningClaims[space.request.uri],
    undefined,
    'the lapsed lease leaves the request unclaimed on B index',
  )
  await b.tick()
  await b.tick()
  assert.deepEqual([...b.claims.heldRequests()], [space.request.uri])
  const dispatchable = b.dispatchable()
  assert.deepEqual(dispatchable.map((item) => item.request.uri), [space.request.uri])
  assert.equal(dispatchable[0].actor.did, AGENT_B)

  await b.fulfil(space.request, space.goal)
  const afterB = await b.tick()
  assert.equal(artifactsIn(afterB).length, 1)
  assert.equal(afterB.goals[0].openRequests.length, 0)

  // And on A's index too, once it comes back: one artifact, from B, and A's own claim settles.
  const afterA = await a.tick()
  assert.equal(artifactsIn(afterA).length, 1)
  assert.equal(artifactsIn(afterA)[0].did, AGENT_B)
  assert.equal(a.claims.get(space.request.uri, AGENT_A).state, 'settled')
  close()
})

// --- F2: competing claims ----------------------------------------------------

it('drill 2: two daemons claim at once — one dispatch, the loser gives up within ONE ingestion cycle', async () => {
  const directory = await mkdtemp(join(tmpdir(), 'radial-drill-'))
  try {
    const { clock, space, a, b, close } = await setup()

    // B claims a second earlier than A, so B wins the tie-break on both indexes. A cannot see it
    // yet — the ordinary state of two operators mid-race.
    await b.tick()
    const theirClaimUri = `at://${AGENT_B}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}`
    a.hidden.add(theirClaimUri)
    b.hidden.add(`at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}`)

    clock.now = T0 + 1_000
    await claimAndConfirm(a)
    assert.deepEqual([...a.claims.heldRequests()], [space.request.uri])

    // A dispatches on the strength of the claim it believes it won. A real container starts.
    let settleTurn
    const running = new Promise((resolve) => {
      settleTurn = resolve
    })
    const killed = []
    const dispatcher = new TurnDispatcher({
      ledger: a.turns,
      concurrency: 1,
      runDirFor: (uri) => join(directory, uri.slice(-8)),
      image: 'radial-turn:test',
      timeoutMs: 60_000,
      allowedSchemes: ['https'],
      heldClaims: () => a.claims.heldRequests(),
      kill: async (label) => {
        killed.push(label)
        settleTurn({ outcome: 'crashed' })
      },
      runTurn: () => running,
      log: () => {},
    })
    a.onAbandon = (uri, reason) => dispatcher.abandon(uri, reason)
    dispatcher.pump(a.index(), a.registry)
    assert.equal(dispatcher.inFlight, 1, 'A launched a turn for the claim it thought it held')

    // Now A ingests B's earlier claim. The bound the exit criterion asserts: ONE ingestion cycle
    // from seeing the losing evidence to giving up — not "eventually".
    a.hidden.delete(theirClaimUri)
    const syncsBefore = a.syncs
    await a.tick()
    assert.equal(a.syncs - syncsBefore, 1)
    assert.equal(a.claims.get(space.request.uri, AGENT_A).state, 'lost')
    assert.equal(a.abandoned.length, 1)
    assert.match(a.abandoned[0].reason, new RegExp(AGENT_B))
    await dispatcher.drain()
    assert.deepEqual(killed.length, 1, 'the container is killed exactly once')
    assert.equal(dispatcher.inFlight, 0)
    assert.equal(a.turns.get(space.request.uri).state, 'crashed')

    // A never dispatches it again, however many ticks pass.
    await a.tick()
    await a.tick()
    assert.deepEqual(a.dispatchable(), [])
    assert.equal(a.abandoned.length, 1, 'abandon is called at most once')

    // B, the winner, finishes the work. Exactly one artifact exists in the space.
    b.hidden.clear()
    await b.tick()
    await b.tick()
    assert.deepEqual([...b.claims.heldRequests()], [space.request.uri])
    await b.fulfil(space.request, space.goal)
    const finalB = await b.tick()
    const finalA = await a.tick()
    assert.equal(artifactsIn(finalB).length, 1)
    assert.equal(artifactsIn(finalA).length, 1)
    assert.equal(artifactsIn(finalA)[0].did, AGENT_B)
    // Both operators' folds agree about everything, which is the real exit criterion.
    assert.deepEqual(indexDigest(finalA), indexDigest(finalB))
    close()
  } finally {
    await rm(directory, { recursive: true, force: true })
  }
})

// --- F3: non-member injection ------------------------------------------------

it('drill 3: a non-member claim never counts, never gates dispatch, and says why in `ignored`', async () => {
  const { space, a, b, close } = await setup()

  // However it reached the store — a stray poll, a stream that was not filtering — the fold is what
  // decides, and an outsider holds no grant.
  const injected = {
    did: OUTSIDER,
    collection: COLLECTIONS.claim,
    rkey: 'claim-injected',
    uri: `at://${OUTSIDER}/${COLLECTIONS.claim}/claim-injected`,
    cid: 'cid-claim-injected',
    rev: '0000000000001',
    value: {
      $type: COLLECTIONS.claim,
      request: { uri: space.request.uri, cid: space.request.cid },
      // Earlier than anything either daemon will write, so an unfiltered tie-break would hand it
      // the request outright.
      expiresAt: '2027-01-01T00:00:00.000Z',
      createdAt: '2025-12-31T00:00:00.000Z',
    },
  }
  for (const op of [a, b]) op.records.put(injected)

  const index = await a.tick()
  assert.equal(index.goals[0].winningClaims[space.request.uri], undefined)
  assert.equal(
    index.ignored.some(
      (entry) => entry.uri === injected.uri && entry.reason === 'author is not an active space member',
    ),
    true,
  )

  // And it changes nothing about who does the work: A claims and confirms exactly as it would have.
  await a.tick()
  await a.tick()
  const claimed = a.index()
  assert.equal(claimed.goals[0].winningClaims[space.request.uri].did, AGENT_A)
  assert.deepEqual(
    a.dispatchable(claimed).map((item) => item.request.uri),
    [space.request.uri],
  )
  // Both operators ignore it identically — the same record, the same reason, on both folds.
  const other = await b.tick()
  assert.deepEqual(
    other.ignored.filter((entry) => entry.uri === injected.uri),
    index.ignored.filter((entry) => entry.uri === injected.uri),
  )
  close()
})

// --- F4: removed member mid-claim --------------------------------------------

it('drill 4: removing the claim holder frees the request at once, and re-adding restores their records', async () => {
  const { repos, clock, space, a, b, close } = await setup()

  // A holds a live, confirmed claim, and has said something in the thread.
  await claimAndConfirm(a)
  const message = await a.client.create(COLLECTIONS.message, {
    $type: COLLECTIONS.message,
    goal: { uri: space.goal.uri, cid: space.goal.cid },
    body: 'Working on it.',
    mentions: [],
    createdAt: '2026-01-01T00:05:00Z',
  })
  await b.tick()
  const beforeRemoval = b.index()
  assert.equal(beforeRemoval.goals[0].winningClaims[space.request.uri].did, AGENT_A)
  assert.equal(beforeRemoval.goals[0].messages.some((entry) => entry.uri === message.uri), true)
  assert.deepEqual(b.dispatchable(beforeRemoval), [], 'B leaves a request A holds alone')

  // The admin removes A. Removal revokes one exact grant and retains nothing.
  seed(repos.get(ROOT), COLLECTIONS.removeMember, 'remove-a', 'cid-remove-a', {
    $type: COLLECTIONS.removeMember,
    space: ref(space.space),
    did: AGENT_A,
    membership: ref(space.memberships.get(AGENT_A)),
    createdAt: '2026-01-01T00:06:00Z',
  })

  // Ingest without pumping, so both folds are read at the same moment: this drill is about what the
  // MATERIALIZERS say the instant the grant is revoked, not about who claims next.
  const afterRemovalB = await b.syncOnly()
  const afterRemovalA = await a.syncOnly()
  for (const index of [afterRemovalB, afterRemovalA]) {
    assert.equal(index.goals[0].winningClaims[space.request.uri], undefined, 'the claim stops counting')
    assert.equal(index.goals[0].openRequests.length, 1, 'the request is back in the open set')
    assert.equal(index.goals[0].messages.some((entry) => entry.uri === message.uri), false)
  }
  // Immediately, on BOTH materializers, with no lease to wait out — a revoked grant is not a lease.
  assert.deepEqual(indexDigest(afterRemovalA), indexDigest(afterRemovalB))

  // B picks the work up.
  await claimAndConfirm(b)
  assert.deepEqual([...b.claims.heldRequests()], [space.request.uri])
  assert.deepEqual(
    b.dispatchable().map((item) => item.request.uri),
    [space.request.uri],
  )

  // Time passes: B renews its lease, A's (still in its repo, still uncounted) lapses.
  clock.now = T0 + 300_000
  await b.tick()
  assert.equal(
    Date.parse(b.claims.get(space.request.uri, AGENT_B).expiresAt),
    clock.now + TIMING.leaseMs,
    'B renewed rather than letting its own lease run down',
  )
  clock.now = T0 + TIMING.leaseMs + 1

  // Re-adding A restores everything A ever wrote — the scenario's re-add rule, end to end.
  seed(repos.get(ROOT), COLLECTIONS.addMember, 'member-a-again', 'cid-member-a-again', {
    $type: COLLECTIONS.addMember,
    space: ref(space.space),
    did: AGENT_A,
    kind: 'agent',
    role: 'agent',
    createdAt: '2026-01-01T00:07:00Z',
  })
  const readded = await b.syncOnly()
  assert.equal(
    readded.goals[0].messages.some((entry) => entry.uri === message.uri),
    true,
    're-adding a member restores every record they wrote',
  )
  // A's restored claim comes back with it — and is ignored on its own merits, because its lease
  // lapsed while A was out. Work that moved on stays moved on.
  assert.equal(readded.goals[0].winningClaims[space.request.uri].did, AGENT_B)
  const restoredA = await a.syncOnly()
  assert.deepEqual(indexDigest(restoredA), indexDigest(readded))
  close()
})

// --- F5: divergence detection -------------------------------------------------

it('drill 5: two folds of the same records digest identically; a withheld record names its section', async () => {
  const { space, a, b, close } = await setup()
  await claimAndConfirm(a)
  await b.tick()

  const left = a.index()
  const right = b.index()
  assert.deepEqual(digestDifferences(indexDigest(left), indexDigest(right)), [])
  assert.equal(indexDigest(left).overall, indexDigest(right).overall)

  // Feed a third store the same records in reverse: order cannot change a digest.
  const reversed = new MemoryRecordStore()
  for (const record of [...a.records.records()].reverse()) reversed.put(record)
  assert.equal(
    indexDigest(materialize(reversed, { spaceUri: SPACE_URI, asOf: new Date(T0).toISOString() })).overall,
    indexDigest(materialize(a.records, { spaceUri: SPACE_URI, asOf: new Date(T0).toISOString() })).overall,
  )

  // Now withhold the claim from one of them: a real sync gap, of the kind an operator has to
  // diagnose. The digests differ, and the SECTION says it is about who holds what — not about
  // membership, and not about the artifacts.
  const gapped = new MemoryRecordStore()
  for (const record of a.records.records()) {
    if (record.collection === COLLECTIONS.claim) continue
    gapped.put(record)
  }
  const partial = indexDigest(
    materialize(gapped, { spaceUri: SPACE_URI, asOf: new Date(T0).toISOString() }),
  )
  const complete = indexDigest(left)
  assert.notEqual(partial.overall, complete.overall)
  assert.deepEqual(digestDifferences(complete, partial), ['winningClaims'])
  assert.equal(partial.sections.members, complete.sections.members)
  assert.equal(partial.sections.openRequests, complete.sections.openRequests)
  assert.equal(space.request.uri.length > 0, true)
  close()
})

// --- F6: the reader's clock runs fast ------------------------------------------

it('drill 6: an operator ten minutes fast still honours a live lease, and drops it on lag not skew', async () => {
  // The exposure this drill exists for: a lease used to be an INSTANT compared against the reader's
  // clock, so the lease B actually honoured was `leaseMs − (B's offset)`. At the 600s default, a B
  // ten minutes fast read a claim written this second as long expired, took the work, and both
  // operators did it. A lease is now a DURATION the writer declares and each observer measures from
  // its own first sighting, so B's offset cancels out entirely.
  const { clock, space, a, b, close } = await setup({ offsetB: 10 * 60_000 })

  await claimAndConfirm(a)
  await b.tick()
  const seen = b.index()
  assert.equal(seen.goals[0].winningClaims[space.request.uri].did, AGENT_A)
  assert.deepEqual(b.dispatchable(seen), [], 'B leaves a request A holds a live claim on')
  assert.deepEqual([...b.claims.heldRequests()], [], 'and does not write a competing claim')
  // Both folds agree about who holds what, at ten minutes of offset — the exit criterion.
  assert.deepEqual(digestDifferences(indexDigest(a.index()), indexDigest(seen)), [])

  // What is left is LAG, and it is bounded the way every other drill bounds it: one ingestion cycle
  // after the true expiry, B reads the request as free. Numerically, in cycles.
  clock.now = T0 + TIMING.leaseMs + 1
  const syncsBefore = b.syncs
  const lapsed = await b.syncOnly()
  assert.equal(b.syncs - syncsBefore, 1)
  assert.equal(
    lapsed.goals[0].winningClaims[space.request.uri],
    undefined,
    'a lease nobody renewed lapses on B within one cycle of its true expiry',
  )
  await b.tick()
  await b.tick()
  await b.tick()
  assert.deepEqual([...b.claims.heldRequests()], [space.request.uri])
  close()
})

// --- F7: a forward clock jump mid-lease ---------------------------------------

it('drill 7: a seven-day forward jump mid-lease buys the claimant one lease, not seven days', async () => {
  const { clock, space, a, b, close } = await setup()
  await claimAndConfirm(a)
  await b.tick()
  assert.deepEqual(b.dispatchable(), [])

  // A's clock steps a week forward between two renewals — one NTP correction, or a resumed VM.
  // Nothing can delete the record it renews, so what it writes there is permanent.
  clock.now = T0 + TIMING.renewIntervalMs + 1
  a.skew(7 * 24 * 60 * 60_000)
  const claimUri = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}`
  const before = a.pds.records.get(claimUri).value
  await a.tick()

  // It does not write next Thursday into it. On its own (wrong) clock the claim it holds is a week
  // old, which is past the protocol horizon — the point beyond which no version of THAT claim can
  // be in contract, however honest the duration it declares — so it retires the claim instead.
  const afterJump = a.pds.records.get(claimUri).value
  assert.deepEqual(afterJump, before, 'the record a jumped clock would have poisoned is untouched')
  assert.equal(
    a.logs.some((line) => /retiring our claim/.test(line)),
    true,
    'loudly, because an operator has to be able to see a claim end this way',
  )
  assert.deepEqual([...a.claims.heldRequests()], [])

  // The request still needs doing, so A claims it again — in a NEW record, at the next generation,
  // with an honest `createdAt` of its own. It is dated a week out, which costs A the tie-break
  // against any competitor and buys it nothing: every observer measures the lease from arrival.
  await a.tick()
  const nextUri = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid, 1)}`
  const reclaimed = a.pds.records.get(nextUri).value
  assert.equal(Date.parse(reclaimed.expiresAt) - Date.parse(reclaimed.renewedAt), TIMING.leaseMs)
  assert.equal(Date.parse(reclaimed.createdAt) > clock.now + 6 * 24 * 60 * 60_000, true)

  // So B honours it for one lease from the moment B saw it, and no longer. (Both of A's records are
  // A's, so while the first is still live B reads the earlier `createdAt` as the winner — which is
  // the tie-break doing exactly what it should.)
  await b.tick()
  assert.equal(b.index().goals[0].winningClaims[space.request.uri].did, AGENT_A)
  clock.now = T0 + TIMING.leaseMs + 1
  assert.equal((await b.syncOnly()).goals[0].winningClaims[space.request.uri].uri, nextUri)
  clock.now += TIMING.leaseMs + 1
  assert.equal(
    (await b.syncOnly()).goals[0].winningClaims[space.request.uri],
    undefined,
    'a claim written on a week-fast clock lapses one lease after it arrived, not one week',
  )
  await b.tick()
  await b.tick()
  await b.tick()
  assert.deepEqual([...b.claims.heldRequests()], [space.request.uri])
  close()
})

it('drill 7d: the jumped clock is corrected, and the claim it wrote is retired rather than held forever', async () => {
  // Walk drill 7 one step further, to where a clock jump normally ends: NTP puts it back. What A is
  // left holding is a record every field of which was stamped by the fast clock — a legal lease
  // inside a legal horizon, so it is IN CONTRACT and therefore invisible to every recovery path
  // there was. Not lapsed (it expires next week), not due for renewal, not refused by the fold (so
  // no repair, and the store would not adopt a shortening rewrite of a conforming version anyway),
  // and no other winner for the loss branch to name. The row would sit `held` forever, gating
  // dispatch behind a claim its own writer can neither renew nor retract — the one recovery this
  // whole change would otherwise have REMOVED rather than added.
  const { clock, space, a, b, close } = await setup()
  a.skew(7 * 24 * 60 * 60_000)
  await claimAndConfirm(a)
  const first = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}`
  const written = a.pds.records.get(first).value
  assert.equal(Date.parse(written.createdAt), clock.now + 7 * 24 * 60 * 60_000)
  assert.deepEqual([...a.claims.heldRequests()], [space.request.uri])

  // The correction lands. A's very next pump ends that claim and starts another beside it, dated by
  // the clock it has now — in the same pump, so nothing is left uncovered for a tick.
  a.skew(0)
  await a.tick()
  assert.equal(
    a.logs.some((line) => /retiring our claim/.test(line) && /into this daemon's own future/.test(line)),
    true,
    'and says which clock it distrusts, since an operator cannot see this any other way',
  )
  const next = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid, 1)}`
  const honest = a.pds.records.get(next).value
  assert.equal(Date.parse(honest.createdAt), clock.now)
  assert.equal(Date.parse(honest.expiresAt) - Date.parse(honest.renewedAt), TIMING.leaseMs)
  assert.deepEqual(a.pds.records.get(first).value, written, 'the record from the future is left alone')

  // And the work goes on: the new claim confirms, A dispatches again, and B — which honours the old
  // record for one lease from arrival and the new one on its earlier `createdAt` — agrees A holds it.
  await a.tick()
  await a.tick()
  assert.deepEqual([...a.claims.heldRequests()], [space.request.uri])
  assert.equal(a.claims.get(space.request.uri, AGENT_A).generation, 1)
  assert.deepEqual(
    a.dispatchable().map((item) => item.request.uri),
    [space.request.uri],
  )
  await b.tick()
  const winner = b.index().goals[0].winningClaims[space.request.uri]
  assert.equal(winner.uri, next)
  assert.equal(winner.did, AGENT_A)
  assert.deepEqual([...b.claims.heldRequests()], [], 'B never contests a request A is covering')
  close()
})

it('drill 7f: a correction the next renewal still reaches costs no claim, and no peer sees anything', async () => {
  // The other side of drill 7d's boundary, and the reason the boundary is where it is. A jump of Δ
  // that the RENEWAL branch caught repairs itself: the renewal trigger is
  // `remaining <= leaseMs − renewIntervalMs`, so it falls due Δ + one interval after the correction,
  // while observers stop honouring the version one declared lease after it arrived. For Δ up to
  // `leaseMs − renewIntervalMs` the repair lands first, so retiring there would buy nothing a peer
  // could ever have seen — and cost a generation plus, since the retired record keeps the earlier
  // `createdAt` and wins A's own tie-break until it lapses, a stretch of A's own dispatch.
  const { clock, space, a, b, close } = await setup()
  await claimAndConfirm(a)
  await b.tick()
  const uri = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}`

  // Two minutes fast when the renewal falls due — a resumed VM, a chronyd step.
  const jump = 2 * 60_000
  assert.ok(jump <= TIMING.leaseMs - TIMING.renewIntervalMs, 'inside what a renewal still reaches')
  clock.now = T0 + TIMING.renewIntervalMs
  a.skew(jump)
  await a.tick()
  const jumped = a.pds.records.get(uri)
  assert.equal(Date.parse(jumped.value.renewedAt), clock.now + jump, 'the renewal came off the fast clock')

  // NTP puts the clock back. What A holds is in contract by both bounds, sits ahead of the clock A
  // now has, and no other branch would ever move it — but the renewal that will move it is due
  // before any observer stops honouring it, so A leaves it exactly where it is.
  a.skew(0)
  await a.tick()
  assert.equal(a.logs.some((line) => /retiring our claim/.test(line)), false, 'nothing is retired')
  assert.equal(a.pds.records.get(uri).cid, jumped.cid, 'and nothing is rewritten yet')
  assert.deepEqual([...a.claims.heldRequests()], [space.request.uri], 'A still holds the work')
  assert.equal(a.claims.get(space.request.uri, AGENT_A).generation, 0, 'no generation is burned')
  assert.deepEqual(
    a.dispatchable().map((item) => item.request.uri),
    [space.request.uri],
    'and it stays dispatchable throughout, which is what retiring here would have cost',
  )

  // The repair is the ordinary renewal, dated by the clock A actually has.
  clock.now = Date.parse(jumped.value.expiresAt) - (TIMING.leaseMs - TIMING.renewIntervalMs)
  await a.tick()
  const repaired = a.pds.records.get(uri).value
  assert.equal(Date.parse(repaired.renewedAt), clock.now)
  assert.equal(Date.parse(repaired.expiresAt) - Date.parse(repaired.renewedAt), TIMING.leaseMs)

  // And B read A as the holder at every step, so there was never a divergence to close.
  await b.tick()
  assert.equal(b.index().goals[0].winningClaims[space.request.uri].uri, uri)
  assert.equal(b.claims.all().length, 0, 'B never contests a request A has held all along')
  close()
})

it('drill 7c: a claim that reaches its horizon is retired and re-claimed, with no gap in the work', async () => {
  // What the horizon costs a healthy claim, and how it is paid. A request can stay open far longer
  // than a day — one waiting on a human answer routinely does — and `createdAt` is frozen within a
  // claim, so at some point the claim's own record can no longer carry a conforming lease. It ends
  // there, and the request is claimed again beside it rather than held by a version every observer
  // would refuse.
  const { clock, space, a, b, close } = await setup()
  await claimAndConfirm(a)
  await b.tick()
  const first = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}`

  // Renewals keep it alive up to the horizon, and the LAST one stops exactly there rather than
  // reaching past it — a lease clamped to where every observer would have stopped honouring it.
  const horizon = T0 + 24 * 60 * 60 * 1000
  clock.now = horizon - 100_000
  await a.tick()
  const last = a.pds.records.get(first)
  assert.equal(Date.parse(last.value.expiresAt), horizon, 'the final lease expires at the horizon exactly')
  assert.equal(Date.parse(last.value.expiresAt) - Date.parse(last.value.createdAt), 24 * 60 * 60 * 1000)

  // And it is written once, not on every tick of the interval before the horizon: a renewal that
  // cannot move `expiresAt` is a write for nothing.
  await a.tick()
  await a.tick()
  assert.equal(a.pds.records.get(first).cid, last.cid)
  await b.tick()
  assert.equal(b.index().goals[0].winningClaims[space.request.uri].uri, first)

  // Past the horizon the claim is over. Retired once — the row is terminal, so no tick after this
  // one repeats it — and never rewritten with a lease the fold would refuse.
  clock.now = horizon + 1
  await a.tick()
  assert.deepEqual(a.pds.records.get(first).value, last.value)
  assert.equal(
    a.logs.filter((line) => /retiring our claim/.test(line)).length,
    1,
    'retired once, not once per tick',
  )
  assert.deepEqual([...a.claims.heldRequests()], [])

  // And the work continues under generation 1: a fresh record, a fresh createdAt, confirmed the
  // ordinary way, and B agrees.
  await a.tick()
  await a.tick()
  const next = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid, 1)}`
  assert.deepEqual([...a.claims.heldRequests()], [space.request.uri])
  assert.equal(a.claims.get(space.request.uri, AGENT_A).generation, 1)
  assert.deepEqual(
    a.dispatchable().map((item) => item.request.uri),
    [space.request.uri],
  )
  await b.tick()
  const winner = b.index().goals[0].winningClaims[space.request.uri]
  assert.equal(winner.uri, next)
  assert.equal(winner.did, AGENT_A)
  close()
})

it('drill 7e: a claim retired while its own turn is running is replaced in the same pump, at concurrency 1', async () => {
  // The routine way drill 7c's retirement arrives: a request answered by a human in the last hour
  // of its claim's day dispatches a turn that is still running when the horizon passes. Retirement
  // is terminal, so nothing renews the old record back — if the daemon could not re-claim until the
  // turn ended, the request would read as free to every peer for the rest of a turn (up to
  // `run.timeoutMs`, 60 minutes by default) rather than for a pump.
  //
  // The turn-eligibility exemption alone does not get there. At the default `concurrency: 1` the
  // running turn IS the daemon's one turn slot, so free capacity is zero for exactly as long as the
  // covering claim is needed: charged against it, the re-claim is refused before the exemption is
  // ever consulted.
  const { clock, space, a, b, close } = await setup()
  await claimAndConfirm(a)
  await b.tick()
  a.turns.markRunning(space.request.uri, space.request.cid, {
    containerLabel: 'radial-turn-1',
    checkoutPath: '/tmp/radial-turn-1',
  })
  assert.equal(a.turns.running().length, CONCURRENCY, 'no free turn capacity while it runs')

  clock.now = T0 + 24 * 60 * 60 * 1000 + 1
  await a.tick()
  assert.equal(a.logs.some((line) => /retiring our claim/.test(line)), true)
  assert.equal(a.claims.get(space.request.uri, AGENT_A).generation, 1, 'covered again in the same pump')
  assert.equal(a.turns.get(space.request.uri).state, 'running', 'and the turn is never killed for it')
  const next = `at://${AGENT_A}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid, 1)}`
  assert.equal(Date.parse(a.pds.records.get(next).value.createdAt), clock.now)

  // B's very next tick — the first moment it could have taken the work — finds it covered: it
  // ingests the replacement before it decides anything, and never writes a claim of its own.
  await b.tick()
  assert.equal(b.index().goals[0].winningClaims[space.request.uri].uri, next)
  assert.deepEqual([...b.claims.heldRequests()], [])
  assert.equal(b.claims.all().length, 0, 'B never contests a request A is still working')
  close()
})

it('drill 7b: a claim record already poisoned by a clock jump is repaired by its own writer', async () => {
  // The wedge the clamp would otherwise create. The rkey is a pure function of the request and
  // nothing deletes records, so a claim written with an absurd `expiresAt` — by a pre-`renewedAt`
  // daemon whose clock jumped — is permanent. If the fold merely ignored it, A's next attempt would
  // collide at that rkey, adopt the record, sit in `confirming` (which has no timeout) forever, and
  // burn an outstanding slot: at the default of one, A would claim nothing in this space ever again.
  const { space, a, b, close } = await setup()
  const rkey = claimRkey(space.request.uri, space.request.cid)
  seed(a.pds, COLLECTIONS.claim, rkey, 'cid-claim-poisoned', {
    $type: COLLECTIONS.claim,
    request: ref(space.request),
    expiresAt: '2026-02-01T00:00:00.000Z',
    createdAt: '2026-01-01T00:00:00.000Z',
  })

  const before = await a.syncOnly()
  assert.equal(before.goals[0].winningClaims[space.request.uri], undefined, 'no observer honours it')
  assert.equal(
    before.ignored.some(
      (entry) =>
        entry.uri === `at://${AGENT_A}/${COLLECTIONS.claim}/${rkey}` &&
        /outside the protocol horizon/.test(entry.reason),
    ),
    true,
    'and the fold says why, rather than leaving the request open with no explanation',
  )

  // A claims, collides with its own poisoned record, and rewrites it with a conforming lease.
  await a.tick()
  assert.equal(
    a.logs.some((line) => /out of contract/.test(line)),
    true,
    'loudly — an operator has to be able to see this happened',
  )
  const repaired = a.pds.records.get(`at://${AGENT_A}/${COLLECTIONS.claim}/${rkey}`).value
  assert.equal(Date.parse(repaired.expiresAt) - Date.parse(repaired.renewedAt), TIMING.leaseMs)
  assert.equal(repaired.createdAt, '2026-01-01T00:00:00.000Z', 'the tie-break position is untouched')

  // And the row is not wedged: it confirms, it is dispatchable, and B agrees A holds it.
  await a.tick()
  await a.tick()
  assert.deepEqual([...a.claims.heldRequests()], [space.request.uri])
  assert.deepEqual(
    a.dispatchable().map((item) => item.request.uri),
    [space.request.uri],
  )
  await b.tick()
  assert.equal(b.index().goals[0].winningClaims[space.request.uri].did, AGENT_A)
  close()
})

// --- F8: the tie-break is self-reported, and stays that way --------------------

it('drill 8: a daemon whose clock runs an hour slow wins every race — the documented, unfixed gap', async () => {
  // Deliberately asserting TODAY's behaviour, permanently. The tie-break is earliest `createdAt`,
  // self-reported, with nothing corroborating it: `rev` is observer-local by construction and the
  // claim rkey is a content hash with no time in it, so "lowest URI" reduces to lowest DID — a fixed
  // bias, not a random one. Hardening it means reintroducing a corroborating time source, which
  // changes a wire-contract rule the digest names; see the plan's §7 and operators.md §8.
  //
  // Note what first-seen liveness does NOT do here: it makes the backward-skewed claim MORE live,
  // not less, because its declared lease is measured from arrival rather than from an `expiresAt`
  // that is already in the reader's past. Fairness is a separate problem from safety.
  for (const attempt of [1, 2, 3]) {
    const { space, a, b, close } = await setup({ offsetB: -60 * 60_000 })
    // A genuine race: each claims in the same true instant without having seen the other, which is
    // the only way to test the tie-break rather than the do-not-contest rule.
    const claimOf = (did) =>
      `at://${did}/${COLLECTIONS.claim}/${claimRkey(space.request.uri, space.request.cid)}`
    a.hidden.add(claimOf(AGENT_B))
    b.hidden.add(claimOf(AGENT_A))
    await a.tick()
    await b.tick()
    a.hidden.clear()
    b.hidden.clear()
    const winners = [await a.syncOnly(), await b.syncOnly()].map(
      (index) => index.goals[0].winningClaims[space.request.uri]?.did,
    )
    assert.deepEqual(winners, [AGENT_B, AGENT_B], `attempt ${attempt}: the slow clock wins, on both folds`)
    // B's claim is an hour old by its own account and still counted, which is the point: nothing
    // corroborates a self-reported `createdAt`, so the bias is permanent rather than accidental.
    const claim = (await a.syncOnly()).goals[0].winningClaims[space.request.uri]
    assert.equal(Date.parse(claim.value.createdAt), T0 - 60 * 60_000)
    close()
  }
})