decision_service

zocrates.rag.decision_service

Decision model service manager for citation verification and local Laya tooling.

Provides lifecycle management for the local Laya decision model server (running as an isolated uv tool), probing tools and health status, and testing decision model connectivity against OpenRouter or local endpoints.

DEFAULT_CITATION_VERIFICATION_LOCAL_URL module-attribute

DEFAULT_CITATION_VERIFICATION_LOCAL_URL = 'http://127.0.0.1:8005/v1/systemone'

DEFAULT_CITATION_VERIFICATION_MODEL module-attribute

DEFAULT_CITATION_VERIFICATION_MODEL = '~typesafe/jev-latest'

__all__ module-attribute

__all__ = ['check_laya_running', 'find_uv_binary', 'get_decision_tool_status', 'install_laya_tool', 'is_laya_installed', 'is_uv_available', 'start_laya_daemon', 'stop_laya_daemon', 'test_decision_model_connectivity']

_daemon_lock module-attribute

_daemon_lock = threading.Lock()

_managed_port module-attribute

_managed_port = None

_managed_process module-attribute

_managed_process = None

logger module-attribute

logger = logging.getLogger(__name__)

_cleanup_managed_daemon

_cleanup_managed_daemon()

Atexit hook to terminate any background Laya process on interpreter exit.

Source code in zocrates/rag/decision_service.py

def _cleanup_managed_daemon() -> None:
    """Atexit hook to terminate any background Laya process on interpreter exit."""
    global _managed_process, _managed_port
    managed_pid = None
    if _managed_process is not None:
        managed_pid = _managed_process.pid
        if _managed_process.poll() is None:
            try:
                _managed_process.terminate()
                _managed_process.wait(timeout=2.0)
            except Exception:
                try:
                    _managed_process.kill()
                except Exception:
                    pass
        _managed_process = None
        _managed_port = None

    try:
        data_dir = get_zocrates_data_dir()
        pid_file = data_dir / "laya.pid"
        port_file = data_dir / "laya.port"
        if pid_file.exists():
            try:
                content = pid_file.read_text(encoding="utf-8").strip()
                if managed_pid is not None and content == str(managed_pid):
                    pid_file.unlink(missing_ok=True)
                    port_file.unlink(missing_ok=True)
            except Exception:
                pass
    except Exception:
        pass

_is_laya_process

_is_laya_process(proc)

Check whether a process is definitively a Laya daemon.

Requires that the executable, process name, or a command-line token specifically represents the laya-serve or laya executable/module, rather than a casual substring match in file paths or arguments.

Source code in zocrates/rag/decision_service.py

def _is_laya_process(proc: Any) -> bool:
    """Check whether a process is definitively a Laya daemon.

    Requires that the executable, process name, or a command-line token
    specifically represents the `laya-serve` or `laya` executable/module,
    rather than a casual substring match in file paths or arguments.
    """
    try:
        # Check process name
        name_val = proc.name() if callable(getattr(proc, "name", None)) else getattr(proc, "name", "")
        name = str(name_val or "").lower()
        if name in ("laya-serve", "laya"):
            return True

        # Check command line tokens
        cmdline_val = proc.cmdline() if callable(getattr(proc, "cmdline", None)) else getattr(proc, "cmdline", [])
        if isinstance(cmdline_val, (list, tuple)) and cmdline_val:
            # Check binary / script name in argv[0]
            first_arg = Path(str(cmdline_val[0])).name.lower()
            if first_arg in ("laya-serve", "laya"):
                return True

            # Check if running python inside isolated uv laya tool
            if "/uv/tools/laya/" in str(cmdline_val[0]).lower().replace("\\", "/"):
                return True

            # Check arguments for exact laya-serve command or laya module
            for i, arg in enumerate(cmdline_val):
                arg_str = str(arg).strip()
                arg_base = Path(arg_str).name.lower()
                if arg_base in ("laya-serve", "laya"):
                    return True
                if arg_str in ("laya.serve", "laya:serve"):
                    return True
                if arg_str == "-m" and i + 1 < len(cmdline_val) and cmdline_val[i + 1] in ("laya", "laya.serve"):
                    return True
                # uv tool run laya-serve
                if arg_str == "run" and i + 1 < len(cmdline_val) and Path(cmdline_val[i + 1]).name.lower() == "laya-serve":
                    return True

        # Check exe path if available
        exe_val = proc.exe() if callable(getattr(proc, "exe", None)) else getattr(proc, "exe", "")
        if exe_val:
            exe_base = Path(str(exe_val)).name.lower()
            if exe_base in ("laya-serve", "laya"):
                return True
            if "/uv/tools/laya/" in str(exe_val).lower().replace("\\", "/"):
                return True
    except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
        return False
    except Exception:
        return False
    return False

_process_is_on_port

_process_is_on_port(proc, port)

Check if process is actively listening on the target port.

Source code in zocrates/rag/decision_service.py

def _process_is_on_port(proc: Any, port: int) -> bool:
    """Check if process is actively listening on the target port."""
    try:
        conn_method = getattr(proc, "net_connections", getattr(proc, "connections", None))
        if callable(conn_method):
            for conn in conn_method(kind="inet"):
                status = getattr(conn, "status", None)
                if status in (psutil.CONN_LISTEN, "LISTEN"):
                    laddr = getattr(conn, "laddr", None)
                    if laddr and getattr(laddr, "port", None) == port:
                        return True
    except (psutil.AccessDenied, psutil.NoSuchProcess, psutil.ZombieProcess, Exception):
        pass

    return False

_terminate_psutil_proc

_terminate_psutil_proc(proc, timeout=3.0)

Safely terminate a process (SIGTERM, then SIGKILL if needed).

Source code in zocrates/rag/decision_service.py

def _terminate_psutil_proc(proc: Any, timeout: float = 3.0) -> bool:
    """Safely terminate a process (SIGTERM, then SIGKILL if needed)."""
    try:
        if callable(getattr(proc, "terminate", None)):
            proc.terminate()
        try:
            if callable(getattr(proc, "wait", None)):
                proc.wait(timeout=timeout)
        except (psutil.TimeoutExpired, subprocess.TimeoutExpired):
            if callable(getattr(proc, "kill", None)):
                proc.kill()
            try:
                if callable(getattr(proc, "wait", None)):
                    proc.wait(timeout=1.0)
            except Exception:
                pass
        return True
    except (psutil.NoSuchProcess, psutil.ZombieProcess):
        return True
    except Exception as exc:
        logger.warning("Error terminating process: %s", exc)
        return False

