overseer: wire overseer_advisor_model through consult-advisor CLI (#2113)

sandbox/agent-config/rules/overseer.md

@@ -81,6 +81,8 @@ egg-orch overseer consult-advisor \
   --output-file /tmp/advisor-verdict.json
 ```
 
+The verb auto-resolves `pipeline_id` from `EGG_PIPELINE_ID` (always set in the overseer context) and reads `PipelineConfig.overseer_advisor_model` from the orchestrator status endpoint to pass the configured alias to the advisor; if both the env var and the optional positional `pipeline_id` are absent — or the orchestrator is unreachable — the verb falls back to the `opus` default.
+
 The verdict is the JSON-serialized `AdvisorVerdict` with the same fields documented below (decision / priority / alert_summary / alert_detail / issue_title / issue_body / reasoning).
 
 **Trigger gate (the load-bearing constraint).** Invoke the advisor only when **both** conditions hold simultaneously:

sandbox/egg_lib/orch_cli.py

@@ -1773,6 +1773,7 @@ def cmd_overseer_consult_advisor(args: argparse.Namespace) -> int:
             as a model-output drift
     """
     import asyncio
+    from types import SimpleNamespace
 
     from egg_overseer.advisor import AdvisorParseError, consult_advisor
 
@@ -1809,6 +1810,57 @@ def cmd_overseer_consult_advisor(args: argparse.Namespace) -> int:
         print("Error: inputs.recent_log_lines must be an array", file=sys.stderr)
         return 2
 
+    # Resolve overseer_advisor_model from PipelineConfig when a
+    # pipeline-id is available (issue #2113). The orchestrator's
+    # status endpoint exposes the overseer-relevant config subset
+    # (orchestrator/routes/pipelines.py); we read overseer_advisor_model
+    # and pass a duck-typed config to consult_advisor. Falling back to
+    # config=None keeps the historic "opus" default for callers that
+    # do not provide a pipeline-id, and for any failure (orchestrator
+    # unreachable, malformed env, missing client module) — never crash
+    # the verb on the lookup path. NOTE: extend SimpleNamespace below
+    # if consult_advisor ever reads more `config.*` attributes; the
+    # duck-typed surface silently falls back to AttributeError today.
+    advisor_config: Any = None
+    pid = getattr(args, "pipeline_id", None) or get_pipeline_id_from_env()
+    if pid and _SAFE_ID_PATTERN.match(pid):
+        # Nested try so ImportError is handled before OrchestratorError is
+        # referenced — combining them in a single except clause raises
+        # NameError when the import itself fails (OrchestratorError is
+        # never bound). See review feedback on PR #2158.
+        try:
+            from egg_lib.orch_client import OrchClient, OrchestratorError
+        except ImportError as exc:
+            print(
+                f"Warning: cannot import egg_lib.orch_client ({exc}); "
+                f"falling back to default advisor model",
+                file=sys.stderr,
+            )
+        else:
+            try:
+                status = OrchClient().get_pipeline_status(quote(pid, safe=""))
+                cfg_dict = status.get("config") if isinstance(status, dict) else None
+                model = (
+                    cfg_dict.get("overseer_advisor_model") if isinstance(cfg_dict, dict) else None
+                )
+                if model:
+                    advisor_config = SimpleNamespace(overseer_advisor_model=model)
+            except OrchestratorError as exc:
+                print(
+                    f"Warning: cannot read PipelineConfig for {pid} "
+                    f"({exc}); falling back to default advisor model",
+                    file=sys.stderr,
+                )
+    elif pid:
+        # Malformed pipeline-id (e.g. corrupted EGG_PIPELINE_ID): skip
+        # the lookup silently rather than crashing via validate_id's
+        # sys.exit(1), which would collide with AdvisorParseError's
+        # exit-code semantics. See review feedback on PR #2158.
+        print(
+            f"Warning: pipeline_id {pid!r} is not a safe ID; falling back to default advisor model",
+            file=sys.stderr,
+        )
+
     recent_log_bytes_cap = getattr(args, "recent_log_bytes_cap", None)
     try:
         verdict = asyncio.run(
@@ -1817,7 +1869,7 @@ def cmd_overseer_consult_advisor(args: argparse.Namespace) -> int:
                 health_alerts=health_alerts,
                 progress_events=progress_events,
                 recent_log_lines=recent_log_lines,
-                config=None,
+                config=advisor_config,
                 recent_log_bytes_cap=recent_log_bytes_cap,
             )
         )
@@ -3103,6 +3155,17 @@ def add_signal_args(p: argparse.ArgumentParser) -> None:
             "to --output-file (or stdout when omitted)."
         ),
     )
+    ov_advisor.add_argument(
+        "pipeline_id",
+        nargs="?",
+        help=(
+            "Optional pipeline ID. When provided (or EGG_PIPELINE_ID is "
+            "set), the verb reads PipelineConfig.overseer_advisor_model "
+            "from the orchestrator status endpoint and passes the "
+            "configured alias to consult_advisor. Omitted: falls back "
+            "to the 'opus' default."
+        ),
+    )
     ov_advisor.add_argument(
         "--inputs-file",
         required=True,

sandbox/tests/test_egg_orch_overseer_consult_advisor.py

@@ -128,16 +128,25 @@ def _make_args(
     inputs_file: str | Path,
     output_file: str | Path | None = None,
     json_flag: bool = True,
+    pipeline_id: str | None = None,
     recent_log_bytes_cap: int | None = None,
 ) -> argparse.Namespace:
     return argparse.Namespace(
         inputs_file=str(inputs_file),
         output_file=str(output_file) if output_file else None,
         json=json_flag,
+        pipeline_id=pipeline_id,
         recent_log_bytes_cap=recent_log_bytes_cap,
     )
 
 
+@pytest.fixture(autouse=True)
+def _clear_egg_pipeline_id_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Stop ambient EGG_PIPELINE_ID from triggering the orch_client lookup
+    in tests that do not exercise the pipeline-id branch (issue #2113)."""
+    monkeypatch.delenv("EGG_PIPELINE_ID", raising=False)
+
+
 def _write_inputs(path: Path, *, data: object | None = None) -> None:
     if data is None:
         data = {
@@ -370,6 +379,198 @@ async def _fake(**_kwargs: object) -> AdvisorVerdict:
         assert rc == 2
         assert "cannot write --output-file" in capsys.readouterr().err
 
+    def test_pipeline_id_resolves_advisor_model_from_config(self, tmp_path: Path) -> None:
+        """Regression for issue #2113.
+
+        When a pipeline-id is provided (positional arg or EGG_PIPELINE_ID),
+        the verb must read PipelineConfig.overseer_advisor_model from the
+        orchestrator status endpoint and pass it through to consult_advisor
+        — previously config=None was hardcoded so the knob silently
+        defaulted to 'opus'.
+        """
+        inputs = tmp_path / "inputs.json"
+        _write_inputs(inputs)
+
+        captured: dict[str, object] = {}
+
+        async def _fake_consult_advisor(**kwargs: object) -> AdvisorVerdict:
+            captured.update(kwargs)
+            return AdvisorVerdict(decision="watch", reasoning="ok")
+
+        class _StubClient:
+            def get_pipeline_status(self, pid: str) -> dict[str, object]:
+                assert pid == "pipe-abc"
+                return {"config": {"overseer_advisor_model": "claude-opus-4-7"}}
+
+        with (
+            patch("egg_overseer.advisor.consult_advisor", side_effect=_fake_consult_advisor),
+            patch("egg_lib.orch_client.OrchClient", lambda: _StubClient()),
+        ):
+            rc = cmd_overseer_consult_advisor(
+                _make_args(inputs_file=inputs, pipeline_id="pipe-abc")
+            )
+
+        assert rc == 0
+        cfg = captured["config"]
+        assert cfg is not None
+        assert getattr(cfg, "overseer_advisor_model", None) == "claude-opus-4-7"
+
+    def test_pipeline_id_from_env_resolves_advisor_model(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """EGG_PIPELINE_ID populates pipeline_id when the positional arg
+        is absent — same plumbing as every other overseer verb."""
+        inputs = tmp_path / "inputs.json"
+        _write_inputs(inputs)
+        monkeypatch.setenv("EGG_PIPELINE_ID", "pipe-env")
+
+        captured: dict[str, object] = {}
+
+        async def _fake(**kwargs: object) -> AdvisorVerdict:
+            captured.update(kwargs)
+            return AdvisorVerdict(decision="watch", reasoning="ok")
+
+        class _StubClient:
+            def get_pipeline_status(self, pid: str) -> dict[str, object]:
+                assert pid == "pipe-env"
+                return {"config": {"overseer_advisor_model": "sonnet"}}
+
+        with (
+            patch("egg_overseer.advisor.consult_advisor", side_effect=_fake),
+            patch("egg_lib.orch_client.OrchClient", lambda: _StubClient()),
+        ):
+            rc = cmd_overseer_consult_advisor(_make_args(inputs_file=inputs))
+
+        assert rc == 0
+        cfg = captured["config"]
+        assert cfg is not None
+        assert getattr(cfg, "overseer_advisor_model", None) == "sonnet"
+
+    def test_pipeline_id_status_failure_falls_back_to_default(
+        self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        """Orchestrator unreachable / 404 / etc. must not crash the verb;
+        the advisor still runs with config=None (default 'opus')."""
+        from egg_lib.orch_client import OrchestratorError
+
+        inputs = tmp_path / "inputs.json"
+        _write_inputs(inputs)
+
+        captured: dict[str, object] = {}
+
+        async def _fake(**kwargs: object) -> AdvisorVerdict:
+            captured.update(kwargs)
+            return AdvisorVerdict(decision="watch", reasoning="ok")
+
+        class _BoomClient:
+            def get_pipeline_status(self, pid: str) -> dict[str, object]:
+                raise OrchestratorError("503 service unavailable")
+
+        with (
+            patch("egg_overseer.advisor.consult_advisor", side_effect=_fake),
+            patch("egg_lib.orch_client.OrchClient", lambda: _BoomClient()),
+        ):
+            rc = cmd_overseer_consult_advisor(
+                _make_args(inputs_file=inputs, pipeline_id="pipe-fail")
+            )
+
+        assert rc == 0
+        assert captured["config"] is None
+        assert "falling back to default advisor model" in capsys.readouterr().err
+
+    def test_malformed_pipeline_id_falls_back_without_crashing(
+        self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        """A pipeline_id that fails the safe-ID regex (e.g. corrupted
+        EGG_PIPELINE_ID containing a stray space or slash) must skip the
+        lookup and fall back to config=None — *not* crash via
+        validate_id's sys.exit(1), which would collide with
+        AdvisorParseError's exit-code semantics. Review feedback on PR
+        #2158.
+        """
+        inputs = tmp_path / "inputs.json"
+        _write_inputs(inputs)
+
+        captured: dict[str, object] = {}
+
+        async def _fake(**kwargs: object) -> AdvisorVerdict:
+            captured.update(kwargs)
+            return AdvisorVerdict(decision="watch", reasoning="ok")
+
+        # OrchClient must NOT be invoked on a malformed id — patch it to
+        # blow up if the verb tries to reach the orchestrator anyway.
+        def _explode() -> object:
+            raise AssertionError("OrchClient must not be called for malformed pipeline_id")
+
+        with (
+            patch("egg_overseer.advisor.consult_advisor", side_effect=_fake),
+            patch("egg_lib.orch_client.OrchClient", _explode),
+        ):
+            rc = cmd_overseer_consult_advisor(
+                _make_args(inputs_file=inputs, pipeline_id="bad id/with/slash"),
+            )
+
+        assert rc == 0
+        assert captured["config"] is None
+        assert "is not a safe ID" in capsys.readouterr().err
+
+    def test_orch_client_import_failure_falls_back_without_crashing(
+        self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        """If ``egg_lib.orch_client`` cannot be imported (e.g. the module
+        is unavailable in a stripped runtime), the verb must warn and
+        fall back to ``config=None`` rather than crashing with NameError
+        from a combined ``except (OrchestratorError, ImportError)`` that
+        cannot resolve ``OrchestratorError``. Review feedback on PR
+        #2158.
+        """
+        import builtins
+
+        inputs = tmp_path / "inputs.json"
+        _write_inputs(inputs)
+
+        captured: dict[str, object] = {}
+
+        async def _fake(**kwargs: object) -> AdvisorVerdict:
+            captured.update(kwargs)
+            return AdvisorVerdict(decision="watch", reasoning="ok")
+
+        real_import = builtins.__import__
+
+        def _import(name: str, *args: object, **kwargs: object) -> object:
+            if name == "egg_lib.orch_client":
+                raise ImportError("simulated missing orch_client")
+            return real_import(name, *args, **kwargs)  # type: ignore[arg-type]
+
+        with (
+            patch("egg_overseer.advisor.consult_advisor", side_effect=_fake),
+            patch.object(builtins, "__import__", side_effect=_import),
+        ):
+            rc = cmd_overseer_consult_advisor(
+                _make_args(inputs_file=inputs, pipeline_id="pipe-noimport")
+            )
+
+        assert rc == 0
+        assert captured["config"] is None
+        err = capsys.readouterr().err
+        assert "cannot import egg_lib.orch_client" in err
+        assert "falling back to default advisor model" in err
+
+    def test_parser_accepts_positional_pipeline_id(self) -> None:
+        """The new positional arg must coexist with the existing flags."""
+        parser = create_parser()
+        ns = parser.parse_args(
+            [
+                "overseer",
+                "consult-advisor",
+                "pipe-xyz",
+                "--inputs-file",
+                "/tmp/inputs.json",
+            ]
+        )
+        assert ns.pipeline_id == "pipe-xyz"
+        assert ns.inputs_file == "/tmp/inputs.json"
+
     def test_recent_log_bytes_cap_flag_forwarded(
         self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
     ) -> None: