import assert from 'node:assert/strict'
import { it } from 'node:test'
import { XrpcError } from '../../atproto/dist/index.js'
import { COLLECTIONS, MemoryRecordStore, claimWithinContract, materialize } from '../../core/dist/index.js'
import {
  ClaimLedger,
  ClaimManager,
  MAX_CLAIM_GENERATIONS,
  TurnLedger,
  actorTypesFor,
  claimRkey,
  selectClaimable,
} from '../dist/index.js'

const ROOT = 'did:plc:root'
const HUMAN = 'did:plc:human'
const AGENT = 'did:plc:agent'
const OTHER = 'did:plc:otheragent'
const SPACE_URI = `at://${ROOT}/${COLLECTIONS.space}/space`
const NOW = Date.parse('2026-01-01T01:00:00.000Z')

function mk(did, collection, rkey, cid, value) {
  return { did, collection, rkey, uri: `at://${did}/${collection}/${rkey}`, cid, rev: '0000000000001', value }
}
const ref = (r) => ({ uri: r.uri, cid: r.cid })

/**
 * One space, one human, two agent members, a goal, and one OPEN (unassigned) plan request.
 *
 * `suffix` gives a second, entirely separate space: every rkey and cid is distinct, so the two
 * spaces share no record URI and one daemon-wide claim ledger can hold rows for both.
 */
