diff --git a/docs/working/TASK-0896/current-state.md b/docs/working/TASK-0896/current-state.md index c91258de..49ab2e6a 100644 --- a/docs/working/TASK-0896/current-state.md +++ b/docs/working/TASK-0896/current-state.md @@ -1,10 +1,7 @@ # current-state — TASK-0896 -- **フェーズ**: C-2 確定反映済み(R-001〜R-010)→ 次 = **C-3 Human 承認待ち** -- **branch**: `docs/task-0896-plan`(origin/main 53879e1 から分岐) -- **Mode**: high-risk / lite_eligible=false / autonomous APPROVE 不可(人間 C-3 必須) -- **Metrics**: 変更ファイル実数 9(見積 9・比率 1.0・採用)。重複 5 点の実在を行番号で裏取り済み -- **設計確定**: c3_contract.py 新設(論点 1 案 A・REQUIRED_KEYS 系含む)/ 三つ組照合は strictness 引数で非対称保存(論点 2 案 A・reviewer 集合非対称も保存)/ 返り値 = 理由文字列リスト + 順序契約(論点 3) -- **C-1**: PASS(WARN 1)/ **C-2**: 2 レーン完了(A=Codex major 4→採用 3・不採用 1〔経路違い〕、B=整合 9/9 照合 OK・minor 2 info 4 全採用)→ 1 回確定反映済み・簡易 C-1 PASS -- **次アクション**: 人間 C-3(論点 1: #873 との実装順 / 論点 2: EPIC #870 追記コメント要否)→ APPROVED なら `bin/plangate approve TASK-0896`(c3.json 発行)→ exec -- **ブロッカー**: C-3 Human 判断待ち +- **フェーズ**: exec 完了(コミット a〜d + 敵対レビュー + handoff 発行)→ 次 = **exec PR 作成 → C-4** +- **branch**: `feat/task-0896-c3-contract`(99816f8・origin/main b632a91 から分岐) +- **検証**: 4 系 green(22/247/30/12)+ run-tests 412 passed 0 failed + 敵対レビュー 2 レーン critical/major 0(behavior diff 37+37 全一致) +- **残**: PR 作成 → C-4 merge → issue #896 close 判定 +- **#873 並行**: c3_contract 先行 merge 後に #873 が rebase(C-3 決定・EPIC #870 コメント済み) diff --git a/docs/working/TASK-0896/decision-log.jsonl b/docs/working/TASK-0896/decision-log.jsonl index 3a9906f3..ee16108b 100644 --- a/docs/working/TASK-0896/decision-log.jsonl +++ b/docs/working/TASK-0896/decision-log.jsonl @@ -5,3 +5,5 @@ {"ts":"2026-07-22T12:40:00+09:00","phase":"C-1","decision":"PASS(WARN 1 = T-11 粒度・exec 時サブ分割)","rationale":"17 項目全チェック・FAIL 0","actor":"agent"} {"ts":"2026-07-22T13:10:00+09:00","phase":"C-2","decision":"2 レーン完了: A=Codex 設計妥当性 major 4(採用 3/不採用 1)・B=コードベース整合 9/9 照合 OK minor 2 info 4 全採用","rationale":"R-002(C1-VERDICT マーカー)は経路違いで不採用 — c3-prime evidence マーカーは ai-loop run 経路用・本タスクは human C-3 経路(前例 TASK-0872 grep 0 件で受理)","actor":"agent"} {"ts":"2026-07-22T13:25:00+09:00","phase":"C-2 確定反映","decision":"R-001/R-003〜R-010 を plan/todo/test-cases へ 1 回確定反映・簡易 C-1 PASS","rationale":"主要変更: REQUIRED_KEYS 系の共通化対象化 / 理由リスト順序契約 + 代表文言回帰 / 純粋性 I/O 封じテスト / 残置列挙の完全化(reviewers 2 者・source_sha 照合)/ ta-55 CI 経路追加(実数 8→9)","actor":"agent"} +{"ts":"2026-07-22T18:40:00+09:00","phase":"exec","decision":"コミット a〜d 完了(91dbd81/673112d/cf925c9/99816f8)・各コミットで 4 系 green 維持","rationale":"1 コミット 1 種類(定数→hash→trio→sync)。run-tests 412 passed 0 failed","actor":"agent"} +{"ts":"2026-07-22T18:55:00+09:00","phase":"敵対レビュー","decision":"2 レーン完了・critical/major 0。AF-1 採用(記録対応)/ AF-2 不採用(R-008 確定設計)/ AF-3 採用(記録)/ AF-4 不採用(スコープ外)","rationale":"独立レーンが新旧 behavior diff 37+37 パターンで tuple/exit code 全一致を実測。Codex も強度低下・偽造受理・fail-open なしを確認","actor":"agent"} diff --git a/docs/working/TASK-0896/evidence/adversarial-review-disposition.md b/docs/working/TASK-0896/evidence/adversarial-review-disposition.md new file mode 100644 index 00000000..ab739c42 --- /dev/null +++ b/docs/working/TASK-0896/evidence/adversarial-review-disposition.md @@ -0,0 +1,33 @@ +# 敵対レビュー disposition — TASK-0896(AC-8) + +> 実施: 2026-07-22 / 2 レーン並列(Codex + 独立 behavior-diff 実測)/ 対象 = feat/task-0896-c3-contract 4 コミット +> 結果: **critical / major 0**。両レーンとも「受理/拒否の強度低下・偽造受理・import fail-open なし」を実測で確認 + +## レーン実測サマリ + +- **独立レーン(behavior diff)**: 新旧実装を同一敵対入力で駆動 — plan_package_check 37/37 tuple 全一致 / c3prime_verify 37/37 exit code 全一致 / producer 出力 byte 同一(sha256 一致)/ bundled 自立 4 系全 OK / cmp 10 ファイル byte 一致 / run-tests 412 passed +- **Codex レーン**: strict/lenient 両経路の偽造(余剰 reviewer・非 dict snapshot・未知キー・空値・None==None)全 reject / import 欠落は未捕捉 ImportError で停止(fail-open でない)/ SnapshotTrioTests 等 13+11 件実行成功 + +## 指摘と裁定 + +| AF | レーン | severity | 指摘 | 裁定 | +|----|--------|----------|------|------| +| AF-1 | 両レーン一致 | minor | 複合異常時の先頭 stderr 理由が変化(旧: per-reviewer で verdict 異常が先 / 新: 全 reviewer trio 検査後に verdict — 例: model_a.verdict 不正 + model_b.source_sha 改竄で新実装は trio 不一致を先に報告)。exit code / 判定結果は全一致 | **採用(記録対応)**: 機械消費者 0 件を rg 全数確認済み(bin/plangate は stderr 転送のみ・test は exit code のみ assert)。R-004 順序契約(キー集合→空値→trio)の設計帰結であり実装は変えない。PR body に「診断優先順の変化(判定結果不変)」を明記 + handoff 既知課題 KI-1。理由コード enum 化(V2)時に受理器レベルでも順序固定 | +| AF-2 | 独立 | minor | arbiter の `import c3_contract` は sys.path 挿入なし — `spec_from_file_location` 等の埋め込み import で ModuleNotFoundError(現実の importer は全て path 挿入済みで無害を実測) | **不採用**: R-008(C-2 確定)で「arbiter に sys.path 操作を追加しない」を設計確定済み。handoff 既知課題 KI-2 に経路制約として記録 | +| AF-3 | 独立 | info | c3_contract 部分配布時、c3prime_verify は traceback exit 1(fail-closed)だが legacy record も exit 10 でなく 1 になる | **採用(記録対応)**: handoff 運用注記。sync 対列挙 + ta-30 で構造的に防止済み | +| AF-4 | 独立 | info | `derived_loopspec_hash` が任意値でも受理 — 新旧同一の既存挙動(本 PR の差分でない) | **不採用(スコープ外)**: handoff V2 候補に記録(契約側の値検証は #874 系で検討) | + +**AC-7 確認**: 偽造 record 群 reject 不変(37 パターン exit code 一致 + test_c3prime_verify 12 不変) + +--- + +## River Review(ローカル / PR #902 作成後・2026-07-22) + +独立再実行(4 スイート全 OK・byte 一致・被参照 grep)付き finding verification 適用済み。**critical/major 0・総合 = マージ可(APPROVE 推奨)**。 + +| RF | severity | 指摘 | 裁定 | +|----|----------|------|------| +| F-1 | minor | strict 経路で「キーは存在するが空値」のとき診断が「キー欠落」と報告されオペレータが誤読しうる(判定・exit code は不変・旧文言の被参照 0 件実測) | **採用(即修正)**: 文言を「キー欠落または空値」へ 1 語修正 + 順序契約テストの assert 更新(コミット e)。契約改版として明示コミット(R-004 Fallback 準拠) | +| F-2 | info | 複合異常時の先頭診断優先順変化 | 既 disposition(AF-1 / KI-1)— 確認のみ | +| F-3 | info | arbiter の sys.path 非挿入(python -P で fail-closed crash を実測) | 既 disposition(AF-2 / KI-2)— 確認のみ | +| F-4 | info | arbiter の共通層 import が中腹配置(sibling は冒頭) | **不採用**: 契約定数の使用箇所直近に置く意図コメントあり・AST テストは位置非依存。V2 の整理時に検討 | diff --git a/docs/working/TASK-0896/evidence/test-runs/final-verification.log b/docs/working/TASK-0896/evidence/test-runs/final-verification.log new file mode 100644 index 00000000..5afc53f3 --- /dev/null +++ b/docs/working/TASK-0896/evidence/test-runs/final-verification.log @@ -0,0 +1,14 @@ +=== final 2026-07-22 17:56 feat/task-0896-c3-contract 99816f8 === +Ran 22 tests in 0.046s + +OK +Ran 247 tests in 0.097s + +OK +Ran 30 tests in 0.178s + +OK +Ran 12 tests in 3.013s + +OK +run-tests: 412 passed, 0 failed(b1x8cmubu 実測・ta-30 scripts=10 / ta-55 新 TC PASS) diff --git a/docs/working/TASK-0896/evidence/test-runs/step0-baseline.log b/docs/working/TASK-0896/evidence/test-runs/step0-baseline.log new file mode 100644 index 00000000..d76dd5cc --- /dev/null +++ b/docs/working/TASK-0896/evidence/test-runs/step0-baseline.log @@ -0,0 +1,15 @@ +=== baseline 2026-07-22 17:42 main b632a91 (repo root 実行) === +Ran 247 tests in 0.087s + +OK +Ran 30 tests in 0.150s + +OK +Ran 12 tests in 2.770s + +OK +--- 注: scripts/ai-loop cwd での test_arbiter 実行は ho-paths 相対解決で FAIL する(repo root 実行が正・CI 同様) --- + [PASS] 手偽造 record (c3_status+未知+欠落) → exit 1 + [PASS] HO 適用後: bin/plangate validate が c3-prime を受理 + +Results: 411 passed, 0 failed diff --git a/docs/working/TASK-0896/handoff.md b/docs/working/TASK-0896/handoff.md new file mode 100644 index 00000000..24fa0526 --- /dev/null +++ b/docs/working/TASK-0896/handoff.md @@ -0,0 +1,46 @@ +# Handoff — TASK-0896(検証ロジック共通契約層化) + +> Issue: [#896](https://github.com/s977043/plangate/issues/896) / Mode: high-risk / branch: `feat/task-0896-c3-contract` +> 発行: 2026-07-22(WF-05) + +## 1. 要件適合確認結果(AC ごと) + +| AC | 判定 | 根拠 | +|----|------|------| +| AC-1 契約定数の単一モジュール定義(REQUIRED_KEYS 系含む) | **PASS** | c3_contract.py に 7 定数群 + TRIO_KEYS。3 消費者は import 参照(is 同一を ConsumerAliasTests で固定・タプルリテラル重複 grep 0 件) | +| AC-2 sha256 / canonical hash 単一実装 | **PASS** | sha256_of_file / canonical_hash に統合。ローカル hashlib 実装 0 件(TC-4 grep)。producer 出力は新旧 byte 同一(敵対レビュー実測) | +| AC-3 三つ組照合の I/O なし共通純関数・両経路同一実装 | **PASS** | check_snapshot_trio(理由リスト・順序契約・I/O 封じテスト)。arbiter=strict_keys=False / c3prime=True の両経路が同一実装 | +| AC-4 既存テスト全 green(振る舞い不変) | **PASS** | test_arbiter 247 / test_plan_package 30 / test_c3prime_verify 12 / run-tests 412(既存 411 + ta-55 追記 1・既存期待値変更ゼロ)。evidence/test-runs/ | +| AC-5 sync 列挙 + 2 回目 no-op + ta-30 自立 PASS | **PASS** | copy + delete 保護の対列挙。sync 2 回目「no changes」実測・ta-30 scripts=10 / TC-08 bundled 自立 PASS・cmp 10 ファイル byte 一致 | +| AC-6 arbiter が I/O あり関数を import / call しない | **PASS** | AST 回帰検査(test_arbiter_does_not_touch_io_layer)で属性・名前参照 0 を固定 | +| AC-7 偽造 record 群の reject 不変 | **PASS** | 敵対レビューで新旧 37 パターン exit code 全一致 + test_c3prime_verify 12 不変 | +| AC-8 敵対レビュー 1 ラウンド以上の disposition 記録 | **PASS** | 2 レーン(Codex + 独立 behavior-diff)実施。evidence/adversarial-review-disposition.md | + +## 2. 既知課題一覧 + +- **KI-1(AF-1・minor)**: 複合異常入力での先頭 stderr 診断の優先順が変化(verdict 異常より trio 不一致が先)。判定結果・exit code は全一致・機械消費者 0 件を rg 全数確認済み。R-004 順序契約の設計帰結 +- **KI-2(AF-2・minor)**: arbiter.py は sys.path 挿入なしで `import c3_contract` する(R-008 の確定設計)。CLI 直実行・test 経由・bundled は全て解決済みだが、`spec_from_file_location` 等の埋め込み import は非サポート経路 +- **KI-3(AF-3・info)**: c3_contract.py の部分配布(sync 漏れ)時は c3prime_verify が traceback exit 1(fail-closed)— legacy record も exit 10 でなく 1 になる。sync 対列挙 + ta-30 で構造的に防止済み + +## 3. V2 候補 + +- 理由コード enum + 呼び出し側文言マッピング(KI-1 の受理器レベル順序固定を含む) +- `derived_loopspec_hash` の値検証(AF-4・既存挙動。#874 RunEvidence 契約側で検討) +- arbiter `plan_package_check` の構造検査(PLAN_PACKAGE_REQUIRED_KEYS)と record 検査の関係整理は issue Non-goals どおり非統合を維持 + +## 4. 妥協点(採用しなかった選択肢と理由) + +- 三つ組照合の strict 統一(案 B): arbiter 側挙動変更になるため不採用 — strict_keys 引数で #889 R2 非対称を保存 +- plan_package.py への集約(案 B): arbiter が producer を import する形になり I/O 層分離が module 境界で保証されないため不採用 +- AF-1 対応のテスト追加: plan Files to Touch 外(test_c3prime_verify.py)への変更となるため見送り、記録対応に留めた + +## 5. 引き継ぎ文書 + +c3-prime 契約の検証規則(定数 / hash / snapshot 三つ組照合)が `scripts/ai-loop/c3_contract.py` に単一定義された。**#873 delivery.py / #874 run_evidence.py は本モジュールを import して実装すること**(sys.path.insert の既存パターン: c3prime_verify.py L25 参照)。新スクリプトを追加したら sync-plugin-plangate.sh の copy 列挙 + delete 保護 case の**両方**へ対で追記(漏れると bundled 側 import エラー = ta-30 が検出)。共通層の改版は test_c3_contract.py の契約固定テスト(値 byte 同一・順序契約・純粋性・AC-6 AST 検査)が drift を検出する。4 コミット構成(a=定数 / b=hash / c=trio / d=sync)で各コミット単独 revert 可能。 + +## 6. テスト結果サマリ + +- ベースライン(main b632a91): 247 / 30 / 12 / 411 全 green(evidence/test-runs/step0-baseline.log。※scripts/ai-loop cwd での test_arbiter 実行は ho-paths 相対解決で FAIL する既知事象 — repo root 実行が正) +- 最終形(99816f8): test_c3_contract 22 / test_arbiter 247 / test_plan_package 30 / test_c3prime_verify 12 / **run-tests 412 passed 0 failed**(evidence/test-runs/final-verification.log) +- 敵対レビュー: 新旧 behavior diff 37+37 パターン全一致・bundled 自立 4 系 OK(evidence/adversarial-review-disposition.md) +- settings タスクロック: `doctor --check-settings` PASS(handoff 前提充足) diff --git a/docs/working/TASK-0896/run.ndjson b/docs/working/TASK-0896/run.ndjson new file mode 100644 index 00000000..e183d3d1 --- /dev/null +++ b/docs/working/TASK-0896/run.ndjson @@ -0,0 +1 @@ +{"ts":"2026-07-22T08:41:42Z","task_id":"TASK-0896","phase":"D","event":"session_started","detail":"agent=codex"} diff --git a/docs/working/TASK-0896/status.md b/docs/working/TASK-0896/status.md index 161a2dbf..d0684b2c 100644 --- a/docs/working/TASK-0896/status.md +++ b/docs/working/TASK-0896/status.md @@ -19,11 +19,23 @@ - 論点 2(EPIC #870 追記): **投稿する**(AI が下書き投稿・論点 1 の決定内容を反映) - c3.json 発行: Human が `bin/plangate approve TASK-0896` を plan branch 上で実行(発行は確定反映 f6bf1a9 の後 = EH-3 整合順序) +## exec 記録(2026-07-22) + +| 日時 | 内容 | +|------|------| +| 2026-07-22 14:30 | c3.json 発行(Human 対話実行)→ PR #899(plan 正式化)CI 全 green → C-4 merge b75f277 | +| 2026-07-22 17:40 | exec 開始(branch `feat/task-0896-c3-contract`)・ベースライン 247/30/12/411 全 green | +| 2026-07-22 18:10 | コミット a 91dbd81(定数集約)/ b 673112d(hash 統合)/ c cf925c9(trio 共通化)/ d 99816f8(sync + ta-55) | +| 2026-07-22 18:55 | 敵対レビュー 2 レーン完了(critical/major 0・AF-1〜4 disposition)・handoff 発行 | + +### 計画からの変更点 + +なし(Files to Touch 9 ファイル内で完結・Replan Trigger 発火なし)。AF-1(診断優先順の変化)は R-004 順序契約の設計帰結として記録対応(KI-1)。 + ## 残タスク -- [ ] c3.json 発行(Human・`bin/plangate approve TASK-0896`) -- [ ] c3.json commit + push → PR 作成(plan 正式化 PR・1 PR 完結方式) -- [ ] C-4 マージ後: exec 開始(todo.md T-1〜T-17・本セッション or `PLANGATE_HOOK_TASK=TASK-0896` 再起動セッション) +- [ ] exec PR 作成 → C-4(Human レビュー・マージ) +- [ ] マージ後: issue #896 に DoD evidence 記録 → close 判定 ## 参照 diff --git a/plugin/plangate/skills/ai-loop-cycle/scripts/arbiter.py b/plugin/plangate/skills/ai-loop-cycle/scripts/arbiter.py index d09c4742..4c31bccf 100644 --- a/plugin/plangate/skills/ai-loop-cycle/scripts/arbiter.py +++ b/plugin/plangate/skills/ai-loop-cycle/scripts/arbiter.py @@ -475,21 +475,12 @@ def plan_quality_check(gates: Any) -> bool: # snapshot 欠落)→ priority 1.65 で BLOCKED(契約 §3/§4 の fail-closed) # escalate / BLOCKED 条件を追加するだけの安全側変更であり、以前 escalate だった # 経路を auto-approve に倒す効果は持たない。 -PLAN_PACKAGE_REQUIRED_KEYS = ( - "plan_hash", - "source_sha", - "plan_package_hash", - "c1_evidence_ref", - "c2_evidence_ref", - "reviewers", -) -SNAPSHOT_REQUIRED_KEYS = ( - "verdict", - "plan_hash", - "source_sha", - "plan_package_hash", - "evidence_ref", -) +# 契約定数は c3_contract が単一定義(TASK-0896 / #896)。arbiter は I/O なし層 +# (定数・純関数)のみを import する(AC-6: sha256_of_file は import / call しない)。 +import c3_contract # noqa: E402 + +PLAN_PACKAGE_REQUIRED_KEYS = c3_contract.PLAN_PACKAGE_REQUIRED_KEYS +SNAPSHOT_REQUIRED_KEYS = c3_contract.SNAPSHOT_KEYS def plan_package_check(plan_package: Any, target_sha: str) -> tuple[bool, bool, str]: @@ -507,19 +498,12 @@ def plan_package_check(plan_package: Any, target_sha: str) -> tuple[bool, bool, reviewers = plan_package.get("reviewers") if not isinstance(reviewers, dict): return True, False, "plan_package.reviewers が object でない" - for model in ("model_a", "model_b"): - snap = reviewers.get(model) - if not isinstance(snap, dict): - return True, False, f"reviewers.{model} の snapshot が欠落(契約 §3: BLOCKED)" - snap_missing = [k for k in SNAPSHOT_REQUIRED_KEYS if not snap.get(k)] - if snap_missing: - return True, False, f"reviewers.{model} の snapshot キー欠落: {', '.join(snap_missing)}" - for key in ("plan_hash", "source_sha", "plan_package_hash"): - if snap.get(key) != plan_package.get(key): - return True, False, ( - f"reviewers.{model}.{key} がトップレベル値と不一致" - "(同一 Plan Package を観ていない = AC-5 違反)" - ) + # snapshot 5 キー整合 + 三つ組一致は共通純関数(strict_keys=False = 余剰キー + # 許容の既存挙動を保存)。理由リスト非空 → integrity NG(先頭を reason に)。 + trio_reasons = c3_contract.check_snapshot_trio( + plan_package, reviewers, strict_keys=False) + if trio_reasons: + return True, False, trio_reasons[0] if plan_package.get("source_sha") != target_sha: return True, False, ( f"plan_package.source_sha ({plan_package.get('source_sha')}) と " diff --git a/plugin/plangate/skills/ai-loop-cycle/scripts/c3_contract.py b/plugin/plangate/skills/ai-loop-cycle/scripts/c3_contract.py new file mode 100644 index 00000000..c9c3cf95 --- /dev/null +++ b/plugin/plangate/skills/ai-loop-cycle/scripts/c3_contract.py @@ -0,0 +1,121 @@ +"""c3-prime 契約の共通契約層(TASK-0896 / #896)。 + +契約正本: docs/workflows/ai-loop/c3-prime-contract.md(本モジュールは実装集約のみ・ +規則は不変)。arbiter.py(入力ブロック検証)/ plan_package.py(producer)/ +c3prime_verify.py(record 受理器)の 3 消費者が import 参照する単一定義。 + +層区分(AC-6): +- 契約定数: I/O なし。全消費者が参照可 +- I/O なし純関数: canonical_hash / check_snapshot_trio。arbiter が import してよいのは + この層まで(arbiter は「入力 dict のみで決定論裁定」の設計を維持し、共通層から + 新たなファイル読取依存を持ち込まない) +- I/O あり関数: sha256_of_file。producer / 受理器のみが使用し、arbiter は + import / call しない(test_c3_contract.py が回帰検査) +""" +from __future__ import annotations + +import hashlib +import json +import pathlib + +# --------------------------------------------------------------------------- +# 契約定数(契約 §1/§2) +# --------------------------------------------------------------------------- + +# 契約 §1: Plan Package 6 要素(key 順は artifact_hashes の表示順にも使う) +ARTIFACTS = ( + "pbi-input.md", + "plan.md", + "todo.md", + "test-cases.md", + "review-self.md", + "review-external.md", +) + +# 契約 §2: decision の 3 値 allowlist(#887 レビュー / 両 Codex major 指摘反映)。 +VALID_DECISIONS = ("AUTO_APPROVED", "HUMAN_ESCALATED", "BLOCKED") +VALID_VERDICTS = ("approve", "reject") + +# 契約 §3: reviewer snapshot の 5 キー(三つ組照合の対象は plan_hash / +# source_sha / plan_package_hash)。 +SNAPSHOT_KEYS = ("verdict", "plan_hash", "source_sha", "plan_package_hash", "evidence_ref") + +# 契約 §2: c3-prime record トップレベルの必須キー(受理器 allowlist の中核)。 +RECORD_REQUIRED_KEYS = ( + "task_id", "approval_kind", "phase", "decision", "source_sha", "plan_hash", + "plan_package_hash", "artifact_hashes", "c1_evidence_ref", "c2_evidence_ref", + "reviewers", "policy_ref", "issued_at", "issued_by", +) +# 任意で許容する追加キー(それ以外の未知キーは reject。`c3_status` は §5 で明示禁止)。 +RECORD_OPTIONAL_KEYS = ("derived_loopspec_hash",) +RECORD_ALLOWED_KEYS = set(RECORD_REQUIRED_KEYS) | set(RECORD_OPTIONAL_KEYS) + +# 契約 §2/§3: arbiter 入力 `plan_package` ブロックの必須キー。 +PLAN_PACKAGE_REQUIRED_KEYS = ( + "plan_hash", + "source_sha", + "plan_package_hash", + "c1_evidence_ref", + "c2_evidence_ref", + "reviewers", +) + +# 三つ組照合の対象キー(snapshot 値 = container トップレベル値の一致要求)。 +TRIO_KEYS = ("plan_hash", "source_sha", "plan_package_hash") + + +# --------------------------------------------------------------------------- +# I/O なし純関数 +# --------------------------------------------------------------------------- + +def canonical_hash(obj) -> str: + """正規化 JSON(sort_keys・区切り最小)の sha256 を返す(契約 §2)。""" + canon = json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8") + return "sha256:" + hashlib.sha256(canon).hexdigest() + + +def check_snapshot_trio(container, reviewers, strict_keys) -> list[str]: + """reviewer snapshot の 5 キー整合 + 三つ組一致を検査し理由リストを返す。 + + 空リスト = OK。判定・終端制御(arbiter の tuple 部分成功 / c3prime_verify の + 即時 reject)は呼び出し側の責務。理由リストの生成順序は検査順で契約固定: + reviewer 順(model_a → model_b)× reviewer 内は + (1) snapshot 型 / キー集合 → (2) 空値 → (3) 三つ組不一致(TASK-0896 R-004)。 + + strict_keys の非対称は #889 R2 由来の意図的設計を保存する(TASK-0896 R-005): + - strict_keys=True(c3prime_verify): snapshot はちょうど 5 キー(余剰 reject) + - strict_keys=False(arbiter): 欠落・空値のみ検査(余剰キー許容) + + 本関数が検査しないもの(呼び出し側残置): verdict 語彙 / evidence_ref 独立性 / + AUTO_APPROVED 整合 / reviewers ちょうど 2 者(strict 側)/ 余剰 reviewer 許容 + (lenient 側)/ PLAN_PACKAGE_REQUIRED_KEYS 構造検査 / source_sha vs target_sha。 + """ + reasons: list[str] = [] + for model in ("model_a", "model_b"): + snap = reviewers.get(model) if isinstance(reviewers, dict) else None + if not isinstance(snap, dict): + reasons.append(f"reviewers.{model} の snapshot が欠落(契約 §3: fail-closed)") + continue + if strict_keys and set(snap) != set(SNAPSHOT_KEYS): + reasons.append(f"reviewers.{model} の snapshot キーが規定 5 キーと不一致") + continue + missing = [k for k in SNAPSHOT_KEYS if not snap.get(k)] + if missing: + reasons.append( + f"reviewers.{model} の snapshot キー欠落または空値: {', '.join(missing)}") + continue + for key in TRIO_KEYS: + if snap.get(key) != (container.get(key) if isinstance(container, dict) else None): + reasons.append( + f"reviewers.{model}.{key} がトップレベル値と不一致" + "(同一 Plan Package を観ていない = AC-5 違反)") + return reasons + + +# --------------------------------------------------------------------------- +# I/O あり関数(arbiter は import / call しない — AC-6) +# --------------------------------------------------------------------------- + +def sha256_of_file(path) -> str: + """ファイル内容の sha256 を `sha256:<64hex>` 形式で返す(契約 §2)。""" + return "sha256:" + hashlib.sha256(pathlib.Path(path).read_bytes()).hexdigest() diff --git a/plugin/plangate/skills/ai-loop-cycle/scripts/c3prime_verify.py b/plugin/plangate/skills/ai-loop-cycle/scripts/c3prime_verify.py index e14a0894..bba38909 100755 --- a/plugin/plangate/skills/ai-loop-cycle/scripts/c3prime_verify.py +++ b/plugin/plangate/skills/ai-loop-cycle/scripts/c3prime_verify.py @@ -16,35 +16,26 @@ """ from __future__ import annotations -import hashlib import json import pathlib import re import sys sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) +import c3_contract # noqa: E402 契約定数の単一定義(TASK-0896 / #896) import plan_package # noqa: E402 受理側でも evidence marker を再検証するため共有 -ARTIFACTS = ( - "pbi-input.md", "plan.md", "todo.md", "test-cases.md", - "review-self.md", "review-external.md", -) -VALID_DECISIONS = ("AUTO_APPROVED", "HUMAN_ESCALATED", "BLOCKED") -VALID_VERDICTS = ("approve", "reject") -SNAPSHOT_KEYS = ("verdict", "plan_hash", "source_sha", "plan_package_hash", "evidence_ref") -# 契約 §2: c3-prime トップレベルの必須キー(allowlist の中核)。 -REQUIRED_KEYS = ( - "task_id", "approval_kind", "phase", "decision", "source_sha", "plan_hash", - "plan_package_hash", "artifact_hashes", "c1_evidence_ref", "c2_evidence_ref", - "reviewers", "policy_ref", "issued_at", "issued_by", -) -# 任意で許容する追加キー(それ以外の未知キーは reject。`c3_status` は §5 で明示禁止)。 -OPTIONAL_KEYS = ("derived_loopspec_hash",) -ALLOWED_KEYS = set(REQUIRED_KEYS) | set(OPTIONAL_KEYS) +# 契約定数は c3_contract が単一定義(値の契約固定は test_c3_contract.py)。 +ARTIFACTS = c3_contract.ARTIFACTS +VALID_DECISIONS = c3_contract.VALID_DECISIONS +VALID_VERDICTS = c3_contract.VALID_VERDICTS +SNAPSHOT_KEYS = c3_contract.SNAPSHOT_KEYS +REQUIRED_KEYS = c3_contract.RECORD_REQUIRED_KEYS +OPTIONAL_KEYS = c3_contract.RECORD_OPTIONAL_KEYS +ALLOWED_KEYS = c3_contract.RECORD_ALLOWED_KEYS -def _sha256(path: pathlib.Path) -> str: - return "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest() +_sha256 = c3_contract.sha256_of_file # 単一実装(TASK-0896 AC-2) def _fail(msg: str) -> int: @@ -148,8 +139,7 @@ def main(argv): return _fail(f"artifact_hashes 不一致: {name}(stale)") # plan_package_hash = artifact_hashes の正規化 JSON の sha256 - canon = json.dumps(ah, sort_keys=True, separators=(",", ":")).encode("utf-8") - if data.get("plan_package_hash") != "sha256:" + hashlib.sha256(canon).hexdigest(): + if data.get("plan_package_hash") != c3_contract.canonical_hash(ah): return _fail("plan_package_hash が artifact_hashes から再計算した値と不一致") # reviewer snapshot 三つ組一致 + decision-verdict 整合 @@ -157,18 +147,14 @@ def main(argv): # model_a / model_b ちょうど 2 者(余剰 reviewer キーは reject / #889 R2 medium)。 if not isinstance(reviewers, dict) or set(reviewers) != {"model_a", "model_b"}: return _fail(f"reviewers は model_a / model_b のちょうど 2 者: {sorted(reviewers) if isinstance(reviewers, dict) else reviewers!r}") + # snapshot 5 キー整合(strict_keys=True = ちょうど 5 キー・未知ネストキーは + # reject / #889 R2 medium)+ 三つ組一致は共通純関数。理由リスト非空 → reject。 + trio_reasons = c3_contract.check_snapshot_trio(data, reviewers, strict_keys=True) + if trio_reasons: + return _fail(trio_reasons[0]) for m in ("model_a", "model_b"): - snap = reviewers.get(m) - # snapshot は 5 キーちょうど(未知ネストキーは reject / #889 R2 medium)。 - if not isinstance(snap, dict) or set(snap) != set(SNAPSHOT_KEYS): - return _fail(f"reviewers.{m} の snapshot キーが規定 5 キーと不一致") - if any(not snap.get(k) for k in SNAPSHOT_KEYS): - return _fail(f"reviewers.{m} の snapshot に空値") - if snap.get("verdict") not in VALID_VERDICTS: + if reviewers[m].get("verdict") not in VALID_VERDICTS: return _fail(f"reviewers.{m}.verdict が approve/reject 以外") - for key in ("plan_hash", "source_sha", "plan_package_hash"): - if snap.get(key) != data.get(key): - return _fail(f"reviewers.{m}.{key} がトップレベル値と不一致(AC-5 違反)") # reviewer 独立性: 両者の evidence_ref が同一なら独立 2 者レビュー偽装 # (#889 R2 high。snapshot hash は同一が正だが evidence は別根拠であるべき)。 if reviewers["model_a"]["evidence_ref"] == reviewers["model_b"]["evidence_ref"]: diff --git a/plugin/plangate/skills/ai-loop-cycle/scripts/plan_package.py b/plugin/plangate/skills/ai-loop-cycle/scripts/plan_package.py index b3555d1a..28571db3 100644 --- a/plugin/plangate/skills/ai-loop-cycle/scripts/plan_package.py +++ b/plugin/plangate/skills/ai-loop-cycle/scripts/plan_package.py @@ -15,7 +15,6 @@ from __future__ import annotations import argparse -import hashlib import json import pathlib import re @@ -23,15 +22,11 @@ TASK_ID_RE = re.compile(r"^TASK-[0-9]{4}$") -# 契約 §1: Plan Package 6 要素(key 順は artifact_hashes の表示順にも使う) -ARTIFACTS = ( - "pbi-input.md", - "plan.md", - "todo.md", - "test-cases.md", - "review-self.md", - "review-external.md", -) +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) +import c3_contract # noqa: E402 契約定数の単一定義(TASK-0896 / #896) + +# 契約 §1: Plan Package 6 要素(単一定義は c3_contract) +ARTIFACTS = c3_contract.ARTIFACTS C1_EVIDENCE = "review-self.md" C2_EVIDENCE = "review-external.md" @@ -78,9 +73,9 @@ def check_presence(task_dir): _C1_PREFIX_RE = re.compile(r"^C1-VERDICT:", re.MULTILINE) _C2_PREFIX_RE = re.compile(r"^C2-VERDICT:", re.MULTILINE) -# 契約 §2: decision の 3 値 allowlist(#887 レビュー / 両 Codex major 指摘反映)。 -VALID_DECISIONS = ("AUTO_APPROVED", "HUMAN_ESCALATED", "BLOCKED") -VALID_VERDICTS = ("approve", "reject") +# 契約 §2: decision の 3 値 allowlist(単一定義は c3_contract)。 +VALID_DECISIONS = c3_contract.VALID_DECISIONS +VALID_VERDICTS = c3_contract.VALID_VERDICTS def _read_evidence_marker(path, marker_re, prefix_re): @@ -134,8 +129,7 @@ def check_evidence(task_dir): return errors -def _sha256_of(path): - return "sha256:" + hashlib.sha256(pathlib.Path(path).read_bytes()).hexdigest() +_sha256_of = c3_contract.sha256_of_file # 単一実装(TASK-0896 AC-2) def compute_hashes(task_dir): @@ -145,12 +139,10 @@ def compute_hashes(task_dir): if presence: raise PlanPackageError(presence) artifact_hashes = {name: _sha256_of(task_dir / name) for name in ARTIFACTS} - canon = json.dumps(artifact_hashes, sort_keys=True, - separators=(",", ":")).encode("utf-8") return { "artifact_hashes": artifact_hashes, "plan_hash": artifact_hashes["plan.md"], - "plan_package_hash": "sha256:" + hashlib.sha256(canon).hexdigest(), + "plan_package_hash": c3_contract.canonical_hash(artifact_hashes), } diff --git a/plugin/plangate/skills/ai-loop-cycle/scripts/test_c3_contract.py b/plugin/plangate/skills/ai-loop-cycle/scripts/test_c3_contract.py new file mode 100644 index 00000000..18ae2b41 --- /dev/null +++ b/plugin/plangate/skills/ai-loop-cycle/scripts/test_c3_contract.py @@ -0,0 +1,215 @@ +"""c3_contract.py の契約固定テスト(TASK-0896)。 + +対象は共通層の純関数境界値と定数契約のみ。偽造 record 14 パターンの受理器 +統合テストは test_c3prime_verify.py に残置する(pbi-input Unknowns 確定)。 +""" +import pathlib +import sys +import unittest + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) +import c3_contract # noqa: E402 + + +class ConstantContractTests(unittest.TestCase): + """契約定数の値固定(コミット a: 移行前実装との byte 同一を将来にわたり固定)。""" + + def test_artifacts(self): + self.assertEqual(c3_contract.ARTIFACTS, ( + "pbi-input.md", "plan.md", "todo.md", "test-cases.md", + "review-self.md", "review-external.md", + )) + + def test_decisions_verdicts(self): + self.assertEqual(c3_contract.VALID_DECISIONS, + ("AUTO_APPROVED", "HUMAN_ESCALATED", "BLOCKED")) + self.assertEqual(c3_contract.VALID_VERDICTS, ("approve", "reject")) + + def test_snapshot_keys(self): + self.assertEqual(c3_contract.SNAPSHOT_KEYS, ( + "verdict", "plan_hash", "source_sha", "plan_package_hash", "evidence_ref")) + + def test_record_keys(self): + self.assertEqual(c3_contract.RECORD_REQUIRED_KEYS, ( + "task_id", "approval_kind", "phase", "decision", "source_sha", "plan_hash", + "plan_package_hash", "artifact_hashes", "c1_evidence_ref", "c2_evidence_ref", + "reviewers", "policy_ref", "issued_at", "issued_by", + )) + self.assertEqual(c3_contract.RECORD_OPTIONAL_KEYS, ("derived_loopspec_hash",)) + self.assertEqual( + c3_contract.RECORD_ALLOWED_KEYS, + set(c3_contract.RECORD_REQUIRED_KEYS) | set(c3_contract.RECORD_OPTIONAL_KEYS)) + + def test_plan_package_keys(self): + self.assertEqual(c3_contract.PLAN_PACKAGE_REQUIRED_KEYS, ( + "plan_hash", "source_sha", "plan_package_hash", + "c1_evidence_ref", "c2_evidence_ref", "reviewers", + )) + + def test_trio_keys(self): + self.assertEqual(c3_contract.TRIO_KEYS, + ("plan_hash", "source_sha", "plan_package_hash")) + + +class ConsumerAliasTests(unittest.TestCase): + """3 消費者が c3_contract の定数を参照している(is 同一 = 別定義でない)。""" + + def test_plan_package_aliases(self): + import plan_package + self.assertIs(plan_package.ARTIFACTS, c3_contract.ARTIFACTS) + self.assertIs(plan_package.VALID_DECISIONS, c3_contract.VALID_DECISIONS) + self.assertIs(plan_package.VALID_VERDICTS, c3_contract.VALID_VERDICTS) + + def test_c3prime_verify_aliases(self): + import c3prime_verify + self.assertIs(c3prime_verify.ARTIFACTS, c3_contract.ARTIFACTS) + self.assertIs(c3prime_verify.VALID_DECISIONS, c3_contract.VALID_DECISIONS) + self.assertIs(c3prime_verify.VALID_VERDICTS, c3_contract.VALID_VERDICTS) + self.assertIs(c3prime_verify.SNAPSHOT_KEYS, c3_contract.SNAPSHOT_KEYS) + self.assertIs(c3prime_verify.REQUIRED_KEYS, c3_contract.RECORD_REQUIRED_KEYS) + self.assertIs(c3prime_verify.OPTIONAL_KEYS, c3_contract.RECORD_OPTIONAL_KEYS) + self.assertEqual(c3prime_verify.ALLOWED_KEYS, c3_contract.RECORD_ALLOWED_KEYS) + + def test_arbiter_aliases(self): + import arbiter + self.assertIs(arbiter.PLAN_PACKAGE_REQUIRED_KEYS, + c3_contract.PLAN_PACKAGE_REQUIRED_KEYS) + self.assertIs(arbiter.SNAPSHOT_REQUIRED_KEYS, c3_contract.SNAPSHOT_KEYS) + + + +class HashTests(unittest.TestCase): + """hash ヘルパーの境界値(コミット b)。""" + + def test_canonical_hash_deterministic_and_order_independent(self): + import hashlib as _h + expected = "sha256:" + _h.sha256(b'{"a":1,"b":2}').hexdigest() + self.assertEqual(c3_contract.canonical_hash({"a": 1, "b": 2}), expected) + self.assertEqual(c3_contract.canonical_hash({"b": 2, "a": 1}), expected) + + def test_canonical_hash_empty(self): + import hashlib as _h + self.assertEqual(c3_contract.canonical_hash({}), + "sha256:" + _h.sha256(b"{}").hexdigest()) + + def test_sha256_of_file_detects_one_byte_change(self): + import tempfile + with tempfile.TemporaryDirectory() as d: + p1 = pathlib.Path(d) / "a.md" + p2 = pathlib.Path(d) / "b.md" + p1.write_bytes(b"plan body\n") + p2.write_bytes(b"plan bodY\n") + h1 = c3_contract.sha256_of_file(p1) + self.assertTrue(h1.startswith("sha256:") and len(h1) == 71) + self.assertEqual(h1, c3_contract.sha256_of_file(p1)) + self.assertNotEqual(h1, c3_contract.sha256_of_file(p2)) + +def _snap(**over): + base = {"verdict": "approve", "plan_hash": "sha256:aa", "source_sha": "bbb", + "plan_package_hash": "sha256:cc", "evidence_ref": "docs/x.md"} + base.update(over) + return base + + +def _container(): + return {"plan_hash": "sha256:aa", "source_sha": "bbb", "plan_package_hash": "sha256:cc"} + + +class SnapshotTrioTests(unittest.TestCase): + """check_snapshot_trio の理由リスト契約(コミット c)。""" + + def test_ok_both_modes(self): + rv = {"model_a": _snap(evidence_ref="docs/a.md"), "model_b": _snap(evidence_ref="docs/b.md")} + self.assertEqual(c3_contract.check_snapshot_trio(_container(), rv, strict_keys=True), []) + self.assertEqual(c3_contract.check_snapshot_trio(_container(), rv, strict_keys=False), []) + + def test_missing_snapshot_and_non_dict(self): + for rv in ({}, None, {"model_a": _snap()}, {"model_a": "str", "model_b": _snap()}): + for strict in (True, False): + reasons = c3_contract.check_snapshot_trio(_container(), rv, strict_keys=strict) + self.assertTrue(reasons, f"rv={rv!r} strict={strict}") + + def test_empty_value(self): + rv = {"model_a": _snap(plan_hash=""), "model_b": _snap()} + for strict in (True, False): + reasons = c3_contract.check_snapshot_trio(_container(), rv, strict_keys=strict) + self.assertEqual(len(reasons), 1) + self.assertIn("model_a", reasons[0]) + + def test_trio_mismatch_each_key(self): + for key in c3_contract.TRIO_KEYS: + rv = {"model_a": _snap(**{key: "sha256:tampered"}), "model_b": _snap()} + for strict in (True, False): + reasons = c3_contract.check_snapshot_trio(_container(), rv, strict_keys=strict) + self.assertTrue(any(f"model_a.{key}" in r for r in reasons), + f"key={key} strict={strict}: {reasons}") + + def test_extra_key_asymmetry_preserved(self): + # #889 R2 由来の意図的非対称の両側固定(TASK-0896 R-005 / TC-6) + rv = {"model_a": _snap(extra="x"), "model_b": _snap()} + self.assertTrue(c3_contract.check_snapshot_trio(_container(), rv, strict_keys=True)) + self.assertEqual(c3_contract.check_snapshot_trio(_container(), rv, strict_keys=False), []) + + def test_container_none_values_do_not_pass(self): + # container 側 None と snapshot 側の値ありは不一致(None==None 偶然一致は + # 空値検査が先に落とすため通過しない) + rv = {"model_a": _snap(), "model_b": _snap()} + reasons = c3_contract.check_snapshot_trio({}, rv, strict_keys=False) + self.assertTrue(reasons) + + def test_reason_order_contract(self): + # 順序契約(R-004): reviewer 順(model_a → model_b)、reviewer 内は + # キー集合 → 空値 → 三つ組の検査順。複合異常で固定する。 + rv = {"model_a": _snap(verdict=""), + "model_b": _snap(plan_hash="sha256:tampered")} + reasons = c3_contract.check_snapshot_trio(_container(), rv, strict_keys=False) + self.assertEqual(len(reasons), 2) + self.assertIn("model_a", reasons[0]) + self.assertIn("キー欠落または空値", reasons[0]) + self.assertIn("model_b.plan_hash", reasons[1]) + + def test_representative_wording_regression(self): + # 代表文言の回帰固定(R-004: 先頭要素が外部へ出るため文言を固定) + rv = {"model_a": _snap(extra="x"), "model_b": _snap()} + strict = c3_contract.check_snapshot_trio(_container(), rv, strict_keys=True) + self.assertEqual(strict[0], "reviewers.model_a の snapshot キーが規定 5 キーと不一致") + rv2 = {"model_a": _snap(source_sha="zzz"), "model_b": _snap()} + trio = c3_contract.check_snapshot_trio(_container(), rv2, strict_keys=False) + self.assertEqual( + trio[0], + "reviewers.model_a.source_sha がトップレベル値と不一致" + "(同一 Plan Package を観ていない = AC-5 違反)") + + def test_purity_no_io(self): + # I/O 封じ純粋性(R-003 / AC-3): open を封じても純関数群は成功する + import builtins + real_open = builtins.open + real_read_bytes = pathlib.Path.read_bytes + + def _blocked(*a, **k): + raise AssertionError("I/O in pure function") + builtins.open = _blocked + pathlib.Path.read_bytes = _blocked + try: + rv = {"model_a": _snap(evidence_ref="a"), "model_b": _snap(evidence_ref="b")} + self.assertEqual(c3_contract.check_snapshot_trio(_container(), rv, strict_keys=True), []) + self.assertTrue(c3_contract.canonical_hash({"k": 1}).startswith("sha256:")) + finally: + builtins.open = real_open + pathlib.Path.read_bytes = real_read_bytes + + def test_arbiter_does_not_touch_io_layer(self): + # AC-6 回帰検査: arbiter のコードが sha256_of_file を参照・呼出しない + # (コメント/docstring の言及は許容するため AST の属性・名前参照で検査) + import ast + import arbiter + src = pathlib.Path(arbiter.__file__).read_text(encoding="utf-8") + for node in ast.walk(ast.parse(src)): + if isinstance(node, ast.Attribute): + self.assertNotEqual(node.attr, "sha256_of_file") + if isinstance(node, ast.Name): + self.assertNotEqual(node.id, "sha256_of_file") + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/ai-loop/arbiter.py b/scripts/ai-loop/arbiter.py index d09c4742..4c31bccf 100644 --- a/scripts/ai-loop/arbiter.py +++ b/scripts/ai-loop/arbiter.py @@ -475,21 +475,12 @@ def plan_quality_check(gates: Any) -> bool: # snapshot 欠落)→ priority 1.65 で BLOCKED(契約 §3/§4 の fail-closed) # escalate / BLOCKED 条件を追加するだけの安全側変更であり、以前 escalate だった # 経路を auto-approve に倒す効果は持たない。 -PLAN_PACKAGE_REQUIRED_KEYS = ( - "plan_hash", - "source_sha", - "plan_package_hash", - "c1_evidence_ref", - "c2_evidence_ref", - "reviewers", -) -SNAPSHOT_REQUIRED_KEYS = ( - "verdict", - "plan_hash", - "source_sha", - "plan_package_hash", - "evidence_ref", -) +# 契約定数は c3_contract が単一定義(TASK-0896 / #896)。arbiter は I/O なし層 +# (定数・純関数)のみを import する(AC-6: sha256_of_file は import / call しない)。 +import c3_contract # noqa: E402 + +PLAN_PACKAGE_REQUIRED_KEYS = c3_contract.PLAN_PACKAGE_REQUIRED_KEYS +SNAPSHOT_REQUIRED_KEYS = c3_contract.SNAPSHOT_KEYS def plan_package_check(plan_package: Any, target_sha: str) -> tuple[bool, bool, str]: @@ -507,19 +498,12 @@ def plan_package_check(plan_package: Any, target_sha: str) -> tuple[bool, bool, reviewers = plan_package.get("reviewers") if not isinstance(reviewers, dict): return True, False, "plan_package.reviewers が object でない" - for model in ("model_a", "model_b"): - snap = reviewers.get(model) - if not isinstance(snap, dict): - return True, False, f"reviewers.{model} の snapshot が欠落(契約 §3: BLOCKED)" - snap_missing = [k for k in SNAPSHOT_REQUIRED_KEYS if not snap.get(k)] - if snap_missing: - return True, False, f"reviewers.{model} の snapshot キー欠落: {', '.join(snap_missing)}" - for key in ("plan_hash", "source_sha", "plan_package_hash"): - if snap.get(key) != plan_package.get(key): - return True, False, ( - f"reviewers.{model}.{key} がトップレベル値と不一致" - "(同一 Plan Package を観ていない = AC-5 違反)" - ) + # snapshot 5 キー整合 + 三つ組一致は共通純関数(strict_keys=False = 余剰キー + # 許容の既存挙動を保存)。理由リスト非空 → integrity NG(先頭を reason に)。 + trio_reasons = c3_contract.check_snapshot_trio( + plan_package, reviewers, strict_keys=False) + if trio_reasons: + return True, False, trio_reasons[0] if plan_package.get("source_sha") != target_sha: return True, False, ( f"plan_package.source_sha ({plan_package.get('source_sha')}) と " diff --git a/scripts/ai-loop/c3_contract.py b/scripts/ai-loop/c3_contract.py new file mode 100644 index 00000000..c9c3cf95 --- /dev/null +++ b/scripts/ai-loop/c3_contract.py @@ -0,0 +1,121 @@ +"""c3-prime 契約の共通契約層(TASK-0896 / #896)。 + +契約正本: docs/workflows/ai-loop/c3-prime-contract.md(本モジュールは実装集約のみ・ +規則は不変)。arbiter.py(入力ブロック検証)/ plan_package.py(producer)/ +c3prime_verify.py(record 受理器)の 3 消費者が import 参照する単一定義。 + +層区分(AC-6): +- 契約定数: I/O なし。全消費者が参照可 +- I/O なし純関数: canonical_hash / check_snapshot_trio。arbiter が import してよいのは + この層まで(arbiter は「入力 dict のみで決定論裁定」の設計を維持し、共通層から + 新たなファイル読取依存を持ち込まない) +- I/O あり関数: sha256_of_file。producer / 受理器のみが使用し、arbiter は + import / call しない(test_c3_contract.py が回帰検査) +""" +from __future__ import annotations + +import hashlib +import json +import pathlib + +# --------------------------------------------------------------------------- +# 契約定数(契約 §1/§2) +# --------------------------------------------------------------------------- + +# 契約 §1: Plan Package 6 要素(key 順は artifact_hashes の表示順にも使う) +ARTIFACTS = ( + "pbi-input.md", + "plan.md", + "todo.md", + "test-cases.md", + "review-self.md", + "review-external.md", +) + +# 契約 §2: decision の 3 値 allowlist(#887 レビュー / 両 Codex major 指摘反映)。 +VALID_DECISIONS = ("AUTO_APPROVED", "HUMAN_ESCALATED", "BLOCKED") +VALID_VERDICTS = ("approve", "reject") + +# 契約 §3: reviewer snapshot の 5 キー(三つ組照合の対象は plan_hash / +# source_sha / plan_package_hash)。 +SNAPSHOT_KEYS = ("verdict", "plan_hash", "source_sha", "plan_package_hash", "evidence_ref") + +# 契約 §2: c3-prime record トップレベルの必須キー(受理器 allowlist の中核)。 +RECORD_REQUIRED_KEYS = ( + "task_id", "approval_kind", "phase", "decision", "source_sha", "plan_hash", + "plan_package_hash", "artifact_hashes", "c1_evidence_ref", "c2_evidence_ref", + "reviewers", "policy_ref", "issued_at", "issued_by", +) +# 任意で許容する追加キー(それ以外の未知キーは reject。`c3_status` は §5 で明示禁止)。 +RECORD_OPTIONAL_KEYS = ("derived_loopspec_hash",) +RECORD_ALLOWED_KEYS = set(RECORD_REQUIRED_KEYS) | set(RECORD_OPTIONAL_KEYS) + +# 契約 §2/§3: arbiter 入力 `plan_package` ブロックの必須キー。 +PLAN_PACKAGE_REQUIRED_KEYS = ( + "plan_hash", + "source_sha", + "plan_package_hash", + "c1_evidence_ref", + "c2_evidence_ref", + "reviewers", +) + +# 三つ組照合の対象キー(snapshot 値 = container トップレベル値の一致要求)。 +TRIO_KEYS = ("plan_hash", "source_sha", "plan_package_hash") + + +# --------------------------------------------------------------------------- +# I/O なし純関数 +# --------------------------------------------------------------------------- + +def canonical_hash(obj) -> str: + """正規化 JSON(sort_keys・区切り最小)の sha256 を返す(契約 §2)。""" + canon = json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8") + return "sha256:" + hashlib.sha256(canon).hexdigest() + + +def check_snapshot_trio(container, reviewers, strict_keys) -> list[str]: + """reviewer snapshot の 5 キー整合 + 三つ組一致を検査し理由リストを返す。 + + 空リスト = OK。判定・終端制御(arbiter の tuple 部分成功 / c3prime_verify の + 即時 reject)は呼び出し側の責務。理由リストの生成順序は検査順で契約固定: + reviewer 順(model_a → model_b)× reviewer 内は + (1) snapshot 型 / キー集合 → (2) 空値 → (3) 三つ組不一致(TASK-0896 R-004)。 + + strict_keys の非対称は #889 R2 由来の意図的設計を保存する(TASK-0896 R-005): + - strict_keys=True(c3prime_verify): snapshot はちょうど 5 キー(余剰 reject) + - strict_keys=False(arbiter): 欠落・空値のみ検査(余剰キー許容) + + 本関数が検査しないもの(呼び出し側残置): verdict 語彙 / evidence_ref 独立性 / + AUTO_APPROVED 整合 / reviewers ちょうど 2 者(strict 側)/ 余剰 reviewer 許容 + (lenient 側)/ PLAN_PACKAGE_REQUIRED_KEYS 構造検査 / source_sha vs target_sha。 + """ + reasons: list[str] = [] + for model in ("model_a", "model_b"): + snap = reviewers.get(model) if isinstance(reviewers, dict) else None + if not isinstance(snap, dict): + reasons.append(f"reviewers.{model} の snapshot が欠落(契約 §3: fail-closed)") + continue + if strict_keys and set(snap) != set(SNAPSHOT_KEYS): + reasons.append(f"reviewers.{model} の snapshot キーが規定 5 キーと不一致") + continue + missing = [k for k in SNAPSHOT_KEYS if not snap.get(k)] + if missing: + reasons.append( + f"reviewers.{model} の snapshot キー欠落または空値: {', '.join(missing)}") + continue + for key in TRIO_KEYS: + if snap.get(key) != (container.get(key) if isinstance(container, dict) else None): + reasons.append( + f"reviewers.{model}.{key} がトップレベル値と不一致" + "(同一 Plan Package を観ていない = AC-5 違反)") + return reasons + + +# --------------------------------------------------------------------------- +# I/O あり関数(arbiter は import / call しない — AC-6) +# --------------------------------------------------------------------------- + +def sha256_of_file(path) -> str: + """ファイル内容の sha256 を `sha256:<64hex>` 形式で返す(契約 §2)。""" + return "sha256:" + hashlib.sha256(pathlib.Path(path).read_bytes()).hexdigest() diff --git a/scripts/ai-loop/c3prime_verify.py b/scripts/ai-loop/c3prime_verify.py index e14a0894..bba38909 100755 --- a/scripts/ai-loop/c3prime_verify.py +++ b/scripts/ai-loop/c3prime_verify.py @@ -16,35 +16,26 @@ """ from __future__ import annotations -import hashlib import json import pathlib import re import sys sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) +import c3_contract # noqa: E402 契約定数の単一定義(TASK-0896 / #896) import plan_package # noqa: E402 受理側でも evidence marker を再検証するため共有 -ARTIFACTS = ( - "pbi-input.md", "plan.md", "todo.md", "test-cases.md", - "review-self.md", "review-external.md", -) -VALID_DECISIONS = ("AUTO_APPROVED", "HUMAN_ESCALATED", "BLOCKED") -VALID_VERDICTS = ("approve", "reject") -SNAPSHOT_KEYS = ("verdict", "plan_hash", "source_sha", "plan_package_hash", "evidence_ref") -# 契約 §2: c3-prime トップレベルの必須キー(allowlist の中核)。 -REQUIRED_KEYS = ( - "task_id", "approval_kind", "phase", "decision", "source_sha", "plan_hash", - "plan_package_hash", "artifact_hashes", "c1_evidence_ref", "c2_evidence_ref", - "reviewers", "policy_ref", "issued_at", "issued_by", -) -# 任意で許容する追加キー(それ以外の未知キーは reject。`c3_status` は §5 で明示禁止)。 -OPTIONAL_KEYS = ("derived_loopspec_hash",) -ALLOWED_KEYS = set(REQUIRED_KEYS) | set(OPTIONAL_KEYS) +# 契約定数は c3_contract が単一定義(値の契約固定は test_c3_contract.py)。 +ARTIFACTS = c3_contract.ARTIFACTS +VALID_DECISIONS = c3_contract.VALID_DECISIONS +VALID_VERDICTS = c3_contract.VALID_VERDICTS +SNAPSHOT_KEYS = c3_contract.SNAPSHOT_KEYS +REQUIRED_KEYS = c3_contract.RECORD_REQUIRED_KEYS +OPTIONAL_KEYS = c3_contract.RECORD_OPTIONAL_KEYS +ALLOWED_KEYS = c3_contract.RECORD_ALLOWED_KEYS -def _sha256(path: pathlib.Path) -> str: - return "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest() +_sha256 = c3_contract.sha256_of_file # 単一実装(TASK-0896 AC-2) def _fail(msg: str) -> int: @@ -148,8 +139,7 @@ def main(argv): return _fail(f"artifact_hashes 不一致: {name}(stale)") # plan_package_hash = artifact_hashes の正規化 JSON の sha256 - canon = json.dumps(ah, sort_keys=True, separators=(",", ":")).encode("utf-8") - if data.get("plan_package_hash") != "sha256:" + hashlib.sha256(canon).hexdigest(): + if data.get("plan_package_hash") != c3_contract.canonical_hash(ah): return _fail("plan_package_hash が artifact_hashes から再計算した値と不一致") # reviewer snapshot 三つ組一致 + decision-verdict 整合 @@ -157,18 +147,14 @@ def main(argv): # model_a / model_b ちょうど 2 者(余剰 reviewer キーは reject / #889 R2 medium)。 if not isinstance(reviewers, dict) or set(reviewers) != {"model_a", "model_b"}: return _fail(f"reviewers は model_a / model_b のちょうど 2 者: {sorted(reviewers) if isinstance(reviewers, dict) else reviewers!r}") + # snapshot 5 キー整合(strict_keys=True = ちょうど 5 キー・未知ネストキーは + # reject / #889 R2 medium)+ 三つ組一致は共通純関数。理由リスト非空 → reject。 + trio_reasons = c3_contract.check_snapshot_trio(data, reviewers, strict_keys=True) + if trio_reasons: + return _fail(trio_reasons[0]) for m in ("model_a", "model_b"): - snap = reviewers.get(m) - # snapshot は 5 キーちょうど(未知ネストキーは reject / #889 R2 medium)。 - if not isinstance(snap, dict) or set(snap) != set(SNAPSHOT_KEYS): - return _fail(f"reviewers.{m} の snapshot キーが規定 5 キーと不一致") - if any(not snap.get(k) for k in SNAPSHOT_KEYS): - return _fail(f"reviewers.{m} の snapshot に空値") - if snap.get("verdict") not in VALID_VERDICTS: + if reviewers[m].get("verdict") not in VALID_VERDICTS: return _fail(f"reviewers.{m}.verdict が approve/reject 以外") - for key in ("plan_hash", "source_sha", "plan_package_hash"): - if snap.get(key) != data.get(key): - return _fail(f"reviewers.{m}.{key} がトップレベル値と不一致(AC-5 違反)") # reviewer 独立性: 両者の evidence_ref が同一なら独立 2 者レビュー偽装 # (#889 R2 high。snapshot hash は同一が正だが evidence は別根拠であるべき)。 if reviewers["model_a"]["evidence_ref"] == reviewers["model_b"]["evidence_ref"]: diff --git a/scripts/ai-loop/plan_package.py b/scripts/ai-loop/plan_package.py index b3555d1a..28571db3 100644 --- a/scripts/ai-loop/plan_package.py +++ b/scripts/ai-loop/plan_package.py @@ -15,7 +15,6 @@ from __future__ import annotations import argparse -import hashlib import json import pathlib import re @@ -23,15 +22,11 @@ TASK_ID_RE = re.compile(r"^TASK-[0-9]{4}$") -# 契約 §1: Plan Package 6 要素(key 順は artifact_hashes の表示順にも使う) -ARTIFACTS = ( - "pbi-input.md", - "plan.md", - "todo.md", - "test-cases.md", - "review-self.md", - "review-external.md", -) +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) +import c3_contract # noqa: E402 契約定数の単一定義(TASK-0896 / #896) + +# 契約 §1: Plan Package 6 要素(単一定義は c3_contract) +ARTIFACTS = c3_contract.ARTIFACTS C1_EVIDENCE = "review-self.md" C2_EVIDENCE = "review-external.md" @@ -78,9 +73,9 @@ def check_presence(task_dir): _C1_PREFIX_RE = re.compile(r"^C1-VERDICT:", re.MULTILINE) _C2_PREFIX_RE = re.compile(r"^C2-VERDICT:", re.MULTILINE) -# 契約 §2: decision の 3 値 allowlist(#887 レビュー / 両 Codex major 指摘反映)。 -VALID_DECISIONS = ("AUTO_APPROVED", "HUMAN_ESCALATED", "BLOCKED") -VALID_VERDICTS = ("approve", "reject") +# 契約 §2: decision の 3 値 allowlist(単一定義は c3_contract)。 +VALID_DECISIONS = c3_contract.VALID_DECISIONS +VALID_VERDICTS = c3_contract.VALID_VERDICTS def _read_evidence_marker(path, marker_re, prefix_re): @@ -134,8 +129,7 @@ def check_evidence(task_dir): return errors -def _sha256_of(path): - return "sha256:" + hashlib.sha256(pathlib.Path(path).read_bytes()).hexdigest() +_sha256_of = c3_contract.sha256_of_file # 単一実装(TASK-0896 AC-2) def compute_hashes(task_dir): @@ -145,12 +139,10 @@ def compute_hashes(task_dir): if presence: raise PlanPackageError(presence) artifact_hashes = {name: _sha256_of(task_dir / name) for name in ARTIFACTS} - canon = json.dumps(artifact_hashes, sort_keys=True, - separators=(",", ":")).encode("utf-8") return { "artifact_hashes": artifact_hashes, "plan_hash": artifact_hashes["plan.md"], - "plan_package_hash": "sha256:" + hashlib.sha256(canon).hexdigest(), + "plan_package_hash": c3_contract.canonical_hash(artifact_hashes), } diff --git a/scripts/ai-loop/test_c3_contract.py b/scripts/ai-loop/test_c3_contract.py new file mode 100644 index 00000000..18ae2b41 --- /dev/null +++ b/scripts/ai-loop/test_c3_contract.py @@ -0,0 +1,215 @@ +"""c3_contract.py の契約固定テスト(TASK-0896)。 + +対象は共通層の純関数境界値と定数契約のみ。偽造 record 14 パターンの受理器 +統合テストは test_c3prime_verify.py に残置する(pbi-input Unknowns 確定)。 +""" +import pathlib +import sys +import unittest + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) +import c3_contract # noqa: E402 + + +class ConstantContractTests(unittest.TestCase): + """契約定数の値固定(コミット a: 移行前実装との byte 同一を将来にわたり固定)。""" + + def test_artifacts(self): + self.assertEqual(c3_contract.ARTIFACTS, ( + "pbi-input.md", "plan.md", "todo.md", "test-cases.md", + "review-self.md", "review-external.md", + )) + + def test_decisions_verdicts(self): + self.assertEqual(c3_contract.VALID_DECISIONS, + ("AUTO_APPROVED", "HUMAN_ESCALATED", "BLOCKED")) + self.assertEqual(c3_contract.VALID_VERDICTS, ("approve", "reject")) + + def test_snapshot_keys(self): + self.assertEqual(c3_contract.SNAPSHOT_KEYS, ( + "verdict", "plan_hash", "source_sha", "plan_package_hash", "evidence_ref")) + + def test_record_keys(self): + self.assertEqual(c3_contract.RECORD_REQUIRED_KEYS, ( + "task_id", "approval_kind", "phase", "decision", "source_sha", "plan_hash", + "plan_package_hash", "artifact_hashes", "c1_evidence_ref", "c2_evidence_ref", + "reviewers", "policy_ref", "issued_at", "issued_by", + )) + self.assertEqual(c3_contract.RECORD_OPTIONAL_KEYS, ("derived_loopspec_hash",)) + self.assertEqual( + c3_contract.RECORD_ALLOWED_KEYS, + set(c3_contract.RECORD_REQUIRED_KEYS) | set(c3_contract.RECORD_OPTIONAL_KEYS)) + + def test_plan_package_keys(self): + self.assertEqual(c3_contract.PLAN_PACKAGE_REQUIRED_KEYS, ( + "plan_hash", "source_sha", "plan_package_hash", + "c1_evidence_ref", "c2_evidence_ref", "reviewers", + )) + + def test_trio_keys(self): + self.assertEqual(c3_contract.TRIO_KEYS, + ("plan_hash", "source_sha", "plan_package_hash")) + + +class ConsumerAliasTests(unittest.TestCase): + """3 消費者が c3_contract の定数を参照している(is 同一 = 別定義でない)。""" + + def test_plan_package_aliases(self): + import plan_package + self.assertIs(plan_package.ARTIFACTS, c3_contract.ARTIFACTS) + self.assertIs(plan_package.VALID_DECISIONS, c3_contract.VALID_DECISIONS) + self.assertIs(plan_package.VALID_VERDICTS, c3_contract.VALID_VERDICTS) + + def test_c3prime_verify_aliases(self): + import c3prime_verify + self.assertIs(c3prime_verify.ARTIFACTS, c3_contract.ARTIFACTS) + self.assertIs(c3prime_verify.VALID_DECISIONS, c3_contract.VALID_DECISIONS) + self.assertIs(c3prime_verify.VALID_VERDICTS, c3_contract.VALID_VERDICTS) + self.assertIs(c3prime_verify.SNAPSHOT_KEYS, c3_contract.SNAPSHOT_KEYS) + self.assertIs(c3prime_verify.REQUIRED_KEYS, c3_contract.RECORD_REQUIRED_KEYS) + self.assertIs(c3prime_verify.OPTIONAL_KEYS, c3_contract.RECORD_OPTIONAL_KEYS) + self.assertEqual(c3prime_verify.ALLOWED_KEYS, c3_contract.RECORD_ALLOWED_KEYS) + + def test_arbiter_aliases(self): + import arbiter + self.assertIs(arbiter.PLAN_PACKAGE_REQUIRED_KEYS, + c3_contract.PLAN_PACKAGE_REQUIRED_KEYS) + self.assertIs(arbiter.SNAPSHOT_REQUIRED_KEYS, c3_contract.SNAPSHOT_KEYS) + + + +class HashTests(unittest.TestCase): + """hash ヘルパーの境界値(コミット b)。""" + + def test_canonical_hash_deterministic_and_order_independent(self): + import hashlib as _h + expected = "sha256:" + _h.sha256(b'{"a":1,"b":2}').hexdigest() + self.assertEqual(c3_contract.canonical_hash({"a": 1, "b": 2}), expected) + self.assertEqual(c3_contract.canonical_hash({"b": 2, "a": 1}), expected) + + def test_canonical_hash_empty(self): + import hashlib as _h + self.assertEqual(c3_contract.canonical_hash({}), + "sha256:" + _h.sha256(b"{}").hexdigest()) + + def test_sha256_of_file_detects_one_byte_change(self): + import tempfile + with tempfile.TemporaryDirectory() as d: + p1 = pathlib.Path(d) / "a.md" + p2 = pathlib.Path(d) / "b.md" + p1.write_bytes(b"plan body\n") + p2.write_bytes(b"plan bodY\n") + h1 = c3_contract.sha256_of_file(p1) + self.assertTrue(h1.startswith("sha256:") and len(h1) == 71) + self.assertEqual(h1, c3_contract.sha256_of_file(p1)) + self.assertNotEqual(h1, c3_contract.sha256_of_file(p2)) + +def _snap(**over): + base = {"verdict": "approve", "plan_hash": "sha256:aa", "source_sha": "bbb", + "plan_package_hash": "sha256:cc", "evidence_ref": "docs/x.md"} + base.update(over) + return base + + +def _container(): + return {"plan_hash": "sha256:aa", "source_sha": "bbb", "plan_package_hash": "sha256:cc"} + + +class SnapshotTrioTests(unittest.TestCase): + """check_snapshot_trio の理由リスト契約(コミット c)。""" + + def test_ok_both_modes(self): + rv = {"model_a": _snap(evidence_ref="docs/a.md"), "model_b": _snap(evidence_ref="docs/b.md")} + self.assertEqual(c3_contract.check_snapshot_trio(_container(), rv, strict_keys=True), []) + self.assertEqual(c3_contract.check_snapshot_trio(_container(), rv, strict_keys=False), []) + + def test_missing_snapshot_and_non_dict(self): + for rv in ({}, None, {"model_a": _snap()}, {"model_a": "str", "model_b": _snap()}): + for strict in (True, False): + reasons = c3_contract.check_snapshot_trio(_container(), rv, strict_keys=strict) + self.assertTrue(reasons, f"rv={rv!r} strict={strict}") + + def test_empty_value(self): + rv = {"model_a": _snap(plan_hash=""), "model_b": _snap()} + for strict in (True, False): + reasons = c3_contract.check_snapshot_trio(_container(), rv, strict_keys=strict) + self.assertEqual(len(reasons), 1) + self.assertIn("model_a", reasons[0]) + + def test_trio_mismatch_each_key(self): + for key in c3_contract.TRIO_KEYS: + rv = {"model_a": _snap(**{key: "sha256:tampered"}), "model_b": _snap()} + for strict in (True, False): + reasons = c3_contract.check_snapshot_trio(_container(), rv, strict_keys=strict) + self.assertTrue(any(f"model_a.{key}" in r for r in reasons), + f"key={key} strict={strict}: {reasons}") + + def test_extra_key_asymmetry_preserved(self): + # #889 R2 由来の意図的非対称の両側固定(TASK-0896 R-005 / TC-6) + rv = {"model_a": _snap(extra="x"), "model_b": _snap()} + self.assertTrue(c3_contract.check_snapshot_trio(_container(), rv, strict_keys=True)) + self.assertEqual(c3_contract.check_snapshot_trio(_container(), rv, strict_keys=False), []) + + def test_container_none_values_do_not_pass(self): + # container 側 None と snapshot 側の値ありは不一致(None==None 偶然一致は + # 空値検査が先に落とすため通過しない) + rv = {"model_a": _snap(), "model_b": _snap()} + reasons = c3_contract.check_snapshot_trio({}, rv, strict_keys=False) + self.assertTrue(reasons) + + def test_reason_order_contract(self): + # 順序契約(R-004): reviewer 順(model_a → model_b)、reviewer 内は + # キー集合 → 空値 → 三つ組の検査順。複合異常で固定する。 + rv = {"model_a": _snap(verdict=""), + "model_b": _snap(plan_hash="sha256:tampered")} + reasons = c3_contract.check_snapshot_trio(_container(), rv, strict_keys=False) + self.assertEqual(len(reasons), 2) + self.assertIn("model_a", reasons[0]) + self.assertIn("キー欠落または空値", reasons[0]) + self.assertIn("model_b.plan_hash", reasons[1]) + + def test_representative_wording_regression(self): + # 代表文言の回帰固定(R-004: 先頭要素が外部へ出るため文言を固定) + rv = {"model_a": _snap(extra="x"), "model_b": _snap()} + strict = c3_contract.check_snapshot_trio(_container(), rv, strict_keys=True) + self.assertEqual(strict[0], "reviewers.model_a の snapshot キーが規定 5 キーと不一致") + rv2 = {"model_a": _snap(source_sha="zzz"), "model_b": _snap()} + trio = c3_contract.check_snapshot_trio(_container(), rv2, strict_keys=False) + self.assertEqual( + trio[0], + "reviewers.model_a.source_sha がトップレベル値と不一致" + "(同一 Plan Package を観ていない = AC-5 違反)") + + def test_purity_no_io(self): + # I/O 封じ純粋性(R-003 / AC-3): open を封じても純関数群は成功する + import builtins + real_open = builtins.open + real_read_bytes = pathlib.Path.read_bytes + + def _blocked(*a, **k): + raise AssertionError("I/O in pure function") + builtins.open = _blocked + pathlib.Path.read_bytes = _blocked + try: + rv = {"model_a": _snap(evidence_ref="a"), "model_b": _snap(evidence_ref="b")} + self.assertEqual(c3_contract.check_snapshot_trio(_container(), rv, strict_keys=True), []) + self.assertTrue(c3_contract.canonical_hash({"k": 1}).startswith("sha256:")) + finally: + builtins.open = real_open + pathlib.Path.read_bytes = real_read_bytes + + def test_arbiter_does_not_touch_io_layer(self): + # AC-6 回帰検査: arbiter のコードが sha256_of_file を参照・呼出しない + # (コメント/docstring の言及は許容するため AST の属性・名前参照で検査) + import ast + import arbiter + src = pathlib.Path(arbiter.__file__).read_text(encoding="utf-8") + for node in ast.walk(ast.parse(src)): + if isinstance(node, ast.Attribute): + self.assertNotEqual(node.attr, "sha256_of_file") + if isinstance(node, ast.Name): + self.assertNotEqual(node.id, "sha256_of_file") + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/sync-plugin-plangate.sh b/scripts/sync-plugin-plangate.sh index 999e90a3..659c737a 100755 --- a/scripts/sync-plugin-plangate.sh +++ b/scripts/sync-plugin-plangate.sh @@ -305,7 +305,9 @@ fi if [ -d "$AI_LOOP_SCRIPTS_DIR" ]; then # plan_package.py + test_plan_package.py(TASK-0872 / R-008: Plan-first 束縛層。 # 明示列挙に無いと plugin 配布物からサイレント欠落するため必ず対で列挙する) - for _f in "$AI_LOOP_SCRIPTS_DIR/arbiter.py" "$AI_LOOP_SCRIPTS_DIR/test_arbiter.py" "$AI_LOOP_SCRIPTS_DIR/metrics.py" "$AI_LOOP_SCRIPTS_DIR/test_metrics.py" "$AI_LOOP_SCRIPTS_DIR/plan_package.py" "$AI_LOOP_SCRIPTS_DIR/test_plan_package.py" "$AI_LOOP_SCRIPTS_DIR/c3prime_verify.py" "$AI_LOOP_SCRIPTS_DIR/test_c3prime_verify.py"; do + # c3_contract.py + test_c3_contract.py(TASK-0896: arbiter/plan_package/ + # c3prime_verify が import する共通契約層。列挙漏れは bundled 側 import エラー) + for _f in "$AI_LOOP_SCRIPTS_DIR/arbiter.py" "$AI_LOOP_SCRIPTS_DIR/test_arbiter.py" "$AI_LOOP_SCRIPTS_DIR/metrics.py" "$AI_LOOP_SCRIPTS_DIR/test_metrics.py" "$AI_LOOP_SCRIPTS_DIR/plan_package.py" "$AI_LOOP_SCRIPTS_DIR/test_plan_package.py" "$AI_LOOP_SCRIPTS_DIR/c3prime_verify.py" "$AI_LOOP_SCRIPTS_DIR/test_c3prime_verify.py" "$AI_LOOP_SCRIPTS_DIR/c3_contract.py" "$AI_LOOP_SCRIPTS_DIR/test_c3_contract.py"; do [ -f "$_f" ] || continue _sync_ai_loop_file "$_f" "$PLUGIN_AI_LOOP_SCRIPTS" "skills/ai-loop-cycle/scripts" done @@ -315,7 +317,7 @@ if [ -d "$PLUGIN_AI_LOOP_SCRIPTS" ]; then [ -f "$_f" ] || continue _base="$(basename "$_f")" case "$_base" in - arbiter.py|test_arbiter.py|metrics.py|test_metrics.py|plan_package.py|test_plan_package.py|c3prime_verify.py|test_c3prime_verify.py) : ;; + arbiter.py|test_arbiter.py|metrics.py|test_metrics.py|plan_package.py|test_plan_package.py|c3prime_verify.py|test_c3prime_verify.py|c3_contract.py|test_c3_contract.py) : ;; *) if [ "$DRY_RUN" = "1" ]; then _drylog "WOULD DELETE: skills/ai-loop-cycle/scripts/$_base" else rm "$_f"; _log "DELETE: skills/ai-loop-cycle/scripts/$_base"; fi diff --git a/tests/extras/ta-55-c3prime-accept.sh b/tests/extras/ta-55-c3prime-accept.sh index 71e9cac1..f87b2d72 100644 --- a/tests/extras/ta-55-c3prime-accept.sh +++ b/tests/extras/ta-55-c3prime-accept.sh @@ -22,6 +22,13 @@ if [ ! -f "$PG_T55_VERIFY" ]; then elif ! command -v python3 >/dev/null 2>&1; then printf ' [SKIP] python3 不在\n' else + # TASK-0896: 共通契約層(c3_contract.py)の単体テストを CI 実行経路に乗せる + if python3 "$PG_T55_ROOT/scripts/ai-loop/test_c3_contract.py" >/dev/null 2>&1; then + t55_pass "test_c3_contract.py 単体テスト(共通契約層)" + else + t55_fail "test_c3_contract.py 単体テスト FAIL" + fi + # sandbox TASK dir を build_c3_prime で生成(plan_package.py を利用) _t55_tmp=$(mktemp -d) register_cleanup "$_t55_tmp"