diff --git a/skills/system-simulation/SKILL.md b/skills/system-simulation/SKILL.md index 94214d8..b93cfbd 100644 --- a/skills/system-simulation/SKILL.md +++ b/skills/system-simulation/SKILL.md @@ -34,12 +34,12 @@ metadata: 4. 用户用显示名称描述组件或变量时，利用检查结果中的稳定 ID、结果 `key`、物理量和单位消歧。存在重名、多个候选或“参数/结果变量”含义不清时，先询问，不能替用户猜。 5. 使用 `simulate` 的事件流持续判断 queued、validating、compiling、integrating 和结束状态。仿真时间暂时不变但内部活动仍增长时，只说明正在处理慢步，不能宣称卡死。 6. 成功运行后交付用户选择的 SVG 曲线和完整 `results.csv`，并简要说明完成状态、实际仿真终点和重要诊断。失败或取消时交付能够安全生成的部分结果；若运行前即失败而没有 CSV，要明确说明原因。 -7. 优化需求优先按自然语言理解：从检查结果补齐稳定结果 `key`、单位和当前参数值，未指定的算法、预算、容差和输出目录采用参考文档中的推荐默认值。不要要求用户填写规格 JSON，也不要追问随机种子、变异因子等已有默认值。 -8. 信息足以形成规格后，直接在内部写入规格并执行无候选仿真的 `plan`，无需先征求生成计划的许可。计划阶段用普通语言只展示目标、可调参数及范围、约束、最多仿真次数、输出位置和重要假设。参数明显是连续物理标量时，把连续性作为计划假设，一次整体执行确认即可覆盖；只有语义确有歧义时才自然地追问。内部声明代码、SHA、`planHash` 和 `confirmationToken` 默认不展示。 -9. 生成计划不等于获准执行。只有用户看过计划摘要并明确表示开始后才能传入 `--confirmed`；用户只要求计划时必须停在计划阶段。计划任一实质内容变化都要重新确认。 +7. 优化需求优先按自然语言理解：从检查结果补齐稳定结果 `key`、单位和当前参数值，未指定的算法、预算、容差和输出目录采用参考文档中的推荐默认值。不要要求用户填写规格 JSON，也不要追问随机种子、变异因子等已有默认值。`plan --present` 成功后，面向用户展示的设计变量当前值和单位必须直接采用 `presentation.designVariables[].current` 与 `unit`；完整审计计划中的对应字段是 `designVariables[].initial` 与 `unit`。不得根据源 JSON 的 `parameterUnits` 再换算或另行推断。 +8. 信息足以形成规格后，直接在内部写入规格并执行无候选仿真的 `plan`，无需先征求生成计划的许可。普通流程必须把完整计划以仅当前用户可读的权限保存到输出目录之外的新内部文件，并让 stdout 只返回展示白名单。计划阶段从回复第一个字起使用用户当前语言并直接展示计划，只列目标、可调参数及范围、约束、仿真预算槽位、搜索启动时限、完整输出位置和重要假设。输出位置必须是 `plan` 返回的完整绝对路径，不用 `...` 缩写。采用默认搜索设置且没有需要用户决策的警告时，只说“采用默认搜索设置”及其执行上限，不显示“无警告”或原始 warnings、算法名称或变体、随机种子、种群、变异/交叉参数、搜索/复验预算拆分、边界处理、端点播种或理论完整代数；若展示视图返回 `nonDefaultSettings`，则必须把其中将被确认的非默认值简明列出。参数明显是连续物理标量时，把连续性作为计划假设，一次整体执行确认即可覆盖，不展示用于作出判断的内部合同字段清单；只有语义确有歧义时才自然地追问。内部声明代码、SHA、`planHash` 和 `confirmationToken` 默认不展示。 +9. 生成计划不等于获准执行。只有用户看过计划摘要并明确表示开始后才能传入 `--confirmed`；用户说只要计划、先看计划且暂时不要运行或其他同等表述时，展示计划后直接停住，不在本轮追问是否开始。计划任一实质内容变化都要重新确认。 10. 优化中的失败、取消、停滞或不完整仿真不计算目标分数。最终只对预算内找到的最佳可行候选做一次绕过缓存的完整复验；复验通过前不把候选称为已验证方案。 -11. 严格区分搜索停止与候选复验：`verified` 只说明最佳搜索候选的新鲜复验通过，不等于搜索收敛、系统达到稳态或全局最优。报告时分别说明搜索候选评估、按阶段拆分的仿真预算槽位占用及完成/失败记录、未形成试验记录的槽位、缓存命中、独立复验、未用预算和真实停止原因；槽位占用不能说成后端已接收或已完成。内部防死循环上限及重复停滞后的种群塌缩都不得解释成“已收敛”。 -12. 目标使用 `final` 时必须报告脚本给出的末段趋势诊断状态；只有完整且覆盖计划终点的新鲜序列才能分析趋势，诊断不可用或样本不足时明确说明且不自行推断。检测到明显变化时，说明终点值只是快照。新鲜复验已经通过后不再建议重复同一复验；最佳点落在边界时，只能说明已采样点的趋势，未经工程可行性确认不得建议放宽边界。 +11. 严格区分搜索停止与候选复验：`verified` 只说明最佳搜索候选的新鲜复验通过，不等于搜索收敛、系统达到稳态或全局最优。优化执行完成后的结果报告分别说明搜索候选评估、按阶段拆分的仿真预算槽位占用及完成/失败记录、未形成试验记录的槽位、缓存命中、独立复验、未用预算和真实停止原因；这些审计明细不属于计划摘要，槽位占用也不能说成后端已接收或已完成。只要 `searchConvergenceEstablished` 为 `false`，计划、进度和结果中都不得说搜索“已收敛”“将收敛”或“大概率收敛”；若只想表达重复运行可能得到相同结果，改说“可能再次找到同一候选”。内部防死循环上限及重复停滞后的种群塌缩都不得解释成收敛。 +12. 目标使用 `final` 时必须报告脚本给出的末段趋势诊断状态；只有完整且覆盖计划终点的新鲜序列才能分析趋势，诊断不可用或样本不足时明确说明且不自行推断。检测到明显变化时，说明终点值只是快照。有限样本只能表述为“在这些已评估点中，参数增大时结果均增大或均减小”，不得称整个范围单调，也不得外推样本之间或未采样位置。新鲜复验已经通过后不再建议重复同一复验；最佳点落在边界时，未经物理、安全和组件合同方面的工程可行性确认，不得建议放宽边界或把它列作默认下一步。 脚本命令统一从仓库根目录运行： diff --git a/skills/system-simulation/references/optimization-workflow.md b/skills/system-simulation/references/optimization-workflow.md index 96951e6..4d379f5 100644 --- a/skills/system-simulation/references/optimization-workflow.md +++ b/skills/system-simulation/references/optimization-workflow.md @@ -49,7 +49,9 @@ 用户描述了响应约束但未指定容差时，默认 `tolerance = 0`；未指定 `scale` 时，取该约束所有非空边界绝对值的最大值，若结果为零则取 `1`。目标或约束接近零、后端存在可观察的不确定性，或用户给出安全裕量时，应提出有物理意义的容差建议，不能用一个跨量纲的非零绝对容差替代判断。 -默认输出目录使用项目文件所在目录下尚不存在的 `optimization-runs/<项目名>-`。源目录不可写时，改用当前可写工作区中的同名新目录，并在计划摘要中说明实际位置。不要让用户命名目录，也不要覆盖既有目录。默认计划可对用户概括为“差分进化、最多 25 次仿真、最长 1 小时，并预留一次独立复验”；除非用户询问或计划产生覆盖不足警告，不主动讲解种群、变异因子、交叉概率、随机种子或代数公式。 +默认输出目录使用项目文件所在目录下尚不存在的 `optimization-runs/<项目名>-`。源目录不可写时，改用当前可写工作区中的同名新目录，并在计划摘要中说明实际位置。不要让用户命名目录，也不要覆盖既有目录。默认计划可对用户概括为“使用默认搜索设置，最多占用 25 个仿真预算槽位；搜索启动时限为 1 小时，并为找到的最佳可行候选预留一次独立复验”。搜索启动时限到达后不会取消正在运行的健康仿真，预留复验也可能在其后执行，所以不能把它称为总耗时硬上限。除非用户询问或计划产生需要用户决策的覆盖不足警告，不主动讲解算法名称或变体、种群、变异因子、交叉概率、随机种子、搜索/复验预算拆分、边界处理、端点播种、理论完整代数或公式。 + +生成新计划不以历史运行作为前置检查；除非用户要求复用、比较或解释旧结果，不主动扫描、校验或汇总旧优化目录。若当前对话已经明确存在源文件与实质规格相同的历史运行，为避免混淆最多用一句话注明它只是历史参考、不属于本次计划；用户未追问时不展开旧候选、停止细节、趋势或复验数据，也不用历史样本预测新搜索会收敛或断言连续区间性质。 ## 优化规格 JSON @@ -137,7 +139,7 @@ 若参数像整数、计数、无量纲模式量、条件控制量，元数据彼此矛盾，或无法判断改变它是否影响端口和拓扑，必须先用自然语言询问，例如：“这个参数可以取任意小数，并且调整时不会切换组件模式或端口吗？”不要向用户显示内部声明代码。`editor`、`options` 和未来可能出现的显式否决只用于拒绝明显不适用的参数，不能证明其余参数连续。 -参数值、初值和边界一律使用线性 SI 合同。如果用户用显示单位给出边界，先换算为 SI 并在计划中展示。后端在计划阶段将源 JSON 转换为基准 System XML v3；每个候选都从该 XML 重新生成，只替换选中 `Parameter/@value` 的 SI 数字，不在前一个候选上累积修改，源 JSON 永不被覆盖。 +参数值、初值和边界一律使用线性 SI 合同。ReactFlow JSON 中普通数值参数已经是 SI 值，`parameterUnits` 只是编辑器显示信息，不能据此把普通数值再次换算；表达式所需的显示单位换算由后端完成。如果用户用显示单位给出边界，才把用户输入换算为 SI。普通 `plan --present` 成功后，`presentation.designVariables[].current` 和 `unit` 是计划摘要中当前值与单位的唯一依据；完整审计计划中的对应字段为 `designVariables[].initial` 和 `unit`。不要从源 JSON 重新计算显示值，若其他信息与它矛盾则先排查而不是向用户展示两套数值。用户给出的边界已经使用该 SI 单位且没有矛盾时，不主动解释 `parameterUnits` 或添加显示单位换算旁注。后端在计划阶段将源 JSON 转换为基准 System XML v3；每个候选都从该 XML 重新生成，只替换选中 `Parameter/@value` 的 SI 数字，不在前一个候选上累积修改，源 JSON 永不被覆盖。 ### 统计量与响应约束 @@ -153,7 +155,7 @@ ## 计划确认 -信息足以形成规格后直接执行 `plan`，不需要用户先批准计划生成。它校验源 JSON、规格、结果键/单位、设计变量合同和边界，并请求后端生成基准 XML，但不开始优化候选仿真。其结构化输出包含源 JSON 与规格 JSON 的绝对路径/SHA-256、解析后规格、基准 XML SHA-256、目标/约束元数据、解析后设计变量、执行上限、绝对输出目录、`parameterContinuity`、`requiredAssertions`，以及值相同的 `planHash` 与 `confirmationToken`。 +信息足以形成规格后直接执行 `plan`，不需要用户先批准计划生成。它校验源 JSON、规格、结果键/单位、设计变量合同和边界，并请求后端生成基准 XML，但不开始优化候选仿真。完整审计计划文件包含源 JSON 与规格 JSON 的绝对路径/SHA-256、解析后规格、基准 XML SHA-256、目标/约束元数据、解析后设计变量、执行上限、绝对输出目录、`parameterContinuity`、`requiredAssertions`，以及值相同的 `planHash` 与 `confirmationToken`；普通流程的 stdout 只返回严格白名单的展示视图，不包含这些执行凭据和搜索内部字段。 计划中的 `OPTIMIZATION_CONTINUITY_USER_ASSERTION` 是给脚本和审计使用的内部标识，不是要求用户照抄的口令。后端不会验证参数的物理/语义连续性，因此面向用户的计划摘要必须用普通语言列出相关假设。若参数语义清楚，用户在看到摘要后明确同意开始运行，即视为同时接受完整计划和这些假设；未得到这次整体确认时不得传入 `--confirmed`。若参数语义不清，则应在执行确认之前先完成自然语言消歧。 @@ -166,11 +168,28 @@ - 要改善哪个结果，用什么统计口径； - 调整哪些参数，各自在什么范围； - 有哪些响应约束； -- 采用默认还是用户指定的搜索配置，最多提交多少次仿真、最长多久； +- 采用默认还是用户指定的搜索配置、最多占用多少仿真预算槽位、停止启动新搜索候选的时限，以及可使总耗时超过该时限的在途仿真和预留复验； - 哪些参数连续性或工程边界属于假设； - 输出写到哪里，并明确源模型不变。 -摘要后只问一次自然问题，例如：“就按这个方案开始吗？”用户明确同意后直接执行，不再追加连续性声明、算法参数或 token 确认。若用户明确只要计划，则交付摘要后停止，不把问题措辞成已经准备执行；等用户之后主动要求开始。 +计划正文只描述即将执行的运行，不把历史结果回顾、搜索审计明细或对本次结果的预测混入计划。默认设置不存在警告时，“采用默认搜索设置”已足够，不再把内部配置、端点播种或合同判定字段展开成技术清单。若安全展示视图包含 `nonDefaultSettings`，只列出其中实际偏离推荐默认值、并会随本计划一起确认的设置；不要反过来读取完整审计计划扩展技术细节。 + +用户尚未限制本轮只做计划、且接下来是否执行需要确认时，摘要后只问一次中性的自然问题，例如：“就按这个方案开始吗？”不要主动把换目标、放宽参数边界或其他扩展范围列成备选项。用户明确同意后直接执行，不再追加连续性声明、算法参数或 token 确认。若用户说“只要计划”“暂时不要运行”或同等意思，则交付摘要后直接陈述会停在计划阶段，不在本轮询问是否开始，等用户之后主动要求。用户主动提出调整时再讨论；涉及放宽工程边界时，必须先确认新的范围符合物理、安全和组件合同。 + +无警告且使用默认设置时，按下列内容边界组织计划回复；可以顺应用户语言调整措辞，但不要增加其他技术段落： + +```text +优化目标：让哪个结果按什么统计口径变大、变小或接近目标值。 +调整参数：参数名称、脚本 plan 返回的当前 SI 值、用户确认的 SI 范围。 +响应约束：列出约束；没有就说无。 +运行上限：采用默认搜索设置，最多占用多少仿真预算槽位；搜索到时后不再启动新候选，但会等在途仿真结束，并为找到的最佳可行候选预留一次独立复验。 +重要假设：用一句普通语言说明参数按连续物理量处理且不改变模式或结构。 +输出：plan 返回的新目录完整绝对路径，源模型不变。 + +结束语：若本轮可以询问执行，则问“就按这个方案开始吗？”；若用户说暂时不要运行，则说“计划已准备好，我会停在这里，等你之后明确说开始。” +``` + +正式计划回复从第一个字起使用用户当前语言并直接进入计划内容；不加过程旁白，不显示 `warnings: []` 等内部状态，不复述内部枚举名，也不在计划后追加单位科普、算法原理、历史回顾、结果预测或调整建议。输出目录照抄 `plan` 返回的完整绝对路径，不用省略号或相对路径。只有真实警告、无法消除的单位歧义或其他需要用户决策的问题，才在相应条目中简短说明。 ## DE/rand/1/bin 搜索 @@ -240,7 +259,7 @@ crossoverProbability = [0, 1] 只有 `solutionStatus == verified` 时才生成 `best-parameters.json`、`best-system.xml`、`best-project.json`、`result.json`、完整 `results.csv` 和目标/响应约束的独立 SVG 曲线。`best-project.json` 将被优化参数的原表达式替换为普通 SI 数值，源 JSON 不变。`optimize` 仅在状态为 `verified` 时返回退出码 `0`，其他结果返回 `4`。 -最终汇报必须使用这一口径： +只有 `solutionStatus == verified` 时，最终汇报才使用这一口径： > 这是实际完成仿真的搜索点中表现最好的可行候选，并已通过一次独立复验；复验不证明搜索收敛、系统达到稳态或全局最优。 @@ -255,10 +274,14 @@ crossoverProbability = [0, 1] ```powershell py -3.12 skills/system-simulation/scripts/optimization_skill.py plan PROJECT.json ` --spec optimization-spec.json ` - --output-dir OUTPUT_DIR + --output-dir OUTPUT_DIR ` + --plan-file PLAN_FILE ` + --present ``` -向用户展示计划并获得明确确认后，原样使用 `plan` 返回的三个值： +普通 Skill 流程必须同时使用 `--plan-file` 和 `--present`：完整审计计划以 `0600` 权限独占写入 `PLAN_FILE`，stdout 只返回用户计划所需的白名单字段。`PLAN_FILE` 必须是位于 `OUTPUT_DIR` 外的新文件，不能覆盖既有文件，也不能与输出目录互为祖先或后代；默认用本次输出目录名加 UTC 时间戳或随机后缀生成同级文件，不复用固定的临时文件名。省略这两个选项的旧式完整 stdout 只用于兼容测试或显式审计排障，不用于普通对话。 + +向用户展示计划并获得明确确认后，在内部从 `PLAN_FILE` 读取并原样使用源 SHA、规格 SHA 和 `confirmationToken`，不要向用户展示： ```powershell py -3.12 skills/system-simulation/scripts/optimization_skill.py optimize PROJECT.json ` @@ -275,7 +298,9 @@ Linux 使用已确认的 Python 3.12 解释器和相同参数： ```bash python3.12 skills/system-simulation/scripts/optimization_skill.py plan PROJECT.json \ --spec optimization-spec.json \ - --output-dir OUTPUT_DIR + --output-dir OUTPUT_DIR \ + --plan-file PLAN_FILE \ + --present python3.12 skills/system-simulation/scripts/optimization_skill.py optimize PROJECT.json \ --spec optimization-spec.json \ @@ -291,7 +316,8 @@ python3.12 skills/system-simulation/scripts/optimization_skill.py optimize PROJE ```bash python3.12 skills/system-simulation/scripts/optimization_skill.py \ --base-url http://127.0.0.1:18082 --timeout 60 \ - plan PROJECT.json --spec optimization-spec.json --output-dir OUTPUT_DIR + plan PROJECT.json --spec optimization-spec.json --output-dir OUTPUT_DIR \ + --plan-file PLAN_FILE --present ``` OpenClaw 中不假设当前目录是仓库根目录，使用 Skill 根目录占位符： @@ -299,9 +325,11 @@ OpenClaw 中不假设当前目录是仓库根目录，使用 Skill 根目录占 ```bash python3.12 "{baseDir}/scripts/optimization_skill.py" plan PROJECT.json \ --spec optimization-spec.json \ - --output-dir OUTPUT_DIR + --output-dir OUTPUT_DIR \ + --plan-file PLAN_FILE \ + --present ``` -也可以先进入本 Skill 目录，再使用 `scripts/optimization_skill.py plan ...` 和 `scripts/optimization_skill.py optimize ...`。不根据用户主目录、OpenClaw 数据目录或仓库名称猜测脚本路径。持续消费 JSONL 进展，定期报告已使用/最大后端提交数、当前代数、最佳可行目标、失败数、缓存命中和内层仿真阶段；不因仿真时间短暂停滞而声称卡死。 +也可以先进入本 Skill 目录，再使用 `scripts/optimization_skill.py plan ...` 和 `scripts/optimization_skill.py optimize ...`。不根据用户主目录、OpenClaw 数据目录或仓库名称猜测脚本路径。持续消费 JSONL 进展，定期报告已占用/最大仿真预算槽位、已完成和失败记录、当前代数、最佳可行目标、缓存命中和内层仿真阶段；不把槽位占用说成后端已接收或已完成，也不因仿真时间短暂停滞而声称卡死。 通常省略 `--optimization-id` 让脚本生成唯一 ID。若显式指定，同一后端任务保留窗口内必须使用新的 ID；快速复用旧 ID 会被后端按冲突拒绝。 diff --git a/skills/system-simulation/scripts/optimization_skill.py b/skills/system-simulation/scripts/optimization_skill.py index 29cb675..b3f6cf5 100644 --- a/skills/system-simulation/scripts/optimization_skill.py +++ b/skills/system-simulation/scripts/optimization_skill.py @@ -55,6 +55,12 @@ MAX_DESIGN_VARIABLES = 16 MAX_RESPONSE_CONSTRAINTS = 16 MAX_SIMULATION_RUNS = 200 MAX_WALL_SECONDS = 7 * 24 * 60 * 60 +RECOMMENDED_ALGORITHM_SEED = 0 +RECOMMENDED_POPULATION_SIZE = 8 +RECOMMENDED_MUTATION_FACTOR = 0.8 +RECOMMENDED_CROSSOVER_PROBABILITY = 0.7 +RECOMMENDED_VALIDATION_RELATIVE_TOLERANCE = 1e-8 +RECOMMENDED_VALIDATION_ABSOLUTE_TOLERANCE = 0.0 OPTIMIZATION_ID_PATTERN = re.compile(r"^[A-Za-z0-9._-]{1,96}$") IDENTIFIER_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9._-]{0,63}$") SUPPORTED_STATISTICS = { @@ -306,6 +312,148 @@ class RuntimePlan: timeout: float confirmation_token: str + def presentation_dict(self) -> dict[str, object]: + """Return the strict allowlist used for an ordinary user-facing plan.""" + objective = self.spec.objective + objective_metadata = self.variables[objective.result_key] + payload: dict[str, object] = { + "objective": { + "resultKey": objective.result_key, + "componentId": objective_metadata.get("componentId"), + "label": objective_metadata.get("label"), + "quantity": objective_metadata.get("quantity"), + "statistic": objective.statistic.as_dict(), + "goal": objective.goal.as_dict(), + "seriesUnit": objective.expected_unit, + "metricUnit": _metric_unit( + objective.expected_unit, + objective.statistic.kind, + ), + }, + "designVariables": [ + { + "id": item.spec.id, + "componentId": item.spec.component_id, + "parameter": item.spec.parameter, + "label": item.label, + "quantity": item.quantity, + "current": item.initial, + "lower": item.spec.lower, + "upper": item.spec.upper, + "unit": item.spec.unit, + } + for item in self.resolved_design_variables + ], + "constraints": [ + { + "id": item.id, + "resultKey": item.result_key, + "componentId": self.variables[item.result_key].get( + "componentId" + ), + "label": self.variables[item.result_key].get("label"), + "quantity": self.variables[item.result_key].get("quantity"), + "statistic": item.statistic.as_dict(), + "lower": item.lower, + "upper": item.upper, + "tolerance": item.tolerance, + "seriesUnit": item.expected_unit, + "metricUnit": _metric_unit( + item.expected_unit, + item.statistic.kind, + ), + } + for item in self.spec.constraints + ], + "runLimits": { + "maxSimulationBudgetSlots": ( + self.spec.budget.max_simulation_runs + ), + "searchLaunchTimeLimitSeconds": ( + self.spec.budget.max_wall_seconds + ), + "inFlightSimulationMayFinishAfterLimit": True, + "freshVerificationSlotsReserved": 1, + "verificationRunsOnlyIfFeasibleCandidateFound": True, + }, + "assumptions": { + "continuousLinearSiDesignVariableIds": [ + item.spec.id for item in self.resolved_design_variables + ], + "portsTopologyModesUnchanged": True, + }, + "outputDirectory": str(self.output_directory), + "sourceWillBeOverwritten": False, + } + non_default_settings: dict[str, object] = {} + search_settings: dict[str, object] = {} + algorithm = self.spec.algorithm + if algorithm.seed != RECOMMENDED_ALGORITHM_SEED: + search_settings["randomSeed"] = algorithm.seed + if algorithm.population_size != RECOMMENDED_POPULATION_SIZE: + search_settings["populationSize"] = algorithm.population_size + if algorithm.mutation_factor != RECOMMENDED_MUTATION_FACTOR: + search_settings["mutationFactor"] = algorithm.mutation_factor + if ( + algorithm.crossover_probability + != RECOMMENDED_CROSSOVER_PROBABILITY + ): + search_settings["crossoverProbability"] = ( + algorithm.crossover_probability + ) + if search_settings: + non_default_settings["search"] = search_settings + + verification_settings: dict[str, object] = {} + validation = self.spec.validation + if ( + validation.relative_tolerance + != RECOMMENDED_VALIDATION_RELATIVE_TOLERANCE + ): + verification_settings["relativeTolerance"] = ( + validation.relative_tolerance + ) + if ( + validation.absolute_tolerance + != RECOMMENDED_VALIDATION_ABSOLUTE_TOLERANCE + ): + verification_settings["absoluteTolerance"] = ( + validation.absolute_tolerance + ) + if verification_settings: + non_default_settings["verification"] = verification_settings + + constraint_scales = [] + for constraint in self.spec.constraints: + bounds = [ + abs(bound) + for bound in (constraint.lower, constraint.upper) + if bound is not None + ] + recommended_scale = max(bounds, default=0.0) or 1.0 + if constraint.scale != recommended_scale: + constraint_scales.append( + {"id": constraint.id, "scale": constraint.scale} + ) + if constraint_scales: + non_default_settings["constraintRankingScales"] = ( + constraint_scales + ) + if non_default_settings: + payload["nonDefaultSettings"] = non_default_settings + + search_run_limit = self.spec.budget.max_simulation_runs - 1 + population_size = self.spec.algorithm.population_size + if (search_run_limit - population_size) // population_size < 1: + payload["attention"] = [ + ( + "The budget can cover the initial candidate set and reserve " + "fresh verification, but it cannot cover one complete search " + "update cycle if every candidate is unique." + ) + ] + return payload + def public_dict(self) -> dict[str, object]: search_run_limit = self.spec.budget.max_simulation_runs - 1 population_size = self.spec.algorithm.population_size @@ -849,6 +997,86 @@ def _validate_output_target(path_text: str) -> Path: return path +def _validate_plan_file_target( + path_text: str, + output_directory_text: str, +) -> Path: + requested_path = Path(path_text).expanduser() + if requested_path.exists() or requested_path.is_symlink(): + raise simulation.InputError( + "OPTIMIZATION_PLAN_FILE_ALREADY_EXISTS", + "The saved optimization plan path must be new.", + {"path": str(requested_path.absolute())}, + ) + try: + path = requested_path.resolve(strict=False) + output_directory = Path(output_directory_text).expanduser().resolve( + strict=False + ) + except (OSError, RuntimeError) as exc: + raise simulation.InputError( + "OPTIMIZATION_PLAN_FILE_PATH_INVALID", + "The saved plan file path could not be resolved safely.", + {"path": str(requested_path.absolute())}, + ) from exc + if path.exists(): + raise simulation.InputError( + "OPTIMIZATION_PLAN_FILE_ALREADY_EXISTS", + "The saved optimization plan path must be new.", + {"path": str(path)}, + ) + if ( + path == output_directory + or path.is_relative_to(output_directory) + or output_directory.is_relative_to(path) + ): + raise simulation.InputError( + "OPTIMIZATION_PLAN_FILE_OUTPUT_CONFLICT", + "The saved plan file and optimization output directory must be separate sibling paths.", + { + "planFile": str(path), + "outputDirectory": str(output_directory), + }, + ) + return path + + +def _write_new_private_file(path: Path, data: bytes) -> None: + """Exclusively create a plan receipt readable only by its owner.""" + created = False + descriptor: int | None = None + try: + path.parent.mkdir(parents=True, exist_ok=True) + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + if hasattr(os, "O_BINARY"): + flags |= os.O_BINARY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + descriptor = os.open(path, flags, 0o600) + created = True + if hasattr(os, "fchmod"): + os.fchmod(descriptor, 0o600) + with os.fdopen(descriptor, "wb") as handle: + descriptor = None + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + except OSError as exc: + if descriptor is not None: + try: + os.close(descriptor) + except OSError: + pass + try: + if created and path.exists(): + path.unlink() + except OSError: + pass + raise simulation.ArtifactError( + "OUTPUT_WRITE_FAILED", str(exc), {"path": str(path)} + ) from exc + + def _parameter_contracts( inspection: Mapping[str, object], ) -> dict[tuple[str, str], tuple[dict[str, object], dict[str, object] | None]]: @@ -1173,6 +1401,18 @@ def build_runtime_plan( def command_plan(args: argparse.Namespace) -> int: + present = bool(getattr(args, "present", False)) + plan_file_text = getattr(args, "plan_file", None) + if present != bool(plan_file_text): + raise simulation.InputError( + "OPTIMIZATION_PLAN_PRESENTATION_OPTIONS_REQUIRED", + "--present and --plan-file must be supplied together.", + ) + plan_file = ( + _validate_plan_file_target(plan_file_text, args.output_dir) + if plan_file_text + else None + ) plan = build_runtime_plan( args.input, args.spec, @@ -1180,7 +1420,30 @@ def command_plan(args: argparse.Namespace) -> int: base_url=args.base_url, timeout=args.timeout, ) - simulation.emit_json(plan.public_dict()) + audit_payload = plan.public_dict() + if plan_file is not None: + serialized = ( + json.dumps( + audit_payload, + ensure_ascii=False, + allow_nan=False, + indent=2, + ) + + "\n" + ).encode("utf-8") + _write_new_private_file(plan_file, serialized) + if present: + simulation.emit_json( + { + "ok": True, + "command": "optimization-plan", + "view": "presentation", + "confirmationRequired": True, + "presentation": plan.presentation_dict(), + } + ) + else: + simulation.emit_json(audit_payload) return 0 @@ -3789,6 +4052,20 @@ def build_parser() -> argparse.ArgumentParser: plan_parser.add_argument("input") plan_parser.add_argument("--spec", required=True) plan_parser.add_argument("--output-dir", required=True) + plan_parser.add_argument( + "--plan-file", + help=( + "Write the complete auditable plan to this new path outside the " + "optimization output directory; requires --present." + ), + ) + plan_parser.add_argument( + "--present", + action="store_true", + help=( + "Emit only the user-facing plan allowlist; requires --plan-file." + ), + ) plan_parser.set_defaults(handler=command_plan) optimize_parser = subparsers.add_parser( diff --git a/tests/test_system_optimization_skill.py b/tests/test_system_optimization_skill.py index c45711c..b5e563c 100644 --- a/tests/test_system_optimization_skill.py +++ b/tests/test_system_optimization_skill.py @@ -14,6 +14,7 @@ import hashlib import importlib.util import json import math +import stat import sys import tempfile import unittest @@ -107,9 +108,15 @@ def _json_bytes(value: object) -> bytes: def _valid_spec( *, seed: int = 12345, + population_size: int = 4, + mutation_factor: float = 0.8, + crossover_probability: float = 0.7, max_simulation_runs: int = 6, lower: float = 0.0, upper: float = 4.0, + relative_tolerance: float = 1e-12, + absolute_tolerance: float = 1e-12, + constraint_scale: float = 1.0, ) -> dict[str, object]: return { "optimizationSchemaVersion": 1, @@ -138,23 +145,23 @@ def _valid_spec( "lower": None, "upper": 4.0, "tolerance": 0.0, - "scale": 1.0, + "scale": constraint_scale, } ], "algorithm": { "name": "differentialEvolution", "seed": seed, - "populationSize": 4, - "mutationFactor": 0.8, - "crossoverProbability": 0.7, + "populationSize": population_size, + "mutationFactor": mutation_factor, + "crossoverProbability": crossover_probability, }, "budget": { "maxSimulationRuns": max_simulation_runs, "maxWallSeconds": 60.0, }, "validation": { - "relativeTolerance": 1e-12, - "absoluteTolerance": 1e-12, + "relativeTolerance": relative_tolerance, + "absoluteTolerance": absolute_tolerance, }, } @@ -207,9 +214,15 @@ def _make_plan( *, output_name: str = "optimization-output", seed: int = 12345, + population_size: int = 4, + mutation_factor: float = 0.8, + crossover_probability: float = 0.7, max_simulation_runs: int = 6, lower: float = 0.0, upper: float = 4.0, + relative_tolerance: float = 1e-12, + absolute_tolerance: float = 1e-12, + constraint_scale: float = 1.0, inspection: dict[str, object] | None = None, ) -> optimization.RuntimePlan: directory.mkdir(parents=True, exist_ok=True) @@ -217,9 +230,15 @@ def _make_plan( source_path.write_bytes(_json_bytes(PROJECT)) spec_payload = _valid_spec( seed=seed, + population_size=population_size, + mutation_factor=mutation_factor, + crossover_probability=crossover_probability, max_simulation_runs=max_simulation_runs, lower=lower, upper=upper, + relative_tolerance=relative_tolerance, + absolute_tolerance=absolute_tolerance, + constraint_scale=constraint_scale, ) spec_path = directory / "optimization-spec.json" spec_path.write_bytes(_json_bytes(spec_payload)) @@ -268,6 +287,18 @@ def _xml_parameter_values(xml: bytes) -> dict[tuple[str, str], str]: return values +def _recursive_mapping_keys(value: object) -> list[str]: + keys: list[str] = [] + if isinstance(value, dict): + for key, child in value.items(): + keys.append(str(key)) + keys.extend(_recursive_mapping_keys(child)) + elif isinstance(value, list): + for child in value: + keys.extend(_recursive_mapping_keys(child)) + return keys + + class OptimizationSpecTests(unittest.TestCase): def test_spec_rejects_unknown_fields_at_every_nested_contract(self) -> None: parsed = optimization.parse_optimization_spec(_valid_spec()) @@ -807,6 +838,458 @@ class CandidateIsolationTests(unittest.TestCase): self.assertEqual(plan.baseline_xml, BASELINE_XML) +class PlanPresentationTests(unittest.TestCase): + _FORBIDDEN_KEY_FRAGMENTS = ( + "sha256", + "hash", + "token", + "algorithm", + "seed", + "population", + "searchrunlimit", + "fullgenerations", + "requiredassertions", + "parametercontinuity", + ) + + @staticmethod + def _recommended_plan( + directory: Path, + *, + max_simulation_runs: int = 25, + output_name: str = "optimization-output", + inspection: dict[str, object] | None = None, + ) -> optimization.RuntimePlan: + return _make_plan( + directory, + output_name=output_name, + seed=optimization.RECOMMENDED_ALGORITHM_SEED, + population_size=optimization.RECOMMENDED_POPULATION_SIZE, + mutation_factor=optimization.RECOMMENDED_MUTATION_FACTOR, + crossover_probability=( + optimization.RECOMMENDED_CROSSOVER_PROBABILITY + ), + max_simulation_runs=max_simulation_runs, + relative_tolerance=( + optimization.RECOMMENDED_VALIDATION_RELATIVE_TOLERANCE + ), + absolute_tolerance=( + optimization.RECOMMENDED_VALIDATION_ABSOLUTE_TOLERANCE + ), + constraint_scale=4.0, + inspection=inspection, + ) + + @staticmethod + def _command_args( + plan: optimization.RuntimePlan, + *, + present: bool = False, + plan_file: Path | None = None, + ) -> argparse.Namespace: + return argparse.Namespace( + input=str(plan.source.path), + spec=str(plan.spec_source.path), + output_dir=str(plan.output_directory), + base_url=plan.base_url, + timeout=plan.timeout, + present=present, + plan_file=str(plan_file) if plan_file is not None else None, + ) + + def assert_safe_presentation(self, payload: object) -> None: + normalized_keys = { + "".join(character for character in key.casefold() if character.isalnum()) + for key in _recursive_mapping_keys(payload) + } + leaked = { + key + for key in normalized_keys + if any( + fragment in key for fragment in self._FORBIDDEN_KEY_FRAGMENTS + ) + } + self.assertEqual(leaked, set()) + + def test_presentation_is_a_closed_human_facing_view(self) -> None: + with tempfile.TemporaryDirectory() as directory_text: + inspection = _inspection() + result_variables = inspection["system"]["resultVariables"] + assert isinstance(result_variables, list) + for variable in result_variables: + assert isinstance(variable, dict) + variable["backendOnly"] = "BACKEND-METADATA-SENTINEL" + plan = self._recommended_plan( + Path(directory_text), + max_simulation_runs=17, + inspection=inspection, + ) + + presentation = plan.presentation_dict() + + self.assertEqual( + set(presentation), + { + "objective", + "designVariables", + "constraints", + "runLimits", + "assumptions", + "outputDirectory", + "sourceWillBeOverwritten", + }, + ) + objective = presentation["objective"] + self.assertIsInstance(objective, dict) + self.assertEqual(objective["resultKey"], "sensor.output") + objective_text = json.dumps(objective, ensure_ascii=False) + self.assertIn("final", objective_text) + self.assertIn("minimize", objective_text) + self.assertIn("m", objective_text) + + design_variables = presentation["designVariables"] + self.assertIsInstance(design_variables, list) + self.assertEqual(len(design_variables), 1) + design_variable = design_variables[0] + self.assertEqual( + set(design_variable), + { + "id", + "componentId", + "parameter", + "label", + "quantity", + "current", + "lower", + "upper", + "unit", + }, + ) + self.assertEqual(design_variable["id"], "gain") + self.assertEqual(design_variable["componentId"], "component-a") + self.assertEqual(design_variable["parameter"], "gain") + self.assertEqual(design_variable["current"], 2.0) + self.assertEqual(design_variable["lower"], 0.0) + self.assertEqual(design_variable["upper"], 4.0) + self.assertEqual(design_variable["unit"], "m") + + constraints = presentation["constraints"] + self.assertIsInstance(constraints, list) + self.assertEqual(len(constraints), 1) + constraint_text = json.dumps(constraints[0], ensure_ascii=False) + self.assertIn("sensor.limit", constraint_text) + self.assertIn("maximum", constraint_text) + self.assertIn("4.0", constraint_text) + + self.assertEqual( + presentation["runLimits"], + { + "maxSimulationBudgetSlots": 17, + "searchLaunchTimeLimitSeconds": 60.0, + "inFlightSimulationMayFinishAfterLimit": True, + "freshVerificationSlotsReserved": 1, + "verificationRunsOnlyIfFeasibleCandidateFound": True, + }, + ) + self.assertEqual( + presentation["assumptions"], + { + "continuousLinearSiDesignVariableIds": ["gain"], + "portsTopologyModesUnchanged": True, + }, + ) + self.assertEqual( + presentation["outputDirectory"], str(plan.output_directory) + ) + self.assertFalse(presentation["sourceWillBeOverwritten"]) + + self.assert_safe_presentation(presentation) + serialized = json.dumps(presentation, ensure_ascii=False) + for forbidden_value in ( + plan.source.sha256, + plan.spec_source.sha256, + plan.confirmation_token, + hashlib.sha256(plan.baseline_xml).hexdigest(), + "differentialEvolution", + "BACKEND-METADATA-SENTINEL", + plan.base_url, + ): + self.assertNotIn(forbidden_value, serialized) + + def test_presentation_only_adds_attention_when_budget_is_insufficient(self) -> None: + with tempfile.TemporaryDirectory() as directory_text: + root = Path(directory_text) + sufficient = self._recommended_plan( + root / "sufficient", max_simulation_runs=17 + ) + insufficient = self._recommended_plan( + root / "insufficient", max_simulation_runs=9 + ) + + sufficient_presentation = sufficient.presentation_dict() + insufficient_presentation = insufficient.presentation_dict() + + self.assertEqual(sufficient.public_dict()["warnings"], []) + self.assertNotIn("warnings", sufficient_presentation) + self.assertNotIn("attention", sufficient_presentation) + + self.assertEqual( + [item["code"] for item in insufficient.public_dict()["warnings"]], + ["OPTIMIZATION_BUDGET_INITIAL_POPULATION_ONLY"], + ) + self.assertNotIn("warnings", insufficient_presentation) + self.assertIn("attention", insufficient_presentation) + self.assertTrue(insufficient_presentation["attention"]) + attention_text = json.dumps( + insufficient_presentation["attention"], ensure_ascii=False + ) + self.assertNotIn( + "OPTIMIZATION_BUDGET_INITIAL_POPULATION_ONLY", attention_text + ) + self.assertNotIn("generation", attention_text.casefold()) + self.assert_safe_presentation(insufficient_presentation) + + def test_presentation_discloses_only_non_default_confirmed_settings( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory_text: + plan = _make_plan(Path(directory_text)) + + presentation = plan.presentation_dict() + + self.assertEqual( + presentation["nonDefaultSettings"], + { + "search": { + "randomSeed": 12345, + "populationSize": 4, + }, + "verification": { + "relativeTolerance": 1e-12, + "absoluteTolerance": 1e-12, + }, + "constraintRankingScales": [ + {"id": "limit", "scale": 1.0} + ], + }, + ) + serialized = json.dumps(presentation, ensure_ascii=False) + for forbidden_value in ( + plan.source.sha256, + plan.spec_source.sha256, + plan.confirmation_token, + plan.base_url, + "differentialEvolution", + ): + self.assertNotIn(forbidden_value, serialized) + + def test_command_plan_preserves_legacy_audit_stdout_without_new_flags( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory_text: + plan = _make_plan(Path(directory_text)) + args = self._command_args(plan) + with mock.patch.object( + optimization, "build_runtime_plan", return_value=plan + ), mock.patch.object( + optimization.simulation, "emit_json" + ) as emit_json: + result = optimization.command_plan(args) + + self.assertEqual(result, 0) + emit_json.assert_called_once_with(plan.public_dict()) + self.assertFalse(plan.output_directory.exists()) + + def test_command_plan_present_saves_audit_plan_and_emits_safe_wrapper( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory_text: + root = Path(directory_text) + plan = self._recommended_plan(root) + plan_file = root / "saved-optimization-plan.json" + args = self._command_args(plan, present=True, plan_file=plan_file) + with mock.patch.object( + optimization, "build_runtime_plan", return_value=plan + ), mock.patch.object( + optimization.simulation, "emit_json" + ) as emit_json, mock.patch.object( + optimization, + "_write_new_private_file", + wraps=optimization._write_new_private_file, + ) as write_new_file: + result = optimization.command_plan(args) + + self.assertEqual(result, 0) + write_new_file.assert_called_once() + written_path, written_bytes = write_new_file.call_args.args + self.assertEqual(written_path, plan_file) + self.assertEqual(json.loads(written_bytes), plan.public_dict()) + self.assertEqual( + json.loads(plan_file.read_text(encoding="utf-8")), + plan.public_dict(), + ) + self.assertEqual(stat.S_IMODE(plan_file.stat().st_mode), 0o600) + self.assertEqual( + list(root.glob(f".{plan_file.name}.*.tmp")), + [], + ) + expected_stdout = { + "ok": True, + "command": "optimization-plan", + "view": "presentation", + "confirmationRequired": True, + "presentation": plan.presentation_dict(), + } + emit_json.assert_called_once_with(expected_stdout) + self.assert_safe_presentation(expected_stdout["presentation"]) + + def test_command_plan_requires_present_and_plan_file_together(self) -> None: + with tempfile.TemporaryDirectory() as directory_text: + root = Path(directory_text) + plan = _make_plan(root) + cases = ( + self._command_args(plan, present=True), + self._command_args( + plan, + plan_file=root / "unpaired-plan.json", + ), + ) + for args in cases: + with self.subTest( + present=args.present, plan_file=args.plan_file + ), mock.patch.object( + optimization, "build_runtime_plan", return_value=plan + ) as build_runtime_plan, mock.patch.object( + optimization.simulation, "emit_json" + ) as emit_json: + with self.assertRaises(optimization.simulation.InputError): + optimization.command_plan(args) + build_runtime_plan.assert_not_called() + emit_json.assert_not_called() + self.assertFalse((root / "unpaired-plan.json").exists()) + + def test_command_plan_rejects_plan_file_inside_or_around_output_directory( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory_text: + root = Path(directory_text) + plan = _make_plan(root, output_name="reserved/output") + cases = { + "same": plan.output_directory, + "descendant": plan.output_directory / "plan.json", + "ancestor": plan.output_directory.parent, + } + for label, plan_file in cases.items(): + args = self._command_args( + plan, present=True, plan_file=plan_file + ) + with self.subTest(label=label), mock.patch.object( + optimization, "build_runtime_plan", return_value=plan + ), mock.patch.object( + optimization.simulation, "emit_json" + ) as emit_json: + with self.assertRaises(optimization.simulation.InputError): + optimization.command_plan(args) + emit_json.assert_not_called() + + def test_command_plan_never_overwrites_an_existing_plan_file(self) -> None: + with tempfile.TemporaryDirectory() as directory_text: + root = Path(directory_text) + plan = _make_plan(root) + plan_file = root / "existing-plan.json" + original = b"keep this exact file\n" + plan_file.write_bytes(original) + args = self._command_args(plan, present=True, plan_file=plan_file) + + with mock.patch.object( + optimization, "build_runtime_plan", return_value=plan + ), mock.patch.object( + optimization.simulation, "emit_json" + ) as emit_json: + with self.assertRaises(optimization.simulation.InputError): + optimization.command_plan(args) + + self.assertEqual(plan_file.read_bytes(), original) + emit_json.assert_not_called() + + def test_command_plan_rejects_a_broken_plan_file_symlink(self) -> None: + with tempfile.TemporaryDirectory() as directory_text: + root = Path(directory_text) + plan = _make_plan(root) + target = root / "missing-target.json" + plan_file = root / "plan-link.json" + plan_file.symlink_to(target) + args = self._command_args(plan, present=True, plan_file=plan_file) + + with mock.patch.object( + optimization, "build_runtime_plan", return_value=plan + ) as build_runtime_plan, mock.patch.object( + optimization.simulation, "emit_json" + ) as emit_json: + with self.assertRaises(optimization.simulation.InputError): + optimization.command_plan(args) + + build_runtime_plan.assert_not_called() + emit_json.assert_not_called() + self.assertTrue(plan_file.is_symlink()) + self.assertFalse(target.exists()) + + def test_command_plan_reports_a_parent_symlink_loop_as_input_error( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory_text: + root = Path(directory_text) + plan = _make_plan(root) + loop = root / "loop" + loop.symlink_to(loop.name) + plan_file = loop / "plan.json" + args = self._command_args(plan, present=True, plan_file=plan_file) + + with mock.patch.object( + optimization, "build_runtime_plan", return_value=plan + ) as build_runtime_plan, mock.patch.object( + optimization.simulation, "emit_json" + ) as emit_json: + with self.assertRaises( + optimization.simulation.InputError + ) as raised: + optimization.command_plan(args) + + self.assertEqual( + raised.exception.code, + "OPTIMIZATION_PLAN_FILE_PATH_INVALID", + ) + build_runtime_plan.assert_not_called() + emit_json.assert_not_called() + + def test_command_plan_does_not_emit_when_plan_file_write_fails(self) -> None: + with tempfile.TemporaryDirectory() as directory_text: + root = Path(directory_text) + plan = _make_plan(root) + plan_file = root / "unwritten-plan.json" + args = self._command_args(plan, present=True, plan_file=plan_file) + write_error = optimization.simulation.ArtifactError( + "OUTPUT_WRITE_FAILED", + "simulated exclusive write failure", + {"path": str(plan_file)}, + ) + + with mock.patch.object( + optimization, "build_runtime_plan", return_value=plan + ), mock.patch.object( + optimization, + "_write_new_private_file", + side_effect=write_error, + ), mock.patch.object( + optimization.simulation, "emit_json" + ) as emit_json: + with self.assertRaises(optimization.simulation.ArtifactError): + optimization.command_plan(args) + + emit_json.assert_not_called() + self.assertFalse(plan_file.exists()) + + class RankingAndConfirmationTests(unittest.TestCase): def test_plan_warns_when_budget_cannot_cover_a_complete_generation(self) -> None: with tempfile.TemporaryDirectory() as directory_text: