filesystem.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/tools/filesystem.py
 
"""
Filesystem tools — all operations are clamped to the working directory.
 
Purpose
-------
Provides :func:`list_dir`, :func:`read_file`, :func:`write_file`,
:func:`edit_file`, :func:`read_chunks`, and :func:`search_files` as both
Python callables and OpenAI tool schemas. Every path is resolved relative
to the project root and validated to prevent escaping the sandbox.
 
Responsibilities
----------------
- :func:`_safe_path` — resolve and reject paths that escape ``ROOT``.
- :func:`_enforce_write_gates` — the single shared write-side M8 gate
  (sandbox clamp, hard-deny protected-write, human-confirm tier) both
  :func:`write_file` and :func:`edit_file` route through.
- :func:`list_dir` — recursive directory walk with depth limit and
  hidden-file filter.
- :func:`read_file`, :func:`read_chunks` — UTF-8 reads with offset
  pagination and natural-boundary lookahead.
- :func:`write_file` — whole-file overwrite/create with informational
  chunk metadata.
- :func:`edit_file` — fail-closed anchored partial edit (exact
  ``old_string`` → ``new_string``); atomic, newline-preserving; the
  primary modify path (plan 171 / F20).
- :func:`search_files` — recursive grep over text files.
- Module-level ``TOOLS`` (callables) and ``SCHEMAS`` (OpenAI shapes)
  for auto-registration.
 
Diagnostics
-----------
Domain: OC_CHAT.TOOLS
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- All paths are validated by :func:`_safe_path`; escapes raise
  :class:`PermissionError` rather than silently allowing access.
- Reads/writes use UTF-8 with ``errors="replace"``; binary files are
  rejected.
- ``read_chunks`` clamps ``offset`` to ``[0, total_chars]`` and never
  reads past EOF.
- ``write_file`` chunk slicing is character-based (not byte-based)
  to handle multi-byte UTF-8 correctly.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.tool_results (ReadFileResult, WriteFileResult and ToolError typed return shapes)
- oc_chat.tools._sandbox (sandbox clamp, protected read and write classes, human-confirm tier and secret redaction)
- oc_chat.core.cancellation (lazy; cooperative Ctrl+C check in list_dir and search_files)
- oc_chat.core.resource_caps (lazy; clamps the model-supplied depth, byte and chunk limits)
 
See Also
--------
oc_chat/docs/conventions/CONTRACTS.md §3 Tool registry contract; §4 Tool return-
value conventions (read_file → §4.2, write_file → §4.3).
"""
import contextlib
import contextvars
from collections.abc import Iterator
from pathlib import Path
import os
 
from oc_diagnostics import _dbg
 
from oc_chat.core.tool_results import ReadFileResult, ToolError, WriteFileResult
from oc_chat.tools._sandbox import (
    SandboxViolation,
    get_session_sandbox_root_override,
    is_human_confirm_write_path,
    is_importable_write_path,
    is_protected_read_path,
    is_protected_write_path,
    redact_secrets,
    require_inside_sandbox,
)
 
_HARDCODED_ROOT = Path(__file__).parent.parent.parent.resolve()  # project root
ROOT = _HARDCODED_ROOT
 
#: Generic denial for the protected-read (secret) class -- no existence or
#: content disclosure (M8 F46; M8.0 WU-2).
_PROTECTED_READ_MSG = "Reading this file is not permitted: it is in the protected-read (secret) class."
_PROTECTED_WRITE_MSG = "Writing this file is not permitted: it is in the protected-write class (integrity / config-as-code / secret)."
 
 
# ── Human-confirm protected-write channel (M8.0 WU-8 / F70 + F70a) ───────────
#: The resolved write path an interactive human approved for one re-dispatch (F70).
#: A ContextVar (NOT a tool argument) the model cannot set; the react-loop sets it
#: after a y/N approval, scoped to the single re-run of write_file.
_human_approved_write: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "oc_chat_human_approved_write", default=None)
 
 
@contextlib.contextmanager
def human_confirmed_write(target: str) -> Iterator[None]:
    """Authorize exactly *target* (a resolved path string) for one write_file
    re-dispatch (F70). For the interactive react-loop only; reset on exit."""
    token = _human_approved_write.set(target)
    try:
        yield
    finally:
        _human_approved_write.reset(token)
 
 
def _plan_envelope_covers_write(target: str) -> bool:
    """F70a forward-compat seam: does a signed ``PlanAutonomyEnvelope`` cover this
    headless write? M8.0 has no envelope machinery (M8.3), so always ``False`` — a
    headless config/importable write is denied, never silently auto-allowed."""
    return False
 
 
def _make_protected_write_block(target: str, reason: str) -> dict:
    """The human-confirm-tier signal write_file returns (relayed by the react-loop;
    failed closed headless by F70a)."""
    return {
        "requires_confirmation": True,
        "gate": "protected_write",
        "path": target,
        "reason": reason,
        "message": (
            f"Writing '{target}' touches the human-confirm protected-write tier "
            # OCT-LINT: disable=no_hardcoded_secrets reason="Fixed user-facing protected-write message; no credential material (operator-ruled 2026-09-28)"
            f"({reason}): config-as-code or an importable-tree write (a write-then-import "
            f"in-process code-execution risk). A human must approve it; headless runs "
            f"fail closed (a signed PlanAutonomyEnvelope is M8.3)."
        ),
    }
 
 
def deny_protected_write(result: dict, *, note: str = "") -> dict:
    """Convert a protected-write block into a terminal ``error`` (headless / declined)."""
    if isinstance(result, dict) and result.get("requires_confirmation"):
        msg = result.get("message") or "This write requires human approval."
        return {"error": (msg + " " + note).strip()}
    return result
 
 
