# tools/file_ops.py - File read/write/delete/metadata and text chunking.
# All path resolution via utils.path_safety (safe_resolve_path, is_dangerous_path).
from __future__ import annotations

import contextvars
import json
import math
import os
import re
import mimetypes
import shutil
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from datetime import datetime
from typing import Any

from langchain_core.tools import tool
from tools._logging import _debug_log, _emit_input_status
from tools.live_view_visuals import emit_live_view_preview, image_to_preview_base64
from utils.path_safety import (
    safe_resolve_path,
    is_dangerous_path,
    is_path_allowed,
    is_plugins_dir_write_blocked,
    is_agent_tool_blocked,
    get_user_output_dir,
    normalize_likely_mugi_temp_upload_path,
)
from utils.text_io import read_text_file_with_encoding_fallback
from utils.user_model import (
    apply_patch as _user_model_apply_patch,
    create_proposal_detailed,
    list_proposals,
)

# Kanban HTML composition workers need enough preview to patch GSAP/Hyperframes markup.
_KANBAN_HTML_READ_PREVIEW_CHARS = 200_000
_kanban_expand_html_read_preview: contextvars.ContextVar[bool] = contextvars.ContextVar(
    "kanban_expand_html_read_preview",
    default=False,
)


def set_kanban_expand_html_read_preview(enabled: bool = True) -> contextvars.Token[bool]:
    return _kanban_expand_html_read_preview.set(bool(enabled))


def reset_kanban_expand_html_read_preview(token: contextvars.Token[bool]) -> None:
    _kanban_expand_html_read_preview.reset(token)


def _read_file_preview_limit(ext: str) -> int:
    if not _kanban_expand_html_read_preview.get():
        return 5000
    if ext in (".html", ".htm", ".css", ".js", ".json"):
        return _KANBAN_HTML_READ_PREVIEW_CHARS
    return 5000

_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff", ".tif"}

_DEFAULT_WORKSPACE = os.path.expanduser("~/.mugi/workspace")
FILE_CHUNK_SIZE = 3500

# Workspace-root files treated as evolution/personality targets for write safety (keep in sync with commit_workspace_evolution).
_EVOLUTION_COMMIT_WHITELIST = (
    "CONSTITUTION.md",
    "PERSONA.md",
    "IDENTITY.md",
    "SOUL.md",
    "USER.md",
    "MEMORY.md",
    "JOURNAL.md",
    "HEARTBEAT.md",
    "TOOLS.md",
    "AGENTS.md",
    "SESSION-STATE.md",
)
_EVOLUTION_SHARD_GITADD_CAP = 200
_EVOLUTION_SKILLS_GITADD_CAP = 200


def _evolution_shard_rel_paths(workspace: str) -> list[str]:
    """Bounded list of memory/shards/*.md for evolution commits (validated)."""
    try:
        from utils.memory_shard_layout import list_shard_rel_paths

        rels = list_shard_rel_paths(workspace)
    except Exception:
        return []
    out: list[str] = []
    for r in rels[:_EVOLUTION_SHARD_GITADD_CAP]:
        if not r.startswith("memory/shards/") or not r.endswith(".md"):
            continue
        if ".." in r.replace("\\", "/"):
            continue
        ap = os.path.join(workspace, r.replace("/", os.sep))
        if os.path.isfile(ap) and not os.path.islink(ap):
            out.append(r.replace("\\", "/"))
    return out


def _is_staged_evolution_memory_shard(p: str) -> bool:
    q = (p or "").replace("\\", "/").strip()
    if not q.startswith("memory/shards/") or not q.endswith(".md"):
        return False
    rest = q[len("memory/shards/") :]
    if not rest or ".." in rest or "/" in rest:
        return False
    return True


def _evolution_skill_markdown_rel_paths(workspace: str) -> list[str]:
    """Bounded list of skills/**/*.md (excluding dot/archived dirs)."""
    skills_root = os.path.join(workspace, "skills")
    if not os.path.isdir(skills_root):
        return []
    out: list[str] = []
    for root, dirs, files in os.walk(skills_root):
        dirs[:] = [d for d in dirs if not d.startswith(".")]
        rel_root = os.path.relpath(root, workspace).replace("\\", "/")
        if rel_root == "skills/.archive" or rel_root.startswith("skills/.archive/"):
            dirs[:] = []
            continue
        for name in files:
            if not name.lower().endswith(".md"):
                continue
            abs_path = os.path.join(root, name)
            if not os.path.isfile(abs_path) or os.path.islink(abs_path):
                continue
            rel = os.path.relpath(abs_path, workspace).replace("\\", "/")
            if ".." in rel or not rel.startswith("skills/"):
                continue
            out.append(rel)
            if len(out) >= _EVOLUTION_SKILLS_GITADD_CAP:
                return out
    return out


def _is_staged_evolution_skill_markdown(p: str) -> bool:
    q = (p or "").replace("\\", "/").strip()
    if not q.startswith("skills/") or not q.lower().endswith(".md"):
        return False
    if "/." in q:
        return False
    if q.startswith("skills/.archive/"):
        return False
    return ".." not in q


def _configured_workspace_dir() -> str:
    try:
        from core.config import get_workspace_path

        return os.path.abspath(os.path.expanduser(get_workspace_path()))
    except Exception:
        return os.path.abspath(os.path.expanduser(_DEFAULT_WORKSPACE))


def _is_evolution_protected_workspace_target(workspace_abs: str, path_abs: str) -> bool:
    w = os.path.abspath(workspace_abs)
    p = os.path.abspath(path_abs)
    if not p.startswith(w + os.sep) and p != w:
        return False
    return os.path.basename(p) in _EVOLUTION_COMMIT_WHITELIST


def _content_smells_like_truncated_read_file_paste(content: str) -> bool:
    if not content or len(content) < 50:
        return False
    c = content
    if "⚠️ TRUNCATED:" in c:
        return True
    if "Do not save this truncated content back" in c:
        return True
    if "File content (first 5000 chars):" in c:
        return True
    if "Only the first 5000 are shown below" in c:
        return True
    return False


def _restricted_background_writes() -> bool:
    try:
        from tool_policy import is_background_personal_proposal_context

        return is_background_personal_proposal_context()
    except Exception:
        return False


def _direct_personal_write_blocked() -> bool:
    try:
        from tool_policy import is_non_foreground_personal_context

        return is_non_foreground_personal_context()
    except Exception:
        return False


def _is_under_mugi_tools_dir(path_abs: str) -> bool:
    tools = os.path.abspath(os.path.expanduser("~/.mugi/tools"))
    p = os.path.abspath(path_abs)
    return p == tools or p.startswith(tools + os.sep)


def _agent_deletable_roots() -> tuple[str, ...]:
    """Directories where delete_file may remove files (workspace + user-visible output)."""
    roots: list[str] = [os.path.abspath(os.path.expanduser(_DEFAULT_WORKSPACE))]
    try:
        roots.append(os.path.abspath(get_user_output_dir()))
        roots.append(os.path.abspath(get_ocr_output_dir()))
    except Exception:
        pass
    # De-dupe preserving order
    seen: set[str] = set()
    out: list[str] = []
    for r in roots:
        if r not in seen:
            seen.add(r)
            out.append(r)
    return tuple(out)


def _mugi_temp_uploads_root() -> str:
    return os.path.abspath(os.path.expanduser("~/.mugi/temp_uploads"))


def _is_registered_chat_attachment_path(path: str) -> bool:
    """True when *path* is a chat attachment staged under ~/.mugi/temp_uploads/."""
    try:
        from attachment_registry import get_record_by_path

        if get_record_by_path(path) is None:
            return False
        root = _mugi_temp_uploads_root()
        real_path = os.path.realpath(os.path.abspath(path))
        return real_path == root or real_path.startswith(root + os.sep)
    except Exception:
        return False


def _path_allowed_for_agent_read(path: str) -> bool:
    if any(is_path_allowed(path, base_dir=root) for root in _agent_deletable_roots()):
        return True
    return _is_registered_chat_attachment_path(path)

# OCR LLM correction: chunk size vs output tokens — keep chunks small enough that the
# model can emit a full corrected region without truncation (copy-edit ≈ similar length).
_DEFAULT_OCR_CORRECTION_CHUNK_CHARS = 5600
_OCR_CORRECTION_WORKERS = max(1, min(8, int(os.environ.get("MUGI_OCR_CORRECTION_WORKERS", "3"))))


def _max_tokens_for_ocr_correction_chunk(chunk: str) -> int:
    """Completion budget for one OCR correction chunk (chars → tokens heuristic)."""
    n = len(chunk or "")
    # ~0.35 tokens/char for English + margin for markdown/footnotes; bounded by backend.
    estimated = int(n * 0.35) + 900
    estimated = max(estimated, 2048)
    try:
        from agent_loop import get_llm_max_tokens

        cap = int(get_llm_max_tokens() or 8192)
    except Exception:
        cap = 8192
    return min(estimated, max(cap, 2048))
_READ_FILE_CHUNK_BATCH_MAX = 8


def _resolve_read(workspace: str, requested: str) -> tuple[str | None, str]:
    """Resolve path for read; return (abs_path, error_msg). Blocks agent-tool-blocked paths."""
    path, err = safe_resolve_path(workspace or _DEFAULT_WORKSPACE, requested, allow_write=False, create_parents=False)
    if err:
        return path, err
    if path and not _path_allowed_for_agent_read(path):
        return None, (
            "Refused: path must be inside ~/.mugi/workspace/ or ~/Documents/Mugi/ (including OCR/), "
            "or be a registered chat attachment. "
            f"Got: {requested}"
        )
    if path and (not os.path.exists(path)):
        repaired = normalize_likely_mugi_temp_upload_path(requested or "")
        if repaired and repaired != (requested or ""):
            repaired_path, repaired_err = safe_resolve_path(
                workspace or _DEFAULT_WORKSPACE,
                repaired,
                allow_write=False,
                create_parents=False,
            )
            if (not repaired_err) and repaired_path and os.path.exists(repaired_path):
                path = repaired_path
    if path and is_agent_tool_blocked(path):
        return None, f"Path is not accessible via file tools: {path}"
    return path, err