check_laya_running

check_laya_running(url=DEFAULT_CITATION_VERIFICATION_LOCAL_URL, timeout=0.5)

Check if the local Laya server is running and reachable.

Source code in zocrates/rag/decision_service.py

def check_laya_running(
    url: str = DEFAULT_CITATION_VERIFICATION_LOCAL_URL,
    timeout: float = 0.5,
) -> bool:
    """Check if the local Laya server is running and reachable."""
    parsed = urllib.parse.urlparse(url)
    health_url = f"{parsed.scheme}://{parsed.netloc}/health"
    try:
        resp = requests.get(health_url, timeout=timeout)
        if resp.status_code == 200:
            try:
                data = resp.json()
                if isinstance(data, dict):
                    if data.get("status") == "ok" and any(
                        k in data for k in ("loaded", "device", "checkpoint_devices", "cpu_fallbacks")
                    ):
                        return True
            except Exception:
                pass
    except Exception:
        pass

    openapi_url = f"{parsed.scheme}://{parsed.netloc}/openapi.json"
    try:
        resp = requests.get(openapi_url, timeout=timeout)
        if resp.status_code == 200:
            try:
                data = resp.json()
                if isinstance(data, dict):
                    info = data.get("info", {})
                    if isinstance(info, dict):
                        title = str(info.get("title", "")).lower()
                        desc = str(info.get("description", "")).lower()
                        if "laya" in title or "laya" in desc:
                            return True
            except Exception:
                pass
    except Exception:
        pass

    return False

find_laya_python

find_laya_python()

Locate the Python executable used by the isolated Laya uv tool environment.

Source code in zocrates/rag/decision_service.py

def find_laya_python() -> str | None:
    """Locate the Python executable used by the isolated Laya uv tool environment."""
    tool_py = Path.home() / ".local" / "share" / "uv" / "tools" / "laya" / "bin" / "python3"
    if tool_py.is_file() and os.access(tool_py, os.X_OK):
        return str(tool_py)

    laya_bin = find_laya_serve_binary()
    if laya_bin:
        try:
            with open(laya_bin, "r", encoding="utf-8") as f:
                first_line = f.readline()
                if first_line.startswith("#!"):
                    shebang = first_line[2:].strip()
                    if os.path.isfile(shebang) and os.access(shebang, os.X_OK):
                        return shebang
        except Exception:
            pass

    return shutil.which("python3")

find_laya_serve_binary

find_laya_serve_binary()

Locate the laya-serve binary on PATH or in standard user tool directories.

Source code in zocrates/rag/decision_service.py

def find_laya_serve_binary() -> str | None:
    """Locate the `laya-serve` binary on PATH or in standard user tool directories."""
    found = shutil.which("laya-serve")
    if found:
        return found
    candidates = [
        Path.home() / ".local" / "bin" / "laya-serve",
        Path.home() / ".cargo" / "bin" / "laya-serve",
    ]
    for candidate in candidates:
        if candidate.is_file() and os.access(candidate, os.X_OK):
            return str(candidate)
    return None

find_uv_binary

find_uv_binary()

Locate the uv executable on PATH or standard install locations.

Source code in zocrates/rag/decision_service.py

def find_uv_binary() -> str | None:
    """Locate the `uv` executable on PATH or standard install locations."""
    found = shutil.which("uv")
    if found:
        return found
    uv_name = "uv.exe" if sys.platform == "win32" else "uv"
    candidates = [
        Path.home() / ".cargo" / "bin" / uv_name,
        Path.home() / ".local" / "bin" / uv_name,
        Path("/usr/local/bin/uv"),
        Path("/opt/homebrew/bin/uv"),
    ]
    try:
        resolved_zotero = resolve_zotero_data_dir()
        if resolved_zotero is not None:
            candidates.append(resolved_zotero / "zocrates-data" / "bin" / uv_name)
    except Exception:
        pass

    for candidate in candidates:
        if candidate.is_file() and os.access(candidate, os.X_OK):
            return str(candidate)
    return None

get_backend_port

get_backend_port(zotero_data_dir=None, cfg=None)

Return configured backend port from environment or zocrates_config.json, or None if dynamic.

Delegates to the shared env/config parse helpers used by _resolve_backend_port (env BACKEND_PORT > config "backend_port") instead of duplicating the parsing logic. None means the dynamic deterministic offset tier applies (see compute_deterministic_port).

Source code in zocrates/config.py

def get_backend_port(
    zotero_data_dir: Path | None = None,
    cfg: dict[str, Any] | None = None,
) -> int | None:
    """Return configured backend port from environment or zocrates_config.json, or None if dynamic.

    Delegates to the shared env/config parse helpers used by
    ``_resolve_backend_port`` (env BACKEND_PORT > config "backend_port")
    instead of duplicating the parsing logic. ``None`` means the dynamic
    deterministic offset tier applies (see ``compute_deterministic_port``).
    """
    env_port = _parse_env_backend_port()
    if env_port is not None:
        return env_port
    if cfg is not None:
        port = cfg.get("backend_port")
        if port is None:
            return None
        try:
            port_int = int(port)
            return port_int if port_int > 0 else None
        except (ValueError, TypeError):
            return None
    return _parse_config_backend_port(zotero_data_dir)

get_decision_tool_status

get_decision_tool_status(config=None)

Inspect local decision model runtime environment and return state metrics.

Source code in zocrates/rag/decision_service.py

