polyc-controller 2026.9.0

//! Reconciler: validate each [`Routine`] spec and report readiness in
//! `status`.
//!
//! Every published routine carries a prompt payload (the `#1591` pivot
//! dissolved the `#1488` reshape's tagged union — fixed content was its sole
//! variant — into a bare prompt struct) and fires in-process off the control
//! plane's scheduler, so this reconciler owns no children of its own kind at
//! all. Its only remaining effect besides the status patch is a one-time
//! migration cleanup: a valid spec's `Apply` branch actively deletes the
//! per-routine trigger `ServiceAccount` and `CronJob` a PRE-#1371 reconcile
//! may have synthesized, so no dormant cron half keeps firing a routine
//! alongside the in-process scheduler. Idempotent against a routine that
//! never had either (a fresh deploy deletes nothing).
//!
//! Same split as the other reconcilers: the *decision* is the pure [`plan`]
//! function (built on the pure [`validate_spec`]), the *effect* is
//! [`reconcile`](fn@reconcile).

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

use futures::StreamExt;
use k8s_openapi::api::batch::v1::CronJob;
use k8s_openapi::api::core::v1::ServiceAccount;
use kube::{
    Api, Client, Resource, ResourceExt,
    api::{DeleteParams, Patch, PatchParams, Preconditions},
    runtime::{
        controller::{Action, Controller},
        watcher,
    },
};
use serde_json::json;

use crate::{
    cron_dow::normalize_cron_dow,
    fanout::{self},
    routine::{
        Routine, RoutinePayload, RoutineSchedule, RoutineSpec, RoutineStatus, RoutineSuspend,
    },
};

/// Errors surfaced by the `Routine` 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("routine has no namespace")]
    NoNamespace,
}

/// What a single reconcile pass should do for a [`Routine`]. Pure output of
/// [`plan`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RoutineAction {
    /// Spec is valid. Reflect `Ready`.
    Apply,
    /// Spec is invalid (the payload is the specific reason). Reflect
    /// `Degraded` and apply nothing — only a spec edit can fix it.
    Invalid(String),
    /// Object is terminating. Nothing to do (this reconciler owns no
    /// downstream resources, so there is nothing to garbage-collect either).
    Noop,
}

/// Decide what to do for `routine`. Pure: no IO, no clock, fully testable.
#[must_use]
pub fn plan(routine: &Routine) -> RoutineAction {
    if routine.meta().deletion_timestamp.is_some() {
        return RoutineAction::Noop;
    }
    match validate_spec(&routine.spec) {
        Err(reason) => RoutineAction::Invalid(reason),
        Ok(()) => RoutineAction::Apply,
    }
}

/// Validate the structured `schedule`: a `cron` variant's expression parses
/// as a standard five-field cron expression and its timezone (if set) is a
/// known IANA zone name; a `once` variant's `at` is a NAIVE local date and
/// time (no offset, no `Z`) whose `timezone` (if set) is a known IANA zone
/// name and in which `at` names a real instant (not a spring-forward gap).
///
/// The expression is checked through [`normalize_cron_dow`] — the same
/// translation [`crate::routine_next_fire`] parses through — so an
/// expression admission accepts is guaranteed to also be one the scheduler
/// can interpret under the same standard-cron day-of-week convention (#1644).
///
/// # Errors
///
/// Returns a human-readable reason for the first rule the schedule violates.
fn validate_schedule(schedule: &RoutineSchedule) -> Result<(), String> {
    match schedule {
        RoutineSchedule::Cron {
            expression,
            timezone,
        } => {
            if normalize_cron_dow(expression)
                .parse::<saffron::Cron>()
                .is_err()
            {
                return Err(format!(
                    "schedule.expression `{expression}` is not a valid cron expression"
                ));
            }
            if let Some(tz) = timezone
                && tz.parse::<chrono_tz::Tz>().is_err()
            {
                return Err(format!(
                    "schedule.timezone `{tz}` is not a known IANA zone name"
                ));
            }
            Ok(())
        }
        RoutineSchedule::Once { at, timezone } => {
            if at.parse::<chrono::DateTime<chrono::Utc>>().is_ok() {
                return Err(format!(
                    "schedule.at `{at}` must be a local date and time with no UTC offset or \
                     `Z` — set schedule.timezone instead"
                ));
            }
            let Ok(naive) = at.parse::<chrono::NaiveDateTime>() else {
                return Err(format!(
                    "schedule.at `{at}` is not a valid local date and time \
                     (e.g. `2026-07-20T09:00:00`)"
                ));
            };
            let Some(tz) = timezone else {
                return Ok(());
            };
            let Ok(zone) = tz.parse::<chrono_tz::Tz>() else {
                return Err(format!(
                    "schedule.timezone `{tz}` is not a known IANA zone name"
                ));
            };
            if crate::routine_next_fire::local_realizations(zone, naive).is_empty() {
                return Err(format!(
                    "schedule.at `{at}` does not exist in schedule.timezone `{tz}` (it falls \
                     inside a spring-forward gap)"
                ));
            }
            Ok(())
        }
    }
}