def _resolve_write(workspace: str, requested: str) -> tuple[str | None, str]:
    """Resolve path for write; return (abs_path, error_msg). Blocks agent-tool-blocked paths."""
    path, err = safe_resolve_path(workspace or _DEFAULT_WORKSPACE, requested, allow_write=True)
    if err:
        return path, err
    if path and not any(is_path_allowed(path, base_dir=root) for root in _agent_deletable_roots()):
        return None, (
            "Refused: path must be inside ~/.mugi/workspace/ or ~/Documents/Mugi/ (including OCR/). "
            f"Got: {requested}"
        )
    if path and is_agent_tool_blocked(path):
        return None, f"Path is not accessible via file tools: {path}"
    return path, err


@tool
def read_file(filepath: str) -> str:
    """Read a local plain-text file. Returns full content if under 5000 characters; otherwise returns only the first 5000 (so context stays small). For large files use read_file_chunk to read in sections and edit with submit_ocr_chunk_corrections or other tools.

    When multiple tools run in parallel in one model round, do not read a path in the same batch as write_text_file/save_to_file for that path—the read may execute before the write finishes and report "path does not exist". Read in a later round or after the write result is visible."""
    _debug_log(f"[DEBUG read_file] Called with filepath: {filepath!r}")
    _emit_input_status("Reading file…")
    try:
        filepath = filepath.strip(' "\n,')
        abs_path, err = _resolve_read(_DEFAULT_WORKSPACE, filepath)
        if err:
            _debug_log(f"[ERROR read_file] {err}", "ERROR")
            return err
        if not os.path.exists(abs_path):
            from utils.path_suggest import enrich_missing_path_error

            msg = f"Path does not exist: {abs_path}"
            return msg + enrich_missing_path_error(abs_path, workspace=_DEFAULT_WORKSPACE)
        if os.path.isdir(abs_path):
            return f"Path is a directory, not a file: {abs_path}\n\nUse 'list_directory' to list directory contents instead."
        if not os.path.isfile(abs_path):
            return f"Path exists but is not a file: {abs_path}"
        if not os.access(abs_path, os.R_OK):
            return f"File not readable: {abs_path}"
        ext = os.path.splitext(abs_path)[1].lower()
        if ext in _IMAGE_EXTENSIONS:
            try:
                with open(abs_path, "rb") as img_f:
                    b64 = image_to_preview_base64(img_f.read(), max_width=400)
                if b64:
                    emit_live_view_preview(
                        f"Image: {os.path.basename(abs_path)}",
                        tool_name="read_file",
                        preview_image_base64=b64,
                    )
            except Exception:
                pass
        content = read_text_file_with_encoding_fallback(abs_path)
        full_length = len(content)
        # Memory inspector trace: byte size on disk / wire (UTF-8), not Unicode code unit count.
        utf8_byte_len = len(content.encode("utf-8", errors="replace"))
        try:
            from utils.memory_turn_trace import (
                is_memory_inspector_workspace_rel,
                record_memory_path_read_file,
                workspace_relative_path,
            )

            wr = workspace_relative_path(_DEFAULT_WORKSPACE, abs_path)
            if wr and is_memory_inspector_workspace_rel(wr):
                record_memory_path_read_file(
                    path_rel=wr,
                    bytes_read=int(utf8_byte_len),
                    truncated=bool(full_length > _read_file_preview_limit(ext)),
                )
        except Exception:
            pass
        preview_limit = _read_file_preview_limit(ext)
        if full_length > preview_limit:
            content = content[:preview_limit] + "\n... (truncated)"
            next_chunk_no_overlap = max(1, math.ceil(preview_limit / FILE_CHUNK_SIZE))
            return (
                f"⚠️ TRUNCATED: This file has {full_length} characters total. "
                f"Only the first {preview_limit} are shown below.\n\n"
                f"Do not save this truncated content back to the same path or you will overwrite the rest. "
                f"To process the whole file use read_file_chunk (chunk_index is 0-based: 0, 1, 2, ...). "
                f"Chunk boundaries use FILE_CHUNK_SIZE={FILE_CHUNK_SIZE}, so they do not align exactly with this preview. "
                f"For less overlap after this preview, start with chunk_index={next_chunk_no_overlap}; "
                f"use chunk_index={max(0, next_chunk_no_overlap - 1)} if overlap is acceptable.\n\n"
                f"When editing, use submit_ocr_chunk_corrections or save to a different path.\n\n"
                f"File content (first {preview_limit} chars):\n{content}"
            )
        return f"File content:\n{content}"
    except Exception as e:
        _debug_log(f"[ERROR read_file] {e}", "ERROR")
        return f"Read failed: {str(e)}."


@tool
def list_directory(dir_path: str) -> str:
    """List the contents of a directory. Returns filenames, sizes, and types (file/dir).
    Use full or relative paths. Refuses dangerous/protected paths."""
    _emit_input_status("Listing directory…")
    try:
        dir_path = (dir_path or "").strip(' "\n,')
        if not dir_path:
            return "No dir_path provided."
        abs_path, err = _resolve_read(_DEFAULT_WORKSPACE, dir_path)
        if err:
            return err
        if not os.path.exists(abs_path):
            from utils.path_suggest import enrich_missing_path_error

            msg = f"Path does not exist: {abs_path}"
            return msg + enrich_missing_path_error(abs_path, workspace=_DEFAULT_WORKSPACE)
        if not os.path.isdir(abs_path):
            return f"Path is not a directory: {abs_path}\n\nUse 'read_file' to read file contents instead."
        entries = []
        try:
            items = sorted(os.listdir(abs_path))
        except PermissionError:
            return f"Permission denied: {abs_path}"
        for item in items:
            full = os.path.join(abs_path, item)
            try:
                if os.path.isdir(full):
                    entries.append(f"  {item}/")
                else:
                    size = os.path.getsize(full)
                    entries.append(f"  {item}  ({size} bytes)")
            except OSError:
                entries.append(f"  {item}  (inaccessible)")
        header = f"Directory: {abs_path}  ({len(items)} items)\n"
        if not entries:
            return header + "  (empty)"
        return header + "\n".join(entries)
    except Exception as e:
        _debug_log(f"[ERROR list_directory] {e}", "ERROR")
        return f"list_directory failed: {str(e)}"


@tool
def delete_file(path: str) -> str:
    """Delete a file at the given path. Allowed inside ~/.mugi/workspace/ and ~/Documents/Mugi/ (including OCR subfolder)—same roots agents use for write_text_file / charts / downloads. Use for removing BOOTSTRAP.md after onboarding or cleaning test files in Documents/Mugi."""
    _emit_input_status("Deleting file…")
    try:
        path = path.strip(' "\n,')
        abs_path, err = _resolve_read(_DEFAULT_WORKSPACE, path)
        if err:
            return err
        if not any(is_path_allowed(abs_path, base_dir=root) for root in _agent_deletable_roots()):
            return (
                "Refused: path must be inside ~/.mugi/workspace/ or ~/Documents/Mugi/ (including OCR/). "
                f"Got: {path}"
            )
        if is_plugins_dir_write_blocked(abs_path):
            return "Blocked: cannot delete files in the plugins directory. Manage plugins in Settings."
        try:
            from utils.user_model import check_user_model_direct_write_allowed

            _ws = _configured_workspace_dir()
            _um_err = check_user_model_direct_write_allowed(_ws, abs_path, mode="delete")
            if _um_err:
                return _um_err
        except Exception as exc:
            _debug_log(f"[ERROR delete_file user model guard] {exc}", "ERROR")
            return "Refused: user model delete guard failed (fail-closed)."
        if not os.path.exists(abs_path):
            return f"Path does not exist: {abs_path}"
        if os.path.isdir(abs_path):
            return f"Path is a directory; delete_file only removes files: {abs_path}"
        os.remove(abs_path)
        if os.path.basename(abs_path).upper() == "BOOTSTRAP.MD":
            try:
                from core.config import get_config, get_workspace_path
                from persistence import sync_onboarding_done_if_bootstrap_removed

                sync_onboarding_done_if_bootstrap_removed(get_config(), get_workspace_path())
            except Exception:
                _debug_log("[delete_file] onboarding sync after bootstrap delete failed", "WARN")
        return f"Deleted: {abs_path}"
    except OSError as e:
        return f"Delete failed: {e}"


@tool
def copy_file(source_path: str, destination_path: str) -> str:
    """Copy a file from source_path to destination_path. Use full paths or paths relative to ~/.mugi/workspace. If destination_path is a directory, the file is copied into it with the same name. Creates parent directories for the destination if needed."""
    _emit_input_status("Copying file…")
    try:
        source_path = (source_path or "").strip()
        destination_path = (destination_path or "").strip()
        if not source_path:
            return "No source_path provided."
        if not destination_path:
            return "No destination_path provided."
        src_abs, err = _resolve_read(_DEFAULT_WORKSPACE, source_path)
        if err:
            return err
        if not os.path.exists(src_abs):
            return f"Source does not exist: {src_abs}"
        if os.path.isdir(src_abs):
            return f"Source is a directory, not a file: {src_abs}"
        if not os.path.isfile(src_abs):
            return f"Source path exists but is not a file: {src_abs}"
        if is_dangerous_path(src_abs):
            return f"Operation blocked: source path is in a protected location: {src_abs}"
        dest_abs, err = _resolve_write(_DEFAULT_WORKSPACE, destination_path)
        if err:
            return err
        if is_dangerous_path(dest_abs):
            return f"Operation blocked: destination path is in a protected location: {dest_abs}"
        if is_plugins_dir_write_blocked(dest_abs):
            return "Blocked: cannot write to the plugins directory. Use create_plugin (evolution mode) to add plugins."
        if os.path.isdir(dest_abs):
            dest_abs = os.path.join(dest_abs, os.path.basename(src_abs))
        if is_plugins_dir_write_blocked(dest_abs):
            return "Blocked: cannot write to the plugins directory."
        try:
            from utils.user_model import check_user_model_direct_write_allowed

            _ws = _configured_workspace_dir()
            _um_err = check_user_model_direct_write_allowed(_ws, dest_abs, mode="overwrite")
            if _um_err:
                return _um_err
        except Exception as exc:
            _debug_log(f"[ERROR copy_file user model guard] {exc}", "ERROR")
            return "Refused: user model write guard failed (fail-closed)."
        shutil.copy2(src_abs, dest_abs)
        size = os.path.getsize(dest_abs)
        return f"Copied to: {dest_abs} ({size} bytes)"
    except OSError as e:
        _debug_log(f"[ERROR copy_file] {e}", "ERROR")
        return f"Copy failed: {e}"


@tool
def move_file(source_path: str, destination_path: str) -> str:
    """Move a file from source_path to destination_path. Use full paths or paths relative to ~/.mugi/workspace. If destination_path is a directory, the file is moved into it with the same name. Creates parent directories for the destination if needed. The source file is removed after a successful move."""
    _emit_input_status("Moving file…")
    try:
        source_path = (source_path or "").strip()
        destination_path = (destination_path or "").strip()
        if not source_path:
            return "No source_path provided."
        if not destination_path:
            return "No destination_path provided."
        src_abs, err = _resolve_read(_DEFAULT_WORKSPACE, source_path)
        if err:
            return err
        if not os.path.exists(src_abs):
            return f"Source does not exist: {src_abs}"
        if os.path.isdir(src_abs):
            return f"Source is a directory, not a file: {src_abs}"
        if not os.path.isfile(src_abs):
            return f"Source path exists but is not a file: {src_abs}"
        if is_dangerous_path(src_abs):
            return f"Operation blocked: source path is in a protected location: {src_abs}"
        if is_plugins_dir_write_blocked(src_abs):
            return "Blocked: cannot move files from the plugins directory. Manage plugins in Settings."
        try:
            from utils.user_model import check_user_model_direct_write_allowed

            _ws = _configured_workspace_dir()
            _um_err = check_user_model_direct_write_allowed(_ws, src_abs, mode="delete")
            if _um_err:
                return _um_err
        except Exception as exc:
            _debug_log(f"[ERROR move_file user model guard] {exc}", "ERROR")
            return "Refused: user model delete guard failed (fail-closed)."
        dest_abs, err = _resolve_write(_DEFAULT_WORKSPACE, destination_path)
        if err:
            return err
        if is_dangerous_path(dest_abs):
            return f"Operation blocked: destination path is in a protected location: {dest_abs}"
        if is_plugins_dir_write_blocked(dest_abs):
            return "Blocked: cannot write to the plugins directory. Use create_plugin (evolution mode) to add plugins."
        if os.path.isdir(dest_abs):
            dest_abs = os.path.join(dest_abs, os.path.basename(src_abs))
        if is_plugins_dir_write_blocked(dest_abs):
            return "Blocked: cannot write to the plugins directory."
        try:
            from utils.user_model import check_user_model_direct_write_allowed

            _um_err = check_user_model_direct_write_allowed(_ws, dest_abs, mode="overwrite")
            if _um_err:
                return _um_err
        except Exception as exc:
            _debug_log(f"[ERROR move_file user model guard] {exc}", "ERROR")
            return "Refused: user model write guard failed (fail-closed)."
        shutil.move(src_abs, dest_abs)
        size = os.path.getsize(dest_abs)
        return f"Moved to: {dest_abs} ({size} bytes)"
    except OSError as e:
        _debug_log(f"[ERROR move_file] {e}", "ERROR")
        return f"Move failed: {e}"


_READ_FILE_CHUNK_BATCH_MAX = 8


# ---------- split_text_file / merge_text_files ----------
@tool
def split_text_file(input_json: str) -> str:
    """Split a .txt or .md file into multiple smaller files.

    Input must be a JSON string with: "file_path", optional "output_dir", "split_method" ("equal"|"lines"|"headers"), "num_parts" (if equal), "lines_per_file" (if lines).
    Example: split_text_file('{"file_path": "/path/to/notes.md", "split_method": "equal", "num_parts": 3}')
    """
    _debug_log(f"[DEBUG split_text_file] Called with input_json: {input_json!r}")
    _emit_input_status("Splitting file…")
    try:
        if isinstance(input_json, dict):
            data = input_json
        elif isinstance(input_json, str) and input_json.strip():
            data = json.loads(input_json)
        else:
            data = {}
        file_path = (data.get("file_path") or "").strip()
        output_dir = (data.get("output_dir") or "").strip()
        split_method = (data.get("split_method") or "equal").strip().lower()
        num_parts = data.get("num_parts")
        lines_per_file = data.get("lines_per_file")
        if not file_path:
            return "Error: 'file_path' is required."
        abs_path, err = _resolve_read(_DEFAULT_WORKSPACE, file_path)
        if err:
            return err
        if not os.path.exists(abs_path):
            return f"File not found: {abs_path}"
        if os.path.isdir(abs_path):
            return f"Path is a directory, not a file: {abs_path}"
        output_dir_abs = None
        if output_dir:
            output_dir_abs, err = _resolve_write(_DEFAULT_WORKSPACE, output_dir)
            if err:
                return err
        else:
            output_dir_abs = os.path.dirname(abs_path)
        if not os.path.exists(output_dir_abs):
            return f"Output directory does not exist: {output_dir_abs}"
        if not os.access(output_dir_abs, os.W_OK):
            return f"Output directory is not writable: {output_dir_abs}"
        with open(abs_path, "r", encoding="utf-8") as f:
            content = f.read()
        lines = content.splitlines(keepends=True)
        if not lines:
            lines = [content] if content else [""]
        total_lines = len(lines)
        source_name = os.path.splitext(os.path.basename(abs_path))[0]
        source_ext = os.path.splitext(os.path.basename(abs_path))[1] or ".txt"
        chunks = []
        if split_method == "equal":
            if not num_parts or num_parts < 2:
                return "Error: 'num_parts' must be at least 2 when using 'equal' split method."
            lines_per_part = total_lines // num_parts
            remainder = total_lines % num_parts
            start = 0
            for i in range(num_parts):
                end = start + lines_per_part + (1 if i < remainder else 0)
                chunks.append("".join(lines[start:end]))
                start = end
        elif split_method == "lines":
            if not lines_per_file or lines_per_file < 1:
                return "Error: 'lines_per_file' must be at least 1 when using 'lines' split method."
            for start in range(0, total_lines, lines_per_file):
                end = min(start + lines_per_file, total_lines)
                chunks.append("".join(lines[start:end]))
        elif split_method == "headers":
            current = []
            for line in lines:
                stripped = line.lstrip()
                if stripped.startswith("## ") or stripped.startswith("### "):
                    if current:
                        chunks.append("".join(current))
                    current = [line]
                else:
                    current.append(line)
            if current:
                chunks.append("".join(current))
            if not chunks:
                chunks = [content]
        else:
            return f"Error: 'split_method' must be 'equal', 'lines', or 'headers', got '{split_method}'"
        output_files = []
        for idx, chunk in enumerate(chunks, 1):
            out_name = f"{source_name}_part{idx}{source_ext}"
            out_path = os.path.join(output_dir_abs, out_name)
            if is_dangerous_path(out_path):
                return f"Output path is in a protected location: {out_path}"
            if is_plugins_dir_write_blocked(out_path):
                return "Blocked: cannot write to the plugins directory. Use create_plugin (evolution mode) to add plugins."
            try:
                from utils.user_model import check_user_model_direct_write_allowed

                _ws = _configured_workspace_dir()
                _um_err = check_user_model_direct_write_allowed(_ws, out_path, mode="overwrite")
                if _um_err:
                    return _um_err
            except Exception as exc:
                _debug_log(f"[ERROR split_text_file user model guard] {exc}", "ERROR")
                return "Refused: user model write guard failed (fail-closed)."
            with open(out_path, "w", encoding="utf-8") as f:
                f.write(chunk)
            output_files.append(out_path)
        result = f"Split successfully into {len(output_files)} file(s):\n"
        for i, path in enumerate(output_files, 1):
            result += f"{i}. {path} ({os.path.getsize(path)} bytes)\n"
        return result
    except json.JSONDecodeError as e:
        return f"Invalid JSON input: {str(e)}"
    except Exception as e:
        _debug_log(f"[ERROR split_text_file] {e}", "ERROR")
        return f"Split text file failed: {str(e)}"


