fix(trajectory): preserve attempt receipts across rejected reuse and retry preparation

docs/reference/trajectory-assessment.md

@@ -186,6 +186,18 @@ parsing and qualification failure. Missing usage stays unknown, and retry
 archives the failed output, receipt and observed usage before a new attempt.
 This does not reconstruct unmeasured historical usage or complete monetary cost.
 
+Rechecking a saved result can fail if its output is missing, malformed or no
+longer qualifies. That refusal is not a new analysis attempt: it preserves the
+previous receipt, observed usage and elapsed time. Retry preparation failures
+also leave that receipt unchanged. Once a new running receipt is persisted,
+process and qualification failures belong to that new attempt. Restoring the
+saved output permits ordinary reuse without another model call; it does not
+turn an invalid result into an accepted one.
+
+中文：已有结果的复核被拒绝，或重试准备失败，不会改写上一轮回执、用量和耗时。
+新运行回执持久化后发生的失败才记入本次尝试。修复原输出后可直接复用，既不自动
+重跑模型，也不放宽结果资格检查。
+
 普通阅读依次呈现指标、研究汇报和完整研究历程；汇报先给概览和可展开结论，「深入解读」
 再用并排对照展开观察／竞争解释及实验条件／预期区分／反证，窄屏保持原顺序纵向阅读。
 两层复用同一结果和原文入口，浏览不触发模型。新分析仍使用既有授权、

loopx/capabilities/benchmark_toolkit/trajectory_assessment.py

@@ -9,6 +9,7 @@
 import json
 import os
 from pathlib import Path
+import shutil
 import subprocess
 import time
 from typing import Any
@@ -126,6 +127,8 @@ def execute(
         }
     os.close(fd)
     process: subprocess.Popen[str] | None = None
+    attempt_started = False
+    prior_archive: Path | None = None
     started = time.monotonic()
     try:
         if (directory / "receipt.json").exists():
