vent_agent_2.0/tests/test_main_fan_sandbox.py

"""自主模式的真实沙箱执行、交付归档和权限边界；不调用外部模型或网关。"""
import asyncio
import json
import socket
from pathlib import Path
from types import SimpleNamespace

import pytest
from PIL import Image

import config
from agents.workspace import session_root
from db import main_fan_store
from extensions.main_fan import exploration, service
from extensions.main_fan.schemas import AgentAssessment
from extensions.registry import load_extensions
from runtime import python_sandbox


@pytest.fixture
def workbench(monkeypatch, tmp_path):
    monkeypatch.setattr(config, 'ROOT', tmp_path)
    monkeypatch.setenv('MAIN_FAN_TARGETS_FILE', '')
    monkeypatch.setenv('MAIN_FAN_RULES_FILE', '')
    user, session = 'sandbox-user', 'sandbox-session'
    workspace = session_root(user, session) / 'workspace'
    uploads = workspace / 'uploads'
    uploads.mkdir(parents=True)
    source = uploads / 'scan.pdf'
    with Image.new('RGB', (120, 160), 'white') as image:
        image.save(source, 'PDF')
    request = {'mine_code': 'sandbox-mine', 'mine_name': '沙箱测试矿',
               'fan_id': 'fan-1', 'fan_name': '1号风机', 'fan_pair': ['fan-1', 'fan-2'],
               'materials': [{'path': '/workspace/uploads/scan.pdf'}],
               'measurements': {'values': [10, 20, 30]}}
    prepared = exploration.prepare(request, user, session)
    return SimpleNamespace(user=user, session=session, workspace=workspace,
                           request=request, manifest=prepared, source=source)


def require_sandbox():
    if not python_sandbox.capability()['available']:
        pytest.skip(python_sandbox.capability()['reason'])


def delivery_script(workbench, *, code='WARNING', summary='脚本实际计算均值20.0', status='completed'):
    """测试模型写出的程序：读原 PDF、渲染、计算并生成所有交付，不调用业务服务。"""
    identifier = workbench.manifest['workflow_id']
    return f'''
from pathlib import Path
import json
import statistics
import pypdfium2 as pdfium
from PIL import Image
import numpy as np
import cv2
base = Path('exploration/{identifier}')
data = json.loads((base/'input.json').read_text())
mean = statistics.mean(data['measurements']['values'])
with pdfium.PdfDocument('uploads/scan.pdf') as pdf:
    pages = len(pdf)
    page = pdf[0]
    try:
        textpage = page.get_textpage()
        try:
            text = textpage.get_text_range()
        finally:
            textpage.close()
        bitmap = page.render(scale=1)
        image = bitmap.to_pil()
        image.save('page.png')
        image.close()
        bitmap.close()
    finally:
        page.close()
facts = {{'page_count': pages, 'text': text, 'mean': mean}}
Path('facts.json').write_text(json.dumps(facts))
report = '# Agent 生成的测试报告\\n\\n均值：' + str(mean) + '；PDF页数：' + str(pages)
Path('report.md').write_text(report)
result = {{'status': {status!r}, 'result_code': {code!r}, 'summary': {summary!r},
    'data_detail': '测试原件及本轮明确提供的数据；非现场结果',
    'evidence': [{{'source': '/workspace/exploration/{identifier}/input.json',
                  'pointer': '/measurements/values', 'excerpt': '10、20、30'}}],
    'details': {{'mean': mean, 'pages': pages, 'custom_method': 'statistics.mean'}}}}
Path('assessment.json').write_text(json.dumps(result, ensure_ascii=False))
print('computed', mean, 'pages', pages)
'''


async def execute_delivery(workbench, **kwargs):
    require_sandbox()
    script = workbench.workspace / 'analyze.py'
    script.write_text(delivery_script(workbench, **kwargs), encoding='utf-8')
    receipt = await python_sandbox.run(workbench.user, workbench.session, '/workspace/analyze.py',
        output_paths=['/workspace/report.md', '/workspace/assessment.json', '/workspace/facts.json', '/workspace/page.png'])
    assert receipt['exit_code'] == 0, receipt['stderr']
    assert not receipt['timed_out']
    exploration.record_python(workbench.manifest['workflow_id'], workbench.user, workbench.session, receipt)
    return receipt


