feat(transport): add Lambda Lang content-type negotiation

[voidborne-d]

Summary

Adds Lambda Lang content-type negotiation for ASAP envelopes, enabling semantic compression of JSON-RPC payloads.

Closes #52

What

Lambda Codec (src/asap/transport/codecs/lambda_codec.py)

• Self-contained encoder/decoder — no external dependencies
• Substitutes common JSON-RPC and ASAP keys with short Lambda Lang atoms
• 100% round-trip fidelity: decode(encode(data)) == data
• Content-Type: application/vnd.asap+lambda
• Version-prefixed format (λ1:) for forward compatibility

Server-side negotiation (server.py)

• Checks Accept: application/vnd.asap+lambda header
• Encodes successful responses with Lambda codec when requested
• Graceful fallback to JSON on encode failure
• Error responses always remain JSON
• Zero impact on existing behavior when Lambda is not requested

Client-side negotiation (client.py)

• New lambda_codec_enabled parameter (default: False)
• When enabled, sends Accept: application/vnd.asap+lambda header
• Automatically decodes Lambda responses based on Content-Type
• Opt-in only — no change to default client behavior

Tests

• 27 unit tests (tests/transport/unit/test_lambda_codec.py): round-trips, atom mappings, edge cases, error handling
• 7 integration tests (tests/transport/integration/test_lambda_negotiation.py): content-type negotiation, fallback, mixed scenarios
• All 803+ existing tests continue to pass
• Full CI: pytest, ruff check, ruff format, mypy — all green

Design Decisions

