_session_drive_support.py

"""Driven-session spine (H0): the daemon DRIVES a local-LLM turn under the authoritative ledger.

OFFLINE tests drive a REAL in-process Supervisor pinned to a fresh scratch KAIZEN_REPO_ROOT in a
SUBPROCESS -- the suite idiom for touching the isolated DB in-process (import-frozen paths.REPO_ROOT
plus deterministic Windows handle release; cf. test_supervisor / test_remote_dispatch /
test_policy._run_policy_snippet). The scenario body runs in the subprocess with the daemon's
_adapter_factory seam injected (mirroring _dispatch_runner): the factory returns a LocalLLMAdapter
over a HERMETIC scripted provider, so the whole loop (prompt -> parse -> decide() -> executor -> T6
funnel) runs with no network, no model, no child process. A fake clock/sleep makes the session/events
long-poll deterministic. Every scenario asserts on a RESULT json line the subprocess prints.

Coverage (the H0 exit matrix):
- start -> events -> final .................... StartEventsFinalTest
- ask -> approve-by-correlation_id -> tool ... ApproveByCorrelationTest
- ask -> deny ................................ DenyTest
- ask timeout fail-closed ................... AskTimeoutTest
- steer mid-turn ............................ SteerTest
- interrupt ................................. InterruptTest
- kill (waiters denied, adapter dead) ...... KillTest
- cursor pagination gapless ................ CursorPaginationTest
- long-poll mid-poll delivery .............. LongPollDeliveryTest
- long-poll timeout returns empty .......... LongPollTimeoutTest
- codex/claude capability-denied, unknown UNKNOWN,
  with NO C1/T5 rows written ............... EngineGateTest
- bad loopback token refused ............... LoopbackTokenTest
- shutdown-with-open-turn .................. ShutdownOpenTurnTest
- double-approve ALREADY_DECIDED ........... DoubleApproveTest
- capabilities shape (3 wire engines) ...... CapabilitiesShapeTest
- capabilities degraded probe (no Ollama) .. CapabilitiesDegradedTest
- start w/ explicit profile persists all
  C1/T5 fields + profile/point FIRST ....... ProfileStartTest
- Full without opt-in DENIED, zero rows .... FullOptInTest
- reasoning_effort on local_llm DENIED ..... ProfileUnsupportedTest
- unknown profile field DENIED, zero rows .. ProfileUnknownFieldTest
- legacy model vs profile.model conflict ... ModelConflictTest
- claude_cli alias -> claude on the wire ... EngineAliasTest
- profile_hash differs plan vs ask ......... ProfileHashDiffersTest

Plus a gated live-smoke class (KAIZEN_RUN_LIVE=1 + KAIZEN_LLM_MODEL/KAIZEN_LLM_BASE_URL): a real Ollama
driven turn incl. an approval round-trip, asserting production-shaped C1/C4/T5/T6 rows in an isolated DB.
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock

sys.path.insert(0, str(Path(__file__).resolve().parent))
from tests._harness import KAIZEN, REPO_ROOT, kaizen, run  # noqa: E402

_LIVE = os.environ.get("KAIZEN_RUN_LIVE") == "1"

# --- subprocess driver harness (scratch plane; the daemon runs in-process in the child) ----------

# The child preamble defines: a scripted chat provider (canned reply list), a driven _adapter_factory
# that wires it into a LocalLLMAdapter with the daemon-supplied recorder (so events funnel to T6), a
# fake clock/sleep pair, and helpers to poll events. BODY is the scenario; it must set ``out``.
_PREAMBLE = r"""
import json, os, sys, time
from types import SimpleNamespace
from kaizen_components.orchestration.supervisor import Supervisor, _LOCAL_LLM_CODE_FEATURE_EVIDENCE
from kaizen_components.orchestration.adapters import local_llm as L
from kaizen_components.orchestration import policy

os.environ.setdefault("KAIZEN_LLM_MODEL", "fixture-model")