def submit_delivery(workbench, run_ids, **kwargs):
    return exploration.submit(workbench.manifest['workflow_id'], workbench.user, workbench.session,
        '/workspace/report.md', '/workspace/assessment.json', run_ids, **kwargs)


def test_only_gateway_batch_and_infrastructure_tools_are_registered():
    catalog = load_extensions()
    names = set(catalog.agents['main_fan'].tools)
    assert names == {'prepare_main_fan_task', 'fetch_main_fan_data', 'query_fan_timeseries',
                     'query_fan_sensor_realtime', 'query_fan_equipment_realtime', 'query_fan_equipment_timeseries',
                     'run_python', 'submit_main_fan_report', 'submit_main_fan_mine_plan',
                     'query_main_fan_history', 'fetch_main_fan_reports', 'archive_main_fan_report'}
    assert names.isdisjoint({'inspect_main_fan_document', 'extract_main_fan_pages',
                             'register_main_fan_facts', 'compare_main_fan_data', 'analyze_main_fan'})
    assert catalog.business('main_fan').test_runner is None


@pytest.mark.asyncio
async def test_direct_pdf_read_returns_recoverable_feedback_without_sending_binary():
    from extensions.main_fan.middleware import MainFanSandboxMiddleware
    middleware = MainFanSandboxMiddleware()
    request = SimpleNamespace(tool_call={'name': 'read_file', 'id': 'pdf-read',
                                        'args': {'file_path': '/workspace/uploads/report.pdf'}})
    async def forbidden(request):
        pytest.fail('不应让框架发送 PDF 二进制附件')
    result = await middleware.awrap_tool_call(request, forbidden)
    assert result.status == 'error' and isinstance(result.content, str)
    assert 'run_python' in result.content and 'PNG' in result.content


def test_prepare_exports_raw_files_and_frozen_delivery_schema(workbench):
    manifest = workbench.manifest
    assert manifest['mode'] == 'sandbox_exploration'
    assert manifest['python_paths']['input'] == manifest['input_path'].removeprefix('/workspace/')
    assert manifest['python_paths']['materials'] == ['uploads/scan.pdf']
    assert 'facts_schema_path' not in manifest
    data = json.loads((workbench.workspace / manifest['data_path'].removeprefix('/workspace/')).read_text())
    assert data['reports'] == []
    schema = json.loads((workbench.workspace / manifest['submission_schema_path'].removeprefix('/workspace/')).read_text())
    assert {'result_code', 'summary'} <= set(schema['required'])
    assert exploration.prepare(workbench.request, workbench.user, workbench.session)['workflow_id'] == manifest['workflow_id']
    trusted = exploration.prepare({'mine_code': 'fake', 'mine_name': '伪造'}, workbench.user, workbench.session,
                                  trusted=workbench.request)
    assert trusted['mine_code'] == 'sandbox-mine'


def test_test_lab_defaults_to_agent_and_cannot_use_legacy_pipeline(monkeypatch):
    from api.test_lab_routes import TestRequest as Request
    from runtime import business_tasks
    from runtime.test_lab import service as lab
    from db import business_store
    monkeypatch.setattr(business_tasks, '_controller', business_tasks.BusinessTaskController())
    request = Request(agent_name='main_fan', input={'mine_code': 'test', 'mine_name': '测试矿'})
    assert request.execution_mode == 'agent'
    result = lab.start_run(request, 'lab-default-test')
    assert business_store.get(result['id'])['source'] == 'test_agent'
    with pytest.raises(ValueError, match='完整智能体'):
        lab.start_run(request.model_copy(update={'execution_mode': 'pipeline'}), 'lab-default-test')


@pytest.mark.asyncio
async def test_real_sandbox_parses_pdf_computes_and_generates_report_without_legacy_pipeline(workbench, monkeypatch):
    def forbidden(*args, **kwargs):
        raise AssertionError('自主模式不能调用固定业务解析、计算或报告生成')
    from algorithms.main_fan import equivalence
    from extensions.main_fan import documents, pipeline, reporting
    for name in ('normalize_and_validate_fan_data', 'compare_basic_parameters', 'analyze_fan_switch_event',
                 'assess_fan_equivalent_capability'):
        monkeypatch.setattr(equivalence, name, forbidden)
    monkeypatch.setattr(documents, '_prepare', forbidden)
    monkeypatch.setattr(pipeline, 'run_pipeline', forbidden)
    monkeypatch.setattr(reporting, 'render_result', forbidden)
    monkeypatch.setattr(service, '_screen', forbidden)
    receipt = await execute_delivery(workbench)
    assert 'computed 20 pages 1' in receipt['stdout']
    assert (workbench.workspace / 'page.png').is_file()
    result = submit_delivery(workbench, [receipt['run_id']], artifacts=['/workspace/facts.json'])
    assert result['status'] == 'completed'
    assert result['analysis'] is None
    assert result['result'] == (workbench.workspace / 'report.md').read_text()
    assert result['assessment']['agent_result']['details']['mean'] == 20
    # 故意不生成旧七阶段结构；平台编码与摘要完全来自 Agent 交付，不被固定算法改成 UNDETERMINED。
    judgment = result['platform_payload']['judgment']
    assert judgment['resultCode'] == 'WARNING'
    assert judgment['result'] == '脚本实际计算均值20.0'
    assert result['assessment']['evidence'][0]['sha256']
    from runtime.test_lab.spreadsheets import result_tables, write_workbook, preview_workbook
    workbook = workbench.workspace / 'export.xlsx'
    write_workbook(workbook, result_tables(result))
    preview = preview_workbook(workbook)
    assert preview['sheets'][1]['name'] == 'Agent分析明细'
    assert ['mean', 20] in preview['sheets'][1]['rows']
    assert len(main_fan_store.workflow(workbench.manifest['workflow_id'], workbench.user, workbench.session)['stages']) == 3
    assert service.validate_result({**result, 'result': '篡改'}, workbench.user, workbench.session) == result
    assert submit_delivery(workbench, [receipt['run_id']]) == result
    with pytest.raises(ValueError, match='不属于'):
        service.validate_result(result, 'another-user', workbench.session)


@pytest.mark.asyncio
async def test_relative_evidence_path_binds_exact_same_trusted_source(workbench):
    require_sandbox()
    code = delivery_script(workbench).replace(
        '/workspace/exploration/' + workbench.manifest['workflow_id'] + '/input.json',
        'exploration/' + workbench.manifest['workflow_id'] + '/input.json')
    (workbench.workspace / 'relative.py').write_text(code)
    receipt = await python_sandbox.run(workbench.user, workbench.session, '/workspace/relative.py',
                                      output_paths=['/workspace/report.md', '/workspace/assessment.json'])
    assert receipt['ok'], receipt['stderr']
    exploration.record_python(workbench.manifest['workflow_id'], workbench.user, workbench.session, receipt)
    result = submit_delivery(workbench, [receipt['run_id']])
    assert result['assessment']['evidence'][0]['source'] == workbench.manifest['input_path']


@pytest.mark.asyncio
@pytest.mark.parametrize('target', ['report.md', 'assessment.json', 'facts.json'])
async def test_changed_script_outputs_cannot_be_submitted(workbench, target):
    receipt = await execute_delivery(workbench)
    (workbench.workspace / target).write_text('tampered')
    with pytest.raises(ValueError):
        submit_delivery(workbench, [receipt['run_id']], artifacts=['/workspace/facts.json'])


@pytest.mark.asyncio
async def test_delivery_requires_successful_execution_from_same_workflow(workbench):
    receipt = await execute_delivery(workbench)
    with pytest.raises(ValueError, match='成功执行记录'):
        submit_delivery(workbench, [])
    second = exploration.prepare({**workbench.request, 'task_description': '独立任务'}, workbench.user, workbench.session)
    with pytest.raises(ValueError, match='不属于本轮'):
        exploration.submit(second['workflow_id'], workbench.user, workbench.session,
                           '/workspace/report.md', '/workspace/assessment.json', [receipt['run_id']])
    with pytest.raises(ValueError, match='不属于'):
        exploration._load(workbench.manifest['workflow_id'], 'someone-else', workbench.session)


@pytest.mark.asyncio
async def test_python_tool_returns_concrete_paths_when_agent_code_fails(workbench):
    require_sandbox()
    from extensions.main_fan.exploration_tools import run_python
    (workbench.workspace / 'wrong-path.py').write_text("from pathlib import Path\nPath('missing-input.json').read_text()")
    runtime = SimpleNamespace(config={'configurable': {'workspace_user': workbench.user,
                                                      'workspace_session': workbench.session}})
    result = await run_python('/workspace/wrong-path.py', runtime, workflow_id=workbench.manifest['workflow_id'])
    assert result['ok'] is False
    assert 'FileNotFoundError' in result['stderr']
    assert result['python_paths']['input'] == workbench.manifest['python_paths']['input']
    assert '不要' in result['next_action']


