Merge pull request #50 from trick77/feat/noergler-pr-rollup

feat(noergler): per-PR rollup event with merged/declined/deleted outcome

Author: trick77

README.md

@@ -81,7 +81,7 @@ the data captured in v1.
 | **Untracked-work rate** | `COUNT(*) WHERE jira_keys = '{}'` over merged PRs — process-compliance signal. |
 | **Per-ticket flow** | `WHERE 'ABC-1234' = ANY(jira_keys)` returns every event for a ticket across Bitbucket / pipeline / Argo (joined via commit_sha). |
 | **Human vs automated split** | `WHERE NOT is_automated` (Renovate / Dependabot / Snyk / Mend / generic-bot detection runs at write time and tags `automation_source`). Default dashboards exclude bots; bot velocity is a separate CI-health view. |
-| **AI reviewer precision** *(noergler)* | `1 - count(noergler_events WHERE event_type='feedback' AND verdict='disagreed') / count(noergler_events WHERE event_type='completed')` per repo × week. Higher = the AI review is more useful. |
+| **AI reviewer precision** *(noergler)* | `1 - count(noergler_events WHERE event_type='feedback' AND verdict='disagreed') / count(noergler_events WHERE event_type='pr_completed')` per repo × week. Higher = the AI review is more useful. Filter on `outcome='merged'` to score precision only on PRs that shipped. |
 
 ### FinOps signals
 
@@ -92,7 +92,9 @@ events arrive pre-priced in USD.
 
 | Signal | How it's computed |
 |---|---|
-| **LLM review spend per model / team** *(noergler)* | `SUM(noergler_events.cost_usd), SUM(prompt_tokens + completion_tokens) GROUP BY model, team` over `event_type = 'completed'`. Pre-priced — no multiplier needed. |
+| **LLM review spend per PR / team** *(noergler)* | `SUM(cost_usd), SUM(prompt_tokens + completion_tokens) GROUP BY team` over `event_type = 'pr_completed'`. Each row is a per-PR rollup (one event per merged / declined / deleted PR). Filter on `outcome='merged'` for "spend that actually shipped"; keep all outcomes for total LLM-review spend including abandoned PRs. Pre-priced — no multiplier needed. |
+| **LLM review cost per KLOC** *(noergler)* | `SUM(cost_usd) / NULLIF(SUM(lines_added + lines_removed), 0) * 1000 GROUP BY team, outcome` over `noergler_events WHERE event_type='pr_completed'`. Diff-size normalised cost — fair comparison across small fixes and large refactors. |
+| **Wasted LLM review** *(noergler)* | `SUM(cost_usd) FROM noergler_events WHERE event_type='pr_completed' AND outcome IN ('declined','deleted')`. Review effort spent on code that never shipped. |
 | **CI compute time per pipeline / team** | `SUM(pipeline_events.duration_seconds) GROUP BY pipeline_name, team`. The unit metric for CI cost attribution. |
 | **Wasted CI** | `SUM(duration_seconds) WHERE status IN ('FAILURE','Failed')` — failed builds × time. Quantifies the cost of flakes / broken tests. |
 | **Bot-driven pipeline churn** | `pipeline_events` joined to `bitbucket_events` via `commit_sha` filtered on `is_automated = true`. Renovate / Dependabot can drive 40–70% of pipeline runs in many orgs; useful input for batching policies. |

migrations/versions/0002_noergler_pr_rollup.py

