Agent-System/internal/engineeringdashboard/collector.go

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
}