Spaces:

User1342
/

distinct

Sleeping

initial commit

2aa8b3a about 2 months ago

8.45 kB

"""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)

	"""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)