INT-C1: retain the installed interpreter in CLI runs

README.md

@@ -63,6 +63,8 @@ hermes-workflows status \
   --id wf_typed_quickstart
 ```
 
+`hermes-workflows run` stays in the Python interpreter and environment that installed the console script. It does not discover or invoke an unrelated `uv` on `PATH`. Activate and install into the environment you want before running the command; `--project-root` controls project/registry discovery, not interpreter selection.
+
 `run` records or replays the workflow instance. It does not pretend the current process is a forever worker. The canonical foreground continuation command is `hermes-workflows runner run --config ...`: it leases queued workflow/step/agent/bash/child-workflow commands, records outputs, and re-enters the workflow until it is waiting for Review Queue input or terminal. The older `hermes-workflows worker --config ...` spelling remains a compatibility alias, but new docs and operators should prefer `runner run` / `runner once`.
 
 Respond to the Review Queue request through the Hermes dashboard/plugin or another configured review adapter, then start the runner again if you used the bounded smoke command above. A recorded operator response always creates a visible durable continuation command; the runner, not a hidden chat callback, consumes that command. In a real supervised setup, keep `runner run` alive under launchd, systemd, s6, tmux, or another supervisor only after the foreground command works in your workspace. The response payload must match the `returns=` dataclass schema and include provenance from the adapter that recorded it.

docs/setup-for-agents.md

@@ -72,6 +72,8 @@ hermes-workflows run typed-quickstart \
   --input-json '{"change":"Expose typed workflow contracts."}'
 ```
 
+The command retains the interpreter, `VIRTUAL_ENV`, and environment that own the installed `hermes-workflows` console script. A `uv` executable elsewhere on `PATH` is ignored. Activate and install into the intended environment first; `--project-root` changes project/registry discovery only and never selects another Python environment.
+
 `run` records the workflow activation and queues missing work. It is not the always-on continuation loop. A run can return `running` before the Review Queue request exists because a runner still needs to execute queued steps and replay the workflow to the next wait.
 
 The exact serialized input and initial `run` result are:

src/hermes_workflows/cli.py

@@ -4,8 +4,6 @@
 import json
 import os
 import platform
-import shutil
-import subprocess
 import sys
 import time
 from collections import Counter
@@ -18,6 +16,7 @@
 from .dashboard import render_dashboard
 from .dashboard_server import serve_dashboard
 from .engine import JsonCodec, RunResult, WorkflowEngine
+from .installed_environment import resolve_installed_execution
 from .invocation import InvocationService, TrustedResumer
 from .registry import WorkflowRegistry, looks_like_path
 from .runner_api import default_db_path, default_workflow_id, infer_project_root, run_workflow_function
@@ -175,34 +174,6 @@ def run_doctor(args: argparse.Namespace) -> int:
     return 0
 
 
-def _uv_cwd_for_run(args: argparse.Namespace) -> Path:
-    if args.project_root is not None:
-        return infer_project_root(args.project_root)
-    if args.config is not None:
-        return infer_project_root(start=args.config)
-    ref = str(args.workflow_ref)
-    path_candidate = ref.rsplit(":", 1)[0] if ":" in ref and ref.rsplit(":", 1)[0].endswith(".py") else ref
-    path = Path(path_candidate).expanduser()
-    if path.suffix == ".py" or path.exists():
-        return infer_project_root(start=path)
-    return Path.cwd()
-
-
-def run_via_uv(raw_argv: list[str], args: argparse.Namespace) -> int:
-    uv = shutil.which("uv")
-    if uv is None:
-        # Keep the CLI usable in minimal environments; tests and normal installs
-        # with uv still exercise the blessed uv path.
-        return main(["_run-engine", *raw_argv[1:]])
-    child_args = ["_run-engine", *raw_argv[1:]]
-    cmd = [uv, "run", "python", "-m", "hermes_workflows", *child_args]
-    env = os.environ.copy()
-    env.pop("VIRTUAL_ENV", None)
-    env["HERMES_WORKFLOWS_UV_CHILD"] = "1"
-    completed = subprocess.run(cmd, env=env, cwd=_uv_cwd_for_run(args))
-    return int(completed.returncode)
-
-
 def resolve_run_invocation(args: argparse.Namespace) -> tuple[Callable[..., Any], str, Path, str, Any]:
     config_path = args.config
     if args.project_root is not None:
@@ -302,6 +273,13 @@ def run_engine_cli(args: argparse.Namespace) -> int:
     return 0
 
 
+def run_installed_cli(args: argparse.Namespace) -> int:
+    execution = resolve_installed_execution()
+    if execution.python_executable != sys.executable or execution.environment != dict(os.environ):
+        raise RuntimeError("installed workflow execution must retain the current interpreter and environment")
+    return run_engine_cli(args)
+
+
 def run_worker_registry_cli(args: argparse.Namespace) -> int:
     registry_obj = WorkflowRegistry.from_sources(config_path=args.config)
     try:
@@ -510,8 +488,7 @@ def add_runner_service_args(parser: argparse.ArgumentParser) -> None:
 
 
 def main(argv: list[str] | None = None) -> int:
-    raw_argv = list(argv) if argv is not None else sys.argv[1:]
-    argv = normalize_agent_value_options(raw_argv)
+    argv = normalize_agent_value_options(list(argv) if argv is not None else sys.argv[1:])
     parser = argparse.ArgumentParser(prog="hermes-workflows")
     visible_commands = (
         "{registry,invoke,resume-trusted,resume-pending,start,run,runner,worker,signal,"
@@ -560,7 +537,10 @@ def main(argv: list[str] | None = None) -> int:
     start.add_argument("--id", required=True, dest="workflow_id")
     start.add_argument("--input-json", required=True)
 
-    run = sub.add_parser("run", help="Run or resume a workflow through uv using a registry name, module ref, or file path")
+    run = sub.add_parser(
+        "run",
+        help="Run or resume a workflow in the installed environment using a registry name, module ref, or file path",
+    )
     run.add_argument("workflow_ref", help="registry alias, module:function, workflow.py, or workflow.py:function")
     run.add_argument("--config", type=Path)
     run.add_argument("--db", help="configured DB alias or explicit local DB path; defaults to <project-root>/.hermes/workflows.sqlite")
@@ -571,20 +551,6 @@ def main(argv: list[str] | None = None) -> int:
     run.add_argument("--watch", action="store_true", help="Keep re-invoking the same workflow entrypoint/DB until terminal or --max-resumes")
     run.add_argument("--poll-interval", type=float, default=1.0)
     run.add_argument("--max-resumes", type=positive_int)
-    run.add_argument("--direct", action="store_true", help=argparse.SUPPRESS)
-
-    run_engine = sub.add_parser("_run-engine", help=argparse.SUPPRESS)
-    run_engine.add_argument("workflow_ref", help=argparse.SUPPRESS)
-    run_engine.add_argument("--config", type=Path)
-    run_engine.add_argument("--db")
-    run_engine.add_argument("--id", dest="workflow_id")
-    run_engine.add_argument("--input-json", default="{}")
-    run_engine.add_argument("--project-root", type=Path)
-    run_engine.add_argument("--no-drain", action="store_true")
-    run_engine.add_argument("--watch", action="store_true")
-    run_engine.add_argument("--poll-interval", type=float, default=1.0)
-    run_engine.add_argument("--max-resumes", type=positive_int)
-    sub._choices_actions = [action for action in sub._choices_actions if action.dest != "_run-engine"]
 
     runner = sub.add_parser("runner", help="Canonical Workflow Runner v2 commands")
     runner_sub = runner.add_subparsers(dest="runner_command", required=True)
@@ -796,12 +762,7 @@ def main(argv: list[str] | None = None) -> int:
     if args.command == "doctor":
         return run_doctor(args)
     if args.command == "run":
-        if args.direct:
-            args.command = "_run-engine"
-            return run_engine_cli(args)
-        return run_via_uv(raw_argv, args)
-    if args.command == "_run-engine":
-        return run_engine_cli(args)
+        return run_installed_cli(args)
     if args.command == "worker" and args.config is not None:
         return run_worker_registry_cli(args)
     if args.command == "worker" and (not args.workflow_ref or not args.db or not args.workflow_id):

tests/test_cli.py

@@ -1352,9 +1352,13 @@ def test_cli_serve_dashboard_can_approve_waiting_workflow(tmp_path):
     }
 
 
