8.6 KiB
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)
@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, notname(core/agent.py:4hashed on name is fragile). Message.seqmonotonic per thread replacesThread.last_readindex hacks.Threadholds nomessages:list; messages live in Store.- Multiple threads between same participants allowed (key is
id, notfrozensetlikecore/app.py:9).
Membership:
Threadis frozen;join/leavereturns newThreadvia Store, not in-place mutation (fixes commentedcore/thread.py:43approach).
4. Store Protocol (core/store.py) — dumb I/O
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)
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)
SQLiteStore(path: Path = Path.home()/".agentmsgs.db")
# PRAGMA journal_mode=WAL for concurrent opencode+codex writers
Schema:
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
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)
@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:
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.pywithInMemoryStore: 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.pywithtempfile+SQLiteStore: same ops suite + concurrent append from two handles (WAL).- Existing
tests/gng/test_usual.pydeleted withgng. - Demo in
src/agentmsgs/__init__.py:1moved toexamples/demo.pyor 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
gngremoval
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__.pydemo logic (move) - Tests: new
tests/test_ops.py,tests/test_store_sqlite.py