refactor: add typed upstream protocol adapter boundary

docs/maintainer-architecture.md

@@ -20,7 +20,8 @@ flowchart TD
     end
 
     subgraph Upstream["OpenCode Upstream Layer"]
-        UpstreamClient["opencode_upstream_client.py"]
+        UpstreamClient["opencode_upstream_client.py (V1 facade)"]
+        UpstreamContracts["upstream/contracts.py"]
     end
 
     subgraph Extensions["Extension / Contract Layer"]
@@ -37,7 +38,8 @@ flowchart TD
 
     Inbound --> Execution
     Inbound --> Extensions
-    Execution --> Upstream
+    Execution --> UpstreamContracts
+    UpstreamClient --> UpstreamContracts
     Inbound --> Persistence
     Execution --> Persistence
 ```
@@ -53,8 +55,9 @@ flowchart TD
    - `execution/coordinator.py`
    - `execution/stream_runtime.py`
    - `execution/tool_orchestration.py`
-5. `opencode_upstream_client.py` sends requests to the upstream OpenCode runtime.
-6. The adapter maps upstream responses back into A2A tasks, messages, artifacts, and stream events.
+5. `opencode_upstream_client.py` preserves the public V1 client facade and sends requests to the OpenCode runtime.
+6. The internal `upstream/contracts.py` separates prompt receipts, terminal outcomes, and normalized execution updates; V1 event decoding occurs before updates reach the shared stream runtime.
+7. The execution layer maps outcomes and updates into A2A tasks, messages, artifacts, and stream events.
 
 ### JSON-RPC Extension Path
 

src/opencode_a2a/execution/coordinator.py

@@ -22,9 +22,23 @@
     TaskStatusUpdateEvent,
 )
 
-from ..opencode_upstream_client import UpstreamConcurrencyLimitError, UpstreamContractError
+from ..opencode_upstream_client import (
+    OpencodeMessage,
+    UpstreamConcurrencyLimitError,
+    UpstreamContractError,
+)
+from ..upstream.contracts import (
+    ExecutionOutcome,
+    ExecutionState,
+    PromptSubmission,
+    UpstreamAdapter,
+)
 from .event_helpers import _enqueue_artifact_update
-from .stream_events import _extract_token_usage, _extract_upstream_error_from_response
+from .stream_events import (
+    _extract_token_usage,
+    _extract_upstream_error_from_response,
+    _task_state_from_execution_state,
+)
 from .stream_state import (
     _build_output_metadata,
     _build_stream_artifact_metadata,
@@ -36,12 +50,14 @@
     _await_stream_terminal_signal,
     _format_upstream_error,
     _StreamTerminalSignal,
+    _UpstreamInBandError,
 )
 
 if TYPE_CHECKING:
     from .executor import OpencodeAgentExecutor
 
 logger = logging.getLogger(__name__)
+_MISSING = object()
 
 
 @dataclass(frozen=True)
@@ -134,19 +150,44 @@ async def run(self) -> None:
                 if self._prepared.streaming_request:
                     send_kwargs["timeout_override"] = self._executor._client.stream_timeout
 
-                if not self._prepared.use_structured_parts and not turn_request_parts:
+                prompt_text = (
+                    user_text
+                    if not self._prepared.use_structured_parts and not turn_request_parts
+                    else user_text or None
+                )
+                prompt_kwargs = dict(send_kwargs)
+                if self._prepared.use_structured_parts or turn_request_parts:
+                    prompt_kwargs["parts"] = turn_request_parts
+
+                upstream: UpstreamAdapter = self._executor._client
+                submit_prompt = getattr(upstream, "submit_prompt", _MISSING)
+                response: ExecutionOutcome | OpencodeMessage
+                if submit_prompt is _MISSING:
                     response = await self._executor._client.send_message(
                         self._session_id,
-                        user_text,
-                        **send_kwargs,
+                        prompt_text,
+                        **prompt_kwargs,
                     )
                 else:
-                    response = await self._executor._client.send_message(
+                    if not callable(submit_prompt):
+                        raise UpstreamContractError(
+                            "OpenCode submit_prompt must be callable when provided"
+                        )
+                    submission = await submit_prompt(
                         self._session_id,
-                        user_text or None,
-                        parts=turn_request_parts,
-                        **send_kwargs,
+                        prompt_text,
+                        **prompt_kwargs,
                     )
+                    if not isinstance(submission, PromptSubmission):
+                        raise UpstreamContractError(
+                            "OpenCode submit_prompt must return PromptSubmission"
+                        )
+                    if submission.outcome is None:
+                        raise RuntimeError(
+                            "Upstream accepted the prompt without a terminal outcome; "
+                            "asynchronous observation is not supported"
+                        )
+                    response = submission.outcome
 
                 if self._pending_preferred_claim:
                     await self._executor._session_manager.finalize_preferred_session_binding(
@@ -156,8 +197,20 @@ async def run(self) -> None:
                     )
                     self._pending_preferred_claim = False
 
+                if (
+                    isinstance(response, ExecutionOutcome)
+                    and response.state != ExecutionState.COMPLETED
+                ):
+                    await self._handle_response(response)
+                    break
+
+                response_payload = (
+                    dict(response.response or {})
+                    if isinstance(response, ExecutionOutcome)
+                    else response.raw
+                )
                 tool_results = await maybe_handle_tools(
-                    response.raw,
+                    response_payload,
                     a2a_client_manager=self._executor._a2a_client_manager,
                 )
                 if tool_results:
@@ -300,18 +353,51 @@ async def _bind_session(self) -> None:
             )
 
     async def _handle_response(self, response: Any) -> None:
-        response_text = response.text or ""
+        response_text = (
+            response.output if isinstance(response, ExecutionOutcome) else response.text
+        ) or ""
         resolved_message_id = self._stream_state.resolve_message_id(response.message_id)
-        response_error = _extract_upstream_error_from_response(response.raw)
+        response_session_id = (
+            response.session.id if isinstance(response, ExecutionOutcome) else response.session_id
+        )
+        response_payload = (
+            dict(response.response or {})
+            if isinstance(response, ExecutionOutcome)
+            else response.raw
+        )
+        response_error: _UpstreamInBandError | None
+        if isinstance(response, ExecutionOutcome) and response.state != ExecutionState.COMPLETED:
+            response_error = _UpstreamInBandError(
+                error_type=response.error_type or "UPSTREAM_EXECUTION_ERROR",
+                state=_task_state_from_execution_state(response.state),
+                message=response.error_message or "OpenCode execution did not complete.",
+                upstream_status=response.upstream_status,
+            )
+        elif (
+            isinstance(response, ExecutionOutcome)
+            and response.error_type is not None
+            and response.error_state is not None
+            and response.error_message is not None
+        ):
+            response_error = _UpstreamInBandError(
+                error_type=response.error_type,
+                state=_task_state_from_execution_state(response.error_state),
+                message=response.error_message,
+                upstream_status=response.upstream_status,
+            )
+        else:
+            response_error = _extract_upstream_error_from_response(response_payload)
         resolved_token_usage = _merge_token_usage(
-            _extract_token_usage(response.raw),
+            response.usage
+            if isinstance(response, ExecutionOutcome)
+            else _extract_token_usage(response_payload),
             self._stream_state.token_usage,
         )
 
         logger.debug(
             "OpenCode response task_id=%s session_id=%s message_id=%s text_len=%s",
             self._task_id,
-            response.session_id,
+            response_session_id,
             resolved_message_id,
             len(response_text),
         )
@@ -331,15 +417,14 @@ async def _handle_response(self, response: Any) -> None:
 
         if self._prepared.streaming_request:
             await self._handle_streaming_response(
-                response=response,
                 response_text=response_text,
                 resolved_message_id=resolved_message_id,
                 resolved_token_usage=resolved_token_usage,
             )
             return
 
         await self._handle_non_streaming_response(
-            response=response,
+            session_id=response_session_id,
             response_text=response_text,
             resolved_message_id=resolved_message_id,
             resolved_token_usage=resolved_token_usage,
@@ -348,14 +433,12 @@ async def _handle_response(self, response: Any) -> None:
     async def _handle_streaming_response(
         self,
         *,
-        response: Any,
         response_text: str,
         resolved_message_id: str,
         resolved_token_usage: Mapping[str, Any] | None,
     ) -> None:
         from .stream_events import BlockType
 
-        del response
         if self._stream_terminal_signal is None:
             raise RuntimeError("Streaming terminal signal was not initialized")
 
@@ -439,7 +522,7 @@ async def _handle_streaming_response(
     async def _handle_non_streaming_response(
         self,
         *,
-        response: Any,
+        session_id: str,
         response_text: str,
         resolved_message_id: str,
         resolved_token_usage: Mapping[str, Any] | None,
@@ -470,7 +553,7 @@ async def _handle_non_streaming_response(
             history=history,
             artifacts=[artifact],
             metadata=_build_output_metadata(
-                session_id=response.session_id,
+                session_id=session_id,
                 usage=resolved_token_usage,
                 include_session_metadata=self._prepared.emit_session_metadata,
                 include_streaming_metadata=self._prepared.emit_streaming_metadata,

src/opencode_a2a/execution/stream_events.py

@@ -7,6 +7,14 @@
 
 from a2a.types import TaskState
 
+from ..upstream.contracts import (
+    ExecutionOutcome,
+    ExecutionState,
+    ExecutionUpdate,
+    Intervention,
+    SessionRef,
+    UpdateKind,
+)
 from .stream_state import BlockType
 from .upstream_error_translator import (
     _format_inband_upstream_error,
@@ -541,3 +549,102 @@ def _extract_tool_part_payload(part: Mapping[str, Any]) -> dict[str, Any] | None
     if not payload:
         return None
     return payload
+
+
+def normalize_v1_event(event: Mapping[str, Any], *, log_limit: int = 0) -> ExecutionUpdate | None:
+    """Decode a V1 event into the protocol-neutral execution update contract."""
+    _log_stream_event_debug(event, limit=log_limit)
+    event_type = event.get("type")
+    props = event.get("properties")
+    if not isinstance(event_type, str) or not isinstance(props, Mapping):
+        return None
+
+    part_value = props.get("part")
+    part = part_value if isinstance(part_value, Mapping) else {}
+    session_id = _extract_event_session_id(event)
+    message_id = _extract_stream_message_id(part, props)
+    part_id = _extract_stream_part_id(part, props)
+    part_type = _extract_stream_part_type(part, props)
+    role = _normalize_role(part.get("role") or props.get("role"))
+    delta_value = props.get("delta")
+    delta = delta_value if isinstance(delta_value, str) else None
+    field_value = props.get("field")
+    delta_field = field_value if isinstance(field_value, str) else None
+
+    kind = UpdateKind.OTHER
+    if event_type == "message.part.updated":
+        kind = UpdateKind.PART_UPDATED
+    elif event_type == "message.part.delta":
+        kind = UpdateKind.PART_DELTA
+
+    error = _extract_upstream_error_from_event(event)
+    signal = _extract_stream_terminal_signal(event)
+    terminal = None
+    if signal is not None:
+        terminal = ExecutionOutcome(
+            session=SessionRef(session_id or ""),
+            state=_execution_state(signal.state),
+            output="",
+            error_type=signal.error_type,
+            error_message=signal.message,
+            error_state=_execution_state(signal.state),
+            upstream_status=signal.upstream_status,
+        )
+
+    asked = _extract_interrupt_asked_event(event)
+    intervention = (
+        Intervention(
+            request_id=asked["request_id"],
+            type=asked["interrupt_type"],
+            details=asked["details"],
+        )
+        if asked is not None
+        else None
+    )
+    resolved = _extract_interrupt_resolved_event(event)
+    progress = _extract_progress_metadata(part, props) if isinstance(part_value, Mapping) else None
+
+    return ExecutionUpdate(
+        kind=kind,
+        session_id=session_id,
+        message_id=message_id,
+        part_id=part_id,
+        part_type=part_type,
+        role=role,
+        delta_field=delta_field,
+        delta=delta,
+        snapshot=_extract_stream_snapshot_text(part),
+        tool_payload=_extract_tool_part_payload(part),
+        progress=progress,
+        progress_identity=(_build_progress_identity(part, props) if progress is not None else None),
+        usage=_extract_token_usage(event),
+        error_type=error.error_type if error is not None else None,
+        error_message=error.message if error is not None else None,
+        error_state=_execution_state(error.state) if error is not None else None,
+        upstream_status=error.upstream_status if error is not None else None,
+        terminal=terminal,
+        intervention=intervention,
+        resolved_request_id=resolved["request_id"] if resolved is not None else None,
+        resolved_intervention_type=(resolved["interrupt_type"] if resolved is not None else None),
+        resolution=resolved["resolution"] if resolved is not None else None,
+    )
+
+
+def _execution_state(state: TaskState) -> ExecutionState:
+    if state == TaskState.TASK_STATE_COMPLETED:
+        return ExecutionState.COMPLETED
+    if state == TaskState.TASK_STATE_CANCELED:
+        return ExecutionState.CANCELLED
+    if state == TaskState.TASK_STATE_AUTH_REQUIRED:
+        return ExecutionState.AUTH_REQUIRED
+    return ExecutionState.FAILED
+
+
+def _task_state_from_execution_state(state: ExecutionState) -> TaskState:
+    if state == ExecutionState.COMPLETED:
+        return TaskState.TASK_STATE_COMPLETED
+    if state == ExecutionState.CANCELLED:
+        return TaskState.TASK_STATE_CANCELED
+    if state == ExecutionState.AUTH_REQUIRED:
+        return TaskState.TASK_STATE_AUTH_REQUIRED
+    return TaskState.TASK_STATE_FAILED