polyc-controller 2026.8.3

//! The `Workflow` custom resource (`polychrome.dev/v1alpha1`).
//!
//! A `Workflow` is the catalog's **multi-stage pipeline noun**: a dependency
//! graph of named *stages*, each of which the controller fans out into one
//! [`ServiceDefinition`](crate::ServiceDefinition) and **sequences by the
//! declared `dependsOn` edges** — a downstream stage's workload is not created
//! until every stage it depends on reports ready. The canonical shape is a
//! *dataset* stage upstream of the *service* stages that consume it: declare
//! the edge once, and the platform stands the dataset up first and the
//! consumers only after it is serving.
//!
//! This is the developer-portal catalog shape — a grouping of components and
//! data-source resources joined by a dependency relation — but realized on the
//! CRD substrate Polychrome already runs, so the relations drive real pods and
//! a signed log rather than the read-mostly metadata a portal settles for. A
//! stage's `kind` (`dataset` for a data source other stages consume, `service`
//! for a consumer) is provenance the agent reads; the `dependsOn` edges are
//! what the reconciler enforces.
//!
//! Same envelope discipline as the rest of the catalog (`ServiceDefinition`,
//! `ToolService`, `Conversation`): `kubectl apply` of one of these is the whole
//! pipeline API, and the controller reflects per-stage readiness into the
//! `status` subresource so an agent can parse each stage of the workload
//! (`status.stages[]`) without reading the fanned-out Deployments.

use std::collections::BTreeMap;

use kube::CustomResource;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};

/// Desired state of one multi-stage workflow.
///
/// The agent-chosen surface is deliberately the stage list and its `dependsOn`
/// edges; everything operational (security context, resource bounds, probes) is
/// stamped uniformly downstream by the `ServiceDefinition` reconciler, exactly
/// as it is for a single scaffolded service.
#[derive(CustomResource, Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
#[kube(
    group = "polychrome.dev",
    version = "v1alpha1",
    kind = "Workflow",
    namespaced,
    status = "WorkflowStatus",
    shortname = "wf",
    category = "polychrome",
    derive = "PartialEq",
    printcolumn = r#"{"name":"Ready","type":"boolean","jsonPath":".status.ready"}"#,
    printcolumn = r#"{"name":"Phase","type":"string","jsonPath":".status.phase"}"#,
    printcolumn = r#"{"name":"Stages","type":"integer","jsonPath":".status.stageCount"}"#,
    printcolumn = r#"{"name":"Age","type":"date","jsonPath":".metadata.creationTimestamp"}"#
)]
#[serde(rename_all = "camelCase")]
pub struct WorkflowSpec {
    /// Human-readable description of what this pipeline does.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    /// Entity reference of the owner that requested this workflow (catalog
    /// `{kind}:{namespace}/{name}` grammar, e.g. `"user:slack/U123"`). The
    /// reconciler itself does not interpret it — enforcement lives one layer
    /// up, in the `polyc-scaffold` MCP connector (`#789`), which stamps this
    /// field from the caller's own attributed identity on create (never a
    /// model-supplied value) and filters every list/get/delete by it. Direct
    /// `kubectl` access bypasses that filter, same as any other RBAC-gated
    /// resource.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub owner: Option<String>,
    /// The stages of the pipeline. Order in the list is not significant — the
    /// execution order is derived from the `dependsOn` edges (a topological
    /// sort), so the same workflow reconciles identically however the stages
    /// are listed. Must be non-empty and form a DAG; a cycle or a dangling
    /// `dependsOn` is surfaced in `status` and nothing is applied.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub stages: Vec<WorkflowStage>,
}

/// One stage of a workflow: a single deployable unit plus the edges that gate
/// it.
///
/// Mirrors [`ServiceDefinitionSpec`](crate::ServiceDefinitionSpec)'s
/// agent-chosen fields (`image`, `port`, `replicas`, `env`, `template`
/// lineage) and adds the two fields a pipeline needs: a `name` unique within
/// the workflow and the `dependsOn` upstreams that must be ready first.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct WorkflowStage {
    /// Stage name, unique within the workflow. DNS-1123-label-shaped: it is
    /// concatenated with the workflow name to mint the fanned-out
    /// `ServiceDefinition` (`{workflow}-{stage}`), so it must be a valid label
    /// fragment.
    pub name: String,
    /// Catalog kind of this stage, for provenance and agent display:
    /// `"dataset"` (a data source other stages consume) or `"service"` (a
    /// consumer). Informational only; the reconciler sequences on `dependsOn`,
    /// not on `kind`. `None` → `"service"`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub kind: Option<String>,
    /// Names of the stages this one depends on. The stage's workload is not
    /// created until every named upstream reports ready. Each name must match
    /// another stage in this workflow; a stage may not depend on itself, and
    /// the edges must not form a cycle.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub depends_on: Vec<String>,
    /// Human-readable description of this stage.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    /// Name of the catalog template this stage was stamped from (lineage, e.g.
    /// `"rust-dataset"`). Carried onto the fanned-out `ServiceDefinition`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub template: Option<String>,
    /// Container image the stage runs. Resolved (from a template, by the
    /// connector) before the workflow is applied — the controller deploys
    /// exactly this reference and never resolves templates itself.
    pub image: String,
    /// HTTP port the container listens on (also used for the readiness probe
    /// and the fronting `Service`). `None` → the `ServiceDefinition` default.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub port: Option<i32>,
    /// Desired replica count for this stage. `None` → the `ServiceDefinition`
    /// default.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub replicas: Option<i32>,
    /// Plain environment variables for the stage container. `SERVICE_NAME` is
    /// injected by the reconciler. Secrets do not belong here.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub env: BTreeMap<String, String>,
}

