vent_agent_2.0/extensions/main_fan/workflow.py

"""主通风机固定工作流：完整 PDF 提取 + 现成监测数据 → 一次 Qwen JSON 研判。"""
from __future__ import annotations

import asyncio
from datetime import datetime
import json
import math
from pathlib import Path
import time
from typing import Literal
from uuid import uuid4
from zoneinfo import ZoneInfo

import httpx
from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator

import config
from agents.model_factory import _is_qwen35_model, _qwen_thinking_payload, normalize_thinking
from billing.meter import UsageMeter
from db import business_store, chat_store, db_conn, main_fan_store
from extensions.main_fan import exploration, history_compaction, history_processing, workflow_json_repair, workflow_monitoring
from extensions.main_fan.batch_context import scope_for
from extensions.main_fan.schemas import MainFanResult
from extensions.main_fan.workflow_documents import OCRSettings, PDFExtractor, service_url
from extensions.main_fan.workflow_page_filter import select_pages
from extensions.registry import load_extensions
from obs import get_logger, get_trace_id, log_context
from runtime import business, execution
from runtime.context_meter import estimate_tokens
from runtime.engine import ExecutionResult
from runtime.python_sandbox import digest, workspace_file
from runtime.result_models import PlatformJudgment, PlatformResultPayload

logger = get_logger(__name__)
VERSION = 'main-fan-workflow-v5'
LOCAL_ZONE = ZoneInfo('Asia/Shanghai')
JUDGMENT_POLICY = {
    'version': 'main-fan-precision-first-v2-monitoring-rules',
    'priority': 'precision',
    'monitoring': '固定代码按本轮冻结规则生成monitoring_assessment；已确认的监测告警不能被模型降级，不增加稳定窗口等新条件。',
    'performance': '辅助性粗略核对；只有清晰、可比且确凿的异常证据才告警，识别不清或资料不足只说明限制。',
    'normal': '有可用数据且未发现确凿异常；不表示未核实项目已被证明合格或安全。',
    'undetermined': '仅当报告、开停、风量三类均没有可用数据。',
}


class WorkflowJudgment(PlatformJudgment):
    # 让请求 schema 与本地约束一致；本地文件没有公开证据链接。
    evidenceLink: Literal[''] = Field(...)


class WorkflowDecision(BaseModel):
    """模型填写业务内容；平台身份、任务标识和完成时刻由宿主填写。"""
    model_config = ConfigDict(extra='forbid', strict=True)
    objectDetail: str = Field(min_length=1)
    dataDetail: str
    judgment: WorkflowJudgment

    @field_validator('objectDetail')
    @classmethod
    def nonblank_object(cls, value):
        if not value.strip():
            raise ValueError('研判对象不能为空白')
        return value

    @field_validator('judgment')
    @classmethod
    def complete_judgment(cls, value):
        if not value.result.strip() or not value.reason.strip():
            raise ValueError('研判摘要和正文不能为空白')
        if value.evidenceLink:
            raise ValueError('本地工作流没有公开证据页，evidenceLink 须为空字符串')
        return value


class WorkflowError(ValueError):
    def __init__(self, code, message, *, details=None):
        self.code = code
        self.details = details or {}
        super().__init__(message)


def publish_enabled():
    # 新工作流默认用于服务器本地核验；不能因为沿用旧环境的总开关而直接上报。
    return (config.get_bool('MAIN_FAN_WORKFLOW_PUBLISH_ENABLED', False)
            and config.get_bool('RESULT_PLATFORM_ENABLED', False))


def qwen_http_client():
    return httpx.AsyncClient(trust_env=False)


def full_monitoring(context, user, session):
    prepared = context['prepared_data']
    compact = exploration._read_json(workspace_file(user, session, prepared['agent_data_path']))
    details_path = compact.get('details_path')
    if details_path:
        if details_path not in context.get('query_exports', {}):
            raise ValueError('监测详情不属于本轮已核验输入')
        details = exploration._read_json(workspace_file(user, session, details_path))
        # 不重新取数或计算切换/比例，只展开固定预处理已经产生的全部结果。
        view = history_compaction.agent_view(details, details_path, full=True)
        compact = history_processing.agent_input(view)
        compact['state_period_semantics'] = view['state_period_semantics']
    if compact.get('switch_candidates_truncated') or any(row.get('truncated') for row in compact['states']):
        raise ValueError('工作流需要完整监测结果，不能将截断预览交给唯一一次模型调用')
    return compact


def build_input(workflow, monitoring, reports, scope, *, availability=None, monitoring_assessment=None):
    context = workflow['context']
    availability = data_availability(reports, monitoring) if availability is None else availability
    monitoring_assessment = (workflow_monitoring.assess(monitoring, context['policy'])
                             if monitoring_assessment is None else monitoring_assessment)
    references = [{'id': row['id'], 'from_fan': row['from_fan'], 'to_fan': row['to_fan'],
                   'flow': row.get('flow', []), 'comparison': row.get('comparison', {})}
                  for row in monitoring['switch_candidates']]
    return {'workflow_version': VERSION, 'workflow_id': workflow['id'],
        'identity': {'mine_code': scope.mine['mine_id'], 'mine_name': scope.mine['mine_name'],
            'dept_id': scope.mine['dept_id'], 'shaft': scope.shaft.model_dump(mode='json'),
            'fan_pair': context['task_input'].get('fan_pair', [])},
        'rules': context['policy'],
        'judgment_policy': JUDGMENT_POLICY,
        'data_availability': availability,
        'monitoring_assessment': monitoring_assessment,
        'pdf': reports,
        'airflow': {'range': monitoring['range'], 'flows': monitoring['flows'],
                    'switch_references': references, 'notes': monitoring.get('notes', [])},
        'switching': {'range': monitoring['range'], 'states': monitoring['states'],
                      'candidates': [{**row, 'duration_minutes': row.get('duration_minutes'),
                          'duration_status': row.get('duration_status') or (
                              'unknown' if row.get('duration_minutes') is None else 'provided')}
                          for row in monitoring['switch_candidates']],
                      'time_range_semantics': '整个候选的采样定位范围，不用起止相减；直接采用duration_minutes及其采样上下界。'
                          'estimated表示按实际开停变化观测估算，只有无法配对时间时才为unknown。',
                      'notes': monitoring.get('notes', []),
                      'state_period_semantics': monitoring.get('state_period_semantics', '')},
        'history_coverage': context.get('history_manifest', {}) and {
            key: context['history_manifest'][key] for key in ('cutoff', 'lookback_days', 'point_count',
                'expected_segments', 'completed_segments', 'failed_segments', 'coverage_complete')},
        'output_schema': decision_schema(availability, monitoring_assessment)}


def decision_schema(availability=None, monitoring_assessment=None):
    schema = WorkflowDecision.model_json_schema()
    if availability is not None:
        schema['$defs']['WorkflowJudgment']['properties']['resultCode']['enum'] = (
            ['NORMAL', 'WARNING', 'ABNORMAL'] if availability['any_available'] else ['UNDETERMINED'])
    if monitoring_assessment and monitoring_assessment['minimum_result_code'] != 'NORMAL':
        minimum = workflow_monitoring.RANK[monitoring_assessment['minimum_result_code']]
        schema['$defs']['WorkflowJudgment']['properties']['resultCode']['enum'] = [
            code for code, rank in workflow_monitoring.RANK.items() if rank >= minimum]
    return schema


def response_format(availability=None, monitoring_assessment=None):
    return {'type': 'json_schema', 'json_schema': {'name': 'main_fan_workflow_decision',
        'strict': True, 'schema': decision_schema(availability, monitoring_assessment)}}


