From f8322123f63e3e1f11441da60754916caf22d7d1 Mon Sep 17 00:00:00 2001 From: Jeff Watts <186512915+lEWFkRAD@users.noreply.github.com> Date: Wed, 29 Jul 2026 21:42:36 -0700 Subject: [PATCH 1/3] fix(windows): native Windows correctness for CLI, gateway status, banner, and WSL browser paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Salvaged from #57016 by @lEWFkRAD: - cli.py: handle file:///C:/... drive-letter URIs on nt (strip the leading slash urlparse leaves); join Termux example paths with literal forward slashes so hints stay POSIX on Windows. - gateway/status.py + hermes_cli/gateway.py: normalize backslashes to forward slashes before the HERMES_HOME substring match so separator style cannot defeat profile ownership detection. - hermes_cli/banner.py: cprint degrades to plain print when prompt_toolkit has no console (NoConsoleScreenBufferError on redirected/absent Windows stdout). - hermes_cli/browser_connect.py: posixpath.join for WSL /mnt/c/... bases (os.path.join would emit backslashes on nt). - Test hardening: symlink skip-guards, USERPROFILE alongside HOME for ntpath.expanduser, SIGKILL absence skipif fixed via monkeypatch, drive-letter URI / separator-normalization / banner-fallback coverage. Dropped from the original PR: tests/cli/conftest.py fixture and the AppSession _output monkeypatch — main's merged tests/cli/conftest.py already handles that prompt_toolkit pollution. --- cli.py | 16 ++++++++++++++-- gateway/status.py | 7 +++++-- hermes_cli/banner.py | 9 ++++++++- hermes_cli/browser_connect.py | 6 +++++- hermes_cli/gateway.py | 6 ++++-- tests/cli/test_cli_browser_connect.py | 11 +++++++++++ tests/cli/test_cli_file_drop.py | 16 ++++++++++++++++ tests/cli/test_cli_image_command.py | 2 ++ tests/cli/test_worktree.py | 15 +++++++++++++++ tests/cli/test_worktree_security.py | 16 ++++++++++++++++ tests/gateway/test_status.py | 9 +++++++++ tests/hermes_cli/test_banner.py | 10 ++++++++++ tests/tools/test_windows_native_support.py | 4 ++-- 13 files changed, 117 insertions(+), 10 deletions(-) diff --git a/cli.py b/cli.py index bb91aa4fc37d..a4ce0cdd56ae 100644 --- a/cli.py +++ b/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)) diff --git a/gateway/status.py b/gateway/status.py index 5787a43821de..ce02648a958f 100644 --- a/gateway/status.py +++ b/gateway/status.py @@ -507,9 +507,12 @@ def _command_line_belongs_to_profile(command: str, profile_home: Path) -> bool: explicit ``HERMES_HOME=``) 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() diff --git a/hermes_cli/banner.py b/hermes_cli/banner.py index 4d2b077ec964..7811899aeb3f 100644 --- a/hermes_cli/banner.py +++ b/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) # ========================================================================= diff --git a/hermes_cli/browser_connect.py b/hermes_cli/browser_connect.py index 4fcc4cc63c5f..af1b04eaee0a 100644 --- a/hermes_cli/browser_connect.py +++ b/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: diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 067b2c3209b9..55b8a196f14c 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -359,7 +359,9 @@ 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 "" @@ -367,7 +369,7 @@ def _scan_gateway_pids( 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 diff --git a/tests/cli/test_cli_browser_connect.py b/tests/cli/test_cli_browser_connect.py index 2f17b0595a45..4bdc56cbcf5f 100644 --- a/tests/cli/test_cli_browser_connect.py +++ b/tests/cli/test_cli_browser_connect.py @@ -83,6 +83,17 @@ def test_linux_candidates_include_official_brave_and_edge_stable_paths(self): assert candidates == [brave, edge] + def test_wsl_install_candidates_keep_posix_separators_on_nt_host(self): + expected = "/mnt/c/Program Files/Google/Chrome/Application/chrome.exe" + + with patch("hermes_cli.browser_connect.shutil.which", return_value=None), \ + patch("hermes_cli.browser_connect.os.path.isfile", side_effect=lambda path: path == expected): + candidates = get_chrome_debug_candidates("Linux") + + assert candidates == [expected] + assert "\\" not in candidates[0] + + def test_wait_for_browser_debug_ready_or_exit_detects_early_exit(self, monkeypatch): class _Proc: def __init__(self): diff --git a/tests/cli/test_cli_file_drop.py b/tests/cli/test_cli_file_drop.py index b6093e215605..00c0fd5d5f8c 100644 --- a/tests/cli/test_cli_file_drop.py +++ b/tests/cli/test_cli_file_drop.py @@ -1,6 +1,7 @@ """Tests for _detect_file_drop — file path detection that prevents dragged/pasted absolute paths from being mistaken for slash commands.""" +import os import pytest @@ -157,6 +158,8 @@ def test_tilde_prefixed_path(self, tmp_path, monkeypatch): img.parent.mkdir(parents=True, exist_ok=True) img.write_bytes(b"\x89PNG\r\n\x1a\n") monkeypatch.setenv("HOME", str(home)) + # ntpath.expanduser ignores HOME (Python 3.8+) — it wants USERPROFILE. + monkeypatch.setenv("USERPROFILE", str(home)) result = _detect_file_drop("~/storage/shared/Pictures/cat.png what is this?") @@ -166,6 +169,19 @@ def test_tilde_prefixed_path(self, tmp_path, monkeypatch): assert result["remainder"] == "what is this?" + @pytest.mark.skipif(os.name != "nt", reason="Windows drive-letter URI contract") + def test_windows_drive_letter_file_uri_drops_url_leading_slash(self, tmp_path): + image = tmp_path / "drive-uri.png" + image.write_bytes(b"\x89PNG\r\n\x1a\n") + uri = image.as_uri() + assert uri.startswith("file:///") and ":/" in uri + + result = _detect_file_drop(uri) + + assert result is not None + assert result["path"] == image + + # --------------------------------------------------------------------------- # Tests: edge cases # --------------------------------------------------------------------------- diff --git a/tests/cli/test_cli_image_command.py b/tests/cli/test_cli_image_command.py index 0af4635dfa90..573efbe77e9c 100644 --- a/tests/cli/test_cli_image_command.py +++ b/tests/cli/test_cli_image_command.py @@ -59,6 +59,8 @@ def test_collect_query_images_supports_tilde_paths(self, tmp_path, monkeypatch): home = tmp_path / "home" img = _make_image(home / "storage" / "shared" / "Pictures" / "cat.png") monkeypatch.setenv("HOME", str(home)) + # ntpath.expanduser ignores HOME (Python 3.8+) — it wants USERPROFILE. + monkeypatch.setenv("USERPROFILE", str(home)) message, images = _collect_query_images("describe this", "~/storage/shared/Pictures/cat.png") diff --git a/tests/cli/test_worktree.py b/tests/cli/test_worktree.py index 6ca7c4514cc1..626ca29bd4a6 100644 --- a/tests/cli/test_worktree.py +++ b/tests/cli/test_worktree.py @@ -416,6 +416,21 @@ def test_ten_concurrent_worktrees(self, git_repo): assert not Path(info["path"]).exists() +def _can_symlink(): + """Check if we can create symlinks (needs admin/dev-mode on Windows).""" + import tempfile + try: + with tempfile.TemporaryDirectory() as d: + src = Path(d) / "src" + src.write_text("x") + lnk = Path(d) / "lnk" + lnk.symlink_to(src) + return True + except OSError: + return False + + +@pytest.mark.skipif(not _can_symlink(), reason="Symlinks need elevated privileges") class TestWorktreeDirectorySymlink: """Test .worktreeinclude with directories (symlinked).""" diff --git a/tests/cli/test_worktree_security.py b/tests/cli/test_worktree_security.py index bd5aae81cd20..c8c2b89f20a1 100644 --- a/tests/cli/test_worktree_security.py +++ b/tests/cli/test_worktree_security.py @@ -6,6 +6,20 @@ import pytest +def _can_symlink(): + """Check if we can create symlinks (needs admin/dev-mode on Windows).""" + import tempfile + try: + with tempfile.TemporaryDirectory() as d: + src = Path(d) / "src" + src.write_text("x") + lnk = Path(d) / "lnk" + lnk.symlink_to(src) + return True + except OSError: + return False + + @pytest.fixture def git_repo(tmp_path): """Create a temporary git repo for testing real cli._setup_worktree behavior.""" @@ -76,6 +90,7 @@ def test_rejects_parent_directory_directory_traversal(self, git_repo): finally: _force_remove_worktree(info) + @pytest.mark.skipif(not _can_symlink(), reason="Symlinks need elevated privileges") def test_rejects_symlink_that_resolves_outside_repo(self, git_repo): import cli as cli_mod @@ -110,6 +125,7 @@ def test_allows_valid_file_include(self, git_repo): finally: _force_remove_worktree(info) + @pytest.mark.skipif(not _can_symlink(), reason="Symlinks need elevated privileges") def test_allows_valid_directory_include(self, git_repo): import cli as cli_mod diff --git a/tests/gateway/test_status.py b/tests/gateway/test_status.py index 64582219c422..237331782bb8 100644 --- a/tests/gateway/test_status.py +++ b/tests/gateway/test_status.py @@ -248,6 +248,15 @@ def test_runtime_status_running_pid_accepts_matching_profile_cmdline(self, monke ), cmdline + def test_command_line_belongs_to_profile_normalizes_separators(self): + """A Windows argv renders HERMES_HOME with backslashes while the + profile's Path may carry forward slashes (and, on Windows, vice + versa). The separator difference must not defeat the match.""" + home = Path("c:/opt/data/profiles/coder") + cmdline = r"hermes_home=c:\opt\data\profiles\coder hermes gateway run --replace" + assert status._command_line_belongs_to_profile(cmdline, home) is True + + def test_write_runtime_status_explicit_none_clears_stale_fields(self, tmp_path, monkeypatch): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) diff --git a/tests/hermes_cli/test_banner.py b/tests/hermes_cli/test_banner.py index 9493d40de3d8..e1dbed119522 100644 --- a/tests/hermes_cli/test_banner.py +++ b/tests/hermes_cli/test_banner.py @@ -9,6 +9,16 @@ import tools.mcp_tool +def test_cprint_falls_back_to_plain_print_when_prompt_toolkit_has_no_console(capsys): + with patch( + "prompt_toolkit.print_formatted_text", + side_effect=RuntimeError("no console screen buffer"), + ): + banner.cprint("fallback text") + + assert capsys.readouterr().out == "fallback text\n" + + diff --git a/tests/tools/test_windows_native_support.py b/tests/tools/test_windows_native_support.py index d5e1f9357e6d..81be4319b557 100644 --- a/tests/tools/test_windows_native_support.py +++ b/tests/tools/test_windows_native_support.py @@ -52,10 +52,10 @@ def _reset_configured(self, monkeypatch): yield sys.modules.pop("hermes_cli.stdio", None) - def test_no_op_on_posix(self): + def test_no_op_on_posix(self, monkeypatch): from hermes_cli import stdio - assert stdio.is_windows() is False + monkeypatch.setattr(stdio, "is_windows", lambda: False) result = stdio.configure_windows_stdio() assert result is False From cad839713ba14cefc596a479160909c5c2bd7e3f Mon Sep 17 00:00:00 2001 From: konsisumer Date: Wed, 29 Jul 2026 21:44:13 -0700 Subject: [PATCH 2/3] docs: purge stale xdist/_enforce_test_timeout test-runner references repo-wide MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The test runner moved to per-file subprocess isolation via scripts/run_tests_parallel.py (hermetic `env -i`, worker count auto-scaled from CPU count, FLAKY-retry policy) — no pytest-xdist, no SIGALRM per-test timeout fixture. Docs still described the old runner in many places: - AGENTS.md: "-n auto xdist workers, in-tree subprocess-isolation plugin" clause replaced with the current per-file-subprocess description; the `::test_x` single-test example now shows file + -k (runner is file-granular). - CONTRIBUTING.md: "hermetic env, 4 xdist workers" comment corrected; `tests/conftest.py::_enforce_test_timeout` reference redirected to the win32 timeout-method shim in `tests/conftest.py::pytest_configure`. - skills/autonomous-ai-agents/hermes-agent/references/contributor-guide.md and windows-quirks.md: same corrections (the bundled skill mirrors the contributor docs); Windows workaround no longer installs pytest-xdist or passes -n 0. - website/docs + zh-Hans i18n mirrors: same fixes in adding-providers.md and the bundled-skill doc pages. - skills/software-development/python-debugpy/SKILL.md (+ zh-Hans mirror): "-p no:xdist"/"-n 0" pdb advice rewritten for the captured per-file subprocess runner. - skills/creative/comfyui/tests/README.md: parent-repo "-n auto by default" rationale updated to past tense. Combined salvage of PR #38295 (konsisumer), PR #51354 (TutkuEroglu, redirected to the current conftest truth and the relocated references/contributor-guide.md), and PR #54956 (waroffchange). Co-authored-by: TutkuEroglu Co-authored-by: waroffchange <116298975+waroffchange@users.noreply.github.com> --- AGENTS.md | 5 +++-- CONTRIBUTING.md | 5 +++-- .../references/contributor-guide.md | 7 ++++--- .../hermes-agent/references/windows-quirks.md | 9 +++++---- skills/creative/comfyui/tests/README.md | 12 +++++++----- .../python-debugpy/SKILL.md | 16 +++++++--------- .../docs/developer-guide/adding-providers.md | 4 ++-- .../developer-guide/adding-providers.md | 4 ++-- .../autonomous-ai-agents-hermes-agent.md | 18 +++++++++--------- .../software-development-python-debugpy.md | 16 +++++++--------- 10 files changed, 49 insertions(+), 47 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index d623ba59bbf1..70fc9bc8d647 100644 --- a/AGENTS.md +++ b/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 ``` diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 46581d820037..4fbf5b5a3da6 100644 --- a/CONTRIBUTING.md +++ b/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")` diff --git a/skills/autonomous-ai-agents/hermes-agent/references/contributor-guide.md b/skills/autonomous-ai-agents/hermes-agent/references/contributor-guide.md index 3bda90211855..a578f06d25c1 100644 --- a/skills/autonomous-ai-agents/hermes-agent/references/contributor-guide.md +++ b/skills/autonomous-ai-agents/hermes-agent/references/contributor-guide.md @@ -84,8 +84,9 @@ run_conversation(): ### Testing -Use the canonical runner — it enforces CI-parity (hermetic env, unset -credentials, TZ=UTC, xdist workers, per-test subprocess isolation): +Use the canonical runner — it enforces CI-parity (hermetic `env -i`, unset +credentials, TZ=UTC, per-file subprocess isolation via +`scripts/run_tests_parallel.py` — no xdist, worker count auto-scaled): ```bash scripts/run_tests.sh # full suite @@ -102,7 +103,7 @@ scripts/run_tests.sh -v --tb=long # pass-through pytest flags **Cross-platform test guards:** tests using POSIX-only syscalls need a skip marker. Common ones already in the codebase: - Symlink creation → `@pytest.mark.skipif(sys.platform == "win32", reason="Symlinks require elevated privileges on Windows")` (see `tests/cron/test_cron_script.py`) - POSIX file modes (0o600, etc.) → `@pytest.mark.skipif(sys.platform.startswith("win"), reason="POSIX mode bits not enforced on Windows")` (see `tests/hermes_cli/test_auth_toctou_file_modes.py`) -- `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`) - Live Winsock / Windows-specific regression tests → `@pytest.mark.skipif(sys.platform != "win32", reason="Windows-specific regression")` **Monkeypatching `sys.platform` is not enough** when the code under test also calls `platform.system()` / `platform.release()` / `platform.mac_ver()`. Those functions re-read the real OS independently, so a test that sets `sys.platform = "linux"` on a Windows runner will still see `platform.system() == "Windows"` and route through the Windows branch. Patch all three together: diff --git a/skills/autonomous-ai-agents/hermes-agent/references/windows-quirks.md b/skills/autonomous-ai-agents/hermes-agent/references/windows-quirks.md index 4cf283e95332..d87f1c0eb5b2 100644 --- a/skills/autonomous-ai-agents/hermes-agent/references/windows-quirks.md +++ b/skills/autonomous-ai-agents/hermes-agent/references/windows-quirks.md @@ -33,13 +33,14 @@ echo `os.environ` inside an `execute_code` block to confirm `SYSTEMROOT` is set. `scripts/run_tests.sh` is POSIX-only (expects `.venv/bin/activate`); the Hermes-installed `venv/Scripts/` has no pip/pytest (stripped for size). -Install pytest into a system Python and run directly with `-n 0` -(`pyproject.toml`'s `addopts` already sets `-n`): +Install pytest into a system Python and run directly (the repo no longer +uses pytest-xdist; the canonical runner does per-file subprocess isolation, +which the POSIX-only wrapper handles): ```bash -"/c/Program Files/Python311/python" -m pip install --user pytest pytest-xdist pyyaml +"/c/Program Files/Python311/python" -m pip install --user pytest pyyaml export PYTHONPATH="$(pwd)" -"/c/Program Files/Python311/python" -m pytest tests/foo/test_bar.py -v --tb=short -n 0 +"/c/Program Files/Python311/python" -m pytest tests/foo/test_bar.py -v --tb=short ``` (POSIX-only tests need skip guards — see the cross-platform guard list in diff --git a/skills/creative/comfyui/tests/README.md b/skills/creative/comfyui/tests/README.md index 833632ae9c41..d27fa97e32aa 100644 --- a/skills/creative/comfyui/tests/README.md +++ b/skills/creative/comfyui/tests/README.md @@ -43,8 +43,10 @@ When you change a script: ## Why the explicit `-c` / `-o`? -The parent hermes-agent repo's `pyproject.toml` enables `pytest-xdist` by -default (`-n auto`). This suite is small enough that parallelism isn't -worth the complexity, and pytest-xdist isn't always installed in the user's -environment. The `-c tests/pytest.ini -o addopts="-p no:xdist"` flags make -the suite run identically regardless of the parent project's config. +The parent hermes-agent repo used to enable `pytest-xdist` by default +(`-n auto`); the canonical runner has since moved to per-file subprocess +isolation via `scripts/run_tests_parallel.py` and no longer uses xdist. +This suite is small enough that parallelism isn't worth the complexity, and +pytest-xdist isn't always installed in the user's environment. The +`-c tests/pytest.ini -o addopts="-p no:xdist"` flags make the suite run +identically regardless of the parent project's config. diff --git a/skills/software-development/python-debugpy/SKILL.md b/skills/software-development/python-debugpy/SKILL.md index e57d8d91e253..9907afdac1d2 100644 --- a/skills/software-development/python-debugpy/SKILL.md +++ b/skills/software-development/python-debugpy/SKILL.md @@ -107,11 +107,9 @@ scripts/run_tests.sh tests/path/to/test_file.py::test_name --trace scripts/run_tests.sh tests/path/to/test_file.py --showlocals --tb=long ``` -Note: `scripts/run_tests.sh` uses xdist (`-n 4`) by default, and pdb does NOT work under xdist. Add `-p no:xdist` or run a single test with `-n 0`: +Note: `scripts/run_tests.sh` runs each test file in a captured subprocess via `run_tests_parallel.py` (no xdist), so interactive pdb does NOT work under the wrapper. Run pytest directly for `--pdb`: ```bash -scripts/run_tests.sh tests/foo_test.py::test_bar --pdb -p no:xdist -# or source .venv/bin/activate python -m pytest tests/foo_test.py::test_bar --pdb ``` @@ -276,7 +274,7 @@ nc 127.0.0.1 4444 ## Debugging Hermes-specific Processes ### Tests -See Recipe 3. Always add `-p no:xdist` or run single tests without xdist. +See Recipe 3. The wrapper captures subprocess output, so run pytest directly for interactive pdb. ### `run_agent.py` / CLI — one-shot Easiest: add `breakpoint()` near the suspect line, then run `hermes` normally. Control returns to your terminal at the pause point. @@ -308,7 +306,7 @@ Long-lived. Use `remote-pdb` at a handler, or `debugpy` with `--wait-for-client` ## Common Pitfalls -1. **pdb under pytest-xdist silently does nothing.** You won't see the prompt, the test just hangs. Always use `-p no:xdist` or `-n 0`. +1. **pdb under a parallel/output-capturing runner silently does nothing.** You won't see the prompt, the test just hangs (true of pytest-xdist and of `scripts/run_tests.sh`'s captured per-file subprocesses). Run pytest directly on a single file for interactive debugging. 2. **`breakpoint()` in CI / non-TTY contexts hangs the process.** Safe locally; never commit it. Add a pre-commit grep as a safety net. @@ -333,7 +331,7 @@ Long-lived. Use `remote-pdb` at a handler, or `debugpy` with `--wait-for-client` - [ ] After `pip install debugpy`, confirm: `python -c "import debugpy; print(debugpy.__version__)"` - [ ] For remote debug, confirm the port is actually listening: `ss -tlnp | grep 5678` -- [ ] First breakpoint actually hits (if it doesn't, you likely have `PYTHONBREAKPOINT=0`, you're under xdist, or execution finished before attach) +- [ ] First breakpoint actually hits (if it doesn't, you likely have `PYTHONBREAKPOINT=0`, you're under a parallel/capturing runner, or execution finished before attach) - [ ] `where` / `w` shows the expected call stack - [ ] Post-debug cleanup: no stray `breakpoint()` / `set_trace()` in committed code ```bash @@ -354,10 +352,10 @@ breakpoint() **"This test passes in isolation but fails in the suite."** ```bash -scripts/run_tests.sh tests/the_test.py --pdb -p no:xdist -# But if it only fails WITH other tests: +scripts/run_tests.sh tests/the_test.py # confirm it fails under the isolated runner first +# For interactive debugging, or if it only fails WITH other tests: source .venv/bin/activate -python -m pytest tests/ -x --pdb -p no:xdist +python -m pytest tests/ -x --pdb # Now it pdb-traps at the exact failing test after state accumulated. ``` diff --git a/website/docs/developer-guide/adding-providers.md b/website/docs/developer-guide/adding-providers.md index 0898d698ac8c..1964c194629d 100644 --- a/website/docs/developer-guide/adding-providers.md +++ b/website/docs/developer-guide/adding-providers.md @@ -338,11 +338,11 @@ For docs-only examples, the exact file set may differ. The point is to cover: - provider:model parsing - any adapter-specific message conversion -Run tests with xdist disabled: +Run the targeted tests (or use `scripts/run_tests.sh`, which runs each file in its own subprocess): ```bash source venv/bin/activate -python -m pytest tests/hermes_cli/test_runtime_provider_resolution.py tests/cli/test_cli_provider_resolution.py tests/hermes_cli/test_setup_model_provider.py tests/run_agent/test_provider_parity.py -n0 -q +python -m pytest tests/hermes_cli/test_runtime_provider_resolution.py tests/cli/test_cli_provider_resolution.py tests/hermes_cli/test_setup_model_provider.py tests/run_agent/test_provider_parity.py -q ``` For deeper changes, run the full suite before pushing: diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/adding-providers.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/adding-providers.md index 04245b32e1cb..638f47df2e70 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/adding-providers.md +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/developer-guide/adding-providers.md @@ -338,11 +338,11 @@ Prompt（提示词）缓存和 provider 专属的调节项很容易出现回归 - `provider:model` 解析 - 任何适配器专属的消息转换 -使用禁用 xdist 的方式运行测试： +运行目标测试（或使用 `scripts/run_tests.sh`，它在独立子进程中运行每个文件）： ```bash source venv/bin/activate -python -m pytest tests/test_runtime_provider_resolution.py tests/test_cli_provider_resolution.py tests/test_cli_model_command.py tests/test_setup_model_selection.py -n0 -q +python -m pytest tests/test_runtime_provider_resolution.py tests/test_cli_provider_resolution.py tests/test_cli_model_command.py tests/test_setup_model_selection.py -q ``` 对于更深层的修改，在推送前运行完整测试套件： diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/skills/bundled/autonomous-ai-agents/autonomous-ai-agents-hermes-agent.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/skills/bundled/autonomous-ai-agents/autonomous-ai-agents-hermes-agent.md index 986eb015d486..c18bb063ce2a 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/skills/bundled/autonomous-ai-agents/autonomous-ai-agents-hermes-agent.md +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/skills/bundled/autonomous-ai-agents/autonomous-ai-agents-hermes-agent.md @@ -698,20 +698,20 @@ mintty / git-bash 行为相同（Alt+Enter 全屏），除非你在选项 → ### 测试/贡献 -**`scripts/run_tests.sh` 在 Windows 上无法直接使用** — 它查找 POSIX venv 布局（`.venv/bin/activate`）。Hermes 安装的 venv 位于 `venv/Scripts/`，也没有 pip 或 pytest（为减小安装体积而精简）。解决方案：将 `pytest + pytest-xdist + pyyaml` 安装到系统 Python 3.11 用户站点，然后设置 `PYTHONPATH` 直接调用 pytest： +**`scripts/run_tests.sh` 在 Windows 上无法直接使用** — 它查找 POSIX venv 布局（`.venv/bin/activate`）。Hermes 安装的 venv 位于 `venv/Scripts/`，也没有 pip 或 pytest（为减小安装体积而精简）。解决方案：将 `pytest + pyyaml` 安装到系统 Python 3.11 用户站点，然后设置 `PYTHONPATH` 直接调用 pytest： ```bash -"/c/Program Files/Python311/python" -m pip install --user pytest pytest-xdist pyyaml +"/c/Program Files/Python311/python" -m pip install --user pytest pyyaml export PYTHONPATH="$(pwd)" -"/c/Program Files/Python311/python" -m pytest tests/foo/test_bar.py -v --tb=short -n 0 +"/c/Program Files/Python311/python" -m pytest tests/foo/test_bar.py -v --tb=short ``` -使用 `-n 0` 而非 `-n 4` — `pyproject.toml` 的默认 `addopts` 已包含 `-n`，且 wrapper 的 CI 一致性保证不适用于非 POSIX 环境。 +仓库已不再使用 pytest-xdist——规范 runner 通过 `run_tests_parallel.py` 做按文件子进程隔离，但该 wrapper 仅支持 POSIX，其 CI 一致性保证不适用于非 POSIX 环境。 **仅 POSIX 的测试需要跳过守卫。** 代码库中已有的常见标记： - 符号链接——Windows 上需要提升权限 - `0o600` 文件模式——POSIX 模式位在 NTFS 上默认不强制执行 -- `signal.SIGALRM`——仅 Unix（参见 `tests/conftest.py::_enforce_test_timeout`） +- `signal.SIGALRM`——仅 Unix（每测试超时不再直接使用它；参见 `tests/conftest.py::pytest_configure` 中的 win32 timeout-method shim） - Winsock / Windows 特有回归——`@pytest.mark.skipif(sys.platform != "win32", ...)` 使用现有的跳过模式风格（`sys.platform == "win32"` 或 `sys.platform.startswith("win")`）以与测试套件其余部分保持一致。 @@ -891,19 +891,19 @@ python -m pytest tests/tools/ -q # 特定区域 - 推送任何变更前运行完整套件 - 使用 `-o 'addopts='` 清除任何内置的 pytest 标志 -**Windows 贡献者：** `scripts/run_tests.sh` 目前查找 POSIX venv（`.venv/bin/activate` / `venv/bin/activate`），在 Windows 上会报错，因为布局是 `venv/Scripts/activate` + `python.exe`。Hermes 安装的 venv 位于 `venv/Scripts/`，也没有 `pip` 或 `pytest`——为终端用户安装体积而精简。解决方案：将 pytest + pytest-xdist + pyyaml 安装到系统 Python 3.11 用户站点（`/c/Program Files/Python311/python -m pip install --user pytest pytest-xdist pyyaml`），然后直接运行测试： +**Windows 贡献者：** `scripts/run_tests.sh` 目前查找 POSIX venv（`.venv/bin/activate` / `venv/bin/activate`），在 Windows 上会报错，因为布局是 `venv/Scripts/activate` + `python.exe`。Hermes 安装的 venv 位于 `venv/Scripts/`，也没有 `pip` 或 `pytest`——为终端用户安装体积而精简。解决方案：将 pytest + pyyaml 安装到系统 Python 3.11 用户站点（`/c/Program Files/Python311/python -m pip install --user pytest pyyaml`），然后直接运行测试： ```bash export PYTHONPATH="$(pwd)" -"/c/Program Files/Python311/python" -m pytest tests/tools/test_foo.py -v --tb=short -n 0 +"/c/Program Files/Python311/python" -m pytest tests/tools/test_foo.py -v --tb=short ``` -使用 `-n 0`（而非 `-n 4`），因为 `pyproject.toml` 的默认 `addopts` 已包含 `-n`，且 wrapper 的 CI 一致性保证不适用于非 POSIX 环境。 +仓库已不再使用 pytest-xdist——规范 runner 通过 `run_tests_parallel.py` 做按文件子进程隔离，但该 wrapper 仅支持 POSIX，其 CI 一致性保证不适用于非 POSIX 环境。 **跨平台测试守卫：** 使用仅 POSIX 系统调用的测试需要跳过标记。代码库中已有的常见标记： - 符号链接创建 → `@pytest.mark.skipif(sys.platform == "win32", reason="Symlinks require elevated privileges on Windows")`（参见 `tests/cron/test_cron_script.py`） - POSIX 文件模式（0o600 等）→ `@pytest.mark.skipif(sys.platform.startswith("win"), reason="POSIX mode bits not enforced on Windows")`（参见 `tests/hermes_cli/test_auth_toctou_file_modes.py`） -- `signal.SIGALRM` → 仅 Unix（参见 `tests/conftest.py::_enforce_test_timeout`） +- `signal.SIGALRM` → 仅 Unix（每测试超时不再直接使用它；参见 `tests/conftest.py::pytest_configure` 中的 win32 timeout-method shim） - 实时 Winsock / Windows 特有回归测试 → `@pytest.mark.skipif(sys.platform != "win32", reason="Windows-specific regression")` **仅 monkeypatch `sys.platform` 是不够的**，当被测代码还调用 `platform.system()` / `platform.release()` / `platform.mac_ver()` 时。这些函数独立重新读取真实 OS，因此在 Windows runner 上将 `sys.platform = "linux"` 的测试仍会看到 `platform.system() == "Windows"` 并走 Windows 分支。需要同时 patch 三者： diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/skills/bundled/software-development/software-development-python-debugpy.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/skills/bundled/software-development/software-development-python-debugpy.md index a8276c5678fb..e3ea93f47b45 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/skills/bundled/software-development/software-development-python-debugpy.md +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/skills/bundled/software-development/software-development-python-debugpy.md @@ -125,11 +125,9 @@ scripts/run_tests.sh tests/path/to/test_file.py::test_name --trace scripts/run_tests.sh tests/path/to/test_file.py --showlocals --tb=long ``` -注意：`scripts/run_tests.sh` 默认使用 xdist（`-n 4`），pdb 在 xdist 下**无法正常工作**。请添加 `-p no:xdist` 或使用 `-n 0` 运行单个测试： +注意：`scripts/run_tests.sh` 通过 `run_tests_parallel.py` 将每个测试文件放在捕获输出的子进程中运行（不使用 xdist），因此交互式 pdb 在 wrapper 下**无法正常工作**。请直接运行 pytest 使用 `--pdb`： ```bash -scripts/run_tests.sh tests/foo_test.py::test_bar --pdb -p no:xdist -# 或 source .venv/bin/activate python -m pytest tests/foo_test.py::test_bar --pdb ``` @@ -294,7 +292,7 @@ nc 127.0.0.1 4444 ## 调试 Hermes 特定进程 ### 测试 -参见方案 3。始终添加 `-p no:xdist` 或在不使用 xdist 的情况下运行单个测试。 +参见方案 3。wrapper 会捕获子进程输出，交互式 pdb 请直接运行 pytest。 ### `run_agent.py` / CLI — 一次性运行 最简单：在可疑行附近添加 `breakpoint()`，然后正常运行 `hermes`。控制权将在暂停点返回到你的终端。 @@ -326,7 +324,7 @@ set_trace(host="127.0.0.1", port=4444) # 在你想捕获的 RPC 处理器中 ## 常见陷阱 -1. **pdb 在 pytest-xdist 下静默失效。** 你不会看到提示符，测试只会挂起。始终使用 `-p no:xdist` 或 `-n 0`。 +1. **pdb 在并行/捕获输出的 runner 下静默失效。** 你不会看到提示符，测试只会挂起（pytest-xdist 与 `scripts/run_tests.sh` 的按文件捕获子进程均如此）。交互式调试请直接对单个文件运行 pytest。 2. **`breakpoint()` 在 CI / 非 TTY 环境中会挂起进程。** 本地使用没问题；永远不要提交它。添加 pre-commit grep 作为安全网。 @@ -351,7 +349,7 @@ set_trace(host="127.0.0.1", port=4444) # 在你想捕获的 RPC 处理器中 - [ ] `pip install debugpy` 后确认：`python -c "import debugpy; print(debugpy.__version__)"` - [ ] 对于远程调试，确认端口确实在监听：`ss -tlnp | grep 5678` -- [ ] 第一个断点确实触发（如果没有，可能是 `PYTHONBREAKPOINT=0`、在 xdist 下运行，或执行在附加前已结束） +- [ ] 第一个断点确实触发（如果没有，可能是 `PYTHONBREAKPOINT=0`、在并行/捕获输出的 runner 下运行，或执行在附加前已结束） - [ ] `where` / `w` 显示预期的调用栈 - [ ] 调试后清理：已提交代码中无残留的 `breakpoint()` / `set_trace()` / `debugpy.listen` ```bash @@ -372,10 +370,10 @@ breakpoint() **"这个测试单独运行通过，但在测试套件中失败。"** ```bash -scripts/run_tests.sh tests/the_test.py --pdb -p no:xdist -# 但如果只有与其他测试一起运行才失败： +scripts/run_tests.sh tests/the_test.py # 先确认它在隔离 runner 下失败 +# 交互式调试，或只有与其他测试一起运行才失败时： source .venv/bin/activate -python -m pytest tests/ -x --pdb -p no:xdist +python -m pytest tests/ -x --pdb # 现在它会在状态积累后的确切失败测试处触发 pdb。 ``` From f874ff99093f9347d780c8a005ff08b5378b6515 Mon Sep 17 00:00:00 2001 From: SSC-ENG <225143396+SSC-ENG@users.noreply.github.com> Date: Wed, 29 Jul 2026 23:23:19 -0700 Subject: [PATCH 3/3] feat(kanban): govern raw intake decomposition --- hermes_cli/config_defaults.py | 5 + hermes_cli/kanban.py | 78 ++-- hermes_cli/kanban_db.py | 130 ++++++ hermes_cli/kanban_decompose.py | 389 +++++++++++------- hermes_cli/kanban_intake.py | 211 ++++++++-- tests/hermes_cli/test_kanban_decompose.py | 12 +- .../hermes_cli/test_kanban_governed_intake.py | 92 +++++ 7 files changed, 708 insertions(+), 209 deletions(-) create mode 100644 tests/hermes_cli/test_kanban_governed_intake.py diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index dd1cfe88c013..34549242bca5 100644 --- a/hermes_cli/config_defaults.py +++ b/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 diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index a57728db6d48..2471a58541b0 100644 --- a/hermes_cli/kanban.py +++ b/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 diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 74bc9d3ca3f6..c26b95840e19 100644 --- a/hermes_cli/kanban_db.py +++ b/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[A-Z][A-Z0-9]+-\d+)\s+" + r"subissues=\[(?P.+)\]", + re.DOTALL, +) +_LINEAR_SUBISSUE_RE = re.compile( + r"(?:key\s*[:=]\s*)?(?P[A-Z][A-Z0-9]+-\d+)\s*" + r"[,;]\s*cptc\s*[:=]\s*(?P1|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( diff --git a/hermes_cli/kanban_decompose.py b/hermes_cli/kanban_decompose.py index ce634073211a..ea4a321c080c 100644 --- a/hermes_cli/kanban_decompose.py +++ b/hermes_cli/kanban_decompose.py @@ -29,9 +29,9 @@ no children created. This makes ``decompose`` a strict superset of ``specify`` from the user's perspective. -* If the LLM picks an assignee that doesn't exist as a profile, we - rewrite it to the configured ``default_assignee`` (or the default - profile if unset). A child task NEVER ends up with ``assignee=None``. +* LLM output is advisory. Tenant, domain, certification, profile, + assignee, graph, fan-out, and PPMA-gate invariants are validated + deterministically before any DB mutation. """ from __future__ import annotations @@ -44,10 +44,28 @@ from typing import Optional from hermes_cli import kanban_db as kb +from hermes_cli import kanban_intake from hermes_cli import profiles as profiles_mod logger = logging.getLogger(__name__) +_PPMA_PROFILE = "paul-park" +_DEFAULT_FANOUT_CAP = 6 +_DEFAULT_DOMAINS = frozenset({ + "program-management", + "engineering", + "security", + "finance", + "marketing", + "operations", + "information-technology", + "customer-service", + "product", + "legal", + "procurement", + "people", +}) + _SYSTEM_PROMPT = """You are the Kanban decomposer for the Hermes Agent board. @@ -70,7 +88,8 @@ "title": "", "body": "", "assignee": "", - "domain": "", + "domain": "", + "required_certification": "", "parents": [, ...] }, ... @@ -78,20 +97,20 @@ } Rules: - - For fanout=true, task index 0 is ALWAYS the PPMA scoping gate. Assign it - to paul-park, give it no parents, and make every execution task depend on - index 0. Its body must require Linear parent + technical sub-issue CPTC - scoping before downstream execution. + - For fanout=true, index 0 is the PPMA scoping gate: assign paul-park, + domain program-management, no parents, certification helios-agent-ppma. + - Every execution task depends directly on index 0. PPMA records the Linear + parent and technical CPTC sub-issues before those tasks become eligible. - "parents" is a list of INDICES (0-based) into this same "tasks" list, expressing actual data dependencies. Tasks with no parents run in PARALLEL. Tasks with parents wait until every parent completes. - - Prefer parallelism. If two tasks can be done independently, give - them no parents so the dispatcher fans them out at once. + - Prefer parallelism after the shared PPMA gate. Independent execution tasks + should list only index 0 as a parent. - Use 2-6 tasks for normal work. Don't create 20 tiny tasks. Don't cram everything into 1 task. - Pick assignees from the roster by matching the task to the profile's - DESCRIPTION (not just the name). When nothing matches well, use null - and the system will route to the default_assignee. + DESCRIPTION (not just the name). When assignment is ambiguous, use null; + deterministic validation routes the item to PPMA with the reason recorded. - Each child task body is what a fresh worker will read with no other context — be specific about goal, approach, and acceptance criteria. @@ -122,7 +141,7 @@ Available profiles (assignees you may pick from): {roster} -Default assignee (used when no profile fits a task): {default_assignee} +Ambiguous-assignment owner: {default_assignee} """ @@ -185,8 +204,7 @@ def _load_config() -> dict: def _resolve_orchestrator_profile(cfg: dict) -> str: """Resolve which profile owns the root/orchestration task after fan-out. - Falls back to the active default profile when ``kanban.orchestrator_profile`` - is unset, so a task is never stranded for lack of an orchestrator. + Ambiguous orchestration is a PPMA concern, never a launch-profile concern. """ kanban_cfg = cfg.get("kanban", {}) if isinstance(cfg, dict) else {} explicit = (kanban_cfg.get("orchestrator_profile") or "").strip() @@ -196,15 +214,11 @@ def _resolve_orchestrator_profile(cfg: dict) -> str: return explicit except Exception: pass - # Fall back to the active default profile. - try: - return profiles_mod.get_active_profile_name() or "default" - except Exception: - return "default" + return _PPMA_PROFILE def _resolve_default_assignee(cfg: dict) -> str: - """Resolve which profile catches child tasks the orchestrator can't route.""" + """Resolve the PPMA owner for ambiguous child assignment.""" kanban_cfg = cfg.get("kanban", {}) if isinstance(cfg, dict) else {} explicit = (kanban_cfg.get("default_assignee") or "").strip() if explicit: @@ -213,26 +227,35 @@ def _resolve_default_assignee(cfg: dict) -> str: return explicit except Exception: pass - try: - return profiles_mod.get_active_profile_name() or "default" - except Exception: - return "default" + return _PPMA_PROFILE -def _build_roster() -> tuple[list[dict], set[str]]: - """Return (roster_for_prompt, valid_assignee_names). +def _profile_certifications(profile_path) -> set[str]: + """Return binding skill-directory names installed for one profile.""" + if profile_path is None: + return set() + skills_root = profile_path / "skills" + if not skills_root.is_dir(): + return set() + certifications: set[str] = set() + for skill_md in skills_root.rglob("SKILL.md"): + certifications.add(skill_md.parent.name) + return certifications + + +def _build_roster() -> tuple[list[dict], dict[str, set[str]]]: + """Return (roster_for_prompt, installed certifications by profile). Each roster entry is ``{name, description, has_description}``. The - valid-set is used after the LLM responds to rewrite invalid - assignees to the default fallback. + The certification map is used by deterministic post-LLM validation. """ roster: list[dict] = [] - valid: set[str] = set() + certifications: dict[str, set[str]] = {} try: all_profiles = profiles_mod.list_profiles() except Exception as exc: logger.warning("decompose: failed to list profiles: %s", exc) - return roster, valid + return roster, certifications for p in all_profiles: desc = (p.description or "").strip() roster.append({ @@ -240,8 +263,13 @@ def _build_roster() -> tuple[list[dict], set[str]]: "description": desc or f"(no description; profile named {p.name!r})", "has_description": bool(desc), }) - valid.add(p.name) - return roster, valid + certifications[p.name] = _profile_certifications(getattr(p, "path", None)) + if p.name == _PPMA_PROFILE: + # The PPMA profile's identity is the authoritative certification + # holder. This also keeps isolated test/profile fixtures honest + # when they model the profile without copying its skill tree. + certifications[p.name].add("helios-agent-ppma") + return roster, certifications def _format_roster(roster: list[dict]) -> str: @@ -273,6 +301,147 @@ def _normalize_assignee_choice( return chosen +def _allowed_domains(cfg: dict, envelope: Optional[kanban_intake.IntakeEnvelope]) -> set[str]: + kanban_cfg = cfg.get("kanban", {}) if isinstance(cfg, dict) else {} + configured = kanban_cfg.get("intake_allowed_domains") or [] + allowed = { + str(value).strip() + for value in configured + if isinstance(value, str) and value.strip() + } + allowed = allowed or set(_DEFAULT_DOMAINS) + if envelope: + allowed.add(envelope.tenant_domain) + allowed.add("program-management") + return allowed + + +def _validate_children( + raw_tasks: list, + *, + cfg: dict, + task: kb.Task, + envelope: Optional[kanban_intake.IntakeEnvelope], + certifications: dict[str, set[str]], +) -> tuple[list[dict], list[dict]]: + """Validate and normalize all LLM graph output before any durable write.""" + kanban_cfg = cfg.get("kanban", {}) if isinstance(cfg, dict) else {} + fanout_cap = int(kanban_cfg.get("intake_fanout_cap") or _DEFAULT_FANOUT_CAP) + if len(raw_tasks) < 2: + raise ValueError("fanout graph must contain PPMA task 0 plus an execution task") + if len(raw_tasks) > fanout_cap: + raise ValueError(f"fanout graph exceeds configured cap of {fanout_cap}") + if _PPMA_PROFILE not in certifications: + raise ValueError("required PPMA profile 'paul-park' is not installed") + if envelope and task.tenant and envelope.tenant_domain != task.tenant: + raise ValueError( + "intake envelope tenant_domain does not match the task tenant" + ) + + allowed_domains = _allowed_domains(cfg, envelope) + allowed_assignees = set(certifications) + configured_assignees = kanban_cfg.get("intake_allowed_assignees") or [] + if configured_assignees: + allowed_assignees &= { + str(value).strip() + for value in configured_assignees + if isinstance(value, str) and value.strip() + } + allowed_assignees.add(_PPMA_PROFILE) + + children: list[dict] = [] + decisions: list[dict] = [] + for idx, entry in enumerate(raw_tasks): + if not isinstance(entry, dict): + raise ValueError(f"tasks[{idx}] is not an object") + title = entry.get("title") + if not isinstance(title, str) or not title.strip(): + raise ValueError(f"tasks[{idx}].title is missing or empty") + body = entry.get("body") if isinstance(entry.get("body"), str) else "" + parents = entry.get("parents") or [] + if not isinstance(parents, list): + raise ValueError(f"tasks[{idx}].parents must be a list") + if any(not isinstance(parent, int) for parent in parents): + raise ValueError(f"tasks[{idx}].parents contains a non-integer index") + if any(parent < 0 or parent >= len(raw_tasks) or parent == idx for parent in parents): + raise ValueError(f"tasks[{idx}].parents contains an invalid index") + if len(set(parents)) != len(parents): + raise ValueError(f"tasks[{idx}].parents contains duplicate indices") + + requested = entry.get("assignee") + requested_name = requested.strip() if isinstance(requested, str) else None + domain = str(entry.get("domain") or "").strip() + required_certification = str(entry.get("required_certification") or "").strip() or None + reason = "validated" + + if idx == 0: + requested_name = _PPMA_PROFILE + domain = "program-management" + required_certification = "helios-agent-ppma" + parents = [] + reason = "ppma_gate_enforced" + gate_text = ( + "Record scope before execution with: LINEAR_SCOPE: parent= " + "subissues=[{key:, cptc:}]. Then complete this gate so " + "the dependent domain tasks can become eligible." + ) + body = f"{body.strip()}\n\n{gate_text}".strip() + elif 0 not in parents: + raise ValueError(f"tasks[{idx}] is not directly gated by PPMA task 0") + + if domain not in allowed_domains: + raise ValueError(f"tasks[{idx}].domain {domain!r} is not allowed") + + resolved = requested_name + if not resolved or resolved not in allowed_assignees: + resolved = _PPMA_PROFILE + reason = "ambiguous_or_disallowed_assignee" + if required_certification: + holder_skills = certifications.get(resolved, set()) + if required_certification not in holder_skills: + resolved = _PPMA_PROFILE + reason = "required_certification_unverified" + if resolved not in certifications: + raise ValueError(f"tasks[{idx}] resolved to missing profile {resolved!r}") + + children.append({ + "title": title.strip()[:200], + "body": body.strip(), + "assignee": resolved, + "parents": parents, + "domain": domain, + "required_certification": required_certification, + }) + decisions.append({ + "index": idx, + "requested_assignee": requested, + "resolved_assignee": resolved, + "required_certification": required_certification, + "reason": reason, + }) + + # DB also checks cycles atomically. Detect here so the complete post-LLM + # validation result is known before attempting the durable decomposition. + indegree = [0] * len(children) + adjacent: list[list[int]] = [[] for _ in children] + for child_idx, child in enumerate(children): + for parent_idx in child["parents"]: + adjacent[parent_idx].append(child_idx) + indegree[child_idx] += 1 + queue = [idx for idx, degree in enumerate(indegree) if degree == 0] + visited = 0 + while queue: + node = queue.pop() + visited += 1 + for neighbor in adjacent[node]: + indegree[neighbor] -= 1 + if indegree[neighbor] == 0: + queue.append(neighbor) + if visited != len(children): + raise ValueError("fanout graph contains a dependency cycle") + return children, decisions + + def decompose_task( task_id: str, *, @@ -300,7 +469,9 @@ def decompose_task( default_assignee = _resolve_default_assignee(cfg) kanban_cfg = cfg.get("kanban", {}) if isinstance(cfg, dict) else {} auto_promote = bool(kanban_cfg.get("auto_promote_children", True)) - roster, valid_names = _build_roster() + roster, certifications = _build_roster() + valid_names = set(certifications) + envelope = kanban_intake.parse_envelope(task.body) try: from agent.auxiliary_client import call_llm # type: ignore @@ -379,37 +550,6 @@ def decompose_task( return DecomposeOutcome( task_id, False, "task moved out of triage before promotion", ) - with kb.connect_closing() as conn: - resolved_assignee = assignee_val or task.assignee - kb._append_event( - conn, - task_id, - "intake_classified", - {"schema_version": 1, "actor": audit_author, "source": "decomposer", - "correlation_id": task_id, "intake_id": task_id, - "domain": "UNKNOWN", "classification": "single_task", - "confidence": None, "rationale_code": "decomposer_result", - "required_gates": []}, - ) - kb._append_event( - conn, - task_id, - "routing_decided", - {"schema_version": 1, "actor": audit_author, "source": "decomposer", - "correlation_id": task_id, "intake_id": task_id, - "task_id": task_id, "requested_assignee": parsed.get("assignee"), - "resolved_assignee": resolved_assignee, - "fallback_used": resolved_assignee != parsed.get("assignee"), - "profile_exists": resolved_assignee in valid_names, - "required_certification": None, "holder_verified": None}, - ) - kb._append_event( - conn, - task_id, - "decomposition_decided", - {"schema_version": 1, "fanout": False, "root_task_id": task_id, - "child_ids": [], "dependency_edges": [], "rationale_code": "single_task"}, - ) return DecomposeOutcome( task_id, True, "single task (no fanout)", fanout=False, new_title=title_val, @@ -421,63 +561,16 @@ def decompose_task( task_id, False, "decomposer returned fanout=true with empty tasks list", ) - # Rewrite invalid assignees to the default fallback. Never leave a - # task with assignee=None — the user explicitly does not want that. - children: list[dict] = [] - for idx, entry in enumerate(raw_tasks): - if not isinstance(entry, dict): - return DecomposeOutcome( - task_id, False, f"tasks[{idx}] is not an object", - ) - title = entry.get("title") - if not isinstance(title, str) or not title.strip(): - return DecomposeOutcome( - task_id, False, f"tasks[{idx}].title is missing or empty", - ) - body = entry.get("body") - if not isinstance(body, str): - body = "" - assignee = entry.get("assignee") - chosen = _normalize_assignee_choice( - assignee, - default_assignee=default_assignee, - valid_names=valid_names, - ) - if ( - isinstance(assignee, str) - and assignee.strip() - and assignee.strip() not in valid_names - ): - logger.info( - "decompose: task %s child %d picked unknown assignee %r — " - "routing to default_assignee %r", - task_id, idx, assignee, default_assignee, - ) - parents = entry.get("parents") or [] - if not isinstance(parents, list): - parents = [] - # Clean parent indices: drop non-int and out-of-range. - clean_parents = [p for p in parents if isinstance(p, int) and 0 <= p < len(raw_tasks) and p != idx] - children.append({ - "title": title.strip()[:200], - "body": body.strip(), - "assignee": chosen, - "parents": clean_parents, - "domain": str(entry.get("domain") or "UNKNOWN").strip() or "UNKNOWN", - }) - - if fanout: - children[0]["assignee"] = "paul-park" - children[0]["domain"] = "program-management" - gate_text = ( - "Scope this intake into Linear before execution: create or verify a " - "plain-language parent and type:technical sub-issue(s) with CPTC. " - "Only after scoping, dispatch the dependent execution tasks." + try: + children, routing_decisions = _validate_children( + raw_tasks, + cfg=cfg, + task=task, + envelope=envelope, + certifications=certifications, ) - children[0]["body"] = f"{children[0]['body']}\n\n{gate_text}".strip() - for idx in range(1, len(children)): - if 0 not in children[idx]["parents"]: - children[idx]["parents"].append(0) + except ValueError as exc: + return DecomposeOutcome(task_id, False, f"post-LLM validation failed: {exc}") try: with kb.connect_closing() as conn: @@ -501,40 +594,32 @@ def decompose_task( ) with kb.connect_closing() as conn: - kb._append_event( - conn, - task_id, - "intake_classified", - {"schema_version": 1, "actor": audit_author, "source": "decomposer", - "correlation_id": task_id, "intake_id": task_id, - "domain": children[0].get("domain") or "UNKNOWN", "classification": "fanout", - "confidence": None, "rationale_code": "decomposer_result", - "required_gates": []}, - ) - for child_id, child in zip(child_ids, children): + with kb.write_txn(conn): kb._append_event( conn, - child_id, - "routing_decided", - {"schema_version": 1, "actor": audit_author, "source": "decomposer", - "correlation_id": task_id, "intake_id": task_id, - "task_id": child_id, "requested_assignee": child.get("assignee"), - "resolved_assignee": child.get("assignee"), "fallback_used": False, - "profile_exists": True, "required_certification": None, - "holder_verified": None}, + task_id, + "scope_recorded", + { + "schema_version": 1, + "scope_source": "intake_envelope" if envelope else "decomposer", + "content_digest": envelope.content_digest if envelope else None, + "tenant_domain": envelope.tenant_domain if envelope else task.tenant, + "sensitivity": envelope.sensitivity if envelope else None, + "validated_child_count": len(child_ids), + }, ) - kb._append_event( - conn, - task_id, - "decomposition_decided", - {"schema_version": 1, "fanout": True, "root_task_id": task_id, - "child_ids": child_ids, - "dependency_edges": [ - {"parent_index": p, "child_index": i} - for i, child in enumerate(children) for p in child.get("parents", []) - ], - "rationale_code": "fanout"}, - ) + for child_id, decision in zip(child_ids, routing_decisions): + kb._append_event( + conn, + child_id, + "handoff_emitted", + { + "schema_version": 1, + "handoff_kind": "domain_execution", + "routing": decision, + "parent_gate_index": 0, + }, + ) return DecomposeOutcome( task_id, True, f"decomposed into {len(child_ids)} children", diff --git a/hermes_cli/kanban_intake.py b/hermes_cli/kanban_intake.py index 406633cfb024..a7ddea38dd45 100644 --- a/hermes_cli/kanban_intake.py +++ b/hermes_cli/kanban_intake.py @@ -1,12 +1,10 @@ -"""Governed raw-work intake for the Kanban triage column. - -This module deliberately normalizes input only. Classification and routing remain -owned by ``kanban_decompose`` and the gateway's existing auto-decomposer. -""" +"""Canonical raw-work envelope for the existing Kanban triage/decomposer path.""" from __future__ import annotations import hashlib +import json import re +from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Iterable, Optional @@ -14,17 +12,160 @@ from hermes_cli import kanban_db as kb +_PREFIX = "INTAKE-ENVELOPE v1\n" +_SUFFIX = "\nEND-INTAKE-ENVELOPE" +_SENSITIVITIES = frozenset({"public", "internal", "confidential", "restricted"}) _LINEAR_URL_RE = re.compile(r"https?://(?:www\.)?linear\.app/[^\s]+/issue/[A-Za-z][A-Za-z0-9_-]*-\d+(?:/[^\s]*)?", re.I) _URL_RE = re.compile(r"https?://[^\s]+", re.I) +@dataclass(frozen=True) +class IntakeEnvelope: + source: str + items: tuple[str, ...] + notes: str + attachment_refs: tuple[str, ...] + content_digest: str + tenant_domain: str + sensitivity: str + idempotency_key: str + + def to_dict(self) -> dict: + return { + "source": self.source, + "items": list(self.items), + "notes": self.notes, + "attachment_refs": list(self.attachment_refs), + "content_digest": self.content_digest, + "tenant_domain": self.tenant_domain, + "sensitivity": self.sensitivity, + "idempotency_key": self.idempotency_key, + } + + +def _clean_list(values: Iterable[str], *, field: str, required: bool = False) -> tuple[str, ...]: + cleaned = tuple(str(value).strip() for value in values if str(value).strip()) + if required and not cleaned: + raise ValueError(f"intake envelope {field} must contain at least one value") + return cleaned + + +def _digest_payload( + *, + source: str, + items: tuple[str, ...], + notes: str, + attachment_refs: tuple[str, ...], + tenant_domain: str, + sensitivity: str, +) -> str: + payload = { + "attachment_refs": list(attachment_refs), + "items": list(items), + "notes": notes, + "sensitivity": sensitivity, + "source": source, + "tenant_domain": tenant_domain, + } + encoded = json.dumps(payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def build_envelope( + *, + source: str, + items: Iterable[str], + notes: str = "", + attachment_refs: Iterable[str] = (), + tenant_domain: str, + sensitivity: str = "internal", + idempotency_key: Optional[str] = None, +) -> IntakeEnvelope: + source = str(source).strip() + tenant_domain = str(tenant_domain).strip() + sensitivity = str(sensitivity).strip().lower() + if not source: + raise ValueError("intake envelope source is required") + if not tenant_domain: + raise ValueError("intake envelope tenant_domain is required") + if sensitivity not in _SENSITIVITIES: + raise ValueError( + f"intake envelope sensitivity must be one of {sorted(_SENSITIVITIES)}" + ) + clean_items = _clean_list(items, field="items", required=True) + clean_refs = _clean_list(attachment_refs, field="attachment_refs") + clean_notes = str(notes or "").strip() + digest = _digest_payload( + source=source, + items=clean_items, + notes=clean_notes, + attachment_refs=clean_refs, + tenant_domain=tenant_domain, + sensitivity=sensitivity, + ) + key = str(idempotency_key or digest).strip() + if not key: + raise ValueError("intake envelope idempotency_key is required") + return IntakeEnvelope( + source=source, + items=clean_items, + notes=clean_notes, + attachment_refs=clean_refs, + content_digest=digest, + tenant_domain=tenant_domain, + sensitivity=sensitivity, + idempotency_key=key, + ) + + +def render_envelope(envelope: IntakeEnvelope) -> str: + payload = json.dumps(envelope.to_dict(), ensure_ascii=False, indent=2, sort_keys=True) + return f"{_PREFIX}{payload}{_SUFFIX}" + + +def parse_envelope(body: Optional[str]) -> Optional[IntakeEnvelope]: + text = body or "" + start = text.find(_PREFIX) + if start < 0: + return None + payload_start = start + len(_PREFIX) + end = text.find(_SUFFIX, payload_start) + if end < 0: + raise ValueError("intake envelope is missing END-INTAKE-ENVELOPE") + try: + raw = json.loads(text[payload_start:end]) + except json.JSONDecodeError as exc: + raise ValueError(f"intake envelope JSON is invalid: {exc.msg}") from exc + if not isinstance(raw, dict): + raise ValueError("intake envelope payload must be an object") + supplied_digest = str(raw.get("content_digest") or "").strip() + supplied_key = str(raw.get("idempotency_key") or "").strip() + raw_items = raw.get("items") + raw_attachment_refs = raw.get("attachment_refs") + envelope = build_envelope( + source=raw.get("source") or "", + items=raw_items if isinstance(raw_items, list) else (), + notes=raw.get("notes") or "", + attachment_refs=( + raw_attachment_refs if isinstance(raw_attachment_refs, list) else () + ), + tenant_domain=raw.get("tenant_domain") or "", + sensitivity=raw.get("sensitivity") or "", + idempotency_key=supplied_key, + ) + if supplied_digest != envelope.content_digest: + raise ValueError("intake envelope content_digest does not match canonical content") + if not supplied_key: + raise ValueError("intake envelope idempotency_key is required") + return envelope + + def _normalize_url(value: str) -> str: parsed = urlsplit(value.strip()) return urlunsplit((parsed.scheme.lower(), parsed.netloc.lower(), parsed.path, parsed.query, parsed.fragment)) def normalize_raw_ref(text: str) -> str: - """Normalize raw text for stable idempotency without changing stored input.""" lines = [line.strip() for line in (text or "").splitlines()] return "\n".join(lines).strip() @@ -69,7 +210,7 @@ def idempotency_key(kind: str, text: str, files: Iterable[Path]) -> str: return hashlib.sha256(f"{kind}\n{ref}".encode("utf-8")).hexdigest() -def build_envelope(*, kind: str, raw_ref_sha256: str, received_by: str, text: str, attachment_ids: list[int]) -> str: +def _legacy_envelope(*, kind: str, raw_ref_sha256: str, received_by: str, text: str, attachment_ids: list[int]) -> str: received_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") header = "\n".join([ "---", @@ -80,7 +221,7 @@ def build_envelope(*, kind: str, raw_ref_sha256: str, received_by: str, text: st f"intake_attachment_ids: {attachment_ids}", "---", ]) - return f"{header}\n{ text }" if text else f"{header}\n" + return f"{header}\n{text}" if text else f"{header}\n" def receive( @@ -94,7 +235,8 @@ def receive( tenant: Optional[str] = None, board: Optional[str] = None, ) -> tuple[str, bool]: - paths = [Path(p).expanduser() for p in files] + """Backward-compatible CLI intake entry point.""" + paths = [Path(path).expanduser() for path in files] for path in paths: if not path.is_file(): raise ValueError(f"intake file does not exist: {path}") @@ -103,31 +245,44 @@ def receive( kind = source_type(text, paths) digest = raw_hash(text, paths) key = idempotency_key(kind, text, paths) - title = title or (f"Raw intake: {kind}") - existing_before = conn.execute( - "SELECT id FROM tasks WHERE idempotency_key = ? AND status != 'archived' LIMIT 1", (key,) + title = title or f"Raw intake: {kind}" + existing = conn.execute( + "SELECT id FROM tasks WHERE idempotency_key = ? AND status != 'archived' LIMIT 1", + (key,), ).fetchone() task_id = kb.create_task( - conn, title=title, body=build_envelope( - kind=kind, raw_ref_sha256=digest, received_by=received_by, - text=text, attachment_ids=[], - ), created_by=received_by, priority=priority, tenant=tenant, - triage=True, idempotency_key=key, + conn, + title=title, + body=_legacy_envelope(kind=kind, raw_ref_sha256=digest, received_by=received_by, text=text, attachment_ids=[]), + created_by=received_by, + priority=priority, + tenant=tenant, + triage=True, + idempotency_key=key, ) - created = existing_before is None - attachment_ids: list[int] = [] - for path in paths: - attachment_ids.append(kb.store_attachment_bytes( - conn, task_id, path.name, path.read_bytes(), - content_type=None, uploaded_by=received_by, board=board, - )) + created = existing is None + attachment_ids = [ + kb.store_attachment_bytes( + conn, + task_id, + path.name, + path.read_bytes(), + content_type=None, + uploaded_by=received_by, + board=board, + ) + for path in paths + ] if created: - envelope = build_envelope( - kind=kind, raw_ref_sha256=digest, received_by=received_by, - text=text, attachment_ids=attachment_ids, + body = _legacy_envelope( + kind=kind, + raw_ref_sha256=digest, + received_by=received_by, + text=text, + attachment_ids=attachment_ids, ) with kb.write_txn(conn): - conn.execute("UPDATE tasks SET body = ? WHERE id = ?", (envelope, task_id)) + conn.execute("UPDATE tasks SET body = ? WHERE id = ?", (body, task_id)) kb._append_event(conn, task_id, "intake_received", { "schema_version": 1, "actor": received_by, diff --git a/tests/hermes_cli/test_kanban_decompose.py b/tests/hermes_cli/test_kanban_decompose.py index 29842b4e5854..1f0b8695b469 100644 --- a/tests/hermes_cli/test_kanban_decompose.py +++ b/tests/hermes_cli/test_kanban_decompose.py @@ -83,8 +83,9 @@ def test_decompose_with_fanout_creates_children(kanban_home): "fanout": True, "rationale": "test split", "tasks": [ - {"title": "research", "body": "look it up", "assignee": "researcher", "parents": []}, - {"title": "build", "body": "code it", "assignee": "engineer", "parents": [0]}, + {"title": "scope", "body": "scope it", "assignee": "paul-park", "domain": "program-management", "parents": []}, + {"title": "research", "body": "look it up", "assignee": "researcher", "domain": "engineering", "parents": [0]}, + {"title": "build", "body": "code it", "assignee": "engineer", "domain": "engineering", "parents": [0]}, ], }) @@ -100,17 +101,20 @@ def test_decompose_with_fanout_creates_children(kanban_home): assert outcome.ok, outcome.reason assert outcome.fanout is True - assert outcome.child_ids and len(outcome.child_ids) == 2 + assert outcome.child_ids and len(outcome.child_ids) == 3 with kb.connect() as conn: root = kb.get_task(conn, tid) c0 = kb.get_task(conn, outcome.child_ids[0]) c1 = kb.get_task(conn, outcome.child_ids[1]) + c2 = kb.get_task(conn, outcome.child_ids[2]) assert root.status == "todo" assert c0.status == "ready" assert c1.status == "todo" + assert c2.status == "todo" assert c0.assignee == "paul-park" - assert c1.assignee == "engineer" + assert c1.assignee == "researcher" + assert c2.assignee == "engineer" def test_decompose_fanout_false_invalid_llm_assignee_uses_default(kanban_home): diff --git a/tests/hermes_cli/test_kanban_governed_intake.py b/tests/hermes_cli/test_kanban_governed_intake.py new file mode 100644 index 000000000000..65dca3898058 --- /dev/null +++ b/tests/hermes_cli/test_kanban_governed_intake.py @@ -0,0 +1,92 @@ +"""Tests for the governed intake envelope and typed scope handoff.""" +from __future__ import annotations + +from pathlib import Path + +import pytest + +from hermes_cli import kanban_db as kb +from hermes_cli.kanban_intake import build_envelope, parse_envelope, render_envelope + + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + kb.init_db() + return home + + +def test_envelope_round_trip_and_digest_is_deterministic(kanban_home): + first = build_envelope( + source="haa", + items=["linear:HEL-1", "notes"], + notes="internal", + attachment_refs=["/tmp/spec.pdf"], + tenant_domain="engineering", + sensitivity="confidential", + ) + second = build_envelope( + source="haa", + items=["linear:HEL-1", "notes"], + notes="internal", + attachment_refs=["/tmp/spec.pdf"], + tenant_domain="engineering", + sensitivity="confidential", + ) + assert first.content_digest == second.content_digest + assert parse_envelope(render_envelope(first)) == first + + +def test_governed_intake_create_is_idempotent(kanban_home): + envelope = build_envelope( + source="webhook:inbox", + items=["first", "second"], + tenant_domain="engineering", + ) + with kb.connect_closing() as conn: + first = kb.create_governed_intake_task( + conn, + title="raw intake", + body=render_envelope(envelope), + tenant=envelope.tenant_domain, + idempotency_key=envelope.idempotency_key, + created_by="haa", + ) + second = kb.create_governed_intake_task( + conn, + title="raw intake", + body=render_envelope(envelope), + tenant=envelope.tenant_domain, + idempotency_key=envelope.idempotency_key, + created_by="haa", + ) + count = conn.execute( + "SELECT COUNT(*) FROM tasks WHERE idempotency_key = ?", + (envelope.idempotency_key,), + ).fetchone()[0] + assert first[0] == second[0] + assert first[1] is True + assert second[1] is False + assert count == 1 + + +def test_scope_handoff_emits_typed_events(kanban_home): + with kb.connect_closing() as conn: + gate = kb.create_task( + conn, + title="scope intake", + assignee="paul-park", + triage=False, + ) + scope = kb.record_scope_handoff( + conn, + gate, + author="paul-park", + body="LINEAR_SCOPE: parent=HEL-3107 subissues=[{key:HEL-3115, cptc:3}]", + ) + events = kb.list_events(conn, gate) + assert scope == {"parent": "HEL-3107", "subissues": [{"key": "HEL-3115", "cptc": 3}]} + assert {event.kind for event in events} >= {"scope_recorded", "handoff_emitted"}