mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
32909145a2
Main's classic CLI stamped its own row: `end_session(sid, "cli_close")` on quit and `new_session` / `resumed_other` / `branched` on a switch. On the gateway only ACP rows were ever ended (the last viewer leaving an idle ACP session), so a CLI quit left every row open. `session.detach` takes an optional `end_reason` (snake_case, 1-32 chars). When the detaching viewer was the last subscriber, the owner ends the idle local row with that reason in the same idle-guarded write ACP uses (`end_idle_local_session`: a queued/started admission or a running worker keeps it open); a viewer that re-attached meanwhile reopens it. ACP rows keep `acp_disconnect`; a detach without `end_reason` is unchanged (only a viewer leaving). Generic name, so the TUI can close with it too. Contracts regenerated. LEDGER: A-072, DB-WRITE-007, DB-WRITE-011, DB-WRITE-014, SLASH-E2-060 (gateway half). (cherry picked from commit 8e99979da6ba08027a42ae0b86d095466fd17bc8)
73 lines
3.6 KiB
Python
73 lines
3.6 KiB
Python
"""ACP v0.9 has no per-session destroy: the last viewer leaving an idle ACP session ends it (#118216).
|
|
|
|
The in-process adapter stamped ``ended_at`` at its stdio shutdown; under the gateway the ACP process
|
|
is only a viewer, so the owner stamps it when the last subscription goes or the gateway stops. Without
|
|
a writer, source='acp' rows stay open forever and the ended-session guard keeps prune/archive away.
|
|
A later ``session/load`` + prompt reopens the row (``session_local_recovery.reopen_local_session``).
|
|
"""
|
|
import logging
|
|
|
|
from hermes_state_local import POLICY_PREFIX, end_idle_local_session, local_receipt
|
|
from hermes_state_runtime import RuntimeStoreError
|
|
|
|
ACP_END_REASON = 'acp_disconnect'
|
|
|
|
|
|
def end_idle_acp_session(authority, session_id):
|
|
"""End *session_id* iff it is an ACP-created local session with no viewer and an idle FIFO."""
|
|
live = authority.sessions.get(session_id)
|
|
if live is not None and live.subscribers:
|
|
return False
|
|
try:
|
|
receipt = local_receipt(authority.db, session_id)
|
|
except RuntimeStoreError:
|
|
return False # not a local session: no ACP policy to end
|
|
if receipt.get('policy', {}).get('source') != 'acp':
|
|
return False
|
|
try:
|
|
return bool(end_idle_local_session(authority.db, epoch=authority.epoch, session_id=session_id,
|
|
target_id=receipt['entry']['session_id'], reason=ACP_END_REASON))
|
|
except (RuntimeStoreError, KeyError, TypeError):
|
|
logging.getLogger(__name__).warning('ACP session %s not ended', session_id, exc_info=True)
|
|
return False
|
|
|
|
|
|
def end_idle_viewer_session(authority, session_id, reason):
|
|
"""``session.detach {end_reason}`` from the last viewer: end *session_id*'s idle local row with the
|
|
viewer's own reason (main's classic CLI ``end_session(sid, 'cli_close')`` and its switch reasons).
|
|
ACP rows keep their own reason; a busy FIFO keeps the row open, as for ACP."""
|
|
live = authority.sessions.get(session_id)
|
|
if live is not None and live.subscribers:
|
|
return False
|
|
try:
|
|
receipt = local_receipt(authority.db, session_id)
|
|
except RuntimeStoreError:
|
|
return False # not a local session: nothing this verb may end
|
|
if receipt.get('policy', {}).get('source') == 'acp':
|
|
return end_idle_acp_session(authority, session_id)
|
|
try:
|
|
return bool(end_idle_local_session(authority.db, epoch=authority.epoch, session_id=session_id,
|
|
target_id=receipt['entry']['session_id'], reason=reason))
|
|
except (RuntimeStoreError, KeyError, TypeError):
|
|
logging.getLogger(__name__).warning('local session %s not ended', session_id, exc_info=True)
|
|
return False
|
|
|
|
|
|
async def reopen_if_reattached(authority, session_id, ended):
|
|
"""On the owner loop after the close committed: a viewer attached meanwhile, so the session
|
|
it watches is live again (the stamp would make active work prune/archive eligible)."""
|
|
live = authority.sessions.get(session_id)
|
|
if ended and live is not None and live.subscribers:
|
|
import asyncio
|
|
target = local_receipt(authority.db, session_id)['entry']['session_id']
|
|
await asyncio.to_thread(authority.db.reopen_session, target)
|
|
return ended
|
|
|
|
|
|
def end_idle_acp_sessions(authority):
|
|
"""Gateway shutdown: every ACP session nobody is viewing, attached this run or not."""
|
|
with authority.db._read_ctx() as conn:
|
|
ids = [row[0][len(POLICY_PREFIX):] for row in conn.execute(
|
|
'SELECT key FROM state_meta WHERE key LIKE ?', (POLICY_PREFIX + '%',))]
|
|
return sum(end_idle_acp_session(authority, sid) for sid in ids)
|