import { createHash } from 'node:crypto'
import { existsSync, readFileSync } from 'node:fs'
import { mkdir, rename, writeFile } from 'node:fs/promises'
import { dirname, join } from 'node:path'
import { XrpcError } from '@radial/atproto'
import {
  activeGoals,
  activeProjects,
  agentTypesFor,
  COLLECTIONS,
  compareCodePoints,
  duplicateReviewRequests,
  type ArtifactRecord,
  type ArtifactRequestRecord,
  type GoalView,
  type IndexedRecord,
  type MaterializedIndex,
  type ProjectView,
  type RadialRecord,
  type StrongRef,
} from '@radial/core'
import { actorAccepts, type ActorRegistry, type LoadedActor } from './actors.js'
import { base32Encode } from './turn-socket.js'

/** The one auto-review request type name the daemon emits (design §7). Matches a `review`-producing
 * agent profile and the materializer's review-request path (materializer.ts:487-502). */
const REVIEW_TYPE = 'review'
export const AUTO_REVIEW_STAGGER_TICKS = 2

/** Stable strongref key (uri#cid): the subject an auto-review request pins is an EXACT artifact
 * version, so every dedup and the observed-subjects ledger key on uri#cid. */
const refKey = (reference: { uri: string; cid: string }): string => `${reference.uri}#${reference.cid}`

const sameRef = (left: StrongRef | undefined, right: StrongRef | undefined): boolean =>
  Boolean(left && right && left.uri === right.uri && left.cid === right.cid)

/**
 * Deterministic rkey for the auto-review request pinning an exact artifact version: `autoreview-`
 * plus 24 base32 chars of sha256(subjectUri#subjectCid[#retractedCount]). Per-actor repos make this
 * stable per daemon-actor, so a restart that re-derives the same rkey collides on the PDS
 * (RecordAlreadyExists) instead of writing a second request — closing the restart-window double-write
 * the in-memory `alreadyWritten` set alone cannot (it is lost on restart). Mirrors `mergeRkey`.
 *
 * `retractedCount` salts the key with how many prior auto-review requests for this exact subject have
 * been retracted: 0 reproduces the historical rkey; after a retraction the rewrite gets a FRESH rkey
 * (design fix) so it does not collide with — and adopt — the tombstoned request, which the
 * materializer keeps out of `openRequests` forever.
 */
export function autoReviewRkey(subjectUri: string, subjectCid: string, retractedCount = 0): string {
  const salt = retractedCount > 0 ? `#${retractedCount}` : ''
  const digest = createHash('sha256').update(`${subjectUri}#${subjectCid}${salt}`).digest()
  return `autoreview-${base32Encode(digest).slice(0, 24)}`
}

export function autoReviewRetractionRkey(requestUri: string, requestCid: string): string {
  const digest = createHash('sha256').update(`${requestUri}#${requestCid}`).digest()
  return `retract-${base32Encode(digest).slice(0, 24)}`
}

/** Space-wide fold-derived reviewer order. Unlike loaded actors, every observer computes this list
 * identically, which gives separate operators distinct fallback ranks. */
export function eligibleReviewerDids(index: MaterializedIndex): string[] {
  const profiled = new Set(
    (index.agents ?? [])
      .filter((agent) => agentTypesFor(agent.value, index.space.uri).includes(REVIEW_TYPE))
      .map((agent) => agent.did),
  )
  return index.members
    .filter((member) => member.active && member.kind === 'agent' && profiled.has(member.did))
    .map((member) => member.did)
    .sort(compareCodePoints)
}

/**
 * The project's per-artifact-type auto-review defaults, read through THIS single helper on purpose.
 * `ProjectView.autoReview` already merges the on-record default with the admin `setAutoReview`
 * overlay (latest-wins per type), so it is the single source of truth.
 */
export function projectAutoReviewConfig(projectView: ProjectView): Record<string, boolean> {
  if (projectView.autoReview) return projectView.autoReview
  const config = projectView.target.value.autoReview
  return config && typeof config === 'object' ? (config as Record<string, boolean>) : {}
}

/**
 * The loaded actors eligible to author a review in this space: each holds an active AGENT-kind
 * membership grant (mirroring dispatch.ts's agent-kind check) and produces `review` artifacts HERE —
 * a profile scoped out of this space is not one of them, however it is configured elsewhere.
 * Sorted by DID, so `[0]` is the deterministic writer and the list is a stable probe order.
 */
export function eligibleReviewActors(index: MaterializedIndex, actors: ActorRegistry): LoadedActor[] {
  const activeAgentDids = new Set(
    index.members.filter((member) => member.active && member.kind === 'agent').map((member) => member.did),
  )
  return actors.all
    .filter(
      (actor) => activeAgentDids.has(actor.did) && actorAccepts(actor, REVIEW_TYPE, index.space.uri),
    )
    .sort((a, b) => compareCodePoints(a.did, b.did))
}

/**
 * ALL loaded actors that produce `review` artifacts, regardless of their CURRENT index eligibility —
 * the set of repos probed for an already-written request before creating a new one. It is
 * deliberately broader than `eligibleReviewActors`: an actor that wrote the request and then lost its
 * membership grant (but is still loaded) must still be probed, or its record is missed and a
 * duplicate is written. Sorted by DID for a stable probe order.
 *
 * Space scoping is deliberately NOT applied here for the same reason membership is not: this is a
 * probe order, not eligibility. A profile scoped out of a space today may well have written the
 * request yesterday, and missing its record writes a duplicate.
 */
export function loadedReviewActors(actors: ActorRegistry): LoadedActor[] {
  return actors.all
    .filter((actor) => actor.artifactTypes.includes(REVIEW_TYPE))
    .sort((a, b) => compareCodePoints(a.did, b.did))
}

/**
 * The request writer is deterministic: among the ELIGIBLE actors (each holds an active AGENT-kind
 * membership grant in the space and produces `review` artifacts) pick the lowest DID. This chooses
 * the repository that authors the open request, not the agent that will execute it.
 */
export function selectReviewWriter(eligible: LoadedActor[]): LoadedActor | undefined {
  return [...eligible].sort((a, b) => compareCodePoints(a.did, b.did))[0]
}

export interface AutoReviewCandidate {
  /** The exact artifact version to request a review of (its uri#cid is the review subject). */
  artifact: IndexedRecord<ArtifactRecord>
  /** The actor whose repo the open review request is written into (see `selectReviewWriter`). */
  writer: LoadedActor
  /** Count of RETRACTED trusted `review` requests already pinning this exact subject — salts the
   * deterministic rkey so a rewrite after a retraction never collides with the tombstoned request. */
  retractedCount: number
  /** Writer position in the space-wide, fold-derived reviewer order. */
  writerRank: number
}

/**
 * Pure selection gate: which artifact versions should have an auto-review request written right now.
 * Scans EVERY version in every goal's and project's artifacts (each prev-chained version is a
 * distinct subject), skipping ended goals and archived projects. An artifact is a candidate iff:
 *   - its effective auto-review config is on: `request.autoReview ?? projectDefault[type] ?? false`,
 *     where `request` is the artifact's fulfilling request (its `request` backref; always resolvable,
 *     the materializer drops artifacts otherwise) and the project default is read through
 *     `projectAutoReviewConfig` (goal artifacts read their goal's project);
 *   - no LIVE-or-FULFILLED trusted `review` request already pins this exact version — a RETRACTED
 *     request does NOT count (it lingers in `view.requests` but is listed in `view.retracted`);
 *   - no trusted review VERDICT already pins this exact version (covers the UI's direct post-a-verdict
 *     path, which has no request);
 *   - it is not in the in-process `alreadyWritten` set; and
 *   - it is not in the observed-subjects `ledger` (i.e. it appeared since the last tick).
 *
 * RETRACTION SEMANTICS (contract). The trigger fires AT MOST ONCE per subject version, decided at
 * landing time. The `ledger`/`alreadyWritten` short-circuit above is checked BEFORE the retracted
 * request is even inspected, so a human retracting the auto-written request after the subject has
 * been observed does NOT re-fire the trigger — a retraction is a deliberate stop signal, and the
 * remedy for wanting a review after all is the UI's Review button. The retracted-request handling and
 * salted rkey below cover only the genuinely-once case: a request created AND retracted before the
 * subject's first observation (so its subject was never in the ledger), where the rewrite must land
 * at a fresh rkey rather than adopt the tombstone.
 *
 * When an artifact would be a candidate but NO eligible review writer exists, `onSkip` is invoked
 * (deduped to once per space by the caller) AND `onDefer` records the subject key so the caller can
 * keep it UNOBSERVED (not ledger-marked) — otherwise it would be lost the moment a reviewer is
 * configured later. Ordered deterministically by subject uri.
 */