function buildScenario({ requestOverrides = {}, extra = [], suffix = '' } = {}) {
  const records = []
  const add = (r) => {
    records.push(r)
    return r
  }
  const space = add(
    mk(ROOT, COLLECTIONS.space, `space${suffix}`, `cid-space${suffix}`, {
      $type: COLLECTIONS.space,
      name: 'Radial',
      description: 'claims test',
      createdAt: '2026-01-01T00:00:00Z',
    }),
  )
  add(
    mk(ROOT, COLLECTIONS.addMember, `member-human${suffix}`, `cid-mh${suffix}`, {
      $type: COLLECTIONS.addMember,
      space: ref(space),
      did: HUMAN,
      kind: 'human',
      role: 'member',
      createdAt: '2026-01-01T00:00:01Z',
    }),
  )
  for (const [index, did] of [AGENT, OTHER].entries()) {
    add(
      mk(ROOT, COLLECTIONS.addMember, `member-${did.split(':').at(-1)}${suffix}`, `cid-m-${index}${suffix}`, {
        $type: COLLECTIONS.addMember,
        space: ref(space),
        did,
        kind: 'agent',
        role: 'agent',
        createdAt: `2026-01-01T00:00:0${index + 2}Z`,
      }),
    )
  }
  add(
    mk(ROOT, COLLECTIONS.artifactType, `type-plan${suffix}`, `cid-type-plan${suffix}`, {
      $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 = add(
    mk(HUMAN, COLLECTIONS.project, `project${suffix}`, `cid-project${suffix}`, {
      $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 = add(
    mk(HUMAN, COLLECTIONS.goal, `goal${suffix}`, `cid-goal${suffix}`, {
      $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',
    }),
  )
  const request = add(
    mk(HUMAN, COLLECTIONS.artifactRequest, `request-1${suffix}`, `cid-request-1${suffix}`, {
      $type: COLLECTIONS.artifactRequest,
      goal: ref(goal),
      type: 'plan',
      basedOn: [],
      createdAt: '2026-01-01T00:03:00Z',
      ...requestOverrides,
    }),
  )
  for (const r of extra) add(r)
  return { records, space, project, goal, request }
}

function indexOf(records, spaceUri = SPACE_URI) {
  const store = new MemoryRecordStore()
  for (const record of records) store.put(record)
  return materialize(store, { spaceUri, asOf: new Date(NOW).toISOString() })
}

const claimRecord = (request, did, { createdAt = '2026-01-01T00:30:00.000Z', expiresAt = '2026-01-01T02:00:00.000Z' } = {}) =>
  mk(did, COLLECTIONS.claim, claimRkey(request.uri, request.cid), `cid-claim-${did.split(':').at(-1)}`, {
    $type: COLLECTIONS.claim,
    request: ref(request),
    expiresAt,
    createdAt,
  })

/**
 * A fake PDS client: records every write, so a test asserts the exact record shape. It also KEEPS
 * what it wrote, keyed by rkey, so a second create at the same rkey collides the way a real repo
 * would — which is the whole mechanism behind adopting a claim after a restart, and behind walking
 * to the next generation when the record at this one is past repair.
 */
function fakeClient(did) {
  const writes = []
  /** Every read-back, so a test can assert a path costs no PDS round trip it did not need. */
  const reads = []
  return {
    writes,
    reads,
    stored: new Map(),
    fail: undefined,
    /** Set when only the CREATE should fail — a collision at the deterministic rkey, which the
     * manager recovers from by writing a `put` that must be allowed to land. */
    failCreateOnly: false,
    existing: undefined,
    async create(collection, value, options = {}) {
      if (this.fail) throw this.fail
      if (this.stored.has(options.rkey)) {
        throw new XrpcError(400, 'RecordAlreadyExists', 'record already exists')
      }
      writes.push({ kind: 'create', collection, value, rkey: options.rkey })
      const cid = `cid-write-${writes.length}`
      this.stored.set(options.rkey, { cid, value })
      return { uri: `at://${did}/${collection}/${options.rkey ?? 'auto'}`, cid }
    },
    async put(collection, uri, value, options = {}) {
      if (this.fail && !this.failCreateOnly) throw this.fail
      writes.push({ kind: 'put', collection, uri, value, swapRecord: options.swapRecord })
      const cid = `cid-write-${writes.length}`
      this.stored.set(uri.split('/').at(-1), { cid, value })
      return { uri, cid }
    },
    async getOwnRecord(collection, rkey) {
      reads.push(rkey)
      const held = this.existing ?? this.stored.get(rkey)
      return held ? { uri: `at://${did}/${collection}/${rkey}`, cid: held.cid, value: held.value } : undefined
    },
  }
}

function actorFor(did, artifactTypes = ['plan'], spaces = undefined) {
  return {
    did,
    profile: 'planner',
    handleName: 'planner',
    artifactTypes,
    ...(spaces ? { spaces } : {}),
    harness: 'claude',
    models: [],
    session: { did, handle: `${did.split(':').at(-1)}.test`, service: 'https://pds.test', accessJwt: 'a', refreshJwt: 'r' },
    client: fakeClient(did),
  }
}

function registryFor(actors) {
  const byDid = new Map()
  for (const actor of actors) {
    const list = byDid.get(actor.did) ?? []
    list.push(actor)
    byDid.set(actor.did, list)
  }
  return {
    all: actors,
    byDid,
    // The real registry's rule, not a simpler one: capabilities are per space (design §11), and a
    // stand-in that ignored the space would let a test pass over a daemon that could not.
    select: (did, type, spaceUri) =>
      byDid.get(did)?.find((actor) => actorTypesFor(actor, spaceUri).includes(type)),
  }
}

const TIMING = { leaseMs: 600_000, renewIntervalMs: 200_000, confirmCycles: 1, maxOutstanding: 2 }

function stack(
  records,
  { actors = [actorFor(AGENT)], timing = TIMING, now = () => NOW, abandon, capacity } = {},
) {
  const claims = new ClaimLedger()
  const turns = new TurnLedger()
  const logs = []
  const manager = new ClaimManager({
    ledger: claims,
    turns,
    timing,
    now,
    ...(abandon ? { abandon } : {}),
    ...(capacity ? { capacity } : {}),
    log: (message) => logs.push(message),
  })
  return {
    claims,
    turns,
    manager,
    logs,
    registry: registryFor(actors),
    actors,
    index: (extra = []) => indexOf([...records, ...extra]),
    close: () => {
      claims.close()
      turns.close()
    },
  }
}

// --- selectClaimable --------------------------------------------------------

it('claims an open unassigned request of a type one of our agents produces', () => {
  const base = buildScenario()
  const s = stack(base.records)
  const candidates = selectClaimable(s.index(), s.registry, s.claims, s.turns, { budget: 5 })
  assert.deepEqual(candidates.map((candidate) => candidate.request.uri), [base.request.uri])
  assert.equal(candidates[0].actor.did, AGENT)
  s.close()
})

it('does not claim work a profile is scoped out of here, and still claims it next door', () => {
  // One profile, two spaces, one config: `plan` everywhere except this space, where it is scoped to
  // implementations only. The open request here is a plan.
  const here = buildScenario()
  const scoped = actorFor(AGENT, ['plan'], { [SPACE_URI]: { artifactTypes: ['implementation'] } })
  const s = stack(here.records, { actors: [scoped] })
  assert.deepEqual(selectClaimable(s.index(), s.registry, s.claims, s.turns, { budget: 5 }), [])
  s.close()

  // The same profile in a space it does not scope: the global list applies and it claims.
  const next = buildScenario({ suffix: 'two' })
  const t = stack(next.records, { actors: [scoped] })
  const candidates = selectClaimable(
    indexOf(next.records, `at://${ROOT}/${COLLECTIONS.space}/spacetwo`),
    t.registry,
    t.claims,
    t.turns,
    { budget: 5 },
  )
  assert.deepEqual(candidates.map((candidate) => candidate.request.uri), [next.request.uri])
  t.close()
})

it('never claims an assigned request — the single-operator path takes no claim at all', () => {
  const base = buildScenario({ requestOverrides: { assignee: AGENT } })
  const s = stack(base.records)
  assert.deepEqual(selectClaimable(s.index(), s.registry, s.claims, s.turns, { budget: 5 }), [])
  s.close()
})

it('does not contest a live claim another operator already won', () => {
  const base = buildScenario()
  const theirs = claimRecord(base.request, OTHER, { createdAt: '2026-01-01T00:10:00.000Z' })
  const s = stack(base.records, { actors: [actorFor(AGENT)] })
  assert.deepEqual(selectClaimable(s.index([theirs]), s.registry, s.claims, s.turns, { budget: 5 }), [])
  s.close()
})

it('claims a request whose winning claim has LAPSED — that is how a crashed daemon work comes back', () => {
  const base = buildScenario()
  const lapsed = claimRecord(base.request, OTHER, {
    createdAt: '2026-01-01T00:10:00.000Z',
    expiresAt: '2026-01-01T00:20:00.000Z', // before NOW
  })
  const s = stack(base.records)
  const candidates = selectClaimable(s.index([lapsed]), s.registry, s.claims, s.turns, { budget: 5 })
  assert.deepEqual(candidates.map((candidate) => candidate.request.uri), [base.request.uri])
  s.close()
})

it('does not claim what we already hold, what we cannot produce, or what the turn ledger gave up on', () => {
  const base = buildScenario()
  const held = stack(base.records)
  held.claims.markClaiming({
    requestUri: base.request.uri,
    requestCid: base.request.cid,
    actorDid: AGENT,
    spaceUri: SPACE_URI,
    rkey: 'claim-x',
    createdAt: '2026-01-01T00:30:00.000Z',
  })
  held.claims.markHeld(base.request.uri, AGENT)
  assert.deepEqual(selectClaimable(held.index(), held.registry, held.claims, held.turns, { budget: 5 }), [])
  held.close()

  const wrongType = stack(base.records, { actors: [actorFor(AGENT, ['adr'])] })
  assert.deepEqual(
    selectClaimable(wrongType.index(), wrongType.registry, wrongType.claims, wrongType.turns, { budget: 5 }),
    [],
  )
  wrongType.close()

  const gaveUp = stack(base.records)
  gaveUp.turns.giveUp(base.request.uri, base.request.cid)
  assert.deepEqual(selectClaimable(gaveUp.index(), gaveUp.registry, gaveUp.claims, gaveUp.turns, { budget: 5 }), [])
  gaveUp.close()
})

it('does not claim a request blocked on our own unanswered question', () => {
  const base = buildScenario()
  const question = mk(AGENT, COLLECTIONS.message, 'q', 'cid-q', {
    $type: COLLECTIONS.message,
    goal: ref(base.goal),
    body: 'which approach?',
    mentions: [],
    re: ref(base.request),
    createdAt: '2026-01-01T00:04:00Z',
  })
  const s = stack([...base.records, question])
  assert.deepEqual(selectClaimable(s.index(), s.registry, s.claims, s.turns, { budget: 5 }), [])
  s.close()
})

it('claims a request whose own turn is RUNNING, and still leaves one in backoff alone', () => {
  // The turn-eligibility gate exists so the daemon does not claim work it would then refuse to
  // dispatch. A turn of ours already running is the opposite situation: we are the ones doing the
  // work, and a claim retired at its horizon mid-turn has to be replaceable straight away —
  // otherwise the request reads as free to every peer for the rest of the turn (up to a full
  // `run.timeoutMs`), and nothing on this daemon can cover it, retirement being terminal.
  const base = buildScenario()
  const running = stack(base.records)
  running.turns.markRunning(base.request.uri, base.request.cid, {
    containerLabel: 'radial-turn-1',
    checkoutPath: '/tmp/radial-turn-1',
  })
  assert.deepEqual(
    selectClaimable(running.index(), running.registry, running.claims, running.turns, { budget: 5 }).map(
      (candidate) => candidate.request.uri,
    ),
    [base.request.uri],
  )
  // And it survives the OTHER cap, which is what quietly undid the exemption: the running turn it
  // covers is itself the slot that makes free capacity zero on the default `concurrency: 1`, so a
  // covering claim charged against turn capacity can never be selected in the one case it is for.
  const covering = selectClaimable(running.index(), running.registry, running.claims, running.turns, {
    budget: 5,
    capacity: 0,
  })
  assert.deepEqual(covering.map((candidate) => candidate.request.uri), [base.request.uri])
  assert.equal(covering[0].covering, true)
  // The outstanding budget still binds it — a covering claim is a claim.
  assert.deepEqual(
    selectClaimable(running.index(), running.registry, running.claims, running.turns, {
      budget: 0,
      capacity: 5,
    }),
    [],
  )
  running.close()

  // A crashed turn inside its cooldown is the ordinary ineligible case, and it stays ineligible:
  // nothing of ours is in flight, so there is nothing to cover.
  const crashed = stack(base.records)
  crashed.turns.markRunning(base.request.uri, base.request.cid, {
    containerLabel: 'radial-turn-1',
    checkoutPath: '/tmp/radial-turn-1',
  })
  crashed.turns.markCrashed(base.request.uri)
  assert.deepEqual(
    selectClaimable(crashed.index(), crashed.registry, crashed.claims, crashed.turns, {
      budget: 5,
      now: new Date(NOW).toISOString(),
    }),
    [],
  )
  crashed.close()
})

it('honours the budget and orders deterministically by request uri', () => {
  const base = buildScenario()
  const second = mk(HUMAN, COLLECTIONS.artifactRequest, 'request-0', 'cid-request-0', {
    $type: COLLECTIONS.artifactRequest,
    goal: ref(base.goal),
    type: 'plan',
    basedOn: [],
    createdAt: '2026-01-01T00:03:30Z',
  })
  const s = stack([...base.records, second])
  const all = selectClaimable(s.index(), s.registry, s.claims, s.turns, { budget: 5 })
  assert.deepEqual(all.map((candidate) => candidate.request.uri), [second.uri, base.request.uri].sort())
  const capped = selectClaimable(s.index(), s.registry, s.claims, s.turns, { budget: 1 })
  assert.equal(capped.length, 1)
  assert.equal(capped[0].request.uri, all[0].request.uri)
  assert.deepEqual(selectClaimable(s.index(), s.registry, s.claims, s.turns, { budget: 0 }), [])

  // With the budget down to one, a request whose turn is already running takes it: `budget` binds
  // covering claims too, and losing the last slot to work not yet started would put a request this
  // daemon is actively working back in front of every peer for the rest of its turn.
  s.turns.markRunning(base.request.uri, base.request.cid, {
    containerLabel: 'radial-turn-1',
    checkoutPath: '/tmp/radial-turn-1',
  })
  const contested = selectClaimable(s.index(), s.registry, s.claims, s.turns, { budget: 1 })
  assert.deepEqual(contested.map((candidate) => candidate.request.uri), [base.request.uri])
  s.close()
})

// --- ClaimManager.pump ------------------------------------------------------

it('writes a claim at the deterministic rkey and waits a full cycle before holding it', async () => {
  const base = buildScenario()
  const s = stack(base.records)
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()

  const write = s.actors[0].client.writes[0]
  assert.equal(write.kind, 'create')
  assert.equal(write.collection, COLLECTIONS.claim)
  assert.equal(write.rkey, claimRkey(base.request.uri, base.request.cid))
  assert.deepEqual(write.value.request, { uri: base.request.uri, cid: base.request.cid })
  assert.equal(write.value.createdAt, '2026-01-01T01:00:00.000Z')
  assert.equal(write.value.expiresAt, '2026-01-01T01:10:00.000Z')
  assert.equal(s.claims.get(base.request.uri, AGENT).state, 'confirming')
  assert.deepEqual([...s.manager.heldClaims()], [])

  // The claim is now in the fold and it is ours — but a confirming row needs `confirmCycles` full
  // ingestion cycles to pass, so the FIRST index that shows it winning is not enough.
  const ours = {
    ...claimRecord(base.request, AGENT, { createdAt: '2026-01-01T01:00:00.000Z' }),
    uri: write.rkey ? `at://${AGENT}/${COLLECTIONS.claim}/${write.rkey}` : undefined,
  }
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.deepEqual([...s.manager.heldClaims()], [])
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.deepEqual([...s.manager.heldClaims()], [base.request.uri])
  s.close()
})

it('loses to an earlier claim, abandons the turn, and stops renewing', async () => {
  const base = buildScenario()
  const abandoned = []
  const s = stack(base.records, { abandon: (uri, reason) => abandoned.push({ uri, reason }) })
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  const rkey = claimRkey(base.request.uri, base.request.cid)
  const ours = {
    ...claimRecord(base.request, AGENT, { createdAt: '2026-01-01T01:00:00.000Z' }),
    uri: `at://${AGENT}/${COLLECTIONS.claim}/${rkey}`,
  }
  s.manager.pump(s.index([ours]), s.registry)
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.deepEqual([...s.manager.heldClaims()], [base.request.uri])

  // The other operator's claim was written EARLIER; once we ingest it, it wins on both indexes.
  const theirs = claimRecord(base.request, OTHER, { createdAt: '2026-01-01T00:59:00.000Z' })
  s.manager.pump(s.index([ours, theirs]), s.registry)
  await s.manager.drain()
  assert.equal(s.claims.get(base.request.uri, AGENT).state, 'lost')
  assert.deepEqual([...s.manager.heldClaims()], [])
  assert.equal(abandoned.length, 1)
  assert.match(abandoned[0].reason, new RegExp(OTHER))

  // Terminal: a second observation of the same loss does not abandon twice or write anything more.
  const writes = s.actors[0].client.writes.length
  s.manager.pump(s.index([ours, theirs]), s.registry)
  await s.manager.drain()
  assert.equal(abandoned.length, 1)
  assert.equal(s.actors[0].client.writes.length, writes)
  s.close()
})

it('renews by moving expiresAt and nothing else, pinned with swapRecord', async () => {
  const base = buildScenario()
  let now = NOW
  const s = stack(base.records, { now: () => now })
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  const rkey = claimRkey(base.request.uri, base.request.cid)
  const ours = {
    ...claimRecord(base.request, AGENT, { createdAt: '2026-01-01T01:00:00.000Z' }),
    uri: `at://${AGENT}/${COLLECTIONS.claim}/${rkey}`,
  }
  s.manager.pump(s.index([ours]), s.registry)
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()

  // Not yet due.
  const before = s.actors[0].client.writes.length
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.equal(s.actors[0].client.writes.length, before)

  now = NOW + TIMING.renewIntervalMs + 1
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  const renewal = s.actors[0].client.writes.at(-1)
  assert.equal(renewal.kind, 'put')
  // Pinned to the CID we last read, so a concurrent rewrite fails the write instead of clobbering it.
  assert.equal(renewal.swapRecord, 'cid-write-1')
  assert.equal(renewal.uri, ours.uri)
  const created = s.actors[0].client.writes[0].value
  // Byte-identical but for the two lease fields — anything else and `store.select` files the whole
  // rewrite as a rejected edit and the lease silently never moves.
  assert.deepEqual(
    { ...renewal.value, expiresAt: undefined, renewedAt: undefined },
    { ...created, expiresAt: undefined, renewedAt: undefined },
  )
  assert.notEqual(renewal.value.expiresAt, created.expiresAt)
  // And they move TOGETHER, from one clock read: the lease this version declares is `leaseMs`
  // however far the daemon's clock has wandered since the claim was created.
  assert.equal(renewal.value.renewedAt, new Date(now).toISOString())
  assert.equal(
    Date.parse(renewal.value.expiresAt) - Date.parse(renewal.value.renewedAt),
    TIMING.leaseMs,
  )
  assert.equal(renewal.value.expiresAt, new Date(now + TIMING.leaseMs).toISOString())
  s.close()
})

it('adopts its own claim after a restart instead of writing a second one', async () => {
  const base = buildScenario()
  const actor = actorFor(AGENT)
  const rkey = claimRkey(base.request.uri, base.request.cid)
  // The rkey is a pure function of the request, so a restart mid-lease collides with our OWN prior
  // claim. Adopting it — including its original `createdAt`, which is our place in the tie-break
  // and what every renewal must reproduce — is what keeps a restart from re-entering the race.
  actor.client.fail = new XrpcError(400, 'RecordAlreadyExists', 'record already exists')
  actor.client.existing = {
    cid: 'cid-claim-from-before-the-restart',
    value: {
      $type: COLLECTIONS.claim,
      request: ref(base.request),
      expiresAt: '2026-01-01T00:40:00.000Z',
      createdAt: '2026-01-01T00:30:00.000Z',
    },
  }
  const s = stack(base.records, { actors: [actor] })
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  const row = s.claims.get(base.request.uri, AGENT)
  assert.equal(row.state, 'confirming')
  assert.equal(row.claimCid, 'cid-claim-from-before-the-restart')
  assert.equal(row.createdAt, '2026-01-01T00:30:00.000Z')
  assert.equal(row.claimUri, `at://${AGENT}/${COLLECTIONS.claim}/${rkey}`)
  assert.equal(actor.client.writes.length, 0)
  s.close()
})

it('adopts its own claim when a PDS reports a duplicate create as an internal server error', async () => {
  const base = buildScenario()
  const actor = actorFor(AGENT)
  const rkey = claimRkey(base.request.uri, base.request.cid)
  // A hosted PDS has been observed returning this response for a duplicate deterministic-rkey
  // create. The read-back is authoritative: if our matching record is there, the create is
  // idempotently complete regardless of the status attached to its response.
  actor.client.fail = new XrpcError(500, 'InternalServerError', 'Internal Server Error')
  actor.client.existing = {
    cid: 'cid-claim-from-before-the-reset',
    value: {
      $type: COLLECTIONS.claim,
      request: ref(base.request),
      expiresAt: '2026-01-01T00:40:00.000Z',
      renewedAt: '2026-01-01T00:30:00.000Z',
      createdAt: '2026-01-01T00:30:00.000Z',
    },
  }
  const s = stack(base.records, { actors: [actor] })
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()

  const row = s.claims.get(base.request.uri, AGENT)
  assert.equal(row.state, 'confirming')
  assert.equal(row.claimCid, 'cid-claim-from-before-the-reset')
  assert.equal(row.claimUri, `at://${AGENT}/${COLLECTIONS.claim}/${rkey}`)
  assert.match(s.logs.join('\n'), /adopted existing claim/)
  assert.equal(actor.client.writes.length, 0)
  s.close()
})

it('repairs an existing claim record whose lease is out of contract, instead of wedging on it', async () => {
  const base = buildScenario()
  const actor = actorFor(AGENT)
  const rkey = claimRkey(base.request.uri, base.request.cid)
  // The same collision, but the record we collide with declares a month-long lease — written by a
  // daemon whose clock jumped before `renewedAt` existed. Nothing can delete it and every observer
  // refuses it, so adopting it as-is would park this row in `confirming` (which has no timeout)
  // against a claim that can never win, burning an outstanding slot forever.
  actor.client.fail = new XrpcError(400, 'RecordAlreadyExists', 'record already exists')
  actor.client.failCreateOnly = true
  actor.client.existing = {
    cid: 'cid-claim-poisoned',
    value: {
      $type: COLLECTIONS.claim,
      request: ref(base.request),
      expiresAt: '2026-02-01T00:00:00.000Z',
      createdAt: '2026-01-01T00:30:00.000Z',
    },
  }
  const s = stack(base.records, { actors: [actor], now: () => NOW })
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()

  const repair = actor.client.writes.at(-1)
  assert.equal(repair.kind, 'put')
  assert.equal(repair.uri, `at://${AGENT}/${COLLECTIONS.claim}/${rkey}`)
  assert.equal(repair.swapRecord, 'cid-claim-poisoned')
  // Shortened to a conforming lease, with the tie-break position left exactly where it was.
  assert.equal(repair.value.createdAt, '2026-01-01T00:30:00.000Z')
  assert.equal(repair.value.renewedAt, new Date(NOW).toISOString())
  assert.equal(repair.value.expiresAt, new Date(NOW + TIMING.leaseMs).toISOString())
  assert.match(s.logs.join('\n'), /out of contract/)

  const row = s.claims.get(base.request.uri, AGENT)
  assert.equal(row.state, 'confirming')
  assert.equal(row.renewedAt, new Date(NOW).toISOString())

  // And the row now progresses: the repaired claim wins the fold and reaches `held`, which is what
  // releases the outstanding budget for anything else in this space.
  const ours = {
    ...claimRecord(base.request, AGENT, {
      createdAt: '2026-01-01T00:30:00.000Z',
      expiresAt: new Date(NOW + TIMING.leaseMs).toISOString(),
    }),
    uri: `at://${AGENT}/${COLLECTIONS.claim}/${rkey}`,
    cid: 'cid-write-1',
    value: undefined,
  }
  const repaired = {
    ...ours,
    value: {
      $type: COLLECTIONS.claim,
      request: ref(base.request),
      expiresAt: new Date(NOW + TIMING.leaseMs).toISOString(),
      renewedAt: new Date(NOW).toISOString(),
      createdAt: '2026-01-01T00:30:00.000Z',
    },
  }
  s.manager.pump(s.index([repaired]), s.registry)
  s.manager.pump(s.index([repaired]), s.registry)
  await s.manager.drain()
  assert.deepEqual([...s.manager.heldClaims()], [base.request.uri])
  s.close()
})

it('repairs an out-of-contract record found under a HELD row, which is where an upgrade finds one', async () => {
  // The migration state the other repair paths miss. A pre-`renewedAt` daemon whose clock jumped
  // wrote a year-long lease and kept working; its ledger row is durable, so after the upgrade the
  // row is `held` against a record every observer (this one included) now refuses. Left alone the
  // row never moves: `expiresAt` is a year out so no renewal is ever due, and the fold names nobody
  // so the loss branch never fires either — and `heldRequests()` gates dispatch, so the daemon
  // would keep dispatching work under a claim nobody honours.
  const base = buildScenario()
  const actor = actorFor(AGENT)
  const s = stack(base.records, { actors: [actor], now: () => NOW })
  const rkey = claimRkey(base.request.uri, base.request.cid)
  const poisoned = {
    ...claimRecord(base.request, AGENT, {
      createdAt: '2026-01-01T00:30:00.000Z',
      expiresAt: '2027-01-01T00:00:00.000Z',
    }),
    cid: 'cid-claim-poisoned',
  }
  s.claims.markClaiming({
    requestUri: base.request.uri,
    requestCid: base.request.cid,
    actorDid: AGENT,
    spaceUri: SPACE_URI,
    rkey,
    createdAt: '2026-01-01T00:30:00.000Z',
  })
  s.claims.markConfirming(base.request.uri, AGENT, {
    uri: poisoned.uri,
    cid: poisoned.cid,
    createdAt: '2026-01-01T00:30:00.000Z',
    expiresAt: '2027-01-01T00:00:00.000Z',
  })
  s.claims.markHeld(base.request.uri, AGENT)

  s.manager.pump(s.index([poisoned]), s.registry)
  await s.manager.drain()
  const repair = actor.client.writes.at(-1)
  assert.equal(repair.kind, 'put')
  assert.equal(repair.swapRecord, 'cid-claim-poisoned')
  assert.equal(repair.value.expiresAt, new Date(NOW + TIMING.leaseMs).toISOString())
  assert.equal(repair.value.createdAt, '2026-01-01T00:30:00.000Z', 'the tie-break position is untouched')
  assert.match(s.logs.join('\n'), /out of contract/)
  // And it stops gating dispatch until the repair has been seen to win, like any other fresh lease.
  assert.equal(s.claims.get(base.request.uri, AGENT).state, 'confirming')
  assert.deepEqual([...s.manager.heldClaims()], [])
  s.close()
})

it('retires a claim its own clock, since corrected, dated into the future — and never re-adopts it', async () => {
  // The record every other branch is blind to. A daemon whose clock stepped a week forward stamped
  // `createdAt`, `renewedAt` and `expiresAt` from that clock, so the version is perfectly IN
  // CONTRACT: a legal lease inside a legal horizon. Correct the clock and nothing fires — the lease
  // is not lapsed (it is a week out), not due for renewal, not out of contract (so no repair, and
  // the store would refuse a shortening rewrite of it anyway), and the fold names no other winner.
  // The row would sit `held` forever, gating dispatch behind a claim its own writer cannot renew.
  const base = buildScenario()
  const actor = actorFor(AGENT)
  const s = stack(base.records, { actors: [actor], now: () => NOW })
  const rkey = claimRkey(base.request.uri, base.request.cid)
  const jumped = NOW + 7 * 24 * 60 * 60_000
  const fromTheFuture = {
    $type: COLLECTIONS.claim,
    request: ref(base.request),
    expiresAt: new Date(jumped + TIMING.leaseMs).toISOString(),
    renewedAt: new Date(jumped).toISOString(),
    createdAt: new Date(jumped).toISOString(),
  }
  assert.equal(claimWithinContract(fromTheFuture), true, 'nothing about the record itself is refusable')
  // It exists in the repo at the deterministic rkey, so the walk has something to collide with.
  actor.client.stored.set(rkey, { cid: 'cid-claim-from-the-future', value: fromTheFuture })
  const ours = {
    ...claimRecord(base.request, AGENT, {
      createdAt: fromTheFuture.createdAt,
      expiresAt: fromTheFuture.expiresAt,
    }),
    cid: 'cid-claim-from-the-future',
    value: fromTheFuture,
  }
  s.claims.markClaiming({
    requestUri: base.request.uri,
    requestCid: base.request.cid,
    actorDid: AGENT,
    spaceUri: SPACE_URI,
    rkey,
    createdAt: fromTheFuture.createdAt,
  })
  s.claims.markConfirming(base.request.uri, AGENT, {
    uri: ours.uri,
    cid: ours.cid,
    createdAt: fromTheFuture.createdAt,
    expiresAt: fromTheFuture.expiresAt,
    renewedAt: fromTheFuture.renewedAt,
  })
  s.claims.markHeld(base.request.uri, AGENT)
  // The fold still hands us the request — this daemon's own index is not what frees the row.
  assert.equal(s.index([ours]).goals[0].winningClaims[base.request.uri].did, AGENT)

  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.match(s.logs.join('\n'), /retiring our claim/)
  assert.match(s.logs.join('\n'), /into this daemon's own future/)
  assert.deepEqual([...s.manager.heldClaims()], [], 'it stops gating dispatch at once')

  // And it is re-claimed in the same pump, under the next generation: the walk asks the SAME
  // question before adopting the record it collides with, so it cannot put the row back where it
  // was. The future-dated record is left exactly as it stands — no rewrite of it would be adopted.
  const row = s.claims.get(base.request.uri, AGENT)
  assert.equal(row.state, 'confirming')
  assert.equal(row.generation, 1)
  assert.equal(row.rkey, claimRkey(base.request.uri, base.request.cid, 1))
  assert.equal(row.createdAt, new Date(NOW).toISOString(), 'the new record is dated by the corrected clock')
  assert.deepEqual(actor.client.stored.get(rkey).value, fromTheFuture)
  assert.deepEqual(
    actor.client.writes.map((write) => [write.kind, write.rkey ?? write.uri]),
    [['create', claimRkey(base.request.uri, base.request.cid, 1)]],
  )
  s.close()
})

it('retires a renewal a modest jump wrote, without waiting for the next renewal to come due', async () => {
  // The jump the RENEWAL branch catches, which is the ordinary one: `createdAt` is frozen, so only
  // `renewedAt` and `expiresAt` come from the fast clock. The version is in contract by both bounds
  // for any jump inside the horizon, so nothing refuses it and nothing can shorten it — and at
  // twenty minutes it is nowhere near `MAX_CLAIM_LEASE_MS`, which is what case 3 used to measure.
  //
  // Left alone the row stays `held`, gating dispatch, until this daemon's own renewal comes due —
  // `remaining <= leaseMs - renewIntervalMs`, so Δ + 200s away — while every observer stops
  // honouring the record one declared lease after it arrived. Bounded, but it is the same
  // divergence the change exists to remove, and at the defaults it lasts up to ~43 minutes.
  const base = buildScenario()
  const actor = actorFor(AGENT)
  const s = stack(base.records, { actors: [actor], now: () => NOW })
  const rkey = claimRkey(base.request.uri, base.request.cid)
  const jump = 20 * 60_000
  const createdAt = new Date(NOW - 5 * 60_000).toISOString() // claimed honestly, five minutes ago
  const renewed = {
    $type: COLLECTIONS.claim,
    request: ref(base.request),
    expiresAt: new Date(NOW + jump + TIMING.leaseMs).toISOString(),
    renewedAt: new Date(NOW + jump).toISOString(),
    createdAt,
  }
  assert.equal(claimWithinContract(renewed), true, 'in contract by both bounds, so nothing refuses it')
  const ours = { ...claimRecord(base.request, AGENT, { createdAt }), cid: 'cid-claim-renewed', value: renewed }
  actor.client.stored.set(rkey, { cid: ours.cid, value: renewed })
  s.claims.markClaiming({
    requestUri: base.request.uri,
    requestCid: base.request.cid,
    actorDid: AGENT,
    spaceUri: SPACE_URI,
    rkey,
    createdAt,
  })
  s.claims.markConfirming(base.request.uri, AGENT, {
    uri: ours.uri,
    cid: ours.cid,
    createdAt,
    expiresAt: renewed.expiresAt,
    renewedAt: renewed.renewedAt,
  })
  s.claims.markHeld(base.request.uri, AGENT)

  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.match(s.logs.join('\n'), /retiring our claim/)
  assert.match(s.logs.join('\n'), /further ahead than the longest lease/)
  assert.deepEqual([...s.manager.heldClaims()], [], 'it stops gating dispatch in the pump that finds it')
  // And it is covered again at once, in a record the corrected clock could have written.
  const row = s.claims.get(base.request.uri, AGENT)
  assert.equal(row.generation, 1)
  assert.equal(row.createdAt, new Date(NOW).toISOString())
  assert.deepEqual(actor.client.stored.get(rkey).value, renewed, 'the jumped record is left as it stands')
  s.close()
})

it('leaves a jump the next renewal still reaches, and lets that renewal repair it', async () => {
  // The other side of the same boundary, and the reason it is not a flat margin. A jump of Δ that
  // the renewal branch caught repairs ITSELF on the next renewal — due `Δ + renewIntervalMs` after
  // the correction, since the trigger is `remaining <= leaseMs - renewIntervalMs` — while observers
  // stop honouring the version one declared lease after it arrived. So for Δ up to
  // `leaseMs - renewIntervalMs` the record heals before anybody's honouring window closes, with
  // nothing for a peer to see. Retiring there would cost a generation and, because the retired record
  // keeps the earlier `createdAt` and so wins our OWN tie-break until it lapses, up to a full lease of
  // this daemon's own dispatch — paid every time a host steps its clock back by a couple of minutes.
  const base = buildScenario()
  const actor = actorFor(AGENT)
  let now = NOW
  const s = stack(base.records, { actors: [actor], now: () => now })
  const rkey = claimRkey(base.request.uri, base.request.cid)
  const jump = 2 * 60_000 // a resumed VM, a chronyd step
  assert.ok(jump <= TIMING.leaseMs - TIMING.renewIntervalMs, 'inside what a renewal still reaches')
  const createdAt = new Date(NOW - 5 * 60_000).toISOString() // claimed honestly, five minutes ago
  const renewed = {
    $type: COLLECTIONS.claim,
    request: ref(base.request),
    expiresAt: new Date(NOW + jump + TIMING.leaseMs).toISOString(),
    renewedAt: new Date(NOW + jump).toISOString(),
    createdAt,
  }
  assert.equal(claimWithinContract(renewed), true, 'in contract, like every record a jump leaves')
  const ours = { ...claimRecord(base.request, AGENT, { createdAt }), cid: 'cid-claim-renewed', value: renewed }
  actor.client.stored.set(rkey, { cid: ours.cid, value: renewed })
  s.claims.markClaiming({
    requestUri: base.request.uri,
    requestCid: base.request.cid,
    actorDid: AGENT,
    spaceUri: SPACE_URI,
    rkey,
    createdAt,
  })
  s.claims.markConfirming(base.request.uri, AGENT, {
    uri: ours.uri,
    cid: ours.cid,
    createdAt,
    expiresAt: renewed.expiresAt,
    renewedAt: renewed.renewedAt,
  })
  s.claims.markHeld(base.request.uri, AGENT)

  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.equal(/retiring our claim/.test(s.logs.join('\n')), false, 'nothing is retired')
  assert.deepEqual([...s.manager.heldClaims()], [base.request.uri], 'the row keeps holding the work')
  assert.equal(actor.client.writes.length, 0, 'and writes nothing: the renewal is not due yet')
  assert.equal(s.claims.get(base.request.uri, AGENT).generation, 0, 'no generation burned')

  // Δ + one renewal interval later the renewal falls due, and it is the repair: both lease fields
  // move from the honest clock while `createdAt` — the tie-break position — stays put.
  now = NOW + jump + TIMING.renewIntervalMs
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.equal(/retiring our claim/.test(s.logs.join('\n')), false)
  const put = actor.client.writes.at(-1)
  assert.equal(put.kind, 'put')
  assert.equal(put.value.renewedAt, new Date(now).toISOString())
  assert.equal(put.value.expiresAt, new Date(now + TIMING.leaseMs).toISOString())
  assert.equal(put.value.createdAt, createdAt)
  assert.equal(claimWithinContract(put.value), true)
  assert.deepEqual([...s.manager.heldClaims()], [base.request.uri])
  assert.equal(s.claims.get(base.request.uri, AGENT).generation, 0)
  s.close()
})

it('re-claims mid-turn when a running turn is what the retired claim was covering', async () => {
  // Retirement is terminal, so nothing renews the old record back into life; the request is free to
  // every peer from that moment. If the daemon could not re-claim until the turn ended, that window
  // would be a whole turn timeout wide (60 minutes by default) rather than a pump.
  const base = buildScenario()
  const actor = actorFor(AGENT)
  const createdAt = '2026-01-01T01:00:00.000Z'
  let now = NOW
  // Wired the way `cli.ts` wires it: `run.concurrency - dispatcher.inFlight`, at the default
  // concurrency of 1. A turn of ours running IS that one slot, so free capacity is zero for exactly
  // as long as the re-claim is needed — which is why a covering claim must not be charged against
  // it. With the dep left off (capacity infinite) this test asserts the exemption and not the path.
  const s = stack(base.records, {
    actors: [actor],
    now: () => now,
    capacity: () => 1 - s.turns.running().length,
  })
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  const ours = {
    ...claimRecord(base.request, AGENT, { createdAt }),
    uri: `at://${AGENT}/${COLLECTIONS.claim}/${claimRkey(base.request.uri, base.request.cid)}`,
  }
  s.manager.pump(s.index([ours]), s.registry)
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.deepEqual([...s.manager.heldClaims()], [base.request.uri])
  // The turn this claim covers is running when the horizon arrives — the routine case for a request
  // answered by a human in the last hour of its claim's day.
  s.turns.markRunning(base.request.uri, base.request.cid, {
    containerLabel: 'radial-turn-1',
    checkoutPath: '/tmp/radial-turn-1',
  })

  now = NOW + 24 * 60 * 60_000 + 60_000
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.match(s.logs.join('\n'), /retiring our claim/)
  const row = s.claims.get(base.request.uri, AGENT)
  assert.equal(row.state, 'confirming', 'covered again in the same pump, not at the end of the turn')
  assert.equal(row.generation, 1)
  assert.equal(s.turns.get(base.request.uri).state, 'running', 'and the turn is left running')
  s.close()
})

it('stops walking generations at the cap, and stops selecting the request too', async () => {
  // The backstop's own failure mode. Recording the exhausted walk only in the log would leave the
  // row at the last generation tried and `settled`, which `selectClaimable` reselects: every pump
  // would then re-walk to the cap through a create that collides and a read that says why — two PDS
  // round trips a tick, forever, after a single log line.
  const base = buildScenario()
  const actor = actorFor(AGENT)
  const now = NOW
  const s = stack(base.records, { actors: [actor], now: () => now })
  // Every generation up to the cap already exists in the repo, each one past its horizon: the shape
  // a request nobody can finish reaches after a fortnight of being claimed and retired.
  for (let generation = 0; generation < MAX_CLAIM_GENERATIONS; generation += 1) {
    actor.client.stored.set(claimRkey(base.request.uri, base.request.cid, generation), {
      cid: `cid-claim-generation-${generation}`,
      value: {
        $type: COLLECTIONS.claim,
        request: ref(base.request),
        // Created two days before `NOW`, so every one of them is past its horizon and none can be
        // adopted, renewed or repaired.
        expiresAt: '2025-12-30T00:10:00.000Z',
        renewedAt: '2025-12-30T00:00:00.000Z',
        createdAt: '2025-12-30T00:00:00.000Z',
      },
    })
  }
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  const row = s.claims.get(base.request.uri, AGENT)
  assert.equal(row.state, 'settled')
  assert.equal(row.generation, MAX_CLAIM_GENERATIONS, 'the cap is on the row, not only in the log')
  assert.equal(s.logs.filter((line) => /leaving it alone/.test(line)).length, 1)
  assert.equal(actor.client.writes.length, 0, 'and nothing was written at any generation')

  // From here the request is skipped outright: no selection, no create, no read, no second log line.
  s.manager.pump(s.index(), s.registry)
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  assert.equal(actor.client.writes.length, 0)
  assert.equal(s.logs.filter((line) => /leaving it alone/.test(line)).length, 1)
  assert.equal(s.claims.get(base.request.uri, AGENT).generation, MAX_CLAIM_GENERATIONS)

  // And the escape hatch the log line promises actually moves it. `reset` cannot delete the sixteen
  // records — nothing deletes records, and every one of them is why the walk stopped — so it keeps
  // the generation and raises the row's ceiling instead. The next pump writes at the first
  // generation nothing was ever written at: one create, no collisions, no second capping.
  s.claims.reset(base.request.uri, { generations: MAX_CLAIM_GENERATIONS })
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  const revived = s.claims.get(base.request.uri, AGENT)
  assert.equal(revived.state, 'confirming')
  assert.equal(revived.generation, MAX_CLAIM_GENERATIONS)
  assert.equal(revived.rkey, claimRkey(base.request.uri, base.request.cid, MAX_CLAIM_GENERATIONS))
  assert.deepEqual(
    actor.client.writes.map((write) => [write.kind, write.rkey]),
    [['create', claimRkey(base.request.uri, base.request.cid, MAX_CLAIM_GENERATIONS)]],
    'one write, and no re-walk of the spent generations to reach it',
  )
  assert.equal(s.logs.filter((line) => /leaving it alone/.test(line)).length, 1)
  s.close()
})

it('reports the cap when a retirement is what spends the last generation', async () => {
  // Retiring now moves the row's generation, which is also how the LAST one is spent — and a row at
  // its ceiling is skipped by `selectClaimable`, so the walk that used to print "leaving it alone"
  // is never entered. The line has to come from the retirement instead, or a request that has run out
  // of generations goes quiet with only a retirement in the log to explain it.
  const base = buildScenario()
  const actor = actorFor(AGENT)
  const createdAt = '2026-01-01T01:00:00.000Z'
  let now = NOW
  const s = stack(base.records, { actors: [actor], now: () => now })
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  // One generation left, the way `radiald claim reset` grants them: the row keeps its own, and its
  // ceiling moves up from there.
  s.claims.reset(base.request.uri, { generations: 1 })
  assert.equal(s.claims.get(base.request.uri, AGENT).generationCap, 1)
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  const adopted = s.claims.get(base.request.uri, AGENT)
  assert.equal(adopted.generation, 0, 'the record already at generation 0 is adopted, not duplicated')
  assert.equal(adopted.state, 'confirming')

  const ours = { ...claimRecord(base.request, AGENT, { createdAt }), uri: `at://${AGENT}/${COLLECTIONS.claim}/${claimRkey(base.request.uri, base.request.cid)}` }
  now = NOW + 24 * 60 * 60_000 + 60_000
  const writes = actor.client.writes.length
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.match(s.logs.join('\n'), /retiring our claim/)
  assert.equal(s.logs.filter((line) => /leaving it alone/.test(line)).length, 1, 'and says it has run out')
  const row = s.claims.get(base.request.uri, AGENT)
  assert.equal(row.state, 'settled')
  assert.equal(row.generation, 1, 'at its ceiling, so nothing selects it again')
  assert.equal(actor.client.writes.length, writes, 'and no record is written at a generation it has not got')

  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.equal(actor.client.writes.length, writes)
  assert.equal(s.logs.filter((line) => /leaving it alone/.test(line)).length, 1, 'once, not once a pump')
  s.close()
})

it('retires a claim at the protocol horizon and re-claims under the next generation', async () => {
  // The cost of bounding every version by the horizon, and how it is paid. `createdAt` is frozen
  // within a claim, so a day after it was written no version of THAT claim can be in contract —
  // renewal cannot fix it, and neither can the repair path. A request that still needs work (one
  // waiting on a human answer routinely does) gets a new record beside the retired one.
  const base = buildScenario()
  const actor = actorFor(AGENT)
  const createdAt = '2026-01-01T01:00:00.000Z'
  let now = NOW
  const s = stack(base.records, { actors: [actor], now: () => now })
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  assert.equal(s.claims.get(base.request.uri, AGENT).generation, 0)
  const ours = { ...claimRecord(base.request, AGENT, { createdAt }), uri: `at://${AGENT}/${COLLECTIONS.claim}/${claimRkey(base.request.uri, base.request.cid)}` }

  // A day and a minute later the lease has run out and cannot be renewed: the horizon has passed.
  now = NOW + 24 * 60 * 60_000 + 60_000
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.match(s.logs.join('\n'), /retiring our claim/)

  // Retired, and re-claimed in the same pump — at generation 1, a new record beside the retired one
  // with a fresh `createdAt` and a fresh place in the tie-break.
  const row = s.claims.get(base.request.uri, AGENT)
  assert.equal(row.state, 'confirming')
  assert.equal(row.generation, 1)
  assert.equal(row.rkey, claimRkey(base.request.uri, base.request.cid, 1))
  assert.notEqual(row.rkey, claimRkey(base.request.uri, base.request.cid))
  assert.match(s.logs.join('\n'), /generation 1/)
  const created = actor.client.writes.at(-1)
  assert.equal(created.kind, 'create')
  assert.equal(created.rkey, claimRkey(base.request.uri, base.request.cid, 1))
  assert.equal(created.value.createdAt, new Date(now).toISOString())
  assert.equal(created.value.renewedAt, new Date(now).toISOString())
  assert.equal(
    actor.client.writes.filter((write) => write.kind === 'put').length,
    0,
    'and nothing was written to the retired record, which no rewrite could put back in contract',
  )
  // The retirement SPENT generation 0, so the re-claim goes straight to 1. Leaving the row where it
  // was sent the walk into a create that collided with the record it had just judged past using and
  // a read that re-derived the same verdict — two PDS round trips and a second identical retirement,
  // on every retirement.
  assert.deepEqual(actor.client.reads, [], 'no read-back of the record we just retired')
  assert.deepEqual(
    actor.client.writes.map((write) => write.rkey),
    [claimRkey(base.request.uri, base.request.cid), claimRkey(base.request.uri, base.request.cid, 1)],
    'one create for the original claim, one for its replacement, and nothing else',
  )
  s.close()
})

it('clamps the last renewal to the horizon instead of writing a lease the fold would refuse', async () => {
  const base = buildScenario()
  const actor = actorFor(AGENT)
  const createdAt = '2026-01-01T01:00:00.000Z'
  const horizon = Date.parse(createdAt) + 24 * 60 * 60_000
  let now = NOW
  const s = stack(base.records, { actors: [actor], now: () => now })
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  const ours = { ...claimRecord(base.request, AGENT, { createdAt }), uri: `at://${AGENT}/${COLLECTIONS.claim}/${claimRkey(base.request.uri, base.request.cid)}` }
  s.manager.pump(s.index([ours]), s.registry)
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.deepEqual([...s.manager.heldClaims()], [base.request.uri])

  // Five minutes short of the horizon, with a ten-minute lease: the renewal stops AT the horizon,
  // which is where every observer would have stopped honouring it anyway.
  now = horizon - 5 * 60_000
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  const renewal = actor.client.writes.at(-1)
  assert.equal(renewal.kind, 'put')
  assert.equal(renewal.value.expiresAt, new Date(horizon).toISOString())
  assert.equal(renewal.value.renewedAt, new Date(now).toISOString())

  // And it is not rewritten on every tick of the interval that is left: there is nowhere to move it.
  const writes = actor.client.writes.length
  now = horizon - 60_000
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.equal(actor.client.writes.length, writes)
  s.close()
})

it('leaves the row claiming when the write fails transiently, so the next pump retries', async () => {
  const base = buildScenario()
  const actor = actorFor(AGENT)
  actor.client.fail = new Error('PDS unreachable')
  const s = stack(base.records, { actors: [actor] })
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  assert.equal(s.claims.get(base.request.uri, AGENT).state, 'claiming')
  assert.match(s.logs.join('\n'), /failed to claim/)

  actor.client.fail = undefined
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  assert.equal(s.claims.get(base.request.uri, AGENT).state, 'confirming')
  s.close()
})

it('revives its own lapsed lease by renewing, never by claiming over its own record', async () => {
  const base = buildScenario()
  let now = NOW
  const s = stack(base.records, { now: () => now })
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  const rkey = claimRkey(base.request.uri, base.request.cid)
  const ours = {
    ...claimRecord(base.request, AGENT, { createdAt: '2026-01-01T01:00:00.000Z' }),
    uri: `at://${AGENT}/${COLLECTIONS.claim}/${rkey}`,
  }
  s.manager.pump(s.index([ours]), s.registry)
  s.manager.pump(s.index([ours]), s.registry)
  await s.manager.drain()
  assert.deepEqual([...s.manager.heldClaims()], [base.request.uri])

  // The daemon was down long enough for its own lease to run out. Nobody else took the work, and
  // the claim record is still there at the deterministic rkey — a create would collide with it and
  // adopt its EXPIRED lease, which would then never confirm and never renew.
  now = NOW + TIMING.leaseMs + 60_000
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  const revival = s.actors[0].client.writes.at(-1)
  assert.equal(revival.kind, 'put')
  assert.equal(revival.value.expiresAt, new Date(now + TIMING.leaseMs).toISOString())
  assert.equal(revival.value.createdAt, '2026-01-01T01:00:00.000Z', 'the tie-break position is unchanged')
  // Back to confirming: a revived lease is not something to work on until the fold agrees it is live.
  assert.equal(s.claims.get(base.request.uri, AGENT).state, 'confirming')
  assert.deepEqual([...s.manager.heldClaims()], [])
  s.close()
})

it('settles a retracted claim and abandons its running turn from the same snapshot', async () => {
  const base = buildScenario()
  const abandoned = []
  const s = stack(base.records, { abandon: (uri, reason) => abandoned.push({ uri, reason }) })
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  assert.equal(s.claims.get(base.request.uri, AGENT).state, 'confirming')

  // The first snapshot that reveals the tombstone no longer lists the request as open. This is the
  // batching shape produced when an auto-review canonical and the loser's retraction arrive in one
  // poll: the open-only duplicate reconciler has never seen the loser, so claim settlement must
  // still stop its already-running container.
  const retraction = mk(HUMAN, COLLECTIONS.retractRequest, 'retract-plan', 'cid-retract-plan', {
    $type: COLLECTIONS.retractRequest,
    request: ref(base.request),
    createdAt: '2026-01-01T01:05:00Z',
  })
  s.manager.pump(s.index([retraction]), s.registry)
  await s.manager.drain()
  assert.equal(s.claims.get(base.request.uri, AGENT).state, 'settled')
  assert.deepEqual(abandoned, [{ uri: base.request.uri, reason: 'request no longer open' }])

  // Terminal rows are skipped on later snapshots, so the stop signal is emitted exactly once.
  s.manager.pump(s.index([retraction]), s.registry)
  await s.manager.drain()
  assert.equal(abandoned.length, 1)
  s.close()
})

it('caps outstanding claims at maxOutstanding and at the dispatcher capacity', async () => {
  const base = buildScenario()
  const second = mk(HUMAN, COLLECTIONS.artifactRequest, 'request-2', 'cid-request-2', {
    $type: COLLECTIONS.artifactRequest,
    goal: ref(base.goal),
    type: 'plan',
    basedOn: [],
    createdAt: '2026-01-01T00:03:30Z',
  })
  const claims = new ClaimLedger()
  const turns = new TurnLedger()
  const actor = actorFor(AGENT)
  const manager = new ClaimManager({
    ledger: claims,
    turns,
    timing: { ...TIMING, maxOutstanding: 1 },
    now: () => NOW,
  })
  const index = indexOf([...base.records, second])
  manager.pump(index, registryFor([actor]))
  await manager.drain()
  assert.equal(claims.all().length, 1)

  // Capacity zero: even with budget left, claiming work we cannot start starves the other operator.
  const capped = new ClaimLedger()
  const cappedManager = new ClaimManager({
    ledger: capped,
    turns,
    timing: TIMING,
    now: () => NOW,
    capacity: () => 0,
  })
  cappedManager.pump(index, registryFor([actorFor(AGENT)]))
  await cappedManager.drain()
  assert.deepEqual(capped.all(), [])

  claims.close()
  capped.close()
  turns.close()
})

// --- two spaces, one manager ------------------------------------------------

it('reconciles a row only against the space it was claimed in — two spaces, one manager', async () => {
  // The daemon-wide ledger plus one `pump` per space per tick (runtime.ts) is the shape that broke:
  // reconciling every row against every index settled space A's live claim the moment space B was
  // pumped, and the row then churned between `claiming` and `confirming` forever without ever
  // reaching `held`, so neither request was ever dispatchable.
  const first = buildScenario()
  const second = buildScenario({ suffix: '-two' })
  const secondSpaceUri = `at://${ROOT}/${COLLECTIONS.space}/space-two`

  const claims = new ClaimLedger()
  const turns = new TurnLedger()
  const logs = []
  const actor = actorFor(AGENT)
  const registry = registryFor([actor])
  const manager = new ClaimManager({ ledger: claims, turns, timing: TIMING, now: () => NOW, log: (m) => logs.push(m) })

  const ourClaim = (scenario) => ({
    ...claimRecord(scenario.request, AGENT, { createdAt: '2026-01-01T01:00:00.000Z' }),
    uri: `at://${AGENT}/${COLLECTIONS.claim}/${claimRkey(scenario.request.uri, scenario.request.cid)}`,
  })

  // Three ticks, each pumping both spaces in `runDaemon`'s order: claim, see it win, confirm.
  for (let tick = 0; tick < 3; tick += 1) {
    const extra = tick === 0 ? [] : [ourClaim(first)]
    const extraTwo = tick === 0 ? [] : [ourClaim(second)]
    manager.pump(indexOf([...first.records, ...extra]), registry)
    manager.pump(indexOf([...second.records, ...extraTwo], secondSpaceUri), registry)
    await manager.drain()
  }

  const rowOne = claims.get(first.request.uri, AGENT)
  const rowTwo = claims.get(second.request.uri, AGENT)
  assert.equal(rowOne.spaceUri, SPACE_URI)
  assert.equal(rowTwo.spaceUri, secondSpaceUri)
  assert.equal(rowOne.state, 'held')
  assert.equal(rowTwo.state, 'held')
  assert.deepEqual([...manager.heldClaims()].sort(), [first.request.uri, second.request.uri].sort())
  assert.equal(
    logs.some((message) => message.includes('no longer an open request')),
    false,
    'a live claim in one space must not be settled by the pump of another',
  )
  // Exactly one create per request: no collision-and-adopt churn at the deterministic rkey.
  assert.deepEqual(
    actor.client.writes.filter((write) => write.kind === 'create').map((write) => write.value.request.uri).sort(),
    [first.request.uri, second.request.uri].sort(),
  )

  claims.close()
  turns.close()
})

it('adopts a row written before claims were space-scoped into the space that knows its request', async () => {
  const base = buildScenario()
  const other = buildScenario({ suffix: '-two' })
  const otherSpaceUri = `at://${ROOT}/${COLLECTIONS.space}/space-two`
  const s = stack(base.records)
  // A row from an older build of the daemon: everything but the space it belongs to.
  s.claims.markClaiming({
    requestUri: base.request.uri,
    requestCid: base.request.cid,
    actorDid: AGENT,
    spaceUri: '',
    rkey: claimRkey(base.request.uri, base.request.cid),
    createdAt: '2026-01-01T00:30:00.000Z',
  })

  // A space that has never heard of the request leaves it alone.
  s.manager.pump(indexOf(other.records, otherSpaceUri), s.registry)
  await s.manager.drain()
  assert.equal(s.claims.get(base.request.uri, AGENT).spaceUri, '')
  assert.equal(s.claims.get(base.request.uri, AGENT).state, 'claiming')

  // Its own space adopts it and reconciles it like any other row — here, retrying the create.
  s.manager.pump(s.index(), s.registry)
  await s.manager.drain()
  const row = s.claims.get(base.request.uri, AGENT)
  assert.equal(row.spaceUri, SPACE_URI)
  assert.equal(row.state, 'confirming')
  s.close()
})