From 81887077acf838fc8b38421ab4517985ccf70723 Mon Sep 17 00:00:00 2001 From: Tom Brandenburg Date: Sun, 25 Oct 2026 11:26:04 +0100 Subject: [PATCH 1/3] refactor(upstream): add typed adapter boundary (#2) --- docs/maintainer-architecture.md | 11 +- src/opencode_a2a/execution/coordinator.py | 94 +++++-- src/opencode_a2a/execution/stream_events.py | 107 ++++++++ src/opencode_a2a/execution/stream_runtime.py | 246 +++++++++---------- src/opencode_a2a/opencode_upstream_client.py | 80 ++++++ src/opencode_a2a/upstream/__init__.py | 25 ++ src/opencode_a2a/upstream/contracts.py | 118 +++++++++ tests/support/streaming_output.py | 61 +++++ tests/upstream/test_upstream_contracts.py | 126 ++++++++++ 9 files changed, 717 insertions(+), 151 deletions(-) create mode 100644 src/opencode_a2a/upstream/__init__.py create mode 100644 src/opencode_a2a/upstream/contracts.py create mode 100644 tests/upstream/test_upstream_contracts.py diff --git a/docs/maintainer-architecture.md b/docs/maintainer-architecture.md index b30054cd..8ad64da5 100644 --- a/docs/maintainer-architecture.md +++ b/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 diff --git a/src/opencode_a2a/execution/coordinator.py b/src/opencode_a2a/execution/coordinator.py index e463f96c..29ede9f2 100644 --- a/src/opencode_a2a/execution/coordinator.py +++ b/src/opencode_a2a/execution/coordinator.py @@ -23,8 +23,17 @@ ) from ..opencode_upstream_client import UpstreamConcurrencyLimitError, UpstreamContractError +from ..upstream.contracts import ( + ExecutionOutcome, + 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,6 +45,7 @@ _await_stream_terminal_signal, _format_upstream_error, _StreamTerminalSignal, + _UpstreamInBandError, ) if TYPE_CHECKING: @@ -134,19 +144,31 @@ 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: - response = await self._executor._client.send_message( - self._session_id, - user_text, - **send_kwargs, - ) - else: - response = await self._executor._client.send_message( + 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", None) + submission = ( + await submit_prompt(self._session_id, prompt_text, **prompt_kwargs) + if callable(submit_prompt) + else None + ) + response = ( + submission.outcome + if isinstance(submission, PromptSubmission) + else await self._executor._client.send_message( self._session_id, - user_text or None, - parts=turn_request_parts, - **send_kwargs, + prompt_text, + **prompt_kwargs, ) + ) if self._pending_preferred_claim: await self._executor._session_manager.finalize_preferred_session_binding( @@ -156,8 +178,13 @@ async def run(self) -> None: ) self._pending_preferred_claim = False + 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 +327,42 @@ 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( + error_type=response.error_type, + state=_task_state_from_execution_state(response.error_state), + message=response.error_message, + upstream_status=response.upstream_status, + ) + if isinstance(response, ExecutionOutcome) + and response.error_type is not None + and response.error_state is not None + and response.error_message is not None + else _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,7 +382,6 @@ 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, @@ -339,7 +389,7 @@ async def _handle_response(self, response: Any) -> None: 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 +398,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 +487,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 +518,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, diff --git a/src/opencode_a2a/execution/stream_events.py b/src/opencode_a2a/execution/stream_events.py index 4d5458b5..40239afc 100644 --- a/src/opencode_a2a/execution/stream_events.py +++ b/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 diff --git a/src/opencode_a2a/execution/stream_runtime.py b/src/opencode_a2a/execution/stream_runtime.py index 54e76055..cf811998 100644 --- a/src/opencode_a2a/execution/stream_runtime.py +++ b/src/opencode_a2a/execution/stream_runtime.py @@ -1,10 +1,11 @@ from __future__ import annotations import asyncio +import inspect import json import logging from collections import defaultdict -from collections.abc import Awaitable, Callable, Mapping +from collections.abc import AsyncIterator, Awaitable, Callable, Mapping from typing import Any from a2a.server.events.event_queue import EventQueue @@ -17,26 +18,12 @@ from ..a2a_utils import make_data_part from ..interrupt_request_tracker import BoundInterruptRequestTracker +from ..upstream.contracts import ExecutionUpdate, UpdateKind from .event_helpers import _enqueue_artifact_update from .stream_events import ( BlockType, - _build_progress_identity, - _extract_event_session_id, - _extract_interrupt_asked_event, - _extract_interrupt_resolved_event, - _extract_progress_metadata, - _extract_stream_message_id, - _extract_stream_part_id, - _extract_stream_part_type, - _extract_stream_session_id, - _extract_stream_snapshot_text, - _extract_stream_terminal_signal, - _extract_token_usage, - _extract_tool_part_payload, - _extract_upstream_error_from_event, - _log_stream_event_debug, _map_part_type_to_block_type, - _normalize_role, + _task_state_from_execution_state, ) from .stream_state import ( _build_output_metadata, @@ -46,11 +33,33 @@ _StreamOutputState, _StreamPartState, ) -from .upstream_error_translator import _StreamTerminalSignal +from .upstream_error_translator import ( + _StreamTerminalSignal, + _UpstreamInBandError, +) logger = logging.getLogger("opencode_a2a.execution.executor") +async def _client_stream_updates(client: Any, **kwargs: Any) -> AsyncIterator[ExecutionUpdate]: + stream_updates = getattr(client, "stream_updates", None) + if callable(stream_updates) and not inspect.iscoroutinefunction(stream_updates): + updates = stream_updates(**kwargs) + if hasattr(updates, "__aiter__"): + async for update in updates: + yield update + return + + from .stream_events import normalize_v1_event + + settings = getattr(client, "settings", None) + log_limit = getattr(settings, "a2a_log_body_limit", 0) + async for event in client.stream_events(**kwargs): + update = normalize_v1_event(event, log_limit=log_limit) + if update is not None: + yield update + + class StreamRuntime: def __init__( self, @@ -262,12 +271,11 @@ def _resolve_stream_artifact_id( def _upsert_part_state( *, part_id: str, - part: Mapping[str, Any], - props: Mapping[str, Any], + part_type: str | None, role: str | None, message_id: str | None, ) -> _StreamPartState | None: - block_type = _map_part_type_to_block_type(_extract_stream_part_type(part, props)) + block_type = _map_part_type_to_block_type(part_type) if block_type is None: return None state = part_states.get(part_id) @@ -368,10 +376,9 @@ def _snapshot_chunks( def _tool_chunks( *, state: _StreamPartState, - part: Mapping[str, Any], + tool_payload: Mapping[str, Any] | None, message_id: str | None, ) -> list[_NormalizedStreamChunk]: - tool_payload = _extract_tool_part_payload(part) if tool_payload is None: return [] content_key = json.dumps( @@ -409,108 +416,102 @@ def _tool_chunks( stream_kwargs["directory"] = directory if workspace_id is not None: stream_kwargs["workspace_id"] = workspace_id - async for event in self._client.stream_events(**stream_kwargs): + async for update in _client_stream_updates( + self._client, + **stream_kwargs, + ): if stop_event.is_set(): break - _log_stream_event_debug( - event, - limit=max(0, self._client.settings.a2a_log_body_limit), - ) - event_type = event.get("type") - if not isinstance(event_type, str): - continue - props = event.get("properties") - if not isinstance(props, Mapping): - continue - event_session_id = _extract_event_session_id(event) - if event_session_id == session_id: - part = props.get("part") - if isinstance(part, Mapping): - progress = _extract_progress_metadata(part, props) - if progress is not None: - progress_identity = _build_progress_identity(part, props) - progress_key = json.dumps( - progress, - ensure_ascii=False, - sort_keys=True, - separators=(",", ":"), + if update.session_id == session_id: + if update.progress is not None: + progress = update.progress + progress_identity = update.progress_identity or "unknown" + progress_key = json.dumps( + progress, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + if ( + stream_state.register_progress( + identity=progress_identity, + content_key=progress_key, ) - if ( - stream_state.register_progress( - identity=progress_identity, - content_key=progress_key, - ) - and emit_streaming_metadata - ): - sequence = stream_state.next_sequence() - await event_queue.enqueue_event( - TaskStatusUpdateEvent( - task_id=task_id, - context_id=context_id, - status=TaskStatus( - state=TaskState.TASK_STATE_WORKING - ), - metadata=_build_output_metadata( - session_id=session_id, - stream={ - "message_id": ( - stream_state.resolve_message_id( - _extract_stream_message_id( - part, props - ) - ) - ), - "event_id": stream_state.build_event_id( - sequence - ), - "sequence": sequence, - }, - progress=dict(progress), - include_session_metadata=( - emit_session_metadata + and emit_streaming_metadata + ): + sequence = stream_state.next_sequence() + await event_queue.enqueue_event( + TaskStatusUpdateEvent( + task_id=task_id, + context_id=context_id, + status=TaskStatus(state=TaskState.TASK_STATE_WORKING), + metadata=_build_output_metadata( + session_id=session_id, + stream={ + "message_id": ( + stream_state.resolve_message_id( + update.message_id + ) ), - include_streaming_metadata=( - emit_streaming_metadata + "event_id": stream_state.build_event_id( + sequence ), - include_interrupt_metadata=( - emit_interrupt_metadata - ), - ), - ) + "sequence": sequence, + }, + progress=dict(progress), + include_session_metadata=emit_session_metadata, + include_streaming_metadata=emit_streaming_metadata, + include_interrupt_metadata=emit_interrupt_metadata, + ), ) - upstream_error = _extract_upstream_error_from_event(event) - if upstream_error is not None and stream_state.upstream_error is None: - stream_state.upstream_error = upstream_error - signal = _extract_stream_terminal_signal(event) - if signal is not None and not terminal_signal.done(): - terminal_signal.set_result(signal) + ) + if ( + update.error_type is not None + and update.error_state is not None + and update.error_message is not None + and stream_state.upstream_error is None + ): + stream_state.upstream_error = _UpstreamInBandError( + error_type=update.error_type, + state=_task_state_from_execution_state(update.error_state), + message=update.error_message, + upstream_status=update.upstream_status, + ) + if update.terminal is not None and not terminal_signal.done(): + outcome = update.terminal + terminal_signal.set_result( + _StreamTerminalSignal( + state=_task_state_from_execution_state(outcome.state), + error_type=outcome.error_type, + message=outcome.error_message, + upstream_status=outcome.upstream_status, + ) + ) stop_event.set() - usage = _extract_token_usage(event) - if usage is not None: - stream_state.ingest_token_usage(usage) - asked = _extract_interrupt_asked_event(event) - if asked is not None: - request_id = asked["request_id"] + if update.usage is not None: + stream_state.ingest_token_usage(update.usage) + if update.intervention is not None: + intervention = update.intervention + request_id = intervention.request_id if stream_state.mark_interrupt_pending(request_id): await interrupt_requests.remember_request( request_id=request_id, session_id=session_id, - interrupt_type=asked["interrupt_type"], + interrupt_type=intervention.type, identity=identity, task_id=task_id, context_id=context_id, - details=asked["details"], + details=dict(intervention.details), ) await _emit_interrupt_status( state=TaskState.TASK_STATE_INPUT_REQUIRED, request_id=request_id, - interrupt_type=asked["interrupt_type"], + interrupt_type=intervention.type, phase="asked", - details=asked["details"], + details=intervention.details, ) - resolved = _extract_interrupt_resolved_event(event) - if resolved is not None: - resolved_request_id = resolved["request_id"] + if update.resolved_request_id is not None: + resolved_request_id = update.resolved_request_id cleared_pending = stream_state.clear_interrupt_pending( resolved_request_id ) @@ -519,25 +520,24 @@ def _tool_chunks( await _emit_interrupt_status( state=TaskState.TASK_STATE_WORKING, request_id=resolved_request_id, - interrupt_type=resolved["interrupt_type"], + interrupt_type=( + update.resolved_intervention_type or "question" + ), phase="resolved", - resolution=resolved["resolution"], + resolution=update.resolution, ) - if event_type not in {"message.part.updated", "message.part.delta"}: + if update.kind not in {UpdateKind.PART_UPDATED, UpdateKind.PART_DELTA}: continue - part = props.get("part") - if not isinstance(part, Mapping): - part = {} - if _extract_stream_session_id(part, props) != session_id: + if update.session_id != session_id: continue - message_id = _extract_stream_message_id(part, props) - part_id = _extract_stream_part_id(part, props) + message_id = update.message_id + part_id = update.part_id if not part_id: continue - if event_type == "message.part.delta": - field = props.get("field") - delta = props.get("delta") + if update.kind == UpdateKind.PART_DELTA: + field = update.delta_field + delta = update.delta if field != "text" or not isinstance(delta, str) or not delta: continue state = part_states.get(part_id) @@ -562,12 +562,10 @@ def _tool_chunks( await _emit_chunks(delta_chunks) continue - role = _normalize_role(part.get("role") or props.get("role")) state = _upsert_part_state( part_id=part_id, - part=part, - props=props, - role=role, + part_type=update.part_type, + role=update.role, message_id=message_id, ) if state is None: @@ -591,7 +589,7 @@ def _tool_chunks( ) ) - delta = props.get("delta") + delta = update.delta if isinstance(delta, str) and delta: chunks.extend( _delta_chunks( @@ -605,12 +603,12 @@ def _tool_chunks( chunks.extend( _tool_chunks( state=state, - part=part, + tool_payload=update.tool_payload, message_id=message_id, ) ) else: - snapshot_text = _extract_stream_snapshot_text(part) + snapshot_text = update.snapshot if snapshot_text is not None: chunks.extend( _snapshot_chunks( diff --git a/src/opencode_a2a/opencode_upstream_client.py b/src/opencode_a2a/opencode_upstream_client.py index 74d22cbe..3007901a 100644 --- a/src/opencode_a2a/opencode_upstream_client.py +++ b/src/opencode_a2a/opencode_upstream_client.py @@ -17,6 +17,14 @@ from .runtime_state import InterruptRequestBinding from .server.state_store import InterruptRequestRepository, MemoryInterruptRequestRepository from .trace_context import current_trace_headers +from .upstream.contracts import ( + ExecutionOutcome, + ExecutionState, + ExecutionUpdate, + PromptReceipt, + PromptSubmission, + SessionRef, +) _UNSET = object() logger = logging.getLogger(__name__) @@ -451,6 +459,24 @@ async def stream_events( data_lines.append(line[5:].lstrip()) continue + async def stream_updates( + self, + stop_event: asyncio.Event | None = None, + *, + directory: str | None = None, + workspace_id: str | None = None, + ) -> AsyncIterator[ExecutionUpdate]: + async for event in self.stream_events( + stop_event, + directory=directory, + workspace_id=workspace_id, + ): + from .execution.stream_events import normalize_v1_event + + update = normalize_v1_event(event, log_limit=self._settings.a2a_log_body_limit) + if update is not None: + yield update + async def create_session( self, title: str | None = None, @@ -879,6 +905,60 @@ async def send_message( raw=data, ) + async def submit_prompt( + self, + session_id: str, + text: str | None = None, + *, + parts: Sequence[Mapping[str, Any]] | None = None, + directory: str | None = None, + workspace_id: str | None = None, + model_override: Mapping[str, Any] | None = None, + timeout_override: float | None | object = _UNSET, + ) -> PromptSubmission: + """Return V1's synchronous prompt result as distinct receipt and outcome values.""" + message = await self.send_message( + session_id, + text, + parts=parts, + directory=directory, + workspace_id=workspace_id, + model_override=model_override, + timeout_override=timeout_override, + ) + from a2a.types import TaskState + + from .execution.stream_events import ( + _extract_token_usage, + _extract_upstream_error_from_response, + ) + + error = _extract_upstream_error_from_response(message.raw) + outcome_state = ExecutionState.COMPLETED + if error is not None: + outcome_state = ( + ExecutionState.AUTH_REQUIRED + if error.state == TaskState.TASK_STATE_AUTH_REQUIRED + else ExecutionState.CANCELLED + if error.state == TaskState.TASK_STATE_CANCELED + else ExecutionState.FAILED + ) + return PromptSubmission( + receipt=PromptReceipt(session=SessionRef(session_id)), + outcome=ExecutionOutcome( + session=SessionRef(session_id), + state=outcome_state, + output=message.text, + message_id=message.message_id, + usage=_extract_token_usage(message.raw), + 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=outcome_state if error is not None else None, + upstream_status=error.upstream_status if error is not None else None, + response=message.raw, + ), + ) + async def permission_reply( self, request_id: str, diff --git a/src/opencode_a2a/upstream/__init__.py b/src/opencode_a2a/upstream/__init__.py new file mode 100644 index 00000000..8c7273ec --- /dev/null +++ b/src/opencode_a2a/upstream/__init__.py @@ -0,0 +1,25 @@ +"""Typed internal contracts for upstream protocol adapters.""" + +from .contracts import ( + ExecutionOutcome, + ExecutionState, + ExecutionUpdate, + Intervention, + PromptReceipt, + PromptSubmission, + SessionRef, + UpdateKind, + UpstreamAdapter, +) + +__all__ = [ + "ExecutionOutcome", + "ExecutionState", + "ExecutionUpdate", + "Intervention", + "PromptReceipt", + "PromptSubmission", + "SessionRef", + "UpdateKind", + "UpstreamAdapter", +] diff --git a/src/opencode_a2a/upstream/contracts.py b/src/opencode_a2a/upstream/contracts.py new file mode 100644 index 00000000..1579654c --- /dev/null +++ b/src/opencode_a2a/upstream/contracts.py @@ -0,0 +1,118 @@ +"""Protocol-neutral data exchanged with an upstream execution adapter.""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator, Mapping, Sequence +from dataclasses import dataclass +from enum import StrEnum +from typing import Any, Protocol + + +class ExecutionState(StrEnum): + COMPLETED = "completed" + FAILED = "failed" + CANCELLED = "cancelled" + AUTH_REQUIRED = "auth_required" + + +class UpdateKind(StrEnum): + OTHER = "other" + PART_UPDATED = "part_updated" + PART_DELTA = "part_delta" + + +@dataclass(frozen=True) +class SessionRef: + id: str + + +@dataclass(frozen=True) +class PromptReceipt: + session: SessionRef + message_id: str | None = None + + +@dataclass(frozen=True) +class ExecutionOutcome: + session: SessionRef + state: ExecutionState + output: str + message_id: str | None = None + usage: Mapping[str, Any] | None = None + error_type: str | None = None + error_message: str | None = None + error_state: ExecutionState | None = None + upstream_status: int | None = None + response: Mapping[str, Any] | None = None + + +@dataclass(frozen=True) +class PromptSubmission: + receipt: PromptReceipt + outcome: ExecutionOutcome + + +@dataclass(frozen=True) +class Intervention: + request_id: str + type: str + details: Mapping[str, Any] + + +@dataclass(frozen=True) +class ExecutionUpdate: + kind: UpdateKind + session_id: str | None = None + message_id: str | None = None + part_id: str | None = None + part_type: str | None = None + role: str | None = None + delta_field: str | None = None + delta: str | None = None + snapshot: str | None = None + tool_payload: Mapping[str, Any] | None = None + progress: Mapping[str, Any] | None = None + progress_identity: str | None = None + usage: Mapping[str, Any] | None = None + error_type: str | None = None + error_message: str | None = None + error_state: ExecutionState | None = None + upstream_status: int | None = None + terminal: ExecutionOutcome | None = None + intervention: Intervention | None = None + resolved_request_id: str | None = None + resolved_intervention_type: str | None = None + resolution: str | None = None + + +class UpstreamAdapter(Protocol): + """Small execution surface shared by upstream protocol implementations.""" + + async def create_session( + self, + title: str | None = None, + *, + directory: str | None = None, + workspace_id: str | None = None, + ) -> str: ... + + async def submit_prompt( + self, + session_id: str, + text: str | None = None, + *, + parts: Sequence[Mapping[str, Any]] | None = None, + directory: str | None = None, + workspace_id: str | None = None, + model_override: Mapping[str, Any] | None = None, + timeout_override: float | None | object = ..., + ) -> PromptSubmission: ... + + def stream_updates( + self, + stop_event: asyncio.Event | None = None, + *, + directory: str | None = None, + workspace_id: str | None = None, + ) -> AsyncIterator[ExecutionUpdate]: ... diff --git a/tests/support/streaming_output.py b/tests/support/streaming_output.py index a84df732..efe4f68c 100644 --- a/tests/support/streaming_output.py +++ b/tests/support/streaming_output.py @@ -1,4 +1,5 @@ import asyncio +from collections.abc import AsyncIterator from a2a.types import ( TaskArtifactUpdateEvent, @@ -6,7 +7,16 @@ ) from google.protobuf.json_format import MessageToDict +from opencode_a2a.execution.stream_events import _extract_token_usage, normalize_v1_event from opencode_a2a.opencode_upstream_client import OpencodeMessage +from opencode_a2a.upstream.contracts import ( + ExecutionOutcome, + ExecutionState, + ExecutionUpdate, + PromptReceipt, + PromptSubmission, + SessionRef, +) from tests.support.helpers import ( DummyEventQueue, ) @@ -75,6 +85,38 @@ async def send_message( raw=self._response_raw, ) + async def submit_prompt( + self, + session_id: str, + text: str | None = None, + *, + parts: list[dict] | None = None, + directory: str | None = None, + workspace_id: str | None = None, + model_override: dict[str, str] | None = None, + timeout_override=None, # noqa: ANN001 + ) -> PromptSubmission: + del workspace_id + message = await self.send_message( + session_id, + text, + parts=parts, + directory=directory, + model_override=model_override, + timeout_override=timeout_override, + ) + return PromptSubmission( + receipt=PromptReceipt(session=SessionRef(session_id)), + outcome=ExecutionOutcome( + session=SessionRef(session_id), + state=ExecutionState.COMPLETED, + output=message.text, + message_id=message.message_id, + usage=_extract_token_usage(message.raw), + response=message.raw, + ), + ) + async def stream_events(self, stop_event=None, *, directory: str | None = None): # noqa: ANN001 del directory for index, event in enumerate(self._stream_events_payload): @@ -91,6 +133,25 @@ async def stream_events(self, stop_event=None, *, directory: str | None = None): ): yield {"type": "session.idle", "properties": {"sessionID": "ses-1"}} + async def stream_updates( + self, + stop_event: asyncio.Event | None = None, + *, + directory: str | None = None, + workspace_id: str | None = None, + ) -> AsyncIterator[ExecutionUpdate]: + del workspace_id + async for event in self.stream_events( + stop_event, + directory=directory, + ): + update = normalize_v1_event( + event, + log_limit=self.settings.a2a_log_body_limit, + ) + if update is not None: + yield update + def _event( *, diff --git a/tests/upstream/test_upstream_contracts.py b/tests/upstream/test_upstream_contracts.py new file mode 100644 index 00000000..b88fd570 --- /dev/null +++ b/tests/upstream/test_upstream_contracts.py @@ -0,0 +1,126 @@ +from unittest.mock import AsyncMock + +import pytest + +from opencode_a2a.execution.stream_events import normalize_v1_event +from opencode_a2a.opencode_upstream_client import OpencodeMessage, OpencodeUpstreamClient +from opencode_a2a.upstream.contracts import ( + ExecutionState, + UpdateKind, +) +from tests.support.settings import make_settings + + +def test_v1_event_decoder_returns_correlated_protocol_neutral_part_update() -> None: + update = normalize_v1_event( + { + "type": "message.part.updated", + "properties": { + "sessionID": "ses-1", + "messageID": "msg-1", + "part": { + "id": "prt-1", + "type": "text", + "role": "assistant", + "text": "answer", + "sessionID": "ses-1", + "messageID": "msg-1", + }, + }, + } + ) + + assert update is not None + assert update.kind == UpdateKind.PART_UPDATED + assert update.session_id == "ses-1" + assert update.message_id == "msg-1" + assert update.part_id == "prt-1" + assert update.part_type == "text" + assert update.role == "agent" + assert update.snapshot == "answer" + + +@pytest.mark.asyncio +async def test_submit_prompt_separates_v1_receipt_from_terminal_outcome(monkeypatch) -> None: + client = OpencodeUpstreamClient(make_settings(test_bearer_token="t-1")) + message = OpencodeMessage( + text="completed output", + session_id="ses-1", + message_id="assistant-msg-1", + raw={"info": {"tokens": {"input": 2, "output": 3}}, "parts": []}, + ) + send_message = AsyncMock(return_value=message) + monkeypatch.setattr(client, "send_message", send_message) + + submission = await client.submit_prompt("ses-1", "input") + + assert submission.receipt.session.id == "ses-1" + assert submission.receipt.message_id is None + assert submission.outcome.state == ExecutionState.COMPLETED + assert submission.outcome.message_id == "assistant-msg-1" + assert submission.outcome.output == "completed output" + assert submission.outcome.usage == {"input_tokens": 2, "output_tokens": 3, "total_tokens": 5} + assert submission.outcome.response is not None + assert submission.outcome.response["parts"] == [] + assert send_message.await_args is not None + assert send_message.await_args.args == ("ses-1", "input") + assert send_message.await_args.kwargs["parts"] is None + assert "timeout_override" in send_message.await_args.kwargs + await client.close() + + +@pytest.mark.asyncio +async def test_submit_prompt_preserves_terminal_v1_error_in_outcome(monkeypatch) -> None: + client = OpencodeUpstreamClient(make_settings(test_bearer_token="t-1")) + message = OpencodeMessage( + text="", + session_id="ses-1", + message_id="assistant-msg-1", + raw={ + "info": { + "error": { + "name": "APIError", + "data": {"message": "upstream denied", "statusCode": 400}, + } + } + }, + ) + monkeypatch.setattr(client, "send_message", AsyncMock(return_value=message)) + + submission = await client.submit_prompt("ses-1", "input") + + assert submission.outcome.state == ExecutionState.FAILED + assert submission.outcome.error_state == ExecutionState.FAILED + assert submission.outcome.error_type == "UPSTREAM_BAD_REQUEST" + assert submission.outcome.upstream_status == 400 + assert "upstream denied" in (submission.outcome.error_message or "") + await client.close() + + +@pytest.mark.asyncio +async def test_stream_updates_exposes_normalized_values_not_v1_event_mapping(monkeypatch) -> None: + client = OpencodeUpstreamClient(make_settings(test_bearer_token="t-1")) + + async def stream_events(*_args, **_kwargs): + yield { + "type": "message.part.delta", + "properties": { + "sessionID": "ses-1", + "messageID": "msg-1", + "partID": "prt-1", + "field": "text", + "delta": "hello", + }, + } + + monkeypatch.setattr(client, "stream_events", stream_events) + + updates = [update async for update in client.stream_updates()] + + assert len(updates) == 1 + assert updates[0].kind == UpdateKind.PART_DELTA + assert updates[0].session_id == "ses-1" + assert updates[0].part_id == "prt-1" + assert updates[0].delta == "hello" + assert not hasattr(updates[0], "properties") + await client.close() From a2f8691ecf530715787de43b51f738448c698eac Mon Sep 17 00:00:00 2001 From: Tom Brandenburg Date: Sun, 25 Oct 2026 11:32:05 +0100 Subject: [PATCH 2/3] fix(upstream): allow receipt-only submissions safely (#2) --- src/opencode_a2a/execution/coordinator.py | 21 +++++++--- src/opencode_a2a/upstream/contracts.py | 2 +- .../test_streaming_output_contract_core.py | 39 +++++++++++++++++++ tests/upstream/test_upstream_contracts.py | 2 + 4 files changed, 57 insertions(+), 7 deletions(-) diff --git a/src/opencode_a2a/execution/coordinator.py b/src/opencode_a2a/execution/coordinator.py index 29ede9f2..1c6b4e3c 100644 --- a/src/opencode_a2a/execution/coordinator.py +++ b/src/opencode_a2a/execution/coordinator.py @@ -22,7 +22,11 @@ TaskStatusUpdateEvent, ) -from ..opencode_upstream_client import UpstreamConcurrencyLimitError, UpstreamContractError +from ..opencode_upstream_client import ( + OpencodeMessage, + UpstreamConcurrencyLimitError, + UpstreamContractError, +) from ..upstream.contracts import ( ExecutionOutcome, PromptSubmission, @@ -160,15 +164,20 @@ async def run(self) -> None: if callable(submit_prompt) else None ) - response = ( - submission.outcome - if isinstance(submission, PromptSubmission) - else await self._executor._client.send_message( + response: ExecutionOutcome | OpencodeMessage + if isinstance(submission, PromptSubmission): + if submission.outcome is None: + raise RuntimeError( + "Upstream accepted the prompt without a terminal outcome; " + "asynchronous observation is not supported" + ) + response = submission.outcome + else: + response = await self._executor._client.send_message( self._session_id, prompt_text, **prompt_kwargs, ) - ) if self._pending_preferred_claim: await self._executor._session_manager.finalize_preferred_session_binding( diff --git a/src/opencode_a2a/upstream/contracts.py b/src/opencode_a2a/upstream/contracts.py index 1579654c..8b517951 100644 --- a/src/opencode_a2a/upstream/contracts.py +++ b/src/opencode_a2a/upstream/contracts.py @@ -50,7 +50,7 @@ class ExecutionOutcome: @dataclass(frozen=True) class PromptSubmission: receipt: PromptReceipt - outcome: ExecutionOutcome + outcome: ExecutionOutcome | None @dataclass(frozen=True) diff --git a/tests/execution/test_streaming_output_contract_core.py b/tests/execution/test_streaming_output_contract_core.py index 997c48d3..f15999a0 100644 --- a/tests/execution/test_streaming_output_contract_core.py +++ b/tests/execution/test_streaming_output_contract_core.py @@ -1,5 +1,6 @@ import asyncio from types import SimpleNamespace +from unittest.mock import AsyncMock import pytest from a2a.server.agent_execution import RequestContext @@ -20,6 +21,7 @@ from opencode_a2a.execution.stream_state import _StreamOutputState from opencode_a2a.server.context_helpers import normalize_server_call_context from opencode_a2a.task_states import TERMINAL_TASK_STATES +from opencode_a2a.upstream.contracts import PromptReceipt, PromptSubmission, SessionRef from tests.support.helpers import ( DummyEventQueue, make_request_context, @@ -45,6 +47,43 @@ def _is_terminal_status_event(event: TaskStatusUpdateEvent) -> bool: ) +@pytest.mark.asyncio +async def test_receipt_only_prompt_submission_fails_without_resending() -> None: + client = DummyStreamingClient(stream_events_payload=[], response_text="") + submit_prompt = AsyncMock( + return_value=PromptSubmission( + receipt=PromptReceipt(session=SessionRef("ses-1")), + outcome=None, + ) + ) + send_message = AsyncMock() + client.submit_prompt = submit_prompt + client.send_message = send_message + executor = OpencodeAgentExecutor(client, streaming_enabled=False) + queue = DummyEventQueue() + + await executor.execute( + make_request_context( + task_id="task-receipt-only", + context_id="ctx-receipt-only", + text="hello", + ), + queue, + ) + + failed_tasks = [ + event + for event in queue.events + if isinstance(event, Task) and event.status.state == TaskState.TASK_STATE_FAILED + ] + assert submit_prompt.await_count == 1 + send_message.assert_not_awaited() + assert len(failed_tasks) == 1 + assert ( + "asynchronous observation is not supported" in failed_tasks[0].status.message.parts[0].text + ) + + @pytest.mark.asyncio async def test_streaming_accepts_file_input_without_breaking_contract() -> None: client = DummyStreamingClient( diff --git a/tests/upstream/test_upstream_contracts.py b/tests/upstream/test_upstream_contracts.py index b88fd570..a67852bb 100644 --- a/tests/upstream/test_upstream_contracts.py +++ b/tests/upstream/test_upstream_contracts.py @@ -54,6 +54,7 @@ async def test_submit_prompt_separates_v1_receipt_from_terminal_outcome(monkeypa submission = await client.submit_prompt("ses-1", "input") + assert submission.outcome is not None assert submission.receipt.session.id == "ses-1" assert submission.receipt.message_id is None assert submission.outcome.state == ExecutionState.COMPLETED @@ -89,6 +90,7 @@ async def test_submit_prompt_preserves_terminal_v1_error_in_outcome(monkeypatch) submission = await client.submit_prompt("ses-1", "input") + assert submission.outcome is not None assert submission.outcome.state == ExecutionState.FAILED assert submission.outcome.error_state == ExecutionState.FAILED assert submission.outcome.error_type == "UPSTREAM_BAD_REQUEST" From 65c15e936582c6c69df6bb923078321d29402df5 Mon Sep 17 00:00:00 2001 From: Tom Brandenburg Date: Sun, 25 Oct 2026 13:17:23 +0100 Subject: [PATCH 3/3] fix(execution): enforce upstream outcome safety --- src/opencode_a2a/execution/coordinator.py | 68 +++++++---- tests/execution/test_agent_errors.py | 9 ++ tests/execution/test_cancellation.py | 4 + tests/execution/test_session_ownership.py | 4 + .../test_streaming_output_contract_core.py | 110 +++++++++++++++++- 5 files changed, 173 insertions(+), 22 deletions(-) diff --git a/src/opencode_a2a/execution/coordinator.py b/src/opencode_a2a/execution/coordinator.py index 1c6b4e3c..e0622591 100644 --- a/src/opencode_a2a/execution/coordinator.py +++ b/src/opencode_a2a/execution/coordinator.py @@ -29,6 +29,7 @@ ) from ..upstream.contracts import ( ExecutionOutcome, + ExecutionState, PromptSubmission, UpstreamAdapter, ) @@ -56,6 +57,7 @@ from .executor import OpencodeAgentExecutor logger = logging.getLogger(__name__) +_MISSING = object() @dataclass(frozen=True) @@ -158,26 +160,34 @@ async def run(self) -> None: prompt_kwargs["parts"] = turn_request_parts upstream: UpstreamAdapter = self._executor._client - submit_prompt = getattr(upstream, "submit_prompt", None) - submission = ( - await submit_prompt(self._session_id, prompt_text, **prompt_kwargs) - if callable(submit_prompt) - else None - ) + submit_prompt = getattr(upstream, "submit_prompt", _MISSING) response: ExecutionOutcome | OpencodeMessage - if isinstance(submission, PromptSubmission): + if submit_prompt is _MISSING: + response = await self._executor._client.send_message( + self._session_id, + prompt_text, + **prompt_kwargs, + ) + else: + if not callable(submit_prompt): + raise UpstreamContractError( + "OpenCode submit_prompt must be callable when provided" + ) + submission = await submit_prompt( + self._session_id, + 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 - else: - response = await self._executor._client.send_message( - self._session_id, - prompt_text, - **prompt_kwargs, - ) if self._pending_preferred_claim: await self._executor._session_manager.finalize_preferred_session_binding( @@ -187,6 +197,13 @@ 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) @@ -348,19 +365,28 @@ async def _handle_response(self, response: Any) -> None: if isinstance(response, ExecutionOutcome) else response.raw ) - response_error = ( - _UpstreamInBandError( - error_type=response.error_type, - state=_task_state_from_execution_state(response.error_state), - message=response.error_message, + 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, ) - if isinstance(response, ExecutionOutcome) + 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 - else _extract_upstream_error_from_response(response_payload) - ) + ): + 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( response.usage if isinstance(response, ExecutionOutcome) diff --git a/tests/execution/test_agent_errors.py b/tests/execution/test_agent_errors.py index 8d116d28..877491b2 100644 --- a/tests/execution/test_agent_errors.py +++ b/tests/execution/test_agent_errors.py @@ -160,6 +160,7 @@ async def test_execute_http_error_maps_to_task_error_type_and_state( exc = httpx.HTTPStatusError("upstream error", request=request, response=response) client = AsyncMock() + del client.submit_prompt async def create_session(title: str | None = None, *, directory: str | None = None) -> str: del title, directory @@ -204,6 +205,7 @@ async def test_streaming_execute_http_error_emits_status_update_with_metadata(ca exc = httpx.HTTPStatusError("upstream error", request=request, response=response) client = AsyncMock() + del client.submit_prompt async def create_session(title: str | None = None, *, directory: str | None = None) -> str: del title, directory @@ -260,6 +262,7 @@ async def test_streaming_execute_upstream_backpressure_emits_status_update_with_ caplog, ) -> None: client = AsyncMock() + del client.submit_prompt async def create_session(title: str | None = None, *, directory: str | None = None) -> str: del title, directory @@ -318,6 +321,7 @@ async def create_session(title: str | None = None, *, directory: str | None = No @pytest.mark.asyncio async def test_execute_upstream_payload_error_maps_to_task_error_type() -> None: client = AsyncMock() + del client.submit_prompt async def create_session(title: str | None = None, *, directory: str | None = None) -> str: del title, directory @@ -360,6 +364,7 @@ async def create_session(title: str | None = None, *, directory: str | None = No @pytest.mark.asyncio async def test_execute_upstream_backpressure_maps_to_task_error_type() -> None: client = AsyncMock() + del client.submit_prompt async def create_session(title: str | None = None, *, directory: str | None = None) -> str: del title, directory @@ -403,6 +408,7 @@ async def create_session(title: str | None = None, *, directory: str | None = No @pytest.mark.asyncio async def test_execute_response_info_error_maps_to_task_failed_state() -> None: client = AsyncMock() + del client.submit_prompt async def create_session(title: str | None = None, *, directory: str | None = None) -> str: del title, directory @@ -463,6 +469,7 @@ async def test_streaming_execute_response_info_error_emits_failed_status_and_err None ): client = AsyncMock() + del client.submit_prompt async def create_session(title: str | None = None, *, directory: str | None = None) -> str: del title, directory @@ -532,6 +539,7 @@ async def stream_events(stop_event=None, *, directory: str | None = None): # no @pytest.mark.asyncio async def test_streaming_execute_session_error_emits_failed_status() -> None: client = AsyncMock() + del client.submit_prompt async def create_session(title: str | None = None, *, directory: str | None = None) -> str: del title, directory @@ -606,6 +614,7 @@ async def stream_events(stop_event=None, *, directory: str | None = None): # no @pytest.mark.asyncio async def test_streaming_execute_message_updated_info_error_emits_failed_status() -> None: client = AsyncMock() + del client.submit_prompt async def create_session(title: str | None = None, *, directory: str | None = None) -> str: del title, directory diff --git a/tests/execution/test_cancellation.py b/tests/execution/test_cancellation.py index 9105f99d..00236e91 100644 --- a/tests/execution/test_cancellation.py +++ b/tests/execution/test_cancellation.py @@ -22,6 +22,7 @@ def _memory_session_repository(executor: OpencodeAgentExecutor) -> MemorySession @pytest.mark.asyncio async def test_cancel_interrupts_running_execute_and_keeps_queue_open(caplog): client = AsyncMock(spec=OpencodeUpstreamClient) + del client.submit_prompt send_started = asyncio.Event() send_cancelled = asyncio.Event() @@ -121,6 +122,7 @@ async def test_cancel_does_not_block_with_real_event_queue() -> None: @pytest.mark.asyncio async def test_cancel_keeps_canceled_status_when_abort_session_fails() -> None: client = AsyncMock(spec=OpencodeUpstreamClient) + del client.submit_prompt send_started = asyncio.Event() send_cancelled = asyncio.Event() @@ -190,6 +192,7 @@ async def send_message( @pytest.mark.asyncio async def test_cancel_remains_responsive_when_abort_session_hangs(caplog) -> None: client = AsyncMock(spec=OpencodeUpstreamClient) + del client.submit_prompt send_started = asyncio.Event() send_cancelled = asyncio.Event() @@ -268,6 +271,7 @@ async def slow_abort_session( @pytest.mark.asyncio async def test_cancel_waiting_for_session_lock_does_not_abort_other_generation() -> None: client = AsyncMock(spec=OpencodeUpstreamClient) + del client.submit_prompt client.create_session.return_value = "session-4" client.send_message = AsyncMock() client.abort_session.return_value = True diff --git a/tests/execution/test_session_ownership.py b/tests/execution/test_session_ownership.py index 236fa246..0dc8380d 100644 --- a/tests/execution/test_session_ownership.py +++ b/tests/execution/test_session_ownership.py @@ -26,6 +26,7 @@ @pytest.fixture def mock_client(): client = AsyncMock(spec=OpencodeUpstreamClient) + del client.submit_prompt # Define sessions to return sessions = ["session-1", "session-2", "session-3"] current_idx = 0 @@ -156,6 +157,7 @@ async def test_session_hijack_prevention(mock_client): @pytest.mark.asyncio async def test_concurrent_session_create_isolated_by_identity(): client = AsyncMock(spec=OpencodeUpstreamClient) + del client.submit_prompt created = 0 async def create_session(title=None, directory=None): @@ -284,6 +286,7 @@ def _now() -> float: @pytest.mark.asyncio async def test_preferred_session_claim_is_released_on_upstream_failure(): client = AsyncMock(spec=OpencodeUpstreamClient) + del client.submit_prompt async def send_message( session_id, @@ -323,6 +326,7 @@ async def send_message( @pytest.mark.asyncio async def test_preferred_session_claim_is_released_on_upstream_cancellation(): client = AsyncMock(spec=OpencodeUpstreamClient) + del client.submit_prompt async def send_message( _session_id, diff --git a/tests/execution/test_streaming_output_contract_core.py b/tests/execution/test_streaming_output_contract_core.py index f15999a0..910136bb 100644 --- a/tests/execution/test_streaming_output_contract_core.py +++ b/tests/execution/test_streaming_output_contract_core.py @@ -21,7 +21,13 @@ from opencode_a2a.execution.stream_state import _StreamOutputState from opencode_a2a.server.context_helpers import normalize_server_call_context from opencode_a2a.task_states import TERMINAL_TASK_STATES -from opencode_a2a.upstream.contracts import PromptReceipt, PromptSubmission, SessionRef +from opencode_a2a.upstream.contracts import ( + ExecutionOutcome, + ExecutionState, + PromptReceipt, + PromptSubmission, + SessionRef, +) from tests.support.helpers import ( DummyEventQueue, make_request_context, @@ -84,6 +90,108 @@ async def test_receipt_only_prompt_submission_fails_without_resending() -> None: ) +@pytest.mark.parametrize( + ("execution_state", "task_state"), + [ + (ExecutionState.FAILED, TaskState.TASK_STATE_FAILED), + (ExecutionState.CANCELLED, TaskState.TASK_STATE_CANCELED), + (ExecutionState.AUTH_REQUIRED, TaskState.TASK_STATE_AUTH_REQUIRED), + ], +) +@pytest.mark.asyncio +async def test_non_completed_outcome_without_error_metadata_is_terminal_before_tools( + monkeypatch, + execution_state: ExecutionState, + task_state: TaskState, +) -> None: + from opencode_a2a.execution import coordinator + + tool_calls: list[dict] = [] + + async def record_tool_call(raw_response: dict, *, a2a_client_manager) -> None: + del a2a_client_manager + tool_calls.append(raw_response) + + monkeypatch.setattr(coordinator, "maybe_handle_tools", record_tool_call) + client = DummyStreamingClient(stream_events_payload=[], response_text="unexpected success") + client.submit_prompt = AsyncMock( + return_value=PromptSubmission( + receipt=PromptReceipt(session=SessionRef("ses-1")), + outcome=ExecutionOutcome( + session=SessionRef("ses-1"), + state=execution_state, + output="unexpected success", + response={ + "parts": [ + { + "type": "tool", + "tool": "a2a_call", + "callID": "call-1", + "state": {"status": "calling"}, + } + ] + }, + ), + ) + ) + executor = OpencodeAgentExecutor(client, streaming_enabled=False) + queue = DummyEventQueue() + + await executor.execute( + make_request_context( + task_id=f"task-outcome-{execution_state.value}", + context_id="ctx-outcome", + text="hello", + ), + queue, + ) + + failed_tasks = [ + event + for event in queue.events + if isinstance(event, Task) and event.status.state == task_state + ] + assert len(failed_tasks) == 1 + assert failed_tasks[0].status.message.parts[0].text == "OpenCode execution did not complete." + assert not any( + isinstance(event, Task) and event.status.state == TaskState.TASK_STATE_COMPLETED + for event in queue.events + ) + assert tool_calls == [] + + +@pytest.mark.asyncio +async def test_invalid_prompt_submission_fails_without_legacy_resend() -> None: + client = DummyStreamingClient(stream_events_payload=[], response_text="unused") + submit_prompt = AsyncMock(return_value=object()) + send_message = AsyncMock() + client.submit_prompt = submit_prompt + client.send_message = send_message + executor = OpencodeAgentExecutor(client, streaming_enabled=False) + queue = DummyEventQueue() + + await executor.execute( + make_request_context( + task_id="task-invalid-submission", + context_id="ctx-invalid-submission", + text="hello", + ), + queue, + ) + + failed_tasks = [ + event + for event in queue.events + if isinstance(event, Task) and event.status.state == TaskState.TASK_STATE_FAILED + ] + assert submit_prompt.await_count == 1 + send_message.assert_not_awaited() + assert len(failed_tasks) == 1 + assert "submit_prompt must return PromptSubmission" in ( + failed_tasks[0].status.message.parts[0].text + ) + + @pytest.mark.asyncio async def test_streaming_accepts_file_input_without_breaking_contract() -> None: client = DummyStreamingClient(