hermes-agent/gateway/session_local.py

"""Fresh local routes owned by the authenticated gateway, not by a viewer.

Source and supported launch settings are frozen per route, never process env.
"""
from __future__ import annotations

import json
import logging
import uuid
from contextlib import contextmanager

from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import BasePlatformAdapter, SendResult
from gateway.session import SessionSource
from gateway.session_contract import CANONICAL_GATEWAY_PROTOCOL
from hermes_state_runtime import RuntimeStoreError

logger = logging.getLogger(__name__)


class LocalSessionAdapter(BasePlatformAdapter):
    """Delivery stays in the canonical event stream; no second local transport."""

    def __init__(self, authority):
        super().__init__(PlatformConfig(enabled=True), Platform.LOCAL)
        self.authority = authority
        # Built outside ``_create_adapter``: without its runner a push wake skips the authority FIFO
        # and falls to a message handler this adapter never has (``deliver_wake``).
        self.gateway_runner = getattr(authority, 'runner', None)
        self.sources = {}
        self.policies = {}

    @staticmethod
    def _source_key(source):
        # A conversation its chat moved past keeps the chat's id, which every other one of that chat
        # shares: its own key names it (gateway/session_detached.py).
        from gateway.session_detached import detached_route_of
        return detached_route_of(source) or source.chat_id

    def authorize_source(self, source):
        saved = self.sources.get(self._source_key(source))
        if saved is None:
            return False
        original, identity = saved
        return (source is original and self._identity(source) == identity
                and any(live.source is source for live in self.authority.sessions.values()))

    @staticmethod
    def _identity(source):
        return source.platform, source.chat_id, source.user_id, source.profile, source.chat_type

    def register_source(self, source):
        self.sources[self._source_key(source)] = source, self._identity(source)

    def kept_source(self, source):
        """The object this route was admitted with when *source* is that same source: a closed
        conversation retired from memory comes back with it, so the results its background work
        still holds pass ``authorize_source`` (an identity check)."""
        saved = self.sources.get(self._source_key(source))
        return saved[0] if saved is not None and saved[0].to_dict() == source.to_dict() else source

    def forget(self, source, route):
        """A deleted conversation's admission source, frozen policy and post-delivery slot (a local
        route names one conversation; a detached copy keeps only its own key, its chat's stays)."""
        key = self._source_key(source)
        if self.sources.get(key, (None,))[0] is source:
            del self.sources[key]
        if key == source.chat_id and source.platform == Platform.LOCAL:
            self.policies.pop(key, None)
            self._post_delivery_callbacks.pop(route, None)

    async def connect(self, *, is_reconnect=False):
        return True

    async def disconnect(self):
        self.sources.clear()

    async def send(self, chat_id, content, reply_to=None, metadata=None):
        # TurnRunner already publishes model/tool deltas and settlement exactly once.
        return SendResult(success=True, message_id=uuid.uuid4().hex)

    async def edit_message(self, chat_id, message_id, content, *, finalize=False):
        return SendResult(success=True, message_id=message_id)

    async def _record_delivery_obligation(self, *args, **kwargs):
        # No platform ACK to lose: viewers replay the durable transcript. A ledger row would only
        # park a raw copy of the reply in state.db that no redelivery can ever use.
        return None

    async def send_typing(self, chat_id, metadata=None):
        return None

    async def get_chat_info(self, chat_id):
        return {'id': chat_id, 'type': 'dm'}

    async def send_exec_approval(self, **kwargs):
        # The existing callback owns registration in PendingControls. An ACK here
        # selects that rendering path without a plaintext fallback or a second waiter.
        return SendResult(success=True, message_id=uuid.uuid4().hex)

    async def send_clarify(self, chat_id, question, choices, clarify_id, session_key, metadata=None):
        error = self._unanswerable_clarify(chat_id)
        if error is not None:
            # Acknowledging this question would park the turn on an answer nobody can submit.
            # A structured decline releases the wait at once and skips the plain-text re-ask
            # (an ordinary failure would fall back to a send() that ACKs, and the turn would
            # wait the whole clarify timeout).
            from gateway.relay.egress import EGRESS_DECLINE_CODE
            return SendResult(success=False, error=error, error_kind='forbidden',
                              raw_response={'success': False, 'code': EGRESS_DECLINE_CODE, 'error': error})
        # Local sessions keep the base prompt: a choice question also captures typed text.
        return await super().send_clarify(chat_id, question, choices, clarify_id, session_key, metadata)

    def _unanswerable_clarify(self, chat_id):
        """Why no one can answer a question asked in this chat now, or None to ask it.

        Hosted rooms (controls expose approvals only) and forwarded A2A conversations (the
        author is a remote peer with no prompt surface) never have an answering viewer. Any
        other chat can be answered by an attached viewer holding ``session:respond``; with none
        attached, a producer turn (Bot Chat DM, plugin/heartbeat injection) declines, while a
        viewer's own turn keeps waiting so its author can re-attach and answer."""
        policy = self.policies.get(chat_id)
        if policy is None:
            return None
        from gateway.session_a2a import is_forward_policy
        if policy.source == 'bot_room':
            return 'Clarification is unavailable in hosted rooms'
        if is_forward_policy(policy):
            return 'Clarification is unavailable to a forwarded peer'
        live = self.authority.sessions.get(chat_id)
        if live is None:
            return None
        with live.event_stream.lock:
            if any('session:respond' in member.capabilities for member in live.subscribers.values()):
                return None
        from hermes_state_runtime import list_session_admissions
        started = next((row for row in list_session_admissions(self.authority.db, session_id=chat_id)
                        if row['status'] == 'started'), None)
        if started is not None and 'local_automation_v1' in started['payload']:
            return 'Clarification is unavailable: no viewer can answer this chat'
        return None


def forget_live_session(authority, session_id, live):
    """What the local adapter and the authority keyed by a deleted live session."""
    authority.launch_sources.pop(session_id, None)
    if live.source is None or live.source.platform != Platform.LOCAL:
        return
    from gateway.session_local_recovery import local_adapter_map
    local = local_adapter_map(authority).get(Platform.LOCAL)
    if local is not None:
        local.forget(live.source, live.route)


def authorize_local_source(runner, source):
    """None for legacy/nonlocal routes and history mounts; registered local adapters fail closed."""
    if source is None or source.platform != Platform.LOCAL:
        return None
    # A forged profile must not miss this adapter and fall into a messaging
    # allow-all policy. Local source identity is owned by this authority alone.
    from gateway.session_authorities import all_authorities
    owners = all_authorities(runner)
    adapter = runner._adapters_for_profile(source.profile).get(Platform.LOCAL)
    if not isinstance(adapter, LocalSessionAdapter):
        # A LOCAL source naming a profile this runtime does not serve must fail closed rather
        # than fall into a messaging allow-all policy; legacy runners (no authority) stay None.
        return False if owners else None
    # The adapter's authority must be one THIS runtime owns (the launch authority or a served
    # secondary's), never a stale/foreign object that happens to carry a LOCAL adapter.
    if not any(adapter.authority is owner for owner in owners):
        return False
    return adapter.authorize_source(source)


def _bypass_policy(params, *, private_secrets, client_local=False):
    """Frozen snapshot for --safe-mode / --ignore-user-config: code defaults plus the explicit
    launch options. The profile's config.yaml/.env are never read, so a malformed profile cannot
    block or shape the session; there is no default model to inherit, so one must be explicit."""
    import copy
    from gateway.session_policy import build_policy
    from hermes_cli.config_defaults import DEFAULT_CONFIG
    if not params.get('model'):
        raise RuntimeStoreError('invalid_params', field='model',
                                detail='a bypass launch has no profile to inherit a model from')
    defaults = copy.deepcopy(DEFAULT_CONFIG)
    if not isinstance(defaults.get('model'), dict):  # code default is the empty string
        defaults['model'] = {}
    return build_policy(params, defaults, private_secrets=private_secrets, profile_terminal=False,
                        client_local=client_local)


def load_profile_config(home=None):
    """The profile's effective config.yaml for a NEW session's policy, ``${VAR}`` refs expanded against
    the profile's CURRENT ``.env`` (the source tool env refreshes from), not the env this daemon
    started with: a single-profile owner binds no secret scope and never reloads ``.env`` for local
    turns. A bound scope (a served profile's owner scope) already is that source and is kept; a miss
    on the launch profile still falls to the process env (systemd / ``op run`` injection)."""
    from pathlib import Path
    from gateway.run import _load_gateway_config
    with profile_env_scope(home):
        return _load_gateway_config(*(() if home is None else (Path(home) / 'config.yaml',)))


