diff --git a/pyproject.toml b/pyproject.toml
index 83b60b6..5a5b742 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -38,6 +38,7 @@ dependencies = [
     "numpy>=1.26.0,<2.0",
     "feedparser>=6.0.0",
     "pypdf>=4.0.0",
+    "html2text>=2020.1.16",  # eager import in tools/webfetch/converter.py
     "jupyter_client>=8.0.0",
     "ipykernel>=6.0.0",
     # Durable sessions: ADK DatabaseSessionService (SQLAlchemy async) needs an
diff --git a/src/agentic_cli/file_utils.py b/src/agentic_cli/file_utils.py
index 540aa8e..aa25e75 100644
--- a/src/agentic_cli/file_utils.py
+++ b/src/agentic_cli/file_utils.py
@@ -3,6 +3,7 @@
 import fcntl
 import json
 import os
+import re
 import tempfile
 import time
 from contextlib import contextmanager
@@ -59,6 +60,34 @@ def sanitize_filename(name: str) -> str:
     return "".join(c if c.isalnum() or c in "-_" else "_" for c in name)
 
 
+def glob_pattern_escapes_root(pattern: str) -> bool:
+    """True if a glob pattern would search outside its base directory.
+
+    Rejects absolute patterns and any pattern containing a ``..`` path
+    component. ``pathlib.Path.glob`` does not normalize ``..``, so a pattern
+    like ``../*`` escapes the base directory even though the permission engine
+    only authorized that base. Callers should reject such patterns before
+    globbing.
+    """
+    if os.path.isabs(pattern):
+        return True
+    return ".." in re.split(r"[\\/]+", pattern)
+
+
+def path_is_within(path: Path, root: Path) -> bool:
+    """True if ``path`` resolves to a location inside ``root``.
+
+    Both operands are fully resolved (following symlinks) before the check, so
+    a symlink under ``root`` that points outside is correctly rejected. Returns
+    False on any resolution error (loop, permission) — fail closed.
+    """
+    try:
+        path.resolve().relative_to(root.resolve())
+        return True
+    except (OSError, ValueError, RuntimeError):
+        return False
+
+
 def _atomic_write(path: Path, content: str) -> None:
     """Write content to a file atomically and durably.
 
diff --git a/src/agentic_cli/tools/document/compile.py b/src/agentic_cli/tools/document/compile.py
index 6a86788..d2ad630 100644
--- a/src/agentic_cli/tools/document/compile.py
+++ b/src/agentic_cli/tools/document/compile.py
@@ -1,16 +1,24 @@
 """Compile a LaTeX document to PDF with a host TeX engine.
 
 ``compile_document`` is a narrow, permission-gated tool: it runs ``latexmk``
-(preferred) or ``pdflatex`` as a guarded subprocess — shell-escape is disabled
-for the **pdflatex** path (``-no-shell-escape`` flag).  For ``latexmk``, the
-engine relies on its default restricted mode; note that ``latexmk`` also reads
-``.latexmkrc`` (arbitrary Perl) from the build directory and home directory, so
-callers with an untrusted ``.tex``/build directory should prefer ``pdflatex``.
-OS-sandbox confinement for the general case is deferred.
+(preferred) or ``pdflatex`` as a guarded subprocess with the two arbitrary-code
+vectors disabled — ``latexmk`` with ``-norc`` (so no ``.latexmkrc`` Perl is read
+from the build directory or home) and ``pdflatex`` with ``-no-shell-escape`` (no
+``\\write18``).  It runs on the host (not the container sandbox — an intentional
+decoupling); OS-sandbox confinement of the build tree is deferred (spec §9).
 
-The tool runs with a wall-clock timeout and a scoped working dir, and returns a
-structured result.  It does NOT execute arbitrary code; a ``report_writer``-style
-agent uses it to turn an authored ``.tex`` into a PDF.
+The tool runs with a wall-clock timeout and a private temp build dir, and
+returns a structured result.  It does not execute arbitrary host code by
+default; a ``report_writer``-style agent uses it to turn an authored ``.tex``
+into a PDF.  Build intermediates (``*.aux``, ``*.log``) are isolated in a
+``tempfile.TemporaryDirectory`` and never written to the source or delivery dir;
+only the final PDF is promoted via ``_deliver_no_follow``.
+
+The subprocess receives only an allowlisted environment (``_ENV_PASSTHROUGH`` +
+the ``TEXMF*``/TeX config vars), not the full host environment, so host secrets
+aren't handed to the TeX process.  Delivery to ``output_pdf`` is a no-follow
+atomic write (temp file + ``os.replace``) so an attacker-placed symlink at the
+destination can't redirect the write.
 
 Provisioning is host-based: the engine must be on ``PATH`` (TeX Live / MacTeX).
 If neither is found the tool returns a structured error with an install hint.
@@ -26,6 +34,7 @@
 import shutil
 import signal
 import subprocess
+import tempfile
 import time
 from pathlib import Path
 from typing import Any
@@ -35,6 +44,98 @@
 
 _ENGINES = ("latexmk", "pdflatex")
 _LOG_TAIL_LINES = 40
+_LOG_TAIL_BYTES = 64 * 1024
+_MAX_CAPTURE_BYTES = 200_000
+# Cap the size of any single file the TeX child writes (bounds runaway-`.tex`
+# disk use, incl. the log/pdf/aux and our redirected stdout/stderr temp files).
+_RLIMIT_FSIZE_BYTES = 500 * 1024 * 1024
+
+# Only these host env vars reach the TeX process. The tool must not hand the
+# whole host environment (API keys, tokens) to a subprocess that — on the
+# latexmk path — can execute arbitrary Perl from a .latexmkrc. PATH/HOME are
+# needed for the engine binary and kpathsea; TEXINPUTS is set explicitly.
+_ENV_PASSTHROUGH = (
+    "PATH", "HOME", "TERM", "TMPDIR", "TEMP", "TMP",
+    "LANG", "LC_ALL", "LC_CTYPE", "SOURCE_DATE_EPOCH",
+    # latexmk is a Perl program; without its module path it can fail to load.
+    "PERL5LIB", "PERLLIB",
+)
+
+# TeX's own search/config vars (kpathsea) that don't fall under the TEXMF*
+# namespace. TEXINPUTS is deliberately excluded — it is set explicitly below.
+_TEX_VARS = (
+    "TEXFONTS", "TEXFORMATS", "TEXPOOL", "TEXPSHEADERS",
+    "TEXCONFIG", "TEXDOCS", "TEXSOURCES",
+)
+
+# The kpathsea TEXMF* configuration variables (exact — a strict secret boundary,
+# so a name like TEXMF_SECRET is not passed through).
+_TEXMF_VARS = (
+    "TEXMFHOME", "TEXMFVAR", "TEXMFCONFIG", "TEXMFCACHE", "TEXMFLOCAL",
+    "TEXMFDIST", "TEXMFMAIN", "TEXMFSYSVAR", "TEXMFSYSCONFIG", "TEXMFDBS",
+    "TEXMFCNF", "TEXMFOUTPUT",
+)
+
+
+def _build_env(assets_dir: str | None, source_dir: str | None = None) -> dict[str, str]:
+    """Minimal, allowlisted environment for the TeX subprocess.
+
+    Passes PATH/HOME/locale plus TeX's own configuration variables — an exact
+    list of ``TEXMF*`` vars and a fixed set of other TeX vars — so a custom
+    ``TEXMFHOME`` etc. keeps working, but not arbitrary host env (a name like
+    ``TEXT_API_TOKEN`` starts with "TEX" yet is not a TeX var), and never the
+    caller's ``TEXINPUTS`` (set explicitly below).
+
+    ``assets_dir`` and ``source_dir`` become TEXINPUTS read roots so figures and
+    ``\\input`` siblings resolve even though the build runs in a private temp dir.
+    """
+    env = {
+        k: v
+        for k, v in os.environ.items()
+        if k in _ENV_PASSTHROUGH or k in _TEXMF_VARS or k in _TEX_VARS
+    }
+    env.setdefault("PATH", os.defpath)
+    roots = [
+        str(Path(r).expanduser().resolve())
+        for r in (assets_dir, source_dir)
+        if r
+    ]
+    if roots:
+        # Trailing empty entry lets kpathsea append its default search path.
+        env["TEXINPUTS"] = os.pathsep.join(roots) + os.pathsep
+    return env
+
+
+def _deliver_no_follow(produced: Path, dest: Path) -> None:
+    """Copy ``produced`` to ``dest`` without following a symlink at ``dest``.
+
+    Writes to a private temp file in dest's directory, then atomically renames
+    it over dest. ``os.replace`` swaps the destination *name*: if dest is a
+    symlink the link itself is replaced (not written through), so an
+    attacker-placed symlink can't redirect the write outside the intended path.
+
+    This protects only the final path component. A symlinked ``dest.parent``
+    (or an ancestor) still redirects the write; the permission engine
+    canonicalizes ``output_pdf`` at check time, but a check→write window
+    remains. Full parent containment is deferred (spec §9, OS-sandbox).
+    """
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp_name = tempfile.mkstemp(
+        dir=str(dest.parent), prefix=f".{dest.name}.", suffix=".tmp"
+    )
+    tmp_path = Path(tmp_name)
+    try:
+        with os.fdopen(fd, "wb") as out, open(produced, "rb") as src:
+            shutil.copyfileobj(src, out)
+            out.flush()
+            os.fsync(out.fileno())
+        # Match the produced PDF's mode/mtime (mkstemp is 0600) so the delivered
+        # file has the readability a consumer expects, as the old copy2 did.
+        shutil.copystat(produced, tmp_path)
+        os.replace(tmp_path, dest)
+    except BaseException:
+        tmp_path.unlink(missing_ok=True)
+        raise
 
 
 def _which(name: str) -> str | None:
@@ -42,23 +143,46 @@ def _which(name: str) -> str | None:
     return shutil.which(name)
 
 
+def _child_rlimits(cpu_seconds: int):
+    """Build a ``preexec_fn`` that caps the child's file size and CPU time, so a
+    runaway ``.tex`` can't fill the disk or burn CPU within the wall-clock
+    timeout. Best-effort (a platform without ``resource`` just skips it)."""
+    def _apply() -> None:  # runs in the forked child, before exec
+        try:
+            import resource
+            resource.setrlimit(
+                resource.RLIMIT_FSIZE, (_RLIMIT_FSIZE_BYTES, _RLIMIT_FSIZE_BYTES)
+            )
+            resource.setrlimit(resource.RLIMIT_CPU, (cpu_seconds, cpu_seconds))
+        except (ValueError, OSError, ImportError):
+            pass
+    return _apply
+
+
 def _run(
     argv: list[str], *, cwd: str, env: dict[str, str], timeout: float
 ) -> subprocess.CompletedProcess:
     """Run a subprocess in its own process group so a timeout kills the whole
-    tree (latexmk + its pdflatex grandchild), not just the direct child. Seam
-    for tests. POSIX (macOS/Linux), which is what the framework targets."""
-    proc = subprocess.Popen(
-        argv, cwd=cwd, env=env,
-        stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
-        start_new_session=True,
-    )
-    try:
-        out, err = proc.communicate(timeout=timeout)
-    except subprocess.TimeoutExpired:
-        os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
-        proc.communicate()  # reap the killed group
-        raise
+    tree (latexmk + its pdflatex grandchild), not just the direct child.
+    stdout/stderr are captured to temp files and only the last
+    _MAX_CAPTURE_BYTES of each are retained, bounding host memory. The child
+    also runs under RLIMIT_FSIZE/RLIMIT_CPU limits. Seam for tests. POSIX
+    (macOS/Linux), which is what the framework targets."""
+    with tempfile.TemporaryFile() as out_f, tempfile.TemporaryFile() as err_f:
+        proc = subprocess.Popen(
+            argv, cwd=cwd, env=env,
+            stdout=out_f, stderr=err_f,
+            start_new_session=True,
+            preexec_fn=_child_rlimits(int(timeout) + 30),
+        )
+        try:
+            proc.wait(timeout=timeout)
+        except subprocess.TimeoutExpired:
+            os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
+            proc.wait()  # reap the killed group
+            raise
+        out = _tail_of_file(out_f, _MAX_CAPTURE_BYTES)
+        err = _tail_of_file(err_f, _MAX_CAPTURE_BYTES)
     return subprocess.CompletedProcess(argv, proc.returncode, stdout=out, stderr=err)
 
 
@@ -73,28 +197,70 @@ def _detect_engine(engine: str | None) -> str | None:
 
 
 def _build_argv(engine: str, source: str) -> list[str]:
-    """Compiler argv — never enables shell-escape."""
+    """Compiler argv — never enables shell-escape.
+
+    latexmk runs with ``-norc`` so it won't read ``.latexmkrc`` (arbitrary
+    Perl) from the build directory or home; pdflatex runs with
+    ``-no-shell-escape``. Neither path executes arbitrary host code by default.
+    """
     if engine == "latexmk":
-        return ["latexmk", "-pdf", "-interaction=nonstopmode", "-halt-on-error", source]
+        return [
+            "latexmk", "-norc", "-pdf", "-interaction=nonstopmode",
+            "-halt-on-error", source,
+        ]
     return [
         "pdflatex", "-no-shell-escape", "-interaction=nonstopmode",
         "-halt-on-error", source,
     ]
 
 
+def _safe_source_arg(name: str) -> str:
+    """Anchor a source filename so a leading ``-`` can't be parsed as an engine
+    option (arbitrary-exec via ``-pdflatex=CMD`` etc.). ``name`` is a basename
+    and the subprocess cwd is the source's directory, so ``./`` resolves it."""
+    return name if name.startswith("./") else f"./{name}"
+
+
 def _parse_errors(log_text: str) -> list[str]:
     """Extract LaTeX error lines (those beginning with '!') from a log."""
     return [ln for ln in log_text.splitlines() if ln.startswith("!")]
 
 
