kami
87 lines
2.2 KiB
Go
87 lines
2.2 KiB
Go
// Package orchestrator connects router lease events to an opaque herdr
|
|
// session. It is deliberately small: scheduling remains in router and the
|
|
// adapter remains the only component that knows how to drive a harness.
|
|
package orchestrator
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"orchestra/internal/domain"
|
|
"orchestra/internal/herdr"
|
|
"orchestra/internal/store"
|
|
"sync"
|
|
)
|
|
|
|
type Worktrees interface {
|
|
Create(context.Context, domain.Task) (string, error)
|
|
}
|
|
type Adapters interface {
|
|
Adapter(string) (herdr.Adapter, error)
|
|
}
|
|
|
|
type Coordinator struct {
|
|
Store *store.Store
|
|
Worktrees Worktrees
|
|
Adapters Adapters
|
|
mu sync.Mutex
|
|
sessions map[string]herdr.Session
|
|
}
|
|
|
|
func (c *Coordinator) Start(ctx context.Context, e domain.Event) error {
|
|
if e.Type != "TaskLeased" {
|
|
return nil
|
|
}
|
|
if c.Store == nil || c.Worktrees == nil || c.Adapters == nil {
|
|
return fmt.Errorf("orchestrator: dependencies required")
|
|
}
|
|
t, ok := c.Store.Task(e.TaskID)
|
|
if !ok {
|
|
return domain.ErrNotFound
|
|
}
|
|
var p struct {
|
|
HarnessID string `json:"harness_id"`
|
|
HandoffRef string `json:"handoff_ref"`
|
|
}
|
|
if err := json.Unmarshal(e.Payload, &p); err != nil || p.HarnessID == "" {
|
|
return fmt.Errorf("orchestrator: invalid lease")
|
|
}
|
|
w, err := c.Worktrees.Create(ctx, t)
|
|
if err != nil {
|
|
return c.block(t, "worktree: "+err.Error())
|
|
}
|
|
a, err := c.Adapters.Adapter(p.HarnessID)
|
|
if err != nil {
|
|
return c.block(t, "adapter: "+err.Error())
|
|
}
|
|
s, err := a.Lease(ctx, t.ID, w)
|
|
if err != nil {
|
|
return c.block(t, "lease: "+err.Error())
|
|
}
|
|
if p.HandoffRef != "" {
|
|
if err = a.Bootstrap(ctx, s, p.HandoffRef); err != nil {
|
|
_ = a.Kill(ctx, s)
|
|
return c.block(t, "bootstrap: "+err.Error())
|
|
}
|
|
}
|
|
c.mu.Lock()
|
|
if c.sessions == nil {
|
|
c.sessions = map[string]herdr.Session{}
|
|
}
|
|
c.sessions[t.ID] = s
|
|
c.mu.Unlock()
|
|
return nil
|
|
}
|
|
|
|
func (c *Coordinator) block(t domain.Task, reason string) error {
|
|
b, _ := json.Marshal(map[string]string{"blocker": reason})
|
|
return c.Store.Append(domain.Event{ID: domain.NewID(), Type: "TaskBlocked", TaskID: t.ID, Version: t.Version + 1, Payload: b})
|
|
}
|
|
|
|
func (c *Coordinator) Session(taskID string) (herdr.Session, bool) {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
s, ok := c.sessions[taskID]
|
|
return s, ok
|
|
}
|