/// Observed state, written back to the `status` subresource by the controller.
#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct WorkflowStatus {
    /// `true` when every stage has reported ready.
    #[serde(default)]
    pub ready: bool,
    /// Lifecycle phase of the whole pipeline: `Pending` (nothing ready yet),
    /// `Progressing` (some stages ready, more to go), `Ready` (all stages
    /// ready), or `Degraded` (the spec is invalid — a cycle or a dangling
    /// `dependsOn` — and nothing was applied).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub phase: Option<String>,
    /// Number of stages in the workflow (printer-column convenience).
    #[serde(default)]
    pub stage_count: i32,
    /// Human-readable status detail (e.g. the validation error that put the
    /// workflow into `Degraded`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
    /// Per-stage observed state, in the workflow's topological order so an agent
    /// reads the pipeline front to back. Empty while the spec is invalid.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub stages: Vec<StageStatus>,
}

/// Observed state of a single stage — the unit an agent parses to follow the
/// workload.
#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq, Eq, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct StageStatus {
    /// The stage's name (matches [`WorkflowStage::name`]).
    pub name: String,
    /// Catalog kind echoed from the spec (`dataset` / `service`), so the status
    /// is self-describing and the per-stage shape an agent reads stays stable
    /// once the controller starts reporting status.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub kind: Option<String>,
    /// Catalog template lineage echoed from the spec, for the same reason as
    /// [`StageStatus::kind`].
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub template: Option<String>,
    /// Where the stage is in its lifecycle: `Blocked` (an upstream is not yet
    /// ready, so no workload has been created), `Pending` (workload created,
    /// not yet ready), or `Ready` (its `ServiceDefinition` reports ready).
    pub phase: String,
    /// `true` when the stage's `ServiceDefinition` reports ready.
    #[serde(default)]
    pub ready: bool,
    /// Name of the `ServiceDefinition` fanned out for this stage, once created
    /// (`{workflow}-{stage}`). `None` while the stage is still `Blocked`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub service_definition: Option<String>,
    /// The upstream stages this one waits on (echoed from the spec so the status
    /// is self-describing).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub depends_on: Vec<String>,
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]
    use super::*;
    use kube::CustomResourceExt;
    use serde_json::{Value, json};

    #[test]
    fn crd_identity_is_polychrome_workflow() {
        let crd = Workflow::crd();
        assert_eq!(crd.spec.group, "polychrome.dev");
        assert_eq!(crd.spec.names.kind, "Workflow");
        assert_eq!(crd.spec.names.plural, "workflows");
    }

    #[test]
    fn stage_round_trips_camel_case() {
        // The wire is camelCase; a typo'd rename would silently drop a field on
        // the way to/from etcd, so pin the exact JSON shape.
        let stage = WorkflowStage {
            name: "ingest".to_owned(),
            kind: Some("dataset".to_owned()),
            depends_on: vec!["source".to_owned()],
            description: Some("ingest the source feed".to_owned()),
            template: Some("rust-dataset".to_owned()),
            image: "example.test/dataset:1".to_owned(),
            port: Some(8080),
            replicas: Some(1),
            env: BTreeMap::from([("FEED".to_owned(), "main".to_owned())]),
        };
        let v = serde_json::to_value(&stage).unwrap();
        assert!(v.get("dependsOn").is_some(), "depends_on → dependsOn");
        assert_eq!(v["dependsOn"][0], "source");
        assert_eq!(serde_json::from_value::<WorkflowStage>(v).unwrap(), stage);
    }

    #[test]
    fn minimal_stage_needs_only_name_and_image() {
        // dependsOn defaults empty (a root stage), kind defaults to service.
        let stage: WorkflowStage =
            serde_json::from_value(json!({ "name": "api", "image": "img:1" })).unwrap();
        assert!(stage.depends_on.is_empty());
        assert!(stage.kind.is_none());
        // Defaults are omitted from the serialized form (a minimal stage stays tiny).
        let out = serde_json::to_value(&stage).unwrap();
        assert!(out.get("dependsOn").is_none());
        assert!(out.get("env").is_none());
    }

    #[test]
    fn status_stages_round_trip() {
        let status = WorkflowStatus {
            ready: false,
            phase: Some("Progressing".to_owned()),
            stage_count: 2,
            message: None,
            stages: vec![
                StageStatus {
                    name: "data".to_owned(),
                    kind: Some("dataset".to_owned()),
                    template: Some("rust-dataset".to_owned()),
                    phase: "Ready".to_owned(),
                    ready: true,
                    service_definition: Some("pipe-data".to_owned()),
                    depends_on: vec![],
                },
                StageStatus {
                    name: "api".to_owned(),
                    kind: Some("service".to_owned()),
                    template: None,
                    phase: "Pending".to_owned(),
                    ready: false,
                    service_definition: Some("pipe-api".to_owned()),
                    depends_on: vec!["data".to_owned()],
                },
            ],
        };
        let v = serde_json::to_value(&status).unwrap();
        assert_eq!(v["stageCount"], 2);
        assert_eq!(v["stages"][1]["serviceDefinition"], "pipe-api");
        assert_eq!(serde_json::from_value::<WorkflowStatus>(v).unwrap(), status);
    }

    #[test]
    fn minimal_workflow_spec_is_valid() {
        let spec: WorkflowSpec = serde_json::from_value(json!({
            "stages": [{ "name": "only", "image": "img:1" }]
        }))
        .unwrap();
        assert_eq!(spec.stages.len(), 1);
        let _: Value = serde_json::to_value(spec).unwrap();
    }
}