Replace Redis Streams queue and controller

.dockerignore

@@ -12,4 +12,3 @@ bin/*
 ax
 ax-server
 ax-task-runner
-ax-controller

.github/workflows/go.yml

@@ -42,7 +42,6 @@ jobs:
     - name: Build local binaries
       run: |
         go build -v ./cmd/ax
-        go build -v ./cmd/ax-controller
         go build -v ./cmd/ax-server
         go build -v ./cmd/ax-task-runner
 

.gitignore

@@ -6,6 +6,5 @@ bin/
 /ax
 /ax-server
 /ax-task-runner
-/ax-controller
 
 cmd/ax-task-runner/__pycache__/antigravity_bootstrap.cpython-313.pyc

.ko.yaml

@@ -16,8 +16,8 @@ defaultBaseImage: cgr.dev/chainguard/static:latest
 baseImageOverrides:
   github.com/google/ax/cmd/ax-task-runner: alpine/git:latest
 builds:
-  - id: ax-controller
-    main: ./cmd/ax-controller
+  - id: ax-server
+    main: ./cmd/ax-server
     env:
       - CGO_ENABLED=0
     flags:

CONTRIBUTING.md

@@ -33,7 +33,7 @@ All submissions, including submissions by project members, require review. We us
 
 ### Building Binaries
 
-Build all local binaries (`bin/ax`, `bin/ax-controller`, `bin/ax-server`):
+Build all local binaries (`bin/ax`, `bin/ax-server`):
 
 ```bash
 make build
@@ -81,7 +81,7 @@ make push-task-runner TASK_RUNNER_REPO=gcr.io/<your-project>/ax-task-runner
 
 ### Deploying to Kubernetes
 
-Deploy Redis, controller, and server components to your cluster in the `ax-system` namespace:
+Deploy Redis and server components to your cluster in the `ax-system` namespace:
 
 ```bash
 make deploy AX_IMAGE_REPO=gcr.io/<your-project>/ax-images

DESIGN.md

@@ -2,7 +2,7 @@
 
 ## Architecture
 
-Storing millions of short-lived tasks as Kubernetes CRDs pushes etcd past its comfort zone (single-digit GB storage limits, write-rate bottlenecks, control plane degradation). AX keeps its state in Redis and uses Redis Streams as the work queue between the API server and a horizontally scaled pool of controllers.
+Storing millions of short-lived tasks as Kubernetes CRDs pushes etcd past its comfort zone (single-digit GB storage limits, write-rate bottlenecks, control plane degradation). AX stores its state in Redis and reconciles directly with Agent Substrate under fine-grained distributed locks.
 
 ```
                       ax apply -f task.yaml
@@ -11,36 +11,22 @@ Storing millions of short-lived tasks as Kubernetes CRDs pushes etcd past its co
                             ax-server
                       (gRPC API + /healthz)
                                 │
-                     store & publish event
-                                │
-                                ▼
-                              Redis
-               (Task Hashes + Event Streams + PubSub)
-                                │
-                      XREADGROUP (Streams)
-                                │
-                                ▼
-                          ax-controller
-                   (Horizontally Scaled Workers)
-                                │
-                        gRPC (Control API)
-                                │
-                                ▼
-                         Agent Substrate
-                ┌───────────────────────────────┐
-                │ • Atespace Provisioning       │
-                │ • Actor Creation & Activation │
-                │ • Worker Assignment           │
-                └───────────────────────────────┘
+         ┌──────────────────────┴──────────────────────┐
+         ▼                                             ▼
+       Redis                                    Agent Substrate
+ (Resource Store,                       ┌───────────────────────────────┐
+  Locks, PubSub)                        │ • Atespace Provisioning       │
+                                        │ • Actor Creation & Activation │
+                                        │ • Worker Assignment           │
+                                        └───────────────────────────────┘
 ```
 
 ## Components
 
 | Binary | Role |
 |---|---|
 | `ax` | Developer CLI. Applies manifests, inspects and watches resources, tunnels to the cluster. |
-| `ax-server` | Stateless gRPC API on port 8080. Validates manifests, persists to Redis, publishes events. |
-| `ax-controller` | Reconciliation workers. Consume the Redis stream, provision atespaces and actors on Agent Substrate, and drive tasks toward desired state. Scale by adding replicas. |
+| `ax-server` | Direct-execution gRPC API on port 8080. Validates manifests, manages distributed locks, reconciles directly with Agent Substrate, and persists state to Redis. |
 | `ax-task-runner` | Entrypoint inside every task container. Bootstraps the workspace, serves metadata, and runs the agent command. A thin wrapper over the `runner` package, which custom images can embed directly. |
 
 ## API reference

Makefile

@@ -19,22 +19,21 @@ AX_IMAGE_REPO ?= gcr.io/ax-substrate/ate-images
 TASK_RUNNER_REPO ?= $(AX_IMAGE_REPO)/ax-task-runner
 CONTAINER_CLI ?= $(shell which podman 2>/dev/null || which docker 2>/dev/null)
 
-.PHONY: all build build-binaries build-task-runner install push push-task-runner deploy deploy-controller deploy-server deploy-redis apply-example test clean
+.PHONY: all build build-binaries build-task-runner install push push-task-runner deploy deploy-server deploy-redis apply-example test clean
 
 all: build
 
 ## --------------------------------------
 ## Build Targets
 ## --------------------------------------
 
-# Build all local binaries (ax CLI, controller, server)
+# Build all local binaries (ax CLI, server)
 build: build-binaries
 
 build-binaries:
-	@echo "==> Building local binaries (ax, ax-controller, ax-server)..."
+	@echo "==> Building local binaries (ax, ax-server)..."
 	@mkdir -p bin
 	go build -trimpath -ldflags="-s -w" -o bin/ax ./cmd/ax
-	go build -trimpath -ldflags="-s -w" -o bin/ax-controller ./cmd/ax-controller
 	go build -trimpath -ldflags="-s -w" -o bin/ax-server ./cmd/ax-server
 
 # Install the ax CLI into $(go env GOPATH)/bin
@@ -64,17 +63,13 @@ push: push-task-runner
 ## Deployment Targets
 ## --------------------------------------
 
-# Deploy all AX components to Kubernetes (Redis, ax-controller, ax-server)
-deploy: deploy-redis deploy-controller deploy-server
+# Deploy all AX components to Kubernetes (Redis, ax-server)
+deploy: deploy-redis deploy-server
 
 deploy-redis:
 	@echo "==> Deploying Redis to ax-system namespace..."
 	kubectl apply -f deploy/redis.yaml
 
-deploy-controller:
-	@echo "==> Building and deploying ax-controller using ko..."
-	KO_DOCKER_REPO=$(AX_IMAGE_REPO) ko apply -f deploy/ax-controller.yaml
-
 deploy-server:
 	@echo "==> Building and deploying ax-server using ko..."
 	KO_DOCKER_REPO=$(AX_IMAGE_REPO) ko apply -f deploy/ax-server.yaml

README.md

@@ -4,9 +4,9 @@
 </h1>
 
 > [!WARNING]
-> We are still actively refining our core concepts, protocols,
-> and specifications. We will likely to introduce major breaking
-> changes prior to a stable release.
+> AX and several of its features are in heavy development. We are actively
+> refining our core concepts, protocols, and specifications, and will likely
+> introduce major breaking changes prior to a stable release.
 
 **Declare an agentic task with workspaces and model specifications. AX sandboxes it, wires up its workspace, and helps running it at scale.**
 

cmd/ax-server/main.go

@@ -24,21 +24,40 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/google/ax/internal/controller"
+	"github.com/google/ax/internal/lock"
 	"github.com/google/ax/internal/server"
 	"github.com/google/ax/internal/store/redis"
+	"github.com/google/ax/internal/substrate"
 	goredis "github.com/redis/go-redis/v9"
 )
 
 func main() {
 	var (
-		listenAddr    string
-		redisAddr     string
-		redisPassword string
+		listenAddr              string
+		redisAddr               string
+		redisPassword           string
+		substrateEndpoint       string
+		substrateAuthority      string
+		substrateTokenFile      string
+		substrateCAFile         string
+		substrateInsecureTLS    bool
+		substratePlaintext      bool
+		defaultTemplate         string
+		defaultTemplateAtespace string
 	)
 
 	flag.StringVar(&listenAddr, "addr", ":8080", "HTTP listen address")
 	flag.StringVar(&redisAddr, "redis-addr", "localhost:6379", "Redis server address")
 	flag.StringVar(&redisPassword, "redis-password", "", "Redis password")
+	flag.StringVar(&substrateEndpoint, "substrate-endpoint", "api.ate-system.svc.cluster.local:443", "Agent Substrate Control API endpoint")
+	flag.StringVar(&substrateAuthority, "substrate-authority", "api.ate-system.svc", "Authority / TLS ServerName for Substrate endpoint")
+	flag.StringVar(&substrateTokenFile, "substrate-token-file", "", "Path to bearer token file for Substrate auth")
+	flag.StringVar(&substrateCAFile, "substrate-ca-file", "", "Path to CA PEM file for Substrate TLS")
+	flag.BoolVar(&substrateInsecureTLS, "substrate-insecure-tls", false, "Skip Substrate TLS verification")
+	flag.BoolVar(&substratePlaintext, "substrate-plaintext", false, "Use insecure plaintext gRPC connection to Substrate")
+	flag.StringVar(&defaultTemplate, "template", "default-template", "Default Substrate ActorTemplate name")
+	flag.StringVar(&defaultTemplateAtespace, "template-atespace", "ax-system", "Default Substrate ActorTemplate atespace")
 	flag.Parse()
 
 	if envAddr := os.Getenv("ADDR"); envAddr != "" {
@@ -54,7 +73,12 @@ func main() {
 	logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
 	slog.SetDefault(logger)
 
-	slog.Info("starting ax-server", "listenAddr", listenAddr, "redisAddr", redisAddr)
+	slog.Info("starting ax-server",
+		"listenAddr", listenAddr,
+		"redisAddr", redisAddr,
+		"substrateEndpoint", substrateEndpoint,
+		"template", defaultTemplate,
+	)
 
 	rClient := goredis.NewClient(&goredis.Options{
 		Addr:     redisAddr,
@@ -63,7 +87,28 @@ func main() {
 	defer rClient.Close()
 
 	rStore := redis.NewStore(rClient, redis.Options{})
-	srv := server.NewServer(rStore)
+	rLocker := lock.NewRedisLocker(rClient, lock.RedisLockerOptions{})
+
+	var reconciler server.Reconciler
+	subClient, err := substrate.NewClientWithOptions(substrate.ClientOptions{
+		Target:      substrateEndpoint,
+		Authority:   substrateAuthority,
+		TokenFile:   substrateTokenFile,
+		CAFile:      substrateCAFile,
+		InsecureTLS: substrateInsecureTLS,
+		Plaintext:   substratePlaintext,
+	})
+	if err != nil {
+		slog.Warn("could not initialize substrate client; running without substrate reconciliation", "error", err)
+	} else {
+		defer subClient.Close()
+		reconciler = controller.NewTaskReconciler(subClient, defaultTemplate, defaultTemplateAtespace)
+	}
+
+	srv := server.NewServer(rStore, server.Options{
+		Locker:     rLocker,
+		Reconciler: reconciler,
+	})
 
 	httpServer := &http.Server{
 		Addr:    listenAddr,

cmd/ax-task-runner/main.go

@@ -16,7 +16,7 @@
 // loads the Task and Workspace specs and hands them to the runner package,
 // which does everything else.
 //
-// The controller delivers the Task as YAML in AX_TASK_YAML and the bound
+// AX delivers the Task as YAML in AX_TASK_YAML and the bound
 // Workspaces as a multi-document YAML stream in AX_WORKSPACES_YAML. For local
 // runs the specs can be read from files instead with --task-file and one or
 // more --workspace-file flags.