polyc-controller 2026.10.0

//! Reconciler: drive each [`Conversation`] toward one substrate actor.
//!
//! 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 prometheus::{IntCounter, register_int_counter};
use serde_json::json;

use crate::conversation::{Condition, ConditionStatus, Conversation, FINALIZER, upsert_condition};
use crate::execution_backend::{ExecutionBackend, UnitReadiness};

/// `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`" (actor
/// creation, 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 and suspended: the actor keeps its state and the next turn resumes it.
///
/// [`ReconcileAction::SuspendIdle`] sets `status.idle_reclaimed`, and the
/// control plane's per-turn activity stamp clears it. The phase shows the
/// field. It is display only.
pub const PHASE_PAUSED: &str = "Paused";
/// Transient display phase set the instant a [`ReconcileAction::RollHarness`]
/// moves the actor onto a new template.
///
/// For `kubectl get conversation` visibility only. The AUTHORITATIVE signal the
/// concurrency cap counts against is `status.rolling_harness` (see
/// `Context::count_in_flight_rolls`).
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`] (fast, so the
/// roll marker clears promptly rather than waiting for the steady poll).
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 for a live conversation. The reconciler has no watch on the
/// actor, so this poll is what observes a crash, an out-of-band delete, and the
/// moment the idle delay passes.
pub(crate) const PENDING_POLL: Duration = Duration::from_secs(30);

/// The longest a turn may run. The harness caps every turn at this value
/// (`HARNESS_TURN_DEADLINE`).
const TURN_DEADLINE: Duration = Duration::from_mins(10);

/// Slack added to [`TURN_DEADLINE`] before an activity stamp with no terminal
/// stamp reads as an orphaned turn. Covers dial time and clock skew.
const ORPHANED_TURN_MARGIN: Duration = Duration::from_mins(2);

/// 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 the actor named `actor_name` from `template`.
    CreateActor {
        /// Name to give the actor (mirrors the conversation's name).
        actor_name: String,
        /// The `ActorTemplate` to create the actor from.
        template: String,
    },
    /// Delete the actor. `remove_finalizer` is set only when the
    /// `Conversation` itself is being deleted (vs merely `closed`).
    Cleanup {
        /// The actor to delete, if one was recorded.
        actor_name: Option<String>,
        /// Whether to also drop our finalizer (true ⇒ object is terminating).
        remove_finalizer: bool,
    },
    /// The actor exists; read its readiness and reflect it into the
    /// `Conversation` status (`harness_ready`). A crashed actor is deleted here
    /// and created again on the next pass.
    SyncStatus {
        /// The actor to read readiness from.
        actor_name: String,
    },
    /// The conversation's last turn ended and the idle delay has passed (or a
    /// turn orphaned past the deadline): suspend the actor. It keeps its
    /// `/workspace`, and the next turn resumes it through the router.
    ///
    /// Never emitted while a turn is in flight. A suspend breaks the stream that
    /// the turn holds open through the router. See [`plan`] for the rule.
    SuspendIdle {
        /// The actor to suspend.
        actor_name: String,
    },
    /// The actor exists but its recorded `status.harnessImageGeneration` no
    /// longer matches [`Context::desired_harness_generation`] — move it onto the
    /// current template (issue #117).
    ///
    /// # Mechanism
    ///
    /// The reconcile effect suspends the actor if it runs, then points it at the
    /// template the control plane registered for the new harness image
    /// (`UpdateActor`). The change takes effect at the next resume. Substrate
    /// keeps the actor's `/workspace` across it and boots the guest fresh from
    /// the new image. The reconcile effect stamps the new generation, and
    /// `status.rolling_harness = true` so the concurrency cap
    /// ([`Context::harness_roll_concurrency`]) can see it. A roll suspends an
    /// actor, so [`plan`] emits it only when no turn is in flight.
    ///
    /// Resume is automatic: there is no separate "resume" step. The next turn's
    /// dial through the router resumes the actor, and [event-log
    /// replay](crate::conversation) rebuilds the conversation state on the fresh
    /// guest.
    RollHarness {
        /// The actor to move onto the current template.
        actor_name: String,
    },
    /// The conversation has been idle past `spec.idleTimeoutSeconds`; mark it
    /// closed so the next reconcile deletes the actor via
    /// [`ReconcileAction::Cleanup`]. This is the idle-GC (issue #265). Closing
    /// deletes the actor and its `/workspace`.
    CloseIdle,
    /// `conv` has been fully closed (its actor already deleted,
    /// `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{actor_name: None, remove_finalizer: true}` — a no-op teardown
    /// (no actor 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` object, never the live actor, so it
/// cannot tell "actor recorded, healthy" from "actor gone". Once
/// `status.actorName` is set, [`plan`] always yields `SyncStatus`, whose only IO
/// is a read. If the actor disappears — deleted out-of-band, or lost with the
/// substrate database — the conversation would wedge un-`Ready` forever. The
/// fix is level-triggered: the `SyncStatus` arm consults this predicate and,
/// when it returns `true`, clears `status.actorName`. The *next* [`plan`] pass
/// then sees `None` and re-enters [`ReconcileAction::CreateActor`]. The
/// `Conversation` object is the durable desired state, so nothing else is
/// needed to rebuild the actor.
///
/// # Distinguishing "gone" from "still starting"
///
/// The discriminator is [`UnitReadiness::unit_present`]. A suspended actor is
/// present. Only a genuinely absent actor reports `false`. Even then the
/// predicate requires *evidence the actor once existed and was healthy*:
/// `status.phase == "Ready"`, or a recorded actor identity. A just-created actor
/// that the read has not yet observed is not thrashed. After a self-heal fires
/// it clears that evidence, so the predicate is one-shot and cannot loop.
#[must_use]
fn unit_is_gone(
    status: &crate::conversation::ConversationStatus,
    readiness: &UnitReadiness,
) -> bool {
    // Unit object still exists (even if crashed or resuming) → 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_identity = status.actor_uid.is_some();
    was_ready || had_identity
}

