From 6e4df6d2ff9f6440386aa93720e1bf4954967b59 Mon Sep 17 00:00:00 2001
From: kami <kami@kvmx.ru>
Date: Sun, 26 Jul 2026 20:32:45 +0400
Subject: [PATCH] add federated worker registration and heartbeats

---
 cmd/orchestra/main.go             | 39 +++++++++++++++++
 internal/federation/federation.go | 70 +++++++++++++++++++++++++++++++
 progress.md                       |  2 +
 3 files changed, 111 insertions(+)
 create mode 100644 internal/federation/federation.go

diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go
index a3a7736..2c8b1a7 100644
--- a/cmd/orchestra/main.go
+++ b/cmd/orchestra/main.go
@@ -10,6 +10,7 @@ import (
 	"orchestra/internal/authz"
 	"orchestra/internal/delivery"
 	"orchestra/internal/domain"
+	"orchestra/internal/federation"
 	"orchestra/internal/herdr"
 	"orchestra/internal/operations"
 	"orchestra/internal/orchestrator"
@@ -81,6 +82,7 @@ func main() {
 		}
 	}
 	mux := http.NewServeMux()
+	workers := &federation.Registry{}
 	providerHealth := map[string]*provider.Supervisor{}
 	surface := func(r *http.Request) authz.Surface {
 		v := authz.ParseSurface(r.Header.Get("X-Orchestra-Surface"))
@@ -330,6 +332,43 @@ func main() {
 		}
 		json.NewEncoder(w).Encode(out)
 	})
+	mux.HandleFunc("/v1/federation/workers", func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodGet {
+			json.NewEncoder(w).Encode(workers.Snapshot())
+			return
+		}
+		if r.Method != http.MethodPost {
+			http.Error(w, "method not allowed", 405)
+			return
+		}
+		var worker federation.Worker
+		if json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&worker) != nil {
+			http.Error(w, "invalid worker", 400)
+			return
+		}
+		if err := workers.Register(worker); err != nil {
+			http.Error(w, err.Error(), 400)
+			return
+		}
+		w.WriteHeader(http.StatusCreated)
+		json.NewEncoder(w).Encode(worker)
+	})
+	mux.HandleFunc("/v1/federation/workers/", func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost || !strings.HasSuffix(r.URL.Path, "/heartbeat") {
+			http.Error(w, "not found", 404)
+			return
+		}
+		parts := strings.Split(strings.Trim(r.URL.Path, "/"), "/")
+		if len(parts) != 5 {
+			http.Error(w, "not found", 404)
+			return
+		}
+		if err := workers.Heartbeat(parts[3]); err != nil {
+			http.Error(w, err.Error(), 404)
+			return
+		}
+		w.WriteHeader(http.StatusNoContent)
+	})
 	if base := os.Getenv("ORCHESTRA_GITEA_URL"); base != "" {
 		g := provider.Gitea{BaseURL: base, Token: os.Getenv("ORCHESTRA_GITEA_TOKEN"), WebhookSecret: os.Getenv("ORCHESTRA_GITEA_WEBHOOK_SECRET"), Owner: os.Getenv("ORCHESTRA_GITEA_OWNER"), Repo: os.Getenv("ORCHESTRA_GITEA_REPO")}
 		reflecting := provider.ReflectingSink{Sink: s, Tasks: s, Reflector: g}
diff --git a/internal/federation/federation.go b/internal/federation/federation.go
new file mode 100644
index 0000000..d9e1837
--- /dev/null
+++ b/internal/federation/federation.go
@@ -0,0 +1,70 @@
+package federation
+
+import (
+	"errors"
+	"sync"
+	"time"
+)
+
+var ErrUnknownWorker = errors.New("unknown worker")
+
+type Worker struct {
+	ID       string    `json:"id"`
+	Address  string    `json:"address"`
+	Capacity int       `json:"capacity"`
+	LastSeen time.Time `json:"last_seen"`
+	Online   bool      `json:"online"`
+}
+
+type Registry struct {
+	mu      sync.Mutex
+	workers map[string]Worker
+	TTL     time.Duration
+}
+
+func (r *Registry) init() {
+	if r.TTL <= 0 {
+		r.TTL = 90 * time.Second
+	}
+	if r.workers == nil {
+		r.workers = map[string]Worker{}
+	}
+}
+func (r *Registry) Register(w Worker) error {
+	if w.ID == "" {
+		return errors.New("worker id required")
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.init()
+	w.LastSeen = time.Now().UTC()
+	w.Online = true
+	r.workers[w.ID] = w
+	return nil
+}
+func (r *Registry) Heartbeat(id string) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.init()
+	w, ok := r.workers[id]
+	if !ok {
+		return ErrUnknownWorker
+	}
+	w.LastSeen = time.Now().UTC()
+	w.Online = true
+	r.workers[id] = w
+	return nil
+}
+func (r *Registry) Snapshot() []Worker {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.init()
+	now := time.Now()
+	out := make([]Worker, 0, len(r.workers))
+	for id, w := range r.workers {
+		w.Online = now.Sub(w.LastSeen) <= r.TTL
+		r.workers[id] = w
+		out = append(out, w)
+	}
+	return out
+}
diff --git a/progress.md b/progress.md
index 27c2565..a8a25bb 100644
--- a/progress.md
+++ b/progress.md
@@ -69,6 +69,8 @@ Notification delivery now supports Telegram and ntfy fan-out for completion, fai
 
 Rotation now honors an optional herdr turn-boundary probe (`pane.status`) before hard-threshold release. Adapters without the optional capability retain occupancy-based fallback behavior.
 
+Federation control-plane foundations now include worker registration, heartbeat updates, TTL-based offline status, and `/v1/federation/workers` plus per-worker heartbeat endpoints. Remote event/lease transport and remote worktree ownership remain to be layered on this registry.
+
 Recommended order:
 
 1. Add the orchestration coordinator: lease → worktree → harness session → bootstrap → lifecycle events.