refactor(policy): centralize sandbox enforcement and validate config

src/opencode_a2a/config.py

@@ -3,10 +3,11 @@
 import json
 from typing import Annotated, Any, Literal, cast
 
-from pydantic import Field, field_validator
+from pydantic import BeforeValidator, Field, model_validator
 from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
 
 from opencode_a2a import __version__
+from opencode_a2a.sandbox_policy import SandboxPolicy
 
 SandboxMode = Literal[
     "unknown",
@@ -34,7 +35,6 @@
     "custom",
 ]
 OutsideWorkspaceAccess = Literal["unknown", "allowed", "disallowed", "custom"]
-DeclaredStringList = Annotated[tuple[str, ...], NoDecode]
 
 
 def _parse_declared_list(value: Any) -> tuple[str, ...]:
@@ -59,6 +59,9 @@ def _parse_declared_list(value: Any) -> tuple[str, ...]:
     raise TypeError("Expected a comma-separated string, JSON array, or sequence.")
 
 
+DeclaredStringList = Annotated[tuple[str, ...], NoDecode, BeforeValidator(_parse_declared_list)]
+
+
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(
         env_prefix="",
@@ -173,18 +176,12 @@ class Settings(BaseSettings):
         alias="A2A_CLIENT_SUPPORTED_TRANSPORTS",
     )
 
-    @field_validator(
-        "a2a_sandbox_writable_roots",
-        "a2a_network_allowed_domains",
-        "a2a_client_supported_transports",
-        mode="before",
-    )
-    @classmethod
-    def _normalize_declared_lists(cls, value: Any) -> tuple[str, ...]:
-        return _parse_declared_list(value)
+    @model_validator(mode="after")
+    def _validate_sandbox_policy(self) -> Settings:
+        SandboxPolicy.from_settings(self).validate_configuration()
+        return self
 
     @classmethod
     def from_env(cls) -> Settings:
-        # BaseSettings constructor loads values from env and applies validation.
         settings_cls: type[BaseSettings] = cls
         return cast(Settings, settings_cls())

src/opencode_a2a/contracts/extensions.py

@@ -302,10 +302,6 @@ class DeploymentConditionalMethod:
     toggle: str
     reason_when_disabled: str = "disabled_by_configuration"
 
