persist and reconcile runtime sessions

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]