polyc-controller 2026.8.3

//! The execution-backend seam.
//!
//! Defines the [`ExecutionBackend`] trait + its readiness DTOs
//! ([`ClaimReadiness`], [`DialAddress`]), and the agent-sandbox implementation
//! [`SandboxClaimBackend`] (with its `SandboxClaim` builders) — the only
//! execution backend. The reconcile state machine that drives it lives in
//! [the reconcile module](mod@crate::reconcile).

use std::collections::BTreeMap;
use std::time::Duration;

use kube::{
    Api, Client, Resource, ResourceExt,
    api::{DeleteParams, Patch, PatchParams, Preconditions},
};
use polyc_k8s_types::sandboxclaim::{
    SandboxClaim, SandboxClaimAdditionalPodMetadata, SandboxClaimLifecycle,
    SandboxClaimLifecycleShutdownPolicy, SandboxClaimSandboxTemplateRef, SandboxClaimSpec,
};

use crate::conversation::Conversation;
use crate::reconcile::{Error, PENDING_POLL, SANDBOX_READY_POLL, ignore_not_found};

/// The pod annotation that carries a conversation's execution audience.
///
/// This key is load-bearing, and both planes read this one definition. The
/// controller writes it into `SandboxClaim.spec.additionalPodMetadata`. The
/// sandbox operator merges that metadata onto the adopted pod and updates the
/// live pod object. A Downward API volume then shows the annotation to the
/// harness, which latches the value for the pod's life.
///
/// An annotation can reach a pod that already runs. An environment variable
/// cannot. That difference is what lets a warm pod serve one conversation.
///
/// The `SandboxTemplate` must never set this key. The operator fails the whole
/// reconcile when the template already holds the key with another value. See
/// `manifests/components/per-conversation-harness/sandbox-template.yaml`.
pub const EXECUTION_AUDIENCE_ANNOTATION: &str = "polychrome.sh/execution-audience";

/// Readiness distilled from an execution unit's status, for mirroring into the
/// `Conversation` status. Pure-function output of [`claim_readiness`].
///
/// The backend-neutral readiness DTO returned by
/// [`ExecutionBackend::readiness`].
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ClaimReadiness {
    /// Whether the execution **unit object** exists at all — the `SandboxClaim`
    /// is present — *independent* of whether its harness is reachable yet.
    ///
    /// This is the gone-vs-transient discriminator for `unit_is_gone`: a unit
    /// whose pod is momentarily restarting still has `unit_present == true`
    /// (claim object lives on), so it is not mistaken for a deleted unit. Only a
    /// genuinely absent unit (no claim) reports `false`.
    pub unit_present: bool,
    /// Where to dial the harness, if known. `None` until an address is
    /// observed.
    pub address: Option<DialAddress>,
    /// `true` when the unit reports its harness is ready to serve.
    pub harness_ready: bool,
}

/// How the control plane reaches a conversation's harness: a bare pod IP,
/// dialed at the configured `harness_port` as `http://{ip}:{port}`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DialAddress {
    /// The `SandboxClaim`'s pod IP.
    PodIp(String),
}

impl DialAddress {
    /// Project to the persisted `(podIp, harnessEndpoint)` status fields.
    /// `harnessEndpoint` stays `None` — no backend populates it — kept as a
    /// pair rather than a single field because the `Conversation` CRD's
    /// status schema already ships both and CRDs are additive-only.
    #[must_use]
    pub fn to_status_fields(&self) -> (Option<String>, Option<String>) {
        match self {
            Self::PodIp(ip) => (Some(ip.clone()), None),
        }
    }
}

/// Distil a `SandboxClaim`'s status into [`ClaimReadiness`]. `None` (claim not
/// yet observed) is treated as not-ready. Pure.
#[must_use]
pub fn claim_readiness(claim: Option<&SandboxClaim>) -> ClaimReadiness {
    // The claim *object* existing is what marks the unit present — a claim whose
    // pod is mid-restart (no IP, no Ready condition yet) is present-but-not-ready,
    // NOT gone. `claim == None` is the only "gone" signal for agent-sandbox.
    let Some(claim) = claim else {
        return ClaimReadiness::default();
    };
    let status = claim.status.as_ref();
    let pod_ip = status
        .and_then(|s| s.sandbox.as_ref())
        .and_then(|s| s.pod_i_ps.as_ref())
        .and_then(|ips| ips.first().cloned());
    let harness_ready = status
        .and_then(|s| s.conditions.as_ref())
        .is_some_and(|cs| cs.iter().any(|c| c.type_ == "Ready" && c.status == "True"));
    ClaimReadiness {
        unit_present: true,
        // agent-sandbox is dialed by pod IP; no router endpoint.
        address: pod_ip.map(DialAddress::PodIp),
        harness_ready,
    }
}

/// The pluggable execution mechanism the reconciler drives each
/// [`Conversation`] onto.
///
/// Today the only execution unit is an agent-sandbox `SandboxClaim`
/// ([`SandboxClaimBackend`]), and it's the only implementation of this trait.
/// The trait still lifts the three IO steps the reconciler performs against
/// that unit — create, read-readiness, delete — behind a seam so a future
/// backend could slot in *without* a rewrite of the
/// [reconcile module](mod@crate::reconcile). The backend-agnostic bits (the
/// finalizer patch and the status patch on the `Conversation` CR itself) stay
/// in [reconcile](mod@crate::reconcile): they touch the `Conversation`, not the
/// execution unit.
///
/// The watch wiring is *not* part of this trait — it is per-backend and lives
/// in [`run`](crate::run) (today `.owns(SandboxClaim)`). The trait abstracts
/// only the reconcile-time IO.
#[async_trait::async_trait]
pub trait ExecutionBackend: Send + Sync {
    /// SSA-apply the execution unit for `conv`, named `name`, from `template`,
    /// owned by `conv`.
    ///
    /// # Errors
    ///
    /// Returns [`Error`] if the apply fails.
    async fn ensure(
        &self,
        conv: &Conversation,
        name: &str,
        template: &str,
        ns: &str,
    ) -> Result<(), Error>;
    /// Read readiness (pod IP + ready) of the named unit; the `Default`
    /// (not-ready) value if not yet observed.
    ///
    /// # Errors
    ///
    /// Returns [`Error`] if the status read fails.
    async fn readiness(&self, name: &str, ns: &str) -> Result<ClaimReadiness, Error>;
    /// Idempotently delete the named unit (a `404` ⇒ already gone ⇒ `Ok`).
    ///
    /// # Errors
    ///
    /// Returns [`Error`] if the delete fails for any reason other than `404`.
    async fn teardown(&self, owner: &Conversation, name: &str, ns: &str) -> Result<(), Error>;
    /// Short identifier for logs/metrics, e.g. `"sandboxclaim"`.
    fn kind(&self) -> &'static str;
    /// How long to wait before re-reading readiness once a unit is *steady-state
    /// `Ready`*. While a unit is still `Pending` the reconciler always polls
    /// tightly (`PENDING_POLL`); this only governs the already-`Ready` case.
    ///
    /// A backend with its own reactive watch (agent-sandbox `.owns(SandboxClaim)`)
    /// learns of changes from the watch and can poll slowly, so it overrides this
    /// to a long interval to shed steady-state read load. A backend with no watch
    /// keeps the default tight cadence so an out-of-band loss is still detected
    /// within it. Defaults to the same tight `PENDING_POLL` cadence.
    fn ready_poll_interval(&self) -> Duration {
        PENDING_POLL
    }
}

/// The agent-sandbox execution backend: each [`Conversation`] maps to one
/// `SandboxClaim`. This is today's only [`ExecutionBackend`].
pub struct SandboxClaimBackend {
    /// Kube client used to build per-namespace `SandboxClaim` APIs.
    client: Client,
}

impl SandboxClaimBackend {
    /// Construct a backend over `client`.
    #[must_use]
    pub const fn new(client: Client) -> Self {
        Self { client }
    }

    /// The `SandboxClaim` API scoped to `ns`.
    fn claims(&self, ns: &str) -> Api<SandboxClaim> {
        Api::namespaced(self.client.clone(), ns)
    }
}

#[async_trait::async_trait]
impl ExecutionBackend for SandboxClaimBackend {
    async fn ensure(
        &self,
        conv: &Conversation,
        name: &str,
        template: &str,
        ns: &str,
    ) -> Result<(), Error> {
        let claim = build_sandbox_claim(conv, name, template, ns);
        // `.force()`: issue #802's drift healing re-applies this SSA patch
        // every `SyncStatus` pass (`reconcile::sync_execution_unit`), not
        // just at creation. An out-of-band edit (`kubectl edit`/`kubectl
        // patch`) takes field ownership under its OWN manager, so a
        // non-forced re-apply from `polychrome.dev/controller` would 409 on
        // that conflict instead of healing it — the drift would persist
        // forever (every reconcile retrying the same conflict). Forcing
        // makes us win every such conflict, restoring the desired spec.
        // (`lease.rs` faces the analogous single-writer problem but resolves
        // it differently — via `resourceVersion`-gated CREATE/REPLACE rather
        // than a forced apply — because a Lease has exactly one legitimate
        // writer at a time and forcing there would let a stale holder
        // clobber the current one; a SandboxClaim spec has exactly one
        // desired state that this controller alone owns, so forcing here
        // only ever overwrites drift, never a legitimate concurrent writer.)
        let pp = PatchParams::apply("polychrome.dev/controller").force();
        self.claims(ns)
            .patch(name, &pp, &Patch::Apply(&claim))
            .await?;
        tracing::info!(claim = name, template, ns, "applied sandbox claim");
        Ok(())
    }