def _enforce_write_gates(path: str) -> "Path | dict":
    """Resolve *path* through the shared write-side M8 gate; return the target or a block.
 
    The SINGLE enforcement path every write-class tool routes through
    (:func:`write_file`, :func:`edit_file`) — one policy, never a parallel path.
    In order (M8.0 WU-5 / WU-8; §5.1.2):
 
    1. Hard-deny the protected-write class BEFORE the sandbox clamp (``.git``
       internals, the protected-read secret class, ``*.pth`` / ``sitecustomize`` /
       ``conftest`` import-exec hooks, the clean-exit marker).
    2. Clamp to the effective sandbox root (a ``..`` / sibling-prefix / symlink
       escape is refused).
    3. Re-check the RESOLVED target (catches a ``..`` / symlink into a ``.git``
       dir the literal string hid).
    4. The human-confirm protected-write TIER (config-as-code + the importable
       tree): the ``_human_approved_write`` ContextVar authorizes EXACTLY the
       approved resolved target on a react-loop re-dispatch (a divergence fails
       closed); otherwise a ``requires_confirmation`` block is returned (the
       react-loop relays a y/N; headless fails closed, F70a).
 
    Returns the resolved absolute :class:`~pathlib.Path` on success, or a
    ``{"error": ...}`` / ``{"requires_confirmation": ...}`` dict on denial / relay.
    """
    if is_protected_write_path(path):
        return {"error": _PROTECTED_WRITE_MSG}
    try:
        target = require_inside_sandbox(path, root=_sandbox_root())
    except SandboxViolation:
        # F25: name the boundary (root + resolved path + whether it exists on
        # disk) instead of the generic escape message, so a write to a sibling
        # subproject reads as "unreachable", not "absent". A secret target never
        # reaches here (hard-denied above), so a None fallback is defensive only.
        return {"error": _outside_sandbox_error(path) or "outside the sandbox"}
    # Re-check the resolved target (catches `..`/symlink into a .git dir etc.).
    if is_protected_write_path(target):
        return {"error": _PROTECTED_WRITE_MSG}
    granted = _human_approved_write.get()
    if granted is not None:
        # Only the EXACT approved target proceeds; a TOCTOU divergence fails closed.
        if str(target) != granted:
            return {"error": "The approved write path changed on re-dispatch; aborted for safety."}
    elif is_importable_write_path(target):
        return _make_protected_write_block(str(target), "importable-tree write")
    elif is_human_confirm_write_path(target):
        return _make_protected_write_block(str(target), "config-as-code")
    return target
 
 
def _safe_path(path: str) -> Path:
    """Resolve *path* against the effective sandbox root; raise if it escapes.
 
    Uses the effective sandbox root (:func:`_sandbox_root` — honouring the
    session override / ``OC_CHAT_SANDBOX_ROOT`` / worktree) and the shared
    :func:`require_inside_sandbox` ``relative_to`` check, NOT a string
    ``startswith`` on the hardcoded package ``ROOT``. The old ``startswith``
    let a sibling directory sharing the root's path prefix escape the sandbox
    (e.g. ``<root>`` vs ``<root>_evil``), and pinned reads to the package root
    while ``write_file`` already honoured the effective root — diverging the
    read and write sandboxes. M8 F4/F4a/F69/F77; M8.0 WU-1.
    """
    try:
        return require_inside_sandbox(path, _sandbox_root())
    except SandboxViolation as exc:
        raise PermissionError(
            f"Access outside working directory is not allowed: {path}"
        ) from exc
 
 
def _find_worktree_root(start: Path | None = None) -> Path | None:
    """Return nearest ancestor containing a Git ``.git`` dir or file."""
    current = (start or Path.cwd()).resolve(strict=False)
    if current.is_file():
        current = current.parent
    for candidate in (current, *current.parents):
        if (candidate / ".git").exists():
            return candidate
    return None
 
 
def _sandbox_root_info() -> tuple[Path, str]:
    """Return the effective sandbox root and its source label."""
    session_override = get_session_sandbox_root_override()
    if session_override is not None:
        return session_override, "session override"
    env = os.environ.get("OC_CHAT_SANDBOX_ROOT")
    if env:
        return Path(env).expanduser().resolve(strict=False), "OC_CHAT_SANDBOX_ROOT"
    if ROOT != _HARDCODED_ROOT:
        return ROOT, "hardcoded default"
    worktree = _find_worktree_root()
    if worktree is not None:
        return worktree, "worktree default"
    return ROOT, "hardcoded default"
 
 
def _sandbox_root() -> Path:
    """Determine the sandbox root for write-side filesystem operations."""
    root, _source = _sandbox_root_info()
    return root
 
 
def _resolve_unclamped(path: str) -> "tuple[Path, Path]":
    """Resolve *path* against the effective sandbox root WITHOUT clamping.
 
    Returns ``(abs_root, abs_candidate)`` — the resolved sandbox root and the
    resolved absolute path *path* names, whether or not it stays inside the root.
    A relative *path* is joined to the root (matching :func:`require_inside_sandbox`);
    symlinks / ``..`` are resolved via ``Path.resolve(strict=False)`` so the file
    need not exist. The counterpart of the clamped :func:`_safe_path`, used only to
    build a truthful boundary error (:func:`_outside_sandbox_error`).
    """
    abs_root = _sandbox_root().resolve(strict=False)
    candidate = Path(path)
    if not candidate.is_absolute():
        candidate = abs_root / candidate
    return abs_root, candidate.resolve(strict=False)
 
 
def _outside_sandbox_error(path: str) -> "str | None":
    """A boundary-truthful error for an out-of-sandbox *path* (F25), or ``None``.
 
    When *path* resolves OUTSIDE the effective sandbox root, return a message that
    NAMES the root + the resolved absolute path and says whether it exists on disk,
    so the model can tell "exists where you cannot reach" from "does not exist"
    instead of reading every boundary denial as nonexistence (the F25 confusion that
    produced the factually-wrong "the worktree lacks oc_diagnostics" report claim).
 
    M8 basis (design foundation V1.1): the sandbox clamps REACH — every filesystem op
    resolves through ``require_inside_sandbox(path, effective_root)`` (§5.1.2) — not
    knowledge; the T1 tier has "no determined in-process adversary" and a git-recoverable
    local blast radius (§3.2). SECRET existence/paths are withheld by the F46 protected-read
    class, which every caller checks FIRST; defensively, this returns ``None`` (disclose
    nothing, fall through to not-found) when the resolved out-of-bounds target is itself in
    the protected-read class (a symlink-hidden secret). No content is read; the existence
    probe is a bare ``os.path.exists`` on the ALREADY-RESOLVED path — no further
    out-of-bounds symlink following.
 
    Returns ``None`` for an in-bounds path (not a boundary case).
    """
    abs_root, abs_candidate = _resolve_unclamped(path)
    try:
        abs_candidate.relative_to(abs_root)
        return None  # in-bounds — not a boundary error
    except ValueError:
        pass
    if is_protected_read_path(abs_candidate):
        return None  # never disclose a secret's existence/path (M8 F46)
    prefix = "exists, but " if os.path.exists(abs_candidate) else ""
    return f"{prefix}outside the sandbox (root: {abs_root}): {abs_candidate}"
 
 
def _was_discovery_rescue(requested: str, resolved: Path) -> bool:
    """``True`` when *resolved* differs from the literal in-sandbox resolution of
    *requested* — i.e. the discovery fallback rescued a wrong/short path (F13).
 
    Lets a read result SAY resolved-via-discovery with both paths (Change 2) so a
    silent rescue stops masking a path error until write time.
    """
    _root, direct = _resolve_unclamped(requested)
    return direct != resolved
 
 
def _discover_file_path(path: str) -> Path:
    """Intelligently discover file path with fallback mechanisms.
    
    Args:
        path: Original path requested by user/tool
        
    Returns:
        Path object for the discovered file location
        
    Raises:
        FileNotFoundError: If file cannot be found in any allowed location
    """
    func = "_discover_file_path"
    
    # First try the direct path
    direct_path = _safe_path(path)
    if direct_path.exists() and not is_protected_read_path(direct_path):
        _dbg("OC_CHAT.TOOLS", func, f"found at direct path: {direct_path}", 3)
        return direct_path
    
    # Try common alternative locations within project
    alternatives = [
        path,  # Original path
        f"oc_chat/{path}",  # Within oc_chat module
        f"oc_chat/docs/{path}",  # In docs directory
        f"oc_chat/roadmap/{path}",  # In roadmap directory
        f"{path}.md",  # With .md extension
        f"oc_chat/{path}.md",  # In oc_chat with .md
        f"oc_chat/docs/{path}.md",  # In docs with .md
        f"oc_chat/roadmap/{path}.md",  # In roadmap with .md
        f"roadmap/{path}",  # Root roadmap directory (if exists)
        f"roadmap/{path}.md",  # Root roadmap directory with .md
    ]
    
    # Try fuzzy matching for common files
    base_name = Path(path).stem
    fuzzy_alternatives = [
        base_name,
        f"{base_name}.md",
        f"ROADMAP-{base_name}",
        f"ROADMAP-{base_name}.md",
        f"{base_name.replace('_', '-')}",  # Replace underscores with dashes
        f"{base_name.replace('-', '_')}",  # Replace dashes with underscores
    ]
    
    # Check all alternatives within allowed directories
    for alt in alternatives:
        try:
            alt_path = _safe_path(alt)
            if alt_path.exists() and not is_protected_read_path(alt_path):
                _dbg("OC_CHAT.TOOLS", func, f"found at alternative: {alt_path}", 3)
                return alt_path
        except PermissionError:
            # Skip paths that escape sandbox
            continue
    
    # Check fuzzy alternatives in key directories
    key_dirs = ["", "oc_chat", "oc_chat/docs", "oc_chat/roadmap", "roadmap"]
    for directory in key_dirs:
        for fuzzy_alt in fuzzy_alternatives:
            try:
                alt_path = _safe_path(f"{directory}/{fuzzy_alt}")
                if alt_path.exists() and not is_protected_read_path(alt_path):
                    _dbg("OC_CHAT.TOOLS", func, f"found via fuzzy match: {alt_path}", 3)
                    return alt_path
            except PermissionError:
                continue
    
    # If nothing found, raise descriptive error
    # M8.0 (F46c): a normalized not-found error — no attempt count (a structure leak).
    raise FileNotFoundError(f"File not found: {path}")
 
 
def _discover_dir_path(path: str) -> Path:
    """Intelligently discover directory path with fallback mechanisms.
    
    Args:
        path: Original directory path requested
        
    Returns:
        Path object for the discovered directory
        
    Raises:
        NotADirectoryError: If directory cannot be found in any allowed location
    """
    func = "_discover_dir_path"
    
    # First try the direct path
    direct_path = _safe_path(path)
    if direct_path.is_dir() and not is_protected_read_path(direct_path):
        _dbg("OC_CHAT.TOOLS", func, f"found at direct path: {direct_path}", 3)
        return direct_path
    
    # Try common alternative locations
    alternatives = [
        path,  # Original path
        f"oc_chat/{path}",  # Within oc_chat module
        f"oc_chat/docs/{path}",  # In docs directory
        f"oc_chat/roadmap/{path}",  # In roadmap directory
    ]
    
    for alt in alternatives:
        try:
            alt_path = _safe_path(alt)
            if alt_path.is_dir() and not is_protected_read_path(alt_path):
                _dbg("OC_CHAT.TOOLS", func, f"found at alternative: {alt_path}", 3)
                return alt_path
        except PermissionError:
            continue
    
    # If nothing found, raise descriptive error
    raise NotADirectoryError(f"Directory not found: {path}")  # F46c: no attempt count
 
 
