teknium1
9ba5850e95
Six of eight memory providers spawned plain threading.Thread for prefetch/sync/
writer work. A plain thread starts with an EMPTY contextvars.Context, so under
multiplex profiles the worker resolved the DEFAULT profile's HERMES_HOME (and
fails closed on scoped secrets). honcho and hindsight had each noticed and
written their own copy_context() wrapper; core had a third in memory_manager.
One canonical pair now lives on the ABC module every provider already imports:
agent/memory_provider.py::ctx_bound / spawn_context_thread. memory_manager,
honcho, hindsight, mem0, retaindb, byterover, supermemory and openviking all use
it; the honcho and hindsight wrappers and memory_manager._ctx_bound are deleted.
Five "json.loads(path.read_text()) or {}" readers (mem0._read_mem0_json,
honcho client/oauth/cli _read_config, hindsight save_config/_load_config) fold
into utils.read_json_or_empty, the read half of every read-merge-atomic_json_write
sidecar store.
holographic.save_config was the only config.yaml writer in the tree that
bypassed hermes_cli.config.save_config: raw open("w") + yaml.dump with no config
lock, no managed-mode refusal, no atomic replace, and a swallowed exception. It
now calls save_config(..., merge_existing=True). Behavior change: a managed
install refuses the write (previously silently rewrote config.yaml); other
sections are deep-merged instead of round-tripped through a raw dump.
openviking._hermes_home_path guarded an impossible ImportError of
hermes_constants (the module already imports agent.*) with a ~/.hermes fallback
that is wrong on Windows and under profile overrides; it is replaced by
get_hermes_home() directly.
Tests: tests/plugins/memory/test_provider_threads_inherit_profile.py drives each
provider's real spawn path with a fake backend and asserts the thread sees the
spawner's HERMES_HOME override (sabotage: retaindb back on threading.Thread ->
red). tests/plugins/memory/test_holographic_save_config.py pins merge-with-
existing-sections and managed-mode refusal (sabotage: raw yaml.dump -> red).
188 lines
10 KiB
Python
188 lines
10 KiB
Python
"""Dialectic layer for HonchoMemoryProvider: multi-pass reasoning, cadence/backoff, liveness."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
import threading
|
||
import time
|
||
|
||
from agent.memory_provider import spawn_context_thread
|
||
|
||
logger = logging.getLogger("plugins.memory.honcho")
|
||
|
||
_STRUCTURE_RE = re.compile(r"^(?:[*-] |\s*\d+\. )", re.MULTILINE) # bullet / ordered-list line
|
||
|
||
|
||
class DialecticMixin:
|
||
"""Dialectic prefetch state machine. State attributes are initialised in
|
||
HonchoMemoryProvider.__init__; this mixin only reads/writes them."""
|
||
|
||
# Reasoning level per (depth, pass) when dialecticDepthLevels is not configured;
|
||
# "base" means dialecticReasoningLevel. Early passes of deeper runs are lighter.
|
||
_PROPORTIONAL_LEVELS: dict[tuple[int, int], str] = {
|
||
(1, 0): "base", (2, 0): "minimal", (2, 1): "base", (3, 0): "minimal", (3, 1): "base", (3, 2): "low",
|
||
}
|
||
_LEVEL_ORDER = ("minimal", "low", "medium", "high", "max")
|
||
# Char-count thresholds for the query-length reasoning heuristic.
|
||
_HEURISTIC_LENGTH_MEDIUM, _HEURISTIC_LENGTH_HIGH = 120, 400
|
||
# A thread older than timeout × multiplier is treated as dead so a hung Honcho call
|
||
# can't block future fires indefinitely.
|
||
_STALE_THREAD_MULTIPLIER = 2.0
|
||
# A pending result fired more than cadence × multiplier turns ago is discarded on read,
|
||
# so a stale conversational pivot isn't injected after trivial-prompt turns.
|
||
_STALE_RESULT_MULTIPLIER = 2
|
||
_BACKOFF_MAX = 8 # ceiling for the empty-streak backoff (× base cadence)
|
||
# Turn-1 budgets: session init + base retrieval share the first; dialectic gets the second.
|
||
_FIRST_TURN_BASE_TIMEOUT, _FIRST_TURN_DIALECTIC_CAP = 3.0, 2.0
|
||
|
||
def _thread_is_live(self) -> bool:
|
||
"""Thread-alive guard that treats threads past the stale threshold as dead."""
|
||
if not self._prefetch_thread or not self._prefetch_thread.is_alive():
|
||
return False
|
||
timeout = self._config.timeout if self._config and self._config.timeout else 8.0
|
||
stale_after = timeout * self._STALE_THREAD_MULTIPLIER
|
||
age = time.monotonic() - self._prefetch_thread_started_at
|
||
if age > stale_after:
|
||
logger.debug("Honcho prefetch thread age %.1fs exceeds stale threshold %.1fs — treating as dead", age, stale_after)
|
||
return False
|
||
return True
|
||
|
||
def _effective_cadence(self) -> int:
|
||
"""Cadence plus empty-streak backoff, capped at _BACKOFF_MAX × base."""
|
||
cadence, streak = self._dialectic_cadence, self._dialectic_empty_streak
|
||
return min(cadence + streak, cadence * self._BACKOFF_MAX) if streak > 0 else cadence
|
||
|
||
def _note_dialectic_failure(self, exc: BaseException) -> None:
|
||
"""Widen the empty-streak backoff after a failed cycle; auth failures are exempt
|
||
because waiting cannot fix a dead token."""
|
||
from plugins.memory.honcho.session import HonchoAuthError
|
||
if isinstance(exc, HonchoAuthError):
|
||
logger.warning("Honcho dialectic auth failure (not counted toward cadence backoff): %s", exc)
|
||
return
|
||
self._dialectic_empty_streak += 1
|
||
|
||
def _spawn_dialectic(
|
||
self, query: str, *, thread_name: str, fired_at: int, log_label: str, use_query_rewrite: bool = True,
|
||
) -> threading.Thread:
|
||
"""Start a background dialectic run that publishes into the pending-result slot. Only a
|
||
non-empty result advances ``_last_dialectic_turn`` (so empty returns retry next turn) and
|
||
resets the empty streak; failures widen the backoff."""
|
||
def _run() -> None:
|
||
try:
|
||
r = self._run_dialectic_depth(query, use_query_rewrite=use_query_rewrite)
|
||
except Exception as exc:
|
||
logger.debug("Honcho %s failed: %s", log_label, exc)
|
||
self._note_dialectic_failure(exc)
|
||
return
|
||
if r and r.strip():
|
||
with self._prefetch_lock:
|
||
self._prefetch_result = r
|
||
self._prefetch_result_fired_at = fired_at
|
||
self._last_dialectic_turn = fired_at
|
||
self._dialectic_empty_streak = 0
|
||
else:
|
||
self._dialectic_empty_streak += 1
|
||
|
||
self._prefetch_thread_started_at = time.monotonic()
|
||
thread = spawn_context_thread(_run, name=thread_name)
|
||
thread.start()
|
||
self._prefetch_thread = thread
|
||
return thread
|
||
|
||
def _consume_pending_dialectic(self) -> str:
|
||
"""Pop the pending dialectic result, or "" when none is ready or it is stale."""
|
||
with self._prefetch_lock:
|
||
dialectic_result, fired_at = self._prefetch_result, self._prefetch_result_fired_at
|
||
self._prefetch_result, self._prefetch_result_fired_at = "", -999
|
||
stale_limit = self._dialectic_cadence * self._STALE_RESULT_MULTIPLIER
|
||
if dialectic_result and fired_at >= 0 and (self._turn_count - fired_at) > stale_limit:
|
||
logger.debug("Honcho pending dialectic discarded as stale: fired_at=%d, turn=%d, limit=%d",
|
||
fired_at, self._turn_count, stale_limit)
|
||
return ""
|
||
return dialectic_result if (dialectic_result and dialectic_result.strip()) else ""
|
||
|
||
def _apply_reasoning_heuristic(self, base: str, query: str) -> str:
|
||
"""Scale ``base`` up by query length (+1 at >=120 chars, +2 at >=400), clamped at reasoning_level_cap."""
|
||
if not self._reasoning_heuristic or not query or base not in self._LEVEL_ORDER:
|
||
return base
|
||
n = len(query)
|
||
bump = 0 if n < self._HEURISTIC_LENGTH_MEDIUM else 1 if n < self._HEURISTIC_LENGTH_HIGH else 2
|
||
cap_idx = self._LEVEL_ORDER.index(self._reasoning_level_cap)
|
||
return self._LEVEL_ORDER[min(self._LEVEL_ORDER.index(base) + bump, cap_idx)]
|
||
|
||
def _resolve_pass_level(self, pass_idx: int, query: str = "") -> str:
|
||
"""Reasoning level for a pass: explicit dialecticDepthLevels win, then the
|
||
proportional table, then the base level scaled by the length heuristic."""
|
||
if self._dialectic_depth_levels and pass_idx < len(self._dialectic_depth_levels):
|
||
return self._dialectic_depth_levels[pass_idx]
|
||
mapping = self._PROPORTIONAL_LEVELS.get((self._dialectic_depth, pass_idx))
|
||
if mapping is None or mapping == "base":
|
||
base = self._config.dialectic_reasoning_level if self._config else "low"
|
||
return self._apply_reasoning_heuristic(base, query)
|
||
return mapping
|
||
|
||
def _build_dialectic_prompt(self, pass_idx: int, prior_results: list[str], is_cold: bool) -> str:
|
||
"""Pass 0: cold (general) or warm (session-scoped) query; pass 1: gap audit
|
||
against the prior result; pass 2: reconciliation across prior passes."""
|
||
if pass_idx == 0:
|
||
if is_cold:
|
||
return ("Who is this person? What are their preferences, goals, and working style? "
|
||
"Focus on facts that would help an AI assistant be immediately useful.")
|
||
return ("Given what's been discussed in this session so far, what context about this user is most "
|
||
"relevant to the current conversation? Prioritize active context over biographical facts.")
|
||
if pass_idx == 1:
|
||
prior = prior_results[-1] if prior_results else ""
|
||
return (f"Given this initial assessment:\n\n{prior}\n\n"
|
||
"What gaps remain in your understanding that would help going forward? Synthesize what you "
|
||
"actually know about the user's current state and immediate needs, grounded in evidence from "
|
||
"recent sessions.")
|
||
p0, p1 = [prior_results[i] if len(prior_results) > i else "(empty)" for i in (0, 1)]
|
||
return (f"Prior passes produced:\n\nPass 1:\n{p0}\n\nPass 2:\n{p1}\n\n"
|
||
"Do these assessments cohere? Reconcile any contradictions and produce a final, concise "
|
||
"synthesis of what matters most for the current conversation.")
|
||
|
||
@staticmethod
|
||
def _signal_sufficient(result: str) -> bool:
|
||
"""True when a pass returned enough signal to skip further passes: >100 chars
|
||
with structure (headers/bullets/ordered list), or >300 chars regardless."""
|
||
if not result or len(result.strip()) < 100:
|
||
return False
|
||
structured = "\n" in result and bool("##" in result or "•" in result or _STRUCTURE_RE.search(result))
|
||
return structured or len(result.strip()) > 300
|
||
|
||
def _run_dialectic_depth(self, query: str, *, use_query_rewrite: bool = True) -> str:
|
||
"""Run up to dialecticDepth .chat() passes, bailing early once a pass returns
|
||
strong signal. Returns the last non-empty result."""
|
||
if not self._manager or not self._session_key:
|
||
return ""
|
||
is_cold = not self._base_context_cache
|
||
results: list[str] = []
|
||
rewritten_query = ""
|
||
if use_query_rewrite and self._query_rewrite_enabled and self._query_rewriter:
|
||
try:
|
||
rewritten_query = self._query_rewriter(query).strip()
|
||
except Exception as exc:
|
||
logger.debug("Honcho query rewriter failed: %s", exc)
|
||
|
||
depth = self._dialectic_depth
|
||
for i in range(depth):
|
||
# Dependent prompts require a non-empty prior result; without one, later
|
||
# passes retry the independent base prompt.
|
||
prior_results = [r for r in results if r and r.strip()]
|
||
if prior_results and self._signal_sufficient(prior_results[-1]):
|
||
logger.debug("Honcho dialectic depth %d: pass %d skipped, prior signal sufficient", depth, i)
|
||
break
|
||
if prior_results:
|
||
prompt = self._build_dialectic_prompt(i, prior_results, is_cold)
|
||
else:
|
||
if i:
|
||
logger.debug("Honcho dialectic depth %d: pass %d has no non-empty prior — "
|
||
"falling back to base prompt", depth, i)
|
||
prompt = rewritten_query or self._build_dialectic_prompt(0, [], is_cold)
|
||
level = self._resolve_pass_level(i, query=query)
|
||
logger.debug("Honcho dialectic depth %d: pass %d, level=%s, cold=%s", depth, i, level, is_cold)
|
||
result = self._manager.dialectic_query(self._session_key, prompt, reasoning_level=level, peer="user")
|
||
results.append(result or "")
|
||
return next((r for r in reversed(results) if r and r.strip()), "")
|