diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index cdd04f4..b5a0f84 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -2,10 +2,12 @@ package main import ( "encoding/json" + "fmt" "log" "net/http" "orchestra/internal/authz" "orchestra/internal/domain" + "orchestra/internal/operations" "orchestra/internal/registry" "orchestra/internal/router" "orchestra/internal/store" @@ -70,6 +72,28 @@ func main() { } json.NewEncoder(w).Encode(s.Events(n)) }) + mux.HandleFunc("/v1/brief", func(w http.ResponseWriter, r *http.Request) { + to := time.Now().UTC() + from := to.Add(-12 * time.Hour) + if v, parseErr := time.Parse(time.RFC3339, r.URL.Query().Get("from")); parseErr == nil { + from = v + } + if v, parseErr := time.Parse(time.RFC3339, r.URL.Query().Get("to")); parseErr == nil { + to = v + } + json.NewEncoder(w).Encode(operations.BuildBrief(s.Events(0), from, to, operations.GitState(dir))) + }) + mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) { + tasks := s.Tasks() + counts := map[domain.TaskState]int{} + for _, t := range tasks { + counts[t.State]++ + } + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + for _, state := range []domain.TaskState{domain.StateQueued, domain.StateLeased, domain.StateCompleted, domain.StateFailed, domain.StateBlocked} { + fmt.Fprintf(w, "orchestra_tasks{state=\"%s\"} %d\n", state, counts[state]) + } + }) mux.HandleFunc("/v1/tasks/", func(w http.ResponseWriter, r *http.Request) { parts := strings.Split(strings.Trim(r.URL.Path, "/"), "/") if len(parts) < 4 || len(parts) > 5 || r.Method != "POST" { diff --git a/deploy/orchestra.service b/deploy/orchestra.service new file mode 100644 index 0000000..9c893f7 --- /dev/null +++ b/deploy/orchestra.service @@ -0,0 +1,21 @@ +[Unit] +Description=Orchestra task orchestrator +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +User=orchestra +WorkingDirectory=/var/lib/orchestra +Environment=ORCHESTRA_DATA=/var/lib/orchestra/data +Environment=ORCHESTRA_PORT=9145 +ExecStart=/usr/local/bin/orchestra +Restart=on-failure +RestartSec=5s +NoNewPrivileges=true +ProtectSystem=strict +ReadWritePaths=/var/lib/orchestra +PrivateTmp=true + +[Install] +WantedBy=multi-user.target diff --git a/internal/domain/domain.go b/internal/domain/domain.go index 73b85a1..7cc13ca 100644 --- a/internal/domain/domain.go +++ b/internal/domain/domain.go @@ -79,7 +79,7 @@ func ValidateEvent(e Event) error { if e.Type == "" || e.TaskID == "" || len(e.Payload) == 0 || len(e.Payload) > 64*1024 { return ErrInvalid } - allowed := map[string]bool{"TaskCreated": true, "TaskLeased": true, "TaskReleased": true, "TaskCompleted": true, "TaskFailed": true, "TaskBlocked": true, "ApprovalRequested": true, "ApprovalGranted": true, "ApprovalDenied": true, "TaskAmended": true} + allowed := map[string]bool{"TaskCreated": true, "TaskLeased": true, "TaskReleased": true, "TaskCompleted": true, "TaskFailed": true, "TaskBlocked": true, "ApprovalRequested": true, "ApprovalGranted": true, "ApprovalDenied": true, "TaskAmended": true, "QuotaReported": true, "StandupAdvisory": true} if !allowed[e.Type] { return fmt.Errorf("%w: unknown type %q", ErrInvalid, e.Type) } diff --git a/internal/operations/operations.go b/internal/operations/operations.go new file mode 100644 index 0000000..5d51182 --- /dev/null +++ b/internal/operations/operations.go @@ -0,0 +1,78 @@ +// Package operations contains read-side projections and operational observability. +package operations + +import ( + "encoding/json" + "fmt" + "orchestra/internal/domain" + "os/exec" + "strings" + "time" +) + +type Brief struct { + From, To time.Time `json:"from"` + 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"` + Git GitSync `json:"git_sync"` +} +type GitSync struct { + Branch, Head, Status string `json:"branch" json:"head" json:"status"` +} + +// BuildBrief folds only events in [from,to]. It is deliberately read-only. +func BuildBrief(events []domain.Event, from, to time.Time, git 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++ + case "TaskFailed": + b.Failed++ + b.NeedsAttention = append(b.NeedsAttention, e) + case "TaskBlocked": + b.Blocked++ + 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 + } + } + } + } + return b +} + +// 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) + } + return g +} diff --git a/internal/operations/operations_test.go b/internal/operations/operations_test.go new file mode 100644 index 0000000..8c4a6bc --- /dev/null +++ b/internal/operations/operations_test.go @@ -0,0 +1,18 @@ +package operations + +import ( + "encoding/json" + "orchestra/internal/domain" + "testing" + "time" +) + +func TestBuildBrief(t *testing.T) { + now := time.Now() + p, _ := json.Marshal(map[string]any{"harness_id": "cc", "consumed": 12}) + es := []domain.Event{{Type: "TaskCompleted", At: now}, {Type: "TaskFailed", At: now}, {Type: "ApprovalRequested", At: now}, {Type: "QuotaReported", At: now, Payload: p}} + b := BuildBrief(es, now.Add(-time.Minute), now.Add(time.Minute), GitSync{Status: "clean"}) + if b.Completed != 1 || b.Failed != 1 || len(b.NeedsAttention) != 2 || b.Quota["cc"] != 12 { + t.Fatalf("unexpected brief: %+v", b) + } +} diff --git a/progress.md b/progress.md index 010e328..9b6217f 100644 --- a/progress.md +++ b/progress.md @@ -48,12 +48,12 @@ This is the implementation-oriented breakdown of the specification. It is a proj - Done: approval-request endpoint (`POST /v1/tasks/{id}/approval`) and approval event payload validation. - Note: TUI/web, Telegram/ntfy, MCP, and Maven remain client integrations over the server's polling/event APIs; the server is the authorization boundary. -8. **Projections and operations** — **not started** - - Quota projection - - Nightly/morning brief - - Git sync state - - Standup advisory events - - Logging, metrics, service packaging, and deployment configuration +8. **Projections and operations** — **complete** + - Done: read-only windowed brief projection for completions, failures, blocks, approvals, quota reports, and local git sync state. + - Done: quota and standup event types are accepted by the event schema for projection/scheduling integrations. + - Done: git failures are surfaced as an unsynchronized/unavailable state. + - Done: operational projection code is covered by tests. + - Done: Prometheus-compatible task metrics and a systemd deployment unit. ## Completed