Coverage for integrations / agent_engine / speculative_dispatcher.py: 62.1%

1""" 

2Unified Agent Goal Engine - Speculative Dispatcher 

3 

4Draft-first + expert-takeover dispatcher. 

5 

6Two entry points, one delivery channel: 

7 

8 dispatch_draft_first(prompt, ...) 

9 DRAFT tier (Qwen3.5-0.8B) replies SYNCHRONOUSLY in ~300ms with an 

10 envelope ``{reply, delegate: none|local|hive, confidence, ...}``. 

11 The reply is the user's standby answer; ``delegate`` is the draft's 

12 self-assessment of whether a heavier model needs to take the turn. 

13 

14 Five guards can promote the turn to expert background regardless of 

15 the draft's own decision: 

16 * refusal pattern in the reply → REFUSAL_OVERRIDE 

17 * delegate=none + confidence < floor → LOW_CONFIDENCE 

18 * delegate=none + agent_bound prompt → AGENT_BOUND 

19 * delegate=none + classifier surfaces actionable intent 

20 (channel_connect / is_create_agent / language_change / 

21 invite / join_room / memory_query) → ACTIONABLE_INTENT 

22 * envelope parse failure → PARSE_FAILURE 

23 The canonical taxonomy lives in 

24 ``integrations.agent_engine.escalation_reasons.EscalationReason``; 

25 callers / observers / WorldModelBridge see the reason in the 

26 return dict and in ``self._active[speculation_id]``. 

27 

28 dispatch_speculative(prompt, ...) 

29 Legacy entry. Fast tier replies synchronously, expert runs in 

30 background. Same delivery channel. 

31 

32Expert background path 

33---------------------- 

34 

35When a turn escalates, ``_expert_background_task`` dispatches via 

36``_dispatch_expert_langchain``: 

37 

38 - **Local expert** (``is_local=True``): routes through the full HARTOS 

39 /chat pipeline — full tool registry, full system prompt, full agent 

40 context. Bundled mode uses an in-process Flask test_client; HTTP 

41 mode POSTs to the configured backend. Re-entry guarded by 

42 ``speculative=False, draft_first=False`` in the payload. 

43 

44 - **Hive expert** (``is_local=False``): registered by 

45 ``HiveExpertDiscovery`` from ``peer.capability.announce`` gossip, 

46 routed via OpenAI-compatible POST to the peer's ``base_url`` with 

47 bearer auth. The hive peer's 27B / fine-tuned model takes the turn 

48 directly — no "review the draft" wrapper. 

49 

50When the expert returns, ``_deliver_expert_response`` bubble-replaces 

51the standby via the existing ``speculation_id`` channel: SSE for the 

52chat UI + TTS pupit topic for voice. The fast_response stays only when 

53the expert returned empty or got guardrail-blocked. 

54 

55Guardrails at every layer 

56------------------------- 

57- ``ConstitutionalFilter.check_prompt`` before dispatch and on expert output 

58- ``HiveCircuitBreaker.is_halted()`` before dispatch and again at task entry 

59- ``HiveEthos.rewrite_prompt_for_togetherness`` on every prompt 

60- Budget enforcement via ``_check_and_reserve_budget`` 

61- Energy + latency tracking on every model call 

62- Per-peer install-validation gate via ``_pass_validation_gate`` 

63 

64Hive expert wiring (subscriber-side complete; producer pending) 

65--------------------------------------------------------------- 

66``HiveExpertDiscovery`` listens for ``peer.capability.announce`` / 

67``peer.capability.revoke`` on the platform EventBus and auto-registers 

68reachable, trust-verified peers as ``ModelTier.EXPERT`` backends. 

69Until peers start emitting the announce gossip (producer daemon ships 

70separately), ``_pick_expert_for_delegate('hive', ...)`` falls back to 

71the local fast model and ``served_by`` in telemetry reads 

72``local_langchain_bg`` 100% of the time. 

73""" 

74import atexit 

75import json 

76import logging 

77import os 

78import re 

79import time 

80import uuid 

81import threading 

82from collections import deque 

83 

84from core.port_registry import get_port 

85from concurrent.futures import ThreadPoolExecutor 

86from typing import Dict, List, Optional 

87 

88# Eager-import the history loader so the per-call inline import in 

89# dispatch_draft_first doesn't hit Python's import lock on every chat 

90# (sys.modules cache makes subsequent calls fast, but the FIRST call 

91# of a freshly-warm process still pays the lock cost). Imported via 

92# try/except to keep the module load tolerant of test environments 

93# that mock out the social subtree. 

94try: 

95 from integrations.channels.memory.shared_history import ( 

96 seed_autogen_from_shared_history) 

97except ImportError: 

98 seed_autogen_from_shared_history = None # type: ignore[assignment] 

99 

100from integrations.agent_engine.escalation_reasons import EscalationReason 

101 

102logger = logging.getLogger('hevolve_social') 

103 

104 

105# Minimum draft confidence required to commit the draft's reply as the 

106# FINAL answer (delegate="none" path). Below this we schedule an expert 

107# verification in the background regardless of what the draft said — 

108# reasoning quality must never regress just because the 0.8B thought 

109# it could handle the question. A confident "none" still takes the 

110# fast path; an unsure "none" is treated as a quiet "local". 

111_DRAFT_CONFIDENCE_FLOOR = 0.85 

112 

113# Refusal patterns the draft must NEVER emit. See the role-contract 

114# block in `_build_draft_classifier_prompt` — the draft is the 

115# first-responder, NOT the authority on system capability. Any reply 

116# that asserts the system can't do something is a prompt-following 

117# failure (or the model slipped a refusal through the system prompt) 

118# and must be replaced with a standby + escalation to the expert. 

119# 

120# Targets HIGH-CONFIDENCE capability refusals only — not legitimate 

121# negative phrasing like "I don't know the answer". Match shape: 

122# "I" + negation + (capability noun OR system-action verb). Word 

123# boundaries + IGNORECASE guard against false positives like 

124# "I cannot wait to help". 

125# Capability verbs the draft might falsely claim it can't perform. 

126# Factored out so the refusal-pattern alternatives stay in sync. 

127_REFUSAL_VERBS = ( 

128 r"access|fetch|reach|browse|connect|connect to|read|retrieve|" 

129 r"download|verify|view|see|check|crawl|open|load|hit|resolve|" 

130 r"directly|currently|presently" 

131) 

