activate_v3_product.py

#!/usr/bin/env python3
"""One-shot, fail-closed activation for the standalone V3/G6 paper product."""

from __future__ import annotations

import argparse
import hashlib
import json
import shutil
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from common import BASE_DIR
from integrity_io import atomic_write_json, exclusive_file_lock
from v3_paper_account import empty_account_state
from v3_robot_equity import empty_robot_state
from v3_continuity_binding import write_binding


def _read_json(path: Path) -> dict[str, Any]:
    payload = json.loads(path.read_text(encoding="utf-8-sig"))
    if not isinstance(payload, dict):
        raise RuntimeError(f"activation_payload_not_object:{path.name}")
    return payload


def _utc(value: Any | None) -> datetime:
    if value is None:
        return datetime.now(timezone.utc)
    if isinstance(value, datetime):
        result = value
    else:
        result = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    if result.tzinfo is None:
        result = result.replace(tzinfo=timezone.utc)
    return result.astimezone(timezone.utc)


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _archive_file(source: Path, destination: Path, rows: list[dict[str, Any]], role: str) -> None:
    if not source.exists() or not source.is_file():
        return
    destination.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(source, destination)
    rows.append({
        "role": role,
        "source": str(source),
        "archive_path": str(destination),
        "bytes": destination.stat().st_size,
        "sha256": _sha256(destination),
    })


def _assert_runtime_ready(
    status: dict[str, Any],
    *,
    run_id: str,
    now: datetime,
    max_age_seconds: int,
) -> None:
    if str(status.get("run_id") or "") != run_id:
        raise RuntimeError("activation_runtime_run_id_mismatch")
    if status.get("paper_trading_only") is not True or status.get("automatic_order_allowed") is not False:
        raise RuntimeError("activation_runtime_paper_boundary_invalid")
    if status.get("status") in {"ERROR", "FIX_INFRASTRUCTURE"}:
        raise RuntimeError(f"activation_runtime_not_ready:{status.get('status')}")
    if status.get("new_formal_positions_allowed") is not False:
        raise RuntimeError("activation_prestart_position_isolation_failed")
    if status.get("formal_forward_ledgers_modified") is not False:
        raise RuntimeError("activation_prestart_formal_ledger_was_modified")
    if status.get("shadow_pipeline_errors"):
        raise RuntimeError("activation_shadow_pipeline_errors_present")
    generated = _utc(status.get("generated_at"))
    age = (now - generated).total_seconds()
    if age < -60 or age > max_age_seconds:
        raise RuntimeError(f"activation_runtime_status_stale:{age:.0f}")


