#!/usr/bin/env python3
"""aem-agentkit-helper - deterministic helper for the aem-agentkit skill.

Reference implementation. See references/helpers.md in the skill bundle
for the full operation spec. POSIX only (Linux, macOS); Windows is
rejected at startup because the symlink-hardening contract requires
O_NOFOLLOW semantics that the Win32 API does not expose in a portable
form.

Protocol
--------
- `aem-agentkit-helper --version`       prints VERSION and exits 0.
- `aem-agentkit-helper`                  reads JSON-line requests from
  stdin until EOF, emits one JSON-line response per request to stdout.
  Exit code 0 if every request returned ok=true, 1 otherwise.

Every request is `{"op": "<name>", ...}` with op-specific fields. Every
response is `{"ok": true, ...}` or `{"ok": false, "error": "..."}`.
"""

import base64
import errno
import fcntl
import fnmatch
import hashlib
import json
import os
import re
import sys
import traceback
import unicodedata

VERSION = "1.0.0-beta"
# Protocol-version is tracked separately from skill version. Bump on op
# add / response-shape changes so the skill driver can pin the wire format
# independently of marketing version bumps. See references/helpers.md § 1.
PROTOCOL_VERSION = "2"

# --------------------------------------------------------------------- #
# Platform gate                                                         #
# --------------------------------------------------------------------- #

# Python version pin. The helper uses no 3.10-only syntax today, but the
# spec requires 3.10+ for forward-compat (PEP 604 unions, structural pattern
# matching in future tests). Fail loud rather than emit obscure SyntaxError.
if sys.version_info < (3, 10):
    sys.stderr.write(
        f"aem-agentkit-helper requires Python 3.10+; got "
        f"{sys.version_info.major}.{sys.version_info.minor}\n"
    )
    sys.exit(1)

# Platform allow-list: only Linux and Darwin support the syscall surface
# the symlink-hardening contract needs (/proc/self/fd or F_GETPATH for the
# TOCTOU re-check; O_NOFOLLOW semantics). Other POSIX variants (FreeBSD,
# OpenBSD, Solaris, AIX) silently degrade and are rejected.
_SUPPORTED_PLATFORMS = {"linux", "darwin"}
if sys.platform not in _SUPPORTED_PLATFORMS:
    sys.stderr.write(
        f"aem-agentkit: platform '{sys.platform}' is unsupported. "
        f"The symlink-hardening contract requires Linux or macOS.\n"
    )
    sys.exit(1)

# --------------------------------------------------------------------- #
# Constants from privacy-and-sanitization.md                            #
# --------------------------------------------------------------------- #

# Unicode code points to strip. See references/privacy-and-sanitization.md
# § 2.1 for the source list. Each block is named so additions stay auditable.
# C0/C1 controls except TAB; line/paragraph separators (Unicode Cc/Cf
# categories); soft hyphen / Mongolian vowel separator; zero-width set
# (U+200B-U+200F covers ZWSP, ZWNJ, ZWJ, LRM, RLM); WORD JOINER, BOM,
# REPLACEMENT CHARACTER; Arabic Letter Mark; bidi overrides (LRE/RLE/PDF/
# LRO/RLO at U+202A-U+202E and isolates LRI/RLI/FSI/PDI at U+2066-U+2069).
_STRIP_CODEPOINTS = (
    list(range(0x00, 0x09)) + list(range(0x0A, 0x20))     # C0 controls except \t (0x09)
    + [0x2028, 0x2029]                                      # LINE SEPARATOR, PARAGRAPH SEPARATOR
    + [0x00AD, 0x180E]                                      # SOFT HYPHEN, MONGOLIAN VOWEL SEP
    + list(range(0x200B, 0x2010))                           # zero-width / directional marks U+200B..U+200F
    + [0x2060, 0xFEFF, 0xFFFD]                              # WORD JOINER, BOM, REPLACEMENT CHAR
    + [0x061C]                                              # ARABIC LETTER MARK
    + list(range(0x202A, 0x202F))                           # bidi overrides U+202A..U+202E
    + list(range(0x2066, 0x206A))                           # bidi isolates U+2066..U+2069
)
STRIP_SET = frozenset(_STRIP_CODEPOINTS)

# Variant of STRIP_SET for whole-FILE-body sanitization (op_read_for_context).
# STRIP_SET strips line/carriage feeds because op_sanitize_string operates on
# single-line fragments; when reading an entire source file into an LLM context
# we must PRESERVE line structure (LF/CR) while still neutralizing bidi /
# zero-width / control / BOM injection. \t is already excluded from STRIP_SET.
_FILE_STRIP_PRESERVE = frozenset({0x0A, 0x0D})  # LINE FEED, CARRIAGE RETURN
FILE_STRIP_SET = STRIP_SET - _FILE_STRIP_PRESERVE

# Deny-list patterns applied per path segment, case-insensitive (ASCII casefold)
DENY_PATTERNS = [
    "env*.json", "secrets*",
    ".env", ".env.*", "*.env", "*.env.*",
    "credential*", "credentials*", "*creds*", "*cred",
    "*secret*", "*secrets", "*password*", "*passwd*", "*token*",
    "api-key*", "api_key*", "apikey*",
    "auth.json", "auth-config*", "auth-tokens*",
    "*.pem", "*.key", "*.p12", "*.pfx", "*.p8", "*.jks", "*.jceks",
    "*.keystore", "*.truststore", "keystore", "truststore", "*.p7b",
    "id_rsa*", "id_dsa*", "id_ecdsa*", "id_ed25519*", "*.ovpn", "*.netrc.gpg",
    "*.key.json", "*-service-account*.json", "*-firebase-adminsdk-*.json",
    "firebase.json", ".firebaserc", "aws-exports.js", "kubeconfig",
    "profiles.yml",
    ".npmrc", ".yarnrc", ".yarnrc.yml", ".pypirc",
    ".dockercfg", "settings.xml", "settings-security.xml",
    ".netrc", "_netrc", ".htpasswd",
    "aio-config.json", "*-private.pem", "*ims*credentials*", "serviceuser*key*",
    "*.tfvars", "*.tfstate", "*.tfstate.backup",
    "*.gpg", "*.asc", "*.kdbx", "wallet.dat", "*.pgp",
    "datasources.local.xml", "sshconfigs.xml", "websservers.xml",
    "security*.xml", "sftp.json", "launch.local.json", "secrets.json",
    "*.bak", "*.orig", "*.swp", "*.swo", ".#*", "*~", "*.rej",
    # Auxiliary credential surfaces (security review I/M findings):
    "github_pat_*", ".vault-token", "*.aio-config", "op-session-*",
]
DENY_PATTERNS_LC = [p.lower() for p in DENY_PATTERNS]

