diff --git a/app/main.py b/app/main.py
index 28fc1a7..0a98b34 100644
--- a/app/main.py
+++ b/app/main.py
@@ -10,6 +10,7 @@
 from app.middleware.body_limit import BodySizeLimitMiddleware
 from app.middleware.rate_limit import RateLimitMiddleware
 from app.middleware.request_id import RequestIdMiddleware
+from app.middleware.verify_echo import VerifyEchoMiddleware
 from app.redis import close_redis
 from app.routers import agent_live_sessions, agents, alerts, auth_routes, cues, device_code, echo, events, executions, health, info, internal_users, messages, usage, webhook_secret, workers
 from app.utils.logging import setup_logging
@@ -70,6 +71,10 @@ async def lifespan(app: FastAPI):
     allow_headers=["*"],
     expose_headers=["X-Request-Id", "X-RateLimit-Limit", "X-RateLimit-Remaining", "X-RateLimit-Reset", "Retry-After", "X-CueAPI-Usage-Warning"],
 )
+# VerifyEcho sits inside rate-limit + body-limit so we don't process
+# oversized/throttled requests; outside SecurityHeaders (when added) so the
+# security-header sweep applies to the possibly-mutated response.
+app.add_middleware(VerifyEchoMiddleware)
 app.add_middleware(RateLimitMiddleware)
 app.add_middleware(BodySizeLimitMiddleware)
 app.add_middleware(RequestIdMiddleware)
diff --git a/app/middleware/verify_echo.py b/app/middleware/verify_echo.py
new file mode 100644
index 0000000..02985c3
--- /dev/null
+++ b/app/middleware/verify_echo.py
@@ -0,0 +1,213 @@
+"""Universal BodyVerify Layer 1.5 middleware.
+
+Phase 1 wired echo-back per-handler on POST /v1/messages + POST /v1/cues/<id>/fire.
+This middleware extends the same primitive to EVERY POST/PATCH/PUT endpoint with
+a JSON body, so the substrate-keystone protection isn't endpoint-by-endpoint.
+
+Design choice: raw ASGI middleware (not BaseHTTPMiddleware) so we can capture
+both the inbound request body bytes and the outbound response stream without
+fighting Starlette's buffering. Same shape as ``BodySizeLimitMiddleware``.
+
+Behavior:
+
+* No-op unless the request carries ``X-CueAPI-Verify-Echo: true`` (case-
+  insensitive, whitespace-stripped). Header absent → zero perf cost path.
+* Captures raw request body bytes BEFORE the handler runs; re-emits them via
+  the ``receive`` callable so handlers can read the body normally.
+* Captures response body bytes AFTER the handler returns; if the response is
+  application/json + dict-shaped + status 2xx, injects ``body_received``
+  (parsed from raw bytes) + ``body_received_sha256`` (SHA256 over canonical
+  JSON of the parsed body, or over the raw bytes for non-JSON payloads).
+* Idempotent — skips injection if response already has ``body_received``
+  (Phase 1 endpoints preserve their per-handler behavior; this middleware is
+  a coverage gap-filler, not a replacement).
+* Only acts on POST/PATCH/PUT (methods with bodies).
+"""
+from __future__ import annotations
+
+import hashlib
+import json as _json
+from typing import Any
+
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+
+VERIFY_ECHO_HEADER = b"x-cueapi-verify-echo"
+_BODY_METHODS = {"POST", "PATCH", "PUT"}
+
+
+def _canonical_json_bytes(value: Any) -> bytes:
+    return _json.dumps(
+        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
+    ).encode("utf-8")
+
+
+def _verify_echo_requested(headers: list[tuple[bytes, bytes]]) -> bool:
+    for k, v in headers:
+        if k.lower() == VERIFY_ECHO_HEADER:
+            return v.strip().lower() == b"true"
+    return False
+
+
+class VerifyEchoMiddleware:
+    """Raw ASGI middleware — universal echo-back primitive."""
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        method = scope.get("method", "").upper()
+        if method not in _BODY_METHODS:
+            await self.app(scope, receive, send)
+            return
+
+        if not _verify_echo_requested(scope.get("headers", [])):
+            await self.app(scope, receive, send)
+            return
+
+        # Capture the request body. We must replay it via receive so the
+        # downstream handler can still read it via `await request.body()`
+        # / Pydantic body parsing.
+        body_chunks: list[bytes] = []
+        more_body = True
+        while more_body:
+            message = await receive()
+            if message["type"] != "http.request":
+                # Non-body lifecycle messages (e.g. http.disconnect) — pass
+                # through to the app as-is. Practically rare on the body
+                # ingest path but handled defensively.
+                break
+            body_chunks.append(message.get("body", b"") or b"")
+            more_body = bool(message.get("more_body", False))
+        request_body_bytes = b"".join(body_chunks)
+
+        # Re-emit the captured body to the handler. After the body is fully
+        # sent (one chunk + more_body=False), subsequent receive() calls
+        # should hang waiting for the next request — typical ASGI pattern
+        # for after-body lifecycle messages — but Starlette doesn't poll
+        # after parsing, so emitting once is sufficient.
+        _replayed = {"sent": False}
+
+        async def receive_replay() -> Message:
+            if not _replayed["sent"]:
+                _replayed["sent"] = True
+                return {
+                    "type": "http.request",
+                    "body": request_body_bytes,
+                    "more_body": False,
+                }
+            # If asked again (rare), defer to the original receive — for
+            # disconnect events etc.
+            return await receive()
+
+        # Capture the response. We need to read the streaming body, decide
+        # whether to inject, and re-emit a possibly-modified body.
+        response_status: dict = {"code": 200}
+        response_headers: dict = {"items": []}
+        response_chunks: list[bytes] = []
+
+        async def send_capture(message: Message) -> None:
+            if message["type"] == "http.response.start":
+                response_status["code"] = message.get("status", 200)
+                response_headers["items"] = list(message.get("headers", []))
+                # Defer sending — we may need to mutate Content-Length.
+                return
+            if message["type"] == "http.response.body":
+                response_chunks.append(message.get("body", b"") or b"")
+                if message.get("more_body", False):
+                    return
+                # Final chunk — process now, then emit.
+                modified = self._maybe_inject(
+                    status_code=response_status["code"],
+                    headers=response_headers["items"],
+                    request_body_bytes=request_body_bytes,
+                    response_body_bytes=b"".join(response_chunks),
+                )
+                # Emit start with possibly-mutated headers
+                await send(
+                    {
+                        "type": "http.response.start",
+                        "status": response_status["code"],
+                        "headers": modified["headers"],
+                    }
+                )
+                await send(
+                    {
+                        "type": "http.response.body",
+                        "body": modified["body"],
+                        "more_body": False,
+                    }
+                )
+                return
+            # Pass through any other message types
+            await send(message)
+
+        await self.app(scope, receive_replay, send_capture)
+
+    @staticmethod
+    def _maybe_inject(
+        *,
+        status_code: int,
+        headers: list[tuple[bytes, bytes]],
+        request_body_bytes: bytes,
+        response_body_bytes: bytes,
+    ) -> dict:
+        """Decide whether to inject echo fields; return final {headers, body}."""
+        # Only inject on 2xx — error responses keep their shape so callers
+        # don't get noisy echo fields on validation errors etc.
+        if not (200 <= status_code < 300):
+            return {"headers": headers, "body": response_body_bytes}
+
+        # Inspect Content-Type — only JSON responses are candidates.
+        content_type = b""
+        for k, v in headers:
+            if k.lower() == b"content-type":
+                content_type = v.lower()
+                break
+        if b"application/json" not in content_type:
+            return {"headers": headers, "body": response_body_bytes}
+
+        # Parse the response. Skip if it isn't a JSON object (lists, scalars,
+        # etc. don't get echo fields).
+        try:
+            resp_dict = _json.loads(response_body_bytes)
+        except _json.JSONDecodeError:
+            return {"headers": headers, "body": response_body_bytes}
+        if not isinstance(resp_dict, dict):
+            return {"headers": headers, "body": response_body_bytes}
+
+        # Idempotent: respect existing body_received from Phase 1 handlers.
+        if "body_received" in resp_dict:
+            return {"headers": headers, "body": response_body_bytes}
+
+        # Parse the request body. Fall back to raw bytes if not valid JSON.
+        body_view: Any
+        sha_input: bytes
+        if not request_body_bytes:
+            body_view = None
+            sha_input = b""
+        else:
+            try:
+                body_view = _json.loads(request_body_bytes)
+                sha_input = _canonical_json_bytes(body_view)
+            except _json.JSONDecodeError:
+                body_view = request_body_bytes.decode("utf-8", errors="replace")
+                sha_input = request_body_bytes
+
+        resp_dict["body_received"] = body_view
+        resp_dict["body_received_sha256"] = hashlib.sha256(sha_input).hexdigest()
+
+        new_body = _json.dumps(resp_dict).encode("utf-8")
+
+        # Update Content-Length header (others preserved).
+        new_headers: list[tuple[bytes, bytes]] = []
+        for k, v in headers:
+            if k.lower() == b"content-length":
+                continue
+            new_headers.append((k, v))
+        new_headers.append((b"content-length", str(len(new_body)).encode("ascii")))
+        return {"headers": new_headers, "body": new_body}
diff --git a/parity-manifest.json b/parity-manifest.json
index 3927efc..a23dd13 100644
--- a/parity-manifest.json
+++ b/parity-manifest.json
@@ -87,14 +87,15 @@
       {"path": "app/auth.py", "private_counterpart": "app/auth.py", "last_synced": "2026-04-16"},
       {"path": "app/config.py", "private_counterpart": "app/config.py", "last_synced": "2026-04-16"},
       {"path": "app/database.py", "private_counterpart": "app/database.py", "last_synced": "2026-04-16"},
