temp/robot/helper_a2a_jsonrpc_wire_format.py

"""Helper script for a2a_jsonrpc_wire_format.robot integration tests.

Each subcommand is a self-contained check that prints a sentinel on success
and exits with code 1 on failure.
"""

from __future__ import annotations

import sys
from pathlib import Path

# Ensure local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from cleveragents.a2a.facade import A2aLocalFacade  # noqa: E402
from cleveragents.a2a.models import (  # noqa: E402
    A2aErrorDetail,
    A2aRequest,
    A2aResponse,
)

# Non-standard field names that must NOT appear in wire output
_BANNED_REQUEST_FIELDS = {"a2a_version", "request_id", "operation", "auth"}
_BANNED_RESPONSE_FIELDS = {"a2a_version", "request_id", "status", "data", "timing_ms"}

# Required JSON-RPC 2.0 field names
_REQUIRED_REQUEST_FIELDS = {"jsonrpc", "id", "method", "params"}
_REQUIRED_SUCCESS_RESPONSE_FIELDS = {"jsonrpc", "id", "result"}
_REQUIRED_ERROR_RESPONSE_FIELDS = {"jsonrpc", "id", "error"}


def request_wire_format() -> None:
    """Verify A2aRequest serialises to JSON-RPC 2.0 wire format."""
    req = A2aRequest(method="_cleveragents/plan/status", params={"plan_id": "P1"})
    wire = req.model_dump()

    # Check required fields
    for field in _REQUIRED_REQUEST_FIELDS:
        if field not in wire:
            print(
                f"FAIL: missing required field '{field}' in request wire format",
                file=sys.stderr,
            )
            print(f"  Got fields: {list(wire.keys())}", file=sys.stderr)
            sys.exit(1)

    # Check jsonrpc value
    if wire["jsonrpc"] != "2.0":
        print(
            f"FAIL: jsonrpc should be '2.0', got '{wire['jsonrpc']}'",
            file=sys.stderr,
        )
        sys.exit(1)

    # Check method value
    if wire["method"] != "_cleveragents/plan/status":
        print(f"FAIL: method mismatch: {wire['method']}", file=sys.stderr)
        sys.exit(1)

    # Check banned fields are absent
    for field in _BANNED_REQUEST_FIELDS:
        if field in wire:
            print(
                f"FAIL: non-standard field '{field}' present in request wire format",
                file=sys.stderr,
            )
            sys.exit(1)

    print("a2a-request-wire-format-ok")


def response_success_wire_format() -> None:
    """Verify A2aResponse (success) serialises to JSON-RPC 2.0 wire format."""
    resp = A2aResponse(id="REQ-001", result={"plan_id": "P1", "phase": "strategize"})
    wire = resp.model_dump(exclude_none=True)

    # Check required fields
    for field in _REQUIRED_SUCCESS_RESPONSE_FIELDS:
        if field not in wire:
            print(
                f"FAIL: missing required field '{field}'"
                " in success response wire format",
                file=sys.stderr,
            )
            print(f"  Got fields: {list(wire.keys())}", file=sys.stderr)
            sys.exit(1)

    # Check jsonrpc value
    if wire["jsonrpc"] != "2.0":
        print(
            f"FAIL: jsonrpc should be '2.0', got '{wire['jsonrpc']}'",
            file=sys.stderr,
        )
        sys.exit(1)

    # Check id value
    if wire["id"] != "REQ-001":
        print(f"FAIL: id mismatch: {wire['id']}", file=sys.stderr)
        sys.exit(1)

    # Check result contains expected data
    if "plan_id" not in wire["result"]:
        print(f"FAIL: result missing plan_id: {wire['result']}", file=sys.stderr)
        sys.exit(1)

    # Check banned fields are absent
    for field in _BANNED_RESPONSE_FIELDS:
        if field in wire:
            print(
                f"FAIL: non-standard field '{field}'"
                " present in success response wire format",
                file=sys.stderr,
            )
            sys.exit(1)

    # Check error is absent in success response
    if "error" in wire:
        print(
            f"FAIL: 'error' field present in success response: {wire['error']}",
            file=sys.stderr,
        )
        sys.exit(1)

    print("a2a-response-success-wire-format-ok")


def response_error_wire_format() -> None:
    """Verify A2aResponse (error) serialises to JSON-RPC 2.0 wire format."""
    resp = A2aResponse(
        id="REQ-002",
        error=A2aErrorDetail(code=-32001, message="Plan not found"),
    )
    wire = resp.model_dump(exclude_none=True)

    # Check required fields
    for field in _REQUIRED_ERROR_RESPONSE_FIELDS:
        if field not in wire:
            print(
                f"FAIL: missing required field '{field}' in error response wire format",
                file=sys.stderr,
            )
            print(f"  Got fields: {list(wire.keys())}", file=sys.stderr)
            sys.exit(1)

    # Check jsonrpc value
    if wire["jsonrpc"] != "2.0":
        print(
            f"FAIL: jsonrpc should be '2.0', got '{wire['jsonrpc']}'",
            file=sys.stderr,
        )
        sys.exit(1)

    # Check error structure
    error = wire["error"]
    if error.get("code") != -32001:
        print(f"FAIL: error code mismatch: {error.get('code')}", file=sys.stderr)
        sys.exit(1)

    # Check banned fields are absent
    for field in _BANNED_RESPONSE_FIELDS:
        if field in wire:
            print(
                f"FAIL: non-standard field '{field}'"
                " present in error response wire format",
                file=sys.stderr,
            )
            sys.exit(1)

    # Check result is absent in error response
    if "result" in wire:
        print(
            f"FAIL: 'result' field present in error response: {wire['result']}",
            file=sys.stderr,
        )
        sys.exit(1)

    print("a2a-response-error-wire-format-ok")


