"""月度本地批次：40天双系统取数、身份边界、PDF接口与跨会话档案复用。"""
from contextlib import asynccontextmanager, contextmanager
import asyncio
from datetime import timedelta
import json
from types import SimpleNamespace

import pytest

import config
from agents.workspace import session_root
from db import business_store, chat_store, main_fan_store
from extensions.main_fan import exploration, gateway_queries, history, report_archive
from extensions.main_fan.batch import BatchRunner
from extensions.main_fan.batch_context import BatchScope, current_scope
from extensions.main_fan.batch_models import BatchRequest, HistoryPoint, MinePlan, ShaftPlan
from extensions.main_fan.batch_tools import checked_plan
from extensions.main_fan.report_api import ReportClient, mock_report_client
from extensions.registry import load_extensions
from extensions.shaanxi.client import GatewayError
from runtime.python_sandbox import capability, digest, run as run_python, workspace_file
from test_main_fan_documents import native_pdf


@pytest.fixture
def batch_workspace(monkeypatch, tmp_path):
    for directory in ('skills', 'db'):
        (tmp_path / directory).symlink_to(config.ROOT / directory, target_is_directory=True)
    monkeypatch.setattr(config, 'ROOT', tmp_path)
    for key, value in {'MAIN_FAN_TARGETS_FILE': '', 'MAIN_FAN_RULES_FILE': '',
        'MAIN_FAN_SERIES_MAX_HOURS': '24', 'MAIN_FAN_HISTORY_CONCURRENCY': '2',
        'RESULT_PLATFORM_ENABLED': '1', 'MAIN_FAN_AUTO_PUBLISH': '1'}.items():
        monkeypatch.setenv(key, value)
    load_extensions()
    source = tmp_path / 'original.pdf'
    native_pdf(source)
    return SimpleNamespace(root=tmp_path, pdf=source,
        mine={'dept_id': '1219', 'mine_id': '00001219', 'mine_name': '测试甲矿'})


@contextmanager
def bound(workspace, *, client=None, shaft=None, days=40, mine=None):
    session = chat_store.create_session('batch-test', 'full')
    mine = mine or workspace.mine
    shaft = shaft or ShaftPlan(shaft_id='north', shaft_name='北风井', identity_status='confirmed', identity_reason='测试明确井号',
        points=[HistoryPoint(system='safety', node_id='state1', role='fan_state'),
                HistoryPoint(system='safety', node_id='state2', role='fan_state'),
                HistoryPoint(system='equipment', node_id='flow1', role='flow'),
                HistoryPoint(system='equipment', node_id='flow2', role='flow')])
    scope = BatchScope('batch-1', mine, 'batch-test', session['id'], workspace.root / 'out',
        workspace.root / 'archives', '2026-10-20T15:00:00+08:00', days, client, shaft)
    token = current_scope.set(scope)
    prepared = exploration.prepare({'mine_code': mine['mine_id'], 'mine_name': mine['mine_name'],
        'dept_id': mine['dept_id'], 'use_gateway': True, 'measurements': {'batch': {
            'batch_id': scope.batch_id, 'cutoff': scope.cutoff, 'lookback_days': scope.lookback_days,
            'shaft_plan': shaft.model_dump(mode='json')}}}, scope.user, scope.session, job_id='local-test-no-push')
    scope.active_workflow_id = prepared['workflow_id']
    try:
        yield scope, prepared['workflow_id']
    finally:
        current_scope.reset(token)


class HistoryGateway:
    def __init__(self):
        self.calls = []
        self.fail = set()

    async def proxy(self, agent, path, *, params):
        self.calls.append((path, params))
        node = params.get('nodeIds') or params['nodeId']
        if (node, params['startTime']) in self.fail:
            raise GatewayError('test', 503, retryable=False)
        if 'nodeIds' in params:
            return {'code': 200, 'data': [[{'NodeID': node, 'DataTime': params['startTime'],
                'RealData': 1, 'KgDesc': '开'}], [{'NodeID': node}]]}
        return {'code': 200, 'data': {'nodeId': node, 'unit': 'm3/min',
            'samples': [{'time': params['startTime'], 'value': 3800}]}}


def test_daily_windows_are_contiguous_and_rolling():
    windows = history.daily_windows('2026-10-20 15:00:00', 40)
    assert len(windows) == 40
    assert windows[0][1].isoformat() == '2026-10-20T15:00:00+08:00'
    assert windows[-1][0].isoformat() == '2026-09-10T15:00:00+08:00'
    assert all(end - start == timedelta(days=1) for start, end in windows)
    assert all(windows[index][0] == windows[index + 1][1] for index in range(39))


@pytest.mark.asyncio
async def test_four_points_160_daily_queries_one_operation_and_resumption(batch_workspace, monkeypatch):
    gateway = HistoryGateway()
    monkeypatch.setattr(history, 'get_service', lambda _: gateway)
    with bound(batch_workspace) as (scope, identifier):
        receipt = await history.query(identifier, scope.user, scope.session)
        assert receipt['coverage_complete'] and receipt['completed_segments'] == 160
        assert receipt['source_policy']['fan_state'] == 'safety'
        assert receipt['source_policy']['flow'] == 'equipment'
        assert len(gateway.calls) == 160
        assert sum('nodeIds' in params for _, params in gateway.calls) == 80
        assert sum('nodeId' in params for _, params in gateway.calls) == 80
        assert all(params['deptId'] == '1219' for _, params in gateway.calls)
        workflow = main_fan_store.workflow(identifier, scope.user, scope.session)
        assert len(workflow['stages']) == 2, '按日传输不消耗160次Agent操作'
        assert len(workflow['context']['history_manifest']['segments']) == 160
        archive = workspace_file(scope.user, scope.session, receipt['evidence_archive_path'])
        assert archive.is_file() and len(list(archive.parent.glob('history/**/*.json'))) == 0
        compact = json.loads(workspace_file(scope.user, scope.session, receipt['agent_data_path']).read_text())
        assert len(compact['states']) == 2 and len(compact['flows']) == 2
        assert len(compact['states'][0]['periods']) == 1
        details = json.loads(workspace_file(scope.user, scope.session, compact['details_path']).read_text())
        assert details['source_policy'] == receipt['source_policy']
        await history.query(identifier, scope.user, scope.session)
        assert len(gateway.calls) == 160, '恢复成功段不能重复发请求'
        exploration.validate_inputs(main_fan_store.workflow(identifier, scope.user, scope.session), scope.user, scope.session)


@pytest.mark.asyncio
async def test_history_failure_keeps_partial_and_only_failed_interval_retries(batch_workspace, monkeypatch):
    gateway = HistoryGateway()
    gateway.fail.add(('flow1', '2026-10-19 15:00:00'))
    monkeypatch.setattr(history, 'get_service', lambda _: gateway)
    with bound(batch_workspace, days=2) as (scope, identifier):
        first = await history.query(identifier, scope.user, scope.session)
        assert first['failed_segments'] == 1 and not first['coverage_complete']
        assert len(gateway.calls) == 8
        gateway.fail.clear()
        second = await history.query(identifier, scope.user, scope.session)
        assert second['completed_segments'] == 8 and second['coverage_complete']
        assert len(gateway.calls) == 9


@pytest.mark.asyncio
async def test_single_query_rejects_another_mine_before_gateway_access(batch_workspace, monkeypatch):
    gateway = HistoryGateway()
    monkeypatch.setattr(gateway_queries, 'get_service', lambda _: gateway)
    with bound(batch_workspace) as (scope, identifier):
        with pytest.raises(ValueError, match='不属于'):
            await gateway_queries.fetch(identifier, scope.user, scope.session, '/ignored', {'deptId': '1338'})
        assert not gateway.calls


@pytest.mark.asyncio
async def test_mock_report_interface_returns_original_bytes_and_fixture_provenance(batch_workspace):
    async with mock_report_client(batch_workspace.pdf) as http:
        client = ReportClient(http)
        files, _, response = await client.list('00001219', 'north')
        assert response['code'] == 200 and files[0]['is_mock']
        assert files[0]['binding'] == 'test_fixture_not_mine_report'
        target = batch_workspace.root / 'download.pdf'
        assert await client.download(files[0], target) == digest(batch_workspace.pdf)
        assert target.read_bytes() == batch_workspace.pdf.read_bytes()
        assert (await http.get('/api/main-fan/report-files/not-found.pdf')).status_code == 404


