feat(protocol): negotiate versions during initialization

CHANGELOG.md

@@ -6,6 +6,8 @@ Execution boundaries and auditable run telemetry.
 
 ### Added
 
+- Added `MAJOR.MINOR` protocol version negotiation to `agent/initialize`, independent of package versions.
+- Added initialize request/result schemas and recorded the negotiated protocol version in audit metadata.
 - Added a run-level wall-clock budget via `--max-duration` / `ECP_MAX_DURATION`, separate from the existing per-RPC `--timeout`.
 - Added a structured audit record covering run id, timestamps, manifest SHA-256 digest, agent metadata, configured limits, per-step latency and `exit_reason`, token usage, and pass/fail totals. Embedded under `audit` in the JSON report and writable standalone with `ecp run --audit-out`.
 - Added `schema/audit.schema.json` and referenced it from `report.schema.json`.
@@ -15,6 +17,7 @@ Execution boundaries and auditable run telemetry.
 
 ### Changed
 
+- Versionless agents remain supported as legacy protocol `0.1` with one warning per run; incompatible major versions stop execution with `VERSION_UNSUPPORTED` (`-32001`).
 - **Breaking (runtime behavior):** a timeout, crash, protocol violation, or JSON-RPC error no longer aborts the whole run. The failing step is recorded as failed, remaining steps in that scenario are marked skipped, and the next scenario proceeds with a fresh agent. Previously one hung agent killed the run and produced no report at all.
 - Steps that never produced a result now contribute a failed `execution` check, so a timed-out step can no longer be counted as a pass by CI.
 - `ECPRunner.run_scenarios()` returns two new keys, `exit_reason` and `audit`. Existing `passed` / `total` / `scenarios` keys are unchanged.

README.md

@@ -48,6 +48,8 @@ ECP is JSON-RPC 2.0 over stdio or Streamable HTTP. The runtime calls:
 - `agent/step`
 - `agent/reset`
 
+During `agent/initialize`, the runtime and agent negotiate an independent `MAJOR.MINOR` protocol version. Versionless agents remain supported as legacy protocol `0.1`.
+
 The agent returns:
 
 - `public_output` - user-visible answer

docs/spec.md

@@ -16,9 +16,35 @@ For Streamable HTTP, the agent runs as an HTTP server and exposes one endpoint,
 
 ### agent/initialize
 
-**Params**: `config` (object, optional)
+**Params**:
 
-**Result**: `{ name, capabilities }`
+- `protocol_version`: highest protocol version supported by the runtime, formatted as `MAJOR.MINOR`
+- `config`: optional configuration object
+
+**Result**: `{ name, protocol_version, capabilities }`
+
+Protocol versions are independent of runtime and SDK package versions. A major-version mismatch is incompatible and produces JSON-RPC error `-32001` (`VERSION_UNSUPPORTED`). When minor versions differ, the lower minor is selected. Agents that omit the field remain supported as legacy protocol `0.1`, with one warning per run.
+
+```json
+{
+  "jsonrpc": "2.0",
+  "id": 1,
+  "method": "agent/initialize",
+  "params": { "protocol_version": "1.0", "config": {} }
+}
+```
+
+```json
+{
+  "jsonrpc": "2.0",
+  "id": 1,
+  "result": {
+    "name": "SupportAgent",
+    "protocol_version": "1.0",
+    "capabilities": {}
+  }
+}
+```
 
 ### agent/step
 
@@ -132,6 +158,8 @@ Latency aggregates only steps that actually executed, so one timeout does not sk
 
 Machine-readable JSON Schemas live in `schema/`:
 
+- `schema/initialize-params.schema.json`
+- `schema/initialize-result.schema.json`
 - `schema/manifest.schema.json`
 - `schema/agent-result.schema.json`
 - `schema/tool-call.schema.json`

examples/protocol_demo.py

@@ -13,7 +13,7 @@
 import subprocess
 import sys
 import threading
-from typing import Dict, Any
+from typing import Any, Dict
 
 # We'll use the existing customer support demo agent as our server
 AGENT_CMD = [sys.executable, "examples/customer_support_demo/agent.py"]