    async fn readiness(&self, name: &str, ns: &str) -> Result<ClaimReadiness, Error> {
        let claim = self.claims(ns).get_opt(name).await?;
        Ok(claim_readiness(claim.as_ref()))
    }

    async fn teardown(&self, owner: &Conversation, name: &str, ns: &str) -> Result<(), Error> {
        let claims = self.claims(ns);
        let Some(claim) = claims.get_opt(name).await? else {
            return Ok(());
        };
        let Some(params) = delete_params_for_owned_claim(owner, &claim) else {
            tracing::warn!(
                claim = name,
                conversation = %owner.name_any(),
                "refusing to delete an execution unit not owned by this conversation incarnation"
            );
            return Ok(());
        };
        ignore_not_found(claims.delete(name, &params).await)
    }

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

    fn ready_poll_interval(&self) -> Duration {
        // The `.owns(SandboxClaim)` watch reports changes reactively; the
        // steady-state poll is only a backstop, so keep it infrequent.
        SANDBOX_READY_POLL
    }
}

/// Builds a target-side compare-and-swap delete for an observed owned claim.
///
/// A stale reconcile for an earlier `Conversation` incarnation may retain the
/// same object name after Kubernetes has recreated both parent and child. The
/// owner UID refuses that cross-incarnation delete. The child's own UID and
/// resource version then pin the delete to the exact object this pass read.
fn delete_params_for_owned_claim(
    owner: &Conversation,
    claim: &SandboxClaim,
) -> Option<DeleteParams> {
    let owner_uid = owner.uid()?;
    let has_matching_owner = claim
        .metadata
        .owner_references
        .iter()
        .flatten()
        .any(|reference| {
            reference.controller == Some(true)
                && reference.kind == "Conversation"
                && reference.uid == owner_uid
        });
    if !has_matching_owner {
        return None;
    }
    let claim_uid = claim.uid()?;
    let claim_resource_version = claim.resource_version()?;
    Some(DeleteParams {
        preconditions: Some(Preconditions {
            uid: Some(claim_uid),
            resource_version: Some(claim_resource_version),
        }),
        ..DeleteParams::default()
    })
}

/// Build the `SandboxClaim` to apply for `conv`. Encapsulates the parent-link
/// metadata for child conversations (handoff): an info log + a
/// `polychrome.dev/parent-conversation` label discoverable via
/// `kubectl get sandboxclaim -l ...`.
///
/// The cross-partition handoff *event* (signed `Handoff` written to the
/// parent's eventlog) is the responsibility of the control plane
/// (`AgentSvc::emit_handoff`); the controller deliberately stays free of any
/// eventlog client and only carries the parent linkage via Kubernetes
/// metadata. This split keeps the reconciler hermetic and unit-testable.
fn build_sandbox_claim(
    conv: &Conversation,
    claim_name: &str,
    template: &str,
    ns: &str,
) -> SandboxClaim {
    if let Some(parent) = conv.spec.parent_conversation_id.as_deref() {
        tracing::info!(
            child_conversation = %claim_name,
            parent_conversation = parent,
            "creating sandbox claim for child conversation (handoff)"
        );
    }
    let mut claim = SandboxClaim::new(
        claim_name,
        SandboxClaimSpec {
            sandbox_template_ref: SandboxClaimSandboxTemplateRef {
                name: template.to_owned(),
            },
            // The audience travels as pod metadata, never as an environment
            // variable. The operator merges this map onto the pod it adopts and
            // then updates the live pod object, so a pod that already runs
            // learns which conversation it serves. An environment variable
            // could not do that: Kubernetes cannot change the environment of a
            // running container.
            additional_pod_metadata: Some(SandboxClaimAdditionalPodMetadata {
                annotations: Some(BTreeMap::from([(
                    EXECUTION_AUDIENCE_ANNOTATION.to_owned(),
                    claim_name.to_owned(),
                )])),
                labels: None,
            }),
            // Execution takes no cluster configuration input, and claim
            // environment injection also blocks warm-pod adoption outright.
            env: None,
            lifecycle: Some(lifecycle_for(conv)),
            // Left unset on purpose, so the CRD default `default` applies.
            //
            // `default` is a POLICY value, not the name of a pool object. It
            // tells the operator to adopt from whichever pool serves this
            // claim's `sandboxTemplateRef`. Do not go looking for a pool called
            // `default`: the object this component ships is
            // `polychrome-harness-warm`
            // (`manifests/components/per-conversation-harness/warmpool.yaml`),
            // and it is matched by template, not by this string.
            //
            // Adoption saves about 86 seconds of a 104 second pod start. Name a
            // value here only to opt this claim out of pool adoption.
            warmpool: None,
        },
    );
    claim.metadata.namespace = Some(ns.to_owned());
    claim.metadata.owner_references = conv.controller_owner_ref(&()).map(|r| vec![r]);
    if let Some(parent) = conv.spec.parent_conversation_id.as_deref() {
        let labels = claim.metadata.labels.get_or_insert_with(Default::default);
        labels.insert(
            "polychrome.dev/parent-conversation".to_owned(),
            parent.to_owned(),
        );
    }
    claim
}

