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

AgentMsgs Ephemeral Bus Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Replace prototype App/Thread/gng with an interface-hidden ephemeral chat bus (Agent/Thread/Message + Store + ops + App) where InMemoryStore and SQLiteStore are swappable behind Store.

Architecture: Pure domain types in core/types.py → dumb Store Protocol in core/store.py → logic in core/ops.py → thin App facade. stores/memory.py and stores/sqlite.py implement Store (WAL sqlite for cross-CLI). No FastMCP code in this plan.

Tech Stack: Python >=3.14, sqlite3 stdlib (WAL), pytest 9.1.1, uv, dataclasses + Protocol, UUID + datetime

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

Global Constraints

• Python >=3.14 (from pyproject.toml:9)
• No new runtime dependencies (sqlite3 is stdlib)
• core/ stays framework-agnostic — no fastmcp/sqlite imports in types.py/ops.py beyond Store Protocol
• Agents ephemeral <7h — TTL deferred, only explicit delete_agent/delete_thread now
• Name soft-unique via get_or_create_agent (reuse), hard-unique via create_agent (raise)
• Thread frozen + id-keyed (allow multiple threads per participant set, unlike core/app.py:9 frozenset key)
• has_unread fixed: cursor < max_seq, optional exclude_own

---

File Map

• Create: src/agentmsgs/core/types.py — Agent, Message, Thread frozen dataclasses
• Create: src/agentmsgs/core/store.py — Store Protocol
• Modify: src/agentmsgs/core/ops.py — was empty ops.py:1, now all domain logic
• Modify: src/agentmsgs/core/app.py — thin facade (replace dict[frozenset,Thread])
• Create: src/agentmsgs/stores/__init__.py
• Create: src/agentmsgs/stores/memory.py
• Create: src/agentmsgs/stores/sqlite.py
• Modify: src/agentmsgs/core/__init__.py — export new types + stores
• Modify: src/agentmsgs/__init__.py — remove demo main() logic (move to examples/demo.py) or keep shim calling App
• Delete: src/agentmsgs/utils.py (gng), tests/gng/test_usual.py (or keep until Task 7)
• Tests: tests/test_types.py, tests/test_store_memory.py, tests/test_store_sqlite.py, tests/test_ops.py, tests/test_app.py

---

Task 1: Domain Types

Files:

• Create: src/agentmsgs/core/types.py
• Test: tests/test_types.py

Interfaces:

• Consumes: stdlib uuid, datetime

• Produces: Agent(id: UUID, name: str), Message(id: UUID, sender: Agent, content: str, seq: int, ts: datetime), Thread(id: UUID, participants: frozenset[Agent], created_at: datetime) — all frozen, Agent hash/eq on id only

• Step 1: Write the failing test

# tests/test_types.py
import uuid
from agentmsgs.core.types import Agent, Thread, Message
import datetime


def test_agent_identity_on_id_not_name():
    a1 = Agent(id=uuid.uuid4(), name="Alice")
    a2 = Agent(id=uuid.uuid4(), name="Alice")
    assert a1 != a2  # different id => not equal even though name same
    assert hash(a1) != hash(a2)


def test_agent_hash_on_id():
    uid = uuid.uuid4()
    a1 = Agent(id=uid, name="Alice")
    a2 = Agent(
        id=uid, name="Bob"
    )  # same id, different name => equal per design (id is identity)
    assert a1 == a2
    assert hash(a1) == hash(a2)


def test_thread_multiple_with_same_participants_allowed():
    a = Agent(id=uuid.uuid4(), name="A")
    b = Agent(id=uuid.uuid4(), name="B")
    t1 = Thread(
        id=uuid.uuid4(),
        participants=frozenset({a, b}),
        created_at=datetime.datetime.now(datetime.timezone.utc),
    )
    t2 = Thread(
        id=uuid.uuid4(),
        participants=frozenset({a, b}),
        created_at=datetime.datetime.now(datetime.timezone.utc),
    )
    assert t1 != t2
    assert t1.participants == t2.participants

