diff --git a/plans/2026-09-04-p0-06-failures-observable.md b/plans/2026-09-04-p0-06-failures-observable.md index 0bd02215..800ab173 100644 --- a/plans/2026-09-04-p0-06-failures-observable.md +++ b/plans/2026-09-04-p0-06-failures-observable.md @@ -315,3 +315,26 @@ subsystem to the registry, the /health command. - configure_logging(): called once from RuneApp.__init__, idempotent, never silent-pass. - Non-vacuity: removing the registry's _surfaced anti-spam set fails test_t1_user_message_fires_once_per_subsystem (verified by mutation probe). - Env note: rune_tui resolves `import rune` via memory_env.scripts_dir() — smoke tests need RUNE_SCRIPTS_DIR override; TUI + package deploy as a unit. + +## Post-quality discharges (2026-09-04; quality gate general-156 Approved with minors) + +- I-1 DISCHARGED (observability-recursion trap): record_failure's WHOLE body is now guarded — + reg.record() and the logging call were unguarded (verified: registry.record raising OSError + propagated out of the caller's except block, masking the original failure). RED-GREEN PROVEN: + revert → OSError propagates; guarded → returns False + last-resort log. Suite 433 maintained. +- M-3 DISCHARGED: protocol-scrub rebuild (rune_operator.py:238-245) now carries + degraded=result.degraded — observability no longer lost on the scrub path. +- M-5 NOT-A-DEFECT: the acceptance test already uses a local registry (reviewer misread). +- SIM105: last-resort handler converted to contextlib.suppress (intent-exact). + +## Pins (hardening plan) +- M-4: operator-channel degraded list is per-turn (TUI renders ⚠ every turn while broken) vs + the doc's once-per-session — reconcile (de-dup in TUI or amend docs) before the follow-up + sweep copies the pattern. +- M-6: configure_logging handler cleanup — the test's removeHandler targets the child logger + but the handler lives on the 'rune' root. +- M-7: real-path integration test (monkeypatch memory_store.connect → run _execute_on_model / + _run_warm_router) — cheap, deterministic; belongs in hardening. +- Plan-level: _surfaced re-arm after N clean turns; remaining ~390 silent handlers (follow-up + sweep); parallel_processor per-item error dicts still invisible; failure-semantics.md should + mark the truthful-names extension list as an amendment (7cd3b67). diff --git a/src/rune/rune_operator.py b/src/rune/rune_operator.py index 516b6bf5..263392c1 100644 --- a/src/rune/rune_operator.py +++ b/src/rune/rune_operator.py @@ -97,6 +97,7 @@ def _filter_safety_noise(thinking: str) -> str: @dataclass class RuneTask: """A task that Rune is managing.""" + task_id: str description: str status: str = "pending" # pending, running, completed, failed @@ -120,6 +121,7 @@ class RuneTask: @dataclass class RuneState: """Rune's persistent state.""" + identity: str = "Rune" role: str = "Operator" current_task: Optional[str] = None @@ -161,14 +163,26 @@ class RuneOperator: role=data.get("role", "Operator"), current_task=data.get("current_task"), active_model=data.get("active_model"), - tasks=[RuneTask(**{k: v for k, v in t.items() if k in RuneTask.__dataclass_fields__}) for t in data.get("tasks", [])], + tasks=[ + RuneTask( + **{ + k: v + for k, v in t.items() + if k in RuneTask.__dataclass_fields__ + } + ) + for t in data.get("tasks", []) + ], pending_task_id=data.get("pending_task_id"), pending_request=data.get("pending_request"), pending_runtime_state=data.get("pending_runtime_state"), ) except Exception as e: import logging - logging.getLogger(__name__).warning("Failed to load operator state: %s", e) + + logging.getLogger(__name__).warning( + "Failed to load operator state: %s", e + ) def _save(self): """Save Rune's state to storage.""" @@ -183,9 +197,15 @@ class RuneOperator: "role": self.state.role, "current_task": self.state.current_task, "active_model": self.state.active_model, - "tasks_pending": len([t for t in self.state.tasks if t.status == "pending"]), - "tasks_running": len([t for t in self.state.tasks if t.status == "running"]), - "tasks_completed": len([t for t in self.state.tasks if t.status == "completed"]), + "tasks_pending": len( + [t for t in self.state.tasks if t.status == "pending"] + ), + "tasks_running": len( + [t for t in self.state.tasks if t.status == "running"] + ), + "tasks_completed": len( + [t for t in self.state.tasks if t.status == "completed"] + ), } def send_task(self, description: str, model: str | None = None) -> Dict[str, Any]: @@ -242,6 +262,7 @@ class RuneOperator: model=result.model or model, recall_count=result.recall_count, override_tier=result.override_tier, + degraded=result.degraded, # P0-06: carry observability through the scrub error="internal tool protocol in final content", ) @@ -350,9 +371,7 @@ class RuneOperator: self.state.pending_request = None self.state.pending_runtime_state = None - def _apply_runtime_result( - self, task: RuneTask, result: OperatorResponse - ) -> None: + def _apply_runtime_result(self, task: RuneTask, result: OperatorResponse) -> None: if result.success and result.pending_approval is not None: self.state.pending_task_id = task.task_id self.state.pending_request = asdict(result.pending_approval) @@ -397,15 +416,22 @@ class RuneOperator: desc_lower = description.lower() # Quick tasks — laptop model - if any(word in desc_lower for word in ["hello", "hi", "thanks", "ok", "status"]): + if any( + word in desc_lower for word in ["hello", "hi", "thanks", "ok", "status"] + ): return "qwen2.5:7b" # Coding tasks — local coding model - if any(word in desc_lower for word in ["fix", "code", "debug", "test", "refactor"]): + if any( + word in desc_lower for word in ["fix", "code", "debug", "test", "refactor"] + ): return "qwen3:14b" # Complex tasks — strong local model - if any(word in desc_lower for word in ["design", "architecture", "research", "analyze"]): + if any( + word in desc_lower + for word in ["design", "architecture", "research", "analyze"] + ): return "hf.co/DavidAU/Qwen3.6-27B-Fable-Fusion-711-Uncensored-Heretic-NM-DAU-NEO-MAX-MTP-GGUF:IQ2_M" # Default to laptop model @@ -433,6 +459,7 @@ class RuneOperator: recall_count = 0 try: import rune.memory_store as memory_store + db = memory_store.connect() try: results = memory_store.cached_recall(db, task.description, budget=3) @@ -505,13 +532,22 @@ class RuneOperator: # Always include personality context from memory (not dependent on user message) try: import rune.memory_store as memory_store + db = memory_store.connect() try: - persona_memories = memory_store.cached_recall(db, "Rune personality voice identity how to speak", budget=3) + persona_memories = memory_store.cached_recall( + db, "Rune personality voice identity how to speak", budget=3 + ) if persona_memories: - persona_context = "\n".join(f"- {r.get('text', '')[:100]}" for r in persona_memories if r.get('text')) + persona_context = "\n".join( + f"- {r.get('text', '')[:100]}" + for r in persona_memories + if r.get("text") + ) if persona_context: - system_parts.append(f"\n## Identity context\n{persona_context}") + system_parts.append( + f"\n## Identity context\n{persona_context}" + ) finally: db.close() except Exception as exc: # P0-06: observable degrade (T1) @@ -525,16 +561,20 @@ class RuneOperator: # Memory-as-a-Tool (arXiv:2601.05960) try: from rune.memory_consolidation import MemoryConsolidation + consolidation = MemoryConsolidation() # Consolidate warm neurons into knowledge try: import warm_neuron_store + neurons = warm_neuron_store.list_neurons() for neuron in neurons[:10]: # Consolidate top 10 knowledge = consolidation.consolidate_neuron(neuron) # Check if already consolidated - existing = [k for k in consolidation.knowledge if k.id == knowledge.id] + existing = [ + k for k in consolidation.knowledge if k.id == knowledge.id + ] if not existing: consolidation.knowledge.append(knowledge) except Exception as exc: # P0-06: observable degrade (T1) @@ -546,14 +586,21 @@ class RuneOperator: # Consolidate graph memory into knowledge try: import graph_memory + db = graph_memory.connect() try: task_words = task.description.lower().split()[:5] for word in task_words: entity = graph_memory.resolve_entity(db, word) if entity: - knowledge = consolidation.consolidate_graph_entity(entity) - existing = [k for k in consolidation.knowledge if k.id == knowledge.id] + knowledge = consolidation.consolidate_graph_entity( + entity + ) + existing = [ + k + for k in consolidation.knowledge + if k.id == knowledge.id + ] if not existing: consolidation.knowledge.append(knowledge) finally: @@ -567,19 +614,28 @@ class RuneOperator: # Consolidate abstraction facts into knowledge try: import abstraction + db = abstraction.connect() try: task_words = task.description.lower().split()[:3] if task_words: - placeholders = " OR ".join(["what LIKE ?" for _ in task_words]) + placeholders = " OR ".join( + ["what LIKE ?" for _ in task_words] + ) params = [f"%{w}%" for w in task_words] facts = db.execute( f"SELECT what FROM facts WHERE status='active' AND ({placeholders}) LIMIT 5", - params + params, ).fetchall() for fact in facts: - knowledge = consolidation.consolidate_abstraction_fact(fact[0]) - existing = [k for k in consolidation.knowledge if k.id == knowledge.id] + knowledge = consolidation.consolidate_abstraction_fact( + fact[0] + ) + existing = [ + k + for k in consolidation.knowledge + if k.id == knowledge.id + ] if not existing: consolidation.knowledge.append(knowledge) finally: @@ -587,22 +643,33 @@ class RuneOperator: except Exception as exc: # P0-06: observable degrade (T1) from rune.subsystem_health import record_failure - record_failure("consolidation", exc, tier=1, context="abstraction facts") + record_failure( + "consolidation", exc, tier=1, context="abstraction facts" + ) degraded.append("consolidation") # Consolidate memory recall into knowledge try: import rune.memory_store as memory_store + db = memory_store.connect() try: - results = memory_store.cached_recall(db, task.description, budget=5) + results = memory_store.cached_recall( + db, task.description, budget=5 + ) if results: for r in results: text = r.get("text", "") score = r.get("score", 0.5) if text: - knowledge = consolidation.consolidate_memory(text, score) - existing = [k for k in consolidation.knowledge if k.id == knowledge.id] + knowledge = consolidation.consolidate_memory( + text, score + ) + existing = [ + k + for k in consolidation.knowledge + if k.id == knowledge.id + ] if not existing: consolidation.knowledge.append(knowledge) finally: @@ -610,14 +677,18 @@ class RuneOperator: except Exception as exc: # P0-06: observable degrade (T1) from rune.subsystem_health import record_failure - record_failure("consolidation", exc, tier=1, context="memory recall") + record_failure( + "consolidation", exc, tier=1, context="memory recall" + ) degraded.append("consolidation") # Save consolidated knowledge consolidation._save_knowledge() # Inject consolidated knowledge into system prompt - consolidated_context = consolidation.format_knowledge_for_prompt(task.description) + consolidated_context = consolidation.format_knowledge_for_prompt( + task.description + ) if consolidated_context: system_parts.append(f"\n{consolidated_context}") @@ -630,22 +701,30 @@ class RuneOperator: # Inject warm neurons — these ARE my neurons, they fire when relevant try: import warm_neuron_store + neurons = warm_neuron_store.list_neurons() if neurons: # Find neurons relevant to the current task task_words = set(task.description.lower().split()) relevant = [] for n in neurons: - triggers = n.get('triggers', []) + triggers = n.get("triggers", []) if any(t in task_words for t in triggers): relevant.append(n) # If no specific match, include top neurons by importance if not relevant: - relevant = sorted(neurons, key=lambda x: x.get('importance', 0), reverse=True)[:5] + relevant = sorted( + neurons, key=lambda x: x.get("importance", 0), reverse=True + )[:5] else: relevant = relevant[:5] - neuron_context = "\n".join(f"- [{n.get('slug', '?')}] {n.get('text', '')[:100]}" for n in relevant) - system_parts.append(f"\n## My neurons (warm, firing)\n{neuron_context}") + neuron_context = "\n".join( + f"- [{n.get('slug', '?')}] {n.get('text', '')[:100]}" + for n in relevant + ) + system_parts.append( + f"\n## My neurons (warm, firing)\n{neuron_context}" + ) except Exception as exc: # P0-06: observable degrade (T1) from rune.subsystem_health import record_failure @@ -655,6 +734,7 @@ class RuneOperator: # Inject graph memory — these ARE my connections between ideas try: import graph_memory + db = graph_memory.connect() try: # Get entities relevant to the task @@ -665,8 +745,13 @@ class RuneOperator: if entity: entities.append(entity) if entities: - entity_context = "\n".join(f"- {e.get('name', '?')}: {e.get('description', '')[:80]}" for e in entities[:5]) - system_parts.append(f"\n## My graph connections\n{entity_context}") + entity_context = "\n".join( + f"- {e.get('name', '?')}: {e.get('description', '')[:80]}" + for e in entities[:5] + ) + system_parts.append( + f"\n## My graph connections\n{entity_context}" + ) finally: db.close() except Exception as exc: # P0-06: observable degrade (T1) @@ -678,6 +763,7 @@ class RuneOperator: # Inject abstraction facts — these ARE my learned knowledge try: import abstraction + db = abstraction.connect() try: task_words = task.description.lower().split()[:3] @@ -686,11 +772,13 @@ class RuneOperator: params = [f"%{w}%" for w in task_words] facts = db.execute( f"SELECT what FROM facts WHERE status='active' AND ({placeholders}) LIMIT 3", - params + params, ).fetchall() if facts: fact_context = "\n".join(f"- {f[0][:100]}" for f in facts) - system_parts.append(f"\n## My learned knowledge\n{fact_context}") + system_parts.append( + f"\n## My learned knowledge\n{fact_context}" + ) finally: db.close() except Exception as exc: # P0-06: observable degrade (T1) @@ -710,24 +798,30 @@ class RuneOperator: # Add ambient context (research-backed: "provide a map, not the territory") # Repo map — always present, gives codebase awareness - if hasattr(task, '_ambient_repo_map') and task._ambient_repo_map: - system_parts.append(f"\n## Codebase structure\n{task._ambient_repo_map[:1500]}") + if hasattr(task, "_ambient_repo_map") and task._ambient_repo_map: + system_parts.append( + f"\n## Codebase structure\n{task._ambient_repo_map[:1500]}" + ) # Auto-read files — files mentioned in the message - if hasattr(task, '_ambient_files') and task._ambient_files: + if hasattr(task, "_ambient_files") and task._ambient_files: files_context = [] for fpath, content in task._ambient_files.items(): files_context.append(f"### {fpath}\n{content[:500]}") if files_context: - system_parts.append(f"\n## Referenced files\n{''.join(files_context)}") + system_parts.append( + f"\n## Referenced files\n{''.join(files_context)}" + ) # Auto-search functions — functions mentioned in the message - if hasattr(task, '_ambient_functions') and task._ambient_functions: + if hasattr(task, "_ambient_functions") and task._ambient_functions: funcs_context = [] for func_name, locations in task._ambient_functions.items(): funcs_context.append(f"- {func_name}: {locations[:200]}") if funcs_context: - system_parts.append(f"\n## Referenced functions\n{''.join(funcs_context)}") + system_parts.append( + f"\n## Referenced functions\n{''.join(funcs_context)}" + ) system_prompt = "\n".join(system_parts) @@ -737,12 +831,14 @@ class RuneOperator: user_prompt = task.description # Log token usage - self._log_token_usage({ - "static": 0, - "dynamic": capsule.token_count, - "user": len(task.description) // 4, - "total": capsule.token_count + len(task.description) // 4, - }) + self._log_token_usage( + { + "static": 0, + "dynamic": capsule.token_count, + "user": len(task.description) // 4, + "total": capsule.token_count + len(task.description) // 4, + } + ) # Get safety override — always start at model profile's tier override = safety_override.SafetyOverride() @@ -750,7 +846,9 @@ class RuneOperator: # Always use model profile's tier as baseline (blanket override) if profile.safety_tier > 0: override_params["tier"] = min(profile.safety_tier, 3) - override_params["reason"] = f"model_profile tier {profile.safety_tier} (blanket)" + override_params["reason"] = ( + f"model_profile tier {profile.safety_tier} (blanket)" + ) # Adaptive timeout based on model size timeout = 30 @@ -832,7 +930,9 @@ class RuneOperator: except Exception as e: model_profiles.record_interaction(model, False, task.description) - return OperatorResponse(success=False, model=model, error=str(e), degraded=sorted(set(degraded))) + return OperatorResponse( + success=False, model=model, error=str(e), degraded=sorted(set(degraded)) + ) @staticmethod def _finalize_runtime_response( @@ -872,18 +972,23 @@ class RuneOperator: # Rotate if file exceeds 10MB try: - if os.path.exists(log_path) and os.path.getsize(log_path) > 10 * 1024 * 1024: + if ( + os.path.exists(log_path) + and os.path.getsize(log_path) > 10 * 1024 * 1024 + ): rotated = log_path + ".old" os.replace(log_path, rotated) except OSError: pass with open(log_path, "a") as f: - f.write(f"[{datetime.now(timezone.utc).isoformat()}] " - f"static={tokens.get('static', 0)} " - f"dynamic={tokens.get('dynamic', 0)} " - f"user={tokens.get('user', 0)} " - f"total={tokens.get('total', 0)}\n") + f.write( + f"[{datetime.now(timezone.utc).isoformat()}] " + f"static={tokens.get('static', 0)} " + f"dynamic={tokens.get('dynamic', 0)} " + f"user={tokens.get('user', 0)} " + f"total={tokens.get('total', 0)}\n" + ) def list_models(self) -> List[Dict[str, Any]]: """List available models.""" @@ -964,7 +1069,9 @@ def main(): print(f"Role: {status['role']}") print(f"Current task: {status['current_task'] or 'None'}") print(f"Active model: {status['active_model'] or 'None'}") - print(f"Tasks: {status['tasks_pending']} pending, {status['tasks_running']} running, {status['tasks_completed']} completed") + print( + f"Tasks: {status['tasks_pending']} pending, {status['tasks_running']} running, {status['tasks_completed']} completed" + ) elif command == "send": if len(sys.argv) < 3: diff --git a/src/rune/subsystem_health.py b/src/rune/subsystem_health.py index 15de65dd..bf26a5f2 100644 --- a/src/rune/subsystem_health.py +++ b/src/rune/subsystem_health.py @@ -14,6 +14,7 @@ failure, or emit. Silent `pass` is a defect. from __future__ import annotations +import contextlib import json import logging import threading @@ -70,27 +71,37 @@ def record_failure( Returns True when the caller should ALSO surface this to the user (T0/T1 first occurrence). T2 never surfaces. """ - reg = registry or _REGISTRY - message = f"{type(error).__name__}: {error}" - if context: - message = f"{message} ({context})" - log = logging.getLogger("rune.subsystem_health") - log.error("subsystem=%s tier=%d %s", subsystem, tier, message) + # The observability layer must NEVER raise inside a caller's except + # block — a raising record_failure masks the very failure it reports + # (P0-06 quality discharge I-1: the observability-recursion trap). try: - from rune import event_bus + reg = registry or _REGISTRY + message = f"{type(error).__name__}: {error}" + if context: + message = f"{message} ({context})" + log = logging.getLogger("rune.subsystem_health") + log.error("subsystem=%s tier=%d %s", subsystem, tier, message) + try: + from rune import event_bus - event_bus.emit( - _TIER_EVENTS.get(tier, "storage.degraded"), - source="subsystem_health", - payload={ - "subsystem": subsystem, - "tier": tier, - "error": message[:300], - }, - ) - except Exception: # the observability layer itself must not raise - log.exception("failed to emit subsystem-health event") - return reg.record(subsystem, message, tier) + event_bus.emit( + _TIER_EVENTS.get(tier, "storage.degraded"), + source="subsystem_health", + payload={ + "subsystem": subsystem, + "tier": tier, + "error": message[:300], + }, + ) + except Exception: + log.exception("failed to emit subsystem-health event") + return reg.record(subsystem, message, tier) + except Exception: + with contextlib.suppress(Exception): + logging.getLogger("rune.subsystem_health").exception( + "record_failure itself failed for subsystem=%s", subsystem + ) + return False # last resort: never mask the caller's original error def configure_logging(log_dir: Path | None = None):