substrate

package
v0.0.0-...-74321ee Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 7, 2026 License: Apache-2.0 Imports: 40 Imported by: 0

Documentation

Index

Constants

View Source 
const DefaultAtenetRouterURL = "http://atenet-router.ate-system.svc:80"

DefaultAtenetRouterURL is the in-cluster HTTP endpoint for Substrate's Envoy router.

View Source 
const GoldenActorAtespace = "ate-golden"

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.

View Source 
const HarnessLabelKey = "kagent.dev/agent-harness"

HarnessLabelKey labels substrate lifecycle managed for an AgentHarness.

View Source 
const OpenClawGatewayPort = 18789

OpenClawGatewayPort is the loopback port the OpenClaw gateway listens on inside a substrate actor. It is a private implementation detail: the in-sandbox `openclaw acp` child connects to it over loopback, while the acp-shim owns the atenet ingress port (acpListenPort). kagent never reaches the gateway directly.

View Source 
const (
	SandboxAgentLabelKey = "kagent.dev/sandbox-agent"
)
View Source 
const (

	// Referenced by generated ActorTemplates to gate scheduling onto a WorkerPool.
	// The kagent Helm chart stamps it on the WorkerPool it manages;
	// externally-owned pools must carry it to remain eligible.
	WorkerPoolLabelKey = "kagent.dev/worker-pool"
)

Variables

View Source 
var (
	AcpSandboxOpenClawImageDigest string
	AcpSandboxHermesImageDigest   string
)

AcpSandboxOpenClawImageDigest and AcpSandboxHermesImageDigest are the link-time-injected image digests (sha256:...) for the acp-sandbox workload images, set via -X ...substrate.AcpSandbox*ImageDigest=... They are empty in source and in unit tests, in which case resolution returns an error rather than an unpinned ref.

View Source 
var ErrActorTemplateReconcilePending = errors.New("actor template reconciliation pending")

ErrActorTemplateReconcilePending indicates ActorTemplate reconciliation started a multi-step recreate (e.g. golden-actor deletion) and callers should requeue.

View Source 
var ErrNoFreeWorkers = errors.New("substrate worker pool has no free workers; try again later or increase WorkerPool replicas")

ErrNoFreeWorkers is returned when ate-api cannot assign a WorkerPool worker to resume an actor.

Functions

func ActorHost 

func ActorHost(atespace, actorID string, suffix string) string

ActorHost returns the atenet router Host header value for the actor. atespace is folded in as a DNS label so the router can parse it out of :authority.

func ActorID 

func ActorID(ah *v1alpha2.AgentHarness) string

ActorID returns a stable DNS-1123 actor id for this harness.

func ActorStatusLabel 

func ActorStatusLabel(status ateapipb.Actor_Status) string

ActorStatusLabel returns a stable human-readable actor status.

func GatewayRouterTarget 

func GatewayRouterTarget(routerURL, atespace, actorID string) (*url.URL, string, error)

GatewayRouterTarget returns the atenet-router reverse-proxy URL and Host header for an actor.

func HarnessNameFromLabels 

func HarnessNameFromLabels(labels map[string]string) string

HarnessNameFromLabels returns the AgentHarness name from generated lifecycle labels.

func ResolveCurrentActorTemplate 

func ResolveCurrentActorTemplate(ctx context.Context, kube client.Client, namespace, agentName string) (*atev1alpha1.ActorTemplate, error)

ResolveCurrentActorTemplate returns the ActorTemplate a SandboxAgent should currently serve from: the template matching the agent's CURRENT desired config whose golden is Ready, else the most-recently-desired Ready template (the previous config) while the desired one is still building — the blue-green pivot, with no downtime and an atomic flip once the new golden is Ready.

"Desired" is tracked by the kagent.dev/desired-generation annotation (the agent generation that last applied the template), NOT creationTimestamp. Creation time is wrong for a flip-back to a retained older config: that template's golden was built earlier, so by-creation ordering would keep serving the newer (now-undesired) config. The desired template is always re-applied with the current (highest) generation, so picking the highest-generation Ready template follows the current config in both directions. Falls back to the highest-generation template when none is Ready yet (first build). Returns (nil, nil) when no template exists.