@@ -158,6 +161,7 @@ def execute(
             # explicitly retrying the same complete analysis job.
             archive = directory / f"attempt-{prior['attempt']}"
             archive.mkdir(exist_ok=False)
+            prior_archive = archive
             for name in (
                 "receipt.json",
                 "events.jsonl",
@@ -167,7 +171,12 @@ def execute(
             ):
                 path = directory / name
                 if path.exists():
-                    path.rename(archive / name)
+                    if name == "receipt.json":
+                        # Status stays readable until the next running receipt
+                        # commits; a process exit cannot look like a fresh job.
+                        shutil.copyfile(path, archive / name)
+                    else:
+                        path.rename(archive / name)
             attempt = prior["attempt"] + 1
         else:
             attempt = 1
@@ -236,6 +245,7 @@ def execute(
             "reused": False,
         }
         _write(directory / "receipt.json", receipt)
+        attempt_started = True
         if reuse_from is not None:
             # Same declared evidence, eligibility, scope and model contract.
             # Requalify against the new exact manifest; never splice labels or
@@ -316,7 +326,9 @@ def execute(
             except subprocess.TimeoutExpired:
                 process.kill()
                 process.wait()
-        if (directory / "receipt.json").exists():
+        # Reuse validation and retry preparation still belong to the old
+        # attempt. Their errors must not rewrite its status, cost or clock.
+        if attempt_started:
             receipt = read_job(directory)
             receipt.update(
                 ok=False,
@@ -326,6 +338,14 @@ def execute(
                 usage=_observed_usage(directory),
             )
             _write(directory / "receipt.json", receipt)
+        elif prior_archive is not None:
+            # Preparation did not create a new attempt. Restore its original
+            # artifacts as well as its unchanged status, cost and clock.
+            for path in prior_archive.iterdir():
+                if path.name != "receipt.json":
+                    path.replace(directory / path.name)
+            (prior_archive / "receipt.json").unlink(missing_ok=True)
+            prior_archive.rmdir()
         raise
     finally:
         lock.unlink(missing_ok=True)

tests/capabilities/test_trajectory_assessment.py

@@ -6,6 +6,7 @@
 import copy
 import os
 from pathlib import Path
+import subprocess
 import sys
 import threading
 import time
@@ -229,6 +230,129 @@ def test_failed_model_output_needs_explicit_retry_and_preserves_failure(
     assert read_job(directory / "attempt-1")["usage"] == failure["usage"]
 
 
+@pytest.mark.parametrize("damage", ["invalid_json", "invalid_result", "missing_output"])
+def test_rejected_reuse_preserves_the_completed_attempt(prepared, tmp_path, damage):
+    first = execute(prepared, tmp_path / "jobs", codex=provider(tmp_path))
+    directory = Path(first["directory"])
+    receipt = (directory / "receipt.json").read_bytes()
+    result = (directory / "assessment.json").read_bytes()
+    output = (directory / "model-output.json").read_bytes()
+    if damage == "missing_output":
+        (directory / "model-output.json").unlink()
+    else:
+        (directory / "model-output.json").write_text(
+            "{" if damage == "invalid_json" else "{}", encoding="utf-8"
+        )
+    with pytest.raises(Exception):
+        execute(prepared, tmp_path / "jobs", codex="must-not-run")
+    # A failed integrity check is not a new provider attempt. Preserve the
+    # completed attempt's status, observed cost, clock and result proof.
+    assert (directory / "receipt.json").read_bytes() == receipt
+    assert (directory / "assessment.json").read_bytes() == result
+    assert not (directory / "execution.lock").exists()
+    (directory / "model-output.json").write_bytes(output)
+    recovered = execute(prepared, tmp_path / "jobs", codex="must-not-run")
+    assert recovered["status"] == "succeeded" and recovered["reused"] is True
+    assert recovered["attempt"] == first["attempt"]
+    assert recovered["analysis_wall_seconds"] == first["analysis_wall_seconds"]
+    assert recovered["usage"] == first["usage"]
+
+
+def test_retry_preparation_failure_does_not_rewrite_the_previous_failure(prepared, tmp_path):
+    with pytest.raises(Exception, match="missing eligible"):
+        execute(prepared, tmp_path / "jobs", codex=provider(tmp_path, invalid=True))
+    directory = tmp_path / "jobs" / prepared["analysis_key"]
+    receipt = (directory / "receipt.json").read_bytes()
+    # A conflicting archive must be diagnosed before a new attempt exists.
+    archive = directory / "attempt-1"
+    archive.mkdir()
+    with pytest.raises(FileExistsError):
+        execute(prepared, tmp_path / "jobs", retry=True, codex="must-not-run")
+    assert (directory / "receipt.json").read_bytes() == receipt
+    assert not (directory / "execution.lock").exists()
+    archive.rmdir()
+    result = execute(prepared, tmp_path / "jobs", retry=True, codex=provider(tmp_path))
+    assert result["status"] == "succeeded" and result["attempt"] == 2
+    assert (archive / "receipt.json").read_bytes() == receipt
+
+
+def test_failure_after_starting_a_new_attempt_is_still_recorded(prepared, tmp_path):
+    with pytest.raises(FileNotFoundError):
+        execute(prepared, tmp_path / "jobs", codex=str(tmp_path / "missing-provider"))
+    directory = tmp_path / "jobs" / prepared["analysis_key"]
+    receipt = read_job(directory)
+    assert receipt["status"] == "failed" and receipt["attempt"] == 1
+    assert receipt["usage"] is None and receipt["analysis_wall_seconds"] is not None
+    assert not (directory / "execution.lock").exists()
+
+
+def test_cli_retry_preparation_preserves_status_and_attempt_sequence(prepared, tmp_path, monkeypatch):
+    with pytest.raises(Exception, match="missing eligible"):
+        execute(prepared, tmp_path / "jobs", codex=provider(tmp_path, invalid=True))
+    directory = tmp_path / "jobs" / prepared["analysis_key"]
+    original = {name: path.read_bytes() for name in ("receipt.json", "events.jsonl", "model-output.json")
+                if (path := directory / name).exists()}
+    binary = tmp_path / "bin"
+    binary.mkdir()
+    (binary / "codex").symlink_to(provider(tmp_path))
+    monkeypatch.setenv("PATH", str(binary) + os.pathsep + os.environ["PATH"])
+    registry = tmp_path / "registry.json"
+    registry.write_text('{"projects":{}}')
+    prefix = [sys.executable, "-m", "loopx.entrypoint", "--registry", str(registry),
+              "--runtime-root", str(tmp_path / "runtime"), "--format", "json", "benchmark"]
+    command = [*prefix, "trajectory-assess", "--input-json", str(tmp_path / "input.json"),
+               "--output-dir", str(tmp_path / "jobs"), "--model", "fixture", "--execute"]
+    obstruction = directory / "input.tmp"
+    obstruction.mkdir()
+    rejected = subprocess.run([*command, "--retry"], capture_output=True, text=True)
+    assert rejected.returncode != 0
+    status = subprocess.run([*prefix, "trajectory-assessment-status", "--directory", str(directory)],
+                            capture_output=True, text=True)
+    assert json.loads(status.stdout) == json.loads(original["receipt.json"])
+    assert all((directory / name).read_bytes() == raw for name, raw in original.items())
+    obstruction.rmdir()
+    ordinary = subprocess.run(command, capture_output=True, text=True)
+    assert json.loads(ordinary.stdout)["attempt"] == 1
+    assert "Explicit retry" in json.loads(ordinary.stdout)["reason"]
+    assert (directory / "receipt.json").read_bytes() == original["receipt.json"]
+    retry = subprocess.run([*command, "--retry"], capture_output=True, text=True, check=True)
+    assert json.loads(retry.stdout)["attempt"] == 2
+    assert json.loads(retry.stdout)["status"] == "succeeded"
+    assert all((directory / "attempt-1" / name).read_bytes() == raw for name, raw in original.items())
+
+
+def test_cli_rejected_reuse_and_source_recovery_preserve_original_receipt(prepared, tmp_path, monkeypatch):
+    first = execute(prepared, tmp_path / "jobs", codex=provider(tmp_path))
+    directory = Path(first["directory"])
+    receipt = (directory / "receipt.json").read_bytes()
+    output = (directory / "model-output.json").read_bytes()
+    (directory / "model-output.json").write_text("{}")
+    # Even an unexpected dispatch in this negative case stays synthetic.
+    binary = tmp_path / "bin"
+    binary.mkdir()
+    called = tmp_path / "unexpected-dispatch"
+    child = binary / "codex"
+    child.write_text(f"#!{sys.executable}\nfrom pathlib import Path\nPath({str(called)!r}).touch()\nraise SystemExit(99)\n")
+    child.chmod(0o700)
+    monkeypatch.setenv("PATH", str(binary) + os.pathsep + os.environ["PATH"])
+    registry = tmp_path / "registry.json"
+    registry.write_text('{"projects":{}}')
+    command = [sys.executable, "-m", "loopx.entrypoint", "--registry", str(registry),
+               "--runtime-root", str(tmp_path / "runtime"), "--format", "json",
+               "benchmark", "trajectory-assess", "--input-json", str(tmp_path / "input.json"),
+               "--output-dir", str(tmp_path / "jobs"), "--model", "fixture", "--execute"]
+    rejected = subprocess.run(command, capture_output=True, text=True)
+    assert rejected.returncode != 0
+    assert (directory / "receipt.json").read_bytes() == receipt
+    assert not called.exists() and not (directory / "execution.lock").exists()
+    (directory / "model-output.json").write_bytes(output)
+    recovered = subprocess.run(command, capture_output=True, text=True, check=True)
+    value = json.loads(recovered.stdout)
+    assert value["status"] == "succeeded" and value["reused"] is True
+    assert value["analysis_wall_seconds"] == first["analysis_wall_seconds"]
+    assert value["usage"] == first["usage"] and not called.exists()
+
+
 def test_invalid_utf8_output_fails_strictly_and_retry_preserves_bytes(
     prepared, tmp_path
 ):

tests/presentation/test_research_analysis_api.py

@@ -237,6 +237,34 @@ def test_display_context_uses_registered_sources_and_the_existing_http_reader(wo
     assert json.loads(path.read_text()) == data
 
 
+@pytest.mark.parametrize("workspace", [False, True], indirect=True)
+def test_http_rejected_reuse_keeps_completed_attempt_and_recovers_without_dispatch(workspace):
+    w, read = workspace, workspace["request"]
+    w["mode"].write_text("ok")
+    read(body=w["operation"])
+    first = read()["pack"]
+    receipts = list((w["tmp_path"] / "runtime" / "benchmark" / "trajectory-assessments").glob("*/jobs/*/receipt.json"))
+    assert len(receipts) == 1
+    receipt_path = receipts[0]
+    receipt = receipt_path.read_bytes()
+    output_path = receipt_path.parent / "model-output.json"
+    output = output_path.read_bytes()
+    output_path.write_text("{}")
+    with pytest.raises(HTTPError) as rejected:
+        read(body=w["operation"])
+    assert rejected.value.code in (400, 503)
+    assert receipt_path.read_bytes() == receipt
+    assert w["count"].read_text().splitlines() == ["started"]
+    assert not (receipt_path.parent / "execution.lock").exists()
+    output_path.write_bytes(output)
+    restored = read()["pack"]
+    assert restored["job"] == first["job"] and restored["result"] == first["result"]
+    assert read(body=w["operation"])["job"]["reused"] is True
+    assert read()["pack"]["job"]["analysis_wall_seconds"] == first["job"]["analysis_wall_seconds"]
+    assert w["count"].read_text().splitlines() == ["started"]
+    assert all(path.read_bytes() == before for path, before in w["before"].items())
+
+
 def test_frontend_language_scopes_process_retry_cache_and_original_citations(workspace):
     w, read = workspace, workspace["request"]
     zh, en, legacy = read(output_language="zh")["pack"], read(output_language="en")["pack"], read()["pack"]