Merge commit 'fe854f2' into support-actortemplate-secret-env

Author: EItanya

.github/workflows/pr-workflow.yaml

@@ -15,6 +15,8 @@
 name: pr-workflow
 on:
   pull_request:
+  push:
+    branches: [main]
 jobs:
   run-tests:
     runs-on: ubuntu-latest
@@ -25,7 +27,6 @@ jobs:
       uses: actions/setup-go@v5
       with:
         go-version: 1.26
-    - run: go vet -v ./...
     - run: go test -v ./...
     - name: verify
       run: hack/verify-all.sh

.golangci.yaml

@@ -26,6 +26,10 @@ linters:
   # the codebase is clean for the basics.
   default: standard
 
+  enable:
+    # Spelling errors in comments, strings, and identifiers.
+    - misspell
+
   settings:
     errcheck:
       # Functions where ignoring the return is idiomatic in this codebase.
@@ -48,6 +52,9 @@ linters:
         - (k8s.io/client-go/tools/cache.SharedIndexInformer).AddEventHandler
         - (k8s.io/client-go/tools/cache.SharedIndexInformer).AddIndexers
 
+    misspell:
+      mode: default
+
   exclusions:
     # Built-in presets that suppress the most common idiomatic false
     # positives across the standard linters.

AGENTS.md

@@ -9,6 +9,34 @@ Agent Substrate relies on the fact that agent-like applications tend to be idle
 For development, it's recommended to read the `README.md` and `CONTRIBUTING.md` in the root folder.
 See `hack/install-ate.sh` and `tools/setup-gcp` for provisioning and deploying clusters and GCP resources.
 
+## Repository Layout
+
+```
+cmd/          # One subdirectory per binary (ateapi, atelet, atenet, …)
+internal/     # Shared packages, internal to this module only
+pkg/          # Shared packages intended for external import
+docs/         # Design docs and developer guides
+hack/         # Dev/CI scripts and code generators
+manifests/    # Kubernetes YAML for deploying Agent Substrate
+demos/        # Self-contained example applications
+benchmarking/ # Load-testing tools and workloads
+tools/        # Standalone Go tools (go run ./tools/<name>) for Dev/CI
+```
+
+**Where to put new Go code — quick rules:**
+
+| Situation | Location |
+|---|---|
+| Only used by one binary | `cmd/<binary>/internal/<pkg>` |
+| Shared across binaries, not for external import | `internal/<pkg>` |
+| Public API for external consumers | `pkg/<pkg>` |
+| Public proto (control-plane gRPC API) | `pkg/proto/<name>` |
+| Internal proto (atelet / ateom) | `internal/proto/<name>` |
+| Dev/CI scripts | `hack/` |
+| Standalone Go dev/CI tools | `tools/<name>` with its own `go.mod` |
+
+See `docs/dev/code-layout.md` for the full rationale and per-directory details.
+
 ## Build and Test Commands
 
 Agent Substrate uses a `Makefile` for its build and test tasks.

LICENSES/github.com/google/go-cmp/cmp/LICENSE

@@ -77,7 +77,6 @@ lint:
 
 .PHONY: verify
 verify: test
-	$(GO) vet ./...
 	bash hack/verify-all.sh
 
 .PHONY: clean

Makefile

@@ -57,13 +57,6 @@ spec:
   - name: sleep
     image: busybox@sha256:1487d0af5f52b4ba31c7e465126ee2123fe3f2305d638e7827681e7cf6c83d5e
     command: ["/bin/sleep", "1000000"]
-  sessionDiscovery:
-    mapping:
-      pathPrefix: "/v1/sleep"
-    protocol:
-      strategy: CEL
-      actorIdentity:
-        cel: "request.host.split('.')[0]"
   workerPoolRef:
     namespace: benchmark-workloads
     name: benchmark-ateom
@@ -90,13 +83,6 @@ spec:
   - name: usermem
     image: containerstack/alpine-stress:latest
     command: ["stress", "--vm", "1", "--vm-bytes", "1G", "--vm-keep", "--vm-hang", "0"]
-  sessionDiscovery:
-    mapping:
-      pathPrefix: "/v1/usermem"
-    protocol:
-      strategy: CEL
-      actorIdentity:
-        cel: "request.host.split('.')[0]"
   workerPoolRef:
     namespace: benchmark-workloads
     name: benchmark-ateom