/// Outcome of [`sync_execution_unit`]: the unit vanished (self-heal — see
/// [`unit_is_gone`]), the unit crashed, or it is present and its
/// [`UnitReadiness`] is returned.
#[derive(Debug, Clone, PartialEq, Eq)]
enum SyncOutcome {
    /// The unit vanished; the caller clears `actorName` so the next [`plan`]
    /// pass creates it again.
    Vanished,
    /// The unit crashed. Substrate never recovers a crashed actor, so the caller
    /// deletes it and clears `actorName`. The next [`plan`] pass creates a new
    /// actor, and the event log rebuilds the conversation on it.
    Crashed,
    /// The unit is present and healthy; its readiness was read.
    Synced(UnitReadiness),
}

/// The `SyncStatus` action's IO against the [`ExecutionBackend`] only — no
/// `Conversation` kube calls. Isolated from [`reconcile`]'s status patches so it
/// is unit-testable with a mocked backend.
///
/// Substrate keeps no spec that this controller could re-apply, so the sync is
/// a read: an actor's template changes only through [`ReconcileAction::RollHarness`].
///
/// # Errors
///
/// Returns [`Error`] if the backend call fails.
async fn sync_execution_unit(
    backend: &dyn ExecutionBackend,
    conv: &Conversation,
    actor_name: &str,
    ns: &str,
) -> Result<SyncOutcome, Error> {
    let current = conv.status.clone().unwrap_or_default();
    let readiness = backend.readiness(actor_name, ns).await?;
    if unit_is_gone(&current, &readiness) {
        return Ok(SyncOutcome::Vanished);
    }
    if readiness.crashed {
        return Ok(SyncOutcome::Crashed);
    }
    Ok(SyncOutcome::Synced(readiness))
}

/// Whether a turn may be running on `status`'s actor. Pure.
///
/// The control plane stamps `lastActivityUnix` as a turn starts and
/// `lastTerminalUnix` as it ends, and it makes the terminal stamp strictly
/// greater. So a turn is in flight when the terminal stamp is absent or not
/// greater than the activity stamp.
const fn turn_in_flight(status: &crate::conversation::ConversationStatus) -> bool {
    match (status.last_terminal_unix, status.last_activity_unix) {
        (Some(terminal), Some(activity)) => terminal <= activity,
        (None, Some(_)) => true,
        (_, None) => false,
    }
}

/// Whether the idle rule says to suspend the actor now. Pure.
///
/// - The actor is not already suspended, and it is ready.
/// - The last turn ended (terminal stamp greater than the activity stamp) and the
///   idle delay has passed since that stamp; or
/// - a turn started, no terminal stamp followed, and the activity is older than
///   [`TURN_DEADLINE`] plus [`ORPHANED_TURN_MARGIN`]. That is a crash backstop: a
///   control plane that died mid-turn never stamps the end.
fn should_suspend(
    status: &crate::conversation::ConversationStatus,
    now_unix: i64,
    suspend_idle_seconds: Option<i64>,
) -> bool {
    let Some(delay) = suspend_idle_seconds else {
        return false;
    };
    if status.idle_reclaimed || !status.harness_ready {
        return false;
    }
    if turn_in_flight(status) {
        let orphaned_after =
            i64::try_from((TURN_DEADLINE + ORPHANED_TURN_MARGIN).as_secs()).unwrap_or(i64::MAX);
        return status
            .last_activity_unix
            .is_some_and(|activity| now_unix.saturating_sub(activity) > orphaned_after);
    }
    status
        .last_terminal_unix
        .is_some_and(|terminal| now_unix.saturating_sub(terminal) > delay)
}