def get_decision_tool_status(config: dict[str, Any] | None = None) -> dict[str, Any]:
    """Inspect local decision model runtime environment and return state metrics."""
    cfg = dict(config) if config is not None else load_zocrates_config()
    local_url = cfg.get("citation_verification_local_url", DEFAULT_CITATION_VERIFICATION_LOCAL_URL)
    port = int(cfg.get("citation_verification_local_port", 8005))

    uv_avail = is_uv_available()
    laya_inst = is_laya_installed()
    running = check_laya_running(local_url, timeout=0.5)

    active_device = "unknown"
    if running:
        parsed = urllib.parse.urlparse(local_url)
        health_url = f"{parsed.scheme}://{parsed.netloc}/health"
        try:
            h_resp = requests.get(health_url, timeout=0.5)
            if h_resp.status_code == 200:
                h_data = h_resp.json()
                active_device = h_data.get("device", "running")
        except Exception:
            active_device = "running"
    elif laya_inst:
        active_device = "installed (offline)"
    else:
        active_device = "not installed"

    global _managed_process
    daemon_managed = _managed_process is not None and _managed_process.poll() is None

    return {
        "status": "success",
        "uv_available": uv_avail,
        "tool_installed": laya_inst,
        "server_running": running,
        "daemon_managed": daemon_managed,
        "daemon_starting": daemon_managed and not running,
        "port": port,
        "local_url": local_url,
        "active_device": active_device,
    }

get_secure_api_key

get_secure_api_key(key_name='openrouter_api_key', env_vars=('OPENROUTER_API_KEY', 'OPEN_ROUTER_API_KEY'))

Load a named API key from in-memory session environment variables.

Zotero Preferences / Login Manager serves as the secure persistent vault. Backend keys are in-memory session variables (os.environ) synced from Zotero.

Parameters:

Name	Type	Description	Default
key_name	str	Entry identifier (e.g., 'openrouter_api_key', 'openai_api_key').	'openrouter_api_key'
env_vars	list[str] | tuple[str, ...] | None	Environment variable names to check.	('OPENROUTER_API_KEY', 'OPEN_ROUTER_API_KEY')

Returns:

Type	Description
str | None	The API key string or None if not found.

Source code in zocrates/config.py

def get_secure_api_key(
    key_name: str = "openrouter_api_key",
    env_vars: list[str] | tuple[str, ...] | None = ("OPENROUTER_API_KEY", "OPEN_ROUTER_API_KEY"),
) -> str | None:
    """Load a named API key from in-memory session environment variables.

    Zotero Preferences / Login Manager serves as the secure persistent vault.
    Backend keys are in-memory session variables (os.environ) synced from Zotero.

    Args:
        key_name: Entry identifier (e.g., 'openrouter_api_key', 'openai_api_key').
        env_vars: Environment variable names to check.

    Returns:
        The API key string or None if not found.
    """
    if env_vars:
        for var in env_vars:
            val = os.getenv(var)
            if val:
                return val
    return None

get_zocrates_data_dir

get_zocrates_data_dir(zotero_data_dir=None)

Return (and create if needed) the Zocrates data directory.

Parameters:

Name	Type	Description	Default
zotero_data_dir	Path | str | None	Root Zotero data directory (Path, string, or None for auto-discovery).	None

Returns:

Type	Description
Path	Path to <zotero_data_dir>/zocrates-data/, created if absent.

Source code in zocrates/config.py

def get_zocrates_data_dir(zotero_data_dir: Path | str | None = None) -> Path:
    """Return (and create if needed) the Zocrates data directory.

    Args:
        zotero_data_dir: Root Zotero data directory (Path, string, or None for auto-discovery).

    Returns:
        Path to ``<zotero_data_dir>/zocrates-data/``, created if absent.
    """
    resolved = resolve_zotero_data_dir(zotero_data_dir)
    if resolved is None:
        raise RuntimeError(
            "Cannot determine Zotero data directory. "
            "Please ensure Zotero is running and registered, or set ZOTERO_DATA_DIR."
        )
    zocrates_dir = resolved / ZOCRATES_DATA_SUBDIR
    zocrates_dir.mkdir(parents=True, exist_ok=True)
    return zocrates_dir

install_laya_tool

install_laya_tool()

Install Laya as an isolated uv tool with FastAPI and Uvicorn.

Source code in zocrates/rag/decision_service.py

def install_laya_tool() -> dict[str, Any]:
    """Install Laya as an isolated uv tool with FastAPI and Uvicorn."""
    uv_bin = find_uv_binary()
    if not uv_bin:
        return {
            "status": "error",
            "message": "The `uv` package manager was not found. Please install `uv` (https://docs.astral.sh/uv/) first.",
        }

    cmd = [
        uv_bin,
        "tool",
        "install",
        "--with",
        "uvicorn",
        "--with",
        "fastapi",
        "laya",
    ]
    logger.info("Installing Laya tool via: %s", " ".join(cmd))
    try:
        proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=300.0,
        )
        if proc.returncode != 0:
            err_msg = proc.stderr.strip() or proc.stdout.strip() or f"Process exited with code {proc.returncode}"
            logger.error("Failed to install Laya uv tool: %s", err_msg)
            return {
                "status": "error",
                "message": f"Installation failed: {err_msg}",
            }
        logger.info("Successfully installed Laya uv tool.")
        return {
            "status": "success",
            "message": "Laya tool installed successfully.",
        }
    except subprocess.TimeoutExpired:
        return {
            "status": "error",
            "message": "Installation timed out after 300 seconds.",
        }
    except Exception as exc:
        logger.error("Exception installing Laya: %s", exc)
        return {
            "status": "error",
            "message": f"Installation error: {exc}",
        }

is_laya_installed

is_laya_installed()

Return whether Laya is installed either as a standalone binary or uv tool.

Source code in zocrates/rag/decision_service.py

def is_laya_installed() -> bool:
    """Return whether Laya is installed either as a standalone binary or uv tool."""
    if find_laya_serve_binary() is not None:
        return True

    uv_bin = find_uv_binary()
    if uv_bin:
        try:
            res = subprocess.run(
                [uv_bin, "tool", "list"],
                capture_output=True,
                text=True,
                timeout=3.0,
            )
            if res.returncode == 0 and "laya" in res.stdout:
                return True
        except Exception:
            pass

    tool_dir = Path.home() / ".local" / "share" / "uv" / "tools" / "laya"
    if tool_dir.exists() and tool_dir.is_dir():
        return True

    return False

is_uv_available

is_uv_available()

Return whether the uv package manager is installed and executable.

Source code in zocrates/rag/decision_service.py

