Aggregate execution group terminal observations

docs/roadmap/implementation.md

@@ -532,6 +532,8 @@ the root observation-boundary tests.
 
 ## Stage 10 — Implement Terminal-Based Group Aggregation
 
+**Status:** Complete.
+
 **Scope:** internal collector, usage semantics, and concurrent snapshots.
 
 1. Add domain summary/member values and the zero-ready internal group collector

internal/domain/observation.go

@@ -61,6 +61,22 @@ type ObservedTokenUsage struct {
 	CacheWriteTokens ObservedTokenCount
 }
 
+// ExecutionGroupMember identifies an observed run and terminal coverage.
+type ExecutionGroupMember struct {
+	RunID            string
+	TerminalObserved bool
+}
+
+// ExecutionGroupSummary contains terminal-only aggregate execution facts.
+type ExecutionGroupSummary struct {
+	Members                  []ExecutionGroupMember
+	TerminalCoverageComplete bool
+	Attempts                 uint64
+	TransientRetries         uint64
+	RepairCalls              uint64
+	Usage                    ObservedTokenUsage
+}
+
 // ExecutionEvent contains metadata only; it carries no mutable containers or
 // error values. Nonterminal events leave aggregate counters at zero.
 type ExecutionEvent struct {

internal/observation/group.go

@@ -0,0 +1,73 @@
+package observation
+
+import (
+	"sync"
+
+	"go.maximumdirect.net/promptkit/internal/domain"
+)
+
+// Group incrementally collects terminal facts for explicitly observed runs.
+// Its zero value is ready; it must not be copied after first use.
+type Group struct {
+	mu                         sync.Mutex
+	members                    map[string]bool
+	terminals                  int
+	attempts, retries, repairs uint64
+	usage                      domain.ObservedTokenUsage
+}
+
+// Observe enrolls a run and incorporates its first terminal event only.
+func (g *Group) Observe(event domain.ExecutionEvent) {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	if g.members == nil {
+		g.members = make(map[string]bool)
+	}
+	terminal := g.members[event.RunID]
+	if _, exists := g.members[event.RunID]; !exists {
+		g.members[event.RunID] = false
+	}
+	if event.Kind != domain.ExecutionFinished || terminal {
+		return
+	}
+	g.members[event.RunID] = true
+	g.attempts += event.Attempts
+	g.retries += event.TransientRetries
+	g.repairs += event.RepairCalls
+	if g.terminals == 0 {
+		complete := domain.ObservedTokenCount{Complete: true}
+		g.usage = domain.ObservedTokenUsage{PromptTokens: complete, CompletionTokens: complete, TotalTokens: complete, CachedTokens: complete, CacheWriteTokens: complete}
+	}
+	mergeGroupUsage(&g.usage, event.Usage)
+	g.terminals++
+}
+
+// Snapshot returns a detached consistent view of currently observed members.
+func (g *Group) Snapshot() domain.ExecutionGroupSummary {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	result := domain.ExecutionGroupSummary{Members: make([]domain.ExecutionGroupMember, 0, len(g.members)), TerminalCoverageComplete: len(g.members) == g.terminals, Attempts: g.attempts, TransientRetries: g.retries, RepairCalls: g.repairs, Usage: g.usage}
+	for id, terminal := range g.members {
+		result.Members = append(result.Members, domain.ExecutionGroupMember{RunID: id, TerminalObserved: terminal})
+	}
+	if len(g.members) == 0 {
+		zero := domain.ObservedTokenCount{Known: true, Complete: true}
+		result.Usage = domain.ObservedTokenUsage{PromptTokens: zero, CompletionTokens: zero, TotalTokens: zero, CachedTokens: zero, CacheWriteTokens: zero}
+	}
+	return result
+}
+
+func mergeGroupUsage(total *domain.ObservedTokenUsage, next domain.ObservedTokenUsage) {
+	merge := func(a *domain.ObservedTokenCount, b domain.ObservedTokenCount) {
+		if b.Known {
+			a.Value += b.Value
+		}
+		a.Known = a.Known || b.Known
+		a.Complete = a.Complete && b.Complete
+	}
+	merge(&total.PromptTokens, next.PromptTokens)
+	merge(&total.CompletionTokens, next.CompletionTokens)
+	merge(&total.TotalTokens, next.TotalTokens)
+	merge(&total.CachedTokens, next.CachedTokens)
+	merge(&total.CacheWriteTokens, next.CacheWriteTokens)
+}

internal/observation/group_test.go

@@ -0,0 +1,77 @@
+package observation
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+
+	"go.maximumdirect.net/promptkit/internal/domain"
+)
+
+func TestGroupTerminalAccounting(t *testing.T) {
+	var group Group
+	empty := group.Snapshot()
+	if !empty.TerminalCoverageComplete || !empty.Usage.PromptTokens.Known || !empty.Usage.PromptTokens.Complete || empty.Usage.PromptTokens.Value != 0 {
+		t.Fatalf("empty = %#v", empty)
+	}
+	group.Observe(domain.ExecutionEvent{Kind: domain.ExecutionAttempt, RunID: "pending", Attempts: 99})
+	pending := group.Snapshot()
+	if pending.TerminalCoverageComplete || pending.Attempts != 0 || pending.Usage.PromptTokens.Known || pending.Usage.PromptTokens.Complete {
+		t.Fatalf("pending = %#v", pending)
+	}
+	zero := domain.ObservedTokenCount{Known: true, Complete: true}
+	known := domain.ObservedTokenCount{Value: 4, Known: true, Complete: true}
+	partial := domain.ObservedTokenCount{Value: 2, Known: true, Complete: false}
+	group.Observe(domain.ExecutionEvent{Kind: domain.ExecutionFinished, RunID: "first", Outcome: domain.ExecutionOutcomeError, Attempts: 2, TransientRetries: 1, RepairCalls: 1, Usage: domain.ObservedTokenUsage{PromptTokens: known, CompletionTokens: zero, TotalTokens: partial, CachedTokens: zero}})
+	group.Observe(domain.ExecutionEvent{Kind: domain.ExecutionFinished, RunID: "first", Attempts: 100})
+	group.Observe(domain.ExecutionEvent{Kind: domain.ExecutionStarted, RunID: "first"})
+	group.Observe(domain.ExecutionEvent{Kind: domain.ExecutionFinished, RunID: "pending", Attempts: 3, Usage: domain.ObservedTokenUsage{PromptTokens: partial, CompletionTokens: zero, TotalTokens: domain.ObservedTokenCount{}, CachedTokens: zero}})
+	got := group.Snapshot()
+	if !got.TerminalCoverageComplete || len(got.Members) != 2 || got.Attempts != 5 || got.TransientRetries != 1 || got.RepairCalls != 1 {
+		t.Fatalf("summary = %#v", got)
+	}
+	if got.Usage.PromptTokens != (domain.ObservedTokenCount{Value: 6, Known: true}) || got.Usage.CompletionTokens != zero || got.Usage.TotalTokens != partial || got.Usage.CachedTokens != zero || got.Usage.CacheWriteTokens != (domain.ObservedTokenCount{}) {
+		t.Fatalf("usage = %#v", got.Usage)
+	}
+	got.Members[0].RunID = "changed"
+	for _, member := range group.Snapshot().Members {
+		if member.RunID == "changed" || !member.TerminalObserved {
+			t.Fatalf("member = %#v", member)
+		}
+	}
+}
+
+func TestGroupConcurrentSnapshots(t *testing.T) {
+	var group Group
+	var wg sync.WaitGroup
+	const count = 100
+	for i := 0; i < count; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			id := fmt.Sprint(i)
+			group.Observe(domain.ExecutionEvent{RunID: id, Kind: domain.ExecutionStarted})
+			group.Observe(domain.ExecutionEvent{RunID: id, Kind: domain.ExecutionFinished, Attempts: 1})
+		}(i)
+	}
+	for i := 0; i < count; i++ {
+		snapshot := group.Snapshot()
+		seen, terminals := map[string]bool{}, 0
+		for _, member := range snapshot.Members {
+			if seen[member.RunID] {
+				t.Fatal("duplicate member")
+			}
+			seen[member.RunID] = true
+			if member.TerminalObserved {
+				terminals++
+			}
+		}
+		if snapshot.Attempts != uint64(terminals) || snapshot.TerminalCoverageComplete != (terminals == len(snapshot.Members)) {
+			t.Fatalf("inconsistent snapshot = %#v", snapshot)
+		}
+	}
+	wg.Wait()
+	if final := group.Snapshot(); len(final.Members) != count || final.Attempts != count || !final.TerminalCoverageComplete {
+		t.Fatalf("final = %#v", final)
+	}
+}