feat(stamphog): wait for and count PostHog Review through a backend evidence hook

products/review_hog/ARCHITECTURE.md

@@ -461,7 +461,9 @@ pr_metadata.head_branch` is threaded (as explicit kwargs, alongside `team_id` /
     suppress the publish or clobber a stranger's comment.
     **Reactions:** once a publishing turn passes every gate, the workflow puts a 👀 on the PR (`add_review_reaction_activity`) and removes it by the returned reaction id when the turn ends, on success, failure, or cancellation (the `finally` around the review stages).
     Both calls are fail-open and gated by the `reviewhog-pr-reactions-2026-10` patch.
-    Stamphog lists `posthog[bot]` in its trusted reviewer bots, so it waits while that 👀 is fresh.
+    The 👀 is a cue for people only. The per-head review state is review_hog's public read API, `facade/review_status.py::review_status_for_head` (contract `HeadReviewStatus` in `facade/contracts.py`, implemented by `head_review_status.py`). Stamphog cannot import it (ReviewHog depends on stamphog, never the reverse), so `receivers.connect()` registers `review_evidence_for_stamphog`, a thin adapter onto stamphog's `facade/review_evidence.py` registry, with the reviewer named "PostHog Review".
+    It answers per head: pending while a request for the head waits in the queue (the queue workflow's `requested_heads` query) or a turn on the head runs, through its publish; completed with the kind, the valid findings count, the published flag, the merge base the diff was taken against and the skipped files (filtered files, files GitHub sent without a patch, and the files of every pipeline chunk that no review unit finished, because a turn completes when only some of its units fail); failed when the turn for the head failed; None for any other head or a repository no project owns.
+    When a publishing turn ends, after publish or after the failure is recorded, `nudge_stamphog_activity` calls stamphog's `request_rereview_after_external_review` (behind the `reviewhog-nudge-stamphog-2026-10` patch, fail-open). Stamphog decides whether a WAIT at that head runs again.
     **Standard turns post no status comment** (the same patch): the 👀 shows the turn runs, and the outcome is the review, a 👍 on the PR when nothing is publishable, or one failure comment per head (`post_failure_comment`, hidden `reviewhog:failed` marker) posted by `fail_status_comment_activity`.
     A later publish with findings, in either mode, first removes the app's own 👍 (`remove_own_pr_reactions`, matched with `is_app_bot_author`).
     Status comments that earlier turns left on the PR stay as they are, and a Standard turn never refreshes them.

products/review_hog/backend/facade/review_status.py

@@ -0,0 +1,25 @@
+"""Facade for reading where PostHog Review's review of one pull request head stands.
+
+Any product may read it, except one that review_hog depends on: stamphog gets the same answer
+through the provider that `receivers.connect()` registers, because importing review_hog there would
+be a dependency cycle. Lives apart from `reviews.py` and `github.py` because the lookup imports the
+Temporal client.
+"""
+
+from products.review_hog.backend.facade.review_status_types import HeadReviewState, HeadReviewStatus
+from products.review_hog.backend.head_review_status import HeadReviewStatusLookup
+from products.review_hog.backend.ownership import RepositoryOwnership
+
+__all__ = ["HeadReviewState", "HeadReviewStatus", "review_status_for_head"]
+
+
+def review_status_for_head(repository: str, pr_number: int, head_sha: str) -> HeadReviewStatus | None:
+    """The status of the review of exactly `head_sha`, or None when none is known.
+
+    None also when no project reviews the repository. Raises when Temporal does not answer the
+    queue query, so a caller cannot read an outage as "nothing pending".
+    """
+    owner = RepositoryOwnership.find_by_name(repository)
+    if owner is None:
+        return None
+    return HeadReviewStatusLookup(owner.team_id, repository, pr_number, head_sha).status()

products/review_hog/backend/facade/review_status_types.py

@@ -0,0 +1,31 @@
+"""Types of the head review status that review_hog exposes to other products. No Django imports."""
+
+from dataclasses import dataclass
+from enum import StrEnum
+
+
+class HeadReviewState(StrEnum):
+    # Accepted but not started, or running. A turn stays pending until its findings are on the PR.
+    PENDING = "pending"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+
+@dataclass(frozen=True, kw_only=True)
+class HeadReviewStatus:
+    """Where PostHog Review's review of one pull request head stands."""
+
+    state: HeadReviewState
+    # "Standard" or "Deep" for a completed turn. Empty while pending or failed.
+    review_kind: str
+    # Valid findings of a completed turn. Zero while pending or failed.
+    findings_count: int
+    # True when the turn posted a review on the PR. A clean turn posts nothing, so it stays False.
+    published: bool
+    head_sha: str
+    # The base branch the completed turn reviewed the head against. Empty when it is not known.
+    base_branch: str = ""
+    # The merge base the completed turn took the diff against. Empty when it is not known.
+    base_sha: str = ""
+    # The changed files the completed turn left out of the review, for example lock files or a chunk no unit read.
+    skipped_paths: tuple[str, ...] = ()

products/review_hog/backend/head_review_status.py

@@ -0,0 +1,201 @@
+"""Where PostHog Review's review of one PR head stands. Read through `facade/review_status.py`.
+
+The answer is per head, unlike `pr_status.PRStatusLookup`, which reports "queued" for the whole PR.
+The database answers for turns that started: a running turn keeps the report ACTIVE on its head
+until its findings are on the PR, and a finished turn leaves a turn marker, the completed head, and
+the published heads. A request that waits in the PR's review queue exists only in the queue
+workflow, so the workflow's `requested_heads` query answers for it.
+"""
+
+from datetime import datetime, timedelta
+
+from posthog.dataclasses import frozen
+
+from products.review_hog.backend.facade.review_status_types import HeadReviewState, HeadReviewStatus
+from products.review_hog.backend.models import ReviewReport, ReviewReportArtefact
+from products.review_hog.backend.reviewer.artefact_content import TurnMarkerArtefact
+from products.review_hog.backend.reviewer.constants import REVIEW_MODE_FLASH, REVIEW_MODE_FULL
+from products.review_hog.backend.reviewer.persistence import (
+    load_chunk_set,
+    load_turn_findings,
+    reviewed_pipeline_chunk_ids,
+)
+from products.review_hog.backend.reviewer.progress import (
+    RUN_OUTCOME_FAILED,
+    RUN_STAGE_REVIEW,
+    finding_counts,
+    in_progress_report_ids,
+    run_outcome_markers,
+)
+from products.review_hog.backend.reviewer.review_design import REVIEW_DESIGN_SINGLE_AGENT
+from products.review_hog.backend.reviewer.review_state import turn_review_mode
+from products.review_hog.backend.temporal.client import queued_review_heads
+from products.review_hog.backend.temporal.types import review_pr_workflow_id
+
+_KIND_NAMES = {REVIEW_MODE_FLASH: "Standard", REVIEW_MODE_FULL: "Deep"}
+# Callers poll, for example stamphog every 30 seconds, so a slow queue answer must not hold them.
+_QUERY_RPC_TIMEOUT = timedelta(seconds=2)
+
+
+@frozen
+class _Turn:
+    marker: TurnMarkerArtefact
+    # When the newest marker row of the turn was written, which is when its last attempt started.
+    started_at: datetime
+
+
+class HeadReviewStatusLookup:
+    def __init__(self, team_id: int, repository: str, pr_number: int, head_sha: str) -> None:
+        self.team_id = team_id
+        self.repository = repository
+        self.pr_number = pr_number
+        self.head_sha = head_sha
+        # Repository casing can differ per trigger, so match it the way the trigger does.
+        self.report = (
+            ReviewReport.objects.for_team(team_id).filter(repository__iexact=repository, pr_number=pr_number).first()
+        )
+        self.report_id = str(self.report.id) if self.report is not None else ""
+
+    def _status(
+        self,
+        state: HeadReviewState,
+        *,
+        review_kind: str = "",
+        findings_count: int = 0,
+        published: bool = False,
+        base_branch: str = "",
+        base_sha: str = "",
+        skipped_paths: tuple[str, ...] = (),
+    ) -> HeadReviewStatus:
+        return HeadReviewStatus(
+            state=state,
+            review_kind=review_kind,
+            findings_count=findings_count,
+            published=published,
+            head_sha=self.head_sha,
+            base_branch=base_branch,
+            base_sha=base_sha,
+            skipped_paths=skipped_paths,
+        )
+
+    def _running_head(self) -> str | None:
+        """The head of the turn that runs now, or None. A turn stays ACTIVE through its publish."""
+        report = self.report
+        if report is None or report.status != ReviewReport.Status.ACTIVE:
+            return None
+        if self.report_id not in in_progress_report_ids(self.team_id, [report]):
+            return None
+        return report.head_sha
+
+    def _queued(self, running_head: str | None) -> bool:
+        owner, _, repo = self.repository.partition("/")
+        workflow_id = review_pr_workflow_id(team_id=self.team_id, owner=owner, repo=repo, pr_number=self.pr_number)
+        heads = queued_review_heads(workflow_id, rpc_timeout=_QUERY_RPC_TIMEOUT)
+        if self.head_sha in heads.active or self.head_sha in heads.pending:
+            return True
+        # A request without a head reviews the live head when its turn starts. A waiting one has not
+        # started yet. The active one is the running turn when a turn runs on another head.
+        return None in heads.pending or (None in heads.active and running_head is None)
+
+    def _turns_on_head(self) -> dict[int, _Turn]:
+        """Each turn on this head by run index, from its marker. The newest row of a retried turn wins."""
+        rows = (
+            ReviewReportArtefact.objects.for_team(self.team_id)
+            .filter(
+                report_id=self.report_id,
+                type=ReviewReportArtefact.ArtefactType.TURN_MARKER,
+                head_sha=self.head_sha,
+            )
+            .order_by("created_at", "id")
+            .values("content", "created_at")
+        )
+        turns: dict[int, _Turn] = {}
+        for row in rows:
+            marker = TurnMarkerArtefact.model_validate_json(row["content"])
+            turns[marker.run_index] = _Turn(marker=marker, started_at=row["created_at"])
+        return turns
+
+    def _skipped_paths(self, turn: _Turn | None) -> tuple[str, ...]:
+        """The files the turn left out, plus the files of every pipeline chunk that no review unit finished.
+
+        The fan-out fails a turn only when most of its units fail, so a turn can complete with a chunk unread.
+        """
+        if turn is None:
+            return ()
+        skipped = list(turn.marker.skipped_paths or ())
+        # A single-agent turn fails when its main session fails, and that session reviews the whole diff.
+        if turn.marker.fingerprint_inputs.get("review_design") == REVIEW_DESIGN_SINGLE_AGENT:
+            return tuple(skipped)
+        chunks = load_chunk_set(team_id=self.team_id, report_id=self.report_id, head_sha=self.head_sha)
+        # Without the chunking the gaps are unknown, so the turn keeps only the files it left out.
+        if chunks is None:
+            return tuple(skipped)
+        reviewed = reviewed_pipeline_chunk_ids(team_id=self.team_id, report_id=self.report_id, head_sha=self.head_sha)
+        for chunk in chunks.chunks:
+            if chunk.chunk_id in reviewed:
+                continue
+            for file in chunk.files:
+                if file.filename not in skipped:
+                    skipped.append(file.filename)
+        return tuple(skipped)
+
+    def _finished(self) -> HeadReviewStatus | None:
+        report = self.report
+        if report is None:
+            return None
+        published_heads = report.published_head_shas or {}
+
+        def posted(run_index: int) -> bool:
+            return published_heads.get(str(run_index)) == self.head_sha
+
+        turns = {run_index: turn for run_index, turn in self._turns_on_head().items() if run_index <= report.run_count}
+        # Only a turn that posts to the PR is a review of the PR. An eval or CLI turn is not, even
+        # with valid findings. A clean publishing turn posts nothing and still counts. A marker
+        # from before `publishes` existed counts only when its turn posted.
+        completed_runs = [
+            run_index
+            for run_index, turn in turns.items()
+            if turn.marker.publishes or (turn.marker.publishes is None and posted(run_index))
+        ]
+        # Turns that started before turn markers existed have none, so the latest turn's head counts too.
+        if (
+            report.run_count
+            and report.completed_head_sha == self.head_sha
+            and report.run_count not in turns
+            and posted(report.run_count)
+        ):
+            completed_runs.append(report.run_count)
+        failed_at: dict[int | None, datetime] = {
+            marker.run_index: marker.created_at
+            for marker in run_outcome_markers(self.team_id, [self.report_id])[self.report_id]
+            if marker.stage == RUN_STAGE_REVIEW
+            and marker.outcome == RUN_OUTCOME_FAILED
+            and marker.head_sha == self.head_sha
+        }
+        if not completed_runs:
+            return self._status(HeadReviewState.FAILED) if failed_at else None
+        run_index = max(completed_runs)
+        turn = turns.get(run_index)
+        # Finalize counts a turn before it publishes, so a failed publish marks a counted turn failed.
+        # A failure from before the turn's last attempt started belongs to an attempt that a
+        # successful retry replaced.
+        if run_index in failed_at and (turn is None or failed_at[run_index] > turn.started_at):
+            return self._status(HeadReviewState.FAILED)
+        pairs = load_turn_findings(team_id=self.team_id, report_id=self.report_id, run_index=run_index)
+        counts, _ = finding_counts(pairs)
+        review_mode = turn.marker.review_mode if turn is not None else turn_review_mode(pairs)
+        return self._status(
+            HeadReviewState.COMPLETED,
+            review_kind=_KIND_NAMES.get(review_mode, ""),
+            findings_count=sum(counts.values()),
+            published=posted(run_index),
+            base_branch=(turn.marker.base_branch or "") if turn is not None else "",
+            base_sha=(turn.marker.base_sha or "") if turn is not None else "",
+            skipped_paths=self._skipped_paths(turn),
+        )
+
+    def status(self) -> HeadReviewStatus | None:
+        running_head = self._running_head()
+        if running_head == self.head_sha or self._queued(running_head):
+            return self._status(HeadReviewState.PENDING)
+        return self._finished()

products/review_hog/backend/receivers.py

@@ -49,6 +49,11 @@
 from products.signals.backend.models import SignalReportArtefact
 from products.signals.backend.report_generation.priority import persisted_report_priority
 from products.stamphog.backend.facade.inbox_hooks import register_inbox_acting_reviewer_resolver
+from products.stamphog.backend.facade.review_evidence import (
+    ReviewEvidence,
+    ReviewEvidenceState,
+    register_review_evidence_provider,
+)
 
 # This module loads during django.setup() (AppConfig.ready() wires the receiver), and
 # posthog/test/repo_invariants/test_startup_import_budget.py forbids temporalio/modal/openai/anthropic at setup —
@@ -57,9 +62,12 @@
 
 logger = logging.getLogger(__name__)
 
+# The name stamphog shows for these reviews, the same name the reviews carry on GitHub.
+REVIEWER_NAME = "PostHog Review"
+
 
 def connect() -> None:
-    """Wire the TaskRun-save receiver and the stamphog toggle hook; called once from `AppConfig.ready()`."""
+    """Wire the TaskRun-save receiver and the stamphog hooks; called once from `AppConfig.ready()`."""
     post_save.connect(
         handle_task_run_saved,
         sender=django_apps.get_model("tasks", "TaskRun"),
@@ -68,9 +76,33 @@
     # Stamphog re-checks this toggle before re-reviewing a self-driving PR. It gets the resolver
     # through a hook because importing review_hog from stamphog would be a dependency cycle.
     register_inbox_acting_reviewer_resolver(resolve_stamphog_acting_reviewer)
+    # Stamphog waits for a pending review of the head it judges and counts a completed one as
+    # assurance. The same dependency cycle rules out a direct import.
+    register_review_evidence_provider(review_evidence_for_stamphog)
+
+
+def review_evidence_for_stamphog(repository: str, pr_number: int, head_sha: str) -> ReviewEvidence | None:
+    """Map review_hog's own head status onto stamphog's evidence contract."""
+    # The lookup reads the review pipeline's modules, which reach openai, anthropic and temporalio.
+    from products.review_hog.backend.facade.review_status import review_status_for_head  # noqa: PLC0415
+
+    status = review_status_for_head(repository, pr_number, head_sha)
+    if status is None:
+        return None
+    return ReviewEvidence(
+        state=ReviewEvidenceState(status.state.value),
+        reviewer=REVIEWER_NAME,
+        review_kind=status.review_kind,
+        findings_count=status.findings_count,
+        published=status.published,
+        head_sha=status.head_sha,
+        base_ref=status.base_branch,
+        base_sha=status.base_sha,
+        skipped_paths=status.skipped_paths,
+    )
 
 
 def handle_task_run_saved(sender: type, instance: Any, created: bool, **kwargs: Any) -> None:
     """Start an inbox review when a signals-origin implementation run records its PR.
 
     Fires on every TaskRun save (a hot model), so the checks run cheapest-first and the whole body is

products/review_hog/backend/reviewer/artefact_content.py

@@ -327,7 +327,10 @@ class PRSnapshotArtefact(BaseModel):
     )
     merge_base_sha: str | None = Field(
         default=None,
-        description="The commit GitHub computes the PR diff against. Fetched for single-agent turns only.",
+        description="The commit GitHub computes the PR diff against. None when GitHub did not return it.",
+    )
+    skipped_paths: list[str] = Field(
+        default_factory=list, description="The changed files the review left out (lock, generated, test files)."
     )
 
 
@@ -347,6 +350,13 @@ class TurnMarkerArtefact(BaseModel):
     fingerprint_inputs: dict[str, Any] = Field(
         default_factory=dict, description="The payload the fingerprint hashes (prompt and skill texts as hashes)."
     )
+    # The fields below are None on markers written before they existed.
+    publishes: bool | None = Field(default=None, description="Whether the turn posts its result to the PR.")
+    base_branch: str | None = Field(default=None, description="The base branch the reviewed diff was taken against.")
+    base_sha: str | None = Field(default=None, description="The merge base the reviewed diff was taken against.")
+    skipped_paths: list[str] | None = Field(
+        default=None, description="The changed files the turn left out of the review."
+    )
 
 
 RUN_OUTCOME_NOTE_AUTHOR = "review_hog_run_outcome"