@contextmanager
def profile_env_scope(home=None):
    """Bind the profile's current ``.env`` (see ``load_profile_config``) unless a scope is bound."""
    from pathlib import Path
    from agent.secret_scope import build_profile_secret_scope, current_secret_scope, reset_secret_scope, set_secret_scope
    from gateway.run import _gateway_config_home
    if current_secret_scope() is not None:
        yield
        return
    home = Path(home) if home is not None else _gateway_config_home()  # default: the active gateway home
    token = set_secret_scope(build_profile_secret_scope(home), profile_home=str(home))
    try:
        yield
    finally:
        reset_secret_scope(token)


def project_plugins_warning(policy):
    """The plain notice for a session whose client opted into project plugins (its forwarded
    ``HERMES_ENABLE_PROJECT_PLUGINS``) in a folder that has some: the gateway never loads them."""
    from pathlib import Path
    from utils import is_truthy_value
    # The client's folder is a path on this host only for a client the gateway proved local.
    env = policy.client_env() if policy is not None and not policy.client_env_remote else None
    folder = Path(policy.cwd) / '.hermes' / 'plugins' if env else None
    if not env or not is_truthy_value(env.get('HERMES_ENABLE_PROJECT_PLUGINS')) or not folder.is_dir():
        return None
    try:
        if not any(folder.iterdir()):
            return None
    except OSError:  # an unlistable folder skips the optional notice, never the session
        return None
    return (f'project plugins in {folder} are not loaded: the gateway that runs Hermes sessions does not '
            'load ./.hermes/plugins, so HERMES_ENABLE_PROJECT_PLUGINS has no effect here.')


def create_local_session(authority, actor, params, *, trusted_policy=None, trusted_secrets=None, skills_prompt=None,
                         client_local=False):
    """Stage, commit and publish a local session inline (synchronous callers)."""
    from gateway.session_local_recovery import restore_local_session
    from hermes_state_local import commit_local_session
    staged = stage_local_session(authority, actor, params, trusted_policy=trusted_policy,
                                 trusted_secrets=trusted_secrets, skills_prompt=skills_prompt,
                                 client_local=client_local)
    if isinstance(staged, dict):
        commit_local_session(authority.db, epoch=authority.epoch, receipt=staged)
        staged = restore_local_session(authority, staged['session_id'])
    return staged


def staged_creation_follows(authority, actor, request_id):
    """Whether a retry of *request_id* may ask to follow the profile.

    True when nothing is staged under that identity, or the staged request already follows.
    False only when an older build committed the session without ``follow_profile_config`` and
    stopped before its title. The next attempt reuses the identity, and a different request is
    refused for good. The retry then carries the frozen request.
    """
    if not isinstance(request_id, str) or not request_id:
        return True
    from gateway.session_local_recovery import local_identity
    from hermes_state_local import local_receipt
    sid = local_identity(authority.profile_id, actor.subject, request_id)
    if authority.db.get_session(sid) is None:
        return True
    try:
        request = json.loads(local_receipt(authority.db, sid)['policy']['request_json'])
    except (KeyError, TypeError, ValueError, RuntimeStoreError):
        return True
    return isinstance(request, dict) and request.get('follow_profile_config') is True