def scripted_provider(replies):
    # Returns canned dicts/strings in order (repeating the last), recording each call's message list.
    state = {"i": 0, "calls": []}
    def provider(messages, **opts):
        state["calls"].append([dict(m) for m in messages])
        idx = min(state["i"], len(replies) - 1)
        state["i"] += 1
        reply = replies[idx]
        return {"text": reply} if isinstance(reply, str) else dict(reply)
    provider.state = state
    return provider

def deterministic_ids():
    counters = {}
    def factory(prefix):
        counters[prefix] = counters.get(prefix, 0) + 1
        return prefix + "-" + str(counters[prefix])
    return factory

def tool_reply(name, **args):
    return json.dumps({"tool": name, "args": args})

def final_reply(answer):
    return json.dumps({"final": answer})

def allow_rule(verb, prefix, rid):
    return {"id": rid, "rule_type": "allow", "verb": verb, "match_kind": "path_prefix",
            "pattern": prefix, "engine": None, "enabled": True}

def make_engine(rules=(), protected=()):
    # vendor=[] so no real ~/.claude/~/.codex resolution in a unit test.
    return policy.PolicyEngine(list(protected), list(rules), [])

def echo_tools():
    # A single ALLOW-scoped no-op tool so a tool intent reaches decide() and runs.
    def _run(args):
        return "ran echo " + str(args.get("path", ""))
    return {"echo": L.ToolSpec("echo", "file_read", "echo a path", _run, arg_hints=("path",))}

def install_local_fixture_capability(sup, models=None, *, generic_models=None):
    model_ids = list(models or [os.environ.get("KAIZEN_LLM_MODEL") or "fixture-model"])
    rows = [{"id": model_id, "label": model_id, "reasoning_efforts": [], "default_effort": None,
             "capabilities": ["completion", "thinking", "tools", "vision"]} for model_id in model_ids]
    binding = sup._local_platform_binding("ollama")
    platform = {"id": "ollama", "label": "Ollama", "protocol": "ollama",
                "availability": {"state": "available", "code": None, "message": ""},
                "models": rows, "default_model": None,
                "_endpoint_config_hash": binding.endpoint_config_hash}
    platforms = [platform]
    if generic_models is not None:
        generic_rows = [{"id": model_id, "label": model_id, "reasoning_efforts": [], "default_effort": None,
                         "capabilities": ["completion"]}
                        for model_id in generic_models]
        generic_binding = sup._local_platform_binding("openai_compatible")
        platforms.append({"id": "openai_compatible", "label": "OpenAI-compatible",
                          "protocol": "openai_compatible",
                          "availability": {"state": "available", "code": None, "message": ""},
                          "models": generic_rows, "default_model": None,
                          "_endpoint_config_hash": generic_binding.endpoint_config_hash})
    local = {"id": "local_llm", "label": "Local Model", "drivable": True,
             "availability": dict(platform["availability"]), "platforms": platforms,
             "default_platform": "ollama", "models": list(rows), "default_model": None,
             "default_reasoning_effort": None, "auth_modes": ["none"],
             "permission_modes": ["plan", "ask", "agent", "full"], "warnings": [],
             "_code_proven_features": dict(_LOCAL_LLM_CODE_FEATURE_EVIDENCE)}
    local = sup._materialize_capabilities([local])[0]
    with sup._capabilities_lock:
        vendors = [item for item in (sup._capabilities or []) if item.get("id") != "local_llm"]
        if not vendors:
            vendors = [sup._vendor_pending_capability("codex"), sup._vendor_pending_capability("claude")]
        sup._capabilities = [local, *vendors]
        sup._capabilities_built_at = time.monotonic()

