Add quota receipt aggregation and approved standup advisories

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)

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,

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)
+	}
+}

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

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
 	}