• Follows the same graceful degradation pattern as compression.py
• Uses asap.observability.get_logger() for structured logging
• Atom delimiters (§…§) chosen because they cannot appear in valid JSON keys
• Conservative negotiation: wildcards (*/*) do NOT trigger Lambda encoding
• No modifications to envelope.py or any existing behavior

[vercel]

@voidborne-d is attempting to deploy a commit to the adrianno's projects Team on Vercel.

A member of the Team first needs to authorize it.

[adriannoes]

This PR perfectly executes the architectural vision outlined in Issue #52, keeping the Lambda Lang encoding cleanly segregated at the transport layer via Content-Type negotiation.

However, before it can be merged into production, the serialization logic needs to be heavily optimized. The current implementation relies on the standard json module and sequential string replacements, which act as a CPU-bound bottleneck that will block the asyncio event loop.

[adriannoes]

This loop iterating over 35 .replace() operations on a potentially massive JSON string is highly CPU-bound. Because FastAPI operates on an asyncio event loop, executing this directly will block the event loop, stalling all concurrent agent requests (DoS risk).

Suggestion: Optimize this to a single pass using Python's built-in re module. Because re is implemented in C under the hood, running one regex substitution is drastically faster than executing a Python for loop and avoids creating 35 intermediate strings in memory:

    ```python
    import re

    # Pre-compile globally
    _ENCODE_PATTERN = re.compile("|".join(map(re.escape, _ENCODE_MAP.keys())))

    def _encode_match(m: re.Match) -> str:
        return _ENCODE_MAP[m.group(0)]

    def encode(json_str: str) -> str:
        # Assuming you accepted the suggestion to take json_str directly
        encoded_str = _ENCODE_PATTERN.sub(_encode_match, json_str)
        return _VERSION_PREFIX + encoded_str
    ```

Even with this optimization, for very large payloads, this should ideally be called via await asyncio.to_thread(lambda_codec.encode, ...) from the routing layers to ensure the event loop is never blocked.

[adriannoes]

The same event loop blocking risk applies here during decoding. Please apply the parallel regex optimization here as well:

    ```python
    _DECODE_PATTERN = re.compile("|".join(map(re.escape, _DECODE_MAP.keys())))

    def _decode_match(m: re.Match) -> str:
        return _DECODE_MAP[m.group(0)]

    def decode(encoded_str: str) -> str:
        # Check version prefix...
        json_str = encoded_str[len(_VERSION_PREFIX):]
        return _DECODE_PATTERN.sub(_decode_match, json_str)
    ```

Problem: Taking the raw string and executing an O(N*K) string replacement loop in an async context.

Rationale: FastAPI operates on an asyncio event loop. Performing extremely heavy CPU-bound string manipulation on potentially massive JSON payloads (e.g., large LLM context windows) will block the event loop entirely. This stalls all concurrent agent requests and introduces a severe DoS vector.

[adriannoes]

Accepting a dict here forces us to call json.dumps() in Python, which is much slower than utilizing Pydantic's native Rust core for serialization.

Suggestion: Change the signature to accept a pre-serialized JSON string. We can rely on the upstream routing layer to dump the model efficiently and pass it here:

    ```python
    def encode(json_str: str) -> str:
        # Drop the local json.dumps() import and execution.
        # (Apply the regex optimization on `json_str` directly)
    ```

[adriannoes]

Running model_dump() creates a massive intermediate Python dictionary just to pass it to the codec for serializing again. This completely bypasses Pydantic v2's Rust-based performance architecture.

My suggestion is to serialize directly to JSON using Pydantic's core and pass the string to the updated codec:

    ```python
    # Leverage Pydantic's native Rust execution for near-native speeds
    encoded_body = lambda_codec.encode(rpc_response.model_dump_json(by_alias=True))
    ```

Problem: The codec accepts a dict and uses standard json.dumps(). In server.py, rpc_response.model_dump() creates an intermediate dict just to be re-serialized.

Rationale:

• Converting Pydantic models to dict and then to JSON in Python adds noticeable layer overhead for large payloads.
• We can keep the codebase lean (without adding external orjson dependencies) by ensuring the custom codec plugs into the pre-serialized output of Pydantic's highly-optimized Rust core (model_dump_json()).

[adriannoes]

Problem: Catching a generic Exception and logging a warning using str(e), completely discarding the stack trace.

Rationale: We heavily depend on observability. Without a stack trace, debugging production encoding failures becomes nearly impossible.

Fix Suggestion:

```python
except Exception as e:
    logger.warning(
        "asap.server.lambda_encode_failed",
        error=str(e),
        error_type=type(e).__name__,
        exc_info=True,  # MUST log the stack trace
    )
```

[adriannoes]

This decode call runs string replacement loops directly on response.text. For large payloads, this will block the client's asyncio event loop while waiting for CPU-bound parsing.

Suggestion: Offload the decoding to a thread to keep the client's async event loop free:

    ```python
    import asyncio

    if LAMBDA_CONTENT_TYPE in response_content_type:
        # Offload CPU-bound decoding to unblock the main event loop
        json_response = await asyncio.to_thread(lambda_codec.decode, response.text)
    else:
        json_response = response.json()
    ```

[adriannoes]

Thanks for the PR, @voidborne-d! The implementation is excellent and perfectly aligns with the Content-Type negotiation approach we discussed in #52.

I've left a few inline comments regarding serialization performance and asyncio event loop blocking and I've provided ready-to-use code snippets in the inline comments to make this easy.

Once those are addressed, this is good to merge!

[voidborne-d]

Thanks for the thorough review @adriannoes! All 6 suggestions addressed in the latest push:

1. ✅ Single-pass regex — pre-compiled re.compile + .sub() replaces the O(N×K) .replace() loop
2. ✅ String-based API — encode(json_str) / decode() -> str, no more json.dumps() inside the codec
3. ✅ model_dump_json(by_alias=True) in server — direct Pydantic Rust-core serialization, no intermediate dict
4. ✅ asyncio.to_thread for client-side decode — event loop stays free
5. ✅ exc_info=True on encode failure — full stack trace in logs
6. ✅ Tests updated for the new string-based API

Used your code snippets directly where provided. Ready for re-review!

[adriannoes]

LGTM