fix(server): PR #233 expert-review followups — security docs + retry/transform tests

Addresses code-reviewer + security-reviewer findings on PR #233.

DOCS (security — moved from docs-only to import-site where callers
see it)
- SkillMiddleware docstring now includes four security callouts:
  (1) do NOT swallow ADCPError — serves fake success for failed
      mutations (IdempotencyConflictError, ADCPTaskError).
  (2) middleware is a data processor for the full skill payload —
      params contain buyer briefs, budgets, PII; context has
      caller_identity and tenant_id. Third-party middleware gets
      the complete surface; treat as controller-processor.
  (3) exception messages land in server logs verbatim via
      logger.exception before client-facing sanitisation — do not
      format params / caller_identity into exception text.
  (4) short-circuit caches MUST key on (skill_name, params,
      caller_identity, tenant_id). skill_name + params alone serves
      principal A's data to principal B.
- Clarify params is request-side only; transforms happen on the
  return side of call_next.
- Note ContextVars propagate through call_next (same asyncio task).

DOCS — docs/handler-authoring.md
- "Semantics worth knowing" expanded with the composition-order WHY
  (audit outermost so rate-limited rejected calls don't disappear
  from audit), short-circuit cache-key requirements, retry support,
  transform-on-return-not-input rule.
- New "Security — middleware is a data processor" callout matches
  the import-site docstring.

TESTS (+2, covering code-reviewer's gap list)
- test_middleware_can_invoke_call_next_multiple_times_for_retry:
  retry-on-transient-error pattern. Middleware calls call_next 3
  times; handler fails twice, succeeds on third. Locks the
  re-entrant composition contract a naive loop-variable closure
  would break.
- test_middleware_can_transform_result_on_return_side: enriching
  middleware wraps the handler's return. Distinct code path from
  short-circuit (which never calls call_next).

NITS
- Removed stale `from a2a.types import TaskStatus  # noqa: F401
  (unused but document` line with truncated comment in
  test_middleware_can_short_circuit_without_invoking_handler.

Deferred (not blocking this PR):
- Protocol class for SkillMiddleware (do alongside ContextFactory to
  keep declaration style consistent).
- Runtime validation of middleware return shape against skill output
  schemas.
- ContextVar-propagation formal docs in a dedicated section.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bokelley

docs/handler-authoring.md

@@ -362,21 +362,52 @@ serve(MyAgent(), transport="a2a", middleware=[audit_middleware])
 
 **Semantics worth knowing:**
 
-- **Composition**: `middleware=[Audit(), RateLimit(), Metrics()]` runs
-  `Audit → RateLimit → Metrics → handler` on the way in and unwinds in
-  the opposite order.
-- **Short-circuit**: a middleware that returns without calling
-  `call_next()` stops the chain. Its return value becomes the dispatch
-  result (serialised back to the client). Rate limiters / feature
-  flags use this.
-- **Exception observation**: catch around `await call_next()` to log
-  failures. Re-raise to let the executor's normal error path take over
-  (`ADCPError` → failed task with `adcp_error` DataPart; other
-  exceptions → opaque failed task). Swallowing the exception and
-  returning a substitute result is allowed but almost always wrong.
+- **Composition — put audit outermost.** `middleware=[Audit(),
+  RateLimit(), Metrics()]` runs `Audit → RateLimit → Metrics →
+  handler` on the way in and unwinds in the opposite order. **If you
+  put rate-limiting before audit, rejected requests disappear from
+  your audit log** — often the most interesting events for security
+  review. Audit always outermost.
+- **Short-circuit — cache keys MUST include principal + tenant.** A
+  middleware that returns without calling `call_next()` stops the
+  chain; its return value becomes the dispatch result. Rate limiters
+  / feature flags use this. **Caching middleware that short-circuits
+  must key on `(skill_name, params, context.caller_identity,
+  context.tenant_id)`** — a cache keyed only on `skill_name + params`
+  serves principal A's data to principal B on a matching-params call.
+- **Exception observation — never swallow an `ADCPError`.** Catch
+  around `await call_next()` to log failures. Re-raise to let the
+  executor's normal error path take over (`ADCPError` → failed task
+  with `adcp_error` DataPart; other exceptions → opaque failed task).
+  Swallowing an `ADCPError` (especially `IdempotencyConflictError` or
+  `ADCPTaskError`) and returning a fake-success dict silently converts
+  a rejected mutation into a "completed" task — double-billing,
+  double-allocation, duplicated side effects. Don't.
+- **Exception messages end up in server logs.** Middleware-raised
+  exceptions flow through `logger.exception` in the executor before
+  client-facing sanitisation. Don't format `params` or
+  `context.caller_identity` into exception text — operators read those
+  logs.
+- **Retry is supported.** Call `call_next()` more than once (e.g.
+  retry-on-transient-error middleware). Each call gets a fresh
+  inner chain — composition is re-entrant by design.
+- **Transform on return, not on input.** `params` passed in is the
+  same dict every middleware sees. Mutating it doesn't change what
+  the next layer receives. Transforms happen on the *return* side by
+  modifying the value of `await call_next()`.
 - **Context access**: the middleware sees the `ToolContext` produced
   by the `context_factory` (or the a2a-sdk fallback). Tenant id,
