fix(evaluation): redact optimizer config snapshots

Author: YAO-001

tests/evaluation/test_agent_optimizer.py

@@ -254,6 +254,8 @@ async def test_facade_persists_artifacts_under_output_dir(tmp_path, monkeypatch)
     """The facade must materialise result.json, summary.txt, rounds/*.json,
     baseline_prompts/, best_prompts/, config.snapshot.json and run.log under
     output_dir for every successful run."""
+    import json
+
     train = EvalSet(eval_set_id="train", eval_cases=[_eval_case("c1")])
     val = EvalSet(eval_set_id="val", eval_cases=[_eval_case("c1")])
     train_path = tmp_path / "train.json"
@@ -287,7 +289,14 @@ async def fake_call_gepa(self, **kwargs):
 
     assert (output_dir / "result.json").is_file()
     assert (output_dir / "summary.txt").is_file()
-    assert (output_dir / "config.snapshot.json").is_file()
+    config_snapshot_path = output_dir / "config.snapshot.json"
+    assert config_snapshot_path.is_file()
+    config_snapshot_text = config_snapshot_path.read_text(encoding="utf-8")
+    assert "test-key" not in config_snapshot_text
+    config_snapshot = json.loads(config_snapshot_text)
+    assert config_snapshot["optimize"]["algorithm"]["reflection_lm"][
+        "api_key"
+    ] == "<redacted>"
     assert (output_dir / "run.log").is_file()
     assert (output_dir / "baseline_prompts" / "instruction.md").is_file()
     assert (output_dir / "best_prompts" / "instruction.md").is_file()
@@ -297,6 +306,114 @@ async def fake_call_gepa(self, **kwargs):
     assert "SUCCEEDED" in log_line
 
 
+def test_copy_config_snapshot_recursively_redacts_common_secret_keys(tmp_path):
+    """Config snapshots must remain useful without publishing credentials."""
+    import json
+
+    payload = {
+        "api_key": "api-secret",
+        "nested": [
+            {
+                "TOKEN": "token-secret",
+                "access-token": "access-secret",
+                "Authorization": "Bearer auth-secret",
+            },
+            {
+                "password": "password-secret",
+                "credentials": {"username": "alice", "password": "nested-secret"},
+                "privateKey": "private-secret",
+            },
+            {
+                "openai_api_key": "namespaced-api-secret",
+                "github_token": "namespaced-token-secret",
+                "aws_secret_access_key": "aws-secret",
+                "db_passwd": "database-secret",
+            },
+        ],
+        "model_name": "gpt-4o",
+        "max_tokens": 128,
+        "token_budget": 256,
+    }
+    config_path = tmp_path / "optimizer.json"
+    config_path.write_text(json.dumps(payload), encoding="utf-8")
+    output_dir = tmp_path / "output"
+    output_dir.mkdir()
+
+    AgentOptimizer._copy_config_snapshot(str(config_path), str(output_dir))
+
+    snapshot_path = output_dir / "config.snapshot.json"
+    snapshot_text = snapshot_path.read_text(encoding="utf-8")
+    snapshot = json.loads(snapshot_text)
+    assert snapshot == {
+        "api_key": "<redacted>",
+        "nested": [
+            {
+                "TOKEN": "<redacted>",
+                "access-token": "<redacted>",
+                "Authorization": "<redacted>",
+            },
+            {
+                "password": "<redacted>",
+                "credentials": "<redacted>",
+                "privateKey": "<redacted>",
+            },
+            {
+                "openai_api_key": "<redacted>",
+                "github_token": "<redacted>",
+                "aws_secret_access_key": "<redacted>",
+                "db_passwd": "<redacted>",
+            },
+        ],
+        "model_name": "gpt-4o",
+        "max_tokens": 128,
+        "token_budget": 256,
+    }
+    assert snapshot_text == (
+        json.dumps(
+            snapshot,
+            ensure_ascii=False,
+            sort_keys=True,
+            indent=2,
+            allow_nan=False,
+        )
+        + "\n"
+    )
+    for secret in (
+        "api-secret",
+        "token-secret",
+        "access-secret",
+        "auth-secret",
+        "password-secret",
+        "nested-secret",
+        "private-secret",
+        "namespaced-api-secret",
+        "namespaced-token-secret",
+        "aws-secret",
+        "database-secret",
+    ):
+        assert secret not in snapshot_text
+
+
+@pytest.mark.parametrize(
+    "invalid_config",
+    [
+        '{"api_key": "secret", invalid}',
+        '{"api_key": NaN}',
+    ],
+)
+def test_copy_config_snapshot_invalid_json_fails_closed(tmp_path, invalid_config):
+    """Malformed source config must never be copied verbatim as an artifact."""
+    config_path = tmp_path / "optimizer.json"
+    config_path.write_text(invalid_config, encoding="utf-8")
+    output_dir = tmp_path / "output"
+    output_dir.mkdir()
+
+    AgentOptimizer._copy_config_snapshot(str(config_path), str(output_dir))
+
+    assert not (output_dir / "config.snapshot.json").exists()
+    assert not (output_dir / "config.snapshot.json.tmp").exists()
+
+
 @pytest.mark.asyncio
 async def test_facade_persists_artifacts_when_algorithm_fails(tmp_path, monkeypatch):
     """Even when the algorithm returns a FAILED result the facade should

trpc_agent_sdk/evaluation/_agent_optimizer.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import inspect
+import json
 import logging
 import os
 import signal
@@ -52,6 +53,56 @@
 
 _PROMPT_FILE_LOGGER = logging.getLogger("trpc_agent_sdk.optimizer")
 
+_REDACTED_CONFIG_VALUE = "<redacted>"
+_SENSITIVE_CONFIG_KEY_SUFFIXES = (
+    "apikey",
+    "authorization",
+    "authtoken",
+    "bearertoken",
+    "clientsecret",
+    "credential",
+    "credentials",
+    "idtoken",
+    "password",
+    "passwd",
+    "privatekey",
+    "refreshtoken",
+    "secret",
+    "secretaccesskey",
+    "secretkey",
+    "token",
+    "accesstoken",
+)
+
+
+def _reject_non_finite_json_constant(value: str) -> Any:
+    """Reject Python's non-standard JSON constants (NaN and infinities)."""
+    raise ValueError(f"non-finite JSON constant is not allowed: {value}")
+
+
+def _redact_config_secrets(value: Any) -> Any:
+    """Return a JSON-compatible copy with credential values redacted.
+
+    Key matching is case-insensitive, ignores separators and accepts
+    namespaced credential suffixes, so ``api_key``, ``API-Key`` and
+    ``openai_api_key`` share the same policy without redacting safe fields
+    such as ``max_tokens``.
+    """
+    if isinstance(value, dict):
+        redacted = {}
+        for key, child in value.items():
+            normalized_key = "".join(
+                character for character in key.casefold() if character.isalnum()
+            )
+            if normalized_key.endswith(_SENSITIVE_CONFIG_KEY_SUFFIXES):
+                redacted[key] = _REDACTED_CONFIG_VALUE
+            else:
+                redacted[key] = _redact_config_secrets(child)
+        return redacted
+    if isinstance(value, list):
+        return [_redact_config_secrets(child) for child in value]
+    return value
+
 
 def _atomic_write_text(path: str, content: str) -> None:
     """Atomically replace ``path`` with ``content`` (UTF-8).
@@ -157,8 +208,8 @@ async def optimize(
             output_dir: Required artifact directory. The facade creates it when
                 missing and persists ``result.json``, ``summary.txt``,
                 ``rounds/`` records, ``baseline_prompts/`` and ``best_prompts/``
-                directories, a ``config.snapshot.json`` copy of the input
-                config, and a ``run.log`` summary line.
+                directories, a redacted ``config.snapshot.json`` copy of the
+                input config, and a ``run.log`` summary line.
             callbacks: Optional evaluator lifecycle callbacks.
             update_source: When True, persist the best candidate back to
                 every registered TargetPrompt field after a SUCCEEDED
@@ -393,7 +444,7 @@ def _persist_artifacts(
                                             (regardless of update_source).
           - ``best_prompts/<name>.md``     Best candidate per field
                                             (only when a result was produced).
-          - ``config.snapshot.json``       Copy of the input config.
+          - ``config.snapshot.json``       Redacted copy of the input config.
           - ``run.log``                    One-line status footer.
 
         SIGINT (Ctrl+C) is masked for the duration of this method so a
@@ -473,12 +524,25 @@ def _write_rounds_directory(result: OptimizeResult, output_dir: str) -> None:
     def _copy_config_snapshot(config_path: str, output_dir: str) -> None:
         target = os.path.join(output_dir, "config.snapshot.json")
         try:
-            # Read + atomic-write rather than shutil.copyfile so an interrupted
-            # copy cannot leave a half-written ``config.snapshot.json``.
-            content = Path(config_path).read_text(encoding="utf-8")
+            # Parse and redact fully before the first write so neither the
+            # destination nor its temporary sibling can ever contain the raw
+            # credential-bearing config. Invalid/non-finite JSON therefore
+            # fails closed instead of falling back to a plaintext copy.
+            raw_config = json.loads(
+                Path(config_path).read_text(encoding="utf-8"),
+                parse_constant=_reject_non_finite_json_constant,
+            )
+            redacted_config = _redact_config_secrets(raw_config)
+            content = json.dumps(
+                redacted_config,
+                ensure_ascii=False,
+                sort_keys=True,
+                indent=2,
+                allow_nan=False,
+            ) + "\n"
             _atomic_write_text(target, content)
         except Exception:  # pragma: no cover
-            _PROMPT_FILE_LOGGER.warning("failed to copy config snapshot", exc_info=True)
+            _PROMPT_FILE_LOGGER.warning("failed to write redacted config snapshot", exc_info=True)
 
     @staticmethod
     def _write_run_log(*, output_dir: str, line: str) -> None: