xc_agent_platform_demo/src/gate.py

"""Dual live-submit gate. Default: dry-run / blocked.

Live is allowed only when ALL of:
  - ALLOW_LIVE_SUBMIT=1
  - LIVE_CONFIRM=I_CONFIRM_LIVE_SUBMIT
  - STRATEGY_ID present

Phase2 client exists but requires explicit injection. No automatic wiring.
Queue ≥95 blocks; at most one injected submit per process.
"""

from __future__ import annotations

import os
import threading
from .queue_depth import queue_depth
from dataclasses import dataclass
from typing import Any, Callable, Dict, Optional, Protocol

from .strategy import strategy_id_present

_submit_count = 0
_submit_lock = threading.Lock()

LIVE_CONFIRM_VALUE = "I_CONFIRM_LIVE_SUBMIT"


class SubmitClient(Protocol):
    """Minimal interface for a submit client (injected fake in tests)."""

    def submit(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        ...


@dataclass(frozen=True)
class GateDecision:
    mode: str  # "dry-run" | "live" | "blocked"
    allowed: bool
    reason: str


def _allow_live_flag() -> bool:
    return os.environ.get("ALLOW_LIVE_SUBMIT", "0").strip() == "1"


def _live_confirm_ok() -> bool:
    return os.environ.get("LIVE_CONFIRM", "").strip() == LIVE_CONFIRM_VALUE


def evaluate_gate() -> GateDecision:
    """Evaluate whether live submit is permitted."""
    if not strategy_id_present():
        return GateDecision(
            mode="blocked",
            allowed=False,
            reason="STRATEGY_ID missing or empty",
        )
    if not _allow_live_flag():
        return GateDecision(
            mode="dry-run",
            allowed=False,
            reason="ALLOW_LIVE_SUBMIT is not 1 (default dry-run)",
        )
    if not _live_confirm_ok():
        return GateDecision(
            mode="blocked",
            allowed=False,
            reason=f"LIVE_CONFIRM must be exactly {LIVE_CONFIRM_VALUE!r}",
        )
    depth = queue_depth()
    if depth is None and ("ASYNC_QUEUE_DEPTH" in os.environ or "QUEUE_SNAPSHOT_JSON" in os.environ):
        return GateDecision("blocked", False, "queue depth invalid/unknown; human confirmation required")
    if depth is not None and depth >= 95:
        return GateDecision("blocked", False, "queue ≥95: 禁 live")
    return GateDecision(
        mode="live",
        allowed=True,
        reason="dual gate open (ALLOW_LIVE_SUBMIT=1 and LIVE_CONFIRM set)",
    )


def attempt_live_submit(
    payload: Dict[str, Any],
    client: Optional[SubmitClient] = None,
) -> Dict[str, Any]:
    """Attempt a live submit only if the gate opens and a client is injected.

    Phase2 TaskAddClient must be explicitly injected and dual-gated.
    Tests must use fake clients or mock HTTP only.
    """
    global _submit_count
    decision = evaluate_gate()
    if not decision.allowed:
        return {
            "ok": False,
            "mode": decision.mode,
            "reason": decision.reason,
            "submitted": False,
        }
    if client is None:
        return {
            "ok": False,
            "mode": "blocked",
            "reason": (
                "Phase2 task/add client exists; explicit client injection required; "
                "no real request without injection and dual gate"
            ),
            "submitted": False,
        }
    with _submit_lock:
        if _submit_count >= 1:
            return dict(ok=False, mode="blocked", submitted=False,
                        reason="process limit: at most 1 real add")
        # Reserve before invocation: a transport failure may still have created a task.
        _submit_count += 1
    result = client.submit(payload)
    return {
        "ok": True,
        "mode": "live",
        "reason": decision.reason,
        "submitted": True,
        "result": result,
    }