diff --git a/.gitignore b/.gitignore index 173d0d3..c8fedf8 100644 --- a/.gitignore +++ b/.gitignore @@ -22,6 +22,7 @@ webclient/navi.db debug/eval/navi.db session_files/ +export_files/ workspace/ .tmp .codex diff --git a/deploy/env.template b/deploy/env.template index a95fb16..aa79ca9 100644 --- a/deploy/env.template +++ b/deploy/env.template @@ -89,3 +89,18 @@ NAVI_MCP_HARD_PANEL_TOKEN= NAVI_MCP_SYNAPSE_TOKEN= NAVI_MCP_TGCLIENT_TOKEN= + +# ─── History export ────────────────────────────────────────────────── +# Bulk snapshot of message text for an external consumer (n-gram tables). +# The snapshot is age-encrypted to the recipient below; its private half stays +# on the consumer, so Navi cannot decrypt what it exports. Both lines are +# required — with either blank nothing is built and the endpoints answer 503. +# Consumer setup: scripts/navi_export.py keygen +#NAVI_EXPORT_AGE_RECIPIENT=age1... +#NAVI_EXPORT_TOKEN= +# Roles in the corpus. Tool output and system prompts are excluded on purpose: +# they are scraped web text and boilerplate, and they drown real dialogue. +#NAVI_EXPORT_ROLES=user,assistant +#NAVI_EXPORT_INTERVAL_HOURS=4 +# Rolling window over whole sessions, newest first. 0 = no cap. +#NAVI_EXPORT_MAX_BYTES=157286400 diff --git a/docs/api.md b/docs/api.md index dd408f0..d64e2f2 100644 --- a/docs/api.md +++ b/docs/api.md @@ -15,6 +15,10 @@ taking it down. Reads in-process state only — no LLM call — so it is cheap enough for a 30 s poll (the embed probe is cached for 10 s). +When the history export is configured, the body also carries an informational +`export` object (`built`, `built_at`, `rows`, `age_seconds`, `stale`) — see +[`export.md`](export.md). It never affects `status`. + **Response `200`** ```json { @@ -936,6 +940,47 @@ --- +### History export + +Two endpoints, both requiring `Authorization: Bearer ` +(machine credential, not user auth; `401` for missing and wrong alike, `503` +when the export is not configured or nothing has been built). Full design, +corpus rules and threat model: [`export.md`](export.md). + +#### `GET /export/v1/manifest` + +Describes the current snapshot without transferring it. + +**Response `200`** +```json +{ + "export_version": 1, + "built_at": "2026-11-04T12:00:00+00:00", + "content_id": "9f2c…", + "file": {"name": "messages-v1.jsonl.gz.age", "bytes": 5242880, "sha256": "…"}, + "rows": 23672, + "sessions": 1177, + "sessions_total": 1177, + "text_bytes": 14680064, + "fields": ["v", "session", "seq", "role", "content", "ts"], + "roles": ["user", "assistant"], + "window": {"from": "…", "to": "…", "truncated": false, "max_bytes": 157286400}, + "age_recipient": "age1…" +} +``` + +--- + +#### `GET /export/v1/messages` + +The age-encrypted gzip-JSONL snapshot, streamed from disk. + +**Headers**: `ETag` (from `content_id`), `X-Export-Content-Id`, `X-Export-Built-At` + +**Response `304`** — empty body, when `If-None-Match` carries the current `ETag` + +--- + ## WebSocket ### `WS /ws/sessions/{session_id}` diff --git a/docs/config.md b/docs/config.md index 1ff65d5..223d9e8 100644 --- a/docs/config.md +++ b/docs/config.md @@ -245,6 +245,22 @@ |---|---|---|---| | `PUBLIC_URL` | str | `http://localhost:8099` | Base URL used by `share_file` to build download links. Set this when behind a reverse proxy. | +## History export + +Bulk snapshot of message text for an external consumer (n-gram tables). Off +unless **both** the recipient key and the token are set — with either blank the +builder never starts and the endpoints answer `503`. See +[`export.md`](export.md). + +| Variable | Type | Default | Description | +|---|---|---|---| +| `NAVI_EXPORT_AGE_RECIPIENT` | str | `""` | age public key (`age1…`) the snapshot is encrypted to. The private half lives on the consumer only. | +| `NAVI_EXPORT_TOKEN` | str | `""` | Bearer token the consumer presents. Compared as a SHA-256 digest; rotate by changing it and restarting. | +| `NAVI_EXPORT_INTERVAL_HOURS` | float | `4.0` | Rebuild period. The first build runs at startup; the floor is 15 minutes. | +| `NAVI_EXPORT_ROLES` | str | `user,assistant` | Comma-separated roles in the corpus. `tool` output and `system` prompts are excluded by default — they skew an n-gram table toward scraped web text and boilerplate. | +| `NAVI_EXPORT_MAX_BYTES` | int | `157286400` (150 MB) | Rolling window: whole sessions, newest first, until this much raw text. `0` = no cap. | +| `NAVI_EXPORT_DIR` | str | `export_files` | Where the ciphertext and `manifest.json` live. Git-ignored. | + ## Context compression | Variable | Type | Default | Description | diff --git a/docs/export.md b/docs/export.md new file mode 100644 index 0000000..cc72538 --- /dev/null +++ b/docs/export.md @@ -0,0 +1,163 @@ +# History export + +A periodically rebuilt, age-encrypted snapshot of message text, served over +HTTP to one machine on the local network that holds the decryption key. Built +for n-gram tables: the consumer needs a corpus of real dialogue, not a copy of +the database. + +## Why not a dump + +The database is ~600 MB on prod, but almost none of it is dialogue: + +| Part | Size | +|---|---| +| `session_messages.content`, all roles | 68 MB | +| `session_messages.thinking` | 40 MB | +| `session_messages.tool_calls` | 16 MB | +| attachments (images/files/metadata) | 95 MB | +| everything else (`sessions` JSON, store, vectors) | ~370 MB | + +The default corpus — `user` + `assistant` content across both message tables — +is **14 MB of UTF-8, 23,672 rows** on prod. That gzips to a few megabytes. + +A dump would also carry `mcp_user_keys` (Fernet-encrypted MCP credentials), +`users` (PII, password hashes) and `api_tokens` (token hashes). Those must not +leave the server in any form, under any key — so the export is a fixed, +versioned `SELECT`, not `pg_dump`. + +`EXPORT_VERSION` (`navi/export/builder.py`) plus the `fields` list in the +manifest is the contract. Adding a field means bumping the version, because the +consumer validates the list. + +## What is in the file + +gzip-JSONL, one object per line: + +```json +{"v":1,"session":"…","seq":42,"role":"assistant","content":"…","ts":"2026-11-04T10:22:31+00:00"} +``` + +Default corpus: `role = user | assistant`, `content` only. `NAVI_EXPORT_ROLES` +adds or removes roles: + +| Role | Default | Why | +|---|---|---| +| `user`, `assistant` | on | the dialogue itself — what an n-gram table should predict | +| `tool` | off | 52 MB of scraped pages, command output and file contents; a different distribution that would dominate the counts | +| `system` | off | near-identical boilerplate (867 copies of the same prompts) | +| `thinking` | never | model reasoning is a separate style; not part of the schema | + +Summary and compression artefacts (`is_summary`, `is_compression`) are always +excluded — they duplicate dialogue in a summarised form. + +## The rolling window + +Whole sessions, newest activity first, until the cumulative text reaches +`NAVI_EXPORT_MAX_BYTES` (150 MB). Today the entire corpus is 14 MB, so nothing +is cut (`window.truncated: false`); the cap exists so the snapshot stays +bounded as history grows. The cut is always on a session boundary — a +conversation never arrives half a story. + +Both `session_messages` and `session_messages_archive` feed the window, so +compressed-away history stays in the corpus. + +## Build + +A background loop (`navi/export/loop.py`) starts at boot and repeats every +`NAVI_EXPORT_INTERVAL_HOURS`. Inside: + +1. one read-only `REPEATABLE READ` transaction, so window and rows agree; +2. rows stream through an asyncpg cursor; +3. gzip is written deterministically (fixed `mtime`, sorted keys) — the same + corpus always yields the same digest; +4. the plaintext is encrypted with `age` to `NAVI_EXPORT_AGE_RECIPIENT` and the + plaintext is deleted; only `messages-v1.jsonl.gz.age` remains. + +`content_id` is the SHA-256 of the plaintext JSONL. When a rebuild produces the +same `content_id`, the ciphertext is left untouched and only `built_at` moves — +so an idle server does not invalidate a consumer's cache. + +If a build fails, the previous file and manifest stay in place. `built_at` keeps +telling the truth, so a consumer sees the snapshot ageing instead of silently +reading stale data as fresh. + +`/health` carries an informational `export` object when this feature is +configured — `built`, `built_at`, `rows`, `age_seconds`, `stale` (> two missed +rebuilds). It never moves the aggregate `status`: a slow snapshot is not the +service being unhealthy. + +## Endpoints + +Both require `Authorization: Bearer `. This is a machine +credential: no cookies, no user, no role. A missing token and a wrong token get +the same `401`. + +### `GET /export/v1/manifest` + +```json +{ + "export_version": 1, + "built_at": "2026-11-04T12:00:00+00:00", + "content_id": "9f2c…", + "file": {"name": "messages-v1.jsonl.gz.age", "bytes": 5242880, "sha256": "…"}, + "rows": 23672, + "sessions": 1177, + "sessions_total": 1177, + "text_bytes": 14680064, + "fields": ["v", "session", "seq", "role", "content", "ts"], + "roles": ["user", "assistant"], + "window": {"from": "…", "to": "…", "truncated": false, "max_bytes": 157286400}, + "age_recipient": "age1…" +} +``` + +`503` when nothing has been built yet, or when the export is not configured. + +### `GET /export/v1/messages` + +The ciphertext, streamed from disk. Headers: `ETag` (derived from +`content_id`), `X-Export-Content-Id`, `X-Export-Built-At`. A conditional +request with the matching `If-None-Match` answers `304` with an empty body — +the normal case on a 4-hour schedule. + +## Consumer + +[`scripts/navi_export.py`](../scripts/navi_export.py) — one file, one +dependency (`pip install pyrage`), Python 3.11+. + +```bash +# once: make a keypair here, put the public half into Navi's .env +./navi_export.py keygen --out ~/.config/navi-export +# -> ~/.config/navi-export/key.txt (secret, 0600; never leaves this machine) +# -> age1… (NAVI_EXPORT_AGE_RECIPIENT on Navi) + +# every run +./navi_export.py pull \ + --url https://navi.example \ + --token-file ~/.config/navi-export/token \ + --identity ~/.config/navi-export/key.txt \ + --out ~/ngram/messages.jsonl.gz +``` + +The pull reads the manifest first and exits `3` with **no download** when +`content_id` matches the local state file. Otherwise it downloads to `.part`, +verifies the manifest's SHA-256, decrypts, fully decompresses once to prove the +gzip is intact, and atomically renames into place. + +Exit codes: `0` updated, `3` unchanged, `1` error. + +Cron: polling every 30 minutes is cheap — the steady state is one manifest GET. + +## Threat model + +- **Navi cannot read its own export.** Only the public half of the age key is on + the server. A full compromise of Navi yields the ciphertext, not the corpus. +- **The token is transport control, not confidentiality.** It gates who may + fetch. Confidentiality is the age key. Rotate the token by editing `.env` and + restarting; rotate the key by generating a new pair and updating + `NAVI_EXPORT_AGE_RECIPIENT` (the old ciphertext becomes unreadable, which is + the point). +- **What a token holder still learns:** metadata in the manifest (row counts, + date range) and the size of the corpus. +- **Nothing about attachment bodies, user accounts, MCP keys or API tokens is + reachable through these endpoints**, by construction of the `SELECT`. diff --git a/docs/index.md b/docs/index.md index ab67924..2299240 100644 --- a/docs/index.md +++ b/docs/index.md @@ -26,6 +26,7 @@ | [`navi_code.md`](navi_code.md) | Navi Code — local terminal mode (profile, .env, security) | | [`navi_code_cli.md`](navi_code_cli.md) | `navi-code` CLI/TUI client — flags, commands, TUI layout, events | | [`api_tokens.md`](api_tokens.md) | API token auth for headless clients — schema, REST endpoints, security | +| [`export.md`](export.md) | History export — encrypted, token-served snapshot of dialogue for n-gram consumers | | [`architecture.md`](architecture.md) | Component diagram, data flow, dependency graph | | [`agent.md`](agent.md) | Agent loop, planning phase, tool execution, subagents, workers | | [`tools.md`](tools.md) | Built-in tools, user tool format, hot-reload | diff --git a/navi/api/routes/export.py b/navi/api/routes/export.py new file mode 100644 index 0000000..ce3a756 --- /dev/null +++ b/navi/api/routes/export.py @@ -0,0 +1,107 @@ +"""Token-gated delivery of the encrypted history snapshot. + +Two endpoints, both requiring `Authorization: Bearer `: + +* `GET /export/v1/manifest` — what is available and whether it changed. Small, + unencrypted (it holds no message text), safe to poll every few minutes. +* `GET /export/v1/messages` — the age file itself, streamed from disk. + +The fast path is the one that transfers nothing: the manifest's `content_id` +is a digest of the plaintext corpus, stable across rebuilds of unchanged data, +and the endpoint echoes it as an ETag so a conditional GET answers 304. + +Auth is intentionally not the user auth stack: this is a machine credential +with no session, no cookies and no role. A wrong token and a missing one get +the same answer, so the endpoint does not confirm that a token exists. +""" + +from __future__ import annotations + +import hashlib +import hmac +from pathlib import Path +from typing import Annotated + +import structlog +from fastapi import APIRouter, Header, HTTPException, Request +from fastapi.responses import FileResponse, JSONResponse, Response + +from navi.config import settings +from navi.export.builder import EXPORT_VERSION, export_paths, read_manifest + +log = structlog.get_logger() + +router = APIRouter(prefix="/export/v1", tags=["export"]) + +_WWW_AUTH = {"WWW-Authenticate": "Bearer"} + + +def _digest(value: str) -> bytes: + return hashlib.sha256(value.encode("utf-8")).digest() + + +def _authorize(authorization: str | None) -> None: + expected = settings.navi_export_token.strip() + if not expected: + raise HTTPException(status_code=503, detail="export is not configured") + given = "" + if authorization and authorization.lower().startswith("bearer "): + given = authorization[7:].strip() + if not given or not hmac.compare_digest(_digest(given), _digest(expected)): + raise HTTPException(status_code=401, detail="invalid token", headers=_WWW_AUTH) + + +def _etag(content_id: str) -> str: + return f'"v{EXPORT_VERSION}-{content_id[:32]}"' + + +def _manifest_or_503() -> tuple[dict, Path]: + manifest = read_manifest() + age_path, _ = export_paths() + if not manifest or not age_path.exists(): + raise HTTPException(status_code=503, detail="no export has been built yet") + return manifest, age_path + + +@router.get("/manifest") +async def export_manifest( + authorization: Annotated[str | None, Header()] = None, +) -> JSONResponse: + """Describe the current snapshot without transferring it.""" + _authorize(authorization) + manifest, _ = _manifest_or_503() + log.info("export.manifest_served", content_id=manifest.get("content_id", "")[:12]) + return JSONResponse(manifest, headers={"Cache-Control": "no-store"}) + + +@router.get("/messages") +async def export_messages( + request: Request, + authorization: Annotated[str | None, Header()] = None, +) -> Response: + """The age-encrypted gzip-JSONL snapshot.""" + _authorize(authorization) + manifest, age_path = _manifest_or_503() + + content_id = manifest.get("content_id", "") + etag = _etag(content_id) + if content_id and request.headers.get("if-none-match") == etag: + return Response(status_code=304, headers={"ETag": etag}) + + log.info( + "export.file_served", + content_id=content_id[:12], + bytes=manifest.get("file", {}).get("bytes"), + client=request.client.host if request.client else None, + ) + return FileResponse( + str(age_path), + media_type="application/octet-stream", + filename=manifest.get("file", {}).get("name", "messages.jsonl.gz.age"), + headers={ + "ETag": etag, + "Cache-Control": "no-store", + "X-Export-Content-Id": content_id, + "X-Export-Built-At": manifest.get("built_at", ""), + }, + ) diff --git a/navi/api/routes/health.py b/navi/api/routes/health.py index e8bc7cd..2c1f3b9 100644 --- a/navi/api/routes/health.py +++ b/navi/api/routes/health.py @@ -7,6 +7,7 @@ """ import time +from datetime import datetime, timezone from fastapi import APIRouter @@ -34,7 +35,7 @@ statuses = [check["status"] for check in (embed_check, hive_check, llm_check)] aggregate = "degraded" if "degraded" in statuses else "ok" - return { + body = { "status": aggregate, "service": "navi", "checks": {"embed": embed_check, "hive": hive_check, "llm": llm_check}, @@ -43,6 +44,39 @@ "embed": embed_status, "hive": hive, } + export_status = _export_status() + if export_status is not None: + # Informational only: a background snapshot is not the service's + # health, so it never moves `status`. + body["export"] = export_status + return body + + +def _export_status() -> dict | None: + """Freshness of the history snapshot, when the export is configured.""" + if not settings.navi_export_configured: + return None + from navi.export.builder import read_manifest + from navi.export.loop import export_interval_seconds + + manifest = read_manifest() + if not manifest: + return {"configured": True, "built": False} + try: + built = datetime.fromisoformat(manifest["built_at"]) + except (KeyError, ValueError): + return {"configured": True, "built": False} + age = (datetime.now(timezone.utc) - built).total_seconds() + return { + "configured": True, + "built": True, + "built_at": manifest["built_at"], + "content_id": manifest.get("content_id", "")[:12], + "rows": manifest.get("rows"), + "age_seconds": int(age), + # Two missed rebuilds, not one: a single slow tick is not news. + "stale": age > export_interval_seconds() * 2, + } def _llm_status() -> dict: diff --git a/navi/config.py b/navi/config.py index e154097..162738a 100644 --- a/navi/config.py +++ b/navi/config.py @@ -103,6 +103,27 @@ # Change if the server is behind a reverse proxy or runs on a different port. public_url: str = "http://localhost:8099" + # --- Export: bulk history snapshot for an external consumer (n-gram tables) + # Every navi_export_interval_hours one gzip-JSONL file of message text is built + # and age-encrypted to navi_export_age_recipient. The private half of that key + # never reaches this server, so Navi cannot decrypt its own export. + # navi_export_token is the bearer token the consumer presents; only its SHA-256 + # digest is compared. Both fields must be set for the builder to run and + # the endpoints to answer at all. + navi_export_age_recipient: str = "" + navi_export_token: str = "" + navi_export_interval_hours: float = 4.0 + # Which roles make the corpus. Tool results and system prompts are opt-in: + # tool output is scraped web text and command logs (52 MB of prod's 74 MB), + # and system messages are near-duplicate boilerplate — both would drown the + # actual dialogue in an n-gram table. + navi_export_roles: str = "user,assistant" + # Rolling window, whole sessions, newest first, until this much raw text. + # A cap that never binds today (the whole corpus is ~13 MB); it exists so + # the snapshot stays bounded as history grows. + navi_export_max_bytes: int = 150 * 1024 * 1024 + navi_export_dir: str = "export_files" + # Gmail IMAP/SMTP (App Password auth) gmail_address: str = "" gmail_app_password: str = "" @@ -265,6 +286,15 @@ return [p.strip() for p in self.fs_allowed_paths.split(",") if p.strip()] @property + def navi_export_roles_list(self) -> list[str]: + return [r.strip() for r in self.navi_export_roles.split(",") if r.strip()] + + @property + def navi_export_configured(self) -> bool: + """Both a recipient key and a token are required — no half-configured export.""" + return bool(self.navi_export_age_recipient.strip() and self.navi_export_token.strip()) + + @property def terminal_allowed_commands_list(self) -> list[str]: return [c.strip() for c in self.terminal_allowed_commands.split(",") if c.strip()] diff --git a/navi/export/__init__.py b/navi/export/__init__.py new file mode 100644 index 0000000..b055a1a --- /dev/null +++ b/navi/export/__init__.py @@ -0,0 +1,11 @@ +"""Bulk history export for an external consumer (n-gram tables and the like). + +The snapshot is *not* a database dump. It is a fixed, versioned SELECT over the +message tables — text only, no attachments, no keys, no user rows — written as +gzip-JSONL and encrypted with `age` to a recipient whose private half never +touches this server. See `docs/export.md`. +""" + +from navi.export.builder import build_export, read_manifest + +__all__ = ["build_export", "read_manifest"] diff --git a/navi/export/builder.py b/navi/export/builder.py new file mode 100644 index 0000000..d91337c --- /dev/null +++ b/navi/export/builder.py @@ -0,0 +1,276 @@ +"""Build the encrypted history snapshot served at /export/v1/messages. + +Shape of a build +---------------- +1. One read-only REPEATABLE READ transaction over both message tables, so the + rolling window and the rows that follow come from the same instant. +2. The window: whole sessions, newest activity first, until the raw text + reaches `navi_export_max_bytes`. Cutting on a session boundary keeps a + conversation from arriving half a story. +3. Rows are streamed through a cursor and serialised one JSON object per line. + The plaintext gzip is built deterministically (fixed mtime, sorted keys) so + the same input always yields the same `content_id` — that is what lets the + consumer answer "nothing changed" without downloading anything. +4. The plaintext is encrypted to `navi_export_age_recipient` and the plaintext is + deleted. Only ciphertext stays on disk. + +Nothing here reads `users`, `mcp_user_keys`, `api_tokens` or attachments. +Adding a field is a schema change: bump EXPORT_VERSION and say so in the +manifest, because the consumer validates `fields`. +""" + +from __future__ import annotations + +import gzip +import hashlib +import json +import os +from datetime import datetime, timezone +from pathlib import Path + +import structlog + +from navi.config import settings + +log = structlog.get_logger() + +EXPORT_VERSION = 1 +FIELDS = ["v", "session", "seq", "role", "content", "ts"] +MANIFEST_NAME = "manifest.json" +FILE_NAME = "messages-v1.jsonl.gz.age" + +# Whole sessions, newest first, while the cumulative text stays under budget. +_WINDOW_SQL = """ +WITH msgs AS ( + SELECT session_id, role, content, created_at, is_summary, is_compression + FROM session_messages + UNION ALL + SELECT session_id, role, content, created_at, is_summary, is_compression + FROM session_messages_archive +), +buckets AS ( + SELECT session_id, + max(created_at) AS last_ts, + sum(octet_length(content))::bigint AS text_bytes + FROM msgs + WHERE role = ANY($1::text[]) + AND content IS NOT NULL AND content <> '' + AND NOT coalesce(is_summary, false) + AND NOT coalesce(is_compression, false) + GROUP BY session_id +), +ordered AS ( + SELECT session_id, last_ts, text_bytes, + sum(text_bytes) OVER (ORDER BY last_ts DESC NULLS LAST, session_id) AS cum + FROM buckets +) +SELECT session_id +FROM ordered +WHERE cum - text_bytes < $2 +ORDER BY last_ts DESC NULLS LAST, session_id +""" + +_SESSIONS_SQL = """ +SELECT count(*) FROM ( + SELECT session_id + FROM ( + SELECT session_id, role, content, created_at, is_summary, is_compression + FROM session_messages + UNION ALL + SELECT session_id, role, content, created_at, is_summary, is_compression + FROM session_messages_archive + ) m + WHERE role = ANY($1::text[]) + AND content IS NOT NULL AND content <> '' + AND NOT coalesce(is_summary, false) + AND NOT coalesce(is_compression, false) + GROUP BY session_id +) t +""" + +_ROWS_SQL = """ +SELECT session_id, sequence_number, role, content, created_at +FROM ( + SELECT session_id, sequence_number, role, content, created_at, is_summary, is_compression + FROM session_messages + UNION + SELECT session_id, sequence_number, role, content, created_at, is_summary, is_compression + FROM session_messages_archive +) m +WHERE session_id = ANY($1::text[]) + AND role = ANY($2::text[]) + AND content IS NOT NULL AND content <> '' + AND NOT coalesce(is_summary, false) + AND NOT coalesce(is_compression, false) +ORDER BY created_at NULLS LAST, session_id, sequence_number +""" + + +def export_paths(out_dir: Path | None = None) -> tuple[Path, Path]: + """(age file, manifest file) in the export directory.""" + base = Path(out_dir if out_dir is not None else settings.navi_export_dir) + return base / FILE_NAME, base / MANIFEST_NAME + + +def read_manifest(out_dir: Path | None = None) -> dict | None: + """The manifest on disk, or None when no export has been built yet.""" + _, manifest_path = export_paths(out_dir) + try: + return json.loads(manifest_path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + + +def _record(row) -> dict: + created = row["created_at"] + ts = created.astimezone(timezone.utc).isoformat() if created else None + return { + "v": EXPORT_VERSION, + "session": row["session_id"], + "seq": row["sequence_number"], + "role": row["role"], + "content": row["content"], + "ts": ts, + } + + +async def build_export(pool, out_dir: Path | None = None) -> dict: + """Build (or confirm) the snapshot and return the manifest. + + Raises when the export is not configured or the recipient key is malformed; + the caller is a background loop that logs and retries on the next tick. + """ + import pyrage + + if not settings.navi_export_configured: + raise RuntimeError("export is not configured (NAVI_EXPORT_AGE_RECIPIENT / NAVI_EXPORT_TOKEN)") + + recipient = pyrage.x25519.Recipient.from_str(settings.navi_export_age_recipient.strip()) + roles = settings.navi_export_roles_list + # 0 (or negative) disables the cap: the window then covers all history. + budget = settings.navi_export_max_bytes if settings.navi_export_max_bytes > 0 else 2**62 + + age_path, manifest_path = export_paths(out_dir) + base = age_path.parent + base.mkdir(parents=True, exist_ok=True) + + plain_tmp = base / (FILE_NAME + ".plain.tmp") + age_tmp = base / (FILE_NAME + ".tmp") + + now = datetime.now(timezone.utc) + rows = 0 + text_bytes = 0 + first_ts: datetime | None = None + last_ts: datetime | None = None + session_ids: list[str] = [] + total_sessions = 0 + + digest = hashlib.sha256() + try: + async with pool.acquire() as conn, conn.transaction( + isolation="repeatable_read", readonly=True + ): + session_ids = [ + r["session_id"] for r in await conn.fetch(_WINDOW_SQL, roles, budget) + ] + total_sessions = await conn.fetchval(_SESSIONS_SQL, roles) or 0 + + # Blocking file I/O in a background job, once every few hours: + # an executor hop here would buy nothing. + with open(plain_tmp, "wb") as plain_fh, gzip.GzipFile( # noqa: ASYNC230 + filename="", + mode="wb", + fileobj=plain_fh, + compresslevel=9, + mtime=0, + ) as gz: + if session_ids: + async for row in conn.cursor(_ROWS_SQL, session_ids, roles, prefetch=500): + line = json.dumps( + _record(row), ensure_ascii=False, sort_keys=True, separators=(",", ":") + ) + "\n" + chunk = line.encode("utf-8") + gz.write(chunk) + digest.update(chunk) + rows += 1 + text_bytes += len(row["content"].encode("utf-8")) + created = row["created_at"] + if created is not None: + if first_ts is None or created < first_ts: + first_ts = created + if last_ts is None or created > last_ts: + last_ts = created + except BaseException: + plain_tmp.unlink(missing_ok=True) + raise + + content_id = digest.hexdigest() + previous = read_manifest(out_dir) + + # Same corpus as last time: keep the ciphertext (and its ETag) untouched. + # Only the manifest's built_at moves, which is what the consumer ignores. + if ( + previous + and previous.get("content_id") == content_id + and age_path.exists() + and previous.get("file", {}).get("name") == FILE_NAME + ): + plain_tmp.unlink(missing_ok=True) + manifest = {**previous, "built_at": now.isoformat()} + _write_manifest(manifest_path, manifest) + log.info("export.unchanged", content_id=content_id[:12], rows=rows) + return manifest + + try: + with open(plain_tmp, "rb") as src, open(age_tmp, "wb") as dst: # noqa: ASYNC230 + pyrage.encrypt_io(src, dst, [recipient]) + file_bytes = age_tmp.stat().st_size + file_sha = _sha256_file(age_tmp) + os.replace(age_tmp, age_path) + finally: + plain_tmp.unlink(missing_ok=True) + age_tmp.unlink(missing_ok=True) + + manifest = { + "export_version": EXPORT_VERSION, + "built_at": now.isoformat(), + "content_id": content_id, + "file": {"name": FILE_NAME, "bytes": file_bytes, "sha256": file_sha}, + "rows": rows, + "sessions": len(session_ids), + "sessions_total": total_sessions, + "text_bytes": text_bytes, + "fields": FIELDS, + "roles": roles, + "window": { + "from": first_ts.astimezone(timezone.utc).isoformat() if first_ts else None, + "to": last_ts.astimezone(timezone.utc).isoformat() if last_ts else None, + "truncated": len(session_ids) < total_sessions, + "max_bytes": budget, + }, + "age_recipient": settings.navi_export_age_recipient.strip(), + } + _write_manifest(manifest_path, manifest) + log.info( + "export.built", + rows=rows, + sessions=len(session_ids), + text_bytes=text_bytes, + file_bytes=file_bytes, + content_id=content_id[:12], + ) + return manifest + + +def _write_manifest(path: Path, manifest: dict) -> None: + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + os.replace(tmp, path) + + +def _sha256_file(path: Path) -> str: + h = hashlib.sha256() + with open(path, "rb") as f: + for block in iter(lambda: f.read(1 << 20), b""): + h.update(block) + return h.hexdigest() diff --git a/navi/export/loop.py b/navi/export/loop.py new file mode 100644 index 0000000..71990f7 --- /dev/null +++ b/navi/export/loop.py @@ -0,0 +1,38 @@ +"""Background rebuild loop for the export snapshot. + +Same shape as the other lifespan loops (cleanup, recall, pending sweep): one +task, never allowed to die, logging and retrying on the next tick. The first +build happens at startup, so a restart is also a refresh. +""" + +import asyncio + +import structlog + +from navi.config import settings +from navi.export.builder import build_export + +log = structlog.get_logger() + +_MIN_INTERVAL_SECONDS = 15 * 60 + + +def export_interval_seconds() -> float: + return max(_MIN_INTERVAL_SECONDS, settings.navi_export_interval_hours * 3600) + + +async def export_loop(store) -> None: + """Rebuild the snapshot every `navi_export_interval_hours`.""" + interval = export_interval_seconds() + while True: + try: + pool = await store._get_pool() + await build_export(pool) + except asyncio.CancelledError: + raise + except Exception: + # A failed build leaves the previous file and manifest in place: + # built_at keeps telling the truth, and the consumer sees the + # snapshot getting older instead of silently stale-new. + log.exception("export.build_failed") + await asyncio.sleep(interval) diff --git a/navi/main.py b/navi/main.py index 246bd7b..a187164 100644 --- a/navi/main.py +++ b/navi/main.py @@ -13,7 +13,7 @@ from fastapi.staticfiles import StaticFiles from navi.api.deps import require_admin -from navi.api.routes import agents, api_tokens, auth, health, messages, mcp_keys, peer, push, sessions, webhooks, synapse +from navi.api.routes import agents, api_tokens, auth, export, health, messages, mcp_keys, peer, push, sessions, webhooks, synapse from navi.api.routes.admin import router as admin_router from navi.api.websocket import router as ws_router from navi.config import settings @@ -197,6 +197,14 @@ ) pending_sweep_task = asyncio.create_task(pending_sweep_loop(container.session_store)) + # History export for external consumers. No key or no token = no task at + # all: an unconfigured export costs nothing and exposes nothing. + export_task: asyncio.Task | None = None + if settings.navi_export_configured: + from navi.export.loop import export_loop + + export_task = asyncio.create_task(export_loop(container.session_store)) + # Swarm announce loop — tell the hive registry (address book) this navi # exists. Never a blocking dependency: no key/no HIVE_URL = no task at all, # and the loop itself survives any number of failed ticks. @@ -241,6 +249,12 @@ await pending_sweep_task except asyncio.CancelledError: pass + if export_task is not None: + export_task.cancel() + try: + await export_task + except asyncio.CancelledError: + pass if ui_mcp_active and ui_server_task is not None: ui_server_task.cancel() @@ -286,6 +300,7 @@ app.include_router(agents.router) app.include_router(sessions.router) app.include_router(messages.router) +app.include_router(export.router) app.include_router(ws_router) app.include_router(push.router) app.include_router(webhooks.router) diff --git a/pyproject.toml b/pyproject.toml index ee7fe64..4a0ce4f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,6 +44,9 @@ # distribution name is gnexus-gauth (import: gnexus_gauth), repo is gnexus-auth-client-py "gnexus-gauth @ git+https://git.gnexus.space/root/gnexus-auth-client-py.git", "cryptography>=42", + # History export: age (X25519 + chunked ChaCha20-Poly1305) in streaming + # mode, so Navi encrypts a snapshot it cannot itself decrypt. + "pyrage>=1.2", # Synapse notification hub — client lib (ingest + s2s webhook verification). # Only the receiving half is used today: no outbound events per user decision. diff --git a/scripts/navi_export.py b/scripts/navi_export.py new file mode 100755 index 0000000..f914d40 --- /dev/null +++ b/scripts/navi_export.py @@ -0,0 +1,249 @@ +#!/usr/bin/env python3 +"""Pull Navi's encrypted history snapshot, decrypt it, leave a gzip-JSONL file. + +Runs on the consumer side — a machine that holds the private half of the age +key and never shares it. Dependencies: Python 3.11+ and `pip install pyrage`. + + # once, on the consumer: make a keypair, send the *public* half to Navi + ./navi_export.py keygen --out ~/.config/navi-export + # -> writes key.txt (secret, mode 600) and prints age1... for NAVI_EXPORT_AGE_RECIPIENT + + # every run (cron-friendly) + ./navi_export.py pull \ + --url https://navi.example \ + --token-file ~/.config/navi-export/token \ + --identity ~/.config/navi-export/key.txt \ + --out ~/ngram/messages.jsonl.gz + +Exit codes: 0 updated, 3 unchanged (nothing downloaded), 1 error. +""" + +from __future__ import annotations + +import argparse +import gzip +import hashlib +import json +import os +import sys +import urllib.error +import urllib.request +from pathlib import Path + +MANIFEST_PATH = "/export/v1/manifest" +MESSAGES_PATH = "/export/v1/messages" +USER_AGENT = "navi-export/1" + + +class PullError(Exception): + pass + + +def _die(message: str) -> int: + print(f"error: {message}", file=sys.stderr) + return 1 + + +def _request(url: str, token: str, *, accept: str = "application/json") -> urllib.request.Request: + return urllib.request.Request( + url, + headers={ + "Authorization": f"Bearer {token}", + "Accept": accept, + "User-Agent": USER_AGENT, + }, + ) + + +def fetch_manifest(base: str, token: str, timeout: float) -> dict: + try: + with urllib.request.urlopen(_request(base + MANIFEST_PATH, token), timeout=timeout) as resp: + return json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + if exc.code == 401: + raise PullError("the server rejected the token (401)") from exc + if exc.code == 503: + raise PullError("the server has no export built yet (503)") from exc + raise PullError(f"manifest request failed: HTTP {exc.code}") from exc + except urllib.error.URLError as exc: + raise PullError(f"cannot reach {base}: {exc.reason}") from exc + + +def _read_token(args) -> str: + if args.token: + return args.token.strip() + if args.token_file: + path = Path(args.token_file).expanduser() + try: + return path.read_text(encoding="utf-8").strip() + except OSError as exc: + raise PullError(f"cannot read token file {path}: {exc}") from exc + raise PullError("one of --token or --token-file is required") + + +def _load_state(path: Path) -> dict: + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + + +def _save_state(path: Path, state: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(json.dumps(state, indent=2) + "\n", encoding="utf-8") + os.replace(tmp, path) + + +def _download(base: str, token: str, timeout: float, dest: Path) -> str: + """Stream the ciphertext to `dest`; return the sha256 of what landed.""" + digest = hashlib.sha256() + try: + with urllib.request.urlopen( + _request(base + MESSAGES_PATH, token, accept="application/octet-stream"), + timeout=timeout, + ) as resp, open(dest, "wb") as out: + while True: + block = resp.read(1 << 20) + if not block: + break + digest.update(block) + out.write(block) + except urllib.error.HTTPError as exc: + dest.unlink(missing_ok=True) + raise PullError(f"download failed: HTTP {exc.code}") from exc + except urllib.error.URLError as exc: + dest.unlink(missing_ok=True) + raise PullError(f"download failed: {exc.reason}") from exc + return digest.hexdigest() + + +def _decrypt(identity_path: Path, src: Path, dest: Path) -> None: + import pyrage + + try: + identity = pyrage.x25519.Identity.from_str( + identity_path.expanduser().read_text(encoding="utf-8").strip() + ) + except (OSError, pyrage.IdentityError) as exc: + raise PullError(f"cannot use identity {identity_path}: {exc}") from exc + + try: + with open(src, "rb") as enc, open(dest, "wb") as out: + pyrage.decrypt_io(enc, out, [identity]) + except pyrage.DecryptError as exc: + dest.unlink(missing_ok=True) + raise PullError( + "decryption failed — the file is not for this identity, or it is corrupt" + ) from exc + + +def _verify_gzip(path: Path) -> int: + """Fully decompress once: catches truncation now, not at n-gram build time.""" + total = 0 + try: + with gzip.open(path, "rb") as f: + for block in iter(lambda: f.read(1 << 20), b""): + total += len(block) + except (OSError, EOFError) as exc: + raise PullError(f"decrypted file is not a valid gzip stream: {exc}") from exc + return total + + +def cmd_pull(args) -> int: + base = args.url.rstrip("/") + out = Path(args.out).expanduser() + state_path = Path(args.state).expanduser() if args.state else out.with_suffix(out.suffix + ".state.json") + token = _read_token(args) + state = _load_state(state_path) + + manifest = fetch_manifest(base, token, args.timeout) + content_id = manifest.get("content_id", "") + if not content_id or not manifest.get("file", {}).get("sha256"): + raise PullError("manifest is missing content_id/sha256 — server and script disagree") + + if not args.force and state.get("content_id") == content_id and out.exists(): + print(f"unchanged since {manifest.get('built_at')} (content_id {content_id[:12]})") + return 3 + + out.parent.mkdir(parents=True, exist_ok=True) + part_enc = out.with_name(out.name + ".part.age") + part_plain = out.with_name(out.name + ".part") + try: + got = _download(base, token, args.timeout, part_enc) + expected = manifest["file"]["sha256"] + if got != expected: + raise PullError(f"checksum mismatch: got {got[:12]}, manifest says {expected[:12]}") + + _decrypt(Path(args.identity), part_enc, part_plain) + plain_bytes = _verify_gzip(part_plain) + os.replace(part_plain, out) + finally: + part_enc.unlink(missing_ok=True) + part_plain.unlink(missing_ok=True) + + _save_state(state_path, {"content_id": content_id, "built_at": manifest.get("built_at")}) + + window = manifest.get("window", {}) + span = f"{window.get('from') or '?'} .. {window.get('to') or '?'}" + if window.get("truncated"): + span += f" (window capped at {window.get('max_bytes', 0) / 1024 / 1024:.0f} MB)" + print( + f"updated {out}: {plain_bytes / 1024 / 1024:.1f} MB, " + f"{manifest.get('rows', 0)} rows, {manifest.get('sessions', 0)} sessions, {span}" + ) + print(f"built_at {manifest.get('built_at')} content_id {content_id[:12]}") + return 0 + + +def cmd_keygen(args) -> int: + import pyrage + + out = Path(args.out).expanduser() + out.mkdir(parents=True, exist_ok=True) + key_path = out / "key.txt" + if key_path.exists() and not args.force: + return _die(f"{key_path} already exists (use --force to overwrite — it invalidates the old export)") + identity = pyrage.x25519.Identity.generate() + key_path.write_text(str(identity) + "\n", encoding="utf-8") + os.chmod(key_path, 0o600) + print(key_path) + print(str(identity.to_public())) + print("Set NAVI_EXPORT_AGE_RECIPIENT on Navi to the line above.", file=sys.stderr) + return 0 + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser(prog="navi-export", description=__doc__.splitlines()[0]) + sub = p.add_subparsers(dest="cmd", required=True) + + pull = sub.add_parser("pull", help="fetch, decrypt and install the snapshot") + pull.add_argument("--url", required=True, help="Navi base URL, e.g. https://navi.example") + pull.add_argument("--token", help="export token (prefer --token-file)") + pull.add_argument("--token-file", help="file holding the export token") + pull.add_argument("--identity", required=True, help="age secret key file") + pull.add_argument("--out", required=True, help="where to write the decrypted gzip-JSONL") + pull.add_argument("--state", help="state file (default: .state.json)") + pull.add_argument("--timeout", type=float, default=120.0, help="per-request timeout, seconds") + pull.add_argument("--force", action="store_true", help="download even if content_id is unchanged") + pull.set_defaults(func=cmd_pull) + + keygen = sub.add_parser("keygen", help="generate an age keypair for the consumer") + keygen.add_argument("--out", default=str(Path.home() / ".config" / "navi-export")) + keygen.add_argument("--force", action="store_true") + keygen.set_defaults(func=cmd_keygen) + return p + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + return args.func(args) + except PullError as exc: + return _die(str(exc)) + except KeyboardInterrupt: + return 130 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/unit/api/test_export.py b/tests/unit/api/test_export.py new file mode 100644 index 0000000..660562d --- /dev/null +++ b/tests/unit/api/test_export.py @@ -0,0 +1,147 @@ +"""Export endpoints: token gate, manifest payload, conditional GET.""" + +from __future__ import annotations + +import json + +import pytest +from fastapi.testclient import TestClient + +from navi.api.routes import export as export_mod +from navi.config import Settings +from navi.export import builder + +TOKEN = "e" * 48 + + +@pytest.fixture +def export_env(tmp_path, monkeypatch): + out = tmp_path / "export_files" + out.mkdir() + settings = Settings( + _env_file=None, + navi_persona_file="", + navi_export_dir=str(out), + navi_export_token=TOKEN, + navi_export_age_recipient="age1qqqq", + ) + monkeypatch.setattr(export_mod, "settings", settings) + monkeypatch.setattr(builder, "settings", settings) + + from navi.main import app + + return TestClient(app), out + + +def _publish(out, content_id="a" * 64, payload=b"ciphertext", built_at="2026-11-04T12:00:00+00:00"): + (out / builder.FILE_NAME).write_bytes(payload) + manifest = { + "export_version": 1, + "built_at": built_at, + "content_id": content_id, + "file": {"name": builder.FILE_NAME, "bytes": len(payload), "sha256": "b" * 64}, + "rows": 3, + "sessions": 1, + "sessions_total": 1, + "fields": builder.FIELDS, + "roles": ["user", "assistant"], + "window": {"from": None, "to": None, "truncated": False, "max_bytes": 1}, + "age_recipient": "age1qqqq", + } + (out / builder.MANIFEST_NAME).write_text(json.dumps(manifest)) + return manifest + + +def test_missing_or_wrong_token_is_rejected(export_env): + client, out = export_env + _publish(out) + assert client.get("/export/v1/manifest").status_code == 401 + bad = client.get("/export/v1/manifest", headers={"Authorization": "Bearer nope"}) + assert bad.status_code == 401 + assert bad.headers["www-authenticate"] == "Bearer" + # Same answer for a wrong token and no token: the endpoint confirms nothing. + assert client.get("/export/v1/messages").status_code == 401 + + +def test_manifest_is_served_with_a_valid_token(export_env): + client, out = export_env + _publish(out) + resp = client.get("/export/v1/manifest", headers={"Authorization": f"Bearer {TOKEN}"}) + assert resp.status_code == 200 + assert resp.json()["content_id"] == "a" * 64 + assert resp.headers["cache-control"] == "no-store" + + +def test_nothing_built_answers_503(export_env): + client, _ = export_env + resp = client.get("/export/v1/manifest", headers={"Authorization": f"Bearer {TOKEN}"}) + assert resp.status_code == 503 + + +def test_file_is_served_with_an_etag_and_304_when_unchanged(export_env): + client, out = export_env + _publish(out) + headers = {"Authorization": f"Bearer {TOKEN}"} + + first = client.get("/export/v1/messages", headers=headers) + assert first.status_code == 200 + assert first.content == b"ciphertext" + etag = first.headers["etag"] + assert etag.startswith('"v1-') + assert first.headers["x-export-content-id"] == "a" * 64 + + again = client.get("/export/v1/messages", headers={**headers, "If-None-Match": etag}) + assert again.status_code == 304 + assert again.content == b"" + + +def test_new_content_id_changes_the_etag(export_env): + client, out = export_env + headers = {"Authorization": f"Bearer {TOKEN}"} + _publish(out, content_id="a" * 64) + first = client.get("/export/v1/messages", headers=headers).headers["etag"] + + out.joinpath(builder.MANIFEST_NAME).unlink() + _publish(out, content_id="c" * 64) + second = client.get("/export/v1/messages", headers=headers).headers["etag"] + assert first != second + + +def test_unconfigured_token_disables_the_endpoints(export_env, monkeypatch): + client, out = export_env + _publish(out) + monkeypatch.setattr( + export_mod, "settings", Settings(_env_file=None, navi_persona_file="", navi_export_token="") + ) + resp = client.get("/export/v1/manifest", headers={"Authorization": "Bearer anything"}) + assert resp.status_code == 503 + + +def test_health_reports_export_freshness_without_touching_status(export_env, monkeypatch): + _, out = export_env + from datetime import datetime, timedelta, timezone + + from navi.api.routes import health as health_mod + from navi.export import loop as loop_mod + + monkeypatch.setattr(health_mod, "settings", builder.settings) + monkeypatch.setattr(loop_mod, "settings", builder.settings) + + assert health_mod._export_status()["built"] is False + + now = datetime.now(timezone.utc) + _publish(out, built_at=now.isoformat()) + fresh = health_mod._export_status() + assert fresh["built"] is True and fresh["stale"] is False and fresh["rows"] == 3 + + _publish(out, built_at=(now - timedelta(hours=24)).isoformat()) + assert health_mod._export_status()["stale"] is True + + +def test_health_omits_export_when_not_configured(export_env, monkeypatch): + from navi.api.routes import health as health_mod + + monkeypatch.setattr( + health_mod, "settings", Settings(_env_file=None, navi_persona_file="") + ) + assert health_mod._export_status() is None diff --git a/tests/unit/export/test_builder.py b/tests/unit/export/test_builder.py new file mode 100644 index 0000000..0c132bf --- /dev/null +++ b/tests/unit/export/test_builder.py @@ -0,0 +1,170 @@ +"""Export builder: window math, deterministic content id, encryption roundtrip.""" + +from __future__ import annotations + +import gzip +import json +from datetime import datetime, timezone + +import pyrage +import pytest + +from navi.config import Settings +from navi.export import builder + +TS = datetime(2026, 11, 1, 12, 0, tzinfo=timezone.utc) + + +class _Tx: + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + +class FakeConn: + """Just enough of an asyncpg connection for the builder's query sequence.""" + + def __init__(self, sessions: list[str], rows: list[dict], total_sessions: int): + self._sessions = sessions + self._rows = rows + self._total = total_sessions + self.calls: list[tuple[str, tuple]] = [] + + def transaction(self, **kwargs): + return _Tx() + + async def fetch(self, sql, *args): + self.calls.append((sql, args)) + assert args[0] == ["user", "assistant"] + return [{"session_id": s} for s in self._sessions] + + async def fetchval(self, sql, *args): + return self._total + + def cursor(self, sql, *args, prefetch=500): + self.calls.append((sql, args)) + conn = self + + async def gen(): + for row in conn._rows: + yield row + + return gen() + + +class FakePool: + def __init__(self, conn): + self._conn = conn + + def acquire(self): + pool = self + + class _Acq: + async def __aenter__(self): + return pool._conn + + async def __aexit__(self, *exc): + return False + + return _Acq() + + +def _row(session: str, seq: int, role: str, content: str) -> dict: + return { + "session_id": session, + "sequence_number": seq, + "role": role, + "content": content, + "created_at": TS, + } + + +@pytest.fixture +def export_settings(tmp_path, monkeypatch): + identity = pyrage.x25519.Identity.generate() + s = Settings( + _env_file=None, + navi_persona_file="", + navi_export_dir=str(tmp_path / "export_files"), + navi_export_token="t" * 32, + navi_export_age_recipient=str(identity.to_public()), + navi_export_roles="user,assistant", + navi_export_max_bytes=10_000, + ) + monkeypatch.setattr(builder, "settings", s) + return s, identity + + +@pytest.mark.asyncio +async def test_build_writes_encrypted_jsonl_and_manifest(export_settings): + _settings, identity = export_settings + rows = [_row("s1", 1, "user", "привет"), _row("s1", 2, "assistant", 'he said "hi"')] + pool = FakePool(FakeConn(["s1"], rows, total_sessions=1)) + + manifest = await builder.build_export(pool) + + age_path, manifest_path = builder.export_paths() + assert age_path.exists() and manifest_path.exists() + assert manifest["rows"] == 2 + assert manifest["sessions"] == 1 + assert manifest["window"]["truncated"] is False + assert manifest["fields"] == builder.FIELDS + assert manifest["age_recipient"] == str(identity.to_public()) + assert manifest_path.read_text().count("content_id") == 1 + + decrypted = pyrage.decrypt(age_path.read_bytes(), [identity]) + lines = [json.loads(line) for line in gzip.decompress(decrypted).decode().splitlines()] + assert lines == [ + {"v": 1, "session": "s1", "seq": 1, "role": "user", "content": "привет", "ts": TS.isoformat()}, + {"v": 1, "session": "s1", "seq": 2, "role": "assistant", "content": 'he said "hi"', "ts": TS.isoformat()}, + ] + # The plaintext never stays on disk. + assert not list(builder.export_paths()[0].parent.glob("*.tmp")) + assert not list(builder.export_paths()[0].parent.glob("*.plain*")) + + +@pytest.mark.asyncio +async def test_same_corpus_keeps_the_ciphertext_and_its_content_id(export_settings): + _, _ = export_settings + rows = [_row("s1", 1, "user", "hello")] + first = await builder.build_export(FakePool(FakeConn(["s1"], rows, 1))) + age_path, _ = builder.export_paths() + before = age_path.read_bytes() + mtime = age_path.stat().st_mtime_ns + + second = await builder.build_export(FakePool(FakeConn(["s1"], rows, 1))) + + assert second["content_id"] == first["content_id"] + assert age_path.read_bytes() == before + assert age_path.stat().st_mtime_ns == mtime + # built_at still moves, so freshness monitoring sees a live builder. + assert second["built_at"] >= first["built_at"] + + +@pytest.mark.asyncio +async def test_changed_corpus_produces_a_new_content_id(export_settings): + _, _ = export_settings + first = await builder.build_export(FakePool(FakeConn(["s1"], [_row("s1", 1, "user", "a")], 1))) + second = await builder.build_export(FakePool(FakeConn(["s1"], [_row("s1", 1, "user", "b")], 1))) + assert first["content_id"] != second["content_id"] + + +@pytest.mark.asyncio +async def test_window_reports_truncation(export_settings): + _, _ = export_settings + manifest = await builder.build_export( + FakePool(FakeConn(["newest"], [_row("newest", 1, "user", "x")], total_sessions=7)) + ) + assert manifest["window"]["truncated"] is True + assert manifest["sessions"] == 1 + assert manifest["sessions_total"] == 7 + + +@pytest.mark.asyncio +async def test_unconfigured_export_refuses_to_build(tmp_path, monkeypatch): + s = Settings(_env_file=None, navi_persona_file="", navi_export_age_recipient="", navi_export_token="") + monkeypatch.setattr(builder, "settings", s) + with pytest.raises(RuntimeError): + await builder.build_export(FakePool(FakeConn([], [], 0))) diff --git a/tests/unit/export/test_cli.py b/tests/unit/export/test_cli.py new file mode 100644 index 0000000..68ab22c --- /dev/null +++ b/tests/unit/export/test_cli.py @@ -0,0 +1,183 @@ +"""End-to-end check of the consumer CLI against the wire contract. + +A stub server stands in for Navi (same paths, same bearer check, same manifest +shape) and the CLI is driven as a module so exit codes are asserted directly. +This is the contract test for `scripts/navi_export.py`. +""" + +from __future__ import annotations + +import gzip +import hashlib +import importlib.util +import json +import threading +from contextlib import contextmanager +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +import pyrage + +SCRIPT = Path(__file__).resolve().parents[3] / "scripts" / "navi_export.py" +TOKEN = "s" * 40 +CORPUS = ( + b'{"v":1,"session":"s1","seq":1,"role":"user","content":"hi","ts":null}\n' + b'{"v":1,"session":"s1","seq":2,"role":"assistant","content":"hello","ts":null}\n' +) + + +def _cli(): + spec = importlib.util.spec_from_file_location("navi_export_cli", SCRIPT) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _pack(identity): + cipher = pyrage.encrypt(gzip.compress(CORPUS, mtime=0), [identity.to_public()]) + manifest = { + "export_version": 1, + "built_at": "2026-11-04T12:00:00+00:00", + "content_id": hashlib.sha256(CORPUS).hexdigest(), + "file": { + "name": "messages-v1.jsonl.gz.age", + "bytes": len(cipher), + "sha256": hashlib.sha256(cipher).hexdigest(), + }, + "rows": 2, + "sessions": 1, + "sessions_total": 1, + "window": {"from": None, "to": None, "truncated": False, "max_bytes": 1}, + "age_recipient": str(identity.to_public()), + } + return manifest, cipher + + +@contextmanager +def _serve(manifest: dict, cipher: bytes): + state = {"downloads": 0} + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def _reply(self, code: int, body: bytes = b"", ctype="application/json"): + self.send_response(code) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self): + if self.headers.get("Authorization") != f"Bearer {TOKEN}": + self._reply(401) + return + if self.path == "/export/v1/manifest": + self._reply(200, json.dumps(manifest).encode()) + elif self.path == "/export/v1/messages": + state["downloads"] += 1 + self._reply(200, cipher, "application/octet-stream") + else: + self._reply(404) + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_address[1]}", state + finally: + server.shutdown() + server.server_close() + + +def _pull(url, identity_path, out, token=TOKEN): + return _cli().main( + [ + "pull", + "--url", url, + "--token", token, + "--identity", str(identity_path), + "--out", str(out), + ] + ) + + +def _identity(tmp_path) -> Path: + path = tmp_path / "key.txt" + path.write_text(str(pyrage.x25519.Identity.generate()) + "\n") + return path + + +def test_pull_downloads_decrypts_and_installs(tmp_path): + identity = pyrage.x25519.Identity.generate() + key = tmp_path / "key.txt" + key.write_text(str(identity) + "\n") + manifest, cipher = _pack(identity) + out = tmp_path / "corpus" / "messages.jsonl.gz" + + with _serve(manifest, cipher) as (url, state): + assert _pull(url, key, out) == 0 + + assert state["downloads"] == 1 + assert gzip.decompress(out.read_bytes()) == CORPUS + assert json.loads((out.parent / (out.name + ".state.json")).read_text())["content_id"] == ( + manifest["content_id"] + ) + assert not list(out.parent.glob("*.part*")) + + +def test_second_pull_downloads_nothing(tmp_path): + identity = pyrage.x25519.Identity.generate() + key = tmp_path / "key.txt" + key.write_text(str(identity) + "\n") + manifest, cipher = _pack(identity) + out = tmp_path / "messages.jsonl.gz" + + with _serve(manifest, cipher) as (url, state): + assert _pull(url, key, out) == 0 + assert _pull(url, key, out) == 3 + assert state["downloads"] == 1 + + +def test_wrong_token_exits_1(tmp_path): + identity = pyrage.x25519.Identity.generate() + key = tmp_path / "key.txt" + key.write_text(str(identity) + "\n") + manifest, cipher = _pack(identity) + + with _serve(manifest, cipher) as (url, _state): + assert _pull(url, key, tmp_path / "out.gz", token="wrong") == 1 + + +def test_checksum_mismatch_exits_1(tmp_path): + identity = pyrage.x25519.Identity.generate() + key = tmp_path / "key.txt" + key.write_text(str(identity) + "\n") + manifest, cipher = _pack(identity) + manifest["file"]["sha256"] = "0" * 64 + out = tmp_path / "out.gz" + + with _serve(manifest, cipher) as (url, _state): + assert _pull(url, key, out) == 1 + assert not out.exists() + + +def test_foreign_identity_exits_1(tmp_path): + owner = pyrage.x25519.Identity.generate() + manifest, cipher = _pack(owner) + key = _identity(tmp_path) # a different key + out = tmp_path / "out.gz" + + with _serve(manifest, cipher) as (url, _state): + assert _pull(url, key, out) == 1 + assert not out.exists() + + +def test_keygen_writes_a_usable_identity(tmp_path): + cli = _cli() + assert cli.main(["keygen", "--out", str(tmp_path)]) == 0 + key = tmp_path / "key.txt" + assert key.read_text().startswith("AGE-SECRET-KEY-") + assert key.stat().st_mode & 0o777 == 0o600 + # Refuses to clobber an existing key unless forced. + assert cli.main(["keygen", "--out", str(tmp_path)]) == 1 diff --git a/uv.lock b/uv.lock index 444494e..d1ab629 100644 --- a/uv.lock +++ b/uv.lock @@ -1922,6 +1922,7 @@ { name = "playwright" }, { name = "pydantic" }, { name = "pydantic-settings" }, + { name = "pyrage" }, { name = "python-dateutil" }, { name = "python-dotenv" }, { name = "python-multipart" }, @@ -1965,6 +1966,7 @@ { name = "playwright", specifier = ">=1.40" }, { name = "pydantic", specifier = ">=2.7" }, { name = "pydantic-settings", specifier = ">=2.3" }, + { name = "pyrage", specifier = ">=1.2" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0" }, { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.23" }, { name = "python-dateutil", specifier = ">=2.8" }, @@ -2518,6 +2520,18 @@ ] [[package]] +name = "pyrage" +version = "1.4.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/15/a4/612910dbedf7d9fb24ec58474849438b6f88e078e1a100f73a54a605d228/pyrage-1.4.0.tar.gz", hash = "sha256:545bf188e8d217ec88c32bc8e8bf56026c8c0b1b1f96d2688cedf1427ab0e827", size = 32198, upload-time = "2026-08-23T22:48:59.906Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/13/bb/6b167a449e95ecf0eaf4277c949a89a29dfc08cc533f710a9e6f12bef513/pyrage-1.4.0-cp310-abi3-macosx_10_12_x86_64.macosx_11_0_arm64.macosx_10_12_universal2.whl", hash = "sha256:bcf4177aea91a9f48951db8001c1385baafbae62cc15d691ef43efebaab6b8db", size = 1701313, upload-time = "2026-08-23T22:48:53.835Z" }, + { url = "https://files.pythonhosted.org/packages/5a/cb/d88aaab0b89141802473f6139fe8e05519d1b6ac06bff8231a1378a5674f/pyrage-1.4.0-cp310-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:f868b360ccd3f836326d42b532fae3e8eb8340a88cca0a39c0ae10fce68f6cc5", size = 882978, upload-time = "2026-08-23T22:48:55.525Z" }, + { url = "https://files.pythonhosted.org/packages/d9/f0/41da7d88cfdad2eba492f3275ec50a65916ccd3a78140fb15875569ad1ff/pyrage-1.4.0-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:7f4eb0cbc2b4aee5deac9862c47b3895866df5554153951af822e9c581e68558", size = 924673, upload-time = "2026-08-23T22:48:57.1Z" }, + { url = "https://files.pythonhosted.org/packages/d3/8c/354fdb9ee7b45a2060272c623536b2cf9f0996de2242eaac177cff3f8003/pyrage-1.4.0-cp310-abi3-win_amd64.whl", hash = "sha256:c4a5dcd24a14b5e0d4cccff13649a2b8f1609ad45d6438bb85f35d84b0c8155e", size = 848219, upload-time = "2026-08-23T22:48:58.702Z" }, +] + +[[package]] name = "pytest" version = "9.1.1" source = { registry = "https://pypi.org/simple" }