aggregator

Aggregator Package

Package aggregator provides snapshot publishing functionality for the commitgraph leaderboard.

Purpose

The aggregator runs every 15 minutes and publishes a full ranked list as a Parquet snapshot to ARMOR. This package handles the metadata construction for the snapshot, including corpus stats, percentile distributions, window information, and histogram invariant assertions.

Integration Points

1. BuildSnapshotMetadata

The core integration point is BuildSnapshotMetadata(ctx, db). This function:

1. Establishes when stats are read: Called after DB connection is available, before snapshot construction
2. Reads the three corpus stat values: Calls pg.ReadCorpusStats(ctx, db) to get totals
3. Makes stats available in scope: Returns metadata struct ready for snapshot construction

2. RunHistogramAssertion

The histogram invariant assertion verifies that for every user, the sum of the 30-element daily_ai_commits histogram array equals the ai_commits_30d scalar field. This catches bugs where the ranking query and histogram query disagree about window boundaries, exclusion filters, or alias merge logic.

Publish Cycle Flow

// 1. Establish database connection
db, err := sql.Open("postgres", dsn)
if err != nil {
    return fmt.Errorf("failed to connect: %w", err)
}
defer db.Close()

ctx := context.Background()

// 2. *** INTEGRATION POINT 1 *** - Read stats into metadata
metadata, err := aggregator.BuildSnapshotMetadata(ctx, db)
if err != nil {
    return fmt.Errorf("failed to build metadata: %w", err)
}

// 3. Stats are now available in scope
fmt.Printf("Total commits: %d\n", metadata.Totals.Commits)
fmt.Printf("Total developers: %d\n", metadata.Totals.Developers)
fmt.Printf("Total repositories: %d\n", metadata.Totals.Repositories)

// 4. *** INTEGRATION POINT 2 *** - Run histogram invariant assertion
assertionResult, err := aggregator.RunHistogramAssertion(ctx, db, aggregator.AssertionOptions{})
if err != nil {
    return fmt.Errorf("failed to run histogram assertion: %w", err)
}

if !assertionResult.Passed {
    // Assertion failed - BLOCK PUBLISH
    return fmt.Errorf("histogram assertion failed: %d violations, blocking publish",
        assertionResult.ViolationCount)
}

// 5. Use metadata when constructing the Parquet snapshot
// (In production: query ranking table, write to Parquet with footer metadata)

Metadata Structure

File-Level Metadata (not per-row)

• WindowStart: Start of 30-day histogram window (current_date - 29 days)
• Totals: Corpus-wide totals from corpus_stats table
  • Commits: Total AI commits in corpus
  • Developers: Total developers with AI commits
  • Repositories: Total repositories with AI commits
• Percentiles: Distribution of ai_commits_30d across all ranked users
  • P50: Median (50th percentile)
  • P75: 75th percentile
  • P90: 90th percentile
  • P95: 95th percentile
  • P99: 99th percentile
• GeneratedAt: When snapshot was created

Acceptance Criteria Met

✅ Stats reader is called at appropriate point: BuildSnapshotMetadata is called after DB is accessible and before snapshot construction

✅ Three stat values available in scope: metadata.Totals.Commits, .Developers, .Repositories are populated and accessible

✅ Integration point is clear and maintainable: Single function call with clear documentation and examples

✅ Histogram assertion runs pre-publish: RunHistogramAssertion is called before snapshot construction to block publishes with data integrity issues

✅ Assertion failure blocks publish: When histogram mismatches are detected, publish is blocked and appropriate error is returned

✅ Bypass mode for emergencies: Assertion can be bypassed with AssertionOptions{Bypass: true, BypassReason: "..."} for emergency situations with audit trail

✅ Failure diagnostics: AssertionResult.FormatViolationLog() provides detailed diagnostic messages including canonical_login, ai_commits_30d, histogram_sum, and difference for each violation

✅ Appropriate error codes: AssertionFailedError is returned for assertion failures (exit code 2), distinct from other errors (exit code 1)

✅ Tested in CI: Unit tests cover assertion logic, bypass mode, violation formatting, and error handling

Usage Example

See example_publish_cycle.go for a complete example of how to integrate this into the publish cycle.

Testing

Run integration tests:

go test ./pkg/aggregator/...

The integration tests verify:

• TestBuildSnapshotMetadata_Integration: Corpus stats reader is called successfully
• TestFormatViolationLog: Violation log formatting with truncation
• TestShouldBlockPublish: Block/publish logic for all assertion states
• TestRunHistogramAssertion_*: Assertion behavior with passing, failing, and bypass modes

Histogram Invariant Assertion

What it checks

The histogram invariant assertion (Invariant 7) verifies that for every published row:

SUM(daily_ai_commits array) == ai_commits_30d

Both values are derived from the same source data (repo_user_daily_tool rollup) over the same 30-day window, so any divergence indicates a bug in:

• Window boundary calculation (off-by-one errors)
• Exclusion filter application (excluded repos leaking in)
• Alias merge logic (same user appearing under multiple logins)
• Array aggregation (missing days or wrong ordering)

Failure modes

When the assertion fails, it returns:

• ViolationCount: Number of users with mismatches
• Violations: Detailed list with canonical_login, ai_commits_30d, histogram_sum, difference
• ShouldBlockPublish(): Returns true to block publish

Example violation log:

Histogram assertion violations (3 users):
  [1] alice@example.com: ai_commits_30d=100, histogram_sum=80, diff=20
  [2] bob@example.com: ai_commits_30d=50, histogram_sum=45, diff=5
  [3] charlie@example.com: ai_commits_30d=30, histogram_sum=35, diff=-5

Bypass mode

For emergencies, the assertion can be bypassed with an audit trail:

result, err := aggregator.RunHistogramAssertion(ctx, db, aggregator.AssertionOptions{
    Bypass:       true,
    BypassReason: "Emergency bypass for data investigation - ticket INC-12345",
})

When bypassed:

• Assertion still runs and records violations
• ShouldBlockPublish() returns false
• Bypass reason is logged for audit trail
• Publish proceeds despite failures

Production usage

In production publish cycle:

1. Run assertion with RunHistogramAssertion(ctx, db, AssertionOptions{})
2. Check assertionResult.ShouldBlockPublish()
3. If true: Block publish, alert operators, log violations, return exit code 2
4. If false: Proceed with publish

Related Documentation

• SQL implementation: migrations/invariant_7_histogram_reconciles.sql
• Test fixtures: pkg/pg/invariant_7_integration_test.go
• Invariant specification: docs/plan/plan.md#L1851-1855
• Bead tracking: cg-3b5y1