fix(evaluation): 加固报告发布与配置快照

guocfu · guocfu · commit 94618bc9a5d0 · 2026-07-19T12:31:15.000+08:00
diff --git a/examples/optimization/eval_optimize_loop/artifact_writer.py b/examples/optimization/eval_optimize_loop/artifact_writer.py
@@ -7,11 +7,15 @@
 
 from __future__ import annotations
 
+import ctypes
+import errno
 import hashlib
+import json
 import os
 from pathlib import Path
 import shutil
-from typing import Literal, TypeAlias
+import sys
+from typing import Callable, Literal, TypeAlias
 from uuid import uuid4
 
 from pydantic import BaseModel
@@ -31,6 +35,20 @@
 ]
 
 _INPUT_COPY_DISABLED = "artifacts.copy_input_files=false"
+_SENSITIVE_CONFIG_KEYS = {"apikey", "authorization", "baseurl"}
+_APPROVED_SENSITIVE_VALUES = {
+    "",
+    "${TRPC_AGENT_API_KEY}",
+    "${TRPC_AGENT_BASE_URL}",
+    "fake-not-used-in-offline-mode",
+}
+_AT_FDCWD = -100
+_RENAME_NOREPLACE = 1
+_RENAMEAT2_UNAVAILABLE = {
+    errno.ENOSYS,
+    errno.EINVAL,
+    getattr(errno, "EOPNOTSUPP", errno.ENOTSUP),
+}
 
 
 class ArtifactWriteError(RuntimeError):
@@ -126,6 +144,87 @@ def _json_text(model: BaseModel) -> str:
     return model.model_dump_json(by_alias=False, indent=2) + "\n"
 
 
+def _normalized_sensitive_key(key: str) -> str:
+    return key.replace("_", "").replace("-", "").casefold()
+
+
+def _validate_sensitive_config_values(value: object, *, path: str = "$") -> None:
+    if isinstance(value, list):
+        for index, item in enumerate(value):
+            _validate_sensitive_config_values(item, path=f"{path}[{index}]")
+        return
+    if not isinstance(value, dict):
+        return
+    for key, item in value.items():
+        item_path = f"{path}.{key}"
+        if _normalized_sensitive_key(key) in _SENSITIVE_CONFIG_KEYS:
+            if not isinstance(item, str) or item not in _APPROVED_SENSITIVE_VALUES:
+                raise ArtifactWriteError(
+                    "sensitive optimizer config value is not an approved "
+                    f"placeholder: {item_path}"
+                )
+        else:
+            _validate_sensitive_config_values(item, path=item_path)
+
+
+def _validate_optimizer_config_for_copy(path: Path) -> None:
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ArtifactWriteError(
+            f"failed to parse optimizer config snapshot: {path}: {exc}"
+        ) from exc
+    _validate_sensitive_config_values(payload)
+
+
+def _target_exists(path: Path) -> bool:
+    return path.exists() or path.is_symlink()
+
+
+def _rename_directory_no_replace(source: Path, target: Path) -> None:
+    """Atomically publish a directory without replacing an existing target.
+
+    Linux uses renameat2 with RENAME_NOREPLACE. Platforms without that primitive
+    use a controlled fallback that rechecks the target immediately before rename.
+    """
+    if sys.platform.startswith("linux"):
+        try:
+            libc = ctypes.CDLL(None, use_errno=True)
+            renameat2 = libc.renameat2
+        except (AttributeError, OSError):
+            renameat2 = None
+        if renameat2 is not None:
+            renameat2.argtypes = [
+                ctypes.c_int,
+                ctypes.c_char_p,
+                ctypes.c_int,
+                ctypes.c_char_p,
+                ctypes.c_uint,
+            ]
+            renameat2.restype = ctypes.c_int
+            ctypes.set_errno(0)
+            result = renameat2(
+                _AT_FDCWD,
+                os.fsencode(source),
+                _AT_FDCWD,
+                os.fsencode(target),
+                _RENAME_NOREPLACE,
+            )
+            if result == 0:
+                return
+            error_number = ctypes.get_errno()
+            if error_number == errno.EEXIST:
+                raise ArtifactWriteError(
+                    f"report directory already exists: {target}"
+                )
+            if error_number not in _RENAMEAT2_UNAVAILABLE:
+                raise OSError(error_number, os.strerror(error_number), target)
+
+    if _target_exists(target):
+        raise ArtifactWriteError(f"report directory already exists: {target}")
+    source.rename(target)
+
+
 def _published_relative_path(root: Path, path: Path) -> str:
     relative = path.relative_to(root)
     if relative.parts and relative.parts[0].startswith(".report.tmp-"):
