deploy: compute-guard BLOCK drop-in (live on Veron since 09-24 00:0xZ)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

api/tests/test_compute_guard.py

@@ -220,3 +220,12 @@ def test_scoped_allow_in_a_real_diff():
 """
     fs = cg.parse_added("windy-pro", diff, ALLOW)
     assert [(f.line, f.match) for f in fs] == [(3, "openrouter.ai")]
+
+
+def test_block_mode_never_blocks_grant_owned(monkeypatch):
+    monkeypatch.setattr(cg, "MODE", "block")
+    g = cg.Finding("src/client/desktop/x.js", 9, "provider host", "api.openai.com")
+    state, desc, f = cg.status_for([], whole_tree=True, grant=[g])
+    assert state == "success" and desc.startswith("⚠ WARN (Grant-owned, not blocking): 1") and f is g
+    lane = cg.Finding("a.py", 3, "provider host", "x")
+    assert cg.status_for([lane], whole_tree=True, grant=[g])[0] == "failure"

api/tests/test_guards_report.py

@@ -65,3 +65,13 @@ def test_render_splits_lane_and_grant_counts():
 def test_windy_pro_root_env_example_is_grant_owned_but_not_the_account_servers():
     assert gr.grant_owned("windy-pro", ".env.example", None, OWNED)
     assert not gr.grant_owned("windy-pro", "account-server/.env.example", None, OWNED)
+
+
+def test_split_grant_sends_desktop_code_to_grant(monkeypatch):
+    F = gr.cg.Finding
+    fs = [F("src/client/desktop/main.js", 3, "provider host", "x"),
+          F("account-server/src/llm.ts", 5, "provider host", "y")]
+    lane, grant = gr.split_grant("windy-pro", "a" * 40, fs)
+    assert [f.path for f in grant] == ["src/client/desktop/main.js"]
+    assert [f.path for f in lane] == ["account-server/src/llm.ts"]
+    assert gr.split_grant("windy-chat", "a" * 40, fs) == (fs, [])

api/tests/test_pr_status_bridge.py

@@ -351,10 +351,12 @@ class _Guard:
         return self.findings
 
     @staticmethod
-    def status_for(findings, whole_tree):
+    def status_for(findings, whole_tree, grant=()):
+        if not findings and grant:
+            return "success", f"GRANT-WARN {len(grant)}", grant[0]
         if not findings:
             return "success", "OK: clean", None
-        return "success", f"WARN {len(findings)}", findings[0]
+        return "failure", f"BLOCK {len(findings)}", findings[0]
 
 
 class _F:
@@ -366,12 +368,12 @@ def test_guard_posts_warn_with_a_link_to_the_first_finding(fake, monkeypatch):
     monkeypatch.setitem(sys.modules, "compute_guard", _Guard([_F()]))
     bridge.post_compute_guard("windy-chat", SHA, "main", False)
     assert [(p["context"], p["state"], p["description"]) for p in f.posted] == [
-        ("windy-git/compute-guard", "success", "WARN 1")]
+        ("windy-git/compute-guard", "failure", "BLOCK 1")]
     assert f.posted[0]["target_url"].endswith(f"/src/commit/{SHA}/app/llm.py#L7")
 
 
 def test_guard_same_status_is_not_reposted(fake, monkeypatch):
-    f = fake(statuses=[{"context": "windy-git/compute-guard", "state": "success", "description": "WARN 1"}])
+    f = fake(statuses=[{"context": "windy-git/compute-guard", "state": "failure", "description": "BLOCK 1"}])
     monkeypatch.setitem(sys.modules, "compute_guard", _Guard([_F()]))
     bridge.post_compute_guard("windy-chat", SHA, "main", False)
     assert f.posted == []
@@ -385,11 +387,11 @@ def test_guard_that_cannot_run_posts_nothing(fake, monkeypatch):
 
 
 def test_ci_hygiene_posts_under_its_own_context(fake, monkeypatch):
-    f = fake(statuses=[{"context": "windy-git/compute-guard", "state": "success", "description": "WARN 1"}])
+    f = fake(statuses=[{"context": "windy-git/compute-guard", "state": "failure", "description": "BLOCK 1"}])
     monkeypatch.setitem(sys.modules, "ci_hygiene", _Guard([_F()]))
     bridge.post_ci_hygiene("windy-chat", SHA, "main", True)
     # the compute-guard status with the same description must not suppress it
-    assert [(p["context"], p["description"]) for p in f.posted] == [("windy-git/ci-hygiene", "WARN 1")]
+    assert [(p["context"], p["description"]) for p in f.posted] == [("windy-git/ci-hygiene", "BLOCK 1")]
 
 
 def test_retargeted_pr_gets_a_fresh_mirror_on_the_new_base(fake):
@@ -428,3 +430,24 @@ def test_named_no_daemon_job_is_not_posted_for_that_repo_only(fake, monkeypatch)
 def test_default_no_daemon_named_is_empty():
     """eternitas converted its ci/build to a no-Docker ci/smoke (#179); nothing left."""
     assert bridge.NO_DAEMON_NAMED == {}
+
+
+def test_grant_owned_findings_never_block(fake, monkeypatch):
+    """Orchestrator 09-23: compute-guard blocks lane-owned code only."""
+    f = fake()
+    monkeypatch.setitem(sys.modules, "compute_guard", _Guard([_F()]))
+    monkeypatch.setitem(sys.modules, "guards_report",
+                        type("GR", (), {"split_grant": staticmethod(lambda r, s, fs: ([], list(fs)))}))
+    bridge.post_compute_guard("windy-pro", SHA, "main", True)
+    assert [(p["state"], p["description"]) for p in f.posted] == [("success", "GRANT-WARN 1")]
+
+
+def test_failed_grant_split_warns_instead_of_blocking(fake, monkeypatch):
+    def boom(*a):
+        raise RuntimeError("no bare clone")
+
+    f = fake()
+    monkeypatch.setitem(sys.modules, "compute_guard", _Guard([_F()]))
+    monkeypatch.setitem(sys.modules, "guards_report", type("GR", (), {"split_grant": staticmethod(boom)}))
+    bridge.post_compute_guard("windy-pro", SHA, "main", True)
+    assert [p["state"] for p in f.posted] == ["success"]

deploy/systemd/windygit-sync.service.d/compute-guard-block.conf

@@ -0,0 +1,3 @@
+[Service]
+# Orchestrator 09-23: compute-guard BLOCKS lane-owned findings; Grant-owned (ci/grant-owned.yml) stay WARN.
+Environment=COMPUTE_GUARD_MODE=block

scripts/ci_hygiene.py

@@ -215,8 +215,13 @@ def _check(repo: str, sha: str, default_branch: str, is_default_head: bool):
                           lambda: cg.scan_added(repo, bare, f"refs/heads/{default_branch}", sha, allow, **kw))
 
 