def data_availability(reports, monitoring):
    def finite(value):
        return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)
    readable_pages = sum(page['status'] == 'completed' and bool(page.get('text', '').strip())
        for report in reports if not report.get('is_mock') for page in report['pages'])
    known_states = {'开', '停', '开机', '停机', '运行', '停止', 'running', 'stopped', 'on', 'off'}
    state_periods = sum(str(period.get('state', '')).strip().lower() in known_states
        for point in monitoring.get('states', []) for period in point.get('periods', []))
    candidates = monitoring.get('switch_candidates', [])
    durations = sum(finite(row.get('duration_minutes')) and row['duration_minutes'] >= 0 for row in candidates)
    samples = sum(max(0, row['samples']) for row in monitoring.get('flows', []) if finite(row.get('samples')))
    values = sum(finite(value) for point in monitoring.get('flows', []) for value in point.get('range', []))
    values += sum(finite(row.get('comparison', {}).get(key)) for row in candidates
                  for key in ('before_value', 'after_value', 'change_percent'))
    values += sum(isinstance(sample, (list, tuple)) and len(sample) == 2 and finite(sample[1])
        for row in candidates for point in row.get('flow', [])
        for sample in (point.get('before'), point.get('after')))
    comparisons = sum(all(finite(row.get('comparison', {}).get(key))
        for key in ('before_value', 'after_value', 'change_percent')) for row in candidates)
    return {'pdf_readable_pages': readable_pages, 'known_state_periods': state_periods,
        'known_switch_intervals': durations, 'flow_numeric_samples': samples, 'flow_numeric_values': values,
        'flow_comparisons': comparisons,
        'any_available': bool(readable_pages or state_periods or durations or samples or values)}


def validate_result_policy(decision, availability):
    undetermined = decision.judgment.resultCode == 'UNDETERMINED'
    if undetermined != (not availability['any_available']):
        raise WorkflowError('workflow_result_policy',
            'UNDETERMINED 仅适用于三类数据均不可用；有可用数据时按明确异常证据判定，资料局部缺失只说明限制',
            details={'data_availability': availability, 'result_code': decision.judgment.resultCode})


def parse_decision(content, *, source_name='qwen-response.json'):
    if not isinstance(content, str) or not content.strip():
        raise WorkflowError('qwen_empty_response', 'Qwen 没有返回研判 JSON，原始响应已保存')
    value = workflow_json_repair.unfenced(content)
    def invalid_number(number):
        raise ValueError('研判 JSON 不接受非有限数值：' + number)
    try:
        raw = json.loads(value, parse_constant=invalid_number)
    except json.JSONDecodeError as exc:
        raise WorkflowError('qwen_invalid_json',
            f'Qwen JSON 语法错误：{exc.msg}，第 {exc.lineno} 行第 {exc.colno} 列（偏移 {exc.pos}）；原文见 {source_name}',
            details={'error_type': 'JSONDecodeError', 'line': exc.lineno, 'column': exc.colno,
                     'offset': exc.pos, 'message': exc.msg}) from exc
    except ValueError as exc:
        raise WorkflowError('qwen_invalid_json', str(exc), details={'error_type': type(exc).__name__}) from exc
    try:
        return WorkflowDecision.model_validate(raw)
    except ValidationError as exc:
        # 不把模型正文或整份输入塞入日志，只记录字段位置和约束错误。
        errors = exc.errors(include_input=False, include_context=False, include_url=False)
        raise WorkflowError('qwen_schema_validation',
            f'Qwen JSON 可解析，但字段不符合研判契约；字段错误见 diagnostics.error_details，原文见 {source_name}',
            details={'error_type': 'ValidationError', 'errors': errors[:8]}) from exc


def build_payload(decision, workflow, scope):
    binding = load_extensions().business('main_fan').result_binding
    task_id = workflow['context']['job_id'] or workflow['id']
    return PlatformResultPayload(agentCode=binding.agent_code, agentName=binding.agent_name,
        taskId=task_id, checkedAt=datetime.now(LOCAL_ZONE).strftime('%Y-%m-%d %H:%M:%S'),
        mineCode=scope.mine['mine_id'], mineName=scope.mine['mine_name'],
        **decision.model_dump(mode='json')).model_dump(mode='json')


