From db4067127679d26f0f239a741e9962cd6b252336 Mon Sep 17 00:00:00 2001 From: Dani Date: Fri, 6 Nov 2026 18:18:02 -0500 Subject: [PATCH] feat(facade): purge, user_id owner guard, and error codes (1.3.0) - purge_request / purge_confirm / purge_get / purge_wait for the memory-space purge. Two calls stay two (no one-call purge), and the scope ids never come from the client's app_id / project_id defaults. purge_confirm takes the purge_request result and raises when the server answers with a different, already running purge (its nonce is spent and nothing of the caller's was queued). purge_wait shares the task_wait polling loop. - update / delete_by_ids take user_id: a memory owned by anyone else is a 404 / skipped. Documented to be set from the caller's session, never from model output. delete_by_ids returns deleted. - EverOSAPIError gains code / message read from the error body, so a caller can branch on e.g. purge_in_progress instead of a bare 409. - quickstart: a purge section, user_id usage, method table rows. Co-Authored-By: Claude Opus 5.5 --- everos_cloud/client.py | 241 ++++++++++++++++++++++++++++++++++++++--- quickstart.md | 66 ++++++++++- tests/test_client.py | 218 +++++++++++++++++++++++++++++++++++++ 3 files changed, 503 insertions(+), 22 deletions(-) diff --git a/everos_cloud/client.py b/everos_cloud/client.py index c90b98c..7aee1b2 100644 --- a/everos_cloud/client.py +++ b/everos_cloud/client.py @@ -18,14 +18,17 @@ style and each name is the last segment of its route — ``update`` / ``delete_by_ids`` / ``feedback`` (1.2.0), ``context`` (1.3.0). Every other resource carries its name as a prefix — ``kb_`` / ``doc_`` / ``task_`` / ``tag_`` — so completion groups by resource and no facade method -collides with the generated method it wraps. +collides with the generated method it wraps. The memory-space purge (1.3.0) is a memory +route but a multi-call flow, so it takes a prefix like ``task_``: ``purge_request`` / +``purge_confirm`` / ``purge_get`` / ``purge_wait``. Errors: every failure raised by this facade derives from :class:`EverOSError` — ``EverOSAPIError`` for HTTP errors, ``EverOSStorageError`` for object-upload failures, -and a plain ``EverOSError`` for a task that fails or outlives its wait timeout. +and a plain ``EverOSError`` for a task or purge that fails or outlives its wait timeout. """ from __future__ import annotations +import json import mimetypes import os import time @@ -55,6 +58,9 @@ KbCreateInput, KbPatchBody, MessageItem, + PurgeConfirmInput, + PurgeRequestData, + PurgeRequestInput, Reason, SearchBody, SearchInput, @@ -73,6 +79,7 @@ DEFAULT_TASK_TIMEOUT = 300.0 # seconds to wait for an async task before giving up DEFAULT_TASK_INTERVAL = 2.0 # seconds before the first task re-poll DEFAULT_TASK_MAX_INTERVAL = 15.0 # ceiling for the backoff between task polls +DEFAULT_PURGE_TIMEOUT = 1800.0 # seconds; a purge of a large memory space takes minutes # Statuses worth retrying DURING a poll rather than aborting the whole wait. A long # ingest polled at a fixed interval is itself a source of 429s, and a wait that dies @@ -98,6 +105,8 @@ "list_documents": "doc_list", "get_document": "doc_get", "update_document": "doc_update", "delete_document": "doc_delete", "get_task_status": "task_get", "list_tasks": "task_list", + "request_purge": "purge_request", "confirm_purge": "purge_confirm", + "get_purge_job": "purge_get", } _API_ATTRS = ("memory", "storage", "knowledge", "tasks") @@ -107,6 +116,8 @@ # and the contract documents `finished_at` as "absent while the task is not in a # terminal state" — so the timestamp, not the enum, is the load-bearing signal. _TERMINAL_TASK_STATUS = frozenset({"success", "failed"}) +# Same reasoning for purge jobs, whose terminal statuses are spelled differently. +_TERMINAL_PURGE_STATUS = frozenset({"succeeded", "failed"}) _OP_CLASSES = {"add": AddOperation, "update": UpdateOperation, "delete": DeleteOperation} @@ -154,17 +165,77 @@ def _task_finished(item: Any) -> bool: return getattr(item, "status", None) in _TERMINAL_TASK_STATUS +def _purge_finished(job: Any) -> bool: + """Has this purge job ended? Same rule as :func:`_task_finished`.""" + if getattr(job, "finished_at", None) is not None: + return True + return getattr(job, "status", None) in _TERMINAL_PURGE_STATUS + + +def _scope_text(scope: tuple) -> str: + """``("project", "a1", "p1")`` -> ``"project a1/p1"``, for messages.""" + kind, app_id, project_id = scope + ids = "/".join(x for x in (app_id, project_id) if x) + return f"{kind} {ids}" if ids else str(kind) + + +def _purge_failure(job: Any) -> str: + """The stores that failed, with their errors, for the exception message.""" + failed = [ + f"{getattr(s, 'store', '?')}: {getattr(s, 'error', None) or 'no reason given'}" + for s in (getattr(job, "steps", None) or []) + if getattr(s, "state", None) == "failed" + ] + return "; ".join(failed) or "no reason given" + + class EverOSError(Exception): """Base class for every error raised by the EverOS facade.""" +def _error_fields(body: Any) -> tuple[str | None, str | None]: + """``(code, message)`` from an error body, or ``(None, None)``. + + The API answers errors in a few shapes: the memory engine's envelope + ``{"request_id": ..., "error": {"code": ..., "message": ...}}``, a flat + ``{"code": ..., "message": ...}`` from the gateway's own checks, and + ``{"error": ""}`` from its quota check (message only). Anything else + (empty, not JSON, an HTML page from a proxy) yields ``None`` rather than raising: + this runs while an error is already being reported. + """ + if isinstance(body, (bytes, bytearray)): + body = body.decode("utf-8", "replace") + if isinstance(body, str): + try: + body = json.loads(body) + except ValueError: + return None, None + if not isinstance(body, Mapping): + return None, None + err = body.get("error") + if isinstance(err, str): + return None, err + src = err if isinstance(err, Mapping) else body + code, message = src.get("code"), src.get("message") + return (code if isinstance(code, str) else None, message if isinstance(message, str) else None) + + class EverOSAPIError(EverOSError): - """A memory endpoint returned an HTTP error (wraps the SDK's ``ApiException``).""" + """A memory endpoint returned an HTTP error (wraps the SDK's ``ApiException``). + + ``status`` is the HTTP status and ``body`` the raw response body. ``code`` and + ``message`` are read from the body when it carries them, so a caller can branch on + the error rather than the status alone — e.g. ``"purge_in_progress"`` (a write + during a purge: retry once it ends) versus another 409. Both are ``None`` when the + body does not say. + """ def __init__(self, status: Any, body: Any = None) -> None: - super().__init__(f"EverOS API error: status={status}") self.status = status self.body = body + self.code, self.message = _error_fields(body) + detail = f" code={self.code}" if self.code else "" + super().__init__(f"EverOS API error: status={status}{detail}" + (f": {self.message}" if self.message else "")) class EverOSStorageError(EverOSError): @@ -486,6 +557,7 @@ def update( summary: str | None = None, subject: str | None = None, reason: ReasonLike | None = None, + user_id: str | None = None, memory_type: str = "episode", ) -> Any: """Patch one episode's ``episode`` text, ``summary`` and/or ``subject``. @@ -494,6 +566,11 @@ def update( ``"wrong_subject"`` or ``{"code": ..., "note": ...}``. Returns ``UpdateData``, whose ``unchanged`` is true when the patch matched what was stored. + ``user_id`` limits the edit to that user's memory: a memory owned by anyone else + is a 404, exactly like a missing one. In an agent tool, set it from your own + session (the signed-in user), never from model output, so a wrong or invented + ``memory_id`` cannot reach another user's memory. + Profile items are edited with :meth:`edit`, not here. """ patch = _clean(episode=episode, summary=summary, subject=subject) @@ -505,6 +582,7 @@ def update( memory_id=memory_id, patch=EpisodePatch(**patch), reason=self._to_reason(reason), + user_id=user_id, ) ) return self._call(self.memory.update_memory, payload).data @@ -514,18 +592,24 @@ def delete_by_ids( memory_ids: Sequence[str], *, reason: ReasonLike | None = None, + user_id: str | None = None, memory_type: str = "episode", ) -> Any: """Soft-delete 1-50 memories by id. Returns ``DeleteByIdsData``. Ids that do not exist are skipped; one malformed id fails the whole batch (422). - To clear a whole user / agent / session, use :meth:`delete`. + ``deleted`` on the result counts what this call removed, so a 0 means nothing + was. ``user_id`` deletes only that user's memories: ids owned by anyone else are + skipped like missing ones. As with :meth:`update`, set it from your own session, + never from model output. To clear a whole user / agent / session, use + :meth:`delete`. """ payload = DeleteByIdsInput( **_clean( memory_type=memory_type, memory_ids=list(memory_ids), reason=self._to_reason(reason), + user_id=user_id, ) ) return self._call(self.memory.delete_memories_by_ids, payload).data @@ -772,35 +856,160 @@ def task_wait( A non-transient HTTP error propagates as ``EverOSAPIError``, as does a transient one that outlives the deadline. """ + item = self._poll( + lambda: self.task_get(task_id), + _task_finished, + what=f"task {task_id}", + attr="task", + timeout=timeout, + interval=interval, + max_interval=max_interval, + ) + if raise_on_failure and getattr(item, "status", None) == "failed": + reason = getattr(item, "error", None) or getattr(item, "error_code", None) or "no reason given" + err = EverOSError(f"task {task_id} failed: {reason}") + err.task = item + raise err + return item + + def _poll( + self, + fetch: Any, + finished: Any, + *, + what: str, + attr: str, + timeout: float, + interval: float, + max_interval: float, + ) -> Any: + """Call ``fetch`` until ``finished(item)``, with backoff; the loop behind every wait. + + Backs off from ``interval`` up to ``max_interval``. A transient failure (429 / + 5xx) is retried on the same schedule; any other HTTP error, or a transient one + past the deadline, propagates as ``EverOSAPIError``. Still unfinished at + ``timeout`` raises ``EverOSError`` with the last item seen on ``err.``. + """ deadline = time.monotonic() + timeout delay = interval item = None while True: try: - item = self.task_get(task_id) + item = fetch() except EverOSAPIError as exc: # Out of budget, or not the kind of failure that will pass: give up. if exc.status not in _TRANSIENT_STATUS or time.monotonic() >= deadline: raise else: - if _task_finished(item): - break + if finished(item): + return item if time.monotonic() >= deadline: err = EverOSError( - f"task {task_id} still {getattr(item, 'status', 'unknown')!r} " - f"after {timeout}s" + f"{what} still {getattr(item, 'status', 'unknown')!r} after {timeout}s" ) - err.task = item + setattr(err, attr, item) raise err time.sleep(delay) delay = min(delay * 2, max_interval) if delay else delay - if raise_on_failure and getattr(item, "status", None) == "failed": - reason = getattr(item, "error", None) or getattr(item, "error_code", None) or "no reason given" - err = EverOSError(f"task {task_id} failed: {reason}") - err.task = item + # -- memory-space purge -------------------------------------------------- + # Permanent: there is no undo. The API makes a purge two calls on purpose (a nonce + # from step 1, confirmed within 5 minutes) so that one mistaken call cannot wipe a + # memory space, and the facade keeps them two: there is deliberately no one-call + # purge. For the same reason the scope ids never come from the client's + # app_id / project_id defaults: what gets purged is always written at the call site. + def purge_request( + self, + scope: str, + *, + app_id: str | None = None, + project_id: str | None = None, + ) -> Any: + """Step 1 of 2: ask to purge. **Deletes nothing.** Returns ``PurgeRequestData``. + + ``scope`` is ``"space"`` (the whole memory space of this API key), ``"app"`` + (pass ``app_id``) or ``"project"`` (pass ``app_id`` and ``project_id``); each + takes exactly its own ids, and a missing or extra one is a 422. The result's + ``summary`` counts what would be deleted right now; check it, then pass the + result (or its ``nonce``) to :meth:`purge_confirm` before ``expires_at`` + (5 minutes). Counting can take up to about 30 seconds on a large memory space; + a count that did not finish is ``None`` and ``summary.complete`` is false, so + check ``complete`` before trusting the numbers. + """ + payload = PurgeRequestInput(**_clean(scope=scope, app_id=app_id, project_id=project_id)) + return self._call(self.memory.request_purge, payload).data + + def purge_confirm(self, request: PurgeRequestData | str) -> Any: + """Step 2 of 2: start the purge. **Permanent.** Returns the queued ``PurgeJobData``. + + Pass the result of :meth:`purge_request` (recommended) or just its ``nonce``. + The scope is the one the nonce was issued for. The nonce works once, only with + the same API key; an unknown, expired or used one is a 404 and nothing happens. + Until the job ends, every write to the memory space is a 409. Follow it with + :meth:`purge_wait` or :meth:`purge_get`. + + Only one purge runs per memory space. Confirming while one is queued or running + uses up the nonce and returns THAT job, whatever its scope, without queuing + yours. Given the request result, this method checks the returned job against + the scope you asked for and raises ``EverOSError`` (the running job on + ``.job``) when they differ: wait for it to end, then call :meth:`purge_request` + again. Given a bare nonce it cannot tell, so compare ``job.scope`` yourself. + """ + nonce = request if isinstance(request, str) else getattr(request, "nonce", None) + if not isinstance(nonce, str): + raise TypeError("purge_confirm takes the purge_request() result or its nonce string") + job = self._call(self.memory.confirm_purge, PurgeConfirmInput(nonce=nonce)).data + if not isinstance(request, str): + asked = (request.scope, getattr(request, "app_id", None), getattr(request, "project_id", None)) + got = (getattr(job, "scope", None), getattr(job, "app_id", None), getattr(job, "project_id", None)) + if got != asked: + err = EverOSError( + f"another purge is already {getattr(job, 'status', 'running')!r} in this " + f"memory space (job {getattr(job, 'job_id', '?')}, scope {_scope_text(got)}); " + f"yours ({_scope_text(asked)}) was not queued and its nonce is used up. " + "Wait for that job to end, then call purge_request again." + ) + err.job = job + raise err + return job + + def purge_get(self, job_id: str) -> Any: + """Read a purge job. Returns ``PurgeJobData`` (``status``, per-store ``steps``).""" + return self._call(self.memory.get_purge_job, job_id).data + + def purge_wait( + self, + job_id: str, + *, + timeout: float = DEFAULT_PURGE_TIMEOUT, + interval: float = DEFAULT_TASK_INTERVAL, + max_interval: float = DEFAULT_TASK_MAX_INTERVAL, + raise_on_failure: bool = True, + ) -> Any: + """Poll a purge job until it ends and return the terminal ``PurgeJobData``. + + Same polling as :meth:`task_wait`. Raises ``EverOSError`` naming the failed + stores if the job fails (unless ``raise_on_failure=False``, which returns it), or + if it is still running at ``timeout``; the last job seen is on the error's + ``.job``. Timing out only ends the wait, not the purge, which keeps running on + the server (it fails on its own only after repeated retries, within about two + hours): call ``purge_wait`` again to keep waiting. A purge that failed partway + stays partly done: request a new one. + """ + job = self._poll( + lambda: self.purge_get(job_id), + _purge_finished, + what=f"purge {job_id}", + attr="job", + timeout=timeout, + interval=interval, + max_interval=max_interval, + ) + if raise_on_failure and getattr(job, "status", None) == "failed": + err = EverOSError(f"purge {job_id} failed: {_purge_failure(job)}") + err.job = job raise err - return item + return job # -- storage ------------------------------------------------------------- def presign(self, objects: Sequence[Any]) -> Any: diff --git a/quickstart.md b/quickstart.md index f14b05c..52b3183 100644 --- a/quickstart.md +++ b/quickstart.md @@ -71,6 +71,15 @@ client.update("mem-1", summary="Moved to Hangzhou; lives near West Lake.", reason="wrong_subject") client.delete_by_ids(["mem-2", "mem-3"], reason="redundant") # 1-50 ids, soft delete +# In an agent tool acting for one user, pass user_id (1.3.0) from YOUR session, never +# from model output: a memory owned by anyone else is then a 404 on update and is +# skipped on delete_by_ids, so a wrong or invented id cannot reach another user's +# memory. `deleted` counts what was actually removed. +client.update("mem-1", summary="...", user_id=current_user_id) +result = client.delete_by_ids(["mem-2"], user_id=current_user_id) +if result.deleted == 0: + ... # nothing of this user's was removed + # ── Rate a memory ───────────────────────────────────────────────────────────── # Record-only: nothing is edited. reason / note / suggestion go with "negative". client.feedback("mem-1", "negative", reason="outdated", @@ -144,6 +153,46 @@ if task.status == "failed": print(task.error) ``` +## Purge a memory space + +**Permanent — there is no undo.** A purge deletes all memory data in the memory space, +one app, or one project, so you can reuse it from empty. It takes two calls on purpose, +and the SDK keeps them two: there is no one-call purge. The scope is only ever what you +pass, never the client's `app_id` / `project_id` defaults. + +```python +# 1. Request: deletes nothing. Returns a nonce and a count of what would be deleted. +req = client.purge_request("project", app_id="my_app", project_id="staging") +print(req.summary.mongo_total, req.summary.complete) # check before you confirm + +# 2. Confirm within 5 minutes: starts the purge. Pass `req`, not just `req.nonce`: +# if another purge is already running, the server returns THAT job and your nonce is +# used up; given `req`, the SDK notices the different scope and raises EverOSError. +job = client.purge_confirm(req) + +# 3. Wait for it. Writes to the memory space are rejected (409) until it ends. +job = client.purge_wait(job.job_id) # raises if it fails or times out +``` + +`scope` is `"space"` (no ids), `"app"` (`app_id`) or `"project"` (`app_id` + +`project_id`); a missing or extra id is a 422. Files uploaded through `upload` are not +deleted. See https://docs.evermind.ai/cloud/purge for what else stays behind. + +While a purge runs, writes fail with HTTP 409. Branch on the error's `code` rather +than the status alone: + +```python +from everos_cloud import EverOSAPIError + +try: + client.add(session_id="s1", messages=[...]) +except EverOSAPIError as e: + if e.code == "purge_in_progress": + ... # retry after the purge ends + else: + raise +``` + ## Memory tags Tags are scoped by the memory ids themselves — no `app_id` / `project_id`. @@ -155,9 +204,10 @@ client.tag_replace("episode", ["mem-2"], ["archived"]) # overwrite ``` Every method returns the endpoint's `.data`. Failures raise `EverOSError` -(`EverOSAPIError` for HTTP errors, `EverOSStorageError` for uploads, and a plain -`EverOSError` for a task that fails or outlives its `task_wait` timeout — the last -task seen is on the error's `.task`). Set a +(`EverOSAPIError` for HTTP errors, with `.status`, and `.code` / `.message` when the +response names them, `EverOSStorageError` for uploads, and a plain +`EverOSError` for a task or purge that fails or outlives its `task_wait` / +`purge_wait` timeout — the last task or job seen is on the error's `.task` / `.job`). Set a per-client request timeout with `EverOS(api_key=..., timeout=30)`, or use it as a context manager (`with EverOS(...) as client:`) to release connections on exit. @@ -167,7 +217,8 @@ Memory methods are bare verbs named after the last segment of their route (`add` `search` / `get` / `flush` / `edit` / `delete` since 1.0.0, `update` / `delete_by_ids` / `feedback` since 1.2.0, `context` since 1.3.0), as are `presign` / `upload` for storage and `close`. Every other resource is named `_`, so typing `client.kb` / `client.doc` / -`client.task` / `client.tag` lists everything for that resource. +`client.task` / `client.tag` / `client.purge` lists everything for that resource. The +purge is a memory route but a multi-call flow, so it is `purge_` like `task_`. | Method | Endpoint | Notes | |---|---|---| @@ -178,8 +229,8 @@ resource is named `_`, so typing `client.kb` / `client.doc` / | `context(query, user_id, ...)` | `POST /api/v2/memory/context` | Multi-round LLM retrieval for multi-hop / temporal questions; user owners only. | | `edit(user_id, operations)` | `POST /api/v2/memory/edit` | Bulk profile add / update / delete. | | `delete(...)` | `POST /api/v2/memory/delete` | Scoped soft-delete (whole user / agent / session). | -| `update(memory_id, ...)` | `POST /api/v2/memory/update` | Patch one episode's text / summary / subject; `unchanged` in the result when nothing differed. | -| `delete_by_ids(memory_ids, ...)` | `POST /api/v2/memory/delete_by_ids` | Soft-delete 1-50 memories by id; unknown ids are skipped. | +| `update(memory_id, ...)` | `POST /api/v2/memory/update` | Patch one episode's text / summary / subject; `unchanged` in the result when nothing differed. `user_id=` limits it to that user's memory. | +| `delete_by_ids(memory_ids, ...)` | `POST /api/v2/memory/delete_by_ids` | Soft-delete 1-50 memories by id; unknown ids are skipped and `deleted` counts the rest. `user_id=` limits it to that user's memories. | | `feedback(memory_id, rating, ...)` | `POST /api/v2/memory/feedback` | Record a positive / negative rating; edits nothing. | | `upload(path)` | `POST /api/v2/object/sign` + S3 | Presign + direct-to-S3, returns `object_key`. | | `tag_bind` / `tag_unbind` / `tag_replace` | `POST /api/v2/memory/tag/*` | Add / remove / overwrite tags on memory ids. | @@ -193,6 +244,9 @@ resource is named `_`, so typing `client.kb` / `client.doc` / | `task_get(task_id)` | `GET /api/v2/tasks/{task_id}` | One task's status. | | `task_list(...)` | `GET /api/v2/tasks` | Filter by status / session / time window. | | `task_wait(task_id, ...)` | polls `GET /api/v2/tasks/{task_id}` | Blocks until terminal, backing off between polls and riding out transient 429/5xx; raises on failure/timeout. | +| `purge_request(scope, ...)` | `POST /api/v2/memory/purge` | Step 1 of a purge: returns a `nonce` and a count; deletes nothing. | +| `purge_confirm(request)` | `POST /api/v2/memory/purge/confirm` | Step 2: starts the permanent purge (202 job). Pass the `purge_request` result so a different running purge is caught. | +| `purge_get(job_id)` / `purge_wait(job_id, ...)` | `GET /api/v2/memory/purge/{job_id}` | Read the job, or poll it until it ends (same polling as `task_wait`). | ## Low-level typed client (advanced) diff --git a/tests/test_client.py b/tests/test_client.py index 6555314..3c6ad8e 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -16,6 +16,8 @@ KbCreateInput, KbPatchBody, MessageItem, + PurgeConfirmInput, + PurgeRequestInput, Reason, SearchBody, SearchInput, @@ -858,3 +860,219 @@ def test_getattr_hook_does_not_break_normal_lookup(): with pytest.raises(AttributeError) as ei: getattr(c, dunder) assert "generated method name" not in str(ei.value) + + +# ── owner guard on update / delete_by_ids (1.3.0) ──────────────────────────── +def test_update_and_delete_by_ids_send_user_id_only_when_given(): + mem = MagicMock() + mem.update_memory.return_value = SimpleNamespace(data="UPDATED") + mem.delete_memories_by_ids.return_value = SimpleNamespace(data="DELETED") + c = _client(memory=mem) + + c.update("m1", subject="t", user_id="u1") + c.delete_by_ids(["m1"], user_id="u1") + assert mem.update_memory.call_args.args[0].user_id == "u1" + assert mem.delete_memories_by_ids.call_args.args[0].user_id == "u1" + + c.update("m1", subject="t") + c.delete_by_ids(["m1"]) + assert "user_id" not in mem.update_memory.call_args.args[0].to_dict() + assert "user_id" not in mem.delete_memories_by_ids.call_args.args[0].to_dict() + + +def test_user_id_guard_never_comes_from_the_client_scope(): + """The guard names a user, not a scope: no client default may fill it in.""" + mem = MagicMock() + c = EverOS("sk-test", app_id="my-app", project_id="p1") + c.memory = mem + c.update("m1", subject="t") + assert mem.update_memory.call_args.args[0].user_id is None + + +# ── memory-space purge (1.3.0) ─────────────────────────────────────────────── +def _job(status, finished_at=None, steps=None): + return SimpleNamespace(status=status, finished_at=finished_at, steps=steps or []) + + +def test_purge_request_sends_exactly_the_scope_it_is_given(): + mem = MagicMock() + mem.request_purge.return_value = SimpleNamespace(data="NONCE") + c = _client(memory=mem) + + assert c.purge_request("project", app_id="a1", project_id="p1") == "NONCE" + payload = mem.request_purge.call_args.args[0] + assert isinstance(payload, PurgeRequestInput) + assert (payload.scope, payload.app_id, payload.project_id) == ("project", "a1", "p1") + + +def test_purge_scope_never_inherits_the_client_defaults(): + """A client built for one app must not turn a space purge into an app purge, or an + app purge into a different app's: the target is only ever what the call names.""" + mem = MagicMock() + c = EverOS("sk-test", app_id="my-app", project_id="p1") + c.memory = mem + + c.purge_request("space") + sent = mem.request_purge.call_args.args[0].to_dict() + assert sent == {"scope": "space"} + + +def test_there_is_no_one_call_purge(): + """The two-step nonce is the safety mechanism; the facade must not collapse it.""" + assert not hasattr(EverOS, "purge") + assert not hasattr(EverOS, "purge_space") + + +def test_purge_confirm_and_get(): + mem = MagicMock() + mem.confirm_purge.return_value = SimpleNamespace(data="JOB") + mem.get_purge_job.return_value = SimpleNamespace(data="STATUS") + c = _client(memory=mem) + + nonce = "3f9c2a7d8e1b4c6a9f0e2d4b6a8c0e1f" + assert c.purge_confirm(nonce) == "JOB" + payload = mem.confirm_purge.call_args.args[0] + assert isinstance(payload, PurgeConfirmInput) and payload.nonce == nonce + + assert c.purge_get("job-1") == "STATUS" + assert mem.get_purge_job.call_args.args[0] == "job-1" + + +def test_purge_wait_polls_until_succeeded(): + mem = MagicMock() + mem.get_purge_job.side_effect = [ + SimpleNamespace(data=_job("queued")), + SimpleNamespace(data=_job("running")), + ApiException(status=503, reason="unavailable"), + SimpleNamespace(data=_job("succeeded", finished_at="2026-11-06T00:00:00Z")), + ] + c = _client(memory=mem) + + assert c.purge_wait("job-1", interval=0).status == "succeeded" + assert mem.get_purge_job.call_count == 4 + + +def test_purge_wait_raises_naming_the_failed_stores(): + mem = MagicMock() + failed = _job( + "failed", + finished_at="2026-11-06T00:00:00Z", + steps=[ + SimpleNamespace(store="mongo:v2_episodes", state="done", error=None), + SimpleNamespace(store="es:episode", state="failed", error="timed out"), + ], + ) + mem.get_purge_job.return_value = SimpleNamespace(data=failed) + c = _client(memory=mem) + + with pytest.raises(EverOSError) as ei: + c.purge_wait("job-1", interval=0) + assert "es:episode: timed out" in str(ei.value) + assert "mongo:v2_episodes" not in str(ei.value) + assert ei.value.job is failed + + assert c.purge_wait("job-1", interval=0, raise_on_failure=False) is failed + + +def test_purge_wait_times_out_with_the_last_job(): + mem = MagicMock() + mem.get_purge_job.return_value = SimpleNamespace(data=_job("running")) + c = _client(memory=mem) + + with pytest.raises(EverOSError) as ei: + c.purge_wait("job-1", timeout=0, interval=0) + assert "running" in str(ei.value) + assert ei.value.job.status == "running" + + +def test_purge_wait_stops_on_an_unknown_terminal_status(): + mem = MagicMock() + mem.get_purge_job.return_value = SimpleNamespace( + data=_job("cancelled", finished_at="2026-11-06T00:00:00Z") + ) + c = _client(memory=mem) + assert c.purge_wait("job-1", interval=0).status == "cancelled" + assert mem.get_purge_job.call_count == 1 + + +def test_generated_purge_names_point_at_the_facade(): + c = EverOS("sk-test") + with pytest.raises(AttributeError) as ei: + c.request_purge + assert "client.purge_request(" in str(ei.value) + + +def _request(scope, app_id=None, project_id=None): + return SimpleNamespace(nonce="3f9c2a7d8e1b4c6a9f0e2d4b6a8c0e1f", scope=scope, + app_id=app_id, project_id=project_id) + + +def test_purge_confirm_takes_the_request_and_returns_a_matching_job(): + mem = MagicMock() + job = SimpleNamespace(job_id="j1", status="queued", scope="project", app_id="a1", project_id="p1") + mem.confirm_purge.return_value = SimpleNamespace(data=job) + c = _client(memory=mem) + + assert c.purge_confirm(_request("project", "a1", "p1")) is job + assert mem.confirm_purge.call_args.args[0].nonce == "3f9c2a7d8e1b4c6a9f0e2d4b6a8c0e1f" + + +def test_purge_confirm_raises_when_another_purge_answered(): + """While a purge runs, confirm spends the nonce and returns THAT job, whatever its + scope. Handed the request, the facade must not let it pass as the caller's purge.""" + mem = MagicMock() + other = SimpleNamespace(job_id="j-other", status="running", scope="app", app_id="a9", project_id=None) + mem.confirm_purge.return_value = SimpleNamespace(data=other) + c = _client(memory=mem) + + with pytest.raises(EverOSError) as ei: + c.purge_confirm(_request("project", "a1", "p1")) + msg = str(ei.value) + assert "j-other" in msg and "app a9" in msg and "project a1/p1" in msg + assert "purge_request again" in msg + assert ei.value.job is other + + +def test_purge_confirm_with_a_bare_nonce_cannot_check(): + mem = MagicMock() + other = SimpleNamespace(job_id="j-other", status="running", scope="space", app_id=None, project_id=None) + mem.confirm_purge.return_value = SimpleNamespace(data=other) + c = _client(memory=mem) + assert c.purge_confirm("3f9c2a7d8e1b4c6a9f0e2d4b6a8c0e1f") is other + + +def test_purge_confirm_rejects_anything_without_a_nonce(): + c = _client(memory=MagicMock()) + with pytest.raises(TypeError): + c.purge_confirm(SimpleNamespace(scope="space")) + + +@pytest.mark.parametrize( + ("body", "code", "message"), + [ + ('{"request_id": "r", "error": {"code": "purge_in_progress", "message": "busy"}}', + "purge_in_progress", "busy"), + (b'{"code": "InvalidParameter", "message": "bad timestamp"}', "InvalidParameter", "bad timestamp"), + ({"error": {"code": "not_found"}}, "not_found", None), + ("bad gateway", None, None), + (None, None, None), + ('{"error": "route quota exceeded"}', None, "route quota exceeded"), + ('{"error": 42}', None, None), + ], +) +def test_api_errors_expose_the_body_code_and_message(body, code, message): + err = EverOSAPIError(409, body) + assert (err.status, err.code, err.message) == (409, code, message) + assert "status=409" in str(err) + if code: + assert code in str(err) + + +def test_a_write_during_a_purge_surfaces_its_code(): + mem = MagicMock() + mem.flush_memory.side_effect = ApiException(status=409, reason="Conflict") + mem.flush_memory.side_effect.body = '{"error": {"code": "purge_in_progress", "message": "purge running"}}' + c = _client(memory=mem) + with pytest.raises(EverOSAPIError) as ei: + c.flush("s1") + assert ei.value.code == "purge_in_progress"