@tool
def merge_text_files(input_json: str) -> str:
    """Merge multiple .txt or .md files into a single file.

    Input must be a JSON string with "file_paths" (list, at least 2), "output_path", optional "separator" (default "\\n\\n").
    """
    _debug_log(f"[DEBUG merge_text_files] Called with input_json: {input_json!r}")
    _emit_input_status("Merging files…")
    try:
        if isinstance(input_json, dict):
            data = input_json
        elif isinstance(input_json, str) and input_json.strip():
            data = json.loads(input_json)
        else:
            data = {}
        file_paths = data.get("file_paths", [])
        output_path = (data.get("output_path") or "").strip()
        separator = data.get("separator", "\n\n")
        if not file_paths or not isinstance(file_paths, list):
            return "Error: 'file_paths' must be a list with at least 2 file paths."
        if len(file_paths) < 2:
            return "Error: At least 2 files are required to merge."
        if not output_path:
            return "Error: 'output_path' is required."
        output_path_abs, err = _resolve_write(_DEFAULT_WORKSPACE, output_path)
        if err:
            return err
        if is_plugins_dir_write_blocked(output_path_abs):
            return "Blocked: cannot write to the plugins directory. Use create_plugin (evolution mode) to add plugins."
        output_dir = os.path.dirname(output_path_abs)
        if not os.path.exists(output_dir):
            os.makedirs(output_dir, exist_ok=True)
        try:
            from utils.user_model import check_user_model_direct_write_allowed

            _ws = _configured_workspace_dir()
            _um_err = check_user_model_direct_write_allowed(_ws, output_path_abs, mode="overwrite")
            if _um_err:
                return _um_err
        except Exception as exc:
            _debug_log(f"[ERROR merge_text_files user model guard] {exc}", "ERROR")
            return "Refused: user model write guard failed (fail-closed)."
        contents = []
        for file_path in file_paths:
            p = (file_path or "").strip()
            abs_p, err = _resolve_read(_DEFAULT_WORKSPACE, p)
            if err:
                return err
            if not os.path.exists(abs_p):
                return f"File not found: {abs_p}"
            if os.path.isdir(abs_p):
                return f"Path is a directory, not a file: {abs_p}"
            with open(abs_p, "r", encoding="utf-8") as f:
                contents.append(f.read())
        merged = separator.join(contents)
        with open(output_path_abs, "w", encoding="utf-8") as f:
            f.write(merged)
        size = os.path.getsize(output_path_abs)
        return (
            f"Text files merged successfully!\n"
            f"Output file: {output_path_abs}\n"
            f"File size: {size} bytes\n"
            f"Merged {len(contents)} file(s) in order."
        )
    except json.JSONDecodeError as e:
        return f"Invalid JSON input: {str(e)}"
    except Exception as e:
        _debug_log(f"[ERROR merge_text_files] {e}", "ERROR")
        return f"Merge text files failed: {str(e)}"


# ---------- save_to_file / write_text_file / append_to_text_file ----------
def _write_verify(abs_path: str, content: str) -> str:
    """Write content to abs_path (already safety-checked), then verify. Returns success message or error."""
    parent = os.path.dirname(abs_path)
    if parent and not os.path.exists(parent):
        os.makedirs(parent, exist_ok=True)
    if os.path.exists(abs_path) and os.path.isfile(abs_path):
        existing_size = os.path.getsize(abs_path)
        if existing_size > 20000 and len(content) < existing_size * 0.5:
            return (
                f"Refusing to overwrite: file has {existing_size} bytes but content is only {len(content)} characters. "
                "Use read_file_chunk and submit_ocr_chunk_corrections to process in sections, or use a different path."
            )
    with open(abs_path, "w", encoding="utf-8") as f:
        f.write(content)
    if not os.path.exists(abs_path):
        return "File was not created after write."
    size = os.path.getsize(abs_path)
    return f"File saved: {abs_path} ({size} bytes)."


@tool
def save_to_file(file_path: str, content: str) -> str:
    """Save (overwrite/create) text content to a file. Use full absolute paths. Creates parent dirs; refuses dangerous paths. Verifies write."""
    return write_text_file.invoke({"file_path": file_path, "content": content})


_IDENTITY_FILE_CROSS_CHECK: dict[str, tuple[str, ...]] = {
    "IDENTITY.MD": ("# USER.md",),
    "PERSONA.MD": ("# USER.md",),
    "USER.MD": ("# IDENTITY.md", "# PERSONA.md"),
}


@tool
def write_text_file(file_path: str, content: str) -> str:
    """Write or overwrite a plain text file. Use full absolute paths (e.g. ~/.mugi/workspace/MEMORY.md). Creates parent dirs; refuses dangerous paths.

    Post-onboarding, USER.md and persona files (PERSONA.md, IDENTITY.md, SOUL.md) are user-owned — use **patch_user_model** (or the "You" pane in Settings → You) for new user-stated personal facts. patch_user_profile is legacy; the first-class User Model + "You" pane are the modern paths.

    Parallel tool rounds: avoid read_file on the same path in the same assistant message as this write; ordering is not guaranteed."""
    _emit_input_status("Writing file…")
    try:
        file_path = (file_path or "").strip()
        if not file_path:
            return "No file_path provided."
        basename_upper = os.path.basename(file_path).upper()
        wrong_headers = _IDENTITY_FILE_CROSS_CHECK.get(basename_upper, ())
        content_lstrip = (content or "").lstrip()
        for wrong_header in wrong_headers:
            if content_lstrip.startswith(wrong_header):
                return (
                    f"Refused: content starts with '{wrong_header}' but you are writing to {os.path.basename(file_path)}. "
                    "You mixed up the file contents. Write the correct content for this file."
                )
        abs_path, err = _resolve_write(_DEFAULT_WORKSPACE, file_path)
        if err:
            return err
        try:
            from utils.kanban_dispatcher import current_claim_token

            norm = abs_path.replace("\\", "/")
            if current_claim_token() and "/web_runs/" in norm:
                return (
                    "Refused: kanban web-act workers must use write_web_act_file for paths under "
                    "web_runs/, not write_text_file."
                )
            if not current_claim_token() and "/web_runs/" in norm:
                import re

                if re.search(r"/web_runs/[^/]+/web_runs/", norm):
                    return (
                        "Refused: nested web_runs path. Write scripts relative to the scope root "
                        "(e.g. explore.py under web_runs/{thread_id}/), not web_runs/.../web_runs/.... "
                        "Use write_web_act_file from chat when web-act is enabled."
                    )
        except Exception:
            pass
        if is_dangerous_path(abs_path):
            return f"Operation blocked: path is in a protected or dangerous location: {abs_path}"
        if is_plugins_dir_write_blocked(abs_path):
            return "Blocked: cannot write to the plugins directory. Use create_plugin (evolution mode) to add plugins."
        ws = _configured_workspace_dir()
        if _is_evolution_protected_workspace_target(ws, abs_path) and _content_smells_like_truncated_read_file_paste(
            content
        ):
            return (
                "Refused: content looks like a truncated read_file preview (or a paste of one). "
                "Read the full file with read_file_chunk, merge edits locally, then save only complete content—never the warning banner or partial preview."
            )
        if _restricted_background_writes() and _is_under_mugi_tools_dir(abs_path):
            return (
                "Refused: writes under ~/.mugi/tools/ are blocked for background, idle, and scheduled runs. "
                "Use a user-facing chat session or approved plugin flows to add tool modules."
            )
        try:
            from core import get_config
            from utils.personality_writes import check_profile_write_allowed
            from utils.user_model import check_user_model_direct_write_allowed

            _cfg = get_config()
            _ws = ws or _configured_workspace_dir()
            _prof_err = check_profile_write_allowed(
                os.path.basename(abs_path),
                abs_path,
                content,
                workspace_dir=_ws,
                config=_cfg,
                mode="overwrite",
            )
            if _prof_err:
                return _prof_err
            _um_err = check_user_model_direct_write_allowed(_ws, abs_path, mode="overwrite")
            if _um_err:
                return _um_err
        except Exception as exc:
            _debug_log(f"[ERROR write_text_file profile guard] {exc}", "ERROR")
            return (
                "Refused: profile write guard failed (fail-closed). "
                "Use Settings → Profile or patch_user_model for personal facts."
            )
        return _write_verify(abs_path, content)
    except Exception as e:
        _debug_log(f"[ERROR write_text_file] {e}", "ERROR")
        return f"Write failed: {str(e)}"


@tool
def append_to_text_file(file_path: str, content: str) -> str:
    """Append text to the end of a file (creates file if it does not exist). Use full absolute paths. Creates parent dirs; refuses dangerous paths."""
    _emit_input_status("Appending to file…")
    try:
        file_path = (file_path or "").strip()
        if not file_path:
            return "No file_path provided."
        abs_path, err = _resolve_write(_DEFAULT_WORKSPACE, file_path)
        if err:
            return err
        if is_dangerous_path(abs_path):
            return f"Operation blocked: path is in a protected or dangerous location: {abs_path}"
        if is_plugins_dir_write_blocked(abs_path):
            return "Blocked: cannot write to the plugins directory. Use create_plugin (evolution mode) to add plugins."
        try:
            from core import get_config
            from utils.personality_writes import check_profile_write_allowed
            from utils.user_model import check_user_model_direct_write_allowed

            _cfg = get_config()
            _ws = _configured_workspace_dir()
            _prof_err = check_profile_write_allowed(
                os.path.basename(abs_path),
                abs_path,
                content or "",
                workspace_dir=_ws,
                config=_cfg,
                mode="append",
            )
            if _prof_err:
                return _prof_err
            _um_err = check_user_model_direct_write_allowed(_ws, abs_path, mode="append")
            if _um_err:
                return _um_err
        except Exception as exc:
            _debug_log(f"[ERROR append_to_text_file profile guard] {exc}", "ERROR")
            return (
                "Refused: profile append guard failed (fail-closed). "
                "Use patch_user_model for user-stated personal facts or Settings → Profile."
            )
        parent = os.path.dirname(abs_path)
        if parent and not os.path.exists(parent):
            os.makedirs(parent, exist_ok=True)
        mode = "a" if os.path.exists(abs_path) else "w"
        with open(abs_path, mode, encoding="utf-8") as f:
            f.write(content)
        return f"Appended to {abs_path} successfully ({len(content)} characters added)."
    except Exception as e:
        _debug_log(f"[ERROR append_to_text_file] {e}", "ERROR")
        return f"Append failed: {str(e)}"