132# Capability nouns paired with "I do(n't| not) have <NOUN>" or 

133# "I lack <NOUN>" / "I have no <NOUN>" — anything that frames an 

134# absent capability rather than negative recall ("I don't know"). 

135_REFUSAL_NOUNS = ( 

136 r"access|tools?|the ability|the capability|permission|a way|any way|" 

137 r"built-?in|external|the internet|web access|internet access|" 

138 r"means|way to" 

139) 

140_REFUSAL_PATTERN = re.compile( 

141 r"\b(?:" 

142 # "I cannot/can't <optional softener> <verb>" 

143 r"I (?:can'?t|cannot)\s+" 

144 r"(?:directly\s+|currently\s+|presently\s+)?" 

145 r"(?:" + _REFUSAL_VERBS + r")" 

146 r"|" 

147 # "I am unable/not able TO <optional softener> <verb>" 

148 # and contraction "I'm unable/not able TO <verb>" 

149 # (regex split because "I'm" has no space between I and m, 

150 # while "I am" requires the space) 

151 r"(?:I'?m|I am)\s+(?:unable|not able)\s+to\s+" 

152 r"(?:directly\s+|currently\s+|presently\s+)?" 

153 r"(?:" + _REFUSAL_VERBS + r")" 

154 r"|" 

155 # "I (don't|do not) have <noun>" e.g. "I don't have built-in tools" 

156 r"I do(?:n'?t| not) have\s+" 

157 r"(?:" + _REFUSAL_NOUNS + r")" 

158 r"|" 

159 # "I lack <noun>" e.g. "I lack access to GitHub" 

160 r"I lack\s+(?:" + _REFUSAL_NOUNS + r")" 

161 r"|" 

162 # "I have no <noun>" e.g. "I have no tools to retrieve…" 

163 r"I have no\s+(?:access|way|tools?|ability|means)" 

164 r"|" 

165 # "I'm just/only a (large language model|LLM|AI|chatbot|…)" 

166 # — split for I'm vs I am as above. 

167 r"(?:I'?m|I am) (?:just|only) (?:a|an) " 

168 r"(?:large language model|LLM|AI|language model|chatbot|" 

169 r"text-based assistant)" 

170 r")", 

171 re.IGNORECASE, 

172) 

173 

174# Generic standby reply substituted when the draft slips a refusal 

175# through. Keeps the user comfortable while the expert path runs. 

176# Intentionally short and capability-neutral — the expert's actual 

177# answer will replace this within the latency budget. 

178_REFUSAL_STANDBY_REPLY = "Let me check that for you…" 

179 

180 

181def _capability_summary_safe() -> str: 

182 """Return the runtime capability summary for the draft prompt, or 

183 an empty string when the helper itself can't be imported. 

184 

185 Lazy import — tool_allowlist pulls model_registry / ModelCatalog / 

186 MCP / channel subsystems on first call. In bare unit-test envs 

187 those aren't loaded, so we treat any failure as 'no summary' and 

188 let the rest of the prompt carry on. Empty string is the signal 

189 the f-string in _build_draft_classifier_prompt skips the section. 

190 """ 

191 try: 

192 from integrations.agent_engine.tool_allowlist import ( 

193 get_capability_summary, 

194 ) 

195 return get_capability_summary() or "" 

196 except Exception: 

197 return "" 

198 

199 

200class SpeculativeDispatcher: 

201 """Fast-first, expert-takeover speculative execution engine. 

202 

203 Every method enforces guardrails — no code path bypasses safety. 

204 """ 

205 

206 def __init__(self, model_registry=None): 

207 from .model_registry import model_registry as _default_registry 

208 self._registry = model_registry or _default_registry 

209 self._expert_pool = ThreadPoolExecutor( 

210 max_workers=int(os.environ.get('HEVOLVE_EXPERT_WORKERS', '4')), 

211 thread_name_prefix='spec_expert', 

212 ) 

213 atexit.register(lambda: self._expert_pool.shutdown(wait=False)) 

214 self._active: Dict[str, dict] = {} # speculation_id → metadata 

215 self._lock = threading.Lock() 

216 self._results: Dict[str, dict] = {} # speculation_id → expert result 

217 self._results_max = 1000 # evict oldest when exceeded 

218 # HiveMind fusion consult results (speculation_id → {result, ts}). 

219 # Populated by the best-effort `_schedule_hive_consult` fired when 

220 # the user selected `intelligence_preference='hive_preferred'` AND 

221 # the draft self-delegated to hive. Read by observers / tests — 

222 # does NOT feed back into the chat reply path (non-blocking; the 

223 # hot-path latency budget is preserved). Capped at 1000 entries 

224 # via the same eviction helper as `_results`. 

225 self._last_hive_consult: Dict[str, dict] = {} 

226 

227 # ─── Gate: should we speculate? ─── 

228 

229 def should_speculate(self, user_id: str, prompt_id: str, 

230 prompt: str, goal: dict = None) -> bool: 

231 """Gate: expert model available + budget remaining + not halted + not casual.""" 

232 # GUARDRAIL: circuit breaker 

233 from security.hive_guardrails import HiveCircuitBreaker 

234 if HiveCircuitBreaker.is_halted(): 

235 return False 

236 

237 # GUARDRAIL: constitutional check on prompt 

238 from security.hive_guardrails import ConstitutionalFilter 

239 passed, _ = ConstitutionalFilter.check_prompt(prompt) 

240 if not passed: 

241 return False 

242 

243 # Need both a fast and expert model 

244 fast = self._registry.get_fast_model() 

245 expert = self._registry.get_expert_model() 

246 if not fast or not expert: 

247 return False 

248 if fast.model_id == expert.model_id: 

249 return False # Same model — no point speculating 

250 

251 # Budget check (if goal has spark budget) 

252 if goal and goal.get('spark_budget', 0) > 0: 

253 spent = goal.get('spark_spent', 0) 

254 remaining = goal['spark_budget'] - spent 

255 # Estimate ~4k tokens per speculation (prompt + response) 

256 if remaining < expert.cost_per_1k_tokens * 4: 

257 return False 

258 

259 return True 

260 

261 # ─── Main entry point ─── 

262 

