ouroboros/tests/test_node_runtime_resolver.py

"""Node branch of the process-interpreter resolver (process_interpreters.py).

Ladder contract under test (plan §2.1 + amendments R1-R8):
a healthy PATH node is a byte-identical no-op (argv AND env), a missing or
probe-dead PATH candidate falls back to the bundled runtime (argv rewrite only
for node/nodejs requests; an attested child-env PATH prepend for EVERY
triggered launch, npm-family and ``sh -c`` bodies included), a non-local
executor backend skips the ladder entirely, and no usable runtime is an
honest as-written launch with disclosed probe facts — never a typed pre-block.

Seam placement is itself part of the contract: the node health check is an
EXECUTION probe of an argv[0]-steered candidate, so it runs only AFTER the
dispatch gates and configured Safety have admitted the call —
a planted PATH shim named ``node`` must never execute on a refused call.
"""
from __future__ import annotations

import json
import os
import pathlib
import shlex
import sys
from typing import Any

import pytest

import ouroboros.process_interpreters as resolver
from ouroboros.platform_layer import PATH_SEP, NodeRuntimeHealth
from ouroboros.process_interpreters import (
    InterpreterResolutionTrace,
    apply_env_path_prepend,
    interpreter_path_overlay,
    record_interpreter_resolution,
    resolve_process_node,
)
from ouroboros.tools.registry import ToolContext, ToolRegistry
from ouroboros.tools.shell_guards import interpreter_family

pytestmark = pytest.mark.skipif(
    sys.platform == "win32", reason="POSIX stub executables drive the ladder"
)


@pytest.fixture(autouse=True)
def _isolated_node_health_memo():
    """T18: the probe memo is a module-level registry — reset it around every
    test so no verdict leaks between tests on the same xdist worker."""
    from ouroboros import node_runtime as _nr

    saved = dict(_nr._NODE_HEALTH_MEMO)
    _nr._NODE_HEALTH_MEMO.clear()
    try:
        yield
    finally:
        _nr._NODE_HEALTH_MEMO.clear()
        _nr._NODE_HEALTH_MEMO.update(saved)



def _stub(path: pathlib.Path, body: str) -> pathlib.Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(f"#!/bin/sh\n{body}", encoding="utf-8")
    path.chmod(0o755)
    return path


def _healthy_stub(path: pathlib.Path) -> pathlib.Path:
    return _stub(path, "echo v24.16.0\n")


def _context(tmp_path: pathlib.Path) -> ToolContext:
    repo = tmp_path / "system_repo"
    data = tmp_path / "data"
    repo.mkdir(exist_ok=True)
    data.mkdir(exist_ok=True)
    return ToolContext(
        repo_dir=repo,
        system_repo_dir=repo,
        drive_root=data,
        task_id="node-resolver-test",
    )


def _tool_args(tool_name: str, *, token: str = "node") -> dict:
    if tool_name == "run_command":
        return {"cmd": [token, "--version"]}
    if tool_name == "run_script":
        return {"script": "console.log('ok')", "interpreter": token}
    if tool_name == "start_service":
        return {"name": "svc", "cmd": [token, "server.js"]}
    if tool_name == "verify_and_record":
        return {"contract_kind": "explicit_command", "check": [token, "--version"]}
    raise AssertionError(tool_name)


@pytest.fixture()
def quiet_bootstrap(monkeypatch):
    """Deterministic PATH: the resolver's idempotent bootstrap becomes a no-op
    so monkeypatched PATH is exactly what the ladder probes."""
    monkeypatch.setattr(resolver, "bootstrap_process_path", lambda: [])


@pytest.mark.parametrize(
    "tool_name", ["run_command", "run_script", "start_service", "verify_and_record"]
)
def test_healthy_path_node_is_byte_identical_noop(tmp_path, monkeypatch, quiet_bootstrap, tool_name):
    bin_dir = tmp_path / "bin"
    node = _healthy_stub(bin_dir / "node")
    monkeypatch.setenv("PATH", str(bin_dir))
    ctx = _context(tmp_path)
    args = _tool_args(tool_name)

    resolved, trace = resolve_process_node(ctx, tool_name, args, runtime_mode="advanced")

    assert resolved == args  # argv byte-identical
    assert trace is not None
    assert trace.family == "node"
    assert trace.reason == "path_node_healthy"
    assert not trace.changed
    assert trace.env_path_prepend == ""
    assert trace.runtime_path == str(node)
    assert trace.runtime_version == "24.16.0"
    assert trace.path_snapshot == str(bin_dir)
    # env byte-identical: no overlay, inherit-env stays inherit (None).
    assert interpreter_path_overlay(trace) is None
    assert apply_env_path_prepend(None, trace) is None


