diff --git a/docs/architecture/configuration-and-compilation.md b/docs/architecture/configuration-and-compilation.md index 41d6a80a28..e4cc9bf668 100644 --- a/docs/architecture/configuration-and-compilation.md +++ b/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//` 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. diff --git a/go/api/clientset/versioned/typed/api/v1alpha3/api_client.go b/go/api/clientset/versioned/typed/api/v1alpha3/api_client.go index 97de970b4a..cfcb3365c3 100644 --- a/go/api/clientset/versioned/typed/api/v1alpha3/api_client.go +++ b/go/api/clientset/versioned/typed/api/v1alpha3/api_client.go @@ -18,6 +18,7 @@ type ApiV1alpha3Interface interface { ModelConfigsGetter RemoteMCPServersGetter SandboxTemplatesGetter + ScheduledRunsGetter } // ApiV1alpha3Client is used to interact with features provided by the api group. @@ -49,6 +50,10 @@ func (c *ApiV1alpha3Client) SandboxTemplates(namespace string) SandboxTemplateIn return newSandboxTemplates(c, namespace) } +func (c *ApiV1alpha3Client) ScheduledRuns(namespace string) ScheduledRunInterface { + return newScheduledRuns(c, namespace) +} + // NewForConfig creates a new ApiV1alpha3Client for the given config. // NewForConfig is equivalent to NewForConfigAndClient(c, httpClient), // where httpClient was generated with rest.HTTPClientFor(c). diff --git a/go/api/clientset/versioned/typed/api/v1alpha3/fake/fake_api_client.go b/go/api/clientset/versioned/typed/api/v1alpha3/fake/fake_api_client.go index 19d7ba79b7..48003331a9 100644 --- a/go/api/clientset/versioned/typed/api/v1alpha3/fake/fake_api_client.go +++ b/go/api/clientset/versioned/typed/api/v1alpha3/fake/fake_api_client.go @@ -36,6 +36,10 @@ func (c *FakeApiV1alpha3) SandboxTemplates(namespace string) v1alpha3.SandboxTem return newFakeSandboxTemplates(c, namespace) } +func (c *FakeApiV1alpha3) ScheduledRuns(namespace string) v1alpha3.ScheduledRunInterface { + return newFakeScheduledRuns(c, namespace) +} + // RESTClient returns a RESTClient that is used to communicate // with API server by this client implementation. func (c *FakeApiV1alpha3) RESTClient() rest.Interface { diff --git a/go/api/clientset/versioned/typed/api/v1alpha3/fake/fake_scheduledrun.go b/go/api/clientset/versioned/typed/api/v1alpha3/fake/fake_scheduledrun.go new file mode 100644 index 0000000000..d97809a3f7 --- /dev/null +++ b/go/api/clientset/versioned/typed/api/v1alpha3/fake/fake_scheduledrun.go @@ -0,0 +1,36 @@ +// Code generated by client-gen. DO NOT EDIT. + +package fake + +import ( + apiv1alpha3 "github.com/kagent-dev/kagent/go/api/clientset/versioned/typed/api/v1alpha3" + v1alpha3 "github.com/kagent-dev/kagent/go/api/v1alpha3" + gentype "k8s.io/client-go/gentype" +) + +// fakeScheduledRuns implements ScheduledRunInterface +type fakeScheduledRuns struct { + *gentype.FakeClientWithList[*v1alpha3.ScheduledRun, *v1alpha3.ScheduledRunList] + Fake *FakeApiV1alpha3 +} + +func newFakeScheduledRuns(fake *FakeApiV1alpha3, namespace string) apiv1alpha3.ScheduledRunInterface { + return &fakeScheduledRuns{ + gentype.NewFakeClientWithList[*v1alpha3.ScheduledRun, *v1alpha3.ScheduledRunList]( + fake.Fake, + namespace, + v1alpha3.SchemeGroupVersion.WithResource("scheduledruns"), + v1alpha3.SchemeGroupVersion.WithKind("ScheduledRun"), + func() *v1alpha3.ScheduledRun { return &v1alpha3.ScheduledRun{} }, + func() *v1alpha3.ScheduledRunList { return &v1alpha3.ScheduledRunList{} }, + func(dst, src *v1alpha3.ScheduledRunList) { dst.ListMeta = src.ListMeta }, + func(list *v1alpha3.ScheduledRunList) []*v1alpha3.ScheduledRun { + return gentype.ToPointerSlice(list.Items) + }, + func(list *v1alpha3.ScheduledRunList, items []*v1alpha3.ScheduledRun) { + list.Items = gentype.FromPointerSlice(items) + }, + ), + fake, + } +} diff --git a/go/api/clientset/versioned/typed/api/v1alpha3/generated_expansion.go b/go/api/clientset/versioned/typed/api/v1alpha3/generated_expansion.go index bc2146b2a4..066f76bb83 100644 --- a/go/api/clientset/versioned/typed/api/v1alpha3/generated_expansion.go +++ b/go/api/clientset/versioned/typed/api/v1alpha3/generated_expansion.go @@ -13,3 +13,5 @@ type ModelConfigExpansion interface{} type RemoteMCPServerExpansion interface{} type SandboxTemplateExpansion interface{} + +type ScheduledRunExpansion interface{} diff --git a/go/api/clientset/versioned/typed/api/v1alpha3/scheduledrun.go b/go/api/clientset/versioned/typed/api/v1alpha3/scheduledrun.go new file mode 100644 index 0000000000..bb8b917d06 --- /dev/null +++ b/go/api/clientset/versioned/typed/api/v1alpha3/scheduledrun.go @@ -0,0 +1,54 @@ +// Code generated by client-gen. DO NOT EDIT. + +package v1alpha3 + +import ( + context "context" + + scheme "github.com/kagent-dev/kagent/go/api/clientset/versioned/scheme" + apiv1alpha3 "github.com/kagent-dev/kagent/go/api/v1alpha3" + v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + types "k8s.io/apimachinery/pkg/types" + watch "k8s.io/apimachinery/pkg/watch" + gentype "k8s.io/client-go/gentype" +) + +// ScheduledRunsGetter has a method to return a ScheduledRunInterface. +// A group's client should implement this interface. +type ScheduledRunsGetter interface { + ScheduledRuns(namespace string) ScheduledRunInterface +} + +// ScheduledRunInterface has methods to work with ScheduledRun resources. +type ScheduledRunInterface interface { + Create(ctx context.Context, scheduledRun *apiv1alpha3.ScheduledRun, opts v1.CreateOptions) (*apiv1alpha3.ScheduledRun, error) + Update(ctx context.Context, scheduledRun *apiv1alpha3.ScheduledRun, opts v1.UpdateOptions) (*apiv1alpha3.ScheduledRun, error) + // Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus(). + UpdateStatus(ctx context.Context, scheduledRun *apiv1alpha3.ScheduledRun, opts v1.UpdateOptions) (*apiv1alpha3.ScheduledRun, error) + Delete(ctx context.Context, name string, opts v1.DeleteOptions) error + DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error + Get(ctx context.Context, name string, opts v1.GetOptions) (*apiv1alpha3.ScheduledRun, error) + List(ctx context.Context, opts v1.ListOptions) (*apiv1alpha3.ScheduledRunList, error) + Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) + Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *apiv1alpha3.ScheduledRun, err error) + ScheduledRunExpansion +} + +// scheduledRuns implements ScheduledRunInterface +type scheduledRuns struct { + *gentype.ClientWithList[*apiv1alpha3.ScheduledRun, *apiv1alpha3.ScheduledRunList] +} + +// newScheduledRuns returns a ScheduledRuns +func newScheduledRuns(c *ApiV1alpha3Client, namespace string) *scheduledRuns { + return &scheduledRuns{ + gentype.NewClientWithList[*apiv1alpha3.ScheduledRun, *apiv1alpha3.ScheduledRunList]( + "scheduledruns", + c.RESTClient(), + scheme.ParameterCodec, + namespace, + func() *apiv1alpha3.ScheduledRun { return &apiv1alpha3.ScheduledRun{} }, + func() *apiv1alpha3.ScheduledRunList { return &apiv1alpha3.ScheduledRunList{} }, + ), + } +} diff --git a/go/api/scheduledrun/schedule.go b/go/api/scheduledrun/schedule.go index 01e73e0aad..2ff6987069 100644 --- a/go/api/scheduledrun/schedule.go +++ b/go/api/scheduledrun/schedule.go @@ -3,11 +3,13 @@ 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" @@ -15,6 +17,36 @@ import ( 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) diff --git a/go/api/v1alpha3/configuration_crd_cel_test.go b/go/api/v1alpha3/configuration_crd_cel_test.go index 02d17d4b57..d22a750492 100644 --- a/go/api/v1alpha3/configuration_crd_cel_test.go +++ b/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 { diff --git a/go/api/v1alpha3/scheduledrun_types.go b/go/api/v1alpha3/scheduledrun_types.go new file mode 100644 index 0000000000..facf451e6b --- /dev/null +++ b/go/api/v1alpha3/scheduledrun_types.go @@ -0,0 +1,127 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package v1alpha3 + +import ( + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" +) + +// +genclient +// +kubebuilder:object:root=true +// +kubebuilder:resource:path=scheduledruns,singular=scheduledrun,categories=kagent +// +kubebuilder:subresource:status +// +kubebuilder:printcolumn:name="Agent",type="string",JSONPath=".spec.agentRef.name" +// +kubebuilder:printcolumn:name="Schedule",type="string",JSONPath=".spec.schedule" +// +kubebuilder:printcolumn:name="Paused",type="boolean",JSONPath=".spec.paused" +// +kubebuilder:printcolumn:name="Ready",type="string",JSONPath=".status.conditions[?(@.type=='Ready')].status" +// +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp" + +// ScheduledRun sends a prompt to an Agent on a cron schedule. Each occurrence +// creates a fresh Session owned by the ScheduledRun rather than by a human user. +// Task state and history belong to upstream A2A. +type ScheduledRun struct { + metav1.TypeMeta `json:",inline"` + // +optional + metav1.ObjectMeta `json:"metadata,omitempty"` + // +required + Spec ScheduledRunSpec `json:"spec"` + // +optional + Status ScheduledRunStatus `json:"status,omitempty"` +} + +// ScheduledRunSpec describes when and what to send. The Agent reference is +// local to the ScheduledRun's namespace. +type ScheduledRunSpec struct { + // AgentRef names the Agent that receives each occurrence. It is immutable: + // reserved occurrences resolve the schedule's Agent when they start, so a + // change could send an earlier prompt to a different Agent. Recreate the + // ScheduledRun to target another Agent. + // +kubebuilder:validation:XValidation:rule="has(self.name) && self.name != ''",message="agentRef.name must not be empty" + // +kubebuilder:validation:XValidation:rule="self == oldSelf",message="agentRef is immutable" + // +required + AgentRef corev1.LocalObjectReference `json:"agentRef"` + // Schedule is a standard five-field cron expression evaluated in TimeZone. + // Macros such as @hourly and seconds fields are not accepted. + // +kubebuilder:validation:MinLength=1 + // +kubebuilder:validation:MaxLength=256 + // +kubebuilder:validation:XValidation:rule="self.split(' ').filter(f, f != '').size() == 5",message="schedule must contain five cron fields" + // +required + Schedule string `json:"schedule"` + // TimeZone is an IANA time zone name such as Europe/Paris. Omitted means UTC. + // +kubebuilder:validation:MinLength=1 + // +kubebuilder:validation:MaxLength=253 + // +optional + TimeZone *string `json:"timeZone,omitempty"` + // Prompt is the user message sent at each occurrence. + // +kubebuilder:validation:MinLength=1 + // +kubebuilder:validation:MaxLength=32768 + // +kubebuilder:validation:Pattern=`\S` + // +required + Prompt string `json:"prompt"` + // Paused keeps the schedule and its history without reserving new + // occurrences. Omitted means false. + // +kubebuilder:default=false + // +optional + Paused *bool `json:"paused,omitempty"` + // ExecutionTimeout bounds one occurrence, including queueing and runtime + // provisioning. Omitted means 15 minutes. + // +kubebuilder:default="15m" + // +kubebuilder:validation:XValidation:rule="duration(self) > duration('0s') && duration(self) <= duration('9223372036s')",message="executionTimeout must be positive and at most 9223372036s" + // +optional + ExecutionTimeout *metav1.Duration `json:"executionTimeout,omitempty"` +} + +const ( + // ScheduledRunConditionAccepted reports whether the spec compiles to a + // valid schedule: cron syntax, time zone, prompt, and timeout bounds. + ScheduledRunConditionAccepted = "Accepted" + // ScheduledRunConditionResolvedRefs reports whether the referenced Agent + // exists. Occurrences of a schedule with a missing Agent fail at run time. + ScheduledRunConditionResolvedRefs = "ResolvedRefs" + // ScheduledRunConditionReady reports whether the schedule is persisted in + // the control plane and will reserve occurrences unless paused. + ScheduledRunConditionReady = "Ready" +) + +// ScheduledRunStatus reports control-plane persistence of a ScheduledRun. +type ScheduledRunStatus struct { + // +optional + ObservedGeneration int64 `json:"observedGeneration,omitempty"` + // +kubebuilder:validation:MaxItems=3 + // +listType=map + // +listMapKey=type + // +optional + Conditions []metav1.Condition `json:"conditions,omitempty"` +} + +// +kubebuilder:object:root=true + +// ScheduledRunList contains ScheduledRun resources. +type ScheduledRunList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []ScheduledRun `json:"items"` +} + +func init() { + SchemeBuilder.Register(func(s *runtime.Scheme) error { + s.AddKnownTypes(GroupVersion, &ScheduledRun{}, &ScheduledRunList{}) + return nil + }) +} diff --git a/go/api/v1alpha3/zz_generated.deepcopy.go b/go/api/v1alpha3/zz_generated.deepcopy.go index f21a0d5b50..57ce68674d 100644 --- a/go/api/v1alpha3/zz_generated.deepcopy.go +++ b/go/api/v1alpha3/zz_generated.deepcopy.go @@ -1698,6 +1698,118 @@ func (in *SandboxTemplateWorkload) DeepCopy() *SandboxTemplateWorkload { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ScheduledRun) DeepCopyInto(out *ScheduledRun) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + in.Spec.DeepCopyInto(&out.Spec) + in.Status.DeepCopyInto(&out.Status) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ScheduledRun. +func (in *ScheduledRun) DeepCopy() *ScheduledRun { + if in == nil { + return nil + } + out := new(ScheduledRun) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *ScheduledRun) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ScheduledRunList) DeepCopyInto(out *ScheduledRunList) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ListMeta.DeepCopyInto(&out.ListMeta) + if in.Items != nil { + in, out := &in.Items, &out.Items + *out = make([]ScheduledRun, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ScheduledRunList. +func (in *ScheduledRunList) DeepCopy() *ScheduledRunList { + if in == nil { + return nil + } + out := new(ScheduledRunList) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *ScheduledRunList) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ScheduledRunSpec) DeepCopyInto(out *ScheduledRunSpec) { + *out = *in + out.AgentRef = in.AgentRef + if in.TimeZone != nil { + in, out := &in.TimeZone, &out.TimeZone + *out = new(string) + **out = **in + } + if in.Paused != nil { + in, out := &in.Paused, &out.Paused + *out = new(bool) + **out = **in + } + if in.ExecutionTimeout != nil { + in, out := &in.ExecutionTimeout, &out.ExecutionTimeout + *out = new(metav1.Duration) + **out = **in + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ScheduledRunSpec. +func (in *ScheduledRunSpec) DeepCopy() *ScheduledRunSpec { + if in == nil { + return nil + } + out := new(ScheduledRunSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *ScheduledRunStatus) DeepCopyInto(out *ScheduledRunStatus) { + *out = *in + if in.Conditions != nil { + in, out := &in.Conditions, &out.Conditions + *out = make([]metav1.Condition, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ScheduledRunStatus. +func (in *ScheduledRunStatus) DeepCopy() *ScheduledRunStatus { + if in == nil { + return nil + } + out := new(ScheduledRunStatus) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *SecretReference) DeepCopyInto(out *SecretReference) { *out = *in diff --git a/go/core/internal/controller/scheduledrun/reconciler.go b/go/core/internal/controller/scheduledrun/reconciler.go new file mode 100644 index 0000000000..dc73ae1ef0 --- /dev/null +++ b/go/core/internal/controller/scheduledrun/reconciler.go @@ -0,0 +1,192 @@ +package scheduledrun + +import ( + "context" + "errors" + "fmt" + "reflect" + + "google.golang.org/protobuf/types/known/durationpb" + apierrors "k8s.io/apimachinery/pkg/api/errors" + apiMeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/builder" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1" + schedules "github.com/kagent-dev/kagent/go/api/scheduledrun" + "github.com/kagent-dev/kagent/go/api/v1alpha3" + "github.com/kagent-dev/kagent/go/core/internal/database" +) + +type reconcilerStore interface { + ApplyScheduledRun(context.Context, *apiv1.ScheduledRun, string) (*apiv1.ScheduledRun, error) + RetireScheduledRuns(context.Context, string) error +} + +// Reconciler persists ScheduledRun resources as control-plane schedules. The +// Scheduler and execution Controller then treat them like API-created +// schedules; only the creator differs, so no human principal can edit them. +type Reconciler struct { + client client.Client + store reconcilerStore +} + +func NewReconciler(client client.Client, store reconcilerStore) *Reconciler { + return &Reconciler{client: client, store: store} +} + +func (r *Reconciler) SetupWithManager(manager ctrl.Manager) error { + return ctrl.NewControllerManagedBy(manager). + For(&v1alpha3.ScheduledRun{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). + // Only Agent existence is observed, so create and delete events suffice. + Watches(&v1alpha3.Agent{}, handler.EnqueueRequestsFromMapFunc(r.requestsForAgent), builder.WithPredicates(predicate.Funcs{ + UpdateFunc: func(event.UpdateEvent) bool { return false }, + })). + Complete(r) +} + +// retirementFinalizer keeps a ScheduledRun until its schedule is retired, so a +// deletion while no controller is running is still observed on restart. +const retirementFinalizer = "kagent.dev/scheduledrun-retirement" + +func (r *Reconciler) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) { + creator := schedules.KubernetesCreator(request.Namespace, request.Name) + schedule := &v1alpha3.ScheduledRun{} + if err := r.client.Get(ctx, request.NamespacedName, schedule); err != nil { + if !apierrors.IsNotFound(err) { + return reconcile.Result{}, err + } + // Covers objects whose finalizer was removed by hand. Idempotent. + return reconcile.Result{}, r.store.RetireScheduledRuns(ctx, creator) + } + if !schedule.DeletionTimestamp.IsZero() { + if !controllerutil.ContainsFinalizer(schedule, retirementFinalizer) { + return reconcile.Result{}, nil + } + // The name cannot be reused until the finalizer is removed, so every live + // row of this creator belongs to the object being deleted. + if err := r.store.RetireScheduledRuns(ctx, creator); err != nil { + return reconcile.Result{}, err + } + controllerutil.RemoveFinalizer(schedule, retirementFinalizer) + return reconcile.Result{}, r.client.Update(ctx, schedule) + } + if controllerutil.AddFinalizer(schedule, retirementFinalizer) { + // Persist the finalizer before the row exists, so it can never outlive the object. + if err := r.client.Update(ctx, schedule); err != nil { + return reconcile.Result{}, fmt.Errorf("add ScheduledRun finalizer: %w", err) + } + } + status := schedule.Status.DeepCopy() + status.ObservedGeneration = schedule.Generation + condition := func(conditionType string, conditionStatus metav1.ConditionStatus, reason, message string) { + apiMeta.SetStatusCondition(&status.Conditions, metav1.Condition{ + Type: conditionType, Status: conditionStatus, Reason: reason, Message: message, + ObservedGeneration: schedule.Generation, + }) + } + + resolved, err := r.resolveAgent(ctx, schedule) + if err != nil { + return reconcile.Result{}, err + } + if resolved { + condition(v1alpha3.ScheduledRunConditionResolvedRefs, metav1.ConditionTrue, "ResolvedRefs", "Agent "+schedule.Spec.AgentRef.Name+" exists") + } else { + condition(v1alpha3.ScheduledRunConditionResolvedRefs, metav1.ConditionFalse, "AgentNotFound", "Agent "+schedule.Spec.AgentRef.Name+" does not exist; occurrences fail until it does") + } + + config := configFromSpec(schedule.Spec) + if err := schedules.Validate(config); err != nil { + // The last accepted schedule, if any, keeps running: deletion is final per + // request, so an invalid edit must not retire the row. + condition(v1alpha3.ScheduledRunConditionAccepted, metav1.ConditionFalse, "InvalidSchedule", err.Error()) + condition(v1alpha3.ScheduledRunConditionReady, metav1.ConditionFalse, "NotAccepted", "The schedule was not accepted") + return reconcile.Result{}, r.updateStatus(ctx, schedule, status) + } + condition(v1alpha3.ScheduledRunConditionAccepted, metav1.ConditionTrue, "Accepted", "The schedule is valid") + + stored, err := r.store.ApplyScheduledRun(ctx, &apiv1.ScheduledRun{ + Creator: creator, + Agent: &apiv1.ResourceReference{Namespace: schedule.Namespace, Name: schedule.Spec.AgentRef.Name}, + Config: config, + }, string(schedule.UID)) + if err != nil { + condition(v1alpha3.ScheduledRunConditionReady, metav1.ConditionFalse, "PersistenceFailed", err.Error()) + return reconcile.Result{}, errors.Join(fmt.Errorf("apply ScheduledRun: %w", err), r.updateStatus(ctx, schedule, status)) + } + state := "active" + if config.GetPaused() { + state = "paused" + } + condition(v1alpha3.ScheduledRunConditionReady, metav1.ConditionTrue, "Scheduled", "Scheduled run "+stored.GetId()+" is "+state) + return reconcile.Result{}, r.updateStatus(ctx, schedule, status) +} + +// resolveAgent reports whether the referenced Agent exists. Readiness is left +// to execution, matching the gRPC service. +func (r *Reconciler) resolveAgent(ctx context.Context, schedule *v1alpha3.ScheduledRun) (bool, error) { + err := r.client.Get(ctx, types.NamespacedName{Namespace: schedule.Namespace, Name: schedule.Spec.AgentRef.Name}, &v1alpha3.Agent{}) + if apierrors.IsNotFound(err) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("resolve Agent %s: %w", schedule.Spec.AgentRef.Name, err) + } + return true, nil +} + +func (r *Reconciler) updateStatus(ctx context.Context, schedule *v1alpha3.ScheduledRun, status *v1alpha3.ScheduledRunStatus) error { + if reflect.DeepEqual(schedule.Status, *status) { + return nil + } + original := schedule.DeepCopy() + schedule.Status = *status + if err := r.client.Status().Patch(ctx, schedule, client.MergeFrom(original)); err != nil { + return fmt.Errorf("update ScheduledRun status: %w", err) + } + return nil +} + +func (r *Reconciler) requestsForAgent(ctx context.Context, object client.Object) []reconcile.Request { + list := &v1alpha3.ScheduledRunList{} + if err := r.client.List(ctx, list, client.InNamespace(object.GetNamespace())); err != nil { + return nil + } + var requests []reconcile.Request + for i := range list.Items { + if list.Items[i].Spec.AgentRef.Name == object.GetName() { + requests = append(requests, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(&list.Items[i])}) + } + } + return requests +} + +// configFromSpec translates the declared spec into the normalized control-plane +// configuration. Omitted optional fields take the same defaults as the gRPC API. +func configFromSpec(spec v1alpha3.ScheduledRunSpec) *apiv1.ScheduledRunConfig { + config := &apiv1.ScheduledRunConfig{Schedule: spec.Schedule, Prompt: spec.Prompt} + if spec.TimeZone != nil { + config.TimeZone = *spec.TimeZone + } + if spec.Paused != nil { + config.Paused = *spec.Paused + } + if spec.ExecutionTimeout != nil { + config.ExecutionTimeout = durationpb.New(spec.ExecutionTimeout.Duration) + } + return schedules.Normalize(config) +} + +var ( + _ reconcile.Reconciler = (*Reconciler)(nil) + _ reconcilerStore = (*database.Client)(nil) +) diff --git a/go/core/internal/controller/scheduledrun/reconciler_test.go b/go/core/internal/controller/scheduledrun/reconciler_test.go new file mode 100644 index 0000000000..b9bd73254b --- /dev/null +++ b/go/core/internal/controller/scheduledrun/reconciler_test.go @@ -0,0 +1,225 @@ +package scheduledrun + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + apiMeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1" + schedules "github.com/kagent-dev/kagent/go/api/scheduledrun" + "github.com/kagent-dev/kagent/go/api/v1alpha3" +) + +type fakeReconcilerStore struct { + applied *apiv1.ScheduledRun + requestID string + applies int + applyErr error + retired []string + retireErr error +} + +func (f *fakeReconcilerStore) ApplyScheduledRun(_ context.Context, request *apiv1.ScheduledRun, requestID string) (*apiv1.ScheduledRun, error) { + f.applies++ + if f.applyErr != nil { + return nil, f.applyErr + } + f.applied, f.requestID = proto.CloneOf(request), requestID + stored := proto.CloneOf(request) + stored.Id = "11111111-1111-1111-1111-111111111111" + return stored, nil +} + +func (f *fakeReconcilerStore) RetireScheduledRuns(_ context.Context, creator string) error { + f.retired = append(f.retired, creator) + return f.retireErr +} + +func testSchedule() *v1alpha3.ScheduledRun { + return &v1alpha3.ScheduledRun{ + ObjectMeta: metav1.ObjectMeta{Namespace: "team-a", Name: "nightly", UID: types.UID("uid-1"), Generation: 2}, + Spec: v1alpha3.ScheduledRunSpec{ + AgentRef: corev1.LocalObjectReference{Name: "report"}, + Schedule: "0 3 * * *", + Prompt: "Summarize yesterday", + }, + } +} + +func testAgent(name string) *v1alpha3.Agent { + return &v1alpha3.Agent{ObjectMeta: metav1.ObjectMeta{Namespace: "team-a", Name: name}} +} + +func testKube(t *testing.T, objects ...client.Object) client.Client { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, v1alpha3.AddToScheme(scheme)) + return fake.NewClientBuilder().WithScheme(scheme). + WithStatusSubresource(&v1alpha3.ScheduledRun{}). + WithObjects(objects...).Build() +} + +func reconcileOnce(t *testing.T, kube client.Client, store *fakeReconcilerStore, schedule *v1alpha3.ScheduledRun) (*v1alpha3.ScheduledRun, error) { + t.Helper() + _, err := NewReconciler(kube, store).Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKeyFromObject(schedule)}) + updated := &v1alpha3.ScheduledRun{} + if getErr := kube.Get(t.Context(), client.ObjectKeyFromObject(schedule), updated); getErr != nil { + return nil, errors.Join(err, getErr) + } + return updated, err +} + +func requireCondition(t *testing.T, schedule *v1alpha3.ScheduledRun, conditionType string, status metav1.ConditionStatus, reason string) { + t.Helper() + condition := apiMeta.FindStatusCondition(schedule.Status.Conditions, conditionType) + require.NotNil(t, condition, "condition %s", conditionType) + require.Equal(t, status, condition.Status, "%s: %s", conditionType, condition.Message) + require.Equal(t, reason, condition.Reason, "%s: %s", conditionType, condition.Message) + require.Equal(t, schedule.Generation, condition.ObservedGeneration) +} + +func TestReconcilePersistsDeclaredSchedule(t *testing.T) { + schedule := testSchedule() + schedule.Spec.TimeZone = new("Europe/Paris") + schedule.Spec.ExecutionTimeout = &metav1.Duration{Duration: time.Minute} + store := &fakeReconcilerStore{} + kube := testKube(t, schedule, testAgent("report")) + + updated, err := reconcileOnce(t, kube, store, schedule) + require.NoError(t, err) + require.Equal(t, "uid-1", store.requestID) + require.Equal(t, schedules.KubernetesCreator("team-a", "nightly"), store.applied.Creator) + require.True(t, proto.Equal(&apiv1.ResourceReference{Namespace: "team-a", Name: "report"}, store.applied.Agent)) + require.True(t, proto.Equal(&apiv1.ScheduledRunConfig{ + Schedule: "0 3 * * *", TimeZone: "Europe/Paris", Prompt: "Summarize yesterday", + ExecutionTimeout: store.applied.Config.ExecutionTimeout, + }, store.applied.Config)) + require.Equal(t, time.Minute, store.applied.Config.ExecutionTimeout.AsDuration()) + require.Equal(t, schedule.Generation, updated.Status.ObservedGeneration) + requireCondition(t, updated, v1alpha3.ScheduledRunConditionAccepted, metav1.ConditionTrue, "Accepted") + requireCondition(t, updated, v1alpha3.ScheduledRunConditionResolvedRefs, metav1.ConditionTrue, "ResolvedRefs") + requireCondition(t, updated, v1alpha3.ScheduledRunConditionReady, metav1.ConditionTrue, "Scheduled") + require.Contains(t, apiMeta.FindStatusCondition(updated.Status.Conditions, v1alpha3.ScheduledRunConditionReady).Message, "11111111-1111-1111-1111-111111111111") +} + +func TestReconcileAppliesDefaultsAndPause(t *testing.T) { + schedule := testSchedule() + schedule.Spec.Paused = new(true) + store := &fakeReconcilerStore{} + updated, err := reconcileOnce(t, testKube(t, schedule, testAgent("report")), store, schedule) + require.NoError(t, err) + require.Equal(t, "UTC", store.applied.Config.TimeZone) + require.Equal(t, 15*time.Minute, store.applied.Config.ExecutionTimeout.AsDuration()) + require.True(t, store.applied.Config.Paused) + require.Contains(t, apiMeta.FindStatusCondition(updated.Status.Conditions, v1alpha3.ScheduledRunConditionReady).Message, "paused") +} + +func TestReconcileReportsMissingAgentButPersists(t *testing.T) { + schedule := testSchedule() + store := &fakeReconcilerStore{} + updated, err := reconcileOnce(t, testKube(t, schedule), store, schedule) + require.NoError(t, err) + require.Equal(t, 1, store.applies) + requireCondition(t, updated, v1alpha3.ScheduledRunConditionResolvedRefs, metav1.ConditionFalse, "AgentNotFound") + requireCondition(t, updated, v1alpha3.ScheduledRunConditionReady, metav1.ConditionTrue, "Scheduled") +} + +func TestReconcileRejectsInvalidScheduleWithoutTouchingStore(t *testing.T) { + for name, mutate := range map[string]func(*v1alpha3.ScheduledRunSpec){ + "cron": func(spec *v1alpha3.ScheduledRunSpec) { spec.Schedule = "61 * * * *" }, + "time zone": func(spec *v1alpha3.ScheduledRunSpec) { spec.TimeZone = new("Mars/Olympus") }, + "local": func(spec *v1alpha3.ScheduledRunSpec) { spec.TimeZone = new("Local") }, + "timeout": func(spec *v1alpha3.ScheduledRunSpec) { + spec.ExecutionTimeout = &metav1.Duration{Duration: time.Nanosecond} + }, + } { + t.Run(name, func(t *testing.T) { + schedule := testSchedule() + mutate(&schedule.Spec) + store := &fakeReconcilerStore{} + updated, err := reconcileOnce(t, testKube(t, schedule, testAgent("report")), store, schedule) + require.NoError(t, err) + require.Zero(t, store.applies) + requireCondition(t, updated, v1alpha3.ScheduledRunConditionAccepted, metav1.ConditionFalse, "InvalidSchedule") + requireCondition(t, updated, v1alpha3.ScheduledRunConditionReady, metav1.ConditionFalse, "NotAccepted") + }) + } +} + +func TestReconcileSurfacesStoreFailure(t *testing.T) { + schedule := testSchedule() + store := &fakeReconcilerStore{applyErr: errors.New("database unavailable")} + updated, err := reconcileOnce(t, testKube(t, schedule, testAgent("report")), store, schedule) + require.ErrorContains(t, err, "database unavailable") + requireCondition(t, updated, v1alpha3.ScheduledRunConditionAccepted, metav1.ConditionTrue, "Accepted") + requireCondition(t, updated, v1alpha3.ScheduledRunConditionReady, metav1.ConditionFalse, "PersistenceFailed") +} + +func TestReconcileRetiresDeletedObject(t *testing.T) { + store := &fakeReconcilerStore{} + _, err := NewReconciler(testKube(t), store).Reconcile(t.Context(), ctrl.Request{NamespacedName: types.NamespacedName{Namespace: "team-a", Name: "nightly"}}) + require.NoError(t, err) + require.Equal(t, []string{schedules.KubernetesCreator("team-a", "nightly")}, store.retired) + require.Zero(t, store.applies) +} + +func TestRequestsForAgentSelectsReferencingSchedules(t *testing.T) { + referencing := testSchedule() + other := testSchedule() + other.Name, other.Spec.AgentRef = "weekly", corev1.LocalObjectReference{Name: "other"} + elsewhere := testSchedule() + elsewhere.Namespace = "team-b" + kube := testKube(t, referencing, other, elsewhere) + requests := NewReconciler(kube, &fakeReconcilerStore{}).requestsForAgent(t.Context(), testAgent("report")) + require.Equal(t, []ctrl.Request{{NamespacedName: client.ObjectKeyFromObject(referencing)}}, requests) +} + +func TestReconcileAddsFinalizerBeforePersisting(t *testing.T) { + schedule := testSchedule() + store := &fakeReconcilerStore{} + updated, err := reconcileOnce(t, testKube(t, schedule, testAgent("report")), store, schedule) + require.NoError(t, err) + require.Contains(t, updated.Finalizers, retirementFinalizer) + require.Equal(t, 1, store.applies) +} + +// A deletion that happened while no controller ran is still pending on restart: +// the finalizer holds the object, so the first reconcile retires the row. +func TestReconcileRetiresObjectDeletedWhileDown(t *testing.T) { + schedule := testSchedule() + schedule.Finalizers = []string{retirementFinalizer} + schedule.DeletionTimestamp = &metav1.Time{Time: time.Now()} + store := &fakeReconcilerStore{} + kube := testKube(t, schedule) + _, err := NewReconciler(kube, store).Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKeyFromObject(schedule)}) + require.NoError(t, err) + require.Equal(t, []string{schedules.KubernetesCreator("team-a", "nightly")}, store.retired) + require.Zero(t, store.applies) + // Removing the last finalizer lets the API server delete the object. + require.True(t, apierrors.IsNotFound(kube.Get(t.Context(), client.ObjectKeyFromObject(schedule), &v1alpha3.ScheduledRun{}))) +} + +func TestReconcileKeepsFinalizerWhenRetirementFails(t *testing.T) { + schedule := testSchedule() + schedule.Finalizers = []string{retirementFinalizer} + schedule.DeletionTimestamp = &metav1.Time{Time: time.Now()} + kube := testKube(t, schedule) + _, err := NewReconciler(kube, &fakeReconcilerStore{retireErr: errors.New("database unavailable")}).Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKeyFromObject(schedule)}) + require.ErrorContains(t, err, "database unavailable") + current := &v1alpha3.ScheduledRun{} + require.NoError(t, kube.Get(t.Context(), client.ObjectKeyFromObject(schedule), current)) + require.Contains(t, current.Finalizers, retirementFinalizer) +} diff --git a/go/core/internal/database/scheduled_runs.go b/go/core/internal/database/scheduled_runs.go index ad8d506eec..4e5fa0ff7d 100644 --- a/go/core/internal/database/scheduled_runs.go +++ b/go/core/internal/database/scheduled_runs.go @@ -42,34 +42,13 @@ func (c *Client) FindScheduledRunRequest(ctx context.Context, creator, requestID // matches, or ErrIdempotencyConflict otherwise. Paused schedules have no due time; // execution is left to the scheduler. func (c *Client) CreateScheduledRun(ctx context.Context, request *apiv1.ScheduledRun, requestID string, hash []byte) (*apiv1.ScheduledRun, error) { - schedule := proto.CloneOf(request) - id := uuid.New() - schedule.Id, schedule.Etag = id.String(), uuid.NewString() - schedule.CreatedAt, schedule.UpdatedAt = nil, nil - schedule.NextExecutionTime, schedule.DeletedAt = nil, nil - data, err := proto.Marshal(schedule) - if err != nil { - return nil, fmt.Errorf("failed to encode schedule: %w", err) - } var row scheduledRunRow - err = c.withTx(ctx, func(tx pgx.Tx) error { - row, err = queryOne(ctx, tx, ` - INSERT INTO scheduled_run (id, creator, request_id, request_hash, data) - VALUES ($1, $2, $3, $4, $5) - ON CONFLICT (creator, request_id) DO NOTHING RETURNING id, creator, request_hash, data, created_at, updated_at, next_execution_time, deleted_at - `, pgx.RowToStructByName[scheduledRunRow], id, schedule.Creator, requestID, hash, data) - if err != nil { - return err - } - row.NextExecutionTime, err = nextExecutionTime(schedule.Config, row.CreatedAt) - if err != nil || row.NextExecutionTime == nil { - return err - } - // Keep the first cron time atomic with creation, using the insert's clock. - return advanceScheduledRun(ctx, tx, row.ID, row.NextExecutionTime) + err := c.withTx(ctx, func(tx pgx.Tx) (err error) { + row, err = insertScheduledRun(ctx, tx, request, requestID, hash) + return err }) - if errors.Is(err, pgx.ErrNoRows) { - return c.FindScheduledRunRequest(ctx, schedule.Creator, requestID, hash) + if errors.Is(err, ErrConflict) { + return c.FindScheduledRunRequest(ctx, request.GetCreator(), requestID, hash) } if err != nil { return nil, fmt.Errorf("failed to create schedule: %w", err) @@ -168,22 +147,7 @@ func (c *Client) DeleteScheduledRun(ctx context.Context, id uuid.UUID, creator s if err != nil { return err } - if row.DeletedAt != nil { - result = row - return nil - } - schedule, err := toScheduledRun(row) - if err != nil { - // Identity and ownership come from columns, not the damaged payload. - result, err = saveScheduledRun(ctx, tx, row.ID, row.Data, nil, true) - return err - } - schedule.Etag, schedule.NextExecutionTime = uuid.NewString(), nil - data, err := proto.Marshal(schedule) - if err != nil { - return err - } - result, err = saveScheduledRun(ctx, tx, row.ID, data, nil, true) + result, err = tombstoneScheduledRun(ctx, tx, row) return err }) if err != nil { @@ -202,6 +166,180 @@ func (c *Client) DeleteScheduledRun(ctx context.Context, id uuid.UUID, creator s return schedule, nil } +// ApplyScheduledRun converges the creator's schedules on one declared schedule +// identified by requestID: it creates the schedule when absent and replaces the +// configuration when it differs, bumping the etag only on change. The agent is +// fixed at creation; a different agent returns ErrFailedPrecondition. +// Other live schedules of the same creator are tombstoned in the same +// transaction, so a declarative owner whose identity changed (for example a +// recreated Kubernetes object) cannot leave the previous schedule firing. A +// tombstoned requestID returns ErrFailedPrecondition because deletion is final +// per request; a concurrent insert of the same requestID returns ErrConflict. +// Like UpdateScheduledRun, the next occurrence only moves when the schedule, +// time zone, or pause setting changes. +func (c *Client) ApplyScheduledRun(ctx context.Context, request *apiv1.ScheduledRun, requestID string) (*apiv1.ScheduledRun, error) { + desired := proto.CloneOf(request) + if desired.GetCreator() == "" || requestID == "" { + return nil, fmt.Errorf("schedule creator and request ID are required: %w", ErrFailedPrecondition) + } + hash, err := scheduledrun.RequestHash(desired.Agent, desired.Config) + if err != nil { + return nil, fmt.Errorf("failed to hash schedule request: %w", err) + } + var result scheduledRunRow + err = c.withTx(ctx, func(tx pgx.Tx) error { + row, err := queryOne(ctx, tx, ` + SELECT id, creator, request_hash, data, created_at, updated_at, next_execution_time, deleted_at + FROM scheduled_run WHERE creator = $1 AND request_id = $2 FOR UPDATE + `, pgx.RowToStructByName[scheduledRunRow], desired.Creator, requestID) + switch { + case errors.Is(err, pgx.ErrNoRows): + result, err = insertScheduledRun(ctx, tx, desired, requestID, hash) + case err != nil: + return err + case row.DeletedAt != nil: + return fmt.Errorf("ScheduledRun request %s was deleted: %w", requestID, ErrFailedPrecondition) + default: + result, err = replaceScheduledRun(ctx, tx, row, desired) + } + if err != nil { + return err + } + return retireScheduledRuns(ctx, tx, desired.Creator, &requestID) + }) + if err != nil { + return nil, fmt.Errorf("failed to apply schedule: %w", err) + } + return toScheduledRun(result) +} + +// RetireScheduledRuns tombstones every live schedule of a creator, retaining +// execution history like DeleteScheduledRun. It succeeds when nothing is live. +func (c *Client) RetireScheduledRuns(ctx context.Context, creator string) error { + if creator == "" { + return fmt.Errorf("schedule creator is required: %w", ErrFailedPrecondition) + } + err := c.withTx(ctx, func(tx pgx.Tx) error { + return retireScheduledRuns(ctx, tx, creator, nil) + }) + if err != nil { + return fmt.Errorf("failed to retire schedules: %w", err) + } + return nil +} + +// insertScheduledRun stores a new schedule and its first due time using the +// insert's clock. Callers hold the transaction; a concurrent insert of the same +// creator/requestID returns ErrConflict. +func insertScheduledRun(ctx context.Context, tx pgx.Tx, request *apiv1.ScheduledRun, requestID string, hash []byte) (scheduledRunRow, error) { + schedule := proto.CloneOf(request) + id := uuid.New() + schedule.Id, schedule.Etag = id.String(), uuid.NewString() + schedule.CreatedAt, schedule.UpdatedAt = nil, nil + schedule.NextExecutionTime, schedule.DeletedAt = nil, nil + data, err := proto.Marshal(schedule) + if err != nil { + return scheduledRunRow{}, fmt.Errorf("failed to encode schedule: %w", err) + } + row, err := queryOne(ctx, tx, ` + INSERT INTO scheduled_run (id, creator, request_id, request_hash, data) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (creator, request_id) DO NOTHING RETURNING id, creator, request_hash, data, created_at, updated_at, next_execution_time, deleted_at + `, pgx.RowToStructByName[scheduledRunRow], id, schedule.Creator, requestID, hash, data) + if errors.Is(err, pgx.ErrNoRows) { + return scheduledRunRow{}, fmt.Errorf("ScheduledRun request %s was created concurrently: %w", requestID, ErrConflict) + } + if err != nil { + return scheduledRunRow{}, err + } + row.NextExecutionTime, err = nextExecutionTime(schedule.Config, row.CreatedAt) + if err != nil || row.NextExecutionTime == nil { + return row, err + } + // Keep the first cron time atomic with creation, using the insert's clock. + return row, advanceScheduledRun(ctx, tx, row.ID, row.NextExecutionTime) +} + +// replaceScheduledRun makes a locked live schedule match the desired +// configuration. An unchanged schedule is returned without a write. The agent +// is immutable because reserved executions resolve it when their session is +// reserved; changing it returns ErrFailedPrecondition. The request hash keeps +// fingerprinting the original creation inputs. +func replaceScheduledRun(ctx context.Context, tx pgx.Tx, row scheduledRunRow, desired *apiv1.ScheduledRun) (scheduledRunRow, error) { + schedule, err := toScheduledRun(row) + if err != nil { + return scheduledRunRow{}, err + } + if !proto.Equal(schedule.Agent, desired.Agent) { + return scheduledRunRow{}, fmt.Errorf("ScheduledRun %s agent cannot change: %w", row.ID, ErrFailedPrecondition) + } + if proto.Equal(schedule.Config, desired.Config) { + return row, nil + } + previous := schedule.Config + schedule.Config, schedule.Etag = proto.CloneOf(desired.Config), uuid.NewString() + data, err := proto.Marshal(schedule) + if err != nil { + return scheduledRunRow{}, err + } + result, err := queryOne(ctx, tx, ` + UPDATE scheduled_run SET data = $2, updated_at = statement_timestamp() + WHERE id = $1 RETURNING id, creator, request_hash, data, created_at, updated_at, next_execution_time, deleted_at + `, pgx.RowToStructByName[scheduledRunRow], row.ID, data) + if err != nil { + return scheduledRunRow{}, err + } + // Prompt and agent edits must not skip an occurrence already due for reservation. + if desired.Config.Schedule == previous.Schedule && desired.Config.TimeZone == previous.TimeZone && desired.Config.Paused == previous.Paused { + return result, nil + } + result.NextExecutionTime, err = nextExecutionTime(desired.Config, result.UpdatedAt) + if err != nil { + return scheduledRunRow{}, err + } + return result, advanceScheduledRun(ctx, tx, result.ID, result.NextExecutionTime) +} + +// retireScheduledRuns tombstones the creator's live schedules, keeping the one +// identified by keepRequestID when given. Rows are locked in ID order. +func retireScheduledRuns(ctx context.Context, tx pgx.Tx, creator string, keepRequestID *string) error { + rows, err := queryMany(ctx, tx, ` + SELECT id, creator, request_hash, data, created_at, updated_at, next_execution_time, deleted_at + FROM scheduled_run WHERE creator = $1 AND deleted_at IS NULL + AND ($2::text IS NULL OR request_id <> $2::text) + ORDER BY id FOR UPDATE + `, pgx.RowToStructByName[scheduledRunRow], creator, keepRequestID) + if err != nil { + return err + } + for _, row := range rows { + if _, err := tombstoneScheduledRun(ctx, tx, row); err != nil { + return err + } + } + return nil +} + +// tombstoneScheduledRun stops future scheduling for a locked schedule and +// returns the tombstone row. Repeated deletion returns the existing tombstone; +// a malformed payload is retained for repair while the columns record deletion. +func tombstoneScheduledRun(ctx context.Context, tx pgx.Tx, row scheduledRunRow) (scheduledRunRow, error) { + if row.DeletedAt != nil { + return row, nil + } + schedule, err := toScheduledRun(row) + if err != nil { + // Identity and ownership come from columns, not the damaged payload. + return saveScheduledRun(ctx, tx, row.ID, row.Data, nil, true) + } + schedule.Etag, schedule.NextExecutionTime = uuid.NewString(), nil + data, err := proto.Marshal(schedule) + if err != nil { + return scheduledRunRow{}, err + } + return saveScheduledRun(ctx, tx, row.ID, data, nil, true) +} + // TriggerScheduledRun reserves a manual execution for an owned schedule, including when // paused, without starting runtime work or changing the next scheduled occurrence. Reusing // requestID for that schedule returns the same execution, even after deletion; new diff --git a/go/core/internal/database/scheduled_runs_test.go b/go/core/internal/database/scheduled_runs_test.go index 80aa0e82a9..12f307c6b7 100644 --- a/go/core/internal/database/scheduled_runs_test.go +++ b/go/core/internal/database/scheduled_runs_test.go @@ -168,6 +168,97 @@ func TestScheduledRunRequestsSurviveEditAndDeletion(t *testing.T) { require.ErrorIs(t, err, ErrNotFound) } +func TestApplyScheduledRunConvergesDeclaredSchedule(t *testing.T) { + c := NewClient(setupTestDB(t), "public") + sessionFixture(t, c, t.Context(), "team-a", "scheduled-revision", "report", "runtime") + const creator = "scheduledrun.api.kagent.dev/team-a/nightly" + desired := &apiv1.ScheduledRun{ + Creator: creator, + Agent: &apiv1.ResourceReference{Namespace: "team-a", Name: "report"}, + Config: scheduledrun.Normalize(&apiv1.ScheduledRunConfig{Schedule: "* * * * *", Prompt: "declared prompt"}), + } + created, err := c.ApplyScheduledRun(t.Context(), desired, "uid-1") + require.NoError(t, err) + require.NotNil(t, created.NextExecutionTime) + require.Equal(t, creator, created.Creator) + + // An unchanged apply is a no-op: same etag, same update time. + same, err := c.ApplyScheduledRun(t.Context(), desired, "uid-1") + require.NoError(t, err) + require.True(t, proto.Equal(created, same)) + + // A prompt edit changes the etag but keeps the pending occurrence. + desired.Config.Prompt = "edited prompt" + edited, err := c.ApplyScheduledRun(t.Context(), desired, "uid-1") + require.NoError(t, err) + require.Equal(t, created.Id, edited.Id) + require.NotEqual(t, created.Etag, edited.Etag) + require.Equal(t, "edited prompt", edited.Config.Prompt) + require.True(t, created.NextExecutionTime.AsTime().Equal(edited.NextExecutionTime.AsTime())) + + // Pausing clears the due time. + desired.Config.Paused = true + paused, err := c.ApplyScheduledRun(t.Context(), desired, "uid-1") + require.NoError(t, err) + require.Nil(t, paused.NextExecutionTime) + + // A reserved firing keeps its target: the agent cannot change in place, and + // the rejected apply leaves the stored schedule untouched. + execution, err := c.TriggerScheduledRun(t.Context(), uuid.MustParse(created.Id), creator, "manual") + require.NoError(t, err) + desired.Agent.Name = "other" + _, err = c.ApplyScheduledRun(t.Context(), desired, "uid-1") + require.ErrorIs(t, err, ErrFailedPrecondition) + unchanged, err := c.GetScheduledRun(t.Context(), uuid.MustParse(created.Id), creator) + require.NoError(t, err) + require.True(t, proto.Equal(paused, unchanged)) + reserved, err := c.ReserveScheduledRunExecutionSession(t.Context(), uuid.MustParse(execution.Id), creator) + require.NoError(t, err) + session, err := c.GetSession(t.Context(), reserved.SessionId, creator) + require.NoError(t, err) + require.Equal(t, "report", session.GetAgent().GetName()) + + // The owning object is recreated with a new UID and another agent: the old + // row is retired and a new identity starts, retaining the old history. + desired.Config.Paused = false + replaced, err := c.ApplyScheduledRun(t.Context(), desired, "uid-2") + require.NoError(t, err) + require.NotEqual(t, created.Id, replaced.Id) + require.NotNil(t, replaced.NextExecutionTime) + old, err := c.GetScheduledRun(t.Context(), uuid.MustParse(created.Id), creator) + require.NoError(t, err) + require.NotNil(t, old.DeletedAt) + live, err := c.ListScheduledRuns(t.Context(), ScheduledRunQuery{Creator: creator, Limit: 50}) + require.NoError(t, err) + require.Len(t, live, 1) + require.Equal(t, replaced.Id, live[0].Id) + retained, err := c.GetScheduledRunExecution(t.Context(), uuid.MustParse(execution.Id), creator) + require.NoError(t, err) + require.Equal(t, created.Id, retained.ScheduledRunId) + + // Deletion is final per request identity. + _, err = c.ApplyScheduledRun(t.Context(), desired, "uid-1") + require.ErrorIs(t, err, ErrFailedPrecondition) + + // Retiring the owner tombstones every live row and is idempotent. + require.NoError(t, c.RetireScheduledRuns(t.Context(), creator)) + require.NoError(t, c.RetireScheduledRuns(t.Context(), creator)) + live, err = c.ListScheduledRuns(t.Context(), ScheduledRunQuery{Creator: creator, Limit: 50}) + require.NoError(t, err) + require.Empty(t, live) + gone, err := c.GetScheduledRun(t.Context(), uuid.MustParse(replaced.Id), creator) + require.NoError(t, err) + require.NotNil(t, gone.DeletedAt) + require.Nil(t, gone.NextExecutionTime) + + // Other creators are untouched. + other, _ := createTestSchedule(t, c) + require.NoError(t, c.RetireScheduledRuns(t.Context(), creator)) + kept, err := c.GetScheduledRun(t.Context(), uuid.MustParse(other.Id), other.Creator) + require.NoError(t, err) + require.Nil(t, kept.DeletedAt) +} + func TestScheduledRunSQLTimestamps(t *testing.T) { db := setupTestDB(t) c := NewClient(db, "public") diff --git a/go/core/internal/service/scheduledrun/service.go b/go/core/internal/service/scheduledrun/service.go index 99046c9730..b012f77f18 100644 --- a/go/core/internal/service/scheduledrun/service.go +++ b/go/core/internal/service/scheduledrun/service.go @@ -2,7 +2,6 @@ package scheduledrun import ( "context" - "crypto/sha256" "encoding/base64" "errors" "time" @@ -59,14 +58,11 @@ func (s *Service) Create(ctx context.Context, request CreateRequest) (*apiv1.Sch return nil, err } // Hash original normalized inputs, not the mutable persisted configuration. - data, err := (proto.MarshalOptions{Deterministic: true}).Marshal(&apiv1.CreateScheduledRunRequest{ - Agent: request.Agent, Config: config, - }) + hash, err := schedules.RequestHash(request.Agent, config) if err != nil { return nil, serviceerrors.NewInvalidArgument("Invalid schedule inputs", err) } - hash := sha256.Sum256(data) - existing, err := s.store.FindScheduledRunRequest(ctx, creator, request.RequestID, hash[:]) + existing, err := s.store.FindScheduledRunRequest(ctx, creator, request.RequestID, hash) if err == nil { return existing, nil } @@ -83,7 +79,7 @@ func (s *Service) Create(ctx context.Context, request CreateRequest) (*apiv1.Sch if err := s.validateTarget(ctx, schedule); err != nil { return nil, err } - result, err := s.store.CreateScheduledRun(ctx, schedule, request.RequestID, hash[:]) + result, err := s.store.CreateScheduledRun(ctx, schedule, request.RequestID, hash) return result, storeError(err) } diff --git a/go/core/pkg/app/app.go b/go/core/pkg/app/app.go index 331251462d..ed40901263 100644 --- a/go/core/pkg/app/app.go +++ b/go/core/pkg/app/app.go @@ -369,6 +369,9 @@ func Run(ctx context.Context, input *configv1.ControllerConfig, opts Options) er gateway, cfg.GetScheduledRuns().GetExecutionInterval().AsDuration())); err != nil { return fmt.Errorf("add scheduled run controller: %w", err) } + if err := scheduledruncontroller.NewReconciler(manager.GetClient(), store).SetupWithManager(manager); err != nil { + return fmt.Errorf("set up ScheduledRun reconciler: %w", err) + } sandboxTemplates := kubecrud.NewService(manager.GetClient(), authorizer, &kagentv1alpha3.SandboxTemplate{}, &kagentv1alpha3.SandboxTemplateList{}, kagentv1alpha3.SandboxTemplateKind) guests, err := sandboxservice.NewGuestDialer(cfg.GetSubstrate().GetRouterUrl(), authenticator) if err != nil { diff --git a/go/core/test/e2e/README.md b/go/core/test/e2e/README.md index d5cb2b9717..666699b6e1 100644 --- a/go/core/test/e2e/README.md +++ b/go/core/test/e2e/README.md @@ -210,6 +210,12 @@ reachable IPv4 address. Its Service and EndpointSlice route to the receiver. `TestMCPInteraction` starts `mockmcp` on the same reachable host, registers it as a `RemoteMCPServer`, and verifies an actual `tools/call` request. +`TestScheduledRunResource` declares a paused `ScheduledRun` resource, waits for +it to become Ready, and reads the persisted schedule through gRPC as the +resource's synthetic creator. It triggers one execution, applies a prompt edit +to the same schedule identity, and checks that deletion retires the schedule +while retaining its execution history. + `TestSessionContextCompaction` clones the `kagent` Harness into one whose `spec.kagent.compaction` fires a sliding window after two turns, with a dedicated summarizer `ModelConfig` pointing at the same mock LLM behind a diff --git a/go/core/test/e2e/scheduled_run_resource_test.go b/go/core/test/e2e/scheduled_run_resource_test.go new file mode 100644 index 0000000000..82010aa759 --- /dev/null +++ b/go/core/test/e2e/scheduled_run_resource_test.go @@ -0,0 +1,137 @@ +package e2e_test + +import ( + "context" + "testing" + "time" + + a2apb "github.com/a2aproject/a2a-go/v2/a2apb/v1" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/metadata" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/wait" + ctrlclient "sigs.k8s.io/controller-runtime/pkg/client" + + apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1" + schedules "github.com/kagent-dev/kagent/go/api/scheduledrun" + "github.com/kagent-dev/kagent/go/api/v1alpha3" +) + +// TestScheduledRunResource declares a schedule in Kubernetes and verifies the +// controller persists it under the resource's own creator, runs it, follows +// edits, and retires it on deletion. +func TestScheduledRunResource(t *testing.T) { + t.Parallel() + forEachHarness(t, func(t *testing.T, harness testHarness) { + t.Parallel() + template := createInteractionTemplate(t, harness, startInteractionMock(t)) + kube := interactionKubeClient(t) + resource := &v1alpha3.ScheduledRun{ + ObjectMeta: metav1.ObjectMeta{GenerateName: "declared-", Namespace: "kagent"}, + Spec: v1alpha3.ScheduledRunSpec{ + AgentRef: corev1.LocalObjectReference{Name: template}, + Schedule: "* * * * *", + Prompt: "What is 2+2?", + Paused: new(true), + }, + } + require.NoError(t, kube.Create(t.Context(), resource)) + t.Cleanup(func() { + if err := kube.Delete(context.Background(), resource); err != nil && !apierrors.IsNotFound(err) { + t.Errorf("delete ScheduledRun: %v", err) + } + }) + waitScheduledRunReady(t, kube, resource) + + // The row is owned by the resource, not by any human principal. + creator := schedules.KubernetesCreator(resource.Namespace, resource.Name) + conn := newControllerConn(t, interactionTarget(t)) + ctx, cancel := context.WithTimeout(metadata.AppendToOutgoingContext(t.Context(), "x-user-id", creator), 5*time.Minute) + t.Cleanup(cancel) + f := &scheduledFixture{ctx: ctx, schedules: apiv1.NewScheduledRunServiceClient(conn), sessions: apiv1.NewSessionServiceClient(conn), system: apiv1.NewSystemServiceClient(conn), tasks: a2apb.NewA2AServiceClient(conn)} + f.schedule = declaredSchedule(t, f) + require.Equal(t, template, f.schedule.GetAgent().GetName()) + require.Equal(t, "kagent", f.schedule.GetAgent().GetNamespace()) + require.True(t, f.schedule.GetConfig().GetPaused()) + require.Equal(t, "UTC", f.schedule.GetConfig().GetTimeZone()) + require.Equal(t, 15*time.Minute, f.schedule.GetConfig().GetExecutionTimeout().AsDuration()) + require.Nil(t, f.schedule.GetNextExecutionTime()) + humanCtx := metadata.AppendToOutgoingContext(t.Context(), "x-user-id", "e2e") + _, err := f.schedules.GetScheduledRun(humanCtx, &apiv1.GetScheduledRunRequest{ScheduledRunId: f.schedule.GetId()}) + require.Error(t, err, "a human principal must not see a declared schedule") + + execution := f.trigger(t) + t.Cleanup(func() { + if execution.GetSessionId() == "" { + return + } + cleanupCtx, cleanupCancel := context.WithTimeout(metadata.AppendToOutgoingContext(context.Background(), "x-user-id", creator), 2*time.Minute) + defer cleanupCancel() + if err := deleteIdleSession(cleanupCtx, f.sessions, execution.GetSessionId()); err != nil { + t.Errorf("delete declared session: %v", err) + } + }) + execution = f.waitExecution(t, execution.GetId(), apiv1.ScheduledRunExecutionState_SCHEDULED_RUN_EXECUTION_STATE_SUCCEEDED) + f.assertCompletedTask(t, execution) + + // A spec edit converges onto the same schedule identity. + require.NoError(t, kube.Get(t.Context(), ctrlclient.ObjectKeyFromObject(resource), resource)) + resource.Spec.Prompt = "What is 3+3?" + require.NoError(t, kube.Update(t.Context(), resource)) + waitScheduledRunReady(t, kube, resource) + edited := declaredSchedule(t, f) + require.Equal(t, f.schedule.GetId(), edited.GetId()) + require.Equal(t, "What is 3+3?", edited.GetConfig().GetPrompt()) + require.NotEqual(t, f.schedule.GetEtag(), edited.GetEtag()) + + // Deleting the resource retires the schedule and retains its history. + require.NoError(t, kube.Delete(t.Context(), resource)) + require.NoError(t, wait.PollUntilContextTimeout(ctx, time.Second, time.Minute, true, func(ctx context.Context) (bool, error) { + result, err := f.schedules.ListScheduledRuns(ctx, &apiv1.ListScheduledRunsRequest{}) + if err != nil { + return false, err + } + return len(result.GetScheduledRuns()) == 0, nil + }), "declared schedule should be retired") + tombstone, err := f.schedules.GetScheduledRun(ctx, &apiv1.GetScheduledRunRequest{ScheduledRunId: f.schedule.GetId()}) + require.NoError(t, err) + require.NotNil(t, tombstone.GetScheduledRun().GetDeletedAt()) + retained, err := f.schedules.GetScheduledRunExecution(ctx, &apiv1.GetScheduledRunExecutionRequest{ExecutionId: execution.GetId()}) + require.NoError(t, err) + require.Equal(t, execution.GetState(), retained.GetExecution().GetState()) + }) +} + +func declaredSchedule(t *testing.T, f *scheduledFixture) *apiv1.ScheduledRun { + t.Helper() + result, err := f.schedules.ListScheduledRuns(f.ctx, &apiv1.ListScheduledRunsRequest{}) + require.NoError(t, err) + require.Len(t, result.GetScheduledRuns(), 1, "a resource owns exactly one live schedule") + return result.GetScheduledRuns()[0] +} + +func waitScheduledRunReady(t *testing.T, kube ctrlclient.Client, resource *v1alpha3.ScheduledRun) { + t.Helper() + err := wait.PollUntilContextTimeout(t.Context(), time.Second, 2*time.Minute, true, func(ctx context.Context) (bool, error) { + if err := kube.Get(ctx, ctrlclient.ObjectKeyFromObject(resource), resource); err != nil { + return false, err + } + if resource.Status.ObservedGeneration != resource.Generation { + return false, nil + } + for _, condition := range resource.Status.Conditions { + if condition.Type == v1alpha3.ScheduledRunConditionReady { + return condition.Status == metav1.ConditionTrue, nil + } + } + return false, nil + }) + if err != nil { + t.Fatalf("wait for ScheduledRun %s/%s: %v; last conditions: %+v", resource.Namespace, resource.Name, err, resource.Status.Conditions) + } + for _, condition := range resource.Status.Conditions { + require.Equal(t, metav1.ConditionTrue, condition.Status, "%s: %s", condition.Type, condition.Message) + } +} diff --git a/helm/kagent-crds/templates/api.kagent.dev_scheduledruns.yaml b/helm/kagent-crds/templates/api.kagent.dev_scheduledruns.yaml new file mode 100644 index 0000000000..4a6ee81f51 --- /dev/null +++ b/helm/kagent-crds/templates/api.kagent.dev_scheduledruns.yaml @@ -0,0 +1,204 @@ +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.22.0 + name: scheduledruns.api.kagent.dev +spec: + group: api.kagent.dev + names: + categories: + - kagent + kind: ScheduledRun + listKind: ScheduledRunList + plural: scheduledruns + singular: scheduledrun + scope: Namespaced + versions: + - additionalPrinterColumns: + - jsonPath: .spec.agentRef.name + name: Agent + type: string + - jsonPath: .spec.schedule + name: Schedule + type: string + - jsonPath: .spec.paused + name: Paused + type: boolean + - jsonPath: .status.conditions[?(@.type=='Ready')].status + name: Ready + type: string + - jsonPath: .metadata.creationTimestamp + name: Age + type: date + name: v1alpha3 + schema: + openAPIV3Schema: + description: |- + ScheduledRun sends a prompt to an Agent on a cron schedule. Each occurrence + creates a fresh Session owned by the ScheduledRun rather than by a human user. + Task state and history belong to upstream A2A. + properties: + apiVersion: + description: |- + APIVersion defines the versioned schema of this representation of an object. + Servers should convert recognized schemas to the latest internal value, and + may reject unrecognized values. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources + type: string + kind: + description: |- + Kind is a string value representing the REST resource this object represents. + Servers may infer this from the endpoint the client submits requests to. + Cannot be updated. + In CamelCase. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds + type: string + metadata: + type: object + spec: + description: |- + ScheduledRunSpec describes when and what to send. The Agent reference is + local to the ScheduledRun's namespace. + properties: + agentRef: + description: |- + AgentRef names the Agent that receives each occurrence. It is immutable: + reserved occurrences resolve the schedule's Agent when they start, so a + change could send an earlier prompt to a different Agent. Recreate the + ScheduledRun to target another Agent. + properties: + name: + default: "" + description: |- + Name of the referent. + This field is effectively required, but due to backwards compatibility is + allowed to be empty. Instances of this type with an empty value here are + almost certainly wrong. + More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names + type: string + type: object + x-kubernetes-map-type: atomic + x-kubernetes-validations: + - message: agentRef.name must not be empty + rule: has(self.name) && self.name != '' + - message: agentRef is immutable + rule: self == oldSelf + executionTimeout: + default: 15m + description: |- + ExecutionTimeout bounds one occurrence, including queueing and runtime + provisioning. Omitted means 15 minutes. + type: string + x-kubernetes-validations: + - message: executionTimeout must be positive and at most 9223372036s + rule: duration(self) > duration('0s') && duration(self) <= duration('9223372036s') + paused: + default: false + description: |- + Paused keeps the schedule and its history without reserving new + occurrences. Omitted means false. + type: boolean + prompt: + description: Prompt is the user message sent at each occurrence. + maxLength: 32768 + minLength: 1 + pattern: \S + type: string + schedule: + description: |- + Schedule is a standard five-field cron expression evaluated in TimeZone. + Macros such as @hourly and seconds fields are not accepted. + maxLength: 256 + minLength: 1 + type: string + x-kubernetes-validations: + - message: schedule must contain five cron fields + rule: self.split(' ').filter(f, f != '').size() == 5 + timeZone: + description: TimeZone is an IANA time zone name such as Europe/Paris. + Omitted means UTC. + maxLength: 253 + minLength: 1 + type: string + required: + - agentRef + - prompt + - schedule + type: object + status: + description: ScheduledRunStatus reports control-plane persistence of a + ScheduledRun. + properties: + conditions: + items: + description: Condition contains details for one aspect of the current + state of this API Resource. + properties: + lastTransitionTime: + description: |- + lastTransitionTime is the last time the condition transitioned from one status to another. + This should be when the underlying condition changed. If that is not known, then using the time when the API field changed is acceptable. + format: date-time + type: string + message: + description: |- + message is a human readable message indicating details about the transition. + This may be an empty string. + maxLength: 32768 + type: string + observedGeneration: + description: |- + observedGeneration represents the .metadata.generation that the condition was set based upon. + For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date + with respect to the current state of the instance. + format: int64 + minimum: 0 + type: integer + reason: + description: |- + reason contains a programmatic identifier indicating the reason for the condition's last transition. + Producers of specific condition types may define expected values and meanings for this field, + and whether the values are considered a guaranteed API. + The value should be a CamelCase string. + This field may not be empty. + maxLength: 1024 + minLength: 1 + pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$ + type: string + status: + description: status of the condition, one of True, False, Unknown. + enum: + - "True" + - "False" + - Unknown + type: string + type: + description: type of condition in CamelCase or in foo.example.com/CamelCase. + maxLength: 316 + pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$ + type: string + required: + - lastTransitionTime + - message + - reason + - status + - type + type: object + maxItems: 3 + type: array + x-kubernetes-list-map-keys: + - type + x-kubernetes-list-type: map + observedGeneration: + format: int64 + type: integer + type: object + required: + - spec + type: object + served: true + storage: true + subresources: + status: {} diff --git a/helm/kagent/templates/NOTES.txt b/helm/kagent/templates/NOTES.txt index eaeff959a6..a2a3389688 100644 --- a/helm/kagent/templates/NOTES.txt +++ b/helm/kagent/templates/NOTES.txt @@ -36,7 +36,7 @@ ENABLED TOOLS: USEFUL COMMANDS: # View all kagent resources - kubectl -n {{ include "kagent.namespace" . }} get agents,agenttemplates,harnesses,modelconfigs,remotemcpservers + kubectl -n {{ include "kagent.namespace" . }} get agents,agenttemplates,harnesses,modelconfigs,remotemcpservers,scheduledruns # Get agent templates and harnesses kubectl -n {{ include "kagent.namespace" . }} get agents,agenttemplates,harnesses diff --git a/helm/kagent/templates/rbac/getter-role.yaml b/helm/kagent/templates/rbac/getter-role.yaml index 1157c787ce..0b152d737c 100644 --- a/helm/kagent/templates/rbac/getter-role.yaml +++ b/helm/kagent/templates/rbac/getter-role.yaml @@ -9,6 +9,7 @@ - modelconfigs - modelproviderconfigs - remotemcpservers + - scheduledruns verbs: - get - list @@ -29,6 +30,7 @@ - modelconfigs/finalizers - modelproviderconfigs/finalizers - remotemcpservers/finalizers + - scheduledruns/finalizers verbs: - update - apiGroups: @@ -45,6 +47,7 @@ - modelconfigs/status - modelproviderconfigs/status - remotemcpservers/status + - scheduledruns/status verbs: - get - patch diff --git a/helm/kagent/templates/rbac/writer-role.yaml b/helm/kagent/templates/rbac/writer-role.yaml index b00658b17a..15de9223a6 100644 --- a/helm/kagent/templates/rbac/writer-role.yaml +++ b/helm/kagent/templates/rbac/writer-role.yaml @@ -9,6 +9,7 @@ - modelconfigs - modelproviderconfigs - remotemcpservers + - scheduledruns verbs: - create - update @@ -39,6 +40,7 @@ - modelconfigs/finalizers - modelproviderconfigs/finalizers - remotemcpservers/finalizers + - scheduledruns/finalizers verbs: - update - apiGroups: diff --git a/helm/kagent/tests/rbac_test.yaml b/helm/kagent/tests/rbac_test.yaml index 687445f988..d81743d0af 100644 --- a/helm/kagent/tests/rbac_test.yaml +++ b/helm/kagent/tests/rbac_test.yaml @@ -142,7 +142,7 @@ tests: value: [api.kagent.dev] - equal: path: rules[0].resources - value: [harnesses, agents, sandboxtemplates, agenttemplates, modelconfigs, modelproviderconfigs, remotemcpservers] + value: [harnesses, agents, sandboxtemplates, agenttemplates, modelconfigs, modelproviderconfigs, remotemcpservers, scheduledruns] - equal: path: rules[1].apiGroups value: [kagent.dev] @@ -162,7 +162,7 @@ tests: value: [api.kagent.dev] - equal: path: rules[0].resources - value: [harnesses, agents, sandboxtemplates, agenttemplates, modelconfigs, modelproviderconfigs, remotemcpservers] + value: [harnesses, agents, sandboxtemplates, agenttemplates, modelconfigs, modelproviderconfigs, remotemcpservers, scheduledruns] - equal: path: rules[1].apiGroups value: [kagent.dev]