def is_uv_available() -> bool:
    """Return whether the `uv` package manager is installed and executable."""
    return find_uv_binary() is not None

load_zocrates_config

load_zocrates_config(zotero_data_dir=None)

Load the Zocrates config from the data directory.

If zotero_data_dir is None, attempts to obtain it from the active LibrarySyncer service or the ZOTERO_DATA_DIR environment variable.

Returns config merged with standard defaults.

Parameters:

Name	Type	Description	Default
zotero_data_dir	Path | str | None	Root Zotero data directory.	None

Returns:

Type	Description
dict	Parsed JSON config dict merged with defaults.

Source code in zocrates/config.py

def load_zocrates_config(zotero_data_dir: Path | str | None = None) -> dict:
    """Load the Zocrates config from the data directory.

    If ``zotero_data_dir`` is ``None``, attempts to obtain it from the active
    ``LibrarySyncer`` service or the ``ZOTERO_DATA_DIR`` environment variable.

    Returns config merged with standard defaults.

    Args:
        zotero_data_dir: Root Zotero data directory.

    Returns:
        Parsed JSON config dict merged with defaults.
    """
    defaults = copy.deepcopy(DEFAULT_CONFIG)

    resolved_dir = resolve_zotero_data_dir(zotero_data_dir)
    if resolved_dir is None:
        return defaults

    config_path = get_config_path(resolved_dir)
    resolved_path = config_path.resolve()
    if not config_path.exists():
        with _CONFIG_CACHE_LOCK:
            _CONFIG_CACHE.pop(resolved_path, None)
            _CONFIG_CACHE_FINGERPRINT.pop(resolved_path, None)
        return defaults

    try:
        stat_res = config_path.stat()
        current_fingerprint = (stat_res.st_mtime_ns, stat_res.st_size)
    except OSError:
        return defaults

    with _CONFIG_CACHE_LOCK:
        if resolved_path in _CONFIG_CACHE and _CONFIG_CACHE_FINGERPRINT.get(resolved_path) == current_fingerprint:
            return copy.deepcopy(_CONFIG_CACHE[resolved_path])

    if config_path.exists():
        try:
            lock_path = str(config_path) + ".lock"
            with FileLock(lock_path):
                try:
                    stat_res = config_path.stat()
                    current_fingerprint = (stat_res.st_mtime_ns, stat_res.st_size)
                except OSError:
                    return defaults

                with _CONFIG_CACHE_LOCK:
                    if resolved_path in _CONFIG_CACHE and _CONFIG_CACHE_FINGERPRINT.get(resolved_path) == current_fingerprint:
                        return copy.deepcopy(_CONFIG_CACHE[resolved_path])

                with open(config_path) as f:
                    loaded = json.load(f)

                # Backfill override_agent_models if missing or incomplete
                needs_save = False

                if "override_agent_models" not in loaded or not isinstance(loaded["override_agent_models"], dict):
                    loaded["override_agent_models"] = copy.deepcopy(DEFAULT_OVERRIDE_AGENT_MODELS)
                    needs_save = True
                else:
                    for role, default_role_cfg in DEFAULT_OVERRIDE_AGENT_MODELS.items():
                        if role not in loaded["override_agent_models"]:
                            loaded["override_agent_models"][role] = copy.deepcopy(default_role_cfg)
                            needs_save = True

                # Backfill search_harness if missing or incomplete in existing config file on disk
                if "search_harness" not in loaded or not isinstance(loaded["search_harness"], dict):
                    loaded["search_harness"] = copy.deepcopy(DEFAULT_SEARCH_HARNESS)
                    needs_save = True
                else:
                    for k, default_val in DEFAULT_SEARCH_HARNESS.items():
                        if k not in loaded["search_harness"]:
                            loaded["search_harness"][k] = copy.deepcopy(default_val)
                            needs_save = True

                # Backfill graph_relations if missing or incomplete in existing config file on disk
                if "graph_relations" not in loaded or not isinstance(loaded["graph_relations"], dict):
                    loaded["graph_relations"] = copy.deepcopy(DEFAULT_GRAPH_RELATIONS)
                    needs_save = True
                else:
                    for k, default_val in DEFAULT_GRAPH_RELATIONS.items():
                        if k not in loaded["graph_relations"]:
                            loaded["graph_relations"][k] = copy.deepcopy(default_val)
                            needs_save = True

                # Backfill llm_num_retries if missing from existing config file on disk
                if "llm_num_retries" not in loaded:
                    loaded["llm_num_retries"] = DEFAULT_LLM_NUM_RETRIES
                    needs_save = True

                # Backfill llm_base_url if missing from existing config file on disk
                if "llm_base_url" not in loaded:
                    loaded["llm_base_url"] = DEFAULT_LLM_BASE_URL
                    needs_save = True

                # Backfill citations options if missing from existing config file on disk
                if "citations_enabled" not in loaded:
                    loaded["citations_enabled"] = DEFAULT_CITATIONS_ENABLED
                    needs_save = True
                if "references_section_enabled" not in loaded:
                    loaded["references_section_enabled"] = DEFAULT_REFERENCES_SECTION_ENABLED
                    needs_save = True
                if "considered_sources_disclosure" not in loaded:
                    loaded["considered_sources_disclosure"] = DEFAULT_CONSIDERED_SOURCES_DISCLOSURE
                    needs_save = True
                if "auto_render_mermaid_diagrams" not in loaded:
                    loaded["auto_render_mermaid_diagrams"] = DEFAULT_AUTO_RENDER_MERMAID_DIAGRAMS
                    needs_save = True
                if "enable_mcp_server" not in loaded:
                    loaded["enable_mcp_server"] = DEFAULT_ENABLE_MCP_SERVER
                    needs_save = True
                if "auto_spawn_backend" not in loaded:
                    loaded["auto_spawn_backend"] = DEFAULT_AUTO_SPAWN_BACKEND
                    needs_save = True
                if "enable_predictive_ghost_input" not in loaded:
                    loaded["enable_predictive_ghost_input"] = DEFAULT_ENABLE_PREDICTIVE_GHOST_INPUT
                    needs_save = True
                if "predictive_prompt_model" not in loaded:
                    loaded["predictive_prompt_model"] = DEFAULT_PREDICTIVE_PROMPT_MODEL
                    needs_save = True
                if "sentence_attribution_model" not in loaded:
                    loaded["sentence_attribution_model"] = DEFAULT_SENTENCE_ATTRIBUTION_MODEL
                    needs_save = True
                if "scholar_polite_email" not in loaded:
                    loaded["scholar_polite_email"] = DEFAULT_SCHOLAR_POLITE_EMAIL
                    needs_save = True
                if "conversion_timeout_seconds" not in loaded:
                    loaded["conversion_timeout_seconds"] = DEFAULT_CONVERSION_TIMEOUT_SECONDS
                    needs_save = True
                if "datalab_merge_cross_page" not in loaded:
                    loaded["datalab_merge_cross_page"] = DEFAULT_DATALAB_MERGE_CROSS_PAGE
                    needs_save = True
                if "user_research_profile" not in loaded:
                    loaded["user_research_profile"] = DEFAULT_USER_RESEARCH_PROFILE
                    needs_save = True
                if "max_research_profile_chars" not in loaded:
                    loaded["max_research_profile_chars"] = DEFAULT_MAX_RESEARCH_PROFILE_CHARS
                    needs_save = True
                if "max_saved_sessions" not in loaded:
                    loaded["max_saved_sessions"] = DEFAULT_MAX_SAVED_SESSIONS
                    needs_save = True
                if "session_retention_days" not in loaded:
                    loaded["session_retention_days"] = DEFAULT_SESSION_RETENTION_DAYS
                    needs_save = True
                if "session_db_auto_vacuum_on_startup" not in loaded:
                    loaded["session_db_auto_vacuum_on_startup"] = DEFAULT_SESSION_DB_AUTO_VACUUM_ON_STARTUP
                    needs_save = True
                if "prune_empty_sessions" not in loaded:
                    loaded["prune_empty_sessions"] = DEFAULT_PRUNE_EMPTY_SESSIONS
                    needs_save = True
                if "backend_port" not in loaded:
                    loaded["backend_port"] = DEFAULT_BACKEND_PORT
                    needs_save = True

                # Backfill Phase 7 onboarding SSOT keys if missing from existing
                # config files on disk (embedding_dimension / sync_auto_abort_failures)
                if "embedding_dimension" not in loaded:
                    loaded["embedding_dimension"] = DEFAULT_CONFIG["embedding_dimension"]
                    needs_save = True
                if "sync_auto_abort_failures" not in loaded:
                    loaded["sync_auto_abort_failures"] = DEFAULT_SYNC_AUTO_ABORT_FAILURES
                    needs_save = True

                if "vlm_captioning_backup_model" not in loaded:
                    loaded["vlm_captioning_backup_model"] = DEFAULT_CONFIG["vlm_captioning_backup_model"]
                    needs_save = True
                if "vlm_captioning_model" not in loaded:
                    loaded["vlm_captioning_model"] = DEFAULT_CONFIG["vlm_captioning_model"]
                    needs_save = True
                if "vlm_captioning_provider" not in loaded:
                    loaded["vlm_captioning_provider"] = DEFAULT_CONFIG["vlm_captioning_provider"]
                    needs_save = True
                if "enable_vlm_captioning" not in loaded:
                    loaded["enable_vlm_captioning"] = DEFAULT_CONFIG["enable_vlm_captioning"]
                    needs_save = True
                if "vlm_max_images_per_doc" not in loaded:
                    loaded["vlm_max_images_per_doc"] = DEFAULT_CONFIG["vlm_max_images_per_doc"]
                    needs_save = True

                # Detect and migrate deprecated compaction keys from disk
                if "compaction_token_threshold" in loaded and "adaptive_compaction_token_ceiling" not in loaded:
                    loaded["adaptive_compaction_token_ceiling"] = loaded["compaction_token_threshold"]
                    needs_save = True
                if "enable_context_compaction" in loaded and "enable_adaptive_compaction" not in loaded:
                    loaded["enable_adaptive_compaction"] = loaded["enable_context_compaction"]
                    needs_save = True
                if "adaptive_decision_model" in loaded and "adaptive_compaction_decision_model" not in loaded:
                    loaded["adaptive_compaction_decision_model"] = loaded.pop("adaptive_decision_model")
                    needs_save = True
                elif "adaptive_decision_model" in loaded:
                    loaded.pop("adaptive_decision_model", None)
                    needs_save = True

                # Purge all 5 deprecated compaction keys
                deprecated_compaction_keys = (
                    "compaction_interval",
                    "compaction_overlap_size",
                    "compaction_token_threshold",
                    "compaction_event_retention_size",
                    "enable_context_compaction",
                )
                for dep_key in deprecated_compaction_keys:
                    if dep_key in loaded:
                        loaded.pop(dep_key, None)
                        needs_save = True

                # Backfill missing adaptive compaction options
                adaptive_compaction_defaults = {
                    "enable_adaptive_compaction": DEFAULT_ENABLE_ADAPTIVE_COMPACTION,
                    "adaptive_compaction_retention_turns": DEFAULT_ADAPTIVE_COMPACTION_RETENTION_TURNS,
                    "adaptive_compaction_min_turns": DEFAULT_ADAPTIVE_COMPACTION_MIN_TURNS,
                    "adaptive_compaction_token_floor": DEFAULT_ADAPTIVE_COMPACTION_TOKEN_FLOOR,
                    "adaptive_compaction_token_ceiling": DEFAULT_ADAPTIVE_COMPACTION_TOKEN_CEILING,
                    "adaptive_compaction_threshold": DEFAULT_ADAPTIVE_COMPACTION_THRESHOLD,
                    "adaptive_compaction_decision_model": DEFAULT_ADAPTIVE_COMPACTION_DECISION_MODEL,
                    "compaction_summarizer_model": DEFAULT_COMPACTION_SUMMARIZER_MODEL,
                }
                for k, default_val in adaptive_compaction_defaults.items():
                    if k not in loaded:
                        loaded[k] = default_val
                        needs_save = True

                # Disallow local Laya for compaction decisions on disk; reset to default remote model
                if is_local_decision_model(loaded.get("adaptive_compaction_decision_model")):
                    loaded["adaptive_compaction_decision_model"] = DEFAULT_ADAPTIVE_COMPACTION_DECISION_MODEL
                    needs_save = True

                # Backfill citation verification options if missing from existing config file on disk
                verification_defaults = {
                    "citation_verification_enabled": DEFAULT_CITATION_VERIFICATION_ENABLED,
                    "citation_verification_provider": DEFAULT_CITATION_VERIFICATION_PROVIDER,
                    "citation_verification_model": DEFAULT_CITATION_VERIFICATION_MODEL,
                    "citation_verification_threshold": DEFAULT_CITATION_VERIFICATION_THRESHOLD,
                    "citation_verification_local_url": DEFAULT_CITATION_VERIFICATION_LOCAL_URL,
                    "citation_verification_local_port": DEFAULT_CITATION_VERIFICATION_LOCAL_PORT,
                    "citation_verification_auto_start_local": DEFAULT_CITATION_VERIFICATION_AUTO_START_LOCAL,
                    "citation_verification_timeout_sec": DEFAULT_CITATION_VERIFICATION_TIMEOUT_SEC,
                    "citation_verification_turn_budget_sec": DEFAULT_CITATION_VERIFICATION_TURN_BUDGET_SEC,
                }
                for k, default_val in verification_defaults.items():
                    if k not in loaded:
                        loaded[k] = default_val
                        needs_save = True
                # Migrate legacy default 3.0s/10.0s timeout and budget to 10.0s/20.0s for CPU inference headroom
                try:
                    if float(loaded.get("citation_verification_timeout_sec", 0)) == 3.0:
                        loaded["citation_verification_timeout_sec"] = DEFAULT_CITATION_VERIFICATION_TIMEOUT_SEC
                        needs_save = True
                except (ValueError, TypeError):
                    pass
                try:
                    if float(loaded.get("citation_verification_turn_budget_sec", 0)) == 10.0:
                        loaded["citation_verification_turn_budget_sec"] = DEFAULT_CITATION_VERIFICATION_TURN_BUDGET_SEC
                        needs_save = True
                except (ValueError, TypeError):
                    pass

                if needs_save:
                    try:
                        with open(config_path, "w") as f_out:
                            json.dump(loaded, f_out, indent=2)
                        stamp_config_mtime(resolved_dir)
                    except Exception as save_err:
                        logger.warning(
                            "Failed to backfill default dialogue agent models to %s: %s",
                            config_path,
                            save_err,
                        )
                    try:
                        stat_res = config_path.stat()
                        current_fingerprint = (stat_res.st_mtime_ns, stat_res.st_size)
                    except OSError:
                        pass

                res = dict(defaults)
                res.update(loaded)
                with _CONFIG_CACHE_LOCK:
                    _CONFIG_CACHE[resolved_path] = copy.deepcopy(res)
                    _CONFIG_CACHE_FINGERPRINT[resolved_path] = current_fingerprint
                return copy.deepcopy(res)
        except Exception as e:
            logger.warning("Failed to load config from %s: %s", config_path, e)
    return defaults