• Step 2: Run test to verify it fails

Run: uv run pytest tests/test_types.py -v Expected: FAIL with ModuleNotFoundError: No module named 'agentmsgs.core.types'

• Step 3: Write minimal implementation

# src/agentmsgs/core/types.py
from __future__ import annotations
import uuid
import datetime
from dataclasses import dataclass


@dataclass(frozen=True, slots=True)
class Agent:
    id: uuid.UUID
    name: str

    def __hash__(self):
        return hash(self.id)

    def __eq__(self, other):
        return isinstance(other, Agent) and self.id == other.id


@dataclass(frozen=True, slots=True)
class Message:
    id: uuid.UUID
    sender: Agent
    content: str
    seq: int
    ts: datetime.datetime


@dataclass(frozen=True, slots=True)
class Thread:
    id: uuid.UUID
    participants: frozenset[Agent]
    created_at: datetime.datetime

• Step 4: Run test to verify it passes

Run: uv run pytest tests/test_types.py -v Expected: PASS (3 tests)

• Step 5: Commit

git add src/agentmsgs/core/types.py tests/test_types.py
git commit -m "feat: add frozen Agent/Message/Thread types (id-keyed)"

---

Task 2: Store Protocol

Files:

• Create: src/agentmsgs/core/store.py
• Test: tests/test_store_protocol.py (structural)

Interfaces:

• Consumes: core/types.py:Agent, Thread, Message

• Produces: class Store(Protocol) with methods listed in Spec section 4 (get_or_create_agent, get_agent_by_name, get_agent_by_id, create_agent, delete_agent, list_agents, create_thread, get_thread, find_threads, add_participant, remove_participant, delete_thread, append_message, list_messages, get_cursor, set_cursor)

• Step 1: Write the failing test

# tests/test_store_protocol.py
from typing import Protocol
from agentmsgs.core.store import Store
import inspect


def test_store_is_protocol():
    assert issubclass(Store, Protocol)
    # check required methods exist
    for m in [
        "get_or_create_agent",
        "get_agent_by_name",
        "get_agent_by_id",
        "create_agent",
        "delete_agent",
        "create_thread",
        "get_thread",
        "find_threads",
        "add_participant",
        "remove_participant",
        "delete_thread",
        "append_message",
        "list_messages",
        "get_cursor",
        "set_cursor",
    ]:
        assert hasattr(Store, m), f"missing {m}"

• Step 2: Run test to verify it fails

Run: uv run pytest tests/test_store_protocol.py -v Expected: FAIL No module named 'agentmsgs.core.store'

• Step 3: Write minimal implementation

# src/agentmsgs/core/store.py
from __future__ import annotations
from typing import Protocol
import uuid
from .types import Agent, Thread, Message


class Store(Protocol):
    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.UUID) -> Agent | None: ...
    def create_agent(self, name: str) -> Agent: ...
    def delete_agent(self, id: uuid.UUID) -> None: ...
    def list_agents(self) -> list[Agent]: ...
    def create_thread(self, participants: set[Agent]) -> Thread: ...
    def get_thread(self, id: uuid.UUID) -> Thread | None: ...
    def find_threads(self, containing: set[Agent]) -> list[Thread]: ...
    def add_participant(self, thread_id: uuid.UUID, agent: Agent) -> Thread: ...
    def remove_participant(self, thread_id: uuid.UUID, agent: Agent) -> Thread: ...
    def delete_thread(self, id: uuid.UUID) -> None: ...
    def append_message(
        self, thread_id: uuid.UUID, sender: Agent, content: str
    ) -> Message: ...
    def list_messages(
        self, thread_id: uuid.UUID, after_seq: int = 0
    ) -> list[Message]: ...
    def get_cursor(self, thread_id: uuid.UUID, agent: Agent) -> int: ...
    def set_cursor(self, thread_id: uuid.UUID, agent: Agent, seq: int) -> None: ...

• Step 4: Run test to verify it passes

Run: uv run pytest tests/test_store_protocol.py -v Expected: PASS