async def execute(*, workflow_id, user_name, session_id, progress=None, qwen_client=None, ocr_client=None):
    """一次业务研判；仅失败扫描页补识别和不合格 JSON 格式修复允许额外调用。"""
    scope = scope_for(user_name, session_id)
    workflow = exploration._load(workflow_id, user_name, session_id, allow_completed=True)
    if workflow['status'] == 'completed':
        if (workflow.get('result') or {}).get('assessment', {}).get('mode') != VERSION:
            raise ValueError('旧执行版本结果不能冒充本版工作流结果，请重新研判')
        result = workflow['result']
        diagnostics = {**result['assessment'].get('diagnostics', {}), 'reused_result': True, 'qwen_calls_this_attempt': 0}
        return ExecutionResult(workflow['context'].get('assessment_workflow', {}).get('run_id', ''),
            session_id, 'success', result['result'], '', {}, result.get('sources', []), [], result, diagnostics)
    context = workflow['context']
    admission = None
    run_id = uuid4().hex
    started = time.monotonic()
    prefix = '/workspace/exploration/' + workflow_id + '/workflow'
    meters = {}
    diagnostics = {'execution_mode': VERSION, 'qwen_calls': 0, 'judgment_calls': 0,
                   'qwen_ocr_calls': 0, 'json_repair_calls': 0, 'tool_calls': 0}
    artifacts = {}
    stage_name = '初始化工作流'

    def write(name, value):
        virtual = prefix + '/' + name
        path = workspace_file(user_name, session_id, virtual)
        exploration._write(path, value)
        context.setdefault('query_exports', {})[virtual] = digest(path)
        artifacts[name] = virtual
        return virtual

    def stage(name, status, result=None):
        nonlocal stage_name
        stage_name = name
        rows = workflow['stages']
        key = 'workflow:' + name
        found = next((row for row in rows if row['name'] == key), None)
        value = {'name': key, 'status': status, 'result': result}
        if found is None:
            rows.append(value)
        else:
            found.clear(); found.update(value)
        context['assessment_workflow'] = {'version': VERSION, 'run_id': run_id,
            'stage': name, 'status': status, 'diagnostics': diagnostics, 'files': artifacts}
        main_fan_store.update_workflow(workflow_id, context=context, stages=rows)
        if progress:
            progress({'stage': name, 'status': status, 'workflow_id': workflow_id})
        logger.info('研判工作流阶段', extra={'kv': {'event': 'workflow.stage', 'stage': name,
            'status': status, 'workflow_id': workflow_id, 'run_id': run_id}})

    def record_usage(model, raw):
        meter = meters.setdefault(model, UsageMeter(user_name, session_id, 'business', model,
            trace_id=get_trace_id(), usage_id=run_id if model == config.MODEL else None))
        if raw:
            meter.add(raw.get('prompt_tokens', 0), raw.get('completion_tokens', 0))

    def record_qwen_call(kind):
        diagnostics['qwen_calls'] += 1
        diagnostics[kind] += 1

    def usage():
        rows = {name: {'input_tokens': meter.input_tokens, 'output_tokens': meter.output_tokens,
                       'total_tokens': meter.total_tokens, 'usage_source': meter.usage_source}
                for name, meter in meters.items()}
        return {'input_tokens': sum(row['input_tokens'] for row in rows.values()),
            'output_tokens': sum(row['output_tokens'] for row in rows.values()),
            'total_tokens': sum(row['total_tokens'] for row in rows.values()), 'models': rows}

    with log_context(run_id=run_id, workflow_id=workflow_id, execution_mode=VERSION):
        try:
            admission = await execution.admit(user_name, session_id)
            admission.business_lease = business.BusinessLease('main_fan')
            admission.business_lease.bind()
            await asyncio.to_thread(business_store.bind_session, session_id, 'main_fan')
            exploration.validate_inputs(workflow, user_name, session_id)
            if not _is_qwen35_model(config.MODEL):
                raise WorkflowError('workflow_model_config', '本工作流请将 DEEPAGENT_MODEL 配置为实际 Qwen3.5 服务模型名')
            endpoint = service_url(config.OPENAI_BASE_URL)
            maximum = config.get_int('MAIN_FAN_WORKFLOW_MAX_TOKENS', 16384)
            timeout = config.get_int('MAIN_FAN_WORKFLOW_TIMEOUT_SECONDS', 600)
            repair_limit = config.get_int('MAIN_FAN_JSON_REPAIR_MAX_ATTEMPTS', workflow_json_repair.MAX_ATTEMPTS)
            if maximum <= 0 or timeout <= 0:
                raise ValueError('工作流输出长度及超时必须大于 0')
            if not 0 <= repair_limit <= workflow_json_repair.MAX_ATTEMPTS:
                raise ValueError('JSON 格式修复次数须为 0～3，不能超过 3 次')
            ocr_settings = OCRSettings.from_config()
            stage('PDF逐页提取', 'running')
            extractor = PDFExtractor(user_name, session_id, prefix, scope.cache_root,
                settings=ocr_settings, client=ocr_client, qwen_client=qwen_client, progress=progress,
                cache_policy=context['task_input'].get('document_cache_policy', 'auto'),
                on_usage=lambda raw: record_usage(ocr_settings.model, raw),
                on_qwen_usage=lambda raw: record_usage(config.MODEL, raw),
                on_qwen_call=lambda: record_qwen_call('qwen_ocr_calls'))
            materials = context['prepared_data']['reports']['reports']
            reports = await extractor.extract(materials)
            context.setdefault('query_exports', {}).update(extractor.files)
            write('pdf-extraction.json', reports)
            diagnostics.update(pdf_count=len(reports), pdf_pages=sum(row['page_count'] for row in reports),
                failed_pages=sum(page['status'] != 'completed' for row in reports for page in row['pages']))
            stage('PDF逐页提取', 'completed', {key: diagnostics[key] for key in ('pdf_count', 'pdf_pages', 'failed_pages')})

            stage('过滤明确无关PDF页', 'running')
            selected_reports, selection = select_pages(reports)
            write('pdf-page-selection.json', selection)
            diagnostics.update(retained_pdf_pages=sum(len(row['pages']) for row in selected_reports),
                excluded_pdf_pages=sum(len(row['excluded_pages']) for row in selection))
            stage('过滤明确无关PDF页', 'completed', {key: diagnostics[key]
                for key in ('retained_pdf_pages', 'excluded_pdf_pages')})

            stage('合并本轮三类输入', 'running')
            monitoring = full_monitoring(context, user_name, session_id)
            availability = data_availability(reports, monitoring)
            stage('监测阈值固定判定', 'running')
            monitoring_assessment = workflow_monitoring.assess(monitoring, context['policy'])
            write('monitoring-assessment.json', monitoring_assessment)
            diagnostics['monitoring_result_code'] = monitoring_assessment['minimum_result_code']
            stage('监测阈值固定判定', 'completed')
            inputs = build_input(workflow, monitoring, selected_reports, scope, availability=availability,
                                 monitoring_assessment=monitoring_assessment)
            diagnostics['data_availability'] = availability
            diagnostics['judgment_policy'] = JUDGMENT_POLICY['version']
            write('qwen-input.json', inputs)
            system_prompt = Path(__file__).with_name('workflow_prompt.md').read_text(encoding='utf-8')
            messages = [{'role': 'system', 'content': system_prompt},
                        {'role': 'user', 'content': json.dumps(inputs, ensure_ascii=False, allow_nan=False)}]
            payload = {'model': config.MODEL, 'messages': messages, 'stream': False,
                'max_tokens': maximum, 'response_format': response_format(availability, monitoring_assessment),
                **_qwen_thinking_payload(normalize_thinking(None))}
            stage('合并本轮三类输入', 'completed')

            headers = {'Authorization': 'Bearer ' + config.OPENAI_API_KEY} if config.OPENAI_API_KEY else {}
            async def invoke(client, request, label, kind):
                # 无 HTTP 自动重试；补识别、格式修复均有明确计数和独立原始响应。
                write(label + '-request.json', request)
                record_qwen_call(kind)
                response = await client.post(endpoint + '/chat/completions', json=request, headers=headers, timeout=timeout)
                name = label + '-response.json'
                virtual = prefix + '/' + name
                local = workspace_file(user_name, session_id, virtual)
                local.parent.mkdir(parents=True, exist_ok=True)
                local.write_text(response.text, encoding='utf-8')
                context['query_exports'][virtual] = digest(local)
                artifacts[name] = virtual
                diagnostics['qwen_http_status'] = response.status_code
                if response.is_error:
                    raise WorkflowError('qwen_http_error', f'Qwen HTTP {response.status_code}，原始错误见 {name}')
                raw = response.json()
                if not isinstance(raw, dict):
                    raise WorkflowError('qwen_unexpected_response', f'Qwen 响应结构无效，详见 {name}')
                choice = raw['choices'][0]
                if not isinstance(choice, dict):
                    raise WorkflowError('qwen_unexpected_response', f'Qwen 候选消息结构无效，详见 {name}')
                message = choice['message']
                if not isinstance(message, dict):
                    raise WorkflowError('qwen_unexpected_response', f'Qwen 消息结构无效，详见 {name}')
                content = message.get('content')
                if raw.get('usage'):
                    record_usage(config.MODEL, raw['usage'])
                else:
                    record_usage(config.MODEL, {})
                    meters[config.MODEL].add(estimate_tokens(json.dumps(request['messages'], ensure_ascii=False)),
                        estimate_tokens((content if isinstance(content, str) else '')
                            + str(message.get('reasoning') or message.get('reasoning_content') or '')), 'estimate')
                if kind == 'judgment_calls':
                    diagnostics.update(response_id=raw.get('id'), finish_reason=choice.get('finish_reason'))
                if choice.get('finish_reason') == 'length':
                    raise WorkflowError('qwen_response_truncated', f'Qwen 输出达到长度上限，不能补造丢失内容；详见 {name}')
                if message.get('tool_calls') or choice.get('finish_reason') not in {'stop', None}:
                    raise WorkflowError('qwen_unexpected_response', f'Qwen 未返回正常结束的正文，详见 {name}')
                return raw, content

            async def judge(client):
                stage('Qwen一次研判', 'running')
                original_raw, original = await invoke(client, payload, 'qwen', 'judgment_calls')
                stage('Qwen一次研判', 'completed', {'response_id': original_raw.get('id'), 'usage': original_raw.get('usage')})
                stage('JSON校验与格式修复', 'running')
                attempts = []
                write('json-repair-history.json', attempts)
                try:
                    decision = parse_decision(original)
                except WorkflowError as exc:
                    if not repair_limit or not workflow_json_repair.can_repair(original, exc.code):
                        raise
                    error = {'error_code': exc.code, 'message': str(exc), 'details': exc.details}
                else:
                    write('qwen-final-response.json', original_raw)
                    return decision
                current = original
                for attempt in range(1, repair_limit + 1):
                    label = f'qwen-repair-{attempt}'
                    entry = {'attempt': attempt, 'request': prefix + '/' + label + '-request.json',
                        'response': prefix + '/' + label + '-response.json', 'status': 'running'}
                    attempts.append(entry)
                    write('json-repair-history.json', attempts)
                    logger.warning('研判 JSON 未通过校验，进行格式修复', extra={'kv': {
                        'event': 'workflow.json.repair.started', 'attempt': attempt, 'max_attempts': repair_limit,
                        'error_code': error['error_code'], 'error_details': error['details']}})
                    if progress:
                        progress({'stage': 'json_format_repair', 'attempt': attempt, 'max_attempts': repair_limit})
                    repair_request = {'model': config.MODEL, 'stream': False, 'max_tokens': maximum,
                        # 格式修复保留原编码；随后由固定监测规则统一合并，修复器不能改业务内容。
                        'messages': workflow_json_repair.messages(current, decision_schema(availability), error),
                        'response_format': response_format(availability), **_qwen_thinking_payload(normalize_thinking(None))}
                    repaired = None
                    try:
                        repaired_raw, repaired = await invoke(client, repair_request, label, 'json_repair_calls')
                        candidate = parse_decision(repaired, source_name=label + '-response.json')
                        changed = workflow_json_repair.preservation_errors(original, candidate)
                        if changed:
                            current = original
                            raise WorkflowError('qwen_repair_content_changed', '格式修复改动了业务内容，不能接受',
                                                details={'changed_content': changed})
                    except (WorkflowError, httpx.HTTPError, ValueError, KeyError, IndexError, TypeError) as exc:
                        error = {'error_code': exc.code if isinstance(exc, WorkflowError) else type(exc).__name__,
                                 'message': str(exc)[:500], 'details': getattr(exc, 'details', {})}
                        entry.update(status='failed', error=error)
                        # 继续修复上一轮字符串；最终始终与最初研判核对内容。
                        if isinstance(exc, WorkflowError) and exc.code in {'qwen_invalid_json', 'qwen_schema_validation'}:
                            if isinstance(repaired, str):
                                current = repaired
                        write('json-repair-history.json', attempts)
                        continue
                    entry.update(status='completed', response_id=repaired_raw.get('id'))
                    write('json-repair-history.json', attempts)
                    write('qwen-final-response.json', repaired_raw)
                    diagnostics['json_repaired'] = True
                    return candidate
                raise WorkflowError('qwen_json_repair_exhausted',
                    f'JSON 格式修复 {repair_limit} 次仍未通过，原始响应及每次修复记录已保存',
                    details={'repair_attempts': repair_limit, 'last_error': error})

            if qwen_client is None:
                async with qwen_http_client() as client:
                    decision = await judge(client)
            else:
                decision = await judge(qwen_client)
            stage('JSON校验与格式修复', 'completed', {'repair_attempts': diagnostics['json_repair_calls']})

            stage('校验并保存平台JSON', 'running')
            resolved, resolution = workflow_monitoring.resolve(decision.model_dump(mode='json'), monitoring_assessment)
            decision = WorkflowDecision.model_validate(resolved)
            write('decision-resolution.json', resolution)
            diagnostics['monitoring_enforced'] = resolution['monitoring_enforced']
            diagnostics['model_result_code'] = resolution['model_result_code']
            if resolution['monitoring_enforced']:
                logger.info('固定监测规则生成正式告警结论', extra={'kv': {
                    'event': 'workflow.monitoring.enforced', 'model_result_code': resolution['model_result_code'],
                    'minimum_result_code': resolution['minimum_result_code'], 'final_result_code': resolution['final_result_code']}})
            validate_result_policy(decision, availability)
            exploration.validate_inputs(workflow, user_name, session_id)
            platform = build_payload(decision, workflow, scope)
            write('platform-payload.json', platform)
            write('usage.json', usage())
            body = '# ' + scope.mine['mine_name'] + ' 主备风机研判报告\n\n' + decision.judgment.result + '\n\n' + decision.judgment.reason
            report_virtual = prefix + '/report.md'
            report_path = workspace_file(user_name, session_id, report_virtual)
            report_path.write_text(body, encoding='utf-8')
            context['query_exports'][report_virtual] = digest(report_path)
            artifacts['report.md'] = report_virtual
            sources = [{'title': row.get('filename') or Path(row['path']).name, 'url': row['path'], 'type': 'file'} for row in materials]
            sources.append({'title': '本轮完整工作流输入', 'url': artifacts['qwen-input.json'], 'type': 'file'})
            value = {'status': 'completed', 'mine_code': scope.mine['mine_id'], 'mine_name': scope.mine['mine_name'],
                'check_object': decision.objectDetail, 'result_title': scope.mine['mine_name'] + ' 主备风机研判报告',
                'result': body, 'result_format': 'report', 'workflow_id': workflow_id, 'sources': sources,
                'check_process': '固定监测预处理 → Python/GLM OCR及Qwen补识别 → 无关页过滤 → 固定监测阈值判定 → Qwen一次研判 → JSON校验及最多3次格式修复 → 固定告警结论合并 → 本地归档',
                'platform_payload': platform, 'analysis': None, 'assessment': {'mode': VERSION,
                    'decision': {'result_code': decision.judgment.resultCode, 'summary': decision.judgment.result},
                    'monitoring_assessment': monitoring_assessment,
                    'history_coverage': inputs['history_coverage'],
                    'report_archives': [{key: row[key] for key in ('path', 'sha256', 'is_mock', 'binding',
                        'page_count', 'complete', 'reused')} for row in reports],
                    'workflow_files': artifacts, 'diagnostics': diagnostics},
                'artifacts': [{'filename': name, 'path': artifacts[name]} for name in (
                    'report.md', 'platform-payload.json', 'qwen-input.json', 'pdf-extraction.json',
                    'pdf-page-selection.json', 'qwen-response.json', 'qwen-final-response.json',
                    'json-repair-history.json', 'monitoring-assessment.json', 'decision-resolution.json') if name in artifacts]}
            with db_conn.transaction():
                saved = main_fan_store.save_analysis(user_name, session_id, value)
                result = MainFanResult.model_validate(saved).model_dump(mode='json')
                stage('校验并保存平台JSON', 'completed', {'analysis_id': saved['analysis_id']})
                main_fan_store.update_workflow(workflow_id, context=context, stages=workflow['stages'], result=result, status='completed')
            try:
                await asyncio.to_thread(chat_store.save_message, session_id, 'user', '执行主通风机固定工作流（本轮数据独立研判）')
                await asyncio.to_thread(chat_store.save_message, session_id, 'assistant', body, citations=sources)
            except Exception:
                logger.warning('工作流结果已归档，但会话展示消息保存失败', exc_info=True)
            logger.info('工作流研判已完成并保存', extra={'kv': {'event': 'workflow.completed',
                'result_code': decision.judgment.resultCode, 'qwen_calls': diagnostics['qwen_calls'],
                'duration_ms': round((time.monotonic() - started) * 1000)}})
            return ExecutionResult(run_id, session_id, 'success', body, '', usage(), sources, [], result, diagnostics)
        except asyncio.CancelledError:
            main_fan_store.update_workflow(workflow_id, context=context, stages=workflow['stages'], status='interrupted')
            raise
        except Exception as exc:
            code = exc.code if isinstance(exc, WorkflowError) else type(exc).__name__
            diagnostics['error'] = str(exc)[:500]
            if isinstance(exc, WorkflowError) and exc.details:
                diagnostics['error_details'] = exc.details
            stage(stage_name, 'failed', {'error_code': code, 'message': str(exc)[:500],
                'error_details': diagnostics.get('error_details', {})})
            write('failure.json', {'error_code': code, 'message': str(exc)[:500], 'diagnostics': diagnostics})
            main_fan_store.update_workflow(workflow_id, context=context, stages=workflow['stages'], status='failed')
            logger.error('工作流研判失败', extra={'kv': {'event': 'workflow.failed', 'error_code': code,
                'stage': stage_name, 'qwen_calls': diagnostics['qwen_calls'], 'message': str(exc)[:500],
                'error_details': diagnostics.get('error_details', {})}})
            return ExecutionResult(run_id, session_id, 'failed', '', code, usage(), [], [], None, diagnostics)
        finally:
            try:
                for meter in meters.values():
                    await asyncio.shield(asyncio.to_thread(meter.settle))
            finally:
                if admission is not None:
                    admission.release()