item 6c: the gateway in the execution path

.github/workflows/ci.yml

@@ -22,7 +22,9 @@ jobs:
       - name: Install
         run: |
           python -m pip install --upgrade pip
-          pip install -e ".[dev]"
+          # The gateway's tests need its extra; T30 still proves `import ctrlrun` does not
+          # touch it, in a subprocess, whether or not it happens to be installed here.
+          pip install -e ".[dev,gateway]"
 
       - name: pytest
         run: pytest
@@ -62,7 +64,7 @@ jobs:
         run: |
           mkdir -p /tmp/sdist && tar xzf dist/*.tar.gz -C /tmp/sdist --strip-components=1
           cd /tmp/sdist
-          pip install ".[dev]"
+          pip install ".[dev,gateway]"
           pytest -q
 
       - name: The quick start works from the wheel

pyproject.toml

@@ -86,6 +86,12 @@ select = ["E", "F", "W", "I", "N", "UP", "B", "ANN", "SIM", "RUF"]
 # SPEC-v0.1 §5.3 freezes `commit_effect(effect_key, action_id, result: Any)`. An executor's
 # return value has no narrower type, and the store only serializes it.
 "src/ctrlrun/state.py" = ["ANN401"]
+# The gateway's subject matter is untyped JSON off a socket: a JSON-RPC `params`, a tool's
+# result, an HTTP client's response object behind a lazy import. Narrowing those to a lie
+# would be worse than saying `Any` and validating at the boundary, which is what mcp.py
+# does. N818: the error names follow this codebase's convention (DuplicateEffect,
+# AmbiguousEffect), not the suffix rule.
+"src/ctrlrun/gateway/*.py" = ["ANN401", "N818"]
 # Type hints are a src/ constraint; tests stay cheap to write. N802 lets acceptance
 # tests carry the SPEC §7 identifier verbatim (test_T7_...).
 "tests/*" = ["ANN", "N802"]

src/ctrlrun/cli/main.py

@@ -442,6 +442,86 @@ def _approval_dict(record: ApprovalRecord) -> dict[str, Any]:
     }
 
 
+@main.command()
+@click.option("--upstream", required=True, help="The MCP server this gateway fronts.")
+@click.option("--alias", required=True, help="Names the upstream in 'mcp.<alias>.<tool>'.")
+@click.option("--listen", default="127.0.0.1:8900", show_default=True, help="HOST:PORT.")
+@click.option("--path", default="/mcp", show_default=True, help="The MCP endpoint path.")
+@click.option("--principal", default=None, help="A fixed agent name, for one tenant.")
+@click.option("--principal-header", default=None, help="Take the agent from this header.")
+@click.option(
+    "--principal-from-client-info",
+    is_flag=True,
+    help="Take it from clientInfo, which the protocol does not verify. Removed in v0.3.",
+)
+@click.option("--user-header", default=None, help="Take principal.user from this header.")
+@click.option("--environment", default="production", show_default=True)
+@click.option("--upstream-timeout", type=float, default=30.0, show_default=True)
+@click.option("--max-body-bytes", type=int, default=1024 * 1024, show_default=True)
+@click.option("--allow-origin", "allow_origins", multiple=True, help="Repeatable.")
+@click.option("--allow-remote", is_flag=True, help="Permit a non-loopback --listen.")
+def gateway(
+    upstream: str,
+    alias: str,
+    listen: str,
+    path: str,
+    principal: str | None,
+    principal_header: str | None,
+    principal_from_client_info: bool,
+    user_header: str | None,
+    environment: str,
+    upstream_timeout: float,
+    max_body_bytes: int,
+    allow_origins: tuple[str, ...],
+    allow_remote: bool,
+) -> None:
+    """Front an MCP server, applying this directory's policy to every tools/call."""
+    host, _, port = listen.rpartition(":")
+    try:
+        from ..gateway import serve
+
+        _announce_actions_without_an_effect()
+        serve(
+            upstream=upstream,
+            alias=alias,
+            host=host or "127.0.0.1",
+            port=int(port),
+            path=path,
+            principal=principal,
+            principal_header=principal_header,
+            principal_from_client_info=principal_from_client_info,
+            user_header=user_header,
+            environment=environment,
+            upstream_timeout=upstream_timeout,
+            max_body_bytes=max_body_bytes,
+            allow_origins=tuple(allow_origins),
+            allow_remote=allow_remote,
+        )
+    except CTRLRunError as exc:
+        raise _fail(exc) from exc
+    except KeyboardInterrupt:  # pragma: no cover - an operator pressing ctrl-c
+        click.echo("")
+
+
+def _announce_actions_without_an_effect() -> None:
+    """SPEC-v0.2 §3.2 — print the actions with no `effect:` at startup.
+
+    A write with no effect key is exactly the configuration this product exists to prevent,
+    and it should be visible on the line that starts the process rather than discovered in a
+    receipt three weeks later.
+    """
+    from ..policy import Policy
+
+    policy = Policy.from_file()
+    keyless = sorted(name for name in policy.actions if policy.effect_template(name) is None)
+    if not keyless:
+        return
+    click.echo(f"{len(keyless)} action(s) have no effect: template and get no reservation:")
+    for name in keyless:
+        click.echo(f"  {name}")
+    click.echo("That is right for a read, and wrong for anything that changes the world.")
+
+
 def _receipt_line(receipt: Receipt) -> str:
     return (
         f"{iso_timestamp(receipt.finished_at)}  {receipt.receipt_id}  {receipt.action}  "

src/ctrlrun/gateway/__init__.py

@@ -14,7 +14,7 @@
 
 import importlib
 from types import ModuleType
-from typing import Final, NoReturn
+from typing import Any, Final
 
 from ..errors import MissingDependency
 
@@ -36,14 +36,24 @@ def http_client() -> ModuleType:
         raise MissingDependency(_HTTP_CLIENT, _EXTRA) from exc
 
 
-def serve(*, upstream: str, alias: str, **options: object) -> NoReturn:
+def serve(*, upstream: str, alias: str, **options: Any) -> None:
     """Run a gateway in front of one upstream MCP server (SPEC-v0.2 §6.1).
 
     The dependency check happens first and unconditionally, so an operator who has not
     installed the extra learns it from one clear line rather than from a stack trace.
+
+    Blocks until interrupted. `Control.from_file()` finds the policy and the store, so a
+    gateway and a decorator-based worker sharing a policy share reservations (§6.1).
     """
     http_client()
-    raise NotImplementedError(
-        "the gateway's request handling arrives with build-list items 6b-6d; "
-        "this release ships the StateStore half it stands on (SPEC-v0.2 §6.9.4, §6.10)"
-    )
+    from ..control import Control
+    from .server import Gateway, GatewayConfig, httpx_forwarder, serve_forever
+
+    config = GatewayConfig(upstream=upstream, alias=alias, **options)
+    control = Control.from_file()
+    forwarder = httpx_forwarder(config)
+    try:
+        serve_forever(Gateway(config, control, forwarder))
+    finally:
+        forwarder.close()
+        control.store.close()