[codex] Harden Retriever harness artifact safety

nemo_retriever/harness/README.md

@@ -211,6 +211,9 @@ Read these files instead of scraping stdout:
 
 Dry-runs write only planning artifacts. They do not create empty `run.log`,
 `beir_metrics.json`, `beir_run.trec`, or `query_results.jsonl` files.
+An explicit `--output-dir` must be absent or empty; the harness refuses to
+reuse a non-empty directory so stale files cannot be reported as current-run
+artifacts.
 
 ## Gates
 

nemo_retriever/src/nemo_retriever/harness/artifact_writer.py

@@ -15,7 +15,28 @@
 from typing import Any, Mapping
 
 from nemo_retriever.harness.contracts import FailurePayload, PHASE_VALUES, STATUS_VALUES
-from nemo_retriever.harness.json_io import jsonable, write_json
+from nemo_retriever.harness.json_io import jsonable, redact, write_json
+
+
+ARTIFACT_NAMES = (
+    "status.json",
+    "events.jsonl",
+    "resolved_benchmark.json",
+    "ingest_plan.json",
+    "query_plan.json",
+    "summary_metrics.json",
+    "environment.json",
+    "runfile.json",
+    "results.json",
+    "run.log",
+    "beir_metrics.json",
+    "beir_run.trec",
+    "query_results.jsonl",
+)
+
+
+class ArtifactWriteError(RuntimeError):
+    """An artifact directory could not be initialized or written safely."""
 
 
 def utc_now() -> str:
@@ -25,7 +46,7 @@ def utc_now() -> str:
 def append_jsonl(path: Path, payload: Mapping[str, Any]) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     with path.open("a", encoding="utf-8") as handle:
-        handle.write(json.dumps(jsonable(payload), sort_keys=False) + "\n")
+        handle.write(json.dumps(redact(jsonable(payload)), sort_keys=False) + "\n")
 
 
 def append_text(path: Path, text: str) -> None:
@@ -88,40 +109,79 @@ def capture_output_to_log(path: Path, *, label: str):
             os.close(saved_stdout)
 
 
-def redact(value: Any) -> Any:
-    if isinstance(value, dict):
-        out: dict[str, Any] = {}
-        for key, nested in value.items():
-            normalized = str(key).lower()
-            if any(token in normalized for token in ("api_key", "password", "secret", "credential", "token")):
-                out[key] = "<redacted>" if nested else nested
-            else:
-                out[key] = redact(nested)
-        return out
-    if isinstance(value, list):
-        return [redact(item) for item in value]
-    if isinstance(value, tuple):
-        return [redact(item) for item in value]
-    return value
+def initialize_artifact_dir(path: Path) -> Path:
+    """Create an artifact directory, refusing to reuse non-empty output."""
+    artifact_dir = path.expanduser().resolve()
+    try:
+        artifact_dir.mkdir(parents=True, exist_ok=False)
+    except FileExistsError as exc:
+        if not artifact_dir.is_dir():
+            raise ArtifactWriteError(f"Artifact path exists and is not a directory: {artifact_dir}") from exc
+        try:
+            next(artifact_dir.iterdir())
+        except StopIteration:
+            pass
+        except OSError as list_exc:
+            raise ArtifactWriteError(f"Cannot inspect artifact directory {artifact_dir}: {list_exc}") from list_exc
+        else:
+            raise ArtifactWriteError(
+                f"Artifact directory is not empty: {artifact_dir}. Choose a new --output-dir."
+            ) from exc
+    except OSError as exc:
+        raise ArtifactWriteError(f"Cannot create artifact directory {artifact_dir}: {exc}") from exc
+    return artifact_dir
 
 
 class ArtifactWriter:
     def __init__(self, *, artifact_dir: Path, run_id: str, benchmark: str) -> None:
-        self.artifact_dir = artifact_dir.expanduser().resolve()
+        self.artifact_dir = initialize_artifact_dir(artifact_dir)
         self.run_id = run_id
         self.benchmark = benchmark
         self.started_at = utc_now()
-        self.artifact_dir.mkdir(parents=True, exist_ok=True)
-        self.events_path = self.artifact_dir / "events.jsonl"
-        if self.events_path.exists():
-            self.events_path.unlink()
+        self._written_artifacts: set[str] = set()
 
     def path(self, name: str) -> Path:
         return self.artifact_dir / name
 
+    def _write_error(self, name: str, exc: Exception) -> ArtifactWriteError:
+        return ArtifactWriteError(f"Failed to write artifact {self.path(name)}: {exc}")
+
+    def write_json(self, name: str, payload: Mapping[str, Any]) -> Path:
+        try:
+            write_json(self.path(name), payload)
+        except Exception as exc:
+            raise self._write_error(name, exc) from exc
+        self._written_artifacts.add(name)
+        return self.path(name)
+
+    def append_jsonl(self, name: str, payload: Mapping[str, Any]) -> Path:
+        try:
+            append_jsonl(self.path(name), payload)
+        except Exception as exc:
+            raise self._write_error(name, exc) from exc
+        self._written_artifacts.add(name)
+        return self.path(name)
+
+    def write_text(self, name: str, text: str) -> Path:
+        try:
+            path = self.path(name)
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(text, encoding="utf-8")
+        except Exception as exc:
+            raise self._write_error(name, exc) from exc
+        self._written_artifacts.add(name)
+        return self.path(name)
+
+    def register_existing(self, name: str) -> None:
+        if self.path(name).exists():
+            self._written_artifacts.add(name)
+
+    def artifact_paths(self) -> dict[str, str]:
+        return {name: str(self.path(name)) for name in ARTIFACT_NAMES if name in self._written_artifacts}
+
     def event(self, phase: str, event: str, message: str, data: Mapping[str, Any] | None = None) -> None:
-        append_jsonl(
-            self.events_path,
+        self.append_jsonl(
+            "events.jsonl",
             {
                 "time": utc_now(),
                 "run_id": self.run_id,
@@ -156,25 +216,6 @@ def status(
             "summary_metrics_path": str(summary_metrics_path) if summary_metrics_path is not None else None,
             "failure": failure.to_dict() if failure is not None else None,
         }
-        write_json(self.path("status.json"), payload)
+        self.write_json("status.json", payload)
         self.event(phase, f"status_{status}", f"status={status} phase={phase}")
         return payload
-
-
-def artifact_paths(writer: ArtifactWriter) -> dict[str, str]:
-    names = (
-        "status.json",
-        "events.jsonl",
-        "resolved_benchmark.json",
-        "ingest_plan.json",
-        "query_plan.json",
-        "summary_metrics.json",
-        "environment.json",
-        "runfile.json",
-        "results.json",
-        "run.log",
-        "beir_metrics.json",
-        "beir_run.trec",
-        "query_results.jsonl",
-    )
-    return {name: str(writer.path(name)) for name in names if writer.path(name).exists() or name == "results.json"}

nemo_retriever/src/nemo_retriever/harness/beir_runner.py

@@ -4,19 +4,17 @@
 
 from __future__ import annotations
 
-from pathlib import Path
 import time
 from typing import Any, Mapping, Sequence
 
-from nemo_retriever.harness.artifact_writer import append_jsonl, ArtifactWriter
+from nemo_retriever.harness.artifact_writer import ArtifactWriter
 from nemo_retriever.harness.contracts import (
     EXIT_EVALUATION_FAILURE,
     EXIT_MISSING_INPUT,
     EXIT_QUERY_FAILURE,
     FailurePayload,
     HarnessRunError,
 )
-from nemo_retriever.harness.json_io import write_json
 from nemo_retriever.query.workflow import ResolvedQueryPlan
 from nemo_retriever.tools.recall.beir import (
     build_beir_run_from_hits,
@@ -25,17 +23,17 @@
 )
 
 
-def _write_trec_run(path: Path, run: Mapping[str, Mapping[str, float]]) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    with path.open("w", encoding="utf-8") as handle:
-        for query_id, docs in run.items():
-            ordered = sorted(docs.items(), key=lambda item: (-item[1], item[0]))
-            for rank, (doc_id, score) in enumerate(ordered, start=1):
-                handle.write(f"{query_id} Q0 {doc_id} {rank} {float(score):.6f} retriever-harness\n")
+def _trec_run_text(run: Mapping[str, Mapping[str, float]]) -> str:
+    lines: list[str] = []
+    for query_id, docs in run.items():
+        ordered = sorted(docs.items(), key=lambda item: (-item[1], item[0]))
+        for rank, (doc_id, score) in enumerate(ordered, start=1):
+            lines.append(f"{query_id} Q0 {doc_id} {rank} {float(score):.6f} retriever-harness\n")
+    return "".join(lines)
 
 
 def _write_query_result(
-    path: Path,
+    writer: ArtifactWriter,
     *,
     query_id: str,
     query_text: str,
@@ -47,8 +45,8 @@ def _write_query_result(
         ranked = dict(hit)
         ranked["rank"] = rank
         ranked_hits.append(ranked)
-    append_jsonl(
-        path,
+    writer.append_jsonl(
+        "query_results.jsonl",
         {
             "query_id": query_id,
             "query": query_text,
@@ -114,9 +112,6 @@ def run_beir_queries(
     query_kwargs = query_plan.query_kwargs()
     raw_hits: list[list[dict[str, Any]]] = []
     latencies_ms: list[float] = []
-    query_results_path = writer.path("query_results.jsonl")
-    if query_results_path.exists():
-        query_results_path.unlink()
     for query_id, query_text in zip(dataset.query_ids, dataset.queries):
         start = time.perf_counter()
         try:
@@ -137,7 +132,7 @@ def run_beir_queries(
         raw_hits.append(hit_dicts)
         latencies_ms.append(latency_ms)
         _write_query_result(
-            query_results_path,
+            writer,
             query_id=query_id,
             query_text=query_text,
             latency_ms=latency_ms,
@@ -160,6 +155,6 @@ def run_beir_queries(
                 debug_artifacts=("query_results.jsonl", "run.log"),
             ),
         ) from exc
-    write_json(writer.path("beir_metrics.json"), metrics)
-    _write_trec_run(writer.path("beir_run.trec"), run)
+    writer.write_json("beir_metrics.json", metrics)
+    writer.write_text("beir_run.trec", _trec_run_text(run))
     return latencies_ms, metrics, len(dataset.queries)