feat(tui): the cognitive loops are actually INVOKED — #2 (contract BUILDER-2026-10-08T13:15:49Z-03dc90)

- The defect: RuneApp instantiated CognitiveLoop/DreamCycle/AutonomyLoop and
  told the user '◆ Cognitive loop active — Rune thinks between turns' — but
  never called think()/dream()/evaluate() anywhere. A dead mechanism wearing
  a live-status costume.
- The fix: an idle-cognition ticker (set_interval 240s) on RuneApp — think()
  every tick (consolidate, probe dead weight, drift check) with the most
  recent user context, dream() every third tick (full consolidation), on a
  worker thread; only dim status lines touch the UI thread; failures degrade
  silently. Input submit captures _cognition_ctx.
- Bonus fix: module-level logger in rune_tui.py — 61 latent F821 NameErrors
  inside exception handlers (any suppressed-exception path that reached
  logger.debug would have raised instead of degrading). Now 0.
- tests/test_cognitive_wiring.py: 5/5 — think() invoked with context, dream()
  gated to the third tick, actions surface as status, silence stays silent,
  the wiring pinned at source level.

src/rune/rune_tui.py

@@ -22,6 +22,8 @@ import os
 import sys
 from datetime import datetime, timezone
 
+logger = logging.getLogger("rune.tui")
+
 # Debug logger for silent exception swallowing
 _log = logging.getLogger("rune_tui")
 
@@ -82,6 +84,8 @@ def _operator_transport():
     tui_transport (RUNE_TUI_TRANSPORT=inprocess|v1, default inprocess —
     zero behavior change until the wave opts in)."""
     return _tui_transport.get_transport()
+
+
 import cron_engine
 import night_mode
 import time_awareness
@@ -160,6 +164,7 @@ def collect_post_response_signals(response: str, messages_count: int) -> list[st
         pass
     return signals
 
+
 from textual.app import App, ComposeResult
 from textual.containers import Horizontal, Vertical, VerticalScroll
 from textual.widgets import Button, Header, Input, ProgressBar, RichLog, Static
@@ -1744,6 +1749,55 @@ class RuneApp(App):
             logger.debug("suppressed exception", exc_info=True)
             self.chat_channel.add_message("system", f"[dim]⚠ Autonomy loop failed: {e}[/dim]")
 
+        # #2 (the dead-mechanism fix): the loops above were instantiated but
+        # NEVER INVOKED — the "thinks between turns" message was a lie. The
+        # idle-cognition ticker invokes them for real: think() every tick,
+        # dream() every third, off the render thread, results as dim status.
+        self._cognition_ctx = ""
+        self._dream_count = 0
+        self.set_interval(240, self._idle_cognition_tick)
+
+    def _idle_cognition_tick(self) -> None:
+        """#2: Rune thinks between turns. Runs the cognitive loop's think()
+        (consolidate, probe dead weight, check drift) every tick and the
+        dream cycle's full consolidation every third tick. Memory work runs
+        on a worker thread; only the dim status lines touch the UI thread."""
+        import threading
+
+        def _tick():
+            try:
+                loop = getattr(self, "cognitive_loop", None)
+                if loop is not None:
+                    res = loop.think(context=getattr(self, "_cognition_ctx", ""))
+                    acted = [
+                        a["type"]
+                        for a in res.get("actions", [])
+                        if isinstance(a, dict) and not str(a.get("type", "")).endswith("_error")
+                    ]
+                    if acted:
+                        self.call_from_thread(
+                            self.chat_channel.add_message,
+                            "system",
+                            f"[dim]◆ think: {', '.join(acted)}[/dim]",
+                        )
+            except Exception as e:  # noqa: BLE001 — idle cognition must never crash the session (degrade-not-crash)
+                logger.debug("cognitive tick failed", exc_info=True)
+            try:
+                dream = getattr(self, "dream_cycle", None)
+                if dream is not None:
+                    self._dream_count = getattr(self, "_dream_count", 0) + 1
+                    if self._dream_count % 3 == 0:
+                        dream.dream()
+                        self.call_from_thread(
+                            self.chat_channel.add_message,
+                            "system",
+                            "[dim]◆ dream: consolidation cycle complete[/dim]",
+                        )
+            except Exception as e:  # noqa: BLE001 — degrade-not-crash
+                logger.debug("dream tick failed", exc_info=True)
+
+        threading.Thread(target=_tick, daemon=True).start()
+
     def on_static_clicked(self, event: Static.Clicked) -> None:
         """Handle refresh button click."""
         if event.widget.id == "refresh-btn":
@@ -1779,12 +1833,12 @@ class RuneApp(App):
             the load_tasks fallback then handles."""
             try:
                 import rune_operator as _ro