@tool
def patch_user_profile(field: str, value: str) -> str:
    """[Legacy / deprecated for new facts]

    Update a single field in the old USER.md profile.

    For new personal facts after the User Model migration, strongly prefer `patch_user_model`
    (the chat tool) or direct editing in Settings → You. This remains only for backward
    compatibility with any residual legacy profile usage.

    Merges into existing USER.md (fills empty fields only; otherwise appends under Important Notes).
    For biographical fields the value must appear in a recent user message.
    Post-onboarding, do not use write_text_file on USER.md — use `patch_user_model` or the You pane."""
    _emit_input_status("Updating user profile…")
    try:
        from core import get_config
        from persistence import secure_atomic_write, invalidate_personality_cache
        from utils.personality_writes import (
            check_profile_write_allowed,
            field_is_sensitive,
            merge_user_profile_field,
            normalize_user_profile_field,
            patch_user_profile_write_context,
            sensitive_field_allowed,
        )

        field_norm = normalize_user_profile_field(field)
        if not field_norm:
            return "No field provided."
        val = (value or "").strip()
        if not val:
            return "No value provided."

        if _direct_personal_write_blocked():
            return (
                "Refused: patch_user_profile is not available during background, idle, scheduled, "
                "messaging, mesh, hook, or subagent runs."
            )

        try:
            from tool_policy import get_current_policy_context

            ctx = get_current_policy_context()
            recent = str(getattr(ctx, "recent_user_text", "") or "") if ctx else ""
        except Exception:
            recent = ""

        if field_is_sensitive(field_norm) and not sensitive_field_allowed(val, recent):
            return (
                "Refused: this field looks biographical/sensitive. Only store it when the user "
                "explicitly stated it in this conversation, or ask them to confirm in Settings → Profile."
            )

        abs_path, err = _resolve_write(_DEFAULT_WORKSPACE, "USER.md")
        if err:
            return err
        ws = _configured_workspace_dir()
        cfg = get_config()
        existing = ""
        if os.path.isfile(abs_path):
            with open(abs_path, "r", encoding="utf-8") as f:
                existing = f.read()

        merged = merge_user_profile_field(existing, field_norm, val)
        with patch_user_profile_write_context():
            prof_err = check_profile_write_allowed(
                "USER.md",
                abs_path,
                merged,
                workspace_dir=ws,
                config=cfg,
                mode="overwrite",
            )
            if prof_err:
                return prof_err
            secure_atomic_write(abs_path, merged)
        try:
            invalidate_personality_cache()
        except Exception:
            pass
        return f"Updated USER.md field {field_norm!r}."
    except ValueError as exc:
        return f"Refused: {exc}"
    except Exception as e:
        _debug_log(f"[ERROR patch_user_profile] {e}", "ERROR")
        return f"Profile update failed: {str(e)}"


@tool
def patch_user_model(patch: dict, allow_bulk_clear: bool = False) -> str:
    """Update the first-class User Model with facts the user explicitly stated in this conversation.

    Preferred tool for user-stated personal information (identity, goals, preferences, relationships, current focus, etc.).
    The patch is a dictionary of top-level sections (e.g. {"identity": {...}, "goals": [...] }).

    For biographical/sensitive fields, the values must appear in recent user messages.
    This tool is not available in background, idle, or scheduled runs — use Settings or direct chat only.

    allow_bulk_clear: leave False. A patch that would clear most/all of the user's
    profile in one step is refused unless this is True. Only set it after the user
    has explicitly confirmed in this conversation that they want their profile wiped.
    """
    _emit_input_status("Updating personal model…")
    try:
        if _direct_personal_write_blocked():
            return (
                "Refused: patch_user_model is not available during background, idle, scheduled, "
                "messaging, mesh, hook, or subagent runs. "
                "Personal facts should be stated directly in chat or updated in Settings → You."
            )
        try:
            from tool_policy import get_current_policy_context

            if get_current_policy_context() is None:
                return (
                    "Refused: patch_user_model requires an active foreground chat session."
                )
        except Exception:
            return "Refused: patch_user_model requires an active foreground chat session."

        if not isinstance(patch, dict) or not patch:
            return "Refused: patch must be a non-empty dictionary of model sections."

        from utils.user_model import patch_has_applicable_sections

        if not patch_has_applicable_sections(patch):
            return "Refused: patch must include at least one well-formed User Model section."

        try:
            from tool_policy import get_current_policy_context
            ctx = get_current_policy_context()
            recent = str(getattr(ctx, "recent_user_text", "") or "") if ctx else ""
        except Exception:
            recent = ""

        # Basic sensitive field protection (reuse existing helpers where applicable)
        # For Phase 2 we apply the same spirit as patch_user_profile for any obvious biographical content.
        from utils.personality_writes import field_is_sensitive, sensitive_field_allowed

        def _iter_field_values(section: str, value: object):
            if isinstance(value, dict):
                for k, v in value.items():
                    yield k, v
            elif isinstance(value, list):
                for item in value:
                    if isinstance(item, dict):
                        for k, v in item.items():
                            yield k, v

        for section, value in patch.items():
            for k, v in _iter_field_values(section, value):
                if field_is_sensitive(k) and not sensitive_field_allowed(str(v), recent):
                    return (
                        f"Refused: field '{k}' looks biographical/sensitive. "
                        "Only store it when the user explicitly stated it in this conversation."
                    )

        ws = _configured_workspace_dir()
        thread_id = None
        try:
            from persistence import get_current_agent_thread_id

            thread_id = get_current_agent_thread_id()
        except Exception:
            pass
        result = _user_model_apply_patch(
            ws,
            patch,
            actor="agent",
            provenance={"source": "chat", "thread_id": thread_id},
            allow_bulk_clear=bool(allow_bulk_clear),
        )

        if result.get("ok"):
            return "User Model updated successfully."
        else:
            return result.get("error", "Failed to update User Model.")

    except Exception as e:
        _debug_log(f"[ERROR patch_user_model] {e}", "ERROR")
        return f"User Model update failed: {str(e)}"


@tool
def propose_user_model_update(changes: dict, rationale: str) -> str:
    """Propose an update to the User Model.

    Use this when you have observed something about the user that should be recorded,
    but you are not 100% certain or it was not explicitly stated by the user in this turn.
    The proposal will be reviewed by the user before being applied.

    changes: dict of top-level User Model sections to update.
    rationale: Clear explanation of why you are proposing this change.
    """
    try:
        if not isinstance(changes, dict) or not changes:
            return "Refused: changes must be a non-empty dictionary of User Model sections."
        if not (rationale or "").strip():
            return "Refused: rationale is required for proposals."

        from utils.user_model import patch_has_applicable_sections

        if not patch_has_applicable_sections(changes):
            return "Refused: changes must include at least one well-formed User Model section."

        if _restricted_background_writes():
            return (
                "Refused: propose_user_model_update is not available during background, idle, "
                "or scheduled runs."
            )

        ws = _configured_workspace_dir()
        result = create_proposal_detailed(ws, changes, rationale.strip(), source="agent")
        status = result.get("status")
        if status == "created":
            return (
                f"Proposal created (id: {result.get('id')}). "
                "The user can review and accept/reject it."
            )
        if status == "duplicate":
            # Not a failure — the user already has an equivalent proposal waiting.
            # Tell the model plainly so it does NOT retry-loop on this.
            return (
                "No new proposal created: "
                f"{result.get('reason') or 'a similar proposal is already pending review'}. "
                "Do not retry — the user will see the existing proposal."
            )
        return (
            "Failed to create proposal: "
            f"{result.get('reason') or 'invalid changes or storage error'}."
        )
    except Exception as e:
        _debug_log(f"[ERROR propose_user_model_update] {e}", "ERROR")
        return f"Proposal failed: {str(e)}"


@tool
def list_pending_proposals() -> str:
    """List all pending proposals for the User Model.

    Use this to see what updates are waiting for user review.
    """
    try:
        from utils.user_model import filter_proposals_for_you_pane, list_proposals

        ws = _configured_workspace_dir()
        proposals = filter_proposals_for_you_pane(ws, list_proposals(ws))
        if not proposals:
            return "No pending proposals."

        lines = ["Pending User Model proposals:"]
        for p in proposals[:10]:  # cap output
            pid = p.get("id") or "?"
            lines.append(f"- {pid} (from {p.get('source', '?')}): {(p.get('rationale') or '')[:100]}")
        if len(proposals) > 10:
            lines.append(f"... and {len(proposals) - 10} more.")
        return "\n".join(lines)
    except Exception as e:
        _debug_log(f"[ERROR list_pending_proposals] {e}", "ERROR")
        return f"Failed to list proposals: {str(e)}"


# ---------- read_file_chunk / submit_ocr_chunk_corrections ----------
@tool
def read_file_chunk(file_path: str, chunk_index: int, chunk_size: int = FILE_CHUNK_SIZE, num_chunks: int = 1) -> str:
    """Read one or more consecutive sections (chunks) of a text file. file_path: full path. chunk_index: 0-based. chunk_size: chars per chunk (default 3500). num_chunks: how many consecutive chunks (default 1, max 8)."""
    _emit_input_status("Reading file chunk…")
    try:
        file_path = file_path.strip(' "\n,')
        abs_path, err = _resolve_read(_DEFAULT_WORKSPACE, file_path)
        if err:
            return err
        if not os.path.exists(abs_path) or not os.path.isfile(abs_path):
            return f"File not found or not a file: {abs_path}"
        with open(abs_path, "r", encoding="utf-8") as f:
            content = f.read()
        total_chunks = max(1, math.ceil(len(content) / chunk_size))
        if chunk_index < 0 or chunk_index >= total_chunks:
            return f"Chunk index {chunk_index} is out of range. File has {total_chunks} chunk(s) (0 to {total_chunks - 1})."
        n = max(1, min(int(num_chunks), _READ_FILE_CHUNK_BATCH_MAX, total_chunks - chunk_index))
        parts = []
        for i in range(n):
            idx = chunk_index + i
            start = idx * chunk_size
            end = min(start + chunk_size, len(content))
            seg = content[start:end]
            if n == 1:
                parts.append(f"Chunk {idx + 1} of {total_chunks} (chars {start}-{end}):\n\n{seg}")
            else:
                parts.append(f"--- Chunk {idx + 1} of {total_chunks} (chars {start}-{end}) ---\n\n{seg}")
        result = "\n\n".join(parts)
        if n > 1:
            result = f"Chunks {chunk_index + 1}–{chunk_index + n} of {total_chunks}:\n\n{result}"
        return result
    except Exception as e:
        _debug_log(f"[ERROR read_file_chunk] {e}", "ERROR")
        return f"read_file_chunk failed: {str(e)}"


@tool
def submit_ocr_chunk_corrections(
    file_path: str,
    chunk_index: int,
    corrected_content: str,
    chunk_size: int = FILE_CHUNK_SIZE,
    output_file: str | None = None,
) -> str:
    """Replace one chunk of a file with new text. Use same chunk_index and chunk_size as in read_file_chunk. output_file: optional path to write to (keeps original unchanged)."""
    _emit_input_status("Applying corrections…")
    try:
        file_path = file_path.strip(' "\n,')
        corrected_content = corrected_content if corrected_content is not None else ""
        abs_path, err = _resolve_read(_DEFAULT_WORKSPACE, file_path)
        if err:
            return err
        if is_dangerous_path(abs_path):
            return f"Path is in a protected location: {abs_path}"
        if not os.path.exists(abs_path) or not os.path.isfile(abs_path):
            return f"File not found or not a file: {abs_path}"
        write_path = abs_path
        if output_file:
            out_abs, err = _resolve_write(_DEFAULT_WORKSPACE, output_file.strip(' "\n,'))
            if err:
                return err
            if is_dangerous_path(out_abs):
                return f"Output path is in a protected location: {out_abs}"
            if is_plugins_dir_write_blocked(out_abs):
                return "Blocked: cannot write to the plugins directory. Use create_plugin (evolution mode) to add plugins."
            try:
                from utils.user_model import check_user_model_direct_write_allowed

                _ws = _configured_workspace_dir()
                _um_err = check_user_model_direct_write_allowed(_ws, out_abs, mode="overwrite")
                if _um_err:
                    return _um_err
            except Exception as exc:
                _debug_log(f"[ERROR submit_ocr_chunk_corrections user model guard] {exc}", "ERROR")
                return "Refused: user model write guard failed (fail-closed)."
            if not os.path.exists(out_abs) or chunk_index == 0:
                shutil.copy2(abs_path, out_abs)
            write_path = out_abs
        try:
            from utils.user_model import check_user_model_direct_write_allowed

            _ws = _configured_workspace_dir()
            _um_err = check_user_model_direct_write_allowed(_ws, write_path, mode="overwrite")
            if _um_err:
                return _um_err
        except Exception as exc:
            _debug_log(f"[ERROR submit_ocr_chunk_corrections user model guard] {exc}", "ERROR")
            return "Refused: user model write guard failed (fail-closed)."
        with open(write_path, "r", encoding="utf-8") as f:
            content = f.read()
        total_chunks = max(1, math.ceil(len(content) / chunk_size))
        if chunk_index < 0 or chunk_index >= total_chunks:
            return f"Chunk index {chunk_index} is out of range. File has {total_chunks} chunk(s)."
        start = chunk_index * chunk_size
        end = min(start + chunk_size, len(content))
        new_content = content[:start] + corrected_content + content[end:]
        with open(write_path, "w", encoding="utf-8") as f:
            f.write(new_content)
        if output_file:
            return f"Updated chunk {chunk_index + 1} of {total_chunks} in {write_path}. Original unchanged. File size now {len(new_content)} characters."
        return f"Updated chunk {chunk_index + 1} of {total_chunks}. File size now {len(new_content)} characters."
    except Exception as e:
        _debug_log(f"[ERROR submit_ocr_chunk_corrections] {e}", "ERROR")
        return f"submit_ocr_chunk_corrections failed: {str(e)}"


# ---------- get_file_metadata / verify_file_exists / get_default_save_location / suggest_save_filename ----------
@tool
def get_file_metadata(path: str) -> str:
    """Retrieve metadata (size, dates, type, permissions) for a file or directory. Input: full path. Refuses protected paths."""
    _emit_input_status("Reading metadata…")
    try:
        path = (path or "").strip(' "\n,')
        abs_path, err = _resolve_read(_DEFAULT_WORKSPACE, path)
        if err:
            return err
        if is_dangerous_path(abs_path):
            return f"Access blocked: path is in a protected or dangerous location: {abs_path}"
        p = Path(abs_path).resolve(strict=False)
        if not p.exists():
            return f"Path does not exist: {abs_path}"
        stat = p.stat()
        info = {
            "absolute_path": str(p),
            "exists": True,
            "is_file": p.is_file(),
            "is_dir": p.is_dir(),
            "is_symlink": p.is_symlink(),
            "size_bytes": stat.st_size if p.is_file() else None,
            "modified_iso": datetime.fromtimestamp(stat.st_mtime).isoformat(),
            "created_iso": datetime.fromtimestamp(stat.st_birthtime).isoformat() if hasattr(stat, "st_birthtime") else None,
            "mime_type_guess": (mimetypes.guess_type(str(p))[0] or "unknown"),
            "permissions_octal": oct(stat.st_mode)[-3:],
        }
        return json.dumps(info, indent=2)
    except Exception as e:
        _debug_log(f"[ERROR get_file_metadata] {e}", "ERROR")
        return f"Metadata retrieval failed: {str(e)}"


@tool
def verify_file_exists(file_path: str) -> str:
    """Verify that a file exists at the given path; return size, modified time, and a short content preview."""
    _emit_input_status("Verifying file…")
    try:
        file_path = (file_path or "").strip(' "\n,')
        abs_path, err = _resolve_read(_DEFAULT_WORKSPACE, file_path)
        if err:
            return err
        if not os.path.exists(abs_path):
            return f"File does NOT exist at: {abs_path}"
        if not os.path.isfile(abs_path):
            return f"Path exists but is not a file: {abs_path}"
        stat = os.stat(abs_path)
        file_size = stat.st_size
        modified_time = datetime.fromtimestamp(stat.st_mtime).strftime("%Y-%m-%d %H:%M:%S")
        preview = ""
        try:
            with open(abs_path, "r", encoding="utf-8", errors="ignore") as f:
                preview_content = f.read(500)
                if preview_content:
                    preview = f"\nContent preview (first 500 chars):\n{preview_content}"
                    if file_size > 500:
                        preview += f"\n... (total {file_size} bytes)"
                else:
                    preview = "\nFile is empty"
        except Exception as read_error:
            preview = f"\nCould not read contents: {read_error}"
        return (
            f"File EXISTS and verified.\n"
            f"Path: {abs_path}\n"
            f"Size: {file_size} bytes\n"
            f"Last modified: {modified_time}\n"
            f"{preview}"
        )
    except Exception as e:
        _debug_log(f"[ERROR verify_file_exists] {e}", "ERROR")
        return f"Verification failed: {str(e)}"


@tool
def get_default_save_location() -> str:
    """Get the default directory for saving files the user will open: ~/Documents/Mugi. Returns a DIRECTORY path; when saving use a full file path including filename. Use this when the user does not specify where to save."""
    _emit_input_status("Finding save location…")
    try:
        out_dir = get_user_output_dir()
        return (
            f"Default save location (use when unsure where to save): {out_dir}\n\n"
            "When saving, include a filename: save_to_file(file_path=\"{path}/filename<extension>\", content=\"...\")"
        )
    except Exception as e:
        _debug_log(f"[ERROR get_default_save_location] {e}", "ERROR")
        return f"Failed to get default save location: {str(e)}"


@tool
def suggest_save_filename(source_file_path: str, suffix: str = "_ocr", extension: str = "") -> str:
    """Suggest a filename for saving OCR or processed results based on a source file. extension: e.g. .md, .txt (must match user request)."""
    _emit_input_status("Suggesting filename…")
    try:
        if extension and not extension.startswith("."):
            extension = "." + extension
        source_path, err = _resolve_read(_DEFAULT_WORKSPACE, source_file_path)
        if err:
            return err
        if is_dangerous_path(source_path):
            return f"Source path is in a protected location: {source_path}"
        source_dir = os.path.dirname(source_path)
        # OCR and other saves: put output in ~/Documents/Mugi/OCR (not hidden .mugi)
        if suffix == "_ocr" or ".mugi" in os.path.abspath(source_dir):
            from utils.path_safety import get_ocr_output_dir
            source_dir = get_ocr_output_dir()
        source_basename = os.path.basename(source_path)
        source_name, _ = os.path.splitext(source_basename)
        if extension:
            suggested_name = f"{source_name}{suffix}{extension}"
        else:
            suggested_name = f"{source_name}{suffix}<extension>"
        suggested_path = os.path.join(source_dir, suggested_name)
        abs_suggested = os.path.abspath(suggested_path)
        if extension:
            return (
                f"Suggested save path: {abs_suggested}\n"
                f"Based on source '{source_basename}'. save_to_file(file_path=\"{abs_suggested}\", content=<text>)"
            )
        return (
            f"Suggested base: {source_name}{suffix}\n"
            f"Directory: {source_dir}\n"
            "Provide the extension the user requested (e.g. .md, .txt)."
        )
    except Exception as e:
        _debug_log(f"[ERROR suggest_save_filename] {e}", "ERROR")
        return f"Failed to suggest save filename: {str(e)}"