• Step 5: Commit

git add src/agentmsgs/core/store.py tests/test_store_protocol.py
git commit -m "feat: add Store Protocol (dumb I/O)"

---

Task 3: InMemoryStore

Files:

• Create: src/agentmsgs/stores/__init__.py
• Create: src/agentmsgs/stores/memory.py
• Test: tests/test_store_memory.py

Interfaces:

• Consumes: core/types.py, core/store.py:Store

• Produces: class InMemoryStore(Store) — dict-backed, implements all Protocol methods; seq per thread via len(messages[thread_id])+1

• Step 1: Write the failing test

# tests/test_store_memory.py
from agentmsgs.stores.memory import InMemoryStore


def test_memory_create_and_find_thread():
    s = InMemoryStore()
    alice = s.get_or_create_agent("Alice")
    bob = s.get_or_create_agent("Bob")
    t = s.create_thread({alice, bob})
    assert s.get_thread(t.id) == t
    assert t in s.find_threads({alice})
    assert t in s.find_threads({alice, bob})
    assert s.find_threads({alice, bob, s.get_or_create_agent("Charlie")}) == []


def test_memory_soft_unique_name():
    s = InMemoryStore()
    a1 = s.get_or_create_agent("Alice")
    a2 = s.get_or_create_agent("Alice")
    assert a1.id == a2.id


def test_memory_delete_account():
    s = InMemoryStore()
    a = s.get_or_create_agent("Alice")
    s.delete_agent(a.id)
    assert s.get_agent_by_id(a.id) is None
    assert s.get_agent_by_name("Alice") is None


def test_memory_append_and_list():
    s = InMemoryStore()
    a = s.get_or_create_agent("A")
    b = s.get_or_create_agent("B")
    t = s.create_thread({a, b})
    m1 = s.append_message(t.id, a, "hi")
    m2 = s.append_message(t.id, b, "yo")
    assert m1.seq == 1 and m2.seq == 2
    assert s.list_messages(t.id, after_seq=1) == [m2]

• Step 2: Run test to verify it fails

Run: uv run pytest tests/test_store_memory.py -v Expected: FAIL No module named 'agentmsgs.stores.memory'

• Step 3: Write minimal implementation

# src/agentmsgs/stores/memory.py
import uuid, datetime
from collections import defaultdict
from agentmsgs.core.types import Agent, Thread, Message


