dispatch.py

#!/usr/bin/env python3
"""zcode-dispatch — dispatch and steer ZCode sessions from any external agent.

Drives the official `zcode-cli app-server` protocol (NDJSON over stdio) to
create/continue ZCode sessions with an explicit model + reasoning level,
answer the protocol's runtime-preference and provider-auth interactions, and
return a sanitized JSON result.

All site-specific values (provider id, models, reasoning default, model title
tags) come from a config file — nothing about a particular station is
hardcoded. See config.example.json.

Usage:
  dispatch.py --cwd DIR --model ID --prompt-file FILE --out OUT.json
              [--config CONFIG.json] [--resume sess_...] [--timeout SEC]
              [--reasoning LEVEL] [--mode MODE]
  dispatch.py --setup [--config CONFIG.json]

Exit codes: 0 completed | 2 arguments | 3 concurrency/resume mismatch |
4 turn failed | 5 timeout (session intact, resumable) | 6 bridge/config error.
"""
# pylint: disable=C0116,C0103,C0209,C0321,C0412,C0415,W0621,W0718
# pylint: disable=R1710,R1732,R0912,R0913,R0914,R0915,R0801,E0606,E0602
import argparse
import base64
import getpass
import glob
import hashlib
import json
import os
import selectors
import sqlite3
import subprocess
import sys
import time
from pathlib import Path

BRIDGE_DIR = Path(__file__).resolve().parent
ZCODE_HOME = os.environ.get("ZCODE_HOME", os.path.expanduser("~/.zcode"))
DB = os.path.join(ZCODE_HOME, "cli", "db", "db.sqlite")
CRED_STORE = os.path.join(ZCODE_HOME, "v2", "credentials.json")
PREFS_RESULT = {"nativeSearchEnhancementsEnabled": False, "memoryEnabled": False,
                "askUserQuestionAutoResolutionEnabled": True,
                "modelContextBudgetStrategy": "preflight-v1"}

DEFAULT_CONFIG = {
    "provider": "account:zai-individual-coding-plan",
    "models": ["GLM-5.3", "GLM-5.3-Flash"],
    "default_reasoning": "max",
    "default_timeout": 300,
    "model_tags": {"GLM-5.3": "glm", "GLM-5.3-Flash": "flash"},
    "builtin_provider_config": None,
    "personal_provider_config": None,
}


def err_exit(code, msg):
    print(f"ERROR: {msg}", file=sys.stderr)
    sys.exit(code)


# ---------------- config ----------------
def load_config(path):
    cfg = dict(DEFAULT_CONFIG)
    candidates = [path] if path else [
        os.environ.get("ZCODE_DISPATCH_CONFIG"),
        os.path.expanduser("~/.config/zcode-dispatch/config.json"),
        str(BRIDGE_DIR / "config.json"),
    ]
    for c in candidates:
        if c and os.path.isfile(c):
            with open(c, encoding="utf-8") as f:
                user = json.load(f)
            for k, v in user.items():
                if v is not None:
                    cfg[k] = v
            break
    return cfg


def detect_builtin_provider_config():
    """Newest zcode-builtin.json across installed desktop runtime versions."""
    base = os.path.join(ZCODE_HOME, "v2", "runtime", "provider")
    hits = []
    for plat in ("linux-x86_64", "linux-aarch64", "darwin-arm64", "win32-x64"):
        d = os.path.join(base, plat)
        if not os.path.isdir(d):
            continue
        for f in glob.glob(os.path.join(d, "*", "endpoint-*", "zcode-builtin.json")):
            rel = f[len(d) + 1:].split(os.sep)[0]
            try:
                key = [int(x) for x in rel.split(".")]
            except ValueError:
                continue
            hits.append((key, f))
    if not hits:
        err_exit(6, "no zcode-builtin.json found under the ZCode runtime dir — "
                    "set builtin_provider_config in the config file")
    hits.sort(key=lambda t: t[0], reverse=True)
    want = os.environ.get("ZCODE_APP_VERSION")
    if want:
        pref = [f for (_, f) in hits if f"/{want}/" in f]
        if pref:
            return pref[0]
    return hits[0][1]


