diff --git a/internal/integration/end_to_end_test.go b/internal/integration/end_to_end_test.go index d1959b0..2ee9d35 100644 --- a/internal/integration/end_to_end_test.go +++ b/internal/integration/end_to_end_test.go @@ -133,7 +133,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: 5, Payload: mustJSON(map[string]string{"report_ref": ref})}); err != nil { + if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCompleted", TaskID: task.ID, Version: got.Version + 1, Payload: mustJSON(map[string]string{"report_ref": ref})}); err != nil { t.Fatal(err) } got, _ = s.Task(task.ID) diff --git a/internal/operations/operations.go b/internal/operations/operations.go index c2bae0e..e862f52 100644 --- a/internal/operations/operations.go +++ b/internal/operations/operations.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "orchestra/internal/domain" + "orchestra/internal/store" "os/exec" "strings" "time" @@ -29,6 +30,31 @@ type StandupItem struct { Title string `json:"title,omitempty"` } +// QuotaReceipt is the native usage receipt produced by a harness. Receipts +// are additive: a task that crosses rotations contributes each rotation's +// receipt to every window containing it. +type QuotaReceipt struct { + HarnessID string `json:"harness_id"` + Consumed float64 `json:"consumed"` + At time.Time `json:"at"` +} + +// AggregateQuota sums native receipts in the requested window. It does not +// de-duplicate rotations or use a cumulative session total. +func AggregateQuota(events []domain.Event, from, to time.Time) map[string]float64 { + out := map[string]float64{} + for _, e := range events { + if e.Type != "QuotaReported" || e.At.Before(from) || e.At.After(to) { + continue + } + var r QuotaReceipt + if json.Unmarshal(e.Payload, &r) == nil && r.HarnessID != "" && r.Consumed >= 0 { + out[r.HarnessID] += r.Consumed + } + } + return out +} + func StandupItems(tasks []domain.Task) []StandupItem { out := make([]StandupItem, 0) for _, t := range tasks { @@ -59,15 +85,73 @@ func BuildBrief(events []domain.Event, from, to time.Time, git GitSync) Brief { b.NeedsAttention = append(b.NeedsAttention, e) case "ApprovalRequested": b.NeedsAttention = append(b.NeedsAttention, e) - case "QuotaReported": - if h, ok := p["harness_id"].(string); ok { - if n, ok := p["consumed"].(float64); ok { - b.Quota[h] += n - } + } + } + b.Quota = AggregateQuota(events, from, to) + return b +} + +// GenerateStandupAdvisory creates a persisted, read-only recommendation. +// Applying it is intentionally a separate approval-gated operation. +func GenerateStandupAdvisory(s *store.Store, at time.Time) (domain.Event, error) { + items := StandupItems(s.Tasks()) + p, err := json.Marshal(map[string]any{"items": items, "generated_at": at.UTC()}) + if err != nil { + return domain.Event{}, err + } + e := domain.Event{ID: domain.NewID(), TaskID: "system", Type: "StandupAdvisory", Payload: p, At: at} + return e, s.Append(e) +} + +// ApplyAdvisory applies only approved title recommendations. Unknown or +// malformed recommendations are ignored, while event conflicts are returned. +func ApplyAdvisory(s *store.Store, advisoryID string) ([]domain.Event, error) { + var advisory *domain.Event + approved := false + for _, e := range s.Events(0) { + if e.ID == advisoryID && e.Type == "StandupAdvisory" { + x := e + advisory = &x + } + if e.Type == "ApprovalGranted" { + var p struct { + SubjectRef string `json:"subject_ref"` + } + _ = json.Unmarshal(e.Payload, &p) + if p.SubjectRef == advisoryID { + approved = true } } } - return b + if advisory == nil { + return nil, fmt.Errorf("advisory %q not found", advisoryID) + } + if !approved { + return nil, fmt.Errorf("advisory %q is not approved", advisoryID) + } + var p struct { + Items []StandupItem `json:"items"` + } + if err := json.Unmarshal(advisory.Payload, &p); err != nil { + return nil, err + } + var out []domain.Event + for _, item := range p.Items { + if item.TaskID == "" || item.Title == "" { + continue + } + t, ok := s.Task(item.TaskID) + if !ok || t.Title == item.Title { + continue + } + b, _ := json.Marshal(map[string]any{"title": item.Title, "advisory_ref": advisoryID}) + e := domain.Event{ID: domain.NewID(), TaskID: item.TaskID, Type: "TaskAmended", Version: t.Version + 1, Payload: b} + if err := s.Append(e); err != nil { + return out, err + } + out = append(out, e) + } + return out, nil } // GitState reports local checkout state for the brief. Git failures are visible, diff --git a/internal/operations/operations_test.go b/internal/operations/operations_test.go index 8c4a6bc..538a880 100644 --- a/internal/operations/operations_test.go +++ b/internal/operations/operations_test.go @@ -3,6 +3,7 @@ package operations import ( "encoding/json" "orchestra/internal/domain" + "orchestra/internal/store" "testing" "time" ) @@ -16,3 +17,45 @@ func TestBuildBrief(t *testing.T) { t.Fatalf("unexpected brief: %+v", b) } } + +func TestAggregateQuotaSumsRotationsAndWindows(t *testing.T) { + now := time.Now().UTC() + enc := func(at time.Time, n float64) domain.Event { + p, _ := json.Marshal(map[string]any{"harness_id": "codex", "consumed": n}) + return domain.Event{Type: "QuotaReported", At: at, Payload: p} + } + es := []domain.Event{enc(now.Add(-2*time.Hour), 4), enc(now.Add(-time.Hour), 6), enc(now.Add(-48*time.Hour), 100)} + got := AggregateQuota(es, now.Add(-3*time.Hour), now) + if got["codex"] != 10 { + t.Fatalf("quota=%v, want 10", got) + } +} + +func TestApplyAdvisoryRequiresApproval(t *testing.T) { + s, err := store.Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + p, _ := json.Marshal(map[string]any{"source": "test", "external_id": "1", "project": "p", "title": "old"}) + if err := s.Append(domain.Event{ID: "task", TaskID: "task", Type: "TaskCreated", Version: 1, Payload: p}); err != nil { + t.Fatal(err) + } + ap, _ := json.Marshal(map[string]any{"items": []StandupItem{{TaskID: "task", Title: "new"}}}) + adv := domain.Event{ID: "adv", TaskID: "system", Type: "StandupAdvisory", Payload: ap} + if err := s.Append(adv); err != nil { + t.Fatal(err) + } + if _, err := ApplyAdvisory(s, "adv"); err == nil { + t.Fatal("unapproved advisory applied") + } + grant, _ := json.Marshal(map[string]any{"subject_ref": "adv"}) + if err := s.Append(domain.Event{ID: "grant", TaskID: "system", Type: "ApprovalGranted", Payload: grant}); err != nil { + t.Fatal(err) + } + if _, err := ApplyAdvisory(s, "adv"); err != nil { + t.Fatal(err) + } + if task, _ := s.Task("task"); task.Title != "new" { + t.Fatalf("title=%q", task.Title) + } +} diff --git a/internal/router/router.go b/internal/router/router.go index b3ee7f3..b1ee185 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -17,8 +17,9 @@ type AlwaysAvailable struct{} func (AlwaysAvailable) Available(registry.Herdr) bool { return true } -// QuotaAvailability applies the conservative 80% rule to the most recent -// native quota report in the configured rolling window. +// QuotaAvailability applies the conservative 80% rule to summed native +// receipts in the configured rolling window. Receipts are additive across +// rotations; a cumulative report must not replace earlier rotations. type QuotaAvailability struct { Store *store.Store Limits map[string]float64 @@ -51,8 +52,8 @@ func (q QuotaAvailability) Available(h registry.Herdr) bool { HarnessID string `json:"harness_id"` Consumed float64 `json:"consumed"` } - if json.Unmarshal(e.Payload, &p) == nil && p.HarnessID == h.ID && p.Consumed > consumed { - consumed = p.Consumed + if json.Unmarshal(e.Payload, &p) == nil && p.HarnessID == h.ID && p.Consumed >= 0 { + consumed += p.Consumed } } return consumed < limit*0.8 diff --git a/internal/store/store.go b/internal/store/store.go index 7b13a01..0e06374 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -94,7 +94,7 @@ func (s *Store) apply(e domain.Event) error { return err } t := s.tasks[e.TaskID] - if e.Type == "QuotaReported" || e.Type == "StandupAdvisory" { + if e.Type == "QuotaReported" || e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied" { return nil } switch e.Type { @@ -196,7 +196,7 @@ func (s *Store) Append(e domain.Event) error { return domain.ErrConflict } } - global := e.Type == "QuotaReported" || e.Type == "StandupAdvisory" + global := e.Type == "QuotaReported" || e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied" if !taskExists && e.Type != "TaskCreated" && !global { return domain.ErrNotFound }