def fixture_local_capability(self):
    model_id = os.environ.get("KAIZEN_LLM_MODEL") or "fixture-model"
    binding = self._local_platform_binding("ollama")
    available = os.environ.get("KAIZEN_TEST_LOCAL_CATALOG_FAILURE") != "1"
    models = [{"id": model_id, "label": model_id, "reasoning_efforts": [], "default_effort": None,
               "capabilities": ["completion", "thinking", "tools", "vision"]}] if available else []
    availability = {"state": "available", "code": None, "message": ""} if available else {
        "state": "unavailable", "code": "DENIED_PLATFORM_UNAVAILABLE",
        "message": "Model catalog is unavailable; check the daemon-owned platform configuration and refresh capabilities.",
    }
    platform = {"id": "ollama", "label": "Ollama", "protocol": "ollama",
                "availability": availability, "models": models, "default_model": None,
                "_endpoint_config_hash": binding.endpoint_config_hash}
    return {"id": "local_llm", "label": "Local Model", "drivable": True,
            "availability": dict(availability), "platforms": [platform], "default_platform": "ollama",
            "models": list(models), "default_model": None, "default_reasoning_effort": None,
            "auth_modes": ["none"], "permission_modes": ["plan", "ask", "agent", "full"],
            "warnings": [] if available else ["Ollama model catalog is not currently available."],
            "_code_proven_features": dict(_LOCAL_LLM_CODE_FEATURE_EVIDENCE)}

Supervisor._local_llm_capability = fixture_local_capability

_production_boot = Supervisor.boot

def fixture_boot(self):
    result = _production_boot(self)
    self._handle_session_capabilities({})
    return result

Supervisor.boot = fixture_boot

def install_factory(sup, provider, *, engine=None, tools=None, timeout_holder=None, catalog_models=None,
                    generic_models=None):
    '''Install a scripted adapter factory while preserving the supervisor's driven-turn options.'''
    # The _adapter_factory seam: build a LocalLLMAdapter over the scripted provider with the daemon's
    # recorder (so events funnel to T6). Honors the daemon-passed kwargs (engine_name/model/
    # approval_timeout/max_turns). A fresh id_factory per adapter keeps ids deterministic per run.
    eng = engine if engine is not None else make_engine()
    install_local_fixture_capability(sup, catalog_models, generic_models=generic_models)
    def factory(agent_run_id, recorder, kwargs):
        adapter = L.LocalLLMAdapter(
            eng, chat_provider=provider, tools=(tools if tools is not None else {}),
            recorder=recorder, logger=(lambda _m: None), id_factory=deterministic_ids(), **kwargs,
        )
        return adapter
    sup._adapter_factory = factory

class FakeClock:
    # A monotonic fake: clock() advances only when sleep() is called (deterministic long-poll).
    def __init__(self):
        self.t = 0.0
    def clock(self):
        return self.t
    def sleep(self, dt):
        self.t += dt

def wait_idle(sup, run_id, budget=8.0):
    # Block until the current turn completes without implicitly finalizing the conversation.
    deadline = time.monotonic() + budget
    while time.monotonic() < deadline:
        sess = sup._get_driven(run_id)
        if sess is not None and sess.turn_state == "idle":
            return sess
        state = sup._safe_reduce(run_id)
        if state is not None and state["terminal"]:
            return None
        time.sleep(0.02)
    return sup._get_driven(run_id)

def wait_terminal(sup, run_id, budget=8.0):
    # Compatibility helper for pre-H2 scenarios: wait for idle, then EXPLICITLY close. Fatal/kill paths
    # may terminalize first. H2-specific tests below use wait_idle directly to prove T8 stays absent.
    deadline = time.monotonic() + budget
    while time.monotonic() < deadline:
        state = sup._safe_reduce(run_id)
        if state is not None and state["terminal"]:
            return state
        sess = sup._get_driven(run_id)
        if sess is not None and sess.turn_state == "idle":
            sup._handle_control({"op": "session/close", "args": {"agent_run_id": run_id}})
            continue
        time.sleep(0.02)
    return sup._safe_reduce(run_id)

def wait_open_approval(sup, run_id, budget=8.0):
    # Block until (a) an approval/open event lands on the run's stream AND (b) the C4 approval row for
    # that correlation is persisted (record_ask runs AFTER the event emit -- local_llm.py:687 vs :693 --
    # so an approve keyed by the event's correlation would race the row). Return the correlation_id (the
    # race-free approve handle the webview uses). None if none appears in budget.
    from kaizen_components import db
    deadline = time.monotonic() + budget
    while time.monotonic() < deadline:
        rows = db.fetch_all(
            "SELECT correlation_id FROM agent_events WHERE agent_run_id = ? AND event_kind = 'approval' "
            "AND marker = 'open' ORDER BY sequence_no LIMIT 1", (run_id,))
        if rows and rows[0][0]:
            corr = rows[0][0]
            # Confirm the C4 row exists (open OR decided) so the alt-key approve resolves deterministically.
            c4 = db.fetch_all(
                "SELECT a.id FROM approval_requests a JOIN agent_runs r ON r.session_id = a.session_id "
                "WHERE r.id = ? AND a.correlation_id = ? LIMIT 1", (run_id, corr))
            if c4:
                return corr
        time.sleep(0.02)
    return None

