Persist runtime owner metadata in sessions

Status	File	+	-
M	`src/loader/runtime/bootstrap.py`	2	1
A	`src/loader/runtime/owner_metadata.py`	73	0
M	`src/loader/runtime/public_shell.py`	21	0
M	`src/loader/runtime/session.py`	33	1
M	`tests/test_runtime_bootstrap.py`	8	2
M	`tests/test_runtime_handle.py`	8	2
M	`tests/test_runtime_launcher.py`	4	1
M	`tests/test_runtime_public_shell.py`	10	0
M	`tests/test_session_state.py`	33	0

src/loader/runtime/bootstrap.pymodified

      RuntimeSafeguardsProtocol,
+ )
  from .events import TurnSummary
 +from .owner_metadata import build_runtime_owner_metadata
  from .permissions import PermissionConfigStatus, PermissionPolicy
  from .reasoning_service import RuntimeReasoningService
  from .session import ConversationSession
          _queue_steering_message=source.queue_steering_message,
          _drain_steering_messages=source.drain_steering_messages,
          _refresh_capability_profile=source.refresh_capability_profile,
 -        metadata={"owner_type": type(source).__name__},
 +        metadata=build_runtime_owner_metadata(source),
+     )

src/loader/runtime/owner_metadata.pyadded

 +"""Shared helpers for runtime-owner metadata."""
++
 +from __future__ import annotations
++
 +from typing import Any
++
 +_KNOWN_RUNTIME_OWNER_PATHS = {
 +    "Agent": "public-agent",
 +    "RuntimeHandle": "runtime-handle",
 +}
++
++
 +def normalize_runtime_owner_type(value: Any) -> str | None:
 +    """Coerce persisted runtime-owner types into optional text."""
++
 +    if value is None:
 +        return None
 +    text = str(value).strip()
 +    return text or None
++
++
 +def normalize_runtime_owner_path(
 +    value: Any,
 +    *,
 +    owner_type: str | None = None,
 +) -> str | None:
 +    """Coerce persisted runtime-owner paths into canonical text."""
++
 +    if value is not None:
 +        text = str(value).strip()
 +        if text:
 +            return text
 +    if owner_type is None:
 +        return None
 +    return _KNOWN_RUNTIME_OWNER_PATHS.get(owner_type, _camel_to_kebab(owner_type))
++
++
 +def build_runtime_owner_metadata(source: Any) -> dict[str, str | None]:
 +    """Build canonical runtime-owner metadata from one shell owner."""
++
 +    owner_type = (
 +        normalize_runtime_owner_type(source)
 +        if isinstance(source, str)
 +        else normalize_runtime_owner_type(type(source).__name__)
 +    )
 +    return {
 +        "owner_type": owner_type,
 +        "owner_path": normalize_runtime_owner_path(None, owner_type=owner_type),
 +    }
++
++
 +def format_runtime_owner_label(
 +    owner_type: str | None,
 +    owner_path: str | None,
 +) -> str | None:
 +    """Render one compact human-readable runtime-owner label."""
++
 +    normalized_type = normalize_runtime_owner_type(owner_type)
 +    normalized_path = normalize_runtime_owner_path(owner_path, owner_type=normalized_type)
 +    if normalized_type and normalized_path:
 +        return f"{normalized_path} ({normalized_type})"
 +    return normalized_path or normalized_type
++
++
 +def _camel_to_kebab(value: str) -> str:
 +    """Convert one CamelCase-ish class name into kebab-case."""
++
 +    chars: list[str] = []
 +    for index, char in enumerate(value):
 +        if char.isupper() and index > 0:
 +            chars.append("-")
 +        chars.append(char.lower())
 +    return "".join(chars)

