diff --git a/sandbox/agent-config/rules/overseer.md b/sandbox/agent-config/rules/overseer.md index ff72f4d69f..987eff367d 100644 --- a/sandbox/agent-config/rules/overseer.md +++ b/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: diff --git a/sandbox/egg_lib/orch_cli.py b/sandbox/egg_lib/orch_cli.py index 86aa441f48..0726300d90 100755 --- a/sandbox/egg_lib/orch_cli.py +++ b/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, diff --git a/sandbox/tests/test_egg_orch_overseer_consult_advisor.py b/sandbox/tests/test_egg_orch_overseer_consult_advisor.py index 7173233f5e..1edd96b73d 100644 --- a/sandbox/tests/test_egg_orch_overseer_consult_advisor.py +++ b/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: