feat: rotate sessions on occupancy threshold
This commit is contained in:
@@ -49,6 +49,15 @@ func main() {
|
||||
}
|
||||
coordinator := &orchestrator.Coordinator{Store: s, 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 {
|
||||
hard = v
|
||||
}
|
||||
go func() {
|
||||
if monitorErr := coordinator.Monitor(context.Background(), hard, 30*time.Second); monitorErr != nil {
|
||||
log.Printf("orchestrator monitor: %v", monitorErr)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
mux := http.NewServeMux()
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Worktrees interface {
|
||||
@@ -68,6 +69,61 @@ type Coordinator struct {
|
||||
sessions map[string]herdr.Session
|
||||
}
|
||||
|
||||
// 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 interval <= 0 {
|
||||
interval = 30 * time.Second
|
||||
}
|
||||
t := time.NewTicker(interval)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-t.C:
|
||||
c.rotate(ctx, hard)
|
||||
}
|
||||
}
|
||||
}
|
||||
func (c *Coordinator) rotate(ctx context.Context, hard float64) {
|
||||
c.mu.Lock()
|
||||
sessions := make(map[string]herdr.Session, len(c.sessions))
|
||||
for id, s := range c.sessions {
|
||||
sessions[id] = s
|
||||
}
|
||||
c.mu.Unlock()
|
||||
for taskID, session := range sessions {
|
||||
task, ok := c.Store.Task(taskID)
|
||||
if !ok || task.State != domain.StateLeased {
|
||||
continue
|
||||
}
|
||||
a, err := c.Adapters.Adapter(session.Harness)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
occupancy, err := a.Occupancy(session)
|
||||
if err != nil || occupancy < hard {
|
||||
continue
|
||||
}
|
||||
ref, err := a.Release(ctx, session)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if ref == "" {
|
||||
continue
|
||||
}
|
||||
b, _ := json.Marshal(map[string]string{"handoff_ref": ref, "reason": "threshold"})
|
||||
e := domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: taskID, Version: task.Version + 1, Payload: b}
|
||||
if c.Store.Append(e) == nil {
|
||||
c.mu.Lock()
|
||||
delete(c.sessions, taskID)
|
||||
c.mu.Unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Coordinator) Start(ctx context.Context, e domain.Event) error {
|
||||
if e.Type != "TaskLeased" {
|
||||
return nil
|
||||
|
||||
+4
-4
@@ -7,22 +7,22 @@ Updated: 2026-07-26
|
||||
`go test ./...` passes, but the implementation is still a tested substrate/router prototype rather than a functioning unattended multi-harness orchestra. The following gaps were verified against `orchestra-spec (1).md` and the current code:
|
||||
|
||||
- **Harness execution is wired for configured Git/Codex deployments.** The router invokes the coordinator; `ORCHESTRA_REPO` + `ORCHESTRA_WORKTREE_ROOT` enable Git worktree creation, configured herdr socket adapters, session creation, and optional bootstrap. Other harness types and monitoring callbacks remain pending.
|
||||
- **Rotation is not implemented.** There is no turn-boundary callback, herdr event subscription, occupancy-triggered rotation, milestone/thrash trigger, or split-then-close coordinator.
|
||||
- **Rotation is partially implemented.** The configured coordinator polls adapter occupancy, releases above `ORCHESTRA_OCCUPANCY_HARD` (default 75%), and publishes a handoff-backed `TaskReleased` event. Turn-boundary callbacks, milestone/thrash triggers, and split-then-close safety remain pending.
|
||||
- **Lifecycle API payloads are invalid.** The release, complete, and block endpoints all emit `{"source":"api"}`, while validation requires `handoff_ref` or `reason`, `report_ref`, and `blocker` respectively. The documented lifecycle endpoints therefore cannot complete successfully.
|
||||
- **Provider integrations are partially wired.** JSONL watching and optional Gitea webhook/poll loops now start from environment configuration, but terminal reflection, provider health, cancellation, and delivery fan-out remain absent.
|
||||
- **CAS references are not content-verified at event append.** Lifecycle events check that referenced files exist, but do not verify that the file content hashes to the supplied reference.
|
||||
- **Replay bypasses event validation.** Startup replay unmarshals and applies events without validating the event envelope, payload schema, or sequence/version invariants.
|
||||
- **Snapshots are written but never loaded or used for replay acceleration.** Startup always replays the complete event log.
|
||||
- **Task creation projection is incomplete.** `parent`, `due`, `inherent_priority`, and `estimate` are defined in the domain model but are not projected from `TaskCreated` payloads.
|
||||
- **Occupancy support is incomplete relative to the spec.** Native readers exist, but Codex active-session discovery, opencode server/SSE plus fallback, and coordinator monitoring are not implemented.
|
||||
- **Occupancy support is incomplete relative to the spec.** Native readers and configured hard-threshold monitoring exist, but Codex active-session discovery, opencode server/SSE plus fallback, and turn-boundary monitoring are not implemented.
|
||||
- **Authorization is mostly enforced at HTTP ingress.** Lifecycle and approval handlers now call `AuthorizeEvent`; an absent surface still defaults to full-control Web, and non-HTTP/event-bus integrations remain unwired.
|
||||
|
||||
The first pass closed the store/API defects (lifecycle defaults, CAS content verification, validated replay, snapshot loading, and projection of task metadata) and added optional Gitea webhook/poll wiring. The remaining server-side gaps are below.
|
||||
|
||||
### Remaining server-side gaps
|
||||
|
||||
- **The orchestration coordinator is operational but incomplete.** It is constructed from deployment configuration, resolves a worktree, invokes `herdr.Adapter.Lease`, bootstraps handoffs, and blocks failed starts. Session mappings are in-memory and turn/lifecycle monitoring is still pending.
|
||||
- **Rotation is still absent.** No adapter turn-boundary callback, occupancy trigger, milestone/thrash trigger, handoff save/validate flow, split-then-close sequence, or lease transfer coordinator is wired into the server. The existing occupancy readers and anchor validator are standalone library primitives.
|
||||
- **The orchestration coordinator is operational but incomplete.** It is constructed from deployment configuration, resolves a worktree, invokes `herdr.Adapter.Lease`, bootstraps handoffs, monitors occupancy, and rotates through handoff references. Session mappings are in-memory and turn-boundary monitoring is still pending.
|
||||
- **Rotation remains incomplete.** Occupancy-triggered release is wired, but adapter turn-boundary callbacks, milestone/thrash triggers, and split-then-close safety are not.
|
||||
- **Harness discovery and registration are not operational.** Static herdr configuration and socket clients exist, but startup does not create adapters, ping configured herdrs, discover active Codex sessions, subscribe to opencode SSE, or run the required fallback/TTL monitoring loop.
|
||||
- **Provider ingestion is only partially wired.** JSONL and Gitea are available when configured, but there is no provider lifecycle management, cancellation, error health projection, terminal-state reflection, or provider fan-out.
|
||||
- **Lifecycle event contracts remain incomplete.** Validation does not enforce the spec's `expected_version`, `ttl`, `anchor_sha`, `receipt`, or optional `handoff_ref` relationships, and the HTTP API does not validate actor/surface authorization at the event construction site. Completion without a report currently creates a generated placeholder artifact rather than requiring the stop-hook/wrapper receipt described by the spec.
|
||||
|
||||
Reference in New Issue
Block a user