/// Validate `suspend`'s pause metadata, if set: a non-empty `pausedBy` and a
/// `pausedAt` that parses as RFC3339. Absent (the routine is active) is
/// always valid — `suspend` is additive (#1493), so admission never requires
/// it.
///
/// # Errors
///
/// Returns a human-readable reason for the first rule the metadata violates.
fn validate_suspend(suspend: Option<&RoutineSuspend>) -> Result<(), String> {
    let Some(suspend) = suspend else {
        return Ok(());
    };
    if suspend.paused_by.trim().is_empty() {
        return Err("suspend.pausedBy must not be empty".to_owned());
    }
    if suspend
        .paused_at
        .parse::<chrono::DateTime<chrono::Utc>>()
        .is_err()
    {
        return Err(format!(
            "suspend.pausedAt `{}` is not a valid RFC3339 instant",
            suspend.paused_at
        ));
    }
    Ok(())
}

/// Validate the payload: the prompt text run at fire time must be non-empty
/// — a routine with nothing to say has no reason to run unattended.
///
/// # Errors
///
/// Returns a human-readable reason if the prompt is empty (or all
/// whitespace).
fn validate_payload(payload: &RoutinePayload) -> Result<(), String> {
    if payload.prompt.trim().is_empty() {
        return Err("payload.prompt must not be empty".to_owned());
    }
    Ok(())
}

/// Validate a routine's spec.
///
/// Checks, in order: the structured `schedule` (`validate_schedule` — both
/// the `cron` and `once` variants); `suspend`'s pause metadata, if set
/// (`validate_suspend`, #1493); and the `payload`'s prompt text
/// (`validate_payload`). Pure.
///
/// `scope` needs no runtime check of its own (#1802): the wire enum names
/// exactly [`crate::routine::RoutineScope::Public`] and
/// [`crate::routine::RoutineScope::Private`], so an old reserved value
/// (`instance`/`persona`/`shared`) is rejected before this validator ever
/// runs — the same structural-schema enforcement the API server applies at
/// `kubectl apply`. The create-by-chat path (`routine_intent::compile_spec`)
/// catches the same class of bad value even earlier, with a purpose-built
/// refusal naming the two valid choices, rather than relying on the raw
/// deserialize error.
///
/// # Errors
///
/// Returns a human-readable reason for the first rule the spec violates.
pub fn validate_spec(spec: &RoutineSpec) -> Result<(), String> {
    validate_schedule(&spec.schedule)?;
    validate_suspend(spec.suspend.as_ref())?;
    validate_payload(&spec.payload)?;
    Ok(())
}

/// The name of the per-routine trigger `ServiceAccount` a PRE-#1371 reconcile
/// may have provisioned — a routine named `standup` got `standup-trigger`.
/// [`reconcile`]'s `Apply` branch deletes it if present; nothing creates one
/// anymore.
fn trigger_service_account_name(routine: &Routine) -> String {
    format!("{}-trigger", routine.name_any())
}

/// Shared reconcile context.
pub struct Context {
    /// Kube client for the migration-cleanup deletes and the status patch.
    pub client: Client,
}

/// Reconcile one [`Routine`] by executing the [`plan`].
///
/// # Errors
///
/// Returns [`Error`] if the migration-cleanup deletes or the status patch
/// fail, or the object lacks a namespace.
#[tracing::instrument(skip_all, fields(routine = %routine.name_any()))]
pub async fn reconcile(routine: Arc<Routine>, ctx: Arc<Context>) -> Result<Action, Error> {
    let ns = routine.namespace().ok_or(Error::NoNamespace)?;
    match plan(&routine) {
        RoutineAction::Noop => Ok(Action::await_change()),
        RoutineAction::Invalid(reason) => {
            sync_status(&ctx.client, &ns, &routine, false, "Degraded", Some(&reason)).await?;
            // Nothing is owned in this branch (a spec that never validated
            // never got its children applied), so only a spec edit (a fresh
            // watch event) can move the routine out of Degraded.
            Ok(Action::await_change())
        }
        RoutineAction::Apply => {
            // Every class fires in-process now (#1369 fixedContent, #1371
            // enrolled) — this reconciler owns no children of its own kind.
            // Actively delete any trigger `ServiceAccount`/`CronJob` a
            // PRE-#1371 reconcile synthesized (no dormant cron half left
            // firing a routine alongside the in-process scheduler);
            // idempotent against a routine that never had either.
            let service_accounts: Api<ServiceAccount> = Api::namespaced(ctx.client.clone(), &ns);
            let service_account_name = trigger_service_account_name(&routine);
            if let Some(service_account) = service_accounts.get_opt(&service_account_name).await? {
                if let Some(params) = legacy_delete_params(&routine, &service_account) {
                    ignore_not_found(
                        service_accounts
                            .delete(&service_account_name, &params)
                            .await,
                    )?;
                } else {
                    tracing::warn!(
                        resource = %service_account_name,
                        "refusing to delete a legacy ServiceAccount not owned by this Routine incarnation"
                    );
                }
            }
            let cronjobs: Api<CronJob> = Api::namespaced(ctx.client.clone(), &ns);
            let cronjob_name = routine.name_any();
            if let Some(cronjob) = cronjobs.get_opt(&cronjob_name).await? {
                if let Some(params) = legacy_delete_params(&routine, &cronjob) {
                    ignore_not_found(cronjobs.delete(&cronjob_name, &params).await)?;
                } else {
                    tracing::warn!(
                        resource = %cronjob_name,
                        "refusing to delete a legacy CronJob not owned by this Routine incarnation"
                    );
                }
            }
            sync_status(&ctx.client, &ns, &routine, true, "Ready", None).await?;
            // Nothing is owned — a fresh spec edit (the next watch event) is
            // the only thing that could reintroduce a need for cleanup, so
            // there is nothing here for a periodic requeue to catch that the
            // watch wouldn't already.
            Ok(Action::await_change())
        }
    }
}

