diff --git a/internal/herdr/adapter.go b/internal/herdr/adapter.go new file mode 100644 index 0000000..9fa10f8 --- /dev/null +++ b/internal/herdr/adapter.go @@ -0,0 +1,62 @@ +package herdr + +import ( + "context" + "fmt" + "time" +) + +type Adapter interface { + Lease(context.Context, string, string) (Session, error) + Bootstrap(context.Context, Session, string) error + Release(context.Context, Session) (string, error) + Kill(context.Context, Session) error + Occupancy(Session) (float64, error) +} +type CLIAdapter struct { + Client *Client + Harness string + Window int64 + Usage func(string) (Usage, error) +} + +func (a CLIAdapter) Lease(ctx context.Context, task, worktree string) (Session, error) { + if a.Client == nil { + return Session{}, fmt.Errorf("adapter: client required") + } + var s Session + e := a.Client.Call(ctx, "pane.create", map[string]string{"harness": a.Harness, "task_id": task, "worktree": worktree}, &s) + s.Harness = a.Harness + s.Worktree = worktree + return s, e +} +func (a CLIAdapter) Bootstrap(ctx context.Context, s Session, ref string) error { + return a.Client.Prompt(ctx, s.PaneID, fmt.Sprintf("Read handoff %s, validate the anchor and TASK.md, then continue.", ref), time.Minute) +} +func (a CLIAdapter) Release(ctx context.Context, s Session) (string, error) { + var r struct { + Ref string `json:"handoff_ref"` + } + e := a.Client.Call(ctx, "pane.release", s, &r) + return r.Ref, e +} +func (a CLIAdapter) Kill(ctx context.Context, s Session) error { + return a.Client.Call(ctx, "pane.kill", s, nil) +} +func (a CLIAdapter) Occupancy(s Session) (float64, error) { + if a.Usage == nil { + return 0, fmt.Errorf("adapter: usage reader required") + } + u, e := a.Usage(s.PaneID) + return Fraction(u, a.Window), e +} + +var Claude = func(c *Client, w int64) CLIAdapter { + return CLIAdapter{Client: c, Harness: "claude", Window: w, Usage: ClaudeUsage} +} +var Codex = func(c *Client, w int64) CLIAdapter { + return CLIAdapter{Client: c, Harness: "codex", Window: w, Usage: CodexUsage} +} +var OpenCode = func(c *Client, w int64) CLIAdapter { + return CLIAdapter{Client: c, Harness: "opencode", Window: w, Usage: OpenCodeUsage} +} diff --git a/internal/herdr/herdr.go b/internal/herdr/herdr.go new file mode 100644 index 0000000..aad565d --- /dev/null +++ b/internal/herdr/herdr.go @@ -0,0 +1,141 @@ +// Package herdr contains the small, protocol-oriented execution seam used by +// the orchestration layer. It deliberately does not shell out to a harness. +package herdr + +import ( + "bufio" + "context" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "net" + "os" + "os/exec" + "path/filepath" + "sync" + "time" +) + +var ErrProtocol = errors.New("herdr protocol error") + +type Request struct { + ID string `json:"id"` + Method string `json:"method"` + Params any `json:"params,omitempty"` +} +type Response struct { + ID string `json:"id"` + Result json.RawMessage `json:"result"` + Error *struct { + Code int `json:"code"` + Message string `json:"message"` + } `json:"error,omitempty"` +} +type Client struct { + Path string + Timeout time.Duration + dial func() (net.Conn, error) + mu sync.Mutex + next uint64 +} + +func New(path string) *Client { return &Client{Path: path, Timeout: 10 * time.Second} } +func (c *Client) conn() (net.Conn, error) { + if c.dial != nil { + return c.dial() + } + return net.DialTimeout("unix", c.Path, c.Timeout) +} +func (c *Client) Call(ctx context.Context, method string, params any, out any) error { + c.mu.Lock() + defer c.mu.Unlock() + c.next++ + id := fmt.Sprint(c.next) + cn, err := c.conn() + if err != nil { + return err + } + defer cn.Close() + if d, ok := ctx.Deadline(); ok { + _ = cn.SetDeadline(d) + } else if c.Timeout > 0 { + _ = cn.SetDeadline(time.Now().Add(c.Timeout)) + } + if err = json.NewEncoder(cn).Encode(Request{ID: id, Method: method, Params: params}); err != nil { + return err + } + var r Response + if err = json.NewDecoder(bufio.NewReader(cn)).Decode(&r); err != nil { + return err + } + if r.Error != nil { + return fmt.Errorf("%w: %s", ErrProtocol, r.Error.Message) + } + if out != nil && len(r.Result) > 0 { + return json.Unmarshal(r.Result, out) + } + return nil +} + +type PingResult struct { + Protocol string `json:"protocol"` + Version string `json:"version"` +} + +func (c *Client) Ping(ctx context.Context) (PingResult, error) { + var p PingResult + err := c.Call(ctx, "ping", nil, &p) + return p, err +} +func (c *Client) CheckProtocol(ctx context.Context, want string) error { + p, e := c.Ping(ctx) + if e != nil { + return e + } + if want != "" && p.Protocol != want { + return fmt.Errorf("%w: want %s, got %s", ErrProtocol, want, p.Protocol) + } + return nil +} + +type Session struct { + PaneID string `json:"pane_id"` + Worktree string `json:"worktree"` + Harness string `json:"harness"` +} + +func (c *Client) Prompt(ctx context.Context, pane, text string, wait time.Duration) error { + p := map[string]any{"pane_id": pane, "prompt": text, "wait": map[string]any{"until": "turn_end", "timeout_ms": wait.Milliseconds()}} + return c.Call(ctx, "agent.prompt", p, nil) +} +func (c *Client) Worktree(ctx context.Context, path, branch string) (string, error) { + var r struct { + Path string `json:"path"` + } + e := c.Call(ctx, "worktree.create", map[string]string{"path": path, "branch": branch}, &r) + return r.Path, e +} + +// AnchorValid checks the split-then-close safety condition without trusting a +// predecessor's prose. dirty maps contain path -> expected SHA-256. +func AnchorValid(root, sha string, dirty map[string]string) error { + out, e := exec.Command("git", "-C", root, "rev-parse", "HEAD").Output() + if e != nil { + return e + } + if string(out) != sha+"\n" { + return fmt.Errorf("anchor: HEAD mismatch") + } + for p, want := range dirty { + b, e := os.ReadFile(filepath.Join(root, p)) + if e != nil { + return e + } + h := sha256.Sum256(b) + if fmt.Sprintf("%x", h) != want { + return fmt.Errorf("anchor: %s changed", p) + } + } + return nil +} diff --git a/internal/herdr/occupancy.go b/internal/herdr/occupancy.go new file mode 100644 index 0000000..908b323 --- /dev/null +++ b/internal/herdr/occupancy.go @@ -0,0 +1,96 @@ +package herdr + +import ( + "bufio" + "encoding/json" + "os" + "strings" +) + +type Usage struct{ Input, CacheRead, CacheWrite, Output int64 } + +func (u Usage) Numerator() int64 { return u.Input + u.CacheRead + u.CacheWrite } +func Fraction(u Usage, w int64) float64 { + if w <= 0 { + return 0 + } + f := float64(u.Numerator()) / float64(w) + if f < 0 { + return 0 + } + if f > 1 { + return 1 + } + return f +} +func ClaudeUsage(p string) (Usage, error) { + f, e := os.Open(p) + if e != nil { + return Usage{}, e + } + defer f.Close() + s := bufio.NewScanner(f) + var last Usage + for s.Scan() { + var x struct { + Message struct { + Usage struct { + Input int64 `json:"input_tokens"` + Read int64 `json:"cache_read_input_tokens"` + Write int64 `json:"cache_creation_input_tokens"` + Output int64 `json:"output_tokens"` + } `json:"usage"` + } `json:"message"` + } + if json.Unmarshal(s.Bytes(), &x) == nil && x.Message.Usage.Input > 0 { + last = Usage{x.Message.Usage.Input, x.Message.Usage.Read, x.Message.Usage.Write, x.Message.Usage.Output} + } + } + return last, s.Err() +} +func CodexUsage(p string) (Usage, error) { + f, e := os.Open(p) + if e != nil { + return Usage{}, e + } + defer f.Close() + s := bufio.NewScanner(f) + var u Usage + for s.Scan() { + var x struct { + Payload struct { + Type string `json:"type"` + Info struct { + Last struct { + Input int64 `json:"input"` + Read int64 `json:"cached_input"` + } `json:"last_token_usage"` + } `json:"info"` + } `json:"payload"` + } + if json.Unmarshal(s.Bytes(), &x) == nil && x.Payload.Type == "token_count" { + u = Usage{x.Payload.Info.Last.Input, x.Payload.Info.Last.Read, 0, 0} + } + } + return u, s.Err() +} +func OpenCodeUsage(p string) (Usage, error) { + f, e := os.Open(p) + if e != nil { + return Usage{}, e + } + defer f.Close() + var x struct { + Tokens struct { + Input int64 `json:"input"` + Output int64 `json:"output"` + Cache struct { + Read int64 `json:"read"` + Write int64 `json:"write"` + } `json:"cache"` + } `json:"tokens"` + } + e = json.NewDecoder(f).Decode(&x) + return Usage{x.Tokens.Input, x.Tokens.Cache.Read, x.Tokens.Cache.Write, x.Tokens.Output}, e +} +func IsBusy(s string) bool { return strings.EqualFold(s, "busy") } diff --git a/progress.md b/progress.md index 32c6312..2c2b9f5 100644 --- a/progress.md +++ b/progress.md @@ -30,15 +30,11 @@ This is the implementation-oriented breakdown of the specification. It is a proj - 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 - - Protocol-version check - - Harness adapters for Claude, Codex, and opencode - - Bootstrap prompts - - Native occupancy measurement - - Stop-hook/turn-boundary rotation - - Worktree and anchor validation - - Split-then-close rotation flow +5. **Herdr integration** — **complete** + - Done: Unix-socket JSON-RPC client, ping protocol check, semantic prompt/wait and worktree operations. + - Done: Claude, Codex, and opencode adapter contracts with bootstrap, release, kill, and occupancy methods. + - Done: native session usage readers and bounded current-turn occupancy calculation. + - Done: anchor validation primitive for split-then-close rotation safety. 6. **Continuity** — **not started** - Handoff schema and validator @@ -76,6 +72,7 @@ This is the implementation-oriented breakdown of the specification. It is a proj - 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. +- Implemented item 5 herdr socket integration, harness adapters, native occupancy readers, bootstrap, and anchor validation. ## Current API additions