def plan_api_key(provider, cred_store):
    """Decrypt the plan API key from the shared ZCode credential store.

    Values are `enc:v1:` AES-256-GCM blobs; the key is SHA-256 of the
    ZCODE_CREDENTIAL_SECRET env value or ZCode's deterministic fallback
    (zcode-credential-fallback:<platform>:<homedir>:<username>).
    The key is used in-process only and never logged or returned to the caller.
    """
    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
    secret = (os.environ.get("ZCODE_CREDENTIAL_SECRET") or "").strip() or (
        f"zcode-credential-fallback:{sys.platform}:"
        f"{os.path.expanduser('~')}:{getpass.getuser()}")
    key = hashlib.sha256(secret.encode()).digest()
    creds = json.load(open(cred_store, encoding="utf-8"))
    prefix = f"account-provider:coding-plan:{provider}:"
    for k, v in creds.items():
        if k.startswith(prefix) and k.endswith(":api-key"):
            if not (isinstance(v, str) and v.startswith("enc:v1:")):
                return v
            iv_b, tag_b, ct_b = v[len("enc:v1:"):].split(".")
            plain = AESGCM(key).decrypt(
                base64.urlsafe_b64decode(iv_b + "=="),
                base64.urlsafe_b64decode(ct_b + "==")
                + base64.urlsafe_b64decode(tag_b + "=="), None)
            return plain.decode("utf-8")
    err_exit(6, f"no plan api-key credential found for {provider} "
                f"(run `zcode-cli login` first)")


def registry_snapshot(cli_log_dir):
    """Latest account-revision snapshot from the CLI log (provider registry)."""
    for lg in sorted(glob.glob(os.path.join(cli_log_dir, "zcode-*.jsonl")), reverse=True):
        rev = None
        with open(lg, "r", encoding="utf-8", errors="replace") as fh:
            for line in fh:
                if "provider_registry.ready" in line and '"accountRevision"' in line:
                    try:
                        rev = json.loads(line)["context"]["accountRevision"]
                    except Exception:
                        pass
        if rev:
            return rev
    err_exit(6, "no provider_registry.ready event found in the CLI logs — "
                "start zcode-cli once so the registry logs itself")


def model_tag(model_id):
    """Short static-title tag for a model id (flash/glm/heuristic)."""
    low = model_id.lower()
    if "flash" in low:
        return "flash"
    if "glm" in low:
        return "glm"
    return "".join(c for c in low if c.isalnum())[:12]


def run_setup(config_path):
    """Detect provider/models from the installed ZCode runtime, write config.

    Reads the newest zcode-builtin.json (provider rules + builtin model rules)
    and checks the shared credential store for plan api-key PRESENCE (never
    values). Picks the first credentialed zhipu-account provider (interactive
    choice when several are credentialed and stdin is a tty) and writes the
    same config shape as config.example.json.
    """
    builtin = detect_builtin_provider_config()
    with open(builtin, encoding="utf-8") as fh:
        data = json.load(fh)
    rules = (((data.get("config") or {}).get("providerConfigRules") or {})
             .get("providerRules") or [])
    model_rules = (((data.get("config") or {}).get("modelConfigRules") or {})
                   .get("builtinProviderModelRules") or [])
    providers = [r for r in rules
                 if ((r.get("config") or {}).get("access") or {})
                 .get("type") == "zhipu-account"]
    if not providers:
        err_exit(6, "no zhipu-account providers found in the builtin config")
    creds = {}
    try:
        with open(CRED_STORE, encoding="utf-8") as fh:
            creds = json.load(fh)
    except (OSError, json.JSONDecodeError):
        pass

    print(f"Builtin config: {builtin}")
    print("Account providers:")
    candidates = []
    for r in providers:
        pid = r.get("providerId")
        models = sorted({m.get("modelId") for m in model_rules
                         if m.get("providerId") == pid
                         and (m.get("config") or {}).get("enabled", True)} - {None})
        if not models:
            continue
        has_key = any(k.startswith(f"account-provider:coding-plan:{pid}:")
                      and k.endswith(":api-key") for k in creds)
        candidates.append((pid, models, has_key))
        print(f"  {pid}  ({r.get('providerName', '?')})  "
              f"credential: {'present' if has_key else 'absent'}  "
              f"models: {', '.join(models)}")
    if not candidates:
        err_exit(6, "no zhipu-account provider with enabled builtin models")

    credentialed = [c for c in candidates if c[2]]
    if credentialed:
        provider, models, _ = credentialed[0]
        if len(credentialed) > 1 and sys.stdin.isatty():
            print("Multiple credentialed providers:")
            for i, c in enumerate(credentialed, 1):
                print(f"  {i}) {c[0]}")
            pick = input(
                f"Choose provider [1-{len(credentialed)}, default 1]: ").strip()
            if pick.isdigit() and 1 <= int(pick) <= len(credentialed):
                provider, models, _ = credentialed[int(pick) - 1]
    else:
        provider, models, _ = candidates[0]
        print("WARNING: no plan credential found for any provider — run "
              "`zcode-cli login` before dispatching.")

    personal = os.path.join(ZCODE_HOME, "v2", "provider_config.json")
    cfg = {
        "provider": provider,
        "models": models,
        "default_reasoning": "max",
        "default_timeout": 300,
        "model_tags": {m: model_tag(m) for m in models},
        "builtin_provider_config": builtin,
        "personal_provider_config": personal if os.path.isfile(personal) else None,
    }
    out = Path(config_path) if config_path else BRIDGE_DIR / "config.json"
    out.write_text(json.dumps(cfg, indent=2, ensure_ascii=False) + "\n",
                   encoding="utf-8")
    print(f"\nWrote config: {out}")
    print("Trim `models` or adjust defaults if needed, then dispatch with:")
    print(f"  dispatch.py --cwd <exclusive-worktree> --model {models[0]} "
          f"--prompt-file task.md --out result.json")


# ---------------- main ----------------
ap = argparse.ArgumentParser(description="Dispatch and steer ZCode sessions")
ap.add_argument("--cwd")
ap.add_argument("--model")
ap.add_argument("--prompt-file")
ap.add_argument("--out")
ap.add_argument("--resume", default=None, metavar="sess_...")
ap.add_argument("--timeout", type=float)
ap.add_argument("--mode", default="yolo",
                choices=["yolo", "build", "edit", "plan"])
ap.add_argument("--reasoning", default=None)
ap.add_argument("--config", default=None)
ap.add_argument("--setup", action="store_true",
                help="detect provider/models/credentials, write the config "
                     "file, then exit")
args = ap.parse_args()

if args.setup:
    run_setup(args.config)
    sys.exit(0)
if not (args.cwd and args.model and args.prompt_file and args.out):
    err_exit(2, "--cwd, --model, --prompt-file and --out are required "
                "(or run --setup to generate a config file first)")

CONFIG = load_config(args.config)
PROVIDER = CONFIG["provider"]
if args.model not in CONFIG["models"]:
    err_exit(2, f"model {args.model!r} not configured — available: "
                f"{', '.join(CONFIG['models'])}")
REASONING = args.reasoning or CONFIG["default_reasoning"]
TIMEOUT = args.timeout or CONFIG["default_timeout"]
MODEL_TAG = CONFIG["model_tags"].get(args.model, args.model.lower())
CWD = os.path.abspath(args.cwd)
OUT = os.path.abspath(args.out)
HELPER = os.path.join(BRIDGE_DIR, "notify_click.sh")
PROJECT = os.path.basename(CWD.rstrip("/")) or CWD

if not os.path.isdir(CWD):
    err_exit(2, f"--cwd does not exist: {CWD}")
try:
    prompt_text = Path(args.prompt_file).read_text(encoding="utf-8")
except OSError as e:
    err_exit(2, f"--prompt-file unreadable: {e}")
if not prompt_text.strip():
    err_exit(2, "--prompt-file is empty")
if TIMEOUT < 20:
    err_exit(2, "--timeout too small (min 20s)")

result = {"status": "starting", "sessionId": None, "cwd": CWD, "model": args.model,
          "reasoning": REASONING, "resumed": bool(args.resume), "turnStatus": None,
          "response": None, "usage": None,
          "actualProvider": None, "actualModel": None, "actualReasoning": None,
          "staticTitle": None, "visibilityHandoff": None,
          "startedAtUtc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
          "finishedAtUtc": None, "error": None}


def save():
    tmp = OUT + ".tmp"
    Path(tmp).write_text(json.dumps(result, indent=1, ensure_ascii=False) + "\n",
                          encoding="utf-8")
    os.replace(tmp, OUT)


save()


def db_query(sql, params=()):
    try:
        con = sqlite3.connect(DB)
        rows = con.execute(sql, params).fetchall()
        con.close()
        return rows
    except sqlite3.Error:
        return []


