wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
707 lines
26 KiB
Python
707 lines
26 KiB
Python
"""
|
||
PocketBase-backed session store.
|
||
|
||
Implements SessionStoreProtocol using PocketBase collections for all durable
|
||
storage. The key performance design:
|
||
|
||
- All methods except ``append_turn_event`` make direct PocketBase HTTP calls.
|
||
These are called at most a handful of times per turn (create, get, update
|
||
status, add message) and the ~5–10 ms overhead is acceptable.
|
||
|
||
- ``append_turn_event`` returns immediately without writing to PocketBase.
|
||
The existing ``_mirror_event_to_workspace`` in turn_runtime.py already
|
||
appends every event to a local ``events.jsonl`` file. When
|
||
``update_turn_status`` finalises a turn it reads that file and batch-posts
|
||
all events to PocketBase ``turn_events`` in a single request, trading
|
||
real-time durability for ~40× lower per-event latency during streaming.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
from pathlib import Path
|
||
import re
|
||
import time
|
||
from typing import Any
|
||
import uuid
|
||
|
||
from deeptutor.services.path_service import get_path_service
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_VALID_ID = re.compile(r"^[a-zA-Z0-9_-]+$")
|
||
|
||
|
||
def _validate_id(value: str, name: str = "id") -> str:
|
||
if not _VALID_ID.match(value):
|
||
raise ValueError(f"Invalid {name}: {value!r}")
|
||
return value
|
||
|
||
|
||
def _json_loads(value: Any, default: Any) -> Any:
|
||
if not value:
|
||
return default
|
||
if isinstance(value, (dict, list)):
|
||
return value
|
||
try:
|
||
return json.loads(value)
|
||
except Exception:
|
||
return default
|
||
|
||
|
||
def _pb():
|
||
"""Return the shared PocketBase client."""
|
||
from deeptutor.services.pocketbase_client import get_pb_client
|
||
|
||
return get_pb_client()
|
||
|
||
|
||
def _to_float(value: Any, default: float = 0.0) -> float:
|
||
try:
|
||
return float(value) if value is not None else default
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def _current_user_id() -> str:
|
||
"""Id of the request-scoped current user, used to isolate session rows.
|
||
|
||
PocketBase is a single shared server queried by one process-wide
|
||
admin-authenticated client, so it has no filesystem-level isolation. Every
|
||
session row is therefore scoped by ``user_id`` (the SQLite backend isolates
|
||
via a per-user database file instead — see ``get_sqlite_session_store``).
|
||
This reads the same ``_current_user`` ContextVar that the SQLite path
|
||
service resolves against, so the two backends share one source of truth and
|
||
are equally reliable across HTTP, WebSocket, and turn-runtime threads. Falls
|
||
back to the local-admin id in single-user / no-auth mode.
|
||
|
||
The id is validated (it always matches ``_VALID_ID`` for real users — a
|
||
PocketBase record id, a ``u_<hex>`` id, or ``local-admin``) so it is safe to
|
||
interpolate into a PocketBase filter string.
|
||
"""
|
||
from deeptutor.multi_user.context import get_current_user
|
||
|
||
return _validate_id(get_current_user().id, "user_id")
|
||
|
||
|
||
def _find_session_record(pb: Any, session_id: str, user_id: str) -> Any | None:
|
||
"""Return the ``sessions`` record for *session_id* owned by *user_id*.
|
||
|
||
Scoping every session lookup by ``user_id`` is the single point that keeps
|
||
one user from reading or mutating another's sessions on the shared
|
||
PocketBase backend. Returns ``None`` when no such row exists for this user.
|
||
"""
|
||
records = pb.collection("sessions").get_full_list(
|
||
query_params={"filter": f'session_id="{session_id}" && user_id="{user_id}"'}
|
||
)
|
||
return records[0] if records else None
|
||
|
||
|
||
class PocketBaseSessionStore:
|
||
"""PocketBase-backed implementation of SessionStoreProtocol."""
|
||
|
||
# ------------------------------------------------------------------
|
||
# Sessions
|
||
# ------------------------------------------------------------------
|
||
|
||
async def create_session(
|
||
self,
|
||
title: str | None = None,
|
||
session_id: str | None = None,
|
||
) -> dict[str, Any]:
|
||
now = time.time()
|
||
resolved_id = session_id or f"unified_{int(now * 1000)}_{uuid.uuid4().hex[:8]}"
|
||
resolved_title = (title or "New conversation").strip() or "New conversation"
|
||
owner_id = _current_user_id()
|
||
|
||
def _create():
|
||
return (
|
||
_pb()
|
||
.collection("sessions")
|
||
.create(
|
||
{
|
||
"session_id": resolved_id,
|
||
"user_id": owner_id,
|
||
"title": resolved_title[:100],
|
||
"compressed_summary": "",
|
||
"summary_up_to_msg_id": 0,
|
||
"preferences_json": {},
|
||
"capability": "",
|
||
"status": "idle",
|
||
}
|
||
)
|
||
)
|
||
|
||
record = await asyncio.to_thread(_create)
|
||
return self._session_record_to_dict(record, resolved_id, resolved_title, now)
|
||
|
||
async def get_session(self, session_id: str) -> dict[str, Any] | None:
|
||
sid = _validate_id(session_id, "session_id")
|
||
uid = _current_user_id()
|
||
|
||
def _get():
|
||
try:
|
||
return _find_session_record(_pb(), sid, uid)
|
||
except Exception:
|
||
return None
|
||
|
||
record = await asyncio.to_thread(_get)
|
||
if record is None:
|
||
return None
|
||
return self._session_record_to_dict(record)
|
||
|
||
async def ensure_session(
|
||
self,
|
||
session_id: str | None = None,
|
||
) -> dict[str, Any]:
|
||
if session_id:
|
||
session = await self.get_session(session_id)
|
||
if session is not None:
|
||
return session
|
||
return await self.create_session()
|
||
|
||
def _session_record_to_dict(
|
||
self,
|
||
record: Any,
|
||
session_id: str | None = None,
|
||
title: str | None = None,
|
||
now: float | None = None,
|
||
) -> dict[str, Any]:
|
||
sid = session_id or getattr(record, "session_id", getattr(record, "id", ""))
|
||
t = title or getattr(record, "title", "New conversation") or "New conversation"
|
||
created = _to_float(getattr(record, "created", None)) or now or time.time()
|
||
updated = _to_float(getattr(record, "updated", None)) or now or time.time()
|
||
preferences_raw = getattr(record, "preferences_json", None)
|
||
return {
|
||
"id": sid,
|
||
"session_id": sid,
|
||
"title": t,
|
||
"created_at": created,
|
||
"updated_at": updated,
|
||
"compressed_summary": getattr(record, "compressed_summary", "") or "",
|
||
"summary_up_to_msg_id": int(getattr(record, "summary_up_to_msg_id", 0) or 0),
|
||
"preferences": _json_loads(preferences_raw, {}),
|
||
"capability": getattr(record, "capability", "") or "",
|
||
"status": getattr(record, "status", "idle") or "idle",
|
||
"active_turn_id": "",
|
||
}
|
||
|
||
async def update_session_title(self, session_id: str, title: str) -> bool:
|
||
sid = _validate_id(session_id, "session_id")
|
||
uid = _current_user_id()
|
||
|
||
def _update():
|
||
record = _find_session_record(_pb(), sid, uid)
|
||
if record is None:
|
||
return False
|
||
_pb().collection("sessions").update(
|
||
record.id, {"title": (title.strip() or "New conversation")[:100]}
|
||
)
|
||
return True
|
||
|
||
try:
|
||
return await asyncio.to_thread(_update)
|
||
except Exception as exc:
|
||
logger.warning(f"update_session_title failed: {exc}")
|
||
return False
|
||
|
||
async def delete_session(self, session_id: str) -> bool:
|
||
sid = _validate_id(session_id, "session_id")
|
||
uid = _current_user_id()
|
||
|
||
def _delete():
|
||
record = _find_session_record(_pb(), sid, uid)
|
||
if record is None:
|
||
return False
|
||
_pb().collection("sessions").delete(record.id)
|
||
return True
|
||
|
||
try:
|
||
return await asyncio.to_thread(_delete)
|
||
except Exception as exc:
|
||
logger.warning(f"delete_session failed: {exc}")
|
||
return False
|
||
|
||
async def list_sessions(
|
||
self,
|
||
limit: int = 50,
|
||
offset: int = 0,
|
||
) -> list[dict[str, Any]]:
|
||
page = (offset // limit) + 1
|
||
uid = _current_user_id()
|
||
|
||
def _list():
|
||
query_params: dict[str, Any] = {"sort": "-updated", "filter": f'user_id="{uid}"'}
|
||
return _pb().collection("sessions").get_list(page, limit, query_params=query_params)
|
||
|
||
try:
|
||
result = await asyncio.to_thread(_list)
|
||
return [self._session_record_to_dict(r) for r in result.items]
|
||
except Exception as exc:
|
||
logger.warning(f"list_sessions failed: {exc}")
|
||
return []
|
||
|
||
async def update_summary(self, session_id: str, summary: str, up_to_msg_id: int) -> bool:
|
||
sid = _validate_id(session_id, "session_id")
|
||
uid = _current_user_id()
|
||
|
||
def _update():
|
||
record = _find_session_record(_pb(), sid, uid)
|
||
if record is None:
|
||
return False
|
||
_pb().collection("sessions").update(
|
||
record.id,
|
||
{
|
||
"compressed_summary": summary,
|
||
"summary_up_to_msg_id": max(0, int(up_to_msg_id)),
|
||
},
|
||
)
|
||
return True
|
||
|
||
try:
|
||
return await asyncio.to_thread(_update)
|
||
except Exception as exc:
|
||
logger.warning(f"update_summary failed: {exc}")
|
||
return False
|
||
|
||
async def update_session_preferences(
|
||
self, session_id: str, preferences: dict[str, Any]
|
||
) -> bool:
|
||
sid = _validate_id(session_id, "session_id")
|
||
|
||
async def _merge():
|
||
session = await self.get_session(sid)
|
||
if session is None:
|
||
return False
|
||
merged = {**session.get("preferences", {}), **(preferences or {})}
|
||
uid = _current_user_id()
|
||
|
||
def _update():
|
||
record = _find_session_record(_pb(), sid, uid)
|
||
if record is None:
|
||
return False
|
||
_pb().collection("sessions").update(record.id, {"preferences_json": merged})
|
||
return True
|
||
|
||
return await asyncio.to_thread(_update)
|
||
|
||
try:
|
||
return await _merge()
|
||
except Exception as exc:
|
||
logger.warning(f"update_session_preferences failed: {exc}")
|
||
return False
|
||
|
||
async def get_session_with_messages(self, session_id: str) -> dict[str, Any] | None:
|
||
session = await self.get_session(session_id)
|
||
if session is None:
|
||
return None
|
||
session["messages"] = await self.get_messages(session_id)
|
||
session["active_turns"] = await self.list_active_turns(session_id)
|
||
return session
|
||
|
||
# ------------------------------------------------------------------
|
||
# Messages
|
||
# ------------------------------------------------------------------
|
||
# Messages/turns/turn_events are keyed by ``session_id`` and are reached
|
||
# from the API only through a session lookup that is already user-scoped
|
||
# (``get_session_with_messages`` returns ``None`` for another user's
|
||
# session before any message is fetched, and ``create_turn`` rejects a
|
||
# session the caller doesn't own). Internal callers always operate on the
|
||
# current user's own session, so these rows don't carry a separate
|
||
# ``user_id`` filter — the session boundary above is the access gate.
|
||
|
||
async def add_message(
|
||
self,
|
||
session_id: str,
|
||
role: str,
|
||
content: str,
|
||
capability: str = "",
|
||
events: list[dict[str, Any]] | None = None,
|
||
attachments: list[dict[str, Any]] | None = None,
|
||
metadata: dict[str, Any] | None = None,
|
||
parent_message_id: int | None = None,
|
||
) -> int:
|
||
# ``parent_message_id`` is accepted to match the protocol shape but is
|
||
# not yet wired through PocketBase storage — branching only works on
|
||
# the SQLite backend today.
|
||
_ = parent_message_id
|
||
sid = _validate_id(session_id, "session_id")
|
||
now = time.time()
|
||
|
||
def _add():
|
||
payload = {
|
||
"session_id": sid,
|
||
"role": role,
|
||
"content": content or "",
|
||
"capability": capability or "",
|
||
"events_json": events or [],
|
||
"attachments_json": attachments or [],
|
||
"metadata_json": metadata or {},
|
||
"msg_created_at": now,
|
||
}
|
||
record = _pb().collection("messages").create(payload)
|
||
# Title generation is owned by the turn runtime (LLM-driven
|
||
# after the first user+assistant pair). Until that runs the
|
||
# session keeps the ``New conversation`` sentinel.
|
||
return record
|
||
|
||
try:
|
||
record = await asyncio.to_thread(_add)
|
||
# Return a synthetic integer id using epoch ms
|
||
return int(now * 1000)
|
||
except Exception as exc:
|
||
logger.warning(f"add_message failed: {exc}")
|
||
return 0
|
||
|
||
async def delete_message(self, message_id: int | str) -> bool:
|
||
def _delete():
|
||
_pb().collection("messages").delete(str(message_id))
|
||
return True
|
||
|
||
try:
|
||
return await asyncio.to_thread(_delete)
|
||
except Exception as exc:
|
||
logger.warning(f"delete_message failed: {exc}")
|
||
return False
|
||
|
||
async def get_last_message(
|
||
self, session_id: str, role: str | None = None
|
||
) -> dict[str, Any] | None:
|
||
sid = _validate_id(session_id, "session_id")
|
||
filter_str = f'session_id="{sid}"'
|
||
if role:
|
||
filter_str += f' && role="{role}"'
|
||
|
||
def _get():
|
||
records = (
|
||
_pb()
|
||
.collection("messages")
|
||
.get_full_list(
|
||
query_params={
|
||
"filter": filter_str,
|
||
"sort": "-msg_created_at",
|
||
"perPage": 1,
|
||
}
|
||
)
|
||
)
|
||
return records[0] if records else None
|
||
|
||
try:
|
||
record = await asyncio.to_thread(_get)
|
||
return self._message_record_to_dict(record) if record is not None else None
|
||
except Exception as exc:
|
||
logger.warning(f"get_last_message failed: {exc}")
|
||
return None
|
||
|
||
async def get_messages(self, session_id: str) -> list[dict[str, Any]]:
|
||
sid = _validate_id(session_id, "session_id")
|
||
|
||
def _get():
|
||
return (
|
||
_pb()
|
||
.collection("messages")
|
||
.get_full_list(
|
||
query_params={
|
||
"filter": f'session_id="{sid}"',
|
||
"sort": "msg_created_at",
|
||
}
|
||
)
|
||
)
|
||
|
||
try:
|
||
records = await asyncio.to_thread(_get)
|
||
return [self._message_record_to_dict(r) for r in records]
|
||
except Exception as exc:
|
||
logger.warning(f"get_messages failed: {exc}")
|
||
return []
|
||
|
||
async def get_messages_for_context(
|
||
self, session_id: str, leaf_message_id: int | None = None
|
||
) -> list[dict[str, Any]]:
|
||
# leaf_message_id (branch-aware context) is not supported on PocketBase
|
||
# yet; fall back to the linear, append-only view.
|
||
_ = leaf_message_id
|
||
messages = await self.get_messages(session_id)
|
||
return [
|
||
{"id": m["id"], "role": m["role"], "content": m["content"] or ""}
|
||
for m in messages
|
||
if m["role"] in ("user", "assistant", "system")
|
||
]
|
||
|
||
def _message_record_to_dict(self, record: Any) -> dict[str, Any]:
|
||
return {
|
||
"id": getattr(record, "id", ""),
|
||
"session_id": getattr(record, "session_id", ""),
|
||
"role": getattr(record, "role", ""),
|
||
"content": getattr(record, "content", "") or "",
|
||
"capability": getattr(record, "capability", "") or "",
|
||
"events": _json_loads(getattr(record, "events_json", None), []),
|
||
"attachments": _json_loads(getattr(record, "attachments_json", None), []),
|
||
"metadata": _json_loads(getattr(record, "metadata_json", None), {}),
|
||
"created_at": _to_float(getattr(record, "msg_created_at", None)),
|
||
}
|
||
|
||
# ------------------------------------------------------------------
|
||
# Turns
|
||
# ------------------------------------------------------------------
|
||
|
||
async def create_turn(self, session_id: str, capability: str = "") -> dict[str, Any]:
|
||
sid = _validate_id(session_id, "session_id")
|
||
uid = _current_user_id()
|
||
now = time.time()
|
||
turn_id = f"turn_{int(now * 1000)}_{uuid.uuid4().hex[:10]}"
|
||
|
||
def _create():
|
||
# Guard: ensure the session exists AND belongs to the current user.
|
||
if _find_session_record(_pb(), sid, uid) is None:
|
||
raise ValueError(f"Session not found: {sid}")
|
||
# Guard: no duplicate active turns
|
||
active = (
|
||
_pb()
|
||
.collection("turns")
|
||
.get_full_list(query_params={"filter": f'session_id="{sid}" && status="running"'})
|
||
)
|
||
if active:
|
||
raise RuntimeError(f"Session already has an active turn: {active[0].turn_id}")
|
||
return (
|
||
_pb()
|
||
.collection("turns")
|
||
.create(
|
||
{
|
||
"turn_id": turn_id,
|
||
"session_id": sid,
|
||
"capability": capability or "",
|
||
"status": "running",
|
||
"error": "",
|
||
"turn_created_at": now,
|
||
"turn_updated_at": now,
|
||
"finished_at": None,
|
||
}
|
||
)
|
||
)
|
||
|
||
await asyncio.to_thread(_create)
|
||
return {
|
||
"id": turn_id,
|
||
"turn_id": turn_id,
|
||
"session_id": sid,
|
||
"capability": capability or "",
|
||
"status": "running",
|
||
"error": "",
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
"finished_at": None,
|
||
"last_seq": 0,
|
||
}
|
||
|
||
async def get_turn(self, turn_id: str) -> dict[str, Any] | None:
|
||
tid = _validate_id(turn_id, "turn_id")
|
||
|
||
def _get():
|
||
records = (
|
||
_pb().collection("turns").get_full_list(query_params={"filter": f'turn_id="{tid}"'})
|
||
)
|
||
return records[0] if records else None
|
||
|
||
record = await asyncio.to_thread(_get)
|
||
return self._turn_record_to_dict(record) if record else None
|
||
|
||
async def get_active_turn(self, session_id: str) -> dict[str, Any] | None:
|
||
sid = _validate_id(session_id, "session_id")
|
||
|
||
def _get():
|
||
records = (
|
||
_pb()
|
||
.collection("turns")
|
||
.get_full_list(
|
||
query_params={
|
||
"filter": f'session_id="{sid}" && status="running"',
|
||
"sort": "-turn_updated_at",
|
||
}
|
||
)
|
||
)
|
||
return records[0] if records else None
|
||
|
||
record = await asyncio.to_thread(_get)
|
||
return self._turn_record_to_dict(record) if record else None
|
||
|
||
async def list_active_turns(self, session_id: str) -> list[dict[str, Any]]:
|
||
sid = _validate_id(session_id, "session_id")
|
||
|
||
def _list():
|
||
return (
|
||
_pb()
|
||
.collection("turns")
|
||
.get_full_list(
|
||
query_params={
|
||
"filter": f'session_id="{sid}" && status="running"',
|
||
"sort": "-turn_updated_at",
|
||
}
|
||
)
|
||
)
|
||
|
||
try:
|
||
records = await asyncio.to_thread(_list)
|
||
return [self._turn_record_to_dict(r) for r in records]
|
||
except Exception:
|
||
return []
|
||
|
||
async def update_turn_status(self, turn_id: str, status: str, error: str = "") -> bool:
|
||
tid = _validate_id(turn_id, "turn_id")
|
||
now = time.time()
|
||
finished_at = now if status in {"completed", "failed", "cancelled"} else None
|
||
|
||
def _update():
|
||
records = (
|
||
_pb().collection("turns").get_full_list(query_params={"filter": f'turn_id="{tid}"'})
|
||
)
|
||
if not records:
|
||
return False
|
||
_pb().collection("turns").update(
|
||
records[0].id,
|
||
{
|
||
"status": status,
|
||
"error": error or "",
|
||
"turn_updated_at": now,
|
||
"finished_at": finished_at,
|
||
},
|
||
)
|
||
return True
|
||
|
||
try:
|
||
updated = await asyncio.to_thread(_update)
|
||
except Exception as exc:
|
||
logger.warning(f"update_turn_status failed: {exc}")
|
||
return False
|
||
|
||
# Batch-flush turn events from local JSONL buffer to PocketBase on finalisation
|
||
if updated and finished_at is not None:
|
||
await self._flush_turn_events(tid)
|
||
|
||
return updated
|
||
|
||
async def _flush_turn_events(self, turn_id: str) -> None:
|
||
"""
|
||
Read the local events.jsonl write-ahead buffer and batch-POST all
|
||
events to PocketBase turn_events collection in a single background call.
|
||
"""
|
||
tid = _validate_id(turn_id, "turn_id")
|
||
try:
|
||
path_service = get_path_service()
|
||
# The JSONL file is written by _mirror_event_to_workspace; we look
|
||
# across all capability workspaces since we only have the turn_id.
|
||
workspace_root = path_service.get_user_root()
|
||
jsonl_files: list[Path] = list(workspace_root.rglob(f"{tid}/events.jsonl"))
|
||
|
||
if not jsonl_files:
|
||
return
|
||
|
||
events: list[dict[str, Any]] = []
|
||
for jsonl_path in jsonl_files:
|
||
try:
|
||
for line in jsonl_path.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if line:
|
||
events.append(json.loads(line))
|
||
except Exception as exc:
|
||
logger.debug(f"Could not read events.jsonl at {jsonl_path}: {exc}")
|
||
|
||
if not events:
|
||
return
|
||
|
||
def _batch_create():
|
||
pb = _pb()
|
||
for event in events:
|
||
try:
|
||
pb.collection("turn_events").create(
|
||
{
|
||
"turn_id": tid,
|
||
"session_id": event.get("session_id", ""),
|
||
"seq": int(event.get("seq", 0)),
|
||
"type": event.get("type", ""),
|
||
"source": event.get("source", ""),
|
||
"stage": event.get("stage", ""),
|
||
"content": str(event.get("content", ""))[:10000],
|
||
"metadata_json": event.get("metadata", {}),
|
||
"event_timestamp": float(event.get("timestamp", 0)),
|
||
}
|
||
)
|
||
except Exception as exc:
|
||
logger.debug(f"turn_events batch item failed: {exc}")
|
||
|
||
await asyncio.to_thread(_batch_create)
|
||
logger.debug(f"Flushed {len(events)} turn events for {tid} to PocketBase")
|
||
|
||
except Exception as exc:
|
||
logger.warning(f"_flush_turn_events failed for {turn_id}: {exc}")
|
||
|
||
def _turn_record_to_dict(self, record: Any) -> dict[str, Any]:
|
||
turn_id = getattr(record, "turn_id", getattr(record, "id", ""))
|
||
return {
|
||
"id": turn_id,
|
||
"turn_id": turn_id,
|
||
"session_id": getattr(record, "session_id", ""),
|
||
"capability": getattr(record, "capability", "") or "",
|
||
"status": getattr(record, "status", "running") or "running",
|
||
"error": getattr(record, "error", "") or "",
|
||
"created_at": _to_float(getattr(record, "turn_created_at", None)),
|
||
"updated_at": _to_float(getattr(record, "turn_updated_at", None)),
|
||
"finished_at": _to_float(getattr(record, "finished_at", None)) or None,
|
||
"last_seq": 0,
|
||
}
|
||
|
||
# ------------------------------------------------------------------
|
||
# Turn events — write-ahead only; batch flush handled in update_turn_status
|
||
# ------------------------------------------------------------------
|
||
|
||
async def append_turn_event(self, turn_id: str, event: dict[str, Any]) -> dict[str, Any]:
|
||
"""
|
||
Assign a monotonic seq number and return the annotated payload.
|
||
|
||
Does NOT write to PocketBase immediately — the caller's
|
||
_mirror_event_to_workspace already appends to events.jsonl, which
|
||
is flushed to PocketBase in bulk when the turn is finalised.
|
||
"""
|
||
payload = dict(event)
|
||
payload.setdefault("turn_id", turn_id)
|
||
# Assign seq if not provided; use timestamp-based counter as fallback.
|
||
if not payload.get("seq"):
|
||
payload["seq"] = int(time.time() * 1000) % 1_000_000
|
||
return payload
|
||
|
||
async def get_turn_events(self, turn_id: str, after_seq: int = 0) -> list[dict[str, Any]]:
|
||
"""Retrieve persisted turn events from PocketBase (post-turn replay)."""
|
||
tid = _validate_id(turn_id, "turn_id")
|
||
|
||
def _get():
|
||
filter_str = f'turn_id="{tid}"'
|
||
if after_seq > 0:
|
||
filter_str += f" && seq > {after_seq}"
|
||
return (
|
||
_pb()
|
||
.collection("turn_events")
|
||
.get_full_list(query_params={"filter": filter_str, "sort": "seq"})
|
||
)
|
||
|
||
try:
|
||
records = await asyncio.to_thread(_get)
|
||
return [
|
||
{
|
||
"type": getattr(r, "type", ""),
|
||
"source": getattr(r, "source", ""),
|
||
"stage": getattr(r, "stage", ""),
|
||
"content": getattr(r, "content", "") or "",
|
||
"metadata": _json_loads(getattr(r, "metadata_json", None), {}),
|
||
"session_id": getattr(r, "session_id", ""),
|
||
"turn_id": tid,
|
||
"seq": int(getattr(r, "seq", 0)),
|
||
"timestamp": _to_float(getattr(r, "event_timestamp", None)),
|
||
}
|
||
for r in records
|
||
]
|
||
except Exception as exc:
|
||
logger.warning(f"get_turn_events failed: {exc}")
|
||
return []
|