Landscape-sweep GPT-6 lane through OmniRoute: lane-local Codex home with the token stack at max effort (stacked on #385)

manifests/evidence.json

@@ -38968,8 +38968,8 @@
     },
     {
       "path": "tests/test_landscape_sweep_harness.py",
-      "sha256": "7f9c7ab0e3555563f1c26a651005cb3affad2ab4f19628bcbdf581fb800e7c17",
-      "bytes": 148389
+      "sha256": "a61c1be98cbb113e7ab574936f809dd23be2bc959c576a361153267a5c6371db",
+      "bytes": 162061
     },
     {
       "path": "tests/test_lane_packets.py",
@@ -39458,13 +39458,13 @@
     },
     {
       "path": "tools/sota-convergence/landscape-sweep/README.md",
-      "sha256": "3e059eb203d373f36e0fa035e372cf494b40d94cfe786caf2eaa2863065546f8",
-      "bytes": 34645
+      "sha256": "f6019a3a5301bf6bdbe20212b52556601b880f535693b4552a773d2473299af7",
+      "bytes": 39012
     },
     {
       "path": "tools/sota-convergence/landscape-sweep/build_args.py",
-      "sha256": "c9d68dfc93e59a994d41e0c9df717fd2fc5e8f934ef7c3da5719b700e98f3f4a",
-      "bytes": 17188
+      "sha256": "5643aa8aa5fe4174457ca0c2b694a49c8c4a020a4f279420e7034f44061a6eb5",
+      "bytes": 30467
     },
     {
       "path": "tools/sota-convergence/landscape-sweep/build_inputs.py",
@@ -39478,8 +39478,8 @@
     },
     {
       "path": "tools/sota-convergence/landscape-sweep/codex_job.py",
-      "sha256": "0a3603ad32aaa4f930beabfe088ff295e3ff31cfecd25adbcc5e542322867e99",
-      "bytes": 30208
+      "sha256": "dce6ca4e908a151c4b509a207a15dc24f0d9f022cbe7031cb5eb87b1a9ec0921",
+      "bytes": 35782
     },
     {
       "path": "tools/sota-convergence/landscape-sweep/convert.py",

tests/test_landscape_sweep_harness.py

@@ -614,7 +614,8 @@ def test_refusals(self):
 config = json.load(open(os.environ["FAKE_CODEX_CONFIG"]))
 stdin, null = os.fstat(0), os.stat(os.devnull)
 record = {{"argv": sys.argv[1:], "cwd": os.getcwd(), "stdin_devnull": (stdin.st_ino, stdin.st_dev) == (null.st_ino, null.st_dev),
-          "stdin_read": sys.stdin.read(), "rust_log": os.environ.get("RUST_LOG")}}
+          "stdin_read": sys.stdin.read(), "rust_log": os.environ.get("RUST_LOG"),
+          "codex_home": os.environ.get("CODEX_HOME"), "api_key_present": bool(os.environ.get("OMNIROUTE_API_KEY"))}}
 if config.get("stubborn"):
     import subprocess
     subprocess.Popen([sys.executable, "-c", "import os, signal, sys, time; signal.signal(signal.SIGTERM, signal.SIG_IGN); "
@@ -653,6 +654,153 @@ def process_alive(pid: int) -> bool:
     return bool(state) and not state.startswith("Z")
 
 
+STACK_WORKER_FIXTURE = """model = "gpt-6-astra"
+model_reasoning_effort = "max"
+web_search = "live"
+
+[mcp_servers.context-mode]
+disabled_tools = ["ctx_upgrade", "ctx_purge"]
+"""
+TOKEN_MCP_SERVERS = ("serena", "ai-memory", "socraticode", "headroom", "codebase-memory", "qmd", "context-mode")
+
+
+class OmniRouteLaneBuildTests(unittest.TestCase):
+    """--gpt6-provider omniroute stages a lane-local CODEX_HOME: the provider block, the rendered token MCP servers
+    and the stack-worker profile, and nothing of the host's interactive trust state."""
+
+    def stage_lane(self, *extra):
+        work = stage_work(self)
+        profile = temp_dir(self) / "stack-worker.config.toml"
+        profile.write_text(STACK_WORKER_FIXTURE, encoding="utf-8")
+        return work, profile, build(work, "--gpt6-provider", "omniroute", "--codex-host", "example",
+                                    "--stack-worker-profile", profile, *extra)
+
+    def test_lane_home_carries_provider_mcp_servers_and_profile(self):
+        work, profile, done = self.stage_lane()
+        self.assertEqual(done.returncode, 0, done.stderr)
+        config = (work / "codex-home" / "config.toml").read_text(encoding="utf-8")
+        for line in ('model = "cx/gpt-6-astra"', 'model_provider = "omniroute"', 'model_reasoning_effort = "max"',
+                     "[model_providers.omniroute]", 'base_url = "http://127.0.0.1:20128/v1"',
+                     'env_key = "OMNIROUTE_API_KEY"', "requires_openai_auth = false", 'wire_api = "responses"'):
+            self.assertIn(line, config)
+        for server in TOKEN_MCP_SERVERS:
+            self.assertIn(f"[mcp_servers.{server}]", config)
+        self.assertNotIn("supports_websockets = true", config)
+        # The key never reaches a shell snapshot file, and GPT-6 Astra's search is Codex's standalone web.run.
+        for line in ("[features]", "shell_snapshot = false", "standalone_web_search = true",
+                     "supports_standalone_web_search = true"):
+            self.assertIn(line, config)
+        self.assertNotIn("[projects.", config)
+        self.assertNotIn("[hooks.state", config)
+        self.assertNotIn("${", config)  # every template placeholder rendered
+        self.assertEqual((work / "codex-home" / "stack-worker.config.toml").read_bytes(), profile.read_bytes())
+        try:
+            import tomllib
+        except ImportError:
+            tomllib = None
+        if tomllib is not None:
+            parsed = tomllib.loads(config)
+            self.assertEqual(parsed["model_providers"]["omniroute"]["env_key"], "OMNIROUTE_API_KEY")
+            self.assertIs(parsed["model_providers"]["omniroute"]["supports_standalone_web_search"], True)
+            self.assertEqual(parsed["features"], {"shell_snapshot": False, "standalone_web_search": True})
+            self.assertEqual(sorted(parsed["mcp_servers"]), sorted(TOKEN_MCP_SERVERS))
+        codex = json.loads((work / "staged.json").read_text())["codex"]
+        self.assertEqual((codex["provider"], codex["codex_home"], codex["profile"], codex["api_key_env"],
+                          codex["model"], codex["effort"]),
+                         ("omniroute", "codex-home", "stack-worker", "OMNIROUTE_API_KEY", "cx/gpt-6-astra", "max"))
+        self.assertEqual(sorted(codex["lane_home"]["mcp_servers"]), sorted(TOKEN_MCP_SERVERS))
+        self.assertNotIn("quota_stop_percent", codex)
+        self.assertEqual(codex["api_key_placeholder"], "local-loopback")  # keyless loopback gateway by default
+        self.assertNotIn("local-loopback", config)  # the placeholder is runtime-only, never in the lane config
+        # The staged runner reads the same settings back.
+        lane = codex_job.settings(work)
+        self.assertEqual((lane["provider"], lane["codex_home"], lane["profile"]),
+                         ("omniroute", work / "codex-home", "stack-worker"))
+
+    def test_lane_refuses_quota_gate_missing_host_and_remote_gateways(self):
+        for extra, needle in ((("--quota-stop-percent", "90"), "--quota-stop-percent"),
+                              (("--omniroute-base-url", "http://10.0.0.5:20128/v1"), "loopback")):
+            _, _, done = self.stage_lane(*extra)
+            self.assertEqual(done.returncode, 2, done.stdout)
+            self.assertIn(needle, done.stderr)
+        work = stage_work(self)
+        done = build(work, "--gpt6-provider", "omniroute")
+        self.assertEqual(done.returncode, 2, done.stdout)
+        self.assertIn("--codex-host", done.stderr)
+
+    def test_codex_host_accepts_a_private_host_value_file_by_path(self):
+        private = temp_dir(self) / "privatehost.json"
+        private.write_text((ROOT / "adoption" / "hosts" / "example.json").read_text(encoding="utf-8"), encoding="utf-8")
+        work = stage_work(self)
+        profile = temp_dir(self) / "stack-worker.config.toml"
+        profile.write_text(STACK_WORKER_FIXTURE, encoding="utf-8")
+        done = build(work, "--gpt6-provider", "omniroute", "--codex-host", private, "--stack-worker-profile", profile)
+        self.assertEqual(done.returncode, 0, done.stderr)
+        config = (work / "codex-home" / "config.toml").read_text(encoding="utf-8")
+        self.assertIn("[mcp_servers.context-mode]", config)
+        self.assertNotIn(str(private.parent), config)  # the private file's location stays out of the lane record
+        codex = json.loads((work / "staged.json").read_text())["codex"]
+        self.assertEqual(codex["lane_home"]["host"], "privatehost")
+        missing = build(stage_work(self), "--gpt6-provider", "omniroute", "--codex-host", str(private) + ".gone.json",
+                        "--stack-worker-profile", profile)
+        self.assertEqual(missing.returncode, 2)
+        self.assertIn("does not exist", missing.stderr)
+
+    @unittest.skipIf(sys.version_info < (3, 11), "tomllib is Python 3.11+")
+    def test_mcp_extraction_accepts_quoted_and_spaced_headers_and_fails_closed(self):
+        import tomllib
+        text = ('model = "x"\n[mcp_servers."demo"]\ncommand = "a"\n[ mcp_servers.plain ]\ncommand = "b"\n'
+                '[mcp_servers.plain.env]\nK = "v"\n[projects."/p"]\ntrust_level = "trusted"\n')
+        sections, names = build_args.mcp_sections(text)
+        self.assertEqual(names, ["demo", "plain"])
+        self.assertNotIn("[projects.", sections)
+        self.assertEqual(tomllib.loads(sections)["mcp_servers"], tomllib.loads(text)["mcp_servers"])
+        # A server the line extraction cannot see (declared with inline-table syntax under [mcp_servers]) fails closed.
+        with self.assertRaises(ValueError):
+            build_args.mcp_sections('[mcp_servers]\nhidden = { command = "c" }\n[mcp_servers.plain]\ncommand = "b"\n')
+
+    @unittest.skipIf(sys.version_info < (3, 11), "tomllib is Python 3.11+")
+    def test_mcp_extraction_keeps_nested_tables_in_every_header_spelling(self):
+        import tomllib
+        # TOML allows a literal-quoted name and spaces around the dots; a server's env table in either spelling is kept.
+        for env_header in ("[mcp_servers.'demo'.env]", "[ mcp_servers . demo . env ]", "[mcp_servers . 'demo' . env]"):
+            text = f'[mcp_servers.demo]\ncommand = "a"\n{env_header}\nK = "v"\n[projects."/p"]\ntrust_level = "t"\n'
+            sections, names = build_args.mcp_sections(text)
+            self.assertEqual(names, ["demo"], env_header)
+            self.assertEqual(tomllib.loads(sections)["mcp_servers"], {"demo": {"command": "a", "env": {"K": "v"}}},
+                             env_header)
+        # A nested table the line extraction cannot see (a quoted top-level key) would lose the env: fail closed.
+        with self.assertRaisesRegex(ValueError, r"differ from the rendered config for \['demo'\]"):
+            build_args.mcp_sections('[mcp_servers.demo]\ncommand = "a"\n["mcp_servers".demo.env]\nK = "v"\n')
+
+    def test_mcp_extraction_refuses_without_tomllib(self):
+        with mock.patch.dict(sys.modules, {"tomllib": None}):  # a None entry makes the import raise ImportError
+            with self.assertRaisesRegex(ValueError, "Python 3.11"):
+                build_args.mcp_sections('[mcp_servers.demo]\ncommand = "a"\n')
+
+    def test_missing_profile_names_the_flag(self):
+        work = stage_work(self)
+        done = build(work, "--gpt6-provider", "omniroute", "--codex-host", "example",
+                     "--stack-worker-profile", temp_dir(self) / "absent.config.toml")
+        self.assertEqual(done.returncode, 2, done.stdout)
+        self.assertIn("pass --stack-worker-profile", done.stderr)
+
+    def test_require_key_stages_no_placeholder(self):
+        work, _, done = self.stage_lane("--omniroute-require-key")
+        self.assertEqual(done.returncode, 0, done.stderr)
+        self.assertNotIn("api_key_placeholder", json.loads((work / "staged.json").read_text())["codex"])
+
+    def test_native_default_is_unchanged(self):
+        work = stage_work(self)
+        done = build(work)
+        self.assertEqual(done.returncode, 0, done.stderr)
+        codex = json.loads((work / "staged.json").read_text())["codex"]
+        self.assertNotIn("provider", codex)
+        self.assertEqual(codex["model"], "gpt-6-astra")
+        self.assertFalse((work / "codex-home").exists())
+        self.assertEqual(codex_job.settings(work)["codex_home"], None)
+
+
 class RunnerCase(unittest.TestCase):
     def setUp(self):
         self.work = temp_dir(self)
@@ -698,6 +846,77 @@ def record(self):
                                                  "reasoning_output_tokens": 3}}
 
 
