Type SESSION_PROPOSE config with a validated SessionProposeConfig model (#18)

examples/session_flow.json

@@ -43,7 +43,7 @@
       "config": {
         "preferred_payload_modes": ["semantic_frame", "text"],
         "ttl_secs": 3600,
-        "required_trust_domain": "research.internal"
+        "trust_domain": "research.internal"
       }
     },
     "payload_mode": "text",

sdk/python/src/ldp_protocol/client.py

@@ -11,7 +11,7 @@
 from ldp_protocol.types.contract import DelegationContract, FailurePolicy
 from ldp_protocol.types.error import LdpError
 from ldp_protocol.types.identity import LdpIdentityCard
-from ldp_protocol.types.messages import LdpEnvelope, LdpMessageBody
+from ldp_protocol.types.messages import LdpEnvelope, LdpMessageBody, SessionProposeConfig
 from ldp_protocol.types.payload import PayloadMode
 from ldp_protocol.types.provenance import Provenance
 from ldp_protocol.types.session import LdpSession, SessionConfig, SessionState
@@ -190,11 +190,11 @@ async def establish_session(self, url: str) -> LdpSession:
             from_id=self.delegate_id,
             to_id=url,
             body=LdpMessageBody.session_propose(
-                config={
-                    "preferred_payload_modes": [m.value for m in advertised_modes],
-                    "ttl_secs": self.config.ttl_secs,
-                    "trust_domain": self.trust_domain.name,
-                }
+                config=SessionProposeConfig(
+                    preferred_payload_modes=advertised_modes,
+                    ttl_secs=self.config.ttl_secs,
+                    trust_domain=self.trust_domain.name,
+                )
             ),
         )
         propose_resp = await self.send_message(url, propose)

sdk/python/src/ldp_protocol/delegate.py

@@ -210,11 +210,25 @@ def _handle_hello(self, envelope: LdpEnvelope) -> LdpEnvelope:
         )
 
     def _handle_session_propose(self, envelope: LdpEnvelope) -> LdpEnvelope:
-        # Validate trust domain
-        remote_domain = "unknown"
-        if envelope.body.config and "trust_domain" in envelope.body.config:
-            remote_domain = envelope.body.config["trust_domain"]
+        # `config` is a typed SessionProposeConfig, validated at the wire
+        # boundary (issue #18): trust_domain and preferred_payload_modes are
+        # already well-formed here, and LdpMessageBody's validator guarantees
+        # config is present for SESSION_PROPOSE (the forward-compat tolerance the
+        # old dict[str, Any] guards provided now lives in the model). The guard
+        # below is a fail-safe backstop for that invariant — fail closed on
+        # trust rather than assume an "unknown" domain.
+        config = envelope.body.config
+        if config is None:
+            return LdpEnvelope.create(
+                session_id=envelope.session_id,
+                from_id=self.identity.delegate_id,
+                to_id=envelope.from_,
+                body=LdpMessageBody.session_reject(
+                    reason="SESSION_PROPOSE requires a config with a trust_domain"
+                ),
+            )
 
+        remote_domain = config.trust_domain
         if not self.identity.trust_domain.trusts(remote_domain):
             return LdpEnvelope.create(
                 session_id=envelope.session_id,
@@ -226,21 +240,7 @@ def _handle_session_propose(self, envelope: LdpEnvelope) -> LdpEnvelope:
             )
 
         session_id = str(uuid4())
-        # Negotiate payload mode. `config` is untrusted wire input
-        # (dict[str, Any], unvalidated), so both the preferred-modes list AND its
-        # items may be malformed. Treat a non-list as "no preference expressed"
-        # and drop non-string / unknown / forward-compat items, so a Mode-3-aware
-        # peer and an older peer still negotiate down to a shared known mode
-        # instead of crashing the handler — a non-iterable would raise on
-        # iteration, and an unhashable item would raise on the membership check.
-        initiator_modes = []
-        if envelope.body.config:
-            raw_modes = envelope.body.config.get("preferred_payload_modes")
-            if isinstance(raw_modes, list):
-                valid_modes = {mode.value for mode in PayloadMode}
-                initiator_modes = [
-                    PayloadMode(m) for m in raw_modes if isinstance(m, str) and m in valid_modes
-                ]
+        initiator_modes = config.preferred_payload_modes
         negotiated = negotiate_payload_mode(
             initiator_modes, self.identity.advertised_payload_modes()
         )

sdk/python/src/ldp_protocol/types/__init__.py

@@ -19,7 +19,7 @@
 from ldp_protocol.types.provenance import Provenance
 from ldp_protocol.types.verification import EvidenceRef, ProvenanceEntry, VerificationStatus
 from ldp_protocol.types.session import LdpSession, SessionConfig, SessionState
-from ldp_protocol.types.messages import LdpEnvelope, LdpMessageBody
+from ldp_protocol.types.messages import LdpEnvelope, LdpMessageBody, SessionProposeConfig
 
 __all__ = [
     "PayloadMode",
@@ -47,4 +47,5 @@
     "SessionState",
     "LdpEnvelope",
     "LdpMessageBody",
+    "SessionProposeConfig",
 ]

sdk/python/src/ldp_protocol/types/messages.py

@@ -6,14 +6,55 @@
 from typing import Any
 from uuid import uuid4
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator, model_validator
 
 from ldp_protocol.types.contract import DelegationContract
 from ldp_protocol.types.error import LdpError
 from ldp_protocol.types.payload import PayloadMode
 from ldp_protocol.types.provenance import Provenance
 
 
+class SessionProposeConfig(BaseModel):
+    """Typed, validated body of a SESSION_PROPOSE message.
+
+    Replaces the former ``config: dict[str, Any]`` hatch so Pydantic enforces
+    the wire shape at the boundary (issue #18). ``trust_domain`` is required and
+    non-blank; ``ttl_secs``, when present, must be positive;
+    ``preferred_payload_modes`` is tolerant of forward-compat peers — unknown or
+    malformed mode entries are dropped rather than failing the whole handshake,
+    so a peer on a newer mode set still negotiates down to a shared known mode.
+    """
+
+    preferred_payload_modes: list[PayloadMode] = Field(default_factory=list)
+    ttl_secs: int | None = Field(default=None, gt=0)
+    trust_domain: str
+
+    @field_validator("trust_domain")
+    @classmethod
+    def _trust_domain_not_blank(cls, value: str) -> str:
+        # Mirror TrustDomain.name: a blank or whitespace-only domain must not
+        # slip through the typed boundary and reach the trust check as "".
+        if not value.strip():
+            raise ValueError("trust_domain must not be blank")
+        return value
+
+    @field_validator("preferred_payload_modes", mode="before")
+    @classmethod
+    def _drop_unrecognized_modes(cls, value: Any) -> Any:
+        # Untrusted wire input. A non-list means "no preference expressed"; any
+        # unknown / forward-compat / non-string item is dropped so a newer peer
+        # degrades to the shared known modes instead of failing validation.
+        # Recognised values fall through to normal PayloadMode coercion.
+        if not isinstance(value, list):
+            return []
+        known = {mode.value for mode in PayloadMode}
+        return [
+            item
+            for item in value
+            if isinstance(item, PayloadMode) or (isinstance(item, str) and item in known)
+        ]
+
+
 class LdpMessageBody(BaseModel):
     """LDP message body — tagged union matching the Rust implementation."""
 
@@ -27,7 +68,7 @@ class LdpMessageBody(BaseModel):
     capabilities: Any | None = None
 
     # SESSION_PROPOSE
-    config: dict[str, Any] | None = None
+    config: SessionProposeConfig | None = None
 
     # SESSION_ACCEPT
     session_id: str | None = None
@@ -58,6 +99,16 @@ class LdpMessageBody(BaseModel):
     claim: Any | None = None
     evidence: Any | None = None
 
