vent_agent_2.0/extensions/main_fan/batch_context.py

"""宿主绑定的矿级执行上下文，不接受模型提供输出路径、凭证或取数范围。"""
from contextvars import ContextVar
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from extensions.main_fan.batch_models import ShaftPlan


@dataclass
class BatchScope:
    batch_id: str
    mine: dict
    user: str
    session: str
    output: Path
    cache_root: Path
    cutoff: str
    lookback_days: int
    report_client: Any
    shaft: ShaftPlan | None = None
    progress: Any = None
    active_workflow_id: str = ''
    report_mode: str = 'mock'
    publish: bool = False
    assessment_revision: str = ''


current_scope: ContextVar[BatchScope | None] = ContextVar('main_fan_batch_scope', default=None)


def scope_for(user, session):
    scope = current_scope.get()
    if scope is None or (scope.user, scope.session) != (user, session):
        raise ValueError('该工具仅供宿主已绑定的主通风机本地批次使用')
    return scope