Skip to content

Python reference

Generated from the source. The prose explanation of how the pieces fit is in Architecture; this is the detail.

Only the modules worth calling from outside are listed. The engine's internals are documented in place — every module has a docstring saying what it is for — but a reference page for two hundred rule functions would be a worse way to read them than the source.

Paths and configuration

citar.paths

Where CITAR keeps its files.

Every directory CITAR reads or writes is resolved here, once, so that the same code works whether it was started from a git checkout or installed from a wheel.

Two kinds of location

Package data ships inside the wheel and is read-only: the ruleset JSON, the browser client, the database migrations, the hardware-collector scripts. It always sits next to this file.

State is everything CITAR writes: saved games, the server registry, benchmark runs, lab results, reports, the usage ledger, the accounts database. Where that goes depends on how CITAR was started:

git checkout      <checkout>/saves, <checkout>/config, <checkout>/benchmarks
installed         a per-user state directory (see ``state_dir``)

The checkout case matters because it is the development workflow: a contributor's games, registry and benchmark history sit beside the code where they can be inspected, diffed and deleted. The installed case matters because site-packages is the wrong place to write a save file and is often not writable at all — on Windows an installer puts it under Program Files, and on Linux a system-wide install is owned by root.

Overrides

Environment variables win over both, and are how the systemd unit points a deployment at /var/lib/citar:

====================== ==================================================================== CITAR_STATE_DIR Base for all state. Sets saves, config and benchmarks in one go. CITAR_SAVE_DIR Saved games, and everything filed under them (maps, scenarios, probes, lab, reports, usage, balance). CITAR_CONFIG_DIR The server registry (servers.json) and its siblings. CITAR_BENCH_DIR Benchmark suites and runs. CITAR_DATA_DIR Private state: the accounts database and the generated secret key. Never the checkout, even in development — see data_dir. ====================== ====================================================================

Nothing here creates a directory on import. Call ensure (or the *_dir helpers, which create on demand) so that importing :mod:citar.paths stays free of side effects — tests and the --help path both rely on that.

package_data 

package_data() -> Path

The ruleset and other JSON shipped with the package (citar/data).

Source code in citar/paths.py 
65
66
67
def package_data() -> Path:
    """The ruleset and other JSON shipped with the package (``citar/data``)."""
    return PACKAGE / "data"

web_dir 

web_dir() -> Path

The browser client served as static files (citar/web).

Source code in citar/paths.py 
70
71
72
def web_dir() -> Path:
    """The browser client served as static files (``citar/web``)."""
    return PACKAGE / "web"

migrations_dir 

migrations_dir() -> Path

The Alembic migration environment (citar/migrations).

Source code in citar/paths.py 
75
76
77
def migrations_dir() -> Path:
    """The Alembic migration environment (``citar/migrations``)."""
    return PACKAGE / "migrations"

collectors_dir 

collectors_dir() -> Path

Hardware-collector scripts offered for download on the Servers page.

Source code in citar/paths.py 
80
81
82
def collectors_dir() -> Path:
    """Hardware-collector scripts offered for download on the Servers page."""
    return PACKAGE / "data" / "collectors"

in_source_checkout 

in_source_checkout() -> bool

True when CITAR is running from a source tree that it may write to.

Both halves matter. A marker file alone is not enough — a checkout mounted read-only, or one owned by another user, would pass that test and then fail on the first save. So the directory is probed for writability as well, and anything unexpected counts as "not a checkout", which falls back to the per-user directory and always works.

Source code in citar/paths.py 
87
88
89
90
91
92
93
94
95
96
97
def in_source_checkout() -> bool:
    """True when CITAR is running from a source tree that it may write to.

    Both halves matter. A marker file alone is not enough — a checkout mounted read-only, or one
    owned by another user, would pass that test and then fail on the first save. So the directory is
    probed for writability as well, and anything unexpected counts as "not a checkout", which falls
    back to the per-user directory and always works.
    """
    if not any((ROOT / marker).exists() for marker in _CHECKOUT_MARKERS):
        return False
    return os.access(ROOT, os.W_OK)

state_dir 

state_dir() -> Path

The base directory for everything CITAR writes.

CITAR_STATE_DIR wins; then a writable source checkout; then the per-user directory that :func:data_dir also uses. The result is not created here — the callers below do that for the specific subdirectory they need.

Source code in citar/paths.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
def state_dir() -> Path:
    """The base directory for everything CITAR writes.

    ``CITAR_STATE_DIR`` wins; then a writable source checkout; then the per-user directory that
    :func:`data_dir` also uses. The result is not created here — the callers below do that for the
    specific subdirectory they need.
    """
    override = _env("CITAR_STATE_DIR")
    if override:
        return Path(override).expanduser()
    if in_source_checkout():
        return ROOT
    return _user_base()

ensure 

ensure(path: Path) -> Path

Create path (and its parents) if it does not exist, and return it.

Source code in citar/paths.py 
133
134
135
136
def ensure(path: Path) -> Path:
    """Create *path* (and its parents) if it does not exist, and return it."""
    path.mkdir(parents=True, exist_ok=True)
    return path

save_dir 

save_dir() -> Path

Saved games, and the folders filed beneath them (maps, scenarios, probes, lab, reports).

Source code in citar/paths.py 
139
140
141
142
def save_dir() -> Path:
    """Saved games, and the folders filed beneath them (maps, scenarios, probes, lab, reports)."""
    override = _env("CITAR_SAVE_DIR")
    return ensure(Path(override).expanduser() if override else state_dir() / "saves")

config_dir 

config_dir() -> Path

The server registry and other operator-edited configuration.

Source code in citar/paths.py 
145
146
147
148
def config_dir() -> Path:
    """The server registry and other operator-edited configuration."""
    override = _env("CITAR_CONFIG_DIR")
    return ensure(Path(override).expanduser() if override else state_dir() / "config")

bench_dir 

bench_dir() -> Path

Benchmark suites, runs and scoring settings.

CITAR_SAVE_DIR is honoured as a fallback so that a test or deployment which redirects saves does not leave benchmark runs behind in a different tree; this was the existing behaviour and tests depend on it.

Source code in citar/paths.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def bench_dir() -> Path:
    """Benchmark suites, runs and scoring settings.

    ``CITAR_SAVE_DIR`` is honoured as a fallback so that a test or deployment which redirects saves
    does not leave benchmark runs behind in a different tree; this was the existing behaviour and
    tests depend on it.
    """
    override = _env("CITAR_BENCH_DIR")
    if override:
        return ensure(Path(override).expanduser())
    saves = _env("CITAR_SAVE_DIR")
    if saves:
        return ensure(Path(saves).expanduser().parent / "benchmarks")
    return ensure(state_dir() / "benchmarks")

data_dir 

data_dir() -> Path

Private state: the accounts database and the locally generated secret key.

This is deliberately not the checkout, even in development. The project folder may sync to OneDrive or Dropbox, and a sync client copying a SQLite file mid-write corrupts it; password hashes and session tokens do not belong in a synced folder either.

An explicit CITAR_STATE_DIR is honoured, because somebody who names one directory for all of CITAR's state means this too — that is how the systemd unit points a deployment at /var/lib/citar. What never happens is falling into a source checkout by accident.

Source code in citar/paths.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def data_dir() -> Path:
    """Private state: the accounts database and the locally generated secret key.

    This is deliberately *not* the checkout, even in development. The project folder may sync to
    OneDrive or Dropbox, and a sync client copying a SQLite file mid-write corrupts it; password
    hashes and session tokens do not belong in a synced folder either.

    An explicit ``CITAR_STATE_DIR`` is honoured, because somebody who names one directory for all of
    CITAR's state means this too — that is how the systemd unit points a deployment at
    ``/var/lib/citar``. What never happens is falling into a source checkout by accident.
    """
    if _env("CITAR_DATA_DIR"):
        return ensure(Path(_env("CITAR_DATA_DIR")).expanduser())
    if _env("CITAR_STATE_DIR"):
        return ensure(Path(_env("CITAR_STATE_DIR")).expanduser())
    return ensure(_user_base())

sub 

sub(name: str) -> Path

A named subdirectory of :func:save_dir, created on demand.

sub("lab") is how the lab, probes, reports, usage ledger and balance reports find their homes, so that redirecting CITAR_SAVE_DIR moves all of them together.

Source code in citar/paths.py 
185
186
187
188
189
190
191
def sub(name: str) -> Path:
    """A named subdirectory of :func:`save_dir`, created on demand.

    ``sub("lab")`` is how the lab, probes, reports, usage ledger and balance reports find their
    homes, so that redirecting ``CITAR_SAVE_DIR`` moves all of them together.
    """
    return ensure(save_dir() / name)

saves_path 

saves_path(*parts: str) -> Path

A path under the save directory, without creating anything.

Source code in citar/paths.py 
201
202
203
204
205
def saves_path(*parts: str) -> Path:
    """A path under the save directory, without creating anything."""
    override = _env("CITAR_SAVE_DIR")
    base = Path(override).expanduser() if override else state_dir() / "saves"
    return base.joinpath(*parts)

config_path 

config_path(*parts: str) -> Path

A path under the configuration directory, without creating anything.

Source code in citar/paths.py 
208
209
210
211
212
def config_path(*parts: str) -> Path:
    """A path under the configuration directory, without creating anything."""
    override = _env("CITAR_CONFIG_DIR")
    base = Path(override).expanduser() if override else state_dir() / "config"
    return base.joinpath(*parts)

bench_path 

bench_path(*parts: str) -> Path

A path under the benchmark directory, without creating anything.

Source code in citar/paths.py 
215
216
217
218
219
220
221
222
223
def bench_path(*parts: str) -> Path:
    """A path under the benchmark directory, without creating anything."""
    override = _env("CITAR_BENCH_DIR")
    if override:
        base = Path(override).expanduser()
    else:
        saves = _env("CITAR_SAVE_DIR")
        base = Path(saves).expanduser().parent / "benchmarks" if saves else state_dir() / "benchmarks"
    return base.joinpath(*parts)

describe 

describe() -> dict[str, str]

Every resolved location, for citar doctor and bug reports.

Source code in citar/paths.py 
226
227
228
229
230
231
232
233
234
235
236
def describe() -> dict[str, str]:
    """Every resolved location, for ``citar doctor`` and bug reports."""
    return {
        "package": str(PACKAGE),
        "source checkout": "yes" if in_source_checkout() else "no (installed)",
        "state": str(state_dir()),
        "saves": str(save_dir()),
        "config": str(config_dir()),
        "benchmarks": str(bench_dir()),
        "private data": str(data_dir()),
    }

citar.settings

Runtime settings for the CITAR server, read from the environment once at import.

CITAR runs in one of two modes:

local     (default) one operator on their own machine. The server binds to loopback, mints a
          session for a single owner account automatically, and generates its own secret key.
          This is the laptop workflow: no login screen, no TLS, no OAuth registration.
server    the public deployment. Sessions are real, TLS is required, and the process refuses to
          start without a secret key and a public origin.

Every authorization check runs the same way in both modes — local mode is a real account with a real session, not a bypass — so a permission bug cannot hide locally and surface in production.

Where things live

The project folder syncs to OneDrive, which is fine for code and game saves but not for a live SQLite database: the sync client copies files mid-write and a WAL database does not survive that. It is also the wrong place for password hashes and session tokens. So state that must be both durable and private goes to a per-user data directory outside the project:

Windows   %LOCALAPPDATA%\CITAR
macOS     ~/Library/Application Support/CITAR
Linux     $XDG_DATA_HOME/citar  (default ~/.local/share/citar)

Override with CITAR_DATA_DIR. On the VPS this is a normal directory under the service account, and it is the only thing that needs backing up besides saves/.

Secrets come from the environment, never from a file in the project folder. citar/env.example lists every variable; on the VPS they are set in the systemd unit's EnvironmentFile with mode 0600.

SettingsError

Bases: RuntimeError

A misconfiguration serious enough that the server must not start.

Source code in citar/settings.py 
47
48
class SettingsError(RuntimeError):
    """A misconfiguration serious enough that the server must not start."""

data_dir 

data_dir() -> Path

The private state directory (database, generated secret key). Created on demand.

Resolved by :func:citar.paths.data_dir, which every other directory also goes through, so there is one answer to "where does CITAR put things" rather than one per module.

Source code in citar/settings.py 
75
76
77
78
79
80
81
82
83
def data_dir() -> Path:
    """The private state directory (database, generated secret key). Created on demand.

    Resolved by :func:`citar.paths.data_dir`, which every other directory also goes through, so
    there is one answer to "where does CITAR put things" rather than one per module.
    """
    from . import paths

    return paths.data_dir()

get 

get() -> Settings

The process-wide settings, loaded on first use.

Source code in citar/settings.py 
339
340
341
342
343
344
def get() -> Settings:
    """The process-wide settings, loaded on first use."""
    global _settings
    if _settings is None:
        _settings = load()
    return _settings

reset 

reset()

Forget the cached settings, so the next read picks up a changed environment.

For tests, and for the setup wizard. Not something a running server does.

Source code in citar/settings.py 
354
355
356
357
358
359
360
def reset():
    """Forget the cached settings, so the next read picks up a changed environment.

    For tests, and for the setup wizard. Not something a running server does.
    """
    global _settings
    _settings = None

