refactor: delegate workspace export

coding-agent/.opencode/prompts/tasks/runtime-workspace-export-service-task-packet.md

@@ -0,0 +1,52 @@
+Goal:
+Move workspace archive export orchestration out of `SessionManager` into a
+runtime/workspace service boundary while preserving active-turn rejection,
+cloud-binding validation, concurrent idle exports, export-in-progress guards,
+and post-export owner revalidation.
+
+Scope:
+- Add `RuntimeWorkspaceExportService` under `coding_agent.runs`.
+- Delegate `SessionManager.export_workspace_archive()` to the new service.
+- Add focused service tests and update ADR-0058 follow-up status.
+
+Out of scope:
+- Do not change HTTP workspace transfer endpoints, archive generation,
+  workspace cleanup, checkpoint behavior, runtime execution, daemon CLI,
+  executor selection, or event streams.
+- Do not change the `RuntimeTurnAdmissionService` export-in-progress guard.
+- Do not touch untracked `webui/` or contract fixture workspaces.
+
+Context:
+- ADRs:
+  - `docs/adr/0058-local-daemon-control-plane-executor-architecture.md`
+- Relevant files:
+  - `src/coding_agent/runs/workspace_export.py`
+  - `src/coding_agent/runs/__init__.py`
+  - `src/coding_agent/server/session_manager.py`
+  - `tests/coding_agent/test_runtime_workspace_export_service.py`
+  - `tests/ui/test_session_manager_public_api.py`
+  - `tests/ui/test_session_manager_runtime.py`
+- Postmortem notes:
+  - `postmortem/patterns/PM-0022-revalidate-event-stream-ownership-after-queue-attach.md`
+  - `postmortem/patterns/PM-0023-make-event-stream-cleanup-and-teardown-idempotent.md`
+
+Target tests:
+- `uv run pytest tests/coding_agent/test_runtime_workspace_export_service.py -v`
+- `uv run pytest tests/ui/test_session_manager_public_api.py -k "export_workspace_archive" -v`
+- `uv run pytest tests/ui/test_session_manager_runtime.py -k "workspace_export" -v`
+- `uv run pytest tests/ui/test_http_server_failover.py -k "events or event_queue or owner" -v`
+- `uv run pytest tests/ui/test_session_manager_public_api.py -k "event_queue or close_session or clear_sessions" -v`
+- `uv run ruff check src/coding_agent/runs/workspace_export.py src/coding_agent/runs/__init__.py src/coding_agent/server/session_manager.py tests/coding_agent/test_runtime_workspace_export_service.py`
+- `uv run ruff format --check src/coding_agent/runs/workspace_export.py src/coding_agent/runs/__init__.py src/coding_agent/server/session_manager.py tests/coding_agent/test_runtime_workspace_export_service.py`
+
+Loop policy:
+- Engineer implements the smallest correct change and runs the target tests.
+- Reviewer reviews only the resulting diff and affected tests.
+- Reviewer reports only P1/P2 findings.
+- Engineer fixes only accepted P1/P2 findings and reruns the same target tests.
+- Verifier reruns the exact target tests and reports pass/fail only.
+
+Stop conditions:
+- At most one review/fix/retest cycle.
+- Escalate architectural redirection or scope expansion to the human.
+- Ignore non-blocking optimization suggestions.

coding-agent/docs/adr/0058-local-daemon-control-plane-executor-architecture.md

@@ -359,6 +359,10 @@ This ADR does not implement that path.
     ensure invocation, runtime builder wiring, and session persistence wiring
     are delegated to `RuntimeEnsureOrchestrationService` instead of living
     inline in `SessionManager.ensure_session_runtime()`.
+  - Completed: workspace archive export admission, cloud-session validation,
+    export-in-progress bookkeeping, thread offload, and post-export owner
+    revalidation are delegated to `RuntimeWorkspaceExportService` instead of
+    living inline in `SessionManager.export_workspace_archive()`.
   - Completed: runtime control service construction for persistence, recovery,
     query, resume, attached-executor, cancel, and task-stopper services is
     delegated to `RuntimeControlServices` instead of `SessionManager` private

coding-agent/src/coding_agent/runs/__init__.py

@@ -66,6 +66,18 @@
     RuntimeEnsureSession,
     RuntimeEnsureSessionLoader,
 )