-def status_for(findings, whole_tree: bool):
+def status_for(findings, whole_tree: bool, grant=()):
+    """Same contract as compute_guard.status_for: `grant` findings never block."""
     scope = "in CI/Dockerfiles" if whole_tree else "added"
+    if not findings and grant:
+        g, n = grant[0], len(grant)
+        desc = f"⚠ WARN (Grant-owned, not blocking): {n} CI hygiene issue{'s' if n > 1 else ''} {scope}, e.g. {g.path}:{g.line} {g.match}"
+        return "success", desc[:140], g
     if not findings:
         return "success", f"OK: no floating install or host-port service {scope}", None
     f = findings[0]

scripts/compute_guard.py

@@ -260,9 +260,19 @@ def check(repo: str, sha: str, default_branch: str, is_default_head: bool) -> li
     )
 
 
-def status_for(findings: list[Finding], whole_tree: bool) -> tuple[str, str, Finding | None]:
+def status_for(findings: list[Finding], whole_tree: bool,
-    """(state, description, first finding) for the GitHub commit status."""
+               grant: list[Finding] = ()) -> tuple[str, str, Finding | None]:
+    """(state, description, first finding) for the GitHub commit status.
+
+    `findings` = lane-owned (these block in MODE=block); `grant` = findings in
+    Grant-owned code (ci/grant-owned.yml): always WARN, never red (orchestrator
+    09-23: his desktop work is never blocked by us)."""
     scope = "in tree" if whole_tree else "added"
+    if not findings and grant:
+        g, n = grant[0], len(grant)
+        desc = (f"⚠ WARN (Grant-owned, not blocking): {n} direct AI-provider use{'s' if n > 1 else ''} "
+                f"{scope}, e.g. {g.path}:{g.line} {g.match}")
+        return "success", desc[:140], g
     if not findings:
         what = "no direct AI-provider use in tree" if whole_tree else "no direct AI-provider use added"
         return "success", f"OK: {what} (Windy Mind is the only door)", None

scripts/guards_report.py

@@ -75,6 +75,25 @@ def grant_owned(repo: str, path: str, job: str | None, owned: list[dict]) -> boo
     return False
 
 
+def split_grant(repo: str, sha: str, findings: list) -> tuple[list, list]:
+    """(lane-owned, Grant-owned) findings at `sha`, by ci/grant-owned.yml, with
+    workflow lines attributed to their job exactly as the status page does."""
+    owned = (yaml.safe_load(OWNED.read_text()) or {}).get("grant_owned") or []
+    if not any(e["repo"] == repo for e in owned):
+        return list(findings), []
+    bare = cg.WORK / f"{repo}.git"
+    texts: dict[str, str] = {}
+    lane, grant = [], []
+    for f in findings:
+        job = None
+        if "/workflows/" in f.path:
+            if f.path not in texts:
+                texts[f.path] = cg._git(bare, "show", f"{sha}:{f.path}")
+            job = job_of(texts[f.path], f.line)
+        (grant if grant_owned(repo, f.path, job, owned) else lane).append(f)
+    return lane, grant
+
+
 def scan(repo: str, owned: list[dict]):
     bare = cg.WORK / f"{repo}.git"
     if not bare.is_dir():

scripts/pr_status_bridge.py

@@ -395,7 +395,14 @@ def _post_guard(modname: str, ctx: str, repo: str, sha: str, default_branch: str
         return
     if findings is None:
         return
-    state, desc, first = g.status_for(findings, whole_tree=is_default_head)
+    try:  # Grant-owned code never blocks (orchestrator 09-23); lazy like the guards
+        import importlib
+
+        lane, grant = importlib.import_module("guards_report").split_grant(repo, sha, findings)
+    except Exception as e:  # noqa: BLE001 — can't tell whose code: warn, never block
+        print(f"  {repo}@{sha[:7]} {ctx}: Grant-owned split failed ({type(e).__name__}); WARN only")
+        lane, grant = [], list(findings)
+    state, desc, first = g.status_for(lane, whole_tree=is_default_head, grant=grant)
     st, existing = github("GET", f"/repos/{GH_OWNER}/{repo}/commits/{sha}/statuses?per_page=100")
     for s in existing or []:  # newest first: compare the latest guard status only
         if s["context"] == ctx: