diff --git a/.changeset/python-server-choices-interaction.md b/.changeset/python-server-choices-interaction.md
new file mode 100644
index 00000000..58712fac
--- /dev/null
+++ b/.changeset/python-server-choices-interaction.md
@@ -0,0 +1,9 @@
+---
+'@smooai/smooth-operator-server': minor
+---
+
+Port the Rich Interactions runtime + the `choices` kind (AskUserQuestion) to the **Python** server (Wave 2 of the polyglot effort), mirroring the Rust reference (PR #475).
+
+New kind-agnostic framework (`interaction.py`): `InteractionKind` (kind / capability / tool_schema / parse_request / validate / fallback_directive), an `InteractionRegistry` catalog (default: `choices`), and a session-keyed `PendingInteractions` park/resume registry that generalizes the write-confirmation `ConfirmationRegistry`. Each turn registers per-kind `request_<kind>` raise tools — parking on a channel that declared the kind's render capability in `supports` (emit `interaction_required`, await `submit_interaction`, resume with the canonical payload), or degrading to the kind's conversational directive on text-only channels, where the model submits through the generic `submit_interaction` tool. A new `submit_interaction` dispatcher action routes values to the kind validator: invalid values emit retryable `interaction_invalid` (turn stays parked), valid values resume the turn.
+
+The `choices` kind (`choices.py`) mirrors `choices.rs`: `request_choices { questions (1–4), reason }` with 2–4 options and an optional `multiSelect`, the shared `validate_choices` (every question answered, labels ∈ options, single-select one pick XOR `other`, multi-select ≥1, blank `other` dropped, all errors in one pass), the enumerated fallback directive, and capability id `choice_chips`. Validated against the shared `spec/interactions/choices.schema.json` + conformance fixtures.
diff --git a/python/server/src/smooth_operator_server/__init__.py b/python/server/src/smooth_operator_server/__init__.py
index 187ca2dc..4e7cf3fe 100644
--- a/python/server/src/smooth_operator_server/__init__.py
+++ b/python/server/src/smooth_operator_server/__init__.py
@@ -25,8 +25,17 @@
     Principal,
 )
 from .backplane import Backplane, InMemoryBackplane
+from .choices import ChoicesKind, validate_choices
 from .coding_tools import coding_tools, coding_tools_from_env, resolve_workspace_path
 from .dispatcher import FrameDispatcher
