orchestra/internal/operations/operations.go

// Package operations contains read-side projections and operational observability.
package operations

import (
	"encoding/json"
	"fmt"
	"orchestra/internal/authz"
	"orchestra/internal/domain"
	"orchestra/internal/store"
	"os/exec"
	"strings"
	"time"
)

type Brief struct {
	From           time.Time           `json:"from"`
	To             time.Time           `json:"to"`
	Completed      int                 `json:"completed"`
	Failed         int                 `json:"failed"`
	Blocked        int                 `json:"blocked"`
	NeedsAttention []domain.Event      `json:"needs_attention"`
	Quota          map[string]float64  `json:"quota_consumed"`
	Receipts       []CompletionReceipt `json:"receipts"`
	// Git is keyed by project ID (spec §7.4: "what pushed, what's on which
	// branch, what workpc still needs to pull" is a per-project question,
	// not a single global answer read off the event-log directory).
	Git map[string]GitSync `json:"git_sync"`
}
type GitSync struct {
	Branch string `json:"branch"`
	Head   string `json:"head"`
	Status string `json:"status"`
	// Ahead/Behind are counts vs the branch's upstream, when one is
	// configured — "what pushed" and "what workpc still needs to pull"
	// (§7.4). Zero when there is no upstream (e.g. detached HEAD).
	Ahead  int `json:"ahead,omitempty"`
	Behind int `json:"behind,omitempty"`
}

// CompletionReceipt names the proof a completion carries (§7.4: "what
// pushed... and the receipts for every completion"), pulled out of
// TaskCompleted's payload rather than just counted.
type CompletionReceipt struct {
	TaskID    string         `json:"task_id"`
	ReportRef string         `json:"report_ref"`
	Receipt   map[string]any `json:"receipt"`
}

type StandupItem struct {
	TaskID string           `json:"task_id"`
	State  domain.TaskState `json:"state"`
	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"`
}

// QuotaWindows publishes the two routing windows from the same additive
// per-lease receipts used by the scheduler.
type QuotaWindows struct {
	FiveHour map[string]float64 `json:"five_hour"`
	Weekly   map[string]float64 `json:"weekly"`
}

func ProjectQuotaWindows(events []domain.Event, now time.Time) QuotaWindows {
	return QuotaWindows{FiveHour: AggregateQuota(events, now.Add(-5*time.Hour), now), Weekly: AggregateQuota(events, now.Add(-7*24*time.Hour), now)}
}

// 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 {
		if t.State == domain.StateQueued || t.State == domain.StateLeased || t.State == domain.StateNeedsAttention || t.State == domain.StateBlocked {
			out = append(out, StandupItem{t.ID, t.State, t.Title})
		}
	}
	return out
}

// BuildBrief folds only events in [from,to]. It is deliberately read-only.
func BuildBrief(events []domain.Event, from, to time.Time, git map[string]GitSync) Brief {
	b := Brief{From: from, To: to, Quota: map[string]float64{}, Git: git}
	for _, e := range events {
		if e.At.Before(from) || e.At.After(to) {
			continue
		}
		var p map[string]any
		_ = json.Unmarshal(e.Payload, &p)
		switch e.Type {
		case "TaskCompleted":
			b.Completed++
			receipt, _ := p["receipt"].(map[string]any)
			reportRef, _ := p["report_ref"].(string)
			b.Receipts = append(b.Receipts, CompletionReceipt{TaskID: e.TaskID, ReportRef: reportRef, Receipt: receipt})
		case "TaskFailed":
			b.Failed++
			b.NeedsAttention = append(b.NeedsAttention, e)
		case "TaskBlocked":
			b.Blocked++
			b.NeedsAttention = append(b.NeedsAttention, e)
		case "TaskNeedsAttention":
			b.NeedsAttention = append(b.NeedsAttention, e)
		case "ApprovalRequested":
			b.NeedsAttention = append(b.NeedsAttention, e)
		}
	}
	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, Surface: string(authz.System)}
	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
			}
		}
	}
	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, Surface: string(authz.System)}
		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,
// never silently interpreted as synchronized.
func GitState(dir string) GitSync {
	g := GitSync{Status: "unavailable"}
	run := func(args ...string) string {
		out, err := exec.Command("git", append([]string{"-C", dir}, args...)...).Output()
		if err != nil {
			return ""
		}
		return strings.TrimSpace(string(out))
	}
	g.Branch, g.Head = run("branch", "--show-current"), run("rev-parse", "HEAD")
	if s := run("status", "--porcelain"); s != "" {
		g.Status = "dirty"
	} else if g.Head != "" {
		g.Status = "clean"
	}
	if g.Head == "" {
		g.Status = fmt.Sprintf("git unavailable (%s)", dir)
	}
	if counts := run("rev-list", "--left-right", "--count", "@{u}...HEAD"); counts != "" {
		var behind, ahead int
		if n, _ := fmt.Sscanf(counts, "%d\t%d", &behind, &ahead); n == 2 {
			g.Behind, g.Ahead = behind, ahead
		}
	}
	return g
}