diff --git a/crates/tracedecay-agent-hosts/src/agents/hermes/lifecycle.rs b/crates/tracedecay-agent-hosts/src/agents/hermes/lifecycle.rs index d83c5f1bd5..8d2c85a565 100644 --- a/crates/tracedecay-agent-hosts/src/agents/hermes/lifecycle.rs +++ b/crates/tracedecay-agent-hosts/src/agents/hermes/lifecycle.rs @@ -110,8 +110,8 @@ mod tests { "config should select tracedecay memory provider:\n{config}" ); assert!( - config.contains("engine: tracedecay"), - "config should select tracedecay context engine:\n{config}" + !config.contains("engine: tracedecay"), + "config must leave Hermes' native context engine selected:\n{config}" ); } @@ -144,7 +144,7 @@ mod tests { let named_config = text(&home.path().join(".hermes/profiles/work/config.yaml")); assert!(named_config.contains("- tracedecay")); assert!(named_config.contains("provider: tracedecay")); - assert!(named_config.contains("engine: tracedecay")); + assert!(!named_config.contains("engine: tracedecay")); assert_eq!( super::super::hermes_registration_state( home.path(), diff --git a/crates/tracedecay-agent-hosts/src/agents/hermes/profile_config.rs b/crates/tracedecay-agent-hosts/src/agents/hermes/profile_config.rs index 2b3b83ec72..f238136ae1 100644 --- a/crates/tracedecay-agent-hosts/src/agents/hermes/profile_config.rs +++ b/crates/tracedecay-agent-hosts/src/agents/hermes/profile_config.rs @@ -85,12 +85,12 @@ pub(super) fn registration_state( .and_then(|memory| string_value(&memory, "provider")) .as_deref() == Some("tracedecay"); - let context = root + let replaces_compressor = root .get_mapping("context") .and_then(|context| string_value(&context, "engine")) .as_deref() == Some("tracedecay"); - if enabled && memory && context { + if enabled && memory && !replaces_compressor { State::Current } else { State::Repairable @@ -145,7 +145,6 @@ pub(super) fn disable_plugin(config_path: &Path) -> Result<()> { // install/update surfaces stay stable. const PLUGINS_ERR: &str = "unsupported Hermes plugins config"; const MEMORY_ERR: &str = "unsupported Hermes memory config"; -const CONTEXT_ERR: &str = "unsupported Hermes context config"; // The mutation path is deliberately text/span oriented rather than CST oriented. // `yaml-edit` (as of 0.2.3) cannot reliably synthesize nested block or flow @@ -159,7 +158,7 @@ const CONTEXT_ERR: &str = "unsupported Hermes context config"; /// The containers enabling TraceDecay may add to a profile, as JSON pointers /// into the YAML document. -const OWNED_CONTAINERS: [&str; 4] = ["/plugins", "/plugins/enabled", "/memory", "/context"]; +const OWNED_CONTAINERS: [&str; 3] = ["/plugins", "/plugins/enabled", "/memory"]; /// The enabled config and the [`OWNED_CONTAINERS`] enabling added to it. fn enable_plugin_config( @@ -216,20 +215,13 @@ fn enable_normalized(existing: &str) -> std::result::Result { "memory", "provider", "tracedecay", - &[], MEMORY_ERR, "Hermes memory provider already configured; refusing to overwrite it", )?; - let text = ensure_scalar( - &text, - "context", - "engine", - "tracedecay", - &["compressor"], - CONTEXT_ERR, - "Hermes context engine already configured; refusing to overwrite it", - )?; - Ok(text) + // TraceDecay owns retrieval and memory, not Hermes' native compressor. + // Repair profiles activated by the former no-op engine while preserving + // every user-selected engine and the rest of the context configuration. + disable_scalar(&text, "context", "engine", false) } fn disable_normalized( @@ -394,7 +386,6 @@ fn ensure_scalar( container: &str, key: &str, value: &str, - allow_overwrite: &[&str], unsupported: &str, conflict: &str, ) -> std::result::Result { @@ -424,21 +415,6 @@ fn ensure_scalar( } match string_value(mapping, key).as_deref() { Some(current) if current == value => Ok(text.to_string()), - Some(current) if allow_overwrite.contains(¤t) => { - let entry = mapping - .find_entry_by_key(key) - .ok_or_else(|| unsupported.to_string())?; - let range = entry - .value_node() - .and_then(|node| node.as_scalar().map(yaml_edit::Scalar::byte_range)) - .ok_or_else(|| unsupported.to_string())?; - Ok(format!( - "{}{}{}", - &text[..range.start as usize], - value, - &text[range.end as usize..] - )) - } Some(_) => Err(conflict.to_string()), None => Err(unsupported.to_string()), } @@ -833,7 +809,7 @@ mod tests { .as_deref(), Some("tracedecay") ); - assert_eq!( + assert_ne!( root.get_mapping("context") .and_then(|context| string_value(&context, "engine")) .as_deref(), @@ -1004,11 +980,11 @@ mod tests { for (original, appended) in [ ( "theme: dark\nplugins:\n enabled:\n - foreign\n", - &["/memory", "/context"][..], + &["/memory"][..], ), ( "model: x\nmemory: {}\n", - &["/plugins", "/plugins/enabled", "/context"][..], + &["/plugins", "/plugins/enabled"][..], ), ] { let (enabled, created) = enable_plugin_config(original).unwrap(); @@ -1020,6 +996,48 @@ mod tests { } } + #[test] + fn activation_preserves_native_and_user_selected_context_engines() { + for context in [ + "", + "context: {engine: compressor, budget: 42}\n", + "context: {engine: custom-engine, budget: 42}\n", + ] { + let original = format!( + "plugins: {{enabled: [tracedecay]}}\nmemory: {{provider: tracedecay}}\n\ + {context}compression: {{enabled: false, threshold: 0.6}}\n" + ); + let (updated, created) = enable_plugin_config(&original).unwrap(); + assert_eq!(updated, original); + assert!(created.is_empty()); + } + } + + #[test] + fn activation_repairs_the_unavailable_context_engine_without_losing_settings() { + use crate::agents::host_bundle::HostBundleRegistrationStateV1 as State; + + let dir = TempDir::new().unwrap(); + let config = dir.path().join("config.yaml"); + let original = concat!( + "plugins: {enabled: [tracedecay]}\n", + "memory: {provider: tracedecay}\n", + "context: {engine: tracedecay, budget: 42}\n", + "compression: {enabled: true, threshold: 0.6}\n", + ); + std::fs::write(&config, original).unwrap(); + assert_eq!(registration_state(&config), State::Repairable); + + enable_plugin(&config).unwrap(); + + assert_eq!(registration_state(&config), State::Current); + let updated = read(&config); + assert_enabled(&updated); + assert!(updated.contains("budget: 42")); + assert!(updated.contains("compression: {enabled: true, threshold: 0.6}")); + assert_eq!(enable_plugin_config(&updated).unwrap().0, updated); + } + #[test] fn disable_empty_mapping_preserves_immediately_following_line() { let input = concat!( diff --git a/crates/tracedecay-agent-hosts/src/agents/hermes/templates.rs b/crates/tracedecay-agent-hosts/src/agents/hermes/templates.rs index aba852798e..b00e619a46 100644 --- a/crates/tracedecay-agent-hosts/src/agents/hermes/templates.rs +++ b/crates/tracedecay-agent-hosts/src/agents/hermes/templates.rs @@ -22,7 +22,7 @@ pub(super) fn plugin_manifest( kind: standalone\n\ version: {version}\n\ generator_commit: {commit}\n\ - description: TraceDecay code-graph tools, memory provider, and LCM context engine for Hermes\n\ + description: TraceDecay code-graph tools, memory provider, and LCM retrieval for Hermes\n\ author: tracedecay (generated by `tracedecay install --agent hermes`)\n\ provides_tools:\n{tools}\n\ provides_hooks:\n\ diff --git a/crates/tracedecay-agent-hosts/src/agents/hermes/templates/plugin_init.py b/crates/tracedecay-agent-hosts/src/agents/hermes/templates/plugin_init.py index 85d64c2ac4..087a312dd7 100644 --- a/crates/tracedecay-agent-hosts/src/agents/hermes/templates/plugin_init.py +++ b/crates/tracedecay-agent-hosts/src/agents/hermes/templates/plugin_init.py @@ -12,6 +12,7 @@ import threading import time from collections import deque +from functools import partial from pathlib import Path from . import schemas, tools @@ -38,12 +39,6 @@ class MemoryProvider: pass -try: - from agent.context_engine import ContextEngine -except Exception: - class ContextEngine: - pass - # Hermes' centralized auxiliary LLM facade is the MODULE-LEVEL # agent.auxiliary_client.call_llm(task=..., messages=..., ...). AIAgent # instances carry no ``auxiliary_client`` attribute and no host call site @@ -229,7 +224,8 @@ def _resolve_auxiliary_client(agent=None): "type": "string", "description": "Optional session id override (for example, expand a cross-session grep hit in its owning session).", }, - "max_tokens": {"type": "integer", "description": "Token budget for returned content.", "default": 4000}, + "max_tokens": {"type": "integer", "description": "Token budget for returned content.", + "default": 4000}, "source_limit": { "type": "integer", "minimum": 1, @@ -320,21 +316,10 @@ def _resolve_auxiliary_client(agent=None): LCM_TOOL_ALIASES = { schema["name"]: "tracedecay_" + schema["name"] for schema in LCM_NATIVE_SCHEMAS } -LCM_DIRECT_TOOL_NAMES = frozenset(LCM_TOOL_ALIASES.values()) LCM_DIRECT_TO_NATIVE = {tracedecay_name: native_name for native_name, tracedecay_name in LCM_TOOL_ALIASES.items()} -# Public LCM readers never depend on forwarding the host's live message list. -MESSAGE_DEPENDENT_TOOLS = frozenset() - STANDARD_HERMES_LCM_PROVIDER = "hermes" -LCM_PROVIDER_LOCAL_TOOL_NAMES = frozenset(( - "tracedecay_lcm_describe", - "tracedecay_lcm_doctor", - "tracedecay_lcm_expand", - "tracedecay_lcm_expand_query", -)) - # Direct duplicates of the memory provider's own tool surface # (fact_store / fact_feedback / memory_status). Skipped at register() time # when tracedecay is the active memory.provider so the same store is not @@ -349,8 +334,6 @@ def _resolve_auxiliary_client(agent=None): # first-turn guidance nudge so it never advertises tools that are not # actually registered. _REGISTERED_TOOL_NAMES = set() -_CONTEXT_TOOL_NAMES = set() -_HOST_FORWARDS_MESSAGES = None def _active_memory_provider(ctx=None): """The memory.provider configured for this profile, if any.""" @@ -371,19 +354,12 @@ def _active_memory_provider(ctx=None): pass return None -def _make_wrapped_lcm_handler(tool_name: str, engine): - def _wrapped(args: dict, **kwargs) -> str: - return engine.handle_tool_call(tool_name, args, **kwargs) - return _wrapped - -def _host_forwards_registered_tool_messages(ctx) -> bool: - capability = getattr(ctx, "context_engine_tool_handlers_receive_messages", False) - if callable(capability): - try: - capability = capability() - except Exception: - return False - return bool(capability) +def _make_wrapped_lcm_handler(tool_name: str, config=None, hermes_home=None): + return _make_project_safe_handler( + LCM_TOOL_ALIASES[tool_name], + partial(_handle_lcm_tool_call, tool_name, config=config), + hermes_home, + ) _NUDGE_CODE_REQUEST_RE = re.compile( r"\b(" @@ -1094,9 +1070,6 @@ def safe_handler(args, **kwargs): "nudge": (True, "Inject a first-turn tracedecay guidance nudge only for codebase/navigation requests."), "sync_turn": (True, "Mirror each completed turn into the LCM raw store."), "prefetch": (True, "Background fact recall injected at turn start."), - "context_threshold": ("", "Compression trigger as a fraction of the context window (default: hermes compression.threshold)."), - "threshold_tokens": ("", "Absolute compression trigger in tokens (overrides context_threshold)."), - "context_length": ("", "Context window override when the host does not report one."), "expansion_model": ("", "Model used for lcm_expand_query synthesis."), "expansion_context_tokens": ("", "Expanded-context budget for lcm_expand_query (default 32000)."), "expansion_timeout_ms": ("", "lcm_expand_query synthesis timeout in milliseconds."), @@ -1148,7 +1121,7 @@ def _with_plugin_block(config, hermes_home=None): """Layer the profile's plugins.tracedecay config block under a host config. Host-provided values always win; the block fills the gaps so profile - config.yaml settings reach the engine/provider without bespoke env vars. + config.yaml settings reach tools and the provider without bespoke env vars. """ block = { key: value @@ -1297,116 +1270,6 @@ def _lcm_int_setting(config, env_key, *names, default=None): pass return _configured_int(config, *names, default=default) -def _config_bool_disabled(value): - if isinstance(value, bool): - return value is False - if isinstance(value, (int, float)): - return value == 0 - if isinstance(value, str): - normalized = value.strip().lower() - if normalized in ("0", "false", "no", "off"): - return True - try: - return float(normalized) == 0 - except ValueError: - return False - return False - -def _hermes_yaml_compression_threshold(default, hermes_home=None): - # Port of hermes-lcm config._hermes_compression_threshold: read the main - # Hermes compression.threshold from the resolved host config when no LCM - # override exists. Disabled Hermes compression must not leak its threshold. - home = ( - hermes_home - or os.path.join(os.path.expanduser("~"), ".hermes") - ) - cfg_path = Path(home) / "config.yaml" - try: - text = cfg_path.read_text() - except Exception: - return default - try: - import yaml - except Exception: - yaml = None - try: - if yaml is not None: - cfg = yaml.safe_load(text) or {} - compression = cfg.get("compression") or {} - if _config_bool_disabled(compression.get("enabled")): - return default - value = compression.get("threshold") - if value is None: - return default - return float(value) - - in_compression = False - direct_indent = None - compression_disabled = False - threshold_value = None - for raw_line in text.splitlines(): - line = raw_line.split('#', 1)[0].rstrip() - if not line.strip(): - continue - if not line.startswith((" ", "\t")): - in_compression = line.strip() == "compression:" - direct_indent = None - continue - if not in_compression: - continue - indent = len(line) - len(line.lstrip(" \t")) - if direct_indent is None: - direct_indent = indent - if indent != direct_indent or ":" not in line: - continue - key, raw_value = line.strip().split(":", 1) - value = raw_value.strip().strip("'\"") - if key == "enabled" and _config_bool_disabled(value): - compression_disabled = True - elif key == "threshold": - threshold_value = value - if compression_disabled or threshold_value is None: - return default - return float(threshold_value) - except Exception: - return default - -def _lcm_context_threshold(config, hermes_home=None): - raw = os.environ.get("LCM_CONTEXT_THRESHOLD") - if raw is not None: - try: - return float(raw) - except (TypeError, ValueError): - pass - configured = _configured_value(config, "context_threshold") - if configured is not None: - try: - return float(configured) - except (TypeError, ValueError): - pass - return _hermes_yaml_compression_threshold(0.75, hermes_home=hermes_home) - -def _configured_threshold_tokens(config, hermes_home=None, context_length_override=None): - explicit = _configured_int(config, "threshold_tokens") - if explicit is not None: - return explicit - context_length = context_length_override - if context_length is None: - context_length = _configured_int( - config, - "context_length", - "max_context_tokens", - "model_context_tokens", - ) - if context_length is None: - return None - try: - return int(int(context_length) * float(_lcm_context_threshold(config, hermes_home=hermes_home))) - except (TypeError, ValueError): - return None - - - def _lcm_expansion_model(config): value = _lcm_str_setting(config, "LCM_EXPANSION_MODEL", "expansion_model", default="") return str(value or "").strip() @@ -1588,9 +1451,20 @@ def _synthesize_expand_query_payload(retrieval, agent=None, **kwargs): def _handle_lcm_expand_query(args, **kwargs) -> str: kwargs = dict(kwargs) agent = kwargs.pop("agent", None) + config = _with_plugin_block(kwargs.pop("config", None), kwargs.pop("hermes_home", None)) args = dict(args or {}) args.setdefault("provider", STANDARD_HERMES_LCM_PROVIDER) - retrieval = call_tracedecay_json("tracedecay_lcm_expand_query", args, **kwargs) + args.setdefault("context_max_tokens", _lcm_expansion_context_tokens(config)) + retrieval = call_tracedecay_json( + "tracedecay_lcm_expand_query", args, + **_project_call_kwargs(kwargs.get("project_root")), + ) + if kwargs.get("model") is None: + model = _lcm_expansion_model(config) + if model: + kwargs["model"] = model + if kwargs.get("timeout") is None and kwargs.get("expansion_timeout") is None: + kwargs["expansion_timeout"] = _lcm_expansion_timeout_ms(config) / 1000 payload = _synthesize_expand_query_payload(retrieval, agent=agent, **kwargs) return json.dumps(payload) @@ -1681,503 +1555,41 @@ def _translate_lcm_args(native_name: str, args: dict) -> dict: return translated return translated -_ENGINE_DEFAULT_SESSION = "__default__" - -class _EngineSessionState: - """Mutable per-conversation engine state. - - The host registers ONE engine instance and shares it across every - AIAgent in the process (gateway sessions, parallel delegate_task - children), so conversation-scoped fields must never live directly on - the engine. - """ - - _FIELDS = ( - "project_root", - "agent", - "model", - "last_prompt_tokens", - "last_real_prompt_tokens", - "last_completion_tokens", - "last_total_tokens", - "context_length", - "threshold_tokens", - "compression_count", - "last_compress_result", - "_last_compress_aborted", - "_last_summary_error", - "_runtime_context_length", - "_session_start_context_length", - ) - - def __init__(self): - self.project_root = None - self.agent = None - self.model = "" - self.last_prompt_tokens = 0 - self.last_real_prompt_tokens = 0 - self.last_completion_tokens = 0 - self.last_total_tokens = 0 - # Host-contract token state (run_agent.py reads these directly; the - # minimum-context guard in agent/agent_init.py checks context_length). - self.context_length = 0 - self.threshold_tokens = 0 - self.compression_count = 0 - # Compression abort/diagnostic state read by - # agent/conversation_compression.py after each compress() call. - self.last_compress_result = None - self._last_compress_aborted = False - self._last_summary_error = None - self._runtime_context_length = None - self._session_start_context_length = None - - def adopt(self, other): - """Carry conversation state across a compression-rotation rebind.""" - for field in self._FIELDS: - setattr(self, field, getattr(other, field)) - -def _engine_session_property(field): - """Engine attribute view onto the calling context's session state. - - Keeps the host contract intact: run_agent.py reads (and - conversation_compression.py occasionally writes) these as plain - attributes on the shared engine instance, and each access resolves to - the session bound to the calling thread. - """ - - def _get(self): - return getattr(self._state(), field) - - def _set(self, value): - setattr(self._state(), field, value) - - return property(_get, _set) - -class TraceDecayContextEngine(ContextEngine): - def __init__(self, config=None, hermes_home=None): - # Per-session mutable state, guarded by an RLock. Calls that carry - # no session id (compress, update_from_response, should_compress, - # ...) resolve the session bound to the calling thread, falling - # back to the most recently bound session. That keeps concurrent - # gateway sessions / delegate_task children from clobbering each - # other whenever the host gives us a binding signal, and at minimum - # serializes state access when it does not. - self._state_lock = threading.RLock() - self._session_states = {} - self._thread_sessions = threading.local() - self.active_session_id = None - self._host_config = config - self.hermes_home = _resolve_hermes_home(config, hermes_home) - self.config = _with_plugin_block(config, self.hermes_home) - self.project_root = _resolved_project_scope( - _configured_project_root(self.config), - self.hermes_home, - ) - - def __deepcopy__(self, memo): - """Create an agent-local engine without copying locks or live agents.""" - clone = type(self)( - config=copy.deepcopy(self._host_config, memo), - hermes_home=self.hermes_home, - ) - memo[id(self)] = clone - clone.config = copy.deepcopy(self.config, memo) - clone.project_root = self.project_root - with self._state_lock: - clone.active_session_id = self.active_session_id - for key, state in self._session_states.items(): - copied_state = _EngineSessionState() - for field in _EngineSessionState._FIELDS: - if field == "agent": - continue - setattr(copied_state, field, copy.deepcopy(getattr(state, field), memo)) - clone._session_states[key] = copied_state - return clone - - project_root = _engine_session_property("project_root") - agent = _engine_session_property("agent") - model = _engine_session_property("model") - last_prompt_tokens = _engine_session_property("last_prompt_tokens") - last_real_prompt_tokens = _engine_session_property("last_real_prompt_tokens") - last_completion_tokens = _engine_session_property("last_completion_tokens") - last_total_tokens = _engine_session_property("last_total_tokens") - context_length = _engine_session_property("context_length") - threshold_tokens = _engine_session_property("threshold_tokens") - compression_count = _engine_session_property("compression_count") - last_compress_result = _engine_session_property("last_compress_result") - _last_compress_aborted = _engine_session_property("_last_compress_aborted") - _last_summary_error = _engine_session_property("_last_summary_error") - _runtime_context_length = _engine_session_property("_runtime_context_length") - _session_start_context_length = _engine_session_property("_session_start_context_length") - - def _session_key(self, session_id=None): - if session_id: - return str(session_id) - bound = getattr(self._thread_sessions, "session_id", None) - if bound: - return bound - return self.active_session_id or _ENGINE_DEFAULT_SESSION - - def _state(self, session_id=None): - key = self._session_key(session_id) - with self._state_lock: - state = self._session_states.get(key) - if state is None: - state = _EngineSessionState() - self._session_states[key] = state - return state - - @property - def name(self) -> str: - return "tracedecay" - - def _bind_session(self, session_id=None, hermes_home=None, project_root=None, **kwargs): - if session_id is not None: - session_key = str(session_id) - with self._state_lock: - if session_key not in self._session_states: - state = _EngineSessionState() - # Compression rotation continues the same conversation: - # carry model/window/counters over from the predecessor - # session's state (old_session_id on boundary starts, - # else whatever this thread was bound to). - source_key = str(kwargs.get("old_session_id") or "") - source = self._session_states.get(source_key) if source_key else None - if source is None and kwargs.get("boundary_reason") == "compression": - source = self._session_states.get(self._session_key()) - if source is None: - source = self._session_states.get(_ENGINE_DEFAULT_SESSION) - if source is not None: - state.adopt(source) - self._session_states[session_key] = state - # Bind the calling thread so session-less calls (compress, - # update_from_response, should_compress) resolve this session. - self._thread_sessions.session_id = session_key - self.active_session_id = session_key - if kwargs.get("config") is not None: - self._host_config = kwargs.get("config") - if "context_length" in kwargs: - applied_context_length = self._apply_context_length(kwargs.get("context_length")) - if applied_context_length is not None: - self._session_start_context_length = applied_context_length - next_agent = kwargs.get("agent") - if next_agent is not None: - self.agent = next_agent - explicit_hermes_home = hermes_home or kwargs.get("hermes_home") - if explicit_hermes_home or kwargs.get("config") is not None or self.hermes_home is None: - next_hermes_home = _resolve_hermes_home( - kwargs.get("config", self._host_config), - explicit_hermes_home, - ) - if next_hermes_home: - self.hermes_home = next_hermes_home - # Re-layer the profile's plugins.tracedecay block now that the host - # config and hermes_home are settled for this session. - self.config = _with_plugin_block(self._host_config, self.hermes_home) - explicit_project_root = project_root or kwargs.get("project_root") - runtime_cwd = kwargs.get("cwd") - configured_project_root = _configured_project_root(self.config) - routing_supplied = bool(explicit_project_root or runtime_cwd) - if explicit_project_root: - route_state, next_project_root = _project_scope_resolution( - explicit_project_root, - self.hermes_home, - ) - if route_state == "unregistered": - next_project_root = _code_project_root( - explicit=explicit_project_root, - hermes_home=self.hermes_home, - ) - elif runtime_cwd: - next_project_root = _resolved_project_scope(runtime_cwd, self.hermes_home) - elif self.project_root: - route_state, next_project_root = _project_scope_resolution( - self.project_root, - self.hermes_home, - ) - if route_state == "unregistered": - next_project_root = _code_project_root( - explicit=self.project_root, - hermes_home=self.hermes_home, - ) - routing_supplied = routing_supplied or next_project_root is None - elif configured_project_root: - next_project_root = _resolved_project_scope( - configured_project_root, - self.hermes_home, - ) - routing_supplied = True - elif session_id is not None and not self.project_root: - next_project_root = _resolved_project_scope( - _runtime_working_directory(), - self.hermes_home, - ) - else: - next_project_root = None - if next_project_root: - self.project_root = next_project_root - elif routing_supplied: - self.project_root = None - - def initialize(self, session_id=None, hermes_home=None, project_root=None, **kwargs): - self._bind_session(session_id, hermes_home, project_root, **kwargs) - - def on_session_start(self, session_id=None, hermes_home=None, project_root=None, **kwargs): - self._bind_session(session_id, hermes_home, project_root, **kwargs) - - def on_session_end(self, session_id=None, messages=None, **kwargs): - _join_host_receipts() - # Real session boundary: drop the per-session record so long-lived - # gateway processes do not accumulate dead conversation state. - key = self._session_key(session_id) - with self._state_lock: - self._session_states.pop(key, None) - if getattr(self._thread_sessions, "session_id", None) == key: - self._thread_sessions.session_id = None - if self.active_session_id == key: - self.active_session_id = None - - def _apply_context_length(self, context_length): - """Track a context length on the host-contract attrs. - - run_agent.py logs ``context_length``/``threshold_tokens`` directly and - the minimum-context guard in agent/agent_init.py reads - ``context_length``, leaving them 0 logged a bogus 0-token window and - silently bypassed that guard. - """ - try: - parsed = int(context_length) - except (TypeError, ValueError): - return None - if parsed <= 0: - return None - self.context_length = parsed - threshold_tokens = _configured_threshold_tokens( - self.config, - hermes_home=self.hermes_home, - context_length_override=parsed, - ) - if threshold_tokens is None: - threshold_tokens = int( - parsed * _lcm_context_threshold(self.config, hermes_home=self.hermes_home) - ) - self.threshold_tokens = threshold_tokens - return parsed - - def update_model(self, model, context_length, base_url="", api_key="", provider="", api_mode=""): - self.model = str(model or "") - self._runtime_context_length = self._apply_context_length(context_length) - - def update_from_response(self, usage): - # Required by newer Hermes ContextEngine ABCs (abstract method). - # Tracks normalized token usage; run_agent.py reads these attrs. - usage = usage or {} - - def _as_int(value): - try: - return int(value) - except (TypeError, ValueError): - return 0 - - self.last_prompt_tokens = _as_int( - usage.get("prompt_tokens") or usage.get("input_tokens") - ) - self.last_real_prompt_tokens = self.last_prompt_tokens - self.last_completion_tokens = _as_int( - usage.get("completion_tokens") or usage.get("output_tokens") - ) - self.last_total_tokens = _as_int(usage.get("total_tokens")) or ( - self.last_prompt_tokens + self.last_completion_tokens - ) - - def _effective_context_length(self): - if self._runtime_context_length is not None: - return self._runtime_context_length - if self._session_start_context_length is not None: - return self._session_start_context_length - return None - - def _tool_args(self, session_id=None): - return { - "session_id": session_id if session_id is not None else self.active_session_id - } - - def should_compress_preflight(self, messages, current_tokens=None, **kwargs): - del messages, current_tokens, kwargs - # Hermes does not expose an authentic raw-compression protocol. Its - # transcript still ingests through the daemon, but compaction is a - # typed unavailable capability. - return False - - def _preflight_probe(self, messages, current_tokens=None, **kwargs): - del messages, current_tokens, kwargs - return { - "status": "unavailable", - "reason": "host_raw_compression_unavailable", - "should_compress": False, - } - - def should_compress(self, prompt_tokens=None, **kwargs): - del prompt_tokens, kwargs - return False - - def has_content_to_compress(self, messages, current_tokens=None, **kwargs): - del current_tokens, kwargs - non_empty = [ - message - for message in (messages or []) - if str((message or {}).get("content") or "").strip() - ] - return len(non_empty) >= 2 - - def should_defer_preflight_to_real_usage(self, rough_tokens=None): - del rough_tokens - return False - - def carry_over_new_session_context(self, old_session_id, new_session_id): - old_session_id = str(old_session_id or "") - new_session_id = str(new_session_id or "") - if not new_session_id: - return - self._bind_session(new_session_id, old_session_id=old_session_id) - - def status(self, session_id=None, **kwargs): - args = self._tool_args(session_id) - args["provider"] = STANDARD_HERMES_LCM_PROVIDER - args = _lcm_store_args(args, kwargs.get("project_root") or self.project_root) - return call_tracedecay_json( - "tracedecay_lcm_status", - args, - **_project_call_kwargs(kwargs.get("project_root") or self.project_root), - ) - - def get_tool_schemas(self): - return list(LCM_NATIVE_SCHEMAS) - - def get_status(self): - last_result = self.last_compress_result - if not isinstance(last_result, dict): - last_result = {"status": "never_ran"} - return { - "engine": self.name, - "session_id": self.active_session_id, - "active_session_id": self.active_session_id, - "project_root": self.project_root, - "tracedecay_binary_path": tools.TRACEDECAY_BIN, - "tracedecay_binary_available": _tracedecay_binary_available(), - "context_engine_tool_names": sorted( - schema["name"] for schema in self.get_tool_schemas() - ), - "last_compress_result": last_result, - "awaiting_real_usage_after_compression": False, - "live_ingest": { - "registered_tool_names": sorted(_REGISTERED_TOOL_NAMES), - "context_tool_names": sorted(_CONTEXT_TOOL_NAMES), - "host_forwards_messages": _HOST_FORWARDS_MESSAGES, - "message_dependent_tools_registered": bool( - MESSAGE_DEPENDENT_TOOLS.intersection(_REGISTERED_TOOL_NAMES) - ), - "gate_reason": ( - "not_registered" - if not _REGISTERED_TOOL_NAMES and not _CONTEXT_TOOL_NAMES - else ( - "host_does_not_forward_messages" - if _HOST_FORWARDS_MESSAGES is False - else "registered" - ) - ), - }, - } - - def handle_tool_call(self, name, arguments=None, **kwargs) -> str: - tool_name, tool_args = _normalize_memory_tool_call(name, arguments) - native_name = tool_name - tracedecay_name = LCM_TOOL_ALIASES.get(native_name) - if tracedecay_name is None and native_name in LCM_DIRECT_TOOL_NAMES: - tracedecay_name = native_name - native_name = LCM_DIRECT_TO_NATIVE.get(native_name, native_name) - if tracedecay_name is None: - return tools.error_payload(f"unknown LCM tool: {tool_name}") - - preflight_kwargs = dict(kwargs) - preflight_kwargs.pop("messages", None) - - tool_args = _translate_lcm_args(native_name, dict(tool_args)) - if tool_args.get("error"): - return json.dumps({"error": tool_args["error"]}) - if tracedecay_name in LCM_PROVIDER_LOCAL_TOOL_NAMES: - tool_args.setdefault("provider", STANDARD_HERMES_LCM_PROVIDER) - if self.active_session_id: - tool_args.setdefault("session_id", self.active_session_id) - tool_args = _lcm_store_args( - tool_args, - preflight_kwargs.get("project_root") or self.project_root, - ) - - if tracedecay_name == "tracedecay_lcm_expand_query": - expand_kwargs = dict(preflight_kwargs) - agent = expand_kwargs.pop("agent", None) or self.agent - return _handle_lcm_expand_query( - tool_args, - agent=agent, - **_project_call_kwargs( - expand_kwargs.get("project_root") or self.project_root - ), - ) - return tools.call_tracedecay_tool( - tracedecay_name, +def _handle_lcm_tool_call(name, arguments=None, config=None, **kwargs) -> str: + """Dispatch retrieval using the calling Hermes session, without owning compaction.""" + tool_name, tool_args = _normalize_memory_tool_call(name, arguments) + native_name = LCM_DIRECT_TO_NATIVE.get(tool_name, tool_name) + tracedecay_name = LCM_TOOL_ALIASES.get(native_name) + if tracedecay_name is None: + return tools.error_payload(f"unknown LCM tool: {tool_name}") + + tool_args = _translate_lcm_args(native_name, dict(tool_args)) + if tool_args.get("error"): + return json.dumps({"error": tool_args["error"]}) + tool_args.setdefault("provider", STANDARD_HERMES_LCM_PROVIDER) + if kwargs.get("session_id"): + tool_args.setdefault("session_id", kwargs["session_id"]) + if ( + native_name == "lcm_grep" + and tool_args.get("scope") == "current" + and not tool_args.get("session_id") + ): + return tools.error_payload("lcm_grep current scope requires a Hermes session_id") + project_root = kwargs.get("project_root") + tool_args = _lcm_store_args(tool_args, project_root) + if tracedecay_name == "tracedecay_lcm_expand_query": + return _handle_lcm_expand_query( tool_args, - **_project_call_kwargs( - preflight_kwargs.get("project_root") or self.project_root - ), - ) - - def expand_query(self, prompt, query=None, node_ids=None, **kwargs): - kwargs = dict(kwargs) - args = self._tool_args(kwargs.pop("session_id", None)) - args["provider"] = STANDARD_HERMES_LCM_PROVIDER - args["prompt"] = prompt - if query is not None: - args["query"] = query - if node_ids is not None: - args["node_ids"] = node_ids - for key in ("max_results", "max_tokens", "context_max_tokens"): - if key in kwargs and kwargs[key] is not None: - args[key] = kwargs[key] - if "context_max_tokens" not in args: - args["context_max_tokens"] = _lcm_expansion_context_tokens(self.config) - args = _lcm_store_args(args, kwargs.get("project_root") or self.project_root) - retrieval = call_tracedecay_json( - "tracedecay_lcm_expand_query", - args, - **_project_call_kwargs(kwargs.get("project_root") or self.project_root), + config=config, + hermes_home=kwargs.get("hermes_home"), + agent=kwargs.get("agent"), + **_project_call_kwargs(project_root), ) - synthesis_kwargs = dict(kwargs) - synthesis_agent = synthesis_kwargs.pop("agent", None) or self.agent - if synthesis_kwargs.get("model") is None: - expansion_model = _lcm_expansion_model(self.config) - if expansion_model: - synthesis_kwargs["model"] = expansion_model - if ( - synthesis_kwargs.get("timeout") is None - and synthesis_kwargs.get("expansion_timeout") is None - ): - synthesis_kwargs["expansion_timeout"] = _lcm_expansion_timeout_ms(self.config) / 1000 - return _synthesize_expand_query_payload(retrieval, agent=synthesis_agent, **synthesis_kwargs) - - def compress(self, messages, current_tokens=None, focus_topic=None, **kwargs): - """Return the unchanged transcript when host compaction is unavailable.""" - del current_tokens, focus_topic, kwargs - original = list(messages or []) - reason = "host_raw_compression_unavailable" - self.last_compress_result = { - "status": "unavailable", - "reason": reason, - "semantic_error": True, - } - self._last_compress_aborted = True - self._last_summary_error = reason - return original + # Retrieval never sends the host's live messages or other execution kwargs + # to the CLI. Completed turns still ingest through the memory provider. + return tools.call_tracedecay_tool( + tracedecay_name, tool_args, **_project_call_kwargs(project_root) + ) class TracedecayMemoryProvider(MemoryProvider): provider_id = "tracedecay" @@ -2585,7 +1997,6 @@ def handle_tool_call(self, name, arguments=None, **kwargs) -> str: ) def register(ctx): - global _HOST_FORWARDS_MESSAGES context_config = getattr(ctx, "config", None) explicit_context_home = ( getattr(ctx, "hermes_home", None) or getattr(ctx, "_hermes_home", None) @@ -2604,6 +2015,7 @@ def bound(*args, **kwargs): return bound ctx.register_hook("pre_llm_call", _pre_llm_call) + ctx.register_hook("on_session_end", lambda **_kwargs: _join_host_receipts()) try: ctx.register_hook("post_tool_call", bind_hermes_home(_post_tool_call)) except Exception as exc: @@ -2635,32 +2047,14 @@ def bound(*args, **kwargs): memory_provider.hermes_home = context_hermes_home ctx.register_memory_provider(memory_provider) - context_engine = TraceDecayContextEngine( - config=context_config, - hermes_home=context_hermes_home, - ) - ctx.register_context_engine(context_engine) - - # Direct tool registration is split by capability: - # - Code-graph / memory / transcript tools register UNCONDITIONALLY. - # They work without host message forwarding, and hermes defers - # registered non-core tools through its tool-search bridge - # (tools/tool_search.py), so schema footprint is not a blocker. - # - Only the live-ingest LCM verbs whose schemas take the in-memory - # ``messages`` list (MESSAGE_DEPENDENT_TOOLS) and the context-engine - # native tool mirrors stay gated behind the message-forwarding - # capability flag, without forwarding their ingest piggyback can - # never fire (the host still mounts the native LCM tools itself via - # context_engine.get_tool_schemas()). + # TraceDecay supplies retrieval and memory. Hermes keeps its configured + # context engine (the built-in compressor by default), which owns native + # context-window pressure, compression, and conversation boundaries. register_tool = getattr(ctx, "register_tool", None) - host_forwards_messages = _host_forwards_registered_tool_messages(ctx) - _HOST_FORWARDS_MESSAGES = host_forwards_messages if callable(register_tool) else None tracedecay_is_memory_provider = _active_memory_provider(ctx) == "tracedecay" if callable(register_tool): for schema in schemas.TOOL_SCHEMAS: name = schema["name"] - if name in MESSAGE_DEPENDENT_TOOLS and not host_forwards_messages: - continue if name in MEMORY_PROVIDER_TOOLS and tracedecay_is_memory_provider: # The active memory provider already exposes this store as # fact_store/fact_feedback/memory_status, registering the @@ -2692,34 +2086,30 @@ def bound(*args, **kwargs): ) else: _REGISTERED_TOOL_NAMES.add(name) - if host_forwards_messages: - for schema in context_engine.get_tool_schemas(): - name = schema["name"] - visible_schema = _agent_visible_schema(schema) - try: - register_tool( - name=name, - toolset="context_engine", - schema=visible_schema, - handler=_make_wrapped_lcm_handler(name, context_engine), - description=schema.get("description", ""), - ) - except Exception as exc: - logger.warning( - "tracedecay LCM tool registration failed for %s; continuing with context-engine schemas: %s", - name, - exc, - ) - else: - _CONTEXT_TOOL_NAMES.add(name) - else: - logger.info( - "tracedecay LCM registered tools skipped: this Hermes host does not forward messages to registered tool handlers; transcript sync remains on the memory-provider hook" - ) + # LCM readers use stored transcripts and forwarded session identity; + # they do not require access to the live in-memory message list. + for schema in LCM_NATIVE_SCHEMAS: + name = schema["name"] + try: + register_tool( + name=name, + toolset="tracedecay", + schema=_agent_visible_schema(schema), + handler=_make_wrapped_lcm_handler( + name, config=context_config, hermes_home=context_hermes_home + ), + description=schema.get("description", ""), + ) + except Exception as exc: + logger.warning( + "tracedecay LCM tool registration failed for %s; continuing: %s", + name, + exc, + ) + else: + _REGISTERED_TOOL_NAMES.add(name) else: - logger.info( - "tracedecay direct tool registration unavailable on this Hermes host; continuing with context-engine schemas" - ) + logger.info("tracedecay direct tool registration unavailable on this Hermes host") register_skill = getattr(ctx, "register_skill", None) skills_dir = Path(__file__).parent / "skills" diff --git a/crates/tracedecay-agent-hosts/src/agents/hermes/templates/skill.md b/crates/tracedecay-agent-hosts/src/agents/hermes/templates/skill.md index 68f1b0fb79..f41570b5c2 100644 --- a/crates/tracedecay-agent-hosts/src/agents/hermes/templates/skill.md +++ b/crates/tracedecay-agent-hosts/src/agents/hermes/templates/skill.md @@ -46,7 +46,7 @@ compresses a session. 2. Hermes exposes native aliases `lcm_grep`, `lcm_load_session`, `lcm_describe`, `lcm_expand`, `lcm_expand_query`, `lcm_status`, and `lcm_doctor`. They dispatch to their matching `tracedecay_lcm_*` commands; - use the Hermes alias and its schema when the context engine offers it, or the + use the Hermes alias and its schema when the plugin offers it, or the canonical command and its schema elsewhere. Do not invent fields by mixing the two surfaces. 3. Narrow temporal evidence with `lcm_grep` / `tracedecay_lcm_grep` (default @@ -87,9 +87,10 @@ and root identity stay inside the mounted daemon authority. Never redirect a profile refresh through whichever project happens to be active, and pass the same `scope`, `session`, `source`, and `target` selectors to `status` and `cancel` as to `begin`. -Leave host context-window -preflight, compression, and boundaries to the Hermes context engine rather -than triggering them during recall. +Hermes keeps its configured context engine, including its built-in compressor +by default. That engine handles context-window preflight, compression, and +boundaries. TraceDecay supplies retrieval tools and memory-provider transcript +ingest without replacing it or triggering compression during recall. ## Storage and project identity diff --git a/crates/tracedecay/tests/hermes_suite/lcm_bridge.rs b/crates/tracedecay/tests/hermes_suite/lcm_bridge.rs index 4c8cfd3113..de49a82b08 100644 --- a/crates/tracedecay/tests/hermes_suite/lcm_bridge.rs +++ b/crates/tracedecay/tests/hermes_suite/lcm_bridge.rs @@ -292,10 +292,9 @@ class NoToolCtx: ctx = NoToolCtx() plugin.register(ctx) -assert [name for name, _ in ctx.hooks] == ["pre_llm_call", "post_tool_call"] +assert [name for name, _ in ctx.hooks] == ["pre_llm_call", "on_session_end", "post_tool_call"] assert len(ctx.memory_providers) == 1 -assert len(ctx.context_engines) == 1 -assert isinstance(ctx.context_engines[0], plugin.TraceDecayContextEngine) +assert ctx.context_engines == [], "TraceDecay must leave Hermes' compressor selected" "#, "generated plugin registration should continue when host lacks register_tool", ); @@ -332,9 +331,9 @@ ctx = RaisingToolCtx() plugin.register(ctx) assert ctx.tool_calls -assert [name for name, _ in ctx.hooks] == ["pre_llm_call", "post_tool_call"] +assert [name for name, _ in ctx.hooks] == ["pre_llm_call", "on_session_end", "post_tool_call"] assert len(ctx.memory_providers) == 1 -assert len(ctx.context_engines) == 1 +assert ctx.context_engines == [] "#, "generated plugin registration should continue when register_tool raises", ); @@ -399,15 +398,6 @@ assert receipt_event == { assert "secret.txt" not in repr(captured) assert "cat " not in repr(captured) -# The engine's session-end boundary drops per-session state without emitting -# a host receipt; the daemon owns any stop-side transcript work. -engine = plugin.TraceDecayContextEngine() -engine.initialize(session_id="session-hermes-native", project_root="/workspace/project") -assert engine.active_session_id == "session-hermes-native" -engine.on_session_end(session_id="session-hermes-native") -assert engine.active_session_id is None -assert len(captured) == 1 - class Ctx: def __init__(self): self.hooks = [] @@ -418,7 +408,10 @@ class Ctx: ctx = Ctx() plugin.register(ctx) -assert [name for name, _ in ctx.hooks] == ["pre_llm_call", "post_tool_call"] +assert [name for name, _ in ctx.hooks] == ["pre_llm_call", "on_session_end", "post_tool_call"] +# Ending a session flushes queued receipts without taking over compaction. +dict(ctx.hooks)["on_session_end"](session_id="session-hermes-native") +assert len(captured) == 1 "#, "generated Hermes callbacks must forward terminal receipt identity without content", ); @@ -499,67 +492,48 @@ finally: } #[test] -fn generated_registration_skips_tools_without_message_forwarding_capability() { +fn generated_registration_keeps_lcm_readers_without_message_forwarding() { run_generated_plugin_script( - "check_registration_capability_gate.py", + "check_generated_registration_keeps_lcm_readers_without_message_forwarding.py", r#" -class UnsafeRegisteredToolCtx: +class Ctx: context_engine_tool_handlers_receive_messages = False - def __init__(self): - self.tools = [] + self.registered = [] self.context_engines = [] - - def register_tool(self, **kwargs): - self.tools.append(kwargs["name"]) - - def register_hook(self, name, handler): - pass - - def register_memory_provider(self, provider): + def register_hook(self, *args, **kwargs): pass - + def register_tool(self, **kwargs): + self.registered.append(kwargs) def register_context_engine(self, engine): self.context_engines.append(engine) -ctx = UnsafeRegisteredToolCtx() +ctx = Ctx() plugin.register(ctx) - -# Code-graph / memory / transcript tools register even without message -# forwarding; only the live-ingest LCM verbs whose schemas carry the -# in-memory messages list (and the context-engine tool mirrors) are gated. -expected_without_messages = { - schema["name"] - for schema in plugin.schemas.TOOL_SCHEMAS - if schema["name"] not in plugin.MESSAGE_DEPENDENT_TOOLS -} -assert expected_without_messages.issubset(ctx.tools) -assert "tracedecay_lcm_compress" not in ctx.tools -assert "tracedecay_lcm_preflight" not in ctx.tools -assert "lcm_grep" not in ctx.tools -assert len(ctx.context_engines) == 1 -engine = ctx.context_engines[0] -assert engine.name == "tracedecay" -assert "lcm_grep" in {schema["name"] for schema in engine.get_tool_schemas()} +expected = {schema["name"] for schema in plugin.schemas.TOOL_SCHEMAS} +expected.update(schema["name"] for schema in plugin.LCM_NATIVE_SCHEMAS) +registered = {entry["name"] for entry in ctx.registered} +assert expected == registered +assert "tracedecay_lcm_compress" not in registered +assert "tracedecay_lcm_preflight" not in registered +assert ctx.context_engines == [] +assert all(entry["toolset"] == "tracedecay" for entry in ctx.registered) "#, - "generated plugin should gate only message-dependent tools when host does not forward messages", + "LCM readers must register independently of live-message forwarding and compaction", ); } #[test] -fn generated_context_engine_exposes_native_lcm_surface_and_dispatch() { +fn generated_lcm_tools_expose_native_surface_and_dispatch() { run_generated_plugin_script( "check_context_engine_native_surface.py", r#" import json plugin._resolved_project_scope = lambda path, *_args: path -engine = plugin.TraceDecayContextEngine() -engine.initialize(session_id="session-1", project_root="/tmp/project") - -assert engine.name == "tracedecay" - -schemas = engine.get_tool_schemas() +from functools import partial +call_tool = partial(plugin._handle_lcm_tool_call, session_id="session-1", project_root="/tmp/project") +schemas = plugin.LCM_NATIVE_SCHEMAS schemas_by_name = {schema["name"]: schema for schema in schemas} schema_names = {schema["name"] for schema in schemas} expected_native = { @@ -615,24 +589,6 @@ doctor_params = schemas_by_name["lcm_doctor"]["parameters"] assert status_params["properties"] == {} assert doctor_params["properties"] == {} -status = engine.get_status() -assert status["engine"] == "tracedecay" -assert status["session_id"] == "session-1" -assert status["active_session_id"] == "session-1" -assert "storage_scope" not in status -assert "hermes_home" not in status -assert "lcm_project_root" not in status -assert status["project_root"] == "/tmp/project" -assert status["tracedecay_binary_path"] == plugin.tools.TRACEDECAY_BIN -assert isinstance(status["tracedecay_binary_available"], bool) -assert status["context_engine_tool_names"] == sorted(schema_names) -assert status["last_compress_result"] == {"status": "never_ran"} -assert status["live_ingest"]["registered_tool_names"] == [] -assert status["live_ingest"]["context_tool_names"] == [] -assert status["live_ingest"]["host_forwards_messages"] is None -assert status["live_ingest"]["message_dependent_tools_registered"] is False -assert status["live_ingest"]["gate_reason"] == "not_registered" - calls = [] def fake_call_tracedecay_tool(name, args, **kwargs): @@ -640,7 +596,7 @@ def fake_call_tracedecay_tool(name, args, **kwargs): return json.dumps({"ok": True, "tool": name}) plugin.tools.call_tracedecay_tool = fake_call_tracedecay_tool -native_result = engine.handle_tool_call( +native_result = call_tool( "lcm_grep", { "query": "orchard", @@ -653,19 +609,19 @@ native_result = engine.handle_tool_call( }, messages=[{"role": "user", "content": "current turn"}], ) -load_result = engine.handle_tool_call( +load_result = call_tool( "lcm_load_session", {"session_id": "session-1", "max_content_chars": 123, "roles": ["user", "tool"], "time_from": 1, "time_to": 2}, messages=[{"role": "assistant", "content": "load turn"}], ) -describe_node_result = engine.handle_tool_call("lcm_describe", {"node_id": 7}) -describe_payload_result = engine.handle_tool_call("lcm_describe", {"externalized_ref": "payload_123.payload"}) -expand_result = engine.handle_tool_call( +describe_node_result = call_tool("lcm_describe", {"node_id": 7}) +describe_payload_result = call_tool("lcm_describe", {"externalized_ref": "payload_123.payload"}) +expand_result = call_tool( "lcm_expand", {"store_id": 42, "session_id": "session-foreign", "max_tokens": 77, "source_offset": 3, "source_limit": 2}, ) -direct_result = engine.handle_tool_call("tracedecay_lcm_grep", {"query": "direct", "session_scope": "all"}) -implicit_current_result = engine.handle_tool_call("lcm_grep", {"query": "implicit"}) +direct_result = call_tool("tracedecay_lcm_grep", {"query": "direct", "session_scope": "all"}) +implicit_current_result = call_tool("lcm_grep", {"query": "implicit"}) assert json.loads(native_result) == {"ok": True, "tool": "tracedecay_lcm_grep"} assert json.loads(load_result) == {"ok": True, "tool": "tracedecay_lcm_load_session"} @@ -731,84 +687,40 @@ assert "project_root" not in calls[6][1] assert calls[6][2] == {"project_root": "/tmp/project"} assert calls[6][1]["session_id"] == "session-1" "#, - "generated context engine should expose Hermes-style native LCM surface", + "generated tools should expose Hermes-style native LCM retrieval", ); } #[test] -fn generated_context_engine_never_uses_hermes_home_as_storage_identity() { +fn generated_lcm_tools_never_use_hermes_home_as_storage_identity() { run_generated_plugin_script( - "check_context_engine_env_home.py", + "check_generated_lcm_tools_never_use_hermes_home_as_storage_identity.py", r#" -import importlib.machinery -import importlib.util import json import os -import pathlib -import sys - -plugin_dir = pathlib.Path(sys.argv[1]) -parent_name = "_hermes_user_env_home" -parent_spec = importlib.machinery.ModuleSpec(parent_name, None, is_package=True) -parent_spec.submodule_search_locations = [] -parent_module = importlib.util.module_from_spec(parent_spec) -sys.modules[parent_name] = parent_module - -module_name = f"{parent_name}.tracedecay" -spec = importlib.util.spec_from_file_location( - module_name, - plugin_dir / "__init__.py", - submodule_search_locations=[str(plugin_dir)], -) -plugin = importlib.util.module_from_spec(spec) -sys.modules[module_name] = plugin -spec.loader.exec_module(plugin) os.environ["HERMES_HOME"] = "/tmp/hermes-from-env" - calls = [] - -def fake_call_tracedecay_tool(name, args, **kwargs): +def fake_call(name, args, **kwargs): calls.append((name, args, kwargs)) - return json.dumps({"content": [{"type": "text", "text": json.dumps({"status": "ok"})}]}) - -plugin.tools.call_tracedecay_tool = fake_call_tracedecay_tool -plugin._resolved_project_scope = lambda path, *_args: ( - str(path) if path and str(path) == "/tmp/project" else None -) - -engine = plugin.TraceDecayContextEngine() -engine.initialize(session_id="session-1") -assert os.path.normcase(os.path.realpath(engine.hermes_home)) == os.path.normcase( - os.path.realpath(str(plugin_dir.parent.parent)) -) -status = engine.get_status() -assert "storage_scope" not in status -assert "hermes_home" not in status -assert "lcm_project_root" not in status -assert status["project_root"] is None - -engine.handle_tool_call( - "lcm_grep", - {"query": "orchard"}, - messages=[{"role": "user", "content": "profile current turn"}], -) - -assert len(calls) == 1 -assert calls[0][0] == "tracedecay_lcm_grep" -assert "project_root" not in calls[0][1] -assert calls[0][1]["storage_scope"] == "user" -assert calls[0][2] == {} -assert "messages" not in calls[0][1] - + return "{}" +plugin.tools.call_tracedecay_tool = fake_call +plugin._project_scope_resolution = lambda *_args: ("rejected", None) +handler = plugin._make_wrapped_lcm_handler("lcm_grep", hermes_home="/tmp/hermes-from-config") +handler({"query": "orchard"}, session_id="session-1", cwd="/tmp/hermes-from-config", + messages=[{"role": "user", "content": "profile current turn"}]) +assert calls == [("tracedecay_lcm_grep", { + "query": "orchard", "scope": "current", "provider": "hermes", + "sort": "relevance", "include_summaries": False, "temporal_mode": {"kind": "current"}, + "session_id": "session-1", "storage_scope": "user", +}, {})], calls os.environ["HERMES_HOME"] = "/tmp/another-hermes-home" -other = plugin.TraceDecayContextEngine() -other.initialize(session_id="session-2") -other.status() +handler({"query": "orchard"}, session_id="session-2") +assert calls[-1][1]["session_id"] == "session-2" assert calls[-1][1]["storage_scope"] == "user" assert calls[-1][2] == {} "#, - "generated context engine must not use HERMES_HOME as a TraceDecay storage identity", + "LCM readers must not use Hermes home as a TraceDecay storage identity", ); } @@ -861,173 +773,67 @@ assert args == {"format": "json"} } #[test] -fn generated_context_engine_resolves_configured_hermes_home_on_registration() { +fn generated_registration_resolves_configured_hermes_home() { run_generated_plugin_script( - "check_context_engine_config_home.py", + "check_generated_registration_resolves_configured_hermes_home.py", r#" -import importlib.machinery -import importlib.util -import pathlib -import sys - -plugin_dir = pathlib.Path(sys.argv[1]) -parent_name = "_hermes_user_config_home" -parent_spec = importlib.machinery.ModuleSpec(parent_name, None, is_package=True) -parent_spec.submodule_search_locations = [] -parent_module = importlib.util.module_from_spec(parent_spec) -sys.modules[parent_name] = parent_module - -module_name = f"{parent_name}.tracedecay" -spec = importlib.util.spec_from_file_location( - module_name, - plugin_dir / "__init__.py", - submodule_search_locations=[str(plugin_dir)], -) -plugin = importlib.util.module_from_spec(spec) -sys.modules[module_name] = plugin -spec.loader.exec_module(plugin) - class Ctx: - def __init__(self): - self.config = {"hermes_home": "/tmp/hermes-from-config"} - self.context_engines = [] + config = {"hermes_home": "/tmp/hermes-from-config"} def register_hook(self, name, handler): pass - def register_context_engine(self, engine): - self.context_engines.append(engine) + def register_memory_provider(self, provider): + self.provider = provider ctx = Ctx() plugin.register(ctx) - -assert len(ctx.context_engines) == 1 -engine = ctx.context_engines[0] -assert engine.name == "tracedecay" -assert engine.hermes_home == "/tmp/hermes-from-config" +ctx.provider.initialize(session_id="session-1") +assert ctx.provider.hermes_home == "/tmp/hermes-from-config" "#, - "generated registration should resolve configured hermes_home", + "generated registration should resolve configured hermes_home without a context engine", ); } #[test] -fn generated_context_engine_registers() { - run_python_check( - "check_context_engine.py", +fn generated_registration_preserves_native_compaction_and_lcm_tools() { + run_generated_plugin_script( + "check_generated_registration_preserves_native_compaction_and_lcm_tools.py", r#" -import importlib.machinery -import importlib.util -import os -import pathlib -import sys -import types - -plugin_dir = pathlib.Path(sys.argv[1]) - -parent_name = "_hermes_user_context" -parent_spec = importlib.machinery.ModuleSpec(parent_name, None, is_package=True) -parent_spec.submodule_search_locations = [] -parent_module = importlib.util.module_from_spec(parent_spec) -sys.modules[parent_name] = parent_module - -class ContextEngine: - pass - -agent_module = types.ModuleType("agent") -agent_module.__path__ = [] -context_engine_module = types.ModuleType("agent.context_engine") -context_engine_module.ContextEngine = ContextEngine -agent_module.context_engine = context_engine_module -sys.modules["agent"] = agent_module -sys.modules["agent.context_engine"] = context_engine_module - -module_name = f"{parent_name}.tracedecay" -spec = importlib.util.spec_from_file_location( - module_name, - plugin_dir / "__init__.py", - submodule_search_locations=[str(plugin_dir)], -) -plugin = importlib.util.module_from_spec(spec) -sys.modules[module_name] = plugin -spec.loader.exec_module(plugin) - -class FullCtx: +import json +class Ctx: + # A retrieval plugin must load on a host that exposes no context-engine API. + config = {"context": {"engine": "compressor"}, "memory": {"provider": "tracedecay"}} def __init__(self): - self.tools = [] - self.hooks = [] - self.context_engines = [] - - def register_tool(self, **kwargs): - self.tools.append(kwargs) - + self.tools = {} def register_hook(self, name, handler): - self.hooks.append((name, handler)) - - def register_context_engine(self, engine): - self.context_engines.append(engine) + pass + def register_tool(self, **kwargs): + self.tools[kwargs["name"]] = kwargs["handler"] + def register_memory_provider(self, provider): + self.provider = provider -ctx = FullCtx() +ctx = Ctx() plugin.register(ctx) -assert len(ctx.context_engines) == 1 -engine = ctx.context_engines[0] -assert isinstance(engine, plugin.TraceDecayContextEngine) -assert isinstance(engine, ContextEngine) - -plugin._resolved_project_scope = lambda path, *_args: path -engine.initialize( - session_id="session-123", - hermes_home="/tmp/hermes-profile", - project_root="/tmp/project", -) -assert engine.active_session_id == "session-123" -assert engine.hermes_home == "/tmp/hermes-profile" -assert engine.project_root == "/tmp/project" - -# Project routing is transport-only and never derived from Hermes home. -assert not hasattr(plugin, "_storage_args") -assert plugin._project_call_kwargs("/tmp/project") == { - "project_root": "/tmp/project", -} - +assert ctx.config["context"]["engine"] == "compressor" +assert ctx.provider.name == "tracedecay" +assert set(plugin.LCM_TOOL_ALIASES).issubset(ctx.tools) calls = [] - -def fake_call_tracedecay_tool(name, args, **kwargs): - calls.append((name, args, kwargs)) - return "{}" - -plugin.tools.call_tracedecay_tool = fake_call_tracedecay_tool -plugin._resolved_project_scope = lambda path, *_args: ( - str(path) if path and str(path) == "/tmp/project" else None -) - -profile_engine = plugin.TraceDecayContextEngine() -profile_engine.on_session_start(session_id="session-1", hermes_home="/tmp/hermes") -assert profile_engine.should_compress_preflight(messages=[], current_tokens=123) is False - -project_engine = plugin.TraceDecayContextEngine() -project_engine.on_session_start( - session_id="session-2", - hermes_home="/tmp/hermes", - project_root="/tmp/project", +plugin._project_scope_resolution = lambda path, *_: ("registered", path) +plugin.tools.call_tracedecay_tool = lambda name, args, **kwargs: ( + calls.append((name, dict(args), kwargs)) or "{}" ) -assert project_engine.should_compress_preflight(messages=[], current_tokens=456) is False - -project_engine = plugin.TraceDecayContextEngine() -project_engine.initialize(session_id="initial", project_root="/tmp/project") -project_engine.on_session_start(session_id="next", project_root="/tmp/project") -assert project_engine.should_compress_preflight(messages=[], current_tokens=789) is False - -profile_engine = plugin.TraceDecayContextEngine() -profile_engine.initialize(session_id="initial", hermes_home="/tmp/hermes") -profile_engine.on_session_start(session_id="next") -assert profile_engine.should_compress_preflight(messages=[], current_tokens=321) is False -assert calls == [] - +ctx.tools["lcm_grep"]({"query": "orchard"}, session_id="session-123", project_root="/tmp/project") +assert calls == [("tracedecay_lcm_grep", { + "query": "orchard", "scope": "current", "provider": "hermes", "session_id": "session-123", + "sort": "relevance", "include_summaries": False, "temporal_mode": {"kind": "current"}, +}, {"project_root": "/tmp/project"})], calls +assert not hasattr(plugin, "_storage_args") "#, - "generated plugin should register a Hermes context engine", + "generated registration must preserve native compaction while exposing LCM retrieval", ); } #[test] -fn context_engine_expand_query_synthesizes_and_degrades() { +fn lcm_expand_query_synthesizes_and_degrades() { run_python_check( "check_expand_query_synthesis.py", r#" @@ -1108,11 +914,12 @@ class Aux: return "hiddenFinal answer from context" agent = type("Agent", (), {"auxiliary_client": Aux()})() -engine = plugin.TraceDecayContextEngine() -engine.initialize(session_id="session-1", project_root="/tmp/project", agent=agent) +def expand_query(**args): + args["session_id"] = "session-1" + return json.loads(plugin._handle_lcm_expand_query(args, project_root="/tmp/project", agent=agent)) responses.append(needs_synthesis()) -answer = engine.expand_query(prompt="What changed?", query="orchard") +answer = expand_query(prompt="What changed?", query="orchard") assert answer["status"] == "ok" assert answer["needs_synthesis"] is False assert answer["answer"] == "Final answer from context" @@ -1127,7 +934,7 @@ assert "EXPANDED CONTEXT" in agent.auxiliary_client.calls[0]["messages"][1]["con agent.auxiliary_client.mode = "timeout" responses.append(needs_synthesis()) -timeout_payload = engine.expand_query(prompt="What changed?", query="orchard") +timeout_payload = expand_query(prompt="What changed?", query="orchard") assert timeout_payload["degraded"] is True assert "timed out" in timeout_payload["error"] assert timeout_payload["timeout_seconds"] > 0 @@ -1135,7 +942,7 @@ assert timeout_payload["needs_synthesis"] is False agent.auxiliary_client.mode = "empty" responses.append(needs_synthesis()) -empty_payload = engine.expand_query(prompt="What changed?", query="orchard") +empty_payload = expand_query(prompt="What changed?", query="orchard") assert empty_payload["degraded"] is True assert "empty answer" in empty_payload["error"] assert empty_payload["needs_synthesis"] is False @@ -1145,13 +952,13 @@ assert empty_payload["needs_synthesis"] is False # exception that loses the retrieval behind a generic registry error. agent.auxiliary_client.mode = "unexpected" responses.append(needs_synthesis()) -failed_payload = engine.expand_query(prompt="What changed?", query="orchard") +failed_payload = expand_query(prompt="What changed?", query="orchard") assert failed_payload["degraded"] is True assert "schema bug" in failed_payload["error"] assert failed_payload["needs_synthesis"] is False assert failed_payload["matches"], "retrieval must survive synthesis failures" "#, - "generated context engine should synthesize and degrade expand-query answers", + "generated retrieval should synthesize and degrade expand-query answers", ); } @@ -1295,7 +1102,7 @@ assert provider.project_root is None } #[test] -fn context_engine_records_all_forwarded_native_lcm_tools() { +fn registration_records_all_native_lcm_tools() { run_generated_plugin_script( "check_context_tool_names_bookkeeping.py", r#" @@ -1324,10 +1131,10 @@ ctx = Ctx() plugin.register(ctx) expected = sorted(schema["name"] for schema in plugin.LCM_NATIVE_SCHEMAS) -assert sorted(plugin._CONTEXT_TOOL_NAMES) == expected +assert set(expected).issubset(plugin._REGISTERED_TOOL_NAMES) assert set(expected).issubset(set(ctx.registered)) "#, - "generated plugin should record every forwarded native LCM context tool", + "generated plugin should record every native LCM retrieval tool", ); } @@ -1397,7 +1204,7 @@ assert "cursor" not in raw } #[test] -fn context_engine_lcm_expand_query_tolerates_forwarded_agent_kwarg() { +fn lcm_expand_query_tolerates_forwarded_agent_kwarg() { run_generated_plugin_script( "check_expand_query_forwarded_agent.py", r#" @@ -1419,24 +1226,23 @@ class Aux: raise AssertionError("synthesis should not run") forwarded_agent = type("ForwardedAgent", (), {"auxiliary_client": Aux()})() -engine = plugin.TraceDecayContextEngine() -engine.initialize(session_id="session-1", project_root="/tmp/project", agent=forwarded_agent) - -raw = engine.handle_tool_call( +raw = plugin._handle_lcm_tool_call( "lcm_expand_query", {"prompt": "What changed?"}, + session_id="session-1", + project_root="/tmp/project", agent=forwarded_agent, ) payload = json.loads(raw) assert payload["status"] == "ok" assert payload["answer"] == "retrieval-only answer" "#, - "generated context engine should not pass duplicate agent kwargs through lcm_expand_query", + "generated retrieval must not pass duplicate agent kwargs through lcm_expand_query", ); } #[test] -fn context_engine_expand_query_uses_expansion_model_context_and_timeout_knobs() { +fn lcm_expand_query_uses_expansion_model_context_and_timeout_knobs() { run_generated_plugin_script( "check_expand_query_expansion_knobs.py", r#" @@ -1501,16 +1307,15 @@ os.environ["LCM_EXPANSION_CONTEXT_TOKENS"] = "4321" os.environ["LCM_EXPANSION_TIMEOUT_MS"] = "9000" agent = FakeAgent() -engine = plugin.TraceDecayContextEngine( +result = json.loads(plugin._handle_lcm_expand_query( + {"prompt": "What changed?", "query": "orchard", "session_id": "session-1"}, + project_root="/tmp/project", agent=agent, config={ "expansion_model": "cfg-expansion-model", "expansion_context_tokens": 32000, "expansion_timeout_ms": 120000, - } -) -engine.initialize(session_id="session-1", project_root="/tmp/project", agent=agent) - -result = engine.expand_query(prompt="What changed?", query="orchard") + }, +)) assert result["status"] == "ok" assert result["needs_synthesis"] is False assert result["answer"] == "synthetic answer" @@ -1702,89 +1507,6 @@ assert tools.call_tracedecay_tool("tracedecay_lcm_status", {}) == "{}" ); } -/// Newer Hermes declares `ContextEngine.update_from_response(usage)` as an -/// abstract method; the generated engine must implement it or the plugin -/// fails to load with "Can't instantiate abstract class". -#[test] -fn generated_context_engine_satisfies_abstract_update_from_response() { - run_python_check( - "check_update_from_response.py", - r#" -import abc -import importlib.machinery -import importlib.util -import pathlib -import sys -import types - -plugin_dir = pathlib.Path(sys.argv[1]) - -# Mimic the newer Hermes ABC *before* the plugin module is executed. -class ContextEngine(abc.ABC): - @abc.abstractmethod - def update_from_response(self, usage): - raise NotImplementedError - -agent_module = types.ModuleType("agent") -agent_module.__path__ = [] -context_engine_module = types.ModuleType("agent.context_engine") -context_engine_module.ContextEngine = ContextEngine -agent_module.context_engine = context_engine_module -sys.modules["agent"] = agent_module -sys.modules["agent.context_engine"] = context_engine_module - -parent_name = "_hermes_user_abc" -parent_spec = importlib.machinery.ModuleSpec(parent_name, None, is_package=True) -parent_spec.submodule_search_locations = [] -parent_module = importlib.util.module_from_spec(parent_spec) -sys.modules[parent_name] = parent_module - -module_name = f"{parent_name}.tracedecay" -spec = importlib.util.spec_from_file_location( - module_name, - plugin_dir / "__init__.py", - submodule_search_locations=[str(plugin_dir)], -) -plugin = importlib.util.module_from_spec(spec) -sys.modules[module_name] = plugin -spec.loader.exec_module(plugin) - -class Ctx: - def __init__(self): - self.context_engines = [] - def register_hook(self, name, handler): - pass - def register_context_engine(self, engine): - self.context_engines.append(engine) - -ctx = Ctx() -# This instantiates TraceDecayContextEngine; an unimplemented abstract method -# raises TypeError here. -plugin.register(ctx) -assert len(ctx.context_engines) == 1 -engine = ctx.context_engines[0] - -engine.update_from_response({"prompt_tokens": 11, "completion_tokens": 7}) -assert engine.last_prompt_tokens == 11 -assert engine.last_completion_tokens == 7 -assert engine.last_total_tokens == 18 - -engine.update_from_response( - {"input_tokens": "3", "output_tokens": "4", "total_tokens": 9} -) -assert engine.last_prompt_tokens == 3 -assert engine.last_completion_tokens == 4 -assert engine.last_total_tokens == 9 - -engine.update_from_response(None) -assert engine.last_prompt_tokens == 0 -assert engine.last_completion_tokens == 0 -assert engine.last_total_tokens == 0 -"#, - "generated engine must satisfy the abstract update_from_response contract", - ); -} - /// Newer Hermes derives the skill namespace from the plugin name and rejects /// ':' inside skill names, so registration must use the bare "tracedecay". #[test] @@ -1926,46 +1648,45 @@ assert "tracedecay:tracedecay" in text, text /// Host runtime config may identify a project, but a real session cwd or /// explicit call wins. The plugin-owned config block is filtered separately. #[test] -fn generated_context_engine_prefers_runtime_project_context() { +fn generated_lcm_tools_scope_each_call_to_its_runtime_session() { run_generated_plugin_script( - "check_context_engine_project_root.py", + "check_generated_lcm_tools_scope_each_call_to_its_runtime_session.py", r#" +import json +from concurrent.futures import ThreadPoolExecutor class Ctx: + config = {"project_root": "/tmp/stale-pinned-project"} def __init__(self): - self.config = {"project_root": "/tmp/pinned-project"} - self.context_engines = [] + self.tools = {} def register_hook(self, name, handler): pass - def register_context_engine(self, engine): - self.context_engines.append(engine) + def register_tool(self, **kwargs): + self.tools[kwargs["name"]] = kwargs["handler"] ctx = Ctx() -plugin._resolved_project_scope = lambda path, *_args: path plugin.register(ctx) -engine = ctx.context_engines[0] - -# Host runtime config applies at registration time. -assert engine.project_root == "/tmp/pinned-project", engine.project_root - -# A real session cwd overrides stale host config. -engine.on_session_start(session_id="s1", cwd="/somewhere/else") -assert engine.project_root == "/somewhere/else", engine.project_root - -# An explicit project_root remains highest priority. -engine.on_session_start(session_id="s2", project_root="/explicit/root") -assert engine.project_root == "/explicit/root", engine.project_root - -# Returning to a prior session restores its route instead of inheriting the -# most recently active session or the host default. -engine.on_session_start(session_id="s1") -assert engine.project_root == "/somewhere/else", engine.project_root - -# Without explicit runtime context, cwd is the fallback. -unpinned = plugin.TraceDecayContextEngine(config={}) -assert unpinned.project_root is None -unpinned.on_session_start(session_id="s3", cwd="/cwd/fallback") -assert unpinned.project_root == "/cwd/fallback", unpinned.project_root +plugin._project_scope_resolution = lambda path, *_: ("registered", path) +plugin.tools.call_tracedecay_tool = lambda name, args, **kwargs: json.dumps({ + "tool": name, "args": args, "route": kwargs, +}) +handler = ctx.tools["lcm_grep"] +def recall(index): + sid = "session-" + str(index) + root = "/tmp/project-" + str(index) + result = json.loads(handler({"query": "q"}, session_id=sid, cwd=root)) + assert result["args"]["session_id"] == sid + assert result["route"] == {"project_root": root} +with ThreadPoolExecutor(max_workers=4) as pool: + list(pool.map(recall, range(16))) +# Explicit selectors and runtime identity are resolved for every call; no +# preceding conversation's session or project can become a fallback. +result = json.loads(handler({"query": "q", "session_id": "explicit"}, + session_id="runtime", cwd="/tmp/cwd", project_root="/tmp/explicit")) +assert result["args"]["session_id"] == "explicit" +assert result["route"] == {"project_root": "/tmp/explicit"} +missing = json.loads(handler({"query": "q"}, cwd="/tmp/other")) +assert "session_id" in missing["error"] "#, - "generated engine must prefer explicit project and real cwd over host config", + "LCM retrieval must preserve runtime project and session scope across concurrent calls", ); } diff --git a/scripts/hermes_plugin_unit_check.py b/scripts/hermes_plugin_unit_check.py index 6b09936b97..b5912b2062 100644 --- a/scripts/hermes_plugin_unit_check.py +++ b/scripts/hermes_plugin_unit_check.py @@ -6,12 +6,12 @@ plugin context and a fake tracedecay binary, so it runs anywhere Python 3 exists. It asserts the host contracts the 2026-06 review found broken: - 1. compress() returns a MESSAGE LIST on success / no-op / error - (hermes ContextEngine ABC), never the raw result dict. + 1. TraceDecay leaves the native context engine selected and registers + LCM retrieval tools independently, scoped to each calling session. 2. Payloads >128 KiB round-trip through `--args @file` and the spill tempfile is cleaned up. 3. Code-graph tools register WITHOUT the message-forwarding capability - flag; the messages-dependent LCM verbs stay gated. + flag; LCM readers use stored transcripts and forwarded session identity. 4. The memory provider implements sync_turn / prefetch / on_memory_write and calls the right subprocess verbs. 5. pre_llm_call returns None on non-first turns and on first-turn greetings @@ -30,7 +30,6 @@ (default: bazel-bin/crates/tracedecay-cli/tracedecay). """ -import copy import json import os import shutil @@ -330,9 +329,6 @@ def raise_resolution(*_args, **_kwargs): ) is None finally: plugin.call_tracedecay_json = real_json - home_engine = plugin.TraceDecayContextEngine(hermes_home=str(host_home)) - home_engine.on_session_start(session_id="home-scope", cwd=str(host_home)) - assert home_engine.project_root is None home_provider = plugin.TracedecayMemoryProvider() home_provider.initialize( session_id="home-scope", hermes_home=str(host_home), cwd=str(host_home) @@ -351,10 +347,8 @@ def _check_registration_split( # duplicates; transcript search has no provider twin and stays. ctx = StubCtx() plugin.register(ctx) - assert ctx.engine.hermes_home == str(host_home), ( - ctx.engine.hermes_home, - str(host_home), - ) + assert ctx.engine is None, "TraceDecay must not replace Hermes' native compressor" + assert ctx.provider.hermes_home == str(host_home) assert ctx.skills == { "managed-test": managed_skill, "tracedecay": plugin_dir / "skills" / "tracedecay" / "SKILL.md", @@ -377,9 +371,7 @@ def _check_registration_split( custom_ctx = StubCtx() custom_ctx.hermes_home = str(custom_home) plugin.register(custom_ctx) - assert custom_ctx.engine.hermes_home == str(custom_home) - custom_ctx.engine.on_session_start(session_id="custom-home", cwd=str(custom_home)) - assert custom_ctx.engine.project_root is None + assert custom_ctx.engine is None custom_ctx.provider.initialize(session_id="custom-home", cwd=str(custom_home)) assert custom_ctx.provider.hermes_home == str(custom_home) assert custom_ctx.provider.project_root is None @@ -433,8 +425,8 @@ def _check_registration_split( ok("registered tools and providers honor a custom Hermes home") assert "tracedecay_lcm_compress" not in ctx.tools, sorted(ctx.tools) assert "tracedecay_lcm_preflight" not in ctx.tools, sorted(ctx.tools) - # Context-engine native mirrors stay gated without the capability flag. - assert "lcm_grep" not in ctx.tools, sorted(ctx.tools) + # Readers work through the ordinary registry without taking over compaction. + assert set(plugin.LCM_TOOL_ALIASES).issubset(ctx.tools), sorted(ctx.tools) assert ctx.provider.get_config_schema() == [], ctx.provider.get_config_schema() schemas = json.dumps([entry["schema"] for entry in ctx.tools.values()]) assert "storage_scope" not in schemas and "hermes_home" not in schemas, schemas @@ -459,7 +451,7 @@ class ForwardingCtx(StubCtx): plugin.register(fwd) assert "tracedecay_lcm_compress" not in fwd.tools, sorted(fwd.tools) assert "lcm_grep" in fwd.tools, sorted(fwd.tools) - ok("read-only LCM mirrors register when the host forwards messages") + ok("LCM readers register regardless of live-message forwarding") return ctx @@ -572,7 +564,7 @@ def _check_canonical_metadata(plugin, plugin_dir: Path, ctx): assert plugin.READ_ONLY_TOOL_NAMES.isdisjoint( {"tracedecay_str_replace", "tracedecay_fact_store_add", "tracedecay_fact_store_curate"} ), sorted(plugin.READ_ONLY_TOOL_NAMES) - assert set(plugin.FACT_STORE_EXACT_ROUTES) >= set(plugin.MEMORY_FACT_ACTIONS.values()) + assert {"add", "search"} <= plugin.FACT_STORE_EXACT_ROUTES.keys() assert all( route == f"tracedecay_fact_store_{action}" and route in plugin.TOOL_SCHEMAS_BY_NAME for action, route in plugin.FACT_STORE_EXACT_ROUTES.items() @@ -701,151 +693,130 @@ def _scripted(name, args, **kwargs): plugin.tools.call_tracedecay_tool = real_tool -def _check_engine_state_and_compress( - plugin, ctx, host_home: Path, runtime_project: Path -) -> list: - # ── 1. compress() message-list contract ────────────────────────────── - engine = ctx.engine - assert engine is not None - engine.initialize(session_id="check-session", hermes_home=str(host_home)) - engine.project_root = str(runtime_project) - engine.context_length = 200_000 - engine.threshold_tokens = 150_000 - engine.agent = object() - cloned_engine = copy.deepcopy(engine) - assert cloned_engine is not engine - assert cloned_engine._state_lock is not engine._state_lock - assert cloned_engine.project_root == str(runtime_project) - assert cloned_engine.context_length == 200_000 - assert cloned_engine.threshold_tokens == 150_000 - assert cloned_engine.agent is None - cloned_engine.context_length = 100_000 - assert engine.context_length == 200_000 - ok("context engine safely deep-copies per-agent budget state") - - real_resolver = plugin._resolved_project_scope +def _check_lcm_retrieval_scope(plugin, ctx, host_home: Path, runtime_project: Path): + assert ctx.engine is None, "TraceDecay must leave native compaction enabled" + calls = [] + real_tool = plugin.tools.call_tracedecay_tool + real_resolution = plugin._project_scope_resolution try: - plugin._resolved_project_scope = lambda path, *_args: ( - str(runtime_project) - if path - and os.path.realpath(str(path)) == os.path.realpath(str(runtime_project)) - else None + plugin._project_scope_resolution = lambda path, *_args: ( + ("rejected", None) if str(path) == str(host_home) else ("registered", str(path)) ) - engine.initialize(session_id="project-session", project_root=str(runtime_project)) - assert engine.project_root == str(runtime_project) - engine.initialize(session_id="untethered-session", cwd=str(host_home)) - assert engine.project_root is None - engine.initialize(session_id="project-session") - assert engine.project_root == str(runtime_project) - engine.initialize(session_id="untethered-session") - assert engine.project_root is None - finally: - plugin._resolved_project_scope = real_resolver - ok("context engine isolates project routing per Hermes session") - - status_calls = [] - real_json = plugin.call_tracedecay_json - try: - plugin.call_tracedecay_json = lambda name, args, **kwargs: ( - status_calls.append((name, args, kwargs)) - or {"status": "not_ingested"} + plugin.tools.call_tracedecay_tool = lambda name, args, **kwargs: ( + calls.append((name, args, kwargs)) or "{}" ) - engine.active_session_id = "status-session" - engine.project_root = str(runtime_project) - assert engine.status() == {"status": "not_ingested"} - finally: - plugin.call_tracedecay_json = real_json - assert status_calls == [ - ( + handler = ctx.tools["lcm_grep"]["handler"] + handler({"query": "orchard"}, session_id="project-session", cwd=str(runtime_project)) + handler({"query": "orchard"}, session_id="chat-session", cwd=str(host_home)) + handler({"query": "orchard"}, session_id="project-session", cwd=str(runtime_project)) + assert [args["session_id"] for _, args, _ in calls] == [ + "project-session", "chat-session", "project-session", + ] + assert calls[0][2] == calls[2][2] == {"project_root": str(runtime_project)} + assert calls[1][2] == {} + assert calls[1][1]["storage_scope"] == "user" + missing = json.loads(handler({"query": "orchard"}, cwd=str(runtime_project))) + assert "session_id" in missing["error"] + assert len(calls) == 3, "missing session identity must never reuse a previous session" + ctx.tools["lcm_status"]["handler"]({}, session_id="status-session", cwd=str(runtime_project)) + assert calls[-1] == ( "tracedecay_lcm_status", {"provider": "hermes", "session_id": "status-session"}, {"project_root": str(runtime_project)}, ) - ], status_calls - ok("context-engine status binds its Hermes provider and session target") - - messages = [ + finally: + plugin.tools.call_tracedecay_tool = real_tool + plugin._project_scope_resolution = real_resolution + ok("retrieval tools retain per-call session/project scope without replacing compaction") + return [ {"role": "user", "content": "hello"}, {"role": "assistant", "content": "hi"}, ] - out = engine.compress(list(messages), current_tokens=10) - assert out == messages, out - assert engine._last_compress_aborted is True - assert engine.last_compress_result == { - "status": "unavailable", - "reason": "host_raw_compression_unavailable", - "semantic_error": True, + + +def _check_lcm_handler_configuration(plugin, host_home: Path, runtime_project: Path): + ctx = StubCtx() + ctx.hermes_home = str(host_home) + ctx.config = { + "expansion_model": "registered-model", + "expansion_context_tokens": 2048, + "expansion_timeout_ms": 3000, } - ok("compress() returns the unchanged transcript with typed unavailability") - - # Host-contract token attrs (minimum-context guard reads these). - engine.update_model("check-model", 128000) - assert engine.context_length == 128000 - assert 0 < engine.threshold_tokens <= 128000 - ok("update_model populates context_length/threshold_tokens") - return messages - - -def _check_engine_thread_isolation(engine): - # Per-session isolation: the singleton engine must not leak one - # conversation's state into another (gateway sessions, delegate_task - # children all share this instance). - import threading as _threading - - seen = {} - - def _other_session(): - engine.on_session_start(session_id="check-session-b") - engine.update_model("other-model", 64000) - engine.update_from_response({"prompt_tokens": 7, "completion_tokens": 3}) - seen["b_context"] = engine.context_length - seen["b_total"] = engine.last_total_tokens - - worker = _threading.Thread(target=_other_session) - worker.start() - worker.join(timeout=10) - assert seen == {"b_context": 64000, "b_total": 10}, seen - # The original thread stays bound to its own session. - assert engine.context_length == 128000, engine.context_length - assert engine.model == "check-model", engine.model - assert engine._last_compress_aborted is True # from the exception check above - ok("per-session engine state stays isolated across threads") - - -def _check_should_compress_gating(plugin, engine): - # Hermes exposes no authenticated raw-compaction payload, so no pressure - # probe can claim that compression is available. - probes = [] - real_probe = plugin.TraceDecayContextEngine._preflight_probe - try: - def _spy_probe(self, *args, **kwargs): - probes.append(args) - return {"status": "ok", "should_compress": True} - - plugin.TraceDecayContextEngine._preflight_probe = _spy_probe - assert engine.should_compress(prompt_tokens=10) is False - assert probes == [], probes - assert engine.should_compress(prompt_tokens=engine.threshold_tokens) is False - assert probes == [], probes - finally: - plugin.TraceDecayContextEngine._preflight_probe = real_probe - ok("should_compress stays false without an authentic host payload") + plugin.register(ctx) + handler = ctx.tools["lcm_expand_query"]["handler"] + requests, config_homes, syntheses = [], [], [] + real_tool = plugin.tools.call_tracedecay_tool + real_block = plugin.tools.plugin_config_block + env_keys = ("LCM_EXPANSION_MODEL", "LCM_EXPANSION_CONTEXT_TOKENS", "LCM_EXPANSION_TIMEOUT_MS") + saved_env = {key: os.environ.pop(key, None) for key in env_keys} + + class Client: + def call_llm(self, **kwargs): + syntheses.append(kwargs) + return "grounded answer" + + class Agent: + auxiliary_client = Client() - # ABC contract: should_compress_preflight returns a BOOL; an error dict - # from the probe must read as False, never truthy. - real_probe = plugin.TraceDecayContextEngine._preflight_probe + def retrieve(name, args, **kwargs): + requests.append((name, dict(args), kwargs)) + payload = {"needs_synthesis": True, "prompt": args["prompt"], "context_blocks": []} + return json.dumps({"content": [{"type": "text", "text": json.dumps(payload)}]}) + + def profile_config(home): + config_homes.append(home) + return { + "expansion_model": "profile-model", + "expansion_context_tokens": 1024, + "expansion_timeout_ms": 2000, + } + + override = { + "expansion_model": "call-model", + "expansion_context_tokens": 4096, + "expansion_timeout_ms": 6000, + } + args = {"prompt": "What did we decide?"} try: - plugin.TraceDecayContextEngine._preflight_probe = ( - lambda self, *a, **k: {"error": "boom"} - ) - assert engine.should_compress_preflight([{"role": "user", "content": "x"}]) is False - plugin.TraceDecayContextEngine._preflight_probe = ( - lambda self, *a, **k: {"status": "ok", "should_compress": True} - ) - assert engine.should_compress_preflight([]) is False + plugin.tools.call_tracedecay_tool = retrieve + plugin.tools.plugin_config_block = profile_config + for index, (kwargs, model, budget, timeout) in enumerate(( + ({}, "registered-model", 2048, 3), + ({"config": override}, "call-model", 4096, 6), + ({"config": None}, "profile-model", 1024, 2), + ({"config": {}}, "profile-model", 1024, 2), + ({}, "registered-model", 2048, 3), + )): + session = f"configuration-session-{index}" + result = json.loads(handler( + args, session_id=session, cwd=str(host_home), + hermes_home=str(runtime_project), agent=Agent(), **kwargs, + )) + assert result["answer"] == "grounded answer", result + assert requests[-1] == ( + "tracedecay_lcm_expand_query", + { + "prompt": args["prompt"], "provider": "hermes", "session_id": session, + "storage_scope": "user", "context_max_tokens": budget, + }, + {}, + ), requests[-1] + assert syntheses[-1]["model"] == model, syntheses[-1] + assert syntheses[-1]["timeout"] == timeout, syntheses[-1] + assert config_homes[-1] == str(host_home), config_homes + assert len(requests) == len(syntheses) == len(config_homes) == 5 + assert args == {"prompt": "What did we decide?"} + assert ctx.config["expansion_model"] == "registered-model" + assert override["expansion_model"] == "call-model" finally: - plugin.TraceDecayContextEngine._preflight_probe = real_probe - ok("should_compress_preflight reports typed capability absence") + plugin.tools.call_tracedecay_tool = real_tool + plugin.tools.plugin_config_block = real_block + for key, value in saved_env.items(): + if value is not None: + os.environ[key] = value + else: + os.environ.pop(key, None) + ok("LCM call configuration overrides registration defaults without changing profile or session") def _check_expand_query_degradation(plugin): @@ -1173,7 +1144,7 @@ def _stop_daemon(proc: subprocess.Popen): proc.wait(timeout=5) -def _check_hermes_lcm_session_census(plugin, ctx, work: Path): +def _check_hermes_lcm_session_census(plugin, work: Path): """One ingested Hermes session shows its message; a session that was not ingested shows none. Status is asked only for the session id.""" home = work / "home" @@ -1196,9 +1167,6 @@ def _check_hermes_lcm_session_census(plugin, ctx, work: Path): stderr=subprocess.PIPE, text=True, ) - engine = ctx.engine - saved_root = engine.project_root - saved_session = engine.active_session_id try: deadline = time.monotonic() + 10 while not (profile / "daemon.sock").exists(): @@ -1207,8 +1175,6 @@ def _check_hermes_lcm_session_census(plugin, ctx, work: Path): if time.monotonic() >= deadline: raise AssertionError("isolated daemon socket did not appear") time.sleep(0.05) - engine.project_root = None - engine.active_session_id = "not-the-queried-session" ingested = _call_until_tool_ready( lambda: plugin.call_tracedecay_json( "tracedecay_hook_runtime", @@ -1230,17 +1196,19 @@ def _check_hermes_lcm_session_census(plugin, ctx, work: Path): ) assert ingested.get("status") in ("accepted", "committed"), ingested present = _call_until_tool_ready( - lambda: engine.status(session_id="gate-hermes-session") + lambda: plugin.call_tracedecay_json("tracedecay_lcm_status", { + "provider": "hermes", "session_id": "gate-hermes-session", "storage_scope": "user", + }) ) absent = _call_until_tool_ready( - lambda: engine.status(session_id="gate-empty-session") + lambda: plugin.call_tracedecay_json("tracedecay_lcm_status", { + "provider": "hermes", "session_id": "gate-empty-session", "storage_scope": "user", + }) ) assert _session_message_count(present, "gate-hermes-session") == 1, present assert _session_message_count(absent, "gate-empty-session") == 0, absent - ok("context-engine status counts the ingested session only") + ok("LCM status counts the ingested session only") finally: - engine.project_root = saved_root - engine.active_session_id = saved_session _stop_daemon(daemon) for key, value in saved_env.items(): if value is None: @@ -1262,16 +1230,15 @@ def run_checks(work: Path): _check_nudge_kill_switch(plugin, ctx) _check_canonical_metadata(plugin, plugin_dir, ctx) _check_response_envelope_table(plugin) - messages = _check_engine_state_and_compress(plugin, ctx, host_home, runtime_project) - _check_engine_thread_isolation(ctx.engine) - _check_should_compress_gating(plugin, ctx.engine) + messages = _check_lcm_retrieval_scope(plugin, ctx, host_home, runtime_project) + _check_lcm_handler_configuration(plugin, host_home, runtime_project) _check_expand_query_degradation(plugin) _check_no_hermes_storage_routing(plugin) _check_provider_verbs(plugin, ctx, host_home, messages) _check_prefetch_cache(plugin, ctx) _check_system_prompt_block(ctx) _check_args_file_spill(plugin, work) - _check_hermes_lcm_session_census(plugin, ctx, work) + _check_hermes_lcm_session_census(plugin, work) if __name__ == "__main__": diff --git a/scripts/hermes_stock_check.py b/scripts/hermes_stock_check.py index f1d6bc7897..7c28ead60d 100644 --- a/scripts/hermes_stock_check.py +++ b/scripts/hermes_stock_check.py @@ -10,16 +10,15 @@ Asserts the surfaces stock Hermes actually exposes: 1. the general PluginManager loads + enables the plugin (hook, command), - 2. the context engine registers and is selected via `context.engine`, + 2. Hermes keeps its native compressor while LCM readers register as tools, 3. the memory provider is discovered via `memory.provider` config (stock routes providers through plugins/memory, not PluginContext), 4. real tool dispatch round-trips through the tracedecay binary - (memory facts, LCM status/preflight/compress, graph status). + (memory facts, LCM status/retrieval, graph status). -Everything runs offline: no model calls (compress stays below threshold). +Everything runs offline: compression pressure checks do not call a model. """ -import copy import json import os import subprocess @@ -88,7 +87,7 @@ def assert_tool_dispatch_success_when_ready(dispatch, *, timeout_seconds=15): def assert_lcm_status_argv(argv, project_root): - """Validate the canonical CLI route used by the stock context engine.""" + """Validate the canonical CLI route used by LCM retrieval.""" assert len(argv) >= 6, argv assert argv[1] == "tool", argv name_index = 2 @@ -162,16 +161,15 @@ def main(): assert managed_view.get("success") is True, managed_view assert "Hermes can load this exported managed skill" in managed_view.get("content", "") ok("plugin-native managed skills resolve through qualified skill_view") - # Code-graph / memory / transcript tools register unconditionally; only - # the live-ingest LCM verbs (whose schemas take the in-memory messages - # list) depend on the context_engine_tool_handlers_receive_messages - # capability, which stock does not advertise. + # Retrieval uses stored transcripts, so all LCM readers register through + # stock Hermes without live-message forwarding or context-engine takeover. registered = set(loaded.tools_registered) assert "tracedecay_search" in registered, sorted(registered) assert "tracedecay_context" in registered, sorted(registered) assert "tracedecay_message_search" in registered, sorted(registered) assert "tracedecay_lcm_compress" not in registered, sorted(registered) assert "tracedecay_lcm_preflight" not in registered, sorted(registered) + assert set(plugin.LCM_TOOL_ALIASES).issubset(registered), sorted(registered) # memory.provider is tracedecay here, so the provider-owned fact trio # must not register as direct duplicates. assert "tracedecay_fact_store" not in registered, sorted(registered) @@ -179,7 +177,7 @@ def main(): assert "tracedecay_fact_feedback" not in registered, sorted(registered) assert "tracedecay_memory_status" not in registered, sorted(registered) ok( - "code-graph tools register on stock; LCM + provider-owned tools stay gated", + "code-graph and LCM readers register on stock; provider-owned tools deduplicate", f"{len(registered)} tools", ) from model_tools import get_tool_definitions @@ -239,50 +237,37 @@ def main(): ) ok("TraceDecay tools participate in stock progressive tool discovery") - # 2. Context engine: registered through the plugin and selected the way - # stock agent/agent_init.py selects it (config-driven, plugin fallback). + # 2. Exercise the real host selector and compressor. Registering a no-op + # plugin engine here would silently disable native automatic compaction. from hermes_cli.config import load_config + from agent.agent_init import _select_context_engine + from agent.context_compressor import ContextCompressor config = load_config() - engine_name = (config.get("context") or {}).get("engine") - assert engine_name == "tracedecay", f"context.engine = {engine_name!r}" - ok("config.yaml selects context.engine: tracedecay") - - from plugins.context_engine import load_context_engine - - assert load_context_engine(engine_name) is None - engine = get_plugin_context_engine() - assert engine is not None and engine.name == engine_name - from agent.context_engine import ContextEngine - - assert isinstance(engine, ContextEngine) - ok("context engine activates via stock plugin fallback") - - engine.initialize(session_id="stock-check-session", project_root=project_root) - assert engine.project_root is not None - assert os.path.realpath(engine.project_root) == os.path.realpath(project_root) - engine.update_model("stock-check-model", 128000) - engine.update_from_response({"prompt_tokens": 120, "completion_tokens": 30}) - assert engine.last_total_tokens == 150 - ok("stock ContextEngine ABC surface works", "update_from_response") - - # Stock Hermes deep-copies the registered plugin singleton for every - # AIAgent. The copy must retain routing/budget state without sharing locks - # or live agent references, otherwise Hermes falls back to its compressor. - engine.agent = object() - cloned_engine = copy.deepcopy(engine) - assert cloned_engine is not engine - assert cloned_engine._state_lock is not engine._state_lock - assert cloned_engine.project_root == engine.project_root - assert cloned_engine.context_length == 128000 - assert cloned_engine.last_total_tokens == 150 - assert cloned_engine.agent is None - ok("context engine deep-copies through the stock Hermes agent contract") - - assert engine.should_compress(1000) is False - ok("should_compress gates locally below the tracked threshold") - assert engine.should_compress_preflight([], current_tokens=1000) is False - ok("should_compress_preflight honors the bool ABC contract") + engine_name = (config.get("context") or {}).get("engine") or "compressor" + assert engine_name == "compressor", f"context.engine = {engine_name!r}" + assert get_plugin_context_engine() is None + assert _select_context_engine(config) is None + compressor = ContextCompressor( + model="stock-check-model", config_context_length=128000, + threshold_percent=0.75, quiet_mode=True, + ) + assert compressor.should_compress(1000) is False + assert compressor.should_compress(compressor.threshold_tokens) is True + ok("stock Hermes retains native compaction and responds to context pressure") + + from tools.registry import registry + + def lcm_status(): + raw = registry.dispatch( + "lcm_status", {}, session_id="stock-check-session", project_root=project_root, + ) + outer = assert_tool_dispatch_success(raw) + payloads = plugin._content_json_payloads(outer["content"]) + assert len(payloads) == 1, outer + payload, error = plugin._application_envelope_payload(payloads[0]) + assert error is None, error + return payload lcm_status_argv = [] real_subprocess_run = plugin.tools.subprocess.run @@ -292,10 +277,10 @@ def capture_lcm_status_argv(argv, *args, **kwargs): return real_subprocess_run(argv, *args, **kwargs) plugin.tools.subprocess.run = capture_lcm_status_argv - status = engine.status() + status = lcm_status() finally: plugin.tools.subprocess.run = real_subprocess_run - assert lcm_status_argv, "context engine did not invoke tracedecay_lcm_status" + assert lcm_status_argv, "registered reader did not invoke tracedecay_lcm_status" status_payload = json.loads(assert_lcm_status_argv(lcm_status_argv[-1], project_root)) assert status_payload == { "provider": "hermes", @@ -305,7 +290,7 @@ def capture_lcm_status_argv(argv, *args, **kwargs): assert_decoded_result(status, "status") if status.get("status") == "not_ingested": assert status.get("store_exists") is False, status - ok("lcm_status dispatch round-trips", "not_ingested before compress") + ok("lcm_status dispatch round-trips", "not_ingested before transcript sync") else: assert status.get("session_id") == "stock-check-session", status ok("lcm_status dispatch round-trips") @@ -314,25 +299,6 @@ def capture_lcm_status_argv(argv, *args, **kwargs): {"role": "user", "content": "hello"}, {"role": "assistant", "content": "hi there"}, ] - compressed = engine.compress(list(messages), current_tokens=50) - # Host ABC contract: compress() returns a MESSAGE LIST the host adopts as - # the live transcript. Hermes exposes no authentic raw-compression - # protocol, so the daemon-owned LCM authority reports host raw compaction - # as a typed unavailable capability and the transcript passes through - # unchanged (the transcript itself still ingests through the daemon). - assert isinstance(compressed, list), type(compressed) - assert compressed == messages, compressed - result = engine.last_compress_result - assert isinstance(result, dict) and result == { - "status": "unavailable", - "reason": "host_raw_compression_unavailable", - "semantic_error": True, - }, result - ok( - "compress returns the unchanged transcript with typed unavailability", - f"reason={result.get('reason')}", - ) - # 3. Memory provider: stock discovers providers via plugins/memory and the # memory.provider config key (the general PluginContext has no # register_memory_provider, so this is the only stock activation path). @@ -510,14 +476,14 @@ def capture_lcm_status_argv(argv, *args, **kwargs): "hello", "hi there", session_id="stock-check-session", messages=messages ) ingest_deadline = time.time() + 30 - ingest_status = engine.status() + ingest_status = lcm_status() while time.time() < ingest_deadline and ( not isinstance(ingest_status, dict) or ingest_status.get("status") in (None, "not_ingested") or _retryable_retained_pressure(ingest_status) ): time.sleep(0.25) - ingest_status = engine.status() + ingest_status = lcm_status() assert isinstance(ingest_status, dict), ingest_status assert ingest_status.get("status") not in (None, "not_ingested"), ingest_status grep = call_when_ready(