polyc-controller 2026.8.3

//! Reconciler: drive each [`Conversation`] toward one agent-sandbox
//! `SandboxClaim`.
//!
//! The *decision* (what should happen) is the pure [`plan`] function, unit-
//! tested without a cluster. The *effect* (applying that decision) is
//! [`reconcile`](fn@reconcile), which talks to the kube API. Keeping them apart
//! means the interesting branching logic is covered by fast, hermetic tests.

use std::{
    fmt::Write as _,
    sync::{Arc, LazyLock, Mutex},
    time::{Duration, Instant},
};

use futures::StreamExt;
use kube::{
    Api, Client, Resource, ResourceExt,
    api::{DeleteParams, ListParams, Patch, PatchParams, Preconditions},
    runtime::{
        controller::{Action, Controller},
        watcher,
    },
};
use polyc_k8s_types::sandboxclaim::SandboxClaim;
use prometheus::{IntCounter, register_int_counter};
use serde_json::json;

use crate::conversation::{Condition, ConditionStatus, Conversation, FINALIZER, upsert_condition};
use crate::execution_backend::{
    ClaimReadiness, DialAddress, ExecutionBackend, SandboxClaimBackend,
};

/// The shared `SandboxTemplate` every conversation's pod is claimed from
/// (pre-provisioned out-of-band; see PRD §15).
pub const DEFAULT_TEMPLATE: &str = "polychrome-harness-default";

/// `status.conditions[].type` for the "can this conversation take a turn
/// right now" signal.
pub const COND_READY: &str = "Ready";
/// `status.conditions[].type` for "actively moving toward `Ready`" (claim
/// creation, harness boot, teardown, idle-close).
pub const COND_PROGRESSING: &str = "Progressing";
/// `status.conditions[].type` for "an anomaly was observed" (today: the
/// issue #2 self-heal, when the execution unit vanished out-of-band).
pub const COND_DEGRADED: &str = "Degraded";

/// `Conversation.status.phase` values. Defined once so the self-heal heuristic
/// (`unit_is_gone`) and the status writer agree on the literal — a rename
/// can't silently desync them.
pub const PHASE_READY: &str = "Ready";
/// The unit is provisioned but not yet `Ready` (also the post-self-heal reset).
pub const PHASE_PENDING: &str = "Pending";
/// Idle, but reclaimed-and-resumable rather than closed.
///
/// No current code path sets `status.idle_reclaimed` true — it is a
/// CRD-additive-only field with no writer left after the backend that used to
/// set it (a resource-reclaiming idle disposition) was removed. This phase and
/// the idle-reaper gate that reads the field stay in place defensively: the
/// field can still be true on an existing object (e.g. a manual edit), and
/// faithfully reflecting that state — a stuck-open, never-reaped conversation
/// — is safer than ignoring it outright.
pub const PHASE_PAUSED: &str = "Paused";
/// Transient display phase set the instant a [`ReconcileAction::RollHarness`]
/// tears the claim down.
///
/// For `kubectl get conversation` visibility only — purely cosmetic. The
/// AUTHORITATIVE signal the concurrency cap counts against is
/// `status.rolling_harness` (see `Context::count_in_flight_rolls`), because
/// this phase string is overwritten back to [`PHASE_PENDING`] by the very next
/// `CreateSandboxClaim` pass, well before the new pod is `Ready`.
pub const PHASE_ROLLING_HARNESS: &str = "RollingHarness";

/// Concurrency cap on simultaneous per-conversation harness rolls, applied
/// when `POLYCHROME_HARNESS_ROLL_CONCURRENCY` is unset, empty, `0`, or fails
/// to parse. See [`Context::harness_roll_concurrency`].
const DEFAULT_HARNESS_ROLL_CONCURRENCY: usize = 5;

/// Default retention window (seconds) a fully-closed, claim-less
/// `Conversation` CR is kept before [`ReconcileAction::DeleteConversation`]
/// removes it — applied when `POLYCHROME_CLOSED_CONVERSATION_RETENTION_SECONDS`
/// is unset (the GC is on by default). See
/// [`parse_closed_conversation_retention_seconds`] for the full parsing rules
/// and [`Context::closed_conversation_retention_seconds`].
const DEFAULT_CLOSED_CONVERSATION_RETENTION_SECS: i64 = 86_400;

/// Parse `POLYCHROME_CLOSED_CONVERSATION_RETENTION_SECONDS`'s raw value (as
/// read by [`run_conversation`]) into the closed-conversation GC's retention
/// window. Pure — testable without touching the environment.
///
/// - Not set (`None`) ⇒ [`DEFAULT_CLOSED_CONVERSATION_RETENTION_SECS`] — the
///   GC is on by default.
/// - Empty or `"0"` ⇒ disabled (kept forever, the pre-fix behavior) — an
///   explicit operator opt-out.
/// - A positive integer ⇒ that many seconds.
/// - Anything else (a negative number, or unparseable/malformed input, e.g. a
///   typo) ⇒ disabled, NOT silently defaulted. The action this window gates
///   is destructive ([`ReconcileAction::DeleteConversation`]), so malformed
///   input fails toward "never delete" rather than toward a default the
///   operator never asked for.
#[must_use]
fn parse_closed_conversation_retention_seconds(raw: Option<&str>) -> Option<i64> {
    match raw {
        None => Some(DEFAULT_CLOSED_CONVERSATION_RETENTION_SECS),
        Some(v) if v.trim().is_empty() => None,
        Some(v) => match v.trim().parse::<i64>() {
            Ok(n) if n > 0 => Some(n),
            Ok(_) | Err(_) => None,
        },
    }
}

/// How long [`Context::count_in_flight_rolls`]'s cached count may be reused
/// before a fresh `Api<Conversation>::list` re-derives it.
const ROLL_COUNT_CACHE_TTL: Duration = Duration::from_secs(5);

/// Requeue cadence after a [`ReconcileAction::RollHarness`] teardown (fast,
/// like [`ReconcileAction::CreateSandboxClaim`]'s [`PENDING_POLL`]-adjacent
/// pattern) so the recreate isn't stalled behind the steady poll cadence.
const ROLL_REQUEUE: Duration = Duration::from_secs(1);

/// Requeue cadence when a would-be roll is deferred by the concurrency cap
/// ([`Context::harness_roll_concurrency`]). Short enough that a freed slot is
/// picked up promptly, long enough not to hammer the list-count cache.
const ROLL_DEFER_POLL: Duration = Duration::from_secs(15);

/// Requeue cadence while a unit is still `Pending` (provisioning / resuming):
/// poll tightly so the `Ready` transition is observed promptly. Applies to every
/// backend regardless of [`ExecutionBackend::ready_poll_interval`].
pub(crate) const PENDING_POLL: Duration = Duration::from_secs(30);

/// Steady-state requeue cadence for an already-`Ready` agent-sandbox unit. Long,
/// because the `.owns(SandboxClaim)` watch already delivers changes reactively —
/// this poll is only a backstop, so it need not be frequent.
pub(crate) const SANDBOX_READY_POLL: Duration = Duration::from_mins(5);

/// How long a `Conversation` may sit in `Terminating` with a failing teardown
/// before the finalizer is force-removed (accepting a possibly-leaked execution
/// unit rather than an undeletable CR). See the `Cleanup` arm in [`reconcile`].
const TEARDOWN_GRACE: Duration = Duration::from_mins(10);

/// How long `conv` has been pending deletion (now − `deletionTimestamp`), or
/// [`Duration::ZERO`] if it isn't being deleted or the clock can't be read.
/// Impure (reads the wall clock) — lives in [`reconcile`], not [`plan`].
fn deletion_age(conv: &Conversation) -> Duration {
    use k8s_openapi::jiff::Timestamp;
    conv.meta()
        .deletion_timestamp
        .as_ref()
        .map_or(Duration::ZERO, |t| {
            Timestamp::now().duration_since(t.0).unsigned_abs()
        })
}

/// The current wall-clock time, RFC3339-formatted, for stamping a fresh
/// `Condition::last_transition_time`. Impure — lives in [`reconcile`], not the
/// pure [`conditions_triad`]/`upsert_condition`, mirroring [`deletion_age`].
fn now_rfc3339() -> String {
    k8s_openapi::jiff::Timestamp::now().to_string()
}

/// Build the `Ready`/`Progressing`/`Degraded` condition triad for a status
/// patch: exactly one pure entry point so every status-patching call site in
/// [`reconcile`] produces the same three condition types with the same
/// upsert semantics (see [`upsert_condition`] — an unchanged condition keeps
/// its `lastTransitionTime`). `reason`/`message` are attached to all three
/// conditions; they narrate the reconcile-time event that produced this
/// triad (e.g. `"HarnessReady"`), not a per-condition-type distinction.
#[must_use]
fn conditions_triad(
    previous: &[Condition],
    ready: bool,
    progressing: bool,
    degraded: bool,
    reason: &str,
    message: &str,
    now: &str,
) -> Vec<Condition> {
    let mut conditions = previous.to_vec();
    upsert_condition(
        &mut conditions,
        COND_READY,
        ConditionStatus::from(ready),
        reason,
        message,
        now,
    );
    upsert_condition(
        &mut conditions,
        COND_PROGRESSING,
        ConditionStatus::from(progressing),
        reason,
        message,
        now,
    );
    upsert_condition(
        &mut conditions,
        COND_DEGRADED,
        ConditionStatus::from(degraded),
        reason,
        message,
        now,
    );
    conditions
}

/// Counts reconcile passes; registered to the prometheus default registry and
/// scraped via the control plane's `/metrics` endpoint.
static RECONCILE_TOTAL: LazyLock<IntCounter> = LazyLock::new(|| {
    register_int_counter!(
        "polychrome_reconcile_total",
        "Total Conversation reconcile passes"
    )
    .expect("register polychrome_reconcile_total")
});

/// Errors surfaced by the reconciler.
#[derive(Debug, thiserror::Error)]
pub enum Error {
    /// A kube API call failed.
    #[error("kube api: {0}")]
    Kube(#[from] kube::Error),
    /// A namespaced object arrived without a namespace (should not happen).
    #[error("conversation has no namespace")]
    NoNamespace,
    /// An [`ExecutionBackend`] operation failed for a reason that isn't a kube
    /// API error. Carries the backend's own error rendered as a string so
    /// [`reconcile`] stays agnostic to which backend produced it.
    #[error("execution backend: {0}")]
    Backend(String),
}

/// Lift the `ToolService` reconciler's `Error` (identical `Kube`/`NoNamespace`
/// shape) into this one, so [`run`] can `?`-propagate its watch's join result
/// instead of hand-writing the match per reconciler.
impl From<crate::toolservice_reconcile::Error> for Error {
    fn from(err: crate::toolservice_reconcile::Error) -> Self {
        match err {
            crate::toolservice_reconcile::Error::Kube(e) => Self::Kube(e),
            crate::toolservice_reconcile::Error::NoNamespace => Self::NoNamespace,
        }
    }
}

/// Lift the `Routine` reconciler's `Error` into this one; see
/// [`From<crate::toolservice_reconcile::Error>`].
impl From<crate::routine_reconcile::Error> for Error {
    fn from(err: crate::routine_reconcile::Error) -> Self {
        match err {
            crate::routine_reconcile::Error::Kube(e) => Self::Kube(e),
            crate::routine_reconcile::Error::NoNamespace => Self::NoNamespace,
        }
    }
}

/// Lift the `ServiceDefinition` reconciler's `Error` into this one; see
/// [`From<crate::toolservice_reconcile::Error>`].
impl From<crate::servicedefinition_reconcile::Error> for Error {
    fn from(err: crate::servicedefinition_reconcile::Error) -> Self {
        match err {
            crate::servicedefinition_reconcile::Error::Kube(e) => Self::Kube(e),
            crate::servicedefinition_reconcile::Error::NoNamespace => Self::NoNamespace,
        }
    }
}

/// Lift the `Workflow` reconciler's `Error` into this one; see
/// [`From<crate::toolservice_reconcile::Error>`].
impl From<crate::workflow_reconcile::Error> for Error {
    fn from(err: crate::workflow_reconcile::Error) -> Self {
        match err {
            crate::workflow_reconcile::Error::Kube(e) => Self::Kube(e),
            crate::workflow_reconcile::Error::NoNamespace => Self::NoNamespace,
        }
    }
}

/// What a single reconcile pass should do for a [`Conversation`]. The pure
/// output of [`plan`]; [`reconcile`] turns it into kube API calls.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ReconcileAction {
    /// Attach our [`FINALIZER`] before creating any owned resources.
    AddFinalizer,
    /// Create a `SandboxClaim` named `claim_name` from `template`.
    CreateSandboxClaim {
        /// Name to give the `SandboxClaim` (mirrors the conversation's name).
        claim_name: String,
        /// `SandboxTemplate` to claim a pod from.
        template: String,
    },
    /// Tear down the owned `SandboxClaim`. `remove_finalizer` is set only when
    /// the `Conversation` itself is being deleted (vs merely `closed`).
    Cleanup {
        /// The claim to delete, if one was recorded.
        claim_name: Option<String>,
        /// Whether to also drop our finalizer (true ⇒ object is terminating).
        remove_finalizer: bool,
    },
    /// The claim exists; read its readiness and reflect it into the
    /// `Conversation` status (pod IP, `harness_ready`).
    SyncStatus {
        /// The owned `SandboxClaim` to read readiness from.
        claim_name: String,
    },
    /// The claim exists but its recorded `status.harnessImageGeneration`
    /// no longer matches [`Context::desired_harness_generation`] — tear it
    /// down so the next pass recreates it at the new image
    /// (`docs/reference/upgrade-delivery.md` §8.1, issue #117 phase 2).
    ///
    /// # Mechanism
    ///
    /// The reconcile effect deletes the claim and clears the pod-address /
    /// readiness / generation status fields (same fields
    /// [`ReconcileAction::Cleanup`]'s non-finalizer-removal branch already
    /// clears), stamping `phase: "RollingHarness"` and
    /// `status.rolling_harness = true` so the concurrency cap
    /// ([`Context::harness_roll_concurrency`]) can see it. The very next
    /// `plan()` pass then observes `claim_name: None` and re-enters
    /// [`ReconcileAction::CreateSandboxClaim`], recreating the pod at
    /// whatever image the shared `SandboxTemplate` now resolves to; the pass
    /// after THAT (`SyncStatus` with a freshly-`None` recorded generation)
    /// re-stamps the new desired generation through the same
    /// adopt-without-rolling path a first-time enable uses (see the
    /// None-vs-Some semantics on [`plan`]) — there is exactly one stamping
    /// path, reused for both. Resume is automatic: there is no separate
    /// "resume" RPC or step, because [event-log
    /// replay](crate::conversation) already reconstructs conversation state
    /// on every fresh harness pod dialed for the same conversation.
    ///
    /// # This may interrupt an in-flight turn — read before enabling this
    ///
    /// There is deliberately **no new "is this conversation mid-turn" signal**
    /// in this change — that is a bigger feature, explicitly out of scope
    /// here. If a turn is running on this conversation's harness pod when a
    /// roll fires, the pod is killed mid-generation. Recovery relies entirely
    /// on the EXISTING kill-and-resume property the event log already
    /// provides on every harness restart. D5 changed what that property
    /// costs. State now commits every accepted step
    /// (`crates/control-plane/src/step_commit.rs`), and the replacement pod
    /// resumes strictly after State's last committed step
    /// (`crates/control-plane/src/grpc/turn.rs`'s `resume_after`). A roll
    /// landing mid-turn therefore repeats only the steps State had not yet
    /// accepted, not the whole turn. The caller still waits for the resumed
    /// steps to finish, so a roll during a turn is still not free; an
    /// operator who cannot tolerate that delay holds back
    /// `POLYCHROME_HARNESS_IMAGE_GENERATION` until a quieter window.
    RollHarness {
        /// The claim to tear down.
        claim_name: String,
    },
    /// The conversation has been idle past `spec.idleTimeoutSeconds`; mark it
    /// closed so the next reconcile tears the harness down via
    /// [`ReconcileAction::Cleanup`]. This
    /// is the working idle-GC (issue #265): the claim's `ttlSecondsAfterFinished`
    /// never fires because a long-running harness never "finishes".
    CloseIdle,
    /// `conv` has been fully closed (its execution unit already torn down,
    /// `status.closedAt` stamped) for longer than the configured retention
    /// window — delete the object outright. This is the closed-conversation
    /// GC: without it, a closed `Conversation` settles into [`Self::Noop`]
    /// forever, since nothing else ever calls delete on it (the idle-GC only
    /// reaps the execution unit, never the CR).
    ///
    /// Distinct from the `being_deleted` branch's
    /// `Cleanup{remove_finalizer: true}`, which only clears the finalizer on
    /// an object ALREADY marked for deletion (`deletionTimestamp` set) — this
    /// action is what actually calls `delete()` in the first place. The
    /// finalizer then runs its normal course on the next pass, since a
    /// deleted-with-finalizer object being deleted again re-enters
    /// `Cleanup{claim_name: None, remove_finalizer: true}` — a no-op teardown
    /// (no claim left) that just clears the finalizer so the object is
    /// actually removed.
    DeleteConversation,
    /// Steady state; nothing to do.
    Noop,
}

/// Decide whether a `SyncStatus` pass has observed that the execution unit
/// has *vanished* (self-heal trigger for issue #2), given the conversation's
/// last-recorded `status` and the freshly-read `readiness`. Pure.
///
/// # Why this lives outside [`plan`]
///
/// [`plan`] sees only the `Conversation` CR, never the live execution unit, so
/// it cannot tell "claim recorded, unit healthy" from "claim recorded, unit
/// GC'd". Once `status.sandboxClaimName` is set, [`plan`] always yields
/// `SyncStatus`, whose only IO is a pure *read* ([`ExecutionBackend::readiness`])
/// — it never re-creates. If the underlying unit disappears (`SandboxClaim`
/// deleted out-of-band) the conversation would be wedged un-`Ready` forever.
/// The fix is level-triggered:
/// the `SyncStatus` reconcile arm consults this predicate and, when it returns
/// `true`, clears `status.sandboxClaimName`; the *next* [`plan`] pass then sees
/// `None` and re-enters [`ReconcileAction::CreateSandboxClaim`], re-creating the
/// unit. Keeping the decision in this pure helper preserves [`plan`]'s purity
/// and keeps the heuristic unit-testable without a cluster.
///
/// # Distinguishing "gone" from "still starting" (and from "transiently down")
///
/// The discriminator is [`ClaimReadiness::unit_present`], set by the backend
/// from whether the unit *object* exists — not from whether its harness is
/// reachable. This is what stops a transient pod restart on the default
/// agent-sandbox backend from looking "gone": the `SandboxClaim` still exists
/// while its pod bounces, so `unit_present` stays `true` and we do nothing. Only
/// a genuinely absent unit (`claim == None`) reports `unit_present == false`.
///
/// Even then we require *evidence the unit once existed and was healthy* —
/// `status.phase == "Ready"`, or a prior address was recorded — before
/// re-creating, so a just-created-not-yet-observed unit isn't thrashed during
/// boot. (After a self-heal fires it nulls that evidence, so the predicate is
/// one-shot and can't loop.)
#[must_use]
fn unit_is_gone(
    status: &crate::conversation::ConversationStatus,
    readiness: &ClaimReadiness,
) -> bool {
    // Unit object still exists (even if its harness is mid-restart) → not gone.
    if readiness.unit_present {
        return false;
    }
    // Unit genuinely absent: only "gone" if we have evidence it once was healthy.
    let was_ready = status.phase.as_deref() == Some(PHASE_READY);
    let had_address = status.pod_ip.is_some() || status.harness_endpoint.is_some();
    was_ready || had_address
}

/// Outcome of [`sync_execution_unit`]: either the unit vanished (self-heal —
/// see [`unit_is_gone`]), or it's present and its freshly re-synced
/// [`ClaimReadiness`] is returned.
#[derive(Debug, Clone, PartialEq, Eq)]
enum SyncOutcome {
    /// The unit vanished out-of-band; the caller clears `sandboxClaimName` so
    /// the next [`plan`] pass re-creates it.
    Vanished,
    /// The unit is present; its spec has been re-applied (drift healed) and
    /// its readiness read.
    Synced(ClaimReadiness),
}

/// The `SyncStatus` action's IO against the [`ExecutionBackend`] only — no
/// `Conversation`-CR kube calls. Isolated from [`reconcile`]'s `Api<Conversation>`
/// status-patch calls so it is unit-testable with a mocked backend, without a
/// live kube API server for the `Conversation` object itself.
///
/// # Order matters
///
/// Readiness is read **first**. If [`unit_is_gone`] says the unit vanished,
/// we return immediately without touching the backend further — calling
/// `ensure` here would silently *resurrect* the vanished unit within this
/// same pass, masking the vanish and defeating the self-heal path (which
/// intentionally recreates it on a LATER pass, after `reconcile` has cleared
/// `sandboxClaimName`). Only once the unit is confirmed present do we
/// `ensure` the desired spec — this is issue #802's drift healing: an
/// out-of-band `SandboxClaim` spec edit (or any other execution-unit drift)
/// is reverted by this idempotent server-side re-apply on every pass a live
/// unit is observed, rather than persisting until the claim is deleted and
/// recreated.
///
/// # Errors
///
/// Returns [`Error`] if either backend call fails.
async fn sync_execution_unit(
    backend: &dyn ExecutionBackend,
    conv: &Conversation,
    claim_name: &str,
    template: &str,
    ns: &str,
) -> Result<SyncOutcome, Error> {
    let current = conv.status.clone().unwrap_or_default();
    let readiness = backend.readiness(claim_name, ns).await?;
    if unit_is_gone(&current, &readiness) {
        return Ok(SyncOutcome::Vanished);
    }
    backend.ensure(conv, claim_name, template, ns).await?;
    Ok(SyncOutcome::Synced(readiness))
}

