Sprint 01: SurrealDB spike — parity adapters, migrations, bench (ADR-01)

services/api/db/migrations/0001_core.surql

@@ -1,4 +1,6 @@
--- 0001_core.surql — verbatim from SDAI-CHAT-003 Block 5 (spec is SSOT)
+-- 0001_core.surql — base is verbatim from SDAI-CHAT-003 Block 5 (spec is SSOT);
+-- Sprint-01 sanctioned changes: additions appended at the bottom, and the
+-- vector index is HNSW (spec said MTREE, which SurrealDB 3.x removed).
 
 DEFINE TABLE message SCHEMAFULL;
 DEFINE FIELD conversation ON message TYPE record<conversation>;
@@ -8,7 +10,7 @@ DEFINE FIELD text    ON message TYPE string ASSERT string::len($value) <= 262144
 DEFINE FIELD turn    ON message TYPE int READONLY;              -- server-stamped, monotonic per conversation
 DEFINE FIELD created_at ON message TYPE datetime VALUE time::now() READONLY;
 DEFINE FIELD embedding  ON message TYPE option<array<float>>;
-DEFINE INDEX msg_embedding ON message FIELDS embedding MTREE DIMENSION 768; -- vector index
+DEFINE INDEX msg_embedding ON message FIELDS embedding HNSW DIMENSION 768; -- vector index (embeddings are L2-normalized)
 -- immutability: no UPDATE permission is ever defined for message content
 
 DEFINE TABLE thread SCHEMAFULL;
@@ -37,3 +39,21 @@ DEFINE TABLE view_block SCHEMAFULL;   -- elements reference atom ids; service ga
 DEFINE TABLE followup   SCHEMAFULL;
 DEFINE TABLE intent_log SCHEMAFULL;   -- the outbox (ADR-03)
 DEFINE TABLE decision_log SCHEMAFULL; -- record-once/replay-many cognition log
+
+-- ---------------------------------------------------------------------------
+-- Sprint-01 sanctioned additions (brief Task 3: minimal fixes, noted in PR).
+-- Spec sync: fold into SDAI-CHAT-003 Block 5 at the 0.1.2 version bump.
+-- ---------------------------------------------------------------------------
+
+-- referenced above as record<conversation> but never defined:
+DEFINE TABLE conversation SCHEMALESS;
+
+-- per-conversation atomic turn counter (implementation detail of append):
+DEFINE TABLE conv_counter SCHEMAFULL;
+DEFINE FIELD n ON conv_counter TYPE int DEFAULT 0;
+
+-- intent_log was SCHEMAFULL with no fields — SCHEMAFULL strips undefined
+-- fields on write, which would silently break the outbox:
+DEFINE FIELD step      ON intent_log TYPE string;
+DEFINE FIELD input_ref ON intent_log TYPE string;
+DEFINE FIELD status    ON intent_log TYPE string ASSERT $value IN ["pending", "claimed", "done"];

services/api/db/surreal.py

@@ -1,11 +1,142 @@
 """SurrealDB engine/session — THE ONLY SurrealQL in the codebase (003 Principle 4).
 
-A static CI check (`test_no_surrealql_outside_db_dir`) keeps this honest.
-Local-first: single node, file-backed, bound to 127.0.0.1.
+All SurrealQL string constants used by repos/surreal_impl.py are defined here;
+surreal_impl.py imports them and never defines query strings.
 
-Open question (003 Block 9): embedded (in-process) vs server mode.
+Targets SurrealDB 3.x (type::record, HNSW). See the 2.x-to-3.x migration
+guide for the renames if this ever runs against an older server.
+
+SDK quirk (verified in CI): query() returns None for any BEGIN/COMMIT batch,
+even with an explicit RETURN. Transactions here are therefore write-only;
+callers read results back with single-statement queries afterwards.
+
+Connect/migrate pattern:
+    client = await connect()
+    applied = await apply_migrations(client)
+"""
+
+import os
+import pathlib
+from typing import Any
+
+from surrealdb import AsyncSurreal  # type: ignore[import-untyped]
+
+_MIGRATIONS_DIR = pathlib.Path(__file__).parent / "migrations"
+
+_MIGRATION_INIT_SQL = """
+DEFINE TABLE IF NOT EXISTS migration SCHEMAFULL;
+DEFINE FIELD IF NOT EXISTS filename ON migration TYPE string;
+DEFINE INDEX IF NOT EXISTS migration_filename ON migration FIELDS filename UNIQUE;
+"""
+
+
+async def connect() -> AsyncSurreal:
+    """Connect, sign in, and select namespace/database from env vars.
+
+    Defaults (Sprint-01 brief): SURREAL_URL=ws://127.0.0.1:8000/rpc,
+    SURREAL_NS=senja, SURREAL_DB=dev, SURREAL_USER/PASS=root/root.
+    """
+    url = os.environ.get("SURREAL_URL", "ws://127.0.0.1:8000/rpc")
+    ns = os.environ.get("SURREAL_NS", "senja")
+    db = os.environ.get("SURREAL_DB", "dev")
+    user = os.environ.get("SURREAL_USER", "root")
+    password = os.environ.get("SURREAL_PASS", "root")
+
+    client = AsyncSurreal(url)
+    await client.connect()
+    await client.signin({"username": user, "password": password})
+    await client.use(ns, db)
+    return client
+
+
+async def apply_migrations(client: AsyncSurreal) -> list[str]:
+    """Execute *.surql files in filename order, idempotently.
+
+    Maintains a `migration` table recording applied filenames; skips files
+    already applied. Returns the filenames applied in this call.
+    """
+    await client.query(_MIGRATION_INIT_SQL)
+
+    applied: list[str] = []
+    for path in sorted(_MIGRATIONS_DIR.glob("*.surql")):
+        filename = path.name
+        existing: Any = await client.query(
+            "SELECT filename FROM migration WHERE filename = $fn LIMIT 1",
+            {"fn": filename},
+        )
+        rows: list[Any] = existing if isinstance(existing, list) else []
+        if rows:
+            continue  # already applied
+
+        sql = path.read_text(encoding="utf-8")
+        # The Python SDK rejects empty queries — skip comment-only files.
+        has_statements = any(
+            line.strip() and not line.strip().startswith("--")
+            for line in sql.splitlines()
+        )
+        if has_statements:
+            await client.query(sql)
+
+        await client.create("migration", {"filename": filename})
+        applied.append(filename)
+
+    return applied
+
+
+# ---------------------------------------------------------------------------
+# SurrealQL constants — imported by repos/surreal_impl.py
+# ---------------------------------------------------------------------------
+
+# Append is ONE write-only transaction: ensure conversation, allocate the turn
+# atomically (conv_counter.n has DEFAULT 0 in 0001_core.surql), create the
+# message under the caller-supplied ULID id. A crash can never leave an
+# allocated turn without its message (no gaps). turn and created_at are
+# server-stamped; the caller reads the message back with SQL_GET_MESSAGE.
+SQL_APPEND_MESSAGE = """
+BEGIN TRANSACTION;
+UPSERT type::record('conversation', $conv_id);
+LET $c = (UPSERT type::record('conv_counter', $conv_id) SET n += 1 RETURN AFTER);
+CREATE type::record('message', $msg_id) CONTENT {
+    conversation: type::record('conversation', $conv_id),
+    author: $author,
+    lane: $lane,
+    text: $text,
+    turn: $c[0].n
+};
+COMMIT TRANSACTION;
 """
 
+# Single-statement read-back by known id (single statements return rows).
+SQL_GET_MESSAGE = """
+SELECT * FROM type::record('message', $msg_id);
+"""
 
-class SurrealSession:  # TODO(003-slice): real session + transaction manager
-    ...
+SQL_SNAPSHOT = """
+SELECT * FROM message
+WHERE conversation = type::record('conversation', $conv_id)
+ORDER BY turn ASC;
+"""
+
+# Intent-log: deterministic record id = step|input_ref makes enqueue idempotent.
+SQL_ENQUEUE = """
+INSERT IGNORE INTO intent_log (id, step, input_ref, status) VALUES (
+    type::record('intent_log', string::concat($step, '|', $input_ref)),
+    $step,
+    $input_ref,
+    'pending'
+);
+"""
+
+# Claim: single-statement conditional update — atomic; RETURN AFTER yields the
+# record only for the caller whose WHERE matched (the claim winner).
+SQL_CLAIM = """
+UPDATE type::record('intent_log', string::concat($step, '|', $input_ref))
+SET status = 'claimed'
+WHERE status = 'pending'
+RETURN AFTER;
+"""
+
+SQL_COMPLETE = """
+UPDATE type::record('intent_log', string::concat($step, '|', $input_ref))
+SET status = 'done';
+"""

services/api/repos/surreal_impl.py

@@ -1,9 +1,183 @@
 """SurrealDB implementation of the repository ports (003 ADR-01).
 
-Transplant note (docs/HARVEST.md): the parity contract-test pattern from the
-quarry proves this implementation against in-memory reference stores — port
-those tests before implementing, then make them green.
+Structurally satisfies MessageStore and IntentLog from repos/ports.py.
+All SurrealQL lives in services/api/db/surreal.py — no query strings here
+(AGENTS.md rule 4). The parity contract tests in tests/contracts/ must pass
+identically against this adapter and the in-memory reference.
 """
 