# Pre-compile the deny-list into a single regex for hot-path lookup. fnmatch
# patterns translate cleanly via fnmatch.translate; we anchor and join with |.
# For a 7000-path workspace, this is ~5-10x faster than iterating fnmatch in
# segment_denied(). The compiled form is matched against the casefolded segment.
DENY_PATTERN_REGEX = re.compile(
    "|".join(f"(?:{fnmatch.translate(p)})" for p in DENY_PATTERNS_LC)
)

# Directory names that prune the entire subtree at every depth
DENY_DIRS = frozenset({
    ".git", "target", "node_modules", "dist", "build", "out",
    "crx-quickstart", ".idea",
    ".terraform", ".gnupg", ".ssh",
    ".aws", ".gcp", ".azure", ".kube", ".aio", ".adobe-aio", ".fbc",
    ".password-store", ".aws-sam", ".m2",
    ".databricks-cfg", ".snowflake", ".dbt",
})

# Special filesystems rejected even when the workspace lives inside them.
# macOS aliases /var/run -> /private/var/run, /tmp -> /private/tmp, etc.
# Realpath resolves through these, so both forms must be rejected.
REJECT_PREFIXES = (
    "/proc/", "/sys/", "/dev/", "/var/run/", "/run/",
    "/private/var/run/", "/private/run/",
)

# Allow-list of write destinations enforced inside op_write_atomic. Every
# write path must match at least one of these globs (after the relative-path
# and dotdot checks). This is the helper-enforced realization of
# SKILL.md § "Hard guarantee - allow-list of paths the skill writes" -
# previously this contract was enforced only by the orchestrating LLM.
WRITE_ALLOWLIST_GLOBS = (
    "AGENTS.md", "*/AGENTS.md", "*/*/AGENTS.md", "*/*/*/AGENTS.md",
    "*/*/*/*/AGENTS.md", "*/*/*/*/*/AGENTS.md",
    # Workspace-root CLAUDE.md only (the consent-gated root-CLAUDE.md write).
    # Intentionally NOT "*/CLAUDE.md" — nested CLAUDE.md remains out of scope,
    # and root AGENTS.md stays owned by ensure-agents-md.
    "CLAUDE.md",
    ".aem/context/*",
    "*/.aem/context/*",
    "*/*/.aem/context/*",
    ".aem/agentkit-overrides.yml",
    "*/.aem/agentkit-overrides.yml",
    ".claude/agents/*", ".claude/commands/*", ".claude/rules/*",
    ".cursor/rules/*", ".cursor/mcp.json",
    ".github/instructions/*", ".github/copilot-instructions.md",
    ".continue/rules/*",
    ".clinerules", ".clinerules.aem-roles-extra.md",
    ".windsurfrules", ".windsurfrules.aem-roles-extra.md",
    "augment.md", "augment.md.aem-roles-extra.md",
    ".mcp.json",
)
# Sidecars (.tmp and .agentkit-new) are derived from allow-list targets and
# share their allow-list status via _is_allowlisted().

# Marker fields removed from a JSON body before checksum
JSON_MARKER_FIELDS = (
    "_generatedBy", "_skillVersion", "schemaVersion",
    "_markerChecksum", "generatedAt", "_static",
)

MAX_BYTES_CEILING = 16 * 1024 * 1024
DEFAULT_MAX_FILES = 100_000
DEFAULT_MAX_DEPTH = 32
DEFAULT_MAX_FILES_PER_SUBTREE = 10_000

# Open lock file descriptors, keyed by absolute lock path. flock(2) is held
# for as long as the fd stays open; the kernel releases it automatically when
# the process dies (crash-safe). op_lock stores the fd here so op_unlock can
# release + close it within the same long-running helper process.
_LOCK_FDS: dict[str, int] = {}

# --------------------------------------------------------------------- #
# Helpers                                                               #
# --------------------------------------------------------------------- #


def casefold_ascii(s: str) -> str:
    """ASCII lowercase casefold (privacy-and-sanitization.md / helpers.md § 3).

    Bytes 0x41..0x5A -> 0x61..0x7A; every other byte unchanged. The input is
    NFC-normalized first so HFS+ (NFD-on-disk) and ext4/APFS (NFC) compare
    identically. Non-ASCII patterns would silently misbehave without this -
    today the deny-list is ASCII-only so the NFC pass is defense-in-depth.
    """
    s = unicodedata.normalize("NFC", s)
    return "".join(c.lower() if "A" <= c <= "Z" else c for c in s)


def segment_denied(segment: str) -> str:
    """Return the matching pattern name if the segment is denied, else "".

    Hot path: called for every entry in op_walk. Uses the pre-compiled
    DENY_PATTERN_REGEX (~5-10x faster than iterating fnmatch per pattern).
    """
    seg_lc = casefold_ascii(segment)
    if seg_lc in DENY_DIRS:
        return seg_lc
    if DENY_PATTERN_REGEX.fullmatch(seg_lc):
        # Return the first matching glob for diagnostic clarity.
        for pat in DENY_PATTERNS_LC:
            if fnmatch.fnmatchcase(seg_lc, pat):
                return pat
    return ""


def _is_allowlisted(rel_posix: str) -> str:
    """Return the matching allow-list glob, or "" if `rel_posix` is not
    a permitted write destination. Sidecars (.tmp, .agentkit-new) inherit
    their target's status.
    """
    candidate = rel_posix
    if candidate.endswith(".tmp"):
        candidate = candidate[:-4]
    elif candidate.endswith(".agentkit-new"):
        candidate = candidate[: -len(".agentkit-new")]
    for pat in WRITE_ALLOWLIST_GLOBS:
        if fnmatch.fnmatchcase(candidate, pat):
            return pat
    return ""


def _resolve_workspace(workspace: str) -> str:
    ws_real = os.path.realpath(workspace)
    if not os.path.isdir(ws_real):
        raise ValueError("workspace is not a directory")
    return ws_real


def _check_special_fs(realpath: str) -> str:
    for prefix in REJECT_PREFIXES:
        if realpath.startswith(prefix):
            return prefix
    return ""


def _fd_realpath(fd: int) -> str:
    """Return the canonical path of an open file descriptor.

    Used for the TOCTOU re-check in op_open (helpers.md § 2.2). Uses the
    stdlib `fcntl` module on Darwin (F_GETPATH = 50) which handles the
    buffer marshalling correctly - the previous ctypes path failed on
    real macOS because fcntl's third arg has variant type and ctypes
    doesn't know to marshal a string buffer as a path argument.
    """
    if sys.platform == "linux":
        return os.readlink(f"/proc/self/fd/{fd}")
    if sys.platform == "darwin":
        import fcntl as _fcntl
        F_GETPATH = getattr(_fcntl, "F_GETPATH", 50)
        # fcntl.fcntl with a bytes arg returns the modified buffer.
        buf = b"\x00" * 1024
        result = _fcntl.fcntl(fd, F_GETPATH, buf)
        return result.rstrip(b"\x00").decode("utf-8")
    raise OSError("fd realpath unsupported on this platform")