+from .workspace_export import (
+    RuntimeWorkspaceArchiveExporter,
+    RuntimeWorkspaceExportBegin,
+    RuntimeWorkspaceExportEnd,
+    RuntimeWorkspaceExportLock,
+    RuntimeWorkspaceExportLockProvider,
+    RuntimeWorkspaceExportOwnerAsserter,
+    RuntimeWorkspaceExportService,
+    RuntimeWorkspaceExportSession,
+    RuntimeWorkspaceExportSessionLoader,
+    RuntimeWorkspaceExportTask,
+)
 from .lifecycle import (
     RuntimeMaintenanceAdmissionBody,
     RuntimeMaintenanceAdmissionService,
@@ -332,6 +344,16 @@
     "RuntimeTurnStartSession",
     "RuntimeTurnStarter",
     "RuntimeTurnSession",
+    "RuntimeWorkspaceArchiveExporter",
+    "RuntimeWorkspaceExportBegin",
+    "RuntimeWorkspaceExportEnd",
+    "RuntimeWorkspaceExportLock",
+    "RuntimeWorkspaceExportLockProvider",
+    "RuntimeWorkspaceExportOwnerAsserter",
+    "RuntimeWorkspaceExportService",
+    "RuntimeWorkspaceExportSession",
+    "RuntimeWorkspaceExportSessionLoader",
+    "RuntimeWorkspaceExportTask",
     "RuntimeWireEventRecorder",
     "RuntimeWireEventSession",
     "RuntimeWireEventStore",

coding-agent/src/coding_agent/runs/workspace_export.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from types import TracebackType
+from typing import Protocol, TypeVar
+
+from coding_agent.environment.execution_binding import CloudWorkspaceBinding
+
+T = TypeVar("T")
+
+
+class RuntimeWorkspaceExportLock(Protocol):
+    def locked(self) -> bool: ...
+
+    async def __aenter__(self) -> object: ...
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> bool | None: ...
+
+
+class RuntimeWorkspaceExportTask(Protocol):
+    def done(self) -> bool: ...
+
+
+class RuntimeWorkspaceExportSession(Protocol):
+    execution_binding: object
+    turn_in_progress: bool
+    task: RuntimeWorkspaceExportTask | None
+
+
+RuntimeWorkspaceExportLockProvider = Callable[[str], RuntimeWorkspaceExportLock]
+RuntimeWorkspaceExportOwnerAsserter = Callable[[str], Awaitable[None]]
+RuntimeWorkspaceExportSessionLoader = Callable[
+    [str],
+    Awaitable[RuntimeWorkspaceExportSession],
+]
+RuntimeWorkspaceExportBegin = Callable[[str], None]
+RuntimeWorkspaceExportEnd = Callable[[str], None]
+RuntimeWorkspaceArchiveExporter = Callable[[CloudWorkspaceBinding], T]
+
+
+@dataclass(frozen=True)
+class RuntimeWorkspaceExportService:
+    turn_lock_for: RuntimeWorkspaceExportLockProvider
+    assert_owner: RuntimeWorkspaceExportOwnerAsserter
+    load_session: RuntimeWorkspaceExportSessionLoader
+    begin_export: RuntimeWorkspaceExportBegin
+    end_export: RuntimeWorkspaceExportEnd
+
+    async def export_archive(
+        self,
+        session_id: str,
+        export_archive: RuntimeWorkspaceArchiveExporter[T],
+    ) -> T:
+        turn_lock = self.turn_lock_for(session_id)
+        if turn_lock.locked():
+            raise RuntimeError("turn already in progress")
+
+        await self.assert_owner(session_id)
+        session = await self.load_session(session_id)
+        if session.turn_in_progress or (
+            session.task is not None and not session.task.done()
+        ):
+            raise RuntimeError("turn already in progress")
+        if not isinstance(session.execution_binding, CloudWorkspaceBinding):
+            raise ValueError("Workspace export requires cloud session")
+
+        self.begin_export(session_id)
+        try:
+            result = await asyncio.to_thread(
+                export_archive,
+                session.execution_binding,
+            )
+            await self.assert_owner(session_id)
+            return result
+        finally:
+            self.end_export(session_id)
+
+
+__all__ = [
+    "RuntimeWorkspaceArchiveExporter",
+    "RuntimeWorkspaceExportBegin",
+    "RuntimeWorkspaceExportEnd",
+    "RuntimeWorkspaceExportLock",
+    "RuntimeWorkspaceExportLockProvider",
+    "RuntimeWorkspaceExportOwnerAsserter",
+    "RuntimeWorkspaceExportService",
+    "RuntimeWorkspaceExportSession",
+    "RuntimeWorkspaceExportSessionLoader",
+    "RuntimeWorkspaceExportTask",
+]

coding-agent/src/coding_agent/server/session_manager.py

@@ -88,6 +88,7 @@
     RuntimeResumeContext as SessionResumeContext,
     RuntimeResumeOrchestrationService,
     RuntimeTurnAdmissionService,
+    RuntimeWorkspaceExportService,
     RuntimeWireEventRecorder,
     RunTarget,
     SessionRuntimeHandle,
@@ -825,6 +826,13 @@ async def runtime_session_is_recoverable(session_id: str) -> bool:
             assert_owner=self._assert_owner,
             load_session=self.get_session_async,
         )
+        self._runtime_workspace_export_service = RuntimeWorkspaceExportService(
+            turn_lock_for=self._turn_lock_for,
+            assert_owner=self._assert_owner,
+            load_session=self.get_session_async,
+            begin_export=self._begin_workspace_export,
+            end_export=self._end_workspace_export,
+        )
         self._runtime_maintenance_admission = RuntimeMaintenanceAdmissionService(
             turn_lock_for=self._turn_lock_for,
             assert_owner=self._assert_owner,
@@ -2794,23 +2802,10 @@ async def export_workspace_archive(
         session_id: str,
         export_archive: Callable[[CloudWorkspaceBinding], T],
     ) -> T:
-        turn_lock = self._turn_lock_for(session_id)
-        if turn_lock.locked():
-            raise RuntimeError("turn already in progress")
-
-        await self._assert_owner(session_id)
-        session = await self.get_session_async(session_id)
-        if session.turn_in_progress or (session.task and not session.task.done()):
-            raise RuntimeError("turn already in progress")
-        if not isinstance(session.execution_binding, CloudWorkspaceBinding):
-            raise ValueError("Workspace export requires cloud session")
-        self._begin_workspace_export(session_id)
-        try:
-            result = await asyncio.to_thread(export_archive, session.execution_binding)
-            await self._assert_owner(session_id)
-            return result
-        finally:
-            self._end_workspace_export(session_id)
+        return await self._runtime_workspace_export_service.export_archive(
+            session_id,
+            export_archive,
+        )
 
     async def list_checkpoints(self, session_id: str) -> list[CheckpointMeta]:
         session = await self.get_session_async(session_id)