@pytest.mark.serial
def test_broken_path_node_falls_back_to_bundled_with_rewrite_and_prepend(
    tmp_path, monkeypatch, quiet_bootstrap,
):
    """A PATH node the kernel kills on launch (the incident class) loses to the
    healthy bundled runtime: argv[0] rewritten, bundled dir attested as prepend."""
    bin_dir = tmp_path / "bin"
    dead = _stub(bin_dir / "node", "kill -9 $$\n")
    bundled = _healthy_stub(tmp_path / "bundle" / "node-standalone" / "bin" / "node")
    monkeypatch.setenv("PATH", str(bin_dir))
    monkeypatch.setattr(resolver, "resolve_bundled_node", lambda: str(bundled))
    ctx = _context(tmp_path)

    resolved, trace = resolve_process_node(
        ctx, "run_command", {"cmd": ["node", "app.js"]}, runtime_mode="advanced"
    )

    assert resolved["cmd"] == [str(bundled), "app.js"]
    assert trace is not None and trace.reason == "bundled_node_fallback"
    assert trace.changed
    assert trace.fallback_reason == f"path_node_broken:signal:SIGKILL:{dead}"
    assert trace.env_path_prepend == str(bundled.parent)
    assert trace.runtime_path == str(bundled)
    assert trace.runtime_version == "24.16.0"
    overlay = interpreter_path_overlay(trace)
    assert overlay == {"PATH": f"{bundled.parent}{PATH_SEP}{bin_dir}"}


def test_missing_path_node_falls_back_to_bundled(tmp_path, monkeypatch, quiet_bootstrap):
    empty = tmp_path / "empty"
    empty.mkdir()
    bundled = _healthy_stub(tmp_path / "bundle" / "node-standalone" / "bin" / "node")
    monkeypatch.setenv("PATH", str(empty))
    monkeypatch.setattr(resolver, "resolve_bundled_node", lambda: str(bundled))
    ctx = _context(tmp_path)

    resolved, trace = resolve_process_node(
        ctx, "run_command", {"cmd": ["node", "--version"]}, runtime_mode="advanced"
    )

    assert resolved["cmd"][0] == str(bundled)
    assert trace is not None and trace.reason == "bundled_node_fallback"
    assert trace.fallback_reason == "path_node_missing:node"


def test_no_usable_node_is_noop_with_disclosed_facts(tmp_path, monkeypatch, quiet_bootstrap):
    empty = tmp_path / "empty"
    empty.mkdir()
    monkeypatch.setenv("PATH", str(empty))
    monkeypatch.setattr(resolver, "resolve_bundled_node", lambda: None)
    ctx = _context(tmp_path)
    args = {"cmd": ["node", "--version"]}

    resolved, trace = resolve_process_node(ctx, "run_command", args, runtime_mode="advanced")

    assert resolved == args  # honest as-written launch, no typed pre-block (R8)
    assert trace is not None and trace.reason == "no_usable_node"
    assert not trace.changed and trace.env_path_prepend == ""
    assert trace.fallback_reason == "path_node_missing:node;bundled_node_missing"
    assert trace.error_reason == ""  # never routed into the python fail-closed branch


@pytest.mark.parametrize(
    ("tool_name", "args"),
    [
        ("run_command", {"cmd": ["/usr/bin/node", "--version"]}),
        ("run_command", {"cmd": ["node20", "--version"]}),
        ("run_command", {"cmd": ["nodemon", "app.js"]}),
        ("run_command", {"cmd": ["sh", "-c", "echo hello"]}),
        ("run_script", {"script": "x", "interpreter": "/opt/node/bin/node"}),
        ("run_script", {"script": "x", "interpreter": "python3"}),
        ("start_service", {"name": "svc", "cmd": ["node18", "server.js"]}),
        ("verify_and_record", {"contract_kind": "artifact_observation", "check": ["node", "-v"]}),
        ("remote_exec", {"cmd": ["node", "-v"]}),
    ],
)
def test_noneligible_invocations_are_byte_for_byte_unchanged(
    tmp_path, monkeypatch, quiet_bootstrap, tool_name, args
):
    """Explicit absolute paths and versioned names are never touched (bug-report
    requirement #5); lookalikes and non-node shells do not trigger the ladder."""
    monkeypatch.setattr(
        resolver, "node_runtime_health",
        lambda *a, **k: (_ for _ in ()).throw(AssertionError("probe must not run")),
    )
    ctx = _context(tmp_path)

    resolved, trace = resolve_process_node(ctx, tool_name, args, runtime_mode="advanced")

    assert resolved == args
    assert trace is None