@pytest.mark.asyncio
async def test_report_archive_reuses_across_sessions_and_corruption_reparses(batch_workspace):
    if not capability()['available']:
        pytest.skip('需要真实Python沙箱')
    async with mock_report_client(batch_workspace.pdf) as http:
        client = ReportClient(http)
        with bound(batch_workspace, client=client) as (scope, identifier):
            first = await report_archive.fetch(identifier, scope.user, scope.session)
            assert first['reports'][0]['mode'] == 'analyze'
            facts = {'source_sha256': digest(batch_workspace.pdf), 'page_count': 1, 'read_pages': [1],
                'page_sources': {'1': '/workspace/page_1.txt'},
                'reports': [{'mine_name': 'Synthetic Native Test Mine', 'report_id': 'NATIVE-TEST-001', 'page': 1}],
                'analysis': {'is_test': True}, 'limitations': ['不同矿的模拟原件']}
            script = workspace_file(scope.user, scope.session, '/workspace/extract.py')
            script.write_text('from pathlib import Path\nimport json\nPath("page_1.txt").write_text("Actual test page text")\nPath("facts.json").write_text(' + repr(json.dumps(facts)) + ')\n')
            execution = await run_python(scope.user, scope.session, '/workspace/extract.py',
                output_paths=['/workspace/facts.json', '/workspace/page_1.txt'])
            assert execution['exit_code'] == 0, execution['stderr']
            exploration.record_python(identifier, scope.user, scope.session, execution)
            with pytest.raises(ValueError, match='没有可核验'):
                report_archive.save(identifier, scope.user, scope.session, '/workspace/facts.json', [execution['run_id']])
            assert not list(scope.cache_root.glob('*.json'))
            from extensions.main_fan.middleware import record_page_read
            from langchain_core.messages import ToolMessage
            record_page_read(SimpleNamespace(tool_call={'name': 'read_file', 'args': {'file_path': '/workspace/page_1.txt'}}),
                ToolMessage(content='Actual test page text', tool_call_id='page-read'))
            receipt = report_archive.save(identifier, scope.user, scope.session, '/workspace/facts.json', [execution['run_id']])
            assert receipt['page_count'] == 1
        with bound(batch_workspace, client=client) as (scope, identifier):
            reused = await report_archive.fetch(identifier, scope.user, scope.session)
            assert reused['reused_count'] == 1 and reused['reports'][0]['mode'] == 'reuse'
            assert 'archive_python_path' in reused['reports'][0]
        archive = next((batch_workspace.root / 'archives').glob('*.json'))
        archive.write_text('{broken')
        with bound(batch_workspace, client=client) as (scope, identifier):
            assert (await report_archive.fetch(identifier, scope.user, scope.session))['reports'][0]['mode'] == 'analyze'


@pytest.mark.asyncio
async def test_report_content_change_invalidates_archive(batch_workspace):
    async with mock_report_client(batch_workspace.pdf) as http:
        with bound(batch_workspace, client=ReportClient(http)) as (scope, identifier):
            first = await report_archive.fetch(identifier, scope.user, scope.session)
            sha = first['reports'][0]['sha256']
            assert not report_archive.load_cached(scope, sha)
    original = batch_workspace.pdf.read_bytes()
    batch_workspace.pdf.write_bytes(original + b'\n% updated fixture\n')
    async with mock_report_client(batch_workspace.pdf) as http:
        with bound(batch_workspace, client=ReportClient(http)) as (scope, identifier):
            second = await report_archive.fetch(identifier, scope.user, scope.session)
            assert second['reports'][0]['sha256'] != sha and second['reused_count'] == 0


@pytest.mark.asyncio
async def test_claiming_26_pages_after_reading_only_10_cannot_create_cache(batch_workspace):
    if not capability()['available']:
        pytest.skip('需要真实沙箱')
    import pypdfium2
    from extensions.main_fan.middleware import record_page_read
    from langchain_core.messages import ToolMessage
    original = batch_workspace.pdf.read_bytes()
    with pypdfium2.PdfDocument(original) as source, pypdfium2.PdfDocument.new() as combined:
        for _ in range(26):
            combined.import_pages(source)
        combined.save(batch_workspace.pdf)
    async with mock_report_client(batch_workspace.pdf) as http:
        with bound(batch_workspace, client=ReportClient(http)) as (scope, identifier):
            await report_archive.fetch(identifier, scope.user, scope.session)
            facts = {'source_sha256': digest(batch_workspace.pdf), 'page_count': 26,
                'read_pages': list(range(1, 27)), 'page_sources': {str(page): f'/workspace/page_{page}.txt' for page in range(1, 27)},
                'reports': [{'report_id': 'synthetic'}]}
            script = workspace_file(scope.user, scope.session, '/workspace/incomplete.py')
            script.write_text('from pathlib import Path\nfor page in range(1,27): Path(f"page_{page}.txt").write_text(f"actual page {page}")\nPath("incomplete.json").write_text(' + repr(json.dumps(facts)) + ')\n')
            receipt = await run_python(scope.user, scope.session, '/workspace/incomplete.py', output_paths=['/workspace/incomplete.json'])
            assert receipt['exit_code'] == 0
            exploration.record_python(identifier, scope.user, scope.session, receipt)
            for page in range(1, 11):
                record_page_read(SimpleNamespace(tool_call={'name': 'read_file', 'args': {'file_path': f'/workspace/page_{page}.txt'}}),
                    ToolMessage(content=f'actual page {page}', tool_call_id=str(page)))
            with pytest.raises(ValueError, match='第11页'):
                report_archive.save(identifier, scope.user, scope.session, '/workspace/incomplete.json', [receipt['run_id']])
            assert not list(scope.cache_root.glob('*.json'))


