feat: ChannelStore inbox (graph->agent) with emit/pull

This commit is contained in:
ethanfel
2026-06-28 13:20:19 +02:00
parent 3ffb1d21c3
commit 055eedee24
3 changed files with 84 additions and 0 deletions
+63
View File
@@ -0,0 +1,63 @@
import threading
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class _Slot:
text: Optional[str] = None
image_path: Optional[str] = None
turn: int = 0
@dataclass
class _Channel:
inbox: _Slot = field(default_factory=_Slot)
outbox: _Slot = field(default_factory=_Slot)
last_consumed_out_turn: int = 0
class ChannelStore:
_singleton: "ChannelStore | None" = None
def __init__(self):
self._lock = threading.Lock()
self._cond = threading.Condition(self._lock)
self._channels: dict[str, _Channel] = {}
# --- singleton plumbing ---
@classmethod
def instance(cls) -> "ChannelStore":
if cls._singleton is None:
cls._singleton = cls()
return cls._singleton
@classmethod
def reset(cls) -> None:
cls._singleton = cls()
def _chan(self, name: str) -> _Channel:
ch = self._channels.get(name)
if ch is None:
ch = _Channel()
self._channels[name] = ch
return ch
# --- inbox: graph -> agent ---
def emit(self, channel: str, text: Optional[str] = None,
image_path: Optional[str] = None) -> int:
with self._cond:
ch = self._chan(channel)
ch.inbox.turn += 1
ch.inbox.text = text
ch.inbox.image_path = image_path
self._cond.notify_all()
return ch.inbox.turn
def pull(self, channel: str) -> dict:
with self._lock:
ch = self._channels.get(channel)
if ch is None:
return {"turn": 0, "text": None, "image_path": None}
s = ch.inbox
return {"turn": s.turn, "text": s.text, "image_path": s.image_path}
View File
+21
View File
@@ -0,0 +1,21 @@
from src.bridge.store import ChannelStore
def test_emit_then_pull_returns_payload():
s = ChannelStore.instance()
s.emit("main", text="hello", image_path="/tmp/a.png")
got = s.pull("main")
assert got["text"] == "hello"
assert got["image_path"] == "/tmp/a.png"
assert got["turn"] == 1
def test_pull_unknown_channel_is_empty_not_error():
s = ChannelStore.instance()
got = s.pull("never")
assert got == {"turn": 0, "text": None, "image_path": None}
def test_emit_bumps_turn_each_time():
s = ChannelStore.instance()
s.emit("main", text="a")
s.emit("main", text="b")
assert s.pull("main")["turn"] == 2
assert s.pull("main")["text"] == "b"