@@ -216,6 +315,7 @@ def _copy_input(
     destination_name: str,
     artifact_id: str,
     produced_by: ReportPhase,
+    content_validator: Callable[[Path], None] | None = None,
 ) -> ArtifactReference:
     if source.is_symlink():
         raise ArtifactWriteError(f"input must not be a symbolic link: {source}")
@@ -225,6 +325,8 @@ def _copy_input(
         raise ArtifactWriteError(f"failed to read input {source}: {exc}") from exc
     if actual_sha256 != expected_sha256:
         raise ArtifactWriteError(f"input hash mismatch: {source}")
+    if content_validator is not None:
+        content_validator(source)
 
     destination = staging / "inputs" / destination_name
     destination.parent.mkdir(parents=True, exist_ok=True)
@@ -387,6 +489,11 @@ def publish_report_bundle(
         )
         for artifact_id, source, expected_hash, destination_name, produced_by in input_specs:
             if copy_input_files:
+                content_validator = (
+                    _validate_optimizer_config_for_copy
+                    if artifact_id == "input.optimizer_config"
+                    else None
+                )
                 references.append(
                     _copy_input(
                         root=root,
@@ -396,6 +503,7 @@ def publish_report_bundle(
                         destination_name=destination_name,
                         artifact_id=artifact_id,
                         produced_by=produced_by,
+                        content_validator=content_validator,
                     )
                 )
             else:
@@ -433,7 +541,7 @@ def publish_report_bundle(
         )
         _validate_available_references(root, staging, validated_index)
 
-        staging.rename(target)
+        _rename_directory_no_replace(staging, target)
         staging = None
         return validated_index
     except Exception as exc:
diff --git a/tests/evaluation/test_eval_optimize_loop_stage5_artifacts.py b/tests/evaluation/test_eval_optimize_loop_stage5_artifacts.py
@@ -2,6 +2,7 @@
 
 from datetime import datetime, timezone
 from hashlib import sha256
+import json
 from pathlib import Path
 import shutil
 
@@ -32,8 +33,7 @@ def _copy_example(tmp_path: Path) -> Path:
     return target
 
 
-async def _build_fake_report(tmp_path: Path, run_id: str):
-    root = _copy_example(tmp_path)
+async def _build_fake_report_for_root(root: Path, run_id: str):
     prepared = prepare_run(root / "pipeline.json", run_id=run_id)
     result = await run_fake_stage(prepared, scenario="improve")
     shutil.rmtree(Path(prepared.workspace.run_dir) / "report", ignore_errors=True)
@@ -60,6 +60,10 @@ async def _build_fake_report(tmp_path: Path, run_id: str):
     return prepared, report
 
 
+async def _build_fake_report(tmp_path: Path, run_id: str):
+    return await _build_fake_report_for_root(_copy_example(tmp_path), run_id)
+
+
 @pytest.mark.asyncio
 async def test_publish_report_bundle_materializes_and_indexes_required_files(tmp_path):
     prepared, report = await _build_fake_report(tmp_path, "artifact_complete")
@@ -71,15 +75,62 @@ async def test_publish_report_bundle_materializes_and_indexes_required_files(tmp
     assert (report_dir / "optimization_report.json").is_file()
     assert (report_dir / "optimization_report.md").is_file()
     assert (report_dir / "artifact_index.json").is_file()
-    for name in (
-        "baseline_train",
-        "baseline_validation",
-        "candidate_train",
-        "candidate_validation",
-    ):
-        assert (report_dir / "evaluations" / f"{name}.json").is_file()
     assert all(ref.relative_path != "report/artifact_index.json" for ref in index.artifacts)
 
+    input_paths = {
+        ref.artifact_id: ref.relative_path
+        for ref in index.artifacts
+        if ref.artifact_type == "input"
+    }
+    assert input_paths == {
+        "input.pipeline_config": "report/inputs/pipeline_config.json",
+        "input.optimizer_config": "report/inputs/optimizer_config.json",
+        "input.train_evalset": "report/inputs/train_evalset.json",
+        "input.validation_evalset": "report/inputs/validation_evalset.json",
+    }
+    assert {
+        ref.artifact_id: ref.relative_path
+        for ref in index.artifacts
+        if ref.artifact_type == "prompt"
+    } == {
+        "prompt.baseline.system_prompt": (
+            "report/prompts/baseline/000-system_prompt.md"
+        ),
+        "prompt.candidate.system_prompt": (
+            "report/prompts/candidate/000-system_prompt.md"
+        ),
+    }
+    baseline_prompt = report_dir / "prompts" / "baseline" / "000-system_prompt.md"
+    candidate_prompt = report_dir / "prompts" / "candidate" / "000-system_prompt.md"
+    assert baseline_prompt.read_text(encoding="utf-8") == (
+        report.input_snapshot.prompt_snapshots[0].content
+    )
+    assert candidate_prompt.read_text(encoding="utf-8") == (
+        report.candidate.prompts["system_prompt"]
+    )
+    evaluation_paths = {
+        ref.artifact_id: ref.relative_path
+        for ref in index.artifacts
+        if ref.artifact_type == "evaluation"
+    }
+    assert evaluation_paths == {
+        f"evaluation.{name}": f"report/evaluations/{name}.json"
+        for name in (
+            "baseline_train",
+            "baseline_validation",
+            "candidate_train",
+            "candidate_validation",
+        )
+    }
+    assert {
+        path.name for path in (report_dir / "evaluations").iterdir()
+    } == {
+        "baseline_train.json",
+        "baseline_validation.json",
+        "candidate_train.json",
+        "candidate_validation.json",
+    }
+
     available = [ref for ref in index.artifacts if ref.status == "available"]
     assert available
     for ref in available:
@@ -110,6 +161,36 @@ async def test_publish_rejects_input_hash_drift(tmp_path):
     assert list(run_dir.glob(".report.tmp-*")) == []
 
 
+@pytest.mark.asyncio
+async def test_publish_rejects_plaintext_secret_in_optimizer_config_without_leaking_it(
+    tmp_path,
+):
+    root = _copy_example(tmp_path)
+    optimizer_path = root / "optimizer.json"
+    payload = json.loads(optimizer_path.read_text(encoding="utf-8"))
+    secret = "sk-real-sentinel-secret-must-not-be-copied"
+    payload["optimize"]["algorithm"]["reflection_lm"]["api_key"] = secret
+    optimizer_path.write_text(
+        json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
+        encoding="utf-8",
+    )
+    prepared, report = await _build_fake_report_for_root(root, "artifact_secret")
+    run_dir = Path(prepared.workspace.run_dir)
+
+    with pytest.raises(ArtifactWriteError, match="sensitive optimizer config"):
+        publish_report_bundle(report, run_dir=run_dir, copy_input_files=True)
+
+    assert secret in optimizer_path.read_text(encoding="utf-8")
+    assert not (run_dir / "report").exists()
+    assert list(run_dir.glob(".report.tmp-*")) == []
+    secret_bytes = secret.encode("utf-8")
+    assert all(
+        secret_bytes not in path.read_bytes()
+        for path in run_dir.rglob("*")
+        if path.is_file()
+    )
+
+
 @pytest.mark.asyncio
 async def test_publish_rejects_existing_report_directory(tmp_path):
     prepared, report = await _build_fake_report(tmp_path, "artifact_existing")
@@ -124,6 +205,34 @@ async def test_publish_rejects_existing_report_directory(tmp_path):
         )
 
 
+@pytest.mark.asyncio
+async def test_publish_does_not_replace_report_directory_created_during_staging(
+    tmp_path, monkeypatch
+):
+    prepared, report = await _build_fake_report(tmp_path, "artifact_report_race")
+    run_dir = Path(prepared.workspace.run_dir)
+    report_dir = run_dir / "report"
+    original_write = artifact_writer._write_text
+
+    def create_competing_report_before_publish(path, content):
+        original_write(path, content)
+        if path.name == "artifact_index.json":
+            report_dir.mkdir()
+
+    monkeypatch.setattr(
+        artifact_writer,
+        "_write_text",
+        create_competing_report_before_publish,
+    )
+
+    with pytest.raises(ArtifactWriteError, match="already exists"):
+        publish_report_bundle(report, run_dir=run_dir, copy_input_files=True)
+
+    assert report_dir.is_dir()
+    assert list(report_dir.iterdir()) == []
+    assert list(run_dir.glob(".report.tmp-*")) == []
+
+
 @pytest.mark.parametrize("target_inside_run", [False, True])
 def test_discovery_rejects_any_file_symlink(tmp_path, target_inside_run):
     run_dir = tmp_path / "run"