def test_plan_requires_real_points_and_complete_pagination(batch_workspace):
    with bound(batch_workspace) as (scope, identifier):
        workflow = main_fan_store.workflow(identifier, scope.user, scope.session)
        plan = MinePlan(shafts=[scope.shaft.model_copy(update={'shaft_id': 'unresolved-main',
            'shaft_name': '主备系统（待确认）', 'identity_status': 'candidate'})])
        with pytest.raises(ValueError, match='两个实时列表'):
            checked_plan(workflow, scope.user, scope.session, plan)
        context = workflow['context']
        for system, collection in [('safety', 'sensor_queries'), ('equipment', 'equipment_queries')]:
            virtual = '/workspace/' + system + '.json'
            local = workspace_file(scope.user, scope.session, virtual)
            exploration._write(local, {'payload': {'rows': []}})
            context[collection] = [{'data_path': virtual, 'page_num': 1, 'next_page': None}]
        with pytest.raises(ValueError, match='未从本矿'):
            checked_plan(workflow, scope.user, scope.session, plan)
        context['sensor_queries'][0]['next_page'] = 2
        with pytest.raises(ValueError, match='未获取分页'):
            checked_plan(workflow, scope.user, scope.session, plan)


@pytest.mark.asyncio
async def test_runner_selected_mines_local_files_no_publish_even_global_auto_enabled(batch_workspace, monkeypatch):
    calls = []
    mine = batch_workspace.mine
    snapshot = {'snapshot_id': 'a' * 32, 'mines': [mine,
        {'dept_id': '1338', 'mine_id': '00001338', 'mine_name': '不应执行的矿'}]}
    plan = MinePlan(shafts=[ShaftPlan(shaft_id=value, shaft_name=value, identity_reason='测试数据明确井号')
                           for value in ('north', 'south')])
    async def discover(*args):
        return plan
    async def prepare(*args):
        return {'agent_data_path': '/workspace/prepared.json'}
    monkeypatch.setattr('extensions.main_fan.batch.discovery.discover', discover)
    monkeypatch.setattr('extensions.main_fan.batch.preparation.assessment', prepare)
    # 本例只验证队列与隔离；模拟结果没有数据库中的真实 workflow。
    monkeypatch.setattr('extensions.main_fan.batch.data_export.export', lambda *args: {'status': 'test_stub'})
    async def executor(**kwargs):
        data = main_fan_store.workflow(kwargs['workflow_id'], kwargs['user_name'], kwargs['session_id'])['context']['task_input']
        phase = data['measurements']['batch']['phase']
        calls.append((data['dept_id'], kwargs['session_id'], phase))
        workspace = session_root(kwargs['user_name'], kwargs['session_id']) / 'workspace'
        workspace.mkdir(exist_ok=True)
        outside = batch_workspace.root / 'not-an-evidence.txt'
        outside.write_text('do not export')
        link = workspace / 'external-link.txt'
        if not link.is_symlink():
            link.symlink_to(outside)
        if phase == 'discovery':
            result = {'assessment': {'plan': plan.model_dump(mode='json')}}
        else:
            result = {'result': '# 本地测试报告', 'analysis_id': 'analysis-test', 'workflow_id': 'workflow-test',
                'platform_payload': {'taskId': 'test-task', 'judgment': {'resultCode': 'UNDETERMINED'}},
                'assessment': {'history_coverage': {'coverage_complete': True}, 'report_archives': []}}
        return SimpleNamespace(status='success', run_id='run-test', error='', output='', usage={}, structured_output=result)
    output = batch_workspace.root / 'output'
    runner = BatchRunner(output, batch_workspace.pdf, executor=executor)
    assert runner.publish is False, '新工作流在全局总开关已开时仍默认不入队'
    request = BatchRequest(dept_ids=['1219'], cutoff='2026-10-20T15:00:00+08:00')
    result = await runner.run(request, snapshot=snapshot)
    assert result['status'] == 'completed' and result['publish'] is False
    assert len(calls) == 2 and {row[0] for row in calls} == {'1219'}
    assert len({row[1] for row in calls}) == 1, '同矿多井复用矿级Agent会话'
    assert (output / '1219/north/报告.md').is_file() and (output / '1219/south/结果.json').is_file()
    assert not (output / '1338').exists()
    assert not (output / '1219/workspace/external-link.txt').exists()
    assert not business_store.list_jobs('main_fan', 'batch-test', 100)
    await runner.run(request)
    assert len(calls) == 2, '恢复已完成批次不能重新调用Agent'
    await runner.run(request.model_copy(update={'reassess': True}))
    assert len(calls) == 4, '重新分析只执行原矿的两井，不重复发现和遍历其他矿'
    assert (output / '1219/north/revisions/analysis-test/报告.md').is_file()


