Merge pull request 'Dev' (#61) from dev into main

Reviewed-on: #61

CLAUDE.md

@@ -42,6 +42,7 @@ Tests use [tests/conftest.py](tests/conftest.py): the `mock_esi` fixture install
 
 ## Git / CI workflow
 
-- Work happens on `dev`; **`main` is not touched until the first complete version** (it triggers the real release publish). Feature branches → PR → fast-forward merge into `dev` → delete branch.
+- Work happens on `dev`. Feature branches → PR → fast-forward merge into `dev` → delete branch. Cut a release by bumping the version in `pyproject.toml` / `src/familiar/__init__.py` and opening a `dev → main` fast-forward PR (publishes the clean version).
 - `main` and `dev` are protected: no direct pushes, **signed commits required** (the user's git auto-signs), merge via PR.
 - CI (`.gitea/workflows/ci.yml`) runs lint + test on every PR/push; on a merge to main/dev it builds and publishes to the Gitea PyPI registry (auth via the `REGISTRY_TOKEN` secret; twine must NOT use `--skip-existing`). dev publishes a unique `.devN` pre-release.
+- **Do NOT attribute commits, merges, or PRs to Claude / Claude Code.** No `Co-Authored-By: Claude ...` trailer on commits and no "Generated with Claude Code" footer in PR descriptions. Keep messages plain and authored by the user.

README.md

@@ -33,6 +33,17 @@ uv run familiar status  # list logged-in characters
 
 Re-run `login` for each character (switch EVE accounts in the browser as needed); they all live in one encrypted store. Changing the requested scopes requires logging in again.
 
+### Offline SDE (optional)
+
+For network-free item / system / station lookups, install a local copy of EVE's Static Data Export (the Fuzzwork SQLite conversion):
+
+```bash
+uv run familiar sde download   # download + decompress the SDE (a few hundred MB)
+uv run familiar sde status     # show whether it's installed and where
+```
+
+Once present, `get_item` / `get_system` / `get_station` serve from it without hitting ESI; familiar falls back to ESI whenever the SDE is absent or doesn't know the id.
+
 ### Connect to Claude
 
 Claude Code:

doc/README.md

@@ -0,0 +1,52 @@
+# familiar — Documentation
+
+> Your familiar across New Eden — an [MCP](https://modelcontextprotocol.io) server that gives a local AI a complete window into your EVE Online account and the New Eden universe, plus opt-in hands to act on it.
+
+This folder contains the full English documentation for the **familiar** project. Start
+with [Getting Started](getting-started.md) if you just want to run it, or [Architecture](architecture.md)
+if you want to understand or extend the code.
+
+## What is familiar?
+
+`familiar` is a local-first MCP server, written in Python with
+[FastMCP](https://github.com/jlowin/fastmcp), that connects a local AI assistant
+(Claude Code, Claude Desktop, Cowork, or any MCP client) to two data sources:
+
+- **ESI** — the [EVE Swagger Interface](https://esi.evetech.net/), CCP's official REST API. Public
+  endpoints (market, universe, industry indices) need no login; character **and corp** data flow
+  through EVE SSO with read **and write** OAuth2 scopes.
+- **SDE** — the [Static Data Export](https://developers.eveonline.com/resource/resources), EVE's
+  static dataset of items, systems and stations, optionally cached locally for fast offline lookups.
+
+It exposes roughly **80 tools** across market, universe, account, industry, trade, structures,
+killmails, profile, corp, communications, contacts, fittings, fleet, UI, and planetary interaction —
+the vast majority read-only, with a small set of **guarded write actions** (mail, waypoints, contacts,
+fittings, fleet, UI windows, PI) gated behind a safety choke point.
+
+## Design principles
+
+- **Safe by default.** Reading is free. Every *write* is previewed before it runs (nothing happens
+  without `confirm=true`), recorded in an audit log, and can be turned off entirely. See
+  [Write Safety](write-safety.md).
+- **Local-first.** Runs on your own machine. Your refresh token and account data never leave your
+  control — the token store is Fernet-encrypted on disk.
+- **Everything, eventually.** The goal is full ESI coverage (character and corp, read and write) plus
+  the complete static universe, built in phases so the project stays shippable.
+
+## Documentation map
+
+| Document | What's in it |
+| --- | --- |
+| [Getting Started](getting-started.md) | Install with `uv`, log in via EVE SSO, connect to Claude, first queries. |
+| [Architecture](architecture.md) | How the server is assembled, the ESI/auth/charutil/writes layers, the module pattern. |
+| [Authentication](authentication.md) | EVE SSO PKCE flow, the encrypted multi-character store, scopes, CLI commands. |
+| [Write Safety](write-safety.md) | The `writes.guard` choke point, dry-run previews, the audit log, enabling/disabling. |
+| [Tool Reference](tool-reference.md) | Every MCP tool, grouped by module, with parameters and descriptions. |
+| [Configuration](configuration.md) | Every environment variable and on-disk file. |
+| [Contributing & Development](contributing.md) | Dev setup, testing conventions, lint/format, the Git/CI workflow. |
+
+## Quick links
+
+- Project root [README.md](../README.md) — the marketing-facing overview and roadmap.
+- [CLAUDE.md](../CLAUDE.md) — guidance for AI coding assistants working in this repo.
+- License: [MIT](../LICENSE).

doc/architecture.md

@@ -0,0 +1,154 @@
+# Architecture
+
+familiar is a small, layered codebase. Every authenticated tool ultimately rests on four shared
+pieces — the server assembly, the ESI client, the auth/token store, and the per-character helpers —
+with a single write-safety choke point on top. This document explains each layer and the conventions
+that keep ~80 tools consistent.
+
+```
+              ┌──────────────────────────────────────────────────┐
+   MCP client │  Claude Code / Claude Desktop / Cowork (stdio)    │
+              └───────────────────────┬──────────────────────────┘
+                                      │  MCP over stdio
+              ┌───────────────────────▼──────────────────────────┐
+              │  server.py — one FastMCP instance, ping + register │
+              └───────────────────────┬──────────────────────────┘
+        register(mcp) on each tool module (market, universe, account, …)
+              ┌───────────────────────┼──────────────────────────┐
+              │                       │                          │
+     ┌────────▼────────┐    ┌─────────▼─────────┐    ┌───────────▼──────────┐
+     │  charutil.py    │    │   writes.py       │    │  lookup.py / sde.py   │
+     │ run_per_*       │    │  guard() gate     │    │  name → id resolution │
+     └────────┬────────┘    └─────────┬─────────┘    └───────────┬──────────┘
+              │                       │                          │
+     ┌────────▼────────┐    ┌─────────▼─────────┐                │
+     │   auth.py       │    │   esi.py / zkill  │◄───────────────┘
+     │ EVE SSO + store │    │  async HTTP wrap  │
+     └─────────────────┘    └───────────────────┘
+```
+
+## Server assembly — `server.py`
+
+[`server.py`](../src/familiar/server.py) creates the single `FastMCP` instance and calls
+`register(mcp)` on each tool module in turn. It also defines the one built-in tool, `ping`, a health
+check that returns the server name, version and UTC time.
+
+Adding tools is mechanical: **write a new module, import it in `server.py`, and call its
+`register(mcp)`.** Nothing else wires it up.
+
+> Note: the `instructions=` string passed to `FastMCP` still describes familiar as "read-only". That
+> text predates the write actions and is historical framing; the real behaviour is read **and**
+> guarded write (see [Write Safety](write-safety.md)).
+
+## The tool-module pattern
+
+Every tool module — market, universe, account, industry, trade, structures, killmails, profile, corp,
+corp_economy, misc, comms, contacts, fittings, fleet, ui, planets, writes — follows the same shape:
+
+```python
+# 1. module-level async functions = the tools (plain, importable, testable)
+async def get_something(character: str | int | None = None) -> dict[str, Any]:
+    """One-line docstring becomes the tool description the AI sees."""
+    ...
+
+# 2. a TOOLS list naming the functions to expose
+TOOLS = [get_something, ...]
+
+# 3. register() loops mcp.tool(fn)
+def register(mcp: FastMCP) -> None:
+    for fn in TOOLS:
+        mcp.tool(fn)
+```
+
+The key idea: **tools are plain functions; registration is separate.** Tests import and call the
+functions directly without any MCP machinery. The function's docstring is what the AI reads to decide
+when to call it, so docstrings are written for that audience.
+
+## ESI access — `esi.py`
+
+[`esi.py`](../src/familiar/esi.py) is a thin async wrapper around `httpx` for the
+[EVE Swagger Interface](https://esi.evetech.net/):
+
+- A shared singleton `EsiClient` via `client()`; `set_client()` lets tests inject an
+  `httpx.MockTransport`.
+- `get` / `post` / `put` / `delete`, plus `get_all_pages` which follows the `X-Pages` header to
+  concatenate every page of a paginated endpoint.
+- Bearer auth via a `token=` argument for authenticated calls; public calls omit it.
+- A uniform `EsiError` carrying the HTTP status.
+- A descriptive `User-Agent` (ESI asks for one).
+- **Response caching** honoring the ESI `Expires` header — GET responses are cached until they
+  expire (disable with `FAMILIAR_DISABLE_CACHE`).
+- **Error-limit back-off** — ESI publishes a rolling error budget in response headers; when it nears
+  exhaustion (`FAMILIAR_ERROR_LIMIT_THRESHOLD`, default 5) the client waits for the window to reset
+  before issuing more requests.
+
+[`zkill.py`](../src/familiar/zkill.py) mirrors this design for the public
+[zKillboard](https://zkillboard.com/) API, used by the killmail tools.
+
+## Authentication — `auth.py`
+
+[`auth.py`](../src/familiar/auth.py) implements EVE SSO with PKCE (no client secret) and a
+**multi-character** encrypted store. Full details in [Authentication](authentication.md). The
+short version:
+
+- `interactive_login()` runs the browser flow once per character and persists a bundle.
+- The store is a dict of characters keyed by id, in one Fernet-encrypted file; a legacy
+  single-character file is migrated on load.
+- `get_auth(character_id)` returns a valid access token, transparently refreshing via the rotating
+  refresh token when the current one is within 60s of expiry.
+- `list_characters` / `resolve_character` / `remove_character` manage the store; `status()` reports
+  it without any network call.
+
+Config (client id, callback host/port, scopes, data dir) lives in
+[`config.py`](../src/familiar/config.py), all environment-overridable — see
+[Configuration](configuration.md).
+
+## Per-character / per-corp pattern — `charutil.py`
+
+[`charutil.py`](../src/familiar/charutil.py) is the backbone of every authenticated tool. Authenticated
+read tools take an **optional** `character` argument and run over one named character or **all**
+logged-in ones:
+
+- `run_per_character(character, fetch)` — resolves the target(s), gets a refreshed token for each,
+  runs `fetch(token, character_id)`, and returns a uniform `{"characters": [...]}` list. Each result
+  is tagged with `character_id` and `character_name`.
+- `run_per_corp(character, fetch)` — same, but also resolves each character's `corporation_id` first
+  (corp endpoints 403 without the right corp role; that surfaces as an `EsiError` for the affected
+  character). `fetch` receives `(token, corporation_id, character_id)`.
+- `one_token(character)` — picks a single `(token, character_id)` for lookups about a *thing* (e.g. a
+  structure) rather than per character.
+- `character_targets(character)` — `(id, name)` pairs without needing a token, for public lookups
+  keyed by character id (e.g. zKillboard).
+- `summarize(items, limit)` — the uniform `{count, sample}` shape used to cap large lists.
+
+Because of this layer, a typical read tool is just a few lines: resolve names to ids, call ESI inside
+a small `fetch`, hand it to `run_per_character`.
+
+## Name resolution — `lookup.py`, `hubs.py`, `sde.py`
+
+- [`lookup.py`](../src/familiar/lookup.py) resolves names to ids via ESI's `/universe/ids/` endpoint,
+  so a user can say "Tritanium" or "Jita" instead of a numeric id. `resolve_id` short-circuits if the
+  value is already numeric and prefers the offline SDE for inventory types when one is present.
+- [`hubs.py`](../src/familiar/hubs.py) holds the static table of NPC trade hubs (Jita, Amarr, Dodixie,
+  Rens, Hek) with their region/system/station ids, used by the market tools.
+- [`sde.py`](../src/familiar/sde.py) is an **optional** local Static Data Export store. When a prepared
+  SQLite copy is on disk (`FAMILIAR_SDE_PATH` or `<data_dir>/sde.sqlite`), familiar resolves type
+  names and serves item details offline. It degrades gracefully: if the file or a needed table is
+  absent, `available()` reports `False` and callers fall back to ESI. `sde_status` is the diagnostic
+  tool reporting this.
+
+## Write safety — `writes.py`
+
+[`writes.py`](../src/familiar/writes.py) provides `guard(action, params, confirm, execute)`, the single
+gate every mutating tool calls. It is the only place a write to ESI is permitted. See
+[Write Safety](write-safety.md) for the full semantics. The rule for contributors: **never call a
+mutating ESI endpoint directly — always route through `writes.guard`.**
+
+## Conventions summary
+
+- **Reads** take an *optional* `character` (omit → all logged-in characters); **writes** take a
+  *required* `character` plus a `confirm` flag.
+- Authenticated tools build on `run_per_character` / `run_per_corp`; thing-scoped ones use `one_token`.
+- Large lists are capped via `summarize(items, limit)`.
+- Docstrings are the AI-facing tool descriptions — keep them one clear sentence.
+- Tests call tool functions directly and inject mock transports via `set_client` / `set_connection`.

doc/getting-started.md

@@ -0,0 +1,109 @@
+# Getting Started
+
+This guide takes you from a fresh clone to a working AI assistant that can read your EVE Online
+account and act on it when you ask.
+
+## 1. Prerequisites
+
+- **[uv](https://docs.astral.sh/uv/)** — the Python project/dependency manager familiar is built
+  around. Install it first (see the uv docs for your platform).
+- **Python 3.11+** — uv will fetch a suitable interpreter if you don't have one.
+- An **EVE Online account** (for the authenticated tools; public market/universe lookups work
+  without one).
+- An MCP-capable client: **Claude Code**, **Claude Desktop**, **Cowork**, or any other MCP client.
+
+## 2. Install
+
+```bash
+uv sync          # install the project and all dependencies (incl. the dev group)
+```
+
+There is no separate build step for local use. `uv run familiar` runs the server straight from
+source.
+
+## 3. Log in (EVE SSO)
+
+familiar uses EVE's Single Sign-On with PKCE — no client secret, nothing to configure. Run the
+login once **per character**:
+
+```bash
+uv run familiar login    # opens your browser to the EVE SSO consent page
+```
+
+The browser opens CCP's login page. Approve the requested scopes; the callback lands on
+`localhost:8765`, the code is exchanged for tokens, and the (rotating) refresh token is stored
+**encrypted** on disk. You'll see a "login complete" page — close the tab and return to the terminal.
+
+To add more characters, switch EVE accounts in the browser and run `login` again. All characters live
+in one encrypted store:
+
+```bash
+uv run familiar status   # list logged-in characters
+uv run familiar logout NAME   # forget one character (by name or id)
+uv run familiar logout        # forget ALL characters
+```
+
+> **Changing scopes:** if the set of requested scopes changes (e.g. a new familiar version adds
+> tools), you must log in again to grant them.
+
+See [Authentication](authentication.md) for the full flow and the scope list.
+
+## 4. Connect to your AI client
+
+### Claude Code
+
+```bash
+claude mcp add familiar -- uv --directory /path/to/familiar run familiar
+```
+
+### Claude Desktop
+
+Add this to `claude_desktop_config.json` (replace the directory with your clone's absolute path):
+
+```json
+{
+  "mcpServers": {
+    "familiar": {
+      "command": "uv",
+      "args": ["--directory", "/path/to/familiar", "run", "familiar"]
+    }
+  }
+}
+```
+
+The client launches `uv run familiar` with no arguments, which starts the MCP server over **stdio** —
+the transport MCP clients speak.
+
+## 5. First queries
+
+Ask in natural language. The AI picks the right tools:
+
+- *"Is the familiar server up?"* → calls `ping`.
+- *"What's Tritanium worth in Jita?"* → `get_market_price`.
+- *"Compare Plex prices across all trade hubs."* → `compare_hubs`.
+- *"How much ISK do I have across all my characters?"* → wallet tools, per character.
+- *"Where are my characters right now?"* → location tools.
+- *"Set a waypoint to Amarr for my main."* → `set_waypoint` (a **write** — previewed first; see below).
+
+## 6. Write actions
+
+Write tools (send mail, set waypoints, edit contacts/fittings, fleet commands, open windows, PI) are
+**enabled by default**, but every write is still previewed unless you pass `confirm=true`, and each
+executed write is logged. To block all writes globally:
+
+```bash
+uv run familiar writes disable    # turn writes off
+uv run familiar writes enable     # turn them back on
+uv run familiar writes            # show current state
+```
+
+Full details in [Write Safety](write-safety.md).
+
+## 7. Verify everything works
+
+```bash
+uv run pytest          # full test suite (fails under 80% coverage)
+uv run familiar        # start the server manually (Ctrl-C to stop) — it waits on stdio
+```
+
+If `uv run familiar` starts without error and `familiar status` shows your characters, you're ready.

src/familiar/__main__.py

@@ -11,13 +11,27 @@ hand in a terminal:
     familiar writes        # show whether write actions are enabled
     familiar writes enable # allow write actions (mail, waypoints, ...)
     familiar writes disable
+    familiar sde download [url]  # download + install the local SDE (offline lookups)
+    familiar sde status          # is the local SDE available, and where
+    familiar sde path            # print the expected SDE path
 """
 
 from __future__ import annotations
 
+import asyncio
 import sys
 
-from . import auth, writes
+from . import auth, sde, writes
+
+
+def _human_size(num: int) -> str:
+    """Render a byte count as a short human-readable string."""
+    size = float(num)
+    for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
+        if size < 1024.0 or unit == "TiB":
+            return f"{size:.1f} {unit}" if unit != "B" else f"{int(size)} {unit}"
+        size /= 1024.0
+    return f"{num} B"
 
 
 def _login() -> int:
@@ -64,9 +78,34 @@ def _writes() -> int:
     return 0
 
 
+def _sde() -> int:
+    sub = sys.argv[2] if len(sys.argv) > 2 else "status"
+    if sub == "download":
+        url = sys.argv[3] if len(sys.argv) > 3 else None
+        result = asyncio.run(sde.download_sde(url))
+        print(f"Wrote {result['path']} ({_human_size(result['bytes'])}).")
+        print("Done.")
+    elif sub == "status":
+        status = sde.sde_status()
+        state = "available" if status["available"] else "not available"
+        print(f"SDE is {state} at {status['path']}.")
+    elif sub == "path":
+        print(sde.sde_path())
+    else:
+        print("Usage: familiar sde [download [url]|status|path]")
+        return 1
+    return 0
+
+
 def main() -> None:
     command = sys.argv[1] if len(sys.argv) > 1 else None
-    handlers = {"login": _login, "status": _status, "logout": _logout, "writes": _writes}
+    handlers = {
+        "login": _login,
+        "status": _status,
+        "logout": _logout,
+        "writes": _writes,
+        "sde": _sde,
+    }
     if command in handlers:
         raise SystemExit(handlers[command]())
     if command in {"-h", "--help", "help"}:

src/familiar/sde.py

@@ -18,21 +18,45 @@ Minimal expected schema (only what familiar reads)::
         groupID   INTEGER,
         volume    REAL
     );
+    CREATE TABLE mapSolarSystems (
+        solarSystemID    INTEGER PRIMARY KEY,
+        solarSystemName  TEXT,
+        regionID         INTEGER,
+        constellationID  INTEGER,
+        security         REAL
+    );
+    CREATE TABLE staStations (
+        stationID       INTEGER PRIMARY KEY,
+        stationName     TEXT,
+        solarSystemID   INTEGER,
+        corporationID   INTEGER
+    );
 
 Anything beyond these columns is ignored, so a full official SDE conversion works
-as well as a hand-built subset.
+as well as a hand-built subset. The Fuzzwork SQLite conversion of the SDE
+(https://www.fuzzwork.co.uk/dump/sqlite-latest.sqlite.bz2) already ships all of
+these tables, so :func:`download_sde` of that dump satisfies the schema above.
 """
 
 from __future__ import annotations
 
+import bz2
 import os
 import sqlite3
+from pathlib import Path
 from typing import Any
 
+import httpx
 from fastmcp import FastMCP
 
 from .config import data_dir
 
+# Fuzzwork's SQLite conversion of the SDE ships invTypes / mapSolarSystems /
+# staStations with the columns familiar reads. Env-overridable for mirrors.
+SDE_URL = os.environ.get(
+    "FAMILIAR_SDE_URL", "https://www.fuzzwork.co.uk/dump/sqlite-latest.sqlite.bz2"
+)
+
 
 def sde_path() -> str:
     """Filesystem path to the SQLite SDE, env-overridable via ``FAMILIAR_SDE_PATH``."""
@@ -97,6 +121,48 @@ def available() -> bool:
     return True
 
 
+async def download_sde(
+    url: str | None = None,
+    dest: Path | None = None,
+    *,
+    transport: httpx.AsyncBaseTransport | None = None,
+) -> dict[str, Any]:
+    """Download and install the SQLite SDE, returning ``{"path", "bytes"}``.
+
+    Streams ``url`` (default :data:`SDE_URL`), bz2-decompresses it incrementally
+    so arbitrarily large dumps work, and writes the plain SQLite database to
+    ``dest`` (default :func:`sde_path`). The Fuzzwork dump (a bz2-compressed
+    SQLite file) satisfies the schema familiar reads.
+
+    ``transport`` mirrors :class:`familiar.esi.EsiClient`: tests inject an
+    ``httpx.MockTransport`` to serve a fixture without touching the network.
+
+    The cached connection is reset on success so :func:`available` and the
+    lookups immediately see the freshly written file.
+    """
+    source = url or SDE_URL
+    target = Path(dest) if dest is not None else Path(sde_path())
+    target.parent.mkdir(parents=True, exist_ok=True)
+
+    decompressor = bz2.BZ2Decompressor()
+    written = 0
+    async with httpx.AsyncClient(transport=transport, timeout=None) as http:
+        async with http.stream("GET", source) as resp:
+            resp.raise_for_status()
+            with open(target, "wb") as out:
+                async for chunk in resp.aiter_bytes():
+                    if not chunk:
+                        continue
+                    plain = decompressor.decompress(chunk)
+                    if plain:
+                        out.write(plain)
+                        written += len(plain)
+
+    # Point the cached connection at the new file.
+    set_connection(None)
+    return {"path": str(target), "bytes": written}
+
+
 def _row_to_type(row: sqlite3.Row | None) -> dict[str, Any] | None:
     """Normalize an ``invTypes`` row to the familiar type dict."""
     if row is None:
@@ -146,6 +212,73 @@ def resolve_type_name(name: str) -> int | None:
     return found["type_id"] if found is not None else None
 
 
+def _row_to_system(row: sqlite3.Row | None) -> dict[str, Any] | None:
+    """Normalize a ``mapSolarSystems`` row to the familiar system dict."""
+    if row is None:
+        return None
+    return {
+        "system_id": row["solarSystemID"],
+        "name": row["solarSystemName"],
+        "region_id": row["regionID"],
+        "constellation_id": row["constellationID"],
+        "security_status": row["security"],
+    }
+
+
+def system_by_id(system_id: int) -> dict[str, Any] | None:
+    """Look up a solar system by id offline, or ``None`` if absent/unavailable."""
+    conn = connect()
+    if conn is None:
+        return None
+    try:
+        cur = conn.execute(
+            "SELECT solarSystemID, solarSystemName, regionID, constellationID, security "
+            "FROM mapSolarSystems WHERE solarSystemID = ?",
+            (system_id,),
+        )
+        return _row_to_system(cur.fetchone())
+    except sqlite3.Error:
+        return None
+
+
+def system_by_name(name: str) -> dict[str, Any] | None:
+    """Look up a solar system by exact (case-insensitive) name offline, or ``None``."""
+    conn = connect()
+    if conn is None:
+        return None
+    try:
+        cur = conn.execute(
+            "SELECT solarSystemID, solarSystemName, regionID, constellationID, security "
+            "FROM mapSolarSystems WHERE solarSystemName = ? COLLATE NOCASE",
+            (name,),
+        )
+        return _row_to_system(cur.fetchone())
+    except sqlite3.Error:
+        return None
+
+
+def station_by_id(station_id: int) -> dict[str, Any] | None:
+    """Look up a station by id offline, or ``None`` if absent/unavailable."""
+    conn = connect()
+    if conn is None:
+        return None
+    try:
+        cur = conn.execute(
+            "SELECT stationID, stationName, solarSystemID FROM staStations WHERE stationID = ?",
+            (station_id,),
+        )
+        row = cur.fetchone()
+    except sqlite3.Error:
+        return None
+    if row is None:
+        return None
+    return {
+        "station_id": row["stationID"],
+        "name": row["stationName"],
+        "system_id": row["solarSystemID"],
+    }
+
+
 def sde_status() -> dict[str, Any]:
     """Report whether the local SDE is available and where it is expected.
 

src/familiar/universe.py

@@ -12,7 +12,7 @@ from fastmcp import FastMCP
 
 from . import sde
 from .esi import client
-from .lookup import resolve_id
+from .lookup import _as_id, resolve_id
 
 ROUTE_FLAGS = {"shortest", "secure", "insecure"}
 
@@ -56,6 +56,26 @@ async def get_item(item: str | int) -> dict[str, Any]:
 
 async def get_system(system: str | int) -> dict[str, Any]:
     """Look up a solar system by name or id: security status, region, counts."""
+    # Serve from the offline SDE when present — no network call. The SDE lacks
+    # security_class and planet/stargate counts, so those come back as
+    # None / 0; ESI is the source of record when the SDE is absent or doesn't
+    # know the system.
+    if sde.available():
+        direct = _as_id(system)
+        offline = (
+            sde.system_by_id(direct) if direct is not None else sde.system_by_name(str(system))
+        )
+        if offline is not None:
+            sec = offline["security_status"]
+            return {
+                "system_id": offline["system_id"],
+                "name": offline["name"],
+                "constellation_id": offline["constellation_id"],
+                "security_status": round(sec, 2) if isinstance(sec, (int, float)) else sec,
+                "security_class": None,
+                "planets": 0,
+                "stargates": 0,
+            }
     system_id = await resolve_id(system, "systems")
     data = await client().get(f"/universe/systems/{system_id}/")
     sec = data.get("security_status")
@@ -72,6 +92,21 @@ async def get_system(system: str | int) -> dict[str, Any]:
 
 async def get_station(station: str | int) -> dict[str, Any]:
     """Look up a station by name or id: which system, type, and services."""
+    # Serve from the offline SDE when the caller passed a numeric id and the
+    # store knows it — no network call. The SDE lacks type_id and services, so
+    # those come back as None; ESI is the source of record otherwise.
+    if sde.available():
+        direct = _as_id(station)
+        if direct is not None:
+            offline = sde.station_by_id(direct)
+            if offline is not None:
+                return {
+                    "station_id": offline["station_id"],
+                    "name": offline["name"],
+                    "system_id": offline["system_id"],
+                    "type_id": None,
+                    "services": None,
+                }
     station_id = await resolve_id(station, "stations")
     data = await client().get(f"/universe/stations/{station_id}/")
     return {

tests/test_sde.py

@@ -1,7 +1,9 @@
 """Local SDE store: offline lookups and SDE-preferred / ESI-fallback wiring."""
 
 import asyncio
+import bz2
 import sqlite3
+from pathlib import Path
 
 import httpx
 import pytest
@@ -21,6 +23,58 @@ def _make_sde(rows=((34, "Tritanium", 18, 0.01), (35, "Pyerite", 18, 0.01))):
     return conn
 
 
+def _build_sde_file(path: Path) -> None:
+    """Write a tiny on-disk SDE with invTypes, mapSolarSystems and staStations."""
+    conn = sqlite3.connect(str(path))
+    conn.executescript(
+        "CREATE TABLE invTypes ("
+        "typeID INTEGER PRIMARY KEY, typeName TEXT, groupID INTEGER, volume REAL);"
+        "CREATE TABLE mapSolarSystems ("
+        "solarSystemID INTEGER PRIMARY KEY, solarSystemName TEXT, regionID INTEGER, "
+        "constellationID INTEGER, security REAL);"
+        "CREATE TABLE staStations ("
+        "stationID INTEGER PRIMARY KEY, stationName TEXT, solarSystemID INTEGER, "
+        "corporationID INTEGER);"
+    )
+    conn.execute("INSERT INTO invTypes VALUES (?, ?, ?, ?)", (34, "Tritanium", 18, 0.01))
+    conn.execute(
+        "INSERT INTO mapSolarSystems VALUES (?, ?, ?, ?, ?)",
+        (30000142, "Jita", 10000002, 20000020, 0.945913),
+    )
+    conn.execute(
+        "INSERT INTO staStations VALUES (?, ?, ?, ?)",
+        (60003760, "Jita IV - Moon 4 - Caldari Navy Assembly Plant", 30000142, 1000035),
+    )
+    conn.commit()
+    conn.close()
+
+
+def _make_sde_full():
+    """In-memory SQLite with invTypes, mapSolarSystems and staStations populated."""
+    conn = sqlite3.connect(":memory:")
+    conn.executescript(
+        "CREATE TABLE invTypes ("
+        "typeID INTEGER PRIMARY KEY, typeName TEXT, groupID INTEGER, volume REAL);"
+        "CREATE TABLE mapSolarSystems ("
+        "solarSystemID INTEGER PRIMARY KEY, solarSystemName TEXT, regionID INTEGER, "
+        "constellationID INTEGER, security REAL);"
+        "CREATE TABLE staStations ("
+        "stationID INTEGER PRIMARY KEY, stationName TEXT, solarSystemID INTEGER, "
+        "corporationID INTEGER);"
+    )
+    conn.execute("INSERT INTO invTypes VALUES (?, ?, ?, ?)", (34, "Tritanium", 18, 0.01))
+    conn.execute(
+        "INSERT INTO mapSolarSystems VALUES (?, ?, ?, ?, ?)",
+        (30000142, "Jita", 10000002, 20000020, 0.945913),
+    )
+    conn.execute(
+        "INSERT INTO staStations VALUES (?, ?, ?, ?)",
+        (60003760, "Jita IV - Moon 4 - CNAP", 30000142, 1000035),
+    )
+    conn.commit()
+    return conn
+
+
 @pytest.fixture
 def use_sde():
     """Inject an in-memory SDE and reset the cached connection afterwards."""
@@ -31,6 +85,16 @@ def use_sde():
     conn.close()
 
 
+@pytest.fixture
+def use_sde_full():
+    """Inject an in-memory SDE with systems + stations; reset afterwards."""
+    conn = _make_sde_full()
+    sde.set_connection(conn)
+    yield conn
+    sde.set_connection(None)
+    conn.close()
+
+
 def test_available_true_with_table(use_sde):
     assert sde.available() is True
 
@@ -123,3 +187,179 @@ def test_get_item_falls_back_to_esi_when_sde_absent(mock_esi):
     result = asyncio.run(universe.get_item("Tritanium"))
     assert result["type_id"] == 34
     assert result["description"] == "The most common ore type."
+
+
+# --- system / station offline lookups -------------------------------------
+
+
+def test_system_by_id(use_sde_full):
+    sysrow = sde.system_by_id(30000142)
+    assert sysrow == {
+        "system_id": 30000142,
+        "name": "Jita",
+        "region_id": 10000002,
+        "constellation_id": 20000020,
+        "security_status": 0.945913,
+    }
+
+
+def test_system_by_name_case_insensitive(use_sde_full):
+    assert sde.system_by_name("jita")["system_id"] == 30000142
+    assert sde.system_by_name("JITA")["system_id"] == 30000142
+
+
+def test_system_lookups_missing(use_sde_full):
+    assert sde.system_by_id(999999) is None
+    assert sde.system_by_name("Nowhere") is None
+
+
+def test_station_by_id(use_sde_full):
+    station = sde.station_by_id(60003760)
+    assert station == {
+        "station_id": 60003760,
+        "name": "Jita IV - Moon 4 - CNAP",
+        "system_id": 30000142,
+    }
+
+
+def test_station_by_id_missing(use_sde_full):
+    assert sde.station_by_id(999999) is None
+
+
+def test_system_station_lookups_degrade_without_table(use_sde):
+    # `use_sde` has only invTypes — the system/station tables are absent.
+    assert sde.system_by_id(30000142) is None
+    assert sde.system_by_name("Jita") is None
+    assert sde.station_by_id(60003760) is None
+
+
+def test_system_station_lookups_degrade_without_connection():
+    sde.set_connection(None)
+    assert sde.system_by_id(30000142) is None
+    assert sde.system_by_name("Jita") is None
+    assert sde.station_by_id(60003760) is None
+
+
+# --- universe SDE fast-path / ESI fallback --------------------------------
+
+
+def test_get_system_uses_sde_without_esi(use_sde_full, mock_esi):
+    def boom(request):  # any ESI call here is a bug
+        raise AssertionError(f"ESI was called: {request.url}")
+
+    mock_esi(boom)
+    result = asyncio.run(universe.get_system("Jita"))
+    assert result["system_id"] == 30000142
+    assert result["name"] == "Jita"
+    assert result["constellation_id"] == 20000020
+    assert result["security_status"] == 0.95  # rounded to 2 dp
+    assert result["security_class"] is None
+    assert result["planets"] == 0
+    assert result["stargates"] == 0
+
+
+def test_get_system_falls_back_to_esi_when_sde_absent(mock_esi):
+    sde.set_connection(None)
+    fixtures = {
+        "/universe/ids/": {"systems": [{"id": 30000142, "name": "Jita"}]},
+        "/universe/systems/": {
+            "system_id": 30000142,
+            "name": "Jita",
+            "constellation_id": 20000020,
+            "security_status": 0.9459,
+            "security_class": "B",
+            "planets": [{"planet_id": 1}],
+            "stargates": [1, 2, 3],
+        },
+    }
+
+    def route(request):
+        for needle, body in fixtures.items():
+            if needle in request.url.path:
+                return httpx.Response(200, json=body)
+        return httpx.Response(404, json={"error": "no fixture"})
+
+    mock_esi(route)
+    result = asyncio.run(universe.get_system("Jita"))
+    assert result["system_id"] == 30000142
+    assert result["security_status"] == 0.95
+    assert result["security_class"] == "B"
+    assert result["planets"] == 1
+    assert result["stargates"] == 3
+
+
+def test_get_station_uses_sde_for_numeric_id_without_esi(use_sde_full, mock_esi):
+    def boom(request):
+        raise AssertionError(f"ESI was called: {request.url}")
+
+    mock_esi(boom)
+    result = asyncio.run(universe.get_station(60003760))
+    assert result["station_id"] == 60003760
+    assert result["system_id"] == 30000142
+    assert result["type_id"] is None
+    assert result["services"] is None
+
+
+def test_get_station_falls_back_to_esi_when_sde_absent(mock_esi):
+    sde.set_connection(None)
+    fixtures = {
+        "/universe/stations/": {
+            "station_id": 60003760,
+            "name": "Jita IV - Moon 4 - CNAP",
+            "system_id": 30000142,
+            "type_id": 1529,
+            "services": ["market", "repair"],
+        },
+    }
+
+    def route(request):
+        for needle, body in fixtures.items():
+            if needle in request.url.path:
+                return httpx.Response(200, json=body)
+        return httpx.Response(404, json={"error": "no fixture"})
+
+    mock_esi(route)
+    result = asyncio.run(universe.get_station(60003760))
+    assert result["type_id"] == 1529
+    assert result["services"] == ["market", "repair"]
+
+
+# --- download_sde ----------------------------------------------------------
+
+
+def test_download_sde_decompresses_and_installs(tmp_path):
+    src = tmp_path / "source.sqlite"
+    _build_sde_file(src)
+    compressed = bz2.compress(src.read_bytes())
+
+    def handler(request):
+        # Serve the bz2 bytes in a couple of chunks to exercise streaming.
+        return httpx.Response(200, content=compressed)
+
+    dest = tmp_path / "installed.sqlite"
+    try:
+        result = asyncio.run(
+            sde.download_sde(
+                url="https://example.invalid/sde.sqlite.bz2",
+                dest=dest,
+                transport=httpx.MockTransport(handler),
+            )
+        )
+        assert result["path"] == str(dest)
+        assert dest.is_file()
+        assert result["bytes"] == dest.stat().st_size
+
+        # Point the live store at the freshly written file and query it.
+        import os
+
+        os.environ["FAMILIAR_SDE_PATH"] = str(dest)
+        sde.set_connection(None)
+        try:
+            assert sde.available() is True
+            assert sde.type_by_id(34)["name"] == "Tritanium"
+            assert sde.system_by_name("Jita")["system_id"] == 30000142
+        finally:
+            del os.environ["FAMILIAR_SDE_PATH"]
+            sde.set_connection(None)
+    finally:
+        sde.set_connection(None)