+
                 # H4: private _task_state peek — stays in-process by design
                 # (the /v1 surface carries the four public calls only).
                 return getattr(_ro.get_operator(), "_task_state", None)
             except Exception as exc:  # noqa: BLE001 — deliberate degradation: startup ordering; the fallback path handles it and the degradation is logged
-                print(f"[task-loader] canonical repo unavailable ({exc}); "
-                      f"falling back to export ledger", flush=True)
+                print(f"[task-loader] canonical repo unavailable ({exc}); falling back to export ledger", flush=True)
                 return None
 
         def load_tasks():
@@ -1798,10 +1852,10 @@ class RuneApp(App):
                     if repo is not None:
                         records = repo.list_all()
                         active = [
-                            {"task_id": r.task_id, "title": r.title,
-                             "status": r.status}
+                            {"task_id": r.task_id, "title": r.title, "status": r.status}
                             for r in records
-                            if r.status in ("created", "running")]
+                            if r.status in ("created", "running")
+                        ]
                         all_pins = pins.list_pins()
                         open_pins = [p for p in all_pins if p.get("status") == "open"]
                         return ("tasks", active, open_pins)
@@ -2149,6 +2203,9 @@ class RuneApp(App):
         if not text:
             return
 
+        # #2: the idle-cognition ticker thinks with the most recent context.
+        self._cognition_ctx = text
+
         # Record in input history (up-arrow recall)
         if isinstance(event.input, HistoryInput):
             event.input.add_to_history(text)
