fix: skip stale auto optimization tasks

Author: Pigbibi

scripts/prepare_auto_optimization_pr.py

@@ -11,6 +11,7 @@
 SUMMARY_RE = re.compile(r"^\s+- Summary: (?P<summary>.+)$")
 SOURCE_RE = re.compile(r"^\s+- Source: \[(?P<label>.+?)\]\((?P<url>[^)]+)\)$")
 MARKER_PREFIX = "<!-- auto-optimization-pr:issue-"
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
 
 
 def parse_actions(issue_body: str) -> list[dict[str, Any]]:
@@ -58,15 +59,56 @@ def parse_actions(issue_body: str) -> list[dict[str, Any]]:
     return actions
 
 
-def build_payload(issue_context: dict[str, Any]) -> dict[str, Any]:
+def _read_text(path: Path) -> str:
+    return path.read_text(encoding="utf-8") if path.exists() else ""
+
+
+def _is_completed_low_risk_task(action: dict[str, Any], repo_root: Path) -> bool:
+    title = str(action.get("title", "")).lower()
+    repo_name = repo_root.name
+
+    if repo_name == "CryptoLeaderRotation":
+        if "shadow/challenger build generation" in title:
+            workflow = _read_text(repo_root / ".github" / "workflows" / "monthly_publish.yml")
+            return "run_monthly_shadow_build.py" in workflow
+        if "deterministic tie-break behavior" in title:
+            readme = _read_text(repo_root / "README.md")
+            runbook = _read_text(repo_root / "docs" / "operator_runbook.md")
+            return (
+                "Monthly ranking tie-break rule for `core_major` live exports:" in readme
+                and "deterministic tie-break" in runbook
+            )
+
+    if repo_name == "BinancePlatform" and "zero-trade diagnostics" in title:
+        monthly_report = _read_text(repo_root / "scripts" / "run_monthly_report_bundle.py")
+        return (
+            "No explicit gating or no-trade reasons were recorded this month." in monthly_report
+            and "gating_summary" in monthly_report
+        )
+
+    return False
+
+
+def build_payload(issue_context: dict[str, Any], repo_root: Path | None = None) -> dict[str, Any]:
+    repo_root = repo_root or PROJECT_ROOT
     issue_number = int(issue_context["number"])
     issue_title = str(issue_context["title"]).strip()
     issue_body = str(issue_context["body"])
     parsed_actions = parse_actions(issue_body)
-    safe_actions = [
-        action for action in parsed_actions
-        if action["risk_level"] == "low" and "auto-pr-safe" in action.get("flags", [])
+    low_safe_actions = [
+        action
+        for action in parsed_actions
+        if action["risk_level"] == "low"
+        and "auto-pr-safe" in action.get("flags", [])
+        and "experiment-only" not in action.get("flags", [])
     ]
+    safe_actions: list[dict[str, Any]] = []
+    skipped_actions: list[dict[str, Any]] = []
+    for action in low_safe_actions:
+        if _is_completed_low_risk_task(action, repo_root):
+            skipped_actions.append({**action, "skip_reason": "already_implemented"})
+        else:
+            safe_actions.append(action)
     return {
         "issue_number": issue_number,
         "issue_title": issue_title,
@@ -75,7 +117,9 @@ def build_payload(issue_context: dict[str, Any]) -> dict[str, Any]:
         "pr_title": f"Draft: address monthly optimization issue #{issue_number}",
         "should_run": bool(safe_actions),
         "safe_task_count": len(safe_actions),
+        "skipped_task_count": len(skipped_actions),
         "safe_actions": safe_actions,
+        "skipped_actions": skipped_actions,
     }
 
 
@@ -86,8 +130,19 @@ def render_task_summary(payload: dict[str, Any]) -> str:
         f"- Issue: #{payload['issue_number']} {payload['issue_title']}",
         f"- Eligible low-risk auto-pr-safe tasks: `{payload['safe_task_count']}`",
     ]
+    if payload["skipped_actions"]:
+        lines.append(f"- Skipped as already implemented: `{payload['skipped_task_count']}`")
     if not payload["safe_actions"]:
