Shared chat-channel foundations

nerve/agent/engine.py

@@ -254,6 +254,7 @@ def __init__(self, config: NerveConfig, db: Database):
         # survives restarts without re-firing on every resume.
         self._observed_models: dict[str, str] = {}
         self._router = None  # ChannelRouter — lazy-initialized via .router property
+        self._channel_runtimes: dict[str, Any] = {}
         # Gateways expose an ephemeral plaintext MCP listener on loopback for
         # co-located Codex processes. The gateway fills this after the listener
         # starts; the callable passed to CodexBackend reads it lazily.
@@ -714,6 +715,14 @@ def register_channel(self, channel: Any) -> None:
         """Register a channel with the router."""
         self.router.register(channel)
 
+    def register_channel_runtime(self, name: str, runtime: Any) -> None:
+        """Publish the lifecycle owner for a dynamically managed channel."""
+        self._channel_runtimes[name] = runtime
+
+    def get_channel_runtime(self, name: str) -> Any | None:
+        """Return a channel lifecycle owner, if one was installed."""
+        return self._channel_runtimes.get(name)
+
     # ------------------------------------------------------------------ #
     #  File snapshot for diff tracking                                     #
     # ------------------------------------------------------------------ #

nerve/channels/access.py

@@ -0,0 +1,165 @@
+"""Pattern-matching primitives for transport access policies.
+
+A gate matches an identity's platform ID and resolved aliases using
+case-insensitive globs. Deny wins, a non-empty allow list must match, and an
+incomplete identity cannot clear a deny list.
+
+Aliases are split by who controls them. A deny rule may match any of them.
+An allow rule may match only the ones the subject cannot set for itself,
+because a grant that rests on a self-set name lets the subject choose its
+own access.
+"""
+
+from __future__ import annotations
+
+import fnmatch
+from dataclasses import dataclass, field
+from typing import Callable
+
+
+def _norm(value: object) -> str:
+    """Normalize a value for case-insensitive matching."""
+    return str(value).strip().lower()
+
+
+def _matches(value: str, pattern: str) -> bool:
+    """Case-insensitive shell-glob match of one value against one pattern."""
+    return fnmatch.fnmatchcase(_norm(value), _norm(pattern))
+
+
+@dataclass(frozen=True)
+class Identity:
+    """A platform ID and resolved names used for policy matching.
+
+    ``names`` hold identity the platform or its administrators control, so a
+    grant may rest on them. ``self_set_names`` hold whatever the subject can
+    set without approval; a deny rule may match those, but an allow rule may
+    not, because the subject would then choose its own access.
+
+    ``complete`` means the candidates cover every name relevant to the active
+    patterns. ID-only policies therefore need no lookup, while deny lists reject
+    an incomplete identity.
+    """
+
+    id: str = ""
+    names: tuple[str, ...] = ()
+    self_set_names: tuple[str, ...] = ()
+    complete: bool = True
+
+    @property
+    def candidates(self) -> tuple[str, ...]:
+        """Every string an allow rule may grant on."""
+        return tuple(v for v in (self.id, *self.names) if v)
+
+    @property
+    def deny_candidates(self) -> tuple[str, ...]:
+        """Every string a deny rule may refuse on, self-set names included."""
+        return tuple(
+            v for v in (self.id, *self.names, *self.self_set_names) if v
+        )
+
+    def __str__(self) -> str:
+        label = next(iter((*self.names, *self.self_set_names)), "")
+        if label and self.id:
+            return f"{label} ({self.id})"
+        return label or self.id or "unknown"
+
+
+@dataclass(frozen=True)
+class Decision:
+    """The outcome of a policy check, with a reason fit for a log line."""
+
+    allowed: bool
+    reason: str = ""
+
+    def __bool__(self) -> bool:
+        return self.allowed
+
+
+@dataclass
+class PatternGate:
+    """Allow/deny matching for a labeled identity."""
+
+    label: str
+    allow: list[str] = field(default_factory=list)
+    deny: list[str] = field(default_factory=list)
+
+    def any_deny_pattern(self, predicate: Callable[[str], bool]) -> bool:
+        """Whether any non-empty deny pattern satisfies *predicate*."""
+        return any(predicate(p) for p in self.deny if p)
+
+    def check(self, who: Identity) -> Decision:
+        """Decide whether *who* passes this gate."""
+        candidates = who.candidates
+
+        for pattern in self.deny:
+            for value in who.deny_candidates:
+                if _matches(value, pattern):
+                    return Decision(
+                        False,
+                        f"{self.label} {who} matches deny pattern {pattern!r}",
+                    )
+
+        # A deny list is only meaningful against a candidate set known to
+        # cover it. Refuse rather than let an unread name walk past the list
+        # that names it.
+        if self.deny and not who.complete:
+            return Decision(
+                False,
+                f"{self.label} {who} could not be fully identified, so the "
+                f"deny list cannot be checked",
+            )
+
+        if self.allow:
+            if not candidates:
+                return Decision(
+                    False, f"{self.label} is unidentified and an allow list is set",
+                )
+            for pattern in self.allow:
+                for value in candidates:
+                    if _matches(value, pattern):
+                        return Decision(
+                            True,
+                            f"{self.label} {who} matches allow pattern {pattern!r}",
+                        )
+            # Say so when the only match was on a name the subject sets,
+            # otherwise the rule looks broken rather than declined.
+            for pattern in self.allow:
+                for value in who.self_set_names:
+                    if _matches(value, pattern):
+                        return Decision(
+                            False,
+                            f"{self.label} {who} matches allow pattern "
+                            f"{pattern!r} only on a profile name it sets "
+                            f"itself; grant on the id, handle, or email",
+                        )
+            return Decision(False, f"{self.label} {who} is not on the allow list")
+
+        return Decision(True, "")
+
+
+def needs_name_resolution(
+    *gates: PatternGate, is_id: Callable[[str], bool] | None = None,
+) -> bool:
+    """Whether any gate pattern requires names beyond the platform ID.
+
+    ``is_id`` recognizes literal IDs. Omitting it forces resolution, as does
+    any glob; extra lookups are safer than skipping one needed by a deny rule.
+    """
+    for gate in gates:
+        for pattern in (*gate.allow, *gate.deny):
+            if not pattern:
+                continue
+            if any(c in pattern for c in "*?["):
+                return True
+            if is_id is None or not is_id(pattern):
+                return True
+    return False
+
+
+__all__ = [
+    "Decision",
+    "Identity",
+    "PatternGate",
+    "needs_name_resolution",
+]

