merge: integrate stochastic robustness experiment

Author: ly028716

README.md

@@ -65,7 +65,8 @@ memory, convergence, and stopping behavior:
 11. [Trace diagnostics](docs/trace-diagnostics.md)
 12. [Diagnosis-driven repair loop](docs/diagnosis-repair-loop.md)
 13. [Multi-repair selection](docs/multi-repair-selection.md)
-14. [Trace difference analysis](docs/trace-diff-analysis.md)
+14. [Stochastic robustness experiment](docs/stochastic-robustness.md)
+15. [Trace difference analysis](docs/trace-diff-analysis.md)
 15. [Semantic regression gate](docs/regression-gate.md)
 15. [Architecture](docs/architecture.md)
 16. [Metrics](docs/metrics.md)

README.zh-CN.md

@@ -1,5 +1,7 @@
 # Loop Engineering Study
 
+[随机性与鲁棒性实验](docs/stochastic-robustness.md)：以固定种子比较策略在随机扰动下的经验表现。
+
 [多修复方案选择](docs/multi-repair-selection.md)：重跑多个确定性候选并按证据选优。
 
 [Trace 差异分析](docs/trace-diff-analysis.md)：对诊断修复前后的 Artifact 定位首个可观察分歧。

docs/experiments.md

@@ -18,6 +18,7 @@ python experiments/sensitivity_analysis.py
 python experiments/trace_diagnostics.py
 python experiments/diagnosis_repair_loop.py
 python experiments/multi_repair_selection.py
