From d40449e665f79c866ba3b43ab9752bfa7899d091 Mon Sep 17 00:00:00 2001 From: Jonathan Jamroga Date: Thu, 2 Jul 2026 14:52:04 -0400 Subject: [PATCH] feat(substrate): bump agent substrate to 0.0.8 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adapts kagent for substrate v0.0.8's atespace-scoped ActorRef identity model (rename of ActorId→ActorRef{Atespace,Name} on all actor RPCs). Maps atespace 1:1 to the SandboxAgent/AgentHarness Kubernetes namespace, adds an EnsureAtespace idempotent helper, and updates the atenet-router Host header shape to include the atespace label. Also fixes a pre-existing kagent bug that PR #2109's ActorTemplate spec immutability change surfaced: SnapshotsConfig.{OnPause,OnCommit} were left zero-value in kagent's desired spec but the API server defaults them to "Full" on admission, causing apiequality.Semantic.DeepEqual to report drift every reconcile and hot-loop delete/recreate the ActorTemplate CR. Verified end-to-end on colima+kind with substrate v0.0.8 published charts: SandboxAgent (declarative Go) and AgentHarness (openclaw) both reach Ready=True and chat round-trip works. Signed-off-by: Jonathan Jamroga --- Makefile | 5 +- examples/substrate-openclaw/README.md | 25 ++++++++- go/api/httpapi/substrate.go | 1 + .../a2a/substrate_sandbox_transport.go | 2 +- go/core/internal/a2a/substrate_transport.go | 4 +- .../handlers/agentharness_gateway.go | 7 ++- .../handlers/agentharness_gateway_test.go | 6 +- .../internal/httpserver/handlers/substrate.go | 1 + go/core/pkg/sandboxbackend/async.go | 6 +- .../substrate/actor_reachability.go | 8 +-- .../substrate/actor_reachability_test.go | 9 ++- .../sandboxbackend/substrate/agent_actor.go | 31 ++++++----- .../substrate/agent_lifecycle.go | 6 ++ .../substrate/agentharness_actor.go | 30 ++++++---- .../pkg/sandboxbackend/substrate/client.go | 38 +++++++++---- .../sandboxbackend/substrate/client_test.go | 55 +++++++++++++++++++ .../sandboxbackend/substrate/delete_actor.go | 20 +++---- .../substrate/delete_actor_test.go | 2 +- .../pkg/sandboxbackend/substrate/gateway.go | 8 ++- .../sandboxbackend/substrate/gateway_test.go | 18 ++++-- .../substrate/lifecycle_actortemplate.go | 6 ++ .../substrate/lifecycle_delete.go | 7 ++- .../substrate/lifecycle_delete_test.go | 26 +++++++-- .../pkg/sandboxbackend/substrate/openclaw.go | 23 +++++--- .../sandboxbackend/substrate/openclaw_test.go | 4 +- go/go.mod | 6 +- go/go.sum | 12 ++-- helm/kagent-crds/Chart-template.yaml | 2 +- helm/kagent/Chart-template.yaml | 2 +- scripts/controller-digest-ldflags.sh | 8 ++- scripts/kind/setup-kind.sh | 15 ++++- 31 files changed, 289 insertions(+), 104 deletions(-) diff --git a/Makefile b/Makefile index 13b7d3b7c4..4befd16380 100644 --- a/Makefile +++ b/Makefile @@ -150,6 +150,7 @@ KMCP_VERSION ?= $(shell $(AWK) '/github\.com\/kagent-dev\/kmcp/ { print substr($ # Substrate SUBSTRATE_ENABLED ?= false SUBSTRATE_VERSION ?= $(shell $(AWK) '/github\.com\/kagent-dev\/substrate/ { print substr($$5, 2) }' go/go.mod) # Substrate version defaults to the replace target in go.mod +SUBSTRATE_REPO ?= oci://ghcr.io/kagent-dev/substrate/helm # Override for local dev when consuming a locally-published chart, e.g. oci://localhost:5001/kagent-dev/substrate/helm HELM_ACTION=upgrade --install @@ -453,8 +454,8 @@ helm-tools: ## Package all tool Helm charts into the dist folder .PHONY: helm-version helm-version: ## Stamp chart versions, update dependencies, and package kagent + kagent-crds helm-version: helm-cleanup helm-agents helm-tools - VERSION=$(VERSION) KMCP_VERSION=$(KMCP_VERSION) SUBSTRATE_VERSION=$(SUBSTRATE_VERSION) envsubst < helm/kagent-crds/Chart-template.yaml > helm/kagent-crds/Chart.yaml - VERSION=$(VERSION) KMCP_VERSION=$(KMCP_VERSION) SUBSTRATE_VERSION=$(SUBSTRATE_VERSION) envsubst < helm/kagent/Chart-template.yaml > helm/kagent/Chart.yaml + VERSION=$(VERSION) KMCP_VERSION=$(KMCP_VERSION) SUBSTRATE_VERSION=$(SUBSTRATE_VERSION) SUBSTRATE_REPO=$(SUBSTRATE_REPO) envsubst < helm/kagent-crds/Chart-template.yaml > helm/kagent-crds/Chart.yaml + VERSION=$(VERSION) KMCP_VERSION=$(KMCP_VERSION) SUBSTRATE_VERSION=$(SUBSTRATE_VERSION) SUBSTRATE_REPO=$(SUBSTRATE_REPO) envsubst < helm/kagent/Chart-template.yaml > helm/kagent/Chart.yaml helm dependency update helm/kagent helm dependency update helm/kagent-crds helm package -d $(HELM_DIST_FOLDER) helm/kagent-crds diff --git a/examples/substrate-openclaw/README.md b/examples/substrate-openclaw/README.md index ca62166fbf..b39c2afe57 100644 --- a/examples/substrate-openclaw/README.md +++ b/examples/substrate-openclaw/README.md @@ -6,14 +6,37 @@ Follow these instructions to install Substrate on a kind cluster. This feature a This assumes you've configured a kind cluster using `make create-kind-cluster`. -Create a `substrate-values.yaml` file: +### Image-pull args for a local (kind) registry + +atelet is what pulls the ActorTemplate container image for each golden actor. It uses +`go-containerregistry` directly (not containerd), so containerd's registry-mirror +config on the kind node does **not** apply to atelet — you have to tell atelet how to +reach the local kind registry with its own flags: ```yaml atelet: + # Skip the GCP application-default-credentials probe. Substrate defaults this + # to true (for GKE + Artifact Registry); on kind it just adds latency and a + # noisy log line before falling back to anonymous auth. + gcpAuthForImagePulls: false + + # Rewrite `localhost:PORT/...` image refs (which is what `make helm-install` + # renders when `--set registry=localhost:5001` is in effect) to a hostname + # that atelet's puller can actually resolve from inside its pod. atelet also + # applies `name.Insecure` for any ref that was originally `localhost:*`, so + # the rewritten `kind-registry:5000` ref is fetched over HTTP (kind-registry + # is `registry:2` with no TLS by default). Without both parts of this — the + # rewrite AND the insecure flag it triggers — atelet errors out with either + # `dial tcp [::1]:5001: connect: connection refused` (no rewrite), or + # `http: server gave HTTP response to HTTPS client` (rewrite without Insecure). extraArgs: - --localhost-registry-replacement=kind-registry:5000 ``` +When installing Substrate as a **subchart** of kagent (i.e. `--set substrate.enabled=true` +on the kagent chart), prefix these keys with `substrate.` — e.g. +`--set-json 'substrate.atelet.extraArgs=["--localhost-registry-replacement=kind-registry:5000"]'`. + Then install the Substrate platform and kagent: ```bash diff --git a/go/api/httpapi/substrate.go b/go/api/httpapi/substrate.go index 3029109f3a..f818dc4447 100644 --- a/go/api/httpapi/substrate.go +++ b/go/api/httpapi/substrate.go @@ -37,6 +37,7 @@ type SubstrateActorTemplateEntry struct { // SubstrateActorEntry is runtime state from ate-api (redis). type SubstrateActorEntry struct { ActorID string `json:"actorId"` + Atespace string `json:"atespace,omitempty"` Status string `json:"status"` ActorTemplateNamespace string `json:"actorTemplateNamespace,omitempty"` ActorTemplateName string `json:"actorTemplateName,omitempty"` diff --git a/go/core/internal/a2a/substrate_sandbox_transport.go b/go/core/internal/a2a/substrate_sandbox_transport.go index adb366c228..ba8317f25f 100644 --- a/go/core/internal/a2a/substrate_sandbox_transport.go +++ b/go/core/internal/a2a/substrate_sandbox_transport.go @@ -72,7 +72,7 @@ func (t *substrateSandboxSessionRoundTripper) RoundTrip(req *http.Request) (*htt // Proxy the A2A request through atenet-router to the session actor. The router // selects the actor by HTTP Host; actor ID comes from EnsureSessionActor above. - actorRT, err := newSubstrateAgentRoundTripper(t.routerURL, res.Handle.ID, t.base) + actorRT, err := newSubstrateAgentRoundTripper(t.routerURL, res.Handle.Atespace, res.Handle.ID, t.base) if err != nil { return nil, err } diff --git a/go/core/internal/a2a/substrate_transport.go b/go/core/internal/a2a/substrate_transport.go index ec6d0a1427..0b43a2fb0a 100644 --- a/go/core/internal/a2a/substrate_transport.go +++ b/go/core/internal/a2a/substrate_transport.go @@ -14,8 +14,8 @@ type substrateAgentRoundTripper struct { base http.RoundTripper } -func newSubstrateAgentRoundTripper(routerURL, actorID string, base http.RoundTripper) (http.RoundTripper, error) { - target, host, err := substrate.GatewayRouterTarget(routerURL, actorID) +func newSubstrateAgentRoundTripper(routerURL, atespace, actorID string, base http.RoundTripper) (http.RoundTripper, error) { + target, host, err := substrate.GatewayRouterTarget(routerURL, atespace, actorID) if err != nil { return nil, err } diff --git a/go/core/internal/httpserver/handlers/agentharness_gateway.go b/go/core/internal/httpserver/handlers/agentharness_gateway.go index c332b8db61..8328d0afbc 100644 --- a/go/core/internal/httpserver/handlers/agentharness_gateway.go +++ b/go/core/internal/httpserver/handlers/agentharness_gateway.go @@ -83,7 +83,7 @@ func (h *Handlers) HandleAgentHarnessGateway(w ErrorResponseWriter, r *http.Requ return } - target, upstreamHost, err := h.resolveSubstrateGatewayTarget(r.Context(), ensureRes.Handle.ID) + target, upstreamHost, err := h.resolveSubstrateGatewayTarget(r.Context(), ensureRes.Handle.Atespace, ensureRes.Handle.ID) if err != nil { log.Info("resolve substrate gateway target failed", "error", err) http.Error(w, err.Error(), http.StatusServiceUnavailable) @@ -99,19 +99,20 @@ func (h *Handlers) HandleAgentHarnessGateway(w ErrorResponseWriter, r *http.Requ // explicitly suspends a session via the suspend endpoint. } -func (h *Handlers) resolveSubstrateGatewayTarget(ctx context.Context, actorID string) (*url.URL, string, error) { +func (h *Handlers) resolveSubstrateGatewayTarget(ctx context.Context, atespace, actorID string) (*url.URL, string, error) { cfg := h.AgentHarnessGateway if cfg == nil { return nil, "", fmt.Errorf("substrate gateway is not configured") } actorID = strings.TrimSpace(actorID) - target, host, err := substrate.GatewayRouterTarget(cfg.AtenetRouterURL, actorID) + target, host, err := substrate.GatewayRouterTarget(cfg.AtenetRouterURL, atespace, actorID) if err != nil { return nil, "", fmt.Errorf("substrate actor %q: %w", actorID, err) } ctrllog.FromContext(ctx).WithName("agentharness-gateway").Info( "proxying via atenet-router", + "atespace", atespace, "actor", actorID, "router", target.String(), "host", host, diff --git a/go/core/internal/httpserver/handlers/agentharness_gateway_test.go b/go/core/internal/httpserver/handlers/agentharness_gateway_test.go index faf6a81418..6f1ebbcb02 100644 --- a/go/core/internal/httpserver/handlers/agentharness_gateway_test.go +++ b/go/core/internal/httpserver/handlers/agentharness_gateway_test.go @@ -14,7 +14,7 @@ import ( func TestACPProxyForwardsToAtenetRouterWithActorHost(t *testing.T) { t.Parallel() - const actorHost = "ahr-kagent-my-claw.actors.resources.substrate.ate.dev" + const actorHost = "ahr-kagent-my-claw.kagent.actors.resources.substrate.ate.dev" var gotHost, gotAuth, gotPath string upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -56,9 +56,9 @@ func TestACPProxyForwardsToAtenetRouterWithActorHost(t *testing.T) { func TestACPProxyRewriteTargetsAtenetRouterHost(t *testing.T) { t.Parallel() - const actorHost = "ahr-kagent-my-claw.actors.resources.substrate.ate.dev" + const actorHost = "ahr-kagent-my-claw.kagent.actors.resources.substrate.ate.dev" - target, host, err := substrate.GatewayRouterTarget(substrate.DefaultAtenetRouterURL, "ahr-kagent-my-claw") + target, host, err := substrate.GatewayRouterTarget(substrate.DefaultAtenetRouterURL, "kagent", "ahr-kagent-my-claw") if err != nil { t.Fatal(err) } diff --git a/go/core/internal/httpserver/handlers/substrate.go b/go/core/internal/httpserver/handlers/substrate.go index 06045b56b3..d3665c3202 100644 --- a/go/core/internal/httpserver/handlers/substrate.go +++ b/go/core/internal/httpserver/handlers/substrate.go @@ -205,6 +205,7 @@ func (h *SubstrateHandler) listAteAPIState(ctx context.Context, namespaces []str func actorEntryFromPB(a *ateapipb.Actor) api.SubstrateActorEntry { return api.SubstrateActorEntry{ ActorID: a.GetActorId(), + Atespace: a.GetAtespace(), Status: substrate.ActorStatusLabel(a.GetStatus()), ActorTemplateNamespace: a.GetActorTemplateNamespace(), ActorTemplateName: a.GetActorTemplateName(), diff --git a/go/core/pkg/sandboxbackend/async.go b/go/core/pkg/sandboxbackend/async.go index a8cc3bf172..fac9a95125 100644 --- a/go/core/pkg/sandboxbackend/async.go +++ b/go/core/pkg/sandboxbackend/async.go @@ -9,8 +9,12 @@ import ( // Handle is the opaque identifier an AsyncBackend uses to address a sandbox // it owns on an external control plane. Persisted in AgentHarness.Status.BackendRef. +// +// For Substrate backends, Atespace scopes the ID — an actor's identity on +// substrate is (Atespace, ID). Non-substrate backends may leave Atespace empty. type Handle struct { - ID string + ID string + Atespace string } // EnsureResult is returned by EnsureAgentHarness. Endpoint (if set) is surfaced diff --git a/go/core/pkg/sandboxbackend/substrate/actor_reachability.go b/go/core/pkg/sandboxbackend/substrate/actor_reachability.go index e32b808e8d..024e4eeb47 100644 --- a/go/core/pkg/sandboxbackend/substrate/actor_reachability.go +++ b/go/core/pkg/sandboxbackend/substrate/actor_reachability.go @@ -18,7 +18,7 @@ const ( ) type actorGetter interface { - GetActor(ctx context.Context, actorID string) (*ateapipb.Actor, error) + GetActor(ctx context.Context, atespace, actorID string) (*ateapipb.Actor, error) } // waitForActorReachableViaAtenet blocks until ate-api reports the actor RUNNING and @@ -27,7 +27,7 @@ func waitForActorReachableViaAtenet( ctx context.Context, actors actorGetter, httpClient *http.Client, - routerURL, actorID string, + routerURL, atespace, actorID string, ) error { if actors == nil { return fmt.Errorf("substrate ate-api client is required") @@ -43,7 +43,7 @@ func waitForActorReachableViaAtenet( waitCtx, cancel := context.WithTimeout(ctx, defaultActorReachabilityTimeout) defer cancel() - target, host, err := GatewayRouterTarget(routerURL, actorID) + target, host, err := GatewayRouterTarget(routerURL, atespace, actorID) if err != nil { return err } @@ -53,7 +53,7 @@ func waitForActorReachableViaAtenet( defer ticker.Stop() for { - actor, getErr := actors.GetActor(waitCtx, actorID) + actor, getErr := actors.GetActor(waitCtx, atespace, actorID) if getErr == nil && actor.GetStatus() == ateapipb.Actor_STATUS_RUNNING { statusCode, probeErr := probeActorViaAtenetRouter(waitCtx, httpClient, probeURL, host) if probeErr == nil && statusCode < http.StatusInternalServerError { diff --git a/go/core/pkg/sandboxbackend/substrate/actor_reachability_test.go b/go/core/pkg/sandboxbackend/substrate/actor_reachability_test.go index eabc187130..55af8de3b8 100644 --- a/go/core/pkg/sandboxbackend/substrate/actor_reachability_test.go +++ b/go/core/pkg/sandboxbackend/substrate/actor_reachability_test.go @@ -15,7 +15,7 @@ type stubActorGetter struct { status atomic.Int32 } -func (s *stubActorGetter) GetActor(context.Context, string) (*ateapipb.Actor, error) { +func (s *stubActorGetter) GetActor(context.Context, string, string) (*ateapipb.Actor, error) { return &ateapipb.Actor{Status: ateapipb.Actor_Status(s.status.Load())}, nil } @@ -33,7 +33,7 @@ func TestProbeActorViaAtenetRouterSetsActorHost(t *testing.T) { context.Background(), srv.Client(), srv.URL+"/health", - "asr-kagent-demo.actors.resources.substrate.ate.dev", + "asr-kagent-demo.kagent.actors.resources.substrate.ate.dev", ) if err != nil { t.Fatalf("probeActorViaAtenetRouter() error = %v", err) @@ -41,7 +41,7 @@ func TestProbeActorViaAtenetRouterSetsActorHost(t *testing.T) { if status != http.StatusOK { t.Fatalf("status = %d, want 200", status) } - if gotHost != "asr-kagent-demo.actors.resources.substrate.ate.dev" { + if gotHost != "asr-kagent-demo.kagent.actors.resources.substrate.ate.dev" { t.Fatalf("Host = %q", gotHost) } } @@ -70,6 +70,7 @@ func TestWaitForActorReachableViaAtenetRetriesUntilHealthy(t *testing.T) { actors, srv.Client(), srv.URL, + "kagent", "asr-kagent-demo", ) if err != nil { @@ -104,6 +105,7 @@ func TestWaitForActorReachableViaAtenetWaitsForRunningStatus(t *testing.T) { actors, srv.Client(), srv.URL, + "kagent", "asr-kagent-demo", ) if err != nil { @@ -130,6 +132,7 @@ func TestWaitForActorReachableViaAtenetTimesOut(t *testing.T) { actors, srv.Client(), srv.URL, + "kagent", "asr-kagent-demo", ) if err == nil { diff --git a/go/core/pkg/sandboxbackend/substrate/agent_actor.go b/go/core/pkg/sandboxbackend/substrate/agent_actor.go index 7199da323a..4c32ce48d2 100644 --- a/go/core/pkg/sandboxbackend/substrate/agent_actor.go +++ b/go/core/pkg/sandboxbackend/substrate/agent_actor.go @@ -65,14 +65,18 @@ func (b *SandboxAgentActorBackend) EnsureSessionActor(ctx context.Context, sa *v return sandboxbackend.EnsureResult{}, err } tmplNS := sa.Namespace + atespace := sa.Namespace created := false - actor, err := b.client.GetActor(ctx, actorID) + actor, err := b.client.GetActor(ctx, atespace, actorID) if err != nil { if status.Code(err) != codes.NotFound { return sandboxbackend.EnsureResult{}, fmt.Errorf("substrate GetActor %q: %w", actorID, err) } - actor, err = b.client.CreateActor(ctx, actorID, tmplNS, tmplName) + if err := b.client.EnsureAtespace(ctx, atespace); err != nil { + return sandboxbackend.EnsureResult{}, fmt.Errorf("substrate EnsureAtespace %q: %w", atespace, err) + } + actor, err = b.client.CreateActor(ctx, atespace, actorID, tmplNS, tmplName) if err != nil { return sandboxbackend.EnsureResult{}, wrapCreateActorError(actorID, err) } @@ -85,13 +89,13 @@ func (b *SandboxAgentActorBackend) EnsureSessionActor(ctx context.Context, sa *v ateapipb.Actor_STATUS_PAUSED, ateapipb.Actor_STATUS_PAUSING: // PAUSED/PAUSING keep a node-local snapshot; ResumeActor brings them back // the same as a suspended actor. - _, err = b.client.ResumeActor(ctx, actorID) + _, err = b.client.ResumeActor(ctx, atespace, actorID) if err != nil { return sandboxbackend.EnsureResult{}, wrapResumeActorError(actorID, err) } } - if err := waitForActorReachableViaAtenet(ctx, b.client, nil, b.atenetRouterURL, actorID); err != nil { + if err := waitForActorReachableViaAtenet(ctx, b.client, nil, b.atenetRouterURL, atespace, actorID); err != nil { return sandboxbackend.EnsureResult{}, err } @@ -104,9 +108,9 @@ func (b *SandboxAgentActorBackend) EnsureSessionActor(ctx context.Context, sa *v b.scheduleReapOrphanedSessionActors(sa, actorID) } - host := ActorHost(actorID, "") + host := ActorHost(atespace, actorID, "") return sandboxbackend.EnsureResult{ - Handle: sandboxbackend.Handle{ID: actorID}, + Handle: sandboxbackend.Handle{ID: actorID, Atespace: atespace}, Endpoint: fmt.Sprintf("atenet-router Host %s", host), }, nil } @@ -175,7 +179,7 @@ func (b *SandboxAgentActorBackend) reapOrphanedSessionActors(ctx context.Context if actor.GetActorTemplateNamespace() == sa.Namespace && actor.GetActorTemplateName() == current.Name { continue // under the current desired config — a live session, keep } - if _, err := deleteActorIfSuspended(ctx, b.client, id); err != nil { + if _, err := deleteActorIfSuspended(ctx, b.client, sa.Namespace, id); err != nil { errs = append(errs, err) } } @@ -191,7 +195,8 @@ func (b *SandboxAgentActorBackend) SuspendSessionActor(ctx context.Context, sa * if err != nil { return err } - actor, err := b.client.GetActor(ctx, actorID) + atespace := sa.Namespace + actor, err := b.client.GetActor(ctx, atespace, actorID) if err != nil { if status.Code(err) == codes.NotFound { return nil @@ -200,7 +205,7 @@ func (b *SandboxAgentActorBackend) SuspendSessionActor(ctx context.Context, sa * } switch actor.GetStatus() { case ateapipb.Actor_STATUS_RUNNING, ateapipb.Actor_STATUS_RESUMING, ateapipb.Actor_STATUS_SUSPENDING: - if err := b.client.SuspendActor(ctx, actorID); err != nil && status.Code(err) != codes.NotFound { + if err := b.client.SuspendActor(ctx, atespace, actorID); err != nil && status.Code(err) != codes.NotFound { return fmt.Errorf("substrate SuspendActor %q: %w", actorID, err) } } @@ -208,11 +213,11 @@ func (b *SandboxAgentActorBackend) SuspendSessionActor(ctx context.Context, sa * } // DeleteSandboxAgentActor deletes a substrate actor by id. -func (b *SandboxAgentActorBackend) DeleteSandboxAgentActor(ctx context.Context, actorID string) (bool, error) { +func (b *SandboxAgentActorBackend) DeleteSandboxAgentActor(ctx context.Context, atespace, actorID string) (bool, error) { if strings.TrimSpace(actorID) == "" { return true, nil } - return deleteActor(ctx, b.client, actorID) + return deleteActor(ctx, b.client, atespace, actorID) } // DeleteSandboxAgentSessionActor deletes the actor(s) for a single chat session. Because the @@ -236,7 +241,7 @@ func (b *SandboxAgentActorBackend) DeleteSandboxAgentSessionActor(ctx context.Co continue } seen[actorID] = struct{}{} - done, err := b.DeleteSandboxAgentActor(ctx, actorID) + done, err := b.DeleteSandboxAgentActor(ctx, sa.Namespace, actorID) if err != nil { return false, err } @@ -319,7 +324,7 @@ func (b *SandboxAgentActorBackend) DeleteAllSandboxAgentActors(ctx context.Conte if !actorBelongsToSandboxAgent(sa, actor, prefix, ownedTemplates) { continue } - done, err := deleteActor(ctx, b.client, id) + done, err := deleteActor(ctx, b.client, sa.Namespace, id) if err != nil { return false, fmt.Errorf("delete substrate actor %q: %w", id, err) } diff --git a/go/core/pkg/sandboxbackend/substrate/agent_lifecycle.go b/go/core/pkg/sandboxbackend/substrate/agent_lifecycle.go index 623d54074b..5baca72289 100644 --- a/go/core/pkg/sandboxbackend/substrate/agent_lifecycle.go +++ b/go/core/pkg/sandboxbackend/substrate/agent_lifecycle.go @@ -113,6 +113,12 @@ func (p *Lifecycle) buildSandboxAgentActorTemplate( WorkerSelector: workerSelectorForPool(wpKey), SnapshotsConfig: atev1alpha1.SnapshotsConfig{ Location: sandboxAgentSnapshotsLocation(sa), + // Mirror substrate's CRD defaults so kagent's spec-drift check + // (apiequality.Semantic.DeepEqual) treats them as equal to the + // values the API server fills in on admission — otherwise kagent + // re-creates the ActorTemplate every reconcile in a hot loop. + OnPause: atev1alpha1.SnapshotScopeFull, + OnCommit: atev1alpha1.SnapshotScopeFull, }, }, } diff --git a/go/core/pkg/sandboxbackend/substrate/agentharness_actor.go b/go/core/pkg/sandboxbackend/substrate/agentharness_actor.go index 9017da67e4..c44547d701 100644 --- a/go/core/pkg/sandboxbackend/substrate/agentharness_actor.go +++ b/go/core/pkg/sandboxbackend/substrate/agentharness_actor.go @@ -54,13 +54,17 @@ func (b *AgentHarnessSessionActorBackend) EnsureSessionActor(ctx context.Context actorID := ActorID(ah) tmplNS, tmplName := generatedActorTemplateKey(ah) + atespace := ah.Namespace - actor, err := b.client.GetActor(ctx, actorID) + actor, err := b.client.GetActor(ctx, atespace, actorID) if err != nil { if status.Code(err) != codes.NotFound { return sandboxbackend.EnsureResult{}, fmt.Errorf("substrate GetActor %q: %w", actorID, err) } - actor, err = b.client.CreateActor(ctx, actorID, tmplNS, tmplName) + if err := b.client.EnsureAtespace(ctx, atespace); err != nil { + return sandboxbackend.EnsureResult{}, fmt.Errorf("substrate EnsureAtespace %q: %w", atespace, err) + } + actor, err = b.client.CreateActor(ctx, atespace, actorID, tmplNS, tmplName) if err != nil { return sandboxbackend.EnsureResult{}, fmt.Errorf("substrate CreateActor %q: %w", actorID, err) } @@ -73,18 +77,18 @@ func (b *AgentHarnessSessionActorBackend) EnsureSessionActor(ctx context.Context ateapipb.Actor_STATUS_PAUSED, ateapipb.Actor_STATUS_PAUSING: // PAUSED/PAUSING keep a node-local snapshot; ResumeActor brings them back // the same as a suspended actor. - if _, err = b.client.ResumeActor(ctx, actorID); err != nil { + if _, err = b.client.ResumeActor(ctx, atespace, actorID); err != nil { return sandboxbackend.EnsureResult{}, wrapResumeActorError(actorID, err) } } - if err := waitForActorReachableViaAtenet(ctx, b.client, nil, b.atenetRouterURL, actorID); err != nil { + if err := waitForActorReachableViaAtenet(ctx, b.client, nil, b.atenetRouterURL, atespace, actorID); err != nil { return sandboxbackend.EnsureResult{}, err } - host := ActorHost(actorID, "") + host := ActorHost(atespace, actorID, "") return sandboxbackend.EnsureResult{ - Handle: sandboxbackend.Handle{ID: actorID}, + Handle: sandboxbackend.Handle{ID: actorID, Atespace: atespace}, Endpoint: fmt.Sprintf("atenet-router Host %s", host), }, nil } @@ -97,7 +101,8 @@ func (b *AgentHarnessSessionActorBackend) SuspendSessionActor(ctx context.Contex return nil } actorID := ActorID(ah) - actor, err := b.client.GetActor(ctx, actorID) + atespace := ah.Namespace + actor, err := b.client.GetActor(ctx, atespace, actorID) if err != nil { if status.Code(err) == codes.NotFound { return nil @@ -106,7 +111,7 @@ func (b *AgentHarnessSessionActorBackend) SuspendSessionActor(ctx context.Contex } switch actor.GetStatus() { case ateapipb.Actor_STATUS_RUNNING, ateapipb.Actor_STATUS_RESUMING, ateapipb.Actor_STATUS_SUSPENDING: - if err := b.client.SuspendActor(ctx, actorID); err != nil && status.Code(err) != codes.NotFound { + if err := b.client.SuspendActor(ctx, atespace, actorID); err != nil && status.Code(err) != codes.NotFound { return fmt.Errorf("substrate SuspendActor %q: %w", actorID, err) } } @@ -114,11 +119,11 @@ func (b *AgentHarnessSessionActorBackend) SuspendSessionActor(ctx context.Contex } // DeleteSessionActor deletes a single per-session actor by id. -func (b *AgentHarnessSessionActorBackend) DeleteSessionActor(ctx context.Context, actorID string) (bool, error) { +func (b *AgentHarnessSessionActorBackend) DeleteSessionActor(ctx context.Context, atespace, actorID string) (bool, error) { if strings.TrimSpace(actorID) == "" { return true, nil } - return deleteActor(ctx, b.client, actorID) + return deleteActor(ctx, b.client, atespace, actorID) } // SessionActorState is the coarse lifecycle state of a chat session's actor as @@ -145,7 +150,8 @@ func (b *AgentHarnessSessionActorBackend) GetSessionActorState(ctx context.Conte return SessionActorStateMissing, fmt.Errorf("session id is required") } actorID := ActorID(ah) - actor, err := b.client.GetActor(ctx, actorID) + atespace := ah.Namespace + actor, err := b.client.GetActor(ctx, atespace, actorID) if err != nil { if status.Code(err) == codes.NotFound { return SessionActorStateMissing, nil @@ -181,7 +187,7 @@ func (b *AgentHarnessSessionActorBackend) DeleteAllAgentHarnessActors(ctx contex if id != prefix && !strings.HasPrefix(id, prefix+"-") { continue } - done, err := deleteActor(ctx, b.client, id) + done, err := deleteActor(ctx, b.client, ah.Namespace, id) if err != nil { return false, fmt.Errorf("delete substrate actor %q: %w", id, err) } diff --git a/go/core/pkg/sandboxbackend/substrate/client.go b/go/core/pkg/sandboxbackend/substrate/client.go index dc34de3742..dc883a275c 100644 --- a/go/core/pkg/sandboxbackend/substrate/client.go +++ b/go/core/pkg/sandboxbackend/substrate/client.go @@ -10,8 +10,10 @@ import ( "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/connectivity" "google.golang.org/grpc/credentials" + "google.golang.org/grpc/status" ) // Client wraps ate-api Control gRPC. @@ -121,21 +123,25 @@ func (c *Client) callCtx(ctx context.Context) (context.Context, context.CancelFu return context.WithTimeout(ctx, c.cfg.CallTimeout) } -func (c *Client) GetActor(ctx context.Context, actorID string) (*ateapipb.Actor, error) { +func actorRef(atespace, actorID string) *ateapipb.ActorRef { + return &ateapipb.ActorRef{Atespace: atespace, Name: actorID} +} + +func (c *Client) GetActor(ctx context.Context, atespace, actorID string) (*ateapipb.Actor, error) { ctx, cancel := c.callCtx(ctx) defer cancel() - resp, err := c.ControlClient.GetActor(ctx, &ateapipb.GetActorRequest{ActorId: actorID}) + resp, err := c.ControlClient.GetActor(ctx, &ateapipb.GetActorRequest{ActorRef: actorRef(atespace, actorID)}) if err != nil { return nil, err } return resp.GetActor(), nil } -func (c *Client) CreateActor(ctx context.Context, actorID, tmplNS, tmplName string) (*ateapipb.Actor, error) { +func (c *Client) CreateActor(ctx context.Context, atespace, actorID, tmplNS, tmplName string) (*ateapipb.Actor, error) { ctx, cancel := c.callCtx(ctx) defer cancel() resp, err := c.ControlClient.CreateActor(ctx, &ateapipb.CreateActorRequest{ - ActorId: actorID, + ActorRef: actorRef(atespace, actorID), ActorTemplateNamespace: tmplNS, ActorTemplateName: tmplName, }) @@ -145,26 +151,38 @@ func (c *Client) CreateActor(ctx context.Context, actorID, tmplNS, tmplName stri return resp.GetActor(), nil } -func (c *Client) ResumeActor(ctx context.Context, actorID string) (*ateapipb.Actor, error) { +func (c *Client) ResumeActor(ctx context.Context, atespace, actorID string) (*ateapipb.Actor, error) { ctx, cancel := c.callCtx(ctx) defer cancel() - resp, err := c.ControlClient.ResumeActor(ctx, &ateapipb.ResumeActorRequest{ActorId: actorID}) + resp, err := c.ControlClient.ResumeActor(ctx, &ateapipb.ResumeActorRequest{ActorRef: actorRef(atespace, actorID)}) if err != nil { return nil, err } return resp.GetActor(), nil } -func (c *Client) SuspendActor(ctx context.Context, actorID string) error { +func (c *Client) SuspendActor(ctx context.Context, atespace, actorID string) error { + ctx, cancel := c.callCtx(ctx) + defer cancel() + _, err := c.ControlClient.SuspendActor(ctx, &ateapipb.SuspendActorRequest{ActorRef: actorRef(atespace, actorID)}) + return err +} + +func (c *Client) DeleteActor(ctx context.Context, atespace, actorID string) error { ctx, cancel := c.callCtx(ctx) defer cancel() - _, err := c.ControlClient.SuspendActor(ctx, &ateapipb.SuspendActorRequest{ActorId: actorID}) + _, err := c.ControlClient.DeleteActor(ctx, &ateapipb.DeleteActorRequest{ActorRef: actorRef(atespace, actorID)}) return err } -func (c *Client) DeleteActor(ctx context.Context, actorID string) error { +// EnsureAtespace idempotently ensures the named atespace exists on the substrate side. +// Actors cannot be created into a nonexistent atespace (FailedPrecondition). +func (c *Client) EnsureAtespace(ctx context.Context, name string) error { ctx, cancel := c.callCtx(ctx) defer cancel() - _, err := c.ControlClient.DeleteActor(ctx, &ateapipb.DeleteActorRequest{ActorId: actorID}) + _, err := c.CreateAtespace(ctx, &ateapipb.CreateAtespaceRequest{Name: name}) + if err != nil && status.Code(err) == codes.AlreadyExists { + return nil + } return err } diff --git a/go/core/pkg/sandboxbackend/substrate/client_test.go b/go/core/pkg/sandboxbackend/substrate/client_test.go index e26063bb20..ced0c47c33 100644 --- a/go/core/pkg/sandboxbackend/substrate/client_test.go +++ b/go/core/pkg/sandboxbackend/substrate/client_test.go @@ -6,6 +6,7 @@ import ( "crypto/rsa" "crypto/tls" "crypto/x509" + "errors" "math/big" "net" "os" @@ -13,9 +14,12 @@ import ( "testing" "time" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "github.com/stretchr/testify/require" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials" + "google.golang.org/grpc/status" ) func TestAteAPITLSConfig(t *testing.T) { @@ -86,6 +90,57 @@ func TestBearerTokenFile(t *testing.T) { }) } +func TestEnsureAtespace(t *testing.T) { + t.Run("returns nil when substrate reports AlreadyExists", func(t *testing.T) { + fake := &createAtespaceFake{err: status.Error(codes.AlreadyExists, "Atespace kagent already exists")} + c := &Client{ControlClient: fake} + + require.NoError(t, c.EnsureAtespace(context.Background(), "kagent")) + require.Equal(t, "kagent", fake.lastName) + }) + + t.Run("returns nil on successful create", func(t *testing.T) { + fake := &createAtespaceFake{} + c := &Client{ControlClient: fake} + + require.NoError(t, c.EnsureAtespace(context.Background(), "kagent")) + }) + + t.Run("propagates non-AlreadyExists errors", func(t *testing.T) { + fake := &createAtespaceFake{err: status.Error(codes.Internal, "boom")} + c := &Client{ControlClient: fake} + + err := c.EnsureAtespace(context.Background(), "kagent") + require.Error(t, err) + require.Equal(t, codes.Internal, status.Code(err)) + }) + + t.Run("propagates non-gRPC errors", func(t *testing.T) { + fake := &createAtespaceFake{err: errors.New("dial failed")} + c := &Client{ControlClient: fake} + + err := c.EnsureAtespace(context.Background(), "kagent") + require.Error(t, err) + require.Contains(t, err.Error(), "dial failed") + }) +} + +// createAtespaceFake is a partial ControlClient stand-in that captures the last +// CreateAtespace request and returns a preset error. All other methods panic. +type createAtespaceFake struct { + ateapipb.ControlClient + lastName string + err error +} + +func (f *createAtespaceFake) CreateAtespace(_ context.Context, in *ateapipb.CreateAtespaceRequest, _ ...grpc.CallOption) (*ateapipb.CreateAtespaceResponse, error) { + f.lastName = in.GetName() + if f.err != nil { + return nil, f.err + } + return &ateapipb.CreateAtespaceResponse{Atespace: &ateapipb.Atespace{Name: in.GetName()}}, nil +} + func newTestTLSCert(t *testing.T) tls.Certificate { t.Helper() key, err := rsa.GenerateKey(rand.Reader, 2048) diff --git a/go/core/pkg/sandboxbackend/substrate/delete_actor.go b/go/core/pkg/sandboxbackend/substrate/delete_actor.go index 9c860407c4..5939d753e0 100644 --- a/go/core/pkg/sandboxbackend/substrate/delete_actor.go +++ b/go/core/pkg/sandboxbackend/substrate/delete_actor.go @@ -11,7 +11,7 @@ import ( // deleteActor performs at most one mutating ate-api step per call. // Returns true when the actor no longer exists. Callers should requeue until true. -func deleteActor(ctx context.Context, c *Client, actorID string) (bool, error) { +func deleteActor(ctx context.Context, c *Client, atespace, actorID string) (bool, error) { if actorID == "" { return true, nil } @@ -19,7 +19,7 @@ func deleteActor(ctx context.Context, c *Client, actorID string) (bool, error) { return false, fmt.Errorf("substrate ate-api client is required") } - actor, err := c.GetActor(ctx, actorID) + actor, err := c.GetActor(ctx, atespace, actorID) if err != nil { if status.Code(err) == codes.NotFound { return true, nil @@ -29,7 +29,7 @@ func deleteActor(ctx context.Context, c *Client, actorID string) (bool, error) { switch actor.GetStatus() { case ateapipb.Actor_STATUS_SUSPENDED, ateapipb.Actor_STATUS_UNSPECIFIED: - if err := c.DeleteActor(ctx, actorID); err != nil { + if err := c.DeleteActor(ctx, atespace, actorID); err != nil { if status.Code(err) == codes.NotFound { return true, nil } @@ -40,32 +40,32 @@ func deleteActor(ctx context.Context, c *Client, actorID string) (bool, error) { } return false, nil case ateapipb.Actor_STATUS_SUSPENDING: - _ = c.SuspendActor(ctx, actorID) + _ = c.SuspendActor(ctx, atespace, actorID) return false, nil case ateapipb.Actor_STATUS_RUNNING, ateapipb.Actor_STATUS_RESUMING: - if err := c.SuspendActor(ctx, actorID); err != nil && status.Code(err) != codes.NotFound { + if err := c.SuspendActor(ctx, atespace, actorID); err != nil && status.Code(err) != codes.NotFound { return false, fmt.Errorf("suspend actor %q: %w", actorID, err) } return false, nil case ateapipb.Actor_STATUS_PAUSED: - if _, err := c.ResumeActor(ctx, actorID); err != nil && status.Code(err) != codes.NotFound { + if _, err := c.ResumeActor(ctx, atespace, actorID); err != nil && status.Code(err) != codes.NotFound { return false, fmt.Errorf("resume paused actor %q before delete: %w", actorID, err) } return false, nil case ateapipb.Actor_STATUS_PAUSING: return false, nil default: - _ = c.SuspendActor(ctx, actorID) + _ = c.SuspendActor(ctx, atespace, actorID) return false, nil } } // deleteActorIfSuspended deletes an actor only when it is in a SUSPENDED (idle) state -func deleteActorIfSuspended(ctx context.Context, c *Client, actorID string) (done bool, err error) { +func deleteActorIfSuspended(ctx context.Context, c *Client, atespace, actorID string) (done bool, err error) { if actorID == "" || c == nil { return true, nil } - actor, err := c.GetActor(ctx, actorID) + actor, err := c.GetActor(ctx, atespace, actorID) if err != nil { if status.Code(err) == codes.NotFound { return true, nil @@ -74,7 +74,7 @@ func deleteActorIfSuspended(ctx context.Context, c *Client, actorID string) (don } switch actor.GetStatus() { case ateapipb.Actor_STATUS_SUSPENDED, ateapipb.Actor_STATUS_UNSPECIFIED: - if err := c.DeleteActor(ctx, actorID); err != nil && status.Code(err) != codes.NotFound { + if err := c.DeleteActor(ctx, atespace, actorID); err != nil && status.Code(err) != codes.NotFound { return false, fmt.Errorf("delete actor %q: %w", actorID, err) } return true, nil diff --git a/go/core/pkg/sandboxbackend/substrate/delete_actor_test.go b/go/core/pkg/sandboxbackend/substrate/delete_actor_test.go index c61ab4edf4..1309cf277d 100644 --- a/go/core/pkg/sandboxbackend/substrate/delete_actor_test.go +++ b/go/core/pkg/sandboxbackend/substrate/delete_actor_test.go @@ -4,7 +4,7 @@ import "testing" func TestDeleteActorEmptyID(t *testing.T) { t.Parallel() - done, err := deleteActor(t.Context(), nil, "") + done, err := deleteActor(t.Context(), nil, "", "") if err != nil { t.Fatalf("unexpected error: %v", err) } diff --git a/go/core/pkg/sandboxbackend/substrate/gateway.go b/go/core/pkg/sandboxbackend/substrate/gateway.go index b1631d94e5..7582334c28 100644 --- a/go/core/pkg/sandboxbackend/substrate/gateway.go +++ b/go/core/pkg/sandboxbackend/substrate/gateway.go @@ -10,7 +10,7 @@ import ( const DefaultAtenetRouterURL = "http://atenet-router.ate-system.svc:80" // GatewayRouterTarget returns the atenet-router reverse-proxy URL and Host header for an actor. -func GatewayRouterTarget(routerURL, actorID string) (*url.URL, string, error) { +func GatewayRouterTarget(routerURL, atespace, actorID string) (*url.URL, string, error) { routerURL = strings.TrimSpace(routerURL) if routerURL == "" { routerURL = DefaultAtenetRouterURL @@ -19,6 +19,10 @@ func GatewayRouterTarget(routerURL, actorID string) (*url.URL, string, error) { if actorID == "" { return nil, "", fmt.Errorf("actor id is required") } + atespace = strings.TrimSpace(atespace) + if atespace == "" { + return nil, "", fmt.Errorf("atespace is required") + } target, err := url.Parse(routerURL) if err != nil { return nil, "", fmt.Errorf("parse atenet-router URL %q: %w", routerURL, err) @@ -26,6 +30,6 @@ func GatewayRouterTarget(routerURL, actorID string) (*url.URL, string, error) { if target.Scheme == "" { return nil, "", fmt.Errorf("atenet-router URL %q must include a scheme (http or https)", routerURL) } - host := ActorHost(actorID, "") + host := ActorHost(atespace, actorID, "") return target, host, nil } diff --git a/go/core/pkg/sandboxbackend/substrate/gateway_test.go b/go/core/pkg/sandboxbackend/substrate/gateway_test.go index 8d89cf7e5b..9dc9044193 100644 --- a/go/core/pkg/sandboxbackend/substrate/gateway_test.go +++ b/go/core/pkg/sandboxbackend/substrate/gateway_test.go @@ -6,36 +6,44 @@ import ( func TestGatewayRouterTarget(t *testing.T) { t.Parallel() - target, host, err := GatewayRouterTarget("", "ahr-kagent-my-claw") + target, host, err := GatewayRouterTarget("", "kagent", "ahr-kagent-my-claw") if err != nil { t.Fatal(err) } if target.String() != DefaultAtenetRouterURL { t.Fatalf("target = %s, want %s", target, DefaultAtenetRouterURL) } - if host != "ahr-kagent-my-claw.actors.resources.substrate.ate.dev" { + if host != "ahr-kagent-my-claw.kagent.actors.resources.substrate.ate.dev" { t.Fatalf("host = %q", host) } } func TestGatewayRouterTargetCustomURL(t *testing.T) { t.Parallel() - target, host, err := GatewayRouterTarget("http://atenet-router.custom.svc:8080", "actor-1") + target, host, err := GatewayRouterTarget("http://atenet-router.custom.svc:8080", "kagent", "actor-1") if err != nil { t.Fatal(err) } if target.Host != "atenet-router.custom.svc:8080" { t.Fatalf("target host = %q", target.Host) } - if host != "actor-1.actors.resources.substrate.ate.dev" { + if host != "actor-1.kagent.actors.resources.substrate.ate.dev" { t.Fatalf("host = %q", host) } } func TestGatewayRouterTargetRejectsEmptyActor(t *testing.T) { t.Parallel() - _, _, err := GatewayRouterTarget("", "") + _, _, err := GatewayRouterTarget("", "kagent", "") if err == nil { t.Fatal("expected error for empty actor id") } } + +func TestGatewayRouterTargetRejectsEmptyAtespace(t *testing.T) { + t.Parallel() + _, _, err := GatewayRouterTarget("", "", "actor-1") + if err == nil { + t.Fatal("expected error for empty atespace") + } +} diff --git a/go/core/pkg/sandboxbackend/substrate/lifecycle_actortemplate.go b/go/core/pkg/sandboxbackend/substrate/lifecycle_actortemplate.go index ccad684c66..d375b2d969 100644 --- a/go/core/pkg/sandboxbackend/substrate/lifecycle_actortemplate.go +++ b/go/core/pkg/sandboxbackend/substrate/lifecycle_actortemplate.go @@ -181,6 +181,12 @@ func (p *Lifecycle) buildActorTemplate(ctx context.Context, ah *v1alpha2.AgentHa WorkerSelector: workerSelectorForPool(wpKey), SnapshotsConfig: atev1alpha1.SnapshotsConfig{ Location: substrateSnapshotsLocation(ah), + // Mirror substrate's CRD defaults so kagent's spec-drift check + // (apiequality.Semantic.DeepEqual) treats them as equal to the + // values the API server fills in on admission — otherwise kagent + // re-creates the ActorTemplate every reconcile in a hot loop. + OnPause: atev1alpha1.SnapshotScopeFull, + OnCommit: atev1alpha1.SnapshotScopeFull, }, }, } diff --git a/go/core/pkg/sandboxbackend/substrate/lifecycle_delete.go b/go/core/pkg/sandboxbackend/substrate/lifecycle_delete.go index 16f47d8b6b..66c0fb0b90 100644 --- a/go/core/pkg/sandboxbackend/substrate/lifecycle_delete.go +++ b/go/core/pkg/sandboxbackend/substrate/lifecycle_delete.go @@ -42,8 +42,13 @@ func (p *Lifecycle) CleanupGeneratedTemplate(ctx context.Context, ah *v1alpha2.A return true, nil } +// GoldenActorAtespace is the reserved substrate atespace that per-template +// golden actors live in. Mirrors substrate's internal/resources.GoldenActorAtespace, +// duplicated here because that package is internal to the substrate module. +const GoldenActorAtespace = "ate-golden" + func deleteGoldenActor(ctx context.Context, ateClient *Client, actorID string) (bool, error) { - return deleteActor(ctx, ateClient, actorID) + return deleteActor(ctx, ateClient, GoldenActorAtespace, actorID) } func (p *Lifecycle) goldenActorID(ctx context.Context, tmplKey types.NamespacedName) (string, error) { diff --git a/go/core/pkg/sandboxbackend/substrate/lifecycle_delete_test.go b/go/core/pkg/sandboxbackend/substrate/lifecycle_delete_test.go index 4eaf9e13c7..3e766d40e2 100644 --- a/go/core/pkg/sandboxbackend/substrate/lifecycle_delete_test.go +++ b/go/core/pkg/sandboxbackend/substrate/lifecycle_delete_test.go @@ -25,19 +25,21 @@ type recordingActorClient struct { } func (r *recordingActorClient) GetActor(_ context.Context, in *ateapipb.GetActorRequest, _ ...grpc.CallOption) (*ateapipb.GetActorResponse, error) { - if slices.Contains(r.deleted, in.GetActorId()) { + name := in.GetActorRef().GetName() + if slices.Contains(r.deleted, name) { return nil, status.Error(codes.NotFound, "actor deleted") } return &ateapipb.GetActorResponse{ Actor: &ateapipb.Actor{ - ActorId: in.GetActorId(), - Status: ateapipb.Actor_STATUS_SUSPENDED, + ActorId: name, + Atespace: in.GetActorRef().GetAtespace(), + Status: ateapipb.Actor_STATUS_SUSPENDED, }, }, nil } func (r *recordingActorClient) DeleteActor(_ context.Context, in *ateapipb.DeleteActorRequest, _ ...grpc.CallOption) (*ateapipb.DeleteActorResponse, error) { - r.deleted = append(r.deleted, in.GetActorId()) + r.deleted = append(r.deleted, in.GetActorRef().GetName()) return &ateapipb.DeleteActorResponse{}, nil } @@ -73,6 +75,22 @@ func (r *recordingActorClient) DebugClear(context.Context, *ateapipb.DebugClearR panic("not used") } +func (r *recordingActorClient) CreateAtespace(_ context.Context, in *ateapipb.CreateAtespaceRequest, _ ...grpc.CallOption) (*ateapipb.CreateAtespaceResponse, error) { + return &ateapipb.CreateAtespaceResponse{Atespace: &ateapipb.Atespace{Name: in.GetName()}}, nil +} + +func (r *recordingActorClient) GetAtespace(context.Context, *ateapipb.GetAtespaceRequest, ...grpc.CallOption) (*ateapipb.GetAtespaceResponse, error) { + panic("not used") +} + +func (r *recordingActorClient) ListAtespaces(context.Context, *ateapipb.ListAtespacesRequest, ...grpc.CallOption) (*ateapipb.ListAtespacesResponse, error) { + panic("not used") +} + +func (r *recordingActorClient) DeleteAtespace(context.Context, *ateapipb.DeleteAtespaceRequest, ...grpc.CallOption) (*ateapipb.DeleteAtespaceResponse, error) { + panic("not used") +} + func TestLifecycleCleanupGeneratedTemplate_DeletesGoldenActor(t *testing.T) { t.Parallel() scheme := runtime.NewScheme() diff --git a/go/core/pkg/sandboxbackend/substrate/openclaw.go b/go/core/pkg/sandboxbackend/substrate/openclaw.go index c3d508bcb7..e93f87f1e8 100644 --- a/go/core/pkg/sandboxbackend/substrate/openclaw.go +++ b/go/core/pkg/sandboxbackend/substrate/openclaw.go @@ -52,13 +52,17 @@ func (b *ClawBackend) EnsureAgentHarness(ctx context.Context, ah *v1alpha2.Agent actorID := ActorID(ah) tmplNS, tmplName := generatedActorTemplateKey(ah) + atespace := ah.Namespace - actor, err := b.client.GetActor(ctx, actorID) + actor, err := b.client.GetActor(ctx, atespace, actorID) if err != nil { if status.Code(err) != codes.NotFound { return sandboxbackend.EnsureResult{}, fmt.Errorf("substrate GetActor %q: %w", actorID, err) } - actor, err = b.client.CreateActor(ctx, actorID, tmplNS, tmplName) + if err := b.client.EnsureAtespace(ctx, atespace); err != nil { + return sandboxbackend.EnsureResult{}, fmt.Errorf("substrate EnsureAtespace %q: %w", atespace, err) + } + actor, err = b.client.CreateActor(ctx, atespace, actorID, tmplNS, tmplName) if err != nil { return sandboxbackend.EnsureResult{}, fmt.Errorf("substrate CreateActor %q: %w", actorID, err) } @@ -68,7 +72,7 @@ func (b *ClawBackend) EnsureAgentHarness(ctx context.Context, ah *v1alpha2.Agent case ateapipb.Actor_STATUS_RUNNING, ateapipb.Actor_STATUS_RESUMING: // already active or waking case ateapipb.Actor_STATUS_SUSPENDED, ateapipb.Actor_STATUS_UNSPECIFIED: - actor, err = b.client.ResumeActor(ctx, actorID) + actor, err = b.client.ResumeActor(ctx, atespace, actorID) if err != nil { return sandboxbackend.EnsureResult{}, fmt.Errorf("substrate ResumeActor %q: %w", actorID, err) } @@ -79,7 +83,7 @@ func (b *ClawBackend) EnsureAgentHarness(ctx context.Context, ah *v1alpha2.Agent endpoint := substrateConnectionEndpoint(ah.Namespace, ah.Name, actor) return sandboxbackend.EnsureResult{ - Handle: sandboxbackend.Handle{ID: actorID}, + Handle: sandboxbackend.Handle{ID: actorID, Atespace: atespace}, Endpoint: endpoint, }, nil } @@ -88,7 +92,7 @@ func (b *ClawBackend) GetStatus(ctx context.Context, h sandboxbackend.Handle) (m if h.ID == "" { return metav1.ConditionUnknown, "ActorHandleMissing", "no substrate actor id recorded yet" } - actor, err := b.client.GetActor(ctx, h.ID) + actor, err := b.client.GetActor(ctx, h.Atespace, h.ID) if err != nil { if status.Code(err) == codes.NotFound { return metav1.ConditionUnknown, "ActorNotFound", fmt.Sprintf("substrate actor %q not found", h.ID) @@ -102,7 +106,7 @@ func (b *ClawBackend) DeleteAgentHarness(ctx context.Context, h sandboxbackend.H if h.ID == "" { return true, nil } - done, err := deleteActor(ctx, b.client, h.ID) + done, err := deleteActor(ctx, b.client, h.Atespace, h.ID) if err != nil { return false, fmt.Errorf("substrate delete actor %q: %w", h.ID, err) } @@ -135,11 +139,12 @@ func ActorID(ah *v1alpha2.AgentHarness) string { } // ActorHost returns the atenet router Host header value for the actor. -func ActorHost(actorID string, suffix string) string { +// atespace is folded in as a DNS label so the router can parse it out of :authority. +func ActorHost(atespace, actorID string, suffix string) string { if suffix == "" { suffix = defaultActorHostSuffix } - return actorID + "." + suffix + return actorID + "." + atespace + "." + suffix } func generatedActorTemplateKey(ah *v1alpha2.AgentHarness) (string, string) { @@ -152,7 +157,7 @@ func substrateConnectionEndpoint(namespace, name string, actor *ateapipb.Actor) return "kagent gateway: " + gw } if actorID := strings.TrimSpace(actor.GetActorId()); actorID != "" { - return fmt.Sprintf("atenet-router Host %s (UI via kagent %s)", ActorHost(actorID, ""), gw) + return fmt.Sprintf("atenet-router Host %s (UI via kagent %s)", ActorHost(actor.GetAtespace(), actorID, ""), gw) } return fmt.Sprintf("kagent gateway: %s (actor status %s)", gw, actor.GetStatus()) } diff --git a/go/core/pkg/sandboxbackend/substrate/openclaw_test.go b/go/core/pkg/sandboxbackend/substrate/openclaw_test.go index 5e5b752f5f..e1a7ab1eb1 100644 --- a/go/core/pkg/sandboxbackend/substrate/openclaw_test.go +++ b/go/core/pkg/sandboxbackend/substrate/openclaw_test.go @@ -26,8 +26,8 @@ func TestActorID(t *testing.T) { } func TestActorHost(t *testing.T) { - got := ActorHost("ahr-kagent-my-claw", "") - if got != "ahr-kagent-my-claw.actors.resources.substrate.ate.dev" { + got := ActorHost("kagent", "ahr-kagent-my-claw", "") + if got != "ahr-kagent-my-claw.kagent.actors.resources.substrate.ate.dev" { t.Fatalf("ActorHost = %q", got) } } diff --git a/go/go.mod b/go/go.mod index 10b447672b..b0f4588dca 100644 --- a/go/go.mod +++ b/go/go.mod @@ -371,8 +371,8 @@ require ( github.com/tidwall/sjson v1.2.5 // indirect github.com/timakin/bodyclose v0.0.0-20260129054331-73d1f95b84b4 // indirect github.com/timonwong/loggercheck v0.11.0 // indirect - github.com/tklauser/go-sysconf v0.3.16 // indirect - github.com/tklauser/numcpus v0.11.0 // indirect + github.com/tklauser/go-sysconf v0.4.0 // indirect + github.com/tklauser/numcpus v0.12.0 // indirect github.com/tomarrell/wrapcheck/v2 v2.12.0 // indirect github.com/tommy-muehle/go-mnd/v2 v2.5.1 // indirect github.com/ultraware/funlen v0.2.0 // indirect @@ -450,4 +450,4 @@ require ( tool sigs.k8s.io/kube-api-linter/cmd/golangci-lint-kube-api-linter -replace github.com/agent-substrate/substrate => github.com/kagent-dev/substrate v0.0.7 +replace github.com/agent-substrate/substrate => github.com/kagent-dev/substrate v0.0.8 diff --git a/go/go.sum b/go/go.sum index b4cca2391e..5e1ecfcb9a 100644 --- a/go/go.sum +++ b/go/go.sum @@ -500,8 +500,8 @@ github.com/kagent-dev/mockllm v0.0.5 h1:mm9Ml3NH6/E/YKVMgMwWYMNsNGkDze6I6TC0ppHZ github.com/kagent-dev/mockllm v0.0.5/go.mod h1:tDLemRsTZa1NdHaDbg3sgFk9cT1QWvMPlBtLVD6I2mA= github.com/kagent-dev/mockmcp v0.0.0-20260520211643-dcd475b74085 h1:5bYxRc6K6+JiFMYGSJpQ7y18PzTGXUmUtVXY7Fqh0Ew= github.com/kagent-dev/mockmcp v0.0.0-20260520211643-dcd475b74085/go.mod h1:vTMste7c+XiDOQyhe6iGPKLbMy3chY/h2fL3Dox5fAE= -github.com/kagent-dev/substrate v0.0.7 h1:zf4NryZxpVy/tLhlWQSPgINkoPAse00e38+EH2UQ/oE= -github.com/kagent-dev/substrate v0.0.7/go.mod h1:2ri5mTz714LWinayPQG1d4G/kMuzbqvRXlrUrpc7N9Y= +github.com/kagent-dev/substrate v0.0.8 h1:rWTLRrf148GQfqKFSTfZx5KQ8jxN7VsP0ICSJrH1eJ8= +github.com/kagent-dev/substrate v0.0.8/go.mod h1:2cvSnnHPwZRAhkrC2LH1bQd1tQBfL3p+zU6pZiHBUkA= github.com/karamaru-alpha/copyloopvar v1.2.2 h1:yfNQvP9YaGQR7VaWLYcfZUlRP2eo2vhExWKxD/fP6q0= github.com/karamaru-alpha/copyloopvar v1.2.2/go.mod h1:oY4rGZqZ879JkJMtX3RRkcXRkmUvH0x35ykgaKgsgJY= github.com/kisielk/errcheck v1.10.0 h1:Lvs/YAHP24YKg08LA8oDw2z9fJVme090RAXd90S+rrw= @@ -802,10 +802,10 @@ github.com/timakin/bodyclose v0.0.0-20260129054331-73d1f95b84b4 h1:SiHe5XLTn9sFW github.com/timakin/bodyclose v0.0.0-20260129054331-73d1f95b84b4/go.mod h1:sDHLK7rb/59v/ZxZ7KtymgcoxuUMxjXq8gtu9VMOK8M= github.com/timonwong/loggercheck v0.11.0 h1:jdaMpYBl+Uq9mWPXv1r8jc5fC3gyXx4/WGwTnnNKn4M= github.com/timonwong/loggercheck v0.11.0/go.mod h1:HEAWU8djynujaAVX7QI65Myb8qgfcZ1uKbdpg3ZzKl8= -github.com/tklauser/go-sysconf v0.3.16 h1:frioLaCQSsF5Cy1jgRBrzr6t502KIIwQ0MArYICU0nA= -github.com/tklauser/go-sysconf v0.3.16/go.mod h1:/qNL9xxDhc7tx3HSRsLWNnuzbVfh3e7gh/BmM179nYI= -github.com/tklauser/numcpus v0.11.0 h1:nSTwhKH5e1dMNsCdVBukSZrURJRoHbSEQjdEbY+9RXw= -github.com/tklauser/numcpus v0.11.0/go.mod h1:z+LwcLq54uWZTX0u/bGobaV34u6V7KNlTZejzM6/3MQ= +github.com/tklauser/go-sysconf v0.4.0 h1:7H0uAN+7RkwWRaxhYXDLqa5V3LPrJeV8wmD9dRUgPQU= +github.com/tklauser/go-sysconf v0.4.0/go.mod h1:8mTNWyog7H+MpKijp4VmKJAd2bbYQ2zuUwkYRbUArPI= +github.com/tklauser/numcpus v0.12.0 h1:NR85qdvHA9pFse3x3weVZ0r0ST8R6l5RHbZrlRaqob4= +github.com/tklauser/numcpus v0.12.0/go.mod h1:ABHeXzJnr/qqwguhClkZKT1/8VABcYrsyUiUGobwWJg= github.com/tomarrell/wrapcheck/v2 v2.12.0 h1:H/qQ1aNWz/eeIhxKAFvkfIA+N7YDvq6TWVFL27Of9is= github.com/tomarrell/wrapcheck/v2 v2.12.0/go.mod h1:AQhQuZd0p7b6rfW+vUwHm5OMCGgp63moQ9Qr/0BpIWo= github.com/tommy-muehle/go-mnd/v2 v2.5.1 h1:NowYhSdyE/1zwK9QCLeRb6USWdoif80Ie+v+yU8u1Zw= diff --git a/helm/kagent-crds/Chart-template.yaml b/helm/kagent-crds/Chart-template.yaml index f84bd5e93e..4ed70033f0 100644 --- a/helm/kagent-crds/Chart-template.yaml +++ b/helm/kagent-crds/Chart-template.yaml @@ -10,5 +10,5 @@ dependencies: condition: kmcp.enabled - name: substrate-crds version: ${SUBSTRATE_VERSION} - repository: oci://ghcr.io/kagent-dev/substrate/helm + repository: ${SUBSTRATE_REPO} condition: substrate.enabled diff --git a/helm/kagent/Chart-template.yaml b/helm/kagent/Chart-template.yaml index 5463c0b593..efa99d89da 100644 --- a/helm/kagent/Chart-template.yaml +++ b/helm/kagent/Chart-template.yaml @@ -10,7 +10,7 @@ dependencies: condition: kmcp.enabled - name: substrate version: ${SUBSTRATE_VERSION} - repository: oci://ghcr.io/kagent-dev/substrate/helm + repository: ${SUBSTRATE_REPO} condition: substrate.enabled - name: kagent-tools version: 0.2.1 diff --git a/scripts/controller-digest-ldflags.sh b/scripts/controller-digest-ldflags.sh index 498d069081..0a8c1e424d 100755 --- a/scripts/controller-digest-ldflags.sh +++ b/scripts/controller-digest-ldflags.sh @@ -19,6 +19,12 @@ CONTAINER_RUNTIME="${CONTAINER_RUNTIME:-docker}" TRANSLATOR_PKG="github.com/kagent-dev/kagent/go/core/internal/controller/translator/agent" SUBSTRATE_PKG="github.com/kagent-dev/kagent/go/core/pkg/sandboxbackend/substrate" MANIFEST_ACCEPT="application/vnd.oci.image.index.v1+json, application/vnd.docker.distribution.manifest.list.v2+json, application/vnd.oci.image.manifest.v1+json, application/vnd.docker.distribution.manifest.v2+json" +# Set DIGEST_CURL_INSECURE=true to skip TLS verification when resolving digests +# against a self-signed local registry. +CURL_INSECURE_ARG="" +if [[ "${DIGEST_CURL_INSECURE:-false}" == "true" ]]; then + CURL_INSECURE_ARG="-k" +fi : "${APP_IMG:?APP_IMG is required}" : "${APP_FULL_IMG:?APP_FULL_IMG is required}" @@ -54,7 +60,7 @@ registry_manifest_digest() { fi if ! headers="$( - curl -fsSI \ + curl -fsSI ${CURL_INSECURE_ARG} \ -H "Accept: ${MANIFEST_ACCEPT}" \ "${scheme}://${registry}/v2/${repository}/manifests/${tag}" )"; then diff --git a/scripts/kind/setup-kind.sh b/scripts/kind/setup-kind.sh index 70c898aa45..ac7f18b179 100755 --- a/scripts/kind/setup-kind.sh +++ b/scripts/kind/setup-kind.sh @@ -11,8 +11,11 @@ KIND_IMAGE_VERSION=${KIND_IMAGE_VERSION:-1.35.0} CONTAINER_RUNTIME=${CONTAINER_RUNTIME:-$(command -v podman >/dev/null 2>&1 && echo podman || echo docker)} # 1. Create registry container unless it already exists -reg_name='kind-registry' -reg_port='5001' +# Override REG_NAME / REG_PORT / REG_SCHEME to reuse an existing local registry +# (e.g. an HTTPS registry on another port) instead of creating a fresh kind-registry. +reg_name="${REG_NAME:-kind-registry}" +reg_port="${REG_PORT:-5001}" +reg_scheme="${REG_SCHEME:-http}" if [ "$("${CONTAINER_RUNTIME}" inspect -f '{{.State.Running}}' "${reg_name}" 2>/dev/null || true)" != 'true' ]; then "${CONTAINER_RUNTIME}" run \ -d --restart=always -p "127.0.0.1:${reg_port}:5000" --network bridge --name "${reg_name}" \ @@ -38,11 +41,16 @@ fi # # We want a consistent name that works from both ends, so we tell containerd to # alias localhost:${reg_port} to the registry container when pulling images +# Internal container port: registry:2 listens on 5000 by default. reg_port is the host-side +# mapping; containerd inside the kind node reaches the registry container by its docker +# network name on the internal port. +reg_internal_port="${REG_INTERNAL_PORT:-5000}" REGISTRY_DIR="/etc/containerd/certs.d/localhost:${reg_port}" for node in $(kind get nodes --name "${KIND_CLUSTER_NAME}"); do "${CONTAINER_RUNTIME}" exec "${node}" mkdir -p "${REGISTRY_DIR}" cat <