polyc-controller 2026.9.0

//! Reconciler: periodically health-check each [`ToolService`] and reflect the
//! result into its `status` subresource.
//!
//! This mirrors [`crate::reconcile`](mod@crate::reconcile) exactly: the *decision* (what should
//! happen) is the pure [`plan`] function, unit-tested without a cluster or a
//! network. The *effect* (applying that decision) is [`reconcile`](fn@reconcile),
//! which dials the connector's MCP endpoint and patches the CR.
//!
//! Where the `Conversation` reconciler distils a `SandboxClaim`'s pod readiness
//! into the `Conversation` status, this one distils an MCP `list_tools`
//! handshake into the `ToolService` status (`healthy`, `available_tools`,
//! `message`). A connector that is unreachable is *not* a reconcile error: it
//! is a `healthy = false` status with a human-readable `message`, so a flaky
//! connector never wedges the controller's work queue.

use std::{
    sync::{Arc, LazyLock},
    time::{Duration, SystemTime},
};

use chrono::{DateTime, SecondsFormat, Utc};
use futures::StreamExt;
use k8s_openapi::api::core::v1::Secret;
use kube::{
    Api, Client, Resource, ResourceExt,
    api::{Patch, PatchParams},
    runtime::{
        controller::{Action, Controller},
        watcher,
    },
};
use polyc_agent::ToolExecutor;
use polyc_tools::mcp_client::McpToolSource;
use prometheus::{IntCounter, register_int_counter};
use serde_json::json;

use crate::reserved_secrets::{ReservedSecret, check_secret_ref};
use crate::toolservice::{Auth, BearerSecretRef, FINALIZER, ToolDescriptor, ToolService};

/// How often a healthy/known `ToolService` is re-checked. A check is also
/// forced (regardless of this interval) whenever the spec `version` differs
/// from the `observed_version` already in status.
pub const CHECK_INTERVAL: Duration = Duration::from_mins(5);

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

/// Errors surfaced by the `ToolService` reconciler.
///
/// Note that a *connector* being unreachable is deliberately **not** an error
/// here — that is reflected as an unhealthy status. Only failures to talk to
/// the kube API (or a missing namespace) abort a reconcile pass.
#[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("toolservice has no namespace")]
    NoNamespace,
}

/// What a single reconcile pass should do for a [`ToolService`]. The pure
/// output of [`plan`]; [`reconcile`] turns it into IO (an MCP handshake and a
/// kube patch).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ToolServiceAction {
    /// Attach our [`FINALIZER`] before recording any controller-side state.
    AddFinalizer,
    /// Drop our finalizer; the object is terminating. (There is no external
    /// resource to delete — the connector is owned out-of-band — so cleanup is
    /// just finalizer removal.)
    Cleanup,
    /// A health check is due: connect to the connector, list tools, and write
    /// the distilled result into status. Carries the first remote URL to dial.
    CheckHealth {
        /// The `remotes[0].url` to connect to.
        url: String,
    },
    /// Steady state; the status is fresh and reflects the current version.
    Noop,
}

/// Readiness distilled from an MCP `list_tools` handshake, for mirroring into
/// the `ToolService` status. The async analogue of
/// [`crate::reconcile`](mod@crate::reconcile)'s `ClaimReadiness`.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ToolServiceReadiness {
    /// `true` when the connector handshook and enumerated its tools.
    pub healthy: bool,
    /// Tools advertised by the connector (empty when unhealthy).
    pub available_tools: Vec<ToolDescriptor>,
    /// Human-readable detail; an error reason when unhealthy, `None` when
    /// healthy.
    pub message: Option<String>,
}

/// Decide what to do for `ts` at instant `now`, re-checking on `interval`.
///
/// Pure: no IO, no ambient clock (the caller passes `now`), fully testable.
///
/// The "health-check due" rule fires when **any** of:
/// * there is no recorded `last_checked_at` (never checked);
/// * `last_checked_at` is unparseable (treat as stale → re-check);
/// * `now - last_checked_at >= interval` (stale);
/// * `observed_version != spec.version` (the connector was re-registered at a
///   new version, so the cached status no longer applies).
#[must_use]
pub fn plan(ts: &ToolService, now: SystemTime, interval: Duration) -> ToolServiceAction {
    let being_deleted = ts.meta().deletion_timestamp.is_some();
    let has_finalizer = ts.finalizers().iter().any(|f| f == FINALIZER);

    if being_deleted {
        return if has_finalizer {
            ToolServiceAction::Cleanup
        } else {
            ToolServiceAction::Noop
        };
    }

    if !has_finalizer {
        return ToolServiceAction::AddFinalizer;
    }

    if check_due(ts, now, interval) {
        // Dial the first remote. A spec with no remotes still warrants a check
        // pass so the status records the misconfiguration (handled in
        // `reconcile` — an empty URL fails the connect and is reported).
        let url = ts
            .spec
            .remotes
            .first()
            .map(|r| r.url.clone())
            .unwrap_or_default();
        return ToolServiceAction::CheckHealth { url };
    }

    ToolServiceAction::Noop
}