def test_windows_launcher_suffixes_normalize_for_token_match(
    tmp_path, monkeypatch, quiet_bootstrap,
):
    """On Windows node.exe rewrites like node and NPM.CMD triggers the family
    prepend (R7); on POSIX the same spellings stay unclassified (T9: exec is
    case-sensitive there and launcher suffixes are a Windows convention)."""
    empty = tmp_path / "empty"
    empty.mkdir()
    bundled = _healthy_stub(tmp_path / "bundle" / "node-standalone" / "bin" / "node")
    monkeypatch.setenv("PATH", str(empty))
    monkeypatch.setattr(resolver, "resolve_bundled_node", lambda: str(bundled))
    ctx = _context(tmp_path)

    monkeypatch.setattr(resolver, "IS_WINDOWS", False)
    posix_args = {"cmd": ["node.exe", "app.js"]}
    unchanged, posix_trace = resolve_process_node(
        ctx, "run_command", posix_args, runtime_mode="advanced"
    )
    assert unchanged == posix_args and posix_trace is None

    monkeypatch.setattr(resolver, "IS_WINDOWS", True)

    rewritten, exe_trace = resolve_process_node(
        ctx, "run_command", {"cmd": ["node.exe", "app.js"]}, runtime_mode="advanced"
    )
    assert rewritten["cmd"][0] == str(bundled)
    assert exe_trace is not None and exe_trace.reason == "bundled_node_fallback"

    family_args = {"cmd": ["NPM.CMD", "ci"]}
    unchanged, npm_trace = resolve_process_node(
        ctx, "run_command", family_args, runtime_mode="advanced"
    )
    assert unchanged == family_args  # family tools are never rewritten
    assert npm_trace is not None and npm_trace.env_path_prepend == str(bundled.parent)


def test_npm_family_gets_prepend_without_rewrite(tmp_path, monkeypatch, quiet_bootstrap):
    empty = tmp_path / "empty"
    empty.mkdir()
    bundled = _healthy_stub(tmp_path / "bundle" / "node-standalone" / "bin" / "node")
    monkeypatch.setenv("PATH", str(empty))
    monkeypatch.setattr(resolver, "resolve_bundled_node", lambda: str(bundled))
    ctx = _context(tmp_path)
    args = {"cmd": ["npm", "ci"]}

    resolved, trace = resolve_process_node(ctx, "run_command", args, runtime_mode="advanced")

    assert resolved == args
    assert trace is not None and trace.reason == "bundled_node_fallback"
    assert not trace.changed
    assert trace.env_path_prepend == str(bundled.parent)
    env = apply_env_path_prepend({"PATH": "ignored-base", "HOME": "/h"}, trace)
    # The prepend rebuilds PATH from the resolver's FROZEN snapshot.
    assert env == {"PATH": f"{bundled.parent}{PATH_SEP}{empty}", "HOME": "/h"}


def test_sh_dash_c_body_triggers_prepend_only(tmp_path, monkeypatch, quiet_bootstrap):
    empty = tmp_path / "empty"
    empty.mkdir()
    bundled = _healthy_stub(tmp_path / "bundle" / "node-standalone" / "bin" / "node")
    monkeypatch.setenv("PATH", str(empty))
    monkeypatch.setattr(resolver, "resolve_bundled_node", lambda: str(bundled))
    ctx = _context(tmp_path)
    args = {"cmd": ["sh", "-c", "npm ci && node app.js"]}

    resolved, trace = resolve_process_node(ctx, "run_command", args, runtime_mode="advanced")

    assert resolved == args  # the shell wrapper argv is never rewritten
    assert trace is not None and trace.reason == "bundled_node_fallback"
    assert trace.env_path_prepend == str(bundled.parent)


def test_run_script_shell_body_triggers_prepend(tmp_path, monkeypatch, quiet_bootstrap):
    empty = tmp_path / "empty"
    empty.mkdir()
    bundled = _healthy_stub(tmp_path / "bundle" / "node-standalone" / "bin" / "node")
    monkeypatch.setenv("PATH", str(empty))
    monkeypatch.setattr(resolver, "resolve_bundled_node", lambda: str(bundled))
    ctx = _context(tmp_path)
    args = {"script": "corepack enable\nyarn install\n", "interpreter": "bash"}

    resolved, trace = resolve_process_node(ctx, "run_script", args, runtime_mode="advanced")

    assert resolved == args
    assert trace is not None and trace.env_path_prepend == str(bundled.parent)


def test_string_check_shell_body_triggers_prepend(tmp_path, monkeypatch, quiet_bootstrap):
    """A string verify check normalizes to ["sh","-c",text]; its body is scanned."""
    empty = tmp_path / "empty"
    empty.mkdir()
    bundled = _healthy_stub(tmp_path / "bundle" / "node-standalone" / "bin" / "node")
    monkeypatch.setenv("PATH", str(empty))
    monkeypatch.setattr(resolver, "resolve_bundled_node", lambda: str(bundled))
    ctx = _context(tmp_path)
    args = {"contract_kind": "explicit_command", "check": "npx --yes jest"}

    resolved, trace = resolve_process_node(ctx, "verify_and_record", args, runtime_mode="advanced")

    assert resolved == args
    assert trace is not None and trace.env_path_prepend == str(bundled.parent)


