| """Dialectic layer for HonchoMemoryProvider: multi-pass reasoning, cadence/backoff, liveness.""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| import re |
| import threading |
| import time |
|
|
| from plugins.memory.honcho.client import spawn_context_thread |
|
|
| logger = logging.getLogger("plugins.memory.honcho") |
|
|
| _STRUCTURE_RE = re.compile(r"^(?:[*-] |\s*\d+\. )", re.MULTILINE) |
|
|
|
|
| class DialecticMixin: |
| """Dialectic prefetch state machine. State attributes are initialised in |
| HonchoMemoryProvider.__init__; this mixin only reads/writes them.""" |
|
|
| |
| |
| _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") |
| |
| _HEURISTIC_LENGTH_MEDIUM, _HEURISTIC_LENGTH_HIGH = 120, 400 |
| |
| |
| _STALE_THREAD_MULTIPLIER = 2.0 |
| |
| |
| _STALE_RESULT_MULTIPLIER = 2 |
| _BACKOFF_MAX = 8 |
| |
| _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, owner=self) |
| 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): |
| |
| |
| 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()), "") |
|
|
