From 04e8f0a64bcdf814f315f22b1b0969f6ef347862 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 17 Sep 2026 03:57:41 +0800 Subject: [PATCH] Handle deferred Global ETF recovery safely Co-Authored-By: Codex --- .../workflows/global_etf_research_codegen.yml | 29 +- README.md | 11 +- scripts/run_global_etf_research_codegen.py | 263 +++++++++++++++-- tests/test_global_etf_research_codegen.py | 264 ++++++++++++++++++ 4 files changed, 525 insertions(+), 42 deletions(-) diff --git a/.github/workflows/global_etf_research_codegen.yml b/.github/workflows/global_etf_research_codegen.yml index 1dc62ed..877bd46 100644 --- a/.github/workflows/global_etf_research_codegen.yml +++ b/.github/workflows/global_etf_research_codegen.yml @@ -13,6 +13,11 @@ on: required: true default: false type: boolean + resume_deferred: + description: Resume one due, pre-execution Codex quota deferral + required: true + default: false + type: boolean permissions: contents: read @@ -49,6 +54,14 @@ jobs: echo "execute_and_auth_only_are_mutually_exclusive" >&2 exit 2 fi + if [ "${{ inputs.resume_deferred }}" = "true" ] && [ "${{ inputs.execute }}" != "true" ]; then + echo "resume_deferred_requires_execute" >&2 + exit 2 + fi + if [ "${{ inputs.resume_deferred }}" = "true" ] && [ "${{ inputs.auth_only }}" = "true" ]; then + echo "resume_deferred_and_auth_only_are_mutually_exclusive" >&2 + exit 2 + fi - name: Checkout frozen UsEquityStrategies base if: inputs.execute == true || inputs.auth_only == true @@ -98,7 +111,11 @@ jobs: run: | set -euo pipefail if [ "${{ inputs.execute }}" = "true" ]; then - "$CASE_ROOT/venv/bin/python" -m scripts.run_global_etf_research_codegen --execute --ues-repo-root "$GITHUB_WORKSPACE/ues-source" + resume_arg=() + if [ "${{ inputs.resume_deferred }}" = "true" ]; then + resume_arg+=(--resume-deferred) + fi + "$CASE_ROOT/venv/bin/python" -m scripts.run_global_etf_research_codegen --execute --ues-repo-root "$GITHUB_WORKSPACE/ues-source" "${resume_arg[@]}" else python3 -m scripts.run_global_etf_research_codegen fi @@ -110,15 +127,7 @@ jobs: umask 077 mkdir -p "$CASE_ROOT/artifact" chmod 700 "$CASE_ROOT/artifact" - python3 - "$HOME/.local/state/aiauditbridge/global-etf-review-20260917-auth-recovery-35124525442/result.json" "$CASE_ROOT/artifact/result.json" <<'PY' - import json, sys - from pathlib import Path - source, target = map(Path, sys.argv[1:]) - if source.is_file(): - value = json.loads(source.read_text(encoding="utf-8")) - value.pop("source", None) - target.write_text(json.dumps(value, ensure_ascii=False, sort_keys=True), encoding="utf-8") - PY + python3 -m scripts.run_global_etf_research_codegen --project-result > "$CASE_ROOT/artifact/result.json" - name: Upload only the bounded advisory result if: always() && inputs.execute == true diff --git a/README.md b/README.md index 19a49d6..36e5914 100644 --- a/README.md +++ b/README.md @@ -62,8 +62,15 @@ The manual workflow also has an `auth_only` path for its existing OIDC and audit-service health check. It cannot be combined with `execute`, and it does not read the research source, create a claim, or call a model. A claim without a terminal result remains unknown and is never retried -automatically. Only the advisory result is projected to a seven-day GitHub -artifact; source body and raw response stay private on VPS. No deployment, +automatically. A Codex response is deferred only when its saved pre-execution +quota record has the exact bounded shape; ordinary gateway failures remain +failed. Default re-entry returns a saved deferred result without another call. +The manual `resume_deferred` input is false by default and can make one due +recovery attempt only after reusing the verified claim source and rerunning the +fixed candidate tests. It keeps the original claim, result, and response, +leaves a permanent same-root recovery lock, and blocks after an unknown outcome +or another model call after a completed recovery. Only the advisory result is projected to a seven-day +GitHub artifact; source body and raw response stay private on VPS. No deployment, trading, or financial promotion authority is granted. ## Bounded SOXL research entry diff --git a/scripts/run_global_etf_research_codegen.py b/scripts/run_global_etf_research_codegen.py index a38caf6..e1be15a 100644 --- a/scripts/run_global_etf_research_codegen.py +++ b/scripts/run_global_etf_research_codegen.py @@ -14,6 +14,7 @@ from html.parser import HTMLParser import hashlib import json +import math import os from pathlib import Path import platform @@ -49,7 +50,10 @@ GLOBAL_ETF_SOURCE_REPOSITORY = "QuantStrategyLab/AIAuditBridge" _REVISION = re.compile(r"^[0-9a-f]{40}$") _SHA256 = re.compile(r"^[0-9a-f]{64}$") -_TERMINAL_STATUSES = frozenset({"review_completed", "failed"}) +_TERMINAL_STATUSES = frozenset({"review_completed", "failed", "deferred"}) +_DEFERRED_RETRY_MAX_SECONDS = 7 * 24 * 60 * 60 +_DEFERRED_QUOTA_ERROR = "codex_quota_reserved" +_DEFERRED_FAILURE_CATEGORY = "quota_or_capacity_failure" class GlobalResearchCodegenError(ValueError): @@ -229,9 +233,87 @@ def _read_json(path: Path, reason: str) -> dict[str, Any]: return value +def _trusted_deferred_retry_at(raw: Any, *, now: float | None = None) -> int | None: + """Return the bounded retry time for the one recognized pre-execution deferral.""" + if not isinstance(raw, Mapping): + return None + retry_at = raw.get("retry_at") + if ( + raw.get("status") != "deferred" + or raw.get("execution_started") is not False + or raw.get("error") != _DEFERRED_QUOTA_ERROR + or raw.get("failure_category") != _DEFERRED_FAILURE_CATEGORY + or isinstance(retry_at, bool) + or not isinstance(retry_at, (int, float)) + or not math.isfinite(retry_at) + or retry_at != int(retry_at) + ): + return None + current = datetime.now(timezone.utc).timestamp() if now is None else now + retry = int(retry_at) + return retry if current < retry <= current + _DEFERRED_RETRY_MAX_SECONDS else None + + +def _stored_deferred_timestamp(result: Mapping[str, Any], claim: Mapping[str, Any]) -> float | None: + value = result.get("deferred_at", claim.get("claimed_at")) + if not isinstance(value, str): + return None + try: + timestamp = datetime.fromisoformat(value).astimezone(timezone.utc).timestamp() + except ValueError: + return None + return timestamp if math.isfinite(timestamp) else None + + +def _deferred_result_from_response( + result: Mapping[str, Any], response: Mapping[str, Any], *, claim: Mapping[str, Any], +) -> dict[str, Any] | None: + """Recognize only the saved, trusted Codex quota deferral without exposing raw data.""" + if result.get("status") == "failed" and result.get("reason") != "global_codegen_gateway_failed": + return None + if result.get("status") not in {"failed", "deferred"}: + return None + if ( + response.get("success") is not False + or response.get("provider") != "codex" + or response.get("output") != "" + ): + return None + saved_at = _stored_deferred_timestamp(result, claim) + retry_at = _trusted_deferred_retry_at(response.get("raw"), now=saved_at) if saved_at is not None else None + if retry_at is None: + return None + deferred = dict(result) + deferred.update(status="deferred", reason="global_codegen_quota_deferred", retry_at=retry_at) + return deferred + + def _recover_existing(root: Path) -> dict[str, Any] | None: result_path = root / "result.json" claim_path = root / "claim.json" + resume_result_path = root / "resume-result.json" + resume_lock_path = root / "resume.lock" + if resume_result_path.exists(): + if not claim_path.exists(): + raise GlobalResearchCodegenError("global_codegen_terminal_claim_missing") + resume_result = _read_json(resume_result_path, "global_codegen_terminal_invalid") + claim = _read_json(claim_path, "global_codegen_claim_invalid") + _validate_stored_identity(resume_result.get("identity")) + _validate_stored_identity(claim.get("identity")) + if claim.get("identity") != resume_result.get("identity"): + raise GlobalResearchCodegenError("global_codegen_terminal_identity_mismatch") + _validate_stored_source(claim.get("source"), resume_result["identity"]) + if resume_result.get("status") not in _TERMINAL_STATUSES: + raise GlobalResearchCodegenError("global_codegen_terminal_invalid") + if resume_result.get("status") == "deferred": + response = _read_json(root / "resume-response.json", "global_codegen_response_unknown") + deferred = _deferred_result_from_response(resume_result, response, claim=claim) + if deferred is None: + raise GlobalResearchCodegenError("global_codegen_terminal_invalid") + return deferred + return resume_result + if resume_lock_path.exists(): + raise GlobalResearchCodegenError("global_codegen_resume_blocked") if result_path.exists(): if not claim_path.exists(): raise GlobalResearchCodegenError("global_codegen_terminal_claim_missing") @@ -246,6 +328,14 @@ def _recover_existing(root: Path) -> dict[str, Any] | None: _validate_stored_source(result.get("source"), result["identity"]) if result.get("status") not in _TERMINAL_STATUSES: raise GlobalResearchCodegenError("global_codegen_terminal_invalid") + response_path = root / "response.json" + if response_path.exists(): + response = _read_json(response_path, "global_codegen_response_unknown") + deferred = _deferred_result_from_response(result, response, claim=claim) + if deferred is not None: + return deferred + if result.get("status") == "deferred": + raise GlobalResearchCodegenError("global_codegen_terminal_invalid") return result if claim_path.exists(): claim = _read_json(claim_path, "global_codegen_claim_invalid") @@ -298,6 +388,65 @@ def _write_terminal(root: Path, result: dict[str, Any]) -> None: raise GlobalResearchCodegenError("global_codegen_terminal_write_unknown") from None +def _write_resume_terminal(root: Path, result: dict[str, Any]) -> None: + result.update( + no_order=True, + promotion_eligible=False, + research_only=True, + live_authority_granted=False, + ) + try: + (root / "resume-result.json").write_text( + json.dumps(result, ensure_ascii=False, sort_keys=True, allow_nan=False), encoding="utf-8", + ) + except (OSError, TypeError, ValueError): + raise GlobalResearchCodegenError("global_codegen_terminal_write_unknown") from None + + +def _write_response(root: Path, response: Any, *, filename: str) -> dict[str, Any]: + saved = { + "success": getattr(response, "success", False), "provider": getattr(response, "provider", ""), + "model": getattr(response, "model", ""), "raw": getattr(response, "raw", {}), + "output": getattr(response, "output", ""), + } + try: + (root / filename).write_text( + json.dumps(saved, ensure_ascii=False, sort_keys=True, allow_nan=False), encoding="utf-8", + ) + except (OSError, TypeError, ValueError): + raise GlobalResearchCodegenError("global_codegen_response_unknown") from None + return saved + + +def _acquire_resume_lock(root: Path, identity: Mapping[str, Any]) -> None: + lock = {"status": "resume_started", "resumed_at": datetime.now(timezone.utc).isoformat(), "identity": dict(identity)} + try: + fd = os.open(root / "resume.lock", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(lock, handle, ensure_ascii=False, sort_keys=True, allow_nan=False) + except FileExistsError: + raise GlobalResearchCodegenError("global_codegen_resume_blocked") from None + except OSError: + raise GlobalResearchCodegenError("global_codegen_resume_lock_failed") from None + + +def _load_resumable_deferred(root: Path, identity: Mapping[str, Any]) -> dict[str, Any]: + claim = _read_json(root / "claim.json", "global_codegen_claim_invalid") + result = _read_json(root / "result.json", "global_codegen_terminal_invalid") + response = _read_json(root / "response.json", "global_codegen_response_unknown") + _validate_stored_identity(claim.get("identity")) + _validate_stored_identity(result.get("identity")) + _validate_stored_source(claim.get("source"), claim["identity"]) + if claim.get("identity") != result.get("identity") or claim.get("identity") != identity: + raise GlobalResearchCodegenError("global_codegen_resume_identity_mismatch") + deferred = _deferred_result_from_response(result, response, claim=claim) + if deferred is None: + raise GlobalResearchCodegenError("global_codegen_resume_unavailable") + if deferred["retry_at"] > datetime.now(timezone.utc).timestamp(): + raise GlobalResearchCodegenError("global_codegen_resume_not_due") + return deferred + + def _prompt(files: Mapping[str, str], source: Mapping[str, Any], tests: Mapping[str, Any]) -> str: materials = "\n".join( f"File: {path}\nSHA256: {hashlib.sha256(content.encode()).hexdigest()}\n{content}" @@ -464,23 +613,41 @@ def run_global_etf_research_codegen_case( source_ref: str, execute: Callable[[str], Any] | None = None, fetch_source: Callable[[], dict[str, Any]] | None = None, candidate_test_runner: Callable[..., Mapping[str, Any]] | None = None, + resume_deferred: bool = False, ) -> dict[str, Any]: """Run one fixed Global codegen attempt with exclusive persistent claim.""" root = Path(run_root).resolve() recovered = _recover_existing(root) - if recovered is not None: + if recovered is not None and not (resume_deferred and recovered.get("status") == "deferred"): return recovered + if resume_deferred and recovered is None: + raise GlobalResearchCodegenError("global_codegen_resume_unavailable") + if resume_deferred and (root / "resume-result.json").exists(): + raise GlobalResearchCodegenError("global_codegen_resume_used") _docker_preflight() source_commit, files = _read_global_base(Path(ues_repo_root)) - source = (fetch_source or fetch_global_research_source)() - identity = _identity(source=source, source_commit=source_commit) - recovered = _claim_or_recover(root, identity, source) - if recovered is not None: - return recovered + resumed = recovered is not None + if resumed: + claim = _read_json(root / "claim.json", "global_codegen_claim_invalid") + _validate_stored_identity(claim.get("identity")) + _validate_stored_source(claim.get("source"), claim["identity"]) + source = dict(claim["source"]) + identity = dict(claim["identity"]) + if identity.get("source_commit") != source_commit: + raise GlobalResearchCodegenError("global_codegen_resume_identity_mismatch") + _load_resumable_deferred(root, identity) + _acquire_resume_lock(root, identity) + else: + source = (fetch_source or fetch_global_research_source)() + identity = _identity(source=source, source_commit=source_commit) + recovered = _claim_or_recover(root, identity, source) + if recovered is not None: + return recovered from scripts.run_new_research import _archive_codegen_base # Validate the published source before spending a model call. No model patch # is applied, and neither the checkout nor the archived source is writable. + test_result: dict[str, Any] | None = None try: with tempfile.TemporaryDirectory(prefix="aab-global-review-") as tmp: baseline = Path(tmp) / "source" @@ -490,26 +657,36 @@ def run_global_etf_research_codegen_case( raise GlobalResearchCodegenError("global_codegen_candidate_tests_failed") except Exception: result = {"status": "failed", "reason": "global_review_tests_failed", "identity": identity} - _write_terminal(root, result) + if test_result is not None: + result["candidate_tests"] = test_result + (_write_resume_terminal if resumed else _write_terminal)(root, result) return result prompt = _prompt(files, source, test_result) try: response = (execute or _codex_execute(source_ref=source_ref))(prompt) except Exception: raise GlobalResearchCodegenError("global_codegen_response_unknown") from None - response_path = root / "response.json" - try: - response_path.write_text(json.dumps({ - "success": getattr(response, "success", False), "provider": getattr(response, "provider", ""), - "model": getattr(response, "model", ""), "raw": getattr(response, "raw", {}), - "output": getattr(response, "output", ""), - }, ensure_ascii=False, sort_keys=True, allow_nan=False), encoding="utf-8") - except (OSError, TypeError, ValueError): - raise GlobalResearchCodegenError("global_codegen_response_unknown") from None - raw = getattr(response, "raw", {}) if isinstance(getattr(response, "raw", {}), Mapping) else {} + saved_response = _write_response(root, response, filename="resume-response.json" if resumed else "response.json") + raw = saved_response["raw"] if isinstance(saved_response["raw"], Mapping) else {} if getattr(response, "success", False) is not True: - result = {"status": "failed", "reason": "global_codegen_gateway_failed", "identity": identity} - _write_terminal(root, result) + retry_at = _trusted_deferred_retry_at(raw) + if ( + retry_at is not None + and saved_response.get("success") is False + and saved_response.get("provider") == "codex" + and saved_response.get("output") == "" + ): + result = { + "status": "deferred", "reason": "global_codegen_quota_deferred", "retry_at": retry_at, + "deferred_at": datetime.now(timezone.utc).isoformat(), "candidate_tests": test_result, + "identity": identity, + } + else: + result = { + "status": "failed", "reason": "global_codegen_gateway_failed", + "candidate_tests": test_result, "identity": identity, + } + (_write_resume_terminal if resumed else _write_terminal)(root, result) return result if ( getattr(response, "provider", "") != "codex" @@ -519,17 +696,19 @@ def run_global_etf_research_codegen_case( or raw.get("research_stage") != "optimization" ): result = {"status": "failed", "reason": "global_codegen_result_invalid", "identity": identity} - _write_terminal(root, result) + result["candidate_tests"] = test_result + (_write_resume_terminal if resumed else _write_terminal)(root, result) return result try: review = _validate_review(getattr(response, "output", "")) except GlobalResearchCodegenError: result = {"status": "failed", "reason": "global_review_invalid", "identity": identity} - _write_terminal(root, result) + result["candidate_tests"] = test_result + (_write_resume_terminal if resumed else _write_terminal)(root, result) return result result = {"status": "review_completed", "review": review, "changed_paths": [], "candidate_tests": test_result, "identity": identity, "advisory_only": True} - _write_terminal(root, result) + (_write_resume_terminal if resumed else _write_terminal)(root, result) return result @@ -543,16 +722,42 @@ def plan() -> dict[str, Any]: } +def _public_result(result: Mapping[str, Any]) -> dict[str, Any]: + public = {key: value for key, value in result.items() if key not in {"identity", "source"}} + public.update(no_order=True, promotion_eligible=False, live_authority_granted=False) + return public + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--execute", action="store_true", help="run only in the fixed main self-hosted workflow") parser.add_argument("--auth-preflight", action="store_true", help="verify audit-service authentication only") + parser.add_argument("--resume-deferred", action="store_true", help="resume one due, trusted quota deferral") + parser.add_argument("--project-result", action="store_true", help="project the sanitized persisted advisory result") parser.add_argument("--ues-repo-root", type=Path, default=Path("/opt/ues-source")) args = parser.parse_args(argv) + if args.project_result: + if args.execute or args.auth_preflight or args.resume_deferred: + print("global_codegen_projection_unavailable") + return 2 + try: + result = _recover_existing(GLOBAL_ETF_STATE_ROOT) + except GlobalResearchCodegenError: + result = {"status": "unknown"} + if result is None: + result = {"status": "unknown"} + print(json.dumps(_public_result(result), ensure_ascii=False, sort_keys=True, separators=(",", ":"))) + return 0 if args.auth_preflight: + if args.resume_deferred: + print("global_codegen_resume_unavailable") + return 2 outcome = _auth_preflight() print(f"auth_preflight_{outcome}") return 0 if outcome == "passed" else 1 + if args.resume_deferred and not args.execute: + print("global_codegen_resume_unavailable") + return 2 if not args.execute: print(json.dumps(plan(), ensure_ascii=False, sort_keys=True, separators=(",", ":"))) return 0 @@ -568,13 +773,11 @@ def main(argv: list[str] | None = None) -> int: print("global_codegen_unavailable") return 2 try: - result = run_global_etf_research_codegen_case(ues_repo_root=args.ues_repo_root, source_ref=environ.get("GITHUB_SHA", "")) - public = { - key: value for key, value in result.items() - if key not in {"identity", "source"} - } - public.update(no_order=True, promotion_eligible=False, live_authority_granted=False) - print(json.dumps(public, ensure_ascii=False, sort_keys=True, separators=(",", ":"))) + result = run_global_etf_research_codegen_case( + ues_repo_root=args.ues_repo_root, source_ref=environ.get("GITHUB_SHA", ""), + resume_deferred=args.resume_deferred, + ) + print(json.dumps(_public_result(result), ensure_ascii=False, sort_keys=True, separators=(",", ":"))) return 1 if result.get("status") == "failed" else 0 except GlobalResearchCodegenError as exc: print(str(exc)) diff --git a/tests/test_global_etf_research_codegen.py b/tests/test_global_etf_research_codegen.py index 7b75baa..567e792 100644 --- a/tests/test_global_etf_research_codegen.py +++ b/tests/test_global_etf_research_codegen.py @@ -4,10 +4,12 @@ import io import json import os +from datetime import timedelta from pathlib import Path import subprocess import sys import textwrap +import threading import urllib.error from contextlib import redirect_stdout from tempfile import TemporaryDirectory @@ -66,6 +68,24 @@ def test_global_codegen_docker_integration_fixture(tmp_path): class GlobalResearchCodegenTests(TestCase): + def _write_legacy_deferred(self, root: Path, *, retry_at: int) -> dict[str, object]: + source = _source() + identity = codegen._identity(source=source, source_commit=codegen.GLOBAL_ETF_RESEARCH_CODEGEN_UES_COMMIT) + codegen._claim_or_recover(root, identity, source) + claim = json.loads((root / "claim.json").read_text(encoding="utf-8")) + claim["claimed_at"] = (codegen.datetime.now(codegen.timezone.utc) - timedelta(seconds=120)).isoformat() + (root / "claim.json").write_text(json.dumps(claim), encoding="utf-8") + codegen._write_terminal(root, { + "status": "failed", "reason": "global_codegen_gateway_failed", "identity": identity, + }) + (root / "response.json").write_text(json.dumps({ + "success": False, "provider": "codex", "model": "", "output": "", "raw": { + "status": "deferred", "execution_started": False, "retry_at": retry_at, + "error": "codex_quota_reserved", "failure_category": "quota_or_capacity_failure", + }, + }), encoding="utf-8") + return identity + def test_auth_preflight_is_a_bounded_execute_or_auth_only_gate_before_research_entry(self): workflow = (Path(__file__).parents[1] / ".github/workflows/global_etf_research_codegen.yml").read_text() preflight = " - name: Verify audit-service authentication before research entry\n" @@ -88,6 +108,17 @@ def test_workflow_rejects_execute_and_auth_only_together(self): self.assertNotEqual(result.returncode, 0) self.assertIn("execute_and_auth_only_are_mutually_exclusive", result.stderr) + def test_workflow_exposes_explicit_resume_only_with_execute(self): + workflow = (Path(__file__).parents[1] / ".github/workflows/global_etf_research_codegen.yml").read_text() + inputs = workflow.split(" inputs:\n", 1)[1].split("\npermissions:", 1)[0] + self.assertIn(" resume_deferred:\n", inputs) + self.assertIn(" default: false\n", inputs.split(" resume_deferred:\n", 1)[1]) + guard = workflow.split(" - name: Reject conflicting execution modes\n", 1)[1].split("\n - name:", 1)[0] + self.assertIn("resume_deferred_requires_execute", guard) + self.assertIn("resume_deferred_and_auth_only_are_mutually_exclusive", guard) + execute = workflow.split(" - name: Run the fixed plan or execute path\n", 1)[1].split("\n - name:", 1)[0] + self.assertIn("resume_arg+=(--resume-deferred)", execute) + def test_auth_preflight_hides_exception_and_stops_before_research(self): class FakeAuthenticationError(Exception): pass @@ -287,6 +318,239 @@ def model(prompt): source_ref="a" * 40, fetch_source=lambda: self.fail("refetched"), execute=lambda _: self.fail("recalled")) self.assertEqual(replay, result) + def test_trusted_quota_deferral_preserves_completed_test_summary_and_replays_without_model(self): + retry_at = int(codegen.datetime.now(codegen.timezone.utc).timestamp()) + 60 + response = SimpleNamespace( + success=False, provider="codex", model="", output="", + raw={ + "status": "deferred", "execution_started": False, + "retry_at": retry_at, "error": "codex_quota_reserved", + "failure_category": "quota_or_capacity_failure", + }, + ) + with TemporaryDirectory() as tmp, patch.object(codegen, "_docker_preflight"), patch.object( + codegen, "_read_global_base", return_value=(codegen.GLOBAL_ETF_RESEARCH_CODEGEN_UES_COMMIT, {})), patch.dict( + sys.modules, {"scripts.run_new_research": SimpleNamespace(_archive_codegen_base=lambda *a, **k: None)}): + calls = [] + result = codegen.run_global_etf_research_codegen_case( + ues_repo_root=tmp, run_root=tmp, source_ref="a" * 40, fetch_source=_source, + candidate_test_runner=lambda *args, **kwargs: {"status": "passed", "profile": "fixed"}, + execute=lambda prompt: calls.append(prompt) or response, + ) + self.assertEqual(result["status"], "deferred") + self.assertEqual(result["reason"], "global_codegen_quota_deferred") + self.assertEqual(result["retry_at"], retry_at) + self.assertEqual(result["candidate_tests"], {"status": "passed", "profile": "fixed"}) + self.assertEqual(len(calls), 1) + replay = codegen.run_global_etf_research_codegen_case( + ues_repo_root=tmp, run_root=tmp, source_ref="a" * 40, + fetch_source=lambda: self.fail("deferred source must not run"), + execute=lambda _: self.fail("deferred model must not run"), + ) + self.assertEqual(replay, result) + + def test_legacy_gateway_failure_is_deferred_only_for_the_exact_saved_quota_shape(self): + source = _source() + identity = codegen._identity(source=source, source_commit=codegen.GLOBAL_ETF_RESEARCH_CODEGEN_UES_COMMIT) + retry_at = int(codegen.datetime.now(codegen.timezone.utc).timestamp()) + 60 + with TemporaryDirectory() as tmp: + root = Path(tmp) + codegen._claim_or_recover(root, identity, source) + codegen._write_terminal(root, { + "status": "failed", "reason": "global_codegen_gateway_failed", "identity": identity, + }) + (root / "response.json").write_text(json.dumps({"success": False, "provider": "codex", "output": "", "raw": { + "status": "deferred", "execution_started": False, "retry_at": retry_at, + "error": "codex_quota_reserved", "failure_category": "quota_or_capacity_failure", + }}), encoding="utf-8") + recovered = codegen._recover_existing(root) + self.assertEqual(recovered["status"], "deferred") + self.assertEqual(recovered["retry_at"], retry_at) + self.assertNotIn("candidate_tests", recovered) + + (root / "response.json").write_text(json.dumps({"success": False, "provider": "codex", "output": "", "raw": { + "status": "deferred", "execution_started": None, "retry_at": retry_at, + "error": "codex_quota_reserved", "failure_category": "quota_or_capacity_failure", + }}), encoding="utf-8") + self.assertEqual(codegen._recover_existing(root)["status"], "failed") + + def test_due_legacy_deferral_resumes_once_with_cached_source_and_preserves_old_evidence(self): + retry_at = int(codegen.datetime.now(codegen.timezone.utc).timestamp()) - 1 + with TemporaryDirectory() as tmp, patch.object(codegen, "_docker_preflight"), patch.object( + codegen, "_read_global_base", return_value=(codegen.GLOBAL_ETF_RESEARCH_CODEGEN_UES_COMMIT, {})), patch.dict( + sys.modules, {"scripts.run_new_research": SimpleNamespace(_archive_codegen_base=lambda *a, **k: None)}): + root = Path(tmp) + self._write_legacy_deferred(root, retry_at=retry_at) + result = codegen.run_global_etf_research_codegen_case( + ues_repo_root=root, run_root=root, source_ref="a" * 40, resume_deferred=True, + fetch_source=lambda: self.fail("resume must use the verified claim source"), + candidate_test_runner=lambda *args, **kwargs: {"status": "passed", "profile": "fixed"}, + execute=lambda _: _response(), + ) + self.assertEqual(result["status"], "review_completed") + self.assertEqual(json.loads((root / "result.json").read_text())["status"], "failed") + self.assertTrue((root / "resume.lock").exists()) + self.assertTrue((root / "resume-response.json").exists()) + self.assertEqual(codegen._recover_existing(root), result) + self.assertEqual(codegen.run_global_etf_research_codegen_case( + ues_repo_root=root, run_root=root, source_ref="a" * 40, resume_deferred=True, + execute=lambda _: self.fail("a completed recovery must not run twice"), + ), result) + + def test_resume_without_an_existing_deferred_record_stops_before_docker_or_model(self): + with TemporaryDirectory() as tmp, patch.object(codegen, "_docker_preflight") as docker: + with self.assertRaisesRegex(codegen.GlobalResearchCodegenError, "resume_unavailable"): + codegen.run_global_etf_research_codegen_case( + ues_repo_root=tmp, run_root=tmp, source_ref="a" * 40, resume_deferred=True, + fetch_source=lambda: self.fail("no source"), execute=lambda _: self.fail("no model"), + ) + docker.assert_not_called() + + def test_deferred_resume_rejects_not_due_or_locked_without_a_model_call(self): + now = int(codegen.datetime.now(codegen.timezone.utc).timestamp()) + for state in ("not_due", "locked"): + with self.subTest(state=state), TemporaryDirectory() as tmp, patch.object(codegen, "_docker_preflight") as docker, patch.object( + codegen, "_read_global_base", return_value=(codegen.GLOBAL_ETF_RESEARCH_CODEGEN_UES_COMMIT, {})): + root = Path(tmp) + self._write_legacy_deferred(root, retry_at=now + 60) + if state == "locked": + codegen._acquire_resume_lock(root, json.loads((root / "claim.json").read_text())["identity"]) + with self.assertRaises(codegen.GlobalResearchCodegenError): + codegen.run_global_etf_research_codegen_case( + ues_repo_root=root, run_root=root, source_ref="a" * 40, resume_deferred=True, + fetch_source=lambda: self.fail("no source"), execute=lambda _: self.fail("no model"), + ) + if state == "not_due": + self.assertTrue(docker.called) + else: + docker.assert_not_called() + + def test_resume_unknown_response_leaves_permanent_lock_and_no_retry_result(self): + retry_at = int(codegen.datetime.now(codegen.timezone.utc).timestamp()) - 1 + with TemporaryDirectory() as tmp, patch.object(codegen, "_docker_preflight"), patch.object( + codegen, "_read_global_base", return_value=(codegen.GLOBAL_ETF_RESEARCH_CODEGEN_UES_COMMIT, {})), patch.dict( + sys.modules, {"scripts.run_new_research": SimpleNamespace(_archive_codegen_base=lambda *a, **k: None)}): + root = Path(tmp) + self._write_legacy_deferred(root, retry_at=retry_at) + with self.assertRaisesRegex(codegen.GlobalResearchCodegenError, "response_unknown"): + codegen.run_global_etf_research_codegen_case( + ues_repo_root=root, run_root=root, source_ref="a" * 40, resume_deferred=True, + candidate_test_runner=lambda *args, **kwargs: {"status": "passed"}, + execute=lambda _: (_ for _ in ()).throw(RuntimeError("unknown")), + ) + self.assertTrue((root / "resume.lock").exists()) + self.assertFalse((root / "resume-result.json").exists()) + with self.assertRaisesRegex(codegen.GlobalResearchCodegenError, "resume_blocked"): + codegen.run_global_etf_research_codegen_case( + ues_repo_root=root, run_root=root, source_ref="a" * 40, resume_deferred=True, + execute=lambda _: self.fail("must not retry"), + ) + + def test_concurrent_resumes_make_one_model_call_and_preserve_original_evidence(self): + retry_at = int(codegen.datetime.now(codegen.timezone.utc).timestamp()) - 1 + with TemporaryDirectory() as tmp, patch.object(codegen, "_docker_preflight"), patch.object( + codegen, "_read_global_base", return_value=(codegen.GLOBAL_ETF_RESEARCH_CODEGEN_UES_COMMIT, {})), patch.dict( + sys.modules, {"scripts.run_new_research": SimpleNamespace(_archive_codegen_base=lambda *a, **k: None)}): + root = Path(tmp) + self._write_legacy_deferred(root, retry_at=retry_at) + original = {name: (root / name).read_bytes() for name in ("claim.json", "response.json", "result.json")} + barrier = threading.Barrier(2) + original_loader = codegen._load_resumable_deferred + calls, outcomes = [], [] + + def synchronized_loader(*args, **kwargs): + value = original_loader(*args, **kwargs) + barrier.wait(timeout=5) + return value + + def resume(): + try: + outcomes.append(codegen.run_global_etf_research_codegen_case( + ues_repo_root=root, run_root=root, source_ref="a" * 40, resume_deferred=True, + candidate_test_runner=lambda *args, **kwargs: {"status": "passed"}, + execute=lambda _: calls.append("model") or _response(), + )) + except codegen.GlobalResearchCodegenError as exc: + outcomes.append(str(exc)) + + with patch.object(codegen, "_load_resumable_deferred", side_effect=synchronized_loader): + threads = [threading.Thread(target=resume), threading.Thread(target=resume)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + self.assertFalse(thread.is_alive()) + self.assertEqual(calls, ["model"]) + self.assertEqual(sum(isinstance(item, dict) for item in outcomes), 1) + self.assertIn("global_codegen_resume_blocked", outcomes) + self.assertEqual({name: (root / name).read_bytes() for name in original}, original) + + def test_resume_quota_deferral_blocks_a_third_model_call(self): + retry_at = int(codegen.datetime.now(codegen.timezone.utc).timestamp()) - 1 + response = SimpleNamespace( + success=False, provider="codex", model="", output="", + raw={ + "status": "deferred", "execution_started": False, + "retry_at": retry_at + 120, "error": "codex_quota_reserved", + "failure_category": "quota_or_capacity_failure", + }, + ) + with TemporaryDirectory() as tmp, patch.object(codegen, "_docker_preflight"), patch.object( + codegen, "_read_global_base", return_value=(codegen.GLOBAL_ETF_RESEARCH_CODEGEN_UES_COMMIT, {})), patch.dict( + sys.modules, {"scripts.run_new_research": SimpleNamespace(_archive_codegen_base=lambda *a, **k: None)}): + root = Path(tmp) + self._write_legacy_deferred(root, retry_at=retry_at) + calls = [] + resumed = codegen.run_global_etf_research_codegen_case( + ues_repo_root=root, run_root=root, source_ref="a" * 40, resume_deferred=True, + candidate_test_runner=lambda *args, **kwargs: {"status": "passed"}, + execute=lambda _: calls.append("model") or response, + ) + self.assertEqual(resumed["status"], "deferred") + self.assertEqual(codegen.run_global_etf_research_codegen_case( + ues_repo_root=root, run_root=root, source_ref="a" * 40, + execute=lambda _: self.fail("default must not call"), + )["status"], "deferred") + with self.assertRaisesRegex(codegen.GlobalResearchCodegenError, "resume_used"): + codegen.run_global_etf_research_codegen_case( + ues_repo_root=root, run_root=root, source_ref="a" * 40, resume_deferred=True, + execute=lambda _: self.fail("third model call"), + ) + self.assertEqual(calls, ["model"]) + + def test_project_result_projects_deferred_resume_and_unknown_without_private_fields(self): + now = int(codegen.datetime.now(codegen.timezone.utc).timestamp()) + with TemporaryDirectory() as tmp, patch.object(codegen, "GLOBAL_ETF_STATE_ROOT", Path(tmp)): + root = Path(tmp) + self._write_legacy_deferred(root, retry_at=now + 60) + output = io.StringIO() + with redirect_stdout(output): + self.assertEqual(codegen.main(["--project-result"]), 0) + deferred = json.loads(output.getvalue()) + self.assertEqual(deferred["status"], "deferred") + self.assertNotIn("source", deferred) + self.assertNotIn("raw", deferred) + + identity = json.loads((root / "claim.json").read_text())["identity"] + codegen._write_resume_terminal(root, {"status": "review_completed", "identity": identity}) + output = io.StringIO() + with redirect_stdout(output): + self.assertEqual(codegen.main(["--project-result"]), 0) + completed = json.loads(output.getvalue()) + self.assertEqual(completed["status"], "review_completed") + self.assertNotIn("source", completed) + self.assertNotIn("raw", completed) + + with TemporaryDirectory() as tmp, patch.object(codegen, "GLOBAL_ETF_STATE_ROOT", Path(tmp)): + (Path(tmp) / "resume.lock").write_text("{}", encoding="utf-8") + output = io.StringIO() + with redirect_stdout(output): + self.assertEqual(codegen.main(["--project-result"]), 0) + self.assertEqual(json.loads(output.getvalue()), { + "live_authority_granted": False, "no_order": True, + "promotion_eligible": False, "status": "unknown", + }) + def test_global_gateway_payload_is_fixed_and_tools_are_disabled(self): payload = { "task": gateway.GLOBAL_ETF_RESEARCH_CODEGEN_TASK,