Coverage for astrocyte/

1"""Team-memory sync helpers (``team-memory.md`` §8): the change-feed cursor (G3) and push limits (G2).

3The cursor is opaque to clients: URL-safe base64 of the JSON object

4``{"changed_at": <ISO 8601>, "id": <memory id>}`` — the ``(changed_at, id)``

5position of the last change a client has seen. Stores return only entries

6strictly after it, ordered by ``(changed_at, id)``.

7"""

9from __future__ import annotations

11import base64

12import binascii

13import hashlib

14import json

15import re

16from datetime import datetime, timezone

18from astrocyte.errors import InvalidCursor

20#: Most changes one ``list_changes`` call returns.

21MAX_CHANGES_LIMIT = 1000

24def encode_cursor(changed_at: datetime, memory_id: str) -> str:

25 if changed_at.tzinfo is None:

26 changed_at = changed_at.replace(tzinfo=timezone.utc)

27 payload = json.dumps(

28 {"changed_at": changed_at.astimezone(timezone.utc).isoformat(), "id": memory_id},

29 separators=(",", ":"),

30 )

31 return base64.urlsafe_b64encode(payload.encode("utf-8")).decode("ascii").rstrip("=")

34def decode_cursor(cursor: str) -> tuple[datetime, str]:

35 """``(changed_at, id)`` from a cursor; :class:`InvalidCursor` if it isn't one."""

36 try:

37 raw = base64.urlsafe_b64decode(cursor + "=" * (-len(cursor) % 4))

38 data = json.loads(raw.decode("utf-8"))

39 changed_at = datetime.fromisoformat(data["changed_at"])

40 memory_id = data["id"]

41 except (binascii.Error, UnicodeError, ValueError, TypeError, KeyError) as exc:

42 raise InvalidCursor() from exc

43 if not isinstance(memory_id, str) or not memory_id or changed_at.tzinfo is None:

44 raise InvalidCursor()

45 return changed_at, memory_id

48# ── push (G2) ─────────────────────────────────────────────────────────────

50#: Most records one ``push_records`` call accepts.

51MAX_PUSH_RECORDS = 100

53#: A pushed id: 8–64 of ``[A-Za-z0-9_-]``. Local stores mint 16 hex

54#: characters (``uuid4().hex[:16]``); the wider set leaves room for other

55#: clients' schemes while keeping ids safe in URLs, logs and SQL.

56PUSH_ID_RE = re.compile(r"[A-Za-z0-9_-]{8,64}")

58#: ``sha256:`` and 64 lowercase hex digits.

59CONTENT_HASH_RE = re.compile(r"sha256:[0-9a-f]{64}")

61#: Underscore-prefixed metadata keys are the system's own. A push keeps only

62#: these from the client: the reader regroups a turn's chunks by

63#: ``_retain_id`` / ``_chunk_index`` and dates them by ``_created_at``.

64#: ``_actor`` is stamped from the authenticated caller (a client's is kept

65#: only without a context, as retain does).

66PUSH_SYSTEM_METADATA_KEYS = frozenset({"_created_at", "_retain_id", "_chunk_index", "_actor"})

69def content_hash(text: str) -> str:

70 return "sha256:" + hashlib.sha256(text.encode("utf-8")).hexdigest()

Coverage for astrocyte/_sync.py: 100%

31 statements