# ---------------- one-driver lock per cwd ----------------
lock_dir = BRIDGE_DIR / "locks"
lock_dir.mkdir(exist_ok=True)
lock_path = lock_dir / ("dispatch-"
                        + hashlib.sha1(CWD.encode()).hexdigest()[:16] + ".lock")
if lock_path.exists():
    try:
        other = int(lock_path.read_text(encoding="utf-8").split()[0])
        os.kill(other, 0)
        err_exit(3, f"cwd is already driven by PID {other} ({lock_path.name}). "
                    f"One driver per cwd/session — wait for it to finish.")
    except ValueError:
        pass
    except ProcessLookupError:
        pass
    except PermissionError:
        err_exit(3, f"lock {lock_path.name} belongs to a live process that "
                    f"cannot be signaled — refusing to race it.")
lock_path.write_text(f"{os.getpid()} {args.model} {time.time()}\n", encoding="utf-8")


def release_lock():
    try:
        if lock_path.exists() and lock_path.read_text(
                encoding="utf-8").split()[0] == str(os.getpid()):
            lock_path.unlink()
    except Exception:
        pass


# ---------------- resume prechecks ----------------
if args.resume:
    rows = db_query("SELECT directory FROM session WHERE id=?", (args.resume,))
    if not rows:
        release_lock()
        err_exit(2, f"resume session not found: {args.resume}")
    if os.path.realpath(rows[0][0] or "") != os.path.realpath(CWD):
        release_lock()
        err_exit(3, f"resume session belongs to cwd '{rows[0][0]}', not '{CWD}'.")
    run = db_query("SELECT status FROM turn_usage WHERE session_id=? AND status='running'",
                   (args.resume,))
    if run:
        release_lock()
        err_exit(3, f"session {args.resume} has a running turn — not driving it in parallel.")

# ---------------- bridge preconditions ----------------
TEST_MODE = bool(os.environ.get("DISPATCH_APP_SERVER_CMD"))
if TEST_MODE:
    # Offline test mode (fake app-server): skip station preconditions entirely.
    API_KEY = "test-mode-key"
    push = {"basedOnZCodeBuiltinRevision": "test", "providers": {},
            "states": {}, "revision": "account:[test,{},{}]"}
else:
    BUILTIN = CONFIG.get("builtin_provider_config") or detect_builtin_provider_config()
if not TEST_MODE:
    PERSONAL = CONFIG.get("personal_provider_config") or os.path.join(
        ZCODE_HOME, "v2", "provider_config.json")
    for f in (BUILTIN, PERSONAL):
        if not os.path.isfile(f):
            release_lock()
            err_exit(6, f"provider config file missing: {f}")
    os.environ["ZCODE_BUILTIN_PROVIDER_CONFIG_FILE"] = BUILTIN
    os.environ["ZCODE_PERSONAL_PROVIDER_CONFIG_FILE"] = PERSONAL

    API_KEY = plan_api_key(PROVIDER, CRED_STORE)  # in-process only

    account_rev = None
    cli_logs = os.path.join(ZCODE_HOME, "cli", "log")
    for lg in sorted(glob.glob(os.path.join(cli_logs, "zcode-*.jsonl")), reverse=True):
        with open(lg, "r", encoding="utf-8", errors="replace") as fh:
            for line in fh:
                if "provider_registry.ready" in line and '"accountRevision"' in line:
                    try:
                        account_rev = json.loads(line)["context"]["accountRevision"]
                    except Exception:
                        pass
        if account_rev:
            break
    if not (account_rev or "").startswith("account:"):
        release_lock()
        err_exit(6, "provider registry revision not found in CLI logs.")
    inner = json.loads(account_rev[len("account:"):])
    based_on, prov_map, _old_states = inner[0], inner[1], inner[2]
    new_providers, our_states = {}, {}
    for entry in prov_map:
        pid, cfg = entry.get("providerId"), dict(entry.get("config") or {})
        acc = dict(cfg.get("access") or {})
        if pid == PROVIDER and acc.get("type") == "zhipu-account":
            acc["entitled"] = True
            our_states[pid] = {"availability": "available", "entitled": True,
                               "current": True}
        cfg["access"] = acc
        new_providers[pid] = {k: cfg[k] for k in
                              ("access", "api", "builtinModelIds", "personalModelIds")
                              if k in cfg}
    if not our_states:
        release_lock()
        err_exit(6, f"{PROVIDER} not present in the provider registry snapshot.")
    def jstr(x):
        return json.dumps(x, separators=(",", ":"), ensure_ascii=False)
    push = {"basedOnZCodeBuiltinRevision": based_on, "providers": new_providers,
            "states": our_states,
            "revision": "account:" + jstr([based_on, jstr(new_providers),
                                           jstr(our_states)])}