/// Decide what to do for `conv`. Pure: no IO. `now_unix` (the clock) is passed
/// in so the function stays fully testable.
///
/// `desired_harness_generation` is [`Context::desired_harness_generation`] —
/// the operator-set `POLYCHROME_HARNESS_IMAGE_GENERATION`, an opaque string
/// compared for equality only, never parsed. Once a claim exists
/// (`status.sandboxClaimName` is `Some`), it governs the choice between
/// [`ReconcileAction::SyncStatus`] and [`ReconcileAction::RollHarness`]:
///
/// | `desired_harness_generation` | recorded `status.harnessImageGeneration` | action |
/// |---|---|---|
/// | `None` (feature off) | *(any)* | `SyncStatus` — unchanged from before this feature existed |
/// | `Some(desired)` | `None` (never recorded) | `SyncStatus` — **adopt without rolling**; the `SyncStatus` reconcile arm stamps `desired` into status on this pass, since `plan` is pure and cannot itself patch status |
/// | `Some(desired)` | `Some(recorded)`, `recorded == desired` | `SyncStatus` — steady state |
/// | `Some(desired)` | `Some(recorded)`, `recorded != desired` | `RollHarness` — the only case that tears the claim down |
///
/// The middle row is the case an adversarial review flagged as the easiest to
/// get wrong: a conversation that predates this feature (or was created
/// before an operator ever set the env var) must NOT be rolled the first time
/// the desired generation becomes known — there is nothing to roll *from*,
/// only a gap to fill in.
///
/// `retention_seconds` is [`Context::closed_conversation_retention_seconds`] —
/// once `conv` is fully closed (`status.closedAt` stamped, no claim left),
/// `Some(n)` deletes it outright ([`ReconcileAction::DeleteConversation`])
/// once `now_unix - closedAt > n`; `None` disables the closed-conversation GC
/// and preserves the old behavior of settling into [`ReconcileAction::Noop`]
/// forever. A `closed` conversation with no `closedAt` stamp (predates this
/// field, or a hypothetical future close path that never stamps it) also
/// settles to `Noop` regardless of `retention_seconds` — no stamp, no GC,
/// mirroring the idle reaper's own rule.
#[must_use]
pub fn plan(
    conv: &Conversation,
    default_template: &str,
    now_unix: i64,
    desired_harness_generation: Option<&str>,
    retention_seconds: Option<i64>,
) -> ReconcileAction {
    let being_deleted = conv.meta().deletion_timestamp.is_some();
    let has_finalizer = conv.finalizers().iter().any(|f| f == FINALIZER);
    let closed = conv.status.as_ref().is_some_and(|s| s.closed);
    let claim_name = conv
        .status
        .as_ref()
        .and_then(|s| s.sandbox_claim_name.clone());

    if being_deleted {
        return if has_finalizer {
            ReconcileAction::Cleanup {
                claim_name,
                remove_finalizer: true,
            }
        } else {
            ReconcileAction::Noop
        };
    }

    if closed {
        // Delete the pod claim but leave the object (and finalizer) until it is
        // actually deleted. Once the claim name is cleared, delete the object
        // outright once it's sat past the retention window — or settle to Noop
        // (the pre-fix behavior) if there's no stamp to measure from, or the GC
        // is disabled.
        return if claim_name.is_some() {
            ReconcileAction::Cleanup {
                claim_name,
                remove_finalizer: false,
            }
        } else if let Some(retention) = retention_seconds
            && let Some(closed_at) = conv.status.as_ref().and_then(|s| s.closed_at)
            && now_unix.saturating_sub(closed_at) > retention
        {
            ReconcileAction::DeleteConversation
        } else {
            ReconcileAction::Noop
        };
    }

    if !has_finalizer {
        return ReconcileAction::AddFinalizer;
    }

    // Idle reap (issue #265): close a conversation whose last turn is older than
    // its idle timeout so the harness pod is torn down via Cleanup. Reap ONLY on
    // a real `last_activity_unix` stamp (the control plane writes it on EVERY
    // turn, incl. the first). NO fallback: using creationTimestamp as a proxy
    // would close OLD-but-ACTIVE conversations whose stamp simply hadn't landed
    // yet — the regression that closed live conversations. No stamp ⇒ never
    // closed. `0` disables.
    let idle_timeout = i64::from(conv.spec.idle_timeout_seconds);
    if idle_timeout > 0
        && let Some(last) = conv.status.as_ref().and_then(|s| s.last_activity_unix)
        && now_unix.saturating_sub(last) > idle_timeout
        // Defensive: no code path sets `idle_reclaimed` true anymore (see
        // `PHASE_PAUSED`'s doc comment), but honor it if an existing object
        // already carries it rather than re-reaping regardless.
        && !conv.status.as_ref().is_some_and(|s| s.idle_reclaimed)
    {
        return ReconcileAction::CloseIdle;
    }

    claim_name.map_or_else(
        || ReconcileAction::CreateSandboxClaim {
            claim_name: conv.name_any(),
            template: default_template.to_owned(),
        },
        |claim_name| {
            // See this function's doc comment for the full None-vs-Some
            // table. Only a RECORDED generation that actively disagrees with
            // the desired one triggers a roll; a never-recorded generation
            // adopts silently via SyncStatus instead.
            if let Some(desired) = desired_harness_generation {
                let recorded = conv
                    .status
                    .as_ref()
                    .and_then(|s| s.harness_image_generation.as_deref());
                if let Some(recorded) = recorded
                    && recorded != desired
                {
                    return ReconcileAction::RollHarness { claim_name };
                }
            }
            ReconcileAction::SyncStatus { claim_name }
        },
    )
}

/// Shared reconcile context.
pub struct Context {
    /// Kube client used for `Conversation`-CR API calls (finalizer + status).
    pub client: Client,
    /// `SandboxTemplate` name to claim pods from.
    pub default_template: String,
    /// The execution backend the reconciler drives conversations onto
    /// (see [`ExecutionBackend`]).
    pub backend: Arc<dyn ExecutionBackend>,
    /// The operator-set desired harness image generation
    /// (`POLYCHROME_HARNESS_IMAGE_GENERATION`), read once at controller
    /// startup. An opaque string (a release version like the workspace's
    /// `CalVer` `2026.7.0`, but never parsed or compared semantically here —
    /// only equality matters). `None` (the env var unset) means the
    /// roll-on-image-change feature (issue #117 phase 2,
    /// `docs/reference/upgrade-delivery.md` §8.1) is OFF entirely: [`plan`]
    /// never emits [`ReconcileAction::RollHarness`].
    pub desired_harness_generation: Option<String>,
    /// Concurrency cap on simultaneous per-conversation harness rolls
    /// (`POLYCHROME_HARNESS_ROLL_CONCURRENCY`, default `DEFAULT_HARNESS_ROLL_CONCURRENCY`),
    /// read once at controller startup. Enforced by [`reconcile`]'s
    /// `RollHarness` arm via `Context::count_in_flight_rolls` — never by
    /// [`plan`], which stays pure and has no cluster-wide view.
    pub harness_roll_concurrency: usize,
    /// Retention window (seconds) a fully-closed, claim-less `Conversation`
    /// CR is kept before [`plan`] emits [`ReconcileAction::DeleteConversation`]
    /// (`POLYCHROME_CLOSED_CONVERSATION_RETENTION_SECONDS`, default
    /// `DEFAULT_CLOSED_CONVERSATION_RETENTION_SECS`), read once at
    /// controller startup. `None` disables the GC entirely — see [`plan`]'s
    /// doc comment for the full semantics.
    pub closed_conversation_retention_seconds: Option<i64>,
    /// Cache backing [`Context::count_in_flight_rolls`]. Not `pub`: purely an
    /// implementation detail of the cap's cost model.
    roll_count_cache: Mutex<RollCountCache>,
}

/// Cached result of the last cluster-wide in-flight-roll count. See
/// [`Context::count_in_flight_rolls`] for the cost model this cache exists
/// to bound.
#[derive(Debug, Default)]
struct RollCountCache {
    /// When the count was last refreshed; `None` before the first read.
    refreshed_at: Option<Instant>,
    /// The count as of `refreshed_at`.
    count: usize,
}

impl Context {
    /// Construct a [`Context`], starting the roll-count cache fresh (empty —
    /// its first read always re-lists). `desired_harness_generation` and
    /// `harness_roll_concurrency` are normally read from the environment by
    /// `run_conversation`'s `POLYCHROME_HARNESS_IMAGE_GENERATION`/
    /// `POLYCHROME_HARNESS_ROLL_CONCURRENCY` lookup before this is called;
    /// this constructor exists so callers outside this module (the
    /// `run_conversation` entrypoint below, plus `tests/e2e.rs`) don't need
    /// to see `roll_count_cache`, which is a private implementation detail.
    #[must_use]
    pub fn new(
        client: Client,
        default_template: String,
        backend: Arc<dyn ExecutionBackend>,
        desired_harness_generation: Option<String>,
        harness_roll_concurrency: usize,
        closed_conversation_retention_seconds: Option<i64>,
    ) -> Self {
        Self {
            client,
            default_template,
            backend,
            desired_harness_generation,
            harness_roll_concurrency,
            closed_conversation_retention_seconds,
            roll_count_cache: Mutex::new(RollCountCache::default()),
        }
    }

    /// A would-be [`ReconcileAction::RollHarness`] teardown for `ns` may
    /// proceed. Counts conversations whose `status.rolling_harness` is
    /// `true` — set the instant a roll tears a claim down, cleared once the
    /// replacement claim resyncs back to `Ready` (see
    /// [`ReconcileAction::RollHarness`]'s doc comment) — against
    /// [`Self::harness_roll_concurrency`].
    ///
    /// # Cost model
    ///
    /// No `kube::runtime::reflector::Store<Conversation>` is wired for this
    /// controller — [`run_conversation`] drives a bare
    /// `Controller::new(...).owns(...)` watch with no exposed cache — so
    /// there is no already-warm, in-memory index of every conversation's
    /// status to count against for free. The only way to get a cluster-wide
    /// count without one is a genuine `Api<Conversation>::list` call. Calling
    /// it on every single reconcile tick (one per conversation, every
    /// `~30s`-`5m` per [`PENDING_POLL`]/[`SANDBOX_READY_POLL`]) would
    /// multiply API-server list load by the size of the fleet. Instead the
    /// count is cached for [`ROLL_COUNT_CACHE_TTL`] and only re-listed once
    /// that staleness window elapses, so list-call volume is bounded by
    /// wall-clock time (at most one list per TTL window, cluster-wide, no
    /// matter how many conversations exist), not by reconcile frequency or
    /// fleet size. The staleness this trades away is bounded and benign: the
    /// cap is a soft concurrency limiter on a background maintenance
    /// operation, not a correctness invariant, so briefly running a handful
    /// of rolls over or under `harness_roll_concurrency` within one TTL
    /// window is harmless.
    ///
    /// # Errors
    ///
    /// Returns [`Error`] if the list call fails.
    async fn count_in_flight_rolls(&self, ns: &str) -> Result<usize, Error> {
        let cached = self
            .roll_count_cache
            .lock()
            .expect("roll count cache lock poisoned")
            .fresh();
        if let Some(count) = cached {
            return Ok(count);
        }
        let convs: Api<Conversation> = Api::namespaced(self.client.clone(), ns);
        let list = convs.list(&ListParams::default()).await?;
        let count = list
            .items
            .iter()
            .filter(|c| c.status.as_ref().is_some_and(|s| s.rolling_harness))
            .count();
        self.roll_count_cache
            .lock()
            .expect("roll count cache lock poisoned")
            .store(count);
        Ok(count)
    }
}

impl RollCountCache {
    /// The cached count, if it hasn't yet aged past [`ROLL_COUNT_CACHE_TTL`].
    fn fresh(&self) -> Option<usize> {
        self.refreshed_at
            .filter(|at| at.elapsed() < ROLL_COUNT_CACHE_TTL)
            .map(|_| self.count)
    }

    /// Record a freshly-read count, timestamped now.
    fn store(&mut self, count: usize) {
        self.refreshed_at = Some(Instant::now());
        self.count = count;
    }
}

/// Whether a would-be [`ReconcileAction::RollHarness`] teardown may proceed
/// given `in_flight` conversations already mid-roll and the configured `cap`.
/// Pure — kept separate from [`Context::count_in_flight_rolls`] so the cap
/// arithmetic itself is unit-testable without a kube API.
#[must_use]
const fn roll_allowed(in_flight: usize, cap: usize) -> bool {
    in_flight < cap
}

