diff --git a/app/services/calculation_worker.py b/app/services/calculation_worker.py index 49912a9..55b2003 100644 --- a/app/services/calculation_worker.py +++ b/app/services/calculation_worker.py @@ -93,8 +93,8 @@ async def execute_child(input_path: str, output_path: str): category = 'EXECUTION_CONTEXT_UNAVAILABLE' payload = {'status': 'failed', 'error': {'code': category}} encoded = json.dumps(payload, allow_nan=False) - if len(encoded.encode()) > 8 * 1024 * 1024: - encoded = json.dumps({'status': 'failed', 'error': {'code': 'RESULT_TOO_LARGE'}}) + # Keep the complete calculation for persistence. MCP result reads apply + # their own summary/page budgets after the result has been saved. Path(output_path).write_text(encoded) @@ -212,8 +212,6 @@ class CalculationWorker: terminal = {'status': 'timed_out', 'error': {'code': 'EXECUTION_TIMEOUT'}} elif process.returncode is not None: try: - if output_path.stat().st_size > 8 * 1024 * 1024: - raise ValueError('Result exceeds persistence limit') terminal = json.loads(output_path.read_text()) except (OSError, ValueError): terminal = {'status': 'failed', 'error': {'code': 'CHILD_PROCESS_FAILED'}} diff --git a/app/services/investigation_graph.py b/app/services/investigation_graph.py index cadc506..65816f5 100644 --- a/app/services/investigation_graph.py +++ b/app/services/investigation_graph.py @@ -39,6 +39,7 @@ def update_investigation_graph( )) nodes = {node.node_id: node.model_copy(deep=True) for node in previous.node_refs} if previous else {} + saved_ids = set(nodes) require(len({node.node_id for node in update.nodes}) == len(update.nodes), "GRAPH_NODE_DUPLICATE", "Graph update contains duplicate node ids", status=422) for node in update.nodes: @@ -81,6 +82,26 @@ def update_investigation_graph( edges.append(edge.model_copy(deep=True)) require(len(nodes) <= 100 and len(edges) <= 200, "GRAPH_CAPACITY_EXCEEDED", "Investigation graph exceeds 100 nodes or 200 edges; existing paths were retained", status=422) + # Validate only newly added steps. Historical snapshots and status-only + # patches stay valid, even if an earlier publisher omitted relationships. + new_ids = set(nodes) - saved_ids + if new_ids and len(nodes) > 1: + neighbours = {node_id: set() for node_id in nodes} + for edge in edges: + neighbours[edge.source].add(edge.target) + neighbours[edge.target].add(edge.source) + connected = set(saved_ids) or {next(iter(nodes))} + pending = list(connected) + while pending: + for node_id in neighbours[pending.pop()] - connected: + connected.add(node_id) + pending.append(node_id) + missing = sorted(new_ids - connected) + require(not missing, "GRAPH_RELATION_REQUIRED", + "Add explicit edges connecting these new steps to the investigation narrative: " + + ", ".join(missing) + + ". Describe branches, parallel findings or retries; edges do not schedule execution.", + status=422) lines = [source or "flowchart TD"] for node in nodes.values(): if in_source(node.node_id): diff --git a/docs/calculation-queue.md b/docs/calculation-queue.md index c74d8c5..1d80deb 100644 --- a/docs/calculation-queue.md +++ b/docs/calculation-queue.md @@ -43,6 +43,9 @@ API 保留 `uvicorn --workers 4`。长计算在项目 BizDB 持久化入队， 调查的排队回执不写成分析证据。成功结果读取时，按原计算来源、任务及运行验证，并幂等地保存证据。未完成的计算阻止提交最终报告、`insufficient_data` 或将运行标记成功；中间进度仍可保存。 +计算子进程的完整结果文件及 supervisor 读取不设 8MiB 整份体积上限，全网分析结果按原值保存到 `calculation_jobs.result`。 +Agent 读取仍使用 `get_calculation_result` 的摘要及 `data_path/limit/offset` 分页预算；完整结果保存与单次工具消息大小分别处理。 + Agent 调查执行在原生回合空闲后由宿主短轮询等待，继续维持调查心跳。新作业进入终态即触发模型读取结果，其余作业继续后台运行。状态读取遇到网络、限流或临时服务错误，最多重试 3 次，采用指数退避和随机抖动；权限拒绝、上下文失效立即停止，重试耗尽明确报错，不重新提交或取消计算。普通聊天展示任务编号，可在当前会话后续查询。异常流结束时若无法安全继续处理新结果，会明确失败而不是虚报完成；Server 中的计算结果仍保留。 ## 排队期限与结果保留 diff --git a/docs/investigation-progress-and-presentation.md b/docs/investigation-progress-and-presentation.md index 87ca368..e210388 100644 --- a/docs/investigation-progress-and-presentation.md +++ b/docs/investigation-progress-and-presentation.md @@ -57,6 +57,12 @@ Server 不验证依赖执行顺序或推断节点完工。 `progress.graph_update` 包含可选 `title`、`nodes` 和 `edges`。节点沿用 `node_id/label/status/evidence_refs/action_refs/map_targets`， 可用 `detail` 记录实际结果或失败原因；连线为 `{source,target,label?}`。仅提交新增或变化的内容， `update_investigation_graph` 合并节点、追加明确连线并生成 Mermaid，保留已有路径和引用。 +新建调查图使用 `flowchart TD`，按自上而下的阅读顺序展示。多步骤图必须显式连接流程关系： +独立核查从共同问题分叉并在判断处汇总，条件分支标注条件，失败尝试连接重试或替代路线。 +新增节点须通过连线接入已有调查脉络；首次发布多个节点时须形成相连的图。 +未接入的新增节点返回 `GRAPH_RELATION_REQUIRED`（422），错误中列出需补关系的节点。 +根据真实关系修正后重试，不按节点列表顺序自动补线；关系仅解释调查，不约束执行先后。 +单个起点、仅更新已有节点、纯文字更新和已保存的历史图仍可不传连线，不重排或改写历史报告。 “保留路径”约束图的结构，不冻结已有节点的标记。`nodes` 按 `node_id` 合并：新增节点必须提供 `label`； 修改已保存节点时可只提交 `node_id` 与变化字段，无需再提供 `label`、全部节点或连线。 省略字段保留原值，`label: null` 也保留名称；`status: null` 显式恢复为参考步骤。 diff --git a/docs/mcp-query-implementation.md b/docs/mcp-query-implementation.md index cbac94f..b765263 100644 --- a/docs/mcp-query-implementation.md +++ b/docs/mcp-query-implementation.md @@ -16,7 +16,7 @@ Agent 根据401 challenge 或路径/根 well-known 发现受保护资源元数 普通查询默认20条；目录类最多500条，显式指定设备的时序查询最多10000条、20个标识、24小时；offset 最大10000。SCADA调查首次省略 `query_scada_readings.device_ids`，由Server读取项目完整设备目录及窗口内全部授权观测，内部自动翻页，`offset` 必须为0。其余查询按问题选定对象和窗口。SQL 有 limit/offset 和 statement_timeout。历史对比在同一条SQL中按序列、时间对数据点分页，返回 page.has_more；非零offset也标明覆盖不完整。历史曲线展示调用初始每页500点，`RESULT_TOO_LARGE` 时原位减半重试，最多读取1000条、发出16次请求，共用120秒超时（单次请求同时受该剩余总预算约束），图表最多100个有效点；这不改变模型首次查询默认值或时序接口上限。 -单份结果超过8MiB报错；超过12KB的调查工具消息默认返回证据资源引用，设备目录的精简路径见下段。完整快照仍保存，Server 的 `resources/read` 可以读取它。Agent 本地 MCP 只提供工具，将上游资源引用转为 `evidence_read` 中的 `mcp_tjwater_get_task_evidence` 参数，不向 OpenCode 泄露不可在本地读取的链接。`get_task_evidence` 在数据库中按 data_path、limit、offset 选页，嵌套集合和超过256字符的字符串返回路径描述；字符串按 Unicode 字符分页。页和完整资源均在数据库返回前检查8MiB上限，保留原证据编号和合成来源；普通查询要求缩小范围。其他结果仍保留执行后的整体8MiB检查，不能把它理解为所有查询均有数据库端字节限制。原始/清洗实测、模型基线、分析结果、比较和 synthetic 均明确标记。非有限数值转为 null 并记录数据缺口。 +同步查询单份结果超过8MiB报错；后台计算作业的完整结果保存不设此上限，读取时仍执行工具消息摘要和分页预算；超过12KB的调查工具消息默认返回证据资源引用，设备目录的精简路径见下段。完整快照仍保存，Server 的 `resources/read` 可以读取它。Agent 本地 MCP 只提供工具，将上游资源引用转为 `evidence_read` 中的 `mcp_tjwater_get_task_evidence` 参数，不向 OpenCode 泄露不可在本地读取的链接。`get_task_evidence` 在数据库中按 data_path、limit、offset 选页，嵌套集合和超过256字符的字符串返回路径描述；字符串按 Unicode 字符分页。页和完整资源均在数据库返回前检查8MiB上限，保留原证据编号和合成来源；普通查询要求缩小范围。其他结果仍保留执行后的整体8MiB检查，不能把它理解为所有查询均有数据库端字节限制。原始/清洗实测、模型基线、分析结果、比较和 synthetic 均明确标记。非有限数值转为 null 并记录数据缺口。 `list_scada_devices` 默认每页100个、最多500个，其他集合仍沿用各自的分页默认值。设备目录超过12,000字节时，先内联本页全部设备的 `device_id`、`device_type` 和 `measurement_unit`，省略其他属性并明确标注；模型映射、坐标等通过 `get_scada_device` 核验，任务证据和幂等回执保留完整属性。此时 `truncated` 表示属性省略，设备是否还有后续页以 `page.has_more` 为准，按本页实际设备数递增 `offset`。精简目录仍超预算时沿用证据引用或缩页路径。 diff --git a/tests/unit/test_calculation_worker.py b/tests/unit/test_calculation_worker.py index 4caa43c..9db47e5 100644 --- a/tests/unit/test_calculation_worker.py +++ b/tests/unit/test_calculation_worker.py @@ -1,5 +1,6 @@ """Worker boundaries use real managed subprocesses, without external databases.""" import asyncio +import json from functools import wraps from types import SimpleNamespace from contextlib import asynccontextmanager @@ -27,6 +28,66 @@ def run_async(function): return wrapper +@run_async +async def test_child_keeps_complete_result_above_eight_mib(monkeypatch, tmp_path): + from app.mcp import calculation_jobs + from app.services import calculation_worker + + data = {"comparisons": "x" * (8 * 1024 * 1024), "map_payload": {"features": ["coverage"]}} + + @asynccontextmanager + async def scope(row): + yield object() + + async def calculate(scope, row): + return SimpleNamespace(model_dump=lambda **kwargs: data) + + monkeypatch.setattr(calculation_worker, "execution_scope", scope) + monkeypatch.setattr(calculation_jobs, "execute_calculation_job", calculate) + input_path, output_path = tmp_path / "input.json", tmp_path / "result.json" + input_path.write_text("{}") + await calculation_worker.execute_child(str(input_path), str(output_path)) + assert output_path.stat().st_size > 8 * 1024 * 1024 + assert json.loads(output_path.read_text()) == {"status": "succeeded", "result": data} + + +@run_async +async def test_supervisor_persists_complete_result_above_eight_mib(monkeypatch, tmp_path): + monkeypatch.setenv("CALCULATION_WORKER_STATE_DIR", str(tmp_path)) + original = asyncio.create_subprocess_exec + + async def spawn(*args, **kwargs): + # A real child writes the same file protocol as the calculation process. + code = ('import json,pathlib,sys; pathlib.Path(sys.argv[1]).write_text(' + 'json.dumps({"status":"succeeded","result":{"comparisons":"x"*(8*1024*1024),' + '"map_payload":{"features":["coverage"]}}}))') + return await original(sys.executable, "-c", code, args[-1], start_new_session=True) + + monkeypatch.setattr(asyncio, "create_subprocess_exec", spawn) + results = [] + + class Repository: + async def heartbeat(self, *args): + return "running" + + async def finish(self, row, owner, **terminal): + results.append(terminal) + return True + + @asynccontextmanager + async def repository(project): + yield Repository() + + worker = CalculationWorker(WorkerSettings(heartbeat_seconds=.02), repository=repository) + await asyncio.wait_for(worker.run_job({"project_id": "project", "job_id": "large-result", + "timeout_seconds": 10}), 15) + assert len(results) == 1 + assert results[0]["status"] == "succeeded" + assert results[0]["result"] == { + "comparisons": "x" * (8 * 1024 * 1024), "map_payload": {"features": ["coverage"]}, + } + + def test_worker_settings_require_valid_concurrency(monkeypatch): settings = WorkerSettings.from_environment() assert settings.max_concurrency == 2 diff --git a/tests/unit/test_investigation_graph.py b/tests/unit/test_investigation_graph.py index f5b3b61..0daeae2 100644 --- a/tests/unit/test_investigation_graph.py +++ b/tests/unit/test_investigation_graph.py @@ -239,6 +239,85 @@ def test_graph_renderer_escapes_labels_and_keeps_explicit_parallel_and_failed_pa assert graph.node_refs[2].status == "reference" +@pytest.mark.parametrize("edges", [[], [{"source": "check", "target": "check"}]]) +def test_new_multistep_graph_requires_relationships_instead_of_isolated_nodes(edges): + with pytest.raises(DomainError) as error: + update_investigation_graph(None, InvestigationGraphUpdate.model_validate({ + "nodes": [{"node_id": "read", "label": "观测核验"}, + {"node_id": "check", "label": "异常判断"}], + "edges": edges, + })) + assert error.value.code == "GRAPH_RELATION_REQUIRED" + assert "check" in str(error.value) + + +def test_new_branch_must_connect_to_the_saved_narrative_without_inventing_edges(): + original = update_investigation_graph(None, InvestigationGraphUpdate.model_validate({ + "nodes": [{"node_id": "root", "label": "调查起点"}], + })) + with pytest.raises(DomainError) as error: + update_investigation_graph(original, InvestigationGraphUpdate.model_validate({ + "nodes": [{"node_id": "left", "label": "独立核查"}, + {"node_id": "right", "label": "核查发现"}], + "edges": [{"source": "left", "target": "right"}], + })) + assert error.value.code == "GRAPH_RELATION_REQUIRED" + assert original.edges == [] + assert len(original.node_refs) == 1 + + +def test_connected_parallel_branches_and_join_render_top_down(): + graph = update_investigation_graph(None, InvestigationGraphUpdate.model_validate({ + "nodes": [{"node_id": key, "label": label} for key, label in ( + ("start", "异常核验"), ("isolation", "隔离分析"), + ("impact", "爆管影响"), ("summary", "汇总判断"))], + "edges": [{"source": source, "target": target, "label": label} + for source, target, label in ( + ("start", "isolation", "并行核查"), ("start", "impact", "并行核查"), + ("isolation", "summary", "汇总依据"), ("impact", "summary", "汇总依据"))], + })) + assert graph.mermaid.startswith("flowchart TD\n") + assert len(graph.edges) == 4 + assert 'start -->|"并行核查"| impact' in graph.mermaid + patched = update_investigation_graph(graph, InvestigationGraphUpdate.model_validate({ + "nodes": [{"node_id": "impact", "status": "recorded"}], + })) + assert patched.edges == graph.edges + + +def test_existing_disconnected_snapshot_remains_readable_and_patchable(): + original = InvestigationDiagram.model_validate({ + "title": "已保存的旧图", "mermaid": 'flowchart TD\n read["观测"]\n check["判断"]', + "edges": [], "node_refs": [{"node_id": "read", "label": "观测"}, + {"node_id": "check", "label": "判断"}], + }) + patched = update_investigation_graph(original, InvestigationGraphUpdate.model_validate({ + "nodes": [{"node_id": "check", "status": "recorded"}], + })) + assert patched.mermaid == original.mermaid + assert patched.edges == original.edges == [] + assert original.node_refs[1].status is None + + +@pytest.mark.anyio +async def test_publication_rejects_missing_links_and_accepts_a_corrected_graph(): + task, repo, service = setup_service() + first = await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "read", "label": "观测核验"}, + ]) + original = first.task.latest_progress.progress.diagram.model_dump(mode="json") + with pytest.raises(DomainError) as error: + await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "check", "label": "异常判断"}, + ]) + assert error.value.code == "GRAPH_RELATION_REQUIRED" + assert (await service.detail(task["task_id"])).latest_progress.progress.diagram.model_dump(mode="json") == original + corrected = await publish(service, repo, task["task_id"], nodes=[ + {"node_id": "check", "label": "异常判断"}, + ], edges=[{"source": "read", "target": "check", "label": "核验依据"}]) + assert 'read -->|"核验依据"| check' in corrected.task.latest_progress.progress.diagram.mermaid + + def test_first_structured_update_preserves_an_existing_mermaid_without_reinterpreting_it(): original = InvestigationDiagram.model_validate({ "title": "原调查路线", "mermaid": 'flowchart LR\nA{核查条件} -->|具备| B[原路径]\nA --> C[另一分支]',