resolve_laya_device

resolve_laya_device()

Determine the optimal and functional execution device for Laya (cuda, mps, xpu, cpu).

Probes actual execution inside the Laya Python environment to guard against CUDA capability mismatches (e.g. sm_61 on PyTorch CUDA >= 7.5 builds) or OOMs.

Source code in zocrates/rag/decision_service.py

def resolve_laya_device() -> str:
    """Determine the optimal and functional execution device for Laya (cuda, mps, xpu, cpu).

    Probes actual execution inside the Laya Python environment to guard against
    CUDA capability mismatches (e.g. sm_61 on PyTorch CUDA >= 7.5 builds) or OOMs.
    """
    env_override = os.environ.get("LAYA_DEVICE")
    if env_override and env_override.strip():
        return env_override.strip().lower()

    py_bin = find_laya_python()
    if not py_bin:
        return "cpu"

    # 1. Probe CUDA execution with an actual tensor operation in an isolated subprocess
    try:
        res = subprocess.run(
            [py_bin, "-c", "import torch; assert torch.cuda.is_available(); (torch.zeros(1).cuda() + 1).item()"],
            capture_output=True,
            text=True,
            timeout=3.0,
        )
        if res.returncode == 0:
            return "cuda"
        logger.info(
            "Laya CUDA probe failed (returncode %d); falling back to alternate devices. Output: %s",
            res.returncode,
            res.stderr.strip() or res.stdout.strip(),
        )
    except Exception as exc:
        logger.info("Laya CUDA probe timed out or errored: %s", exc)

    # 2. Probe MPS (Apple Silicon)
    try:
        res = subprocess.run(
            [py_bin, "-c", "import torch; assert hasattr(torch.backends, 'mps') and torch.backends.mps.is_available(); (torch.zeros(1).to('mps') + 1).item()"],
            capture_output=True,
            text=True,
            timeout=3.0,
        )
        if res.returncode == 0:
            return "mps"
    except Exception:
        pass

    # 3. Probe XPU (Intel GPU)
    try:
        res = subprocess.run(
            [py_bin, "-c", "import torch; assert hasattr(torch, 'xpu') and torch.xpu.is_available(); (torch.zeros(1).to('xpu') + 1).item()"],
            capture_output=True,
            text=True,
            timeout=3.0,
        )
        if res.returncode == 0:
            return "xpu"
    except Exception:
        pass

    return "cpu"