263 def dispatch_speculative(self, prompt: str, user_id: str, prompt_id: str, 

264 goal_id: str = None, goal_type: str = 'general', 

265 node_id: str = None) -> dict: 

266 """ 

267 1. Guardrail-check the prompt 

268 2. Pick fast model → dispatch synchronously → user gets response 

269 3. Record compute contribution for hive node (ad revenue) 

270 4. Pick expert model → dispatch in background thread 

271 5. Return fast response immediately 

272 

273 Returns: 

274 { 

275 'response': str, # Fast agent's response 

276 'speculation_id': str, # Track the background expert 

277 'fast_model': str, # Which model served fast 

278 'expert_pending': bool, # True if expert is working 

279 'latency_ms': float, # Fast response latency 

280 'energy_kwh': float, # Energy consumed 

281 } 

282 """ 

283 speculation_id = str(uuid.uuid4())[:12] 

284 

285 # GUARDRAIL: circuit breaker 

286 from security.hive_guardrails import HiveCircuitBreaker 

287 if HiveCircuitBreaker.is_halted(): 

288 return {'response': '', 'speculation_id': speculation_id, 

289 'error': 'Hive is halted', 'expert_pending': False} 

290 

291 # GUARDRAIL: constitutional filter 

292 from security.hive_guardrails import ConstitutionalFilter 

293 passed, reason = ConstitutionalFilter.check_prompt(prompt) 

294 if not passed: 

295 return {'response': '', 'speculation_id': speculation_id, 

296 'error': reason, 'expert_pending': False} 

297 

298 # GUARDRAIL: rewrite prompt for togetherness 

299 from security.hive_guardrails import HiveEthos 

300 prompt = HiveEthos.rewrite_prompt_for_togetherness(prompt) 

301 

302 # ── FAST PATH ── 

303 fast_model = self._registry.get_fast_model() 

304 if not fast_model: 

305 return {'response': '', 'speculation_id': speculation_id, 

306 'error': 'No fast model available', 'expert_pending': False} 

307 

308 start = time.time() 

309 fast_response = self._dispatch_to_model( 

310 fast_model, prompt, user_id, prompt_id, goal_type, goal_id) 

311 elapsed_ms = (time.time() - start) * 1000 

312 self._track_call_telemetry(fast_model, elapsed_ms, node_id) 

313 

314 # ── EXPERT PATH (background) ── 

315 expert_model = self._registry.get_expert_model() 

316 expert_pending = self._schedule_expert_background( 

317 speculation_id=speculation_id, 

318 prompt=prompt, 

319 fast_response=fast_response, 

320 expert_model=expert_model, 

321 user_id=user_id, prompt_id=prompt_id, 

322 goal_id=goal_id, goal_type=goal_type, 

323 origin_model_id=fast_model.model_id, 

324 origin_model_role='fast_model', 

325 ) 

326 

327 return { 

328 'response': fast_response, 

329 'speculation_id': speculation_id, 

330 'fast_model': fast_model.model_id, 

331 'expert_pending': expert_pending, 

332 'latency_ms': round(elapsed_ms, 1), 

333 'energy_kwh': round( 

334 self._registry.get_total_energy_kwh(hours=0.01), 6), 

335 } 

336 

337 # ─── Draft-first dispatch (Qwen3.5-0.8B standby + delegate signal) ─── 

338 

339 def dispatch_draft_first(self, prompt: str, user_id: str, prompt_id: str, 

340 goal_id: str = None, goal_type: str = 'general', 

341 node_id: str = None, 

342 agent_persona: Optional[str] = None, 

343 preferred_lang: str = 'en', 

344 user_pref: str = 'auto', 

345 agent_bound: bool = False) -> dict: 

346 """Draft-first dispatch: tiny model answers immediately, signals whether 

347 to delegate. 

348 

349 Architecture (the piece the user asked for on top of the speculative 

350 dispatcher): 

351 

352 1. The DRAFT tier model (Qwen3.5-0.8B) receives a wrapped prompt that 

353 asks it to emit JSON: 

354 { "reply": "...", 

355 "delegate": "none" | "local" | "hive", 

356 "confidence": 0.0-1.0 } 

357 `delegate` is the draft's self-assessment of its place in the 

358 hierarchy: can it answer, or should a bigger model take over? 

359 2. Regardless of the delegate signal, the draft's ``reply`` is 

360 returned SYNCHRONOUSLY as a standby response — the user sees 

361 something within ~300ms even when delegation is needed. 

362 3. When ``delegate != "none"`` (or the JSON can't be parsed), a 

363 background expert task runs on the local FAST tier or is 

364 dispatched to the hive — same code path as dispatch_speculative. 

365 4. Both the draft's reply AND the eventual expert reply get fed 

366 through ``WorldModelBridge.record_interaction`` with distinct 

367 model_id tags so HevolveAI's continual learner can distill 

368 the expert's improvements back into the draft over time. 

369 

370 Guardrails: the outer /chat handler already ran GuardrailEnforcer + 

371 prompt_guard before calling us, so we only re-check constitutional 

372 filter (cheap) and circuit breaker here. Budget + hive_ethos still 

373 apply on the expert path via dispatch_speculative's helpers. 

374 

375 Returns a dict shaped like dispatch_speculative's response plus 

376 ``delegate``, ``draft_model``, and ``draft_confidence`` fields so 

377 callers can discriminate. 

378 """ 

379 speculation_id = str(uuid.uuid4())[:12] 

380 

381 # ── 1. Local preconditions (cheapest first, so a missing registry 

382 # entry never triggers an unnecessary network probe). 

383 draft_model = self._registry.get_draft_model() 

384 if draft_model is None: 

385 return { 

386 'response': '', 'speculation_id': speculation_id, 

387 'delegate': 'none', 'error': 'no_draft_model', 

388 'expert_pending': False, 

389 } 

390 

391 # ── Language guard: skip 0.8B draft for non-Latin scripts ── 

392 # Qwen3.5-0.8B-UD-Q4_K_XL has weak Unicode coverage for the 

393 # scripts listed in core.constants.NON_LATIN_SCRIPT_LANGS — 

394 # falls back to Latin-transliterated output ("Vanakkam! Nan 

395 # ungal nanban..." for Tamil) which is unusable for TTS + UX. 

396 # The 4B main has native Unicode coverage. Canonical set lives 