The command line

citar.cli

The citar command.

One entry point in front of everything CITAR can do, so that an installed copy needs no knowledge of the module layout::

citar                     start the server and open the browser
citar serve               the same, without opening a browser
citar setup               interactive first-time configuration
citar doctor              check this installation and print what it found
citar admin …             operator commands (accounts, invitations, worker tokens)
citar worker …            run the worker agent that serves local models to a remote CITAR
citar mcp …               the MCP bridge, for Claude Code and other MCP clients
citar bench …             command-line model benchmark
citar sim …               one headless bot-vs-bot game in the console
citar balance …           parallel bot games, for balancing the scripted bot
citar lab …               the long-running bot experiment runner
citar where               print every directory CITAR reads or writes

Sub-commands are thin: each one hands its remaining arguments to the module that already owned that interface, so citar bench --all-models and python -m citar.bench --all-models are the same program. python -m citar.server also still works, because the deployed systemd unit invokes it that way and a packaging change should not require touching a running server.

main 

main(argv: Optional[Sequence[str]] = None) -> int

Dispatch a citar command line. Returns the process exit status.

Source code in citar/cli.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def main(argv: Optional[Sequence[str]] = None) -> int:
    """Dispatch a ``citar`` command line. Returns the process exit status."""
    argv = list(sys.argv[1:] if argv is None else argv)

    # No arguments is the home-user path: start the server and open the lobby. Anything that looks
    # like an option (``citar --version``) still goes through argparse.
    if not argv:
        return _serve([], open_browser=True)

    command, rest = argv[0], argv[1:]

    if command in DELEGATES:
        module_path, attr, _ = DELEGATES[command]
        return _delegate(module_path, attr, f"citar {command}", rest)
    if command == "serve":
        return _serve(rest, open_browser=False)
    if command == "setup":
        from .wizard import cli as wizard_cli

        return wizard_cli.main(rest)
    if command == "doctor":
        from .doctor import main as doctor_main

        return doctor_main(rest)
    if command == "where":
        return _where()

    # Not a command: either an option for the top-level parser (``--version``, ``--help``) or a
    # mistake. argparse prints the right thing and exits in both cases.
    _build_parser().parse_args(argv)
    return 0

citar.doctor

citar doctor - check an installation and explain anything that is wrong.

The point of this command is to turn "it doesn't work" into a specific sentence. Every check prints one line: what was tested, what was found, and - when something is wrong - the one thing to do about it. Nothing here changes state, so it is safe to run against a live server, and its output is what a bug report should contain.

Checks, in the order a failure would stop you:

  1. Python and CITAR versions, and whether this is a checkout or an installed copy.
  2. Dependencies - every import CITAR needs, named individually rather than as one traceback.
  3. Directories - where state lives and whether it is writable.
  4. Configuration - the mode, and in server mode the settings that must be present.
  5. Database - that it opens and its schema is current.
  6. Ruleset - that the packaged data loads and how much of it there is.
  7. Model providers - every server in the registry that CITAR can reach right now.
  8. Port - whether the configured port is free, or already has a CITAR on it.

Report

Collects check results and decides the exit status.

Exit status is what a health check or an installer looks at: 0 when everything that matters works, 1 when something is actually broken. Warnings never fail the run - a missing optional provider is a fact about the setup, not a fault in it.

Source code in citar/doctor.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class Report:
    """Collects check results and decides the exit status.

    Exit status is what a health check or an installer looks at: ``0`` when everything that matters
    works, ``1`` when something is actually broken. Warnings never fail the run - a missing optional
    provider is a fact about the setup, not a fault in it.
    """

    def __init__(self) -> None:
        self.failures = 0
        self.warnings = 0

    def line(self, status: str, label: str, detail: str = "", fix: str = "") -> None:
        """Print one check's result, and count it."""
        print(f"[{status}] {label}" + (f": {detail}" if detail else ""))
        if fix:
            print(f"         -> {fix}")
        if status is FAIL:
            self.failures += 1
        elif status is WARN:
            self.warnings += 1

    def section(self, title: str) -> None:
        """Print a section heading."""
        print(f"\n{title}\n{'-' * len(title)}")

line 

line(
    status: str, label: str, detail: str = "", fix: str = ""
) -> None

Print one check's result, and count it.

Source code in citar/doctor.py 
68
69
70
71
72
73
74
75
76
def line(self, status: str, label: str, detail: str = "", fix: str = "") -> None:
    """Print one check's result, and count it."""
    print(f"[{status}] {label}" + (f": {detail}" if detail else ""))
    if fix:
        print(f"         -> {fix}")
    if status is FAIL:
        self.failures += 1
    elif status is WARN:
        self.warnings += 1

section 

section(title: str) -> None

Print a section heading.

Source code in citar/doctor.py 
78
79
80
def section(self, title: str) -> None:
    """Print a section heading."""
    print(f"\n{title}\n{'-' * len(title)}")

main 

main(argv: Optional[Sequence[str]] = None) -> int

Run every check and return 0 when nothing is broken.

Source code in citar/doctor.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
def main(argv: Optional[Sequence[str]] = None) -> int:
    """Run every check and return ``0`` when nothing is broken."""
    import argparse

    parser = argparse.ArgumentParser(prog="citar doctor", description=__doc__.splitlines()[0])
    parser.add_argument("--quiet", action="store_true", help="only print problems")
    # None means "read the command line" (``python -m citar.doctor``); an empty list is a bare
    # ``citar doctor``, which has passed its own arguments along already.
    args = parser.parse_args(sys.argv[1:] if argv is None else list(argv))

    # The Windows console is often not UTF-8, and a diagnostic that dies on its own output is
    # worse than useless. Everything printed below is ASCII; this makes the rest survive too.
    try:
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
    except (AttributeError, OSError):
        pass

    report = Report()
    if args.quiet:
        original = report.line

        def quiet_line(status, label, detail="", fix=""):
            """Print only the checks that are not passing."""
            if status is not OK:
                original(status, label, detail, fix)

        report.line = quiet_line                                # type: ignore[method-assign]
        report.section = lambda title: None                     # type: ignore[assignment]

    print(f"CITAR {__version__} - checking this installation\n")
    for check in (_check_versions, _check_dependencies, _check_directories, _check_configuration,
                  _check_database, _check_ruleset, _check_providers, _check_port):
        try:
            check(report)
        except Exception as exc:
            report.line(FAIL, check.__name__.lstrip("_"), f"check crashed: {type(exc).__name__}: {exc}")

    print()
    if report.failures:
        print(f"{report.failures} problem(s) found" +
              (f", {report.warnings} warning(s)" if report.warnings else "") + ".")
        return 1
    if report.warnings:
        print(f"No problems. {report.warnings} warning(s) - see above.")
    else:
        print("Everything checks out.")
    return 0

Setup

citar.wizard

First-time setup: work out what this machine can do, then configure CITAR for it.

The wizard exists because CITAR spans two very different installations. Somebody who wants to play a few games against a model on their own PC needs a model endpoint and nothing else — no domain, no TLS, no accounts. Somebody standing up a public server needs all of that and will get a broken site if any one piece is missing. Asking every question to everybody guarantees that both audiences answer questions that do not apply to them, so the first question decides which of the three flows runs and the rest follow from it:

citar.wizard.local one person, this computer, models on localhost or an API key. citar.wizard.server a public deployment: domain, TLS, sign-in, the first administrator. citar.wizard.worker this machine's GPU serving models to a CITAR server somewhere else.

Two rules hold across all three:

Nothing is written until the end. Each flow collects a plan, shows it, and asks once. A wizard that edits as it goes leaves a half-configured machine behind when somebody presses Ctrl-C.

Every question has a non-interactive equivalent. The installers (install.sh, the Windows installer, the Docker image) run the same code with --non-interactive and flags, so there is one implementation of "set CITAR up", not one per packaging format.

citar.wizard.detect

What can this machine do? — the questions the wizard answers before asking any.

Setup goes badly when the software makes the person describe their own computer. Everything here is something CITAR can find out for itself: which model servers are already running, what the GPU is, how much of a model will fit in it. A wizard that opens with "I found LM Studio with 6 models" is a different experience from one that opens with "enter your base URL".

Nothing in this module changes anything, and every probe has a short timeout: a model server that is not running should cost the wizard a fraction of a second, not a stall on a dead TCP port.

Endpoint dataclass

A model server found (or not found) on this machine.

Source code in citar/wizard/detect.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@dataclass
class Endpoint:
    """A model server found (or not found) on this machine."""

    provider: str
    label: str
    base_url: str
    install_url: str
    reachable: bool = False
    models: list[str] = field(default_factory=list)
    error: str = ""

    @property
    def summary(self) -> str:
        """This endpoint in one line, as the wizard shows it."""
        if not self.reachable:
            return f"{self.label}: not running"
        if not self.models:
            return f"{self.label}: running, but no model is loaded"
        head = ", ".join(self.models[:3])
        more = f" (+{len(self.models) - 3} more)" if len(self.models) > 3 else ""
        return f"{self.label}: {len(self.models)} model(s) - {head}{more}"

summary property 

summary: str

This endpoint in one line, as the wizard shows it.

port_open 

port_open(
    host: str, port: int, timeout: float = 0.35
) -> bool

Is something listening? Checked first, because a closed port answers instantly.

An HTTP request to a closed port takes as long as the connect timeout, and the wizard probes half a dozen of them. Testing the socket first keeps detection under a second in the normal case, where none of them are running.

Source code in citar/wizard/detect.py 
66
67
68
69
70
71
72
73
74
75
76
77
def port_open(host: str, port: int, timeout: float = 0.35) -> bool:
    """Is something listening? Checked first, because a closed port answers instantly.

    An HTTP request to a closed port takes as long as the connect timeout, and the wizard probes
    half a dozen of them. Testing the socket first keeps detection under a second in the normal
    case, where none of them are running.
    """
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False

probe 

probe(
    provider: str,
    label: str,
    base_url: str,
    install_url: str = "",
    timeout: float = 2.0,
) -> Endpoint

Ask one endpoint what models it has. Never raises.

Source code in citar/wizard/detect.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def probe(provider: str, label: str, base_url: str, install_url: str = "",
          timeout: float = 2.0) -> Endpoint:
    """Ask one endpoint what models it has. Never raises."""
    endpoint = Endpoint(provider, label, base_url, install_url)
    host, port = _split(base_url)
    if not port_open(host, port):
        return endpoint
    try:
        import httpx

        response = httpx.get(base_url.rstrip("/") + "/models", timeout=timeout)
        if response.status_code >= 400:
            endpoint.error = f"HTTP {response.status_code}"
            return endpoint
        payload = response.json() or {}
        endpoint.reachable = True
        endpoint.models = [m.get("id", "") for m in payload.get("data", []) if m.get("id")]
    except Exception as exc:
        endpoint.error = f"{type(exc).__name__}: {exc}"
    return endpoint

find_endpoints 

find_endpoints(
    include_unreachable: bool = False,
) -> list[Endpoint]

Probe every well-known local model server.

Duplicates are collapsed by base URL: Ollama's OpenAI-compatible endpoint and LM Studio can be configured onto the same port, and offering the same URL twice under two names would be a confusing way to start.

Source code in citar/wizard/detect.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def find_endpoints(include_unreachable: bool = False) -> list[Endpoint]:
    """Probe every well-known local model server.

    Duplicates are collapsed by base URL: Ollama's OpenAI-compatible endpoint and LM Studio can be
    configured onto the same port, and offering the same URL twice under two names would be a
    confusing way to start.
    """
    from concurrent.futures import ThreadPoolExecutor

    targets = []
    seen: set[str] = set()
    for provider, label, base_url, install_url in KNOWN_ENDPOINTS:
        if base_url in seen:
            continue
        seen.add(base_url)
        targets.append((provider, label, base_url, install_url))

    # Probed in parallel. Serially this is the slowest thing the wizard does, because a model
    # server that is running can take a couple of seconds to list its catalogue and the wizard
    # would pay that for each one in turn.
    with ThreadPoolExecutor(max_workers=len(targets)) as pool:
        results = list(pool.map(lambda t: probe(*t), targets))
    return [e for e in results if e.reachable or include_unreachable]

hardware 

hardware() -> dict

This machine's CPU, RAM and GPUs, via the same collector the Servers page uses.

Source code in citar/wizard/detect.py 
135
136
137
138
139
140
141
142
def hardware() -> dict:
    """This machine's CPU, RAM and GPUs, via the same collector the Servers page uses."""
    from .. import hwinfo

    try:
        return hwinfo.collect()
    except Exception:
        return {}

usable_vram_gb 

usable_vram_gb(info: Optional[dict] = None) -> float

The largest single GPU's VRAM, in GB.

The largest single card, not the total: a model has to fit in one of them unless the runtime is set up to split it, which is not something to assume during first-time setup. Integrated graphics are skipped — their "VRAM" is a slice of system RAM and produces a suggestion the machine cannot honour.

