"""Deterministic local planning and automation runtime for Gate 10B."""

from __future__ import annotations

from dataclasses import dataclass, field
from difflib import unified_diff
import json
from pathlib import Path
import re
from typing import Iterable

from .ledger import stable_hash


GATE10B_APPROVAL_TOKEN = "APPROVE_GATE10B_LOCAL_AUTOMATION"

_SIDE_EFFECT_ACTIONS = {"WRITE_LOCAL_BRIEF", "APPLY_PATCH_IN_ISOLATED_COPY", "WRITE_LOCAL_SCRIPT", "PERSIST_LOCAL_UPDATE"}
_NO_SIDE_EFFECT_ACTIONS = {
    "INSPECT_LOCAL_INPUTS",
    "EXTRACT_RELEVANT_CONTEXT",
    "PREPARE_LOCAL_VERIFIER_PLAN",
    "DRAFT_LOCAL_AUTOMATION_PROPOSAL",
}
_ALLOWED_ACTIONS = _NO_SIDE_EFFECT_ACTIONS | _SIDE_EFFECT_ACTIONS
_SAFE_TARGET_RE = re.compile(r"^[A-Za-z0-9._/-]+$")


def _stable_id(prefix: str, *parts: str) -> str:
    return f"{prefix}_{stable_hash('::'.join(parts))[:12]}"


def _normalize_relative_target(target_path: str) -> str:
    raw = target_path.strip()
    if not raw:
        raise ValueError("empty target path is not allowed")
    if raw.startswith(("/", "\\")):
        raise ValueError("absolute targets are not allowed")
    if raw.startswith("//") or raw.startswith("\\\\"):
        raise ValueError("UNC-style targets are not allowed")
    if "\\" in raw:
        raise ValueError("windows separators are not allowed")
    if "//" in raw:
        raise ValueError("empty target path components are not allowed")
    if ":" in raw:
        raise ValueError("drive paths are not allowed")
    if not _SAFE_TARGET_RE.fullmatch(raw):
        raise ValueError("unsafe characters are not allowed in target paths")
    normalized = Path(raw).as_posix()
    parts = normalized.split("/")
    if any(part in {"", ".", ".."} for part in parts):
        raise ValueError("path traversal is not allowed")
    return "/".join(parts)


def _path_is_relative_to(path: Path, root: Path) -> bool:
    try:
        path.relative_to(root)
    except ValueError:
        return False
    return True


def _resolve_state_dir_target(root: Path, target_path: str) -> Path:
    normalized = _normalize_relative_target(target_path)
    root_resolved = root.resolve()
    candidate = (root_resolved / normalized).resolve()
    if not _path_is_relative_to(candidate, root_resolved):
        raise ValueError("resolved target escapes the state directory workspace")
    candidate.parent.mkdir(parents=True, exist_ok=True)
    return candidate