/// Reconcile one [`Conversation`] by executing the [`plan`].
///
/// # Errors
///
/// Returns [`Error`] if any kube API call fails or the object lacks a namespace.
// The reconcile effect dispatcher is one match over every `ReconcileAction`
// arm; splitting the arms into free functions would scatter the kube IO without
// making it clearer. The decision logic lives in the pure `plan`.
#[allow(clippy::too_many_lines)]
#[tracing::instrument(skip_all, fields(conversation = %conv.name_any()))]
pub async fn reconcile(conv: Arc<Conversation>, ctx: Arc<Context>) -> Result<Action, Error> {
    RECONCILE_TOTAL.inc();
    let ns = conv.namespace().ok_or(Error::NoNamespace)?;
    let name = conv.name_any();
    let convs: Api<Conversation> = Api::namespaced(ctx.client.clone(), &ns);
    let pp = PatchParams::apply("polychrome.dev/controller");

    let now_unix = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX));

    match plan(
        &conv,
        &ctx.default_template,
        now_unix,
        ctx.desired_harness_generation.as_deref(),
        ctx.closed_conversation_retention_seconds,
    ) {
        ReconcileAction::AddFinalizer => {
            let patch = json!({ "metadata": { "finalizers": [FINALIZER] } });
            convs.patch(&name, &pp, &Patch::Merge(&patch)).await?;
            Ok(Action::requeue(Duration::from_secs(1)))
        }
        ReconcileAction::CreateSandboxClaim {
            claim_name,
            template,
        } => {
            // Logged here (not only in the backend) so the decision to create
            // is traceable even if a future backend's own `ensure` doesn't log
            // — this arm alone answers "was a SandboxClaim decided-on for
            // conversation X, and when" from logs.
            tracing::info!(
                conversation = %name,
                claim = %claim_name,
                template = %template,
                "creating sandbox claim for conversation"
            );
            ctx.backend
                .ensure(&conv, &claim_name, &template, &ns)
                .await?;
            let conditions = conditions_triad(
                &conv
                    .status
                    .as_ref()
                    .map(|s| s.conditions.clone())
                    .unwrap_or_default(),
                false,
                true,
                false,
                "CreatingExecutionUnit",
                "waiting for the execution unit to become ready",
                &now_rfc3339(),
            );
            let status = json!({
                "status": {
                    "sandboxClaimName": claim_name,
                    "phase": PHASE_PENDING,
                    "conditions": conditions,
                }
            });
            convs
                .patch_status(&name, &PatchParams::default(), &Patch::Merge(&status))
                .await?;
            Ok(Action::requeue(PENDING_POLL))
        }
        ReconcileAction::Cleanup {
            claim_name,
            remove_finalizer,
        } => {
            if let Some(claim) = claim_name {
                // Normally a teardown failure propagates (`?`) to requeue and
                // retry — we keep retrying delete until it (or NOT_FOUND) wins.
                // But a unit that *never* tears down must not wedge the
                // Conversation in `Terminating` forever. Once deletion has been
                // pending past [`TEARDOWN_GRACE`],
                // force the finalizer off and accept a possibly-leaked unit
                // (logged loudly for an operator to reconcile) — the lesser evil
                // than an undeletable CR.
                if let Err(e) = ctx.backend.teardown(&conv, &claim, &ns).await {
                    if deletion_age(&conv) >= TEARDOWN_GRACE {
                        tracing::error!(
                            claim = %claim,
                            error = %e,
                            grace_secs = TEARDOWN_GRACE.as_secs(),
                            "teardown still failing past grace; force-removing finalizer — \
                             execution unit may be leaked, operator should verify"
                        );
                        let patch = json!({ "metadata": { "finalizers": [] } });
                        convs.patch(&name, &pp, &Patch::Merge(&patch)).await?;
                        return Ok(Action::await_change());
                    }
                    return Err(e);
                }
            }
            if remove_finalizer {
                let patch = json!({ "metadata": { "finalizers": [] } });
                convs.patch(&name, &pp, &Patch::Merge(&patch)).await?;
                Ok(Action::await_change())
            } else {
                // Tearing the harness down MUST also clear the readiness
                // pointers: the pod behind `podIp` is gone, so leaving
                // `harnessReady: true` makes a resumed (re-opened) conversation
                // advertise a ready harness at a dead IP — the control plane
                // then dials the dead pod and the turn dies on the harness
                // deadline. Clearing them forces the resume to wait for a fresh
                // claim (re-created by the CreateSandboxClaim path once `closed`
                // flips back to false) before dialing.
                //
                // This branch (remove_finalizer: false) is reached only via
                // `plan`'s `closed` path, so this is the exact moment the
                // conversation finishes closing — stamp `closedAt` here so the
                // closed-conversation GC (`DeleteConversation`) has a real
                // clock to measure retention from.
                let current = conv.status.clone().unwrap_or_default();
                let conditions = conditions_triad(
                    &current.conditions,
                    false,
                    true,
                    false,
                    "TearingDown",
                    "deleting the execution unit",
                    &now_rfc3339(),
                );
                let patch = json!({ "status": {
                    "sandboxClaimName": null,
                    "podIp": null,
                    "harnessReady": false,
                    "phase": "Closing",
                    "closedAt": now_unix,
                    "conditions": conditions,
                } });
                convs
                    .patch_status(&name, &PatchParams::default(), &Patch::Merge(&patch))
                    .await?;
                Ok(Action::requeue(Duration::from_secs(5)))
            }
        }
        ReconcileAction::SyncStatus { claim_name } => {
            let current = conv.status.clone().unwrap_or_default();

            // Issue #802: `sync_execution_unit` reads readiness AND — unless
            // the unit vanished — re-applies the desired spec each pass
            // (drift healing: an out-of-band SandboxClaim spec edit is healed
            // on the very next reconcile). See its doc comment for why the
            // ordering (read-then-conditionally-heal) matters for the
            // self-heal path below.
            let outcome = sync_execution_unit(
                ctx.backend.as_ref(),
                &conv,
                &claim_name,
                &ctx.default_template,
                &ns,
            )
            .await?;

            let readiness = match outcome {
                // Issue #2 self-heal: a unit that was once healthy now reads
                // as empty ⇒ it vanished (GC'd / api-server restart / deleted
                // out-of-band). Clear `sandboxClaimName` so the NEXT plan()
                // pass re-enters CreateSandboxClaim and re-creates it. We
                // requeue promptly (1 s) so the re-create happens without
                // waiting the steady poll. See `unit_is_gone` for the
                // gone-vs-starting heuristic.
                SyncOutcome::Vanished => {
                    let conditions = conditions_triad(
                        &current.conditions,
                        false,
                        true,
                        true,
                        "ExecutionUnitVanished",
                        "the execution unit disappeared out-of-band; recreating it",
                        &now_rfc3339(),
                    );
                    let patch = json!({ "status": {
                        "sandboxClaimName": null,
                        "podIp": null,
                        "harnessEndpoint": null,
                        "harnessReady": false,
                        "phase": PHASE_PENDING,
                        "conditions": conditions,
                    } });
                    convs
                        .patch_status(&name, &PatchParams::default(), &Patch::Merge(&patch))
                        .await?;
                    tracing::warn!(
                        claim = %claim_name,
                        "execution unit vanished; cleared claim name to trigger re-create"
                    );
                    return Ok(Action::requeue(Duration::from_secs(1)));
                }
                SyncOutcome::Synced(readiness) => readiness,
            };

            // Project the backend's single dial address onto the two persisted
            // status fields (exactly one is `Some`).
            let (pod_ip, harness_endpoint) = readiness
                .address
                .as_ref()
                .map_or((None, None), DialAddress::to_status_fields);
            // Desired display phase. While `idle_reclaimed` is set, show
            // `Paused` regardless of readiness; once a turn clears the flag,
            // fall back to readiness. No code path sets this field true
            // anymore (see `PHASE_PAUSED`'s doc comment) — this only matters
            // for an existing object that already carries it.
            let phase = if current.idle_reclaimed {
                PHASE_PAUSED
            } else if readiness.harness_ready {
                PHASE_READY
            } else {
                PHASE_PENDING
            };
            // The None-vs-Some adopt path (see `plan`'s doc comment):
            // `plan` only ever routes here with a recorded generation that's
            // either absent or already equal to the desired one — a genuine
            // mismatch instead returns `RollHarness`. So the only stamping
            // this arm ever needs to do is fill in a still-`None` recorded
            // generation once a desired one is known — this is BOTH the
            // first-time-enable adopt and the post-`RollHarness` re-stamp
            // (the same path, reused). `Some(_)` only when there's actually
            // something to write, so the patch below never emits a stray
            // `null` that would clobber an already-correct stamp.
            let generation_to_stamp = ctx
                .desired_harness_generation
                .as_deref()
                .filter(|_| current.harness_image_generation.is_none());
            // Resynced back to Ready: clear the roll-in-flight marker the
            // `RollHarness` arm set, so the concurrency cap stops counting
            // this conversation as mid-roll. Reusing `current.rolling_harness`
            // as the trigger (rather than gating on `readiness.harness_ready`
            // alone) keeps this idempotent — once cleared, re-observing
            // Ready again is a no-op.
            let clear_rolling = current.rolling_harness && readiness.harness_ready;
            // Only patch when something actually changed, to avoid a
            // status-write → watch → reconcile churn loop.
            if current.pod_ip != pod_ip
                || current.harness_endpoint != harness_endpoint
                || current.harness_ready != readiness.harness_ready
                || current.phase.as_deref() != Some(phase)
                || generation_to_stamp.is_some()
                || clear_rolling
            {
                let (ready, progressing, reason, message) = if current.idle_reclaimed {
                    (
                        false,
                        false,
                        "IdleReclaimed",
                        "the execution unit's worker was reclaimed while idle; it resumes on the next turn",
                    )
                } else if readiness.harness_ready {
                    (true, false, "HarnessReady", "the harness is dialable")
                } else {
                    (
                        false,
                        true,
                        "WaitingForHarness",
                        "waiting for the harness to become dialable",
                    )
                };
                let conditions = conditions_triad(
                    &current.conditions,
                    ready,
                    progressing,
                    false,
                    reason,
                    message,
                    &now_rfc3339(),
                );
                let mut patch = json!({ "status": {
                    "podIp": pod_ip,
                    "harnessEndpoint": harness_endpoint,
                    "harnessReady": readiness.harness_ready,
                    "phase": phase,
                    "conditions": conditions,
                } });
                if let Some(desired) = generation_to_stamp {
                    patch["status"]["harnessImageGeneration"] = json!(desired);
                }
                if clear_rolling {
                    patch["status"]["rollingHarness"] = json!(false);
                }
                convs
                    .patch_status(&name, &PatchParams::default(), &Patch::Merge(&patch))
                    .await?;
                tracing::info!(
                    pod_ip = ?pod_ip,
                    harness_endpoint = ?harness_endpoint,
                    harness_ready = readiness.harness_ready,
                    generation_stamped = ?generation_to_stamp,
                    "synced conversation status from execution unit"
                );
            }
            // Steady-state poll cadence. While a unit is still `Pending` every
            // backend polls tightly; once `Ready` the cadence is the backend's
            // own — agent-sandbox stretches it out, since its
            // `.owns(SandboxClaim)` watch reports changes reactively (out-of-band
            // loss still surfaces via the `unit_is_gone` self-heal above), so the
            // poll is just a backstop.
            let next = if readiness.harness_ready {
                ctx.backend.ready_poll_interval()
            } else {
                PENDING_POLL
            };
            Ok(Action::requeue(next))
        }
        ReconcileAction::RollHarness { claim_name } => {
            // Concurrency cap (issue #117 phase 2, `docs/reference/upgrade-delivery.md`
            // §8.1): `plan` is pure and has no cluster-wide view, so the cap is
            // enforced here against reconciler-visible status — see
            // `Context::count_in_flight_rolls` for the cost model. At/over cap,
            // defer: no teardown call, just a short requeue so a freed slot is
            // picked up promptly without hammering the list-count cache.
            let in_flight = ctx.count_in_flight_rolls(&ns).await?;
            if !roll_allowed(in_flight, ctx.harness_roll_concurrency) {
                tracing::info!(
                    claim = %claim_name,
                    in_flight,
                    cap = ctx.harness_roll_concurrency,
                    "deferring harness roll: concurrency cap reached"
                );
                return Ok(Action::requeue(ROLL_DEFER_POLL));
            }
            // Same teardown call `Cleanup` already uses; the next
            // `CreateSandboxClaim` pass recreates the claim at the new image
            // (see `ReconcileAction::RollHarness`'s doc comment for the full
            // mechanism, including the turn-interruption tradeoff).
            ctx.backend.teardown(&conv, &claim_name, &ns).await?;
            let current = conv.status.clone().unwrap_or_default();
            let conditions = conditions_triad(
                &current.conditions,
                false,
                true,
                false,
                "RollingHarnessImage",
                "tearing the execution unit down to roll it onto a newer harness image",
                &now_rfc3339(),
            );
            let patch = json!({ "status": {
                "sandboxClaimName": null,
                "podIp": null,
                "harnessEndpoint": null,
                "harnessReady": false,
                "harnessImageGeneration": null,
                "phase": PHASE_ROLLING_HARNESS,
                "rollingHarness": true,
                "conditions": conditions,
            } });
            convs
                .patch_status(&name, &PatchParams::default(), &Patch::Merge(&patch))
                .await?;
            tracing::info!(
                claim = %claim_name,
                "tore down execution unit to roll it onto a new harness image"
            );
            Ok(Action::requeue(ROLL_REQUEUE))
        }
        ReconcileAction::CloseIdle => {
            // Idempotent: set closed=true (a bool, so no status-write churn).
            // The next reconcile sees `closed` and runs the Cleanup path that
            // tears down the unit + harness pod.
            let current = conv.status.clone().unwrap_or_default();
            let conditions = conditions_triad(
                &current.conditions,
                false,
                true,
                false,
                "ClosingIdleConversation",
                "idle past the timeout; tearing the execution unit down",
                &now_rfc3339(),
            );
            let patch = json!({ "status": {
                "closed": true,
                "phase": "Closing",
                "conditions": conditions,
            } });
            convs
                .patch_status(&name, &PatchParams::default(), &Patch::Merge(&patch))
                .await?;
            tracing::info!(conversation = %name, "closing idle conversation (idle GC)");
            Ok(Action::requeue(Duration::from_secs(5)))
        }
        ReconcileAction::DeleteConversation => {
            // Sets `deletionTimestamp`; the finalizer already present on this
            // object means the very next `plan` pass re-enters the
            // `being_deleted` branch and clears it via a no-op
            // `Cleanup{claim_name: None, remove_finalizer: true}` (no claim
            // left to tear down).
            //
            // Guarded by a resourceVersion precondition: `plan` decided to
            // delete from a snapshot that may be stale by the time this call
            // lands, and Kubernetes has no "undelete" once `deletionTimestamp`
            // is set — a resume racing in between (`kube_harness::
            // stamp_activity` flipping `closed` back to `false`) would
            // otherwise be silently discarded by an in-flight deletion the
            // reconciler can no longer stop. The precondition turns the
            // delete into a compare-and-swap on the exact object this pass
            // observed: if anything touched it since (a resume, an
            // idle-reclaim, a manual edit), the API server refuses with
            // `Conflict` instead of deleting it.
            let dp = DeleteParams {
                preconditions: Some(Preconditions {
                    resource_version: conv.resource_version(),
                    uid: conv.uid(),
                }),
                ..DeleteParams::default()
            };
            match convs.delete(&name, &dp).await {
                Ok(_) => {
                    tracing::info!(
                        conversation = %name,
                        "deleting fully-closed conversation past its retention window"
                    );
                    Ok(Action::await_change())
                }
                // A benign race, not a failure: something changed the object
                // since `plan` decided to delete it. Re-evaluate fresh on the
                // next pass rather than treating this as an error to retry
                // with backoff.
                Err(kube::Error::Api(e)) if e.code == 409 => {
                    tracing::info!(
                        conversation = %name,
                        "conversation changed since it was planned for deletion \
                         (likely resumed); skipping this pass"
                    );
                    Ok(Action::requeue(Duration::from_secs(5)))
                }
                Err(e) => Err(Error::Kube(e)),
            }
        }
        ReconcileAction::Noop => Ok(Action::requeue(Duration::from_mins(5))),
    }
}