-  caller identity, anything your factory populates.
+  caller identity, anything your factory populates. `ContextVar`s set
+  before `call_next()` propagate to the handler — no `asyncio.create_task`
+  needed.
+
+**Security — middleware is a data processor for the full skill
+payload.** `params` carries decoded buyer briefs, budgets, brand
+refs, proposal text, PII in message parts. `context` carries
+`caller_identity` + `tenant_id`. Installing a third-party middleware
+(SaaS audit, observability vendor, bespoke tracing) hands that vendor
+the complete skill surface. Treat it as a data processor under your
+GDPR/CCPA controller-processor agreements.
 
 MCP transport has its own middleware story (see "Pattern 2 —
 in-process HTTP middleware" above); `SkillMiddleware` is A2A-only.

src/adcp/server/serve.py

@@ -81,25 +81,70 @@ async def middleware(
     ) -> Any:
         ...
 
-Middleware wraps ``call_next()`` — call it (possibly more than once, or
-never) to invoke the rest of the chain plus the underlying handler.
-Anything the middleware returns becomes the dispatch result the A2A
-transport serialises back to the client, so middleware can short-circuit
-(skip the handler entirely) or transform the result.
+Middleware wraps ``call_next()`` — call it (possibly more than once to
+implement retry, or never to short-circuit) to invoke the rest of the
+chain plus the underlying handler. Anything the middleware returns
+becomes the dispatch result the A2A transport serialises back to the
+client, so middleware can short-circuit (skip the handler entirely) or
+transform the result on the return side.
 
 Middleware observes both success and failure — catch exceptions around
 ``call_next()`` to implement audit-on-failure or retry-classifier hooks.
 Middleware re-raising propagates to the executor's normal error path
 (application ``ADCPError`` → failed task w/ ``adcp_error`` DataPart;
 other exceptions → opaque failed task per the spec's error-sanitisation
-rule).
+rule). **Swallowing an exception and returning a substitute result is
+allowed but almost always wrong** — in particular, swallowing
+``ADCPError`` subclasses (``IdempotencyConflictError``,
+``ADCPTaskError``) serves a fake success for a failed mutation, which
+double-bills / double-allocates in production.
+
+``params`` is the parsed request dict passed to every middleware in
+the chain and to the handler. Middleware cannot mutate what the next
+layer sees by mutating ``params`` — transforms happen on the return
+side only, by modifying the value returned from ``call_next()``.
 
 Multiple middlewares compose outermost-first, matching Starlette/ASGI
 semantics — if you pass ``middleware=[Audit(), RateLimit(), Metrics()]``,
 the runtime order is::
 
     Audit.__call__ →  RateLimit.__call__ →  Metrics.__call__ →  handler
 
+**Put audit outermost.** Middleware that short-circuits (rate limiter,
+feature-flag gate) never calls ``call_next()``, so anything deeper in
+the chain never sees the request. If your audit middleware sits
+*after* the rate limiter, rejected calls disappear from the audit
+trail — often the most interesting events for security review.
+
+``call_next()`` runs in the same asyncio task as the middleware that
+invoked it, so ``ContextVar`` values set before the call are visible
+to downstream middleware and the handler. Don't ``asyncio.create_task``
+your way around this unless you need the isolation.
+
+**Security — middleware is a data processor for the full skill payload.**
+``params`` is decoded business content (buyer briefs, budgets, brand
+references, proposal text, PII in message parts). ``context`` carries
+``caller_identity``, ``tenant_id``, and anything your ``context_factory``
+populates. Installing a third-party middleware (observability vendor,
+SaaS audit pipeline, external tracing) hands that vendor the complete
+skill payload surface — treat it as a data processor under your
+GDPR/CCPA controller-processor relationships and review the blast
+radius before wiring vendors here.
+
+**Security — do not format ``params`` or ``context.caller_identity``
+into exception messages.** Middleware-raised exceptions pass through
+``logger.exception`` in the executor (server-side trace with the raw
+message) before the executor's sanitisation kicks in for the client
+response. Exception text ends up in operator logs verbatim; keep it
+opaque.
+
+**Security — short-circuit caches MUST include principal + tenant in
+the cache key.** A middleware that caches on ``skill_name + params``
+alone and returns a cached result without calling ``call_next()``
+will serve principal A's data to principal B on a matching-params
+call. Key on ``(skill_name, params, context.caller_identity,
+context.tenant_id)``.
+
 Example — audit logging with exception capture::
 
     from adcp.server import SkillMiddleware, ToolContext