def _validate_path(workspace: str, path: str) -> dict:
    """Run the realpath gauntlet on `path` against `workspace`. Return dict.

    The resolved-realpath deny-list re-check (security finding C1) catches
    in-workspace symlinks that route around the deny-list - e.g.,
    `<ws>/innocent -> <ws>/.git`. The literal entry name "innocent" passes
    the segment check; without re-walking the realpath segments, descent
    into `.git` would surface its contents.
    """
    try:
        ws_real = _resolve_workspace(workspace)
    except (OSError, ValueError) as e:
        return {"ok": False, "error": f"workspace invalid: {e}"}

    try:
        path_real = os.path.realpath(path, strict=True)
    except OSError as e:
        return {"ok": False, "error": f"realpath failed: {e.strerror or e}", "errno": e.errno}

    parts = path_real.split(os.sep)
    if ".." in parts:
        return {"ok": False, "error": "resolved path contains .."}

    if path_real != ws_real and not path_real.startswith(ws_real + os.sep):
        return {"ok": False, "error": "path escapes workspace root"}

    bad_prefix = _check_special_fs(path_real)
    if bad_prefix:
        return {"ok": False, "error": f"path traverses rejected filesystem {bad_prefix}"}

    rel = "" if path_real == ws_real else os.path.relpath(path_real, ws_real)
    if rel:
        for seg in rel.split(os.sep):
            denied = segment_denied(seg)
            if denied:
                return {"ok": False, "error": f"deny-list match on segment: {seg} (pattern: {denied})"}

    # Case-collision detection (QA finding Q11): on case-insensitive
    # filesystems (default macOS APFS, NTFS) the realpath of `agents.md`
    # and `AGENTS.md` is the same file. If the requested basename differs
    # from the realpath basename byte-for-byte, surface a warning so the
    # caller can decide whether to proceed.
    case_collision = False
    requested_base = os.path.basename(path.rstrip(os.sep))
    real_base = os.path.basename(path_real)
    if (requested_base and real_base
            and requested_base != real_base
            and unicodedata.normalize("NFC", requested_base).lower()
                == unicodedata.normalize("NFC", real_base).lower()):
        case_collision = True

    try:
        is_symlink = os.path.islink(path)
    except OSError:
        is_symlink = False
    is_dir = os.path.isdir(path_real)

    return {
        "ok": True,
        "realpath": path_real,
        "workspaceRelative": rel.replace(os.sep, "/"),
        "isSymlink": is_symlink,
        "isDir": is_dir,
        "workspaceRealpath": ws_real,
        "caseCollision": case_collision,
    }


def _validate_segments(rel_posix: str) -> str:
    """Walk a workspace-relative POSIX path through segment_denied. Return
    the matching pattern, or "". Used for write-time policy enforcement and
    for op_match_deny ENOENT fallback (paths that don't exist yet).
    """
    if not rel_posix or rel_posix == ".":
        return ""
    for seg in rel_posix.split("/"):
        if seg in ("", "."):
            continue
        denied = segment_denied(seg)
        if denied:
            return denied
    return ""


# --------------------------------------------------------------------- #
# Operations                                                            #
# --------------------------------------------------------------------- #


def op_realpath(req):
    res = _validate_path(req["workspace"], req["path"])
    res.pop("workspaceRealpath", None)
    return res


def op_match_deny(req):
    workspace = req["workspace"]
    path = req["path"]
    res = _validate_path(workspace, path)
    if res["ok"]:
        return {"ok": True, "denied": False, "matchedPattern": None, "matchedSegment": None}
    err = res.get("error", "")
    m = re.match(r"deny-list match on segment: (.+) \(pattern: (.+)\)$", err)
    if m:
        return {"ok": True, "denied": True, "matchedSegment": m.group(1), "matchedPattern": m.group(2)}
    # ENOENT fallback (QA finding Q10): pre-flight checks need a clean
    # denied/allowed answer for paths that may not exist yet. Walk up to
    # the nearest existing ancestor, realpath it (to handle /tmp -> /private/tmp
    # on macOS and similar aliases), then re-attach the missing tail.
    if res.get("errno") == errno.ENOENT:
        try:
            ws_real = _resolve_workspace(workspace)
        except (OSError, ValueError):
            return res
        if os.path.isabs(path):
            candidate = path
        else:
            candidate = os.path.join(ws_real, path)
        ancestor = candidate
        while ancestor and not os.path.exists(ancestor):
            parent = os.path.dirname(ancestor)
            if parent == ancestor:
                break
            ancestor = parent
        try:
            anc_real = os.path.realpath(ancestor) if os.path.exists(ancestor) else ancestor
        except OSError:
            return res
        if candidate == ancestor:
            norm = anc_real
        else:
            tail = os.path.relpath(candidate, ancestor)
            norm = os.path.normpath(os.path.join(anc_real, tail))
        if not (norm == ws_real or norm.startswith(ws_real + os.sep)):
            return {"ok": False, "error": "path escapes workspace root"}
        rel = "" if norm == ws_real else os.path.relpath(norm, ws_real)
        rel_posix = rel.replace(os.sep, "/")
        denied = _validate_segments(rel_posix)
        if denied:
            offender = next(
                (s for s in rel_posix.split("/") if s and segment_denied(s)),
                "",
            )
            return {"ok": True, "denied": True, "matchedSegment": offender, "matchedPattern": denied}
        return {"ok": True, "denied": False, "matchedPattern": None, "matchedSegment": None}
    return res