/// Derive the `SandboxClaim.spec.lifecycle` for `conv`.
///
/// **Defence-in-depth (review item A1).** The control plane tears a harness
/// down via the `closed` → [`ReconcileAction::Cleanup`] path, but if it ever
/// misses that signal (crash, lost watch event) the pod would leak. Setting
/// `lifecycle` makes agent-sandbox reap the sandbox on its own, so an idle or
/// finished sandbox is never orphaned regardless of the control plane.
///
/// Field semantics, per the vendored agent-sandbox v0.4.6 CRD
/// (`crates/k8s-types/crds/extensions.agents.x-k8s.io_sandboxclaims.yaml`,
/// `spec.lifecycle`):
///
/// * `shutdownPolicy` — `enum [Delete, DeleteForeground, Retain]`,
///   **`default: Retain`**. What agent-sandbox does to the underlying
///   `Sandbox` when the claim's lifecycle fires. We override the `Retain`
///   default with `Delete`: a Retained sandbox would outlive its claim and
///   leak, which is the exact failure A1 guards against.
/// * `shutdownTime` — `format: date-time` (an RFC3339 *absolute* timestamp).
///   The instant at which the sandbox should shut down. Unsuitable here: it
///   needs a clock to compute (`now + idle_timeout`), and [`plan`] /
///   `build_sandbox_claim` are deliberately pure (no clock, fully testable).
/// * `ttlSecondsAfterFinished` — `format: int32`, `minimum: 0`. Reap the
///   sandbox this many seconds after the workload finishes. This is a
///   *relative* duration, needs no clock, and matches our intent ("reap N
///   seconds after idle/finished") — so we map `idle_timeout_seconds` here.
///
/// The `u32 → i32` cast is guarded: an `idle_timeout_seconds` larger than
/// `i32::MAX` saturates to `i32::MAX` rather than wrapping negative (the CRD
/// requires `minimum: 0`).
fn lifecycle_for(conv: &Conversation) -> SandboxClaimLifecycle {
    SandboxClaimLifecycle {
        shutdown_policy: Some(SandboxClaimLifecycleShutdownPolicy::Delete),
        shutdown_time: None,
        ttl_seconds_after_finished: Some(
            i32::try_from(conv.spec.idle_timeout_seconds).unwrap_or(i32::MAX),
        ),
    }
}

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

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

    #[test]
    fn claim_binds_the_audience_as_pod_metadata_and_keeps_pool_adoption() {
        let owner = Conversation::new(
            "conversation-a",
            crate::conversation::ConversationSpec {
                model: String::new(),
                principal_ref: "persona-1".to_owned(),
                idle_timeout_seconds: 300,
                tools_enabled: Vec::new(),
                tools_disabled: Vec::new(),
                parent_conversation_id: None,
                agent_id: None,
            },
        );
        let claim = build_sandbox_claim(&owner, "conversation-a", TPL, "polychrome");

        // The annotation carries the audience, so the operator can bind a pod
        // that already runs.
        assert_eq!(
            claim
                .spec
                .additional_pod_metadata
                .as_ref()
                .and_then(|metadata| metadata.annotations.as_ref())
                .and_then(|annotations| annotations
                    .get("polychrome.sh/execution-audience")
                    .map(String::as_str)),
            Some("conversation-a"),
            "the claim must deliver the audience as pod metadata"
        );

        // The operator refuses to adopt a warm pod for any claim that sets
        // environment variables, whatever the variables say.
        assert_eq!(
            claim.spec.env, None,
            "claim environment injection blocks warm-pod adoption"
        );

        // An unset field selects the CRD default pool, which is the pool this
        // component ships. Any value here would opt the claim out of adoption.
        assert_eq!(
            claim.spec.warmpool, None,
            "the claim must accept the default warm pool"
        );
    }

    #[test]
    fn claim_readiness_none_is_not_ready() {
        assert_eq!(claim_readiness(None), ClaimReadiness::default());
    }

    #[test]
    fn claim_readiness_reads_pod_ip_and_ready_condition() {
        use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, Time};
        use polyc_k8s_types::sandboxclaim::{
            SandboxClaimSpec, SandboxClaimStatus, SandboxClaimStatusSandbox,
        };

        let mut claim = SandboxClaim::new(
            "c1",
            SandboxClaimSpec {
                sandbox_template_ref: super::SandboxClaimSandboxTemplateRef {
                    name: TPL.to_owned(),
                },
                additional_pod_metadata: None,
                env: None,
                lifecycle: None,
                warmpool: None,
            },
        );
        claim.status = Some(SandboxClaimStatus {
            conditions: Some(vec![Condition {
                type_: "Ready".to_owned(),
                status: "True".to_owned(),
                reason: "PodRunning".to_owned(),
                message: String::new(),
                observed_generation: None,
                last_transition_time: Time("2026-05-27T00:00:00Z".parse().unwrap()),
            }]),
            sandbox: Some(SandboxClaimStatusSandbox {
                name: Some("c1-sbx".to_owned()),
                pod_i_ps: Some(vec!["10.4.2.7".to_owned(), "fd00::7".to_owned()]),
            }),
        });

        assert_eq!(
            claim_readiness(Some(&claim)),
            ClaimReadiness {
                unit_present: true,
                address: Some(DialAddress::PodIp("10.4.2.7".to_owned())),
                harness_ready: true,
            }
        );
    }

    #[test]
    fn claim_readiness_not_ready_without_ready_condition() {
        use polyc_k8s_types::sandboxclaim::{
            SandboxClaimSpec, SandboxClaimStatus, SandboxClaimStatusSandbox,
        };

        let mut claim = SandboxClaim::new(
            "c1",
            SandboxClaimSpec {
                sandbox_template_ref: super::SandboxClaimSandboxTemplateRef {
                    name: TPL.to_owned(),
                },
                additional_pod_metadata: None,
                env: None,
                lifecycle: None,
                warmpool: None,
            },
        );
        claim.status = Some(SandboxClaimStatus {
            conditions: None,
            sandbox: Some(SandboxClaimStatusSandbox {
                name: None,
                pod_i_ps: Some(vec!["10.4.2.7".to_owned()]),
            }),
        });

        assert_eq!(
            claim_readiness(Some(&claim)),
            ClaimReadiness {
                unit_present: true,
                address: Some(DialAddress::PodIp("10.4.2.7".to_owned())),
                harness_ready: false,
            }
        );
    }

    #[test]
    fn claim_delete_is_pinned_to_the_observed_owned_incarnation() {
        let mut owner = Conversation::new(
            "c1",
            crate::conversation::ConversationSpec {
                model: String::new(),
                principal_ref: "persona-1".to_owned(),
                idle_timeout_seconds: 300,
                tools_enabled: Vec::new(),
                tools_disabled: Vec::new(),
                parent_conversation_id: None,
                agent_id: None,
            },
        );
        owner.metadata.uid = Some("conversation-uid".to_owned());
        let mut claim = build_sandbox_claim(&owner, "c1", TPL, "polychrome");
        claim.metadata.uid = Some("claim-uid".to_owned());
        claim.metadata.resource_version = Some("claim-rv".to_owned());

        let params = delete_params_for_owned_claim(&owner, &claim).expect("owned live claim");
        let preconditions = params.preconditions.expect("delete preconditions");
        assert_eq!(preconditions.uid.as_deref(), Some("claim-uid"));
        assert_eq!(preconditions.resource_version.as_deref(), Some("claim-rv"));

        owner.metadata.uid = Some("replacement-conversation".to_owned());
        assert!(
            delete_params_for_owned_claim(&owner, &claim).is_none(),
            "a stale reconcile must not delete a replacement owner's claim"
        );
    }
}