diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml
index 862f7b1e..e876138f 100644
--- a/.gitea/workflows/ci.yml
+++ b/.gitea/workflows/ci.yml
@@ -89,6 +89,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 \
diff --git a/config/v1_authority_map.json b/config/v1_authority_map.json
index 5fadbca4..d792ba5f 100644
--- a/config/v1_authority_map.json
+++ b/config/v1_authority_map.json
@@ -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."
         }
       ]
     },
diff --git a/docs/runtime-objects.md b/docs/runtime-objects.md
new file mode 100644
index 00000000..82df48fd
--- /dev/null
+++ b/docs/runtime-objects.md
@@ -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).
\ No newline at end of file
diff --git a/src/rune/canary_p1_07.py b/src/rune/canary_p1_07.py
new file mode 100644
index 00000000..c9d82400
--- /dev/null
+++ b/src/rune/canary_p1_07.py
@@ -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())
diff --git a/src/rune/rune_operator.py b/src/rune/rune_operator.py
index 4211ed99..545382dc 100644
--- a/src/rune/rune_operator.py
+++ b/src/rune/rune_operator.py
@@ -35,14 +35,28 @@ import tool_executor  # noqa: E402
 
 try:
     from .local_tool_runtime import LocalToolRuntime, OllamaChatTransport
-    from .tool_protocol import OperatorResponse, contains_internal_protocol
+    from .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:
@@ -99,6 +113,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(
@@ -360,6 +498,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",
             )
 
@@ -402,6 +541,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:
@@ -1226,6 +1383,38 @@ 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:
+                    runtime_objects: list = []
+                    task_id = task.task_id
+                    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()
+                        )
+                    for observation in _observations_from_degraded(response.degraded, task_id):
+                        runtime_objects.append(observation.to_dict())
+                    if response.pending_approval is None:
+                        # Turn finished from the model's side — emit the
+                        # verification claim (only when a check actually
+                        # ran) and the provisional outcome (send_task
+                        # back-fills the final transition status).
+                        verification, outcome = _verification_and_outcome(
+                            response, task_id, response.degraded
+                        )
+                        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:
+                    from .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):
diff --git a/src/rune/rune_tui.py b/src/rune/rune_tui.py
index 47d0d03b..a1d9a98b 100644
--- a/src/rune/rune_tui.py
+++ b/src/rune/rune_tui.py
@@ -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,156 @@ class ChannelView(Vertical):
         )
 
 
+# Section order for the work pane's operational index (P1-07 #30):
+# Outcomes, Verification, Observations, Memory queries, Tool calls.
+_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)."""
+        object_id = self._object_id(obj)
+        is_new = object_id not in self._entries
+        self._entries[object_id] = dict(obj)
+        if self.is_mounted:
+            self.refresh()
+        return is_new
+
+    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 +1450,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 +1468,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 +1508,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 +1522,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 +3690,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 +3828,36 @@ 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):
+                    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 +3883,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 +3960,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 +3969,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 +3979,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)
diff --git a/src/rune/tool_protocol.py b/src/rune/tool_protocol.py
index 68aee7b0..1c23d8ab 100644
--- a/src/rune/tool_protocol.py
+++ b/src/rune/tool_protocol.py
@@ -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}
diff --git a/tests/test_p1_07_runtime_objects.py b/tests/test_p1_07_runtime_objects.py
new file mode 100644
index 00000000..aaeb1750
--- /dev/null
+++ b/tests/test_p1_07_runtime_objects.py
@@ -0,0 +1,442 @@
+"""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 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