def test_docker_executor_skips_ladder_without_probing(tmp_path, monkeypatch, quiet_bootstrap):
    """R2/Q2-3: a non-local backend resolves node in its own filesystem — no
    host probe runs and no host path can leak into the container argv."""
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    ctx = _context(tmp_path)
    ctx.workspace_root = workspace
    ctx.workspace_mode = "external"
    ctx.executor_ref = {
        "type": "docker_exec",
        "id": "bench",
        "container_name": "bench",
        "network": "none",
        "workspace_host_path": str(workspace),
        "workspace_backend_path": "/workspace",
    }
    monkeypatch.setattr(
        resolver, "node_runtime_health",
        lambda *a, **k: (_ for _ in ()).throw(AssertionError("probe must not run")),
    )
    args = {"cmd": ["node", "app.js"]}

    resolved, trace = resolve_process_node(ctx, "run_command", args, runtime_mode="advanced")

    assert resolved == args
    assert trace is not None and trace.reason == "executor_backend_node"
    assert trace.environment == "backend_path"
    assert trace.env_path_prepend == ""


def test_local_executor_continues_ladder(tmp_path, monkeypatch, quiet_bootstrap):
    """A local executor runs on THIS host, so skipping the ladder would keep the
    broken-PATH bug alive there (amendment R2)."""
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    bin_dir = tmp_path / "bin"
    node = _healthy_stub(bin_dir / "node")
    monkeypatch.setenv("PATH", str(bin_dir))
    ctx = _context(tmp_path)
    ctx.workspace_root = workspace
    ctx.workspace_mode = "external"
    ctx.executor_ref = {
        "type": "local",
        "id": "local",
        "workspace_host_path": str(workspace),
        "workspace_backend_path": "/workspace",
    }

    resolved, trace = resolve_process_node(
        ctx, "run_command", {"cmd": ["node", "app.js"]}, runtime_mode="advanced"
    )

    assert trace is not None and trace.reason == "path_node_healthy"
    assert trace.runtime_path == str(node)


def test_verify_check_args_never_clobbered_by_rewrite(tmp_path, monkeypatch, quiet_bootstrap):
    """R4: the check text is the receipt's identity — the substitution lives in
    the trace (resolved_interpreter) and reaches execution via the attestation."""
    empty = tmp_path / "empty"
    empty.mkdir()
    bundled = _healthy_stub(tmp_path / "bundle" / "node-standalone" / "bin" / "node")
    monkeypatch.setenv("PATH", str(empty))
    monkeypatch.setattr(resolver, "resolve_bundled_node", lambda: str(bundled))
    ctx = _context(tmp_path)
    args = {"contract_kind": "explicit_command", "check": ["node", "--test"]}

    resolved, trace = resolve_process_node(ctx, "verify_and_record", args, runtime_mode="advanced")

    assert resolved == args  # args["check"] untouched
    assert trace is not None and trace.changed
    assert trace.resolved_interpreter == str(bundled)
    assert trace.env_path_prepend == str(bundled.parent)


def test_apply_env_path_prepend_windows_path_key_casing(monkeypatch):
    monkeypatch.setattr(resolver, "IS_WINDOWS", True)
    trace = InterpreterResolutionTrace(
        tool="run_command",
        requested_interpreter="node",
        resolved_interpreter="C:\\bundle\\node.exe",
        surface="system_repo",
        environment="bundled_node",
        reason="bundled_node_fallback",
        family="node",
        path_snapshot="C:\\base",
        env_path_prepend="C:\\bundle",
    )

    env = apply_env_path_prepend({"Path": "stale", "HOME": "h"}, trace)

    assert env is not None
    assert "Path" not in env  # no case-variant duplicate for CreateProcess
    assert env["PATH"] == f"C:\\bundle{PATH_SEP}C:\\base"
    assert env["HOME"] == "h"


def test_recorder_writes_family_specific_event_types(tmp_path):
    ctx = _context(tmp_path)
    base = dict(
        tool="run_command",
        requested_interpreter="node",
        resolved_interpreter="node",
        surface="system_repo",
        environment="host_path",
        reason="path_node_healthy",
    )
    record_interpreter_resolution(
        ctx, InterpreterResolutionTrace(family="node", path_snapshot="/bin", **base)
    )
    record_interpreter_resolution(
        ctx,
        InterpreterResolutionTrace(
            **{**base, "requested_interpreter": "python", "resolved_interpreter": "python",
               "environment": "ouroboros_agent", "reason": "agent_python"},
        ),
    )

    lines = (ctx.drive_logs() / "events.jsonl").read_text(encoding="utf-8").splitlines()
    node_event, python_event = (json.loads(line) for line in lines[-2:])
    assert node_event["type"] == "node_runtime_resolution"
    assert node_event["family"] == "node"
    assert node_event["path_snapshot"] == "/bin"
    assert python_event["type"] == "python_interpreter_resolution"
    # The historic python event payload gains no generalization keys.
    assert "family" not in python_event
    assert "path_snapshot" not in python_event
    assert "env_path_prepend" not in python_event


# ---- registry seam: gates first, then the probe; guard/handler family parity ----


def _advanced_registry(tmp_path, monkeypatch) -> tuple[ToolRegistry, ToolContext]:
    ctx = _context(tmp_path)
    monkeypatch.setenv("OUROBOROS_RUNTIME_MODE", "advanced")
    monkeypatch.setattr("ouroboros.safety.check_safety", lambda *a, **k: (True, ""))
    registry = ToolRegistry(repo_dir=ctx.repo_dir, drive_root=ctx.drive_root)
    registry.set_context(ctx)
    # Campaign owner: the safety check is a registry_guard_process module
    # function whose no-block value is None (upstream's method returned "").
    monkeypatch.setattr(
        "ouroboros.tools.registry_guard_process._run_shell_safety_check",
        lambda *a, **k: None,
    )
    return registry, ctx