@@ -70,7 +70,12 @@ def main() -> None:
 
     try:
         # 3. Send the Initialization message
-        send_rpc(process, "agent/initialize", {"config": {}}, msg_id=1)
+        send_rpc(
+            process,
+            "agent/initialize",
+            {"protocol_version": "1.0", "config": {}},
+            msg_id=1,
+        )
 
         # 4. Wait a moment, then send the Step message (the actual evaluation task)
         import time

runtime/python/README.md

@@ -32,6 +32,8 @@ ecp conformance --target "python examples/customer_support_demo/agent.py"
 ecp doctor
 ```
 
+The runtime negotiates protocol compatibility during `agent/initialize`. It sends protocol `1.0`, accepts lower minor versions within major `1`, and continues to support versionless agents as legacy protocol `0.1` with one warning per run.
+
 Manifest `target` values may be either a command for the default stdio transport or an ECP Streamable HTTP endpoint:
 
 ```yaml
@@ -79,4 +81,3 @@ $env:ECP_LLM_JUDGE_MODEL="gpt-4o-mini"
 - Documentation: https://evaluationcontextprotocol.io/
 - Repository: https://github.com/evaluation-context-protocol/ecp
 - Issues: https://github.com/evaluation-context-protocol/ecp/issues
-

runtime/python/src/ecp_runtime/cli.py

@@ -14,11 +14,12 @@
 from .conformance import (
     build_conformance_report,
     conformance_check,
-    validate_initialize_result,
+    validate_initialize_negotiation,
     validate_reset_result,
     validate_step_result,
 )
 from .manifest import ECPManifest
+from .protocol import initialize_params
 from .reporter import HTMLReporter
 from .runner import ECPRunner, resolve_rpc_timeout
 from .trend import RunTrendAnalyzer
@@ -335,8 +336,8 @@ def conformance(
             agent,
             "initialize response",
             "agent/initialize",
-            {"config": {}},
-            result_validator=validate_initialize_result,
+            initialize_params(),
+            result_validator=validate_initialize_negotiation,
         )
         checks.append(initialize)
         if initialize["passed"]:

runtime/python/src/ecp_runtime/conformance.py

@@ -2,6 +2,9 @@
 
 from typing import Any, Callable, Dict, List, Optional
 
+from .errors import ECPVersionUnsupported
+from .protocol import negotiate_protocol_version, parse_protocol_version
+
 VALID_STATUSES = {"done", "paused"}
 USAGE_FIELDS = ("input_tokens", "output_tokens", "total_tokens")
 
@@ -62,9 +65,24 @@ def validate_initialize_result(result: Any) -> Dict[str, Any]:
         raise ValueError("agent/initialize result name must be a non-empty string")
     if not isinstance(result.get("capabilities"), dict):
         raise ValueError("agent/initialize result capabilities must be an object")
+    if "protocol_version" in result:
+        parse_protocol_version(
+            result["protocol_version"],
+            field="agent/initialize result protocol_version",
+        )
     return result
 
 
+def validate_initialize_negotiation(result: Any) -> Dict[str, Any]:
+    """Validate initialization metadata and reject incompatible major versions."""
+    validated = validate_initialize_result(result)
+    try:
+        negotiate_protocol_version(validated.get("protocol_version"))
+    except ECPVersionUnsupported as exc:
+        raise ValueError(str(exc)) from exc
+    return validated
+
+
 def validate_reset_result(result: Any) -> bool:
     if result is not True:
         raise ValueError("agent/reset result must be true")

runtime/python/src/ecp_runtime/errors.py

@@ -34,6 +34,12 @@ class ECPProtocolError(ECPExecutionError):
     exit_reason = "protocol_error"
 
 
+class ECPVersionUnsupported(ECPProtocolError):
+    """The runtime and agent do not share a compatible protocol major version."""
+
+    code = -32001
+
+
 class ECPAgentError(ECPExecutionError):
     """The agent returned a well-formed JSON-RPC error response."""
 

runtime/python/src/ecp_runtime/protocol.py