def _safe_open_bytes(workspace, path, max_bytes):
    """Validate + safely open `path` within `workspace` and return its bytes.

    Runs the full security gauntlet: _validate_path, O_NOFOLLOW open of the
    resolved leaf, and the fail-closed _fd_realpath TOCTOU re-check. Returns
    {"ok": True, "data": <bytes>} on success or an error dict ({"ok": False,
    ...}) on any failure. Shared by op_open and op_read_for_context so both
    inherit identical workspace-boundary, deny-list, and TOCTOU guarantees.
    """
    val = _validate_path(workspace, path)
    if not val["ok"]:
        return val
    target = val["realpath"]

    # Intra-workspace symlinks via intermediate directories are legitimate
    # (pnpm, yarn workspaces, dispatcher submodules). Drop O_NOFOLLOW_ANY -
    # which rejected ANY symlink in the path - and open the FULLY RESOLVED
    # target instead of the requested path. O_NOFOLLOW on the leaf still
    # rejects the target itself being a symlink (which would defeat the
    # workspace-boundary check). See QA finding Q6.
    flags = os.O_RDONLY | os.O_NOFOLLOW

    try:
        fd = os.open(target, flags)
    except OSError as e:
        return {"ok": False, "error": f"open failed: {e.strerror or e}"}

    try:
        try:
            fd_real = _fd_realpath(fd)
        except OSError as e:
            # Fail-closed (security finding I4 / QA Q5). The TOCTOU re-check
            # is advertised as a hard contract; silently degrading to a
            # best-effort check would mean callers can't trust the
            # security guarantees of any platform where /proc/self/fd or
            # F_GETPATH is masked.
            return {
                "ok": False,
                "error": f"TOCTOU re-check unavailable: {e.strerror or e}",
            }
        if fd_real != target:
            return {"ok": False, "error": "TOCTOU mismatch: descriptor path differs from resolved path"}

        # Stream-read into a single bytearray with hard size tracking so
        # the helper does not transiently hold > max_bytes (memory cap fix
        # for SE4 / M4). On a 100 MB file with a 16 MB ceiling, prior code
        # accumulated chunks until len > max_bytes; we now stop at the
        # exact threshold.
        buf = bytearray()
        while len(buf) <= max_bytes:
            chunk = os.read(fd, min(65536, max_bytes + 1 - len(buf)))
            if not chunk:
                break
            buf.extend(chunk)
        if len(buf) > max_bytes:
            # Best-effort report of actual size for diagnostics (Q23).
            try:
                actual = os.fstat(fd).st_size
            except OSError:
                actual = len(buf)
            return {
                "ok": False,
                "error": f"file exceeds maxBytes ({max_bytes}); actual size {actual}",
            }
        return {"ok": True, "data": bytes(buf)}
    finally:
        os.close(fd)


def op_open(req):
    workspace = req["workspace"]
    path = req["path"]
    max_bytes = min(int(req.get("maxBytes", MAX_BYTES_CEILING)), MAX_BYTES_CEILING)

    res = _safe_open_bytes(workspace, path, max_bytes)
    if not res["ok"]:
        return res
    data = res["data"]

    return {
        "ok": True,
        "bytes": base64.b64encode(data).decode("ascii"),
        "sha256": hashlib.sha256(data).hexdigest(),
        "toctouVerified": True,
    }


def op_read_for_context(req):
    """Read source into an LLM context with dangerous code points neutralized.

    The skill's job is feeding customer Java/HTL/POM into a model to generate
    AGENTS.md. op_open returns RAW bytes; a bidi-override, zero-width, or
    control-char payload buried in a code comment would flow straight into the
    model. This op decodes the bytes (errors="replace"), NFC-normalizes, then
    removes every STRIP_SET code point (bidi overrides, zero-width marks, C0/C1
    controls, BOM, etc.) and reports how many were stripped.

    NOTE: this neutralizes *dangerous code points* only. It does NOT and cannot
    defend against natural-language prompt injection (e.g. an English sentence
    "ignore previous instructions" in a comment). The orchestrator MUST still
    treat returned content as untrusted. The sha256 is over the ORIGINAL raw
    bytes so callers can correlate with op_open / on-disk state.
    """
    workspace = req["workspace"]
    path = req["path"]
    max_bytes = min(int(req.get("maxBytes", MAX_BYTES_CEILING)), MAX_BYTES_CEILING)

    res = _safe_open_bytes(workspace, path, max_bytes)
    if not res["ok"]:
        return res
    data = res["data"]

    text = unicodedata.normalize("NFC", data.decode("utf-8", errors="replace"))
    kept = []
    stripped = 0
    for ch in text:
        if ord(ch) in FILE_STRIP_SET:
            stripped += 1
            continue
        kept.append(ch)
    sanitized = "".join(kept)

    # Self-validate: no FILE_STRIP_SET survivors. NFC normalization can in
    # principle re-introduce a composed form, so re-check fail-closed.
    if any(ord(ch) in FILE_STRIP_SET for ch in sanitized):
        return {"ok": False, "error": "sanitization left dangerous code points"}

    return {
        "ok": True,
        "text": sanitized,
        "sha256": hashlib.sha256(data).hexdigest(),
        "stripped": stripped,
        "toctouVerified": True,
    }


