deploy/kubernetes: dependency-light reconcile core (alpha) (#4853)

Adds Reconcile(desired, observed) — the pure decision an operator's reconcile loop runs: given a desired Agent/Service/Flow resource and the observed cluster state, it returns the one action to converge (create / update / noop) plus Ready/Error status conditions. No controller-runtime, no client-go: the decision is a pure function of desired + observed, so it's fully unit-testable without a cluster. A future operator binary supplies Observed from the live cluster and applies the Action; only that adapter needs the Kubernetes client — keeping the heavy dependency out of the core module. Covers #4842 (Option B). Tests: create-when-absent, noop-when-matched-and- ready, update-on-drift, progressing-when-under-replicated, error-on-invalid -spec. Claude-Session: https://claude.ai/code/session_01CmdEY7pYmV5zzwCjNJ4ykL Co-authored-by: Claude <noreply@anthropic.com>
2026-07-15 15:12:20 +01:00
parent 7e2346b8c8
commit 1250d33f86
4 changed files with 201 additions and 2 deletions
@@ -24,6 +24,7 @@ below is kept current between tags and rolled into the next version when it ship
 - **CLI input resume for agent runs** — the CLI can resume agent runs that require additional user input. (`cmd/micro/`, `agent/`)
 - **A2A inbound AP2 mandate verification (opt-in)** — set `Options.AP2PublicKey` (or `a2a.WithPushURLPolicy`'s sibling `a2a.WithAP2PublicKey` for embedded handlers) and the gateway verifies AP2 payment/checkout mandates carried on incoming messages — signature and task/context binding — recording the outcome in each task's `ap2Verifications`, with the x402 settlement rail carried through for the paid path. Off by default; mandates are otherwise carried unverified. (`gateway/a2a/`)
 - **Flow human-in-the-loop pause/resume** — a flow step can suspend a run for external input with `flow.Await(key, prompt)` (or `flow.AwaitStep`): the run checkpoints with status `waiting` and `Execute` returns cleanly. `Flow.Waiting` lists suspended runs with what they await, and `Flow.ResumeWith(ctx, runID, input)` injects the input and continues from the next step. Recovery (`ResumePending`) skips waiting runs since they need input, not a restart. (`flow/`)
+- **Kubernetes reconcile core (alpha)** — `kubernetes.Reconcile(desired, observed)` decides the single action needed to converge an `Agent`/`Service`/`Flow` resource toward its Deployment (create / update / noop) and returns `Ready`/`Error` status conditions. Dependency-free (no controller-runtime / client-go) and fully unit-testable; a future operator binary supplies observed state and applies the action. (`deploy/kubernetes/`)

 ### Changed
 - **Remote agent chat streaming** — `micro chat` now streams replies from remote agents instead of waiting for the full response. (`cmd/micro/`, `agent/`)
@@ -11,8 +11,13 @@ production defaults.
  `services.micro.dev`, and `flows.micro.dev`.
 - A small dependency-free mapper that turns a desired Go Micro resource into the
  Kubernetes `Deployment` shape an operator reconciliation loop will own.
- Unit tests that validate the structural CRD fragments and dry-run the
-  Agent-to-Deployment mapping.
+- A dependency-free `Reconcile(desired, observed)` core that decides the one
+  action needed to converge (create / update / noop) and the `Ready`/`Error`
+  status conditions — no controller-runtime, no client-go, fully unit-testable.
+  A future operator binary supplies the observed state and applies the action;
+  only that adapter needs the Kubernetes client.
+- Unit tests that validate the structural CRD fragments, the Agent-to-Deployment
+  mapping, and the reconcile decision/conditions.

 ## Local validation

@@ -0,0 +1,105 @@
+package kubernetes
+
+import (
+	"fmt"
+	"reflect"
+)
+
+// Reconcile is the pure decision core an operator's reconcile loop runs: given
+// a desired resource and the currently observed cluster state, it computes the
+// one action needed to converge (create / update / nothing) plus the status
+// conditions to publish. It does not talk to a cluster — no controller-runtime,
+// no client-go — so the whole convergence decision is unit-testable. An adapter
+// binary supplies Observed from the live cluster and applies the returned
+// Action; that adapter is the only piece that needs the Kubernetes client.
+
+// ActionType is the change a reconcile wants applied.
+type ActionType string
+
+const (
+	// ActionCreate means the workload does not exist yet and should be created.
+	ActionCreate ActionType = "create"
+	// ActionUpdate means the workload exists but drifts from desired.
+	ActionUpdate ActionType = "update"
+	// ActionNoop means the workload already matches desired.
+	ActionNoop ActionType = "noop"
+)
+
+// Action is the change Reconcile decided on, carrying the desired Deployment.
+type Action struct {
+	Type       ActionType
+	Deployment Deployment
+}
+
+// Observed is the current cluster state Reconcile compares against. The adapter
+// fills it from the live cluster; a nil Deployment means "not created yet".
+type Observed struct {
+	// Deployment is the workload as it currently exists, or nil if absent.
+	Deployment *Deployment
+	// ReadyReplicas is how many pods are ready, from the live Deployment status.
+	ReadyReplicas int32
+}
+
+// Condition is a status condition to publish on the resource — the ready/error
+// signal for the inner-loop and deploy story. It mirrors the Kubernetes
+// condition shape without importing the API types.
+type Condition struct {
+	Type    string `json:"type"`   // "Ready" | "Error"
+	Status  string `json:"status"` // "True" | "False" | "Unknown"
+	Reason  string `json:"reason"`
+	Message string `json:"message,omitempty"`
+}
+
+// Reconcile computes the action to bring observed toward desired, plus the
+// status conditions. A spec that fails to map returns an Error condition and
+// the error (no action).
+func Reconcile(desired Resource, observed Observed) (Action, []Condition, error) {
+	want, err := MapDeployment(desired)
+	if err != nil {
+		return Action{}, []Condition{{
+			Type: "Error", Status: "True", Reason: "InvalidSpec", Message: err.Error(),
+		}}, err
+	}
+
+	var action Action
+	switch {
+	case observed.Deployment == nil:
+		action = Action{Type: ActionCreate, Deployment: want}
+	case deploymentDiffers(*observed.Deployment, want):
+		action = Action{Type: ActionUpdate, Deployment: want}
+	default:
+		action = Action{Type: ActionNoop, Deployment: want}
+	}
+
+	return action, conditions(want, observed), nil
+}
+
+// conditions derives the Ready condition from observed state against desired.
+func conditions(want Deployment, observed Observed) []Condition {
+	switch {
+	case observed.Deployment == nil:
+		return []Condition{{
+			Type: "Ready", Status: "False", Reason: "Creating",
+			Message: "workload not yet created",
+		}}
+	case observed.ReadyReplicas < want.Replicas:
+		return []Condition{{
+			Type: "Ready", Status: "False", Reason: "Progressing",
+			Message: fmt.Sprintf("%d/%d replicas ready", observed.ReadyReplicas, want.Replicas),
+		}}
+	default:
+		return []Condition{{
+			Type: "Ready", Status: "True", Reason: "Available",
+			Message: fmt.Sprintf("%d/%d replicas ready", observed.ReadyReplicas, want.Replicas),
+		}}
+	}
+}
+
+// deploymentDiffers reports whether the observed deployment drifts from desired
+// on the fields this operator manages (replicas, container, labels). Fields the
+// cluster owns (status, cluster-assigned metadata) are intentionally ignored.
+func deploymentDiffers(current, want Deployment) bool {
+	return current.Replicas != want.Replicas ||
+		!reflect.DeepEqual(current.Pod.Container, want.Pod.Container) ||
+		!reflect.DeepEqual(current.Labels, want.Labels)
+}
@@ -0,0 +1,88 @@
+package kubernetes
+
+import "testing"
+
+func agentResource() Resource {
+	return Resource{
+		Kind:      KindAgent,
+		Name:      "support",
+		Namespace: "agents",
+		Spec:      WorkloadSpec{Image: "example/support:v1", Replicas: 2, Registry: "kubernetes"},
+	}
+}
+
+func TestReconcileCreatesWhenAbsent(t *testing.T) {
+	action, conds, err := Reconcile(agentResource(), Observed{Deployment: nil})
+	if err != nil {
+		t.Fatalf("Reconcile: %v", err)
+	}
+	if action.Type != ActionCreate {
+		t.Fatalf("action = %q, want create", action.Type)
+	}
+	if action.Deployment.Name != "support" || action.Deployment.Replicas != 2 {
+		t.Fatalf("desired deployment = %+v", action.Deployment)
+	}
+	if ready := findCondition(conds, "Ready"); ready == nil || ready.Status != "False" || ready.Reason != "Creating" {
+		t.Fatalf("ready condition = %+v, want False/Creating", ready)
+	}
+}
+
+func TestReconcileNoopWhenMatchedAndReady(t *testing.T) {
+	want, _ := MapDeployment(agentResource())
+	action, conds, err := Reconcile(agentResource(), Observed{Deployment: &want, ReadyReplicas: 2})
+	if err != nil {
+		t.Fatalf("Reconcile: %v", err)
+	}
+	if action.Type != ActionNoop {
+		t.Fatalf("action = %q, want noop", action.Type)
+	}
+	if ready := findCondition(conds, "Ready"); ready == nil || ready.Status != "True" || ready.Reason != "Available" {
+		t.Fatalf("ready condition = %+v, want True/Available", ready)
+	}
+}
+
+func TestReconcileUpdatesOnDrift(t *testing.T) {
+	current, _ := MapDeployment(agentResource())
+	current.Pod.Container.Image = "example/support:v0" // stale image → drift
+	action, _, err := Reconcile(agentResource(), Observed{Deployment: &current, ReadyReplicas: 2})
+	if err != nil {
+		t.Fatalf("Reconcile: %v", err)
+	}
+	if action.Type != ActionUpdate {
+		t.Fatalf("action = %q, want update", action.Type)
+	}
+	if action.Deployment.Pod.Container.Image != "example/support:v1" {
+		t.Fatalf("update should carry the desired image, got %q", action.Deployment.Pod.Container.Image)
+	}
+}
+
+func TestReconcileProgressingWhenUnderReplicated(t *testing.T) {
+	want, _ := MapDeployment(agentResource())
+	_, conds, err := Reconcile(agentResource(), Observed{Deployment: &want, ReadyReplicas: 1})
+	if err != nil {
+		t.Fatalf("Reconcile: %v", err)
+	}
+	if ready := findCondition(conds, "Ready"); ready == nil || ready.Status != "False" || ready.Reason != "Progressing" {
+		t.Fatalf("ready condition = %+v, want False/Progressing", ready)
+	}
+}
+
+func TestReconcileErrorOnInvalidSpec(t *testing.T) {
+	// Missing image → MapDeployment fails → Error condition, no action.
+	_, conds, err := Reconcile(Resource{Kind: KindService, Name: "api"}, Observed{})
+	if err == nil {
+		t.Fatal("Reconcile should error on an invalid spec")
+	}
+	if e := findCondition(conds, "Error"); e == nil || e.Status != "True" || e.Reason != "InvalidSpec" {
+		t.Fatalf("error condition = %+v, want True/InvalidSpec", e)
+	}
+}
+
+func findCondition(conds []Condition, typ string) *Condition {
+	for i := range conds {
+		if conds[i].Type == typ {
+			return &conds[i]
+		}
+	}
+	return nil
+}