def start_sse_server(script):
    # F5: an in-process real-socket SSE endpoint; `script(handler)` writes the streamed body. Returns
    # (server, base_url); the caller shuts it down. Real sockets are required because the silent-
    # transport cancellation cadence shortens the live response socket's timeout.
    import http.server
    import threading as _threading

    class Handler(http.server.BaseHTTPRequestHandler):
        protocol_version = "HTTP/1.0"

        def do_POST(self):
            self.rfile.read(int(self.headers.get("Content-Length") or 0))
            try:
                script(self)
            except (ConnectionError, OSError):
                pass  # the interrupted client closed the socket mid-write

        def log_message(self, *_args):
            pass

    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    server.daemon_threads = True
    _threading.Thread(target=server.serve_forever, daemon=True).start()
    return server, "http://127.0.0.1:%d/v1" % server.server_address[1]

def sse_headers(handler):
    handler.send_response(200)
    handler.send_header("Content-Type", "text/event-stream")
    handler.end_headers()
    handler.wfile.flush()

def sse_frame(text=None, reasoning=None):
    delta = {"content": text} if text is not None else {"reasoning": reasoning}
    return ("data: " + json.dumps({"model": os.environ.get("KAIZEN_LLM_MODEL") or "fixture-model",
                                   "choices": [{"delta": delta}]}) + "\n").encode("utf-8")

def sse_provider(base_url, timeout=30.0):
    # A REAL OpenAICompatClient-backed streaming provider (the default_chat_provider closure shape),
    # so a driven turn exercises the exact transport chain incl. the per-frame interrupt checkpoint.
    from kaizen_components.backends.openai_compat import OpenAICompatClient
    model = os.environ.get("KAIZEN_LLM_MODEL") or "fixture-model"
    client = OpenAICompatClient(base_url, timeout=timeout)

    def provider(messages, **opts):
        return client.chat(messages, model, **opts)

    def stream(messages, on_delta, **opts):
        return client.chat_stream(messages, model, on_delta, **opts)

    provider.stream_chat = stream
    return provider

def assistant_chat_messages(run_id):
    # Durable assistant chat_message bodies for the run (partial-persistence proof, ARC-033).
    from kaizen_components import db
    rows = db.fetch_all(
        "SELECT body FROM agent_events WHERE agent_run_id = ? AND event_kind = 'chat_message' "
        "ORDER BY sequence_no", (run_id,))
    bodies = [json.loads(row[0]) for row in rows]
    return [body for body in bodies if body.get("role") == "assistant"]

def wait_turn_done(sup, run_id, budget=10.0):
    # Block until the current turn publishes its result; returns elapsed seconds (latency probe).
    started = time.monotonic()
    sess = sup._get_driven(run_id)
    if sess is None or not sess.turn_done.wait(budget):
        return None
    return time.monotonic() - started

def wait_waiter_parked(sup, run_id, correlation, budget=8.0):
    # Block until the driven session has PARKED the approval waiter for `correlation` (the adapter's
    # on_approval resolver thread runs slightly after the approval/open event, so kill/approve must wait
    # for the parked waiter to observe it deterministically). Returns True once parked.
    deadline = time.monotonic() + budget
    while time.monotonic() < deadline:
        sess = sup._get_driven(run_id)
        if sess is not None:
            with sess.lock:
                if correlation in sess.waiters:
                    return True
        time.sleep(0.02)
    return False

