From ad32f29cd5c93b405b0837e6d91aa47f85c3792a Mon Sep 17 00:00:00 2001
From: kami <kami@kvmx.ru>
Date: Sun, 26 Jul 2026 20:28:41 +0400
Subject: [PATCH] persist and reconcile runtime sessions

---
 cmd/orchestra/main.go                 |  3 +-
 internal/orchestrator/orchestrator.go | 64 ++++++++++++++++++++++++++-
 progress.md                           |  2 +
 3 files changed, 67 insertions(+), 2 deletions(-)

diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go
index ad9bd00..56b100c 100644
--- a/cmd/orchestra/main.go
+++ b/cmd/orchestra/main.go
@@ -17,6 +17,7 @@ import (
 	"orchestra/internal/router"
 	"orchestra/internal/store"
 	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"time"
@@ -65,7 +66,7 @@ func main() {
 					log.Printf("herdr %s has unsupported harness %q", h.ID, h.Harness)
 				}
 			}
-			coordinator := &orchestrator.Coordinator{Store: s, Worktrees: orchestrator.GitWorktrees{Root: root, Repo: repo}, Adapters: orchestrator.AdapterFactory{Herdrs: adapters}}
+			coordinator := &orchestrator.Coordinator{Store: s, StatePath: filepath.Join(dir, "runtime-sessions.json"), Worktrees: orchestrator.GitWorktrees{Root: root, Repo: repo}, Adapters: orchestrator.AdapterFactory{Herdrs: adapters}}
 			rt.OnLease = func(e domain.Event) error { return coordinator.Start(context.Background(), e) }
 			hard := 0.75
 			if v, parseErr := strconv.ParseFloat(os.Getenv("ORCHESTRA_OCCUPANCY_HARD"), 64); parseErr == nil && v > 0 && v < 1 {
diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go
index 876c5ec..4dbe4d0 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -65,14 +65,71 @@ type Coordinator struct {
 	Store     *store.Store
 	Worktrees Worktrees
 	Adapters  Adapters
+	StatePath string
 	mu        sync.Mutex
 	sessions  map[string]herdr.Session
+	loaded    bool
+}
+
+func (c *Coordinator) loadSessions() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.loaded {
+		return
+	}
+	c.loaded = true
+	c.sessions = map[string]herdr.Session{}
+	if c.StatePath == "" {
+		return
+	}
+	b, err := os.ReadFile(c.StatePath)
+	if err != nil {
+		return
+	}
+	_ = json.Unmarshal(b, &c.sessions)
+}
+
+func (c *Coordinator) saveSessionsLocked() error {
+	if c.StatePath == "" {
+		return nil
+	}
+	b, err := json.Marshal(c.sessions)
+	if err != nil {
+		return err
+	}
+	tmp := c.StatePath + ".tmp"
+	if err = os.WriteFile(tmp, b, 0600); err != nil {
+		return err
+	}
+	return os.Rename(tmp, c.StatePath)
+}
+
+// Reconcile drops mappings whose task lease did not survive restart and kills
+// their recoverable herdr sessions so an orphan cannot keep consuming a slot.
+func (c *Coordinator) Reconcile(ctx context.Context) error {
+	c.loadSessions()
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	for taskID, session := range c.sessions {
+		t, ok := c.Store.Task(taskID)
+		if ok && t.State == domain.StateLeased {
+			continue
+		}
+		if a, err := c.Adapters.Adapter(session.Harness); err == nil {
+			_ = a.Kill(ctx, session)
+		}
+		delete(c.sessions, taskID)
+	}
+	return c.saveSessionsLocked()
 }
 
 // Monitor performs conservative hard-threshold rotation. The adapter owns
 // the handoff creation; the coordinator only publishes its content address
 // and frees the lease for pickup by the router.
 func (c *Coordinator) Monitor(ctx context.Context, hard float64, interval time.Duration) error {
+	if err := c.Reconcile(ctx); err != nil {
+		return err
+	}
 	if interval <= 0 {
 		interval = 30 * time.Second
 	}
@@ -88,6 +145,7 @@ func (c *Coordinator) Monitor(ctx context.Context, hard float64, interval time.D
 	}
 }
 func (c *Coordinator) rotate(ctx context.Context, hard float64) {
+	c.loadSessions()
 	c.mu.Lock()
 	sessions := make(map[string]herdr.Session, len(c.sessions))
 	for id, s := range c.sessions {
@@ -119,6 +177,7 @@ func (c *Coordinator) rotate(ctx context.Context, hard float64) {
 		if c.Store.Append(e) == nil {
 			c.mu.Lock()
 			delete(c.sessions, taskID)
+			_ = c.saveSessionsLocked()
 			c.mu.Unlock()
 		}
 	}
@@ -131,6 +190,7 @@ func (c *Coordinator) Start(ctx context.Context, e domain.Event) error {
 	if c.Store == nil || c.Worktrees == nil || c.Adapters == nil {
 		return fmt.Errorf("orchestrator: dependencies required")
 	}
+	c.loadSessions()
 	t, ok := c.Store.Task(e.TaskID)
 	if !ok {
 		return domain.ErrNotFound
@@ -165,8 +225,9 @@ func (c *Coordinator) Start(ctx context.Context, e domain.Event) error {
 		c.sessions = map[string]herdr.Session{}
 	}
 	c.sessions[t.ID] = s
+	err = c.saveSessionsLocked()
 	c.mu.Unlock()
-	return nil
+	return err
 }
 
 func (c *Coordinator) block(t domain.Task, reason string) error {
@@ -175,6 +236,7 @@ func (c *Coordinator) block(t domain.Task, reason string) error {
 }
 
 func (c *Coordinator) Session(taskID string) (herdr.Session, bool) {
+	c.loadSessions()
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	s, ok := c.sessions[taskID]
diff --git a/progress.md b/progress.md
index 7bb1008..b12cc56 100644
--- a/progress.md
+++ b/progress.md
@@ -59,6 +59,8 @@ Harness registration now uses configured `harness` and `protocol` fields, pings
 
 Added bounded `POST /v1/artifacts` CAS upload support for report/handoff evidence. It returns the verified content hash used by lifecycle events and rejects empty or oversized uploads.
 
+The coordinator now persists active task→herdr session mappings in an atomic runtime state file, reloads them after restart, reconciles them against durable task leases, kills stale recoverable sessions, and removes orphan mappings before monitoring begins.
+
 Recommended order:
 
 1. Add the orchestration coordinator: lease → worktree → harness session → bootstrap → lifecycle events.