export function selectAutoReviewCandidates(
  index: MaterializedIndex,
  actors: ActorRegistry,
  ledger: Set<string>,
  alreadyWritten: Set<string>,
  onSkip?: (reason: string) => void,
  onDefer?: (subjectKey: string) => void,
): AutoReviewCandidate[] {
  const writer = selectReviewWriter(eligibleReviewActors(index, actors))
  const reviewerOrder = eligibleReviewerDids(index)

  // An artifact's `request` backref resolves against the fulfilling requests of ANY target (a
  // project-scoped artifact's request lives under the project; a goal artifact's under the goal).
  const requestByRef = new Map(
    [...index.goals.flatMap((view) => view.requests), ...index.projects.flatMap((view) => view.requests)].map(
      (request) => [refKey(request), request],
    ),
  )
  const projectViewByUri = new Map(index.projects.map((view) => [view.target.uri, view]))

  const candidates: AutoReviewCandidate[] = []
  const scan = (view: GoalView | ProjectView, projectConfig: Record<string, boolean>): void => {
    const retractedUris = new Set(view.retracted.map((request) => request.uri))
    for (const artifact of view.artifacts) {
      const key = refKey(artifact)
      if (ledger.has(key) || alreadyWritten.has(key)) continue
      const request = requestByRef.get(refKey(artifact.value.request))
      if (!request) continue // defensive: the materializer guarantees this resolves.
      const effective = request.value.autoReview ?? projectConfig[artifact.value.type] ?? false
      if (!effective) continue
      // (a) A live-or-fulfilled review request already pins this exact version. Retracted requests
      // linger in `requests` but are excluded from `openRequests`; exclude them here identically.
      const hasReviewRequest = view.requests.some(
        (candidate) =>
          candidate.value.type === REVIEW_TYPE &&
          candidate.value.subject !== undefined &&
          refKey(candidate.value.subject) === key &&
          !retractedUris.has(candidate.uri),
      )
      if (hasReviewRequest) continue
      // (b) A trusted verdict already pins this exact version (direct post-a-verdict path, no request).
      if (view.reviews.some((review) => refKey(review.value.subject) === key)) continue
      if (!writer) {
        onSkip?.('no loaded actor holds an active agent membership grant and produces review artifacts')
        // Defer, don't drop: keep this subject unobserved so it fires once a reviewer is available.
        onDefer?.(key)
        continue
      }
      // Salt the rkey with how many auto-review requests for this exact subject were retracted, so a
      // rewrite after a retraction gets a fresh rkey (never collides with the tombstoned request).
      const retractedCount = view.retracted.filter(
        (request) =>
          request.value.type === REVIEW_TYPE &&
          request.value.subject !== undefined &&
          refKey(request.value.subject) === key,
      ).length
      const rank = reviewerOrder.indexOf(writer.did)
      candidates.push({
        artifact,
        writer,
        retractedCount,
        // A locally loaded writer can precede its still-uningested agent profile. Treat that
        // uncertainty as the end of the known order instead of letting it bypass staggering.
        writerRank: rank === -1 ? reviewerOrder.length : rank,
      })
    }
  }

  // Shelved goals are set aside: agents stop reviewing their artifacts. `activeGoals` carries the
  // whole rule — ended, or hanging off a project its author archived.
  for (const goalView of activeGoals(index)) {
    const projectView = projectViewByUri.get(goalView.target.value.project.uri)
    scan(goalView, projectView ? projectAutoReviewConfig(projectView) : {})
  }
  for (const projectView of activeProjects(index)) {
    scan(projectView, projectAutoReviewConfig(projectView))
  }

  return candidates.sort((a, b) => compareCodePoints(a.artifact.uri, b.artifact.uri))
}

