fix(acp): preload startup skills

acp_adapter/entry.py

@@ -138,9 +138,30 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         help="Accept all prompts (currently used by --setup-browser to skip the "
              "~400 MB Chromium download confirmation).",
     )
+    parser.add_argument(
+        "--skills",
+        "-s",
+        action="append",
+        default=None,
+        help="Preload one or more skills for every ACP session (repeat flag or comma-separate)",
+    )
     return parser.parse_args(argv)
 
 
+def _parse_skills_argument(skills: list[str] | None) -> list[str]:
+    """Normalize repeated/comma-separated ACP --skills flags."""
+    parsed: list[str] = []
+    seen: set[str] = set()
+    for raw in skills or []:
+        for part in str(raw).split(","):
+            normalized = part.strip()
+            if not normalized or normalized in seen:
+                continue
+            seen.add(normalized)
+            parsed.append(normalized)
+    return parsed
+
+
 def _print_version() -> None:
     from hermes_cli import __version__ as hermes_version
 
@@ -265,6 +286,7 @@ def main(argv: list[str] | None = None) -> None:
 
     import acp
     from .server import HermesACPAgent
+    from .session import SessionManager
 
     # MCP tool discovery from config.yaml — run before asyncio.run() so
     # it's safe to use blocking waits.  (ACP also registers per-session
@@ -277,7 +299,8 @@ def main(argv: list[str] | None = None) -> None:
     except Exception:
         logger.debug("MCP tool discovery failed at ACP startup", exc_info=True)
 
-    agent = HermesACPAgent()
+    startup_skills = _parse_skills_argument(args.skills)
+    agent = HermesACPAgent(session_manager=SessionManager(preloaded_skills=startup_skills))
     try:
         asyncio.run(acp.run_agent(agent, use_unstable_protocol=True))
     except KeyboardInterrupt:

acp_adapter/session.py

@@ -191,19 +191,21 @@ class SessionManager:
     via ``session_search``.
     """
 
-    def __init__(self, agent_factory=None, db=None):
+    def __init__(self, agent_factory=None, db=None, preloaded_skills: List[str] | None = None):
         """
         Args:
             agent_factory: Optional callable that creates an AIAgent-like object.
                            Used by tests. When omitted, a real AIAgent is created
                            using the current Hermes runtime provider configuration.
             db:            Optional SessionDB instance. When omitted, the default
                            SessionDB (``~/.hermes/state.db``) is lazily created.
