fix(cron): retain Bot Chat output while a CLI owner is open

Keep never-started output behind unsupported owners and drain in admission
order after release. Persist claims before execution and never replay uncertain
started turns. Existing supported-owner receipts keep their authority.

Credits 686f6c61's residual queue proposal in #100319. This is a scoped
implementation, not general retry of failed CLI subprocesses.

Native Electron before/after: CLI-owned target previously returned
SESSION_NOT_OWNED and remained empty after release/tick; now its queued
output and reply appear once in the target Bot Chat. Nested quiet CLI
message_agent delivery to a named Desktop owner also passes on base.
This commit is contained in:
teknium1
2026-09-14 15:36:32 -07:00
committed by Teknium
parent e7bd70d92a
commit 5d8390d1a4
6 changed files with 199 additions and 14 deletions
+14 -12
View File
@@ -25,12 +25,8 @@ _OWNER_KEYS = ("profile_home", "session_id", "lease_id", "live_session_id")
_TERMINAL = frozenset({"settled", "failed", "cancelled", "ambiguous"})
def find_canonical_live_owner(profile_home: Path | str) -> dict[str, Any] | None:
"""Resolve exact Bot Chat's compression tip without creating/migrating its DB.
Capability advertisement is mandatory; old Desktop/TUI processes must not
receive work they cannot consume. Registry errors propagate, failing closed.
"""
def find_canonical_owner(profile_home: Path | str) -> dict[str, Any] | None:
"""Return the exact Bot Chat tip's lease, including unsupported CLI owners."""
from hermes_cli.active_sessions import active_session_registry_snapshot
from hermes_state import SessionDB
@@ -46,12 +42,18 @@ def find_canonical_live_owner(profile_home: Path | str) -> dict[str, Any] | None
if not session_id:
return None
for entry in active_session_registry_snapshot(registry_home=home):
meta = entry.get("metadata") or {}
if (entry["session_id"] == session_id
and meta.get("bot_live_delivery_consumer") is True
and meta.get("live_session_id")):
return dict(profile_home=str(home), session_id=session_id,
lease_id=entry["lease_id"], live_session_id=meta["live_session_id"])
if entry["session_id"] == session_id:
return {**entry, "profile_home": str(home)}
return None
def find_canonical_live_owner(profile_home: Path | str) -> dict[str, Any] | None:
"""Only advertised consumers may receive owner-pinned mailbox deliveries."""
entry = find_canonical_owner(profile_home)
meta = (entry or {}).get("metadata") or {}
if entry and meta.get("bot_live_delivery_consumer") is True and meta.get("live_session_id"):
return {key: entry[key] for key in ("profile_home", "session_id", "lease_id")} | {
"live_session_id": meta["live_session_id"]}
return None