-    @property
-    def availability(self) -> str:
-        return "enabled" if self.enabled else "disabled"
-
     def control_method_flag(self) -> dict[str, Any]:
         return {
             "enabled_by_default": False,
@@ -315,7 +311,7 @@ def control_method_flag(self) -> dict[str, Any]:
     def method_retention(self) -> dict[str, Any]:
         return {
             "surface": "extension",
-            "availability": self.availability,
+            "availability": "enabled" if self.enabled else "disabled",
             "retention": "deployment-conditional",
             "extension_uri": self.extension_uri,
             "toggle": self.toggle,
@@ -411,7 +407,7 @@ def build_capability_snapshot(*, runtime_profile: RuntimeProfile) -> JsonRpcCapa
         conditional_methods={
             SESSION_CONTROL_METHODS["shell"]: DeploymentConditionalMethod(
                 method=SESSION_CONTROL_METHODS["shell"],
-                enabled=runtime_profile.session_shell_enabled,
+                enabled=runtime_profile.session_shell.enabled,
                 extension_uri=SESSION_QUERY_EXTENSION_URI,
                 toggle=SESSION_SHELL_TOGGLE,
             )

src/opencode_a2a/execution/executor.py

@@ -34,8 +34,8 @@
     map_a2a_parts_to_opencode_parts,
     summarize_a2a_parts,
 )
+from ..sandbox_policy import SandboxPolicy
 from .event_helpers import _enqueue_artifact_update
-from .policy import PolicyEnforcer
 from .request_context import (
     _build_history,
     _extract_opencode_directory,
@@ -207,7 +207,7 @@ async def run(self) -> None:
                     )
 
                 if self._pending_preferred_claim:
-                    await self._executor._finalize_preferred_session_binding(
+                    await self._executor._session_manager.finalize_preferred_session_binding(
                         identity=self._prepared.identity,
                         context_id=self._context_id,
                         session_id=self._session_id,
@@ -296,14 +296,16 @@ async def _bind_session(self) -> None:
         (
             self._session_id,
             self._pending_preferred_claim,
-        ) = await self._executor._get_or_create_session(
+        ) = await self._executor._session_manager.get_or_create_session(
             self._prepared.identity,
             self._context_id,
             self._prepared.session_title or self._prepared.user_text,
             preferred_session_id=self._prepared.bound_session_id,
             directory=self._prepared.directory,
         )
-        self._session_lock = await self._executor._get_session_lock(self._session_id)
+        self._session_lock = await self._executor._session_manager.get_session_lock(
+            self._session_id
+        )
         await self._session_lock.acquire()
         async with self._executor._lock:
             self._executor._running_session_ids[self._execution_key] = self._session_id
@@ -500,7 +502,7 @@ async def _handle_non_streaming_response(
     async def _cleanup(self) -> None:
         if self._pending_preferred_claim and self._session_id:
             with suppress(Exception):
-                await self._executor._release_preferred_session_claim(
+                await self._executor._session_manager.release_preferred_session_claim(
                     identity=self._prepared.identity,
                     session_id=self._session_id,
                 )
@@ -534,7 +536,10 @@ def __init__(
         self._streaming_enabled = streaming_enabled
         self._cancel_abort_timeout_seconds = max(0.0, float(cancel_abort_timeout_seconds))
         self._a2a_client_manager = a2a_client_manager
-        self._policy = PolicyEnforcer(client=client)
+        self._sandbox_policy = SandboxPolicy.from_settings(
+            client.settings,
+            workspace_root=client.directory,
+        )
         self._session_manager = SessionManager(
             client=client,
             session_cache_ttl_seconds=session_cache_ttl_seconds,
@@ -560,23 +565,6 @@ def _emit_metric(
     ) -> None:
         _emit_metric(name, value, **labels)
 
-    def _resolve_and_validate_directory(self, requested: str | None) -> str | None:
-        return self._policy.resolve_directory(requested)
-
-    def resolve_directory_for_control(self, requested: str | None) -> str | None:
-        return self._policy.resolve_directory_for_control(requested)
-
-    async def claim_session_for_control(self, *, identity: str, session_id: str) -> bool:
-        return await self._claim_preferred_session(identity=identity, session_id=session_id)
-
-    async def finalize_session_for_control(self, *, identity: str, session_id: str) -> None:
-        """Finalize control-session ownership after upstream call succeeds."""
-        await self._finalize_session_claim(identity=identity, session_id=session_id)
-
-    async def release_session_for_control(self, *, identity: str, session_id: str) -> None:
-        """Release pending control-session ownership on failure."""
-        await self._release_preferred_session_claim(identity=identity, session_id=session_id)
-
     async def _maybe_handle_tools(
         self, raw_response: dict[str, Any]
     ) -> list[dict[str, Any]] | None:
@@ -777,7 +765,10 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non
         requested_dir = _extract_opencode_directory(context)
 
         try:
-            directory = self._resolve_and_validate_directory(requested_dir)
+            directory = self._sandbox_policy.resolve_directory(
+                requested_dir,
+                default_directory=self._client.directory,
+            )
         except ValueError as e:
             logger.warning("Directory validation failed: %s", e)
             await self._emit_error(
@@ -952,57 +943,6 @@ async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None
                 abort_outcome=abort_outcome,
             )
 
-    async def _get_or_create_session(
-        self,
-        identity: str,
-        context_id: str,
-        title: str,
-        *,
-        preferred_session_id: str | None = None,
-        directory: str | None = None,
-    ) -> tuple[str, bool]:
-        return await self._session_manager.get_or_create_session(
-            identity,
-            context_id,
-            title,
-            preferred_session_id=preferred_session_id,
-            directory=directory,
-        )
-
-    async def _finalize_preferred_session_binding(
-        self,
-        *,
-        identity: str,
-        context_id: str,
-        session_id: str,
-    ) -> None:
-        await self._session_manager.finalize_preferred_session_binding(
-            identity=identity,
-            context_id=context_id,
-            session_id=session_id,
-        )
-
-    async def _claim_preferred_session(self, *, identity: str, session_id: str) -> bool:
-        return await self._session_manager.claim_preferred_session(
-            identity=identity,
-            session_id=session_id,
-        )
-
-    async def _finalize_session_claim(self, *, identity: str, session_id: str) -> None:
-        await self._session_manager.finalize_session_claim(
-            identity=identity,
-            session_id=session_id,
-        )
-
-    async def _release_preferred_session_claim(self, *, identity: str, session_id: str) -> None:
-        await self._session_manager.release_preferred_session_claim(
-            identity=identity,
-            session_id=session_id,
-        )
-
-    async def _get_session_lock(self, session_id: str) -> asyncio.Lock:
-        return await self._session_manager.get_session_lock(session_id)
-
     async def _emit_error(
         self,
         event_queue: EventQueue,

src/opencode_a2a/profile/runtime.py

@@ -4,6 +4,7 @@
 from typing import Any
 
 from ..config import Settings
+from ..sandbox_policy import SandboxPolicy
 
 PROFILE_ID = "opencode-a2a-single-tenant-coding-v1"
 DEPLOYMENT_ID = "single_tenant_shared_workspace"
@@ -165,10 +166,6 @@ class RuntimeProfile:
     service_features: ServiceFeaturesProfile
     runtime_context: RuntimeContext
 
-    @property
-    def session_shell_enabled(self) -> bool:
-        return self.session_shell.enabled
-
     def runtime_features_dict(self) -> dict[str, Any]:
         return {
             "directory_binding": self.directory_binding.as_dict(),
@@ -206,6 +203,10 @@ def health_payload(
 
 
 def build_runtime_profile(settings: Settings) -> RuntimeProfile:
+    sandbox_policy = SandboxPolicy.from_settings(settings)
+    shell_enabled = sandbox_policy.is_session_shell_enabled(
+        enabled_by_config=settings.a2a_enable_session_shell,
+    )
     directory_scope = (
         "workspace_root_or_descendant"
         if settings.a2a_allow_directory_override
@@ -219,8 +220,8 @@ def build_runtime_profile(settings: Settings) -> RuntimeProfile:
             scope=directory_scope,
         ),
         session_shell=SessionShellProfile(
-            enabled=settings.a2a_enable_session_shell,
-            availability="enabled" if settings.a2a_enable_session_shell else "disabled",
+            enabled=shell_enabled,
+            availability="enabled" if shell_enabled else "disabled",
         ),
         execution_environment=ExecutionEnvironmentProfile(
             sandbox=SandboxProfile(