+from .interaction import (
+    InteractionFieldError,
+    InteractionKind,
+    InteractionOutcome,
+    InteractionRegistry,
+    InteractionRequest,
+    PendingInteractions,
+)
 from .otp import (
     OtpChannel,
     OtpContact,
@@ -67,6 +76,14 @@
     "coding_tools_from_env",
     "resolve_workspace_path",
     "FrameDispatcher",
+    "ChoicesKind",
+    "validate_choices",
+    "InteractionFieldError",
+    "InteractionKind",
+    "InteractionOutcome",
+    "InteractionRegistry",
+    "InteractionRequest",
+    "PendingInteractions",
     "OtpChannel",
     "OtpContact",
     "OtpDelivery",
diff --git a/python/server/src/smooth_operator_server/choices.py b/python/server/src/smooth_operator_server/choices.py
new file mode 100644
index 00000000..38b27e3a
--- /dev/null
+++ b/python/server/src/smooth_operator_server/choices.py
@@ -0,0 +1,283 @@
+"""Choices — a structured multiple-choice ask (modeled on Claude Code's
+``AskUserQuestion``): the reference **Rich Interaction kind** (see :mod:`interaction`
+and the Rust reference ``rust/smooth-operator/src/choices.rs``, mirrored exactly).
+
+The agent asks 1–4 short questions, each with 2–4 labeled options; the turn parks until
+the visitor picks. Every question also carries an implicit free-text **"Other"** escape
+hatch, so the visitor can answer outside the enumerated options (exactly as
+``AskUserQuestion`` always offers "Other").
+
+- On a channel that declared the ``choice_chips`` capability, the ``request_choices``
+  tool parks the turn and the server emits ``interaction_required { kind: "choices" }``;
+  the client's chip/menu card resumes with a ``submit_interaction`` action.
+- On a **text-only** channel the same raise degrades to a conversational directive that
+  enumerates the questions + options, and the model submits the picks through the generic
+  ``submit_interaction`` *tool*.
+
+Both paths validate through :func:`validate_choices` — one implementation, one behavior —
+and resume the turn with the same structured payload.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from .interaction import InteractionFieldError, InteractionKind, InteractionRequest
+
+#: Max length of a question's short ``header`` label (chip/tab caption).
+HEADER_MAX_CHARS = 12
+
+
+def _selection_count(answer: dict[str, Any]) -> int:
+    """Total picks in a normalized answer: selected labels + one for a non-blank
+    "Other"."""
+    return len(answer.get("options", [])) + (1 if answer.get("other") else 0)
+
+
+def _normalize_answer(raw: dict[str, Any]) -> dict[str, Any]:
+    """Trim the header + labels, drop blank labels, and collapse a blank/whitespace
+    "Other" to absent (mirrors the Rust normalization pass)."""
+    header = str(raw.get("header", "")).strip()
+    options = [o.strip() for o in raw.get("options", []) if isinstance(o, str) and o.strip()]
+    other_raw = raw.get("other")
+    other = other_raw.strip() if isinstance(other_raw, str) and other_raw.strip() else None
+    normalized: dict[str, Any] = {"header": header, "options": options}
+    if other is not None:
+        normalized["other"] = other
+    return normalized
+
+
+def validate_choices(
+    questions: list[dict[str, Any]], values: dict[str, Any]
+) -> tuple[dict[str, Any] | None, list[InteractionFieldError]]:
+    """Validate submitted ``values`` against the raised ``questions``, returning
+    ``(normalized_values, [])`` or ``(None, errors)`` with every per-question failure.
+
+    Rules (mirrors ``choices.rs`` byte-for-byte):
+    - **every** question must be answered (a selection or a non-blank "Other");
+    - each selected label must be one of that question's option labels;
+    - single-select (``multiSelect: false``): exactly one pick (one label XOR "Other");
+      multi-select: one or more picks (labels and/or "Other");
+    - a blank/whitespace "Other" is treated as absent; labels are trimmed.
+
+    When ``questions`` is empty (a prior-turn fallback raise whose spec is gone),
+    validation degrades to **format-only**: labels can't be checked for membership, so any
+    answer with at least one pick is accepted as-is."""
+    normalized = [_normalize_answer(a) for a in values.get("answers", []) if isinstance(a, dict)]
+
+    # Format-only path: no spec to check membership/required-ness against.
+    if not questions:
+        errors: list[InteractionFieldError] = []
+        for answer in normalized:
+            if _selection_count(answer) == 0:
+                errors.append(InteractionFieldError(answer["header"], "select an option or provide an 'other' answer"))
+        if not normalized:
+            errors.append(InteractionFieldError("answers", "provide an answer for each question, or declined=true"))
+        if errors:
+            return None, errors
+        return {"answers": normalized}, []
+
+    errors = []
+    out: list[dict[str, Any]] = []
+    for question in questions:
+        header = question.get("header", "")
+        answer = next((a for a in normalized if a["header"] == header), None)
+        if answer is None:
+            errors.append(InteractionFieldError(header, "this question must be answered"))
+            continue
+
+        # Every selected label must be one of the enumerated options.
+        option_labels = [o.get("label") for o in question.get("options", []) if isinstance(o, dict)]
+        bad_label = False
+        for label in answer["options"]:
+            if label not in option_labels:
+                bad_label = True
+                errors.append(InteractionFieldError(header, f"'{label}' is not one of the offered options"))
+
+        count = _selection_count(answer)
+        if count == 0:
+            errors.append(InteractionFieldError(header, "select an option or provide an 'other' answer"))
+        elif not question.get("multiSelect", False) and count > 1:
+            errors.append(InteractionFieldError(header, "this question takes a single answer"))
+
+        if not bad_label:
+            out.append(answer)
+
+    if errors:
+        return None, errors
+    return {"answers": out}, []
+
+
+def parse_questions(raw: Any) -> list[dict[str, Any]]:
+    """Parse the raise tool's ``questions`` argument into validated question dicts.
+
+    Enforces the LLM-facing contract so the model produces usable cards: 1–4 questions,
+    each with a non-empty prompt, a non-empty header ≤12 chars (unique within the raise),
+    and 2–4 options with non-empty labels. Raises :class:`ValueError` on any violation
+    (the engine surfaces the text to the model). Mirrors ``choices.rs::parse_questions``,
+    including the shorthand where a bare string is accepted as an option label."""
+    if not isinstance(raw, list):
+        raise ValueError("'questions' must be an array")
+    if not 1 <= len(raw) <= 4:
+        raise ValueError("'questions' must contain between 1 and 4 questions")
+    questions: list[dict[str, Any]] = []
+    seen_headers: list[str] = []
+    for item in raw:
+        if not isinstance(item, dict):
+            raise ValueError("each question must be an object")
+        question = str(item.get("question") or "").strip()
+        if not question:
+            raise ValueError("each question needs a non-empty 'question'")
+        header = str(item.get("header") or "").strip()
+        if not header:
+            raise ValueError("each question needs a non-empty 'header'")
+        if len(header) > HEADER_MAX_CHARS:
+            raise ValueError(f"header '{header}' is too long (max {HEADER_MAX_CHARS} characters)")
+        if header in seen_headers:
+            raise ValueError(f"duplicate question header '{header}'")
+        seen_headers.append(header)
+
+        raw_options = item.get("options")
+        if not isinstance(raw_options, list):
+            raise ValueError(f"question '{header}' needs an 'options' array")
+        if not 2 <= len(raw_options) <= 4:
+            raise ValueError(f"question '{header}' must offer between 2 and 4 options")
+        options: list[dict[str, str]] = []
+        for opt in raw_options:
+            # Accept the object form `{ label, description? }` and the shorthand bare
+            # string the model sometimes emits.
+            if isinstance(opt, str):
+                option = {"label": opt.strip(), "description": ""}
+            elif isinstance(opt, dict):
+                option = {
+                    "label": str(opt.get("label") or "").strip(),
+                    "description": str(opt.get("description") or "").strip(),
+                }
+            else:
+                raise ValueError(f"invalid option entry in '{header}': {opt!r}")
+            if not option["label"]:
+                raise ValueError(f"an option in '{header}' has an empty label")
+            options.append(option)
+
+        parsed: dict[str, Any] = {"question": question, "header": header, "options": options}
+        if item.get("multiSelect") is True:
+            parsed["multiSelect"] = True
+        questions.append(parsed)
+    return questions
+
+
+class ChoicesKind(InteractionKind):
+    """The ``choices`` Rich Interaction kind — a structured multiple-choice ask modeled on
+    ``AskUserQuestion`` (see the module docs and
+    ``spec/interactions/choices.schema.json``)."""
+
+    def kind(self) -> str:
+        return "choices"
+
+    def capability(self) -> str:
+        return "choice_chips"
+
+    def tool_schema(self) -> dict[str, Any]:
+        return {
+            "name": "request_choices",
+            "description": (
+                "Ask the visitor a structured multiple-choice question (1–4 questions, each with "
+                "2–4 labeled options) and wait for their pick. On channels that can render "
+                "chips/menus the visitor taps an option; on text channels you will be told to "
+                "enumerate the options and accept a natural-language answer. An implicit free-text "
+                '"Other" is always available, so use this whenever the answer is likely (but not '
+                "certainly) one of a small set — never free-form the menu yourself."
+            ),
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "questions": {
+                        "type": "array",
+                        "minItems": 1,
+                        "maxItems": 4,
+                        "description": "The questions to ask, in order (1–4).",
+                        "items": {
+                            "type": "object",
+                            "properties": {
+                                "question": {
+                                    "type": "string",
+                                    "description": "The question prompt shown to the visitor.",
+                                },
+                                "header": {
+                                    "type": "string",
+                                    "maxLength": HEADER_MAX_CHARS,
+                                    "description": "A short label (≤12 chars), unique within the raise. Used as the answer key and the chip/tab caption.",
+                                },
+                                "options": {
+                                    "type": "array",
+                                    "minItems": 2,
+                                    "maxItems": 4,
+                                    "description": "The 2–4 options to offer. A free-text 'Other' is always available in addition.",
+                                    "items": {
+                                        "type": "object",
+                                        "properties": {
+                                            "label": {
+                                                "type": "string",
+                                                "description": "The option label (the value submitted).",
+                                            },
+                                            "description": {
+                                                "type": "string",
+                                                "description": "A short gloss for the option.",
+                                            },
+                                        },
+                                        "required": ["label"],
+                                    },
+                                },
+                                "multiSelect": {
+                                    "type": "boolean",
+                                    "description": "Allow selecting more than one option (default false).",
+                                },
+                            },
+                            "required": ["question", "header", "options"],
+                        },
+                    },
+                    "reason": {
+                        "type": "string",
+                        "description": 'Why you\'re asking, phrased for the visitor (e.g. "to route you to the right team").',
+                    },
+                },
+                "required": ["questions", "reason"],
+            },
+        }
+
+    def parse_request(self, args: dict[str, Any]) -> InteractionRequest:
+        questions = parse_questions(args.get("questions"))
+        reason = str(args.get("reason") or "").strip() or "to help you better"
+        return InteractionRequest(kind=self.kind(), spec={"questions": questions}, reason=reason)
+
+    def validate(
+        self, spec: dict[str, Any] | None, values: dict[str, Any]
+    ) -> tuple[dict[str, Any] | None, list[InteractionFieldError]]:
+        questions = (spec or {}).get("questions") or []
+        if not isinstance(values, dict):
+            return None, [InteractionFieldError("values", "invalid values shape: expected an object")]
+        return validate_choices(questions, values)
+
+    def fallback_directive(self, spec: dict[str, Any], reason: str) -> str:
+        lines = []
+        for q in spec.get("questions", []):
+            question = q.get("question")
+            if not question:
+                continue
+            header = q.get("header") or question
+            multi = q.get("multiSelect", False)
+            opts = ", ".join(o.get("label", "") for o in q.get("options", []) if isinstance(o, dict))
+            suffix = " (choose one or more)" if multi else ""
+            lines.append(f"- [{header}] {question} Options: {opts}{suffix}.")
+        enumerated = "\n".join(lines)
+        return (
+            "This visitor's channel cannot display choice chips. Ask the following question(s) "
+            f"conversationally, naturally weaving in the reason ({reason}), and read out each option "
+            f"so the visitor can pick:\n{enumerated}\nThe visitor may also answer with something not "
+            "listed (that's fine — capture it as their 'other' answer). When you have their pick(s), "
+            'call the `submit_interaction` tool with kind "choices" and `values.answers` — one entry '
+            'per question `{ header, options: [chosen label(s)], other?: "their free-text answer" }`. '
+            "It validates each answer and will tell you if a pick isn't offered so you can re-ask. If "
+            "the visitor declines to choose, call `submit_interaction` with declined=true and continue "
+            "helping them."
+        )
diff --git a/python/server/src/smooth_operator_server/dispatcher.py b/python/server/src/smooth_operator_server/dispatcher.py
index a6ad82e1..d7611541 100644
--- a/python/server/src/smooth_operator_server/dispatcher.py
+++ b/python/server/src/smooth_operator_server/dispatcher.py
@@ -32,6 +32,7 @@
 from .auth import AccessContext, normalize_email
 from .backplane import Target
 from .confirmation import ConfirmationRegistry
+from .interaction import InteractionOutcome, InteractionRegistry, PendingInteractions
 from .otp import OtpContact, OtpInvalid, OtpService, OtpVerified
 from .session_store import SessionStore
 from .turn_runner import Sink, TurnContext, TurnRunner
@@ -56,6 +57,8 @@ def __init__(
         tools: list[Any] | None = None,
         confirm_tools: list[str] | None = None,
         confirmations: ConfirmationRegistry | None = None,
+        interactions: InteractionRegistry | None = None,
+        interaction_pending: PendingInteractions | None = None,
         agent_config_resolver: AgentConfigResolver | None = None,
         session_authenticator: SessionAuthenticator | None = None,
         judge_model: str | None = None,
@@ -75,6 +78,17 @@ def __init__(
         #: `confirm_tool_action` frame resolves the future a parked turn awaits.
         #: Created on demand (one per connection) when HITL is enabled.
         self._confirmations = confirmations if confirmations is not None else ConfirmationRegistry()
+        #: Rich Interactions catalog (default: the reference ``choices`` kind) + the
+        #: session-keyed park registry. A turn registers the per-kind raise tools; a
+        #: ``submit_interaction`` frame resolves the park a raise tool awaits — always on
+        #: this same connection, so (like confirmations) the registry is connection-local.
+        self._interactions = interactions if interactions is not None else InteractionRegistry.default()
+        self._interaction_pending = interaction_pending if interaction_pending is not None else PendingInteractions()
+        #: ``supports`` (client render capabilities) declared per session at
+        #: ``create_conversation_session``, connection-local: park/resume happen on this
+        #: connection, so this need not be persisted. Gates each kind's rich-vs-fallback
+        #: path per turn.
+        self._session_supports: dict[str, list[str]] = {}
         #: Per-agent config resolver (SMOODEV-590). Resolved per turn from the session's
         #: agent; the default (empty static resolver) returns None → the server-wide
         #: default prompt drives every turn.
@@ -190,6 +204,8 @@ async def dispatch(self, raw_frame: str, sink: Sink) -> None:
                     await self._handle_send_message(frame, request_id, sink)
             elif action == "confirm_tool_action":
                 self._handle_confirm_tool_action(frame, request_id, sink)
+            elif action == "submit_interaction":
+                await self._handle_submit_interaction(frame, request_id, sink)
             elif action == "verify_otp":
                 await self._handle_verify_otp(frame, request_id, sink)
             elif action is None:
@@ -315,6 +331,13 @@ async def _handle_create_session(self, frame: dict, request_id: str | None, sink
             org_id=self._access.principal.org,
         )
         await self._associate_session(session)
+        # Capture the session's declared render capabilities (``supports``), connection-
+        # local, to gate Rich Interactions per turn. Unknown values are kept as-is and
+        # simply never match a kind's capability (forward-compatible). A non-list is
+        # ignored (no capabilities → every kind degrades to its conversational fallback).
+        raw_supports = frame.get("supports")
+        if isinstance(raw_supports, list):
+            self._session_supports[session.session_id] = [s for s in raw_supports if isinstance(s, str)]
         data = {
             "sessionId": session.session_id,
             "conversationId": session.conversation_id,
@@ -536,6 +559,9 @@ async def _handle_send_message(self, frame: dict, request_id: str | None, sink:
             agent_config=agent_config,
             judge_model=self._judge_model,
             org_id=self._access.principal.org,
+            interactions=self._interactions,
+            interaction_pending=self._interaction_pending,
+            capabilities=self._session_supports.get(session_id),
         )
 
         # Run the turn as a background task, NOT awaited inline. A turn that calls a
@@ -731,6 +757,110 @@ def _handle_confirm_tool_action(self, frame: dict, request_id: str | None, sink:
             )
         )
 
+    async def _handle_submit_interaction(self, frame: dict, request_id: str | None, sink: Sink) -> None:
+        """``submit_interaction`` — the single Rich Interactions resume verb: resolve a
+        turn parked on a raised interaction with the visitor's values (or ``declined``).
+
+        Per ``spec/actions/submit-interaction.schema.json`` the client replies with
+        ``{action, sessionId, requestId, interactionId, kind?, values?, declined?}`` to an
+        ``interaction_required`` event. Flow mirrors the Rust ``handle_submit_interaction``
+        exactly: require ``requestId`` + ``sessionId`` + ``interactionId``; the session
+        must be visible; PEEK the park (don't consume) so an invalid submit leaves the
+        turn parked; the ``interactionId`` (and ``kind``, if given) must match the park
+        (else ``INTERACTION_MISMATCH``, so a stale card can't resolve a newer park). A
+        decline resolves with a declined payload; values route to the kind's validator —
+        invalid values emit ``interaction_invalid`` and KEEP the turn parked (retryable,
+        never a terminal ``error``, like ``otp_invalid``); valid values resolve the park
+        with the canonical payload. Continuation is signalled by the resumed streaming
+        sequence; we ack with an ``immediate_response``."""
+        # requestId echoes the originating interaction_required — require it, don't invent.
+        if not request_id:
+            sink(protocol.error(None, "VALIDATION_ERROR", "submit_interaction requires a 'requestId'"))
+            return
+        session_id = frame.get("sessionId")
+        if not session_id:
+            sink(protocol.error(request_id, "VALIDATION_ERROR", "submit_interaction requires a 'sessionId'"))
+            return
+        interaction_id = frame.get("interactionId")
+        if not interaction_id:
+            sink(protocol.error(request_id, "VALIDATION_ERROR", "submit_interaction requires an 'interactionId'"))
+            return
+
+        # Not-yours is reported identically to not-found — no existence oracle.
+        session = await self._visible_session(session_id)
+        if session is None:
+            sink(protocol.error(request_id, "SESSION_NOT_FOUND", f"session '{session_id}' not found"))
+            return
+
+        # Peek (don't consume): an invalid submit must leave the turn parked for a resubmit.
+        pending = self._interaction_pending.peek(session_id)
+        if pending is None:
+            sink(
+                protocol.error(
+                    request_id,
+                    "NO_PENDING_INTERACTION",
+                    f"no interaction is awaiting a submit for session '{session_id}'",
+                )
+            )
+            return
+        if interaction_id != pending.interaction_id:
+            sink(
+                protocol.error(
+                    request_id,
+                    "INTERACTION_MISMATCH",
+                    "interactionId does not match the parked interaction",
+                )
+            )
+            return
+        claimed_kind = frame.get("kind")
+        if isinstance(claimed_kind, str) and claimed_kind and claimed_kind != pending.kind:
+            sink(protocol.error(request_id, "INTERACTION_MISMATCH", "kind does not match the parked interaction"))
+            return
+
+        # Decline: resume the turn with a declined payload so the agent proceeds gracefully.
+        if frame.get("declined") is True:
+            self._interaction_pending.resolve(session_id, InteractionOutcome.declined())
+            sink(protocol.immediate_response(request_id, 200, "Interaction declined", {"sessionId": session_id}))
+            return
+
+        values = frame.get("values")
+        if not isinstance(values, dict):
+            sink(
+                protocol.error(
+                    request_id, "VALIDATION_ERROR", "submit_interaction requires 'values' unless 'declined' is true"
+                )
+            )
+            return
+
+        kind = self._interactions.get(pending.kind)
+        if kind is None:  # a hosted-kind park whose kind was since removed — defensive.
+            sink(protocol.error(request_id, "NO_PENDING_INTERACTION", f"unknown interaction kind '{pending.kind}'"))
+            return
+        canonical, errors = kind.validate(pending.spec, values)
+        if errors:
+            # Retryable: re-render the card with the per-field errors; the turn stays parked.
+            sink(
+                protocol.interaction_invalid(
+                    request_id,
+                    pending.interaction_id,
+                    pending.kind,
+                    [e.to_dict() for e in errors],
+                    "Some fields need attention.",
+                )
+            )
+            return
+
+        # Valid: consume the park and resume the turn with the canonical payload.
+        self._interaction_pending.resolve(session_id, InteractionOutcome.submitted(canonical or {}))
+        sink(
+            protocol.immediate_response(
+                request_id,
+                200,
+                "Interaction submitted",
+                {"sessionId": session_id, "interactionId": pending.interaction_id, "values": canonical},
+            )
+        )
+
 
 #: Default cap for list_conversations when the caller doesn't ask for a specific limit.
 _DEFAULT_LIST_LIMIT = 50
diff --git a/python/server/src/smooth_operator_server/interaction.py b/python/server/src/smooth_operator_server/interaction.py
new file mode 100644
index 00000000..18255fff
--- /dev/null
+++ b/python/server/src/smooth_operator_server/interaction.py
@@ -0,0 +1,241 @@
+"""Rich Interactions — the extensible structured-interaction framework.
+
+One pattern, many kinds (see the Rust reference ``rust/smooth-operator/src/interaction.rs``
+and ``docs/Architecture/Rich Interactions.md``): an agent raises a **structured
+interaction** (identity intake, a date picker, choice chips, …). On a channel whose
+client declared the kind's render capability (``supports`` at
+``create_conversation_session``), the turn parks and the client renders a rich card
+(``interaction_required`` → ``submit_interaction``). On a text-only channel the same
+raise degrades to the kind's **conversational fallback**: a directive the model follows
+turn by turn, submitting through the generic ``submit_interaction`` *tool*. Both paths
+run the kind's **server-side validator** and resume the turn with the same canonical
+payload.
+
+Adding a kind = implementing :class:`InteractionKind` (one module) and registering it in
+an :class:`InteractionRegistry`. No new protocol events, no new client verbs.
+
+The park/resume half (:class:`PendingInteractions`) is the kind-agnostic generalization
+of :mod:`smooth_operator_server.confirmation`: where the confirmation registry parks a
+turn on a ``bool`` verdict keyed by session, this parks a turn on an
+:class:`InteractionOutcome` keyed by session — the async analog of the Rust server's
+``PendingInteraction`` map (``register_interaction`` / ``pending_interaction`` /
+``take_interaction`` / ``clear_interaction``).
+"""
+
+from __future__ import annotations
+
+import asyncio
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Any
+
+#: How long a parked interaction waits for the client's ``submit_interaction`` before the
+#: raise tool unblocks with a ``no_response`` payload. Matches the Rust server's
+#: ``INTERACTION_TIMEOUT`` and the write-confirmation window (300s).
+INTERACTION_TIMEOUT = 300.0
+
+#: The tool-result guidance the raise tool returns for each non-submitted outcome —
+#: byte-for-byte the Rust ``RequestInteractionTool::execute`` messages, so the model
+#: reads the same instruction on every engine.
+DECLINED_MESSAGE = "The visitor declined. Continue helping them without this and do not ask again this conversation."
+NO_RESPONSE_MESSAGE = (
+    "The visitor did not respond to the card. Continue without it; you may offer again later if it becomes relevant."
+)
+
+
+@dataclass(frozen=True)
+class InteractionFieldError:
+    """A single per-field validation failure, carried on ``interaction_invalid`` and in
+    the conversational tool's error result. ``field`` is a kind-specific key (for
+    ``choices`` it is the question ``header``)."""
+
+    field: str
+    message: str
+
+    def to_dict(self) -> dict[str, str]:
+        return {"field": self.field, "message": self.message}
+
+
+@dataclass(frozen=True)
+class InteractionRequest:
+    """A parsed raise: what the agent asked the visitor for. ``spec`` shape per
+    ``spec/interactions/<kind>.schema.json#/$defs/Spec``."""
+
+    kind: str
+    spec: dict[str, Any]
+    reason: str
+
+
+@dataclass(frozen=True)
+class InteractionOutcome:
+    """How a parked interaction resolved — the value the raise tool's awaited future
+    settles to. ``status`` is one of ``submitted`` / ``declined`` / ``no_response``
+    (the ``Payload.status`` enum); ``values`` is present only when submitted."""
+
+    status: str
+    values: dict[str, Any] | None = None
+
+    @classmethod
+    def submitted(cls, values: dict[str, Any]) -> InteractionOutcome:
+        return cls("submitted", values)
+
+    @classmethod
+    def declined(cls) -> InteractionOutcome:
+        return cls("declined")
+
+    @classmethod
+    def no_response(cls) -> InteractionOutcome:
+        return cls("no_response")
+
+    def to_payload(self) -> dict[str, Any]:
+        """The canonical validated payload the parked turn resumes with — the string the
+        raise tool hands back to the model (identical on the card and conversational
+        paths). Matches ``spec/interactions/choices.schema.json#/$defs/Payload``."""
+        if self.status == "submitted":
+            return {"status": "submitted", "values": self.values or {"answers": []}}
+        if self.status == "declined":
+            return {"status": "declined", "message": DECLINED_MESSAGE}
+        return {"status": "no_response", "message": NO_RESPONSE_MESSAGE}
+
+
+class InteractionKind(ABC):
+    """One interaction kind — the extension seam of the Rich Interactions pattern.
+
+    A kind supplies exactly the pieces that differ per interaction; ALL park / resume /
+    event / registry machinery is shared and kind-agnostic: identity (:meth:`kind` /
+    :meth:`capability`), the LLM-facing raise-tool surface (:meth:`tool_schema` +
+    :meth:`parse_request`), the server-side :meth:`validate`, and the conversational
+    :meth:`fallback_directive`."""
+
+    @abstractmethod
+    def kind(self) -> str:
+        """The wire kind id (e.g. ``choices``). Selects the client card + the validator."""
+
+    @abstractmethod
+    def capability(self) -> str:
+        """The client render capability that gates the rich path (e.g. ``choice_chips``).
+        A session that declared it in ``supports`` gets the parked card; anything else
+        gets the conversational fallback."""
+
+    @abstractmethod
+    def tool_schema(self) -> dict[str, Any]:
+        """The raise tool's LLM-facing schema — ``{name, description, parameters}``.
+        Convention: name it ``request_<kind>``."""
+
+    @abstractmethod
+    def parse_request(self, args: dict[str, Any]) -> InteractionRequest:
+        """Parse + canonicalize the raise tool's arguments into the kind's ``spec`` and
+        the human-readable reason. Raises :class:`ValueError` on malformed arguments (the
+        engine surfaces the text to the model)."""
+
+    @abstractmethod
+    def validate(
+        self, spec: dict[str, Any] | None, values: dict[str, Any]
+    ) -> tuple[dict[str, Any] | None, list[InteractionFieldError]]:
+        """Validate ``values`` against ``spec``, returning ``(canonical, [])`` on success
+        or ``(None, errors)`` with EVERY failed field (so a card annotates all in one
+        round-trip). ``spec`` may be ``None`` on the conversational path when the raise
+        happened in an earlier turn — the kind then applies format-only validation."""
+
+    @abstractmethod
+    def fallback_directive(self, spec: dict[str, Any], reason: str) -> str:
+        """The conversational-degradation directive for text-only channels: instructions
+        the model follows to collect the same information turn by turn, then submit
+        through the ``submit_interaction`` tool."""
+
+
+class InteractionRegistry:
+    """The catalog of interaction kinds a server hosts. The default catalog is the
+    reference kind (``choices``); a host may extend or replace it. The Python analog of
+    the Rust ``InteractionRegistry``."""
+
+    def __init__(self) -> None:
+        self._kinds: dict[str, InteractionKind] = {}
+
+    def with_kind(self, kind: InteractionKind) -> InteractionRegistry:
+        """Register a kind (builder). A later registration with the same id replaces the
+        earlier one."""
+        self._kinds[kind.kind()] = kind
+        return self
+
+    def get(self, kind: str) -> InteractionKind | None:
+        return self._kinds.get(kind)
+
+    def kinds(self) -> list[InteractionKind]:
+        """Every registered kind, in registration order."""
+        return list(self._kinds.values())
+
+    @classmethod
+    def default(cls) -> InteractionRegistry:
+        """The reference catalog: ``choices``."""
+        from .choices import ChoicesKind
+
+        return cls().with_kind(ChoicesKind())
+
+
+@dataclass
+class _Pending:
+    """One parked interaction: the id echoed on submit, the kind + spec the submit
+    validates against, and the future the raise tool awaits."""
+
+    interaction_id: str
+    kind: str
+    spec: dict[str, Any]
+    future: asyncio.Future[InteractionOutcome]
+
+
+@dataclass
+class PendingInteractions:
+    """The session-keyed registry of parked interactions — the kind-agnostic
+    generalization of :class:`~smooth_operator_server.confirmation.ConfirmationRegistry`.
+
+    Single-threaded under the asyncio event loop (every method runs on the loop, so the
+    dict needs no locking). At most one outstanding interaction per session; a new park
+    supersedes any prior one (its turn unblocks with ``no_response``)."""
+
+    _pending: dict[str, _Pending] = field(default_factory=dict)
+
+    def register(
+        self, session_id: str, interaction_id: str, kind: str, spec: dict[str, Any]
+    ) -> asyncio.Future[InteractionOutcome]:
+        """Register (and return) a fresh outcome future for ``session_id``. Any prior park
+        is superseded — resolved ``no_response`` so its raise tool can never dangle
+        (mirrors the Rust ``register_interaction`` replacing the prior responder)."""
+        prior = self._pending.pop(session_id, None)
+        if prior is not None and not prior.future.done():
+            prior.future.set_result(InteractionOutcome.no_response())
+        future: asyncio.Future[InteractionOutcome] = asyncio.get_running_loop().create_future()
+        self._pending[session_id] = _Pending(interaction_id, kind, spec, future)
+        return future
+
+    def peek(self, session_id: str) -> _Pending | None:
+        """The parked interaction for ``session_id`` WITHOUT consuming it — the submit
+        handler peeks to validate, so an invalid submit leaves the turn parked."""
+        return self._pending.get(session_id)
+
+    def resolve(self, session_id: str, outcome: InteractionOutcome) -> bool:
+        """Consume the park for ``session_id`` and settle its future with ``outcome``.
+        Returns ``True`` if a park was resolved, ``False`` if none was awaiting (a
+        duplicate/stale submit). Taking it out makes a duplicate a clean no-op (mirrors
+        the Rust ``take_interaction``)."""
+        pending = self._pending.pop(session_id, None)
+        if pending is None or pending.future.done():
+            return False
+        pending.future.set_result(outcome)
+        return True
+
+    def clear(self, session_id: str) -> None:
+        """Drop any park for ``session_id`` (turn ended). Idempotent. Resolves a still-
+        pending future ``no_response`` so a raise tool awaiting at teardown unblocks."""
+        pending = self._pending.pop(session_id, None)
+        if pending is not None and not pending.future.done():
+            pending.future.set_result(InteractionOutcome.no_response())
+
+    def reject_all(self) -> None:
+        """Resolve every outstanding park ``no_response`` (connection torn down) so any
+        turn parked on an interaction unparks and finishes cleanly — fail soft (the agent
+        continues without the answer), never leave a turn hung."""
+        for pending in tuple(self._pending.values()):
+            if not pending.future.done():
+                pending.future.set_result(InteractionOutcome.no_response())
+        self._pending.clear()
diff --git a/python/server/src/smooth_operator_server/interaction_tools.py b/python/server/src/smooth_operator_server/interaction_tools.py
new file mode 100644
index 00000000..6211f6a7
--- /dev/null
+++ b/python/server/src/smooth_operator_server/interaction_tools.py
@@ -0,0 +1,164 @@
+"""Per-turn Rich Interaction tools — the model-facing surface, built fresh each turn.
+
+Two model-callable tools, mirroring the Rust core crate (``tools/interaction.rs``):
+
+- **``request_<kind>``** (one per hosted kind): the raise tool. On a **rich** channel
+  (the kind's capability is in the session's ``supports``) it parks the turn — registers
+  a park, emits ``interaction_required``, and awaits the client's ``submit_interaction``
+  — returning the kind's canonical payload. On a **fallback** channel it returns the
+  kind's conversational directive immediately (no park) and stashes the spec so the
+  generic submit tool can validate it fully.
+- **``submit_interaction``** (one, registered only when ≥1 kind is on the fallback path):
+  the conversational-path submit. Routes the model's values to the kind's validator and
+  returns the same canonical payload — or a tool error the model relays and re-asks on.
+
+The park is folded straight into the raise tool's coroutine (an ``await`` on the park
+future), so — unlike the Rust bridge task + mpsc channel — no extra task is needed: the
+turn coroutine suspends on the future and the connection read loop stays free to receive
+the ``submit_interaction`` action. Same shape as the write-confirmation park.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import uuid
+from typing import Any, Callable
+
+from smooth_operator_core import FunctionTool
+
+from . import protocol
+from .interaction import (
+    INTERACTION_TIMEOUT,
+    InteractionKind,
+    InteractionOutcome,
+    InteractionRegistry,
+    PendingInteractions,
+)
+
+#: The sink type the raise tool emits ``interaction_required`` through.
+Sink = Callable[[dict[str, Any]], None]
+
+#: submit_interaction TOOL messages — byte-for-byte the Rust ``SubmitInteractionTool``.
+_SUBMIT_DECLINED_MESSAGE = "Noted. Continue helping the visitor without this and do not ask again this conversation."
+
+
+def _request_tool(
+    kind: InteractionKind,
+    rich: bool,
+    session_id: str,
+    request_id: str,
+    sink: Sink,
+    pending: PendingInteractions,
+    raised_specs: dict[str, dict[str, Any]],
+) -> FunctionTool:
+    """Build one ``request_<kind>`` raise tool bound to this turn's session/sink."""
+    schema = kind.tool_schema()
+
+    async def _run(args: dict[str, Any]) -> str:
+        request = kind.parse_request(args)  # ValueError → surfaced to the model
+        if not rich:
+            # Fallback: no card can render — hand the model the conversational directive
+            # and stash the spec so the submit tool validates with full required-ness.
+            raised_specs[request.kind] = request.spec
+            return json.dumps(
+                {
+                    "mode": "conversational",
+                    "kind": request.kind,
+                    "spec": request.spec,
+                    "reason": request.reason,
+                    "instructions": kind.fallback_directive(request.spec, request.reason),
+                }
+            )
+        # Rich: park. Register the park, emit the event, await the client's submit.
+        interaction_id = str(uuid.uuid4())
+        future = pending.register(session_id, interaction_id, request.kind, request.spec)
+        sink(protocol.interaction_required(request_id, interaction_id, request.kind, request.spec, request.reason))
+        try:
+            outcome = await asyncio.wait_for(future, INTERACTION_TIMEOUT)
+        except (asyncio.TimeoutError, asyncio.CancelledError):
+            # Our own timeout / turn teardown reads as no answer to the card. Drop any
+            # lingering park so a late submit can't resolve a dead future.
+            pending.clear(session_id)
+            outcome = InteractionOutcome.no_response()
+        return json.dumps(outcome.to_payload())
+
+    return FunctionTool(
+        name=schema["name"],
+        description=schema["description"],
+        parameters=schema["parameters"],
+        func=_run,
+    )
+
+
+def _submit_tool(
+    kinds: InteractionRegistry,
+    raised_specs: dict[str, dict[str, Any]],
+) -> FunctionTool:
+    """Build the generic ``submit_interaction`` tool (conversational-fallback submit)."""
+    hosted = [k.kind() for k in kinds.kinds()]
+
+    async def _run(args: dict[str, Any]) -> str:
+        kind_id = args.get("kind")
+        kind = kinds.get(kind_id) if isinstance(kind_id, str) else None
+        if kind is None:
+            raise ValueError(f"unknown interaction kind {kind_id!r}; hosted kinds: {', '.join(hosted)}")
+        if args.get("declined") is True:
+            return json.dumps({"status": "declined", "message": _SUBMIT_DECLINED_MESSAGE})
+        values = args.get("values")
+        if not isinstance(values, dict):
+            raise ValueError("'values' object is required unless declined=true")
+        # The raise may have happened this turn (spec stashed) or a prior turn (spec gone
+        # → format-only validation).
+        spec = raised_specs.get(kind_id)
+        canonical, errors = kind.validate(spec, values)
+        if errors:
+            detail = "; ".join(f"{e.field}: {e.message}" for e in errors)
+            raise ValueError(
+                f"validation failed — {detail}. Re-ask the visitor for the corrected value(s) and submit again."
+            )
+        return json.dumps({"status": "submitted", "values": canonical})
+
+    return FunctionTool(
+        name="submit_interaction",
+        description=(
+            "Submit the visitor's answer(s) to an interaction you raised conversationally (a "
+            "request_<kind> tool told you the channel cannot render a card). Provide the `kind` and "
+            "the `values` for that kind, or `declined: true` if the visitor refused. Returns the "
+            "validated result, or an error naming what to re-ask."
+        ),
+        parameters={
+            "type": "object",
+            "properties": {
+                "kind": {"type": "string", "enum": hosted, "description": "The interaction kind being answered."},
+                "values": {"type": "object", "description": "The kind-specific submitted values."},
+                "declined": {"type": "boolean", "description": "True when the visitor refused the interaction."},
+            },
+            "required": ["kind"],
+        },
+        func=_run,
+    )
+
+
+def build_interaction_tools(
+    kinds: InteractionRegistry,
+    capabilities: set[str],
+    session_id: str,
+    request_id: str,
+    sink: Sink,
+    pending: PendingInteractions,
+) -> list[FunctionTool]:
+    """Build this turn's Rich Interaction tools: one ``request_<kind>`` per hosted kind
+    (rich when the kind's capability is declared, else fallback), plus the generic
+    ``submit_interaction`` tool when at least one kind is on the fallback path. Mirrors the
+    Rust runner's per-turn registration + ``any_fallback`` gate."""
+    raised_specs: dict[str, dict[str, Any]] = {}
+    tools: list[FunctionTool] = []
+    any_fallback = False
+    for kind in kinds.kinds():
+        rich = kind.capability() in capabilities
+        any_fallback = any_fallback or not rich
+        tools.append(_request_tool(kind, rich, session_id, request_id, sink, pending, raised_specs))
+    if any_fallback:
+        tools.append(_submit_tool(kinds, raised_specs))
+    return tools
diff --git a/python/server/src/smooth_operator_server/protocol.py b/python/server/src/smooth_operator_server/protocol.py
index 39c96f64..f3417caa 100644
--- a/python/server/src/smooth_operator_server/protocol.py
+++ b/python/server/src/smooth_operator_server/protocol.py
@@ -160,6 +160,58 @@ def write_confirmation_required(request_id: str, tool_id: str, action_descriptio
     }
 
 
+def interaction_required(
+    request_id: str, interaction_id: str, kind: str, spec: dict[str, Any], reason: str
+) -> dict[str, Any]:
+    """``interaction_required`` — the Rich Interactions envelope, emitted mid-turn when
+    the agent raises a structured interaction (kind ``choices``, ``identity_intake``, …)
+    on a session that declared the kind's render capability in ``supports``. The turn is
+    **parked** until the client replies with a ``submit_interaction`` action carrying the
+    same ``requestId`` + ``interactionId`` (values or ``declined: true``). Sessions
+    without the capability never receive this — the server degrades that kind to its
+    conversational fallback instead.
+
+    Wire shape matches ``spec/events/interaction-required.schema.json`` and the Rust
+    reference's ``protocol::interaction_required`` byte-for-byte: the prompt detail is
+    double-nested under ``data.data.{interactionId, kind, spec, reason}``.
+    ``interactionId`` is a server-generated id echoed on the submit so a stale card can
+    never resolve a newer park; ``kind`` selects the client card + the server validator;
+    ``spec`` is the kind-specific render payload."""
+    return {
+        "type": "interaction_required",
+        "requestId": request_id,
+        "data": {
+            "requestId": request_id,
+            "data": {"interactionId": interaction_id, "kind": kind, "spec": spec, "reason": reason},
+        },
+        "timestamp": _now_ms(),
+    }
+
+
+def interaction_invalid(
+    request_id: str, interaction_id: str, kind: str, errors: list[dict[str, Any]], message: str
+) -> dict[str, Any]:
+    """``interaction_invalid`` — a ``submit_interaction`` carried values that failed the
+    kind's server-side validation. The turn **stays parked** (the client re-renders the
+    card with the per-field errors and lets the visitor resubmit); like ``otp_invalid``,
+    invalid input is a retryable state, never a terminal ``error`` event.
+
+    Wire shape matches ``spec/events/interaction-invalid.schema.json`` and the Rust
+    reference byte-for-byte: the detail is double-nested under
+    ``data.data.{interactionId, kind, errors, message}``, where each ``errors`` entry is
+    ``{field, message}`` (``field`` is a kind-specific key — for ``choices`` the question
+    ``header``)."""
+    return {
+        "type": "interaction_invalid",
+        "requestId": request_id,
+        "data": {
+            "requestId": request_id,
+            "data": {"interactionId": interaction_id, "kind": kind, "errors": errors, "message": message},
+        },
+        "timestamp": _now_ms(),
+    }
+
+
 def otp_verification_required(
     request_id: str,
     tool_id: str,
diff --git a/python/server/src/smooth_operator_server/server.py b/python/server/src/smooth_operator_server/server.py
index 49f77340..f5d22ac1 100644
--- a/python/server/src/smooth_operator_server/server.py
+++ b/python/server/src/smooth_operator_server/server.py
@@ -37,6 +37,7 @@
 from .coding_tools import coding_tools_from_env
 from .confirmation import ConfirmationRegistry
 from .dispatcher import FrameDispatcher
+from .interaction import InteractionRegistry, PendingInteractions
 from .otp import OtpService
 from .session_store import InMemorySessionStore, SessionStore
 from .workflow import WORKFLOW_JUDGE_MODEL
@@ -88,6 +89,11 @@ class ServerState:
     otp_service: OtpService | None = None
     #: Fast/cheap model for the post-turn workflow judge (default haiku-tier).
     judge_model: str = WORKFLOW_JUDGE_MODEL
+    #: Rich Interactions catalog — the kinds this server hosts (default: the reference
+    #: ``choices`` kind). Each turn registers the per-kind ``request_<kind>`` raise tools;
+    #: a client submits back through the ``submit_interaction`` action / tool. Extend or
+    #: replace to host more kinds.
+    interactions: InteractionRegistry = field(default_factory=InteractionRegistry.default)
     cancel: asyncio.Event = field(default_factory=asyncio.Event)
 
 
@@ -147,6 +153,10 @@ async def associate(target: Target) -> None:
     # frame and the parked turn it resumes are always on the same connection (the
     # session id keys within it), so the registry need not be server-wide.
     confirmations = ConfirmationRegistry()
+    # One pending-interaction registry per connection: a `submit_interaction` frame and
+    # the parked turn it resumes are always on the same connection (session-keyed within
+    # it), so — like confirmations — it need not be server-wide.
+    interaction_pending = PendingInteractions()
     dispatcher = FrameDispatcher(
         state.store,
         state.chat_client,
@@ -157,6 +167,8 @@ async def associate(target: Target) -> None:
         tools=state.tools,
         confirm_tools=state.confirm_tools,
         confirmations=confirmations,
+        interactions=state.interactions,
+        interaction_pending=interaction_pending,
         agent_config_resolver=state.agent_config_resolver,
         session_authenticator=state.session_authenticator,
         judge_model=state.judge_model,
@@ -207,6 +219,9 @@ async def associate(target: Target) -> None:
         # graceful-drain "in-flight turn finishes" contract now that turns run as
         # background tasks rather than inline).
         confirmations.reject_all()
+        # Likewise unpark any turn parked on a raised interaction (fail soft — the agent
+        # continues without the answer) so its `eventual_response` flushes before drain.
+        interaction_pending.reject_all()
         await dispatcher.wait_for_turns()
         # Stop the writer (drain any already-queued events first), then detach —
         # the detach-after-loop runs regardless of how the loop exited.
diff --git a/python/server/src/smooth_operator_server/turn_runner.py b/python/server/src/smooth_operator_server/turn_runner.py
index 8d624e08..c072a863 100644
--- a/python/server/src/smooth_operator_server/turn_runner.py
+++ b/python/server/src/smooth_operator_server/turn_runner.py
@@ -39,6 +39,8 @@
 from .agent_config import AgentConfig, filter_tools
 from .confirmation import ConfirmationRegistry
 from .extensions import build_extension_host
+from .interaction import InteractionRegistry, PendingInteractions
+from .interaction_tools import build_interaction_tools
 from .model_info import model_output_ceiling
 from .session_store import MessageDirection, SessionStore
 from .workflow import (
@@ -248,6 +250,9 @@ def __init__(
         tool_hooks: list[Any] | None = None,
         org_id: str | None = None,
         executor: AgentExecutor | None = None,
+        interactions: InteractionRegistry | None = None,
+        interaction_pending: PendingInteractions | None = None,
+        capabilities: list[str] | None = None,
     ) -> None:
         self._chat_client = chat_client
         self._store = store
@@ -280,6 +285,16 @@ def __init__(
         #: span so the observability studio groups turns by org (mirrors the Rust
         #: runner's ``org_id`` span field). ``None`` ⇒ the attribute is omitted.
         self._org_id = org_id
+        #: Rich Interactions catalog + the session-keyed park registry (shared with the
+        #: dispatcher's ``submit_interaction`` handler). When both are wired, this turn
+        #: registers the per-kind ``request_<kind>`` raise tools + the generic
+        #: ``submit_interaction`` fallback tool. ``None`` ⇒ no interaction tools (behavior
+        #: unchanged).
+        self._interactions = interactions
+        self._interaction_pending = interaction_pending
+        #: The client render capabilities this session declared in ``supports`` — gates
+        #: whether each kind's raise tool parks (rich) or degrades (fallback).
+        self._capabilities = set(capabilities or [])
 
     def _is_gated(self, tool_name: str) -> bool:
         """True when ``tool_name`` matches a confirmation-gated pattern (substring,
@@ -328,6 +343,25 @@ async def run(
         if ext_turn is not None:
             agent_tools.extend(filter_tools(ext_turn.host.tools(), self._agent_config))
 
+        # Rich Interactions: register this turn's per-kind `request_<kind>` raise tools
+        # (rich when the session declared the kind's capability, else conversational
+        # fallback) + the generic `submit_interaction` tool. Framework tools, appended
+        # raw (not through the agent allow-list) so they are always available, mirroring
+        # the Rust runner's unconditional registration. `interaction_required` events
+        # emit through this connection's sink; a rich raise parks on
+        # `_interaction_pending` until the dispatcher's `submit_interaction` resolves it.
+        if self._interactions is not None and self._interaction_pending is not None:
+            agent_tools.extend(
+                build_interaction_tools(
+                    self._interactions,
+                    self._capabilities,
+                    confirm_session,
+                    request_id,
+                    sink,
+                    self._interaction_pending,
+                )
+            )
+
         # 1. Build the agent. The knowledge base (when present) auto-injects the
         #    top hits into the system prompt — the engine handles retrieval + rerank
         #    internally, mirroring the C# `new SmoothAgent(..., Knowledge = ...)`.
@@ -532,6 +566,11 @@ async def _gate(req: HumanApprovalRequest) -> HumanApprovalResponse:
             # `(cfg.clear)(session_id)` at turn end). No-op when HITL is off.
             if self._confirmations is not None:
                 self._confirmations.clear(confirm_session)
+            # Drop any interaction still parked (mirrors the Rust `(cfg.clear)` at turn
+            # end): a raise tool awaiting at teardown unblocks `no_response`, and a stale
+            # registration can't mis-route a later `submit_interaction`. No-op when off.
+            if self._interaction_pending is not None:
+                self._interaction_pending.clear(confirm_session)
             # SEP — stop the extension subprocesses this turn spawned (and clear any
             # ui/confirm still parked). No-op when no host was built (default deny).
             if ext_turn is not None:
diff --git a/python/server/tests/test_choices_validator.py b/python/server/tests/test_choices_validator.py
new file mode 100644
index 00000000..116d167c
--- /dev/null
+++ b/python/server/tests/test_choices_validator.py
@@ -0,0 +1,166 @@
+"""Unit tests for the ``choices`` Rich Interaction kind's validator + parser.
+
+Mirrors the Rust reference tests in ``rust/smooth-operator/src/choices.rs``, and
+additionally validates the shared conformance fixtures
+(``spec/conformance/fixtures.json``) so the Python validator agrees with the golden
+spec every language checks against.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from smooth_operator_server.choices import ChoicesKind, parse_questions, validate_choices
+
+_SPEC = Path(__file__).resolve().parents[3] / "spec" / "conformance" / "fixtures.json"
+
+
+@pytest.fixture(scope="module")
+def fixtures() -> dict:
+    return json.loads(_SPEC.read_text())
+
+
+def _q(header: str, labels: list[str], multi: bool = False) -> dict:
+    return {
+        "question": f"{header}?",
+        "header": header,
+        "options": [{"label": label, "description": ""} for label in labels],
+        **({"multiSelect": True} if multi else {}),
+    }
+
+
+def _a(header: str, options: list[str], other: str | None = None) -> dict:
+    ans: dict = {"header": header, "options": options}
+    if other is not None:
+        ans["other"] = other
+    return ans
+
+
+def test_valid_single_select_normalizes() -> None:
+    canonical, errors = validate_choices([_q("Plan", ["Basic", "Pro"])], {"answers": [_a("Plan", ["  Pro  "])]})
+    assert errors == []
+    assert canonical == {"answers": [{"header": "Plan", "options": ["Pro"]}]}
+
+
+def test_valid_multi_select_keeps_all_picks() -> None:
+    canonical, errors = validate_choices(
+        [_q("Topics", ["Sales", "Support", "Billing"], multi=True)],
+        {"answers": [_a("Topics", ["Sales", "Billing"])]},
+    )
+    assert errors == []
+    assert canonical["answers"][0]["options"] == ["Sales", "Billing"]
+
+
+def test_other_escape_hatch_is_accepted() -> None:
+    canonical, errors = validate_choices(
+        [_q("Plan", ["Basic", "Pro"])], {"answers": [_a("Plan", [], other="  Enterprise, actually  ")]}
+    )
+    assert errors == []
+    assert canonical["answers"][0]["options"] == []
+    assert canonical["answers"][0]["other"] == "Enterprise, actually"
+
+
+def test_unknown_label_is_a_field_error() -> None:
+    canonical, errors = validate_choices([_q("Plan", ["Basic", "Pro"])], {"answers": [_a("Plan", ["Platinum"])]})
+    assert canonical is None
+    assert len(errors) == 1
+    assert errors[0].field == "Plan"
+    assert "not one of the offered" in errors[0].message
+
+
+def test_single_select_rejects_multiple_picks() -> None:
+    _, errors = validate_choices([_q("Plan", ["Basic", "Pro"])], {"answers": [_a("Plan", ["Basic", "Pro"])]})
+    assert any("single answer" in e.message for e in errors)
+
+
+def test_unanswered_question_is_required() -> None:
+    _, errors = validate_choices(
+        [_q("Plan", ["Basic", "Pro"]), _q("Size", ["S", "M"])], {"answers": [_a("Plan", ["Pro"])]}
+    )
+    assert len(errors) == 1
+    assert errors[0].field == "Size"
+    assert "must be answered" in errors[0].message
+
+
+def test_empty_answer_needs_a_pick_or_other() -> None:
+    _, errors = validate_choices([_q("Plan", ["Basic", "Pro"])], {"answers": [_a("Plan", [])]})
+    assert any("select an option" in e.message for e in errors)
+
+
+def test_format_only_when_spec_is_gone() -> None:
+    # No questions (prior-turn fallback raise) → membership can't be checked; any answer
+    # with a pick is accepted as-is.
+    canonical, errors = validate_choices([], {"answers": [_a("Plan", ["Anything"])]})
+    assert errors == []
+    assert canonical["answers"][0]["options"] == ["Anything"]
+    # …but a pickless answer still fails.
+    _, errors2 = validate_choices([], {"answers": [_a("Plan", [])]})
+    assert any("select an option" in e.message for e in errors2)
+
+
+def test_parse_questions_enforces_the_contract() -> None:
+    # Happy path with shorthand string options.
+    qs = parse_questions([{"question": "Which plan?", "header": "Plan", "options": ["Basic", "Pro"]}])
+    assert len(qs) == 1
+    assert qs[0]["options"][0]["label"] == "Basic"
+    assert "multiSelect" not in qs[0]
+
+    with pytest.raises(ValueError):  # too many questions
+        parse_questions([{"question": "q", "header": f"H{i}", "options": ["a", "b"]} for i in range(5)])
+    with pytest.raises(ValueError):  # too few options
+        parse_questions([{"question": "q", "header": "H", "options": ["only"]}])
+    with pytest.raises(ValueError):  # header too long
+        parse_questions([{"question": "q", "header": "ThisHeaderIsWayTooLong", "options": ["a", "b"]}])
+    with pytest.raises(ValueError):  # duplicate headers
+        parse_questions(
+            [
+                {"question": "q1", "header": "H", "options": ["a", "b"]},
+                {"question": "q2", "header": "H", "options": ["a", "b"]},
+            ]
+        )
+
+
+def test_kind_wires_the_reference_surface() -> None:
+    kind = ChoicesKind()
+    assert kind.kind() == "choices"
+    assert kind.capability() == "choice_chips"
+    assert kind.tool_schema()["name"] == "request_choices"
+
+    req = kind.parse_request(
+        {
+            "questions": [
+                {
+                    "question": "Which plan interests you?",
+                    "header": "Plan",
+                    "options": [{"label": "Basic"}, {"label": "Pro"}],
+                }
+            ],
+            "reason": "to route you",
+        }
+    )
+    assert req.kind == "choices"
+    assert req.reason == "to route you"
+    assert req.spec["questions"][0]["header"] == "Plan"
+
+    canonical, errors = kind.validate(req.spec, {"answers": [{"header": "Plan", "options": ["Pro"]}]})
+    assert errors == []
+    assert canonical["answers"][0]["options"][0] == "Pro"
+
+    directive = kind.fallback_directive(req.spec, "to route you")
+    assert "Basic, Pro" in directive
+    assert "submit_interaction" in directive
+
+
+def test_shared_fixtures_validate_to_the_canonical_payload(fixtures: dict) -> None:
+    """The shared ``choices`` fixtures: the golden spec + values must validate to exactly
+    the golden payload's values (the same golden shapes the Rust/Go/TS/C# engines check)."""
+    spec = fixtures["choices_spec"]["instance"]
+    values = fixtures["choices_values"]["instance"]
+    expected = fixtures["choices_payload"]["instance"]["values"]
+
+    canonical, errors = ChoicesKind().validate(spec, values)
+    assert errors == []
+    assert canonical == expected
diff --git a/python/server/tests/test_submit_interaction.py b/python/server/tests/test_submit_interaction.py
new file mode 100644
index 00000000..c9368f47
--- /dev/null
+++ b/python/server/tests/test_submit_interaction.py
@@ -0,0 +1,244 @@
+"""Rich Interactions (``choices`` kind) — the raise → park → ``submit_interaction`` →
+resume path, end-to-end over the real Python WS server.
+
+Boots the server with a scripted :class:`~smooth_operator_core.MockLlmProvider` (so the
+turn runs offline) and drives the full seam over a real ``websockets`` client:
+
+  - **Rich path** (session declared ``choice_chips``): ``request_choices`` parks the turn
+    and the server emits ``interaction_required``; a ``submit_interaction`` action resumes
+    it — the raise tool's canonical ``submitted`` payload reaches the model, the turn
+    streams the final reply and completes.
+  - **Invalid then resubmit**: a bad submit emits ``interaction_invalid`` and the turn
+    STAYS parked (retryable, never a terminal ``error``); a corrected submit resumes it.
+  - **Fallback path** (no capability): ``request_choices`` returns the conversational
+    directive with NO ``interaction_required``; the model submits through the generic
+    ``submit_interaction`` *tool* and the turn completes.
+
+The Python analog of the Rust ``submit_interaction`` tests. Like the write-confirmation
+test, the ``submit_interaction`` frame arrives on the same connection's reader while the
+turn is parked — proving the turn runs as a background task so the reader stays free.
+"""
+
+from __future__ import annotations
+
+import json
+
+import websockets
+from smooth_operator_core import MockLlmProvider
+
+from smooth_operator_server import ServerState, serve
+from smooth_operator_server.session_store import InMemorySessionStore
+
+_QUESTIONS = [
+    {"question": "Which plan interests you?", "header": "Plan", "options": [{"label": "Basic"}, {"label": "Pro"}]}
+]
+_RAISE_ARGS = json.dumps({"questions": _QUESTIONS, "reason": "to route you"})
+
+
+async def _start(mock: MockLlmProvider) -> tuple:
+    state = ServerState(store=InMemorySessionStore(), chat_client=mock)
+    server = await serve(state, "127.0.0.1", 0)
+    return server, state
+
+
+async def _create_session(ws, supports: list[str] | None = None) -> str:
+    frame = {
+        "action": "create_conversation_session",
+        "requestId": "r-create",
+        "agentId": "11111111-1111-1111-1111-111111111111",
+        "userName": "Alice",
+        "userEmail": "alice@example.com",
+    }
+    if supports is not None:
+        frame["supports"] = supports
+    await ws.send(json.dumps(frame))
+    while True:
+        event = json.loads(await ws.recv())
+        if event.get("type") == "immediate_response":
+            return event["data"]["sessionId"]
+
+
+async def _recv(ws):
+    while True:
+        event = json.loads(await ws.recv())
+        if event.get("type") not in ("keepalive", "pong"):
+            return event
+
+
+async def _send_message(ws) -> None:
+    await ws.send(
+        json.dumps({"action": "send_message", "requestId": "r-msg", "sessionId": _SID, "message": "help me pick"})
+    )
+
+
+# module-level session id shared across helpers, set per test
+_SID = ""
+
+
+async def test_rich_path_parks_emits_interaction_required_and_resumes() -> None:
+    mock = MockLlmProvider()
+    mock.push_tool_call("call-1", "request_choices", _RAISE_ARGS)
+    mock.push_text("Great — I've noted your pick.")
+    server, _ = await _start(mock)
+    global _SID
+    try:
+        async with websockets.connect(server.ws_url()) as ws:
+            _SID = await _create_session(ws, supports=["choice_chips"])
+            await _send_message(ws)
+
+            ack = await _recv(ws)
+            assert ack["type"] == "immediate_response" and ack["status"] == 202
+
+            # Park: an interaction_required arrives (a toolCall chunk may precede it).
+            event = await _recv(ws)
+            if event["type"] == "stream_chunk":
+                event = await _recv(ws)
+            assert event["type"] == "interaction_required"
+            assert event["requestId"] == "r-msg"
+            inner = event["data"]["data"]
+            assert inner["kind"] == "choices"
+            assert inner["spec"]["questions"][0]["header"] == "Plan"
+            assert inner["reason"] == "to route you"
+            interaction_id = inner["interactionId"]
+            assert interaction_id
+
+            # Resume: submit the pick. The reader was free to receive this while parked.
+            await ws.send(
+                json.dumps(
+                    {
+                        "action": "submit_interaction",
+                        "requestId": "r-msg",
+                        "sessionId": _SID,
+                        "interactionId": interaction_id,
+                        "kind": "choices",
+                        "values": {"answers": [{"header": "Plan", "options": ["Pro"]}]},
+                    }
+                )
+            )
+
+            tokens: list[str] = []
+            tool_results: list[dict] = []
+            saw_submit_ack = False
+            while True:
+                event = await _recv(ws)
+                etype = event["type"]
+                if etype == "immediate_response" and event["status"] == 200:
+                    saw_submit_ack = True
+                    assert event["data"]["values"]["answers"][0]["options"] == ["Pro"]
+                elif etype == "stream_chunk":
+                    tr = event["data"]["state"].get("rawResponse", {}).get("toolResult")
+                    if tr:
+                        tool_results.append(tr)
+                elif etype == "stream_token":
+                    tokens.append(event["token"])
+                elif etype == "eventual_response":
+                    break
+
+            assert saw_submit_ack, "the submit_interaction ack must arrive"
+            assert "".join(tokens) == "Great — I've noted your pick."
+            # The raise tool returned the canonical submitted payload to the model.
+            payload = json.loads(next(tr["result"] for tr in tool_results if tr["name"] == "request_choices"))
+            assert payload["status"] == "submitted"
+            assert payload["values"]["answers"][0]["options"] == ["Pro"]
+    finally:
+        await server.shutdown()
+
+
+async def test_invalid_submit_stays_parked_then_resubmit_resumes() -> None:
+    mock = MockLlmProvider()
+    mock.push_tool_call("call-1", "request_choices", _RAISE_ARGS)
+    mock.push_text("Locked in.")
+    server, _ = await _start(mock)
+    global _SID
+    try:
+        async with websockets.connect(server.ws_url()) as ws:
+            _SID = await _create_session(ws, supports=["choice_chips"])
+            await _send_message(ws)
+            assert (await _recv(ws))["status"] == 202
+            event = await _recv(ws)
+            if event["type"] == "stream_chunk":
+                event = await _recv(ws)
+            assert event["type"] == "interaction_required"
+            interaction_id = event["data"]["data"]["interactionId"]
+
+            # A bad pick (not an offered option) → interaction_invalid, turn stays parked.
+            await ws.send(
+                json.dumps(
+                    {
+                        "action": "submit_interaction",
+                        "requestId": "r-msg",
+                        "sessionId": _SID,
+                        "interactionId": interaction_id,
+                        "values": {"answers": [{"header": "Plan", "options": ["Platinum"]}]},
+                    }
+                )
+            )
+            invalid = await _recv(ws)
+            assert invalid["type"] == "interaction_invalid"
+            assert invalid["data"]["data"]["errors"][0]["field"] == "Plan"
+            assert invalid["data"]["data"]["message"] == "Some fields need attention."
+
+            # Corrected submit → resumes (proves the park survived the invalid attempt).
+            await ws.send(
+                json.dumps(
+                    {
+                        "action": "submit_interaction",
+                        "requestId": "r-msg",
+                        "sessionId": _SID,
+                        "interactionId": interaction_id,
+                        "values": {"answers": [{"header": "Plan", "options": ["Basic"]}]},
+                    }
+                )
+            )
+            while True:
+                event = await _recv(ws)
+                if event["type"] == "eventual_response":
+                    assert event["data"]["data"]["response"]["responseParts"] == ["Locked in."]
+                    break
+    finally:
+        await server.shutdown()
+
+
+async def test_fallback_path_without_capability_uses_the_submit_tool() -> None:
+    mock = MockLlmProvider()
+    mock.push_tool_call("call-1", "request_choices", _RAISE_ARGS)
+    mock.push_tool_call(
+        "call-2",
+        "submit_interaction",
+        json.dumps({"kind": "choices", "values": {"answers": [{"header": "Plan", "options": ["Pro"]}]}}),
+    )
+    mock.push_text("Thanks for letting me know.")
+    server, _ = await _start(mock)
+    global _SID
+    try:
+        async with websockets.connect(server.ws_url()) as ws:
+            # No `supports` → text-only channel → choices degrades to the conversational
+            # fallback (no card, no park).
+            _SID = await _create_session(ws)
+            await _send_message(ws)
+            assert (await _recv(ws))["status"] == 202
+
+            events: list[dict] = []
+            tool_results: list[dict] = []
+            while True:
+                event = await _recv(ws)
+                events.append(event)
+                if event["type"] == "stream_chunk":
+                    tr = event["data"]["state"].get("rawResponse", {}).get("toolResult")
+                    if tr:
+                        tool_results.append(tr)
+                if event["type"] == "eventual_response":
+                    break
+
+            # The fallback path NEVER emits interaction_required.
+            assert all(e["type"] != "interaction_required" for e in events)
+            # request_choices returned the conversational directive…
+            raise_result = json.loads(next(tr["result"] for tr in tool_results if tr["name"] == "request_choices"))
+            assert raise_result["mode"] == "conversational"
+            assert "submit_interaction" in raise_result["instructions"]
+            # …and the generic submit_interaction TOOL returned the submitted payload.
+            submit_result = json.loads(next(tr["result"] for tr in tool_results if tr["name"] == "submit_interaction"))
+            assert submit_result["status"] == "submitted"
+            assert submit_result["values"]["answers"][0]["options"] == ["Pro"]
+    finally:
+        await server.shutdown()
diff --git a/python/server/tests/test_turn_runner_agent_config.py b/python/server/tests/test_turn_runner_agent_config.py
index 50478a0f..c05fcd77 100644
--- a/python/server/tests/test_turn_runner_agent_config.py
+++ b/python/server/tests/test_turn_runner_agent_config.py
@@ -254,6 +254,16 @@ async def test_tool_config_filters_tools_per_agent() -> None:
         )
         await dispatcher.wait_for_turns()
 
-    # agent-a restricted to its allow-list; agent-b (no config) sees the full set.
-    assert _sent_tool_names(mock.calls[0]) == ["crm"]
-    assert _sent_tool_names(mock.calls[1]) == ["crm", "knowledge_search", "notify_humans"]
+    # agent-a restricted to its allow-list; agent-b (no config) sees the full set. The
+    # Rich Interaction framework tools (request_<kind> + submit_interaction) are always
+    # registered — NOT subject to the agent allow-list, mirroring the Rust runner — so
+    # assert the AGENT tool subset with those filtered out.
+    _INTERACTION = {"request_choices", "submit_interaction"}
+    assert [n for n in _sent_tool_names(mock.calls[0]) if n not in _INTERACTION] == ["crm"]
+    assert [n for n in _sent_tool_names(mock.calls[1]) if n not in _INTERACTION] == [
+        "crm",
+        "knowledge_search",
+        "notify_humans",
+    ]
+    # …and the interaction tools ARE present regardless of the agent's allow-list.
+    assert _INTERACTION.issubset(set(_sent_tool_names(mock.calls[0])))
diff --git a/python/server/uv.lock b/python/server/uv.lock
index 56979f21..6ee689b6 100644
--- a/python/server/uv.lock
+++ b/python/server/uv.lock
@@ -846,7 +846,7 @@ wheels = [
 
 [[package]]
 name = "smooai-smooth-operator-server"
-version = "1.49.1"
+version = "1.52.3"
 source = { editable = "." }
 dependencies = [
     { name = "opentelemetry-api" },