def test_plan_rejects_vibration_as_fan_state_and_fan_number_as_shaft(batch_workspace):
    with bound(batch_workspace) as (scope, identifier):
        workflow = main_fan_store.workflow(identifier, scope.user, scope.session)
        for collection in ('sensor_queries', 'equipment_queries'):
            virtual = '/workspace/' + collection + '.json'
            exploration._write(workspace_file(scope.user, scope.session, virtual), {'payload': {'rows': [
                {'nodeId': 'vibration', 'nodeTypeName': '水平振动', 'sensorUnit': 'mm/s'}]}})
            workflow['context'][collection] = [{'data_path': virtual, 'page_num': 1, 'next_page': None}]
        shaft = ShaftPlan(shaft_id='unresolved-main', shaft_name='主备系统（待确认）', identity_reason='没有井号',
            points=[HistoryPoint(system='equipment', node_id='vibration', role='fan_state')])
        with pytest.raises(ValueError, match='实际类型'):
            checked_plan(workflow, scope.user, scope.session, MinePlan(shafts=[shaft]))
        shaft.points = []
        shaft.shaft_id, shaft.shaft_name = 'shaft-1', '1号通风井'
        with pytest.raises(ValueError, match='风机号不能'):
            checked_plan(workflow, scope.user, scope.session, MinePlan(shafts=[shaft]))


@pytest.mark.parametrize('system,role,row', [
    ('equipment', 'fan_state', {'nodeId': 'wrong-source', 'nodeTypeName': '风机运行状态'}),
    ('safety', 'flow', {'nodeId': 'wrong-source', 'sensorName': '风量'}),
])
def test_plan_rejects_valid_measurement_from_wrong_authoritative_system(batch_workspace, system, role, row):
    with bound(batch_workspace) as (scope, identifier):
        workflow = main_fan_store.workflow(identifier, scope.user, scope.session)
        for name, collection in [('safety', 'sensor_queries'), ('equipment', 'equipment_queries')]:
            virtual = '/workspace/' + collection + '.json'
            exploration._write(workspace_file(scope.user, scope.session, virtual),
                {'payload': {'rows': [row] if name == system else []}})
            workflow['context'][collection] = [{'data_path': virtual, 'page_num': 1, 'next_page': None}]
        shaft = ShaftPlan(shaft_id='unresolved-main', shaft_name='主备系统（待确认）', identity_reason='没有井号',
            points=[HistoryPoint(system=system, node_id='wrong-source', role=role)])
        with pytest.raises(ValueError, match='开停以安全监控为准'):
            checked_plan(workflow, scope.user, scope.session, MinePlan(shafts=[shaft]))


@pytest.mark.asyncio
async def test_history_rejects_old_equipment_state_plan_before_gateway_access(batch_workspace, monkeypatch):
    gateway = HistoryGateway()
    monkeypatch.setattr(history, 'get_service', lambda _: gateway)
    shaft = ShaftPlan(shaft_id='unresolved-main', shaft_name='主备系统（待确认）', identity_reason='旧计划',
        points=[HistoryPoint(system='equipment', node_id='state1', role='fan_state')])
    with bound(batch_workspace, shaft=shaft) as (scope, identifier):
        with pytest.raises(ValueError, match='开停以安全监控为准'):
            await history.query(identifier, scope.user, scope.session)
        assert gateway.calls == []


@pytest.mark.asyncio
async def test_history_reuses_old_archive_but_excludes_equipment_states_from_analysis(batch_workspace, monkeypatch):
    gateway = HistoryGateway()
    monkeypatch.setattr(history, 'get_service', lambda _: gateway)
    with bound(batch_workspace, days=2) as (scope, identifier):
        scope.shaft.points.append(HistoryPoint(system='equipment', node_id='old-state', role='fan_state'))
        # 先按旧版本行为生成真实格式的合成归档，再应用当前来源选择，检验历史兼容。
        with monkeypatch.context() as legacy:
            legacy.setattr(history, 'select_history_points', lambda points: list(points))
            old = await history.query(identifier, scope.user, scope.session)
        assert old['completed_segments'] == 10 and len(gateway.calls) == 10
        archive = workspace_file(scope.user, scope.session, old['evidence_archive_path'])
        archived_bytes = archive.read_bytes()
        current = await history.query(identifier, scope.user, scope.session)
        assert current['coverage_complete'] and current['completed_segments'] == 8
        assert current['point_count'] == 4 and len(gateway.calls) == 10
        assert archive.read_bytes() == archived_bytes, '来源调整不应重拉或删除旧原件'
        workflow = main_fan_store.workflow(identifier, scope.user, scope.session)
        assert len(workflow['context']['history_index']) == 10
        manifest = workflow['context']['history_manifest']
        assert len(manifest['segments']) == 8
        assert manifest['ignored_plan_points'][0]['node_id'] == 'old-state'
        compact = json.loads(workspace_file(scope.user, scope.session, current['agent_data_path']).read_text())
        assert all(p['point'] != 'old-state' for p in compact['states'])
        exploration.validate_inputs(workflow, scope.user, scope.session)