-def test_hermes_workflows_run_uses_uv_and_project_default_db_for_registry_alias(tmp_path):
+def test_hermes_workflows_run_retains_installed_interpreter_without_invoking_uv(tmp_path):
+    execution_record = tmp_path / "installed-execution.json"
     (tmp_path / "alias_wf.py").write_text(
-        "from hermes_workflows import wait_for, workflow\n"
+        "import json, os, sys\n"
+        "from pathlib import Path\n"
+        "from hermes_workflows import workflow\n"
+        f"Path({str(execution_record)!r}).write_text(json.dumps({{'python_executable': sys.executable, 'virtual_env': os.environ.get('VIRTUAL_ENV')}}))\n"
         "@workflow\n"
         "async def alias_workflow(inputs):\n"
         "    return {'message': inputs['message']}\n"
@@ -1375,17 +1379,15 @@ def test_hermes_workflows_run_uses_uv_and_project_default_db_for_registry_alias(
     )
     fake_bin = tmp_path / "bin"
     fake_bin.mkdir()
-    marker = tmp_path / "uv-called.json"
+    marker = tmp_path / "uv-called"
     fake_uv = fake_bin / "uv"
     fake_uv.write_text(
-        "#!/usr/bin/env python3\n"
-        "import json, os, subprocess, sys\n"
-        f"open({str(marker)!r}, 'w').write(json.dumps({{'args': sys.argv[1:], 'cwd': os.getcwd()}}))\n"
-        "args = sys.argv[1:]\n"
-        "assert args[0] == 'run'\n"
-        "raise SystemExit(subprocess.run(args[1:]).returncode)\n"
+        "#!/bin/sh\n"
+        f"touch {marker}\n"
+        "exit 97\n"
     )
     fake_uv.chmod(0o755)
+    installed_env = tmp_path / "installed-env"
 
     payload = json.loads(
         run_cli(
@@ -1396,12 +1398,21 @@ def test_hermes_workflows_run_uses_uv_and_project_default_db_for_registry_alias(
             str(registry),
             "--project-root",
             str(tmp_path),
-            env_extra={"PATH": f"{fake_bin}:{os.environ['PATH']}"},
+            env_extra={
+                "PATH": f"{fake_bin}:{os.environ['PATH']}",
+                "VIRTUAL_ENV": str(installed_env),
+            },
         ).stdout
     )
 
     assert payload["status"] == "running"
     assert payload["waiting_on"] is None
+    execution = json.loads(execution_record.read_text())
+    assert execution == {
+        "python_executable": sys.executable,
+        "virtual_env": str(installed_env),
+    }
+    assert not marker.exists()
     assert (tmp_path / ".hermes" / "workflows.sqlite").exists()
     completed_payload = run_worker(
         tmp_path,
@@ -1412,11 +1423,6 @@ def test_hermes_workflows_run_uses_uv_and_project_default_db_for_registry_alias(
     )
     assert completed_payload["status"] == "completed"
     assert completed_payload["result"] == {"message": "from-registry"}
-    uv_call = json.loads(marker.read_text())
-    uv_args = uv_call["args"]
-    assert uv_call["cwd"] == str(tmp_path)
-    assert uv_args[:4] == ["run", "python", "-m", "hermes_workflows"]
-    assert "_run-engine" in uv_args
 
 
 def test_workflow_run_helper_supports_direct_uv_script_style_and_default_db(tmp_path):
@@ -1727,7 +1733,7 @@ def test_direct_workflow_run_defaults_db_to_workflow_file_project(tmp_path):
     assert not (caller / ".hermes" / "workflows.sqlite").exists()
 
 
-def test_run_via_uv_uses_config_project_as_child_process_cwd(tmp_path):
+def test_run_with_external_config_ignores_path_uv_and_defaults_db_to_config_project(tmp_path):
     project = tmp_path / "workflow_project"
     caller = tmp_path / "caller"
     project.mkdir()
@@ -1746,23 +1752,20 @@ def test_run_via_uv_uses_config_project_as_child_process_cwd(tmp_path):
                 "workflows": {
                     "configured": {
                         "workflow_ref": str(project / "configured_wf.py") + ":configured_wf",
-                        "default_input": {"project": "uv-cwd"},
+                        "default_input": {"project": "installed-environment"},
                     }
                 }
             }
         )
     )
     fake_bin = tmp_path / "bin"
     fake_bin.mkdir()