@@ -123,13 +109,6 @@ spec:
   - name: kernelmem
     image: containerstack/alpine-stress:latest
     command: ["sh", "-c", "find /; for i in $(seq 50000); do sleep 10000 & done; sleep 10000"]
-  sessionDiscovery:
-    mapping:
-      pathPrefix: "/v1/kernelmem"
-    protocol:
-      strategy: CEL
-      actorIdentity:
-        cel: "request.host.split('.')[0]"
   workerPoolRef:
     namespace: benchmark-workloads
     name: benchmark-ateom

benchmarking/workloads/manifests/full_workloads.yaml.tmpl

@@ -48,13 +48,6 @@ spec:
   - name: sleep
     image: busybox@sha256:1487d0af5f52b4ba31c7e465126ee2123fe3f2305d638e7827681e7cf6c83d5e
     command: ["/bin/sleep", "1000000"]
-  sessionDiscovery:
-    mapping:
-      pathPrefix: "/v1/sleep"
-    protocol:
-      strategy: CEL
-      actorIdentity:
-        cel: "request.host.split('.')[0]"
   workerPoolRef:
     namespace: benchmark-workloads
     name: benchmark-ateom

benchmarking/workloads/manifests/workloads.yaml.tmpl

@@ -68,6 +68,9 @@ func (s *WorkerPoolSyncer) Start(ctx context.Context) {
 				return
 			}
 			slog.InfoContext(ctx, "Syncer: removing worker from store", slog.String("worker", pod.Namespace+"/"+pod.Name))