/// Whether a health check is due for `ts` at `now`. Pure helper for [`plan`].
fn check_due(ts: &ToolService, now: SystemTime, interval: Duration) -> bool {
    let Some(status) = ts.status.as_ref() else {
        return true; // never reconciled
    };
    if status.observed_version.as_deref() != Some(ts.spec.version.as_str()) {
        return true; // version bump (or never recorded) ⇒ re-check
    }
    let Some(last) = status.last_checked_at.as_deref() else {
        return true; // healthy flag set but no timestamp ⇒ re-check
    };
    let Ok(parsed) = DateTime::parse_from_rfc3339(last) else {
        return true; // unparseable ⇒ treat as stale
    };
    let now: DateTime<Utc> = now.into();
    // `signed_duration_since` is negative if `last` is in the future (clock
    // skew); only a positive elapsed >= interval is "stale".
    now.signed_duration_since(parsed.with_timezone(&Utc))
        .to_std()
        .is_ok_and(|elapsed| elapsed >= interval)
}

/// Connect to the MCP connector at `url`, list its tools, and distil the
/// outcome into a [`ToolServiceReadiness`].
///
/// A connection or `list_tools` failure (and an empty `url`) yields
/// `healthy = false` with the error in `message` — it never returns `Err`, so
/// an unreachable connector cannot abort a reconcile pass.
pub async fn health_check(url: &str, bearer: Option<String>) -> ToolServiceReadiness {
    if url.is_empty() {
        return ToolServiceReadiness {
            healthy: false,
            available_tools: Vec::new(),
            message: Some("no remote URL configured".to_owned()),
        };
    }
    // Bounded connect: a connector that accepts the socket but never completes
    // the MCP handshake must not hang the reconcile pass (and pin a controller
    // worker) — mirror the harness's bounded-connect contract. The `bearer` is
    // resolved from the connector's `auth.bearerSecretRef` so an auth-protected
    // server lists its tools here instead of 401-ing (which would report it
    // unhealthy with 0 tools). It is bound to `url` (RFC 8707) so the same
    // audience-scoping the harness enforces also holds on the health-check dial.
    // `false` approval: the controller only lists tools; per-tool approval is
    // the harness's concern.
    let bearer = match bearer {
        Some(token) => match polyc_tools::AudienceBoundToken::new(token, url) {
            Ok(token) => Some(token),
            Err(err) => {
                return ToolServiceReadiness {
                    healthy: false,
                    available_tools: Vec::new(),
                    message: Some(err.to_string()),
                };
            }
        },
        None => None,
    };
    match McpToolSource::connect(
        url.to_owned(),
        polyc_tools::ConnectOptions {
            // No label: the status `availableTools` describes THIS server's own
            // raw catalogue, not a namespaced model surface. The harness applies
            // the `<label>__` connector prefix when it composes many connectors
            // for the model; here a single known connector is introspected, so
            // the raw names the remote registered are the operator-facing answer.
            bearer,
            ..polyc_tools::ConnectOptions::default()
        },
    )
    .await
    {
        Ok(source) => {
            let available_tools = source
                .specs()
                .into_iter()
                .map(descriptor_from_spec)
                .collect();
            // Politely cancel the transport now that we have the catalogue;
            // we hold the source only long enough to read its specs.
            source.shutdown();
            ToolServiceReadiness {
                healthy: true,
                available_tools,
                message: None,
            }
        }
        Err(err) => ToolServiceReadiness {
            healthy: false,
            available_tools: Vec::new(),
            message: Some(err.to_string()),
        },
    }
}

/// Reflect one connector [`ToolSpec`] into its catalog [`ToolDescriptor`],
/// carrying the argument schema (serialized to a JSON string) and the MCP
/// behavioral annotations the harness's gates key on.
fn descriptor_from_spec(spec: polyc_llm::ToolSpec) -> ToolDescriptor {
    ToolDescriptor {
        name: spec.name,
        description: (!spec.description.is_empty()).then_some(spec.description),
        // A schema that fails to serialize (it came from a parsed MCP object, so
        // this is not expected) reflects as empty rather than aborting the check.
        input_schema: serde_json::to_string(&spec.schema_json).unwrap_or_default(),
        title: spec.title,
        read_only: spec.read_only,
        destructive: spec.destructive,
        open_world: spec.open_world,
    }
}

