ouyangbo/EvoScientist
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 3 Wiki Activity

fix(cli): forward async-notifier replies back to originating channel (#244)

Browse Source 
* fix(cli): forward async-notifier replies back to originating channel

When PR #214's auto-notifier fires a synthetic agent turn after a
channel-originated conversation, the synthesized response only rendered
to the local CLI/TUI — the channel user (iMessage etc.) saw nothing
and had to manually re-prompt to find out what happened.

Adds a per-thread channel-origin registry in cli/channel.py and wires
the three notifier paths (Rich CLI / TUI / serve) to publish the final
response back via bus.publish_outbound when the originating thread was
started by a channel turn. Publish is fire-and-forget (scheduled on the
bus loop + done-callback for failure logging) so the notifier turn
doesn't block on the asyncio / textual event loop.

The registry is cleared on /new and /resume rotation so stale entries
don't accumulate.

* fix(cli): address review feedback on channel-origin forwarding

Follow-up to the review on #244 (din0s, X-iZhang):

- Guard the /resume origin cleanup on a real thread change in Rich CLI
  and TUI (serve mode already did via thread_changed). Resuming the
  already-active thread no longer wipes its still-live origin, which
  would otherwise silently drop a later async-notifier forward — the
  exact gap this PR closes.
- Re-bind the now-current thread to its channel after a channel-issued
  /new or /resume slash command (which rotates the thread inside the
  dispatch), so notifier turns on the rotated thread still forward.
- Guard the publish done-callback against a cancelled future, whose
  .exception() raises CancelledError (rather than returning it) on
  bus-loop teardown, so the intended warning still logs.
- Mirror the normal reply path's manager.record_message(channel, "sent")
  for forwarded notifications so per-channel stats stay accurate.
- Print the closing "[channel: Replied to ...]" line in all three
  notifier paths (Rich CLI / TUI / serve) when a forward actually
  happened, so the forwarded block reads as terminated on screen.

Adds test_publish_records_sent_metric. ruff clean; notification-origin
suite (10) + related channel/CLI/serve suites (728) pass.

* fix(cli): store sender information separately from chat_id in channel origin

---------

Co-authored-by: X-iZhang <zacharyzhang2022@gmail.com>
This commit is contained in:
Ziheng Zhang
2026-05-31 21:43:42 +08:00
committed by GitHub
parent 721a03c25b
commit 2364e6b130
5 changed files with 492 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
EvoScientist/cli/channel.py
+131
View File
@@ -447,6 +447,137 @@ _ASK_USER_TIMEOUT = (
_STOP_COMMANDS = frozenset(("/stop", "/cancel"))
# ---------------------------------------------------------------------------
# Per-thread channel-origin registry
# ---------------------------------------------------------------------------
# When a channel-originated message starts an agent turn, the turn's
# thread_id is remembered against its (channel_type, chat_id, metadata).
# Later, when an async sub-agent notification fires a synthetic agent turn
# for that same thread_id, the notifier path pushes the synthesized final
# response back to the same chat — otherwise the follow-up would only render
# locally and the channel user would never see it. v1 forwards only the
# final response (no mid-turn thinking/todo/media).
@dataclass(frozen=True)
class _ChannelOrigin:
"""Channel destination remembered for a thread, for notifier push-back."""
channel_type: str
chat_id: str
sender: str
metadata: dict | None = None
_thread_channel_origins: dict[str, _ChannelOrigin] = {}
_thread_channel_origins_lock = threading.Lock()
def remember_channel_origin(thread_id: str | None, msg: ChannelMessage) -> None:
"""Record that ``thread_id`` is currently bound to ``msg``'s channel chat.
Called on entry to each channel-triggered agent turn (Rich CLI / TUI /
serve). The latest channel turn for a given thread wins — re-registering
is intentional, since the user can keep talking on the same thread from
the same channel and we always want the most recent metadata.
"""
if not thread_id:
return
with _thread_channel_origins_lock:
_thread_channel_origins[thread_id] = _ChannelOrigin(
channel_type=msg.channel_type,
chat_id=msg.chat_id,
sender=msg.sender,
metadata=dict(msg.metadata) if msg.metadata else None,
)
def get_channel_origin(thread_id: str | None) -> _ChannelOrigin | None:
"""Return the channel origin remembered for ``thread_id``, or ``None``."""
if not thread_id:
return None
with _thread_channel_origins_lock:
return _thread_channel_origins.get(thread_id)
def forget_channel_origin(thread_id: str | None) -> None:
"""Drop the registry entry for ``thread_id`` (e.g. on ``/new`` rotation)."""
if not thread_id:
return
with _thread_channel_origins_lock:
_thread_channel_origins.pop(thread_id, None)
def publish_to_channel_origin(thread_id: str | None, content: str) -> bool:
"""Schedule pushing ``content`` to the channel remembered for ``thread_id``.
Fire-and-forget: returns ``True`` iff a publish coroutine was scheduled
on the bus loop; returns ``False`` if no origin is registered, the bus
isn't running, ``content`` is empty/whitespace, or scheduling itself
fails. The publish runs asynchronously — failures inside the coroutine
are logged via a done-callback so callers (which are often on event
loops that must not block) don't pay any latency.
"""
from ..channels.bus.events import OutboundMessage
if not content or not content.strip():
return False
origin = get_channel_origin(thread_id)
if origin is None:
return False
loop = _bus_loop
manager = _manager
if loop is None or manager is None:
return False
bus = getattr(manager, "bus", None)
if bus is None:
return False
async def _publish_and_record() -> None:
await bus.publish_outbound(
OutboundMessage(
channel=origin.channel_type,
chat_id=origin.chat_id,
content=content,
metadata=origin.metadata or {},
)
)
# Mirror the normal channel-reply path, which records a "sent"
# message after a successful publish so per-channel stats stay
# accurate for forwarded notifications too.
manager.record_message(origin.channel_type, "sent")
try:
future = asyncio.run_coroutine_threadsafe(_publish_and_record(), loop)
except Exception as exc:
_channel_logger.warning(
"Async notification publish to %s:%s failed to schedule: %s",
origin.channel_type,
origin.chat_id,
exc,
)
return False
def _on_publish_done(fut) -> None:
"""Log any exception raised by the fire-and-forget publish coroutine."""
# A cancelled future raises CancelledError from .exception() rather
# than returning it (e.g. bus loop torn down mid-publish); treat that
# as a benign shutdown, not a failure to log.
if fut.cancelled():
return
exc = fut.exception()
if exc is not None:
_channel_logger.warning(
"Async notification publish to %s:%s failed: %s",
origin.channel_type,
origin.chat_id,
exc,
)
future.add_done_callback(_on_publish_done)
return True
def _is_stop_command(content: str | None) -> bool:
"""Whether incoming content is a stop/cancel slash command."""
return (content or "").strip().lower() in _STOP_COMMANDS
EvoScientist/cli/commands.py
+25 -2
View File
@@ -37,6 +37,10 @@ from .channel import (
channel_ask_user_prompt,
channel_hitl_prompt,
dispatch_channel_slash_command,
forget_channel_origin,
get_channel_origin,
publish_to_channel_origin,
remember_channel_origin,
)
from .mcp_ui import (
_mcp_add_server_from_kwargs,
@@ -790,6 +794,7 @@ def _make_serve_start_new_session_cb(
from ..sessions import generate_thread_id
new_tid = generate_thread_id()
forget_channel_origin(agent_holder.get("thread_id"))
agent_holder["thread_id"] = new_tid
if channel_runtime is not None:
channel_runtime.thread_id = new_tid
@@ -837,6 +842,7 @@ def _make_serve_cmd_completed_hook(
new_tid = getattr(ctx, "thread_id", None)
thread_changed = bool(new_tid) and new_tid != agent_holder.get("thread_id")
if thread_changed:
forget_channel_origin(agent_holder.get("thread_id"))
agent_holder["thread_id"] = new_tid
if channel_runtime is not None:
channel_runtime.thread_id = new_tid
@@ -896,6 +902,8 @@ def _serve_process_message(
if not _claim_or_complete_channel_request(msg):
return
remember_channel_origin(agent_holder.get("thread_id"), msg)
runtime_workspace = agent_holder.get("workspace_dir") or workspace_dir
console.print(
@@ -1015,6 +1023,10 @@ def _serve_process_message(
return
if _slash_handled:
# A channel-issued /new or /resume rotates the thread inside the
# dispatch above; re-bind the now-current thread to this channel
# so async-notifier turns on it still forward back here.
remember_channel_origin(agent_holder["thread_id"], msg)
console.print(f"[dim][{msg.channel_type}] Replied to {msg.sender}[/dim]")
return
@@ -1079,18 +1091,29 @@ def _serve_drain_notifications(
# session-rebind callback), falling back to the startup value.
runtime_workspace = agent_holder.get("workspace_dir") or workspace_dir
meta = build_metadata(runtime_workspace, model)
tid = agent_holder["thread_id"]
try:
run_streaming(
response = run_streaming(
ui_backend="cli",
agent=agent_holder["agent"],
message=text,
thread_id=agent_holder["thread_id"],
thread_id=tid,
show_thinking=show_thinking,
interactive=True,
metadata=meta,
)
except Exception as exc:
_serve_logger.warning("Notification agent turn failed: %s", exc)
return
if publish_to_channel_origin(tid, response or ""):
# Mirror a normal channel turn's closing "Replied to" line so the
# forwarded notification reads as terminated in the serve log.
origin = get_channel_origin(tid)
if origin is not None:
console.print(
f"[dim][{origin.channel_type}] Replied to "
f"{origin.sender or origin.chat_id}[/dim]"
)
async def _run_notification_message_async(text: str, notifs: list) -> None:
await _aio.to_thread(_run_notification_message, text, notifs)
EvoScientist/cli/interactive.py
+26 -1
View File
@@ -604,6 +604,7 @@ def cmd_interactive(
and kick off background agent reload. The dispatch block
refreshes the status bar post-execute (symmetric with
/compact)."""
_ch_mod.forget_channel_origin(state.get("thread_id"))
if not workspace_fixed:
state["workspace_dir"] = _create_session_workspace(run_name)
state["thread_id"] = generate_thread_id()
@@ -665,6 +666,11 @@ def cmd_interactive(
console.print(f"[red]{exc}[/red]")
return
state["workspace_dir"] = workspace_dir
if thread_id != state.get("thread_id"):
# Only drop the origin on a real thread change — resuming
# the already-active thread must keep its live origin so a
# later async-notifier turn still forwards to the channel.
_ch_mod.forget_channel_origin(state.get("thread_id"))
state["thread_id"] = thread_id
state["resumed"] = True
state["status_started_at"] = datetime.now()
@@ -801,6 +807,8 @@ def cmd_interactive(
if not _ch_mod._claim_or_complete_channel_request(msg):
return
_ch_mod.remember_channel_origin(state["thread_id"], msg)
try:
# Clear the waiting ❯ prompt line
sys.stdout.write("\r\033[2K")
@@ -943,6 +951,11 @@ def cmd_interactive(
channel_runtime=channel_runtime,
)
if _slash_handled:
# A channel-issued /new or /resume rotates the thread
# inside the dispatch above; re-bind the now-current
# thread to this channel so async-notifier turns on it
# still forward back here.
_ch_mod.remember_channel_origin(state["thread_id"], msg)
_print_separator()
sys.stdout.write("\033[34;1m❯\033[0m ")
sys.stdout.flush()
@@ -1010,7 +1023,7 @@ def cmd_interactive(
console.print(line_text, style=line_style, markup=False)
meta = build_metadata(state["workspace_dir"], model)
await _refresh_status_snapshot(text, reset_streaming_text=True)
run_streaming(
response = run_streaming(
ui_backend=state["ui_backend"],
agent=await _await_agent_ready(),
message=text,
@@ -1026,6 +1039,18 @@ def cmd_interactive(
on_stream_event=_handle_stream_status_event,
status_footer_builder=_stream_status_footer,
)
_notif_tid = target_thread_id or state["thread_id"]
if _ch_mod.publish_to_channel_origin(_notif_tid, response):
# Forwarded to a channel — print the same closing
# "Replied to" line a normal channel turn shows, so the
# forwarded block reads as terminated on screen.
_origin = _ch_mod.get_channel_origin(_notif_tid)
if _origin is not None:
tx = Text()
tx.append(f"[{_origin.channel_type}: Replied to ", style="dim")
tx.append(_origin.sender or _origin.chat_id, style="cyan")
tx.append("]", style="dim")
console.print(tx)
await _refresh_status_snapshot(reset_streaming_text=True)
console.print()
_print_separator()
EvoScientist/cli/tui_interactive.py
+38 -6
View File
@@ -590,6 +590,7 @@ def run_textual_interactive(
# Clear all widgets except #welcome
self.clear_chat()
_ch_mod.forget_channel_origin(self._conversation_tid)
if not workspace_fixed:
self._workspace_dir = create_session_workspace(run_name)
self._conversation_tid = generate_thread_id()
@@ -656,6 +657,11 @@ def run_textual_interactive(
await sync_widget.cleanup()
self._workspace_dir = workspace_dir
if thread_id != self._conversation_tid:
# Only drop the origin on a real thread change — resuming the
# already-active thread must keep its live origin so a later
# async-notifier turn still forwards to the channel.
_ch_mod.forget_channel_origin(self._conversation_tid)
self._conversation_tid = thread_id
# Background reload: history renders immediately; next turn awaits.
self._start_background_agent_load(self._workspace_dir)
@@ -895,14 +901,27 @@ def run_textual_interactive(
# get viewport-follow during streaming (only after completion).
# Mark busy synchronously so the next poll tick doesn't re-enter.
self._busy = True
self._run_task = asyncio.ensure_future(
self._run_turn(
effective_tid = target_thread_id or self._conversation_tid
async def _run_and_publish() -> None:
response = await self._run_turn(
text,
skip_user_message=True,
resolve_mentions=False,
thread_id_override=target_thread_id,
)
)
if _ch_mod.publish_to_channel_origin(effective_tid, response or ""):
# Mirror a normal channel turn's closing "Replied to" line
# so the forwarded notification reads as terminated.
_origin = _ch_mod.get_channel_origin(effective_tid)
if _origin is not None:
self._append_system(
f"[{_origin.channel_type}: Replied to "
f"{_origin.sender or _origin.chat_id}]",
style="dim",
)
self._run_task = asyncio.ensure_future(_run_and_publish())
async def _read_async_tasks_tui(
self, target_thread_id: str | None
@@ -1949,9 +1968,14 @@ def run_textual_interactive(
skip_user_message: bool = False,
resolve_mentions: bool = True,
thread_id_override: str | None = None,
) -> None:
) -> str:
"""Handle a user turn: stream agent response with widgets.
Returns the final response text from ``_stream_with_widgets`` so
callers (notably the async-notifier path) can forward it onward —
e.g. push it back to the originating channel. Returns ``""`` if
the turn was cancelled or the agent failed to load.
Args:
user_text: The user's message text.
skip_user_message: If True, suppress the UserMessage widget echo
@@ -1968,6 +1992,7 @@ def run_textual_interactive(
to the live tid when ``None``.
"""
cancelled = False
response = ""
try:
self._busy = True
self._turn_started_at = datetime.now()
@@ -1992,9 +2017,9 @@ def run_textual_interactive(
try:
await self._await_agent_ready()
except Exception:
return
return ""
await self._stream_with_widgets(
response = await self._stream_with_widgets(
message_to_send,
display_text=user_text,
file_warnings=file_warnings,
@@ -2018,6 +2043,8 @@ def run_textual_interactive(
self._render_queue_indicator()
self._run_task = asyncio.ensure_future(self._run_turn(next_msg))
return response
async def _process_channel_message(self, msg: ChannelMessage) -> None:
"""Process a channel message: stream agent response and reply.
@@ -2030,6 +2057,7 @@ def run_textual_interactive(
prompt_widget = None
if not _ch_mod._claim_or_complete_channel_request(msg):
return
_ch_mod.remember_channel_origin(self._conversation_tid, msg)
try:
self._busy = True
self._turn_started_at = datetime.now()
@@ -2135,6 +2163,10 @@ def run_textual_interactive(
channel_runtime=self._channel_runtime,
)
if _slash_handled:
# A channel-issued /new or /resume rotates the thread in
# the dispatch above; re-bind the now-current thread to
# this channel so async-notifier turns on it still forward.
_ch_mod.remember_channel_origin(self._conversation_tid, msg)
return # outer finally handles _busy / widget cleanup
# Non-slash message — streams through the agent, so wait
tests/test_channel_notification_origin.py
+272
View File
@@ -0,0 +1,272 @@
"""Tests for the per-thread channel-origin registry used by the async-notifier
push-back path."""
from __future__ import annotations
import asyncio
import threading
from unittest.mock import MagicMock
import pytest
from EvoScientist.cli import channel as channel_cli
from EvoScientist.cli.channel import ChannelMessage
@pytest.fixture(autouse=True)
def _restore_channel_globals():
"""Restore bus globals + the origin registry between tests."""
original = {
"_manager": channel_cli._manager,
"_bus_loop": channel_cli._bus_loop,
"_bus_thread": channel_cli._bus_thread,
}
with channel_cli._thread_channel_origins_lock:
original_origins = dict(channel_cli._thread_channel_origins)
channel_cli._thread_channel_origins.clear()
yield
channel_cli._manager = original["_manager"]
channel_cli._bus_loop = original["_bus_loop"]
channel_cli._bus_thread = original["_bus_thread"]
with channel_cli._thread_channel_origins_lock:
channel_cli._thread_channel_origins.clear()
channel_cli._thread_channel_origins.update(original_origins)
def _make_msg(
*,
channel_type: str = "imessage",
chat_id: str = "+15551234567",
metadata: dict | None = None,
) -> ChannelMessage:
return ChannelMessage(
msg_id="msg-1",
content="hi",
sender="alice",
channel_type=channel_type,
metadata=metadata if metadata is not None else {"foo": "bar"},
chat_id=chat_id,
)
def _install_fake_bus(
monkeypatch,
) -> tuple[asyncio.AbstractEventLoop, MagicMock, threading.Thread]:
"""Spin up a real asyncio loop on a background thread + a stub bus.
Returns (loop, publish_outbound_mock, thread). The caller is responsible
for stopping the loop at the end of the test.
"""
loop = asyncio.new_event_loop()
ready = threading.Event()
def _runner():
asyncio.set_event_loop(loop)
ready.set()
loop.run_forever()
thread = threading.Thread(target=_runner, daemon=True)
thread.start()
ready.wait(timeout=2)
publish_outbound = MagicMock()
async def _publish_outbound(msg):
publish_outbound(msg)
bus = MagicMock()
bus.publish_outbound = _publish_outbound
manager = MagicMock()
manager.bus = bus
monkeypatch.setattr(channel_cli, "_bus_loop", loop)
monkeypatch.setattr(channel_cli, "_manager", manager)
return loop, publish_outbound, thread
def _stop_loop(loop: asyncio.AbstractEventLoop, thread: threading.Thread) -> None:
loop.call_soon_threadsafe(loop.stop)
thread.join(timeout=2)
loop.close()
def _wait_for_publish(
mock: MagicMock, *, expected: int = 1, timeout: float = 2.0
) -> None:
"""Block until ``mock`` has been called ``expected`` times (or timeout)."""
import time
deadline = time.time() + timeout
while mock.call_count < expected and time.time() < deadline:
time.sleep(0.01)
def test_remember_and_publish_roundtrip(monkeypatch):
loop, publish_outbound, thread = _install_fake_bus(monkeypatch)
try:
msg = _make_msg(metadata={"thread_root": "root-1"})
channel_cli.remember_channel_origin("tid-1", msg)
assert channel_cli.publish_to_channel_origin("tid-1", "hello") is True
_wait_for_publish(publish_outbound)
publish_outbound.assert_called_once()
sent = publish_outbound.call_args.args[0]
assert sent.channel == "imessage"
assert sent.chat_id == "+15551234567"
assert sent.content == "hello"
assert sent.metadata == {"thread_root": "root-1"}
assert sent.reply_to is None
finally:
_stop_loop(loop, thread)
def test_origin_remembers_sender_distinct_from_chat_id():
"""The origin stores the human-readable sender separately from chat_id, so
the notifier closer can show the same handle a normal turn shows.
Regression: iMessage exposes an internal chat id (e.g. "1") as chat_id
while the handle (phone number) lives on sender. The closer must display
sender, not chat_id, or the user sees a confusing "Replied to 1".
"""
channel_cli.remember_channel_origin(
"tid-sender",
_make_msg(chat_id="1"), # iMessage-style internal chat id
)
origin = channel_cli.get_channel_origin("tid-sender")
assert origin is not None
assert origin.chat_id == "1" # routing id preserved
assert origin.sender == "alice" # human handle preserved for display
def test_publish_records_sent_metric(monkeypatch):
"""A forwarded notification records a "sent" metric, mirroring the normal
channel-reply path (``manager.record_message``)."""
loop, _publish_outbound, thread = _install_fake_bus(monkeypatch)
try:
channel_cli.remember_channel_origin(
"tid-metric", _make_msg(channel_type="telegram")
)
assert channel_cli.publish_to_channel_origin("tid-metric", "done") is True
# record_message runs right after publish_outbound inside the same
# coroutine; sync on it before asserting.
_wait_for_publish(channel_cli._manager.record_message)
channel_cli._manager.record_message.assert_called_once_with("telegram", "sent")
finally:
_stop_loop(loop, thread)
def test_publish_returns_false_without_origin(monkeypatch):
loop, publish_outbound, thread = _install_fake_bus(monkeypatch)
try:
assert channel_cli.publish_to_channel_origin("unknown-tid", "hi") is False
publish_outbound.assert_not_called()
finally:
_stop_loop(loop, thread)
def test_publish_returns_false_when_bus_down(monkeypatch):
monkeypatch.setattr(channel_cli, "_bus_loop", None)
monkeypatch.setattr(channel_cli, "_manager", None)
channel_cli.remember_channel_origin("tid-2", _make_msg())
assert channel_cli.publish_to_channel_origin("tid-2", "hi") is False
def test_publish_returns_false_for_empty_content(monkeypatch):
loop, publish_outbound, thread = _install_fake_bus(monkeypatch)
try:
channel_cli.remember_channel_origin("tid-3", _make_msg())
assert channel_cli.publish_to_channel_origin("tid-3", "") is False
assert channel_cli.publish_to_channel_origin("tid-3", " \n ") is False
publish_outbound.assert_not_called()
finally:
_stop_loop(loop, thread)
def test_forget_origin(monkeypatch):
loop, publish_outbound, thread = _install_fake_bus(monkeypatch)
try:
channel_cli.remember_channel_origin("tid-4", _make_msg())
channel_cli.forget_channel_origin("tid-4")
assert channel_cli.publish_to_channel_origin("tid-4", "hi") is False
publish_outbound.assert_not_called()
finally:
_stop_loop(loop, thread)
def test_remember_overwrites_same_thread(monkeypatch):
loop, publish_outbound, thread = _install_fake_bus(monkeypatch)
try:
channel_cli.remember_channel_origin(
"tid-5", _make_msg(channel_type="telegram", chat_id="111")
)
channel_cli.remember_channel_origin(
"tid-5", _make_msg(channel_type="imessage", chat_id="222")
)
assert channel_cli.publish_to_channel_origin("tid-5", "hi") is True
_wait_for_publish(publish_outbound)
sent = publish_outbound.call_args.args[0]
assert sent.channel == "imessage"
assert sent.chat_id == "222"
finally:
_stop_loop(loop, thread)
def test_publish_swallows_bus_error(monkeypatch, caplog):
loop = asyncio.new_event_loop()
ready = threading.Event()
def _runner():
asyncio.set_event_loop(loop)
ready.set()
loop.run_forever()
thread = threading.Thread(target=_runner, daemon=True)
thread.start()
ready.wait(timeout=2)
async def _boom(_msg):
raise RuntimeError("bus dead")
bus = MagicMock()
bus.publish_outbound = _boom
manager = MagicMock()
manager.bus = bus
monkeypatch.setattr(channel_cli, "_bus_loop", loop)
monkeypatch.setattr(channel_cli, "_manager", manager)
channel_cli.remember_channel_origin("tid-6", _make_msg())
try:
with caplog.at_level("WARNING", logger="EvoScientist.cli.channel"):
# The publish is scheduled (returns True); the coroutine raises
# asynchronously and the done-callback logs the failure.
assert channel_cli.publish_to_channel_origin("tid-6", "hi") is True
# Give the bus thread a moment to run the coroutine + callback.
import time
deadline = time.time() + 2.0
while (
not any(
"Async notification publish" in r.message for r in caplog.records
)
and time.time() < deadline
):
time.sleep(0.01)
assert any("Async notification publish" in r.message for r in caplog.records)
finally:
_stop_loop(loop, thread)
def test_remember_with_falsy_thread_id_noop(monkeypatch):
"""Defensive: a falsy thread_id must not pollute the registry."""
msg = _make_msg()
channel_cli.remember_channel_origin(None, msg)
channel_cli.remember_channel_origin("", msg)
with channel_cli._thread_channel_origins_lock:
assert channel_cli._thread_channel_origins == {}
def test_get_channel_origin_returns_none_for_unknown():
assert channel_cli.get_channel_origin("never-registered") is None
assert channel_cli.get_channel_origin(None) is None