agentmsgs/docs/superpowers/specs/2026-09-01-agentmsgs-ephemeral-bus-design.md

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, 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

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.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