+    @model_validator(mode="after")
+    def _require_config_for_session_propose(self) -> LdpMessageBody:
+        # A SESSION_PROPOSE must carry a config (and therefore a trust_domain):
+        # a config-less proposal is rejected at the wire boundary as malformed
+        # rather than silently negotiating as the "unknown" trust domain. Keeps
+        # the trust_domain-required contract honest end to end (issue #18).
+        if self.type == "SESSION_PROPOSE" and self.config is None:
+            raise ValueError("SESSION_PROPOSE requires a config with a trust_domain")
+        return self
+
     @classmethod
     def hello(cls, delegate_id: str, supported_modes: list[PayloadMode]) -> LdpMessageBody:
         return cls(type="HELLO", delegate_id=delegate_id, supported_modes=supported_modes)
@@ -67,7 +118,8 @@ def capability_manifest(cls, capabilities: Any) -> LdpMessageBody:
         return cls(type="CAPABILITY_MANIFEST", capabilities=capabilities)
 
     @classmethod
-    def session_propose(cls, config: dict[str, Any]) -> LdpMessageBody:
+    def session_propose(cls, config: SessionProposeConfig | dict[str, Any]) -> LdpMessageBody:
+        # A dict is coerced to SessionProposeConfig (and validated) on construction.
         return cls(type="SESSION_PROPOSE", config=config)
 
     @classmethod

sdk/python/tests/test_client.py

@@ -56,7 +56,7 @@ async def fake_discover(self, url):
 
 def _proposed_modes(sent: list) -> list[str]:
     propose = next(e for e in sent if e.body.type == "SESSION_PROPOSE")
-    return propose.body.config["preferred_payload_modes"]
+    return [m.value for m in propose.body.config.preferred_payload_modes]
 
 
 def _client() -> LdpClient:

sdk/python/tests/test_types.py

@@ -1,5 +1,8 @@
 """Tests for LDP protocol types."""
 
