fix(observability): record_failure never raises; degraded survives scrub

Discharges P0-06 quality I-1 (whole body guarded — observability-recursion
trap closed, red-green proven both directions) + M-3 (protocol-scrub carries
degraded). SIM105 via contextlib.suppress. Suite 433 maintained.
This commit is contained in:
Nick committed 2026-09-04 18:05:51 +01:00
1 parent 7cd3b67766
commit 2d7744d5fd
3 files changed
+216 -75

No files matched your search

@@ -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).
+163 -56
View File
@@ -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:
+30 -19
View File
@@ -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):