@@ -2220,8 +2277,7 @@ class RuneApp(App):
                 if _mode.mode in ("PROJECT_BUILD", "EXISTING_PROJECT_CHANGE"):
                     ch.add_message(
                         "system",
-                        f"[yellow]🔨 Build mode recommended "
-                        f"(classified: {_mode.mode})[/yellow]",
+                        f"[yellow]🔨 Build mode recommended (classified: {_mode.mode})[/yellow]",
                     )
                     ch.add_message(
                         "system",
@@ -3267,48 +3323,39 @@ class RuneApp(App):
             from rune import testing_mode as tm
             from rune.behaviour_ledger import BehaviourLedger
             from rune.project_registry import ProjectRegistry
+
             registry = ProjectRegistry("store/scaffold")
             ch = self._ch()
             if arg.strip():
                 # project filter: read THAT project's ledger
                 entry = registry.get(arg.strip())
                 if entry is None:
-                    ch.add_message("system",
-                        f"[yellow]No registered scaffolded project "
-                        f"{arg.strip()!r}[/yellow]")
+                    ch.add_message("system", f"[yellow]No registered scaffolded project {arg.strip()!r}[/yellow]")
                     return
                 loc = entry["scaffold_location"]
                 if not (_Path(loc) / "behaviour_ledger.json").is_file():
-                    ch.add_message("system",
-                        f"[yellow]{arg.strip()} has no behaviour ledger "
-                        f"yet[/yellow]")
+                    ch.add_message("system", f"[yellow]{arg.strip()} has no behaviour ledger yet[/yellow]")
                     return
                 beh = BehaviourLedger(loc)
                 items = beh.outstanding()
-                ch.add_message("system",
-                    f"[bold]{arg.strip()}: {len(items)} outstanding "
-                    f"validation(s)[/bold]")
+                ch.add_message("system", f"[bold]{arg.strip()}: {len(items)} outstanding validation(s)[/bold]")
             else:
                 rows = tm.aggregate(registry, _Path("store/scaffold"))
-                ch.add_message("system", "[bold]Testing aggregate "
-                               "(registered scaffolded projects)[/bold]")
+                ch.add_message("system", "[bold]Testing aggregate (registered scaffolded projects)[/bold]")
                 for r in rows:
                     if r["outstanding"] is None:
-                        ch.add_message("system",
-                            f"  {r['project']}: no behaviour ledger yet")
+                        ch.add_message("system", f"  {r['project']}: no behaviour ledger yet")
                     else:
-                        ch.add_message("system",
-                            f"  {r['project']}: {r['outstanding']} "
-                            f"outstanding")
+                        ch.add_message("system", f"  {r['project']}: {r['outstanding']} outstanding")
                 items = []
             for b in items[:8]:
-                ch.add_message("system",
+                ch.add_message(
+                    "system",
                     f"  • {b['behaviour_id']}: {b['intended_behaviour'][:70]} "
-                    f"[{b['state']}]"
-                    + (f" — procedure: {b['procedure'][:60]}" if b.get("procedure") else ""))
+                    f"[{b['state']}]" + (f" — procedure: {b['procedure'][:60]}" if b.get("procedure") else ""),
+                )
         except Exception as exc:  # noqa: BLE001 — user-visible surface: report, never crash the TUI
-            self._ch().add_message("system",
-                f"[red]/testing failed: {exc}[/red]")
+            self._ch().add_message("system", f"[red]/testing failed: {exc}[/red]")
 
     def _cmd_build(self, arg):
         """Build mode — full build pipeline with worktree isolation."""
@@ -3349,20 +3396,22 @@ class RuneApp(App):
                 # ledger remain its unique capabilities).
                 try:
                     from rune.construction_interface import ConstructionInterface
+
                     iface = ConstructionInterface(builder="tui-build-cmd")
                     envelope = iface.classify(description)
                     state = envelope.get("state") or {}
                     mode_name = state.get("mode", "UNKNOWN")
                     probes = state.get("probe_questions", [])
                 except Exception as exc:  # noqa: BLE001 — classification
-                    mode_name = "UNKNOWN"   # unavailable: refuse, fail closed
+                    mode_name = "UNKNOWN"  # unavailable: refuse, fail closed
                     probes = []
-                    self._ch().add_message("system",
-                        f"[red]Construction classification unavailable ({exc}) — refusing to start a build[/red]")
+                    self._ch().add_message(
+                        "system",
+                        f"[red]Construction classification unavailable ({exc}) — refusing to start a build[/red]",
+                    )
                     return
                 if mode_name in ("NOT_CONSTRUCTION", "UNKNOWN"):
-                    msg = (f"[yellow]'/build start' refused: classified "
-                           f"{mode_name}.[/yellow]")
+                    msg = f"[yellow]'/build start' refused: classified {mode_name}.[/yellow]"
                     for q in probes[:3]:
                         msg += f"\n  - {q}"
                     self._ch().add_message("system", msg)
@@ -3381,26 +3430,31 @@ class RuneApp(App):
                 # BuildManager remains a bounded worktree/ledger adapter
                 # that is only ever reachable WITH a governing contract.
                 import time as _t
+
                 _cid = f"build-{int(_t.time())}"
                 try:
                     from rune.construction_interface import ConstructionInterface
+
                     _iface = ConstructionInterface(builder="tui-build-cmd")
                     _doc = _iface.scaffold.load()
                     _iface.scaffold.new_contract(
-                        _doc, contract_id=_cid, need=description,
-                        origin="user", scope=[],
-                        requirement="/build start (classified: "
-                                    f"{mode_name})")
+                        _doc,
+                        contract_id=_cid,
+                        need=description,
+                        origin="user",
+                        scope=[],
+                        requirement=f"/build start (classified: {mode_name})",
+                    )
                     _iface.scaffold.save(_doc)
-                    _iface.transition(_cid, "approved",
-                                      evidence_ref="user:/build-start")
-                    _iface.transition(_cid, "building",
-                                      evidence_ref="build-manager-adapter")
+                    _iface.transition(_cid, "approved", evidence_ref="user:/build-start")
+                    _iface.transition(_cid, "building", evidence_ref="build-manager-adapter")
                 except Exception as exc:  # noqa: BLE001 — T19 19A: contract mint failure FAILS CLOSED (refuse, never degrade to ungoverned construction)
-                    self._ch().add_message("system",
+                    self._ch().add_message(
+                        "system",
                         f"[red]'/build start' refused: canonical contract "
                         f"mint failed ({exc}) — construction cannot start "
-                        f"without its governing contract.[/red]")
+                        f"without its governing contract.[/red]",
+                    )
                     return
                 manager = BuildManager()
                 build_id = manager.start_build(description)

tests/test_cognitive_wiring.py