+import pytest
+from pydantic import ValidationError
+
 from ldp_protocol.types import (
     LdpCapability,
     LdpEnvelope,
@@ -9,6 +12,7 @@
     Provenance,
     QualityMetrics,
     SessionConfig,
+    SessionProposeConfig,
     TrustDomain,
     negotiate_payload_mode,
     register_payload_mode,
@@ -619,7 +623,9 @@ def test_all_message_factory_methods(self):
         types_list = [
             LdpMessageBody.hello("id", [PayloadMode.TEXT]),
             LdpMessageBody.capability_manifest({"caps": []}),
-            LdpMessageBody.session_propose({"ttl": 3600}),
+            LdpMessageBody.session_propose(
+                {"preferred_payload_modes": ["text"], "trust_domain": "default"}
+            ),
             LdpMessageBody.session_accept("s1", PayloadMode.TEXT),
             LdpMessageBody.session_reject("no"),
             LdpMessageBody.task_submit("t1", "echo", {"data": 1}),
@@ -715,3 +721,114 @@ def test_backward_compat_no_verification_fields(self):
         p = Provenance.model_validate(old)
         assert p.verification_status == VerificationStatus.UNVERIFIED
         assert p.lineage == []
+
+
+class TestSessionProposeConfig:
+    """The typed SESSION_PROPOSE config (issue #18): strict trust_domain,
+    tolerant payload-mode handling so forward-compat peers still negotiate."""
+
+    def test_valid_config(self):
+        c = SessionProposeConfig(
+            preferred_payload_modes=[PayloadMode.SEMANTIC_GRAPH, PayloadMode.TEXT],
+            ttl_secs=120,
+            trust_domain="default",
+        )
+        assert c.preferred_payload_modes == [PayloadMode.SEMANTIC_GRAPH, PayloadMode.TEXT]
+        assert c.ttl_secs == 120
+        assert c.trust_domain == "default"
+
+    def test_trust_domain_is_required(self):
+        with pytest.raises(ValidationError):
+            SessionProposeConfig(preferred_payload_modes=[PayloadMode.TEXT])
+
+    def test_string_modes_are_coerced(self):
+        c = SessionProposeConfig(
+            preferred_payload_modes=["semantic_graph", "text"], trust_domain="d"
+        )
+        assert c.preferred_payload_modes == [PayloadMode.SEMANTIC_GRAPH, PayloadMode.TEXT]
+
+    def test_unknown_modes_are_dropped(self):
+        # Forward-compat: a peer on a newer mode set must not fail the handshake.
+        c = SessionProposeConfig(
+            preferred_payload_modes=["wormhole_mode", "semantic_graph", "text"],
+            trust_domain="d",
+        )
+        assert c.preferred_payload_modes == [PayloadMode.SEMANTIC_GRAPH, PayloadMode.TEXT]
+
+    def test_non_string_items_are_dropped(self):
+        c = SessionProposeConfig(
+            preferred_payload_modes=[{"mode": "semantic_graph"}, ["text"], 3, "text"],
+            trust_domain="d",
+        )
+        assert c.preferred_payload_modes == [PayloadMode.TEXT]
+
+    def test_non_list_modes_becomes_empty(self):
+        c = SessionProposeConfig(preferred_payload_modes=3, trust_domain="d")
+        assert c.preferred_payload_modes == []
+
+    def test_modes_default_to_empty(self):
+        assert SessionProposeConfig(trust_domain="d").preferred_payload_modes == []
+
+    def test_parses_through_envelope_boundary(self):
+        # The point of #18: a malformed-but-tolerable config validates at the
+        # wire boundary, dropping the unknown mode rather than crashing.
+        env = LdpEnvelope.create(
+            "s1",
+            "client",
+            "delegate",
+            body=LdpMessageBody.session_propose(
+                {"preferred_payload_modes": ["bogus", "text"], "trust_domain": "default"}
+            ),
+        )
+        parsed = LdpEnvelope.model_validate(env.model_dump(by_alias=True))
+        assert parsed.body.config is not None
+        assert parsed.body.config.preferred_payload_modes == [PayloadMode.TEXT]
+        assert parsed.body.config.trust_domain == "default"
+
+    def test_envelope_missing_trust_domain_is_rejected(self):
+        # Composes with #19: this raises at the wire boundary, and the HTTP
+        # route maps the ValidationError to a 400 rather than a 500.
+        wire = {
+            "session_id": "s1",
+            "from": "client",
+            "to": "delegate",
+            "body": {"type": "SESSION_PROPOSE", "config": {"preferred_payload_modes": ["text"]}},
+        }
+        with pytest.raises(ValidationError):
+            LdpEnvelope.model_validate(wire)
+
+    def test_blank_trust_domain_is_rejected(self):
+        # A blank or whitespace-only trust_domain must not slip through the typed
+        # boundary and reach the trust check as "" — reject it as malformed,
+        # mirroring TrustDomain.name's own non-blank rule.
+        with pytest.raises(ValidationError):
+            SessionProposeConfig(trust_domain="")
+        with pytest.raises(ValidationError):
+            SessionProposeConfig(trust_domain="   \t")
+
+    def test_non_positive_ttl_is_rejected(self):
+        with pytest.raises(ValidationError):
+            SessionProposeConfig(trust_domain="d", ttl_secs=0)
+        with pytest.raises(ValidationError):
+            SessionProposeConfig(trust_domain="d", ttl_secs=-1)
+
+    def test_factory_dict_missing_trust_domain_raises(self):
+        # The session_propose(dict) convenience path coerces + validates on
+        # construction, so the missing-trust_domain contract holds there too.
+        with pytest.raises(ValidationError):
+            LdpMessageBody.session_propose({"preferred_payload_modes": ["text"]})
+
+    def test_session_propose_requires_config(self):
+        # trust_domain-required, enforced end to end: a config-less SESSION_PROPOSE
+        # is rejected at the wire boundary (becomes a 400 via the route), instead
+        # of silently negotiating as the "unknown" trust domain.
+        with pytest.raises(ValidationError):
+            LdpMessageBody(type="SESSION_PROPOSE")
+        wire = {
+            "session_id": "s1",
+            "from": "client",
+            "to": "delegate",
+            "body": {"type": "SESSION_PROPOSE"},
+        }
+        with pytest.raises(ValidationError):
+            LdpEnvelope.model_validate(wire)