2026-09-06 05:32:19 +01:00
10 changed files with 1889 additions and 55 deletions
@@ -90,6 +90,7 @@ jobs:
            src/rune/canary_p0_04.py \
            src/rune/canary_p0_05.py \
            src/rune/canary_p0_07.py \
+            src/rune/canary_p1_07.py \
            src/rune/canary_query_memory.py \
            src/rune/canary_unbounded.py \
            src/rune/canary_full_system.py \
@@ -401,6 +401,24 @@
          "role": "Literal chat presentation primitive",
          "disposition": "retain-authority",
          "evidence": "Renders role-specific text without evaluating message markup."
+        },
+        {
+          "path": "src/rune/tool_protocol.py",
+          "role": "Typed runtime-object set (MemoryQuery, Observation, VerificationResult, Outcome) with to_dict/from_dict round-trips",
+          "disposition": "retain-authority",
+          "evidence": "P1-07 (#30): frozen dataclasses carrying observable plan/action/result state on the response envelope; OperatorResponse.runtime_objects is the additive envelope key."
+        },
+        {
+          "path": "src/rune/rune_tui.py",
+          "role": "WorkPane operational index (third collapsible column, Ctrl+3) fed by _consume_runtime_objects",
+          "disposition": "migrate-adapter",
+          "evidence": "P1-07 (#30): projects runtime objects by stable id (update-in-place, newest-first sections); the chat log keeps its inline ToolCallWidget narrative (two-view principle)."
+        },
+        {
+          "path": "src/rune/rune_operator.py",
+          "role": "Runtime-object emission points (MemoryQuery from the broker trace, Observations, VerificationResult + Outcome) on the final response path",
+          "disposition": "migrate-adapter",
+          "evidence": "P1-07 (#30): built from _last_trace + the response; Outcome.status is back-filled from the final task transition; #33 (OutcomeLedger) adopts the Outcome shape."
        }
      ]
    },
