feat(kanban): govern raw intake decomposition

AGENTS.md

@@ -1284,14 +1284,15 @@ def profile_env(tmp_path, monkeypatch):
 ### Python
 **ALWAYS use `scripts/run_tests.sh`** — do not call `pytest` directly. The script enforces
 hermetic environment parity with CI (unset credential vars, TZ=UTC, LANG=C.UTF-8,
-`-n auto` xdist workers, in-tree subprocess-isolation plugin). Direct `pytest`
+per-file subprocess isolation via `scripts/run_tests_parallel.py` — no xdist,
+worker count auto-scaled from CPU count). Direct `pytest`
 on a 16+ core developer machine with API keys set diverges from CI in ways
 that have caused multiple "works locally, fails in CI" incidents (and the reverse).
 
 ```bash
 scripts/run_tests.sh                                  # full suite, CI-parity
 scripts/run_tests.sh tests/gateway/                   # one directory
-scripts/run_tests.sh tests/agent/test_foo.py::test_x  # one test
+scripts/run_tests.sh tests/agent/test_foo.py -k test_x  # one test (file + -k; the runner is file-granular)
 scripts/run_tests.sh -v --tb=long                     # pass-through pytest flags
 ```
 

CONTRIBUTING.md

@@ -201,7 +201,8 @@ ln -sf "$(pwd)/venv/bin/hermes" ~/.local/bin/hermes
 ### Run tests
 
 ```bash
-# Preferred — matches CI (hermetic env, 4 xdist workers); see AGENTS.md
+# Preferred — matches CI (hermetic `env -i`, per-file subprocess isolation
+# via run_tests_parallel.py, worker count auto-scaled); see AGENTS.md
 scripts/run_tests.sh
 
 # Alternative (activate the venv first). The wrapper is still recommended
@@ -848,7 +849,7 @@ that touches the OS, assume *any* platform can hit your code path.
 Tests that use POSIX-only syscalls need a skip marker. Common ones:
 - Symlinks → `@pytest.mark.skipif(sys.platform == "win32", ...)`
 - `0o600` file modes → `@pytest.mark.skipif(sys.platform.startswith("win"), ...)`
-- `signal.SIGALRM` → Unix-only (see `tests/conftest.py::_enforce_test_timeout`)
+- `signal.SIGALRM` → Unix-only (per-test timeouts no longer use it directly; see the win32 timeout-method shim in `tests/conftest.py::pytest_configure`)
 - `os.setsid` / `os.fork` → Unix-only
 - Live Winsock / Windows-specific regression tests →
   `@pytest.mark.skipif(sys.platform != "win32", reason="Windows-specific regression")`

cli.py

@@ -3235,10 +3235,12 @@ def _termux_example_image_path(filename: str = "cat.png") -> str:
         "/storage/emulated/0",
         "/storage/self/primary",
     ]
+    # Termux/Android roots are POSIX paths — join with literal forward
+    # slashes so the hint stays correct even when this renders on Windows.
     for root in candidates:
         if os.path.isdir(root):
-            return os.path.join(root, "Pictures", filename)
-    return os.path.join("~/storage/shared", "Pictures", filename)
+            return f"{root}/Pictures/{filename}"
+    return f"~/storage/shared/Pictures/{filename}"
 
 
 def _split_path_input(raw: str) -> tuple[str, str]:
@@ -3309,6 +3311,16 @@ def _resolve_attachment_path(raw_path: str) -> Path | None:
                 expanded = unquote(parsed.path or "")
                 if parsed.netloc and os.name == "nt":
                     expanded = f"//{parsed.netloc}{expanded}"
+                elif (
+                    os.name == "nt"
+                    and len(expanded) >= 3
+                    and expanded[0] == "/"
+                    and expanded[1].isalpha()
+                    and expanded[2] == ":"
+                ):
+                    # file:///C:/... parses to path "/C:/..." — drop the
+                    # leading slash so it resolves as a drive-letter path.
+                    expanded = expanded[1:]
         except Exception:
             expanded = token
     expanded = os.path.expandvars(os.path.expanduser(expanded))

gateway/status.py

@@ -507,9 +507,12 @@ def _command_line_belongs_to_profile(command: str, profile_home: Path) -> bool:
     explicit ``HERMES_HOME=<path>``) on its argv; the default/root gateway runs
     bare with no profile flag.
     """
-    command_lc = command.lower()
+    # Normalize separators before the substring match: on Windows,
+    # str(Path) renders backslashes while a HERMES_HOME= value on the argv
+    # may carry forward slashes (Git Bash, JSON configs) — and vice versa.
+    command_lc = command.lower().replace("\\", "/")
     profile_name = _profile_name_for_home(profile_home)
-    home_lc = str(profile_home).lower()
+    home_lc = str(profile_home).lower().replace("\\", "/")
 
     if profile_name is not None and profile_name != "default":
         profile_lc = profile_name.lower()

hermes_cli/banner.py

@@ -40,7 +40,14 @@ def cprint(text: str):
     """Print ANSI-colored text through prompt_toolkit's renderer."""
     from prompt_toolkit import print_formatted_text as _pt_print
     from prompt_toolkit.formatted_text import ANSI as _PT_ANSI
-    _pt_print(_PT_ANSI(text))
+    try:
+        _pt_print(_PT_ANSI(text))
+    except Exception:
+        # prompt_toolkit needs a real console. On Windows, a redirected or
+        # absent stdout (pythonw.exe, CI, `hermes ... > file`) raises
+        # NoConsoleScreenBufferError from its Win32Output — display helpers
+        # must never crash the caller over that, so degrade to plain print.
+        print(text)
 
 
 # =========================================================================

hermes_cli/browser_connect.py

@@ -5,6 +5,7 @@
 import logging
 import os
 import platform
+import posixpath
 import shlex
 import shutil
 import subprocess
@@ -95,7 +96,10 @@ def add_windows_install_paths(
         for _, group in install_groups:
             for base in filter(None, bases):
                 for parts in group:
-                    add(os.path.join(base, *parts))
+                    # Only called with WSL ``/mnt/c/...`` bases — those are
+                    # POSIX paths regardless of the host OS, so join with
+                    # posixpath (os.path.join would emit backslashes on nt).
+                    add(posixpath.join(base, *parts))
 
     if system == "Darwin":
         for app in _DARWIN_APPS:

hermes_cli/config_defaults.py

@@ -2176,6 +2176,11 @@
         # assignee to any installed profile. When unset, falls back to the
         # default profile. A task never ends up with assignee=None.
         "default_assignee": "",
+        # Governed raw-intake rules. Empty values use the built-in safe
+        # vocabulary and route ambiguity to PPMA, never the launch profile.
+        "intake_fanout_cap": 6,
+        "intake_allowed_domains": [],
+        "intake_allowed_assignees": [],
         # Per-profile concurrency cap (#21582). When set to a positive int,
         # no single profile can have more than N workers running at once,
         # even if the global max_in_progress / max_spawn caps would allow

hermes_cli/gateway.py

@@ -359,15 +359,17 @@ def _scan_gateway_pids(
         looks_like_gateway_runtime_command_line,
     )
     current_home = str(get_hermes_home().resolve())
-    current_home_lc = current_home.lower()
+    # Forward slashes on both sides of the HERMES_HOME= match — see
+    # gateway.status._command_line_belongs_to_profile, which this mirrors.
+    current_home_lc = current_home.lower().replace("\\", "/")
     current_profile_arg = _profile_arg(current_home)
     current_profile_name = (
         current_profile_arg.split()[-1] if current_profile_arg else ""
     )
     current_profile_name_lc = current_profile_name.lower()
 
     def _matches_current_profile(command: str) -> bool:
-        command_lc = command.lower()
+        command_lc = command.lower().replace("\\", "/")
         if current_profile_name:
             return (
                 f"--profile {current_profile_name_lc}" in command_lc

hermes_cli/kanban.py

@@ -1519,30 +1519,49 @@ def _cmd_create(args: argparse.Namespace) -> int:
         )
         return 2
     with kb.connect_closing() as conn:
-        task_id = kb.create_task(
-            conn,
-            title=args.title,
-            body=args.body,
-            assignee=args.assignee,
-            created_by=args.created_by or _profile_author(),
-            workspace_kind=ws_kind,
-            workspace_path=ws_path,
-            branch_name=branch_name,
-            project_id=getattr(args, "project", None),
-            tenant=args.tenant,
-            priority=args.priority,
-            parents=tuple(args.parent or ()),
-            triage=bool(getattr(args, "triage", False)),
-            idempotency_key=getattr(args, "idempotency_key", None),
-            max_runtime_seconds=max_runtime,
-            skills=getattr(args, "skills", None) or None,
-            max_retries=max_retries,
-            model_override=getattr(args, "model_override", None),
-            provider_override=getattr(args, "provider_override", None),
-            goal_mode=bool(getattr(args, "goal_mode", False)),
-            goal_max_turns=getattr(args, "goal_max_turns", None),
-            initial_status=getattr(args, "initial_status", "running"),
-        )
+        intake_envelope = None
+        if getattr(args, "triage", False) and args.body:
+            from hermes_cli import kanban_intake
+            try:
+                intake_envelope = kanban_intake.parse_envelope(args.body)
+            except ValueError as exc:
+                print(f"kanban: invalid intake envelope: {exc}", file=sys.stderr)
+                return 2
+        if intake_envelope:
+            task_id, _ = kb.create_governed_intake_task(
+                conn,
+                title=args.title,
+                body=args.body,
+                tenant=intake_envelope.tenant_domain,
+                idempotency_key=intake_envelope.idempotency_key,
+                created_by=args.created_by or _profile_author(),
+                priority=args.priority,
+            )
+        else:
+            task_id = kb.create_task(
+                conn,
+                title=args.title,
+                body=args.body,
+                assignee=args.assignee,
+                created_by=args.created_by or _profile_author(),
+                workspace_kind=ws_kind,
+                workspace_path=ws_path,
+                branch_name=branch_name,
+                project_id=getattr(args, "project", None),
+                tenant=args.tenant,
+                priority=args.priority,
+                parents=tuple(args.parent or ()),
+                triage=bool(getattr(args, "triage", False)),
+                idempotency_key=getattr(args, "idempotency_key", None),
+                max_runtime_seconds=max_runtime,
+                skills=getattr(args, "skills", None) or None,
+                max_retries=max_retries,
+                model_override=getattr(args, "model_override", None),
+                provider_override=getattr(args, "provider_override", None),
+                goal_mode=bool(getattr(args, "goal_mode", False)),
+                goal_max_turns=getattr(args, "goal_max_turns", None),
+                initial_status=getattr(args, "initial_status", "running"),
+            )
         task = kb.get_task(conn, task_id)
     if getattr(args, "json", False):
         print(json.dumps(_task_to_dict(task), indent=2, ensure_ascii=False))
@@ -2127,7 +2146,16 @@ def _cmd_comment(args: argparse.Namespace) -> int:
             body = body[: max(0, args.max_len - len(suffix))].rstrip() + suffix
     author = args.author or _profile_author()
     with kb.connect_closing() as conn:
-        kb.add_comment(conn, args.task_id, author, body)
+        task = kb.get_task(conn, args.task_id)
+        if task and task.assignee == "paul-park" and body.startswith("LINEAR_SCOPE:"):
+            kb.record_scope_handoff(
+                conn,
+                args.task_id,
+                author=author,
+                body=body,
+            )
+        else:
+            kb.add_comment(conn, args.task_id, author, body)
     print(f"Comment added to {args.task_id}")
     return 0
 

hermes_cli/kanban_db.py

@@ -3608,6 +3608,82 @@ def add_comment(
         return int(cur.lastrowid or 0)
 
 
+_LINEAR_SCOPE_RE = re.compile(
+    r"LINEAR_SCOPE:\s*parent=(?P<parent>[A-Z][A-Z0-9]+-\d+)\s+"
+    r"subissues=\[(?P<subissues>.+)\]",
+    re.DOTALL,
+)
+_LINEAR_SUBISSUE_RE = re.compile(
+    r"(?:key\s*[:=]\s*)?(?P<key>[A-Z][A-Z0-9]+-\d+)\s*"
+    r"[,;]\s*cptc\s*[:=]\s*(?P<cptc>1|2|3|5|8|13)\b",
+    re.IGNORECASE,
+)
+
+
+def parse_linear_scope(body: str) -> Optional[dict]:
+    """Parse PPMA's structured Linear/CPTC handoff, or return ``None``."""
+    match = _LINEAR_SCOPE_RE.search(body or "")
+    if not match:
+        return None
+    subissues = [
+        {"key": item.group("key").upper(), "cptc": int(item.group("cptc"))}
+        for item in _LINEAR_SUBISSUE_RE.finditer(match.group("subissues"))
+    ]
+    if not subissues:
+        return None
+    return {"parent": match.group("parent").upper(), "subissues": subissues}
+
+
+def record_scope_handoff(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    author: str,
+    body: str,
+) -> dict:
+    """Persist a validated PPMA scope comment and typed handoff events."""
+    scope = parse_linear_scope(body)
+    if scope is None:
+        raise ValueError(
+            "PPMA scope handoff must contain LINEAR_SCOPE with at least one "
+            "{key, cptc} technical sub-issue"
+        )
+    now = int(time.time())
+    with write_txn(conn):
+        task = conn.execute(
+            "SELECT status, assignee FROM tasks WHERE id = ?", (task_id,)
+        ).fetchone()
+        if task is None:
+            raise ValueError(f"unknown task {task_id}")
+        if task["assignee"] != "paul-park":
+            raise ValueError("scope handoff is only valid for a PPMA gate task")
+        existing = conn.execute(
+            "SELECT 1 FROM task_events WHERE task_id = ? AND kind = 'scope_recorded'",
+            (task_id,),
+        ).fetchone()
+        if existing:
+            return scope
+        conn.execute(
+            "INSERT INTO task_comments (task_id, author, body, created_at) "
+            "VALUES (?, ?, ?, ?)",
+            (task_id, author.strip(), body.strip(), now),
+        )
+        payload = {"schema_version": 1, **scope, "recorded_by": author.strip()}
+        _append_event(conn, task_id, "scope_recorded", payload)
+        _append_event(
+            conn,
+            task_id,
+            "handoff_emitted",
+            {
+                "schema_version": 1,
+                "handoff_kind": "linear_scope",
+                "downstream_task_ids": child_ids(conn, task_id),
+                "scope_parent": scope["parent"],
+            },
+        )
+    return scope
+
+
 def list_comments(conn: sqlite3.Connection, task_id: str) -> list[Comment]:
     rows = conn.execute(
         "SELECT * FROM task_comments WHERE task_id = ? ORDER BY created_at ASC",
@@ -6455,6 +6531,60 @@ def decompose_triage_task(
     return child_ids
 
 
+def create_governed_intake_task(
+    conn: sqlite3.Connection,
+    *,
+    title: str,
+    body: str,
+    tenant: str,
+    idempotency_key: str,
+    created_by: Optional[str] = None,
+    priority: int = 0,
+) -> tuple[str, bool]:
+    """Atomically deduplicate and create one governed raw-intake task.
+
+    ``create_task`` intentionally has legacy best-effort idempotency semantics.
+    Governed intake needs a stronger contract because duplicate webhook/feed
+    deliveries may race. Serialize the lookup+insert under one write transaction
+    without changing the compatibility behavior of the general task API.
+    """
+    if not idempotency_key or not idempotency_key.strip():
+        raise ValueError("governed intake requires an idempotency_key")
+    now = int(time.time())
+    with write_txn(conn):
+        existing = conn.execute(
+            "SELECT id FROM tasks WHERE idempotency_key = ? "
+            "AND status != 'archived' ORDER BY created_at DESC LIMIT 1",
+            (idempotency_key.strip(),),
+        ).fetchone()
+        if existing:
+            return existing["id"], False
+        task_id = _new_task_id()
+        conn.execute(
+            "INSERT INTO tasks "
+            "(id, title, body, status, workspace_kind, tenant, priority, "
+            " created_at, created_by, idempotency_key) "
+            "VALUES (?, ?, ?, 'triage', 'scratch', ?, ?, ?, ?, ?)",
+            (
+                task_id,
+                title.strip(),
+                body,
+                tenant.strip(),
+                int(priority),
+                now,
+                created_by,
+                idempotency_key.strip(),
+            ),
+        )
+        _append_event(
+            conn,
+            task_id,
+            "created",
+            {"by": created_by, "assignee": None, "status": "triage", "parents": []},
+        )
+        return task_id, True
+
+
 def archive_task(conn: sqlite3.Connection, task_id: str) -> bool:
     with write_txn(conn):
         cur = conn.execute(