resolve_zotero_data_dir

resolve_zotero_data_dir(zotero_data_dir=None)

Resolve the active Zotero data directory with auto-discovery fallback.

Parameters:

Name	Type	Description	Default
zotero_data_dir	Path | str | None	Explicit data directory path or None.	None

Returns:

Type	Description
Path | None	Resolved Path to the Zotero data directory, or None if not located.

Source code in zocrates/config.py

def resolve_zotero_data_dir(zotero_data_dir: Path | str | None = None) -> Path | None:
    """Resolve the active Zotero data directory with auto-discovery fallback.

    Args:
        zotero_data_dir: Explicit data directory path or None.

    Returns:
        Resolved Path to the Zotero data directory, or None if not located.
    """
    if zotero_data_dir is not None:
        return Path(zotero_data_dir)
    try:
        from zocrates.rag.service_manager import get_syncer
        syncer = get_syncer()
        if syncer and getattr(syncer, "_zotero_data_dir", None):
            return Path(syncer._zotero_data_dir)
    except Exception:
        pass
    if ZOTERO_DATA_DIR:
        return Path(ZOTERO_DATA_DIR)
    if env_dir := os.getenv("ZOTERO_DATA_DIR"):
        return Path(env_dir)
    # Auto-discover active Zotero profile in user home directory (fallback for offline scripts)
    candidates = (
        list(Path.home().glob("Zotero*/zocrates-data"))
        + list(Path.home().glob("Zotero*/*/zocrates-data"))
        + list(Path.home().glob(".var/app/org.zotero.Zotero/data/zotero/zocrates-data"))
        + list(Path.home().glob(".zotero/zotero/*/zocrates-data"))
    )
    if candidates:
        selected = candidates[0].parent
        if len(candidates) > 1:
            logger.warning(
                "Multiple Zotero profiles found with zocrates-data (%s). "
                "Defaulting to %s. Set ZOTERO_DATA_DIR to select a specific profile.",
                [str(c.parent) for c in candidates],
                selected,
            )
        else:
            logger.debug("Auto-discovered Zotero data directory: %s", selected)
        return selected
    return None