@@ -114,7 +159,10 @@ async def audit_middleware(
         try:
             result = await call_next()
         except Exception as exc:
-            audit_log.failure(skill_name, context.caller_identity, exc)
+            # Keep exception text opaque — this ends up in server logs.
+            audit_log.failure(
+                skill_name, context.caller_identity, type(exc).__name__
+            )
             raise
         audit_log.success(
             skill_name,

tests/test_a2a_server.py

@@ -951,8 +951,6 @@ async def test_middleware_can_short_circuit_without_invoking_handler():
     """Middleware that returns without calling ``call_next`` stops the
     chain and its return value becomes the dispatch result. Rate
     limiters and feature flags rely on this."""
-    from a2a.types import TaskStatus  # noqa: F401 (unused but document
-
     handler_called = False
 
     class _TrackingHandler(ADCPHandler):
@@ -1054,3 +1052,74 @@ async def noop_mw(skill_name, params, context, call_next):
     executor = handler.agent_executor
     assert isinstance(executor, ADCPAgentExecutor)
     assert executor._middleware == (noop_mw,)
+
+
+async def test_middleware_can_invoke_call_next_multiple_times_for_retry():
+    """Retry-on-transient-error middleware calls ``call_next()`` more
+    than once — each call builds a fresh inner chain. This locks the
+    re-entrant composition contract a naive loop-variable closure would
+    break."""
+    call_counts = {"mw": 0, "handler": 0}
+
+    async def retry_middleware(skill_name, params, context, call_next):
+        last_exc: Exception | None = None
+        for _ in range(3):
+            call_counts["mw"] += 1
+            try:
+                return await call_next()
+            except RuntimeError as exc:
+                last_exc = exc
+        raise last_exc if last_exc else RuntimeError("unreachable")
+
+    class _TransientFailHandler(ADCPHandler):
+        async def get_adcp_capabilities(self, params: Any, context: Any = None) -> Any:
+            return {}
+
+        async def get_products(self, params: Any, context: Any = None) -> Any:
+            call_counts["handler"] += 1
+            if call_counts["handler"] < 3:
+                raise RuntimeError("transient")
+            return {"products": [{"id": "finally"}]}
+
+    executor = ADCPAgentExecutor(_TransientFailHandler(), middleware=[retry_middleware])
+    ctx = RequestContext(request=MessageSendParams(message=_make_datapart_msg("get_products")))
+    queue = EventQueue()
+    await executor.execute(ctx, queue)
+
+    assert call_counts["mw"] == 3
+    assert call_counts["handler"] == 3
+    event = await queue.dequeue_event(no_wait=True)
+    assert isinstance(event, Task)
+    assert event.status.state == "completed"
+
+
+async def test_middleware_can_transform_result_on_return_side():
+    """Middleware can mutate or replace the value of ``call_next()``
+    before returning it. The transformed value is what the client
+    sees — covers the annotation / enrichment use case distinct from
+    short-circuiting."""
+
+    async def enriching_middleware(skill_name, params, context, call_next):
+        result = await call_next()
+        # Wrap handler's return with a marker the test observes.
+        if isinstance(result, dict):
+            return {**result, "middleware_marker": "wrapped"}
+        return result
+
+    executor = ADCPAgentExecutor(_TestHandler(), middleware=[enriching_middleware])
+    ctx = RequestContext(request=MessageSendParams(message=_make_datapart_msg("get_products")))
+    queue = EventQueue()
+    await executor.execute(ctx, queue)
+
+    event = await queue.dequeue_event(no_wait=True)
+    assert isinstance(event, Task)
+    assert event.status.state == "completed"
+    data_parts = [
+        p.root
+        for p in event.artifacts[0].parts
+        if hasattr(p.root, "data") and isinstance(p.root.data, dict)
+    ]
+    result = data_parts[0].data
+    assert result["middleware_marker"] == "wrapped"
+    # And the handler's original payload is still there.
+    assert result["products"][0]["id"] == "p1"