-# TODO(003-slice): implement MessageStore, ThreadIndexStore, AtomStore,
-# IntentLog, DecisionLog against services/api/db/surreal.py
+from __future__ import annotations
+
+import asyncio
+import os
+import time
+from datetime import datetime, timezone
+from typing import Any
+
+from surrealdb import AsyncSurreal  # type: ignore[import-untyped]
+
+from db import surreal as _sql
+from models import MAX_MESSAGE_BYTES, Message, MessageTooLarge, Snapshot
+
+_TXN_RETRIES = 50
+_TXN_BACKOFF_S = 0.004  # linear: 4ms, 8ms, ... — counter contention clears fast
+
+_ULID_ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"  # Crockford base32
+
+
+def _new_ulid() -> str:
+    """Dependency-free ULID (ADR-04): 48-bit ms timestamp + 80 bits randomness,
+    26 chars Crockford base32 — lexicographically sortable by creation time."""
+    value = (int(time.time() * 1000) << 80) | int.from_bytes(os.urandom(10), "big")
+    return "".join(
+        _ULID_ALPHABET[(value >> (5 * shift)) & 31] for shift in range(25, -1, -1)
+    )
+
+
+# ---------------------------------------------------------------------------
+# Result-shape helpers (SDK returns vary by statement kind and version)
+# ---------------------------------------------------------------------------
+
+
+def _normalise_rows(result: Any) -> list[dict[str, Any]]:
+    if result is None:
+        return []
+    if isinstance(result, dict):
+        return [result]
+    if isinstance(result, (list, tuple)):
+        out: list[dict[str, Any]] = []
+        for item in result:
+            out.extend(_normalise_rows(item))
+        return out
+    return []
+
+
+def _record_key(raw_id: Any) -> str:
+    """'message:⟨01H…⟩' / RecordID -> bare key string."""
+    s = str(raw_id)
+    if ":" in s:
+        s = s.split(":", 1)[1]
+    return s.strip("⟨⟩`")
+
+
+def _parse_datetime(value: Any) -> datetime:
+    if isinstance(value, datetime):
+        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
+    if isinstance(value, str):
+        return datetime.fromisoformat(value.replace("Z", "+00:00"))
+    raise TypeError(f"unexpected created_at shape: {value!r}")
+
+
+def _record_to_message(conv_id: str, row: dict[str, Any]) -> Message:
+    return Message(
+        id=_record_key(row["id"]),
+        conversation_id=conv_id,
+        author=str(row["author"]),
+        lane=str(row["lane"]),
+        text=str(row["text"]),
+        turn=int(row["turn"]),
+        created_at=_parse_datetime(row["created_at"]),
+    )
+
+
+def _is_txn_conflict(exc: Exception) -> bool:
+    msg = str(exc).lower()
+    return "retr" in msg or "conflict" in msg  # 'retried'/'retry'/'conflict'
+
+
+# ---------------------------------------------------------------------------
+# SurrealMessageStore
+# ---------------------------------------------------------------------------
+
+
+class SurrealMessageStore:
+    """Append-only message store. Turn allocation and message creation happen
+    in ONE database transaction (SQL_APPEND_MESSAGE) — monotonic, gapless,
+    duplicate-free under concurrency, and crash-safe (a crash can never leave
+    an allocated turn without its message).
+
+    SDK reality (verified in CI): transaction batches always return None,
+    and a conflicted transaction aborts WITHOUT raising. The adapter
+    therefore generates the ULID id itself (ruling #5), commits the
+    write-only transaction, and reads the message back by known id — the
+    read-back doubles as the success check. Absent message = conflict
+    abort = retry the whole transaction.
+
+    Safety: an aborted transaction creates nothing and rolls back the
+    counter (no turn gaps); a duplicate commit would fail loudly on the
+    existing ULID rather than double-append.
+
+    turn (counter) and created_at (schema VALUE) are server-stamped.
+    No update or delete methods exist — append-only by construction (Principle 1).
+    """
+
+    def __init__(self, client: AsyncSurreal) -> None:
+        self._client = client
+
+    async def append(
+        self, conversation_id: str, text: str, lane: str, author: str
+    ) -> Message:
+        if len(text.encode("utf-8")) > MAX_MESSAGE_BYTES:
+            raise MessageTooLarge(f"message exceeds {MAX_MESSAGE_BYTES} bytes")
+
+        msg_id = _new_ulid()
+        params = {
+            "conv_id": conversation_id,
+            "msg_id": msg_id,
+            "author": author,
+            "lane": lane,
+            "text": text,
+        }
+        for attempt in range(_TXN_RETRIES):
+            try:
+                await self._client.query(_sql.SQL_APPEND_MESSAGE, params)
+            except Exception as exc:  # loud conflict — retry; anything else raises
+                if _is_txn_conflict(exc):
+                    await asyncio.sleep(_TXN_BACKOFF_S * (attempt + 1))
+                    continue
+                raise
+
+            # Read-back by known id is the success check: conflicted
+            # transactions abort silently (SDK returns None either way).
+            rows = _normalise_rows(
+                await self._client.query(_sql.SQL_GET_MESSAGE, {"msg_id": msg_id})
+            )
+            if rows:
+                return _record_to_message(conversation_id, rows[0])
+            await asyncio.sleep(_TXN_BACKOFF_S * (attempt + 1))
+
+        raise RuntimeError(
+            f"append transaction still conflicting after {_TXN_RETRIES} retries"
+        )
+
+    async def snapshot(self, conversation_id: str) -> Snapshot:
+        rows = await self._client.query(_sql.SQL_SNAPSHOT, {"conv_id": conversation_id})
+        messages = [_record_to_message(conversation_id, r) for r in _normalise_rows(rows)]
+        messages.sort(key=lambda m: m.turn)  # belt-and-braces over ORDER BY
+        return Snapshot(conversation_id=conversation_id, messages=messages)
+
+    # No update / delete methods — append-only by construction (003 Principle 1).
+
+
+# ---------------------------------------------------------------------------
+# SurrealIntentLog
+# ---------------------------------------------------------------------------
+
+
+class SurrealIntentLog:
+    """Transactional outbox (003 ADR-03). Deterministic record id per
+    (step, input_ref); claim is a single conditional UPDATE, so exactly one
+    concurrent caller sees a non-empty RETURN AFTER."""
+
+    def __init__(self, client: AsyncSurreal) -> None:
+        self._client = client
+
+    async def enqueue(self, step: str, input_ref: str) -> None:
+        await self._client.query(_sql.SQL_ENQUEUE, {"step": step, "input_ref": input_ref})
+
+    async def claim(self, step: str, input_ref: str) -> bool:
+        result = await self._client.query(_sql.SQL_CLAIM, {"step": step, "input_ref": input_ref})
+        return len(_normalise_rows(result)) > 0
+
+    async def complete(self, step: str, input_ref: str) -> None:
+        await self._client.query(_sql.SQL_COMPLETE, {"step": step, "input_ref": input_ref})