Source code in citar/wizard/detect.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def usable_vram_gb(info: Optional[dict] = None) -> float:
    """The largest single GPU's VRAM, in GB.

    The largest *single* card, not the total: a model has to fit in one of them unless the runtime
    is set up to split it, which is not something to assume during first-time setup. Integrated
    graphics are skipped — their "VRAM" is a slice of system RAM and produces a suggestion the
    machine cannot honour.
    """
    info = hardware() if info is None else info
    best = 0.0
    for gpu in info.get("gpus") or []:
        vendor = (gpu.get("vendor") or "").lower()
        name = (gpu.get("name") or "").lower()
        if "intel" in vendor and "arc" not in name:
            continue                                            # integrated; not a target
        if gpu.get("vram_gb"):
            best = max(best, float(gpu["vram_gb"]))
    if best:
        return best
    # Apple silicon shares memory between CPU and GPU, and roughly two-thirds of it can be used for
    # a model before the system starts swapping.
    memory = info.get("memory") or {}
    if memory.get("unified") and memory.get("ram_gb"):
        return round(float(memory["ram_gb"]) * 0.66, 1)
    return 0.0

suggest_model 

suggest_model(
    vram_gb: Optional[float] = None,
) -> tuple[str, str]

A model to download, and why, for the VRAM this machine has.

Returns (name, reason). These are starting points rather than recommendations: the Models page ranks what actually plays well once there are games to compare.

Source code in citar/wizard/detect.py 
172
173
174
175
176
177
178
179
180
181
182
def suggest_model(vram_gb: Optional[float] = None) -> tuple[str, str]:
    """A model to download, and why, for the VRAM this machine has.

    Returns ``(name, reason)``. These are starting points rather than recommendations: the Models
    page ranks what actually plays well once there are games to compare.
    """
    vram = usable_vram_gb() if vram_gb is None else vram_gb
    for low, high, name, reason in VRAM_SUGGESTIONS:
        if low <= vram < high:
            return name, reason
    return VRAM_SUGGESTIONS[0][2], VRAM_SUGGESTIONS[0][3]

describe_hardware 

describe_hardware(info: Optional[dict] = None) -> list[str]

Human-readable lines about this machine, for the wizard's opening screen.

Source code in citar/wizard/detect.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def describe_hardware(info: Optional[dict] = None) -> list[str]:
    """Human-readable lines about this machine, for the wizard's opening screen."""
    info = hardware() if info is None else info
    lines = []
    cpu = info.get("cpu") or {}
    if cpu.get("model"):
        threads = cpu.get("threads")
        lines.append(f"CPU: {cpu['model']}" + (f" ({threads} threads)" if threads else ""))
    memory = info.get("memory") or {}
    if memory.get("ram_gb"):
        kind = "Unified memory" if memory.get("unified") else "RAM"
        lines.append(f"{kind}: {memory['ram_gb']} GB")
    for gpu in info.get("gpus") or []:
        vram = f", {gpu['vram_gb']} GB VRAM" if gpu.get("vram_gb") else ""
        lines.append(f"GPU: {gpu.get('name', 'unknown')}{vram}")
    if not lines:
        lines.append("Could not read this machine's hardware; that only affects suggestions.")
    return lines

port_free 

port_free(port: int, host: str = '127.0.0.1') -> bool

Can CITAR bind here? Used to pick a default port that will actually start.

Source code in citar/wizard/detect.py 
205
206
207
def port_free(port: int, host: str = "127.0.0.1") -> bool:
    """Can CITAR bind here? Used to pick a default port that will actually start."""
    return not port_open(host, port, timeout=0.2)

first_free_port 

first_free_port(start: int = 8765, tries: int = 20) -> int

The first free port at or after start, so a second CITAR does not collide with the first.

Source code in citar/wizard/detect.py 
210
211
212
213
214
215
def first_free_port(start: int = 8765, tries: int = 20) -> int:
    """The first free port at or after *start*, so a second CITAR does not collide with the first."""
    for offset in range(tries):
        if port_free(start + offset):
            return start + offset
    return start

citar.wizard.prompts

Terminal prompts for the setup wizard.

Deliberately small and dependency-free: CITAR is often set up over SSH on a box with nothing installed, and a setup wizard that needs a TUI library to ask "which port?" is a setup wizard that fails at the moment it is most needed.

Three things every prompt here gets right, which hand-rolled input() calls usually do not:

  • A default you can accept with Return, shown in the prompt, so the whole wizard can be walked through by pressing Return when the defaults are right.
  • Ctrl-C and EOF end the wizard cleanly, with a message saying nothing was changed, rather than a traceback. citar setup piped from a script hits EOF on the first question; that must not look like a crash.
  • Non-interactive mode, where a question with no answer supplied is an error naming the flag that would have supplied it, instead of a hang waiting on a terminal nobody is watching.

Cancelled

Bases: Exception

The operator ended the wizard. Nothing has been written.

Source code in citar/wizard/prompts.py 
26
27
class Cancelled(Exception):
    """The operator ended the wizard. Nothing has been written."""

NeedsAnswer

Bases: Exception

A non-interactive run reached a question it had no answer for.

Source code in citar/wizard/prompts.py 
30
31
32
33
34
35
36
class NeedsAnswer(Exception):
    """A non-interactive run reached a question it had no answer for."""

    def __init__(self, question: str, flag: str = "") -> None:
        super().__init__(
            f"{question}\nThis is a non-interactive run, so there is nobody to ask."
            + (f" Pass {flag}." if flag else ""))

say 

say(text: str = '') -> None

Print a line of wizard output.

Source code in citar/wizard/prompts.py 
39
40
41
def say(text: str = "") -> None:
    """Print a line of wizard output."""
    print(text)

heading 

heading(text: str) -> None

A section heading, so a long wizard reads as steps rather than a wall of questions.

Source code in citar/wizard/prompts.py 
44
45
46
def heading(text: str) -> None:
    """A section heading, so a long wizard reads as steps rather than a wall of questions."""
    print(f"\n{text}\n{'=' * len(text)}")

note 

note(text: str) -> None

An indented explanation under a question.

Source code in citar/wizard/prompts.py 
49
50
51
52
def note(text: str) -> None:
    """An indented explanation under a question."""
    for line in text.splitlines():
        print(f"    {line}")

ask 

ask(
    question: str,
    default: str = "",
    flag: str = "",
    validate: Optional[
        Callable[[str], Optional[str]]
    ] = None,
) -> str

Ask for a line of text.

validate returns an error message for a bad answer, or None when it is acceptable; the question is asked again until it passes. That keeps validation next to the question rather than in a second pass at the end, where the operator has forgotten what they typed.

Source code in citar/wizard/prompts.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def ask(question: str, default: str = "", flag: str = "",
        validate: Optional[Callable[[str], Optional[str]]] = None) -> str:
    """Ask for a line of text.

    *validate* returns an error message for a bad answer, or ``None`` when it is acceptable; the
    question is asked again until it passes. That keeps validation next to the question rather than
    in a second pass at the end, where the operator has forgotten what they typed.
    """
    if NON_INTERACTIVE:
        if default:
            return default
        raise NeedsAnswer(question, flag)
    suffix = f" [{default}]" if default else ""
    while True:
        answer = _read(f"{question}{suffix}: ").strip() or default
        if validate:
            problem = validate(answer)
            if problem:
                note(problem)
                continue
        return answer

ask_yes_no 

ask_yes_no(
    question: str, default: bool = True, flag: str = ""
) -> bool

Ask a yes/no question. Returns the default on a bare Return.

Source code in citar/wizard/prompts.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def ask_yes_no(question: str, default: bool = True, flag: str = "") -> bool:
    """Ask a yes/no question. Returns the default on a bare Return."""
    if NON_INTERACTIVE:
        return default
    suffix = " [Y/n]" if default else " [y/N]"
    while True:
        answer = _read(f"{question}{suffix}: ").strip().lower()
        if not answer:
            return default
        if answer in ("y", "yes"):
            return True
        if answer in ("n", "no"):
            return False
        note("Please answer y or n.")

ask_choice 

ask_choice(
    question: str,
    options: Sequence[tuple[str, str, str]],
    default: str = "",
    flag: str = "",
) -> str

Ask for one of several options.

Each option is (key, title, explanation). The explanation is printed under the title because the whole reason a wizard beats a config file is that it can say what a choice means at the moment the choice is made.

Source code in citar/wizard/prompts.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def ask_choice(question: str, options: Sequence[tuple[str, str, str]], default: str = "",
               flag: str = "") -> str:
    """Ask for one of several options.

    Each option is ``(key, title, explanation)``. The explanation is printed under the title because
    the whole reason a wizard beats a config file is that it can say what a choice means at the
    moment the choice is made.
    """
    if NON_INTERACTIVE:
        if default:
            return default
        raise NeedsAnswer(question, flag)
    keys = [key for key, _, _ in options]
    print()
    print(question)
    print()
    for index, (key, title, explanation) in enumerate(options, start=1):
        marker = " (default)" if key == default else ""
        print(f"  {index}. {title}{marker}")
        for line in explanation.splitlines():
            print(f"     {line}")
        print()
    while True:
        answer = _read(f"Choose 1-{len(options)}" + (f" [{keys.index(default) + 1}]" if default else "") + ": ").strip()
        if not answer and default:
            return default
        if answer.isdigit() and 1 <= int(answer) <= len(options):
            return keys[int(answer) - 1]
        if answer.lower() in keys:
            return answer.lower()
        note(f"Please enter a number from 1 to {len(options)}.")

ask_secret 

ask_secret(
    question: str, flag: str = "", allow_empty: bool = False
) -> str

Ask for something that must not be echoed or stored in shell history.

Falls back to a visible prompt only when there is no terminal to hide the typing — and says so, because somebody pasting an API key deserves to know it is on screen.

