feat: declare scheduled runs with the ScheduledRun CRD

docs/architecture/configuration-and-compilation.md

@@ -353,3 +353,61 @@ Agent's last successful revision.
 This replaces implicit label selection and the old template/Harness inputs as a
 breaking API change. SandboxTemplate hosting configuration is not part of this
 change.
+
+## Declared schedules
+
+`ScheduledRun` is an `api.kagent.dev/v1alpha3` resource that declares a cron
+schedule for one Agent in the same namespace. It is the YAML counterpart of the
+gRPC `ScheduledRunService`: both produce rows in the same `scheduled_run` table,
+and the same Scheduler and execution Controller fire them.
+
+```yaml
+apiVersion: api.kagent.dev/v1alpha3
+kind: ScheduledRun
+metadata:
+  name: nightly-report
+  namespace: kagent
+spec:
+  agentRef:
+    name: assistant
+  schedule: "0 3 * * *"
+  timeZone: Europe/Paris
+  prompt: Summarize yesterday's incidents.
+  paused: false
+  executionTimeout: 15m
+```
+
+`schedule` is a five-field cron expression; macros and seconds are rejected at
+admission. `timeZone` defaults to UTC and `executionTimeout` to 15 minutes, the
+same defaults the gRPC API applies. The reconciler validates the cron expression
+and time zone with the shared `go/api/scheduledrun` semantics and reports the
+result in the `Accepted` condition. `ResolvedRefs` reports whether the Agent
+exists; a missing Agent does not block persistence, matching gRPC updates, and
+occurrences fail at run time until it exists. `Ready` is true once the row is
+persisted for the current generation.
+
+Ownership is the main difference from API-created schedules. A resource has no
+human creator, so its row is owned by the synthetic creator
+`scheduledrun.api.kagent.dev/<namespace>/<name>` and keyed by the object UID as
+its request ID. Sessions created for its occurrences carry that owner. The
+creator-scoped gRPC API therefore never shows a declared schedule to a human
+principal, and no `kubectl` user can assign a schedule to another user's
+identity. `kubectl get scheduledruns` is the observation surface.
+
+The reconciler converges rather than edits: it applies the desired configuration
+in one store transaction, bumps the etag only when something changed, and
+tombstones any live row of the same owner with a different UID. A deleted and
+recreated object therefore starts a new schedule identity while the previous
+execution history is retained.
+
+`agentRef` is immutable, at admission and in the store. A reserved execution
+resolves its schedule's Agent only when its session is reserved, so changing the
+target in place could send an earlier prompt to a different Agent. To target
+another Agent, recreate the ScheduledRun. The gRPC API likewise fixes the Agent
+at creation.
+
+The `kagent.dev/scheduledrun-retirement` finalizer is added before the row is
+first persisted. It holds a deleted object until its row is tombstoned, so a
+deletion while no controller is running is retired on restart. An invalid edit
+leaves the last accepted schedule running, because deletion is final per
+request identity and must not be triggered by a typo.

go/api/scheduledrun/schedule.go

@@ -3,18 +3,50 @@
 package scheduledrun
 
 import (
+	"crypto/sha256"
 	"fmt"
 	"strings"
 	"time"
 	_ "time/tzdata" // Scheduling must work in images without system timezone data.
 
+	"buf.build/go/protovalidate"
 	"github.com/robfig/cron/v3"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/durationpb"
 
 	apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1"
 )
 
+// KubernetesCreator is the owner recorded for schedules declared by a
+// ScheduledRun resource. It is namespaced so that control-plane rows cannot be
+// read or edited through the creator-scoped gRPC API by a human principal.
+func KubernetesCreator(namespace, name string) string {
+	return "scheduledrun.api.kagent.dev/" + namespace + "/" + name
+}
+
+// Validate reports configuration the store rejects: protobuf field constraints
+// and scheduling semantics. Callers outside the gRPC interceptor need it before
+// persisting a normalized configuration.
+func Validate(config *apiv1.ScheduledRunConfig) error {
+	if err := protovalidate.Validate(config); err != nil {
+		return err
+	}
+	_, err := Next(config, time.Now())
+	return err
+}
+
+// RequestHash fingerprints the immutable creation inputs of a schedule for
+// request deduplication. It hashes the normalized inputs, never the mutable
+// persisted configuration.
+func RequestHash(agent *apiv1.ResourceReference, config *apiv1.ScheduledRunConfig) ([]byte, error) {
+	data, err := (proto.MarshalOptions{Deterministic: true}).Marshal(&apiv1.CreateScheduledRunRequest{Agent: agent, Config: config})
+	if err != nil {
+		return nil, err
+	}
+	sum := sha256.Sum256(data)
+	return sum[:], nil
+}
+
 // Normalize supplies defaults without mutating a caller's request.
 func Normalize(config *apiv1.ScheduledRunConfig) *apiv1.ScheduledRunConfig {
 	result := proto.CloneOf(config)

go/api/v1alpha3/configuration_crd_cel_test.go

@@ -56,7 +56,7 @@ func TestConfigurationCRDValidation(t *testing.T) {
 				names = append(names, resource.Name)
 			}
 		}
