Fix stale workflow history in fleet observation

fleet/github.py

@@ -88,11 +88,20 @@ def content_text(payload: dict[str, Any]) -> str:
         raise GitHubError(f"Unsupported GitHub contents encoding: {encoding}")
 
     def node_snapshot(self, repository: str, branch: str, history: int) -> dict[str, Any]:
-        encoded_ref = urllib.parse.quote(branch, safe="")
         workflows = self.get_json(f"repos/{repository}/actions/workflows?per_page=100")
+        # GitHub's server-side `branch=` filter has intermittently returned stale
+        # workflow-run history for otherwise current public repositories. Fetch the
+        # recent repository history without that filter and apply the branch match
+        # locally using each run's explicit head_branch instead.
         runs = self.get_json(
-            f"repos/{repository}/actions/runs?branch={encoded_ref}&per_page={history}"
+            f"repos/{repository}/actions/runs?per_page={history}"
         )
+        raw_runs = runs.get("workflow_runs", []) if isinstance(runs, dict) else []
+        branch_runs = [
+            run
+            for run in raw_runs
+            if isinstance(run, dict) and run.get("head_branch") == branch
+        ]
         config = self.get_content(repository, "config.py", branch)
         core = self.get_content(repository, "oswm_codebase", branch)
         managed = self.get_content(repository, ".oswm-managed-files.json", branch)
@@ -101,7 +110,7 @@ def node_snapshot(self, repository: str, branch: str, history: int) -> dict[str,
             "core_sha": core.get("sha"),
             "managed_text": self.content_text(managed),
             "workflows": workflows.get("workflows", []) if isinstance(workflows, dict) else [],
-            "runs": runs.get("workflow_runs", []) if isinstance(runs, dict) else [],
+            "runs": branch_runs,
         }
 
     def probe_url(self, url: str) -> dict[str, Any]:

tests/test_fleet_management.py

@@ -8,6 +8,7 @@
 import pytest
 import yaml
 
+from fleet.github import GitHubClient
 from fleet.reconcile import markdown_summary, previous_scheduled_time, reconcile
 from fleet.registry import load_registry
 
@@ -265,6 +266,52 @@ def test_fleet_workflows_are_parseable_and_secretless():
     assert "${{ secrets" not in status_source
 
 
+def test_node_snapshot_filters_run_branch_locally_without_server_branch_query():
+    class RecordingClient(GitHubClient):
+        def __init__(self):
+            super().__init__(token=None)
+            self.paths = []
+
+        def get_json(self, path):
+            self.paths.append(path)
+            if "actions/workflows" in path:
+                return {"workflows": []}
+            if "actions/runs" in path:
+                return {
+                    "workflow_runs": [
+                        {
+                            "id": 1,
+                            "head_branch": "main",
+                            "path": ".github/workflows/data_daily_updating.yml",
+                        },
+                        {
+                            "id": 2,
+                            "head_branch": "feature/test",
+                            "path": ".github/workflows/data_daily_updating.yml",
+                        },
+                    ]
+                }
+            raise AssertionError(path)
+
+        def get_content(self, repository, path, branch):
+            payloads = {
+                "config.py": {"content": "", "encoding": "utf-8"},
+                "oswm_codebase": {"sha": REVISION},
+                ".oswm-managed-files.json": {
+                    "content": '{"managed_revision": 2}',
+                    "encoding": "utf-8",
+                },
+            }
+            return payloads[path]
+
+    client = RecordingClient()
+    snapshot = client.node_snapshot("example/test-node", "main", 20)
+
+    run_paths = [path for path in client.paths if "actions/runs" in path]
+    assert run_paths == ["repos/example/test-node/actions/runs?per_page=20"]
+    assert [run["id"] for run in snapshot["runs"]] == [1]
+
+
 def test_registry_toml_is_standard_library_parseable():
     data = tomllib.loads((ROOT / "fleet/registry.toml").read_text())
     assert data["schema_version"] == 1