@@ -0,0 +1,70 @@
+"""noergler events: switch from per-run 'completed' to per-PR 'pr_completed' rollup
+
+Adds outcome / lines_added / lines_removed / files_changed / total_runs /
+merge_commit_sha / models_used / first_review_at columns to noergler_events.
+The pre-existing prompt_tokens / completion_tokens / elapsed_ms / findings_count /
+cost_usd columns are reused as aggregated totals — no rename, just a
+semantic shift.
+
+Revision ID: 0002
+Revises: 0001
+Create Date: 2026-05-17
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects.postgresql import ARRAY
+
+revision: str = "0002"
+down_revision: str | None = "0001"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column("noergler_events", sa.Column("outcome", sa.String, nullable=True))
+    op.add_column("noergler_events", sa.Column("merge_commit_sha", sa.String, nullable=True))
+    op.add_column("noergler_events", sa.Column("lines_added", sa.Integer, nullable=True))
+    op.add_column("noergler_events", sa.Column("lines_removed", sa.Integer, nullable=True))
+    op.add_column("noergler_events", sa.Column("files_changed", sa.Integer, nullable=True))
+    op.add_column("noergler_events", sa.Column("total_runs", sa.Integer, nullable=True))
+    op.add_column(
+        "noergler_events",
+        sa.Column("models_used", ARRAY(sa.String), nullable=True),
+    )
+    op.add_column(
+        "noergler_events",
+        sa.Column("first_review_at", sa.DateTime(timezone=True), nullable=True),
+    )
+
+    op.create_index("ix_noergler_events_outcome", "noergler_events", ["outcome"])
+    # The 'model' (singular) index made sense when each row was a single
+    # review run. Per-PR rollups carry models_used (an array) instead, so
+    # the singular-model index no longer matches the query patterns.
+    op.drop_index("ix_noergler_events_model", table_name="noergler_events")
+
+    # run_id was the per-run idempotency key for the old 'completed' event.
+    # The rollup uses (pr_key, outcome) — run_id is dead. 'model' (singular)
+    # is superseded by models_used; the original value is still recoverable
+    # from the payload JSONB on historical rows if anyone ever needs it.
+    op.drop_column("noergler_events", "run_id")
+    op.drop_column("noergler_events", "model")
+
+
+def downgrade() -> None:
+    op.add_column("noergler_events", sa.Column("model", sa.String, nullable=True))
+    op.add_column("noergler_events", sa.Column("run_id", sa.String, nullable=True))
+    op.create_index("ix_noergler_events_model", "noergler_events", ["model"])
+    op.drop_index("ix_noergler_events_outcome", table_name="noergler_events")
+
+    op.drop_column("noergler_events", "first_review_at")
+    op.drop_column("noergler_events", "models_used")
+    op.drop_column("noergler_events", "total_runs")
+    op.drop_column("noergler_events", "files_changed")
+    op.drop_column("noergler_events", "lines_removed")
+    op.drop_column("noergler_events", "lines_added")
+    op.drop_column("noergler_events", "merge_commit_sha")
+    op.drop_column("noergler_events", "outcome")

src/riptide_collector/models.py

@@ -115,22 +115,34 @@ class NoerglerEvent(Base):
         Index("ix_noergler_events_event_type", "event_type"),
         Index("ix_noergler_events_pr_key", "pr_key"),
         Index("ix_noergler_events_commit_sha", "commit_sha"),
-        Index("ix_noergler_events_model", "model"),
+        Index("ix_noergler_events_outcome", "outcome"),
     )
 
     id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
     delivery_id: Mapped[str] = mapped_column(String, nullable=False)
+    # 'pr_completed' | 'feedback' — historical rows may carry the deprecated
+    # 'completed' value before the per-run → per-PR rollup switch.
     event_type: Mapped[str] = mapped_column(String, nullable=False)
     pr_key: Mapped[str | None] = mapped_column(String, nullable=True)
     repo: Mapped[str | None] = mapped_column(String, nullable=True)
+    # pr_completed: source-branch HEAD at close time. feedback: optional join key.
     commit_sha: Mapped[str | None] = mapped_column(String, nullable=True)
-    run_id: Mapped[str | None] = mapped_column(String, nullable=True)
-    model: Mapped[str | None] = mapped_column(String, nullable=True)
+    # pr_completed-only:
+    outcome: Mapped[str | None] = mapped_column(String, nullable=True)
+    merge_commit_sha: Mapped[str | None] = mapped_column(String, nullable=True)
+    lines_added: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    lines_removed: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    files_changed: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    total_runs: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    models_used: Mapped[list[str] | None] = mapped_column(ARRAY(String), nullable=True)
+    first_review_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    # Aggregated finops totals across all review runs that fed into this PR rollup.
     prompt_tokens: Mapped[int | None] = mapped_column(Integer, nullable=True)
     completion_tokens: Mapped[int | None] = mapped_column(Integer, nullable=True)
     elapsed_ms: Mapped[int | None] = mapped_column(Integer, nullable=True)
     findings_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
     cost_usd: Mapped[Decimal | None] = mapped_column(Numeric(12, 6), nullable=True)
+    # feedback-only:
     finding_id: Mapped[str | None] = mapped_column(String, nullable=True)
     verdict: Mapped[str | None] = mapped_column(String, nullable=True)
     actor: Mapped[str | None] = mapped_column(String, nullable=True)

src/riptide_collector/routers/noergler.py

@@ -8,8 +8,8 @@
 from riptide_collector.models import NoerglerEvent
 from riptide_collector.parsers import lower
 from riptide_collector.schemas.noergler import (
-    NoerglerCompleted,
     NoerglerFeedback,
+    NoerglerPrCompleted,
     NoerglerWebhook,
 )
 
@@ -25,16 +25,16 @@ def make_router(
     @router.post(
         "/noergler",
         status_code=status.HTTP_202_ACCEPTED,
-        summary="Noergler PR-review webhook sink (cost + reviewer-precision)",
+        summary="Noergler PR-review webhook sink (per-PR rollup + reviewer-precision)",
     )
     async def noergler_webhook(  # pyright: ignore[reportUnusedFunction]
         event: NoerglerWebhook,
         caller_team: str = Depends(auth_dep),
     ) -> dict[str, str]:
         raw = event.model_dump(mode="json")
 
-        if isinstance(event, NoerglerCompleted):
-            values = _values_completed(event, caller_team, raw)
+        if isinstance(event, NoerglerPrCompleted):
+            values = _values_pr_completed(event, caller_team, raw)
         else:
             values = _values_feedback(event, caller_team, raw)
 
@@ -70,27 +70,36 @@ async def noergler_webhook(  # pyright: ignore[reportUnusedFunction]
     return router
 
 
-def _values_completed(
-    event: NoerglerCompleted, caller_team: str, raw: dict[str, Any]
+def _values_pr_completed(
+    event: NoerglerPrCompleted, caller_team: str, raw: dict[str, Any]
 ) -> dict[str, Any]:
-    # run_id is unique per noergler review run, so it alone is a stable
-    # idempotency key. Prefix with event_type so a future event re-using a
-    # run_id can't collide.
-    delivery_id = f"completed#{event.run_id}"
+    # A PR has at most one terminal outcome, but Bitbucket can redeliver the
+    # lifecycle webhook (e.g. retries). Idempotency keyed on (pr_key, outcome)
+    # — including outcome in the key lets a deleted-after-decline edge case
+    # land two distinct rows if the operator wires both events. pr_key is
+    # lowercased to match the stored column so a casing-flipped redelivery
+    # (PROJ/... vs proj/...) still dedupes.
+    delivery_id = f"pr_completed#{lower(event.pr_key)}#{event.outcome}"
     return {
         "delivery_id": delivery_id,
-        "event_type": "completed",
+        "event_type": "pr_completed",
+        "outcome": event.outcome,
         "pr_key": lower(event.pr_key),
         "repo": lower(event.repo),
-        "commit_sha": lower(event.commit_sha),
-        "run_id": event.run_id,
-        "model": event.model,
-        "prompt_tokens": event.prompt_tokens,
-        "completion_tokens": event.completion_tokens,
-        "elapsed_ms": event.elapsed_ms,
-        "findings_count": event.findings_count,
-        "cost_usd": event.cost_usd,
-        "occurred_at": event.finished_at,
+        "commit_sha": lower(event.source_commit_sha),
+        "merge_commit_sha": lower(event.merge_commit_sha) if event.merge_commit_sha else None,
+        "lines_added": event.lines_added,
+        "lines_removed": event.lines_removed,
+        "files_changed": event.files_changed,
+        "total_runs": event.total_runs,
+        "models_used": event.models_used,
+        "prompt_tokens": event.total_prompt_tokens,
+        "completion_tokens": event.total_completion_tokens,
+        "elapsed_ms": event.total_elapsed_ms,
+        "findings_count": event.total_findings_count,
+        "cost_usd": event.total_cost_usd,
+        "first_review_at": event.first_review_at,
+        "occurred_at": event.closed_at,
         "team": caller_team,
         "payload": raw,
     }

src/riptide_collector/schemas/noergler.py

@@ -1,12 +1,13 @@
 """Noergler PR-review payloads.
 
-Two event types, both keyed off review activity (not PR lifecycle — PR
-lifecycle already comes in via Bitbucket):
+Two event types:
 
-- `completed`: emitted after an LLM review run finishes. Carries the finops
-  signal — model, token counts, elapsed time, cost.
+- `pr_completed`: emitted once per PR lifecycle (merged / declined / deleted).
+  Carries the roll-up finops signal — aggregated tokens, elapsed time, cost,
+  findings and the final PR diff-size — plus an `outcome` so consumers can
+  filter for merged PRs vs. abandoned ones.
 - `feedback`: emitted when a reviewer disagrees with or acknowledges a
-  finding. Carries the reviewer-precision signal.
+  finding. Reactive and not tied to the PR lifecycle.
 """
 
 from __future__ import annotations
@@ -15,7 +16,7 @@
 from decimal import Decimal
 from typing import Annotated, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 
 def _to_utc(value: datetime) -> datetime:
@@ -29,25 +30,72 @@ class _Common(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
 
-class NoerglerCompleted(_Common):
-    event_type: Literal["completed"]
+class NoerglerPrCompleted(_Common):
+    event_type: Literal["pr_completed"]
+    outcome: Literal["merged", "declined", "deleted"] = Field(
+        ...,
+        description=(
+            "PR lifecycle outcome. Only 'merged' PRs reach production; "
+            "'declined' and 'deleted' incurred LLM cost without shipping code "
+            "and must be filtered out of throughput / DORA metrics."
+        ),
+    )
     pr_key: str = Field(..., min_length=1, description="Bitbucket PR key, e.g. 'PROJ/repo#42'")
     repo: str = Field(..., min_length=1)
-    commit_sha: str = Field(..., min_length=7)
-    run_id: str = Field(..., min_length=1, description="noergler review-run id (idempotency key)")
-    model: str = Field(..., min_length=1, description="LLM identifier, e.g. 'gpt-4o-2024-08-06'")
-    prompt_tokens: int = Field(..., ge=0)
-    completion_tokens: int = Field(..., ge=0)
-    elapsed_ms: int = Field(..., ge=0)
-    findings_count: int = Field(..., ge=0)
-    cost_usd: Decimal = Field(..., ge=0)
-    finished_at: datetime
-
-    @field_validator("finished_at")
+    source_commit_sha: str = Field(
+        ...,
+        min_length=7,
+        description="Last reviewed source-branch commit (HEAD of fromRef when the PR closed).",
+    )
+    merge_commit_sha: str | None = Field(
+        default=None,
+        min_length=7,
+        description="Merge commit SHA. Set only when outcome='merged'.",
+    )
+    lines_added: int = Field(..., ge=0, description="Final cumulative PR diff: lines added.")
+    lines_removed: int = Field(..., ge=0, description="Final cumulative PR diff: lines removed.")
+    files_changed: int = Field(..., ge=0, description="Final cumulative PR diff: files changed.")
+    total_runs: int = Field(..., ge=1, description="Number of review runs aggregated.")
+    total_prompt_tokens: int = Field(..., ge=0)
+    total_completion_tokens: int = Field(..., ge=0)
+    total_elapsed_ms: int = Field(..., ge=0)
+    total_findings_count: int = Field(..., ge=0)
+    total_cost_usd: Decimal = Field(..., ge=0)
+    models_used: list[str] = Field(
+        ...,
+        min_length=1,
+        description="Distinct LLM identifiers used across the PR's review runs.",
+    )
+    first_review_at: datetime
+    closed_at: datetime = Field(
+        ...,
+        description="Timestamp the PR reached its terminal outcome (merged/declined/deleted).",
+    )
+
+    @field_validator("first_review_at", "closed_at")
     @classmethod
     def _normalise_tz(cls, v: datetime) -> datetime:
         return _to_utc(v)
 
+    @field_validator("models_used")
+    @classmethod
+    def _models_non_empty(cls, v: list[str]) -> list[str]:
+        if any(not m.strip() for m in v):
+            raise ValueError("models_used entries must be non-empty strings")
+        return v
+
+    @model_validator(mode="after")
+    def _check_merge_commit_consistency(self) -> NoerglerPrCompleted:
+        # Catch sender bugs: only merged PRs produce a merge commit, and a
+        # merged PR is exactly the case where the sender must supply one.
+        # Without this, "outcome=declined + merge_commit_sha=<sha>" would
+        # silently land in the DB.
+        if self.outcome == "merged" and not self.merge_commit_sha:
+            raise ValueError("merge_commit_sha is required when outcome='merged'")
+        if self.outcome != "merged" and self.merge_commit_sha is not None:
+            raise ValueError(f"merge_commit_sha must be null when outcome='{self.outcome}'")
+        return self
+
 
 class NoerglerFeedback(_Common):
     event_type: Literal["feedback"]
@@ -83,6 +131,6 @@ def _normalise_tz(cls, v: datetime) -> datetime:
 
 
 NoerglerWebhook = Annotated[
-    NoerglerCompleted | NoerglerFeedback,
+    NoerglerPrCompleted | NoerglerFeedback,
     Field(discriminator="event_type"),
 ]

tests/fixtures/noergler_completed.json

@@ -0,0 +1,19 @@
+{
+  "event_type": "pr_completed",
+  "outcome": "declined",
+  "pr_key": "PROJ/payments-api#43",
+  "repo": "acme/payments-api",
+  "source_commit_sha": "f00ba1234567890abc1234567890abc123456789",
+  "lines_added": 40,
+  "lines_removed": 5,
+  "files_changed": 2,
+  "total_runs": 1,
+  "total_prompt_tokens": 4200,
+  "total_completion_tokens": 180,
+  "total_elapsed_ms": 3100,
+  "total_findings_count": 2,
+  "total_cost_usd": "0.042000",
+  "models_used": ["gpt-4o-2024-08-06"],
+  "first_review_at": "2026-04-30T09:10:00Z",
+  "closed_at": "2026-04-30T09:25:00Z"
+}

tests/fixtures/noergler_pr_completed_declined.json

@@ -0,0 +1,19 @@
+{
+  "event_type": "pr_completed",
+  "outcome": "deleted",
+  "pr_key": "PROJ/payments-api#44",
+  "repo": "acme/payments-api",
+  "source_commit_sha": "deadbeef1234567890abc1234567890abc123456",
+  "lines_added": 0,
+  "lines_removed": 0,
+  "files_changed": 0,
+  "total_runs": 1,
+  "total_prompt_tokens": 1500,
+  "total_completion_tokens": 60,
+  "total_elapsed_ms": 1700,
+  "total_findings_count": 0,
+  "total_cost_usd": "0.015000",
+  "models_used": ["gpt-4o-2024-08-06"],
+  "first_review_at": "2026-05-01T08:00:00Z",
+  "closed_at": "2026-05-01T08:05:00Z"
+}