@@ -0,0 +1,60 @@
+"""Wire-level protocol version negotiation for the ECP runtime."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Optional, Tuple
+
+from .errors import ECPVersionUnsupported
+
+PROTOCOL_VERSION = "1.0"
+LEGACY_PROTOCOL_VERSION = "0.1"
+VERSION_UNSUPPORTED_CODE = ECPVersionUnsupported.code
+
+_PROTOCOL_VERSION_PATTERN = re.compile(r"^(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)$")
+
+
+@dataclass(frozen=True)
+class ProtocolNegotiation:
+    """The version selected for a runtime-agent session."""
+
+    version: str
+    legacy: bool = False
+
+
+def parse_protocol_version(
+    value: object, *, field: str = "protocol_version"
+) -> Tuple[int, int]:
+    """Parse a ``MAJOR.MINOR`` version or raise a contract validation error."""
+    if not isinstance(value, str) or _PROTOCOL_VERSION_PATTERN.fullmatch(value) is None:
+        raise ValueError(f"{field} must be a MAJOR.MINOR string")
+    major, minor = value.split(".", 1)
+    return int(major), int(minor)
+
+
+def negotiate_protocol_version(agent_version: Optional[str]) -> ProtocolNegotiation:
+    """Select the common protocol version, preserving versionless legacy agents."""
+    if agent_version is None:
+        return ProtocolNegotiation(LEGACY_PROTOCOL_VERSION, legacy=True)
+
+    runtime_major, runtime_minor = parse_protocol_version(PROTOCOL_VERSION)
+    agent_major, agent_minor = parse_protocol_version(
+        agent_version,
+        field="agent/initialize result protocol_version",
+    )
+    if runtime_major != agent_major:
+        raise ECPVersionUnsupported(
+            f"VERSION_UNSUPPORTED ({VERSION_UNSUPPORTED_CODE}): "
+            f"runtime supports protocol {PROTOCOL_VERSION}, but the agent selected {agent_version}"
+        )
+
+    return ProtocolNegotiation(f"{runtime_major}.{min(runtime_minor, agent_minor)}")
+
+
+def initialize_params(config: Optional[dict] = None) -> dict:
+    """Build the normative ``agent/initialize`` request parameters."""
+    return {
+        "protocol_version": PROTOCOL_VERSION,
+        "config": {} if config is None else config,
+    }

runtime/python/src/ecp_runtime/pytest_plugin.py

@@ -1,8 +1,11 @@
 import os
+import warnings
 from typing import Any, Dict
 
 import pytest
 
+from .conformance import validate_initialize_result
+from .protocol import initialize_params, negotiate_protocol_version
 from .runner import _create_agent
 
 
@@ -23,8 +26,16 @@ def __init__(self, target: str, rpc_timeout: float = 30.0):
     def start(self):
         self._agent = _create_agent(self.target, self.rpc_timeout)
         self._agent.start()
-        resp = self._agent.send_rpc("agent/initialize", {"config": {}})
+        resp = self._agent.send_rpc("agent/initialize", initialize_params())
         self._ensure_rpc_success(resp, "agent/initialize")
+        result = validate_initialize_result(resp.get("result"))
+        negotiation = negotiate_protocol_version(result.get("protocol_version"))
+        if negotiation.legacy:
+            warnings.warn(
+                f"Agent omitted protocol_version; treating it as legacy protocol {negotiation.version}",
+                RuntimeWarning,
+                stacklevel=2,
+            )
         return self
 
     def stop(self):

runtime/python/src/ecp_runtime/runner.py

@@ -38,9 +38,15 @@
     ECPProtocolError,
     ECPTimeoutError,
     ECPTransportError,
+    ECPVersionUnsupported,
     exit_reason_for,
 )
 from .graders import evaluate_step
+from .protocol import (
+    VERSION_UNSUPPORTED_CODE,
+    initialize_params,
+    negotiate_protocol_version,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -278,16 +284,21 @@ def __init__(
         self.max_duration = resolve_max_duration(max_duration)
         self.manifest_path = manifest_path
         self.agent_info: Dict[str, Any] = {}
+        self._legacy_version_warned = False
 
     def run_scenarios(self):
+        # A runner can be reused by API consumers. Negotiation metadata and
+        # warning suppression are scoped to one run, not to the object lifetime.
+        self.agent_info = {}
+        self._legacy_version_warned = False
         started_at = audit_module.utc_now()
         run_start = time.perf_counter()
         deadline = run_start + self.max_duration if self.max_duration else None
 
         report_data: List[Dict[str, Any]] = []
         run_exit_reason = audit_module.STATUS_OK
 
-        for scenario in self.manifest.scenarios:
+        for scenario_index, scenario in enumerate(self.manifest.scenarios):
             if deadline is not None and time.perf_counter() >= deadline:
                 run_exit_reason = ECPBudgetExceeded.exit_reason
                 logger.error(
@@ -305,7 +316,20 @@ def run_scenarios(self):
                 continue
 
             logger.info("Scenario: %s", scenario.name)
-            record = self._run_scenario(scenario, deadline)
+            try:
+                record = self._run_scenario(scenario, deadline)
+            except ECPVersionUnsupported as exc:
+                message = str(exc)
+                logger.error("Run aborted before execution: %s", message)
+                report_data.append(
+                    self._skipped_scenario(scenario, ECPProtocolError.exit_reason, message)
+                )
+                for remaining in self.manifest.scenarios[scenario_index + 1 :]:
+                    report_data.append(
+                        self._skipped_scenario(remaining, ECPProtocolError.exit_reason, message)
+                    )
+                run_exit_reason = ECPProtocolError.exit_reason
+                break
             report_data.append(record)
 
             if record["exit_reason"] == ECPBudgetExceeded.exit_reason:
@@ -358,19 +382,38 @@ def _run_scenario(self, scenario, deadline: Optional[float]) -> Dict[str, Any]:
             agent = self._create_agent(self.manifest.target, rpc_timeout=self.rpc_timeout)
             agent.start()
             started = True
-            init_resp = self._call_rpc(agent, "agent/initialize", {"config": {}}, scenario.name, None)
+            init_resp = self._call_rpc(
+                agent,
+                "agent/initialize",
+                initialize_params(),
+                scenario.name,
+                None,
+            )
             init_result = init_resp["result"]
             try:
                 validate_initialize_result(init_result)
             except ValueError as exc:
                 raise ECPProtocolError(
                     f"Invalid agent/initialize result at scenario='{scenario.name}': {exc}"
                 ) from exc
+            negotiation = negotiate_protocol_version(init_result.get("protocol_version"))
+            if negotiation.legacy and not self._legacy_version_warned:
+                logger.warning(
+                    "Agent '%s' omitted protocol_version; treating it as legacy protocol %s",
+                    init_result.get("name"),
+                    negotiation.version,
+                )
+                self._legacy_version_warned = True
             if not self.agent_info:
                 self.agent_info = {
                     "name": init_result.get("name"),
+                    "protocol_version": negotiation.version,
                     "capabilities": init_result.get("capabilities", {}),
                 }
+        except ECPVersionUnsupported:
+            if started and agent is not None:
+                agent.stop()
+            raise
         except ECPExecutionError as exc:
             logger.error("Scenario '%s' could not start: %s", scenario.name, exc)
             if started and agent is not None:
@@ -537,6 +580,13 @@ def _ensure_rpc_success(
         except ValueError as exc:
             # A well-formed JSON-RPC error is the agent reporting failure;
             # anything else means the envelope itself broke the contract.
+            error_code = None
+            if isinstance(rpc_resp, dict) and isinstance(rpc_resp.get("error"), dict):
+                error_code = rpc_resp["error"].get("code")
+            if error_code == VERSION_UNSUPPORTED_CODE:
+                raise ECPVersionUnsupported(
+                    f"VERSION_UNSUPPORTED ({VERSION_UNSUPPORTED_CODE}) at {where}: {exc}"
+                ) from exc
             failure = ECPAgentError if isinstance(rpc_resp, dict) and "error" in rpc_resp else ECPProtocolError
             raise failure(f"RPC call failed ({method}) at {where}: {exc}") from exc