def list_dir(path: str = ".", max_depth: int = 2, exclude_hidden: bool = True) -> dict:
    """List files and subdirectories within the working directory.
 
    Returns a dict keyed by relative path, each value being
    ``{"dirs": [...], "files": [...]}``.  ``max_depth=0`` lists only the
    target directory itself.
    """
    func = "list_dir"
    from oc_chat.core.resource_caps import clamp_depth
    max_depth = clamp_depth(max_depth)  # M8.0 WU-9: bound the directory-walk depth
    _dbg("OC_CHAT.TOOLS", func, f"path={path} max_depth={max_depth}", 3)
 
    # First try direct path to check if it's a file vs directory
    try:
        direct_path = _safe_path(path)
        if direct_path.exists() and not direct_path.is_dir():
            # Path exists but is not a directory - this should raise NotADirectoryError
            _dbg("OC_CHAT.TOOLS", func, f"path exists but is not directory: {path}", 2)
            raise NotADirectoryError(f"Not a directory: {path}")
    except PermissionError:
        pass  # Continue to discovery if direct path escapes sandbox
    
    try:
        target_path = _discover_dir_path(path)
    except (NotADirectoryError, PermissionError) as e:
        # F25: an out-of-sandbox dir is unreachable, not absent — name the
        # boundary. (`_safe_path` inside discovery raises PermissionError on a
        # sandbox escape; a genuinely-absent in-bounds path keeps not-found.)
        boundary = _outside_sandbox_error(path)
        if boundary is not None:
            _dbg("OC_CHAT.TOOLS", func, f"outside-sandbox list: {boundary}", 2)
            raise NotADirectoryError(boundary) from e
        _dbg("OC_CHAT.TOOLS", func, f"directory discovery failed: {e}", 2)
        raise
 
    result = {}
    base = target_path  # Define base variable for depth calculation
    from oc_chat.core.cancellation import tool_cancel_requested
    for root, dirs, files in os.walk(target_path):
        if tool_cancel_requested():  # cooperative Ctrl+C — return partial tree
            break
        if exclude_hidden:
            dirs[:]  = [d for d in dirs  if not d.startswith(".")]
            files[:] = [f for f in files if not f.startswith(".")]
 
        # Protected-read class: omit secret files + never descend credential
        # dirs (.ssh/.gnupg/.aws), regardless of exclude_hidden (M8 F46; WU-2).
        dirs[:] = [d for d in dirs if not is_protected_read_path(Path(root) / d)]
        files[:] = [f for f in files if not is_protected_read_path(Path(root) / f)]
 
        depth = len(Path(root).relative_to(base).parts)
        if depth > max_depth:
            dirs.clear()
            continue
 
        key = str(Path(root).relative_to(_sandbox_root()))
        result[key] = {"dirs": sorted(dirs), "files": sorted(files)}
 
    # Change 2 (F13/F25): echo the RESOLVED absolute target + a discovery-rescue
    # flag into the ROOT map entry (the return stays a path->{dirs,files} map, so
    # the positional `/ls` consumer and existing tests are byte-compatible; the
    # extra keys are additive metadata on the target's own entry).
    root_key = str(target_path.relative_to(_sandbox_root()))
    if root_key in result:
        result[root_key]["resolved_path"] = str(target_path)
        if _was_discovery_rescue(path, target_path):
            result[root_key]["resolved_via_discovery"] = True
            result[root_key]["requested_path"] = path
 
    return result
 
 
def read_file(path: str, max_bytes: int = 50_000, offset: int = 0) -> ReadFileResult | ToolError:
    """Read a text file within the working directory with pagination support.
 
    Returns ``{"content": ..., "truncated": bool, "total_size": int, "offset": int,
    "next_offset": int|None, "lines_count": int}`` or ``{"error": ...}``.
    
    Args:
        path: File path relative to working directory
        max_bytes: Maximum bytes to read per request (default 50_000 to avoid API truncation)
        offset: Byte offset to start reading from (default 0)
    """
    from oc_chat.core.resource_caps import clamp_bytes
    max_bytes = clamp_bytes(max_bytes)  # M8.0 WU-9a: model arg cannot exceed 5 MB
    if is_protected_read_path(path):
        return {"error": _PROTECTED_READ_MSG}
    try:
        p = _discover_file_path(path)
    except (FileNotFoundError, PermissionError) as e:
        # F25: a path OUTSIDE the sandbox is "exists where you cannot reach",
        # not "does not exist" — say so (a literal out-of-sandbox path raises
        # PermissionError from _safe_path; a genuinely-absent in-bounds path
        # raises FileNotFoundError and keeps the not-found message).
        return {"error": _outside_sandbox_error(path) or str(e)}
    if p.is_dir():
        return {"error": f"Path is a directory: {path}"}
    if is_protected_read_path(p):
        return {"error": _PROTECTED_READ_MSG}
 
    try:
        full_data = p.read_bytes()
        total_size = len(full_data)
    except Exception as e:
        return {"error": str(e)}
 
    # Binary detection: null byte in first 512 bytes
    if b"\0" in full_data[:512]:
        return {"error": "Binary file — cannot read as text."}
 
    # Handle offset and chunking. offset=0 on an empty file is fine —
    # it returns empty content, untruncated. Only reject offsets that
    # actually skip past the end of a non-empty file.
    if offset > 0 and offset >= total_size:
        return {"error": f"Offset {offset} exceeds file size {total_size}"}
    
    data = full_data[offset:offset + max_bytes]
    truncated = offset + len(data) < total_size
 
    try:
        content = data.decode("utf-8")
    except UnicodeDecodeError:
        content = data.decode("utf-8", errors="replace")
        content = "[decoded with replacement characters; source is not valid UTF-8]\n" + content
 
    # Add pagination info
    lines_count = content.count('\n') + (1 if content and not content.endswith('\n') else 0)
    next_offset = offset + len(data) if truncated else None
 
    # Secret-pattern redaction of returned content (M8 F46/F51; WU-2). Applied
    # after the pagination math, which is keyed to original-file byte offsets,
    # not the redacted display string.
    result: ReadFileResult = {
        "content": redact_secrets(content),
        "truncated": truncated,
        "total_size": total_size,
        "offset": offset,
        "next_offset": next_offset,
        "lines_count": lines_count,
        # Change 2 (F13/F25): echo the RESOLVED absolute path so a discovery
        # rescue is VISIBLE ground truth, not a silent wrong-path mask.
        "path": str(p),
    }
    if _was_discovery_rescue(path, p):
        result["resolved_via_discovery"] = True
        result["requested_path"] = path
    return result
 
 
def _normalize_newlines(content: str) -> str:
    """Normalize line endings to prevent accumulation of \r\n sequences."""
    return content.replace('\r\n', '\n').replace('\r', '\n')
 
 