src/loader/runtime/public_shell.pymodified

  from .dod import DefinitionOfDoneStore
  from .events import AgentEvent, TurnSummary
  from .launcher import build_runtime_launcher
 +from .owner_metadata import build_runtime_owner_metadata
  from .permissions import PermissionConfigStatus, PermissionMode, PermissionPolicy
  from .prompt_history import PromptSnapshot
  from .prompting import build_system_prompt_result
      prompt_format: str | None,
      prompt_sections: list[str],
      workflow_mode: str,
 +    runtime_owner_type: str | None,
 +    runtime_owner_path: str | None,
      rotate_after_bytes: int,
      auto_compaction_input_tokens_threshold: int,
      compaction_keep_last_messages: int,
          few_shot_factory=few_shot_factory,
          project_root=project_root,
          messages=messages or [],
 +        runtime_owner_type=runtime_owner_type,
 +        runtime_owner_path=runtime_owner_path,
          permission_mode=permission_policy.active_mode.as_str(),
          permission_prompting_enabled=permission_policy.prompting_enabled,
          permission_rule_counts=_copy_rule_counts(permission_policy.rule_counts()),
      prompt_format: str | None,
      prompt_sections: list[str],
      workflow_mode: str,
 +    runtime_owner_type: str | None,
 +    runtime_owner_path: str | None,
      rotate_after_bytes: int,
      auto_compaction_input_tokens_threshold: int,
      compaction_keep_last_messages: int,
          prompt_format=prompt_format,
          prompt_sections=prompt_sections,
          workflow_mode=workflow_mode,
 +        runtime_owner_type=runtime_owner_type,
 +        runtime_owner_path=runtime_owner_path,
          rotate_after_bytes=rotate_after_bytes,
          auto_compaction_input_tokens_threshold=(
              auto_compaction_input_tokens_threshold
      owner.prompt_sections = list(install.restored.prompt_sections)
      owner.last_turn_summary = install.restored.last_turn_summary
      owner._system_message = None
 +    owner_metadata = build_runtime_owner_metadata(owner)
 +    if (
 +        install.session.runtime_owner_type != owner_metadata["owner_type"]
 +        or install.session.runtime_owner_path != owner_metadata["owner_path"]
 +    ):
 +        install.session.update_runtime_state(
 +            runtime_owner_type=owner_metadata["owner_type"],
 +            runtime_owner_path=owner_metadata["owner_path"],
 +        )
  def build_fresh_runtime_session_install(
  ) -> RuntimeSessionInstall:
      """Build a fresh runtime session install from the current public shell."""
 +    owner_metadata = build_runtime_owner_metadata(owner)
      return create_runtime_session_install(
          project_root=owner.project_root,
          messages=messages,
          prompt_format=owner.prompt_format,
          prompt_sections=list(owner.prompt_sections),
          workflow_mode=workflow_mode or owner.workflow_mode,
 +        runtime_owner_type=owner_metadata["owner_type"],
 +        runtime_owner_path=owner_metadata["owner_path"],
          rotate_after_bytes=owner.config.session_rotate_after_bytes,
          auto_compaction_input_tokens_threshold=(
              owner.config.session_auto_compaction_input_tokens_threshold

src/loader/runtime/session.pymodified

      has_canonical_completion_trace,
      normalize_completion_trace,
+ )
 +from .owner_metadata import (
 +    normalize_runtime_owner_path,
 +    normalize_runtime_owner_type,
 +)
  from .prompt_history import PromptSnapshot, normalize_prompt_history
  from .workflow_ledger import WorkflowLedger
  from .workflow_policy import WorkflowTimelineEntry
 -SESSION_VERSION = 10
 +SESSION_VERSION = 11
  DEFAULT_ROTATE_AFTER_BYTES = 256 * 1024
  MAX_ROTATED_FILES = 3
  _UNSET = object()
      usage: dict[str, int] = field(default_factory=dict)
      active_dod_path: str | None = None
      current_task: str | None = None
 +    runtime_owner_type: str | None = None
 +    runtime_owner_path: str | None = None
      workflow_mode: str = "execute"
      permission_mode: str = "workspace-write"
      permission_prompting_enabled: bool = False
              "usage": dict(self.usage),
              "active_dod_path": self.active_dod_path,
              "current_task": self.current_task,
 +            "runtime_owner_type": self.runtime_owner_type,
 +            "runtime_owner_path": self.runtime_owner_path,
              "workflow_mode": self.workflow_mode,
              "permission_mode": self.permission_mode,
              "permission_prompting_enabled": self.permission_prompting_enabled,
              usage=normalize_usage(data.get("usage")),
              active_dod_path=data.get("active_dod_path"),
              current_task=data.get("current_task"),
 +            runtime_owner_type=normalize_runtime_owner_type(
 +                data.get("runtime_owner_type")
 +            ),
 +            runtime_owner_path=normalize_runtime_owner_path(
 +                data.get("runtime_owner_path"),
 +                owner_type=normalize_runtime_owner_type(
 +                    data.get("runtime_owner_type")
 +                ),
 +            ),
              workflow_mode=str(data.get("workflow_mode", "execute")),
              permission_mode=str(data.get("permission_mode", "workspace-write")),
              permission_prompting_enabled=bool(
      usage_totals: dict[str, int] = field(default_factory=dict)
      active_dod_path: str | None = None
      current_task: str | None = None
 +    runtime_owner_type: str | None = None
 +    runtime_owner_path: str | None = None
      workflow_mode: str = "execute"
      permission_mode: str = "workspace-write"
      permission_prompting_enabled: bool = False
          *,
          active_dod_path: str | None = None,
          current_task: str | None = None,
 +        runtime_owner_type: str | None = None,
 +        runtime_owner_path: str | None = None,
          workflow_mode: str | None = None,
          permission_mode: str | None = None,
          permission_prompting_enabled: bool | None = None,
              self.active_dod_path = active_dod_path
          if current_task is not None:
              self.current_task = current_task
 +        if runtime_owner_type is not None:
 +            self.runtime_owner_type = normalize_runtime_owner_type(runtime_owner_type)
 +        if runtime_owner_path is not None or runtime_owner_type is not None:
 +            self.runtime_owner_path = normalize_runtime_owner_path(
 +                runtime_owner_path,
 +                owner_type=self.runtime_owner_type,
 +            )
          if workflow_mode is not None:
              self.workflow_mode = workflow_mode
          if permission_mode is not None:
              usage=dict(self.usage_totals),
              active_dod_path=self.active_dod_path,
              current_task=self.current_task,
 +            runtime_owner_type=self.runtime_owner_type,
 +            runtime_owner_path=self.runtime_owner_path,
              workflow_mode=self.workflow_mode,
              permission_mode=self.permission_mode,
              permission_prompting_enabled=self.permission_prompting_enabled,
          instance.usage_totals = dict(snapshot.usage)
          instance.active_dod_path = snapshot.active_dod_path
          instance.current_task = snapshot.current_task
 +        instance.runtime_owner_type = snapshot.runtime_owner_type
 +        instance.runtime_owner_path = snapshot.runtime_owner_path
          instance.workflow_mode = snapshot.workflow_mode
          instance.permission_mode = snapshot.permission_mode
          instance.permission_prompting_enabled = snapshot.permission_prompting_enabled

tests/test_runtime_bootstrap.pymodified

      assert context.workflow_mode == agent.workflow_mode
      assert context.prompt_format == agent.prompt_format
      assert context.prompt_sections == agent.prompt_sections
 -    assert source.metadata == {"owner_type": "Agent"}
 +    assert source.metadata == {
 +        "owner_type": "Agent",
 +        "owner_path": "public-agent",
 +    }
  def test_sync_runtime_context_refreshes_prompt_and_capability_state(
      assert isinstance(launcher, RuntimeLauncher)
      assert isinstance(launcher.source, RuntimeBootstrapView)
      assert launcher.source is not agent
 -    assert launcher.source.metadata == {"owner_type": "Agent"}
 +    assert launcher.source.metadata == {
 +        "owner_type": "Agent",
 +        "owner_path": "public-agent",
 +    }

tests/test_runtime_handle.pymodified

      assert isinstance(launcher, RuntimeLauncher)
      assert isinstance(launcher.source, RuntimeBootstrapView)
      assert launcher.source is not handle
 -    assert launcher.source.metadata == {"owner_type": "RuntimeHandle"}
 +    assert launcher.source.metadata == {
 +        "owner_type": "RuntimeHandle",
 +        "owner_path": "runtime-handle",
 +    }
      assert context.project_root == temp_dir.resolve()
      assert context.backend is handle.backend
      assert context.registry is handle.registry
+     )
      assert summary.final_response == "Runtime handle reply."
 -    assert runtime.source.metadata == {"owner_type": "RuntimeHandle"}
 +    assert runtime.source.metadata == {
 +        "owner_type": "RuntimeHandle",
 +        "owner_path": "runtime-handle",
 +    }
      assert any(event.type == "response" for event in events)

tests/test_runtime_launcher.pymodified

      assert isinstance(launcher, RuntimeLauncher)
      assert isinstance(launcher.source, RuntimeBootstrapView)
      assert launcher.source is not agent
 -    assert launcher.source.metadata == {"owner_type": "Agent"}
 +    assert launcher.source.metadata == {
 +        "owner_type": "Agent",
 +        "owner_path": "public-agent",
 +    }
  @pytest.mark.asyncio

tests/test_runtime_public_shell.pymodified

          prompt_format="native",
          prompt_sections=["Runtime Config", "Workflow Context"],
          workflow_mode="execute",
 +        runtime_owner_type="RuntimeHandle",
 +        runtime_owner_path="runtime-handle",
          rotate_after_bytes=handle.config.session_rotate_after_bytes,
          auto_compaction_input_tokens_threshold=(
              handle.config.session_auto_compaction_input_tokens_threshold
      assert session.permission_mode == handle.active_permission_mode
      assert session.permission_prompting_enabled is handle.permission_policy.prompting_enabled
      assert session.permission_rule_counts == handle.permission_policy.rule_counts()
 +    assert session.runtime_owner_type == "RuntimeHandle"
 +    assert session.runtime_owner_path == "runtime-handle"
      assert session.prompt_format == "native"
      assert session.prompt_sections == ["Runtime Config", "Workflow Context"]
          prompt_format="native",
          prompt_sections=["Runtime Config", "Workflow Context"],
          workflow_mode="execute",
 +        runtime_owner_type="RuntimeHandle",
 +        runtime_owner_path="runtime-handle",
          rotate_after_bytes=handle.config.session_rotate_after_bytes,
          auto_compaction_input_tokens_threshold=(
              handle.config.session_auto_compaction_input_tokens_threshold
          prompt_format="native",
          prompt_sections=["Runtime Config", "Workflow Context"],
          workflow_mode="plan",
 +        runtime_owner_type="Agent",
 +        runtime_owner_path="public-agent",
          rotate_after_bytes=handle.config.session_rotate_after_bytes,
          auto_compaction_input_tokens_threshold=(
              handle.config.session_auto_compaction_input_tokens_threshold
      assert handle.active_permission_mode == "prompt"
      assert handle.prompt_format == "native"
      assert handle.prompt_sections == ["Runtime Config", "Workflow Context"]
 +    assert handle.session.runtime_owner_type == "RuntimeHandle"
 +    assert handle.session.runtime_owner_path == "runtime-handle"
  def test_build_fresh_runtime_session_install_uses_current_owner_shell_state(

tests/test_session_state.pymodified

  from loader.runtime.completion_trace import CompletionTraceEntry
  from loader.runtime.evidence_provenance import EvidenceProvenance
  from loader.runtime.prompt_history import PromptSnapshot
 +from loader.runtime.runtime_handle import RuntimeHandle
  from loader.runtime.session import ConversationSession
  from loader.runtime.workflow_ledger import WorkflowLedger, WorkflowLedgerItem
  from loader.runtime.workflow_policy import WorkflowTimelineEntry
      session.update_runtime_state(
          current_task="Inspect permission history",
 +        runtime_owner_type="RuntimeHandle",
          permission_mode="allow",
          permission_prompting_enabled=True,
          permission_rule_counts={"allow": 2, "deny": 1, "ask": 4},
      assert reloaded.permission_rules_source == str(
          temp_dir / ".loader" / "permission-rules.json"
+     )
 +    assert reloaded.runtime_owner_type == "RuntimeHandle"
 +    assert reloaded.runtime_owner_path == "runtime-handle"
      assert reloaded.prompt_format == "native"
      assert reloaded.prompt_sections == [
          "Runtime Config",
+     ]
 +def test_resume_session_updates_runtime_owner_metadata(temp_dir: Path) -> None:
 +    agent = Agent(
 +        backend=ScriptedBackend(),
 +        config=AgentConfig(auto_context=False, stream=False),
 +        project_root=temp_dir,
 +    )
 +    agent.session.persist()
 +    session_id = agent.session.session_id
++
 +    handle = RuntimeHandle(
 +        backend=ScriptedBackend(),
 +        config=AgentConfig(auto_context=False, stream=False),
 +        project_root=temp_dir,
 +    )
++
 +    assert handle.resume_session(session_id) is True
++
 +    reloaded = ConversationSession.load(
 +        project_root=temp_dir,
 +        system_message_factory=_dummy_system,
 +        few_shot_factory=_dummy_few_shots,
 +        session_id=session_id,
 +    )
++
 +    assert reloaded is not None
 +    assert reloaded.runtime_owner_type == "RuntimeHandle"
 +    assert reloaded.runtime_owner_path == "runtime-handle"
++
++
  def test_session_prefers_canonical_workflow_timeline_for_completion_trace(
      temp_dir: Path,
  ) -> None:

tenseleyflow/loader / `c694b5a`

9 changed files