HW-03: retain the installed interpreter for workflow runs

docs/operations/installed-environment.md

@@ -0,0 +1,29 @@
+# Installed environment contract
+
+The `hermes-workflows` console script belongs to the Python environment that installed the wheel. Normal workflow execution must retain that process's `sys.executable`, `VIRTUAL_ENV`, and environment. Discovering an unrelated `uv` on `PATH` is diagnostic only; it must not cause project discovery, a subprocess trampoline, or environment mutation.
+
+`hermes_workflows.installed_environment.resolve_installed_execution()` captures this contract without executing anything found on `PATH`. `installed_environment_report()` emits the non-secret identity needed for operations:
+
+- exact Python executable and virtual environment;
+- installed package origin and version;
+- foundation package-manifest SHA-256 / ownership key;
+- absolute registry path and byte fingerprint;
+- configured DB alias; and
+- the visible `uv` path, if any, as evidence that its presence does not control execution.
+
+The report deliberately omits the full environment because it may contain credentials. It reports the DB alias, not registry-v2 catalog semantics. Reported aliases use the same bounded identifier shape as registry-location aliases: 1–64 lowercase ASCII letters, digits, underscores, or hyphens, beginning with a letter. Invalid values fail with a fixed, non-reflective error.
+
+## Verification
+
+Run:
+
+    uv run pytest -q tests/test_installed_cli_environment.py
+    python tests/probes/installed_cli_smoke.py
+
+The probe normalizes its work root once before creating paths. It builds a wheel, installs it without an editable checkout into a new virtual environment, puts a marker-writing fake `uv` first on `PATH`, and exercises the packaged typed quickstart through the installed interpreter and console script. Success means the workflow reaches `signal:operator.response:review_release_note`, both the interpreter and package origin are beneath the temporary virtual environment, identity fingerprints are present, and the fake marker is absent.
+
+The probe may not use private parser switches, private engine commands, or child-marker environment variables to evade the public execution contract. Central CLI deletion and wiring remain owned by INT-C1; this module is the isolated resolver and evidence seam that integration consumes.
+
+## Rollback
+
+Rollback may remove this resolver only together with its unconsumed integration. It must not restore implicit `uv` execution or environment stripping. A project-specific environment runner, if ever needed, must be a separate explicit command and contract.

src/hermes_workflows/installed_environment.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+import hashlib
+import os
+import re
+import shutil
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Dict, Mapping, Optional, Union
+
+from .package_resources import foundation_manifest, installed_package_version, ownership_key
+
+
+PathLike = Union[str, Path]
+_DB_ALIAS_PATTERN = re.compile(r"^[a-z][a-z0-9_-]{0,63}$")
+
+
+@dataclass(frozen=True)
+class InstalledExecution:
+    """The interpreter and environment owned by the installed console script."""
+
+    python_executable: str
+    environment: Dict[str, str]
+    visible_uv: Optional[str]
+
+
+@dataclass(frozen=True)
+class InstalledEnvironmentReport:
+    """Non-secret identity fields for an installed workflow process."""
+
+    schema_version: int
+    python_executable: str
+    virtual_env: Optional[str]
+    visible_uv: Optional[str]
+    package_origin: str
+    package_version: str
+    package_manifest_sha256: str
+    package_ownership_key: str
+    registry_path: str
+    registry_sha256: str
+    db_alias: str
+
+    def to_dict(self) -> Dict[str, object]:
+        return {
+            "schema_version": self.schema_version,
+            "python_executable": self.python_executable,
+            "virtual_env": self.virtual_env,
+            "visible_uv": self.visible_uv,
+            "package_origin": self.package_origin,
+            "package_version": self.package_version,
+            "package_manifest_sha256": self.package_manifest_sha256,
+            "package_ownership_key": self.package_ownership_key,
+            "registry_path": self.registry_path,
+            "registry_sha256": self.registry_sha256,
+            "db_alias": self.db_alias,
+        }
+
+
+def resolve_installed_execution(
+    *,
+    environ: Optional[Mapping[str, str]] = None,
+    python_executable: Optional[str] = None,
+) -> InstalledExecution:
+    """Retain the current interpreter and environment without project discovery.
+
+    The returned mapping is a copy so callers may add explicit child-process
+    settings without mutating the console script's process environment.
+    """
+
+    source = os.environ if environ is None else environ
+    environment = {str(key): str(value) for key, value in source.items()}
+    executable = sys.executable if python_executable is None else str(python_executable)
+    if not executable:
+        raise ValueError("python_executable must be nonblank")
+    visible_uv = shutil.which("uv", path=environment.get("PATH"))
+    return InstalledExecution(
+        python_executable=executable,
+        environment=environment,
+        visible_uv=visible_uv,
+    )
+
+
+def installed_environment_report(
+    *,
+    registry_path: PathLike,
+    db_alias: str,
+    environ: Optional[Mapping[str, str]] = None,
+    python_executable: Optional[str] = None,
+) -> InstalledEnvironmentReport:
+    """Resolve safe package, interpreter, registry, and DB-alias identity."""
+
+    if not isinstance(db_alias, str) or _DB_ALIAS_PATTERN.fullmatch(db_alias) is None:
+        raise ValueError(f"db_alias must match {_DB_ALIAS_PATTERN.pattern}")
+    alias = db_alias
+    registry = Path(registry_path).expanduser().resolve(strict=True)
+    if not registry.is_file():
+        raise ValueError("registry_path must identify a file")
+
+    execution = resolve_installed_execution(
+        environ=environ,
+        python_executable=python_executable,
+    )
+    registry_sha256 = hashlib.sha256(registry.read_bytes()).hexdigest()
+    package_manifest_sha256 = ownership_key(foundation_manifest())
+    return InstalledEnvironmentReport(
+        schema_version=1,
+        python_executable=execution.python_executable,
+        virtual_env=execution.environment.get("VIRTUAL_ENV"),
+        visible_uv=execution.visible_uv,
+        package_origin=str(Path(__file__).resolve()),
+        package_version=installed_package_version(),
+        package_manifest_sha256=package_manifest_sha256,
+        package_ownership_key=package_manifest_sha256,
+        registry_path=str(registry),
+        registry_sha256=registry_sha256,
+        db_alias=alias,
+    )

tests/fixtures/install_smoke_registry_v2.json

@@ -0,0 +1 @@
+{"db":{"alias":"smoke-db","relative_path":"state/workflows.sqlite"},"input":{"change":"Expose typed workflow contracts."},"registry_location":{"registry_file":".hermes/workflows.registry.json","state_root":"state"},"schema_version":2,"workflow":{"alias":"install-smoke","workflow_ref":"hermes_workflows.examples.install_smoke:release_note_workflow"}}

tests/probes/installed_cli_smoke.py

@@ -0,0 +1,217 @@
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import shlex
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+FIXTURE = REPO_ROOT / "tests" / "fixtures" / "install_smoke_registry_v2.json"
+WORKFLOW_ID = "wf_installed_cli_smoke"
+EXPECTED_WAIT = "signal:operator.response:review_release_note"
+
+
+def _run(command: List[str], *, cwd: Path, env: Optional[Dict[str, str]] = None) -> subprocess.CompletedProcess:
+    return subprocess.run(
+        command,
+        cwd=cwd,
+        env=env,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=True,
+    )
+
+
+def _venv_executable(venv: Path, name: str) -> Path:
+    scripts = venv / ("Scripts" if os.name == "nt" else "bin")
+    suffix = ".exe" if os.name == "nt" else ""
+    return scripts / f"{name}{suffix}"
+
+
+def _is_beneath(path: Path, parent: Path) -> bool:
+    absolute_path = Path(os.path.abspath(str(path)))
+    absolute_parent = Path(os.path.abspath(str(parent)))
+    candidates = (
+        (absolute_path, absolute_parent),
+        (absolute_path.resolve(), absolute_parent.resolve()),
+    )
+    for candidate, candidate_parent in candidates:
+        try:
+            candidate.relative_to(candidate_parent)
+        except ValueError:
+            continue
+        return True
+    return False
+
+
+def _identity_command(python: Path, registry: Path, db_alias: str) -> List[str]:
+    program = (
+        "import json; "
+        "from hermes_workflows.installed_environment import installed_environment_report; "
+        "print(json.dumps(installed_environment_report("
+        "registry_path=" + repr(str(registry)) + ",db_alias=" + repr(db_alias) + ").to_dict(),sort_keys=True))"
+    )
+    return [str(python), "-c", program]
+
+
+def run_probe(work_root: Path) -> Dict[str, Any]:
+    work_root = Path(os.path.abspath(str(work_root.expanduser())))
+    work_root.mkdir(parents=True, exist_ok=False)
+    dist = work_root / "dist"
+    dist.mkdir()
+    build = _run(
+        [sys.executable, "-m", "build", "--wheel", "--outdir", str(dist)],
+        cwd=REPO_ROOT,
+    )
+    wheels = sorted(dist.glob("*.whl"))
+    if len(wheels) != 1:
+        raise RuntimeError(f"expected one built wheel, found {len(wheels)}")
+    wheel = wheels[0]
+
+    venv = work_root / "venv"
+    _run([sys.executable, "-m", "venv", str(venv)], cwd=work_root)
+    python = _venv_executable(venv, "python")
+    installed_cli = _venv_executable(venv, "hermes-workflows")
+    install = _run(
+        [str(python), "-m", "pip", "install", "--disable-pip-version-check", "--no-deps", str(wheel)],
+        cwd=work_root,
+    )
+
+    fake_bin = work_root / "fake-bin"
+    fake_bin.mkdir()
+    marker = work_root / "fake-uv-called"
+    fake_uv = fake_bin / "uv"
+    fake_uv.write_text(
+        "#!/bin/sh\ntouch " + shlex.quote(str(marker)) + "\nexit 97\n",
+        encoding="utf-8",
+    )
+    fake_uv.chmod(0o755)
+
+    workspace = work_root / "workspace"
+    registry = workspace / ".hermes" / "workflows.registry.json"
+    registry.parent.mkdir(parents=True)
+    registry.write_bytes(FIXTURE.read_bytes())
+    fixture = json.loads(registry.read_text(encoding="utf-8"))
+    db_path = registry.parent / fixture["db"]["relative_path"]
+    db_path.parent.mkdir(parents=True)
+
+    env = os.environ.copy()
+    env.pop("PYTHONPATH", None)
+    env["PATH"] = os.pathsep.join((str(fake_bin), env.get("PATH", "")))
+    env["VIRTUAL_ENV"] = str(venv)
+    input_json = json.dumps(fixture["input"], sort_keys=True, separators=(",", ":"))
+
+    started = _run(
+        [
+            str(python),
+            "-m",
+            "hermes_workflows.examples.install_smoke",
+            "--db",
+            str(db_path),
+            "--id",
+            WORKFLOW_ID,
+            "--input-json",
+            input_json,
+        ],
+        cwd=workspace,
+        env=env,
+    )
+    worker = _run(
+        [
+            str(installed_cli),
+            "worker",
+            fixture["workflow"]["workflow_ref"],
+            "--db",
+            str(db_path),
+            "--id",
+            WORKFLOW_ID,
+            "--max-commands",
+            "10",
+        ],
+        cwd=workspace,
+        env=env,
+    )
+    status = _run(
+        [
+            str(installed_cli),
+            "status",
+            "--db",
+            str(db_path),
+            "--id",
+            WORKFLOW_ID,
+        ],
+        cwd=workspace,
+        env=env,
+    )
+    identity = _run(
+        _identity_command(python, registry, fixture["db"]["alias"]),
+        cwd=workspace,
+        env=env,
+    )
+
+    started_payload = json.loads(started.stdout)
+    worker_payload = json.loads(worker.stdout)
+    status_payload = json.loads(status.stdout)
+    identity_payload = json.loads(identity.stdout)
+    if started_payload["status"] != "running":
+        raise RuntimeError(f"unexpected start status: {started_payload}")
+    if worker_payload["status"] != "waiting" or worker_payload["waiting_on"] != EXPECTED_WAIT:
+        raise RuntimeError(f"typed wait was not reached: {worker_payload}")
+    if status_payload["status"] != "waiting" or status_payload["waiting_on"] != EXPECTED_WAIT:
+        raise RuntimeError(f"status did not preserve typed wait: {status_payload}")
+
+    package_origin = Path(identity_payload["package_origin"])
+    cli_shebang = installed_cli.read_text(encoding="utf-8").splitlines()[0]
+    expected_shebang = "#!" + str(python)
+    if cli_shebang != expected_shebang:
+        raise RuntimeError(f"installed console script uses an unexpected interpreter: {cli_shebang!r}")
+    result = {
+        **identity_payload,
+        "status": status_payload["status"],
+        "waiting_on": status_payload["waiting_on"],
+        "installed_cli_executable": str(python),
+        "installed_cli_path": str(installed_cli),
+        "installed_cli_shebang": cli_shebang,
+        "interpreter_under_venv": _is_beneath(Path(identity_payload["python_executable"]), venv),
+        "package_origin_under_venv": _is_beneath(package_origin, venv),
+        "fixture_sha256": hashlib.sha256(FIXTURE.read_bytes()).hexdigest(),
+        "wheel_sha256": hashlib.sha256(wheel.read_bytes()).hexdigest(),
+        "fake_uv_visible": identity_payload["visible_uv"] == str(fake_uv),
+        "fake_uv_called": marker.exists(),
+        "hidden_bypasses_used": [],
+        "commands": {
+            "build": build.args,
+            "install": install.args,
+            "start": started.args,
+            "worker": worker.args,
+            "status": status.args,
+        },
+    }
+    if result["fake_uv_called"]:
+        raise RuntimeError("the unrelated uv executable was invoked")
+    return result
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--work-root", type=Path)
+    args = parser.parse_args(argv)
+    if args.work_root is not None:
+        payload = run_probe(args.work_root)
+    else:
+        with tempfile.TemporaryDirectory(prefix="hermes-workflows-installed-") as temp_dir:
+            payload = run_probe(Path(temp_dir) / "probe")
+    print(json.dumps(payload, sort_keys=True, separators=(",", ":")))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())