def write_file(path: str, content: str, chunk_size: int = 0, max_chunks: int = 0) -> WriteFileResult | ToolError:
    """Write (or overwrite) a text file within a working directory tree.
 
    Files whose basename starts with ``_tmp`` are automatically redirected
    to ``logs/`` so that repository root stays tidy.
    
    Args:
        path: Destination path relative to working directory
        content: Text content to write
        chunk_size: Target chunk size in bytes for large content (default 0 = no chunking)
        max_chunks: Maximum number of chunks to return (default 0 = no chunking)
    """
    from datetime import datetime
    
    # Normalize line endings to prevent blank line multiplication
    content = _normalize_newlines(content)
    
    original_path = path
    redirected = False
 
    # M8.0 WU-5 (§5.1.2/G-4): hard-deny the protected-write class BEFORE the
    # sandbox clamp -- .git internals, secrets, *.pth / sitecustomize / conftest
    # import-exec hooks, provider-endpoint config, the clean-exit marker.
    if is_protected_write_path(path):
        return {"error": _PROTECTED_WRITE_MSG}
 
    # Content size validation
    content_size = len(content.encode('utf-8'))
    if content_size > 1_000_000:  # 1MB limit
        return {"error": f"Content too large: {content_size} bytes (max 1,000,000)"}
    
    # Handle chunking for large content
    chunks = []
    if chunk_size > 0 and len(content) > chunk_size:
        try:
            for i in range(0, min(max_chunks, (len(content) + chunk_size - 1) // chunk_size)):
                start = i * chunk_size
                end = min(start + chunk_size, len(content))
                chunk_content = content[start:end]
                chunks.append({
                    "chunk_index": i,
                    "content": chunk_content,
                    "size": len(chunk_content)
                })
        except Exception as e:
            return {"error": f"Chunking failed: {str(e)}"}
    
    # Redirect temporary workspace files to logs/
    if Path(path).name.startswith("_tmp"):
        path = str(Path("logs") / Path(path).name)
        redirected = True
 
    # H.2a + M8.0 WU-5/WU-8: clamp to the sandbox root, re-check the resolved
    # target for the hard-deny protected-write class, and apply the human-confirm
    # protected-write TIER (config-as-code + importable-tree) — the SAME shared
    # enforcement path edit_file routes through (no parallel policy).
    gate = _enforce_write_gates(path)
    if isinstance(gate, dict):
        return gate
    target = gate
    sandbox_root = _sandbox_root()
    # Suppress the informational chunk metadata for a human-confirm/importable
    # target (only reachable here in an approved re-dispatch), matching pre-refactor
    # behaviour where a protected human-confirm target is never chunk-tagged.
    _suppress_chunk_meta = is_importable_write_path(target) or is_human_confirm_write_path(target)
 
    target.parent.mkdir(parents=True, exist_ok=True)
 
    # Always write the COMPLETE content to the target as ONE file. (The legacy
    # "chunked" path recursively wrote separate ``*_chunk_N`` files, dropped the
    # directory of a nested target — `target.stem` — and never wrote the target
    # itself, yet reported the target as written: a real bug. ``chunk_size`` /
    # ``max_chunks`` now only yield informational split metadata; the on-disk file is
    # the complete target. A protected human-confirm target is never chunk-tagged.)
    target.write_text(content, encoding="utf-8")
    actual_path = str(target.relative_to(sandbox_root))
    chunk_meta = (
        [{"chunk_index": c["chunk_index"], "size": c["size"]} for c in chunks]
        if (chunks and not _suppress_chunk_meta) else []
    )
    return {
        "written": actual_path,
        "original_path": original_path,
        "redirected": redirected,
        "actual_path": actual_path,
        "size_bytes": content_size,
        "timestamp": datetime.now().isoformat(),
        "chunked": bool(chunk_meta),
        "chunks": chunk_meta,
    }
 
 
# ── edit_file: anchored partial edits (plan 171; F20 cure class) ─────────────
 
def _match_start_lines(text: str, anchor: str) -> list[int]:
    """1-based line numbers where each non-overlapping occurrence of *anchor* begins.
 
    Advances by ``len(anchor)`` so the count matches :meth:`str.count` /
    :meth:`str.replace` (both non-overlapping), keeping the reported line numbers
    consistent with the occurrence count the caller gated on.
    """
    lines: list[int] = []
    step = max(1, len(anchor))
    idx = text.find(anchor)
    while idx != -1:
        lines.append(text.count("\n", 0, idx) + 1)
        idx = text.find(anchor, idx + step)
    return lines
 
 
def _edit_near_miss_message(text: str, anchor: str, path: str) -> str:
    """Name the closest existing line to *anchor*'s first line + the first divergence.
 
    A 0-match re-anchor aid (plan 171): the model gets the nearest line number, a
    similarity score, and the column where its ``old_string`` first diverges from
    the file, so it can copy an exact anchor from a fresh ``read_file`` instead of
    guessing blindly.
    """
    import difflib
    anchor_first = anchor.split("\n", 1)[0]
    file_lines = text.split("\n")
    matcher = difflib.SequenceMatcher()
    matcher.set_seq2(anchor_first)
    best_i, best_ratio = -1, -1.0
    for i, line in enumerate(file_lines):
        matcher.set_seq1(line)
        ratio = matcher.ratio()
        if ratio > best_ratio:
            best_ratio, best_i = ratio, i
    if best_i < 0:
        return f"old_string not found in {path} (file is empty)."
    near = file_lines[best_i]
    col = 1
    for expected_ch, file_ch in zip(anchor_first, near):
        if expected_ch != file_ch:
            break
        col += 1
    return (
        f"old_string not found in {path}. Closest line is {best_i + 1} "
        f"(similarity {best_ratio:.0%}); first divergence at column {col}.\n"
        f"  expected: {anchor_first!r}\n"
        f"  file    : {near!r}\n"
        f"Re-anchor old_string to match the file exactly (copy from a fresh read_file)."
    )
 
 
def _edit_context_snippet(edited_text: str, replacement: str, start_line: int, radius: int = 2) -> str:
    """A ±*radius*-line, line-numbered snippet of the applied change (ground-truth echo).
 
    Spans the replacement's own line range plus *radius* lines on each side of the
    first edited line, so the caller sees WHAT changed and WHERE without a re-read
    (F16/F25 direction; 172 generalizes this echo to reads).
    """
    lines = edited_text.split("\n")
    span = replacement.count("\n")
    lo = max(0, start_line - 1 - radius)
    hi = min(len(lines), start_line + span + radius)
    return "\n".join(f"{i + 1:>5} | {lines[i]}" for i in range(lo, hi))
 
 
def _atomic_write_bytes(target: Path, data: bytes) -> None:
    """Write *data* to *target* atomically: temp file in the SAME dir + ``os.replace``.
 
    A same-directory temp keeps the final rename on one filesystem (so
    :func:`os.replace` is atomic); on any failure the temp is unlinked, so a failed
    edit never leaves a partial file — the target keeps its original bytes.
    """
    import tempfile
    target.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp_name = tempfile.mkstemp(dir=str(target.parent), prefix=".oc_edit_", suffix=".tmp")
    try:
        with os.fdopen(fd, "wb") as handle:
            handle.write(data)
        os.replace(tmp_name, target)
    except BaseException:
        with contextlib.suppress(OSError):
            os.unlink(tmp_name)
        raise
 
 
def edit_file(path: str, old_string: str, new_string: str, expected_count: int = 1) -> dict:
    """Replace an exact ``old_string`` with ``new_string`` in an EXISTING file.
 
    The anchored-edit primary modify path (plan 171 / F20): the tool-call payload
    is the anchor + replacement, not the whole file, so it does not hit the
    large-write provider-truncation class. Fail-closed always — ``old_string`` must
    occur exactly *expected_count* times (default 1) or nothing is written:
 
    * 0 matches → an error naming the closest near-miss line (first divergence) so
      the model can re-anchor.
    * more than *expected_count* → an error listing the occurrence line numbers; no
      write. (Set ``expected_count`` to the reported count to replace them all.)
    * ``old_string == new_string`` or an empty ``old_string`` → error (this tool
      never creates files; ``write_file`` owns creation).
 
    Routes through the SAME :func:`_enforce_write_gates` policy as ``write_file``
    (sandbox clamp, hard-deny protected-write, human-confirm tier). The write is
    atomic (temp + ``os.replace``) and preserves the file's existing newline style
    and UTF-8 encoding; binary files are refused. The result echoes the RESOLVED
    absolute path + the match line number(s) + a ±2-line context snippet of the
    applied change.
 
    Args:
        path: File path (relative to the working directory) to edit in place.
        old_string: The exact text to find (anchor). Must match the file byte-for-byte
            after newline normalization; include enough surrounding context to be unique.
        new_string: The replacement text.
        expected_count: Required number of occurrences of ``old_string`` (default 1).
    """
    func = "edit_file"
    from datetime import datetime
 
    # ── Argument validation (edit never creates; a no-op is an error) ───────
    if old_string == "":
        return {"error": "edit_file cannot create files: old_string is empty. Use write_file to create a file."}
    if old_string == new_string:
        return {"error": "edit_file is a no-op: old_string and new_string are identical."}
    try:
        expected_count = int(expected_count)
    except (TypeError, ValueError):
        return {"error": f"expected_count must be an integer, got {expected_count!r}."}
    if expected_count < 1:
        return {"error": f"expected_count must be >= 1, got {expected_count}."}
 
    # ── The shared write-side M8 gate (identical policy to write_file) ──────
    gate = _enforce_write_gates(path)
    if isinstance(gate, dict):
        return gate
    target = gate
    sandbox_root = _sandbox_root()
 
    # ── The file must already exist (edit is not a create path) ─────────────
    if not target.exists():
        return {"error": f"File not found: {path}. edit_file modifies existing files; use write_file to create."}
    if target.is_dir():
        return {"error": f"Path is a directory: {path}"}
 
    # ── Read + binary/encoding guards (parity with read_file) ───────────────
    try:
        raw = target.read_bytes()
    except OSError as exc:
        return {"error": str(exc)}
    if b"\0" in raw[:512]:
        return {"error": "Binary file — cannot edit as text."}
    try:
        original = raw.decode("utf-8")
    except UnicodeDecodeError:
        return {"error": "Non-UTF-8 file — cannot edit as text."}
 
    # ── Detect the file's newline style; match/apply in normalized (\n) space ─
    newline = "\r\n" if "\r\n" in original else "\n"
    norm = original.replace("\r\n", "\n").replace("\r", "\n")
    anchor = old_string.replace("\r\n", "\n").replace("\r", "\n")
    replacement = new_string.replace("\r\n", "\n").replace("\r", "\n")
 
    count = norm.count(anchor)
    if count == 0:
        return {"error": _edit_near_miss_message(norm, anchor, path)}
    if count != expected_count:
        found_lines = _match_start_lines(norm, anchor)
        return {"error": (
            f"Expected {expected_count} occurrence(s) of old_string in {path}, found {count} "
            f"(at line(s) {', '.join(map(str, found_lines))}). No write performed. Add more "
            f"surrounding context to old_string to anchor a single occurrence, or set "
            f"expected_count={count} to replace all."
        )}
 
    # ── Apply (full replace; count == expected_count guaranteed) + atomic write ─
    match_lines = _match_start_lines(norm, anchor)
    edited_norm = norm.replace(anchor, replacement)
    out_text = edited_norm.replace("\n", newline) if newline != "\n" else edited_norm
    out_bytes = out_text.encode("utf-8")
    try:
        _atomic_write_bytes(target, out_bytes)
    except OSError as exc:
        return {"error": f"Atomic write failed (file unchanged): {exc}"}
 
    actual_path = str(target.relative_to(sandbox_root))
    context = _edit_context_snippet(edited_norm, replacement, match_lines[0])
    _dbg("OC_CHAT.TOOLS", func, f"edited {actual_path}: {count} replacement(s) at {match_lines}", 3)
    return {
        "edited": actual_path,
        "path": str(target),           # RESOLVED absolute path — model-independent ground truth (F16/F25)
        "actual_path": actual_path,
        "match_line": match_lines[0],
        "match_lines": match_lines,
        "replacements": count,
        "context": context,
        "size_bytes": len(out_bytes),
        "newline": "crlf" if newline == "\r\n" else "lf",
        "timestamp": datetime.now().isoformat(),
    }
 
 
def file_info(path: str) -> dict:
    """Return file metadata without reading content."""
    func = "file_info"
    if is_protected_read_path(path):
        return {"error": _PROTECTED_READ_MSG}
    try:
        p = _discover_file_path(path)
    except (FileNotFoundError, PermissionError) as e:
        return {"error": _outside_sandbox_error(path) or str(e)}  # F25: name the boundary
    if p.is_dir():
        return {"error": f"Path is a directory: {path}"}
    if is_protected_read_path(p):
        return {"error": _PROTECTED_READ_MSG}
 
    try:
        stat = p.stat()
 
        # Detect if file is binary
        try:
            with open(p, 'rb') as f:
                binary = b'\0' in f.read(512)
        except OSError as e:
            # VALIDATION_ERROR — unreadable; treat as binary so we don't try to count lines
            _dbg("OC_CHAT.TOOLS", func, f"VALIDATION_ERROR: binary-probe: {type(e).__name__}: {e}", 3)
            binary = True
 
        # Estimate line count for text files
        lines = None
        if not binary:
            try:
                with open(p, 'r', encoding='utf-8') as f:
                    lines = sum(1 for _ in f)
            except (OSError, UnicodeDecodeError) as e:
                # VALIDATION_ERROR — encoding mismatch or read error; line count unavailable
                _dbg("OC_CHAT.TOOLS", func, f"VALIDATION_ERROR: line-count: {type(e).__name__}: {e}", 3)
                lines = None
        
        # Estimate optimal chunk size
        size = stat.st_size
        if size < 10_000:
            chunk_size = size
            chunks = 1
        else:
            chunk_size = 50_000  # Match read_file default
            chunks = (size + chunk_size - 1) // chunk_size
            
        info = {
            "size_bytes": size,
            "lines": lines,
            "binary": binary,
            "readable_chunks": chunks,
            "recommended_chunk_size": chunk_size,
            "modified_time": stat.st_mtime,
            "created_time": stat.st_ctime,
            # Change 2 (F13/F25): the RESOLVED absolute path — a discovery rescue
            # is visible, not silently masked.
            "path": str(p),
        }
        if _was_discovery_rescue(path, p):
            info["resolved_via_discovery"] = True
            info["requested_path"] = path
        return info
    except Exception as e:
        return {"error": str(e)}
 
 
def read_chunks(path: str, chunk_size: int = 10_000, max_chunks: int = 10, offset: int = 0) -> dict:
    """Read file in optimized chunks with natural boundaries.
    
    Args:
        path: File path relative to working directory
        chunk_size: Target chunk size in characters (default 10000)
        max_chunks: Maximum number of chunks to return (default 10)
        offset: Character offset to start reading from (default 0);
                pass the ``next_offset`` from a previous result to paginate.
    """
    # M8.0 WU-9 review fix (Appendix B): clamp the model-supplied chunk parameters so
    # the total output (chunk_size * max_chunks) cannot exceed the 5 MB ceiling.
    from oc_chat.core.resource_caps import MAX_BYTES, clamp_bytes
    chunk_size = max(1, clamp_bytes(chunk_size))
    try:
        max_chunks = int(max_chunks)
    except (TypeError, ValueError, OverflowError):
        max_chunks = 1
    max_chunks = max(1, min(max_chunks, MAX_BYTES // chunk_size))
    if is_protected_read_path(path):
        return {"error": _PROTECTED_READ_MSG}
    p = _safe_path(path)
    if is_protected_read_path(p):
        return {"error": _PROTECTED_READ_MSG}
    if not p.exists():
        return {"error": f"File not found: {path}"}
    if p.is_dir():
        return {"error": f"Path is a directory: {path}"}
 
    try:
        full_data = p.read_bytes()
        
        # Binary detection
        if b"\0" in full_data[:512]:
            return {"error": "Binary file — cannot read as text."}
        
        # Decode once — all offsets below are character-based on this string
        content = full_data.decode("utf-8")
        total_chars = len(content)
        
        # Clamp starting offset
        if offset < 0:
            offset = 0
        if offset > 0 and offset >= total_chars:
            return {"error": f"Offset {offset} exceeds file length {total_chars}"}
        
        # Smart chunking - try to break at natural boundaries
        chunks = []
        pos = offset
        
        for i in range(max_chunks):
            if pos >= total_chars:
                break
                
            # Calculate tentative end of this chunk (character-based)
            chunk_end = min(pos + chunk_size, total_chars)
            chunk_data = content[pos:chunk_end]
            
            # Try to extend to natural boundary if not at end
            if chunk_end < total_chars:
                # Look for newline, paragraph break, or sentence end
                lookahead = min(total_chars, chunk_end + 1000)
                next_newline = content.find('\n', chunk_end, lookahead)
                next_para = content.find('\n\n', chunk_end, lookahead)
                next_sentence = content.find('. ', chunk_end, min(total_chars, chunk_end + 500))
                
                # Choose the closest natural boundary
                candidates = [next_newline, next_para, next_sentence]
                candidates = [c for c in candidates if c != -1]
                if candidates:
                    best_boundary = min(candidates)
                    if best_boundary - chunk_end < 200:  # Don't extend too much
                        chunk_end = best_boundary + 1
                        chunk_data = content[pos:chunk_end]
            
            chunks.append({
                # Secret-pattern redaction of returned content (M8 F46/F51;
                # WU-2). size/lines stay keyed to the original chunk text.
                "content": redact_secrets(chunk_data),
                "offset": pos,
                "size": len(chunk_data),
                "lines": chunk_data.count('\n') + (1 if chunk_data and not chunk_data.endswith('\n') else 0)
            })
            
            pos = chunk_end
        
        return {
            "chunks": chunks,
            "total_size": len(full_data),   # byte size for backward compatibility
            "total_chars": total_chars,     # character count
            "total_chunks": len(chunks),
            "has_more": pos < total_chars,
            "next_offset": pos if pos < total_chars else None
        }
        
    except UnicodeDecodeError:
        return {"error": "Non-UTF-8 file."}
    except Exception as e:
        return {"error": str(e)}
 
 
def search_files(root_dir: str = ".", pattern: str = "*") -> dict:
    """Search recursively for files matching a glob pattern."""
    from oc_chat.core.cancellation import tool_cancel_requested
    root = _safe_path(root_dir)
    matches: list[str] = []
    for p in root.glob(f"**/{pattern}"):
        if tool_cancel_requested():  # cooperative Ctrl+C — return partial matches
            return {"matches": sorted(matches), "count": len(matches), "cancelled": True}
        if p.is_file():
            if is_protected_read_path(p):
                continue  # omit protected-read (secret) files (M8 F46; WU-2)
            matches.append(str(p.relative_to(_sandbox_root())))
    return {"matches": sorted(matches), "count": len(matches)}
 
 
def read_prompt_file(filepath: str, *, allow_absolute: bool = False) -> str:
    """Read a file destined for the provider-bound prompt, with M8.0 WU-2b (F73)
    enforcement.
 
    Prompt-composition reads (``system_prompt_file``, ``include_files``) happen
    before the tool gate and their content egresses to the model provider, so
    they must honour the same floor as the read tools:
 
    * **No absolute paths** -- there is no out-of-band allowlist yet, so
      ``allow_absolute`` is effectively always ``False`` (the seam for a future
      OOB-allow; F73 "unless OOB-allowed").
    * **Clamp to the effective sandbox root** (``_sandbox_root()``); a ``..`` /
      sibling-prefix / symlink escape is refused.
    * **Deny the protected-read (secret) class** on the final resolved path.
    * **Redact secret patterns** from the returned content (defence in depth).
 
    Returns the file's text, stripped + redacted. Raises :class:`PermissionError`
    on a policy denial (absolute, sandbox escape, or protected-read) and
    :class:`FileNotFoundError` when the file is missing or empty -- the callers
    fall back to the built-in default / skip the include, so a denied secret is
    never inserted into the prompt.
    """
    raw = str(filepath).strip()
    if not raw:
        raise FileNotFoundError("empty prompt-file path")
    if Path(raw).is_absolute() and not allow_absolute:
        raise PermissionError(
            f"Absolute prompt-composition paths are not allowed: {filepath}"
        )
    try:
        p = require_inside_sandbox(raw, _sandbox_root())
    except SandboxViolation as exc:
        raise PermissionError(
            f"Prompt-composition path outside the working directory is not allowed: {filepath}"
        ) from exc
    if is_protected_read_path(p):
        raise PermissionError(
            f"Reading this file for prompt composition is not permitted "
            f"(protected-read/secret class): {filepath}"
        )
    content = p.read_text(encoding="utf-8").strip()
    if not content:
        raise FileNotFoundError(f"prompt file missing or empty: {filepath}")
    return redact_secrets(content)
 
 
# ── OpenAI tool definitions ───────────────────────────────────────────────────
 
TOOLS = {
    "list_dir":     list_dir,
    "read_file":    read_file,
    "write_file":   write_file,
    "edit_file":    edit_file,
    "search_files": search_files,
    "file_info":    file_info,
    "read_chunks":  read_chunks,
}
 
SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "list_dir",
            "description": (
                "List files and subdirectories within the working directory. "
                "Returns a dict keyed by relative path, each with 'dirs' and 'files' lists."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "path":           {"type": "string",  "description": "Directory to list (relative, default '.')"},
                    "max_depth":      {"type": "integer", "description": "Recursion depth (0 = flat listing, default 2)"},
                    "exclude_hidden": {"type": "boolean", "description": "Exclude dot-files/dirs (default true)"},
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "Read the text content of a file within the working directory with pagination support. Use offset for chunked reading of large files.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path":      {"type": "string",  "description": "File path relative to working directory"},
                    "max_bytes": {"type": "integer", "description": "Maximum bytes to read per request (default 50000, reduced to avoid API truncation)"},
                    "offset":    {"type": "integer", "description": "Byte offset to start reading from (default 0)"},
                },
                "required": ["path"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "Write (or overwrite) a text file within a working directory. The complete content is always written to one file. Files starting with '_tmp' are redirected to 'logs/'. Returns detailed metadata including the actual write location.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path":    {"type": "string", "description": "Destination path relative to working directory"},
                    "content": {"type": "string", "description": "Text content to write"},
                    # OCT-LINT: disable=no_hardcoded_secrets reason="write_file tool-schema description of the chunk_size parameter; no credential material (operator-ruled 2026-09-28)"
                    "chunk_size": {"type": "integer", "description": "Optional: target chunk size (chars) for INFORMATIONAL split metadata in the response; the file is still written whole (default 0 = none)"},
                    "max_chunks": {"type": "integer", "description": "Optional: cap on the informational chunk-metadata entries (default 0 = none)"},
                },
                "required": ["path", "content"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "edit_file",
            "description": (
                "Anchored partial edit: replace an exact old_string with new_string in an "
                "EXISTING file. Preferred over write_file for MODIFYING a file because the "
                "tool-call payload is only the anchor + replacement, not the whole file — this "
                "avoids the large-write truncation failure class. Fail-closed: old_string must "
                "occur exactly expected_count times (default 1) or nothing is written (0 matches "
                "reports the closest near-miss line; >expected_count reports the match line "
                "numbers). Never creates files (use write_file). Preserves the file's newline "
                "style; the result echoes the resolved path + match line + a context snippet."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "path":           {"type": "string",  "description": "File path (relative to the working directory) to edit in place"},
                    "old_string":     {"type": "string",  "description": "Exact text to find (anchor); include enough surrounding context to be unique"},
                    "new_string":     {"type": "string",  "description": "Replacement text"},
                    "expected_count": {"type": "integer", "description": "Required number of occurrences of old_string (default 1); set to the reported count to replace all"},
                },
                "required": ["path", "old_string", "new_string"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_files",
            "description": "Search recursively for files matching a glob pattern.",
            "parameters": {
                "type": "object",
                "properties": {
                    "root_dir": {"type": "string", "description": "Search root (relative, default '.')"},
                    "pattern":  {"type": "string", "description": "Glob pattern e.g. '*.py', '*.md'"},
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "file_info",
            "description": "Get file metadata without reading content. Returns size, line count, binary detection, and chunking recommendations.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "File path relative to working directory"},
                },
                "required": ["path"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "read_chunks",
            "description": "Read file in optimized chunks with natural boundaries. Returns multiple chunks at once for efficient processing of large files.",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "File path relative to working directory"},
                    "chunk_size": {"type": "integer", "description": "Target chunk size in characters (default 10000)"},
                    "max_chunks": {"type": "integer", "description": "Maximum number of chunks to return (default 10)"},
                    "offset": {"type": "integer", "description": "Character offset to start reading from (default 0). Pass the next_offset from a previous result to paginate."},
                },
                "required": ["path"],
            },
        },
    },
]