_CORRECT_OCR_SYSTEM_PROMPT = """You are a text correction assistant. Correct OCR and typo errors in the text the user provides. Preserve meaning, formatting, and structure. Preserve footnote references exactly (e.g. [^1], [^2], [^*]) and the **Footnotes** section at the end; only fix OCR and typo errors in the wording.

Do not invent or add footnote content. If the text contains a footnote reference (e.g. [^1], [^*]) but there is no corresponding entry in the **Footnotes** section, leave the reference as-is and do not write any footnote entry for it. If a footnote entry already exists, only fix OCR and typo errors in that existing text; do not expand or rewrite it. Never fabricate footnote text that does not appear in the source. Do not use footnote bodies to explain or comment on the document (e.g. do not write "This footnote reference appears in the text but is not listed" or "preserved as-is" or similar meta-commentary as footnote content).

Also fix line breaks and hyphenation caused by page layout: (1) Rejoin words that were split across lines by a hyphen (e.g. "human-\\nity" → "humanity", "trib-\\nalisms" → "tribalisms"). (2) Remove arbitrary line breaks that break sentences—join lines that are clearly part of the same sentence or paragraph so the text reads as continuous prose. Keep intentional paragraph breaks (blank lines between paragraphs).

Output only the corrected text. Do not add any preamble (e.g. "Here is the corrected version"), summary, or closing note (e.g. "Note: The original text contains..."). The first line of your response must be the first line of the corrected document; the last line must be the last line of the corrected document."""


def _rejoin_hyphenation(text: str) -> str:
    """Rejoin words broken by end-of-line hyphenation (e.g. human-\\nity → humanity)."""
    if not text or not text.strip():
        return text
    text = re.sub(r"(\w+)-\s*\n\s*(\w+)", r"\1\2", text)
    text = re.sub(r"\s*-\s*\n\s*", " ", text)
    return text


_THINKING_BLOCK_RE = re.compile(
    r"<think>.*?</think>\s*"
    r"|<\|channel>thought.*?<channel\|>\s*",
    re.DOTALL | re.IGNORECASE,
)
_THINKING_PREFIX_RE = re.compile(
    r"^(?:thinking process|thought process|reasoning|analysis):\s*",
    re.IGNORECASE,
)
_REASONING_PARAGRAPH_RE = re.compile(
    r"^(?:"
    r"\d+[\.\)]\s"
    r"|[-*]\s"
    r"|step\s+\d+"
    r"|thinking process"
    r"|thought process"
    r"|reasoning:"
    r"|analysis:"
    r")",
    re.IGNORECASE,
)
_TRANSITION_PARAGRAPH_RE = re.compile(
    r"^(?:"
    r"here\s+is\s+the\s+corrected"
    r"|below\s+is\s+the"
    r"|corrected\s+(?:text|version|document)"
    r"|the\s+corrected\s+(?:text|version|document)"
    r")",
    re.IGNORECASE,
)
# Substrings that almost only appear in leaked meta-instructions (avoid bare "task:" /
# "constraints:" — those appear in real documents).
_META_REASONING_HINTS = (
    "analyze the request",
    "step-by-step correction",
    "correction strategy",
    "preserve meaning, formatting, and structure",
    "output only the corrected text",
    "first line of your response",
    "do not invent or add footnote content",
    "re-reading the instruction",
    "wait, check",
)
_META_REASONING_LINE_RE = re.compile(
    r"^\s*(?:"
    r"\d+[\.\)]\s+\*\*analyze the request"
    r"|[-*]\s+\*\*wait,"
    r"|[-*]\s+\*\*task"
    r"|[-*]\s+\*\*constraints"
    r"|[-*]\s+\*\*input"
    r"|step\s+\d+"
    r"|i will\s+"
    r"|let me\s+"
    r")",
    re.IGNORECASE,
)


def _strip_thinking_from_completion(text: str) -> str:
    """Remove thinking/reasoning wrappers for OCR correction completions.

    This OCR-specific stripper is intentionally separate from chat-response
    finalization in agent_loop/utils.
    """
    if not text:
        return text
    text = _THINKING_BLOCK_RE.sub("", text)
    text = text.strip()

    # Split into paragraphs and drop leading meta-reasoning paragraphs when
    # present (common across many "thinking" model families).
    paragraphs = re.split(r"\n{2,}", text)
    drop_until = 0
    saw_reasoning = bool(_THINKING_PREFIX_RE.match(text))
    for i, para in enumerate(paragraphs):
        stripped = para.strip()
        if not stripped:
            if saw_reasoning:
                drop_until = i + 1
            continue
        lowered = stripped.lower()
        is_reasoning_para = (
            _REASONING_PARAGRAPH_RE.match(stripped) is not None
            or _TRANSITION_PARAGRAPH_RE.match(stripped) is not None
            or _META_REASONING_LINE_RE.match(stripped) is not None
            or any(h in lowered for h in _META_REASONING_HINTS)
        )
        if is_reasoning_para:
            saw_reasoning = True
            drop_until = i + 1
            continue
        if saw_reasoning:
            return "\n\n".join(p for p in paragraphs[drop_until:] if p.strip()).strip()
        return text

    # Everything looked like reasoning; return best-effort stripped text.
    return _THINKING_PREFIX_RE.sub("", text).strip() if saw_reasoning else text


def _looks_like_reasoning_leak(content: str) -> bool:
    """Heuristic: detect meta-reasoning output instead of corrected document text."""
    if not isinstance(content, str):
        return False
    head = content[:4000]
    lowered = head.lower()
    if lowered.startswith(("thinking process:", "thought process:", "reasoning:", "analysis:")):
        return True

    hint_hits = sum(1 for h in _META_REASONING_HINTS if h in lowered)
    if hint_hits >= 2:
        return True

    reasoning_lines = 0
    for line in head.splitlines()[:160]:
        if _META_REASONING_LINE_RE.match(line):
            reasoning_lines += 1
            if reasoning_lines >= 3:
                return True
    return False


def _validate_correction_content(content: str) -> str:
    """Strip thinking wrappers, then validate corrected text."""
    content = _strip_thinking_from_completion(content)
    if not content:
        raise ValueError("Correction returned empty content.")
    if _looks_like_reasoning_leak(content):
        raise ValueError("OCR correction model returned reasoning text instead of corrected document.")
    return content


_OCR_STRICT_RETRY_USER = (
    "Your previous reply was rejected because it contained analysis or instructions instead of document text. "
    "Reply again with ONLY the corrected passage — same structure as the input — no preamble, checklist, or commentary."
)


def _openai_ocr_correction_raw(
    client: Any,
    model: str,
    messages: list[dict],
    max_tokens: int,
) -> str:
    """Single chat.completions call; returns raw message content (may be invalid)."""
    last_main_error: Exception | None = None
    request_variants = [
        {
            "model": model,
            "messages": messages,
            "max_tokens": max_tokens,
            "temperature": 0.2,
            "extra_body": {
                "enable_thinking": False,
                "chat_template_kwargs": {"enable_thinking": False},
            },
        },
        {
            "model": model,
            "messages": messages,
            "max_tokens": max_tokens,
            "temperature": 0.2,
        },
    ]
    resp = None
    for idx, kwargs in enumerate(request_variants):
        try:
            resp = client.chat.completions.create(**kwargs)
            break
        except Exception as e:
            last_main_error = e
            if idx == 0:
                _debug_log(
                    "[_apply_ocr_correction_impl] main model rejected thinking-suppression params; retrying without them.",
                    "WARN",
                )
                continue
            raise
    if resp is None:
        raise last_main_error or ValueError("Correction request failed.")
    choice = (resp.choices or [None])[0]
    if choice is None or getattr(choice, "message", None) is None:
        raise ValueError("Correction returned no content.")
    return getattr(choice.message, "content", None) or ""


def _apply_ocr_correction_impl(chunk: str) -> str:
    """Run LLM correction on a single chunk (caller ensures chunk is within token limits). Raises on failure."""
    base_messages: list[dict] = [
        {
            "role": "system",
            "content": (
                _CORRECT_OCR_SYSTEM_PROMPT
                + "\n\nReturn only corrected document text. Do NOT include analysis, reasoning, checklists, "
                "or step-by-step commentary."
            ),
        },
        {"role": "user", "content": chunk},
    ]

    # Primary path: use the main (larger) model for faithful text reproduction.
    try:
        from agent_loop import (
            get_backend_base_url,
            get_llmster_model,
            _get_openai_client,
        )

        base_url = get_backend_base_url()
        is_lmstudio = "1234" in (base_url or "")
        api_key = "lm-studio"
        model = (get_llmster_model() or "local-model") if is_lmstudio else "local-model"
        client = _get_openai_client(base_url, api_key)
        max_tokens = _max_tokens_for_ocr_correction_chunk(chunk)

        messages = list(base_messages)
        raw = _openai_ocr_correction_raw(client, model, messages, max_tokens)
        try:
            return _validate_correction_content(raw)
        except ValueError as ve:
            if "reasoning" not in str(ve).lower():
                raise
            _debug_log(
                "[_apply_ocr_correction_impl] main model output looked like reasoning; one strict retry.",
                "WARN",
            )
            messages = list(base_messages) + [{"role": "user", "content": _OCR_STRICT_RETRY_USER}]
            raw2 = _openai_ocr_correction_raw(client, model, messages, max_tokens)
            return _validate_correction_content(raw2)
    except Exception as _main_err:
        _debug_log(f"[_apply_ocr_correction_impl] main model failed: {_main_err}", "WARN")

    # Fallback: utility model (smaller/local) when the main model is unavailable.
    messages = base_messages
    try:
        from utility_model import utility_complete

        util_cap = min(_max_tokens_for_ocr_correction_chunk(chunk), 8192)
        utility_text = utility_complete(
            messages,
            max_tokens=util_cap,
            temperature=0.2,
            activity_operation="ocr_text_correction",
        )
        if utility_text:
            return _validate_correction_content(utility_text)
    except Exception as _util_err:
        _debug_log(f"[_apply_ocr_correction_impl] utility model failed: {_util_err}", "WARN")

    raise ValueError("Both main model and utility model failed to correct OCR text.")