397 # in core.constants; this file imports rather than duplicates. 

398 from core.constants import NON_LATIN_SCRIPT_LANGS 

399 _lang_key = (preferred_lang or 'en').split('-')[0].lower() 

400 if _lang_key and _lang_key != 'en' and _lang_key in NON_LATIN_SCRIPT_LANGS: 

401 logger.info( 

402 f"Skipping 0.8B draft for preferred_lang={_lang_key!r} " 

403 f"(weak Unicode script coverage); routing direct to 4B.", 

404 ) 

405 return { 

406 'response': '', 'speculation_id': speculation_id, 

407 'delegate': 'none', 'error': 'draft_skipped_non_english', 

408 'expert_pending': False, 

409 } 

410 

411 # ── 2. Gate checks (constitutional + circuit breaker + draft server probe) 

412 gate_error = self._check_draft_first_gates(prompt) 

413 if gate_error is not None: 

414 return { 

415 'response': '', 'speculation_id': speculation_id, 

416 'delegate': 'none', 'error': gate_error, 'expert_pending': False, 

417 } 

418 

419 # ── 2. Load recent conversation history (best-effort, non-fatal) ── 

420 # Single source of truth — same seed_autogen_from_shared_history the 

421 # autogen GroupChat uses. Without this the draft sees each turn as 

422 # first-contact and emits generic greetings for follow-ups (witnessed 

423 # 2026-05-09 09:35 — user asked about WhatsApp at 09:34 then "what's 

424 # happening?" at 09:35; draft replied with a generic "Nothing 

425 # unusual…" because it had no memory of the WhatsApp turn 60s prior). 

426 # Cap at 4 messages — the 0.8B context budget can't fit more without 

427 # crowding out the answering rules + JSON schema. 

428 recent_turns: List[Dict] = [] 

429 try: 

430 if seed_autogen_from_shared_history is None: 

431 raise ImportError( 

432 "shared_history.seed_autogen_from_shared_history " 

433 "not importable in this environment") 

434 recent_turns = seed_autogen_from_shared_history( 

435 user_id, max_messages=4) or [] 

436 # NO dedup against the current prompt — users genuinely 

437 # repeat themselves (rephrasing, asking again, "hi" / "hi" 

438 # in casual back-and-forth). Dropping a recent turn just 

439 # because its text matches the current prompt would silently 

440 # lose legitimate conversation history. If the writer hook 

441 # in get_response_group / chatbot_routes persisted the 

442 # current turn before this dispatch ran, the prompt may 

443 # appear twice in the LLM's input — that's the correct 

444 # natural-language signal ("user said X, then said X again") 

445 # and the model should treat it as such. 

446 except Exception as _hist_err: 

447 logger.debug(f"draft history load failed: {_hist_err}") 

448 recent_turns = [] 

449 

450 # ── 3. Dispatch the draft with the classifier prompt ── 

451 draft_prompt = self._build_draft_classifier_prompt( 

452 prompt, agent_persona=agent_persona, preferred_lang=preferred_lang, 

453 recent_turns=recent_turns) 

454 start = time.time() 

455 draft_raw = self._dispatch_to_model( 

456 draft_model, draft_prompt, user_id, prompt_id, goal_type, goal_id) 

457 draft_latency_ms = (time.time() - start) * 1000 

458 self._track_call_telemetry(draft_model, draft_latency_ms, node_id) 

459 

460 # ── 3. Parse envelope + record draft interaction ── 

461 parsed = self._parse_draft_envelope(draft_raw) 

462 draft_reply = parsed.get('reply') or draft_raw.strip()[:500] 

463 # Track whether the envelope parsed at all — an empty parsed dict 

464 # means we fell through to the delegate='local' default below, 

465 # which is materially different from the model emitting an 

466 # explicit 'local' decision. Stamp the reason for downstream 

467 # telemetry / continual-learning before any guard runs. 

468 parse_failed = not parsed 

469 delegate = parsed.get('delegate', 'local') # default on parse fail 

470 confidence = float(parsed.get('confidence') or 0.0) 

471 # Canonical reason value — refined by each guard below. Starts 

472 # at PARSE_FAILURE when the envelope didn't parse, otherwise the 

473 # baseline CLASSIFIER_DELEGATE. Only meaningful when we end up 

474 # in the delegate in ('local', 'hive') branch — None otherwise. 

475 escalation_reason: Optional[EscalationReason] = ( 

476 EscalationReason.PARSE_FAILURE if parse_failed 

477 else EscalationReason.CLASSIFIER_DELEGATE 

478 ) 

479 

480 # REFUSAL GUARD: the draft is the first-responder role, NOT the 

481 # authority on system capability. Any reply that asserts the 

482 # system can't do something is a prompt-following failure — the 

483 # role contract in the classifier prompt explicitly forbids 

484 # refusals of this shape. When the model slips one through 

485 # anyway (typically when the user asks for a tool-bound 

486 # capability the draft can't see — URL fetch, file read, 

487 # GitHub PR check, etc.), we replace the standby with a generic 

488 # holding reply and force escalation to the local expert. The 

489 # user never sees the refusal; the expert (with full tool 

490 # access) produces the real answer and the SSE/WAMP fan-out 

491 # delivers it to replace the standby. 

492 # Size-agnostic — same rule applies whether the draft slot 

493 # holds a 0.8B, 4B, or 27B model. None of them see the full 

494 # tool registry the expert binds. 

495 refusal_overridden = False 

496 if draft_reply and _REFUSAL_PATTERN.search(draft_reply): 

497 logger.info( 

498 "draft-first: refusal detected in draft reply " 

499 "(delegate=%r, conf=%.2f) — replacing with standby + " 

500 "forcing delegate=local. Original reply prefix: %r", 

501 delegate, confidence, draft_reply[:120], 

502 ) 

503 draft_reply = _REFUSAL_STANDBY_REPLY 

504 delegate = 'local' 

505 refusal_overridden = True 

506 escalation_reason = EscalationReason.REFUSAL_OVERRIDE 

507 

508 # REASONING-QUALITY GUARD: an unsure "none" is not good enough to 

509 # ship as the final answer. Promote it to "local" so an expert 

510 # verifier still runs in the background. Keeps the single 

511 # dispatch path — this is just delegate normalization, no new 

512 # branch below. Ensures the draft model can never regress the 