/// Pins a legacy-child delete to the observed child and Routine incarnation.
fn legacy_delete_params<K>(routine: &Routine, child: &K) -> Option<DeleteParams>
where
    K: ResourceExt,
{
    let routine_uid = routine.uid()?;
    let owned = child
        .meta()
        .owner_references
        .iter()
        .flatten()
        .any(|reference| {
            reference.controller == Some(true)
                && reference.kind == "Routine"
                && reference.uid == routine_uid
        });
    if !owned {
        return None;
    }
    let child_uid = child.uid()?;
    let child_resource_version = child.resource_version()?;
    Some(DeleteParams {
        preconditions: Some(Preconditions {
            uid: Some(child_uid),
            resource_version: Some(child_resource_version),
        }),
        ..DeleteParams::default()
    })
}

/// Delete-if-exists: a 404 from the delete call is success (nothing to clean
/// up), any other error propagates. Mirrors
/// `crate::reconcile::ignore_not_found`'s idiom, kept local since that one
/// returns `crate::reconcile::Error`, not this module's own [`Error`].
fn ignore_not_found<T>(res: Result<T, kube::Error>) -> Result<(), Error> {
    match res {
        Ok(_) => Ok(()),
        Err(kube::Error::Api(e)) if e.code == 404 => Ok(()),
        Err(e) => Err(Error::Kube(e)),
    }
}

/// Write the routine's `status` subresource, but only when something
/// changed, to avoid a status-write → watch → reconcile churn loop.
async fn sync_status(
    client: &Client,
    ns: &str,
    routine: &Routine,
    ready: bool,
    phase: &str,
    message: Option<&str>,
) -> Result<(), Error> {
    let default = RoutineStatus::default();
    let current = routine.status.as_ref().unwrap_or(&default);
    let unchanged = current.ready == ready
        && current.phase.as_deref() == Some(phase)
        && current.message.as_deref() == message;
    if unchanged {
        return Ok(());
    }
    let routines: Api<Routine> = Api::namespaced(client.clone(), ns);
    let status = json!({ "status": {
        "ready": ready,
        "phase": phase,
        "message": message,
    } });
    routines
        .patch_status(
            &routine.name_any(),
            &PatchParams::default(),
            &Patch::Merge(&status),
        )
        .await?;
    tracing::info!(%phase, ready, "synced routine status");
    Ok(())
}

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

/// Run the `Routine` controller over `namespace` until its watch streams end.
///
/// `namespace` is the control plane's operating namespace — the same
/// namespace `Agent` and `ToolService` CRs live in.
///
/// `client` is the unary (#785-bounded) client — every reconcile-time
/// get/list/patch/delete call, via `Context`, rides it. `watch_client` has no
/// read timeout and backs only the `Routine` watch `Api` handle passed to
/// `Controller::new` below (see [`crate::reconcile::run`]'s doc comment for
/// why the two must not cross). This reconciler owns no children of its own
/// kind (#1371), so there is nothing left to `.owns()`.
///
/// # Errors
///
/// Returns [`Error`] only on fatal setup failure; per-item errors go through
/// [`error_policy`].
pub async fn run_routine(
    client: Client,
    watch_client: Client,
    namespace: &str,
) -> Result<(), Error> {
    let routines: Api<Routine> = Api::namespaced(watch_client, namespace);

    // Pre-flight, same as the other fan-out reconcilers: stay dormant until
    // the watch is serveable instead of error-looping.
    fanout::await_watchable(&routines, namespace, "Routine").await;

    let ctx = Arc::new(Context { client });

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

#[cfg(test)]
mod tests;