fix: detect grok background work from pinned status

2 files changed, 124 insertions, 2 deletions

diff --git a/scripts/agent_detection_manifest_check.py b/scripts/agent_detection_manifest_check.py
index d1346a94..f180318b 100644
--- a/scripts/agent_detection_manifest_check.py
+++ b/scripts/agent_detection_manifest_check.py
@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
 import re
 import sys
 import tomllib
@@ -39,9 +40,12 @@ REGION_RE = re.compile(
     r"before_current_prompt_marker|current_prompt_block_marker|after_current_prompt_block_marker|"
     r"prompt_box_body|above_prompt_box|last_non_empty_above_prompt_box|after_last_horizontal_rule|"
     r"osc_title|osc_progress|"
-    r"bottom_lines\([1-9][0-9]*\)|bottom_non_empty_lines\([1-9][0-9]*\))$"
+    r"bottom_lines\([1-9][0-9]*\)|bottom_non_empty_lines\([1-9][0-9]*\)|"
+    r"top_non_empty_lines\([1-9][0-9]*\))$"
 )
+REGION_COUNT_RE = re.compile(r"\(([1-9][0-9]*)\)$")
 VERSION_RE = re.compile(r"^[0-9]+(?:\.[0-9]+)*$")
+MAX_TOP_REGION_LINE_COUNT = 65_535
 MAX_RULES_PER_MANIFEST = 128
 MAX_GATE_DEPTH = 8
 MAX_TOTAL_GATES = 512
@@ -49,6 +53,17 @@ MAX_MATCHERS_PER_GATE = 32
 MAX_TOTAL_MATCHERS = 1024
 MAX_MATCHER_CHARS = 512
 
+# Keep engine-2 clients on the OSC-capable manifest until an engine-3 release
+# can consume top_non_empty_lines. Remove this entry when the website publishes
+# the bundled Grok manifest.
+STAGED_WEBSITE_MANIFESTS = {
+    "grok": (
+        "2026.07.16.2",
+        "2026.07.16.1",
+        "1f35b3271a96cf830c64bed78751619bfd8013c277c0d7c0f999b7a433895f28",
+    ),
+}
+
 
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description=__doc__)
@@ -136,6 +151,11 @@ def validate_manifest(path: Path, engine_version: int) -> dict:
     complexity = {"gates": 0, "matchers": 0}
     for index, rule in enumerate(rules):
         validate_rule(path, index, rule, complexity)
+        region = rule.get("region", "whole_recent")
+        if region.startswith("top_non_empty_lines(") and min_engine < 3:
+            raise CheckError(
+                f"{path}: rule {rule['id']} region {region!r} requires min_engine_version 3"
+            )
 
     return manifest
 
