agent/onboarding_record.py

Rama main · réplica de lectura del repositorio original.

"""Where a first-run onboarding run lives: one ``onboarding_run`` key in the session's
``sessions.model_config`` (no schema change; a compression rotation carries it, see
``hermes_state_compression._CONVERSATION_MODEL_CONFIG_KEYS``).

``advance`` is the one writer after ``start``: load, ``onboarding_flow.apply``, save, and when the
run ends write the install-level flag in the root config so the next launch does not reopen it.
"""

from __future__ import annotations

import json
import logging
import time
import uuid
from collections.abc import Callable, Mapping
from typing import Any

from agent.onboarding_flow import (
    Applied, End, Event, Facts, Step, applies_to, apply, is_active, new_record,
)
from agent.onboarding_text import reminder

logger = logging.getLogger(__name__)

KEY = "onboarding_run"

# Told when a run stops being in progress: ``{run_id, stored_session_id, outcome}``. The gateway sends
# it to every window (tui_gateway/methods_onboarding.py::register).
_ended_listeners: list[Callable[[dict], None]] = []


def on_ended(listener: Callable[[dict], None]) -> None:
    _ended_listeners.append(listener)


def load(db, session_id: str) -> dict | None:
    record = db.get_session_model_config_value(session_id, KEY) if db is not None and session_id else None
    return record if isinstance(record, dict) else None


def _announce(run_id: str | None, session_id: str, outcome: str) -> None:
    ended = {"run_id": run_id or "", "stored_session_id": session_id, "outcome": outcome}
    for listener in _ended_listeners:
        listener(ended)


def _transition(db, session_id: str, event: Event) -> Step | None:
    """Apply *event* and write the result in one database transaction at the compression tip, so Skip
    setup pressed while a card answer is being applied is never overwritten by the older copy, and an id
    read before a rotation published still lands on the copy the chat reads. An ended run then closes the
    install flag (a Skip or Done retried after a failed config write repairs it), unless another run
    is still open: the flag belongs to that run, and an old chat must not end it."""
    holder: dict[str, Any] = {}

    def update(current: Any) -> Any:
        holder.clear()
        if not isinstance(current, dict):
            return current
        holder["was_active"] = is_active(current)
        holder["step"] = apply(current, event, Facts.from_stored(current["facts"]), now=time.time())
        return holder["step"].record

    db.update_conversation_model_config_value(session_id, KEY, update)
    step = holder.get("step")
    # Told before the flag write, which can fail: the record already says the run ended.
    if step is not None and holder["was_active"] and not is_active(step.record):
        _announce(step.record.get("run_id"), session_id, step.record["ended_by"])
    if step is not None and not is_active(step.record) and active_session_id(db) is None:
        _close_install_flag()
    return step


def _close_install_flag() -> None:
    from hermes_cli import config as config_mod
    from hermes_cli.onboarding_run import set_run
    from hermes_constants import get_default_hermes_root

    from agent.onboarding import PROFILE_BUILD_FLAG

    section = config_mod.read_user_config_raw(get_default_hermes_root() / "config.yaml").get("onboarding")
    section = section if isinstance(section, dict) else {}
    seen = section.get("seen") if isinstance(section.get("seen"), dict) else {}
    if section.get("run") is not False or not seen.get(PROFILE_BUILD_FLAG):
        set_run(False, mark_profile_offered=True)


def start(db, session_id: str, facts: Facts) -> dict:
    """Write a fresh run record (the row must exist) and mark the install as mid-onboarding, so a
    crash or a quit before the end runs onboarding again on the next launch (which starts over:
    ``onboarding.start`` with ``fresh_launch``)."""
    from hermes_cli.onboarding_run import set_run

    record = new_record(facts, now=time.time(), run_id=uuid.uuid4().hex)
    db.patch_session_model_config(session_id, {KEY: record})
    set_run(True)
    return record


def advance(db, session_id: str, event: Event) -> Step:
    """Apply one event to the stored run; LookupError when this session has no run."""
    step = _transition(db, session_id, event) if db is not None and session_id else None
    if step is None:
        raise LookupError(f"session {session_id} has no onboarding run")
    return step


def open_session_ids(db) -> list[str]:
    """The chats whose run is still in progress, newest first. Rotation leaves the active copy on the
    compression parent, so each id is read at its compression tip."""
    tips: set[str] = set()
    open_ids: list[str] = []
    for sid in db.session_ids_with_model_config_key(KEY):
        tip = db.resolve_resume_session_id(sid) or sid
        if tip not in tips:
            tips.add(tip)
            if is_active(load(db, tip)):
                open_ids.append(tip)
    return open_ids


def active_session_id(db) -> str | None:
    return next(iter(open_session_ids(db)), None)


def delete_chat(db, session_id: str, delete: Callable[[], bool]) -> bool:
    """Run *delete*; when it removed the chat holding the live run, end that run (``deleted``) on every
    compression copy and tell windows once. Only the compression tip holds the live run, so deleting an
    older copy ends nothing. The install flag closes before the delete when no other run is open: a
    failed flag write keeps the chat for a retry, and a refused delete sets the flag again."""
    from hermes_cli.onboarding_run import set_run

    record = load(db, session_id)
    live = is_active(record) and (db.resolve_resume_session_id(session_id) or session_id) == session_id
    last = live and all(sid == session_id for sid in open_session_ids(db))
    if last:
        _close_install_flag()
    deleted = False
    try:
        deleted = delete()
    finally:
        if last and not deleted:
            set_run(True)
    if not deleted:
        return False
    if live:
        # Each older copy is written on its own row: a tip-routed write would land every copy on the
        # newest one left.
        for sid in db.session_ids_with_model_config_key(KEY) if record.get("run_id") else ():
            if ((older := load(db, sid)) or {}).get("run_id") == record["run_id"]:
                db.patch_session_model_config(sid, {KEY: _end_deleted(older)})
        _announce(record.get("run_id"), session_id, "deleted")
    return True


def _end_deleted(current: Any) -> Any:
    if not is_active(current):
        return current
    return apply(current, End("deleted"), Facts.from_stored(current["facts"]), now=time.time()).record


def observe_tool_result(agent, name: str, args: Mapping[str, Any], result: Any) -> Any:
    """Settle a beat from a general tool's result (``set_theme``, ``manage_connections`` display,
    ``gui_tour``). Runs for every tool result; only an onboarding chat (one whose agent carries
    ``setup_beat``) reads its record. Returns *result*, with the next-step reminder added when it
    settled a beat."""
    if not applies_to(name) or "setup_beat" not in (getattr(agent, "valid_tool_names", None) or ()):
        return result
    db, session_id = getattr(agent, "_session_db", None), getattr(agent, "session_id", "") or ""
    try:
        parsed = json.loads(result) if isinstance(result, str) else result
        if isinstance(parsed, dict) and load(db, session_id) is not None:
            step = advance(db, session_id, Applied(name, dict(args or {}), parsed))
            text = reminder(step.reminder, step.record, Facts.from_stored(step.record["facts"]))
            if text and isinstance(result, str):
                return json.dumps({**parsed, "reminder": text}, ensure_ascii=False)
    except Exception:
        logger.warning("onboarding: %s result did not update the run record", name, exc_info=True)
    return result