/// Choose the catalog to persist, version-keying it (invariant 8) and gating
/// adoption on the check having actually succeeded: the freshly listed tools
/// are adopted only when a **healthy** check ran during a version change (or
/// the first-ever check, where `current.observed_version` is `None`); a
/// same-version periodic re-check, OR a *failed* check that happened to land
/// on a version bump, KEEPS the previously reviewed catalog untouched.
///
/// Without the health gate, a transient connector outage colliding with a
/// version bump (`health_check` returns `available_tools: []` on any dial or
/// list failure) would adopt that empty listing as "the fresh catalog" and,
/// paired with [`observed_version_after_check`] settling `observed_version`
/// unconditionally, freeze it in permanently once the connector recovers —
/// see `version_keyed_catalog_freezes_prior_catalog_on_failed_check_during_version_bump`.
///
/// Pure so the version-keying invariant is unit-tested without a cluster.
#[must_use]
fn version_keyed_catalog(
    current: &crate::toolservice::ToolServiceStatus,
    spec_version: &str,
    readiness: &ToolServiceReadiness,
) -> Vec<ToolDescriptor> {
    let version_matches = current.observed_version.as_deref() == Some(spec_version);
    if version_matches || !readiness.healthy {
        // Same version: freeze the catalog to what was reviewed under it. A
        // failed check — whether same-version or colliding with a bump —
        // never overwrites a catalog with a busted (empty) listing; it just
        // leaves the prior catalog in place until a healthy check lists it.
        current.available_tools.clone()
    } else {
        // A *successful* check on a version bump (or first-ever check):
        // adopt the fresh list under the new version.
        readiness.available_tools.clone()
    }
}

/// The `observedVersion` to persist after a check, gating settlement on the
/// check having succeeded. A check that ran during a version bump (or the
/// very first check) but FAILED must not settle `observed_version` to the
/// new `spec_version` — that would make [`check_due`] treat the object as
/// current on the next pass, permanently freezing whatever catalog
/// [`version_keyed_catalog`] retained instead of retrying at the fast,
/// always-due cadence a version mismatch forces.
///
/// Pure so the retry-forcing invariant is unit-tested without a cluster.
#[must_use]
fn observed_version_after_check(
    current: &crate::toolservice::ToolServiceStatus,
    spec_version: &str,
    healthy: bool,
) -> Option<String> {
    let version_matches = current.observed_version.as_deref() == Some(spec_version);
    if version_matches || healthy {
        Some(spec_version.to_owned())
    } else {
        // Failed check colliding with a version bump (or the first-ever
        // check): hold the prior `observed_version` (possibly `None`) so
        // `check_due` keeps forcing retries.
        current.observed_version.clone()
    }
}

/// Shared reconcile context.
pub struct Context {
    /// Kube client used for `ToolService`-CR API calls (finalizer + status).
    pub client: Client,
    /// How often to re-check a known-good connector.
    pub check_interval: Duration,
}

/// Screen a connector's author-supplied `auth.bearerSecretRef` before anything
/// reads it: `Ok(None)` when the spec configures no auth, `Ok(Some(_))` for a
/// reference the resolver may read, and `Err` for one naming platform key
/// material.
///
/// Pure, like [`plan`], so the refusal is unit-tested without a cluster —
/// [`resolve_bearer`] is then only the read this admits.
///
/// # Errors
///
/// Returns [`ReservedSecret`] when the reference names platform key material.
fn admissible_bearer_ref(auth: Option<&Auth>) -> Result<Option<&BearerSecretRef>, ReservedSecret> {
    let Some(auth) = auth else {
        return Ok(None);
    };
    let r = &auth.bearer_secret_ref;
    check_secret_ref(&r.name)?;
    Ok(Some(r))
}

/// Resolve a connector's bearer token from its `auth.bearerSecretRef` (a
/// namespace-local `Secret`), so the health check dials it the same way the
/// harness does. `Ok(None)` when there's no auth configured; a
/// missing/unreadable Secret is logged and treated as `Ok(None)` (the check
/// then reports unhealthy, which is the correct signal that the connector
/// can't be reached as configured).
///
/// The `bearerSecretRef` name is author-supplied, so
/// [`check_secret_ref`](crate::reserved_secrets::check_secret_ref) runs before
/// the read: a reference to platform key material returns `Err` and the caller
/// never dials, rather than resolving to `None` and dialing the author's URL
/// unauthenticated.
///
/// # Errors
///
/// Returns [`ReservedSecret`] when the reference names platform key material.
async fn resolve_bearer(
    client: &Client,
    namespace: &str,
    auth: Option<&Auth>,
) -> Result<Option<String>, ReservedSecret> {
    let Some(r) = admissible_bearer_ref(auth)? else {
        return Ok(None);
    };
    let secrets: Api<Secret> = Api::namespaced(client.clone(), namespace);
    match secrets.get(&r.name).await {
        Ok(secret) => Ok(secret
            .data
            .as_ref()
            .and_then(|d| d.get(&r.key))
            .and_then(|b| String::from_utf8(b.0.clone()).ok())
            .map(|s| s.trim().to_owned())),
        Err(e) => {
            tracing::warn!(secret = %r.name, key = %r.key, error = %e,
                "toolservice health check: failed to read bearer secret");
            Ok(None)
        }
    }
}