+			if err := s.releaseActorOnDeadWorker(ctx, pod.Namespace, pod.Labels[workerPodLabel], pod.Name); err != nil {
+				slog.ErrorContext(ctx, "Failed to release actor bound to deleted worker", slog.Any("err", err))
+			}
 			err := s.persistence.DeleteWorker(ctx, pod.Namespace, pod.Labels[workerPodLabel], pod.Name)
 			if err != nil {
 				slog.ErrorContext(ctx, "Failed to delete worker from store during delete event", slog.Any("err", err))
@@ -97,6 +100,9 @@ func (s *WorkerPoolSyncer) syncWorkerToStore(ctx context.Context, pod *corev1.Po
 
 	if pod.DeletionTimestamp != nil {
 		slog.InfoContext(ctx, "Syncer: removing worker from store (pod deleting)", slog.String("worker", pod.Namespace+"/"+pod.Name))
+		if err := s.releaseActorOnDeadWorker(ctx, pod.Namespace, pod.Labels[workerPodLabel], pod.Name); err != nil {
+			slog.ErrorContext(ctx, "Failed to release actor bound to soft-deleting worker", slog.Any("err", err))
+		}
 		err := s.persistence.DeleteWorker(ctx, pod.Namespace, pod.Labels[workerPodLabel], pod.Name)
 		if err != nil {
 			slog.ErrorContext(ctx, "Failed to delete worker from store during update event (deleting)", slog.Any("err", err))
@@ -136,3 +142,50 @@ func (s *WorkerPoolSyncer) syncWorkerToStore(ctx context.Context, pod *corev1.Po
 func isWorkerEligible(pod *corev1.Pod) bool {
 	return pod.Status.PodIP != ""
 }
+
+// releaseActorOnDeadWorker resets the actor bound to a vanishing worker
+// pod back to STATUS_SUSPENDED so the next request reassigns it.
+//
+// UpdateActor uses optimistic version checking. A concurrent SuspendActor
+// or ResumeActor wins; we drop this attempt silently.
+//
+// Best-effort only. The caller always proceeds to DeleteWorker after this
+// returns, so any non-contention failure leaves the actor stranded
+// (STATUS_RUNNING, pointer at a pod that no longer exists). Recovery
+// then needs a manual SuspendActor.
+//
+// The long-term fix is a finalizer-based controller that holds the pod
+// in Terminating state until the actor is gracefully suspended. Tracked
+// in https://github.com/agent-substrate/substrate/issues/23.
+func (s *WorkerPoolSyncer) releaseActorOnDeadWorker(ctx context.Context, namespace, pool, podName string) error {
+	worker, err := s.persistence.GetWorker(ctx, namespace, pool, podName)
+	if err != nil {
+		if errors.Is(err, store.ErrNotFound) {
+			return nil
+		}
+		return err
+	}
+	if worker.GetActorId() == "" {
+		return nil
+	}
+	actor, err := s.persistence.GetActor(ctx, worker.GetActorId())
+	if err != nil {
+		if errors.Is(err, store.ErrNotFound) {
+			return nil
+		}
+		return err
+	}
+	// Skip if a concurrent SuspendActor already cleared the pointer.
+	if actor.GetAteomPodNamespace() != namespace || actor.GetAteomPodName() != podName {
+		return nil
+	}
+	actor.Status = ateapipb.Actor_STATUS_SUSPENDED
+	actor.AteomPodNamespace = ""
+	actor.AteomPodName = ""
+	actor.AteomPodIp = ""
+	actor.InProgressSnapshot = ""
+	if err := s.persistence.UpdateActor(ctx, actor, actor.GetVersion()); err != nil && !errors.Is(err, store.ErrPersistenceRetry) {
+		return err
+	}
+	return nil
+}

cmd/ateapi/internal/controlapi/syncer.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest"
+	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -154,3 +155,62 @@ func TestSyncer_Lifecycle(t *testing.T) {
 		t.Fatalf("Worker still found in Redis after deletion: %v", err)
 	}
 }
+
+func TestSyncer_DeleteBoundWorker_ClearsActor(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	persistence, fakeK8s, cleanup := setupSyncerTest(t, ctx)
+	defer cleanup()
+
+	ns, pool, pod, ip := "ns-orphan", "pool1", "worker-orphan", "10.0.0.1"
+	if _, err := fakeK8s.CoreV1().Pods(ns).Create(ctx, &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{Name: pod, Namespace: ns,
+			Labels: map[string]string{workerPodLabel: pool}},
+		Status: corev1.PodStatus{Phase: corev1.PodRunning, PodIP: ip,
+			PodIPs: []corev1.PodIP{{IP: ip}}},
+	}, metav1.CreateOptions{}); err != nil {
+		t.Fatalf("create pod: %v", err)
+	}
+	if err := wait.PollUntilContextTimeout(ctx, 50*time.Millisecond, 2*time.Second, true, func(c context.Context) (bool, error) {
+		_, gerr := persistence.GetWorker(c, ns, pool, pod)
+		return gerr == nil, nil
+	}); err != nil {
+		t.Fatalf("worker row not materialised: %v", err)
+	}
+	actorID := "actor-orphan"
+	if err := persistence.CreateActor(ctx, &ateapipb.Actor{
+		ActorId: actorID, ActorTemplateNamespace: ns, ActorTemplateName: "tmpl",
+		Status:            ateapipb.Actor_STATUS_RUNNING,
+		AteomPodNamespace: ns, AteomPodName: pod, AteomPodIp: ip,
+		LastSnapshot: "gs://snapshots/last", InProgressSnapshot: "gs://snapshots/partial",
+	}); err != nil {
+		t.Fatalf("create actor: %v", err)
+	}
+	w, _ := persistence.GetWorker(ctx, ns, pool, pod)
+	w.ActorId, w.ActorNamespace, w.ActorTemplate = actorID, ns, "tmpl"
+	if err := persistence.UpdateWorker(ctx, w, w.Version); err != nil {
+		t.Fatalf("update worker: %v", err)
+	}
+
+	if err := fakeK8s.CoreV1().Pods(ns).Delete(ctx, pod, metav1.DeleteOptions{}); err != nil {
+		t.Fatalf("delete pod: %v", err)
+	}
+	var got *ateapipb.Actor
+	if err := wait.PollUntilContextTimeout(ctx, 50*time.Millisecond, 2*time.Second, true, func(c context.Context) (bool, error) {
+		a, gerr := persistence.GetActor(c, actorID)
+		if gerr != nil {
+			return false, gerr
+		}
+		got = a
+		return a.GetStatus() == ateapipb.Actor_STATUS_SUSPENDED, nil
+	}); err != nil {
+		t.Fatalf("actor not reset to SUSPENDED: %v", err)
+	}
+	if got.AteomPodName != "" || got.AteomPodNamespace != "" || got.AteomPodIp != "" || got.InProgressSnapshot != "" {
+		t.Errorf("bind fields not cleared: %+v", got)
+	}
+	if got.LastSnapshot == "" {
+		t.Errorf("LastSnapshot must be preserved")
+	}
+}