@pytest.mark.asyncio
async def test_preexisting_files_are_not_mislabeled_as_script_generated_outputs(workbench):
    require_sandbox()
    (workbench.workspace / 'existing.md').write_text('只是之前写入的文件')
    (workbench.workspace / 'noop.py').write_text("print('no output written')")
    receipt = await python_sandbox.run(workbench.user, workbench.session, '/workspace/noop.py',
                                      output_paths=['/workspace/existing.md'])
    assert receipt['ok']
    assert receipt['outputs'] == []
    assert receipt['unchanged_outputs'] == ['/workspace/existing.md']


@pytest.mark.parametrize('number', ['NaN', 'Infinity', '1e999'])
def test_delivery_json_rejects_nonfinite_numbers(tmp_path, number):
    path = tmp_path / 'result.json'
    path.write_text('{"details":{"value":' + number + '}}')
    with pytest.raises(ValueError, match='非有限'):
        exploration._read_json(path)


@pytest.mark.asyncio
async def test_raw_sources_and_frozen_inputs_cannot_be_replaced(workbench):
    receipt = await execute_delivery(workbench)
    original = workbench.source.read_bytes()
    workbench.source.write_bytes(b'changed pdf')
    with pytest.raises(ValueError, match='已被修改'):
        submit_delivery(workbench, [receipt['run_id']])
    workbench.source.write_bytes(original)
    rules = workbench.workspace / workbench.manifest['rules_path'].removeprefix('/workspace/')
    rules.write_text('{}')
    with pytest.raises(ValueError, match='已被修改'):
        submit_delivery(workbench, [receipt['run_id']])


@pytest.mark.asyncio
async def test_unknown_evidence_is_rejected_and_agent_can_correct_and_resubmit(workbench):
    receipt = await execute_delivery(workbench)
    # 修改生成脚本后重新执行，保证失败来自来源归属而非哈希变化。
    script = delivery_script(workbench).replace('/input.json', '/other-task.json')
    (workbench.workspace / 'analyze.py').write_text(script)
    bad = await python_sandbox.run(workbench.user, workbench.session, '/workspace/analyze.py',
                                  output_paths=['/workspace/report.md', '/workspace/assessment.json'])
    assert bad['exit_code'] == 0, bad['stderr']
    exploration.record_python(workbench.manifest['workflow_id'], workbench.user, workbench.session, bad)
    with pytest.raises(ValueError, match='证据来源'):
        submit_delivery(workbench, [bad['run_id']])
    corrected = await execute_delivery(workbench)
    assert submit_delivery(workbench, [corrected['run_id']])['analysis_id']


@pytest.mark.asyncio
async def test_gateway_fetch_still_uses_host_connector_and_updates_raw_files(workbench, monkeypatch):
    import hashlib
    from extensions.main_fan import data_source
    inputs = {**workbench.request, 'use_gateway': True, 'materials': []}
    prepared = exploration.prepare(inputs, workbench.user, workbench.session)
    calls = []
    class Gateway:
        async def download(self, agent_code, source, local):
            calls.append(('download', agent_code, source.file_id))
            local.write_bytes(workbench.source.read_bytes())
            return {'sha256': hashlib.sha256(local.read_bytes()).hexdigest()}
    gateway = Gateway()
    async def fetch(connector, agent_code, mine_code):
        assert connector is gateway
        calls.append(('fetch', agent_code, mine_code))
        return {'devices': [], 'points': [], 'operating_series': [], 'switch_records': [], 'fan_state_records': [],
                'files': [{'file_id': 'file-1', 'version': 'v1', 'filename': 'remote.pdf', 'path': '/files/1'}]}
    monkeypatch.setattr(exploration, 'get_service', lambda name: gateway)
    monkeypatch.setattr(data_source, 'fetch', fetch)
    first = await exploration.fetch(prepared['workflow_id'], workbench.user, workbench.session)
    second = await exploration.fetch(prepared['workflow_id'], workbench.user, workbench.session)
    assert len(first['materials']) == len(second['materials']) == 1
    assert [call[0] for call in calls] == ['fetch', 'download']
    material = first['materials'][0]
    assert (workbench.workspace / material['path'].removeprefix('/workspace/')).is_file()
    raw = json.loads((workbench.workspace / first['input_path'].removeprefix('/workspace/')).read_text())
    assert raw['materials'][0]['sha256'] == material['sha256']