@@ -0,0 +1,125 @@
+"""test_cognitive_wiring.py — #2: the cognitive loop is actually INVOKED.
+
+The defect: RuneApp instantiated CognitiveLoop/DreamCycle but never called
+think()/dream() — the "[dim]◆ Cognitive loop active[/dim]" message was a
+dead mechanism. The fix: an idle-cognition ticker that invokes think() every
+tick and dream() every third, off the render thread.
+
+These tests pin the INVOCATION SEMANTICS with stub loops (no live TUI, no
+live memory DBs): the tick must call think() with the captured context,
+call dream() only on the third tick, and post status only for real actions.
+"""
+
+import sys
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+import importlib.util
+
+spec = importlib.util.spec_from_file_location("rune_tui", ROOT / "src/rune/rune_tui.py")
+rune_tui = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(rune_tui)
+
+
+class _StubLoop:
+    def __init__(self):
+        self.calls = []
+
+    def think(self, context: str = "") -> dict:
+        self.calls.append(("think", context))
+        return {"actions": [{"type": "curiosity_probe", "probed": 5, "found": 2}]}
+
+
+class _StubDream:
+    def __init__(self):
+        self.calls = 0
+
+    def dream(self) -> dict:
+        self.calls += 1
+        return {"actions": [{"type": "consolidation"}]}
+
+
+class _StubApp:
+    """The tick method's dependency surface, and nothing else."""
+
+    def __init__(self):
+        self.cognitive_loop = _StubLoop()
+        self.dream_cycle = _StubDream()
+        self.autonomy_loop = None
+        self._cognition_ctx = "hello world"
+        self._dream_count = 0
+        self.messages = []
+
+        def call_from_thread(fn, *args):
+            self.messages.append((fn, args))
+
+        self.call_from_thread = call_from_thread
+
+        class _Chan:
+            def __init__(self, sink):
+                self._sink = sink
+
+            def add_message(self, *args):
+                self._sink(("msg", args))
+
+        self.chat_channel = _Chan(self.messages.append)
+
+    _idle_cognition_tick = rune_tui.RuneApp._idle_cognition_tick
+
+
+class TestCognitiveWiring(unittest.TestCase):
+    def _drain(self, app):
+        """The tick spawns a worker thread; join it deterministically."""
+        import time
+
+        for _ in range(50):
+            time.sleep(0.05)
+            if not any(t.is_alive() for t in __import__("threading").enumerate() if t.name.startswith("Thread")):
+                break
+        time.sleep(0.2)
+
+    def test_tick_invokes_think_with_context(self):
+        app = _StubApp()
+        app._idle_cognition_tick()
+        self._drain(app)
+        self.assertEqual(app.cognitive_loop.calls, [("think", "hello world")])
+
+    def test_dream_only_on_third_tick(self):
+        app = _StubApp()
+        app._idle_cognition_tick()
+        self._drain(app)
+        self.assertEqual(app.dream_cycle.calls, 0, "dream must NOT fire on tick 1")
+        app._idle_cognition_tick()
+        self._drain(app)
+        self.assertEqual(app.dream_cycle.calls, 0)
+        app._idle_cognition_tick()
+        self._drain(app)
+        self.assertEqual(app.dream_cycle.calls, 1, "dream fires on the third tick")
+
+    def test_think_actions_surface_as_status(self):
+        app = _StubApp()
+        app._idle_cognition_tick()
+        self._drain(app)
+        texts = [args[1] for (_fn, args) in app.messages if isinstance(args, tuple) and len(args) >= 2]
+        self.assertTrue(any("curiosity_probe" in t for t in texts), f"real think actions must surface: {app.messages}")
+
+    def test_silent_loops_post_nothing(self):
+        app = _StubApp()
+        app.cognitive_loop = _StubLoop()
+        app.cognitive_loop.think = lambda context="": {"actions": []}
+        app._idle_cognition_tick()
+        self._drain(app)
+        texts = [args[1] for (_fn, args) in app.messages if isinstance(args, tuple) and len(args) >= 2]
+        self.assertEqual([t for t in texts if "think:" in t], [], "no actions -> no status spam")
+
+    def test_source_wires_the_ticker(self):
+        src = (ROOT / "src/rune/rune_tui.py").read_text()
+        self.assertIn("set_interval(240, self._idle_cognition_tick)", src)
+        self.assertIn("self._cognition_ctx = text", src, "input submit must capture context for the ticker")
+
+
+if __name__ == "__main__":
+    unittest.main()