diff --git a/docs/reference/trajectory-assessment.md b/docs/reference/trajectory-assessment.md index d442b71df8..fdc375952b 100644 --- a/docs/reference/trajectory-assessment.md +++ b/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. + +中文：已有结果的复核被拒绝，或重试准备失败，不会改写上一轮回执、用量和耗时。 +新运行回执持久化后发生的失败才记入本次尝试。修复原输出后可直接复用，既不自动 +重跑模型，也不放宽结果资格检查。 + 普通阅读依次呈现指标、研究汇报和完整研究历程；汇报先给概览和可展开结论，「深入解读」 再用并排对照展开观察／竞争解释及实验条件／预期区分／反证，窄屏保持原顺序纵向阅读。 两层复用同一结果和原文入口，浏览不触发模型。新分析仍使用既有授权、 diff --git a/loopx/capabilities/benchmark_toolkit/trajectory_assessment.py b/loopx/capabilities/benchmark_toolkit/trajectory_assessment.py index 69ad903a00..77541ddec3 100644 --- a/loopx/capabilities/benchmark_toolkit/trajectory_assessment.py +++ b/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) diff --git a/tests/capabilities/test_trajectory_assessment.py b/tests/capabilities/test_trajectory_assessment.py index a745b79241..5636f6bae1 100644 --- a/tests/capabilities/test_trajectory_assessment.py +++ b/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 ): diff --git a/tests/presentation/test_research_analysis_api.py b/tests/presentation/test_research_analysis_api.py index 9f56f31feb..aaf3e0fe31 100644 --- a/tests/presentation/test_research_analysis_api.py +++ b/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"]