diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index bed5d00..66f926f 100644 --- a/docs/roadmap/implementation.md +++ b/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 diff --git a/internal/domain/observation.go b/internal/domain/observation.go index 3f21926..b722505 100644 --- a/internal/domain/observation.go +++ b/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 { diff --git a/internal/observation/group.go b/internal/observation/group.go new file mode 100644 index 0000000..def85a6 --- /dev/null +++ b/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) +} diff --git a/internal/observation/group_test.go b/internal/observation/group_test.go new file mode 100644 index 0000000..e5e58d7 --- /dev/null +++ b/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) + } +}