/**
 * The artifact versions the ledger is allowed to mark as observed: every artifact under a live
 * project plus every artifact under a live goal. Artifacts under a shelved target are deliberately
 * EXCLUDED — they stay unobserved so that if the goal reopens, or the project is un-archived (or the
 * artifact and its ending arrived in the same first sync), they become candidates then, instead of
 * being permanently wedged out by an early mark.
 */
const observableArtifacts = (index: MaterializedIndex): Array<IndexedRecord<ArtifactRecord>> => [
  ...activeProjects(index).flatMap((view) => view.artifacts),
  ...activeGoals(index).flatMap((view) => view.artifacts),
]

/** On-disk shape of a per-space observed-subjects ledger. Presence of the file means the space has
 * been seeded (pre-existing artifacts recorded, never auto-reviewed). */
interface LedgerFile {
  space: string
  subjects: string[]
}

/** In-memory per-space observed-subjects ledger: the set of subject refKeys (uri#cid) ever seen,
 * plus whether a persisted snapshot already existed (seeded). */
class ObservedLedger {
  readonly subjects = new Set<string>()
  seeded = false
  constructor(readonly path: string) {}
}

export interface AutoReviewTriggerDeps {
  /** Daemon state directory; per-space ledgers live under `<stateDir>/auto-review/`. */
  stateDir: string
  /** Injectable ISO-8601 clock for the written request's `createdAt`. */
  now?: () => string
  log?: (message: string) => void
  /** Stops a locally running turn when its auto-review request loses deduplication. */
  onAbandon?: (requestUri: string, reason: string) => void
}

/**
 * Auto-review trigger (design §7), modeled on `MergePoller`: `pump` selects the artifact versions
 * whose effective auto-review config is on and that landed since the last tick, and fires off exactly
 * one open `type: "review"` `artifactRequest` per subject (authored by one of the daemon's eligible
 * agent identities, with the artifact as `subject`). Fire-and-forget: `pump` never awaits a write, so it
 * never blocks the ingestion loop.
 *
 * "Landed" is defined by a persisted per-space observed-subjects ledger, NOT a timestamp. Each tick,
 * AFTER selection, every OBSERVABLE artifact version is recorded and persisted EXCEPT the ones the
 * ledger must not swallow: current candidates (recorded only once their write lands), subjects with a
 * write still in flight from an overlapping pump, and subjects deferred for want of an eligible
 * reviewer. A failed write therefore leaves its subject unmarked and the next pump reselects and
 * retries it (the deterministic rkey + adoption make that retry safe against duplicates).
 * Artifacts under an ended goal are never marked, so they become candidates if the goal reopens. The
 * first-ever snapshot of a space seeds the ledger without triggering anything (pre-existing artifacts
 * are never auto-reviewed); thereafter an artifact is a candidate only if it is not yet in the ledger
 * — which also covers artifacts that landed during daemon downtime (still reviewed on restart) and
 * prevents a backfill storm when a project flips auto-review on. The trigger fires AT MOST ONCE per
 * subject version, decided at landing time; a post-landing retraction is a stop signal, not a
 * re-trigger (see `selectAutoReviewCandidates`).
 *
 * Double-write defense is layered: the in-memory `#alreadyWritten` set closes the within-process
 * window before the request is ingested; before creating, the DETERMINISTIC rkey is probed via
 * `getOwnRecord` across every LOADED review-capable actor's repo (`loadedReviewActors` — broader than
 * current index eligibility, so an actor that wrote the request and then lost its grant is still
 * probed); and a collision on create is likewise adopted. A record is adopted only when it matches
 * exactly — same subject, type `review`, goal/project target, `basedOn` of exactly `[subject]`, and
 * either no `assignee` (the canonical open shape) or a legacy `assignee` equal to the repo it was
 * found in; a mismatched record in one repo does not stop the probe (a valid match in another repo
 * is still adopted). ACCEPTED residual corner: if the prior
 * writer is removed from the daemon config ENTIRELY between crash and restart, it is no longer loaded
 * and cannot be probed, so at most one duplicate review request may be written — the same bounded,
 * attributable corner `MergePoller` accepts, and harmless (a redundant review request).
 */
