checkpoint: multi-repo Gitea ingestion, per-project repos, rotation anchor_sha fix
Pre-existing uncommitted work found at session start: rotation now emits anchor_sha on TaskReleased (previously silently dropped by store.Append validation), multi-repo Gitea provider support, per-project git worktree roots, and associated test coverage. Committing as a checkpoint before starting remediation work tracked in AUDIT.md.
This commit is contained in:
kami
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"orchestra/internal/authz"
|
||||
"orchestra/internal/domain"
|
||||
"orchestra/internal/herdr"
|
||||
"orchestra/internal/orchestrator"
|
||||
@@ -128,7 +129,7 @@ func TestEndToEndIngestRouteLeaseRotateAndComplete(t *testing.T) {
|
||||
}
|
||||
got, _ := s.Task(task.ID)
|
||||
if got.State == domain.StateLeased && h.releases == 1 {
|
||||
if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: task.ID, Version: got.Version + 1, Payload: mustJSON(map[string]string{
|
||||
if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: task.ID, Version: got.Version + 1, Surface: string(authz.System), Payload: mustJSON(map[string]string{
|
||||
"handoff_ref": h.ref,
|
||||
"anchor_sha": "0123456789012345678901234567890123456789",
|
||||
})}); err != nil {
|
||||
@@ -153,7 +154,7 @@ func TestEndToEndIngestRouteLeaseRotateAndComplete(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: got.Version + 1, Payload: mustJSON(map[string]any{
|
||||
if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: got.Version + 1, Surface: string(authz.System), Payload: mustJSON(map[string]any{
|
||||
"report_ref": ref,
|
||||
"receipt": map[string]any{"harness_id": "h1", "consumed": 1},
|
||||
})}); err != nil {
|
||||
@@ -181,7 +182,7 @@ func TestRestartReplayAndReconcileKillsOrphan(t *testing.T) {
|
||||
}
|
||||
// A fresh coordinator sees the durable session, then drops it once the lease is gone.
|
||||
ref, _ := s.PutArtifact([]byte("handoff"))
|
||||
if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: task.ID, Version: 3, Payload: mustJSON(map[string]string{
|
||||
if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: task.ID, Version: 3, Surface: string(authz.System), Payload: mustJSON(map[string]string{
|
||||
"handoff_ref": ref,
|
||||
"anchor_sha": "0123456789012345678901234567890123456789",
|
||||
})}); err != nil {
|
||||
@@ -207,10 +208,10 @@ func TestRestartReplayAndReconcileKillsOrphan(t *testing.T) {
|
||||
func TestProviderRetryReflectionQuotaAndVersionConflict(t *testing.T) {
|
||||
s, r, _ := setup(t)
|
||||
task := ingest(t, s, "retry")
|
||||
if err := s.Append(domain.Event{Type: "QuotaReported", TaskID: "quota", Version: 1, Payload: mustJSON(map[string]any{"harness_id": "h1", "consumed": 90.0})}); err != nil {
|
||||
if err := s.Append(domain.Event{Type: "QuotaReported", TaskID: "quota", Version: 1, Surface: string(authz.System), Payload: mustJSON(map[string]any{"harness_id": "h1", "consumed": 90.0})}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rt := router.Router{Store: s, Registry: r, Reachability: alwaysReachable{}, Availability: router.QuotaAvailability{Store: s, Limits: map[string]float64{"h1": 100}, Now: time.Now}}
|
||||
rt := router.Router{Store: s, Registry: r, Reachability: alwaysReachable{}, Availability: router.QuotaAvailability{Store: s, Limits: map[string]router.QuotaWindowLimits{"h1": {Weekly: 100}}, Now: time.Now}}
|
||||
if got, _ := rt.AssignPending(); len(got) != 0 {
|
||||
t.Fatalf("quota assigned=%d", len(got))
|
||||
}
|
||||
@@ -225,7 +226,7 @@ func TestProviderRetryReflectionQuotaAndVersionConflict(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
reflector := &fakeReflector{}
|
||||
if err := (provider.ReflectingSink{Sink: s, Tasks: s, Reflector: reflector}).Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: 3, Payload: mustJSON(map[string]any{
|
||||
if err := (provider.ReflectingSink{Sink: s, Tasks: s, Reflector: reflector}).Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: 3, Surface: string(authz.System), Payload: mustJSON(map[string]any{
|
||||
"report_ref": ref,
|
||||
"receipt": map[string]any{"harness_id": "h1", "consumed": 1},
|
||||
})}); err != nil {
|
||||
|
||||
@@ -0,0 +1,213 @@
|
||||
package integration
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"orchestra/internal/authz"
|
||||
"orchestra/internal/domain"
|
||||
"orchestra/internal/federation"
|
||||
"orchestra/internal/herdr"
|
||||
"orchestra/internal/operations"
|
||||
"orchestra/internal/orchestrator"
|
||||
"orchestra/internal/registry"
|
||||
"orchestra/internal/router"
|
||||
"orchestra/internal/store"
|
||||
"os/exec"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func gitInit(t *testing.T, dir, marker string) string {
|
||||
t.Helper()
|
||||
run := func(args ...string) {
|
||||
cmd := exec.Command("git", append([]string{"-C", dir}, args...)...)
|
||||
if out, err := cmd.CombinedOutput(); err != nil {
|
||||
t.Fatalf("git %v: %v: %s", args, err, out)
|
||||
}
|
||||
}
|
||||
run("init")
|
||||
run("config", "user.email", "t@t")
|
||||
run("config", "user.name", "t")
|
||||
run("commit", "--allow-empty", "-m", "init: "+marker)
|
||||
head, err := herdr.HeadSHA(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return head
|
||||
}
|
||||
|
||||
type fixedWorktree struct{ path string }
|
||||
|
||||
func (w fixedWorktree) Create(context.Context, domain.Task) (string, error) { return w.path, nil }
|
||||
|
||||
// machineAdapter is like harness but actually threads the worktree it was
|
||||
// leased against into the Session, the way a real herdr adapter must — the
|
||||
// shared harness fixture ignores it, which is fine for single-checkout
|
||||
// tests but would silently defeat this one's anchor_sha assertions.
|
||||
type machineAdapter struct{ *harness }
|
||||
|
||||
func (a machineAdapter) Lease(_ context.Context, _ string, worktree string) (herdr.Session, error) {
|
||||
return herdr.Session{Harness: "h1", PaneID: "pane-1", Worktree: worktree}, nil
|
||||
}
|
||||
|
||||
type singleAdapter struct{ a herdr.Adapter }
|
||||
|
||||
func (s singleAdapter) Adapter(string) (herdr.Adapter, error) { return s.a, nil }
|
||||
|
||||
// TestCrossMachineLeaseAnchorAndQuotaArePerHost exercises spec §2.1/§9 open
|
||||
// question 8 end to end using two independent local git checkouts standing
|
||||
// in for homesrv and workpc: each machine's Coordinator only ever validates
|
||||
// and stamps anchor_sha from its OWN local checkout (never the other
|
||||
// machine's), and quota consumption is accounted strictly per harness/host
|
||||
// so one machine exhausting its window cannot block the other from being
|
||||
// leased work. Federation registration/heartbeat plumbing (already covered
|
||||
// by internal/federation's own tests) is exercised alongside this to prove
|
||||
// the pieces fit together, not just in isolation.
|
||||
func TestCrossMachineLeaseAnchorAndQuotaArePerHost(t *testing.T) {
|
||||
homesrvRepo, workpcRepo := t.TempDir(), t.TempDir()
|
||||
homesrvHead := gitInit(t, homesrvRepo, "homesrv")
|
||||
workpcHead := gitInit(t, workpcRepo, "workpc")
|
||||
if homesrvHead == workpcHead {
|
||||
t.Fatal("test setup: expected distinct checkouts")
|
||||
}
|
||||
|
||||
s, err := store.Open(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// The federation registry is homesrv's view of workpc as an intermittent
|
||||
// worker (spec §2.1): it must be registered and reachable before the
|
||||
// router would ever consider leasing to it.
|
||||
workers := &federation.Registry{}
|
||||
if err := workers.Register(federation.Worker{ID: "workpc", Address: "workpc.mesh", Token: "secret"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := workers.Heartbeat("workpc"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
online := false
|
||||
for _, w := range workers.Snapshot() {
|
||||
if w.ID == "workpc" && w.Online {
|
||||
online = true
|
||||
}
|
||||
}
|
||||
if !online {
|
||||
t.Fatal("workpc worker should be online after heartbeat")
|
||||
}
|
||||
|
||||
// Two tasks, one leased to each machine's harness.
|
||||
mk := func(id string) domain.Task {
|
||||
b, _ := json.Marshal(map[string]any{"source": "jsonl", "external_id": id, "project": "p"})
|
||||
if err := s.Append(domain.Event{ID: id, Type: "TaskCreated", TaskID: id, Version: 1, Surface: string(authz.System), Payload: b}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
task, _ := s.Task(id)
|
||||
return task
|
||||
}
|
||||
homesrvTask := mk("home-task")
|
||||
workpcTask := mk("workpc-task")
|
||||
|
||||
homesrvAdapter := &harness{occupancy: .95, boundary: true, ref: mustArtifact(t, s, "home-handoff")}
|
||||
workpcAdapter := &harness{occupancy: .95, boundary: true, ref: mustArtifact(t, s, "workpc-handoff")}
|
||||
|
||||
homesrvCoord := &orchestrator.Coordinator{Store: s, Worktrees: fixedWorktree{homesrvRepo}, Adapters: singleAdapter{machineAdapter{homesrvAdapter}}, StatePath: t.TempDir() + "/home-sessions.json"}
|
||||
workpcCoord := &orchestrator.Coordinator{Store: s, Worktrees: fixedWorktree{workpcRepo}, Adapters: singleAdapter{machineAdapter{workpcAdapter}}, StatePath: t.TempDir() + "/workpc-sessions.json"}
|
||||
|
||||
homesrvLease, err := s.Lease(homesrvTask.ID, "homesrv-h1", time.Minute)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := homesrvCoord.Start(context.Background(), homesrvLease); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
workpcLease, err := s.Lease(workpcTask.ID, "workpc-h1", time.Minute)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := workpcCoord.Start(context.Background(), workpcLease); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
ctxHome, cancelHome := context.WithCancel(context.Background())
|
||||
defer cancelHome()
|
||||
ctxWork, cancelWork := context.WithCancel(context.Background())
|
||||
defer cancelWork()
|
||||
go homesrvCoord.Monitor(ctxHome, .8, time.Millisecond)
|
||||
go workpcCoord.Monitor(ctxWork, .8, time.Millisecond)
|
||||
|
||||
waitQueued := func(id string) domain.Task {
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if t, ok := s.Task(id); ok && t.State == domain.StateQueued {
|
||||
return t
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
t.Fatalf("task %s never rotated back to queued", id)
|
||||
return domain.Task{}
|
||||
}
|
||||
waitQueued(homesrvTask.ID)
|
||||
waitQueued(workpcTask.ID)
|
||||
|
||||
// Each machine's coordinator must have stamped anchor_sha from its OWN
|
||||
// checkout — never the other machine's HEAD, and never each other's.
|
||||
anchorFor := func(taskID string) string {
|
||||
for _, e := range s.Events(0) {
|
||||
if e.TaskID != taskID || e.Type != "TaskReleased" {
|
||||
continue
|
||||
}
|
||||
var p map[string]any
|
||||
_ = json.Unmarshal(e.Payload, &p)
|
||||
return p["anchor_sha"].(string)
|
||||
}
|
||||
t.Fatalf("no TaskReleased found for %s", taskID)
|
||||
return ""
|
||||
}
|
||||
if got := anchorFor(homesrvTask.ID); got != homesrvHead {
|
||||
t.Fatalf("homesrv anchor_sha=%s want=%s (must validate against its own checkout)", got, homesrvHead)
|
||||
}
|
||||
if got := anchorFor(workpcTask.ID); got != workpcHead {
|
||||
t.Fatalf("workpc anchor_sha=%s want=%s (must validate against its own checkout, not homesrv's)", got, workpcHead)
|
||||
}
|
||||
|
||||
// Quota is accounted per host/harness: exhausting homesrv-h1 must not
|
||||
// affect workpc-h1's availability, and vice versa (spec §2.1, §7.2).
|
||||
quotaReport := func(harness string, consumed float64) {
|
||||
p, _ := json.Marshal(map[string]any{"harness_id": harness, "consumed": consumed})
|
||||
if err := s.Append(domain.Event{ID: domain.NewID(), Type: "QuotaReported", TaskID: "quota", Version: 1, Surface: string(authz.System), Payload: p}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
quotaReport("homesrv-h1", 95)
|
||||
limits := map[string]router.QuotaWindowLimits{
|
||||
"homesrv-h1": {Weekly: 100},
|
||||
"workpc-h1": {Weekly: 100},
|
||||
}
|
||||
avail := router.QuotaAvailability{Store: s, Limits: limits}
|
||||
if avail.Available(registry.Herdr{ID: "homesrv-h1"}) {
|
||||
t.Fatal("homesrv-h1 should be quota-exhausted at 95/100")
|
||||
}
|
||||
if !avail.Available(registry.Herdr{ID: "workpc-h1"}) {
|
||||
t.Fatal("workpc-h1 exhaustion leaked from homesrv-h1's per-host accounting")
|
||||
}
|
||||
|
||||
// The windowed quota aggregation used by the brief (spec §7.4) must also
|
||||
// keep the two hosts separate.
|
||||
sums := operations.AggregateQuota(s.Events(0), time.Now().Add(-time.Hour), time.Now().Add(time.Hour))
|
||||
if sums["homesrv-h1"] != 95 {
|
||||
t.Fatalf("homesrv-h1 aggregate=%v want 95", sums["homesrv-h1"])
|
||||
}
|
||||
if sums["workpc-h1"] != 0 {
|
||||
t.Fatalf("workpc-h1 aggregate=%v want 0 (must not inherit homesrv-h1's receipts)", sums["workpc-h1"])
|
||||
}
|
||||
}
|
||||
|
||||
func mustArtifact(t *testing.T, s *store.Store, content string) string {
|
||||
t.Helper()
|
||||
ref, err := s.PutArtifact([]byte(content))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return ref
|
||||
}
|
||||
Reference in New Issue
Block a user