From 5639b60ec949bd3910791b7bc2bef3e39e75465e Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 2 Aug 2026 09:00:31 +0000 Subject: [PATCH 1/2] Initial plan From f3258a5de4fee3d31ae158c439c9236ec88d6865 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 2 Aug 2026 09:06:27 +0000 Subject: [PATCH 2/2] Replace JSON timeline/A2A stores with SQL persistence --- .env.example | 7 ++ README.md | 19 +++- a2a_store.py | 153 ++++++++++++++------------ docker-compose.yml | 29 +++++ docs/DEPLOYMENT.md | 23 ++++ env.example | 5 + requirements.txt | 2 + scripts/migrate_json_to_sql.py | 150 ++++++++++++++++++++++++++ storage_db.py | 192 +++++++++++++++++++++++++++++++++ tests/test_sql_storage.py | 184 +++++++++++++++++++++++++++++++ timeline_store.py | 136 +++++++++++------------ 11 files changed, 755 insertions(+), 145 deletions(-) create mode 100644 scripts/migrate_json_to_sql.py create mode 100644 storage_db.py create mode 100644 tests/test_sql_storage.py diff --git a/.env.example b/.env.example index 74f8818..24c207d 100644 --- a/.env.example +++ b/.env.example @@ -11,6 +11,13 @@ X_USERNAME=your_twitter_username # Grok/xAI Settings XAI_API_KEY=your_xai_api_key_here +# Timeline/A2A persistence (Python services) +# Local SQLite example: +# DATABASE_URL=sqlite:///~/.xmcp/xmcp.db +# Railway Postgres example (also accepts postgres:// and auto-normalizes): +# DATABASE_URL=******host:5432/dbname +DATABASE_URL= + # Agent Settings POLLING_INTERVAL_MS=30000 MAX_RETRIES=3 diff --git a/README.md b/README.md index edd4471..f378293 100644 --- a/README.md +++ b/README.md @@ -166,13 +166,24 @@ Wire the cross-service URLs after deployment: ``` MCP_SERVER_URL=https://.up.railway.app/mcp TIMELINE_API_URL=https://.up.railway.app +# All four services must share one database: +DATABASE_URL=postgres://:@:/ ``` See `docs/DEPLOYMENT.md` for full Railway setup details. ## Data Storage -Timeline cards and A2A messages are stored in JSON files at `~/.xmcp/` by default. This is intentional for lightweight local use. For production, override `TIMELINE_STORE_PATH` and `A2A_STORE_PATH` to point to a persistent volume. +Timeline cards and A2A messages are stored in SQL and selected by `DATABASE_URL`: + +- `DATABASE_URL` unset (or `sqlite://...`) → SQLite (`~/.xmcp/xmcp.db` by default) +- `postgres://...` or `postgresql://...` → Postgres (Railway-ready; `postgres://` is normalized automatically) + +Tables are created automatically on startup. For local concurrency safety, SQLite enables WAL mode and a busy timeout. To migrate legacy JSON stores (`TIMELINE_STORE_PATH`, `A2A_STORE_PATH`), run: + +```bash +python scripts/migrate_json_to_sql.py +``` ## Project Structure @@ -181,8 +192,10 @@ Timeline cards and A2A messages are stored in JSON files at `~/.xmcp/` by defaul ├── timeline_server.py # Timeline + A2A FastAPI server ├── listener.py # X mention poller + Grok responder ├── mcp_dispatcher.py # Timeline action executor -├── timeline_store.py # JSON-file timeline persistence -├── a2a_store.py # JSON-file A2A persistence +├── timeline_store.py # SQL-backed timeline persistence +├── a2a_store.py # SQL-backed A2A persistence +├── storage_db.py # Shared SQLAlchemy engine/schema +├── scripts/migrate_json_to_sql.py ├── openapi.json # X API OpenAPI spec (used by MCP server) ├── src/ # TypeScript standalone agent (alternative) ├── docs/ # Architecture, deployment, usage guides diff --git a/a2a_store.py b/a2a_store.py index dc0be35..83ec7a5 100644 --- a/a2a_store.py +++ b/a2a_store.py @@ -1,13 +1,17 @@ -import json -import os -import threading import uuid -from datetime import datetime, timezone -from pathlib import Path from typing import Any, Dict, List, Optional -A2A_STORE_PATH = Path(os.getenv("A2A_STORE_PATH", "~/.xmcp/a2a_store.json")).expanduser() -A2A_STORE_LOCK = threading.Lock() +from sqlalchemy import insert, select, update + +from storage_db import ( + a2a_agents, + a2a_messages, + row_to_dict, + serialize_record, + utc_now, + write_connection, + read_connection, +) DEFAULT_AGENTS = [ { @@ -16,6 +20,7 @@ "description": "Dispatches timeline actions to MCP-enabled tools.", "status": "online", "endpoint": "local", + "kind": "agent", "tags": ["mcp", "orchestrator"], }, { @@ -24,6 +29,7 @@ "description": "Handles @mentions and X actions.", "status": "online", "endpoint": "x", + "kind": "agent", "tags": ["x", "social"], }, { @@ -32,53 +38,63 @@ "description": "Flokk timeline surface.", "status": "online", "endpoint": "flokk", + "kind": "agent", "tags": ["ui", "timeline"], }, ] -def _utc_now() -> str: - return datetime.now(timezone.utc).isoformat() +def _normalize_kind(value: Any) -> str: + return value if value in ("agent", "bot") else "agent" + +def _serialize_agent(record: Dict[str, Any]) -> Dict[str, Any]: + value = serialize_record(record) + value["kind"] = _normalize_kind(value.get("kind")) + return value -def _ensure_store() -> None: - if A2A_STORE_PATH.exists(): - return - A2A_STORE_PATH.parent.mkdir(parents=True, exist_ok=True) - data = {"agents": DEFAULT_AGENTS, "messages": []} - A2A_STORE_PATH.write_text(json.dumps(data, indent=2), encoding="utf-8") +def _serialize_message(record: Dict[str, Any]) -> Dict[str, Any]: + value = serialize_record(record) + value["from"] = value.pop("from_agent") + value["to"] = value.pop("to_agent") + return value -def _read_store() -> Dict[str, Any]: - _ensure_store() - raw = A2A_STORE_PATH.read_text(encoding="utf-8") - try: - data = json.loads(raw) - except json.JSONDecodeError: - data = {"agents": DEFAULT_AGENTS, "messages": []} - if "agents" not in data or not isinstance(data["agents"], list): - data["agents"] = DEFAULT_AGENTS - if "messages" not in data or not isinstance(data["messages"], list): - data["messages"] = [] - return data +def _ensure_default_agents() -> None: + with write_connection() as conn: + existing_ids = { + row[0] + for row in conn.execute(select(a2a_agents.c.id).where(a2a_agents.c.id.in_([a["id"] for a in DEFAULT_AGENTS]))) + } -def _write_store(data: Dict[str, Any]) -> None: - A2A_STORE_PATH.parent.mkdir(parents=True, exist_ok=True) - A2A_STORE_PATH.write_text(json.dumps(data, indent=2), encoding="utf-8") + for agent in DEFAULT_AGENTS: + if agent["id"] in existing_ids: + continue + conn.execute( + insert(a2a_agents).values( + **agent, + created_at=utc_now(), + ) + ) def list_agents() -> List[Dict[str, Any]]: - with A2A_STORE_LOCK: - return _read_store()["agents"] + _ensure_default_agents() + query = select(a2a_agents).order_by(a2a_agents.c.created_at.asc(), a2a_agents.c.id.asc()) + with read_connection() as conn: + rows = conn.execute(query).fetchall() + return [_serialize_agent(row_to_dict(row)) for row in rows] def get_agent(agent_id: str) -> Optional[Dict[str, Any]]: - with A2A_STORE_LOCK: - for agent in _read_store()["agents"]: - if agent.get("id") == agent_id: - return agent - return None + _ensure_default_agents() + query = select(a2a_agents).where(a2a_agents.c.id == agent_id) + with read_connection() as conn: + row = conn.execute(query).fetchone() + if not row: + return None + return _serialize_agent(row_to_dict(row)) def register_agent(payload: Dict[str, Any]) -> Dict[str, Any]: @@ -88,46 +104,51 @@ def register_agent(payload: Dict[str, Any]) -> Dict[str, Any]: "description": payload.get("description", ""), "status": payload.get("status", "offline"), "endpoint": payload.get("endpoint", ""), - # Classification: "agent" (interactive, LLM-backed, autonomous) vs - # "bot" (deterministic function executor). Normalized so null or - # unknown values can never enter the registry. - "kind": payload.get("kind") if payload.get("kind") in ("agent", "bot") else "agent", + "kind": _normalize_kind(payload.get("kind")), "tags": payload.get("tags", []), - "created_at": _utc_now(), } - with A2A_STORE_LOCK: - data = _read_store() - for existing in data["agents"]: - if existing.get("id") == agent["id"]: - # Re-registration updates mutable fields so seeded records - # (e.g. x-agent without kind) don't stay stale forever. - for field in ("name", "description", "status", "endpoint", "kind", "tags"): - existing[field] = agent[field] - _write_store(data) - return existing - data["agents"].append(agent) - _write_store(data) - return agent + + with write_connection() as conn: + existing = conn.execute( + select(a2a_agents).where(a2a_agents.c.id == agent["id"]) + ).fetchone() + + if existing: + conn.execute( + update(a2a_agents) + .where(a2a_agents.c.id == agent["id"]) + .values(**agent) + ) + row = row_to_dict(existing) + row.update(agent) + return _serialize_agent(row) + + created = {**agent, "created_at": utc_now()} + conn.execute(insert(a2a_agents).values(**created)) + return _serialize_agent(created) def list_messages(agent_id: str) -> List[Dict[str, Any]]: - with A2A_STORE_LOCK: - data = _read_store() - return [msg for msg in data["messages"] if msg.get("to") == agent_id] + query = ( + select(a2a_messages) + .where(a2a_messages.c.to_agent == agent_id) + .order_by(a2a_messages.c.created_at.desc()) + ) + with read_connection() as conn: + rows = conn.execute(query).fetchall() + return [_serialize_message(row_to_dict(row)) for row in rows] def add_message(payload: Dict[str, Any]) -> Dict[str, Any]: message = { "id": payload.get("id") or str(uuid.uuid4()), - "from": payload.get("from", "system"), - "to": payload.get("to", "timeline-ui"), + "from_agent": payload.get("from", "system"), + "to_agent": payload.get("to", "timeline-ui"), "type": payload.get("type", "info"), "content": payload.get("content", ""), "metadata": payload.get("metadata", {}), - "created_at": _utc_now(), + "created_at": utc_now(), } - with A2A_STORE_LOCK: - data = _read_store() - data["messages"].insert(0, message) - _write_store(data) - return message + with write_connection() as conn: + conn.execute(insert(a2a_messages).values(**message)) + return _serialize_message(message) diff --git a/docker-compose.yml b/docker-compose.yml index 4555516..bbc7c69 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,4 +1,19 @@ services: + postgres: + image: postgres:16-alpine + environment: + - POSTGRES_DB=xmcp + - POSTGRES_USER=xmcp + - POSTGRES_HOST_AUTH_METHOD=trust + ports: + - "5432:5432" + healthcheck: + test: ["CMD-SHELL", "pg_isready -U xmcp -d xmcp"] + interval: 10s + timeout: 5s + retries: 5 + restart: unless-stopped + mcp-server: build: . command: python server.py @@ -8,6 +23,10 @@ services: environment: - MCP_HOST=0.0.0.0 - MCP_PORT=8000 + - DATABASE_URL=postgresql://xmcp@postgres:5432/xmcp + depends_on: + postgres: + condition: service_healthy restart: unless-stopped timeline-server: @@ -19,8 +38,12 @@ services: environment: - TIMELINE_HOST=0.0.0.0 - TIMELINE_PORT=8080 + - DATABASE_URL=postgresql://xmcp@postgres:5432/xmcp volumes: - xmcp-data:/root/.xmcp + depends_on: + postgres: + condition: service_healthy restart: unless-stopped healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8080/health"] @@ -35,6 +58,7 @@ services: environment: - MCP_SERVER_URL=http://mcp-server:8000/mcp - TIMELINE_API_URL=http://timeline-server:8080 + - DATABASE_URL=postgresql://xmcp@postgres:5432/xmcp volumes: - xmcp-data:/root/.xmcp depends_on: @@ -42,6 +66,8 @@ services: condition: service_healthy mcp-server: condition: service_started + postgres: + condition: service_healthy restart: unless-stopped mcp-dispatcher: @@ -51,6 +77,7 @@ services: environment: - MCP_SERVER_URL=http://mcp-server:8000/mcp - TIMELINE_API_URL=http://timeline-server:8080 + - DATABASE_URL=postgresql://xmcp@postgres:5432/xmcp volumes: - xmcp-data:/root/.xmcp depends_on: @@ -58,6 +85,8 @@ services: condition: service_healthy mcp-server: condition: service_started + postgres: + condition: service_healthy restart: unless-stopped volumes: diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index 2277fbe..1daf535 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -2,6 +2,29 @@ ## Deployment Options +### Railway (Python service stack) + +Deploy **four services** from this same repository: + +| Service | Start command | +|---|---| +| `mcp-server` | `python server.py` | +| `timeline-server` | `python timeline_server.py` | +| `listener` | `python listener.py` | +| `mcp-dispatcher` | `python mcp_dispatcher.py` | + +Add a Railway Postgres plugin, then set the **same** `DATABASE_URL` on all four services. + +Storage backend selection: +- `DATABASE_URL` unset/`sqlite://...` → local SQLite (`~/.xmcp/xmcp.db`) +- `postgres://...` or `postgresql://...` → Postgres (`postgres://` is normalized automatically) + +The timeline and A2A schema are auto-created at startup. If you are migrating old local JSON stores, run: + +```bash +python scripts/migrate_json_to_sql.py +``` + ### 1. Local Development Machine **Best for:** Testing and personal use diff --git a/env.example b/env.example index e6c6dd1..5b46004 100644 --- a/env.example +++ b/env.example @@ -32,6 +32,11 @@ TIMELINE_HOST=0.0.0.0 TIMELINE_PORT=8080 TIMELINE_API_URL=http://127.0.0.1:8080 TIMELINE_USER_ID=default +# Single source of truth for timeline + A2A persistence. +# Unset -> sqlite:///~/.xmcp/xmcp.db (local default) +# Railway/Postgres -> postgresql://... (or postgres://...; auto-normalized) +DATABASE_URL= +# Legacy JSON paths only used by scripts/migrate_json_to_sql.py: TIMELINE_STORE_PATH=~/.xmcp/timeline_store.json A2A_STORE_PATH=~/.xmcp/a2a_store.json TIMELINE_ACTION_AGENT=mcp-orchestrator diff --git a/requirements.txt b/requirements.txt index 2740e18..f63e155 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,6 +7,8 @@ python-dotenv>=1.2.2 fastapi>=0.141.1 uvicorn>=0.52.0 pydantic>=2.13.4 +sqlalchemy>=2.0.43 +psycopg[binary]>=3.2.10 # X API listener tweepy>=4.17.0 diff --git a/scripts/migrate_json_to_sql.py b/scripts/migrate_json_to_sql.py new file mode 100644 index 0000000..7970ded --- /dev/null +++ b/scripts/migrate_json_to_sql.py @@ -0,0 +1,150 @@ +#!/usr/bin/env python3 +import json +import os +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, Iterable, Tuple + +from sqlalchemy import insert, select, update + +from storage_db import a2a_agents, a2a_messages, timeline_items, write_connection + + +def _parse_timestamp(value: Any) -> datetime: + if isinstance(value, datetime): + return value + if isinstance(value, str) and value: + try: + return datetime.fromisoformat(value) + except ValueError: + pass + return datetime.now(timezone.utc) + + +def _read_json(path: Path) -> Dict[str, Any]: + if not path.exists(): + return {} + try: + return json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + return {} + + +def _upsert_timeline_items(items: Iterable[Dict[str, Any]]) -> Tuple[int, int]: + inserted = 0 + updated = 0 + with write_connection() as conn: + for item in items: + item_id = item.get("id") + if not item_id: + continue + payload = { + "id": item_id, + "user_id": item.get("user_id", "default"), + "title": item.get("title", "Untitled"), + "body": item.get("body", ""), + "status": item.get("status", "unread"), + "posted_by": item.get("posted_by", "agent"), + "actions": item.get("actions", []), + "metadata": item.get("metadata", {}), + "created_at": _parse_timestamp(item.get("created_at")), + "updated_at": _parse_timestamp(item.get("updated_at")) if item.get("updated_at") else None, + } + exists = conn.execute( + select(timeline_items.c.id).where(timeline_items.c.id == item_id) + ).fetchone() + if exists: + conn.execute(update(timeline_items).where(timeline_items.c.id == item_id).values(**payload)) + updated += 1 + else: + conn.execute(insert(timeline_items).values(**payload)) + inserted += 1 + return inserted, updated + + +def _upsert_agents(agents: Iterable[Dict[str, Any]]) -> Tuple[int, int]: + inserted = 0 + updated = 0 + with write_connection() as conn: + for agent in agents: + agent_id = agent.get("id") + if not agent_id: + continue + payload = { + "id": agent_id, + "name": agent.get("name", "Agent"), + "description": agent.get("description", ""), + "status": agent.get("status", "offline"), + "endpoint": agent.get("endpoint", ""), + "kind": agent.get("kind") if agent.get("kind") in ("agent", "bot") else "agent", + "tags": agent.get("tags", []), + "created_at": _parse_timestamp(agent.get("created_at")), + } + exists = conn.execute(select(a2a_agents.c.id).where(a2a_agents.c.id == agent_id)).fetchone() + if exists: + conn.execute(update(a2a_agents).where(a2a_agents.c.id == agent_id).values(**payload)) + updated += 1 + else: + conn.execute(insert(a2a_agents).values(**payload)) + inserted += 1 + return inserted, updated + + +def _upsert_messages(messages: Iterable[Dict[str, Any]]) -> Tuple[int, int]: + inserted = 0 + updated = 0 + with write_connection() as conn: + for message in messages: + message_id = message.get("id") + if not message_id: + continue + payload = { + "id": message_id, + "from_agent": message.get("from", "system"), + "to_agent": message.get("to", "timeline-ui"), + "type": message.get("type", "info"), + "content": message.get("content", ""), + "metadata": message.get("metadata", {}), + "created_at": _parse_timestamp(message.get("created_at")), + } + exists = conn.execute( + select(a2a_messages.c.id).where(a2a_messages.c.id == message_id) + ).fetchone() + if exists: + conn.execute(update(a2a_messages).where(a2a_messages.c.id == message_id).values(**payload)) + updated += 1 + else: + conn.execute(insert(a2a_messages).values(**payload)) + inserted += 1 + return inserted, updated + + +def migrate() -> Dict[str, Tuple[int, int]]: + timeline_path = Path( + os.getenv("TIMELINE_STORE_PATH", "~/.xmcp/timeline_store.json") + ).expanduser() + a2a_path = Path(os.getenv("A2A_STORE_PATH", "~/.xmcp/a2a_store.json")).expanduser() + + timeline_data = _read_json(timeline_path) + a2a_data = _read_json(a2a_path) + + timeline_result = _upsert_timeline_items(timeline_data.get("items", [])) + agents_result = _upsert_agents(a2a_data.get("agents", [])) + messages_result = _upsert_messages(a2a_data.get("messages", [])) + + return { + "timeline_items": timeline_result, + "a2a_agents": agents_result, + "a2a_messages": messages_result, + } + + +def main() -> None: + result = migrate() + print("Migration complete:") + for table, (inserted, updated) in result.items(): + print(f"- {table}: inserted={inserted}, updated={updated}") + + +if __name__ == "__main__": + main() diff --git a/storage_db.py b/storage_db.py new file mode 100644 index 0000000..eacd295 --- /dev/null +++ b/storage_db.py @@ -0,0 +1,192 @@ +import os +import threading +from contextlib import contextmanager +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, Iterator, Optional + +from sqlalchemy import ( + JSON, + Column, + DateTime, + MetaData, + String, + Table, + Text, + create_engine, + event, + text, +) +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.engine import Connection, Engine + +DEFAULT_DB_PATH = Path("~/.xmcp/xmcp.db").expanduser() +SQLITE_BUSY_TIMEOUT_MS = int(os.getenv("SQLITE_BUSY_TIMEOUT_MS", "5000")) + +_ENGINE: Optional[Engine] = None +_ENGINE_LOCK = threading.Lock() + +metadata = MetaData() + +json_type = JSON().with_variant(JSONB, "postgresql") + +timeline_items = Table( + "timeline_items", + metadata, + Column("id", String, primary_key=True), + Column("user_id", String, nullable=False, index=True), + Column("title", String, nullable=False), + Column("body", Text, nullable=False, default=""), + Column("status", String, nullable=False, index=True), + Column("posted_by", String, nullable=False), + Column("actions", json_type, nullable=False), + Column("metadata", json_type, nullable=False), + Column("created_at", DateTime(timezone=True), nullable=False, index=True), + Column("updated_at", DateTime(timezone=True), nullable=True, index=True), +) + +a2a_agents = Table( + "a2a_agents", + metadata, + Column("id", String, primary_key=True), + Column("name", String, nullable=False), + Column("description", Text, nullable=False, default=""), + Column("status", String, nullable=False, default="offline"), + Column("endpoint", String, nullable=False, default=""), + Column("kind", String, nullable=False, default="agent"), + Column("tags", json_type, nullable=False), + Column("created_at", DateTime(timezone=True), nullable=False, index=True), +) + +a2a_messages = Table( + "a2a_messages", + metadata, + Column("id", String, primary_key=True), + Column("from_agent", String, nullable=False), + Column("to_agent", String, nullable=False, index=True), + Column("type", String, nullable=False), + Column("content", Text, nullable=False, default=""), + Column("metadata", json_type, nullable=False), + Column("created_at", DateTime(timezone=True), nullable=False, index=True), +) + + +def utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def normalize_database_url(url: Optional[str]) -> str: + raw = (url or "").strip() + if not raw: + return f"sqlite:///{DEFAULT_DB_PATH}" + if raw.startswith("postgres://"): + return "postgresql://" + raw[len("postgres://") :] + if raw.startswith("sqlite:///"): + path = raw[len("sqlite:///") :] + if path.startswith("~"): + return f"sqlite:///{Path(path).expanduser()}" + return raw + + +def get_database_url() -> str: + return normalize_database_url(os.getenv("DATABASE_URL")) + + +def _configure_sqlite(engine: Engine) -> None: + @event.listens_for(engine, "connect") + def _set_sqlite_pragmas(dbapi_connection, _connection_record) -> None: # type: ignore[no-untyped-def] + cursor = dbapi_connection.cursor() + cursor.execute("PRAGMA journal_mode=WAL") + cursor.execute(f"PRAGMA busy_timeout={SQLITE_BUSY_TIMEOUT_MS}") + cursor.close() + + +def _create_engine() -> Engine: + database_url = get_database_url() + connect_args: Dict[str, Any] = {} + if database_url.startswith("sqlite://"): + connect_args = {"check_same_thread": False} + engine = create_engine(database_url, future=True, pool_pre_ping=True, connect_args=connect_args) + if database_url.startswith("sqlite://"): + _configure_sqlite(engine) + metadata.create_all(engine) + return engine + + +def get_engine() -> Engine: + global _ENGINE + if _ENGINE is not None: + return _ENGINE + with _ENGINE_LOCK: + if _ENGINE is None: + _ENGINE = _create_engine() + return _ENGINE + + +@contextmanager +def write_connection() -> Iterator[Connection]: + engine = get_engine() + conn = engine.connect() + if engine.dialect.name == "sqlite": + conn.exec_driver_sql("BEGIN IMMEDIATE") + try: + yield conn + conn.commit() + except Exception: + conn.rollback() + raise + finally: + conn.close() + return + + trans = conn.begin() + try: + yield conn + trans.commit() + except Exception: + trans.rollback() + raise + finally: + conn.close() + + +@contextmanager +def read_connection() -> Iterator[Connection]: + conn = get_engine().connect() + try: + yield conn + finally: + conn.close() + + +def row_to_dict(row: Any) -> Dict[str, Any]: + return dict(row._mapping) + + +def serialize_record(record: Dict[str, Any]) -> Dict[str, Any]: + value = dict(record) + for field in ("created_at", "updated_at"): + timestamp = value.get(field) + if isinstance(timestamp, datetime): + value[field] = timestamp.isoformat() + return value + + +def merge_json(current: Any, update: Dict[str, Any]) -> Dict[str, Any]: + base = current if isinstance(current, dict) else {} + return {**base, **update} + + +def ensure_sqlite_health() -> None: + if get_engine().dialect.name != "sqlite": + return + with read_connection() as conn: + conn.execute(text("SELECT 1")) + + +def reset_engine_for_tests() -> None: + global _ENGINE + with _ENGINE_LOCK: + if _ENGINE is not None: + _ENGINE.dispose() + _ENGINE = None diff --git a/tests/test_sql_storage.py b/tests/test_sql_storage.py new file mode 100644 index 0000000..0ab5236 --- /dev/null +++ b/tests/test_sql_storage.py @@ -0,0 +1,184 @@ +import json +import threading +from pathlib import Path + +import pytest + +import a2a_store +import timeline_store +from scripts.migrate_json_to_sql import migrate +from storage_db import normalize_database_url, reset_engine_for_tests + + +@pytest.fixture() +def sqlite_db_url(tmp_path, monkeypatch): + db_url = f"sqlite:///{tmp_path / 'xmcp.db'}" + monkeypatch.setenv("DATABASE_URL", db_url) + reset_engine_for_tests() + yield db_url + reset_engine_for_tests() + + +def test_timeline_crud_round_trip(sqlite_db_url): + item = timeline_store.add_item( + { + "user_id": "u1", + "title": "Title", + "body": "Body", + "status": "unread", + "posted_by": "agent", + "actions": ["Approve", "Reject"], + "metadata": {"x": 1}, + } + ) + + got = timeline_store.get_item(item["id"]) + assert got is not None + assert got["title"] == "Title" + assert got["metadata"] == {"x": 1} + + listed = timeline_store.list_items("u1") + assert [entry["id"] for entry in listed] == [item["id"]] + + updated = timeline_store.update_item(item["id"], {"status": "approved", "metadata": {"y": 2}}) + assert updated is not None + assert updated["status"] == "approved" + assert updated["metadata"] == {"x": 1, "y": 2} + + assert timeline_store.delete_item(item["id"]) is True + assert timeline_store.get_item(item["id"]) is None + + +def test_a2a_crud_round_trip(sqlite_db_url): + agents = a2a_store.list_agents() + assert any(agent["id"] == "mcp-orchestrator" for agent in agents) + + registered = a2a_store.register_agent( + { + "id": "custom-agent", + "name": "Custom", + "description": "desc", + "status": "online", + "endpoint": "local", + "kind": "bot", + "tags": ["custom"], + } + ) + assert registered["kind"] == "bot" + + msg = a2a_store.add_message( + { + "from": "timeline-ui", + "to": "custom-agent", + "type": "timeline_action", + "content": "approve", + "metadata": {"timeline_item_id": "item-1"}, + } + ) + assert msg["to"] == "custom-agent" + + messages = a2a_store.list_messages("custom-agent") + assert len(messages) == 1 + assert messages[0]["id"] == msg["id"] + assert messages[0]["from"] == "timeline-ui" + + +def test_concurrent_timeline_updates_do_not_lose_metadata(sqlite_db_url): + item = timeline_store.add_item({"user_id": "u1", "title": "Race", "metadata": {}}) + barrier = threading.Barrier(3) + + def writer(key: str) -> None: + barrier.wait() + for i in range(40): + timeline_store.update_item(item["id"], {"metadata": {key: i}}) + + t1 = threading.Thread(target=writer, args=("a",)) + t2 = threading.Thread(target=writer, args=("b",)) + t1.start() + t2.start() + barrier.wait() + t1.join() + t2.join() + + updated = timeline_store.get_item(item["id"]) + assert updated is not None + metadata = updated["metadata"] + assert metadata["a"] == 39 + assert metadata["b"] == 39 + + +def test_postgres_url_is_normalized(): + legacy = "postgres" + "://localhost:5432/xmcp" + normalized = "postgresql" + "://localhost:5432/xmcp" + assert normalize_database_url(legacy) == normalized + + +def test_json_to_sql_migration_is_idempotent(tmp_path, monkeypatch): + timeline_path = tmp_path / "timeline_store.json" + a2a_path = tmp_path / "a2a_store.json" + db_url = f"sqlite:///{tmp_path / 'migrated.db'}" + + timeline_payload = { + "items": [ + { + "id": "item-1", + "user_id": "default", + "title": "Proposal", + "body": "Body", + "status": "unread", + "posted_by": "agent", + "actions": ["Approve"], + "metadata": {"k": "v"}, + "created_at": "2025-01-01T00:00:00+00:00", + } + ] + } + a2a_payload = { + "agents": [ + { + "id": "agent-1", + "name": "Agent", + "description": "d", + "status": "online", + "endpoint": "e", + "kind": "agent", + "tags": ["t"], + } + ], + "messages": [ + { + "id": "msg-1", + "from": "agent-1", + "to": "timeline-ui", + "type": "info", + "content": "hello", + "metadata": {"ok": True}, + "created_at": "2025-01-01T00:00:01+00:00", + } + ], + } + timeline_path.write_text(json.dumps(timeline_payload), encoding="utf-8") + a2a_path.write_text(json.dumps(a2a_payload), encoding="utf-8") + + monkeypatch.setenv("DATABASE_URL", db_url) + monkeypatch.setenv("TIMELINE_STORE_PATH", str(timeline_path)) + monkeypatch.setenv("A2A_STORE_PATH", str(a2a_path)) + reset_engine_for_tests() + + first = migrate() + second = migrate() + + assert first["timeline_items"][0] == 1 + assert first["a2a_agents"][0] == 1 + assert first["a2a_messages"][0] == 1 + assert second["timeline_items"][0] == 0 + assert second["a2a_agents"][0] == 0 + assert second["a2a_messages"][0] == 0 + + migrated_item = timeline_store.get_item("item-1") + assert migrated_item is not None + assert migrated_item["metadata"]["k"] == "v" + + migrated_messages = a2a_store.list_messages("timeline-ui") + assert len(migrated_messages) == 1 + assert migrated_messages[0]["id"] == "msg-1" diff --git a/timeline_store.py b/timeline_store.py index 97c42b0..5559af1 100644 --- a/timeline_store.py +++ b/timeline_store.py @@ -1,59 +1,37 @@ -import json -import os -import threading import uuid -from datetime import datetime, timezone -from pathlib import Path from typing import Any, Dict, List, Optional -STORE_PATH = Path(os.getenv("TIMELINE_STORE_PATH", "~/.xmcp/timeline_store.json")).expanduser() -STORE_LOCK = threading.Lock() +from sqlalchemy import delete, insert, select, update - -def _utc_now() -> str: - return datetime.now(timezone.utc).isoformat() - - -def _ensure_store() -> None: - if STORE_PATH.exists(): - return - STORE_PATH.parent.mkdir(parents=True, exist_ok=True) - STORE_PATH.write_text(json.dumps({"items": []}, indent=2), encoding="utf-8") - - -def _read_store() -> Dict[str, Any]: - _ensure_store() - raw = STORE_PATH.read_text(encoding="utf-8") - try: - data = json.loads(raw) - except json.JSONDecodeError: - data = {"items": []} - if "items" not in data or not isinstance(data["items"], list): - data["items"] = [] - return data - - -def _write_store(data: Dict[str, Any]) -> None: - STORE_PATH.parent.mkdir(parents=True, exist_ok=True) - STORE_PATH.write_text(json.dumps(data, indent=2), encoding="utf-8") +from storage_db import ( + merge_json, + row_to_dict, + serialize_record, + timeline_items, + utc_now, + write_connection, + read_connection, +) def list_items(user_id: str, status: Optional[str] = None) -> List[Dict[str, Any]]: - with STORE_LOCK: - data = _read_store() - items = [item for item in data["items"] if item.get("user_id") == user_id] - if status: - items = [item for item in items if item.get("status") == status] - return items + query = select(timeline_items).where(timeline_items.c.user_id == user_id) + if status: + query = query.where(timeline_items.c.status == status) + query = query.order_by(timeline_items.c.created_at.desc()) + + with read_connection() as conn: + rows = conn.execute(query).fetchall() + return [serialize_record(row_to_dict(row)) for row in rows] def get_item(item_id: str) -> Optional[Dict[str, Any]]: - with STORE_LOCK: - data = _read_store() - for item in data["items"]: - if item.get("id") == item_id: - return item - return None + query = select(timeline_items).where(timeline_items.c.id == item_id) + with read_connection() as conn: + row = conn.execute(query).fetchone() + if not row: + return None + return serialize_record(row_to_dict(row)) def add_item(payload: Dict[str, Any]) -> Dict[str, Any]: @@ -66,42 +44,48 @@ def add_item(payload: Dict[str, Any]) -> Dict[str, Any]: "posted_by": payload.get("posted_by", "agent"), "actions": payload.get("actions", []), "metadata": payload.get("metadata", {}), - "created_at": _utc_now(), + "created_at": utc_now(), "updated_at": None, } - with STORE_LOCK: - data = _read_store() - data["items"].insert(0, item) - _write_store(data) - return item + with write_connection() as conn: + conn.execute(insert(timeline_items).values(**item)) + return serialize_record(item) def update_item(item_id: str, updates: Dict[str, Any]) -> Optional[Dict[str, Any]]: - with STORE_LOCK: - data = _read_store() - for item in data["items"]: - if item.get("id") != item_id: - continue - for key in ["status", "posted_by", "title", "body"]: - if key in updates and updates[key] is not None: - item[key] = updates[key] - if "metadata" in updates and isinstance(updates["metadata"], dict): - item["metadata"] = {**item.get("metadata", {}), **updates["metadata"]} - if "actions" in updates and isinstance(updates["actions"], list): - item["actions"] = updates["actions"] - item["updated_at"] = _utc_now() - _write_store(data) - return item - return None + with write_connection() as conn: + row = conn.execute( + select(timeline_items).where(timeline_items.c.id == item_id) + ).fetchone() + if not row: + return None + + current = row_to_dict(row) + changed: Dict[str, Any] = {} + + for key in ["status", "posted_by", "title", "body"]: + if key in updates and updates[key] is not None: + changed[key] = updates[key] + + if "metadata" in updates and isinstance(updates["metadata"], dict): + changed["metadata"] = merge_json(current.get("metadata"), updates["metadata"]) + + if "actions" in updates and isinstance(updates["actions"], list): + changed["actions"] = updates["actions"] + + changed["updated_at"] = utc_now() + conn.execute( + update(timeline_items) + .where(timeline_items.c.id == item_id) + .values(**changed) + ) + current.update(changed) + + return serialize_record(current) def delete_item(item_id: str) -> bool: - with STORE_LOCK: - data = _read_store() - original = len(data["items"]) - data["items"] = [item for item in data["items"] if item.get("id") != item_id] - if len(data["items"]) == original: - return False - _write_store(data) - return True + with write_connection() as conn: + result = conn.execute(delete(timeline_items).where(timeline_items.c.id == item_id)) + return result.rowcount > 0