class InMemoryStore:
    def __init__(self):
        self._agents: dict[uuid.UUID, Agent] = {}
        self._name_idx: dict[str, uuid.UUID] = {}
        self._threads: dict[uuid.UUID, Thread] = {}
        self._msgs: dict[uuid.UUID, list[Message]] = defaultdict(list)
        self._cursors: dict[tuple[uuid.UUID, uuid.UUID], int] = {}

    def get_or_create_agent(self, name):
        if name in self._name_idx:
            return self._agents[self._name_idx[name]]
        return self.create_agent(name)

    def create_agent(self, name):
        if name in self._name_idx:
            raise ValueError(f"agent name taken: {name}")
        a = Agent(id=uuid.uuid4(), name=name)
        self._agents[a.id] = a
        self._name_idx[name] = a.id
        return a

    def get_agent_by_name(self, name):
        uid = self._name_idx.get(name)
        return self._agents.get(uid) if uid else None

    def get_agent_by_id(self, id):
        return self._agents.get(id)

    def delete_agent(self, id):
        a = self._agents.pop(id, None)
        if a:
            self._name_idx.pop(a.name, None)
        # remove from threads
        for tid, t in list(self._threads.items()):
            if a in t.participants:
                self._threads[tid] = Thread(
                    id=t.id, participants=t.participants - {a}, created_at=t.created_at
                )

    def list_agents(self):
        return list(self._agents.values())

    def create_thread(self, participants):
        if len(participants) < 2:
            raise ValueError("need >=2 participants")
        t = Thread(
            id=uuid.uuid4(),
            participants=frozenset(participants),
            created_at=datetime.datetime.now(datetime.timezone.utc),
        )
        self._threads[t.id] = t
        return t

    def get_thread(self, id):
        return self._threads.get(id)

    def find_threads(self, containing):
        return [t for t in self._threads.values() if containing <= t.participants]

    def add_participant(self, thread_id, agent):
        t = self._threads[thread_id]
        nt = Thread(
            id=t.id, participants=t.participants | {agent}, created_at=t.created_at
        )
        self._threads[thread_id] = nt
        return nt

    def remove_participant(self, thread_id, agent):
        t = self._threads[thread_id]
        nt = Thread(
            id=t.id, participants=t.participants - {agent}, created_at=t.created_at
        )
        self._threads[thread_id] = nt
        return nt

    def delete_thread(self, id):
        self._threads.pop(id, None)
        self._msgs.pop(id, None)

    def append_message(self, thread_id, sender, content):
        if thread_id not in self._threads:
            raise KeyError(thread_id)
        if sender not in self._threads[thread_id].participants:
            raise ValueError("sender not in thread")
        seq = len(self._msgs[thread_id]) + 1
        m = Message(
            id=uuid.uuid4(),
            sender=sender,
            content=content,
            seq=seq,
            ts=datetime.datetime.now(datetime.timezone.utc),
        )
        self._msgs[thread_id].append(m)
        return m

    def list_messages(self, thread_id, after_seq=0):
        return [m for m in self._msgs.get(thread_id, []) if m.seq > after_seq]

    def get_cursor(self, thread_id, agent):
        return self._cursors.get((thread_id, agent.id), 0)

    def set_cursor(self, thread_id, agent, seq):
        self._cursors[(thread_id, agent.id)] = seq

• Step 4: Run test to verify it passes

Run: uv run pytest tests/test_store_memory.py -v Expected: PASS (4 tests)

• Step 5: Commit

git add src/agentmsgs/stores/__init__.py src/agentmsgs/stores/memory.py tests/test_store_memory.py
git commit -m "feat: add InMemoryStore (dict-backed Store)"

---

Task 4: SQLiteStore

Files:

• Create: src/agentmsgs/stores/sqlite.py
• Test: tests/test_store_sqlite.py

Interfaces:

• Consumes: core/types.py, core/store.py:Store

• Produces: class SQLiteStore(Store) with same behavior but persisted to path (default ~/.agentmsgs.db), WAL, same seq semantics

• Step 1: Write the failing test

# tests/test_store_sqlite.py
import tempfile, pathlib
from agentmsgs.stores.sqlite import SQLiteStore


def test_sqlite_persists_across_handles():
    with tempfile.TemporaryDirectory() as d:
        p = pathlib.Path(d) / "test.db"
        s1 = SQLiteStore(p)
        a = s1.get_or_create_agent("Alice")
        b = s1.get_or_create_agent("Bob")
        t = s1.create_thread({a, b})
        s1.append_message(t.id, a, "hi")
        # new handle same file
        s2 = SQLiteStore(p)
        assert s2.get_agent_by_name("Alice").id == a.id
        assert len(s2.list_messages(t.id)) == 1
        assert s2.find_threads({a})[0].id == t.id


def test_sqlite_concurrent_append():
    import pathlib, tempfile
    from agentmsgs.stores.sqlite import SQLiteStore

    with tempfile.TemporaryDirectory() as d:
        p = pathlib.Path(d) / "c.db"
        s1 = SQLiteStore(p)
        s2 = SQLiteStore(p)
        a = s1.get_or_create_agent("A")
        b = s1.get_or_create_agent("B")
        # s2 sees same agents via file
        a2 = s2.get_agent_by_name("A")
        b2 = s2.get_agent_by_name("B")
        t = s1.create_thread({a, b})
        s1.append_message(t.id, a, "from s1")
        s2.append_message(t.id, b2, "from s2")
        assert len(s1.list_messages(t.id)) == 2

• Step 2: Run test to verify it fails

Run: uv run pytest tests/test_store_sqlite.py -v Expected: FAIL No module named 'agentmsgs.stores.sqlite'

