Keep challenger DB limits compatible with OOS retrieval

scripts/run_leo_agent_challenger_loop.py

@@ -1183,11 +1183,38 @@ print(json.dumps(run_suite(), sort_keys=True, default=str))
 '''
 
 
+def _patch_db_context_limits(source: str) -> str:
+    constant_limits = {
+        "CONTEXT_CLAIM_LIMIT": 4,
+        "CONTEXT_ROW_LIMIT": 6,
+    }
+    constant_patterns = {name: rf"(?m)^{name}\s*=\s*\d+\s*$" for name in constant_limits}
+    matched_constants = {name: bool(re.search(pattern, source)) for name, pattern in constant_patterns.items()}
+    if any(matched_constants.values()):
+        if not all(matched_constants.values()):
+            raise RuntimeError("database context source has an incomplete named-limit contract")
+        patched = source
+        for name, value in constant_limits.items():
+            patched, count = re.subn(constant_patterns[name], f"{name} = {value}", patched, count=1)
+            if count != 1:
+                raise RuntimeError(f"database context source has an ambiguous {name} contract")
+        return patched
+
+    replacements = (
+        ('"--limit",\n        "0",', '"--limit",\n        "4",'),
+        ('"--context-limit",\n        "0",', '"--context-limit",\n        "6",'),
+    )
+    patched = source
+    for old, new in replacements:
+        if patched.count(old) != 1:
+            raise RuntimeError("database context source no longer matches the legacy bounded-limit contract")
+        patched = patched.replace(old, new, 1)
+    return patched
+
+
 def _patched_db_context_source() -> str:
     source = DB_CONTEXT_PLUGIN.read_text(encoding="utf-8")
-    patched = source.replace('"--limit",\n        "0",', '"--limit",\n        "4",').replace(
+    patched = _patch_db_context_limits(source)
-        '"--context-limit",\n        "0",', '"--context-limit",\n        "6",'
-    )
     patched = patched.replace(
         'def _pre_llm_call(**kwargs: Any) -> dict[str, str]:\n    user_message = str(kwargs.get("user_message") or "")\n',
         "def _pre_llm_call(**kwargs: Any) -> dict[str, str]:\n"

tests/test_run_leo_agent_challenger_loop.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import ast
 import json
 
 import pytest
@@ -7,6 +8,18 @@ import pytest
 from scripts import run_leo_agent_challenger_loop as runner
 
 
+def _module_int_constant(source: str, name: str) -> int:
+    for node in ast.parse(source).body:
+        if not isinstance(node, ast.Assign) or len(node.targets) != 1:
+            continue
+        target = node.targets[0]
+        if isinstance(target, ast.Name) and target.id == name:
+            value = ast.literal_eval(node.value)
+            if isinstance(value, int) and not isinstance(value, bool):
+                return value
+    raise AssertionError(f"missing integer module constant: {name}")
+
+
 def test_remote_script_embeds_stateless_challenger_read_only_guard_and_bounded_context() -> None:
     script = runner.build_remote_script("abc123", challenger_max_tokens=512)
 
@@ -14,8 +27,8 @@ def test_remote_script_embeds_stateless_challenger_read_only_guard_and_bounded_c
     assert "gatewayrunner_temp_profile_no_model_tools" in script
     assert "blocked by isolated challenger-loop read-only guard" in script
     context_source = runner._patched_db_context_source()
-    assert '"--limit",\n        "4",' in context_source
+    assert _module_int_constant(context_source, "CONTEXT_CLAIM_LIMIT") == 4
-    assert '"--context-limit",\n        "6",' in context_source
+    assert _module_int_constant(context_source, "CONTEXT_ROW_LIMIT") == 6
     assert "CHALLENGER_MAX_TOKENS = 512" in script
     assert "posted_to_telegram" in script
     assert "db_fingerprint_unchanged" in script
@@ -31,6 +44,19 @@ def test_remote_script_embeds_stateless_challenger_read_only_guard_and_bounded_c
     assert "challenger_semantic_gate_passed_before_leo_advance" in script
 
 
+@pytest.mark.parametrize(
+    "source",
+    (
+        "",
+        "CONTEXT_CLAIM_LIMIT = 4\n",
+        'command = ["--limit", "0"]\n',
+    ),
+)
+def test_db_context_limit_patch_rejects_incomplete_or_unknown_contract(source: str) -> None:
+    with pytest.raises(RuntimeError, match=r"bounded-limit contract|incomplete named-limit contract"):
+        runner._patch_db_context_limits(source)
+
+
 def test_remote_script_rejects_unbounded_tokens_and_unsafe_report_prefix() -> None:
     with pytest.raises(ValueError, match="between 128 and 2048"):
         runner.build_remote_script("abc123", challenger_max_tokens=4096)