Something went wrong. Try again.
Hermes WebUI: The best way to use Hermes Agent from the web or from your phone!
Something went wrong. Try again.
4.0 kB · 118 lines
Python
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119"""Hermes Web UI -- Optional state.db sync bridge.
Mirrors WebUI session metadata (token usage, title, model) into thehermes-agent state.db so that /insights, session lists, and costtracking include WebUI activity.
This is opt-in via the 'sync_to_insights' setting (default: off).All operations are wrapped in try/except -- if state.db is unavailable,locked, or the schema doesn't match, the WebUI continues normally.
The bridge uses absolute token counts (not deltas) because the WebUISession object already accumulates totals across turns. This avoidsany double-counting risk."""import loggingimport osfrom pathlib import Path
logger = logging.getLogger(__name__)
def _get_state_db(): """Get a SessionDB instance for the active profile's state.db. Returns None if hermes_state is not importable or DB is unavailable. Each caller is responsible for calling db.close() when done. """ try: from hermes_state import SessionDB except ImportError: return None
try: from api.profiles import get_active_hermes_home hermes_home = Path(get_active_hermes_home()).expanduser().resolve() except Exception: logger.debug("Failed to resolve hermes home, using default") hermes_home = Path(os.getenv('HERMES_HOME', str(Path.home() / '.hermes')))
db_path = hermes_home / 'state.db' if not db_path.exists(): return None
try: return SessionDB(db_path) except Exception: logger.debug("Failed to open state.db") return None
def sync_session_start(session_id: str, model=None) -> None: """Register a WebUI session in state.db (idempotent). Called when a session's first message is sent. """ db = _get_state_db() if not db: return try: db.ensure_session( session_id=session_id, source='webui', model=model, ) except Exception: logger.debug("Failed to sync session start to state.db") finally: try: db.close() except Exception: logger.debug("Failed to close state.db")
def sync_session_usage(session_id: str, input_tokens: int=0, output_tokens: int=0, estimated_cost=None, model=None, title: str=None, message_count: int=None) -> None: """Update token usage and title for a WebUI session in state.db. Called after each turn completes. Uses absolute=True to set totals (the WebUI Session already accumulates across turns). """ db = _get_state_db() if not db: return try: # Ensure session exists first (idempotent) db.ensure_session(session_id=session_id, source='webui', model=model) # Set absolute token counts db.update_token_counts( session_id=session_id, input_tokens=input_tokens, output_tokens=output_tokens, estimated_cost_usd=estimated_cost, model=model, absolute=True, ) # Update title if we have one, using the public API if title: try: db.set_session_title(session_id, title) except Exception: logger.debug("Failed to sync session title to state.db") # Update message count if message_count is not None: try: def _set_msg_count(conn): conn.execute( "UPDATE sessions SET message_count = ? WHERE id = ?", (message_count, session_id), ) db._execute_write(_set_msg_count) except Exception: logger.debug("Failed to sync message count to state.db") except Exception: logger.debug("Failed to sync session usage to state.db") finally: try: db.close() except Exception: logger.debug("Failed to close state.db")