From 13f679084502fa1daf7e217cceaf6b0558fc1d91 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Sun, 4 Oct 2026 20:58:43 +0800 Subject: [PATCH] feat: evolve with one model and executable objectives, automatic adoption and rollback --- doc/architecture/SPEC/config.md | 4 + doc/architecture/SPEC/engine/executor.md | 4 + doc/architecture/SPEC/improve.md | 14 +- doc/architecture/SPEC/kb_service.md | 4 + doc/guide/self-evolution.md | 132 ++--- src/infermatrix_copilot/config.py | 5 + src/infermatrix_copilot/engine/executor.py | 8 +- src/infermatrix_copilot/improve/cli.py | 5 +- .../improve/coordinator.py | 29 +- src/infermatrix_copilot/improve/cycle.py | 7 +- src/infermatrix_copilot/improve/drivers.py | 16 +- src/infermatrix_copilot/improve/enroll.py | 9 + src/infermatrix_copilot/improve/evolution.py | 57 ++- .../improve/evolution_publish.py | 14 +- src/infermatrix_copilot/improve/isolation.py | 2 + src/infermatrix_copilot/improve/objectives.py | 426 ++++++++++++++++ src/infermatrix_copilot/improve/runtime.py | 408 ++++++++++++++++ src/infermatrix_copilot/improve/worker.py | 13 +- .../improve/workflows/workflow-improve.yaml | 12 +- src/infermatrix_copilot/kb_service/runtime.py | 29 +- test/test_autonomous_evolution.py | 460 ++++++++++++++++++ test/test_evolution.py | 2 +- 22 files changed, 1527 insertions(+), 133 deletions(-) create mode 100644 src/infermatrix_copilot/improve/objectives.py create mode 100644 src/infermatrix_copilot/improve/runtime.py create mode 100644 test/test_autonomous_evolution.py diff --git a/doc/architecture/SPEC/config.md b/doc/architecture/SPEC/config.md index 7782ead7..e381657f 100644 --- a/doc/architecture/SPEC/config.md +++ b/doc/architecture/SPEC/config.md @@ -95,3 +95,7 @@ provider 的缺省模型必须在**这一处**解析，否则 `ResolvedTarget.mo ## 自进化接入（2026-10-04） `TRACE_STORE_ROOT` 控制 trace/1 全量采集。`IMPROVE_ENABLED=false` 是总停止开关，`IMPROVE_EVOLVE_ENABLED=false` 控制代码进化；源码、数据和发布 outbox 各有独立目录。周预算默认 20 美元，周度槽位默认周一 05:00 UTC。`PR_CONTEXT_SOURCE=snapshot` 与 `IMPROVE_SHADOW` 由可信实验控制器设置；`KB_DRAFT_MAX_OPERATIONS` 默认 6。 + +## 自主进化接入 + +自主进化新增 `improve_evaluation_mode=objective`、`improve_promotion_mode=automatic`、`improve_evolve_model`；固定单模型，无需判官。总开关与周预算默认不变。 diff --git a/doc/architecture/SPEC/engine/executor.md b/doc/architecture/SPEC/engine/executor.md index 6a22a028..d5b8a0c1 100644 --- a/doc/architecture/SPEC/engine/executor.md +++ b/doc/architecture/SPEC/engine/executor.md @@ -58,3 +58,7 @@ helper：`_eval_when`、`_merge`。 ## 自进化接入（2026-10-04） 配置 trace 根时为整次执行绑定 store，并为每步绑定工作流、item、unit_id 和配置指纹；终局 `step_result` 保存实际报告与发现。影子运行在进入步骤前拒绝非 read/report 风险。已采用的配置覆盖按工作流限定，同时复制 LLM 的 settings，采集指纹与实际执行配置一致。 + +## 自主进化接入 + +启用自主进化时，原生 PR 评审先捕获固定 Git 对象／知识快照；已采用的制品在可信代理的沙箱中执行。安全快照不可得时记录原因；活动制品自动恢复上一版并阻断本次调用。 diff --git a/doc/architecture/SPEC/improve.md b/doc/architecture/SPEC/improve.md index 7870aa2c..491da976 100644 --- a/doc/architecture/SPEC/improve.md +++ b/doc/architecture/SPEC/improve.md @@ -5,11 +5,19 @@ `设计：/data/zhoutaichang/copilot/meta-improvement-engine-design.md v1（GPT-6 sol 批准 2026-09-29） · refactor-status: building (P0–P4 已落地)` ## 职责 -面向任意 trace/1 工作流的"取证—候选—实验—人审 PR—上线观察"循环。使用入口和数据契约见 [自进化使用指南](../../guide/self-evolution.md)。引擎的写权限限于追加自己的 trace/1 记录、 -写自己的账本和源码制品目录、在 OS 隔离环境跑候选；产线变更一律走人审 PR。 +面向任意 trace/1 工作流的"取证—候选—可信目标实验—自动采用—上线观察／自动回滚"循环。使用入口和数据契约见 [自进化使用指南](../../guide/self-evolution.md)。引擎的写权限限于追加自己的 trace/1 记录、 +写自己的账本和源码制品目录、在 OS 隔离环境跑候选；自主模式仅采用已评估制品，并在无凭据沙箱执行；生产写权限仍归宿主。PR 是可选审计，旧金标／人审模式保留。 新增 `artifacts.py`（不可变源码及补丁）、`isolation.py/worker.py`（无凭据沙箱与可信预算代理）、`drivers.py`（PR/知识/元基准驱动及父进程评分）、`evolution.py`（周度候选状态机）、`coordinator.py`（统一续跑周期）、`annotations.py`（历史 trace 导出与人工标签导入）、`evolution_publish.py`（独立 `evolve-outbox/1` 与部署跟踪）。`IMPROVE_EVOLVE_ENABLED` 默认关闭，`IMPROVE_ENABLED` 保留总停止作用。 +## 自主目标与制品运行时 + +`objectives.py` 是受保护的父进程目标控制器（`objective/1`）：自动导入可信 replay 输入、生成有因果见证的故障／负例、固定一个 API 模型、预注册三次配对及 item 聚类功效、核验实际源码与输入、独立计算分数及护栏。见证不传生成器或候选；候选数值分数无效。PR 与知识指标仅证明结构契约、产物保留与资源收益，`semantic_quality_claim=false`。引擎候选每次只修改归因或 lint；lint 既有正例及负例受保护。功效不足会存档并等待新鲜样本，不重放完成的付费调用。 + +`runtime.py` 重新应用补丁核验实际评估树，原子采用本地不可变 release，原生 PR 评审、知识草稿、lint 和归因在活动制品的沙箱中运行。宿主记录运行指纹才更新部署基线。首次八个不同 item 与上一版比较，七日缺证据自动恢复；保留后每周最多八个新 item 持续比较。契约、凭据／隔离／预算、指纹失败或资源显著退化自动恢复上一版；恢复制品不可验证则禁用执行。执行／切换锁及持久化激活、回滚意图支持重启恢复。 + +默认 `IMPROVE_EVALUATION_MODE=objective`、`IMPROVE_PROMOTION_MODE=automatic`，不要求人工标签、判官或 performance 模型。总开关仍关闭，固定模型取 `IMPROVE_EVOLVE_MODEL` 或 eco；每周一个候选、一次修复及共享 20 美元预算不变。`gold/pr` 模式保留原有双家族取证及人工流程。无隔离生产驱动的机械工作流仍可 lint，但不自动采用。当前主机命名空间关闭，代码候选实际执行延期。 + ## 模块 | 文件 | 阶段 | 作用 | |---|---|---| @@ -56,4 +64,4 @@ - lint 只读 trace，不调模型；无证据形态的 lint 不猜。 ## 测试 -`test_improve_p0.py`、`test_improve_p0b.py`、`test_improve_p1.py`、`test_improve_p2.py`、`test_improve_p3.py`、`test_improve_p4.py`、`test_improve_kb_intake_adapter.py`、`test_evolution.py`。 +`test_improve_p0.py`、`test_improve_p0b.py`、`test_improve_p1.py`、`test_improve_p2.py`、`test_improve_p3.py`、`test_improve_p4.py`、`test_improve_kb_intake_adapter.py`、`test_evolution.py`、`test_autonomous_evolution.py`。 diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 53bd61bf..0bcde4ee 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -772,3 +772,7 @@ dry run 为旧快照建本地临时 baseline，再生成相同升级串与全量 ## 自进化接入（2026-10-04） 起草单元携带 `kb-intake.draft` 的 item、workflow 和配置指纹；起草使用该工作流已采用的最大操作数覆盖。调度器在同一服务租约内检查周度改进槽位；进化启用时调用可续跑协调器，未完成检查点优先续跑，异常单独记录。 + +## 自主进化接入 + +自主进化开启时起草前捕获事件与知识快照；已采用制品通过隔离代理起草 typed operations，宿主沿用已有知识发布权限与门禁。进化实验只在内存应用，不写产品知识库。 diff --git a/doc/guide/self-evolution.md b/doc/guide/self-evolution.md index 3ecceb84..71e06f69 100644 --- a/doc/guide/self-evolution.md +++ b/doc/guide/self-evolution.md @@ -1,144 +1,90 @@ # 用自进化引擎改进 Copilot 工作流 -引擎根据 trace 中的损失提出一个机制假设，生成允许范围内的源码或配置补丁，再用冻结输入比较两个独立源码制品。通过可信回归检查和预注册评估后，它准备 PR 草稿，由人类评审和合并。相同机制用于 PR 评审、知识摄取及引擎自身。 +默认评估模式为 `objective`，采用模式为 `automatic`。开启进化后，系统自行收集输入、提出假设、生成补丁、运行隔离实验、采用通过评估的源码制品，并观察生产调用、自动回滚。整个闭环不要求人工标注、人工批准或更强的判官模型。生成、归因和业务执行复用一个固定 API 模型。 -默认关闭进化。每周最多生成一个候选，开发检查失败最多修复一次；所有工作流共用既有每周 20 美元预算。预算、样本或隔离不足会延期，CLI 退出码为 3；输入损坏等命令错误退出码为 1。不会自动提高预算。 +进化总开关仍默认关闭。每周最多一个新候选，开发检查失败最多修复一次。候选生成、实验、隔离生产请求及影子比较共用每周 20 美元预算；不足时延期或恢复上一版，不自行增加预算。 ## 开始使用 -先从干净、已提交的 Copilot 源码检出运行。建议在该检出中创建专用环境，安装源码与仓库声明的可选依赖： +从干净、已提交的 Copilot 源码运行，安装仓库的固定依赖： ```bash python -m venv .venv source .venv/bin/activate python -m pip install -e '.[dev,mcp,kb]' -``` - -源码制品只包含 `src`、`playbooks`、`adapters`、`skills`，不包含产品知识库、评估标签或账本。使用当前 Python 环境的固定依赖版本；依赖指纹变化会拒绝执行。 -```bash export IMPROVE_ENABLED=true export IMPROVE_EVOLVE_ENABLED=true export IMPROVE_EVOLVE_SOURCE_DIR="$PWD" -export IMPROVE_EVOLVE_DATA_DIR=/srv/copilot/evolution-data export IMPROVE_LEDGER_DIR=/srv/copilot/improve-ledger export TRACE_STORE_ROOT=/srv/copilot/traces -# 复用现有 provider 配置，显式固定两条 API 路由；比较实验还需固定判官。 -export ECO_MODEL=YOUR_PINNED_GENERATOR_MODEL -export PERFORMANCE_MODEL=YOUR_PINNED_INVESTIGATOR_MODEL -export IMPROVE_JUDGE=api:YOUR_PINNED_JUDGE_MODEL +# 沿用现有 provider 凭据与一个固定模型，无需设置判官或 performance 模型。 +export ECO_MODEL=YOUR_PINNED_API_MODEL +# 可选：明确指定同一个进化模型。 +# export IMPROVE_EVOLVE_MODEL=YOUR_PINNED_API_MODEL infermatrix-copilot improve workflows list -infermatrix-copilot improve workflows check --workflow pr-review.agent.review_diff +infermatrix-copilot improve workflows prepare --workflow all +infermatrix-copilot improve workflows check --workflow workflow-improve.improve.forensics +infermatrix-copilot improve evolve run --workflow workflow-improve.improve.forensics infermatrix-copilot improve evolve run --workflow pr-review.agent.review_diff infermatrix-copilot improve evolve run --workflow kb-intake.draft -infermatrix-copilot improve evolve run --workflow workflow-improve.improve.forensics infermatrix-copilot improve evolve run --workflow all --repo OWNER/REPO infermatrix-copilot improve evolve list infermatrix-copilot improve evolve show CANDIDATE_ID ``` -`workflows check` 报告 trace、完整指纹、开发输入、新鲜留出样本、人工标注／策展金标、可信测试、成本估计及沙箱缺项。已具备结果适配器的工作流也可通过 `mechanical_readiness` 独立检查机械修复资格，不需要质量留出集。产生 trace 的未声明工作流仍被周度 lint 检查；比较质量需要额外接入契约。 - -本机 `max_user_namespaces=0`，因此检查会报告 `sandbox-unavailable`。隔离依赖 Linux user namespace 与 bubblewrap；恢复主机隔离能力后须重新检查。程序不会修改内核配置，不会回退到无沙箱执行。 - -## 冻结实验输入 - -每个工作流使用 `//dataset.json`，输入 JSON 为该目录下的相对路径。输入文件的 SHA-256 和完整条目元数据共同形成快照版本。 - -```json -{ - "workflow": "kb-intake.draft", - "items": [ - { - "item": "OWNER/REPO#123", - "split": "development", - "input": "event-123.json", - "sha256": "INPUT_FILE_SHA256" - }, - { - "item": "OWNER/REPO#124", - "split": "holdout", - "input": "event-124.json", - "sha256": "INPUT_FILE_SHA256", - "gold": { - "entries": [{"gold_id": "rule-1", "path": "repos/demo/core/rules.md", "concern": "人工核实的完整规则契约"}] - } - } - ] -} -``` - -输入不得携带 `labels`、`gold`、`scores`、`judgments` 或 `report` 字段。可信父进程持有留出标签和金标，只向隔离工作进程传入业务输入。生成器只看到开发样本、损失证据和允许修改的源码。晋级留出集被使用后记录到 `evolution/holdouts.json`，下一候选必须使用新样本；修改标签或输入会使已注册实验失效。 +`prepare` 自动导入已有 replay trace，并生成引擎自身的受控故障及无故障对照。业务样本只能来自真实输入，不用模型编造标签。`evolve run` 经完整周期执行这些阶段；重复调用会刷新免费输入采集并继续已有候选，不重复完成的付费调用。`--repo` 只选择指定仓库的业务证据与输入。 -| 工作流 | 输入 JSON 内容 | 父进程验证 | -|---|---|---| -| PR 评审 | `repo`、`pr`、原始 `base_sha/head_sha`、完整 `base_files/head_files`、`diff`、`context_text`、`gate_report`、`knowledge_files` | 在临时仓库重建固定两个树，执行实际评审步骤；盲配对判官计算召回与精确率。原始 SHA 是快照身份，重建仓库的提交 SHA 用于本地源码工具 | -| 知识摄取 | `repo`、`repo_dir`、`event_id`、`generator_model`、`evidence`、`files` 知识快照、`release`、`today`、可选 `external_texts` | 工作进程仅起草 typed operations；父进程在内存树运行生产知识操作与质量门，金标覆盖只计通过的规则，不写生产知识库；`generator_model` 必须与可信 API 代理的 eco 模型一致 | -| 引擎归因 | `records`、`blobs`（引用到文本映射）、`concerns`、`cells` | 条目另存 `labels` 与 `label_source: human`；完整人工标注分母，缺失、弃答、分歧和无效证据均记错 | +`check` 报告固定 API 路由、冻结输入、测试、源码、预算、沙箱和功效样本缺项。CLI 延期退出码为 3，命令错误为 1。未配置固定模型时明确延期。混合模型与 harness 路由尚未接入此代理，不会静默调用其他模型。 -在预注册前按 item 的固定排序选择达到功效要求的新鲜样本，避免把整个留出池都计入一周预算。每 item 固定三次成对运行；只有三对完整的 item 进入聚类统计。主要指标必须 `supported` 且达到最小效果；精确率、有效规则覆盖及无效规则数的显著退化会阻止晋级。成本也有保护指标。评分不采信工作进程返回的数字。隔离、源码或导入路径核验失败使实验无效。 +当前主机 `max_user_namespaces=0`。恢复 Linux user namespace 与 bubblewrap 隔离前，只能准备数据、查看证据和执行可信测试；生成的代码不会执行，也不会采用。程序不会修改主机内核配置或回退到无沙箱执行。 -默认最小效果为 0.05。功效计算可能要求几十个新鲜 item；在缺少历史成本时，保守估计可能超过 20 美元。应先收集真实调用成本并补齐样本，不能通过降低可信保护或自行提高预算跳过检查。声明可提供由维护者核实的 `cost_per_unit_usd` 估计，实际请求始终受共享预算限制。 +## 系统如何判断改进 -## 历史 trace 与人工标注 +没有独立证据时，模型的自我评价不能证明进步。这里用可信父进程持有的可执行目标代替金标与判官；候选只返回业务产物，不能修改评分、测试、预算、隔离、发布权限或控制器。 -当前仓库的元基准目录没有已提交的真实案例。工具不会生成或猜测人工标签。先人工策展真实 item 的金标，再导出完整历史单元： +| 工作流 | 自动得到的实验输入 | 可证明的收益与保护 | +|---|---|---| +| 引擎归因／lint | 控制器注入输出截断、候选丢弃、规划回退，随机化记录标识与参数；同时保留无故障对照 | 对已知干预正确归因并引用实际证据，或正确检出对应 lint；漏答计错，负例误报计错，既有正例不得丢失。每次只改归因或 lint，避免未评估的混合变更 | +| PR 评审 | 原生步骤冻结 base/head Git 树、diff、上下文、已有发现和知识快照，写入 replay trace | 非空报告，文件／行号／证据契约正确；有效基线上的发现、处置、复查与 verdict 必须保留；比较成本或耗时。不把结构通过率当作真实问题召回率 | +| 知识草稿 | 原生起草冻结事件、证据、知识文件、版本与生成模型 | typed operations 在内存知识树可应用，范围及保护路径正确；已有有效操作不得丢失，空草稿可以正确。不奖励编造规则，不宣称新规则语义正确；实验不写生产知识库 | +| CI debug、rebase、issue 回答等 | 既有 trace 及可信回归测试 | 持续获得 Tier 1 机械检查。已有测试的基线失败／候选通过可证明限定缺陷修复；尚无隔离生产驱动的工作流不能自动采用，不宣称比较质量提升 | -```bash -infermatrix-copilot improve meta export --case incident-001 \ - --unit-id HISTORICAL_UNIT_ID --gold-file /srv/curated/item.gold.json \ - --meta-dir /srv/copilot/human-meta --trace-root /srv/copilot/traces +引擎默认生成受控案例，干预见证只在父进程清单中。生成器只能看到开发样本、trace 证据和允许源码；工作进程只收到冻结业务输入，不能访问晋级案例的见证或报告。候选返回的分数被忽略，费用、调用数、耗时、源码与导入路径由可信控制器核验。 -# 人工检查 trace 后写 labels.json，例如 {"gold-id": "S2"}；阶段定义见 forensics.STAGES。 -infermatrix-copilot improve meta annotate --case incident-001 \ - --labels-file /srv/curated/labels.json --human-verified --split holdout \ - --meta-dir /srv/copilot/human-meta +默认目标为契约通过率；效率提案使用资源收益。每 item 三次配对，按 item 聚类计算置信区间，提升必须获得 `supported` 且达到预注册最小效果（默认 0.05）。已有契约及输出不得退化。若实际方差使实验功效不足，保留旧实验，用新鲜独立样本注册下一次；不重新评分已消费的留出集，也不重复生成补丁。受控场景可自动补充，真实业务样本须等待 trace。 -export IMPROVE_META_DIR=/srv/copilot/human-meta -infermatrix-copilot improve workflows import-meta --data-dir /srv/copilot/evolution-data -``` +上述结论只覆盖可执行契约、受控故障和资源收益。没有可验证结果信号时，系统记录限制，不声称任意 PR 召回率或知识语义质量变好。生产知识发布仍沿用宿主已有权限及质量门；进化评分本身不调用该门的模型判官，也不授权候选写知识库。 -开发集与留出集须由人工分别安排。未指定 split 的旧案例默认进入开发集；空标注不满足晋级条件。知识摄取及 PR 的输入 JSON 也需由维护者从固定事件、源码和知识快照导入，金标只写在父进程清单中。 +## 输入、声明与可信边界 -## 工作流声明与机械修复 +默认输入位于 `/evolution/inputs//dataset.json`，可用 `IMPROVE_EVOLVE_DATA_DIR` 指定其他目录。每个输入文件、条目元数据和源码制品都有内容指纹。重复业务 item 归为同一聚类；旧 trace 缺少完整输入时不会猜测快照。 -扩展声明位于 `src/infermatrix_copilot/improve/workflows/*.yaml`，保留既有工作单元、item、fingerprint、capture 与 outcome_adapter，新增 `experiment_driver` 和 `evolution`：允许的 `paths/settings`、可信 `tests`、主要 `metric/direction/min_effect`、保护 `guards`、共享源码 `dependencies`。候选改动共享运行时后，受影响工作流的测试合并执行并记在候选中。 +PR 重放读取固定 Git 对象，拒绝符号链接、凭据文件、非 UTF-8 或超限文件；知识输入冻结当前生成模型，必须与代理固定模型一致。模型输出与输入都不携带生产凭据。候选在无网络、受限文件系统的沙箱执行，模型请求只能经过宿主代理及共享预算。 -首版 API 代理接受固定 eco/performance 两条模型路由；使用混合成员、harness lens 或其他专用评审模型的配置会报告代理驱动缺项，不能静默改成默认路由。并发评审请求经代理串行预算结算。 +工作流声明在 `src/infermatrix_copilot/improve/workflows/*.yaml`。`evolution` 指定可改路径、配置键与可信测试；共享代码变更合并所有受影响工作流的回归检查。可选 `objective` 仅接受 `metric: contract_success|resource_gain`、`min_effect`、`prior_sd`；这类控制策略不在候选允许修改的范围内。 -当前可信驱动为 `pr-review`、`kb-intake`、`meta` 和 `mechanical`。新增比较驱动需要维护者实现并审查父进程的快照、预测和评分契约，再通过声明选择；未知驱动不会回退成 PR 评审。 +失败候选、修复尝试、补丁及报告保留在账本旁，不进入产品知识库。同一失败补丁不能通过换提案重试。结束的候选自动关闭对应提案；中断的生成不会重发，下周可继续提出新的改动。晋级使用的新鲜留出集记录在 `evolution/holdouts.json`。 -CI debug、rebase 和 issue 回答已声明机械改动范围。无结果适配器时，在 dataset.json 额外提供 `reproducer`（Python 测试文件相对路径）及 `reproducer_sha256`。该可信复现测试必须在基线返回 pytest 失败码 1，在候选返回 0，且工作流回归检查通过。产物明确写 `quality_claim: false`。CI debug 和 rebase 首版不运行比较质量驱动，不在生产仓库执行修改或推送。 +## 自动采用、观察与回滚 -评分、基准、预算、隔离、发布、控制器及权限源码受保护。引擎自身只能改归因、lint 和声明允许的模型配置。失败候选、每次尝试的补丁、源码制品和测试报告会保留。同一失败提案不再次自动生成；同一基线上的已失败补丁也不会以新提案重复执行。 +通过测试及可信实验后，控制器重新应用允许范围内的补丁，核验与实际评估制品相同，然后把它复制为不可变本地 release。`evolution/active.json` 原子记录当前指针、上一版和谱系；不必创建或合并 PR。 -## 周期续跑、发布和上线观察 +Copilot 原生评审、知识起草、周期 lint 与引擎诊断会在隔离进程执行活动制品。宿主仍控制生产写操作。真实执行 trace 确认源码指纹后才记录部署基线；指针激活本身不是部署证明。原始 Git 检出不会被自动改写。 -`improve cycle`、`workflow-improve` playbook 和 `kb serve` 周度槽位共用协调器：lint → 已注册实验 → 取证 → 候选生成／验证／实验 → 草稿。阶段及每一对实验均落盘；重启跳过完成阶段，中断的付费调用不自动重发。每周最多一个新候选；有待评审候选的工作流直接返回已有候选。 +首次试运行对同一冻结输入运行上一版，至少八个不同 item 的有效配对才能标记 `retained`。输出契约失败、已有有效产物变化、制品被篡改、隔离或预算失效，立即恢复确切上一版并阻断本次调用。七天没有足够试运行证据也自动恢复。保留后，每周最多八个新 item 继续配对；统计显著的资源退化触发回滚。无法验证上一版时关闭活动制品，不执行未知源码。 -候选目录在 `/evolution/candidates//`，含 `candidate.json`、`incumbent/`、`arm/`、`candidate.patch`、`PR.md`、`bundle.json`；配对进度、判定、源码／依赖哈希、测试报告、失败原因和谱系都可检查。 +关闭 `IMPROVE_ENABLED` 或 `IMPROVE_EVOLVE_ENABLED` 后，原生流程停止使用进化制品。候选目录 `/evolution/candidates//` 包含补丁、两个源码制品、测试报告、`objective.json`、历次实验、候选状态和 `PR.md` 审计草稿；活动制品在 `evolution/releases/`。续跑、并发与采用操作使用锁和持久化阶段进度，中断的付费调用不自动重发。 -请求发布： - -```bash -export ALLOW_POST=true -export ALLOW_PUSH=true -export IMPROVE_PROPOSAL_REPO=OWNER/InferMatrixCopilot -export IMPROVE_EVOLVE_OUTBOX_DIR=/srv/copilot/evolution-outbox -infermatrix-copilot improve evolve run --workflow pr-review.agent.review_diff --post - -# omni-maintainer，使用与候选基线相同且可信的源码仓库 -omni-maintainer evolve --outbox /srv/copilot/evolution-outbox \ - --source /srv/repos/InferMatrixCopilot -``` +## 可选人工评估与 PR 审计 -新协议为 `evolve-outbox/1`，既有 `improve-outbox/1` 提案协议保留。maintainer 默认 `phase.evolution_prs_live=false`；只有维护者按仓库策略开启后才推送并创建草稿 PR。它核验实际补丁对应的源码树，拒绝过期主分支、受保护文件和哈希不符，重试复用同一候选分支／PR。进化 PR 禁止 arbiter 自主晋级，仍需人类 go。 +旧金标流程保留供主动选择：`IMPROVE_EVALUATION_MODE=gold`、`IMPROVE_PROMOTION_MODE=pr`。此模式才要求人工标签、策展金标及相应判官。旧 `meta export/annotate` 与 `workflows import-meta` 命令保持兼容；自主模式无需使用。 -PR 合并只标记 `merged`。生产 trace 出现候选源码树 SHA，且采用配置与指纹清单一致后，才标记 `deployed` 并记录账本部署基线。下一周期读取生产样本及已有可信结果；不足八个共同 item 或缺少结果时报告等待，不宣称上线收益。有显著退化则生成回滚建议，由人类决定执行。 +`--post` 可额外生成 `evolve-outbox/1` 审计动作，仍遵守 `ALLOW_POST`、`ALLOW_PUSH` 和 maintainer 策略；不会绕过 GitHub 仓库发布权限。自主采用不等待这一审计 PR。旧 `improve-outbox/1` 与手动 PR 合并／部署跟踪保留。 ## 验证范围 -`test/test_evolution.py` 包括三个驱动的离线闭环（脚本化模型／工作进程）、可信 PR 与知识评分、完整归因分母、受保护改动、指纹造假、续跑、并发去重、预算拒绝、一次修复、机械复现及合并未部署。真实 OS 沙箱测试在主机不可用时明确 skip。maintainer 测试覆盖补丁实际应用、哈希、过期基线和发布去重。 +`test/test_autonomous_evolution.py` 验证无人工／无判官的三个驱动闭环、可信评分、受控故障与负例、固定模型、实际制品采用、指纹篡改、自动恢复、观察、预算／隔离阻断和中断续跑。测试使用脚本化模型和工作进程来验证协议与状态机；真实 OS 沙箱不可用时相应测试明确跳过。 -离线闭环能检验协议与状态机；真实模型、真实人工基准及隔离执行的三个端到端验收，仍须在具备沙箱和评估数据的主机完成。 +当前主机尚未执行真实模型生成代码的生产闭环。上线前必须在具备隔离、真实业务输入及现有预算的主机完成实际运行；脚本化测试不能替代这一证明。 diff --git a/src/infermatrix_copilot/config.py b/src/infermatrix_copilot/config.py index 4d9fc655..ac86fc3e 100644 --- a/src/infermatrix_copilot/config.py +++ b/src/infermatrix_copilot/config.py @@ -589,6 +589,11 @@ def _parse_mixture(cls, v): # improve_ledger_dir (default ~/.infermatrix-copilot/improve). improve_enabled: bool = False improve_evolve_enabled: bool = False + # Autonomous evaluations use executable contracts and one pinned API model. + # "gold" retains the optional P0–P4 semantic benchmark path. + improve_evaluation_mode: Literal["objective", "gold"] = "objective" + improve_promotion_mode: Literal["automatic", "pr"] = "automatic" + improve_evolve_model: str = "" improve_evolve_source_dir: str = "" improve_evolve_data_dir: str = "" improve_evolve_outbox_dir: str = "" diff --git a/src/infermatrix_copilot/engine/executor.py b/src/infermatrix_copilot/engine/executor.py index fccc0260..7a3841b1 100644 --- a/src/infermatrix_copilot/engine/executor.py +++ b/src/infermatrix_copilot/engine/executor.py @@ -278,7 +278,13 @@ async def _run_step(self, spec, params: dict, state: dict, item, try: with tracing.span("step", step=spec.name, attempt=attempt, **ident), \ trace_context(attempt=attempt, **unit): - last = await spec.handler(ctx) + from ..improve import objectives + last = None + if spec.name == "agent.review_diff" and objectives.enabled(ctx.settings) and ctx.settings.improve_enabled and ctx.settings.improve_evolve_enabled: + from ..improve.runtime import review_step + last = await review_step(ctx) + if last is None: + last = await spec.handler(ctx) except Exception as exc: # handler bug != typed failure last = StepResult(False, FailureKind.BLOCKED, f"unhandled error: {type(exc).__name__}: {exc}") diff --git a/src/infermatrix_copilot/improve/cli.py b/src/infermatrix_copilot/improve/cli.py index b25bd5a4..f0b46f5d 100644 --- a/src/infermatrix_copilot/improve/cli.py +++ b/src/infermatrix_copilot/improve/cli.py @@ -108,7 +108,7 @@ def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="infermatrix-copilot improve", description="operate the meta-improvement engine's trace index") sub = parser.add_subparsers(dest="command", required=True) - for name, choices in (("workflows", ["list", "check", "import-meta"]), ("evolve", ["run", "list", "show"])): + for name, choices in (("workflows", ["list", "check", "prepare", "import-meta"]), ("evolve", ["run", "list", "show"])): p = sub.add_parser(name) p.add_argument("action", choices=choices) p.add_argument("id", nargs="?", default="") @@ -193,6 +193,9 @@ def main(argv: list[str] | None = None) -> int: if args.command == "workflows": if args.action == "list": result = evolution.list_workflows(settings, store) elif args.action == "check": result = evolution.check(settings, store, args.workflow, repo=args.repo) + elif args.action == "prepare": + from .objectives import prepare + result = prepare(settings, store, workflow=args.workflow, repo=args.repo) else: from .drivers import export_meta if not settings.improve_evolve_data_dir: raise ValueError("import-meta needs --data-dir or IMPROVE_EVOLVE_DATA_DIR") diff --git a/src/infermatrix_copilot/improve/coordinator.py b/src/infermatrix_copilot/improve/coordinator.py index f54d4a98..0b79977c 100644 --- a/src/infermatrix_copilot/improve/coordinator.py +++ b/src/infermatrix_copilot/improve/coordinator.py @@ -47,7 +47,34 @@ def run(settings, store, *, workflow="all", repo="", post=False, now=None, llm=N state["stages"]["experiments"] = completed atomic_json(path, state) forensic_key = "forensics" if workflow == "all" and not repo else f"forensics:{workflow}:{repo}" - if forensic_key not in state["stages"]: + from . import objectives + if objectives.enabled(settings): + # No semantic gold matrix, cross-family investigator, or judge. + # Controller scenarios and trace replay supply falsifiable goals. + # Input collection is free and must refresh while a candidate waits. + state["stages"]["objective_inputs"] = objectives.prepare(settings, store, workflow=workflow, repo=repo) + from .ledger import Ledger + ledger = Ledger(root, store) + completed_candidates = evolution.candidates(settings) + attempted_proposals = {c.get("proposal") for c in completed_candidates} + for completed in completed_candidates: + if completed["state"] in evolution.TERMINAL: + for proposal in ledger.load(completed["workflow"]).proposals: + if proposal.id == completed.get("proposal") and proposal.state in ("open", "experiment-registered", "neutral", "underpowered"): + ledger.transition(completed["workflow"], proposal.id, "closed", candidate=completed["id"], outcome=completed["state"]) + for name in state["stages"]["objective_inputs"]: + if not any(p.state == "open" and p.id not in attempted_proposals for p in ledger.load(name).proposals): + records = store.query(workflow=name, limit=3) + if records: + ledger.open_proposal(name, tier=1, stage="efficiency", claim="Reduce resource use while preserving executable contracts and existing outputs", evidence=[r["id"] for r in records], loss=1) + atomic_json(path, state) + if forensic_key not in state["stages"]: + from .runtime import diagnose + state["stages"][forensic_key] = {"ok": False, "summary": "diagnosis-interrupted; no paid-call replay"} + atomic_json(path, state) + state["stages"][forensic_key] = {"ok": True, "summary": "program-verified hypotheses; no human labels or evaluator model", "outputs": diagnose(settings, store, llm)} + atomic_json(path, state) + elif forensic_key not in state["stages"]: # Reuse the actual playbook handler; expose the original lint window. from ..engine.step import StepContext from ..engine.steps.improve import _forensics diff --git a/src/infermatrix_copilot/improve/cycle.py b/src/infermatrix_copilot/improve/cycle.py index 7ce4ecd1..90140e2e 100644 --- a/src/infermatrix_copilot/improve/cycle.py +++ b/src/infermatrix_copilot/improve/cycle.py @@ -193,7 +193,12 @@ def _run_cycle_locked(store: TraceStore, settings: Any, ledger_dir: Path, *, now settings=settings) usd_samples, seconds_samples, parse_failed, parse_total = [], [], 0, 0 for unit in stats.units: - findings = run_lints(unit, store, baseline) + from . import objectives + if objectives.enabled(settings) and settings.improve_evolve_enabled: + from .runtime import run_lints as runtime_lints + findings = runtime_lints(unit, store, baseline) + else: + findings = run_lints(unit, store, baseline) stats.findings.extend(findings) if any(f.quarantine for f in findings): stats.quarantined.append(unit.unit_id) diff --git a/src/infermatrix_copilot/improve/drivers.py b/src/infermatrix_copilot/improve/drivers.py index c0f4a0ea..d4861091 100644 --- a/src/infermatrix_copilot/improve/drivers.py +++ b/src/infermatrix_copilot/improve/drivers.py @@ -12,6 +12,14 @@ from .artifacts import ArtifactError, digest, safe_path +def input_path(raw): + """Read-only replay files can include .github; never overwrite Git metadata.""" + from pathlib import PurePosixPath + p = PurePosixPath(raw) + if not raw or p.is_absolute() or ".." in p.parts or ".git" in p.parts or "\\" in raw or any(ord(c) < 32 for c in raw): + raise ArtifactError("unsafe replay path") + return p.as_posix() + def dataset(settings, workflow: str) -> list[dict]: root = Path(settings.improve_evolve_data_dir).expanduser() if settings.improve_evolve_data_dir else None if root is None: @@ -231,13 +239,14 @@ def git(*args): for p in repo.rglob("*"): if p.is_file() and ".git" not in p.parts: p.unlink() for name, content in files.items(): - p = repo / safe_path(name); p.parent.mkdir(parents=True, exist_ok=True); p.write_text(content) + p = repo / input_path(name); p.parent.mkdir(parents=True, exist_ok=True); p.write_text(content) git("add", "-A"); git("commit", "--allow-empty", "-qm", "frozen input") revisions.append(subprocess.check_output(["git", "-C", str(repo), "rev-parse", "HEAD"], text=True).strip()) state = {"playbook": "pr-review", "repo_path": str(repo), "diff_text": data["diff"], "pr_base_sha": revisions[0], "pr_head_sha": revisions[1], "pr_context": data.get("context_text", ""), "gate_report": data.get("gate_report", ""), - "task_spec": {"kind": "pr_review", "repo": data["repo"], "pr": data["pr"]}} + "pr_state": data.get("pr_state", ""), + "task_spec": {"kind": "pr_review", "repo": data["repo"], "pr": data["pr"], "params": data.get("task_params", {})}} settings = settings.model_copy(update={"improve_shadow": True, "run_root": Path("/tmp/runs"), "knowledge_dir": Path("/tmp/knowledge"), "skills_dir": Path("/candidate/skills"), "playbooks_dir": Path("/candidate/playbooks"), "adapters_dir": Path("/candidate/adapters"), "memory_db": Path("/tmp/memory.db")}) @@ -248,5 +257,6 @@ def git(*args): result = asyncio.run(_review_diff(ctx)) if not result.ok: raise ArtifactError(result.summary) return {"review_text": result.outputs.get("review_text") or state.get("review_text", ""), - "review_comments": result.outputs.get("review_comments", [])} + "review_comments": result.outputs.get("review_comments", []), + **{k: state.get(k) for k in ("review_summary", "review_verdict", "review_finding_dispositions", "review_carried_findings", "review_finding_rechecks", "review_recheck_missing")}} raise ArtifactError(f"worker has no driver {driver}") diff --git a/src/infermatrix_copilot/improve/enroll.py b/src/infermatrix_copilot/improve/enroll.py index a464b4c9..f9e02c30 100644 --- a/src/infermatrix_copilot/improve/enroll.py +++ b/src/infermatrix_copilot/improve/enroll.py @@ -105,6 +105,15 @@ def parse_declaration(doc: dict, source: str = "") -> WorkflowDeclaratio raise DeclarationError(f"{source}: evolution.{key} must be a string list") from .artifacts import validate_policy validate_policy(evolution) + objective = doc.get("objective", {}) + if not isinstance(objective, dict) or set(objective) - {"metric", "min_effect", "prior_sd"}: + raise DeclarationError(f"{source}: unsupported objective policy") + if objective.get("metric", "contract_success") not in ("contract_success", "resource_gain"): + raise DeclarationError(f"{source}: objective metric must be executable") + import math + for key in ("min_effect", "prior_sd"): + if key in objective and (not math.isfinite(float(objective[key])) or not 0 < float(objective[key]) <= 1): + raise DeclarationError(f"{source}: objective.{key} must be finite in (0,1]") return WorkflowDeclaration( workflow=workflow, kind=kind, unit=unit, item_key=str(_require(doc, "item_key", source)), fingerprint_covers=tuple(covers), capture=capture, diff --git a/src/infermatrix_copilot/improve/evolution.py b/src/infermatrix_copilot/improve/evolution.py index 9ce4873d..aa2597df 100644 --- a/src/infermatrix_copilot/improve/evolution.py +++ b/src/infermatrix_copilot/improve/evolution.py @@ -1,4 +1,4 @@ -"""Resumable, human-promoted evolution for business workflows and the engine itself.""" +"""Resumable, objective-verified evolution for business workflows and the engine itself.""" from __future__ import annotations from dataclasses import asdict @@ -17,7 +17,7 @@ from .isolation import Sandbox, SandboxUnavailable from .ledger import Ledger -TERMINAL = {"rejected", "closed", "deployed"} +TERMINAL = {"rejected", "closed", "deployed", "retained", "rolled-back"} def directory(settings) -> Path: return ledger_dir_for(settings) / "evolution" @@ -38,6 +38,9 @@ def save(settings, c: dict, store=None) -> None: result={"type": "evolution", "candidate": c["id"], "state": c["state"], "reason": c.get("reason", "")}) def check(settings, store, workflow: str, *, repo: str = "", sandbox=None, mechanical=False) -> dict: + from . import objectives + if objectives.enabled(settings): + return objectives.check(settings, store, workflow, repo=repo, sandbox=sandbox) decl = declarations_for(settings).get(workflow) reasons = [] if not decl: @@ -298,18 +301,35 @@ def run(settings, store, *, workflow="all", repo="", post=False, llm=None, sandb return _run(settings, store, workflow, repo, post, llm, sandbox or Sandbox(), now) def _run(settings, store, workflow, repo, post, llm, sandbox, now): + from . import objectives + autonomous = objectives.enabled(settings) + if autonomous: + if settings.review_lens_backends or settings.llm_mixture: + return {'state': 'deferred', 'reason': 'mixed-review-routes-not-supported-by-objective-proxy'} + try: + settings = objectives.execution_settings(settings) + except (ValueError, RuntimeError) as exc: + return {'state': 'deferred', 'reason': 'fixed-model-unconfigured: ' + str(exc)} + from .runtime import sync as runtime_sync + runtime_sync(settings, store, now=now) from .evolution_publish import sync, publish sync(settings, store) active = [c for c in candidates(settings) if c["state"] not in TERMINAL and (workflow == "all" or c["workflow"] == workflow) and (not repo or c.get("repo") == repo)] if active: c = active[0] if c.get("generation_interrupted"): + if autonomous: + c.update(state="rejected", reason="generation-interrupted; no paid-call replay") + save(settings, c, store) return c - if c["state"] in ("pr-ready", "published", "merged"): + if c["state"] in ("pr-ready", "published", "merged", "canary", "retained", "rolled-back"): + if autonomous and c["state"] == "pr-ready" and settings.improve_promotion_mode == "automatic": + from .runtime import promote + return promote(settings, store, c, sandbox=sandbox) if post: publish(settings, store, c) return c if c["state"] == "generating": - c.update(state="deferred", generation_interrupted=True, reason="generation-interrupted; paid request is not automatically repeated") + c.update(state="rejected" if autonomous else "deferred", generation_interrupted=True, reason="generation-interrupted; paid request is not automatically repeated") save(settings, c, store); return c else: if any(iso_week(c["created_at"]) == iso_week(now) for c in candidates(settings)): @@ -338,6 +358,9 @@ def _run(settings, store, workflow, repo, post, llm, sandbox, now): save(settings, c, store) source = Path(settings.improve_evolve_source_dir or artifacts.source_root()).resolve() decl = declarations_for(settings)[c["workflow"]] + if autonomous: + from dataclasses import replace + decl = replace(decl, evolution={**decl.evolution, "settings": [k for k in decl.evolution.get("settings", []) if k not in ("ECO_MODEL", "PERFORMANCE_MODEL", "AGENT_MODEL")]}) work = directory(settings) / "candidates" / c["id"] governor = governor_for(settings, ledger_dir_for(settings)) try: @@ -345,9 +368,13 @@ def _run(settings, store, workflow, repo, post, llm, sandbox, now): if c["state"] in ("new", "deferred") and "generated" not in c: status = check(settings, store, c["workflow"], repo=repo, sandbox=sandbox, mechanical=c.get("mode") == "mechanical") if not status["ready"]: c.update(state="deferred", reason="; ".join(status["reasons"])); save(settings, c, store); return c - base = artifacts.baseline(source, work / "incumbent") + if autonomous: + from .runtime import baseline + base = baseline(settings, c["workflow"], work / "incumbent") + else: + base = artifacts.baseline(source, work / "incumbent") c["base_revision"] = base["revision"] - rows = drivers.dataset(settings, c["workflow"]) + rows = objectives.rows(settings, c["workflow"]) if autonomous else drivers.dataset(settings, c["workflow"]) development = [r for r in rows if r["split"] == "development" and (not repo or r["item"].split("#")[0] == repo)] evidence_ids = set(c["evidence"]) try: @@ -358,12 +385,17 @@ def _run(settings, store, workflow, repo, post, llm, sandbox, now): raise artifacts.ArtifactError("promotion labels/reports cannot be used as generation evidence") c["evidence_excerpt"] = evidence c["state"] = "generating"; save(settings, c, store) + if autonomous: + from dataclasses import replace + decl = replace(decl, evolution={**decl.evolution, "settings": [k for k in decl.evolution.get("settings", []) if k not in ("ECO_MODEL", "PERFORMANCE_MODEL", "AGENT_MODEL")]}) c["generated"] = generate(settings, decl, work / "incumbent", c, development, llm, governor) c["state"] = "generated"; save(settings, c, store) if artifacts.git(source, "rev-parse", "HEAD") != c["base_revision"]: raise artifacts.ArtifactError("source baseline advanced; candidate needs re-evaluation") if "verified" not in c: arm = artifacts.apply_candidate(work / "incumbent", work / "arm", c["generated"], decl.evolution, workflow=c["workflow"]) + if autonomous and c['workflow'] == objectives.ENGINE and 'src/infermatrix_copilot/improve/lints.py' in arm['paths'] and len(arm['paths']) > 1: + raise artifacts.ArtifactError('one self-evolution candidate must target either lints or attribution, with an independent objective') affected = [d for d in declarations_for(settings).values() if any(fnmatch.fnmatchcase(path, pattern) for path in arm["paths"] for pattern in [*d.evolution.get("paths", []), *d.evolution.get("dependencies", [])])] tests = sorted(set(decl.evolution["tests"]) | {t for d in affected for t in d.evolution.get("tests", [])}) @@ -398,6 +430,19 @@ def _run(settings, store, workflow, repo, post, llm, sandbox, now): (work / "candidate.patch").write_text(artifacts.patch(work / "incumbent", work / "arm")) c["patch_sha"] = artifacts.digest((work / "candidate.patch").read_bytes()) c["state"] = "verified"; save(settings, c, store) + if autonomous: + c["evaluation"] = objectives.evaluate(settings, store, c, sandbox, llm) + c["state"] = "pr-ready" if c["evaluation"].get("promotable") else "deferred" if c["evaluation"].get("blocking") else "rejected" + c["reason"] = "" if c["state"] == "pr-ready" else "objective experiment lacks sufficient evidence or improvement" + save(settings, c, store) + from .evolution_publish import draft + draft(settings, c) + if c["state"] == "pr-ready" and settings.improve_promotion_mode == "automatic": + from .runtime import promote + c = promote(settings, store, c, sandbox=sandbox) + if post and c["state"] == "canary": publish(settings, store, c) + elif post: publish(settings, store, c) + return c if c.get("experiment"): previous = experiments.load(ledger_dir_for(settings), c["experiment"]) if previous.state == "underpowered" or previous.result.get("blocking") in ("isolation", "data"): diff --git a/src/infermatrix_copilot/improve/evolution_publish.py b/src/infermatrix_copilot/improve/evolution_publish.py index e3b78d91..e673ebaa 100644 --- a/src/infermatrix_copilot/improve/evolution_publish.py +++ b/src/infermatrix_copilot/improve/evolution_publish.py @@ -16,8 +16,8 @@ def draft(settings, c): f"Base: `{c['base_revision']}`. Candidate source: `{c['artifact']['tree_sha']}`.\n\n" f"Affected workflows: {', '.join(c.get('affected_workflows', []))}.\n\n" "Validation\n\n```json\n" + json.dumps({"tests": c.get("verified"), "experiment": evaluation}, indent=2) + "\n```\n\n" - "Rollback: revert this PR and restore the preceding configuration. Human approval is required.\n\n" - f"\n") + + ("Deployment: verified local sandbox release with automatic canary and rollback; this PR is an optional audit record.\n\n" if getattr(settings, "improve_promotion_mode", "pr") == "automatic" and getattr(settings, "improve_evaluation_mode", "gold") == "objective" else "Rollback: revert this PR and restore the preceding configuration. Human approval is required.\n\n") + + f"\n") (root / "PR.md").write_text(body, encoding="utf-8") atomic_json(root / "bundle.json", {"protocol": PROTOCOL, "candidate": c["id"], "workflow": c["workflow"], "base_revision": c["base_revision"], "source_sha": c["artifact"]["tree_sha"], @@ -28,8 +28,8 @@ def publish(settings, store, c): from .evolution import directory, save from .ledger import Ledger from .cycle import ledger_dir_for - if c["state"] != "pr-ready" or not settings.allow_post or not settings.allow_push: - return {"published": False, "reason": "requires pr-ready, ALLOW_POST=1 and ALLOW_PUSH=1"} + if c["state"] not in ("pr-ready", "canary", "retained") or not settings.allow_post or not settings.allow_push: + return {"published": False, "reason": "requires evaluated candidate, ALLOW_POST=1 and ALLOW_PUSH=1"} if Ledger(ledger_dir_for(settings)).load(c["workflow"]).hold: return {"published": False, "reason": "workflow publication hold"} if not settings.improve_evolve_outbox_dir or not settings.improve_proposal_repo: @@ -62,8 +62,10 @@ def sync(settings, store): if ack_path.is_file(): ack = json.loads(ack_path.read_text()) if ack.get("candidate") == c["id"] and ack.get("source_sha") == c.get("artifact", {}).get("tree_sha"): - if ack.get("ok") and not ack.get("dry_run") and c["state"] == "pr-ready": - c.update(state="published", pr=ack["url"], reason="") + if ack.get("ok") and not ack.get("dry_run"): + if c["state"] == "pr-ready": c.update(state="published", pr=ack["url"], reason="") + elif c["state"] in ("canary", "retained", "rolled-back"): + c["pr"] = ack["url"] # audit never changes runtime adoption obs_path = root / "inbox" / f"{c['id']}.json" if obs_path.is_file(): obs = json.loads(obs_path.read_text()) diff --git a/src/infermatrix_copilot/improve/isolation.py b/src/infermatrix_copilot/improve/isolation.py index 693785dd..b40343e4 100644 --- a/src/infermatrix_copilot/improve/isolation.py +++ b/src/infermatrix_copilot/improve/isolation.py @@ -131,6 +131,8 @@ def drain(): if result.get("source_sha") != artifact["tree_sha"] or not result.get("package_path", "").startswith("/candidate/src/"): raise SandboxUnavailable("worker source fingerprint/import mismatch") verify(source) + # Counted by the parent; a worker cannot forge resource accounting. + result["_broker_calls"] = calls if payload.get("mode") == "tests": result["report"] = "".join(stderr_tail)[-12000:] return result finally: diff --git a/src/infermatrix_copilot/improve/objectives.py b/src/infermatrix_copilot/improve/objectives.py new file mode 100644 index 00000000..d948b6e3 --- /dev/null +++ b/src/infermatrix_copilot/improve/objectives.py @@ -0,0 +1,426 @@ +"""Trusted executable objectives. No human labels, model judges or candidate scores. + +Fault witnesses stay in the parent. Tests exercise bounded contracts, never claim +semantic recall on arbitrary PRs or correctness of newly generated knowledge. +""" +from __future__ import annotations +from dataclasses import asdict +import copy +import json +from pathlib import Path +import time +import uuid + +from . import artifacts, drivers, stats +from .budget import BudgetRefused +from .cycle import governor_for, ledger_dir_for +from .enroll import declarations_for +from .isolation import Sandbox, SandboxUnavailable +from ..trace_store import file_lock, bind_store, trace_context + +PROTOCOL = 'objective/1' +ENGINE = 'workflow-improve.improve.forensics' +DRIVERS = {'pr-review': 'pr-review', 'kb-intake': 'kb-intake', 'meta': 'objective-meta', 'mechanical': 'objective-tests'} +FAULTS = {'ceiling': ('S4', 'L01'), 'discard': ('S3', 'L09'), 'fallback': ('S7', 'L07')} + + +def enabled(settings): + return getattr(settings, 'improve_evaluation_mode', 'gold') == 'objective' + + +def execution_settings(settings): + """One API route for generation, inference and diagnosis; no stronger tier.""" + from .experiments import settings_for + model = settings.improve_evolve_model or settings.tier_target('eco').model + patched, _ = settings_for(settings, {'ECO_MODEL': model, 'PERFORMANCE_MODEL': model, 'AGENT_MODEL': model}) + return patched.model_copy(update={'review_planner_model': model, 'review_promotion_model': model, + 'review_lens_backends': {}, 'llm_mixture': {}}) + + +def data_root(settings): + return Path(settings.improve_evolve_data_dir).expanduser() if settings.improve_evolve_data_dir else ledger_dir_for(settings) / 'evolution' / 'inputs' + + +def rows(settings, workflow): + return drivers.dataset(settings.model_copy(update={'improve_evolve_data_dir': str(data_root(settings))}), workflow) + + +def fault_case(seed: str, fault: str): + """Known intervention and a clean control; only the intervention's input travels.""" + uid, rid, gid = 'unit-' + seed, 'event-' + seed, 'concern-' + seed + reply = json.dumps({'summary': 'completed', 'review_comments': []}) + ref = 'sha256:' + artifacts.digest(reply.encode()) + ctx = {'workflow': ENGINE, 'playbook': 'workflow-improve', 'unit_id': uid, 'item': seed, 'fingerprint': seed} + model = {'id': rid, 'kind': 'model_call', 'schema': 'trace/1', 'at': 1, 'context': ctx, + 'seconds': 1, 'model': {'model': 'fixed', 'served_model': 'fixed'}, + 'usage': {'input_tokens': 20, 'output_tokens': 20}, 'inputs': {}, 'outputs': {'reply': ref}, + 'result': {'stop_reason': 'end_turn', 'max_tokens': 100}} + terminal = {'id': 'terminal-' + seed, 'kind': 'decision', 'schema': 'trace/1', 'at': 2, + 'context': ctx, 'inputs': {}, 'outputs': {}, 'result': {'type': 'step_result', 'status': 'ok'}} + clean = {'records': [model, terminal], 'blobs': {ref: reply}, 'concerns': [], 'cells': []} + mutated = copy.deepcopy(clean) + if fault == 'ceiling': + mutated['records'][0]['result'].update(stop_reason='max_tokens', max_tokens=20) + evidence_id = rid + elif fault == 'discard': + mutated['records'][1]['result'] = {'type': 'cap', 'dropped': 1 + int(seed[:2], 16) % 12, 'cap': 3} + evidence_id = terminal['id'] + elif fault == 'fallback': + mutated['records'][1]['result'] = {'type': 'planner_zero_output', 'reason': 'injected planner failure'} + evidence_id = terminal['id'] + else: + raise artifacts.ArtifactError('unknown controlled fault') + mutated.update(concerns=[{'gold_id': gid, 'path': 'runtime', 'concern': 'The unit failed to retain a complete executable result.'}], cells=[gid]) + stage, lint = FAULTS[fault] + return mutated, clean, {'cell': gid, 'stage': stage, 'lint': lint, 'evidence': evidence_id} + + +def capture(store, workflow, item, payload): + """Record a replay envelope in a trusted host, before invoking candidate code.""" + if store is None: return + forbidden = {'labels', 'gold', 'scores', 'judgments', 'report'} + if forbidden & set(payload): raise artifacts.ArtifactError('replay input includes evaluator material') + raw = json.dumps(payload, ensure_ascii=False) + if len(raw.encode()) > 16_000_000: raise artifacts.ArtifactError('replay input exceeds capture limit') + store.append('decision', context={'workflow': workflow, 'item': item}, + inputs={'evolution_input': raw}, result={'type': 'replay_input', 'input_sha': artifacts.digest(raw.encode())}) + + +def collect(settings, store, workflow): + """Automatically import immutable trace inputs; never guess missing snapshots.""" + root = data_root(settings) / workflow + with file_lock(root / 'collection.lock', blocking=True) as held: + if not held: return 0 + path = root / 'dataset.json' + doc = json.loads(path.read_text()) if path.exists() else {'workflow': workflow, 'items': []} + known = {r['item'] for r in doc['items']} + added = 0 + for r in store.query(workflow=workflow, limit=10000): + if r.get('result', {}).get('type') != 'replay_input': continue + ref = r.get('inputs', {}).get('evolution_input') + if not ref: continue + payload = json.loads(store.blob(ref)) + source_id = str(r['id']) + # Cluster all captures of one business item together. A retry or a + # changed snapshot must not become an independent holdout item. + item = str(r.get('context', {}).get('item') or source_id) + if item in known: continue + name = artifacts.digest(item.encode()) + '.json' + artifacts.atomic_json(root / name, payload) + doc['items'].append({'item': item, 'split': 'development' if int(artifacts.digest(item.encode())[:8], 16) % 5 == 0 else 'holdout', + 'input': name, 'sha256': artifacts.digest((root / name).read_bytes()), + 'origin': {'record': source_id, 'kind': 'trace-replay'}, 'objective': {}}) + known.add(item); added += 1 + artifacts.atomic_json(path, doc) + return added + + +def prepare(settings, store, *, workflow='all', repo=''): + """Generate controlled engine cases and collect available business snapshots.""" + report = {} + for name, decl in declarations_for(settings).items(): + if workflow != 'all' and name != workflow: continue + if not decl.evolution: continue + if name == ENGINE: + root = data_root(settings) / name + path = root / 'dataset.json' + with file_lock(root / 'collection.lock', blocking=True) as held: + if not held: continue + doc = json.loads(path.read_text()) if path.exists() else {'workflow': name, 'items': []} + # A fresh, bounded pool for each candidate. Labels are causal + # witnesses of the controller's intervention, never model text. + used_path = ledger_dir_for(settings) / 'evolution' / 'holdouts.json' + used = json.loads(used_path.read_text()) if used_path.exists() else {} + fresh = [r for r in doc['items'] if r['split'] == 'holdout' and not any(k.startswith(name + ':' + r['item'] + ':') for k in used)] + floor = max([80, *[trial.get('n_required', 0) for cpath in (ledger_dir_for(settings) / 'evolution' / 'candidates').glob('*/candidate.json') + for trial in json.loads(cpath.read_text()).get('objective_trials', []) + if json.loads(cpath.read_text()).get('workflow') == name]]) + if len(fresh) < floor: + # Grow only the controller's scenarios; business inputs + # must arrive from real traces. Budget still gates calls. + for i in range(max(100, floor - len(fresh) + 10)): + seed = uuid.uuid4().hex + fault = tuple(FAULTS)[i % len(FAULTS)] + payload, clean, witness = fault_case(seed, fault) + if i % 4 == 0: + payload, witness = clean, {'negative': True} + artifacts.atomic_json(root / (seed + '.json'), payload) + doc['items'].append({'item': 'controlled:' + seed, 'split': 'development' if i < 10 else 'holdout', + 'input': seed + '.json', 'sha256': artifacts.digest((root / (seed + '.json')).read_bytes()), + 'origin': {'kind': 'controlled-intervention', 'version': 1}, 'objective': witness}) + artifacts.atomic_json(path, doc) + report[name] = {'items': len(doc['items']), 'source': 'controlled-interventions'} + store.append('decision', context={'workflow': name}, result={'type': 'objective_inputs', 'items': len(doc['items'])}) + else: + report[name] = {'imported': collect(settings, store, name), 'source': 'trace-replay'} + return report + + +def policy(decl, mechanism=''): + supplied = decl.raw.get('objective') or {} + goal = 'resource_gain' if mechanism in ('L11', 'L15', 'efficiency') else supplied.get('metric', 'contract_success') + return {'metric': goal, 'min_effect': float(supplied.get('min_effect', .05)), 'sd': float(supplied.get('prior_sd', .15)), + 'guards': {'contract_success': 'higher'}, 'replicates': 3} + + +def check(settings, store, workflow, *, repo='', sandbox=None): + decl = declarations_for(settings).get(workflow) + if not decl or not decl.evolution: + return {'workflow': workflow, 'ready': False, 'reasons': ['no-objective-mutation-policy'], 'evaluation_mode': 'objective'} + reasons = [] + st = None + try: + st = execution_settings(settings) + if st.tier_target('eco').kind != 'api': reasons.append('objective-mode-requires-one-API-route') + if settings.review_lens_backends or settings.llm_mixture: reasons.append('mixed-review-routes-not-supported-by-objective-proxy') + except (ValueError, RuntimeError): reasons.append('fixed-model-unconfigured') + data = [r for r in rows(settings, workflow) if not repo or r['item'].split('#')[0] == repo] + required_keys = {'pr-review': {'repo', 'pr', 'base_sha', 'head_sha', 'base_files', 'head_files', 'diff', 'knowledge_files'}, + 'kb-intake': {'repo', 'repo_dir', 'generator_model', 'evidence', 'files', 'release', 'today'}, + 'meta': {'records', 'blobs', 'concerns', 'cells'}}.get(decl.experiment_driver, set()) + if any(required_keys - set(r['payload']) for r in data): reasons.append('incomplete-driver-snapshot') + if st and decl.experiment_driver == 'kb-intake' and any(r['payload'].get('generator_model') != st.tier_target('eco').model for r in data): + reasons.append('pinned-generator-model-does-not-match-broker') + dev, hold = [r for r in data if r['split'] == 'development'], [r for r in data if r['split'] == 'holdout'] + used_path = ledger_dir_for(settings) / 'evolution' / 'holdouts.json' + used = json.loads(used_path.read_text()) if used_path.exists() else {} + hold = [r for r in hold if f"{workflow}:{r['item']}:{r['version']}" not in used] + p = policy(decl) + required = stats.items_required(p['sd'], p['min_effect']) + if not dev: reasons.append('missing-replay-development-inputs') + if len(hold) < required: reasons.append(f'insufficient-fresh-objective-items: {len(hold)} < {required}') + driver = DRIVERS.get(decl.experiment_driver) + if driver == 'objective-tests': + reasons = [r for r in reasons if not r.startswith('insufficient-fresh-objective-items') and r != 'missing-replay-development-inputs'] + required = 0 + if settings.improve_promotion_mode == 'automatic': reasons.append('missing-sandboxed-production-driver') + if not driver: reasons.append('missing-objective-driver') + source = Path(settings.improve_evolve_source_dir or artifacts.source_root()) + try: + revision = artifacts.git(source, 'rev-parse', 'HEAD') + if artifacts.git(source, 'status', '--porcelain', '--untracked-files=no', '--', 'src', 'playbooks', 'adapters', 'skills'): reasons.append('uncommitted-source-baseline') + except artifacts.ArtifactError: + revision = ''; reasons.append('source-baseline-not-a-repository') + for test in decl.evolution['tests']: + if not (source / test).is_file(): reasons.append('missing-regression-test:' + test) + isolation = (sandbox or Sandbox()).check() + if not isolation['ready']: reasons.append('sandbox-unavailable:' + isolation['reason']) + budget = governor_for(settings, ledger_dir_for(settings)).remaining() + if budget['usd_remaining'] <= 0: reasons.append('budget-exhausted') + if not store or not store.query(workflow=workflow, limit=1): reasons.append('missing-workflow-traces') + return {'workflow': workflow, 'tier': 1, 'evaluation_mode': 'objective', 'ready': not reasons, + 'reasons': reasons, 'driver': driver, 'source_revision': revision, 'development_items': len(dev), + 'holdout_items': len(hold), 'n_required': required, 'sandbox': isolation, 'budget': budget, + 'semantic_quality_claim': False, 'evolution': decl.evolution} + + +def canonical(driver, output): + if driver == 'pr-review': return {k: output.get(k, []) for k in ('review_text', 'review_summary', 'review_comments', 'review_verdict', 'review_finding_dispositions', 'review_carried_findings', 'review_finding_rechecks', 'review_recheck_missing')} + if driver == 'kb-intake': return output.get('operations', []) + return output.get('attributions', []) + + +def score(driver, row, output, *, lint=False): + """All scores are computed here; candidate numeric fields are ignored.""" + if driver == 'meta': + expected = row.get('objective') or {} + if row.get('origin', {}).get('kind') != 'controlled-intervention': raise artifacts.ArtifactError('missing causal witness') + if lint: + found = output.get('findings') + if not isinstance(found, list): return 0.0 + if expected.get('negative'): return float(not found) + ids = {r['id'] for r in row['payload']['records']} + return float(any(f.get('lint') == expected['lint'] and expected['evidence'] in f.get('evidence', []) for f in found) + and all(f.get('evidence') and set(f['evidence']).issubset(ids) and f.get('lint') != 'L00' for f in found)) + found = output.get('attributions') + if not isinstance(found, list): return 0.0 + if expected.get('negative'): return float(not found) + matches = [a for a in found if a.get('gold_id') == expected['cell']] + ids = {r['id'] for r in row['payload']['records']} + return float(len(found) == len(row['payload']['cells']) and len(matches) == 1 and matches[0].get('stage') == expected['stage'] and not matches[0].get('disputed') + and expected['evidence'] in matches[0].get('evidence', []) and set(matches[0]['evidence']).issubset(ids)) + if driver == 'pr-review': + data = row['payload']; findings = output.get('review_comments') + if not isinstance(findings, list) or not isinstance(output.get('review_text'), str) or not output['review_text'].strip(): return 0.0 + for f in findings: + path, line = f.get('file', f.get('path')), f.get('line') + if path not in data['head_files'] or not isinstance(line, int) or isinstance(line, bool) or not 1 <= line <= len(data['head_files'][path].splitlines()): return 0.0 + if f.get('severity') not in ('blocker', 'major', 'minor', 'nit') or not f.get('comment') or f.get('disposition', 'publish') != 'publish': return 0.0 + anchor = f.get('anchor_snippet') + if anchor and anchor not in data['head_files'][path]: return 0.0 + return 1.0 + if driver == 'kb-intake': + from ..knowledge_service.ops import KnowledgeOperation, apply_operations + data = row['payload'] + raw = output.get('operations') + if not isinstance(raw, list) or len(raw) > 6 or output.get('rejected'): return 0.0 + try: + operations = [KnowledgeOperation.from_dict(o) for o in raw] + if any(o.allow_protected or any(not artifacts.safe_path(p).startswith(data['repo_dir'].rstrip('/') + '/') for p in (o.page, o.new_page) if p) for o in operations): return 0.0 + apply_operations(data['files'], operations, release=data['release'], today=data['today']) + except (ValueError, KeyError, TypeError): return 0.0 + return 1.0 + raise artifacts.ArtifactError('unknown executable objective') + + +def evaluate(settings, store, c, sandbox, llm): + """Pre-register once, checkpoint each paid call, paired item-level inference.""" + from .evolution import directory + decl = declarations_for(settings)[c['workflow']] + work = directory(settings) / 'candidates' / c['id'] + path = work / 'objective.json' + governor = governor_for(settings, ledger_dir_for(settings)) + st = execution_settings(settings) + if any(k in c['generated'].get('overrides', {}) for k in ('ECO_MODEL', 'PERFORMANCE_MODEL', 'AGENT_MODEL')): + raise artifacts.ArtifactError('objective candidates cannot change the pinned model') + from .evolution import public_settings + source = Path(settings.improve_evolve_source_dir or artifacts.source_root()) + if decl.experiment_driver == 'mechanical': + # Existing trusted tests are an automatic executable reproducer. A + # passing incumbent cannot establish a defect repair on this path. + runs = {} + for side in ('incumbent', 'arm'): + runs[side] = sandbox.run(work / side, {'mode': 'tests', 'tests': decl.evolution['tests'], 'settings': public_settings(st)}, + work / 'objective-tests' / side, tests=source / 'test') + verdict = {'protocol': PROTOCOL, 'label': 'supported' if runs['incumbent'].get('rc') == 1 and runs['arm'].get('rc') == 0 else 'neutral', + 'promotable': runs['incumbent'].get('rc') == 1 and runs['arm'].get('rc') == 0, + 'kind': 'existing-contract-repair', 'semantic_quality_claim': False, 'reproducer': runs} + artifacts.atomic_json(path, {'sources': {side: artifacts.verify(work / side)['tree_sha'] for side in ('incumbent', 'arm')}, 'result': verdict}) + return verdict + p = policy(decl, c['mechanism']) + all_rows = {r['item']: r for r in rows(settings, c['workflow'])} + previous_required = max([0, *[t['n_required'] for t in c.get('objective_trials', [])]]) + if path.exists() and json.loads(path.read_text()).get('result', {}).get('blocking') == 'data': + old = json.loads(path.read_text()) + previous_required = old['result']['n_required'] + c.setdefault('objective_trials', []).append({'path': str(path.with_name('objective-' + str(len(c.get('objective_trials', []))) + '.json')), 'n_required': previous_required}) + path.replace(Path(c['objective_trials'][-1]['path'])) + from .evolution import save + save(settings, c, store) + if path.exists(): + exp = json.loads(path.read_text()) + # Frozen registry + source + execution settings are never refreshed + # under the same paid experiment, even after a process restart. + if exp['policy'] != p or exp['settings_sha'] != artifacts.digest(json.dumps(public_settings(st), sort_keys=True).encode()): + raise artifacts.ArtifactError('objective registration settings changed') + else: + required = max(previous_required, stats.items_required(p['sd'], p['min_effect'])) + fresh = [r for r in all_rows.values() if r['split'] == 'holdout' and (not c.get('repo') or r['item'].split('#')[0] == c['repo'])] + used_path = directory(settings) / 'holdouts.json' + with file_lock(directory(settings) / 'holdouts.lock', blocking=True) as held: + if not held: raise SandboxUnavailable('holdout registry locked') + used = json.loads(used_path.read_text()) if used_path.exists() else {} + fresh = [r for r in fresh if f"{c['workflow']}:{r['item']}:{r['version']}" not in used] + selected = sorted(fresh, key=lambda r: r['item'])[:required] + if len(selected) < required: raise SandboxUnavailable('insufficient fresh objective scenarios') + exp = {'protocol': PROTOCOL, 'id': c['id'] + '-objective-' + uuid.uuid4().hex[:8], 'state': 'registered', 'policy': p, 'n_required': required, + 'items': {r['item']: r['version'] for r in selected}, 'sources': {side: artifacts.verify(work / side)['tree_sha'] for side in ('incumbent', 'arm')}, + 'settings_sha': artifacts.digest(json.dumps(public_settings(st), sort_keys=True).encode()), 'progress': {}, 'at': time.time()} + # Registration is durable before consuming holdouts. Resuming also + # verifies/repairs this index, so a crash between writes is safe. + artifacts.atomic_json(path, exp) + for r in selected: used[f"{c['workflow']}:{r['item']}:{r['version']}"] = exp['id'] + artifacts.atomic_json(used_path, used) + for item, version in exp['items'].items(): + if item not in all_rows or all_rows[item]['version'] != version: raise artifacts.ArtifactError('objective snapshot changed') + with file_lock(directory(settings) / 'holdouts.lock', blocking=True) as held: + if not held: raise SandboxUnavailable('holdout registry locked') + used_path = directory(settings) / 'holdouts.json' + used = json.loads(used_path.read_text()) if used_path.exists() else {} + for item, version in exp['items'].items(): + key = f"{c['workflow']}:{item}:{version}" + if key in used and used[key] != exp['id']: raise artifacts.ArtifactError('objective holdout reused') + used[key] = exp['id'] + artifacts.atomic_json(used_path, used) + for side in ('incumbent', 'arm'): + if artifacts.verify(work / side)['tree_sha'] != exp['sources'][side]: raise artifacts.ArtifactError('objective source changed') + if exp.get('result'): return exp['result'] + lint = decl.experiment_driver == 'meta' and set(c['artifact']['paths']) == {'src/infermatrix_copilot/improve/lints.py'} + driver = 'objective-lints' if lint else DRIVERS[decl.experiment_driver] + for item in exp['items']: + for rep in range(3): + key = item + '/' + str(rep) + sample = exp['progress'].get(key) + if sample: + if sample['state'] == 'running': + exp['progress'][key] = {'state': 'excluded', 'reason': 'interrupted paid call; no replay'} + artifacts.atomic_json(path, exp) + continue + before = governor.remaining()['usd_remaining'] + if before <= 0: raise BudgetRefused('objective experiment budget exhausted') + exp['progress'][key] = {'state': 'running'}; exp['state'] = 'running'; artifacts.atomic_json(path, exp) + sample = {'state': 'scored', 'item': item, 'outputs': {}, 'scores': {}, 'seconds': {}, 'usd': {}, 'calls': {}} + order = ('arm', 'incumbent') if (int(artifacts.digest(item.encode())[:8], 16) + rep) % 2 else ('incumbent', 'arm') + for side in order: + started, settled = time.monotonic(), governor.remaining()['usd_settled'] + arm_settings = settings_for_source(st, work / side, c['workflow']) + context = {'workflow': c['workflow'], 'unit_id': f"exp-objective:{c['id']}:{key}:{side}", 'item': item} + with bind_store(store), trace_context(**context): + output = sandbox.run(work / side, {'mode': 'predict', 'driver': driver, 'capture_errors': True, 'input': all_rows[item]['payload'], + 'settings': public_settings(arm_settings), 'trace_context': context}, + work / 'objective-runs' / artifacts.digest(key.encode())[:16] / side, + llm=llm, settings=arm_settings, governor=governor) + if output.get('source_sha') != exp['sources'][side]: raise artifacts.ArtifactError('objective worker fingerprint mismatch') + sample['seconds'][side] = time.monotonic() - started + sample['usd'][side] = governor.remaining()['usd_settled'] - settled + sample['calls'][side] = output.get('_broker_calls', 0) + sample['outputs'][side] = canonical(decl.experiment_driver, output) + sample['scores'][side] = score(decl.experiment_driver, all_rows[item], output, lint=lint) + # A resource optimization cannot shrink artifacts to buy a score. + # For semantic tasks require exact baseline finding/operation sets. + equivalent = decl.experiment_driver == 'meta' or sample['outputs']['arm'] == sample['outputs']['incumbent'] + sample['no_regression'] = sample['scores']['arm'] >= sample['scores']['incumbent'] and (equivalent or not sample['scores']['incumbent']) + a, b = sample['usd']['arm'], sample['usd']['incumbent'] + if b <= 0: a, b = sample['seconds']['arm'], sample['seconds']['incumbent'] + sample['resource_gain'] = (b - a) / max(b, 1e-9) + sample['outputs'] = {side: artifacts.digest(json.dumps(value, sort_keys=True).encode()) for side, value in sample['outputs'].items()} + exp['progress'][key] = sample; artifacts.atomic_json(path, exp) + complete = [item for item in exp['items'] if all(exp['progress'][item + '/' + str(r)]['state'] == 'scored' for r in range(3))] + deltas = {item: [exp['progress'][item + '/' + str(rep)]['resource_gain'] if p['metric'] == 'resource_gain' else + exp['progress'][item + '/' + str(rep)]['scores']['arm'] - exp['progress'][item + '/' + str(rep)]['scores']['incumbent'] for rep in range(3)] for item in complete} + result = stats.paired(deltas, p['metric']) + required = max(exp['n_required'], stats.items_required(result.sd, p['min_effect'])) + label = stats.label(result, n_required=required) + guard = all(s.get('no_regression') for s in exp['progress'].values() if s['state'] == 'scored') + verdict = {'protocol': PROTOCOL, **asdict(result), 'label': label, 'metric': p['metric'], 'min_effect': p['min_effect'], + 'n_required': required, 'n_registered': len(exp['items']), 'replicates': 3, 'snapshot_versions': exp['items'], + 'no_regression': guard, 'promotable': label == 'supported' and result.mean >= p['min_effect'] and guard, + 'sources': exp['sources'], 'semantic_quality_claim': False, + 'scope': 'controlled-interventions' if decl.experiment_driver == 'meta' else 'replay-contracts-and-output-preservation'} + if len(complete) < required: verdict['blocking'] = 'data' + exp['state'] = label; exp['result'] = verdict; artifacts.atomic_json(path, exp) + store.append('decision', result={'type': 'objective_verdict', 'candidate': c['id'], **verdict}) + return verdict + + +def worker(payload, settings, llm): + """Protected driver called by the trusted launcher, not a mutable score function.""" + from .meta import _unit_from + from ..trace_store import TraceStore + data = payload['input'] + store = TraceStore(Path('/tmp/objective-traces')) + for ref, content in data['blobs'].items(): + if store.put_blob(content) != ref: raise artifacts.ArtifactError('objective blob mismatch') + unit = _unit_from(data['records']) + if payload['driver'] == 'objective-lints': + from .lints import run_lints, Baseline + return {'findings': [asdict(f) for f in run_lints(unit, store, Baseline(settings=settings, **data.get('baseline', {})))]} + from .forensics import attribute, Cell + from .adapters import Gold, GoldEntry + from ..agent_loop import run_agent + from ..run_trace import RunTrace + target = settings.tier_target('eco') + def agent(system, prompt, scope, extra_tools, max_iters): + return run_agent(llm, system=system, prompt=prompt, scope=scope, extra_tools=extra_tools, + max_iters=max_iters, model=target.model, trace=RunTrace(Path('/tmp/objective-agent.jsonl'))).text + concerns = Gold(unit.item, tuple(GoldEntry(**g) for g in data['concerns'])) + attributed = attribute(store, {unit.unit_id: unit}, {unit.item: concerns}, + [Cell(g, unit.unit_id, 'miss') for g in data['cells']], agents={'fixed-model': agent}, max_cells=len(data['cells'])) + return {'attributions': [asdict(a) for a in attributed]} + + +def settings_for_source(settings, source, workflow): + st = artifacts.runtime_settings(settings.model_copy(update={'playbooks_dir': Path(source) / 'playbooks'}), workflow) + if st.tier_target('eco').model != settings.tier_target('eco').model or st.tier_target('performance').model != settings.tier_target('eco').model: + raise artifacts.ArtifactError('adopted source requests a different objective model') + return st diff --git a/src/infermatrix_copilot/improve/runtime.py b/src/infermatrix_copilot/improve/runtime.py new file mode 100644 index 00000000..db3652eb --- /dev/null +++ b/src/infermatrix_copilot/improve/runtime.py @@ -0,0 +1,408 @@ +"""Trusted local deployment of evaluated artifacts, with automatic rollback. + +Active bytes execute only in the existing credential-free sandbox. The host +keeps publishing/knowledge writes and the model broker. A PR is optional audit. +""" +from __future__ import annotations +import json +from pathlib import Path +import shutil +import time +import uuid + +from . import artifacts, objectives +from .cycle import ledger_dir_for, governor_for +from .isolation import Sandbox, SandboxUnavailable +from ..trace_store import bind_store, trace_context, file_lock + + +def registry(settings): + return ledger_dir_for(settings) / 'evolution' / 'active.json' + + +def active(settings): + path = registry(settings) + return json.loads(path.read_text()) if path.exists() else {} + + +def baseline(settings, workflow, dest): + current = active(settings) + if current.get('artifact'): + source = Path(current['artifact']) + meta = artifacts.verify(source) + if meta['revision'] != artifacts.git(Path(settings.improve_evolve_source_dir or artifacts.source_root()), 'rev-parse', 'HEAD'): + raise artifacts.ArtifactError('active release belongs to an outdated source baseline') + if dest.exists(): + if artifacts.verify(dest)['tree_sha'] != meta['tree_sha']: raise artifacts.ArtifactError('active baseline changed while candidate was generated') + else: shutil.copytree(source, dest) + return meta + return artifacts.baseline(Path(settings.improve_evolve_source_dir or artifacts.source_root()), dest) + + +def promote(settings, store, c, *, sandbox=None): + with file_lock(registry(settings).with_name('execution.lock'), blocking=True) as held: + if not held: raise SandboxUnavailable('release execution is busy') + return _promote(settings, store, c, sandbox=sandbox) + + +def _promote(settings, store, c, *, sandbox=None): + from .evolution import directory, save + if not settings.improve_enabled or not settings.improve_evolve_enabled or not objectives.enabled(settings) or settings.improve_promotion_mode != 'automatic': + return c + if c['workflow'] not in ('pr-review.agent.review_diff', 'kb-intake.draft', objectives.ENGINE): + raise artifacts.ArtifactError('workflow lacks a sandboxed production driver') + status = (sandbox or Sandbox()).check() + if not status['ready']: + c.update(state='deferred', reason='automatic deployment requires isolation: ' + status['reason']); save(settings, c, store); return c + work = directory(settings) / 'candidates' / c['id'] + if artifacts.git(Path(settings.improve_evolve_source_dir or artifacts.source_root()), 'rev-parse', 'HEAD') != c['base_revision']: + raise artifacts.ArtifactError('source baseline advanced before activation') + proof = json.loads((work / 'objective.json').read_text()) + verdict = proof.get('result', {}) + if verdict != c.get('evaluation') or verdict.get('protocol') != objectives.PROTOCOL or not verdict.get('promotable'): + raise artifacts.ArtifactError('automatic promotion requires the persisted controller verdict') + arm, incumbent = artifacts.verify(work / 'arm'), artifacts.verify(work / 'incumbent') + if arm['tree_sha'] != c['artifact']['tree_sha'] or proof.get('sources') != {'arm': arm['tree_sha'], 'incumbent': incumbent['tree_sha']}: + raise artifacts.ArtifactError('promotion source differs from evaluated bytes') + if arm['dependency_sha'] != artifacts.dependency_hash() or c.get('verified', {}).get('rc') != 0: + raise artifacts.ArtifactError('promotion dependency/regression verification failed') + if artifacts.digest((work / 'candidate.patch').read_bytes()) != c['patch_sha']: + raise artifacts.ArtifactError('evaluated patch changed') + # Reapply in a separate tree. A forged arm manifest cannot expand the + # mutation scope or modify the protected evaluator/runtime. + from .enroll import declarations_for + decl = declarations_for(settings)[c['workflow']] + with file_lock(registry(settings).with_suffix('.lock'), blocking=True) as held: + if not held: raise SandboxUnavailable('deployment registry locked') + current = active(settings) + if current.get('candidate') == c['id']: + c.update(state='canary', activated_at=current['at']); save(settings, c, store); return c + expected = artifacts.verify(Path(current['artifact']))['tree_sha'] if current.get('artifact') else incumbent['tree_sha'] + if incumbent['tree_sha'] != expected: raise artifacts.ArtifactError('deployment baseline advanced') + check_path = work / ('reapply-' + uuid.uuid4().hex) + try: + reconstructed = artifacts.apply_candidate(work / 'incumbent', check_path, c['generated'], decl.evolution, workflow=c['workflow']) + if reconstructed['tree_sha'] != arm['tree_sha']: raise artifacts.ArtifactError('release mutation proof mismatch') + finally: shutil.rmtree(check_path, ignore_errors=True) + release = ledger_dir_for(settings) / 'evolution' / 'releases' / arm['tree_sha'] + if not release.exists(): + staged = release.with_name('.' + release.name + '-' + uuid.uuid4().hex) + staged.parent.mkdir(parents=True, exist_ok=True) + shutil.copytree(work / 'arm', staged) + artifacts.verify(staged); staged.replace(release) + artifacts.verify(release) + previous = current if current else {'artifact': str(work / 'incumbent'), 'source_sha': incumbent['tree_sha'], 'workflows': [], 'candidate': None} + entry = {'candidate': c['id'], 'artifact': str(release), 'source_sha': arm['tree_sha'], 'at': time.time(), + 'workflows': sorted(set(previous.get('workflows', [])) | set(c.get('affected_workflows', [])) | {c['workflow']}), + 'previous': previous, 'evaluation_sha': artifacts.digest(json.dumps(verdict, sort_keys=True).encode())} + # Persist lineage before activation; recovery accepts exactly this + # candidate pointer and never deploys an unevaluated replacement. + c.update(state='canary', previous_release=previous, activated_at=entry['at'], release=str(release), incumbent_sha=incumbent['tree_sha']) + save(settings, c, store) + artifacts.atomic_json(registry(settings), entry) + store.append('decision', result={'type': 'automatic_activation', 'candidate': c['id'], 'source_sha': arm['tree_sha']}) + return c + + +def rollback(settings, store, c, reason): + from .evolution import save + with file_lock(registry(settings).with_suffix('.lock'), blocking=True) as held: + if not held: raise SandboxUnavailable('rollback registry locked') + current = active(settings) + if current.get('candidate') != c['id']: + c.update(state='rollback-blocked', reason='active release advanced; refusing to overwrite its lineage') + save(settings, c, store); return + try: + previous = current['previous'] + if artifacts.verify(Path(previous['artifact']))['tree_sha'] != previous['source_sha']: + raise artifacts.ArtifactError('previous release fingerprint mismatch') + c['rollback_pending'] = reason + save(settings, c, store) + artifacts.atomic_json(registry(settings), previous) + c.pop('rollback_pending', None) + c.update(state='rolled-back', reason=reason, rolled_back_at=time.time()) + from .ledger import Ledger + ledger = Ledger(ledger_dir_for(settings)) + with ledger.locked(c['workflow']): + row = ledger.load(c['workflow']) + if row.baseline.get('deployed_source', {}).get('candidate') == c['id']: + row.baseline.pop('deployed_source', None) + ledger.save(row) + except (ValueError, OSError, KeyError) as exc: + # Disable the affected executable release if exact restoration is + # unavailable. Subsequent invocations block instead of running it. + artifacts.atomic_json(registry(settings), {**current, 'disabled': True, 'reason': str(exc)}) + c.update(state='rollback-blocked', reason=str(exc)) + save(settings, c, store) + store.append('decision', result={'type': 'automatic_rollback', 'candidate': c['id'], 'state': c['state'], 'reason': reason}) + + +def execute(settings, store, workflow, data, llm, *, sandbox=None, driver_override=None): + if not settings.improve_enabled or not settings.improve_evolve_enabled: return None + with file_lock(registry(settings).with_name('execution.lock'), blocking=True) as held: + if not held: raise SandboxUnavailable('release execution is busy') + return _execute(settings, store, workflow, data, llm, sandbox=sandbox, driver_override=driver_override) + + +def _execute(settings, store, workflow, data, llm, *, sandbox=None, driver_override=None): + """Production prediction. Host stamps metrics/bytes and validates outputs.""" + from .evolution import candidate, save, directory, public_settings + entry = active(settings) + if not entry or workflow not in entry.get('workflows', []): return None + if entry.get('disabled'): raise SandboxUnavailable('active runtime disabled: ' + entry.get('reason', '')) + c = candidate(settings, entry['candidate']) + sandbox = sandbox or Sandbox() + root = Path(entry['artifact']) + try: + if artifacts.verify(root)['tree_sha'] != entry['source_sha']: raise artifacts.ArtifactError('active release changed') + st = objectives.settings_for_source(objectives.execution_settings(settings), root, workflow) + from .enroll import declarations_for + decl = declarations_for(st)[workflow] + driver = driver_override or objectives.DRIVERS.get(decl.experiment_driver) + if driver == 'objective-tests': raise artifacts.ArtifactError('workflow lacks a sandboxed production driver') + before, started = governor_for(st, ledger_dir_for(st)).remaining()['usd_settled'], time.monotonic() + context = {'workflow': workflow, 'unit_id': 'live-evolve:' + uuid.uuid4().hex} + with bind_store(store), trace_context(**context): + result = sandbox.run(root, {'mode': 'predict', 'driver': driver, 'capture_errors': True, 'input': data, 'settings': public_settings(st), 'trace_context': context}, + directory(st) / 'production' / context['unit_id'], llm=llm, settings=st, governor=governor_for(st, ledger_dir_for(st))) + if result.get('source_sha') != entry['source_sha']: raise artifacts.ArtifactError('production worker fingerprint mismatch') + if result.get('execution_error'): raise artifacts.ArtifactError(result['execution_error']) + paired = None + item = str(data.get('repo', '') + '#' + str(data.get('pr', data.get('event_id', '')))) if decl.experiment_driver != 'meta' else str(data['records'][0].get('context', {}).get('unit_id', '')) + eligible = workflow == c['workflow'] and (workflow != objectives.ENGINE or driver == ('objective-lints' if set(c['artifact']['paths']) == {'src/infermatrix_copilot/improve/lints.py'} else 'objective-meta')) + # During canary, compare the actual predecessor on the same frozen + # input. Both calls share the existing budget; no evaluator model. + # Continue checking fresh production items after retention, bounded + # to eight pairs per week and the same shared budget. + from .budget import iso_week + week = iso_week(time.time()) + observe = c['state'] == 'canary' or (c['state'] == 'retained' and c.get('monitor_week') != week) + if observe and c['state'] == 'retained': + c.update(monitor_week=week, monitor_items=[], monitor_records=[]) + observe = c['state'] == 'canary' or (c['state'] == 'retained' and len(c.get('monitor_items', [])) < 8) + observed_items = c.get('canary_items', []) if c['state'] == 'canary' else c.get('monitor_items', []) + if observe and eligible and item not in observed_items: + previous = entry['previous'] + previous_settings = objectives.settings_for_source(objectives.execution_settings(settings), previous['artifact'], workflow) + previous_started = time.monotonic() + settled = governor_for(st, ledger_dir_for(st)).remaining()['usd_settled'] + reference = sandbox.run(Path(previous['artifact']), {'mode': 'predict', 'driver': driver, 'capture_errors': True, 'input': data, + 'settings': public_settings(objectives.settings_for_source(objectives.execution_settings(settings), previous['artifact'], workflow)), 'trace_context': {**context, 'unit_id': 'exp-canary:' + context['unit_id']}}, + directory(st) / 'production' / (context['unit_id'] + '-previous'), llm=llm, settings=previous_settings, governor=governor_for(st, ledger_dir_for(st))) + if reference.get('source_sha') != previous['source_sha']: raise artifacts.ArtifactError('canary predecessor changed') + if decl.experiment_driver in ('pr-review', 'kb-intake') and objectives.canonical(decl.experiment_driver, reference) != objectives.canonical(decl.experiment_driver, result): + raise artifacts.ArtifactError('canary output differs from predecessor; semantic preservation unverified') + arm_usd = settled - before + old_usd = governor_for(st, ledger_dir_for(st)).remaining()['usd_settled'] - settled + paired = {'item': item, 'arm_usd': arm_usd, 'previous_usd': old_usd, + 'arm_seconds': previous_started - started, 'previous_seconds': time.monotonic() - previous_started, + 'previous_calls': reference['_broker_calls']} + c.setdefault('canary_items' if c['state'] == 'canary' else 'monitor_items', []).append(item) + if driver == 'objective-meta': + ids = {r['id'] for r in data['records']} + found = result.get('attributions', []) + if len(found) != len(data['cells']) or {a.get('gold_id') for a in found} != set(data['cells']) or any(a.get('stage') not in ('S0', 'S1', 'S2', 'S3', 'S4', 'S5', 'S6', 'S7', 'S8', 'S9', 'S10') or not set(a.get('evidence', [])).issubset(ids) for a in found): + raise artifacts.ArtifactError('diagnostic output violates the evidence contract') + if decl.experiment_driver in ('pr-review', 'kb-intake'): + # Empty KB drafts may be correct on production inputs. Promotion + # never receives a yield reward for fabricating rules. + valid = objectives.score(decl.experiment_driver, {'payload': data}, result) + if decl.experiment_driver == 'kb-intake' and result.get('operations') == [] and not result.get('rejected'): valid = 1 + if not valid: raise artifacts.ArtifactError('production output violates executable contract') + metrics = {'type': 'runtime_result', 'candidate': c['id'], 'source_sha': entry['source_sha'], 'contract_success': 1, + 'broker_calls': result['_broker_calls'], 'usd': governor_for(st, ledger_dir_for(st)).remaining()['usd_settled'] - before, + 'seconds': time.monotonic() - started, 'pair': paired, 'driver': driver} + record = store.append('decision', context=context, result=metrics) + if c['state'] in ('canary', 'retained') and eligible: + if c['state'] == 'canary': c.setdefault('production_records', []).append(record['id']) + elif paired: c.setdefault('monitor_records', []).append(record['id']) + c['deployed_at'] = c.get('deployed_at', record['at']); c['deployed_record'] = record['id'] + # Actual sandbox execution, rather than activation/PR merge, + # attests the deployed baseline. + from .ledger import Ledger + ledger = Ledger(ledger_dir_for(st)) + with ledger.locked(workflow): + row = ledger.load(workflow) + row.baseline['deployed_source'] = {'candidate': c['id'], 'tree_sha': entry['source_sha'], 'record': record['id'], 'at': record['at']} + ledger.save(row) + save(st, c, store) + return result + except Exception as exc: + # Budget/isolation/validation failures are visible and restore the + # previous pointer; this call blocks instead of replaying a paid call. + rollback(settings, store, c, f'{type(exc).__name__}: {exc}') + raise + + +def sync(settings, store, *, now=None): + with file_lock(registry(settings).with_name('execution.lock'), blocking=True) as held: + if not held: return + return _sync(settings, store, now=now) + + +def _sync(settings, store, *, now=None): + from .evolution import candidates, save + now = time.time() if now is None else now + for c in candidates(settings): + if c['state'] not in ('canary', 'retained'): continue + entry = active(settings) + if c.get('rollback_pending') and entry.get('source_sha') == c.get('previous_release', {}).get('source_sha'): + c.update(state='rolled-back', reason=c.pop('rollback_pending'), rolled_back_at=now) + save(settings, c, store); continue + if entry.get('candidate') != c['id']: + # Recover a crash after saving activation intent but before the + # registry pointer. Revalidation precedes any later activation. + if c['state'] == 'canary' and not c.get('deployed_record') and (not entry or entry.get('source_sha') == c.get('incumbent_sha')): + c.update(state='pr-ready', reason='activation interrupted; exact release will be revalidated') + save(settings, c, store) + continue + if now - c['activated_at'] >= 7 * 86400 and c['state'] == 'canary': + rollback(settings, store, c, 'canary lacks verified production evidence after seven days'); continue + records = [store.get(rid) for rid in c.get('monitor_records' if c['state'] == 'retained' else 'production_records', [])] + verified = [r for r in records if r.get('result', {}).get('source_sha') == entry['source_sha'] and r['result'].get('contract_success') == 1] + c['observation'] = {'state': 'waiting-for-production-samples', 'verified_units': len(verified), 'semantic_quality_claim': False} + if len(verified) >= 8: + from . import stats + pairs = [r['result']['pair'] for r in verified if r['result'].get('pair')] + deltas = {} + for pair in pairs: + a, b = pair['arm_usd'], pair['previous_usd'] + if b <= 0: a, b = pair['arm_seconds'], pair['previous_seconds'] + deltas.setdefault(pair['item'], []).append((b - a) / max(b, 1e-9)) + measured = stats.paired(deltas, 'production_resource_gain') + c['observation']['resource_gain'] = {'mean': measured.mean, 'lo': measured.lo, 'hi': measured.hi, 'n_items': measured.n_items} + if measured.n_items >= 8 and measured.hi < 0: + rollback(settings, store, c, 'statistically significant production resource regression'); continue + if measured.n_items >= 8: + c.update(state='retained', reason='', retained_at=now) + c['observation']['state'] = 'production-contracts-verified' + save(settings, c, store) + + +def freeze_review(settings, state): + """Freeze local Git objects and knowledge; never read PR checkout symlinks.""" + import re + import subprocess + repository = Path(state['repo_path']) + revisions = [state['pr_base_sha'], state['pr_head_sha']] + if any(not re.fullmatch(r'[a-fA-F0-9]{40}', revision) for revision in revisions): raise artifacts.ArtifactError('review replay needs fixed base/head SHAs') + trees, total = [], 0 + for revision in revisions: + listing = subprocess.check_output(['git', '-C', str(repository), 'ls-tree', '-rz', '--full-tree', revision], timeout=60) + files = {} + for record in listing.split(b'\0'): + if not record: continue + meta, name = record.split(b'\t', 1) + mode, kind, oid = meta.split() + name = name.decode() + # Refuse credential/config payloads and binary/oversize material + # rather than silently replaying a different repository. + if mode == b'120000' or kind != b'blob': raise artifacts.ArtifactError('review snapshot contains non-regular source') + from .drivers import input_path + input_path(name) + if any(part == '.env' or part.startswith('.env.') for part in Path(name).parts): raise artifacts.ArtifactError('review snapshot contains credential files') + size = int(subprocess.check_output(['git', '-C', str(repository), 'cat-file', '-s', oid.decode()], timeout=20)) + total += size + if size > 1_000_000 or total > 12_000_000: raise artifacts.ArtifactError('review snapshot exceeds replay bound') + files[name] = subprocess.check_output(['git', '-C', str(repository), 'cat-file', 'blob', oid.decode()], timeout=20).decode('utf-8') + trees.append(files) + knowledge = {} + kroot = Path(settings.knowledge_dir).resolve() + if kroot.is_dir(): + for p in sorted(kroot.rglob('*.md')): + if p.is_symlink() or not p.resolve().is_relative_to(kroot): raise artifacts.ArtifactError('knowledge snapshot path escaped') + total += p.stat().st_size + if total > 16_000_000: raise artifacts.ArtifactError('knowledge snapshot exceeds replay bound') + knowledge[p.relative_to(kroot).as_posix()] = p.read_text() + spec = state['task_spec'] + return {'repo': spec['repo'], 'pr': spec['pr'], 'base_sha': revisions[0], 'head_sha': revisions[1], + 'base_files': trees[0], 'head_files': trees[1], 'diff': state['diff_text'], 'knowledge_files': knowledge, + 'context_text': state.get('pr_context', ''), 'gate_report': state.get('gate_report', ''), + 'task_params': spec.get('params', {}), 'pr_state': state.get('pr_state', '')} + + +async def review_step(ctx): + """Use an active sandbox release for the read-only review step.""" + from ..engine.step import StepResult, FailureKind + from ..trace_store import current_store + store = current_store() + if not store: return None + workflow = 'pr-review.agent.review_diff' + entry = active(ctx.settings) + is_active = workflow in entry.get('workflows', []) + try: + data = freeze_review(ctx.settings, ctx.state) + item = f"{data['repo']}#{data['pr']}@{data['head_sha']}" + objectives.capture(store, workflow, item, data) + except (ValueError, OSError, KeyError) as exc: + store.append('decision', result={'type': 'replay_capture_missing', 'workflow': workflow, 'reason': str(exc)}) + if is_active: + from .evolution import candidate + rollback(ctx.settings, store, candidate(ctx.settings, entry['candidate']), 'review replay capture failed: ' + str(exc)) + return StepResult(False, FailureKind.BLOCKED, 'active review lacks a safe frozen replay input') + return None + if not is_active: return None + import asyncio + try: + prediction = await asyncio.to_thread(execute, ctx.settings, store, workflow, data, ctx.llm) + except Exception as exc: + return StepResult(False, FailureKind.BLOCKED, 'autonomous review rolled back: ' + str(exc)) + if prediction is None: return None + outputs = {k: prediction.get(k) for k in ('review_text', 'review_comments', 'review_summary', 'review_verdict', + 'review_finding_dispositions', 'review_carried_findings', 'review_finding_rechecks', 'review_recheck_missing')} + ctx.state.update(outputs) + return StepResult(True, summary='review produced by verified autonomous source', outputs={**outputs, 'state_updates': outputs}) + + +def run_lints(unit, store, baseline): + """Production lint code is loaded from the active release in a sandbox.""" + from .lints import run_lints as original, Finding, catalogue + settings = baseline.settings + entry = active(settings) if settings else {} + workflow = objectives.ENGINE + if not settings or workflow not in entry.get('workflows', []): return original(unit, store, baseline) + from .evolution import candidate + c = candidate(settings, entry['candidate']) + try: + blobs = {ref: store.blob(ref) for r in unit.records for ref in [*r.get('inputs', {}).values(), *r.get('outputs', {}).values()]} + data = {'records': unit.records, 'blobs': blobs, 'cells': [], 'concerns': [], 'baseline': {'usd': baseline.usd, 'seconds': baseline.seconds, 'parse_failure_rate': baseline.parse_failure_rate}} + output = execute(settings, store, workflow, data, None, driver_override='objective-lints') + known, ids = {r['id'] for r in catalogue()}, {r['id'] for r in unit.records} + findings = [Finding(**f) for f in output['findings']] + if any(f.lint not in known or f.unit_id != unit.unit_id or f.workflow != unit.workflow or not f.evidence or not set(f.evidence).issubset(ids) for f in findings): + raise artifacts.ArtifactError('active lint returned invalid evidence') + expected = {(f.lint, tuple(f.evidence)) for f in original(unit, store, baseline)} + if not expected.issubset({(f.lint, tuple(f.evidence)) for f in findings}): raise artifacts.ArtifactError('active lint lost an existing positive') + return findings + except Exception as exc: + if active(settings).get('candidate') == c['id']: rollback(settings, store, c, 'production lint failed: ' + str(exc)) + # Diagnostics continue using the exact preceding host code; generated + # code is never executed outside the sandbox. + store.append('decision', result={'type': 'runtime_lint_restored', 'reason': str(exc)}) + return original(unit, store, baseline) + + +def diagnose(settings, store, llm=None): + """One-model hypotheses on observed mechanical defects, without gold labels.""" + if objectives.ENGINE not in active(settings).get('workflows', []): return {'state': 'no-active-diagnostic-release'} + from .reader import units_between + from .lints import run_lints as original, Baseline + from ..llm import LLM + report = [] + for unit in units_between(store, time.time() - 7 * 86400, time.time()).values(): + if len(report) >= 3: break + if unit.playbook == 'workflow-improve' or unit.unit_id.startswith(('exp-', 'live-evolve:')): continue + findings = original(unit, store, Baseline(settings=settings)) + if not findings: continue + try: + blobs = {ref: store.blob(ref) for r in unit.records for ref in [*r.get('inputs', {}).values(), *r.get('outputs', {}).values()]} + except (ValueError, OSError) as exc: + store.append('decision', result={'type': 'diagnosis_input_missing', 'unit': unit.unit_id, 'reason': str(exc)}) + continue + concerns = [{'gold_id': 'observed-' + str(i), 'path': 'runtime', 'concern': f'{f.lint}: {f.detail}'} for i, f in enumerate(findings[:3])] + data = {'records': unit.records, 'blobs': blobs, 'concerns': concerns, 'cells': [g['gold_id'] for g in concerns]} + result = execute(settings, store, objectives.ENGINE, data, llm or LLM(objectives.execution_settings(settings))) + store.append('decision', context={'workflow': objectives.ENGINE}, result={'type': 'objective_diagnosis', 'source_unit': unit.unit_id, + 'hypotheses': result['attributions'], 'semantic_quality_claim': False}) + report.append(unit.unit_id) + return {'state': 'diagnosed', 'units': report, 'semantic_quality_claim': False} diff --git a/src/infermatrix_copilot/improve/worker.py b/src/infermatrix_copilot/improve/worker.py index 1683e710..a0cf26dd 100644 --- a/src/infermatrix_copilot/improve/worker.py +++ b/src/infermatrix_copilot/improve/worker.py @@ -43,8 +43,17 @@ def for_target(self, target): import pytest result["rc"] = pytest.main(["-q", "-p", "no:cacheprovider", "--confcutdir=/tests", *["/tests/" + p.removeprefix("test/") for p in payload["tests"]]]) else: - from infermatrix_copilot.improve.drivers import worker_run - result = worker_run(payload, settings, LLM(settings)) + try: + if payload.get("driver", "").startswith("objective-"): + from infermatrix_copilot.improve.objectives import worker + result = worker(payload, settings, LLM(settings)) + else: + from infermatrix_copilot.improve.drivers import worker_run + result = worker_run(payload, settings, LLM(settings)) + except Exception as exc: + if not payload.get("capture_errors"): + raise + result = {"execution_error": f"{type(exc).__name__}: {str(exc)[:500]}"} result.update(source_sha=payload["source_sha"], package_path=str(infermatrix_copilot.__file__)) wire.write(json.dumps({"type": "result", "result": result}, ensure_ascii=False) + "\n"); wire.flush() diff --git a/src/infermatrix_copilot/improve/workflows/workflow-improve.yaml b/src/infermatrix_copilot/improve/workflows/workflow-improve.yaml index f59b654f..90b5b718 100644 --- a/src/infermatrix_copilot/improve/workflows/workflow-improve.yaml +++ b/src/infermatrix_copilot/improve/workflows/workflow-improve.yaml @@ -1,13 +1,5 @@ -# The engine's own forensics role (design §11.2, self-enrolment): every -# `improve.forensics` step call is a unit — a cycle's Tier 2 pass in -# production, or ONE frozen meta-benchmark case in a shadow self-experiment -# (`--task-param meta_case=`, item `meta:`). The outcome adapter -# scores a meta case by the stage labels' agreement with the HUMAN labels -# (recall_gold = share of cells agreed, kappa_review = Cohen's kappa) and the -# Tier 1 catalogue's recall on the injected lint samples (lint_recall_review). -# No judge is involved: the labels are human, so the cross-family rule for -# self-experiments is satisfied by construction. The benchmark is read-only -# to the engine; a fingerprint diff touching it refuses registration. +# Self-enrolled engine. Autonomous objective mode uses controller interventions +# and one fixed model; the adapter below preserves optional gold-mode support. workflow: workflow-improve.improve.forensics kind: dynamic unit: step_call diff --git a/src/infermatrix_copilot/kb_service/runtime.py b/src/infermatrix_copilot/kb_service/runtime.py index 9fd67ae5..90b14f44 100644 --- a/src/infermatrix_copilot/kb_service/runtime.py +++ b/src/infermatrix_copilot/kb_service/runtime.py @@ -250,11 +250,30 @@ def _run_intake_locked(rt: KbRuntime, lifecycle: RepoLifecycle, *, max_events: i from ..improve.artifacts import runtime_settings draft_settings = runtime_settings(getattr(rt, "settings", None) or Settings(), "kb-intake.draft") with trace_context(draft_key=key, _accepted=holder, step="draft"): - drafts.append(draft_changes( - repo=lifecycle.repo, repo_dir=lifecycle.knowledge_dir, event_id=event["id"], - evidence=event["payload"], files=base, gateway=rt.gateway, - generator=rt.generator, release=release, today=today, - max_operations=max(1, min(6, draft_settings.kb_draft_max_operations)))) + from ..improve import objectives + prediction = None + if objectives.enabled(draft_settings) and draft_settings.improve_enabled and draft_settings.improve_evolve_enabled and rt.traces is not None: + from ..improve.runtime import execute + payload = {"repo": lifecycle.repo, "repo_dir": lifecycle.knowledge_dir, "event_id": event["id"], + "evidence": event["payload"], "files": base, "release": release, "today": today, + "external_texts": external, "generator_model": rt.generator.model} + objectives.capture(rt.traces, "kb-intake.draft", f"{lifecycle.repo}#{event['id']}", payload) + from ..llm import LLM + try: + prediction = execute(draft_settings, rt.traces, "kb-intake.draft", payload, LLM(draft_settings)) + except Exception as exc: + raise ModelUnavailable("autonomous draft rolled back: " + str(exc)) from exc + if prediction is not None: + from .intake import Draft + from ..knowledge_service.ops import KnowledgeOperation + drafts.append(Draft(event_ids=[event["id"]], operations=[KnowledgeOperation.from_dict(o) for o in prediction["operations"]], + result=None, rationale=prediction.get("rationale", ""), rejected=prediction.get("rejected", False))) + else: + drafts.append(draft_changes( + repo=lifecycle.repo, repo_dir=lifecycle.knowledge_dir, event_id=event["id"], + evidence=event["payload"], files=base, gateway=rt.gateway, + generator=rt.generator, release=release, today=today, + max_operations=max(1, min(6, draft_settings.kb_draft_max_operations)))) if accepted_key(key, holder): accepted[event["id"]] = accepted_key(key, holder) except ModelUnavailable as exc: diff --git a/test/test_autonomous_evolution.py b/test/test_autonomous_evolution.py new file mode 100644 index 00000000..4b3492c4 --- /dev/null +++ b/test/test_autonomous_evolution.py @@ -0,0 +1,460 @@ +"""Autonomous objectives, deployed bytes, and recovery without judges/annotations.""" +from __future__ import annotations +import copy +import json +from pathlib import Path +from types import SimpleNamespace +import pytest +import yaml + +from test_evolution import bench, Model +from infermatrix_copilot.improve import artifacts, objectives, runtime, evolution +from infermatrix_copilot.improve.isolation import SandboxUnavailable +from infermatrix_copilot.config import Settings + + +class ObjectiveWorker: + def __init__(self): self.calls = []; self.bad = False; self.interrupt = False + def check(self): return {'ready': True, 'reason': ''} + def run(self, source, payload, work, **kwargs): + self.calls.append(payload) + assert 'objective' not in payload['input'] if payload['mode'] != 'tests' else True + meta = artifacts.verify(source) + arm = '# improved' in (source / 'src/infermatrix_copilot/improve/forensics.py').read_text() + result = {'source_sha': meta['tree_sha'], 'package_path': '/candidate/src/infermatrix_copilot/__init__.py', '_broker_calls': 1} + if payload['mode'] == 'tests': return {**result, 'rc': 0} + if self.interrupt: + self.interrupt = False + raise KeyboardInterrupt() + if self.bad: return {**result, 'execution_error': 'injected failure'} + data = payload['input'] + if payload['driver'] == 'objective-lints': return {**result, 'findings': []} + stage, evidence = 'S0', [] + for record in data['records']: + kind = record.get('result', {}).get('type') + if record.get('result', {}).get('stop_reason') == 'max_tokens': stage, evidence = 'S4', [record['id']] + if kind == 'cap': stage, evidence = 'S3', [record['id']] + if kind == 'planner_zero_output': stage, evidence = 'S7', [record['id']] + if not arm: stage, evidence = 'S0', [] + return {**result, 'scores': {'contract_success': 999}, 'attributions': [dict(gold_id=gid, stage=stage, evidence=evidence, disputed=False) for gid in data['cells']]} + + +@pytest.fixture +def autonomous(bench): + st, store, _, model = bench() + st = st.model_copy(update={'improve_evaluation_mode': 'objective', 'improve_promotion_mode': 'pr', 'improve_judge': '', 'performance_model': ''}) + path = Path(st.improve_workflows_dirs) / 'workflow.yaml' + doc = yaml.safe_load(path.read_text()); doc['objective'] = {'min_effect': .5, 'prior_sd': .15} + path.write_text(yaml.safe_dump(doc)) + root = objectives.data_root(st) / objectives.ENGINE + (root / 'dataset.json').unlink() + objectives.prepare(st, store, workflow=objectives.ENGINE) + manifest = json.loads((root / 'dataset.json').read_text()) + # A deterministic positive cohort for the large-effect integration case; + # negative controls are independently asserted below. + manifest['items'] = [r for r in manifest['items'] if not r['objective'].get('negative')] + artifacts.atomic_json(root / 'dataset.json', manifest) + return st, store, model, ObjectiveWorker() + + +def test_default_is_one_model_objective_and_automatic(): + st = Settings(_env_file=None, eco_model='claude-sonnet-5', performance_model='') + assert st.improve_evaluation_mode == 'objective' and st.improve_promotion_mode == 'automatic' + single = objectives.execution_settings(st) + assert single.tier_target('eco').model == single.tier_target('performance').model == 'claude-sonnet-5' + + +def test_readiness_has_no_human_gold_or_judge_requirement(autonomous): + st, store, _, worker = autonomous + report = evolution.check(st, store, objectives.ENGINE, sandbox=worker) + assert report['ready'], report + assert report['evaluation_mode'] == 'objective' + assert not any('human' in r or 'gold' in r or 'judge' in r for r in report['reasons']) + assert not report['semantic_quality_claim'] + + +@pytest.mark.parametrize('fault', objectives.FAULTS) +def test_causal_witness_scores_complete_denominator_and_negative_controls(fault): + data, clean, witness = objectives.fault_case('abcdef0123456789', fault) + row = {'payload': data, 'objective': witness, 'origin': {'kind': 'controlled-intervention'}} + correct = {'attributions': [{'gold_id': witness['cell'], 'stage': witness['stage'], 'evidence': [witness['evidence']]}]} + assert objectives.score('meta', row, correct) == 1 + assert objectives.score('meta', row, {'scores': {'accuracy': 1}, 'attributions': []}) == 0 + wrong = copy.deepcopy(correct); wrong['attributions'][0]['evidence'] = ['not-in-trace'] + assert objectives.score('meta', row, wrong) == 0 + negative = {'payload': clean, 'objective': {'negative': True}, 'origin': row['origin']} + assert objectives.score('meta', negative, {'attributions': []}) == 1 + assert objectives.score('meta', negative, correct) == 0 + + +def test_generate_evaluate_with_one_model_no_judge_no_annotations(autonomous): + st, store, model, worker = autonomous + result = evolution.run(st, store, workflow=objectives.ENGINE, llm=model, sandbox=worker) + assert result['state'] == 'pr-ready', result + assert result['evaluation']['promotable'] and result['evaluation']['protocol'] == objectives.PROTOCOL + assert result['evaluation']['replicates'] == 3 + assert result['evaluation']['n_items'] >= 8 + assert len(model.calls) == 1 + assert model.calls[0]['model'] == st.eco_model + assert all(p.get('settings', {}).get('performance_model') == st.eco_model for p in worker.calls) + assert not result['evaluation']['semantic_quality_claim'] + path = evolution.directory(st) / 'candidates' / result['id'] / 'PR.md' + assert path.exists() + + +def verified(autonomous): + st, store, model, worker = autonomous + c = evolution.run(st, store, workflow=objectives.ENGINE, llm=model, sandbox=worker) + assert c['state'] == 'pr-ready', c + return st.model_copy(update={'improve_promotion_mode': 'automatic'}), store, model, worker, c + + +def test_automatic_activation_reuses_exact_artifact_without_pr(autonomous): + st, store, model, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + assert c['state'] == 'canary' + entry = runtime.active(st) + assert entry['source_sha'] == c['artifact']['tree_sha'] and not c.get('pr') + assert runtime.baseline(st, c['workflow'], Path(st.improve_ledger_dir) / 'next-base')['tree_sha'] == entry['source_sha'] + runtime.promote(st, store, c, sandbox=worker) + assert runtime.active(st)['candidate'] == c['id'] + assert len(model.calls) == 1 + + +def test_tampered_verdict_cannot_activate(autonomous): + st, store, _, worker, c = verified(autonomous) + c['evaluation']['mean'] = 900 + with pytest.raises(artifacts.ArtifactError, match='persisted controller verdict'): + runtime.promote(st, store, c, sandbox=worker) + assert not runtime.active(st) + + +def test_live_failure_restores_exact_previous_bytes(autonomous): + st, store, model, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + previous = runtime.active(st)['previous']['source_sha'] + data, _, _ = objectives.fault_case('abcdef0123456789', 'ceiling') + worker.bad = True + with pytest.raises(artifacts.ArtifactError): runtime.execute(st, store, c['workflow'], data, model, sandbox=worker) + assert runtime.active(st)['source_sha'] == previous + assert evolution.candidate(st, c['id'])['state'] == 'rolled-back' + + +def test_merged_or_activated_does_not_attest_deployment(autonomous): + st, store, model, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + assert not c.get('deployed_record') + data, _, _ = objectives.fault_case('abcdef0123456789', 'ceiling') + runtime.execute(st, store, c['workflow'], data, model, sandbox=worker) + updated = evolution.candidate(st, c['id']) + record = store.get(updated['deployed_record']) + assert record['result']['source_sha'] == c['artifact']['tree_sha'] + assert record['result']['pair']['previous_calls'] == 1 + + +def test_canary_timeout_rolls_back_without_human(autonomous): + st, store, _, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + runtime.sync(st, store, now=c['activated_at'] + 8 * 86400) + assert evolution.candidate(st, c['id'])['state'] == 'rolled-back' + + +def test_eight_distinct_verified_production_pairs_retain_release(autonomous): + st, store, model, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + # Deterministic clock removes process timing noise from this retention test. + from unittest.mock import patch + counter = iter(range(10000)) + with patch.object(runtime.time, 'monotonic', side_effect=lambda: next(counter)): + for i in range(8): + data, _, _ = objectives.fault_case(f'{i:016x}', 'discard') + runtime.execute(st, store, c['workflow'], data, model, sandbox=worker) + runtime.sync(st, store) + updated = evolution.candidate(st, c['id']) + assert updated['state'] == 'retained', updated + assert updated['observation']['resource_gain']['n_items'] == 8 + + +def test_interrupted_paid_pair_is_excluded_and_never_replayed(autonomous): + st, store, model, worker = autonomous + worker.interrupt = True + with pytest.raises(KeyboardInterrupt): evolution.run(st, store, workflow=objectives.ENGINE, llm=model, sandbox=worker) + result = evolution.run(st, store, workflow=objectives.ENGINE, llm=model, sandbox=worker) + assert len(model.calls) == 1 + exp = json.loads((evolution.directory(st) / 'candidates' / result['id'] / 'objective.json').read_text()) + assert any(r['state'] == 'excluded' for r in exp['progress'].values()) + assert not result['evaluation']['promotable'] + + +def test_trace_inputs_are_automatically_frozen_and_clustered(tmp_path): + from infermatrix_copilot.trace_store import TraceStore + st = Settings(_env_file=None, improve_ledger_dir=str(tmp_path / 'ledger')) + store = TraceStore(tmp_path / 'trace') + for _ in range(3): objectives.capture(store, 'kb-intake.draft', 'repo#1', {'files': {}, 'repo': 'repo'}) + assert objectives.collect(st, store, 'kb-intake.draft') == 1 + assert len(objectives.rows(st, 'kb-intake.draft')) == 1 + row = objectives.rows(st, 'kb-intake.draft')[0] + (objectives.data_root(st) / 'kb-intake.draft' / row['input']).write_text('{}') + with pytest.raises(artifacts.ArtifactError): objectives.rows(st, 'kb-intake.draft') + + +def test_empty_kb_is_not_rewarded_as_an_error(): + row = {'payload': {'files': {}, 'repo_dir': 'repos/demo', 'release': 'v1', 'today': '2026-10-04'}} + assert objectives.score('kb-intake', row, {'operations': [], 'rejected': False}) == 1 + assert objectives.score('kb-intake', row, {'operations': [], 'rejected': True}) == 0 + + +def test_resource_change_preserves_carried_findings_and_verdict(): + original = {'review_comments': [], 'review_verdict': 'APPROVE', 'review_recheck_missing': ['id']} + modified = {**original, 'review_recheck_missing': []} + assert objectives.canonical('pr-review', original) != objectives.canonical('pr-review', modified) + + +def test_kill_switch_prevents_live_candidate_execution(autonomous): + st, store, model, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + calls = len(worker.calls) + assert runtime.execute(st.model_copy(update={'improve_enabled': False}), store, c['workflow'], {}, model, sandbox=worker) is None + assert len(worker.calls) == calls + + +def test_injected_lint_faults_and_clean_controls_are_real(tmp_path): + from infermatrix_copilot.trace_store import TraceStore + from infermatrix_copilot.improve.meta import _unit_from + from infermatrix_copilot.improve.lints import run_lints, Baseline + for fault in objectives.FAULTS: + positive, clean, witness = objectives.fault_case('abcdef0123456789', fault) + store = TraceStore(tmp_path / fault) + for ref, text in positive['blobs'].items(): assert store.put_blob(text) == ref + findings = run_lints(_unit_from(positive['records']), store, Baseline(settings=Settings(_env_file=None))) + assert witness['lint'] in {f.lint for f in findings} + assert not run_lints(_unit_from(clean['records']), store, Baseline(settings=Settings(_env_file=None))) + + +def test_three_objective_workflows_need_no_labels_or_judge(bench): + for driver, workflow in [('pr-review', 'pr-review.agent.review_diff'), ('kb-intake', 'kb-intake.draft')]: + st, store, _, model = bench(driver, workflow) + st = st.model_copy(update={'improve_evaluation_mode': 'objective', 'improve_promotion_mode': 'pr', 'improve_judge': '', 'performance_model': ''}) + decl_path = Path(st.improve_workflows_dirs) / 'workflow.yaml' + doc = yaml.safe_load(decl_path.read_text()); doc['objective'] = {'min_effect': .5, 'prior_sd': .15}; decl_path.write_text(yaml.safe_dump(doc)) + manifest_path = objectives.data_root(st) / workflow / 'dataset.json' + manifest = json.loads(manifest_path.read_text()) + for row in manifest['items']: + for key in ('labels', 'label_source', 'gold'): row.pop(key, None) + artifacts.atomic_json(manifest_path, manifest) + class Worker: + def check(self): return {'ready': True, 'reason': ''} + def run(self, source, payload, work, **kwargs): + meta = artifacts.verify(source) + improved = '# improved' in next((source / 'src').rglob('*.py')).read_text() + result = {'source_sha': meta['tree_sha'], '_broker_calls': 1} + if payload['mode'] == 'tests': return {**result, 'rc': 0} + if driver == 'pr-review': return {**result, 'review_text': 'complete report' if improved else '', 'review_comments': []} + return {**result, 'operations': [], 'rejected': not improved} + result = evolution.run(st, store, workflow=workflow, llm=model, sandbox=Worker()) + assert result['state'] == 'pr-ready', result + assert result['evaluation']['promotable'] and not result['evaluation']['semantic_quality_claim'] + assert len(model.calls) == 1 + + +def test_active_source_tampering_rolls_back_without_running_it(autonomous): + st, store, model, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + root = Path(runtime.active(st)['artifact']) + (root / 'src/infermatrix_copilot/improve/forensics.py').write_text('tampered') + calls = len(worker.calls) + data, _, _ = objectives.fault_case('abcdef0123456789', 'ceiling') + with pytest.raises(artifacts.ArtifactError): runtime.execute(st, store, c['workflow'], data, model, sandbox=worker) + assert len(worker.calls) == calls + assert evolution.candidate(st, c['id'])['state'] == 'rolled-back' + + +def test_rollback_failure_disables_release_instead_of_running_unverified_bytes(autonomous): + st, store, _, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + previous = Path(runtime.active(st)['previous']['artifact']) + (previous / 'src/infermatrix_copilot/improve/forensics.py').write_text('corrupt baseline') + runtime.rollback(st, store, c, 'regression') + assert runtime.active(st)['disabled'] and c['state'] == 'rollback-blocked' + + +def test_isolation_loss_blocks_automatic_promotion(autonomous): + st, store, _, worker, c = verified(autonomous) + worker.check = lambda: {'ready': False, 'reason': 'user namespaces disabled'} + runtime.promote(st, store, c, sandbox=worker) + assert c['state'] == 'deferred' and not runtime.active(st) + + +def test_better_model_override_is_rejected_before_evaluation(autonomous): + st, store, model, worker = autonomous + model.proposal['overrides'] = {'PERFORMANCE_MODEL': 'stronger-model'} + result = evolution.run(st, store, workflow=objectives.ENGINE, llm=model, sandbox=worker) + assert result['state'] == 'rejected' and len(worker.calls) == 0 + + +def test_canary_production_resource_regression_is_automatically_reverted(autonomous): + st, store, _, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + c['production_records'] = [] + for i in range(8): + r = store.append('decision', result={'type': 'runtime_result', 'source_sha': c['artifact']['tree_sha'], 'contract_success': 1, + 'pair': {'item': str(i), 'arm_usd': 2, 'previous_usd': 1, 'arm_seconds': 1, 'previous_seconds': 1}}) + c['production_records'].append(r['id']) + evolution.save(st, c, store) + runtime.sync(st, store) + assert evolution.candidate(st, c['id'])['state'] == 'rolled-back' + + +def test_retained_release_keeps_bounded_weekly_shadow_comparison(autonomous): + from unittest.mock import patch + st, store, model, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + c.update(state='retained', retained_at=c['activated_at']) + evolution.save(st, c, store) + counter = iter(range(10000)) + before = len(worker.calls) + with patch.object(runtime.time, 'monotonic', side_effect=lambda: next(counter)): + for i in range(10): + data, _, _ = objectives.fault_case(f'{i:016x}', 'discard') + runtime.execute(st, store, c['workflow'], data, model, sandbox=worker) + assert len(worker.calls) - before == 18 # ten live calls + eight shadow pairs + updated = evolution.candidate(st, c['id']) + assert len(updated['monitor_records']) == 8 + runtime.sync(st, store) + assert evolution.candidate(st, c['id'])['state'] == 'retained' + + +def test_activation_intent_recovers_before_registry_write(autonomous): + st, store, _, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + previous = runtime.active(st)['previous'] + artifacts.atomic_json(runtime.registry(st), previous) + runtime.sync(st, store) + assert evolution.candidate(st, c['id'])['state'] == 'pr-ready' + runtime.promote(st, store, evolution.candidate(st, c['id']), sandbox=worker) + assert runtime.active(st)['candidate'] == c['id'] + + +def test_rollback_intent_recovers_after_registry_restore(autonomous): + st, store, _, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + previous = runtime.active(st)['previous'] + c['rollback_pending'] = 'interrupted restore' + evolution.save(st, c, store) + artifacts.atomic_json(runtime.registry(st), previous) + runtime.sync(st, store) + assert evolution.candidate(st, c['id'])['state'] == 'rolled-back' + + +def test_underpowered_resume_uses_fresh_scenarios_without_regeneration(autonomous): + st, store, model, worker = autonomous + worker.interrupt = True + with pytest.raises(KeyboardInterrupt): evolution.run(st, store, workflow=objectives.ENGINE, llm=model, sandbox=worker) + first = evolution.run(st, store, workflow=objectives.ENGINE, llm=model, sandbox=worker) + assert first['evaluation']['blocking'] == 'data' + first_ids = set(first['evaluation']['snapshot_versions']) + objectives.prepare(st, store, workflow=objectives.ENGINE) + manifest_path = objectives.data_root(st) / objectives.ENGINE / 'dataset.json' + manifest = json.loads(manifest_path.read_text()) + manifest['items'] = [r for r in manifest['items'] if not r['objective'].get('negative')] + artifacts.atomic_json(manifest_path, manifest) + second = evolution.run(st, store, workflow=objectives.ENGINE, llm=model, sandbox=worker) + assert second['state'] == 'pr-ready' and len(model.calls) == 1 + assert not first_ids.intersection(second['evaluation']['snapshot_versions']) + assert len(second['objective_trials']) == 1 + + +def test_mixed_routes_are_explicitly_deferred(autonomous): + st, store, model, worker = autonomous + st = st.model_copy(update={'review_lens_backends': {'correctness': 'harness'}}) + result = evolution.run(st, store, workflow=objectives.ENGINE, llm=model, sandbox=worker) + assert result['state'] == 'deferred' and not worker.calls and not model.calls + + +def test_objective_score_rejects_extra_predictions_and_report_loss(): + data, _, witness = objectives.fault_case('abcdef0123456789', 'ceiling') + row = {'payload': data, 'objective': witness, 'origin': {'kind': 'controlled-intervention'}} + a = {'gold_id': witness['cell'], 'stage': witness['stage'], 'evidence': [witness['evidence']]} + assert objectives.score('meta', row, {'attributions': [a, {**a, 'gold_id': 'extra'}]}) == 0 + base = {'review_text': 'A supported finding', 'review_comments': []} + assert objectives.canonical('pr-review', base) != objectives.canonical('pr-review', {**base, 'review_text': 'No finding'}) + + +def test_kb_objective_refuses_rename_outside_owner(): + row = {'payload': {'files': {}, 'repo_dir': 'repos/demo', 'release': 'v1', 'today': '2026-10-04'}} + output = {'operations': [{'kind': 'rename_page', 'page': 'repos/demo/rules.md', 'new_page': 'repos/other/rules.md'}]} + assert objectives.score('kb-intake', row, output) == 0 + + +def test_optional_audit_pr_does_not_interrupt_canary(autonomous): + from infermatrix_copilot.improve import evolution_publish + st, store, _, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + outbox = Path(st.improve_ledger_dir) / 'audit' + st = st.model_copy(update={'improve_evolve_outbox_dir': str(outbox)}) + artifacts.atomic_json(outbox / 'acks' / (c['id'] + '.json'), {'candidate': c['id'], 'source_sha': c['artifact']['tree_sha'], + 'ok': True, 'url': 'https://example.test/audit/1'}) + evolution_publish.sync(st, store) + updated = evolution.candidate(st, c['id']) + assert updated['state'] == 'canary' and updated['pr'].endswith('/1') + + +def test_objective_coordinator_never_calls_legacy_dual_model_forensics(autonomous, monkeypatch): + from infermatrix_copilot.improve import coordinator + from infermatrix_copilot.engine.steps import improve + st, store, model, worker = autonomous + def forbidden(*args, **kwargs): raise AssertionError('legacy model or human benchmark requested') + monkeypatch.setattr(improve, '_forensics', forbidden) + result = coordinator.run(st, store, workflow=objectives.ENGINE, llm=model, sandbox=worker) + assert result['stages']['evolve']['state'] == 'pr-ready', result + count = len(model.calls) + resumed = coordinator.run(st, store, workflow=objectives.ENGINE, llm=model, sandbox=worker) + assert resumed['state'] == 'complete' and len(model.calls) == count + + +def test_interrupted_generation_is_retired_without_manual_unblocking(autonomous): + st, store, _, worker = autonomous + class Interrupted: + def create(self, **kwargs): raise KeyboardInterrupt() + with pytest.raises(KeyboardInterrupt): evolution.run(st, store, workflow=objectives.ENGINE, llm=Interrupted(), sandbox=worker) + result = evolution.run(st, store, workflow=objectives.ENGINE, llm=Interrupted(), sandbox=worker) + assert result['state'] == 'rejected' and result['generation_interrupted'] + assert evolution.run(st, store, workflow=objectives.ENGINE, llm=Interrupted(), sandbox=worker)['reason'] == 'weekly-candidate-limit' + + +def test_unchanged_lints_cannot_consume_attribution_canary_items(autonomous): + st, store, model, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + data, _, _ = objectives.fault_case('abcdef0123456789', 'ceiling') + count = len(worker.calls) + runtime.execute(st, store, c['workflow'], data, model, sandbox=worker, driver_override='objective-lints') + assert len(worker.calls) == count + 1 + runtime.execute(st, store, c['workflow'], data, model, sandbox=worker) + updated = evolution.candidate(st, c['id']) + assert len(updated['canary_items']) == 1 and len(updated['production_records']) == 1 + assert store.get(updated['production_records'][0])['result']['pair'] + + +def test_rollback_clears_current_deployment_attestation(autonomous): + from infermatrix_copilot.improve.ledger import Ledger + st, store, model, worker, c = verified(autonomous) + runtime.promote(st, store, c, sandbox=worker) + data, _, _ = objectives.fault_case('abcdef0123456789', 'ceiling') + runtime.execute(st, store, c['workflow'], data, model, sandbox=worker) + assert Ledger(Path(st.improve_ledger_dir)).load(c['workflow']).baseline['deployed_source']['candidate'] == c['id'] + runtime.rollback(st, store, evolution.candidate(st, c['id']), 'verified regression') + assert 'deployed_source' not in Ledger(Path(st.improve_ledger_dir)).load(c['workflow']).baseline + + +def test_finished_candidate_does_not_block_following_week_hypothesis(autonomous): + from infermatrix_copilot.improve import coordinator + from infermatrix_copilot.improve.ledger import Ledger + st, store, model, worker = autonomous + c = evolution.run(st, store, workflow=objectives.ENGINE, llm=model, sandbox=worker) + c['state'] = 'rejected' + evolution.save(st, c, store) + result = coordinator.run(st, store, workflow=objectives.ENGINE, llm=model, sandbox=worker, now=c['created_at'] + 8 * 86400) + next_candidate = result['stages']['evolve'] + assert next_candidate.get('id') != c['id'] and next_candidate.get('proposal') != c['proposal'] + proposals = Ledger(Path(st.improve_ledger_dir)).load(c['workflow']).proposals + assert next(p for p in proposals if p.id == c['proposal']).state == 'closed' diff --git a/test/test_evolution.py b/test/test_evolution.py index 6cb8e552..4e0ff1a3 100644 --- a/test/test_evolution.py +++ b/test/test_evolution.py @@ -89,7 +89,7 @@ def build(driver="meta", workflow="workflow-improve.improve.forensics"): "sha256": artifacts.digest((data / f"{n}.json").read_bytes()), "labels": {"g": "S2"}, "label_source": "human", "gold": {"entries": [{"gold_id": "g", "path": "a.py", "concern": "concern"}]}}) artifacts.atomic_json(data / "dataset.json", {"workflow": workflow, "items": rows}) - st = Settings(_env_file=None, improve_enabled=True, improve_evolve_enabled=True, + st = Settings(_env_file=None, improve_enabled=True, improve_evolve_enabled=True, improve_evaluation_mode="gold", improve_evolve_source_dir=str(source), improve_evolve_data_dir=str(data.parent), improve_workflows_dirs=str(defs), improve_ledger_dir=str(tmp_path / driver / "ledger"), improve_budget_usd_week=20, improve_judge="api:claude-sonnet-5", eco_model="claude-sonnet-5",