fix: live pre-merge build-status revalidation (BERTE-602)

CLAUDE.md

@@ -0,0 +1,20 @@
+# bert-e — Claude Code notes
+
+## Running tests
+
+pytest is at `.venv/bin/pytest` (not in system PATH):
+
+```sh
+.venv/bin/pytest
+```
+
+## Creating a GitHub release
+
+`gh release create` requires the **full** commit SHA for `--target` — short SHAs are rejected with HTTP 422. Always resolve the full SHA first:
+
+```sh
+FULL_SHA=$(git rev-parse origin/main)
+gh release create <tag> --target "$FULL_SHA" --prerelease --generate-notes --title "<tag>"
+```
+
+Follow the checklist in `devdocs/docs/tools/bert-e/release.md` for the full release process (tag naming, pre-release flag, monitoring the Actions workflow, then updating devinfra).

bert_e/git_host/github/__init__.py

@@ -717,6 +717,39 @@ def state(self):
         else:
             return 'NOTSTARTED'
 
+    def state_for_branch(self, branch_name: str) -> str:
+        """Return aggregated state for runs on a specific branch only.
+
+        Filters to branch_name before deduplicating, so a sibling branch
+        that shares the same SHA cannot cause a false SUCCESSFUL result.
+        Unlike .state, this never returns SUCCESSFUL for a different branch.
+        """
+        conclusion_ranking = {
+            'success': 4, 'skipped': 4, None: 3, 'failure': 2, 'cancelled': 1
+        }
+        runs = [
+            r for r in self._workflow_runs
+            if r['head_branch'] == branch_name and
+            r.get('event') != 'workflow_dispatch'
+        ]
+        best: dict = {}
+        for run in runs:
+            wid = run['workflow_id']
+            conclusion = run['conclusion']
+            if (wid not in best or
+                    conclusion_ranking[conclusion] >
+                    conclusion_ranking[best[wid]['conclusion']]):
+                best[wid] = run
+        branch_runs = list(best.values())
+        LOG.info(
+            "state_for_branch: branch=%s runs=%s",
+            branch_name,
+            [{'id': r['id'], 'status': r['status'],
+              'conclusion': r['conclusion']}
+             for r in branch_runs],
+        )
+        return self.branch_state(branch_runs)
+
     @property
     def description(self) -> str:
         return 'github actions CI'

bert_e/tests/unit/test_github_build_status.py

@@ -227,3 +227,130 @@ def test_skipped_workflow_treated_as_success(client):
     assert workflow_runs.state == 'SUCCESSFUL'
     assert workflow_runs.is_pending() is False
     assert workflow_runs.is_queued() is False
+
+
+def _make_run(*, run_id, sha, branch, status, conclusion, event='push',
+              workflow_id):
+    return {
+        'id': run_id,
+        'head_sha': sha,
+        'head_branch': branch,
+        'status': status,
+        'event': event,
+        'workflow_id': workflow_id,
+        'check_suite_id': run_id,
+        'conclusion': conclusion,
+        'pull_requests': [],
+        'repository': {
+            'full_name': 'org/repo',
+            'owner': {'login': 'org'},
+            'name': 'repo',
+        },
+    }
+
+
+SHARED_SHA = 'aabb' * 10
+W_BRANCH = 'w/4.3/bugfix/BERTE-602'
+
+
+def test_cross_branch_sha_poisoning_state_property(client):
+    """Document the existing .state bug: a sibling branch's SUCCESSFUL run
+    on the same SHA causes .state to return SUCCESSFUL even though the
+    w-branch is still in_progress (the PR 5155 hollow-success shape).
+    """
+    data = {
+        'total_count': 2,
+        'workflow_runs': [
+            _make_run(
+                run_id=1, sha=SHARED_SHA, branch=W_BRANCH,
+                status='in_progress', conclusion=None, workflow_id=10),
+            _make_run(
+                run_id=2, sha=SHARED_SHA, branch='development/4.3',
+                status='completed', conclusion='success', workflow_id=20),
+        ],
+    }
+    runs = AggregatedWorkflowRuns(client, **data)
+    # The existing .state property returns SUCCESSFUL because development/4.3
+    # succeeded — this is the source of the stale-cache poisoning.
+    assert runs.state == 'SUCCESSFUL'
+
+
+def test_state_for_branch_catches_inprogress(client):
+    """state_for_branch filters strictly to the requested branch so an
+    in-progress run is not masked by a sibling branch's success.
+    """
+    data = {
+        'total_count': 2,
+        'workflow_runs': [
+            _make_run(
+                run_id=1, sha=SHARED_SHA, branch=W_BRANCH,
+                status='in_progress', conclusion=None, workflow_id=10),
+            _make_run(
+                run_id=2, sha=SHARED_SHA, branch='development/4.3',
+                status='completed', conclusion='success', workflow_id=20),
+        ],
+    }
+    runs = AggregatedWorkflowRuns(client, **data)
+    assert runs.state_for_branch(W_BRANCH) == 'INPROGRESS'
+    assert runs.state_for_branch('development/4.3') == 'SUCCESSFUL'
+
+
+def test_state_for_branch_no_runs_returns_notstarted(client):
+    data = {
+        'total_count': 1,
+        'workflow_runs': [
+            _make_run(
+                run_id=1, sha=SHARED_SHA, branch='development/4.3',
+                status='completed', conclusion='success', workflow_id=20),
+        ],
+    }
+    runs = AggregatedWorkflowRuns(client, **data)
+    assert runs.state_for_branch(W_BRANCH) == 'NOTSTARTED'
+
+
+def test_state_for_branch_workflow_dispatch_excluded(client):
+    data = {
+        'total_count': 1,
+        'workflow_runs': [
+            _make_run(
+                run_id=1, sha=SHARED_SHA, branch=W_BRANCH,
+                status='completed', conclusion='success',
+                event='workflow_dispatch', workflow_id=10),
+        ],
+    }
+    runs = AggregatedWorkflowRuns(client, **data)
+    assert runs.state_for_branch(W_BRANCH) == 'NOTSTARTED'
+
+
+def test_state_for_branch_deduplicates_within_branch(client):
+    """Same workflow_id on the same branch: keep the best conclusion."""
+    data = {
+        'total_count': 2,
+        'workflow_runs': [
+            _make_run(
+                run_id=1, sha=SHARED_SHA, branch=W_BRANCH,
+                status='completed', conclusion='cancelled', workflow_id=10),
+            _make_run(
+                run_id=2, sha=SHARED_SHA, branch=W_BRANCH,
+                status='completed', conclusion='success', workflow_id=10),
+        ],
+    }
+    runs = AggregatedWorkflowRuns(client, **data)
+    assert runs.state_for_branch(W_BRANCH) == 'SUCCESSFUL'
+
+
+def test_state_for_branch_clean_pr_passes(client):
+    """A PR where the w-branch completed successfully passes revalidation."""
+    data = {
+        'total_count': 2,
+        'workflow_runs': [
+            _make_run(
+                run_id=1, sha=SHARED_SHA, branch=W_BRANCH,
+                status='completed', conclusion='success', workflow_id=10),
+            _make_run(
+                run_id=2, sha=SHARED_SHA, branch=W_BRANCH,
+                status='completed', conclusion='skipped', workflow_id=20),
+        ],
+    }
+    runs = AggregatedWorkflowRuns(client, **data)
+    assert runs.state_for_branch(W_BRANCH) == 'SUCCESSFUL'