diff --git a/adapters/README.md b/adapters/README.md
index 3ea402ec..af4c7e7a 100644
--- a/adapters/README.md
+++ b/adapters/README.md
@@ -227,7 +227,7 @@ adapters/
 ├── uv.lock                         # exact dependency versions, with hashes
 ├── src/acs/
 │   ├── core/                       # no I/O: envelopes, signing, decisions, error codes, bundled schemas
-│   ├── client/                     # I/O: settings and fail posture, the secret, the handshake, the round trip, session state, audit
+│   ├── client/                     # I/O: settings and fail posture, the secret, the handshake, the round trip, session state and turns, audit
 │   ├── integrations/<framework>/
 │   │   ├── hook.py                 # the installed hook command (Claude Code, Cursor)
 │   │   ├── adapter.py              # the hook, or the NAT middleware
@@ -252,6 +252,6 @@ Installing the package (`uv tool install ./adapters`, or `pip install ./adapters
 5. Add a row to the status table above.
 6. Open a PR against `GenAI-Security-Project/agent-control-standard`.
 
-Keep module-level state constant or set once at startup. Read `ACS_GUARDIAN_URL`, `ACS_DEFAULT_DENY`, `ACS_STARTUP_POSTURE`, `ACS_HANDSHAKE` and `ACS_DISABLED` once with `acs.client.Config.from_env` at the entry point, pass the `Config` and `FailPosture` down, gate each step on `Config.gate(open_session(...))`, take each round trip's outcome from `acs.client.exchange`, and settle every step without a usable decision through `FailPosture.resolve`.
+Keep module-level state constant or set once at startup. Read `ACS_GUARDIAN_URL`, `ACS_DEFAULT_DENY`, `ACS_STARTUP_POSTURE`, `ACS_HANDSHAKE` and `ACS_DISABLED` once with `acs.client.Config.from_env` at the entry point, pass the `Config` and `FailPosture` down, gate each step on `Config.gate(open_session(...))`, build envelopes with `acs.core.request_envelope` and send them unsigned to `acs.client.round_trip`, which signs as the last step and returns either the decision or the `FailPosture.resolve` resolution to write in the host's shape, and keep the open turn with `acs.client.open_turn` / `current_turn` / `close_turn`.
 
 The bar for "reference implementation" status is: round-trip tests pass against the example Guardian, documented configuration for users, and an explicit conformance posture statement matching the format in the Claude Code adapter's README.
diff --git a/adapters/src/acs/client/__init__.py b/adapters/src/acs/client/__init__.py
index 3b3ac88e..4ff2d676 100644
--- a/adapters/src/acs/client/__init__.py
+++ b/adapters/src/acs/client/__init__.py
@@ -20,13 +20,17 @@
     "Undecided": "_outcome",
     "audit_event": "_audit",
     "classify_response": "_exchange",
+    "close_turn": "_store",
+    "current_turn": "_store",
     "exchange": "_exchange",
     "load_hmac_secret": "_secrets",
     "load_session_state": "_store",
     "open_session": "_handshake",
+    "open_turn": "_store",
     "ping": "_transport",
     "post": "_exchange",
     "record_step": "_store",
+    "round_trip": "_exchange",
     "save_session_state": "_store",
     "sign_envelope": "_signing",
     "validate_guardian_url": "_urls",
diff --git a/adapters/src/acs/client/_config.py b/adapters/src/acs/client/_config.py
index d9241da3..cba545f5 100644
--- a/adapters/src/acs/client/_config.py
+++ b/adapters/src/acs/client/_config.py
@@ -25,6 +25,7 @@ class Resolution:
     audit_type: str
     cause: str
     fields: Mapping[str, Any]
+    message: str = ""
 
     @property
     def label(self) -> str:
@@ -65,10 +66,10 @@ def negotiated(self, server_hello: Mapping[str, Any] | None) -> FailPosture:
     def resolve(self, outcome: Undecided | Refused) -> Resolution:
         """The one place a step without a usable decision is settled: a refusal blocks, anything else follows the posture."""
         if isinstance(outcome, Refused):
-            return Resolution(False, "guardian_refusal_fail_closed", outcome.cause, outcome.fields)
+            return Resolution(False, "guardian_refusal_fail_closed", outcome.cause, outcome.fields, outcome.message)
         if self.deny_on_failure:
-            return Resolution(False, "decision_failure_fail_closed", outcome.cause, outcome.fields)
-        return Resolution(True, "fail_open_bypass", outcome.cause, outcome.fields)
+            return Resolution(False, "decision_failure_fail_closed", outcome.cause, outcome.fields, outcome.message)
+        return Resolution(True, "fail_open_bypass", outcome.cause, outcome.fields, outcome.message)
 
 
 @dataclass(frozen=True)
diff --git a/adapters/src/acs/client/_exchange.py b/adapters/src/acs/client/_exchange.py
index 8487f0ff..2e0da788 100644
--- a/adapters/src/acs/client/_exchange.py
+++ b/adapters/src/acs/client/_exchange.py
@@ -10,8 +10,9 @@
 from dataclasses import replace
 from typing import Any
 
+from acs.client._config import FailPosture, Resolution
 from acs.client._outcome import Decided, Outcome, Refused, Undecided
-from acs.client._signing import verify_signature
+from acs.client._signing import sign_envelope, verify_signature
 from acs.client._urls import validate_guardian_url
 from acs.core import MAX_REQUEST_BODY_BYTES, guardian_error_cause, is_guardian_refusal, response_matches_request
 
@@ -180,6 +181,18 @@ def _read(resp: Any, cancelled: threading.Event) -> bytes:
     return b"".join(chunks)
 
 
+def round_trip(request: dict[str, Any], *, guardian_url: str, posture: FailPosture,
+               session_id: str | None) -> Decided | Resolution:
+    """Sign the request as the last step before sending, exchange it within its method's timeout, and resolve anything but a usable decision with the posture."""
+    try:
+        sign_envelope(request, session_id=session_id)
+    except Exception as e:  # noqa: BLE001
+        return posture.resolve(Undecided("adapter_exception", {"error": str(e)}, f"adapter error: {e}"))
+    outcome = exchange(request, guardian_url=guardian_url,
+                       timeout_s=posture.timeout_for(request.get("method", "")), session_id=session_id)
+    return outcome if isinstance(outcome, Decided) else posture.resolve(outcome)
+
+
 def exchange(request: dict[str, Any], *, guardian_url: str, timeout_s: float,
              session_id: str | None) -> Outcome:
     """POST one request and classify the answer; never raises."""
diff --git a/adapters/src/acs/client/_store.py b/adapters/src/acs/client/_store.py
index b8540d04..ed2aeba7 100644
--- a/adapters/src/acs/client/_store.py
+++ b/adapters/src/acs/client/_store.py
@@ -107,3 +107,26 @@ def record_step(session_id: str, step_id: str, *, workspace: str | None = None)
             del seen[: len(seen) - 1000]
     st["last_step_id"] = step_id
     save_session_state(session_id, st, workspace=workspace)
+
+
+# ----- The open turn -----
+
+def current_turn(session_id: str | None, *, workspace: str | None = None) -> str | None:
+    if not session_id:
+        return None
+    turn_id = load_session_state(session_id, workspace=workspace).get("turn_id")
+    return turn_id if isinstance(turn_id, str) else None
+
+
+def open_turn(session_id: str, turn_id: str, *, workspace: str | None = None) -> None:
+    """Record turn_id as the session's open turn, for later steps to stamp and the turn end to close."""
+    state = load_session_state(session_id, workspace=workspace)
+    state["turn_id"] = turn_id
+    save_session_state(session_id, state, workspace=workspace)
+
+
+def close_turn(session_id: str, *, workspace: str | None = None) -> None:
+    """Forget the session's open turn once its turnEnd has been sent or recorded."""
+    state = load_session_state(session_id, workspace=workspace)
+    state.pop("turn_id", None)
+    save_session_state(session_id, state, workspace=workspace)
diff --git a/adapters/src/acs/core/__init__.py b/adapters/src/acs/core/__init__.py
index 7598f8ab..e4f1a6c5 100644
--- a/adapters/src/acs/core/__init__.py
+++ b/adapters/src/acs/core/__init__.py
@@ -6,6 +6,7 @@
 _EXPORTS = {
     "ACS_VERSION": "_envelope",
     "DEFAULT_SKEW_WINDOW_MS": "_envelope",
+    "Decision": "_decision",
     "GUARDIAN_ERROR_CAUSE": "_errors",
     "GUARDIAN_REFUSAL_CODES": "_errors",
     "KNOWN_DECISIONS": "_decision",
@@ -23,10 +24,12 @@
     "modify_composition_violation": "_decision",
     "normalize_decision": "_decision",
     "parse_iso8601": "_envelope",
+    "request_envelope": "_envelope",
     "response_matches_request": "_envelope",
     "schemas_dir": "_schemas",
     "server_hello_problem": "_handshake",
     "sign_with_key": "_signing",
+    "turn_start_envelope": "_envelope",
     "verify_with_key": "_signing",
 }
 
diff --git a/adapters/src/acs/core/_decision.py b/adapters/src/acs/core/_decision.py
index 519ea35e..0ad04331 100644
--- a/adapters/src/acs/core/_decision.py
+++ b/adapters/src/acs/core/_decision.py
@@ -3,6 +3,7 @@
 
 import json
 import re
+from dataclasses import dataclass, field
 from typing import Any
 
 
@@ -13,7 +14,30 @@
 KNOWN_DECISIONS = frozenset({"allow", "deny", "modify", "ask", "defer"})
 
 
-def normalize_decision(result: Any) -> tuple[str, str, dict]:
+@dataclass(frozen=True)
+class Decision:
+    """A Guardian result's decision fields, each coerced to a safe type.
+
+    - disposition: stripped, lower-cased str; "" when the field is not a
+      usable string, which routes the caller to its unusable→DENY branch.
+    - reasoning: always a str (JSON repr if the Guardian sent a
+      non-string, so it can go into a `reason` field unchanged).
+    - modifications: always a dict ({} when the Guardian sent a
+      non-object, which makes a MODIFY with no usable mutation
+      substitute to DENY per §6.3:146).
+    """
+
+    disposition: str = ""
+    reasoning: str = ""
+    modifications: dict = field(default_factory=dict)
+
+    @property
+    def known(self) -> bool:
+        """Whether the disposition is one of the five ACS dispositions (§6.3)."""
+        return self.disposition in KNOWN_DECISIONS
+
+
+def normalize_decision(result: Any) -> Decision:
     """Coerce a Guardian result's decision fields to safe types, never
     raising.
 
@@ -22,17 +46,10 @@ def normalize_decision(result: Any) -> tuple[str, str, dict]:
     intent, and an uncaught AttributeError exits 1, which the shell
     hosts treat as proceed (fail-open).
 
-    Returns (decision, reasoning, modifications):
-      - decision: stripped, lower-cased str; "" when the field is not a
-        usable string, which routes the caller to its unusable→DENY branch.
-      - reasoning: always a str (JSON repr if the Guardian sent a
-        non-string, so it can go into a `reason` field unchanged).
-      - modifications: always a dict ({} when the Guardian sent a
-        non-object, which makes a MODIFY with no usable mutation
-        substitute to DENY per §6.3:146).
+    Returns a Decision with every field set.
     """
     if not isinstance(result, dict):
-        return "", "", {}
+        return Decision()
     raw_decision = result.get("decision")
     decision = raw_decision.strip().lower() if isinstance(raw_decision, str) else ""
     raw_reason = result.get("reasoning", "")
@@ -40,7 +57,7 @@ def normalize_decision(result: Any) -> tuple[str, str, dict]:
                  else json.dumps(raw_reason, default=str))
     raw_mods = result.get("modifications", {})
     modifications = raw_mods if isinstance(raw_mods, dict) else {}
-    return decision, reasoning, modifications
+    return Decision(decision, reasoning, modifications)
 
 
 def modify_composition_violation(modifications: dict) -> str | None:
diff --git a/adapters/src/acs/core/_envelope.py b/adapters/src/acs/core/_envelope.py
index 36c200e8..1cac29a2 100644
--- a/adapters/src/acs/core/_envelope.py
+++ b/adapters/src/acs/core/_envelope.py
@@ -1,8 +1,9 @@
-"""Envelope constants, response binding, timestamps and IDs."""
 from __future__ import annotations
 
 import datetime
 import uuid
+from collections.abc import Mapping
+from typing import Any
 
 
 ACS_VERSION = "0.1.0"
@@ -14,7 +15,28 @@
 MAX_REQUEST_BODY_BYTES = 1_048_576  # 1 MiB
 
 
-# ----- Response↔request binding -----
+def request_envelope(method: str, *, request_id: str, metadata: Mapping[str, Any],
+                     payload: Mapping[str, Any]) -> dict[str, Any]:
+    """An unsigned step request; round_trip signs it just before sending."""
+    return {
+        "jsonrpc": "2.0",
+        "id": str(uuid.uuid4()),
+        "method": method,
+        "params": {
+            "acs_version": ACS_VERSION,
+            "request_id": request_id,
+            "timestamp": iso8601_now(),
+            "metadata": dict(metadata),
+            "payload": dict(payload),
+        },
+    }
+
+
+def turn_start_envelope(metadata: Mapping[str, Any], turn_id: str) -> dict[str, Any]:
+    """turn_id doubles as request_id, as turn-start.json recommends."""
+    return request_envelope("steps/turnStart", request_id=turn_id, metadata={**metadata, "turn_id": turn_id},
+                            payload={"turn_id": turn_id, "triggered_by": "user_message"})
+
 
 def response_matches_request(request: dict, response: dict) -> bool:
     """Bind a Guardian response to the request that elicited it.
diff --git a/adapters/src/acs/integrations/claude_code/adapter.py b/adapters/src/acs/integrations/claude_code/adapter.py
index 4556653c..c7776199 100755
--- a/adapters/src/acs/integrations/claude_code/adapter.py
+++ b/adapters/src/acs/integrations/claude_code/adapter.py
@@ -67,26 +67,25 @@
 
 from acs.client import (
     Config,
-    Decided,
     FailPosture,
     Handshake,
     Refused,
     Resolution,
     Undecided,
     audit_event,
-    exchange,
-    load_session_state,
+    close_turn,
+    current_turn,
     open_session,
-    save_session_state,
-    sign_envelope,
+    open_turn,
+    round_trip,
 )
 from acs.core import (
-    ACS_VERSION,
-    KNOWN_DECISIONS,
-    iso8601_now,
+    Decision,
     method_evaluated,
     modify_composition_violation,
     normalize_decision,
+    request_envelope,
+    turn_start_envelope,
 )
 from acs.integrations.claude_code.hook import _event_name
 
@@ -283,31 +282,13 @@ def _turn_ws(event: dict[str, Any]) -> str | None:
     return event.get("cwd") or None
 
 
-def _current_turn_id(session_id: str | None, event: dict[str, Any]) -> str | None:
-    if not session_id:
-        return None
-    return load_session_state(session_id, workspace=_turn_ws(event)).get("turn_id")
-
-
-def _open_turn(session_id: str, turn_id: str, event: dict[str, Any]) -> None:
-    st = load_session_state(session_id, workspace=_turn_ws(event))
-    st["turn_id"] = turn_id
-    save_session_state(session_id, st, workspace=_turn_ws(event))
-
-
-def _close_turn(session_id: str, event: dict[str, Any]) -> None:
-    st = load_session_state(session_id, workspace=_turn_ws(event))
-    st.pop("turn_id", None)
-    save_session_state(session_id, st, workspace=_turn_ws(event))
-
-
 def _payload_turn_end(event: dict[str, Any]) -> dict[str, Any]:
     """steps/turnEnd — closes the turn opened at the last UserPromptSubmit.
     turn_id comes from session state (empty if no turn is open, in which
     case build_request skips the emission). Claude's Stop fires on normal
     turn completion and carries no error discriminator, so outcome is
     'completed'."""
-    turn_id = _current_turn_id(event.get("session_id"), event)
+    turn_id = current_turn(event.get("session_id"), workspace=_turn_ws(event))
     return {"turn_id": turn_id or "", "outcome": "completed"}
 
 
@@ -401,25 +382,17 @@ def build_request(event: dict[str, Any]) -> dict[str, Any]:
     if subagent_spawn:
         method = "steps/subagentStart"
 
-    metadata: dict[str, Any] = {
-        "agent_id": _agent_id(event),
-        "session_id": session_id,
-        "platform": "claude-code",
-        "adapter_version": ADAPTER_VERSION,
-    }
-    for k in ("cwd", "transcript_path", "permission_mode"):
-        if event.get(k):
-            metadata[k] = event[k]
+    metadata = _metadata(event)
 
     # Stamp the open turn's id on in-turn steps (request-envelope.json:69).
     if method in _IN_TURN_METHODS:
-        tid = _current_turn_id(session_id, event)
+        tid = current_turn(session_id, workspace=_turn_ws(event))
         if tid:
             metadata["turn_id"] = tid
 
     # steps/turnEnd needs a turn_id (>=1); if no turn is open there is
     # nothing to close honestly, so skip the emission (main audits it).
-    if method == "steps/turnEnd" and not _current_turn_id(session_id, event):
+    if method == "steps/turnEnd" and not current_turn(session_id, workspace=_turn_ws(event)):
         return {}
 
     # For PreToolUse (including the Task→subagentStart remap), pin
@@ -442,52 +415,24 @@ def build_request(event: dict[str, Any]) -> dict[str, Any]:
             payload["subagent_session_id"] = str(uuid.uuid5(
                 uuid.NAMESPACE_URL, f"claude-code:subagent:{request_id}"))
 
-    envelope = {
-        "jsonrpc": "2.0",
-        "id": str(uuid.uuid4()),
-        "method": method,
-        "params": {
-            "acs_version": ACS_VERSION,
-            "request_id": request_id,
-            "timestamp": iso8601_now(),
-            "metadata": metadata,
-            "payload": payload,
-        },
-    }
-    sign_envelope(envelope, session_id=session_id)
-    return envelope
+    return request_envelope(method, request_id=request_id, metadata=metadata, payload=payload)
 
 
-def build_turn_start_request(event: dict[str, Any], turn_id: str) -> dict[str, Any]:
-    """steps/turnStart opening the turn for a UserPromptSubmit. turn_id
-    SHOULD equal the envelope's request_id (turn-start.json), so we use it
-    for both. triggered_by is user_message — a user prompt opened this
-    turn. Decision-eligible: a Guardian MAY deny to block the whole turn."""
-    session_id = event.get("session_id") or ""
+def _metadata(event: dict[str, Any]) -> dict[str, Any]:
     metadata: dict[str, Any] = {
         "agent_id": _agent_id(event),
-        "session_id": session_id,
+        "session_id": event.get("session_id") or "",
         "platform": "claude-code",
         "adapter_version": ADAPTER_VERSION,
-        "turn_id": turn_id,
     }
     for k in ("cwd", "transcript_path", "permission_mode"):
         if event.get(k):
             metadata[k] = event[k]
-    envelope = {
-        "jsonrpc": "2.0",
-        "id": str(uuid.uuid4()),
-        "method": "steps/turnStart",
-        "params": {
-            "acs_version": ACS_VERSION,
-            "request_id": turn_id,
-            "timestamp": iso8601_now(),
-            "metadata": metadata,
-            "payload": {"turn_id": turn_id, "triggered_by": "user_message"},
-        },
-    }
-    sign_envelope(envelope, session_id=session_id)
-    return envelope
+    return metadata
+
+
+def build_turn_start_request(event: dict[str, Any], turn_id: str) -> dict[str, Any]:
+    return turn_start_envelope(_metadata(event), turn_id)
 
 
 def _open_session(event: dict[str, Any], config: Config) -> Handshake:
@@ -507,29 +452,27 @@ def _open_session(event: dict[str, Any], config: Config) -> Handshake:
 
 # ─── Response translation — dispatch table, one function per hook ─────────
 
-def _translate_pretool(decision: str, reasoning: str,
-                        modifications: dict,
-                        original_input: dict) -> dict[str, Any]:
-    if decision in PRETOOL_PERMISSION_MAP:
-        return _pretool_response(PRETOOL_PERMISSION_MAP[decision], reasoning)
-    if decision == "defer":
+def _translate_pretool(decision: Decision, original_input: dict) -> dict[str, Any]:
+    if decision.disposition in PRETOOL_PERMISSION_MAP:
+        return _pretool_response(PRETOOL_PERMISSION_MAP[decision.disposition], decision.reasoning)
+    if decision.disposition == "defer":
         # No native defer in permissionDecision; spec default
         # timeout_decision is deny (§6). Audited so substitution rate
         # is machine-visible.
         audit_event("defer_substituted_deny", hook="PreToolUse")
         return _pretool_response(
-            "deny", f"DEFER substituted to DENY (no native defer): {reasoning}")
-    if decision == "modify":
+            "deny", f"DEFER substituted to DENY (no native defer): {decision.reasoning}")
+    if decision.disposition == "modify":
         # §6.3:146 — a modifications object that breaks the composition
         # rules (both shapes, or overlapping structured targets) is
         # un-interpretable; fail closed to DENY rather than half-apply it.
-        violation = modify_composition_violation(modifications)
+        violation = modify_composition_violation(decision.modifications)
         if violation:
             audit_event("modify_composition_invalid", hook="PreToolUse",
                         detail=violation)
             return _pretool_response(
                 "deny", f"MODIFY refused ({violation}); failing closed per §6.3:146")
-        if modifications.get("redactions"):
+        if decision.modifications.get("redactions"):
             # Claude's updatedInput can replace arguments, but it exposes no
             # native JSON-Pointer redaction operation.  Applying sibling
             # parameter_overrides while dropping redactions would change the
@@ -539,81 +482,73 @@ def _translate_pretool(decision: str, reasoning: str,
             return _pretool_response(
                 "deny", "MODIFY refused: Claude Code cannot apply the "
                 "requested redactions without dropping part of the decision")
-        overrides = modifications.get("parameter_overrides")
+        overrides = decision.modifications.get("parameter_overrides")
         # A non-dict override cannot be applied as updatedInput; §6.3:146
         # fails an unrealizable modification closed to DENY.
         if isinstance(overrides, dict):
             # updatedInput REPLACES the whole input, so merge the per-
             # argument overrides (modifications.json) onto the original.
             merged = {**original_input, **overrides}
-            return _pretool_response("allow", reasoning, updated_input=merged)
+            return _pretool_response("allow", decision.reasoning, updated_input=merged)
         return _pretool_response(
             "deny",
-            f"MODIFY substituted to DENY (no usable parameter_overrides): {reasoning}",
+            f"MODIFY substituted to DENY (no usable parameter_overrides): {decision.reasoning}",
         )
     return {}
 
 
-def _translate_posttool(decision: str, reasoning: str,
-                         modifications: dict,
-                         original_input: dict) -> dict[str, Any]:
-    if decision == "deny":
-        return _block_response(reasoning or "blocked by Guardian", "PostToolUse")
-    if decision == "modify":
-        violation = modify_composition_violation(modifications)
+def _translate_posttool(decision: Decision, original_input: dict) -> dict[str, Any]:
+    if decision.disposition == "deny":
+        return _block_response(decision.reasoning or "blocked by Guardian", "PostToolUse")
+    if decision.disposition == "modify":
+        violation = modify_composition_violation(decision.modifications)
         if violation:
             audit_event("modify_composition_invalid", hook="PostToolUse",
                         detail=violation)
             return _block_response(
                 f"MODIFY refused ({violation}); failing closed per §6.3:146")
-        updated = modifications.get("modified_content")
+        updated = decision.modifications.get("modified_content")
         if updated is not None:
             hso = {
                 "hookEventName": "PostToolUse",
                 "updatedToolOutput": str(updated),
             }
-            if reasoning:
-                hso["additionalContext"] = reasoning
+            if decision.reasoning:
+                hso["additionalContext"] = decision.reasoning
             return {"hookSpecificOutput": hso}
         return _block_response(
-            f"MODIFY substituted to DENY (no modified_content): {reasoning}")
-    if decision in ("ask", "defer"):
+            f"MODIFY substituted to DENY (no modified_content): {decision.reasoning}")
+    if decision.disposition in ("ask", "defer"):
         return _block_response(
-            f"{decision} on post-tool not supported by Claude Code: {reasoning}")
+            f"{decision.disposition} on post-tool not supported by Claude Code: {decision.reasoning}")
     return {}
 
 
-def _translate_user_prompt(decision: str, reasoning: str,
-                            modifications: dict,
-                            original_input: dict) -> dict[str, Any]:
-    if decision == "allow":
+def _translate_user_prompt(decision: Decision, original_input: dict) -> dict[str, Any]:
+    if decision.disposition == "allow":
         return {}
-    if decision == "deny":
-        return _block_response(reasoning or "blocked by Guardian")
-    # The prompt hook can only allow or block, so ask/defer/modify fail
-    # closed to a block, audited — an arrived restrictive decision the
-    # host cannot apply must not execute as allow (§6.4).
+    if decision.disposition == "deny":
+        return _block_response(decision.reasoning or "blocked by Guardian")
+    # The prompt hook can only allow or block, so ask, defer and modify block, audited (§6.4).
     audit_event("prompt_decision_substituted_block",
-                disposition=decision, hook="UserPromptSubmit")
+                disposition=decision.disposition, hook="UserPromptSubmit")
     return _block_response(
-        f"{decision} on user prompt not applicable at this hook — blocking: {reasoning}")
+        f"{decision.disposition} on user prompt not applicable at this hook — blocking: {decision.reasoning}")
 
 
-def _translate_session_stop(decision: str, reasoning: str,
-                             modifications: dict,
-                             original_input: dict) -> dict[str, Any]:
+def _translate_session_stop(decision: Decision, original_input: dict) -> dict[str, Any]:
     """Stop (steps/turnEnd) is not decision-eligible — the turn has
     already happened (hooks.md:290). A DENY cannot be enforced here, and
     Claude Code reads {"decision":"block"} on Stop as "do not stop,
     continue the conversation", so emitting one would turn a denial into
     an instruction to keep working. Record it and emit nothing."""
-    if decision in ("deny", "ask", "defer"):
-        audit_event("unenforceable_decision", decision=decision,
-                    reason=reasoning or "(none)")
+    if decision.disposition in ("deny", "ask", "defer"):
+        audit_event("unenforceable_decision", decision=decision.disposition,
+                    reason=decision.reasoning or "(none)")
     return {}
 
 
-_TRANSLATORS: dict[str, Callable[[str, str, dict, dict], dict[str, Any]]] = {
+_TRANSLATORS: dict[str, Callable[[Decision, dict], dict[str, Any]]] = {
     "PreToolUse":       _translate_pretool,
     "PostToolUse":      _translate_posttool,
     "UserPromptSubmit": _translate_user_prompt,
@@ -626,16 +561,15 @@ def translate_response(acs_response: dict[str, Any], hook_event: str,
     # normalize_decision never raises: this runs outside main()'s
     # try/except and an uncaught error is exit 1, which Claude Code
     # proceeds on.
-    decision, reasoning, modifications = normalize_decision(
-        acs_response.get("result"))
+    decision = normalize_decision(acs_response.get("result"))
 
     # An arrived decision outside the five dispositions fails closed;
     # whether §6.4:156 (posture) or §6.3:146 (fail closed) governs is
     # unsettled and tracked in issue #32 — deny is the safe reading.
-    if decision not in KNOWN_DECISIONS:
+    if not decision.known:
         audit_event("unusable_disposition",
-                    disposition=decision or "(missing)", hook=hook_event)
-        reason = (f"unusable Guardian disposition '{decision}' — cannot be "
+                    disposition=decision.disposition or "(missing)", hook=hook_event)
+        reason = (f"unusable Guardian disposition '{decision.disposition}' — cannot be "
                   f"interpreted; failing closed (see issue #32)")
         if hook_event == "PreToolUse":
             return _pretool_response("deny", reason)
@@ -648,33 +582,20 @@ def translate_response(acs_response: dict[str, Any], hook_event: str,
 
     translator = _TRANSLATORS.get(hook_event)
     if translator:
-        return translator(decision, reasoning, modifications, original_input or {})
+        return translator(decision, original_input or {})
 
     # Informational hooks (SessionStart, SessionEnd, Notification):
     # surface the Guardian's `reasoning` as additionalContext when present.
-    if reasoning:
+    if decision.reasoning:
         return {"hookSpecificOutput": {
             "hookEventName": hook_event,
-            "additionalContext": reasoning,
+            "additionalContext": decision.reasoning,
         }}
     return {}
 
 
 # ─── Main flow ──────────────────────────────────────────────────────────────
 
-def _guardian_roundtrip(request: dict[str, Any], hook_name: str, session_id: str | None,
-                        guardian_url: str, posture: FailPosture) -> tuple[dict | None, int | None]:
-    """POST one request. Returns (response, None) for a usable decision, or
-    (None, exit_code) once a failure has been settled in the hook's shape;
-    shared by the turnStart pre-step and the main step."""
-    outcome = exchange(request, guardian_url=guardian_url, timeout_s=posture.timeout_for(request.get("method", "")),
-                       session_id=session_id)
-    if isinstance(outcome, Decided):
-        return outcome.response, None
-    sys.stderr.write(f"acs-adapter: {outcome.message or outcome.cause}\n")
-    return None, _settle(posture.resolve(outcome), hook_name, session_id)
-
-
 def main() -> int:
     config = Config.from_env(os.environ)
     if len(sys.argv) > 1 and sys.argv[1] == "--version":
@@ -739,25 +660,22 @@ def _govern(event: dict[str, Any], hook_name: str, session_id: str | None,
     if hook_name == "UserPromptSubmit":
         turn_id = str(uuid.uuid4())
         if method_evaluated(server_hello, "steps/turnStart"):
-            ts_response, terminal = _guardian_roundtrip(
-                build_turn_start_request(event, turn_id), "UserPromptSubmit", session_id,
-                config.guardian_url, posture)
-            if terminal is not None:
-                return terminal
-            ts_decision, ts_reason, _ = normalize_decision(ts_response.get("result"))
-            if ts_decision != "allow":
-                # turnStart is a gate boundary — the host can't apply
-                # modify/ask/defer to "a turn is starting", so any non-allow
-                # blocks the prompt (fail closed). Audited.
+            result = round_trip(build_turn_start_request(event, turn_id), guardian_url=config.guardian_url,
+                                posture=posture, session_id=session_id)
+            if isinstance(result, Resolution):
+                return _settle(result, "UserPromptSubmit", session_id)
+            turn_start = normalize_decision(result.response.get("result"))
+            if turn_start.disposition != "allow":
+                # Only allow applies to a turn start, so any other disposition blocks the prompt.
                 audit_event("turn_start_blocked", turn_id=turn_id,
-                            disposition=ts_decision or "(unusable)")
+                            disposition=turn_start.disposition or "(unusable)")
                 _emit(_block_response(
-                    f"turn blocked at start ({ts_decision or 'unusable'}): {ts_reason}"))
+                    f"turn blocked at start ({turn_start.disposition or 'unusable'}): {turn_start.reasoning}"))
                 return 0
         else:
             audit_event("method_not_evaluated", method="steps/turnStart",
                         hook=hook_name, session_id=session_id)
-        _open_turn(session_id, turn_id, event)
+        open_turn(session_id, turn_id, workspace=_turn_ws(event))
 
     try:
         request = build_request(event)
@@ -781,23 +699,21 @@ def _govern(event: dict[str, Any], hook_name: str, session_id: str | None,
         audit_event("method_not_evaluated", method=method, hook=hook_name,
                     session_id=session_id)
         if hook_name == "Stop":
-            _close_turn(session_id, event)
+            close_turn(session_id, workspace=_turn_ws(event))
         return 0
     if method == "steps/toolCallResult":
         origin = ("steps/subagentStart" if _is_subagent_tool(event.get("tool_name"))
                   else "steps/toolCallRequest")
-        if (not method_evaluated(server_hello, origin)
-                and request["params"]["payload"].pop("request_id_ref", None) is not None):
+        if not method_evaluated(server_hello, origin):
             # The originating request was never sent, so citing it would dangle.
-            sign_envelope(request, session_id=request["params"]["metadata"]["session_id"])
+            request["params"]["payload"].pop("request_id_ref", None)
 
-    response, terminal = _guardian_roundtrip(request, hook_name, session_id,
-                                             config.guardian_url, posture)
-    if terminal is not None:
-        return terminal
+    result = round_trip(request, guardian_url=config.guardian_url, posture=posture, session_id=session_id)
+    if isinstance(result, Resolution):
+        return _settle(result, hook_name, session_id)
+    response = result.response
 
-    # translate_response runs outside _guardian_roundtrip's try/except and
-    # exit 1 means the host proceeds, so a translation bug fails closed here.
+    # A translation bug would exit 1, which the host reads as proceed, so it fails closed here.
     try:
         _emit(translate_response(response, hook_name,
                                  original_input=event.get("tool_input") or {}))
@@ -808,7 +724,7 @@ def _govern(event: dict[str, Any], hook_name: str, session_id: str | None,
 
     # Close the turn after the turnEnd has been emitted/recorded.
     if hook_name == "Stop":
-        _close_turn(session_id, event)
+        close_turn(session_id, workspace=_turn_ws(event))
     return 0
 
 
@@ -822,6 +738,8 @@ def _settle(resolution: Resolution, hook_name: str, session_id: str | None) -> i
     what went wrong, so an unreachable Guardian (an ops issue) reads apart from
     a malformed envelope (a client bug).
     """
+    if resolution.message:
+        sys.stderr.write(f"acs-adapter: {resolution.message}\n")
     if not resolution.proceed:
         _emit_deny(hook_name, f"ACS adapter: {resolution.label} ({resolution.cause})")
     audit_event(resolution.audit_type, cause=resolution.cause, hook=hook_name,
diff --git a/adapters/src/acs/integrations/cursor/adapter.py b/adapters/src/acs/integrations/cursor/adapter.py
index e63d19c1..b2d7d901 100755
--- a/adapters/src/acs/integrations/cursor/adapter.py
+++ b/adapters/src/acs/integrations/cursor/adapter.py
@@ -72,28 +72,28 @@
 
 from acs.client import (
     Config,
-    Decided,
     FailPosture,
     Handshake,
     Refused,
     Resolution,
     Undecided,
     audit_event,
-    exchange,
+    close_turn,
+    current_turn,
     load_session_state,
     open_session,
+    open_turn,
     record_step,
-    save_session_state,
-    sign_envelope,
+    round_trip,
 )
 from acs.core import (
-    ACS_VERSION,
-    KNOWN_DECISIONS,
+    Decision,
     coerce_uuid,
-    iso8601_now,
     method_evaluated,
     modify_composition_violation,
     normalize_decision,
+    request_envelope,
+    turn_start_envelope,
 )
 from acs.integrations.cursor.hook import PERMISSION_EVENTS
 
@@ -264,27 +264,6 @@ def _int_ms(raw: Any) -> int | None:
 })
 
 
-def _current_turn_id(event: dict[str, Any]) -> str | None:
-    sid = _session_id(event)
-    if not sid:
-        return None
-    return load_session_state(sid, workspace=_workspace(event)).get("turn_id")
-
-
-def _open_turn(event: dict[str, Any], turn_id: str) -> None:
-    sid = _session_id(event)
-    st = load_session_state(sid, workspace=_workspace(event))
-    st["turn_id"] = turn_id
-    save_session_state(sid, st, workspace=_workspace(event))
-
-
-def _close_turn(event: dict[str, Any]) -> None:
-    sid = _session_id(event)
-    st = load_session_state(sid, workspace=_workspace(event))
-    st.pop("turn_id", None)
-    save_session_state(sid, st, workspace=_workspace(event))
-
-
 def _wrap_arguments(raw: dict[str, Any]) -> dict[str, Any]:
     """tool-call-request.json:26-37 — each arg is {value, provenance?}."""
     return {k: {"value": v} for k, v in (raw or {}).items()}
@@ -499,7 +478,7 @@ def _payload_turn_end(event: dict[str, Any]) -> dict[str, Any]:
     2026-08-22)."""
     outcome = _CURSOR_STOP_STATUS_TO_OUTCOME.get(
         (event.get("status") or "").lower(), "completed")
-    return {"turn_id": _current_turn_id(event) or "", "outcome": outcome}
+    return {"turn_id": current_turn(_session_id(event), workspace=_workspace(event)) or "", "outcome": outcome}
 
 
 def _payload_subagent_start(event: dict[str, Any]) -> dict[str, Any]:
@@ -603,26 +582,16 @@ def build_request(event_name: str, event: dict[str, Any]) -> dict[str, Any]:
     if not session_id:
         return {}
 
-    metadata: dict[str, Any] = {
-        "agent_id": _agent_id(event),
-        "session_id": session_id,
-        "platform": "cursor",
-        "adapter_version": ADAPTER_VERSION,
-        "cursor_event": event_name,
-    }
-    ws = event.get("cwd") or _first_workspace_root(event)
-    if ws:
-        metadata["workspace_path"] = ws
+    metadata = _metadata(event_name, event)
 
     # Stamp the open turn's id on in-turn steps (request-envelope.json:69).
     if method in _IN_TURN_METHODS:
-        tid = _current_turn_id(event)
+        tid = current_turn(_session_id(event), workspace=_workspace(event))
         if tid:
             metadata["turn_id"] = tid
 
-    # steps/turnEnd needs a turn_id (>=1); with no open turn there is
-    # nothing to close honestly, so skip the emission (main audits it).
-    if method == "steps/turnEnd" and not _current_turn_id(event):
+    # turnEnd needs an open turn; with none the emission is skipped, and main audits it.
+    if method == "steps/turnEnd" and not current_turn(_session_id(event), workspace=_workspace(event)):
         return {}
 
     # Pin *Request request_ids deterministically (uuid5 of `tool_use_id`
@@ -650,53 +619,29 @@ def build_request(event_name: str, event: dict[str, Any]) -> dict[str, Any]:
                            "honestly populate entries_to_compact (>=1 required)")
         return {}
 
-    envelope = {
-        "jsonrpc": "2.0",
-        "id": str(uuid.uuid4()),
-        "method": method,
-        "params": {
-            "acs_version": ACS_VERSION,
-            "request_id": request_id,
-            "timestamp": iso8601_now(),
-            "metadata": metadata,
-            "payload": payload,
-        },
-    }
-    sign_envelope(envelope, session_id=session_id)
-    return envelope
+    return request_envelope(method, request_id=request_id, metadata=metadata, payload=payload)
 
 
-def build_turn_start_request(event: dict[str, Any], turn_id: str) -> dict[str, Any]:
-    """steps/turnStart opening the turn for a beforeSubmitPrompt. turn_id
-    SHOULD equal the envelope's request_id (turn-start.json), so we use it
-    for both. Decision-eligible: a Guardian MAY deny to block the turn
-    (blocks the prompt via exit code 2)."""
-    session_id = _session_id(event)
+def _metadata(event_name: str, event: dict[str, Any]) -> dict[str, Any]:
     metadata: dict[str, Any] = {
         "agent_id": _agent_id(event),
-        "session_id": session_id,
+        "session_id": _session_id(event),
         "platform": "cursor",
         "adapter_version": ADAPTER_VERSION,
-        "cursor_event": "beforeSubmitPrompt",
-        "turn_id": turn_id,
+        "cursor_event": event_name,
     }
     ws = event.get("cwd") or _first_workspace_root(event)
     if ws:
         metadata["workspace_path"] = ws
-    envelope = {
-        "jsonrpc": "2.0",
-        "id": str(uuid.uuid4()),
-        "method": "steps/turnStart",
-        "params": {
-            "acs_version": ACS_VERSION,
-            "request_id": turn_id,
-            "timestamp": iso8601_now(),
-            "metadata": metadata,
-            "payload": {"turn_id": turn_id, "triggered_by": "user_message"},
-        },
-    }
-    sign_envelope(envelope, session_id=session_id)
-    return envelope
+    return metadata
+
+
+def build_turn_start_request(event: dict[str, Any], turn_id: str) -> dict[str, Any]:
+    """steps/turnStart opening the turn for a beforeSubmitPrompt. turn_id
+    SHOULD equal the envelope's request_id (turn-start.json), so we use it
+    for both. Decision-eligible: a Guardian MAY deny to block the turn
+    (blocks the prompt via exit code 2)."""
+    return turn_start_envelope(_metadata("beforeSubmitPrompt", event), turn_id)
 
 
 def _open_session(event: dict[str, Any], config: Config) -> Handshake:
@@ -726,112 +671,105 @@ def _open_session(event: dict[str, Any], config: Config) -> Handshake:
 #     internal __exit_code/_reasoning keys consumed by main()
 #   - everything else: no response shape (Stop, SessionStart, SessionEnd, etc.)
 
-def _translate_permission(decision: str, reasoning: str,
-                           modifications: dict, event_name: str,
-                           original_input: dict) -> dict[str, Any]:
-    if decision in PERMISSION_MAP:
-        return _permission_response(PERMISSION_MAP[decision], reasoning)
-    if decision == "modify":
+def _translate_permission(decision: Decision, event_name: str, original_input: dict) -> dict[str, Any]:
+    if decision.disposition in PERMISSION_MAP:
+        return _permission_response(PERMISSION_MAP[decision.disposition], decision.reasoning)
+    if decision.disposition == "modify":
         # §6.3:146 — refuse an un-interpretable modifications object
         # (both shapes, or overlapping structured targets) rather than
         # half-apply it.
-        violation = modify_composition_violation(modifications)
+        violation = modify_composition_violation(decision.modifications)
         if violation:
             audit_event("modify_composition_invalid", event=event_name,
                         detail=violation)
             return _permission_response(
                 "deny", f"MODIFY refused ({violation}); failing closed per §6.3:146")
-        if modifications.get("redactions"):
+        if decision.modifications.get("redactions"):
             # updated_input cannot express JSON-Pointer redactions; never
             # apply only the override half of a compound decision.
             audit_event("modify_unapplied_redactions", event=event_name)
             return _permission_response(
                 "deny", "MODIFY refused: Cursor cannot apply the requested "
                 "redactions without dropping part of the decision")
-        overrides = modifications.get("parameter_overrides")
+        overrides = decision.modifications.get("parameter_overrides")
         # A non-dict override cannot be applied — fail closed per §6.3:146.
         if isinstance(overrides, dict) and event_name == "preToolUse":
             # parameter_overrides are per-argument EDITS while Cursor's
             # updated_input REPLACES the whole input, so merge onto the
             # original.
             merged = {**original_input, **overrides}
-            return _permission_response("allow", reasoning, updated_input=merged)
+            return _permission_response("allow", decision.reasoning, updated_input=merged)
         return _permission_response("deny",
-                                     f"MODIFY substituted to DENY: {reasoning}")
+                                     f"MODIFY substituted to DENY: {decision.reasoning}")
     return {}
 
 
-def _translate_post_tool(decision: str, reasoning: str,
-                          modifications: dict, event_name: str) -> dict[str, Any]:
+def _translate_post_tool(decision: Decision, event_name: str) -> dict[str, Any]:
     """Only `postToolUse` documents output fields (`additional_context`,
     and `updated_mcp_tool_output` for MCP tools); every other post-tool
     event documents "No output fields currently supported" (docs
     2026-08-22), so an arrived restrictive decision there is recorded and
     nothing is emitted."""
     if event_name != "postToolUse":
-        if decision in ("deny", "modify", "ask", "defer"):
-            audit_event("unenforceable_decision", decision=decision,
-                        event=event_name, reason=reasoning or "(none)")
+        if decision.disposition in ("deny", "modify", "ask", "defer"):
+            audit_event("unenforceable_decision", decision=decision.disposition,
+                        event=event_name, reason=decision.reasoning or "(none)")
         return {}
-    if decision == "deny":
+    if decision.disposition == "deny":
         # postToolUse fires after execution — no block exists; the denial
         # reaches the conversation as additional_context and is audited.
         audit_event("unenforceable_decision", decision="deny",
-                    event=event_name, reason=reasoning or "(none)")
+                    event=event_name, reason=decision.reasoning or "(none)")
         return _post_tool_response(
             additional_context=f"ACS Guardian denied this result: "
-                               f"{reasoning or '(no reasoning)'}")
-    if decision == "modify":
-        violation = modify_composition_violation(modifications)
+                               f"{decision.reasoning or '(no reasoning)'}")
+    if decision.disposition == "modify":
+        violation = modify_composition_violation(decision.modifications)
         if violation:
             audit_event("modify_composition_invalid", event=event_name,
                         detail=violation)
             return _post_tool_response(
                 additional_context=f"MODIFY refused ({violation})")
-        updated = modifications.get("modified_content")
+        updated = decision.modifications.get("modified_content")
         if updated is not None:
             # Docs show an OBJECT value; ACS modified_content is a string.
             # Parse JSON content when it is JSON, else pass the string.
             return _post_tool_response(
                 updated_mcp_tool_output=_parse_json_field(updated))
         return _post_tool_response(
-            additional_context=f"MODIFY received: {reasoning}")
-    if reasoning:
-        return _post_tool_response(additional_context=reasoning)
+            additional_context=f"MODIFY received: {decision.reasoning}")
+    if decision.reasoning:
+        return _post_tool_response(additional_context=decision.reasoning)
     return {}
 
 
-def _translate_before_submit_prompt(decision: str, reasoning: str,
-                                     modifications: dict,
-                                     event_name: str) -> dict[str, Any]:
+def _translate_before_submit_prompt(decision: Decision, event_name: str) -> dict[str, Any]:
     """The prompt hook can only allow (exit 0) or block (exit 2), so every
     non-allow decision fails closed per §6.4; returns internal
     __exit_code/_reasoning keys for main()."""
-    if decision == "allow":
+    if decision.disposition == "allow":
         return {"__exit_code": 0, "_reasoning": None}
-    if decision != "deny":
-        # Audit the substitution for decisions the prompt hook can't apply.
+    if decision.disposition != "deny":
         audit_event("prompt_decision_substituted_block",
-                    disposition=decision, event="beforeSubmitPrompt")
+                    disposition=decision.disposition, event="beforeSubmitPrompt")
     return {"__exit_code": 2,
-            "_reasoning": reasoning or f"{decision} on prompt not applicable — blocking"}
+            "_reasoning": decision.reasoning or f"{decision.disposition} on prompt not applicable — blocking"}
 
 
 def translate_response(acs_response: dict[str, Any], event_name: str,
                        original_input: dict | None = None) -> dict[str, Any]:
     # normalize_decision coerces every field to a safe type and never
     # raises — an uncaught throw here would exit 1 and Cursor would proceed.
-    decision, reasoning, modifications = normalize_decision(
-        acs_response.get("result"))
+    decision = normalize_decision(acs_response.get("result"))
     original_input = original_input or {}
 
     # An arrived-but-uninterpretable decision fails closed; the exact
     # spec basis is unsettled (§6.4:156 vs. arrived-and-unusable),
     # tracked in issue #32.
-    if decision not in KNOWN_DECISIONS:
+    if not decision.known:
         audit_event("unusable_disposition",
-                    disposition=decision or "(missing)", event=event_name)
-        reason = (f"unusable Guardian disposition '{decision}' — cannot be "
+                    disposition=decision.disposition or "(missing)", event=event_name)
+        reason = (f"unusable Guardian disposition '{decision.disposition}' — cannot be "
                   f"interpreted; failing closed (see issue #32)")
         if event_name in PERMISSION_EVENTS:
             return _permission_response("deny", reason)
@@ -842,12 +780,11 @@ def translate_response(acs_response: dict[str, Any], event_name: str,
         return {}
 
     if event_name in PERMISSION_EVENTS:
-        return _translate_permission(decision, reasoning, modifications,
-                                     event_name, original_input)
+        return _translate_permission(decision, event_name, original_input)
     if event_name in POST_TOOL_EVENTS:
-        return _translate_post_tool(decision, reasoning, modifications, event_name)
+        return _translate_post_tool(decision, event_name)
     if event_name == "beforeSubmitPrompt":
-        return _translate_before_submit_prompt(decision, reasoning, modifications, event_name)
+        return _translate_before_submit_prompt(decision, event_name)
 
     # Remaining events are observational with no response shape — empty
     # dict skips stdout emission.
@@ -856,20 +793,6 @@ def translate_response(acs_response: dict[str, Any], event_name: str,
 
 # ─── Main flow ──────────────────────────────────────────────────────────────
 
-def _guardian_roundtrip(request: dict[str, Any], event_name: str, event: dict[str, Any],
-                        guardian_url: str, posture: FailPosture) -> tuple[dict | None, int | None]:
-    """POST one request. Returns (response, None) for a usable decision, or
-    (None, exit_code) once a failure has been settled in the event's shape;
-    shared by the turnStart pre-step and the main step."""
-    session_id = _session_id(event)
-    outcome = exchange(request, guardian_url=guardian_url, timeout_s=posture.timeout_for(request.get("method", "")),
-                       session_id=session_id)
-    if isinstance(outcome, Decided):
-        return outcome.response, None
-    sys.stderr.write(f"acs-adapter: {outcome.message or outcome.cause}\n")
-    return None, _settle(posture.resolve(outcome), event_name, session_id)
-
-
 def main() -> int:
     config = Config.from_env(os.environ)
     if len(sys.argv) > 1 and sys.argv[1] == "--version":
@@ -936,20 +859,19 @@ def _govern(event_name: str, event: dict[str, Any], session_id: str | None,
     if event_name == "beforeSubmitPrompt":
         turn_id = str(uuid.uuid4())
         if method_evaluated(server_hello, "steps/turnStart"):
-            ts_response, terminal = _guardian_roundtrip(
-                build_turn_start_request(event, turn_id), "beforeSubmitPrompt", event,
-                config.guardian_url, posture)
-            if terminal is not None:
-                return terminal
-            ts_decision, ts_reason, _ = normalize_decision(ts_response.get("result"))
-            if ts_decision != "allow":
+            result = round_trip(build_turn_start_request(event, turn_id), guardian_url=config.guardian_url,
+                                posture=posture, session_id=session_id)
+            if isinstance(result, Resolution):
+                return _settle(result, "beforeSubmitPrompt", session_id)
+            turn_start = normalize_decision(result.response.get("result"))
+            if turn_start.disposition != "allow":
                 # A turn boundary can't be modify/ask'd — any non-allow
                 # blocks the prompt (exit 2).
                 audit_event("turn_start_blocked", turn_id=turn_id,
-                            disposition=ts_decision or "(unusable)",
+                            disposition=turn_start.disposition or "(unusable)",
                             event="beforeSubmitPrompt")
-                msg = (f"turn blocked at start ({ts_decision or 'unusable'}): "
-                       f"{ts_reason}")
+                msg = (f"turn blocked at start ({turn_start.disposition or 'unusable'}): "
+                       f"{turn_start.reasoning}")
                 sys.stderr.write(f"acs-adapter: {msg}\n")
                 sys.stdout.write(json.dumps({"continue": False,
                                              "user_message": msg}) + "\n")
@@ -957,7 +879,7 @@ def _govern(event_name: str, event: dict[str, Any], session_id: str | None,
         else:
             audit_event("method_not_evaluated", method="steps/turnStart",
                         event=event_name, session_id=session_id)
-        _open_turn(event, turn_id)
+        open_turn(session_id, turn_id, workspace=_workspace(event))
 
     try:
         request = build_request(event_name, event)
@@ -980,13 +902,11 @@ def _govern(event_name: str, event: dict[str, Any], session_id: str | None,
         audit_event("method_not_evaluated", method=method, event=event_name,
                     session_id=session_id)
         if event_name == "stop":
-            _close_turn(event)
+            close_turn(session_id, workspace=_workspace(event))
         return _explicit_allow(event_name)
-    if (method == "steps/toolCallResult"
-            and not method_evaluated(server_hello, "steps/toolCallRequest")
-            and request["params"]["payload"].pop("request_id_ref", None) is not None):
+    if method == "steps/toolCallResult" and not method_evaluated(server_hello, "steps/toolCallRequest"):
         # The originating request was never sent, so citing it would dangle.
-        sign_envelope(request, session_id=request["params"]["metadata"]["session_id"])
+        request["params"]["payload"].pop("request_id_ref", None)
 
     # Track this step in session state so subsequent subagentStart /
     # preCompact events can cite a real parent_step_id / entries_to_compact.
@@ -997,10 +917,10 @@ def _govern(event_name: str, event: dict[str, Any], session_id: str | None,
     except Exception:  # noqa: BLE001
         pass
 
-    response, terminal = _guardian_roundtrip(request, event_name, event,
-                                             config.guardian_url, posture)
-    if terminal is not None:
-        return terminal
+    result = round_trip(request, guardian_url=config.guardian_url, posture=posture, session_id=session_id)
+    if isinstance(result, Resolution):
+        return _settle(result, event_name, session_id)
+    response = result.response
 
     # A bug escaping translate_response would exit 1 = Cursor proceeds;
     # this guard makes any translation bug fail closed instead.
@@ -1014,7 +934,7 @@ def _govern(event_name: str, event: dict[str, Any], session_id: str | None,
 
     # Close the turn after the turnEnd has been emitted/recorded.
     if event_name == "stop":
-        _close_turn(event)
+        close_turn(session_id, workspace=_workspace(event))
 
     if event_name == "beforeSubmitPrompt":
         exit_code = out.pop("__exit_code", 0)
@@ -1052,6 +972,8 @@ def _settle(resolution: Resolution, event_name: str, session_id: str | None) ->
     A proceed carries Cursor's explicit allow: failClosed covers hook crashes
     and timeouts, and the posture already decided this step.
     """
+    if resolution.message:
+        sys.stderr.write(f"acs-adapter: {resolution.message}\n")
     if resolution.proceed:
         audit_event(resolution.audit_type, cause=resolution.cause, event=event_name,
                     session_id=session_id, **resolution.fields)
diff --git a/adapters/src/acs/integrations/nat/adapter.py b/adapters/src/acs/integrations/nat/adapter.py
index 79cab72a..b034c8d7 100644
--- a/adapters/src/acs/integrations/nat/adapter.py
+++ b/adapters/src/acs/integrations/nat/adapter.py
@@ -61,23 +61,20 @@
 try:
     from acs.client import (
         Config,
-        Decided,
         FailPosture,
         Handshake,
         Negotiated,
         Resolution,
         Undecided,
         audit_event,
-        exchange,
         open_session,
-        sign_envelope,
+        round_trip,
     )
     from acs.core import (
-        ACS_VERSION,
-        iso8601_now as _common_iso8601_now,
         method_evaluated,
         modify_composition_violation,
         normalize_decision,
+        request_envelope,
     )
 except ImportError as _acs_import_error:
     # Without acs.core (rfc8785, §10) NAT would run ungoverned, so the import fails loudly.
@@ -185,10 +182,6 @@ class ACSMiddlewareConfig(FunctionMiddlewareBaseConfig, name="acs_guardian"):  #
 # ----- Helpers (module-scope so tests can exercise them without instantiating the middleware) -----
 
 
-def _iso8601_now() -> str:
-    return _common_iso8601_now()
-
-
 def _coerce_uuid(raw: str | None) -> str:
     """request-envelope.json:66 wants session_id as UUID. Accept a UUID
     directly; otherwise derive a stable UUID5 from whatever NAT gave us."""
@@ -231,9 +224,6 @@ def _stringify_step_data(data: Any) -> str:
     return str(data)
 
 
-KNOWN_DECISIONS = frozenset({"allow", "deny", "modify", "ask", "defer"})
-
-
 def _extract_arguments(context: Any) -> dict[str, Any]:
     """Build a flat {arg_name: value} dict from NAT's invocation context.
 
@@ -505,26 +495,11 @@ def _emit_lifecycle_hook(self, method: str, payload: dict) -> None:
             audit_event("method_not_evaluated", method=method,
                         session_id=self._session_id)
             return
-        request = {
-            "jsonrpc": "2.0",
-            "id": str(uuid.uuid4()),
-            "method": method,
-            "params": {
-                "acs_version": ACS_VERSION,
-                "request_id": str(uuid.uuid4()),
-                "timestamp": _iso8601_now(),
-                "metadata": {
-                    "agent_id": self._agent_id,
-                    "session_id": self._session_id,
-                    "platform": "nat",
-                },
-                "payload": payload,
-            },
-        }
-        sign_envelope(request, session_id=self._session_id)
-        outcome = self._exchange(request)
-        if not isinstance(outcome, Decided):
-            audit_event("lifecycle_hook_failed", **{**outcome.fields, "cause": outcome.cause,
+        request = request_envelope(method, request_id=str(uuid.uuid4()), metadata=self._metadata(),
+                                   payload=payload)
+        result = self._round_trip(request)
+        if isinstance(result, Resolution):
+            audit_event("lifecycle_hook_failed", **{**result.fields, "cause": result.cause,
                                                     "method": method, "session_id": self._session_id})
 
     def _correlation_request_id(self, context) -> str:
@@ -588,23 +563,22 @@ async def pre_invoke(self, context):
                 request_id=correlation_id,
             )
         except Exception as e:  # noqa: BLE001
-            outcome = Undecided("adapter_exception", {"error": str(e)}, f"adapter error: {e}")
+            result = self._posture.resolve(Undecided("adapter_exception", {"error": str(e)},
+                                                     f"adapter error: {e}"))
         else:
-            outcome = self._exchange(request)
-        if not isinstance(outcome, Decided):
-            return self._settle_pre(context, self._posture.resolve(outcome), outcome.message)
-        response = outcome.response
+            result = self._round_trip(request)
+        if isinstance(result, Resolution):
+            return self._settle_pre(context, result)
+        response = result.response
 
-        # Total coercion: malformed verdict fields must not throw here.
-        decision, reasoning, mods = normalize_decision(
-            (response or {}).get("result"))
+        decision = normalize_decision((response or {}).get("result"))
 
-        if decision == "allow":
+        if decision.disposition == "allow":
             return None
-        if decision == "deny":
-            return self._block(context, reasoning or "denied by Guardian")
-        if decision == "modify":
-            violation = modify_composition_violation(mods)
+        if decision.disposition == "deny":
+            return self._block(context, decision.reasoning or "denied by Guardian")
+        if decision.disposition == "modify":
+            violation = modify_composition_violation(decision.modifications)
             if violation:
                 audit_event("modify_composition_invalid",
                             session_id=self._session_id,
@@ -612,7 +586,7 @@ async def pre_invoke(self, context):
                 return self._block(
                     context,
                     f"MODIFY refused ({violation}); failing closed per §6.3:146")
-            if mods.get("redactions"):
+            if decision.modifications.get("redactions"):
                 # NAT has no JSON-Pointer redaction primitive; applying only
                 # parameter_overrides would drop part of the compound decision.
                 audit_event("modify_unapplied_redactions",
@@ -621,25 +595,24 @@ async def pre_invoke(self, context):
                 return self._block(
                     context, "MODIFY refused: NAT cannot apply the requested "
                     "redactions without dropping part of the decision")
-            overrides = mods.get("parameter_overrides")
+            overrides = decision.modifications.get("parameter_overrides")
             if isinstance(overrides, dict):
                 _apply_overrides_to_context(context, overrides)
                 return context
-            return self._block(context, f"MODIFY substituted to DENY: {reasoning}")
-        if decision in ("ask", "defer"):
-            # No pause-and-resume primitive on this boundary: substitute
-            # block; ASK/DEFER deployments compose with NAT's HITL middleware.
-            return self._block(context, f"{decision}: {reasoning}")
+            return self._block(context, f"MODIFY substituted to DENY: {decision.reasoning}")
+        if decision.disposition in ("ask", "defer"):
+            # No pause primitive here, so ask and defer block; NAT's HITL middleware composes for approvals.
+            return self._block(context, f"{decision.disposition}: {decision.reasoning}")
 
         # An arrived-but-uninterpretable decision fails closed; the spec
         # basis is tracked in issue #32.
         audit_event("unusable_disposition",
-                    disposition=decision or "(missing)",
+                    disposition=decision.disposition or "(missing)",
                     session_id=self._session_id,
                     method="steps/toolCallRequest")
         return self._block(
             context,
-            f"unusable Guardian disposition '{decision}' — cannot be "
+            f"unusable Guardian disposition '{decision.disposition}' — cannot be "
             f"interpreted; failing closed (see issue #32)")
 
     async def post_invoke(self, context):
@@ -679,29 +652,26 @@ async def post_invoke(self, context):
                 request_id_ref=correlation_id if request_sent else None,
             )
         except Exception as e:  # noqa: BLE001
-            outcome = Undecided("adapter_exception", {"error": str(e)}, f"adapter error: {e}")
+            result = self._posture.resolve(Undecided("adapter_exception", {"error": str(e)},
+                                                     f"adapter error: {e}"))
         else:
-            outcome = self._exchange(request)
-        if not isinstance(outcome, Decided):
-            return self._settle_post(context, self._posture.resolve(outcome))
-        response = outcome.response
-
-        decision, reasoning, mods = normalize_decision(
-            (response or {}).get("result"))
-
-        if decision == "deny":
-            # The tool already executed; the redaction signal is
-            # output=None plus the audit event (strict Pydantic model —
-            # ad-hoc attributes raise ValidationError).
+            result = self._round_trip(request)
+        if isinstance(result, Resolution):
+            return self._settle_post(context, result)
+        response = result.response
+
+        decision = normalize_decision((response or {}).get("result"))
+
+        if decision.disposition == "deny":
             _redact_output(context)
             audit_event("post_invoke_redacted",
                         cause="guardian_deny",
                         session_id=self._session_id,
                         method="steps/toolCallResult",
-                        reasoning=reasoning or "output redacted by Guardian")
+                        reasoning=decision.reasoning or "output redacted by Guardian")
             return context
-        if decision == "modify":
-            violation = modify_composition_violation(mods)
+        if decision.disposition == "modify":
+            violation = modify_composition_violation(decision.modifications)
             if violation:
                 audit_event("modify_composition_invalid",
                             session_id=self._session_id,
@@ -712,7 +682,7 @@ async def post_invoke(self, context):
                             session_id=self._session_id,
                             method="steps/toolCallResult")
                 return context
-            modified_content = mods.get("modified_content")
+            modified_content = decision.modifications.get("modified_content")
             if modified_content is not None:
                 context.output = modified_content
                 return context
@@ -724,11 +694,9 @@ async def post_invoke(self, context):
                         session_id=self._session_id,
                         method="steps/toolCallResult")
             return context
-        if decision not in KNOWN_DECISIONS:
-            # Arrived-but-uninterpretable decision: redact (fail closed);
-            # spec basis tracked in issue #32.
+        if not decision.known:
             audit_event("unusable_disposition",
-                        disposition=decision or "(missing)",
+                        disposition=decision.disposition or "(missing)",
                         session_id=self._session_id,
                         method="steps/toolCallResult")
             _redact_output(context)
@@ -736,15 +704,20 @@ async def post_invoke(self, context):
                         cause="unusable_disposition",
                         session_id=self._session_id,
                         method="steps/toolCallResult",
-                        decision=decision)
+                        decision=decision.disposition)
             return context
         return None
 
     # ----- helpers -----
 
-    def _exchange(self, request: dict):
-        return exchange(request, guardian_url=self._acs.guardian_url,
-                        timeout_s=self._posture.timeout_for(request.get("method", "")), session_id=self._session_id)
+    def _round_trip(self, request: dict):
+        return round_trip(request, guardian_url=self._acs.guardian_url, posture=self._posture,
+                          session_id=self._session_id)
+
+    def _metadata(self) -> dict[str, Any]:
+        """The envelope metadata every step of this middleware carries."""
+        return {"agent_id": self._agent_id, "session_id": self._session_id,
+                "platform": "nat", "adapter_version": ADAPTER_VERSION}
 
     def _settle_pre(self, context, resolution: Resolution, message: str = ""):
         """Settle a pre_invoke step without a usable decision: block unless the resolution proceeds, and record why."""
@@ -753,7 +726,7 @@ def _settle_pre(self, context, resolution: Resolution, message: str = ""):
                                               "method": "steps/toolCallRequest"})
         if resolution.proceed:
             return None
-        return self._block(context, message or resolution.cause)
+        return self._block(context, message or resolution.message or resolution.cause)
 
     def _settle_post(self, context, resolution: Resolution):
         """Settle a post_invoke step without a usable decision: redact the output the Guardian never vetted unless the resolution lets it through, and record why."""
@@ -782,13 +755,7 @@ def _build_request(
         request_id: str | None = None,
         request_id_ref: str | None = None,
     ) -> dict:
-        """Build a signed ACS request envelope matching request-envelope.json."""
-        metadata = {
-            "agent_id": self._agent_id,
-            "session_id": self._session_id,
-            "platform": "nat",
-            "adapter_version": ADAPTER_VERSION,
-        }
+        """Build an ACS request envelope matching request-envelope.json; the round trip signs it."""
         if method == "steps/toolCallRequest":
             payload: dict[str, Any] = {
                 "tool": {"name": tool_name},
@@ -809,20 +776,8 @@ def _build_request(
             if request_id_ref:
                 payload["request_id_ref"] = request_id_ref
 
-        envelope = {
-            "jsonrpc": "2.0",
-            "id": str(uuid.uuid4()),
-            "method": method,
-            "params": {
-                "acs_version": ACS_VERSION,
-                "request_id": request_id or str(uuid.uuid4()),
-                "timestamp": _iso8601_now(),
-                "metadata": metadata,
-                "payload": payload,
-            },
-        }
-        sign_envelope(envelope, session_id=self._session_id)
-        return envelope
+        return request_envelope(method, request_id=request_id or str(uuid.uuid4()),
+                                metadata=self._metadata(), payload=payload)
 
 # ----- NAT registration -----
 
diff --git a/adapters/tests/client/test_round_trip.py b/adapters/tests/client/test_round_trip.py
new file mode 100644
index 00000000..1128c974
--- /dev/null
+++ b/adapters/tests/client/test_round_trip.py
@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import io
+import json
+import os
+import tempfile
+import unittest
+import urllib.error
+from pathlib import Path
+from typing import Any
+from unittest import mock
+
+from acs.client import (Decided, FailPosture, Resolution, close_turn, current_turn, open_turn, round_trip,
+                        verify_signature)
+from acs.client import _exchange, _store
+from acs.core import derive_session_key, request_envelope, sign_with_key
+
+SECRET = "round-trip-secret"
+SESSION = "44444444-4444-4444-8444-444444444444"
+URL = "http://127.0.0.1:8787/acs"
+PROCEED = FailPosture(deny_on_failure=False, timeout_s=2.0, per_method_timeout_s={"steps/toolCallResult": 0.3})
+DENY = FailPosture(deny_on_failure=True, timeout_s=2.0)
+
+
+def _request(**payload: Any) -> dict:
+    return request_envelope("steps/toolCallResult", request_id="55555555-5555-4555-8555-555555555555",
+                            metadata={"agent_id": "a", "session_id": SESSION, "platform": "test"},
+                            payload=payload)
+
+
+@mock.patch.dict(os.environ, {"ACS_HMAC_SECRET": SECRET, "ACS_HMAC_SECRET_FILE": ""})
+class RoundTrip(unittest.TestCase):
+    def _send(self, request: dict, answer, posture: FailPosture = PROCEED):
+        sent: list[tuple[dict, float]] = []
+
+        def urlopen(req: Any, timeout: float):
+            body = json.loads(req.data)
+            sent.append((body, timeout))
+            return answer(body)
+
+        with mock.patch.object(_exchange, "_open", side_effect=urlopen):
+            result = round_trip(request, guardian_url=URL, posture=posture, session_id=SESSION)
+        return result, sent
+
+    @staticmethod
+    def _signed_allow(body: dict):
+        response = sign_with_key({"jsonrpc": "2.0", "id": body["id"], "result": {
+            "request_id": body["params"]["request_id"], "decision": "allow"}},
+            derive_session_key(SECRET.encode(), SESSION))
+        return io.BytesIO(json.dumps(response).encode())
+
+    def test_an_edit_after_building_is_covered_by_the_one_signature(self) -> None:
+        request = _request(request_id_ref="66666666-6666-4666-8666-666666666666", tool={"name": "x"})
+        request["params"]["payload"].pop("request_id_ref")
+        result, sent = self._send(request, self._signed_allow)
+        self.assertIsInstance(result, Decided)
+        wire = sent[0][0]
+        self.assertNotIn("request_id_ref", wire["params"]["payload"])
+        self.assertTrue(verify_signature(wire, session_id=SESSION), "the signature must cover what was sent")
+
+    def test_each_method_waits_for_its_negotiated_timeout(self) -> None:
+        _, sent = self._send(_request(), self._signed_allow)
+        self.assertEqual(sent[0][1], 0.3)
+
+    def test_failures_resolve_with_their_message(self) -> None:
+        def refused(body: dict):
+            raise urllib.error.URLError(ConnectionRefusedError(61, "refused"))
+        for posture, proceed, audit in ((PROCEED, True, "fail_open_bypass"),
+                                        (DENY, False, "decision_failure_fail_closed")):
+            with self.subTest(audit):
+                result, _ = self._send(_request(), refused, posture)
+                self.assertIsInstance(result, Resolution)
+                self.assertEqual((result.proceed, result.audit_type, result.cause), (proceed, audit, "transport_failure"))
+                self.assertIn("Guardian unreachable", result.message)
+
+    def test_a_request_that_cannot_be_signed_is_an_adapter_error(self) -> None:
+        request = _request(note="\ud800")
+        result, sent = self._send(request, self._signed_allow, DENY)
+        self.assertEqual(sent, [], "nothing unsigned goes on the wire")
+        self.assertEqual((result.proceed, result.cause), (False, "adapter_exception"))
+
+
+class Turns(unittest.TestCase):
+    def test_open_current_close_per_workspace(self) -> None:
+        with tempfile.TemporaryDirectory() as d, mock.patch.object(_store, "_SESSION_STATE_DIR", Path(d)):
+            self.assertIsNone(current_turn(SESSION, workspace="/a"))
+            open_turn(SESSION, "turn-a", workspace="/a")
+            open_turn(SESSION, "turn-b", workspace="/b")
+            self.assertEqual((current_turn(SESSION, workspace="/a"), current_turn(SESSION, workspace="/b")),
+                             ("turn-a", "turn-b"))
+            close_turn(SESSION, workspace="/a")
+            self.assertEqual((current_turn(SESSION, workspace="/a"), current_turn(SESSION, workspace="/b")),
+                             (None, "turn-b"))
+            self.assertIsNone(current_turn(None, workspace="/b"))
+
+    def test_a_turn_id_that_is_not_a_string_is_no_turn(self) -> None:
+        with tempfile.TemporaryDirectory() as d, mock.patch.object(_store, "_SESSION_STATE_DIR", Path(d)):
+            _store.save_session_state(SESSION, {"turn_id": ["not", "a", "turn"]})
+            self.assertIsNone(current_turn(SESSION))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/adapters/tests/common/test_session_gate.py b/adapters/tests/common/test_session_gate.py
index dcfff84b..3d559d6b 100644
--- a/adapters/tests/common/test_session_gate.py
+++ b/adapters/tests/common/test_session_gate.py
@@ -272,6 +272,31 @@ def test_a_later_hooks_required_profile_refuses_a_stored_negotiation(self) -> No
                 p, denied, _ = run(guardian.url(), str(uuid.uuid4()), ACS_REQUIRE_PROFILES="acs-core")
                 self.assertTrue(denied, "control: a fresh session with the requirement is refused too")
 
+    def test_a_denied_turn_start_blocks_the_prompt_and_sends_no_message(self) -> None:
+        guardian = ProgrammableGuardian(hmac_secret=SECRET)
+        guardian.handlers["steps/turnStart"] = lambda req: {
+            "type": "final", "acs_version": "0.1.0", "request_id": req["params"]["request_id"],
+            "decision": "deny", "reasoning": "policy", "chain_hash": "0" * 64}
+        guardian.start()
+        self.addCleanup(guardian.stop)
+        runs = {"claude-code": ([str(BIN / "acs-claude-code-hook")],
+                                {"session_id": str(uuid.uuid4()), "cwd": "/tmp",
+                                 "hook_event_name": "UserPromptSubmit", "prompt": "hello"}),
+                "cursor": ([str(BIN / "acs-cursor-hook"), "beforeSubmitPrompt"],
+                           {"conversation_id": str(uuid.uuid4()), "workspace_roots": ["/tmp"], "prompt": "hello"})}
+        for name, (argv, event) in runs.items():
+            with self.subTest(name):
+                guardian.reset()
+                p = subprocess.run(argv, input=json.dumps(event), capture_output=True, text=True,
+                                   env=self._env(guardian.url()), timeout=30)
+                out = json.loads(p.stdout or "{}")
+                blocked = (out.get("decision") == "block" if name == "claude-code"
+                           else p.returncode == 2 and out.get("continue") is False)
+                self.assertTrue(blocked, (p.returncode, p.stdout, p.stderr[-300:]))
+                self.assertIn("turn_start_blocked", [r["acs_audit_event"] for r in _audit_records(p.stderr)])
+                self.assertEqual(guardian.methods_seen(), ["handshake/hello", "steps/turnStart"],
+                                 "a blocked prompt's userMessage is never sent")
+
     def test_negotiated_per_method_timeouts_bound_their_own_methods(self) -> None:
         """A 300 ms toolCallRequest timeout ends that wait; toolCallResult waits under the 5 s default and its deny applies."""
         guardian = ProgrammableGuardian(hmac_secret=SECRET)
diff --git a/adapters/tests/core/test_decision.py b/adapters/tests/core/test_decision.py
new file mode 100644
index 00000000..ade66d2a
--- /dev/null
+++ b/adapters/tests/core/test_decision.py
@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import dataclasses
+import json
+import unittest
+
+from acs.core import Decision, normalize_decision, schemas_dir
+
+
+class NormalizeDecision(unittest.TestCase):
+    def test_the_known_dispositions_are_the_schema_enum(self) -> None:
+        schema = json.loads((schemas_dir() / "response-envelope.json").read_text())
+        enum = schema["$defs"]["AcsResult"]["properties"]["decision"]["enum"]
+        for disposition in enum:
+            with self.subTest(disposition):
+                self.assertTrue(normalize_decision({"decision": disposition}).known)
+        self.assertFalse(normalize_decision({"decision": "maybe"}).known)
+
+    def test_a_disposition_is_stripped_and_lower_cased(self) -> None:
+        decision = normalize_decision({"decision": " DENY ", "reasoning": "no"})
+        self.assertEqual(decision, Decision("deny", "no", {}))
+        self.assertTrue(decision.known)
+
+    def test_an_unusable_disposition_is_empty_and_unknown(self) -> None:
+        for raw in (None, 1, ["allow"], {"value": "allow"}):
+            with self.subTest(raw=raw):
+                decision = normalize_decision({"decision": raw})
+                self.assertEqual(decision.disposition, "")
+                self.assertFalse(decision.known)
+
+    def test_a_result_that_is_not_an_object_is_an_empty_decision(self) -> None:
+        for result in (None, "allow", ["allow"], 7):
+            with self.subTest(result=result):
+                self.assertEqual(normalize_decision(result), Decision("", "", {}))
+
+    def test_reasoning_is_always_a_string(self) -> None:
+        self.assertEqual(normalize_decision({"decision": "deny"}).reasoning, "")
+        self.assertEqual(normalize_decision({"decision": "deny", "reasoning": {"rule": 4}}).reasoning,
+                         '{"rule": 4}')
+
+    def test_modifications_are_always_an_object(self) -> None:
+        mods = {"parameter_overrides": {"path": "/tmp"}}
+        self.assertEqual(normalize_decision({"decision": "modify", "modifications": mods}).modifications, mods)
+        for raw in (None, [], "redact", 3):
+            with self.subTest(raw=raw):
+                self.assertEqual(normalize_decision({"decision": "modify", "modifications": raw}).modifications, {})
+
+    def test_a_decision_is_immutable_and_owns_its_default(self) -> None:
+        decision = normalize_decision(None)
+        with self.assertRaises(dataclasses.FrozenInstanceError):
+            decision.disposition = "allow"  # type: ignore[misc]
+        self.assertIsNot(Decision().modifications, Decision().modifications)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/adapters/tests/core/test_envelope.py b/adapters/tests/core/test_envelope.py
new file mode 100644
index 00000000..42faa764
--- /dev/null
+++ b/adapters/tests/core/test_envelope.py
@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import unittest
+
+from acs.core import request_envelope, turn_start_envelope
+from acs.testing.harness import validate_hook_payload, validate_request_envelope
+
+METADATA = {"agent_id": "agent", "session_id": "11111111-1111-4111-8111-111111111111", "platform": "test"}
+
+
+class RequestEnvelopes(unittest.TestCase):
+    def test_a_step_envelope_validates(self) -> None:
+        envelope = request_envelope("steps/sessionStart", request_id="22222222-2222-4222-8222-222222222222",
+                                    metadata=METADATA, payload={})
+        self.assertEqual(validate_request_envelope(envelope), [])
+        self.assertNotIn("signature", envelope["params"], "the client signs as the last step before sending")
+
+    def test_turn_start_carries_its_turn_id_three_ways(self) -> None:
+        turn_id = "33333333-3333-4333-8333-333333333333"
+        envelope = turn_start_envelope(METADATA, turn_id)
+        self.assertEqual(validate_request_envelope(envelope), [])
+        self.assertEqual(validate_hook_payload(envelope["params"]["payload"], "turn-start.json"), [])
+        self.assertEqual((envelope["params"]["request_id"], envelope["params"]["metadata"]["turn_id"],
+                          envelope["params"]["payload"]["turn_id"]), (turn_id, turn_id, turn_id))
+
+    def test_envelopes_copy_their_inputs(self) -> None:
+        payload = {"k": "v"}
+        envelope = request_envelope("steps/sessionStart", request_id="r", metadata=METADATA, payload=payload)
+        envelope["params"]["payload"]["k"] = "changed"
+        self.assertEqual(payload, {"k": "v"})
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/adapters/tests/integrations/nat/test_dispositions_live.py b/adapters/tests/integrations/nat/test_dispositions_live.py
index 6bf0de44..3dccc560 100644
--- a/adapters/tests/integrations/nat/test_dispositions_live.py
+++ b/adapters/tests/integrations/nat/test_dispositions_live.py
@@ -275,6 +275,19 @@ class ShellInput(BaseModel):
             "the secret/sensitive value flows through despite Guardian "
             "demanding redaction")
 
+    def test_post_invoke_unusable_disposition_redacts_output_under_fail_open(self) -> None:
+        """An arrived decision outside the five dispositions fails closed, whatever the posture."""
+        class ShellInput(BaseModel):
+            command: str
+        for decision in ("maybe", None):
+            with self.subTest(decision=decision):
+                ctx = _make_ctx_with_pydantic_input(
+                    "Bash", ShellInput, ShellInput(command="cat /etc/secret"))
+                ctx.output = "AWS_SECRET_ACCESS_KEY=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+                self._set_response("steps/toolCallResult", {"decision": decision, "reasoning": "?"})
+                asyncio.run(self._mw(default_deny=False).post_invoke(ctx))
+                self.assertIsNone(ctx.output, "an unusable disposition let the output through")
+
     def test_post_invoke_structured_modify_never_leaks_unmodified_output(self) -> None:
         """A structured result redaction NAT cannot realize fails closed."""
         class ShellInput(BaseModel):