def op_walk(req):
    """Bounded directory walk.

    Glob dialect: Python `fnmatch.fnmatchcase` against the workspace-relative
    POSIX path. `*` matches any character INCLUDING `/`, so `*.java` matches
    `core/A.java` AND `core/sub/B.java`. This is NOT shell-glob (where `*`
    stops at `/`) and NOT git-style `**` (which is unsupported). To restrict
    a walk to a single sub-tree, pass it as a root; do not rely on the glob
    for path-segment scoping. See references/helpers.md § 2.3.
    """
    workspace = req["workspace"]
    roots = req.get("roots", ["."])
    max_files = int(req.get("maxFiles", DEFAULT_MAX_FILES))
    max_depth = int(req.get("maxDepth", DEFAULT_MAX_DEPTH))
    per_subtree = int(req.get("maxFilesPerSubtree", DEFAULT_MAX_FILES_PER_SUBTREE))
    globs = req.get("globs", []) or []

    try:
        ws_real = _resolve_workspace(workspace)
    except (OSError, ValueError) as e:
        return {"ok": False, "error": f"workspace invalid: {e}"}

    files = []
    visited = set()
    warnings = []
    truncated_subtrees = []
    truncated_global = False

    def matches_any_glob(rel_posix):
        if not globs:
            return True
        return any(fnmatch.fnmatchcase(rel_posix, g) for g in globs)

    for root in roots:
        if truncated_global:
            break
        root_path = root if os.path.isabs(root) else os.path.join(ws_real, root)
        root_val = _validate_path(workspace, root_path)
        if not root_val["ok"]:
            warnings.append(f"root rejected: {root}: {root_val.get('error')}")
            continue
        root_real = root_val["realpath"]
        subtree_count = 0
        subtree_truncated = False
        subtree_label = root_val["workspaceRelative"] or "."

        stack = [(root_real, 0)]
        while stack and not subtree_truncated and not truncated_global:
            current, depth = stack.pop()
            if depth > max_depth:
                warnings.append(f"depth cap reached at {os.path.relpath(current, ws_real)}")
                continue
            try:
                entries = sorted(os.listdir(current))
            except OSError as e:
                warnings.append(f"cannot list {os.path.relpath(current, ws_real)}: {e.strerror}")
                continue
            for name in entries:
                full = os.path.join(current, name)
                # Run the full validation gauntlet on every entry so an
                # in-workspace symlink (e.g. `safe -> .git`) cannot escape
                # the deny-list. The prior code only checked the literal
                # entry name `name`; the resolved realpath's segments were
                # not re-checked. See security finding C1.
                child_val = _validate_path(workspace, full)
                if not child_val["ok"]:
                    err = child_val.get("error", "")
                    rel_for_warn = os.path.relpath(full, ws_real)
                    if "deny-list" in err:
                        warnings.append(f"deny-list rejected: {rel_for_warn}: {err}")
                    elif "escapes workspace" in err:
                        warnings.append(f"escape rejected: {rel_for_warn}")
                    elif "rejected filesystem" in err:
                        warnings.append(f"special-fs rejected: {rel_for_warn}")
                    else:
                        warnings.append(f"rejected: {rel_for_warn}: {err}")
                    continue
                real = child_val["realpath"]
                if real in visited:
                    continue
                visited.add(real)
                if child_val["isDir"]:
                    stack.append((real, depth + 1))
                    continue
                rel_posix = child_val["workspaceRelative"]
                if matches_any_glob(rel_posix):
                    files.append(rel_posix)
                    subtree_count += 1
                    if subtree_count >= per_subtree:
                        # Per-subtree cap fired: terminate THIS subtree's
                        # walk only. Leave the outer-loop stack alone so
                        # subsequent roots still get a chance. Previously
                        # stack.clear() short-circuited all remaining
                        # roots (QA finding Q12).
                        truncated_subtrees.append(subtree_label)
                        warnings.append(f"per-subtree cap reached: {subtree_label}")
                        subtree_truncated = True
                        break
                    if len(files) >= max_files:
                        # Global cap is workspace-wide. DON'T tag the
                        # current subtree as "truncated" - the subtree
                        # might be complete; the cap fired on its last
                        # entry. Set the global truncated flag only.
                        # See staff-engineer finding SE1.
                        truncated_global = True
                        warnings.append("global file-walk cap reached")
                        break

    files.sort()
    return {
        "ok": True,
        "files": files,
        "truncated": truncated_global or bool(truncated_subtrees),
        "truncatedSubtrees": sorted(set(truncated_subtrees)),
        "globalCapReached": truncated_global,
        "warnings": sorted(warnings),
    }