513 # reasoning quality the user gets — worst case they see the 

514 # draft reply briefly as a standby and it's replaced when the 

515 # 4B expert finishes via the existing crossbar delivery. 

516 if delegate == 'none' and confidence < _DRAFT_CONFIDENCE_FLOOR: 

517 logger.info( 

518 f"draft-first: low-confidence 'none' ({confidence:.2f} < " 

519 f"{_DRAFT_CONFIDENCE_FLOOR}) → escalating to local verifier" 

520 ) 

521 delegate = 'local' 

522 escalation_reason = EscalationReason.LOW_CONFIDENCE 

523 

524 # AGENT-BINDING GUARD: when the caller bound this turn to a 

525 # specific agent (prompt_id resolves to a real agent on disk, 

526 # not the request-id fallback), the user has chosen a 

527 # specialist and expects THAT specialist's voice — not the 

528 # 0.8B draft answering in its generic voice. Even a trivial 

529 # greeting like "hi" should pass through the specialist so 

530 # its persona / system prompt / tool registry shapes the 

531 # reply. Promote delegate=none → local so the expert path 

532 # always takes the turn for agent-bound requests. 

533 # 

534 # When agent_bound=False (no specific agent in scope, e.g. 

535 # default chat or guest free-floating), the draft's "none" 

536 # decision stays — the 0.8B can handle trivial questions 

537 # without paying the 4B cost. 

538 if delegate == 'none' and agent_bound: 

539 logger.info( 

540 "draft-first: prompt_id=%r is bound to a specific " 

541 "agent — escalating delegate=none → 'local' so the " 

542 "agent's expert path takes the turn instead of the " 

543 "0.8B draft's generic voice.", 

544 prompt_id, 

545 ) 

546 delegate = 'local' 

547 escalation_reason = EscalationReason.AGENT_BOUND 

548 

549 # ACTIONABLE-INTENT GUARD: when the draft's own classifier 

550 # surfaces an actionable intent flag (channel_connect, 

551 # is_create_agent, language_change), answering in-band would 

552 # orphan the action — there is no way to invoke the matching 

553 # tool (Connect_Channel / Create_Agent) from the casual draft 

554 # path because casual_conv=True skips the full tool registry 

555 # in hart_intelligence_entry.get_ans (see the is_first=True 

556 # branch). Promote delegate=none → 'local' so the expert 

557 # turn binds the full registry — Connect_Channel for 

558 # channel-add intents, Create_Agent for agent-build intents — 

559 # and actually fires the tool the LLM would otherwise have 

560 # described in free-form text. 

561 # 

562 # The draft's reply is also replaced with the standby (same 

563 # rationale as REFUSAL GUARD above): the draft on the 

564 # casual path has no tool access, so any in-band reply on 

565 # an actionable-intent turn is a verified-signal anti-pattern 

566 # — it claims action while taking none. The expert reply 

567 # replaces the standby via the existing speculation_id 

568 # bubble-replacement path (#204 already shipped). 

569 # Helper: treat the sentinel string 'none' as "no intent", matching 

570 # the classifier's own contract (it returns 'none' for *_intent / 

571 # memory_query when there's no actionable intent). Without this, 

572 # the literal 'none'.strip() evaluates truthy and the guard fires 

573 # on every casual turn → every reply gets replaced with the 

574 # "Let me check that for you…" placeholder, and the user never 

575 # receives the actual answer because the expert background task 

576 # has nothing real to refine (live evidence 2026-05-11 22:36:41 

577 # request 897fc534 — speculation_id 9a418ac4-1eb, message 'strange'). 

578 def _intent_set(value) -> bool: 

579 if not value or value is False: 

580 return False 

581 v = str(value).strip().lower() 

582 return bool(v) and v != 'none' 

583 

584 if delegate == 'none' and ( 

585 _intent_set(parsed.get('channel_connect')) 

586 or parsed.get('is_create_agent') 

587 or _intent_set(parsed.get('language_change')) 

588 or _intent_set(parsed.get('invite_intent')) 

589 or _intent_set(parsed.get('join_room_intent')) 

590 or _intent_set(parsed.get('memory_query')) 

591 ): 

592 logger.info( 

593 "draft-first: actionable intent flag set " 

594 "(channel_connect=%r, is_create_agent=%r, " 

595 "language_change=%r, invite_intent=%r, " 

596 "join_room_intent=%r, memory_query=%r) — escalating " 

597 "delegate=none → 'local' so the expert tool registry " 

598 "handles the turn.", 

599 parsed.get('channel_connect'), 

600 parsed.get('is_create_agent'), 

601 parsed.get('language_change'), 

602 parsed.get('invite_intent'), 

603 parsed.get('join_room_intent'), 

604 parsed.get('memory_query'), 

605 ) 

606 draft_reply = _REFUSAL_STANDBY_REPLY 

607 delegate = 'local' 

608 escalation_reason = EscalationReason.ACTIONABLE_INTENT 

609 

610 # Non-Latin languages skip draft entirely (hart_intelligence_entry.py) 

611 # so this code path is only reached for English/Latin-script languages. 

612 

613 # ── Draft telemetry: log full envelope for offline calibration ── 

614 # The data scientist requires this to build a confidence calibration 

615 # curve and detect intent classification drift over time. 

616 try: 

617 _telemetry = { 

618 'speculation_id': speculation_id, 

619 'user_id': user_id, 

620 'confidence': confidence, 

621 'delegate': delegate, 

622 'is_casual': parsed.get('is_casual'), 

623 'is_correction': parsed.get('is_correction'), 

624 'is_create_agent': parsed.get('is_create_agent'), 

625 'channel_connect': parsed.get('channel_connect'), 

626 'language_change': parsed.get('language_change'), 

627 'draft_model': draft_model.model_id if draft_model else None, 

628 'latency_ms': draft_latency_ms, 

629 'reply_len': len(draft_reply) if draft_reply else 0, 

630 'escalated': delegate != parsed.get('delegate', 'local'), 

631 # refusal_overridden lets us calibrate per-model adherence 

632 # to the role contract. A draft model with the right 

633 # prompt should hit this near-zero — sustained non-zero 

634 # rate is a signal that either the prompt isn't being 

635 # followed (model too small / fine-tune mismatch) or the 

636 # model is the wrong fit for the draft slot on this 

637 # hardware. 

638 'refusal_overridden': refusal_overridden, 

639 } 

