fix(p1-11): redact CLI top-level exception text before stderr

autosearch/cli/main.py

@@ -12,6 +12,7 @@
 from autosearch.core import secrets_store
 from autosearch.core.environment_probe import probe_environment
 from autosearch.core.models import SearchMode
+from autosearch.core.redact import redact
 from autosearch.core.scope_clarifier import ScopeClarifier
 from autosearch.core.search_scope import (
     ChannelScope,
@@ -797,18 +798,19 @@ def _is_tty() -> bool:
 
 
 def _exit_query_failure(message: str, *, exit_code: int, json_output: bool) -> None:
+    safe_message = redact(message)
     if json_output:
         typer.echo(
             json.dumps(
                 {
                     "delivery_status": "error",
-                    "error": message,
+                    "error": safe_message,
                     "exit_code": exit_code,
                 }
             )
         )
     else:
-        typer.echo(message, err=True)
+        typer.echo(safe_message, err=True)
     raise typer.Exit(code=exit_code)
 
 

tests/smoke/test_first_use_flow.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import json
+import os
 import subprocess
 
 import pytest
@@ -77,6 +78,47 @@ def test_query_json_loop_emits_valid_envelope() -> None:
     assert payload["evidence_count"] == len(payload["evidence"])
 
 
+@pytest.mark.smoke
+def test_first_use_error_path_redacted(tmp_path) -> None:
+    leaked_key = "sk-FAKEKEY" + "1234567890abcdef"
+    (tmp_path / "sitecustomize.py").write_text(
+        """
+import autosearch.cli.query_pipeline as query_pipeline
+from autosearch.cli.query_pipeline import QueryResult
+
+_LEAKED_KEY = "sk-FAKEKEY" + "1234567890abcdef"
+
+
+async def _failing_run_query(_query: str, **_kwargs: object) -> QueryResult:
+    raise RuntimeError(f"upstream returned token {_LEAKED_KEY}")
+
+
+query_pipeline.run_query = _failing_run_query
+""",
+        encoding="utf-8",
+    )
+    env = smoke_env(AUTOSEARCH_LLM_MODE="dummy")
+    env["PYTHONPATH"] = f"{tmp_path}{os.pathsep}{env['PYTHONPATH']}"
+
+    result = subprocess.run(
+        [
+            *console_script_command("autosearch", "autosearch.cli.main"),
+            "query",
+            "smoke redaction query",
+        ],
+        capture_output=True,
+        text=True,
+        env=env,
+        timeout=60,
+    )
+
+    assert result.returncode == 1, (
+        f"query exited {result.returncode}; stderr:\n{result.stderr}\nstdout:\n{result.stdout}"
+    )
+    assert leaked_key not in result.stderr
+    assert "[REDACTED]" in result.stderr
+
+
 @pytest.mark.smoke
 def test_query_help_lists_subcommand() -> None:
     """`autosearch query --help` proves the v2 thin-orchestration CLI is wired up."""

tests/unit/test_mcp_error_redaction.py

@@ -12,8 +12,10 @@
 from __future__ import annotations
 
 import asyncio
+import json
 
 import pytest
+from typer.testing import CliRunner
 
 
 @pytest.fixture(autouse=True)
@@ -28,7 +30,7 @@
     if hasattr(result, "content"):
         for c in result.content:
             if hasattr(c, "text"):
                 import json
 
                 return json.loads(c.text)
     return result.model_dump() if hasattr(result, "model_dump") else dict(result)
@@ -93,3 +95,41 @@
     ).read_text(encoding="utf-8")
     assert "sk-ant-secretvalue" not in patterns
     assert "REDACTED" in patterns
+
+
+def test_cli_query_top_level_exception_redacted(monkeypatch: pytest.MonkeyPatch) -> None:
+    from autosearch.cli.main import app
+    from autosearch.cli.query_pipeline import QueryResult
+
+    leaked_key = "sk-FAKEKEY" + "1234567890abcdef"
+
+    async def _failing_run_query(_query: str, **_kwargs: object) -> QueryResult:
+        raise RuntimeError(f"upstream returned token {leaked_key}")
+
+    monkeypatch.setattr("autosearch.cli.query_pipeline.run_query", _failing_run_query)
+
+    result = CliRunner().invoke(app, ["query", "redaction smoke"])
+
+    combined_output = (result.output or "") + (result.stderr or "")
+    assert result.exit_code == 1
+    assert leaked_key not in combined_output
+    assert "REDACTED" in combined_output
+
+
+def test_cli_query_json_error_envelope_redacted(monkeypatch: pytest.MonkeyPatch) -> None:
+    from autosearch.cli.main import app
+    from autosearch.cli.query_pipeline import QueryResult
+
+    leaked_key = "sk-FAKEKEY" + "1234567890abcdef"
+
+    async def _failing_run_query(_query: str, **_kwargs: object) -> QueryResult:
+        raise RuntimeError(f"upstream returned token {leaked_key}")
+
+    monkeypatch.setattr("autosearch.cli.query_pipeline.run_query", _failing_run_query)
+
+    result = CliRunner().invoke(app, ["query", "redaction smoke", "--json"])
+
+    assert result.exit_code == 1
+    assert leaked_key not in result.stdout
+    payload = json.loads(result.stdout)
+    assert "[REDACTED]" in payload["error"]