+def _tail_of_file(f, limit: int) -> str:
+    """Return the last ``limit`` bytes of an open binary temp file, decoded."""
+    size = f.seek(0, os.SEEK_END)
+    f.seek(max(0, size - limit))
+    return f.read().decode("utf-8", errors="replace")
+
+
+def _read_log_tail(log_path: Path, fallback: str) -> str:
+    """Return at most the last _LOG_TAIL_BYTES of the log (decoded), else
+    ``fallback``. Bounds memory on a runaway compiler log."""
+    try:
+        if not log_path.is_file():
+            return fallback
+        with open(log_path, "rb") as f:
+            size = f.seek(0, os.SEEK_END)
+            f.seek(max(0, size - _LOG_TAIL_BYTES))
+            return f.read().decode("utf-8", errors="replace")
+    except OSError:
+        return fallback
+
+
 @register_tool(
     category=ToolCategory.EXECUTION,
-    capabilities=[Capability("document.compile", target_arg="source_path")],
+    capabilities=[
+        Capability("document.compile", target_arg="source_path"),
+        # The tool also reads assets_dir and writes output_pdf when those are
+        # supplied; scope them explicitly (optional → not exercised when absent).
+        Capability("filesystem.read", target_arg="assets_dir", optional=True),
+        Capability("filesystem.write", target_arg="output_pdf", optional=True),
+    ],
     description=(
         "Compile a LaTeX source file to PDF using a host TeX engine (latexmk or "
-        "pdflatex). Shell-escape is disabled for pdflatex (-no-shell-escape); "
-        "latexmk uses its default restricted mode but also reads .latexmkrc from "
-        "the build/home directory. Returns the PDF path plus any compiler errors. "
+        "pdflatex). Runs on the host: latexmk with -norc (no .latexmkrc) and "
+        "pdflatex with -no-shell-escape, so it does not execute arbitrary host "
+        "code by default. Returns the PDF path plus any compiler errors. "
         "Requires TeX Live/MacTeX on PATH."
     ),
 )
