diff --git a/scripts/research/local_quant_research/decision.py b/scripts/research/local_quant_research/decision.py deleted file mode 100644 index d43bab4..0000000 --- a/scripts/research/local_quant_research/decision.py +++ /dev/null @@ -1,179 +0,0 @@ -from __future__ import annotations - -import hashlib -import json -import os -import re -import shutil -import uuid -from pathlib import Path -from typing import Mapping - - -_SHA256 = re.compile(r"[0-9a-f]{64}") -_PROJECT_ID = re.compile(r"[a-z0-9][a-z0-9._-]{0,63}") -_DECISIONS = { - "proceed_to_joinquant", - "revise_and_reassess", - "stop_evidence_insufficient", -} -_REQUIRED = { - "decision", - "candidate_focus", - "baseline_action", - "reason", - "confirmed_by", - "confirmed_at", -} - - -class DecisionError(ValueError): - """Raised when a human decision does not match immutable research evidence.""" - - -def _canonical(value: object) -> bytes: - return json.dumps( - value, - ensure_ascii=False, - sort_keys=True, - separators=(",", ":"), - allow_nan=False, - ).encode("utf-8") - - -def _digest(value: object) -> str: - return hashlib.sha256(_canonical(value)).hexdigest() - - -def _file_digest(path: Path) -> str: - try: - return hashlib.sha256(path.read_bytes()).hexdigest() - except OSError as exc: - raise DecisionError("research evidence is missing") from exc - - -def _recommendation(run_dir: Path) -> tuple[dict[str, object], str]: - path = Path(run_dir) / "recommendation.json" - try: - value = json.loads(path.read_text(encoding="utf-8")) - except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: - raise DecisionError("recommendation evidence is invalid") from exc - if not isinstance(value, dict): - raise DecisionError("recommendation evidence is invalid") - identity = value.get("identity") - run_id = identity.get("run_id") if isinstance(identity, Mapping) else None - if not isinstance(run_id, str) or _SHA256.fullmatch(run_id) is None: - raise DecisionError("recommendation run identity is invalid") - return value, run_id - - -def _document( - run_dir: Path, - project_id: str, - decision: Mapping[str, object], -) -> dict[str, object]: - if _PROJECT_ID.fullmatch(project_id) is None: - raise DecisionError("project_id is invalid") - if set(decision) != _REQUIRED: - raise DecisionError("human decision fields are incomplete or unknown") - if decision["decision"] not in _DECISIONS: - raise DecisionError("human decision value is invalid") - if not isinstance(decision["candidate_focus"], list) or any( - not isinstance(item, str) or not item for item in decision["candidate_focus"] - ): - raise DecisionError("candidate_focus must be a string array") - for field in ("baseline_action", "reason", "confirmed_by", "confirmed_at"): - if not isinstance(decision[field], str) or not str(decision[field]).strip(): - raise DecisionError(f"{field} must be non-empty") - _, run_id = _recommendation(run_dir) - payload = { - "schema_version": 1, - "project_id": project_id, - "run_id": run_id, - "report_sha256": _file_digest(Path(run_dir) / "local-research-report.md"), - "recommendation_sha256": _file_digest(Path(run_dir) / "recommendation.json"), - **dict(decision), - } - payload["decision_id"] = _digest(payload) - payload["document_sha256"] = _digest(payload) - return payload - - -def record_human_decision( - *, - run_dir: Path, - decision_root: Path, - project_id: str, - decision: Mapping[str, object], -) -> Path: - document = _document(Path(run_dir), project_id, decision) - target = ( - Path(decision_root) - / project_id - / str(document["run_id"]) - / str(document["decision_id"]) - ) - output = target / "human-decision.json" - if output.exists(): - if ( - validate_human_decision( - run_dir=run_dir, - project_id=project_id, - decision_path=output, - ) - != document - ): - raise DecisionError( - "existing human decision conflicts with requested decision" - ) - return output - target.parent.mkdir(parents=True, exist_ok=True) - temporary = target.parent / f".{target.name}.tmp-{uuid.uuid4().hex}" - temporary.mkdir() - try: - (temporary / "human-decision.json").write_text( - json.dumps(document, ensure_ascii=False, sort_keys=True, indent=2) + "\n", - encoding="utf-8", - ) - os.replace(temporary, target) - except Exception: - shutil.rmtree(temporary, ignore_errors=True) - raise - return output - - -def validate_human_decision( - *, - run_dir: Path, - project_id: str, - decision_path: Path, -) -> dict[str, object]: - try: - document = json.loads(Path(decision_path).read_text(encoding="utf-8")) - except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: - raise DecisionError("human decision is invalid") from exc - if not isinstance(document, dict): - raise DecisionError("human decision is invalid") - semantic = { - key: value for key, value in document.items() if key != "document_sha256" - } - if document.get("document_sha256") != _digest(semantic): - raise DecisionError("human decision document digest mismatch") - _, run_id = _recommendation(Path(run_dir)) - if document.get("run_id") != run_id: - raise DecisionError("human decision run identity mismatch") - if document.get("project_id") != project_id: - raise DecisionError("human decision project identity mismatch") - if document.get("report_sha256") != _file_digest( - Path(run_dir) / "local-research-report.md" - ) or document.get("recommendation_sha256") != _file_digest( - Path(run_dir) / "recommendation.json" - ): - raise DecisionError("human decision evidence digest mismatch") - project_path = Path(decision_path).parents[2] - if project_path.name != project_id: - raise DecisionError("human decision path does not match its project") - expected_parent = project_path / run_id / str(document.get("decision_id")) - if Path(decision_path).parent.resolve() != expected_parent.resolve(): - raise DecisionError("human decision path does not match its identity") - return document diff --git a/scripts/research/local_quant_research/evidence.py b/scripts/research/local_quant_research/evidence.py index 1c978b3..5b50512 100644 --- a/scripts/research/local_quant_research/evidence.py +++ b/scripts/research/local_quant_research/evidence.py @@ -291,13 +291,13 @@ def validate_complete_run( raise EvidenceError("completed output digest mismatch") status = _load_json(Path(run_dir) / "project-status.json") - expected_status = {"schema_version": 1, "status": "complete", "reason_codes": []} - accepted_statuses = [ - expected_status, - {**expected_status, "next_action": "human_confirmation_required"}, - {**expected_status, "next_action": "return_to_caller"}, - ] - if status not in accepted_statuses: + expected_status = { + "schema_version": 1, + "status": "complete", + "reason_codes": [], + "next_action": "return_to_caller", + } + if status != expected_status: raise EvidenceError("completed project status is invalid") expected_files = {"run-manifest.json", "project-status.json"} for item in outputs: diff --git a/scripts/research/local_quant_research/runner.py b/scripts/research/local_quant_research/runner.py index 8291c0f..6dfee6c 100644 --- a/scripts/research/local_quant_research/runner.py +++ b/scripts/research/local_quant_research/runner.py @@ -683,21 +683,12 @@ def _project_status(staging: Path) -> tuple[str, tuple[str, ...]]: if status == "complete" and reasons: raise EvidenceError("complete project status must not contain reasons") next_action = document.get("next_action") - if next_action is not None and ( - status != "complete" - or next_action - not in {"human_confirmation_required", "return_to_caller"} - ): + expected_next_action = "return_to_caller" if status == "complete" else None + if next_action != expected_next_action: raise EvidenceError("project status next_action is invalid") return status, tuple(reasons) -def _project_next_action(staging: Path) -> str | None: - document = json.loads((Path(staging) / "project-status.json").read_text(encoding="utf-8")) - value = document.get("next_action") - return value if isinstance(value, str) else None - - def _actual_staging_files(staging: Path) -> set[str]: files: set[str] = set() for path in staging.rglob("*"): @@ -842,7 +833,7 @@ def run_project(config_path: Path, *, repo_root: Path) -> RunResult: reused=True, reasons=(), stages=complete_stages, - next_action=_project_next_action(run_dir), + next_action="return_to_caller", ) project_root.mkdir(parents=True, exist_ok=True) @@ -1013,7 +1004,6 @@ def run_project(config_path: Path, *, repo_root: Path) -> RunResult: try: project_status, reason_codes = _project_status(staging) - next_action = _project_next_action(staging) except EvidenceError: return _attempt_result( repo_root=repo_root, @@ -1148,5 +1138,5 @@ def run_project(config_path: Path, *, repo_root: Path) -> RunResult: reused=False, reasons=(), stages=tuple(StageRecord(name, "complete") for name in _COMPLETE_STAGE_NAMES), - next_action=next_action, + next_action="return_to_caller", ) diff --git a/tests/local_quant_research/test_human_decision.py b/tests/local_quant_research/test_human_decision.py deleted file mode 100644 index 7ed905a..0000000 --- a/tests/local_quant_research/test_human_decision.py +++ /dev/null @@ -1,98 +0,0 @@ -from __future__ import annotations - -import hashlib -import json -from pathlib import Path - -import pytest - -from scripts.research.local_quant_research.decision import ( - DecisionError, - record_human_decision, - validate_human_decision, -) - - -def _run(tmp_path: Path) -> tuple[Path, str]: - run_id = "a" * 64 - run_dir = tmp_path / "run" - run_dir.mkdir() - (run_dir / "local-research-report.md").write_text("report\n", encoding="utf-8") - (run_dir / "recommendation.json").write_text( - json.dumps( - {"identity": {"run_id": run_id}, "recommendation": "revise_and_reassess"} - ), - encoding="utf-8", - ) - return run_dir, run_id - - -def test_human_decision_is_append_only_outside_immutable_run(tmp_path: Path) -> None: - run_dir, run_id = _run(tmp_path) - decision_root = tmp_path / "decisions" - decision = { - "decision": "revise_and_reassess", - "candidate_focus": ["entry-40"], - "baseline_action": "retain_frozen_baseline", - "reason": "等待人工复核稳健性反证", - "confirmed_by": "research-owner", - "confirmed_at": "2026-07-14T12:00:00+08:00", - } - - path = record_human_decision( - run_dir=run_dir, - decision_root=decision_root, - project_id="strategy-003", - decision=decision, - ) - document = validate_human_decision( - run_dir=run_dir, - project_id="strategy-003", - decision_path=path, - ) - - assert path.parent.parent.name == run_id - assert not path.is_relative_to(run_dir) - assert document["decision"] == "revise_and_reassess" - assert ( - document["report_sha256"] - == hashlib.sha256( - (run_dir / "local-research-report.md").read_bytes() - ).hexdigest() - ) - assert ( - record_human_decision( - run_dir=run_dir, - decision_root=decision_root, - project_id="strategy-003", - decision=decision, - ) - == path - ) - - -def test_human_decision_rejects_changed_report_or_recommendation( - tmp_path: Path, -) -> None: - run_dir, _ = _run(tmp_path) - path = record_human_decision( - run_dir=run_dir, - decision_root=tmp_path / "decisions", - project_id="strategy-003", - decision={ - "decision": "stop_evidence_insufficient", - "candidate_focus": [], - "baseline_action": "retain_frozen_baseline", - "reason": "证据不足", - "confirmed_by": "research-owner", - "confirmed_at": "2026-07-14T12:00:00+08:00", - }, - ) - (run_dir / "local-research-report.md").write_text("changed\n", encoding="utf-8") - - with pytest.raises(DecisionError, match="digest"): - validate_human_decision( - run_dir=run_dir, - project_id="strategy-003", - decision_path=path, - ) diff --git a/tests/local_quant_research/test_runner.py b/tests/local_quant_research/test_runner.py index d5ca5d6..b7cf3b0 100644 --- a/tests/local_quant_research/test_runner.py +++ b/tests/local_quant_research/test_runner.py @@ -12,9 +12,12 @@ import pyarrow.parquet as pq from scripts.research.local_quant_research.contracts import OutputSpec -from scripts.research.local_quant_research.evidence import collect_output_evidence +from scripts.research.local_quant_research.evidence import ( + EvidenceError, + canonical_digest, + collect_output_evidence, +) from scripts.research.local_quant_research.runner import _project_status, run_project -from scripts.research.local_quant_research.evidence import canonical_digest from scripts.research.market_data.contracts import SnapshotSelection from scripts.research.market_data.storage import create_snapshot, import_batch @@ -34,9 +37,15 @@ "high_limit", "low_limit", ) +COMPLETE_STATUS = { + "schema_version": 1, + "status": "complete", + "reason_codes": [], + "next_action": "return_to_caller", +} -def test_complete_project_status_can_declare_human_confirmation_next_action( +def test_complete_project_status_rejects_human_confirmation_next_action( tmp_path: Path, ) -> None: _write_json( @@ -49,7 +58,8 @@ def test_complete_project_status_can_declare_human_confirmation_next_action( }, ) - assert _project_status(tmp_path) == ("complete", ()) + with pytest.raises(EvidenceError, match="next_action"): + _project_status(tmp_path) def test_complete_project_status_can_return_single_scenario_to_caller( @@ -185,7 +195,7 @@ def fake_run(command: list[str], **kwargs): output_dir = _output_dir(command) _write_json( output_dir / "project-status.json", - {"schema_version": 1, "status": "complete", "reason_codes": []}, + COMPLETE_STATUS, ) _write_json(output_dir / "result.json", {"answer": 42}) return subprocess.CompletedProcess(command, 0, stdout="ignored", stderr="ignored") @@ -454,6 +464,35 @@ def test_same_complete_identity_is_revalidated_and_reused_without_execution( assert {path.name: _sha256(path) for path in second.run_path.iterdir()} == before +def test_complete_run_with_human_confirmation_status_is_not_reused( + repo_root: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_root, config_path, _ = _build_repo(tmp_path, repo_root) + monkeypatch.setattr(subprocess, "run", _successful_process()) + first = run_project(config_path, repo_root=fake_root) + _write_json( + first.run_path / "project-status.json", + { + "schema_version": 1, + "status": "complete", + "reason_codes": [], + "next_action": "human_confirmation_required", + }, + ) + monkeypatch.setattr( + subprocess, + "run", + lambda *args, **kwargs: pytest.fail("invalid complete run must not execute"), + ) + + result = run_project(config_path, repo_root=fake_root) + + assert result.status == "failed" + assert result.reused is False + + def test_tampered_complete_output_fails_without_overwriting_old_run( repo_root: Path, tmp_path: Path, @@ -642,7 +681,7 @@ def test_missing_required_output_is_failed(tmp_path: Path, repo_root: Path, monk def process(command: list[str], **kwargs): _write_json( _output_dir(command) / "project-status.json", - {"schema_version": 1, "status": "complete", "reason_codes": []}, + COMPLETE_STATUS, ) return subprocess.CompletedProcess(command, 0, stdout="", stderr="") @@ -696,7 +735,7 @@ def process(command: list[str], **kwargs): output_dir = _output_dir(command) _write_json( output_dir / "project-status.json", - {"schema_version": 1, "status": "complete", "reason_codes": []}, + COMPLETE_STATUS, ) _write_json(output_dir / "result.json", {"input": value_used}) return subprocess.CompletedProcess(command, 0, stdout="", stderr="")