export class AutoReviewTrigger {
  readonly #deps: AutoReviewTriggerDeps
  readonly #ledgers = new Map<string, ObservedLedger>()
  readonly #alreadyWritten = new Set<string>()
  readonly #inFlight = new Map<string, Promise<void>>()
  readonly #retractions = new Map<string, Promise<void>>()
  readonly #alreadyRetracted = new Set<string>()
  /** Duplicate requests whose locally running turn has already been abandoned. */
  readonly #alreadyAbandoned = new Set<string>()
  readonly #staggerTicks = new Map<string, number>()
  /** Per-space persist tail: chained so overlapping ticks never race the tmp+rename. */
  readonly #persists = new Map<string, Promise<void>>()
  /** Skip/error reasons already logged, so a stable condition is reported once. */
  readonly #loggedSkips = new Set<string>()

  constructor(deps: AutoReviewTriggerDeps) {
    this.#deps = deps
  }

  get inFlight(): number {
    return this.#inFlight.size
  }

  #now(): string {
    return this.#deps.now?.() ?? new Date().toISOString()
  }

  #logOnce(message: string): void {
    if (this.#loggedSkips.has(message)) return
    this.#loggedSkips.add(message)
    this.#deps.log?.(message)
  }

  #ledgerPathFor(spaceUri: string): string {
    const hex = createHash('sha256').update(spaceUri).digest('hex').slice(0, 16)
    return join(this.#deps.stateDir, 'auto-review', `${hex}.json`)
  }

  /** Loads (once per process) the persisted ledger for a space; a present file means seeded. */
  #loadLedger(spaceUri: string): ObservedLedger {
    const existing = this.#ledgers.get(spaceUri)
    if (existing) return existing
    const ledger = new ObservedLedger(this.#ledgerPathFor(spaceUri))
    if (existsSync(ledger.path)) {
      try {
        const parsed = JSON.parse(readFileSync(ledger.path, 'utf8')) as LedgerFile
        if (Array.isArray(parsed.subjects)) for (const subject of parsed.subjects) ledger.subjects.add(subject)
        ledger.seeded = true
      } catch (error) {
        // Corrupt ledger: treat as unseeded (re-seed silently) rather than crash the daemon.
        this.#deps.log?.(
          `auto-review: unreadable ledger ${ledger.path}; re-seeding: ${error instanceof Error ? error.message : String(error)}`,
        )
      }
    }
    this.#ledgers.set(spaceUri, ledger)
    return ledger
  }

  /** Selects new auto-review candidates from `index` and fires their writes; returns immediately. */
  pump(index: MaterializedIndex, actors: ActorRegistry): void {
    const spaceUri = index.space.uri
    const ledger = this.#loadLedger(spaceUri)
    this.#reconcileDuplicates(index, actors)

    // Select against the ledger as it stood at the START of this tick. A first-ever snapshot
    // (unseeded) triggers nothing: pre-existing artifacts are seeded below and never reviewed.
    const deferred = new Set<string>()
    const candidates = ledger.seeded
      ? selectAutoReviewCandidates(
          index,
          actors,
          ledger.subjects,
          this.#alreadyWritten,
          (reason) => this.#logOnce(`not auto-reviewing: ${reason}`),
          (key) => deferred.add(key),
        )
      : []
    const ready: AutoReviewCandidate[] = []
    for (const candidate of candidates) {
      const key = refKey(candidate.artifact)
      const delay = candidate.writerRank * AUTO_REVIEW_STAGGER_TICKS
      const ticks = this.#staggerTicks.get(key) ?? 0
      if (ticks < delay) {
        this.#staggerTicks.set(key, ticks + 1)
        deferred.add(key)
        this.#deps.log?.(`auto-review: deferring ${candidate.artifact.uri} (rank ${candidate.writerRank}, tick ${ticks + 1}/${delay})`)
      } else {
        this.#staggerTicks.delete(key)
        ready.push(candidate)
      }
    }
    const candidateKeys = new Set(ready.map((candidate) => refKey(candidate.artifact)))

    // AFTER selection, record every OBSERVABLE artifact version (live goal + project; never
    // under an ended goal) EXCEPT the ones the ledger must not swallow: this tick's candidates (marked
    // only once their write lands — see `#write`), subjects deferred for want of a reviewer (must fire
    // once one is configured), and subjects with a write still in flight from an overlapping pump (so
    // a failed write always retries). This keeps a failed/crashed/deferred subject reselectable.
    for (const artifact of observableArtifacts(index)) {
      const key = refKey(artifact)
      if (candidateKeys.has(key) || deferred.has(key) || this.#inFlight.has(key)) continue
      // The common stagger path ends when another writer's canonical request arrives. Selection
      // then omits the subject, so discard its counter as soon as the subject becomes observable.
      this.#staggerTicks.delete(key)
      ledger.subjects.add(key)
    }
    ledger.seeded = true
    this.#persist(spaceUri, ledger)

    // Probe across ALL loaded review-capable actors (not just those currently index-eligible), so an
    // actor that wrote the request and then lost its grant is still probed before we re-create.
    const probeActors = loadedReviewActors(actors)
    for (const candidate of ready) {
      const key = refKey(candidate.artifact)
      if (this.#inFlight.has(key)) continue
      const promise = this.#write(candidate, spaceUri, probeActors).finally(() => this.#inFlight.delete(key))
      this.#inFlight.set(key, promise)
    }
  }

  #reconcileDuplicates(index: MaterializedIndex, actors: ActorRegistry): void {
    const byDid = new Map(actors.all.map((actor) => [actor.did, actor]))
    for (const view of [...index.goals, ...index.projects]) {
      for (const { duplicate, canonical } of duplicateReviewRequests(view, index.members).values()) {
        const reason = `duplicate auto-review request; canonical is ${canonical.uri}`
        // Execution and authorship are independent: any operator may be running this open request,
        // while only its author can retract it. Every daemon therefore abandons first; the author
        // gate below applies only to the protocol write.
        if (!this.#alreadyAbandoned.has(duplicate.uri)) {
          this.#alreadyAbandoned.add(duplicate.uri)
          this.#deps.onAbandon?.(duplicate.uri, reason)
        }
        const writer = byDid.get(duplicate.did)
        if (!writer || this.#retractions.has(duplicate.uri) || this.#alreadyRetracted.has(duplicate.uri)) continue
        const promise = this.#retract(writer, duplicate, reason).finally(() => this.#retractions.delete(duplicate.uri))
        this.#retractions.set(duplicate.uri, promise)
      }
    }
  }

  async #retract(
    writer: LoadedActor,
    request: IndexedRecord<ArtifactRequestRecord>,
    reason: string,
  ): Promise<void> {
    try {
      await writer.client.create(
        COLLECTIONS.retractRequest,
        { $type: COLLECTIONS.retractRequest, request: { uri: request.uri, cid: request.cid }, createdAt: this.#now() },
        { rkey: autoReviewRetractionRkey(request.uri, request.cid) },
      )
      this.#alreadyRetracted.add(request.uri)
      this.#deps.log?.(`auto-review: retracting duplicate review request ${request.uri}: ${reason}`)
    } catch (error) {
      if (error instanceof XrpcError && error.status === 400 && error.error === 'RecordAlreadyExists') {
        this.#alreadyRetracted.add(request.uri)
        return
      }
      this.#deps.log?.(`auto-review: failed to retract duplicate review request ${request.uri}: ${error instanceof Error ? error.message : String(error)}`)
    }
  }

  /** Snapshots and persists the ledger, chained per space (tmp+rename can't race a prior tick). The
   * returned tail is tracked for `drain`; `#write` awaits it so a mark-on-success is durable before
   * its in-flight promise settles. */
  #persist(spaceUri: string, ledger: ObservedLedger): Promise<void> {
    const snapshot: LedgerFile = { space: spaceUri, subjects: [...ledger.subjects].sort() }
    const tail = (this.#persists.get(spaceUri) ?? Promise.resolve())
      .then(() => this.#writeLedgerFile(ledger.path, snapshot))
      .catch((error: unknown) =>
        this.#deps.log?.(
          `auto-review: failed to persist ledger ${ledger.path}: ${error instanceof Error ? error.message : String(error)}`,
        ),
      )
    this.#persists.set(spaceUri, tail)
    return tail
  }

  /** Marks a subject observed (ledger + in-process guard) after its write landed, and awaits the
   * ledger persist so the durability the deterministic-rkey retry relies on is in place. */
  async #mark(spaceUri: string, key: string): Promise<void> {
    this.#alreadyWritten.add(key)
    const ledger = this.#ledgers.get(spaceUri)
    if (!ledger) return
    ledger.subjects.add(key)
    await this.#persist(spaceUri, ledger)
  }

  async #writeLedgerFile(path: string, snapshot: LedgerFile): Promise<void> {
    await mkdir(dirname(path), { recursive: true })
    const tmp = `${path}.tmp`
    await writeFile(tmp, JSON.stringify(snapshot))
    await rename(tmp, path) // atomic replace so a crash mid-write never leaves a torn ledger.
  }

  /** Whether an already-existing record found in `repoDid`'s repo is the auto-review request we would
   * have written for this subject: same type, subject, goal/project target, `basedOn` of exactly
   * `[subject]`, and either no assignment or legacy self-assignment to `repoDid`. Anything else is a
   * foreign record at the rkey. */
  #adoptable(
    value: ArtifactRequestRecord,
    subject: StrongRef,
    artifact: IndexedRecord<ArtifactRecord>,
    repoDid: string,
  ): boolean {
    const targetMatches = artifact.value.goal
      ? sameRef(value.goal, artifact.value.goal)
      : sameRef(value.project, artifact.value.project)
    const basedOnMatches =
      Array.isArray(value.basedOn) && value.basedOn.length === 1 && sameRef(value.basedOn[0], subject)
    return (
      value.type === REVIEW_TYPE &&
      sameRef(value.subject, subject) &&
      targetMatches &&
      basedOnMatches &&
      (value.assignee === undefined || value.assignee === repoDid)
    )
  }

  async #write(candidate: AutoReviewCandidate, spaceUri: string, probeActors: LoadedActor[]): Promise<void> {
    const { artifact, writer, retractedCount } = candidate
    const subject: StrongRef = { uri: artifact.uri, cid: artifact.cid }
    const key = refKey(artifact)
    const rkey = autoReviewRkey(subject.uri, subject.cid, retractedCount)

    // Probe the deterministic rkey across EVERY loaded review-capable actor's repo before creating, so
    // a restart that changed the lowest-DID writer (or an actor that lost its grant) still surfaces a
    // peer's earlier request instead of writing a second one that would never collide with it. A
    // mismatched record in one repo does NOT stop the probe — keep scanning; a valid match anywhere
    // wins. Only if none of the repos holds a valid match do we fall through to a create.
    let sawMismatch = false
    for (const probe of probeActors) {
      let existing: { uri: string; cid: string; value: RadialRecord } | undefined
      try {
        existing = await probe.client.getOwnRecord(COLLECTIONS.artifactRequest, rkey)
      } catch (error) {
        // A probe failure means we cannot rule out an existing request; do not risk a duplicate —
        // leave the subject unmarked and let the next pump retry.
        this.#deps.log?.(
          `auto-review: probe of ${probe.did} for ${subject.uri} failed: ${error instanceof Error ? error.message : String(error)}; retrying next tick`,
        )
        return
      }
      if (!existing) continue
      if (this.#adoptable(existing.value as ArtifactRequestRecord, subject, artifact, probe.did)) {
        await this.#mark(spaceUri, key)
        this.#deps.log?.(`auto-review: adopted existing review request of ${subject.uri} in ${probe.did}: ${existing.uri}`)
        return
      }
      sawMismatch = true // foreign record at this rkey here; keep probing the remaining repos.
    }
    if (sawMismatch) {
      // No repo held a valid match, but at least one held a foreign record at the deterministic rkey
      // (should be impossible — the rkey is a pure function of subject + retraction count). Do NOT
      // create (a peer's foreign record is unexpected), do NOT mark; log once per subject.
      this.#logOnce(
        `auto-review: refusing to adopt mismatched record at ${rkey} for subject ${subject.uri}`,
      )
      return
    }

    // The review request's goal/project target is copied EXACTLY from the subject artifact's own ref
    // — the materializer silently drops a review request whose target does not match its subject's
    // (materializer.ts:487-502).
    const record: ArtifactRequestRecord = {
      $type: COLLECTIONS.artifactRequest,
      ...(artifact.value.goal ? { goal: artifact.value.goal } : { project: artifact.value.project as StrongRef }),
      type: REVIEW_TYPE,
      subject,
      basedOn: [subject], // provenance recorded-when-known: the subject is what this review is of.
      createdAt: this.#now(),
    }
    // This create is the daemon's ONLY `artifactRequest` write, and its type is the constant three
    // lines up. That is the invariant keeping the `answer` turn out of design §14's rejected
    // alternative C (design §10): an answer request is only ever written by a human, so a reply can
    // never commission the next reply and a conversation cannot sustain itself without a person
    // writing a record each time. It cannot be asserted at runtime here — there is no input that
    // would make this record's type anything but `review` — so it is asserted where it can actually
    // fail, over the source: `test/auto-review.test.mjs`, "the daemon writes no other request".
    let ref: StrongRef
    try {
      ref = await writer.client.create(COLLECTIONS.artifactRequest, record, { rkey })
    } catch (error) {
      if (error instanceof XrpcError && error.status === 400 && error.error === 'RecordAlreadyExists') {
        // Raced with a write between the probe and this create: re-fetch from the selected actor.
        const existing = await writer.client.getOwnRecord(COLLECTIONS.artifactRequest, rkey).catch(() => undefined)
        if (existing && this.#adoptable(existing.value as ArtifactRequestRecord, subject, artifact, writer.did)) {
          await this.#mark(spaceUri, key)
          return
        }
        this.#logOnce(
          `auto-review: refusing to adopt mismatched record at ${rkey} in ${writer.did} for subject ${subject.uri}`,
        )
        return
      }
      // A transient failure: leave the subject UNMARKED so the next pump reselects and retries it.
      this.#deps.log?.(
        `auto-review: failed to write review request for ${subject.uri}: ${error instanceof Error ? error.message : String(error)}; retrying next tick`,
      )
      return
    }
    await this.#mark(spaceUri, key)
    this.#deps.log?.(`auto-review: authored open review request for ${subject.uri} as ${writer.did}: ${ref.uri}`)
  }

  /** Awaits every in-flight write and ledger persist (best-effort graceful shutdown). */
  async drain(): Promise<void> {
    await Promise.all([...this.#inFlight.values(), ...this.#retractions.values(), ...this.#persists.values()])
  }
}