@pytest.mark.asyncio
async def test_legacy_agent_graph_with_explicit_executor_and_local_delivery(batch_workspace, monkeypatch, pricing_ready):
    if not capability()['available']:
        pytest.skip('需要实际沙箱')
    from agents import agent_factory, agent_registry
    from runtime.environment import RuntimeEnvironment
    from sandbox_agent_helpers import tool_call
    from test_main_fan_timeseries import ToolModel
    monkeypatch.setenv('RESULT_PLATFORM_ENABLED', '0')
    points = [HistoryPoint(system='safety', node_id='state1', role='fan_state'),
              HistoryPoint(system='equipment', node_id='flow1', role='flow', unit='m3/min')]
    plan = MinePlan(shafts=[ShaftPlan(shaft_id='unresolved-main', shaft_name='主备系统（待确认）',
        identity_reason='实际列表未给井号', points=points)])
    gateway = HistoryGateway()
    original_proxy = gateway.proxy
    async def proxy(agent, path, *, params):
        if path.endswith('getNodeRealtimeDetails'):
            return {'code': 200, 'total': 1, 'rows': [{'nodeId': 'state1', 'sensorTypeId': '1010', 'nodePlace': '主通1一级'}]}
        if path.endswith('selectSbNodeList'):
            return {'code': 200, 'total': 1, 'rows': [{'nodeId': 'flow1', 'nodeTypeName': '风量', 'nodeTypeNo': '30101', 'sensorUnit': 'm3/min'}]}
        return await original_proxy(agent, path, params=params)
    gateway.proxy = proxy
    monkeypatch.setenv('SHANXI_GATEWAY_ENABLED', '1')
    monkeypatch.setenv('SHANXI_GATEWAY_URL', 'https://gateway.invalid')
    monkeypatch.setenv('SHANXI_GATEWAY_TOKEN', 'batch-test-not-a-real-token')
    @asynccontextmanager
    async def gateway_service(dependencies):
        yield gateway
    monkeypatch.setattr('extensions.shaanxi.extension.gateway_service', gateway_service)
    monkeypatch.setattr(gateway_queries, 'get_service', lambda _: gateway)
    monkeypatch.setattr(history, 'get_service', lambda _: gateway)
    def latest():
        from db import db_conn
        row = db_conn.query_one("SELECT id,user_name,session_id FROM main_fan_workflows WHERE status='running' ORDER BY created_at DESC LIMIT 1")
        return main_fan_store.workflow(row['id'], row['user_name'], row['session_id'])
    def messages():
        yield tool_call('prepare_main_fan_task', {'target': {}})
        workflow = latest(); identifier = workflow['id']
        yield tool_call('fetch_main_fan_reports', {'workflow_id': identifier})
        yield tool_call('query_main_fan_history', {'workflow_id': identifier})
        workflow = latest(); pdf = workflow['context']['report_archives'][0]
        prefix = 'exploration/' + identifier
        code = f'''from pathlib import Path
import json, hashlib, pypdfium2 as pdfium
source=Path({pdf['python_path']!r})
with pdfium.PdfDocument(source) as document:
    texts=[document[index].get_textpage().get_text_range() for index in range(len(document))]
facts={{'source_sha256':hashlib.sha256(source.read_bytes()).hexdigest(),'page_count':len(texts),
       'read_pages':list(range(1,len(texts)+1)),'page_sources':{{'1':'/workspace/page_1.txt'}},'reports':[{{'raw_text':texts[0],'page':1}}],
       'analysis':{{}},'limitations':['模拟报告，不移植矿井身份']}}
manifest=json.loads(Path({(prefix + '/history-manifest.json')!r}).read_text())
compact=json.loads(Path({(prefix + '/history-agent.json')!r}).read_text())
details=json.loads(Path(compact['details_path'].removeprefix('/workspace/')).read_text())
counts={{row['node_id']:row['quality']['raw_sample_count'] for row in details['points']}}
Path('facts.json').write_text(json.dumps(facts,ensure_ascii=False))
Path('page_1.txt').write_text(texts[0])
Path('report.md').write_text('# 本地判识\\n\\n完整查询并保留归属限制。')
assessment={{'result_code':'UNDETERMINED','summary':'已查询原件与时序，通风井及整机归属待确认。',
 'details':{{'segments':manifest['expected_segments'],'total_records':sum(counts.values()),'switch_analysis':{{'status':'insufficient_data','method':'单级状态不能代替整机','events':[],'limitations':['测试仅有单级状态']}},'flow_analysis':{{'status':'unrecognized_structure','raw_sample_count':None,'limitations':['保留未知格式，不当空数据']}}}},
 'evidence':[{{'source':'/workspace/{prefix}/history-manifest.json'}}]}}
Path('assessment.json').write_text(json.dumps(assessment,ensure_ascii=False))
'''
        yield tool_call('write_file', {'file_path': '/workspace/analyze.py', 'content': code})
        yield tool_call('run_python', {'workflow_id': identifier, 'script_path': '/workspace/analyze.py',
            'output_paths': ['/workspace/facts.json', '/workspace/report.md', '/workspace/assessment.json', '/workspace/page_1.txt']})
        receipt = latest()['context']['python_runs'][-1]
        assert receipt['exit_code'] == 0
        yield tool_call('read_file', {'file_path': '/workspace/page_1.txt'})
        yield tool_call('archive_main_fan_report', {'workflow_id': identifier, 'facts_path': '/workspace/facts.json',
            'python_run_ids': [receipt['run_id']]})
        # 即使宿主全局自动推送开关为真，本地批次的task_id也必须阻止入队。
        monkeypatch.setenv('RESULT_PLATFORM_ENABLED', '1')
        yield tool_call('submit_main_fan_report', {'workflow_id': identifier, 'report_path': '/workspace/report.md',
            'calculations_path': '/workspace/assessment.json', 'python_run_ids': [receipt['run_id']]})
    model = ToolModel(messages=messages())
    monkeypatch.setattr(agent_factory.model_factory, 'get_model', lambda: model)
    agent_registry.invalidate()
    async def legacy_executor(*, workflow_id, user_name, session_id, progress=None):
        # 显式保留原交互 Agent/沙箱回归；生产批次默认走新的固定 workflow。
        from runtime import execution
        task = main_fan_store.workflow(workflow_id, user_name, session_id)['context']
        return await execution.execute(agent_name='main_fan', user_name=user_name, session_id=session_id,
            message='按原有 Skill 读取准备好的本轮原件并提交报告。', kind='business',
            graph_config={'recursion_limit': 480, 'configurable': {
                'business_task_input': task['task_input'], 'business_task_id': task['job_id']}})
    output = batch_workspace.root / 'graph-output'
    snapshot = {'snapshot_id': 'b' * 32, 'mines': [batch_workspace.mine]}
    async with RuntimeEnvironment(prewarm=False, billing_retry=False, recover_interrupted=False):
        result = await BatchRunner(output, batch_workspace.pdf, executor=legacy_executor).run(
            {'dept_ids': ['1219'], 'cutoff': '2026-10-20 15:00:00', 'lookback_days': 2}, snapshot=snapshot)
        assert result['status'] == 'completed', result
        from db import db_conn
        assert db_conn.query_one('SELECT COUNT(*) AS n FROM business_jobs')['n'] == 0
    delivered = json.loads((output / '1219/unresolved-main/结果.json').read_text())
    assert delivered['assessment']['agent_result']['details']['segments'] == 4
    assert delivered['assessment']['history_coverage']['coverage_complete']
    assert delivered['assessment']['report_archives'][0]['archived']
    assert (output / 'index.html').is_file()


