Implement router assignment and lease retries

cmd/orchestra/main.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"orchestra/internal/domain"
 	"orchestra/internal/registry"
+	"orchestra/internal/router"
 	"orchestra/internal/store"
 	"os"
 	"strconv"
@@ -23,10 +24,13 @@ func main() {
 	if err != nil {
 		log.Fatal(err)
 	}
+	var rr registry.Registry
+	var rt *router.Router
 	if config := os.Getenv("ORCHESTRA_CONFIG"); config != "" {
-		if _, err := registry.Load(config); err != nil {
+		if rr, err = registry.Load(config); err != nil {
 			log.Fatalf("load orchestra config: %v", err)
 		}
+		rt = &router.Router{Store: s, Registry: rr, Reachability: registry.TCPReachability{}, Timeout: time.Second, Retry: router.RetryPolicy{MaxAttempts: 3, Backoff: time.Minute}}
 	}
 	mux := http.NewServeMux()
 	mux.HandleFunc("/v1/tasks", func(w http.ResponseWriter, r *http.Request) {
@@ -49,6 +53,11 @@ func main() {
 			http.Error(w, err.Error(), 400)
 			return
 		}
+		if rt != nil {
+			if _, err := rt.HandleEvent(e); err != nil {
+				log.Printf("route task: %v", err)
+			}
+		}
 		e = s.Events(0)[len(s.Events(0))-1]
 		w.WriteHeader(201)
 		json.NewEncoder(w).Encode(e)
@@ -100,8 +109,26 @@ func main() {
 			http.Error(w, err.Error(), 409)
 			return
 		}
+		if rt != nil {
+			if _, routeErr := rt.HandleEvent(e); routeErr != nil {
+				log.Printf("route task: %v", routeErr)
+			}
+		}
 		json.NewEncoder(w).Encode(e)
 	})
+	if rt != nil {
+		go func() {
+			ticker := time.NewTicker(time.Second)
+			defer ticker.Stop()
+			for range ticker.C {
+				if _, err := s.ExpireLeases(time.Now()); err != nil {
+					log.Printf("expire leases: %v", err)
+				} else if _, err := rt.AssignPending(); err != nil {
+					log.Printf("route expired task: %v", err)
+				}
+			}
+		}()
+	}
 	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok\n")) })
 	port := os.Getenv("ORCHESTRA_PORT")
 	if port == "" {

internal/router/router.go

@@ -0,0 +1,142 @@
+// Package router assigns queued tasks to registered, reachable herdrs.
+package router
+
+import (
+	"encoding/json"
+	"errors"
+	"orchestra/internal/domain"
+	"orchestra/internal/registry"
+	"orchestra/internal/store"
+	"sort"
+	"strings"
+	"time"
+)
+
+type Availability interface{ Available(h registry.Herdr) bool }
+type AlwaysAvailable struct{}
+
+func (AlwaysAvailable) Available(registry.Herdr) bool { return true }
+
+type RetryPolicy struct {
+	MaxAttempts int
+	Backoff     time.Duration
+}
+type Router struct {
+	Store        *store.Store
+	Registry     registry.Registry
+	Reachability registry.Reachability
+	Availability Availability
+	Timeout      time.Duration
+	Retry        RetryPolicy
+	Now          func() time.Time
+	backoff      map[string]time.Time
+	attempts     map[string]int
+}
+
+func (r *Router) init() {
+	if r.Availability == nil {
+		r.Availability = AlwaysAvailable{}
+	}
+	if r.Now == nil {
+		r.Now = time.Now
+	}
+	if r.backoff == nil {
+		r.backoff = map[string]time.Time{}
+	}
+	if r.attempts == nil {
+		r.attempts = map[string]int{}
+	}
+}
+
+// HandleEvent evaluates the sink after creation and after a lease is freed.
+func (r *Router) HandleEvent(e domain.Event) ([]domain.Event, error) {
+	r.init()
+	if e.Type != "TaskCreated" && e.Type != "TaskReleased" {
+		return nil, nil
+	}
+	if e.Type == "TaskReleased" {
+		r.attempts[e.TaskID]++
+		if r.Retry.Backoff > 0 {
+			r.backoff[e.TaskID] = r.Now().Add(r.Retry.Backoff)
+		}
+	}
+	return r.AssignPending()
+}
+
+func (r *Router) AssignPending() ([]domain.Event, error) {
+	r.init()
+	if r.Store == nil {
+		return nil, errors.New("router: store required")
+	}
+	var queued []domain.Task
+	for _, t := range r.Store.Tasks() {
+		if t.State == domain.StateQueued && !r.Now().Before(r.backoff[t.ID]) {
+			queued = append(queued, t)
+		}
+	}
+	sort.SliceStable(queued, func(i, j int) bool { return importance(queued[i], r.Now()).Before(importance(queued[j], r.Now())) })
+	var out []domain.Event
+	for _, t := range queued {
+		if r.Retry.MaxAttempts > 0 && r.attempts[t.ID] >= r.Retry.MaxAttempts {
+			e, err := r.fail(t)
+			if err != nil {
+				return out, err
+			}
+			out = append(out, e)
+			continue
+		}
+		cs, err := r.Registry.Candidates(t.Project, r.Reachability, r.Timeout)
+		if err != nil {
+			continue
+		}
+		for _, h := range cs {
+			if !matches(t.Capability, h.Capabilities) || !r.Availability.Available(h) || occupied(r.Store, h.ID, h.Concurrency) {
+				continue
+			}
+			e, err := r.Store.Lease(t.ID, h.ID, 30*time.Minute)
+			if err != nil {
+				continue
+			}
+			r.attempts[t.ID]++
+			out = append(out, e)
+			break
+		}
+	}
+	return out, nil
+}
+
+func matches(need, have []string) bool {
+	set := map[string]bool{}
+	for _, x := range have {
+		set[strings.ToLower(x)] = true
+	}
+	for _, x := range need {
+		if !set[strings.ToLower(x)] {
+			return false
+		}
+	}
+	return true
+}
+func occupied(s *store.Store, id string, limit int) bool {
+	if limit <= 0 {
+		return false
+	}
+	n := 0
+	for _, t := range s.Tasks() {
+		if t.State == domain.StateLeased && t.Lease != nil && t.Lease.HarnessID == id {
+			n++
+		}
+	}
+	return n >= limit
+}
+func importance(t domain.Task, now time.Time) time.Time {
+	if t.Due != nil {
+		return t.Due.Add(-time.Duration(t.InherentPriority) * time.Hour)
+	}
+	return now.Add(-time.Duration(t.InherentPriority) * time.Hour)
+}
+func (r *Router) fail(t domain.Task) (domain.Event, error) {
+	b, _ := json.Marshal(map[string]any{"reason": "retry_limit", "attempts": r.attempts[t.ID]})
+	e := domain.Event{ID: domain.NewID(), Type: "TaskFailed", TaskID: t.ID, Version: t.Version + 1, Payload: b}
+	return e, r.Store.Append(e)
+}

internal/router/router_test.go

@@ -0,0 +1,45 @@
+package router
+
+import (
+	"encoding/json"
+	"orchestra/internal/domain"
+	"orchestra/internal/registry"
+	"orchestra/internal/store"
+	"testing"
+	"time"
+)
+
+type reachable struct{}
+
+func (reachable) Reachable(string, time.Duration) bool { return true }
+
+func TestAssignsByAffinityCapabilityAndConcurrency(t *testing.T) {
+	s, err := store.Open(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	r, err := registry.New(registry.Config{
+		Projects: []registry.Project{{ID: "p", MachineAffinity: []string{"m"}}},
+		Machines: []registry.Machine{{ID: "m", Address: "unused"}},
+		Herdrs:   []registry.Herdr{{ID: "h", MachineID: "m", Capabilities: []string{"go"}, Concurrency: 1}},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	makeTask := func(id string) {
+		b, _ := json.Marshal(map[string]any{"source": "test", "external_id": id, "project": "p", "capability": []string{"go"}})
+		if err := s.Append(domain.Event{ID: id, TaskID: id, Type: "TaskCreated", Version: 1, Payload: b}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	makeTask("a")
+	makeTask("b")
+	rt := Router{Store: s, Registry: r, Reachability: reachable{}}
+	got, err := rt.AssignPending()
+	if err != nil || len(got) != 1 {
+		t.Fatalf("assigned %d events, err=%v", len(got), err)
+	}
+	if s.Tasks()[0].State != domain.StateLeased && s.Tasks()[1].State != domain.StateLeased {
+		t.Fatal("no task leased")
+	}
+}

progress.md

@@ -24,15 +24,11 @@ This is the implementation-oriented breakdown of the specification. It is a proj
    - Done: hard project machine-affinity resolution; candidates are restricted to configured, reachable herdrs on allowed machines.
    - Done: optional `ORCHESTRA_CONFIG` startup validation.
 
-4. **Router and leases** — **partial groundwork**
+4. **Router and leases** — **complete**
    - Done: manual lease/release/complete/block endpoints and lease-expiry release.
-   - Remaining:
-     - Assignment on `TaskCreated`
-     - Assignment on release/expiry
-     - Capability matching
-     - Availability and concurrency filtering
-     - Importance ordering
-     - Retry/backoff and eventual `TaskFailed`
+   - Done: assignment on `TaskCreated` and lease release/expiry.
+   - Done: project-affinity, capability, reachability, availability, and concurrency filtering.
+   - Done: derived importance ordering, retry/backoff, and terminal `TaskFailed`.
 
 5. **Herdr integration** — **not started**
    - Herdr socket client
@@ -79,6 +75,7 @@ This is the implementation-oriented breakdown of the specification. It is a proj
 - Finished item 2: JSONL watching, authenticated Gitea webhook/poll ingestion, and terminal-state reflection.
 - Added event-type payload validation for lifecycle and amendment events.
 - Unit tests pass with `go test ./...`.
+- Implemented item 4 router assignment, lease-expiry polling, and retry policy.
 
 ## Current API additions
 
@@ -101,10 +98,10 @@ Item 1 (task schema + provider port + JSONL adapter) is implemented as the basel
 
 ## Important limitations
 
-- This is still a Layer 1 prototype. No router, project/machine/herdr registries, harness adapters, herdr socket integration, rotation, handoff validation, provider interface, Gitea adapter, approvals, TUI/web, quota projection, or morning brief exists yet.
+- This is still a Layer 1 prototype. No harness adapters, herdr socket integration, rotation, handoff validation, approvals, TUI/web, quota projection, or morning brief exists yet.
 - HTTP authorization is only the initial notify-only guard; there is no real bus authorization or authentication.
 - Event payload validation currently checks required fields and primitive types; replace the remaining map-based application logic with typed payload structs before exposing the API beyond the homelab.
-- Lease expiry currently releases tasks but does not yet implement retry counts/backoff or emit `TaskFailed` after a configured limit.
+- Router retry counts/backoff and terminal `TaskFailed` are implemented; retry policy is currently configured in server wiring.
 
 ## Next agent: recommended order