def _nfc_normalize_leaves(obj):
    """Walk a JSON-decoded structure and NFC-normalize every string leaf.

    Without this, identical logical content can hash differently between
    HFS+ (NFD) and ext4/APFS (NFC) — see QA finding Q7. The strip-list
    pass on op_sanitize_string covers extracted string fragments before
    they enter a JSON body; this pass covers the JSON body holistically
    so re-runs on macOS NFD-on-disk don't churn `.agentkit-new` sidecars.
    """
    if isinstance(obj, str):
        return unicodedata.normalize("NFC", obj)
    if isinstance(obj, dict):
        return {k: _nfc_normalize_leaves(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_nfc_normalize_leaves(v) for v in obj]
    return obj


def _canonical_body_sha(raw: bytes, kind):
    """Compute the canonical-body sha256 for `raw` under the given `kind`.

    Returns the hex digest string on success, or None on any structural
    failure (no marker newline, BOM, parse error, non-object JSON, unknown
    kind). This is the single source of truth for both op_sha256_canonical
    (the external op) and _is_skill_owned (the overwrite-protection check),
    so the two can never diverge in how they derive the body checksum.
    """
    if raw.startswith(b"\xef\xbb\xbf"):
        return None

    if kind == "markdown":
        # The marker MUST be on the first non-empty line. Skip leading blank
        # lines so a stray newline from an IDE auto-prettier doesn't reclass
        # the file as human-curated (QA finding Q22).
        pos = 0
        while pos < len(raw) and raw[pos:pos + 1] in (b"\n", b"\r"):
            pos += 1
        nl = raw.find(b"\n", pos)
        if nl < 0:
            return None
        body = raw[nl + 1:]
        return hashlib.sha256(body).hexdigest()

    if kind == "json":
        try:
            obj = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            return None
        if not isinstance(obj, dict):
            return None
        # JSON_MARKER_FIELDS is stripped at the TOP LEVEL only. Nested keys
        # with the same names are preserved by design — if a customer added
        # a nested `componentDefinition._markerChecksum`, the checksum is
        # legitimately part of the body. See security finding M2.
        cleaned = {k: v for k, v in obj.items() if k not in JSON_MARKER_FIELDS}
        cleaned = _nfc_normalize_leaves(cleaned)
        emitted = json.dumps(
            cleaned, sort_keys=True, indent=2, ensure_ascii=False,
            separators=(",", ": "),
        ).encode("utf-8") + b"\n"
        return hashlib.sha256(emitted).hexdigest()

    return None


def op_sha256_canonical(req):
    kind = req.get("kind")
    try:
        raw = base64.b64decode(req["bytes"], validate=True)
    except Exception as e:
        return {"ok": False, "error": f"bytes must be valid base64: {e}"}

    if raw.startswith(b"\xef\xbb\xbf"):
        return {"ok": False, "error": "UTF-8 BOM not allowed"}

    if kind == "markdown":
        sha = _canonical_body_sha(raw, "markdown")
        if sha is None:
            return {"ok": False, "error": "markdown body missing a newline-terminated marker line"}
        return {"ok": True, "sha256": sha}

    if kind == "json":
        # Preserve the granular error messages the external op contract
        # advertises (parse failure vs. non-object top-level); these are
        # exercised by existing tests. _canonical_body_sha collapses both
        # to None, so re-derive the specific cause here.
        try:
            obj = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as e:
            return {"ok": False, "error": f"json parse failed: {e}"}
        if not isinstance(obj, dict):
            return {"ok": False, "error": "json top-level must be an object"}
        sha = _canonical_body_sha(raw, "json")
        return {"ok": True, "sha256": sha}

    return {"ok": False, "error": f"unknown kind: {kind}"}


# Version-agnostic marker prefix and checksum locator for markdown marker
# lines. The version digits after `v` are intentionally not matched so that
# a file generated by any aem-agentkit version is recognized.
_MD_MARKER_PREFIX = b"<!-- aem-agentkit: generated v"
_MD_CHECKSUM_RE = re.compile(rb"checksum:\s*([0-9a-fA-F]{64})\b")


def _is_skill_owned(full_path: str, kind) -> bool:
    """Return True iff the EXISTING file at `full_path` is authentically
    skill-generated, per collision-rules.md: a file is "skill-owned" only
    when its marker prefix matches AND the embedded sha256 recomputes over
    the canonical body. Anything that fails — wrong prefix, missing/malformed
    checksum, a second marker occurrence, a mismatch, an unreadable file, a
    symlink, or an unknown kind — is treated as human-curated (return False).

    Fail-closed in every ambiguous case: when in doubt, refuse the overwrite.
    """
    # Fail-closed open: O_NOFOLLOW rejects a symlink at the leaf; any open
    # error (permission, missing, ENOENT race) is treated as human-curated.
    try:
        fd = os.open(full_path, os.O_RDONLY | os.O_NOFOLLOW)
    except OSError:
        return False
    try:
        try:
            raw = os.read(fd, MAX_BYTES_CEILING)
        except OSError:
            return False
    finally:
        os.close(fd)

    if kind == "markdown":
        # The marker prefix must occur exactly once anywhere in the file; a
        # second occurrence is a spoofing signal -> human-curated.
        if raw.count(_MD_MARKER_PREFIX) != 1:
            return False
        # The marker must be on the first NON-EMPTY line.
        pos = 0
        while pos < len(raw) and raw[pos:pos + 1] in (b"\n", b"\r"):
            pos += 1
        nl = raw.find(b"\n", pos)
        if nl < 0:
            return False
        first_line = raw[pos:nl]
        if not first_line.startswith(_MD_MARKER_PREFIX):
            return False
        m = _MD_CHECKSUM_RE.search(first_line)
        if not m:
            return False
        embedded = m.group(1).decode("ascii").lower()
        recomputed = _canonical_body_sha(raw, "markdown")
        if recomputed is None:
            return False
        return embedded == recomputed

    if kind == "json":
        try:
            obj = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            return False
        if not isinstance(obj, dict):
            return False
        if obj.get("_generatedBy") != "aem-agentkit":
            return False
        embedded = obj.get("_markerChecksum")
        if not isinstance(embedded, str):
            return False
        recomputed = _canonical_body_sha(raw, "json")
        if recomputed is None:
            return False
        return embedded.lower() == recomputed

    # Unknown / missing kind -> conservatively human-curated.
    return False


def op_write_atomic(req):
    """Atomic write with helper-enforced allow-list and deny-list.

    Previously the helper trusted the orchestrating LLM to send only
    allow-listed paths. A bug or prompt injection could ask the helper to
    write `.git/hooks/post-commit` or `.env`, and the helper would comply.
    The allow-list (security C2 / QA Q2) and per-segment deny-list checks
    are now ENFORCED here — the helper is the trust anchor for the
    "no writes outside the allow-list" hard guarantee in SKILL.md.
    """
    workspace = req["workspace"]
    rel = req["path"]
    kind = req.get("kind")
    try:
        data = base64.b64decode(req["bytes"], validate=True)
    except Exception as e:
        return {"ok": False, "error": f"bytes must be valid base64: {e}"}

    try:
        ws_real = _resolve_workspace(workspace)
    except (OSError, ValueError) as e:
        return {"ok": False, "error": f"workspace invalid: {e}"}

    if os.path.isabs(rel) or ".." in rel.split("/"):
        return {"ok": False, "error": "path must be relative without .."}

    rel_posix = rel.replace(os.sep, "/")

    # Deny-list segment check first (applies even before allow-list because
    # the deny-list is a hard refusal regardless of intent).
    denied = _validate_segments(rel_posix)
    if denied:
        return {"ok": False, "error": f"deny-list match on path: {rel_posix} (pattern: {denied})"}

    # Allow-list check. Caller may opt out per request with
    # `enforceAllowlist: false`, but the default is the safe one. Setting
    # this to false is intended only for test fixtures.
    enforce_allowlist = bool(req.get("enforceAllowlist", True))
    matched_allow = _is_allowlisted(rel_posix)
    if enforce_allowlist and not matched_allow:
        return {
            "ok": False,
            "error": (
                f"path not in write allow-list: {rel_posix}. See SKILL.md "
                f"'Hard guarantee' for the permitted destinations."
            ),
        }

    full = os.path.join(ws_real, *rel.split("/"))
    parent = os.path.dirname(full)
    if not parent:
        return {"ok": False, "error": "path lacks a parent directory"}

    # Validate-before-makedirs (security I5): walk up to the nearest
    # existing ancestor, realpath-check it stays inside the workspace.
    # The prior code ran makedirs first; if any parent component was a
    # symlink to outside, makedirs created directories there before the
    # realpath check rejected. Now no side effects happen until we know
    # the path is safe.
    ancestor = parent
    while ancestor and not os.path.exists(ancestor):
        ancestor = os.path.dirname(ancestor)
    if not ancestor:
        return {"ok": False, "error": "no existing ancestor for parent"}
    try:
        ancestor_real = os.path.realpath(ancestor, strict=True)
    except OSError as e:
        return {"ok": False, "error": f"ancestor realpath failed: {e.strerror or e}"}
    if ancestor_real != ws_real and not ancestor_real.startswith(ws_real + os.sep):
        return {"ok": False, "error": "ancestor escapes workspace root"}
    # Reject if any segment between the validated ancestor and the parent
    # is a symlink (a malicious customer can't insert one, but a stray
    # workspace symlink could).
    walked = ancestor_real
    rel_from_ancestor = os.path.relpath(parent, ancestor) if parent != ancestor else ""
    if rel_from_ancestor and rel_from_ancestor != ".":
        for seg in rel_from_ancestor.split(os.sep):
            if not seg:
                continue
            candidate = os.path.join(walked, seg)
            if os.path.islink(candidate):
                return {"ok": False, "error": f"refused: intermediate dir is a symlink: {seg}"}
            walked = candidate

    try:
        os.makedirs(parent, exist_ok=True)
    except OSError as e:
        return {"ok": False, "error": f"parent dir create failed: {e.strerror or e}"}

    # Post-makedirs sanity check (cheap defense-in-depth).
    parent_real = os.path.realpath(parent)
    if parent_real != ws_real and not parent_real.startswith(ws_real + os.sep):
        return {"ok": False, "error": "parent escapes workspace root"}

    # Case-collision detection (Q11): on case-insensitive filesystems,
    # writing `AGENTS.md` when `agents.md` already exists silently
    # overwrites it. Detect and refuse unless the caller opted in.
    case_collision_detected = False
    target_base = os.path.basename(full)
    if os.path.exists(full):
        try:
            existing_real_base = os.path.basename(os.path.realpath(full))
        except OSError:
            existing_real_base = target_base
        if (existing_real_base != target_base
                and unicodedata.normalize("NFC", existing_real_base).lower()
                    == unicodedata.normalize("NFC", target_base).lower()):
            case_collision_detected = True
            if not bool(req.get("allowCaseCollision", False)):
                return {
                    "ok": False,
                    "error": (
                        f"case-insensitive filesystem collision: requested "
                        f"{target_base!r} but pre-existing entry is "
                        f"{existing_real_base!r}. Refusing to silently rename."
                    ),
                }

    # Marker-aware, fail-closed overwrite protection (collision-rules.md):
    # never silently overwrite a human-curated file. A file is skill-owned
    # only when its marker prefix matches AND the embedded sha256 recomputes
    # over the canonical body; anything that fails is human-curated and must
    # not be clobbered. The orchestrator may force an overwrite with
    # allowOverwriteHumanCurated:true (the diagnostic field below records it).
    # Fresh paths (target does not exist) skip the check entirely.
    allow_overwrite_human = bool(req.get("allowOverwriteHumanCurated", False))
    overwrote_human_curated = False
    if os.path.exists(full):
        skill_owned = _is_skill_owned(full, kind)
        if not skill_owned:
            if not allow_overwrite_human:
                return {
                    "ok": False,
                    "error": (
                        f"refusing to overwrite human-curated file: {rel_posix}. "
                        f"It exists and is not skill-owned (no valid aem-agentkit marker). "
                        f"Pass allowOverwriteHumanCurated:true to force, or write to a "
                        f".agentkit-new sidecar."
                    ),
                }
            overwrote_human_curated = True

    tmp = full + ".tmp"
    try:
        fd = os.open(tmp, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
    except FileExistsError:
        return {"ok": False, "error": ".tmp already exists; aborting"}
    except OSError as e:
        return {"ok": False, "error": f"tmp open failed: {e.strerror or e}"}

    try:
        os.write(fd, data)
        os.fsync(fd)
    finally:
        os.close(fd)

    try:
        os.rename(tmp, full)
    except OSError as e:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        return {"ok": False, "error": f"rename failed: {e.strerror or e}"}

    try:
        dfd = os.open(parent, os.O_RDONLY)
        try:
            os.fsync(dfd)
        finally:
            os.close(dfd)
    except OSError:
        pass

    return {
        "ok": True,
        "sha256": hashlib.sha256(data).hexdigest(),
        "allowlistMatch": matched_allow,
        "caseCollision": case_collision_detected,
        "overwroteHumanCurated": overwrote_human_curated,
    }


def op_cleanup_tmp(req):
    """Delete orphan `.tmp` sidecars adjacent to marker-bearing targets,
    plus orphan `.tmp` files where the target does not exist (a crash
    artifact from a previous run that crashed mid-write-atomic).

    Previously, op_cleanup_tmp ONLY deleted .tmp when the target existed
    AND carried the marker prefix. After a crash mid-write-atomic the
    target was never created, so the .tmp sat there forever - and the
    next write-atomic to that path hits O_CREAT|O_EXCL on the tmp and
    aborts permanently. The skill becomes wedged. See QA finding Q4.

    Adds realpath validation per visited candidate (security I3): the
    op now uses the same bounded-walk + per-entry _validate_path that
    op_walk does, so a customer symlink chain cannot trick the cleanup
    into reading or deleting outside the workspace.

    The marker prefix `<!-- aem-agentkit: generated v` is intentionally
    version-agnostic. Cross-version cleanup is the right default: a tmp
    written by v0.9.0 should still be cleaned by v1.0.0 if its target
    was written by either version.
    """
    workspace = req["workspace"]
    try:
        ws_real = _resolve_workspace(workspace)
    except (OSError, ValueError) as e:
        return {"ok": False, "error": f"workspace invalid: {e}"}

    deleted = []
    orphans_recovered = []

    # Bounded walk, same gauntlet as op_walk (security I3): every visited
    # entry must pass _validate_path, so a customer-controlled symlink
    # chain `safe -> /etc` cannot reach .tmp files outside the workspace.
    stack = [(ws_real, 0)]
    visited = set()
    while stack:
        current, depth = stack.pop()
        if depth > DEFAULT_MAX_DEPTH:
            continue
        try:
            entries = sorted(os.listdir(current))
        except OSError:
            continue
        for name in entries:
            full = os.path.join(current, name)
            child_val = _validate_path(workspace, full)
            if not child_val["ok"]:
                continue
            real = child_val["realpath"]
            if real in visited:
                continue
            visited.add(real)
            if child_val["isDir"]:
                stack.append((real, depth + 1))
                continue
            if not name.endswith(".tmp"):
                continue

            target_name = name[:-4]
            target_path = os.path.join(current, target_name)
            tmp_path = full
            rel_tmp = os.path.relpath(tmp_path, ws_real).replace(os.sep, "/")

            target_exists = os.path.exists(target_path)
            if target_exists:
                try:
                    with open(target_path, "rb") as f:
                        head = f.read(4096)
                except OSError:
                    continue
                if (b"<!-- aem-agentkit: generated v" in head
                        or b'"_generatedBy": "aem-agentkit"' in head):
                    try:
                        os.unlink(tmp_path)
                        deleted.append(rel_tmp)
                    except OSError:
                        pass
            else:
                # Orphan .tmp with no target. Recover ONLY when the .tmp
                # sits at an allow-listed path - that's the signal it was
                # written by a prior crashed write-atomic run, not some
                # unrelated customer file ending in .tmp. See Q4.
                if _is_allowlisted(rel_tmp):
                    try:
                        os.unlink(tmp_path)
                        orphans_recovered.append(rel_tmp)
                    except OSError:
                        pass

    return {
        "ok": True,
        "deleted": sorted(deleted),
        "orphansRecovered": sorted(orphans_recovered),
    }


def op_sanitize_string(req):
    value = req.get("value", "")
    if not isinstance(value, str):
        return {"ok": False, "error": "value must be a string"}
    if value == "":
        return {"ok": True, "value": "", "dropped": True, "reason": "empty"}

    normalized = unicodedata.normalize("NFC", value)

    # Strip-list check
    for ch in normalized:
        if ord(ch) in STRIP_SET:
            return {"ok": True, "value": "", "dropped": True, "reason": "stripped"}

    truncated = False
    if len(normalized) > 80:
        normalized = normalized[:79] + "…"
        truncated = True

    # Inline-code wrap with appropriate fence length
    backticks = re.findall(r"`+", normalized)
    longest = max((len(b) for b in backticks), default=0)
    fence = "`" * (longest + 1)
    wrapped = f"{fence}{normalized}{fence}"

    # Self-validate (no strip-list survivors)
    for ch in wrapped:
        if ord(ch) in STRIP_SET:
            return {"ok": True, "value": "", "dropped": True, "reason": "stripped"}

    reason = "length" if truncated else "ok"
    return {"ok": True, "value": wrapped, "dropped": False, "reason": reason}


def op_lock(req):
    """Advisory lock at .aem/context/.agentkit.lock via fcntl.flock.

    Uses a real `flock(LOCK_EX | LOCK_NB)` on an open file descriptor that
    the helper holds for the lifetime of the lock. flock is advisory and the
    kernel releases it automatically when the holding process dies, so a
    SIGKILL'd helper cannot leave a stale lock behind - crash-safety comes
    for free and there is no stale-classification / PID-reuse logic to get
    wrong. The PID written into the file is diagnostic only.
    """
    workspace = req["workspace"]
    lock_rel = req.get("path", ".aem/context/.agentkit.lock")
    try:
        ws_real = _resolve_workspace(workspace)
    except (OSError, ValueError) as e:
        return {"ok": False, "acquired": False, "error": f"workspace invalid: {e}"}

    lock_full = os.path.join(ws_real, *lock_rel.split("/"))
    parent = os.path.dirname(lock_full)
    try:
        os.makedirs(parent, exist_ok=True)
    except OSError as e:
        return {"ok": False, "acquired": False, "error": f"lock parent: {e.strerror or e}"}

    pid = os.getpid()
    try:
        fd = os.open(lock_full, os.O_CREAT | os.O_RDWR, 0o644)
    except OSError as e:
        return {"ok": False, "acquired": False, "error": f"lock open failed: {e.strerror or e}"}

    try:
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except (BlockingIOError, OSError) as e:
        # EAGAIN / EWOULDBLOCK: another process holds the lock.
        os.close(fd)
        if isinstance(e, OSError) and e.errno not in (errno.EAGAIN, errno.EWOULDBLOCK):
            return {"ok": False, "acquired": False, "error": f"flock failed: {e.strerror or e}"}
        return {"ok": False, "acquired": False, "error": "another invocation is already running"}

    # Acquired. Record diagnostic PID (not load-bearing) and hold the fd open -
    # closing it would release the flock.
    try:
        os.ftruncate(fd, 0)
        os.write(fd, f"{pid}\n".encode("ascii"))
        os.fsync(fd)
    except OSError:
        pass
    _LOCK_FDS[lock_full] = fd
    return {"ok": True, "acquired": True, "pid": pid}


def op_unlock(req):
    workspace = req["workspace"]
    lock_rel = req.get("path", ".aem/context/.agentkit.lock")
    try:
        ws_real = _resolve_workspace(workspace)
    except (OSError, ValueError) as e:
        return {"ok": False, "error": f"workspace invalid: {e}"}
    lock_full = os.path.join(ws_real, *lock_rel.split("/"))

    fd = _LOCK_FDS.pop(lock_full, None)
    if fd is not None:
        try:
            fcntl.flock(fd, fcntl.LOCK_UN)
        except OSError:
            pass
        try:
            os.close(fd)
        except OSError:
            pass
    try:
        os.unlink(lock_full)
    except FileNotFoundError:
        pass
    except OSError:
        pass
    return {"ok": True}


# --------------------------------------------------------------------- #
# Dispatch                                                              #
# --------------------------------------------------------------------- #

def op_protocol_version(req):
    """Return the protocol version so the driver can pin separately
    from the skill version (architect rec: decouple cadences)."""
    return {"ok": True, "protocolVersion": PROTOCOL_VERSION, "skillVersion": VERSION}


OPS = {
    "realpath": op_realpath,
    "open": op_open,
    "read-for-context": op_read_for_context,
    "walk": op_walk,
    "sha256-canonical": op_sha256_canonical,
    "write-atomic": op_write_atomic,
    "cleanup-tmp": op_cleanup_tmp,
    "sanitize-string": op_sanitize_string,
    "lock": op_lock,
    "unlock": op_unlock,
    "match-deny": op_match_deny,
    "protocol-version": op_protocol_version,
}

# Per-invocation cap on stdin ops (security M5). Prevents a runaway
# orchestrator (or one being prompt-injected via repo content) from
# coercing the helper into millions of operations.
MAX_OPS_PER_INVOCATION = 100_000


def handle(req):
    op = req.get("op")
    fn = OPS.get(op)
    if fn is None:
        return {"ok": False, "error": f"unknown op: {op!r}"}
    try:
        return fn(req)
    except KeyError as e:
        return {"ok": False, "error": f"missing required field: {e}"}
    except Exception as e:
        # Print traceback to stderr so customer terminals (and CI logs) get
        # actionable context. Stdout is reserved for the JSON-line protocol;
        # stderr is invisible to the orchestrator-driven flow but visible
        # to humans. See SE finding 8.
        traceback.print_exc(file=sys.stderr)
        return {"ok": False, "error": f"internal helper error: {type(e).__name__}: {e}"}


def main():
    if len(sys.argv) > 1:
        if sys.argv[1] == "--version":
            sys.stdout.write(VERSION + "\n")
            return 0
        if sys.argv[1] == "--protocol-version":
            sys.stdout.write(PROTOCOL_VERSION + "\n")
            return 0
        if sys.argv[1] == "--self-test":
            # Self-test entry point so downstream consumers can run the
            # gate without the JS toolchain. Returns 0 if all unit tests
            # pass, non-zero otherwise.
            here = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
            return os.spawnvp(os.P_WAIT, sys.executable, [
                sys.executable, "-m", "unittest", "discover",
                "-s", os.path.join(here, "tests"),
                "-p", "test_*.py", "-v",
            ])
        if sys.argv[1] in ("-h", "--help"):
            sys.stdout.write(__doc__)
            return 0
        sys.stderr.write(f"aem-agentkit-helper: unknown argument {sys.argv[1]!r}\n")
        return 1

    all_ok = True
    op_count = 0
    for line in sys.stdin:
        op_count += 1
        if op_count > MAX_OPS_PER_INVOCATION:
            sys.stderr.write(
                f"aem-agentkit-helper: op-count cap reached "
                f"({MAX_OPS_PER_INVOCATION}); aborting\n"
            )
            return 1
        line = line.strip()
        if not line:
            continue
        try:
            req = json.loads(line)
        except json.JSONDecodeError as e:
            resp = {"ok": False, "error": f"invalid json: {e}"}
        else:
            resp = handle(req)
        if not resp.get("ok"):
            all_ok = False
        sys.stdout.write(json.dumps(resp, sort_keys=True) + "\n")
        sys.stdout.flush()
    return 0 if all_ok else 1


if __name__ == "__main__":
    sys.exit(main())
