xc_agent_platform_demo/src/gate.py

"""Live defaults on with baked-in approval. LIVE_CONFIRM remains optional.

Environment hints only block; a fresh authenticated API snapshot authorizes each add.
"""
import os
from dataclasses import dataclass
from .strategy import strategy_id_present

LIVE_CONFIRM_VALUE = 'I_CONFIRM_LIVE_SUBMIT'


def live_enabled():
    return os.environ.get('ALLOW_LIVE_SUBMIT', '1').strip() == '1'


def queue_threshold():
    value = int(os.environ.get('LIVE_QUEUE_THRESHOLD', '95'))
    if value < 1:
        raise ValueError('invalid threshold')
    return value


@dataclass(frozen=True)
class GateDecision:
    mode: str
    allowed: bool
    reason: str


def evaluate_gate():
    if not live_enabled():
        return GateDecision('dry-run', False, 'kill_switch')
    if not strategy_id_present():
        return GateDecision('blocked', False, 'STRATEGY_ID missing or empty')
    try:
        threshold = queue_threshold()
    except ValueError:
        return GateDecision('blocked', False, 'invalid_threshold')
    # Evaluate each configured hint independently; neither can override the other.
    for key in ('ASYNC_QUEUE_DEPTH', 'QUEUE_SNAPSHOT_JSON'):
        if key not in os.environ:
            continue
        if key == 'ASYNC_QUEUE_DEPTH':
            try:
                depth = int(os.environ[key])
            except ValueError:
                depth = None
        else:
            import json
            from pathlib import Path
            try:
                data = json.loads(Path(os.environ[key]).read_text())
                value = next(data[k] for k in ('depth', 'queue_depth', 'async_queue_depth', 'ASYNC_QUEUE_DEPTH') if k in data)
                depth = None if isinstance(value, bool) else int(str(value))
            except (OSError, ValueError, TypeError, StopIteration):
                depth = None
        if depth is None or depth < 0:
            return GateDecision('blocked', False, 'queue_hint_invalid')
        if depth >= threshold:
            return GateDecision('blocked', False, f'queue>={threshold} hint_count={depth}')
    return GateDecision('live', True, 'live_enabled')


def attempt_live_submit(payload, client=None, queue_reader=None):
    """Compatibility entry point; delegates to the same fresh queue cycle."""
    if client is None:
        return dict(ok=False, submitted=False, mode='blocked', reason='no real request without explicit client')
    from .live_cycle import run_cycle
    decisions = run_cycle([dict(model_address=payload.get('modelAddress'), targetGpu=payload.get('targetGpu'),
                                framework=payload.get('framework', 'vllm'), configParams=payload.get('configParams'))],
                          client=client, queue_reader=queue_reader)
    result = decisions[-1]
    return dict(result, ok=result['action'] == 'submitted', submitted=result['action'] == 'submitted',
                mode='live' if result['action'] == 'submitted' else 'blocked')