def _artifact_json(path: Path, payload: dict[str, object]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")


def _write_text(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")


def _render_brief_content(plan: AutomationPlan, step: PlanStep) -> str:
    return (
        "# Local Brief\n\n"
        f"- action: {step.action_type}\n"
        f"- plan_id: {plan.plan_id}\n"
        f"- task_id: {plan.task_id}\n"
        f"- target_path: {step.target_path}\n\n"
        "## Scope\n"
        f"This deterministic brief addresses: {plan.request}.\n"
        "It was written only inside the Gate 10B state-dir workspace.\n"
    )


def _render_script_stub_content(plan: AutomationPlan, step: PlanStep) -> str:
    return (
        "# Local Script Stub\n\n"
        f"- action: {step.action_type}\n"
        f"- plan_id: {plan.plan_id}\n"
        f"- target_path: {step.target_path}\n\n"
        "Checklist:\n"
        "1. Review the local target scope.\n"
        "2. Confirm the change remains inside the state-dir workspace.\n"
        f"3. Record the deterministic follow-up for: {plan.request}.\n"
    )


def _render_update_record_content(plan: AutomationPlan, step: PlanStep) -> str:
    return (
        "# Local Update Record\n\n"
        f"- action: {step.action_type}\n"
        f"- task_id: {plan.task_id}\n"
        f"- target_path: {step.target_path}\n"
        "- status: approved-local-only\n\n"
        f"Update summary: {plan.request}.\n"
    )


def _apply_isolated_patch(plan: AutomationPlan, workspace_root: Path, step: PlanStep, artifact_path: Path) -> tuple[tuple[str, ...], dict[str, object]]:
    fixture_path = _resolve_state_dir_target(workspace_root / "isolated_copy", step.target_path)
    before_path = fixture_path.with_suffix(f"{fixture_path.suffix}.before")
    diff_path = artifact_path.with_suffix(".diff")
    original_lines = [
        "# Isolated Fixture",
        f"task_id: {plan.task_id}",
        "status: draft",
        f"request: {plan.request}",
    ]
    patched_lines = [
        "# Isolated Fixture",
        f"task_id: {plan.task_id}",
        "status: approved-local-only",
        f"request: {plan.request}",
        "patch_note: deterministic bounded patch applied inside state_dir",
    ]
    _write_text(before_path, "\n".join(original_lines) + "\n")
    _write_text(fixture_path, "\n".join(patched_lines) + "\n")
    diff_text = "".join(
        unified_diff(
            [f"{line}\n" for line in original_lines],
            [f"{line}\n" for line in patched_lines],
            fromfile=str(before_path.name),
            tofile=str(fixture_path.name),
        )
    )
    _write_text(diff_path, diff_text)
    return (
        (str(fixture_path), str(before_path), str(diff_path)),
        {
            "written_path": str(fixture_path),
            "before_path": str(before_path),
            "diff_path": str(diff_path),
            "diff_preview": diff_text,
        },
    )


def _execute_approved_side_effect(
    plan: AutomationPlan,
    workspace_root: Path,
    step: PlanStep,
    artifact_path: Path,
) -> tuple[tuple[str, ...], dict[str, object]]:
    if step.action_type == "WRITE_LOCAL_BRIEF":
        output_path = _resolve_state_dir_target(workspace_root, step.target_path)
        _write_text(output_path, _render_brief_content(plan, step))
        return ((str(output_path),), {"written_path": str(output_path), "content_kind": "brief"})
    if step.action_type == "WRITE_LOCAL_SCRIPT":
        output_path = _resolve_state_dir_target(workspace_root, step.target_path)
        _write_text(output_path, _render_script_stub_content(plan, step))
        return ((str(output_path),), {"written_path": str(output_path), "content_kind": "script_stub"})
    if step.action_type == "PERSIST_LOCAL_UPDATE":
        output_path = _resolve_state_dir_target(workspace_root, step.target_path)
        _write_text(output_path, _render_update_record_content(plan, step))
        return ((str(output_path),), {"written_path": str(output_path), "content_kind": "update_record"})
    if step.action_type == "APPLY_PATCH_IN_ISOLATED_COPY":
        return _apply_isolated_patch(plan, workspace_root, step, artifact_path)
    raise ValueError(f"unsupported action type: {step.action_type}")


@dataclass(frozen=True, slots=True)
class PlanningTask:
    task_id: str
    category: str
    request: str
    target_path: str
    expected_side_effect_action: str

    def to_dict(self) -> dict[str, object]:
        return {
            "task_id": self.task_id,
            "category": self.category,
            "request": self.request,
            "target_path": self.target_path,
            "expected_side_effect_action": self.expected_side_effect_action,
        }

    @classmethod
    def from_dict(cls, data: dict[str, object]) -> "PlanningTask":
        return cls(
            task_id=str(data["task_id"]),
            category=str(data["category"]),
            request=str(data["request"]),
            target_path=str(data["target_path"]),
            expected_side_effect_action=str(data["expected_side_effect_action"]),
        )


@dataclass(frozen=True, slots=True)
class PlanStep:
    step_id: str
    title: str
    action_type: str
    detail: str
    target_path: str
    requires_approval: bool
    side_effect: bool
    local_only: bool = True
    state_dir_only_artifact: bool = False
    no_network_access: bool = True
    no_background_daemon: bool = True
    no_unrestricted_shell: bool = True
    no_main_checkout_mutation: bool = True

    def to_dict(self) -> dict[str, object]:
        return {
            "step_id": self.step_id,
            "title": self.title,
            "action_type": self.action_type,
            "detail": self.detail,
            "target_path": self.target_path,
            "requires_approval": self.requires_approval,
            "side_effect": self.side_effect,
            "local_only": self.local_only,
            "state_dir_only_artifact": self.state_dir_only_artifact,
            "no_network_access": self.no_network_access,
            "no_background_daemon": self.no_background_daemon,
            "no_unrestricted_shell": self.no_unrestricted_shell,
            "no_main_checkout_mutation": self.no_main_checkout_mutation,
        }

    @classmethod
    def from_dict(cls, data: dict[str, object]) -> "PlanStep":
        return cls(
            step_id=str(data["step_id"]),
            title=str(data["title"]),
            action_type=str(data["action_type"]),
            detail=str(data["detail"]),
            target_path=str(data["target_path"]),
            requires_approval=bool(data["requires_approval"]),
            side_effect=bool(data["side_effect"]),
            local_only=bool(data.get("local_only", True)),
            state_dir_only_artifact=bool(data.get("state_dir_only_artifact", False)),
            no_network_access=bool(data.get("no_network_access", True)),
            no_background_daemon=bool(data.get("no_background_daemon", True)),
            no_unrestricted_shell=bool(data.get("no_unrestricted_shell", True)),
            no_main_checkout_mutation=bool(data.get("no_main_checkout_mutation", True)),
        )


@dataclass(frozen=True, slots=True)
class AutomationPlan:
    plan_id: str
    task_id: str
    category: str
    request: str
    summary: str
    decomposition: tuple[str, ...]
    steps: tuple[PlanStep, ...]
    dry_run_default: bool = True
    approval_required_for_side_effects: bool = True
    local_only: bool = True
    no_network_access: bool = True
    no_background_daemon: bool = True
    no_unrestricted_shell: bool = True
    no_main_checkout_mutation: bool = True
    no_learned_runtime_authority: bool = True
    no_learned_answer_authority: bool = True
    no_modrux_demo_expansion: bool = True

    def to_dict(self) -> dict[str, object]:
        return {
            "plan_id": self.plan_id,
            "task_id": self.task_id,
            "category": self.category,
            "request": self.request,
            "summary": self.summary,
            "decomposition": list(self.decomposition),
            "steps": [step.to_dict() for step in self.steps],
            "dry_run_default": self.dry_run_default,
            "approval_required_for_side_effects": self.approval_required_for_side_effects,
            "local_only": self.local_only,
            "no_network_access": self.no_network_access,
            "no_background_daemon": self.no_background_daemon,
            "no_unrestricted_shell": self.no_unrestricted_shell,
            "no_main_checkout_mutation": self.no_main_checkout_mutation,
            "no_learned_runtime_authority": self.no_learned_runtime_authority,
            "no_learned_answer_authority": self.no_learned_answer_authority,
            "no_modrux_demo_expansion": self.no_modrux_demo_expansion,
        }

    @classmethod
    def from_dict(cls, data: dict[str, object]) -> "AutomationPlan":
        return cls(
            plan_id=str(data["plan_id"]),
            task_id=str(data["task_id"]),
            category=str(data["category"]),
            request=str(data["request"]),
            summary=str(data["summary"]),
            decomposition=tuple(str(item) for item in data.get("decomposition", [])),
            steps=tuple(PlanStep.from_dict(dict(item)) for item in data.get("steps", [])),
            dry_run_default=bool(data.get("dry_run_default", True)),
            approval_required_for_side_effects=bool(data.get("approval_required_for_side_effects", True)),
            local_only=bool(data.get("local_only", True)),
            no_network_access=bool(data.get("no_network_access", True)),
            no_background_daemon=bool(data.get("no_background_daemon", True)),
            no_unrestricted_shell=bool(data.get("no_unrestricted_shell", True)),
            no_main_checkout_mutation=bool(data.get("no_main_checkout_mutation", True)),
            no_learned_runtime_authority=bool(data.get("no_learned_runtime_authority", True)),
            no_learned_answer_authority=bool(data.get("no_learned_answer_authority", True)),
            no_modrux_demo_expansion=bool(data.get("no_modrux_demo_expansion", True)),
        )


@dataclass(frozen=True, slots=True)
class StepExecutionRecord:
    step_id: str
    action_type: str
    dry_run: bool
    executed: bool
    simulated: bool
    requires_approval: bool
    approval_granted: bool
    side_effect: bool
    side_effect_executed: bool
    artifact_path: str
    output_paths: tuple[str, ...] = ()
    execution_evidence: dict[str, object] = field(default_factory=dict)
    refusal_reason: str = ""

    def to_dict(self) -> dict[str, object]:
        return {
            "step_id": self.step_id,
            "action_type": self.action_type,
            "dry_run": self.dry_run,
            "executed": self.executed,
            "simulated": self.simulated,
            "requires_approval": self.requires_approval,
            "approval_granted": self.approval_granted,
            "side_effect": self.side_effect,
            "side_effect_executed": self.side_effect_executed,
            "artifact_path": self.artifact_path,
            "output_paths": list(self.output_paths),
            "execution_evidence": dict(self.execution_evidence),
            "refusal_reason": self.refusal_reason,
        }

    @classmethod
    def from_dict(cls, data: dict[str, object]) -> "StepExecutionRecord":
        return cls(
            step_id=str(data["step_id"]),
            action_type=str(data["action_type"]),
            dry_run=bool(data["dry_run"]),
            executed=bool(data["executed"]),
            simulated=bool(data["simulated"]),
            requires_approval=bool(data["requires_approval"]),
            approval_granted=bool(data["approval_granted"]),
            side_effect=bool(data["side_effect"]),
            side_effect_executed=bool(data["side_effect_executed"]),
            artifact_path=str(data["artifact_path"]),
            output_paths=tuple(str(item) for item in data.get("output_paths", [])),
            execution_evidence={str(key): value for key, value in dict(data.get("execution_evidence", {})).items()},
            refusal_reason=str(data.get("refusal_reason", "")),
        )


@dataclass(frozen=True, slots=True)
class PlanExecutionResult:
    plan_id: str
    task_id: str
    dry_run: bool
    approval_required_for_side_effects: bool
    step_records: tuple[StepExecutionRecord, ...]
    side_effect_count: int
    blocked_side_effect_count: int

    def to_dict(self) -> dict[str, object]:
        return {
            "plan_id": self.plan_id,
            "task_id": self.task_id,
            "dry_run": self.dry_run,
            "approval_required_for_side_effects": self.approval_required_for_side_effects,
            "step_records": [record.to_dict() for record in self.step_records],
            "side_effect_count": self.side_effect_count,
            "blocked_side_effect_count": self.blocked_side_effect_count,
        }

    @classmethod
    def from_dict(cls, data: dict[str, object]) -> "PlanExecutionResult":
        return cls(
            plan_id=str(data["plan_id"]),
            task_id=str(data["task_id"]),
            dry_run=bool(data["dry_run"]),
            approval_required_for_side_effects=bool(data["approval_required_for_side_effects"]),
            step_records=tuple(StepExecutionRecord.from_dict(dict(item)) for item in data.get("step_records", [])),
            side_effect_count=int(data["side_effect_count"]),
            blocked_side_effect_count=int(data["blocked_side_effect_count"]),
        )


def _category_family(category: str) -> tuple[str, str]:
    families = {
        "document_brief": ("docs", "WRITE_LOCAL_BRIEF"),
        "repo_patch": ("repo", "APPLY_PATCH_IN_ISOLATED_COPY"),
        "workspace_cleanup": ("workspace", "WRITE_LOCAL_SCRIPT"),
        "memory_skill_review": ("memory", "PERSIST_LOCAL_UPDATE"),
    }
    return families[category]


def build_gate10b_tasks() -> list[PlanningTask]:
    tasks: list[PlanningTask] = []
    families = {
        "document_brief": [
            "summarize onboarding notes",
            "extract release checklist evidence",
            "prepare a local architecture brief",
            "draft a regression summary",
            "consolidate verifier findings",
            "prepare a local incident recap",
            "compile repo task observations",
            "summarize doc research findings",
            "draft a local validation memo",
            "prepare a maintenance brief",
        ],
        "repo_patch": [
            "fix a missing import in a local service",
            "propose a bounded patch for a flaky test fixture",
            "draft a local-only config default repair",
            "prepare a repo patch for a wrong symbol export",
            "inspect a data shape mismatch and draft a patch",
            "prepare a verifier-backed documentation patch",
            "draft a safe refactor proposal for a helper module",
            "inspect a repo regression and propose a patch",
            "prepare a bounded test repair plan",
            "draft a local fixture patch for a parsing bug",
        ],
        "workspace_cleanup": [
            "prepare a cleanup script for stale local artifacts",
            "draft a local-only archive rotation plan",
            "prepare a temp-file cleanup checklist",
            "draft a local report export plan",
            "prepare a deterministic backup script draft",
            "draft a workspace manifest refresh plan",
            "prepare a local diff packaging plan",
            "draft a fixture reset script",
            "prepare a log compaction script draft",
            "draft a local-only notes export",
        ],
        "memory_skill_review": [
            "prepare a memory review packet for repeated failures",
            "draft a local skill evidence update",
            "prepare a memory invalidation review",
            "draft a skill quarantine review note",
            "prepare a local pattern library update",
            "draft a reusable checklist update",
            "prepare a memory coverage review",
            "draft a skill freshness audit note",
            "prepare a local-only replay review",
            "draft a deterministic evidence bundle update",
        ],
    }
    for category, requests in families.items():
        family_root, side_effect_action = _category_family(category)
        for index, request in enumerate(requests, start=1):
            target_path = f"{family_root}/{category}_{index:02d}.md"
            task_id = _stable_id("gate10b_task", category, str(index), request)
            tasks.append(
                PlanningTask(
                    task_id=task_id,
                    category=category,
                    request=request,
                    target_path=target_path,
                    expected_side_effect_action=side_effect_action,
                )
            )
    return tasks


def build_automation_plan(task: PlanningTask) -> AutomationPlan:
    target_path = _normalize_relative_target(task.target_path)
    side_effect_action = task.expected_side_effect_action
    if side_effect_action not in _SIDE_EFFECT_ACTIONS:
        raise ValueError(f"unsupported side-effect action: {side_effect_action}")
    summary = f"Plan a local-only automation workflow for {task.request}."
    decomposition = (
        "Inspect local inputs relevant to the task.",
        "Extract only task-local evidence needed for planning.",
        "Draft a bounded local-only automation proposal.",
        "Require explicit approval before any side-effectful step.",
    )
    steps = (
        PlanStep(
            step_id=_stable_id("step", task.task_id, "inspect"),
            title="Inspect local inputs",
            action_type="INSPECT_LOCAL_INPUTS",
            detail=f"Inspect only the local inputs needed to {task.request}.",
            target_path=target_path,
            requires_approval=False,
            side_effect=False,
        ),
        PlanStep(
            step_id=_stable_id("step", task.task_id, "extract"),
            title="Extract relevant context",
            action_type="EXTRACT_RELEVANT_CONTEXT",
            detail="Extract deterministic task-local context for the plan.",
            target_path=target_path,
            requires_approval=False,
            side_effect=False,
        ),
        PlanStep(
            step_id=_stable_id("step", task.task_id, "proposal"),
            title="Draft local automation proposal",
            action_type="DRAFT_LOCAL_AUTOMATION_PROPOSAL",
            detail="Draft the local-only automation proposal and bounded verifier checklist.",
            target_path=target_path,
            requires_approval=False,
            side_effect=False,
            state_dir_only_artifact=True,
        ),
        PlanStep(
            step_id=_stable_id("step", task.task_id, "effect"),
            title="Approval-gated local automation step",
            action_type=side_effect_action,
            detail="Execute only after explicit approval and only within the bounded local target.",
            target_path=target_path,
            requires_approval=True,
            side_effect=True,
        ),
    )
    return AutomationPlan(
        plan_id=_stable_id("plan", task.task_id, task.category, task.request),
        task_id=task.task_id,
        category=task.category,
        request=task.request,
        summary=summary,
        decomposition=decomposition,
        steps=steps,
    )


def validate_plan_schema(plan: AutomationPlan) -> list[str]:
    errors: list[str] = []
    if not plan.plan_id or not plan.task_id or not plan.summary:
        errors.append("missing required plan metadata")
    if len(plan.decomposition) < 4:
        errors.append("task decomposition is incomplete")
    if len(plan.steps) < 4:
        errors.append("plan must contain at least four steps")
    step_ids = [step.step_id for step in plan.steps]
    if len(step_ids) != len(set(step_ids)):
        errors.append("step IDs must be unique")
    if not plan.dry_run_default:
        errors.append("dry-run must be the default mode")
    if not plan.approval_required_for_side_effects:
        errors.append("approval gating must be enabled")
    safety_flags = (
        plan.local_only,
        plan.no_network_access,
        plan.no_background_daemon,
        plan.no_unrestricted_shell,
        plan.no_main_checkout_mutation,
        plan.no_learned_runtime_authority,
        plan.no_learned_answer_authority,
        plan.no_modrux_demo_expansion,
    )
    if not all(safety_flags):
        errors.append("plan safety flags must remain enabled")
    for step in plan.steps:
        if step.action_type not in _ALLOWED_ACTIONS:
            errors.append(f"unsupported action type: {step.action_type}")
        try:
            _normalize_relative_target(step.target_path)
        except ValueError as exc:
            errors.append(f"unsafe target path for {step.step_id}: {exc}")
        if step.side_effect and not step.requires_approval:
            errors.append(f"side-effect step {step.step_id} must require approval")
        if step.action_type in _NO_SIDE_EFFECT_ACTIONS and step.side_effect:
            errors.append(f"non-side-effect action {step.step_id} marked as side effect")
        if step.action_type in _SIDE_EFFECT_ACTIONS and not step.side_effect:
            errors.append(f"side-effect action {step.step_id} missing side-effect flag")
        step_safety = (
            step.local_only,
            step.no_network_access,
            step.no_background_daemon,
            step.no_unrestricted_shell,
        )
        if not all(step_safety):
            errors.append(f"step {step.step_id} violates local-only safety flags")
    return errors


def execute_automation_plan(
    plan: AutomationPlan,
    state_dir: Path,
    *,
    dry_run: bool = True,
    approve_side_effects: bool = False,
    approval_token: str = "",
) -> PlanExecutionResult:
    records: list[StepExecutionRecord] = []
    execution_mode = "dry_run" if dry_run else ("approved" if approve_side_effects else "unapproved")
    execution_dir = state_dir / "planning_runtime" / "artifacts" / plan.plan_id / execution_mode
    workspace_root = state_dir / "planning_runtime" / "workspace" / plan.plan_id
    execution_dir.mkdir(parents=True, exist_ok=True)
    side_effect_count = 0
    blocked_side_effect_count = 0
    approval_valid = approve_side_effects and approval_token == GATE10B_APPROVAL_TOKEN
    for step in plan.steps:
        artifact_path = execution_dir / f"{step.step_id}.json"
        executed = not step.side_effect or (not dry_run and approval_valid and step.requires_approval)
        simulated = dry_run or (step.side_effect and not executed)
        side_effect_executed = bool(step.side_effect and executed and not dry_run)
        output_paths: tuple[str, ...] = ()
        execution_evidence: dict[str, object] = {}
        refusal_reason = ""
        if step.side_effect and not side_effect_executed:
            blocked_side_effect_count += 1
            if dry_run:
                refusal_reason = "dry-run mode blocks side effects"
            elif not approval_valid:
                refusal_reason = "explicit approval token required"
        if side_effect_executed:
            side_effect_count += 1
            workspace_root.mkdir(parents=True, exist_ok=True)
            output_paths, execution_evidence = _execute_approved_side_effect(plan, workspace_root, step, artifact_path)
        else:
            execution_evidence = {"state_dir_workspace_root": str(workspace_root)}
        _artifact_json(
            artifact_path,
            {
                "step_id": step.step_id,
                "action_type": step.action_type,
                "target_path": step.target_path,
                "dry_run": dry_run,
                "executed": executed,
                "simulated": simulated,
                "approval_granted": approval_valid,
                "side_effect_executed": side_effect_executed,
                "written_paths": list(output_paths),
                "execution_evidence": execution_evidence,
                "refusal_reason": refusal_reason,
            },
        )
        records.append(
            StepExecutionRecord(
                step_id=step.step_id,
                action_type=step.action_type,
                dry_run=dry_run,
                executed=executed,
                simulated=simulated,
                requires_approval=step.requires_approval,
                approval_granted=approval_valid,
                side_effect=step.side_effect,
                side_effect_executed=side_effect_executed,
                artifact_path=str(artifact_path),
                output_paths=output_paths,
                execution_evidence=execution_evidence,
                refusal_reason=refusal_reason,
            )
        )
    return PlanExecutionResult(
        plan_id=plan.plan_id,
        task_id=plan.task_id,
        dry_run=dry_run,
        approval_required_for_side_effects=plan.approval_required_for_side_effects,
        step_records=tuple(records),
        side_effect_count=side_effect_count,
        blocked_side_effect_count=blocked_side_effect_count,
    )


def plan_has_local_only_proposal(plan: AutomationPlan) -> bool:
    return (
        plan.local_only
        and plan.no_network_access
        and plan.no_background_daemon
        and plan.no_unrestricted_shell
        and plan.no_learned_runtime_authority
        and any(step.action_type == "DRAFT_LOCAL_AUTOMATION_PROPOSAL" for step in plan.steps)
    )


def approval_required_for_all_side_effects(plan: AutomationPlan) -> bool:
    side_effect_steps = [step for step in plan.steps if step.side_effect]
    return bool(side_effect_steps) and all(step.requires_approval for step in side_effect_steps)


def write_jsonl(path: Path, rows: Iterable[dict[str, object]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8") as handle:
        for row in rows:
            handle.write(json.dumps(row, sort_keys=True) + "\n")
