fix(moa): scope quiet relay to machine-readable CLI

Keep MoA reference display events off the machine-readable -Q stdout
surface (platform=cli with tool_progress_mode=off) while preserving them
everywhere else. Extracts the relay into module-level helpers so the
policy is testable.

Salvaged from #67334.
This commit is contained in:
sgtworkman 2026-07-23 12:01:49 -07:00 committed by Teknium
parent 2487dea9ce
commit 3dfe712384
2 changed files with 118 additions and 0 deletions

View file

@ -69,6 +69,43 @@ def _ra():
return run_agent
def _moa_reference_output_allowed(agent: Any) -> bool:
"""Keep MoA display events off only the machine-readable ``-Q`` surface."""
return not (
getattr(agent, "platform", None) == "cli"
and getattr(agent, "tool_progress_mode", "all") == "off"
)
def _relay_moa_reference_event(agent: Any, event: str, **kwargs: Any) -> None:
"""Relay MoA display events while preserving the ``-Q`` stdout contract."""
if not _moa_reference_output_allowed(agent):
return
cb = getattr(agent, "tool_progress_callback", None)
if cb is None:
return
try:
if event == "moa.reference":
cb(
"moa.reference",
str(kwargs.get("label") or ""),
str(kwargs.get("text") or ""),
None,
moa_index=kwargs.get("index"),
moa_count=kwargs.get("count"),
)
elif event == "moa.aggregating":
cb(
"moa.aggregating",
str(kwargs.get("aggregator") or ""),
None,
None,
moa_ref_count=kwargs.get("ref_count"),
)
except Exception:
pass
def _normalize_route_base_url(base_url: Any) -> str:
"""Canonicalize an endpoint URL for model-route identity comparisons."""
return normalize_route_base_url(base_url)

View file

@ -0,0 +1,81 @@
"""Regression coverage for machine-readable MoA quiet output."""
from __future__ import annotations
from types import SimpleNamespace
import unittest
from agent.agent_init import _relay_moa_reference_event
class MoAQuietReferenceOutputTests(unittest.TestCase):
@staticmethod
def _agent(*, platform: str, tool_progress_mode: str, quiet_mode: bool = True):
calls = []
def callback(*args, **kwargs):
calls.append((args, kwargs))
return SimpleNamespace(
platform=platform,
tool_progress_mode=tool_progress_mode,
quiet_mode=quiet_mode,
tool_progress_callback=callback,
), calls
def test_machine_readable_cli_suppresses_reference_relay(self) -> None:
agent, calls = self._agent(platform="cli", tool_progress_mode="off")
_relay_moa_reference_event(
agent,
"moa.reference",
label="local:advisor",
text="hidden",
index=1,
count=1,
)
self.assertEqual(calls, [])
def test_interactive_cli_delivers_reference_relay(self) -> None:
agent, calls = self._agent(
platform="cli",
tool_progress_mode="all",
quiet_mode=True,
)
_relay_moa_reference_event(
agent,
"moa.reference",
label="local:advisor",
text="visible",
index=1,
count=2,
)
self.assertEqual(
calls,
[
(
("moa.reference", "local:advisor", "visible", None),
{"moa_index": 1, "moa_count": 2},
)
],
)
def test_gateway_delivers_even_when_progress_mode_is_off(self) -> None:
agent, calls = self._agent(platform="discord", tool_progress_mode="off")
_relay_moa_reference_event(
agent,
"moa.aggregating",
aggregator="local:aggregator",
ref_count=2,
)
self.assertEqual(
calls,
[
(
("moa.aggregating", "local:aggregator", None, None),
{"moa_ref_count": 2},
)
],
)
if __name__ == "__main__":
unittest.main()