+python experiments/stochastic_robustness.py
 python experiments/trace_diff_analysis.py
 python experiments/regression_gate.py
 ```

docs/stochastic-robustness.md

@@ -0,0 +1,23 @@
+# 随机性与鲁棒性实验
+
+本实验以固定随机种子评估四种策略在随机动作失败与有界数值噪声下的经验鲁棒性。
+
+## 运行实验
+
+```powershell
+python experiments/stochastic_robustness.py
+```
+
+报告写入 `.loop/runs/stochastic-robustness/report.json`，并保存每次运行的 Artifact。
+
+## 扰动矩阵
+
+实验运行低、中、高三档扰动，每档覆盖 `fixed`、`error_aware`、`memory_aware`、`adaptive` 四种策略及 8 个固定种子，共 96 次运行。
+
+## 如何阅读报告
+
+每组汇总包含成功率、平均/最差/P90 成本与平均/P90 步数。每个档位的排名优先成功率，再比较成本 P90、平均步数和策略声明顺序。
+
+## 解释边界
+
+结果仅描述固定扰动模型和有限种子集下的经验表现，不构成真实环境分布或统计显著性的证明。

docs/superpowers/sdd/progress.md

@@ -21,3 +21,4 @@
 - Phase 2 semantic regression gate: complete (four semantic contracts, isolated child outputs, and 65 passing tests verified)
 - Phase 2 Trace difference analysis: complete (read-only first-difference comparison, three diagnosis-repair Artifact pairs, structured JSON report, and 73 passing tests verified)
 - Phase 2 multi-repair selection: complete (six deterministic candidate reruns across three cases, stable evidence-based ranking, and 77 passing tests verified)
+- Phase 2 stochastic robustness: complete (96 seeded stochastic runs, 12 strategy-level summaries, stable rankings, and 78 passing tests verified)

experiments/stochastic_robustness.py

@@ -0,0 +1,116 @@
+"""Reproducible random-failure and noisy-execution robustness experiment."""
+
+from __future__ import annotations
+
+import random
+import json
+from pathlib import Path
+
+if __package__:
+    from ._bootstrap import prepare_script_imports
+else:
+    from _bootstrap import prepare_script_imports
+
+    prepare_script_imports(__file__)
+
+from loop_engineering.actions import Action, ActionResult
+from loop_engineering.artifacts import save_run_artifact
+from loop_engineering.evaluators import GoalEvaluator
+from loop_engineering.metrics import MetricReport
+from loop_engineering.models import LoopState
+from loop_engineering.policies import Decision
+from loop_engineering.runner import LoopRunner
+from loop_engineering.stopping import MaxSteps, SuccessReached
+
+from experiments.adaptive_strategy import (
+    AdaptivePolicy,
+    ErrorAwarePolicy,
+    FixedPolicy,
+    MemoryAwarePolicy,
+)
+
+STRATEGIES = ("fixed", "error_aware", "memory_aware", "adaptive")
+SEEDS = (101, 211, 307, 401, 503, 601, 701, 809)
+LEVELS = (("low", 0.05, 0.10), ("medium", 0.15, 0.30), ("high", 0.30, 0.60))
+
+
+class StochasticAction(Action):
+    """Apply an increment with independent random failure and bounded noise."""
+
+    def __init__(
+        self, failure_rate: float, noise_amplitude: float, rng: random.Random
+    ) -> None:
+        if not 0.0 <= failure_rate <= 1.0:
+            raise ValueError("failure_rate must be between 0 and 1.")
+        if noise_amplitude < 0.0:
+            raise ValueError("noise_amplitude must be non-negative.")
+        self._failure_rate = failure_rate
+        self._noise_amplitude = noise_amplitude
+        self._rng = rng
+
+    def apply(self, state: LoopState, decision: Decision) -> ActionResult:
+        if self._rng.random() < self._failure_rate:
+            return ActionResult(
+                state=state.with_value(state.value, stochastic_failure=True),
+                success=False,
+                cost=0.0,
+            )
+        amount = float(decision.parameters["amount"])
+        actual_amount = amount + self._rng.uniform(
+            -self._noise_amplitude, self._noise_amplitude
+        )
+        return ActionResult(
+            state=state.with_value(state.value + actual_amount),
+            success=True,
+            cost=abs(actual_amount),
+        )
+
+
+def _policy_for(strategy: str):
+    return {
+        "fixed": FixedPolicy,
+        "error_aware": ErrorAwarePolicy,
+        "memory_aware": MemoryAwarePolicy,
+        "adaptive": AdaptivePolicy,
+    }[strategy]()
+
+
+def _run(root: Path, level: str, failure_rate: float, noise: float, strategy: str, seed: int) -> dict[str, object]:
+    trace = LoopRunner(
+        policy=_policy_for(strategy),
+        action=StochasticAction(failure_rate, noise, random.Random(seed)),
+        evaluator=GoalEvaluator(tolerance=0.25),
+        stop_conditions=[SuccessReached(), MaxSteps(8)],
+    ).run(LoopState(step=0, value=0.0, goal=6.0))
+    metrics = MetricReport.from_trace(trace)
+    artifact = save_run_artifact(root / f"{level}--{strategy}--{seed}.json", trace, metrics)
+    return {"level": level, "strategy": strategy, "seed": seed, "success": metrics.success, "cost": metrics.cost, "steps": metrics.steps, "final_score": metrics.final_score, "artifact_path": str(artifact)}
+
+
+def _summary(level: str, strategy: str, runs: list[dict[str, object]]) -> dict[str, object]:
+    costs = sorted(float(item["cost"]) for item in runs)
+    steps = sorted(int(item["steps"]) for item in runs)
+    index = 7
+    return {"level": level, "strategy": strategy, "run_count": len(runs), "success_count": sum(bool(item["success"]) for item in runs), "success_rate": sum(bool(item["success"]) for item in runs) / len(runs), "mean_cost": sum(costs) / len(costs), "worst_cost": costs[-1], "cost_p90": costs[index], "mean_steps": sum(steps) / len(steps), "steps_p90": steps[index]}
+
+
+def run_stochastic_robustness(output_dir: str | Path = ".loop/runs/stochastic-robustness") -> dict[str, object]:
+    """Run the fixed stochastic matrix and persist every run artifact."""
+    root = Path(output_dir).resolve()
+    runs = [_run(root, level, rate, noise, strategy, seed) for level, rate, noise in LEVELS for strategy in STRATEGIES for seed in SEEDS]
+    summaries = [_summary(level, strategy, [item for item in runs if item["level"] == level and item["strategy"] == strategy]) for level, _, _ in LEVELS for strategy in STRATEGIES]
+    rankings = {level: sorted([item for item in summaries if item["level"] == level], key=lambda item: (-float(item["success_rate"]), float(item["cost_p90"]), float(item["mean_steps"]), STRATEGIES.index(str(item["strategy"])))) for level, _, _ in LEVELS}
+    result = {"levels": [item[0] for item in LEVELS], "strategies": list(STRATEGIES), "runs": runs, "summaries": summaries, "rankings": rankings}
+    root.mkdir(parents=True, exist_ok=True)
+    (root / "report.json").write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    return result
+
+
+def main() -> None:
+    """Print the complete stochastic robustness report as JSON."""
+
+    print(json.dumps(run_stochastic_robustness(), ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

tests/test_stochastic_robustness.py

@@ -0,0 +1,38 @@
+import random
+
+import pytest
+
+from experiments.stochastic_robustness import StochasticAction
+from loop_engineering.models import LoopState
+from loop_engineering.policies import Decision
+
+
+def test_stochastic_action_is_reproducible_for_the_same_seed() -> None:
+    decision = Decision(name="increment", parameters={"amount": 1.0})
+    first = StochasticAction(0.15, 0.30, random.Random(101))
+    second = StochasticAction(0.15, 0.30, random.Random(101))
+
+    first_results = [first.apply(LoopState(0, 0.0, 6.0), decision) for _ in range(3)]
+    second_results = [second.apply(LoopState(0, 0.0, 6.0), decision) for _ in range(3)]
+
+    assert first_results == second_results
+
+
+@pytest.mark.parametrize("failure_rate,noise", [(-0.01, 0.1), (1.01, 0.1), (0.1, -0.1)])
+def test_stochastic_action_rejects_invalid_parameters(failure_rate: float, noise: float) -> None:
+    with pytest.raises(ValueError):
+        StochasticAction(failure_rate, noise, random.Random(1))
+
+
+def test_robustness_matrix_is_complete_and_reproducible(tmp_path) -> None:
+    from experiments.stochastic_robustness import run_stochastic_robustness
+
+    first = run_stochastic_robustness(tmp_path / "first")
+    second = run_stochastic_robustness(tmp_path / "second")
+
+    assert len(first["runs"]) == 96
+    assert len(first["summaries"]) == 12
+    assert all(item["run_count"] == 8 for item in first["summaries"])
+    assert [item["success"] for item in first["runs"]] == [
+        item["success"] for item in second["runs"]
+    ]