commit 68e2a87c46079275006ef5499ed920b576003147 Author: lda Date: Tue Sep 1 01:24:36 2026 +0700 spec: ephemeral bus interface-over-file design (sqlite store, ops owns logic) diff --git a/docs/superpowers/specs/2026-09-01-agentmsgs-ephemeral-bus-design.md b/docs/superpowers/specs/2026-09-01-agentmsgs-ephemeral-bus-design.md new file mode 100644 index 0000000..080018e --- /dev/null +++ b/docs/superpowers/specs/2026-09-01-agentmsgs-ephemeral-bus-design.md @@ -0,0 +1,165 @@ +# AgentMsgs Ephemeral Bus — Design Spec +**Date:** 2026-09-01 +**Status:** Draft, approved sections 1-6 +**Context:** chat bridge `opencode <-> codex` via shared file, ephemeral agents (task 20m-7h max), no long-lived accounts. + +## 1. Goal +Replace current prototype (`src/agentmsgs/core/app.py:8` `dict[frozenset->Thread]`, `src/agentmsgs/utils.py:5` `gng`, `src/agentmsgs/core/thread.py:10` buggy unread) with a deep module where persistence is hidden behind an interface. `App` is consumed later by `FastMCP` server (out of scope) — `core/` stays framework-agnostic. + +Non-goals: TTL GC (deferred), auth, network transport, FastMCP tool definitions. + +## 2. Architecture +``` +src/agentmsgs/core/types.py # pure domain: Agent, Thread, Message (frozen) +src/agentmsgs/core/store.py # Protocol Store (dumb I/O) +src/agentmsgs/core/ops.py # domain logic (validates, owns has_unread/join/name-unique) +src/agentmsgs/core/app.py # thin facade delegating to ops +src/agentmsgs/stores/memory.py # InMemoryStore(dicts) — tests / single-process +src/agentmsgs/stores/sqlite.py # SQLiteStore — cross-CLI file sharing +``` +`App(store: Store)` is the only injection point: `App(InMemoryStore())` vs `App(SQLiteStore(Path("~/.agentmsgs.db")))`. Callers import `ops`/`App`, never a concrete store. + +`utils.py:gng` is deleted. If undirected-pair indexing is ever needed it lives privately inside `SQLiteStore`. + +## 3. Domain Types (`core/types.py`) +```python +@dataclass(frozen=True, slots=True) +class Agent: id: UUID; name: str # __hash__/__eq__ on id + +@dataclass(frozen=True, slots=True) +class Message: id: UUID; sender: Agent; content: str; seq: int; ts: datetime + +@dataclass(frozen=True) +class Thread: id: UUID; participants: frozenset[Agent]; created_at: datetime +``` +Changes from current: +- Identity on `UUID`, not `name` (`core/agent.py:4` hashed on name is fragile). +- `Message.seq` monotonic per thread replaces `Thread.last_read` index hacks. +- `Thread` holds no `messages:list`; messages live in Store. +- Multiple threads between same participants allowed (key is `id`, not `frozenset` like `core/app.py:9`). + +Membership: +- `Thread` is frozen; `join/leave` returns new `Thread` via Store, not in-place mutation (fixes commented `core/thread.py:43` approach). + +## 4. Store Protocol (`core/store.py`) — dumb I/O +```python +class Store(Protocol): + # agents + def get_or_create_agent(self, name: str) -> Agent: ... + def get_agent_by_name(self, name: str) -> Agent | None: ... + def get_agent_by_id(self, id: UUID) -> Agent | None: ... + def create_agent(self, name: str) -> Agent: ... # raises if name taken + def delete_agent(self, id: UUID) -> None: ... # explicit, per user request + def list_agents(self) -> list[Agent]: ... + + # threads + def create_thread(self, participants: set[Agent]) -> Thread: ... + def get_thread(self, id: UUID) -> Thread | None: ... + def find_threads(self, containing: set[Agent]) -> list[Thread]: ... + def add_participant(self, thread_id: UUID, agent: Agent) -> Thread: ... + def remove_participant(self, thread_id: UUID, agent: Agent) -> Thread: ... + def delete_thread(self, id: UUID) -> None: ... + + # messages / cursors + def append_message(self, thread_id: UUID, sender: Agent, content: str) -> Message: ... + def list_messages(self, thread_id: UUID, after_seq: int = 0) -> list[Message]: ... + def get_cursor(self, thread_id: UUID, agent: Agent) -> int: ... + def set_cursor(self, thread_id: UUID, agent: Agent, seq: int) -> None: ... +``` +Single `Store` keeps switch cost to one arg. It does **not** own business rules. No `has_unread` here (see ops). Logical grouping Agent/Thread/Message retained for readability; can split into 3 Protocols later with zero caller change. + +## 5. Stores + +### 5.1 InMemoryStore (`stores/memory.py`) +```python +agents: dict[UUID, Agent]; name_idx: dict[str, UUID] +threads: dict[UUID, Thread] +messages: dict[UUID, list[Message]] # thread_id -> list +cursors: dict[tuple[UUID,UUID], int] +``` +No locks. Used in unit tests. + +### 5.2 SQLiteStore (`stores/sqlite.py`) +```python +SQLiteStore(path: Path = Path.home()/".agentmsgs.db") +# PRAGMA journal_mode=WAL for concurrent opencode+codex writers +``` +Schema: +```sql +agents(id TEXT PK, name TEXT UNIQUE, created_at TEXT) +threads(id TEXT PK, created_at TEXT) +thread_participants(thread_id TEXT, agent_id TEXT, PRIMARY KEY(thread_id, agent_id)) +messages(id TEXT PK, thread_id TEXT, sender_id TEXT, seq INTEGER, content TEXT, ts TEXT) +cursors(thread_id TEXT, agent_id TEXT, seq INTEGER, PRIMARY KEY(thread_id, agent_id)) +-- seq is per-thread MAX(seq)+1 on append +-- find_threads(containing) via GROUP BY/HAVING +``` +Atomicity via SQLite transactions; no manual tmp-rename. GC/TTL deferred — no `DELETE WHERE created_at <` job now. Explicit `delete_agent`/`delete_thread` only. + +## 6. Ops (`core/ops.py`) — where logic lives +```python +def get_or_create_agent(store, name): # soft-unique: reuse id if name exists (shouldn't, not mustn't) +def create_agent(store, name): # hard: raise ValueError if name exists +def delete_agent(store, id): # remove from agents + thread_participants; threads remain unless deleted +def create_thread(store, participants): # validate len>=2, all agents exist +def join_thread(store, thread_id, agent): # validate not already in +def leave_thread(store, thread_id, agent): ... +def append_message(store, thread_id, sender, content): # validate sender in thread +def list_messages(store, thread_id, after_seq=0): ... +def mark_read(store, thread_id, agent, seq): # set_cursor, validate seq bounds +def has_unread(store, thread_id, agent, exclude_own: bool=False) -> bool: + # cursor < max_seq; if exclude_own, filter messages where sender==agent + # fixes core/thread.py:39 bug (all(m.sender==who)) +``` +All validation before Store mutation. Errors: `ValueError` (sender not in thread, <2 participants, duplicate via create_agent), `KeyError` (missing thread/agent). + +## 7. App Facade (`core/app.py`) +```python +@dataclass +class App: + store: Store = field(default_factory=InMemoryStore) + def get_or_create_agent(self, name): return ops.get_or_create_agent(self.store, name) + def delete_account(self, agent_id): return ops.delete_agent(self.store, agent_id) + def create_thread(self, *agents): return ops.create_thread(self.store, set(agents)) + def find_threads(self, *agents): return ops.find_threads(self.store, set(agents)) + def join_thread(self, tid, agent): return ops.join_thread(self.store, tid, agent) + def append(self, tid, sender, content): return ops.append_message(self.store, tid, sender, content) + def poll(self, tid, agent, after_seq=0): return ops.list_messages(self.store, tid, after_seq) + def mark_read(self, tid, agent, seq): return ops.mark_read(self.store, tid, agent, seq) + def has_unread(self, tid, agent, exclude_own=False): return ops.has_unread(self.store, tid, agent, exclude_own) +``` +Current `add_thread`/`create_thread`/`add_thread_2`/`get_thread`/`find_thread` confusion collapsed to `create_thread`/`find_threads`/`get_thread`. + +Flow example: +```python +app = App(store=SQLiteStore()) +alice = app.get_or_create_agent("opencode-agent-123") +bob = app.get_or_create_agent("codex-agent-456") +t = app.create_thread(alice, bob) +app.append(t.id, alice, "check thread.py:10") +for m in app.poll(t.id, bob): print(m.content) +app.mark_read(t.id, bob, m.seq) +app.join_thread(t.id, charlie) +app.delete_account(alice.id) # explicit cleanup, no TTL +``` + +## 8. Testing +- `tests/test_ops.py` with `InMemoryStore`: create/find/join/leave/append/poll/mark_read/has_unread (both exclude_own modes), soft-unique name, delete_account, delete_thread. +- `tests/test_store_sqlite.py` with `tempfile` + `SQLiteStore`: same ops suite + concurrent append from two handles (WAL). +- Existing `tests/gng/test_usual.py` deleted with `gng`. +- Demo in `src/agentmsgs/__init__.py:1` moved to `examples/demo.py` or removed. + +## 9. Future — FastMCP +`mcp/server.py` (not in this change) will import `App` and expose tools 1:1: `create_thread`, `join_thread`, `append`, `poll`, `has_unread`, `delete_account`. No `fastmcp` dependency in `core/`. + +## 10. Out of Scope / Deferred +- TTL auto-GC (prob not now) +- Message GC apart from `delete_thread`/`delete_agent` +- Multiple threads per exact participant set is supported by design but not stressed +- `gng` removal + +## 11. File Touches +- Modify: `core/agent.py`, `core/message.py`, `core/thread.py`, `core/app.py`, `core/ops.py`, `core/__init__.py`, `pyproject.toml` (add no new deps, sqlite3 is stdlib) +- Add: `core/types.py`, `core/store.py`, `stores/memory.py`, `stores/sqlite.py` +- Remove: `utils.py`, `src/agentmsgs/__init__.py` demo logic (move) +- Tests: new `tests/test_ops.py`, `tests/test_store_sqlite.py`