/// The readiness a refused `bearerSecretRef` settles into: unhealthy, no
/// catalog, and the refusal itself as the status message the resource's author
/// reads. Nothing is dialed and nothing is read.
fn refused_readiness(err: &ReservedSecret) -> ToolServiceReadiness {
    ToolServiceReadiness {
        healthy: false,
        available_tools: Vec::new(),
        message: Some(err.to_string()),
    }
}

/// Reconcile one [`ToolService`] by executing the [`plan`].
///
/// # Errors
///
/// Returns [`Error`] if any kube API call fails or the object lacks a
/// namespace. A connector being unreachable is reflected into status, not
/// returned as an error.
#[tracing::instrument(skip_all, fields(toolservice = %ts.name_any()))]
pub async fn reconcile(ts: Arc<ToolService>, ctx: Arc<Context>) -> Result<Action, Error> {
    TOOLSERVICE_RECONCILE_TOTAL.inc();
    let ns = ts.namespace().ok_or(Error::NoNamespace)?;
    let name = ts.name_any();
    let api: Api<ToolService> = Api::namespaced(ctx.client.clone(), &ns);
    let pp = PatchParams::apply("polychrome.dev/controller");

    match plan(&ts, SystemTime::now(), ctx.check_interval) {
        ToolServiceAction::AddFinalizer => {
            let patch = json!({ "metadata": { "finalizers": [FINALIZER] } });
            api.patch(&name, &pp, &Patch::Merge(&patch)).await?;
            Ok(Action::requeue(Duration::from_secs(1)))
        }
        ToolServiceAction::Cleanup => {
            let patch = json!({ "metadata": { "finalizers": [] } });
            api.patch(&name, &pp, &Patch::Merge(&patch)).await?;
            Ok(Action::await_change())
        }
        ToolServiceAction::CheckHealth { url } => {
            // Fail closed on a reserved reference: no Secret read, no dial —
            // the refusal becomes the status the author reads.
            let readiness = match resolve_bearer(&ctx.client, &ns, ts.spec.auth.as_ref()).await {
                Ok(bearer) => health_check(&url, bearer).await,
                Err(err) => {
                    tracing::error!(
                        toolservice = %name,
                        secret = %err.secret,
                        "refused a toolservice bearer reference to platform key material"
                    );
                    refused_readiness(&err)
                }
            };
            let version = ts.spec.version.clone();
            let now = Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true);
            let current = ts.status.clone().unwrap_or_default();

            // Version-key the catalog: adopt the freshly listed tools only on a
            // *healthy* version change (or first check); a same-version
            // re-check, or a failed check that landed on a bump, keeps the
            // reviewed catalog frozen, so health can flip without rewriting
            // the catalog under an unchanged version (invariant 8) and a
            // transient outage can never freeze an empty catalog in.
            let catalog = version_keyed_catalog(&current, &version, &readiness);
            let observed_version =
                observed_version_after_check(&current, &version, readiness.healthy);

            // Only patch when something actually changed, to avoid a
            // status-write → watch → reconcile churn loop. `lastCheckedAt`
            // alone is deliberately excluded from the comparison: bumping it
            // every pass would defeat the de-bounce, and the requeue interval
            // already bounds check freshness.
            let changed = current.healthy != readiness.healthy
                || current.available_tools != catalog
                || current.message != readiness.message
                || current.observed_version != observed_version;

            if changed {
                let patch = json!({ "status": {
                    "healthy": readiness.healthy,
                    "lastCheckedAt": now,
                    "availableTools": catalog,
                    "message": readiness.message,
                    "observedVersion": observed_version,
                } });
                api.patch_status(&name, &PatchParams::default(), &Patch::Merge(&patch))
                    .await?;
                tracing::info!(
                    healthy = readiness.healthy,
                    tools = catalog.len(),
                    message = ?readiness.message,
                    "synced toolservice status from MCP health check"
                );
            } else {
                // Nothing material changed, but record that we checked so the
                // next `plan` sees a fresh timestamp and settles to Noop until
                // the interval elapses again.
                let patch = json!({ "status": { "lastCheckedAt": now } });
                api.patch_status(&name, &PatchParams::default(), &Patch::Merge(&patch))
                    .await?;
            }
            Ok(Action::requeue(ctx.check_interval))
        }
        ToolServiceAction::Noop => Ok(Action::requeue(ctx.check_interval)),
    }
}