def stage_local_session(authority, actor, params, *, trusted_policy=None, trusted_secrets=None, skills_prompt=None,
                        client_local=False, desktop_local=False):
    """Everything before the creation commit: the existing session's ref on an exact retry, else the
    receipt ``commit_local_session`` writes (the async RPC commits it off the owner loop).
    ``client_local``: the gateway's own proof that the client is on this host (its connection redeemed a
    ticket minted over the private control socket), never a value the client sends. ``desktop_local``:
    that client is also the Desktop app (``process_identity.local_desktop_client``); a Desktop chat
    records it so its turns run as work for the local Desktop (``session_policy.local_desktop_turn``)."""
    if actor.profile_id != authority.profile_id:
        raise RuntimeStoreError('profile_mismatch')
    if 'session:create' not in actor.capabilities:
        raise RuntimeStoreError('permission_denied')
    from gateway.session_policy import build_policy, BYPASS_FIELDS, FOLLOW_ADOPTED, LOCAL_DESKTOP_CHAT
    from gateway.run import _resolve_gateway_model
    from dataclasses import replace
    private_secrets = dict(trusted_secrets or {})
    client_key = params.get('api_key')  # the client's own; an alias's key is the owner's expansion
    if trusted_policy is not None:
        policy = trusted_policy
    elif any(params.get(name) is True for name in BYPASS_FIELDS):
        policy = _bypass_policy(params, private_secrets=private_secrets, client_local=client_local)
    else:
        from gateway.session_local_route import resolve_launch_route
        raw_model, sent = params.get('model'), params
        with profile_env_scope():  # an alias's ``key_env`` / ``${VAR}`` key reads the same .env
            config = load_profile_config()
            params = resolve_launch_route(params, config)
        # A client's session.create is the one caller whose named folder may select the coding posture.
        policy = build_policy(params, config, private_secrets=private_secrets, workspace_posture=True,
                              client_local=client_local)
        if params is not sent:  # resolved, even to a model of the alias's own name
            # Owner-stamped provenance (not create fields): a remote client cannot re-resolve the typed
            # `-m` alias against its own home, so resume and an exact retry compare a repeat against the
            # name, and ``route_derived`` names the fields the route (not the client) filled.
            import json
            derived = sorted(key for key in ('provider', 'base_url') if key in params and key not in sent)
            policy = replace(policy, request_json=json.dumps(
                {**json.loads(policy.request_json), 'model_requested': raw_model, 'route_derived': derived},
                sort_keys=True))
        marks = {LOCAL_DESKTOP_CHAT: True} if desktop_local and policy.source == 'gui' else {}
        if policy.model is None:
            from hermes_cli.anon_auth import GUEST_MODEL
            policy = replace(policy, model=_resolve_gateway_model(policy.config()) or None)
            if policy.model == GUEST_MODEL:
                # Follows its profile, so the provider the user connects later replaces the trial model.
                marks[FOLLOW_ADOPTED] = True
        if marks:
            import json
            policy = replace(policy, config_json=json.dumps({**json.loads(policy.config_json), **marks}))
    request_id = params.get('request_id', uuid.uuid4().hex)
    if not isinstance(request_id, str) or not request_id or len(request_id) > 256:
        raise RuntimeStoreError('invalid_params', field='request_id',
                                detail='must be a non-empty string of at most 256 characters')
    from dataclasses import asdict
    from gateway.session import SessionEntry
    from gateway.session_lifecycle import _now
    from gateway.session_local_recovery import local_identity, restore_local_session
    authority._require_admission_open()
    sid = local_identity(authority.profile_id, actor.subject, request_id)
    from gateway.session_policy import bind_launch_key, restore_policy, launch_key
    if authority.db.get_session(sid) is not None:
        from hermes_state_local import local_receipt
        saved = local_receipt(authority.db, sid)
        original = restore_policy(saved['policy'])
        from hermes_state_local import same_create_retry
        if not same_create_retry(saved['policy'], policy.request_json, params):
            raise RuntimeStoreError('invalid_params', field='request_id',
                                    detail='a resumed request_id must carry the same frozen request')
        import hmac
        import json
        if 'model_requested' in json.loads(original.request_json):
            # An owner-resolved alias: its key came from the profile (a `.env` rotation is not a changed
            # request) and the client sent none, so one it sends now is a different request.
            if client_key is not None:
                raise RuntimeStoreError('admission_conflict')
        elif original.credential_ref is not None or params.get('api_key') is not None:
            key = launch_key(authority, original)
            if key is None or not hmac.compare_digest(key, params.get('api_key') or ''):
                raise RuntimeStoreError('admission_conflict')
        return restore_local_session(authority, sid)
    from gateway.session_policy import launch_skills
    if launch_skills(params) and skills_prompt is None:
        # `-s` is rendered off-loop by the RPC handler (render_launch_skills); a caller that
        # skipped it must not mint a session silently missing its preload.
        raise RuntimeStoreError('invalid_params', field='skills',
                                detail='the caller must render -s skills before minting the session')
    policy = replace(policy, skills_prompt=skills_prompt)
    if warning := project_plugins_warning(policy):
        logger.warning('session %s: %s', sid, warning)
    client_request = {k: params[k] for k in ('client_env',) if k in params}
    policy = bind_launch_key(authority, sid, policy, params.get("api_key"), config_secrets=private_secrets,
                             owner_key=client_key is None,  # an overlapping exact retry takes the held one
                             client_request=client_request)
    from gateway.session_local_recovery import local_source
    source = local_source(authority, sid, actor.subject)
    route = authority.runner.session_store._generate_session_key(source)
    # Same clock as every other routing entry: the recovery sweeps compare against a
    # naive cutoff, and one aware entry aborts the whole iteration.
    now = _now()
    entry = SessionEntry(route, sid, now, now, origin=source, platform=Platform.LOCAL)
    # Prefill is read now, at creation (main's CLI: at startup), and held only if this attempt commits.
    from gateway.session_worker_construct import stage_session_prefill
    staged = stage_session_prefill(authority, policy, {
        'profile_id': authority.profile_id, 'principal_id': actor.subject, 'request_id': request_id,
        'session_id': sid, 'route': route, 'entry': entry.to_dict(), 'policy': asdict(policy)})
    # Owner memory, never stored: an overlapping create's commit judges this request's own map by the
    # winner's committed rules, as a sequential retry is.
    staged.client_request = client_request
    return staged