start_laya_daemon

start_laya_daemon(port=8005)

Spawn laya-serve in the background.

Source code in zocrates/rag/decision_service.py

def start_laya_daemon(port: int = 8005) -> dict[str, Any]:
    """Spawn laya-serve in the background."""
    global _managed_process, _managed_port

    backend_port = get_backend_port()
    if backend_port is not None and port == backend_port:
        return {
            "status": "error",
            "message": f"Port {port} conflicts with the active Zocrates backend server.",
        }

    target_url = f"http://127.0.0.1:{port}/v1/systemone"
    if check_laya_running(target_url, timeout=0.5):
        return {
            "status": "success",
            "message": f"Laya server is already running on port {port}.",
            "port": port,
        }

    laya_bin = find_laya_serve_binary()
    uv_bin = find_uv_binary()

    if not laya_bin and not uv_bin:
        return {
            "status": "error",
            "message": "Neither `laya-serve` binary nor `uv` tool runner found. Please install Laya first.",
        }

    if laya_bin:
        cmd = [laya_bin]
    else:
        cmd = [uv_bin, "tool", "run", "laya-serve"]

    env = os.environ.copy()
    env["LAYA_HOST"] = "127.0.0.1"
    env["LAYA_PORT"] = str(port)
    env["LAYA_MODELS"] = "english"
    device = resolve_laya_device()
    env["LAYA_DEVICE"] = device
    local_bin = str(Path.home() / ".local" / "bin")
    if local_bin not in env.get("PATH", ""):
        env["PATH"] = f"{local_bin}{os.pathsep}{env.get('PATH', '')}"

    with _daemon_lock:
        if _managed_process is not None and _managed_process.poll() is None:
            return {
                "status": "success",
                "message": f"Laya server is already running on port {_managed_port}.",
                "port": _managed_port,
            }
        _managed_process = None
        _managed_port = None

        logger.info("Spawning Laya daemon: %s on port %s with LAYA_DEVICE=%s", " ".join(cmd), port, device)
        try:
            proc = subprocess.Popen(
                cmd,
                env=env,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                start_new_session=True,
            )
            _managed_process = proc
            _managed_port = port
            try:
                data_dir = get_zocrates_data_dir()
                (data_dir / "laya.pid").write_text(str(proc.pid), encoding="utf-8")
                (data_dir / "laya.port").write_text(str(port), encoding="utf-8")
            except Exception as write_err:
                logger.warning("Could not write laya.pid/laya.port: %s", write_err)
        except Exception as exc:
            logger.error("Failed to spawn Laya daemon: %s", exc)
            return {
                "status": "error",
                "message": f"Failed to spawn Laya daemon: {exc}",
            }

    deadline = time.time() + 30.0
    while time.time() < deadline:
        if proc.poll() is not None:
            try:
                data_dir = get_zocrates_data_dir()
                (data_dir / "laya.pid").unlink(missing_ok=True)
                (data_dir / "laya.port").unlink(missing_ok=True)
            except Exception:
                pass
            return {
                "status": "error",
                "message": f"Laya daemon process terminated unexpectedly with code {proc.poll()}.",
            }
        if check_laya_running(target_url, timeout=0.3):
            return {
                "status": "success",
                "message": f"Laya daemon started successfully on port {port}.",
                "port": port,
            }
        time.sleep(0.3)

    return {
        "status": "success",
        "message": f"Laya daemon process launched (PID: {proc.pid}), awaiting model initialization.",
        "port": port,
    }

stop_laya_daemon

stop_laya_daemon(port=8005)

Terminate the managed daemon or verified Laya process listening on port.

Source code in zocrates/rag/decision_service.py