/// Requeue policy on reconcile failure: retry with a fixed short backoff.
#[must_use]
pub fn error_policy(_ts: Arc<ToolService>, err: &Error, _ctx: Arc<Context>) -> Action {
    tracing::warn!(error = %err, "toolservice reconcile failed; requeuing");
    Action::requeue(Duration::from_secs(10))
}

/// Run the `ToolService` controller until its watch stream ends. Blocks for
/// the process lifetime in a real deployment.
///
/// The watcher is namespaced (`Api::namespaced`) for the same least-privilege
/// reasons documented on [`crate::reconcile::run`]. Unlike the `Conversation`
/// controller this one `.owns(...)` nothing: a `ToolService` has no
/// controller-created child object — its "child" is an external MCP server
/// reached over the network, reflected into status rather than managed as a CR.
///
/// `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 `Api` handle passed to `Controller::new`
/// below (see [`crate::reconcile::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`].
pub async fn run_toolservice(
    client: Client,
    watch_client: Client,
    namespace: &str,
) -> Result<(), Error> {
    let api: Api<ToolService> = Api::namespaced(watch_client, namespace);
    let ctx = Arc::new(Context {
        client,
        check_interval: CHECK_INTERVAL,
    });

    Controller::new(api, watcher::Config::default())
        .run(reconcile, error_policy, ctx)
        .for_each(|res| async move {
            if let Err(e) = res {
                tracing::warn!(error = %e, "toolservice reconcile stream item errored");
            }
        })
        .await;
    Ok(())
}

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

    use std::time::{Duration, SystemTime};

    use chrono::{SecondsFormat, Utc};
    use kube::api::ObjectMeta;

    use super::*;
    use crate::toolservice::{Remote, ToolServiceSpec, ToolServiceStatus};

    const INTERVAL: Duration = Duration::from_secs(300);

    fn ts(name: &str) -> ToolService {
        ToolService::new(
            name,
            ToolServiceSpec {
                server_name: Some("io.github.acme/search".to_owned()),
                description: Some("Search connector".to_owned()),
                version: "1.0.0".to_owned(),
                remotes: vec![Remote {
                    url: "https://acme.example/mcp".to_owned(),
                    transport: "streamable-http".to_owned(),
                }],
                needs_approval: false,
                approval_tools: Vec::new(),
                default_enabled: false,
                auth: None,
                tool_expansions: std::collections::BTreeMap::new(),
            },
        )
    }

    fn finalized(name: &str) -> ToolService {
        let mut t = ts(name);
        t.metadata.finalizers = Some(vec![FINALIZER.to_owned()]);
        t
    }

    fn rfc3339(at: SystemTime) -> String {
        let dt: chrono::DateTime<Utc> = at.into();
        dt.to_rfc3339_opts(SecondsFormat::Secs, true)
    }

    #[test]
    fn fresh_toolservice_gets_a_finalizer_first() {
        assert_eq!(
            plan(&ts("s1"), SystemTime::now(), INTERVAL),
            ToolServiceAction::AddFinalizer
        );
    }

    #[test]
    fn finalized_without_status_checks_health() {
        assert_eq!(
            plan(&finalized("s1"), SystemTime::now(), INTERVAL),
            ToolServiceAction::CheckHealth {
                url: "https://acme.example/mcp".to_owned(),
            }
        );
    }

    #[test]
    fn fresh_status_settles_to_noop() {
        let mut t = finalized("s1");
        let now = SystemTime::now();
        t.status = Some(ToolServiceStatus {
            healthy: true,
            last_checked_at: Some(rfc3339(now)),
            observed_version: Some("1.0.0".to_owned()),
            ..Default::default()
        });
        assert_eq!(plan(&t, now, INTERVAL), ToolServiceAction::Noop);
    }

    #[test]
    fn stale_status_checks_health() {
        let mut t = finalized("s1");
        let checked = SystemTime::now() - Duration::from_secs(600);
        t.status = Some(ToolServiceStatus {
            healthy: true,
            last_checked_at: Some(rfc3339(checked)),
            observed_version: Some("1.0.0".to_owned()),
            ..Default::default()
        });
        assert_eq!(
            plan(&t, SystemTime::now(), INTERVAL),
            ToolServiceAction::CheckHealth {
                url: "https://acme.example/mcp".to_owned(),
            }
        );
    }

    #[test]
    fn version_bump_checks_health_even_when_timestamp_is_fresh() {
        let mut t = finalized("s1");
        let now = SystemTime::now();
        t.status = Some(ToolServiceStatus {
            healthy: true,
            last_checked_at: Some(rfc3339(now)),
            observed_version: Some("0.9.0".to_owned()), // stale version
            ..Default::default()
        });
        assert_eq!(
            plan(&t, now, INTERVAL),
            ToolServiceAction::CheckHealth {
                url: "https://acme.example/mcp".to_owned(),
            }
        );
    }

    #[test]
    fn missing_timestamp_checks_health() {
        let mut t = finalized("s1");
        t.status = Some(ToolServiceStatus {
            healthy: true,
            last_checked_at: None,
            observed_version: Some("1.0.0".to_owned()),
            ..Default::default()
        });
        assert_eq!(
            plan(&t, SystemTime::now(), INTERVAL),
            ToolServiceAction::CheckHealth {
                url: "https://acme.example/mcp".to_owned(),
            }
        );
    }

    #[test]
    fn unparseable_timestamp_checks_health() {
        let mut t = finalized("s1");
        t.status = Some(ToolServiceStatus {
            healthy: true,
            last_checked_at: Some("not-a-timestamp".to_owned()),
            observed_version: Some("1.0.0".to_owned()),
            ..Default::default()
        });
        assert_eq!(
            plan(&t, SystemTime::now(), INTERVAL),
            ToolServiceAction::CheckHealth {
                url: "https://acme.example/mcp".to_owned(),
            }
        );
    }

    #[test]
    fn check_due_with_no_remotes_yields_empty_url() {
        let mut t = finalized("s1");
        t.spec.remotes.clear();
        assert_eq!(
            plan(&t, SystemTime::now(), INTERVAL),
            ToolServiceAction::CheckHealth { url: String::new() }
        );
    }

    #[test]
    fn deletion_with_finalizer_cleans_up() {
        let mut t = finalized("s1");
        t.metadata.deletion_timestamp = Some(k8s_openapi::apimachinery::pkg::apis::meta::v1::Time(
            "2026-05-27T00:00:00Z".parse().unwrap(),
        ));
        assert_eq!(
            plan(&t, SystemTime::now(), INTERVAL),
            ToolServiceAction::Cleanup
        );
    }

    #[test]
    fn deletion_without_finalizer_is_noop() {
        let mut t = ts("s1");
        t.metadata = ObjectMeta {
            name: Some("s1".to_owned()),
            deletion_timestamp: Some(k8s_openapi::apimachinery::pkg::apis::meta::v1::Time(
                "2026-05-27T00:00:00Z".parse().unwrap(),
            )),
            ..Default::default()
        };
        assert_eq!(
            plan(&t, SystemTime::now(), INTERVAL),
            ToolServiceAction::Noop
        );
    }

    fn descriptor(name: &str) -> ToolDescriptor {
        ToolDescriptor {
            name: name.to_owned(),
            description: Some("d".to_owned()),
            input_schema: r#"{"type":"object"}"#.to_owned(),
            title: None,
            read_only: false,
            destructive: false,
            open_world: true,
        }
    }

    #[test]
    fn descriptor_from_spec_carries_schema_and_annotations() {
        let mut spec = polyc_llm::ToolSpec::new(
            "delete_file",
            "Delete a file.",
            serde_json::json!({ "type": "object", "properties": { "path": { "type": "string" } } }),
        );
        spec.title = Some("Delete a file".to_owned());
        spec.destructive = true;
        spec.open_world = true;
        let d = descriptor_from_spec(spec);
        assert_eq!(d.name, "delete_file");
        assert_eq!(d.description.as_deref(), Some("Delete a file."));
        assert_eq!(d.title.as_deref(), Some("Delete a file"));
        assert!(d.destructive);
        assert!(d.open_world);
        assert!(!d.read_only);
        // The schema round-trips through the string form.
        let schema: serde_json::Value = serde_json::from_str(&d.input_schema).unwrap();
        assert_eq!(schema["properties"]["path"]["type"], "string");
    }

    fn healthy_readiness(tools: Vec<ToolDescriptor>) -> ToolServiceReadiness {
        ToolServiceReadiness {
            healthy: true,
            available_tools: tools,
            message: None,
        }
    }

    fn failed_readiness() -> ToolServiceReadiness {
        ToolServiceReadiness {
            healthy: false,
            available_tools: Vec::new(),
            message: Some("dial failed".to_owned()),
        }
    }

    #[test]
    fn version_keyed_catalog_freezes_on_unchanged_version() {
        // Invariant 8: a same-version re-check must NOT rewrite the catalog,
        // even if a live re-list drifted.
        let current = ToolServiceStatus {
            observed_version: Some("1.0.0".to_owned()),
            available_tools: vec![descriptor("echo")],
            ..Default::default()
        };
        let drifted = healthy_readiness(vec![descriptor("echo"), descriptor("sneaky_new_tool")]);
        let kept = version_keyed_catalog(&current, "1.0.0", &drifted);
        assert_eq!(
            kept,
            vec![descriptor("echo")],
            "catalog frozen to the reviewed version"
        );
    }

    #[test]
    fn version_keyed_catalog_adopts_on_version_bump() {
        let current = ToolServiceStatus {
            observed_version: Some("1.0.0".to_owned()),
            available_tools: vec![descriptor("echo")],
            ..Default::default()
        };
        let fresh = vec![descriptor("echo"), descriptor("new_tool")];
        let adopted = version_keyed_catalog(&current, "2.0.0", &healthy_readiness(fresh.clone()));
        assert_eq!(
            adopted, fresh,
            "a version bump adopts the freshly listed catalog"
        );
    }

    #[test]
    fn version_keyed_catalog_adopts_on_first_check() {
        // No prior status: `observed_version` is None, so the first check adopts
        // the fresh list under whatever version the spec declares.
        let current = ToolServiceStatus::default();
        let fresh = vec![descriptor("echo")];
        assert_eq!(
            version_keyed_catalog(&current, "1.0.0", &healthy_readiness(fresh.clone())),
            fresh
        );
    }

    #[test]
    fn version_keyed_catalog_keeps_catalog_on_failed_same_version_recheck() {
        // #635 existence stickiness: a connector outage never changes the
        // answer to "what tools exist". The steady-state outage case — an
        // UNHEALTHY periodic re-check on the SAME version — must leave the
        // reviewed catalog untouched even though the failed check listed zero
        // tools; health drives advertisement (harness-side), never existence.
        let current = ToolServiceStatus {
            observed_version: Some("1.0.0".to_owned()),
            available_tools: vec![descriptor("echo")],
            healthy: true,
            ..Default::default()
        };
        assert_eq!(
            version_keyed_catalog(&current, "1.0.0", &failed_readiness()),
            vec![descriptor("echo")],
            "an outage must not remove tools from the catalog"
        );
    }

    #[test]
    fn version_keyed_catalog_freezes_prior_catalog_on_failed_check_during_version_bump() {
        // A version bump colliding with a transient connector outage must not
        // freeze an EMPTY catalog into the reflected status: the health check
        // failed, so its (empty) listing is not "the freshly listed tools" —
        // the prior catalog stays authoritative until a check actually
        // succeeds. This is the exact scenario the happy-path-only tests
        // above miss.
        let current = ToolServiceStatus {
            observed_version: Some("1.0.0".to_owned()),
            available_tools: vec![descriptor("echo")],
            healthy: true,
            ..Default::default()
        };
        let kept = version_keyed_catalog(&current, "2.0.0", &failed_readiness());
        assert_eq!(
            kept,
            vec![descriptor("echo")],
            "a failed check on a version bump must not wipe the prior catalog"
        );

        // Once the connector recovers, a SUCCESSFUL check at the same bumped
        // version adopts the real list.
        let recovered = healthy_readiness(vec![descriptor("echo"), descriptor("new_tool")]);
        let adopted = version_keyed_catalog(&current, "2.0.0", &recovered);
        assert_eq!(adopted, recovered.available_tools);
    }

    #[test]
    fn version_keyed_catalog_stays_empty_on_failed_first_check() {
        // First-ever check (no prior status) that fails: there is no prior
        // catalog to keep, so it correctly stays empty. `observed_version`
        // must not settle to the new version either (covered below), so a
        // later successful check still adopts rather than freezing.
        let current = ToolServiceStatus::default();
        assert_eq!(
            version_keyed_catalog(&current, "1.0.0", &failed_readiness()),
            Vec::<ToolDescriptor>::new()
        );
    }

    #[test]
    fn observed_version_settles_on_healthy_check() {
        let current = ToolServiceStatus::default();
        assert_eq!(
            observed_version_after_check(&current, "1.0.0", true),
            Some("1.0.0".to_owned())
        );
    }

    #[test]
    fn observed_version_does_not_settle_on_failed_version_bump() {
        // The bug: a health-check failure colliding with a version bump must
        // NOT settle `observed_version` to the new version — doing so makes
        // `check_due` treat the object as current, permanently freezing the
        // (empty) catalog `version_keyed_catalog` adopted for the failed
        // check instead of retrying at the fast, always-due cadence a
        // version mismatch forces.
        let current = ToolServiceStatus {
            observed_version: Some("1.0.0".to_owned()),
            ..Default::default()
        };
        assert_eq!(
            observed_version_after_check(&current, "2.0.0", false),
            Some("1.0.0".to_owned()),
            "keep the prior observed_version so check_due keeps forcing retries"
        );
    }

    #[test]
    fn observed_version_stays_unset_on_failed_first_check() {
        let current = ToolServiceStatus::default();
        assert_eq!(observed_version_after_check(&current, "1.0.0", false), None);
    }

    #[test]
    fn observed_version_settles_on_same_version_even_if_unhealthy() {
        // A same-version periodic re-check that happens to fail still settles
        // `observed_version` (it was already correct) — only a *bump*
        // colliding with a failure must hold back.
        let current = ToolServiceStatus {
            observed_version: Some("1.0.0".to_owned()),
            ..Default::default()
        };
        assert_eq!(
            observed_version_after_check(&current, "1.0.0", false),
            Some("1.0.0".to_owned())
        );
    }

    #[tokio::test]
    async fn health_check_empty_url_is_unhealthy_not_an_error() {
        let r = health_check("", None).await;
        assert!(!r.healthy);
        assert!(r.available_tools.is_empty());
        assert_eq!(r.message.as_deref(), Some("no remote URL configured"));
    }

    fn auth_for(secret: &str) -> Auth {
        Auth {
            bearer_secret_ref: BearerSecretRef {
                name: secret.to_owned(),
                key: "token".to_owned(),
            },
        }
    }

    /// The controller resolves the same author-supplied `bearerSecretRef` the
    /// control plane does, so the same refusal must hold here. Each reserved
    /// name — and the control-signing prefix — is refused rather than admitted,
    /// so [`resolve_bearer`] never reads the Secret and the reconcile never
    /// dials the author's URL.
    #[test]
    fn the_bearer_ref_screen_refuses_every_reserved_secret() {
        let wallet = format!(
            "{}{}",
            crate::reserved_secrets::CONTROL_KEY_SECRET_PREFIX,
            "ab".repeat(32)
        );
        let reserved = crate::reserved_secrets::RESERVED_SECRET_NAMES
            .iter()
            .map(|s| (*s).to_owned())
            .chain(std::iter::once(wallet));

        for secret in reserved {
            let auth = auth_for(&secret);
            let err = admissible_bearer_ref(Some(&auth))
                .expect_err("a reserved reference must be refused, not admitted");
            assert_eq!(err.secret, secret);
            assert!(
                err.to_string().contains(&secret),
                "the refusal must name the Secret it refused"
            );

            // The refusal is what the resource's author reads back in status:
            // unhealthy, no catalog, and the actionable message.
            let readiness = refused_readiness(&err);
            assert!(!readiness.healthy);
            assert!(readiness.available_tools.is_empty());
            assert_eq!(readiness.message.as_deref(), Some(err.to_string().as_str()));
        }
    }

    /// The other direction: an ordinary Secret name an author created for
    /// their own connector is admitted, reference intact, so the read step
    /// proceeds exactly as it did before the guard existed. No auth at all is
    /// likewise never a refusal.
    #[test]
    fn the_bearer_ref_screen_admits_an_author_chosen_secret() {
        let auth = auth_for("acme-mcp-token");
        let admitted = admissible_bearer_ref(Some(&auth))
            .expect("an ordinary reference is never refused")
            .expect("configured auth yields a reference to read");
        assert_eq!(admitted.name, "acme-mcp-token");
        assert_eq!(admitted.key, "token");

        assert!(
            admissible_bearer_ref(None)
                .expect("no auth is never a refusal")
                .is_none()
        );
    }

    /// The refusal is wired into the reconcile's health-check step: a
    /// `ToolService` naming reserved material settles into an unhealthy status
    /// carrying the refusal, and its catalog is frozen rather than replaced —
    /// the same treatment any failed check gets.
    #[test]
    fn a_refused_toolservice_settles_unhealthy_with_its_catalog_frozen() {
        let auth = auth_for(crate::reserved_secrets::STATE_ATTESTATION_SECRET);
        let err = admissible_bearer_ref(Some(&auth)).expect_err("refused");
        let readiness = refused_readiness(&err);

        let current = ToolServiceStatus {
            healthy: true,
            observed_version: Some("1.0.0".to_owned()),
            available_tools: vec![ToolDescriptor {
                name: "find".to_owned(),
                description: None,
                input_schema: String::new(),
                title: None,
                read_only: true,
                destructive: false,
                open_world: false,
            }],
            ..Default::default()
        };
        // A refusal on a version bump must not adopt an empty catalog, and must
        // not settle the new version — the check keeps retrying at the forced
        // cadence, exactly like any other failed check.
        let catalog = version_keyed_catalog(&current, "2.0.0", &readiness);
        assert_eq!(catalog, current.available_tools);
        assert_eq!(
            observed_version_after_check(&current, "2.0.0", readiness.healthy),
            Some("1.0.0".to_owned())
        );
    }

    #[tokio::test]
    async fn health_check_unreachable_url_is_unhealthy_not_an_error() {
        // A connection failure must distil to an unhealthy status, never panic
        // or hang the reconcile. (Reserved/unroutable host.)
        let r = health_check("http://127.0.0.1:1/mcp", None).await;
        assert!(!r.healthy);
        assert!(r.available_tools.is_empty());
        assert!(r.message.is_some());
    }
}