@@ -110,7 +276,7 @@ def compile_document(
     Args:
         source_path: Path to the .tex file to compile.
         output_pdf: If set, the produced PDF is copied here (parents created);
-            build intermediates stay in the source's directory.
+            build intermediates are isolated in a private temp dir.
         assets_dir: Directory prepended to TEXINPUTS so figures/resources resolve
             by bare name (e.g. an artifacts dir).
         engine: Force an engine ("latexmk"/"pdflatex"); default auto-detects
@@ -129,6 +295,16 @@ def compile_document(
             "duration_ms": 0,
         }
 
+    if assets_dir and os.pathsep in assets_dir:
+        # A path-list separator would turn one authorized filesystem.read target
+        # into several TEXINPUTS search roots (e.g. "assets:/etc" also reads /etc).
+        return {
+            "success": False,
+            "error": f"assets_dir must be a single path (no {os.pathsep!r}): {assets_dir}",
+            "pdf_path": None, "engine": None, "log_tail": "", "errors": [],
+            "duration_ms": 0,
+        }
+
     chosen = _detect_engine(engine)
     if chosen is None:
         looked = engine or "/".join(_ENGINES)
@@ -142,64 +318,63 @@ def compile_document(
             "duration_ms": 0,
         }
 
-    work_dir = src.parent
-    env = dict(os.environ)
-    if assets_dir:
-        prev = env.get("TEXINPUTS", "")
-        # Prepend assets_dir; trailing empty entry preserves the default path.
-        env["TEXINPUTS"] = f"{assets_dir}{os.pathsep}{prev}{os.pathsep}"
-
-    argv = _build_argv(chosen, src.name)
+    env = _build_env(assets_dir, source_dir=str(src.parent))
+    argv = _build_argv(chosen, _safe_source_arg(src.name))
     start = time.monotonic()
-    try:
-        proc = _run(argv, cwd=str(work_dir), env=env, timeout=float(timeout_s))
-    except subprocess.TimeoutExpired:
-        return {
-            "success": False, "error": f"Compilation timed out after {timeout_s}s",
-            "pdf_path": None, "engine": chosen, "log_tail": "", "errors": [],
-            "duration_ms": int((time.monotonic() - start) * 1000),
-        }
-    except OSError as exc:
-        return {
-            "success": False, "error": f"Failed to run {chosen}: {exc}",
-            "pdf_path": None, "engine": chosen, "log_tail": "", "errors": [],
-            "duration_ms": int((time.monotonic() - start) * 1000),
-        }
-    duration_ms = int((time.monotonic() - start) * 1000)
 
-    log_path = work_dir / (src.stem + ".log")
-    try:
-        log_text = log_path.read_text(errors="replace") if log_path.is_file() else (proc.stdout or "")
-    except OSError:
-        log_text = proc.stdout or ""
-    log_tail = "\n".join(log_text.splitlines()[-_LOG_TAIL_LINES:])
-    produced = work_dir / (src.stem + ".pdf")
-    success = proc.returncode == 0 and produced.is_file()
+    with tempfile.TemporaryDirectory(prefix="texbuild-", ignore_cleanup_errors=True) as build_dir:
+        build = Path(build_dir)
+        try:
+            shutil.copy2(src, build / src.name)
+        except OSError as exc:
+            return {
+                "success": False, "error": f"Failed to stage source: {exc}",
+                "pdf_path": None, "engine": chosen, "log_tail": "", "errors": [],
+                "duration_ms": int((time.monotonic() - start) * 1000),
+            }
 
-    if not success:
-        return {
-            "success": False, "pdf_path": None, "engine": chosen,
-            "log_tail": log_tail, "errors": _parse_errors(log_text),
-            "duration_ms": duration_ms, "error": None,
-        }
+        try:
+            proc = _run(argv, cwd=str(build), env=env, timeout=float(timeout_s))
+        except subprocess.TimeoutExpired:
+            return {
+                "success": False, "error": f"Compilation timed out after {timeout_s}s",
+                "pdf_path": None, "engine": chosen, "log_tail": "", "errors": [],
+                "duration_ms": int((time.monotonic() - start) * 1000),
+            }
+        except OSError as exc:
+            return {
+                "success": False, "error": f"Failed to run {chosen}: {exc}",
+                "pdf_path": None, "engine": chosen, "log_tail": "", "errors": [],
+                "duration_ms": int((time.monotonic() - start) * 1000),
+            }
+        duration_ms = int((time.monotonic() - start) * 1000)
+
+        log_path = build / (src.stem + ".log")
+        log_text = _read_log_tail(log_path, fallback=proc.stdout or "")
+        log_tail = "\n".join(log_text.splitlines()[-_LOG_TAIL_LINES:])
+        produced = build / (src.stem + ".pdf")
+        success = proc.returncode == 0 and produced.is_file()
+
+        if not success:
+            return {
+                "success": False, "pdf_path": None, "engine": chosen,
+                "log_tail": log_tail, "errors": _parse_errors(log_text),
+                "duration_ms": duration_ms, "error": None,
+            }
 
-    final = produced
-    if output_pdf:
-        dest = Path(output_pdf)
+        dest = Path(output_pdf) if output_pdf else (src.parent / (src.stem + ".pdf"))
         try:
-            dest.parent.mkdir(parents=True, exist_ok=True)
-            shutil.copy2(produced, dest)
+            _deliver_no_follow(produced, dest)
         except OSError as exc:
             return {
                 "success": False,
-                "error": f"Failed to deliver PDF to {output_pdf}: {exc}",
-                "pdf_path": str(produced), "engine": chosen,
+                "error": f"Failed to deliver PDF to {dest}: {exc}",
+                "pdf_path": None, "engine": chosen,
                 "log_tail": log_tail, "errors": [], "duration_ms": duration_ms,
             }
-        final = dest
 
     return {
-        "success": True, "pdf_path": str(final), "engine": chosen,
+        "success": True, "pdf_path": str(dest), "engine": chosen,
         "log_tail": log_tail, "errors": [], "duration_ms": duration_ms,
         "error": None,
     }
diff --git a/src/agentic_cli/tools/glob_tool.py b/src/agentic_cli/tools/glob_tool.py
index 057e03e..784acc0 100644
--- a/src/agentic_cli/tools/glob_tool.py
+++ b/src/agentic_cli/tools/glob_tool.py
@@ -9,12 +9,15 @@
 from pathlib import Path
 from typing import Any, Literal
 
+from agentic_cli.file_utils import glob_pattern_escapes_root, path_is_within
 from agentic_cli.tools.registry import (
     ToolCategory,
     register_tool,
 )
 from agentic_cli.workflow.permissions import Capability
 
+_MAX_SCAN = 10_000  # hard ceiling on matches materialized before sorting/limiting
+
 
 @register_tool(
     category=ToolCategory.READ,
@@ -72,14 +75,38 @@ def glob(
             "path": str(search_path),
         }
 
-    # Find matching files
-    matches = list(search_path.glob(pattern))
+    # The permission engine authorizes only `path`; a pattern like "../*" or an
+    # absolute pattern would escape that authorized root, so reject it.
+    if glob_pattern_escapes_root(pattern):
+        return {
+            "success": False,
+            "error": f"Pattern escapes the search root: {pattern!r}",
+            "path": str(search_path),
+        }
+
+    # Find matching files, capping how many we materialize (a pathological
+    # pattern like "**/*" over a huge tree must not exhaust memory).
+    matches = []
+    scan_truncated = False
+    for p in search_path.glob(pattern):
+        matches.append(p)
+        if len(matches) >= _MAX_SCAN:
+            scan_truncated = True
+            break
 
     # Filter results
     filtered = []
     for match in matches:
-        # Skip hidden files if not requested
-        if not include_hidden and match.name.startswith("."):
+        # Drop anything resolving outside the authorized root (e.g. a symlink
+        # inside `path` that points elsewhere) — defense in depth.
+        if not path_is_within(match, search_path):
+            continue
+
+        # Skip results with any hidden component, not just a hidden basename —
+        # a pattern like "**/*" otherwise leaks files under a dot-directory.
+        if not include_hidden and any(
+            part.startswith(".") for part in match.relative_to(search_path).parts
+        ):
             continue
 
         # Skip directories if not requested
@@ -97,7 +124,7 @@ def glob(
         filtered.sort(key=lambda p: p.stat().st_mtime, reverse=True)
 
     # Truncate if needed
-    truncated = len(filtered) > max_results
+    truncated = scan_truncated or len(filtered) > max_results
     filtered = filtered[:max_results]
 
     # Format output
diff --git a/src/agentic_cli/tools/grep_tool.py b/src/agentic_cli/tools/grep_tool.py
index 81e93c1..eb48448 100644
--- a/src/agentic_cli/tools/grep_tool.py
+++ b/src/agentic_cli/tools/grep_tool.py
@@ -5,17 +5,25 @@
 """
 
 import functools
-import re
+import os
+import signal
 import subprocess
+import tempfile
+import re
 from pathlib import Path
 from typing import Any, Literal
 
+from agentic_cli.file_utils import glob_pattern_escapes_root, path_is_within
 from agentic_cli.tools.registry import (
     ToolCategory,
     register_tool,
 )
 from agentic_cli.workflow.permissions import Capability
 
+_MAX_FILES = 10_000        # cap the number of files the Python fallback scans
+_MAX_FILE_BYTES = 5_000_000  # skip files larger than this (avoid reading whole huge files)
+_MAX_RG_OUTPUT_BYTES = 10_000_000  # cap ripgrep JSON we read into memory
+
 
 @register_tool(
     category=ToolCategory.READ,
@@ -70,6 +78,18 @@ def grep(
             "path": str(search_path),
         }
 
+    # The permission engine authorizes only `path`; a file_pattern like "../*"
+    # or an absolute pattern would escape that authorized root, so reject it.
+    if file_pattern and glob_pattern_escapes_root(file_pattern):
+        return {
+            "success": False,
+            "error": f"File pattern escapes the search root: {file_pattern!r}",
+            "matches": [],
+            "total_matches": 0,
+            "files_searched": 0,
+            "truncated": False,
+        }
+
     # Try to use ripgrep if available (faster, respects .gitignore)
     if _ripgrep_available():
         return _grep_with_ripgrep(
@@ -145,22 +165,35 @@ def _grep_with_ripgrep(
     cmd.append("--")
     cmd.append(str(path))
 
+    # Drop RIPGREP_CONFIG_PATH so a host config can't inject flags (e.g.
+    # --follow, which would make rg traverse symlinks out of the authorized
+    # root). Containment below is the backstop; this removes the vector.
+    rg_env = {k: v for k, v in os.environ.items() if k != "RIPGREP_CONFIG_PATH"}
+    # Capture rg output to a temp file and read back at most _MAX_RG_OUTPUT_BYTES
+    # so a large tree can't allocate unbounded JSON in host memory.
+    rg_truncated = False
     try:
-        result = subprocess.run(
-            cmd,
-            capture_output=True,
-            text=True,
-            timeout=30,
-        )
-    except subprocess.TimeoutExpired:
-        return {
-            "success": False,
-            "error": "Search timed out after 30 seconds",
-            "matches": [],
-            "total_matches": 0,
-            "files_searched": 0,
-            "truncated": False,
-        }
+        with tempfile.TemporaryFile(mode="w+b") as out_f:
+            proc = subprocess.Popen(
+                cmd, stdout=out_f, stderr=subprocess.DEVNULL,
+                env=rg_env, start_new_session=True,
+            )
+            try:
+                proc.wait(timeout=30)
+            except subprocess.TimeoutExpired:
+                os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
+                proc.wait()
+                return {
+                    "success": False,
+                    "error": "Search timed out after 30 seconds",
+                    "matches": [],
+                    "total_matches": 0,
+                    "files_searched": 0,
+                    "truncated": False,
+                }
+            out_f.seek(0)
+            raw = out_f.read(_MAX_RG_OUTPUT_BYTES)
+            rg_truncated = bool(out_f.read(1))  # more output than the cap remained
     except FileNotFoundError:
         # Ripgrep not found, fall back to Python
         return _grep_python(
@@ -175,6 +208,8 @@ def _grep_with_ripgrep(
             output_mode=output_mode,
         )
 
+    stdout_text = raw.decode("utf-8", errors="replace")
+
     # Parse ripgrep JSON output
     import json
 
@@ -183,7 +218,7 @@ def _grep_with_ripgrep(
     file_counts: dict[str, int] = {}
     total_matches = 0
 
-    for line in result.stdout.strip().split("\n"):
+    for line in stdout_text.strip().split("\n"):
         if not line:
             continue
         try:
@@ -194,6 +229,14 @@ def _grep_with_ripgrep(
         if data.get("type") == "match":
             match_data = data.get("data", {})
             file_path = match_data.get("path", {}).get("text", "")
+            # Drop files resolving outside the authorized root (e.g. a symlink
+            # rg followed) — parity with the Python fallback's containment. rg
+            # may report a path relative to the search root, so resolve it there.
+            if file_path:
+                fp = Path(file_path)
+                abs_fp = fp if fp.is_absolute() else (path / fp)
+                if not path_is_within(abs_fp, path):
+                    continue
             files_searched.add(file_path)
             file_counts[file_path] = file_counts.get(file_path, 0) + 1
             total_matches += 1
@@ -219,7 +262,7 @@ def _grep_with_ripgrep(
         "matches": matches,
         "total_matches": total_matches,
         "files_searched": len(files_searched),
-        "truncated": len(matches) >= max_results,
+        "truncated": rg_truncated or len(matches) >= max_results,
     }
 
 
@@ -257,25 +300,42 @@ def _grep_python(
     total_matches = 0
     file_counts: dict[str, int] = {}
 
-    # Get files to search
+    # Get files to search, capping how many we materialize.
     if path.is_file():
-        files = [path]
+        candidates = iter([path])
+    elif file_pattern:
+        candidates = path.rglob(file_pattern) if recursive else path.glob(file_pattern)
     else:
-        if file_pattern:
-            if recursive:
-                files = list(path.rglob(file_pattern))
-            else:
-                files = list(path.glob(file_pattern))
-        else:
-            if recursive:
-                files = [f for f in path.rglob("*") if f.is_file()]
-            else:
-                files = [f for f in path.iterdir() if f.is_file()]
+        candidates = path.rglob("*") if recursive else path.iterdir()
+
+    files = []
+    scan_truncated = False
+    for f in candidates:
+        # Count only files against the budget — directories must not exhaust it
+        # (otherwise dirs before the files silently drop matches).
+        if not f.is_file():
+            continue
+        files.append(f)
+        if len(files) >= _MAX_FILES:
+            scan_truncated = True
+            break
 
     for file_path in files:
         if not file_path.is_file():
             continue
 
+        # Skip files resolving outside the authorized root (e.g. a symlink
+        # under `path` pointing elsewhere) — defense in depth.
+        if not path_is_within(file_path, path):
+            continue
+
+        # Skip files that are too large to read whole (reliability bound).
+        try:
+            if file_path.stat().st_size > _MAX_FILE_BYTES:
+                continue
+        except OSError:
+            continue
+
         try:
             content = file_path.read_text()
             lines = content.splitlines()
@@ -325,5 +385,5 @@ def _grep_python(
         "matches": matches,
         "total_matches": total_matches,
         "files_searched": files_searched,
-        "truncated": total_matches > max_results,
+        "truncated": scan_truncated or total_matches > max_results,
     }
diff --git a/src/agentic_cli/workflow/adk/permission_plugin.py b/src/agentic_cli/workflow/adk/permission_plugin.py
index cfab22f..1088e10 100644
--- a/src/agentic_cli/workflow/adk/permission_plugin.py
+++ b/src/agentic_cli/workflow/adk/permission_plugin.py
@@ -5,7 +5,8 @@
 2. Unregistered MCP toolset tool → gate through the engine under a synthetic
    ``mcp`` capability (no rule → ASK).
 3. Tool has no capability declaration → deny (author error, loud).
-4. Engine absent from service registry → allow (test/dev fallback).
+4. Engine absent from service registry → fail closed (deny) when permissions
+   are enabled; allow only when permissions are disabled.
 5. Otherwise call engine.check() and return None on allow, error dict on deny.
 """
 
@@ -59,6 +60,25 @@ def _is_mcp_tool(tool: "BaseTool") -> bool:
 _MCP_TARGET_ARG = "__mcp_target__"
 
 
+def _no_engine_result(tool_name: str) -> dict | None:
+    """Return value when the permission engine is absent from the registry.
+
+    Fail closed: if permissions are enabled but no engine is wired, deny the
+    call. In production ``base_manager`` always constructs the engine, so a
+    missing engine with permissions on is a misconfiguration — allowing would
+    silently bypass all gating. Only permissions-disabled allows (None).
+    """
+    from agentic_cli.config import get_settings
+
+    if get_settings().permissions_enabled:
+        logger.warning("permission_engine_missing", tool=tool_name)
+        return {
+            "success": False,
+            "error": "Permission denied: permission engine unavailable",
+        }
+    return None
+
+
 class PermissionPlugin(BasePlugin):
     """ADK plugin: gates every tool call through :class:`PermissionEngine`."""
 
@@ -91,7 +111,7 @@ async def before_tool_callback(
 
         engine = get_service(PERMISSION_ENGINE)
         if engine is None:
-            return None  # test/dev fallback
+            return _no_engine_result(tool.name)
 
         result = await engine.check(tool.name, caps, tool_args)
         if result.allowed:
@@ -102,7 +122,7 @@ async def _check_mcp(self, tool: "BaseTool") -> dict | None:
         """Gate an MCP tool through the engine under a synthetic capability."""
         engine = get_service(PERMISSION_ENGINE)
         if engine is None:
-            return None  # test/dev fallback
+            return _no_engine_result(tool.name)
         caps = [Capability("mcp", target_arg=_MCP_TARGET_ARG)]
         result = await engine.check(tool.name, caps, {_MCP_TARGET_ARG: tool.name})
         if result.allowed:
diff --git a/src/agentic_cli/workflow/langgraph/permission_wrap.py b/src/agentic_cli/workflow/langgraph/permission_wrap.py
index 08600fa..d04d6b7 100644
--- a/src/agentic_cli/workflow/langgraph/permission_wrap.py
+++ b/src/agentic_cli/workflow/langgraph/permission_wrap.py
@@ -4,7 +4,8 @@
 Adapter check order matches ADK's PermissionPlugin:
 1. EXEMPT tool → returned unwrapped (no engine call ever).
 2. Tool has no capability declaration → wrapper returns deny dict at call time.
-3. Engine absent from service registry → wrapper runs the original tool (fallback).
+3. Engine absent from service registry → fail closed (deny) when permissions
+   are enabled; run the tool only when permissions are disabled.
 4. Otherwise call engine.check(); return on allow, deny dict on deny.
 """
 
@@ -47,7 +48,19 @@ async def _guarded(*args: Any, **kwargs: Any) -> Any:
                 "error": "Permission denied: tool has no capability declaration",
             }
         engine = get_service(PERMISSION_ENGINE)
-        if engine is not None:
+        if engine is None:
+            # Fail closed: permissions on but no engine wired is a
+            # misconfiguration (base_manager always builds one) — deny rather
+            # than run ungated. Only permissions-disabled falls through to run.
+            from agentic_cli.config import get_settings
+
+            if get_settings().permissions_enabled:
+                logger.warning("permission_engine_missing", tool=name)
+                return {
+                    "success": False,
+                    "error": "Permission denied: permission engine unavailable",
+                }
+        else:
             result = await engine.check(name, caps, kwargs)
             if not result.allowed:
                 return {"success": False, "error": f"Permission denied: {result.reason}"}
diff --git a/src/agentic_cli/workflow/permissions/capabilities.py b/src/agentic_cli/workflow/permissions/capabilities.py
index 0cb1de8..f07aa8e 100644
--- a/src/agentic_cli/workflow/permissions/capabilities.py
+++ b/src/agentic_cli/workflow/permissions/capabilities.py
@@ -21,6 +21,11 @@ class Capability:
 
     name: str                         # e.g. "filesystem.read"
     target_arg: str | None = None     # arg name holding the target; None → target "*"
+    optional: bool = False            # when the target arg is absent/empty, skip
+    #                                   this capability (the side effect isn't
+    #                                   performed) instead of resolving it to a
+    #                                   spurious target. Only for genuinely
+    #                                   optional args (e.g. an output path).
 
 
 @dataclass(frozen=True)
diff --git a/src/agentic_cli/workflow/permissions/engine.py b/src/agentic_cli/workflow/permissions/engine.py
index a1d3b8b..3f338d0 100644
--- a/src/agentic_cli/workflow/permissions/engine.py
+++ b/src/agentic_cli/workflow/permissions/engine.py
@@ -148,6 +148,11 @@ async def check(
         resolved = self._resolve(capabilities, args)
         outcomes = self._evaluate(resolved)
 
+        # No capabilities to evaluate (e.g. every cap is optional and its target
+        # arg was absent) → nothing to gate, allow.
+        if not outcomes:
+            return CheckResult(True, "no applicable capabilities")
+
         # DENY wins.
         deny_hits = [(c, r) for c, r in outcomes if r is not None and r.effect is Effect.DENY]
         if deny_hits:
@@ -175,6 +180,10 @@ def _resolve(
                 resolved.append(ResolvedCapability(cap.name, "*"))
                 continue
             value = args.get(cap.target_arg, "")
+            if cap.optional and (value is None or value == ""):
+                # Optional target not supplied → the side effect isn't performed
+                # this call, so don't resolve (and don't spuriously prompt) it.
+                continue
             matcher = get_matcher(cap.name)
             items = value if isinstance(value, (list, tuple)) else [value]
             for item in items:
diff --git a/tests/integration/test_permission_adk.py b/tests/integration/test_permission_adk.py
index 5d36ade..02ffb8e 100644
--- a/tests/integration/test_permission_adk.py
+++ b/tests/integration/test_permission_adk.py
@@ -115,7 +115,8 @@ def writer_x(path: str):
         assert result == {"success": False, "error": "Permission denied: rule: builtin/deny"}
 
     @pytest.mark.asyncio
-    async def test_engine_absent_allows(self, monkeypatch):
+    async def test_engine_absent_denies_when_permissions_enabled(self, monkeypatch):
+        """Fail closed: permissions on but no engine wired -> deny, not allow."""
         from agentic_cli.tools.registry import get_registry
         from agentic_cli.workflow.adk.permission_plugin import PermissionPlugin
 
@@ -123,19 +124,54 @@ async def test_engine_absent_allows(self, monkeypatch):
             "agentic_cli.workflow.adk.permission_plugin.get_service",
             lambda k: None,
         )
+        monkeypatch.setattr(
+            "agentic_cli.config.get_settings",
+            lambda: SimpleNamespace(permissions_enabled=True),
+        )
+        reg = get_registry()
+
+        @reg.register(
+            name="reader_y_deny",
+            capabilities=[Capability("filesystem.read", target_arg="path")],
+        )
+        def reader_y_deny(path: str):
+            return {}
+
+        plugin = PermissionPlugin()
+        result = await plugin.before_tool_callback(
+            tool=SimpleNamespace(name="reader_y_deny"),
+            tool_args={"path": "/tmp/x"},
+            tool_context=None,
+        )
+        assert result is not None and result["success"] is False
+
+    @pytest.mark.asyncio
+    async def test_engine_absent_allows_when_permissions_disabled(self, monkeypatch):
+        """Permissions off is the only case a missing engine allows."""
+        from agentic_cli.tools.registry import get_registry
+        from agentic_cli.workflow.adk.permission_plugin import PermissionPlugin
+
+        monkeypatch.setattr(
+            "agentic_cli.workflow.adk.permission_plugin.get_service",
+            lambda k: None,
+        )
+        monkeypatch.setattr(
+            "agentic_cli.config.get_settings",
+            lambda: SimpleNamespace(permissions_enabled=False),
+        )
         reg = get_registry()
 
         @reg.register(
-            name="reader_y",
+            name="reader_y_allow",
             capabilities=[Capability("filesystem.read", target_arg="path")],
         )
-        def reader_y(path: str):
+        def reader_y_allow(path: str):
             return {}
 
         plugin = PermissionPlugin()
         result = await plugin.before_tool_callback(
-            tool=SimpleNamespace(name="reader_y"),
+            tool=SimpleNamespace(name="reader_y_allow"),
             tool_args={"path": "/tmp/x"},
             tool_context=None,
         )
-        assert result is None  # fallback allow
+        assert result is None
diff --git a/tests/integration/test_permission_langgraph.py b/tests/integration/test_permission_langgraph.py
index a9c4e45..4b6765e 100644
--- a/tests/integration/test_permission_langgraph.py
+++ b/tests/integration/test_permission_langgraph.py
@@ -88,7 +88,10 @@ def write_lg(path: str):
         assert result == {"success": False, "error": "Permission denied: rule: builtin/deny"}
 
     @pytest.mark.asyncio
-    async def test_engine_absent_allows(self, monkeypatch):
+    async def test_engine_absent_denies_when_permissions_enabled(self, monkeypatch):
+        """Fail closed: permissions on but no engine wired -> deny, not run."""
+        from types import SimpleNamespace
+
         from agentic_cli.tools.registry import get_registry
         from agentic_cli.workflow.langgraph.permission_wrap import wrap_tool_for_permission
 
@@ -96,15 +99,48 @@ async def test_engine_absent_allows(self, monkeypatch):
             "agentic_cli.workflow.langgraph.permission_wrap.get_service",
             lambda k: None,
         )
+        monkeypatch.setattr(
+            "agentic_cli.config.get_settings",
+            lambda: SimpleNamespace(permissions_enabled=True),
+        )
+        reg = get_registry()
+
+        @reg.register(
+            name="read_lg_deny",
+            capabilities=[Capability("filesystem.read", target_arg="path")],
+        )
+        def read_lg_deny(path: str):
+            return {"ok": True}
+
+        wrapped = wrap_tool_for_permission(read_lg_deny)
+        result = await wrapped(path="/x")
+        assert result["success"] is False
+
+    @pytest.mark.asyncio
+    async def test_engine_absent_allows_when_permissions_disabled(self, monkeypatch):
+        """Permissions off is the only case a missing engine runs the tool."""
+        from types import SimpleNamespace
+
+        from agentic_cli.tools.registry import get_registry
+        from agentic_cli.workflow.langgraph.permission_wrap import wrap_tool_for_permission
+
+        monkeypatch.setattr(
+            "agentic_cli.workflow.langgraph.permission_wrap.get_service",
+            lambda k: None,
+        )
+        monkeypatch.setattr(
+            "agentic_cli.config.get_settings",
+            lambda: SimpleNamespace(permissions_enabled=False),
+        )
         reg = get_registry()
 
         @reg.register(
-            name="read_lg2",
+            name="read_lg_allow",
             capabilities=[Capability("filesystem.read", target_arg="path")],
         )
-        def read_lg2(path: str):
+        def read_lg_allow(path: str):
             return {"ok": True}
 
-        wrapped = wrap_tool_for_permission(read_lg2)
+        wrapped = wrap_tool_for_permission(read_lg_allow)
         result = await wrapped(path="/x")
         assert result == {"ok": True}
diff --git a/tests/permissions/test_engine.py b/tests/permissions/test_engine.py
index 6b1f030..6a5aaa5 100644
--- a/tests/permissions/test_engine.py
+++ b/tests/permissions/test_engine.py
@@ -469,3 +469,62 @@ async def test_reloaded_wildcard_rule_still_matches(self, ctx, tmp_path, monkeyp
         )
         assert result.allowed is True
         w2.request_user_input.assert_not_called()
+
+
+class TestOptionalCapability:
+    """A capability marked optional is only exercised when its target arg is
+    supplied — so a tool with an optional output/asset path doesn't prompt for a
+    write/read it isn't performing this call."""
+
+    def _engine(self, ctx):
+        return PermissionEngine(
+            settings=_stub_settings(), workflow=_stub_workflow(), ctx=ctx,
+        )
+
+    def test_optional_cap_skipped_when_arg_absent(self, ctx):
+        engine = self._engine(ctx)
+        resolved = engine._resolve(
+            [Capability("filesystem.write", target_arg="output_pdf", optional=True)],
+            {},  # output_pdf not supplied
+        )
+        assert resolved == []
+
+    def test_optional_cap_skipped_when_arg_empty(self, ctx):
+        engine = self._engine(ctx)
+        resolved = engine._resolve(
+            [Capability("filesystem.write", target_arg="output_pdf", optional=True)],
+            {"output_pdf": None},
+        )
+        assert resolved == []
+
+    def test_optional_cap_resolved_when_arg_present(self, ctx, tmp_path):
+        engine = self._engine(ctx)
+        target = str(tmp_path / "out.pdf")
+        resolved = engine._resolve(
+            [Capability("filesystem.write", target_arg="output_pdf", optional=True)],
+            {"output_pdf": target},
+        )
+        assert len(resolved) == 1
+        assert resolved[0].name == "filesystem.write"
+
+    def test_required_cap_still_resolves_when_arg_absent(self, ctx):
+        """Non-optional (default) behavior is unchanged: an absent target still
+        resolves (to be evaluated/asked), never silently skipped."""
+        engine = self._engine(ctx)
+        resolved = engine._resolve(
+            [Capability("filesystem.write", target_arg="path")],  # optional=False
+            {},
+        )
+        assert len(resolved) == 1
+
+    @pytest.mark.asyncio
+    async def test_check_allows_when_all_optional_caps_absent(self, ctx):
+        """All-optional caps with absent args resolve to [] — check() must not
+        crash (IndexError on outcomes[0]) and should allow (nothing to gate)."""
+        engine = self._engine(ctx)
+        result = await engine.check(
+            "some_tool",
+            [Capability("filesystem.write", target_arg="out", optional=True)],
+            {},
+        )
+        assert result.allowed is True
diff --git a/tests/test_grep_tool_security.py b/tests/test_grep_tool_security.py
index 75fff5a..df08b30 100644
--- a/tests/test_grep_tool_security.py
+++ b/tests/test_grep_tool_security.py
@@ -6,24 +6,32 @@
 """
 
 from pathlib import Path
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
 from agentic_cli.tools.grep_tool import grep
 
 
+class _FakeProc:
+    pid = 4321
+    returncode = 0
+
+    def wait(self, timeout=None):
+        return 0
+
+
 def _run_grep_capturing_argv(tmp_path: Path, pattern: str, rg_stdout: str = ""):
     """Call grep() forcing the ripgrep path and capture the argv built."""
     captured = {}
 
-    def fake_run(cmd, *args, **kwargs):
+    def fake_popen(cmd, *args, **kwargs):
         captured["cmd"] = cmd
-        result = MagicMock()
-        result.stdout = rg_stdout
-        result.returncode = 0
-        return result
+        out = kwargs.get("stdout")
+        if out is not None and rg_stdout:
+            out.write(rg_stdout.encode())
+        return _FakeProc()
 
     with patch("agentic_cli.tools.grep_tool._ripgrep_available", return_value=True), \
-            patch("agentic_cli.tools.grep_tool.subprocess.run", side_effect=fake_run):
+            patch("agentic_cli.tools.grep_tool.subprocess.Popen", side_effect=fake_popen):
         out = grep(pattern=pattern, path=str(tmp_path))
     return captured["cmd"], out
 
diff --git a/tests/test_packaging.py b/tests/test_packaging.py
new file mode 100644
index 0000000..5ee3b5d
--- /dev/null
+++ b/tests/test_packaging.py
@@ -0,0 +1,39 @@
+"""Packaging consistency regression pins.
+
+A clean ``pip install`` uses only pyproject.toml (not environment.yml), so an
+eagerly-imported third-party module missing from ``[project.dependencies]``
+breaks on first import. This pins the specific modules that regressed; it is
+not an exhaustive import-vs-dependency audit.
+"""
+from __future__ import annotations
+
+import tomllib
+from pathlib import Path
+
+_PYPROJECT = Path(__file__).resolve().parent.parent / "pyproject.toml"
+
+
+def _declared_dependencies() -> list[str]:
+    data = tomllib.loads(_PYPROJECT.read_text())
+    return list(data.get("project", {}).get("dependencies", []))
+
+
+def _dep_names() -> set[str]:
+    """Normalized distribution names from the dependency specifiers."""
+    names = set()
+    for spec in _declared_dependencies():
+        # Strip version/extras/markers: name is the leading run of allowed chars.
+        name = spec.split(";")[0].strip()
+        for sep in ("[", ">", "<", "=", "!", "~", " "):
+            name = name.split(sep)[0]
+        names.add(name.strip().lower())
+    return names
+
+
+def test_html2text_is_declared():
+    """converter.py imports html2text at module load (tools/webfetch/converter.py)."""
+    assert "html2text" in _dep_names(), (
+        "html2text is imported eagerly by tools/webfetch/converter.py but is not "
+        "declared in pyproject.toml [project.dependencies] — a clean pip install "
+        "breaks when the webfetch tools are imported."
+    )
diff --git a/tests/tools/test_document_compile.py b/tests/tools/test_document_compile.py
index 1a66369..cc1446d 100644
--- a/tests/tools/test_document_compile.py
+++ b/tests/tools/test_document_compile.py
@@ -1,6 +1,8 @@
 """Offline tests for compile_document — subprocess and engine lookup faked."""
 from __future__ import annotations
 
+import os
+import signal
 import subprocess
 from pathlib import Path
 
@@ -26,12 +28,13 @@ def test_missing_source_returns_error(tmp_path):
     assert r["success"] is False and "not found" in r["error"]
 
 
-def test_success_places_pdf_and_keeps_intermediates(monkeypatch, tmp_path):
+def test_success_delivers_pdf_and_isolates_intermediates(monkeypatch, tmp_path):
     _fake_engine(monkeypatch)
     tex = tmp_path / "r.tex"
     tex.write_text("\\documentclass{article}\\begin{document}hi\\end{document}")
 
     def fake_run(argv, *, cwd, env, timeout):
+        # fake_run receives the private build dir as cwd; write artifacts there
         (Path(cwd) / "r.pdf").write_bytes(b"%PDF-1.5 fake")
         (Path(cwd) / "r.log").write_text("output written on r.pdf")
         (Path(cwd) / "r.aux").write_text("\\relax")
@@ -42,9 +45,9 @@ def fake_run(argv, *, cwd, env, timeout):
     r = compile_document(str(tex), output_pdf=str(out), assets_dir=str(tmp_path / "assets"))
     assert r["success"] is True
     assert r["pdf_path"] == str(out)
-    assert out.is_file()                             # PDF promoted to delivery dir
-    assert (tmp_path / "r.aux").is_file()            # intermediates stay in build dir
-    assert not (out.parent / "r.aux").exists()       # not beside the delivered PDF
+    assert out.is_file()                              # PDF promoted to delivery dir
+    assert not (tmp_path / "r.aux").exists()          # intermediates NOT in the source dir
+    assert not (out.parent / "r.aux").exists()        # nor beside the delivered PDF
 
 
 def test_failure_parses_errors(monkeypatch, tmp_path):
@@ -139,12 +142,13 @@ def fake_run(argv, *, cwd, env, timeout):
         return subprocess.CompletedProcess(argv, 0, stdout="ok", stderr="")
 
     monkeypatch.setattr(mod, "_run", fake_run)
-    monkeypatch.setattr(mod.shutil, "copy2", lambda src, dst: (_ for _ in ()).throw(OSError("disk full")))
+    # Delivery is a no-follow atomic write; force the atomic rename to fail.
+    monkeypatch.setattr(mod.os, "replace", lambda src, dst: (_ for _ in ()).throw(OSError("disk full")))
 
     r = compile_document(str(tex), output_pdf=str(out))
     assert r["success"] is False
     assert "deliver" in r["error"].lower() or str(out) in r["error"]
-    assert r["pdf_path"] is not None   # PDF still exists in build dir
+    assert r["pdf_path"] is None   # build dir (with the PDF) is cleaned up on return
     assert "duration_ms" in r
 
 
@@ -159,8 +163,9 @@ def test_forced_unsupported_engine_rejected(monkeypatch, tmp_path):
 
 # --- Fix wave — final review (I1/I2/M1/M2/M3) ---
 
-def test_run_group_kill_on_timeout(monkeypatch):
-    """I1: _run must start a new session and kill the process group on timeout."""
+def test_run_group_kill_on_timeout(monkeypatch, tmp_path):
+    """I1 regression: _run starts a new session and kills the process group on
+    timeout (now via proc.wait, not communicate)."""
     import subprocess as _subprocess
 
     popen_kwargs: dict = {}
@@ -172,20 +177,304 @@ class FakePopen:
         def __init__(self, argv, **kwargs):
             popen_kwargs.update(kwargs)
 
-        def communicate(self, timeout=None):
+        def wait(self, timeout=None):
             if timeout is not None:
                 raise _subprocess.TimeoutExpired([], timeout)
-            # reap call after kill
-            return ("", "")
+            return 0
 
     monkeypatch.setattr(mod.subprocess, "Popen", FakePopen)
     monkeypatch.setattr(mod.os, "getpgid", lambda pid: pid)
     monkeypatch.setattr(mod.os, "killpg", lambda pgid, sig: killpg_calls.append((pgid, sig)))
 
     try:
-        mod._run(["latexmk"], cwd="/tmp", env={}, timeout=1.0)
+        mod._run(["latexmk"], cwd=str(tmp_path), env={}, timeout=1.0)
     except _subprocess.TimeoutExpired:
         pass  # expected
 
-    assert popen_kwargs.get("start_new_session") is True, "Popen must use start_new_session=True"
-    assert len(killpg_calls) >= 1, "os.killpg must be called on timeout"
+    assert popen_kwargs.get("start_new_session") is True
+    assert len(killpg_calls) >= 1
+    assert killpg_calls[0][1] == signal.SIGKILL
+
+
+def test_run_caps_captured_output(tmp_path):
+    import os as _os
+    import sys as _sys
+
+    argv = [_sys.executable, "-c", "print('x' * 1_000_000)"]
+    r = mod._run(argv, cwd=str(tmp_path), env={"PATH": _os.environ.get("PATH", "")}, timeout=30)
+    assert r.returncode == 0
+    assert len(r.stdout) <= mod._MAX_CAPTURE_BYTES + 1  # bounded (byte cap + trailing newline)
+    assert r.stdout.rstrip().endswith("x")              # tail retained
+
+
+# --- P0-3 hardening: env allowlist, no-follow delivery, capability scope ---
+
+def test_env_is_allowlisted_not_full_environ(monkeypatch, tmp_path):
+    """Env inheritance leak: the TeX process must not receive host secrets;
+    PATH is preserved and assets_dir still reaches TEXINPUTS."""
+    _fake_engine(monkeypatch)
+    monkeypatch.setenv("MY_SECRET_TOKEN", "sk-must-not-leak")
+    monkeypatch.setenv("PATH", "/custom/bin")
+    monkeypatch.setenv("PERL5LIB", "/opt/perl/lib")  # latexmk (Perl) needs this
+    tex = tmp_path / "r.tex"; tex.write_text("x")
+    captured = {}
+
+    def fake_run(argv, *, cwd, env, timeout):
+        captured["env"] = env
+        (Path(cwd) / "r.pdf").write_bytes(b"%PDF")
+        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")
+
+    monkeypatch.setattr(mod, "_run", fake_run)
+    compile_document(str(tex), assets_dir="/tmp/assets")
+    assert "MY_SECRET_TOKEN" not in captured["env"]
+    assert captured["env"]["PATH"] == "/custom/bin"
+    assert captured["env"]["PERL5LIB"] == "/opt/perl/lib"
+    assert "/tmp/assets" in captured["env"]["TEXINPUTS"]
+
+
+def test_delivery_does_not_follow_output_symlink(monkeypatch, tmp_path):
+    """output_pdf may be an attacker-placed symlink; delivery must not write
+    through it to the link target."""
+    _fake_engine(monkeypatch)
+    tex = tmp_path / "r.tex"; tex.write_text("x")
+    outside = tmp_path / "outside.pdf"; outside.write_bytes(b"ORIGINAL")
+    deliver = tmp_path / "deliver"; deliver.mkdir()
+    link = deliver / "report.pdf"; link.symlink_to(outside)
+
+    def fake_run(argv, *, cwd, env, timeout):
+        (Path(cwd) / "r.pdf").write_bytes(b"%PDF-NEW")
+        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")
+
+    monkeypatch.setattr(mod, "_run", fake_run)
+    r = compile_document(str(tex), output_pdf=str(link))
+    assert r["success"] is True
+    assert outside.read_bytes() == b"ORIGINAL"   # link target NOT overwritten
+    assert link.read_bytes() == b"%PDF-NEW"      # PDF delivered to the path
+    assert not link.is_symlink()                 # symlink replaced by a real file
+
+
+def test_capabilities_scope_assets_and_output():
+    """document.compile alone under-authorizes: reading assets_dir and writing
+    output_pdf need explicit (optional) filesystem capabilities."""
+    from agentic_cli.tools.registry import get_registry
+
+    defn = get_registry().get("compile_document")
+    caps = {(c.name, c.target_arg, c.optional) for c in defn.capabilities}
+    assert ("document.compile", "source_path", False) in caps
+    assert ("filesystem.read", "assets_dir", True) in caps
+    assert ("filesystem.write", "output_pdf", True) in caps
+
+
+def test_env_passes_tex_config_but_not_texinputs(monkeypatch, tmp_path):
+    """TeX's own TEX* config vars pass through (so a custom TEXMFHOME works),
+    but a caller-inherited TEXINPUTS is dropped in favor of our controlled one."""
+    _fake_engine(monkeypatch)
+    monkeypatch.setenv("TEXMFHOME", "/home/user/texmf")
+    monkeypatch.setenv("TEXINPUTS", "/evil/inputs")
+    tex = tmp_path / "r.tex"; tex.write_text("x")
+    captured = {}
+
+    def fake_run(argv, *, cwd, env, timeout):
+        captured["env"] = env
+        (Path(cwd) / "r.pdf").write_bytes(b"%PDF")
+        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")
+
+    monkeypatch.setattr(mod, "_run", fake_run)
+    compile_document(str(tex), assets_dir="/tmp/assets")
+    assert captured["env"].get("TEXMFHOME") == "/home/user/texmf"
+    assert "/evil/inputs" not in captured["env"]["TEXINPUTS"]
+    assert "/tmp/assets" in captured["env"]["TEXINPUTS"]
+
+
+def test_delivery_preserves_readable_mode(monkeypatch, tmp_path):
+    """Delivered PDF keeps the produced file's mode (copystat), not the
+    mkstemp default 0600 — a report is not a secret and consumers expect it
+    readable."""
+    import os as _os
+
+    _fake_engine(monkeypatch)
+    tex = tmp_path / "r.tex"; tex.write_text("x")
+    out = tmp_path / "deliver" / "report.pdf"
+
+    def fake_run(argv, *, cwd, env, timeout):
+        p = Path(cwd) / "r.pdf"
+        p.write_bytes(b"%PDF")
+        _os.chmod(p, 0o644)
+        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")
+
+    monkeypatch.setattr(mod, "_run", fake_run)
+    r = compile_document(str(tex), output_pdf=str(out))
+    assert r["success"] is True
+    assert out.stat().st_mode & 0o044  # group/other readable, not 0600
+
+
+# --- P0-3 re-review: env scope, assets_dir separator, latexmk -norc ---
+
+def test_env_excludes_nontex_vars_starting_with_tex(monkeypatch, tmp_path):
+    """`k.startswith('TEX')` is too broad — TEXT_API_TOKEN etc. must NOT leak;
+    only real TeX vars (TEXMF*/known) pass."""
+    _fake_engine(monkeypatch)
+    monkeypatch.setenv("TEXT_API_TOKEN", "sk-must-not-leak")
+    monkeypatch.setenv("TEXMFHOME", "/home/u/texmf")
+    tex = tmp_path / "r.tex"; tex.write_text("x")
+    captured = {}
+
+    def fake_run(argv, *, cwd, env, timeout):
+        captured["env"] = env
+        (Path(cwd) / "r.pdf").write_bytes(b"%PDF")
+        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")
+
+    monkeypatch.setattr(mod, "_run", fake_run)
+    compile_document(str(tex))
+    assert "TEXT_API_TOKEN" not in captured["env"]
+    assert captured["env"].get("TEXMFHOME") == "/home/u/texmf"
+
+
+def test_env_texmf_uses_exact_allowlist_not_prefix(monkeypatch, tmp_path):
+    """A real TEXMF var passes; a TEXMF-prefixed non-var (potential secret) does not."""
+    _fake_engine(monkeypatch)
+    monkeypatch.setenv("TEXMFHOME", "/home/u/texmf")
+    monkeypatch.setenv("TEXMF_SECRET", "leak")
+    tex = tmp_path / "r.tex"; tex.write_text("x")
+    captured = {}
+
+    def fake_run(argv, *, cwd, env, timeout):
+        captured["env"] = env
+        (Path(cwd) / "r.pdf").write_bytes(b"%PDF")
+        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")
+
+    monkeypatch.setattr(mod, "_run", fake_run)
+    compile_document(str(tex))
+    assert captured["env"].get("TEXMFHOME") == "/home/u/texmf"
+    assert "TEXMF_SECRET" not in captured["env"]
+
+
+def test_assets_dir_with_path_separator_rejected(monkeypatch, tmp_path):
+    """assets_dir authorized as one filesystem path must not smuggle extra
+    TEXINPUTS roots via os.pathsep (e.g. 'assets:/etc')."""
+    import os as _os
+
+    _fake_engine(monkeypatch)
+    tex = tmp_path / "r.tex"; tex.write_text("x")
+    r = compile_document(str(tex), assets_dir=f"assets{_os.pathsep}/etc")
+    assert r["success"] is False
+    assert "assets_dir" in r["error"].lower()
+
+
+def test_latexmk_uses_norc(monkeypatch, tmp_path):
+    """latexmk must run with -norc so a build-dir/home .latexmkrc (arbitrary
+    Perl) is not executed."""
+    _fake_engine(monkeypatch)
+    tex = tmp_path / "r.tex"; tex.write_text("x")
+    captured = {}
+
+    def fake_run(argv, *, cwd, env, timeout):
+        captured["argv"] = argv
+        (Path(cwd) / "r.pdf").write_bytes(b"%PDF")
+        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")
+
+    monkeypatch.setattr(mod, "_run", fake_run)
+    compile_document(str(tex))
+    assert "-norc" in captured["argv"]
+
+
+def test_option_like_source_name_not_treated_as_flag(monkeypatch, tmp_path):
+    """A source basename starting with '-' must be anchored (./) so the engine
+    parses it as a file, not an option — otherwise '-pdflatex=CMD.tex' et al.
+    execute arbitrary host commands (which -norc does NOT prevent)."""
+    _fake_engine(monkeypatch)
+    tex = tmp_path / "-pdflatex=evil.tex"
+    tex.write_text("x")
+    captured = {}
+
+    def fake_run(argv, *, cwd, env, timeout):
+        captured["argv"] = argv
+        (Path(cwd) / (tex.stem + ".pdf")).write_bytes(b"%PDF")
+        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")
+
+    monkeypatch.setattr(mod, "_run", fake_run)
+    r = compile_document(str(tex))
+    assert "-pdflatex=evil.tex" not in captured["argv"]        # never a bare option-like token
+    assert "./-pdflatex=evil.tex" in captured["argv"]          # anchored as a path
+    assert r["success"] is True
+
+
+def test_default_delivery_to_source_dir_without_intermediates(monkeypatch, tmp_path):
+    """No output_pdf → PDF lands at <source dir>/<stem>.pdf, but the build's
+    intermediates never touch the source dir."""
+    _fake_engine(monkeypatch)
+    tex = tmp_path / "r.tex"; tex.write_text("x")
+
+    def fake_run(argv, *, cwd, env, timeout):
+        (Path(cwd) / "r.pdf").write_bytes(b"%PDF")
+        (Path(cwd) / "r.aux").write_text("aux")
+        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")
+
+    monkeypatch.setattr(mod, "_run", fake_run)
+    r = compile_document(str(tex))                     # no output_pdf
+    assert r["success"] is True
+    assert r["pdf_path"] == str(tmp_path / "r.pdf")
+    assert (tmp_path / "r.pdf").is_file()              # delivered to source dir
+    assert not (tmp_path / "r.aux").exists()           # intermediate isolated in temp
+
+
+# --- P0-3 hardening: tail-read .log to bound memory on runaway compiler logs ---
+
+
+def test_read_log_tail_bounds_large_log(tmp_path):
+    log = tmp_path / "big.log"
+    log.write_text("START\n" + ("x" * 200_000) + "\n! Real error.\nEND\n")
+    out = mod._read_log_tail(log, fallback="FB")
+    assert "END" in out and "! Real error." in out       # tail retained
+    assert "START" not in out                              # head dropped
+    assert len(out) <= mod._LOG_TAIL_BYTES + 3             # bounded
+
+
+def test_read_log_tail_missing_returns_fallback(tmp_path):
+    assert mod._read_log_tail(tmp_path / "nope.log", fallback="FB") == "FB"
+
+
+def test_read_log_tail_oserror_returns_fallback(monkeypatch, tmp_path):
+    """An existing-but-unreadable log (open raises OSError) returns fallback, not raises."""
+    log = tmp_path / "x.log"; log.write_text("data")
+
+    def boom(*a, **k):
+        raise PermissionError("nope")
+
+    monkeypatch.setattr("builtins.open", boom)
+    assert mod._read_log_tail(log, fallback="FB") == "FB"
+
+
+def test_texinputs_roots_are_absolute_for_relative_source(monkeypatch, tmp_path):
+    """Relative source_path/assets_dir must resolve to ABSOLUTE TEXINPUTS roots
+    (the build runs in a temp dir, so relative roots would resolve there)."""
+    _fake_engine(monkeypatch)
+    monkeypatch.chdir(tmp_path)
+    (tmp_path / "assets").mkdir()
+    (tmp_path / "r.tex").write_text("x")
+    captured = {}
+
+    def fake_run(argv, *, cwd, env, timeout):
+        captured["env"] = env
+        (Path(cwd) / "r.pdf").write_bytes(b"%PDF")
+        return subprocess.CompletedProcess(argv, 0, stdout="", stderr="")
+
+    monkeypatch.setattr(mod, "_run", fake_run)
+    compile_document("r.tex", assets_dir="assets")  # relative paths
+    entries = [e for e in captured["env"]["TEXINPUTS"].split(os.pathsep) if e]
+    assert entries
+    assert all(os.path.isabs(e) for e in entries)
+    assert str((tmp_path / "assets").resolve()) in entries
+
+
+def test_run_limits_child_file_size(monkeypatch, tmp_path):
+    """A runaway child writing beyond RLIMIT_FSIZE is killed (SIGXFSZ), not
+    allowed to fill the disk."""
+    import os as _os
+    import sys as _sys
+
+    monkeypatch.setattr(mod, "_RLIMIT_FSIZE_BYTES", 4096)
+    argv = [_sys.executable, "-c", "open('big.bin','wb').write(b'x' * 1_000_000)"]
+    r = mod._run(argv, cwd=str(tmp_path), env={"PATH": _os.environ.get("PATH", "")}, timeout=30)
+    assert r.returncode != 0                              # killed by the file-size limit
+    assert (tmp_path / "big.bin").stat().st_size <= 4096 * 8   # capped, not 1MB
diff --git a/tests/tools/test_document_compile_latex.py b/tests/tools/test_document_compile_latex.py
index 8fec1c2..5e72181 100644
--- a/tests/tools/test_document_compile_latex.py
+++ b/tests/tools/test_document_compile_latex.py
@@ -31,5 +31,7 @@ def test_compiles_minimal_document(tmp_path):
     assert r["success"] is True, r
     assert Path(r["pdf_path"]).is_file() and Path(r["pdf_path"]).stat().st_size > 0
     assert out.is_file()
-    assert (build / "r.log").is_file()               # intermediates in build dir
+    # Isolation contract: the build runs in a private temp dir (cleaned up), so
+    # intermediates never land in the source dir or beside the delivered PDF.
+    assert not (build / "r.log").exists()            # source dir stays clean
     assert not (out.parent / "r.log").exists()       # not beside delivered PDF
diff --git a/tests/tools/test_glob_grep_containment.py b/tests/tools/test_glob_grep_containment.py
new file mode 100644
index 0000000..ab09e3f
--- /dev/null
+++ b/tests/tools/test_glob_grep_containment.py
@@ -0,0 +1,253 @@
+"""P0-4: glob/grep patterns must not escape the permission-authorized root.
+
+The permission engine authorizes only the ``path`` argument. A ``pattern`` /
+``file_pattern`` of ``../*`` or an absolute path would let the tool read
+outside the granted directory. These tests pin the containment behavior.
+"""
+from __future__ import annotations
+
+import json
+import subprocess
+
+import agentic_cli.tools.glob_tool as glob_mod
+import agentic_cli.tools.grep_tool as grep_mod
+from agentic_cli.tools.glob_tool import glob
+from agentic_cli.tools.grep_tool import grep
+
+
+class _FakeProc:
+    """Stand-in for a ripgrep subprocess (grep now uses Popen + a temp file)."""
+    pid = 4321
+    returncode = 0
+
+    def wait(self, timeout=None):
+        return 0
+
+
+def test_glob_rejects_parent_escape(tmp_path):
+    root = tmp_path / "root"
+    root.mkdir()
+    (root / "inside.txt").write_text("x")
+    (tmp_path / "secret.txt").write_text("SECRET")
+
+    r = glob(pattern="../*", path=str(root))
+
+    assert r["success"] is False
+    assert "secret" not in str(r).lower()
+
+
+def test_glob_rejects_absolute_pattern(tmp_path):
+    root = tmp_path / "root"
+    root.mkdir()
+    r = glob(pattern="/etc/*", path=str(root))
+    assert r["success"] is False
+
+
+def test_glob_normal_pattern_still_works(tmp_path):
+    root = tmp_path / "root"
+    root.mkdir()
+    (root / "a.py").write_text("x")
+    (root / "b.txt").write_text("y")
+    r = glob(pattern="*.py", path=str(root))
+    assert r["success"] is True
+    assert r["files"] == ["a.py"]
+
+
+def test_glob_recursive_pattern_still_works(tmp_path):
+    root = tmp_path / "root"
+    (root / "sub").mkdir(parents=True)
+    (root / "sub" / "deep.py").write_text("x")
+    r = glob(pattern="**/*.py", path=str(root))
+    assert r["success"] is True
+    assert "sub/deep.py" in r["files"]
+
+
+def test_glob_skips_symlink_escaping_root(tmp_path):
+    root = tmp_path / "root"
+    root.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (outside / "passwd").write_text("SECRET")
+    (root / "link").symlink_to(outside, target_is_directory=True)
+
+    r = glob(pattern="link/*", path=str(root))
+
+    # Pattern itself is legal, but the symlinked result resolves outside root.
+    assert r["success"] is True
+    assert all("passwd" not in str(f) for f in r["files"])
+
+
+def test_grep_rejects_parent_escape_file_pattern(tmp_path, monkeypatch):
+    root = tmp_path / "root"
+    root.mkdir()
+    (root / "a.txt").write_text("needle")
+    (tmp_path / "secret.txt").write_text("needle SECRET")
+    # Force the Python fallback — that's the path that follows ../ literally.
+    monkeypatch.setattr(grep_mod, "_ripgrep_available", lambda: False)
+
+    r = grep(pattern="needle", path=str(root), file_pattern="../*")
+
+    assert r["success"] is False
+
+
+def test_grep_python_skips_symlink_escaping_root(tmp_path, monkeypatch):
+    root = tmp_path / "root"
+    root.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (outside / "secret.txt").write_text("needle SECRET")
+    (root / "link.txt").symlink_to(outside / "secret.txt")
+    (root / "real.txt").write_text("needle here")
+    monkeypatch.setattr(grep_mod, "_ripgrep_available", lambda: False)
+
+    r = grep(pattern="needle", path=str(root))
+
+    assert r["success"] is True
+    files = {m["file"] for m in r["matches"]}
+    # link.txt resolves to outside/secret.txt (outside root) → must be skipped,
+    # even though its own path name doesn't contain "secret".
+    assert not any("link.txt" in f for f in files)
+    assert any("real.txt" in f for f in files)
+
+
+def test_grep_ripgrep_filters_outside_root(tmp_path, monkeypatch):
+    """The ripgrep path (used when rg is installed) must apply the same
+    containment as the Python fallback — a followed symlink that rg reports
+    with an outside-root path must be dropped."""
+    root = tmp_path / "root"
+    root.mkdir()
+    (root / "real.txt").write_text("needle here")
+    monkeypatch.setattr(grep_mod, "_ripgrep_available", lambda: True)
+
+    outside = tmp_path / "outside" / "secret.txt"
+    inside = root / "real.txt"
+
+    def fake_popen(cmd, **kwargs):
+        out = kwargs["stdout"]
+        lines = "\n".join([
+            json.dumps({"type": "match", "data": {
+                "path": {"text": str(outside)}, "line_number": 1,
+                "lines": {"text": "needle SECRET\n"}}}),
+            json.dumps({"type": "match", "data": {
+                "path": {"text": str(inside)}, "line_number": 1,
+                "lines": {"text": "needle here\n"}}}),
+        ]) + "\n"
+        out.write(lines.encode())
+        return _FakeProc()
+
+    monkeypatch.setattr(grep_mod.subprocess, "Popen", fake_popen)
+    r = grep(pattern="needle", path=str(root))
+    files = {m["file"] for m in r["matches"]}
+    assert any("real.txt" in f for f in files)
+    assert not any("secret.txt" in f for f in files)
+
+
+def test_grep_ripgrep_scrubs_config_path_env(tmp_path, monkeypatch):
+    """A host RIPGREP_CONFIG_PATH could inject --follow (defeating containment);
+    it must not be inherited by the rg subprocess."""
+    root = tmp_path / "root"
+    root.mkdir()
+    monkeypatch.setenv("RIPGREP_CONFIG_PATH", "/home/user/.rgrc")
+    monkeypatch.setattr(grep_mod, "_ripgrep_available", lambda: True)
+    captured = {}
+
+    def fake_popen(cmd, **kwargs):
+        captured["env"] = kwargs.get("env")
+        return _FakeProc()
+
+    monkeypatch.setattr(grep_mod.subprocess, "Popen", fake_popen)
+    grep(pattern="x", path=str(root))
+    assert captured["env"] is not None
+    assert "RIPGREP_CONFIG_PATH" not in captured["env"]
+
+
+def test_grep_ripgrep_output_bounded(tmp_path, monkeypatch):
+    """rg output beyond _MAX_RG_OUTPUT_BYTES is not read into memory whole; the
+    result is flagged truncated."""
+    root = tmp_path / "root"
+    root.mkdir()
+    inside = root / "a.txt"
+    inside.write_text("needle")
+    monkeypatch.setattr(grep_mod, "_ripgrep_available", lambda: True)
+    monkeypatch.setattr(grep_mod, "_MAX_RG_OUTPUT_BYTES", 200)
+
+    def fake_popen(cmd, **kwargs):
+        out = kwargs["stdout"]
+        line = (json.dumps({"type": "match", "data": {
+            "path": {"text": str(inside)}, "line_number": 1,
+            "lines": {"text": "needle\n"}}}) + "\n").encode()
+        for _ in range(50):  # well over the 200-byte cap
+            out.write(line)
+        return _FakeProc()
+
+    monkeypatch.setattr(grep_mod.subprocess, "Popen", fake_popen)
+    r = grep(pattern="needle", path=str(root))
+    assert r["truncated"] is True
+
+
+def test_glob_excludes_hidden_ancestor(tmp_path):
+    """include_hidden=False must drop results with a hidden ANCESTOR, not just
+    a hidden basename (e.g. .hidden/secret.txt via **/*)."""
+    root = tmp_path / "root"
+    (root / ".hidden").mkdir(parents=True)
+    (root / ".hidden" / "secret.txt").write_text("s")
+    (root / "visible.txt").write_text("v")
+    r = glob(pattern="**/*", path=str(root), include_hidden=False)
+    assert r["success"] is True
+    assert all(".hidden" not in f for f in r["files"])
+    assert any("visible.txt" in f for f in r["files"])
+
+
+def test_glob_caps_scanned_matches(tmp_path, monkeypatch):
+    root = tmp_path / "root"
+    root.mkdir()
+    for i in range(6):
+        (root / f"f{i}.txt").write_text("x")
+    monkeypatch.setattr(glob_mod, "_MAX_SCAN", 3)
+    r = glob(pattern="*", path=str(root), max_results=100)
+    assert r["success"] is True
+    assert len(r["files"]) == 3   # exactly the ceiling (6 files, all pass filters)
+    assert r["truncated"] is True
+
+
+def test_grep_python_skips_oversized_files(tmp_path, monkeypatch):
+    root = tmp_path / "root"
+    root.mkdir()
+    (root / "small.txt").write_text("needle here")
+    (root / "big.txt").write_text("needle " + ("x" * 1000))
+    monkeypatch.setattr(grep_mod, "_ripgrep_available", lambda: False)
+    monkeypatch.setattr(grep_mod, "_MAX_FILE_BYTES", 100)
+    r = grep(pattern="needle", path=str(root))
+    files = {m["file"] for m in r["matches"]}
+    assert any("small.txt" in f for f in files)
+    assert not any("big.txt" in f for f in files)   # oversized file skipped
+
+
+def test_grep_python_reports_truncated_when_file_cap_hit(tmp_path, monkeypatch):
+    """Hitting the _MAX_FILES scan ceiling must be reported as truncated, not
+    silently dropped (else a partial result claims to be complete)."""
+    root = tmp_path / "root"
+    root.mkdir()
+    for i in range(3):
+        (root / f"f{i}.txt").write_text("needle")
+    monkeypatch.setattr(grep_mod, "_ripgrep_available", lambda: False)
+    monkeypatch.setattr(grep_mod, "_MAX_FILES", 2)  # fewer than the 3 files
+    r = grep(pattern="needle", path=str(root))
+    assert r["truncated"] is True
+
+
+def test_grep_python_directories_do_not_consume_file_budget(tmp_path, monkeypatch):
+    """Directories must not count against _MAX_FILES — otherwise dirs before the
+    files exhaust the budget and matches are silently missed."""
+    root = tmp_path / "root"
+    root.mkdir()
+    for i in range(20):
+        (root / f"dir{i}").mkdir()
+    (root / "a.txt").write_text("needle")
+    (root / "b.txt").write_text("needle")
+    monkeypatch.setattr(grep_mod, "_ripgrep_available", lambda: False)
+    monkeypatch.setattr(grep_mod, "_MAX_FILES", 2)  # exactly the 2 real files
+    r = grep(pattern="needle", path=str(root))
+    files = {m["file"] for m in r["matches"]}
+    assert any("a.txt" in f for f in files)
+    assert any("b.txt" in f for f in files)
diff --git a/tests/workflow/test_adk_mcp_permissions.py b/tests/workflow/test_adk_mcp_permissions.py
index 00e943f..7fd8e04 100644
--- a/tests/workflow/test_adk_mcp_permissions.py
+++ b/tests/workflow/test_adk_mcp_permissions.py
@@ -91,8 +91,31 @@ async def test_no_rule_asks_user_and_denies(self, tmp_path):
         res = await _check(eng, _MCPTool("notion_search"))
         assert res is not None and res["success"] is False
 
-    async def test_engine_absent_allows(self, tmp_path):
-        # No engine in the registry -> test/dev fallback allows.
+    async def test_engine_absent_denies_when_permissions_enabled(self, tmp_path, monkeypatch):
+        # No engine but permissions enabled -> fail closed (deny), not allow.
+        from types import SimpleNamespace
+
+        monkeypatch.setattr(
+            "agentic_cli.config.get_settings",
+            lambda: SimpleNamespace(permissions_enabled=True),
+        )
+        token = set_service_registry({})
+        try:
+            res = await PermissionPlugin().before_tool_callback(
+                tool=_MCPTool("notion_search"), tool_args={}, tool_context=None
+            )
+        finally:
+            token.var.reset(token)
+        assert res is not None and res["success"] is False
+
+    async def test_engine_absent_allows_when_permissions_disabled(self, tmp_path, monkeypatch):
+        # Permissions off is the only case a missing engine allows.
+        from types import SimpleNamespace
+
+        monkeypatch.setattr(
+            "agentic_cli.config.get_settings",
+            lambda: SimpleNamespace(permissions_enabled=False),
+        )
         token = set_service_registry({})
         try:
             res = await PermissionPlugin().before_tool_callback(