@pytest.mark.asyncio
async def test_reassessment_reuses_same_batch_history_but_other_batch_refetches(batch_workspace, monkeypatch):
    gateway = HistoryGateway()
    monkeypatch.setattr(history, 'get_service', lambda _: gateway)
    with bound(batch_workspace, days=2) as (scope, identifier):
        await history.query(identifier, scope.user, scope.session)
        assert len(gateway.calls) == 8
        old = main_fan_store.workflow(identifier, scope.user, scope.session)
        main_fan_store.update_workflow(identifier, status='completed')
        data = old['context']['task_input']
        second = exploration.prepare(data, scope.user, scope.session, job_id='local-review')
        scope.active_workflow_id = second['workflow_id']
        receipt = await history.query(second['workflow_id'], scope.user, scope.session)
        assert receipt['coverage_complete'] and len(gateway.calls) == 8
        scope.batch_id = 'different-batch'
        data['measurements']['batch']['batch_id'] = scope.batch_id
        third = exploration.prepare(data, scope.user, scope.session, job_id='other-month')
        scope.active_workflow_id = third['workflow_id']
        await history.query(third['workflow_id'], scope.user, scope.session)
        assert len(gateway.calls) == 16, '不同月度批次仍重新取得当期原件'


def test_equipment_sbnodecurve_zero_samples_are_data_not_empty():
    from extensions.main_fan.equipment import _curve_data
    body = {'code': 200, 'data': [{'nodeId': 'flow1', 'nodePlace': '1号风机风量', 'sbNodeCurve': [
        {'dataTime': '2026-10-19 16:23:20', 'realData': '0'},
        {'dataTime': '2026-10-19 17:13:21', 'realData': '0'}]}]}
    data, summary = _curve_data(body, 'flow1')
    assert summary['sample_count'] == 2 and summary['normalization'] == 'sbNodeCurve'
    assert data['payload'] is body and data['series'][0]['samples'][0]['value'] == '0'
    assert data['series'][0]['samples'][0]['time'] == '2026-10-19T16:23:20+08:00'
    unknown, metadata = _curve_data({'code': 200, 'data': [{'otherCurve': [1, 2]}]})
    assert unknown['series'] is None and metadata['sample_count'] is None
    with pytest.raises(ValueError, match='测点与本次请求'):
        _curve_data(body, 'another-point')


