Let configuration answers own model/context status labels

src/agent_comms/acp_extension.py

@@ -18,6 +18,7 @@
 )
 from .agent_events import CompactionEvent
 from .declared_family import DeclaredFamily
+from .context_reading import ContextReading
 from .input_attempt import ACPInputIdText, InputAttempt
 from .input_origin import HumanInputOrigin, InputOrigin, UnattributedInputOrigin
 from .field_codec import FieldCodec
@@ -446,26 +447,20 @@ def for_session(self, session_id: str) -> InputStartedUpdate:
         )
 
 
-@dataclass(frozen=True)
-class ContextUsage:
-    used: int
-    size: int
-
-    @property
-    def label(self) -> str:
-        return f"{self.used:,}/{self.size:,} tokens ({self.used / self.size * 100:.1f}%)"
-
-
 @dataclass(frozen=True)
 class CoordinationChangedUpdate(AgentCommsUpdate):
     thread: ThreadIncarnation
     wire_root: str
     owner_pid: int
     worktree: str
-    model: str | None
+    model: str
     thinking_level: str | None
     title: str
-    context_usage: ContextUsage | None
+    context_usage: ContextReading
+
+    @property
+    def status_label(self) -> str:
+        return " · ".join((self.model, *self.context_usage.label_parts()))
 
     @property
     def source_identity(self):

src/agent_comms/agent_activity.py

@@ -24,7 +24,7 @@
 from .errors import RelationViolationError
 from .registry_document import RegistrySnapshot
 from .routing import TurnRouting
-from .runtime_info import AgentRuntimeInfo, RuntimeInfoStore
+from .runtime_info import RuntimeInfo, RuntimeInfoStore
 from .store_files import _store_lock
 from .thread_identity import OwnerIdentity, ThreadIncarnation
 from .threads import Thread
@@ -266,5 +266,5 @@ def observe_runtime_info(self, thread: str, observation: RuntimeInfoObservation)
             canonical = self.registry.require(thread).name
             self.runtime_info.observe(canonical, time.time(), observation)
 
-    def agent_info_of(self, thread: str) -> AgentRuntimeInfo | None:
-        return self.runtime_info.read().get(self.registry.require(thread).name)
+    def agent_info_of(self, thread: str) -> RuntimeInfo:
+        return self.runtime_info.for_thread(self.registry.require(thread).name)

src/agent_comms/context_reading.py

@@ -0,0 +1,43 @@
+"""Native context observations own their presentation without nullable readings."""
+
+from abc import abstractmethod
+from dataclasses import dataclass
+
+from .declared_family import DeclaredFamily
+
+
+class ContextReading(DeclaredFamily):
+    @abstractmethod
+    def label_parts(self) -> tuple[str, ...]: ...
+
+    @abstractmethod
+    def status_text(self) -> str: ...
+
+
+@dataclass(frozen=True)
+class ContextUsage(ContextReading):
+    used: int
+    size: int
+
+    @property
+    def context_percent(self) -> float:
+        return self.used / self.size * 100
+
+    @property
+    def label(self) -> str:
+        return f"{self.used:,}/{self.size:,} tokens ({self.context_percent:.1f}%)"
+
+    def label_parts(self) -> tuple[str, ...]:
+        return (self.label,)
+
+    def status_text(self) -> str:
+        return self.label
+
+
+@dataclass(frozen=True)
+class UnmeasuredContext(ContextReading):
+    def label_parts(self) -> tuple[str, ...]:
+        return ()
+
+    def status_text(self) -> str:
+        return "Native owner has not reported context usage"

src/agent_comms/runtime_info.py

@@ -2,23 +2,48 @@
 
 from __future__ import annotations
 
+from abc import ABC, abstractmethod
 from collections.abc import Sequence
 from dataclasses import dataclass, field, replace
 from typing import TYPE_CHECKING, ClassVar
 
+from .context_reading import ContextReading, ContextUsage, UnmeasuredContext
 from .errors import RelationViolationError
 from .field_codec import FieldCodec
 from .locked_store import LockedStore
 from .thread_owned_state import ThreadOwnedState
 
 if TYPE_CHECKING:
     from .agent_events import RuntimeInfoObservation
-    from .acp_extension import ContextUsage
     from .threads import Thread
 
 
+class RuntimeInfo(ABC):
+    """A runtime-info read is observed or explicitly unobserved."""
+
+    __slots__ = ()
+
+    @abstractmethod
+    def context_reading(self) -> ContextReading: ...
+
+    @abstractmethod
+    def model_or(self, default: str) -> str: ...
+
+
+class UnobservedRuntimeInfo(RuntimeInfo):
+    """No observation exists; do not invent a model or observation timestamp."""
+
+    __slots__ = ()
+
+    def context_reading(self) -> ContextReading:
+        return UnmeasuredContext()
+
+    def model_or(self, default: str) -> str:
+        return default
+
+
 @dataclass(frozen=True, slots=True)
-class AgentRuntimeInfo:
+class AgentRuntimeInfo(RuntimeInfo):
     """Declares mutable runtime metadata for one registered thread."""
 
     thread: str
@@ -43,17 +68,13 @@ def context_percent(self) -> float | None:
             return None
         return self.context_used / self.context_size * 100
 
-    @property
-    def context_usage(self) -> ContextUsage | None:
+    def context_reading(self) -> ContextReading:
         if self.context_used is None or not self.context_size:
-            return None
-        from .acp_extension import ContextUsage
-
+            return UnmeasuredContext()
         return ContextUsage(self.context_used, self.context_size)
 
-    @classmethod
-    def optional_context_usage(cls, info: AgentRuntimeInfo | None) -> ContextUsage | None:
-        return info.context_usage if info is not None else None
+    def model_or(self, default: str) -> str:
+        return self.model or default
 
 
 @dataclass(frozen=True, slots=True)
@@ -70,6 +91,9 @@ def record_type(self) -> type[dict[str, AgentRuntimeInfo]]:
     def empty(self) -> dict[str, AgentRuntimeInfo]:
         return {}
 
+    def for_thread(self, thread: str) -> RuntimeInfo:
+        return self.read().get(thread, UnobservedRuntimeInfo())
+
     def observe(self, thread: str, timestamp: float, observation: RuntimeInfoObservation) -> None:
         """Apply only this observation's fields in the original store transaction."""
         def apply(values: dict[str, AgentRuntimeInfo]) -> dict[str, AgentRuntimeInfo]:

src/agent_comms/thread_configuration.py

@@ -8,7 +8,6 @@
 from .acp_extension import CoordinationChangedUpdate
 from .pi_vocabulary import ThinkingLevel
 from .registration import Registration
-from .runtime_info import AgentRuntimeInfo
 from .thread_identity import ThreadIncarnation
 from .thread_management import ThreadManagement
 from .agent_activity import AgentActivity
@@ -24,7 +23,6 @@ def __init__(self, root: Path, registry: Registration, threads: ThreadManagement
     def answer(self, name: str) -> CoordinationChangedUpdate:
         thread = self.registry.require(name)
         info = self.agents.agent_info_of(thread.name)
-        usage = AgentRuntimeInfo.optional_context_usage(info)
         return CoordinationChangedUpdate(
             ThreadIncarnation(thread.name, thread.created_at),
             str(self.root.resolve()),
@@ -33,5 +31,5 @@ def answer(self, name: str) -> CoordinationChangedUpdate:
             self.threads.resolve_thread_model(thread.name),
             ThinkingLevel.optional_name(thread.thinking_level),
             thread.title or thread.name,
-            usage,
+            info.context_reading(),
         )

src/agent_comms/thread_management.py

@@ -32,6 +32,7 @@
 from .input_disposition import InputDispositions
 from .message_bus import MessageBus
 from .native_arguments import NativeArguments
+from .native_settings import NativeThreadDefaults
 from .native_session_files import NativeSessionFiles
 from .launch_runtime import LaunchRuntime
 from .owner_lifecycle import OwnerLifecycle
@@ -541,15 +542,15 @@ def initialize_native_configuration(
                 self.registry.register(updated, self.registry.status(thread.name))
             return updated
 
-    def resolve_thread_model(self, name: str, default: str | None = None) -> str | None:
+    def resolve_thread_model(self, name: str, default: str = NativeThreadDefaults.model) -> str:
         """Prefer a saved selection, then the resumed session's last model."""
         thread = self.registry.require(name)
         if thread.model is not None:
             return thread.model
         if thread.session_file and (model := _session_model(Path(thread.session_file))):
             return "/".join(model)
         info = self.agents.agent_info_of(thread.name)
-        return (info.model if info else None) or default
+        return info.model_or(default)
 
     def attach_session(self, original: Thread, session_file: str, *, pid: int | None = None) -> Thread:
         """Attach authoritative Pi runtime state to an existing thread."""

tests/test_acp_extension.py

@@ -26,13 +26,15 @@
     SelectedWriteAcceptedUpdate,
     TextRouteUpdate,
     TranscriptChangedUpdate,
+    TranscriptDraftUpdate,
     TranscriptSnapshotUpdate,
     TurnChangedUpdate,
     UnavailableCursorObservation,
     decode_updates,
     encode_updates,
 )
 from agent_comms.acp_failure import BackendDeliveryFailure
+from agent_comms.context_reading import UnmeasuredContext
 from agent_comms.agent_events import CompactionStart
 from agent_comms.pi_payloads import McpLiveReceipt
 from agent_comms.thread_identity import AdmissionIdentity, ThreadIncarnation
@@ -70,17 +72,18 @@ def test_declared_family_roundtrip_and_strict_boundary(tmp_path):
             "/home/ts/wt/pilot",
             123,
             "/home/ts/wt/pilot",
-            None,
+            "openai-codex/gpt-6.1-sol",
             None,
             "Pilot",
-            None,
+            UnmeasuredContext(),
         ),
         GoalChangedUpdate(None, None),
         CompactionChangedUpdate(CompactionStart()),
         CompactionCommittedUpdate("summary", "first-kept"),
         TranscriptSnapshotUpdate(
             read.read(), read.identity
         ),
+        TranscriptDraftUpdate(read.identity, None),
         InputDeliveryChangedUpdate("input"),
         McpClientReceiptUpdate(
             "turn", McpLiveReceipt(1, "pi-mcp-client", "a" * 32, "running", "turn", ())
@@ -104,7 +107,7 @@ def test_declared_family_roundtrip_and_strict_boundary(tmp_path):
         {"agentComms": {"updates": [{"kind": "unregistered"}]}},
         {"agentComms": {"updates": [], "extra": True}},
     ):
-        with pytest.raises(ValueError):
+        with pytest.raises((TypeError, ValueError)):
             decode_updates(invalid)