-    marker = tmp_path / "uv-cwd.json"
+    marker = tmp_path / "uv-called"
     fake_uv = fake_bin / "uv"
     fake_uv.write_text(
-        "#!/usr/bin/env python3\n"
-        "import json, os, subprocess, sys\n"
-        f"open({str(marker)!r}, 'w').write(json.dumps({{'args': sys.argv[1:], 'cwd': os.getcwd()}}))\n"
-        "args = sys.argv[1:]\n"
-        "assert args[0] == 'run'\n"
-        "raise SystemExit(subprocess.run(args[1:]).returncode)\n"
+        "#!/bin/sh\n"
+        f"touch {marker}\n"
+        "exit 97\n"
     )
     fake_uv.chmod(0o755)
 
@@ -1783,8 +1786,7 @@ def test_run_via_uv_uses_config_project_as_child_process_cwd(tmp_path):
     assert payload["status"] == "running"
     assert payload["waiting_on"] is None
     assert payload["result"] is None
-    uv_call = json.loads(marker.read_text())
-    assert uv_call["cwd"] == str(project)
+    assert not marker.exists()
     assert (project / ".hermes" / "workflows.sqlite").exists()
     assert not (caller / ".hermes" / "workflows.sqlite").exists()
 
@@ -1807,7 +1809,6 @@ def test_module_ref_run_defaults_db_to_imported_module_project(tmp_path):
             [Path.cwd() / "src", project],
             "run",
             "module_wf:module_workflow",
-            "--direct",
             "--id",
             "wf_module_ref",
         ).stdout
@@ -1820,7 +1821,7 @@ def test_module_ref_run_defaults_db_to_imported_module_project(tmp_path):
     assert not (caller / ".hermes" / "workflows.sqlite").exists()
 
 
-def test_internal_run_engine_command_is_hidden_from_top_level_help(tmp_path):
+def test_removed_internal_run_engine_command_is_absent_from_top_level_help(tmp_path):
     completed = run_cli_from(tmp_path, [Path.cwd() / "src", tmp_path], "--help")
     assert "_run-engine" not in completed.stdout
 

tests/test_uv_run_contract.py

@@ -1,5 +1,8 @@
+import inspect
 from pathlib import Path
 
+from hermes_workflows import cli
+
 try:
     import tomllib
 except ModuleNotFoundError:  # pragma: no cover
@@ -12,3 +15,19 @@ def test_pyproject_declares_uv_default_dev_group_for_plain_uv_run_pytest():
     assert any(str(dep).startswith('pytest') for dep in dev_group), (
         'plain `uv run pytest` must install/use the project pytest, not leak to an active external venv PATH pytest'
     )
+
+
+def test_public_run_uses_installed_environment_without_uv_trampoline_or_hidden_bypass():
+    installed_run_source = inspect.getsource(cli.run_installed_cli)
+    cli_source = Path(cli.__file__).read_text()
+
+    assert "resolve_installed_execution()" in installed_run_source
+    for obsolete_token in (
+        "run_via_uv",
+        "_run-engine",
+        "--direct",
+        "HERMES_WORKFLOWS_UV_CHILD",
+        'shutil.which("uv")',
+        '"uv", "run", "python"',
+    ):
+        assert obsolete_token not in cli_source