def publish_local_policy(authority, session_id, *, evict=True):
    from hermes_state_local import local_receipt
    from gateway.session_policy import restore_policy
    live = authority.sessions[session_id]
    from gateway.session_local_recovery import local_adapter_map
    adapter = local_adapter_map(authority)[Platform.LOCAL]
    adapter.policies[live.source.chat_id] = restore_policy(local_receipt(authority.db, session_id)['policy'])
    if evict:
        authority.runner._evict_cached_agent(live.route)


def _lazy_model(authority, policy):
    """The model a not-yet-built session will run: with no provider pinned, the free tier's agent
    build pins ``nous/welcome`` (``pin_model_for_route``), so the configured default is not it."""
    import json
    model = policy.current_model(authority) if policy else None
    if (policy is None or json.loads(policy.request_json).get('model')
            or str(policy.provider or '').strip().lower() not in {'', 'auto'}):
        return model
    from gateway.session_authorities import owner_scope
    from hermes_cli.anon_auth import GUEST_MODEL, free_tier_route
    with owner_scope(authority):
        return GUEST_MODEL if free_tier_route() else model


def _reasoning_fields(agent, policy, model, authority=None, override=None):
    """Main's session-info effort/tier fields, from the built agent or (lazy) what the policy's next turn
    resolves (a following session's current profile): a resumed Desktop slice holds its reasoning pill
    pending until ``reasoning_effort`` arrives."""
    from agent.fast_mode import parse_service_tier
    from agent.voice_turn_route import session_runtime_view
    if agent is None and policy is None:
        return {}
    view = session_runtime_view(agent) if agent is not None else None
    if view is None:
        from hermes_constants import resolve_reasoning_config
        effective, next_model = policy.effective(authority, secrets=False)
        route = effective.get('model') or {}
    config = (getattr(view, 'reasoning_config', None) if view is not None
              else override if override is not None else resolve_reasoning_config(effective, next_model or ''))
    tier = (getattr(view, 'service_tier', None) if view is not None
            else parse_service_tier((effective.get('agent') or {}).get('service_tier'))) or ''
    provider = getattr(view, 'provider', None) if view is not None else route.get('provider')
    base_url = getattr(view, 'base_url', None) if view is not None else route.get('base_url')
    return effort_fields(config, tier, model, provider, base_url, getattr(view, 'api_mode', None))


def effort_fields(config, tier, model, provider, base_url, api_mode=None):
    """Session-info ``reasoning_effort`` / ``_wire`` / ``service_tier`` / ``fast`` for one resolved route."""
    from agent.fast_mode import STATIC_TIERS
    from agent.reasoning_effort import clamp_effort, route_supported_efforts
    from hermes_cli.models import resolve_fast_mode_overrides
    tier = tier or ''
    effort = ''
    if isinstance(config, dict):
        effort = 'none' if config.get('enabled') is False else str(config.get('effort') or '')
    wire = ''
    if effort and effort != 'none':
        wire = str(clamp_effort(effort, route_supported_efforts(provider, model, api_mode)) or '')
    try:
        fast = tier in STATIC_TIERS and resolve_fast_mode_overrides(
            model, provider=provider or None, base_url=base_url or None, tier=tier) is not None
    except Exception:  # health: allow BLE001 -- main's _fast_tier_applies gate: unknown means not Fast
        fast = False
    return {'reasoning_effort': effort, 'reasoning_effort_wire': wire, 'service_tier': tier, 'fast': fast}