# ---------------- app-server ----------------
cmd = json.loads(os.environ["DISPATCH_APP_SERVER_CMD"]) \
    if os.environ.get("DISPATCH_APP_SERVER_CMD") \
    else ["zcode-cli", "app-server", "--surface", "terminal", "--cwd", CWD]
p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                     stderr=subprocess.PIPE, env=dict(os.environ), bufsize=0)


def send(x):
    p.stdin.write((json.dumps(x) + "\n").encode())
    p.stdin.flush()


sel = selectors.DefaultSelector()
sel.register(p.stdout, selectors.EVENT_READ)
sel.register(p.stderr, selectors.EVENT_READ)
buf = {p.stdout: b"", p.stderr: b""}
deadline = time.monotonic() + TIMEOUT
phase = {"v": "push-wait"}
last_err = {"v": None}
turn_baseline = {"v": 0}
err_delta = {}
assistant = {"text": None, "finish": None, "model": None}
start_notified = {"v": False}


def note(msg):
    print(f"[{time.strftime('%H:%M:%S')}] {msg}", file=sys.stderr, flush=True)


def notify_started():
    """One-shot STARTED desktop notification (fail-safe, click-through)."""
    if start_notified["v"] or not result.get("sessionId") \
            or not os.path.exists(HELPER):
        return
    start_notified["v"] = True
    try:
        kind = "Steer" if args.resume else "New"
        title = f"🚀 ZCode STARTED 🟢🔵 {(result.get('staticTitle') or PROJECT)[:70]}"
        body = f"Type: {kind} · Model: {args.model} · Project: {PROJECT}"
        subprocess.Popen(["setsid", "bash", HELPER, result["sessionId"],
                          title, body], stdout=subprocess.DEVNULL,
                         stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL,
                         start_new_session=True)
    except Exception:
        pass


def on_line(d):
    if d.get("method") == "session/requestRuntimePreferences":
        send({"id": d["id"], "result": PREFS_RESULT})
    elif d.get("method") == "interaction/requestProviderRuntimeHeaders":
        ok = (d.get("params", {}).get("providerId") == PROVIDER
              and d.get("params", {}).get("reason") == "model-request"
              and d.get("params", {}).get("modelId", args.model) == args.model)
        send({"id": d["id"], "result": {"headersApplied": bool(ok),
             **({"requestAuth": {"apiKey": API_KEY}} if ok else
                {"errorMessage": "declined: request context does not match the "
                                 "dispatched provider/model"})}})
    elif d.get("id") == "push":
        if d.get("error"):
            last_err["v"] = "updateAccountConfig: " + json.dumps(d["error"])[:200]
            phase["v"] = "push-failed"
        else:
            phase["v"] = "resume" if args.resume else "create"
    elif d.get("id") == "bridge-create":
        if d.get("error"):
            last_err["v"] = "create: " + json.dumps(d["error"])[:200]
            phase["v"] = "create-failed"
        else:
            result["sessionId"] = ((d.get("result") or {}).get("session") or {}) \
                .get("sessionId")
            result["status"] = "session-created"
            save()
            phase["v"] = "send"
    elif d.get("id") == "bridge-resume":
        if d.get("error"):
            last_err["v"] = "resume: " + json.dumps(d["error"])[:200]
            phase["v"] = "create-failed"
        else:
            result["sessionId"] = args.resume
            result["status"] = "session-resumed"
            save()
            phase["v"] = "send"
    elif d.get("id") == "bridge-send":
        if d.get("error"):
            last_err["v"] = "send: " + json.dumps(d["error"])[:200]
            phase["v"] = "send-failed"
        else:
            phase["v"] = "turn-wait"
            result["status"] = "turn-running"
            save()
            notify_started()
    elif d.get("id") == "bridge-usage":
        r = d.get("result")
        if r:
            result["usage"] = {"modelRequests": r.get("modelRequestCount"),
                               "inputTokens": r.get("inputTokens"),
                               "outputTokens": r.get("outputTokens"),
                               "reasoningTokens": r.get("reasoningTokens"),
                               "totalTokens": r.get("totalTokens"),
                               "modelErrors": r.get("modelErrorCount")}
            if "base" not in err_delta:
                err_delta["base"] = r.get("modelErrorCount") or 0
            elif (r.get("modelErrorCount") or 0) > err_delta["base"]:
                last_err["v"] = (f"model error during turn (delta modelErrors="
                                 f"{(r.get('modelErrorCount') or 0) - err_delta['base']})")
                phase["v"] = "turn-failed"


