Initial commit of Agent Substrate

This is the initial release of the Agent Substrate.

Agent substrate is a system built on top of Kubernetes which manages agent-like
workloads to achieve higher scale and efficiency than Kubernetes alone can
offer, with lower latency.  It builds on top of Kubernetes features like
Pods and Pod autoscaling, but takes the Kubernetes control-plane out of the
critical path to achieve lower latency.

It can run on any Kubernetes cluster and does not inhibit “regular” use of
Kubernetes in any way. Kubernetes provides the infrastructure provisioning and
management for all types of workloads, while Agent Substrate provides
agent-specific scheduling and control.

At its core, Agent Substrate maps a larger set of “actors” (applications such
as agents) onto a smaller set of ready “workers” (Kubernetes Pods), relying on
the fact that agent-like applications tend to be idle most of the time to
achieve heavy multiplexing.  It provides functionality to manage an actor’s
lifecycle (e.g. create/destroy, suspend/resume), to assign actors to workers in real
time, and to route incoming traffic to them.

Agent Substrate is intended to be a low-opinion system.  The workloads it
manages don't have to be literal AI agents, but those are the best example of
the kind of applications it is designed for.  It is not an SDK for building
agents, but rather a system for running them at scale.

Agent Substrate is currently in VERY early development.  It is not ready for
production use, and the APIs are almost guaranteed to change.  We are not
making any guarantees about backward compatibility at this stage, and
everything in this project may be changed.

Co-authored-by: Alex Bulankou <alexbu@google.com>
Co-authored-by: Benjamin Elder <bentheelder@google.com>
Co-authored-by: Bowei Du <bowei@google.com>
Co-authored-by: Dmitry Berkovich <dberkov@google.com>
Co-authored-by: Fabricio Voznika <fvoznika@google.com>
Co-authored-by: Francisco Cabrera <fclieutier@google.com>
Co-authored-by: Haven Xia <haoyuxia@google.com>
Co-authored-by: Julian Gutierrez Oschmann <juliangut@google.com>
Co-authored-by: Kevin Steuer <ksteuer@google.com>
Co-authored-by: Max Smythe <smythe@google.com>
Co-authored-by: Maya Wang <mymaya@google.com>
Co-authored-by: Michael Taufen <mtaufen@google.com>
Co-authored-by: Shruti Nair <shrutinair@google.com>
Co-authored-by: Taahir Ahmed <taahm@google.com>
Co-authored-by: Tim Hockin <thockin@google.com>
Co-authored-by: Zoe Zhao <zoezhao@google.com>

internal/memorypullcache/memorypullcache.go