func SandboxAgentActorID 

func SandboxAgentActorID(sa *v1alpha2.SandboxAgent) string

SandboxAgentActorID returns the legacy stable actor id prefix for a SandboxAgent.

func SandboxAgentNameFromLabels 

func SandboxAgentNameFromLabels(labels map[string]string) string

SandboxAgentNameFromLabels returns the SandboxAgent name from generated lifecycle labels.

func SandboxAgentSessionActorID 

func SandboxAgentSessionActorID(sa *v1alpha2.SandboxAgent, sessionID string) string

SandboxAgentSessionActorID returns the ate-api actor id for a SandboxAgent chat session, derived from the session alone: one session ⇔ one actor for the session's entire life, across config AND shape rollouts (the actor's template binding lives on the actor record, not in the id). The id keeps the agent prefix (asr-<ns>-<name>-) so per-agent cleanup still matches.

Types

type AgentHarnessLifecycle 

type AgentHarnessLifecycle interface {
	EnsureGeneratedTemplate(ctx context.Context, ah *v1alpha2.AgentHarness) (LifecycleState, error)
	CleanupGeneratedTemplate(ctx context.Context, ah *v1alpha2.AgentHarness) (bool, error)
}

AgentHarnessLifecycle is the substrate lifecycle surface used by the AgentHarness controller.

type AgentHarnessSessionActorBackend 

type AgentHarnessSessionActorBackend struct {
	// contains filtered or unexported fields
}

AgentHarnessSessionActorBackend manages the single shared ate-api actor for an AgentHarness. The AgentHarness is a template: its generated ActorTemplate is the golden snapshot, and one actor is spun from it per harness. Every chat is an ACP session inside that one actor's long-lived child process (created via session/new and resumed via session/load), so a Telegram/Slack gateway and its single getUpdates consumer also stay singular per harness.

func NewAgentHarnessSessionActorBackend 

func NewAgentHarnessSessionActorBackend(client *Client, atenetRouterURL string) *AgentHarnessSessionActorBackend

NewAgentHarnessSessionActorBackend returns a backend that ensures the shared AgentHarness actor on ate-api.

func (*AgentHarnessSessionActorBackend) DeleteAllAgentHarnessActors 

func (b *AgentHarnessSessionActorBackend) DeleteAllAgentHarnessActors(ctx context.Context, ah *v1alpha2.AgentHarness) (bool, error)

DeleteAllAgentHarnessActors deletes the legacy single harness actor and every per-session actor belonging to the AgentHarness. It is best-effort and returns false while any actor is still terminating.

func (*AgentHarnessSessionActorBackend) DeleteSessionActor 

func (b *AgentHarnessSessionActorBackend) DeleteSessionActor(ctx context.Context, atespace, actorID string) (bool, error)

DeleteSessionActor deletes a single per-session actor by id.

func (*AgentHarnessSessionActorBackend) EnsureSessionActor 

func (b *AgentHarnessSessionActorBackend) EnsureSessionActor(ctx context.Context, ah *v1alpha2.AgentHarness, sessionID string) (sandboxbackend.EnsureResult, error)

EnsureSessionActor creates (if needed) and resumes the harness's single shared actor, then waits for it to be reachable via atenet-router. The sessionID identifies the chat for logging only; chats are multiplexed as ACP sessions inside the one actor, so they all resolve to the same ActorID(ah).

func (*AgentHarnessSessionActorBackend) GetSessionActorState 

func (b *AgentHarnessSessionActorBackend) GetSessionActorState(ctx context.Context, ah *v1alpha2.AgentHarness, sessionID string) (SessionActorState, error)

GetSessionActorState reports whether the harness's shared actor is running, suspended, or not yet created.

func (*AgentHarnessSessionActorBackend) SuspendSessionActor 

func (b *AgentHarnessSessionActorBackend) SuspendSessionActor(ctx context.Context, ah *v1alpha2.AgentHarness, sessionID string) error