# ---------------- static title: pin at create, re-pin at finish ----------------
def pin_static_title(sid):
    tag = MODEL_TAG
    pfl = next((ln.strip() for ln in prompt_text.splitlines() if ln.strip()), PROJECT)
    result["staticTitle"] = f"{PROJECT} · {pfl[:55]} ({tag})"[:90]
    db_query("UPDATE session SET title=?, title_source='custom',"
             " time_title_updated=? WHERE id=?",
             (result["staticTitle"], int(time.time() * 1000), sid))


try:
    send({"id": "push", "method": "provider/updateAccountConfig", "params": push})
    poll_deadline = None
    while time.monotonic() < deadline:
        for key, _ in sel.select(1):
            try:
                chunk = os.read(key.fd, 65536)
            except OSError:
                continue
            if not chunk:
                sel.unregister(key.fileobj)
                continue
            buf[key.fileobj] += chunk
            while b"\n" in buf[key.fileobj]:
                line, buf[key.fileobj] = buf[key.fileobj].split(b"\n", 1)
                if key.fileobj == p.stderr:
                    continue
                try:
                    d = json.loads(line)
                except json.JSONDecodeError:
                    continue
                on_line(d)
        if phase["v"] in ("push-failed", "create-failed", "send-failed",
                          "turn-failed"):
            break
        if phase["v"] == "create":
            pin_static_title("")  # sid unknown yet — stored in result only
            send({"id": "bridge-create", "method": "session/create", "params": {
                "workspace": {"workspacePath": CWD, "workspaceKey": CWD},
                "mode": args.mode,
                "model": {"providerId": PROVIDER, "modelId": args.model,
                          "options": {"reasoningLevel": REASONING}},
                "thoughtLevel": REASONING, "titleGenerationEnabled": False}})
            phase["v"] = "create-wait"
        elif phase["v"] == "resume":
            send({"id": "bridge-resume", "method": "session/resume", "params": {
                "sessionId": args.resume,
                "workspace": {"workspacePath": CWD, "workspaceKey": CWD}}})
            phase["v"] = "resume-wait"
        elif phase["v"] == "send" and result["sessionId"]:
            if not args.resume:
                pin_static_title(result["sessionId"])
            send({"id": "bridge-send", "method": "session/send",
                  "params": {"sessionId": result["sessionId"], "content": prompt_text}})
            phase["v"] = "send-ack"
        elif phase["v"] == "turn-wait" and time.monotonic() > (poll_deadline or 0):
            poll_deadline = time.monotonic() + 4
            send({"id": "bridge-usage", "method": "session/usage",
                  "params": {"sessionId": result["sessionId"]}})
            tr = db_query("SELECT status FROM turn_usage WHERE session_id=? AND"
                          " started_at>? ORDER BY started_at DESC LIMIT 1",
                          (result["sessionId"], turn_baseline["v"]))
            if tr and tr[0][0] in ("completed", "failed", "error", "aborted",
                                   "cancelled"):
                phase["v"] = "turn-done"
                note(f"turn end detected: {tr[0][0]}")
                break
    if phase["v"] == "turn-wait" and time.monotonic() >= deadline:
        result["status"] = "timeout"
        result["error"] = (f"timeout after {TIMEOUT}s — turn may still be "
                           f"running; session stays resumable.")
finally:
    if result.get("sessionId") and result["status"] != "timeout":
        try:
            send({"id": "bridge-close", "method": "session/close",
                  "params": {"sessionId": result["sessionId"]}})
            time.sleep(0.4)
        except Exception:
            pass
    try:
        p.stdin.close()
    except Exception:
        pass
    try:
        p.wait(timeout=4)
    except Exception:
        p.terminate()
        try:
            p.wait(timeout=3)
        except Exception:
            p.kill()
    release_lock()