-      {"path": "app/main.py", "private_counterpart": "app/main.py", "last_synced": "2026-04-16"},
+      {"path": "app/main.py", "private_counterpart": "app/main.py", "last_synced": "2026-05-11", "ported_in": "bodyverify-layer-1-5-universal-middleware", "deviation": "BodyVerify port: VerifyEchoMiddleware registered between CORS/RequestId and RateLimit. Private has additional middlewares (SecurityHeaders, AccessAudit) hosted-only; ordering preserved relative to shared middlewares."},
       {"path": "app/redis.py", "private_counterpart": "app/redis.py", "last_synced": "2026-04-16"}
     ],
     "middleware": [
       {"path": "app/middleware/__init__.py", "private_counterpart": "app/middleware/__init__.py", "last_synced": "2026-04-16"},
       {"path": "app/middleware/body_limit.py", "private_counterpart": "app/middleware/body_limit.py", "last_synced": "2026-04-16"},
       {"path": "app/middleware/rate_limit.py", "private_counterpart": "app/middleware/rate_limit.py", "last_synced": "2026-04-16"},
-      {"path": "app/middleware/request_id.py", "private_counterpart": "app/middleware/request_id.py", "last_synced": "2026-04-16"}
+      {"path": "app/middleware/request_id.py", "private_counterpart": "app/middleware/request_id.py", "last_synced": "2026-04-16"},
+      {"path": "app/middleware/verify_echo.py", "private_counterpart": "app/middleware/verify_echo.py", "last_synced": "2026-05-11", "ported_in": "bodyverify-layer-1-5-universal-middleware"}
     ],
     "models": [
       {"path": "app/models/__init__.py", "private_counterpart": "app/models/__init__.py", "last_synced": "2026-04-16"},
diff --git a/tests/test_verify_echo_middleware.py b/tests/test_verify_echo_middleware.py
new file mode 100644
index 0000000..1864c70
--- /dev/null
+++ b/tests/test_verify_echo_middleware.py
@@ -0,0 +1,452 @@
+"""Tests for BodyVerify Layer 1.5 — universal middleware.
+
+Phase 1 wired echo-back per-handler on POST /v1/messages + POST /v1/cues/<id>/fire.
+This phase ships ``VerifyEchoMiddleware`` so the primitive applies to every
+POST/PATCH/PUT JSON endpoint without per-handler integration.
+
+Coverage targets:
+
+- ``VerifyEchoMiddleware`` happy path: header present + dict response → injects
+  ``body_received`` + ``body_received_sha256`` on at least 3 representative
+  endpoints (POST /v1/agents, POST /v1/cues, PATCH /v1/auth/me).
+- Backwards-compat: header absent → no echo fields. GET unaffected.
+- Method gating: header set on GET → no echo (only POST/PATCH/PUT).
+- Status gating: 4xx/5xx responses NOT echoed (validation errors stay clean).
+- Content-type gating: non-JSON 2xx responses NOT echoed (e.g. HTML pages).
+- Idempotency: Phase 1 endpoints (messages + fire) already echo; middleware
+  must NOT double-inject — existing ``body_received`` wins.
+- Raw body preservation: invalid JSON body still passes through but echo is
+  the raw string + hash of raw bytes.
+- 6 metachar classes on a representative non-Phase-1 endpoint (POST /v1/agents
+  via the display_name field).
+- Empty body handling.
+"""
+from __future__ import annotations
+
+import hashlib
+import uuid
+
+import pytest
+
+
+VERIFY_ECHO_HEADER_KEY = "X-CueAPI-Verify-Echo"
+
+
+# ───────────────────────────────────────────────────────────────────────
+# Backwards-compat — header absent path
+# ───────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_no_header_no_echo_on_agents_create(client, auth_headers):
+    """Default behavior: POST /v1/agents without header has no echo fields."""
+    r = await client.post(
+        "/v1/agents",
+        json={"display_name": f"NoEcho {uuid.uuid4().hex[:6]}", "metadata": {}},
+        headers=auth_headers,
+    )
+    assert r.status_code == 201
+    data = r.json()
+    assert "body_received" not in data
+    assert "body_received_sha256" not in data
+
+
+@pytest.mark.asyncio
+async def test_no_header_no_echo_on_cue_create(client, auth_headers):
+    """Default behavior: POST /v1/cues without header has no echo fields."""
+    r = await client.post(
+        "/v1/cues",
+        json={
+            "name": f"no-echo-{uuid.uuid4().hex[:6]}",
+            "schedule": {"type": "recurring", "cron": "0 * * * *"},
+            "callback": {"url": "https://example.com/webhook"},
+            "payload": {"task": "test"},
+        },
+        headers=auth_headers,
+    )
+    assert r.status_code == 201
+    data = r.json()
+    assert "body_received" not in data
+    assert "body_received_sha256" not in data
+
+
+# ───────────────────────────────────────────────────────────────────────
+# Happy path — header present + dict response → echo fields injected
+# ───────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_echo_on_agents_create(client, auth_headers):
+    """POST /v1/agents with X-CueAPI-Verify-Echo: true → echo fields present."""
+    display = f"EchoAgent {uuid.uuid4().hex[:6]}"
+    r = await client.post(
+        "/v1/agents",
+        json={"display_name": display, "metadata": {}},
+        headers={**auth_headers, VERIFY_ECHO_HEADER_KEY: "true"},
+    )
+    assert r.status_code == 201
+    data = r.json()
+    assert "body_received" in data
+    assert data["body_received"]["display_name"] == display
+    assert isinstance(data["body_received_sha256"], str)
+    assert len(data["body_received_sha256"]) == 64
+
+
+@pytest.mark.asyncio
+async def test_echo_on_cue_create(client, auth_headers):
+    """POST /v1/cues with X-CueAPI-Verify-Echo: true → echo fields present."""
+    name = f"echo-cue-{uuid.uuid4().hex[:6]}"
+    r = await client.post(
+        "/v1/cues",
+        json={
+            "name": name,
+            "schedule": {"type": "recurring", "cron": "0 * * * *"},
+            "callback": {"url": "https://example.com/webhook"},
+            "payload": {"task": "test"},
+        },
+        headers={**auth_headers, VERIFY_ECHO_HEADER_KEY: "true"},
+    )
+    assert r.status_code == 201
+    data = r.json()
+    assert "body_received" in data
+    assert data["body_received"]["name"] == name
+    assert len(data["body_received_sha256"]) == 64
+
+
+# ───────────────────────────────────────────────────────────────────────
+# Method gating — only POST/PATCH/PUT
+# ───────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_get_request_no_echo_even_with_header(client, auth_headers):
+    """GET requests bypass the middleware — header set, no fields injected."""
+    r = await client.get(
+        "/v1/usage",
+        headers={**auth_headers, VERIFY_ECHO_HEADER_KEY: "true"},
+    )
+    assert r.status_code == 200
+    data = r.json()
+    assert "body_received" not in data
+    assert "body_received_sha256" not in data
+
+
+# ───────────────────────────────────────────────────────────────────────
+# Status gating — non-2xx responses should NOT carry echo fields
+# ───────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_validation_error_no_echo(client, auth_headers):
+    """422 / 4xx validation responses stay clean — middleware doesn't inject."""
+    r = await client.post(
+        "/v1/agents",
+        json={},  # missing required display_name
+        headers={**auth_headers, VERIFY_ECHO_HEADER_KEY: "true"},
+    )
+    assert r.status_code in (400, 422)
+    data = r.json()
+    assert "body_received" not in data
+    assert "body_received_sha256" not in data
+
+
+# ───────────────────────────────────────────────────────────────────────
+# Idempotency — Phase 1 endpoints already echo; middleware must not double
+# ───────────────────────────────────────────────────────────────────────
+
+
+async def _make_agent(client, headers, slug=None):
+    payload = {"display_name": f"Agent {uuid.uuid4().hex[:6]}", "metadata": {}}
+    if slug:
+        payload["slug"] = slug
+    r = await client.post("/v1/agents", json=payload, headers=headers)
+    assert r.status_code == 201, r.text
+    return r.json()
+
+
+def _from_header(agent):
+    return {"X-Cueapi-From-Agent": agent["id"]}
+
+
+@pytest.mark.asyncio
+async def test_phase_1_messages_endpoint_idempotent_under_middleware(
+    client, auth_headers
+):
+    """Phase 1's send_message handler already injects echo — middleware must
+    NOT overwrite the handler-supplied ``body_received``. The handler echoes
+    the parsed Pydantic dict; middleware would echo the raw JSON. We pin the
+    handler-supplied shape (sentinel: handler's view includes ``body`` key
+    inside ``body_received``)."""
+    sender = await _make_agent(
+        client, auth_headers, slug=f"idem-s-{uuid.uuid4().hex[:6]}"
+    )
+    recipient = await _make_agent(
+        client, auth_headers, slug=f"idem-r-{uuid.uuid4().hex[:6]}"
+    )
+    r = await client.post(
+        "/v1/messages",
+        json={"to": recipient["id"], "body": "idem test"},
+        headers={
+            **auth_headers,
+            **_from_header(sender),
+            VERIFY_ECHO_HEADER_KEY: "true",
+        },
+    )
+    assert r.status_code == 201
+    data = r.json()
+    assert "body_received" in data
+    # Phase 1 handler dumps the Pydantic MessageCreate, so body_received is
+    # a dict with the parsed fields. Middleware-injected echo would also be
+    # a dict (parsed from raw JSON) — but we want to be sure idempotency
+    # preserved the handler's view.
+    assert data["body_received"]["body"] == "idem test"
+
+
+# ───────────────────────────────────────────────────────────────────────
+# 6 metachar classes on a non-Phase-1 endpoint
+# ───────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "metachar_class, payload_value",
+    [
+        ("backticks", "agent `backticks` literal"),
+        ("dollar_paren", "agent $(echo X) literal"),
+        ("dollar_brace", "agent ${VAR} literal"),
+        ("backslash", "agent \\n \\t literal"),
+        ("quotes", "agent 'single' and \"double\" literal"),
+        ("mixed", "agent mixed `cmd` $(sub) ${ref} \\esc \"q\" 'q' literal"),
+    ],
+)
+@pytest.mark.asyncio
+async def test_echo_six_metachar_classes_on_agents_create(
+    client, auth_headers, metachar_class, payload_value
+):
+    """Definition of Done item 1 extension: byte-identical round-trip on a
+    non-Phase-1 endpoint exercised by the middleware."""
+    r = await client.post(
+        "/v1/agents",
+        json={"display_name": payload_value, "metadata": {}},
+        headers={**auth_headers, VERIFY_ECHO_HEADER_KEY: "true"},
+    )
+    assert r.status_code == 201, r.text
+    data = r.json()
+    assert (
+        data["body_received"]["display_name"] == payload_value
+    ), f"metachar class {metachar_class} did NOT survive middleware echo"
+
+
+# ───────────────────────────────────────────────────────────────────────
+# Empty body handling
+# ───────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_echo_empty_body_post(client, auth_headers, db_session):
+    """POST with no body + verify-echo header: server returns 200, echo is
+    body_received=None + SHA256 of empty bytes."""
+    # Create a worker-transport cue we can fire without body
+    cr = await client.post(
+        "/v1/cues",
+        json={
+            "name": f"empty-body-{uuid.uuid4().hex[:6]}",
+            "schedule": {"type": "recurring", "cron": "0 * * * *"},
+            "transport": "worker",
+            "payload": {"task": "test"},
+        },
+        headers=auth_headers,
+    )
+    assert cr.status_code == 201, cr.text
+    cue_id = cr.json()["id"]
+    # Fire with no body — Phase 1's per-handler echo handles this case;
+    # middleware should see existing body_received and skip.
+    r = await client.post(
+        f"/v1/cues/{cue_id}/fire",
+        headers={**auth_headers, VERIFY_ECHO_HEADER_KEY: "true"},
+    )
+    assert r.status_code == 200, r.text
+    data = r.json()
+    assert "body_received" in data
+    assert data["body_received"] is None
+    assert data["body_received_sha256"] == hashlib.sha256(b"").hexdigest()
+
+
+# ───────────────────────────────────────────────────────────────────────
+# Case-insensitive header value
+# ───────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_header_value_case_insensitive(client, auth_headers):
+    """Header value 'TRUE' / 'True' triggers echo (case-insensitive match)."""
+    r = await client.post(
+        "/v1/agents",
+        json={"display_name": f"CaseAgent {uuid.uuid4().hex[:6]}", "metadata": {}},
+        headers={**auth_headers, VERIFY_ECHO_HEADER_KEY: "TRUE"},
+    )
+    assert r.status_code == 201
+    assert "body_received" in r.json()
+
+
+@pytest.mark.asyncio
+async def test_header_value_non_true_no_echo(client, auth_headers):
+    """Header value 'false' / '1' / 'yes' → middleware bypasses."""
+    for val in ("false", "1", "yes", ""):
+        r = await client.post(
+            "/v1/agents",
+            json={
+                "display_name": f"NonTrue {uuid.uuid4().hex[:6]}",
+                "metadata": {},
+            },
+            headers={**auth_headers, VERIFY_ECHO_HEADER_KEY: val},
+        )
+        assert r.status_code == 201, f"value={val!r}: {r.text}"
+        assert (
+            "body_received" not in r.json()
+        ), f"value={val!r} unexpectedly triggered echo"
+
+
+# ───────────────────────────────────────────────────────────────────────
+# Direct unit tests on _maybe_inject — covers branch logic without going
+# through ASGI dispatch (per CLAUDE.md ASGI coverage discipline).
+# ───────────────────────────────────────────────────────────────────────
+
+
+def _ct_header(value: str) -> list[tuple[bytes, bytes]]:
+    return [(b"content-type", value.encode("ascii")), (b"content-length", b"99")]
+
+
+def test_maybe_inject_non_2xx_returns_unchanged():
+    """4xx/5xx responses bypass injection."""
+    from app.middleware.verify_echo import VerifyEchoMiddleware
+
+    result = VerifyEchoMiddleware._maybe_inject(
+        status_code=422,
+        headers=_ct_header("application/json"),
+        request_body_bytes=b'{"x": 1}',
+        response_body_bytes=b'{"detail": "validation_error"}',
+    )
+    # Headers and body unchanged
+    assert result["body"] == b'{"detail": "validation_error"}'
+
+
+def test_maybe_inject_non_json_content_type_returns_unchanged():
+    """HTML/text responses bypass injection."""
+    from app.middleware.verify_echo import VerifyEchoMiddleware
+
+    result = VerifyEchoMiddleware._maybe_inject(
+        status_code=200,
+        headers=_ct_header("text/html; charset=utf-8"),
+        request_body_bytes=b'{"x": 1}',
+        response_body_bytes=b"<html>ok</html>",
+    )
+    assert result["body"] == b"<html>ok</html>"
+
+
+def test_maybe_inject_non_dict_json_response_returns_unchanged():
+    """JSON array / scalar responses bypass injection."""
+    from app.middleware.verify_echo import VerifyEchoMiddleware
+
+    result = VerifyEchoMiddleware._maybe_inject(
+        status_code=200,
+        headers=_ct_header("application/json"),
+        request_body_bytes=b'{"x": 1}',
+        response_body_bytes=b'[1, 2, 3]',
+    )
+    assert result["body"] == b'[1, 2, 3]'
+
+
+def test_maybe_inject_malformed_response_json_returns_unchanged():
+    """Defensive — response not parseable as JSON despite content-type."""
+    from app.middleware.verify_echo import VerifyEchoMiddleware
+
+    result = VerifyEchoMiddleware._maybe_inject(
+        status_code=200,
+        headers=_ct_header("application/json"),
+        request_body_bytes=b'{"x": 1}',
+        response_body_bytes=b'not valid json',
+    )
+    assert result["body"] == b'not valid json'
+
+
+def test_maybe_inject_existing_body_received_preserved():
+    """Phase 1 handler-supplied body_received wins over middleware injection."""
+    from app.middleware.verify_echo import VerifyEchoMiddleware
+    import json as _j
+
+    existing = {
+        "id": "msg_x",
+        "body_received": {"body": "phase-1-shape"},
+        "body_received_sha256": "phase-1-hash",
+    }
+    result = VerifyEchoMiddleware._maybe_inject(
+        status_code=201,
+        headers=_ct_header("application/json"),
+        request_body_bytes=b'{"body": "raw-shape"}',
+        response_body_bytes=_j.dumps(existing).encode("utf-8"),
+    )
+    # Body unchanged — Phase 1 view preserved
+    parsed = _j.loads(result["body"])
+    assert parsed["body_received"] == {"body": "phase-1-shape"}
+    assert parsed["body_received_sha256"] == "phase-1-hash"
+
+
+def test_maybe_inject_empty_request_body():
+    """No request body + header set → body_received=None + SHA256 empty."""
+    from app.middleware.verify_echo import VerifyEchoMiddleware
+    import json as _j
+
+    result = VerifyEchoMiddleware._maybe_inject(
+        status_code=200,
+        headers=_ct_header("application/json"),
+        request_body_bytes=b'',
+        response_body_bytes=b'{"id": "x"}',
+    )
+    parsed = _j.loads(result["body"])
+    assert parsed["body_received"] is None
+    assert parsed["body_received_sha256"] == hashlib.sha256(b"").hexdigest()
+
+
+def test_maybe_inject_invalid_json_request_body_falls_back_to_string():
+    """Malformed JSON request body → body_received is decoded string + raw-bytes hash."""
+    from app.middleware.verify_echo import VerifyEchoMiddleware
+    import json as _j
+
+    raw = b'this is not valid {{{ json'
+    result = VerifyEchoMiddleware._maybe_inject(
+        status_code=200,
+        headers=_ct_header("application/json"),
+        request_body_bytes=raw,
+        response_body_bytes=b'{"id": "x"}',
+    )
+    parsed = _j.loads(result["body"])
+    assert parsed["body_received"] == raw.decode("utf-8")
+    assert parsed["body_received_sha256"] == hashlib.sha256(raw).hexdigest()
+
+
+def test_maybe_inject_content_length_header_updated():
+    """When echo fields are injected, Content-Length is recomputed."""
+    from app.middleware.verify_echo import VerifyEchoMiddleware
+
+    headers_in = [
+        (b"content-type", b"application/json"),
+        (b"content-length", b"10"),  # stale
+        (b"x-custom", b"preserved"),
+    ]
+    result = VerifyEchoMiddleware._maybe_inject(
+        status_code=200,
+        headers=headers_in,
+        request_body_bytes=b'{"k": "v"}',
+        response_body_bytes=b'{"id": "x"}',
+    )
+    # New body is longer than original (echo fields added)
+    assert len(result["body"]) > 10
+    # Content-Length updated to match new body
+    cl = None
+    for k, v in result["headers"]:
+        if k.lower() == b"content-length":
+            cl = int(v)
+    assert cl == len(result["body"])
+    # Other headers preserved
+    custom = next((v for k, v in result["headers"] if k.lower() == b"x-custom"), None)
+    assert custom == b"preserved"