"""Hermes Web UI -- SSE streaming engine and agent thread runner.Includes Sprint 10 cancel support via CANCEL_FLAGS."""import base64import contextlibimport contextvarsimport jsonimport loggingimport mimetypesimport osimport queueimport reimport shleximport sysimport subprocessimport threadingimport timeimport tracebackimport copyfrom pathlib import Pathfrom typing import Optional logger = logging.getLogger(__name__) from api.config import (    get_config,    STREAMS, STREAMS_LOCK, CANCEL_FLAGS, AGENT_INSTANCES, STREAM_PARTIAL_TEXT,    STREAM_REASONING_TEXT, STREAM_LIVE_TOOL_CALLS,    STREAM_GOAL_RELATED, PENDING_GOAL_CONTINUATION,    STREAM_LAST_EVENT_ID,    LOCK, SESSIONS, SESSIONS_MAX, SESSION_DIR,    _get_session_agent_lock, _set_thread_env, _clear_thread_env,    register_active_run, update_active_run, unregister_active_run,    unregister_stream_owner,    SESSION_AGENT_LOCKS, SESSION_AGENT_LOCKS_LOCK,    resolve_model_provider,    resolve_custom_provider_connection,    model_with_provider_context,    warm_models_catalog_provenance_if_cold,    load_settings,    parse_reasoning_effort,    coerce_reasoning_effort_for_model,    _main_model_request_overrides,)from api.helpers import redact_session_data, _redact_textfrom api.compression_anchor import is_context_compression_marker, visible_messages_for_anchorfrom api.compression_recovery import stamp_compression_exhausted_recoveryfrom api.metering import meterfrom api.run_journal import RunJournalWriterfrom api.todo_state import attach_todo_state, emit_todo_statefrom api.turn_journal import append_turn_journal_event_for_streamfrom api.usage import prompt_cache_hit_percentfrom api.models import (    _is_empty_partial_activity_message,    _evict_sessions_over_cap,    clear_process_wakeup_pause,    get_state_db_session_messages,    record_process_wakeup_provider_unavailable_pause,    reconciled_state_db_messages_for_session,)from api.session_ops import mark_session_title_generated, session_has_manual_title  def _session_payload_with_full_messages(session, *, tool_calls=None):    """Return compact session metadata plus the embedded full transcript.     ``Session.compact()`` may intentionally use metadata-only counts from an    index/sidebar load. A settled SSE payload that embeds ``session.messages``    must report the count of that embedded transcript, otherwise completion and    reconcile paths can mistake a complete payload for a stale short window.    """    messages = list(getattr(session, 'messages', None) or [])    raw = session.compact() | {        'messages': messages,        'message_count': len(messages),    }    attach_todo_state(raw, messages)    if tool_calls is not None:        raw['tool_calls'] = tool_calls    return raw  def _compact_for_echo_compare(value: str) -> str:    """Normalize visible stream text for duplicate echo detection."""    return re.sub(r'\s+', '', str(value or ''))  def _strip_compact_echo_suffix(value: str, suffix: str, *, search_window: int = 4096) -> tuple[str, bool]:    """Remove ``suffix`` from ``value`` when they match after whitespace folding."""    raw = str(value or '')    candidate = _compact_for_echo_compare(suffix)    if not raw or not candidate:        return raw, False    tail = raw[-max(len(str(suffix or '')) * 3, search_window):]    offset = len(raw) - len(tail)    for idx in range(len(tail) + 1):        if _compact_for_echo_compare(tail[idx:]) == candidate:            return raw[: offset + idx].rstrip(), True    return raw, False  def _redacted_session_payload_with_full_messages(session, *, tool_calls=None) -> dict | None:    """Best-effort terminal SSE session payload for already-persisted state."""    try:        return redact_session_data(            _session_payload_with_full_messages(session, tool_calls=tool_calls)        )    except Exception:        logger.debug("Failed to build redacted session payload", exc_info=True)        return None  def _cancel_event_payload(    message: str = "Cancelled by user",    *,    session: dict | None = None,) -> dict:    """Return base cancel terminal event metadata."""    payload = {        'message': message,        'type': 'cancelled',        'status': 'cancelled',    }    if session:        payload['session'] = session        payload['session_id'] = session.get('session_id')    return payload  # Global lock for os.environ writes. Per-session locks (_agent_lock) prevent# concurrent runs of the SAME session, but two DIFFERENT sessions can still# interleave their os.environ writes. This global lock serializes the env# save/restore — held only briefly across the env-mutation critical section,# NOT for the entire agent run. The agent runs outside the lock; the finally# block re-acquires to atomically restore env vars. See narrow-lock pattern# in _run_agent_streaming (line ~2719) and profile_env_for_background_worker# (api/profiles.py:715)._ENV_LOCK = threading.Lock() _KEYLESS_CUSTOM_API_KEY = "dummy-key"_STREAM_WRITEBACK_DIAG_DEFAULT_THRESHOLD_MS = 250.0 _STREAMING_CRON_PROFILE_HOME: contextvars.ContextVar[str | None] = contextvars.ContextVar(    "webui_streaming_cron_profile_home",    default=None,)_STREAMING_CRONJOB_WRAPPER_INSTALLED = False  def _stream_writeback_diag_threshold_seconds(environ=None):    if environ is None:        environ = os.environ    raw = str(        environ.get(            "HERMES_WEBUI_STREAM_WRITEBACK_DIAG_MS",            _STREAM_WRITEBACK_DIAG_DEFAULT_THRESHOLD_MS,        )    ).strip()    try:        threshold_ms = float(raw)    except (TypeError, ValueError):        threshold_ms = _STREAM_WRITEBACK_DIAG_DEFAULT_THRESHOLD_MS    if threshold_ms < 0:        return None    return threshold_ms / 1000.0  @contextlib.contextmanagerdef _stream_writeback_stage(timings, name, *, clock=time.perf_counter):    started = clock()    try:        yield    finally:        try:            timings.append((str(name), max(0.0, float(clock() - started))))        except Exception:            pass  def _log_stream_writeback_timings(    session_id,    stream_id,    timings,    started,    *,    clock=time.perf_counter,    log=logger,    environ=None,):    threshold = _stream_writeback_diag_threshold_seconds(environ=environ)    if threshold is None:        return False    try:        total_seconds = max(0.0, float(clock() - started))    except Exception:        return False    if total_seconds < threshold:        return False    parts = []    for name, elapsed in timings or []:        try:            parts.append(f"{name}={float(elapsed) * 1000.0:.1f}ms")        except Exception:            continue    log.debug(        "stream final writeback timing session=%s stream=%s total=%.1fms stages=%s",        session_id,        stream_id,        total_seconds * 1000.0,        " ".join(parts),    )    return True  def _install_streaming_cronjob_profile_wrapper() -> None:    """Wrap the agent cronjob tool so calls run under the streaming profile.     The in-chat agent run already binds per-turn contextvars for other    session-scoped state. Cron jobs are special because ``cron.jobs`` snapshots    path constants at import time, so the model-facing ``cronjob`` tool must    enter the existing WebUI cron profile context at the tool-call boundary.    That context uses the cron-specific lock and restores the module caches as    soon as the single cron tool call returns, avoiding long-lived global path    mutation for the whole agent turn.    """    global _STREAMING_CRONJOB_WRAPPER_INSTALLED    if _STREAMING_CRONJOB_WRAPPER_INSTALLED:        return    try:        from tools.registry import registry    except Exception:        logger.debug("streaming cronjob wrapper: tools registry unavailable", exc_info=True)        return     entry = registry.get_entry("cronjob")    if entry is None:        try:            import tools.cronjob_tools  # noqa: F401        except Exception:            logger.debug("streaming cronjob wrapper: cronjob tool import failed", exc_info=True)        entry = registry.get_entry("cronjob")    if entry is None:        logger.debug("streaming cronjob wrapper: cronjob tool not registered")        return    original_handler = entry.handler    if getattr(original_handler, "_webui_streaming_profile_wrapper", False):        _STREAMING_CRONJOB_WRAPPER_INSTALLED = True        return     # This relies on the agent tool executor's ``propagate_context_to_thread``    # using an unfiltered ``contextvars.copy_context()`` so this WebUI-owned    # contextvar reaches the sync cronjob handler even when the tool call runs    # on the agent's ThreadPoolExecutor worker.    def _profile_scoped_cronjob_handler(args, **kwargs):        profile_home = _STREAMING_CRON_PROFILE_HOME.get()        if not profile_home:            return original_handler(args, **kwargs)        from api.profiles import cron_profile_context_for_home        with cron_profile_context_for_home(Path(profile_home)):            return original_handler(args, **kwargs)     _profile_scoped_cronjob_handler.__dict__["_webui_streaming_profile_wrapper"] = True    _profile_scoped_cronjob_handler.__dict__["_webui_original_handler"] = original_handler    registry.register(        name=entry.name,        toolset=entry.toolset,        schema=entry.schema,        handler=_profile_scoped_cronjob_handler,        check_fn=entry.check_fn,        requires_env=entry.requires_env,        is_async=entry.is_async,        description=entry.description,        emoji=entry.emoji,        max_result_size_chars=entry.max_result_size_chars,        dynamic_schema_overrides=entry.dynamic_schema_overrides,    )    _STREAMING_CRONJOB_WRAPPER_INSTALLED = True  _PERSISTENT_MEMORY_FILES = (    ("memory", ("memories", "MEMORY.md")),    ("user", ("memories", "USER.md")),    ("soul", ("SOUL.md",)),)  def _file_signature(path: Path) -> tuple[int, int] | None:    try:        st = path.stat()        return (int(st.st_mtime_ns), int(st.st_size))    except OSError:        return None  def _persistent_state_snapshot(profile_home: str | None) -> dict:    """Capture lightweight memory/skill file signatures for save toasts."""    if not profile_home:        return {"memory": {}, "skills": {}}    root = Path(profile_home)    memory = {}    for key, parts in _PERSISTENT_MEMORY_FILES:        sig = _file_signature(root.joinpath(*parts))        if sig is not None:            memory[key] = sig    skills = {}    skills_dir = root / "skills"    try:        for skill_md in skills_dir.rglob("SKILL.md"):            try:                rel = str(skill_md.relative_to(skills_dir)).replace("\\", "/")            except ValueError:                rel = str(skill_md)            sig = _file_signature(skill_md)            if sig is not None:                skills[rel] = sig    except OSError:        pass    return {"memory": memory, "skills": skills}  def _persistent_state_changes(before: dict | None, after: dict | None) -> dict:    before = before or {"memory": {}, "skills": {}}    after = after or {"memory": {}, "skills": {}}    memory_before = before.get("memory") or {}    memory_after = after.get("memory") or {}    skills_before = before.get("skills") or {}    skills_after = after.get("skills") or {}    memory_changed = any(memory_before.get(key) != sig for key, sig in memory_after.items())    skills = []    for rel, sig in skills_after.items():        old_sig = skills_before.get(rel)        if old_sig == sig:            continue        name = Path(rel).parent.name or Path(rel).stem        skills.append({            "name": name,            "path": rel,            "action": "created" if old_sig is None else "updated",        })    return {"memory_saved": memory_changed, "skills": skills[:10]}  def _apply_profile_provider_context_to_streaming_model(    model: str | None,    provider_context: str | None,    profile_provider: str | None,    profile_default_model: str | None,) -> tuple[str | None, str | None, bool]:    """Attach profile provider context and repair stale cross-provider models."""    if provider_context or not profile_provider:        return model, provider_context, False     provider_context = profile_provider.lower()    if not profile_default_model:        return model, provider_context, False     from api.routes import _normalize_provider_id     profile_provider_normalized = _normalize_provider_id(profile_provider)    model_lower = (model or "").lower()    # Only run the bare-prefix family match on un-namespaced model ids. A custom    # namespace like "gemini_cli/..." or "claude-relay/..." merely *starts with* a    # first-party token; matching it here would clobber the model to the profile    # default on the send path (the #4278 collision — the slash-qualified branch    # below routes through the fixed _normalize_provider_id instead).    if "/" not in model_lower:        for prefix in ("gpt", "claude", "gemini"):            if model_lower.startswith(prefix):                if _normalize_provider_id(prefix) != profile_provider_normalized:                    return profile_default_model, provider_context, True                return model, provider_context, False     if "/" in model_lower:        slash_prefix = model_lower.split("/", 1)[0]        if provider_context == "openai-codex" and slash_prefix == "openai":            return profile_default_model, provider_context, True         slash_provider = _normalize_provider_id(slash_prefix)        if (            slash_provider            and slash_provider != profile_provider_normalized            and profile_provider_normalized not in {"openrouter", "custom", ""}        ):            return profile_default_model, provider_context, True     return model, provider_context, False  def _apply_profile_home_context_to_streaming_model(    model: str | None,    provider_context: str | None,    profile_home: str | None,    has_profile: bool,) -> tuple[str | None, str | None, bool]:    """Apply profile provider/model context from a profile config if present."""    if not (profile_home and has_profile and not provider_context):        return model, provider_context, False     try:        import yaml as _yaml_pp         _pp_cfg_path = Path(profile_home) / "config.yaml"        if not _pp_cfg_path.is_file():            return model, provider_context, False         _pp_cfg = _yaml_pp.safe_load(_pp_cfg_path.read_text(encoding="utf-8")) or {}        if not isinstance(_pp_cfg, dict):            return model, provider_context, False         _pp = (_pp_cfg.get("model", {}).get("provider") or "").strip()        if not _pp:            return model, provider_context, False         _pp_default = (_pp_cfg.get("model", {}).get("default") or "").strip()        return _apply_profile_provider_context_to_streaming_model(            model,            provider_context,            _pp,            _pp_default,        )    except Exception:        logger.warning("profile provider read failed", exc_info=True)        return model, provider_context, False  def _resolve_custom_provider_runtime_overrides(    resolved_provider: str | None,    resolved_api_key: str | None,    resolved_base_url: str | None,) -> tuple[str | None, str | None, str | None]:    """Return provider/key/base_url overrides for ``custom:*`` endpoints.     Hermes Agent treats named custom providers as routing hints around an    OpenAI-compatible base URL.  Local OpenAI-compatible servers often run    without authentication, so a missing key should not fail before the first    request; pass a harmless placeholder to the SDK and let the endpoint accept    it or return its own auth error.    """    if not (isinstance(resolved_provider, str) and resolved_provider.startswith("custom:")):        return resolved_provider, resolved_api_key, resolved_base_url     _cp_key, _cp_base = resolve_custom_provider_connection(resolved_provider)    if not resolved_api_key and _cp_key:        resolved_api_key = _cp_key    if not resolved_base_url and _cp_base:        resolved_base_url = _cp_base    if resolved_base_url:        # Route through the generic custom OpenAI-compatible client once the        # named provider has supplied the concrete endpoint. Keeping the        # provider as custom:<slug> would make Agent init synthesize invalid        # env-var hints like CUSTOM:SOMETHING-8000_API_KEY on keyless setups.        resolved_provider = "custom"        if not resolved_api_key:            resolved_api_key = _KEYLESS_CUSTOM_API_KEY    return resolved_provider, resolved_api_key, resolved_base_url  def _same_base_url_endpoint(url_a: str, url_b: str) -> bool:    """True if two base URLs point at the same scheme+host+port endpoint.     Used to decide whether a runtime base_url is just a normalized form of the    configured one (e.g. OpenCode-Go's ``/v1`` de-duplication on the same host)    versus a genuinely different endpoint (an explicit ``providers.<id>.base_url``    override at a different host/port that must be preserved). Path/query are    intentionally ignored — the normalization #3895 fixes is path-only.    """    from urllib.parse import urlsplit    try:        a = urlsplit((url_a or "").strip())        b = urlsplit((url_b or "").strip())    except Exception:        return False    _default_port = {"http": 80, "https": 443}    a_host = (a.hostname or "").lower()    b_host = (b.hostname or "").lower()    a_scheme = (a.scheme or "").lower()    b_scheme = (b.scheme or "").lower()    a_port = a.port or _default_port.get(a_scheme)    b_port = b.port or _default_port.get(b_scheme)    return bool(a_host) and a_host == b_host and a_scheme == b_scheme and a_port == b_port  def _runtime_preferred_base_url(    runtime_provider: dict | None,    resolved_provider: str | None,    configured_base_url: str | None,) -> str | None:    """Prefer the runtime-normalized base_url, but never override an explicit    configured endpoint that points somewhere genuinely different.     The #3895 bug was that WebUI used the *configured* base_url (which can carry a    duplicated ``/v1``) instead of the runtime provider's per-model-normalized    base_url, 404ing OpenCode-Go. But blindly preferring the runtime URL would    clobber a legitimate ``providers.<id>.base_url`` override (e.g. LM Studio at a    LAN IP, an OpenRouter mirror). So:      - no runtime URL            -> keep configured      - no configured URL         -> use runtime (all we have)      - named ``custom:`` endpoint -> configured wins (then runtime as fallback)      - same scheme+host+port     -> runtime wins (it's the normalized/corrected

Lines 1–500 of 10945. Load more