SuspendSessionActor checkpoints and frees the worker for the harness's shared actor. It is resumed automatically on the next EnsureSessionActor. Because the actor is shared, suspending affects every chat in the harness.

type AgentsBackend 

type AgentsBackend struct {
	Lifecycle *Lifecycle
	AteClient *Client
}

AgentsBackend implements sandboxbackend.Backend for declarative/BYO SandboxAgents on Agent Substrate.

func NewAgentsBackend 

func NewAgentsBackend(lifecycle *Lifecycle, ate *Client) *AgentsBackend

NewAgentsBackend returns a substrate sandbox backend for SandboxAgent resources.

func (*AgentsBackend) BuildSandbox 

func (b *AgentsBackend) BuildSandbox(ctx context.Context, in sandboxbackend.BuildInput) ([]client.Object, error)

func (*AgentsBackend) ComputeReady 

func (b *AgentsBackend) ComputeReady(ctx context.Context, cl client.Client, nn types.NamespacedName) (metav1.ConditionStatus, string, string)

func (*AgentsBackend) GetOwnedResourceTypes 

func (b *AgentsBackend) GetOwnedResourceTypes() []client.Object

func (*AgentsBackend) OwnedResourceTypesFor 

func (b *AgentsBackend) OwnedResourceTypesFor(_ v1alpha2.AgentObject) ([]client.Object, error)

OwnedResourceTypesFor returns no types: substrate ActorTemplates are intentionally excluded from the reconciler's generic prune so a config change does not delete the currently-serving template. A config change creates a new config-hashed template; superseded templates and their (suspended) goldens are stateful and pin no workers, so they are retained — not retired — and removed only when the SandboxAgent is deleted (DeleteAllSandboxAgentActors + CleanupSandboxAgentTemplate, plus owner-reference GC of the template objects). ActorTemplate remains in GetOwnedResourceTypes for watches.

func (*AgentsBackend) ReconcileActorTemplate 

func (b *AgentsBackend) ReconcileActorTemplate(ctx context.Context, desired client.Object) error

func (*AgentsBackend) SessionDBURL 

func (b *AgentsBackend) SessionDBURL(agent v1alpha2.AgentObject) string

SessionDBURL returns the durable-dir session-store URL the translator bakes into the rendered config (AgentConfig.session_db_url) before building the config Secret. The value is runtime-specific: python's google-adk DatabaseSessionService uses SQLAlchemy's async engine, so the URL must name an async driver (aiosqlite, a core google-adk dependency); the Go ADK's local store parses either form.

type ClawBackend 

type ClawBackend struct {
	// contains filtered or unexported fields
}

ClawBackend implements AsyncBackend for OpenClaw on Agent Substrate.

func NewOpenClawBackend 

func NewOpenClawBackend(client *Client, backend v1alpha2.AgentHarnessBackendType, recorder record.EventRecorder) *ClawBackend

NewOpenClawBackend returns a substrate backend for openclaw harness types.

func (*ClawBackend) DeleteAgentHarness 

func (b *ClawBackend) DeleteAgentHarness(ctx context.Context, h sandboxbackend.Handle) (bool, error)

func (*ClawBackend) EnsureAgentHarness 

func (b *ClawBackend) EnsureAgentHarness(ctx context.Context, ah *v1alpha2.AgentHarness) (sandboxbackend.EnsureResult, error)

func (*ClawBackend) GetStatus 

func (b *ClawBackend) GetStatus(ctx context.Context, h sandboxbackend.Handle) (metav1.ConditionStatus, string, string)

func (*ClawBackend) Name 

func (b *ClawBackend) Name() v1alpha2.AgentHarnessBackendType

func (*ClawBackend) OnAgentHarnessReady 

func (b *ClawBackend) OnAgentHarnessReady(_ context.Context, _ *v1alpha2.AgentHarness, _ sandboxbackend.Handle) error

type Client 

type Client struct {
	ateapipb.ControlClient
	// contains filtered or unexported fields
}

Client wraps ate-api Control gRPC.

func Dial 

func Dial(ctx context.Context, cfg Config) (*Client, error)

Dial connects to the ate-api server.

func (*Client) Close 

func (c *Client) Close() error