def _mock_health(monkeypatch, verdicts: dict[str, NodeRuntimeHealth]) -> None:
    def fake_health(path: str, timeout_sec: float = 10) -> NodeRuntimeHealth:
        return verdicts[str(path)]

    monkeypatch.setattr(resolver, "node_runtime_health", fake_health)


def test_registry_guard_sees_family_stable_argv_and_handler_gets_rewrite(
    tmp_path, monkeypatch, quiet_bootstrap,
):
    """The guard inspects the ORIGINAL bare argv (the node step runs post-gates);
    the handler executes the resolver's substitution, which classifies into the
    SAME interpreter family — the disclosed guard/handler delta contract."""
    # Campaign call site: registry_core reads the guard-args builder through
    # the shell_guards module (upstream read it off the registry facade).
    import ouroboros.tools.shell_guards as registry_module

    bin_dir = tmp_path / "bin"
    dead = _stub(bin_dir / "node", "exit 1\n")
    bundled = _healthy_stub(tmp_path / "bundle" / "node-standalone" / "bin" / "node")
    monkeypatch.setenv("PATH", str(bin_dir))
    monkeypatch.setattr(resolver, "resolve_bundled_node", lambda: str(bundled))
    _mock_health(monkeypatch, {
        str(dead): NodeRuntimeHealth(status="broken", reason="signal:SIGKILL", path=str(dead)),
        str(bundled): NodeRuntimeHealth(status="healthy", version="24.16.0", path=str(bundled)),
    })
    registry, ctx = _advanced_registry(tmp_path, monkeypatch)
    captured: dict[str, list[str]] = {}
    original_guard = registry_module.process_shell_guard_args

    def capture_guard(name, args, **kwargs):
        guarded = original_guard(name, args, **kwargs)
        captured["guard"] = list(guarded["cmd"])
        return guarded

    def capture_handler(_ctx, cmd, _resolved_binding=None, **_kwargs):
        captured["handler"] = list(cmd)
        captured["attested"] = getattr(_ctx, "_active_interpreter_resolution", None)
        return "ok"

    monkeypatch.setattr(registry_module, "process_shell_guard_args", capture_guard)
    import dataclasses
    registry._entries["run_command"] = dataclasses.replace(
        registry._entries["run_command"], handler=capture_handler,
    )

    result = registry.execute("run_command", {"cmd": ["node", "--version"]})

    assert result == "ok"
    assert captured["guard"] == ["node", "--version"]
    assert captured["handler"] == [str(bundled), "--version"]
    assert interpreter_family(captured["guard"][0]) == "node"
    assert interpreter_family(captured["handler"][0]) == "node"
    attested = captured["attested"]
    assert isinstance(attested, InterpreterResolutionTrace)
    assert attested.family == "node" and attested.reason == "bundled_node_fallback"
    assert not hasattr(ctx, "_active_interpreter_resolution")
    event = json.loads(
        (ctx.drive_logs() / "events.jsonl").read_text(encoding="utf-8").splitlines()[-1]
    )
    assert event["type"] == "node_runtime_resolution"
    assert event["env_path_prepend"] == str(bundled.parent)


def test_registry_healthy_path_handler_argv_is_untouched(tmp_path, monkeypatch, quiet_bootstrap):
    bin_dir = tmp_path / "bin"
    node = _healthy_stub(bin_dir / "node")
    monkeypatch.setenv("PATH", str(bin_dir))
    _mock_health(monkeypatch, {
        str(node): NodeRuntimeHealth(status="healthy", version="24.16.0", path=str(node)),
    })
    registry, ctx = _advanced_registry(tmp_path, monkeypatch)
    observed: dict[str, Any] = {}

    def handler(_ctx, cmd, _resolved_binding=None, **_kwargs):
        observed["cmd"] = list(cmd)
        observed["attested"] = getattr(_ctx, "_active_interpreter_resolution", None)
        return "ok"

    import dataclasses
    registry._entries["run_command"] = dataclasses.replace(
        registry._entries["run_command"], handler=handler,
    )

    assert registry.execute("run_command", {"cmd": ["node", "--version"]}) == "ok"
    assert observed["cmd"] == ["node", "--version"]
    attested = observed["attested"]
    assert attested is not None and attested.reason == "path_node_healthy"
    assert attested.env_path_prepend == ""