/// 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 an actor exists
/// (`status.actorName` 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 moves the actor onto a new template, and only while no turn is in flight |
///
/// 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 actor 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.
///
/// `suspend_idle_seconds` is [`Context::suspend_idle_seconds`]. `Some(n)`
/// suspends a ready actor once its last turn ended more than `n` seconds ago
/// ([`ReconcileAction::SuspendIdle`]). It never suspends while a turn is in
/// flight, and `None` turns idle suspend off. See the idle suspend rule (INV-SUB4).
#[must_use]
pub fn plan(
    conv: &Conversation,
    default_template: &str,
    now_unix: i64,
    desired_harness_generation: Option<&str>,
    retention_seconds: Option<i64>,
    suspend_idle_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 actor_name = conv.status.as_ref().and_then(|s| s.actor_name.clone());

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

    if closed {
        // Delete the actor but leave the object (and finalizer) until it is
        // actually deleted. Once the actor 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 actor_name.is_some() {
            ReconcileAction::Cleanup {
                actor_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 actor is deleted 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
    {
        // A suspended actor is idle too, so it closes at the same timeout. The
        // close policy is unchanged by suspend: closing deletes the actor and
        // its `/workspace`.
        return ReconcileAction::CloseIdle;
    }

    let status = conv.status.clone().unwrap_or_default();
    actor_name.map_or_else(
        || ReconcileAction::CreateActor {
            actor_name: conv.name_any(),
            template: default_template.to_owned(),
        },
        |actor_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. A roll suspends the
            // actor, so it waits for the turn to end.
            if let Some(desired) = desired_harness_generation {
                let recorded = status.harness_image_generation.as_deref();
                if let Some(recorded) = recorded
                    && recorded != desired
                    && !turn_in_flight(&status)
                {
                    return ReconcileAction::RollHarness { actor_name };
                }
            }
            if should_suspend(&status, now_unix, suspend_idle_seconds) {
                return ReconcileAction::SuspendIdle { actor_name };
            }
            ReconcileAction::SyncStatus { actor_name }
        },
    )
}

/// Shared reconcile context.
pub struct Context {
    /// Kube client used for `Conversation`-CR API calls (finalizer + status).
    pub client: Client,
    /// The `ActorTemplate` name that new actors are created from and that a
    /// roll moves an actor onto.
    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) 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>,
    /// Idle delay, in seconds, after a turn ends before the actor is suspended
    /// (`POLYCHROME_ACTOR_SUSPEND_IDLE_SECONDS`, default 300), read once at
    /// controller startup. `None` turns idle suspend off.
    pub suspend_idle_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>,
        suspend_idle_seconds: Option<i64>,
    ) -> Self {
        Self {
            client,
            default_template,
            backend,
            desired_harness_generation,
            harness_roll_concurrency,
            closed_conversation_retention_seconds,
            suspend_idle_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,
        ctx.suspend_idle_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::CreateActor {
            actor_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 an actor decided-on for
            // conversation X, and when" from logs.
            tracing::info!(
                conversation = %name,
                actor = %actor_name,
                template = %template,
                "creating actor for conversation"
            );
            if let Some(parent) = conv.spec.parent_conversation_id.as_deref() {
                tracing::info!(
                    child_conversation = %name,
                    parent_conversation = parent,
                    "creating actor for child conversation (handoff)"
                );
            }
            let ensured = ctx
                .backend
                .ensure(&conv, &actor_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(),
            );
            // The actor identity lands in the same patch as the actor name, so a
            // later delete can pin itself to this incarnation.
            let status = json!({
                "status": {
                    "actorName": actor_name,
                    "actorUid": ensured.uid,
                    "phase": PHASE_PENDING,
                    "conditions": conditions,
                }
            });
            convs
                .patch_status(&name, &PatchParams::default(), &Patch::Merge(&status))
                .await?;
            // A created actor is dialable at once, so read it back soon.
            Ok(Action::requeue(Duration::from_secs(1)))
        }
        ReconcileAction::Cleanup {
            actor_name,
            remove_finalizer,
        } => {
            if let Some(actor) = actor_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, &actor, &ns).await {
                    if deletion_age(&conv) >= TEARDOWN_GRACE {
                        tracing::error!(
                            actor = %actor,
                            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 actor down MUST also clear the readiness pointers:
                // the actor behind `actorName` is gone, so leaving
                // `harnessReady: true` makes a resumed (re-opened) conversation
                // advertise a ready harness that does not exist. Clearing them
                // forces the resume to wait for a fresh actor (created by the
                // CreateActor 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": {
                    "actorName": null,
                    "actorUid": null,
                    "harnessReady": false,
                    "idleReclaimed": 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 { actor_name } => {
            let current = conv.status.clone().unwrap_or_default();

            // `sync_execution_unit` reads readiness and classifies the unit.
            // See its doc comment for the vanished and crashed outcomes.
            let outcome =
                sync_execution_unit(ctx.backend.as_ref(), &conv, &actor_name, &ns).await?;

            let readiness = match outcome {
                // Issue #2 self-heal: a unit that was once healthy now reads
                // as absent ⇒ it vanished (deleted out-of-band, or the
                // substrate database was lost). Clear `actorName` so the NEXT
                // plan() pass re-enters CreateActor and creates it again. 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 actor disappeared; creating it again",
                        &now_rfc3339(),
                    );
                    let patch = json!({ "status": {
                        "actorName": null,
                        "actorUid": null,
                        "harnessReady": false,
                        "idleReclaimed": false,
                        "phase": PHASE_PENDING,
                        "conditions": conditions,
                    } });
                    convs
                        .patch_status(&name, &PatchParams::default(), &Patch::Merge(&patch))
                        .await?;
                    tracing::warn!(
                        actor = %actor_name,
                        "actor vanished; cleared its name to trigger re-create"
                    );
                    return Ok(Action::requeue(Duration::from_secs(1)));
                }
                // Substrate never recovers a crashed actor. Delete it, then
                // clear the name so the next pass creates a new one. The
                // conversation's `/workspace` is lost with the actor; the event
                // log rebuilds the conversation itself.
                SyncOutcome::Crashed => {
                    ctx.backend.teardown(&conv, &actor_name, &ns).await?;
                    let conditions = conditions_triad(
                        &current.conditions,
                        false,
                        true,
                        true,
                        "ExecutionUnitCrashed",
                        "the actor crashed; deleting it and creating a new one",
                        &now_rfc3339(),
                    );
                    let patch = json!({ "status": {
                        "actorName": null,
                        "actorUid": null,
                        "harnessReady": false,
                        "idleReclaimed": false,
                        "phase": PHASE_PENDING,
                        "conditions": conditions,
                    } });
                    convs
                        .patch_status(&name, &PatchParams::default(), &Patch::Merge(&patch))
                        .await?;
                    tracing::warn!(
                        actor = %actor_name,
                        "actor crashed; deleted it and cleared its name to trigger re-create"
                    );
                    return Ok(Action::requeue(Duration::from_secs(1)));
                }
                SyncOutcome::Synced(readiness) => readiness,
            };

            // Desired display phase. While `idle_reclaimed` is set, show
            // `Paused` regardless of readiness; once a turn clears the flag,
            // fall back to readiness.
            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` (or waits for the turn to
            // end). So the only stamping this arm ever needs to do is fill in a
            // still-`None` recorded generation once a desired one is known.
            // `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;
            // Adopt the actor identity when a lost status patch left it unset.
            let uid_to_stamp = readiness
                .uid
                .as_deref()
                .filter(|uid| current.actor_uid.as_deref() != Some(uid));
            // Only patch when something actually changed, to avoid a
            // status-write → watch → reconcile churn loop.
            if current.harness_ready != readiness.harness_ready
                || current.phase.as_deref() != Some(phase)
                || generation_to_stamp.is_some()
                || clear_rolling
                || uid_to_stamp.is_some()
            {
                let (ready, progressing, reason, message) = if current.idle_reclaimed {
                    (
                        true,
                        false,
                        "Suspended",
                        "the actor is suspended 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": {
                    "harnessReady": readiness.harness_ready,
                    "phase": phase,
                    "conditions": conditions,
                } });
                if let Some(desired) = generation_to_stamp {
                    patch["status"]["harnessImageGeneration"] = json!(desired);
                }
                if let Some(uid) = uid_to_stamp {
                    patch["status"]["actorUid"] = json!(uid);
                }
                if clear_rolling {
                    patch["status"]["rollingHarness"] = json!(false);
                }
                convs
                    .patch_status(&name, &PatchParams::default(), &Patch::Merge(&patch))
                    .await?;
                tracing::info!(
                    harness_ready = readiness.harness_ready,
                    generation_stamped = ?generation_to_stamp,
                    "synced conversation status from execution unit"
                );
            }
            // The reconciler has no watch on the actor, so it polls. The same
            // cadence observes a crash, an out-of-band delete, and the moment
            // the idle delay passes.
            Ok(Action::requeue(PENDING_POLL))
        }
        ReconcileAction::SuspendIdle { actor_name } => {
            // `plan` emits this only when no turn is in flight (INV-SUB4). The
            // backend suspend is idempotent for an actor that is not running.
            ctx.backend.suspend(&actor_name, &ns).await?;
            let current = conv.status.clone().unwrap_or_default();
            let conditions = conditions_triad(
                &current.conditions,
                true,
                false,
                false,
                "Suspended",
                "the actor is suspended while idle; it resumes on the next turn",
                &now_rfc3339(),
            );
            let patch = json!({ "status": {
                "idleReclaimed": true,
                "phase": PHASE_PAUSED,
                "conditions": conditions,
            } });
            convs
                .patch_status(&name, &PatchParams::default(), &Patch::Merge(&patch))
                .await?;
            tracing::info!(actor = %actor_name, "suspended idle actor");
            Ok(Action::requeue(PENDING_POLL))
        }
        ReconcileAction::RollHarness { actor_name } => {
            // Concurrency cap (issue #117): `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 backend 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!(
                    actor = %actor_name,
                    in_flight,
                    cap = ctx.harness_roll_concurrency,
                    "deferring harness roll: concurrency cap reached"
                );
                return Ok(Action::requeue(ROLL_DEFER_POLL));
            }
            // Suspend if running, then point the actor at the current template
            // (see `ReconcileAction::RollHarness`'s doc comment). The actor keeps
            // its `/workspace`.
            ctx.backend
                .retemplate(&actor_name, &ctx.default_template, &ns)
                .await?;
            let current = conv.status.clone().unwrap_or_default();
            let conditions = conditions_triad(
                &current.conditions,
                true,
                true,
                false,
                "RollingHarnessImage",
                "moved the actor onto the current harness template; it resumes on the next turn",
                &now_rfc3339(),
            );
            let mut patch = json!({ "status": {
                "idleReclaimed": true,
                "phase": PHASE_ROLLING_HARNESS,
                "rollingHarness": true,
                "conditions": conditions,
            } });
            if let Some(desired) = ctx.desired_harness_generation.as_deref() {
                patch["status"]["harnessImageGeneration"] = json!(desired);
            }
            convs
                .patch_status(&name, &PatchParams::default(), &Patch::Merge(&patch))
                .await?;
            tracing::info!(
                actor = %actor_name,
                "moved actor onto the current harness template"
            );
            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
            // deletes the actor.
            let current = conv.status.clone().unwrap_or_default();
            let conditions = conditions_triad(
                &current.conditions,
                false,
                true,
                false,
                "ClosingIdleConversation",
                "idle past the timeout; deleting the execution unit",
                &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{actor_name: None, remove_finalizer: true}` (no actor
            // 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))),
    }
}

/// 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)
}

/// What the `Conversation` reconciler creates and drives: the execution backend
/// and the template new units come from.
///
/// The composition root builds this, because it owns the connection to the
/// backend and registers the template. [`run`] takes it as an `Option`: a
/// deployment with no execution backend reconciles no `Conversation`.
pub struct ExecutionUnits {
    /// The backend that creates, reads, suspends, and deletes units.
    pub backend: Arc<dyn ExecutionBackend>,
    /// The template name that new units are created from.
    pub template: String,
}

/// Default idle delay before an actor is suspended: five minutes.
const DEFAULT_SUSPEND_IDLE_SECONDS: i64 = 300;

/// Parse `POLYCHROME_ACTOR_SUSPEND_IDLE_SECONDS`'s raw value. Pure.
///
/// - Not set (`None`) ⇒ [`DEFAULT_SUSPEND_IDLE_SECONDS`].
/// - A positive integer ⇒ that many seconds.
/// - Empty, `"0"`, negative, or malformed ⇒ `None`, which turns idle suspend
///   off. A typo fails toward "never suspend", because a suspend that fires too
///   early costs every conversation a resume.
#[must_use]
fn parse_suspend_idle_seconds(raw: Option<&str>) -> Option<i64> {
    raw.map_or(Some(DEFAULT_SUSPEND_IDLE_SECONDS), |value| {
        match value.trim().parse::<i64>() {
            Ok(seconds) if seconds > 0 => Some(seconds),
            Ok(_) | Err(_) => None,
        }
    })
}

/// 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 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,
    connector_transport: Arc<dyn polyc_tools::mcp_client::ConnectorTransport>,
    execution: Option<ExecutionUnits>,
) -> 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.
    // With no execution backend, no conversation has an actor to reconcile, so
    // the `Conversation` controller does not start. The other controllers do.
    let conversation_client = client.clone();
    let conversation_watch = watch_client.clone();
    let conversations = async move {
        match execution {
            Some(execution) => {
                run_conversation(
                    conversation_client,
                    conversation_watch,
                    namespace,
                    execution,
                )
                .await
            }
            None => Ok(()),
        }
    };
    let toolservices = crate::toolservice_reconcile::run_toolservice(
        client.clone(),
        watch_client.clone(),
        namespace,
        connector_transport,
    );
    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,
    execution: ExecutionUnits,
) -> 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(),
    );
    // `POLYCHROME_ACTOR_SUSPEND_IDLE_SECONDS` unset ⇒ 300. `0`, empty, or an
    // unparseable value turns idle suspend off, so a typo never suspends
    // sooner than the operator asked.
    let suspend_idle_seconds = parse_suspend_idle_seconds(
        std::env::var("POLYCHROME_ACTOR_SUSPEND_IDLE_SECONDS")
            .ok()
            .as_deref(),
    );
    tracing::info!(
        desired_harness_generation = ?desired_harness_generation,
        harness_roll_concurrency,
        closed_conversation_retention_seconds = ?closed_conversation_retention_seconds,
        suspend_idle_seconds = ?suspend_idle_seconds,
        template = %execution.template,
        backend = execution.backend.kind(),
        "harness image-generation roll, idle suspend, and closed-conversation GC configured"
    );

    // Build the reconcile context. There is no owned-resource watch: an actor
    // is not a Kubernetes object, so the reconciler polls it.
    let ctx = Arc::new(Context::new(
        client,
        execution.template,
        execution.backend,
        desired_harness_generation,
        harness_roll_concurrency,
        closed_conversation_retention_seconds,
        suspend_idle_seconds,
    ));
    let controller = Controller::new(convs, 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-abc123";

    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, 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, None),
            ReconcileAction::CreateActor {
                actor_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 {
            actor_name: Some("c1".to_owned()),
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, None, None, None),
            ReconcileAction::SyncStatus {
                actor_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 {
            actor_name: Some("c1".to_owned()),
            harness_image_generation: Some("2026.6.0".to_owned()),
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, None, None, None),
            ReconcileAction::SyncStatus {
                actor_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 {
            actor_name: Some("c1".to_owned()),
            harness_image_generation: None,
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, Some("2026.7.0"), None, None),
            ReconcileAction::SyncStatus {
                actor_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 {
            actor_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, None),
            ReconcileAction::SyncStatus {
                actor_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 {
            actor_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, None),
            ReconcileAction::RollHarness {
                actor_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 {
            actor_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, None),
            ReconcileAction::CloseIdle
        );
        // 200s since last turn (< 300) → still active, normal sync.
        assert_eq!(
            plan(&c, TPL, 1_000 + 200, None, None, None),
            ReconcileAction::SyncStatus {
                actor_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 {
            actor_name: Some("c1".to_owned()),
            last_activity_unix: None,
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 10_000_000_000, None, None, None),
            ReconcileAction::SyncStatus {
                actor_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 {
            actor_name: Some("c1".to_owned()),
            closed: true,
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, None, None, None),
            ReconcileAction::Cleanup {
                actor_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, 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), None),
            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), None),
            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), None),
            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), None),
            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, 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 {
            actor_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), None),
            ReconcileAction::Cleanup {
                actor_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 {
            actor_name: Some("c1".to_owned()),
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, None, None, None),
            ReconcileAction::Cleanup {
                actor_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, None),
            ReconcileAction::AddFinalizer
        );
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        // Once finalised the plan creates an actor, same as a parent.
        assert_eq!(
            plan(&c, TPL, 0, None, None, None),
            ReconcileAction::CreateActor {
                actor_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 {
            actor_name: Some("c1".to_owned()),
            phase: Some("Ready".to_owned()),
            ..Default::default()
        };
        assert!(unit_is_gone(&status, &UnitReadiness::default()));

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

    #[test]
    fn unit_gone_after_a_recorded_identity_triggers_reclaim() {
        // A recorded actor identity is also evidence the unit once existed.
        let status = ConversationStatus {
            actor_name: Some("c1".to_owned()),
            actor_uid: Some("uid-1".to_owned()),
            ..Default::default()
        };
        assert!(unit_is_gone(&status, &UnitReadiness::default()));
    }

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

    #[test]
    fn healthy_unit_read_is_not_gone() {
        // A present unit is never "gone", even if previously Ready.
        let status = ConversationStatus {
            actor_name: Some("c1".to_owned()),
            actor_uid: Some("uid-1".to_owned()),
            phase: Some(PHASE_READY.to_owned()),
            harness_ready: true,
            ..Default::default()
        };
        assert!(!unit_is_gone(&status, &ready_unit()));
    }

    #[test]
    fn present_but_transiently_unready_unit_is_not_gone() {
        // A crashed or resuming actor still exists, so it must NOT read as gone
        // even though the conversation was previously Ready. The crash path, not
        // the vanish path, handles a crashed actor.
        let status = ConversationStatus {
            actor_name: Some("c1".to_owned()),
            actor_uid: Some("uid-1".to_owned()),
            phase: Some(PHASE_READY.to_owned()),
            harness_ready: true,
            ..Default::default()
        };
        let unready = UnitReadiness {
            unit_present: true,
            harness_ready: false,
            ..Default::default()
        };
        assert!(!unit_is_gone(&status, &unready));
    }

    fn ready_unit() -> UnitReadiness {
        UnitReadiness {
            unit_present: true,
            address: Some(crate::DialAddress::Actor {
                atespace: "ns".to_owned(),
                name: "c1".to_owned(),
            }),
            harness_ready: true,
            crashed: false,
            uid: Some("uid-1".to_owned()),
        }
    }

    #[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, None), ReconcileAction::Noop);
    }

    /// A mocked [`ExecutionBackend`] that records every 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,
        suspend_calls: std::sync::atomic::AtomicUsize,
        retemplate_calls: std::sync::atomic::AtomicUsize,
        readiness: UnitReadiness,
    }

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

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

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

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

        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"
        }
    }

    /// A present, healthy unit syncs without any write to the backend. Substrate
    /// keeps no spec for this controller to re-apply, so the sync is a read.
    #[tokio::test]
    async fn sync_status_reads_a_present_unit_and_writes_nothing() {
        let backend = RecordingBackend {
            readiness: ready_unit(),
            ..Default::default()
        };
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            actor_name: Some("c1".to_owned()),
            phase: Some(PHASE_PENDING.to_owned()),
            ..Default::default()
        });

        let outcome = sync_execution_unit(&backend, &c, "c1", "ns")
            .await
            .expect("sync_execution_unit");
        assert!(matches!(outcome, SyncOutcome::Synced(_)));
        assert_eq!(
            backend
                .ensure_calls
                .load(std::sync::atomic::Ordering::SeqCst),
            0,
            "the sync path only reads"
        );
    }

    /// Self-heal (issue #2): a genuinely vanished unit is reported as vanished
    /// and NOT re-created within the same pass. That would mask the vanish.
    #[tokio::test]
    async fn sync_status_does_not_recreate_a_vanished_unit_in_the_same_pass() {
        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 {
            actor_name: Some("c1".to_owned()),
            phase: Some(PHASE_READY.to_owned()),
            ..Default::default()
        });

        let outcome = sync_execution_unit(&backend, &c, "c1", "ns")
            .await
            .expect("sync_execution_unit");
        assert!(matches!(outcome, SyncOutcome::Vanished));
        assert_eq!(
            backend
                .ensure_calls
                .load(std::sync::atomic::Ordering::SeqCst),
            0
        );
    }

    /// ADR 0035 decision 3: the substrate database is disposable. After a loss,
    /// the recorded phase is Ready and the read returns nothing. The pass reports
    /// the unit vanished, and the pass after that creates the actor again from the
    /// `Conversation`, which is the durable desired state.
    #[tokio::test]
    async fn sync_recreates_after_substrate_database_loss() {
        let backend = RecordingBackend::default();
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        c.status = Some(ConversationStatus {
            actor_name: Some("c1".to_owned()),
            actor_uid: Some("uid-lost".to_owned()),
            phase: Some(PHASE_READY.to_owned()),
            harness_ready: true,
            ..Default::default()
        });
        assert!(matches!(
            sync_execution_unit(&backend, &c, "c1", "ns").await.unwrap(),
            SyncOutcome::Vanished
        ));
        // The reconcile effect clears the recorded identity, as it does for a
        // vanish. The next plan then creates the actor.
        c.status = Some(ConversationStatus {
            phase: Some(PHASE_PENDING.to_owned()),
            ..Default::default()
        });
        assert_eq!(
            plan(&c, TPL, 0, None, None, None),
            ReconcileAction::CreateActor {
                actor_name: "c1".to_owned(),
                template: TPL.to_owned(),
            }
        );
    }

    /// Substrate never recovers a crashed actor, so the sync reports it for a
    /// delete and a re-create. It does not report it as vanished, because the
    /// actor still exists.
    #[tokio::test]
    async fn sync_reports_a_crashed_actor_for_delete_and_recreate() {
        let backend = RecordingBackend {
            readiness: UnitReadiness {
                unit_present: true,
                crashed: true,
                uid: Some("uid-1".to_owned()),
                ..Default::default()
            },
            ..Default::default()
        };
        let mut c = conv("c1");
        c.status = Some(ConversationStatus {
            actor_name: Some("c1".to_owned()),
            actor_uid: Some("uid-1".to_owned()),
            phase: Some(PHASE_READY.to_owned()),
            ..Default::default()
        });
        assert!(matches!(
            sync_execution_unit(&backend, &c, "c1", "ns").await.unwrap(),
            SyncOutcome::Crashed
        ));
    }

    // ---- idle suspend (INV-SUB4) ------------------------------------------

    const NOW: i64 = 1_800_000_000;
    const DELAY: i64 = 300;
    /// The wall clock `reconcile` plans at. The `reconcile`-level tests stamp
    /// relative to it, because `reconcile` reads the real clock.
    fn real_now() -> i64 {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX))
    }

    fn ready_conv(activity: Option<i64>, terminal: Option<i64>) -> Conversation {
        let mut c = conv("c1");
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        // A long idle timeout keeps the idle reaper out of these cases.
        c.spec.idle_timeout_seconds = 0;
        c.status = Some(ConversationStatus {
            actor_name: Some("c1".to_owned()),
            actor_uid: Some("uid-1".to_owned()),
            harness_ready: true,
            phase: Some(PHASE_READY.to_owned()),
            last_activity_unix: activity,
            last_terminal_unix: terminal,
            ..Default::default()
        });
        c
    }

    fn suspend() -> ReconcileAction {
        ReconcileAction::SuspendIdle {
            actor_name: "c1".to_owned(),
        }
    }

    fn sync() -> ReconcileAction {
        ReconcileAction::SyncStatus {
            actor_name: "c1".to_owned(),
        }
    }

    #[test]
    fn plan_never_suspends_while_a_turn_is_in_flight() {
        // A turn started 400 seconds ago and has no terminal stamp after it.
        let c = ready_conv(Some(NOW - 400), None);
        assert_eq!(plan(&c, TPL, NOW, None, None, Some(DELAY)), sync());
        // The previous turn's terminal stamp is older than the new activity.
        let c = ready_conv(Some(NOW - 400), Some(NOW - 900));
        assert_eq!(plan(&c, TPL, NOW, None, None, Some(DELAY)), sync());
        // Equal stamps count as in flight: the control plane makes a real
        // terminal stamp strictly greater, so a tie means a new turn began in
        // the second the last one ended.
        let c = ready_conv(Some(NOW - 400), Some(NOW - 400));
        assert_eq!(plan(&c, TPL, NOW, None, None, Some(DELAY)), sync());
    }

    #[test]
    fn plan_suspends_only_after_terminal_plus_idle_delay() {
        // The turn ended 200 seconds ago: inside the delay.
        let c = ready_conv(Some(NOW - 250), Some(NOW - 200));
        assert_eq!(plan(&c, TPL, NOW, None, None, Some(DELAY)), sync());
        // Exactly at the delay: still not past it.
        let c = ready_conv(Some(NOW - 350), Some(NOW - DELAY));
        assert_eq!(plan(&c, TPL, NOW, None, None, Some(DELAY)), sync());
        // Past the delay.
        let c = ready_conv(Some(NOW - 400), Some(NOW - DELAY - 1));
        assert_eq!(plan(&c, TPL, NOW, None, None, Some(DELAY)), suspend());
    }

    #[test]
    fn plan_suspends_a_turn_orphaned_past_the_turn_deadline() {
        let orphaned_after =
            i64::try_from((TURN_DEADLINE + ORPHANED_TURN_MARGIN).as_secs()).unwrap();
        // A turn that never stamped its end, still inside deadline plus margin.
        let c = ready_conv(Some(NOW - orphaned_after), None);
        assert_eq!(plan(&c, TPL, NOW, None, None, Some(DELAY)), sync());
        // Past it: the control plane died mid-turn, so the actor is reclaimed.
        let c = ready_conv(Some(NOW - orphaned_after - 1), None);
        assert_eq!(plan(&c, TPL, NOW, None, None, Some(DELAY)), suspend());
    }

    #[test]
    fn plan_does_not_resuspend_a_reclaimed_actor() {
        let mut c = ready_conv(Some(NOW - 900), Some(NOW - 800));
        c.status.as_mut().unwrap().idle_reclaimed = true;
        assert_eq!(plan(&c, TPL, NOW, None, None, Some(DELAY)), sync());
    }

    #[test]
    fn plan_does_not_suspend_an_actor_that_is_not_ready_yet() {
        let mut c = ready_conv(Some(NOW - 900), Some(NOW - 800));
        c.status.as_mut().unwrap().harness_ready = false;
        assert_eq!(plan(&c, TPL, NOW, None, None, Some(DELAY)), sync());
    }

    #[test]
    fn plan_with_idle_suspend_off_never_suspends() {
        let c = ready_conv(Some(NOW - 9_000), Some(NOW - 8_000));
        assert_eq!(plan(&c, TPL, NOW, None, None, None), sync());
    }

    #[test]
    fn a_conversation_with_no_stamp_is_never_suspended() {
        let c = ready_conv(None, None);
        assert_eq!(plan(&c, TPL, NOW, None, None, Some(DELAY)), sync());
    }

    #[test]
    fn a_suspended_conversation_still_closes_at_the_idle_timeout() {
        // Suspend does not change the close policy: closing deletes the actor
        // and its workspace once the conversation idles past its timeout.
        let mut c = ready_conv(Some(NOW - 900), Some(NOW - 800));
        c.spec.idle_timeout_seconds = 600;
        c.status.as_mut().unwrap().idle_reclaimed = true;
        assert_eq!(
            plan(&c, TPL, NOW, None, None, Some(DELAY)),
            ReconcileAction::CloseIdle
        );
    }

    #[test]
    fn plan_defers_a_roll_while_a_turn_is_in_flight_and_rolls_when_idle() {
        let mut in_flight = ready_conv(Some(NOW - 10), None);
        in_flight.status.as_mut().unwrap().harness_image_generation = Some("old".to_owned());
        assert_eq!(
            plan(&in_flight, TPL, NOW, Some("new"), None, Some(DELAY)),
            sync(),
            "a roll suspends the actor, so it waits for the turn to end"
        );
        let mut idle = ready_conv(Some(NOW - 20), Some(NOW - 10));
        idle.status.as_mut().unwrap().harness_image_generation = Some("old".to_owned());
        assert_eq!(
            plan(&idle, TPL, NOW, Some("new"), None, Some(DELAY)),
            ReconcileAction::RollHarness {
                actor_name: "c1".to_owned()
            }
        );
    }

    #[test]
    fn suspend_idle_env_parsing_fails_toward_never_suspend() {
        assert_eq!(parse_suspend_idle_seconds(None), Some(300));
        assert_eq!(parse_suspend_idle_seconds(Some("120")), Some(120));
        assert_eq!(parse_suspend_idle_seconds(Some(" 45 ")), Some(45));
        for off in ["", "0", "-5", "soon"] {
            assert_eq!(parse_suspend_idle_seconds(Some(off)), None, "{off:?}");
        }
    }

    #[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,
            Some(DELAY),
        ))
    }

    /// The concurrency cap (issue #117): at cap, a would-be roll is
    /// DEFERRED — the reconcile arm returns before ever calling the backend's
    /// `retemplate` 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: UnitReadiness::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 {
            actor_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
                .retemplate_calls
                .load(std::sync::atomic::Ordering::SeqCst),
            0,
            "a roll deferred by the concurrency cap must never reach the backend"
        );
        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 `retemplate` is called exactly
    /// once, and the status patch stamps the new generation, marks
    /// `rollingHarness: true` (the concurrency cap's own signal), and sets the
    /// transient display phase. It never deletes the actor, so `/workspace`
    /// survives the roll.
    #[tokio::test]
    async fn roll_harness_proceeds_under_cap_and_keeps_the_actor() {
        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: UnitReadiness::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 {
            actor_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
                .retemplate_calls
                .load(std::sync::atomic::Ordering::SeqCst),
            1,
            "under cap, the roll must move the actor onto the new template exactly once"
        );
        assert_eq!(
            backend
                .teardown_calls
                .load(std::sync::atomic::Ordering::SeqCst),
            0,
            "a roll must never delete the actor: its workspace has to survive"
        );
        let recorded = patches.lock().unwrap();
        assert_eq!(recorded.len(), 1, "exactly one status patch");
        let status = &recorded[0]["status"];
        assert_eq!(
            status["harnessImageGeneration"],
            serde_json::Value::String("2026.7.0".to_owned())
        );
        assert_eq!(
            status["rollingHarness"],
            serde_json::Value::Bool(true),
            "the concurrency cap's own signal must be set the instant the roll fires"
        );
        assert_eq!(
            status["phase"],
            serde_json::Value::String(PHASE_ROLLING_HARNESS.to_owned())
        );
    }

    /// The suspend arm calls the backend once and records the suspend, so the
    /// control plane's next turn resumes the actor and clears the flag.
    #[tokio::test]
    async fn suspend_idle_suspends_the_actor_and_marks_the_conversation_paused() {
        let patches = Arc::new(Mutex::new(Vec::new()));
        let client = fake_conversations::client(vec![], patches.clone());
        let backend = Arc::new(RecordingBackend::default());
        let ctx = fake_ctx(client, backend.clone(), None, 5);
        let mut c = ready_conv(Some(real_now() - 900), Some(real_now() - 800));
        c.metadata.namespace = Some("test-ns".to_owned());

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

        assert_eq!(
            backend
                .suspend_calls
                .load(std::sync::atomic::Ordering::SeqCst),
            1
        );
        let recorded = patches.lock().unwrap();
        assert_eq!(recorded.len(), 1);
        assert_eq!(
            recorded[0]["status"]["idleReclaimed"],
            serde_json::Value::Bool(true)
        );
        assert_eq!(
            recorded[0]["status"]["phase"],
            serde_json::Value::String(PHASE_PAUSED.to_owned())
        );
    }

    /// A crashed actor is deleted, and the conversation forgets it so the next
    /// pass creates a new one.
    #[tokio::test]
    async fn sync_deletes_then_recreates_a_crashed_actor() {
        let patches = Arc::new(Mutex::new(Vec::new()));
        let client = fake_conversations::client(vec![], patches.clone());
        let backend = Arc::new(RecordingBackend {
            readiness: UnitReadiness {
                unit_present: true,
                crashed: true,
                uid: Some("uid-1".to_owned()),
                ..Default::default()
            },
            ..Default::default()
        });
        let ctx = fake_ctx(client, backend.clone(), None, 5);
        let mut c = ready_conv(Some(real_now() - 10), Some(real_now() - 5));
        c.metadata.namespace = Some("test-ns".to_owned());

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

        assert_eq!(
            backend
                .teardown_calls
                .load(std::sync::atomic::Ordering::SeqCst),
            1,
            "a crashed actor is deleted"
        );
        let recorded = patches.lock().unwrap();
        assert_eq!(recorded.len(), 1);
        let status = &recorded[0]["status"];
        assert_eq!(status["actorName"], serde_json::Value::Null);
        assert_eq!(status["actorUid"], serde_json::Value::Null);
        assert_eq!(status["harnessReady"], serde_json::Value::Bool(false));
    }

    /// Creating an actor records its name and its identity in one status patch.
    #[tokio::test]
    async fn create_actor_records_the_name_and_the_incarnation_uid() {
        let patches = Arc::new(Mutex::new(Vec::new()));
        let client = fake_conversations::client(vec![], patches.clone());
        let backend = Arc::new(RecordingBackend::default());
        let ctx = fake_ctx(client, backend.clone(), None, 5);
        let mut c = conv("c1");
        c.metadata.namespace = Some("test-ns".to_owned());
        c.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);

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

        assert_eq!(
            backend
                .ensure_calls
                .load(std::sync::atomic::Ordering::SeqCst),
            1
        );
        let recorded = patches.lock().unwrap();
        assert_eq!(recorded.len(), 1);
        assert_eq!(
            recorded[0]["status"]["actorName"],
            serde_json::Value::String("c1".to_owned())
        );
        assert_eq!(
            recorded[0]["status"]["actorUid"],
            serde_json::Value::String("uid-new".to_owned())
        );
    }

    /// The None-vs-Some adopt path, exercised at the `reconcile` level (not
    /// just `plan`): a conversation whose actor is already `Ready` but whose
    /// `harnessImageGeneration` was never recorded gets the desired
    /// generation STAMPED via a status patch — even though nothing else
    /// (actor identity, 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: ready_unit(),
            ..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 {
            actor_name: Some("c1".to_owned()),
            harness_image_generation: None,
            actor_uid: Some("uid-1".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 delete the actor"
        );
        let recorded = patches.lock().unwrap();
        assert_eq!(
            recorded.len(),
            1,
            "the adopt stamp must still send a patch even though the actor identity, readiness, and 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),
            None,
        ));

        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),
            None,
        ));

        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"
        );
    }
}