/// Drop a `404 Not Found` on delete (the unit is already gone); propagate every
/// other error. Shared by the reconcile cleanup path and `SandboxClaimBackend`.
pub(crate) fn ignore_not_found<T>(res: Result<T, kube::Error>) -> Result<(), Error> {
    match res {
        Ok(_) => Ok(()),
        Err(kube::Error::Api(e)) if e.code == 404 => Ok(()),
        Err(e) => Err(Error::Kube(e)),
    }
}

/// Lower bound of the failure-requeue window (issue #8).
const ERROR_BACKOFF_MIN_SECS: u64 = 10;
/// Upper bound (ceiling) of the failure-requeue window (issue #8).
const ERROR_BACKOFF_MAX_SECS: u64 = 60;

/// Spread a conversation's failure requeue across `[MIN, MAX]` seconds using a
/// cheap, deterministic hash of its name. Pure.
///
/// This is the de-lockstep half of issue #8's fix: a persistent backend outage
/// (ate-api-server down, missing CA) fails *every* conversation's reconcile,
/// and a fixed interval would have them all retry in the same instant, hammering
/// the backend in a synchronised thundering herd. Deriving the delay from the
/// name jitters each conversation to a different point in the window, so retries
/// arrive spread out rather than as one synchronised burst.
fn error_backoff(name: &str) -> Duration {
    crate::jitter::within_window(name, ERROR_BACKOFF_MIN_SECS, ERROR_BACKOFF_MAX_SECS)
}

