"""What a double-clicked executable asks, and what it remembers. **The problem this solves.** A worker is started with a line like:: distinct-agent --server https://... --pair K4-7... --approve --allow-remote-server which is fine for somebody who reads a README and lives in a terminal, and is an immediate dead end for the person this is actually asking for help: they downloaded one file and double-clicked it. A program that answers that with ``usage: distinct-agent [-h] ...`` has told them nothing they can act on. So a run with no arguments asks, in order, the three things it cannot work out for itself, remembers the two that are worth remembering, and starts. **What is saved and what is not.** The server address and the worker's own name are saved, because they are the same next week and typing them again is a tax. The pairing code is **not** saved and must not be: it is one-use, it lives ten minutes, and the transport asks the server for a fresh pairing on every start. Writing a dead code into a file on somebody's disk would achieve nothing except to put a credential-shaped string somewhere it does not belong. A stored credential *could* remove the second question. ``ServerRegistry`` already keeps a per-server secret, and a resume path that used it would let a worker rejoin silently. That is a change to the pairing protocol rather than to this screen, it is the kind of change that quietly weakens a handshake if it is made in passing, and it is not made here. The consequence is stated on the screen instead: the code is asked for each time, and the screen says why. The questions themselves are in `distinct_agent/tui.py`, because they are a screen. This module is the part that has to be right whether or not anybody is looking at it: where the file lives, what goes in it, and what command line it turns into. **Settings become command-line arguments, not a second code path.** :func:`to_argv` turns the saved answers into exactly the flags a person would have typed, and ``cli.main`` parses them the way it parses anything else. So every validation, every refusal and every safety check that applies to a typed command applies identically to a double-clicked one. The alternative, threading a settings object into ``main`` beside ``argv``, would create a second way to configure a worker and therefore a second place for a check to be missing. """ from __future__ import annotations import json import os from dataclasses import asdict, dataclass from pathlib import Path from typing import Any, Mapping, Optional, Sequence #: An operator who wants the file somewhere else says so. Mostly useful for #: tests, which must never read or write the real one. CONFIG_ENV_VAR = "DISTINCT_WORKER_CONFIG" #: The file's own version, so a later format can recognise an earlier one #: rather than guessing from which keys happen to be present. CONFIG_VERSION = 1 CONFIG_FILENAME = "worker.json" def config_path(environ: Optional[Mapping[str, str]] = None) -> Path: """Where this machine keeps its answers. Per user rather than beside the executable, because the executable may live in Downloads, in Program Files, or on a memory stick, and none of those is a place a program should be writing to. """ environ = os.environ if environ is None else environ override = environ.get(CONFIG_ENV_VAR) if override: return Path(override).expanduser() if os.name == "nt": base = environ.get("APPDATA") or environ.get("LOCALAPPDATA") root = Path(base) if base else Path.home() / "AppData" / "Roaming" return root / "distinct" / CONFIG_FILENAME base = environ.get("XDG_CONFIG_HOME") root = Path(base) if base else Path.home() / ".config" return root / "distinct" / CONFIG_FILENAME @dataclass(frozen=True) class WorkerSettings: """The answers, and nothing that is a secret. Every field here is something the operator typed or chose, and every one of them is visible on the screen while the worker runs. Nothing in this object would matter if the file were read by somebody else, which is the property that lets it be an ordinary file with ordinary permissions. """ server: str = "" name: str = "" #: Approving in advance is what lets an unattended worker start. It is #: recorded because a person who chose it once meant it, and asked again on #: the setup screen rather than assumed. approve: bool = False #: Off by default and asked separately, exactly as the two flags are. allow_remote: bool = False allow_plaintext: bool = False models: tuple[str, ...] = () def is_complete(self) -> bool: """Whether this is enough to start a worker without asking again.""" return bool(self.server.strip()) def load(path: Optional[Path] = None, *, environ: Optional[Mapping[str, str]] = None) -> WorkerSettings: """Read the saved answers, or return the defaults. Never raises. A missing file is the first run; a corrupt one is a first run with a bad file in the way, and the right response to both is to ask the questions again. Refusing to start because a settings file was truncated by a power cut would be a program choosing its own convenience over the person's. """ path = config_path(environ) if path is None else path try: document = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError): return WorkerSettings() if not isinstance(document, Mapping): return WorkerSettings() models = document.get("models") return WorkerSettings( server=str(document.get("server", "") or ""), name=str(document.get("name", "") or ""), approve=bool(document.get("approve", False)), allow_remote=bool(document.get("allow_remote", False)), allow_plaintext=bool(document.get("allow_plaintext", False)), models=tuple(str(item) for item in models) if isinstance(models, Sequence) and not isinstance(models, (str, bytes)) else (), ) def save( settings: WorkerSettings, path: Optional[Path] = None, *, environ: Optional[Mapping[str, str]] = None, ) -> Optional[Path]: """Write the answers, and say where, or return None having failed quietly. A worker that cannot write its settings still works; it just asks again next time. That is a small annoyance, and it is a much better outcome than refusing to start because a directory was read-only. """ path = config_path(environ) if path is None else path document: dict[str, Any] = {"version": CONFIG_VERSION} document.update(asdict(settings)) document["models"] = list(settings.models) try: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8") except OSError: return None return path def to_argv(settings: WorkerSettings, pairing_code: str) -> list[str]: """The command line this would have been, had somebody typed it. One code path. `cli.main` sees arguments and cannot tell whether a person or this screen produced them, which is the point: the checks that protect a typed command protect a double-clicked one identically. """ argv = ["--server", settings.server.strip(), "--pair", pairing_code.strip()] if settings.name.strip(): argv += ["--name", settings.name.strip()] if settings.approve: argv.append("--approve") if settings.allow_remote: argv.append("--allow-remote-server") if settings.allow_plaintext: argv.append("--allow-plaintext-server") for model_id in settings.models: argv += ["--models", model_id] return argv def needs_remote_permission(server: str) -> bool: """Whether this address is somewhere other than this machine. Asked so the screen can explain the choice at the moment it matters, rather than making somebody discover the flag from a refusal. """ from urllib.parse import urlsplit from distinct_protocol.netpolicy import is_loopback_host host = urlsplit(server.strip()).hostname or "" return bool(host) and not is_loopback_host(host) def needs_plaintext_permission(server: str) -> bool: """Whether this address would send prompts in the clear.""" return server.strip().lower().startswith("http://") and needs_remote_permission(server)