def run_once(
    confirm_run_id: str,
    *,
    base_dir: Path = BASE_DIR,
    now: Any | None = None,
    regenerate_acceptance: bool = True,
    max_status_age_seconds: int = 3600,
) -> dict[str, Any]:
    base_dir = Path(base_dir).resolve()
    now = _utc(now)
    config_dir = base_dir / "config"
    results_dir = base_dir / "results"
    candidate_path = config_dir / "v3_product_candidate.json"
    runtime_path = config_dir / "v3_product_runtime_manifest.json"
    validation_path = config_dir / "v3_validation_registry.json"
    decision_path = config_dir / "v3_decision_policy.json"
    acceptance_path = results_dir / "v3_product_acceptance.json"
    runtime_status_path = results_dir / "v3_shadow_runtime_status.json"
    receipt_path = results_dir / "v3_product_activation_receipt.json"

    with exclusive_file_lock(results_dir / "v3_product_activation.lock"):
        candidate = _read_json(candidate_path)
        run_id = str(candidate.get("run_id") or "")
        if not run_id or confirm_run_id != run_id:
            raise RuntimeError("activation_exact_run_id_confirmation_required")
        if candidate.get("status") == "FROZEN_ACTIVE":
            if receipt_path.exists():
                receipt = _read_json(receipt_path)
                if str(receipt.get("run_id") or "") == run_id:
                    return receipt
            raise RuntimeError("activation_active_without_matching_receipt")
        if candidate.get("status") != "PREPARED_NOT_ACTIVE":
            raise RuntimeError(f"activation_candidate_status_invalid:{candidate.get('status')}")
        if candidate.get("paper_trading_only") is not True:
            raise RuntimeError("activation_candidate_not_paper_only")

        if regenerate_acceptance:
            if base_dir != Path(BASE_DIR).resolve():
                raise RuntimeError("activation_acceptance_regeneration_requires_runtime_base")
            from v3_product_acceptance import run_once as run_acceptance

            run_acceptance(acceptance_path)
        acceptance = _read_json(acceptance_path)
        if acceptance.get("status") != "PASS" or str(acceptance.get("run_id") or "") != run_id:
            raise RuntimeError("activation_product_acceptance_not_passed")
        if acceptance.get("order_sent") is not False or acceptance.get("paper_trading_only") is not True:
            raise RuntimeError("activation_acceptance_boundary_invalid")
        runtime_status = _read_json(runtime_status_path)
        _assert_runtime_ready(
            runtime_status,
            run_id=run_id,
            now=now,
            max_age_seconds=max_status_age_seconds,
        )

        stamp = now.strftime("%Y%m%dT%H%M%SZ")
        artifact_dir = base_dir / "artifacts" / run_id / "activation" / stamp
        preactivation_dir = artifact_dir / "preactivation"
        archived: list[dict[str, Any]] = []
        sources = {
            "paper_account.json": base_dir / "data" / "v3_paper_next" / "paper_account.json",
            "robot_accounts.json": base_dir / "data" / "v3_robot_subaccounts" / "accounts.json",
            "shadow_event_ledger.jsonl": base_dir / "data" / "v3_shadow" / "robot_event_ledger.jsonl",
            "independent_forward_events.jsonl": results_dir / "v3_independent_forward_events.jsonl",
            "shadow_runtime_status.json": runtime_status_path,
            "independent_forward_status.json": results_dir / "v3_independent_forward_status.json",
            "product_acceptance.json": acceptance_path,
        }
        for name, source in sources.items():
            _archive_file(source, preactivation_dir / name, archived, "legacy_or_pre_activation_read_only")

        atomic_write_json(
            base_dir / "data" / "v3_paper_next" / "paper_account.json",
            empty_account_state(run_id=run_id, evidence_role="formal_v3_product_forward"),
        )
        atomic_write_json(
            base_dir / "data" / "v3_robot_subaccounts" / "accounts.json",
            empty_robot_state(run_id=run_id, evidence_role="formal_v3_product_forward"),
        )

        formal_start = now.isoformat(timespec="seconds")
        candidate.update({
            "status": "FROZEN_ACTIVE",
            "frozen_at": formal_start,
            "formal_forward_start": formal_start,
            "first_partial_session_counts_as_complete_session": False,
            "legacy_archive": str(preactivation_dir),
            "activation_receipt": str(receipt_path),
        })
        runtime = _read_json(runtime_path)
        runtime.update({
            "mode": "INDEPENDENT_FORMAL_FORWARD_PAPER",
            "formal_forward_collection_active": True,
            "formal_forward_start": formal_start,
            "formal_forward_evidence_role": "prospective_only_after_product_activation",
        })
        validation = _read_json(validation_path)
        validation.update({
            "run_id": run_id,
            "parent_run_id": candidate.get("parent_run_id"),
            "status": "FROZEN_SCORER_READY_VALIDATION_ACCUMULATING",
        })
        decision = _read_json(decision_path)
        decision.update({
            "run_id": run_id,
            "parent_run_id": candidate.get("parent_run_id"),
            "formal_forward_evidence": True,
        })
        for path, payload in (
            (candidate_path, candidate),
            (runtime_path, runtime),
            (validation_path, validation),
            (decision_path, decision),
        ):
            atomic_write_json(path, payload)

        continuity_binding = write_binding(
            config_dir / "v3_p0_continuity_binding.json",
            run_id=run_id,
            effective_from=formal_start,
            base_dir=base_dir,
            repair_run_id="activation_source_identity",
        )

        frozen_files: list[dict[str, Any]] = []
        for path in (
            candidate_path,
            runtime_path,
            validation_path,
            decision_path,
            config_dir / "product_policy_v1.json",
            config_dir / "v3_institutional_group.json",
        ):
            _archive_file(path, artifact_dir / "frozen_config" / path.name, frozen_files, "frozen_product_config")
        manifest = {
            "schema_version": "v3_product_activation_manifest_1",
            "run_id": run_id,
            "parent_run_id": candidate.get("parent_run_id"),
            "activated_at": formal_start,
            "formal_forward_start": formal_start,
            "first_partial_session_counts_as_complete_session": False,
            "formal_accounts_started_clean": True,
            "shadow_and_independent_ledgers_preserved": True,
            "legacy_evidence_excluded_from_new_run": True,
            "v2_runtime_dependency_allowed": False,
            "order_sent": False,
            "paper_trading_only": True,
            "continuity_binding": {
                "path": str(config_dir / "v3_p0_continuity_binding.json"),
                "evidence_epoch_id": continuity_binding["evidence_epoch_id"],
                "source_identity_sha256": continuity_binding["source_identity"]["tree_sha256"],
            },
            "archived_files": archived,
            "frozen_files": frozen_files,
        }
        manifest_path = artifact_dir / "ACTIVATION_MANIFEST.json"
        atomic_write_json(manifest_path, manifest)
        receipt = {
            "schema_version": "v3_product_activation_receipt_1",
            "status": "FROZEN_ACTIVE",
            "run_id": run_id,
            "activated_at": formal_start,
            "formal_forward_start": formal_start,
            "manifest": str(manifest_path),
            "archive": str(preactivation_dir),
            "paper_trading_only": True,
            "order_sent": False,
            "evidence_epoch_id": continuity_binding["evidence_epoch_id"],
            "source_identity_sha256": continuity_binding["source_identity"]["tree_sha256"],
        }
        atomic_write_json(receipt_path, receipt)
        return receipt


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--confirm-run-id", required=True)
    parser.add_argument("--max-status-age-seconds", type=int, default=3600)
    args = parser.parse_args()
    result = run_once(
        args.confirm_run_id,
        max_status_age_seconds=args.max_status_age_seconds,
    )
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()