@@ -155,6 +175,13 @@ def validate_rule(path: Path, index: int, rule: object, complexity: dict[str, in
     region = rule.get("region", "whole_recent")
     if not isinstance(region, str) or not REGION_RE.fullmatch(region):
         raise CheckError(f"{path}: rule {rule_id} has invalid region {region!r}")
+    count_match = REGION_COUNT_RE.search(region)
+    if (
+        region.startswith("top_non_empty_lines(")
+        and count_match
+        and int(count_match.group(1)) > MAX_TOP_REGION_LINE_COUNT
+    ):
+        raise CheckError(f"{path}: rule {rule_id} has invalid region {region!r}")
     if rule.get("skip_state_update"):
         if state != "unknown":
             raise CheckError(f"{path}: rule {rule_id} skip_state_update requires state unknown")
@@ -295,7 +322,15 @@ def validate_catalog(
 
         bundled_path, bundled_manifest = bundled[agent_id]
         cmp = compare_versions(manifest["version"], bundled_manifest["version"], manifest_path)
-        if cmp < 0:
+        staged_manifest = STAGED_WEBSITE_MANIFESTS.get(agent_id)
+        website_digest = hashlib.sha256(manifest_path.read_bytes()).hexdigest()
+        stages_new_engine_manifest = (
+            staged_manifest
+            == (bundled_manifest["version"], manifest["version"], website_digest)
+            and bundled_manifest["min_engine_version"] == engine_version
+            and manifest["min_engine_version"] < bundled_manifest["min_engine_version"]
+        )
+        if cmp < 0 and not stages_new_engine_manifest:
             raise CheckError(
                 f"{manifest_path}: website version {manifest['version']} is lower than bundled "
                 f"{bundled_manifest['version']} in {bundled_path}"
diff --git a/scripts/test_agent_detection_manifest_check.py b/scripts/test_agent_detection_manifest_check.py
index 3ab69b55..a9ec38d1 100644
--- a/scripts/test_agent_detection_manifest_check.py
+++ b/scripts/test_agent_detection_manifest_check.py
@@ -27,6 +27,21 @@ path = "{path}"
 '''
 
 
+def staged_grok_dirs(root: Path) -> tuple[Path, Path]:
+    bundled = root / "bundled"
+    website = root / "website"
+    bundled.mkdir()
+    website.mkdir()
+    (bundled / "grok.toml").write_bytes(
+        (check.DEFAULT_BUNDLED_DIR / "grok.toml").read_bytes()
+    )
+    (website / "grok.toml").write_bytes(
+        (check.DEFAULT_WEBSITE_DIR / "grok.toml").read_bytes()
+    )
+    (website / "index.toml").write_text(catalog("grok", "grok.toml"))
+    return bundled, website
+
+
 class AgentDetectionManifestCheckTests(unittest.TestCase):
     def test_validates_bundled_and_matching_website_catalog(self):
         with tempfile.TemporaryDirectory() as tmp:
@@ -58,6 +73,41 @@ class AgentDetectionManifestCheckTests(unittest.TestCase):
             with self.assertRaisesRegex(check.CheckError, "lower than bundled"):
                 check.validate_catalog(website, bundled_manifests, engine_version=1)
 
+    def test_allows_explicitly_staged_website_manifest(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            bundled, website = staged_grok_dirs(Path(tmp))
+
+            bundled_manifests = check.load_manifest_dir(bundled, engine_version=3)
+            check.validate_catalog(website, bundled_manifests, engine_version=3)
+
+    def test_rejects_mutated_staged_website_manifest(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            bundled, website = staged_grok_dirs(Path(tmp))
+            with (website / "grok.toml").open("a") as manifest_file:
+                manifest_file.write("\n# unexpected mutation\n")
+
+            bundled_manifests = check.load_manifest_dir(bundled, engine_version=3)
+            with self.assertRaisesRegex(check.CheckError, "lower than bundled"):
+                check.validate_catalog(website, bundled_manifests, engine_version=3)
+
+    def test_rejects_unlisted_website_manifest_lag_for_new_engine(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            bundled = root / "bundled"
+            website = root / "website"
+            bundled.mkdir()
+            website.mkdir()
+            bundled_content = manifest("codex", "2026.06.10.2").replace(
+                "min_engine_version = 1", "min_engine_version = 2"
+            )
+            (bundled / "codex.toml").write_text(bundled_content)
+            (website / "codex.toml").write_text(manifest("codex", "2026.06.10.1"))
+            (website / "index.toml").write_text(catalog())
+
+            bundled_manifests = check.load_manifest_dir(bundled, engine_version=2)
+            with self.assertRaisesRegex(check.CheckError, "lower than bundled"):
+                check.validate_catalog(website, bundled_manifests, engine_version=2)
+
     def test_rejects_same_version_content_drift(self):
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp)
@@ -101,6 +151,43 @@ class AgentDetectionManifestCheckTests(unittest.TestCase):
             with self.assertRaisesRegex(check.CheckError, "exceeds engine"):
                 check.load_manifest_dir(bundled, engine_version=1)
 
+    def test_rejects_top_non_empty_lines_below_engine_three(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            bundled = Path(tmp) / "bundled"
+            bundled.mkdir()
+            content = manifest("codex", "2026.06.10.1").replace(
+                'contains = ["ready"]',
+                'region = "top_non_empty_lines(1)"\ncontains = ["ready"]',
+            )
+            (bundled / "codex.toml").write_text(content)
+
+            with self.assertRaisesRegex(check.CheckError, "requires min_engine_version 3"):
+                check.load_manifest_dir(bundled, engine_version=3)
+
+    def test_top_non_empty_lines_requires_canonical_positive_bounded_count(self):
+        base_rule = {
+            "id": "test",
+            "state": "working",
+            "contains": ["ready"],
+        }
+        name = "top_non_empty_lines"
+        for count in ("1", str(check.MAX_TOP_REGION_LINE_COUNT)):
+            rule = {**base_rule, "region": f"{name}({count})"}
+            check.validate_rule(Path("test.toml"), 0, rule, {"gates": 0, "matchers": 0})
+        for count in (
+            "0",
+            "01",
+            "+1",
+            str(check.MAX_TOP_REGION_LINE_COUNT + 1),
+            "9" * 40,
+        ):
+            rule = {**base_rule, "region": f"{name}({count})"}
+            with self.subTest(region=rule["region"]):
+                with self.assertRaisesRegex(check.CheckError, "invalid region"):
+                    check.validate_rule(
+                        Path("test.toml"), 0, rule, {"gates": 0, "matchers": 0}
+                    )
+
 
 if __name__ == "__main__":
     unittest.main()