@pytest.mark.serial
def test_light_shell_request_reaches_node_probe_and_executes_once(tmp_path, monkeypatch):
    """A guessed write in shell text does not veto the chosen Safety decision."""
    monkeypatch.setenv("OUROBOROS_RUNTIME_MODE", "light")
    monkeypatch.setattr("ouroboros.safety.check_safety", lambda *a, **k: (True, ""))
    repo = tmp_path / "repo"
    repo.mkdir()
    registry = ToolRegistry(repo_dir=repo, drive_root=tmp_path / "drive")
    registry._ctx.task_id = "t-node-order"
    marker = tmp_path / "shim_executed"
    source = repo / "x.py"
    bin_dir = tmp_path / "bin"
    marker_arg = shlex.quote(str(marker))
    _stub(bin_dir / "node", (
        'if [ "$1" = "--version" ]; then\n'
        f"  printf 'probe\\n' >> {marker_arg}\n"
        "  printf 'v24.16.0\\n'\n"
        "else\n"
        f"  printf 'execute\\n' >> {marker_arg}\n"
        f"  printf x > {shlex.quote(str(source))}\n"
        "fi\n"
    ))
    monkeypatch.setenv("PATH", f"{bin_dir}{os.pathsep}{os.environ.get('PATH', '')}")

    result = registry.execute("run_command", {
        "cmd": ["node", f"--eval=require('node:fs').writeFileSync('{repo}/x.py','x')"],
    })

    assert "exit_code=0" in result, result
    assert marker.read_text(encoding="utf-8").splitlines() == ["probe", "execute"]
    assert source.read_bytes() == b"x"


# ---- handler env application ----


def test_run_shell_applies_attested_prepend_and_healthy_env_is_untouched(
    tmp_path, monkeypatch,
):
    import ouroboros.tools.shell as shell

    ctx = _context(tmp_path)
    seen: dict[str, Any] = {}

    def fake_run(cmd, **kwargs):
        seen["cmd"] = list(cmd)
        seen["env"] = kwargs.get("env")

        class _Res:
            returncode = 0
            stdout = "ok"
            stderr = ""
            args = cmd

        return _Res()

    monkeypatch.setattr(shell, "_tracked_subprocess_run", fake_run)
    prepend_dir = str(tmp_path / "bundle" / "bin")
    ctx._active_interpreter_resolution = InterpreterResolutionTrace(
        tool="run_command",
        requested_interpreter="npm",
        resolved_interpreter="npm",
        surface="system_repo",
        environment="bundled_node",
        reason="bundled_node_fallback",
        family="node",
        path_snapshot="/frozen",
        env_path_prepend=prepend_dir,
    )
    result = shell._run_shell(ctx, ["npm", "--version"], cwd="system_repo")
    assert "exit_code=0" in result
    assert seen["env"] is not None
    assert seen["env"]["PATH"] == f"{prepend_dir}{PATH_SEP}/frozen"

    del ctx._active_interpreter_resolution
    seen.clear()
    expected_env = dict(os.environ)
    result = shell._run_shell(ctx, ["npm", "--version"], cwd="system_repo")
    assert "exit_code=0" in result
    assert seen["env"] == expected_env  # explicit task env; no Node PATH rewrite


def test_verify_executes_resolved_argv_but_receipt_keeps_original_check(
    tmp_path, monkeypatch,
):
    from ouroboros.outcomes import verification_receipts_path
    from ouroboros.tools.verify import _verify_and_record

    ctx = _context(tmp_path)
    bundled = str(tmp_path / "bundle" / "node-standalone" / "bin" / "node")
    seen: dict[str, Any] = {}

    def fake_run(cmd, **kwargs):
        seen["cmd"] = list(cmd)
        seen["env"] = kwargs.get("env")

        class _Res:
            returncode = 0
            stdout = "v24.16.0"
            stderr = ""
            args = cmd

        return _Res()

    import ouroboros.tools.shell as shell

    monkeypatch.setattr(shell, "_tracked_subprocess_run", fake_run)
    ctx._active_interpreter_resolution = InterpreterResolutionTrace(
        tool="verify_and_record",
        requested_interpreter="node",
        resolved_interpreter=bundled,
        surface="system_repo",
        environment="bundled_node",
        reason="bundled_node_fallback",
        family="node",
        path_snapshot="/frozen",
        env_path_prepend=str(pathlib.Path(bundled).parent),
    )

    result = _verify_and_record(
        ctx,
        contract_kind="explicit_command",
        check=["node", "--version"],
        expected="v24",
    )

    assert "PASS" in result
    assert seen["cmd"] == [bundled, "--version"]
    assert seen["env"]["PATH"].startswith(f"{pathlib.Path(bundled).parent}{PATH_SEP}")
    receipts = verification_receipts_path(ctx.drive_root, "node-resolver-test")
    receipt = json.loads(receipts.read_text(encoding="utf-8").splitlines()[-1])
    # R4: the receipt's identity is the ORIGINAL check text, not the rewrite.
    assert receipt["check"] == "node --version"


