From 79ee72267c240570bd36754754f7576a806b0e05 Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 17 Sep 2026 07:04:46 +0000 Subject: [PATCH] feat(engineering-dashboard): U5 reconcile and snapshot publisher Merge docs/git/runtime facts into an immutable snapshot with conflict retention, last-known stale handling, and acceptance-sample coverage. Co-authored-by: Cursor --- internal/engineeringdashboard/collector.go | 257 ++++++++++++++++++ .../engineeringdashboard/reconcile_test.go | 171 ++++++++++++ 2 files changed, 428 insertions(+) create mode 100644 internal/engineeringdashboard/collector.go create mode 100644 internal/engineeringdashboard/reconcile_test.go diff --git a/internal/engineeringdashboard/collector.go b/internal/engineeringdashboard/collector.go new file mode 100644 index 00000000..2be3051a --- /dev/null +++ b/internal/engineeringdashboard/collector.go @@ -0,0 +1,257 @@ +package engineeringdashboard + +import ( + "context" + "fmt" + "sort" + "sync" + "sync/atomic" + "time" +) + +// Collector coordinates parallel source collection and atomic snapshot publish. +type Collector struct { + RepoRoot string + Runner CommandRunner + TTL time.Duration + + mu sync.Mutex + current *Snapshot + lastKnown map[string]SourceState + gen atomic.Uint64 + collecting atomic.Bool + singleflight chan struct{} +} + +// NewCollector builds a coordinator for one repo root. +func NewCollector(repoRoot string, runner CommandRunner) *Collector { + if runner == nil { + runner = ExecRunner{} + } + return &Collector{ + RepoRoot: repoRoot, + Runner: runner, + TTL: 3 * time.Second, + lastKnown: map[string]SourceState{}, + singleflight: make(chan struct{}, 1), + } +} + +// Current returns the latest published snapshot (may be nil on cold start). +func (c *Collector) Current() *Snapshot { + c.mu.Lock() + defer c.mu.Unlock() + return c.current +} + +// Refresh builds and publishes a new snapshot. Concurrent callers share one flight. +func (c *Collector) Refresh(ctx context.Context) (*Snapshot, error) { + select { + case c.singleflight <- struct{}{}: + defer func() { <-c.singleflight }() + default: + // another collection in flight — return current + if s := c.Current(); s != nil { + return s, nil + } + // wait briefly for first + select { + case c.singleflight <- struct{}{}: + defer func() { <-c.singleflight }() + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return c.collectAndPublish(ctx) +} + +func (c *Collector) collectAndPublish(ctx context.Context) (*Snapshot, error) { + c.collecting.Store(true) + defer c.collecting.Store(false) + + start := time.Now() + now := start.UTC() + deadline, cancel := context.WithTimeout(ctx, 6*time.Second) + defer cancel() + + var docs DocsCollectResult + var git GitCollectResult + var runtime RuntimeCollectResult + var wg sync.WaitGroup + wg.Add(3) + go func() { + defer wg.Done() + docs = CollectDocs(c.RepoRoot, now) + }() + go func() { + defer wg.Done() + git = CollectGit(deadline, c.RepoRoot, c.Runner, now) + }() + go func() { + defer wg.Done() + runtime = CollectRuntime(deadline, c.RepoRoot, c.Runner, now) + }() + wg.Wait() + + docs.Source = c.applyLastKnown(docs.Source, now) + git.Source = c.applyLastKnown(git.Source, now) + runtime.Source = c.applyLastKnown(runtime.Source, now) + + snap := Reconcile(ReconcileInput{ + RepoRoot: c.RepoRoot, + Now: now, + Duration: time.Since(start), + Docs: docs, + Git: git, + Runtime: runtime, + Generation: c.gen.Add(1), + }) + c.mu.Lock() + c.current = &snap + c.mu.Unlock() + return &snap, nil +} + +func (c *Collector) applyLastKnown(src SourceState, now time.Time) SourceState { + c.mu.Lock() + defer c.mu.Unlock() + if src.Status == SourceOK || src.Status == SourceDegraded { + cp := src + cp.LastSuccessAt = &now + c.lastKnown[src.Name] = cp + return cp + } + prev, ok := c.lastKnown[src.Name] + if !ok { + src.Freshness = FreshnessError + if src.Status == "" { + src.Status = SourceError + } + return src + } + // retain last-known markers without rewriting observed_at of cached facts + src.Status = SourceDegraded + src.Freshness = FreshnessStale + src.LastSuccessAt = prev.LastSuccessAt + if src.Error == "" { + src.Error = "using last-known; current collection failed" + } + return src +} + +// ReconcileInput feeds the pure reconcile function. +type ReconcileInput struct { + RepoRoot string + Now time.Time + Duration time.Duration + Docs DocsCollectResult + Git GitCollectResult + Runtime RuntimeCollectResult + Generation uint64 +} + +// Reconcile merges source facts into an immutable snapshot without picking winners on conflicts. +func Reconcile(in ReconcileInput) Snapshot { + repo := in.Git.Repository + if repo.Root == "" { + repo.Root = in.RepoRoot + } + + tasks := append([]TaskState(nil), in.Docs.Tasks...) + gitByName := map[string]GitBranchView{} + for _, g := range in.Git.Branches { + gitByName[g.Name] = g + } + + var conflicts []Conflict + var facts []Fact + facts = append(facts, in.Docs.Facts...) + facts = append(facts, in.Git.Facts...) + facts = append(facts, in.Runtime.Facts...) + + for i := range tasks { + t := &tasks[i] + declaredBranch := t.Claim.Branch + if declaredBranch != "" { + if g, ok := gitByName[declaredBranch]; ok { + t.Git.Branch = g.Name + t.Git.Worktree = g.Worktree + t.Git.Ahead = g.Ahead + t.Git.Behind = g.Behind + t.Git.RemoteNote = g.RemoteNote + if g.Dirty { + t.Git.Cleanliness = "dirty" + } else { + t.Git.Cleanliness = "clean" + } + if t.Claim.Worktree != "" && g.Worktree != "" && t.Claim.Worktree != g.Worktree && t.Claim.Worktree != "-" { + f1 := FactID(t.ID, "worktree", "docs") + f2 := FactID(t.ID, "worktree", "git") + facts = append(facts, + Fact{ID: f1, Subject: t.ID, Field: "worktree", Value: t.Claim.Worktree, Provenance: ProvenanceDeclared, SourceRef: t.Path, ObservedAt: in.Now}, + Fact{ID: f2, Subject: t.ID, Field: "worktree", Value: g.Worktree, Provenance: ProvenanceObserved, SourceRef: "git", ObservedAt: in.Now}, + ) + conflicts = append(conflicts, NewConflict(t.ID, "worktree", ConflictTaskVsGit, []string{f1, f2}, "declared worktree differs from observed")) + } + } else if t.DocStatus == DocStatusInProgress || t.Claim.Claimed { + f1 := FactID(t.ID, "branch", "docs") + facts = append(facts, Fact{ID: f1, Subject: t.ID, Field: "branch", Value: declaredBranch, Provenance: ProvenanceDeclared, SourceRef: t.Path, ObservedAt: in.Now}) + conflicts = append(conflicts, NewConflict(t.ID, "branch", ConflictMissingEvidence, []string{f1}, "declared branch not observed locally")) + } + } + // claimed but no runtime activity source + if t.Claim.Claimed && t.Claim.Activity == "" { + t.Claim.Activity = "unknown" + } + // never promote doc done into deploy + if t.DocStatus == DocStatusDone { + for env, d := range t.Deployment { + if d.AppReadiness == "" { + d.AppReadiness = "unknown" + t.Deployment[env] = d + } + } + } + } + + envs := append([]EnvironmentState(nil), in.Runtime.Environments...) + for i := range envs { + e := &envs[i] + if e.Facets.AppReadiness == "ok" && e.Facets.ContainerStatus == "partial" { + f1 := FactID("env:"+e.Name, "readiness", "ready") + f2 := FactID("env:"+e.Name, "container_status", "docker") + conflicts = append(conflicts, NewConflict("env:"+e.Name, "health", ConflictDeclaredVsObserved, []string{f1, f2}, + "app /ready ok while containers partial — no aggregate green")) + } + if e.Receipt != "" && e.Facets.ArtifactID != "" { + // receipt present; identity note already set by runtime collector when mismatch unknown + } + } + + sources := []SourceState{in.Docs.Source, in.Git.Source, in.Runtime.Source} + sort.Slice(sources, func(i, j int) bool { return sources[i].Name < sources[j].Name }) + + var warnings []string + warnings = append(warnings, in.Docs.Warnings...) + warnings = append(warnings, in.Git.Warnings...) + warnings = append(warnings, in.Runtime.Warnings...) + + snap := Snapshot{ + SchemaVersion: SchemaVersion, + GenerationID: fmt.Sprintf("gen_%d", in.Generation), + GeneratedAt: in.Now, + CollectionDuration: in.Duration, + Repository: repo, + Sources: sources, + Facts: facts, + Conflicts: conflicts, + Documents: in.Docs.Documents, + Modules: in.Docs.Modules, + Tasks: tasks, + Git: in.Git.Branches, + Environments: envs, + Warnings: warnings, + } + snap.RevisionID = ComputeRevisionID(snap) + return snap +} diff --git a/internal/engineeringdashboard/reconcile_test.go b/internal/engineeringdashboard/reconcile_test.go new file mode 100644 index 00000000..a110317d --- /dev/null +++ b/internal/engineeringdashboard/reconcile_test.go @@ -0,0 +1,171 @@ +package engineeringdashboard + +import ( + "strings" + "testing" + "time" +) + +func TestReconcile_RDevPlannedUnclaimed(t *testing.T) { + task := SampleRDevTaskFixture() + in := ReconcileInput{ + RepoRoot: "/repo", + Now: time.Date(2026, 9, 17, 8, 0, 0, 0, time.UTC), + Generation: 1, + Docs: DocsCollectResult{ + Source: SourceState{Name: "docs", Status: SourceOK, Freshness: FreshnessFresh}, + Tasks: []TaskState{task}, + Documents: []DocumentNode{{ + ID: task.Path, Path: task.Path, Title: task.Title, Status: "planned", + }}, + }, + Git: GitCollectResult{ + Source: SourceState{Name: "git", Status: SourceOK, Freshness: FreshnessFresh}, + Repository: RepositoryIdentity{Root: "/repo", Branch: "dev", Head: "h1"}, + }, + Runtime: RuntimeCollectResult{ + Source: SourceState{Name: "runtime", Status: SourceOK, Freshness: FreshnessFresh}, + }, + } + snap := Reconcile(in) + if len(snap.Tasks) != 1 { + t.Fatal(snap.Tasks) + } + got := snap.Tasks[0] + if got.DocStatus != DocStatusPlanned || got.Claim.Claimed { + t.Fatalf("expected planned/unclaimed: %+v", got) + } + if got.Git.Cleanliness != "unknown" { + t.Fatalf("must not invent git implementation: %+v", got.Git) + } + for _, d := range got.Deployment { + if d.AppReadiness != "unknown" { + t.Fatalf("must not invent deploy: %+v", d) + } + } +} + +func TestReconcile_DevAheadSample(t *testing.T) { + g := SampleDevAheadGitFixture() + in := ReconcileInput{ + RepoRoot: "/repo", + Now: time.Now().UTC(), + Generation: 2, + Git: GitCollectResult{ + Source: SourceState{Name: "git", Status: SourceOK, Freshness: FreshnessFresh}, + Repository: RepositoryIdentity{Root: "/repo", Branch: "dev", Head: g.Head}, + Branches: []GitBranchView{g}, + }, + Docs: DocsCollectResult{Source: SourceState{Name: "docs", Status: SourceOK, Freshness: FreshnessFresh}}, + Runtime: RuntimeCollectResult{Source: SourceState{Name: "runtime", Status: SourceOK, Freshness: FreshnessFresh}}, + } + snap := Reconcile(in) + if len(snap.Git) != 1 || snap.Git[0].Ahead != 3 { + t.Fatalf("%+v", snap.Git) + } + if !strings.Contains(snap.Git[0].RemoteNote, "unconfirmed") { + t.Fatal(snap.Git[0].RemoteNote) + } +} + +func TestReconcile_MainEventsForkConflict(t *testing.T) { + env := SampleMainEventsForkFixture() + in := ReconcileInput{ + RepoRoot: "/repo", + Now: time.Now().UTC(), + Generation: 3, + Docs: DocsCollectResult{Source: SourceState{Name: "docs", Status: SourceOK, Freshness: FreshnessFresh}}, + Git: GitCollectResult{Source: SourceState{Name: "git", Status: SourceOK, Freshness: FreshnessFresh}, Repository: RepositoryIdentity{Root: "/repo"}}, + Runtime: RuntimeCollectResult{ + Source: SourceState{Name: "runtime", Status: SourceOK, Freshness: FreshnessFresh}, + Environments: []EnvironmentState{env}, + Facts: []Fact{ + {ID: FactID("env:main", "readiness", "ready"), Subject: "env:main", Field: "readiness", Value: "ok", Provenance: ProvenanceObserved}, + {ID: FactID("env:main", "container_status", "docker"), Subject: "env:main", Field: "container_status", Value: "partial", Provenance: ProvenanceObserved}, + }, + }, + } + snap := Reconcile(in) + found := false + for _, c := range snap.Conflicts { + if c.Subject == "env:main" && c.Kind == ConflictDeclaredVsObserved { + found = true + if len(c.FactRefs) < 2 { + t.Fatalf("conflict must retain both refs: %+v", c) + } + } + } + if !found { + t.Fatalf("expected health conflict, got %+v", snap.Conflicts) + } + for _, e := range snap.Environments { + if e.Name == "main" && e.Facets.ServicesOK != "partial" { + t.Fatalf("no aggregate green: %+v", e.Facets) + } + } +} + +func TestReconcile_StableAcrossInputOrder(t *testing.T) { + now := time.Date(2026, 9, 17, 9, 0, 0, 0, time.UTC) + mk := func(facts []Fact) Snapshot { + return Reconcile(ReconcileInput{ + RepoRoot: "/repo", Now: now, Generation: 9, + Docs: DocsCollectResult{ + Source: SourceState{Name: "docs", Status: SourceOK, Freshness: FreshnessFresh}, + Facts: facts, + }, + Git: GitCollectResult{Source: SourceState{Name: "git", Status: SourceOK, Freshness: FreshnessFresh}, Repository: RepositoryIdentity{Root: "/repo", Branch: "dev", Head: "x"}}, + Runtime: RuntimeCollectResult{Source: SourceState{Name: "runtime", Status: SourceOK, Freshness: FreshnessFresh}}, + }) + } + f1 := Fact{ID: "a", Subject: "s", Field: "f", Value: "1", Provenance: ProvenanceDeclared, SourceRef: "d", ObservedAt: now} + f2 := Fact{ID: "b", Subject: "s", Field: "f", Value: "2", Provenance: ProvenanceObserved, SourceRef: "g", ObservedAt: now} + a := mk([]Fact{f1, f2}) + b := mk([]Fact{f2, f1}) + if a.RevisionID != b.RevisionID { + t.Fatalf("revision unstable: %s vs %s", a.RevisionID, b.RevisionID) + } +} + +func TestReconcile_DocDoneDoesNotFillDeploy(t *testing.T) { + task := SampleRDevTaskFixture() + task.DocStatus = DocStatusDone + snap := Reconcile(ReconcileInput{ + RepoRoot: "/repo", Now: time.Now().UTC(), Generation: 4, + Docs: DocsCollectResult{Source: SourceState{Name: "docs", Status: SourceOK, Freshness: FreshnessFresh}, Tasks: []TaskState{task}}, + Git: GitCollectResult{Source: SourceState{Name: "git", Status: SourceOK, Freshness: FreshnessFresh}, Repository: RepositoryIdentity{Root: "/repo"}}, + Runtime: RuntimeCollectResult{Source: SourceState{Name: "runtime", Status: SourceOK, Freshness: FreshnessFresh}}, + }) + for _, d := range snap.Tasks[0].Deployment { + if d.AppReadiness != "unknown" { + t.Fatalf("doc done must not imply deploy: %+v", d) + } + } +} + +func TestCollector_LastKnownStaleOnFailure(t *testing.T) { + c := NewCollector("/repo", &FakeRunner{Map: map[string]FakeResult{}}) + ok := SourceState{Name: "git", Status: SourceOK, Freshness: FreshnessFresh} + now := time.Now().UTC() + got := c.applyLastKnown(ok, now) + if got.LastSuccessAt == nil { + t.Fatal("expected last success") + } + fail := SourceState{Name: "git", Status: SourceError, Freshness: FreshnessError, Error: "boom"} + stale := c.applyLastKnown(fail, now.Add(time.Second)) + if stale.Freshness != FreshnessStale || stale.Status != SourceDegraded { + t.Fatalf("%+v", stale) + } + if stale.LastSuccessAt == nil || !stale.LastSuccessAt.Equal(now) { + t.Fatalf("must keep prior last-success: %+v", stale) + } +} + +func TestCollector_ColdStartFailureNoFakeHistory(t *testing.T) { + c := NewCollector("/repo", &FakeRunner{}) + fail := SourceState{Name: "git", Status: SourceError, Error: "missing"} + got := c.applyLastKnown(fail, time.Now().UTC()) + if got.Freshness != FreshnessError || got.LastSuccessAt != nil { + t.Fatalf("cold start must not invent last-known: %+v", got) + } +}