diff --git a/skills/system-simulation/SKILL.md b/skills/system-simulation/SKILL.md new file mode 100644 index 0000000..cf96c3d --- /dev/null +++ b/skills/system-simulation/SKILL.md @@ -0,0 +1,40 @@ +--- +name: system-simulation +description: 读取、校验并简要解释 SystemSimulationApp 工程 JSON v1 或 System XML v3，安全规范化文件文本，并在用户选定结果曲线后启动、监视或取消仿真及导出 CSV。适用于检查模型文件、修复编码或换行、运行仿真和获取结果；不用于旧格式迁移、任意语义修复、自动迭代或网页自动预装。 +--- + +# 系统仿真 + +使用本 Skill 随附的确定性脚本检查模型、调用现有后端并保存结果；不要让语言模型自行重写模型或猜测求解数据。 + +## 基本边界 + +- 仅处理 ReactFlow 工程 JSON v1 和 System XML v3。版本缺失、不受支持或模型版本不匹配时，说明问题并停止，不进行迁移猜测。 +- 组件参数是仿真前设定的固定输入；结果变量才是可随时间绘制的量。不要把“参数”当成结果曲线。 +- 文件通过格式校验不等于物理系统一定可求解。不要隐瞒编译或运行阶段的诊断。 +- 不直接覆盖源文件，不自行修改参数、连接、组件类型、模型版本或求解设置。 +- 本版不支持把模型自动注入网页、生成可直接打开的预装页面、任意损坏文件修复、模型迁移或自动调参迭代。明确告知用户这些能力尚未实现，不要用手工网页操作冒充支持。 + +处理文件、解释格式或选择结果变量时，读取 [references/file-contracts.md](references/file-contracts.md)。请求文件修复时，再读取 [references/repair-policy.md](references/repair-policy.md)。需要运行、监视、取消仿真或交付结果时，读取 [references/workflows.md](references/workflows.md)。 + +## 工作原则 + +1. 先用 `inspect` 确认输入格式、版本、结构和诊断，再基于检查结果简要解释组件、连接与仿真设置。 +2. 如果用户要求修复，只能执行文本规范化。先展示预览和源文件 SHA-256，获得针对该预览的明确确认后，才可写入另一个输出路径；随后重新 `inspect`。 +3. 仿真前必须让用户选择直接曲线查看方式，并解析具体结果变量： + - 分别查看所选变量； + - 将多个同单位、可比较的变量叠加； + - 将不同物理量或单位的变量上下排列。 +4. 用户用显示名称描述组件或变量时，利用检查结果中的稳定 ID、结果 `key`、物理量和单位消歧。存在重名、多个候选或“参数/结果变量”含义不清时，先询问，不能替用户猜。 +5. 使用 `simulate` 的事件流持续判断 queued、validating、compiling、integrating 和结束状态。仿真时间暂时不变但内部活动仍增长时，只说明正在处理慢步，不能宣称卡死。 +6. 成功运行后交付用户选择的 SVG 曲线和完整 `results.csv`，并简要说明完成状态、实际仿真终点和重要诊断。失败或取消时交付能够安全生成的部分结果；若运行前即失败而没有 CSV，要明确说明原因。 + +脚本命令统一从仓库根目录运行： + +```powershell +py -3.12 skills/system-simulation/scripts/simulation_skill.py --help +``` + +Windows 优先使用仓库 `.venv-win\Scripts\python.exe`（若存在），否则使用 `py -3.12`；Linux 优先使用 `.venv/bin/python`，否则使用 `python3.12`。不要调用未经版本确认的 `python`，本项目要求 Python 3.12。 + +优先依赖脚本返回的结构化 JSON/JSONL、稳定错误码和退出码做判断，不解析中文提示文本来驱动下一步。 diff --git a/skills/system-simulation/agents/openai.yaml b/skills/system-simulation/agents/openai.yaml new file mode 100644 index 0000000..171568c --- /dev/null +++ b/skills/system-simulation/agents/openai.yaml @@ -0,0 +1,7 @@ +interface: + display_name: "系统仿真文件助手" + short_description: "读取与校验模型文件，监视仿真并导出结果曲线和 CSV 文件" + default_prompt: "使用 $system-simulation 检查我的模型文件，并在我选定结果曲线后运行和监视仿真。" + +policy: + allow_implicit_invocation: true diff --git a/skills/system-simulation/references/file-contracts.md b/skills/system-simulation/references/file-contracts.md new file mode 100644 index 0000000..2ced179 --- /dev/null +++ b/skills/system-simulation/references/file-contracts.md @@ -0,0 +1,95 @@ +# 文件合同与解释规则 + +## 支持范围 + +本 Skill 只接受以下两种当前格式： + +| 格式 | 版本标志 | 用途 | +| --- | --- | --- | +| ReactFlow 工程 JSON | 顶层 `projectSchemaVersion: 1` | 保存组件、画布、端口显示快照、连线和仿真设置，适合继续编辑 | +| System XML | 根元素 `System/@schemaVersion="3"` 且 `unitSystem="SI"` | 保存可执行模型，适合校验、编译和求解 | + +默认让 `inspect --format auto` 根据内容和扩展名识别格式。若内容与扩展名不一致、无法唯一识别或用户明确指定格式，则报告实际证据，不悄悄按另一种格式解释。 + +System XML v1/v2、缺少 `projectSchemaVersion` 的旧工程、字符串端口和不匹配的组件 `modelVersion` 均不属于本 Skill 的迁移范围。不能只改版本号使其看似当前格式。 + +## 工程 JSON v1 + +顶层合同为： + +```text +projectSchemaVersion = 1 +name +nodes[] +edges[] +simulation { t_start, t_stop, step, max_step, method } +``` + +重要规则： + +- 节点的 `id` 是实例稳定标识；显示标签不能替代它。 +- `data.modelType` 标识注册模型，`data.modelVersion` 必须与当前组件目录精确匹配，执行前不得自动补成当前版本。 +- `data.parameters` 保存输入值；`parameterUnits`、科学计数法偏好、坐标、旋转和镜像属于编辑显示信息。 +- 连接必须保留两端组件及 Handle。不能根据节点位置猜测缺失端口。 +- `simulation.step` 是结果采样间隔；`max_step` 是求解器内部步长上限，两者不能混用。 + +工程 JSON 可以导出为 System XML v3，但转换后不会保留全部画布显示信息的对等逆转换合同。 + +## System XML v3 + +XML v3 只描述“求解什么”： + +- 每个 `Component` 必须有唯一 `id`、注册 `type`、精确 `modelVersion` 和完整 SI 参数； +- 每条连接由两个 `Endpoint(component, port)` 组成；端口类型、方向和物理合同由后端注册表恢复； +- `Simulation/@sampleStep` 对应工程 JSON 的 `simulation.step`； +- 不保存组件位置、旋转、镜像、显示单位或端口显示快照； +- 当前后端固定按 v3 校验，不会根据文件内容选择旧解析器。 + +XML 校验依次覆盖安全/语法、XSD 和语义层。通过这些检查后，编译和求解仍可能发现未连接端口、缺少储能锚点、方程结构或数值问题。 + +## 简要解释模型 + +解释必须依据 `inspect` 的结构化输出以及组件目录，而不是仅凭组件名称推测。优先说明： + +1. 文件格式、版本和项目名； +2. 仿真起止时间、采样间隔、最大内部步长和算法； +3. 组件数量、稳定 ID、模型类型和主要输入参数； +4. 连接数量、连接端点及能够确定的物理域； +5. 错误、警告，以及它们属于格式、语义、编译还是运行阶段。 + +保持“文件合同正确”和“物理模型合理”两个结论分开。没有组件文档或注册元数据支持时，不声称某个参数具有推测出的物理效果。 + +## 参数与结果变量 + +必须明确区分： + +- **参数**：仿真开始前设定的固定输入，例如质量、初始压力、摩擦选项；通常没有时间序列。 +- **结果变量**：仿真返回的时间序列，例如位移、速度、压力或流量；只有这类量可以选作曲线。 + +选择曲线时以结果元数据为准，至少核对： + +```text +key + componentId + componentType + label/quantity + unit +``` + +稳定 `key` 是传给 `simulate --variables` 的最终标识。用户只说“质量块的速度”而存在多个质量块，或一个组件存在多个符合描述的速度结果时，列出候选的组件 ID、结果名称和单位，请用户消歧。 + +`inspect` 默认对组件摘要、连接和结果变量分页。先读取 `componentTypes` 了解完整模型的组件类型分布，再根据 `componentPage`、`connectionPage` 或 `resultVariablePage` 的 `nextOffset` 翻页。优先使用 `--variable-query` 按组件 ID、标签、物理量或单位缩小范围；只有用户点名组件时才使用 `--component` 读取该组件的完整源数据和可用的编译合同。 + +组件摘要中的 `compiledForSimulation` 表示该节点是否进入动态求解网络。介质/物性配置节点仍属于工程，因此会保留在组件总数和列表中，但通常标记为 `false`；这不表示组件丢失或编译失败。 + +曲线模式约束： + +- `separate`：每个所选结果变量分别成图； +- `overlay`：只叠加单位相同且含义可比较的结果变量； +- `stacked`：不同物理量或不同单位上下排列，避免共用一个纵轴造成误读。 + +本版运行 `simulate` 时必须指定至少一个 `--variables` 稳定键，避免在大型模型上无意生成成百上千张曲线。完整 CSV 仍包含全部可用结果变量。 + +## 权威来源 + +- 工程 JSON 请求合同：`app/main.py` 中的 `ReactFlowProjectPayload` +- 组件目录：`GET /api/components/catalog` +- XML v3：`schemas/system-simulation-v3.xsd`、`docs/standard/system-xml-v3.md` +- 接口边界：`docs/standard/backend-interface-version-spec-v1.md` +- 结果变量：组件注册合同中的 `RESULT_VARIABLES` 及仿真结果元数据 diff --git a/skills/system-simulation/references/repair-policy.md b/skills/system-simulation/references/repair-policy.md new file mode 100644 index 0000000..5616fb4 --- /dev/null +++ b/skills/system-simulation/references/repair-policy.md @@ -0,0 +1,48 @@ +# 安全文本规范化策略 + +## 目的 + +`repair-format` 只解决可解析 JSON v1 或 XML v3 的文本层问题，使文件采用稳定的 UTF-8 和跨平台文本格式。它不是模型迁移器，也不是语义修复器。 + +## 允许的修改 + +仅允许脚本已经证明不会改变解析后数据合同的规范化，例如： + +- 将可安全解码的输入统一写为 UTF-8； +- 统一 BOM 和换行表现； +- 规范化文件末尾换行； +- 对可解析内容采用脚本规定的稳定文本序列化形式。 + +以脚本返回的预览、变更摘要和哈希为准；不要在脚本外另写正则替换或自制格式化器。若文件连语法都无法可靠解析，停止并报告诊断，不能尝试猜测闭合括号、XML 标签或截断内容。 + +## 禁止的修改 + +本版不得自动执行下列动作： + +- 新增、删除、更换或重命名组件和连接； +- 修改组件 ID、类型、端口、`modelVersion` 或 Schema 版本； +- 填猜缺失参数、改变数值、单位、离散选项或介质引用； +- 修改仿真起止时间、采样间隔、最大步长或求解算法； +- 把 XML v1/v2 或旧工程升级到当前版本； +- 根据报错放宽容差、删除失败组件或改变物理拓扑； +- 覆盖源文件，即使用户给出的输出路径通过大小写、相对路径或符号链接指向源文件也不行。 + +发现上述问题时，可以解释和给出人工处理建议，但不能借“修复格式”的名义实施。 + +## 强制确认流程 + +1. 对源文件运行 `inspect`，记录格式、诊断和 SHA-256。 +2. 生成或读取 `repair-format` 的规范化预览，向用户说明只会改变哪些文本表现，并展示目标输出路径。 +3. 等待用户针对该预览明确确认。笼统的“帮我看看”或先前对其他版本的确认不能复用。 +4. 使用同一个源文件 SHA-256、预览返回的 `confirmationToken`、`--confirmed` 和预览中相同的 `--output` 路径执行写入。token 绑定源哈希、规范化输出哈希和目标绝对路径。 +5. 如果哈希、规范化结果或目标路径已变化，停止并重新预览；不能绕过 `--expected-sha256` 或确认 token。 +6. 对输出文件重新运行 `inspect`。只有重新校验通过且解析后的模型语义未改变时，才能报告完成。 + +示例命令形状见 [workflows.md](workflows.md)。 + +## 输出与交付 + +- 输出名称建议为原名加 `.normalized`，例如 `plant.normalized.json` 或 `plant.normalized.xml`。 +- 保留源文件；清楚列出新文件、源 SHA-256、输出 SHA-256 和重新校验结果。 +- 如果没有文本差异，说明文件无需规范化，不制造副本冒充修复结果。 +- 如果写入失败或输出校验失败，不能把不完整文件当作成功结果交付。 diff --git a/skills/system-simulation/references/workflows.md b/skills/system-simulation/references/workflows.md new file mode 100644 index 0000000..66feb6b --- /dev/null +++ b/skills/system-simulation/references/workflows.md @@ -0,0 +1,142 @@ +# 命令与对话工作流 + +## CLI 合同 + +从仓库根目录调用： + +```powershell +py -3.12 skills/system-simulation/scripts/simulation_skill.py [--base-url URL] [--timeout SECONDS] COMMAND +``` + +若仓库已有 `.venv-win\Scripts\python.exe`，Windows 应优先用它替换 `py -3.12`；Linux 使用 `.venv/bin/python` 或 `python3.12`。不要使用本机可能指向旧版本的裸 `python`。默认服务地址为 `http://127.0.0.1:8000`，超时参数不得低于 10 秒，以保持在后端 5 秒心跳间隔之上。`inspect`、`repair-format`、`status` 和 `cancel` 在标准输出返回一个结构化 JSON；`simulate` 在标准输出给出节流后的 JSONL 进展和精简完成摘要。输出目录的 `progress.jsonl` 保留完整进展/错误事件，但只保存精简结果摘要；完整数值结果另存为 `result.json`，避免时间序列重复占用空间和智能体上下文。 + +退出码： + +| 退出码 | 含义 | +| --- | --- | +| `0` | 命令按合同成功完成 | +| `2` | 输入、参数或安全前置条件错误 | +| `3` | HTTP、连接或后端结构化错误 | +| `4` | 仿真事件流报告失败 | +| `5` | 本地结果文件写入失败 | + +不要只看退出码 `0` 就声称仿真数值成功；还要检查最终事件和 `result.json` 中的状态。不要通过匹配本地化消息文本判断状态。 + +## 检查与解释 + +```powershell +py -3.12 skills/system-simulation/scripts/simulation_skill.py inspect INPUT --format auto +``` + +`--format` 可为 `auto`、`json` 或 `xml`。完成后按 [file-contracts.md](file-contracts.md) 解释模型。错误和警告应保留层级、稳定错误码、路径或行号；不要只复述最后一句消息。 + +默认只返回首批 50 个紧凑组件、25 条连接和 20 个结果变量，避免大型工程输出撑满上下文。翻阅模型摘要、按组件查看完整合同或搜索结果变量时使用： + +```powershell +py -3.12 skills/system-simulation/scripts/simulation_skill.py inspect INPUT ` + --component COMPONENT_ID ` + --component-offset 0 ` + --component-limit 50 ` + --connection-offset 0 ` + --connection-limit 25 ` + --variable-query QUERY ` + --variable-offset 0 ` + --variable-limit 20 +``` + +分别依据 `componentPage`、`connectionPage` 和 `resultVariablePage` 的 `hasMore`、`nextOffset` 继续分页，不要为寻找一个组件或变量请求全部详细合同。`componentTypes` 始终汇总完整模型，可先用它判断系统构成。`--component` 返回该 ID 的源文件数据和（若参与求解）编译合同，因此物性介质等配置节点也能查看参数。 + +## 文本规范化修复 + +先检查并取得源文件 SHA-256。第一次不带 `--confirmed` 调用只返回差异预览、不会写文件： + +```powershell +py -3.12 skills/system-simulation/scripts/simulation_skill.py repair-format INPUT ` + --format auto ` + --output OUTPUT ` + --expected-sha256 SHA256 +``` + +向用户展示预览中的目标路径、源/输出哈希和 `confirmationToken`，取得明确确认后，再用原样 token 执行写入： + +```powershell +py -3.12 skills/system-simulation/scripts/simulation_skill.py repair-format INPUT ` + --format auto ` + --output OUTPUT ` + --expected-sha256 SHA256 ` + --confirmation-token PREVIEW_TOKEN ` + --confirmed +``` + +token 同时绑定源文件哈希、规范化输出哈希和目标绝对路径；任何一项变化都必须重新预览和确认。脚本拒绝覆盖源文件、哈希/token 不匹配和未确认写入。完整边界见 [repair-policy.md](repair-policy.md)。写入后再次运行 `inspect OUTPUT`。 + +## 仿真前对话 + +运行前必须完成以下判断： + +1. `inspect` 通过，并取得可用结果变量清单。 +2. 用户选择 `separate`、`overlay` 或 `stacked`。 +3. 把自然语言对象解析为稳定结果 `key`；重名、缺单位或把输入参数误称为曲线时先澄清。 +4. 向用户复述将运行的文件、仿真时段、算法、所选结果变量和曲线方式。 + +本版没有网页自动预装能力。用户要求“网页查看”时，说明当前只能直接交付 SVG 曲线与 CSV；不要启动浏览器、生成临时 URL，或声称现有页面会自动载入文件。 + +## 启动并监视仿真 + +```powershell +py -3.12 skills/system-simulation/scripts/simulation_skill.py simulate INPUT ` + --format auto ` + --output-dir OUTPUT_DIR ` + --variables RESULT_KEY_1 RESULT_KEY_2 ` + --chart-mode overlay ` + --simulation-id SIMULATION_ID +``` + +`--variables` 接受结果变量稳定 `key`；不能传组件参数名。`--simulation-id` 可省略并由脚本生成，但应保存最终 ID，供恢复查询或取消使用。 + +监视规则： + +- 消费 JSONL，记录最新 `progress`、`phase`、`simulatedTime/totalTime`、心跳和内部活动快照； +- 长任务期间定期向用户给出简短进展，避免逐条转发事件； +- 仅有仿真时间平台期不能证明卡死。活动序号、RHS、solver step、Jacobian 或闭合计数仍增长时，应报告“正在处理慢步”； +- 网络读取中断后，用已知 simulation ID 查询一次任务快照，再决定是否继续说明、恢复结果或报告连接问题； +- 不因运行缓慢自动取消。只有用户明确要求取消，或既有系统已经把任务判定为 stalled 时，才使用对应取消原因； +- `completed` 才表示完整完成；`stopped`、`stalled`、`failed` 都必须标明是非完整结果。 + +当前任务状态保存在后端进程内，终态记录只短期保留，服务重启后也不能恢复。本 Skill 不承诺跨进程或长期断线续传；需要查询时应及时保存 simulation ID、事件日志和已经写出的结果文件。 + +`status` 对已完成任务只输出结果摘要，不在终端重复打印整套时间序列；正常 `simulate` 流程会把完整数据保存为 `result.json` 和 `results.csv`。 + +恢复查询： + +```powershell +py -3.12 skills/system-simulation/scripts/simulation_skill.py status SIMULATION_ID +``` + +用户要求取消时： + +```powershell +py -3.12 skills/system-simulation/scripts/simulation_skill.py cancel SIMULATION_ID --reason user +``` + +`--reason stalled` 只用于已有充分停滞证据的内部流程，不用来表达普通的用户取消。 + +## 结果文件与交付 + +正常运行目录应包含： + +- `progress.jsonl`：原始进度、心跳和结束事件； +- `result.json`：最终结构化结果； +- `results.csv`：全部可用结果变量的 UTF-8 CSV； +- 根据 `separate`、`overlay` 或 `stacked` 生成的 SVG 曲线。 + +交付时： + +1. 说明最终状态和实际计算到的仿真时间； +2. 返回用户选择的 SVG 曲线； +3. 无论用户只选了几条曲线，都同时返回完整 `results.csv`； +4. 若失败或取消但存在部分序列，明确标注曲线和 CSV 是部分结果； +5. 若没有产生可用时间序列，明确说明没有 CSV，不能创建空文件冒充结果； +6. 保留 `result.json` 和 `progress.jsonl` 作为诊断依据，但通常无需把完整事件日志逐行展示给用户。 + +本版不会根据结果自动改变模型并重试。诊断后若要改参数、拓扑或算法，先把建议交给用户，等待后续迭代能力或单独授权的人工修改流程。 diff --git a/skills/system-simulation/scripts/simulation_skill.py b/skills/system-simulation/scripts/simulation_skill.py new file mode 100644 index 0000000..6f99a6a --- /dev/null +++ b/skills/system-simulation/scripts/simulation_skill.py @@ -0,0 +1,1933 @@ +#!/usr/bin/env python3 +"""Deterministic file and simulation helper for the system-simulation skill. + +The script intentionally talks to the existing FastAPI service instead of +importing solver internals. That keeps the skill boundary small and makes the +backend the authority for component contracts, XML validation and simulation. +""" + +from __future__ import annotations + +import argparse +import codecs +import difflib +import hashlib +import html +import io +import json +import math +import os +import re +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +import uuid +import xml.etree.ElementTree as ET +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterable, Mapping, Sequence + + +MAX_INPUT_BYTES = 5 * 1024 * 1024 +DEFAULT_BASE_URL = "http://127.0.0.1:8000" +DEFAULT_TIMEOUT_SECONDS = 30.0 +MIN_TIMEOUT_SECONDS = 10.0 +ALLOWED_BACKEND_HOSTS = {"127.0.0.1", "localhost", "::1"} +SUPPORTED_FORMATS = {"json", "xml"} +SIMULATION_ID_PATTERN = re.compile(r"^[A-Za-z0-9._-]{1,128}$") +XML_FORBIDDEN_DECLARATION = re.compile(br" None: + super().__init__(message) + self.code = code + self.message = message + self.exit_code = exit_code + self.details = details + + +class InputError(SkillCliError): + def __init__(self, code: str, message: str, details: object | None = None) -> None: + super().__init__(code, message, exit_code=2, details=details) + + +class BackendError(SkillCliError): + def __init__(self, code: str, message: str, details: object | None = None) -> None: + super().__init__(code, message, exit_code=3, details=details) + + +class SimulationError(SkillCliError): + def __init__(self, code: str, message: str, details: object | None = None) -> None: + super().__init__(code, message, exit_code=4, details=details) + + +class ArtifactError(SkillCliError): + def __init__(self, code: str, message: str, details: object | None = None) -> None: + super().__init__(code, message, exit_code=5, details=details) + + +@dataclass(frozen=True) +class SourceFile: + path: Path + raw: bytes + text: str + encoding: str + had_bom: bool + sha256: str + format: str + parsed: object + + +def emit_json(payload: object, *, stream: io.TextIOBase = sys.stdout) -> None: + serialized = json.dumps(payload, ensure_ascii=False, allow_nan=False) + "\n" + try: + stream.write(serialized) + stream.flush() + except UnicodeEncodeError: + binary_stream = getattr(stream, "buffer", None) + if binary_stream is None: + raise + binary_stream.write(serialized.encode("utf-8")) + binary_stream.flush() + + +def configure_standard_streams() -> None: + """Make structured JSON reliable on Windows terminals using legacy codepages.""" + + for stream in (sys.stdout, sys.stderr): + reconfigure = getattr(stream, "reconfigure", None) + if callable(reconfigure): + reconfigure(encoding="utf-8", errors="backslashreplace") + + +def stable_error_payload(error: SkillCliError) -> dict[str, object]: + payload: dict[str, object] = { + "ok": False, + "error": { + "code": error.code, + "message": error.message, + }, + } + if error.details is not None: + payload["error"]["details"] = error.details # type: ignore[index] + return payload + + +def _sha256(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _finite_json_constant(value: str) -> None: + raise ValueError(f"JSON contains non-finite numeric constant {value}.") + + +def _unique_json_object(pairs: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + duplicates: list[str] = [] + for key, value in pairs: + if key in result: + duplicates.append(key) + result[key] = value + if duplicates: + names = ", ".join(sorted(set(duplicates))) + raise ValueError(f"JSON contains duplicate object keys: {names}.") + return result + + +def parse_json_text(text: str) -> object: + try: + return json.loads( + text, + parse_constant=_finite_json_constant, + object_pairs_hook=_unique_json_object, + ) + except (json.JSONDecodeError, ValueError) as exc: + raise InputError("JSON_SYNTAX_INVALID", str(exc)) from exc + + +def _decode_source(raw: bytes) -> tuple[str, str, bool]: + if raw.startswith(codecs.BOM_UTF8): + encoding, had_bom = "utf-8-sig", True + elif raw.startswith(codecs.BOM_UTF16_LE): + encoding, had_bom = "utf-16-le", True + elif raw.startswith(codecs.BOM_UTF16_BE): + encoding, had_bom = "utf-16-be", True + else: + encoding, had_bom = "utf-8", False + try: + text = raw.decode(encoding) + except UnicodeDecodeError as exc: + raise InputError( + "TEXT_ENCODING_UNSUPPORTED", + "Input must be UTF-8, UTF-8 with BOM, or BOM-marked UTF-16.", + {"start": exc.start, "end": exc.end}, + ) from exc + if text.startswith("\ufeff"): + text = text[1:] + return text, encoding, had_bom + + +def _content_format(text: str) -> str | None: + stripped = text.lstrip() + if stripped.startswith("{"): + return "json" + if stripped.startswith("<"): + return "xml" + return None + + +def _suffix_format(path: Path) -> str | None: + suffix = path.suffix.casefold() + if suffix == ".json": + return "json" + if suffix == ".xml": + return "xml" + return None + + +def _parse_xml(raw: bytes) -> ET.Element: + # Removing NUL bytes also exposes ASCII declarations in BOM-marked UTF-16 + # input without trying to interpret arbitrary encodings here. + if XML_FORBIDDEN_DECLARATION.search(raw.replace(b"\x00", b"")): + raise InputError( + "XML_EXTERNAL_DECLARATION_FORBIDDEN", + "DOCTYPE and ENTITY declarations are not accepted.", + ) + try: + parser = ET.XMLParser( + target=ET.TreeBuilder(insert_comments=True, insert_pis=True) + ) + return ET.fromstring(raw, parser=parser) + except ET.ParseError as exc: + details: dict[str, object] = {} + if getattr(exc, "position", None): + details["line"], details["column"] = exc.position + raise InputError("XML_SYNTAX_INVALID", str(exc), details or None) from exc + + +def load_source(path_text: str, requested_format: str = "auto") -> SourceFile: + path = Path(path_text).expanduser() + try: + resolved = path.resolve(strict=True) + except (FileNotFoundError, OSError) as exc: + raise InputError("INPUT_FILE_NOT_FOUND", f"Input file was not found: {path}") from exc + if not resolved.is_file(): + raise InputError("INPUT_NOT_A_FILE", f"Input path is not a file: {resolved}") + try: + size = resolved.stat().st_size + except OSError as exc: + raise InputError("INPUT_FILE_UNREADABLE", str(exc)) from exc + if size > MAX_INPUT_BYTES: + raise InputError( + "INPUT_FILE_TOO_LARGE", + f"Input exceeds the {MAX_INPUT_BYTES}-byte safety limit.", + {"sizeBytes": size, "maximumBytes": MAX_INPUT_BYTES}, + ) + try: + raw = resolved.read_bytes() + except OSError as exc: + raise InputError("INPUT_FILE_UNREADABLE", str(exc)) from exc + if not raw.strip(): + raise InputError("INPUT_FILE_EMPTY", "Input file is empty.") + + text, encoding, had_bom = _decode_source(raw) + detected = _content_format(text) + if detected is None: + raise InputError( + "INPUT_FORMAT_UNRECOGNIZED", + "Input content is neither a JSON object nor an XML document.", + ) + if requested_format != "auto" and requested_format != detected: + raise InputError( + "INPUT_FORMAT_MISMATCH", + f"Requested {requested_format}, but the content looks like {detected}.", + ) + + if detected == "json": + parsed = parse_json_text(text) + if not isinstance(parsed, dict): + raise InputError("JSON_ROOT_NOT_OBJECT", "Project JSON root must be an object.") + else: + parsed = _parse_xml(raw) + + return SourceFile( + path=resolved, + raw=raw, + text=text, + encoding=encoding, + had_bom=had_bom, + sha256=_sha256(raw), + format=detected, + parsed=parsed, + ) + + +def validate_base_url(value: str) -> str: + parsed = urllib.parse.urlparse(value) + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + raise InputError("BACKEND_URL_INVALID", "Backend URL must be an HTTP(S) URL.") + if parsed.username or parsed.password: + raise InputError("BACKEND_URL_CREDENTIALS_FORBIDDEN", "Credentials are not allowed in the backend URL.") + if parsed.query or parsed.fragment or parsed.params or parsed.path not in {"", "/"}: + raise InputError( + "BACKEND_URL_PATH_FORBIDDEN", + "Backend URL must contain only scheme, loopback host and optional port.", + ) + if parsed.hostname.casefold() not in ALLOWED_BACKEND_HOSTS: + raise InputError( + "REMOTE_BACKEND_FORBIDDEN", + "This base version only connects to a loopback backend.", + {"host": parsed.hostname}, + ) + return value.rstrip("/") + + +def _backend_url(base_url: str, path: str) -> str: + return f"{base_url}{path}" + + +def _decode_backend_error(data: bytes, status: int) -> object: + try: + parsed = json.loads(data.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + text = data.decode("utf-8", errors="replace").strip() + return {"status": status, "message": text[:4000]} + return {"status": status, "response": parsed} + + +def http_request( + base_url: str, + path: str, + *, + method: str = "GET", + body: bytes | None = None, + content_type: str | None = None, + timeout: float, + headers: Mapping[str, str] | None = None, +) -> tuple[bytes, Mapping[str, str]]: + request_headers = {"Accept": "application/json"} + if content_type: + request_headers["Content-Type"] = content_type + if headers: + request_headers.update(headers) + request = urllib.request.Request( + _backend_url(base_url, path), + data=body, + headers=request_headers, + method=method, + ) + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + return response.read(), dict(response.headers.items()) + except urllib.error.HTTPError as exc: + data = exc.read() + raise BackendError( + "BACKEND_HTTP_ERROR", + f"Backend returned HTTP {exc.code}.", + _decode_backend_error(data, exc.code), + ) from exc + except (urllib.error.URLError, TimeoutError, OSError) as exc: + raise BackendError("BACKEND_UNAVAILABLE", str(exc)) from exc + + +def http_json( + base_url: str, + path: str, + *, + method: str = "GET", + payload: object | None = None, + timeout: float, +) -> object: + body = None + content_type = None + if payload is not None: + try: + body = json.dumps(payload, ensure_ascii=False, allow_nan=False).encode("utf-8") + except (TypeError, ValueError) as exc: + raise InputError("REQUEST_JSON_INVALID", str(exc)) from exc + content_type = "application/json" + data, _ = http_request( + base_url, + path, + method=method, + body=body, + content_type=content_type, + timeout=timeout, + ) + try: + return json.loads(data.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise BackendError( + "BACKEND_RESPONSE_INVALID", + "Backend response was not valid UTF-8 JSON.", + ) from exc + + +def _canonical_json(value: object) -> bytes: + try: + return ( + json.dumps( + value, + ensure_ascii=False, + allow_nan=False, + indent=2, + sort_keys=False, + ) + + "\n" + ).encode("utf-8") + except (TypeError, ValueError) as exc: + raise InputError("JSON_VALUE_UNSUPPORTED", str(exc)) from exc + + +def _canonical_xml(source: SourceFile) -> bytes: + root = _parse_xml(source.raw) + ET.indent(root, space=" ") + rendered = ET.tostring( + root, + encoding="utf-8", + xml_declaration=True, + short_empty_elements=True, + ) + return rendered.rstrip(b"\r\n") + b"\n" + + +def canonical_bytes(source: SourceFile) -> bytes: + if source.format == "json": + return _canonical_json(source.parsed) + return _canonical_xml(source) + + +def _line_ending_kind(text: str) -> str: + crlf = text.count("\r\n") + without_crlf = text.replace("\r\n", "") + lf = without_crlf.count("\n") + cr = without_crlf.count("\r") + kinds = sum(count > 0 for count in (crlf, lf, cr)) + if kinds > 1: + return "mixed" + if crlf: + return "crlf" + if cr: + return "cr" + return "lf" + + +def text_normalization_report(source: SourceFile, canonical: bytes) -> dict[str, object]: + issues: list[dict[str, object]] = [] + if source.had_bom: + issues.append( + { + "code": "TEXT_BOM_PRESENT", + "message": "The normalized file uses UTF-8 without a BOM.", + "repairable": True, + } + ) + if source.encoding not in {"utf-8", "utf-8-sig"}: + issues.append( + { + "code": "TEXT_ENCODING_NOT_UTF8", + "message": f"The source encoding is {source.encoding}; normalized output uses UTF-8.", + "repairable": True, + } + ) + line_endings = _line_ending_kind(source.text) + if line_endings != "lf": + issues.append( + { + "code": "TEXT_LINE_ENDINGS_NOT_LF", + "message": f"The source uses {line_endings} line endings; normalized output uses LF.", + "repairable": True, + } + ) + if not source.text.endswith(("\n", "\r")): + issues.append( + { + "code": "TEXT_FINAL_NEWLINE_MISSING", + "message": "The normalized file ends with one LF newline.", + "repairable": True, + } + ) + if source.raw != canonical and not issues: + issues.append( + { + "code": "TEXT_LAYOUT_NOT_CANONICAL", + "message": "Whitespace or serialization layout differs from the canonical form.", + "repairable": True, + } + ) + return { + "needed": source.raw != canonical, + "issues": issues, + "canonicalSha256": _sha256(canonical), + "canonicalSizeBytes": len(canonical), + } + + +def _json_request_bytes(value: object) -> bytes: + try: + return json.dumps(value, ensure_ascii=False, allow_nan=False).encode("utf-8") + except (TypeError, ValueError) as exc: + raise InputError("JSON_VALUE_UNSUPPORTED", str(exc)) from exc + + +def _expect_mapping(value: object, *, code: str, message: str) -> dict[str, object]: + if not isinstance(value, dict): + raise BackendError(code, message, {"receivedType": type(value).__name__}) + return value + + +def _xml_inspection( + source: SourceFile, + *, + base_url: str, + timeout: float, +) -> tuple[dict[str, object], dict[str, object]]: + parsed_raw, _ = http_request( + base_url, + "/api/system-xml/parse", + method="POST", + body=source.raw, + content_type="application/xml", + timeout=timeout, + ) + compiled_raw, _ = http_request( + base_url, + "/api/system-xml/compile-model", + method="POST", + body=source.raw, + content_type="application/xml", + timeout=timeout, + ) + try: + parsed = _expect_mapping( + json.loads(parsed_raw.decode("utf-8")), + code="BACKEND_PARSE_RESPONSE_INVALID", + message="XML parse response must be an object.", + ) + compiled = _expect_mapping( + json.loads(compiled_raw.decode("utf-8")), + code="BACKEND_COMPILE_RESPONSE_INVALID", + message="XML compile response must be an object.", + ) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise BackendError( + "BACKEND_RESPONSE_INVALID", + "XML inspection response was not valid JSON.", + ) from exc + model = _expect_mapping( + parsed.get("model"), + code="BACKEND_MODEL_MISSING", + message="XML parse response did not contain a model object.", + ) + return model, compiled + + +def _json_inspection( + source: SourceFile, + *, + base_url: str, + timeout: float, +) -> tuple[dict[str, object], dict[str, object]]: + project = _expect_mapping( + source.parsed, + code="JSON_ROOT_NOT_OBJECT", + message="Project JSON root must be an object.", + ) + raw, _ = http_request( + base_url, + "/api/reactflow/compile-model", + method="POST", + body=_json_request_bytes(project), + content_type="application/json", + timeout=timeout, + ) + try: + compiled = _expect_mapping( + json.loads(raw.decode("utf-8")), + code="BACKEND_COMPILE_RESPONSE_INVALID", + message="Project compile response must be an object.", + ) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise BackendError( + "BACKEND_RESPONSE_INVALID", + "Project compile response was not valid JSON.", + ) from exc + return project, compiled + + +def _result_variables(compiled: Mapping[str, object]) -> list[dict[str, object]]: + result: list[dict[str, object]] = [] + components = compiled.get("components") + if not isinstance(components, list): + return result + for component in components: + if not isinstance(component, dict): + continue + variables = component.get("resultVariables") + if not isinstance(variables, list): + continue + for variable in variables: + if isinstance(variable, dict): + result.append(dict(variable)) + return result + + +def _collection_size(value: object) -> int: + return len(value) if isinstance(value, (dict, list)) else 0 + + +def _compiled_component_index( + compiled: Mapping[str, object], +) -> dict[str, Mapping[str, object]]: + components = compiled.get("components") + if not isinstance(components, list): + return {} + return { + str(component.get("id")): component + for component in components + if isinstance(component, dict) and component.get("id") is not None + } + + +def _compact_compiled_components( + compiled: Mapping[str, object], +) -> list[dict[str, object]]: + compact: list[dict[str, object]] = [] + for component in _compiled_component_index(compiled).values(): + variables = component.get("resultVariables") + parameters = component.get("parameters") + ports = component.get("ports") + compact.append( + { + "id": component.get("id"), + "type": component.get("type"), + "label": component.get("label"), + "modelVersion": component.get("modelVersion"), + "parameterCount": _collection_size(parameters), + "portCount": _collection_size(ports), + "resultVariableCount": _collection_size(variables), + "compiledForSimulation": True, + } + ) + return compact + + +def _compact_components( + model: Mapping[str, object], + compiled: Mapping[str, object], + *, + source_format: str, +) -> list[dict[str, object]]: + """Merge source components with compiler metadata without dropping resources. + + Medium/property nodes intentionally do not appear in the executable network, + but they are still part of the user's project and must remain visible in an + inspection summary. + """ + + compiled_by_id = _compiled_component_index(compiled) + raw_components = model.get("nodes" if source_format == "json" else "components") + if not isinstance(raw_components, list): + return _compact_compiled_components(compiled) + + compact: list[dict[str, object]] = [] + for raw_component in raw_components: + if not isinstance(raw_component, dict): + continue + if source_format == "json": + data = raw_component.get("data") + source_data = data if isinstance(data, dict) else {} + component_id = raw_component.get("id") + component_type = ( + source_data.get("componentType") + or source_data.get("modelType") + or raw_component.get("type") + ) + component_label = source_data.get("label") or raw_component.get("label") + model_version = source_data.get("modelVersion") or raw_component.get("modelVersion") + source_parameters = source_data.get("parameters") + source_ports = source_data.get("ports") + else: + source_data = raw_component + component_id = raw_component.get("id") + component_type = raw_component.get("model_type") or raw_component.get("type") + component_label = raw_component.get("label") + model_version = raw_component.get("model_version") or raw_component.get("modelVersion") + source_parameters = raw_component.get("parameters") + source_ports = raw_component.get("ports") + + compiled_component = compiled_by_id.get(str(component_id)) + compiled_parameters = ( + compiled_component.get("parameters") if compiled_component is not None else None + ) + compiled_ports = ( + compiled_component.get("ports") if compiled_component is not None else None + ) + variables = ( + compiled_component.get("resultVariables") + if compiled_component is not None + else None + ) + compact.append( + { + "id": component_id, + "type": component_type + or (compiled_component.get("type") if compiled_component is not None else None), + "label": component_label + or (compiled_component.get("label") if compiled_component is not None else None), + "modelVersion": model_version + or ( + compiled_component.get("modelVersion") + if compiled_component is not None + else None + ), + "parameterCount": _collection_size(source_parameters) + or _collection_size(compiled_parameters), + "portCount": _collection_size(source_ports) + or _collection_size(compiled_ports), + "resultVariableCount": _collection_size(variables), + "compiledForSimulation": compiled_component is not None, + } + ) + return compact + + +def _page_metadata(total: int, offset: int, limit: int) -> dict[str, object]: + returned = max(0, min(limit, total - offset)) + has_more = offset + returned < total + return { + "offset": offset, + "limit": limit, + "total": total, + "hasMore": has_more, + "nextOffset": offset + returned if has_more else None, + } + + +def _component_type_counts(components: Sequence[Mapping[str, object]]) -> list[dict[str, object]]: + counts: dict[str, int] = {} + for component in components: + model_type = str(component.get("type") or "unknown") + counts[model_type] = counts.get(model_type, 0) + 1 + return [ + {"type": model_type, "count": count} + for model_type, count in sorted(counts.items()) + ] + + +def _matches_variable_query(variable: Mapping[str, object], query: str | None) -> bool: + if not query: + return True + needle = query.casefold() + searchable = ( + variable.get("key"), + variable.get("componentId"), + variable.get("componentType"), + variable.get("portName"), + variable.get("name"), + variable.get("label"), + variable.get("quantity"), + variable.get("unit"), + ) + return any(needle in str(value).casefold() for value in searchable if value is not None) + + +def _variable_page( + variables: Sequence[dict[str, object]], + *, + query: str | None, + offset: int, + limit: int, +) -> tuple[list[dict[str, object]], dict[str, object]]: + filtered = [variable for variable in variables if _matches_variable_query(variable, query)] + items = filtered[offset : offset + limit] + return items, { + "query": query, + **_page_metadata(len(filtered), offset, limit), + } + + +def _source_component_index( + model: Mapping[str, object], + *, + source_format: str, +) -> dict[str, Mapping[str, object]]: + raw_components = model.get("nodes" if source_format == "json" else "components") + if not isinstance(raw_components, list): + return {} + return { + str(component.get("id")): component + for component in raw_components + if isinstance(component, dict) and component.get("id") is not None + } + + +def _component_details( + model: Mapping[str, object], + compiled: Mapping[str, object], + component_ids: Sequence[str], + *, + source_format: str, +) -> tuple[list[dict[str, object]], list[str]]: + if not component_ids: + return [], [] + wanted = set(component_ids) + source_by_id = _source_component_index(model, source_format=source_format) + compiled_by_id = _compiled_component_index(compiled) + details: list[dict[str, object]] = [] + found: set[str] = set() + for component_id in component_ids: + source_component = source_by_id.get(component_id) + compiled_component = compiled_by_id.get(component_id) + if source_component is None and compiled_component is None: + continue + found.add(component_id) + details.append( + { + "id": component_id, + "compiledForSimulation": compiled_component is not None, + "source": dict(source_component) if source_component is not None else None, + "compiled": dict(compiled_component) if compiled_component is not None else None, + } + ) + return details, sorted(wanted - found) + + +def inspect_source( + source: SourceFile, + *, + base_url: str, + timeout: float, + component_ids: Sequence[str] = (), + variable_query: str | None = None, + variable_offset: int = 0, + variable_limit: int = 20, + component_offset: int = 0, + component_limit: int = 50, + connection_offset: int = 0, + connection_limit: int = 25, +) -> dict[str, object]: + canonical = canonical_bytes(source) + if source.format == "json": + model, compiled = _json_inspection(source, base_url=base_url, timeout=timeout) + nodes = model.get("nodes") if isinstance(model.get("nodes"), list) else [] + edges = model.get("edges") if isinstance(model.get("edges"), list) else [] + simulation = model.get("simulation") + version: object = model.get("projectSchemaVersion") + name: object = model.get("name", "untitled") + component_count = len(nodes) + connection_count = len(edges) + else: + model, compiled = _xml_inspection(source, base_url=base_url, timeout=timeout) + components = model.get("components") if isinstance(model.get("components"), list) else [] + connections = model.get("connections") if isinstance(model.get("connections"), list) else [] + simulation = model.get("simulation") + root = source.parsed + version = root.get("schemaVersion") if isinstance(root, ET.Element) else None + name = model.get("name", "untitled") + component_count = len(components) + connection_count = len(connections) + + extension_format = _suffix_format(source.path) + source_issues: list[dict[str, object]] = [] + if extension_format is not None and extension_format != source.format: + source_issues.append( + { + "code": "FILE_EXTENSION_MISMATCH", + "message": f"File extension suggests {extension_format}, but content is {source.format}.", + "repairable": False, + } + ) + backend_validation = compiled.get("validation") + if isinstance(backend_validation, dict): + backend_issues = backend_validation.get("issues") + if isinstance(backend_issues, list): + source_issues.extend( + dict(issue) for issue in backend_issues if isinstance(issue, dict) + ) + all_variables = _result_variables(compiled) + result_variables, variable_page = _variable_page( + all_variables, + query=variable_query, + offset=variable_offset, + limit=variable_limit, + ) + component_details, missing_component_details = _component_details( + model, + compiled, + component_ids, + source_format=source.format, + ) + compact_components = _compact_components( + model, + compiled, + source_format=source.format, + ) + raw_connections = compiled.get("connections") + connections = ( + [dict(connection) for connection in raw_connections if isinstance(connection, dict)] + if isinstance(raw_connections, list) + else [] + ) + system_payload: dict[str, object] = { + "name": name, + "formatVersion": version, + "simulation": simulation, + "componentCount": component_count, + "connectionCount": connection_count, + "componentTypes": _component_type_counts(compact_components), + "components": compact_components[component_offset : component_offset + component_limit], + "componentPage": _page_metadata( + len(compact_components), component_offset, component_limit + ), + "connections": connections[connection_offset : connection_offset + connection_limit], + "connectionPage": _page_metadata( + len(connections), connection_offset, connection_limit + ), + "unconnectedPorts": compiled.get("unconnectedPorts", []), + "resultVariables": result_variables, + "resultVariablePage": variable_page, + } + if component_ids: + system_payload["componentDetails"] = component_details + system_payload["missingComponentDetails"] = missing_component_details + return { + "ok": True, + "command": "inspect", + "source": { + "path": str(source.path), + "format": source.format, + "sizeBytes": len(source.raw), + "sha256": source.sha256, + "encoding": source.encoding, + "hadBom": source.had_bom, + }, + "validation": { + "valid": True, + "issues": source_issues, + }, + "textNormalization": text_normalization_report(source, canonical), + "system": system_payload, + } + + +def command_inspect(args: argparse.Namespace) -> int: + source = load_source(args.input, args.format) + emit_json( + inspect_source( + source, + base_url=args.base_url, + timeout=args.timeout, + component_ids=args.component, + variable_query=args.variable_query, + variable_offset=args.variable_offset, + variable_limit=args.variable_limit, + component_offset=args.component_offset, + component_limit=args.component_limit, + connection_offset=args.connection_offset, + connection_limit=args.connection_limit, + ) + ) + return 0 + + +def _semantic_model( + source: SourceFile, + *, + base_url: str, + timeout: float, +) -> object: + if source.format == "json": + return source.parsed + model, _ = _xml_inspection(source, base_url=base_url, timeout=timeout) + return model + + +def _normalized_preview(source: SourceFile, canonical: bytes) -> dict[str, object]: + before = source.text.splitlines(keepends=True) + after = canonical.decode("utf-8").splitlines(keepends=True) + lines = list( + difflib.unified_diff( + before, + after, + fromfile=str(source.path), + tofile=f"{source.path.name}.normalized", + n=3, + ) + ) + truncated = len(lines) > 200 + diff_text = "".join(lines[:200]) + if len(diff_text) > 20000: + diff_text = diff_text[:20000] + truncated = True + return { + "sourceSha256": source.sha256, + "outputSha256": _sha256(canonical), + "sourceSizeBytes": len(source.raw), + "outputSizeBytes": len(canonical), + "unifiedDiff": diff_text, + "diffTruncated": truncated, + } + + +def _repair_confirmation_token( + *, + source_sha256: str, + output_sha256: str, + target: Path, +) -> str: + binding = { + "version": 1, + "sourceSha256": source_sha256, + "outputSha256": output_sha256, + "targetPath": os.path.normcase(str(target)), + } + return _sha256( + json.dumps(binding, ensure_ascii=False, sort_keys=True).encode("utf-8") + ) + + +def _ensure_distinct_new_output(source: Path, output_text: str) -> Path: + output = Path(output_text).expanduser().resolve(strict=False) + if output == source.resolve(strict=True): + raise InputError("SOURCE_OVERWRITE_FORBIDDEN", "Output must not overwrite the source file.") + if output.exists(): + raise InputError("OUTPUT_ALREADY_EXISTS", f"Output already exists: {output}") + return output + + +def _write_new_file(path: Path, data: bytes) -> None: + created = False + try: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("xb") as handle: + created = True + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + except OSError as exc: + try: + if created and path.exists(): + path.unlink() + except OSError: + pass + raise ArtifactError("OUTPUT_WRITE_FAILED", str(exc), {"path": str(path)}) from exc + + +def command_repair_format(args: argparse.Namespace) -> int: + source = load_source(args.input, args.format) + if args.expected_sha256.casefold() != source.sha256: + raise InputError( + "SOURCE_HASH_MISMATCH", + "The source changed after the repair preview; inspect it again.", + {"expected": args.expected_sha256, "actual": source.sha256}, + ) + output = _ensure_distinct_new_output(source.path, args.output) + canonical = canonical_bytes(source) + inspection = inspect_source(source, base_url=args.base_url, timeout=args.timeout) + preview = _normalized_preview(source, canonical) + confirmation_token = _repair_confirmation_token( + source_sha256=source.sha256, + output_sha256=str(preview["outputSha256"]), + target=output, + ) + preview["confirmationToken"] = confirmation_token + report: dict[str, object] = { + "ok": True, + "command": "repair-format", + "confirmed": bool(args.confirmed), + "changed": source.raw != canonical, + "source": inspection["source"], + "targetPath": str(output), + "textNormalization": inspection["textNormalization"], + "preview": preview, + } + if source.raw == canonical: + report["written"] = False + report["message"] = "The source is already in canonical text form." + emit_json(report) + return 0 + if not args.confirmed: + report["written"] = False + report["requiresConfirmation"] = True + emit_json(report) + return 0 + + if not args.confirmation_token: + raise InputError( + "REPAIR_CONFIRMATION_TOKEN_REQUIRED", + "Confirmed repair requires the confirmation token returned by its preview.", + ) + if args.confirmation_token != confirmation_token: + raise InputError( + "REPAIR_CONFIRMATION_TOKEN_MISMATCH", + "The repair preview no longer matches the source, normalized output, or target path.", + ) + + if source.format == "json": + candidate = SourceFile( + path=output, + raw=canonical, + text=canonical.decode("utf-8"), + encoding="utf-8", + had_bom=False, + sha256=_sha256(canonical), + format="json", + parsed=parse_json_text(canonical.decode("utf-8")), + ) + else: + candidate = SourceFile( + path=output, + raw=canonical, + text=canonical.decode("utf-8"), + encoding="utf-8", + had_bom=False, + sha256=_sha256(canonical), + format="xml", + parsed=_parse_xml(canonical), + ) + before_model = _semantic_model(source, base_url=args.base_url, timeout=args.timeout) + after_model = _semantic_model(candidate, base_url=args.base_url, timeout=args.timeout) + if before_model != after_model: + raise InputError( + "NORMALIZATION_CHANGED_MODEL", + "The candidate did not preserve the parsed model; no file was written.", + ) + _write_new_file(output, canonical) + report["written"] = True + report["output"] = { + "path": str(output), + "sha256": _sha256(canonical), + "sizeBytes": len(canonical), + } + emit_json(report) + return 0 + + +def _prepare_output_directory(path_text: str) -> Path: + path = Path(path_text).expanduser().resolve(strict=False) + if path.exists(): + if not path.is_dir(): + raise InputError("OUTPUT_DIR_NOT_DIRECTORY", f"Output path is not a directory: {path}") + try: + if any(path.iterdir()): + raise InputError( + "OUTPUT_DIR_NOT_EMPTY", + "Simulation output directory must be new or empty.", + {"path": str(path)}, + ) + except OSError as exc: + raise ArtifactError("OUTPUT_DIR_UNREADABLE", str(exc)) from exc + else: + try: + path.mkdir(parents=True) + except OSError as exc: + raise ArtifactError("OUTPUT_DIR_CREATE_FAILED", str(exc)) from exc + return path + + +def _project_name(inspection: Mapping[str, object]) -> str: + system = inspection.get("system") + if isinstance(system, Mapping): + name = system.get("name") + if isinstance(name, str) and name.strip(): + return name.strip() + return "simulation" + + +def _available_variables(inspection: Mapping[str, object]) -> dict[str, dict[str, object]]: + system = inspection.get("system") + if not isinstance(system, Mapping): + return {} + variables = system.get("resultVariables") + if not isinstance(variables, list): + return {} + return { + str(variable["key"]): dict(variable) + for variable in variables + if isinstance(variable, dict) and isinstance(variable.get("key"), str) + } + + +def _json_to_xml( + source: SourceFile, + *, + base_url: str, + timeout: float, +) -> bytes: + raw, _ = http_request( + base_url, + "/api/reactflow/system-xml", + method="POST", + body=_json_request_bytes(source.parsed), + content_type="application/json", + timeout=timeout, + ) + if not raw.strip().startswith(b" dict[str, object] | None: + try: + value = http_json( + base_url, + f"/api/system-xml/simulations/{urllib.parse.quote(simulation_id, safe='')}", + timeout=timeout, + ) + except BackendError: + return None + return value if isinstance(value, dict) else None + + +def _public_result_event( + event: Mapping[str, object], + simulation_id: str, + *, + full_result: str | None = "result.json", +) -> dict[str, object]: + """Return result completion metadata without echoing every time-series sample.""" + + result = event.get("result") + if not isinstance(result, Mapping): + return dict(event) + variables = result.get("variables") + series = result.get("series") + times = series.get("time") if isinstance(series, Mapping) else None + actual_end_time = result.get("actualEndTime") + if actual_end_time is None: + actual_end_time = result.get("simulatedUntil") + summary: dict[str, object] = { + "event": "result", + "simulationId": event.get("simulationId") or simulation_id, + "status": result.get("status"), + "success": result.get("success"), + "actualEndTime": actual_end_time, + "sampleCount": len(times) if isinstance(times, list) else None, + "resultVariableCount": len(variables) if isinstance(variables, list) else None, + } + if full_result is not None: + summary["fullResult"] = full_result + return summary + + +def _should_emit_progress( + event: Mapping[str, object], + *, + previous_phase: str | None, + previous_progress: float | None, + seconds_since_emit: float, +) -> bool: + phase = str(event.get("phase") or "") + raw_progress = event.get("progress") + progress = ( + float(raw_progress) + if isinstance(raw_progress, (int, float)) and not isinstance(raw_progress, bool) + else None + ) + return ( + previous_phase is None + or phase != previous_phase + or (progress is not None and previous_progress is not None and progress >= previous_progress + 5.0) + or progress == 100.0 + or seconds_since_emit >= 5.0 + ) + + +def _read_simulation_stream( + base_url: str, + xml: bytes, + simulation_id: str, + timeout: float, + progress_path: Path, +) -> tuple[dict[str, object] | None, dict[str, object] | None]: + final_result: dict[str, object] | None = None + final_error: dict[str, object] | None = None + public_phase: str | None = None + public_progress: float | None = None + public_emit_time = 0.0 + try: + progress = progress_path.open("x", encoding="utf-8", newline="\n") + except OSError as exc: + raise ArtifactError( + "PROGRESS_LOG_CREATE_FAILED", + str(exc), + {"path": str(progress_path)}, + ) from exc + try: + response = _open_simulation_stream(base_url, xml, simulation_id, timeout) + with response, progress: + for raw_line in response: + if not raw_line.strip(): + continue + try: + event = json.loads(raw_line.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise SimulationError( + "SIMULATION_EVENT_INVALID", + "Simulation stream returned an invalid JSONL event.", + ) from exc + if not isinstance(event, dict) or event.get("event") not in { + "progress", + "result", + "error", + }: + raise SimulationError( + "SIMULATION_EVENT_INVALID", + "Simulation stream returned an event outside the contract.", + event, + ) + event_kind = event.get("event") + logged_event = ( + _public_result_event(event, simulation_id) + if event_kind == "result" + else event + ) + serialized = json.dumps(logged_event, ensure_ascii=False, allow_nan=False) + progress.write(serialized + "\n") + progress.flush() + if event_kind == "progress": + now = time.monotonic() + if _should_emit_progress( + event, + previous_phase=public_phase, + previous_progress=public_progress, + seconds_since_emit=now - public_emit_time, + ): + emit_json(event) + public_phase = str(event.get("phase") or "") + raw_progress = event.get("progress") + if isinstance(raw_progress, (int, float)) and not isinstance(raw_progress, bool): + public_progress = float(raw_progress) + public_emit_time = now + elif event_kind == "result": + emit_json(logged_event) + else: + emit_json(event) + if event_kind == "result": + result = event.get("result") + if isinstance(result, dict): + final_result = result + else: + final_error = { + "code": "SIMULATION_RESULT_MISSING", + "message": "Result event did not contain a result object.", + } + elif event_kind == "error": + final_error = event + except SkillCliError: + progress.close() + raise + except (urllib.error.URLError, TimeoutError, OSError) as exc: + snapshot = _recover_task_snapshot(base_url, simulation_id, timeout) + if snapshot: + result = snapshot.get("result") + error = snapshot.get("error") + if isinstance(result, dict): + return result, error if isinstance(error, dict) else None + if isinstance(error, dict): + return None, error + raise BackendError( + "SIMULATION_STREAM_INTERRUPTED", + str(exc), + {"simulationId": simulation_id, "task": snapshot}, + ) from exc + return final_result, final_error + + +def _validate_result_shape(result: Mapping[str, object]) -> tuple[list[dict[str, object]], dict[str, list[float]]]: + raw_variables = result.get("variables") + raw_series = result.get("series") + if not isinstance(raw_variables, list) or not isinstance(raw_series, dict): + raise SimulationError( + "SIMULATION_RESULT_INVALID", + "Simulation result does not contain variables and series.", + ) + variables = [dict(value) for value in raw_variables if isinstance(value, dict)] + series: dict[str, list[float]] = {} + for key, values in raw_series.items(): + if not isinstance(key, str) or not isinstance(values, list): + raise SimulationError( + "SIMULATION_RESULT_INVALID", + "Simulation series must map string keys to numeric arrays.", + ) + converted: list[float] = [] + for value in values: + if not isinstance(value, (int, float)) or isinstance(value, bool) or not math.isfinite(float(value)): + raise SimulationError( + "SIMULATION_RESULT_NONFINITE", + f"Series {key} contains a non-finite or non-numeric value.", + ) + converted.append(float(value)) + series[key] = converted + times = series.get("time") + if not times: + raise SimulationError("SIMULATION_RESULT_HAS_NO_TIME", "Simulation produced no usable time series.") + return variables, series + + +def _variable_metadata_by_key(variables: Sequence[Mapping[str, object]]) -> dict[str, dict[str, object]]: + return { + str(variable["key"]): dict(variable) + for variable in variables + if isinstance(variable.get("key"), str) + } + + +def _write_result_json(path: Path, result: Mapping[str, object]) -> None: + try: + data = (json.dumps(result, ensure_ascii=False, allow_nan=False, indent=2) + "\n").encode("utf-8") + except (TypeError, ValueError) as exc: + raise SimulationError("SIMULATION_RESULT_NOT_JSON", str(exc)) from exc + _write_new_file(path, data) + + +def _download_csv( + base_url: str, + timeout: float, + project_name: str, + variables: Sequence[Mapping[str, object]], + series: Mapping[str, Sequence[float]], +) -> bytes: + raw, _ = http_request( + base_url, + "/api/simulation-results/csv", + method="POST", + body=_json_request_bytes( + { + "projectName": project_name, + "variables": list(variables), + "series": dict(series), + } + ), + content_type="application/json", + timeout=timeout, + ) + return raw + + +def _safe_stem(value: str) -> str: + value = re.sub(r"[^A-Za-z0-9._-]+", "_", value).strip("._") + return value[:80] or "curve" + + +def _decimated_points(times: Sequence[float], values: Sequence[float], maximum: int = 4000) -> list[tuple[float, float]]: + count = min(len(times), len(values)) + if count <= maximum: + indices = range(count) + else: + indices = sorted({round(index * (count - 1) / (maximum - 1)) for index in range(maximum)}) + return [(float(times[index]), float(values[index])) for index in indices] + + +def _domain(values: Iterable[float]) -> tuple[float, float]: + items = list(values) + minimum = min(items) + maximum = max(items) + if minimum == maximum: + padding = max(1.0, abs(minimum) * 0.05) + return minimum - padding, maximum + padding + padding = (maximum - minimum) * 0.05 + return minimum - padding, maximum + padding + + +def _polyline( + points: Sequence[tuple[float, float]], + *, + x_min: float, + x_max: float, + y_min: float, + y_max: float, + left: float, + top: float, + width: float, + height: float, +) -> str: + x_span = x_max - x_min or 1.0 + y_span = y_max - y_min or 1.0 + return " ".join( + f"{left + (x - x_min) * width / x_span:.2f},{top + height - (y - y_min) * height / y_span:.2f}" + for x, y in points + ) + + +def _variable_label(variable: Mapping[str, object]) -> str: + component = str(variable.get("componentId", "component")) + port = variable.get("portName") + label = str(variable.get("label") or variable.get("name") or variable.get("key")) + return f"{component}.{port}.{label}" if port else f"{component}.{label}" + + +def _svg_document( + times: Sequence[float], + selected: Sequence[tuple[Mapping[str, object], Sequence[float]]], + *, + stacked: bool, +) -> str: + width = 1000 + panel_height = 280 + height = panel_height * len(selected) if stacked else 540 + left, right = 82.0, 32.0 + plot_width = width - left - right + x_min, x_max = _domain(times) + panels = list(enumerate(selected)) if stacked else [(0, selected[0])] + parts = [ + '', + f'', + '', + '', + ] + if stacked: + for panel_index, (variable, values) in panels: + top = panel_index * panel_height + 34.0 + plot_height = panel_height - 78.0 + y_min, y_max = _domain(values) + parts.extend(_svg_panel_axes(left, top, plot_width, plot_height, x_min, x_max, y_min, y_max)) + points = _decimated_points(times, values) + coords = _polyline(points, x_min=x_min, x_max=x_max, y_min=y_min, y_max=y_max, left=left, top=top, width=plot_width, height=plot_height) + label = html.escape(_variable_label(variable)) + unit = html.escape(str(variable.get("unit", ""))) + parts.append(f'{label} [{unit}]') + parts.append(f'') + else: + top = 60.0 + plot_height = height - 130.0 + all_values = [value for _, values in selected for value in values] + y_min, y_max = _domain(all_values) + parts.extend(_svg_panel_axes(left, top, plot_width, plot_height, x_min, x_max, y_min, y_max)) + for index, (variable, values) in enumerate(selected): + coords = _polyline(_decimated_points(times, values), x_min=x_min, x_max=x_max, y_min=y_min, y_max=y_max, left=left, top=top, width=plot_width, height=plot_height) + color = CHART_COLORS[index % len(CHART_COLORS)] + parts.append(f'') + legend_y = 23 + index * 20 + label = html.escape(_variable_label(variable)) + parts.append(f'') + parts.append(f'{label}') + parts.append('') + return "\n".join(parts) + "\n" + + +def _svg_panel_axes( + left: float, + top: float, + width: float, + height: float, + x_min: float, + x_max: float, + y_min: float, + y_max: float, +) -> list[str]: + parts: list[str] = [] + for index in range(6): + x = left + width * index / 5 + y = top + height * index / 5 + x_value = x_min + (x_max - x_min) * index / 5 + y_value = y_max - (y_max - y_min) * index / 5 + parts.append(f'') + parts.append(f'') + parts.append(f'{x_value:.5g}') + parts.append(f'{y_value:.5g}') + parts.append(f'') + parts.append(f'') + parts.append(f'time [s]') + return parts + + +def _write_charts( + output_dir: Path, + mode: str, + selected_keys: Sequence[str], + variables: Mapping[str, Mapping[str, object]], + series: Mapping[str, Sequence[float]], +) -> list[str]: + times = series["time"] + selected = [(variables[key], series[key]) for key in selected_keys] + artifacts: list[str] = [] + if mode == "overlay": + units = {str(variable.get("unit", "")) for variable, _ in selected} + if len(units) > 1: + raise InputError( + "OVERLAY_UNITS_MISMATCH", + "Overlay requires all selected result variables to use the same unit; use stacked instead.", + {"units": sorted(units)}, + ) + path = output_dir / "curves-overlay.svg" + _write_new_file(path, _svg_document(times, selected, stacked=False).encode("utf-8")) + artifacts.append(str(path)) + elif mode == "stacked": + path = output_dir / "curves-stacked.svg" + _write_new_file(path, _svg_document(times, selected, stacked=True).encode("utf-8")) + artifacts.append(str(path)) + else: + for variable, values in selected: + key = str(variable.get("key", "curve")) + key_digest = hashlib.sha256(key.encode("utf-8")).hexdigest()[:8] + path = output_dir / f"curve-{_safe_stem(key)}-{key_digest}.svg" + _write_new_file(path, _svg_document(times, [(variable, values)], stacked=False).encode("utf-8")) + artifacts.append(str(path)) + return artifacts + + +def command_simulate(args: argparse.Namespace) -> int: + source = load_source(args.input, args.format) + inspection = inspect_source( + source, + base_url=args.base_url, + timeout=args.timeout, + variable_limit=1_000_000, + ) + available = _available_variables(inspection) + requested = list(dict.fromkeys(args.variables)) + missing = [key for key in requested if key not in available] + if missing: + raise InputError( + "RESULT_VARIABLE_UNKNOWN", + "One or more requested result keys are not declared by the compiled model.", + {"missing": missing, "available": sorted(available)}, + ) + if args.chart_mode == "overlay": + units = {str(available[key].get("unit", "")) for key in requested} + if len(units) > 1: + raise InputError( + "OVERLAY_UNITS_MISMATCH", + "Overlay requires result variables with the same unit; choose stacked instead.", + {"units": sorted(units)}, + ) + simulation_id = args.simulation_id or uuid.uuid4().hex + if not SIMULATION_ID_PATTERN.fullmatch(simulation_id): + raise InputError("SIMULATION_ID_INVALID", "Simulation ID must match [A-Za-z0-9._-]{1,128}.") + output_dir = _prepare_output_directory(args.output_dir) + xml = source.raw if source.format == "xml" else _json_to_xml(source, base_url=args.base_url, timeout=args.timeout) + emit_json( + { + "event": "submitted", + "simulationId": simulation_id, + "input": str(source.path), + "selectedVariables": requested, + "chartMode": args.chart_mode, + "outputDirectory": str(output_dir), + } + ) + result, error = _read_simulation_stream( + args.base_url, + xml, + simulation_id, + args.timeout, + output_dir / "progress.jsonl", + ) + if result is None: + raise SimulationError( + "SIMULATION_FAILED_WITHOUT_RESULT", + "Simulation ended without a usable result.", + {"simulationId": simulation_id, "backendError": error}, + ) + result_path = output_dir / "result.json" + _write_result_json(result_path, result) + variables, series = _validate_result_shape(result) + metadata = _variable_metadata_by_key(variables) + missing_after_run = [key for key in requested if key not in metadata or key not in series] + if missing_after_run: + raise SimulationError( + "RESULT_VARIABLE_MISSING", + "Requested variables were declared before the run but are missing from the result.", + {"missing": missing_after_run}, + ) + + csv_path = output_dir / "results.csv" + csv_bytes = _download_csv( + args.base_url, + args.timeout, + _project_name(inspection), + variables, + series, + ) + _write_new_file(csv_path, csv_bytes) + chart_paths = _write_charts( + output_dir, + args.chart_mode, + requested, + metadata, + series, + ) + status = str(result.get("status") or ("completed" if result.get("success") else "failed")) + artifact_event = { + "event": "artifacts", + "simulationId": simulation_id, + "status": status, + "partial": status != "completed", + "result": str(result_path), + "csv": str(csv_path), + "charts": chart_paths, + "progress": str(output_dir / "progress.jsonl"), + "selectedVariables": requested, + } + emit_json(artifact_event) + return 0 if status == "completed" and bool(result.get("success")) else 4 + + +def _public_task_snapshot(task: Mapping[str, object]) -> dict[str, object]: + snapshot = {key: value for key, value in task.items() if key != "result"} + result = task.get("result") + if isinstance(result, Mapping): + simulation_id = str(task.get("simulationId") or "unknown") + snapshot["resultSummary"] = _public_result_event( + {"event": "result", "result": result}, + simulation_id, + full_result=None, + ) + return snapshot + + +def command_status(args: argparse.Namespace) -> int: + simulation_id = args.simulation_id + if not SIMULATION_ID_PATTERN.fullmatch(simulation_id): + raise InputError("SIMULATION_ID_INVALID", "Simulation ID must match [A-Za-z0-9._-]{1,128}.") + result = http_json( + args.base_url, + f"/api/system-xml/simulations/{urllib.parse.quote(simulation_id, safe='')}", + timeout=args.timeout, + ) + task = _expect_mapping( + result, + code="BACKEND_TASK_RESPONSE_INVALID", + message="Simulation task response must be an object.", + ) + emit_json({"ok": True, "command": "status", "task": _public_task_snapshot(task)}) + return 0 + + +def command_cancel(args: argparse.Namespace) -> int: + simulation_id = args.simulation_id + if not SIMULATION_ID_PATTERN.fullmatch(simulation_id): + raise InputError("SIMULATION_ID_INVALID", "Simulation ID must match [A-Za-z0-9._-]{1,128}.") + result = http_json( + args.base_url, + f"/api/system-xml/simulations/{urllib.parse.quote(simulation_id, safe='')}/cancel", + method="POST", + payload={"reason": args.reason}, + timeout=args.timeout, + ) + emit_json({"ok": True, "command": "cancel", "task": result}) + return 0 + + +def positive_timeout(value: str) -> float: + try: + parsed = float(value) + except ValueError as exc: + raise argparse.ArgumentTypeError("timeout must be numeric") from exc + if not math.isfinite(parsed) or parsed < MIN_TIMEOUT_SECONDS: + raise argparse.ArgumentTypeError( + f"timeout must be finite and at least {MIN_TIMEOUT_SECONDS:g} seconds " + "so it remains above the backend heartbeat interval" + ) + return parsed + + +def nonnegative_integer(value: str) -> int: + try: + parsed = int(value) + except ValueError as exc: + raise argparse.ArgumentTypeError("value must be an integer") from exc + if parsed < 0: + raise argparse.ArgumentTypeError("value must be zero or greater") + return parsed + + +def page_limit(value: str) -> int: + parsed = nonnegative_integer(value) + if parsed < 1 or parsed > 200: + raise argparse.ArgumentTypeError("page limit must be between 1 and 200") + return parsed + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Inspect, safely normalize and simulate SystemSimulationApp JSON/XML files." + ) + parser.add_argument( + "--base-url", + default=DEFAULT_BASE_URL, + help="Loopback FastAPI base URL (default: %(default)s).", + ) + parser.add_argument( + "--timeout", + default=DEFAULT_TIMEOUT_SECONDS, + type=positive_timeout, + help="HTTP/heartbeat read timeout in seconds (default: %(default)s).", + ) + subparsers = parser.add_subparsers(dest="command", required=True) + + inspect_parser = subparsers.add_parser("inspect", help="Validate and summarize a project file.") + inspect_parser.add_argument("input") + inspect_parser.add_argument("--format", choices=("auto", "json", "xml"), default="auto") + inspect_parser.add_argument( + "--component", + action="append", + default=[], + help="Include full compiled details for this component ID (repeatable).", + ) + inspect_parser.add_argument( + "--variable-query", + help="Filter result variables by key, component, label, quantity or unit.", + ) + inspect_parser.add_argument( + "--variable-offset", + type=nonnegative_integer, + default=0, + help="Result-variable page offset (default: %(default)s).", + ) + inspect_parser.add_argument( + "--variable-limit", + type=page_limit, + default=20, + help="Result-variable page size, 1-200 (default: %(default)s).", + ) + inspect_parser.add_argument( + "--component-offset", + type=nonnegative_integer, + default=0, + help="Component-summary page offset (default: %(default)s).", + ) + inspect_parser.add_argument( + "--component-limit", + type=page_limit, + default=50, + help="Component-summary page size, 1-200 (default: %(default)s).", + ) + inspect_parser.add_argument( + "--connection-offset", + type=nonnegative_integer, + default=0, + help="Connection page offset (default: %(default)s).", + ) + inspect_parser.add_argument( + "--connection-limit", + type=page_limit, + default=25, + help="Connection page size, 1-200 (default: %(default)s).", + ) + inspect_parser.set_defaults(handler=command_inspect) + + repair_parser = subparsers.add_parser( + "repair-format", + help="Preview or write a semantics-preserving UTF-8/LF normalization.", + ) + repair_parser.add_argument("input") + repair_parser.add_argument("--format", choices=("auto", "json", "xml"), default="auto") + repair_parser.add_argument("--output", required=True) + repair_parser.add_argument("--expected-sha256", required=True) + repair_parser.add_argument( + "--confirmation-token", + help="Token from the matching preview; required together with --confirmed.", + ) + repair_parser.add_argument( + "--confirmed", + action="store_true", + help="Write the new file. Without this flag, only return a diff preview.", + ) + repair_parser.set_defaults(handler=command_repair_format) + + simulate_parser = subparsers.add_parser("simulate", help="Run and monitor a simulation.") + simulate_parser.add_argument("input") + simulate_parser.add_argument("--format", choices=("auto", "json", "xml"), default="auto") + simulate_parser.add_argument("--output-dir", required=True) + simulate_parser.add_argument("--variables", nargs="+", required=True) + simulate_parser.add_argument( + "--chart-mode", + choices=("separate", "overlay", "stacked"), + required=True, + ) + simulate_parser.add_argument("--simulation-id") + simulate_parser.set_defaults(handler=command_simulate) + + status_parser = subparsers.add_parser("status", help="Read a known simulation task.") + status_parser.add_argument("simulation_id") + status_parser.set_defaults(handler=command_status) + + cancel_parser = subparsers.add_parser("cancel", help="Cancel a known simulation task.") + cancel_parser.add_argument("simulation_id") + cancel_parser.add_argument("--reason", choices=("user", "stalled"), default="user") + cancel_parser.set_defaults(handler=command_cancel) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + configure_standard_streams() + parser = build_parser() + args = parser.parse_args(argv) + try: + args.base_url = validate_base_url(args.base_url) + return int(args.handler(args)) + except SkillCliError as exc: + emit_json(stable_error_payload(exc), stream=sys.stderr) + return exc.exit_code + except KeyboardInterrupt: + emit_json( + stable_error_payload( + SimulationError( + "CLIENT_INTERRUPTED", + "The local monitor was interrupted. The backend task may still be running; query its simulation ID.", + ) + ), + stream=sys.stderr, + ) + return 4 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_system_simulation_skill.py b/tests/test_system_simulation_skill.py new file mode 100644 index 0000000..36b475e --- /dev/null +++ b/tests/test_system_simulation_skill.py @@ -0,0 +1,1068 @@ +"""Standard-library tests for the system-simulation skill helper. + +These tests deliberately mock only the HTTP transport boundary. Parsing, +normalization, simulation event monitoring, artifact creation, and command +validation therefore run through the same code used by the skill. +""" + +from __future__ import annotations + +import argparse +import importlib.util +import io +import json +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +SCRIPT_PATH = ( + REPOSITORY_ROOT + / "skills" + / "system-simulation" + / "scripts" + / "simulation_skill.py" +) +MODULE_NAME = "system_simulation_skill_under_test" +MODULE_SPEC = importlib.util.spec_from_file_location(MODULE_NAME, SCRIPT_PATH) +if MODULE_SPEC is None or MODULE_SPEC.loader is None: # pragma: no cover + raise RuntimeError(f"Cannot load skill helper from {SCRIPT_PATH}") +skill = importlib.util.module_from_spec(MODULE_SPEC) +sys.modules[MODULE_NAME] = skill +MODULE_SPEC.loader.exec_module(skill) + + +BASE_URL = "http://127.0.0.1:8000" +RESULT_VARIABLES = [ + { + "key": "mass.position", + "componentId": "mass", + "name": "position", + "label": "Position", + "quantity": "displacement", + "unit": "m", + }, + { + "key": "mass.velocity", + "componentId": "mass", + "name": "velocity", + "label": "Velocity", + "quantity": "velocity", + "unit": "m/s", + }, +] +COMPILED_MODEL = { + "validation": {"valid": True, "issues": []}, + "components": [ + { + "id": "mass", + "type": "MECMAS21", + "resultVariables": RESULT_VARIABLES, + } + ], + "connections": [{"id": "connection-1"}], + "unconnectedPorts": [], +} +PROJECT = { + "projectSchemaVersion": 1, + "name": "json-system", + "nodes": [ + {"id": "mass", "type": "component", "data": {"componentType": "MECMAS21"}} + ], + "edges": [{"id": "connection-1", "source": "mass", "target": "mass"}], + "simulation": {"startTime": 0.0, "endTime": 1.0}, +} +SYSTEM_XML = b""" + + + +""" + + +def json_bytes(value: object) -> bytes: + return json.dumps(value, ensure_ascii=False).encode("utf-8") + + +def compile_http_response( + _base_url: str, + path: str, + **_kwargs: object, +) -> tuple[bytes, dict[str, str]]: + if path == "/api/reactflow/compile-model": + return json_bytes(COMPILED_MODEL), {} + raise AssertionError(f"Unexpected HTTP request: {path}") + + +class FileRecognitionTests(unittest.TestCase): + def test_detects_content_format_independently_of_extension(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "project.xml" + path.write_text(json.dumps(PROJECT), encoding="utf-8") + + source = skill.load_source(str(path), "auto") + + self.assertEqual(source.format, "json") + self.assertEqual(source.parsed, PROJECT) + with self.assertRaises(skill.InputError) as caught: + skill.load_source(str(path), "xml") + self.assertEqual(caught.exception.code, "INPUT_FORMAT_MISMATCH") + + def test_rejects_duplicate_json_keys(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "duplicate.json" + path.write_text('{"name":"first","name":"second"}', encoding="utf-8") + + with self.assertRaises(skill.InputError) as caught: + skill.load_source(str(path)) + + self.assertEqual(caught.exception.code, "JSON_SYNTAX_INVALID") + self.assertIn("duplicate", caught.exception.message.lower()) + + def test_rejects_xml_doctype_before_parsing(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "unsafe.xml" + path.write_bytes( + b']>' + b"&xxe;" + ) + + with self.assertRaises(skill.InputError) as caught: + skill.load_source(str(path)) + + self.assertEqual( + caught.exception.code, + "XML_EXTERNAL_DECLARATION_FORBIDDEN", + ) + + def test_rejects_non_loopback_backend_url(self) -> None: + with self.assertRaises(skill.InputError) as caught: + skill.validate_base_url("https://simulation.example.com") + + self.assertEqual(caught.exception.code, "REMOTE_BACKEND_FORBIDDEN") + self.assertEqual(skill.validate_base_url("http://localhost:8000/"), "http://localhost:8000") + + def test_timeout_must_remain_above_backend_heartbeat(self) -> None: + with self.assertRaises(argparse.ArgumentTypeError): + skill.positive_timeout("5") + self.assertEqual(skill.positive_timeout("10"), 10.0) + + def test_inspect_parser_exposes_bounded_paging_defaults(self) -> None: + args = skill.build_parser().parse_args(["inspect", "model.xml"]) + + self.assertEqual(args.variable_limit, 20) + self.assertEqual(args.component_limit, 50) + self.assertEqual(args.connection_limit, 25) + with mock.patch("sys.stderr", new=io.StringIO()), self.assertRaises(SystemExit): + skill.build_parser().parse_args( + ["inspect", "model.xml", "--component-limit", "201"] + ) + + def test_json_output_falls_back_to_utf8_on_legacy_console(self) -> None: + class LegacyConsole: + def __init__(self) -> None: + self.buffer = io.BytesIO() + + def write(self, value: str) -> None: + raise UnicodeEncodeError("gbk", value, 0, 1, "unsupported") + + def flush(self) -> None: + return None + + console = LegacyConsole() + skill.emit_json({"unit": "kg/m³"}, stream=console) + self.assertEqual( + json.loads(console.buffer.getvalue().decode("utf-8")), + {"unit": "kg/m³"}, + ) + + +class InspectionTests(unittest.TestCase): + def test_inspect_json_returns_system_summary(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "project.json" + path.write_text(json.dumps(PROJECT), encoding="utf-8") + source = skill.load_source(str(path)) + + with mock.patch.object( + skill, + "http_request", + side_effect=compile_http_response, + ): + report = skill.inspect_source(source, base_url=BASE_URL, timeout=1.0) + + self.assertTrue(report["ok"]) + self.assertEqual(report["source"]["format"], "json") + self.assertEqual(report["system"]["name"], "json-system") + self.assertEqual(report["system"]["formatVersion"], 1) + self.assertEqual(report["system"]["componentCount"], 1) + self.assertEqual(report["system"]["connectionCount"], 1) + self.assertEqual( + [variable["key"] for variable in report["system"]["resultVariables"]], + ["mass.position", "mass.velocity"], + ) + + def test_inspect_xml_returns_backend_and_document_summary(self) -> None: + parsed_model = { + "name": "xml-system", + "components": [{"id": "mass"}, {"id": "fixed"}], + "connections": [{"id": "connection-1"}], + "simulation": {"startTime": 0.0, "endTime": 1.0}, + } + + def xml_http_response( + _base_url: str, + path: str, + **_kwargs: object, + ) -> tuple[bytes, dict[str, str]]: + if path == "/api/system-xml/parse": + return json_bytes({"model": parsed_model}), {} + if path == "/api/system-xml/compile-model": + return json_bytes(COMPILED_MODEL), {} + raise AssertionError(f"Unexpected HTTP request: {path}") + + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "system.xml" + path.write_bytes(SYSTEM_XML) + source = skill.load_source(str(path)) + + with mock.patch.object( + skill, + "http_request", + side_effect=xml_http_response, + ) as request: + report = skill.inspect_source(source, base_url=BASE_URL, timeout=1.0) + + self.assertEqual(request.call_count, 2) + self.assertEqual(report["source"]["format"], "xml") + self.assertEqual(report["system"]["name"], "xml-system") + self.assertEqual(report["system"]["formatVersion"], "3") + self.assertEqual(report["system"]["componentCount"], 2) + self.assertEqual(report["system"]["connectionCount"], 1) + + def test_inspect_pages_result_variables_and_keeps_components_compact(self) -> None: + variables = [ + { + "key": f"mass.value_{index}", + "componentId": "mass", + "label": f"Value {index}", + "quantity": "dimensionless", + "unit": "", + } + for index in range(75) + ] + compiled = { + **COMPILED_MODEL, + "components": [ + { + "id": "mass", + "type": "MECMAS21", + "parameters": [{"name": "mass", "value": 1.0}], + "ports": [{"name": "port_1"}], + "resultVariables": variables, + } + ], + } + + def response(_base_url: str, path: str, **_kwargs: object): + self.assertEqual(path, "/api/reactflow/compile-model") + return json_bytes(compiled), {} + + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "project.json" + path.write_text(json.dumps(PROJECT), encoding="utf-8") + source = skill.load_source(str(path)) + with mock.patch.object(skill, "http_request", side_effect=response): + report = skill.inspect_source(source, base_url=BASE_URL, timeout=10.0) + + self.assertEqual(len(report["system"]["resultVariables"]), 20) + self.assertTrue(report["system"]["resultVariablePage"]["hasMore"]) + self.assertEqual(report["system"]["resultVariablePage"]["nextOffset"], 20) + self.assertEqual(report["system"]["componentPage"]["total"], 1) + self.assertEqual(report["system"]["connectionPage"]["total"], 1) + self.assertNotIn("parameters", report["system"]["components"][0]) + + def test_inspect_keeps_source_only_medium_component_visible(self) -> None: + project = { + **PROJECT, + "nodes": [ + { + "id": "medium", + "type": "simulationComponent", + "data": { + "label": "Helium properties", + "componentType": "amesim_helium_medium", + "modelVersion": "0.1.0", + "parameters": {"property_model": 0}, + "ports": [], + }, + }, + *PROJECT["nodes"], + ], + } + + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "project.json" + path.write_text(json.dumps(project), encoding="utf-8") + source = skill.load_source(str(path)) + with mock.patch.object( + skill, + "http_request", + side_effect=compile_http_response, + ): + report = skill.inspect_source( + source, + base_url=BASE_URL, + timeout=10.0, + component_ids=("medium",), + ) + + self.assertEqual(report["system"]["componentCount"], 2) + self.assertEqual(report["system"]["componentPage"]["total"], 2) + self.assertEqual(report["system"]["components"][0]["id"], "medium") + self.assertEqual(report["system"]["components"][0]["label"], "Helium properties") + self.assertEqual(report["system"]["components"][0]["modelVersion"], "0.1.0") + self.assertFalse(report["system"]["components"][0]["compiledForSimulation"]) + self.assertEqual(report["system"]["missingComponentDetails"], []) + medium_detail = report["system"]["componentDetails"][0] + self.assertFalse(medium_detail["compiledForSimulation"]) + self.assertEqual(medium_detail["source"]["data"]["parameters"], {"property_model": 0}) + self.assertIsNone(medium_detail["compiled"]) + + def test_variable_query_is_applied_before_nonzero_final_page(self) -> None: + matching = [ + { + "key": f"mass.speed_{index}", + "componentId": "mass", + "label": f"Speed {index}", + "quantity": "velocity", + "unit": "m/s", + } + for index in range(5) + ] + unrelated = [ + { + "key": f"mass.position_{index}", + "componentId": "mass", + "label": f"Position {index}", + "quantity": "displacement", + "unit": "m", + } + for index in range(8) + ] + compiled = { + **COMPILED_MODEL, + "components": [ + { + "id": "mass", + "type": "MECMAS21", + "resultVariables": [*unrelated, *matching], + } + ], + } + + def response(_base_url: str, path: str, **_kwargs: object): + self.assertEqual(path, "/api/reactflow/compile-model") + return json_bytes(compiled), {} + + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "project.json" + path.write_text(json.dumps(PROJECT), encoding="utf-8") + source = skill.load_source(str(path)) + with mock.patch.object(skill, "http_request", side_effect=response): + report = skill.inspect_source( + source, + base_url=BASE_URL, + timeout=10.0, + variable_query="speed", + variable_offset=3, + variable_limit=2, + ) + + page = report["system"]["resultVariablePage"] + self.assertEqual( + [variable["key"] for variable in report["system"]["resultVariables"]], + ["mass.speed_3", "mass.speed_4"], + ) + self.assertEqual(page["query"], "speed") + self.assertEqual(page["offset"], 3) + self.assertEqual(page["total"], 5) + self.assertFalse(page["hasMore"]) + self.assertIsNone(page["nextOffset"]) + + def test_xml_source_only_medium_survives_on_final_component_page(self) -> None: + parsed_model = { + "name": "xml-medium-system", + "components": [ + {"id": "mass", "model_type": "amesim_mecmas21"}, + {"id": "fixed", "model_type": "amesim_fixed"}, + { + "id": "medium", + "model_type": "amesim_helium_medium", + "parameters": {"property_model": 0}, + }, + ], + "connections": [], + "simulation": {"t_start": 0.0, "t_stop": 1.0}, + } + compiled = { + **COMPILED_MODEL, + "components": [ + {"id": "mass", "type": "amesim_mecmas21", "resultVariables": []}, + {"id": "fixed", "type": "amesim_fixed", "resultVariables": []}, + ], + "connections": [], + } + + def response( + _base_url: str, + path: str, + **_kwargs: object, + ) -> tuple[bytes, dict[str, str]]: + if path == "/api/system-xml/parse": + return json_bytes({"model": parsed_model}), {} + if path == "/api/system-xml/compile-model": + return json_bytes(compiled), {} + raise AssertionError(f"Unexpected HTTP request: {path}") + + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "system.xml" + path.write_bytes(SYSTEM_XML) + source = skill.load_source(str(path)) + with mock.patch.object(skill, "http_request", side_effect=response): + report = skill.inspect_source( + source, + base_url=BASE_URL, + timeout=10.0, + component_offset=2, + component_limit=1, + ) + + system = report["system"] + self.assertEqual(system["componentPage"]["total"], 3) + self.assertEqual(system["componentPage"]["offset"], 2) + self.assertFalse(system["componentPage"]["hasMore"]) + self.assertIsNone(system["componentPage"]["nextOffset"]) + self.assertEqual(len(system["components"]), 1) + self.assertEqual(system["components"][0]["id"], "medium") + self.assertEqual(system["components"][0]["type"], "amesim_helium_medium") + self.assertFalse(system["components"][0]["compiledForSimulation"]) + self.assertIn( + {"type": "amesim_helium_medium", "count": 1}, + system["componentTypes"], + ) + + +class TextNormalizationTests(unittest.TestCase): + def _repair_args( + self, + source: Path, + output: Path, + expected_hash: str, + *, + confirmed: bool, + confirmation_token: str | None = None, + ) -> argparse.Namespace: + return argparse.Namespace( + input=str(source), + format="auto", + output=str(output), + expected_sha256=expected_hash, + confirmed=confirmed, + confirmation_token=confirmation_token, + base_url=BASE_URL, + timeout=1.0, + ) + + def test_preview_does_not_write_and_confirmation_writes_new_file(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source_path = root / "source.json" + output_path = root / "normalized.json" + raw = b"\xef\xbb\xbf" + json.dumps(PROJECT, separators=(",", ":")).encode("utf-8") + b"\r\n" + source_path.write_bytes(raw) + source_hash = skill.load_source(str(source_path)).sha256 + + with ( + mock.patch.object( + skill, + "http_request", + side_effect=compile_http_response, + ), + mock.patch.object(skill, "emit_json") as emit, + ): + preview_code = skill.command_repair_format( + self._repair_args( + source_path, + output_path, + source_hash, + confirmed=False, + ) + ) + preview = emit.call_args.args[0] + + self.assertEqual(preview_code, 0) + self.assertFalse(output_path.exists()) + self.assertFalse(preview["written"]) + self.assertTrue(preview["requiresConfirmation"]) + self.assertEqual(preview["preview"]["sourceSha256"], source_hash) + self.assertTrue(preview["preview"]["unifiedDiff"]) + + emit.reset_mock() + write_code = skill.command_repair_format( + self._repair_args( + source_path, + output_path, + source_hash, + confirmed=True, + confirmation_token=preview["preview"]["confirmationToken"], + ) + ) + written = emit.call_args.args[0] + + normalized = output_path.read_bytes() + self.assertEqual(write_code, 0) + self.assertTrue(written["written"]) + self.assertFalse(normalized.startswith(b"\xef\xbb\xbf")) + self.assertNotIn(b"\r\n", normalized) + self.assertTrue(normalized.endswith(b"\n")) + self.assertEqual(json.loads(normalized), PROJECT) + self.assertEqual(source_path.read_bytes(), raw) + + def test_stale_confirmation_hash_and_source_overwrite_are_rejected(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source_path = root / "source.json" + source_path.write_text(json.dumps(PROJECT), encoding="utf-8") + source_hash = skill.load_source(str(source_path)).sha256 + + with self.assertRaises(skill.InputError) as stale: + skill.command_repair_format( + self._repair_args( + source_path, + root / "output.json", + "0" * 64, + confirmed=True, + ) + ) + self.assertEqual(stale.exception.code, "SOURCE_HASH_MISMATCH") + + with self.assertRaises(skill.InputError) as overwrite: + skill.command_repair_format( + self._repair_args( + source_path, + source_path, + source_hash, + confirmed=True, + ) + ) + self.assertEqual(overwrite.exception.code, "SOURCE_OVERWRITE_FORBIDDEN") + + def test_confirmation_token_binds_the_preview_target(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source_path = root / "source.json" + preview_target = root / "preview-target.json" + changed_target = root / "changed-target.json" + source_path.write_text(json.dumps(PROJECT, separators=(",", ":")), encoding="utf-8") + source_hash = skill.load_source(str(source_path)).sha256 + + with ( + mock.patch.object( + skill, + "http_request", + side_effect=compile_http_response, + ), + mock.patch.object(skill, "emit_json") as emit, + ): + skill.command_repair_format( + self._repair_args( + source_path, + preview_target, + source_hash, + confirmed=False, + ) + ) + token = emit.call_args.args[0]["preview"]["confirmationToken"] + + with self.assertRaises(skill.InputError) as caught: + skill.command_repair_format( + self._repair_args( + source_path, + changed_target, + source_hash, + confirmed=True, + confirmation_token=token, + ) + ) + + self.assertEqual(caught.exception.code, "REPAIR_CONFIRMATION_TOKEN_MISMATCH") + self.assertFalse(preview_target.exists()) + self.assertFalse(changed_target.exists()) + + def test_output_race_never_deletes_a_file_created_by_another_process(self) -> None: + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "race.json" + + def raced_open(*_args: object, **_kwargs: object): + with io.open(output, "wb") as handle: + handle.write(b"other-process") + raise FileExistsError("simulated output race") + + with mock.patch("pathlib.Path.open", side_effect=raced_open): + with self.assertRaises(skill.ArtifactError): + skill._write_new_file(output, b"skill-output") + + self.assertEqual(output.read_bytes(), b"other-process") + + +class SimulationTests(unittest.TestCase): + def _simulate_args( + self, + source: Path, + output: Path, + *, + variables: list[str], + chart_mode: str, + ) -> argparse.Namespace: + return argparse.Namespace( + input=str(source), + format="auto", + output_dir=str(output), + variables=variables, + chart_mode=chart_mode, + simulation_id="simulation-test-1", + base_url=BASE_URL, + timeout=1.0, + ) + + def test_streaming_simulation_writes_result_full_csv_and_svg(self) -> None: + result = { + "status": "completed", + "success": True, + "actualEndTime": 1.0, + "variables": RESULT_VARIABLES, + "series": { + "time": [0.0, 0.5, 1.0], + "mass.position": [0.0, 0.25, 1.0], + "mass.velocity": [0.0, 1.0, 2.0], + }, + } + stream = io.BytesIO( + b'{"event":"progress","phase":"integrating","progress":50}\n' + + json_bytes({"event": "result", "result": result}) + + b"\n" + ) + csv_payloads: list[dict[str, object]] = [] + + def simulation_http_response( + _base_url: str, + path: str, + **kwargs: object, + ) -> tuple[bytes, dict[str, str]]: + if path == "/api/reactflow/compile-model": + return json_bytes(COMPILED_MODEL), {} + if path == "/api/reactflow/system-xml": + return SYSTEM_XML, {"Content-Type": "application/xml"} + if path == "/api/simulation-results/csv": + body = kwargs.get("body") + if not isinstance(body, bytes): + raise AssertionError("CSV request did not contain a JSON body") + csv_payloads.append(json.loads(body.decode("utf-8"))) + return b"time,mass.position,mass.velocity\n0,0,0\n0.5,0.25,1\n1,1,2\n", {} + raise AssertionError(f"Unexpected HTTP request: {path}") + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source_path = root / "project.json" + output_path = root / "artifacts" + source_path.write_text(json.dumps(PROJECT), encoding="utf-8") + + with ( + mock.patch.object( + skill, + "http_request", + side_effect=simulation_http_response, + ), + mock.patch.object( + skill, + "_open_simulation_stream", + return_value=stream, + ) as open_stream, + mock.patch.object(skill, "emit_json") as emit, + ): + exit_code = skill.command_simulate( + self._simulate_args( + source_path, + output_path, + variables=["mass.position"], + chart_mode="separate", + ) + ) + + self.assertEqual(exit_code, 0) + open_stream.assert_called_once_with( + BASE_URL, + SYSTEM_XML, + "simulation-test-1", + 1.0, + ) + self.assertEqual(json.loads((output_path / "result.json").read_text("utf-8")), result) + self.assertIn("mass.velocity", (output_path / "results.csv").read_text("utf-8")) + charts = list(output_path.glob("curve-mass.position-*.svg")) + self.assertEqual(len(charts), 1) + chart = charts[0] + self.assertIn(" None: + event = {"event": "progress", "phase": "integrating", "progress": 4} + + self.assertFalse( + skill._should_emit_progress( + event, + previous_phase="integrating", + previous_progress=0.0, + seconds_since_emit=1.0, + ) + ) + self.assertTrue( + skill._should_emit_progress( + {**event, "progress": 5}, + previous_phase="integrating", + previous_progress=0.0, + seconds_since_emit=1.0, + ) + ) + self.assertTrue( + skill._should_emit_progress( + event, + previous_phase="compilation", + previous_progress=0.0, + seconds_since_emit=1.0, + ) + ) + self.assertTrue( + skill._should_emit_progress( + event, + previous_phase="integrating", + previous_progress=0.0, + seconds_since_emit=5.0, + ) + ) + + def test_real_stream_throttles_stdout_but_logs_every_progress_event(self) -> None: + progress_events = [ + {"event": "progress", "phase": "validation", "progress": 0}, + {"event": "progress", "phase": "validation", "progress": 1}, + {"event": "progress", "phase": "validation", "progress": 4}, + {"event": "progress", "phase": "validation", "progress": 5}, + {"event": "progress", "phase": "integrating", "progress": 6}, + {"event": "progress", "phase": "integrating", "progress": 7}, + {"event": "progress", "phase": "integrating", "progress": 100}, + ] + result = { + "status": "completed", + "success": True, + "actualEndTime": 1.0, + "variables": RESULT_VARIABLES, + "series": { + "time": [0.0, 1.0], + "mass.position": [0.0, 1.0], + "mass.velocity": [0.0, 2.0], + }, + } + lines = [json_bytes(event) + b"\n" for event in progress_events] + lines.append(json_bytes({"event": "result", "result": result}) + b"\n") + + with tempfile.TemporaryDirectory() as directory: + progress_path = Path(directory) / "progress.jsonl" + with ( + mock.patch.object( + skill, + "_open_simulation_stream", + return_value=io.BytesIO(b"".join(lines)), + ), + mock.patch.object( + skill.time, + "monotonic", + side_effect=[100.0, 101.0, 102.0, 103.0, 104.0, 110.0, 111.0], + ), + mock.patch.object(skill, "emit_json") as emit, + ): + returned_result, returned_error = skill._read_simulation_stream( + BASE_URL, + SYSTEM_XML, + "simulation-test-1", + 10.0, + progress_path, + ) + + logged = [ + json.loads(line) + for line in progress_path.read_text("utf-8").splitlines() + ] + + self.assertEqual(returned_result, result) + self.assertIsNone(returned_error) + self.assertEqual(logged[:7], progress_events) + self.assertEqual(len(logged), 8) + self.assertEqual(logged[-1]["event"], "result") + self.assertNotIn("result", logged[-1]) + public = [call.args[0] for call in emit.call_args_list] + self.assertEqual( + [event.get("progress") for event in public if event["event"] == "progress"], + [0, 5, 6, 7, 100], + ) + self.assertEqual(public[-1]["event"], "result") + self.assertNotIn("result", public[-1]) + self.assertEqual(public[-1]["sampleCount"], 2) + + def test_overlay_and_stacked_positive_chart_generation(self) -> None: + variables = { + "mass.position": RESULT_VARIABLES[0], + "reference.position": { + **RESULT_VARIABLES[0], + "key": "reference.position", + "componentId": "reference", + }, + "mass.velocity": RESULT_VARIABLES[1], + } + series = { + "time": [0.0, 0.5, 1.0], + "mass.position": [0.0, 0.25, 1.0], + "reference.position": [1.0, 0.5, 0.0], + "mass.velocity": [0.0, 1.0, 2.0], + } + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + overlay_paths = skill._write_charts( + root, + "overlay", + ["mass.position", "reference.position"], + variables, + series, + ) + stacked_paths = skill._write_charts( + root, + "stacked", + ["mass.position", "mass.velocity"], + variables, + series, + ) + + overlay = Path(overlay_paths[0]).read_text("utf-8") + stacked = Path(stacked_paths[0]).read_text("utf-8") + + self.assertEqual([Path(path).name for path in overlay_paths], ["curves-overlay.svg"]) + self.assertEqual([Path(path).name for path in stacked_paths], ["curves-stacked.svg"]) + self.assertEqual(overlay.count(" None: + partial_result = { + "status": "stopped", + "success": False, + "partial": True, + "simulatedUntil": 0.5, + "variables": RESULT_VARIABLES, + "series": { + "time": [0.0, 0.5], + "mass.position": [0.0, 0.25], + "mass.velocity": [0.0, 1.0], + }, + } + inspection = { + "system": { + "name": "partial-system", + "resultVariables": RESULT_VARIABLES, + } + } + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source_path = root / "project.json" + output_path = root / "partial-artifacts" + source_path.write_text(json.dumps(PROJECT), encoding="utf-8") + + with ( + mock.patch.object(skill, "inspect_source", return_value=inspection), + mock.patch.object(skill, "_json_to_xml", return_value=SYSTEM_XML), + mock.patch.object( + skill, + "_read_simulation_stream", + return_value=(partial_result, None), + ), + mock.patch.object( + skill, + "_download_csv", + return_value=b"time,mass.position,mass.velocity\n0,0,0\n0.5,0.25,1\n", + ), + mock.patch.object(skill, "emit_json") as emit, + ): + exit_code = skill.command_simulate( + self._simulate_args( + source_path, + output_path, + variables=["mass.position"], + chart_mode="separate", + ) + ) + + artifact_event = emit.call_args.args[0] + self.assertEqual(exit_code, 4) + self.assertEqual(artifact_event["event"], "artifacts") + self.assertEqual(artifact_event["status"], "stopped") + self.assertTrue(artifact_event["partial"]) + self.assertTrue((output_path / "result.json").is_file()) + self.assertTrue((output_path / "results.csv").is_file()) + self.assertEqual(len(list(output_path.glob("curve-mass.position-*.svg"))), 1) + + def test_overlay_unit_mismatch_is_rejected_before_simulation(self) -> None: + mismatched = { + "system": { + "name": "unit-test", + "resultVariables": [ + {"key": "pressure", "unit": "Pa"}, + {"key": "flow", "unit": "kg/s"}, + ], + } + } + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + source_path = root / "project.json" + output_path = root / "must-not-exist" + source_path.write_text(json.dumps(PROJECT), encoding="utf-8") + + with ( + mock.patch.object(skill, "inspect_source", return_value=mismatched), + mock.patch.object(skill, "_read_simulation_stream") as run, + ): + with self.assertRaises(skill.InputError) as caught: + skill.command_simulate( + self._simulate_args( + source_path, + output_path, + variables=["pressure", "flow"], + chart_mode="overlay", + ) + ) + + self.assertEqual(caught.exception.code, "OVERLAY_UNITS_MISMATCH") + run.assert_not_called() + self.assertFalse(output_path.exists()) + + +class TaskControlTests(unittest.TestCase): + def test_status_queries_the_requested_task(self) -> None: + args = argparse.Namespace( + simulation_id="run-123", + base_url=BASE_URL, + timeout=2.0, + ) + task = {"simulationId": "run-123", "status": "integrating"} + + with ( + mock.patch.object(skill, "http_json", return_value=task) as http_json, + mock.patch.object(skill, "emit_json") as emit, + ): + exit_code = skill.command_status(args) + + self.assertEqual(exit_code, 0) + http_json.assert_called_once_with( + BASE_URL, + "/api/system-xml/simulations/run-123", + timeout=2.0, + ) + self.assertEqual( + emit.call_args.args[0], + {"ok": True, "command": "status", "task": task}, + ) + + def test_status_compacts_a_completed_result(self) -> None: + args = argparse.Namespace( + simulation_id="run-123", + base_url=BASE_URL, + timeout=2.0, + ) + task = { + "simulationId": "run-123", + "status": "completed", + "result": { + "status": "completed", + "success": True, + "simulatedUntil": 1.0, + "variables": RESULT_VARIABLES, + "series": {"time": [0.0, 1.0], "mass.position": [0.0, 1.0]}, + }, + } + + with ( + mock.patch.object(skill, "http_json", return_value=task), + mock.patch.object(skill, "emit_json") as emit, + ): + exit_code = skill.command_status(args) + + self.assertEqual(exit_code, 0) + public_task = emit.call_args.args[0]["task"] + self.assertNotIn("result", public_task) + self.assertNotIn("fullResult", public_task["resultSummary"]) + self.assertEqual(public_task["resultSummary"]["sampleCount"], 2) + self.assertEqual(public_task["resultSummary"]["actualEndTime"], 1.0) + + def test_cancel_posts_reason_to_the_requested_task(self) -> None: + args = argparse.Namespace( + simulation_id="run-123", + reason="stalled", + base_url=BASE_URL, + timeout=2.0, + ) + task = {"simulationId": "run-123", "status": "cancelling"} + + with ( + mock.patch.object(skill, "http_json", return_value=task) as http_json, + mock.patch.object(skill, "emit_json") as emit, + ): + exit_code = skill.command_cancel(args) + + self.assertEqual(exit_code, 0) + http_json.assert_called_once_with( + BASE_URL, + "/api/system-xml/simulations/run-123/cancel", + method="POST", + payload={"reason": "stalled"}, + timeout=2.0, + ) + self.assertEqual( + emit.call_args.args[0], + {"ok": True, "command": "cancel", "task": task}, + ) + + +if __name__ == "__main__": + unittest.main()