Share the round trip, sign once, and return a Decision

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.

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",

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)

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."""

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)

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",
 }
 

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,25 +46,18 @@ 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", "")
     reasoning = (raw_reason if isinstance(raw_reason, str)
                  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:

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.