Improvement Executor

[
  {
    "type": "coverage",
    "succeeded": true,
    "context": {
      "dimension": "coverage",
      "scores": {
        "coverage": 1.0,
        "accuracy": 0.8,
        "efficiency": 1.0,
        "reliability": 1.0,
        "security": 0.8
      }
    },
    "timestamp": "2026-04-02T12:14:23Z",
    "hit_count": 1
  },
  {
    "type": "instruction",
    "succeeded": true,
    "context": {
      "dimension": "accuracy",
      "scores": {
        "coverage": 1.0,
        "accuracy": 0.8,
        "efficiency": 1.0,
        "reliability": 1.0,
        "security": 0.8
      }
    },
    "timestamp": "2026-04-02T12:14:23Z",
    "hit_count": 1
  }
]

#!/usr/bin/env python3
"""Executor for low-risk generic-skill candidates."""

from __future__ import annotations

import argparse
import difflib
import sys
from pathlib import Path

_REPO_ROOT = Path(__file__).resolve().parents[3]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from lib.common import (
    EXECUTOR_SUPPORTED_CATEGORIES,
    SCHEMA_VERSION,
    read_json,
    read_text,
    utc_now_iso,
    write_json,
    write_text,
)
from lib.state_machine import (
    DEFAULT_STATE_ROOT,
    backup_file,
    ensure_tree,
    update_state,
)


def capture_execution_trace(candidate: dict, result: dict, error: str | None = None) -> dict:
    """Capture structured execution trace for failure feedback."""
    return {
        "type": "execution_trace",
        "candidate_id": candidate.get("id", "unknown"),
        "category": candidate.get("category", "unknown"),
        "target_path": candidate.get("target_path", ""),
        "action": candidate.get("execution_plan", {}).get("action", "unknown"),
        "execution_status": result.get("status", "unknown"),
        "modified": result.get("modified", False),
        "diff_summary": result.get("diff_summary", {}),
        "error": error,
        "timestamp": utc_now_iso(),
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Execute a ranked generic-skill candidate")
    parser.add_argument("--input", required=True, help="Ranking artifact JSON")
    parser.add_argument("--candidate-id", required=True, help="Candidate id to execute")
    parser.add_argument("--state-root", default=str(DEFAULT_STATE_ROOT))
    parser.add_argument("--output", default=None)
    parser.add_argument("--force", action="store_true", help="Allow execution even when critic did not accept")
    return parser.parse_args()


def append_markdown_section(target_path: Path, plan: dict) -> dict:
    before = read_text(target_path)
    heading = plan["section_heading"].strip()
    content_lines = plan.get("content_lines", [])
    if heading in before:
        return {
            "status": "no_change",
            "modified": False,
            "diff": "",
            "diff_summary": {
                "reason": f"section `{heading}` already present",
                "added_lines": 0,
            },
            "after_content": before,
        }
    section = heading + "\n\n" + "\n".join(f"- {line}" for line in content_lines)
    after = before.rstrip() + "\n\n" + section + "\n"
    write_text(target_path, after)
    diff = "".join(
        difflib.unified_diff(
            before.splitlines(keepends=True),
            after.splitlines(keepends=True),
            fromfile=str(target_path),
            tofile=str(target_path),
        )
    )
    return {
        "status": "success",
        "modified": True,
        "diff": diff,
        "diff_summary": {
            "reason": f"appended markdown section `{heading}`",
            "added_lines": len(content_lines) + 2,
        },
        "after_content": after,
    }



def _make_diff(before, after, path):
    """Generate a unified diff string between before/after content."""
    return "\n".join(difflib.unified_diff(
        before.split("\n"), after.split("\n"),
        fromfile=str(path), tofile=str(path), lineterm="",
    ))


def replace_markdown_section(target_path, plan):
    """Replace a section identified by heading with new content."""
    content = read_text(target_path)
    heading = plan.get("section_heading", "")
    new_lines = plan.get("content_lines", [])
    lines = content.split("\n")
    start_idx = None
    end_idx = len(lines)
    heading_level = len(heading) - len(heading.lstrip("#"))
    for i, line in enumerate(lines):
        if line.strip() == heading.strip():
            start_idx = i
        elif start_idx is not None and line.startswith("#") and (len(line) - len(line.lstrip("#"))) <= heading_level:
            end_idx = i
            break
    if start_idx is None:
        return {"status": "no_change", "modified": False, "diff": "", "diff_summary": {"reason": f"section '{heading}' not found", "changed_lines": 0}}
    replacement = [heading] + [f"- {line}" if not line.startswith("-") else line for line in new_lines]
    new_content = "\n".join(lines[:start_idx] + replacement + lines[end_idx:])
    diff = _make_diff(content, new_content, target_path)
    write_text(target_path, new_content)
    return {"status": "success", "modified": True, "diff": diff, "diff_summary": {"reason": "replaced_section", "changed_lines": len(new_lines)}}


def insert_before_section(target_path, plan):
    """Insert content before a specified section heading."""
    content = read_text(target_path)
    heading = plan.get("section_heading", "")
    new_lines = plan.get("content_lines", [])
    lines = content.split("\n")
    insert_idx = None
    for i, line in enumerate(lines):
        if line.strip() == heading.strip():
            insert_idx = i
            break
    if insert_idx is None:
        return {"status": "no_change", "modified": False, "diff": "", "diff_summary": {"reason": f"section '{heading}' not found", "changed_lines": 0}}
    insertion = [f"- {line}" if not line.startswith("-") and not line.startswith("#") else line for line in new_lines]
    insertion.append("")
    new_content = "\n".join(lines[:insert_idx] + insertion + lines[insert_idx:])
    diff = _make_diff(content, new_content, target_path)
    write_text(target_path, new_content)
    return {"status": "success", "modified": True, "diff": diff, "diff_summary": {"reason": "inserted_before_section", "changed_lines": len(new_lines)}}


def update_yaml_frontmatter(target_path, plan):
    """Update YAML frontmatter fields in a markdown file."""
    content = read_text(target_path)
    updates = plan.get("frontmatter_updates", {})
    if not content.startswith("---"):
        return {"status": "no_change", "modified": False, "diff": "", "diff_summary": {"reason": "no frontmatter found", "changed_lines": 0}}
    parts = content.split("---", 2)
    if len(parts) < 3:
        return {"status": "no_change", "modified": False, "diff": "", "diff_summary": {"reason": "malformed frontmatter", "changed_lines": 0}}
    try:
        import yaml
    except ImportError:
        return {"status": "error", "modified": False, "diff": "", "diff_summary": {"reason": "PyYAML not available", "changed_lines": 0}}
    frontmatter = yaml.safe_load(parts[1]) or {}
    frontmatter.update(updates)
    new_fm = yaml.dump(frontmatter, default_flow_style=False, allow_unicode=True)
    new_content = "---\n" + new_fm + "---" + parts[2]
    diff = _make_diff(content, new_content, target_path)
    write_text(target_path, new_content)
    return {"status": "success", "modified": True, "diff": diff, "diff_summary": {"reason": "updated_frontmatter", "changed_lines": len(updates)}}


ACTION_HANDLERS = {
    "append_markdown_section": append_markdown_section,
    "replace_markdown_section": replace_markdown_section,
    "insert_before_section": insert_before_section,
    "update_yaml_frontmatter": update_yaml_frontmatter,
}


def main() -> int:
    args = parse_args()
    state_root = Path(args.state_root).expanduser().resolve()
    paths = ensure_tree(state_root)

    ranking_artifact = read_json(Path(args.input).expanduser().resolve())
    run_id = ranking_artifact["run_id"]
    target_path = ranking_artifact["target"]["path"]
    candidate = next((item for item in ranking_artifact.get("scored_candidates", []) if item["id"] == args.candidate_id), None)
    if not candidate:
        raise SystemExit(f"candidate not found: {args.candidate_id}")

    recommendation = candidate.get("recommendation")
    if recommendation != "accept_for_execution" and not args.force:
        execution_artifact = {
            "schema_version": SCHEMA_VERSION,
            "lane": ranking_artifact.get("lane", "generic-skill"),
            "run_id": run_id,
            "stage": "executed",
            "status": "unsupported",
            "created_at": utc_now_iso(),
            "candidate_id": candidate["id"],
            "candidate": candidate,
            "source_ranking_artifact": args.input,
            "result": {
                "status": "unsupported",
                "modified": False,
                "reason": f"critic recommendation is `{recommendation}`; use --force to override",
            },
            "next_step": "apply_gate",
            "next_owner": "gate",
        }
        output_path = Path(args.output).expanduser().resolve() if args.output else paths["executions"] / f"{run_id}-{candidate['id']}.json"
        execution_artifact["truth_anchor"] = str(output_path)
        write_json(output_path, execution_artifact)
        update_state(
            state_root,
            run_id=run_id,
            stage="executed",
            status="unsupported",
            target_path=target_path,
            truth_anchor=str(output_path),
            extra={"candidate_id": candidate["id"]},
        )
        print(str(output_path))
        return 0

    category = candidate.get("category")
    if category not in EXECUTOR_SUPPORTED_CATEGORIES:
        result = {
            "status": "unsupported",
            "modified": False,
            "reason": f"category `{category}` is not supported by the first runnable executor",
        }
        output_path = Path(args.output).expanduser().resolve() if args.output else paths["executions"] / f"{run_id}-{candidate['id']}.json"
        artifact = {
            "schema_version": SCHEMA_VERSION,
            "lane": ranking_artifact.get("lane", "generic-skill"),
            "run_id": run_id,
            "stage": "executed",
            "status": result["status"],
            "created_at": utc_now_iso(),
            "candidate_id": candidate["id"],
            "candidate": candidate,
            "source_ranking_artifact": args.input,
            "result": result,
            "next_step": "apply_gate",
            "next_owner": "gate",
            "truth_anchor": str(output_path),
        }
        write_json(output_path, artifact)
        update_state(
            state_root,
            run_id=run_id,
            stage="executed",
            status="unsupported",
            target_path=target_path,
            truth_anchor=str(output_path),
            extra={"candidate_id": candidate["id"]},
        )
        print(str(output_path))
        return 0

    target_file = Path(candidate["target_path"]).expanduser().resolve()
    if not target_file.exists() or not target_file.is_file():
        result = {
            "status": "error",
            "modified": False,
            "reason": f"target file not found: {target_file}",
        }
        output_path = Path(args.output).expanduser().resolve() if args.output else paths["executions"] / f"{run_id}-{candidate['id']}.json"
        artifact = {
            "schema_version": SCHEMA_VERSION,
            "lane": ranking_artifact.get("lane", "generic-skill"),
            "run_id": run_id,
            "stage": "executed",
            "status": "error",
            "created_at": utc_now_iso(),
            "candidate_id": candidate["id"],
            "target_path": str(target_file),
            "result": result,
            "truth_anchor": str(output_path),
        }
        write_json(output_path, artifact)
        update_state(
            state_root, run_id=run_id, stage="executed", status="error",
            target_path=str(target_file), truth_anchor=str(output_path),
            extra={"candidate_id": candidate["id"]},
        )
        print(str(output_path))
        return 1

    backup_path = paths["executions"] / "backups" / run_id / f"{candidate['id']}-{target_file.name}"
    backup_file(target_file, backup_path)

    plan = candidate.get("execution_plan", {})
    action = plan.get("action")
    handler = ACTION_HANDLERS.get(action)
    if handler is not None:
        result = handler(target_file, plan)
    else:
        result = {
            "status": "unsupported",
            "modified": False,
            "reason": f"action `{action}` is not supported",
        }

    error_msg = None if result["status"] == "success" else result.get("reason")
    execution_trace = capture_execution_trace(candidate, result, error=error_msg)

    output_path = Path(args.output).expanduser().resolve() if args.output else paths["executions"] / f"{run_id}-{candidate['id']}.json"
    execution_artifact = {
        "schema_version": SCHEMA_VERSION,
        "lane": ranking_artifact.get("lane", "generic-skill"),
        "run_id": run_id,
        "stage": "executed",
        "status": result["status"],
        "created_at": utc_now_iso(),
        "candidate_id": candidate["id"],
        "candidate": candidate,
        "source_ranking_artifact": args.input,
        "result": {
            **{key: value for key, value in result.items() if key != "after_content"},
            "backup_path": str(backup_path),
            "rollback_pointer": {
                "method": "restore_backup_file",
                "backup_path": str(backup_path),
                "target_path": str(target_file),
            },
        },
        "execution_trace": execution_trace,
        "next_step": "apply_gate",
        "next_owner": "gate",
        "truth_anchor": str(output_path),
    }
    write_json(output_path, execution_artifact)
    update_state(
        state_root,
        run_id=run_id,
        stage="executed",
        status=result["status"],
        target_path=target_path,
        truth_anchor=str(output_path),
        extra={
            "candidate_id": candidate["id"],
            "execution_modified": result.get("modified", False),
        },
    )
    print(str(output_path))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""
Rollback Script — generic-skill lane 回滚工具

支持依据 receipt / backup / rollback pointer 恢复目标文件。

当前支持范围:
- generic-skill lane 的文档类修改回滚
- 基于 backup 文件恢复
- 基于 receipt 追溯回滚信息

用法:
    python rollback.py --receipt <receipt-path>
    python rollback.py --backup <backup-path> --target <target-path>
    python rollback.py --run-id <run-id> --candidate-id <candidate-id>
    python rollback.py --help
"""

import argparse
import json
import os
import shutil
import sys
from pathlib import Path
from datetime import datetime

_REPO_ROOT = Path(__file__).resolve().parents[3]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from lib.common import read_json, write_json


# 默认根目录
DEFAULT_ROOT = Path(os.environ.get("OPENCLAW_ROOT", os.path.expanduser("~/.openclaw"))) / "shared-context/intel/auto-improvement/generic-skill"


def find_receipt(run_id: str, candidate_id: str) -> Path:
    """根据 run_id 和 candidate_id 查找 receipt"""
    receipts_dir = DEFAULT_ROOT / "receipts"
    pattern = f"{run_id}-{candidate_id}-gate.json"
    
    for receipt_file in receipts_dir.glob(pattern):
        return receipt_file
    
    # 尝试模糊匹配
    for receipt_file in receipts_dir.glob(f"*{run_id}*{candidate_id}*.json"):
        return receipt_file
    
    raise FileNotFoundError(f"Receipt not found: {run_id}-{candidate_id}")


def rollback_from_receipt(receipt_path: Path, dry_run: bool = False) -> dict:
    """
    从 receipt 执行回滚
    
    参数:
        receipt_path: receipt JSON 文件路径
        dry_run: 是否仅模拟不实际执行
    
    返回:
        回滚结果字典
    """
    receipt = read_json(receipt_path)
    
    # 检查是否需要回滚
    decision = receipt.get("decision", "")
    if decision == "keep":
        return {
            "status": "skipped",
            "reason": "Receipt decision is 'keep', no rollback needed",
            "receipt_path": str(receipt_path)
        }
    
    # 查找 execution artifact
    execution_path = receipt.get("truth_anchor", {}).get("execution_path")
    if not execution_path:
        # 尝试从 receipt 中推断
        run_id = receipt.get("run_id")
        candidate_id = receipt.get("candidate_id")
        if run_id and candidate_id:
            executions_dir = DEFAULT_ROOT / "executions"
            for exec_file in executions_dir.glob(f"{run_id}-{candidate_id}.json"):
                execution_path = str(exec_file)
                break
    
    if not execution_path:
        return {
            "status": "error",
            "reason": "Cannot find execution artifact",
            "receipt_path": str(receipt_path)
        }
    
    execution = read_json(Path(execution_path))
    
    # 获取 rollback pointer from result
    result = execution.get("result", {})
    rollback_pointer = result.get("rollback_pointer", {})
    backup_path = rollback_pointer.get("backup_path")

    if not backup_path:
        return {
            "status": "error",
            "reason": "No backup information found in execution artifact",
            "execution_path": execution_path
        }

    # 获取目标路径 (truth_anchor is a string path, not a dict)
    target_path = rollback_pointer.get("target_path")
    if not target_path:
        return {
            "status": "error",
            "reason": "No target path found in execution artifact",
            "execution_path": execution_path
        }
    
    # 执行回滚
    backup_file = Path(backup_path)
    target_file = Path(target_path)
    
    if not backup_file.exists():
        return {
            "status": "error",
            "reason": f"Backup file not found: {backup_path}",
            "backup_path": backup_path
        }
    
    result = {
        "status": "dry_run" if dry_run else "success",
        "action": "rollback",
        "target_path": str(target_file),
        "backup_path": str(backup_file),
        "rollback_pointer": rollback_pointer,
        "receipt_path": str(receipt_path),
        "dry_run": dry_run
    }
    
    if not dry_run:
        # 创建目标文件的备份（回滚前的状态）
        pre_rollback_backup = target_file.parent / f"{target_file.name}.pre-rollback-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
        if target_file.exists():
            shutil.copy2(target_file, pre_rollback_backup)
            result["pre_rollback_backup"] = str(pre_rollback_backup)
        
        # 恢复 backup
        shutil.copy2(backup_file, target_file)
        result["restored_from"] = str(backup_file)
    
    return result


def rollback_from_backup(backup_path: str, target_path: str, dry_run: bool = False) -> dict:
    """
    直接从 backup 文件回滚
    
    参数:
        backup_path: backup 文件路径
        target_path: 目标文件路径
        dry_run: 是否仅模拟不实际执行
    
    返回:
        回滚结果字典
    """
    backup_file = Path(backup_path)
    target_file = Path(target_path)
    
    if not backup_file.exists():
        return {
            "status": "error",
            "reason": f"Backup file not found: {backup_path}"
        }
    
    result = {
        "status": "dry_run" if dry_run else "success",
        "action": "rollback",
        "target_path": str(target_file),
        "backup_path": str(backup_file),
        "dry_run": dry_run
    }
    
    if not dry_run:
        # 创建目标文件的备份（回滚前的状态）
        pre_rollback_backup = target_file.parent / f"{target_file.name}.pre-rollback-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
        if target_file.exists():
            shutil.copy2(target_file, pre_rollback_backup)
            result["pre_rollback_backup"] = str(pre_rollback_backup)
        
        # 恢复 backup
        shutil.copy2(backup_file, target_file)
        result["restored_from"] = str(backup_file)
    
    return result


def main():
    parser = argparse.ArgumentParser(
        description="Rollback tool for generic-skill lane",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # 从 receipt 回滚
  python rollback.py --receipt /path/to/receipt.json
  
  # 从 backup 文件回滚
  python rollback.py --backup /path/to/backup.bak --target /path/to/target.md
  
  # 从 run-id 和 candidate-id 回滚
  python rollback.py --run-id run-20260401-143022 --candidate-id cand-001
  
  # 模拟回滚（不实际执行）
  python rollback.py --receipt /path/to/receipt.json --dry-run

支持范围:
  - generic-skill lane 的文档类修改回滚
  - 基于 backup 文件恢复
  - 基于 receipt 追溯回滚信息
  
注意:
  - 回滚前会自动创建当前状态的备份（pre-rollback-*.bak）
  - 仅支持有 backup 信息的 execution
  - 默认 dry-run 模式，需要实际执行请去掉 --dry-run 标志
        """
    )
    
    parser.add_argument(
        "--receipt",
        type=Path,
        help="Receipt JSON file path"
    )
    
    parser.add_argument(
        "--backup",
        type=str,
        help="Backup file path"
    )
    
    parser.add_argument(
        "--target",
        type=str,
        help="Target file path (required with --backup)"
    )
    
    parser.add_argument(
        "--run-id",
        type=str,
        help="Run ID (alternative to --receipt)"
    )
    
    parser.add_argument(
        "--candidate-id",
        type=str,
        help="Candidate ID (required with --run-id)"
    )
    
    parser.add_argument(
        "--dry-run",
        action="store_true",
        default=True,  # 默认 dry-run 模式
        help="Simulate rollback without actual file operations (default: True)"
    )
    
    parser.add_argument(
        "--execute",
        action="store_true",
        help="Actually execute the rollback (override --dry-run)"
    )
    
    args = parser.parse_args()
    
    # 处理 execute 标志
    if args.execute:
        args.dry_run = False
    
    # 验证参数组合
    if args.receipt:
        result = rollback_from_receipt(args.receipt, dry_run=args.dry_run)
    elif args.backup and args.target:
        result = rollback_from_backup(args.backup, args.target, dry_run=args.dry_run)
    elif args.run_id and args.candidate_id:
        try:
            receipt_path = find_receipt(args.run_id, args.candidate_id)
            result = rollback_from_receipt(receipt_path, dry_run=args.dry_run)
        except FileNotFoundError as e:
            result = {"status": "error", "reason": str(e)}
    else:
        parser.print_help()
        sys.exit(1)
    
    # 输出结果
    print(json.dumps(result, indent=2, ensure_ascii=False))
    
    # 错误时退出码非零
    if result.get("status") == "error":
        sys.exit(1)
    
    sys.exit(0)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Tests for the improvement-executor execute module."""

from __future__ import annotations

import importlib.util
import sys
from pathlib import Path

import pytest

# repo root for lib.common imports
_REPO_ROOT = Path(__file__).resolve().parents[3]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))
REPO_ROOT = _REPO_ROOT

_execute_path = REPO_ROOT / "skills" / "improvement-executor" / "scripts" / "execute.py"
_spec = importlib.util.spec_from_file_location("execute", _execute_path)
execute = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(execute)


# ---------------------------------------------------------------------------
# capture_execution_trace
# ---------------------------------------------------------------------------


class TestCaptureExecutionTrace:
    def test_basic_structure(self):
        candidate = {
            "id": "cand-01-readme",
            "category": "docs",
            "target_path": "/tmp/skill/README.md",
            "execution_plan": {"action": "append_markdown_section"},
        }
        result = {
            "status": "success",
            "modified": True,
            "diff_summary": {"reason": "appended section", "added_lines": 4},
        }
        trace = execute.capture_execution_trace(candidate, result)

        assert trace["type"] == "execution_trace"
        assert trace["candidate_id"] == "cand-01-readme"
        assert trace["category"] == "docs"
        assert trace["target_path"] == "/tmp/skill/README.md"
        assert trace["action"] == "append_markdown_section"
        assert trace["execution_status"] == "success"
        assert trace["modified"] is True
        assert trace["diff_summary"]["added_lines"] == 4
        assert trace["error"] is None
        assert "timestamp" in trace

    def test_with_error(self):
        candidate = {"id": "cand-02-ref", "category": "reference"}
        result = {"status": "failed", "modified": False}
        trace = execute.capture_execution_trace(
            candidate, result, error="file not found"
        )
        assert trace["error"] == "file not found"
        assert trace["execution_status"] == "failed"

    def test_missing_fields_default_gracefully(self):
        trace = execute.capture_execution_trace({}, {})
        assert trace["candidate_id"] == "unknown"
        assert trace["category"] == "unknown"
        assert trace["action"] == "unknown"
        assert trace["execution_status"] == "unknown"
        assert trace["modified"] is False


# ---------------------------------------------------------------------------
# append_markdown_section
# ---------------------------------------------------------------------------


class TestAppendMarkdownSection:
    def test_appends_new_section(self, tmp_path):
        md_file = tmp_path / "README.md"
        md_file.write_text("# My Skill\n\nSome content.\n", encoding="utf-8")

        plan = {
            "section_heading": "## Operator Notes",
            "content_lines": [
                "This skill is advisory.",
                "Pair with external tooling.",
            ],
        }
        result = execute.append_markdown_section(md_file, plan)

        assert result["status"] == "success"
        assert result["modified"] is True
        assert result["diff_summary"]["added_lines"] == 4  # heading + blank + 2 lines
        after = md_file.read_text(encoding="utf-8")
        assert "## Operator Notes" in after
        assert "- This skill is advisory." in after
        assert "- Pair with external tooling." in after

    def test_no_change_when_section_exists(self, tmp_path):
        md_file = tmp_path / "README.md"
        md_file.write_text(
            "# My Skill\n\n## Operator Notes\n\n- Already here.\n",
            encoding="utf-8",
        )

        plan = {
            "section_heading": "## Operator Notes",
            "content_lines": ["New line."],
        }
        result = execute.append_markdown_section(md_file, plan)

        assert result["status"] == "no_change"
        assert result["modified"] is False
        assert result["diff_summary"]["added_lines"] == 0

    def test_diff_is_valid_unified_diff(self, tmp_path):
        md_file = tmp_path / "test.md"
        md_file.write_text("# Title\n", encoding="utf-8")

        plan = {
            "section_heading": "## New Section",
            "content_lines": ["Line one."],
        }
        result = execute.append_markdown_section(md_file, plan)

        assert result["diff"].startswith("---")
        assert "@@" in result["diff"]
        assert "+## New Section" in result["diff"]


# ---------------------------------------------------------------------------
# replace_markdown_section
# ---------------------------------------------------------------------------


class TestReplaceMarkdownSection:
    def test_replaces_existing_section(self, tmp_path):
        md_file = tmp_path / "README.md"
        md_file.write_text(
            "# Title\n\n## Notes\n\n- Old note 1.\n- Old note 2.\n\n## Footer\n\nEnd.\n",
            encoding="utf-8",
        )
        plan = {
            "section_heading": "## Notes",
            "content_lines": ["New note A.", "New note B.", "New note C."],
        }
        result = execute.replace_markdown_section(md_file, plan)

        assert result["status"] == "success"
        assert result["modified"] is True
        assert result["diff_summary"]["reason"] == "replaced_section"
        assert result["diff_summary"]["changed_lines"] == 3

        after = md_file.read_text(encoding="utf-8")
        assert "- New note A." in after
        assert "Old note 1" not in after
        assert "## Footer" in after

    def test_section_not_found(self, tmp_path):
        md_file = tmp_path / "README.md"
        md_file.write_text("# Title\n\nSome content.\n", encoding="utf-8")
        plan = {"section_heading": "## Nonexistent", "content_lines": ["X"]}
        result = execute.replace_markdown_section(md_file, plan)

        assert result["status"] == "no_change"
        assert "not found" in result["diff_summary"]["reason"]

    def test_replaces_last_section(self, tmp_path):
        md_file = tmp_path / "README.md"
        md_file.write_text(
            "# Title\n\n## Last Section\n\n- Old line.\n",
            encoding="utf-8",
        )
        plan = {"section_heading": "## Last Section", "content_lines": ["Replaced."]}
        result = execute.replace_markdown_section(md_file, plan)

        assert result["status"] == "success"
        assert "- Replaced." in md_file.read_text(encoding="utf-8")


# ---------------------------------------------------------------------------
# insert_before_section
# ---------------------------------------------------------------------------


class TestInsertBeforeSection:
    def test_inserts_before_heading(self, tmp_path):
        md_file = tmp_path / "README.md"
        md_file.write_text(
            "# Title\n\n## Section A\n\nContent A.\n",
            encoding="utf-8",
        )
        plan = {
            "section_heading": "## Section A",
            "content_lines": ["Inserted line 1.", "Inserted line 2."],
        }
        result = execute.insert_before_section(md_file, plan)

        assert result["status"] == "success"
        after = md_file.read_text(encoding="utf-8")
        assert after.index("Inserted line 1") < after.index("## Section A")

    def test_section_not_found(self, tmp_path):
        md_file = tmp_path / "README.md"
        md_file.write_text("# Title\n", encoding="utf-8")
        plan = {"section_heading": "## Missing", "content_lines": ["X"]}
        result = execute.insert_before_section(md_file, plan)

        assert result["status"] == "no_change"


# ---------------------------------------------------------------------------
# update_yaml_frontmatter
# ---------------------------------------------------------------------------


class TestUpdateYamlFrontmatter:
    def test_updates_existing_frontmatter(self, tmp_path):
        md_file = tmp_path / "SKILL.md"
        md_file.write_text(
            "---\ntitle: My Skill\nversion: 1\n---\n\n# Body\n",
            encoding="utf-8",
        )
        plan = {"frontmatter_updates": {"version": 2, "author": "auto"}}
        result = execute.update_yaml_frontmatter(md_file, plan)

        assert result["status"] == "success"
        after = md_file.read_text(encoding="utf-8")
        assert "version: 2" in after
        assert "author: auto" in after
        assert "# Body" in after

    def test_no_frontmatter(self, tmp_path):
        md_file = tmp_path / "SKILL.md"
        md_file.write_text("# No Frontmatter\n\nJust body.\n", encoding="utf-8")
        plan = {"frontmatter_updates": {"title": "test"}}
        result = execute.update_yaml_frontmatter(md_file, plan)

        assert result["status"] == "no_change"
        assert "no frontmatter" in result["diff_summary"]["reason"]

    def test_malformed_frontmatter(self, tmp_path):
        md_file = tmp_path / "SKILL.md"
        md_file.write_text("---\ntitle: broken", encoding="utf-8")
        plan = {"frontmatter_updates": {"title": "fixed"}}
        result = execute.update_yaml_frontmatter(md_file, plan)

        assert result["status"] == "no_change"


# ---------------------------------------------------------------------------
# ACTION_HANDLERS dispatch
# ---------------------------------------------------------------------------


class TestActionDispatch:
    def test_all_actions_registered(self):
        for name in [
            "append_markdown_section",
            "replace_markdown_section",
            "insert_before_section",
            "update_yaml_frontmatter",
        ]:
            assert name in execute.ACTION_HANDLERS

    def test_handlers_are_callable(self):
        for name, handler in execute.ACTION_HANDLERS.items():
            assert callable(handler), f"{name} handler is not callable"

    def test_unknown_action_not_in_table(self):
        assert execute.ACTION_HANDLERS.get("delete_file") is None

#!/usr/bin/env python3
"""Tests for the improvement-executor rollback module."""

from __future__ import annotations

import importlib.util
import json
import sys
from pathlib import Path

import pytest

# repo root for lib.common imports
_REPO_ROOT = Path(__file__).resolve().parents[3]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))
REPO_ROOT = _REPO_ROOT

_rollback_path = REPO_ROOT / "skills" / "improvement-executor" / "scripts" / "rollback.py"
_spec = importlib.util.spec_from_file_location("rollback", _rollback_path)
rollback = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(rollback)


# ---------------------------------------------------------------------------
# rollback_from_receipt
# ---------------------------------------------------------------------------


class TestRollbackFromReceipt:
    """Test receipt-based rollback."""

    def _make_receipt(self, tmp_path, *, decision="revert", execution_path=None):
        """Create a minimal receipt JSON."""
        receipt = {
            "decision": decision,
            "run_id": "run-001",
            "candidate_id": "cand-001",
        }
        if execution_path:
            receipt["truth_anchor"] = {"execution_path": str(execution_path)}
        receipt_path = tmp_path / "receipt.json"
        receipt_path.write_text(json.dumps(receipt), encoding="utf-8")
        return receipt_path

    def _make_execution(self, tmp_path, backup_path, target_path, rollback_pointer="ptr-abc"):
        """Create a minimal execution artifact JSON."""
        execution = {
            "result": {
                "rollback_pointer": {
                    "backup_path": str(backup_path),
                    "target_path": str(target_path),
                    "method": "restore_backup_file",
                },
            },
            "truth_anchor": str(tmp_path / "execution.json"),
        }
        exec_path = tmp_path / "execution.json"
        exec_path.write_text(json.dumps(execution), encoding="utf-8")
        return exec_path

    def test_skip_when_decision_is_keep(self, tmp_path):
        receipt_path = self._make_receipt(tmp_path, decision="keep")
        result = rollback.rollback_from_receipt(receipt_path)
        assert result["status"] == "skipped"
        assert "keep" in result["reason"].lower()

    def test_error_when_no_execution_path(self, tmp_path):
        receipt_path = self._make_receipt(tmp_path, decision="revert")
        result = rollback.rollback_from_receipt(receipt_path)
        assert result["status"] == "error"
        assert "execution" in result["reason"].lower() or "Cannot find" in result["reason"]

    def test_error_when_backup_file_missing(self, tmp_path):
        target = tmp_path / "target.md"
        target.write_text("current content", encoding="utf-8")
        backup = tmp_path / "nonexistent_backup.md"
        exec_path = self._make_execution(tmp_path, backup, target)
        receipt_path = self._make_receipt(tmp_path, decision="revert", execution_path=exec_path)

        result = rollback.rollback_from_receipt(receipt_path)
        assert result["status"] == "error"
        assert "not found" in result["reason"].lower()

    def test_successful_rollback(self, tmp_path):
        target = tmp_path / "target.md"
        target.write_text("modified content", encoding="utf-8")
        backup = tmp_path / "backup.md"
        backup.write_text("original content", encoding="utf-8")
        exec_path = self._make_execution(tmp_path, backup, target)
        receipt_path = self._make_receipt(tmp_path, decision="revert", execution_path=exec_path)

        result = rollback.rollback_from_receipt(receipt_path, dry_run=False)
        assert result["status"] == "success"
        assert result["dry_run"] is False
        # Target should be restored
        assert target.read_text(encoding="utf-8") == "original content"

    def test_rollback_pointer_in_result(self, tmp_path):
        target = tmp_path / "target.md"
        target.write_text("v2", encoding="utf-8")
        backup = tmp_path / "backup.md"
        backup.write_text("v1", encoding="utf-8")
        exec_path = self._make_execution(tmp_path, backup, target)
        receipt_path = self._make_receipt(tmp_path, decision="revert", execution_path=exec_path)

        result = rollback.rollback_from_receipt(receipt_path, dry_run=False)
        assert result["rollback_pointer"] is not None
        assert result["rollback_pointer"]["target_path"] == str(target)


# ---------------------------------------------------------------------------
# rollback_from_backup
# ---------------------------------------------------------------------------


class TestRollbackFromBackup:
    """Test direct backup-based rollback."""

    def test_error_when_backup_missing(self, tmp_path):
        result = rollback.rollback_from_backup(
            str(tmp_path / "missing.bak"),
            str(tmp_path / "target.md"),
        )
        assert result["status"] == "error"
        assert "not found" in result["reason"].lower()

    def test_successful_restore(self, tmp_path):
        backup = tmp_path / "backup.md"
        backup.write_text("original", encoding="utf-8")
        target = tmp_path / "target.md"
        target.write_text("changed", encoding="utf-8")

        result = rollback.rollback_from_backup(str(backup), str(target), dry_run=False)
        assert result["status"] == "success"
        assert target.read_text(encoding="utf-8") == "original"
        assert result["restored_from"] == str(backup)

    def test_creates_pre_rollback_backup(self, tmp_path):
        backup = tmp_path / "backup.md"
        backup.write_text("original", encoding="utf-8")
        target = tmp_path / "target.md"
        target.write_text("changed", encoding="utf-8")

        result = rollback.rollback_from_backup(str(backup), str(target), dry_run=False)
        assert "pre_rollback_backup" in result
        pre_rb = Path(result["pre_rollback_backup"])
        assert pre_rb.exists()
        assert pre_rb.read_text(encoding="utf-8") == "changed"


# ---------------------------------------------------------------------------
# Dry-run mode
# ---------------------------------------------------------------------------


class TestDryRun:
    """Dry-run should report success without modifying files."""

    def test_dry_run_does_not_modify_target(self, tmp_path):
        backup = tmp_path / "backup.md"
        backup.write_text("original", encoding="utf-8")
        target = tmp_path / "target.md"
        target.write_text("modified", encoding="utf-8")

        result = rollback.rollback_from_backup(str(backup), str(target), dry_run=True)
        assert result["status"] == "dry_run"
        assert result["dry_run"] is True
        # Target must be unchanged
        assert target.read_text(encoding="utf-8") == "modified"
        # No pre-rollback backup should be created
        assert "pre_rollback_backup" not in result

    def test_dry_run_receipt_does_not_modify_target(self, tmp_path):
        target = tmp_path / "target.md"
        target.write_text("v2", encoding="utf-8")
        backup = tmp_path / "backup.md"
        backup.write_text("v1", encoding="utf-8")

        execution = {
            "result": {
                "rollback_pointer": {
                    "backup_path": str(backup),
                    "target_path": str(target),
                    "method": "restore_backup_file",
                },
            },
            "truth_anchor": str(tmp_path / "exec.json"),
        }
        exec_path = tmp_path / "exec.json"
        exec_path.write_text(json.dumps(execution), encoding="utf-8")

        receipt = {
            "decision": "revert",
            "truth_anchor": {"execution_path": str(exec_path)},
        }
        receipt_path = tmp_path / "receipt.json"
        receipt_path.write_text(json.dumps(receipt), encoding="utf-8")

        result = rollback.rollback_from_receipt(receipt_path, dry_run=True)
        assert result["status"] == "dry_run"
        assert result["dry_run"] is True
        assert target.read_text(encoding="utf-8") == "v2"