func (*Client) CreateActor 

func (c *Client) CreateActor(ctx context.Context, atespace, actorID, tmplNS, tmplName string) (*ateapipb.Actor, error)

func (*Client) DeleteActor 

func (c *Client) DeleteActor(ctx context.Context, atespace, actorID string) error

func (*Client) EnsureAtespace 

func (c *Client) EnsureAtespace(ctx context.Context, name string) error

EnsureAtespace idempotently ensures the named atespace exists on the substrate side. Actors cannot be created into a nonexistent atespace (FailedPrecondition).

func (*Client) GetActor 

func (c *Client) GetActor(ctx context.Context, atespace, actorID string) (*ateapipb.Actor, error)

func (*Client) ListActors 

func (c *Client) ListActors(ctx context.Context, atespace string) ([]*ateapipb.Actor, error)

ListActors returns all actors in the given atespace (empty atespace = all atespaces, including substrate's reserved golden atespace). The list API is paginated — pages are followed until the token drains, since a single page may miss actors.

func (*Client) ListWorkers 

func (c *Client) ListWorkers(ctx context.Context) ([]*ateapipb.Worker, error)

ListWorkers returns all workers reflected in ate-api.

func (*Client) ResumeActor 

func (c *Client) ResumeActor(ctx context.Context, atespace, actorID string) (*ateapipb.Actor, error)

func (*Client) SuspendActor 

func (c *Client) SuspendActor(ctx context.Context, atespace, actorID string) error

type Config 

type Config struct {
	// AteAPIEndpoint is a gRPC target (e.g. dns:///api.ate-system.svc:443).
	AteAPIEndpoint string
	// TokenFile is a path to a file containing a bearer token for ate-api.
	TokenFile   string
	Insecure    bool
	DialTimeout time.Duration
	CallTimeout time.Duration
}

Config holds connection settings for Agent Substrate ate-api.

type Lifecycle 

type Lifecycle struct {
	Client    client.Client
	Defaults  LifecycleDefaults
	AteClient *Client
}

Lifecycle reconciles the Kubernetes lifecycle that kagent owns for a substrate AgentHarness. WorkerPools are externally owned; this helper only resolves the selected WorkerPool.

func NewLifecycle 

func NewLifecycle(kube client.Client, defaults LifecycleDefaults, ateClient *Client) *Lifecycle

func (*Lifecycle) ActorTemplateReady 

func (p *Lifecycle) ActorTemplateReady(ctx context.Context, key types.NamespacedName) (bool, error)

ActorTemplateReady reports whether the ActorTemplate golden snapshot is ready.

func (*Lifecycle) CleanupGeneratedTemplate 

func (p *Lifecycle) CleanupGeneratedTemplate(ctx context.Context, ah *v1alpha2.AgentHarness) (bool, error)

CleanupGeneratedTemplate removes external Substrate actors that Kubernetes garbage collection cannot see. The generated ActorTemplate CR is deleted by owner-reference garbage collection after the AgentHarness finalizer is removed. WorkerPools are externally owned and are never deleted here.

func (*Lifecycle) CleanupSandboxAgentTemplate 

func (p *Lifecycle) CleanupSandboxAgentTemplate(ctx context.Context, sa *v1alpha2.SandboxAgent) (bool, error)

CleanupSandboxAgentTemplate removes external Substrate actors tied to a generated SandboxAgent ActorTemplate.

func (*Lifecycle) EnsureGeneratedTemplate 

func (p *Lifecycle) EnsureGeneratedTemplate(ctx context.Context, ah *v1alpha2.AgentHarness) (LifecycleState, error)

EnsureGeneratedTemplate creates or updates the generated ActorTemplate and reports whether it is Ready.

type LifecycleDefaults 

type LifecycleDefaults struct {
	PauseImage           string
	DefaultWorkloadImage string
	DefaultWorkerPool    types.NamespacedName
	// ImageRegistry and ImageRepository are the runtime registry/repository used
	// to compose digest-pinned acp-sandbox workload image refs (from
	// --image-registry/--image-repository). ImageRepository is the agent app
	// repository (e.g. "kagent-dev/kagent/app"); see acpSandboxImageConfig.
	ImageRegistry   string
	ImageRepository string
}

LifecycleDefaults are cluster-wide defaults for generated ActorTemplate lifecycle.

type LifecycleState 

type LifecycleState struct {
	ActorTemplateReady bool
}

LifecycleState describes the generated Substrate lifecycle for an AgentHarness.

type SandboxAgentActorBackend 

type SandboxAgentActorBackend struct {
	// contains filtered or unexported fields
}

SandboxAgentActorBackend manages ate-api actors for SandboxAgent workloads.

func NewSandboxAgentActorBackend 

func NewSandboxAgentActorBackend(client *Client, kube client.Client, atenetRouterURL string) *SandboxAgentActorBackend

NewSandboxAgentActorBackend returns a backend that ensures SandboxAgent actors on ate-api. kube is used to resolve the agent's current (config-hashed) ActorTemplate.

func (*SandboxAgentActorBackend) DeleteAllSandboxAgentActors 

func (b *SandboxAgentActorBackend) DeleteAllSandboxAgentActors(ctx context.Context, sa *v1alpha2.SandboxAgent) (bool, error)

DeleteAllSandboxAgentActors deletes legacy per-agent actors and all session actors for a SandboxAgent.

func (*SandboxAgentActorBackend) DeleteSandboxAgentActor 

func (b *SandboxAgentActorBackend) DeleteSandboxAgentActor(ctx context.Context, atespace, actorID string) (bool, error)

DeleteSandboxAgentActor deletes a substrate actor by id.

func (*SandboxAgentActorBackend) DeleteSandboxAgentSessionActor 

func (b *SandboxAgentActorBackend) DeleteSandboxAgentSessionActor(ctx context.Context, sa *v1alpha2.SandboxAgent, sessionID string) (bool, error)

DeleteSandboxAgentSessionActor deletes the actor for a single chat session. One session ⇔ one actor with a session-derived id, so a single deterministic delete covers the session's whole life regardless of how many shape rollouts it survived.

func (*SandboxAgentActorBackend) EnsureSessionActor 

func (b *SandboxAgentActorBackend) EnsureSessionActor(ctx context.Context, sa *v1alpha2.SandboxAgent, sessionID string) (sandboxbackend.EnsureResult, error)

EnsureSessionActor creates (or resumes) the per-session actor for a SandboxAgent chat and waits for it to be reachable. One session ⇔ one actor, for the session's entire life: the actor id is derived from the session alone, and an existing actor is always resumed — substrate rebuilds its workload spec from the actor's BIRTH template (the actor record stores the template name), so a session stays pinned to the shape it was created under across shape rollouts. Only when no actor exists yet is one created from the agent's current (newest Ready) ActorTemplate — during a shape rollout, new sessions keep landing on the previous Ready golden until the new one is Ready.

If the WorkerPool has no free worker, CreateActor/ResumeActor surface ErrNoFreeWorkers and this returns it immediately (no buffering). On a single-replica pool the lone worker may be busy building a new golden, so a shape change can briefly make chat return "no free workers"; on a multi-replica pool the spare workers keep serving existing actors, so a rollout does not hit that error. Scaling the WorkerPool is the remedy for capacity pressure, not in-process retries.

func (*SandboxAgentActorBackend) SuspendSessionActor 

func (b *SandboxAgentActorBackend) SuspendSessionActor(ctx context.Context, sa *v1alpha2.SandboxAgent, sessionID string) error

SuspendSessionActor checkpoints and frees the worker for a chat session actor.

type SessionActorState 

type SessionActorState string

SessionActorState is the coarse lifecycle state of a chat session's actor as surfaced to the UI. 

const (
	// SessionActorStateRunning means the actor is running or waking up.
	SessionActorStateRunning SessionActorState = "running"
	// SessionActorStateSuspended means the actor is checkpointed/freed and will
	// resume on the next connect.
	SessionActorStateSuspended SessionActorState = "suspended"
	// SessionActorStateMissing means no actor exists yet for the session.
	SessionActorStateMissing SessionActorState = "missing"
)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.