refactor: restore runtime module boundaries and remove obsolete paths

AGENTS.md

@@ -105,7 +105,9 @@ execution, history, serialization, or the upstream dependency; see
   `app/access`, `app/assistant`, `app/streaming`, `app/voice`, `app/ingress`, `app/tasks`, `app/heartbeat`,
   `app/event_log`;
   the leaf modules `base`, `artifacts`, `host_context`, `principal`, `config`
-  and the `app/routes` package are exempt).
+  and the `app/routes` package are exempt). Ingress has no configurable
+  options and omits `config.py`; MCP uses its server configuration file and
+  omits `config.py` and `deps.py`.
   `config.py` (a frozen pydantic
   model nested into `AppSettings`), `deps.py` (FastAPI `Depends` accessors
   reading `app.state`), `factory.py` (`register_<name>(app_state,
@@ -118,7 +120,8 @@ execution, history, serialization, or the upstream dependency; see
   database, oauth, llm, history, media, decisions, actions, host_state, mcp, artifacts, tools,
   assistant, streaming, voice, ingress, tasks, heartbeat, event_log.
   `LifecycleManager` starts in that order, stops in reverse, and rolls back
-  on a failed start. `RuntimeSettings` is created after `start_all()` and
+  on a failed or cancelled start, including the partially started component.
+  `RuntimeSettings` is created after `start_all()` and
   attached through each service's `set_runtime_settings()`.
 - **Layering, bottom up.** `base` (lifecycle, protocols, resilience,
   exceptions, the domain event hub), `artifacts` (assistant profiles: the artifact schema, the
@@ -137,8 +140,11 @@ execution, history, serialization, or the upstream dependency; see
   are the top. `tests/unit/test_imports.py` asserts that nothing below the
   top layer imports `app` and that `_`-prefixed files stay inside their
   module (what another module needs is re-exported from the module's
-  `__init__.py` or a public file such as `services/tools/request_context.py`);
-  a new cross-package import must keep it green. `main.py` builds nothing at
+  `__init__.py` or a public file such as `services/tools/request_context.py`).
+  It also enforces the application-module dependency graph, including that
+  transport edges are never imported by application services. Ingress receives
+  its event sink from composition; Socket.IO delivery stays at the edge.
+  A new cross-package import must keep it green. `main.py` builds nothing at
   import: uvicorn runs the `create_asgi_app` factory, and loading `.env` is
   the CLI's job.
 - **Configuration has three tiers**, resolved once per request by
@@ -311,6 +317,8 @@ execution, history, serialization, or the upstream dependency; see
   `TASKS__MAX_CONCURRENT` at once; a turn that asks the host for an action
   fails its task. `task_finished` is published on `app.state.events` and is
   not forwarded to Socket.IO; nothing is steered into the parent session.
+  Its private `_runner.py` owns queue reservations, ordering, capacity and
+  cancellation; the service owns admission, records and request construction.
   With Postgres a restart marks unfinished tasks `interrupted` and replays
   nothing; without it tasks live in process memory. A *persistent agent*
   is the continuing form: the host starts it for a (profile, subject), one
@@ -456,7 +464,9 @@ names, ids or private hostnames).
 - **Shared turn pipeline.** Every request kind (new message, host-tool
   continuation, promoted steering) is described by `TurnPlanner`
   (`app/streaming/_turn.py`) and executed by `TurnRunner`
-  (`app/streaming/_runner.py`); `_agent_run.py` maps public Pydantic AI events
+  (`app/streaming/_runner.py`), which delegates native message capture and
+  persistence to `_snapshot.py` and diagnostics to `_telemetry.py`;
+  `_agent_run.py` maps public Pydantic AI events
   and applies steering/screen policy through native capability hooks;
   `_host_tool.py` keeps the pending host-tool state. The non-streaming
   `POST /api/chat` collects the same stream through

README.md

@@ -108,6 +108,13 @@ src/assistant_runtime/cli/       chat, serve, doctor, docs and migrations
 docs/ and alembic/               documentation and database migrations
 ```
 
+Within each service, `interface.py` owns the public operations and lifecycle;
+private modules hold the implementation. Database queries are grouped by record
+domain behind `services/database/repositories`. The turn runner delegates native
+message snapshots and diagnostics to separate modules, and background tasks use
+one scheduler for ordering, capacity and cancellation. Transport adapters own
+Socket.IO and voice protocol details. The import checks enforce these boundaries.
+
 ## License
 
 MIT — see [LICENSE](https://github.com/eandualem/assistant-runtime/blob/main/LICENSE).

src/assistant_runtime/app/assistant/_prompt_builder.py

@@ -22,7 +22,6 @@
     NavigationTarget,
 )
 from assistant_runtime.services.history.models import WorkingMemory
-from assistant_runtime.services.tools.models import ToolSet
 
 # --- Fragment builders ---
 
@@ -236,7 +235,6 @@ def host_context_prompt(host_context: dict[str, Any] | None) -> str:
 
 def build_system_prompt(
     *,
-    available_tools: ToolSet,
     session_context: dict[str, Any],
     working_memory: bool = False,
     artifacts: dict[str, str],
@@ -249,7 +247,6 @@ def build_system_prompt(
     memory (dynamic, last) when ``working_memory`` is set.
 
     Args:
-        available_tools: Tools available for this request.
         session_context: Session context dict (may contain working memory).
         working_memory: Whether the session's working memory is rendered.
         artifacts: Artifact name→text map, normally ``ArtifactService.active_texts()``.

src/assistant_runtime/app/assistant/_serialization.py

@@ -17,6 +17,7 @@
 from pydantic_ai.messages import (
     BinaryContent,
     DocumentUrl,
+    FileUrl,
     ImageUrl,
     ModelMessage,
     ModelMessagesTypeAdapter,
@@ -364,16 +365,8 @@ def _omit_tool_output_media(tool_name: str, content: Any) -> Any:
         return "[Inspected current screen]"
     if isinstance(content, BinaryContent):
         return "[Binary content omitted]"
-    try:
-        import pydantic_ai.messages as pydantic_messages
-    except ImportError:  # pragma: no cover
-        pydantic_messages = None  # type: ignore[assignment]
-    file_url_cls = getattr(pydantic_messages, "FileUrl", None)
-    if file_url_cls is not None and isinstance(content, file_url_cls):
-        result: dict[str, Any] = {"url": content.url}
-        if hasattr(content, "kind"):
-            result["kind"] = content.kind
-        return result
+    if isinstance(content, FileUrl):
+        return {"url": content.url, "kind": content.kind}
     if isinstance(content, dict):
         return {key: _omit_tool_output_media(tool_name, value) for key, value in content.items()}
     if isinstance(content, (list, tuple)):

src/assistant_runtime/app/assistant/_session_store.py

@@ -45,7 +45,7 @@
 _MAX_MEMORY_SESSIONS = 200
 
 
-# Stable prompt texts kept in process; without a database they are the only copy.
+# Reloadable database prompt cache; memory-only texts live with their session.
 _SNAPSHOT_CACHE_SIZE = 256
 
 
@@ -413,22 +413,24 @@ async def update_message(
             ]
             return record
 
-    async def save_prompt_snapshot(self, snapshot_hash: str, content: str) -> None:
-        """Keep the stable text a prompt record names (stored once per hash)."""
+    async def save_prompt_snapshot(self, session_id: str, snapshot_hash: str, content: str) -> None:
+        """Keep stable prompt text for as long as its session can reference it."""
+        if self._db is None:
+            self.get_context(session_id).setdefault("prompt_snapshots", {})[snapshot_hash] = content
+            return
         self._snapshots[snapshot_hash] = content
         self._snapshots.move_to_end(snapshot_hash)
         while len(self._snapshots) > _SNAPSHOT_CACHE_SIZE:
             self._snapshots.popitem(last=False)
-        if self._db is not None:
-            await self._db.save_prompt_snapshot(snapshot_hash, content)
+        await self._db.save_prompt_snapshot(snapshot_hash, content)
 
-    async def prompt_snapshot(self, snapshot_hash: str) -> str | None:
-        """The stable text under ``snapshot_hash``, from the cache or the database."""
+    async def prompt_snapshot(self, session_id: str, snapshot_hash: str) -> str | None:
+        """The stable text from its session, or the reloadable database cache."""
+        if self._db is None:
+            return self._sessions.get(session_id, {}).get("prompt_snapshots", {}).get(snapshot_hash)
         if snapshot_hash in self._snapshots:
             return self._snapshots[snapshot_hash]
-        if self._db is not None:
-            return await self._db.prompt_snapshot(snapshot_hash)
-        return None
+        return await self._db.prompt_snapshot(snapshot_hash)
 
     def get_message(self, session_id: str, message_id: str) -> MessageRecord | None:
         """The cached record of one message, or None when the session or message is not cached."""

src/assistant_runtime/app/assistant/interface.py

@@ -6,7 +6,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import inspect
 import time
 from typing import TYPE_CHECKING, Any
@@ -163,7 +162,7 @@ async def message_prompt(self, session_id: str, message_id: str) -> dict[str, An
         prompt = (record or {}).get("prompt")
         if not prompt:
             return None
-        snapshot = await self._sessions.prompt_snapshot(prompt["snapshot_hash"])
+        snapshot = await self._sessions.prompt_snapshot(session_id, prompt["snapshot_hash"])
         return {
             **prompt,
             "message_id": message_id,
@@ -177,7 +176,6 @@ async def preview_prompt(self, profile: str | None, subject: str | None) -> dict
         ).for_subject(subject)
         texts, extras, versions = await artifacts.prompt_inputs()
         prompt = build_system_prompt(
-            available_tools=self._tools.get_available_tools(None),
             session_context={},
             artifacts=texts,
             profile=artifacts.profile,
@@ -200,7 +198,6 @@ async def profile_prompt(self, profile: str) -> tuple[str, dict[str, int | None]
         artifacts = self._artifacts.for_profile(profile)
         texts, extras, versions = await artifacts.prompt_inputs()
         prompt = build_system_prompt(
-            available_tools=self._tools.get_available_tools(None),
             session_context={},
             artifacts=texts,
             profile=artifacts.profile,
@@ -268,9 +265,7 @@ async def prepare_agent_context(
                     if inspect.isawaitable(deps):
                         deps = await deps
 
-            inputs_task = asyncio.create_task(artifacts_service.prompt_inputs())
-
-            # 2. Config resolution can run while prompt inputs load.
+            # 2. Resolve synchronous configuration before loading prompt inputs.
             effective = resolve_effective_config(
                 self._config, self._runtime_settings, request_config
             )
@@ -282,10 +277,9 @@ async def prepare_agent_context(
             )
 
             # 3. Artifacts + system prompt
-            artifacts, artifact_extras, artifact_versions = await inputs_task
+            artifacts, artifact_extras, artifact_versions = await artifacts_service.prompt_inputs()
 
             prompt_result = build_system_prompt(
-                available_tools=available_tools,
                 session_context=session_context,
                 working_memory=effective.enable_working_memory,
                 artifacts=artifacts,

src/assistant_runtime/app/event_log/_store.py

@@ -8,6 +8,7 @@
 import asyncio
 import contextlib
 from collections.abc import AsyncIterator
+from copy import deepcopy
 from dataclasses import fields, replace
 from datetime import UTC, datetime
 from typing import Any, Protocol
@@ -52,19 +53,19 @@ async def transaction(self) -> AsyncIterator[InMemoryEventStore]:
     async def create_if_new(self, **values: Any) -> tuple[EventRecord, bool]:
         key = (values["source"], values["event_id"])
         if key in self._keys:
-            return self._events[self._keys[key]], False
-        record = EventRecord(id=self._next, created_at=datetime.now(UTC), **values)
+            return deepcopy(self._events[self._keys[key]]), False
+        record = EventRecord(id=self._next, created_at=datetime.now(UTC), **deepcopy(values))
         self._events[record.id] = record
         self._keys[key] = record.id
         self._next += 1
-        return record, True
+        return deepcopy(record), True
 
     async def get(self, event_id: int, *, lock: bool = False) -> EventRecord | None:
-        return self._events.get(event_id)
+        return deepcopy(self._events.get(event_id))
 
     async def update(self, event_id: int, **values: Any) -> EventRecord:
-        record = self._events[event_id] = replace(self._events[event_id], **values)
-        return record
+        record = self._events[event_id] = replace(self._events[event_id], **deepcopy(values))
+        return deepcopy(record)
 
     async def list(self, *, after: int, limit: int, **filters: Any) -> list[EventRecord]:
         records = [
@@ -73,7 +74,7 @@ async def list(self, *, after: int, limit: int, **filters: Any) -> list[EventRec
             if r.id > after
             and all(value is None or getattr(r, name) == value for name, value in filters.items())
         ]
-        return sorted(records, key=lambda r: r.id)[:limit]
+        return deepcopy(sorted(records, key=lambda r: r.id)[:limit])
 
 
 class DatabaseEventStore:

src/assistant_runtime/app/ingress/factory.py

@@ -17,7 +17,7 @@ async def register_ingress(app_state: Any, lifecycle: LifecycleManager) -> None:
         assistant_service=app_state.assistant_service,
         streaming_service=app_state.streaming_service,
         database_service=getattr(app_state, "database_service", None),
-        socket_server=getattr(app_state, "sio", None),
+        event_sink=getattr(app_state, "ingress_event_sink", None),
     )
     app_state.ingress_service = service
     app_state.streaming_service.attach_ingress(service)

src/assistant_runtime/app/ingress/interface.py

@@ -14,13 +14,13 @@
 import asyncio
 import contextlib
 import uuid
+from collections.abc import Awaitable, Callable
 from typing import TYPE_CHECKING, Any
 
 from loguru import logger
 
 from assistant_runtime.app.assistant.exceptions import SessionError
 from assistant_runtime.app.assistant.models import AssistantRequest
-from assistant_runtime.app.socketio_server import socket_event_name
 
 if TYPE_CHECKING:
     from assistant_runtime.app.assistant.interface import AssistantService
@@ -42,12 +42,12 @@ def __init__(
         assistant_service: AssistantService,
         streaming_service: StreamingService,
         database_service: DatabaseService | None = None,
-        socket_server: Any | None = None,
+        event_sink: Callable[[str, dict[str, Any]], Awaitable[None]] | None = None,
     ) -> None:
         self._assistant = assistant_service
         self._streaming = streaming_service
         self._db = database_service
-        self._sio = socket_server
+        self._event_sink = event_sink
         self._queued: list[dict[str, Any]] = []  # the inbox when there is no database
         self._background: set[asyncio.Task[None]] = set()
         self._started = False
@@ -186,15 +186,12 @@ def _run_in_background(self, request: AssistantRequest) -> None:
         task.add_done_callback(self._background.discard)
 
     async def _run(self, request: AssistantRequest) -> None:
-        """Run the promoted turn; events go to the session's Socket.IO room, if any."""
-        room = f"session:{request.session_id}"
+        """Run the promoted turn and forward its events through the configured sink."""
         try:
             async for event in self._streaming.stream_message(request):
-                if self._sio is not None:
+                if self._event_sink is not None:
                     with contextlib.suppress(Exception):
-                        await self._sio.emit(
-                            socket_event_name(event), event, room=room, namespace="/assistant"
-                        )
+                        await self._event_sink(request.session_id, event)
         except asyncio.CancelledError:
             raise
         except Exception as e: