Opencode/maintainer improvements 20261018

docs/security/m15-authority-coverage.json

@@ -2167,14 +2167,14 @@
   },
   {
    "file": "isycode/tui_app_sessions.py",
-   "line": 934,
+   "line": 963,
    "callable": "isycode.tui_app_sessions.SessionMixin._delete_chat_session",
    "action": "session.delete",
    "owner": "session_delete"
   },
   {
    "file": "isycode/tui_app_sessions.py",
-   "line": 974,
+   "line": 1003,
    "callable": "isycode.tui_app_sessions.SessionMixin._delete_iteration_session",
    "action": "session.delete",
    "owner": "session_delete"

src/isycode/harness_graph.py

@@ -193,3 +193,44 @@ def repair_compose(semantic_id: str, harnesses: Iterable[str]) -> str:
             "human controls and limitations. Keep unrelated files intact. If the runtime "
             "cannot validate the required capability, stop before that effect and present "
             "the concrete change for review. Do not claim a gap fixed until verified.\n")
+
+
+def repair_compose_all(rows: Iterable[dict[str, Any]]) -> str:
+    """Build one unified review brief covering every actionable gap in the map."""
+    actionable = [row for row in rows if row["status"] in {"ADD", "WATCH"}]
+    if not actionable:
+        return "No gaps to repair · the harness map is aligned."
+    parts = [f"ISyCode harness repair proposal: all gaps\n"
+             f"Meaning: unified brief for every actionable gap in the Multi Harness map\n"
+             f"Gaps covered: {len(actionable)}\n"]
+    for row in actionable:
+        parts.append("\n---\n")
+        parts.append(repair_compose(row["semantic_id"], row["harnesses"]))
+    return "".join(parts)
+
+
+def repair_targets(rows: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
+    """List every cli/file touched by an actionable gap, for the confirm screen.
+
+    Each entry is ``{"semantic_id": str, "title": str, "harnesses": tuple[str, ...],
+    "status": str, "files": [str, ...]}``. Files come from the seed option's
+    ``isycode_target``; ``"absent"`` means there is no native file yet.
+    """
+    targets: list[dict[str, Any]] = []
+    for row in rows:
+        if row["status"] not in {"ADD", "WATCH"}:
+            continue
+        option = SEED_OPTIONS.get(row["semantic_id"])
+        raw_target = option.isycode_target if option is not None else row.get("target_label", "")
+        if not raw_target or raw_target == "absent":
+            files: list[str] = []
+        else:
+            files = [item.strip() for item in raw_target.split("|") if item.strip()]
+        targets.append({
+            "semantic_id": row["semantic_id"],
+            "title": row["title"],
+            "harnesses": tuple(row["harnesses"]),
+            "status": row["status"],
+            "files": files,
+        })
+    return targets

src/isycode/hive/__init__.py

@@ -0,0 +1,41 @@
+"""ISyCode Hive — on-disk multi-agent coordination layer.
+
+A git-backed mailbox system where agents exchange messages via inbox/outbox,
+coordinated by a single-committer router. Human-in-the-loop approval surfaces
+in the agent's own terminal (ISyCode lane) and can be approved remotely.
+
+Design mirrors isyco-worlds/src/main/hive.ts but in Python, integrated with
+ISyCode's lanes, Sentinel, and the harness compose pipeline.
+"""
+from __future__ import annotations
+
+from .manager import HiveManager, HiveConfig
+from .types import (
+    HiveMessage,
+    HiveTask,
+    AgentMeta,
+    Registry,
+    RegistryAgent,
+    MessageAct,
+    SpawnInjection,
+)
+from .router import Router
+from .guard import hive_write_decision, HiveWriteInput
+from .spawn import spawn_hive_agent, SpawnResult
+
+__all__ = [
+    "HiveManager",
+    "HiveConfig",
+    "HiveMessage",
+    "HiveTask",
+    "AgentMeta",
+    "Registry",
+    "RegistryAgent",
+    "MessageAct",
+    "SpawnInjection",
+    "Router",
+    "hive_write_decision",
+    "HiveWriteInput",
+    "spawn_hive_agent",
+    "SpawnResult",
+]

src/isycode/hive/guard.py

@@ -0,0 +1,90 @@
+"""Hive write guard — refuses file writes outside the agent's own hive workspace.
+
+Ported from isyco-worlds/src/main/hiveGuard.ts. Only guards Claude's file tools
+(Write, Edit, MultiEdit, NotebookEdit). Shell commands that write files are not
+judged from their text. The PreToolUse hook fails OPEN when the shim can't
+reach the app's socket, so this is a strong guard, not an airtight one.
+"""
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any
+
+from .types import HiveWriteInput, HiveWriteDecision
+
+GUARDED_TOOLS: frozenset[str] = frozenset({"Write", "Edit", "MultiEdit", "NotebookEdit"})
+
+
+def target_path(tool_input: dict[str, Any] | None, cwd: str | None) -> str | None:
+    """Return the absolute path a tool call would write, or None if it names none."""
+    if not tool_input or not isinstance(tool_input, dict):
+        return None
+    p = tool_input.get("file_path") or tool_input.get("notebook_path")
+    if not isinstance(p, str):
+        return None
+    if os.path.isabs(p):
+        return os.path.normpath(p)
+    if cwd:
+        return os.path.normpath(os.path.join(cwd, p))
+    return None
+
+
+def is_inside(parent: str, child: str, case_insensitive: bool) -> bool:
+    """Check if child path is inside parent directory."""
+    norm = (lambda p: p.lower()) if case_insensitive else (lambda p: p)
+    parent_n = os.path.normpath(norm(parent))
+    child_n = os.path.normpath(norm(child))
+    try:
+        rel = os.path.relpath(child_n, parent_n)
+    except ValueError:
+        return False
+    return rel == "." or (not rel.startswith("..") and not os.path.isabs(rel))
+
+
+def is_plumbing(rel: str, agent_id: str, is_god: bool) -> bool:
+    """Files the hive protocol allows an agent to write."""
+    own = f"agents/{agent_id}/"
+    if rel == f"{own}memory.md":
+        return True
+    if rel.startswith(f"{own}outbox/") or rel.startswith(f"{own}inbox/"):
+        return True
+    if rel == "tasks.json":
+        return True
+    if is_god and (rel == "board.md" or rel.startswith("spawn-requests/")):
+        return True
+    return False
+
+
+def hive_write_decision(i: HiveWriteInput) -> HiveWriteDecision:
+    """Decide whether to deny a file write inside the hive."""
+    if i["tool"] not in GUARDED_TOOLS:
+        return {"deny": False, "reason": None}
+
+    target = target_path(i["toolInput"], i.get("cwd"))
+    if not target:
+        return {"deny": False, "reason": None}
+
+    norm = (lambda p: p.lower()) if i["caseInsensitive"] else (lambda p: p)
+    hive_root_n = os.path.normpath(norm(i["hiveRoot"]))
+    target_n = os.path.normpath(norm(target))
+
+    if not is_inside(hive_root_n, target_n, i["caseInsensitive"]):
+        return {"deny": False, "reason": None}
+
+    try:
+        rel = os.path.relpath(target_n, hive_root_n).replace(os.sep, "/")
+    except ValueError:
+        return {"deny": False, "reason": None}
+
+    if is_plumbing(rel, i["agentId"], i["isGod"]):
+        return {"deny": False, "reason": None}
+
+    reason = (
+        "The hive folder is only for coordination: your own memory.md, inbox and outbox, "
+        "and tasks.json"
+        + (" (plus board.md and spawn-requests/ for you)" if i["isGod"] else "")
+        + ". To reach another agent, write a message into your outbox. "
+        "Save documents and other work in your working folder instead."
+    )
+    return {"deny": True, "reason": reason}

src/isycode/hive/hook_server.py

@@ -0,0 +1,170 @@
+"""Hook server — Unix domain socket server that receives hook events from agents.
+
+Runs in the ISyCode main process (like the Electron main in isyco-worlds).
+Handles PreToolUse (write guard), PostToolUse (live status), Stop (inbox drain).
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import socket
+import time
+from pathlib import Path
+from typing import Any
+
+from .manager import HiveManager, HiveConfig
+from .guard import hive_write_decision, HiveWriteInput
+from .types import HiveMessage, MessageAct
+
+
+class HookServer:
+    """Async Unix domain socket server for hive hooks."""
+
+    def __init__(self, manager: HiveManager) -> None:
+        self.manager = manager
+        self._server: asyncio.Server | None = None
+        self._task: asyncio.Task | None = None
+        self._running = False
+
+    async def start(self) -> None:
+        """Start the hook server."""
+        sock_path = self.manager.sock_path()
+        if not sock_path or os.name == "nt":
+            # Windows: named pipes not supported in stdlib asyncio yet
+            # Fallback: file-based polling or skip
+            print("[hive] Hook server not started on Windows (no UDS support)")
+            return
+
+        # Remove stale socket
+        try:
+            os.unlink(sock_path)
+        except OSError:
+            pass
+
+        self._server = await asyncio.start_unix_server(
+            self._handle_client, sock_path
+        )
+        self._running = True
+        self._task = asyncio.create_task(self._serve())
+        print(f"[hive] Hook server listening on {sock_path}")
+
+    async def stop(self) -> None:
+        self._running = False
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+        if self._server:
+            self._server.close()
+            await self._server.wait_closed()
+        sock_path = self.manager.sock_path()
+        if sock_path:
+            try:
+                os.unlink(sock_path)
+            except OSError:
+                pass
+
+    async def _serve(self) -> None:
+        if self._server:
+            async with self._server:
+                await self._server.serve_forever()
+
+    async def _handle_client(
+        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+    ) -> None:
+        try:
+            while self._running:
+                line = await reader.readline()
+                if not line:
+                    break
+                try:
+                    payload = json.loads(line.decode("utf-8"))
+                except json.JSONDecodeError:
+                    continue
+                await self._dispatch(payload)
+        except Exception:
+            pass
+        finally:
+            writer.close()
+            await writer.wait_closed()
+
+    async def _dispatch(self, payload: dict[str, Any]) -> None:
+        """Dispatch hook event to appropriate handler."""
+        evt_type = payload.get("type")
+        agent_id = payload.get("agent_id")
+
+        if not agent_id or agent_id not in self.manager.registry.agents:
+            return
+
+        agent = self.manager.registry.agents[agent_id]
+
+        if evt_type == "pre_tool_use":
+            await self._handle_pre_tool_use(payload, agent)
+        elif evt_type == "post_tool_use":
+            await self._handle_post_tool_use(payload, agent)
+        elif evt_type == "stop":
+            await self._handle_stop(payload, agent)
+
+    async def _handle_pre_tool_use(self, payload: dict[str, Any], agent: RegistryAgent) -> None:
+        """PreToolUse: check write guard, update agent status."""
+        tool_name = payload.get("tool_name", "")
+        tool_input = payload.get("tool_input", {})
+        cwd = payload.get("cwd", agent.cwd or os.getcwd())
+
+        # Write guard check
+        if tool_name in {"Write", "Edit", "MultiEdit", "NotebookEdit"}:
+            inp: HiveWriteInput = {
+                "tool": tool_name,
+                "toolInput": tool_input,
+                "cwd": cwd,
+                "agentId": agent.id,
+                "isGod": agent.is_god,
+                "hiveRoot": str(self.manager.hive_root),
+                "caseInsensitive": os.name == "nt",
+            }
+            decision = hive_write_decision(inp)
+            if decision["deny"]:
+                # Send denial back via a response channel (not implemented yet)
+                print(f"[hive] DENY {agent.id} {tool_name}: {decision['reason']}")
+                # For now, we can't easily send back to the hook; the guard
+                # in isyco-worlds works because the hook reads the decision
+                # from the shim's stdout. Our shim does that.
+                return
+
+        # Update status to working
+        self.manager.set_agent_status(agent.id, "working")
+        self.manager._append_log({
+            "kind": "tool_start",
+            "agentId": agent.id,
+            "tool": tool_name,
+            "ts": time.time(),
+        })
+
+    async def _handle_post_tool_use(self, payload: dict[str, Any], agent: RegistryAgent) -> None:
+        """PostToolUse: update status, log."""
+        tool_name = payload.get("tool_name", "")
+        self.manager.set_agent_status(agent.id, "idle")
+        self.manager._append_log({
+            "kind": "tool_end",
+            "agentId": agent.id,
+            "tool": tool_name,
+            "ts": time.time(),
+        })
+
+    async def _handle_stop(self, payload: dict[str, Any], agent: RegistryAgent) -> None:
+        """Stop: session idle → drain inbox (handled by router, but we can nudge)."""
+        self.manager.set_agent_status(agent.id, "idle")
+        session_id = payload.get("session_id")
+        if session_id:
+            self.manager.set_agent_session(agent.id, session_id)
+        self.manager._append_log({
+            "kind": "session_stop",
+            "agentId": agent.id,
+            "sessionId": session_id,
+            "ts": time.time(),
+        })
+        # Nudge router to process inbox immediately
+        # (router runs on interval anyway)