From 81887077acf838fc8b38421ab4517985ccf70723 Mon Sep 17 00:00:00 2001 From: Tom Brandenburg Date: Sun, 25 Oct 2026 11:26:04 +0100 Subject: [PATCH 1/9] 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/9] 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 767e70691d8a09624d25ae4c7ba44a477d3d9ab2 Mon Sep 17 00:00:00 2001 From: Tom Brandenburg Date: Sun, 25 Oct 2026 11:49:23 +0100 Subject: [PATCH 3/9] feat(upstream): add opt-in OpenCode V2 session transport (#3) --- docs/compatibility.md | 8 + docs/guide.md | 1 + src/opencode_a2a/config.py | 2 + src/opencode_a2a/execution/executor.py | 15 ++ src/opencode_a2a/jsonrpc/application.py | 8 +- src/opencode_a2a/server/agent_card.py | 29 ++- src/opencode_a2a/server/application.py | 14 +- src/opencode_a2a/server/middleware.py | 5 +- src/opencode_a2a/server/openapi.py | 7 +- src/opencode_a2a/upstream/v2.py | 230 ++++++++++++++++++ tests/config/test_settings.py | 27 ++ tests/execution/test_v2_prompt_failfast.py | 32 +++ .../test_jsonrpc_unsupported_method.py | 34 +++ tests/server/test_agent_card.py | 18 ++ tests/upstream/test_v2_upstream_adapter.py | 212 ++++++++++++++++ 15 files changed, 634 insertions(+), 8 deletions(-) create mode 100644 src/opencode_a2a/upstream/v2.py create mode 100644 tests/execution/test_v2_prompt_failfast.py create mode 100644 tests/upstream/test_v2_upstream_adapter.py diff --git a/docs/compatibility.md b/docs/compatibility.md index 39117a43..17e9bed1 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -9,6 +9,14 @@ This document defines the compatibility promises `opencode-a2a` currently uphold - Supported A2A protocol line: `1.0` - OpenCode runtime line: `1.18.x` (verified with `1.18.19`) +OpenCode V2 is an opt-in, experimental session-transport slice selected with +`OPENCODE_API_VERSION=v2`. It implements `/api/info`, `/api/location`, and +session create/get/list transport only; A2A prompt execution and all +OpenCode-specific JSON-RPC extensions remain unavailable. V1 stays the +default, and the adapter does not auto-detect or fall back between versions. +The V2 session transport was exercised against OpenCode `2.0.28`; that +experimental check does not add V2 to the supported runtime line above. + This is the single canonical place where the supported upstream OpenCode version line is declared. Keep it updated together with the live integration smoke check (`./scripts/live_opencode_smoke.sh`) and do not duplicate the version line elsewhere. The SDK-owned core JSON-RPC method set follows the supported SDK line and is locked by repository tests so SDK upgrades trigger an explicit compatibility review. diff --git a/docs/guide.md b/docs/guide.md index fb4bc0ce..7561bb27 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -29,6 +29,7 @@ Key variables to understand protocol behavior: - `A2A_STATIC_AUTH_CREDENTIALS`: required static credential registry in JSON array form. Supports multiple `bearer` / `basic` credentials, bearer-only required `principal`, optional `credential_id`, optional `capabilities`, and optional `enabled=false` for explicit disablement. - `OPENCODE_BASE_URL`: upstream OpenCode HTTP endpoint. Default: `http://127.0.0.1:4096`. In two-process deployments, set it explicitly. +- `OPENCODE_API_VERSION`: upstream API selector, `v1` (default) or `v2`. Selection is explicit; the service never detects or falls back to another API version. V2 is experimental session transport only: prompt execution and OpenCode-specific extensions are rejected/hidden until later support lands. - `OPENCODE_WORKSPACE_ROOT`: service-level default workspace root exposed to OpenCode when clients do not request a narrower directory override. - `A2A_ALLOW_DIRECTORY_OVERRIDE`: controls whether clients may pass `metadata.opencode.directory`. - `A2A_ENABLE_SESSION_SHELL`: gates high-risk JSON-RPC method `opencode.sessions.shell`. diff --git a/src/opencode_a2a/config.py b/src/opencode_a2a/config.py index 84f1775f..ea746cb4 100644 --- a/src/opencode_a2a/config.py +++ b/src/opencode_a2a/config.py @@ -37,6 +37,7 @@ OutsideWorkspaceAccess = Literal["unknown", "allowed", "disallowed", "custom"] TaskStoreBackend = Literal["memory", "database"] StaticAuthScheme = Literal["bearer", "basic"] +OpencodeApiVersion = Literal["v1", "v2"] def _parse_declared_list(value: Any) -> tuple[str, ...]: @@ -148,6 +149,7 @@ class Settings(BaseSettings): ) # OpenCode settings + opencode_api_version: OpencodeApiVersion = Field(default="v1", alias="OPENCODE_API_VERSION") opencode_base_url: str = Field(default="http://127.0.0.1:4096", alias="OPENCODE_BASE_URL") opencode_workspace_root: str | None = Field(default=None, alias="OPENCODE_WORKSPACE_ROOT") opencode_agent: str | None = Field(default=None, alias="OPENCODE_AGENT") diff --git a/src/opencode_a2a/execution/executor.py b/src/opencode_a2a/execution/executor.py index 19fffcd2..1826ce72 100644 --- a/src/opencode_a2a/execution/executor.py +++ b/src/opencode_a2a/execution/executor.py @@ -109,6 +109,21 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non ) return + if not getattr(self._client, "prompt_execution_supported", True): + await self._emit_error( + event_queue, + task_id=task_id, + context_id=context_id, + message=( + "OpenCode V2 prompt execution is not implemented; session transport is " + "experimental and prompt execution will be available in a later release." + ), + state=TaskState.TASK_STATE_FAILED, + error_type="UPSTREAM_UNSUPPORTED", + streaming_request=self._should_stream(context), + ) + return + call_context = context.call_context identity = (call_context.state.get("identity") if call_context else None) or "anonymous" credential_id = call_context.state.get("credential_id") if call_context else None diff --git a/src/opencode_a2a/jsonrpc/application.py b/src/opencode_a2a/jsonrpc/application.py index 21087464..701188f2 100644 --- a/src/opencode_a2a/jsonrpc/application.py +++ b/src/opencode_a2a/jsonrpc/application.py @@ -64,6 +64,7 @@ def __init__( upstream_client: OpencodeUpstreamClient, methods: dict[str, str], supported_methods: list[str], + extensions_enabled: bool = True, directory_resolver: Callable[[str | None], str | None] | None = None, session_claim: Callable[..., Awaitable[bool]] | None = None, session_claim_finalize: Callable[..., Awaitable[None]] | None = None, @@ -76,6 +77,7 @@ def __init__( super().__init__(request_handler=http_handler, **kwargs) self._http_handler = http_handler self._upstream_client = upstream_client + self._extensions_enabled = extensions_enabled self._method_session_status = methods["status"] self._method_list_sessions = methods["list_sessions"] self._method_get_session = methods["get_session"] @@ -421,7 +423,11 @@ async def handle_requests(self, request: Request) -> Response: except Exception: return await super().handle_requests(request) - extension_spec = self._extension_method_registry.resolve(base_request.method) + extension_spec = ( + self._extension_method_registry.resolve(base_request.method) + if self._extensions_enabled + else None + ) if extension_spec is None: return await self._handle_core_request( request, diff --git a/src/opencode_a2a/server/agent_card.py b/src/opencode_a2a/server/agent_card.py index e7032cc4..daf02cce 100644 --- a/src/opencode_a2a/server/agent_card.py +++ b/src/opencode_a2a/server/agent_card.py @@ -66,6 +66,11 @@ def _build_agent_card_description( include_detailed_contracts: bool, ) -> str: base = (settings.a2a_description or "").strip() or "OpenCode A2A runtime." + if settings.opencode_api_version == "v2": + return ( + f"{base} OpenCode V2 session transport is experimental; A2A prompt execution and " + "OpenCode-specific extensions are not implemented." + ) if not include_detailed_contracts: public_parts: list[str] = [ base, @@ -356,6 +361,20 @@ def _build_agent_skills( capability_snapshot: JsonRpcCapabilitySnapshot, include_detailed_contracts: bool, ) -> list[AgentSkill]: + if settings.opencode_api_version == "v2": + return [ + AgentSkill( + id="v2-session-transport", + name="OpenCode V2 Session Transport", + description=( + "Experimental upstream session transport for adapter integrations; " + "no A2A session or prompt operation is currently available." + ), + input_modes=list(_JSON_RPC_MODES), + output_modes=list(_JSON_RPC_MODES), + tags=["experimental", "transport"], + ) + ] if not include_detailed_contracts: return [ AgentSkill( @@ -533,9 +552,13 @@ def build_agent_card( capabilities=AgentCapabilities( streaming=True, extended_agent_card=True, - extensions=_build_agent_extensions( - runtime_profile=runtime_profile, - include_detailed_contracts=include_detailed_contracts, + extensions=( + [] + if settings.opencode_api_version == "v2" + else _build_agent_extensions( + runtime_profile=runtime_profile, + include_detailed_contracts=include_detailed_contracts, + ) ), ), skills=_build_agent_skills( diff --git a/src/opencode_a2a/server/application.py b/src/opencode_a2a/server/application.py index 431a8850..9b5fc4e2 100644 --- a/src/opencode_a2a/server/application.py +++ b/src/opencode_a2a/server/application.py @@ -20,6 +20,7 @@ from ..a2a_protocol import ( AGENT_CARD_WELL_KNOWN_PATH, + CORE_JSONRPC_METHODS, EXTENDED_AGENT_CARD_PATH, ) from ..config import Settings @@ -38,6 +39,7 @@ from ..profile.runtime import build_runtime_profile from ..protocol_versions import A2A_PROTOCOL_VERSION from ..trace_context import install_log_record_factory +from ..upstream.v2 import V2UpstreamAdapter from .agent_card import ( build_agent_card, ) @@ -172,7 +174,10 @@ def create_app(settings: Settings) -> FastAPI: settings, engine=database_engine, ) - upstream_client = OpencodeUpstreamClient( + upstream_client_type = ( + V2UpstreamAdapter if settings.opencode_api_version == "v2" else OpencodeUpstreamClient + ) + upstream_client = upstream_client_type( settings, interrupt_request_repository=runtime_state_runtime.interrupt_request_repository, ) @@ -218,7 +223,12 @@ def create_app(settings: Settings) -> FastAPI: http_handler=handler, context_builder=context_builder, upstream_client=upstream_client, - supported_methods=capability_snapshot.supported_jsonrpc_methods(), + supported_methods=( + list(CORE_JSONRPC_METHODS) + if settings.opencode_api_version == "v2" + else capability_snapshot.supported_jsonrpc_methods() + ), + extensions_enabled=settings.opencode_api_version == "v1", directory_resolver=( partial( executor._sandbox_policy.resolve_directory, diff --git a/src/opencode_a2a/server/middleware.py b/src/opencode_a2a/server/middleware.py index 176693cf..fe61246b 100644 --- a/src/opencode_a2a/server/middleware.py +++ b/src/opencode_a2a/server/middleware.py @@ -494,7 +494,10 @@ def _resolve_activated_extensions(request: Request) -> tuple[str, ...]: return () path = request.url.path - if path == AGENT_CARD_WELL_KNOWN_PATH: + supported_extensions: tuple[str, ...] + if settings.opencode_api_version == "v2": + supported_extensions = () + elif path == AGENT_CARD_WELL_KNOWN_PATH: supported_extensions = PUBLIC_EXTENSION_URIS elif path == EXTENDED_AGENT_CARD_PATH: supported_extensions = ALL_EXTENSION_URIS diff --git a/src/opencode_a2a/server/openapi.py b/src/opencode_a2a/server/openapi.py index 4e305603..daf5b3e0 100644 --- a/src/opencode_a2a/server/openapi.py +++ b/src/opencode_a2a/server/openapi.py @@ -244,7 +244,6 @@ def _patch_jsonrpc_openapi_contract( *, runtime_profile: RuntimeProfile, ) -> None: - del settings session_binding = build_session_binding_extension_params( runtime_profile=runtime_profile, ) @@ -316,6 +315,12 @@ def custom_openapi() -> dict[str, Any]: ), }, } + if settings.opencode_api_version == "v2": + post["description"] = ( + "OpenCode V2 session transport is experimental. A2A prompt " + "execution and extension methods are not implemented." + ) + post.pop("x-a2a-extension-contracts", None) request_body = post.setdefault("requestBody", {}) if isinstance(request_body, dict): diff --git a/src/opencode_a2a/upstream/v2.py b/src/opencode_a2a/upstream/v2.py new file mode 100644 index 00000000..85448744 --- /dev/null +++ b/src/opencode_a2a/upstream/v2.py @@ -0,0 +1,230 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Mapping, Sequence +from typing import Any, cast +from urllib.parse import quote + +import httpx + +from ..config import Settings +from ..opencode_upstream_client import ( + OpencodeMessage, + OpencodeUpstreamClient, + UpstreamContractError, +) +from ..sandbox_policy import SandboxPolicy +from ..server.state_store import InterruptRequestRepository +from .contracts import PromptSubmission + + +class UnsupportedUpstreamOperationError(RuntimeError): + """Raised when the selected upstream protocol has no implemented operation.""" + + +class V2UpstreamAdapter(OpencodeUpstreamClient): + """Explicit V2 session transport using the shared V1 HTTP facade infrastructure.""" + + prompt_execution_supported = False + + def __init__( + self, + settings: Settings, + *, + interrupt_request_repository: InterruptRequestRepository | None = None, + ) -> None: + super().__init__(settings, interrupt_request_repository=interrupt_request_repository) + self._v2_server_info: Mapping[str, Any] | None = None + self._v2_server_info_lock = asyncio.Lock() + + def _resolve_location(self, directory: str | None, workspace_id: str | None) -> str: + if workspace_id is not None and workspace_id.strip(): + raise UnsupportedUpstreamOperationError( + "OpenCode V2 session transport does not support V1 workspace IDs." + ) + return ( + SandboxPolicy.from_settings( + self.settings, + workspace_root=self.directory, + ).resolve_directory(directory, default_directory=self.directory) + or "" + ) + + @staticmethod + def _unwrap_data(payload: Any, *, endpoint: str) -> Any: + if not isinstance(payload, Mapping) or "data" not in payload: + raise UpstreamContractError( + f"OpenCode V2 {endpoint} response must contain a data envelope" + ) + return payload["data"] + + async def _get_v2_json( + self, + path: str, + *, + endpoint: str, + params: Mapping[str, Any] | None = None, + ) -> Any: + if endpoint != "/api/info": + await self.get_server_info() + try: + return await self._get_json(path, endpoint=endpoint, params=params) + except httpx.HTTPStatusError as exc: + if exc.response.status_code == 404: + raise UpstreamContractError( + f"OpenCode V2 {endpoint} returned HTTP 404; verify the upstream is an " + "OpenCode V2 server and the requested location/session exists" + ) from exc + raise + + async def _post_v2_json( + self, + path: str, + *, + endpoint: str, + json_body: Any, + ) -> Any: + await self.get_server_info() + try: + return await self._post_json( + path, + endpoint=endpoint, + json_body=json_body, + ) + except httpx.HTTPStatusError as exc: + if exc.response.status_code == 404: + raise UpstreamContractError( + f"OpenCode V2 {endpoint} returned HTTP 404; verify the upstream is an " + "OpenCode V2 server and the requested location/session exists" + ) from exc + raise + + async def get_server_info(self) -> Mapping[str, Any]: + async with self._v2_server_info_lock: + if self._v2_server_info is not None: + return self._v2_server_info + payload = await self._get_v2_json("/api/info", endpoint="/api/info") + version = payload.get("version") if isinstance(payload, Mapping) else None + if not isinstance(version, str): + raise UpstreamContractError( + "OpenCode V2 /api/info response must include a string version" + ) + if not version.startswith("2."): + raise UpstreamContractError( + f"OpenCode V2 was selected but /api/info reports version {version!r}" + ) + server_info = cast(Mapping[str, Any], payload) + self._v2_server_info = server_info + return server_info + + async def get_location(self, *, directory: str | None = None) -> Mapping[str, Any]: + resolved_directory = self._resolve_location(directory, None) + payload = await self._get_v2_json( + "/api/location", + endpoint="/api/location", + params={"location[directory]": resolved_directory}, + ) + if not isinstance(payload, Mapping): + raise UpstreamContractError("OpenCode V2 /api/location response must be an object") + return payload + + async def create_session( + self, + title: str | None = None, + *, + directory: str | None = None, + workspace_id: str | None = None, + ) -> str: + resolved_directory = self._resolve_location(directory, workspace_id) + payload: dict[str, Any] = {"location": {"directory": resolved_directory}} + if title: + payload["title"] = title + response = await self._post_v2_json( + "/api/session", + endpoint="/api/session", + json_body=payload, + ) + data = self._unwrap_data(response, endpoint="/api/session") + session_id = data.get("id") if isinstance(data, Mapping) else None + if not isinstance(session_id, str) or not session_id.strip(): + raise UpstreamContractError("OpenCode V2 /api/session data is missing session id") + return session_id + + async def get_session( + self, + session_id: str, + *, + directory: str | None = None, + workspace_id: str | None = None, + ) -> Any: + normalized_id = session_id.strip() + if not normalized_id: + raise ValueError("session_id must not be empty") + resolved_directory = self._resolve_location(directory, workspace_id) + path = f"/api/session/{quote(normalized_id, safe='')}" + response = await self._get_v2_json( + path, + endpoint="/api/session/{sessionID}", + params={"location[directory]": resolved_directory}, + ) + return self._unwrap_data(response, endpoint="/api/session/{sessionID}") + + async def list_sessions( + self, + *, + params: dict[str, Any] | None = None, + directory: str | None = None, + workspace_id: str | None = None, + ) -> Any: + resolved_directory = self._resolve_location(directory, workspace_id) + query: dict[str, Any] = {"location[directory]": resolved_directory} + if params: + if any( + key in {"directory", "workspace", "location"} or key.startswith("location[") + for key in params + ): + raise ValueError("V2 session list location is selected by the location parameter") + query.update(params) + response = await self._get_v2_json( + "/api/session", + endpoint="/api/session", + params=query, + ) + data = self._unwrap_data(response, endpoint="/api/session") + if not isinstance(data, list): + raise UpstreamContractError("OpenCode V2 /api/session data must be an array") + return {"data": data, **{key: value for key, value in response.items() if key != "data"}} + + async def send_message( + 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 = ..., + ) -> OpencodeMessage: + del session_id, text, parts, directory, workspace_id, model_override, timeout_override + raise UnsupportedUpstreamOperationError( + "OpenCode V2 prompt execution is not implemented; use OPENCODE_API_VERSION=v1 " + "or wait for the V2 execution support release." + ) + + 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: + del session_id, text, parts, directory, workspace_id, model_override, timeout_override + raise UnsupportedUpstreamOperationError( + "OpenCode V2 prompt execution is not implemented; use OPENCODE_API_VERSION=v1 " + "or wait for the V2 execution support release." + ) diff --git a/tests/config/test_settings.py b/tests/config/test_settings.py index b9ef5cd5..65fcf041 100644 --- a/tests/config/test_settings.py +++ b/tests/config/test_settings.py @@ -26,6 +26,7 @@ def test_settings_missing_required(): def test_settings_use_bounded_upstream_defaults() -> None: settings = make_settings() + assert settings.opencode_api_version == "v1" assert settings.opencode_timeout_stream == 900.0 assert settings.opencode_max_concurrent_requests == 32 assert settings.opencode_max_concurrent_streams == 8 @@ -151,6 +152,32 @@ def test_settings_opencode_auth_defaults_to_opencode_username_without_password() assert settings.opencode_auth_password is None +@pytest.mark.parametrize("version", ("v1", "v2")) +def test_settings_accepts_explicit_opencode_api_version(version: str) -> None: + env = { + "OPENCODE_API_VERSION": version, + "A2A_STATIC_AUTH_CREDENTIALS": json.dumps( + [{"scheme": "bearer", "token": "test-token", "principal": "automation"}] + ), + } + with mock.patch.dict(os.environ, env, clear=True): + settings = Settings() + + assert settings.opencode_api_version == version + + +def test_settings_rejects_unknown_opencode_api_version() -> None: + env = { + "OPENCODE_API_VERSION": "auto", + "A2A_STATIC_AUTH_CREDENTIALS": json.dumps( + [{"scheme": "bearer", "token": "test-token", "principal": "automation"}] + ), + } + with mock.patch.dict(os.environ, env, clear=True): + with pytest.raises(ValidationError, match="OPENCODE_API_VERSION"): + Settings() + + def test_settings_opencode_auth_repr_hides_password() -> None: settings = make_settings( opencode_auth_username="opencode", diff --git a/tests/execution/test_v2_prompt_failfast.py b/tests/execution/test_v2_prompt_failfast.py new file mode 100644 index 00000000..fe3bd53e --- /dev/null +++ b/tests/execution/test_v2_prompt_failfast.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from unittest.mock import AsyncMock + +import pytest +from a2a.types import Task, TaskState + +from opencode_a2a.execution.executor import OpencodeAgentExecutor +from tests.support.helpers import DummyEventQueue, make_request_context +from tests.support.settings import make_settings + + +@pytest.mark.asyncio +async def test_v2_prompt_is_failed_before_session_or_upstream_side_effect() -> None: + client = AsyncMock() + client.settings = make_settings(opencode_api_version="v2") + client.directory = "/tmp/opencode-v2-workspace" + client.prompt_execution_supported = False + executor = OpencodeAgentExecutor(client, streaming_enabled=True) + queue = DummyEventQueue() + + await executor.execute( + make_request_context(task_id="task-v2", context_id="ctx-v2", text="hello"), + queue, + ) + + tasks = [event for event in queue.events if isinstance(event, Task)] + assert tasks + assert tasks[-1].status.state == TaskState.TASK_STATE_FAILED + assert "V2 prompt execution is not implemented" in tasks[-1].status.message.parts[0].text + client.create_session.assert_not_awaited() + client.send_message.assert_not_awaited() diff --git a/tests/jsonrpc/test_jsonrpc_unsupported_method.py b/tests/jsonrpc/test_jsonrpc_unsupported_method.py index 9599c539..75850183 100644 --- a/tests/jsonrpc/test_jsonrpc_unsupported_method.py +++ b/tests/jsonrpc/test_jsonrpc_unsupported_method.py @@ -1,8 +1,10 @@ import httpx import pytest +from opencode_a2a.contracts.extensions import SESSION_BINDING_EXTENSION_URI from opencode_a2a.protocol_versions import A2A_PROTOCOL_VERSION from opencode_a2a.server.application import create_app +from opencode_a2a.upstream.v2 import V2UpstreamAdapter from tests.support.settings import make_settings @@ -36,6 +38,38 @@ async def test_unsupported_method_returns_unified_error() -> None: assert data["protocolVersion"] == A2A_PROTOCOL_VERSION +@pytest.mark.asyncio +async def test_v2_rejects_unimplemented_opencode_extension_methods() -> None: + app = create_app(make_settings(opencode_api_version="v2")) + assert isinstance(app.state.upstream_client, V2UpstreamAdapter) + openapi = app.openapi() + assert "x-a2a-extension-contracts" not in openapi["paths"]["/"]["post"] + transport = httpx.ASGITransport(app=app) + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.post( + "/", + headers={ + "Authorization": "Bearer test-token", + "A2A-Extensions": SESSION_BINDING_EXTENSION_URI, + }, + json={ + "jsonrpc": "2.0", + "id": 123, + "method": "opencode.sessions.prompt_async", + "params": {"sessionID": "ses-1", "parts": [{"type": "text", "text": "hello"}]}, + }, + ) + + assert response.status_code == 200 + body = response.json() + assert body["error"]["code"] == -32601 + assert body["error"]["message"] == "Method not found" + assert "opencode.sessions.prompt_async" not in body["error"]["data"]["supportedMethods"] + assert "opencode.sessions.list" not in body["error"]["data"]["supportedMethods"] + assert "A2A-Extensions" not in response.headers + + @pytest.mark.asyncio async def test_unsupported_method_uses_requested_protocol_version() -> None: settings = make_settings(test_bearer_token="test-token") diff --git a/tests/server/test_agent_card.py b/tests/server/test_agent_card.py index 38c70400..0ae21502 100644 --- a/tests/server/test_agent_card.py +++ b/tests/server/test_agent_card.py @@ -60,6 +60,24 @@ def test_agent_card_description_reflects_actual_transport_capabilities() -> None assert skills_by_id["opencode.interrupt.callback"].output_modes == ["application/json"] +def test_v2_agent_card_does_not_advertise_unimplemented_chat_or_extensions() -> None: + card = build_agent_card(make_settings(opencode_api_version="v2")) + extended_card = build_agent_card( + make_settings(opencode_api_version="v2"), + include_detailed_contracts=True, + ) + + assert "session transport is experimental" in card.description + assert ( + "prompt execution and OpenCode-specific extensions are not implemented" in card.description + ) + assert card.capabilities.extensions == [] + assert extended_card.capabilities.extensions == [] + assert [skill.id for skill in card.skills] == ["v2-session-transport"] + assert [skill.id for skill in extended_card.skills] == ["v2-session-transport"] + assert "no A2A session or prompt operation is currently available" in card.skills[0].description + + def test_agent_card_declares_optional_basic_auth_when_configured() -> None: card = build_agent_card( make_settings( diff --git a/tests/upstream/test_v2_upstream_adapter.py b/tests/upstream/test_v2_upstream_adapter.py new file mode 100644 index 00000000..07a4e922 --- /dev/null +++ b/tests/upstream/test_v2_upstream_adapter.py @@ -0,0 +1,212 @@ +from __future__ import annotations + +import json +from typing import Any + +import httpx +import pytest + +from opencode_a2a.opencode_upstream_client import UpstreamContractError +from opencode_a2a.upstream.v2 import UnsupportedUpstreamOperationError, V2UpstreamAdapter +from tests.support.settings import make_settings + + +def _make_client( + monkeypatch: pytest.MonkeyPatch, + handler, + *, + workspace_root: str, +) -> tuple[V2UpstreamAdapter, list[httpx.Request]]: + requests: list[httpx.Request] = [] + + async def capture(request: httpx.Request) -> httpx.Response: + requests.append(request) + return await handler(request) + + client = V2UpstreamAdapter( + make_settings( + opencode_workspace_root=workspace_root, + opencode_auth_username="gateway", + opencode_auth_password="test-password", # pragma: allowlist secret + ) + ) + monkeypatch.setattr(client._client, "_transport", httpx.MockTransport(capture)) + return client, requests + + +@pytest.mark.asyncio +async def test_v2_session_calls_use_documented_routes_locations_and_data_envelopes( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + + async def respond(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/info": + payload: Any = {"version": "2.0.28", "pid": 123, "urls": []} + elif request.url.path == "/api/location": + payload = {"directory": str(workspace)} + elif request.method == "POST": + payload = {"data": {"id": "ses-created", "title": "topic"}} + elif request.url.path == "/api/session": + payload = {"data": [{"id": "ses-created"}], "cursor": {"next": "cursor-2"}} + else: + payload = {"data": {"id": "ses-created", "title": "topic"}} + return httpx.Response(200, json=payload) + + client, requests = _make_client(monkeypatch, respond, workspace_root=str(workspace)) + try: + identity = await client.get_server_info() + location = await client.get_location() + session_id = await client.create_session("topic") + session = await client.get_session(session_id) + page = await client.list_sessions(directory=str(workspace), params={"limit": 5}) + finally: + await client.close() + + assert identity["version"] == "2.0.28" + assert location["directory"] == str(workspace) + assert session_id == "ses-created" + assert session["title"] == "topic" + assert page == {"data": [{"id": "ses-created"}], "cursor": {"next": "cursor-2"}} + assert [request.url.path for request in requests] == [ + "/api/info", + "/api/location", + "/api/session", + "/api/session/ses-created", + "/api/session", + ] + assert requests[1].url.params["location[directory]"] == str(workspace) + assert json.loads(requests[2].content) == { + "location": {"directory": str(workspace)}, + "title": "topic", + } + assert requests[3].url.params["location[directory]"] == str(workspace) + assert requests[4].url.params["location[directory]"] == str(workspace) + assert requests[4].url.params["limit"] == "5" + assert requests[0].headers["authorization"].startswith("Basic ") + + +@pytest.mark.asyncio +async def test_v2_rejects_outside_workspace_and_v1_workspace_id_before_http( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + + async def respond(_request: httpx.Request) -> httpx.Response: + raise AssertionError("unauthorized request reached the upstream") + + client, requests = _make_client(monkeypatch, respond, workspace_root=str(workspace)) + escaped = tmp_path / "outside" + escaped.mkdir() + (workspace / "escape").symlink_to(escaped, target_is_directory=True) + try: + with pytest.raises(ValueError, match="outside the allowed workspace"): + await client.create_session("topic", directory=str(tmp_path)) + with pytest.raises(ValueError, match="outside the allowed workspace"): + await client.create_session("topic", directory=str(workspace / "escape")) + with pytest.raises(UnsupportedUpstreamOperationError, match="workspace IDs"): + await client.create_session("topic", workspace_id="workspace-1") + with pytest.raises(ValueError, match="location is selected"): + await client.list_sessions(params={"location[directory]": str(tmp_path)}) + finally: + await client.close() + + assert requests == [] + + +@pytest.mark.asyncio +async def test_v2_requires_data_envelope_and_reports_mismatched_route( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + + async def missing_envelope(_request: httpx.Request) -> httpx.Response: + if _request.url.path == "/api/info": + return httpx.Response(200, json={"version": "2.0.28"}) + return httpx.Response(200, json={"id": "ses-unwrapped"}) + + client, _ = _make_client(monkeypatch, missing_envelope, workspace_root=str(workspace)) + try: + with pytest.raises(UpstreamContractError, match="data envelope"): + await client.create_session("topic") + finally: + await client.close() + + async def html_response(_request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, text="wrong upstream", headers={"content-type": "text/html"} + ) + + client, _ = _make_client(monkeypatch, html_response, workspace_root=str(workspace)) + try: + with pytest.raises(UpstreamContractError, match="returned non-JSON response"): + await client.get_server_info() + finally: + await client.close() + + async def not_found(_request: httpx.Request) -> httpx.Response: + return httpx.Response(404, json={"error": "not found"}) + + client, _ = _make_client(monkeypatch, not_found, workspace_root=str(workspace)) + try: + with pytest.raises(UpstreamContractError, match="verify the upstream is an OpenCode V2"): + await client.get_server_info() + finally: + await client.close() + + +@pytest.mark.asyncio +async def test_v2_prompt_methods_fail_without_http_side_effect( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + + async def respond(_request: httpx.Request) -> httpx.Response: + raise AssertionError("prompt endpoint must not be reached") + + client, requests = _make_client(monkeypatch, respond, workspace_root=str(workspace)) + try: + with pytest.raises( + UnsupportedUpstreamOperationError, + match="prompt execution is not implemented", + ): + await client.send_message("ses-1", "hello") + with pytest.raises( + UnsupportedUpstreamOperationError, + match="prompt execution is not implemented", + ): + await client.submit_prompt("ses-1", "hello") + finally: + await client.close() + + assert requests == [] + + +@pytest.mark.asyncio +async def test_v2_refuses_to_use_session_route_when_info_reports_v1( + monkeypatch: pytest.MonkeyPatch, + tmp_path, +) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + + async def v1_server(request: httpx.Request) -> httpx.Response: + assert request.url.path == "/api/info" + return httpx.Response(200, json={"version": "1.18.19"}) + + client, requests = _make_client(monkeypatch, v1_server, workspace_root=str(workspace)) + try: + with pytest.raises(UpstreamContractError, match="V2 was selected but /api/info reports"): + await client.create_session("topic") + finally: + await client.close() + + assert [request.url.path for request in requests] == ["/api/info"] From a1362e8bd122cec190655b062d97415693a47978 Mon Sep 17 00:00:00 2001 From: Tom Brandenburg Date: Sun, 25 Oct 2026 11:52:25 +0100 Subject: [PATCH 4/9] fix(card): avoid advertising V2 execution capabilities (#3) --- src/opencode_a2a/server/agent_card.py | 18 ++++++++---------- tests/server/test_agent_card.py | 15 ++++++++++----- 2 files changed, 18 insertions(+), 15 deletions(-) diff --git a/src/opencode_a2a/server/agent_card.py b/src/opencode_a2a/server/agent_card.py index daf02cce..e6cf3eaf 100644 --- a/src/opencode_a2a/server/agent_card.py +++ b/src/opencode_a2a/server/agent_card.py @@ -68,8 +68,8 @@ def _build_agent_card_description( base = (settings.a2a_description or "").strip() or "OpenCode A2A runtime." if settings.opencode_api_version == "v2": return ( - f"{base} OpenCode V2 session transport is experimental; A2A prompt execution and " - "OpenCode-specific extensions are not implemented." + f"{base} Only internal experimental OpenCode V2 session transport exists; " + "A2A message execution and OpenCode-specific extensions are unsupported." ) if not include_detailed_contracts: public_parts: list[str] = [ @@ -364,15 +364,13 @@ def _build_agent_skills( if settings.opencode_api_version == "v2": return [ AgentSkill( - id="v2-session-transport", - name="OpenCode V2 Session Transport", + id="a2a-execution-unavailable", + name="A2A Execution Status", description=( - "Experimental upstream session transport for adapter integrations; " - "no A2A session or prompt operation is currently available." + "Informational only: internal experimental OpenCode V2 session transport " + "exists, but no A2A message or session operation is available." ), - input_modes=list(_JSON_RPC_MODES), - output_modes=list(_JSON_RPC_MODES), - tags=["experimental", "transport"], + tags=["status"], ) ] if not include_detailed_contracts: @@ -550,7 +548,7 @@ def build_agent_card( default_input_modes=list(_CHAT_INPUT_MODES), default_output_modes=list(_CHAT_OUTPUT_MODES), capabilities=AgentCapabilities( - streaming=True, + streaming=settings.opencode_api_version == "v1", extended_agent_card=True, extensions=( [] diff --git a/tests/server/test_agent_card.py b/tests/server/test_agent_card.py index 0ae21502..8de60e0e 100644 --- a/tests/server/test_agent_card.py +++ b/tests/server/test_agent_card.py @@ -67,15 +67,20 @@ def test_v2_agent_card_does_not_advertise_unimplemented_chat_or_extensions() -> include_detailed_contracts=True, ) - assert "session transport is experimental" in card.description + assert "Only internal experimental OpenCode V2 session transport exists" in card.description assert ( - "prompt execution and OpenCode-specific extensions are not implemented" in card.description + "A2A message execution and OpenCode-specific extensions are unsupported" in card.description ) + assert card.capabilities.streaming is False + assert extended_card.capabilities.streaming is False assert card.capabilities.extensions == [] assert extended_card.capabilities.extensions == [] - assert [skill.id for skill in card.skills] == ["v2-session-transport"] - assert [skill.id for skill in extended_card.skills] == ["v2-session-transport"] - assert "no A2A session or prompt operation is currently available" in card.skills[0].description + for v2_card in (card, extended_card): + assert [skill.id for skill in v2_card.skills] == ["a2a-execution-unavailable"] + skill = v2_card.skills[0] + assert "no A2A message or session operation is available" in skill.description + assert skill.input_modes == [] + assert skill.output_modes == [] def test_agent_card_declares_optional_basic_auth_when_configured() -> None: From 5d832336b6b5475c0086f359835f5d3e47c09a0a Mon Sep 17 00:00:00 2001 From: Tom Brandenburg Date: Sun, 25 Oct 2026 11:54:43 +0100 Subject: [PATCH 5/9] test(upstream): cover V2 session status failures (#3) --- docs/compatibility.md | 4 ++ tests/upstream/test_v2_upstream_adapter.py | 46 ++++++++++++++++++++++ 2 files changed, 50 insertions(+) diff --git a/docs/compatibility.md b/docs/compatibility.md index 17e9bed1..f235787d 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -16,6 +16,10 @@ OpenCode-specific JSON-RPC extensions remain unavailable. V1 stays the default, and the adapter does not auto-detect or fall back between versions. The V2 session transport was exercised against OpenCode `2.0.28`; that experimental check does not add V2 to the supported runtime line above. +OpenCode V2 allows `external_directory` by default. Before any future release of +V2 prompt execution, operators must configure an explicit restrictive upstream +V2 `permissions` policy. The current session-only slice does not perform model +execution. This is the single canonical place where the supported upstream OpenCode version line is declared. Keep it updated together with the live integration smoke check (`./scripts/live_opencode_smoke.sh`) and do not duplicate the version line elsewhere. diff --git a/tests/upstream/test_v2_upstream_adapter.py b/tests/upstream/test_v2_upstream_adapter.py index 07a4e922..5b25636c 100644 --- a/tests/upstream/test_v2_upstream_adapter.py +++ b/tests/upstream/test_v2_upstream_adapter.py @@ -161,6 +161,52 @@ async def not_found(_request: httpx.Request) -> httpx.Response: await client.close() +@pytest.mark.asyncio +@pytest.mark.parametrize("status_code", (401, 409)) +async def test_v2_session_http_status_errors_are_typed_without_leaking_request_data( + status_code: int, + monkeypatch: pytest.MonkeyPatch, + tmp_path, +) -> None: + workspace = tmp_path / "workspace" + workspace.mkdir() + request_title = "private-session-request-title" + observed_requests: list[httpx.Request] = [] + + async def respond(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/info": + return httpx.Response(200, json={"version": "2.0.28"}) + observed_requests.append(request) + echoed_request = request.content.decode() + echoed_authorization = request.headers.get("authorization", "") + return httpx.Response( + status_code, + json={ + "message": "upstream rejected request", + "request": echoed_request, + "authorization": echoed_authorization, + }, + ) + + client, _ = _make_client(monkeypatch, respond, workspace_root=str(workspace)) + try: + with pytest.raises(httpx.HTTPStatusError) as exc_info: + await client.create_session(request_title) + finally: + await client.close() + + error = exc_info.value + message = str(error) + assert error.response.status_code == status_code + assert error.request.url.path == "/api/session" + assert len(observed_requests) == 1 + assert request_title.encode() in observed_requests[0].content + assert "Basic " in observed_requests[0].headers["authorization"] + assert request_title not in message + assert observed_requests[0].headers["authorization"] not in message + assert "test-password" not in message + + @pytest.mark.asyncio async def test_v2_prompt_methods_fail_without_http_side_effect( monkeypatch: pytest.MonkeyPatch, From 73ffcd8ca97170f6e12caf91cd7e0c43d7a3ef14 Mon Sep 17 00:00:00 2001 From: Tom Brandenburg Date: Sun, 25 Oct 2026 13:17:23 +0100 Subject: [PATCH 6/9] 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( From 8d5494a2c279b297fd8394fc23f2c27642651e64 Mon Sep 17 00:00:00 2001 From: Tom Brandenburg Date: Sun, 25 Oct 2026 12:57:55 +0100 Subject: [PATCH 7/9] fix(compatibility): gate extension capabilities by API version (#6) --- docs/compatibility.md | 39 +++++- docs/extension-specifications.md | 7 + docs/guide.md | 2 +- .../contracts/extensions/__init__.py | 7 +- .../contracts/extensions/capabilities.py | 120 +++++++++++++--- .../contracts/extensions/compatibility.py | 41 +++--- .../contracts/extensions/private_params.py | 44 ++++-- src/opencode_a2a/jsonrpc/application.py | 23 +-- src/opencode_a2a/server/agent_card.py | 42 ++++-- src/opencode_a2a/server/application.py | 47 +++---- src/opencode_a2a/server/middleware.py | 9 +- src/opencode_a2a/server/openapi.py | 119 ++++++++++------ .../test_extension_contract_consistency.py | 96 +++++++++++++ .../test_jsonrpc_unsupported_method.py | 132 +++++++++++++++++- tests/server/test_agent_card.py | 12 +- tests/server/test_app_behaviors.py | 2 + 16 files changed, 591 insertions(+), 151 deletions(-) diff --git a/docs/compatibility.md b/docs/compatibility.md index f235787d..2becf290 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -5,16 +5,47 @@ This document defines the compatibility promises `opencode-a2a` currently uphold ## Runtime Support - Python versions: 3.11, 3.12, 3.13, 3.14 -- A2A SDK line: `1.x.y` +- A2A Python SDK: `a2a-sdk==1.2.3` (SDK package version; not an OpenCode API version) - Supported A2A protocol line: `1.0` - OpenCode runtime line: `1.18.x` (verified with `1.18.19`) OpenCode V2 is an opt-in, experimental session-transport slice selected with `OPENCODE_API_VERSION=v2`. It implements `/api/info`, `/api/location`, and session create/get/list transport only; A2A prompt execution and all -OpenCode-specific JSON-RPC extensions remain unavailable. V1 stays the -default, and the adapter does not auto-detect or fall back between versions. -The V2 session transport was exercised against OpenCode `2.0.28`; that +OpenCode-specific JSON-RPC extensions remain unavailable. This remains an +explicit safety gate, not a claim that V2 prompt execution works. + +The V2 selector is **OpenCode's upstream HTTP API version**, not "A2A SDK V2". +The runtime continues to expose A2A protocol 1.0 interfaces. Its V2 Agent Card +contains no extension declarations and retains only an informational execution +status skill. The A2A Python SDK card contract used here requires populated +default input/output mode fields and a skill entry; those required card fields +are not a claim that V2 accepts A2A messages. `SendMessage` enters the SDK task +lifecycle and returns a failed task with the fail-closed explanation before +session creation or prompt submission. `SendStreamingMessage` is rejected as +unsupported because the card declares `streaming=false`. Neither path makes an +upstream HTTP request. Direct +`opencode.*` JSON-RPC calls return `-32601 Method not found` without an upstream +HTTP request. The same extension hiding applies to shared interrupt callbacks. + +| Capability family | OpenCode API V1 (`1.18.x`, verified `1.18.19`) | OpenCode API V2 (`2.0.28` session-transport exercise only) | +| --- | --- | --- | +| A2A protocol interfaces | HTTP+JSON and JSON-RPC, protocol `1.0`; current core behavior | Interfaces remain protocol `1.0`; message send/stream fails before upstream HTTP | +| Shared session/model/stream/interrupt extensions | Declared and implemented as documented | Omitted from cards/OpenAPI; direct calls are unsupported | +| `opencode.sessions.*` (17 catalog methods) | Supported; `opencode.sessions.shell` is deployment-conditional | Omitted and rejected before upstream HTTP | +| Provider/model discovery (2 methods) | Supported | Omitted and rejected before upstream HTTP | +| Project/workspace/worktree (9 methods) | Project discovery stable; workspace/worktree discovery experimental; five mutations deployment-conditional | Omitted and rejected before upstream HTTP | +| Interrupt recovery (2 methods) | Adapter-local and caller-scoped | Omitted and rejected before upstream HTTP | +| `a2a.interrupt.*` callbacks (3 methods) | Supported through the shared interrupt extension | Omitted and rejected before upstream HTTP | +| Internal V2 session transport | Not applicable | `/api/info`, `/api/location`, session create/get/list only; exercised against `2.0.28`, not a model execution test | + +V2 support does **not** claim A2A execution, task-scoped cancellation, +permission/form interventions, request-scoped model overrides, structured tool +execution, or outbound `a2a_call` support. `2.0.28` evidence is limited to +session transport; no V2 model-backed acceptance run is claimed. + +V1 stays the default, and the adapter does not auto-detect or fall back between +versions. The V2 session transport was exercised against OpenCode `2.0.28`; that experimental check does not add V2 to the supported runtime line above. OpenCode V2 allows `external_directory` by default. Before any future release of V2 prompt execution, operators must configure an explicit restrictive upstream diff --git a/docs/extension-specifications.md b/docs/extension-specifications.md index 6ead3b76..64adf70f 100644 --- a/docs/extension-specifications.md +++ b/docs/extension-specifications.md @@ -16,6 +16,13 @@ Provider-private contract note: - Shared `metadata.shared.*` contracts are intended to remain low-risk and transportable. - Compatibility and wire-contract URIs are descriptive metadata contracts, not activatable runtime capabilities. - URI path segments identify the contract, not its auth or disclosure tier; public-vs-extended disclosure is controlled by Agent Card and OpenAPI surfaces. +- These declarations describe the V1 extension contract. With `OPENCODE_API_VERSION=v2`, the public and authenticated Agent Cards and OpenAPI publish no extension declarations or extension method metadata. Direct `opencode.*` and `a2a.interrupt.*` JSON-RPC calls return `-32601 Method not found` before any upstream HTTP request. + +OpenCode API V2 is not A2A SDK V2 and does not enable this extension index. +V2's informational status skill and its non-empty Agent Card default modes are +schema/discovery requirements only; they do not claim A2A message execution, +session methods, model overrides, cancellation, or intervention support. See +[`compatibility.md`](./compatibility.md#runtime-support) for the current matrix. ## SDK and Discovery Compatibility diff --git a/docs/guide.md b/docs/guide.md index 7561bb27..b6c158ae 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -29,7 +29,7 @@ Key variables to understand protocol behavior: - `A2A_STATIC_AUTH_CREDENTIALS`: required static credential registry in JSON array form. Supports multiple `bearer` / `basic` credentials, bearer-only required `principal`, optional `credential_id`, optional `capabilities`, and optional `enabled=false` for explicit disablement. - `OPENCODE_BASE_URL`: upstream OpenCode HTTP endpoint. Default: `http://127.0.0.1:4096`. In two-process deployments, set it explicitly. -- `OPENCODE_API_VERSION`: upstream API selector, `v1` (default) or `v2`. Selection is explicit; the service never detects or falls back to another API version. V2 is experimental session transport only: prompt execution and OpenCode-specific extensions are rejected/hidden until later support lands. +- `OPENCODE_API_VERSION`: OpenCode upstream API selector, `v1` (default) or `v2`; this is unrelated to the `a2a-sdk` package version. Selection is explicit, with no auto-detection or fallback. V2 currently provides internal session transport only, not A2A execution. The A2A 1.0 interfaces remain on the card; the V2 status skill is informational, and SDK-required non-empty default input/output modes do not promise that messages are executable. `SendMessage` returns a failed task before session creation or prompt HTTP; `SendStreamingMessage` is rejected because V2 declares `streaming=false`. Direct extension calls return `-32601 Method not found` before upstream HTTP. OpenCode-specific and shared extension declarations remain hidden. See [compatibility](./compatibility.md#runtime-support) for pinned evidence and limitations. - `OPENCODE_WORKSPACE_ROOT`: service-level default workspace root exposed to OpenCode when clients do not request a narrower directory override. - `A2A_ALLOW_DIRECTORY_OVERRIDE`: controls whether clients may pass `metadata.opencode.directory`. - `A2A_ENABLE_SESSION_SHELL`: gates high-risk JSON-RPC method `opencode.sessions.shell`. diff --git a/src/opencode_a2a/contracts/extensions/__init__.py b/src/opencode_a2a/contracts/extensions/__init__.py index feddae75..cded8241 100644 --- a/src/opencode_a2a/contracts/extensions/__init__.py +++ b/src/opencode_a2a/contracts/extensions/__init__.py @@ -1,7 +1,12 @@ from __future__ import annotations # ruff: noqa: F401 -from .capabilities import JsonRpcCapabilitySnapshot, build_capability_snapshot +from .capabilities import ( + DECLARED_EXTENSION_URIS, + JsonRpcCapabilitySnapshot, + OpenCodeApiVersion, + build_capability_snapshot, +) from .catalog import ( COMMAND_REQUEST_ALLOWED_FIELDS, INTERRUPT_CALLBACK_METHODS, diff --git a/src/opencode_a2a/contracts/extensions/capabilities.py b/src/opencode_a2a/contracts/extensions/capabilities.py index 735d03ee..f906b9d4 100644 --- a/src/opencode_a2a/contracts/extensions/capabilities.py +++ b/src/opencode_a2a/contracts/extensions/capabilities.py @@ -1,8 +1,9 @@ from __future__ import annotations -from collections.abc import Collection +from collections.abc import Collection, Mapping from dataclasses import dataclass -from typing import Any +from types import MappingProxyType +from typing import Any, Literal from ...profile.runtime import ( SESSION_SHELL_TOGGLE, @@ -11,6 +12,29 @@ ) from . import catalog, identifiers +OpenCodeApiVersion = Literal["v1", "v2"] + +DECLARED_EXTENSION_URIS: tuple[str, ...] = ( + identifiers.SESSION_BINDING_EXTENSION_URI, + identifiers.MODEL_SELECTION_EXTENSION_URI, + identifiers.STREAMING_EXTENSION_URI, + identifiers.SESSION_MANAGEMENT_EXTENSION_URI, + identifiers.PROVIDER_DISCOVERY_EXTENSION_URI, + identifiers.WORKSPACE_CONTROL_EXTENSION_URI, + identifiers.INTERRUPT_RECOVERY_EXTENSION_URI, + identifiers.INTERRUPT_CALLBACK_EXTENSION_URI, +) + +_ALL_EXTENSION_METHODS = frozenset( + ( + *catalog.SESSION_METHODS.values(), + *catalog.PROVIDER_DISCOVERY_METHODS.values(), + *catalog.WORKSPACE_CONTROL_METHODS.values(), + *catalog.INTERRUPT_RECOVERY_METHODS.values(), + *catalog.INTERRUPT_CALLBACK_METHODS.values(), + ) +) + @dataclass(frozen=True) class DeploymentConditionalMethod: @@ -46,47 +70,94 @@ def disabled_wire_contract_entry(self) -> dict[str, str] | None: @dataclass(frozen=True) class JsonRpcCapabilitySnapshot: - conditional_methods: dict[str, DeploymentConditionalMethod] + upstream_api_version: OpenCodeApiVersion + conditional_methods: Mapping[str, DeploymentConditionalMethod] + + @property + def extensions_enabled(self) -> bool: + return self.upstream_api_version == "v1" + + @property + def a2a_message_execution_enabled(self) -> bool: + return self.upstream_api_version == "v1" + + @property + def streaming_enabled(self) -> bool: + return self.upstream_api_version == "v1" + + @property + def declared_extension_uris(self) -> tuple[str, ...]: + return DECLARED_EXTENSION_URIS if self.extensions_enabled else () def is_method_enabled(self, method: str) -> bool: + if method in _ALL_EXTENSION_METHODS and not self.extensions_enabled: + return False conditional_method = self.conditional_methods.get(method) if conditional_method is None: return True return conditional_method.enabled def session_management_methods(self) -> dict[str, str]: + if not self.extensions_enabled: + return {} methods = dict(catalog.SESSION_METHODS) if not self.is_method_enabled(catalog.SESSION_METHODS["shell"]): methods.pop("shell", None) return methods + def dispatch_method_bindings(self) -> dict[str, str]: + """Return handler registry bindings; every call is still checked against this policy.""" + methods = self.session_management_methods() + if not self.extensions_enabled: + methods = dict(catalog.SESSION_METHODS) + if not self._deployment_method_enabled(catalog.SESSION_METHODS["shell"]): + methods.pop("shell", None) + methods.update(catalog.PROVIDER_DISCOVERY_METHODS) + methods.update(self.dispatch_workspace_control_methods()) + methods.update(catalog.INTERRUPT_RECOVERY_METHODS) + methods.update(catalog.INTERRUPT_CALLBACK_METHODS) + return methods + + def _deployment_method_enabled(self, method: str) -> bool: + conditional_method = self.conditional_methods.get(method) + return conditional_method is None or conditional_method.enabled + + def dispatch_workspace_control_methods(self) -> dict[str, str]: + methods = dict(catalog.WORKSPACE_DISCOVERY_METHODS) + for key, method in catalog.WORKSPACE_MUTATION_METHODS.items(): + if self._deployment_method_enabled(method): + methods[key] = method + return methods + def session_control_methods(self) -> dict[str, str]: + if not self.extensions_enabled: + return {} methods = dict(catalog.SESSION_CONTROL_METHODS) if not self.is_method_enabled(catalog.SESSION_CONTROL_METHODS["shell"]): methods.pop("shell", None) return methods def workspace_control_methods(self) -> dict[str, str]: - methods = dict(catalog.WORKSPACE_DISCOVERY_METHODS) - for key, method in catalog.WORKSPACE_MUTATION_METHODS.items(): - if self.is_method_enabled(method): - methods[key] = method - return methods + if not self.extensions_enabled: + return {} + return self.dispatch_workspace_control_methods() def supported_jsonrpc_methods(self) -> list[str]: - methods = [ - *catalog.CORE_JSONRPC_METHODS, - *(method for key, method in catalog.SESSION_METHODS.items() if key != "shell"), - *catalog.PROVIDER_DISCOVERY_METHODS.values(), - *self.workspace_control_methods().values(), - *catalog.INTERRUPT_RECOVERY_METHODS.values(), - *catalog.INTERRUPT_CALLBACK_METHODS.values(), - ] + methods = list(catalog.CORE_JSONRPC_METHODS) + if not self.extensions_enabled: + return methods + methods.extend(method for key, method in catalog.SESSION_METHODS.items() if key != "shell") + methods.extend(catalog.PROVIDER_DISCOVERY_METHODS.values()) + methods.extend(self.workspace_control_methods().values()) + methods.extend(catalog.INTERRUPT_RECOVERY_METHODS.values()) + methods.extend(catalog.INTERRUPT_CALLBACK_METHODS.values()) if self.is_method_enabled(catalog.SESSION_CONTROL_METHODS["shell"]): methods.append(catalog.SESSION_CONTROL_METHODS["shell"]) return methods def extension_jsonrpc_methods(self) -> list[str]: + if not self.extensions_enabled: + return [] methods = [ *(method for key, method in catalog.SESSION_METHODS.items() if key != "shell"), *catalog.PROVIDER_DISCOVERY_METHODS.values(), @@ -99,6 +170,8 @@ def extension_jsonrpc_methods(self) -> list[str]: return methods def conditionally_available_methods(self) -> dict[str, dict[str, str]]: + if not self.extensions_enabled: + return {} return { method: disabled_entry for method, conditional_method in self.conditional_methods.items() @@ -106,6 +179,8 @@ def conditionally_available_methods(self) -> dict[str, dict[str, str]]: } def method_flags(self, methods: Collection[str]) -> dict[str, dict[str, Any]]: + if not self.extensions_enabled: + return {} return { method: conditional_method.control_method_flag() for method, conditional_method in self.conditional_methods.items() @@ -113,13 +188,19 @@ def method_flags(self, methods: Collection[str]) -> dict[str, dict[str, Any]]: } def conditional_method_retention(self) -> dict[str, dict[str, Any]]: + if not self.extensions_enabled: + return {} return { method: conditional_method.method_retention() for method, conditional_method in self.conditional_methods.items() } -def build_capability_snapshot(*, runtime_profile: RuntimeProfile) -> JsonRpcCapabilitySnapshot: +def build_capability_snapshot( + *, + runtime_profile: RuntimeProfile, + upstream_api_version: OpenCodeApiVersion = "v1", +) -> JsonRpcCapabilitySnapshot: conditional_methods = { catalog.SESSION_CONTROL_METHODS["shell"]: DeploymentConditionalMethod( method=catalog.SESSION_CONTROL_METHODS["shell"], @@ -139,4 +220,7 @@ def build_capability_snapshot(*, runtime_profile: RuntimeProfile) -> JsonRpcCapa for method in catalog.WORKSPACE_MUTATION_METHODS.values() } ) - return JsonRpcCapabilitySnapshot(conditional_methods=conditional_methods) + return JsonRpcCapabilitySnapshot( + upstream_api_version=upstream_api_version, + conditional_methods=MappingProxyType(conditional_methods), + ) diff --git a/src/opencode_a2a/contracts/extensions/compatibility.py b/src/opencode_a2a/contracts/extensions/compatibility.py index 328b751c..5a45ba9f 100644 --- a/src/opencode_a2a/contracts/extensions/compatibility.py +++ b/src/opencode_a2a/contracts/extensions/compatibility.py @@ -5,17 +5,10 @@ from ...profile.runtime import RuntimeProfile from ...protocol_versions import A2A_PROTOCOL_VERSION, normalize_protocol_version from . import catalog, identifiers -from .capabilities import JsonRpcCapabilitySnapshot, build_capability_snapshot - -DECLARED_EXTENSION_URIS: tuple[str, ...] = ( - identifiers.SESSION_BINDING_EXTENSION_URI, - identifiers.MODEL_SELECTION_EXTENSION_URI, - identifiers.STREAMING_EXTENSION_URI, - identifiers.SESSION_MANAGEMENT_EXTENSION_URI, - identifiers.PROVIDER_DISCOVERY_EXTENSION_URI, - identifiers.WORKSPACE_CONTROL_EXTENSION_URI, - identifiers.INTERRUPT_RECOVERY_EXTENSION_URI, - identifiers.INTERRUPT_CALLBACK_EXTENSION_URI, +from .capabilities import ( + JsonRpcCapabilitySnapshot, + OpenCodeApiVersion, + build_capability_snapshot, ) @@ -115,7 +108,11 @@ def _build_method_retention_records( } -def _build_extension_retention() -> dict[str, dict[str, Any]]: +def _build_extension_retention( + capability_snapshot: JsonRpcCapabilitySnapshot, +) -> dict[str, dict[str, Any]]: + if not capability_snapshot.extensions_enabled: + return {} return { identifiers.SESSION_BINDING_EXTENSION_URI: _build_retention_record( surface="core-runtime-metadata", @@ -163,6 +160,8 @@ def _build_method_retention( surface="core", retention="required", ) + if not capability_snapshot.extensions_enabled: + return method_retention method_retention.update( _build_method_retention_records( [method for key, method in catalog.SESSION_METHODS.items() if key != "shell"], @@ -224,6 +223,8 @@ def build_compatibility_profile_params( runtime_profile: RuntimeProfile, supported_protocol_versions: tuple[str, ...] | list[str] | None = None, default_protocol_version: str | None = None, + capability_snapshot: JsonRpcCapabilitySnapshot | None = None, + upstream_api_version: OpenCodeApiVersion = "v1", ) -> dict[str, Any]: ( declared_protocol_version, @@ -238,7 +239,10 @@ def build_compatibility_profile_params( supported_protocol_versions=declared_supported_protocol_versions, default_protocol_version=declared_default_protocol_version, ) - capability_snapshot = build_capability_snapshot(runtime_profile=runtime_profile) + capability_snapshot = capability_snapshot or build_capability_snapshot( + runtime_profile=runtime_profile, + upstream_api_version=upstream_api_version, + ) service_behaviors = build_service_behavior_contract_params() return { **runtime_profile.summary_dict(protocol_version=declared_protocol_version), @@ -246,7 +250,7 @@ def build_compatibility_profile_params( "supported_protocol_versions": declared_supported_protocol_versions, "protocol_compatibility": protocol_compatibility, "core": _build_core_contract_surface(), - "extension_retention": _build_extension_retention(), + "extension_retention": _build_extension_retention(capability_snapshot), "method_retention": _build_method_retention(capability_snapshot), "service_behaviors": service_behaviors, "consumer_guidance": [ @@ -350,6 +354,8 @@ def build_wire_contract_params( runtime_profile: RuntimeProfile, supported_protocol_versions: tuple[str, ...] | list[str] | None = None, default_protocol_version: str | None = None, + capability_snapshot: JsonRpcCapabilitySnapshot | None = None, + upstream_api_version: OpenCodeApiVersion = "v1", ) -> dict[str, Any]: ( declared_protocol_version, @@ -364,7 +370,10 @@ def build_wire_contract_params( supported_protocol_versions=declared_supported_protocol_versions, default_protocol_version=declared_default_protocol_version, ) - capability_snapshot = build_capability_snapshot(runtime_profile=runtime_profile) + capability_snapshot = capability_snapshot or build_capability_snapshot( + runtime_profile=runtime_profile, + upstream_api_version=upstream_api_version, + ) service_behaviors = build_service_behavior_contract_params() return { @@ -381,7 +390,7 @@ def build_wire_contract_params( "conditionally_available_methods": ( capability_snapshot.conditionally_available_methods() ), - "extension_uris": list(DECLARED_EXTENSION_URIS), + "extension_uris": list(capability_snapshot.declared_extension_uris), }, "all_jsonrpc_methods": capability_snapshot.supported_jsonrpc_methods(), "service_behaviors": service_behaviors, diff --git a/src/opencode_a2a/contracts/extensions/private_params.py b/src/opencode_a2a/contracts/extensions/private_params.py index 0b4ef66a..04c3bc73 100644 --- a/src/opencode_a2a/contracts/extensions/private_params.py +++ b/src/opencode_a2a/contracts/extensions/private_params.py @@ -4,7 +4,7 @@ from ...profile.runtime import RuntimeProfile from . import catalog, identifiers, public_params -from .capabilities import build_capability_snapshot +from .capabilities import JsonRpcCapabilitySnapshot, build_capability_snapshot from .contract_docs import build_method_contract_docs @@ -19,11 +19,18 @@ def build_session_management_extension_params( *, runtime_profile: RuntimeProfile, context_id_prefix: str, + capability_snapshot: JsonRpcCapabilitySnapshot | None = None, ) -> dict[str, Any]: - capability_snapshot = build_capability_snapshot(runtime_profile=runtime_profile) + capability_snapshot = capability_snapshot or build_capability_snapshot( + runtime_profile=runtime_profile + ) methods = capability_snapshot.session_management_methods() - read_methods = dict(catalog.SESSION_READ_METHODS) - mutation_methods = dict(catalog.SESSION_MUTATION_METHODS) + read_methods = ( + dict(catalog.SESSION_READ_METHODS) if capability_snapshot.extensions_enabled else {} + ) + mutation_methods = ( + dict(catalog.SESSION_MUTATION_METHODS) if capability_snapshot.extensions_enabled else {} + ) control_methods = capability_snapshot.session_control_methods() active_session_methods = set(methods.values()) pagination_applies_to: list[str] = [] @@ -77,11 +84,19 @@ def build_session_management_extension_params( def build_interrupt_recovery_extension_params( *, runtime_profile: RuntimeProfile, + capability_snapshot: JsonRpcCapabilitySnapshot | None = None, ) -> dict[str, Any]: + capability_snapshot = capability_snapshot or build_capability_snapshot( + runtime_profile=runtime_profile + ) + methods = ( + dict(catalog.INTERRUPT_RECOVERY_METHODS) if capability_snapshot.extensions_enabled else {} + ) return { - "methods": dict(catalog.INTERRUPT_RECOVERY_METHODS), + "methods": methods, "method_contracts": build_method_contract_docs( - catalog.INTERRUPT_RECOVERY_METHOD_CONTRACTS.values() + catalog.INTERRUPT_RECOVERY_METHOD_CONTRACTS.values(), + active_methods=set(methods.values()), ), "supported_metadata": [], "provider_private_metadata": [], @@ -129,11 +144,19 @@ def build_interrupt_recovery_extension_params( def build_provider_discovery_extension_params( *, runtime_profile: RuntimeProfile, + capability_snapshot: JsonRpcCapabilitySnapshot | None = None, ) -> dict[str, Any]: + capability_snapshot = capability_snapshot or build_capability_snapshot( + runtime_profile=runtime_profile + ) + methods = ( + dict(catalog.PROVIDER_DISCOVERY_METHODS) if capability_snapshot.extensions_enabled else {} + ) return { - "methods": dict(catalog.PROVIDER_DISCOVERY_METHODS), + "methods": methods, "method_contracts": build_method_contract_docs( - catalog.PROVIDER_DISCOVERY_METHOD_CONTRACTS.values() + catalog.PROVIDER_DISCOVERY_METHOD_CONTRACTS.values(), + active_methods=set(methods.values()), ), "supported_metadata": ["opencode.directory", "opencode.workspace.id"], "provider_private_metadata": ["opencode.directory", "opencode.workspace.id"], @@ -190,8 +213,11 @@ def build_provider_discovery_extension_params( def build_workspace_control_extension_params( *, runtime_profile: RuntimeProfile, + capability_snapshot: JsonRpcCapabilitySnapshot | None = None, ) -> dict[str, Any]: - capability_snapshot = build_capability_snapshot(runtime_profile=runtime_profile) + capability_snapshot = capability_snapshot or build_capability_snapshot( + runtime_profile=runtime_profile + ) methods = capability_snapshot.workspace_control_methods() active_workspace_methods = set(methods.values()) diff --git a/src/opencode_a2a/jsonrpc/application.py b/src/opencode_a2a/jsonrpc/application.py index 701188f2..94776643 100644 --- a/src/opencode_a2a/jsonrpc/application.py +++ b/src/opencode_a2a/jsonrpc/application.py @@ -24,7 +24,7 @@ from starlette.requests import Request from starlette.responses import Response -from ..contracts.extensions import ALL_EXTENSION_URIS +from ..contracts.extensions import ALL_EXTENSION_URIS, JsonRpcCapabilitySnapshot from ..extension_negotiation import ( requested_extensions_from_call_context, ) @@ -63,8 +63,7 @@ def __init__( http_handler, upstream_client: OpencodeUpstreamClient, methods: dict[str, str], - supported_methods: list[str], - extensions_enabled: bool = True, + capability_snapshot: JsonRpcCapabilitySnapshot, directory_resolver: Callable[[str | None], str | None] | None = None, session_claim: Callable[..., Awaitable[bool]] | None = None, session_claim_finalize: Callable[..., Awaitable[None]] | None = None, @@ -77,7 +76,7 @@ def __init__( super().__init__(request_handler=http_handler, **kwargs) self._http_handler = http_handler self._upstream_client = upstream_client - self._extensions_enabled = extensions_enabled + self._capability_snapshot = capability_snapshot self._method_session_status = methods["status"] self._method_list_sessions = methods["list_sessions"] self._method_get_session = methods["get_session"] @@ -111,7 +110,7 @@ def __init__( self._method_reply_permission = methods["reply_permission"] self._method_reply_question = methods["reply_question"] self._method_reject_question = methods["reject_question"] - self._supported_methods = list(supported_methods) + self._supported_methods = capability_snapshot.supported_jsonrpc_methods() missing_control_hooks = [ name for name, hook in ( @@ -352,7 +351,9 @@ async def _handle_core_request( self._context_builder.build(request) ) request.state.activated_extensions = tuple( - value for value in requested_extensions if value in ALL_EXTENSION_URIS + value + for value in requested_extensions + if self._capability_snapshot.extensions_enabled and value in ALL_EXTENSION_URIS ) return JSONResponse( { @@ -423,11 +424,11 @@ async def handle_requests(self, request: Request) -> Response: except Exception: return await super().handle_requests(request) - extension_spec = ( - self._extension_method_registry.resolve(base_request.method) - if self._extensions_enabled - else None - ) + extension_spec = self._extension_method_registry.resolve(base_request.method) + if extension_spec is not None and not self._capability_snapshot.is_method_enabled( + base_request.method + ): + extension_spec = None if extension_spec is None: return await self._handle_core_request( request, diff --git a/src/opencode_a2a/server/agent_card.py b/src/opencode_a2a/server/agent_card.py index e6cf3eaf..6c51753b 100644 --- a/src/opencode_a2a/server/agent_card.py +++ b/src/opencode_a2a/server/agent_card.py @@ -63,13 +63,16 @@ def _build_agent_card_description( settings: Settings, runtime_profile: RuntimeProfile, *, + capability_snapshot: JsonRpcCapabilitySnapshot, include_detailed_contracts: bool, ) -> str: base = (settings.a2a_description or "").strip() or "OpenCode A2A runtime." - if settings.opencode_api_version == "v2": + if not capability_snapshot.a2a_message_execution_enabled: return ( f"{base} Only internal experimental OpenCode V2 session transport exists; " - "A2A message execution and OpenCode-specific extensions are unsupported." + "A2A SendMessage and SendStreamingMessage fail fast before any upstream request, " + "and OpenCode-specific extensions are unsupported. Agent Card default input/output " + "modes are required SDK schema fields, not a claim that message execution works." ) if not include_detailed_contracts: public_parts: list[str] = [ @@ -177,8 +180,11 @@ def _build_workspace_control_skill_examples(*, capability_snapshot) -> list[str] def _build_agent_extensions( *, runtime_profile: RuntimeProfile, + capability_snapshot: JsonRpcCapabilitySnapshot, include_detailed_contracts: bool, ) -> list[AgentExtension]: + if not capability_snapshot.declared_extension_uris: + return [] session_binding_extension_params = build_session_binding_extension_params( runtime_profile=runtime_profile, ) @@ -189,15 +195,19 @@ def _build_agent_extensions( session_management_extension_params = build_session_management_extension_params( runtime_profile=runtime_profile, context_id_prefix=SESSION_CONTEXT_PREFIX, + capability_snapshot=capability_snapshot, ) provider_discovery_extension_params = build_provider_discovery_extension_params( runtime_profile=runtime_profile, + capability_snapshot=capability_snapshot, ) workspace_control_extension_params = build_workspace_control_extension_params( runtime_profile=runtime_profile, + capability_snapshot=capability_snapshot, ) interrupt_recovery_extension_params = build_interrupt_recovery_extension_params( runtime_profile=runtime_profile, + capability_snapshot=capability_snapshot, ) interrupt_callback_extension_params = build_interrupt_callback_extension_params( runtime_profile=runtime_profile, @@ -205,10 +215,12 @@ def _build_agent_extensions( compatibility_profile_params = build_compatibility_profile_params( protocol_version=A2A_PROTOCOL_VERSION, runtime_profile=runtime_profile, + capability_snapshot=capability_snapshot, ) wire_contract_params = build_wire_contract_params( protocol_version=A2A_PROTOCOL_VERSION, runtime_profile=runtime_profile, + capability_snapshot=capability_snapshot, ) public_extensions = [ @@ -361,14 +373,15 @@ def _build_agent_skills( capability_snapshot: JsonRpcCapabilitySnapshot, include_detailed_contracts: bool, ) -> list[AgentSkill]: - if settings.opencode_api_version == "v2": + if not capability_snapshot.a2a_message_execution_enabled: return [ AgentSkill( id="a2a-execution-unavailable", name="A2A Execution Status", description=( "Informational only: internal experimental OpenCode V2 session transport " - "exists, but no A2A message or session operation is available." + "exists, but no A2A message or session operation is available. " + "SendMessage and SendStreamingMessage fail before any upstream request." ), tags=["status"], ) @@ -497,9 +510,14 @@ def build_agent_card( settings: Settings, *, include_detailed_contracts: bool = False, + capability_snapshot: JsonRpcCapabilitySnapshot | None = None, ) -> AgentCard: public_url = settings.a2a_public_url.rstrip("/") runtime_profile = build_runtime_profile(settings) + capability_snapshot = capability_snapshot or build_capability_snapshot( + runtime_profile=runtime_profile, + upstream_api_version=settings.opencode_api_version, + ) security_schemes: dict[str, SecurityScheme] = {} security_requirements: list[SecurityRequirement] = [] configured_auth_schemes = { @@ -522,13 +540,12 @@ def build_agent_card( ) ) security_requirements.append(SecurityRequirement(schemes=cast(Any, {"basicAuth": {}}))) - capability_snapshot = build_capability_snapshot(runtime_profile=runtime_profile) - return AgentCard( name=settings.a2a_title, description=_build_agent_card_description( settings, runtime_profile, + capability_snapshot=capability_snapshot, include_detailed_contracts=include_detailed_contracts, ), documentation_url=settings.a2a_documentation_url, @@ -548,15 +565,12 @@ def build_agent_card( default_input_modes=list(_CHAT_INPUT_MODES), default_output_modes=list(_CHAT_OUTPUT_MODES), capabilities=AgentCapabilities( - streaming=settings.opencode_api_version == "v1", + streaming=capability_snapshot.streaming_enabled, extended_agent_card=True, - extensions=( - [] - if settings.opencode_api_version == "v2" - else _build_agent_extensions( - runtime_profile=runtime_profile, - include_detailed_contracts=include_detailed_contracts, - ) + extensions=_build_agent_extensions( + runtime_profile=runtime_profile, + capability_snapshot=capability_snapshot, + include_detailed_contracts=include_detailed_contracts, ), ), skills=_build_agent_skills( diff --git a/src/opencode_a2a/server/application.py b/src/opencode_a2a/server/application.py index 9b5fc4e2..f65df15f 100644 --- a/src/opencode_a2a/server/application.py +++ b/src/opencode_a2a/server/application.py @@ -20,16 +20,10 @@ from ..a2a_protocol import ( AGENT_CARD_WELL_KNOWN_PATH, - CORE_JSONRPC_METHODS, EXTENDED_AGENT_CARD_PATH, ) from ..config import Settings -from ..contracts.extensions import ( - INTERRUPT_CALLBACK_METHODS, - INTERRUPT_RECOVERY_METHODS, - PROVIDER_DISCOVERY_METHODS, - build_capability_snapshot, -) +from ..contracts.extensions import build_capability_snapshot from ..execution.executor import OpencodeAgentExecutor from ..execution.metrics import render_prometheus_metrics from ..jsonrpc.application import ( @@ -182,8 +176,17 @@ def create_app(settings: Settings) -> FastAPI: interrupt_request_repository=runtime_state_runtime.interrupt_request_repository, ) client_manager = A2AClientManager(settings) - agent_card = build_agent_card(settings) - extended_agent_card = build_agent_card(settings, include_detailed_contracts=True) + runtime_profile = build_runtime_profile(settings) + capability_snapshot = build_capability_snapshot( + runtime_profile=runtime_profile, + upstream_api_version=settings.opencode_api_version, + ) + agent_card = build_agent_card(settings, capability_snapshot=capability_snapshot) + extended_agent_card = build_agent_card( + settings, + include_detailed_contracts=True, + capability_snapshot=capability_snapshot, + ) executor = OpencodeAgentExecutor( upstream_client, streaming_enabled=True, @@ -207,28 +210,14 @@ def create_app(settings: Settings) -> FastAPI: ) context_builder = IdentityAwareCallContextBuilder() - runtime_profile = build_runtime_profile(settings) - capability_snapshot = build_capability_snapshot(runtime_profile=runtime_profile) - - jsonrpc_methods = { - **capability_snapshot.session_management_methods(), - **dict(PROVIDER_DISCOVERY_METHODS), - **capability_snapshot.workspace_control_methods(), - **dict(INTERRUPT_RECOVERY_METHODS), - **dict(INTERRUPT_CALLBACK_METHODS), - } + jsonrpc_methods = capability_snapshot.dispatch_method_bindings() # Build JSON-RPC app (POST / by default) and attach REST endpoints (HTTP+JSON) to the same app. jsonrpc_app = OpencodeSessionManagementJSONRPCApplication( http_handler=handler, context_builder=context_builder, upstream_client=upstream_client, - supported_methods=( - list(CORE_JSONRPC_METHODS) - if settings.opencode_api_version == "v2" - else capability_snapshot.supported_jsonrpc_methods() - ), - extensions_enabled=settings.opencode_api_version == "v1", + capability_snapshot=capability_snapshot, directory_resolver=( partial( executor._sandbox_policy.resolve_directory, @@ -359,12 +348,18 @@ async def authenticated_extended_agent_card_route() -> JSONResponse: app.state.agent_executor = executor app.state.upstream_client = upstream_client app.state.a2a_client_manager = client_manager - _patch_jsonrpc_openapi_contract(app, settings, runtime_profile=runtime_profile) + _patch_jsonrpc_openapi_contract( + app, + settings, + runtime_profile=runtime_profile, + capability_snapshot=capability_snapshot, + ) install_runtime_middlewares( app, settings, public_card_etag=public_card_etag, extended_card_etag=extended_card_etag, + capability_snapshot=capability_snapshot, ) @app.get("/health") diff --git a/src/opencode_a2a/server/middleware.py b/src/opencode_a2a/server/middleware.py index fe61246b..e22fb231 100644 --- a/src/opencode_a2a/server/middleware.py +++ b/src/opencode_a2a/server/middleware.py @@ -22,7 +22,11 @@ build_static_auth_credentials, ) from ..client.network_policy import matches_allowed_host -from ..contracts.extensions import ALL_EXTENSION_URIS, PUBLIC_EXTENSION_URIS +from ..contracts.extensions import ( + ALL_EXTENSION_URIS, + PUBLIC_EXTENSION_URIS, + JsonRpcCapabilitySnapshot, +) from ..execution.metrics import emit_metric from ..jsonrpc.error_responses import ( adapt_jsonrpc_error, @@ -330,6 +334,7 @@ def install_runtime_middlewares( *, public_card_etag: str, extended_card_etag: str, + capability_snapshot: JsonRpcCapabilitySnapshot, ) -> None: def _requires_protocol_negotiation(request: Request) -> bool: if request.url.path == "/" and request.method == "POST": @@ -495,7 +500,7 @@ def _resolve_activated_extensions(request: Request) -> tuple[str, ...]: path = request.url.path supported_extensions: tuple[str, ...] - if settings.opencode_api_version == "v2": + if not capability_snapshot.extensions_enabled: supported_extensions = () elif path == AGENT_CARD_WELL_KNOWN_PATH: supported_extensions = PUBLIC_EXTENSION_URIS diff --git a/src/opencode_a2a/server/openapi.py b/src/opencode_a2a/server/openapi.py index daf5b3e0..bc9a27b2 100644 --- a/src/opencode_a2a/server/openapi.py +++ b/src/opencode_a2a/server/openapi.py @@ -12,6 +12,8 @@ MODEL_SELECTION_EXTENSION_URI, SESSION_BINDING_EXTENSION_URI, STREAMING_EXTENSION_URI, + JsonRpcCapabilitySnapshot, + build_capability_snapshot, build_interrupt_callback_extension_params, build_model_selection_extension_params, build_public_interrupt_callback_extension_params, @@ -25,7 +27,16 @@ from ..profile.runtime import RuntimeProfile -def _build_jsonrpc_extension_openapi_description() -> str: +def _build_jsonrpc_extension_openapi_description( + capability_snapshot: JsonRpcCapabilitySnapshot | None = None, +) -> str: + if capability_snapshot is not None and not capability_snapshot.extensions_enabled: + return ( + "A2A JSON-RPC retains the A2A protocol 1.0 method surface. With OpenCode API V2, " + "SendMessage and SendStreamingMessage fail before any upstream request; direct " + "opencode.* extension calls return JSON-RPC -32601 Method not found. No OpenCode " + "extension contracts are advertised." + ) interrupt_methods = ", ".join(sorted(INTERRUPT_CALLBACK_METHODS.values())) core_methods = ", ".join(CORE_JSONRPC_METHODS) return ( @@ -243,7 +254,12 @@ def _patch_jsonrpc_openapi_contract( settings: Settings, *, runtime_profile: RuntimeProfile, + capability_snapshot: JsonRpcCapabilitySnapshot | None = None, ) -> None: + capability_snapshot = capability_snapshot or build_capability_snapshot( + runtime_profile=runtime_profile, + upstream_api_version=settings.opencode_api_version, + ) session_binding = build_session_binding_extension_params( runtime_profile=runtime_profile, ) @@ -275,51 +291,50 @@ def custom_openapi() -> dict[str, Any]: post = root_path.get("post") if isinstance(post, dict): post["summary"] = "Handle A2A JSON-RPC Requests" - post["description"] = _build_jsonrpc_extension_openapi_description() - post["x-a2a-extension-contracts"] = { - "session_binding": { - "extension_uri": SESSION_BINDING_EXTENSION_URI, - **build_public_session_binding_extension_params(session_binding), - }, - "model_selection": { - "extension_uri": MODEL_SELECTION_EXTENSION_URI, - **select_public_extension_params( - model_selection, - keys=( - "metadata_field", - "behavior", - "applies_to_methods", - "supported_metadata", - "provider_private_metadata", - "fields", - ), - ), - }, - "streaming": { - "extension_uri": STREAMING_EXTENSION_URI, - **build_public_streaming_extension_params(streaming), - }, - "interrupt_callback": { - "extension_uri": INTERRUPT_CALLBACK_EXTENSION_URI, - **select_public_extension_params( - build_public_interrupt_callback_extension_params( - interrupt_callback + post["description"] = _build_jsonrpc_extension_openapi_description( + capability_snapshot + ) + if capability_snapshot.extensions_enabled: + post["x-a2a-extension-contracts"] = { + "session_binding": { + "extension_uri": SESSION_BINDING_EXTENSION_URI, + **build_public_session_binding_extension_params(session_binding), + }, + "model_selection": { + "extension_uri": MODEL_SELECTION_EXTENSION_URI, + **select_public_extension_params( + model_selection, + keys=( + "metadata_field", + "behavior", + "applies_to_methods", + "supported_metadata", + "provider_private_metadata", + "fields", + ), ), - keys=( - "methods", - "supported_interrupt_events", - "request_id_field", - "interrupt_metadata_field", - "interrupt_fields", + }, + "streaming": { + "extension_uri": STREAMING_EXTENSION_URI, + **build_public_streaming_extension_params(streaming), + }, + "interrupt_callback": { + "extension_uri": INTERRUPT_CALLBACK_EXTENSION_URI, + **select_public_extension_params( + build_public_interrupt_callback_extension_params( + interrupt_callback + ), + keys=( + "methods", + "supported_interrupt_events", + "request_id_field", + "interrupt_metadata_field", + "interrupt_fields", + ), ), - ), - }, - } - if settings.opencode_api_version == "v2": - post["description"] = ( - "OpenCode V2 session transport is experimental. A2A prompt " - "execution and extension methods are not implemented." - ) + }, + } + else: post.pop("x-a2a-extension-contracts", None) request_body = post.setdefault("requestBody", {}) @@ -330,7 +345,10 @@ def custom_openapi() -> dict[str, Any]: app_json = content.setdefault("application/json", {}) if isinstance(app_json, dict): app_json["schema"] = {"$ref": "#/components/schemas/A2ARequest"} - app_json["examples"] = _build_jsonrpc_extension_openapi_examples() + if capability_snapshot.extensions_enabled: + app_json["examples"] = ( + _build_jsonrpc_extension_openapi_examples() + ) rest_post_contracts: dict[str, dict[str, Any]] = { "/message:send": { @@ -352,6 +370,14 @@ def custom_openapi() -> dict[str, Any]: "schema_ref": "#/components/schemas/SendStreamingMessageRequest", }, } + if not capability_snapshot.a2a_message_execution_enabled: + v2_message_description = ( + "OpenCode API V2 session transport is experimental. This A2A message " + "endpoint fails before any upstream request; message execution is not " + "implemented." + ) + for contract in rest_post_contracts.values(): + contract["description"] = v2_message_description rest_examples = _build_rest_message_openapi_examples() for rest_path, contract in rest_post_contracts.items(): rest_path_item = paths.get(rest_path) @@ -374,7 +400,8 @@ def custom_openapi() -> dict[str, Any]: if not isinstance(app_json, dict): continue app_json["schema"] = {"$ref": contract["schema_ref"]} - app_json["examples"] = rest_examples + if capability_snapshot.a2a_message_execution_enabled: + app_json["examples"] = rest_examples app.openapi_schema = schema return schema diff --git a/tests/contracts/test_extension_contract_consistency.py b/tests/contracts/test_extension_contract_consistency.py index a2ed109c..e2761726 100644 --- a/tests/contracts/test_extension_contract_consistency.py +++ b/tests/contracts/test_extension_contract_consistency.py @@ -37,6 +37,7 @@ build_streaming_extension_params, build_wire_contract_params, build_workspace_control_extension_params, + catalog, select_public_extension_params, ) from opencode_a2a.jsonrpc.methods import SESSION_CONTEXT_PREFIX @@ -408,6 +409,101 @@ async def test_runtime_supported_methods_align_with_capability_snapshot( assert error["data"]["supportedMethods"] == wire_contract["all_jsonrpc_methods"] +def test_v1_capability_snapshot_preserves_published_method_order_and_is_immutable() -> None: + runtime_profile = build_runtime_profile(make_settings(test_bearer_token="test-token")) + snapshot = build_capability_snapshot(runtime_profile=runtime_profile) + + assert snapshot.upstream_api_version == "v1" + assert snapshot.supported_jsonrpc_methods() == [ + "SendMessage", + "SendStreamingMessage", + "GetTask", + "ListTasks", + "CancelTask", + "CreateTaskPushNotificationConfig", + "GetTaskPushNotificationConfig", + "ListTaskPushNotificationConfigs", + "DeleteTaskPushNotificationConfig", + "SubscribeToTask", + "GetExtendedAgentCard", + "opencode.sessions.status", + "opencode.sessions.list", + "opencode.sessions.messages.list", + "opencode.sessions.get", + "opencode.sessions.children", + "opencode.sessions.todo", + "opencode.sessions.diff", + "opencode.sessions.messages.get", + "opencode.sessions.prompt_async", + "opencode.sessions.command", + "opencode.sessions.fork", + "opencode.sessions.share", + "opencode.sessions.unshare", + "opencode.sessions.summarize", + "opencode.sessions.revert", + "opencode.sessions.unrevert", + "opencode.providers.list", + "opencode.models.list", + "opencode.projects.list", + "opencode.projects.current", + "opencode.workspaces.list", + "opencode.worktrees.list", + "opencode.permissions.list", + "opencode.questions.list", + "a2a.interrupt.permission.reply", + "a2a.interrupt.question.reply", + "a2a.interrupt.question.reject", + ] + with pytest.raises(TypeError): + snapshot.conditional_methods["opencode.sessions.shell"] = object() + + +def test_v2_capability_snapshot_disables_every_catalogued_extension() -> None: + runtime_profile = build_runtime_profile(make_settings(test_bearer_token="test-token")) + snapshot = build_capability_snapshot( + runtime_profile=runtime_profile, + upstream_api_version="v2", + ) + extension_methods = { + *catalog.SESSION_METHODS.values(), + *catalog.PROVIDER_DISCOVERY_METHODS.values(), + *catalog.WORKSPACE_CONTROL_METHODS.values(), + *catalog.INTERRUPT_RECOVERY_METHODS.values(), + *catalog.INTERRUPT_CALLBACK_METHODS.values(), + } + + assert snapshot.supported_jsonrpc_methods() == list(catalog.CORE_JSONRPC_METHODS) + assert snapshot.extension_jsonrpc_methods() == [] + assert snapshot.declared_extension_uris == () + assert set(snapshot.dispatch_method_bindings().values()) == extension_methods - set( + snapshot.conditional_methods + ) + assert set(snapshot.conditional_methods) == { + catalog.SESSION_CONTROL_METHODS["shell"], + *catalog.WORKSPACE_MUTATION_METHODS.values(), + } + assert all(not binding.enabled for binding in snapshot.conditional_methods.values()) + wire_contract = build_wire_contract_params( + protocol_version=A2A_PROTOCOL_VERSION, + runtime_profile=runtime_profile, + capability_snapshot=snapshot, + ) + compatibility_profile = build_compatibility_profile_params( + protocol_version=A2A_PROTOCOL_VERSION, + runtime_profile=runtime_profile, + capability_snapshot=snapshot, + ) + assert wire_contract["extensions"] == { + "jsonrpc_methods": [], + "conditionally_available_methods": {}, + "extension_uris": [], + } + assert wire_contract["all_jsonrpc_methods"] == list(catalog.CORE_JSONRPC_METHODS) + assert compatibility_profile["extension_retention"] == {} + assert set(compatibility_profile["method_retention"]) == set(catalog.CORE_JSONRPC_METHODS) + assert all(not snapshot.is_method_enabled(method) for method in extension_methods) + + @pytest.mark.asyncio @pytest.mark.parametrize( ("method", "params", "interrupt_type"), diff --git a/tests/jsonrpc/test_jsonrpc_unsupported_method.py b/tests/jsonrpc/test_jsonrpc_unsupported_method.py index 75850183..467e1a5c 100644 --- a/tests/jsonrpc/test_jsonrpc_unsupported_method.py +++ b/tests/jsonrpc/test_jsonrpc_unsupported_method.py @@ -1,12 +1,20 @@ import httpx import pytest -from opencode_a2a.contracts.extensions import SESSION_BINDING_EXTENSION_URI +from opencode_a2a.contracts.extensions import SESSION_BINDING_EXTENSION_URI, catalog from opencode_a2a.protocol_versions import A2A_PROTOCOL_VERSION from opencode_a2a.server.application import create_app from opencode_a2a.upstream.v2 import V2UpstreamAdapter from tests.support.settings import make_settings +_V2_EXTENSION_METHODS = ( + *catalog.SESSION_METHODS.values(), + *catalog.PROVIDER_DISCOVERY_METHODS.values(), + *catalog.WORKSPACE_CONTROL_METHODS.values(), + *catalog.INTERRUPT_RECOVERY_METHODS.values(), + *catalog.INTERRUPT_CALLBACK_METHODS.values(), +) + @pytest.mark.asyncio async def test_unsupported_method_returns_unified_error() -> None: @@ -43,10 +51,34 @@ async def test_v2_rejects_unimplemented_opencode_extension_methods() -> None: app = create_app(make_settings(opencode_api_version="v2")) assert isinstance(app.state.upstream_client, V2UpstreamAdapter) openapi = app.openapi() - assert "x-a2a-extension-contracts" not in openapi["paths"]["/"]["post"] + jsonrpc_contract = openapi["paths"]["/"]["post"] + assert "x-a2a-extension-contracts" not in jsonrpc_contract + assert "fail before any upstream request" in jsonrpc_contract["description"] + assert "examples" not in jsonrpc_contract["requestBody"]["content"]["application/json"] + for path in ("/message:send", "/message:stream"): + rest_contract = openapi["paths"][path]["post"] + assert "fails before any upstream request" in rest_contract["description"] + assert "examples" not in rest_contract["requestBody"]["content"]["application/json"] transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + extended_card = await client.get( + "/extendedAgentCard", + headers={"Authorization": "Bearer test-token"}, + ) + rpc_extended_card = await client.post( + "/", + headers={ + "Authorization": "Bearer test-token", + "A2A-Extensions": SESSION_BINDING_EXTENSION_URI, + }, + json={ + "jsonrpc": "2.0", + "id": 122, + "method": "GetExtendedAgentCard", + "params": {}, + }, + ) response = await client.post( "/", headers={ @@ -61,6 +93,14 @@ async def test_v2_rejects_unimplemented_opencode_extension_methods() -> None: }, ) + assert extended_card.status_code == 200 + assert not extended_card.json()["capabilities"].get("extensions") + assert [skill["id"] for skill in extended_card.json()["skills"]] == [ + "a2a-execution-unavailable" + ] + assert rpc_extended_card.status_code == 200 + assert "A2A-Extensions" not in rpc_extended_card.headers + assert not rpc_extended_card.json()["result"]["capabilities"].get("extensions") assert response.status_code == 200 body = response.json() assert body["error"]["code"] == -32601 @@ -70,6 +110,94 @@ async def test_v2_rejects_unimplemented_opencode_extension_methods() -> None: assert "A2A-Extensions" not in response.headers +@pytest.mark.asyncio +@pytest.mark.parametrize("method", _V2_EXTENSION_METHODS) +async def test_v2_rejects_every_catalogued_extension_before_upstream_http( + method: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + app = create_app(make_settings(opencode_api_version="v2")) + upstream_calls: list[tuple[str, str]] = [] + + async def record_upstream_call(path: str, **_kwargs) -> None: + upstream_calls.append((method, path)) + raise AssertionError("V2 extension dispatch must not make an upstream HTTP request") + + monkeypatch.setattr(app.state.upstream_client, "_get_json", record_upstream_call) + monkeypatch.setattr(app.state.upstream_client, "_post_json", record_upstream_call) + transport = httpx.ASGITransport(app=app) + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.post( + "/", + headers={ + "Authorization": "Bearer test-token", + "A2A-Extensions": SESSION_BINDING_EXTENSION_URI, + }, + json={"jsonrpc": "2.0", "id": 123, "method": method, "params": {}}, + ) + + assert response.status_code == 200 + assert response.json()["error"]["code"] == -32601 + assert response.json()["error"]["message"] == "Method not found" + assert method not in response.json()["error"]["data"]["supportedMethods"] + assert response.json()["error"]["data"]["supportedMethods"] == list( + catalog.CORE_JSONRPC_METHODS + ) + assert upstream_calls == [] + + +@pytest.mark.asyncio +async def test_v2_core_message_paths_fail_before_upstream_http( + monkeypatch: pytest.MonkeyPatch, +) -> None: + app = create_app(make_settings(opencode_api_version="v2")) + upstream_calls: list[tuple[str, str]] = [] + + async def record_upstream_call(path: str, **_kwargs) -> None: + upstream_calls.append(("message", path)) + raise AssertionError("V2 message execution must not make an upstream HTTP request") + + monkeypatch.setattr(app.state.upstream_client, "_get_json", record_upstream_call) + monkeypatch.setattr(app.state.upstream_client, "_post_json", record_upstream_call) + transport = httpx.ASGITransport(app=app) + message = { + "messageId": "msg-v2-failfast", + "role": "ROLE_USER", + "parts": [{"text": "hello"}], + } + + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + send_message = await client.post( + "/", + headers={"Authorization": "Bearer test-token"}, + json={ + "jsonrpc": "2.0", + "id": 124, + "method": "SendMessage", + "params": {"message": message}, + }, + ) + send_streaming = await client.post( + "/", + headers={"Authorization": "Bearer test-token"}, + json={ + "jsonrpc": "2.0", + "id": 125, + "method": "SendStreamingMessage", + "params": {"message": message}, + }, + ) + + assert send_message.status_code == 200 + failed_task = send_message.json()["result"]["task"] + assert failed_task["status"]["state"] == "TASK_STATE_FAILED" + assert "No prompt was submitted" in failed_task["status"]["message"]["parts"][0]["text"] + assert send_streaming.status_code == 200 + assert send_streaming.json()["error"]["message"] == "Streaming is not supported by the agent" + assert upstream_calls == [] + + @pytest.mark.asyncio async def test_unsupported_method_uses_requested_protocol_version() -> None: settings = make_settings(test_bearer_token="test-token") diff --git a/tests/server/test_agent_card.py b/tests/server/test_agent_card.py index 8de60e0e..6ed2a94d 100644 --- a/tests/server/test_agent_card.py +++ b/tests/server/test_agent_card.py @@ -69,12 +69,22 @@ def test_v2_agent_card_does_not_advertise_unimplemented_chat_or_extensions() -> assert "Only internal experimental OpenCode V2 session transport exists" in card.description assert ( - "A2A message execution and OpenCode-specific extensions are unsupported" in card.description + "A2A SendMessage and SendStreamingMessage fail fast before any upstream request" + in card.description ) + assert "default input/output modes are required SDK schema fields" in card.description assert card.capabilities.streaming is False assert extended_card.capabilities.streaming is False assert card.capabilities.extensions == [] assert extended_card.capabilities.extensions == [] + assert [ + (item.protocol_binding, item.protocol_version) for item in card.supported_interfaces + ] == [ + ("HTTP+JSON", "1.0"), + ("JSONRPC", "1.0"), + ] + assert card.default_input_modes == ["text/plain", "application/octet-stream"] + assert card.default_output_modes == ["text/plain", "application/json"] for v2_card in (card, extended_card): assert [skill.id for skill in v2_card.skills] == ["a2a-execution-unavailable"] skill = v2_card.skills[0] diff --git a/tests/server/test_app_behaviors.py b/tests/server/test_app_behaviors.py index 2a2fe438..d2d70a23 100644 --- a/tests/server/test_app_behaviors.py +++ b/tests/server/test_app_behaviors.py @@ -253,6 +253,7 @@ def test_agent_card_helper_builders_cover_optional_branches() -> None: public_description = _build_agent_card_description( settings, runtime_profile, + capability_snapshot=build_capability_snapshot(runtime_profile=runtime_profile), include_detailed_contracts=False, ) assert "Deployment project: alpha." in public_description @@ -262,6 +263,7 @@ def test_agent_card_helper_builders_cover_optional_branches() -> None: extended_description = _build_agent_card_description( settings, runtime_profile, + capability_snapshot=build_capability_snapshot(runtime_profile=runtime_profile), include_detailed_contracts=True, ) assert "Deployment project: alpha." in extended_description From 7845d0ede672c7a80bd2a7a4981abb96b0b299b0 Mon Sep 17 00:00:00 2001 From: Tom Brandenburg Date: Sun, 25 Oct 2026 15:53:46 +0100 Subject: [PATCH 8/9] test(compatibility): retain session-slice fail-closed wording (#6) --- tests/jsonrpc/test_jsonrpc_unsupported_method.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/jsonrpc/test_jsonrpc_unsupported_method.py b/tests/jsonrpc/test_jsonrpc_unsupported_method.py index 467e1a5c..a7218541 100644 --- a/tests/jsonrpc/test_jsonrpc_unsupported_method.py +++ b/tests/jsonrpc/test_jsonrpc_unsupported_method.py @@ -192,7 +192,9 @@ async def record_upstream_call(path: str, **_kwargs) -> None: assert send_message.status_code == 200 failed_task = send_message.json()["result"]["task"] assert failed_task["status"]["state"] == "TASK_STATE_FAILED" - assert "No prompt was submitted" in failed_task["status"]["message"]["parts"][0]["text"] + assert "OpenCode V2 prompt execution is not implemented" in ( + failed_task["status"]["message"]["parts"][0]["text"] + ) assert send_streaming.status_code == 200 assert send_streaming.json()["error"]["message"] == "Streaming is not supported by the agent" assert upstream_calls == [] From cc8470d7b631b00e580de741324d5956ef673780 Mon Sep 17 00:00:00 2001 From: Tom Brandenburg Date: Sun, 25 Oct 2026 15:54:05 +0100 Subject: [PATCH 9/9] style(tests): format session capability assertion --- tests/jsonrpc/test_jsonrpc_unsupported_method.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/jsonrpc/test_jsonrpc_unsupported_method.py b/tests/jsonrpc/test_jsonrpc_unsupported_method.py index a7218541..739749c1 100644 --- a/tests/jsonrpc/test_jsonrpc_unsupported_method.py +++ b/tests/jsonrpc/test_jsonrpc_unsupported_method.py @@ -192,8 +192,9 @@ async def record_upstream_call(path: str, **_kwargs) -> None: assert send_message.status_code == 200 failed_task = send_message.json()["result"]["task"] assert failed_task["status"]["state"] == "TASK_STATE_FAILED" - assert "OpenCode V2 prompt execution is not implemented" in ( - failed_task["status"]["message"]["parts"][0]["text"] + assert ( + "OpenCode V2 prompt execution is not implemented" + in (failed_task["status"]["message"]["parts"][0]["text"]) ) assert send_streaming.status_code == 200 assert send_streaming.json()["error"]["message"] == "Streaming is not supported by the agent"