-		require.ElementsMatch(t, []string{"agents", "agenttemplates", "harnesses", "modelconfigs", "modelproviderconfigs", "remotemcpservers", "sandboxtemplates"}, names)
+		require.ElementsMatch(t, []string{"agents", "agenttemplates", "harnesses", "modelconfigs", "modelproviderconfigs", "remotemcpservers", "sandboxtemplates", "scheduledruns"}, names)
 		_, err = discoveryClient.ServerResourcesForGroupVersion("kagent.dev/v1alpha3")
 		require.True(t, apierrors.IsNotFound(err), "the new CRDs must not publish the legacy group: %v", err)
 	})
@@ -507,6 +507,56 @@ func TestConfigurationCRDValidation(t *testing.T) {
 			}
 		}
 	})
+
+	t.Run("ScheduledRun admission", func(t *testing.T) {
+		for _, tc := range []struct {
+			name       string
+			mutate     func(*ScheduledRunSpec)
+			wantReject string
+		}{
+			{name: "valid"},
+			{name: "whitespace", mutate: func(spec *ScheduledRunSpec) { spec.Schedule = " 0   3 * *  * " }},
+			{name: "missing agent", mutate: func(spec *ScheduledRunSpec) { spec.AgentRef.Name = "" }, wantReject: "agentRef.name must not be empty"},
+			{name: "four fields", mutate: func(spec *ScheduledRunSpec) { spec.Schedule = "0 3 * *" }, wantReject: "schedule must contain five cron fields"},
+			{name: "macro", mutate: func(spec *ScheduledRunSpec) { spec.Schedule = "@hourly" }, wantReject: "schedule must contain five cron fields"},
+			{name: "blank prompt", mutate: func(spec *ScheduledRunSpec) { spec.Prompt = "   " }, wantReject: "spec.prompt"},
+			{name: "empty time zone", mutate: func(spec *ScheduledRunSpec) { spec.TimeZone = new("") }, wantReject: "spec.timeZone"},
+			{name: "zero timeout", mutate: func(spec *ScheduledRunSpec) { spec.ExecutionTimeout = &metav1.Duration{} }, wantReject: "executionTimeout must be positive"},
+			{name: "negative timeout", mutate: func(spec *ScheduledRunSpec) { spec.ExecutionTimeout = &metav1.Duration{Duration: -time.Second} }, wantReject: "executionTimeout must be positive"},
+		} {
+			t.Run(tc.name, func(t *testing.T) {
+				schedule := &ScheduledRun{
+					ObjectMeta: metav1.ObjectMeta{GenerateName: "schedule-", Namespace: namespace},
+					Spec:       ScheduledRunSpec{AgentRef: corev1.LocalObjectReference{Name: "report"}, Schedule: "0 3 * * *", Prompt: "Summarize"},
+				}
+				if tc.mutate != nil {
+					tc.mutate(&schedule.Spec)
+				}
+				err := cl.Create(ctx, schedule)
+				if tc.wantReject != "" {
+					require.ErrorContains(t, err, tc.wantReject)
+					return
+				}
+				require.NoError(t, err)
+				require.NotNil(t, schedule.Spec.Paused)
+				require.False(t, *schedule.Spec.Paused)
+				require.NotNil(t, schedule.Spec.ExecutionTimeout)
+				require.Equal(t, 15*time.Minute, schedule.Spec.ExecutionTimeout.Duration)
+				require.Nil(t, schedule.Spec.TimeZone)
+			})
+		}
+		t.Run("agentRef is immutable", func(t *testing.T) {
+			schedule := &ScheduledRun{
+				ObjectMeta: metav1.ObjectMeta{GenerateName: "schedule-", Namespace: namespace},
+				Spec:       ScheduledRunSpec{AgentRef: corev1.LocalObjectReference{Name: "report"}, Schedule: "0 3 * * *", Prompt: "Summarize"},
+			}
+			require.NoError(t, cl.Create(ctx, schedule))
+			schedule.Spec.Prompt = "Edited"
+			require.NoError(t, cl.Update(ctx, schedule))
+			schedule.Spec.AgentRef.Name = "other"
+			require.ErrorContains(t, cl.Update(ctx, schedule), "agentRef is immutable")
+		})
+	})
 }
 
 func sandboxTemplateForValidation(namespace, name string, mutate func(*SandboxTemplateSpec)) *SandboxTemplate {