+            preloaded_skills: Skill identifiers to inject into every ACP session.
         """
         self._sessions: Dict[str, SessionState] = {}
         self._lock = Lock()
         self._agent_factory = agent_factory
         self._db_instance = db  # None → lazy-init on first use
+        self._preloaded_skills = list(preloaded_skills or [])
 
     # ---- public API ---------------------------------------------------------
 
@@ -574,6 +576,7 @@ def _make_agent(
             return self._agent_factory()
 
         from run_agent import AIAgent
+        from agent.skill_commands import build_preloaded_skills_prompt
         from hermes_cli.config import load_config
         from hermes_cli.runtime_provider import resolve_runtime_provider
 
@@ -605,6 +608,17 @@ def _make_agent(
             "model": model or default_model,
         }
 
+        if self._preloaded_skills:
+            skills_prompt, loaded_skills, missing_skills = build_preloaded_skills_prompt(
+                self._preloaded_skills,
+                task_id=session_id,
+            )
+            if missing_skills:
+                missing_display = ", ".join(missing_skills)
+                raise ValueError(f"Unknown skill(s): {missing_display}")
+            if skills_prompt:
+                kwargs["ephemeral_system_prompt"] = skills_prompt
+
         try:
             runtime = resolve_runtime_provider(requested=requested_provider or config_provider)
             kwargs.update(
@@ -622,6 +636,8 @@ def _make_agent(
 
         _register_task_cwd(session_id, cwd)
         agent = AIAgent(**kwargs)
+        if self._preloaded_skills:
+            agent.preloaded_skills = loaded_skills
         # ACP stdio transport requires stdout to remain protocol-only JSON-RPC.
         # Route any incidental human-readable agent output to stderr instead.
         agent._print_fn = _acp_stderr_print

hermes_cli/main.py

@@ -11829,6 +11829,8 @@ def cmd_acp(args):
                 acp_argv.append("--setup-browser")
             if getattr(args, "assume_yes", False):
                 acp_argv.append("--yes")
+            for skill in getattr(args, "skills", None) or []:
+                acp_argv.extend(["--skills", skill])
             acp_main(acp_argv)
         except ImportError:
             print("ACP dependencies not installed.", file=sys.stderr)

tests/acp/test_entry.py

@@ -12,6 +12,7 @@ def test_main_enables_unstable_protocol(monkeypatch):
     calls = {}
 
     async def fake_run_agent(agent, **kwargs):
+        calls["agent"] = agent
         calls["kwargs"] = kwargs
 
     monkeypatch.setattr(entry, "_setup_logging", lambda: None)
@@ -23,6 +24,30 @@ async def fake_run_agent(agent, **kwargs):
     assert calls["kwargs"]["use_unstable_protocol"] is True
 
 
+def test_main_passes_startup_skills_to_session_manager(monkeypatch):
+    calls = {}
+
+    class FakeSessionManager:
+        def __init__(self, *, preloaded_skills=None):
+            calls["preloaded_skills"] = preloaded_skills
+
+    async def fake_run_agent(agent, **kwargs):
+        calls["agent"] = agent
+
+    monkeypatch.setattr(entry, "_setup_logging", lambda: None)
+    monkeypatch.setattr(entry, "_load_env", lambda: None)
+    monkeypatch.setattr("acp_adapter.session.SessionManager", FakeSessionManager)
+    monkeypatch.setattr(acp, "run_agent", fake_run_agent)
+
+    entry.main(["--skills", "plan,github-pr-workflow", "-s", "hermes-agent"])
+
+    assert calls["preloaded_skills"] == [
+        "plan",
+        "github-pr-workflow",
+        "hermes-agent",
+    ]
+
+
 def test_main_version_prints_without_starting_server(monkeypatch, capsys):
     monkeypatch.setattr(entry, "_setup_logging", lambda: (_ for _ in ()).throw(AssertionError("started server")))
 

tests/acp/test_session.py

@@ -43,6 +43,35 @@ def test_create_session_registers_task_cwd(self, manager, monkeypatch):
         state = manager.create_session(cwd="/tmp/work")
         assert calls == [(state.session_id, "/tmp/work")]
 
+    def test_create_session_preloads_startup_skills(self, tmp_path, monkeypatch):
+        captured_kwargs = {}
+
+        class FakeAgent:
+            model = "test-model"
+
+            def __init__(self, **kwargs):
+                captured_kwargs.update(kwargs)
+
+        monkeypatch.setattr("hermes_cli.config.load_config", lambda: {"model": "test-model"})
+        monkeypatch.setattr(
+            "hermes_cli.runtime_provider.resolve_runtime_provider",
+            lambda requested=None: {},
+        )
+        monkeypatch.setattr(
+            "agent.skill_commands.build_preloaded_skills_prompt",
+            lambda skills, task_id=None: ("PRELOADED SKILL PROMPT", ["plan"], []),
+        )
+
+        manager = SessionManager(
+            db=SessionDB(tmp_path / "state.db"),
+            preloaded_skills=["plan"],
+        )
+        with patch("run_agent.AIAgent", FakeAgent):
+            state = manager.create_session(cwd="/tmp/work")
+
+        assert captured_kwargs["ephemeral_system_prompt"] == "PRELOADED SKILL PROMPT"
+        assert state.agent.preloaded_skills == ["plan"]
+
 
     def test_register_task_cwd_translates_windows_drive_for_wsl_tools(self, monkeypatch):
         captured = {}