Source code in citar/wizard/prompts.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def ask_secret(question: str, flag: str = "", allow_empty: bool = False) -> str:
    """Ask for something that must not be echoed or stored in shell history.

    Falls back to a visible prompt only when there is no terminal to hide the typing — and says so,
    because somebody pasting an API key deserves to know it is on screen.
    """
    if NON_INTERACTIVE:
        if allow_empty:
            return ""
        raise NeedsAnswer(question, flag)
    import getpass

    while True:
        try:
            answer = getpass.getpass(f"{question} (not shown): ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            raise Cancelled("Setup cancelled. Nothing was changed.") from None
        except Exception:
            note("This terminal cannot hide input, so what you type will be visible.")
            answer = _read(f"{question}: ").strip()
        if answer or allow_empty:
            return answer
        note("That cannot be empty.")

ask_port 

ask_port(
    question: str, default: int, flag: str = ""
) -> int

Ask for a TCP port, and reject the ones that will not work.

Source code in citar/wizard/prompts.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def ask_port(question: str, default: int, flag: str = "") -> int:
    """Ask for a TCP port, and reject the ones that will not work."""

    def check(value: str) -> Optional[str]:
        """Validate a port number, rejecting the ones that will not work."""
        if not value.isdigit():
            return "A port is a number, such as 8765."
        port = int(value)
        if not 1 <= port <= 65535:
            return "Ports run from 1 to 65535."
        if port < 1024 and sys.platform != "win32":
            return "Ports below 1024 need root. Use something above 1024 and put a proxy in front."
        return None

    return int(ask(question, str(default), flag, validate=check))

confirm_plan 

confirm_plan(
    title: str, lines: Sequence[str], default: bool = True
) -> bool

Show everything that is about to happen, and ask once.

This is the only place the wizard asks for permission to change the machine, so it lists absolutely everything it will write — including files it will create outside the project.

Source code in citar/wizard/prompts.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
def confirm_plan(title: str, lines: Sequence[str], default: bool = True) -> bool:
    """Show everything that is about to happen, and ask once.

    This is the only place the wizard asks for permission to change the machine, so it lists
    absolutely everything it will write — including files it will create outside the project.
    """
    print()
    print(title)
    print("-" * len(title))
    for line in lines:
        print(f"  {line}")
    print()
    return ask_yes_no("Go ahead?", default)

The game engine

citar.engine.rules

Loads the rule set: UnCiv-derived data in citar/data/ruleset, CITAR additions in citar/data/custom and CITAR constants in citar/data/game.json. Every ruleset object is keyed by its display name (as UnCiv does, because uniques refer to objects by name) and also has a snake_case "id" that tools accept.

RULES_VERSION module-attribute 

RULES_VERSION = 2

get_rules cached 

get_rules(data_dir: str = str(DATA_DIR)) -> Rules

The ruleset, loaded once and cached.

Source code in citar/engine/rules.py 
334
335
336
337
@lru_cache(maxsize=4)
def get_rules(data_dir: str = str(DATA_DIR)) -> Rules:
    """The ruleset, loaded once and cached."""
    return Rules(data_dir)

citar.engine.tools

The single registry of player tools (queries and actions).

Every interface — the browser UI, REST API, MCP server and LLM adapter — dispatches through execute(), so rules are enforced identically for humans and AIs, and new tools automatically reach every client.

tool 

tool(
    name: str,
    description: str,
    properties: Optional[dict] = None,
    required=(),
    kind="action",
    any_time=False,
    category="general",
)

Register a function as a player tool.

This decorator is the whole interface. Registering a function makes it callable from the browser, from MCP and from the LLM adapter at once, with the schema generated from properties - so there is no way to add an action to one interface and forget another, and no way for a model to have a power a human does not.

Source code in citar/engine/tools.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def tool(name: str, description: str, properties: Optional[dict] = None, required=(), kind="action", any_time=False,
         category="general"):
    """Register a function as a player tool.

    This decorator is the whole interface. Registering a function makes it callable from the browser,
    from MCP and from the LLM adapter at once, with the schema generated from *properties* - so there is
    no way to add an action to one interface and forget another, and no way for a model to have a power
    a human does not.
    """
    def deco(fn):
        """Record the function in the registry and return it unchanged."""
        REGISTRY[name] = Tool(name, description, properties or {}, list(required), fn, kind, any_time, category)
        return fn
    return deco

Servers and costing

citar.servers

The server registry (config/servers.json): every machine or online service that can run a model for CITAR, with everything needed to use it and to cost it.

A server has connection how to reach it: provider (lmstudio | ollama | openai_compatible | anthropic | dryrun | none), base URL, LM Link device (a model on another PC reached through this PC's LM Studio), API key backend (see keystore.py), how many games may use it at once, whether CITAR loads models itself hardware CPU, GPUs, RAM (or unified memory), disks... collected by hwinfo.py or typed in power idle watts and the extra watts at full CPU / full GPU load, used to estimate energy wherever it isn't sampled live components owned hardware with price, purchase date, lifespan and resale value (depreciated straight-line) costs effective-dated cost periods: electricity plan, fixed monthly costs (lease, maintenance), an hourly rate while in use (rented machines), per-token prices (APIs) restricted_hours windows (per weekday) when queued work must pause on it, optionally unloading its models models the model catalog: key, label, inference defaults and load profiles (context, GPU offload...)

Electricity plans are shared (both home PCs are on the same bill). One server is the CITAR host: the machine running this program, whose CPU runs the game engine and the scripted bots.

load 

load() -> dict

The registry (cached until the file changes). Creates the default registry the first time.

Source code in citar/servers.py 
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
def load() -> dict:
    """The registry (cached until the file changes). Creates the default registry the first time."""
    with _lock:
        mtime = PATH.stat().st_mtime if PATH.exists() else None
        if _cache["data"] is not None and _cache["mtime"] == mtime:
            return _cache["data"]
        raw = _read()
        if not raw:
            data = bootstrap()
            save(data)
            return data
        data = empty_registry()
        data.update({k: raw[k] for k in ("currency", "currency_symbol", "host_server_id") if k in raw})
        data["electricity_plans"] = [normalize_plan(p) for p in raw.get("electricity_plans") or []]
        data["servers"] = []
        for s in raw.get("servers") or []:
            sv = normalize_server(s)
            sv["updated"] = s.get("updated") or sv["updated"]
            data["servers"].append(sv)
        _cache.update(mtime=mtime, data=data)
        return data

save 

save(data: dict)

Write the registry atomically.

Atomic because the web page reads it while the lab writes it; a partial file would be read as a registry with no servers in it.

Source code in citar/servers.py 
440
441
442
443
444
445
446
447
448
449
450
def save(data: dict):
    """Write the registry atomically.

    Atomic because the web page reads it while the lab writes it; a partial file would be read as a
    registry with no servers in it.
    """
    from .fsutil import write_text
    with _lock:
        CONFIG_DIR.mkdir(parents=True, exist_ok=True)
        write_text(PATH, json.dumps(data, indent=1))
        _cache.update(mtime=PATH.stat().st_mtime, data=data)

list_servers 

list_servers() -> list[dict]

Every server in the registry.

Source code in citar/servers.py 
487
488
489
def list_servers() -> list[dict]:
    """Every server in the registry."""
    return load()["servers"]

get 

get(server_id: str) -> dict

One server by id, raising if it does not exist.

Source code in citar/servers.py 
492
493
494
495
496
497
def get(server_id: str) -> dict:
    """One server by id, raising if it does not exist."""
    for sv in load()["servers"]:
        if sv["id"] == server_id:
            return sv
    raise ServerError(f"No server '{server_id}'.")

upsert 

upsert(data: dict) -> dict

Create or replace a server, validating it and its references first.

Source code in citar/servers.py 
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
def upsert(data: dict) -> dict:
    """Create or replace a server, validating it and its references first."""
    with _lock:
        reg = snapshot()
        sv = normalize_server(data)
        for plan_id in {p.get("electricity_plan_id") for p in sv["costs"]} - {None}:
            if not any(pl["id"] == plan_id for pl in reg["electricity_plans"]):
                raise ServerError(f"Unknown electricity plan '{plan_id}'.")
        idx = next((i for i, s in enumerate(reg["servers"]) if s["id"] == sv["id"]), None)
        if idx is None:
            reg["servers"].append(sv)
        else:
            sv["created"] = reg["servers"][idx].get("created") or sv["created"]
            reg["servers"][idx] = sv
        if data.get("is_host"):
            reg["host_server_id"] = sv["id"]
        elif reg.get("host_server_id") == sv["id"] and data.get("is_host") is False:
            reg["host_server_id"] = None
        save(reg)
        return sv

delete 

delete(server_id: str)

Remove a server.

Source code in citar/servers.py 
539
540
541
542
543
544
545
546
def delete(server_id: str):
    """Remove a server."""
    with _lock:
        reg = snapshot()
        reg["servers"] = [s for s in reg["servers"] if s["id"] != server_id]
        if reg.get("host_server_id") == server_id:
            reg["host_server_id"] = None
        save(reg)

resolve_llm 

resolve_llm(llm: dict, with_key: bool = True) -> dict

Turn a seat's llm block into the agent config: provider, endpoint, key, model and inference settings. Seat blocks without a server_id (tests, scripts) are passed through unchanged.

Source code in citar/servers.py 
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
def resolve_llm(llm: dict, with_key: bool = True) -> dict:
    """Turn a seat's llm block into the agent config: provider, endpoint, key, model and inference settings. Seat
    blocks without a server_id (tests, scripts) are passed through unchanged."""
    if not llm or not llm.get("server_id"):
        return dict(llm or {})
    if find(llm["server_id"]) is None:
        # not in the registry: a machine from the Servers page, played through its helper
        from .pool import seats as pool_seats
        pooled = pool_seats.lookup(llm["server_id"])
        if pooled is not None:
            return pool_seats.resolve(llm, pooled)
    sv = get(llm["server_id"])
    m = model_entry(sv, llm.get("model_id") or llm.get("model"))
    if m is None:
        # a model that isn't in the catalog (yet): usable with catalog defaults
        m = default_model(llm.get("model") or "", sv["connection"]["provider"])
        m["id"] = None
    conn = sv["connection"]
    provider = conn["provider"]
    cfg = {"server_id": sv["id"], "server": sv["name"], "model_id": m["id"], "model": m["key"],
           "provider": {"lmstudio": "openai_compatible", "ollama": "openai_compatible"}.get(provider, provider),
           "base_url": conn.get("base_url") or None, "timeout": conn.get("timeout")}
    inf = m.get("inference") or {}
    for k in ("reasoning_effort", "effort", "max_tokens", "temperature", "max_tool_calls_per_turn"):
        if inf.get(k) not in (None, ""):
            cfg[k] = inf[k]
    tool_mode = inf.get("tool_mode") or "auto"
    for k in SEAT_OVERRIDES:
        if llm.get(k) not in (None, ""):
            cfg[k] = llm[k]
    if cfg.get("tool_mode"):
        tool_mode = cfg["tool_mode"]
    if cfg["provider"] == "openai_compatible":
        if tool_mode == "auto":
            if provider == "lmstudio":
                from .server import lmstudio
                tool_mode = lmstudio.tool_mode_for(conn.get("base_url") or "", m["key"], "auto")
            else:
                tool_mode = "native"
        cfg["tool_mode"] = tool_mode
    else:
        cfg.pop("tool_mode", None)
        cfg.pop("reasoning_effort", None)
    if cfg["provider"] != "anthropic":
        cfg.pop("effort", None)
    if provider == "dryrun":
        cfg["dry_run_delay"] = conn.get("dry_run_delay", 1.0) if llm.get("dry_run_delay") is None else llm["dry_run_delay"]
    prof = profile(m, llm.get("profile_id"))
    if prof:
        cfg["profile_id"] = prof["id"]
        cfg["load"] = dict(prof)
    if with_key and conn["key"]["backend"] != "none":
        from . import keystore
        key = keystore.get(sv)
        if key:
            cfg["api_key"] = key
        elif conn["key"]["backend"] == "env" and conn["key"].get("env"):
            cfg["api_key_env"] = conn["key"]["env"]
    return {k: v for k, v in cfg.items() if v is not None}

seat_ref 

seat_ref(
    server_id: str,
    model_ref: str,
    profile_id: Optional[str] = None,
    **extra,
) -> dict

The small, saveable description of an LLM seat: which server, model and load profile, plus seat overrides.

Source code in citar/servers.py 
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
def seat_ref(server_id: str, model_ref: str, profile_id: Optional[str] = None, **extra) -> dict:
    """The small, saveable description of an LLM seat: which server, model and load profile, plus seat overrides."""
    if find(server_id) is None:
        from .pool import seats as pool_seats
        pooled = pool_seats.lookup(server_id)
        if pooled is not None:
            # a machine from the Servers page: the model is named by its key, and there are no load profiles
            d = {"server_id": pooled["id"], "model_id": model_ref, "model": model_ref, "profile_id": None,
                 "server": pooled["name"], "provider": "worker"}
            d.update({k: v for k, v in extra.items() if v is not None})
            return d
    sv = get(server_id)
    m = model_entry(sv, model_ref)
    if m is None:
        raise ServerError(f"{sv['name']} has no model '{model_ref}'.")
    d = {"server_id": sv["id"], "model_id": m["id"], "model": m["key"], "profile_id": (profile(m, profile_id) or {}).get("id"),
         "server": sv["name"], "provider": sv["connection"]["provider"]}
    d.update({k: v for k, v in extra.items() if v is not None})
    return d

restricted 

restricted(
    server: Optional[dict], when: Optional[datetime] = None
) -> Optional[datetime]

When the server is inside one of its restricted windows: the time the window ends. Otherwise None.

Source code in citar/servers.py 
628
629
630
631
632
633
634
635
636
def restricted(server: Optional[dict], when: Optional[datetime] = None) -> Optional[datetime]:
    """When the server is inside one of its restricted windows: the time the window ends. Otherwise None."""
    if not server:
        return None
    rh = server.get("restricted_hours") or {}
    if not rh.get("enabled") or not rh.get("windows"):
        return None
    hit = _window_hit(rh["windows"], when or datetime.now())
    return hit[1] if hit else None

default_server 

default_server(
    kind: str = "owned", provider: str = "lmstudio", **kw
) -> dict

A new server entry of a kind and provider, with sensible defaults throughout.

Source code in citar/servers.py 
179
180
181
182
183
184
185
186
187
188
189
190
def default_server(kind: str = "owned", provider: str = "lmstudio", **kw) -> dict:
    """A new server entry of a kind and provider, with sensible defaults throughout."""
    now = time.time()
    sv = {"id": new_id("sv_"), "name": "New server", "description": "", "kind": kind,
          "connection": {"provider": provider, "base_url": "http://localhost:1234/v1" if provider == "lmstudio" else "",
                         "lmstudio_device": "", "manage_loading": provider == "lmstudio", "max_parallel": 1,
                         "key": {"backend": "none", "env": ""}, "dry_run_delay": 1.0, "timeout": None},
          "hardware": {"source": "none"}, "power": default_power(), "components": [],
          "costs": [default_cost_period(kind)], "restricted_hours": default_restricted(), "models": [],
          "notes": "", "created": now, "updated": now}
    sv.update(kw)
    return sv

default_model 

default_model(
    key: str, provider: str = "lmstudio", **kw
) -> dict

A model entry with inference defaults suited to where it runs.

Local models default to low reasoning effort and automatic tool mode, because that is what makes them usable - about four times faster with no loss in play quality. Hosted models default to native tool calling, which they do well.

Source code in citar/servers.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def default_model(key: str, provider: str = "lmstudio", **kw) -> dict:
    """A model entry with inference defaults suited to where it runs.

    Local models default to low reasoning effort and automatic tool mode, because that is what makes
    them usable - about four times faster with no loss in play quality. Hosted models default to native
    tool calling, which they do well.
    """
    local = provider in ("lmstudio", "ollama", "openai_compatible")
    m = {"id": new_id("m_"), "key": key, "label": "", "enabled": True, "info": {},
         "inference": {"tool_mode": "auto" if local else "native", "reasoning_effort": "low" if local else "",
                       "effort": "", "max_tokens": None, "temperature": None, "max_tool_calls_per_turn": 150},
         "profiles": [default_profile()] if provider == "lmstudio" else [],
         "default_profile": "p_default" if provider == "lmstudio" else None, "price": None}
    m.update(kw)
    return m

empty_registry 

empty_registry() -> dict

A registry with no servers in it.

Source code in citar/servers.py 
202
203
204
205
def empty_registry() -> dict:
    """A registry with no servers in it."""
    return {"format": "citar-servers", "version": 1, "currency": "USD", "currency_symbol": "$", "host_server_id": None,
            "electricity_plans": [], "servers": []}

citar.usage

Usage ledger: what every activity (game, benchmark game, probe run, lab game, report) used on which server, over time. Costs are not stored here: reports price the ledger with the server configs in force at the time, so fixing a wrong electricity rate later fixes every report.

Files (saves/usage/, JSON lines, one writer per file so processes never interleave): server-YYYY-MM.jsonl written by the CITAR server (games, benchmarks, probes, reports) lab-YYYY-MM.jsonl written by the lab runner (bot-vs-bot games) power--YYYY-MM.jsonl one-minute power samples of a machine (GPU watts from nvidia-smi, CPU utilisation)

Row kinds

{"k": "act", "id", "kind", "name", "parent": {kind, id, name}, "ref": {...}, "t", ...} an activity; later rows for the same id update it {"k": "span", "act", "srv", "model", "t0", "t1", "held", "busy", "cpu", "in", "out", "rsn", "cr", "cw", "req"} one flush window (about a minute): seconds the activity held the server, seconds the model was generating, CPU seconds on the host, tokens (input, output, reasoning, cache read, cache write) and model requests {"k": "pw", "srv", "t0", "t1", "n", "gpu_w", "cpu_util"} one minute of power samples

Tracker

Collects usage in the CITAR server process and writes it to the ledger about once a minute.

Source code in citar/usage.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
class Tracker:
    """Collects usage in the CITAR server process and writes it to the ledger about once a minute."""

    def __init__(self, writer: str = "server"):
        self.writer = writer
        self.lock = threading.RLock()
        self.holders: dict[str, tuple] = {}               # key -> (act, fn() -> {srv: {"threads": [...]}} or None)
        self.acc: dict = defaultdict(lambda: {"held": 0.0, "busy": 0.0, "cpu": 0.0, "models": defaultdict(lambda: defaultdict(float))})
        self.window_start = time.time()
        self._last_sample = time.time()
        self._thread_cpu: dict = {}                        # native thread id -> last cpu seconds
        self._rows: list = []
        self._thread: Optional[threading.Thread] = None
        self._stop = threading.Event()

    # ---- activities
    def activity(self, act_id: str, kind: str, name: str = "", parent: Optional[dict] = None, ref: Optional[dict] = None,
                 **fields):
        """Begin recording one activity: a game, a benchmark, a probe run, a lab game."""
        row = {"k": "act", "id": act_id, "kind": kind, "name": name, "t": time.time()}
        if parent:
            row["parent"] = parent
        if ref:
            row["ref"] = ref
        row.update({k: v for k, v in fields.items() if v is not None})
        with self.lock:
            self._rows.append(row)

    def update(self, act_id: str, **fields):
        """Add or change fields on an activity in progress."""
        row = {"k": "act", "id": act_id, "t": time.time()}
        row.update({k: v for k, v in fields.items() if v is not None})
        with self.lock:
            self._rows.append(row)

    def register(self, act_id: str, holder: Callable, key: Optional[str] = None):
        """Count the servers `holder()` reports as held toward `act_id` (several holders may feed one activity, e.g.
        the cases of a probe run: give each its own key)."""
        with self.lock:
            self.holders[key or act_id] = (act_id, holder)
        self.start()

    def unregister(self, key: str):
        """Stop tracking something that has ended."""
        with self.lock:
            fn = self.holders.pop(key, None)
        if fn is not None:
            self._sample()

    # ---- model calls
    def llm(self, act_id: Optional[str], server_id: Optional[str], model: Optional[str], seconds: float,
            input_tokens: int = 0, output_tokens: int = 0, reasoning_tokens: int = 0, cache_read: int = 0,
            cache_write: int = 0, requests: int = 1):
        """Record one model call: which server and model, how long, and how many tokens."""
        if not act_id:
            return
        with self.lock:
            a = self.acc[(act_id, server_id or "unassigned")]
            a["busy"] += max(0.0, seconds)
            m = a["models"][model or "?"]
            m["busy"] += max(0.0, seconds)
            m["in"] += input_tokens or 0
            m["out"] += output_tokens or 0
            m["rsn"] += reasoning_tokens or 0
            m["cr"] += cache_read or 0
            m["cw"] += cache_write or 0
            m["req"] += requests
        self.start()

    # ---- background
    def start(self):
        """Start the sampling thread."""
        if self._thread is None or not self._thread.is_alive():
            self._thread = threading.Thread(target=self._loop, daemon=True, name=f"usage-{self.writer}")
            self._thread.start()

    def stop(self):
        """Stop sampling and flush what has not been written."""
        self._stop.set()
        self.flush()

    def _loop(self):
        """The sampling thread: take a sample, write what is due, repeat."""
        while not self._stop.wait(SAMPLE_SECONDS):
            try:
                self._sample()
                if time.time() - self.window_start >= FLUSH_SECONDS:
                    self.flush()
            except Exception:
                traceback.print_exc()

    def _thread_times(self) -> dict:
        """CPU time per thread, for attributing the host's work to the right activity."""
        try:
            import psutil
            return {t.id: t.user_time + t.system_time for t in psutil.Process().threads()}
        except Exception:
            return {}

    def _sample(self):
        """Take one sample of CPU and power."""
        now = time.time()
        with self.lock:
            dt = min(now - self._last_sample, SAMPLE_SECONDS * 4)
            self._last_sample = now
            holders = dict(self.holders)
        if dt <= 0:
            return
        times = self._thread_times()
        done = []
        for key, (act, fn) in holders.items():
            try:
                held = fn()
            except Exception:
                held = {}
            if held is None:
                done.append(key)
                continue
            for srv, info in held.items():
                with self.lock:
                    a = self.acc[(act, srv)]
                    a["held"] += dt
                    for tid in (info or {}).get("threads") or []:
                        cur = times.get(tid)
                        if cur is None:
                            continue
                        prev = self._thread_cpu.get(tid)
                        self._thread_cpu[tid] = cur
                        if prev is not None and cur >= prev:
                            a["cpu"] += cur - prev
        with self.lock:
            for act in done:
                self.holders.pop(act, None)

    def flush(self):
        """Write pending ledger entries to disk."""
        now = time.time()
        with self.lock:
            rows, self._rows = self._rows, []
            t0 = self.window_start
            self.window_start = now
            acc, self.acc = self.acc, defaultdict(lambda: {"held": 0.0, "busy": 0.0, "cpu": 0.0,
                                                           "models": defaultdict(lambda: defaultdict(float))})
        for (act, srv), a in acc.items():
            base = {"k": "span", "act": act, "srv": srv, "t0": round(t0, 1), "t1": round(now, 1),
                    "held": round(a["held"], 1), "cpu": round(a["cpu"], 2)}
            if not a["models"]:
                if base["held"] or base["cpu"]:
                    rows.append({**base, "busy": 0.0})
                continue
            first = True
            for model, m in a["models"].items():
                row = {**base, "model": model, "busy": round(m["busy"], 2),
                       **{f: int(m[f]) for f in TOKEN_FIELDS if m.get(f)}}
                if not first:
                    row["held"] = 0.0          # the hold belongs to the server once, not to each model on it
                    row["cpu"] = 0.0
                first = False
                rows.append(row)
        append(self.writer, rows)

activity 

activity(
    act_id: str,
    kind: str,
    name: str = "",
    parent: Optional[dict] = None,
    ref: Optional[dict] = None,
    **fields,
)

Begin recording one activity: a game, a benchmark, a probe run, a lab game.

Source code in citar/usage.py 
117
118
119
120
121
122
123
124
125
126
127
def activity(self, act_id: str, kind: str, name: str = "", parent: Optional[dict] = None, ref: Optional[dict] = None,
             **fields):
    """Begin recording one activity: a game, a benchmark, a probe run, a lab game."""
    row = {"k": "act", "id": act_id, "kind": kind, "name": name, "t": time.time()}
    if parent:
        row["parent"] = parent
    if ref:
        row["ref"] = ref
    row.update({k: v for k, v in fields.items() if v is not None})
    with self.lock:
        self._rows.append(row)

update 

update(act_id: str, **fields)

Add or change fields on an activity in progress.

Source code in citar/usage.py 
129
130
131
132
133
134
def update(self, act_id: str, **fields):
    """Add or change fields on an activity in progress."""
    row = {"k": "act", "id": act_id, "t": time.time()}
    row.update({k: v for k, v in fields.items() if v is not None})
    with self.lock:
        self._rows.append(row)

register 

register(
    act_id: str, holder: Callable, key: Optional[str] = None
)

Count the servers holder() reports as held toward act_id (several holders may feed one activity, e.g. the cases of a probe run: give each its own key).

Source code in citar/usage.py 
136
137
138
139
140
141
def register(self, act_id: str, holder: Callable, key: Optional[str] = None):
    """Count the servers `holder()` reports as held toward `act_id` (several holders may feed one activity, e.g.
    the cases of a probe run: give each its own key)."""
    with self.lock:
        self.holders[key or act_id] = (act_id, holder)
    self.start()

unregister 

unregister(key: str)

Stop tracking something that has ended.

Source code in citar/usage.py 
143
144
145
146
147
148
def unregister(self, key: str):
    """Stop tracking something that has ended."""
    with self.lock:
        fn = self.holders.pop(key, None)
    if fn is not None:
        self._sample()

llm 

llm(
    act_id: Optional[str],
    server_id: Optional[str],
    model: Optional[str],
    seconds: float,
    input_tokens: int = 0,
    output_tokens: int = 0,
    reasoning_tokens: int = 0,
    cache_read: int = 0,
    cache_write: int = 0,
    requests: int = 1,
)

Record one model call: which server and model, how long, and how many tokens.

Source code in citar/usage.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def llm(self, act_id: Optional[str], server_id: Optional[str], model: Optional[str], seconds: float,
        input_tokens: int = 0, output_tokens: int = 0, reasoning_tokens: int = 0, cache_read: int = 0,
        cache_write: int = 0, requests: int = 1):
    """Record one model call: which server and model, how long, and how many tokens."""
    if not act_id:
        return
    with self.lock:
        a = self.acc[(act_id, server_id or "unassigned")]
        a["busy"] += max(0.0, seconds)
        m = a["models"][model or "?"]
        m["busy"] += max(0.0, seconds)
        m["in"] += input_tokens or 0
        m["out"] += output_tokens or 0
        m["rsn"] += reasoning_tokens or 0
        m["cr"] += cache_read or 0
        m["cw"] += cache_write or 0
        m["req"] += requests
    self.start()

start 

start()

Start the sampling thread.

Source code in citar/usage.py 
171
172
173
174
175
def start(self):
    """Start the sampling thread."""
    if self._thread is None or not self._thread.is_alive():
        self._thread = threading.Thread(target=self._loop, daemon=True, name=f"usage-{self.writer}")
        self._thread.start()

stop 

stop()

Stop sampling and flush what has not been written.

Source code in citar/usage.py 
177
178
179
180
def stop(self):
    """Stop sampling and flush what has not been written."""
    self._stop.set()
    self.flush()

flush 

flush()

Write pending ledger entries to disk.

Source code in citar/usage.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def flush(self):
    """Write pending ledger entries to disk."""
    now = time.time()
    with self.lock:
        rows, self._rows = self._rows, []
        t0 = self.window_start
        self.window_start = now
        acc, self.acc = self.acc, defaultdict(lambda: {"held": 0.0, "busy": 0.0, "cpu": 0.0,
                                                       "models": defaultdict(lambda: defaultdict(float))})
    for (act, srv), a in acc.items():
        base = {"k": "span", "act": act, "srv": srv, "t0": round(t0, 1), "t1": round(now, 1),
                "held": round(a["held"], 1), "cpu": round(a["cpu"], 2)}
        if not a["models"]:
            if base["held"] or base["cpu"]:
                rows.append({**base, "busy": 0.0})
            continue
        first = True
        for model, m in a["models"].items():
            row = {**base, "model": model, "busy": round(m["busy"], 2),
                   **{f: int(m[f]) for f in TOKEN_FIELDS if m.get(f)}}
            if not first:
                row["held"] = 0.0          # the hold belongs to the server once, not to each model on it
                row["cpu"] = 0.0
            first = False
            rows.append(row)
    append(self.writer, rows)

PowerSampler

Samples this machine's power draw every few seconds and writes one row a minute: GPU watts (nvidia-smi, all GPUs) and CPU utilisation. Only one process on the machine samples at a time (a heartbeat lock file), so the CITAR server and the lab runner can both start one.

Source code in citar/usage.py 
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
class PowerSampler:
    """Samples this machine's power draw every few seconds and writes one row a minute: GPU watts (nvidia-smi, all
    GPUs) and CPU utilisation. Only one process on the machine samples at a time (a heartbeat lock file), so the
    CITAR server and the lab runner can both start one."""

    def __init__(self, server_id: str, writer_lock: Optional[Path] = None):
        self.server_id = server_id
        self.lock_path = writer_lock or USAGE_DIR / f"power-{server_id}.lock"
        self._thread: Optional[threading.Thread] = None
        self._stop = threading.Event()
        self.nvsmi = None
        import shutil
        self.nvsmi = shutil.which("nvidia-smi")

    def start(self):
        """Start sampling."""
        if self._thread is None:
            self._thread = threading.Thread(target=self._loop, daemon=True, name="power-sampler")
            self._thread.start()

    def stop(self):
        """Stop sampling."""
        self._stop.set()

    def _own_lock(self) -> bool:
        """Whether this process is the one sampling, so two CITARs do not both measure."""
        me = f"{os.getpid()}"
        try:
            if self.lock_path.exists():
                txt = self.lock_path.read_text(encoding="utf-8").split()
                if txt and txt[0] != me and time.time() - float(txt[1]) < 45:
                    return False
            # refresh the heartbeat every 20 s, not every sample (the folder may sync to the cloud)
            if time.time() - getattr(self, "_beat", 0) > 20:
                USAGE_DIR.mkdir(parents=True, exist_ok=True)
                self.lock_path.write_text(f"{me} {time.time():.0f}", encoding="utf-8")
                self._beat = time.time()
            return True
        except (OSError, ValueError, IndexError):
            return False

    def _gpu_watts(self) -> Optional[float]:
        """GPU power from nvidia-smi, or None where it cannot be read."""
        if not self.nvsmi:
            return None
        import subprocess
        try:
            r = subprocess.run([self.nvsmi, "--query-gpu=power.draw", "--format=csv,noheader,nounits"], capture_output=True,
                               stdin=subprocess.DEVNULL, timeout=10)
            vals = [float(x) for x in r.stdout.decode().split() if x.replace(".", "", 1).isdigit()]
            return sum(vals) if vals else None
        except (OSError, ValueError, subprocess.SubprocessError):
            return None

    def _loop(self):
        """The sampling thread."""
        try:
            import psutil
        except ImportError:
            psutil = None
        gpu, cpu, n = [], [], 0
        start = time.time()
        if psutil:
            psutil.cpu_percent(None)
        while not self._stop.wait(SAMPLE_SECONDS):
            try:
                from . import servers as S
                sv = S.find(self.server_id)
                if sv is None or sv["power"].get("sampling") == "off" or not self._own_lock():
                    gpu, cpu, n, start = [], [], 0, time.time()
                    continue
                w = self._gpu_watts()
                if w is not None:
                    gpu.append(w)
                if psutil:
                    cpu.append(psutil.cpu_percent(None) / 100.0)
                n += 1
                if time.time() - start >= 60:
                    row = {"k": "pw", "srv": self.server_id, "t0": round(start, 1), "t1": round(time.time(), 1), "n": n,
                           "cpu_util": round(sum(cpu) / len(cpu), 4) if cpu else None,
                           "gpu_w": round(sum(gpu) / len(gpu), 2) if gpu else None}
                    append(f"power-{self.server_id}", [row])
                    gpu, cpu, n, start = [], [], 0, time.time()
            except Exception:
                traceback.print_exc()

start 

start()

Start sampling.

Source code in citar/usage.py 
338
339
340
341
342
def start(self):
    """Start sampling."""
    if self._thread is None:
        self._thread = threading.Thread(target=self._loop, daemon=True, name="power-sampler")
        self._thread.start()

stop 

stop()

Stop sampling.

Source code in citar/usage.py 
344
345
346
def stop(self):
    """Stop sampling."""
    self._stop.set()

append 

append(writer: str, rows: list[dict])

Append rows to this writer's ledger file for the current month.

Source code in citar/usage.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def append(writer: str, rows: list[dict]):
    """Append rows to this writer's ledger file for the current month."""
    if not rows:
        return
    USAGE_DIR.mkdir(parents=True, exist_ok=True)
    path = USAGE_DIR / f"{writer}-{_month(time.time())}.jsonl"
    text = "".join(json.dumps(r, separators=(",", ":"), default=str) + "\n" for r in rows)
    for k in range(40):
        try:
            with open(path, "a", encoding="utf-8") as f:
                f.write(text)
            return
        except PermissionError:          # a sync client briefly holding the file
            time.sleep(0.1)

read 

read(
    since: Optional[float] = None,
    until: Optional[float] = None,
) -> dict

All ledger rows between two timestamps: {"acts": {id: merged}, "spans": [...], "power": {srv: [...]}}.

Source code in citar/usage.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def read(since: Optional[float] = None, until: Optional[float] = None) -> dict:
    """All ledger rows between two timestamps: {"acts": {id: merged}, "spans": [...], "power": {srv: [...]}}."""
    acts: dict = {}
    spans: list = []
    power: dict = defaultdict(list)
    if not USAGE_DIR.exists():
        return {"acts": acts, "spans": spans, "power": power}
    lo = _month(since) if since else "0000-00"
    hi = _month(until) if until else "9999-99"
    for p in sorted(USAGE_DIR.glob("*.jsonl")):
        month = p.stem[-7:]
        if not (lo <= month <= hi):
            continue
        try:
            lines = p.read_text(encoding="utf-8").splitlines()
        except OSError:
            continue
        for line in lines:
            try:
                r = json.loads(line)
            except ValueError:
                continue
            k = r.get("k")
            if k == "act":
                cur = acts.setdefault(r["id"], {})
                for key, v in r.items():
                    if isinstance(v, dict) and isinstance(cur.get(key), dict):
                        cur[key] = {**cur[key], **v}
                    elif v is not None:
                        cur[key] = v
                cur.setdefault("first_t", r.get("t"))
            elif k == "span":
                if (since and r["t1"] < since) or (until and r["t0"] > until):
                    continue
                spans.append(r)
            elif k == "pw":
                if (since and r["t1"] < since) or (until and r["t0"] > until):
                    continue
                power[r["srv"]].append(r)
    return {"acts": acts, "spans": spans, "power": dict(power)}

tracker 

tracker() -> Tracker

The usage tracker, created on first use.

Source code in citar/usage.py 
267
268
269
270
271
272
273
def tracker() -> Tracker:
    """The usage tracker, created on first use."""
    global _tracker
    with _tlock:
        if _tracker is None:
            _tracker = Tracker("server")
        return _tracker

session_servers 

session_servers(s) -> dict

What a running game session holds right now: the host (its driver thread's CPU) and each LLM seat's server. Returns None when the session is finished (the tracker then drops it).

Source code in citar/usage.py 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def session_servers(s) -> dict:
    """What a running game session holds right now: the host (its driver thread's CPU) and each LLM seat's server.
    Returns None when the session is finished (the tracker then drops it)."""
    if s.stopped:
        return None
    g = s.game
    if s.paused or g.s.phase != "playing":
        return {}
    from . import servers as S
    out: dict = {}
    reg = S.load()
    host = reg.get("host_server_id")
    if host:
        drv = getattr(s, "_driver", None)
        out[host] = {"threads": [drv.native_id] if drv is not None and drv.native_id else []}
    for seat in s.seats:
        if seat.type == "llm" and seat.llm.get("server_id") and g.player(seat.player).alive:
            out.setdefault(seat.llm["server_id"], {"threads": []})
    return out

track_session 

track_session(
    s,
    kind: str = "game",
    parent: Optional[dict] = None,
    ref: Optional[dict] = None,
    name: str = "",
)

Start (or continue, for a reloaded save) the usage record of a game session.

Source code in citar/usage.py 
298
299
300
301
302
303
304
305
306
307
def track_session(s, kind: str = "game", parent: Optional[dict] = None, ref: Optional[dict] = None, name: str = ""):
    """Start (or continue, for a reloaded save) the usage record of a game session."""
    act = getattr(s, "usage_act", None) or f"game:{s.id}"
    s.usage_act = act
    t = tracker()
    seats = [{"player": seat.player, "type": seat.type, "server_id": seat.llm.get("server_id") if seat.type == "llm" else None,
              "model": seat.llm.get("model") if seat.type == "llm" else None} for seat in s.seats]
    t.activity(act, kind, name or s.name, parent=parent, ref={"game_id": s.id, **(ref or {})}, seats=seats)
    t.register(act, lambda: session_servers(s))
    return act

finish_session 

finish_session(s, **fields)

Close out a game session's ledger entry.

Source code in citar/usage.py 
310
311
312
313
314
315
316
317
318
319
320
def finish_session(s, **fields):
    """Close out a game session's ledger entry."""
    act = getattr(s, "usage_act", None)
    if not act:
        return
    g = s.game
    majors = [p for p in g.s.players if p.kind == "major"]
    t = tracker()
    t.update(act, turn=g.turn, phase=g.s.phase, winner=g.s.winner, victory=g.s.victory,
             ended=time.time() if g.s.phase != "playing" else None,
             alive={str(p.id): p.alive for p in majors}, **fields)

start_power_sampler 

start_power_sampler()

Sample the host machine's power if the registry has a host with sampling on.

Source code in citar/usage.py 
414
415
416
417
418
419
420
421
422
423
424
425
426
def start_power_sampler():
    """Sample the host machine's power if the registry has a host with sampling on."""
    global _sampler
    try:
        from . import servers as S
        h = S.host()
    except Exception:
        return None
    if h is None or _sampler is not None:
        return _sampler
    _sampler = PowerSampler(h["id"])
    _sampler.start()
    return _sampler

lab_game_rows 

lab_game_rows(
    exp: str,
    i: int,
    res: dict,
    host_id: Optional[str],
    labels: Optional[list] = None,
) -> list[dict]

Ledger rows for one finished lab game (called by the lab runner).

Source code in citar/usage.py 
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
def lab_game_rows(exp: str, i: int, res: dict, host_id: Optional[str], labels: Optional[list] = None) -> list[dict]:
    """Ledger rows for one finished lab game (called by the lab runner)."""
    fin = res.get("finished")
    try:
        t1 = datetime.fromisoformat(fin).timestamp() if fin else time.time()
    except ValueError:
        t1 = time.time()
    secs = float(res.get("seconds") or 0)
    t0 = float(res.get("started_ts") or (t1 - secs))
    act = f"lab:{exp}:{i}"
    rows = [{"k": "act", "id": act, "kind": "lab", "name": f"{exp} #{i}", "t": t1,
             "parent": {"kind": "lab_experiment", "id": exp, "name": exp},
             "ref": {"exp": exp, "i": i, "seed": res.get("seed"), "map": res.get("map")},
             "turns": res.get("turns"), "winner_label": res.get("winner_label"), "victory": res.get("victory"),
             "ended": t1, "crash": bool(res.get("crash"))}]
    if host_id and secs > 0:
        rows.append({"k": "span", "act": act, "srv": host_id, "t0": round(t0, 1), "t1": round(t1, 1), "held": round(secs, 1),
                     "cpu": round(float(res.get("cpu_s") or secs), 2), "busy": 0.0, "est_cpu": res.get("cpu_s") is None})
    return rows

citar.costing

Prices the usage ledger (usage.py) with the server registry (servers.py).

Method, per server and per minute: * Fixed costs are charged on a calendar basis: a server's hourly fixed rate is its depreciation (straight-line per component over its lifespan, less resale) + fixed monthly costs (lease, maintenance, API subscription, a share of the electricity bill's fixed fee) spread over the hours of that month. An activity pays that rate for the time it held the server; activities holding it at the same time split the minute. Time nobody held is "unallocated". * Rented machines' hourly usage rate is charged the same way, but only for time in use. * Energy: the machine's power for the minute is measured (GPU watts sampled with nvidia-smi plus CPU utilisation times the server's CPU watts) or, where there are no samples, estimated from what CITAR had it doing (model generation seconds for the GPU, CPU seconds for the CPU). Idle power is split like fixed costs (by time held); the extra "dynamic" power above idle goes to the activities that did work in that minute, by their share of it. The energy price comes from the electricity plan in force (flat, time-of-use or tiered) plus, if chosen, the plan's fixed fee spread over the household's monthly kWh. * API servers: tokens x the model's per-million prices (input, output, cache reads, cache writes). Everything is computed with the configuration in force at the time of use (effective-dated cost periods).

hours_in_month 

hours_in_month(when: datetime) -> float

Hours in the month containing a moment, for spreading fixed costs.

Source code in citar/costing.py 
45
46
47
def hours_in_month(when: datetime) -> float:
    """Hours in the month containing a moment, for spreading fixed costs."""
    return calendar.monthrange(when.year, when.month)[1] * 24.0

depreciation_rate 

depreciation_rate(
    server: dict,
    when: datetime,
    lifespan_override: Optional[float] = None,
) -> float

Currency per hour of calendar time for the server's owned components at when.

Source code in citar/costing.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def depreciation_rate(server: dict, when: datetime, lifespan_override: Optional[float] = None) -> float:
    """Currency per hour of calendar time for the server's owned components at `when`."""
    rate = 0.0
    for c in server.get("components") or []:
        price = c.get("price")
        if not price:
            continue
        years = float(lifespan_override or c.get("lifespan_years") or 4)
        bought = _parse_date(c.get("purchased"))
        retired = _parse_date(c.get("retired") or "")
        if retired and when >= retired:
            continue
        if bought:
            if when < bought or when >= bought + timedelta(days=365.25 * years):
                continue
        rate += max(0.0, price - (c.get("resale") or 0.0)) / (years * HOURS_PER_YEAR)
    return rate

energy_price 

energy_price(
    plan: Optional[dict], when: datetime
) -> tuple[Optional[float], float]

(price per kWh incl. any fixed-fee adder, fee per hour charged by share) for an electricity plan at when.

Source code in citar/costing.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def energy_price(plan: Optional[dict], when: datetime) -> tuple[Optional[float], float]:
    """(price per kWh incl. any fixed-fee adder, fee per hour charged by share) for an electricity plan at `when`."""
    if not plan:
        return None, 0.0
    p = S.period_at(plan["periods"], when)
    if not p:
        return None, 0.0
    rate = p.get("rate_kwh")
    if p["type"] == "tou" and p.get("tou"):
        hit = S._window_hit(p["tou"], when)
        if hit:
            rate = hit[0].get("rate_kwh", rate)
    elif p["type"] == "tiered" and p.get("tiers"):
        usage = p.get("household_kwh_month") or 0
        tiers = sorted(p["tiers"], key=lambda t: (t.get("up_to_kwh") is None, t.get("up_to_kwh") or 0))
        rate = tiers[-1]["rate_kwh"]
        for t in tiers:
            if t.get("up_to_kwh") is None or usage <= t["up_to_kwh"]:
                rate = t["rate_kwh"]
                break
    fee_hourly = 0.0
    if p.get("fixed_monthly"):
        if p["fee_allocation"] == "household_kwh" and p.get("household_kwh_month"):
            rate = (rate or 0.0) + p["fixed_monthly"] / p["household_kwh_month"]
        elif p["fee_allocation"] == "share":
            fee_hourly = p["fixed_monthly"] * (p.get("share_pct") or 0) / 100.0 / hours_in_month(when)
    return rate, fee_hourly

fixed_rates 

fixed_rates(
    server: dict,
    reg: dict,
    when: datetime,
    lifespan_override: Optional[float] = None,
) -> dict

Currency per hour: depreciation, other fixed (lease, maintenance, subscription, electricity fee share) and the rented-machine usage rate; plus the energy price per kWh.

Source code in citar/costing.py 
 98
 99
100
101
102
103
104
105
106
107
108
def fixed_rates(server: dict, reg: dict, when: datetime, lifespan_override: Optional[float] = None) -> dict:
    """Currency per hour: depreciation, other fixed (lease, maintenance, subscription, electricity fee share) and the
    rented-machine usage rate; plus the energy price per kWh."""
    per = S.period_at(server["costs"], when) or {}
    hm = hours_in_month(when)
    plan = next((p for p in reg["electricity_plans"] if p["id"] == per.get("electricity_plan_id")), None)
    kwh_price, fee_hourly = energy_price(plan, when)
    return {"depreciation": depreciation_rate(server, when, lifespan_override) if server["kind"] == "owned" else 0.0,
            "fixed": (per.get("fixed_monthly") or 0.0) / hm + (per.get("api_fixed_monthly") or 0.0) / hm + fee_hourly,
            "hourly": (per.get("hourly_rate") or 0.0) if server["kind"] == "leased" else 0.0,
            "kwh": kwh_price, "plan": plan["name"] if plan else None}

total 

total(c: dict, basis: str = 'full') -> float

The total of a cost breakdown, on the chosen energy basis.

full counts all the energy the machine drew; marginal counts only the extra caused by the work. Both are defensible and they differ by a lot, which is why it is a choice rather than a constant.

Source code in citar/costing.py 
124
125
126
127
128
129
130
131
132
133
134
def total(c: dict, basis: str = "full") -> float:
    """The total of a cost breakdown, on the chosen energy basis.

    ``full`` counts all the energy the machine drew; ``marginal`` counts only the extra caused by the
    work. Both are defensible and they differ by a lot, which is why it is a choice rather than a
    constant.
    """
    t = c["depreciation"] + c["fixed"] + c["hourly"] + c["energy_dynamic"] + c["tokens"]
    if basis == "full":
        t += c["energy_idle"]
    return t

compute 

compute(
    since: Optional[float] = None,
    until: Optional[float] = None,
    reg: Optional[dict] = None,
    ledger: Optional[dict] = None,
    lifespan_override: Optional[float] = None,
) -> dict

Cost every activity in [since, until]. Returns { "acts": {act_id: {"info": act row, "servers": {srv: cost}, "models": {model: {...}}}}, "servers": {srv: {"allocated": cost, "calendar": {...}, "power": {...}}}, "days": {date: {srv: cost}}, "notes": [...]}

Source code in citar/costing.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def compute(since: Optional[float] = None, until: Optional[float] = None, reg: Optional[dict] = None,
            ledger: Optional[dict] = None, lifespan_override: Optional[float] = None) -> dict:
    """Cost every activity in [since, until]. Returns {
        "acts": {act_id: {"info": act row, "servers": {srv: cost}, "models": {model: {...}}}},
        "servers": {srv: {"allocated": cost, "calendar": {...}, "power": {...}}},
        "days": {date: {srv: cost}}, "notes": [...]}"""
    reg = reg or S.with_pooled()
    ledger = ledger or U.read(since, until)
    servers = {s["id"]: s for s in reg["servers"]}
    aliases = reg.get("aliases") or {}           # a re-registered machine's old ids -> its current one
    acts_out: dict = {}
    notes: set = set()

    def act_entry(act_id):
        """The cost entry for one activity, created on first use."""
        if act_id not in acts_out:
            acts_out[act_id] = {"info": ledger["acts"].get(act_id) or {"id": act_id, "kind": "unknown"},
                                "servers": defaultdict(_new_cost), "models": defaultdict(_new_cost)}
        return acts_out[act_id]

    # ---- per server, per minute: who held it and what work they did
    days: dict = defaultdict(lambda: defaultdict(float))     # date -> srv -> cost (full basis)
    hours: dict = defaultdict(lambda: defaultdict(float))    # "YYYY-MM-DD HH:00" -> srv -> cost
    buckets: dict = defaultdict(lambda: defaultdict(lambda: defaultdict(lambda: [0.0, 0.0, 0.0])))   # srv -> m -> act -> [held, busy, cpu]
    for sp in ledger["spans"]:
        srv = aliases.get(sp.get("srv"), sp.get("srv")) or "unassigned"
        t0, t1 = float(sp["t0"]), float(sp["t1"])
        if since:
            t0c = max(t0, since)
        else:
            t0c = t0
        t1c = min(t1, until) if until else t1
        dur = max(1e-6, t1 - t0)
        frac_range = max(0.0, (t1c - t0c) / dur)
        if frac_range <= 0:
            continue
        a = act_entry(sp["act"])
        sc = a["servers"][srv]
        model = sp.get("model")
        mc = a["models"][model] if model and model != "?" else None
        # tokens are priced at the span's time on API servers
        toks = {f: (sp.get(f) or 0) * frac_range for f in U.TOKEN_FIELDS}
        for f, v in toks.items():
            sc[f] += v
            if mc is not None:
                mc[f] += v
        sv = servers.get(srv)
        busy = (sp.get("busy") or 0.0) * frac_range
        if mc is not None:
            mc["busy_h"] += busy / 3600
        if sv and sv["kind"] == "api" and model:
            price = S.model_price(sv, model, _dt(t1))
            if price:
                c = (toks["in"] * price["input"] + toks["out"] * price["output"] + toks["cr"] * price.get("cache_read", 0)
                     + toks["cw"] * price.get("cache_write", 0)) / 1e6
                sc["tokens"] += c
                if mc is not None:
                    mc["tokens"] += c
                days[_dt(t1).strftime("%Y-%m-%d")][srv] += c
                hours[_dt(t1).strftime("%Y-%m-%d %H:00")][srv] += c
            elif toks["in"] or toks["out"]:
                sc["unpriced_tokens"] += toks["in"] + toks["out"]
                notes.add(f"No token price for {model} on {sv['name']}; its tokens are uncosted.")
        if sv is None:
            if srv == "unassigned":
                notes.add("Some model usage has no server (a seat not chosen from the Servers list); it is uncosted.")
            else:
                notes.add(f"Usage on a deleted server ({srv}) is uncosted.")
        # spread held/busy/cpu over the minutes of the span
        held, cpu = (sp.get("held") or 0.0), (sp.get("cpu") or 0.0)
        if not (held or busy or cpu):
            continue
        m0, m1 = int(t0c // BUCKET), int(max(t0c, t1c - 1e-6) // BUCKET)
        for m in range(m0, m1 + 1):
            lo, hi = max(t0c, m * BUCKET), min(t1c, (m + 1) * BUCKET)
            if hi <= lo:
                continue
            f = (hi - lo) / dur
            cell = buckets[srv][m][sp["act"]]
            cell[0] += held * f
            cell[1] += busy * f
            cell[2] += cpu * f

    # ---- measured power, per server per minute
    power: dict = {}
    for srv, rows in (ledger.get("power") or {}).items():
        srv = aliases.get(srv, srv)
        sv = servers.get(srv)
        if not sv:
            continue
        gpu_vals = sorted(r["gpu_w"] for r in rows if r.get("gpu_w") is not None)
        gpu_idle = gpu_vals[int(len(gpu_vals) * 0.05)] if gpu_vals else 0.0
        by_min = {}
        for r in rows:          # sample windows don't line up with clock minutes: cover every minute they overlap
            for m in range(int(r["t0"] // BUCKET), int((r["t1"] - 1e-6) // BUCKET) + 1):
                by_min.setdefault(m, r)
        power[srv] = {"rows": by_min, "raw": rows, "gpu_idle": gpu_idle}

    server_out: dict = {}
    rate_cache: dict = {}

    def rates(sv, m):
        """The rates applying to a server and model at this moment."""
        key = (sv["id"], m // 60)          # rates are hourly-stable
        if key not in rate_cache:
            rate_cache[key] = fixed_rates(sv, reg, _dt(m * BUCKET), lifespan_override)
        return rate_cache[key]

    for srv, mins in buckets.items():
        sv = servers.get(srv)
        if not sv:
            continue
        pw = sv["power"]
        idle_w = pw.get("idle_w")
        cpu_max = pw.get("cpu_max_w") or 0.0
        gpu_max = pw.get("gpu_max_w") or 0.0
        oh = 1 + (pw.get("measured_overhead_pct") or 0) / 100.0
        threads = _threads(sv)
        parallel = max(1, sv["connection"].get("max_parallel") or 1)
        agg = server_out.setdefault(srv, {"allocated": _new_cost(), "active_minutes": 0, "measured_minutes": 0})
        meas = power.get(srv)
        for m, cells in mins.items():
            r = rates(sv, m)
            H = sum(c[0] for c in cells.values())
            if H <= 0:
                continue
            agg["active_minutes"] += 1
            scale = min(1.0, BUCKET / H)          # concurrent holders split the minute
            day = _dt(m * BUCKET).strftime("%Y-%m-%d")
            # power for this minute
            energy_ok = idle_w is not None and sv["kind"] in ("owned", "leased")
            dyn_w = 0.0
            measured = False
            if energy_ok:
                row = meas["rows"].get(m) if meas else None
                if row is not None:
                    measured = True
                    agg["measured_minutes"] += 1
                    dyn_w = (row.get("cpu_util") or 0.0) * cpu_max
                    if row.get("gpu_w") is not None:
                        dyn_w += max(0.0, row["gpu_w"] - meas["gpu_idle"]) * oh
                else:
                    busy_total = sum(c[1] for c in cells.values())
                    cpu_total = sum(c[2] for c in cells.values())
                    dyn_w = min(1.0, busy_total / (BUCKET * parallel)) * gpu_max + min(1.0, cpu_total / (BUCKET * threads)) * cpu_max
            W = sum(c[1] + c[2] for c in cells.values())
            for act, (held, busy, cpu) in cells.items():
                share = held * scale                   # seconds of the minute this activity is charged for
                c = acts_out[act]["servers"][srv]
                c["held_h"] += held / 3600
                c["busy_h"] += busy / 3600
                c["cpu_h"] += cpu / 3600
                c["depreciation"] += r["depreciation"] * share / 3600
                c["fixed"] += r["fixed"] * share / 3600
                c["hourly"] += r["hourly"] * share / 3600
                if energy_ok:
                    kwh_idle = idle_w * share / 3600 / 1000
                    wshare = (busy + cpu) / W if W > 0 else held / H
                    kwh_dyn = dyn_w * BUCKET / 3600 / 1000 * wshare
                    c["kwh_idle"] += kwh_idle
                    c["kwh_dynamic"] += kwh_dyn
                    if r["kwh"] is None:
                        c["unpriced_kwh"] += kwh_idle + kwh_dyn
                    else:
                        c["energy_idle"] += kwh_idle * r["kwh"]
                        c["energy_dynamic"] += kwh_dyn * r["kwh"]
                    if measured:
                        c["energy_measured_h"] += share / 3600
                    else:
                        c["energy_estimated_h"] += share / 3600
                spent = (r["depreciation"] + r["fixed"] + r["hourly"]) * share / 3600
                if energy_ok and r["kwh"] is not None:
                    spent += (kwh_idle + kwh_dyn) * r["kwh"]
                days[day][srv] += spent
                hours[_dt(m * BUCKET).strftime("%Y-%m-%d %H:00")][srv] += spent

    # ---- roll up: per activity totals, per server allocated, per day energy/tokens
    for act, a in acts_out.items():
        a["servers"] = {k: dict(v) for k, v in a["servers"].items()}
        a["models"] = {k: dict(v) for k, v in a["models"].items()}
        tot = _new_cost()
        for srv, c in a["servers"].items():
            for k in tot:
                tot[k] += c[k]
            agg = server_out.setdefault(srv, {"allocated": _new_cost(), "active_minutes": 0, "measured_minutes": 0})
            for k in tot:
                agg["allocated"][k] += c[k]
        a["total"] = tot

    # ---- calendar view of each server over the range: what its fixed costs were, allocated or not
    if since and until:
        for sv in reg["servers"]:
            cal = {"depreciation": 0.0, "fixed": 0.0, "hours": (until - since) / 3600}
            t = since
            while t < until:
                step = min(3600.0, until - t)
                r = fixed_rates(sv, reg, _dt(t), lifespan_override)
                cal["depreciation"] += r["depreciation"] * step / 3600
                cal["fixed"] += r["fixed"] * step / 3600
                t += step
            out = server_out.setdefault(sv["id"], {"allocated": _new_cost(), "active_minutes": 0, "measured_minutes": 0})
            out["calendar"] = cal
            out["utilization"] = min(1.0, out["active_minutes"] / 60 / max(1e-9, cal["hours"]))
            meas = power.get(sv["id"])
            if meas and sv["power"].get("idle_w") is not None:
                kwh = 0.0
                cost = 0.0
                for row in meas["raw"]:
                    m = int(row["t0"] // BUCKET)
                    w = sv["power"]["idle_w"] + (row.get("cpu_util") or 0) * (sv["power"].get("cpu_max_w") or 0)
                    if row.get("gpu_w") is not None:
                        w += max(0.0, row["gpu_w"] - meas["gpu_idle"]) * (1 + (sv["power"].get("measured_overhead_pct") or 0) / 100)
                    e = w * (row["t1"] - row["t0"]) / 3600 / 1000
                    kwh += e
                    price = rates(sv, m)["kwh"]
                    cost += e * (price or 0)
                out["metered"] = {"kwh": kwh, "cost": cost, "minutes": len(meas["raw"]), "gpu_idle_w": meas["gpu_idle"]}
    return {"acts": acts_out, "servers": server_out, "days": {d: dict(v) for d, v in days.items()},
            "hours": {h: dict(v) for h, v in hours.items()},
            "notes": sorted(notes), "currency": reg.get("currency", "USD"), "symbol": reg.get("currency_symbol", "$")}

hourly_profile 

hourly_profile(
    server: dict,
    reg: dict,
    when: Optional[datetime] = None,
    busy_fraction: float = 1.0,
) -> dict

What an hour on this server costs at when with the model busy busy_fraction of the time (for what-if comparisons): fixed + depreciation + usage rate + estimated energy.

Source code in citar/costing.py 
359
360
361
362
363
364
365
366
367
368
369
370
def hourly_profile(server: dict, reg: dict, when: Optional[datetime] = None, busy_fraction: float = 1.0) -> dict:
    """What an hour on this server costs at `when` with the model busy `busy_fraction` of the time (for what-if
    comparisons): fixed + depreciation + usage rate + estimated energy."""
    when = when or datetime.now()
    r = fixed_rates(server, reg, when)
    pw = server["power"]
    energy = None
    if pw.get("idle_w") is not None and server["kind"] in ("owned", "leased"):
        watts = pw["idle_w"] + busy_fraction * (pw.get("gpu_max_w") or 0) + 0.1 * (pw.get("cpu_max_w") or 0)
        energy = watts / 1000 * (r["kwh"] or 0)
    return {"depreciation": r["depreciation"], "fixed": r["fixed"], "hourly": r["hourly"], "energy": energy,
            "total": r["depreciation"] + r["fixed"] + r["hourly"] + (energy or 0), "kwh_price": r["kwh"]}

price_tokens 

price_tokens(price: dict, toks: dict) -> float

What a number of tokens costs at a model's prices.

Source code in citar/costing.py 
373
374
375
376
def price_tokens(price: dict, toks: dict) -> float:
    """What a number of tokens costs at a model's prices."""
    return (toks.get("in", 0) * price["input"] + toks.get("out", 0) * price["output"]
            + toks.get("cr", 0) * price.get("cache_read", 0) + toks.get("cw", 0) * price.get("cache_write", 0)) / 1e6

Hardware

citar.hwinfo

CITAR hardware collector: what a machine has (CPU, GPUs, memory, disks, local model runtimes) and a first guess at its power draw, as JSON for the Servers page.

This file is deliberately standalone (standard library only; uses psutil when installed) so it can be copied to any Windows, Linux or macOS machine and run there:

python collect_hardware.py > my-machine.json        # then import the file on the Servers page

On the machine that runs CITAR, the Servers page calls collect() directly.

collect 

collect() -> dict

Everything this machine is, as one dictionary.

Standard library only, so it can be downloaded as a single file and run on a machine that has nothing else installed - which is exactly the case it exists for.

Source code in citar/hwinfo.py 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
def collect() -> dict:
    """Everything this machine is, as one dictionary.

    Standard library only, so it can be downloaded as a single file and run on a machine that has
    nothing else installed - which is exactly the case it exists for.
    """
    info = {"format": "citar-hardware", "collector_version": COLLECTOR_VERSION,
            "collected_at": time.strftime("%Y-%m-%dT%H:%M:%S"), "hostname": socket.gethostname(),
            "os": {"system": platform.system(), "release": platform.release(), "version": platform.version(),
                   "platform": platform.platform(), "arch": platform.machine()},
            "system": {}, "cpu": {"model": platform.processor() or None, "cores": None, "threads": os.cpu_count()},
            "memory": {"ram_gb": None, "unified": False}, "gpus": _nvidia(), "physical_disks": [], "volumes": [],
            "runtimes": [], "power": {"gpu_sampling": any(g.get("power_readable") for g in _nvidia())}}
    ps = _psutil()
    if ps:
        info["cpu"]["cores"] = ps.cpu_count(logical=False)
        info["cpu"]["threads"] = ps.cpu_count(logical=True)
        info["memory"]["ram_gb"] = _gb(ps.virtual_memory().total)
        try:
            freq = ps.cpu_freq()
            if freq and freq.max:
                info["cpu"]["max_ghz"] = round(freq.max / 1000, 2)
        except Exception:
            pass
        try:
            if ps.sensors_battery() is not None:
                info["system"]["form"] = "laptop"
        except Exception:
            pass
    try:
        if sys.platform.startswith("win"):
            _windows(info)
        elif sys.platform == "darwin":
            _macos(info)
        else:
            _linux(info)
    except Exception as e:      # partial data beats none
        info["warnings"] = [f"{type(e).__name__}: {e}"]
    _volumes(info)
    _runtimes(info)
    info["power"]["suggested"] = guess_power(info)
    return info

guess_power 

guess_power(info: dict) -> dict

Rough wall-power figures to start from; replace them with measured numbers (a plug-in power meter) if you can.

Source code in citar/hwinfo.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
def guess_power(info: dict) -> dict:
    """Rough wall-power figures to start from; replace them with measured numbers (a plug-in power meter) if you can."""
    form = info["system"].get("form") or "desktop"
    cpu = (info["cpu"].get("model") or "").lower()
    threads = info["cpu"].get("threads") or 8
    if info["memory"].get("unified"):
        idle, cpu_max = 8, 30 if form == "laptop" else 60
    elif form == "laptop":
        idle = 12
        cpu_max = 55 if re.search(r"\b\d{4,5}hx?\b|hx\b|ryzen \d \d{4}h", cpu) else 25
    else:
        idle = 55
        cpu_max = 65 if threads <= 8 else 125 if threads <= 24 else 200
        if "threadripper" in cpu or "xeon" in cpu or "epyc" in cpu:
            idle, cpu_max = 110, 280
    gpu_max = 0.0
    for g in info["gpus"]:
        if g.get("power_limit_w"):
            gpu_max += 0.85 * g["power_limit_w"]
        elif g.get("vendor") in ("NVIDIA", "AMD") and (g.get("vram_gb") or 0) >= 6:
            gpu_max += 80 if form == "laptop" else 220
    if info["memory"].get("unified"):
        gpu_max = max(gpu_max, 25 if form == "laptop" else 60)
    return {"idle_w": idle, "cpu_max_w": cpu_max, "gpu_max_w": round(gpu_max), "source": "guess",
            "note": "Estimated from the hardware class; measure idle and busy wall power with a plug-in meter for better numbers."}

Keys

citar.keystore

API key storage for servers. Keys never live in the project folder (it may sync to the cloud) and are never sent back to the browser: the web form posts a key once, and afterwards the UI only learns whether one is stored.

Backends (chosen per server):

keyring  the operating system's credential store via the `keyring` package: Windows Credential Manager,
         macOS Keychain, or the Linux Secret Service / KWallet. The default when available.
env      an environment variable named in the server config (nothing is stored by CITAR).
file     a file in the user's config directory (outside the project) encrypted with a passphrase
         (scrypt + Fernet). For headless Linux boxes without a keyring. The passphrase comes from
         CITAR_KEYS_PASSPHRASE or is entered in the web UI once per server start (kept in memory only).

backends 

backends() -> dict

Which backends work on this machine, for the Servers page.

Source code in citar/keystore.py 
28
29
30
31
32
33
34
35
36
37
38
def backends() -> dict:
    """Which backends work on this machine, for the Servers page."""
    kr = _keyring()
    return {
        "keyring": {"available": kr is not None, "name": _keyring_name(kr),
                    "note": None if kr else "Install the 'keyring' package (pip install keyring) to use the OS credential store."},
        "env": {"available": True, "name": "Environment variable"},
        "file": {"available": _fernet_available(), "name": f"Encrypted file ({_file_path()})",
                 "unlocked": _passphrase is not None, "exists": _file_path().exists(),
                 "note": None if _fernet_available() else "Needs the 'cryptography' package."},
    }

store 

store(server_id: str, backend: str, secret: str)

Store a key for a server. The env backend can't store anything (set the variable yourself).

Source code in citar/keystore.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def store(server_id: str, backend: str, secret: str):
    """Store a key for a server. The env backend can't store anything (set the variable yourself)."""
    secret = (secret or "").strip()
    if not secret:
        raise ValueError("The key is empty.")
    with _lock:
        if backend == "keyring":
            kr = _keyring()
            if kr is None:
                raise ValueError("No OS credential store is available here; use an environment variable or the encrypted file.")
            kr.set_password(SERVICE, _account(server_id), secret)
        elif backend == "file":
            if _passphrase is None:
                raise ValueError("Unlock the encrypted key file with its passphrase first.")
            keys = _file_read(_passphrase)
            keys[_account(server_id)] = secret
            _file_write(_passphrase, keys)
        else:
            raise ValueError("Keys for the env backend are read from the environment variable; nothing to store.")

get 

get(server: dict) -> Optional[str]

The API key for a server config (its connection.key block), or None.

Source code in citar/keystore.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def get(server: dict) -> Optional[str]:
    """The API key for a server config (its `connection.key` block), or None."""
    k = ((server.get("connection") or {}).get("key") or {})
    backend = k.get("backend") or "none"
    try:
        if backend == "keyring":
            kr = _keyring()
            return kr.get_password(SERVICE, _account(server["id"])) if kr else None
        if backend == "env":
            return os.environ.get(k.get("env") or "") or None
        if backend == "file":
            if _passphrase is None:
                return None
            return _file_read(_passphrase).get(_account(server["id"]))
    except Exception:
        return None
    return None

delete 

delete(server_id: str, backend: str)

Remove a stored key.

Source code in citar/keystore.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def delete(server_id: str, backend: str):
    """Remove a stored key."""
    with _lock:
        if backend == "keyring":
            kr = _keyring()
            if kr is not None:
                try:
                    kr.delete_password(SERVICE, _account(server_id))
                except Exception:
                    pass
        elif backend == "file" and _passphrase is not None:
            keys = _file_read(_passphrase)
            if keys.pop(_account(server_id), None) is not None:
                _file_write(_passphrase, keys)

status 

status(server: dict) -> dict

What the UI may know: whether a key is present (never the key itself), plus a masked hint.

Source code in citar/keystore.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def status(server: dict) -> dict:
    """What the UI may know: whether a key is present (never the key itself), plus a masked hint."""
    k = ((server.get("connection") or {}).get("key") or {})
    backend = k.get("backend") or "none"
    if backend == "none":
        return {"backend": "none", "present": False}
    key = get(server)
    out = {"backend": backend, "present": bool(key)}
    if key and len(key) > 8:
        out["hint"] = "…" + key[-4:]
    if backend == "env":
        out["env"] = k.get("env")
    if backend == "file" and _passphrase is None:
        out["locked"] = True
    return out

unlock 

unlock(passphrase: str) -> bool

Remember the file passphrase for this server process (checked against the file when one exists).

Source code in citar/keystore.py 
135
136
137
138
139
140
141
142
143
def unlock(passphrase: str) -> bool:
    """Remember the file passphrase for this server process (checked against the file when one exists)."""
    global _passphrase
    if not passphrase:
        raise ValueError("Enter a passphrase.")
    if _file_path().exists():
        _file_read(passphrase)      # raises on a wrong passphrase
    _passphrase = passphrase
    return True

lock 

lock()

Forget the passphrase, so the key file cannot be read again without it.

Source code in citar/keystore.py 
146
147
148
149
def lock():
    """Forget the passphrase, so the key file cannot be read again without it."""
    global _passphrase
    _passphrase = None