def test_independent_verification_excludes_conflicting_states_and_keeps_zero_flow(tmp_path):
    from scripts.verify_main_fan_batch import point_statistics
    path = tmp_path / 'workspace' / 'samples.json'
    path.parent.mkdir()
    segment = {'status': 'completed', 'node_id': 'flow', 'data_path': '/workspace/samples.json',
        'start_time': '2026-10-01T00:00:00+08:00', 'end_time': '2026-10-02T00:00:00+08:00'}
    path.write_text(json.dumps({'samples': [
        {'time': '2026-10-01T01:00:00+08:00', 'value_label': '开', 'status_label': '正常'},
        {'time': '2026-10-01T02:00:00+08:00', 'value_label': '开', 'status_label': '正常'},
        {'time': '2026-10-01T02:00:00+08:00', 'value_label': '停', 'status_label': '正常'},
        {'time': '2026-10-01T03:00:00+08:00', 'value_label': '停', 'status_label': '故障'},
        {'time': '2026-09-30T03:00:00+08:00', 'value_label': '停', 'status_label': '正常'},
    ]}))
    safety = point_statistics([segment], tmp_path, 'safety')
    assert safety['raw_sample_count'] == 5 and safety['unique_usable_sample_count'] == 1
    assert safety['conflicting_timestamp_count'] == 1 and safety['invalid_status_count'] == 1
    assert safety['out_of_window_count'] == 1 and safety['transitions'] == []
    path.write_text(json.dumps({'payload': {'data': [{'nodeId': 'flow', 'sbNodeCurve': [
        {'dataTime': '2026-10-01 01:00:00', 'realData': '0'},
        {'dataTime': '2026-10-01 01:00:00', 'realData': '0'},
        {'dataTime': '2026-10-01 02:00:00', 'realData': 'NaN'},
    ]}]}}))
    flow = point_statistics([segment], tmp_path, 'equipment')
    assert flow['raw_sample_count'] == 3 and flow['zero_count'] == 1
    assert flow['repeated_timestamp_count'] == 1 and flow['unrecognized_value_count'] == 1


@pytest.mark.asyncio
async def test_monthly_local_queue_idempotence_and_shutdown_before_start(batch_workspace, monkeypatch):
    from extensions.main_fan import batch_service
    monkeypatch.setenv('MAIN_FAN_MOCK_REPORT_PDF', str(batch_workspace.pdf))
    monkeypatch.setenv('MAIN_FAN_BATCH_OUTPUT_DIR', str(batch_workspace.root / 'service-output'))
    monkeypatch.setenv('MAIN_FAN_BATCH_DEPT_IDS', '1219')
    calls = []
    class Runner:
        def __init__(self, folder, pdf, **kwargs):
            self.folder = folder
        async def run(self, request):
            calls.append(request.dept_ids)
            exploration._write(self.folder / 'batch.json', {'status': 'completed'})
            return {'status': 'completed'}
    monkeypatch.setattr(batch_service, 'BatchRunner', Runner)
    controller = batch_service.LocalBatchController()
    monkeypatch.setattr('extensions.get_service', lambda _: controller)
    first = await batch_service.monthly_callback()
    second = await batch_service.monthly_callback()
    assert first['batch_id'] == second['batch_id'] and len(controller.tasks) == 1
    await asyncio.gather(*list(controller.tasks.values()))
    assert (await batch_service.monthly_callback())['status'] == 'completed'
    assert calls == [['1219']]
    queued = controller.submit({'dept_ids': ['1219']})
    await controller.close()
    assert controller.get(queued['batch_id'])['status'] == 'interrupted'
    assert not controller.tasks