🔒 Fix Command Injection in ACP Runner terminal creation

zenith/src/zenith_harness/acp_runner.py

@@ -95,14 +95,18 @@ def _augment_acp_command(command: str, provider) -> str:
     For codex-acp this is the no-ask, no-sandbox combo — equivalent to
     `codex --dangerously-bypass-approvals-and-sandbox`, which codex-acp
     does not expose as a flag but accepts via `-c` overrides.
+
+    For hermes the command is passed through unchanged.
     """
-    if getattr(provider, "name", None) == "codex":
+    name = getattr(provider, "name", None)
+    if name == "codex":
         return (
             command
             + ' -c sandbox_mode="danger-full-access"'
             + ' -c approval_policy="never"'
             + ' -c model_reasoning_effort="xhigh"'
         )
+    # hermes: no-op
     return command
 
 
@@ -113,12 +117,16 @@ def _acp_subprocess_env(provider) -> dict[str, str]:
     `/usr/bin/env node`, and pass sandbox-disable hints through env. The
     command line also receives `sandbox_mode="danger-full-access"` in
     `_augment_acp_command`.
+
+    For hermes the env is passed through unchanged.
     """
     env = os.environ.copy()
-    if getattr(provider, "name", None) == "codex":
+    name = getattr(provider, "name", None)
+    if name == "codex":
         # Env-var hints — harmless if codex ignores them.
         env["CODEX_SANDBOX"] = "danger-full-access"
         env["CODEX_DISABLE_SANDBOX"] = "1"
+    # hermes: no special env needed
     return env
 
 