640 logger.info(f"draft-telemetry: {json.dumps(_telemetry)}") 

641 except Exception: 

642 pass # telemetry must never break the hot path 

643 

644 # When the draft path's net effect is "draft answered, no 

645 # escalation" (delegate=='none' and none of the guards fired), 

646 # the escalation_reason is meaningless — clear it so the 

647 # WorldModelBridge sees a clean "this draft was the final 

648 # answer" record. 

649 recorded_reason = ( 

650 escalation_reason.value if delegate in ('local', 'hive') 

651 else None 

652 ) 

653 self._record_interaction_safely( 

654 user_id=user_id, prompt_id=prompt_id, prompt=prompt, 

655 response=draft_reply, model_id=draft_model.model_id, 

656 latency_ms=draft_latency_ms, node_id=node_id, goal_id=goal_id, 

657 escalation_reason=recorded_reason, 

658 ) 

659 

660 # ── 4. Schedule expert if the draft self-delegated ── 

661 expert_pending = False 

662 hive_consult_scheduled = False 

663 if delegate in ('local', 'hive'): 

664 expert_model = self._pick_expert_for_delegate( 

665 delegate, user_pref=user_pref) 

666 expert_pending = self._schedule_expert_background( 

667 speculation_id=speculation_id, 

668 prompt=prompt, 

669 fast_response=draft_reply, 

670 expert_model=expert_model, 

671 user_id=user_id, prompt_id=prompt_id, 

672 goal_id=goal_id, goal_type=goal_type, 

673 origin_model_id=draft_model.model_id, 

674 origin_model_role='draft_model', 

675 delegate=delegate, 

676 escalation_reason=escalation_reason, 

677 ) 

678 # When the user explicitly asked for `hive_preferred` AND the 

679 # draft self-delegated to hive, also fire a best-effort MoE 

680 # HiveMind fusion consult in the background. The consult 

681 # result is stored on self._last_hive_consult for observers 

682 # and future wiring; it does NOT replace the expert's reply 

683 # on the hot path (preserves the 1.5s chat budget). Safe on 

684 # `auto` and `local_only` paths — gated by user_pref so 

685 # existing callers see no behavior change. 

686 if delegate == 'hive' and user_pref == 'hive_preferred': 

687 hive_consult_scheduled = self._schedule_hive_consult( 

688 prompt=prompt, 

689 user_id=user_id, 

690 speculation_id=speculation_id, 

691 ) 

692 

693 # Channel name defensively coerced: draft model sometimes emits None, 

694 # null, or a capitalised string. Normalise to a lowercased str so 

695 # callers can treat an empty string as "no channel connect intent". 

696 _channel = parsed.get('channel_connect') or '' 

697 if not isinstance(_channel, str): 

698 _channel = '' 

699 # Language change — same defensive coercion as channel_connect. 

700 # Validated against the canonical SUPPORTED_LANG_DICT (single 

701 # source of truth for language codes, lives in hart_intelligence_entry). 

702 _lang = parsed.get('language_change') or '' 

703 if not isinstance(_lang, str): 

704 _lang = '' 

705 _lang = _lang.strip().lower()[:5] 

706 if _lang: 

707 from core.safe_hartos_attr import safe_hartos_attr 

708 SUPPORTED_LANG_DICT = safe_hartos_attr('SUPPORTED_LANG_DICT') 

709 if SUPPORTED_LANG_DICT is not None: 

710 if _lang not in SUPPORTED_LANG_DICT: 

711 logger.debug( 

712 "draft: language_change '%s' not in " 

713 "SUPPORTED_LANG_DICT — ignoring", _lang) 

714 _lang = '' 

715 # else: HARTOS not yet loaded — accept the code as-is, same 

716 # fall-through the original ImportError branch had. 

717 return { 

718 'response': draft_reply, 

719 'speculation_id': speculation_id, 

720 'draft_model': draft_model.model_id, 

721 'delegate': delegate, 

722 'draft_confidence': confidence, 

723 'is_correction': bool(parsed.get('is_correction', False)), 

724 'is_casual': bool(parsed.get('is_casual', False)), 

725 'is_create_agent': bool(parsed.get('is_create_agent', False)), 

726 'channel_connect': _channel.strip().lower(), 

727 'language_change': _lang.strip().lower(), 

728 'expert_pending': expert_pending, 

729 # Additive field: observers (J282, telemetry, admin/diag) can 

730 # tell a hive fusion consult was fired in background. Legacy 

731 # callers that don't read this field are unaffected. 

732 'hive_consult_scheduled': hive_consult_scheduled, 

733 # Additive field: which guard promoted this turn to expert 

734 # (or ``None`` when the draft answered as final). Canonical 

735 # values come from ``EscalationReason`` (see 

736 # ``integrations.agent_engine.escalation_reasons``). 

737 'escalation_reason': recorded_reason, 

738 'latency_ms': round(draft_latency_ms, 1), 

739 'energy_kwh': round( 

740 self._registry.get_total_energy_kwh(hours=0.01), 6), 

741 } 

742 

743 # ─── SRP helpers extracted from dispatch_draft_first ─── 

744 

745 # Class-level toggle the health probe can flip off in tests. Prod 

746 # leaves it enabled so dead-port POSTs short-circuit cleanly; unit 

747 # tests that mock _dispatch_to_model set it to False so the mocked 

748 # dispatch actually runs. Kept as a class attribute (not instance) 

749 # so fixtures can patch once for the whole suite. 

750 _health_probe_enabled: bool = True 

751 

752 def _check_draft_first_gates(self, prompt: str) -> Optional[str]: 

753 """Run the cheap gates (circuit breaker + constitutional filter + 

754 draft-server health probe) that must pass before we spend any 

755 model time. 

756 

757 Returns None on success, or an error string identifying which gate 

758 rejected the request. Keeps dispatch_draft_first's orchestration 

759 thin — this method owns "is the system healthy enough to proceed". 

760 """ 

761 from security.hive_guardrails import HiveCircuitBreaker, ConstitutionalFilter 

762 if HiveCircuitBreaker.is_halted(): 

763 return 'Hive is halted' 

764 passed, reason = ConstitutionalFilter.check_prompt(prompt) 