+class OmniRouteLaneRunnerTests(RunnerCase):
+    """A gateway lane runs Codex with CODEX_HOME set to the staged lane-local home, -p stack-worker and no
+    --ignore-user-config, and never starts without its key variable."""
+
+    def lane(self):
+        home = self.work / "codex-home"
+        home.mkdir(exist_ok=True)
+        (home / "config.toml").write_text('model = "cx/gpt-6-astra"\nmodel_provider = "omniroute"\n', encoding="utf-8")
+        (home / "stack-worker.config.toml").write_text(STACK_WORKER_FIXTURE, encoding="utf-8")
+        self.settings({"provider": "omniroute", "codex_home": "codex-home", "profile": "stack-worker",
+                       "api_key_env": "OMNIROUTE_API_KEY", "model": "cx/gpt-6-astra"})
+
+    def test_lane_uses_its_codex_home_and_profile(self):
+        self.lane()
+        self.env["OMNIROUTE_API_KEY"] = "fixture-not-a-key"
+        result = self.job("gpt6-probe", last=LAST, events=[COMPLETED])
+        directory = self.work / "gpt6" / "gpt6-probe"
+        self.assertEqual(self.record()["argv"], [
+            "exec", "-p", "stack-worker", "--skip-git-repo-check", "-s", "read-only",
+            "-m", "cx/gpt-6-astra", "-c", 'model_reasoning_effort="max"', "-c", 'web_search="live"',
+            "--output-schema", str(directory / "schema.json"), "-o", str(directory / "last.json"), "--json",
+            "Reply in JSON."])
+        self.assertEqual(self.record()["codex_home"], str(self.work / "codex-home"))
+        self.assertTrue(self.record()["api_key_present"])
+        self.assertEqual((result["status"], result["exit"], result["model"]), ("done", 0, "cx/gpt-6-astra"))
+        self.assertEqual(json.loads((directory / "inputs.json").read_text())["provider"], "omniroute")
+
+    def test_lane_without_its_key_never_starts_codex(self):
+        self.lane()
+        self.env.pop("OMNIROUTE_API_KEY", None)
+        result = self.job("gpt6-probe", last=LAST, events=[COMPLETED])
+        self.assertEqual(result["exit"], codex_job.EXIT_NO_KEY)
+        self.assertIn("OMNIROUTE_API_KEY is not set", result["stderr_tail"])
+        self.assertFalse((self.bin / "record.json").exists())  # the fake codex never ran
+
+    def test_keyless_lane_uses_the_placeholder_only_when_the_variable_is_unset(self):
+        self.lane()
+        self.settings({"provider": "omniroute", "codex_home": "codex-home", "profile": "stack-worker",
+                       "api_key_env": "OMNIROUTE_API_KEY", "api_key_placeholder": "local-loopback",
+                       "model": "cx/gpt-6-astra"})
+        self.env.pop("OMNIROUTE_API_KEY", None)
+        result = self.job("gpt6-probe", last=LAST, events=[COMPLETED])
+        self.assertEqual(result["exit"], 0)
+        self.assertTrue(self.record()["api_key_present"])
+        lane = codex_job.settings(self.work)
+        saved = os.environ.pop("OMNIROUTE_API_KEY", None)
+        try:
+            self.assertEqual(codex_job.codex_env(lane)["OMNIROUTE_API_KEY"], "local-loopback")
+            os.environ["OMNIROUTE_API_KEY"] = "   "  # blank counts as unset: Codex rejects a blank env_key value
+            self.assertEqual(codex_job.codex_env(lane)["OMNIROUTE_API_KEY"], "local-loopback")
+            os.environ["OMNIROUTE_API_KEY"] = "operator-value"
+            self.assertEqual(codex_job.codex_env(lane)["OMNIROUTE_API_KEY"], "operator-value")  # a real key wins
+        finally:
+            os.environ.pop("OMNIROUTE_API_KEY", None)
+            if saved is not None:
+                os.environ["OMNIROUTE_API_KEY"] = saved
+
+    def test_lane_settings_refuse_a_quota_gate_and_a_missing_home(self):
+        self.lane()
+        self.settings({"provider": "omniroute", "codex_home": "codex-home", "profile": "stack-worker",
+                       "api_key_env": "OMNIROUTE_API_KEY", "quota_stop_percent": 90})
+        with self.assertRaises(codex_job.UsageError):
+            codex_job.settings(self.work)
+        self.settings({"provider": "omniroute", "codex_home": "missing-home", "api_key_env": "OMNIROUTE_API_KEY"})
+        with self.assertRaises(codex_job.UsageError):
+            codex_job.settings(self.work)
+        self.settings({"provider": "omniroute", "codex_home": "../outside", "api_key_env": "OMNIROUTE_API_KEY"})
+        with self.assertRaises(codex_job.UsageError):
+            codex_job.settings(self.work)
+
+
 class RunnerTests(RunnerCase):
     def test_codex_command_line_stdin_and_directory(self):
         result = self.job("gpt6-probe", last=LAST, events=[{"type": "thread.started", "thread_id": "fixture"}, COMPLETED])