nerve/channels/archives.py

@@ -0,0 +1,187 @@
+"""Bounded ZIP unpacking, shared by the chat channels.
+
+Slack and Telegram both accept an attached archive and both cap the file
+they download. That cap is on the *compressed* bytes, so it says nothing
+about what the archive expands to: a few megabytes of zeros expand to
+gigabytes, and reading them into base64 blocks exhausts the daemon.
+
+Every limit here is checked against the archive's own directory before any
+entry is read, and the read itself is bounded as well, because a ZIP header
+can under-report an entry's size. An entry past a limit is refused with a
+line saying so; nothing is silently cut short.
+"""
+
+from __future__ import annotations
+
+import base64
+import io
+import logging
+import zipfile
+
+logger = logging.getLogger(__name__)
+
+TEXT_EXTENSIONS: frozenset[str] = frozenset({
+    ".txt", ".py", ".js", ".ts", ".jsx", ".tsx", ".json", ".yaml", ".yml",
+    ".toml", ".xml", ".html", ".htm", ".css", ".scss", ".less",
+    ".md", ".rst", ".csv", ".tsv", ".sql", ".sh", ".bash", ".zsh",
+    ".rb", ".go", ".rs", ".java", ".kt", ".c", ".cpp", ".h", ".hpp",
+    ".swift", ".lua", ".r", ".m", ".pl", ".php", ".env", ".ini", ".cfg",
+    ".conf", ".log", ".diff", ".patch", ".vue", ".svelte",
+})
+
+IMAGE_EXT_TO_MIME: dict[str, str] = {
+    ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
+    ".png": "image/png", ".gif": "image/gif", ".webp": "image/webp",
+}
+
+# Inline text budget for the whole archive.
+MAX_TEXT_SIZE = 512 * 1024
+# Files in one archive. A directory listing longer than this is a machine
+# dump, not something a person meant to show the agent.
+MAX_ENTRIES = 100
+# Uncompressed bytes for one entry, and for the archive as a whole.
+# Images and PDFs reach the model as base64, which is 4/3 of the bytes
+# read, so the total is set from what the prompt can carry after that
+# expansion. The channels cap the download at about 20 MB of compressed
+# bytes, which says nothing about what the archive expands to.
+MAX_ENTRY_SIZE = 5_000_000
+MAX_TOTAL_SIZE = 12_000_000
+# Uncompressed / compressed for one entry, checked only above RATIO_FLOOR.
+# Small files reach a high ratio for ordinary reasons, such as generated
+# code or a log of one repeated line, and refusing those loses real
+# content. MAX_ENTRY_SIZE and MAX_TOTAL_SIZE already bound what any entry
+# adds to the prompt, so the ratio only has to catch the large entries.
+MAX_RATIO = 100
+RATIO_FLOOR = 1_000_000
+
+
+class _EntryTooLarge(Exception):
+    """An entry produced more bytes than its directory record promised."""
+
+
+def _read_bounded(zf: zipfile.ZipFile, info: zipfile.ZipInfo, limit: int) -> bytes:
+    """Read one entry, refusing more than *limit* bytes.
+
+    The bound is on what comes out, not only on the size the central
+    directory declares, so the caller's budget holds even for an archive
+    whose records do not describe its contents.
+    """
+    with zf.open(info) as handle:
+        raw = handle.read(limit + 1)
+    if len(raw) > limit:
+        raise _EntryTooLarge(info.filename)
+    return raw
+
+
+def _refusal(info: zipfile.ZipInfo, reason: str) -> str:
+    return f"- {info.filename} ({info.file_size} bytes) [{reason}]"
+
+
+def extract_zip(data: bytes, meta_line: str) -> tuple[list[dict[str, str]], str]:
+    """Unpack a ZIP one level — text inline, images and PDFs as blocks.
+
+    Returns ``(content_blocks, context_text)``. ``meta_line`` is the caller's
+    one-line description of the archive and heads the context text.
+    """
+    buf = io.BytesIO(data)
+    if not zipfile.is_zipfile(buf):
+        return [], f"{meta_line}\n(Invalid or corrupted ZIP archive)"
+    buf.seek(0)
+
+    blocks: list[dict[str, str]] = []
+    parts: list[str] = [meta_line]
+    try:
+        with zipfile.ZipFile(buf) as zf:
+            entries = [
+                i for i in zf.infolist()
+                if not i.is_dir() and not i.filename.startswith("__MACOSX/")
+            ]
+            if len(entries) > MAX_ENTRIES:
+                return [], (
+                    f"{meta_line}\n(Archive holds {len(entries)} files; "
+                    f"the limit is {MAX_ENTRIES})"
+                )
+
+            parts.append(f"Archive contains {len(entries)} file(s):")
+            total_text = 0
+            total_read = 0
+            for info in entries:
+                name = info.filename
+                size = info.file_size
+                ext = ""
+                if "." in name.rsplit("/", 1)[-1]:
+                    ext = "." + name.rsplit(".", 1)[-1].lower()
+                wanted = ext in TEXT_EXTENSIONS or ext in IMAGE_EXT_TO_MIME or ext == ".pdf"
+
+                if not wanted:
+                    parts.append(f"- {name} ({size} bytes)")
+                    continue
+
+                # Every bound below is read off the central directory, so a
+                # refusal costs nothing but the listing itself.
+                if size > MAX_ENTRY_SIZE:
+                    parts.append(_refusal(info, "too large to read"))
+                    continue
+                if (
+                    size > RATIO_FLOOR
+                    and info.compress_size
+                    and size / info.compress_size > MAX_RATIO
+                ):
+                    logger.warning(
+                        "Refusing ZIP entry %s: %d bytes from %d compressed",
+                        name, size, info.compress_size,
+                    )
+                    parts.append(_refusal(info, "compression ratio too high"))
+                    continue
+                # The type-specific budget is checked first so a refusal names
+                # the limit the entry actually hit.
+                if ext in TEXT_EXTENSIONS and total_text + size > MAX_TEXT_SIZE:
+                    parts.append(_refusal(info, "text, too large to inline"))
+                    continue
+                if total_read + size > MAX_TOTAL_SIZE:
+                    parts.append(_refusal(info, "archive size budget spent"))
+                    continue
+
+                try:
+                    raw = _read_bounded(zf, info, size)
+                except _EntryTooLarge:
+                    logger.warning(
+                        "Refusing ZIP entry %s: it expands past the %d bytes "
+                        "its header declares", name, size,
+                    )
+                    parts.append(_refusal(info, "larger than it declares"))
+                    continue
+                except RuntimeError:
+                    # A password-protected archive fails the same way on
+                    # every entry, so it is reported once, below.
+                    raise
+                except Exception:
+                    parts.append(_refusal(info, "read error"))
+                    continue
+                total_read += len(raw)
+
+                if ext in TEXT_EXTENSIONS:
+                    total_text += len(raw)
+                    parts.append(
+                        f"--- {name} ({size} bytes) ---\n"
+                        f"```\n{raw.decode('utf-8', errors='replace')}\n```"
+                    )
+                else:
+                    is_pdf = ext == ".pdf"
+                    blocks.append({
+                        "type": "base64",
+                        "media_type": (
+                            "application/pdf" if is_pdf else IMAGE_EXT_TO_MIME[ext]
+                        ),
+                        "data": base64.b64encode(raw).decode("utf-8"),
+                    })
+                    parts.append(
+                        f"- {name} ({size} bytes) [{'PDF' if is_pdf else 'image'}]"
+                    )
+    except zipfile.BadZipFile:
+        return [], f"{meta_line}\n(Invalid or corrupted ZIP archive)"
+    except RuntimeError as e:
+        # Password-protected archives.
+        return [], f"{meta_line}\n(Cannot extract: {e})"
+
+    return blocks, "\n".join(parts)