From 1fd72eac0bf73e4527e2f3f7fafe1f7f53aed7a0 Mon Sep 17 00:00:00 2001
From: ethanfel <ethan.fel@ts-pc.fr>
Date: Sun, 28 Jun 2026 13:20:55 +0200
Subject: [PATCH] feat: ChannelStore outbox (agent->graph) with consume +
 blocking receive

---
 src/bridge/store.py               | 28 ++++++++++++++++++++++
 tests/bridge/test_store_outbox.py | 40 +++++++++++++++++++++++++++++++
 2 files changed, 68 insertions(+)
 create mode 100644 tests/bridge/test_store_outbox.py

diff --git a/src/bridge/store.py b/src/bridge/store.py
index a79d036..4df6edc 100644
--- a/src/bridge/store.py
+++ b/src/bridge/store.py
@@ -1,4 +1,5 @@
 import threading
+import time
 from dataclasses import dataclass, field
 from typing import Optional
 
@@ -61,3 +62,30 @@ class ChannelStore:
                 return {"turn": 0, "text": None, "image_path": None}
             s = ch.inbox
             return {"turn": s.turn, "text": s.text, "image_path": s.image_path}
+
+    # --- outbox: agent -> graph ---
+    def push(self, channel: str, text: Optional[str] = None,
+             image_path: Optional[str] = None) -> int:
+        with self._cond:
+            ch = self._chan(channel)
+            ch.outbox.turn += 1
+            ch.outbox.text = text
+            ch.outbox.image_path = image_path
+            self._cond.notify_all()
+            return ch.outbox.turn
+
+    def receive(self, channel: str, wait_seconds: float = 0.0) -> dict:
+        empty = {"turn": 0, "text": None, "image_path": None}
+        deadline = time.monotonic() + max(0.0, wait_seconds)
+        with self._cond:
+            while True:
+                ch = self._channels.get(channel)
+                if ch is not None and ch.outbox.turn > ch.last_consumed_out_turn:
+                    ch.last_consumed_out_turn = ch.outbox.turn
+                    s = ch.outbox
+                    return {"turn": s.turn, "text": s.text,
+                            "image_path": s.image_path}
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    return empty
+                self._cond.wait(timeout=remaining)
diff --git a/tests/bridge/test_store_outbox.py b/tests/bridge/test_store_outbox.py
new file mode 100644
index 0000000..5840759
--- /dev/null
+++ b/tests/bridge/test_store_outbox.py
@@ -0,0 +1,40 @@
+import threading
+import time
+from src.bridge.store import ChannelStore
+
+def test_push_then_receive_consumes_once():
+    s = ChannelStore.instance()
+    s.push("main", text="result", image_path="/tmp/out.png")
+    got = s.receive("main", wait_seconds=0)
+    assert got["text"] == "result"
+    assert got["image_path"] == "/tmp/out.png"
+    # second receive with no new push -> empty (already consumed)
+    again = s.receive("main", wait_seconds=0)
+    assert again["text"] is None and again["image_path"] is None
+
+def test_receive_nonblocking_empty_channel():
+    s = ChannelStore.instance()
+    got = s.receive("nope", wait_seconds=0)
+    assert got == {"turn": 0, "text": None, "image_path": None}
+
+def test_receive_blocks_until_push():
+    s = ChannelStore.instance()
+
+    def push_later():
+        time.sleep(0.2)
+        s.push("main", text="late")
+
+    threading.Thread(target=push_later, daemon=True).start()
+    start = time.monotonic()
+    got = s.receive("main", wait_seconds=2.0)
+    elapsed = time.monotonic() - start
+    assert got["text"] == "late"
+    assert 0.15 < elapsed < 1.5  # returned shortly after the push, not at timeout
+
+def test_receive_times_out_returns_empty():
+    s = ChannelStore.instance()
+    start = time.monotonic()
+    got = s.receive("main", wait_seconds=0.3)
+    elapsed = time.monotonic() - start
+    assert got["text"] is None and got["image_path"] is None
+    assert elapsed >= 0.3