feat(eval): warn when a trajectory_id appears more than once in a run

docs/outputs.md

@@ -14,6 +14,7 @@ simulated conversation, each carrying:
 - `simulation_traces` (JSON): per-turn / per-call telemetry (judge-gate ratings, fourth-wall pre-filter triggers, capitulation checks, tool-call verification, etc.).
 - A content-hashed `persona_uuid` (stable across runs) and a deterministic `trajectory_id` (hash of `persona + probe + resolved model identities + prompt version`). These are the simulator's idempotency keys.
 - In a run of stored rows (`usersim simulate --inputs`), `trajectory_id` is instead a hash of the stored row and of the setup it ran with: models as sent, the rows' shared settings, asset contents, `USERSIM_*` variables, and prompt and package versions (see [Running stored rows](engine/EXTERNAL_PROBE_RUNTIME.md#running-stored-rows)). `input_id` holds the stored row's own `trajectory_id`, so two setups run over the same stored rows pair on it. A row keeps the `usersim_provenance` it was materialized with, while the run's manifest records the code and versions the run itself used.
+- One `trajectory_id` is one conversation, once per run. A probe or extension that writes its own rows should keep what the conversation stands for (for a multi-session benchmark, say: member, arc and session, not the attempt) in `input_id`, and derive `trajectory_id` from it plus the setup, as `--inputs` does, so runs of different setups don't share ids. It should replace a restarted conversation's row rather than add one beside it. `usersim eval` warns, naming the ids, when one appears more than once in a run, and evaluates each copy as a row of its own.
 
 > **Reasoning-trace export:** `usersim select` and its Hugging Face upload path
 > retain `conversation_messages`, so captured `reasoning_content` travels with

src/usersim/cli/evaluate.py

@@ -32,6 +32,8 @@
 
 #: Trajectory columns reattached to every evaluator row for partitioning and joins.
 _KEY_COLS = ("trajectory_id", "locale", "probe_family")
+#: How many repeated ids the warning names before it only counts the rest.
+_REPEATED_IDS_SHOWN = 5
 
 
 def register(subparsers: argparse._SubParsersAction) -> None:
@@ -183,6 +185,7 @@ def _choose_rows(args: argparse.Namespace, seed_path: Path) -> tuple[Any, Any, i
         raise ValueError("--num-records and --num-records-per-locale are mutually exclusive")
     if args.num_records_per_locale is None:
         keys = pd.read_parquet(seed_path, columns=list(_KEY_COLS))
+        _warn_on_repeated_ids(keys["trajectory_id"])
         total = len(keys)
         num_records = total if args.num_records is None else min(int(args.num_records), total)
         if num_records <= 0:
@@ -195,6 +198,7 @@ def _choose_rows(args: argparse.Namespace, seed_path: Path) -> tuple[Any, Any, i
     if per_locale <= 0:
         raise ValueError("--num-records-per-locale must be positive when provided")
     full = pd.read_parquet(seed_path)
+    _warn_on_repeated_ids(full["trajectory_id"])
     sampled = pd.concat(
         [
             group.sample(n=min(per_locale, len(group)), random_state=_SAMPLE_SEED)
@@ -205,6 +209,30 @@ def _choose_rows(args: argparse.Namespace, seed_path: Path) -> tuple[Any, Any, i
     return sampled[list(_KEY_COLS)].reset_index(drop=True), sampled, len(sampled), record
 
 
+def _warn_on_repeated_ids(trajectory_ids: Any) -> None:
+    """Warn, naming them, when ids appear more than once among the run's rows.
+
+    One ``trajectory_id`` stands for one conversation. A writer that adds a
+    restarted conversation's row beside the earlier one, instead of replacing
+    it, gives that conversation two rows, and each is evaluated, so it counts
+    twice. The rows are evaluated as they are; the warning says which.
+    """
+    counts = trajectory_ids.astype(str).value_counts()
+    repeated = counts[counts > 1]
+    if repeated.empty:
+        return
+    shown = ", ".join(f"{tid} ({n} rows)" for tid, n in repeated.head(_REPEATED_IDS_SHOWN).items())
+    more = f", and {len(repeated) - _REPEATED_IDS_SHOWN} more" if len(repeated) > _REPEATED_IDS_SHOWN else ""
+    logger.warning(
+        "%d trajectory_id(s) appear more than once in this run: %s%s. Each copy is evaluated as a row of "
+        "its own, so such a conversation can count more than once; a restarted conversation's row should "
+        "replace the earlier one.",
+        len(repeated),
+        shown,
+        more,
+    )
+
+
 def _report_judge_verdicts(cells: Iterable[Any]) -> bool:
     """Warn about rows with rubric axes that no judge scored; False when no row got a verdict."""
     from usersim.engine.evaluator.generator import JudgeVerdict, judge_verdict

tests/test_cli_eval_sampling.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import json
+import logging
 from types import SimpleNamespace
 
 import pandas as pd
@@ -64,6 +65,37 @@ def test_a_per_locale_sample(self, tmp_path, monkeypatch, trajectory_df):
         assert recorded.random_seed == 42
 
 
+class TestEvalWarnsOnRepeatedIds:
+    """One ``trajectory_id`` is one conversation; a run that repeats one would score it more than once."""
+
+    @staticmethod
+    def _warnings(caplog):
+        return [r.getMessage() for r in caplog.records if "more than once" in r.getMessage()]
+
+    @pytest.mark.parametrize("flags", [(), ("--num-records", "1"), ("--num-records-per-locale", "1")])
+    def test_a_repeated_id_is_named_whatever_rows_the_pass_picks(
+        self, tmp_path, monkeypatch, trajectory_df, caplog, flags
+    ):
+        repeated = trajectory_df.copy()
+        twice = repeated.loc[0, "trajectory_id"]
+        repeated.loc[5, "trajectory_id"] = twice
+        with caplog.at_level(logging.WARNING, logger="usersim.cli.eval"):
+            out, _ = _evaluate(tmp_path, monkeypatch, repeated, *flags)
+
+        (warning,) = self._warnings(caplog)
+        assert f"{twice} (2 rows)" in warning
+        assert repeated.loc[1, "trajectory_id"] not in warning
+        if not flags:
+            # Every row is still evaluated.
+            assert len(read_partitioned_dataset(out, run=RUN_ID)) == len(repeated)
+
+    def test_a_run_without_repeats_gives_no_warning(self, tmp_path, monkeypatch, trajectory_df, caplog):
+        with caplog.at_level(logging.WARNING, logger="usersim.cli.eval"):
+            _evaluate(tmp_path, monkeypatch, trajectory_df)
+
+        assert self._warnings(caplog) == []
+
+
 @pytest.mark.parametrize("recorded", ["per_locale", None], ids=["recorded", "not_recorded"])
 def test_report_labels_the_run_by_its_recorded_pass(tmp_path, trajectory_df, evaluator_df, recorded):
     """A pass that recorded nothing about its sampling is not labelled full."""