-        lines.extend(["", "No eligible low-risk [auto-pr-safe] tasks were found in this issue."])
+        lines.extend(["", "No eligible low-risk [auto-pr-safe] tasks remain for draft PR generation."])
+        if payload["skipped_actions"]:
+            lines.extend(["", "## Skipped Tasks"])
+            for action in payload["skipped_actions"]:
+                lines.extend(
+                    [
+                        f"- `{action['risk_level']}` {action['title']}",
+                        f"  - Reason: {action['skip_reason']}",
+                    ]
+                )
         return "\n".join(lines).strip() + "\n"
 
     lines.extend(["", "## Selected Tasks"])

tests/test_prepare_auto_optimization_pr.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import tempfile
 import unittest
+from pathlib import Path
 
 from scripts.prepare_auto_optimization_pr import build_payload, parse_actions, render_pr_body
 
@@ -9,19 +11,19 @@ class PrepareAutoOptimizationPrTests(unittest.TestCase):
     def setUp(self) -> None:
         self.issue_context = {
             "number": 15,
-            "title": "Monthly Optimization Tasks · BinancePlatform: 2026-04-01 / 2026-03",
-            "body": """# Monthly Optimization Tasks · BinancePlatform
+            "title": "Monthly Optimization Tasks · Sandbox",
+            "body": """# Monthly Optimization Tasks · Sandbox
 
 ## Actions
-- [ ] `high` Reconcile March cash flows and open-position state
-  - Summary: Pull Binance transaction history for March.
-  - Source: [QuantStrategyLab/BinancePlatform #9](https://github.com/QuantStrategyLab/BinancePlatform/issues/9)
+- [ ] `high` Investigate an upstream issue
+  - Summary: Manual follow-up only.
+  - Source: [Sandbox #0](https://example.com/issues/0)
 - [ ] `low` Add zero-trade diagnostics to the report [auto-pr-safe]
   - Summary: Include the top failed gating reason counts.
-  - Source: [QuantStrategyLab/BinancePlatform #9](https://github.com/QuantStrategyLab/BinancePlatform/issues/9)
+  - Source: [Sandbox #1](https://example.com/issues/1)
 - [ ] `low` Add a boundary tracker [auto-pr-safe, experiment-only]
   - Summary: Track near-cutoff symbols monthly.
-  - Source: [QuantStrategyLab/CryptoLeaderRotation #11](https://github.com/QuantStrategyLab/CryptoLeaderRotation/issues/11)
+  - Source: [Sandbox #2](https://example.com/issues/2)
 """,
         }
 
@@ -32,18 +34,21 @@ def test_parse_actions_preserves_risk_flags_and_source(self) -> None:
         self.assertEqual(actions[0]["risk_level"], "high")
         self.assertEqual(actions[1]["flags"], ["auto-pr-safe"])
         self.assertEqual(actions[2]["flags"], ["auto-pr-safe", "experiment-only"])
-        self.assertEqual(actions[2]["source_label"], "QuantStrategyLab/CryptoLeaderRotation #11")
+        self.assertEqual(actions[2]["source_label"], "Sandbox #2")
 
-    def test_build_payload_selects_only_low_auto_pr_safe_actions(self) -> None:
-        payload = build_payload(self.issue_context)
+    def test_build_payload_selects_only_non_experiment_low_auto_pr_safe_actions(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            payload = build_payload(self.issue_context, repo_root=Path(temp_dir))
 
         self.assertTrue(payload["should_run"])
-        self.assertEqual(payload["safe_task_count"], 2)
+        self.assertEqual(payload["safe_task_count"], 1)
+        self.assertEqual(payload["skipped_task_count"], 0)
         self.assertEqual(payload["branch_name"], "automation/monthly-optimization-issue-15")
         self.assertEqual(payload["safe_actions"][0]["title"], "Add zero-trade diagnostics to the report")
 
     def test_render_pr_body_contains_marker_and_issue_reference(self) -> None:
-        payload = build_payload(self.issue_context)
+        with tempfile.TemporaryDirectory() as temp_dir:
+            payload = build_payload(self.issue_context, repo_root=Path(temp_dir))
         body = render_pr_body(payload)
 
         self.assertIn("<!-- auto-optimization-pr:issue-15 -->", body)