765 if not passed: 

766 return reason or 'constitutional filter' 

767 # Fast TCP probe against the draft server. If the 0.8B caption 

768 # server (port 8081) isn't listening, fall through to the normal 

769 # 4B path instead of POSTing to a dead port and waiting for a 

770 # socket timeout on every chat request. Cache the result for 30s 

771 # so we don't probe on every message. 

772 if self._health_probe_enabled and not self._draft_server_alive(): 

773 return 'draft_server_offline' 

774 return None 

775 

776 _draft_probe_ts: float = 0.0 

777 _draft_probe_ok: bool = False 

778 

779 def _draft_server_alive(self) -> bool: 

780 """Cheap TCP probe against the draft server endpoint. Cached 

781 for 30s so the dispatcher stays responsive under chat load. 

782 Returns True if a connect() to the draft host:port succeeds.""" 

783 import socket 

784 import time as _t 

785 now = _t.time() 

786 if now - self._draft_probe_ts < 30.0: 

787 return self._draft_probe_ok 

788 ok = False 

789 try: 

790 from core.port_registry import get_local_draft_url 

791 url = get_local_draft_url() 

792 # http://host:port/v1 → (host, port) 

793 _body = url.split('://', 1)[-1].split('/', 1)[0] 

794 host, _, port_s = _body.partition(':') 

795 port = int(port_s) if port_s else 80 

796 with socket.create_connection((host, port), timeout=0.5): 

797 ok = True 

798 except Exception: 

799 ok = False 

800 self.__class__._draft_probe_ts = now 

801 self.__class__._draft_probe_ok = ok 

802 return ok 

803 

804 def _build_draft_classifier_prompt( 

805 self, user_prompt: str, agent_persona: Optional[str] = None, 

806 preferred_lang: str = 'en', 

807 recent_turns: Optional[List[Dict]] = None, 

808 ) -> str: 

809 """Wrap the user prompt with the draft-first classifier instruction. 

810 

811 The draft's job is twofold: (a) produce a short standby reply fit 

812 for simple chat, (b) self-assess whether a bigger model is needed. 

813 The JSON schema is flat so a 0.8B model can reliably emit it. 

814 

815 If ``agent_persona`` is provided, it's prepended to the instruction 

816 so the draft's reply is in the voice of the custom / system agent 

817 the user is talking to instead of a generic first-responder. Used 

818 for the Path-2 system-agent case (e.g. Nunba personality agent). 

819 

820 If ``recent_turns`` is provided, prior conversation context is 

821 rendered as a "Recent conversation" block before the current user 

822 prompt so the draft can answer follow-ups in context (e.g. user 

823 asks "what's happening?" 60s after asking about WhatsApp — without 

824 history the draft would treat each turn as first-contact and emit 

825 a generic greeting). Capped at 4 turns to fit the 0.8B context 

826 budget; oldest first; long messages are truncated to 400 chars. 

827 

828 Owns ONLY prompt construction — no I/O, no side effects. 

829 """ 

830 # Default brand identity when no explicit persona is supplied. 

831 # Without this fall-back, the user's "who are you?" turn drops 

832 # through to the underlying model's training-default name 

833 # ("I'm Qwen3.5...") because cf3e337 deliberately removed every 

834 # "You are <internal-role>" sentence to fix an identity-leak 

835 # where the draft echoed "first-responder" architecture jargon. 

836 # That fix was correct in spirit but went one step too far — 

837 # the BRAND identity (the user-facing product name "Nunba") is 

838 # not architecture jargon and is exactly what the user expects 

839 # to hear when no per-agent persona is selected. The 

840 # ``agent_persona`` branch below overrides this for any turn 

841 # where a specific persona is in scope, so explicit personas 

842 # are unaffected. 

843 # 

844 # Single source of truth — core.constants.NUNBA_BRAND_IDENTITY. 

845 # Same constant is imported by Nunba's _fallback_chat in 

846 # routes/hartos_backend_adapter.py so the two paths can never 

847 # drift on brand wording. 

848 from core.constants import NUNBA_BRAND_IDENTITY 

849 persona_block = f"{NUNBA_BRAND_IDENTITY}\n\n" 

850 if agent_persona: 

851 # Cap the persona at ~800 chars so a long system prompt doesn't 

852 # blow the 0.8B model's context budget on a single-turn call. 

853 snippet = agent_persona.strip()[:800] 

854 persona_block = ( 

855 "You are playing the following persona — reply in this " 

856 "voice, but keep the JSON schema below exactly as " 

857 "specified. Persona:\n" 

858 f"{snippet}\n\n" 

859 ) 

860 lang_block = '' 

861 if preferred_lang and not preferred_lang.startswith('en'): 

862 try: 

863 from core.constants import SUPPORTED_LANG_DICT 

864 lang_name = SUPPORTED_LANG_DICT.get(preferred_lang[:2], preferred_lang) 

865 except ImportError: 

866 lang_name = preferred_lang 

867 # Same language + tone prompt the 4B path uses (with examples, code-mixing rules) 

868 _tone = '' 

869 try: 

870 from core.agent_personality import get_regional_tone_prompt 

871 _tone = get_regional_tone_prompt(preferred_lang) 

872 except Exception: 

873 pass 

874 lang_block = ( 

875 f"Answer questions accurately and respond as quickly as possible in {lang_name}. " 

876 f"Keep responses under 200 words. Be colloquial and natural.\n" 

877 f"{_tone}\n\n" 

878 ) 

879 

880 # Compute the runtime capability summary ONCE per prompt build so 

881 # the static-tools / ModelCatalog / MCP / channel walks don't run 

882 # twice for the conditional injection below. 

883 cap_summary = _capability_summary_safe() 

884 cap_block = ( 

885 f"Available capabilities (the system can do these via the " 

886 f"routing path below): {cap_summary}.\n\n" 

887 if cap_summary else "" 

888 ) 

889 

890 # Recent conversation context — single source via 

891 # seed_autogen_from_shared_history (the autogen path uses the same 

892 # call), formatted into a flat User:/Assistant: transcript so the 

893 # 0.8B can read follow-ups in context. Capped at 4 turns + 400 

894 # chars/turn to fit the draft's context budget. 

895 history_block = "" 

896 if recent_turns: 

897 _hist_lines = [] 