def facade_dispatch_jsonrpc() -> None:
    """Verify end-to-end: facade.dispatch() returns JSON-RPC 2.0 compliant response."""
    facade = A2aLocalFacade()
    req = A2aRequest(method="_cleveragents/health/check", id="test-req-001", params={})
    resp = facade.dispatch(req)

    # Check jsonrpc field
    if resp.jsonrpc != "2.0":
        print(
            f"FAIL: response jsonrpc should be '2.0', got '{resp.jsonrpc}'",
            file=sys.stderr,
        )
        sys.exit(1)

    # Check id matches request
    if resp.id != "test-req-001":
        print(
            f"FAIL: response id should be 'test-req-001', got '{resp.id}'",
            file=sys.stderr,
        )
        sys.exit(1)

    # Check result is set (success path)
    if resp.result is None:
        print(
            f"FAIL: response result should be set, got None. Error: {resp.error}",
            file=sys.stderr,
        )
        sys.exit(1)

    # Check error is None (success path)
    if resp.error is not None:
        print(
            f"FAIL: response error should be None, got: {resp.error}",
            file=sys.stderr,
        )
        sys.exit(1)

    # Verify no non-standard fields on the model
    wire = resp.model_dump(exclude_none=True)
    for field in _BANNED_RESPONSE_FIELDS:
        if field in wire:
            print(
                f"FAIL: non-standard field '{field}' in facade response wire format",
                file=sys.stderr,
            )
            sys.exit(1)

    print("a2a-facade-dispatch-jsonrpc-ok")


def request_deserialise() -> None:
    """Verify A2aRequest can be constructed from a JSON-RPC 2.0 dict."""
    payload = {
        "jsonrpc": "2.0",
        "id": "inbound-42",
        "method": "message/send",
        "params": {"content": "hello"},
    }
    req = A2aRequest.model_validate(payload)

    if req.method != "message/send":
        print(f"FAIL: method mismatch: {req.method}", file=sys.stderr)
        sys.exit(1)
    if req.id != "inbound-42":
        print(f"FAIL: id mismatch: {req.id}", file=sys.stderr)
        sys.exit(1)
    if req.jsonrpc != "2.0":
        print(f"FAIL: jsonrpc mismatch: {req.jsonrpc}", file=sys.stderr)
        sys.exit(1)
    if req.params.get("content") != "hello":
        print(f"FAIL: params mismatch: {req.params}", file=sys.stderr)
        sys.exit(1)

    print("a2a-request-deserialise-ok")


def response_deserialise() -> None:
    """Verify A2aResponse can be constructed from a JSON-RPC 2.0 dict."""
    # Success response
    success_payload = {
        "jsonrpc": "2.0",
        "id": "resp-42",
        "result": {"status": "accepted"},
    }
    resp = A2aResponse.model_validate(success_payload)
    if resp.result is None or resp.result.get("status") != "accepted":
        print(f"FAIL: result mismatch: {resp.result}", file=sys.stderr)
        sys.exit(1)
    if resp.error is not None:
        print(f"FAIL: error should be None: {resp.error}", file=sys.stderr)
        sys.exit(1)

    # Error response
    error_payload = {
        "jsonrpc": "2.0",
        "id": "resp-43",
        "error": {"code": -32001, "message": "Plan not found"},
    }
    err_resp = A2aResponse.model_validate(error_payload)
    if err_resp.error is None:
        print("FAIL: error should be set", file=sys.stderr)
        sys.exit(1)
    if err_resp.result is not None:
        print(f"FAIL: result should be None: {err_resp.result}", file=sys.stderr)
        sys.exit(1)

    print("a2a-response-deserialise-ok")


def request_rejects_old_fields() -> None:
    """Verify A2aRequest no longer has old non-standard field names."""
    req = A2aRequest(method="_cleveragents/plan/status", params={"plan_id": "P1"})

    # Verify old attributes don't exist on the model
    old_attrs = ["a2a_version", "request_id", "operation", "auth"]
    for attr in old_attrs:
        if hasattr(req, attr):
            print(
                f"FAIL: A2aRequest still has old attribute '{attr}'",
                file=sys.stderr,
            )
            sys.exit(1)

    # Verify new attributes exist
    new_attrs = ["jsonrpc", "id", "method", "params"]
    for attr in new_attrs:
        if not hasattr(req, attr):
            print(
                f"FAIL: A2aRequest missing new attribute '{attr}'",
                file=sys.stderr,
            )
            sys.exit(1)

    print("a2a-request-rejects-old-fields-ok")


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

_COMMANDS = {
    "request-wire-format": request_wire_format,
    "response-success-wire-format": response_success_wire_format,
    "response-error-wire-format": response_error_wire_format,
    "facade-dispatch-jsonrpc": facade_dispatch_jsonrpc,
    "request-deserialise": request_deserialise,
    "response-deserialise": response_deserialise,
    "request-rejects-old-fields": request_rejects_old_fields,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <command>", file=sys.stderr)
        print(f"Commands: {', '.join(_COMMANDS)}", file=sys.stderr)
        sys.exit(1)
    _COMMANDS[sys.argv[1]]()