def test_run_script_accepts_attested_bundled_node_and_allows_node_exe(
    tmp_path, monkeypatch,
):
    """Resolved and explicitly selected runtime paths share one launch surface."""
    import ouroboros.tools.shell as shell

    ctx = _context(tmp_path)
    monkeypatch.setattr(shell, "_run_shell", lambda *_a, **_k: "ok")
    bundled = str(_healthy_stub(tmp_path / "bundle" / "node-standalone" / "bin" / "node"))
    ctx._active_interpreter_resolution = InterpreterResolutionTrace(
        tool="run_script",
        requested_interpreter="node",
        resolved_interpreter=bundled,
        surface="system_repo",
        environment="bundled_node",
        reason="bundled_node_fallback",
        family="node",
        env_path_prepend=str(pathlib.Path(bundled).parent),
    )
    attested = shell._run_script(ctx, "console.log(1)", interpreter=bundled)
    assert "RUN_SCRIPT_BLOCKED" not in attested

    del ctx._active_interpreter_resolution
    plain = shell._run_script(ctx, "console.log(1)", interpreter="node.exe")
    assert "RUN_SCRIPT_BLOCKED" not in plain

    odd_basename = str(tmp_path / "opt" / "bundle" / "node24")
    explicit = shell._run_script(ctx, "console.log(1)", interpreter=odd_basename)
    assert "RUN_SCRIPT_BLOCKED" not in explicit
    ctx._active_interpreter_resolution = InterpreterResolutionTrace(
        tool="run_script",
        requested_interpreter="node",
        resolved_interpreter=odd_basename,
        surface="system_repo",
        environment="bundled_node",
        reason="bundled_node_fallback",
        family="node",
    )
    admitted = shell._run_script(ctx, "console.log(1)", interpreter=odd_basename)
    assert "RUN_SCRIPT_BLOCKED" not in admitted


@pytest.mark.serial
def test_workspace_executor_local_applies_env_overlay(tmp_path):
    from ouroboros.workspace_executor import ExecutorRef, PathMapping, _execute_local

    executor = ExecutorRef(
        kind="local",
        executor_id="t",
        network="host",
        mappings=(PathMapping(host_path=tmp_path, backend_path="/workspace"),),
    )
    argv = [
        sys.executable,
        "-c",
        "import os; print(os.environ.get('OURO_NODE_TEST', 'missing'))",
    ]

    plain = _execute_local(executor, argv, tmp_path, 30, drive_root=None)
    overlaid = _execute_local(
        executor, argv, tmp_path, 30, drive_root=None,
        env_overlay={"OURO_NODE_TEST": "prepended"},
    )

    assert plain.returncode == 0 and plain.stdout.strip() == "missing"
    assert overlaid.returncode == 0 and overlaid.stdout.strip() == "prepended"


# ---- rename hygiene ----


def test_no_stale_python_interpreter_module_references():
    """The module moved to process_interpreters.py with no compatibility shim;
    a stale import would crash at runtime on the next release."""
    import importlib.util

    # Built dynamically so this test's own source never matches its scan.
    stale_module = "python" + "_interpreter"
    stale_dotted = f"ouroboros.{stale_module}"
    repo_root = pathlib.Path(__file__).resolve().parent.parent
    assert importlib.util.find_spec("ouroboros.process_interpreters") is not None
    assert not (repo_root / "ouroboros" / f"{stale_module}.py").exists()
    # An editable install registers a meta-path finder for its OWN checkout, so
    # in a dev environment `find_spec` can resurrect the old name from a
    # different tree; only a spec originating in THIS tree is a rename failure.
    stale = importlib.util.find_spec(stale_dotted)
    assert stale is None or not str(stale.origin or "").startswith(str(repo_root))
    offenders: list[str] = []
    for base in ("ouroboros", "supervisor", "tests"):
        for path in sorted((repo_root / base).rglob("*.py")):
            text = path.read_text(encoding="utf-8", errors="replace")
            if stale_dotted in text or f"from ouroboros import {stale_module}" in text:
                offenders.append(str(path.relative_to(repo_root)))
    server = repo_root / "server.py"
    if server.is_file() and stale_dotted in server.read_text(encoding="utf-8", errors="replace"):
        offenders.append("server.py")
    assert offenders == []


def test_registry_bridges_resolved_runtime_slot_for_observability(tmp_path):
    """Synthesis pin (streams A+B): when the node trace records a substitution
    (argv rewrite or emergency prepend), `_invoke_builtin_handler` publishes the
    ONE string slot ``ctx._process_resolved_runtime`` for the duration of the
    handler call — the slot the typed process facts and the verify receipt
    disclose — and restores it afterwards; a no-op trace publishes nothing."""
    from dataclasses import replace
    from types import SimpleNamespace

    from ouroboros.process_interpreters import InterpreterResolutionTrace
    from ouroboros.tools.registry import ToolRegistry
    from ouroboros.tools.shell import get_tools

    registry = ToolRegistry.__new__(ToolRegistry)
    registry._ctx = SimpleNamespace(repo_dir=tmp_path, drive_root=tmp_path / "data")
    seen = {}

    def handler(ctx, **kwargs):
        seen["slot"] = getattr(ctx, "_process_resolved_runtime", None)
        return "ok"

    entry = replace(next(tool for tool in get_tools() if tool.name == "run_command"), handler=handler)
    changed = InterpreterResolutionTrace(
        tool="run_command", requested_interpreter="node",
        resolved_interpreter="/bundle/bin/node", surface="external_workspace",
        environment="bundled_node", reason="bundled_node_fallback", family="node",
    )
    err, result = registry._invoke_builtin_handler(
        "run_command", entry, {}, None, changed, None)
    assert err is None and result == "ok"
    assert seen["slot"] == "/bundle/bin/node"
    assert not hasattr(registry._ctx, "_process_resolved_runtime")

    healthy = InterpreterResolutionTrace(
        tool="run_command", requested_interpreter="node",
        resolved_interpreter="node", surface="external_workspace",
        environment="target_path", reason="path_node_healthy", family="node",
    )
    err, result = registry._invoke_builtin_handler(
        "run_command", entry, {}, None, healthy, None)
    assert err is None and seen["slot"] is None
    assert not hasattr(registry._ctx, "_process_resolved_runtime")