/// Render `err` together with its full `source()` chain, one `; caused by:`
/// segment per link.
///
/// kube-runtime's `Controller::run` stream surfaces a stalled watch (e.g. the
/// shared client's read timeout firing on an idle long-lived watch) as
/// `Error::QueueError`, whose derived `Display` is the fixed literal `"event
/// queue error"` — the actual cause (a specific hyper/reqwest timeout, a `410
/// Gone`, ...) is attached via `#[source]` but never appears in `%err` alone.
/// Walking the chain here is what makes that cause visible in logs.
fn error_chain(err: &(dyn std::error::Error + 'static)) -> String {
    let mut msg = err.to_string();
    let mut cause = err.source();
    while let Some(e) = cause {
        write!(msg, "; caused by: {e}").expect("String write is infallible");
        cause = e.source();
    }
    msg
}

/// Requeue policy on reconcile failure (issue #8).
///
/// `error_policy` is a pure fn with no per-object attempt counter to thread an
/// exponential backoff through, so a true per-conversation exponential would
/// require plumbing state (a shared attempt map) that controller-runtime
/// doesn't hand us here. Instead we widen the fixed 10s into a jittered
/// `[10s, 60s]` window keyed on the conversation name (see `error_backoff`):
/// this bounds the retry rate per object and, crucially, breaks the lockstep so
/// a cluster-wide backend outage no longer produces a synchronised retry storm.
/// (controller-runtime's own default error backoff is exponential; this keeps
/// the spirit — a bounded, de-synchronised retry — without the plumbing.)
// Signature is fixed by `Controller::run`'s error-policy contract (owned
// `Arc<Conversation>`), so we can't take `conv` by reference here.
#[allow(clippy::needless_pass_by_value)]
#[must_use]
pub fn error_policy(conv: Arc<Conversation>, err: &Error, _ctx: Arc<Context>) -> Action {
    let delay = error_backoff(&conv.name_any());
    tracing::warn!(error = %err, requeue_secs = delay.as_secs(), "reconcile failed; requeuing");
    Action::requeue(delay)
}

/// Run the control-plane controllers until their watch streams end. Blocks for
/// the process lifetime in a real deployment.
///
/// This drives **both** reconcilers concurrently: the `Conversation` reconciler
/// and the `ToolService` health-check reconciler
/// ([`crate::toolservice_reconcile`]). The control-plane binary calls this once.
///
/// # Scoping
///
/// The watchers are intentionally namespaced (`Api::namespaced`), not
/// cluster-wide (`Api::all`). Two reasons:
///
/// 1. **Principle of least privilege.** The per-reconcile API calls
///    already use `Api::namespaced(ctx.client, &ns)` against the
///    `Conversation`'s own namespace, so cluster-wide visibility was
///    never used. Tightening the watcher matches the actual blast
///    radius and lets the polychrome SA bind to a namespaced `Role`
///    instead of a `ClusterRole`.
///
/// 2. **Polychrome's single-namespace design.** Every resource the
///    manifests provision lives in the `polychrome` namespace (set at
///    the kustomize layer). Cross-namespace `Conversation` CRs aren't
///    a supported deployment shape today; if multi-namespace ever lands
///    it'll be a deliberate change with the matching `ClusterRole`.
///
/// `namespace` is the namespace the control plane operates in, resolved by
/// the binary's own config (explicit config → `POLYCHROME_NAMESPACE` →
/// Downward-API namespace → `polychrome`) — *not* the kube client default.
/// We deliberately avoid `Api::default_namespaced`, whose namespace comes
/// from the client config: in-cluster that's the SA namespace (correct), but
/// out-of-cluster it's the dev's kubeconfig *context* namespace (e.g.
/// `default`), which would silently watch the wrong namespace and never see
/// the `Conversation` CRs that `polychrome up` creates in `polychrome`. The
/// caller resolves and supplies the operating namespace so the namespace we
/// WATCH always equals the one Conversations are deployed into.
///
/// `client` and `watch_client` are two distinct clients the control-plane
/// binary builds from one inferred `kube::Config` (see its
/// `derive_kube_configs`): `client` keeps the tight #785 read-timeout bound
/// and backs every reconcile-time get/list/patch call across all five
/// controllers; `watch_client` has no read timeout and backs every
/// controller's long-lived watch/owned-watch stream — the Conversation
/// reconciler's (the internal `run_conversation` helper below) plus
/// `ToolService`, Routine, `ServiceDefinition`, and Workflow's
/// (`crate::toolservice_reconcile::run_toolservice`,
/// `crate::routine_reconcile::run_routine`,
/// `crate::servicedefinition_reconcile::run_servicedefinition`,
/// `crate::workflow_reconcile::run_workflow`). Sharing the tight-timeout
/// client with a watch stream was the #1002/#1003 incident (an idle watch
/// spuriously errored and paused watch-triggered reconcile dispatch); #1004
/// finished migrating the four reconcilers #1003 left on `client`. Each
/// reconciler's `Context` (and every reconcile-time get/list/patch call it
/// makes) still rides `client`, unchanged — only the watch sockets moved.
///
/// # Errors
///
/// Returns [`Error`] only on fatal setup failure; per-item errors go through
/// [`error_policy`].
// polychrome-work-family: kubernetes-reconcilers
pub async fn run(
    client: Client,
    watch_client: Client,
    namespace: &str,
    apps_namespace: &str,
) -> Result<(), Error> {
    // Five independent controllers share this process: the Conversation
    // reconciler (below), the ToolService health-check reconciler
    // ([`crate::toolservice_reconcile::run_toolservice`]), the Routine
    // definition-validation reconciler
    // ([`crate::routine_reconcile::run_routine`]), the ServiceDefinition
    // fan-out reconciler
    // ([`crate::servicedefinition_reconcile::run_servicedefinition`]), and the
    // Workflow pipeline reconciler
    // ([`crate::workflow_reconcile::run_workflow`]). They watch different CRs
    // and never touch each other's state, so we drive all of them concurrently
    // on this task with `tokio::join!`. Keeping the join here — rather than
    // asking the binary to spawn more futures — means the single
    // `reconcile::run` call site in the control-plane binary launches every
    // watch with no change to the binary.
    //
    // The ServiceDefinition and Workflow watches run over `apps_namespace`, NOT
    // the control namespace: agent-scaffolded workloads (and the pipelines that
    // compose them) live in their own namespace with their own quota/RBAC blast
    // radius, away from the control plane. The Workflow reconciler emits
    // ServiceDefinitions the ServiceDefinition reconciler then fans out, so the
    // two cooperate through the CR rather than touching each other directly.
    // `Routine` is a catalog entry like `Agent`/`ToolService` (operator-
    // published, not agent-scaffolded), so it watches the control `namespace`
    // — the same namespace the control plane's `resolve_routine` reads from.
    let conversations = run_conversation(client.clone(), watch_client.clone(), namespace);
    let toolservices = crate::toolservice_reconcile::run_toolservice(
        client.clone(),
        watch_client.clone(),
        namespace,
    );
    let routines =
        crate::routine_reconcile::run_routine(client.clone(), watch_client.clone(), namespace);
    let servicedefinitions = crate::servicedefinition_reconcile::run_servicedefinition(
        client.clone(),
        watch_client.clone(),
        apps_namespace,
    );
    let workflows = crate::workflow_reconcile::run_workflow(client, watch_client, apps_namespace);
    let (conv_res, ts_res, rtn_res, sd_res, wf_res) = tokio::join!(
        conversations,
        toolservices,
        routines,
        servicedefinitions,
        workflows
    );
    conv_res?;
    ts_res?;
    rtn_res?;
    sd_res?;
    wf_res?;
    Ok(())
}

/// Run the `Conversation` controller until its watch streams end.
///
/// `client` is the unary (#785-bounded) client — every reconcile-time
/// get/list/patch call, via `Context`, rides it. `watch_client` has no read
/// timeout and backs only the watch/owned-watch `Api` handles passed to
/// `Controller::new(...).owns(...)` below (see [`run`]'s doc comment for
/// why the two must not cross).
///
/// # Errors
///
/// Returns [`Error`] only on fatal setup failure; per-item errors go through
/// [`error_policy`].
async fn run_conversation(
    client: Client,
    watch_client: Client,
    namespace: &str,
) -> Result<(), Error> {
    let convs: Api<Conversation> = Api::namespaced(watch_client.clone(), namespace);

    // Read the harness-roll knobs once at startup, directly from the
    // environment, rather than threading them through the control-plane
    // binary's layered `figment` `Config` (that pipeline is deliberately not
    // extended for every knob; this one is reconciler-internal, not
    // user-facing config).
    //
    // `POLYCHROME_HARNESS_IMAGE_GENERATION` unset/empty ⇒ `None` ⇒ the
    // roll-on-image-change feature is OFF (see `Context::desired_harness_generation`).
    let desired_harness_generation = std::env::var("POLYCHROME_HARNESS_IMAGE_GENERATION")
        .ok()
        .filter(|v| !v.trim().is_empty());
    // `POLYCHROME_HARNESS_ROLL_CONCURRENCY` unset, empty, `0`, or unparseable
    // ⇒ `DEFAULT_HARNESS_ROLL_CONCURRENCY`.
    let harness_roll_concurrency = std::env::var("POLYCHROME_HARNESS_ROLL_CONCURRENCY")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .filter(|&n| n > 0)
        .unwrap_or(DEFAULT_HARNESS_ROLL_CONCURRENCY);
    let closed_conversation_retention_seconds = parse_closed_conversation_retention_seconds(
        std::env::var("POLYCHROME_CLOSED_CONVERSATION_RETENTION_SECONDS")
            .ok()
            .as_deref(),
    );
    tracing::info!(
        desired_harness_generation = ?desired_harness_generation,
        harness_roll_concurrency,
        closed_conversation_retention_seconds = ?closed_conversation_retention_seconds,
        "harness image-generation roll + closed-conversation GC configured"
    );

    // Build the reconcile context + the owned `SandboxClaim` watch.
    let claims: Api<SandboxClaim> = Api::namespaced(watch_client, namespace);
    let ctx = Arc::new(Context::new(
        client.clone(),
        DEFAULT_TEMPLATE.to_owned(),
        Arc::new(SandboxClaimBackend::new(client)),
        desired_harness_generation,
        harness_roll_concurrency,
        closed_conversation_retention_seconds,
    ));
    let controller =
        Controller::new(convs, watcher::Config::default()).owns(claims, watcher::Config::default());

    controller
        .run(reconcile, error_policy, ctx)
        .for_each(|res| async move {
            if let Err(e) = res {
                // `%e` alone would only ever print kube-runtime's outer
                // `Display` (e.g. the fixed "event queue error" literal for a
                // stalled watch stream) — `error_chain` walks `#[source]` so
                // the real underlying cause is what lands in the logs.
                tracing::warn!(error = %error_chain(&e), "reconcile stream item errored");
            }
        })
        .await;
    Ok(())
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]

    use kube::api::ObjectMeta;

    use super::*;
    use crate::conversation::{ConversationSpec, ConversationStatus};

    const TPL: &str = "polychrome-harness-default";

    fn conv(name: &str) -> Conversation {
        Conversation::new(
            name,
            ConversationSpec {
                model: "fast-2".to_owned(),
                principal_ref: "persona-test-1".to_owned(),
                idle_timeout_seconds: 300,
                tools_enabled: vec![],
                tools_disabled: vec![],
                parent_conversation_id: None,
                agent_id: None,
            },
        )
    }

    #[derive(Debug, thiserror::Error)]
    #[error("root cause")]
    struct RootCause;

    #[derive(Debug, thiserror::Error)]
    #[error("middle layer")]
    struct MiddleLayer(#[source] RootCause);

    #[derive(Debug, thiserror::Error)]
    #[error("event queue error")]
    struct OuterQueueError(#[source] MiddleLayer);

    #[test]
    fn error_chain_surfaces_every_source_not_just_the_outer_display() {
        // Regression guard for the gap this fixes: kube-runtime's
        // `Error::QueueError`'s derived `Display` is a fixed, useless literal
        // ("event queue error") that never changes regardless of the real
        // underlying cause — `%err` alone would print only that literal. This
        // constructs an analogous three-deep chain and asserts every link's
        // message, not just the outermost one, appears in the rendered string.
        let err = OuterQueueError(MiddleLayer(RootCause));
        let rendered = error_chain(&err);
        assert!(rendered.contains("event queue error"));
        assert!(rendered.contains("middle layer"));
        assert!(rendered.contains("root cause"));
    }

    #[test]
    fn fresh_conversation_gets_a_finalizer_first() {
        assert_eq!(
            plan(&conv("c1"), TPL, 0, None, None),
            ReconcileAction::AddFinalizer
        );
    }

    #[test]
    fn with_finalizer_and_no_claim_creates_one() {
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        assert_eq!(
            plan(&c, TPL, 0, None, None),
            ReconcileAction::CreateSandboxClaim {
                claim_name: "c1".to_owned(),
                template: TPL.to_owned(),
            }
        );
    }

    #[test]
    fn with_claim_recorded_syncs_status() {
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, None, None),
            ReconcileAction::SyncStatus {
                claim_name: "c1".to_owned(),
            }
        );
    }

    /// `desired_harness_generation: None` (the operator has never set
    /// `POLYCHROME_HARNESS_IMAGE_GENERATION`) means the roll-on-image-change
    /// feature is off entirely — `plan` must always return `SyncStatus`, even
    /// when a recorded generation exists and differs from nothing in
    /// particular. Regression guard: this must never emit `RollHarness`.
    #[test]
    fn desired_generation_none_never_rolls() {
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            harness_image_generation: Some("2026.6.0".to_owned()),
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, None, None),
            ReconcileAction::SyncStatus {
                claim_name: "c1".to_owned(),
            }
        );
    }

    /// THE case the adversarial review named as the easiest to get wrong:
    /// a conversation with a claim already recorded, but no
    /// `harness_image_generation` ever stamped (predates the feature, or
    /// predates the operator ever setting the env var) must be ADOPTED, not
    /// rolled, the first time a desired generation becomes known. There is
    /// nothing to roll *from* — only a gap to fill in. `plan` must return
    /// `SyncStatus` (never `RollHarness`); the actual stamping happens in the
    /// `SyncStatus` reconcile arm (see `sync_status_adopts_a_never_recorded_generation_without_rolling`).
    #[test]
    fn recorded_none_desired_some_adopts_without_rolling() {
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            harness_image_generation: None,
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, Some("2026.7.0"), None),
            ReconcileAction::SyncStatus {
                claim_name: "c1".to_owned(),
            },
            "a never-recorded generation must adopt via SyncStatus, never RollHarness"
        );
    }

    /// Steady state: the recorded generation already matches the desired
    /// one. No roll — `SyncStatus` (also stamp-idempotent, no-op patch).
    #[test]
    fn recorded_equals_desired_stays_steady_state() {
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            harness_image_generation: Some("2026.7.0".to_owned()),
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, Some("2026.7.0"), None),
            ReconcileAction::SyncStatus {
                claim_name: "c1".to_owned(),
            }
        );
    }

    /// The only case that actually rolls: a recorded generation that
    /// disagrees with the desired one.
    #[test]
    fn recorded_differs_from_desired_rolls() {
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            harness_image_generation: Some("2026.6.0".to_owned()),
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, Some("2026.7.0"), None),
            ReconcileAction::RollHarness {
                claim_name: "c1".to_owned(),
            }
        );
    }

    #[test]
    fn idle_conversation_past_timeout_is_closed() {
        // conv() sets idle_timeout_seconds = 300.
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            last_activity_unix: Some(1_000),
            ..Default::default()
        });
        // 301s since last turn (> 300) → reap.
        assert_eq!(
            plan(&c, TPL, 1_000 + 301, None, None),
            ReconcileAction::CloseIdle
        );
        // 200s since last turn (< 300) → still active, normal sync.
        assert_eq!(
            plan(&c, TPL, 1_000 + 200, None, None),
            ReconcileAction::SyncStatus {
                claim_name: "c1".to_owned(),
            }
        );
    }

    #[test]
    fn no_activity_stamp_is_never_closed() {
        // Regression guard: a conversation with NO last_activity_unix must never
        // be reaped, even at an absurdly large `now`. The removed creationTimestamp
        // fallback is what closed old-but-active conversations whose stamp hadn't
        // landed yet — there must be no proxy for a real activity stamp.
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            last_activity_unix: None,
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 10_000_000_000, None, None),
            ReconcileAction::SyncStatus {
                claim_name: "c1".to_owned(),
            }
        );
    }

    #[test]
    fn closed_with_claim_cleans_up_but_keeps_finalizer() {
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            closed: true,
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, None, None),
            ReconcileAction::Cleanup {
                claim_name: Some("c1".to_owned()),
                remove_finalizer: false,
            }
        );
    }

    #[test]
    fn closed_after_claim_cleared_settles_to_noop() {
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            closed: true,
            ..Default::default()
        });
        assert_eq!(plan(&c, TPL, 0, None, None), ReconcileAction::Noop);
    }

    #[test]
    fn closed_with_no_closed_at_stamp_never_gcs_even_with_retention_configured() {
        // A CR that predates `closedAt` (or a hypothetical future close path
        // that never stamps it) must not be deleted just because retention is
        // configured — no stamp, no GC, mirroring the idle reaper's own rule.
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            closed: true,
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 10_000_000, None, Some(60)),
            ReconcileAction::Noop
        );
    }

    #[test]
    fn closed_within_retention_window_stays_noop() {
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            closed: true,
            closed_at: Some(1_000),
            ..Default::default()
        });
        // 1_000 + 60 (retention) = 1_060; now is still inside the window.
        assert_eq!(plan(&c, TPL, 1_050, None, Some(60)), ReconcileAction::Noop);
    }

    #[test]
    fn closed_exactly_at_retention_boundary_stays_noop() {
        // `plan` uses a strict `>` so a conversation exactly at the boundary
        // is not yet deleted — the next requeue (5 minutes later, per the
        // `Noop` requeue cadence) will cross it.
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            closed: true,
            closed_at: Some(1_000),
            ..Default::default()
        });
        assert_eq!(plan(&c, TPL, 1_060, None, Some(60)), ReconcileAction::Noop);
    }

    #[test]
    fn closed_past_retention_window_deletes_the_conversation() {
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            closed: true,
            closed_at: Some(1_000),
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 1_061, None, Some(60)),
            ReconcileAction::DeleteConversation
        );
    }

    #[test]
    fn closed_past_retention_window_stays_noop_when_gc_disabled() {
        // `retention_seconds: None` disables the GC regardless of how long a
        // conversation has been closed — the pre-fix behavior, preserved as
        // an explicit operator opt-out.
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            closed: true,
            closed_at: Some(1_000),
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 100_000_000, None, None),
            ReconcileAction::Noop
        );
    }

    #[test]
    fn closed_with_claim_still_present_ignores_retention() {
        // A claim still needing teardown takes priority over the retention
        // check — `Cleanup` must run (and stamp `closedAt`) before there is
        // anything to measure retention from.
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            closed: true,
            closed_at: Some(1_000),
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 100_000_000, None, Some(60)),
            ReconcileAction::Cleanup {
                claim_name: Some("c1".to_owned()),
                remove_finalizer: false,
            }
        );
    }

    #[test]
    fn retention_env_unset_defaults_to_enabled() {
        assert_eq!(
            parse_closed_conversation_retention_seconds(None),
            Some(DEFAULT_CLOSED_CONVERSATION_RETENTION_SECS)
        );
    }

    #[test]
    fn retention_env_empty_or_zero_disables() {
        assert_eq!(parse_closed_conversation_retention_seconds(Some("")), None);
        assert_eq!(
            parse_closed_conversation_retention_seconds(Some("   ")),
            None
        );
        assert_eq!(parse_closed_conversation_retention_seconds(Some("0")), None);
    }

    #[test]
    fn retention_env_positive_integer_is_used_verbatim() {
        assert_eq!(
            parse_closed_conversation_retention_seconds(Some("3600")),
            Some(3_600)
        );
    }

    #[test]
    fn retention_env_negative_or_malformed_fails_safe_to_disabled() {
        // The gated action is destructive (`DeleteConversation`), so garbage
        // input must never silently fall back to the default-enabled
        // behavior — a typo should turn the GC off, not on.
        assert_eq!(
            parse_closed_conversation_retention_seconds(Some("-1")),
            None
        );
        assert_eq!(
            parse_closed_conversation_retention_seconds(Some("off")),
            None
        );
        assert_eq!(
            parse_closed_conversation_retention_seconds(Some("disabled")),
            None
        );
    }

    #[test]
    fn deleting_with_finalizer_cleans_up_and_removes_finalizer() {
        let mut c = conv("c1");
        c.metadata = ObjectMeta {
            name: Some("c1".to_owned()),
            finalizers: Some(vec![FINALIZER.to_owned()]),
            deletion_timestamp: Some(k8s_openapi::apimachinery::pkg::apis::meta::v1::Time(
                "2026-05-27T00:00:00Z".parse().unwrap(),
            )),
            ..Default::default()
        };
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, None, None),
            ReconcileAction::Cleanup {
                claim_name: Some("c1".to_owned()),
                remove_finalizer: true,
            }
        );
    }

    #[test]
    fn child_conversation_plans_same_as_parent() {
        // A child conversation (spawned by a parent's handoff) still flows
        // through the same finalizer → claim → status path. The parent
        // linkage is metadata that the reconcile *effect* (not the pure
        // plan) acts on by labelling the claim.
        let mut c = Conversation::new(
            "child-7",
            ConversationSpec {
                model: "fast-2".to_owned(),
                principal_ref: "persona-test-1".to_owned(),
                idle_timeout_seconds: 300,
                tools_enabled: vec![],
                tools_disabled: vec![],
                parent_conversation_id: Some("parent-1".to_owned()),
                agent_id: Some("researcher".to_owned()),
            },
        );
        // No finalizer yet → the plan adds one.
        assert_eq!(plan(&c, TPL, 0, None, None), ReconcileAction::AddFinalizer);
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        // Once finalised the plan creates a SandboxClaim, same as a parent.
        assert_eq!(
            plan(&c, TPL, 0, None, None),
            ReconcileAction::CreateSandboxClaim {
                claim_name: "child-7".to_owned(),
                template: TPL.to_owned(),
            }
        );
    }

    #[test]
    fn unit_gone_after_previously_ready_triggers_reclaim() {
        // Was Ready, now reads empty ⇒ gone ⇒ clear so plan() re-creates.
        let status = ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            phase: Some("Ready".to_owned()),
            ..Default::default()
        };
        assert!(unit_is_gone(&status, &ClaimReadiness::default()));

        // Simulate the reconcile clearing the claim name: the next plan()
        // (finalizer present, no claim) re-enters CreateSandboxClaim.
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: None,
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, None, None),
            ReconcileAction::CreateSandboxClaim {
                claim_name: "c1".to_owned(),
                template: TPL.to_owned(),
            }
        );
    }

    #[test]
    fn unit_gone_after_prior_address_triggers_reclaim() {
        // A prior pod_ip is also evidence the unit once existed.
        let status = ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            pod_ip: Some("10.4.2.7".to_owned()),
            ..Default::default()
        };
        assert!(unit_is_gone(&status, &ClaimReadiness::default()));

        let status = ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            harness_endpoint: Some("http://router/c1:8080".to_owned()),
            ..Default::default()
        };
        assert!(unit_is_gone(&status, &ClaimReadiness::default()));
    }

    #[test]
    fn still_starting_unit_is_not_treated_as_gone() {
        // Never became ready, no address recorded yet, reads empty ⇒ ambiguous
        // (still booting) ⇒ do NOT clear / re-create (no thrash).
        let status = ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            phase: Some("Pending".to_owned()),
            ..Default::default()
        };
        assert!(!unit_is_gone(&status, &ClaimReadiness::default()));
    }

    #[test]
    fn healthy_unit_read_is_not_gone() {
        // A present unit is never "gone", even if previously Ready.
        let status = ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            phase: Some(PHASE_READY.to_owned()),
            pod_ip: Some("10.4.2.7".to_owned()),
            harness_ready: true,
            ..Default::default()
        };
        let readiness = ClaimReadiness {
            unit_present: true,
            address: Some(DialAddress::PodIp("10.4.2.7".to_owned())),
            harness_ready: true,
        };
        assert!(!unit_is_gone(&status, &readiness));
    }

    #[test]
    fn present_but_transiently_unready_unit_is_not_gone() {
        // Issue #4: on the default agent-sandbox backend a pod restart makes a
        // still-existing claim read present-but-not-ready (no address, not
        // ready). Even though the conversation was previously Ready, the unit
        // object still exists (`unit_present`), so it must NOT be treated as
        // gone — otherwise a routine pod bounce would thrash the claim.
        let status = ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            phase: Some(PHASE_READY.to_owned()),
            pod_ip: Some("10.4.2.7".to_owned()),
            harness_ready: true,
            ..Default::default()
        };
        let restarting = ClaimReadiness {
            unit_present: true,
            address: None,
            harness_ready: false,
        };
        assert!(!unit_is_gone(&status, &restarting));
    }

    #[test]
    fn error_backoff_is_bounded_to_window() {
        for n in 0..1000 {
            let d = error_backoff(&format!("conv-{n}")).as_secs();
            assert!(
                (ERROR_BACKOFF_MIN_SECS..=ERROR_BACKOFF_MAX_SECS).contains(&d),
                "backoff {d}s out of [{ERROR_BACKOFF_MIN_SECS}, {ERROR_BACKOFF_MAX_SECS}]"
            );
        }
    }

    #[test]
    fn error_backoff_is_deterministic_and_desynchronised() {
        // Same name ⇒ same delay (deterministic).
        assert_eq!(error_backoff("c1"), error_backoff("c1"));
        // Different names spread across the window (not all identical).
        let delays: std::collections::HashSet<u64> = (0..200)
            .map(|n| error_backoff(&format!("conv-{n}")).as_secs())
            .collect();
        assert!(delays.len() > 1, "all conversations retried in lockstep");
    }

    #[test]
    fn deleting_without_finalizer_is_noop() {
        let mut c = conv("c1");
        c.metadata.deletion_timestamp = Some(k8s_openapi::apimachinery::pkg::apis::meta::v1::Time(
            "2026-05-27T00:00:00Z".parse().unwrap(),
        ));
        assert_eq!(plan(&c, TPL, 0, None, None), ReconcileAction::Noop);
    }

    /// A mocked [`ExecutionBackend`] that records every `ensure` call, so a
    /// `SyncStatus` pass's IO against the backend can be asserted without a
    /// live kube API server for the `Conversation` object itself (only the
    /// backend seam is exercised — see `sync_execution_unit`).
    #[derive(Default)]
    struct RecordingBackend {
        ensure_calls: std::sync::atomic::AtomicUsize,
        teardown_calls: std::sync::atomic::AtomicUsize,
        readiness: ClaimReadiness,
    }

    #[async_trait::async_trait]
    impl ExecutionBackend for RecordingBackend {
        async fn ensure(
            &self,
            _conv: &Conversation,
            _name: &str,
            _template: &str,
            _ns: &str,
        ) -> Result<(), Error> {
            self.ensure_calls
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Ok(())
        }

        async fn readiness(&self, _name: &str, _ns: &str) -> Result<ClaimReadiness, Error> {
            Ok(self.readiness.clone())
        }

        async fn teardown(
            &self,
            _owner: &Conversation,
            _name: &str,
            _ns: &str,
        ) -> Result<(), Error> {
            self.teardown_calls
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Ok(())
        }

        fn kind(&self) -> &'static str {
            "recording-mock"
        }
    }

    /// Issue #802: a `SandboxClaim` edited out-of-band (e.g. `kubectl edit`)
    /// must be healed on the next reconcile pass. Before this fix, the
    /// `SyncStatus` arm only ever called `readiness()` — a pure read — and
    /// never re-applied the desired spec, so drift persisted until the claim
    /// was deleted and recreated. This fails today because `ensure` is never
    /// invoked from the sync path.
    #[tokio::test]
    async fn sync_status_heals_spec_drift_by_reapplying_desired_spec() {
        let backend = RecordingBackend {
            readiness: ClaimReadiness {
                unit_present: true,
                address: None,
                harness_ready: false,
            },
            ..Default::default()
        };
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            phase: Some(PHASE_PENDING.to_owned()),
            ..Default::default()
        });

        let outcome = sync_execution_unit(&backend, &c, "c1", TPL, "ns")
            .await
            .expect("sync_execution_unit");
        assert!(
            matches!(outcome, SyncOutcome::Synced(_)),
            "unit is present; must not be treated as vanished"
        );
        assert_eq!(
            backend
                .ensure_calls
                .load(std::sync::atomic::Ordering::SeqCst),
            1,
            "SyncStatus must re-apply the desired spec (drift healing), not just read readiness"
        );
    }

    /// Self-heal (issue #2) still takes priority: a genuinely vanished unit
    /// must NOT be resurrected within the same pass — that would mask the
    /// vanish and defeat `unit_is_gone`'s next-pass recreate path.
    #[tokio::test]
    async fn sync_status_does_not_heal_a_vanished_unit() {
        let backend = RecordingBackend::default(); // readiness defaults to unit_present: false
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            phase: Some(PHASE_READY.to_owned()),
            ..Default::default()
        });

        let outcome = sync_execution_unit(&backend, &c, "c1", TPL, "ns")
            .await
            .expect("sync_execution_unit");
        assert!(matches!(outcome, SyncOutcome::Vanished));
        assert_eq!(
            backend
                .ensure_calls
                .load(std::sync::atomic::Ordering::SeqCst),
            0,
            "a vanished unit must not be re-created within the same pass"
        );
    }

    #[test]
    fn conditions_triad_marks_ready_when_ready_true() {
        let conditions = conditions_triad(&[], true, false, false, "HarnessReady", "", "now");
        assert_eq!(conditions.len(), 3);
        let get = |t: &str| conditions.iter().find(|c| c.type_ == t).unwrap().status;
        assert_eq!(get(COND_READY), crate::conversation::ConditionStatus::True);
        assert_eq!(
            get(COND_PROGRESSING),
            crate::conversation::ConditionStatus::False
        );
        assert_eq!(
            get(COND_DEGRADED),
            crate::conversation::ConditionStatus::False
        );
    }

    #[test]
    fn conditions_triad_preserves_transition_time_across_unchanged_passes() {
        let first = conditions_triad(&[], false, true, false, "Provisioning", "", "t0");
        let second = conditions_triad(&first, false, true, false, "Provisioning", "", "t1");
        let progressing = |cs: &[crate::conversation::Condition]| {
            cs.iter()
                .find(|c| c.type_ == COND_PROGRESSING)
                .unwrap()
                .last_transition_time
                .clone()
        };
        assert_eq!(progressing(&first), "t0");
        assert_eq!(
            progressing(&second),
            "t0",
            "status unchanged across passes must not bump lastTransitionTime"
        );

        // Now it actually flips Ready: the Ready condition's transition time
        // advances, Progressing's still doesn't (it also flips, so it does
        // too) — assert the Ready flip specifically.
        let third = conditions_triad(&second, true, false, false, "HarnessReady", "", "t2");
        let ready = third
            .iter()
            .find(|c| c.type_ == COND_READY)
            .unwrap()
            .last_transition_time
            .clone();
        assert_eq!(ready, "t2");
    }

    #[test]
    fn roll_allowed_under_and_at_cap() {
        assert!(roll_allowed(0, 5), "well under cap");
        assert!(roll_allowed(4, 5), "one below cap");
        assert!(!roll_allowed(5, 5), "at cap must defer");
        assert!(!roll_allowed(6, 5), "over cap must defer");
    }

    /// A hand-rolled fake `kube::Client` for `Api<Conversation>`, so the
    /// `RollHarness` concurrency cap and the `SyncStatus` adopt-stamp can be
    /// exercised through the real `reconcile` dispatcher rather than only
    /// through `plan`. Mirrors `polyc_control_plane`'s `lease.rs`
    /// `fakes::fake_client` `tower::service_fn` pattern — no reusable kube-API
    /// mock exists in this crate; the only other option is the live-cluster
    /// `e2e` feature (`tests/e2e.rs`), which is out of scope for a fast unit
    /// test. Serves exactly the two request shapes `reconcile`'s
    /// `Api<Conversation>` calls make: a collection `GET`
    /// (`Context::count_in_flight_rolls`'s `list`, answered from a fixed set
    /// of canned items) and any `PATCH` (a status or metadata patch, recorded
    /// into `patches` and echoed back as a minimally valid `Conversation` so
    /// kube's response deserialization succeeds — the echoed body's content
    /// is never asserted on, only `patches`).
    mod fake_conversations {
        use std::sync::{Arc, Mutex};

        use http::{Method, StatusCode};
        use serde_json::{Value, json};

        pub(super) fn client(items: Vec<Value>, patches: Arc<Mutex<Vec<Value>>>) -> kube::Client {
            let svc = tower::service_fn(move |req: http::Request<kube::client::Body>| {
                let items = items.clone();
                let patches = patches.clone();
                async move {
                    let method = req.method().clone();
                    let path = req.uri().path().to_owned();
                    let body = req
                        .into_body()
                        .collect_bytes()
                        .await
                        .map(|b| b.to_vec())
                        .unwrap_or_default();
                    let (status, payload) = if method == Method::GET {
                        let list = json!({
                            "apiVersion": "polychrome.dev/v1alpha1",
                            "kind": "ConversationList",
                            "items": items,
                        });
                        (StatusCode::OK, serde_json::to_vec(&list).unwrap())
                    } else if method == Method::PATCH {
                        let patch: Value = serde_json::from_slice(&body).unwrap_or_default();
                        patches.lock().expect("patches lock poisoned").push(patch);
                        let name = path
                            .trim_end_matches("/status")
                            .rsplit('/')
                            .next()
                            .unwrap_or("c1");
                        let echoed = json!({
                            "apiVersion": "polychrome.dev/v1alpha1",
                            "kind": "Conversation",
                            "metadata": { "name": name },
                            "spec": { "model": "m", "idleTimeoutSeconds": 300 },
                            "status": {},
                        });
                        (StatusCode::OK, serde_json::to_vec(&echoed).unwrap())
                    } else {
                        (StatusCode::METHOD_NOT_ALLOWED, Vec::new())
                    };
                    let resp = http::Response::builder()
                        .status(status)
                        .header("content-type", "application/json")
                        .body(kube::client::Body::from(payload))
                        .expect("build fake response");
                    Ok::<_, std::convert::Infallible>(resp)
                }
            });
            kube::Client::new(svc, "test-ns")
        }

        /// A canned `Conversation` list item, shaped just enough for
        /// `Context::count_in_flight_rolls`'s filter (`status.rollingHarness`).
        pub(super) fn item(name: &str, rolling_harness: bool) -> Value {
            json!({
                "apiVersion": "polychrome.dev/v1alpha1",
                "kind": "Conversation",
                "metadata": { "name": name },
                "spec": { "model": "m", "idleTimeoutSeconds": 300 },
                "status": { "rollingHarness": rolling_harness },
            })
        }
    }

    /// Build a `Context` wired to a fake `Api<Conversation>` client and a
    /// fresh (unshared) `RollCountCache`, for the `reconcile`-level tests
    /// below.
    fn fake_ctx(
        client: kube::Client,
        backend: Arc<RecordingBackend>,
        desired_harness_generation: Option<&str>,
        harness_roll_concurrency: usize,
    ) -> Arc<Context> {
        Arc::new(Context::new(
            client,
            TPL.to_owned(),
            backend,
            desired_harness_generation.map(str::to_owned),
            harness_roll_concurrency,
            None,
        ))
    }

    /// The concurrency cap (issue #117 phase 2): at cap, a would-be roll is
    /// DEFERRED — the reconcile arm returns before ever calling the backend's
    /// `teardown` or sending a `Conversation` status patch.
    #[tokio::test]
    async fn roll_harness_defers_at_cap_without_tearing_down() {
        let patches = Arc::new(Mutex::new(Vec::new()));
        // One conversation already mid-roll; cap of 1 ⇒ already at cap.
        let client = fake_conversations::client(
            vec![fake_conversations::item("other-conv", true)],
            patches.clone(),
        );
        let backend = Arc::new(RecordingBackend {
            readiness: ClaimReadiness::default(),
            ..Default::default()
        });
        let ctx = fake_ctx(client, backend.clone(), Some("2026.7.0"), 1);

        let mut c = conv("c1");
        c.metadata.namespace = Some("test-ns".to_owned());
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            harness_image_generation: Some("2026.6.0".to_owned()),
            ..Default::default()
        });

        reconcile(Arc::new(c), ctx)
            .await
            .expect("reconcile succeeds even when the roll is deferred");

        assert_eq!(
            backend
                .teardown_calls
                .load(std::sync::atomic::Ordering::SeqCst),
            0,
            "a roll deferred by the concurrency cap must never reach the backend's teardown"
        );
        assert!(
            patches.lock().unwrap().is_empty(),
            "a deferred roll must not touch the Conversation status at all"
        );
    }

    /// Under cap, the roll proceeds: the backend's `teardown` is called
    /// exactly once, and the status patch tears down the pod-address /
    /// readiness / generation fields, marks `rollingHarness: true` (the
    /// concurrency cap's own signal), and sets the transient display phase.
    #[tokio::test]
    async fn roll_harness_proceeds_under_cap() {
        let patches = Arc::new(Mutex::new(Vec::new()));
        // No conversations currently mid-roll ⇒ well under any cap ≥ 1.
        let client = fake_conversations::client(vec![], patches.clone());
        let backend = Arc::new(RecordingBackend {
            readiness: ClaimReadiness::default(),
            ..Default::default()
        });
        let ctx = fake_ctx(client, backend.clone(), Some("2026.7.0"), 5);

        let mut c = conv("c1");
        c.metadata.namespace = Some("test-ns".to_owned());
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            harness_image_generation: Some("2026.6.0".to_owned()),
            ..Default::default()
        });

        reconcile(Arc::new(c), ctx).await.expect("reconcile");

        assert_eq!(
            backend
                .teardown_calls
                .load(std::sync::atomic::Ordering::SeqCst),
            1,
            "under cap, the roll must tear the claim down exactly once"
        );
        let recorded = patches.lock().unwrap();
        assert_eq!(recorded.len(), 1, "exactly one status patch");
        let status = &recorded[0]["status"];
        assert_eq!(status["sandboxClaimName"], serde_json::Value::Null);
        assert_eq!(status["harnessReady"], serde_json::Value::Bool(false));
        assert_eq!(status["harnessImageGeneration"], serde_json::Value::Null);
        assert_eq!(
            status["rollingHarness"],
            serde_json::Value::Bool(true),
            "the concurrency cap's own signal must be set the instant the teardown fires"
        );
        assert_eq!(
            status["phase"],
            serde_json::Value::String(PHASE_ROLLING_HARNESS.to_owned())
        );
    }

    /// The None-vs-Some adopt path, exercised at the `reconcile` level (not
    /// just `plan`): a conversation whose claim is already `Ready` but whose
    /// `harnessImageGeneration` was never recorded gets the desired
    /// generation STAMPED via a status patch — even though nothing else
    /// (pod IP, readiness, phase) changed — and the backend's `teardown` is
    /// never called. This is the exact scenario the adversarial review named
    /// as the easiest to get wrong.
    #[tokio::test]
    async fn sync_status_adopts_a_never_recorded_generation_without_rolling() {
        let patches = Arc::new(Mutex::new(Vec::new()));
        let client = fake_conversations::client(vec![], patches.clone());
        let backend = Arc::new(RecordingBackend {
            readiness: ClaimReadiness {
                unit_present: true,
                address: Some(DialAddress::PodIp("10.4.2.7".to_owned())),
                harness_ready: true,
            },
            ..Default::default()
        });
        let ctx = fake_ctx(client, backend.clone(), Some("2026.7.0"), 5);

        let mut c = conv("c1");
        c.metadata.namespace = Some("test-ns".to_owned());
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            sandbox_claim_name: Some("c1".to_owned()),
            harness_image_generation: None,
            pod_ip: Some("10.4.2.7".to_owned()),
            harness_ready: true,
            phase: Some(PHASE_READY.to_owned()),
            ..Default::default()
        });

        reconcile(Arc::new(c), ctx).await.expect("reconcile");

        assert_eq!(
            backend
                .teardown_calls
                .load(std::sync::atomic::Ordering::SeqCst),
            0,
            "adopting a never-recorded generation must never tear the claim down"
        );
        let recorded = patches.lock().unwrap();
        assert_eq!(
            recorded.len(),
            1,
            "the adopt stamp must still send a patch even though pod_ip/harness_ready/phase are unchanged"
        );
        assert_eq!(
            recorded[0]["status"]["harnessImageGeneration"],
            serde_json::Value::String("2026.7.0".to_owned())
        );
    }

    /// A fake `Api<Conversation>` client that answers exactly one `DELETE`
    /// with `response_status`, recording the request body (the `DeleteParams`
    /// JSON, including any `preconditions`) into `deletes`. Purpose-built for
    /// the `DeleteConversation` reconcile-arm tests below — kept separate
    /// from `fake_conversations::client` (which never needs to see a DELETE)
    /// rather than growing that shared helper's signature for one arm.
    fn fake_delete_client(
        response_status: http::StatusCode,
        deletes: Arc<Mutex<Vec<serde_json::Value>>>,
    ) -> kube::Client {
        let svc = tower::service_fn(move |req: http::Request<kube::client::Body>| {
            let deletes = deletes.clone();
            let status = response_status;
            async move {
                let body = req
                    .into_body()
                    .collect_bytes()
                    .await
                    .map(|b| b.to_vec())
                    .unwrap_or_default();
                let parsed: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default();
                deletes.lock().expect("deletes lock poisoned").push(parsed);
                let payload = if status.is_success() {
                    serde_json::json!({
                        "apiVersion": "polychrome.dev/v1alpha1",
                        "kind": "Conversation",
                        "metadata": { "name": "c1" },
                        "spec": { "model": "m", "idleTimeoutSeconds": 300 },
                        "status": {},
                    })
                } else {
                    serde_json::json!({
                        "kind": "Status",
                        "apiVersion": "v1",
                        "status": "Failure",
                        "message": "the object has been modified; please apply your changes \
                                     to the latest version and try again",
                        "reason": "Conflict",
                        "code": status.as_u16(),
                    })
                };
                let resp = http::Response::builder()
                    .status(status)
                    .header("content-type", "application/json")
                    .body(kube::client::Body::from(
                        serde_json::to_vec(&payload).unwrap(),
                    ))
                    .expect("build fake response");
                Ok::<_, std::convert::Infallible>(resp)
            }
        });
        kube::Client::new(svc, "test-ns")
    }

    /// A `Conversation` shaped to make `plan` emit `DeleteConversation`: closed,
    /// claim already cleared, `closedAt` long enough ago to clear a 60s window.
    fn conv_past_retention() -> Conversation {
        let mut c = conv("c1");
        c.metadata.namespace = Some("test-ns".to_owned());
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.metadata.resource_version = Some("42".to_owned());
        c.metadata.uid = Some("uid-c1".to_owned());
        c.status = Some(ConversationStatus {
            closed: true,
            closed_at: Some(1_000),
            ..Default::default()
        });
        c
    }

    #[tokio::test]
    async fn delete_conversation_sends_a_resource_version_precondition() {
        let deletes = Arc::new(Mutex::new(Vec::new()));
        let client = fake_delete_client(http::StatusCode::OK, deletes.clone());
        let backend = Arc::new(RecordingBackend::default());
        let ctx = Arc::new(Context::new(
            client,
            TPL.to_owned(),
            backend,
            None,
            DEFAULT_HARNESS_ROLL_CONCURRENCY,
            Some(60),
        ));

        reconcile(Arc::new(conv_past_retention()), ctx)
            .await
            .expect("a successful delete must not error");

        let recorded = deletes.lock().unwrap();
        assert_eq!(recorded.len(), 1, "exactly one delete call");
        assert_eq!(
            recorded[0]["preconditions"]["resourceVersion"],
            serde_json::Value::String("42".to_owned()),
            "the delete must be conditioned on the exact object `plan` observed, so a \
             concurrent resume (a status patch bumping resourceVersion) makes it fail \
             instead of silently discarding the resume"
        );
    }

    #[tokio::test]
    async fn delete_conversation_conflict_from_a_concurrent_change_is_not_an_error() {
        let deletes = Arc::new(Mutex::new(Vec::new()));
        let client = fake_delete_client(http::StatusCode::CONFLICT, deletes.clone());
        let backend = Arc::new(RecordingBackend::default());
        let ctx = Arc::new(Context::new(
            client,
            TPL.to_owned(),
            backend,
            None,
            DEFAULT_HARNESS_ROLL_CONCURRENCY,
            Some(60),
        ));

        let action = reconcile(Arc::new(conv_past_retention()), ctx)
            .await
            .expect(
                "a 409 from a concurrent change (e.g. a resume racing the delete) must be \
                 treated as benign, not propagated as a reconcile error",
            );

        assert_eq!(
            action,
            Action::requeue(Duration::from_secs(5)),
            "must requeue promptly to re-evaluate fresh state, not back off as if this were \
             a real failure"
        );
    }
}