def stop_laya_daemon(port: int = 8005) -> dict[str, Any]:
    """Terminate the managed daemon or verified Laya process listening on port."""
    global _managed_process, _managed_port

    stopped = False
    with _daemon_lock:
        if _managed_process is not None:
            if _managed_port is None or _managed_port == port:
                try:
                    if _managed_process.poll() is None:
                        _managed_process.terminate()
                        try:
                            _managed_process.wait(timeout=3.0)
                        except subprocess.TimeoutExpired:
                            _managed_process.kill()
                            _managed_process.wait(timeout=1.0)
                        stopped = True
                except Exception as exc:
                    logger.warning("Error stopping managed Laya process: %s", exc)
                finally:
                    _managed_process = None
                    _managed_port = None

    try:
        data_dir = get_zocrates_data_dir()
        pid_file = data_dir / "laya.pid"
        port_file = data_dir / "laya.port"
        if pid_file.exists():
            recorded_port = None
            if port_file.exists():
                try:
                    recorded_port = int(port_file.read_text(encoding="utf-8").strip())
                except Exception:
                    pass

            # Only terminate if port matches recorded port or no specific recorded port
            if recorded_port is None or recorded_port == port:
                try:
                    pid_str = pid_file.read_text(encoding="utf-8").strip()
                    if pid_str.isdigit():
                        pid = int(pid_str)
                        if psutil.pid_exists(pid):
                            try:
                                proc = psutil.Process(pid)
                                if _is_laya_process(proc):
                                    if _terminate_psutil_proc(proc, timeout=3.0):
                                        stopped = True
                                        pid_file.unlink(missing_ok=True)
                                        port_file.unlink(missing_ok=True)
                                else:
                                    logger.warning(
                                        "Process PID %d recorded in laya.pid is not a Laya process; refusing to terminate.",
                                        pid,
                                    )
                                    pid_file.unlink(missing_ok=True)
                                    port_file.unlink(missing_ok=True)
                            except (psutil.NoSuchProcess, psutil.ZombieProcess):
                                pid_file.unlink(missing_ok=True)
                                port_file.unlink(missing_ok=True)
                            except Exception as exc:
                                logger.warning("Error terminating process %d from laya.pid: %s", pid, exc)
                        else:
                            pid_file.unlink(missing_ok=True)
                            port_file.unlink(missing_ok=True)
                except Exception as read_err:
                    logger.warning("Error reading laya.pid: %s", read_err)
        elif port_file.exists():
            port_file.unlink(missing_ok=True)
    except Exception as exc:
        logger.warning("Error checking or cleaning laya.pid: %s", exc)

    target_url = f"http://127.0.0.1:{port}/v1/systemone"
    if check_laya_running(target_url, timeout=0.3):
        try:
            for p in psutil.process_iter(["pid", "name", "cmdline"]):
                try:
                    if _is_laya_process(p) and _process_is_on_port(p, port):
                        logger.info("Terminating Laya process %d found listening on port %d", p.pid, port)
                        if _terminate_psutil_proc(p, timeout=3.0):
                            stopped = True
                except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
                    continue
        except Exception as exc:
            logger.warning("Error inspecting processes listening on port %d: %s", port, exc)

    return {
        "status": "success",
        "message": "Laya daemon stopped." if stopped else "Laya daemon was not running.",
        "port": port,
    }

test_decision_model_connectivity

test_decision_model_connectivity(provider='openrouter', model=None, local_url=None, openrouter_api_key=None)

Test connectivity to decision model (remote OpenRouter or local Laya).

Source code in zocrates/rag/decision_service.py

def test_decision_model_connectivity(
    provider: str = "openrouter",
    model: str | None = None,
    local_url: str | None = None,
    openrouter_api_key: str | None = None,
) -> dict[str, Any]:
    """Test connectivity to decision model (remote OpenRouter or local Laya)."""
    cfg = load_zocrates_config()
    provider = (provider or cfg.get("citation_verification_provider", "openrouter")).lower()

    source_text = "Transformers rely entirely on self-attention mechanisms without recurrence or convolution."
    claim = "Transformers eliminate recurrence in favor of self-attention mechanisms."

    t0 = time.perf_counter()

    if provider == "local_laya":
        effective_url = local_url or cfg.get("citation_verification_local_url", DEFAULT_CITATION_VERIFICATION_LOCAL_URL)
        effective_model = "laya"
        parsed = urllib.parse.urlparse(effective_url)
        health_url = f"{parsed.scheme}://{parsed.netloc}/health"
        if not check_laya_running(effective_url, timeout=1.0):
            return {
                "ok": False,
                "status": "error",
                "provider": "local_laya",
                "message": f"Local Laya server is not reachable at {effective_url} (probe failed on {health_url}). Make sure the daemon is running.",
            }

        from zocrates.agents.library_search.citation_verifier import _check_support_local_laya

        prob = _check_support_local_laya(
            claim=claim,
            source_text=source_text,
            model=effective_model,
            local_url=effective_url,
            timeout=8.0,
            skip_health_probe=True,
        )
        latency_ms = (time.perf_counter() - t0) * 1000.0

        if prob is not None:
            return {
                "ok": True,
                "status": "ok",
                "provider": "local_laya",
                "model": effective_model,
                "local_url": effective_url,
                "support_probability": prob,
                "latency_ms": round(latency_ms, 1),
                "message": f"Successfully verified test claim with local Laya (support: {prob:.1%}, latency: {latency_ms:.0f}ms)",
            }
        else:
            return {
                "ok": False,
                "status": "error",
                "provider": "local_laya",
                "message": (
                    f"Laya server at {effective_url} failed to evaluate test decision (HTTP 500 inference failed). "
                    "This typically occurs if the GPU compute capability is unsupported (e.g. sm_61) or VRAM is exhausted. "
                    "Restart Laya with LAYA_DEVICE=cpu (or click 'Start Local Server' in Preferences)."
                ),
            }

    else:
        effective_model = model or cfg.get("citation_verification_model", DEFAULT_CITATION_VERIFICATION_MODEL)
        effective_key = (
            openrouter_api_key
            or get_secure_api_key("openrouter_api_key", ("OPENROUTER_API_KEY", "OPEN_ROUTER_API_KEY"))
            or os.environ.get("OPENROUTER_API_KEY")
        )
        if not effective_key:
            return {
                "ok": False,
                "status": "error",
                "provider": "openrouter",
                "message": "OpenRouter API key is missing. Please provide a key in OpenRouter settings.",
            }

        from zocrates.agents.library_search.citation_verifier import _check_support_openrouter

        prob = _check_support_openrouter(
            claim=claim,
            source_text=source_text,
            model=effective_model,
            api_key=effective_key,
            timeout=6.0,
        )
        latency_ms = (time.perf_counter() - t0) * 1000.0

        if prob is not None:
            return {
                "ok": True,
                "status": "ok",
                "provider": "openrouter",
                "model": effective_model,
                "support_probability": prob,
                "latency_ms": round(latency_ms, 1),
                "message": f"Successfully verified test claim with OpenRouter ({effective_model}, support: {prob:.1%}, latency: {latency_ms:.0f}ms)",
            }
        else:
            return {
                "ok": False,
                "status": "error",
                "provider": "openrouter",
                "message": f"OpenRouter decision model ({effective_model}) failed to respond or returned an error.",
            }