@pytest.mark.asyncio
async def test_sandbox_denies_host_secrets_network_and_process_execution(workbench, monkeypatch):
    require_sandbox()
    secret = workbench.workspace.parent / 'host-secret.txt'
    secret.write_text('host-only-value')
    monkeypatch.setenv('SANDBOX_TEST_SECRET', 'not-for-child')
    listener = socket.socket()
    listener.bind(('127.0.0.1', 0))
    listener.listen()
    port = listener.getsockname()[1]
    program = '''
from pathlib import Path
import os, socket, subprocess, sys
assert os.getenv('SANDBOX_TEST_SECRET') is None
try:
    Path('../host-secret.txt').read_text()
except (PermissionError, FileNotFoundError):
    print('host-file-denied')
else:
    raise AssertionError('host file readable')
try:
    socket.create_connection(('127.0.0.1', TEST_PORT), timeout=1)
except OSError:
    print('network-denied')
else:
    raise AssertionError('network accessible')
if sys.platform == 'darwin':
    try:
        subprocess.run(['/bin/sh', '-c', 'true'], check=True)
    except PermissionError:
        print('process-denied')
    else:
        raise AssertionError('arbitrary process allowed')
else:
    print('processes-remain-in-namespace')
'''
    (workbench.workspace / 'boundary.py').write_text(program.replace('TEST_PORT', str(port)))
    try:
        receipt = await python_sandbox.run(workbench.user, workbench.session, '/workspace/boundary.py')
        listener.settimeout(.1)
        with pytest.raises(TimeoutError):
            listener.accept()
    finally:
        listener.close()
    assert receipt['exit_code'] == 0, receipt['stderr']
    assert 'host-file-denied' in receipt['stdout']
    assert 'network-denied' in receipt['stdout']
    assert 'process-denied' in receipt['stdout'] or 'processes-remain-in-namespace' in receipt['stdout']


@pytest.mark.asyncio
async def test_sandbox_failure_timeout_and_path_escape_are_real(workbench):
    require_sandbox()
    script = workbench.workspace / 'failure.py'
    script.write_text("raise RuntimeError('real-script-failure')")
    failed = await python_sandbox.run(workbench.user, workbench.session, '/workspace/failure.py')
    assert failed['exit_code'] != 0 and 'real-script-failure' in failed['stderr']
    with pytest.raises(ValueError, match='失败或超时'):
        python_sandbox.execution_receipt(workbench.user, workbench.session, failed['run_id'])
    script.write_text('import time\ntime.sleep(10)')
    timed = await python_sandbox.run(workbench.user, workbench.session, '/workspace/failure.py', timeout_seconds=1)
    assert timed['timed_out']
    with pytest.raises(ValueError, match='越出'):
        python_sandbox.workspace_file(workbench.user, workbench.session, '/workspace/../../secret')
    (workbench.workspace / 'escape.py').symlink_to(workbench.workspace.parent / 'outside.py')
    with pytest.raises(ValueError, match='越出'):
        await python_sandbox.run(workbench.user, workbench.session, '/workspace/escape.py')


@pytest.mark.asyncio
async def test_missing_sandbox_never_falls_back_to_host_execution(workbench, monkeypatch):
    monkeypatch.setattr(python_sandbox.shutil, 'which', lambda name: None)
    (workbench.workspace / 'touch.py').write_text("from pathlib import Path\nPath('unsafe-marker').touch()")
    with pytest.raises(ValueError, match='沙箱执行'):
        await python_sandbox.run(workbench.user, workbench.session, '/workspace/touch.py')
    assert not (workbench.workspace / 'unsafe-marker').exists()


@pytest.mark.parametrize('value', [
    {'result_code': 'NORMAL', 'summary': '无来源'},
    {'result_code': 'BAD', 'summary': '未知编码'},
    {'result_code': 'UNDETERMINED', 'summary': '  '},
    {'status': 'needs_materials', 'result_code': 'WARNING', 'summary': '矛盾状态',
     'evidence': [{'source': '/workspace/input.json'}]},
])
def test_delivery_contract_rejects_invalid_metadata(value):
    with pytest.raises(ValueError):
        AgentAssessment.model_validate(value)