diff --git a/common/storage/README.md b/common/storage/README.md index e6f2c28..839b1c7 100644 --- a/common/storage/README.md +++ b/common/storage/README.md @@ -20,6 +20,7 @@ The following storage backends are supported: - **Redis** (`redis_storage/`): In-memory data structure store - **SFTP** (`sftp/`): Secure file transfer protocol storage - **File** (`file/`): Local file system storage +- **Utopia** (`utopia/`): Bitemporal knowledge graph, vCons rendered as dated documents ## Configuration diff --git a/common/storage/utopia/README.md b/common/storage/utopia/README.md new file mode 100644 index 0000000..f2d497c --- /dev/null +++ b/common/storage/utopia/README.md @@ -0,0 +1,80 @@ +# Utopia Storage Module for vCon Server + +Sends each vCon to a [Utopia](https://github.com/deeplethe/utopia) knowledge base. Utopia is a self-hosted bitemporal knowledge graph: it extracts facts from documents, dates each one on both a world clock and a record clock, and keeps the quoted evidence for every fact. Point it at your conversations and you can ask what was said about a customer, product or issue as of any date, with the source vCon behind every answer. + +## How a vCon becomes a Utopia document + +Utopia reads prose, not JSON, and it dates a document only from a leading `YYYY-MM-DD` line in a Markdown file. So `save` renders the vCon as dated sentences and uploads it as `.md`: + +``` +2026-09-18 + +# Conversation 8f752fc9-d6d8-4d2c-a9d0-d859f61d7c51 + ++15085550100 took part. +On 2026-09-18 at 14:02, +15085550100 and Support had a recorded call lasting 240 seconds. + +On 2026-09-18, the conversation was summarised: ... + +Transcript of the conversation on 2026-09-18 at 14:02: + +On 2026-09-18, Speaker 0: ... + +Lawful basis: consent, granted for recording, analysis, expires 2027-01-01. +Source: vCon 8f752fc9-d6d8-4d2c-a9d0-d859f61d7c51, created 2026-09-18T14:02:00Z. +``` + +Sentences rather than key-value pairs, because `On 2026-09-18, X called Y` extracts as a fact with a start date, and `created_at: 2026-09-18` leaves the model guessing. Utopia extracts chunk by chunk, so every summary and transcript paragraph repeats the date rather than relying on a heading in another chunk. + +The renderer reads `wtf_transcription`, OpenAI, Deepgram and legacy `transcript` bodies, `summary` analysis, the `tags` attachment and the `lawful_basis` attachment. It accepts legacy records (`vcon: "0.0.1"`, attachment `type` instead of `purpose`, dict bodies). + +## Behaviour + +- **save** uploads the rendering. Utopia skips content it already holds (same sha256 in the KB), so an unchanged vCon is a no-op. When the rendering changed, for example after a new analysis or an amendment, the new document is uploaded first and the older `.md` is then deleted, so a failed upload loses nothing. +- **delete** deletes every live `.md`. A Utopia delete is a tombstone: it invalidates facts that no other document supports and can be undone from the Utopia console. +- **get** is not implemented. Utopia holds the rendering, not the vCon. + +Failures raise, so the conserver's storage retry and dead-letter handling apply. + +## Configuration + +| Option | Description | Default | +| --- | --- | --- | +| base_url | Utopia REST API root | `http://127.0.0.1:1516/api/v1` | +| kb_id | Knowledge base UUID. Required | `""` | +| email | Utopia service account. Required | `""` | +| password | Service account password. Required | `""` | +| timeout | Request timeout in seconds | `30` | +| transient_retries | Retries for connection errors and 429/502/503/504 | `3` | +| transient_backoff_base_s | Exponential backoff factor in seconds | `0.5` | +| require_analysis_grant | Skip vCons whose lawful basis does not grant `analysis` | `false` | + +```yaml +storages: + utopia: + module: storage.utopia + options: + base_url: http://utopia:1516/api/v1 + kb_id: 00000000-0000-0000-0000-000000000000 + email: conserver@example.com + password: + require_analysis_grant: false +``` + +Add `utopia` to a chain's `storages` list. + +The service account needs the Editor role on the knowledge base. Utopia's REST API accepts session tokens only, not the personal access tokens it issues for MCP, so the module logs in with the account and logs in again when its token expires (Utopia tokens last 7 days). + +## Lawful basis + +Every rendering ends with one lawful basis sentence, either the basis and its grants or `Lawful basis: none recorded.` A missing basis is never filled in. With `require_analysis_grant: true`, a vCon whose basis does not grant `analysis`, or has expired, is not sent and a warning names its uuid. The default is off because most existing corpora predate the lawful basis extension; turn it on for regulated data. + +## Before you send real conversations + +- Utopia sends document chunks to a language model to extract facts. Point Utopia at a local model endpoint (Ollama, vLLM) if transcripts may not leave your environment, and run a redaction link earlier in the chain. +- Utopia is pre-1.0. Pin its image version, and do not expose it to the public internet. + +## Not yet handled + +- A `redacted` vCon should purge the original's document permanently rather than tombstone it. That needs the Admin role and Utopia's purge endpoint. +- Party phone numbers and emails are rendered in text but not yet seeded as Utopia entity aliases. diff --git a/common/storage/utopia/__init__.py b/common/storage/utopia/__init__.py new file mode 100644 index 0000000..0ba1c7e --- /dev/null +++ b/common/storage/utopia/__init__.py @@ -0,0 +1,361 @@ +""" +Utopia storage module for vcon-server + +Pushes each vCon into a Utopia knowledge base (https://github.com/deeplethe/utopia), +a bitemporal knowledge graph that extracts dated facts from documents. + +Utopia reads prose, not JSON, and dates a document only from a leading +``YYYY-MM-DD`` line in a Markdown file. So the vCon is rendered here as dated +sentences and uploaded as ``.md``. The filename is the only link between +the vCon and Utopia's own document id: ``get`` is not offered, because Utopia +holds our rendering, not the vCon. + +Endpoints used (all under ``base_url``): +- POST auth/login - Trade the service account for a JWT +- GET kbs/{kb_id}/documents?q= - Find this vCon's document by filename +- POST kbs/{kb_id}/documents - Upload the rendering (multipart) +- DELETE documents/{id} - Tombstone a superseded rendering + +Configuration options: +- base_url: Utopia REST API root (default: http://127.0.0.1:1516/api/v1) +- kb_id: Knowledge base UUID. Required +- email, password: Utopia service account with Editor role on the KB. Required. + Utopia's REST API takes session JWTs only, not personal access tokens +- timeout: Request timeout in seconds (default: 30) +- transient_retries: Retries for transient failures (default: 3) +- transient_backoff_base_s: Backoff factor in seconds (default: 0.5) +- require_analysis_grant: Skip vCons whose lawful basis does not grant + ``analysis`` (default: False) +""" + +import json +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional, Tuple + +from lib.logging_utils import init_logger +from lib.vcon_redis import VconRedis +# ponytail: shared retry policy. POST retries are safe here too: Utopia skips +# an upload whose sha256 is already in the KB. +from storage.vcon_mcp import _session +from vcon import Vcon + +logger = init_logger(__name__) + +default_options: Dict[str, Any] = { + "base_url": "http://127.0.0.1:1516/api/v1", + "kb_id": "", + "email": "", + "password": "", + "timeout": 30, + "transient_retries": 3, + "transient_backoff_base_s": 0.5, + "require_analysis_grant": False, +} + +TRANSCRIPT_TYPES = ("transcript", "wtf_transcription") + +# (base_url, email) -> JWT. ponytail: per process, re-login on 401. Utopia's +# tokens last 7 days, so this logs in about once a week per worker. +_tokens: Dict[Tuple[str, str], str] = {} + + +# ---------------------------------------------------------------- rendering + + +def _date(ts: Any) -> str: + return str(ts)[:10] if ts else "" + + +def _when(ts: Any) -> str: + """'2025-02-27 at 16:44' from an ISO 8601 instant, or just the day.""" + ts = str(ts or "") + return f"{ts[:10]} at {ts[11:16]}" if len(ts) >= 16 else ts[:10] + + +def _party_label(parties: List[dict], index: Any) -> str: + if isinstance(index, int) and 0 <= index < len(parties): + p = parties[index] or {} + return p.get("name") or p.get("tel") or p.get("mailto") or f"party {index}" + return f"party {index}" + + +def _party_sentence(p: dict) -> str: + who = p.get("name") or p.get("tel") or p.get("mailto") or "An unnamed party" + sentence = f"{who} took part" + if p.get("role"): + sentence += f" as {p['role']}" + reach = [f"phone {p['tel']}" if p.get("tel") and p["tel"] != who else "", + f"email {p['mailto']}" if p.get("mailto") and p["mailto"] != who else ""] + reach = [r for r in reach if r] + if reach: + sentence += ", reachable by " + " and ".join(reach) + return sentence + "." + + +def _flat_ints(value: Any) -> List[int]: + """dialog.parties may nest lists (one per channel); flatten to indexes.""" + if isinstance(value, int): + return [value] + if isinstance(value, list): + return [i for v in value for i in _flat_ints(v)] + return [] + + +def _dialog_sentence(d: dict, parties: List[dict]) -> str: + """'On 2025-02-27 at 16:44, Ann and Bob had a recorded call lasting 240 seconds.'""" + kind = {"recording": "a recorded call", "text": "a text conversation", + "transfer": "a transfer", "incomplete": "an incomplete call"}.get( + d.get("type"), "a conversation") + names = [_party_label(parties, i) for i in _flat_ints(d.get("parties"))] + sentence = f"{' and '.join(names)} had {kind}" if names else f"There was {kind}" + duration = d.get("duration") + if isinstance(duration, (int, float)) and duration > 0: + sentence += f" lasting {round(duration)} seconds" + if d.get("start"): + sentence = f"On {_when(d['start'])}, {sentence}" + return sentence + "." + + +def _transcript_lines(body: Any) -> List[str]: + """Tolerates wtf_transcription, OpenAI, Deepgram and legacy bodies.""" + if isinstance(body, str): + return [body.strip()] if body.strip() else [] + if not isinstance(body, dict): + return [] + lines = [] + for s in body.get("segments") or []: + if isinstance(s, dict) and str(s.get("text") or "").strip(): + who = f"Speaker {s['speaker']}: " if s.get("speaker") is not None else "" + lines.append(who + str(s["text"]).strip()) + if lines: + return lines + text = body.get("transcript") + if isinstance(text, dict): + text = text.get("text") + text = text or body.get("text") + return [text.strip()] if isinstance(text, str) and text.strip() else [] + + +def _body(entry: dict) -> Any: + try: + body = Vcon.decoded_body(entry) + except ValueError: # malformed json body; render nothing rather than fail + return None + if isinstance(body, str) and entry.get("encoding") not in ("none", "base64url"): + try: + return json.loads(body) + except ValueError: + pass + return body + + +def _purpose(att: dict) -> Optional[str]: + return att.get("purpose") or att.get("type") + + +def _lawful_basis(attachments: List[dict]) -> Tuple[str, bool]: + """One sentence for the rendering, and whether ``analysis`` is granted now. + + Never fills a gap: no attachment reads as "none recorded". + """ + for att in attachments: + if _purpose(att) != "lawful_basis": + continue + body = _body(att) + if not isinstance(body, dict): + continue + granted = [g.get("purpose") for g in body.get("purpose_grants") or [] + if isinstance(g, dict) and g.get("granted") is True and g.get("purpose")] + expiration = body.get("expiration") + expired = False + if expiration: + try: + exp = datetime.fromisoformat(str(expiration).replace("Z", "+00:00")) + if exp.tzinfo is None: + exp = exp.replace(tzinfo=timezone.utc) + expired = exp <= datetime.now(timezone.utc) + except ValueError: + expired = True # an unreadable expiry is not a valid grant + sentence = f"Lawful basis: {body.get('lawful_basis') or 'unspecified'}" + sentence += f", granted for {', '.join(granted)}" if granted else ", nothing granted" + if expiration: + sentence += f", {'expired' if expired else 'expires'} {_date(expiration)}" + return sentence + ".", "analysis" in granted and not expired + return "Lawful basis: none recorded.", False + + +def _tags(attachments: List[dict]) -> List[str]: + tags = [] + for att in attachments: + if _purpose(att) != "tags": + continue + body = _body(att) + if isinstance(body, dict): + tags += [f"{k}:{v}" for k, v in body.items()] + elif isinstance(body, list): + tags += [str(t) for t in body] + return tags + + +def render(vcon: dict) -> Tuple[str, bool]: + """Render a vCon dict as dated Markdown sentences. + + Returns (markdown, analysis_granted). + """ + parties = vcon.get("parties") or [] + dialogs = vcon.get("dialog") or [] + analysis = vcon.get("analysis") or [] + attachments = vcon.get("attachments") or [] + uuid = vcon.get("uuid", "unknown") + + lines: List[str] = [] + # First non-empty line is the date Utopia uses as the document's date. + day = _date(vcon.get("created_at")) or next( + (_date(d.get("start")) for d in dialogs if d.get("start")), "") + if day: + lines += [day, ""] + lines += [f"# Conversation {uuid}", ""] + if vcon.get("subject"): + lines += [f"Subject: {vcon['subject']}.", ""] + lines += [_party_sentence(p or {}) for p in parties] + lines += [_dialog_sentence(d or {}, parties) for d in dialogs] + + def dialog_of(a: dict) -> dict: + i = a.get("dialog") + return (dialogs[i] or {}) if isinstance(i, int) and 0 <= i < len(dialogs) else {} + + # Utopia extracts chunk by chunk, so a date in a heading does not reach facts + # drawn from later paragraphs. Every summary and transcript paragraph carries + # its own "On ," so the facts in it get a world time. + for a in analysis: + if a.get("type") == "summary": + text = _transcript_lines(_body(a)) + on = _date(dialog_of(a).get("start")) or day + if text: + lines += ["", (f"On {on}, the conversation was summarised: " if on else "Summary: ") + + " ".join(text)] + for a in analysis: + if a.get("type") in TRANSCRIPT_TYPES: + text = _transcript_lines(_body(a)) + if not text: + continue + d = dialog_of(a) + on = _date(d.get("start")) or day + heading = f"Transcript of the conversation on {_when(d.get('start'))}:" if d.get("start") else "Transcript:" + paragraphs = [p.strip() for t in text for p in t.split("\n\n") if p.strip()] + lines += ["", heading, ""] + [f"On {on}, {p}" if on else p for p in paragraphs] + + tags = _tags(attachments) + if tags: + lines += ["", "Tagged " + ", ".join(tags) + "."] + basis_sentence, analysis_granted = _lawful_basis(attachments) + lines += ["", basis_sentence] + source = f"Source: vCon {uuid}" + if vcon.get("created_at"): + source += f", created {vcon['created_at']}" + for key in ("amended", "redacted"): + ref = vcon.get(key) + if isinstance(ref, dict) and ref.get("uuid"): + source += f", {key} from vCon {ref['uuid']}" + lines.append(source + ".") + return "\n".join(lines) + "\n", analysis_granted + + +# --------------------------------------------------------------------- HTTP + + +def _opts(opts: Optional[Dict[str, Any]]) -> Dict[str, Any]: + opts = {**default_options, **(opts or {})} + missing = [k for k in ("kb_id", "email", "password") if not opts.get(k)] + if missing: + raise ValueError(f"utopia storage: missing option(s) {', '.join(missing)}") + return opts + + +def _url(opts: Dict[str, Any], path: str) -> str: + return f"{opts['base_url'].rstrip('/')}/{path.lstrip('/')}" + + +def _filename(vcon_uuid: str) -> str: + return f"{vcon_uuid}.md" + + +def _request(session, opts: Dict[str, Any], method: str, path: str, **kwargs): + """Authenticated request; logs in on first use and once more on a 401.""" + key = (opts["base_url"], opts["email"]) + for attempt in (0, 1): + token = _tokens.get(key) + if not token: + resp = session.post( + _url(opts, "auth/login"), + json={"email": opts["email"], "password": opts["password"]}, + timeout=opts["timeout"], + ) + resp.raise_for_status() + token = _tokens[key] = resp.json()["token"] + resp = session.request( + method, _url(opts, path), + headers={"Authorization": f"Bearer {token}"}, + timeout=opts["timeout"], **kwargs, + ) + if resp.status_code != 401 or attempt: + return resp + _tokens.pop(key, None) + + +def _find(session, opts: Dict[str, Any], vcon_uuid: str) -> List[str]: + """Ids of live Utopia documents holding this vCon's rendering.""" + resp = _request(session, opts, "GET", f"kbs/{opts['kb_id']}/documents", + params={"q": vcon_uuid, "limit": 200}) + resp.raise_for_status() + name = _filename(vcon_uuid) + return [d["id"] for d in resp.json().get("docs", []) if d.get("filename") == name] + + +def save(vcon_uuid: str, opts: Dict[str, Any] = None) -> None: + """Upload the vCon's rendering; tombstone the one it supersedes. + + Upload first, delete second, so a failed upload loses nothing. Identical + content comes back from Utopia as skipped, which is a no-op here. + """ + opts = _opts(opts) + vcon = VconRedis().get_vcon(vcon_uuid) + if not vcon: + raise ValueError(f"vCon {vcon_uuid} not found in Redis") + text, analysis_granted = render(vcon.to_dict()) + if not analysis_granted: + if opts["require_analysis_grant"]: + logger.warning("utopia storage: skipped vCon %s, no lawful basis grants analysis", vcon_uuid) + return + logger.info("utopia storage: vCon %s has no lawful basis granting analysis", vcon_uuid) + + with _session(opts) as session: + previous = _find(session, opts, vcon_uuid) + resp = _request( + session, opts, "POST", f"kbs/{opts['kb_id']}/documents", + files={"file": (_filename(vcon_uuid), text.encode("utf-8"), "text/markdown")}, + ) + resp.raise_for_status() + created = {d["id"] for d in resp.json().get("created") or []} + if not created: + logger.info("utopia storage: vCon %s unchanged in Utopia", vcon_uuid) + return + for doc_id in previous: + if doc_id not in created: + _request(session, opts, "DELETE", f"documents/{doc_id}").raise_for_status() + logger.info("utopia storage: saved vCon %s, superseded %d", vcon_uuid, len(previous)) + + +def delete(vcon_uuid: str, opts: Dict[str, Any] = None) -> bool: + """Tombstone this vCon's rendering in Utopia. + + A Utopia delete is reversible: it invalidates facts with no other source + but keeps the content. Purging (for redaction) needs KB Admin and is not + done here. + """ + opts = _opts(opts) + with _session(opts) as session: + ids = _find(session, opts, vcon_uuid) + for doc_id in ids: + _request(session, opts, "DELETE", f"documents/{doc_id}").raise_for_status() + return bool(ids) diff --git a/common/storage/utopia/test_utopia.py b/common/storage/utopia/test_utopia.py new file mode 100644 index 0000000..e6a5975 --- /dev/null +++ b/common/storage/utopia/test_utopia.py @@ -0,0 +1,219 @@ +"""Utopia storage: rendering, and save/delete against a loopback fake Utopia.""" + +from contextlib import contextmanager +from hashlib import sha256 +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import re +from threading import Thread +from unittest.mock import patch +from urllib.parse import parse_qs, urlparse +import uuid as uuidlib + +import pytest + +import storage.utopia as utopia +from vcon import Vcon + +KB = "11111111-1111-1111-1111-111111111111" +VCON_UUID = "8f752fc9-d6d8-4d2c-a9d0-d859f61d7c51" + + +# A legacy record in the shape of the vcon-dev/fake-vcons corpus. +LEGACY = { + "uuid": VCON_UUID, + "vcon": "0.0.1", + "created_at": "2025-02-27T16:44:59.773035", + "parties": [ + {"tel": "+15085550199", "name": "Jane Doe", "role": "customer"}, + {"mailto": "agent@example.com", "name": "John Roe", "role": "agent"}, + ], + "dialog": [{"type": "text", "start": "2025-02-27T16:44:59.773035", "parties": [1, 0]}], + "analysis": [ + {"type": "transcript", "dialog": 0, "body": {"transcript": "Hello, how can I help?\n\nI need a pickup."}}, + {"type": "summary", "dialog": 0, "body": "The agent answered a question.", "encoding": "none"}, + ], + "attachments": [{"type": "tags", "body": ["disposition:VM Left"]}], +} + + +def current(grants=("recording", "analysis"), expiration="2999-01-01T00:00:00Z", text="Hello there."): + return { + "uuid": VCON_UUID, + "vcon": "0.4.0", + "created_at": "2026-09-18T14:02:00Z", + "parties": [{"tel": "+15085550100"}, {"name": "Support"}], + "dialog": [{"type": "recording", "start": "2026-09-18T14:02:00Z", + "duration": 240.4, "parties": [0, 1]}], + "analysis": [{ + "type": "wtf_transcription", "dialog": 0, "encoding": "json", + "body": json.dumps({"transcript": {"text": text}, + "segments": [{"text": text, "speaker": 0}, + {"text": "Hi.", "speaker": 1}]}), + }], + "attachments": [{ + "purpose": "lawful_basis", "encoding": "json", + "body": json.dumps({ + "lawful_basis": "consent", "expiration": expiration, + "purpose_grants": [{"purpose": p, "granted": True} for p in grants], + "proof_mechanisms": [{"mechanism_type": "audio_recording"}], + }), + }], + } + + +def test_render_legacy_record(): + text, granted = utopia.render(LEGACY) + lines = text.splitlines() + assert lines[0] == "2025-02-27" # Utopia takes the document date from line one + assert "Jane Doe took part as customer, reachable by phone +15085550199." in lines + assert "On 2025-02-27 at 16:44, John Roe and Jane Doe had a text conversation." in lines + assert "On 2025-02-27, Hello, how can I help?" in lines + assert "On 2025-02-27, I need a pickup." in lines # each paragraph dated + assert "On 2025-02-27, the conversation was summarised: The agent answered a question." in lines + assert "Tagged disposition:VM Left." in lines + assert "Lawful basis: none recorded." in lines + assert granted is False + + +def test_render_current_record_with_lawful_basis(): + text, granted = utopia.render(current()) + assert text.startswith("2026-09-18\n") + assert "On 2026-09-18 at 14:02, +15085550100 and Support had a recorded call lasting 240 seconds." in text + assert "+15085550100 took part.\n" in text + assert "On 2026-09-18, Speaker 0: Hello there.\nOn 2026-09-18, Speaker 1: Hi." in text + assert "Lawful basis: consent, granted for recording, analysis, expires 2999-01-01." in text + assert granted is True + + +def test_expired_or_missing_grant_is_not_granted(): + assert utopia.render(current(expiration="2020-01-01T00:00:00Z"))[1] is False + text, granted = utopia.render(current(grants=("recording",))) + assert granted is False and "granted for recording," in text + + +@contextmanager +def fake_utopia(): + """Login, list, upload and delete, with Utopia's sha256 dedup and tombstones.""" + state = {"docs": {}, "logins": 0, "tokens": set()} + + class Handler(BaseHTTPRequestHandler): + def _send(self, status, payload): + data = json.dumps(payload).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def _authed(self): + token = (self.headers.get("Authorization") or "").removeprefix("Bearer ") + if token in state["tokens"]: + return True + self._send(401, {"error": "unauthorized"}) + return False + + def _body(self): + return self.rfile.read(int(self.headers.get("Content-Length") or 0)) + + def do_POST(self): + path = urlparse(self.path).path + body = self._body() + if path == "/api/v1/auth/login": + state["logins"] += 1 + token = f"t{state['logins']}" + state["tokens"].add(token) + return self._send(200, {"token": token}) + if path != f"/api/v1/kbs/{KB}/documents": + return self._send(404, {}) + if not self._authed(): + return None + name = re.search(rb'filename="([^"]+)"', body).group(1).decode() + content = body.split(b"\r\n\r\n", 1)[1].rsplit(b"\r\n--", 1)[0] + digest = sha256(content).hexdigest() + if any(d["sha256"] == digest and not d["deleted"] for d in state["docs"].values()): + return self._send(200, {"created": [], "skipped": [{"filename": name}]}) + doc = {"id": str(uuidlib.uuid4()), "filename": name, "sha256": digest, + "content": content.decode(), "deleted": False} + state["docs"][doc["id"]] = doc + return self._send(200, {"created": [doc], "skipped": []}) + + def do_GET(self): + if not self._authed(): + return + q = parse_qs(urlparse(self.path).query).get("q", [""])[0] + docs = [d for d in state["docs"].values() if q in d["filename"] and not d["deleted"]] + self._send(200, {"docs": docs, "total": len(docs)}) + + def do_DELETE(self): + if not self._authed(): + return + state["docs"][urlparse(self.path).path.rsplit("/", 1)[1]]["deleted"] = True + self._send(200, {"ok": True}) + + def log_message(self, *_args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + Thread(target=server.serve_forever, daemon=True).start() + try: + yield f"http://127.0.0.1:{server.server_port}/api/v1", state + finally: + server.shutdown() + server.server_close() + + +@pytest.fixture +def env(): + utopia._tokens.clear() + with fake_utopia() as (base_url, state), patch("storage.utopia.VconRedis") as redis: + opts = {"base_url": base_url, "kb_id": KB, "email": "svc@example.com", + "password": "secret-password", "transient_retries": 0} + yield opts, state, redis.return_value.get_vcon + + +def live(state): + return [d for d in state["docs"].values() if not d["deleted"]] + + +def test_save_resave_and_delete(env): + opts, state, get_vcon = env + get_vcon.return_value = Vcon(current()) + + utopia.save(VCON_UUID, opts) + assert [d["filename"] for d in live(state)] == [f"{VCON_UUID}.md"] + assert live(state)[0]["content"].startswith("2026-09-18\n") + + utopia.save(VCON_UUID, opts) # identical content: Utopia skips, nothing tombstoned + assert len(state["docs"]) == 1 and len(live(state)) == 1 + + get_vcon.return_value = Vcon(current(text="An amended line.")) + utopia.save(VCON_UUID, opts) # new content supersedes the old rendering + assert len(state["docs"]) == 2 + assert len(live(state)) == 1 and "Speaker 0: An amended line." in live(state)[0]["content"] + + assert utopia.delete(VCON_UUID, opts) is True + assert live(state) == [] + assert utopia.delete(VCON_UUID, opts) is False + + +def test_expired_token_logs_in_again(env): + opts, state, get_vcon = env + get_vcon.return_value = Vcon(current()) + utopia.save(VCON_UUID, opts) + state["tokens"].clear() # Utopia restarted or the 7-day JWT ran out + get_vcon.return_value = Vcon(current(text="Later.")) + utopia.save(VCON_UUID, opts) + assert state["logins"] == 2 and len(live(state)) == 1 + + +def test_require_analysis_grant_skips_upload(env): + opts, state, get_vcon = env + get_vcon.return_value = Vcon(LEGACY) + utopia.save(VCON_UUID, {**opts, "require_analysis_grant": True}) + assert state["docs"] == {} and state["logins"] == 0 + + +def test_missing_options_fail_loudly(): + with pytest.raises(ValueError, match="kb_id, email, password"): + utopia.save(VCON_UUID, {})