@@ -0,0 +1,217 @@
+//  Copyright 2026 Google LLC
+//
+//  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 memorypullcache
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"log/slog"
+	"net"
+	"runtime"
+	"strings"
+
+	"cloud.google.com/go/storage"
+	"github.com/google/go-containerregistry/pkg/authn"
+	"github.com/google/go-containerregistry/pkg/name"
+	v1 "github.com/google/go-containerregistry/pkg/v1"
+	"github.com/google/go-containerregistry/pkg/v1/mutate"
+	"github.com/google/go-containerregistry/pkg/v1/remote"
+	"k8s.io/utils/lru"
+)
+
+type MemoryPullCache struct {
+	gcsClient *storage.Client
+
+	gcpAuthenticator authn.Authenticator
+
+	localhostRegistryReplacement string
+
+	// Map from hexadecimal sha256 hash of image to byte contents of composed
+	// tarball
+	cache *lru.Cache
+}
+
+func NewMemoryPullCache(ctx context.Context, gcpAuthenticator authn.Authenticator, localhostRegistryReplacement string) (*MemoryPullCache, error) {
+	c := &MemoryPullCache{
+		// TODO: Need a smarter cache with bounds on total consumed size, not
+		// just number of entries.  Potentially also try to share the cache
+		// across ateoms on the same machine.
+		//
+		// It would have to be a directory with files named after the sha256
+		// hash.  The benefit would be that a read might be found in the
+		// filesystem cache, or perhaps the folder could be on SSD.
+		//
+		// From the perspective of stable operation, without hidden kernel
+		// caches that could fill up or have weird behavior, it might be better
+		// to just have two levels.  Store some images in ateom memory, and the
+		// rest are kept in a shared GCS cache.
+		cache:                        lru.New(256),
+		localhostRegistryReplacement: localhostRegistryReplacement,
+	}
+
+	c.gcpAuthenticator = gcpAuthenticator
+
+	return c, nil
+}
+
+func (c *MemoryPullCache) Fetch(ctx context.Context, ref string) (io.ReadCloser, error) {
+	// when running in kind we need to rewrite the registry endpoint similar to the
+	// containerd mirror config used in https://kind.sigs.k8s.io/docs/user/local-registry/
+	// for now we have simple opt-in support to rewrite local registries
+	rewritten := false
+	if c.localhostRegistryReplacement != "" {
+		newRef := c.rewriteLocalRegistry(ref)
+		if newRef != ref {
+			ref = newRef
+			rewritten = true
+		}
+	}
+	var nameOpts []name.Option
+	// match docker behavior, permit http image pulls for local registries
+	// this avoids needing to distribute TLS certs all around for local development
+	if rewritten || isLocalRegistry(ref) {
+		nameOpts = append(nameOpts, name.Insecure)
+	}
+
+	parsedRef, err := name.ParseReference(ref, nameOpts...)
+	if err != nil {
+		return nil, fmt.Errorf("while parsing reference: %w", err)
+	}
+
+	// If the image ref included a digest, check for a hit in the pull cache.
+	requestedDigest, digestWasIncluded := parsedRef.(name.Digest)
+	if digestWasIncluded {
+		slog.InfoContext(
+			ctx,
+			"Ref includes digest, checking for cache hit",
+			slog.String("ref", ref),
+			slog.String("digest", requestedDigest.DigestStr()),
+		)
+
+		if vAny, ok := c.cache.Get(requestedDigest.DigestStr()); ok {
+			slog.InfoContext(
+				ctx,
+				"Cache hit",
+				slog.String("ref", ref),
+				slog.String("digest", requestedDigest.DigestStr()),
+			)
+			return io.NopCloser(bytes.NewReader(vAny.([]byte))), nil
+		}
+	}
+
+	slog.InfoContext(
+		ctx,
+		"Cache miss",
+		slog.String("ref", ref),
+	)
+
+	// If we didn't have a cache hit, we are on the slow path of pulling the
+	// image from the registry.  This is a chatty process, with multiple round
+	// trips to the registry.
+
+	var remoteOptions []remote.Option
+	remoteOptions = append(remoteOptions, remote.WithPlatform(v1.Platform{
+		Architecture: runtime.GOARCH,
+		OS:           "linux",
+	}))
+
+	registry := parsedRef.Context().Registry.RegistryStr()
+	if registry == "gcr.io" || strings.HasSuffix(registry, ".gcr.io") || registry == "pkg.dev" || strings.HasSuffix(registry, ".pkg.dev") {
+		if c.gcpAuthenticator != nil {
+			remoteOptions = append(remoteOptions, remote.WithAuth(c.gcpAuthenticator))
+		}
+	}
+
+	img, err := remote.Image(parsedRef, remoteOptions...)
+	if err != nil {
+		return nil, fmt.Errorf("in remote.Image: %w", err)
+	}
+
+	size, err := img.Size()
+	if err != nil {
+		return nil, fmt.Errorf("in img.Size(): %w", err)
+	}
+	if size > 100*1024*1024 {
+		slog.InfoContext(ctx,
+			"Image is too large to cache",
+			slog.String("ref", ref),
+			slog.Int64("size", size),
+		)
+		return mutate.Extract(img), err
+	}
+
+	tarData := mutate.Extract(img)
+	defer tarData.Close()
+
+	memData, err := io.ReadAll(tarData)
+	if err != nil {
+		return nil, fmt.Errorf("while reading image: %w", err)
+	}
+
+	if digestWasIncluded {
+		// If the user requested multi-arch image, the digest they request will
+		// not be the same as the digest of the image we actually downloaded
+		// from the registry.  We need to place the cache entry under the digest
+		// they requested.
+		c.cache.Add(requestedDigest.DigestStr(), memData)
+		slog.InfoContext(
+			ctx,
+			"Populated image cache",
+			slog.String("ref", ref),
+			slog.String("digest", requestedDigest.DigestStr()),
+		)
+	}
+
+	return io.NopCloser(bytes.NewReader(memData)), nil
+}
+
+func registryHost(ref string) string {
+	parts := strings.SplitN(ref, "/", 2)
+	reg, err := name.NewRegistry(parts[0], name.Insecure)
+	if err != nil {
+		return ""
+	}
+	hostPart := reg.Name()
+	if h, _, err := net.SplitHostPort(hostPart); err == nil {
+		return h
+	}
+	return hostPart
+}
+
+func isLocalhostOrLoopback(host string) bool {
+	if host == "localhost" {
+		return true
+	}
+	if ip := net.ParseIP(host); ip != nil && ip.IsLoopback() {
+		return true
+	}
+	return false
+}
+
+func isLocalRegistry(ref string) bool {
+	// by default docker permits localhost and 127.0.0.0/8
+	// we also permit IPv6 loopback here
+	return isLocalhostOrLoopback(registryHost(ref))
+}
+
+func (c *MemoryPullCache) rewriteLocalRegistry(ref string) string {
+	if isLocalRegistry(ref) {
+		parts := strings.SplitN(ref, "/", 2)
+		return c.localhostRegistryReplacement + "/" + parts[1]
+	}
+	return ref
+}