Fix cross-repository PR cost coverage

docs/cloud-data-contract.md

@@ -249,6 +249,18 @@ identifiers that did not report cost; it is metadata-only and must never
 contain commands, paths, auth output, or secrets.  Missing cost is omitted,
 never serialized as zero.
 
+Shared local spend ledgers may contain attempts for several repositories. An
+attempt with an explicit different `repo_slug` is isolated to that repository
+and does not suppress coverage in the repository currently being synced. A
+builder run recorded before PR creation may omit `pr_number`; the command links
+it only when its recorded branch equals exactly one GitHub `headRefName` in the
+fetched PR population. Missing or ambiguous branch matches remain fail-closed.
+No branch prefix, author, PR body marker, or issue prose is used for this join.
+
+Dollar coverage remains provider-reported. Subscription access, token counts,
+or elapsed time are not silently converted to `cost_usd`; an explicit,
+versioned allocation policy would be a separate data source and contract.
+
 Builder evidence fails closed: a `*.cloud-event.json` file that cannot be
 read, parsed, or recognized as a `builder_run` event is never silently
 omitted.  A failure attributable to a PR via its filename is recorded on that

src/code_mower/cloud_client/operations.py

@@ -922,6 +922,11 @@ def _pr_number_from_run_event(event: Mapping[str, Any]) -> str:
     return canonical_pr_number(event.get("pr_number"))
 
 
+def _branch_from_builder_run(event: Mapping[str, Any]) -> str:
+    dimensions = _mapping(event.get("dimensions"))
+    return str(dimensions.get("branch") or event.get("branch") or "").strip()
+
+
 def _is_positive_int_pr_number(number: str) -> bool:
     """Return True only for a canonical positive integer PR number.
 
@@ -933,12 +938,11 @@ def _is_positive_int_pr_number(number: str) -> bool:
 
 
 def _is_associable_builder_run(payload: Any) -> bool:
-    """Return True when a parsed record has the fields needed to associate it.
+    """Return True when a parsed record can be associated directly or by branch.
 
-    A parseable but structurally unusable record -- such as one that only
-    carries ``event_type=builder_run``, or one whose PR number is not a
-    positive integer -- must not be silently filtered out; it routes through
-    filename-attributed or unattributable fail-closed evidence handling.
+    Builder records may be written before a PR exists.  A non-empty branch can
+    later identify one exact GitHub PR; the upload path performs that bounded
+    lookup.  Records without either identity remain fail-closed evidence.
     """
 
     if not isinstance(payload, dict):
@@ -947,9 +951,9 @@ def _is_associable_builder_run(payload: Any) -> bool:
         return False
     if not str(payload.get("repo_slug") or "").strip():
         return False
-    if not _is_positive_int_pr_number(_pr_number_from_run_event(payload)):
-        return False
-    return True
+    return _is_positive_int_pr_number(
+        _pr_number_from_run_event(payload)
+    ) or bool(_branch_from_builder_run(payload))
 
 
 def _builder_run_events(
@@ -1056,6 +1060,12 @@ def pr_outcomes_upload(
         limit=limit,
         repo_path=repo_path,
     )
+    pr_numbers_by_branch: dict[str, set[str]] = {}
+    for pr in pr_records:
+        branch = str(pr.get("headRefName") or "").strip()
+        pr_number = canonical_pr_number(pr.get("number"))
+        if branch and pr_number:
+            pr_numbers_by_branch.setdefault(branch, set()).add(pr_number)
 
     observation_state_path = repo_path / DEFAULT_OBSERVATION_STATE_PATH
     observation_lock_path = observation_state_path.with_name(
@@ -1115,11 +1125,28 @@ def pr_outcomes_upload(
             unattributable_evidence_count = 0
             for event in [*builder_events, *spend_events]:
                 event_repo = str(event.get("repo_slug") or "").strip()
+                # Shared local ledgers may intentionally contain evidence for
+                # several repositories.  An explicit different repository is
+                # attributable there, not missing evidence for this repo.
+                if event_repo and event_repo != detected_repo_slug:
+                    continue
                 event_pr = _pr_number_from_run_event(event)
+                if (
+                    not event_pr
+                    and event.get("event_type") == "builder_run"
+                ):
+                    branch = _branch_from_builder_run(event)
+                    matches = pr_numbers_by_branch.get(branch, set())
+                    if len(matches) == 1:
+                        event = dict(event)
+                        dimensions = dict(_mapping(event.get("dimensions")))
+                        dimensions["pr_number"] = next(iter(matches))
+                        event["dimensions"] = dimensions
+                        event_pr = _pr_number_from_run_event(event)
                 if not _is_positive_int_pr_number(event_pr):
                     unattributable_evidence_count += 1
                     continue
-                if event_repo != detected_repo_slug:
+                if not event_repo:
                     unattributable_evidence_count += 1
                     continue
                 run_events.append(event)

tests/test_pr_outcome_coverage.py

@@ -2575,7 +2575,7 @@ def test_unattributable_reviewer_spend_suppresses_complete(self) -> None:
             )
             self.assertNotIn(str(repo_path), " ".join(result["errors"]))
 
-    def test_mismatched_repo_slug_suppresses_complete_coverage(self) -> None:
+    def test_mismatched_repo_slug_isolated_from_current_repo(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             repo_path = Path(tmp)
             builder_dir = repo_path / ".code-mower" / "builder-runs"
@@ -2605,17 +2605,112 @@ def test_mismatched_repo_slug_suppresses_complete_coverage(self) -> None:
             result = json.loads(raw)
             events = self._emitted_events(result)
             pr2 = events["2"]
-            self.assertEqual(pr2["dimensions"]["cost_coverage"], "partial")
-            self.assertEqual(pr2["metrics"]["cost_expected_run_count"], 2)
-            self.assertIn(
-                "unreadable-evidence",
-                pr2["dimensions"]["missing_cost_sources"],
+            self.assertEqual(pr2["dimensions"]["cost_coverage"], "complete")
+            self.assertEqual(pr2["metrics"]["cost_expected_run_count"], 1)
+            self.assertEqual(pr2["metrics"]["cost_reported_run_count"], 1)
+            self.assertEqual(result["errors"], [])
+
+    def test_shared_spend_rows_for_other_repo_do_not_suppress_current_repo(
+        self,
+    ) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            repo_path = Path(tmp)
+            spend_path = repo_path / ".code-mower" / "reviewer-spend.json"
+            spend_path.parent.mkdir(parents=True)
+            spend_path.write_text(
+                json.dumps(
+                    {
+                        "schema": reviewer_spend.SPEND_SCHEMA,
+                        "runs": [
+                            {
+                                "run_id": "current",
+                                "lane": "claude-audit",
+                                "repo": "owner/repo",
+                                "pr_number": 2,
+                                "head_sha": "abc123",
+                                "model": "sonnet",
+                                "wall_seconds": 1.0,
+                                "verdict": "PASS",
+                                "created_at": "2026-09-03T11:00:00Z",
+                                "cost_usd": 0.05,
+                            },
+                            {
+                                "run_id": "other",
+                                "lane": "claude-audit",
+                                "repo": "owner/other-repo",
+                                "pr_number": 9,
+                                "head_sha": "def456",
+                                "model": "sonnet",
+                                "wall_seconds": 1.0,
+                                "verdict": "PASS",
+                                "created_at": "2026-09-03T11:00:00Z",
+                                "cost_usd": 0.07,
+                            },
+                        ],
+                    }
+                ),
+                encoding="utf-8",
             )
-            self.assertTrue(pr2["dimensions"]["evidence_incomplete"])
+
+            code, raw = self._run_upload(
+                repo_path, repo_path / "bundle", [self._merged_pr("2")]
+            )
+            self.assertEqual(code, 0, raw)
+            result = json.loads(raw)
+            pr2 = self._emitted_events(result)["2"]
+            self.assertEqual(pr2["dimensions"]["cost_coverage"], "complete")
+            self.assertEqual(pr2["metrics"]["cost_expected_run_count"], 1)
+            self.assertEqual(pr2["metrics"]["reported_cost_usd"], 0.05)
+            self.assertEqual(result["errors"], [])
+
+    def test_builder_branch_links_only_one_exact_github_pr(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            repo_path = Path(tmp)
+            builder_dir = repo_path / ".code-mower" / "builder-runs"
+            builder_dir.mkdir(parents=True)
+            builder = _builder_run_event("b2", "", 0.10)
+            builder["dimensions"]["branch"] = "codex/issue-1"
+            (builder_dir / "codex-issue-1.cloud-event.json").write_text(
+                json.dumps(builder), encoding="utf-8"
+            )
+            pr = {**self._merged_pr("2"), "headRefName": "codex/issue-1"}
+
+            code, raw = self._run_upload(repo_path, repo_path / "bundle", [pr])
+            self.assertEqual(code, 0, raw)
+            result = json.loads(raw)
+            pr2 = self._emitted_events(result)["2"]
+            self.assertEqual(pr2["dimensions"]["cost_coverage"], "complete")
+            self.assertEqual(pr2["metrics"]["cost_expected_run_count"], 1)
+            self.assertEqual(pr2["metrics"]["reported_cost_usd"], 0.10)
+            self.assertEqual(result["errors"], [])
+
+    def test_ambiguous_builder_branch_remains_fail_closed(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            repo_path = Path(tmp)
+            builder_dir = repo_path / ".code-mower" / "builder-runs"
+            builder_dir.mkdir(parents=True)
+            builder = _builder_run_event("b2", "", 0.10)
+            builder["dimensions"]["branch"] = "shared/branch"
+            (builder_dir / "shared-branch.cloud-event.json").write_text(
+                json.dumps(builder), encoding="utf-8"
+            )
+            prs = [
+                {**self._merged_pr("2"), "headRefName": "shared/branch"},
+                {**self._merged_pr("3"), "headRefName": "shared/branch"},
+            ]
+
+            code, raw = self._run_upload(repo_path, repo_path / "bundle", prs)
+            self.assertEqual(code, 0, raw)
+            result = json.loads(raw)
+            events = self._emitted_events(result)
+            self.assertEqual(events["2"]["dimensions"]["cost_coverage"], "unknown")
+            self.assertEqual(events["3"]["dimensions"]["cost_coverage"], "unknown")
             self.assertTrue(
-                any("parsed attempt(s)" in e for e in result["errors"])
+                all(event["dimensions"]["evidence_incomplete"] for event in events.values())
+            )
+            self.assertTrue(
+                any("parsed attempt(s)" in error for error in result["errors"])
             )
-            self.assertNotIn(str(repo_path), " ".join(result["errors"]))
 
     def test_truncated_pr_can_retry_and_emit_later(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
@@ -2931,6 +3026,22 @@ def test_valid_builder_runs_directory_returns_events(self) -> None:
             self.assertEqual(events[0]["event_id"], "b1")
             self.assertEqual(unreadable, [])
 
+    def test_pre_pr_builder_run_with_branch_is_valid_evidence(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            repo_path = Path(tmp)
+            builder_dir = repo_path / ".code-mower" / "builder-runs"
+            builder_dir.mkdir(parents=True)
+            builder = _builder_run_event("b1", "", 0.10)
+            builder["dimensions"]["branch"] = "codex/issue-1"
+            (builder_dir / "codex-issue-1.cloud-event.json").write_text(
+                json.dumps(builder), encoding="utf-8"
+            )
+
+            events, unreadable = _builder_run_events(repo_path)
+            self.assertEqual(len(events), 1)
+            self.assertEqual(events[0]["event_id"], "b1")
+            self.assertEqual(unreadable, [])
+
 
 class PrOutcomeIdentifierP2Tests(unittest.TestCase):
     def test_secret_like_lane_identifiers_become_unknown_source(self) -> None: