Protect the branch a split pushed, and stop on a write cap or breaker

ion/execute/split.py

@@ -5,7 +5,7 @@
 from pathlib import Path
 
 from .. import proposals
-from ..util import Refused, SafetyError, filter_repo_path
+from ..util import Refused, SafetyError, Waiting, filter_repo_path
 from ..writes import WriteRequest
 from .base import Executor, Plan, Step, filter_args
 from .provenance import origin_callback
@@ -59,9 +59,11 @@ def verify():
 
         def policy():
             notes = []
-            for f in self._policy_fixes(N, private):
+            for f in self._policy_fixes(N, private, base):
                 try:
                     notes.append(apply_policy_fix(self.ctx, f, pid))
+                except (Waiting, Refused):
+                    raise           # a write cap defers the step to the next run; the breaker or kill switch stops it
                 except Exception as e:  # noqa: BLE001
                     notes.append(f"{f['class']}: {e}")
             return "; ".join(notes) or "nothing to apply"
@@ -77,12 +79,12 @@ def policy():
             Step("apply-policy", f"apply the org policy to {N}", None, policy),
         ])
 
-    def _policy_fixes(self, repo: str, private: bool) -> list[dict]:
+    def _policy_fixes(self, repo: str, private: bool, branch: str) -> list[dict]:
+        """Policy for the new repo. Protection goes on `branch`, the branch the history was pushed to."""
         pol = self.cfg["policy"]
         fixes = []
         if pol.get("branch_protection") and not (private and self.cfg.get("plan", "free") == "free"):
-            fixes.append({"class": "branch_protection", "repo": repo, "branch": self.cfg["policy"]["default_branch"],
-                          "settings": pol["branch_protection"]})
+            fixes.append({"class": "branch_protection", "repo": repo, "branch": branch, "settings": pol["branch_protection"]})
         if pol.get("dependabot_alerts"):
             fixes.append({"class": "dependabot_alerts", "repo": repo})
         if pol.get("secret_scanning") == "public_only" and not private:

tests/test_execute.py

@@ -282,6 +282,47 @@ def test_split_verify_failure_stops_before_removal(ctx, world, monkeypatch):
     assert not any(b.endswith("-remove") for b in branches(world, "mono"))
 
 
+def test_split_protects_the_branch_it_pushed(ctx, world):
+    g = str(world / "mono.git")
+    sh("git", "--git-dir", g, "branch", "-m", "main", "master")
+    sh("git", "--git-dir", g, "symbolic-ref", "HEAD", "refs/heads/master")
+    p = prep(ctx, "split", {**SPLIT, "source_branch": "master"})                # policy default_branch is still main
+    executor_for(ctx, "split").apply(p, dry_run=False)
+    assert branches(world, "mono-engine") == ["master"]
+    assert ctx.gh.writes_to("PUT", "/repos/acme/mono-engine/branches/master/protection")
+    assert not ctx.gh.writes_to("PUT", "/repos/acme/mono-engine/branches/main/protection")
+
+
+def test_split_policy_step_waits_for_write_cap_instead_of_skipping(ctx, world):
+    p = prep(ctx, "split", SPLIT)
+    ex = executor_for(ctx, "split")
+    ctx.cfg["writes"] = {"caps": {"put_branch_protection": 0}}
+    res = ex.apply(p, dry_run=False)
+    assert res["waiting"] and "put_branch_protection" in res["waiting"]
+    assert not ctx.db.ledger_done(p["id"], "create.apply-policy")
+    assert proposals.get(ctx.db, p["id"])["state"].get("phase", "create") == "create"
+    ctx.cfg["writes"] = {}                                                       # next run: the cap is fresh
+    ctx.writes.reset()
+    ex.apply(proposals.get(ctx.db, p["id"]), dry_run=False)
+    assert ctx.gh.writes_to("PUT", "/repos/acme/mono-engine/branches/main/protection")
+    assert ctx.db.ledger_done(p["id"], "create.apply-policy")
+    assert proposals.get(ctx.db, p["id"])["state"]["phase"] == "pointer"
+
+
+def test_split_policy_step_does_not_swallow_a_tripped_breaker(ctx, world, monkeypatch):
+    import ion.execute.split as split_mod
+    from ion.writes import BreakerTripped
+
+    def tripped(*a, **k):
+        raise BreakerTripped("the run would touch too many repos")
+    monkeypatch.setattr(split_mod, "apply_policy_fix", tripped)
+    p = prep(ctx, "split", SPLIT)
+    with pytest.raises(BreakerTripped):
+        executor_for(ctx, "split").apply(p, dry_run=False)
+    assert not ctx.db.ledger_done(p["id"], "create.apply-policy")
+    assert proposals.get(ctx.db, p["id"])["status"] == "failed"
+
+
 # ---- others ---------------------------------------------------------------------------------
 
 def test_archive_executor_never_deletes(ctx):