out = None
exec(BODY)
print("RESULT " + json.dumps(out))
"""

class _DrivenSubprocess(unittest.TestCase):
    """A fresh scratch KAIZEN_REPO_ROOT, K1-initialized; ``drive(body)`` runs the scenario in a child
    process pinned to that plane and returns the parsed RESULT."""

    def setUp(self) -> None:
        self.root = Path(tempfile.mkdtemp(prefix="kaizen-h0-"))
        self.addCleanup(_rmtree, self.root)
        self.assertEqual(kaizen(self.root, "K1")[0], 0)

    def drive(self, body: str, *, env: dict | None = None, timeout: float = 120.0) -> dict:
        script = "BODY = " + repr(body) + "\n" + _PREAMBLE
        full_env = dict(os.environ)
        full_env["KAIZEN_REPO_ROOT"] = str(self.root)
        full_env["KAIZEN_LLM_MODEL"] = "fixture-model"
        full_env["KAIZEN_LLM_BASE_URL"] = "http://127.0.0.1:1/v1"
        if env:
            full_env.update(env)
        proc = subprocess.run(
            [sys.executable, "-c", script], capture_output=True, text=True,
            cwd=str(REPO_ROOT), env=full_env, timeout=timeout,
        )
        for line in proc.stdout.splitlines():
            if line.startswith("RESULT "):
                return json.loads(line[len("RESULT "):])
        self.fail(f"no RESULT.\nstdout:\n{proc.stdout}\nstderr:\n{proc.stderr[-2000:]}")

def _rmtree(path: Path) -> None:
    """Best-effort removal of a scratch session-drive plane, tolerating residual Windows handles during cleanup."""
    import shutil
    shutil.rmtree(path)

# --- 1. start -> events -> final ---------------------------------------------------------------

# --- 2. ask -> approve-by-correlation_id -> tool runs ------------------------------------------

# --- 2b. approve BEFORE the waiter parks (the on-park re-check race fix) -----------------------

# --- 3. ask -> deny ---------------------------------------------------------------------------

# --- 4. ask timeout fail-closed ---------------------------------------------------------------

# --- 5. steer mid-turn ------------------------------------------------------------------------

# --- 6. interrupt -----------------------------------------------------------------------------

# --- 6b. F5 stop-interrupt spine: stream/silence cancellation + acceptance/clear race ----------

# --- 7. kill (waiters denied, adapter dead) ---------------------------------------------------

# --- 8. cursor pagination gapless -------------------------------------------------------------

# --- 9/10. long-poll (mid-poll delivery + timeout returns empty) ------------------------------

# --- 11. engine gate: NO C1/T5 rows on a denied engine ----------------------------------------

# --- 11b. deterministic arg validation: NO C1/T5 rows on a malformed arg -----------------------

# --- 11c. compensating finalization: a post-insert startup failure never dangles the run -------

# --- 12. bad loopback token refused -----------------------------------------------------------

# --- 13. shutdown with an open turn -----------------------------------------------------------

# --- 13b. shutdown of an idle-successful conversation reads completed, not canceled -----------

# --- 14. double-approve -> ALREADY_DECIDED ----------------------------------------------------

# --- canonical gateway on the real (no-factory) driven lane ------------------------------------

# --- H2.1 capabilities + profile ---------------------------------------------------------------

# A deterministically-unreachable Ollama endpoint so the capabilities model probe DEGRADES in the
# scratch subprocess regardless of whether the dev box has a live Ollama on the default port.
_DEAD_OLLAMA = {"KAIZEN_LLM_BASE_URL": "http://127.0.0.1:1/v1"}

# Hermetic vendor-gate env: a PATH with no codex/claude so installed-binary probes fail closed
# deterministically (unavailable -> DENIED_ENGINE_UNAVAILABLE) on every machine, capability calls never
# spawn a real vendor binary, and an offline session/start can never launch a billed vendor child.
# Installed-binary truth lives in test_codex_live.py / test_claude_live.py.
_NO_VENDOR_BINARIES = {"PATH": str(REPO_ROOT)}

# --- P0 workspace-writer lease ---------------------------------------------------------------

# --- gated live smoke (real Ollama) -----------------------------------------------------------

__all__ = tuple(name for name in globals() if not name.startswith('__'))