Skip to content

fix(buzz): keep progress heartbeats and background-delegation completions in the originating thread #80651

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rbaset5 wants to merge 2 commits into
base: main
Choose a base branch
from
+90 −26
Open

fix(buzz): keep progress heartbeats and background-delegation completions in the originating thread #80651

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
58a1398
fix(gateway): anchor Buzz progress heartbeats to the triggering message
rbaset5 Aug 6, 2026
9e25991
fix(delegation): carry the reply anchor through async delegation comp…
rbaset5 Aug 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 42 additions & 25 deletions gateway/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24373,32 +24373,41 @@ def _generic_status_phrase(kind: str, *, tool_name: str | None = None, preview:
and not source.thread_id
else None
)
_progress_metadata = (
self._thread_metadata_for_source(source, event_message_id)
if _progress_thread_id == source.thread_id
else self._thread_metadata_for_target(
source.platform,
source.chat_id,
_progress_thread_id,
chat_type=getattr(source, "chat_type", None),
reply_to_message_id=event_message_id,
)
) if _progress_thread_id else None
if _progress_metadata is None and _relay_prospective_thread_id:
# No real thread yet, but the connector will auto-thread on the
# reply anchor; carry it so progress joins that thread.
_progress_metadata = {"reply_to_message_id": event_message_id}
_progress_metadata = _non_conversational_metadata(_progress_metadata, platform=source.platform)
_progress_reply_to = (
event_message_id
if (
source.platform in (Platform.FEISHU, Platform.MATTERMOST)
and source.thread_id
and event_message_id
if source.platform == Platform("buzz") and event_message_id:
# Buzz has no channel-level thread_id — every reply is anchored
# per-message via a Nostr "e" reply tag (mirrors the
# _status_thread_metadata buzz branch further below). Without
# this, the progress heartbeat ("Working — N min") posted flat/
# top-level while only the final answer threaded correctly.
_progress_metadata = {"thread_id": event_message_id}
_progress_reply_to = event_message_id
else:
_progress_metadata = (
self._thread_metadata_for_source(source, event_message_id)
if _progress_thread_id == source.thread_id
else self._thread_metadata_for_target(
source.platform,
source.chat_id,
_progress_thread_id,
chat_type=getattr(source, "chat_type", None),
reply_to_message_id=event_message_id,
)
) if _progress_thread_id else None
if _progress_metadata is None and _relay_prospective_thread_id:
# No real thread yet, but the connector will auto-thread on the
# reply anchor; carry it so progress joins that thread.
_progress_metadata = {"reply_to_message_id": event_message_id}
_progress_metadata = _non_conversational_metadata(_progress_metadata, platform=source.platform)
_progress_reply_to = (
event_message_id
if (
source.platform in (Platform.FEISHU, Platform.MATTERMOST)
and source.thread_id
and event_message_id
)
or _relay_prospective_thread_id
else None
)
or _relay_prospective_thread_id
else None
)

async def write_tool_log():
"""Drain log_queue and append tool-call lines to tool_calls.log.
Expand Down Expand Up @@ -24505,6 +24514,14 @@ async def write_tool_log():
"thread_id": _progress_thread_id,
"reply_to_message_id": event_message_id,
}
elif source.platform == Platform("buzz") and event_message_id:
# Buzz has no channel-level thread_id concept — every reply is
# anchored per-message via a Nostr "e" reply tag. Without this,
# interim/progress bubbles carry no reply_to at all and post as
# new top-level messages while only the final answer threads
# correctly (the adapter's send() reads metadata["thread_id"]
# as its reply-to fallback — see plugins/platforms/buzz/adapter.py).
_status_thread_metadata = {"thread_id": event_message_id}
else:
_status_thread_metadata = (
self._thread_metadata_for_source(source, event_message_id)
Expand Down
49 changes: 48 additions & 1 deletion tools/async_delegation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,7 +206,13 @@ def _persist_dispatch(record: Dict[str, Any]) -> None:
owner_started_at = None
task_payload = {
key: record.get(key)
for key in ("goal", "goals", "context", "toolsets", "role", "model", "is_batch")
for key in (
"goal", "goals", "context", "toolsets", "role", "model", "is_batch",
# Reply anchor — no dedicated column, so it rides along in
# task_json purely to survive into recover_abandoned_delegations()
# if the owning process dies before a normal completion.
"origin_thread_id", "origin_message_id",
)
if key in record
}
with _DB_LOCK, _transaction() as conn:
Expand Down Expand Up @@ -326,6 +332,8 @@ def recover_abandoned_delegations() -> int:
"goals": task.get("goals"), "context": task.get("context"),
"toolsets": task.get("toolsets"), "role": task.get("role"),
"model": task.get("model"), "is_batch": bool(task.get("is_batch")),
"thread_id": task.get("origin_thread_id", ""),
"message_id": task.get("origin_message_id", ""),
"status": "unknown", "summary": None,
"error": "Delegation owner exited before recording a terminal result; outcome unknown.",
"dispatched_at": dispatched_at, "completed_at": now,
Expand Down Expand Up @@ -734,6 +742,21 @@ def dispatch_async_delegation(
"""
delegation_id = _new_delegation_id()
dispatched_at = time.time()
# Captured on the parent thread BEFORE dispatch — the daemon worker won't
# carry the contextvar (same reasoning as session_key above). Without
# this, the completion notification that re-enters the chat when the
# child finishes has no anchor to the originating message/thread and
# platforms with per-message reply semantics (e.g. Buzz's Nostr "e" tags)
# start a brand-new disconnected thread instead of continuing the one the
# delegation was dispatched from. Mirrors terminal_tool.py's
# notify_on_complete/watch_patterns capture of the same two vars.
try:
from gateway.session_context import get_session_env as _gse
_origin_thread_id = _gse("HERMES_SESSION_THREAD_ID", "")
_origin_message_id = _gse("HERMES_SESSION_MESSAGE_ID", "")
except Exception:
_origin_thread_id = ""
_origin_message_id = ""
record: Dict[str, Any] = {
"delegation_id": delegation_id,
"goal": goal,
Expand All @@ -745,6 +768,8 @@ def dispatch_async_delegation(
"origin_ui_session_id": origin_ui_session_id,
"origin_session_id": origin_session_id,
"parent_session_id": parent_session_id,
"origin_thread_id": _origin_thread_id,
"origin_message_id": _origin_message_id,
"status": "running",
"dispatched_at": dispatched_at,
"completed_at": None,
Expand Down Expand Up @@ -892,6 +917,13 @@ def _push_completion_event(
"origin_ui_session_id": record.get("origin_ui_session_id", ""),
"origin_session_id": record.get("origin_session_id", ""),
"parent_session_id": record.get("parent_session_id"),
# Reply anchor captured at dispatch time (see dispatch_async_delegation) —
# lets _inject_watch_notification stamp the synthetic re-entry message
# with the same id the originating turn was replying to, so per-message
# reply platforms (Buzz) chain into the existing thread instead of
# starting a new one.
"thread_id": record.get("origin_thread_id", ""),
"message_id": record.get("origin_message_id", ""),
"goal": record.get("goal", ""),
"context": record.get("context"),
"toolsets": record.get("toolsets"),
Expand Down Expand Up @@ -973,6 +1005,15 @@ def dispatch_async_delegation_batch(
combined_goal = (
goals[0] if n == 1 else f"{n} parallel subagents: " + "; ".join(g[:40] for g in goals)
)
# Captured on the parent thread BEFORE dispatch — see the identical
# capture (and its rationale) in dispatch_async_delegation above.
try:
from gateway.session_context import get_session_env as _gse
_origin_thread_id = _gse("HERMES_SESSION_THREAD_ID", "")
_origin_message_id = _gse("HERMES_SESSION_MESSAGE_ID", "")
except Exception:
_origin_thread_id = ""
_origin_message_id = ""
record: Dict[str, Any] = {
"delegation_id": delegation_id,
"goal": combined_goal,
Expand All @@ -985,6 +1026,8 @@ def dispatch_async_delegation_batch(
"origin_ui_session_id": origin_ui_session_id,
"origin_session_id": origin_session_id,
"parent_session_id": parent_session_id,
"origin_thread_id": _origin_thread_id,
"origin_message_id": _origin_message_id,
"status": "running",
"dispatched_at": dispatched_at,
"completed_at": None,
Expand Down Expand Up @@ -1097,6 +1140,10 @@ def _push_batch_completion_event(
"origin_ui_session_id": event_record.get("origin_ui_session_id", ""),
"origin_session_id": event_record.get("origin_session_id", ""),
"parent_session_id": event_record.get("parent_session_id"),
# See _push_completion_event — same reply-anchor plumbing for the
# batch completion path.
"thread_id": event_record.get("origin_thread_id", ""),
"message_id": event_record.get("origin_message_id", ""),
"goal": event_record.get("goal", ""),
"goals": event_record.get("goals"),
"context": event_record.get("context"),
Expand Down