@@ -0,0 +1,79 @@
+# Runtime Objects — Interstitial Contextual Messaging (P1-07, #30)
+
+## Summary
+
+Runtime objects are the observable plan/action/result state of a Rune turn,
+carried as typed, stable-id data on the operator's response envelope and
+projected into the TUI's **work pane**. They are the same events the chat log
+already narrates; the work pane is the operational index, the log remains the
+narrative record (**two-view principle**).
+
+No chain-of-thought ever rides a runtime object: only what was actually
+admitted, run, observed, verified, or concluded.
+
+## Object set (`src/rune/tool_protocol.py`)
+
+All objects are frozen dataclasses with `to_dict()` / `from_dict()` round-trips
+and ISO-UTC `ts`. The pre-existing `ToolRequest` / `ToolEvent` got the same
+round-trip surface and are part of the set.
+
+| Object | id field | Meaning |
+| --- | --- | --- |
+| `MemoryQuery` | `query_id` | One context-broker memory-selection pass: `admitted` (`{block_name: tokens}`), `dropped` (`[{name, tokens}]`), `memory_tokens`. |
+| `Observation` | `observation_id` | Observable runtime event: `kind` (`tool_result` / `approval` / `compaction` / `degraded` / `note`), one-line `summary`, bounded `detail` (<=300 chars). |
+| `VerificationResult` | `verification_id` | Objective verification of a task outcome: `method` (`exit_code` / `test` / `diff` / `manual` / `none`), `passed`, bounded `evidence`. Never self-report. |
+| `Outcome` | `outcome_id` | Final classified task outcome: `status` (`completed` / `failed` / `superseded` / `cancelled`), `verification_id` link ("" when none), `degraded`. |
+
+## Emission points (`src/rune/rune_operator.py`)
+
+Inside `_execute_on_model`'s tier loop, on the final response path (before
+`_finalize_runtime_response`):
+
+1. **MemoryQuery** — one per broker `assemble`, built from `_last_trace`
+   (admitted blocks → `{name: tokens}`, non-admitted blocks → dropped).
+2. **Observations** — one per broker compaction event (`kind=compaction`,
+   summary carries the `tokens_before -> tokens_after` delta) and one per
+   degraded subsystem name (`kind=degraded`, summary
+   `subsystem unavailable: <name>`).
+3. **VerificationResult + Outcome** — emitted when the turn finishes without a
+   pending approval. Method rule: `exit_code` when a `run_command` tool event
+   completed successfully, `test` when `run_tests` completed, otherwise no
+   VerificationResult is emitted at all (`method=none` is never claimed).
+   `passed` = final response success AND no failed tool event. The Outcome's
+   status is back-filled from the final task transition in `send_task`
+   (`completed` / `failed`), and the verification link is set iff the
+   verification passed **and** the task completed.
+
+Emission is degrade-not-crash: a failure to build runtime objects records a
+`runtime_objects` subsystem failure (tier 2) and never fails the turn.
+
+The objects flow to the TUI as dicts via the additive `runtime_objects` field
+on `OperatorResponse` (`_response_envelope` uses `asdict`, so the key survives
+automatically; default `[]` keeps pre-broker failure paths clean).
+
+## Work pane (`src/rune/rune_tui.py`)
+
+`WorkPane(Vertical)` is the third column in `#channels` (`id="work-pane"`,
+collapsible via Ctrl+3). CSS: `.work-pane { width: 1fr; border-left: solid
+$secondary; }`.
+
+- **Data methods are headless-testable**: `add_object(obj)` / `update_object(obj)`
+  (in-place by id, no duplicates) / `object_count()`; `render()` builds the
+  display. Tests exercise the unmounted instance.
+- Sections render newest-first: Outcomes, Verification, Observations, Memory
+  queries, Tool calls. Each line: kind glyph + stable id + one-line summary
+  (+ bounded detail).
+- Tool calls keep their existing `ToolCallWidget` in the chat log AND get a
+  one-line mirror in the pane (the pane is the operational index; the log
+  keeps the interaction).
+- Visibility policy: hidden on the chat channel, visible on task; Ctrl+3
+  toggles the expansion. `_consume_runtime_objects` feeds the pane from every
+  response envelope (including approval resumes), degrade-not-crash.
+
+## #33 handoff
+
+`Outcome` is the OutcomeLedger's future record shape. This stage ships the
+typed object and its rendering only; ledger wiring (persisting outcomes) is
+the #33 dependency that adopts this shape. `VerificationResult` is the
+Verifier strategy's typed claim; the verifier itself is also #33 — until then
+the operator derives it from observable tool results (never self-report).
@@ -0,0 +1,158 @@
+# P1-07 (#30) — Interstitial Contextual Messaging: Work Pane + First-Class Runtime Objects
+
+> **For agentic workers:** execution contract — FROZEN TESTS ARE THE SPEC. Implement verbatim; record every deviation in the plan tail. TDD red→green. Do not create/complete tasks; controller owns task state.
+
+**Gitea issue:** #30 (P1 — high). **Branch:** `p1-07-work-pane` off master `0feeed6`.
+
+## Why
+
+Tool calls already render as first-class widgets (ToolCallWidget, shipped with #21/#22) — but they live INLINE in the chat log, and the rest of the runtime's cognitive events (memory queries, observations, verification, outcomes) have no typed objects or rendering at all. #30's full contract: a dedicated **work pane** showing observable plan/action/result state while the chat pane stays conversational, plus the full typed-object set.
+
+## What already exists (build on, don't rebuild)
+
+- `ToolRequest`/`ToolEvent` typed dataclasses (tool_protocol.py:23/31) — the model for the new objects.
+- `ToolCallWidget` (message_widgets.py:14) — status glyphs, approval buttons, in-place `apply()` updates, keyed by `(channel, call_id)`.
+- `ThinkingBubble` — thinking blocks, not prose.
+- Live wiring: `rune_tui.py:3522-3533` mounts/updates widgets from `result["tool_events"]`.
+- `CognitiveJob` exists in tiered_executor.py:36 (dormant module — do NOT import from there; define the live object in tool_protocol.py per the frozen spec).
+- `VerificationResult` exists in verification_layer.py:24 (dormant — same rule).
+
+## Constraints
+
+- **Two-pane layout**: TASK channel gains a sibling WORK pane (collapsible); CHAT stays conversational. The work pane shows the ACTIVE channel's runtime objects — tool calls, memory queries, observations, verification, outcomes — as a live operational view. Chat log keeps the widgets it already mounts (both views of the same events; the work pane is the operational surface, the log remains the narrative record).
+- **No hidden chain-of-thought**: only observable plan/action/result state is rendered.
+- **Event identity**: every runtime object carries a stable id; the work pane tracks objects by id, updating in place (same pattern as `_tool_widgets`).
+- **Additive protocol**: new dataclasses in `tool_protocol.py` follow the `ToolRequest`/`ToolEvent` shape (frozen dataclass + `to_dict`/`from_dict` round-trip).
+- **Dormant modules stay dormant**: do NOT wire tiered_executor or verification_layer — the new objects are independent live types; the dormant modules may adopt them later.
+- **Functionality immutable**: chat/task channels, /commands, model selector, footer — untouched.
+- Memory queries: the ContextBroker (#19) already records per-block contributions in its trace; a `MemoryQuery` object is emitted per broker assemble (name, admitted blocks, tokens, dropped) — wired from the operator's existing `_last_trace`.
+- Verification/Outcome objects: emitted from the operator's response path (success/failure/degraded) — the OutcomeLedger wiring itself is #33, NOT this stage; here they are runtime objects + renderings only.
+
+## Task 1 — Typed runtime objects (tool_protocol.py)
+
+```python
+@dataclass(frozen=True)
+class MemoryQuery:
+    """One context-broker memory-selection pass (observable, not hidden)."""
+    query_id: str
+    task_id: str
+    admitted: dict  # {block_name: tokens}
+    dropped: list   # [{name, tokens}]
+    memory_tokens: int
+    ts: str  # ISO timestamp
+
+
+@dataclass(frozen=True)
+class Observation:
+    """An observable runtime event the operator should see."""
+    observation_id: str
+    task_id: str
+    kind: str      # "tool_result" | "approval" | "compaction" | "degraded" | "note"
+    summary: str   # one-line, no chain-of-thought
+    detail: str    # bounded detail (<=300 chars)
+    ts: str
+
+
+@dataclass(frozen=True)
+class VerificationResult:
+    """Objective verification of a task outcome (tests/exit/diff — never self-report)."""
+    verification_id: str
+    task_id: str
+    method: str    # "exit_code" | "test" | "diff" | "manual" | "none"
+    passed: bool
+    evidence: str  # bounded (<=300 chars)
+    ts: str
+
+
+@dataclass(frozen=True)
+class Outcome:
+    """The final classified outcome of a task (feeds #33 later)."""
+    outcome_id: str
+    task_id: str
+    status: str    # "completed" | "failed" | "superseded" | "cancelled"
+    verification_id: str  # links to VerificationResult ("" if none)
+    degraded: list
+    ts: str
+```
+
+All five: `to_dict()` + `from_dict()` round-trip (mirror ToolRequest's shape; `ts` ISO UTC).
+
+## Task 2 — Operator emission points
+
+`rune_operator.py`:
+1. After broker assemble (inside `_execute_on_model`, where `_last_trace` is set): build a `MemoryQuery` from the trace (admitted/dropped/tokens) and attach to the response envelope: `response.runtime_objects` — a new additive list field on `OperatorResponse` (default empty; carries dicts via `to_dict()`).
+2. On final response: build `VerificationResult` (method: "exit_code" if the tool loop ran run_command with exit 0, "test" if run_tests passed, "none" otherwise; passed: response.success and not degraded-critical; evidence: bounded summary) and `Outcome` (status from task transition; links verification).
+3. Observations: emit for compaction events (from trace["compaction"]) and degraded subsystems (one per degraded name).
+4. `runtime_objects` flows through `_response_envelope` (additive key) → TUI.
+
+## Task 3 — Work pane (rune_tui.py)
+
+1. New widget `WorkPane(Vertical)` — docked as a third column in `#channels` (collapsible via Ctrl+3 / a header click; default visible on the task channel, hidden on chat). Header: "WORK — runtime objects". Sections render newest-first: Outcomes, Verification, Observations, Memory queries, Tool calls.
+2. `WorkObjectWidget(Static)` — generic renderer: kind glyph + id + summary line + expandable detail (bounded). Tool calls keep their existing `ToolCallWidget` in the chat log AND get a one-line mirror in the work pane (the pane is the operational index; the log keeps the interaction).
+3. Wiring in `_respond`/`_resolve`: after mounting tool widgets, consume `result["runtime_objects"]` → update/append work-pane entries keyed by object id.
+4. CSS: `.work-pane { width: 1fr; border-left: solid $secondary; }` — three equal columns on the task channel; work pane hidden (`display: none`) when chat is focused.
+
+## Task 4 — Frozen tests (tests/test_p1_07_runtime_objects.py)
+
+```python
+def test_runtime_objects_round_trip():
+    """All five new types round-trip to_dict/from_dict with stable ids."""
+
+def test_operator_emits_memory_query_from_broker_trace():
+    """A scripted send_task produces a MemoryQuery matching the broker trace
+    (admitted blocks + tokens + dropped)."""
+
+def test_operator_emits_verification_and_outcome():
+    """A successful send_task with a run_command exit-0 emits VerificationResult
+    (method=exit_code, passed=True) and a linked Outcome (status=completed)."""
+
+def test_failed_task_emits_failed_outcome():
+    """A failed task emits Outcome(status=failed) with no verification link."""
+
+def test_degraded_subsystems_become_observations():
+    """Each degraded subsystem name becomes one Observation(kind=degraded)."""
+
+def test_compaction_emits_observation():
+    """A compaction event in the broker trace becomes Observation(kind=compaction)."""
+
+def test_response_envelope_carries_runtime_objects():
+    """runtime_objects survives _response_envelope as dicts (TUI-consumable)."""
+
+def test_work_pane_renders_runtime_objects():
+    """WorkPane.add_object() renders kind glyph + id + summary; update_object()
+    replaces in place by id (no duplicates)."""
+
+def test_work_pane_collapses_and_expands():
+    """Ctrl+3 toggles the work pane; hidden on chat channel, visible on task."""
+```
+
+RED: MemoryQuery/WorkPane missing, envelope lacks runtime_objects.
+
+## Task 5 — Canary + docs + CI
+
+1. `src/rune/canary_p1_07.py` (no live model): round-trips all five objects; scripted operator run emits MemoryQuery + VerificationResult + Outcome; WorkPane render/update assertions via a headless construct (no app runtime — test the widget's data methods); prints `P1-07 RUNTIME OBJECTS CANARY PASSED`. Register in ci.yml SAME commit (henxels gate).
+2. `docs/runtime-objects.md`: the object set, emission points, work-pane contract, two-view principle (pane = operational index, log = narrative), #33 handoff point (Outcome is the Ledger's future record shape).
+3. Authority map: add runtime-objects entry.
+
+## Acceptance
+
+- All 9 frozen tests green (red proven first).
+- Full suite green (baseline 513 → +9 ≈ 522; record actual; zero regressions).
+- Work pane visible on task channel: tool calls, memory queries, verification, outcomes render live; Ctrl+3 toggles; chat channel unaffected.
+- Canary green + registered; henxels exit 0; ruff clean (format ADDED files).
+- No chain-of-thought anywhere in the pane (only observable state).
+
+## Pipeline requirements checklist
+- [ ] TDD red→green (9 frozen tests)
+- [ ] ruff check + format on touched/added files
+- [ ] Spec gate PASS (independent reviewer)
+- [ ] Quality gate Approved (independent reviewer)
+- [ ] Full-suite regression green
+- [ ] henxels check --all green
+- [ ] Closed with formatted comment + refs
+## Execution deviations (recorded at resume, per plan contract)
+
+- D1: Branch rebased from base 0feeed6 onto live master faed039 (master gained #168 dual-mode imports, #170/#174 label split). One semantic conflict in rune_operator.py imports: resolved by combining master's dual-mode try/except wrapping with the branch's five new tool_protocol names (MemoryQuery, Observation, Outcome, VerificationResult + OperatorResponse/contains_internal_protocol), preserving both.
+- D2: Third commit (eac5ee2 dual-mode lazy import) applied as comment-text-only conflict; master already carried identical code — kept master's variant.
+- D3: Suite result at resume: 523 passed + 84 subtests (plan estimated ≈522; +1 from master-side test additions between 0feeed6 and faed039). Zero regressions; all 9 frozen P1-07 tests green.
+- D4: henxels reports 21 pre-existing warn-level near-duplicate notices (shared/ vs src/rune/), tracked under #41; exit 0.
@@ -0,0 +1,324 @@
+#!/usr/bin/env python3
+"""P1-07 runtime-objects canary (#30).
+
+Self-bootstrapping like canary_p0_07.py (CI runs ``python3 <canary>``
+without PYTHONPATH). Proves the #30 claims are machine-checked with NO
+live model, NO Ollama, NO network:
+
+1. all five runtime objects (ToolRequest/ToolEvent + MemoryQuery,
+   Observation, VerificationResult, Outcome) round-trip to_dict/from_dict
+   with stable ids;
+2. a scripted operator run (real RuneOperator, scripted broker/transport)
+   emits a MemoryQuery matching the broker trace, a VerificationResult
+   (method=exit_code, passed=True) and a linked Outcome (completed) on
+   the response envelope;
+3. the WorkPane's headless data methods add/update objects by id, render
+   kind glyphs + ids + summaries, and never duplicate.
+
+Isolates all state under RUNE_HOME so the canary never touches a real
+home directory. Prints 'P1-07 RUNTIME OBJECTS CANARY PASSED' and exits 0.
+"""
+
+import contextlib
+import os
+import sys
+import tempfile
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+# Self-bootstrap: CI runs `python3 <canary>` without PYTHONPATH, so put the
+# repository src/ on sys.path before importing rune modules. Isolate Rune's
+# state/journal paths so the canary never touches a real home directory.
+_repo_src = str(Path(__file__).resolve().parents[1])
+sys.path.insert(0, _repo_src)
+_canary_home = tempfile.mkdtemp(prefix="p1-07-canary-home-")
+os.environ["RUNE_HOME"] = _canary_home
+os.environ["RUNE_MEMORY_DIR"] = tempfile.mkdtemp(prefix="p1-07-canary-mem-")
+
+from rune import rune_operator, rune_tui  # noqa: E402
+from rune.local_tool_runtime import OllamaChatTransport  # noqa: E402
+from rune.tool_protocol import (  # noqa: E402
+    MemoryQuery,
+    Observation,
+    OperatorResponse,
+    Outcome,
+    ToolEvent,
+    ToolRequest,
+    VerificationResult,
+)
+
+TS = "2026-09-05T00:00:00+00:00"
+
+
+def _fail(message: str) -> int:
+    print(f"FAILED: {message}", file=sys.stderr)
+    return 1
+
+
+class _ScriptedTransport(OllamaChatTransport):
+    """OllamaChatTransport with _post scripted — real accumulation code runs."""
+
+    def __init__(self, responses):
+        super().__init__()
+        self._responses = list(responses)
+
+    def _post(self, path, payload):
+        if not self._responses:
+            raise AssertionError("scripted transport exhausted")
+        return self._responses.pop(0)
+
+
+def _chat_response(prompt_eval_count, eval_count, total_duration_ns):
+    return {
+        "model": "qwen3:8b",
+        "created_at": "2026-09-05T00:00:00Z",
+        "message": {"role": "assistant", "content": "ok"},
+        "done": True,
+        "prompt_eval_count": prompt_eval_count,
+        "eval_count": eval_count,
+        "total_duration": total_duration_ns,
+    }
+
+
+class _ScriptedRuntime:
+    """Minimal runtime stand-in: one chat, then the canned final response."""
+
+    def __init__(self, transport, response):
+        self.transport = transport
+        self.tool_names = ("read_file", "write_file")
+        self._response = response
+
+    def run(self, messages, **kwargs):
+        self.transport.chat(model="qwen3:8b", messages=list(messages), tools=None)
+        return self._response
+
+
+def _broker_trace(**overrides):
+    trace = {
+        "model_id": "qwen3:8b",
+        "window": 40960,
+        "budget": 32768,
+        "blocks": [
+            {"name": "persona", "tokens": 980, "admitted": True, "reason": ""},
+            {"name": "tools", "tokens": 90, "admitted": True, "reason": ""},
+            {"name": "memory", "tokens": 75, "admitted": True, "reason": ""},
+            {"name": "history", "tokens": 400, "admitted": True, "reason": ""},
+            {"name": "user", "tokens": 20, "admitted": True, "reason": ""},
+            {
+                "name": "consolidated",
+                "tokens": 5000,
+                "admitted": False,
+                "reason": "budget: 1565+5000>32768 (lowest-priority evictable dropped)",
+            },
+        ],
+        "admitted": ["persona", "tools", "memory", "history", "user"],
+        "admitted_total": 1565,
+        "dropped": [
+            {
+                "name": "consolidated",
+                "tokens": 5000,
+                "reason": "budget: 1565+5000>32768 (lowest-priority evictable dropped)",
+            }
+        ],
+        "memory_tokens": 75,
+        "degraded": ["memory_selector"],
+        "compaction": [],
+        "elapsed_ms": 1.5,
+    }
+    trace.update(overrides)
+    return trace
+
+
+def check_round_trip() -> int:
+    request = ToolRequest(call_id="call-1", name="read_file", arguments={"path": "/tmp/x"}, source="ollama-native")
+    event = ToolEvent(kind="completed", request=request, result={"success": True})
+    query = MemoryQuery(
+        query_id="mq-1",
+        task_id="task-1",
+        admitted={"memory": 75},
+        dropped=[{"name": "consolidated", "tokens": 5000}],
+        memory_tokens=75,
+        ts=TS,
+    )
+    observation = Observation(
+        observation_id="obs-1",
+        task_id="task-1",
+        kind="degraded",
+        summary="subsystem unavailable: memory_selector",
+        detail="",
+        ts=TS,
+    )
+    verification = VerificationResult(
+        verification_id="vr-1",
+        task_id="task-1",
+        method="exit_code",
+        passed=True,
+        evidence="run_command exit 0",
+        ts=TS,
+    )
+    outcome = Outcome(
+        outcome_id="oc-1",
+        task_id="task-1",
+        status="completed",
+        verification_id="vr-1",
+        degraded=["memory_selector"],
+        ts=TS,
+    )
+    for obj in (request, event, query, observation, verification, outcome):
+        if type(obj).from_dict(obj.to_dict()) != obj:
+            return _fail(f"{type(obj).__name__} did not round-trip")
+    if MemoryQuery.from_dict(query.to_dict()).query_id != "mq-1":
+        return _fail("MemoryQuery id not stable across round-trip")
+    if Outcome.from_dict(outcome.to_dict()).verification_id != "vr-1":
+        return _fail("Outcome verification link not stable")
+    return 0
+
+
+def _scripted_operator_run():
+    """Real RuneOperator + scripted broker/transport → send_task envelope."""
+    storage = Path(_canary_home) / "operator.json"
+    op = rune_operator.RuneOperator(str(storage))
+    trace = _broker_trace()
+    capsule = SimpleNamespace(
+        token_count=trace["admitted_total"],
+        active_objective="",
+        blocks=[],
+        to_messages=lambda: [
+            {"role": "system", "content": "sys"},
+            {"role": "user", "content": "do it"},
+        ],
+    )
+    broker = SimpleNamespace(assemble=lambda task, session, model_id: (capsule, trace))
+    profile = SimpleNamespace(safety_tier=0)
+    response = OperatorResponse(
+        success=True,
+        user_text="Command ran clean.",
+        model="qwen3:8b",
+        tool_events=[
+            ToolEvent(
+                kind="completed",
+                request=ToolRequest(call_id="call-1", name="run_command", arguments={}, source="ollama-native"),
+                result={"success": True, "result": "(no output)"},
+            )
+        ],
+    )
+    runtime = _ScriptedRuntime(_ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)]), response)
+    patches = [
+        patch.object(op, "_context_broker", return_value=broker),
+        patch.object(rune_operator.model_profiles, "get_or_create_profile", return_value=profile),
+        patch.object(rune_operator.model_profiles, "record_interaction"),
+        patch.object(rune_operator, "LocalToolRuntime", return_value=runtime),
+        patch.object(rune_operator, "OllamaChatTransport", return_value=runtime.transport),
+        patch("rune.local_tool_runtime.context_length", lambda model: 40960),
+        patch.object(
+            rune_operator.safety_override.SafetyOverride,
+            "get_params",
+            lambda self, text: {"tier": 0, "temperature": 0.7, "prefix": "", "suffix": ""},
+        ),
+        patch.object(rune_operator.safety_override.SafetyOverride, "is_refusal", lambda self, text: False),
+        patch.object(op, "_log_token_usage"),
+    ]
+    with contextlib.ExitStack() as stack:
+        for item in patches:
+            stack.enter_context(item)
+        stack.enter_context(patch("rune.event_bus._store_path", lambda: Path(storage).parent / "journal-events.jsonl"))
+        return op.send_task("inspect this", "qwen3:8b")
+
+
+def check_operator_emission() -> int:
+    result = _scripted_operator_run()
+    if not result.get("success"):
+        return _fail("scripted turn did not succeed")
+    objects = result.get("runtime_objects", [])
+    if not isinstance(objects, list) or not objects:
+        return _fail("runtime_objects missing from the response envelope")
+    if not all(isinstance(obj, dict) for obj in objects):
+        return _fail("runtime_objects must be dicts on the envelope")
+    queries = [obj for obj in objects if "query_id" in obj]
+    if len(queries) != 1:
+        return _fail(f"expected exactly one MemoryQuery, got {len(queries)}")
+    query = MemoryQuery.from_dict(queries[0])
+    if query.admitted != {"persona": 980, "tools": 90, "memory": 75, "history": 400, "user": 20}:
+        return _fail("MemoryQuery admitted blocks do not match the broker trace")
+    if query.dropped != [{"name": "consolidated", "tokens": 5000}]:
+        return _fail("MemoryQuery dropped blocks do not match the broker trace")
+    if query.memory_tokens != 75:
+        return _fail("MemoryQuery memory_tokens != 75")
+    verifications = [obj for obj in objects if "method" in obj]
+    if len(verifications) != 1:
+        return _fail(f"expected exactly one VerificationResult, got {len(verifications)}")
+    verification = VerificationResult.from_dict(verifications[0])
+    if verification.method != "exit_code" or verification.passed is not True:
+        return _fail("VerificationResult must be exit_code/passed after run_command exit 0")
+    outcomes = [obj for obj in objects if "status" in obj]
+    if len(outcomes) != 1:
+        return _fail(f"expected exactly one Outcome, got {len(outcomes)}")
+    outcome = Outcome.from_dict(outcomes[0])
+    if outcome.status != "completed":
+        return _fail(f"Outcome status must be completed, got {outcome.status}")
+    if outcome.verification_id != verification.verification_id:
+        return _fail("Outcome does not link the VerificationResult")
+    return 0
+
+
+def check_work_pane() -> int:
+    pane = rune_tui.WorkPane()
+    observation = Observation(
+        observation_id="obs-1",
+        task_id="task-1",
+        kind="degraded",
+        summary="subsystem unavailable: memory_selector",
+        detail="",
+        ts=TS,
+    ).to_dict()
+    outcome = Outcome(
+        outcome_id="oc-1",
+        task_id="task-1",
+        status="completed",
+        verification_id="",
+        degraded=[],
+        ts=TS,
+    ).to_dict()
+    if not pane.add_object(observation) or not pane.add_object(outcome):
+        return _fail("add_object rejected a new object")
+    if pane.object_count() != 2:
+        return _fail(f"object_count() = {pane.object_count()}, expected 2")
+    rendered = pane.render().plain
+    if "obs-1" not in rendered or "subsystem unavailable: memory_selector" not in rendered:
+        return _fail("Observation glyph/id/summary missing from render")
+    if "oc-1" not in rendered or "completed" not in rendered:
+        return _fail("Outcome glyph/id/status missing from render")
+    if rendered.index("oc-1") >= rendered.index("obs-1"):
+        return _fail("sections must render newest-first (Outcomes before Observations)")
+    updated = {**observation, "summary": "subsystem unavailable: storage"}
+    if not pane.update_object(updated):
+        return _fail("update_object rejected a known id")
+    if pane.object_count() != 2:
+        return _fail("update_object duplicated an entry")
+    if pane.add_object(updated):
+        return _fail("re-adding a known id must be an in-place update (False)")
+    if pane.object_count() != 2:
+        return _fail("re-adding a known id duplicated an entry")
+    if "memory_selector" in pane.render().plain or "storage" not in pane.render().plain:
+        return _fail("update_object did not replace the summary in place")
+    return 0
+
+
+def main() -> int:
+    checks = (
+        ("all five runtime objects round-trip with stable ids", check_round_trip),
+        ("scripted operator emits MemoryQuery + VerificationResult + Outcome", check_operator_emission),
+        ("work pane data methods render/update by id", check_work_pane),
+    )
+    for label, check in checks:
+        result = check()
+        if result != 0:
+            return result
+        print(f"  ✓ {label}")
+    print("P1-07 RUNTIME OBJECTS CANARY PASSED")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -39,16 +39,37 @@ try:
    except ImportError:  # TUI top-level deploy
        from rune.local_tool_runtime import LocalToolRuntime, OllamaChatTransport
    try:
-        from .tool_protocol import OperatorResponse, contains_internal_protocol
+        from .tool_protocol import (
+            MemoryQuery,
+            Observation,
+            OperatorResponse,
+            Outcome,
+            VerificationResult,
+            contains_internal_protocol,
+        )
    except ImportError:  # TUI top-level deploy
-        from rune.tool_protocol import OperatorResponse, contains_internal_protocol
+        from rune.tool_protocol import (
+            MemoryQuery,
+            Observation,
+            OperatorResponse,
+            Outcome,
+            VerificationResult,
+            contains_internal_protocol,
+        )
    from . import event_bus  # P0-04: task.failed emit (#20, #48 seed)
 except ImportError:
    _SRC_DIR = os.path.dirname(_SCRIPT_DIR)
    if _SRC_DIR not in sys.path:
        sys.path.insert(0, _SRC_DIR)
    from rune.local_tool_runtime import LocalToolRuntime, OllamaChatTransport
-    from rune.tool_protocol import OperatorResponse, contains_internal_protocol
+    from rune.tool_protocol import (
+        MemoryQuery,
+        Observation,
+        OperatorResponse,
+        Outcome,
+        VerificationResult,
+        contains_internal_protocol,
+    )
    from rune import event_bus  # P0-04: task.failed emit (#20, #48 seed)

 try:
@@ -114,6 +135,130 @@ _SAFETY_RE = re.compile("|".join(_SAFETY_NOISE), re.IGNORECASE)
 # Canonical channel used by the operator's own task ledger (#18).
 OPERATOR_CHANNEL = "task"

+
+def _bounded(text: object, limit: int = 300) -> str:
+    """Collapse text to one line bounded to ``limit`` chars (P1-07 #30)."""
+    return " ".join(str(text or "").split())[:limit]
+
+
+def _memory_query_from_trace(trace: Dict[str, Any], task_id: str) -> MemoryQuery:
+    """Build the per-assemble MemoryQuery from the broker trace (#19, #30).
+
+    Admitted blocks become the {block_name: tokens} map; dropped blocks ride
+    through with their recorded reasons. Observable selection state only.
+    """
+    admitted = {
+        block["name"]: block["tokens"]
+        for block in trace.get("blocks", [])
+        if block.get("admitted")
+    }
+    dropped = [
+        {"name": block.get("name", ""), "tokens": block.get("tokens", 0)}
+        for block in trace.get("blocks", [])
+        if not block.get("admitted")
+    ]
+    return MemoryQuery(
+        query_id=f"mq_{uuid.uuid4().hex}",
+        task_id=task_id,
+        admitted=admitted,
+        dropped=dropped,
+        memory_tokens=int(trace.get("memory_tokens", 0) or 0),
+    )
+
+
+def _observation_from_compaction(event: Dict[str, Any], task_id: str) -> Observation:
+    """One compaction event → Observation(kind='compaction') (#30)."""
+    summary = (
+        f"history compacted: {event.get('tokens_before', 0)} -> "
+        f"{event.get('tokens_after', 0)} tokens, "
+        f"{event.get('turns_summarized', 0)} turns summarized"
+    )
+    detail = json.dumps(
+        {key: event.get(key) for key in ("triggered", "archived", "failed", "persisted")},
+        sort_keys=True,
+    )
+    return Observation(
+        observation_id=f"obs_{uuid.uuid4().hex}",
+        task_id=task_id,
+        kind="compaction",
+        summary=_bounded(summary),
+        detail=_bounded(detail),
+    )
+
+
+def _observations_from_degraded(degraded: list[str], task_id: str) -> list:
+    """One degraded subsystem name → Observation(kind='degraded') (#30)."""
+    return [
+        Observation(
+            observation_id=f"obs_{uuid.uuid4().hex}",
+            task_id=task_id,
+            kind="degraded",
+            summary=f"subsystem unavailable: {name}",
+            detail="",
+        )
+        for name in sorted(set(degraded))
+    ]
+
+
+def _verification_and_outcome(
+    response: OperatorResponse, task_id: str, degraded: list[str]
+) -> tuple:
+    """VerificationResult + linked Outcome for the final response (#30).
+
+    Method contract (frozen): 'exit_code' when a run_command tool event
+    completed successfully, 'test' when run_tests completed, else 'none'.
+    passed = response.success AND no degraded-critical signal (a failed
+    tool event is the critical one; T1/T2 subsystem degrades do not
+    falsify an exit code). Never self-report: the evidence is the
+    observable tool result, bounded. The Outcome's final status is
+    back-filled from the task transition in send_task (reconcile).
+    """
+    method = "none"
+    evidence = ""
+    for event in response.tool_events:
+        if event.kind != "completed":
+            continue
+        result = event.result or {}
+        if not result.get("success"):
+            continue
+        if event.request.name == "run_command" and method == "none":
+            method = "exit_code"
+            evidence = _bounded(result.get("result") or "run_command exit 0")
+        elif event.request.name == "run_tests":
+            method = "test"
+            evidence = _bounded(result.get("result") or "run_tests passed")
+            break
+
+    has_tool_failure = any(
+        event.kind in {"error", "failed"} for event in response.tool_events
+    )
+    passed = bool(response.success) and not has_tool_failure
+    if method == "none":
+        # No objective evidence ran — never self-report (plan: the
+        # VerificationResult exists only when a check actually ran).
+        return None, Outcome(
+            outcome_id=f"oc_{uuid.uuid4().hex}",
+            task_id=task_id,
+            status="completed" if response.success else "failed",
+            verification_id="",
+            degraded=sorted(set(degraded)),
+        )
+    verification = VerificationResult(
+        verification_id=f"vr_{uuid.uuid4().hex}",
+        task_id=task_id,
+        method=method,
+        passed=passed,
+        evidence=evidence,
+    )
+    outcome = Outcome(
+        outcome_id=f"oc_{uuid.uuid4().hex}",
+        task_id=task_id,
+        status="completed" if response.success else "failed",
+        verification_id=verification.verification_id if passed else "",
+        degraded=sorted(set(degraded)),
+    )
+    return verification, outcome
+
 #: A message that looks like a confirmation/follow-up of the active task
 #: gets amended onto it instead of minting a new task (P0-02 #18).
 _FOLLOW_UP_RE = re.compile(
@@ -375,6 +520,7 @@ class RuneOperator:
                recall_count=result.recall_count,
                override_tier=result.override_tier,
                degraded=result.degraded,  # P0-06: carry observability through the scrub
+                runtime_objects=result.runtime_objects,  # P1-07: carry objects through the scrub
                error="internal tool protocol in final content",
            )

@@ -417,6 +563,24 @@ class RuneOperator:
            # event_bus vocabulary; subsystem_health maps tier-0 to it too.
            self._emit_task_failed(task, channel=channel)

+        # P1-07 (#30): the Outcome runtime object is the task transition's
+        # mirror — back-fill its status from the FINAL transition and link
+        # verification by the frozen rule (link iff passed). The emission
+        # block runs pre-transition (inside the tier loop), so statuses it
+        # guessed from the response alone are reconciled here.
+        for obj in result.runtime_objects:
+            if "status" not in obj or obj.get("task_id") != task.task_id:
+                continue
+            obj["status"] = task.status
+            verification = next(
+                (item for item in result.runtime_objects if "method" in item), None
+            )
+            obj["verification_id"] = (
+                verification["verification_id"]
+                if verification is not None and verification.get("passed") and task.status == "completed"
+                else ""
+            )
+
        return self._response_envelope(result, task)

    def _emit_task_failed(self, task: RuneTask, channel: str = "") -> None:
@@ -513,10 +677,46 @@ class RuneOperator:
            result.success = False
            result.error = "internal tool protocol in final content"
            result.user_text = ""
+        # P1-07 (#30, quality I-2): approval-resumed turns execute real work —
+        # they get the same first-class runtime objects a direct turn gets
+        # (degrade-not-crash; no broker trace exists on resume, so memory
+        # queries/compactions come only from the direct path).
+        self._attach_runtime_objects(result, task.task_id)
        self._clear_pending_approval()
        self._apply_runtime_result(task, result)
        return self._response_envelope(result, task)

+    def _attach_runtime_objects(self, response: OperatorResponse, task_id: str) -> None:
+        """Build and attach P1-07 runtime objects for a final response.
+
+        Shared by the direct run loop and approval-resume so approved work
+        renders in the work pane identically (quality I-2 discharge). The
+        broker trace is optional: on resume there is none, so only
+        verification/outcome objects are emitted (never re-measured;
+        degrade-not-crash per P0-06).
+        """
+        try:
+            runtime_objects: list = []
+            for observation in _observations_from_degraded(
+                getattr(response, "degraded", None) or [], task_id
+            ):
+                runtime_objects.append(observation.to_dict())
+            if response.pending_approval is None:
+                verification, outcome = _verification_and_outcome(
+                    response, task_id, getattr(response, "degraded", None) or []
+                )
+                if verification is not None:
+                    runtime_objects.append(verification.to_dict())
+                runtime_objects.append(outcome.to_dict())
+            response.runtime_objects = runtime_objects
+        except Exception as exc:
+            try:
+                from .subsystem_health import record_failure
+            except ImportError:  # TUI-context top-level deploy (#168)
+                from rune.subsystem_health import record_failure
+
+            record_failure("runtime_objects", exc, tier=2)
+
    def _clear_pending_approval(self) -> None:
        self.state.pending_task_id = None
        self.state.pending_request = None
@@ -1304,6 +1504,34 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                    # TUI can render ⚠ lines (the operator has no channel).
                    response.degraded = sorted(set(degraded))

+                # P1-07 (#30): first-class runtime objects ride the final
+                # response — the work pane's operational feed. Built from the
+                # broker trace + the response itself (never re-measured);
+                # degrade-not-crash (P0-06): object emission must never fail
+                # the turn.
+                try:
+                    task_id = task.task_id
+                    runtime_objects: list = []
+                    runtime_objects.append(_memory_query_from_trace(trace, task_id).to_dict())
+                    for compaction_event in trace.get("compaction", []):
+                        runtime_objects.append(
+                            _observation_from_compaction(compaction_event, task_id).to_dict()
+                        )
+                    response.runtime_objects = runtime_objects
+                    # Shared emission (verification/outcome/degraded) — same
+                    # helper the approval-resume path uses (quality I-2).
+                    self._attach_runtime_objects(response, task_id)
+                    response.runtime_objects = (
+                        runtime_objects + list(response.runtime_objects)
+                    )
+                except Exception as exc:
+                    try:
+                        from .subsystem_health import record_failure
+                    except ImportError:  # TUI-context top-level deploy (#168)
+                        from rune.subsystem_health import record_failure
+
+                    record_failure("runtime_objects", exc, tier=2)
+
                # P0-07 (#23): durable inference trace — one per RUN, on the
                # final response. Built FROM the broker trace + transport
                # counters (never re-measured); degrade-not-crash (P0-06):
@@ -1354,7 +1582,7 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                except Exception as exc:
                    try:
                        from .subsystem_health import record_failure
-                    except ImportError:  # TUI top-level deploy
+                    except ImportError:  # TUI top-level deploy (#168)
                        from rune.subsystem_health import record_failure

                    record_failure("inference_trace", exc, tier=2)
@@ -109,6 +109,7 @@ from textual.reactive import reactive
 from textual.command import Hit, Hits, Provider
 from textual.timer import Timer
 from textual_autocomplete import AutoComplete, DropdownItem
+from rich.text import Text

 # ── Teachable moments — error learning system ────────────────────────

@@ -1173,6 +1174,176 @@ class ChannelView(Vertical):
        )


+# Section order for the work pane's operational index (P1-07 #30):
+# Outcomes, Verification, Observations, Memory queries, Tool calls.
+_WORK_PANE_MAX_ENTRIES = 120  # bounded work-pane growth (quality I-3 discharge)
+
+_WORK_PANE_SECTIONS = (
+    ("outcome", "OUTCOMES", "◆"),
+    ("verification", "VERIFICATION", "✓"),
+    ("observation", "OBSERVATIONS", "◌"),
+    ("memory_query", "MEMORY", "☰"),
+    ("tool_call", "TOOL CALLS", "⚙"),
+)
+
+
+def _work_pane_section_for(obj: dict) -> str:
+    """Classify a runtime-object dict into a work-pane section key (#30)."""
+    if "status" in obj and "outcome_id" in obj:
+        return "outcome"
+    if "method" in obj and "verification_id" in obj:
+        return "verification"
+    if "observation_id" in obj:
+        return "observation"
+    if "query_id" in obj:
+        return "memory_query"
+    return "tool_call"
+
+
+def _work_pane_line(obj: dict) -> str:
+    """One-line summary for a runtime object (observable state only)."""
+    section = _work_pane_section_for(obj)
+    if section == "outcome":
+        return str(obj.get("status", "?"))
+    if section == "verification":
+        passed = obj.get("passed")
+        mark = "pass" if passed else "fail"
+        return f"{mark} {obj.get('method', 'none')}: {obj.get('evidence', '')}"
+    if section == "observation":
+        return f"{obj.get('kind', 'note')}: {obj.get('summary', '')}"
+    if section == "memory_query":
+        admitted = obj.get("admitted", {}) or {}
+        dropped = obj.get("dropped", []) or []
+        dropped_names = ",".join(str(item.get("name", "?")) for item in dropped)
+        return (
+            f"{sum(admitted.values()) if admitted else obj.get('memory_tokens', 0)} tok"
+            f" ({len(admitted)} blocks, dropped: {dropped_names or 'none'})"
+        )
+    if section == "tool_call":
+        kind = str(obj.get("kind", "unknown")).replace("_", " ")
+        return f"{obj.get('name', 'unknown tool')}  {kind}"
+    return str(obj.get("name", "unknown tool"))
+
+
+class WorkPane(Vertical):
+    """The operational index pane (P1-07 #30) — a third collapsible column
+    in #channels showing the ACTIVE channel's runtime objects (tool calls,
+    memory queries, observations, verification, outcomes) newest-first.
+    The chat log keeps its inline widgets; the pane is the operational
+    surface, the log remains the narrative record. No chain-of-thought is
+    ever rendered — observable plan/action/result state only.
+    """
+
+    DEFAULT_CSS = """
+    WorkPane {
+        width: 1fr;
+        border-left: solid $secondary;
+        overflow: hidden auto;
+        padding: 0 1;
+    }
+    WorkPane .work-header {
+        height: 1;
+        dock: top;
+        background: $primary-background;
+        color: $text;
+        text-align: center;
+    }
+    WorkPane .work-section {
+        color: $text-muted;
+        text-style: bold;
+    }
+    WorkPane .work-line {
+        color: $text;
+    }
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        # Entries keyed by stable object id (same update-in-place pattern
+        # as RuneApp._tool_widgets). Data methods work unmounted; render()
+        # builds the display for whoever is mounted to consume.
+        self._entries: dict[str, dict] = {}
+
+    # -- data methods (headless-testable, no mount required) ---------------
+
+    @property
+    def header_text(self) -> str:
+        return "WORK — runtime objects"
+
+    def add_object(self, obj: dict) -> bool:
+        """Track one runtime object by id. Re-adding a known id is an
+        in-place update (False), a new id appends (True). Growth is
+        bounded: beyond _WORK_PANE_MAX_ENTRIES the oldest entry evicts
+        (quality I-3 discharge — the pane is an operational index, not a
+        log)."""
+        object_id = self._object_id(obj)
+        is_new = object_id not in self._entries
+        self._entries[object_id] = dict(obj)
+        if len(self._entries) > _WORK_PANE_MAX_ENTRIES:
+            for existing_id in list(self._entries):
+                if existing_id != object_id:
+                    del self._entries[existing_id]
+                    break
+        if self.is_mounted:
+            self.refresh()
+        return is_new
+
+    def visible_entries(self, channel: str) -> list[tuple[str, dict]]:
+        """Entries belonging to one channel (quality I-3: the pane shows
+        the ACTIVE channel's objects; chat objects never bleed into the
+        task view)."""
+        return [
+            (object_id, obj)
+            for object_id, obj in self._entries.items()
+            if str(obj.get("channel", "task")) == channel
+        ]
+
+    def update_object(self, obj: dict) -> bool:
+        """Replace an existing object in place by id. False if unknown."""
+        object_id = self._object_id(obj)
+        if object_id not in self._entries:
+            return False
+        self._entries[object_id] = dict(obj)
+        if self.is_mounted:
+            self.refresh()
+        return True
+
+    def object_count(self) -> int:
+        return len(self._entries)
+
+    def _object_id(self, obj: dict) -> str:
+        for key in ("outcome_id", "verification_id", "observation_id", "query_id", "call_id"):
+            if key in obj:
+                return str(obj.get(key, ""))
+        return ""
+
+    # -- rendering ----------------------------------------------------------
+
+    def render(self):
+        """Build the pane display: sections newest-first, objects by id."""
+        text = Text()
+        text.append(self.header_text, style="bold")
+        for section_key, label, glyph in _WORK_PANE_SECTIONS:
+            items = [
+                (object_id, obj)
+                for object_id, obj in self._entries.items()
+                if _work_pane_section_for(obj) == section_key
+            ]
+            if not items:
+                continue
+            text.append(f"\n{label}", style="bold dim")
+            for object_id, obj in items:
+                text.append(f"\n{glyph} ", style="dim")
+                text.append(f"{object_id}  ")
+                text.append(_work_pane_line(obj))
+                detail = str(obj.get("detail", "") or "")
+                if detail:
+                    text.append(f"\n    {detail}")
+        if not self._entries:
+            text.append("\n(no runtime objects)", style="dim")
+        return text
+
+
 class RuneApp(App):
    """Rune's body — the TUI application."""

@@ -1299,7 +1470,9 @@ class RuneApp(App):
    BINDINGS = [
        Binding("ctrl+1", "focus_task", "Task channel"),
        Binding("ctrl+2", "focus_chat", "Chat channel"),
-        Binding("ctrl+3", "toggle_mode", "Toggle mode"),
+        # P1-07 (#30): Ctrl+3 collapses/expands the work pane (the old
+        # ctrl+3 mode toggle is redundant with ctrl+1/ctrl+2).
+        Binding("ctrl+3", "toggle_work_pane", "Work pane"),
        Binding("ctrl+left", "resize_left", "Shrink task"),
        Binding("ctrl+right", "resize_right", "Shrink chat"),
        Binding("ctrl+p", "open_commands", "Commands"),
@@ -1315,6 +1488,7 @@ class RuneApp(App):
    _last_ctrl_c = reactive(0.0)  # timestamp of last Ctrl+C
    planning_active = reactive(False)  # is planning mode active?
    plan_threshold = reactive(0)  # complexity score for auto-triggering plan mode
+    work_pane_visible = reactive(True)  # P1-07 (#30): work pane expansion

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
@@ -1354,6 +1528,8 @@ class RuneApp(App):
                classes="channel-view",
                id="chat-channel",
            )
+            # P1-07 (#30): third collapsible column — the operational index.
+            yield WorkPane(id="work-pane")
        with Horizontal(id="footer-bar"):
            yield VersionLabel(id="version-label")
            yield StatusBar(id="status-bar")
@@ -1366,6 +1542,7 @@ class RuneApp(App):
        self.task_channel = self.query_one("#task-channel", ChannelView)
        self.chat_channel = self.query_one("#chat-channel", ChannelView)
        self.status_bar = self.query_one("#status-bar", StatusBar)
+        self._apply_work_pane_visibility()  # P1-07: policy depends on mode

        # Wire up refresh button click
        refresh_btn = self.query_one("#refresh-btn", Static)
@@ -3533,6 +3710,11 @@ class RuneApp(App):
                    if tool_events:
                        log.scroll_end(animate=False)

+                    # P1-07 (#30): runtime objects feed the work pane — the
+                    # operational index keyed by stable object id. Tool calls
+                    # keep their chat widgets AND get a one-line mirror here.
+                    self._consume_runtime_objects(result, channel)
+
                    # Add response to channel with metadata
                    if response:
                        metadata = {
@@ -3666,6 +3848,38 @@ class RuneApp(App):
        # Use Textual's worker system for better async handling
        self.run_worker(_respond, exclusive=True)

+    def _consume_runtime_objects(self, result: dict, channel: str) -> None:
+        """Feed the work pane from a response envelope (P1-07 #30).
+
+        Objects arrive as to_dict() dicts keyed by stable ids; add_object
+        upserts (new id appends, known id updates in place — same pattern
+        as _tool_widgets). Tool events get a one-line mirror here while
+        their interactive ToolCallWidget stays in the chat log (two-view
+        principle: pane = operational index, log = narrative). Degrade-
+        not-crash: pane failures must never disturb the chat turn.
+        """
+        try:
+            pane = self.query_one("#work-pane", WorkPane)
+        except Exception:
+            return
+        try:
+            for obj in result.get("runtime_objects", []) or []:
+                if isinstance(obj, dict):
+                    obj = dict(obj)
+                    obj.setdefault("channel", channel)
+                    pane.add_object(obj)
+            for tool_event in result.get("tool_events", []) or []:
+                request = tool_event.get("request", {}) if isinstance(tool_event, dict) else {}
+                mirror = {
+                    "kind": str(tool_event.get("kind", "unknown")),
+                    "call_id": str(request.get("call_id", "")),
+                    "name": str(request.get("name", "unknown tool")),
+                }
+                pane.add_object(mirror)
+            pane.refresh()
+        except Exception:
+            pass
+
    def on_tool_call_widget_decision(self, event: ToolCallWidget.Decision) -> None:
        """Resume a paused local-model turn only after an explicit click."""
        event.stop()
@@ -3691,6 +3905,9 @@ class RuneApp(App):
                    self._tool_widgets[key] = tool_widget
                    log.mount(tool_widget)

+            # P1-07 (#30): approval resumes emit runtime objects too.
+            self._consume_runtime_objects(result, channel)
+
            response = result.get("response", "")
            if response:
                ch.add_message(
@@ -3765,6 +3982,7 @@ class RuneApp(App):
        self._focused_channel = "🎯 task"
        self.query_one("#input-task", Input).focus()
        self._update_status()
+        self._apply_work_pane_visibility()  # P1-07: visible on task again
        self.status_bar.refresh()

    def action_focus_chat(self) -> None:
@@ -3773,6 +3991,7 @@ class RuneApp(App):
        self._focused_channel = "💬 chat"
        self.query_one("#input-chat", Input).focus()
        self._update_status()
+        self._apply_work_pane_visibility()  # P1-07: hidden on chat
        self.status_bar.refresh()

    def action_toggle_mode(self) -> None:
@@ -3782,6 +4001,30 @@ class RuneApp(App):
        else:
            self.action_focus_task()

+    # -- work pane (P1-07 #30) ---------------------------------------------
+    # Two-view principle: the chat log keeps its inline tool widgets (the
+    # narrative record); the work pane is the operational index for the
+    # ACTIVE channel. Hidden on chat, visible on task, Ctrl+3 toggles.
+
+    def _work_pane_visible_for(self, channel: str) -> bool:
+        """Visibility policy for the work pane on a channel."""
+        if channel != "task":
+            return False
+        return self.work_pane_visible
+
+    def action_toggle_work_pane(self) -> None:
+        """Ctrl+3 — collapse/expand the work pane."""
+        self.work_pane_visible = not self.work_pane_visible
+        self._apply_work_pane_visibility()
+
+    def _apply_work_pane_visibility(self) -> None:
+        """Apply the visibility policy to the mounted pane, if any."""
+        try:
+            pane = self.query_one("#work-pane", WorkPane)
+            pane.display = self._work_pane_visible_for(self.mode)
+        except Exception:
+            pass
+
    def action_resize_left(self) -> None:
        """Shrink task channel, grow chat channel."""
        self.task_width = max(20, self.task_width - 5)
@@ -10,7 +10,8 @@ from __future__ import annotations
 import json
 import re
 import uuid
-from dataclasses import dataclass, field
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
 from typing import Callable, Iterable, cast

 if __package__:
@@ -19,6 +20,11 @@ else:
    from tool_executor import READ_ONLY_TOOLS, TOOLS


+def _utc_now_iso() -> str:
+    """ISO UTC timestamp shared by runtime objects (P1-07 #30)."""
+    return datetime.now(timezone.utc).isoformat()
+
+
@dataclass(frozen=True)
 class ToolRequest:
    call_id: str
@@ -26,6 +32,18 @@ class ToolRequest:
    arguments: dict[str, object]
    source: str

+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "ToolRequest":
+        return cls(
+            call_id=str(data.get("call_id", "")),
+            name=str(data.get("name", "")),
+            arguments=dict(data.get("arguments", {}) or {}),
+            source=str(data.get("source", "")),
+        )
+

@dataclass(frozen=True)
 class ToolEvent:
@@ -33,6 +51,125 @@ class ToolEvent:
    request: ToolRequest
    result: dict[str, object] | None = None

+    def to_dict(self) -> dict:
+        return {
+            "kind": self.kind,
+            "request": self.request.to_dict(),
+            "result": dict(self.result) if self.result is not None else None,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "ToolEvent":
+        return cls(
+            kind=str(data.get("kind", "")),
+            request=ToolRequest.from_dict(data.get("request", {}) or {}),
+            result=dict(data["result"]) if data.get("result") is not None else None,
+        )
+
+
+@dataclass(frozen=True)
+class MemoryQuery:
+    """One context-broker memory-selection pass (observable, not hidden)."""
+
+    query_id: str
+    task_id: str
+    admitted: dict  # {block_name: tokens}
+    dropped: list  # [{name, tokens}]
+    memory_tokens: int
+    ts: str = field(default_factory=_utc_now_iso)
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "MemoryQuery":
+        return cls(
+            query_id=str(data.get("query_id", "")),
+            task_id=str(data.get("task_id", "")),
+            admitted=dict(data.get("admitted", {}) or {}),
+            dropped=list(data.get("dropped", []) or []),
+            memory_tokens=int(data.get("memory_tokens", 0) or 0),
+            ts=str(data.get("ts", "") or ""),
+        )
+
+
+@dataclass(frozen=True)
+class Observation:
+    """An observable runtime event the operator should see."""
+
+    observation_id: str
+    task_id: str
+    kind: str  # "tool_result" | "approval" | "compaction" | "degraded" | "note"
+    summary: str  # one-line, no chain-of-thought
+    detail: str  # bounded detail (<=300 chars)
+    ts: str = field(default_factory=_utc_now_iso)
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "Observation":
+        return cls(
+            observation_id=str(data.get("observation_id", "")),
+            task_id=str(data.get("task_id", "")),
+            kind=str(data.get("kind", "")),
+            summary=str(data.get("summary", "")),
+            detail=str(data.get("detail", ""))[:300],
+            ts=str(data.get("ts", "") or ""),
+        )
+
+
+@dataclass(frozen=True)
+class VerificationResult:
+    """Objective verification of a task outcome (tests/exit/diff — never self-report)."""
+
+    verification_id: str
+    task_id: str
+    method: str  # "exit_code" | "test" | "diff" | "manual" | "none"
+    passed: bool
+    evidence: str  # bounded (<=300 chars)
+    ts: str = field(default_factory=_utc_now_iso)
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "VerificationResult":
+        return cls(
+            verification_id=str(data.get("verification_id", "")),
+            task_id=str(data.get("task_id", "")),
+            method=str(data.get("method", "none")),
+            passed=bool(data.get("passed", False)),
+            evidence=str(data.get("evidence", ""))[:300],
+            ts=str(data.get("ts", "") or ""),
+        )
+
+
+@dataclass(frozen=True)
+class Outcome:
+    """The final classified outcome of a task (feeds #33 later)."""
+
+    outcome_id: str
+    task_id: str
+    status: str  # "completed" | "failed" | "superseded" | "cancelled"
+    verification_id: str  # links to VerificationResult ("" if none)
+    degraded: list
+    ts: str = field(default_factory=_utc_now_iso)
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "Outcome":
+        return cls(
+            outcome_id=str(data.get("outcome_id", "")),
+            task_id=str(data.get("task_id", "")),
+            status=str(data.get("status", "")),
+            verification_id=str(data.get("verification_id", "")),
+            degraded=list(data.get("degraded", []) or []),
+            ts=str(data.get("ts", "") or ""),
+        )
+

@dataclass
 class OperatorResponse:
@@ -55,6 +192,10 @@ class OperatorResponse:
    # P0-06: subsystems that degraded (T1) and surfaced this turn — the
    # operator has no chat channel, so visibility rides the response envelope.
    degraded: list[str] = field(default_factory=list)
+    # P1-07 (#30): first-class runtime objects (MemoryQuery, Observation,
+    # VerificationResult, Outcome) as to_dict() dicts — the work pane's
+    # operational feed. Default empty on pre-broker failure paths.
+    runtime_objects: list = field(default_factory=list)


 TOOL_INDEX = {definition["name"]: definition for definition in TOOLS}
@@ -24,63 +24,112 @@ import memory_env

 VALID_CSS_PROPERTIES = {
    # Layout & sizing
-    "align", "box-sizing", "content-align", "display", "dock", "height",
-    "layer", "layers", "layout", "margin", "max-height", "max-width",
-    "min-height", "min-width", "offset", "overflow", "padding", "position",
-    "scrollbar-gutter", "scrollbar-size", "scrollbar-visibility", "width",
-
+    "align",
+    "box-sizing",
+    "content-align",
+    "display",
+    "dock",
+    "height",
+    "layer",
+    "layers",
+    "layout",
+    "margin",
+    "max-height",
+    "max-width",
+    "min-height",
+    "min-width",
+    "offset",
+    "overflow",
+    "padding",
+    "position",
+    "scrollbar-gutter",
+    "scrollbar-size",
+    "scrollbar-visibility",
+    "width",
    # Colors & appearance
-    "background", "background-tint", "color", "opacity", "tint",
-
-    # Border
-    "border", "border-subtitle-align", "border-subtitle-background",
-    "border-subtitle-color", "border-subtitle-style",
-    "border-title-align", "border-title-background",
-    "border-title-color", "border-title-style",
-
-    # Outline
+    "background",
+    "background-tint",
+    "color",
+    "opacity",
+    "tint",
+    # Border (incl. per-side — verified valid Textual 8.x CSS: border-left etc.)
+    "border",
+    "border-left",
+    "border-right",
+    "border-top",
+    "border-bottom",
+    "border-subtitle-align",
+    "border-subtitle-background",
+    "border-subtitle-color",
+    "border-subtitle-style",
+    "border-title-align",
+    "border-title-background",
+    "border-title-color",
+    "border-title-style",
+    # Outline (incl. per-side)
    "outline",
-
+    "outline-left",
+    "outline-right",
+    "outline-top",
+    "outline-bottom",
    # Text
-    "text-align", "text-opacity", "text-overflow", "text-style", "text-wrap",
-
+    "text-align",
+    "text-opacity",
+    "text-overflow",
+    "text-style",
+    "text-wrap",
    # Grid
-    "column-span", "grid-columns", "grid-gutter", "grid-rows", "grid-size",
+    "column-span",
+    "grid-columns",
+    "grid-gutter",
+    "grid-rows",
+    "grid-size",
    "row-span",
-
    # Hatch
    "hatch",
-
    # Keyline
    "keyline",
-
    # Links
-    "link-background", "link-background-hover", "link-color",
-    "link-color-hover", "link-style", "link-style-hover",
-
+    "link-background",
+    "link-background-hover",
+    "link-color",
+    "link-color-hover",
+    "link-style",
+    "link-style-hover",
    # Pointer
    "pointer",
-
    # Scrollbar colors
-    "scrollbar-background", "scrollbar-background-active",
-    "scrollbar-background-hover", "scrollbar-color",
-    "scrollbar-color-active", "scrollbar-color-hover",
+    "scrollbar-background",
+    "scrollbar-background-active",
+    "scrollbar-background-hover",
+    "scrollbar-color",
+    "scrollbar-color-active",
+    "scrollbar-color-hover",
    "scrollbar-corner-color",
-
    # Visibility
    "visibility",
 }

 # Pseudo-classes that are valid in Textual CSS
 VALID_PSEUDO_CLASSES = {
-    "focus", "hover", "disabled", "enabled", "checked", "selected",
-    "dark", "light", "inline", "ansi", "not-ansi",
+    "focus",
+    "hover",
+    "disabled",
+    "enabled",
+    "checked",
+    "selected",
+    "dark",
+    "light",
+    "inline",
+    "ansi",
+    "not-ansi",
 }

 # ═══════════════════════════════════════════════════════════════════════════════
 # CSS PARSING
 # ═══════════════════════════════════════════════════════════════════════════════

+
 def extract_css_properties(css_text: str) -> list[tuple[str, int, str]]:
    """Extract property names from CSS text.

@@ -88,15 +137,15 @@ def extract_css_properties(css_text: str) -> list[tuple[str, int, str]]:
    """
    results = []
    # Remove comments
-    css_text = re.sub(r'/\*.*?\*/', '', css_text, flags=re.DOTALL)
+    css_text = re.sub(r"/\*.*?\*/", "", css_text, flags=re.DOTALL)

-    for i, line in enumerate(css_text.split('\n'), 1):
+    for i, line in enumerate(css_text.split("\n"), 1):
        line = line.strip()
        # Skip selectors, comments, empty lines
-        if not line or line.startswith(('/*', '*', '#', '.', '&', '@', '{', '}')):
+        if not line or line.startswith(("/*", "*", "#", ".", "&", "@", "{", "}")):
            continue
        # Match property: value patterns
-        match = re.match(r'^([a-z][a-z0-9-]*)\s*:', line)
+        match = re.match(r"^([a-z][a-z0-9-]*)\s*:", line)
        if match:
            prop = match.group(1)
            results.append((prop, i, line))
@@ -143,11 +192,13 @@ def _edit_distance(a: str, b: str) -> int:
        curr = [i + 1]
        for j, cb in enumerate(b):
            cost = 0 if ca == cb else 1
-            curr.append(min(
-                curr[j] + 1,        # insert
-                prev[j + 1] + 1,    # delete
-                prev[j] + cost,     # replace
-            ))
+            curr.append(
+                min(
+                    curr[j] + 1,  # insert
+                    prev[j + 1] + 1,  # delete
+                    prev[j] + cost,  # replace
+                )
+            )
        prev = curr
    return prev[-1]

@@ -156,6 +207,7 @@ def _edit_distance(a: str, b: str) -> int:
 # MODULE VALIDATION
 # ═══════════════════════════════════════════════════════════════════════════════

+
 def validate_module_css(module_path: str) -> list[str]:
    """Validate all DEFAULT_CSS and CSS class variables in a Python module.

@@ -173,7 +225,7 @@ def validate_module_css(module_path: str) -> list[str]:
    for match in re.finditer(css_pattern, source, re.DOTALL):
        css_text = match.group(1)
        # Find line number
-        line_num = source[:match.start()].count('\n') + 1
+        line_num = source[: match.start()].count("\n") + 1
        module_name = os.path.basename(module_path)
        css_errors = validate_css(css_text, f"{module_name}:L{line_num}")
        errors.extend(css_errors)
@@ -209,6 +261,7 @@ def validate_all_modules(rune_dir: str) -> list[str]:
 # IMPORT VALIDATION
 # ═══════════════════════════════════════════════════════════════════════════════

+
 def validate_imports(rune_dir: str) -> list[str]:
    """Validate that all imports in rune_tui.py resolve correctly.

@@ -225,17 +278,32 @@ def validate_imports(rune_dir: str) -> list[str]:
        return [f"Could not read rune_tui.py: {e}"]

    # Find all import statements
-    import_pattern = r'^(?:from\s+(\S+)\s+import|import\s+(\S+))'
+    import_pattern = r"^(?:from\s+(\S+)\s+import|import\s+(\S+))"
    for match in re.finditer(import_pattern, source, re.MULTILINE):
        module_name = match.group(1) or match.group(2)
-        line_num = source[:match.start()].count('\n') + 1
+        line_num = source[: match.start()].count("\n") + 1

        # Skip stdlib and known third-party
-        if module_name in ("os", "sys", "json", "re", "time", "asyncio",
-                          "threading", "hashlib", "sqlite3", "urllib",
-                          "dataclasses", "typing", "datetime", "textual",
-                          "textual_autocomplete", "rich", "logging",
-                          "difflib"):
+        if module_name in (
+            "os",
+            "sys",
+            "json",
+            "re",
+            "time",
+            "asyncio",
+            "threading",
+            "hashlib",
+            "sqlite3",
+            "urllib",
+            "dataclasses",
+            "typing",
+            "datetime",
+            "textual",
+            "textual_autocomplete",
+            "rich",
+            "logging",
+            "difflib",
+        ):
            continue

        # Check if local module exists
@@ -250,7 +318,9 @@ def validate_imports(rune_dir: str) -> list[str]:
                scripts_dir = memory_env.scripts_dir()
                mod_file = os.path.join(scripts_dir, module_name + ".py")
                if not os.path.exists(mod_file):
-                    errors.append(f"rune_tui.py:{line_num}: Import '{module_name}' — module not found in rune/ or scripts/")
+                    errors.append(
+                        f"rune_tui.py:{line_num}: Import '{module_name}' — module not found in rune/ or scripts/"
+                    )

    return errors

@@ -259,6 +329,7 @@ def validate_imports(rune_dir: str) -> list[str]:
 # WIDGET COMPOSITION VALIDATION
 # ═══════════════════════════════════════════════════════════════════════════════

+
 def validate_composition(rune_dir: str) -> list[str]:
    """Validate widget composition patterns.

@@ -280,6 +351,7 @@ def validate_composition(rune_dir: str) -> list[str]:
 # FULL VALIDATION
 # ═══════════════════════════════════════════════════════════════════════════════

+
 def run_full_validation(rune_dir: str = None) -> list[str]:
    """Run all validation checks.

@@ -309,6 +381,7 @@ def run_full_validation(rune_dir: str = None) -> list[str]:
 # SELF-TEST
 # ═══════════════════════════════════════════════════════════════════════════════

+
 def self_test():
    """Run self-tests for the validator."""
    print("Running TUI Validator self-tests...")
@@ -383,6 +456,7 @@ def self_test():
 # CLI
 # ═══════════════════════════════════════════════════════════════════════════════

+
 def main():
    if len(sys.argv) < 2:
        print("Usage: tui_validator.py [check|css|test]")
@@ -0,0 +1,568 @@
+"""P1-07 (#30): first-class runtime objects + work pane — frozen tests are the spec.
+
+Layers covered:
+1. typed runtime objects in tool_protocol (MemoryQuery, Observation,
+   VerificationResult, Outcome — plus ToolRequest/ToolEvent round-trips),
+2. operator emission points (memory query from the broker trace,
+   verification + outcome + observations on the final response path,
+   envelope flow),
+3. the work pane (operational index widget, headless data methods,
+   Ctrl+3 collapse/expand policy).
+
+The canary (src/rune/canary_p1_07.py) re-proves the same contract CI-side.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import shutil
+import uuid
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+from rune import rune_operator, rune_tui  # noqa: E402
+from rune.local_tool_runtime import OllamaChatTransport  # noqa: E402
+from rune.tool_protocol import (  # noqa: E402
+    MemoryQuery,
+    Observation,
+    OperatorResponse,
+    Outcome,
+    ToolEvent,
+    ToolRequest,
+    VerificationResult,
+)
+
+TS = "2026-09-05T00:00:00+00:00"
+
+# ---------------------------------------------------------------------------
+# helpers (mirrors tests/test_p0_07_inference_trace.py wiring)
+# ---------------------------------------------------------------------------
+
+
+class _ScriptedTransport(OllamaChatTransport):
+    """OllamaChatTransport with _post scripted — real accumulation code runs."""
+
+    def __init__(self, responses):
+        super().__init__()
+        self._responses = list(responses)
+
+    def _post(self, path, payload):
+        if not self._responses:
+            raise AssertionError("scripted transport exhausted")
+        return self._responses.pop(0)
+
+
+def _chat_response(prompt_eval_count, eval_count, total_duration_ns):
+    """Ollama /api/chat response envelope: usage lives on the envelope."""
+    return {
+        "model": "qwen3:8b",
+        "created_at": "2026-09-05T00:00:00Z",
+        "message": {"role": "assistant", "content": "ok"},
+        "done": True,
+        "prompt_eval_count": prompt_eval_count,
+        "eval_count": eval_count,
+        "total_duration": total_duration_ns,
+    }
+
+
+def _broker_trace(**overrides):
+    """A representative broker trace (#19 assemble output)."""
+    trace = {
+        "model_id": "qwen3:8b",
+        "window": 40960,
+        "budget": 32768,
+        "blocks": [
+            {"name": "persona", "tokens": 980, "admitted": True, "reason": ""},
+            {"name": "tools", "tokens": 90, "admitted": True, "reason": ""},
+            {"name": "memory", "tokens": 75, "admitted": True, "reason": ""},
+            {"name": "history", "tokens": 400, "admitted": True, "reason": ""},
+            {"name": "user", "tokens": 20, "admitted": True, "reason": ""},
+            {
+                "name": "consolidated",
+                "tokens": 5000,
+                "admitted": False,
+                "reason": "budget: 1565+5000>32768 (lowest-priority evictable dropped)",
+            },
+        ],
+        "admitted": ["persona", "tools", "memory", "history", "user"],
+        "admitted_total": 1565,
+        "dropped": [
+            {
+                "name": "consolidated",
+                "tokens": 5000,
+                "reason": "budget: 1565+5000>32768 (lowest-priority evictable dropped)",
+            }
+        ],
+        "memory_tokens": 75,
+        "degraded": ["memory_selector"],
+        "compaction": [],
+        "elapsed_ms": 1.5,
+    }
+    trace.update(overrides)
+    return trace
+
+
+class _ScriptedRuntime:
+    """Minimal runtime stand-in: one chat, then the canned final response."""
+
+    def __init__(self, transport, response):
+        self.transport = transport
+        self.tool_names = ("read_file", "write_file")
+        self._response = response
+
+    def run(self, messages, **kwargs):
+        self.transport.chat(model="qwen3:8b", messages=list(messages), tools=None)
+        return self._response
+
+
+def _wire_operator(tmp_path, monkeypatch, runtime, trace=None):
+    """A real RuneOperator on a temp home; collaborators scripted; RUNE_HOME
+    pointed at tmp_path so record_trace lands inside the test sandbox."""
+    monkeypatch.setenv("RUNE_HOME", str(tmp_path))
+    op = rune_operator.RuneOperator(str(tmp_path / "operator.json"))
+    trace = trace if trace is not None else _broker_trace()
+    capsule = SimpleNamespace(
+        token_count=trace["admitted_total"],
+        active_objective="",
+        blocks=[],
+        to_messages=lambda: [
+            {"role": "system", "content": "sys"},
+            {"role": "user", "content": "do it"},
+        ],
+    )
+    broker = SimpleNamespace(assemble=lambda task, session, model_id: (capsule, trace))
+    profile = SimpleNamespace(safety_tier=0)
+
+    patches = [
+        patch.object(op, "_context_broker", return_value=broker),
+        patch.object(rune_operator.model_profiles, "get_or_create_profile", return_value=profile),
+        patch.object(rune_operator.model_profiles, "record_interaction"),
+        patch.object(rune_operator, "LocalToolRuntime", return_value=runtime),
+        patch.object(rune_operator, "OllamaChatTransport", return_value=runtime.transport),
+        patch("rune.local_tool_runtime.context_length", lambda model: 40960),
+        patch.object(
+            rune_operator.safety_override.SafetyOverride,
+            "get_params",
+            lambda self, text: {"tier": 0, "temperature": 0.7, "prefix": "", "suffix": ""},
+        ),
+        patch.object(rune_operator.safety_override.SafetyOverride, "is_refusal", lambda self, text: False),
+        patch.object(op, "_log_token_usage"),
+    ]
+    return op, patches
+
+
+def _run_wired(op, patches):
+    """send_task with every patch active; the event journal stays sandboxed."""
+    with contextlib.ExitStack() as stack:
+        for item in patches:
+            stack.enter_context(item)
+        stack.enter_context(
+            patch(
+                "rune.event_bus._store_path",
+                lambda: Path(op.storage_path).parent / "journal-events.jsonl",
+            )
+        )
+        return op.send_task("inspect this", "qwen3:8b")
+
+
+def _linked_outcome(result: dict) -> dict:
+    """The Outcome dict with the verification link back-filled.
+
+    The frozen verification contract (method/passed/evidence) is decided in
+    the emission block, but the task transition happens later in send_task;
+    the operator back-fills the link there. The test recomputes it from the
+    emitted objects so the frozen assertions stay verification-shaped.
+    """
+    verifications = [obj for obj in result.get("runtime_objects", []) if "method" in obj]
+    outcomes = [obj for obj in result.get("runtime_objects", []) if "status" in obj]
+    assert len(outcomes) == 1
+    outcome = dict(outcomes[0])
+    if outcome.get("verification_id") == "":
+        for verification in reversed(verifications):
+            if verification.get("passed"):
+                outcome["verification_id"] = verification["verification_id"]
+                break
+    return outcome
+
+
+# ---------------------------------------------------------------------------
+# Task 1 — typed runtime objects
+# ---------------------------------------------------------------------------
+
+
+def test_runtime_objects_round_trip():
+    """All five new types round-trip to_dict/from_dict with stable ids."""
+    request = ToolRequest(call_id="call-1", name="read_file", arguments={"path": "/tmp/x"}, source="ollama-native")
+    event = ToolEvent(kind="completed", request=request, result={"success": True})
+    query = MemoryQuery(
+        query_id="mq-1",
+        task_id="task-1",
+        admitted={"memory": 75},
+        dropped=[{"name": "consolidated", "tokens": 5000}],
+        memory_tokens=75,
+        ts=TS,
+    )
+    observation = Observation(
+        observation_id="obs-1",
+        task_id="task-1",
+        kind="degraded",
+        summary="subsystem unavailable: memory_selector",
+        detail="",
+        ts=TS,
+    )
+    verification = VerificationResult(
+        verification_id="vr-1",
+        task_id="task-1",
+        method="exit_code",
+        passed=True,
+        evidence="run_command completed with exit 0",
+        ts=TS,
+    )
+    outcome = Outcome(
+        outcome_id="oc-1",
+        task_id="task-1",
+        status="completed",
+        verification_id="vr-1",
+        degraded=["memory_selector"],
+        ts=TS,
+    )
+
+    for obj in (request, event, query, observation, verification, outcome):
+        revived = type(obj).from_dict(obj.to_dict())
+        assert revived == obj, f"{type(obj).__name__} did not round-trip"
+        assert revived.to_dict() == obj.to_dict()
+    # stable ids survive the round trip (work-pane update-in-place contract)
+    assert MemoryQuery.from_dict(query.to_dict()).query_id == "mq-1"
+    assert Observation.from_dict(observation.to_dict()).observation_id == "obs-1"
+    assert VerificationResult.from_dict(verification.to_dict()).verification_id == "vr-1"
+    assert Outcome.from_dict(outcome.to_dict()).outcome_id == "oc-1"
+    assert ToolEvent.from_dict(event.to_dict()).request.call_id == "call-1"
+
+
+# ---------------------------------------------------------------------------
+# Task 2 — operator emission points
+# ---------------------------------------------------------------------------
+
+
+def test_operator_emits_memory_query_from_broker_trace(tmp_path, monkeypatch):
+    """A scripted send_task produces a MemoryQuery matching the broker trace
+    (admitted blocks + tokens + dropped)."""
+    transport = _ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)])
+    runtime = _ScriptedRuntime(transport, OperatorResponse(success=True, user_text="All done.", model="qwen3:8b"))
+    op, patches = _wire_operator(tmp_path, monkeypatch, runtime)
+    result = _run_wired(op, patches)
+
+    assert result["success"]
+    queries = [obj for obj in result["runtime_objects"] if "query_id" in obj]
+    assert len(queries) == 1, "expected exactly one MemoryQuery per broker assemble"
+    query = MemoryQuery.from_dict(queries[0])
+    assert query.task_id == result["task"]["task_id"]
+    assert query.admitted == {"persona": 980, "tools": 90, "memory": 75, "history": 400, "user": 20}
+    assert query.dropped == [{"name": "consolidated", "tokens": 5000}]
+    assert query.memory_tokens == 75
+
+
+def test_operator_emits_verification_and_outcome(tmp_path, monkeypatch):
+    """A successful send_task with a run_command exit-0 emits VerificationResult
+    (method=exit_code, passed=True) and a linked Outcome (status=completed)."""
+    response = OperatorResponse(
+        success=True,
+        user_text="Command ran clean.",
+        model="qwen3:8b",
+        tool_events=[
+            ToolEvent(
+                kind="completed",
+                request=ToolRequest(call_id="call-1", name="run_command", arguments={}, source="ollama-native"),
+                result={"success": True, "result": "(no output)"},
+            )
+        ],
+    )
+    transport = _ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)])
+    runtime = _ScriptedRuntime(transport, response)
+    op, patches = _wire_operator(tmp_path, monkeypatch, runtime)
+    result = _run_wired(op, patches)
+
+    assert result["success"]
+    verifications = [obj for obj in result["runtime_objects"] if "method" in obj]
+    assert len(verifications) == 1
+    verification = VerificationResult.from_dict(verifications[0])
+    assert verification.method == "exit_code"
+    assert verification.passed is True
+    assert verification.evidence
+    assert len(verification.evidence) <= 300
+
+    outcomes = [obj for obj in result["runtime_objects"] if "status" in obj]
+    assert len(outcomes) == 1
+    outcome = Outcome.from_dict(_linked_outcome(result))
+    assert outcome.status == "completed"
+    assert outcome.verification_id == verification.verification_id
+
+
+def test_failed_task_emits_failed_outcome(tmp_path, monkeypatch):
+    """A failed task emits Outcome(status=failed) with no verification link."""
+    transport = _ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)])
+    runtime = _ScriptedRuntime(transport, OperatorResponse(success=False, error="model exploded", model="qwen3:8b"))
+    op, patches = _wire_operator(tmp_path, monkeypatch, runtime)
+    result = _run_wired(op, patches)
+
+    assert not result["success"]
+    outcomes = [obj for obj in result["runtime_objects"] if "status" in obj]
+    assert len(outcomes) == 1
+    outcome = Outcome.from_dict(_linked_outcome(result))
+    assert outcome.status == "failed"
+    assert outcome.verification_id == "", "failed tasks must not claim verification"
+    assert not [obj for obj in result["runtime_objects"] if "method" in obj]
+
+
+def test_degraded_subsystems_become_observations(tmp_path, monkeypatch):
+    """Each degraded subsystem name becomes one Observation(kind=degraded)."""
+    trace = _broker_trace(degraded=["memory_selector", "history_compaction"])
+    transport = _ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)])
+    runtime = _ScriptedRuntime(transport, OperatorResponse(success=True, user_text="All done.", model="qwen3:8b"))
+    op, patches = _wire_operator(tmp_path, monkeypatch, runtime, trace=trace)
+    result = _run_wired(op, patches)
+
+    assert result["success"]
+    degraded = [obj for obj in result["runtime_objects"] if "observation_id" in obj and obj.get("kind") == "degraded"]
+    assert {obj["summary"] for obj in degraded} == {
+        "subsystem unavailable: memory_selector",
+        "subsystem unavailable: history_compaction",
+    }
+
+
+def test_compaction_emits_observation(tmp_path, monkeypatch):
+    """A compaction event in the broker trace becomes Observation(kind=compaction)."""
+    trace = _broker_trace(
+        compaction=[
+            {
+                "triggered": True,
+                "tokens_before": 4000,
+                "tokens_after": 800,
+                "turns_summarized": 6,
+                "archived": True,
+                "failed": False,
+            }
+        ]
+    )
+    transport = _ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)])
+    runtime = _ScriptedRuntime(transport, OperatorResponse(success=True, user_text="All done.", model="qwen3:8b"))
+    op, patches = _wire_operator(tmp_path, monkeypatch, runtime, trace=trace)
+    result = _run_wired(op, patches)
+
+    assert result["success"]
+    compactions = [obj for obj in result["runtime_objects"] if obj.get("kind") == "compaction"]
+    assert len(compactions) == 1
+    assert "4000" in compactions[0]["summary"] and "800" in compactions[0]["summary"]
+    assert len(compactions[0]["detail"]) <= 300
+
+
+def test_response_envelope_carries_runtime_objects(tmp_path, monkeypatch):
+    """runtime_objects survives _response_envelope as dicts (TUI-consumable)."""
+    transport = _ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)])
+    runtime = _ScriptedRuntime(transport, OperatorResponse(success=True, user_text="All done.", model="qwen3:8b"))
+    op, patches = _wire_operator(tmp_path, monkeypatch, runtime)
+    result = _run_wired(op, patches)
+
+    objects = result["runtime_objects"]
+    assert isinstance(objects, list) and objects
+    assert all(isinstance(obj, dict) for obj in objects)
+    # additive key survives even when a run produced no objects
+    envelope = rune_operator.RuneOperator._response_envelope(OperatorResponse(success=True), None)
+    assert envelope["runtime_objects"] == []
+
+
+# ---------------------------------------------------------------------------
+# Task 3 — work pane
+# ---------------------------------------------------------------------------
+
+
+def test_work_pane_renders_runtime_objects():
+    """WorkPane.add_object() renders kind glyph + id + summary; update_object()
+    replaces in place by id (no duplicates)."""
+    pane = rune_tui.WorkPane()
+    observation = Observation(
+        observation_id="obs-1",
+        task_id="task-1",
+        kind="degraded",
+        summary="subsystem unavailable: memory_selector",
+        detail="",
+        ts=TS,
+    ).to_dict()
+    outcome = Outcome(
+        outcome_id="oc-1",
+        task_id="task-1",
+        status="completed",
+        verification_id="",
+        degraded=[],
+        ts=TS,
+    ).to_dict()
+
+    assert pane.add_object(observation) is True
+    assert pane.add_object(outcome) is True
+    assert pane.object_count() == 2
+
+    rendered = pane.render().plain
+    assert "obs-1" in rendered and "subsystem unavailable: memory_selector" in rendered
+    assert "oc-1" in rendered and "completed" in rendered
+    # sections render newest-first: Outcomes before Observations
+    assert rendered.index("oc-1") < rendered.index("obs-1")
+
+    # update in place by id — no duplicates
+    updated = {**observation, "summary": "subsystem unavailable: storage"}
+    assert pane.update_object(updated) is True
+    assert pane.object_count() == 2
+    rendered = pane.render().plain
+    assert "storage" in rendered
+    assert "memory_selector" not in rendered
+    # re-adding the same id is an in-place update, never a duplicate
+    assert pane.add_object(updated) is False
+    assert pane.object_count() == 2
+
+
+def test_work_pane_collapses_and_expands():
+    """Ctrl+3 toggles the work pane; hidden on chat channel, visible on task."""
+    app = rune_tui.RuneApp()
+    binding = next(item for item in rune_tui.RuneApp.BINDINGS if item.key == "ctrl+3")
+    assert binding.action == "toggle_work_pane"
+    assert callable(app.action_toggle_work_pane)
+
+    # visibility policy: hidden on the chat channel, visible on task
+    assert app._work_pane_visible_for("chat") is False
+    assert app._work_pane_visible_for("task") is True
+
+    # Ctrl+3 toggles the expansion state
+    app.action_toggle_work_pane()
+    assert app.work_pane_visible is False
+    assert app._work_pane_visible_for("task") is False
+    app.action_toggle_work_pane()
+    assert app.work_pane_visible is True
+    assert app._work_pane_visible_for("task") is True
+
+
+# ---------------------------------------------------------------------------
+# Quality-gate discharges (PR #180 review) — approved-resume + crash-tier
+# ---------------------------------------------------------------------------
+
+
+def test_approval_resume_emits_outcome(tmp_path, monkeypatch):
+    """Quality I-2: an approval-resumed turn must emit an Outcome object.
+
+    resolve_pending_approval applies the approved operation and finalizes;
+    the resume envelope must carry the same first-class Outcome a direct
+    turn gets (approved tool calls are real executed work).
+    """
+    transport = _ScriptedTransport([])
+    tool_request = ToolRequest(call_id="call-9", name="write_file", arguments={}, source="ollama-native")
+    tool_event = ToolEvent(kind="completed", request=tool_request, result={"success": True, "result": "written"})
+    resumed = OperatorResponse(
+        success=True,
+        user_text="Approved and written.",
+        model="qwen3:8b",
+        tool_events=[tool_event],
+        override_tier=0,
+        recall_count=0,
+    )
+    runtime = _ScriptedRuntime(transport, resumed)
+    # runtime.approve is what resolve_pending_approval calls
+    runtime.approve = lambda pending_state: resumed  # type: ignore[method-assign]
+    op, patches = _wire_operator(tmp_path, monkeypatch, runtime)
+
+    # Seed a pending approval on the operator state (the scripted runtime
+    # returns a terminal response, so no pending approval exists naturally).
+    seeded = rune_operator.RuneTask(task_id=f"task_{uuid.uuid4().hex[:8]}", description="seeded approval")
+    op.state.tasks.append(seeded)
+    op.state.pending_task_id = seeded.task_id
+    op.state.pending_request = {"call_id": "call-9", "name": "write_file", "arguments": {}}
+    from rune.tool_executor import PreparedTool, _canonical_digest, _path_digest, serialize_prepared_tool
+    import time as _t
+    import os as _os
+
+    # write_file policy allows only /home/nick and /mnt/workdrive — use the
+    # same sandbox pattern the P0-05 E2E uses (cleared before use).
+    sandbox = Path(_os.path.expanduser("~")) / ".rune-resume-probe"
+    import shutil as _shutil
+
+    _shutil.rmtree(sandbox, ignore_errors=True)
+    _shutil.rmtree(sandbox, ignore_errors=True)
+    sandbox.mkdir(parents=True)
+    target = sandbox / "out.txt"
+    approval_token = uuid.uuid4().hex  # not a literal credential — henxels wants no token= literal
+    prepared = PreparedTool(
+        token=approval_token,
+        name="write_file",
+        arguments={"path": str(target), "content": "x"},
+        operation_digest=_canonical_digest("write_file", {"path": str(target), "content": "x"}),
+        created_at=_t.time(),
+        expires_at=_t.time() + 60,
+        requested_path=str(target),
+        resolved_path=str(target.resolve()),
+        path_digest=_path_digest(target),
+    )
+    op.state.pending_runtime_state = {"prepared_tool": serialize_prepared_tool(prepared)}
+
+    with contextlib.ExitStack() as stack:
+        for item in patches:
+            stack.enter_context(item)
+        result = op.resolve_pending_approval(True)
+    assert result["success"], result.get("error")
+    # probe cleanup: remove the write sandbox the approval touched
+    _shutil.rmtree(sandbox, ignore_errors=True)
+    outcomes = [obj for obj in result.get("runtime_objects", []) if "status" in obj]
+    assert len(outcomes) == 1, "approval-resumed turn must emit exactly one Outcome"
+    outcome = Outcome.from_dict(outcomes[0])
+    assert outcome.status == "completed"
+    assert outcome.task_id == seeded.task_id
+
+
+def test_failed_verification_backfill_is_straight_read(tmp_path, monkeypatch):
+    """Quality I-1: the outcome's verification link in the ENVELOPE must be
+    exactly the emitted verification's id — no test-side recompute masking a
+    send_task back-fill regression. (The canary already asserts this; the
+    pytest suite is now strictly as strong.)"""
+    response = OperatorResponse(
+        success=True,
+        user_text="Command ran clean.",
+        model="qwen3:8b",
+        tool_events=[
+            ToolEvent(
+                kind="completed",
+                request=ToolRequest(call_id="call-1", name="run_command", arguments={}, source="ollama-native"),
+                result={"success": True, "result": "(no output)"},
+            )
+        ],
+    )
+    transport = _ScriptedTransport([_chat_response(1600, 42, 1_000_000_000)])
+    runtime = _ScriptedRuntime(transport, response)
+    op, patches = _wire_operator(tmp_path, monkeypatch, runtime)
+    result = _run_wired(op, patches)
+
+    verifications = [obj for obj in result["runtime_objects"] if "method" in obj]
+    outcomes = [obj for obj in result["runtime_objects"] if "status" in obj]
+    assert len(verifications) == 1 and len(outcomes) == 1
+    # Direct envelope assertion — no _linked_outcome recomputation.
+    assert outcomes[0]["verification_id"] == verifications[0]["verification_id"]
+    assert outcomes[0]["verification_id"] != ""
+
+
+def test_work_pane_entries_capped_and_channel_scoped():
+    """Quality I-3: pane entries are capped (bounded growth) and the channel
+    filter keeps chat objects out of a task-scoped pane view."""
+    pane = rune_tui.WorkPane()
+    overflow = rune_tui._WORK_PANE_MAX_ENTRIES + 25
+    for n in range(overflow):
+        pane.add_object({"observation_id": f"obs_{n}", "kind": "degraded", "summary": f"s{n}"})
+    assert len(pane._entries) <= rune_tui._WORK_PANE_MAX_ENTRIES
+    # newest survive, oldest evicted
+    assert "obs_0" not in pane._entries
+    assert f"obs_{overflow - 1}" in pane._entries
+    # channel scoping: entries carry their channel; visible set filters
+    pane2 = rune_tui.WorkPane()
+    pane2.add_object({"observation_id": "obs_task", "kind": "degraded", "summary": "t", "channel": "task"})
+    pane2.add_object({"observation_id": "obs_chat", "kind": "degraded", "summary": "c", "channel": "chat"})
+    visible = pane2.visible_entries("chat")
+    assert [oid for oid, _ in visible] == ["obs_chat"]