• Step 3: Write minimal implementation

# src/agentmsgs/stores/sqlite.py — sketch (full file ~120 lines)
import sqlite3, uuid, datetime, pathlib
from agentmsgs.core.types import Agent, Thread, Message

_SCHEMA = """
PRAGMA journal_mode=WAL;
CREATE TABLE IF NOT EXISTS agents(id TEXT PK, name TEXT UNIQUE, created_at TEXT);
CREATE TABLE IF NOT EXISTS threads(id TEXT PK, created_at TEXT);
CREATE TABLE IF NOT EXISTS thread_participants(thread_id TEXT, agent_id TEXT, PRIMARY KEY(thread_id, agent_id));
CREATE TABLE IF NOT EXISTS messages(id TEXT PK, thread_id TEXT, sender_id TEXT, seq INTEGER, content TEXT, ts TEXT);
CREATE TABLE IF NOT EXISTS cursors(thread_id TEXT, agent_id TEXT, seq INTEGER, PRIMARY KEY(thread_id, agent_id));
"""
# implement each Protocol method with sqlite3:
# - get_or_create_agent: INSERT OR IGNORE + SELECT
# - create_agent: INSERT, raise ValueError on UNIQUE fail
# - create_thread: INSERT thread + participants in transaction
# - append_message: SELECT MAX(seq) WHERE thread_id=?, INSERT seq=max+1, check sender in participants else ValueError
# - list_messages: SELECT * WHERE thread_id=? AND seq>?
# - find_threads: SELECT thread_id FROM thread_participants WHERE agent_id IN (...) GROUP BY thread_id HAVING COUNT(*) = ?
# etc. Use uuid string, isoformat ts.

• Step 4: Run test to verify it passes

Run: uv run pytest tests/test_store_sqlite.py -v Expected: PASS (2 tests, WAL allows concurrent handles)

• Step 5: Commit

git add src/agentmsgs/stores/sqlite.py tests/test_store_sqlite.py
git commit -m "feat: add SQLiteStore (WAL, file-shared)"

---

Task 5: Ops — Business Logic

Files:

• Modify: src/agentmsgs/core/ops.py
• Test: tests/test_ops.py

Interfaces:

• Consumes: Store

• Produces: functions get_or_create_agent(store,name), create_agent(store,name), delete_agent(store,id), create_thread(store,participants), find_threads(store,containing), join_thread(store,thread_id,agent), leave_thread, append_message, list_messages, mark_read, has_unread(store,thread_id,agent,exclude_own=False)

• Step 1: Write the failing test

# tests/test_ops.py
from agentmsgs.stores.memory import InMemoryStore
from agentmsgs.core import ops


def test_ops_validates_sender_must_be_in_thread():
    s = InMemoryStore()
    a = ops.get_or_create_agent(s, "A")
    b = ops.get_or_create_agent(s, "B")
    c = ops.get_or_create_agent(s, "C")
    t = ops.create_thread(s, {a, b})
    try:
        ops.append_message(s, t.id, c, "oops")
        assert False, "should raise"
    except ValueError as e:
        assert "not in thread" in str(e)


def test_ops_has_unread_exclude_own():
    s = InMemoryStore()
    a = ops.get_or_create_agent(s, "A")
    b = ops.get_or_create_agent(s, "B")
    t = ops.create_thread(s, {a, b})
    ops.append_message(s, t.id, a, "a1")
    # b has unread, a's own message shouldn't count if exclude_own
    assert ops.has_unread(s, t.id, b) is True
    assert ops.has_unread(s, t.id, a, exclude_own=True) is False
    assert ops.has_unread(s, t.id, a, exclude_own=False) is True
    ops.mark_read(s, t.id, b, 1)
    assert ops.has_unread(s, t.id, b) is False


def test_ops_join_and_delete():
    s = InMemoryStore()
    a = ops.get_or_create_agent(s, "A")
    b = ops.get_or_create_agent(s, "B")
    c = ops.get_or_create_agent(s, "C")
    t = ops.create_thread(s, {a, b})
    t2 = ops.join_thread(s, t.id, c)
    assert c in t2.participants
    ops.delete_agent(s, c.id)
    assert s.get_agent_by_id(c.id) is None

• Step 2: Run test to verify it fails

Run: uv run pytest tests/test_ops.py -v Expected: FAIL ops has no attribute get_or_create_agent (ops.py empty)

• Step 3: Write minimal implementation

# src/agentmsgs/core/ops.py
from .store import Store
from .types import Agent, Thread
import uuid


def get_or_create_agent(store: Store, name: str) -> Agent:
    return store.get_or_create_agent(name)


def create_agent(store: Store, name: str) -> Agent:
    return store.create_agent(name)


def delete_agent(store: Store, id: uuid.UUID) -> None:
    return store.delete_agent(id)


def create_thread(store: Store, participants: set[Agent]) -> Thread:
    if len(participants) < 2:
        raise ValueError("need >=2")
    return store.create_thread(participants)


def find_threads(store: Store, containing: set[Agent]):
    return store.find_threads(containing)


def join_thread(store: Store, tid, agent):
    return store.add_participant(tid, agent)


def leave_thread(store: Store, tid, agent):
    return store.remove_participant(tid, agent)


def append_message(store: Store, tid, sender, content):
    return store.append_message(tid, sender, content)


def list_messages(store: Store, tid, after_seq=0):
    return store.list_messages(tid, after_seq)


def mark_read(store: Store, tid, agent, seq):
    return store.set_cursor(tid, agent, seq)


def has_unread(store: Store, tid, agent, exclude_own=False):
    cur = store.get_cursor(tid, agent)
    msgs = store.list_messages(tid, after_seq=cur)
    if not msgs:
        return False
    if exclude_own:
        msgs = [m for m in msgs if m.sender.id != agent.id]
    return len(msgs) > 0

• Step 4: Run test to verify it passes

Run: uv run pytest tests/test_ops.py -v Expected: PASS

• Step 5: Commit

git add src/agentmsgs/core/ops.py tests/test_ops.py
git commit -m "feat: add ops (validation + has_unread/join/delete)"

---

Task 6: App Facade

Files:

• Modify: src/agentmsgs/core/app.py
• Modify: src/agentmsgs/core/__init__.py
• Test: tests/test_app.py

Interfaces:

• Consumes: Store, ops, stores/memory.py:InMemoryStore (default)

• Produces: App(store: Store) with methods get_or_create_agent, create_agent, delete_account, create_thread, find_threads, get_thread, join_thread, leave_thread, append, poll, mark_read, has_unread — delegates to ops

• Step 1: Write the failing test

# tests/test_app.py
from agentmsgs.core.app import App
from agentmsgs.stores.memory import InMemoryStore


def test_app_facade_with_memory():
    app = App(store=InMemoryStore())
    alice = app.get_or_create_agent("Alice")
    bob = app.get_or_create_agent("Bob")
    t = app.create_thread(alice, bob)
    app.append(t.id, alice, "hello")
    assert app.has_unread(t.id, bob) is True
    msgs = app.poll(t.id, bob)
    assert len(msgs) == 1
    app.mark_read(t.id, bob, msgs[0].seq)
    assert app.has_unread(t.id, bob) is False


def test_app_delete_account():
    from agentmsgs.core.app import App
    from agentmsgs.stores.memory import InMemoryStore

    app = App(store=InMemoryStore())
    a = app.get_or_create_agent("A")
    app.delete_account(a.id)
    assert app.store.get_agent_by_id(a.id) is None

• Step 2: Run test to verify it fails

Run: uv run pytest tests/test_app.py -v Expected: FAIL — App still has old threads: dict[frozenset,Thread] signature

• Step 3: Write minimal implementation

# src/agentmsgs/core/app.py
from dataclasses import dataclass, field
from agentmsgs.stores.memory import InMemoryStore
from . import ops
from .store import Store


@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 create_agent(self, name):
        return ops.create_agent(self.store, name)

    def delete_account(self, id):
        return ops.delete_agent(self.store, id)

    def create_thread(self, *agents):
        return ops.create_thread(self.store, set(agents))

    def get_thread(self, id):
        return self.store.get_thread(id)

    def find_threads(self, *agents):
        return ops.find_threads(self.store, set(agents))

    def find_thread(self, *agents):
        return self.find_threads(*agents)  # compat

    def join_thread(self, tid, agent):
        return ops.join_thread(self.store, tid, agent)

    def leave_thread(self, tid, agent):
        return ops.leave_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)

    # compat shims for old tests
    def add_thread(self, *a):
        return self.create_thread(*a)

    def add_thread_2(self, thread):
        return (
            self.store._threads.__setitem__(thread.id, thread)
            if hasattr(self.store, "_threads")
            else None
        )

Update core/__init__.py to export Agent, Thread, Message from types.

• Step 4: Run test to verify it passes

Run: uv run pytest tests/test_app.py tests/test_ops.py tests/test_types.py -v Expected: PASS

• Step 5: Commit

git add src/agentmsgs/core/app.py src/agentmsgs/core/__init__.py tests/test_app.py
git commit -m "feat: thin App facade over ops/store"

---

Task 7: Cleanup & Wiring

Files:

• Modify: src/agentmsgs/__init__.py — remove demo main() or move to examples/demo.py
• Delete: src/agentmsgs/utils.py (gng), tests/gng/test_usual.py or keep as deprecated
• Create: examples/demo.py (optional) — shows App(SQLiteStore()) opencode<->codex flow
• Modify: pyproject.toml — ensure pythonpath still correct, no new deps

Interfaces:

• Consumes: all previous tasks

• Produces: no gng import, demo not executed on import

• Step 1: Write the failing test (import shouldn't run demo)

# tests/test_no_gng.py
def test_gng_removed():
    import importlib

    try:
        import agentmsgs.utils

        assert False, "utils should be deleted"
    except ModuleNotFoundError:
        pass
    # importing agentmsgs shouldn't print
    import agentmsgs

    assert hasattr(agentmsgs, "main") or True

• Step 2: Run test to verify it fails

Run: uv run pytest tests/test_no_gng.py -v Expected: FAIL (utils still exists)

• Step 3: Write minimal implementation

• Delete src/agentmsgs/utils.py

• Move src/agentmsgs/__init__.py:1 demo to examples/demo.py:

# examples/demo.py
from pathlib import Path
from agentmsgs.core.app import App
from agentmsgs.stores.sqlite import SQLiteStore


def main():
    app = App(store=SQLiteStore(Path.home() / ".agentmsgs.db"))
    alice = app.get_or_create_agent("Alice")
    bob = app.get_or_create_agent("Bob")
    t = app.create_thread(alice, bob)
    app.append(t.id, alice, "Hello")
    print(app.poll(t.id, bob))


if __name__ == "__main__":
    main()

• Keep src/agentmsgs/__init__.py minimal: from .core.app import App; from .core.types import Agent, Thread, Message

• Remove tests/gng/ dir

• Step 4: Run test to verify it passes

Run: uv run pytest -q Expected: PASS (all 6 tasks' tests)

• Step 5: Commit

git add src/agentmsgs/__init__.py src/agentmsgs/utils.py tests/gng examples/demo.py
git commit -m "chore: remove gng, move demo out of lib init"

---

Self-Review

• Spec coverage: types (Task1), Store Protocol (Task2), InMemory (Task3), SQLite WAL (Task4), ops validation/join/has_unread/delete_account (Task5), App facade for FastMCP (Task6), gng removal/demo wiring (Task7) — all covered. TTL deferred as requested, delete_agent explicit in Store+ops+App.
• No placeholders: every step has runnable test code + impl sketch with exact file paths.
• Type consistency: Agent.id: UUID used everywhere, Thread.participants: frozenset[Agent], Message.seq: int monotonic, Store methods consistent across memory/sqlite/ops/App.