def finish_eval(status):
    # Final title pin (the app server's runtime first-input title would
    # otherwise win over the pinned static title)
    try:
        if result.get("staticTitle") and not args.resume and result.get("sessionId"):
            db_query("UPDATE session SET title=?, title_source='custom' WHERE id=?",
                     (result["staticTitle"], result["sessionId"]))
    except Exception:
        pass
    # Desktop notifications (fail-safe: never touches dispatcher/agent flow)
    try:
        if result.get("sessionId") and os.path.exists(HELPER):
            kind = "Steer" if args.resume else "New"
            emoji = {"completed": "✅ ZCode COMPLETED 🟢🟩",
                     "failed": "❌ ZCode FAILED 🔴🔴",
                     "timeout": "🟡 ZCode TIMEOUT 🟡"}.get(status,
                                                          "• ZCode " + status.upper())
            body = f"Type: {kind} · Model: {args.model} · Project: {PROJECT}"
            u = result.get("usage") or {}
            if u.get("totalTokens"):
                body += (f"\nTokens: {u.get('inputTokens','?')} in / "
                         f"{u.get('outputTokens','?')} out")
            resp = (result.get("response") or "").replace("\n", " ")[:100]
            if resp:
                body += f"\n↳ {resp}…"
            subprocess.Popen(["setsid", "bash", HELPER, result["sessionId"],
                              f"{emoji}: {(result.get('staticTitle') or PROJECT)[:70]}",
                              body], stdout=subprocess.DEVNULL,
                             stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL,
                             start_new_session=True)
    except Exception:
        pass
    result["status"] = status
    result["finishedAtUtc"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    save()
    lean = {k: result[k] for k in ("status", "sessionId", "cwd", "model",
                                   "reasoning", "actualProvider", "actualModel",
                                   "actualReasoning", "turnStatus", "response",
                                   "usage", "error", "staticTitle",
                                   "visibilityHandoff") if result.get(k) is not None}
    print(json.dumps(lean, ensure_ascii=False, indent=1))
    sys.exit({"completed": 0, "failed": 4, "timeout": 5}.get(status, 6))


if last_err["v"]:
    result["error"] = last_err["v"]
    result["turnStatus"] = result.get("turnStatus") or "none"
    finish_eval("failed")
if phase["v"] not in ("turn-wait", "turn-done"):
    result["error"] = result["error"] or f"internal phase '{phase['v']}'"
    finish_eval("failed")
if result["status"] == "timeout":
    finish_eval("timeout")

sid = result["sessionId"]
sel_rows = db_query(
    "SELECT DISTINCT json_extract(data,'$.modelSelection.providerId'),"
    " json_extract(data,'$.modelSelection.modelId'),"
    " json_extract(data,'$.modelSelection.options.reasoningLevel')"
    " FROM message WHERE session_id=? AND"
    " json_extract(data,'$.modelSelection.modelId') IS NOT NULL", (sid,))
if sel_rows:
    result["actualProvider"], result["actualModel"], result["actualReasoning"] = sel_rows[-1]
txt_rows = db_query(
    "SELECT json_extract(p.data,'$.text') FROM part p JOIN message m"
    " ON m.id=p.message_id WHERE p.session_id=? AND json_extract(p.data,'$.type')='text'"
    " AND json_extract(m.data,'$.role')='assistant' AND m.time_created > ?"
    " ORDER BY p.sequence", (sid, turn_baseline["v"]))
result["response"] = "\n".join(t for (t,) in txt_rows if t) or None
turn = db_query("SELECT status, input_tokens, output_tokens, computed_total_tokens"
                " FROM turn_usage WHERE session_id=? AND started_at>?"
                " ORDER BY started_at DESC LIMIT 1", (sid, turn_baseline["v"]))
if turn:
    result["turnStatus"], tin, tout, ttotal = turn[0]
    result["usage"] = result["usage"] or {}
    result["usage"].update({"inputTokens": tin, "outputTokens": tout,
                            "totalTokens": ttotal})
if (result.get("usage") or {}).get("modelErrors"):
    result["error"] = "model error during turn"
    finish_eval("failed")
if result["turnStatus"] not in ("completed",):
    result["error"] = f"turn status '{result['turnStatus']}' — not completed."
    finish_eval("failed" if result["turnStatus"] in ("failed", "error", "aborted")
                else "timeout")
if result.get("actualModel") not in CONFIG["models"] or \
        result.get("actualReasoning") != REASONING:
    result["error"] = "model/reasoning not verified as dispatched."
    finish_eval("failed")
finish_eval("completed")