hart_intelligence_entry.py
Rama main · réplica de lectura del repositorio original.
# Fix Windows encoding for non-ASCII characters (Telugu, emojis, etc.)
import sys
# Windows: use the SelectorEventLoop policy process-wide. autobahn's asyncio
# WAMP transport breaks on the default Windows ProactorEventLoop — the
# EventBus WAMP bridge (core/platform/events.py connect_wamp) opens the TCP
# socket to the relay but the session never joins (on_join never fires), so
# the internet relay stays dark on Windows while a macOS/Linux peer connects
# fine. autobahn resolves its loop through the global policy, so this must be
# set before any loop is created; hypercorn serves cleanly on a selector loop.
if sys.platform == 'win32':
import asyncio as _asyncio_boot
_asyncio_boot.set_event_loop_policy(_asyncio_boot.WindowsSelectorEventLoopPolicy())
from core.subprocess_safe import run_bounded
from core.platform_paths import under_test as _under_test
import io
if sys.platform == 'win32' and not _under_test():
# Force UTF-8 encoding for stdout/stderr to prevent crashes with non-ASCII characters
# Skip when running under pytest — pytest wraps stdout/stderr for capture,
# and replacing them here closes pytest's file handles.
# Skip in cx_Freeze windowed builds — sys.stdout.buffer is closed (no console),
# and TextIOWrapper on a closed buffer raises ValueError at import time.
# Only re-wrap a stream that is NOT already UTF-8. The hasattr('buffer')
# test above is NOT sufficient to skip frozen builds: app.py installs a
# _CappedStream as sys.stdout/stderr, and its __getattr__ delegates every
# unknown attribute to the inner file handle -- so `hasattr(.., 'buffer')`
# is True and `.buffer` hands back the INNER handle. Wrapping that inner
# handle escapes the wrapper: when _CappedStream._rotate() later closes and
# reopens the inner handle at the 20MB cap, this TextIOWrapper is left
# holding the CLOSED one, and every write after that raises
# "ValueError: I/O operation on closed file" for the rest of the process.
# The try/except below cannot catch it either -- at import time the buffer
# is still open, so the wrap succeeds and the failure lands hours later.
#
# Measured 2026-08-29 on the installed build: frozen_debug.log rotated at
# 08:37:09, and both langchain tool turns after it (08:58:17, 09:04:27)
# died in warnings.warn -> sys.stderr.write during LLMChain's deprecation
# warning, so /chat 500'd and fell back to a tool-less '_tier: direct'.
# This is the consumer that task #621 predicted but never located.
#
# _CappedStream and the devnull fallbacks are all opened encoding='utf-8',
# so skipping them loses nothing; a real cp1252 Windows console still gets
# wrapped, which is the only case this code was written for.
def _needs_utf8_wrap(stream):
if stream is None or not hasattr(stream, 'buffer'):
return False
enc = (getattr(stream, 'encoding', '') or '').lower().replace('-', '')
return enc != 'utf8'
if _needs_utf8_wrap(sys.stdout):
try:
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
except (ValueError, OSError):
pass
if _needs_utf8_wrap(sys.stderr):
try:
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
except (ValueError, OSError):
pass
# ── Defang importlib.metadata.packages_distributions() before transformers ──
# transformers/utils/import_utils.py:45 calls
# PACKAGE_DISTRIBUTION_MAPPING = importlib.metadata.packages_distributions()
# at module-import time, with NO guard. The function walks every
# *.dist-info/METADATA in site-packages and accesses
# `dist.metadata['Name']`. ANY corrupt dist-info (METADATA missing,
# Name: header malformed, or pip rewriting it concurrently) raises
# KeyError('Name') → transformers crashes its import → langchain →
# hart_intelligence_entry crashes → Tier-1 HARTOS init dies → agent
# daemon never starts → admin Agent Dashboard stays empty for the rest
# of the process lifetime (Tier-1 init is one-shot).
#
# Regression observed 2026-04-26 23:05:23 in dev .venv (transient
# corrupt dist-info). The fix monkey-patches the stdlib function with
# a try/except wrapper so a single bad dist-info no longer cascades to
# a full HARTOS Tier-1 outage. Running BEFORE any langchain /
# transformers import in this module guarantees transformers picks up
# the wrapped version.
import importlib.metadata as _md_safe
_orig_pd = getattr(_md_safe, 'packages_distributions', None)
if _orig_pd is not None and not getattr(_orig_pd, '_hartos_guarded', False):
def _safe_packages_distributions():
try:
return _orig_pd()
except Exception:
# Best-effort fallback: empty mapping. Auto-docstring
# lookups against {} return generic guesses (worse but
# nonfatal).
return {}
_safe_packages_distributions._hartos_guarded = True
_md_safe.packages_distributions = _safe_packages_distributions
del _md_safe
# ── Defang transformers `_LazyModule.__getattr__` recursion ──
# transformers ships a `_LazyModule` class whose `__getattr__` does
# ``hasattr(self, candidate_name)`` which itself triggers
# ``__getattr__`` again on the not-yet-bound attribute → ~1500-frame
# recursion that holds the `transformers` per-package import lock for
# minutes on cold disk. Triggered by ANY downstream
# ``from transformers import GPT2TokenizerFast`` (langchain_core does
# this transitively). Live thread dump 2026-04-28 22:18 captured the
# hartos-init thread in 28+ frames of this recursion, blocking every
# concurrent caller of ``get_world_model_bridge`` (15 dashboard
# requests + peer_discovery + telemetry) for the rest of the process
# lifetime.
#
# Fix: import `GPT2TokenizerFast` from its underlying submodule and
# write it to ``transformers.__dict__`` BEFORE any lazy lookup fires.
# Once the attribute is bound, `hasattr()` short-circuits at the dict
# and never calls `__getattr__`. Every downstream caller gets the
# bound class via plain dict lookup, no recursion.
#
# This used to live in Nunba's `app.py::_prewarm_hartos_chain` (commit
# c0894a2a, 2026-04-28). That prewarm was deleted in favour of the
# `hartos_bootstrap` facade — but the direct-bind got deleted with it,
# letting the recursion regress. The right home for this is HARTOS,
# not Nunba: it must run before HARTOS's own
# `from langchain_classic.llms import OpenAI` (next stmt below) AND
# before any consumer's deferred bootstrap touches transformers.
try:
import transformers as _tf_safe
from transformers.models.gpt2.tokenization_gpt2_fast import (
GPT2TokenizerFast as _GPT2TokenizerFast_direct,
)
if not hasattr(_tf_safe, '__dict__') or \
'GPT2TokenizerFast' not in _tf_safe.__dict__:
_tf_safe.__dict__['GPT2TokenizerFast'] = _GPT2TokenizerFast_direct
del _tf_safe, _GPT2TokenizerFast_direct
except Exception:
# If transformers isn't installed or moved the symbol, fall through.
# Worker threads will hit the lazy path and pay the recursion once;
# bad but not fatal (and surfaced via the hartos_init_error.log).
import logging as _early_logging
_early_logging.getLogger(__name__).warning(
"transformers GPT2 tokenizer defang skipped (symbol absent/moved)", exc_info=True)
# ── Defang transformers `_LazyModule.__getattr__` re-entry recursion ──
#
# Symptom: Tier-1 init thread spends minutes (and burns the GIL the
# whole time, starving Hypercorn workers) inside
# ``transformers.utils.import_utils.py:2215`` —
#
# transformers_module = sys.modules.get("transformers")
# if transformers_module and hasattr(transformers_module, candidate_name):
# ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
#
# That ``hasattr(transformers_module, candidate_name)`` triggers
# ``transformers.__getattr__(candidate_name)`` (because transformers'
# top-level module object IS a ``_LazyModule`` with a custom
# ``__getattr__``), which hits the SAME ``hasattr`` line on a different
# candidate, which recurses again. 130+ frames deep, all
# ``active+gil``, no other thread can run. The recursion is
# bounded (SLOW_TO_FAST_CONVERTERS is finite) so ``RecursionError``
# never fires — it just burns CPU forever on a graph walk that should
# be a single dict lookup.
#
# Why the GPT2TokenizerFast bind above isn't enough: that fix only
# covers ONE leaf of the lazy graph. Whichever ``*TokenizerFast``
# is requested first triggers the same recursion via different
# candidate paths.
#
# Architectural note (impact-assessed): we considered three other
# approaches and rejected each — see
# ``memory/feedback_langchain_uplift_path.md``:
# 1. ``sys.setrecursionlimit(N)`` — recursion is bounded, never
# raises; forcing N too low breaks legitimate deep stacks.
# 2. Lazy-import langchain in HARTOS — multi-week refactor; AND
# HevolveAI (encrypted .so.enc, NOT refactor-able) is a second
# transformers consumer that would still trigger this recursion
# at runtime, so the layer-A fix doesn't cover all consumers.
# 3. Uplift langchain-classic → langchain-core + LangGraph — same
# multi-week scope as (2), still doesn't cover HevolveAI.
#
# (1)–(3) are all consumer-side fixes. This patch is at the
# ``transformers._LazyModule`` layer, which means BOTH consumers
# (open langchain + closed-source HevolveAI) benefit without changes
# to either. That's why a third-party monkey-patch is the
# architecturally correct answer here, not a band-aid.
#
# Mechanism: install a re-entry guard on ``_LazyModule.__getattr__``.
# When a thread is already mid-``__getattr__`` for a (module, name)
# pair and the same pair is requested again (which is what
# ``hasattr`` does inside the same method), raise ``AttributeError``
# immediately. ``hasattr`` swallows ``AttributeError`` and returns
# ``False`` — semantically identical to "the attribute hasn't been
# bound yet" which is exactly what the calling code is asking.
#
# Cross-name recursion (``__getattr__('A')`` triggering
# ``__getattr__('B')``) is preserved — only same-name re-entry on
# the same thread is short-circuited. That's the precise shape of
# the bug.
#
# TODO(upstream): file an issue with HuggingFace transformers
# proposing the fix at the source — replace the offending
# ``hasattr(transformers_module, candidate_name)`` with
# ``candidate_name in transformers_module.__dict__``. Once a
# released transformers carries the upstream fix, this monkey-patch
# becomes vestigial and can be removed. Track via
# ``memory/feedback_transformers_lazy_module_patch.md``.
try:
import threading as _tf_threading
from transformers.utils import import_utils as _tf_import_utils
_tf_lazy_module_class = getattr(_tf_import_utils, '_LazyModule', None)
if _tf_lazy_module_class is not None and not getattr(
_tf_lazy_module_class, '_hartos_reentry_guarded', False):
_orig_getattr = _tf_lazy_module_class.__getattr__
_resolving = _tf_threading.local()
# Bind both the original method AND the threading.local() store
# via default arguments so the closure carries its own
# references. Module-body cleanup (`del _orig_getattr` etc.)
# below is then safe — without this, Python's closure-by-name
# lookup would NameError when the lazy module later actually
# invokes __getattr__ at runtime (symptom:
# ``Tier-1 FAILED — name '_orig_getattr' is not defined``).
def _hartos_reentry_guarded_getattr(
self, name, _orig=_orig_getattr, _local=_resolving):
in_progress = getattr(_local, 'set', None)
if in_progress is None:
in_progress = set()
_local.set = in_progress
key = (id(self), name)
if key in in_progress:
# Same (module, name) re-entry on this thread —
# ``hasattr`` probe inside ``__getattr__``. The
# caller is asking "is this already bound?"; we are
# mid-resolution so the answer is "not yet". Raising
# AttributeError tells hasattr→False, which is the
# semantic the caller wanted.
raise AttributeError(
f"module {self.__name__!r} has no attribute {name!r} "
f"(HARTOS re-entry guard: same-name __getattr__ "
f"recursion broken)"
)
in_progress.add(key)
try:
return _orig(self, name)
finally:
in_progress.discard(key)
_tf_lazy_module_class.__getattr__ = _hartos_reentry_guarded_getattr
_tf_lazy_module_class._hartos_reentry_guarded = True
# NOTE: not deleting _orig_getattr / _resolving — the patched
# method's default args already hold references; deleting the
# module-scope names would leave them dangling at module-body cleanup
# but otherwise harmless. Leave them visible (underscore-prefixed,
# excluded from `from … import *` semantics) to keep the call site
# debuggable via dir().
del _tf_threading, _tf_import_utils, _tf_lazy_module_class
except Exception:
# transformers not installed, version moved _LazyModule, or some
# other surprise — fall through. We've still got the
# GPT2TokenizerFast direct-bind above; recursion may resurface
# but won't crash boot.
import logging as _early_logging
_early_logging.getLogger(__name__).warning(
"transformers _LazyModule re-entry defang skipped (not installed/moved)", exc_info=True)
from bs4 import BeautifulSoup
from enum import Enum
from hartos.cultural_wisdom import get_cultural_prompt_compact
from hartos.agent_identity import build_identity_prompt, SECRETS_GUARDRAIL, extract_owner_name
# The one description of an agent's credential ask (core.agent_tools shares it).
from hartos.ai_key_vault import REQUEST_RESOURCE_DESCRIPTION
# langchain_classic — pydantic v2-compatible fork of langchain 0.0.230.
# Fail-safe (#99): in a frozen env where langchain_classic is absent (the HART OS
# ISO's pinned June-2025 nixpkgs predates the split package), import the module
# ANYWAY so the backend serves its non-langchain surface (social / sync / status /
# daemon) instead of being dead. The langchain chat path then errors at call-time
# (gated by _LANGCHAIN_OK), not at import. The try-block IS the original import
# list, so any build that HAS langchain_classic is byte-for-byte unchanged — this
# can only help the no-langchain build, never regress the working one. The real
# fix (so chat actually works on the ISO) is still to package langchain_classic
# into the Nix env or bump the pin; this just stops a missing dep from killing the
# whole backend.
try:
from langchain_classic.llms import OpenAI
from langchain_classic.chains import LLMChain
from langchain_classic.prompts import PromptTemplate
from langchain_classic.agents import (
ZeroShotAgent, Tool, AgentExecutor, ConversationalAgent,
ConversationalChatAgent, LLMSingleActionAgent, AgentOutputParser,
load_tools, initialize_agent, AgentType
)
from langchain_classic.prompts import (
ChatPromptTemplate, MessagesPlaceholder,
SystemMessagePromptTemplate, HumanMessagePromptTemplate
)
from langchain_classic.chains import LLMMathChain
from langchain_classic.chains.conversation.memory import ConversationSummaryMemory, ConversationBufferWindowMemory
from langchain_classic.chat_models import ChatOpenAI
from langchain_classic.llms.base import LLM
from langchain_classic.memory import ConversationBufferMemory, ReadOnlySharedMemory
from langchain_classic.schema import AgentAction, AgentFinish, OutputParserException, HumanMessage, AIMessage, SystemMessage
from langchain_classic.tools import OpenAPISpec, APIOperation, StructuredTool
from langchain_classic.utilities import GoogleSearchAPIWrapper
_LANGCHAIN_OK = True
except ImportError as _lc_err:
import logging as _logging
_logging.getLogger(__name__).error(
"langchain_classic unavailable (%s) — backend boots WITHOUT the langchain "
"chat path; social/sync/status/daemon still serve (#99).", _lc_err,
exc_info=True)
# 2026-08-24: the message alone cannot be root-caused — the frozen
# install failed with "No module named 'yaml.error'" while every yaml
# copy on every path (frozen lib/, python-embed site-packages, user
# site, ~/.nunba) had a complete package incl. error.py, and the same
# import block passed on bare python-embed. Whatever yaml object the
# app actually resolved is the evidence; capture it.
try:
import sys as _lc_sys
_lc_y = _lc_sys.modules.get('yaml')
_logging.getLogger(__name__).error(
"langchain import-failure context: yaml=%r file=%s path=%s "
"spec=%s | sys.path[:6]=%s",
_lc_y, getattr(_lc_y, '__file__', None),
getattr(_lc_y, '__path__', None),
getattr(_lc_y, '__spec__', None), _lc_sys.path[:6])
except Exception:
pass
_LANGCHAIN_OK = False
class _LangchainUnavailable:
"""A langchain symbol used at call-time raises clearly (not a bare
NameError). The four module-level `class X(<base>)` subclasses below
inherit plain ``object`` so the module still imports."""
def __init__(self, *a, **k):
raise RuntimeError("langchain_classic is not installed in this build")
def __call__(self, *a, **k):
raise RuntimeError("langchain_classic is not installed in this build")
# Base classes subclassed at module level (ChatQwen3VL/CustomGPT -> LLM,
# CustomAgentExecutor -> AgentExecutor, CustomConvoOutputParser ->
# AgentOutputParser) MUST be real, plain classes so the defs import.
LLM = AgentExecutor = AgentOutputParser = object
# Everything else: a sentinel that fails loudly only if actually used.
OpenAI = LLMChain = PromptTemplate = ZeroShotAgent = Tool = \
ConversationalAgent = ConversationalChatAgent = LLMSingleActionAgent = \
load_tools = initialize_agent = AgentType = ChatPromptTemplate = \
MessagesPlaceholder = SystemMessagePromptTemplate = \
HumanMessagePromptTemplate = LLMMathChain = ConversationSummaryMemory = \
ConversationBufferWindowMemory = ChatOpenAI = ConversationBufferMemory = \
ReadOnlySharedMemory = AgentAction = AgentFinish = OutputParserException = \
HumanMessage = AIMessage = SystemMessage = OpenAPISpec = APIOperation = \
StructuredTool = GoogleSearchAPIWrapper = _LangchainUnavailable
import time
# ChatGroq - optional import (version compatibility issues)
try:
from langchain_groq import ChatGroq
except Exception:
ChatGroq = None
try:
from langchain_classic.requests import Requests
except (ImportError, AttributeError):
Requests = None
from flask import Flask, jsonify, request, g
from functools import wraps
import json
import os
import re
import secrets
import logging
import threading
import atexit
import requests
import pytz
import hashlib
from security.node_integrity import compute_code_hash, compute_file_manifest, verify_json_signature
from security import master_key
from core.platform_paths import get_coding_workspace_dir
# Module logger. 13 sites already call `logger.` / `klogger.` and NOTHING
# bound either name at module scope -- ruff F821 reports them, and 12 of the 13
# sit inside `except` handlers, so the NameError replaced the real diagnosis.
# Traced: an exception in _wire_qr_pair_emitter hits `logger.debug` (:3520) ->
# NameError -> caught at :3525 -> NameError -> caught at :4165 -> surfaces to
# the user as "Channel connect error: name 'logger' is not defined". In
# _init_skills and _init_runtime_tools (thread entry points) it kills the
# thread silently. Spelled the way the other ~40 sites in this file spell it;
# _validate_startup's own local binding still wins inside that function.
logger = logging.getLogger(__name__)
# --- Hevolve Boot Integrity Verification ---
_boot_logger = logging.getLogger("hevolve_integrity")
def hevolve_verify_boot():
mode = (os.getenv("HEVOLVE_ENFORCEMENT_MODE") or "warn").lower()
tier = (os.getenv("HEVOLVE_NODE_TIER") or "unknown").lower()
manifest_path = master_key.RELEASE_MANIFEST_FILENAME
if not os.path.exists(manifest_path):
msg = f"[HevolveIntegrity] release_manifest.json missing (tier={tier}, mode={mode})"
if mode == "hard":
raise RuntimeError(msg)
_boot_logger.warning(msg)
return
with open(manifest_path) as f:
d = json.load(f)
pub_hex = d.get("master_public_key")
sig_hex = d.get("master_signature")
payload_obj = {k: d[k] for k in d.keys() if k != "master_signature"}
payload_json = json.dumps(payload_obj, sort_keys=True, separators=(",", ":"))
try:
verify_json_signature(pub_hex, payload_json, sig_hex)
except Exception as e:
msg = f"[HevolveIntegrity] Signature verification failed: {e}"
if mode == "hard":
raise RuntimeError(msg)
_boot_logger.warning(msg)
return
# force_walk: the boot check must hash the bytes it is about to run, never
# a cache. compute_code_hash's mtime cache lives in agent_data/, which the
# deploy bind-mounts into EVERY container, so the cache one image writes is
# read by the next. Measured 2026-09-22 15:56Z on central: a cancelled
# deploy's script kept running and booted its image at 15:37Z, writing the
# cache; the next image's checkout carried an earlier mtime (15:32Z, its
# git reset), so this line returned the PREVIOUS image's hash, called the
# new image tampered ("CODE_HASH mismatch"), and the deploy rolled back
# into a crash loop. A forced walk is a pure read: one directory walk.
current_code_hash = compute_code_hash(force_walk=True)
if d.get("code_hash") != current_code_hash:
msg = "[HevolveIntegrity] CODE_HASH mismatch"
if mode == "hard":
raise RuntimeError(msg)
_boot_logger.warning(msg)
fm = compute_file_manifest()
current_fm_hash = hashlib.sha256(json.dumps(fm, sort_keys=True).encode()).hexdigest()
if d.get("file_manifest_hash") != current_fm_hash:
msg = "[HevolveIntegrity] FILE_MANIFEST_HASH mismatch"
if mode == "hard":
raise RuntimeError(msg)
_boot_logger.warning(msg)
_boot_logger.info(f"[HevolveIntegrity] Boot verification OK (tier={tier}, mode={mode})")
# Defer boot verification to main() — do not run at import time
# --- End Boot Integrity Verification ---
from core.http_pool import LLM_COMPLETION_TIMEOUT, pooled_get, pooled_post
from core.auth_local import (
_is_local_request, require_local_or_token, require_local_or_token_csrf_safe,
)
from datetime import datetime, timezone
from typing import List, Union, Optional, Mapping, Any, Dict
# Conversational chat imports
try:
from langchain_classic.agents.conversational_chat.output_parser import ConvoOutputParser
from langchain_classic.agents.conversational_chat.prompt import FORMAT_INSTRUCTIONS
from langchain_classic.output_parsers.json import parse_json_markdown
except (ImportError, AttributeError):
ConvoOutputParser = None
FORMAT_INSTRUCTIONS = None
parse_json_markdown = None
# Tools imports
try:
from langchain_classic.tools.requests.tool import RequestsGetTool
except (ImportError, AttributeError):
RequestsGetTool = None
try:
from langchain_classic.utilities.requests import TextRequestsWrapper
except (ImportError, AttributeError):
TextRequestsWrapper = None
try:
import tiktoken
except ImportError:
tiktoken = None
from pytz import timezone
from datetime import datetime, timedelta
from waitress import serve
from logging.handlers import RotatingFileHandler
# Pydantic v2 imports (we have pydantic 2.12.5)
try:
from pydantic import BaseModel, Field, field_validator as root_validator
except ImportError:
from pydantic import BaseModel, Field, root_validator
from hartos.threadlocal import thread_local_data
# Crossbar HTTP publisher — canonical path is `crossbarhttp3.CrossbarHttpPublisher`
# (the same publisher used by integrations/social/realtime.py:25). We fall
# back to the legacy `crossbarhttp.Client` API only if the canonical
# publisher is not installed. Either path is OPTIONAL — if neither is
# importable the module-load must NOT crash; this layer is dead-code in
# flat mode where MessageBus uses LOCAL EventBus + PeerLink without HTTP.
#
# Regression 2026-04-26: a `.venv` with broken `crossbarhttp` (no
# __init__.py, Client at .crossbarhttp.Client) made the previous
# `crossbarhttp.Client(...)` module-level call raise AttributeError,
# crashing the whole HARTOS import (Tier-1 dead → agent daemon dead →
# admin dashboard empty). Wrapping in try/except + using the canonical
# crossbarhttp3 first kills that failure mode.
_crossbar_publisher = None
try:
from crossbarhttp3 import CrossbarHttpPublisher as _CrossbarPub
except Exception:
_CrossbarPub = None
try:
import crossbarhttp as _legacy_cb # may be a broken stub
except Exception:
_legacy_cb = None
from PIL import Image
import numpy as np
# Cohere rerank - make optional to avoid pydantic v2 incompatibility with old langchain
try:
from langchain_community.retrievers.document_compressors import cohere_rerank
except Exception:
cohere_rerank = None # Not available
import asyncio
import aiohttp
import redis
import pickle
from threading import Thread
from concurrent.futures import ThreadPoolExecutor
from dotenv import load_dotenv
try:
import autogen
except ImportError:
autogen = None # Optional dependency
from typing import Dict, Tuple
load_dotenv()
#autogen requirements
try:
from hartos.create_recipe import recipe, time_based_execution as time_execution, visual_execution
from hartos.reuse_recipe import chat_agent, crossbar_multiagent, time_based_execution, visual_based_execution
except ImportError as e:
# print() went nowhere: frozen windowed builds have no real stdout, so the
# ONE line explaining why agent creation is dead was invisible. MEASURED
# 2026-08-19: "Could not import recipe modules" appears 0 times across
# langchain.log/gui_app.log/server.log while recipe=None caused 54
# `TypeError: 'NoneType' object is not callable` at the two unguarded call
# sites (:9351, :9578), each one a 500 on /chat. Same logger idiom this
# file already uses at line 261; exc_info keeps the real ImportError cause.
logging.getLogger(__name__).error(
"Could not import recipe modules -- agent creation and recipe "
"execution are DISABLED for this process: %s", e, exc_info=True)
recipe = None
time_execution = None
visual_execution = None
chat_agent = None
crossbar_multiagent = None
time_based_execution = None
visual_based_execution = None
try:
from autobahn.asyncio.component import Component, run
except ImportError:
Component = None
run = None
import threading
try:
from hartos.helper import retrieve_json, PROMPTS_DIR, safe_prompt_path, _is_terminate_msg
except Exception:
retrieve_json = None
safe_prompt_path = None
# helper failed to import — resolve PROMPTS_DIR via the SAME canonical resolver
# helper uses, so we NEVER fall back to the read-only /nix/store package dir on
# the embedded OS (the boot crash this guards). get_recipe_prompts_dir handles
# frozen desktop + embedded HART OS (/nix/store or /etc/hartos-release) + dev.
from core.platform_paths import get_recipe_prompts_dir
PROMPTS_DIR = os.path.abspath(get_recipe_prompts_dir())
# Ensure prompts directory exists (agent creation writes JSON here)
os.makedirs(PROMPTS_DIR, exist_ok=True)
# Boot-time prompts/ snapshot (best-effort, non-blocking). Bounds
# data loss to "since last reboot" if the user wipes data dir or
# hits a corruption. Fires at module-import time so it runs in
# BOTH the CLI hevolve-server path AND the Nunba bundled path
# (Nunba imports hart_intelligence_entry as a library; the CLI
# main() is never called in that case). See core/prompts_backup.py.
try:
from core.prompts_backup import snapshot_at_boot as _hartos_snapshot_at_boot
_hartos_snapshot_at_boot()
except Exception as _snap_err:
logging.getLogger('hevolve_core').debug(
f'prompts_backup at module-load skipped: {_snap_err}')
# Once, after the snapshot above: move aside the VLM learnings that do not say
# which action they ran under (a next-free-slot walker filed them under other
# actions' ids; the loader refuses them either way). Moved, never deleted;
# marker-guarded. See hartos.helper.quarantine_unproven_vlm_learnings_once.
try:
from hartos.helper import quarantine_unproven_vlm_learnings_once as _vlm_q_once
_vlm_q_once(PROMPTS_DIR)
except Exception as _vlm_q_err:
logging.getLogger('hevolve_core').warning(
f'VLM learning quarantine at module-load skipped: {_vlm_q_err}')
# Google A2A integration (from gpt4.1)
try:
from integrations.google_a2a import initialize_a2a_server, get_a2a_server, register_all_agents
except ImportError:
initialize_a2a_server = None
get_a2a_server = None
register_all_agents = None
# os.environ['LANGCHAIN_TRACING_V2'] = 'true'
# os.environ['LANGCHAIN_ENDPOINT'] = 'https://api.smith.langchain.com'
# os.environ['LANGCHAIN_API_KEY'] = os.getenv("LANGCHAIN_API_KEY")
# os.environ["LANGCHAIN_PROJECT"] = os.getenv("LANGCHAIN_PROJECT")
groq_api_key = os.environ.get('GROQ_API_KEY', '')
# ============================================================================
# MemoryGraph — Framework-agnostic provenance-aware memory layer
# ============================================================================
_memory_graphs = {} # Cache: "user_id_prompt_id" → MemoryGraph instance
_memory_graph_lock = threading.Lock()
def _get_or_create_graph(user_id, prompt_id=None):
"""Get or create a MemoryGraph instance for a user/prompt session."""
try:
from integrations.channels.memory.memory_graph import MemoryGraph
session_key = f"{user_id}_{prompt_id}" if prompt_id else str(user_id)
with _memory_graph_lock:
if session_key not in _memory_graphs:
from core.platform_paths import get_memory_graph_dir
db_path = get_memory_graph_dir(session_key)
_memory_graphs[session_key] = MemoryGraph(
db_path=db_path,
user_id=str(user_id),
)
return _memory_graphs[session_key]
except Exception as e:
logging.getLogger(__name__).debug(f"MemoryGraph init skipped: {e}")
return None
def _record_lifecycle(status, user_id, prompt_id, details=''):
"""Record agent lifecycle event in MemoryGraph (fire-and-forget, zero latency)."""
def _bg():
try:
graph = _get_or_create_graph(user_id, prompt_id)
if graph:
graph.register_lifecycle(
event=status,
agent_id=str(user_id),
session_id=f"{user_id}_{prompt_id}",
details=details,
)
except Exception:
logging.getLogger(__name__).exception("_bg: swallowed Exception")
threading.Thread(target=_bg, daemon=True).start()
# ============================================================================
# Custom Qwen3-VL LangChain Wrapper
# ============================================================================
class ChatQwen3VL(LLM):
"""
LangChain LLM wrapper over the ONE configured LLM (an OpenAI-compatible
chat/completions endpoint: the configured API, or the local llama-server
when none is configured; see core.autogen_config).
Posts through _pooled_post_with_refusal_check like every other
chat/completions caller in this module, so it shares the refusal guard,
the completion timeout and the auth header. base_url / model_name are
left None to follow GPT_API / LLM_MODEL_NAME; set them only for an
explicit, deliberate override.
"""
base_url: str = None # None: GPT_API (the configured backend)
model_name: str = None # None: LLM_MODEL_NAME
temperature: float = 0.7
max_tokens: int = 1500
@property
def _llm_type(self) -> str:
return "qwen3.5"
def _call(self, prompt: str, stop: list = None) -> str:
"""
Call the Qwen3.5 API with the given prompt.
Args:
prompt: The input text prompt
stop: Optional stop sequences
Returns:
The generated response text
"""
_url = f"{self.base_url}/chat/completions" if self.base_url else GPT_API
payload = {
"model": self.model_name or LLM_MODEL_NAME,
"messages": [{"role": "user", "content": prompt}],
"temperature": self.temperature,
"max_tokens": self.max_tokens
}
if stop:
payload["stop"] = stop
_log = logging.getLogger(__name__)
# Route through _pooled_post_with_refusal_check (same helper CustomGPT
# uses) so this langchain wrapper gets the SAME refusal-override guard;
# previously only CustomGPT had it, so a draft refusal from THIS wrapper
# flowed straight to the user (the drifted parallel path).
try:
response = _pooled_post_with_refusal_check(
_url,
json=payload,
timeout=60
)
response.raise_for_status()
data = response.json()
return data["choices"][0]["message"]["content"]
except Exception as e:
_log.warning(f"[LLM] {_url} unavailable: {e}")
# Local kind only: re-resolve the llama URL (port conflicts, warm
# starts). With a configured API there is nothing else to try: the
# single API is the only LLM this node talks to, never a local
# fallback behind the user's back.
if LLM_KIND != 'local':
raise
from core.port_registry import get_local_llm_url
_llm_url = get_local_llm_url().rstrip('/') + '/chat/completions'
if _llm_url != _url:
try:
_log.info(f"[LocalLLM] Falling back to llama.cpp at {_llm_url}")
response = _pooled_post_with_refusal_check(
_llm_url,
json=payload,
timeout=120
)
response.raise_for_status()
data = response.json()
return data["choices"][0]["message"]["content"]
except Exception as e2:
_log.error(f"[LocalLLM] llama.cpp fallback also failed: {e2}")
raise
raise
@property
def _identifying_params(self) -> dict:
"""Return identifying parameters."""
return {
"model_name": self.model_name,
"base_url": self.base_url,
"temperature": self.temperature,
"max_tokens": self.max_tokens
}
def get_llm(temperature=0.7, max_tokens=1500):
"""
Get LLM instance based on configuration.
Priority:
1. Wizard-configured cloud provider (HEVOLVE_ACTIVE_CLOUD_PROVIDER env var)
through its vendor SDK. The model is the one the vault exported; there
is no vendor default to fall back on.
2. ChatQwen3VL over the ONE configured backend (GPT_API / LLM_MODEL_NAME).
"""
_active = os.environ.get('HEVOLVE_ACTIVE_CLOUD_PROVIDER', '')
if _active == 'anthropic' and os.environ.get('ANTHROPIC_API_KEY'):
try:
from langchain_anthropic import ChatAnthropic
return ChatAnthropic(
model=os.environ.get('ANTHROPIC_MODEL') or LLM_MODEL_NAME,
api_key=os.environ['ANTHROPIC_API_KEY'],
temperature=temperature,
max_tokens=max_tokens,
)
except ImportError:
logging.getLogger(__name__).debug("get_llm: swallowed ImportError")
if _active == 'google_gemini' and os.environ.get('GOOGLE_API_KEY'):
try:
from langchain_google_genai import ChatGoogleGenerativeAI
return ChatGoogleGenerativeAI(
model=os.environ.get('GOOGLE_MODEL') or LLM_MODEL_NAME,
google_api_key=os.environ['GOOGLE_API_KEY'],
temperature=temperature,
max_output_tokens=max_tokens,
)
except ImportError:
logging.getLogger(__name__).debug("get_llm: swallowed ImportError")
if _active == 'groq' and os.environ.get('GROQ_API_KEY'):
try:
from langchain_groq import ChatGroq
return ChatGroq(
model=os.environ.get('GROQ_MODEL') or LLM_MODEL_NAME,
api_key=os.environ['GROQ_API_KEY'],
temperature=temperature,
max_tokens=max_tokens,
)
except ImportError:
logging.getLogger(__name__).debug("get_llm: swallowed ImportError")
if _active in ('openai', 'azure_openai', 'custom_openai') and os.environ.get('OPENAI_API_KEY'):
_kwargs = dict(
model_name=os.environ.get('OPENAI_MODEL') or LLM_MODEL_NAME,
temperature=temperature,
max_tokens=max_tokens,
)
if _active == 'custom_openai':
# The endpoint the vault exported, the same one autogen uses.
_kwargs['openai_api_base'] = _llm_entry.get('base_url') or os.environ.get('CUSTOM_LLM_BASE_URL', '')
return ChatOpenAI(**_kwargs)
return ChatQwen3VL(temperature=temperature, max_tokens=max_tokens)
# ============================================================================
class RequestLogRecord(logging.LogRecord):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Safely get the req_id from thread-local storage
self.req_id = thread_local_data.get_request_id()
# logging info
# Use the custom log record factory
logging.setLogRecordFactory(RequestLogRecord)
# In bundled/pip-installed mode (NUNBA_BUNDLED env set by main.py), redirect logs
# to the shared Nunba log directory; standalone keeps default behavior.
_is_bundled = bool(os.environ.get('NUNBA_BUNDLED') or getattr(sys, 'frozen', False))
if _is_bundled:
from core.platform_paths import get_log_dir as _get_log_dir_lc
_nunba_log_dir = _get_log_dir_lc()
try:
os.makedirs(_nunba_log_dir, exist_ok=True)
except PermissionError:
_nunba_log_dir = os.path.join(os.path.expanduser('~'), '.nunba', 'logs')
os.makedirs(_nunba_log_dir, exist_ok=True)
_langchain_log_path = os.path.join(_nunba_log_dir, 'langchain.log')
else:
# Non-bundled: write to the canonical WRITABLE data dir, NEVER the CWD — on the
# embedded OS the CWD is the read-only /nix/store package dir, so a relative
# 'langchain.log' crashes boot with OSError [Errno 30] Read-only file system.
try:
from core.platform_paths import get_data_dir
_ld = os.path.join(get_data_dir(), 'logs')
os.makedirs(_ld, exist_ok=True)
_langchain_log_path = os.path.join(_ld, 'langchain.log')
except Exception:
_langchain_log_path = os.devnull # last resort: a log path must never brick boot
handler = RotatingFileHandler(_langchain_log_path, maxBytes=5_000_000, backupCount=2, encoding='utf-8')
handler.setLevel(logging.INFO)
# Prevent UnicodeEncodeError on Windows (cp1252) when LLM output contains emoji.
# In cx_Freeze frozen builds, stdout/stderr may be closed — redirect to devnull.
try:
if sys.stdout is None or sys.stdout.closed:
sys.stdout = open(os.devnull, 'w', encoding='utf-8', errors='replace')
else:
sys.stdout.reconfigure(errors='replace')
except Exception:
sys.stdout = open(os.devnull, 'w', encoding='utf-8', errors='replace')
try:
if sys.stderr is None or sys.stderr.closed:
sys.stderr = open(os.devnull, 'w', encoding='utf-8', errors='replace')
except Exception:
logging.getLogger(__name__).exception("<module>: swallowed Exception")
stream_handler = logging.StreamHandler(sys.stdout)
# Create a logging format.
# [%(threadName)s] is included so EVERY line says which thread it ran on —
# without it, interleaved background threads (agent_daemon,
# resource_governor_monitor, compute_optimizer, TTS _bg, hevolveai-supervisor)
# all log as "RequestID: None" and are indistinguishable from the chat
# request thread, making per-thread latency attribution undecidable (the
# 2026-05-31 "why is hi slow" mis-attribution: a 17s gap was blamed on the
# wrong subsystem purely from log adjacency). threadName is a built-in
# LogRecord attribute (always present) so this is free + safe.
formatter = logging.Formatter(
'%(asctime)s - %(name)s- [RequestID: %(req_id)s] [%(threadName)s] - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
stream_handler.setFormatter(formatter)
# Configure root logger: drop OUR OWN previously-installed handlers to prevent
# duplicates on re-import, then attach them once. In bundled mode the root
# logger gets both stream + file so that ALL module loggers are captured. In
# standalone mode only Flask's app.logger gets the handlers.
#
# NEVER call `_root.handlers.clear()` here. This module runs at IMPORT time,
# and when a host application has already configured root logging, a blanket
# clear() silently steals it.
#
# Incident 2026-08-03 (task #489): Nunba attaches gui_app.log + server.log to
# the ROOT logger at startup. Nunba imports hart_intelligence_entry lazily,
# ~140s into every run — at which point this block detached both handlers.
# Measured on a live frozen build: gui_app.log and server.log both stopped
# dead at boot+140s and stayed silent for the next 134 minutes of uptime,
# while named-logger files (agent_system.log) kept writing. Every Nunba root
# log line was silently relocated into langchain.log. The two files CLAUDE.md
# documents as THE diagnostic logs were dark for ~98% of each session, which
# is why several investigations "found nothing in the logs".
#
# Tagging our handlers keeps the original de-duplication intent (a second
# import removes exactly what the first import added) without touching
# handlers owned by anyone else.
_HARTOS_HANDLER_TAG = '_hartos_root_handler'
def _install_root_handlers(root, bundled, file_handler, console_handler):
"""Attach this module's handlers to the ROOT logger, without stealing anyone's.
Three launch shapes, three answers:
* bundled (cx_Freeze): root gets file + console, so every module logger
lands in langchain.log and the console;
* a host that already configured root (Nunba's gui_app.log/server.log,
pytest's capture): root is left exactly as found, per the 2026-08-03
incident above;
* a bare root with NO handler at all: root gets the console handler.
The third case is the OS and Docker launch (`python hart_intelligence_entry.py`
under systemd). Before 2026-09-24 it got nothing, so every INFO line the
root logger saw during main() went to Python's lastResort handler, which
prints WARNING and above only. Measured on the hart-ota-central nixos test
(2026-09-23 run, shard 0): "STARTUP VALIDATION WARNINGS" reached the journal
as bare text, and "[Guardrail] hash verified", "Platform bootstrapped" and
"Local subscribers bootstrapped: ... ota-push" never appeared at all, until a
background thread's import called logging.basicConfig() some two seconds
after Waitress bound and INFO started flowing in basicConfig's format. The
OTA test waited 240 s for a line the process had already dropped, and the
boot narrative that hart-backend.nix exempts from the journald rate limit
was never being written in the first place.
Attaching only when root is empty is the same rule basicConfig() follows, so
a later basicConfig() call becomes the no-op it was always meant to be and
the journal keeps ONE format (the RequestID/threadName one) end to end.
"""
for existing in list(root.handlers):
if getattr(existing, _HARTOS_HANDLER_TAG, False):
root.removeHandler(existing)
root.setLevel(logging.INFO)
if bundled:
setattr(file_handler, _HARTOS_HANDLER_TAG, True)
setattr(console_handler, _HARTOS_HANDLER_TAG, True)
root.addHandler(file_handler)
root.addHandler(console_handler)
else:
# The bare-root case is ONE rule shared with the agent daemon's
# entrypoint (core.log_handlers); it attaches `console_handler`
# only when root has no handler at all, tagged with the same tag.
from core.log_handlers import install_console_handler_if_bare
install_console_handler_if_bare(root, handler=console_handler,
tag=_HARTOS_HANDLER_TAG)
_root = logging.getLogger()
_install_root_handlers(_root, _is_bundled, handler, stream_handler)
app = Flask(__name__)
app.config['SECRET_KEY'] = os.environ.get('FLASK_SECRET_KEY') or secrets.token_hex(32)
# One payload policy for app AND transport: core.serve passes the same
# constant to Hypercorn's WSGI middleware (whose 64 KB library default
# otherwise transport-rejects bodies Flask's own cap here would accept).
from core.constants import (
AGENT_WORK_TOOL_NAMES, AGENTIC_ROUTER_TOOL_NAME, CREATE_AGENT_TOOL_NAME,
MAX_PAYLOAD_BYTES, SHELL_COMMAND_TIMEOUT_S,
)
app.config['MAX_CONTENT_LENGTH'] = MAX_PAYLOAD_BYTES # 2MB default, HEVOLVE_MAX_PAYLOAD_BYTES to override
# ...for a remote caller. A local caller's body has no cap (owner
# 2026-10-10); the per-request rule is core.serve.max_body_size.
from core.serve import caller_capped_request
app.request_class = caller_capped_request(app.request_class)
# ── LLM outbound logger ───────────────────────────────────────────────
# Monkey-patches httpx.Client.send (+ AsyncClient) to (a) inject the
# thread-local HARTOS request_id as the OpenAI ``user`` field on every
# POST to :8082/v1/chat/completions and (b) append the full request
# body to ~/Documents/Nunba/logs/llm_outbound.jsonl. Covers autogen,
# langchain, openai-python — every Python-originated LLM call funnels
# through httpx. See ``core/llm_outbound_logger.py`` for the full
# rationale (the 2026-05-12 IPL ctx-overflow incident motivated this).
try:
from core.llm_outbound_logger import install as _install_outbound_hook
_install_outbound_hook()
except Exception as _outbound_err:
logging.getLogger(__name__).debug(
"llm_outbound install skipped: %s", _outbound_err)
# ── Proxy-lookup cache (process-wide) ─────────────────────────────────
# requests/urllib re-read the Windows registry for proxy settings on EVERY
# HTTP call (getproxies_registry); the 2026-06-13 sluggishness dig caught the
# superadmin-report loop burning CPU there, and every other requests call pays
# it too. Cache the lookup with a TTL — same proxies, just not re-resolved per
# call. See core/proxy_cache.py.
try:
from core.proxy_cache import install_proxy_cache
install_proxy_cache()
except Exception as _proxy_err:
logging.getLogger(__name__).debug(
"proxy_cache install skipped: %s", _proxy_err)
# ── Boot-time tool-registry warmer ────────────────────────────────────
# Kicks a background daemon thread to pre-fill the 4 process-level tool
# caches (service_tools / system_introspect / skills / provider_gateway)
# so the FIRST user chat turn doesn't pay the ~80-100s cold-start
# penalty observed on 2026-05-13 09:55:23 (IPL turn 301eeed0). The
# warmer itself takes ~100s but runs in parallel with everything else;
# by the time a user types, the caches are warm. Failures are
# swallowed silently — the lazy on-call path retries any registry
# that was temporarily unavailable at boot.
def _spawn_tool_warmup_when_ready():
"""Deferred warmup spawn — runs once get_tools symbols are defined.
The actual warmer function `_warmup_tool_registries_in_background`
is defined further down in this file, so we wrap the spawn in a
short delay-thread that resolves it from globals() at call time.
"""
try:
import threading as _t
def _deferred_spawn():
try:
time.sleep(1)
except Exception:
logging.getLogger(__name__).exception("_deferred_spawn: swallowed Exception")
warmer = globals().get('_warmup_tool_registries_in_background')
if warmer:
try:
warmer()
except Exception as _e:
logging.getLogger(__name__).debug(
"tool-warmup invocation failed: %s", _e)
_t.Thread(target=_deferred_spawn, daemon=True,
name='tool-warmup-spawner').start()
except Exception as _ws_err:
logging.getLogger(__name__).debug(
"tool-warmup spawner failed: %s", _ws_err)
_spawn_tool_warmup_when_ready()
def _json_endpoint(f):
"""Wrap a Flask view so unhandled exceptions return ``{'error': ...}, 500``.
Must be defined before any module-level route registration block uses it
as a decorator — Python resolves decorator names at function-definition
time, not at call time.
"""
@wraps(f)
def _wrapped(*args, **kwargs):
try:
return f(*args, **kwargs)
except Exception as e:
return jsonify({'error': str(e)}), 500
return _wrapped
if _is_bundled:
# Bundled: root owns the handlers; let app.logger propagate to root
# so ALL module loggers (hevolveai, langchain, etc.) are captured.
app.logger.propagate = True
else:
# Standalone: only Flask's logger gets the handlers
app.logger.addHandler(stream_handler)
app.logger.addHandler(handler)
app.logger.propagate = False
# Security: Audit logging — redact API keys, JWTs, passwords from all logs
try:
from security.audit_log import apply_sensitive_filter_to_all
apply_sensitive_filter_to_all()
except Exception:
logging.getLogger(__name__).exception("<module>: swallowed Exception") # Degrade gracefully — logs will still work, just unredacted
# Test logging
app.logger.info('Logger initialized')
# Security: Apply middleware (headers, CORS, CSRF, host validation, API auth)
try:
from security.middleware import apply_security_middleware
apply_security_middleware(app)
app.logger.info("Security middleware applied")
except Exception as e:
app.logger.warning(f"Security middleware not applied: {e}")
# Boot-time tier safety check: warn if HEVOLVE_NODE_TIER is unset on non-bundled.
# A misconfigured central/regional node defaults to 'flat' and silently accepts
# body-sourced user_id without auth — the CISO flagged this as a data-exfil risk.
_boot_tier = os.environ.get('HEVOLVE_NODE_TIER')
_is_bundled = bool(os.environ.get('NUNBA_BUNDLED') or getattr(sys, 'frozen', False))
if not _boot_tier and not _is_bundled:
app.logger.warning(
'HEVOLVE_NODE_TIER not set — defaulting to "flat" (no auth on /chat). '
'Set HEVOLVE_NODE_TIER=regional or =central for multi-user deployments.')
elif _boot_tier in ('regional', 'central'):
# Verify JWT secret exists, otherwise auth enforcement is pointless
_jwt_secret = os.environ.get('SECRET_KEY', '') or app.config.get('SECRET_KEY', '')
if not _jwt_secret:
app.logger.error(
f'HEVOLVE_NODE_TIER={_boot_tier} but no SECRET_KEY configured! '
f'JWT auth will reject ALL requests. Set SECRET_KEY env var.')
# Security: Block inspect.getsource() for protected packages (hevolveai)
try:
from security.source_protection import install_source_guards
install_source_guards()
except Exception:
logging.getLogger(__name__).exception("<module>: swallowed Exception") # Non-fatal — source stripping is the primary defense
# ============================================================================
# HevolveSocial - Agent Social Network
# ============================================================================
try:
from integrations.social import social_bp, init_social
app.register_blueprint(social_bp)
init_social(app)
app.logger.info("HevolveSocial registered at /api/social")
except Exception as e:
app.logger.warning(f"HevolveSocial init skipped (non-critical): {e}")
try:
from integrations.distributed_agent import distributed_agent_bp
app.register_blueprint(distributed_agent_bp)
app.logger.info("distributed_agent_bp registered at /api/distributed")
except ImportError:
app.logger.info("distributed_agent module not available, skipping")
except Exception as e:
app.logger.warning(f"distributed_agent init skipped: {e}")
try:
from integrations.social.api_provision import provision_bp
app.register_blueprint(provision_bp)
app.logger.info("provision_bp registered at /api/provision")
except ImportError:
app.logger.info("Provision module not available, skipping")
except Exception as e:
app.logger.warning(f"Provision init skipped: {e}")
# Legacy ``register_consent_routes`` block removed in the
# consent-surface consolidation (orchestrator review acd11f55,
# 2026-04-25). The HTTP write surface for user consent now lives at
# ``integrations.social.consent_api`` (consent_bp, JWT-authed,
# append-only, mounted at ``/api/social/consent``). ``init_social``
# in ``integrations/social/__init__.py`` registers the new blueprint;
# no extra wiring is required here.
# Claude Code frontier endpoint — OpenAI-compatible /chat/completions backed by
# the authorized `claude -p` (flow #3: HARTOS wears Claude as its EXPERT tier).
# The EXPERT ModelBackend (model_registry) points its base_url here; autogen
# calls it like any local llama.cpp endpoint. Failures degrade to the in-house
# LLM via dispatch.py's existing fallback ladder.
try:
from integrations.providers.claude_code_endpoint import claude_code_bp
app.register_blueprint(claude_code_bp)
app.logger.info("Claude Code frontier endpoint registered at /api/claude/v1")
except ImportError:
app.logger.info("Claude Code endpoint not available, skipping")
except Exception as e:
app.logger.warning(f"Claude Code endpoint init skipped: {e}")
# Agent-engine API — the ledger surface the MCP daemon probe already depends on.
#
# WHY THIS WAS MISSING AND WHY IT MATTERS. integrations/mcp/_tool_impls.py
# documents a known false negative: `probe_agent_daemon()` reads `_running` /
# `_tick_count` off an imported module, and when Python (or, on HART OS, a
# SECOND PROCESS) resolves a different agent_daemon instance than the live one,
# it reports `daemon_enabled=false, _tick_count=0` while the daemon is happily
# ticking. Its defence is to ALSO fetch canonical ledger stats over Flask
# loopback, "shadow-immune" because the request lands on whichever singleton
# Flask actually resolved.
#
# That defence has never been able to run here: agent_engine_bp was registered
# nowhere. Measured on the box 2026-09-09 -- `list_routes` on this backend
# returns 829 routes and ZERO agent-engine, and /api/agent-engine/ledger/stats
# 404s on :6777 AND on the nunba socket. So the corrective probe 404'd, the
# module-attr view was the only view, and the agent engine was structurally
# unobservable from outside the process.
#
# The ledger is DB-backed (HEVOLVE_DB_PATH, shared by both processes), so
# serving it here answers for the node rather than for this process. Registered
# in its own try/except like every neighbour: a route-drop must never take the
# app down.
try:
from integrations.agent_engine import get_engine_blueprint as _get_agent_engine_bp
app.register_blueprint(_get_agent_engine_bp())
app.logger.info("Agent-engine API registered at /api/agent-engine/")
except ImportError:
app.logger.info("Agent-engine API not available, skipping")
except Exception as e:
app.logger.warning(f"Agent-engine API init skipped: {e}")
# MCP HTTP Bridge — exposes local MCP tools via REST for Nunba/external clients
try:
from integrations.mcp.mcp_http_bridge import mcp_local_bp, auto_register_local_mcp
app.register_blueprint(mcp_local_bp)
auto_register_local_mcp()
app.logger.info("MCP HTTP bridge registered at /api/mcp/local")
except ImportError:
app.logger.info("MCP HTTP bridge not available, skipping")
except Exception as e:
app.logger.warning(f"MCP HTTP bridge init skipped: {e}")
# Model Onboarding API — HuggingFace → GGUF → llama.cpp in one call
try:
from integrations.service_tools.model_onboarding import get_blueprint as _get_onboard_bp
app.register_blueprint(_get_onboard_bp())
app.logger.info("Model onboarding API registered at /api/models/")
except ImportError:
app.logger.info("Model onboarding not available, skipping")
except Exception as e:
app.logger.warning(f"Model onboarding init skipped: {e}")
# Robot Intelligence API — 7 intelligences fuse in parallel for any embodied AI
try:
from integrations.robotics.intelligence_api import robot_intelligence_bp
app.register_blueprint(robot_intelligence_bp)
app.logger.info("Robot Intelligence API registered at /api/robot/")
except ImportError:
app.logger.info("Robot Intelligence API not available, skipping")
except Exception as e:
app.logger.warning(f"Robot Intelligence API init skipped: {e}")
# Commercial API — the paid /api/v1/intelligence/* surface.
# Was coded but never registered with Flask, leaving 8 revenue endpoints
# (chat, analyze, generate, hivemind, usage, keys CRUD) dead. Routes:
# POST /api/v1/intelligence/chat
# POST /api/v1/intelligence/analyze
# POST /api/v1/intelligence/generate
# GET /api/v1/intelligence/hivemind
# GET /api/v1/intelligence/usage
# POST /api/v1/intelligence/keys
# GET /api/v1/intelligence/keys
# DELETE /api/v1/intelligence/keys/<key_id>
# Tiers: free / starter / pro / enterprise (see TIER_CONFIG in
# integrations/agent_engine/commercial_api.py:32-36). Revenue split via
# revenue_aggregator: 90% users / 9% infra / 1% central.
# Free tier is always free — gatekeeping intelligence is not the goal;
# sustaining the hive is. This blueprint registration is the single
# wiring change between "revenue surface dead" and "customers can land".
try:
from integrations.agent_engine.commercial_api import commercial_api_bp
app.register_blueprint(commercial_api_bp)
app.logger.info(
"Commercial API registered at /api/v1/intelligence/ — "
"free/starter/pro/enterprise tiers, 90/9/1 revenue split")
except ImportError:
app.logger.info("Commercial API not available, skipping")
except Exception as e:
app.logger.warning(f"Commercial API init skipped: {e}")
# Central OTA control — GET /api/ota/latest (PUBLIC pointer nodes poll on their
# hart-ota-check timer), POST /api/ota/publish (account-gated, kicks the upgrade
# pipeline + fans a signed firmware_update fleet command), GET /api/ota/nodes
# (live per-node rollout) — is served by the fleet_update blueprint, registered
# inside integrations/social/__init__.py alongside the other social surfaces.
# (There is ONE OTA backend; the former agent_engine/ota_api.py duplicate was
# removed during integration — Flask routed to the social blueprint anyway, so
# it was dead code; its pipeline-kick + audit folded into api_fleet_update.py.)
# Hive Session API — Claude Code as hive worker node
try:
from integrations.coding_agent.claude_hive_session import get_blueprint as _get_hive_bp
app.register_blueprint(_get_hive_bp())
app.logger.info("Hive Session API registered at /api/hive/session/")
except ImportError:
app.logger.info("Hive Session API not available, skipping")
except Exception as e:
app.logger.warning(f"Hive Session API init skipped: {e}")
# Hive Signal Bridge API — channel signals dashboard
try:
from integrations.channels.hive_signal_bridge import create_signal_blueprint
app.register_blueprint(create_signal_blueprint())
app.logger.info("Hive Signal Bridge registered at /api/hive/signals/")
except ImportError:
app.logger.info("Hive Signal Bridge not available, skipping")
except Exception as e:
app.logger.warning(f"Hive Signal Bridge init skipped: {e}")
# Hive Benchmark Prover API — distributed benchmark proving
try:
from integrations.agent_engine.hive_benchmark_prover import create_benchmark_prover_blueprint
_bench_bp = create_benchmark_prover_blueprint()
if _bench_bp:
app.register_blueprint(_bench_bp)
app.logger.info("Benchmark Prover API registered at /api/hive/benchmark/")
except ImportError:
app.logger.info("Benchmark Prover not available, skipping")
except Exception as e:
app.logger.warning(f"Benchmark Prover init skipped: {e}")
# Robotics Hardware Bridge API — sensor/actuator registration and action execution
try:
from integrations.robotics.hardware_bridge import _create_blueprint as _create_hw_bp
_hw_bp = _create_hw_bp()
if _hw_bp:
app.register_blueprint(_hw_bp)
app.logger.info("Robotics Hardware Bridge registered at /api/robot/")
except ImportError:
app.logger.info("Robotics Hardware Bridge not available, skipping")
except Exception as e:
app.logger.warning(f"Robotics Hardware Bridge init skipped: {e}")
# Compute Optimizer API — system resource optimization
try:
from core.compute_optimizer import create_optimizer_blueprint, get_optimizer
_opt_bp = create_optimizer_blueprint()
if _opt_bp:
app.register_blueprint(_opt_bp)
_optimizer = get_optimizer()
_optimizer.start()
app.logger.info("Compute Optimizer registered and started")
except ImportError:
app.logger.info("Compute Optimizer not available, skipping")
except Exception as e:
app.logger.warning(f"Compute Optimizer init skipped: {e}")
# App Marketplace — publish, compare, distribute, auto-promote AI apps
try:
from integrations.agent_engine.app_marketplace import marketplace_bp
app.register_blueprint(marketplace_bp)
app.logger.info("App Marketplace registered at /api/marketplace/")
except ImportError:
app.logger.info("App Marketplace not available, skipping")
except Exception as e:
app.logger.warning(f"App Marketplace init skipped: {e}")
# McGroce agentic commerce — /api/commerce/session (Tomcat -> HARTOS token
# exchange), /mandates/<id>, /health. Also in blueprint_registry for the
# Nunba path; _try_register skips it there when this one already ran.
try:
from integrations.commerce.commerce_api import commerce_bp
app.register_blueprint(commerce_bp)
app.logger.info("Commerce API registered at /api/commerce/")
except ImportError:
app.logger.info("Commerce API not available, skipping")
except Exception as e:
app.logger.warning(f"Commerce API init skipped: {e}")
# Resource Governor — start background monitoring (task #260: watchdog-registered)
try:
from core.resource_governor import get_governor
_gov = get_governor()
_gov.start()
app.logger.info(f"Resource Governor started (mode={_gov.get_mode()})")
# Register both daemon loops with the watchdog. Monitor runs every 5s;
# proactive waits up to 5s per iteration. Allow 4× grace for noisy
# system-state probes (nvidia-smi, /proc).
try:
from security.node_watchdog import get_watchdog
_wd = get_watchdog()
if _wd and _gov._running:
_wd.register('resource_governor_monitor',
expected_interval=20,
restart_fn=_gov.start, stop_fn=_gov.stop)
_wd.register('resource_governor_proactive',
expected_interval=20,
restart_fn=_gov.start, stop_fn=_gov.stop)
app.logger.info("Resource Governor registered with watchdog")
except Exception as e:
app.logger.debug(f"Governor watchdog registration skipped: {e}")
except ImportError:
logging.getLogger(__name__).debug("<module>: swallowed ImportError")
except Exception as e:
app.logger.warning(f"Resource Governor start skipped: {e}")
# NOTE — the AI-senses cross-process authority server (Phase 7) is NOT started
# here. core.ai_sensing._state is mutated ONLY by POST /api/shell/ai-sensing,
# which is served by the LiquidUI shell process (liquid_ui_service.py), not this
# :6777 backend. The authority socket must reflect the SAME _state the human
# mutates, so start_authority_server() is called THERE (the canonical holder).
# Starting it here too would bind a SECOND copy of _state that the kill-switch
# never updates — reporting a stale "screen allowed" after the human cut it. One
# writer, one authority. (Co-located bundles where both run in one process are
# unaffected: a single start_authority_server() in the shared heap still wins.)
# Central Orchestrator Client — heartbeat to hevolve.ai central + master
# kill-switch polling. Env-gated: no-op when HEVOLVE_CENTRAL_ORCHESTRATOR_URL
# is unset (default flat-mode install), so we burn ZERO resources on a
# poll loop nobody asked for. When configured, the client posts small
# heartbeats (node_id + guardrail_hash + benchmark best-scores + halted
# flag), polls /halt for a master-signed stop signal, and routes that
# through HiveCircuitBreaker.halt_network() — which itself verifies the
# master-key signature before tripping. See docs/ml_intern_brief §5-D.
try:
from core import central_orchestrator_client as _coc_mod
_coc = _coc_mod.get_client()
if _coc.is_configured():
if _coc.start():
app.logger.info(
"Central Orchestrator Client started "
"(env-gated: HEVOLVE_CENTRAL_ORCHESTRATOR_URL set)"
)
# Watchdog — catch silent thread death. Expected interval
# is 2× heartbeat (heartbeat loop wakes at min(heartbeat,
# halt_poll, 5s) so 120s gives comfortable headroom without
# triggering on a single missed poll.
try:
from security.node_watchdog import get_watchdog
_wd = get_watchdog()
if _wd:
_wd.register(
'central_orchestrator_client',
expected_interval=120,
restart_fn=_coc.start,
stop_fn=_coc.stop,
)
app.logger.info(
"Central Orchestrator Client registered with watchdog"
)
except ImportError:
logging.getLogger(__name__).debug("<module>: swallowed ImportError")
except Exception as e:
app.logger.debug(
f"Central Orchestrator watchdog registration skipped: {e}"
)
# atexit — signal the loop to exit cleanly on process shutdown.
# Daemon thread exits with the process regardless; this just
# flushes any in-flight heartbeat.
import atexit as _atexit
_atexit.register(_coc.stop)
else:
# start() returned False — either already running, or node
# tier is 'central' itself (brief's design: central doesn't
# self-heartbeat). Both paths log inside start().
pass
else:
app.logger.debug(
"Central Orchestrator Client inactive "
"(HEVOLVE_CENTRAL_ORCHESTRATOR_URL unset) — flat-mode default"
)
except ImportError:
app.logger.debug(
"Central Orchestrator Client not available, skipping"
)
except Exception as e:
app.logger.warning(f"Central Orchestrator Client init skipped: {e}")
# Instruction Queue API — never miss a user instruction
try:
from integrations.agent_engine.instruction_queue import (
get_queue, enqueue_instruction, pull_user_batch,
)
@app.route('/api/instructions/enqueue', methods=['POST'])
def _api_enqueue_instruction():
data = request.get_json(silent=True) or {}
user_id = str(data.get('user_id', '1'))
text = data.get('text', '')
if not text:
return jsonify({'error': 'text is required'}), 400
inst = enqueue_instruction(
user_id, text,
priority=data.get('priority', 5),
tags=data.get('tags'),
context=data.get('context'),
)
return jsonify({'success': True, 'instruction': inst.to_dict()})
@app.route('/api/instructions/pending', methods=['GET'])
def _api_pending_instructions():
user_id = request.args.get('user_id', '1')
q = get_queue(user_id)
pending = q.get_pending()
return jsonify({
'success': True,
'pending': [i.to_dict() for i in pending],
'stats': q.stats(),
})
@app.route('/api/instructions/batch', methods=['POST'])
def _api_pull_batch():
data = request.get_json(silent=True) or {}
user_id = str(data.get('user_id', '1'))
max_tokens = data.get('max_tokens', 8000)
batch = pull_user_batch(user_id, max_tokens=max_tokens)
if not batch:
return jsonify({'success': True, 'batch': None, 'message': 'No pending instructions'})
return jsonify({'success': True, 'batch': batch.to_dict()})
@app.route('/api/instructions/cancel', methods=['POST'])
def _api_cancel_instruction():
data = request.get_json(silent=True) or {}
user_id = str(data.get('user_id', '1'))
instruction_id = data.get('instruction_id', '')
if not instruction_id:
return jsonify({'error': 'instruction_id is required'}), 400
q = get_queue(user_id)
q.cancel(instruction_id)
return jsonify({'success': True})
@app.route('/api/instructions/plan', methods=['GET'])
def _api_execution_plan():
"""Pull execution plan — agents query this to see what's available.
Returns dependency-aware waves so the agent can dispatch
independent items in parallel and dependent items in order.
This is a PULL API — agents decide when to fetch work.
"""
user_id = request.args.get('user_id', '1')
max_tokens = int(request.args.get('max_tokens', '8000'))
q = get_queue(user_id)
plan = q.pull_execution_plan(max_tokens=max_tokens)
if not plan:
return jsonify({'success': True, 'plan': None, 'message': 'No pending instructions'})
return jsonify({'success': True, 'plan': plan.to_dict()})
@app.route('/api/instructions/complete', methods=['POST'])
def _api_complete_instruction():
"""Mark a single instruction as done (used by agents after execution).
Notifies SmartLedger to unblock dependent instructions.
"""
data = request.get_json(silent=True) or {}
user_id = str(data.get('user_id', '1'))
instruction_id = data.get('instruction_id', '')
result = data.get('result')
if not instruction_id:
return jsonify({'error': 'instruction_id is required'}), 400
q = get_queue(user_id)
q.complete_instruction(instruction_id, result=result)
return jsonify({'success': True})
@app.route('/api/instructions/fail', methods=['POST'])
def _api_fail_instruction():
"""Mark a single instruction as failed — returns to queue for retry."""
data = request.get_json(silent=True) or {}
user_id = str(data.get('user_id', '1'))
instruction_id = data.get('instruction_id', '')
error = data.get('error', 'unknown')
if not instruction_id:
return jsonify({'error': 'instruction_id is required'}), 400
q = get_queue(user_id)
q.fail_instruction(instruction_id, error=error)
return jsonify({'success': True})
@app.route('/api/instructions/drain', methods=['POST'])
def _api_drain_queue():
"""Trigger wave-based drain — dispatches with parallel + sequential ordering.
Agent or daemon calls this to execute all pending instructions.
Independent instructions dispatch in parallel threads.
Dependent instructions wait for prerequisites.
"""
data = request.get_json(silent=True) or {}
user_id = str(data.get('user_id', '1'))
max_tokens = data.get('max_tokens', 8000)
try:
from integrations.agent_engine.dispatch import drain_instruction_queue
result = drain_instruction_queue(user_id, max_tokens=max_tokens)
if result:
return jsonify({'success': True, 'result': result[:2000]})
return jsonify({'success': True, 'result': None, 'message': 'Queue empty or all failed'})
except Exception as e:
return jsonify({'error': str(e)}), 500
app.logger.info("Instruction queue API routes registered (8 endpoints)")
except Exception as e:
app.logger.warning(f"Instruction queue init skipped: {e}")
# ── Credential Vault API — frontend submits missing credentials ──────
try:
from hartos.ai_key_vault import AIKeyVault as _VaultCls, is_local_request
@app.route('/api/credentials/submit', methods=['POST'])
@_json_endpoint
def _api_credentials_submit():
"""Accept a credential from the frontend — LOCALHOST ONLY.
Secrets never leave the user's device. This endpoint rejects
any request that doesn't originate from the local machine.
"""
if not is_local_request():
return jsonify({'error': 'Credential endpoints are localhost only. '
'Secrets never leave your device.'}), 403
data = request.get_json(silent=True) or {}
key_name = (data.get('key_name') or '').strip()
value = (data.get('value') or '').strip()
if not key_name or not value:
return jsonify({'error': 'key_name and value are required'}), 400
vault = _VaultCls.get_instance()
# A name the process holds that this vault did not store (PATH,
# HTTPS_PROXY...) is left alone by store_credential and answered as
# any other, so this endpoint does not say which variables exist.
resolved = vault.store_credential(
key_name=key_name,
value=value,
channel_type=data.get('channel_type', ''),
)
return jsonify({'success': True, 'key_name': resolved})
@app.route('/api/credentials/pending', methods=['GET'])
@_json_endpoint
def _api_credentials_pending():
"""List pending credential requests — LOCALHOST ONLY."""
if not is_local_request():
return jsonify({'error': 'Credential endpoints are localhost only.'}), 403
vault = _VaultCls.get_instance()
return jsonify({'pending': vault.get_pending_requests()})
app.logger.info("Credential vault API routes registered (2 endpoints)")
except ImportError:
logging.getLogger(__name__).debug("<module>: swallowed ImportError")
except Exception as e:
app.logger.warning(f"Credential vault API init skipped: {e}")
# ── Kong API Gateway — Completions API proxy + metering ───────────────
try:
@app.route('/v1/chat/completions', methods=['POST'])
@_json_endpoint
def _completions_proxy():
"""OpenAI-compatible completions — proxied through Kong metering.
SDK clients hit Kong → Kong routes here → we forward to HevolveAI.
Token usage is metered for 90/9/1 revenue split.
"""
import requests as _req
data = request.get_json(silent=True) or {}
# Resolve the LLM the same way every other consumer does, instead of a
# private HEVOLVE_API_URL that defaulted to localhost:8000. That default
# is correct for an on-device install (llama.cpp on :8000 is the whole
# point of local Nunba), but on the central tier nothing runs there, so
# this proxy 502'd while Langchain/Autogen worked off the gateway. The
# resolver probes candidates and picks the first reachable one — the
# local model on a device, the gateway on central — so a single source
# of truth is correct in both topologies. Explicit HEVOLVE_API_URL still
# wins if set, for operators who want to pin it.
base = os.environ.get('HEVOLVE_API_URL')
if base:
base = base.rstrip('/')
# A pinned base may or may not end in /v1; build the completions
# path without doubling it.
target = base + ('/chat/completions' if base.endswith('/v1')
else '/v1/chat/completions')
headers = {'Content-Type': 'application/json'}
else:
# The ONE configured backend (core.autogen_config): the local
# model on a device, the configured API on central, same as every
# other chat/completions caller in this module.
target = GPT_API
headers = dict(LLM_AUTH_HEADERS)
headers['Content-Type'] = 'application/json'
try:
resp = _req.post(
target,
json=data,
headers=headers,
timeout=120
)
result = resp.json()
except Exception as fwd_err:
return jsonify({'error': f'HevolveAI backend unavailable: {fwd_err}'}), 502
# The serving half of a hive expert exchange: a peer's person asked,
# this node's operator earns the compute served (owner ruling
# 2026-09-26). No requester header (an SDK client) earns nothing.
if resp.status_code == 200:
from integrations.agent_engine.budget_gate import credit_served_completion
credit_served_completion(request.headers, data, result)
# Meter usage for the 90/9/1 revenue split. This passed keywords
# record_metered_usage does not accept (provider/model/tokens/source);
# the TypeError was swallowed below, so SDK usage was NEVER metered
# (hevolveai Master 11.435 S2). meter_llm_call owns the contract;
# 'hive' is the settled cross-operator source. Two things stay open
# for the owner: MeteredAPIUsage has no consumer field (the old
# 'sdk:<consumer>' tag had nowhere to go), and a LOCAL model prices at
# 0 via spark_per_1k, so SDK use of a local model records no revenue.
usage = result.get('usage', {}) or {}
total_tokens = usage.get('total_tokens', 0) or 0
if total_tokens > 0:
try:
from integrations.agent_engine.budget_gate import meter_llm_call
tokens_in = usage.get('prompt_tokens', 0) or 0
tokens_out = usage.get('completion_tokens',
max(0, total_tokens - tokens_in)) or 0
meter_llm_call(
model=data.get('model', 'hevolve'),
tokens_in=tokens_in,
tokens_out=tokens_out,
task_source='hive',
)
except Exception:
logging.getLogger(__name__).exception("_completions_proxy: swallowed Exception") # metering failure must not block response
return jsonify(result)
@app.route('/api/gateway/metering', methods=['GET'])
@_json_endpoint
def _gateway_metering():
"""SDK usage metering stats for billing dashboard."""
try:
from integrations.social.models import db_session
from integrations.agent_engine.budget_gate import metered_usage_by_model
with db_session(commit=False) as session:
return jsonify({'providers': metered_usage_by_model(session)})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/gateway/register', methods=['POST'])
@_json_endpoint
def _gateway_register_node():
"""Register a compute node as Kong upstream target.
Called by compute_mesh when a node joins the hive.
"""
data = request.get_json(silent=True) or {}
target = data.get('target') # e.g. "192.168.1.5:8000"
if not target:
return jsonify({'error': 'target is required'}), 400
kong_admin = os.environ.get('KONG_ADMIN_URL', 'http://localhost:8001')
upstream = data.get('upstream', 'hevolve-nodes')
try:
import requests as _req
resp = _req.post(
f'{kong_admin}/upstreams/{upstream}/targets',
json={'target': target, 'weight': 100},
timeout=10
)
return jsonify({'registered': True, 'status': resp.status_code})
except Exception as e:
return jsonify({'error': f'Kong admin unreachable: {e}'}), 502
app.logger.info("Kong gateway API routes registered (3 endpoints: completions proxy, metering, node registration)")
except Exception as e:
app.logger.warning(f"Kong gateway API init skipped: {e}")
# ============================================================================
# Google A2A Protocol Initialization
# ============================================================================
# Initialize A2A server for cross-platform agent communication
try:
app.logger.info("Initializing Google A2A Protocol server...")
from core.port_registry import get_port as _a2a_get_port
a2a_server = initialize_a2a_server(app, base_url=f"http://localhost:{_a2a_get_port('backend')}")
app.logger.info("Google A2A Protocol server initialized successfully")
# Register all agents with A2A
register_all_agents()
app.logger.info("All agents registered with Google A2A Protocol")
except Exception as e:
app.logger.warning(f"Google A2A Protocol initialization error (non-critical): {e}")
# openAPI spec
try:
spec = OpenAPISpec.from_file(
"./openapi.yaml"
)
except Exception as e:
app.logger.warning(f"Could not load OpenAPI spec: {e}")
spec = None
# Load config.json — graceful fallback when keys are absent (e.g. bundled Nunba install
# where the config.json is Nunba's URL-template file, not langchain's API-key file).
# In frozen Nunba builds, langchain config is bundled as langchain_config.json to avoid
# collision with the project root config.json (which has IP_ADDRESS settings).
config = {}
for _cfg_name in ('langchain_config.json', 'config.json'):
try:
with open(_cfg_name, 'r') as f:
_loaded = json.load(f)
# Distinguish langchain config (has API keys) from Nunba URL config (has IP_ADDRESS)
if 'IP_ADDRESS' not in _loaded or any(k.endswith('_API_KEY') for k in _loaded):
config = _loaded
break
elif not config:
config = _loaded # fall through to try langchain_config.json first
except Exception:
logging.getLogger(__name__).exception("<module>: swallowed Exception")
# global variables
# Token counting: single source — core.token_utils.count_tokens_for_text
# (tiktoken with a word-split fallback). Replaces the old module-global
# `encoding` object + the inline `len(encoding.encode(x)) if encoding else
# x.split()` chains (#112), and removes the import-time crash when tiktoken was
# unavailable (the old except ran tiktoken.get_encoding on a None tiktoken).
from core.token_utils import count_tokens_for_text
# api and keys — use config if available, otherwise keep existing env vars / empty
# A key absent from this tuple can be set in config.json and still never reach
# the code that reads it. That is how SEARXNG_URL sat configured-but-dead: the
# reader consults os.environ only, and nothing copied it across. Add the env
# name here whenever a new provider starts reading one.
# The LLM itself is NOT configured here: HEVOLVE_LLM_ENDPOINT_URL /
# HEVOLVE_LLM_MODEL_NAME / HEVOLVE_LLM_API_KEY are the ONE configured LLM
# (core/autogen_config.py); a vendor-named key (QWEN_*, GLM_*, GROQ_*) picks
# nothing any more (#69).
for _cfg_key in ('OPENAI_API_KEY', 'GOOGLE_CSE_ID', 'GOOGLE_API_KEY',
'NEWS_API_KEY', 'SERPAPI_API_KEY'):
if _cfg_key in config:
os.environ[_cfg_key] = config[_cfg_key]
else:
os.environ.setdefault(_cfg_key, '')
# Mode-aware inference: pass LLM endpoint to HevolveAI for non-flat deployments
_node_tier = os.environ.get('HEVOLVE_NODE_TIER', 'flat')
_active_cloud = os.environ.get('HEVOLVE_ACTIVE_CLOUD_PROVIDER', '')
if _node_tier in ('regional', 'central'):
# No default model name: the configured endpoint decides the model, and an
# endpoint without a name is reported by core.autogen_config, not guessed.
os.environ.setdefault('HEVOLVE_LLM_ENDPOINT_URL', config.get('OPENAI_API_BASE', ''))
os.environ.setdefault('HEVOLVE_LLM_API_KEY', config.get('OPENAI_API_KEY', ''))
os.environ.setdefault('HEVOLVE_LLM_MODEL_NAME', config.get('OPENAI_MODEL', ''))
elif _active_cloud and os.environ.get('HEVOLVE_LLM_API_KEY'):
# Wizard-configured cloud provider (flat mode desktop user).
# Vault already populated HEVOLVE_LLM_* env vars via export_to_env() in app.py.
# HevolveAI's create_learning_llm_config() reads these automatically.
pass
# Cloud fallback for adaptive routing (flat mode — offload when local CPU overloaded)
if config.get('CLOUD_FALLBACK_URL'):
os.environ.setdefault('HEVOLVE_CLOUD_FALLBACK_URL', config['CLOUD_FALLBACK_URL'])
os.environ.setdefault('HEVOLVE_CLOUD_FALLBACK_KEY', config.get('CLOUD_FALLBACK_KEY', ''))
os.environ.setdefault('HEVOLVE_CLOUD_FALLBACK_MODEL', config.get('CLOUD_FALLBACK_MODEL', ''))
# Zep removed — replaced by SimpleMem (local, zero-latency)
# ZEP_API_URL / ZEP_API_KEY no longer needed
# API endpoints — fall back to IP_ADDRESS sub-dict or empty when keys missing.
# In bundled Nunba mode the config.json is URL-template format with IP_ADDRESS dict;
# in cloud/dev mode it has flat top-level keys.
_ip = config.get('IP_ADDRESS', {})
def _resolve_llm_endpoint(registry_fn_name: str, env_var: str) -> str:
"""Resolve an LLM chat-completions URL via port_registry → env var fallback.
Both GPT_API and DRAFT_GPT_API use the same 2-step resolution:
1. core.port_registry.<registry_fn_name>() → base URL → + /chat/completions
2. os.environ[env_var] → base URL → + /chat/completions
Returns '' if neither resolves.
"""
url = ''
try:
import importlib
_mod = importlib.import_module('core.port_registry')
_base = getattr(_mod, registry_fn_name)()
if _base:
url = _base.rstrip('/') + '/chat/completions'
except Exception:
logging.getLogger(__name__).exception("_resolve_llm_endpoint: swallowed Exception")
if not url:
_env = os.environ.get(env_var, '')
if _env:
url = _env.rstrip('/') + '/chat/completions'
return url
# The ONE LLM this process talks to, decided in core.autogen_config from the
# configured trio (HEVOLVE_LLM_ENDPOINT_URL / _MODEL_NAME / _API_KEY). Every
# chat/completions POST in this module goes to GPT_API or DRAFT_GPT_API with
# LLM_MODEL_NAME in the body, and _pooled_post_with_refusal_check adds
# LLM_AUTH_HEADERS on the way out; no site picks a model or a port of its own.
from core.autogen_config import (
resolve_llm_backend, llm_http_target, with_local_fallback,
get_autogen_config_list)
LLM_KIND, _llm_entry = resolve_llm_backend()
LLM_MODEL_NAME = _llm_entry['model']
if LLM_KIND == 'api':
# A configured endpoint (central, a regional node with a key, a desktop
# with a cloud provider): the single API serves every role, so the draft
# slot IS the main model. config.json's GPT_API (the bundled llama URL)
# must not win here: on central it named a server that never ran inside
# the container (#69).
GPT_API, LLM_AUTH_HEADERS, _ = llm_http_target()
DRAFT_GPT_API = GPT_API
else:
LLM_AUTH_HEADERS = {}
# Main LLM endpoint (4B on :8080). Config.json first, then registry/env fallback.
GPT_API = config.get('GPT_API', _ip.get('gpt3_url', ''))
if not GPT_API:
GPT_API = _resolve_llm_endpoint('get_local_llm_url', 'HEVOLVE_LOCAL_LLM_URL')
# Draft LLM endpoint: the 0.8B on :8081 that serves casual_conv=True.
# When casual_conv=True, CustomGPT._call() posts here instead of GPT_API.
# The 0.8B responds in ~300ms with a short reply. If the response signals
# delegate=local (needs tools/reasoning), the caller re-dispatches with
# casual_conv=False which hits GPT_API (the 4B on :8080).
DRAFT_GPT_API = _resolve_llm_endpoint('get_local_draft_url', 'HEVOLVE_LOCAL_DRAFT_URL')
FAV_TEACHER_API = config.get('FAV_TEACHER_API', '')
DREAMBOOTH_API = config.get('DREAMBOOTH_API', '')
STABLE_DIFF_API = config.get('STABLE_DIFF_API', '')
# CRAWLAB_API removed — web crawling is now in-process via integrations.web_crawler
RAG_API = config.get('RAG_API', '')
# Endpoint resolution — single source of truth in core/config_cache.py
# Automatically resolves to localhost:5000 in bundled mode, cloud URLs otherwise.
from core.config_cache import (
get_db_url, get_action_api, get_student_api,
get_vision_api, is_bundled as _config_is_bundled,
get_central_db_url, flag_value,
)
DB_URL = get_db_url()
# Distinct from DB_URL: always points at the central cloud (kong-routed)
# instead of collapsing to localhost in bundled mode. Used by /prompts
# and /prompts/public to merge in agents the user owns on hevolve.ai
# but never synced down to this machine. Empty string disables the
# cross-device merge (local-only listing) — matches the same fail-quiet
# contract _merge_prompts_with_cloud already has on cloud failures.
CENTRAL_DB_URL = get_central_db_url()
ACTION_API = get_action_api()
STUDENT_API = get_student_api()
LLAVA_API = get_vision_api()
if _config_is_bundled():
logging.getLogger(__name__).info(
f"Bundled mode: DB/Action/Student/Vision APIs → {DB_URL}"
)
# ============================================================================
# Embodied AI Learning Pipeline (HevolveAI — in-process, no extra port)
# ============================================================================
_learning_provider = None
_hive_mind = None
_trace_recorder = None
def _is_bundled() -> bool:
"""Detect whether we are pip-installed inside Nunba (flat mode).
When Nunba imports us via ``hartos_backend_adapter``, that module is
already in ``sys.modules`` by the time our daemon thread runs. In
standalone mode (``python hart_intelligence_entry.py``, ``start_with_tracing.bat``)
it is absent. No env-vars or mode flags required.
"""
return 'hartos_backend_adapter' in sys.modules
def _has_cloud_api() -> bool:
"""Return True if the user configured an external cloud / API endpoint."""
return bool(os.environ.get('HEVOLVE_LLM_ENDPOINT_URL', '').strip())
def _wait_for_llm_server(url=None, timeout=15):
if url is None:
from core.port_registry import get_local_llm_url
url = get_local_llm_url().replace('/v1', '')
"""Wait for llama.cpp server, giving parent process time to start it.
In Nunba (flat mode), the desktop app starts llama.cpp in a background
thread. This function polls the health endpoint so HevolveAI sees an
existing server and reuses it instead of auto-starting a second one.
In standalone mode nobody else starts the server, so after *timeout*
seconds we return False and let HevolveAI auto-start as usual.
Returns True if server found, False if timeout expired.
"""
import urllib.request
import urllib.error
_logger = logging.getLogger(__name__)
for i in range(timeout):
try:
req = urllib.request.urlopen(f'{url}/health', timeout=2)
if req.status == 200:
_logger.info(
f"[EmbodiedAI] llama.cpp server ready at {url} "
f"(waited {i}s)")
return True
except (urllib.error.URLError, OSError):
logging.getLogger(__name__).warning("_wait_for_llm_server: swallowed urllib.error.URLError, OSError", exc_info=True)
time.sleep(1)
_logger.info(
f"[EmbodiedAI] No server on {url} after {timeout}s "
"\u2014 HevolveAI will auto-start")
return False
def _init_learning_pipeline():
"""Initialize HevolveAI's learning pipeline in-process.
Instead of starting a separate server on port 8000,
we import and initialize the learning components directly.
world_model_bridge calls these functions without HTTP overhead.
Behaviour depends on context (auto-detected, no env vars needed):
**Bundled (Nunba, flat mode):**
Nunba owns the llama.cpp lifecycle. We wait up to 30 s for
Nunba to start it. If it appears we reuse it. If it never
appears we skip the learning provider entirely — chat still
works, only RL-EF / hivemind is disabled. We NEVER auto-start
a second server on the user's machine.
**Standalone (start_with_tracing.bat, ``python hart_intelligence_entry.py``):**
Brief 5 s wait (in case user already has llama.cpp running).
If nothing responds, ``create_learning_llm_config()`` calls
HevolveAI which auto-starts its own server. Default behaviour,
no mode config needed.
**Cloud API configured (``HEVOLVE_LLM_ENDPOINT_URL``):**
Skip local server wait entirely — HevolveAI's Priority 0 path
routes to the external endpoint.
"""
global _learning_provider, _hive_mind, _trace_recorder
# RFC-B (2026-08-25): one learning home in bundled mode. The docstring's
# "instead of starting a separate server on port 8000" predates
# hevolveai_supervisor, which now always spawns (or adopts) that server
# in bundled Nunba and exports HEVOLVEAI_API_URL before this delayed
# init runs (measured: export 19:15:49 vs this init 19:19:13; export
# ~18:23 vs init 18:27:15). Running BOTH stacks grew commit ~23GB/h
# (snapshots 21:40→22:17) and OOM-killed the app at 22:2x. When bundled
# and a server is designated, the child is the sole learning home; the
# WorldModelBridge already carries stats/experiences/skills over HTTP
# (the http_disabled heal + the Direction-B poller, both live-proven).
if _is_bundled() and os.environ.get('HEVOLVEAI_API_URL'):
logging.getLogger(__name__).info(
"learning pipeline: skipping in-process init — supervisor-managed "
"hevolveai at %s is the learning home (RFC-B)",
os.environ['HEVOLVEAI_API_URL'])
return
try:
# G2: Ensure HevolveArmor's import hook is installed before the first
# `import hevolveai.*` statement runs below. `try_import_hevolveai`
# is idempotent and a no-op when the package is already loaded or
# no armored bundle is present on disk.
try:
from security.native_hive_loader import try_import_hevolveai
try_import_hevolveai('hevolveai')
except ImportError:
# security package missing — fall through to the raw import path
logging.getLogger(__name__).debug("_init_learning_pipeline: swallowed ImportError")
# Bypass rl_ef/__init__.py's graceful-degradation swallow
# (which binds every symbol to None on any ImportError /
# AttributeError / OSError / RuntimeError, then leaves us with
# the unhelpful "rl_ef symbols are None" warning that the guard
# at L1703 emits). Direct-module import surfaces the real
# import-time failure so it lands in the `except Exception as e`
# at L1757 with exc_info=True.
from hevolveai.embodied_ai.rl_ef.learning_llm_provider import (
create_learning_llm_config,
register_learning_provider,
)
from hevolveai.embodied_ai.monitoring.trace_recorder import get_trace_recorder
from hevolveai.embodied_ai.learning.hive_mind import HiveMind, AgentCapability
_logger = logging.getLogger(__name__)
bundled = _is_bundled()
cloud = _has_cloud_api()
_logger.info(
f"[EmbodiedAI] Initializing learning pipeline "
f"(bundled={bundled}, cloud_api={cloud})...")
# ── Decide how to handle the local llama.cpp server ──
if cloud:
# Cloud endpoint configured — HevolveAI uses it directly,
# no need to wait for or start a local server.
_logger.info(
"[EmbodiedAI] Cloud API configured — skipping local server wait")
elif bundled:
# Nunba owns the llama.cpp lifecycle (port 8080).
# Wait generously, but NEVER auto-start a second server.
server_found = _wait_for_llm_server(timeout=30)
if not server_found:
_logger.warning(
"[EmbodiedAI] Nunba's llama.cpp not ready after 30 s "
"— learning disabled (chat still works)")
return
else:
# Standalone — brief courtesy wait then let HevolveAI auto-start.
_wait_for_llm_server(timeout=5)
# Trace recorder
recordings_dir = os.path.join(
os.path.expanduser('~'), '.hevolveai', 'recordings')
os.makedirs(recordings_dir, exist_ok=True)
_trace_recorder = get_trace_recorder(recordings_dir)
# Learning provider (wraps llama.cpp on port 8080 or cloud endpoint)
#
# rl_ef/__init__.py uses a graceful-degradation pattern: when an
# optional sibling import (torch/cuda/transformers stack) fails,
# it binds create_learning_llm_config = register_learning_provider
# = None instead of raising. So a `from … import X` succeeds with
# X bound to None. Calling None(...) is a TypeError that crashes
# the *whole* pipeline (HiveMind included) — guard explicitly so
# only the learning-provider step is skipped.
if create_learning_llm_config is None or register_learning_provider is None:
_logger.warning(
"[EmbodiedAI] rl_ef symbols are None — an optional dep "
"(likely torch/cuda) failed to load inside hevolveai. "
"Trace recording + HiveMind continue; learning provider "
"skipped.")
else:
_domain = 'general'
_wizard_key = os.environ.get('HEVOLVE_LLM_API_KEY', '') or None
llm_config = create_learning_llm_config(
domain=_domain, fallback_api_key=_wizard_key)
if '_provider' in llm_config:
_learning_provider = llm_config['_provider']
register_learning_provider(_domain, _learning_provider)
_logger.info(
"[EmbodiedAI] Learning provider ready (RL-EF + episodic memory)")
else:
_logger.warning(
"[EmbodiedAI] Learning provider init returned no provider")
# HiveMind
if HiveMind is None or AgentCapability is None:
_logger.warning(
"[EmbodiedAI] HiveMind/AgentCapability symbols are None — "
"skipping HiveMind registration.")
else:
import uuid
instance_id = f"hevolve_{uuid.uuid4().hex[:8]}"
# HiveMind signature changed: the old `max_agents=100` kwarg
# is no longer accepted (live evidence frozen_debug.log 7×
# 2026-05-13/14: `TypeError: __init__() got an unexpected
# keyword argument 'max_agents'`). Today's HiveMind is a
# latent-space collaboration substrate (Procrustes-aligned)
# whose __init__ takes (common_dim=256, num_shards=4,
# fusion_method="attention", device="cpu") — no agent cap
# because cardinality is bounded by register_agent calls,
# not a constructor arg. Verified against
# hevolveai/src/.../learning/hive_mind.py:583-589. Defaults
# are sane for the desktop tier; device stays 'cpu' so we
# don't fight TTS for VRAM. Drop the kwarg.
_hive_mind = HiveMind()
_hive_mind.register_agent(
agent_id=instance_id,
agent_type='hevolve_orchestrator',
latent_dim=2048,
capabilities=[
AgentCapability.ENCODE,
AgentCapability.DECODE,
AgentCapability.REASON,
],
)
_logger.info(f"[EmbodiedAI] HiveMind registered as {instance_id}")
except ImportError as e:
logging.getLogger(__name__).warning(
f"[EmbodiedAI] HevolveAI not installed — learning disabled: {e}")
except Exception as e:
logging.getLogger(__name__).error(
f"[EmbodiedAI] Learning pipeline init failed: {e}",
exc_info=True)
def get_learning_provider():
"""Get the in-process learning provider (for world_model_bridge)."""
return _learning_provider
def get_hive_mind():
"""Get the in-process HiveMind instance (for world_model_bridge)."""
return _hive_mind
# Boot learning pipeline in background — delayed to avoid torch circular import.
# Threads that import torch race with the main thread's own torch imports,
# causing "partially initialized module 'torch'" errors.
def _delayed_learning_init():
import time
time.sleep(5) # Let main thread finish all imports first
_init_learning_pipeline()
threading.Thread(
target=_delayed_learning_init, daemon=True,
name='embodied_ai_init').start()
# ============================================================================
# Vision Pipeline (VisionService — MiniCPM sidecar + FrameStore)
# ============================================================================
_vision_service = None
def _init_vision_service():
"""Start VisionService in standalone mode only.
In bundled mode (Nunba), Nunba owns the VisionService lifecycle
and stores it at ``__main__._vision_service``. We skip here to
prevent double-start.
"""
global _vision_service
if _is_bundled():
logging.getLogger(__name__).info(
"[Vision] Bundled mode — Nunba owns VisionService lifecycle")
return
try:
from integrations.vision import VisionService
_vision_service = VisionService()
_vision_service.start()
logging.getLogger(__name__).info(
"[Vision] VisionService started (standalone mode)")
except ImportError:
logging.getLogger(__name__).info(
"[Vision] vision module not available — disabled")
except Exception as e:
logging.getLogger(__name__).warning(
f"[Vision] VisionService failed to start: {e}")
def get_vision_service():
"""Get the active VisionService instance.
Checks module-level var first, then falls back to Nunba's
``__main__._vision_service``; if that misses (cx_Freeze on macOS
mounts ``main.py`` under an opaque module name so neither ``__main__``
nor ``main`` matches), scans the live object graph for a
``VisionService`` instance. Safe because ``_start_vision_service``
only ever creates one.
"""
if _vision_service is not None:
return _vision_service
main_mod = sys.modules.get('__main__')
if main_mod:
_svc = getattr(main_mod, '_vision_service', None)
if _svc is not None:
return _svc
# cx_Freeze fallback: scan the heap.
try:
import gc
from integrations.vision.vision_service import VisionService
for _obj in gc.get_objects():
if isinstance(_obj, VisionService):
return _obj
except Exception:
pass
return None
def get_frame_store():
"""Get the active FrameStore (from VisionService)."""
svc = get_vision_service()
return svc.store if svc else None
def _wire_vision_to_learning():
"""Connect FrameStore to HevolveAI's video learning pipeline.
Waits up to 60s for both VisionService and LearningLLMProvider,
then calls start_video_learning().
"""
_logger = logging.getLogger(__name__)
# Fast-fail: if hevolveai isn't installed, learning is impossible — don't wait 60s
import importlib.util
if not importlib.util.find_spec('hevolveai'):
_logger.info("[Wiring] hevolveai not installed — skipping vision-learning wire")
return
for _ in range(60):
svc = get_vision_service()
if svc and _learning_provider:
break
time.sleep(1)
else:
_logger.info(
"[Wiring] Vision or learning not ready after 60s — skipping")
return
try:
if hasattr(_learning_provider, 'start_video_learning'):
_learning_provider.start_video_learning(
frame_store=get_frame_store(),
frame_store_user_id='default',
)
_logger.info(
"[Wiring] FrameStore → HevolveAI video learning connected")
else:
_logger.info(
"[Wiring] LearningProvider has no start_video_learning — skip")
except Exception as e:
_logger.warning(f"[Wiring] FrameStore→learning failed: {e}")
# Boot vision pipeline in background — delayed like learning init
def _delayed_vision_init():
import time
time.sleep(6) # After learning init delay (5s) to stagger torch imports
_init_vision_service()
threading.Thread(
target=_delayed_vision_init, daemon=True,
name='vision_init').start()
# Wire FrameStore to HevolveAI after both subsystems are ready
threading.Thread(
target=_wire_vision_to_learning, daemon=True,
name='vision_learning_wire').start()
# ============================================================================
# Speaker Diarization Pipeline (sidecar subprocess)
# ============================================================================
_diarization_service = None
def _init_diarization_service():
"""Start DiarizationService in standalone mode only.
In bundled mode (Nunba), Nunba owns the lifecycle.
Skips if HEVOLVE_DIARIZATION_URL is already set (external service).
"""
global _diarization_service
if _is_bundled():
logging.getLogger(__name__).info(
"[Diarization] Bundled mode — Nunba owns lifecycle")
return
# If user already configured an external diarization URL, don't start sidecar
if os.environ.get('HEVOLVE_DIARIZATION_URL', '').strip():
logging.getLogger(__name__).info(
"[Diarization] External URL configured — skipping sidecar")
return
try:
from integrations.audio import DiarizationService
_diarization_service = DiarizationService()
_diarization_service.start()
logging.getLogger(__name__).info(
"[Diarization] DiarizationService starting (standalone mode)")
except ImportError:
logging.getLogger(__name__).info(
"[Diarization] audio module not available — disabled")
except Exception as e:
logging.getLogger(__name__).warning(
f"[Diarization] DiarizationService failed to start: {e}")
def get_diarization_service():
"""Get the active DiarizationService instance."""
if _diarization_service is not None:
return _diarization_service
main_mod = sys.modules.get('__main__')
if main_mod:
return getattr(main_mod, '_diarization_service', None)
return None
# Boot diarization sidecar in background
threading.Thread(
target=_init_diarization_service, daemon=True,
name='diarization_init').start()
# task scheduling and logging
class TaskStatus(Enum):
INITIALIZED = "INITIALIZED"
SCHEDULED = "SCHEDULED"
EXECUTING = "EXECUTING"
TIMEOUT = "TIMEOUT"
COMPLETED = "COMPLETED"
ERROR = "ERROR"
class TaskNames(Enum):
GET_ACTION_USER_DETAILS = "GET_ACTION_USER_DETAILS"
GET_TIME_BASED_HISTORY = "GET_TIME_BASED_HISTORY"
ANIMATE_CHARACTER = "ANIMATE_CHARACTER"
STABLE_DIFF = "STABLE_DIFF"
LLAVA = "LLAVA"
CRAWLAB = "CRAWLAB"
USER_ID_RETRIEVER = "USER_ID_RETRIEVER"
# google search API — graceful when GOOGLE_API_KEY is missing (e.g. local-only mode)
try:
search = GoogleSearchAPIWrapper(k=4)
except Exception as e:
logging.getLogger(__name__).info(f"Google Search unavailable (expected in local mode): {e}")
search = None
# constants
# llm = ChatOpenAI(model_name="gpt-3.5-turbo-16k")
# llm = ChatOpenAI(temperature=0, model="gpt-4")
# llm = CustomGPT()
# The above code is creating an instance of the `LLMMathChain` class with an `open_ai_llm` attribute
# initialized with a `ChatOpenAI` object using the model name "gpt-3.5-turbo".
# llm_math = LLMMathChain(ChatOpenAI(model_name="gpt-3.5-turbo"))
# Old: llm_math = LLMMathChain(llm=ChatOpenAI(model_name="gpt-3.5-turbo"))
# get_llm() serves the configured backend (core.autogen_config); no model name here
try:
llm_math = LLMMathChain(llm=get_llm())
except Exception as _llm_math_err:
# NEVER silent. On the 2026-08-07 nightly this except swallowed the
# constructor error, llm_math stayed None for the whole boot, and every
# Tier-1 chat died at `llm_math.run` (AttributeError) while --validate
# stayed green — import succeeds, so the breakage left zero trace until
# a live user traceback. The Calculator tool is optional; chat is not.
# Guarded use sites: get_tools (x2) + tools_func — same `is not None`
# idiom as `chain` below.
llm_math = None
logging.getLogger(__name__).warning(
"LLMMathChain init failed — Calculator tool disabled for this boot "
"(chat continues without it): %s", _llm_math_err)
# llm_math = LLMMathChain(llm= ChatGroq(groq_api_key=groq_api_key,
# model_name = "mixtral-8x7b-32768"))
# llm = ChatGroq(groq_api_key=groq_api_key,
# model_name="llama-3.1-8b-instant", temperature=0.3)
# app.logger.info(llm.invoke("hi how are you?"))
# OpenAPI chain — deprecated (get_openapi_chain removed from langchain)
chain = None
# URL precedence: WAMP_URL env (set by Nunba desktop / regional / central) →
# legacy cloud default. Flat mode (no cloud) sets WAMP_URL to localhost so
# DNS to aws_rasa.hertzai.com is never attempted (#323).
try:
from core.wamp_url import resolve_publish_url as _resolve_wamp_publish
except ImportError: # cx_Freeze defence — never let realtime kill boot
def _resolve_wamp_publish():
return os.environ.get('WAMP_URL') or 'http://localhost:8088/publish'
# The run scripts export ws://…:8088/ws here; this normalises that to the
# HTTP bridge instead of POSTing at a router socket. See core/wamp_url.py.
_wamp_url = _resolve_wamp_publish()
try:
if _CrossbarPub is not None:
client = _CrossbarPub(_wamp_url)
elif _legacy_cb is not None and hasattr(_legacy_cb, 'Client'):
client = _legacy_cb.Client(_wamp_url, timeout=2.0)
else:
# Legacy package may expose Client at .crossbarhttp.Client
# (broken namespace install). Probe before giving up.
_nested = getattr(_legacy_cb, 'crossbarhttp', None) if _legacy_cb else None
client = _nested.Client(_wamp_url, timeout=2.0) if (_nested and hasattr(_nested, 'Client')) else None
except Exception as _cb_err:
client = None
try:
import logging as _logging_cb_warn
_logging_cb_warn.getLogger(__name__).warning(
f"Crossbar HTTP publisher init skipped: {type(_cb_err).__name__}: {_cb_err} — "
f"MessageBus LOCAL+PEERLINK transports remain active.")
except Exception:
logging.getLogger(__name__).exception("<module>: swallowed Exception")
# Create thread pool executor for async Crossbar publishing
crossbar_executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix='crossbar_publish')
atexit.register(lambda: crossbar_executor.shutdown(wait=False))
_crossbar_client_lock = threading.Lock()
def _http_crossbar_publish(topic: str, payload: str, timeout: float = 2.0):
"""Use the publisher's timeout; never change another socket's defaults.
An unreachable Crossbar is an ordinary state (an offline node), so a
topic's failure is announced once when it starts and once when it ends,
not as a traceback per publish. Tracked per topic (``.failing_topics``,
under the client lock the publish already holds): one global flag let two
workers both warn, and let one always-failing topic beside healthy ones
flip it -- a warning and a "recovered" line per publish.
"""
if client is None:
return
error = None
# The installed crossbarhttp3 distribution exposes crossbarhttp.Client.
# Its timeout and signed-message sequence are mutable client state.
# Serialize only this cloud leg; local/SSE/PeerLink fanout stays async.
with _crossbar_client_lock:
try:
if hasattr(client, 'timeout'):
previous = client.timeout
try:
client.timeout = timeout
client.publish(topic, payload)
finally:
client.timeout = previous
else:
# Preserve alternate SDK compatibility. Its implementation
# owns its timeout; process-global socket mutation is unsafe.
client.publish(topic, payload)
except Exception as e:
error = e
failing = _http_crossbar_publish.__dict__.setdefault(
'failing_topics', set())
log = logging.getLogger(__name__)
if error is None:
if topic in failing:
failing.discard(topic)
log.info("Crossbar publish recovered for %s", topic)
elif topic in failing or len(failing) >= 1000:
log.debug("Crossbar publish still failing for %s: %s", topic, error)
else:
failing.add(topic)
log.warning("Crossbar publish failing for %s (cloud copy not "
"sent; local/PeerLink delivery unaffected): %s: %s -- "
"repeats are logged at debug until it recovers",
topic, type(error).__name__, error)
# Inject HTTP transport into MessageBus (avoids Layer 2 importing Layer 3)
def _inject_http_transport():
try:
from core.peer_link.message_bus import get_message_bus
bus = get_message_bus()
bus.set_http_transport(lambda t, p: crossbar_executor.submit(_http_crossbar_publish, t, p))
except Exception:
logging.getLogger(__name__).exception("_inject_http_transport: swallowed Exception")
_inject_http_transport()
# Topic resolution uses the canonical resolve_legacy_topic() from message_bus.py
# (single source of truth — no duplicate mapping table here)
def publish_async(topic, message, timeout=2.0, skip_crossbar=False):
"""
Publish to all available transports: LOCAL EventBus + PeerLink + Crossbar.
Routes through MessageBus first (local + multi-device delivery),
then falls back to direct HTTP Crossbar for cloud telemetry.
Works fully offline — LOCAL delivery always succeeds.
Also publishes to the confirmation topic (mirrors cloud chatbot.py:publish()).
SIDE-EFFECT — Nunba monkey-patches this at runtime:
Nunba's ``routes/hartos_backend_adapter.py`` wraps this
function in-place to also call ``_capture_thinking(message)``
for every published message. That feeds Nunba's per-request
thinking-trace buffer that the ``/chat`` HTTP response embeds.
Implication for callers: bypassing this function (publishing
directly via ``MessageBus.publish``) silently skips the
capture. If you change the chat-bubble path to call the bus
directly, Nunba's HTTP /chat responses lose their thinking
traces. Either keep going through ``publish_async`` or
migrate Nunba's interceptor to a bus subscriber on
``chat.response`` (tracked in
``memory/project_publish_aop_migration.md`` — the right
long-term shape).
Args:
topic: Crossbar topic to publish to (legacy format). Use
``core.peer_link.message_bus.chat_topic_for(user_id)``
to build per-user chat topic strings instead of inline
f-strings.
message: Message payload (JSON string or dict)
timeout: Maximum time for HTTP Crossbar publish (default: 2.0 seconds)
skip_crossbar: True keeps the message on this node's own legs -- the
MessageBus (LOCAL + the user's own devices over PeerLink) and the
desktop's SSE -- and off the HTTP Crossbar leg, which reaches
central (Reel Studio decision U4(d): a reel push names her
clients). Every stamp above still applies. Nunba's wrapper must
forward it; callers pass it only when True, so a wrapper that
takes (topic, message, timeout) keeps working for everyone else.
Where a chat reply came from: every chat.* envelope but the voice (a
'TTS' payload) leaves here stamped ``served_by`` = local | hive | cloud
(core.constants.canonical_served_by).
The envelope's own key wins; else the thread's reply_from() context
(hartos.threadlocal) -- a context and NOT a keyword, because the wrapper
Nunba installs above takes (topic, message, timeout) only; else the node's
own LLM. The Crossbar HTTP leg below still sends a bare-text message as
the bare text it always was.
"""
# Parse message if JSON string
data = message
if isinstance(message, str):
try:
data = json.loads(message)
except (json.JSONDecodeError, TypeError):
data = {'raw': message}
# The user the PAYLOAD names, read before the bus leg below stamps the
# topic suffix into it: the egress rule compares it with the owner of
# the Crossbar URI (a payload naming nobody on a declared per-user URI
# is that URI's user's).
_claimed_user = data.get('user_id', '') if isinstance(data, dict) else ''
# 1. Route through MessageBus (LOCAL + PEERLINK — always works offline)
try:
from core.peer_link.message_bus import get_message_bus, resolve_legacy_topic
except ImportError:
resolve_legacy_topic = None
bus_topic, user_id = resolve_legacy_topic(topic) if resolve_legacy_topic else (None, '')
if bus_topic:
# Where a chat reply came from: local | hive | cloud, named by the
# ONE rule (core.constants.canonical_served_by). Stamped before the
# bus is touched so the SSE leg below carries it even when the bus
# is down. The envelope's own key, else the dispatch site's
# reply_from() context, else the node's own LLM. Never absent: the
# page's fallback for a missing value is 'cloud', which badged a
# reply the local llama-server wrote "Cloud" (2026-10-04). Local
# imports: test_egress_one_rule exec's this function's source in a
# namespace that holds none of this module's globals.
# A failure here costs the badge, never the delivery.
# Not the voice: a 'TTS' payload is played and never drawn as a
# bubble, and it reaches here on chat.pupit, from the synthesis
# thread that no longer knows the reply's origin -- a hive expert's
# spoken answer was stamped 'local' on the bus envelope and its
# task.confirmation copy (review of 028a8a959).
if (isinstance(data, dict) and bus_topic.startswith('chat.')
and data.get('action') != 'TTS'):
try:
from core.constants import canonical_served_by
from hartos.threadlocal import thread_local_data as _tld_origin
data['served_by'] = canonical_served_by(
data.get('served_by') or _tld_origin.get_served_by(),
data.get('node_tier'))
except Exception as _origin_err:
app.logger.warning(
f"publish_async: served_by not stamped: {_origin_err}")
try:
bus = get_message_bus()
# Ensure user_id is in data for per-user routing
if user_id and isinstance(data, dict):
data.setdefault('user_id', user_id)
# Stamp node_tier on every chat-topic envelope so the SPA's
# formatTier helper (task #146/#148) can render the badge
# without per-callsite plumbing. Closes #153. setdefault
# preserves any explicit value the caller already set —
# this is a strict additive safety net for the audit gap
# task #149 might have missed.
#
# served_by is stamped above, before the bus is touched: the
# dispatch site names it through reply_from() (speculative_
# dispatcher's expert delivery), canonical_served_by defaults it.
if (isinstance(data, dict)
and bus_topic
and bus_topic.startswith('chat.')):
data.setdefault(
'node_tier',
os.environ.get('HEVOLVE_NODE_TIER', 'flat'))
msg_id = bus.publish(
bus_topic, data,
user_id=user_id,
skip_crossbar=True, # We handle Crossbar below directly
)
# Publish confirmation tracking (mirrors cloud chatbot.py:publish())
if bus_topic not in ('task.confirmation', 'task.progress'):
conf_data = dict(data) if isinstance(data, dict) else {'raw': str(data)}
conf_data['confirmation'] = False
conf_data['topic_name'] = topic
conf_data['msg_id'] = msg_id
bus.publish('task.confirmation', conf_data, skip_crossbar=True)
except Exception as e:
app.logger.debug(f"MessageBus publish failed (offline OK): {e}")
# 2. SSE / local-WAMP fan-out for the Nunba desktop SPA.
#
# MessageBus (above) is the in-process EventBus + PeerLink path —
# it does NOT reach the SPA's EventSource subscribers nor the
# embedded wamp_router :8088 that crossbarWorker.js connects to.
# Nunba's main.py exposes ``broadcast_sse_event`` which fans out to
# BOTH legs (publish_local → embedded WAMP topic, plus the SSE
# queue), so calling the canonical helper here gives every publish
# site real-time delivery to the desktop UI without per-call wiring.
#
# Before this leg existed, only TTS bridged to SSE explicitly
# (see TTS publish below); thinking traces, channel events, and
# every other publish_async caller silently dropped on the desktop.
# The publisher abstraction is the canonical home — no parallel
# path at each call site.
# TTS skips this leg — TTS owns its own broadcast_sse_safe call at the
# TTS publish site (search "TTS async: published" below). Routing TTS
# through here in addition was producing two audio playbacks per turn
# (the SPA's request_id-keyed dedup didn't catch them when envelope
# shapes diverged across transports) and racing with the typewriter
# animation. Keep TTS on its dedicated path; this leg covers
# everything else (thinking traces, channel events, capability updates).
if isinstance(data, dict) and data.get('user_id') and data.get('action') != 'TTS':
try:
from core.platform.events import broadcast_sse_safe
# event_type mirrors the canonical chatbot_routes.publish_to_crossbar
# contract (Nunba routes/chatbot_routes.py:733): use payload['type']
# when present, otherwise default to 'message'. The SPA's SSE
# consumer only registers listeners for 'message' (onmessage),
# 'notification', and 'setup_progress'; everything else is routed
# via the local-WAMP leg of broadcast_sse_event (which keys off
# data.get('action') for TTS pupit / chat / social). Using
# 'action' as the SSE event_type would silently drop on SSE
# clients — keep this aligned with the chatbot_routes default.
_evt = data.get('type') if data.get('type') else 'message'
broadcast_sse_safe(_evt, data, user_id=data.get('user_id'))
except Exception as _sse_err:
app.logger.debug(f"SSE fan-out skipped: {_sse_err}")
# 3. HTTP Crossbar for cloud telemetry + legacy mobile (when internet
# available) -- unless the caller keeps this message on the node.
if client is None or skip_crossbar:
return
raw_message = message if isinstance(message, str) else json.dumps(message, default=str)
# Egress (owner ruling 2026-09-26): a topic that is not the message
# user's own reaches other people's subscribers, so it carries the
# scrubbed copy, or nothing when the scrub failed. Asked of the one rule
# (security.edge_privacy.crossbar_egress_copy) the MessageBus Crossbar
# leg asks; the user's own topic still goes byte-identical.
_wire, _wire_is_json = message, not isinstance(message, str)
if isinstance(message, str):
try:
_wire, _wire_is_json = json.loads(message), True
except (json.JSONDecodeError, TypeError):
pass
try:
from security.edge_privacy import crossbar_egress_copy
_out = crossbar_egress_copy(topic, _wire, _claimed_user)
except ImportError as _ep_err:
app.logger.warning(f"Egress rule unavailable ({_ep_err}); not "
f"publishing {topic} to Crossbar")
return
if _out is None:
return
if _out is not _wire:
raw_message = json.dumps(_out, default=str) if _wire_is_json else _out
crossbar_executor.submit(_http_crossbar_publish, topic, raw_message, timeout)
def _get_dynamic_capability_prompt() -> str:
"""Auto-generate capability context for the LLM based on live node state.
- Bundled (Nunba): includes audio synthesis instructions + dynamic capabilities
- Text-only (Hevolve web/channels): only includes non-audio capabilities
- Empty string if nothing special is available (zero prompt bloat)
New models/services appear automatically — no hardcoding per model.
"""
parts = []
# Audio mode context (Nunba only — verify TTS is actually available)
if _is_bundled() if callable(_is_bundled) else _is_bundled:
_tts_ok = False
try:
from tts.tts_engine import get_tts_engine
_tts_ok = get_tts_engine() is not None
except Exception:
logging.getLogger(__name__).exception("_get_dynamic_capability_prompt: swallowed Exception")
if _tts_ok:
parts.append(
'Your text output is auto-synthesized as audio. You can freely '
'mix languages (the system detects scripts and routes each to '
'the right TTS engine). For rich audio, embed tags: '
'<music genre="jazz" duration="10">mood</music> for music, '
'<sing duration="15">lyrics</sing> for singing voice.')
# Dynamic capabilities from orchestrator (any mode)
try:
from integrations.service_tools.model_orchestrator import get_orchestrator
cap_prompt = get_orchestrator().capability_prompt()
if cap_prompt:
parts.append(cap_prompt)
except Exception:
logging.getLogger(__name__).exception("_get_dynamic_capability_prompt: swallowed Exception")
return '\n'.join(parts)
# create prompt
def create_prompt(tools):
# No per-query classification here — the 30s cache + 1.5s hot-path
# budget inside get_user_context are the only speed wins. Repeat
# constructions within the TTL window never re-hit the backend.
user_details, actions = get_action_user_details(
user_id=thread_local_data.get_user_id())
# Build dynamic identity from the ACTIVE agent's config (prompts/{id}.json),
# loaded by the request's prompt_id via the mtime-cached, self-refreshing
# core.cache_loaders.load_agent_config. Replaces a dead threadlocal stub
# (get_agent_config was never defined → hasattr always False → always None),
# which silently dropped the agent's persona from casual replies even though
# the full-agent path injects it via build_personality_prompt. Fenced: the
# hot path never breaks if the load hiccups.
try:
from core.cache_loaders import load_agent_config as _load_agent_config
_active_agent_config = _load_agent_config(thread_local_data.get_prompt_id())
except Exception:
_active_agent_config = None
_owner_name = extract_owner_name(user_details)
_dynamic_identity = build_identity_prompt(_active_agent_config, _owner_name, user_details)
prefix = f"""Below is an instruction that describes a task, paired with an input that provides further context. Write a response that appropriately completes the request.
<GENERAL_INSTRUCTION_START>
{_dynamic_identity}
Consider the consequences of each response you provide.
Your answers must be meaningful and delivered as quickly as possible.
Never refer to the user as a human or yourself as mere AI.
Your response should not be more than 200 words.
{get_cultural_prompt_compact()}
<GENERAL_INSTRUCTION_END>
User details:
<USER_DETAILS_START>
{user_details}
<USER_DETAILS_END>
<CONTEXT_START>
You can help with anything — answering questions, teaching, coding, research, creative writing, data analysis, building agents, brainstorming, planning, and more.
Your expertise draws from various knowledge sources like books, websites, and white papers. Your responses will be conveyed to the user through a video, using an avatar and text-to-speech technology, and can be translated into various languages.
Consider the user's location, time and context of previous dialogues with time to create a proper prompt for tools and follow up in-context questions. You have the ability to see using Visual_Context_Camera tool.
If your response contains abbreviated words, please separate them with spaces, like T T S.
{_get_dynamic_capability_prompt()}
<CONTEXT_END>
These are all the actions that the user has performed up to now:
<PREVIOUS_USER_ACTION_START>
{actions}
Conversation History:
<HISTORY_START>
"""
suffix = """
<HISTORY_END>
Only if this above conversation history is not sufficient to fulfill the user's request then use below FULL_HISTORY tool. Important: If results can be accomplished with above information skip tools section and move to format instructions.
TOOLS
------
Assistant can use tools to look up information that may be helpful in answering the user's
question. The tools you can use are:
<TOOLS_START>
{{tools}}
<TOOLS_END>
<FORMAT_INSTRUCTION_START>
{format_instructions}
<FORMAT_INSTRUCTION_END>
always create parsable output.
Here is the User and AI conversation in reverse chronological order:
USER'S INPUT:
-------------
<USER_INPUT_START>
Latest USER'S INPUT For which you need to respond (consult recent history only when needed for more context): {{{{input}}}}
<USER_INPUT_END>
"""
prompt = ConversationalChatAgent.create_prompt(
tools,
system_message=prefix,
human_message=suffix,
input_variables=["input", "agent_scratchpad", "chat_history"]
)
# prompt_string = prompt.render()
# prompt.rende
return prompt
def _suggest_share_worthy_content(input_text: str) -> str:
"""Tool handler: find high-engagement, under-shared posts the user could share.
Queries the social DB for posts with strong engagement (upvotes > 5,
comments > 3) but few share links (< 3), then returns a human-friendly
suggestion.
"""
try:
from integrations.social.models import get_db, Post, ShareableLink
from sqlalchemy import func, outerjoin
db = get_db()
try:
# Subquery: count share links per post
share_counts = (
db.query(
ShareableLink.resource_id,
func.count(ShareableLink.id).label('link_count'),
)
.filter(ShareableLink.resource_type == 'post')
.group_by(ShareableLink.resource_id)
.subquery()
)
# Main query: high engagement, low shares, not deleted/hidden
posts = (
db.query(Post, share_counts.c.link_count)
.outerjoin(share_counts, Post.id == share_counts.c.resource_id)
.filter(
Post.is_deleted == False,
Post.is_hidden == False,
Post.upvotes > 5,
Post.comment_count > 3,
)
.filter(
(share_counts.c.link_count == None) | # noqa: E711
(share_counts.c.link_count < 3)
)
.order_by(Post.score.desc())
.limit(3)
.all()
)
if not posts:
return ("No under-shared high-engagement content found right now. "
"Keep creating great posts and the community will notice!")
suggestions = []
for post, link_count in posts:
title = (post.title or post.content or '')[:80].strip()
shares = link_count or 0
suggestions.append(
f"- \"{title}\" ({post.upvotes} upvotes, "
f"{post.comment_count} comments, only {shares} shares) "
f"[post_id: {post.id}]"
)
header = ("These posts are resonating with the community but haven't "
"been shared much yet. Consider sharing them with your network:\n")
return header + "\n".join(suggestions)
finally:
db.close()
except Exception as e:
logging.debug(f"Suggest_Share_Worthy_Content failed: {e}")
return f"Could not fetch share-worthy content right now: {e}"
def _observe_user_experience(input_text: str) -> str:
"""Record a user experience observation for self-improvement."""
try:
import json as _json
data = _json.loads(input_text)
event = data.get('event', 'unknown')
page = data.get('page', '')
duration_ms = data.get('duration_ms', 0)
outcome = data.get('outcome', '')
observation = f"User {event} on {page} ({duration_ms}ms): {outcome}"
# Use MemoryGraph if available
try:
user_id = thread_local_data.get_user_id()
prompt_id = thread_local_data.get_prompt_id()
graph = _get_or_create_graph(user_id, prompt_id)
if graph:
session_id = f"{user_id}_{prompt_id}" if prompt_id else str(user_id)
memory_id = graph.register(
content=observation,
metadata={
'memory_type': 'observation',
'source_agent': 'agent',
'session_id': session_id,
'page': page,
'event': event,
},
context_snapshot=f"UX observation during session {session_id}",
)
return f"Observation recorded (id: {memory_id}): {observation}"
except Exception:
logging.getLogger(__name__).exception("_observe_user_experience: swallowed Exception")
return f"Observation noted: {observation}"
except Exception:
return f"Observation noted: {input_text}"
def _self_critique_and_enhance(input_text: str) -> str:
"""Review past suggestions and outcomes to improve future behavior."""
try:
user_id = thread_local_data.get_user_id()
prompt_id = thread_local_data.get_prompt_id()
graph = _get_or_create_graph(user_id, prompt_id)
if not graph:
return "Self-critique unavailable: no memory graph for this session."
session_id = f"{user_id}_{prompt_id}" if prompt_id else str(user_id)
# Recall past suggestions and observations
suggestions = graph.recall(input_text or 'suggestions made outcomes', mode='semantic', top_k=10)
observations = graph.recall('user experience observation', mode='semantic', top_k=10)
if not suggestions and not observations:
return "No past interactions to critique yet. Will observe and learn."
# Format findings for agent reasoning
critique = "Self-critique findings:\n"
if suggestions:
critique += f"Past suggestions ({len(suggestions)}):\n"
for s in suggestions[:5]:
critique += f" - {s.content[:100]}\n"
if observations:
critique += f"User observations ({len(observations)}):\n"
for o in observations[:5]:
critique += f" - {o.content[:100]}\n"
# Store the critique itself as an insight
insight = f"Self-critique on: {input_text}"
graph.register(
content=insight,
metadata={
'memory_type': 'insight',
'source_agent': 'agent',
'session_id': session_id,
'type': 'self_critique',
},
context_snapshot=f"Self-critique during session {session_id}",
)
return critique
except Exception as e:
return f"Self-critique unavailable: {str(e)}"
# Single source of truth for the Create_Agent tool description.
# Previously duplicated at lines ~4580 and ~5042 — a DRY violation
# that meant fixing the trigger heuristic required two edits and
# allowed the two copies to drift.
#
# The wording is intentionally restrictive (PERSISTENT + MULTI-STEP)
# because the prior version triggered for any "create / build / set
# up / deploy" verb — including code-generation requests like "write
# a palindrome function" that should be answered inline by the chat
# LLM, not routed into the create_recipe peer-review pipeline. The
# 2026-05-28 palindrome cycle (request_id c4b09d1f-…) is the
# repro: 4 review rounds + 1 escalation = 10 thinking-events, then
# the force-approved plan called execute_windows_or_android_command
# 9 times trying to type a palindrome check into a terminal.
_CREATE_AGENT_TOOL_DESCRIPTION = (
"Use this tool ONLY when the user wants a PERSISTENT MULTI-STEP agent "
"that runs over time or across messages — e.g. 'set up a daily news "
"summarizer', 'create a trading bot that posts updates', 'build an "
"agent that monitors my inbox and replies to customer questions'.\n\n"
"Do NOT use this tool for any of:\n"
" - One-shot code generation: 'write a palindrome function', "
" 'write a quicksort', 'show me a Fibonacci implementation'. "
" Just write the code in your chat reply.\n"
" - Single questions / explanations: 'what is X', 'explain Y', "
" 'how does Z work'. Answer directly.\n"
" - Math / logic computations: palindrome checks, factorials, "
" string reversal, search algorithms — compute the answer "
" in your response, don't spawn an agent.\n"
" - Direct documentation, refactoring, code review of pasted text.\n\n"
"Input should be the description of what the persistent agent should do.\n"
"If the user uses keywords like 'automatically', 'autonomous', "
"'do it for me', 'handle it', 'just create it', include them in your "
"input so the downstream pipeline runs in autonomous mode."
)
def _without_agent_work_tools(tools):
"""``tools`` minus the agent-work pair (core.constants.AGENT_WORK_TOOL_NAMES:
Create_Agent, an interview or an autonomous build saved under the
caller's id, and Agentic_Router, a Plan Mode consent question); every
other tool keeps its place. Objects without a ``name`` are kept. Both
get_tools branches offer the pair to a person; a metered Hive API call
must not get it, since its caller is a program that can answer none of
those and the tool's output would be the billed reply (task #120; the
review of 7e7ce19bc drove it). The ONE filter both branches apply."""
return [t for t in tools if getattr(t, 'name', None) not in AGENT_WORK_TOOL_NAMES]
def _handle_create_agent_tool(input_text):
"""Tool handler: LLM decided user wants to create an agent.
Called by the LangChain agent when it detects agent creation intent.
Sets thread-local flags that the /chat handler checks after get_ans() returns.
"""
lower = input_text.lower()
autonomous = any(w in lower for w in [
'autonomous', 'automatic', 'automatically', 'do it for me',
'handle it', 'just create', 'create it yourself', 'auto',
])
thread_local_data.set_creation_requested(description=input_text, autonomous=autonomous)
if autonomous:
return f"Agent creation initiated autonomously for: {input_text}. I will set up the agent creation workflow and handle all the details automatically."
return f"Agent creation initiated for: {input_text}. I will set up the agent creation workflow. Let me gather the required details."
# Signals from reuse agent that suggest creating a new agent
_RESPONSE_CREATION_SIGNALS = [
'need a new agent', 'create a new agent', 'requires a different agent',
'beyond my capabilities', 'specialized agent', 'need a specialized',
'suggest creating', 'recommend creating a new',
]
def _response_signals_creation(response_text):
"""Check if the reuse agent's response suggests creating a new agent."""
lower = response_text.lower()
return any(signal in lower for signal in _RESPONSE_CREATION_SIGNALS)
# ─── Perception Watchers (Future TTL) ───────────────────────────────────
_active_watchers = {} # user_id -> [{trigger_id, expires_at, condition, action, modality, callback}]
def _handle_visual_watcher_tool(input_text):
"""Register a visual/audio watcher trigger with TTL.
Input format: 'CONDITION: <condition> | ACTION: <action> | TTL: <minutes>'
Optional: '| MODALITY: visual|audio|both' (defaults to visual)
"""
import re
import time as _time
parts = {}
for segment in input_text.split('|'):
segment = segment.strip()
if ':' in segment:
key, val = segment.split(':', 1)
parts[key.strip().upper()] = val.strip()
condition = parts.get('CONDITION', input_text)
action_text = parts.get('ACTION', 'notify user')
ttl_minutes = int(parts.get('TTL', '30'))
modality = parts.get('MODALITY', 'visual').lower()
user_id = thread_local_data.get_user_id() or 'default'
trigger_id = f'watcher_{user_id}_{int(_time.time())}'
expires_at = _time.time() + (ttl_minutes * 60)
def _on_trigger(description, **kwargs):
"""Callback fired when visual trigger matches."""
try:
from core.platform.events import emit_event
emit_event('tts.speak', {'user_id': user_id, 'text': action_text})
emit_event('perception.watcher.fired', {
'user_id': user_id, 'condition': condition,
'action': action_text, 'trigger_id': trigger_id,
})
except Exception:
logging.getLogger(__name__).exception("_on_trigger: swallowed Exception")
# Register with VisionService for visual watchers
if modality in ('visual', 'both'):
try:
from integrations.vision.vision_service import VisionService
from core.platform.registry import ServiceRegistry
vs = ServiceRegistry.get('VisionService')
if vs:
condition_words = [w for w in condition.lower().split() if len(w) > 3]
vs.register_visual_trigger(
channel='camera', callback=_on_trigger,
keywords=condition_words, cooldown_seconds=5,
name=trigger_id,
)
except Exception:
logging.getLogger(__name__).exception("_handle_visual_watcher_tool: swallowed Exception")
watcher_entry = {
'trigger_id': trigger_id, 'expires_at': expires_at,
'condition': condition, 'action': action_text,
'modality': modality, 'callback': _on_trigger,
}
if user_id not in _active_watchers:
_active_watchers[user_id] = []
_active_watchers[user_id].append(watcher_entry)
return (
f"Watcher registered: watching for '{condition}' → will '{action_text}'. "
f"TTL: {ttl_minutes} minutes. Modality: {modality}."
)
def _evaluate_audio_watchers(user_id, transcript):
"""LLM-powered evaluation of audio watchers against transcript."""
import time as _time
watchers = [w for w in _active_watchers.get(user_id, [])
if w['modality'] in ('audio', 'both') and _time.time() < w['expires_at']]
if not watchers:
return
conditions = "\n".join(f"- Watcher {i+1}: {w['condition']}" for i, w in enumerate(watchers))
try:
llm = get_llm(temperature=0.1, max_tokens=200)
result = llm.invoke(
f"The user just said: \"{transcript}\"\n\n"
f"Active watchers:\n{conditions}\n\n"
f"Which watcher conditions (if any) are semantically triggered by what the user said? "
f"Return ONLY a JSON array of triggered watcher numbers, e.g. [1, 3]. "
f"Return [] if none match."
)
import re as _re
text = (result.content if hasattr(result, 'content') else str(result)).strip()
match = _re.search(r'\[.*?\]', text)
if match:
triggered = json.loads(match.group())
for idx in triggered:
if 1 <= idx <= len(watchers):
w = watchers[idx - 1]
if w.get('callback'):
w['callback'](transcript)
except Exception as e:
app.logger.debug(f"Audio watcher eval failed: {e}")
def _push_workflow_flowchart(user_id, prompt_id, request_id=None):
"""Push recipe JSON to frontend via Crossbar for interactive flowchart rendering."""
try:
recipe_path = os.path.join(PROMPTS_DIR, f'{prompt_id}.json')
if not os.path.isfile(recipe_path):
return
with open(recipe_path) as f:
recipe_data = json.load(f)
crossbar_message = {
"text": ["Agent workflow ready"],
"priority": 50,
"action": "WorkflowFlowchart",
"recipe": recipe_data,
"request_id": request_id or "0",
"prompt_id": prompt_id,
"historical_request_id": [],
"options": [], "newoptions": [],
}
from core.peer_link.message_bus import chat_topic_for
publish_async(chat_topic_for(user_id), json.dumps(crossbar_message))
except Exception:
logging.getLogger(__name__).exception("_push_workflow_flowchart: swallowed Exception")
def _request_consent(action: str, label: str, input_text: str) -> str:
"""Ask this computer's owner for a capability, through ConsentService.
Same canonical path as every other ask -- integrations.vlm.safety
.computer_control_block, vision_service's capture loop, whisper_tool,
social/auth, ai_governance, hive_guardrails -- so the ask is a
UserConsent row the owner can see on the privacy page and revoke, it
carries an immutable audit entry, and it leaves as `consent.request`,
the one payload type every surface renders.
It used to push a bare `{'type': 'approval'}` component through
LiquidUIService with a broadcast_sse_safe fallback and touch no database
at all. That meant: nothing to revoke, nothing on the privacy page, and
a payload only AgentOverlay knows -- so the floating companion, which
exists precisely for when Nunba is NOT the window in front, dropped every
camera and screen ask (#863). The canonical emit that fixed the sibling
asks landed upstream on 2026-08-26 (e991309da); this path never migrated.
Whose permission: the owner of this desktop (HEVOLVE_OWNER_USER_ID, which
Nunba exports at boot), never the caller -- the camera and the screen are
this machine's, and a remote user must not be asked to lend them. Same
rule and same fail-closed answer as computer_control_block and
/visual_agent (#66). Which agent is asking comes from the request's own
prompt id, not the capability label the tool wrapper passes.
"""
# The ask itself is consent_service.ask_owner, shared with the camera
# tool's own ask (hartos.helper), so the two cannot drift apart.
from integrations.social.consent_service import (
ASK_FAILED, NO_OWNER, NOT_HERE, OWNER_ALLOWED, OWNER_SAID_NO,
ask_owner, central_lacks, consent_type_for_action)
log = logging.getLogger(__name__)
consent_type = consent_type_for_action(action)
if consent_type is None:
log.error('_request_consent: %r maps to no consent type', action)
return (f"Could not request {label.lower()} access — this computer "
f"records no permission for '{action}'.")
outcome = ask_owner(consent_type, f'{label} access needed: {input_text}'.strip(),
thread_local_data.get_prompt_id())
if outcome == NOT_HERE:
return f"Not asked: central has no {central_lacks(consent_type)}."
if outcome == NO_OWNER:
log.warning('%s consent not asked: no owner identity '
'(HEVOLVE_OWNER_USER_ID is not set)', label)
return (f"Not asked: nobody is signed in on this computer who could "
f"allow {label.lower()} access.")
if outcome == OWNER_ALLOWED:
return f"{label} access is already allowed — go ahead."
if outcome == OWNER_SAID_NO:
return (f"The owner of this computer said no to "
f"{label.lower()} access.")
if outcome == ASK_FAILED:
return (f"Could not request {label.lower()} access — the permission "
f"system is unavailable.")
return (f"{label} access request sent to the owner of this computer. "
f"Waiting for their answer.")
def _request_capability_consent(input_text: str) -> str:
return _request_consent('enable_camera', 'Camera', input_text)
def _request_screen_consent(input_text: str) -> str:
return _request_consent('enable_screen', 'Screen', input_text)
def _handle_screenshot_tool(input_text: str) -> str:
"""Take screenshot and describe using VLM."""
# AI senses kill-switch: refuse to grab the screen when the human has cut
# the AI's 'screen' sense — "humans are always in control", enforced at the
# grab (not advisory). Mirrors the mic gate at /api/voice.
try:
from core.ai_sensing import allowed
if not allowed('screen'):
return "Screen access is turned off by the user (AI senses kill-switch)."
except Exception:
logging.getLogger(__name__).exception("_handle_screenshot_tool: swallowed Exception")
try:
from PIL import ImageGrab
import base64, io
screenshot = ImageGrab.grab()
buf = io.BytesIO()
from integrations.vlm.local_computer_tool import VLM_IMG_W, VLM_IMG_H
screenshot.resize((VLM_IMG_W, VLM_IMG_H)).save(buf, 'JPEG', quality=50)
b64 = base64.b64encode(buf.getvalue()).decode('ascii')
resp = pooled_post(
GPT_API,
json={
'model': LLM_MODEL_NAME,
'messages': [{'role': 'user', 'content': [
{'type': 'text', 'text': f'Describe what you see on this screen. Question: {input_text}'},
{'type': 'image_url', 'image_url': {'url': f'data:image/jpeg;base64,{b64}'}}
]}],
'max_tokens': 300,
},
headers=LLM_AUTH_HEADERS,
timeout=15,
)
if resp.status_code == 200:
return resp.json().get('choices', [{}])[0].get('message', {}).get('content', 'Could not describe screen.')
return "Screenshot taken but VLM unavailable for description."
except Exception as e:
return f"Screenshot error: {str(e)[:200]}"
def _handle_shell_command_tool(input_text: str) -> str:
"""Execute a shell / PowerShell / bash command on the user's computer
and return its stdout+stderr. This is the FAST path for anything that
can be done programmatically — `notepad file.txt` is one command vs
the 18-iteration VLM loop Computer_Action would do for the same task.
Safety:
- Input is NFKC-normalized + lowercased before the denylist regex
runs, so unicode lookalikes (full-width 'ｒｍ', cyrillic 'р') can't
slip past patterns written in ASCII.
- Hard-denylist of clearly destructive patterns (format, rm -rf /,
shred, dd if=/dev/, mkfs, del /s C:\\ etc.). Denylisted commands
return an explanation without running.
- 30-second timeout. Longer-running work should use Execute_Coding_Task.
- Output truncated to 4000 characters so a `tree /` style flood can't
blow the LLM context budget.
- Logged via _with_tool_logging on the Tool wrapper so every
invocation is auditable.
Input formats:
- 'notepad' / 'calc.exe' / 'explorer C:\\Users' — plain command string
- 'powershell: Get-Process | Select-Object -First 5' — explicit shell
- 'bash: ls -la ~ | head' — explicit shell
On Windows we use cmd.exe by default; 'powershell:' prefix switches to
pwsh. On Unix we use /bin/sh; 'bash:' prefix switches to bash.
"""
import re as _re_shell
import unicodedata as _unicodedata
if not input_text or not isinstance(input_text, str):
return "Shell_Command: empty input. Pass a command string like 'notepad file.txt'."
text = input_text.strip()
# Explicit shell selector, in EITHER spelling. Both resolve HERE, into
# the one argv builder below — a second dispatch path is exactly the
# drift this closes.
#
# colon 'powershell: <cmd>' this tool's own documented form
# native 'powershell -Command "<cmd>"' what a model actually writes
#
# D73, live-measured 2026-09-11 (agent 89091774807, 09:55:19). Only the
# colon form was understood, so the native form fell through to the
# Windows default and ran nested as
# `cmd /c powershell -Command "..."`. Reproduced byte-exact:
#
# rc = 0
# stdout = 'Get-PSDrive -Name C | Select-Object -ExpandProperty FreeGB'
# stderr = ''
#
# Exit 0 with the command echoed back as its own output — worse than an
# error, because every honesty gate downstream reads it as SUCCESS. The
# VLM concluded "The output shows 'FreeGB : 100.0'" (a figure present
# nowhere in that output), reported exit_reason=done, FAB-GUARD passed
# the action because the tool HAD executed, and the user was told the
# machine had 127.4 GB free against a real 6.32 GB.
#
# Dispatched directly the same command returns rc=1 with 'Property
# "FreeGB" cannot be found' — a failure the model can act on.
shell_override = None
# 'shell:' / 'terminal:' name the DEFAULT shell: the VLM loop's models
# write the tool's name in front of the command ("shell: dir", measured
# 2026-10-04 in the daemon's runs), and cmd.exe answered "'shell:' is not
# recognized" to every one. They select nothing (the argv builder below
# falls to its default for any other spelling) and only lose the prefix.
m = _re_shell.match(r'^(powershell|pwsh|bash|sh|cmd|shell|terminal)\s*:\s*(.+)$',
text, _re_shell.IGNORECASE)
if m:
shell_override = m.group(1).lower()
text = m.group(2).strip()
# Having named no shell, a 'shell:' / 'terminal:' command is read exactly
# like an unprefixed one, the native form included. Stripping the prefix
# and stopping there ran `shell: powershell -Command "Get-Date -Format
# yyyy"` as `cmd /c powershell -Command "..."`: exit 0 with the command
# text as its output, the D73 false success below (review of 0cfa782cf,
# measured on the MSI 2026-10-05; before that commit the same input failed
# honestly with exit 1).
if shell_override in (None, 'shell', 'terminal'):
# -enc / -EncodedCommand is deliberately NOT accepted as a selector:
# it is obfuscation, and leaving it unstripped keeps it in front of
# the denylist pattern that blocks it. Only the execute-this-string
# switches (-Command / -c / /c) are consumed.
m = _re_shell.match(
r'^(powershell|pwsh|bash|sh|cmd)(?:\.exe)?\s+'
r'(?:-(?:NoProfile|NonInteractive|NoLogo|ExecutionPolicy\s+\S+|'
r'WindowStyle\s+\S+)\s+)*'
r'(?:-Command|-c|/c)\s+(.+)$',
text, _re_shell.IGNORECASE)
if m:
shell_override = m.group(1).lower()
inner = m.group(2).strip()
# Strip ONE matching pair of wrapping quotes, as the shell would.
if len(inner) >= 2 and inner[0] == inner[-1] and inner[0] in '"\'':
inner = inner[1:-1].strip()
# The denylist below now sees the INNER command. Its patterns are
# substring searches, so destructive text is still caught — pinned
# by TestNativeShellInvocationIsUnderstood's denylist cases.
text = inner or text
# --- Denylist of destructive patterns (case-insensitive, conservative)
_DENY_PATTERNS = [
r'\bformat\s+[a-z]:', # format C:
r'\brm\s+-rf\s+/', # rm -rf / (anywhere in string)
r'\brm\s+-rf\s+--no-preserve-root',
r'\brm\s+-rf\s+~', # rm -rf ~
r'\bdel\s+/s\s+/q\s+[a-z]:\\', # del /s /q C:\
r'\brmdir\s+/s\s+/q\s+[a-z]:\\',
r'\bmkfs\.', # mkfs.*
r'\bdd\s+if=/dev/(zero|random|urandom)\s+of=/dev/[sh]d',
r'\bshred\b',
r':\(\)\s*\{\s*:\|:&\s*\}\s*;:', # fork bomb
r'\bhalt\b', r'\bpoweroff\b',
r'\breboot\b', # conservative — often benign but blocking
r'\bshutdown\s+-[srh]',
r'\breset-computermachinepassword', # dangerous ps cmdlet (lowercased)
r'\bformat-volume',
r'\bremove-item\s+-recurse\s+-force\s+[a-z]:\\',
# Chaining patterns — ethical hacker flagged that 'echo hi && rm -rf /'
# bypassed the old $-anchored pattern. These catch destructive commands
# anywhere in a chained expression (&&, ;, |, ||, backtick, $()).
r'\bmv\s+/\s+/dev/null', # mv / /dev/null
r'\bchmod\s+-[rR]\s+000\s+/', # chmod -R 000 /
r'\bfind\s+/\s+-delete', # find / -delete
r'\bcurl\s+.*\|\s*bash', # curl ... | bash (pipe to shell)
r'\bwget\s+.*\|\s*bash', # wget ... | bash
# Interpreter wrappers — ethical hacker found python -c / perl -e / ruby -e
# bypass all above patterns by wrapping destructive code inside an interpreter.
r'\bpython[23]?\s+-c\s', # python -c "os.system(...)"
r'\bperl\s+-e\s', # perl -e "system(...)"
r'\bruby\s+-e\s', # ruby -e "system(...)"
r'\bnode\s+-e\s', # node -e "require('child_process')..."
r'\bpowershell.*-enc', # powershell -EncodedCommand (obfuscation)
]
# NFKC maps full-width / compatibility chars to their ASCII equivalents
# ('ｒｍ' → 'rm', '＞' → '>'), defeating the common homoglyph bypass.
# We also strip zero-width characters (ZWJ, ZWNJ, BOM) that would
# otherwise split matches like 'r\u200cm'. Strip control chars too.
#
# IMPORTANT: we normalize for the denylist CHECK only. The command
# actually executed is still the raw user text — so a legitimate
# filename with ligatures ('file.txt') or full-width chars still
# resolves to the real file on disk. If the raw text happened to
# contain a homoglyph version of a denylisted token, the denylist
# sees it after normalization and blocks before execution.
_normalized = _unicodedata.normalize('NFKC', text)
_normalized = ''.join(
ch for ch in _normalized
if _unicodedata.category(ch) not in ('Cf', 'Cc') or ch in ('\t', '\n')
)
text_l = _normalized.lower()
for pat in _DENY_PATTERNS:
if _re_shell.search(pat, text_l):
return (
f"Shell_Command refused: the command matches a destructive "
f"pattern ('{pat}') and is blocked by the built-in safety list. "
f"If you really need this, ask the user to run it manually."
)
# --- The shared hard-deny policy every computer-use dispatcher calls
# (power/reset/erase, and stopping the assistant's own processes, #877).
# The LangChain Shell_Command tool reached run_bounded without it.
from integrations.vlm.safety import computer_operation_refusal
_refusal = computer_operation_refusal({'command': text})
if _refusal is not None:
return f"Shell_Command refused: {_refusal}"
# --- The owner's permission: the same computer_control consent the VLM
# loop checks (integrations.vlm.safety.computer_control_block). After
# the denylist on purpose, so a destructive command is refused without
# asking anyone. Inside a VLM run this thread's prompt_id is the run's
# agent (run_local_agentic_loop sets it); for the LangChain tool it is
# the chat turn's agent, stamped by the /chat handler.
from integrations.vlm.safety import computer_control_block
_refusal = computer_control_block(thread_local_data.get_prompt_id())
if _refusal is not None:
return f"Shell_Command not run: {_refusal}"
# --- Choose shell + argv
if sys.platform == 'win32':
if shell_override in ('powershell', 'pwsh'):
argv = ['powershell', '-NoProfile', '-Command', text]
elif shell_override in ('bash', 'sh'):
argv = ['bash', '-c', text]
else:
argv = ['cmd', '/c', text]
else:
if shell_override in ('powershell', 'pwsh'):
argv = ['pwsh', '-NoProfile', '-Command', text]
elif shell_override == 'bash':
argv = ['bash', '-c', text]
else:
argv = ['/bin/sh', '-c', text]
# Bounded through core.subprocess_safe — a plain subprocess.run(timeout=30)
# CANNOT enforce that 30s on Windows. When the deadline fires, run() kills
# the direct child and then calls process.communicate() a SECOND time with
# NO timeout (CPython subprocess.py:559) to drain the pipes. kill() does
# not reach a grandchild, and a surviving grandchild holds the inherited
# stdout/stderr WRITE handles open, so the reader threads never see EOF and
# that drain blocks forever — the `except TimeoutExpired` below could never
# be reached. Measured live 2026-09-09: agent 33323830039's reuse turn sat
# in exactly this frame for 21 minutes while llama-server was idle,
# wedging that user's whole chat turn (thread dump, D35).
#
# run_bounded kills AND explicitly closes the parent-side handles, which is
# what actually releases the readers; its boundedness is already proven by
# tests/unit/test_subprocess_safe.py::TestRunProbeBoundedness. It also
# pins stdin=DEVNULL, so a command that unexpectedly reads stdin gets EOF
# instead of hanging — a second way this could freeze, also closed.
#
# argv still routes through the chosen shell, and run_bounded never passes
# shell=True, so the string is not double-parsed. no_window_kwargs() is
# applied INSIDE run_bounded; repeating it here would be a second copy of
# that decision.
# Announce it like any other computer-use step, instead of poking the
# ribbon directly. Two things were wrong with the bare poke: the floating
# companion window, which reads the computer_use.update topic, never heard
# shell commands at ALL; and its `finally` hide took the ribbon down even
# when this ran INSIDE a VLM run that was still driving the machine.
# current_run joins that enclosing run when there is one (this tool
# reaches here on the loop's own thread) and opens a run of its own when
# there is not, so neither case needs a branch here.
from integrations.vlm.activity_stream import current_run
_caption = f"Shell command: {text[:60]}"
_run = current_run(user_id=thread_local_data.get_user_id() or '',
prompt_id=thread_local_data.get_prompt_id() or '')
_run.step(iteration=1, action='shell', phase='executing', caption=_caption)
_phase, _err = 'completed', ''
# Inside a computer-use run the command runs in the run's workspace
# (hartos.threadlocal workspace, set by integrations.vlm.local_loop),
# where the run's file actions resolve too: the prompt's route for a
# refused python -c is "write_file the script, then shell it".
# Measured 2026-10-05: every shell step ran in the process cwd, the
# install folder. Only an existing folder is handed to the OS; a missing
# one would fail the spawn and read as "interpreter not found".
# Outside a run (the chat's Shell_Command) nothing changes.
_bounded_kw = {}
_workspace = thread_local_data.get_workspace()
if isinstance(_workspace, str) and _workspace and os.path.isdir(_workspace):
_bounded_kw['cwd'] = _workspace
try:
proc = run_bounded(argv, timeout=SHELL_COMMAND_TIMEOUT_S, **_bounded_kw)
except FileNotFoundError as e:
_phase, _err = 'failed', f'interpreter not found — {e}'
return f"Shell_Command: interpreter not found — {e}"
except Exception as e:
_phase, _err = 'failed', f'{type(e).__name__}: {str(e)[:200]}'
return f"Shell_Command error: {type(e).__name__}: {str(e)[:200]}"
finally:
_run.step(iteration=1, action='shell', phase=_phase,
caption=_caption, error=_err)
# Only closes a run this call opened; inside a VLM run the loop still
# owns the ribbon and the ledger task.
_run.finish(exit_reason='done' if _phase == 'completed' else 'action_error',
iteration=1, action='shell', error=_err)
# run_bounded never raises TimeoutExpired — it reports the kill this way.
if proc.timed_out:
return (
f"Shell_Command timed out after {SHELL_COMMAND_TIMEOUT_S}s. "
"For long-running work use "
"Execute_Coding_Task instead, which has a longer budget."
)
out = (proc.stdout or '').strip()
err = (proc.stderr or '').strip()
rc = proc.returncode
# Truncate combined output to 4000 chars max
combined = []
if out:
combined.append(out[:2000])
if err:
combined.append(f"[stderr]\n{err[:2000]}")
body = '\n'.join(combined) if combined else '(no output)'
header = f"Exit code: {rc}"
if len(body) > 4000:
body = body[:4000] + '\n...(truncated)'
return f"{header}\n{body}"
def _handle_navigate_app_tool(input_text: str) -> str:
"""Resolve a spoken destination to a Nunba SPA route and attach a
structured ui_action the frontend LiquidActionBar can render.
The heavy lifting lives in integrations.ui_actions.navigate_tool so
autogen and LangChain share exactly one code path. Thread-local
ui_actions are read by the /chat handler on the way out."""
try:
from integrations.ui_actions import navigate_tool as _nt
role = getattr(thread_local_data, 'get_user_role', lambda: 'flat')() or 'flat'
return _nt.handle_navigate_app(
input_text, user_role=role, thread_local=thread_local_data,
)
except ImportError:
return "Navigate_App is not available on this node."
except Exception as e:
return f"Navigate_App error: {str(e)[:200]}"
def _handle_computer_action_tool(input_text: str) -> str:
"""Execute GUI action on user's computer via VLM agentic loop.
Reuses the same pipeline as execute_windows_or_android_command (autogen):
vlm_adapter → local_loop → point_and_act → pyautogui.
"""
user_id = thread_local_data.get_user_id()
prompt_id = thread_local_data.get_prompt_id()
try:
os.environ.setdefault('HEVOLVE_VLM_UNIFIED', 'true')
from integrations.vlm.vlm_adapter import (
execute_vlm_instruction, resolve_task_workspace)
# 180s is the floor for cold-start runs: first VLM forward pass alone
# can take 8-15s on the 4B model, and a realistic 12-iteration loop
# needs ~2.5 minutes. HEVOLVE_COMPUTER_ACTION_ETA lets the user raise
# it further without touching code.
max_eta = int(os.environ.get('HEVOLVE_COMPUTER_ACTION_ETA', '180'))
message = {
'instruction_to_vlm_agent': input_text,
'workspace_root': resolve_task_workspace(prompt_id=prompt_id),
'user_id': str(user_id or 'guest'),
'prompt_id': str(prompt_id or 0),
'os_to_control': 'windows' if sys.platform == 'win32' else (
'macos' if sys.platform == 'darwin' else 'linux'),
'max_ETA_in_seconds': max_eta,
}
result = execute_vlm_instruction(message)
if not (result and isinstance(result, dict)):
return "Computer action completed (no detailed response)."
exit_reason = result.get('exit_reason', 'done')
responses = result.get('extracted_responses', [])
elapsed = result.get('execution_time_seconds', 0)
n_actions = len(responses)
if exit_reason == 'done':
if responses:
last_content = responses[-1].get('content', '')
if isinstance(last_content, str):
return last_content[:500]
return f"Done — {n_actions} actions in {elapsed:.0f}s."
return f"Done in {elapsed:.0f}s."
if exit_reason == 'consent_required':
# The owner has not allowed agents to control this computer
# (integrations.vlm.safety.computer_control_block); the loop's one
# error record says so. Shell_Command needs the same permission,
# so no other route is offered.
return next((r.get('content') for r in responses
if r.get('type') == 'error' and r.get('content')),
'Not run: the owner has not allowed agents to control '
'this computer.')
# Non-done paths: be honest to the router so it doesn't confabulate.
# The router sees this string as the tool's final answer and should
# turn it into a truthful user reply instead of pretending success.
tail_action = None
for resp in reversed(responses):
if resp.get('type') == 'action':
tail_action = resp.get('content', {})
break
reason_msg = {
'timeout': f"I tried for {elapsed:.0f}s ({n_actions} actions) but ran out of time before finishing.",
'max_iterations': f"I tried {n_actions} actions but the task didn't reach a clear completion.",
'action_error': f"I hit errors on 3 consecutive actions after trying {n_actions} steps and stopped to avoid breaking things.",
'grounding_failed': f"I couldn't reliably locate the UI element after {n_actions} attempts.",
}.get(exit_reason, f"Loop exited with reason '{exit_reason}' after {n_actions} actions.")
if tail_action:
reason_msg += f" Last attempt: {tail_action.get('action', '?')} ({tail_action.get('reasoning', '')[:120]})."
reason_msg += " Want me to try a different approach, or should I describe the steps so you can do it manually?"
return reason_msg
except ImportError:
return "Computer use unavailable — VLM adapter not installed."
except Exception as e:
return f"Computer use error: {str(e)[:200]}"
def _handle_list_pending_actions_tool(input_text: str) -> str:
"""List the user's pending scheduled messages / reminders / jobs.
Answers questions like "what's next for me?", "what reminders do I
have today?", "show my upcoming tasks". Pulls from
ScheduledMessageManager (now durable via JSON persistence) and
apscheduler jobs if the scheduler service is running. The LLM gets
a plain-text summary with ISO timestamps so it can re-format for
the user in whatever style fits.
Input: optional filter hint like 'today', 'this week', 'tomorrow',
or '' for all pending. The filter is lexical — the tool lists
every pending message and lets the LLM narrow if it wants.
"""
try:
hint = (input_text or '').strip().lower()
lines: list[str] = []
# ── ScheduledMessageManager (durable JSON) ───────────────────
try:
from integrations.channels.automation.scheduled_messages import (
ScheduledMessageManager,
)
mgr = ScheduledMessageManager()
pending = mgr.list_pending()
if pending:
lines.append(f"Scheduled messages ({len(pending)} pending):")
for m in pending[:20]:
when = m.scheduled_time.isoformat() if m.scheduled_time else '?'
chan = m.channel_id or 'default'
snip = (m.content or '')[:120]
lines.append(f" • {when} [{chan}] {snip}")
if len(pending) > 20:
lines.append(f" … and {len(pending) - 20} more")
else:
lines.append("No scheduled messages pending.")
except ImportError:
logging.getLogger(__name__).debug("_handle_list_pending_actions_tool: swallowed ImportError")
except Exception as e:
lines.append(f"(scheduled messages unavailable: {str(e)[:80]})")
# ── APScheduler jobs (if active) ─────────────────────────────
try:
from integrations.channels.scheduler import get_scheduler
sched = get_scheduler()
if sched is not None:
jobs = list(sched.get_jobs())
if jobs:
lines.append(f"\nBackground jobs ({len(jobs)} active):")
for j in jobs[:20]:
next_run = getattr(j, 'next_run_time', None)
when = next_run.isoformat() if next_run else 'no next run'
lines.append(f" • {when} {j.id} {j.name or ''}")
except ImportError:
logging.getLogger(__name__).debug("_handle_list_pending_actions_tool: swallowed ImportError")
except Exception as e:
lines.append(f"(scheduler jobs unavailable: {str(e)[:80]})")
if not lines:
return "Nothing scheduled. Your action queue is empty."
if hint:
lines.insert(0, f"Filter hint: {hint}")
return "\n".join(lines)
except Exception as e:
return f"List_Pending_Actions error: {str(e)[:200]}"
def _wire_qr_pair_emitter(channel_type: str, meta: dict) -> None:
"""Hook the live adapter's existing `set_qr_callback` to the
Liquid UI `agent_ui_update` pipe so the user sees a QR right
where they typed "connect <channel>" — no admin-page navigation.
Single helper for every auth_method='qr_session' channel
(whatsapp, telegram_user, discord_user) — the per-channel adapter
already exposes the callback hook, this just bridges it to the
Liquid UI emitter that the form-based flows already use.
Best-effort: if the adapter isn't registered yet (e.g. fresh
bind before bootstrap), the QR will surface on next adapter
spawn via the same callback.
"""
try:
from core.platform.registry import get_registry as _platform_registry
from integrations.channels.registry import get_registry
_lui = _platform_registry().get_or_none('LiquidUIService')
if _lui is None:
return
adapter = get_registry().get(channel_type)
if adapter is None or not hasattr(adapter, 'set_qr_callback'):
return
# The card is FOR this user; None lets agent_ui_update fall back to
# the resolved owner rather than route to a user named 'system'.
_owner = thread_local_data.get_user_id() or None
user_id = _owner or 'system'
display_name = meta.get('display_name') or channel_type
def _emit_qr(qr: str) -> None:
try:
_lui.agent_ui_update(user_id, {
'type': 'qr_pair',
'channel': channel_type,
'title': f"Scan to connect {display_name}",
'help': (
f"Open {display_name} on your phone → Linked "
"devices → Link a device → scan this code."
),
'qr': qr,
}, user_id=_owner)
except Exception as e:
logger.debug("qr_pair emit failed: %s", e)
adapter.set_qr_callback(_emit_qr)
# Trigger an immediate (re)connect so the adapter spins up its
# WAHA / MTProto session and pushes the first QR via the
# callback. Reuses the same loop pattern the admin
# /reconnect endpoint uses — single source of truth.
try:
import asyncio as _asyncio
loop = _asyncio.new_event_loop()
try:
try:
loop.run_until_complete(adapter.disconnect())
except Exception:
logging.getLogger(__name__).exception("_wire_qr_pair_emitter: swallowed Exception")
loop.run_until_complete(adapter.connect())
finally:
loop.close()
except Exception as e:
logger.debug(
"qr_pair reconnect deferred (adapter spawns on bootstrap): %s",
e,
)
except Exception as e:
logger.debug("Connect_Channel: _wire_qr_pair_emitter failed: %s", e)
# A phone or tablet: its browser (Android / iPhone / iPad / iPod / Mobile) or
# a React Native app's HTTP client (okhttp on Android, CFNetwork on iOS).
# A desktop browser, Nunba's desktop webview and a server-side python client
# match none of these.
_PHONE_USER_AGENT = re.compile(
r'\b(?:Android|iPhone|iPad|iPod|Mobile|okhttp|CFNetwork)\b', re.IGNORECASE)
def _user_agent_is_phone(user_agent) -> bool:
"""Does this User-Agent belong to a phone or tablet? One that is the
device a WhatsApp QR would have to be scanned WITH, so showing it a QR
leaves nothing to scan it."""
return bool(_PHONE_USER_AGENT.search(user_agent or ''))
def _start_gateway_qr_pair_push(channel_type: str, meta: dict,
phone: str = None, owner=None):
"""Link a gateway_qr channel (WhatsApp) from chat, by the method that
works on the device the user is holding.
* A number is known -- given (the phone form, or 'whatsapp +91...'),
HEVOLVE_WHATSAPP_PHONE (operator choice) or the user's profile --
the pair code: chat card, consent.* notification that puts the code
on the phone's clipboard with an "Open WhatsApp" banner, and the iOS
fleet command. One tap on the phone, on any device.
* No number and the request came from a phone or tablet
(_user_agent_is_phone): the form asking for the number. A QR shown
on the phone cannot be scanned by that same phone.
* No number, on a desktop: WhatsApp's linked-device QR (``qr_pair``
cards), refreshed as WhatsApp rotates it (~20s) until it is scanned,
with "Link with phone number" on the card as the alternative.
The gateway serves the two methods on different sockets
(request-pair-code tears the QR socket down), so one attempt uses one.
One watcher polls /api/sessions/<sid>/status; when WhatsApp confirms, it
registers the binding, wires the live adapter and sends the
channel_connected card. Every failure (gateway down, no code, the
window running out, a card or push not delivered) is a warning in the
log, and the ones the user must act on are a toast in chat.
Args:
phone: explicit number for "Link with phone number"; any format,
digits are kept.
owner: the user the cards are for, when the caller is a request
handler without a thread-local user (the re-entry route).
Returns 'pair_code', 'qr' or 'ask_phone': what the user was shown, for
the reply to describe. None when the link could not start (the user
was told why in chat).
"""
import os
import json as _json
import logging as _logging
import threading
import time
# The ONE gateway client (api_channels): same base URL resolution as
# the /whatsapp/qr route and the X-Api-Key a remote WAHA needs. Never
# raises; an unreachable gateway is (None, 503).
from integrations.social.api_channels import (
_proxy_gateway, _whatsapp_gateway_base,
)
# Self-contained logger so the helper works whether the module's
# global `logger` is initialised yet or not (helps in standalone
# probes that load this helper before full HARTOS bootstrap).
_log = _logging.getLogger(__name__)
# _owner is who the cards are FOR (None -> agent_ui_update's resolved
# owner); user_id keeps its 'system' placeholder for the gateway session.
_owner = owner or thread_local_data.get_user_id() or None
user_id = _owner or 'system'
sid = user_id if str(user_id).startswith('user_') else f"user_{user_id}"
display_name = meta.get('display_name') or channel_type
icon = meta.get('icon') or channel_type
color = meta.get('color') or '#25D366'
deeplink = (
'whatsapp://settings/linked-devices'
if channel_type == 'whatsapp'
else meta.get('mobile_deeplink')
)
def _emit(payload: dict) -> None:
"""One card to this user's chat; a missing service is logged."""
try:
from core.platform.registry import get_registry
_lui = get_registry().get_or_none('LiquidUIService')
if _lui is None:
_log.warning("gateway_qr: %s card not delivered, no "
"LiquidUIService", payload.get('type'))
return
_lui.agent_ui_update(user_id, payload, user_id=_owner)
except Exception as e:
_log.warning("gateway_qr: %s card not delivered: %s",
payload.get('type'), e)
def _fail(text: str) -> None:
_log.warning("gateway_qr: %s: %s", channel_type, text)
_emit({'type': 'toast', 'severity': 'error',
'channel': channel_type, 'channel_type': channel_type,
'message': text})
# The number, in order: given, the operator's, the user's profile.
# WhatsApp's pair-code endpoint REQUIRES E.164 digits-only.
phone = phone or os.environ.get('HEVOLVE_WHATSAPP_PHONE', '') or ''
if not phone and _owner:
try:
from integrations.social.models import db_session, User as _SocialUser
with db_session(commit=False) as db:
u = db.query(_SocialUser).filter_by(id=str(_owner)).first()
phone = (getattr(u, 'phone_number', '') or getattr(u, 'phone', '') or '') if u else ''
except Exception as e:
_log.warning("gateway_qr: profile phone not read for %s (%s); "
"linking without it", _owner, e)
phone = ''.join(ch for ch in str(phone) if ch.isdigit())
if not phone and _user_agent_is_phone(thread_local_data.get_client_user_agent()):
# On the phone itself: ask for the number, then send the pair code.
# The form posts to the re-entry route, which calls back here with
# the number.
_emit({
'type': 'form',
'title': f"Connect {display_name}",
'channel': channel_type,
'fields': [{
'name': 'phone',
'label': f'Your {display_name} number, with country code',
'placeholder': '+91 ...',
'help': ("Used once to get a linking code for this phone. "
"Not stored."),
'type': 'tel',
'secret': False,
}],
'submit_label': 'Send me a code',
'action': f'/api/social/channels/{channel_type}/connect-pair-code',
})
return 'ask_phone'
base = _whatsapp_gateway_base()
started, start_status = _proxy_gateway('POST', f'/api/sessions/{sid}/start')
if started is None or start_status >= 400:
_fail(f"Couldn't reach the {display_name} link service on this "
f"computer (HTTP {start_status}"
f"{': ' + str(started.get('error')) if started and started.get('error') else ''}"
f"). It may still be starting; try again in a minute.")
return None
# One live attempt per session: a newer connect (say, the phone form
# after the QR) supersedes this watcher, so two never both announce.
# An attempt removes itself when it ends (_retire), so the map holds
# only live attempts.
_attempts = _start_gateway_qr_pair_push.__dict__.setdefault('attempts', {})
attempt = object()
_attempts[sid] = attempt
def _retire() -> None:
if _attempts.get(sid) is attempt:
_attempts.pop(sid, None)
if phone:
# Baileys needs ~3s for the WA noise handshake before
# requestPairingCode succeeds — manual probe earlier today (the
# FA9K4NHK code) verified this timing.
time.sleep(3)
body, _st = _proxy_gateway(
'POST', f'/api/sessions/{sid}/request-pair-code',
json={'phone': phone})
body = body or {}
code = body.get('code') if _st < 400 else None
if not code:
_retire()
_fail(f"{display_name} didn't return a pairing code "
f"({body.get('error') or 'no code'}). Say 'connect "
f"{channel_type}' without a number to scan a QR instead.")
return None
push_payload = {
'kind': 'channel_pair_code',
'channel': channel_type,
'display_name': display_name,
'color': color,
'icon': icon,
'code': code,
'clipboard_payload': code,
'deeplink': deeplink,
'expires_in': 60,
}
# Mobile push via existing NotificationService. Type uses the
# canonical 'consent.*' prefix so the existing NotificationBell
# resolveTargetPath route (consent.* → /admin/consent/{ref}) and
# AutobahnConnectionManager 'consent.*' Intent dispatch (P0-D)
# catch it without per-type bell code. target_id carries the
# channel name so /admin/consent/{whatsapp} lands on the right
# channel admin context. Message JSON keeps the deeplink so the
# generic P0-C deeplink override in resolveTargetPath wins on
# click — that takes the user to the channel pair page directly.
#
# Order matters: NotificationService.create runs BEFORE the
# LiquidUI emit so we can pass notif.id into the chat-card
# payload. PairCodeOverlay (P1-S3) calls notificationsApi.markRead
# with that id on countdown expiry so the bell doesn't carry an
# orphan unread row for an expired pair code.
notif_id = None
try:
from integrations.social.services import NotificationService
from integrations.social.models import db_session
with db_session() as db:
notif = NotificationService.create(
db,
user_id=str(user_id),
type='consent.channel_pair_code',
target_id=channel_type,
target_type='channel',
message=_json.dumps(push_payload),
)
notif_id = getattr(notif, 'id', None)
except Exception as e:
_log.warning("gateway_qr: pair code not pushed to %s's phone "
"(it is still in chat): %s", user_id, e)
# In-chat card via the existing LiquidUI emit pipe.
try:
from core.platform.registry import get_registry
_lui = get_registry().get('LiquidUIService')
if _lui:
_lui.agent_ui_update(
user_id,
{
'type': 'pair_code',
'channel': channel_type,
'channel_type': channel_type,
'display_name': display_name,
'color': color,
'icon': icon,
'code': code,
'expires_in': 60,
'clipboard_payload': code,
'deeplink': deeplink,
'notification_id': notif_id,
'instructions': (
f"Open {display_name} on your phone → "
f"Settings → Linked Devices → Link a Device → "
f"Link with phone number → enter {code} "
f"(60-second window). I've also pushed it "
f"to your phone with auto-copy to clipboard."
),
}, user_id=_owner,
)
except Exception as e:
_log.warning("gateway_qr: pair-code card not delivered: %s", e)
# ── 2026-05-26 P0-E ──────────────────────────────────────────────
# Fleet fanout for iOS native (Nunba-Companion-iOS) which
# subscribes to com.hertzai.hevolve.fleet.user.{user_id} and routes
# the existing 'agent_consent' cmd_type via FleetCommandReceiver
# allowlist — no iOS code change required. Per
# core/peer_link/ui_commands.py module docstring: consent flows
# MUST use cmd_type='agent_consent' (NOT ui_overlay_show); the
# helper is "future work" per that docstring, so we publish via
# message_bus directly with the documented shape. Web/desktop
# already get this card via the agent_ui_update + Notification
# emits above; Android gets it via the consent.* WAMP dispatcher
# (P0-D); this closes the iOS gap with one canonical route.
try:
from core.peer_link.message_bus import get_message_bus
get_message_bus().publish(
'fleet.command.user',
{
'cmd_type': 'agent_consent',
'id': f"consent-{channel_type}-{int(time.time())}",
'title': f"Connect {display_name}",
'body': (
f"Code {code} auto-copied to clipboard. Open "
f"{display_name} → Linked Devices and enter the "
f"code within 60s."
),
'code': code,
'clipboard_payload': code,
'deeplink': deeplink,
'expires_in_seconds': 60,
'channel': channel_type,
'display_name': display_name,
},
user_id=str(user_id),
)
except Exception as e:
_log.warning("gateway_qr: pair code not sent to the iOS "
"companion: %s", e)
def _watch():
"""Show each fresh QR; on pairing, register, go live and confirm.
Window covers 4 pair-code retries (each WA expires after ~60s) and
several QR rotations. Daemon, so shutdown never waits on it.
"""
deadline = time.time() + 240
last_qr = None
shown = 0
while time.time() < deadline:
if _attempts.get(sid) is not attempt:
_log.info("gateway_qr: %s watcher for %s superseded by a "
"newer connect", channel_type, sid)
return
body, _st = _proxy_gateway('GET', f'/api/sessions/{sid}/status')
if body is None or _st >= 400:
_log.debug("gateway_qr: status poll HTTP %s", _st)
body = {}
if body.get('authenticated'):
# Register binding so Hevolve/Nunba see channel as
# connected. Re-uses the SAME register_channel
# closure Connect_Channel calls — single code path.
try:
from integrations.channels.agent_tools import (
build_channel_tool_closures,
)
_tools = build_channel_tool_closures(
{'user_id': str(user_id),
'prompt_id': thread_local_data.get_prompt_id()},
) or []
_reg = next(
(t[2] for t in _tools
if isinstance(t, tuple) and len(t) >= 3
and t[0] == 'register_channel'),
None,
)
if _reg is not None:
reg_out = _reg(channel_type, '{}')
_log.info(
"gateway_qr: register_channel after "
"pair: %s", str(reg_out)[:200])
except Exception as reg_err:
_log.warning(
"gateway_qr: register_channel after pair "
"failed: %s", reg_err)
# Live adapter: register_channel above only writes the
# UserChannelBinding row (so the UI shows "connected")
# — it never wires a real WhatsAppAdapter into the
# running ChannelRegistry, so inbound messages never
# reached the agent. This is the actual transport.
if channel_type == 'whatsapp':
_adapter_out = _ensure_whatsapp_live_adapter(
user_id, sid=sid, base=base)
_log.info(
"gateway_qr: live adapter registration: %s",
_adapter_out)
# Chat success card.
_emit({
'type': 'channel_connected',
'channel': channel_type,
'channel_type': channel_type,
'display_name': display_name,
'color': color,
'icon': icon,
'message': f"✅ {display_name} connected.",
})
_retire()
return
qr = body.get('qr')
if qr and qr != last_qr and not phone:
last_qr = qr
shown += 1
_emit({
'type': 'qr_pair',
'channel': channel_type,
'channel_type': channel_type,
'title': f"Scan to connect {display_name}",
'help': (f"Open {display_name} on your phone → Settings "
f"→ Linked devices → Link a device, and scan "
f"this code."),
'qr': qr,
# One id per code; the overlay replaces the channel's
# card in place with each newer code.
'msg_id': f"qr_pair-{channel_type}-{sid}-{shown}",
# The alternative for someone whose only device is the
# phone: "Link with phone number" posts {phone} here.
'pair_code_action': (
f'/api/social/channels/{channel_type}/connect-pair-code'),
})
time.sleep(2)
if _attempts.get(sid) is not attempt:
return
_retire()
_fail(f"The {display_name} link expired before it was completed. "
f"Say 'connect {channel_type}' to get a new "
f"{'code' if phone else 'QR code'}.")
threading.Thread(
target=_watch, daemon=True,
name=f'connect_channel_poll_{channel_type}',
).start()
return 'pair_code' if phone else 'qr'
_whatsapp_adapter_lock = threading.Lock()
def _fetch_whatsapp_own_identity(gw_base: str, sid: str) -> dict:
"""GET the gateway's own-identity for ``sid``.
Returns ``{'own_jid': ..., 'own_lid': ...}`` (either may be None while
the gateway is still linking), or ``{}`` on any failure, logged. ONE
bounded request; the caller decides when to try again.
"""
import json as _json
import urllib.request as _urlreq
log = logging.getLogger(__name__)
try:
with _urlreq.urlopen(
f"{gw_base.rstrip('/')}/api/sessions/{sid}/status",
timeout=5.0,
) as _resp:
_status = _json.loads(_resp.read().decode('utf-8', errors='replace'))
return {'own_jid': _status.get('own_jid'), 'own_lid': _status.get('own_lid')}
except Exception as _status_err:
log.warning(
"_ensure_whatsapp_live_adapter: could not fetch own identity "
"from gateway (self-chat detection is off until a later poll "
"fetches it): %s", _status_err)
return {}
def _ensure_whatsapp_live_adapter(
user_id, sid: str = None, base: str = None,
) -> dict:
"""Idempotently construct + register a live WhatsAppAdapter for an
already-authenticated gateway_qr session, wiring it into the
process-global ChannelRegistry so inbound WhatsApp messages reach
the agent (via FlaskChannelIntegration._handle_message) and
replies actually get sent back.
register_channel (agent_tools.py) only ever wrote the
UserChannelBinding DB row — it never did this. Without a live
adapter subscribed to the gateway's WebSocket, "connected" in the
DB/UI was cosmetic: nothing was listening for real messages.
Known limitation (unchanged by this fix, not attempted here):
ChannelRegistry keys adapters by channel TYPE, not per user, so
only one live WhatsApp session can be registered process-wide at
a time. Fine for a single-user desktop/local deployment; a real
multi-tenant cloud deployment needs a per-user registry — separate,
larger piece of work.
"""
import asyncio as _aio
import os as _os
sid = sid or (
str(user_id) if str(user_id).startswith('user_')
else f"user_{user_id}"
)
log = logging.getLogger(__name__)
try:
from integrations.channels.flask_integration import (
get_channel_integration,
)
integration = get_channel_integration()
# This function is called on every /whatsapp/qr poll (the mobile
# app's QRScannerScreen polls every ~3s) plus the chat-based
# Connect_Channel path — under concurrent requests, the
# check-then-register below was NOT atomic: two threads could
# both see "not registered yet" before either finished
# registering, each construct + register their OWN adapter with
# its OWN WebSocket connection to the gateway. The gateway then
# broadcasts every inbound message to ALL connected sockets, so
# N racing registrations meant N independent self_chat.handle()
# calls (and N separate /chat + LLM calls) for the SAME message —
# confirmed live 2026-07-28 as repeated near-duplicate replies
# sent to a real WhatsApp self-chat a few seconds apart. The lock
# makes the whole check+register sequence atomic so only ONE
# adapter (and one WebSocket connection) is ever created per
# process, no matter how many requests race here.
gw_base = base or _os.environ.get(
'WHATSAPP_GATEWAY_URL',
f"http://127.0.0.1:"
f"{_os.environ.get('WHATSAPP_GATEWAY_PORT', '3000')}",
)
with _whatsapp_adapter_lock:
existing = integration.registry.get('whatsapp')
if existing is not None:
# A gateway that was still linking when the adapter was
# registered gave no identity. The adapter stays (messages
# must flow), and because every later poll lands here, this
# is the only place the identity can still be adopted: one
# bounded fetch per poll until the adapter carries one.
if not existing.has_owner_identity():
identity = _fetch_whatsapp_own_identity(gw_base, sid)
if identity.get('own_jid') or identity.get('own_lid'):
existing.set_owner_identity(
identity.get('own_jid'), identity.get('own_lid'))
log.info(
"whatsapp adapter adopted its owner identity on a "
"later poll (account_id=%s)", sid)
return {
'success': True,
'message': f'whatsapp adapter already registered '
f'(account_id={getattr(existing, "_account_id", "?")})',
}
from integrations.channels.whatsapp_adapter import (
create_whatsapp_adapter,
)
# Fetch the gateway's own-identity lookup so self-chat detection
# (SelfChatHandler.is_self_message) has something to match against
# — without this, owner_phone/owner_lid are never set and a
# self-chat message can never be recognized, regardless of which
# JID scheme WhatsApp uses for that account (own_jid for
# phone-based accounts, own_lid for LID/privacy-ID accounts).
identity = _fetch_whatsapp_own_identity(gw_base, sid)
adapter = create_whatsapp_adapter(
api_url=gw_base, account_id=sid,
phone_number=identity.get('own_jid'),
owner_lid=identity.get('own_lid'),
)
integration.registry.register(adapter)
# Entry points that skip hartos_bootstrap never call start(), so
# the loop may not exist yet; ensure_running() starts it on demand.
loop, started_integration = integration.ensure_running()
if loop is None:
return {
'success': False,
'error': 'channel event loop failed to start '
'(FlaskChannelIntegration.start() ran but '
'_loop never became live)',
}
# A newly started integration invokes registry.start_all(), which
# starts this adapter exactly once. An already-running loop needs
# this explicit start because registry.start_all() has completed.
if not started_integration:
_aio.run_coroutine_threadsafe(adapter.start(), loop)
log.info(
"whatsapp live adapter registration scheduled "
"(account_id=%s, base=%s)", sid, gw_base,
)
return {
'success': True,
'message': f'whatsapp adapter registration scheduled for {sid}',
}
except Exception as e:
log.warning("_ensure_whatsapp_live_adapter failed: %r", e)
return {'success': False, 'error': repr(e)[:300]}
def _backfill_channel_binding_post_pair(
channel_type: str, user_id: str,
) -> dict:
"""One-shot reconciliation: gateway says authenticated:true but
HARTOS has no UserChannelBinding for this channel/user, OR the
binding exists but no live adapter is wired in (e.g. a pairing
done before _ensure_whatsapp_live_adapter existed, or an
out-of-band pair via manual curl / a prior session where the
polling thread never ran). Calls register_channel('whatsapp', '{}')
to close the binding gap, then wires the live adapter. Idempotent.
"""
try:
from integrations.channels.agent_tools import (
build_channel_tool_closures,
)
tools = build_channel_tool_closures(
{'user_id': str(user_id), 'prompt_id': None},
) or []
reg = next(
(t[2] for t in tools
if isinstance(t, tuple) and len(t) >= 3
and t[0] == 'register_channel'),
None,
)
if reg is None:
return {'success': False, 'error': 'register_channel unavailable'}
out = reg(channel_type, '{}')
adapter_out = None
if channel_type == 'whatsapp':
adapter_out = _ensure_whatsapp_live_adapter(user_id)
return {
'success': True,
'message': str(out)[:400],
'adapter': adapter_out,
}
except Exception as e:
return {'success': False, 'error': repr(e)[:300]}
def _handle_connect_channel_tool(input_text: str) -> str:
"""Register / connect a messaging channel (WhatsApp, Telegram, Discord,
Slack, etc.) from natural-language chat. Delegates to the existing
`register_channel` closure in integrations.channels.agent_tools so there's
exactly ONE registration code path — this is just the LangChain facade.
Input formats accepted:
- "whatsapp" → start setup, returns missing-fields
- "telegram {\"bot_token\":...}" → register with config
- '{"channel": "slack",
"config": {"bot_token":...}}' → full JSON dict
"""
import json as _json
try:
channel_type = None
config_json = '{}'
text = (input_text or '').strip()
if text.startswith('{'):
try:
parsed = _json.loads(text)
channel_type = (parsed.get('channel') or parsed.get('channel_type') or '').lower()
cfg = parsed.get('config') or {}
config_json = _json.dumps(cfg) if isinstance(cfg, dict) else str(cfg)
except _json.JSONDecodeError:
logging.getLogger(__name__).debug("_handle_connect_channel_tool: swallowed _json.JSONDecodeError", exc_info=True)
if not channel_type:
parts = text.split(None, 1)
channel_type = parts[0].lower() if parts else ''
if len(parts) > 1:
config_json = parts[1].strip() or '{}'
if not channel_type:
return (
"Which channel should I connect? Say 'whatsapp', 'telegram', 'slack', "
"'discord', etc. You can also include credentials like "
"'telegram {\"bot_token\": \"123:ABC\"}'."
)
from integrations.channels.agent_tools import build_channel_tool_closures
ctx = {
'user_id': thread_local_data.get_user_id(),
'prompt_id': thread_local_data.get_prompt_id(),
}
tools = build_channel_tool_closures(ctx) or []
register_fn = next(
(t[2] for t in tools
if isinstance(t, tuple) and len(t) >= 3 and t[0] == 'register_channel'),
None,
)
if register_fn is None:
return "Channel registration is not available on this node."
result = register_fn(channel_type, config_json)
# Liquid UI co-pilot: when the tool reports missing-credential
# fields, surface a floating form via agent_ui_update so the
# existing AgentOverlay FormOverlay (case 'form' renderer)
# collects the credentials inline in chat instead of forcing
# the user to navigate to the admin Channels page. The text
# return is preserved unchanged so the LangChain agent can
# summarize it to the user — the form is the IMPERATIVE next
# step, the text is context. Best-effort: never raise, never
# block the tool return. Same emit pattern model_orchestrator
# already uses (see notify_loaded line ~872).
try:
if isinstance(result, str) and 'Missing:' in result:
from integrations.channels.metadata import (
get_channel_metadata, is_oauth_configured,
)
meta = get_channel_metadata(channel_type) or {}
# ── OAuth click-through fork (PR O) ────────────────
# When the operator has set HARTOS_OAUTH_CLIENT_<TYPE>
# env, prefer click-through over paste-form. Calls the
# SAME ``build_authorize_url`` helper /api/oauth/start
# uses (oauth_api.build_authorize_url) — single source
# of URL truth. Emits a Liquid UI ``oauth_link``
# payload that AgentOverlay's case 'oauth_link' renders.
# Falls through to the paste-form on any failure.
if is_oauth_configured(channel_type):
try:
from integrations.channels.oauth_api import (
build_authorize_url,
)
authorize_url, _state = build_authorize_url(
user_id=int(thread_local_data.get_user_id() or 0),
channel_type=channel_type,
)
from core.platform.registry import get_registry
_lui = get_registry().get_or_none('LiquidUIService')
if _lui:
_lui.agent_ui_update(
thread_local_data.get_user_id() or 'system',
{
'type': 'oauth_link',
'channel': channel_type,
'channel_type': channel_type,
'display_name': meta.get('display_name') or channel_type,
'color': meta.get('color') or '#6c63ff',
'icon': meta.get('icon') or 'link',
'url': authorize_url,
# COMPONENT_TYPES['oauth_link'] declares
# authorize_url/provider/title
# (liquid_ui_service.py:690) and the web
# renderer builds its ONLY action from
# `data.authorize_url`
# (AgentOverlay.jsx:736). Only `url` was
# emitted, so actions was [] and the card
# rendered "Sign in to service" with
# nothing to click -- the handshake could
# not be completed on the desktop. `url`
# stays: the RN card reads it
# (AgentInlineChatCard.js:382).
'authorize_url': authorize_url,
'provider': meta.get('display_name') or channel_type,
'title': f"Sign in to {meta.get('display_name') or channel_type}",
'external_url': meta.get('external_url'),
'cta_label': f"Connect with {meta.get('display_name') or channel_type}",
},
user_id=thread_local_data.get_user_id() or None,
)
return (
f"To connect {meta.get('display_name') or channel_type}, "
f"click the button I just sent — it will open the provider's "
f"sign-in page in your browser. After you authorize, the "
f"channel will be connected automatically."
)
except Exception as oauth_err:
logger.debug(
"Connect_Channel: OAuth fork failed, falling back to form: %s",
oauth_err,
)
# Fall through to the paste-form path below.
setup_fields = meta.get('setup_fields') or []
# Skip auto:True fields (WAHA api_url, access_token, etc)
# — these are server-provisioned via env-var defaults in
# register_channel; surfacing them in the chat form would
# ask the user about gateway infrastructure they shouldn't
# need to know about. Operator override stays available
# via the admin Channels page (which shows ALL fields).
visible_fields = [f for f in setup_fields if not f.get('auto')]
if visible_fields:
from core.platform.registry import get_registry
from security.edge_privacy import (
is_secret_key as _is_secret_key,
secret_key_form as _secret_key_form)
_lui = get_registry().get_or_none('LiquidUIService')
if _lui:
_lui.agent_ui_update(
thread_local_data.get_user_id() or 'system',
{
'type': 'form',
'title': f"Connect {meta.get('display_name') or channel_type}",
'channel': channel_type,
'fields': [
{
'name': f.get('key'),
'label': f.get('label') or f.get('key'),
'placeholder': f.get('placeholder') or '',
# The one secret-name rule, with any
# '*_key' masked too: a channel's
# signing or app key is typed into
# this box ('password' was not).
'secret': bool(
f.get('secret')
or _is_secret_key(
_secret_key_form(f.get('key') or ''),
('key',))
),
'help': f.get('help') or '',
'type': f.get('type') or 'text',
'default': f.get('default'),
}
for f in visible_fields
],
# external_url surfaces a "Open <Provider>"
# button at the top of the form so users
# can launch BotFather / Discord dev portal
# / Slack app installer in their browser
# before pasting the resulting token.
'external_url': meta.get('external_url'),
'submit_label': 'Connect',
# The renderer POSTs the field values to
# `action` (Nunba FormOverlay). The old
# submit_action name was read by nothing,
# so Connect dropped the typed token.
'action': f'/api/social/channels/{channel_type}/connect',
},
user_id=thread_local_data.get_user_id() or None,
)
except Exception as e:
logger.debug("Connect_Channel: liquid UI emit skipped: %s", e)
# ── QR-session post-success hookup ────────────────────────
# For channels with auth_method='qr_session' (whatsapp,
# telegram_user, discord_user) the binding alone isn't enough —
# the user also needs to scan a QR with their existing client
# to authorize the session. Reuse the existing
# `set_qr_callback` on each adapter and the existing
# `agent_ui_update` Liquid UI emit pipe (same emitter the
# form uses, just a new payload kind). Single hook for all
# three channels — DRY.
try:
if isinstance(result, str) and 'registered and enabled' in result:
from integrations.channels.metadata import get_channel_metadata
meta = get_channel_metadata(channel_type) or {}
if meta.get('auth_method') == 'qr_session':
_wire_qr_pair_emitter(channel_type, meta)
except Exception as e:
logger.debug("Connect_Channel: qr_pair wire skipped: %s", e)
# ── gateway_qr conversational pair-code push ───────────────
# NEW (2026-05-25): for channels with auth_method='gateway_qr'
# (WhatsApp via the embedded Baileys gateway) the canonical
# onboarding is an 8-char pair-code OTP rather than scanning
# a QR. Rather than dumping the code into the React UI and
# making the user retype, we:
# 1. mint the code at the local gateway,
# 2. emit an agent_ui_update kind='pair_code' so the chat
# panel renders a copy-button card,
# 3. persist a Notification of type='consent.channel_pair_code'
# with the payload as JSON in the message field (the
# consent.* prefix is what every surface's existing route
# catches — bell, Android overlay dispatcher, iOS fleet
# handler). RN
# ConsentOverlayService consumes that, does
# Clipboard.setString(code), and shows an "Open WhatsApp"
# banner whose tap navigates to the deeplink with the
# code already on the clipboard for paste,
# 4. spawn a daemon thread polling the gateway session;
# on authenticated:true emit kind='channel_connected' so
# chat shows the success card automatically.
# The full chain (gateway require()→import() patch,
# baileys 7.0.0-rc13 upgrade, auto-reconnect on code 515
# restart-required) lands the user in state='connected' with
# zero technical detail surfaced to them.
try:
from integrations.channels.metadata import get_channel_metadata
meta = get_channel_metadata(channel_type) or {}
if meta.get('auth_method') == 'gateway_qr':
# A number in the input asks for a pair code. Accepts the
# JSON form ('whatsapp {"phone": "+91..."}') or a bare number
# ('whatsapp +91 90030 54371'); fewer than 8 digits is not a
# phone number. With none, the helper picks by the number on
# file and the device (_start_gateway_qr_pair_push).
try:
_cfg = _json.loads(config_json or '{}')
except ValueError:
_cfg = config_json
if isinstance(_cfg, dict):
_cfg = _cfg.get('phone') or _cfg.get('phone_number') or ''
_phone = ''.join(ch for ch in str(_cfg or '') if ch.isdigit())
if len(_phone) < 8:
_phone = ''
_how = _start_gateway_qr_pair_push(channel_type, meta,
phone=_phone or None)
_name = meta.get('display_name') or channel_type
result = f"{result} " + {
'pair_code': (
f"I sent a linking code to this chat and to your "
f"phone, where it is already copied: in {_name} open "
f"Linked devices > Link a device > Link with phone "
f"number and paste it."),
'ask_phone': (
f"Enter your {_name} number in the form I sent and "
f"I'll send you a linking code: a QR code can't be "
f"scanned by the phone showing it."),
'qr': (
f"Scan the QR code in this chat from {_name} on your "
f"phone (Linked devices > Link a device). If you "
f"can't scan, use 'Link with phone number' on the "
f"card."),
}.get(_how, f"But the {_name} link couldn't start; the "
f"message in chat says why.")
except Exception as e:
logger.warning("Connect_Channel: %s link not started: %s",
channel_type, e)
return result
except Exception as e:
return f"Channel connect error: {str(e)[:200]}"
def _handle_invite_friend_tool(input_text: str) -> str:
"""Generate a shareable Nunba invite link the user can send to a friend.
Mirrors the _handle_connect_channel_tool pattern (G1, plan
elegant-spinning-avalanche). Reuses the canonical referral primitive
``DistributionService.get_or_create_referral_code`` (idempotent —
same user → same code) and the canonical URL builder
``invite_share_url`` so the link format stays a single-writer.
Input formats:
- "" (empty) — generic "share with anyone" link
- "<context, e.g. 'work friend' or 'family'>" — same link, just
echoed back in the friendly reply for the LLM to weave in
Returns text including the share URL so the LangChain agent can
summarize for the user. Also fires a Liquid UI ``form`` card via
``agent_ui_update`` (same emit pattern as Connect_Channel) so the
AgentOverlay renders an inline floating share-card with copy +
"share via channel" actions.
"""
try:
uid = thread_local_data.get_user_id()
if not uid:
return "Cannot generate invite — no signed-in user in this session."
from integrations.social.distribution_service import (
DistributionService, invite_share_url,
)
from integrations.social.models import get_db
db = get_db()
try:
ref = DistributionService.get_or_create_referral_code(db, str(uid))
db.commit()
finally:
db.close()
code = ref.get('code') or ''
if not code:
return "Failed to generate invite code — please try again."
url = invite_share_url(code)
# Liquid UI co-pilot: emit a share card so the user can copy /
# send the link with one click. Same pattern Connect_Channel
# uses (line 2891-2925), same canonical AgentOverlay 'form'
# renderer. Best-effort — never blocks the tool return.
try:
from core.platform.registry import get_registry
_lui = get_registry().get('LiquidUIService')
if _lui:
_lui.agent_ui_update(
str(uid),
{
'type': 'form',
'title': 'Share invite',
'channel': 'invite',
'fields': [
{
'name': 'invite_url',
'label': 'Your invite link',
'value': url,
'readonly': True,
},
],
'submit_label': 'Copy link',
'submit_action': 'copy_invite_url',
}, user_id=str(uid),
)
except Exception as e:
logger.debug("Invite_Friend: liquid UI emit skipped: %s", e)
context_hint = (input_text or '').strip()
suffix = f" (context: {context_hint})" if context_hint else ""
return (
f"Here is your shareable invite link: {url}{suffix}. "
f"Send it via any channel — anyone who joins via this link "
f"is credited as your referral."
)
except Exception as e:
return f"Invite_Friend error: {str(e)[:200]}"
def _handle_join_external_room_tool(input_text: str) -> str:
"""Join an external room (Discord channel, Slack channel, Telegram
super-group, Matrix room, Teams meet, WhatsApp group, etc.) on
behalf of the user, with consent + announcement guardrails.
Mirrors the _handle_connect_channel_tool pattern (G2, plan
elegant-spinning-avalanche). Reuses canonical primitives:
- room_presence_service.gate (consent check, fail-closed)
- room_presence_service.announce_presence (HIVE MISSION
announcement on join)
- room_presence_service.listen_for_objection (auto-leave
on participant 'no AI' reply)
- ChannelRegistry.get / RoomCapableAdapter.join_room (transport)
Input formats:
- "<platform> <room_id_or_url>" (default role=co_pilot)
- "<platform> <room_id> <role>"
- JSON: {"platform":..., "room":..., "role":...}
Roles ∈ {co_pilot, participant, note_taker, silent_observer, writer}.
Returns a human-readable string the LangChain agent can summarize.
Best-effort: never raises out, always returns a string.
"""
import json as _json
try:
uid = thread_local_data.get_user_id()
if not uid:
return "Cannot join external room — no signed-in user in this session."
platform = ''
room_id = ''
role = 'co_pilot'
text = (input_text or '').strip()
if text.startswith('{'):
try:
parsed = _json.loads(text)
platform = (parsed.get('platform') or parsed.get('channel') or '').lower()
room_id = str(parsed.get('room') or parsed.get('room_id') or '')
role = (parsed.get('role') or 'co_pilot').lower()
except _json.JSONDecodeError:
logging.getLogger(__name__).debug("_handle_join_external_room_tool: swallowed _json.JSONDecodeError", exc_info=True)
if not platform or not room_id:
parts = text.split(None, 2)
if len(parts) >= 1:
platform = platform or parts[0].lower()
if len(parts) >= 2:
room_id = room_id or parts[1]
if len(parts) >= 3 and not role:
role = parts[2].lower()
if not platform:
return (
"Which platform should I join? Tell me the platform "
"(discord, slack, telegram, matrix, teams, whatsapp) "
"and the room id or URL."
)
if not room_id:
return f"Which {platform} room should I join? Send the room id or URL."
# 1. Consent gate (canonical path)
from integrations.social.room_presence_service import (
announce_presence, gate, listen_for_objection,
)
allowed, reason = gate(str(uid), platform, room_id, role)
if not allowed:
# Surface a Liquid UI consent card so the user can grant in one click.
try:
from core.platform.registry import get_registry
_lui = get_registry().get_or_none('LiquidUIService')
if _lui:
# A NOTIFICATION, not an approval card: the gate already
# decided, and this only points the owner at where to
# grant. It was typed 'approval', which renders
# AgentOverlay's ApprovalOverlay -- a card that reads
# `description` (absent here, so a blank body), never
# renders `actions` at all, and offers Approve/Deny
# buttons that POST agent_id=undefined, action=undefined
# to /api/agent/approval. 'notification' is the type
# whose contract is exactly this payload's shape:
# title + message + severity + actions
# (liquid_ui_service COMPONENT_TYPES).
#
# And the action key is `kind`, not `action`: the
# renderer dispatches on kind=='navigate' and falls
# through to an "unhandled action" warning otherwise, so
# the button did nothing even once the type was right.
_lui.agent_ui_update(
str(uid),
{
'type': 'notification',
'title': 'Permission needed',
'message': reason,
'severity': 'info',
'actions': [
{'label': 'Open Privacy Settings',
'kind': 'navigate',
'target': '/social/settings/privacy'},
],
},
user_id=str(uid),
)
except Exception as e:
logger.debug("Join_External_Room: consent UI emit skipped: %s", e)
return reason
# 2. Resolve adapter + check room-capable
try:
from integrations.channels.registry import get_registry
from integrations.channels.room_capable import (
UnsupportedRoomError, is_room_capable,
)
registry = get_registry()
adapter = registry.get(platform)
except Exception as e:
return f"Could not look up {platform} adapter: {str(e)[:160]}"
if adapter is None:
return (
f"No {platform} channel is configured. Set it up first "
f"via Connect_Channel ('add channel {platform}') and try again."
)
if not is_room_capable(adapter):
return (
f"The {platform} adapter is configured but does not yet "
f"support room/channel join semantics. Platform support "
f"is pending — for now you can use Connect_Channel for "
f"1:1 messaging."
)
# 3. Join via the adapter (RoomCapableAdapter.join_room)
import asyncio
try:
try:
loop = asyncio.get_event_loop()
if loop.is_running():
fut = asyncio.run_coroutine_threadsafe(
adapter.join_room(room_id, role), loop)
joined = fut.result(timeout=30)
else:
joined = loop.run_until_complete(
adapter.join_room(room_id, role))
except RuntimeError:
joined = asyncio.run(adapter.join_room(room_id, role))
except UnsupportedRoomError as e:
return f"{platform} doesn't support rooms here: {e}"
except Exception as e:
return f"Failed to join {platform}/{room_id}: {str(e)[:160]}"
if not joined:
return (
f"Could not join {platform}/{room_id} — the adapter "
f"refused the join (rate limit, missing permission, or "
f"network). Try again in a moment."
)
# 4. Announce presence (HIVE MISSION — non-optional)
# Resolve the owner's display name from the user record via the same
# regex helper the chat-identity path uses (single source).
# get_user_display_name was a dead threadlocal stub that never existed,
# so owner_name was always None here. Fenced: a lookup miss/failure
# falls back to None, exactly as before.
owner_name = None
try:
_ud, _ = get_action_user_details(user_id=str(uid))
owner_name = extract_owner_name(_ud) or None
except Exception:
owner_name = None
announced = announce_presence(
adapter, room_id, str(uid), role,
owner_display_name=owner_name)
if not announced:
# Required announcement failed — leave to honor the policy.
try:
if loop and loop.is_running():
asyncio.run_coroutine_threadsafe(
adapter.leave_room(room_id), loop).result(timeout=15)
else:
asyncio.run(adapter.leave_room(room_id))
except Exception:
logging.getLogger(__name__).exception("_handle_join_external_room_tool: swallowed Exception")
return (
f"Joined {platform}/{room_id} but could not post the "
f"required AI-presence announcement. Left the room to "
f"honor the privacy policy. Please try again."
)
# 5. Listen for objection
agent_id_for_room = f"{platform}:{room_id}"
def _on_detach(reason: str):
try:
if loop and loop.is_running():
asyncio.run_coroutine_threadsafe(
adapter.leave_room(room_id), loop).result(timeout=15)
else:
asyncio.run(adapter.leave_room(room_id))
except Exception as e:
logger.debug("Join_External_Room: leave_room failed: %s", e)
listen_for_objection(
adapter, room_id, str(uid), agent_id_for_room,
on_detach=_on_detach)
return (
f"Joined {platform}/{room_id} as {role}. I posted an "
f"AI-presence announcement and will leave automatically if "
f"any participant says 'no AI' or '/agent-out'."
)
except Exception as e:
return f"Join_External_Room error: {str(e)[:200]}"
def _handle_agentic_router_tool(input_text):
"""Tool handler: LLM detected a multi-step agentic task.
Called by the LangChain agent when a prompt needs multi-step execution.
Builds a plan using LLM-powered agent matching + plan generation,
sets thread-local flags that the /chat handler checks after get_ans() returns.
"""
try:
from integrations.agentic_router import build_agentic_plan_bounded
# Bounded in agentic_router, on core.subprocess_safe.call_bounded.
# This was `with ThreadPoolExecutor(...) as ex: ex.submit(...)
# .result(timeout=15)`, whose `with` exit joins the stuck worker, so
# the 15 s never released this turn (review F7, 2026-09-27).
plan = build_agentic_plan_bounded(input_text, PROMPTS_DIR)
if plan is None:
return f"I'll help you with: {input_text}. Let me work on this directly."
thread_local_data.set_agentic_routing(
task_description=plan['task_description'],
plan_steps=plan['steps'],
matched_agent_id=plan.get('matched_agent_id'),
)
# Build a human-readable summary for the LLM's response
steps_text = "\n".join(
f" {s['step_num']}. {s['description']}"
for s in plan['steps']
)
agent_note = ""
if plan.get('matched_agent_name'):
agent_note = f"\nMatched agent: {plan['matched_agent_name']} ({plan['matched_agent_source']})"
elif plan.get('requires_new_agent'):
agent_note = "\nNo existing agent matches — a new agent will be created if you approve."
return (
f"I've analyzed your request and prepared a plan:\n\n"
f"Task: {input_text}\n\n"
f"Steps:\n{steps_text}\n"
f"{agent_note}\n\n"
f"Would you like me to proceed with this plan?"
)
except Exception as e:
app.logger.error(f"Agentic router failed: {e}")
return f"I'll help you with: {input_text}. Let me work on this directly."
def _handle_request_resource(input_text: str) -> str:
"""Request_Resource: the agent needs a credential, API key or token.
hartos.ai_key_vault.request_credential_for_agent does the work, for this
tool and for core.agent_tools request_resource alike: an ask no person
could act on goes back to the agent; any other is asked on the owner's
consent card and the agent gets only the {{secret:NAME}} alias.
"""
from hartos.ai_key_vault import request_credential_for_agent
from integrations.social.consent_service import known_agent_id
return request_credential_for_agent(
input_text, agent_id=known_agent_id(thread_local_data.get_prompt_id()))
# Module-level cache for the google-search tool list. Process-wide
# memoization captured 2026-05-08 after observing
# ``get_tools(is_first=True)`` taking 33.4s in casual_conv hot path —
# the slow step is ``load_tools(["google-search"])`` which triggers
# the LangChain GoogleSearchAPIWrapper import chain + env-var probe
# for ``GOOGLE_API_KEY`` / ``GOOGLE_CSE_ID`` on every call. The
# return value is invariant per-process (env doesn't change at
# runtime, the wrapper class doesn't reload), so caching collapses
# the cost from 33s+ to ~zero on the second call onward. Design
# intent ("casual_conv → minimal tool set in ~2.5s") preserved —
# we just stop rebuilding the same answer on every chat turn.
_GOOGLE_SEARCH_TOOLS_CACHE: list | None = None
_GOOGLE_SEARCH_CACHE_LOCK = threading.Lock()
def _safe_load_google_search():
"""Load google-search tool, returning empty list if package is missing.
Memoized per-process — see ``_GOOGLE_SEARCH_TOOLS_CACHE`` above
for the full rationale. Result is invariant under runtime env
changes (env vars are read once at LangChain wrapper init); a
config change requires a process restart anyway.
In bundled/flat mode (NUNBA_BUNDLED=1 or HEVOLVE_NODE_TIER=flat),
the Google API key is expected to be missing — log at DEBUG instead
of WARNING so the log doesn't fill with noise every request. On
cloud/regional tiers, a missing key IS a warning worth surfacing.
"""
global _GOOGLE_SEARCH_TOOLS_CACHE
# Fast path — uncontended read after the first call.
if _GOOGLE_SEARCH_TOOLS_CACHE is not None:
return _GOOGLE_SEARCH_TOOLS_CACHE
with _GOOGLE_SEARCH_CACHE_LOCK:
# Double-check inside the lock so concurrent first-callers
# don't both pay the load_tools cost.
if _GOOGLE_SEARCH_TOOLS_CACHE is not None:
return _GOOGLE_SEARCH_TOOLS_CACHE
try:
result = load_tools(["google-search"])
except (ImportError, Exception) as e:
_bundled = os.environ.get('NUNBA_BUNDLED') or os.environ.get('HEVOLVE_NODE_TIER', 'flat') == 'flat'
if _bundled:
logging.debug(f"Google Search tool unavailable (expected in local mode): {e}")
else:
logging.warning(f"Google Search tool unavailable: {e}")
result = []
_GOOGLE_SEARCH_TOOLS_CACHE = result
return result
def _with_tool_logging(func, tool_name):
"""LangChain tool wrapper — delegates to the canonical core.tool_logging
(#114, single tool-log impl; no more parallel emit/error logic here).
LangChain ``Tool`` objects carry a ``.name`` distinct from ``func.__name__``
(-> name=tool_name for the correct UI label), and the agent_executor wants a
plain-string error rather than the autogen JSON envelope (-> plain_errors).
The per-tool UI stage emit + TOOL_LABELS lookup now live once in
core.tool_logging._emit_tool_call_stage.
"""
from core.tool_logging import log_tool_execution
return log_tool_execution(func, name=tool_name, plain_errors=True)
# Per-(user, prompt) cache for the casual_conv `is_first=True` tool
# list. Captured 2026-05-08 after observing 33.4s on every casual
# chat turn in the bundled env — the heavy steps inside this branch
# (LangChain ``load_tools(['google-search'])`` wrapper init + 30
# hardcoded Tool() objects + service_tool_registry +
# system_introspect + skill_registry + provider_gateway) have NO
# per-call variation once the process is up. Per-user memory tools
# bind to the live MemoryGraph object, so caching the tool function
# is safe: the function queries the graph at CALL time, not at
# build time, so newly added memories are still visible.
#
# Bounded to 32 entries (typical max sessions per Nunba install) +
# FIFO eviction so a long-running instance never grows unbounded.
# Design-intent preserved per
# `tts_engine.py:5722-5729` comment: casual_conv still gets the
# minimal tool set, just memoized so the build cost is paid ONCE
# per (user, prompt) pair instead of every chat turn.
_FIRST_TOOLS_CACHE: dict[str, list] = {}
_FIRST_TOOLS_CACHE_MAX = 32
_FIRST_TOOLS_CACHE_LOCK = threading.Lock()
# ── Process-level caches for slow registry probes ─────────────────────
# Each of these registries does heavy first-call work (Crawl4AI/AceStep
# health probes, langchain google-search wrapper init, system_introspect
# admin-API queries, skill_registry filesystem walk + GitHub probes,
# provider gateway enumeration). Together they account for ~80-100s
# of the 105s `get_tools(is_first=True)` cold-start latency observed
# on the 2026-05-13 09:55:23 IPL turn. Caching the result process-wide
# turns the cold path into a one-time pay; every subsequent (user,
# prompt) build pays only the per-user MemoryGraph open (~50ms).
# Filled lazily on first call (or eagerly by the boot warmer thread
# at startup — see `_warmup_tool_registries_in_background` below).
_SVC_TOOLS_CACHE = None
_SVC_TOOLS_LOCK = threading.Lock()
_INTROSPECT_TOOLS_CACHE = None
_INTROSPECT_TOOLS_LOCK = threading.Lock()
_SKILL_TOOLS_CACHE = None
_SKILL_TOOLS_LOCK = threading.Lock()
_PROVIDER_TOOLS_CACHE = None
_PROVIDER_TOOLS_LOCK = threading.Lock()
def _cached_service_tools():
"""Memoized service_tool_registry.get_langchain_tools()."""
global _SVC_TOOLS_CACHE
if _SVC_TOOLS_CACHE is not None:
return _SVC_TOOLS_CACHE
with _SVC_TOOLS_LOCK:
if _SVC_TOOLS_CACHE is not None:
return _SVC_TOOLS_CACHE
try:
from integrations.service_tools import service_tool_registry
result = list(service_tool_registry.get_langchain_tools())
except (ImportError, Exception):
result = []
_SVC_TOOLS_CACHE = result
return result
def _cached_introspect_tools():
"""Memoized system_introspect_tool.get_langchain_tools()."""
global _INTROSPECT_TOOLS_CACHE
if _INTROSPECT_TOOLS_CACHE is not None:
return _INTROSPECT_TOOLS_CACHE
with _INTROSPECT_TOOLS_LOCK:
if _INTROSPECT_TOOLS_CACHE is not None:
return _INTROSPECT_TOOLS_CACHE
try:
from integrations.service_tools.system_introspect_tool import (
get_langchain_tools as _get_introspect_tools,
)
result = list(_get_introspect_tools())
except (ImportError, Exception):
result = []
_INTROSPECT_TOOLS_CACHE = result
return result
def _cached_skill_tools():
"""Memoized skill_registry.get_langchain_tools()."""
global _SKILL_TOOLS_CACHE
if _SKILL_TOOLS_CACHE is not None:
return _SKILL_TOOLS_CACHE
with _SKILL_TOOLS_LOCK:
if _SKILL_TOOLS_CACHE is not None:
return _SKILL_TOOLS_CACHE
try:
from integrations.skills import skill_registry
result = list(skill_registry.get_langchain_tools())
except (ImportError, Exception):
result = []
_SKILL_TOOLS_CACHE = result
return result
def _cached_provider_tools():
"""Memoized integrations.providers.agent_tools.get_provider_tools()."""
global _PROVIDER_TOOLS_CACHE
if _PROVIDER_TOOLS_CACHE is not None:
return _PROVIDER_TOOLS_CACHE
with _PROVIDER_TOOLS_LOCK:
if _PROVIDER_TOOLS_CACHE is not None:
return _PROVIDER_TOOLS_CACHE
try:
from integrations.providers.agent_tools import get_provider_tools
result = list(get_provider_tools())
except Exception:
result = []
_PROVIDER_TOOLS_CACHE = result
return result
# ── Combined heavy-tools group (non-blocking lazy load) ────────────────
# The 3 suspects responsible for the 103-second silent gap during
# get_tools(is_first=True) on the 2026-05-13 09:55:23 IPL turn:
# 1. _safe_load_google_search() ← cold-import cascade
# (langchain_classic.agents.load_tools transitively pulls in
# google-api-python-client + httplib2 + oauth2client; on Windows
# with Defender scanning DLLs this can be 30-80s.)
# 2. _cached_skill_tools() ← potential GitHub discovery / fs walk
# 3. _cached_provider_tools() ← provider enumeration + connect probes
#
# Grouped here so get_tools NEVER blocks on them — even on the "warm"
# path the heavy cache is consulted with an O(1) read; if not yet
# filled, get_tools proceeds without these tools and a background
# daemon thread continues the build. Subsequent get_tools calls will
# pick them up once the build finishes.
_HEAVY_TOOLS_CACHE: list = []
_HEAVY_TOOLS_LOCK = threading.Lock()
_HEAVY_TOOLS_BUILD_KICKED = False
_HEAVY_TOOLS_READY = False
def _kick_heavy_tools_build_once():
"""Kick the heavy-tools background build if not already running.
Idempotent — safe to call multiple times from any thread."""
global _HEAVY_TOOLS_BUILD_KICKED
with _HEAVY_TOOLS_LOCK:
if _HEAVY_TOOLS_BUILD_KICKED:
return
_HEAVY_TOOLS_BUILD_KICKED = True
def _build():
global _HEAVY_TOOLS_CACHE, _HEAVY_TOOLS_READY
import time as _time
_start = _time.time()
tools = []
try:
tools += list(_safe_load_google_search() or [])
except Exception:
logging.getLogger(__name__).exception("_build: swallowed Exception")
_t_google = _time.time() - _start
try:
tools += list(_cached_skill_tools())
except Exception:
logging.getLogger(__name__).exception("_build: swallowed Exception")
_t_skill = _time.time() - _start - _t_google
try:
tools += list(_cached_provider_tools())
except Exception:
logging.getLogger(__name__).exception("_build: swallowed Exception")
_t_provider = _time.time() - _start - _t_google - _t_skill
with _HEAVY_TOOLS_LOCK:
_HEAVY_TOOLS_CACHE = tools
_HEAVY_TOOLS_READY = True
try:
app.logger.info(
"[HEAVY-TOOLS-LAZY] loaded %d tools "
"(google=%.1fs skill=%.1fs provider=%.1fs total=%.1fs)",
len(tools), _t_google, _t_skill, _t_provider,
_time.time() - _start,
)
except Exception:
logging.getLogger(__name__).exception("_build: swallowed Exception")
try:
import threading as _t
_t.Thread(target=_build, daemon=True,
name='heavy-tools-lazy-build').start()
except Exception:
# Spawn failed — reset the kicked flag so the next call retries.
with _HEAVY_TOOLS_LOCK:
_HEAVY_TOOLS_BUILD_KICKED = False
def _get_heavy_tools_nonblocking():
"""Return the heavy-tools cache contents — non-blocking, never waits.
Reader-only: the build is kicked EAGERLY at boot by
`_warmup_tool_registries_in_background()`, not lazily on first chat.
Chat path returns whatever's already loaded — [] if the background
builder hasn't finished yet, the full list once it has.
Replaces inline _safe_load_google_search / _cached_skill_tools /
_cached_provider_tools so the user never waits for the 80-100s
cold-start cascade.
"""
# Read without lock — list assignment is atomic in CPython, and a
# partially-built cache would still be a list of valid tools.
return _HEAVY_TOOLS_CACHE
def _warmup_tool_registries_in_background():
"""Pre-fill the 4 process-level tool caches in a daemon thread.
Hides the 80-100s cold-start latency that otherwise hits the FIRST
user chat turn. Call once at app boot; subsequent calls are no-ops
because the caches are already filled. Failures are swallowed
silently — the lazy on-call path will retry if a registry was
temporarily unavailable at boot.
"""
try:
import threading as _t
def _warm():
# Brief grace so the rest of bootstrap completes first.
try:
time.sleep(2)
except Exception:
logging.getLogger(__name__).exception("_warm: swallowed Exception")
# Fast registries — these probe at register-time, not
# iteration-time, so iterating them now is cheap. Kept
# synchronous in the warmer thread.
try:
_cached_service_tools()
except Exception:
logging.getLogger(__name__).exception("_warm: swallowed Exception")
try:
_cached_introspect_tools()
except Exception:
logging.getLogger(__name__).exception("_warm: swallowed Exception")
# The 3 heavy ones (google-search cold imports + skill
# registry GitHub/fs walk + provider gateway probes) — do
# NOT load them. Loading them eagerly burns the GIL for
# ~100s during boot (transformers _LazyModule recursion +
# registry probes); loading them lazily blocks the user's
# first chat for ~100s. Either way the net work is the
# same. Default: skip entirely.
#
# google_search is already available via top5_results in
# the inline labeled_tools list (hart_intelligence_entry:
# L4413), so no functionality loss for casual chat.
# skill_registry and provider_gateway are unused in this
# path on the flat/desktop tier.
#
# Opt-in for users who actually need the heavy group:
# $env:HEVOLVE_LOAD_HEAVY_TOOLS = "1"
if os.environ.get('HEVOLVE_LOAD_HEAVY_TOOLS', '').strip().lower() in ('1', 'true', 'yes', 'on'):
try:
_kick_heavy_tools_build_once()
app.logger.info(
"[TOOL-WARMUP] heavy group kicked (opt-in via "
"HEVOLVE_LOAD_HEAVY_TOOLS)")
except Exception:
logging.getLogger(__name__).exception("_warm: swallowed Exception")
try:
app.logger.info(
"[TOOL-WARMUP] fast tool registries ready "
"(svc=%d introspect=%d); heavy group "
"(google+skill+provider) skipped — opt-in via "
"HEVOLVE_LOAD_HEAVY_TOOLS",
len(_SVC_TOOLS_CACHE or []),
len(_INTROSPECT_TOOLS_CACHE or []),
)
except Exception:
logging.getLogger(__name__).exception("_warm: swallowed Exception")
_t.Thread(target=_warm, daemon=True, name='tool-registry-warmup').start()
except Exception as _w_err:
logging.getLogger(__name__).debug(
"tool-registry-warmup spawn skipped: %s", _w_err)
def get_tools(req_tool, is_first: bool = False, agent_tools=None):
"""The tool set for a turn.
``agent_tools`` False withholds the agent-work pair
(core.constants.AGENT_WORK_TOOL_NAMES) on either branch; the metered Hive
API asks for that (task #120). None, the default, reads the turn's own
answer from the request thread-local (thread_local_data.get_agent_tools,
set once by /chat from the body), so every caller of this turn agrees:
get_ans's two calls and CustomGPT._call's prompt rebuild on later steps,
which passes no flag and, while the flag travelled as a keyword,
re-listed both tools in the prompt after the first step (the review of
5a66a3308 measured it). Nothing is added by asking: an explicit False
withholds the pair on any turn, and an explicit True cannot bring it
back on a turn that declined (the review of df2474851 measured that it
did; no caller passes a value today)."""
turn_allows = thread_local_data.get_agent_tools()
agent_tools = turn_allows if agent_tools is None else (bool(agent_tools) and turn_allows)
if is_first:
# ── Timing instrumentation (#185) ─────────────────────────────
# The 2026-05-13 IPL turn showed `get_tools loaded` taking 118s
# despite the heavy-tools group being correctly skipped. Real
# culprit unknown; speculation cost us multiple iterations.
# Capture per-step deltas so the next test gives ground truth.
# Cheap (~12 time.time() calls); always-on for now, demote to
# debug after root cause is fixed.
_gt_t0 = time.time()
_gt_steps = [] # list of (label, elapsed_since_t0_seconds)
def _gt_mark(label):
_gt_steps.append((label, time.time() - _gt_t0))
# Per-(user, prompt) memoization — see _FIRST_TOOLS_CACHE
# docstring above. thread_local_data may be unset in
# degraded/test contexts; fall through to a non-cached
# build in that case.
_cache_key = None
try:
_uid = thread_local_data.get_user_id()
_pid = thread_local_data.get_prompt_id()
# The flag is part of the key: a set built without the agent-work
# pair must never be served to a turn that may use it, or back.
_cache_key = f"{_uid}:{_pid}:{'all' if agent_tools else 'no-agent-work'}"
except Exception:
_cache_key = None
if _cache_key and _cache_key in _FIRST_TOOLS_CACHE:
return _FIRST_TOOLS_CACHE[_cache_key]
_gt_mark('cache_check')
# Heavy tools (google-search + skill_registry + provider_gateway)
# come from the non-blocking heavy-tools cache filled by the boot
# warmer. If not ready yet, returns [] — chat path proceeds without
# them. Once the background build finishes, future first calls
# pick up the full list. Replaces the inline _safe_load_google_search
# call here AND the inline _cached_skill_tools / _cached_provider_tools
# calls further down.
tools = list(_get_heavy_tools_nonblocking())
_gt_mark('heavy_tools_nonblocking')
tool = []
# llm_math is None when its module-level init failed (logged there).
# Same guard shape as `chain` just below: the Calculator is optional,
# but evaluating `llm_math.run` on None kills the WHOLE tool list and
# with it every chat turn (live 2026-08-07: AttributeError on each
# /chat for a full 7h boot).
if llm_math is not None:
tool.append(labeled_tool(
name='Calculator',
func=llm_math.run,
description='Useful for when you need to answer questions about math.',
ui_label='Calculating…',
))
# Only add OpenAPI tool if chain is initialized
if chain is not None:
tool.append(labeled_tool(
name="OpenAPI_Specification",
func=chain.run,
description="Use this feature only when the user's request specifically pertains to one of the following scenarios:\
Image Creation: When a request involves generating an image using text, this feature should be engaged. The entire text prompt must be used as it is unless otherwise requested to enhance further detail of prompt for the image generation process. If additional enahancement is needed , enrich the prompt to image generation with greater detail for learning.\
Student Information: If a request is made for information regarding students, this functionality should be utilized to retrieve the necessary details.\
Query Available Books: When the user is inquiring about available books, this feature should be used to locate and provide information about the required texts.\
Any CRUD operation which is not a READ or anything related to curriculum should not use this tool, It is vital to ensure that the intent precisely falls within one of the above categories before engaging this functionality.\
Don't use this to create a custom curriculum for user",
ui_label='Calling the OpenAPI service…',
))
tool += [
labeled_tool(
name="FULL_HISTORY",
func=parsing_string,
description=f"""Utilize this tool exclusively when the information required predates the current day & pertains to the ongoing user query or when there is a need to recall certain things we spoke earlier. The necessary input for this tool comprises a list of values separated by commas.
The list should encompass a user-generated query, designated by user input text, a commencement date denoted as start_date, and an end date labeled as end_date. The start_date denotes the initiation date for the user information search and should consistently adhere to the ISO 8601 format. Meanwhile, the end_date, also conforming to the ISO 8601 format, signifies the conclusion date for the search.
In cases where the end_date is indeterminable, the current datetime should be employed. For example, if the objective is to retrieve a user's dialogue spanning from the preceding day up to the present day (assuming today's date is 2023-07-13T10:19:56.732291Z), the input would resemble: 'what we discussed about the project, 2023-07-12T10:19:56.00000Z, 2023-07-13T10:19:56.732291Z'. If query has any form of date or time by user, then start end datetime can be exact rather than till today for more accurate results. Remove any references to time based words (e.g. yesterday, today, datetimes, last year) since the date range you provide already accounts for that. e.g. if user has asked what did we discuss the day before yesterday then the text argument should just be empty since it does not have any named entity for fuzzy search followed by start and end datetime.
Strive to apply this tool judiciously for scenarios in which retrospective user information is imperative. If Full history tool response is present, forget other histories, the inputs should be meticulously arranged to facilitate the extraction of accurate and pertinent data within the specified timeframe. Never use this tool for what is the response to my last comment?
Remember whatever user query is regarding search history understand what user is asking about and rephrase it properly then send to tool. Before framing the final tool response from this tool consult corresponding created_at date time to give more accurate response""",
ui_label='Searching your message history…',
),
labeled_tool(
name="Text to image",
func=parse_text_to_image,
description="Based on user query generate visual representation of text. Extract prompt from user query and use it as input for function",
ui_label='Generating an image from text…',
),
labeled_tool(
name="Animate_Character",
func=parse_character_animation,
description='''Use this tool exclusively for animating the selected character or teacher as requested by the user. The user should specify their animation request in a query, such as 'Show me in a spacesuit' or 'Animate yourself as a cartoon standing in front of the Taj Mahal.' This tool handles requests involving animating a pre-selected character and should not be used for general image generation tasks. For example, use it for 'Show me a picture of yourself dancing in the rain' but not for 'Generate an image of a sunset.' input''',
ui_label='Animating your character…',
),
labeled_tool(
name="Image_Inference_Tool",
func=parse_image_to_text,
description='''When a user provides a query containing an image download URL and a related question about that image, utilize this tool for support. Your objective is to extract both the image URL and the user's inquiry or prompt pertaining to that image from their query, and then convert these elements into comma seperated string. The format should be as follows: "image_url, user_query".
''',
ui_label='Analyzing the image…',
),
labeled_tool(
name="Data_Extraction_From_URL",
func=parse_link_for_crwalab,
description='''
Your task is to extract a URL and its type (either 'pdf' or 'website') from a user's query. Upon receiving a query that contains a URL and a specified URL type, you are to use a tool designed for this purpose. The objective is to accurately identify both the URL and its type from the query. Once identified, these elements should be formatted into a comma-separated string, adhering to the format: "url, url_type".
''',
ui_label='Extracting data from URL…',
),
# Direct google-search inline — bypasses langchain_classic.agents.load_tools
# which on first call triggers the transformers _LazyModule re-entry
# recursion (see lines 95-157 docstring). `top5_results` is defined
# in this same file (L6107) — Python resolves the symbol at call
# time, so forward-referencing it here is safe. Without this entry
# the casual_conv langchain agent has no web-search affordance,
# forcing refusals on any "fetch live data" query (2026-05-13 IPL
# turn 301eeed0 forensic).
labeled_tool(
name="google_search",
func=top5_results,
description="Search Google for recent results and retrieve URLs that are suitable for web crawling. Returns the top 5 result URLs + snippets. Use whenever you need live or recent information from the web — sports standings, current news, prices, schedules, anything time-sensitive. Always present source URLs in the response as HTML anchor tags for attribution. Input: the search query as a string.",
ui_label='Searching the web…',
),
labeled_tool(
name="User_details_tool",
func=parse_user_id,
description="If a request is made for information regarding students or users, this functionality should be utilized to retrieve the necessary details. input for this api should Always be current user_id. Except current user id you should say you cannot have access other user's details.",
ui_label='Looking up user details…',
),
labeled_tool(
name="List_Agents",
func=_parse_list_agents,
description=_LIST_AGENTS_TOOL_DESCRIPTION,
ui_label='Listing your agents…',
),
labeled_tool(
name="Visual_Context_Camera",
func=parse_visual_context,
description="To see user or if there is a need to look at user camera feed for vision and understanding scene, visual question answering, seeing user, recognise visual objects and activity then this should be utilised. Input to this tool function should be the user query/input. Only if last 16 seconds Visual Context information is present & is enough, then use that to craft a better creative, better, cohesive, correlated , summarised natural response, format this tool response togather with Previous 15 minutes Visual Context information if you are seeing the scene via videocall from the other end. If there are more than 1 person try to give an identity to each across frames to track the subjects through time by framing the tool input accordingly.",
ui_label='Looking through camera…',
),
labeled_tool(
name="Visual_Context_Watcher",
func=_handle_visual_watcher_tool,
description=(
"Register a visual or audio trigger: continuously watch what the user is doing "
"via camera or listen to what they say, and perform an action when a condition is met. "
"Input format: 'CONDITION: <what to watch for> | ACTION: <what to do> | TTL: <minutes>'. "
"Optional: '| MODALITY: visual|audio|both' (defaults to visual). "
"Example: 'CONDITION: user raises hand | ACTION: say banana | TTL: 30'. "
"Example: 'CONDITION: user mentions their dog | ACTION: remind them about vet appointment | TTL: 60 | MODALITY: audio'. "
"The watcher runs in the background and fires the action whenever the condition "
"is detected. TTL auto-expires the watcher."
),
ui_label='Setting up a visual watcher…',
),
labeled_tool(
name=CREATE_AGENT_TOOL_NAME,
func=_handle_create_agent_tool,
description=_CREATE_AGENT_TOOL_DESCRIPTION,
ui_label='Starting agent creation…',
),
labeled_tool(
name="Request_Resource",
func=_handle_request_resource,
description=REQUEST_RESOURCE_DESCRIPTION,
ui_label='Requesting a resource…',
),
labeled_tool(
name="Suggest_Share_Worthy_Content",
func=_suggest_share_worthy_content,
description=(
"Use this tool when the user asks about content worth sharing, what to share, "
"or when you want to proactively suggest high-engagement posts that deserve "
"wider reach. Finds posts with strong community engagement (many upvotes and "
"comments) but low share count, and suggests them for sharing. "
"Input can be any text — it is not used for filtering."
),
ui_label='Finding share-worthy content…',
),
labeled_tool(
name="Observe_User_Experience",
func=_observe_user_experience,
description=(
"Record a user experience observation. Input: JSON with event, page, "
"duration_ms, outcome. Used for self-improvement and understanding user "
"behavior patterns."
),
ui_label='Recording an observation…',
),
labeled_tool(
name="Self_Critique_And_Enhance",
func=_self_critique_and_enhance,
description=(
"Review past agent suggestions and user behavior observations to improve "
"future recommendations. Input: topic or area to critique. Helps the agent "
"learn from its own interactions."
),
ui_label='Reflecting on past suggestions…',
),
labeled_tool(
name=AGENTIC_ROUTER_TOOL_NAME,
func=_handle_agentic_router_tool,
description=(
"Use when the user's request requires multi-step execution such as building "
"an application, writing code, conducting research, creating a marketing "
"campaign, or any complex task that cannot be answered in a single response. "
"Input: the user's full request describing what they want accomplished. "
"Output: a structured plan with steps. Do NOT use for simple questions, "
"greetings, or tasks that can be answered directly."
),
ui_label='Planning the next steps…',
),
labeled_tool(
name="Shell_Command",
func=_handle_shell_command_tool,
description=(
"Run a single shell / PowerShell / bash command on the user's "
"computer. ALWAYS prefer this over Computer_Action when the task "
"is expressible as a command — it's deterministic, 100ms instead "
"of 60+ seconds of VLM grounding, and doesn't need to see the "
"screen. Use Shell_Command for: launching apps (notepad, calc, "
"chrome, explorer C:\\path), opening files (notepad file.txt), "
"listing directories (dir, ls), system info (systeminfo, "
"Get-Process), creating/deleting files, running scripts, git "
"operations, package installs, etc. Input formats: plain command "
"string ('notepad hello.txt'), 'powershell: Get-Process | Select "
"-First 5' to force PowerShell, 'bash: ls -la ~' to force bash. "
"Destructive patterns (format C:, rm -rf /, fork bomb, etc.) are "
"denied by a built-in safety list. 30-second timeout. Output "
"truncated to 4000 chars. For long-running coding work use "
"Execute_Coding_Task instead."
),
ui_label='Running a command…',
),
labeled_tool(
name="Computer_Action",
func=_handle_computer_action_tool,
description=(
"Execute a GUI action on the user's computer: click buttons, "
"type text, scroll, interact with visible UI elements that can "
"ONLY be reached through the graphical interface. Uses a VLM to "
"identify what to click, so it's ~10-60 seconds per task. "
"IMPORTANT: before picking this tool, ask yourself whether "
"Shell_Command could do the same job — launching apps, opening "
"files, running scripts, changing settings, etc. are almost "
"always faster and more reliable via Shell_Command. Reserve "
"Computer_Action for tasks that genuinely require visual "
"grounding: clicking a specific button inside an already-running "
"app's UI, picking a visible option from a GUI menu, dragging "
"something, etc. Input: natural language instruction (e.g. "
"'click the Save button in the open document', 'select the "
"second item in the dropdown')."
),
ui_label='Acting on your screen…',
),
labeled_tool(
name="Computer_Screenshot",
func=_handle_screenshot_tool,
description=(
"Take a screenshot of the user's screen and describe what you see. "
"Use when you need to understand what's on screen before acting. "
"Input: question about the screen (e.g. 'what apps are open?')."
),
ui_label='Taking a screenshot…',
),
labeled_tool(
name="Request_Camera_Access",
func=_request_capability_consent,
description=(
"Request user permission to access their camera for visual "
"understanding. Use when you need to see the user but camera is "
"not active. Input: reason for needing camera access."
),
ui_label='Requesting camera access…',
),
labeled_tool(
name="Request_Screen_Access",
func=_request_screen_consent,
description=(
"Request user permission to see their screen for computer use. "
"Use when you need to see the screen but screen sharing is not "
"active. Input: reason for needing screen access."
),
ui_label='Requesting screen access…',
),
labeled_tool(
name="Connect_Channel",
func=_handle_connect_channel_tool,
description=(
"Connect / register a messaging channel like WhatsApp, Telegram, "
"Discord, Slack, Email, SMS, Teams, etc. so the user can chat with "
"the agent outside Nunba. Use this tool whenever the user asks to "
"'connect <service>', 'add <service>', 'link my <service>', "
"'set up <service>', 'hook up <service>', or similar — especially "
"for any of the 31 supported channels. Input: the channel name "
"alone (e.g. 'whatsapp') to start setup and get the list of "
"required credentials, OR 'channel_name <json_config>' to register "
# The doubled braces {{...}} are REQUIRED — LangChain's ReAct
# prompt template passes every tool description through
# str.format(), which treats literal { and } as placeholder
# markers. A single `{"bot_token":"..."}` in this string
# raises ValueError: Missing some input keys: {'"bot_token"'}
# at prompt-construction time and crashes /chat before any
# LLM call. Doubling escapes them to literals. See
# langchain.log 2026-04-11 22:46:01 for the crash.
"with credentials (e.g. 'telegram {{\"bot_token\":\"123:ABC\"}}'). "
"For WhatsApp specifically, passing just 'whatsapp' shows a QR "
"code to scan; if the user can't scan (WhatsApp is on the same "
"phone) or asks to link with their number, pass 'whatsapp "
"<their phone number>' to send a pairing code instead. "
"Do NOT ask the user for credentials first — "
"call this tool with just the channel name and it will tell you "
"what's needed."
),
ui_label='Connecting channel…',
),
labeled_tool(
name="Invite_Friend",
func=_handle_invite_friend_tool,
description=(
"Generate a shareable Nunba invite link the user can send to a "
"friend, colleague, or family member so they can join Nunba "
"and (optionally) credit the inviter as their referrer. Use "
"whenever the user says things like 'invite a friend', 'share "
"Nunba with my colleague', 'give me an invite link', 'how do "
"I refer people', 'send my friend an invite', or similar. "
"Input: optional short context describing who the invite is "
"for (e.g. 'work friend', 'family'), or empty string for a "
"generic shareable link. The tool returns a URL the user can "
"paste into any channel; a floating share-card also appears "
"in chat with a one-click Copy button."
),
ui_label='Generating an invite link…',
),
labeled_tool(
name="Join_External_Room",
func=_handle_join_external_room_tool,
description=(
"Join an external room/channel — Discord channel, Slack "
"channel, Telegram super-group, Matrix room, Teams meeting, "
"WhatsApp group, etc. — on behalf of the user as an AI "
"co-pilot, note-taker, participant, silent observer, or "
"writer. Use whenever the user says things like 'join my "
"Discord audio room', 'attend my Teams meet', 'take notes "
"in the WhatsApp family group', 'co-pilot my Slack channel', "
"or similar. Always asks the user for consent first via a "
"Liquid UI prompt, then announces its presence in the room "
"(per HIVE AI MISSION — never silent). Leaves automatically "
"if any participant says 'no AI' / '/agent-out'. Input: "
"either '<platform> <room_id_or_url>' or "
"'<platform> <room> <role>' or a full JSON dict "
"{{\"platform\":..., \"room\":..., \"role\":...}}. Roles: "
"co_pilot (default), participant, note_taker, "
"silent_observer, writer."
),
ui_label='Joining the room…',
),
labeled_tool(
name="Navigate_App",
func=_handle_navigate_app_tool,
description=(
"Open a different page of the Nunba app for the user. Use this "
"whenever the user says things like 'take me to the social feed', "
"'open model management', 'show me the channels page', 'go to "
"admin', 'show the marketplace', 'open games', etc. Input: the "
"destination in natural language (e.g. 'model management' or "
"'social feed'). The tool resolves it against the app's page "
"registry and makes the frontend open that route — you don't "
"need to give a URL, just say the page. After calling this tool, "
"tell the user you're opening that page in a single short sentence."
),
ui_label='Opening the requested page…',
),
labeled_tool(
name="List_Pending_Actions",
func=_handle_list_pending_actions_tool,
description=(
"List the user's upcoming scheduled messages, reminders, and "
"background jobs. Use when the user asks 'what's next', "
"'what do I have scheduled', 'any reminders for today', "
"'what's coming up', 'show my upcoming tasks', etc. "
"Input: optional filter hint ('today', 'this week', "
"'tomorrow') or empty string for all pending. Returns a "
"plain-text list with ISO timestamps — reformat for the "
"user in a natural style. Pulls from both the durable "
"ScheduledMessageManager (reminders, channel messages) "
"and the live apscheduler job store (periodic tasks)."
),
ui_label='Listing pending actions…',
),
]
_gt_mark('inline_labeled_tools')
# Service Tools: Add HTTP microservice tools (Crawl4AI, AceStep, etc.)
# Now memoized at module level — see _cached_service_tools. Eliminates
# the per-(user, prompt) probe cost (Crawl4AI/AceStep health checks)
# observed at ~30s/call on cold first turn.
tool += list(_cached_service_tools())
_gt_mark('cached_service_tools')
# System Introspection Tools: agent self-awareness — GPU tier,
# active models, TTS backend, boot-decision rationale. Lets the
# LLM answer "what model is running?", "why is speculation off?",
# "do I have a GPU?" from real live admin-API data instead of
# hallucinating. Module-level cached.
tool += list(_cached_introspect_tools())
_gt_mark('cached_introspect_tools')
# HART Skills: Ingest agent skills (Claude Code, Markdown, GitHub).
# Moved into the heavy-tools group (loaded eagerly at boot, non-
# blocking on chat path) — see _get_heavy_tools_nonblocking above.
# Already included in the `tools` list at the top of this branch.
# Memory Tools: Add MemoryGraph-backed tools (remember, recall, backtrace)
try:
user_id = thread_local_data.get_user_id()
prompt_id = thread_local_data.get_prompt_id()
_gt_mark('memory_get_ids')
graph = _get_or_create_graph(user_id, prompt_id)
_gt_mark('memory_get_or_create_graph')
if graph:
from integrations.channels.memory.agent_memory_tools import (
create_memory_tools, create_langchain_tools,
)
_gt_mark('memory_tools_import')
session_id = f"{user_id}_{prompt_id}" if prompt_id else str(user_id)
mem_tools_dict = create_memory_tools(graph, str(user_id), session_id)
_gt_mark('memory_create_memory_tools')
lc_mem_tools = create_langchain_tools(mem_tools_dict)
_gt_mark('memory_create_langchain_tools')
tool += lc_mem_tools
except Exception:
_gt_mark('memory_exception')
pass # Non-blocking — memory tools are optional
tools += tool
if not agent_tools:
tools = _without_agent_work_tools(tools)
# Provider gateway tools (Cloud_LLM, Generate_Image, etc.).
# Moved into the heavy-tools group (loaded eagerly at boot, non-
# blocking on chat path) — already included in `tools` at the top
# of this branch via _get_heavy_tools_nonblocking().
# Wrap all tool functions with logging
for t in tools:
if hasattr(t, 'func') and callable(t.func):
t.func = _with_tool_logging(t.func, t.name)
elif hasattr(t, '_run') and callable(t._run):
t._run = _with_tool_logging(t._run, t.name)
_gt_mark('wrap_logging_loop')
# Memoize the wrapped result for this (user, prompt) so the
# next casual_conv chat turn returns in micro-seconds instead
# of re-burning the 30s registry/skills/provider build.
# Bounded FIFO eviction keeps memory flat under heavy
# multi-session usage; the per-user MemoryGraph queried by
# cached memory-tool functions stays live (lookups happen at
# call time, not build time).
if _cache_key:
with _FIRST_TOOLS_CACHE_LOCK:
if (
len(_FIRST_TOOLS_CACHE) >= _FIRST_TOOLS_CACHE_MAX
and _cache_key not in _FIRST_TOOLS_CACHE
):
_FIRST_TOOLS_CACHE.pop(next(iter(_FIRST_TOOLS_CACHE)))
_FIRST_TOOLS_CACHE[_cache_key] = tools
_gt_mark('cache_save')
# Emit timing breakdown — DELTA between consecutive marks so each
# row shows that step's cost, not cumulative. Output keyed by
# `[GET_TOOLS_TIMING]` for easy grep.
try:
_prev = 0.0
_parts = []
for _label, _t in _gt_steps:
_delta = _t - _prev
_parts.append(f"{_label}={_delta:.2f}s")
_prev = _t
app.logger.info(
"[GET_TOOLS_TIMING] total=%.2fs tools=%d steps: %s",
_gt_steps[-1][1] if _gt_steps else 0.0,
len(tools),
" ".join(_parts),
)
except Exception:
logging.getLogger(__name__).exception("get_tools: swallowed Exception")
return tools
else:
tools_dict = {1: 'google_search', 2: 'Calculator', 3: 'OpenAPI_Specification', 4: 'FULL_HISTORY', 5: 'Text to image',
6: 'Image_Inference_Tool', 7: 'Data_Extraction_From_URL', 8: 'User_details_tool', 9: 'Visual_Context_Camera'}
tool_desc = {
'google_search': '''Search Google for recent results and retrieve URLs that are suitable for web crawling. Ensure that the search responses include the source URL from which the data was extracted. Always present this URL in the response as an HTML anchor tag. This approach ensures clear attribution and easy navigation to the original source for each piece of extracted information. Give urls for the source''',
'Calculator': '''Useful for when you need to answer questions about math.''',
'OpenAPI_Specification': '''Use this feature only when the user's request specifically pertains to one of the following scenarios:\
Image Creation: When a request involves generating an image using text, this feature should be engaged. The entire text prompt must be used as it is unless otherwise requested to enhance further detail of prompt for the image generation process. If additional enahancement is needed , enrich the prompt to image generation with greater detail for learning.\
Student Information: If a request is made for information regarding students, this functionality should be utilized to retrieve the necessary details.\
Query Available Books: When the user is inquiring about available books, this feature should be used to locate and provide information about the required texts.\
Any CRUD operation which is not a READ or anything related to curriculum should not use this tool, It is vital to ensure that the intent precisely falls within one of the above categories before engaging this functionality.\
Don't use this to create a custom curriculum for user''',
'FULL_HISTORY': '''Utilize this tool exclusively when the information required predates the current day & pertains to the ongoing user query or when there is a need to recall certain things we spoke earlier. The necessary input for this tool comprises a list of values separated by commas.
The list should encompass a user-generated query, designated by user input text, a commencement date denoted as start_date, and an end date labeled as end_date. The start_date denotes the initiation date for the user information search and should consistently adhere to the ISO 8601 format. Meanwhile, the end_date, also conforming to the ISO 8601 format, signifies the conclusion date for the search.
In cases where the end_date is indeterminable, the current datetime should be employed. For example, if the objective is to retrieve a user's dialogue spanning from the preceding day up to the present day (assuming today's date is 2023-07-13T10:19:56.732291Z), the input would resemble: 'what we discussed about the project, 2023-07-12T10:19:56.00000Z, 2023-07-13T10:19:56.732291Z'. If query has any form of date or time by user, then start end datetime can be exact rather than till today for more accurate results. Remove any references to time based words (e.g. yesterday, today, datetimes, last year) since the date range you provide already accounts for that. e.g. if user has asked what did we discuss the day before yesterday then the text argument should just be empty since it does not have any named entity for fuzzy search followed by start and end datetime.
Strive to apply this tool judiciously for scenarios in which retrospective user information is imperative. If Full history tool response is present, forget other histories, the inputs should be meticulously arranged to facilitate the extraction of accurate and pertinent data within the specified timeframe. Never use this tool for what is the response to my last comment?
Remember whatever user query is regarding search history understand what user is asking about and rephrase it properly then send to tool. Before framing the final tool response from this tool consult corresponding created_at date time to give more accurate response''',
'Text to image': '''Based on user query generate visual representation of text. Extract prompt from user query and use it as input for function''',
'Image_Inference_Tool': '''When a user provides a query containing an image download URL and a related question about that image, utilize this tool for support. Your objective is to extract both the image URL and the user's inquiry or prompt pertaining to that image from their query, and then convert these elements into comma seperated string. The format should be as follows: "image_url, user_query".''',
'Data_Extraction_From_URL': '''Your task is to extract a URL and its type (either 'pdf' or 'website') from a user's query. Upon receiving a query that contains a URL and a specified URL type, you are to use a tool designed for this purpose. The objective is to accurately identify both the URL and its type from the query. Once identified, these elements should be formatted into a comma-separated string, adhering to the format: "url, url_type".''',
'User_details_tool': '''If a request is made for information regarding students or users, this functionality should be utilized to retrieve the necessary details. input for this api should Always be current user_id. Except current user id you should say you cannot have access other user's details.''',
'Visual_Context_Camera': '''This tool captures the user's visual context during a video call, providing real-time captions. Use it for visual question answering, scene understanding, recognizing objects, activities, & monitoring the user. Input will be the user's input/query. If the last 16 seconds of visual context are available and sufficient, it crafts a creative, cohesive response. If not, inform the user of the glitch accessing the current camera feed and guess using the Last_5_Minutes_Visual_Context. Ensure responses are natural, avoiding lists of captions, and format them as if you are seeing the user scene via video call. Analyze the current tool response and previous visual context captions to recognize user activities and infer actions from multiple frames. If the user requests continuous narration without active input, adapt the response to include past, present, and future tenses for dynamic and contextually aware commentary.'''
}
tools_func = {
'google_search': top5_results,
'FULL_HISTORY': parsing_string,
'Text to image': parse_text_to_image,
'Image_Inference_Tool': parse_image_to_text,
'Data_Extraction_From_URL': parse_link_for_crwalab,
'User_details_tool': parse_user_id,
'Visual_Context_Camera': parse_visual_context
}
# Calculator/OpenAPI only when their chains initialized — same guard
# shape; llm_math None = failed boot init (logged at module level).
if llm_math is not None:
tools_func['Calculator'] = llm_math.run
# Only add OpenAPI_Specification to tools_func if chain is initialized
if chain is not None:
tools_func['OpenAPI_Specification'] = chain.run
if req_tool == "google_search":
req_tool = "Google Search Snippets"
# `req_tool in tools_func` guards the tools_func[req_tool] lookup two
# lines down: with llm_math None a Calculator req_tool would KeyError
# here; instead it takes the else-branch (google-search only), which
# is the same degraded-but-alive behaviour as the tool list above.
# When every chain initialized, the maps agree and this is a no-op.
if req_tool is not None and req_tool in tools_dict.values() \
and req_tool in tools_func:
tool_description = tool_desc[req_tool]
tool_func = tools_func[req_tool]
req_tool_from_user = [
Tool(
name=req_tool,
func=tool_func,
description=tool_description
)
]
tools = _safe_load_google_search()
tools += req_tool_from_user
else:
tool_description = ""
tool_func = ""
tools = _safe_load_google_search()
# tools += req_tool_from_user
tool = []
# Same None-guard as the first-tools branch above — see the comment
# there and at the module-level init.
if llm_math is not None:
tool.append(Tool(
name='Calculator',
func=llm_math.run,
description='Useful for when you need to answer questions about math.'
))
# Only add OpenAPI tool if chain is initialized
if chain is not None:
tool.append(Tool(
name="OpenAPI_Specification",
func=chain.run,
description="Use the specialized feature for image generation, student information retrieval, and querying available books, while avoiding its use for non-READ CRUD operations or custom curriculum creation.",
))
tool += [
Tool(
name="FULL_HISTORY",
func=parsing_string,
description=f"""Use the tool for retrieving historical user information within a specified timeframe, including user-generated queries, start and end dates in ISO 8601 format, and carefully rephrase queries related to search history for accurate responses, avoiding use for responses to the last comment."""
),
Tool(
name="Text to image",
func=parse_text_to_image,
description="Based on user query generate visual representation of text. Extract prompt from user query and use it as input for function"
),
Tool(
name="Animate_Character",
func=parse_character_animation,
description='''Use this tool exclusively for animating the selected AI character or teacher as requested by the user; it is not intended for general requests or for animating random images or individuals other than AI teacher avatars. The user should specify their animation request in a query, e.g. 'Show me yourself in a spacesuit' or 'Animate yourself as a person riding a bike.' Once the request is made, the tool will generate the animation and return an URL link to the user that directs them to the animated image. This tool should not be used for general image generation tasks that don't pertain to animating the user's chosen character or teacher. For example, if a user queries 'Show me dancing in the rain,' and they have previously selected a specific character or teacher, the tool should be used to generate this animated scenario. However, if the user's request is something like 'Generate an image of a sunset,' which does not directly involve animating the selected character or teacher, then this tool should not be used.'''
),
Tool(
name="Image_Inference_Tool",
func=parse_image_to_text,
description='''Utilize the tool to extract and format both the image URL and the user's inquiry from a query containing an image download URL into a comma-separated string: "image_url, user_query".'''
),
Tool(
name="Data_Extraction_From_URL",
func=parse_link_for_crwalab,
description='''Utilize the designated tool to extract and format a URL and its type (either 'pdf' or 'website') from a user's query into a comma-separated string: "url, url_type".'''
),
Tool(
name="User_details_tool",
func=parse_user_id,
description="Utilize this functionality to retrieve information about students or users, requiring the current user_id as the only acceptable input; access to other user details is not allowed."
),
Tool(
name="List_Agents",
func=_parse_list_agents,
description=_LIST_AGENTS_TOOL_DESCRIPTION,
),
Tool(
name="Visual_Context_Camera",
func=parse_visual_context,
description="This tool captures the user's visual context during a video call, providing real-time captions. Use it for visual question answering, scene understanding, recognizing objects, activities, & monitoring the user. Input will be the user's input/query. If the last 16 seconds of visual context are available and sufficient, it crafts a creative, cohesive response. If not, inform the user of the glitch accessing the current camera feed and guess using the Last_5_Minutes_Visual_Context. Ensure responses are natural, avoiding lists of captions, and format them as if you are seeing the user scene via video call. Analyze the current tool response and previous visual context captions to recognize user activities and infer actions from multiple frames. If the user requests continuous narration without active input, adapt the response to include past, present, and future tenses for dynamic and contextually aware commentary."
),
Tool(
name=CREATE_AGENT_TOOL_NAME,
func=_handle_create_agent_tool,
description=_CREATE_AGENT_TOOL_DESCRIPTION,
),
Tool(
name="Request_Resource",
func=_handle_request_resource,
description=REQUEST_RESOURCE_DESCRIPTION,
),
Tool(
name="Suggest_Share_Worthy_Content",
func=_suggest_share_worthy_content,
description=(
"Use this tool when the user asks about content worth sharing, what to share, "
"or when you want to proactively suggest high-engagement posts that deserve "
"wider reach. Finds posts with strong community engagement (many upvotes and "
"comments) but low share count, and suggest