@@ -429,9 +437,9 @@ async def _handle_terminal_create(self, params: dict[str, Any]) -> dict[str, str
         env = os.environ.copy()
         for var in env_list:
             env[var["name"]] = var["value"]
-        full_cmd = " ".join([cmd, *args])
-        process = await asyncio.create_subprocess_shell(
-            full_cmd,
+        process = await asyncio.create_subprocess_exec(
+            cmd,
+            *args,
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.STDOUT,
             cwd=cwd,

zenith/src/zenith_harness/assets.py

@@ -111,6 +111,9 @@ def bundled_agents_dir(self, provider_name: str) -> Path:
         """Per-provider subagents directory (`bundled/providers/<name>/agents/`)."""
         return self.config.bundled_dir / "providers" / provider_name / "agents"
 
+    def bundled_prompts_dir(self) -> Path:
+        return self.config.bundled_dir / "prompts"
+
     def _iter_skill_dirs(
         self, project_id: str | None
     ) -> list[tuple[Literal["project", "personal", "bundled"], Path]]:

zenith/src/zenith_harness/bundled/prompts/orchestrator/hermes_addendum.md

@@ -0,0 +1,72 @@
+# Hermes Addendum — Orchestrator Recall & Session Memory
+
+This addendum applies when Hermes is the orchestrator provider. It extends the base orchestrator prompt with Hermes-specific recall mechanisms.
+
+## Hermes Total Recall
+
+Hermes maintains a persistent session database (`~/.hermes/state.db` with FTS5) that survives across sessions, compressions, and restarts. Use it as a **queryable append-only log** of every zenith state read.
+
+### Recall Protocol
+
+**Before every `inspect_project` / `advance_project` / `decide_attention` call:**
+```
+session_search(query="project_id mission_id", role_filter="user,assistant,tool")
+```
+
+**After every MCP tool call that returns zenith state:**
+```
+memory(action="add", target="memory", content="zenith state: {project_id} {mission_id} {state} {key_findings}")
+```
+
+### Query Patterns
+
+| Need | Query |
+|------|-------|
+| Last known state | `session_search(query="project_id state", limit=1, sort="newest")` |
+| All attention items | `session_search(query="attention_needed project_id", limit=10)` |
+| Decision history | `session_search(query="decide_attention project_id", role_filter="assistant,tool")` |
+| Contract changes | `session_search(query="contract VAL- project_id", limit=20)` |
+| Failed tasks | `session_search(query="failed project_id", limit=10)` |
+
+### Memory Promotion
+
+When Hermes session compression drops context, **promote critical findings to Zenith disk state first:**
+
+```python
+# Before compression triggers (at ~50% threshold):
+# 1. Write to mission.md / MEMORY.md / decisions/
+# 2. Then let Hermes compress
+```
+
+The Zenith disk state (`brief.md`, `mission.md`, `contract/`, `decisions/`, `MEMORY.md`) is the **canonical memory**. Hermes session DB is the **audit trail**.
+
+### Skills Loading
+
+Hermes loads skills from:
+- `.hermes/skills/` (project-specific, written by `zenith init`)
+- `~/.hermes/skills/` (user-installed)
+- Bundled skills (copied to `.hermes/skills/` by `zenith init`)
+
+To load a skill mid-mission: `/skill engineering-mission-playbook` or `hermes -s engineering-mission-playbook chat ...`
+
+### Compression Awareness
+
+Default threshold: 50% context window. Lower for zenith missions:
+```bash
+hermes config set compression.threshold 0.3
+hermes config set compression.target_ratio 0.2
+```
+
+### Slash Commands in Mission
+
+| Command | Use |
+|---------|-----|
+| `/inspect` | Show current session info (tokens, tools, skills) |
+| `/memory` | Search/add memory entries |
+| `/session_search` | Query past zenith sessions |
+| `/skill <name>` | Load a skill |
+| `/reload-skills` | Re-scan skill directories |
+
+---
+
+**Rule:** Every zenith state transition (start_project → submit_plan → advance_project → attention → decide → advance → end_mission) should leave a trace in both Hermes session DB *and* Zenith disk state. The disk state is the contract; the session DB is the witness.

zenith/src/zenith_harness/cli.py

@@ -442,6 +442,11 @@ def _setup_provider_assets(
         if not path.exists():
             path.parent.mkdir(parents=True, exist_ok=True)
             body = loader.load_prompt_file("orchestrator", "system_prompt.md")
+            # Append Hermes-specific addendum for Hermes provider
+            if provider.name == "hermes":
+                addendum_path = loader.bundled_prompts_dir() / "orchestrator" / "hermes_addendum.md"
+                if addendum_path.exists():
+                    body += "\n\n---\n\n" + addendum_path.read_text(encoding="utf-8")
             path.write_text(body, encoding="utf-8")
             click.echo(f"Created {path}")
 

zenith/src/zenith_harness/providers.py

@@ -3,16 +3,18 @@
 from dataclasses import dataclass
 from typing import Literal
 
-ProviderName = Literal["claude", "codex"]
+ProviderName = Literal["claude", "codex", "hermes"]
 ConfigFormat = Literal["mcp_json", "codex_config"]
 
 ORCHESTRATOR_PROVIDER_NAMES: tuple[ProviderName, ...] = (
     "claude",
     "codex",
+    "hermes",
 )
 WORKER_PROVIDER_NAMES: tuple[ProviderName, ...] = (
     "claude",
     "codex",
+    "hermes",
 )
 
 
@@ -138,6 +140,17 @@ def _dedupe_paths(paths: tuple[str, ...]) -> tuple[str, ...]:
         orchestrator_prompt_output_path=".codex/orchestrator_prompt.md",
         acp_supports_system_prompt=False,
     ),
+    "hermes": ProviderDefinition(
+        name="hermes",
+        skill_dirs=(".hermes/skills", ".agents/skills"),
+        skill_alias_dirs=(".hermes/skills", ".agents/skills"),
+        config_format="mcp_json",
+        default_worker_acp_command="hermes acp",
+        agent_output_dir=".hermes/agents",
+        orchestrator_prompt_output_path=".hermes/orchestrator_prompt.md",
+        acp_supports_system_prompt=True,
+        acp_runtime_mode=None,
+    ),
 }
 
 

zenith/tests/test_cli.py

@@ -1,4 +1,5 @@
 """CLI integration tests — init / list-projects / show-project / install-skills."""
+
 from __future__ import annotations
 
 import json
@@ -64,19 +65,13 @@ def test_init_does_not_touch_gitignore(
         gitignore = workspace / ".gitignore"
         gitignore.write_text("node_modules/\n")
         original = gitignore.read_text()
-        r = runner.invoke(
-            cli, ["init", "--workspace-dir", str(workspace), "--agent", "claude"]
-        )
+        r = runner.invoke(cli, ["init", "--workspace-dir", str(workspace), "--agent", "claude"])
         assert r.exit_code == 0, r.output
         assert gitignore.read_text() == original
 
-    def test_idempotent(
-        self, runner: CliRunner, workspace: Path, env: dict[str, str]
-    ) -> None:
+    def test_idempotent(self, runner: CliRunner, workspace: Path, env: dict[str, str]) -> None:
         for _ in range(2):
-            r = runner.invoke(
-                cli, ["init", "--workspace-dir", str(workspace), "--agent", "claude"]
-            )
+            r = runner.invoke(cli, ["init", "--workspace-dir", str(workspace), "--agent", "claude"])
             assert r.exit_code == 0, r.output
         # .mcp.json preserved across reruns.
         assert (workspace / ".mcp.json").exists()
@@ -94,7 +89,10 @@ def test_codex_writes_codex_config(
         assert server["args"] == _expected_mcp_server_args()
         assert f"Initialized v5 project workspace at {workspace}" in r.output
         assert "Start your agent from the initialized project workspace" in r.output
-        assert "Read .codex/orchestrator_prompt.md and use Zenith to run this mission." in r.output
+        assert (
+            "Read the .codex/orchestrator_prompt.md and treat it as your primary role, then use Zenith to run this mission."
+            in r.output
+        )
 
     def test_claude_init_writes_runtime_validator_env_names(
         self, runner: CliRunner, workspace: Path, env: dict[str, str]