chore: simplify redundant code and upgrade vulnerable dependencies

pyproject.toml

@@ -26,10 +26,10 @@ classifiers = [
 dependencies = [
     "a2a-sdk==1.0.2",
     "aiosqlite>=0.20,<1.0",
-    "fastapi>=0.110,<1.0",
+    "fastapi>=0.139,<1.0",
     "httpx>=0.27,<1.0",
     "pydantic>=2.6,<3.0",
-    "pydantic-settings>=2.2,<3.0",
+    "pydantic-settings>=2.14.2,<3.0",
     "protobuf>=6.33.5,<7.0",
     "sqlalchemy>=2.0,<3.0",
     "sse-starlette>=2.1,<4.0",

src/opencode_a2a/a2a_protocol.py

@@ -5,5 +5,4 @@
 
 AGENT_CARD_WELL_KNOWN_PATH = SDK_AGENT_CARD_WELL_KNOWN_PATH
 EXTENDED_AGENT_CARD_PATH = "/extendedAgentCard"
-EXTENDED_AGENT_CARD_PATHS = (EXTENDED_AGENT_CARD_PATH,)
 CORE_JSONRPC_METHODS = tuple(JsonRpcDispatcher.METHOD_TO_MODEL)

src/opencode_a2a/cli.py

@@ -103,7 +103,6 @@
 
 ROOT_HELP_EPILOG = "Command-specific help:\n  opencode-a2a serve --help\n  opencode-a2a call --help"
 SERVE_HELP_EPILOG = f"{OPENCODE_SETUP_HELP}\n\n{SERVE_ENVIRONMENT_HELP}"
-CALL_HELP_EPILOG = CALL_HELP
 
 
 class RootHelpFormatter(
@@ -264,7 +263,7 @@ def build_parser() -> argparse.ArgumentParser:
         help="Call an A2A agent.",
         description="Call an A2A agent using the A2A protocol. A2A Protocol 1.0 only.",
         formatter_class=argparse.RawDescriptionHelpFormatter,
-        epilog=CALL_HELP_EPILOG,
+        epilog=CALL_HELP,
     )
     call_parser.add_argument(
         "agent_url",

src/opencode_a2a/execution/coordinator.py

@@ -280,7 +280,7 @@ async def _bind_session(self) -> None:
         if self._prepared.streaming_request:
             self._stream_terminal_signal = asyncio.get_running_loop().create_future()
             self._stream_task = asyncio.create_task(
-                self._executor._consume_opencode_stream(
+                self._executor._stream_runtime.consume(
                     session_id=self._session_id,
                     identity=self._prepared.identity,
                     task_id=self._task_id,

src/opencode_a2a/execution/executor.py

@@ -46,12 +46,6 @@
 from .metrics import emit_metric
 from .session_manager import SessionManager
 from .stream_runtime import StreamRuntime
-from .stream_state import (
-    _StreamOutputState,
-)
-from .upstream_error_translator import (
-    _StreamTerminalSignal,
-)
 
 logger = logging.getLogger(__name__)
 _TEXT_PLAIN_MEDIA_TYPE = "text/plain"
@@ -497,40 +491,3 @@ def _should_stream(self, context: RequestContext) -> bool:
         # JSON-RPC transport sets method in call context state.
         method = call_context.state.get("method")
         return method == "SendStreamingMessage"
-
-    async def _consume_opencode_stream(
-        self,
-        *,
-        session_id: str,
-        identity: str,
-        task_id: str,
-        context_id: str,
-        artifact_id: str,
-        stream_state: _StreamOutputState,
-        event_queue: EventQueue,
-        stop_event: asyncio.Event,
-        terminal_signal: asyncio.Future[_StreamTerminalSignal],
-        directory: str | None = None,
-        workspace_id: str | None = None,
-        allow_structured_output: bool = True,
-        emit_session_metadata: bool = True,
-        emit_streaming_metadata: bool = True,
-        emit_interrupt_metadata: bool = True,
-    ) -> None:
-        await self._stream_runtime.consume(
-            session_id=session_id,
-            identity=identity,
-            task_id=task_id,
-            context_id=context_id,
-            artifact_id=artifact_id,
-            stream_state=stream_state,
-            event_queue=event_queue,
-            stop_event=stop_event,
-            terminal_signal=terminal_signal,
-            directory=directory,
-            workspace_id=workspace_id,
-            allow_structured_output=allow_structured_output,
-            emit_session_metadata=emit_session_metadata,
-            emit_streaming_metadata=emit_streaming_metadata,
-            emit_interrupt_metadata=emit_interrupt_metadata,
-        )

src/opencode_a2a/extension_negotiation.py

@@ -26,8 +26,7 @@
     WORKSPACE_CONTROL_METHODS,
 )
 
-_STREAMING_SHARED_METADATA_KEYS = frozenset({"stream", "progress", "usage"})
-_INTERRUPT_SHARED_METADATA_KEYS = frozenset({"interrupt"})
+_STREAMING_SHARED_METADATA_KEYS = ("stream", "progress", "usage")
 
 JSONRPC_EXTENSION_URI_BY_METHOD: dict[str, str] = {
     **{method: SESSION_MANAGEMENT_EXTENSION_URI for method in SESSION_METHODS.values()},
@@ -107,8 +106,7 @@ def _set_filtered_metadata(
             for key in _STREAMING_SHARED_METADATA_KEYS:
                 filtered_shared.pop(key, None)
         if INTERRUPT_CALLBACK_EXTENSION_URI not in requested_extensions:
-            for key in _INTERRUPT_SHARED_METADATA_KEYS:
-                filtered_shared.pop(key, None)
+            filtered_shared.pop("interrupt", None)
         if filtered_shared:
             filtered_metadata["shared"] = filtered_shared
         else:

src/opencode_a2a/output_modes.py

@@ -32,9 +32,6 @@
 _OPENCODE_METADATA_KEY = "opencode"
 _APPLICATION_JSON_MEDIA_TYPE = "application/json"
 _TEXT_PLAIN_MEDIA_TYPE = "text/plain"
-_STREAM_METADATA_SHARED_KEY = "shared"
-_STREAM_METADATA_STREAM_KEY = "stream"
-_STREAM_METADATA_BLOCK_TYPE_KEY = "block_type"
 
 
 def normalize_accepted_output_modes(source: Any) -> tuple[str, ...] | None:
@@ -208,13 +205,13 @@ def _extract_artifact_stream_block_type(artifact: Artifact | None) -> str | None
     metadata = _normalize_metadata_mapping(artifact.metadata)
     if not metadata:
         return None
-    shared = metadata.get(_STREAM_METADATA_SHARED_KEY)
+    shared = metadata.get("shared")
     if not isinstance(shared, dict):
         return None
-    stream = shared.get(_STREAM_METADATA_STREAM_KEY)
+    stream = shared.get("stream")
     if not isinstance(stream, dict):
         return None
-    block_type = stream.get(_STREAM_METADATA_BLOCK_TYPE_KEY)
+    block_type = stream.get("block_type")
     return block_type.strip() if isinstance(block_type, str) and block_type.strip() else None
 
 

src/opencode_a2a/server/middleware.py

@@ -14,7 +14,7 @@
 
 from ..a2a_protocol import (
     AGENT_CARD_WELL_KNOWN_PATH,
-    EXTENDED_AGENT_CARD_PATHS,
+    EXTENDED_AGENT_CARD_PATH,
 )
 from ..auth import (
     authenticate_static_credential,
@@ -289,7 +289,7 @@ def _resolve_activated_extensions(request: Request) -> tuple[str, ...]:
         path = request.url.path
         if path == AGENT_CARD_WELL_KNOWN_PATH:
             supported_extensions = PUBLIC_EXTENSION_URIS
-        elif path in EXTENDED_AGENT_CARD_PATHS:
+        elif path == EXTENDED_AGENT_CARD_PATH:
             supported_extensions = ALL_EXTENSION_URIS
         elif path == "/" or path.startswith("/v1/"):
             supported_extensions = PUBLIC_EXTENSION_URIS
@@ -313,7 +313,7 @@ async def cache_agent_card_responses(request: Request, call_next):
 
         path = request.url.path
         is_public_card = path == AGENT_CARD_WELL_KNOWN_PATH
-        is_extended_card = path in EXTENDED_AGENT_CARD_PATHS
+        is_extended_card = path == EXTENDED_AGENT_CARD_PATH
         if not is_public_card and not is_extended_card:
             return await call_next(request)
 

src/opencode_a2a/server/task_store_sdk_compat.py

@@ -71,13 +71,12 @@ async def validate_schema(self) -> None:
             self._task_store.engine.url.render_as_string(hide_password=True)
         )
         table_name = self._shape.task_model.__table__.name
-        required_indexes = frozenset({f"idx_{table_name}_owner_last_updated"})
         async with self._task_store.engine.begin() as conn:
             await conn.run_sync(
                 lambda sync_conn: _validate_sdk_task_table_schema(
                     sync_conn,
                     table_name=table_name,
-                    required_indexes=required_indexes,
+                    required_index=f"idx_{table_name}_owner_last_updated",
                     database_url=database_url,
                 )
             )
@@ -247,7 +246,7 @@ def _validate_sdk_task_table_schema(
     connection: Any,
     *,
     table_name: str,
-    required_indexes: frozenset[str],
+    required_index: str,
     database_url: str,
 ) -> None:
     inspector = inspect(connection)
@@ -257,7 +256,7 @@ def _validate_sdk_task_table_schema(
     existing_columns = {column["name"] for column in inspector.get_columns(table_name)}
     existing_indexes = {index["name"] for index in inspector.get_indexes(table_name)}
     missing_columns = sorted(_REQUIRED_SCHEMA_COLUMNS - existing_columns)
-    missing_indexes = sorted(required_indexes - existing_indexes)
+    missing_indexes = [] if required_index in existing_indexes else [required_index]
     if not missing_columns and not missing_indexes:
         return
 

tests/execution/test_metrics.py

@@ -18,7 +18,8 @@
     TaskStatus,
 )
 
-from opencode_a2a.execution.executor import OpencodeAgentExecutor, _StreamOutputState
+from opencode_a2a.execution.executor import OpencodeAgentExecutor
+from opencode_a2a.execution.stream_state import _StreamOutputState
 from opencode_a2a.server.application import OpencodeRequestHandler
 from tests.support.helpers import DummyEventQueue
 from tests.support.settings import make_settings
@@ -156,7 +157,7 @@ async def discard_interrupt_request(self, request_id: str) -> None:
     terminal_signal = asyncio.get_running_loop().create_future()
 
     with caplog.at_level(logging.DEBUG, logger="opencode_a2a.execution.executor"):
-        await executor._consume_opencode_stream(
+        await executor._stream_runtime.consume(
             session_id="ses-1",
             identity="user-1",
             task_id="task-1",
@@ -207,7 +208,7 @@ async def _fast_sleep(_seconds: float) -> None:
     terminal_signal = asyncio.get_running_loop().create_future()
 
     with caplog.at_level(logging.DEBUG, logger="opencode_a2a.execution.executor"):
-        await executor._consume_opencode_stream(
+        await executor._stream_runtime.consume(
             session_id="ses-1",
             identity="user-1",
             task_id="task-1",