898 for _turn in recent_turns[-4:]: 

899 _role = _turn.get('role') or '' 

900 _content = (_turn.get('content') or '').strip() 

901 if not _content: 

902 continue 

903 if len(_content) > 400: 

904 _content = _content[:400] + '…' 

905 if _role == 'user': 

906 _hist_lines.append(f"User: {_content}") 

907 elif _role == 'assistant': 

908 _hist_lines.append(f"Assistant: {_content}") 

909 if _hist_lines: 

910 history_block = ( 

911 "Recent conversation (oldest first; use this to " 

912 "interpret the user's current message in context — " 

913 "follow-ups like 'what's happening?' or 'why?' refer " 

914 "back to these turns):\n" 

915 + "\n".join(_hist_lines) 

916 + "\n\n" 

917 ) 

918 

919 return ( 

920 persona_block 

921 + lang_block 

922 # ── Job + answering rules — size-agnostic, identity-free ───── 

923 # Same wording works whether the draft is 0.8B, 4B, or 27B. 

924 # The model in this slot does NOT see the full tool registry 

925 # (web fetch, code exec, GitHub, filesystem, vision, computer 

926 # control, MCP servers, channels), the user's loaded persona, 

927 # multi-turn memory, or the ReAct loop — so it must never 

928 # refuse on behalf of the system. 

929 # 

930 # The 3ea8648 prompt opened with "You are a fast local 

931 # first-responder" and the model would echo it verbatim on 

932 # "who are you?" → "I'm your fast local first-responder, 

933 # ready to assist you right away." cf3e337 fixed that by 

934 # removing every internal-role identity sentence, but went 

935 # one step too far — with NO identity at all the 0.8B fell 

936 # through to its training-default name ("I'm Qwen…"). The 

937 # default brand identity ("You are Nunba…") now lives in 

938 # persona_block above, so this section deliberately does 

939 # NOT add another "You are <X>" line — only the BRAND 

940 # identity above is allowed; INTERNAL-ROLE jargon 

941 # ("first-responder", "draft", "classifier") stays out. 

942 # All instructions below are phrased as the *job* and 

943 # *rules*, never as architecture identity. 

944 + "Your job is to produce a short reply to the user AND " 

945 "classify the user's intent on several independent axes. The " 

946 "classification flags route the message downstream — be " 

947 "accurate.\n\n" 

948 # Positive capability summary — primary teaching mechanism so 

949 # the model knows what the system CAN do. Auto-discovered: 

950 # static tool list + ModelCatalog (TTS/STT/VLM/video/audio, 

951 # rolled up by type) + MCP servers + channels + expert-agent 

952 # categories. Computed once into cap_block above. 

953 + cap_block 

954 + "ANSWERING RULES — READ BEFORE REPLYING:\n" 

955 "You only see this single turn. The system's actual tool / " 

956 "integration / capability set is dynamic and not visible from " 

957 "here — so you don't get to decide what the system can or " 

958 "can't do. Therefore:\n" 

959 "- NEVER write 'I cannot', 'I don't have access', 'I'm unable', " 

960 "'I'm just a', 'I do not have the ability', or any phrase " 

961 "asserting the system can't do something.\n" 

962 "- NEVER claim no internet/tools/file access; you have no way " 

963 "to verify what is or isn't reachable.\n" 

964 "- You MAY answer directly ONLY for trivial recall, simple " 

965 "math, greetings, or questions that need NO live data, NO " 

966 "external system access, NO filesystem, NO code execution, " 

967 "and NO per-user state beyond this single turn.\n" 

968 "- For ANYTHING else (URLs, code, files, current state, " 

969 "multi-step reasoning, capability questions): set " 

970 "delegate=\"local\" (or \"hive\" for very large requests) " 

971 "and write a brief standby reply such as \"Let me check that " 

972 "for you…\", \"Looking that up…\", or \"One moment…\". The " 

973 "standby is replaced by the authoritative answer automatically " 

974 "— your only job is to keep the user comfortable while that " 

975 "runs.\n" 

976 "- Refusals are not your call. If you ever feel the urge to " 

977 "refuse: pick a standby instead and delegate.\n\n" 

978 + history_block 

979 + f"User: {user_prompt}\n\n" 

980 "Respond with ONE JSON object on a single line and NOTHING else:\n" 

981 '{"reply": "<your short reply to the user, 1-3 sentences>", ' 

982 '"delegate": "none" OR "local" OR "hive", ' 

983 '"confidence": <float 0-1>, ' 

984 '"is_correction": true OR false, ' 

985 '"is_casual": true OR false, ' 

986 '"is_create_agent": true OR false, ' 

987 '"channel_connect": "<channel name or empty string>", ' 

988 '"language_change": "<ISO 639-1 code or empty string>", ' 

989 '"invite_intent": "<short context if user wants invite link, or empty>", ' 

990 '"join_room_intent": "<platform + room/url if user wants agent to join, or empty>", ' 

991 '"memory_query": "<short context if user asks about past conversations, or empty>", ' 

992 '"reason": "<why you chose this delegate value>"}\n\n' 

993 # ── delegate ──────────────────────────────────────────────── 

994 "delegate: Use \"none\" for greetings, small-talk, factual " 

995 "questions you can fully answer yourself, or anything that needs " 

996 "no external tools. Use \"local\" if the request needs tools, " 

997 "code, reasoning, or multi-step work the 4B model can handle. " 

998 "Use \"hive\" if it needs large-model expertise, long-context " 

999 "research, or specialized skill distribution.\n\n" 

1000 # ── is_correction ──────────────────────────────────────────── 

1001 "is_correction: true when the user is telling you something " 

1002 "in the previous assistant turn was wrong, inaccurate, or " 

1003 "they're restating what they actually meant (e.g. 'no that's " 

1004 "wrong', 'actually, I meant X', 'not quite', 'you got it " 

1005 "wrong'). Otherwise false. Routes the turn into the hive's " 

1006 "real-time learning pipeline, so prefer false when unsure.\n\n" 

1007 # ── is_casual ──────────────────────────────────────────────── 

1008 "is_casual: true when the message is pure chit-chat, a " 

1009 "greeting, an acknowledgement, or anything that clearly " 

1010 "doesn't need any tools, search, computer control, agent " 

1011 "creation, or multi-step reasoning. Used to skip the heavy "