def test_run_script_schema_exposes_an_executable_without_a_language_allowlist():
    from ouroboros.tools import shell as shell_mod

    entry = next(e for e in shell_mod.get_tools() if e.name == "run_script")
    parameter = entry.schema["parameters"]["properties"]["interpreter"]
    assert parameter["type"] == "string" and parameter["default"] == "python3"
    assert "enum" not in parameter
    assert "script filename" in parameter["description"]


def test_whitespace_padded_head_is_not_classified(tmp_path, monkeypatch, quiet_bootstrap):
    """T8 pin: ' node ' must NOT produce a node trace — a padded head is run
    as written, so no attestation may claim a substituted runtime for it."""
    empty = tmp_path / "empty"
    empty.mkdir()
    bundled = _healthy_stub(tmp_path / "bundle" / "node-standalone" / "bin" / "node")
    monkeypatch.setenv("PATH", str(empty))
    monkeypatch.setattr(resolver, "resolve_bundled_node", lambda: str(bundled))
    ctx = _context(tmp_path)
    for tool, args in (
        ("run_command", {"cmd": [" node ", "--version"]}),
        ("start_service", {"cmd": [" node ", "server.js"]}),
        # explicit_command IS in _VERIFY_RUN_KINDS — the padded head must be
        # the ONLY reason this stays unclassified (delta finding D2-3).
        ("verify_and_record", {"contract_kind": "explicit_command", "check": [" node ", "--version"]}),
    ):
        resolved, trace = resolve_process_node(ctx, tool, args, runtime_mode="advanced")
        assert resolved == args and trace is None, tool
    # Positive control: the same verify kind with an UNPADDED head classifies,
    # proving the run-kind gate above is actually open for these cases.
    control = {"contract_kind": "explicit_command", "check": ["node", "--version"]}
    _resolved, control_trace = resolve_process_node(
        ctx, "verify_and_record", control, runtime_mode="advanced"
    )
    assert control_trace is not None


def test_relative_path_which_result_is_a_noop(tmp_path, monkeypatch, quiet_bootstrap):
    """T10 pin: if which() still returns a relative candidate after binding
    PATH to the launch cwd, keep the unprovable launch as written."""
    bundled = _healthy_stub(tmp_path / "bundle" / "node-standalone" / "bin" / "node")
    monkeypatch.setattr(resolver, "resolve_bundled_node", lambda: str(bundled))
    monkeypatch.setattr(resolver.shutil, "which", lambda tok, path=None: "bin/node")
    ctx = _context(tmp_path)
    args = {"cmd": ["node", "app.js"]}
    resolved, trace = resolve_process_node(ctx, "run_command", args, runtime_mode="advanced")
    assert resolved == args
    assert trace is not None
    assert trace.reason == "path_node_relative_entry_unprovable"
    assert trace.env_path_prepend in ("", None)


def test_wrapper_matching_covers_abs_paths_and_zsh(tmp_path, monkeypatch, quiet_bootstrap):
    """F-1 pin: /bin/sh -c and zsh -c bodies naming node-family tools trigger
    the family prepend exactly like bare sh; a wrapper hit only rides the env
    prepend (argv untouched)."""
    empty = tmp_path / "empty"
    empty.mkdir()
    bundled = _healthy_stub(tmp_path / "bundle" / "node-standalone" / "bin" / "node")
    monkeypatch.setenv("PATH", str(empty))
    monkeypatch.setattr(resolver, "resolve_bundled_node", lambda: str(bundled))
    ctx = _context(tmp_path)
    for head in ("/bin/sh", "zsh", "dash"):
        args = {"cmd": [head, "-c", "npm ci"]}
        resolved, trace = resolve_process_node(ctx, "run_command", args, runtime_mode="advanced")
        assert resolved == args, head
        assert trace is not None and trace.env_path_prepend == str(bundled.parent), head
    # A non-wrapper absolute head with a node body stays unclassified.
    plain = {"cmd": ["/usr/bin/env", "node", "app.js"]}
    _r, none_trace = resolve_process_node(ctx, "run_command", plain, runtime_mode="advanced")
    assert none_trace is None