def local_session_info(authority, ref):
    live = authority.sessions[ref.session_id]
    agent = authority.agent(ref)
    from gateway.session_authorities import served_profile_name
    from gateway.session_policy import policy_for_source
    policy = policy_for_source(authority.runner, live.source)
    if live.source is None:  # a history mount: what its stored row says
        source = authority.db.get_session(ref.session_id)['source']
    else:
        source = policy.source if policy else live.source.platform.value
    info = {'source': source,
            'model': (getattr(agent, 'model', policy.current_model(authority) if policy else None) if agent is not None
                      else _lazy_model(authority, policy)),
            'lazy': agent is None,
            'profile_id': authority.profile_id, 'desktop_protocol': CANONICAL_GATEWAY_PROTOCOL,
            # Main's TUI labels a named profile's composer (``alpha ❯``) from this field.
            'profile_name': served_profile_name(authority.profile_id),
            **({'cwd': policy.cwd} if policy else {}),
            # An ACP session's edit approval policy: every attached surface reads the same one.
            **({'edit_approval_policy': policy.edit_approval_policy} if policy and policy.source == 'acp' else {})}
    peek = getattr(authority.runner, '_peek_session_state', None)
    state = peek(live.route) if peek is not None else None
    override = state.conversation.reasoning_override if state is not None else None  # a typed /reasoning
    if agent is None and policy is None and live.source is not None and live.source.platform != Platform.LOCAL:
        # A messaging chat not built yet: what its next turn resolves (its /model and /reasoning picks
        # included), so a viewer's model and reasoning pill never wait on a field that never comes.
        from gateway.session_messaging_picks import messaging_route_info
        info.update(messaging_route_info(authority.runner, live))
    else:
        info.update(_reasoning_fields(agent, policy, info['model'], authority, override))
    if policy:
        # The creation request the route was frozen with (secrets already extracted), so a
        # resuming client can tell "same flags again" from a real override attempt.
        import json
        info['launch_request'] = json.loads(policy.request_json)
        from agent.session_client_env import NOT_LOCAL_WARNING
        # Session-start notices main's TUI already renders as ``config_warning``.
        notices = (project_plugins_warning(policy), NOT_LOCAL_WARNING if policy.client_env_remote else None)
        if warning := ' '.join(filter(None, notices)):
            info['config_warning'] = warning
    info.update(_classic_cli_fields(authority, ref, live, agent, policy))
    return info


def _classic_cli_fields(authority, ref, live, agent, policy):
    """What main's HermesCLI read off itself for /status, the status bar and the exit summary: the
    stored title and times, the session's yolo, api_mode, the loaded tool count and a masked
    credential hint (never the key). Lazy sessions report what is known without building an agent."""
    import contextlib
    stored_id = ref.session_id
    with contextlib.suppress(Exception):
        from hermes_state_local import local_receipt
        stored_id = local_receipt(authority.db, ref.session_id)['entry']['session_id'] or stored_id
    row = {}
    with contextlib.suppress(Exception):
        row = authority.db.get_session(stored_id) or {}
    yolo = bool(policy is not None and getattr(policy, 'yolo', False))
    with contextlib.suppress(Exception):
        from tools.approval import is_session_yolo_enabled
        yolo = yolo or is_session_yolo_enabled(live.route)
    api_mode = getattr(agent, 'api_mode', None) if agent is not None else None
    if api_mode is None and policy is not None:
        with contextlib.suppress(Exception):
            api_mode = (policy.effective(authority, secrets=False)[0].get('model') or {}).get('api_mode')
    hint = None
    if agent is not None and getattr(agent, 'api_key', None):
        from agent.redact import mask_secret
        hint = mask_secret(str(agent.api_key))
    tools = getattr(agent, 'tools', None) if agent is not None else None
    started = row.get('started_at')
    return {'title': row.get('title'), 'started_at': started,
            'last_active_at': row.get('last_activity_at') or started, 'yolo': yolo,
            'api_mode': api_mode or None, 'tool_count': len(tools) if isinstance(tools, list) else None,
            'credential_hint': hint}