def _chunks_for_correction(text: str, max_chars: int) -> list[str]:
    """Split text into chunks of at most max_chars, preferring paragraph boundaries."""
    if len(text) <= max_chars:
        return [text] if text.strip() else []
    chunks = []
    current = []
    current_len = 0
    for para in text.split("\n\n"):
        need = len(para) + 2
        if current_len + need <= max_chars:
            current.append(para)
            current_len += need
        else:
            if current:
                chunks.append("\n\n".join(current))
            if len(para) <= max_chars:
                current = [para]
                current_len = need
            else:
                for i in range(0, len(para), max_chars):
                    chunks.append(para[i : i + max_chars])
                current = []
                current_len = 0
    if current:
        chunks.append("\n\n".join(current))
    return chunks


def apply_ocr_correction(text: str, max_chars: int | None = None) -> str:
    """Correct OCR/typos in text via LLM. Chunks if needed; on failure returns original text. Used by complex OCR and Kraken pipelines.

    *max_chars* defaults to a value aligned with the completion token budget so chunks are
    less likely to be truncated mid-passage. Set ``MUGI_OCR_CORRECTION_CHUNK_CHARS`` to override.
    Parallelism: ``MUGI_OCR_CORRECTION_WORKERS`` (default 3, max 8).
    """
    if max_chars is None:
        max_chars = max(
            2000,
            int(os.environ.get("MUGI_OCR_CORRECTION_CHUNK_CHARS", str(_DEFAULT_OCR_CORRECTION_CHUNK_CHARS))),
        )
    if not (text or "").strip():
        return text or ""
    text = (text or "").strip()
    text = _rejoin_hyphenation(text)
    if len(text) <= max_chars:
        try:
            _emit_input_status("Correcting OCR text…")
            emit_live_view_preview("Correcting OCR text…", tool_name="ocr_cleanup", progress_pct=0)
            result = _apply_ocr_correction_impl(text)
            emit_live_view_preview("OCR correction complete", tool_name="ocr_cleanup", progress_pct=100)
            return result
        except Exception as e:
            _debug_log(f"[apply_ocr_correction] single chunk failed: {e}", "WARN")
            return text
    chunks = _chunks_for_correction(text, max_chars)
    if not chunks:
        return text
    _emit_input_status("Correcting OCR text…")
    n = len(chunks)
    emit_live_view_preview("Correcting OCR text…", tool_name="ocr_cleanup", progress_pct=0)
    workers = min(_OCR_CORRECTION_WORKERS, n)

    def _correct_one(idx_chunk: tuple[int, str]) -> tuple[int, str]:
        idx, chunk = idx_chunk
        try:
            return idx, _apply_ocr_correction_impl(chunk)
        except Exception as e:
            _debug_log(f"[apply_ocr_correction] chunk {idx + 1}/{n} failed: {e}", "WARN")
            return idx, chunk

    if workers <= 1:
        results = [""] * n
        for i, chunk in enumerate(chunks):
            emit_live_view_preview(
                f"Correcting OCR text (chunk {i + 1}/{n})…",
                tool_name="ocr_cleanup",
                progress_pct=int(100 * (i + 1) / n) if n else 100,
            )
            _, out = _correct_one((i, chunk))
            results[i] = out
    else:
        lock = threading.Lock()
        done = 0

        def _correct_one_tracked(ic: tuple[int, str]) -> tuple[int, str]:
            nonlocal done
            idx, out = _correct_one(ic)
            with lock:
                done += 1
                emit_live_view_preview(
                    f"Correcting OCR text ({done}/{n} chunks, parallel)…",
                    tool_name="ocr_cleanup",
                    progress_pct=int(100 * done / n) if n else 100,
                )
            return idx, out

        results = [""] * n
        with ThreadPoolExecutor(max_workers=workers) as ex:
            futs = [
                ex.submit(contextvars.copy_context().run, _correct_one_tracked, (i, chunks[i]))
                for i in range(n)
            ]
            for fut in as_completed(futs):
                idx, out = fut.result()
                results[idx] = out

    emit_live_view_preview("OCR correction complete", tool_name="ocr_cleanup", progress_pct=100)
    return "\n\n".join(results)


@tool
def correct_ocr_text(text: str, max_chars: int = _DEFAULT_OCR_CORRECTION_CHUNK_CHARS) -> str:
    """Correct OCR and typo errors in text using context. Preserves meaning, formatting, and footnotes. Use on a chunk of text (e.g. from read_file_chunk); then write the result with submit_ocr_chunk_corrections or write_text_file. Input: text to correct; max_chars caps input size (default ~5600 chars, aligned with correction token budget)."""
    if not (text or "").strip():
        return "No text provided to correct."
    text = (text or "").strip()
    if len(text) > max_chars:
        text = text[:max_chars] + "\n\n[Truncated for correction; correct this portion then process the next chunk.]"
    text = _rejoin_hyphenation(text)
    try:
        _emit_input_status("Correcting OCR text…")
        emit_live_view_preview("Correcting OCR text…", tool_name="correct_ocr_text", progress_pct=0)
        result = _apply_ocr_correction_impl(text)
        emit_live_view_preview("OCR correction complete", tool_name="correct_ocr_text", progress_pct=100)
        return result
    except Exception as e:
        _debug_log(f"[ERROR correct_ocr_text] {e}", "ERROR")
        return f"Correction failed: {str(e)}"


@tool
def commit_workspace_evolution(message: str) -> str:
    """Stage and commit identity/soul/memory files in the workspace (which is a git repo). Call this after you change IDENTITY.md, SOUL.md, MEMORY.md, files under memory/shards/, or other core files. No push is performed."""
    try:
        import subprocess
        from core import get_workspace_path, get_config
        from persistence import ensure_workspace_git
        workspace = get_workspace_path()
        config = get_config()
        ensure_workspace_git(workspace, config)
        git_dir = os.path.join(workspace, ".git")
        if not os.path.isdir(git_dir):
            return "Workspace is not a git repository. Enable evolution mode in Settings to initialize git in the workspace."
        existing = [f for f in _EVOLUTION_COMMIT_WHITELIST if os.path.isfile(os.path.join(workspace, f))]
        shard_paths = _evolution_shard_rel_paths(workspace)
        skill_paths = _evolution_skill_markdown_rel_paths(workspace)
        to_stage = list(dict.fromkeys(existing + shard_paths + skill_paths))
        if not to_stage:
            return "No whitelisted files exist in workspace; nothing to commit."
        try:
            result = subprocess.run(
                ["git", "add"] + to_stage,
                cwd=workspace,
                capture_output=True,
                text=True,
                timeout=30,
            )
            if result.returncode != 0:
                return f"git add failed: {result.stderr or result.stdout or 'unknown'}"
            result = subprocess.run(
                ["git", "commit", "-m", message, "--no-verify"],
                cwd=workspace,
                capture_output=True,
                text=True,
                timeout=30,
            )
            if result.returncode != 0:
                combined = (result.stderr or "") + (result.stdout or "")
                if "nothing to commit" in combined or "no changes added" in combined.lower():
                    return "No changes to commit — files are identical to the last recorded state."
                return f"git commit failed: {combined or 'unknown'}"
            listed = subprocess.run(
                ["git", "show", "--pretty=format:", "--name-only", "-1", "HEAD"],
                cwd=workspace,
                capture_output=True,
                text=True,
                timeout=30,
            )
            paths = (
                [ln.strip() for ln in (listed.stdout or "").splitlines() if ln.strip()]
                if listed.returncode == 0
                else []
            )
            if not paths:
                listed2 = subprocess.run(
                    ["git", "log", "-1", "--name-only", "--pretty=format:"],
                    cwd=workspace,
                    capture_output=True,
                    text=True,
                    timeout=30,
                )
                if listed2.returncode == 0:
                    paths = [ln.strip() for ln in (listed2.stdout or "").splitlines() if ln.strip()]
            whitelist_set = set(_EVOLUTION_COMMIT_WHITELIST)
            committed = [
                p
                for p in paths
                if p in whitelist_set
                or _is_staged_evolution_memory_shard(p)
                or _is_staged_evolution_skill_markdown(p)
            ]
            if not committed and paths:
                committed = paths
            if not committed:
                return "Commit recorded, but could not list changed paths (unexpected)."
            return f"Committed {len(committed)} file(s): {', '.join(committed)}"
        except subprocess.TimeoutExpired:
            return "git command timed out."
        except FileNotFoundError:
            return "git not found; install git to use version history."
    except Exception as e:
        _debug_log(f"[ERROR commit_workspace_evolution] {e}", "ERROR")
        return f"Commit failed: {str(e)}"


__all__ = [
    "read_file",
    "delete_file",
    "copy_file",
    "move_file",
    "FILE_CHUNK_SIZE",
    "split_text_file",
    "merge_text_files",
    "save_to_file",
    "write_text_file",
    "append_to_text_file",
    "patch_user_profile",
    "read_file_chunk",
    "submit_ocr_chunk_corrections",
    "correct_ocr_text",
    "apply_ocr_correction",
    "get_file_metadata",
    "verify_file_exists",
    "get_default_save_location",
    "suggest_save_filename",
    "commit_workspace_evolution",
]
