diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/.gitignore b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/.gitignore
new file mode 100644
index 000000000..1299eac71
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/.gitignore
@@ -0,0 +1,16 @@
+# Python
+__pycache__/
+*.py[cod]
+.venv/
+venv/
+
+# Tooling caches
+.ruff_cache/
+.pytest_cache/
+.mypy_cache/
+
+# Generated at runtime by deploy.py / consumed by invoke.py + cleanup.py
+runtime_config.json
+
+# OS
+.DS_Store
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/README.md b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/README.md
new file mode 100644
index 000000000..1d541598f
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/README.md
@@ -0,0 +1,276 @@
+# Egress-Controlled Code Execution
+
+## Overview
+
+This sample demonstrates how to run **untrusted code** securely inside an Amazon Bedrock
+AgentCore Runtime microVM. Three independently deployable containers collaborate so that
+customer code never touches the network, credentials, or the container control plane
+directly:
+
+- **Supervisor** — trusted orchestrator. Runs as a `BedrockAgentCoreApp`, receives
+  customer commands via the AgentCore Runtime API, and drives containerd (via `ctr`) to
+  pull and manage the broker and agent containers.
+- **Broker** — trusted egress proxy. All external access from the agent flows through it,
+  and every request is validated against a runtime-configurable allowlist.
+- **Agent** — the untrusted sandbox. No network, no credentials, no containerd socket —
+  it can only reach the broker over a Unix-domain IPC socket.
+
+The working Phase 1 operation is `ping_domain`, which exercises the full
+supervisor → agent → broker → egress path with policy enforcement. The allow-vs-deny
+contrast (an allowed domain pings, a blocked domain is denied by the broker) is the
+security boundary this sample demonstrates.
+
+### Use case details
+
+| Information         | Details                                                                 |
+|---------------------|-------------------------------------------------------------------------|
+| Use case type       | Security / sandboxed execution                                          |
+| Agent type          | Multi-container (supervisor / broker / agent)                           |
+| Use case components | AgentCore Runtime, containerd (`ctr`), Unix-socket IPC, egress allowlist |
+| Use case vertical   | Platform / infrastructure security                                      |
+| Example complexity  | Advanced                                                                |
+| SDK used            | Amazon Bedrock AgentCore SDK (`bedrock-agentcore`), boto3               |
+
+## How this sample is different
+
+Unlike the other runtime samples in this folder — which ship a single agent as code
+(zipped to S3 via `codeConfiguration`) — this sample is **container-based**. You build and
+push **three** `linux/arm64` images to ECR, and the runtime is created from the
+`supervisor` image via `containerConfiguration`. There is no `entryPoint`/zip step; each
+image runs its own Dockerfile `CMD`.
+
+It also relies on one platform assumption that the other samples do not: **the supervisor
+drives `containerd` inside the AgentCore Runtime microVM**. When the supervisor receives
+`start_broker` / `start_agent`, it shells out to `ctr` (against
+`/run/containerd/containerd.sock`) to pull each image from ECR and launch the broker and
+agent as sibling containers *inside the same microVM*. The broker is launched with
+`--net-host` (it can reach the network); the agent is launched **without** it (isolated
+network namespace, loopback only) — that asymmetry is the security boundary. If the
+runtime environment does not expose a usable containerd socket to the supervisor, the
+`start_*` commands will fail; this is expected and central to how the sample works.
+
+## Key features
+
+- **Network-isolated agent** — the untrusted container runs with its own network
+  namespace (loopback only); it has no route to the internet or to IMDS.
+- **Broker-mediated egress** — the agent can only act on the outside world by asking the
+  broker, which enforces an allowlist per operation.
+- **Runtime-configurable policy** — allowlists are updated live via the `configure_broker`
+  command, no container restart required.
+- **Length-prefixed JSON IPC** — a small shared protocol (`shared/ipc.py`) over Unix
+  domain sockets connects the components.
+
+## Architecture
+
+![Architecture: the supervisor launches the broker and agent inside the microVM; only the broker reaches the internet, the agent is network-isolated](images/architecture.png)
+
+The supervisor is the only container the runtime starts; it launches the broker and agent
+itself via containerd. Only the broker reaches the internet — the agent is network-isolated
+and must ask the broker over `broker.sock`. The supervisor commands, IPC protocol, and full
+security model are detailed in the sections below.
+
+## Project Structure
+
+```
+12-egress-coding-execution/
+├── deploy.py            # Check images in ECR, create the runtime, wait until READY
+├── invoke.py            # Run the demo: start broker/agent, configure, ping allow + deny
+├── cleanup.py           # Stop containers, delete the runtime (--delete-ecr also drops repos)
+├── requirements.txt     # boto3 (for the deploy/invoke/cleanup scripts)
+├── requirements-dev.txt # + pytest (for running the unit tests)
+├── shared/
+│   └── ipc.py                 # Length-prefixed JSON IPC framing (send/recv), shared by all three
+├── supervisor/                # TRUSTED orchestrator image (the one the runtime starts)
+│   ├── Dockerfile             # python:3.12-slim + containerd client (ctr)
+│   ├── requirements.txt       # image deps: bedrock-agentcore, boto3
+│   └── src/
+│       ├── main.py            # BedrockAgentCoreApp entrypoint + command router
+│       ├── containerd.py      # ctr wrappers: pull / start / stop / status
+│       ├── ecr.py             # fetch an ECR auth token (boto3) for image pulls
+│       ├── profiles.py        # build the ctr run args (broker = --net-host, agent = isolated)
+│       └── task_dispatch.py   # supervisor→agent IPC over supervisor.sock (TaskDispatcher)
+├── broker/                    # TRUSTED egress proxy image
+│   ├── Dockerfile             # python:3.12-slim + iputils-ping (stdlib-only otherwise)
+│   └── src/
+│       ├── main.py            # start the asyncio Unix-socket IPC server
+│       ├── ipc_server.py      # accept agent connections, route methods to handlers
+│       ├── config.py          # BrokerConfig: mutable allowlist state + configure handler
+│       └── handlers/
+│           ├── ping.py        # ping proxy, enforces ALLOWED_PING_DOMAINS
+│           └── http.py        # HTTP proxy (Phase 2 stub → NOT_IMPLEMENTED)
+├── agent/                     # UNTRUSTED sandbox image (no network, no creds, non-root)
+│   ├── Dockerfile             # python:3.12-slim, runs as USER sandbox (UID 1000)
+│   └── src/
+│       ├── agent_main.py      # task loop: wait on supervisor.sock, proxy pings via broker.sock
+│       └── ipc_client.py      # Unix-socket client (connect / request / wait_for_message)
+├── images/                    # architecture diagram (architecture.png)
+└── tests/                     # Unit tests for the IPC framing and the broker allowlist
+```
+
+## Prerequisites
+
+| Requirement | Description |
+|-------------|-------------|
+| Python 3.12+ | To run the `deploy.py` / `invoke.py` / `cleanup.py` scripts |
+| Docker with buildx | To build the three `linux/arm64` container images |
+| AWS account + credentials | For ECR image pulls and Bedrock AgentCore Runtime invocations |
+| An Amazon ECR repository | Three repos: `egress-coding-execution/{supervisor,broker,agent}` |
+| An IAM execution role | `deploy.py` creates one (ECR pull + CloudWatch Logs) if you don't set `ROLE_ARN` |
+
+Install the script dependencies:
+
+```bash
+cd amazon-bedrock-agentcore-samples/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution
+pip install -r requirements.txt
+```
+
+## Building and pushing the images
+
+All images are built for `linux/arm64` (AgentCore Runtime is ARM64) and pushed to ECR.
+Run from the sample root, and replace `<account-id>` and the region with your own. On a
+non-arm64 host, enable emulation first with
+`docker run --privileged --rm tonistiigi/binfmt --install arm64`.
+
+```bash
+REGION=us-east-1
+REGISTRY=<account-id>.dkr.ecr.${REGION}.amazonaws.com/egress-coding-execution
+
+# Create the three repositories (once)
+for c in supervisor broker agent; do
+  aws ecr create-repository --repository-name egress-coding-execution/$c --region ${REGION}
+done
+
+# Authenticate with ECR
+aws ecr get-login-password --region ${REGION} \
+  | docker login --username AWS --password-stdin <account-id>.dkr.ecr.${REGION}.amazonaws.com
+
+# Build & push each image
+docker buildx build --platform linux/arm64 -f supervisor/Dockerfile -t ${REGISTRY}/supervisor:latest --push .
+docker buildx build --platform linux/arm64 -f broker/Dockerfile     -t ${REGISTRY}/broker:latest --push .
+docker buildx build --platform linux/arm64 -f agent/Dockerfile      -t ${REGISTRY}/agent:latest --push .
+```
+
+## Deployment
+
+The three scripts share a `runtime_config.json` written by `deploy.py`. Set your AWS
+region and credentials in the environment first (the scripts read the default boto3
+session). Edit the parameters at the top of `deploy.py` (`ROLE_NAME`, `ECR_REPO`,
+`RUNTIME_NAME`) if you used different names.
+
+```bash
+# 1. Create the AgentCore Runtime from the supervisor image (waits until READY)
+python deploy.py
+
+# 2. Run the end-to-end demo (start broker + agent, configure the allowlist,
+#    then ping an allowed domain and a blocked one)
+python invoke.py
+
+# 3. Tear down the runtime. Add --delete-ecr to also delete the three ECR repos,
+#    and --delete-role to delete the IAM role (only if deploy.py created it).
+python cleanup.py
+```
+
+`deploy.py` verifies the three images exist in ECR, creates (or reuses) the runtime, and
+waits for status `READY`. `invoke.py` drives one session through the full flow:
+
+```
+start_broker → start_agent → configure_broker (allowlist: amazon.com) → status
+  → ping_domain amazon.com      (ALLOW: broker permits it, ping executes)
+  → ping_domain aws.amazon.com  (DENY:  blocked — subdomain not in the allowlist)
+```
+
+Both domains are Amazon's. The allowlist entry `amazon.com` is an **exact match**, so the
+subdomain `aws.amazon.com` is denied — demonstrating that the policy is a strict allowlist,
+not a substring/suffix match (use a glob like `*.amazon.com` to allow subdomains).
+
+The **ALLOW** result confirms the broker let the request through and the ping executed —
+we do not assert ICMP reachability, because inside the microVM the echo replies may not
+complete (a non-zero exit code there is a network-path artifact, not a policy denial). The
+**DENY** result is the broker rejecting the domain against its allowlist:
+
+```json
+{"status":"ok","result":{"exit_code":-1,"stdout":"","stderr":"Domain not in ping allowlist: aws.amazon.com","domain":"aws.amazon.com","count":3,"timeout":5}}
+```
+
+## Running tests
+
+The unit tests cover the two security-critical, pure-Python pieces without any AWS calls:
+the length-prefixed IPC framing (`shared/ipc.py`) and the broker's allowlist policy
+(`broker/src/config.py`, including glob matching and default-deny).
+
+```bash
+pip install -r requirements-dev.txt
+pytest
+```
+
+## Supervisor commands
+
+The supervisor is a stateless command router (`BedrockAgentCoreApp` entrypoint). Every
+command is a JSON payload `{"command": ..., "params": {...}}` sent via
+`invoke_agent_runtime` on one session:
+
+| Command | Parameters | Action |
+|---------|-----------|--------|
+| `start_broker` | `image_uri`, `env` | Pull the broker image from ECR and start it with the trusted profile (`--net-host`). `env` is passed as container env vars. |
+| `start_agent` | `image_uri` | Pull the agent image and start it with the untrusted profile (no `--net-host`). |
+| `configure_broker` | `allowed_ping_domains`, `allowed_domains` | Update the broker allowlists at runtime via IPC — no restart. |
+| `ping_domain` | `domain`, `count`, `timeout` | Dispatch a ping task to the agent (proxied through the broker); returns the result synchronously. |
+| `status` | — | Report the state of the broker and agent containers. |
+| `list_containers` | — | Return the raw containerd view (`ctr tasks ls` / `containers ls`) for debugging. |
+| `stop_agent` / `stop_broker` | — | Kill and remove the container. |
+
+`ping_domain` exercises the full path: the supervisor dispatches the task to the agent
+over `supervisor.sock`; the agent asks the broker to `ping` over `broker.sock`; the broker
+validates the domain against `ALLOWED_PING_DOMAINS` (fnmatch globs like `*.amazon.com`) and
+only then runs `ping`. A denied domain never reaches `ping`.
+
+## IPC protocol
+
+The three containers talk over two Unix-domain sockets on a shared bind mount
+(`/run/containerd/sandbox-ipc` on the microVM → `/tmp/ipc` in each container):
+
+| Socket | Listener | Connector | Purpose |
+|--------|----------|-----------|---------|
+| `broker.sock` | Broker | Agent | Egress proxy (ping; HTTP/secrets are Phase 2) |
+| `supervisor.sock` | Supervisor | Agent | Task dispatch (`ping_domain`) |
+
+Framing is a 4-byte big-endian length prefix followed by a UTF-8 JSON payload
+(`shared/ipc.py`). Requests carry `{id, method, params}`; responses carry
+`{id, status, result}` or `{id, status:"error", error:{code, message}}`.
+
+## Security model
+
+| Component | Trust | Network | Credentials | containerd socket |
+|-----------|-------|---------|-------------|-------------------|
+| Supervisor | Trusted | Host (`--net-host`) | Execution role (IMDS) | Yes |
+| Broker | Trusted | Host (`--net-host`) | Optional, scoped | No |
+| Agent | **Untrusted** | **None (loopback only)** | **None** | **No** |
+
+| Attack vector | Mitigation |
+|---------------|------------|
+| Agent reaches the network / IMDS | Isolated network namespace (no `--net-host`) → no route to `169.254.169.254` |
+| Agent drives containerd | Socket is never mounted into the agent |
+| Agent reads broker/supervisor state | Separate mount and PID namespaces |
+| Agent escalates privileges | Non-root (UID 1000), no added capabilities |
+| Agent reaches unauthorized domains | Broker validates every request against the allowlist |
+
+**Phase 1 hardening relies on** omitting `--net-host` for network isolation and the
+Dockerfile `USER sandbox` (UID 1000) for non-root execution. Explicit namespace flags,
+capability dropping, and cgroup memory/CPU limits are **deferred to Phase 2** and should be
+added before any non-experimental use.
+
+## Current status
+
+Only **Phase 1** is implemented: the supervisor/broker/agent architecture with
+`ping_domain` as the working egress-controlled operation. HTTP proxying, secret
+retrieval, rate limiting, and explicit cgroup/namespace hardening are stubbed or deferred
+to Phase 2.
+
+## Disclaimer
+
+The examples provided in this repository are for **experimental and educational purposes
+only**. They demonstrate concepts and techniques but are not intended for direct use in
+production environments. Before adapting this pattern, complete the Phase 2 hardening
+described in the [Security model](#security-model) above, and use
+[Amazon Bedrock Guardrails](https://docs.aws.amazon.com/bedrock/latest/userguide/prompt-injection.html)
+to protect against prompt injection.
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/agent/src/__init__.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/agent/src/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/agent/src/agent_main.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/agent/src/agent_main.py
new file mode 100644
index 000000000..802eec6d6
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/agent/src/agent_main.py
@@ -0,0 +1,106 @@
+import logging
+import os
+import sys
+import time
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+from .ipc_client import IPCClient
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [agent] %(levelname)s: %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+SUPERVISOR_SOCKET = os.environ.get("SUPERVISOR_SOCKET_PATH", "/tmp/ipc/supervisor.sock")
+BROKER_SOCKET = os.environ.get("BROKER_SOCKET_PATH", "/tmp/ipc/broker.sock")
+
+
+def _handle_ping_domain(params: dict) -> dict:
+    domain = params.get("domain")
+    count = params.get("count", 3)
+    timeout = params.get("timeout", 5)
+
+    if not domain:
+        return {
+            "exit_code": -1,
+            "stdout": "",
+            "stderr": "Missing 'domain' parameter",
+            "domain": "",
+            "count": count,
+            "timeout": timeout,
+        }
+
+    client = IPCClient(BROKER_SOCKET)
+    try:
+        client.connect()
+        response = client.request("ping", {"domain": domain, "count": count, "timeout": timeout})
+    finally:
+        client.close()
+
+    if response and response.get("status") == "ok":
+        result: dict = response.get("result", {})
+        return result
+    else:
+        error = response.get("error", {}) if response else {}
+        return {
+            "exit_code": -1,
+            "stdout": "",
+            "stderr": error.get("message", "Ping request failed"),
+            "domain": domain,
+            "count": count,
+            "timeout": timeout,
+        }
+
+
+def main():
+    logger.info("Agent starting, connecting to supervisor at %s", SUPERVISOR_SOCKET)
+
+    supervisor = IPCClient(SUPERVISOR_SOCKET)
+
+    retries = 0
+    while retries < 30:
+        try:
+            supervisor.connect()
+            logger.info("Connected to supervisor")
+            break
+        except (ConnectionRefusedError, FileNotFoundError):
+            retries += 1
+            time.sleep(1)
+    else:
+        logger.error("Failed to connect to supervisor after 30 retries")
+        sys.exit(1)
+
+    try:
+        while True:
+            logger.info("Waiting for task...")
+            msg = supervisor.wait_for_message()
+            if msg is None:
+                logger.info("Supervisor disconnected, shutting down")
+                break
+
+            request_id = msg.get("id", "unknown")
+            method = msg.get("method", "")
+            params = msg.get("params", {})
+
+            logger.info("Received task %s: method=%s", request_id, method)
+
+            if method == "ping_domain":
+                result = _handle_ping_domain(params)
+                supervisor.send_response(request_id, "ok", result=result)
+            else:
+                supervisor.send_response(
+                    request_id,
+                    "error",
+                    error={
+                        "code": "UNKNOWN_METHOD",
+                        "message": f"Unknown method: {method}",
+                    },
+                )
+    finally:
+        supervisor.close()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/agent/src/ipc_client.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/agent/src/ipc_client.py
new file mode 100644
index 000000000..0648eb7e4
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/agent/src/ipc_client.py
@@ -0,0 +1,58 @@
+import os
+import socket
+import sys
+import uuid
+from typing import Optional
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+from shared.ipc import recv_message, send_message
+
+
+class IPCClient:
+    def __init__(self, socket_path: str):
+        self._socket_path = socket_path
+        self._sock: Optional[socket.socket] = None
+
+    def connect(self) -> None:
+        self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        self._sock.connect(self._socket_path)
+
+    def close(self) -> None:
+        if self._sock:
+            self._sock.close()
+            self._sock = None
+
+    def _require_sock(self) -> socket.socket:
+        if self._sock is None:
+            raise RuntimeError("IPC client is not connected")
+        return self._sock
+
+    def request(self, method: str, params: dict) -> Optional[dict]:
+        sock = self._require_sock()
+        msg = {
+            "id": str(uuid.uuid4()),
+            "method": method,
+            "params": params,
+        }
+        send_message(sock, msg)
+        return recv_message(sock)
+
+    def send_response(
+        self,
+        request_id: str,
+        status: str,
+        result: Optional[dict] = None,
+        error: Optional[dict] = None,
+    ) -> None:
+        sock = self._require_sock()
+        msg: dict = {"id": request_id, "status": status}
+        if result is not None:
+            msg["result"] = result
+        if error is not None:
+            msg["error"] = error
+        send_message(sock, msg)
+
+    def wait_for_message(self) -> Optional[dict]:
+        sock = self._require_sock()
+        return recv_message(sock)
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/__init__.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/config.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/config.py
new file mode 100644
index 000000000..decc883d2
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/config.py
@@ -0,0 +1,44 @@
+import fnmatch
+import os
+from typing import Optional
+
+
+class BrokerConfig:
+    def __init__(self):
+        self._allowed_ping_domains = self._parse_env("ALLOWED_PING_DOMAINS", "")
+        self._allowed_domains = self._parse_env("ALLOWED_DOMAINS", "")
+
+    def _parse_env(self, key: str, default: str) -> list:
+        raw = os.environ.get(key, default)
+        return [p.strip() for p in raw.split(",") if p.strip()]
+
+    @property
+    def allowed_ping_domains(self) -> list:
+        return list(self._allowed_ping_domains)
+
+    @property
+    def allowed_domains(self) -> list:
+        return list(self._allowed_domains)
+
+    def update(
+        self,
+        allowed_ping_domains: Optional[list] = None,
+        allowed_domains: Optional[list] = None,
+    ) -> list:
+        updated = []
+        if allowed_ping_domains is not None:
+            self._allowed_ping_domains = list(allowed_ping_domains)
+            updated.append("allowed_ping_domains")
+        if allowed_domains is not None:
+            self._allowed_domains = list(allowed_domains)
+            updated.append("allowed_domains")
+        return updated
+
+    def is_ping_domain_allowed(self, domain: str) -> bool:
+        for pattern in self.allowed_ping_domains:
+            if fnmatch.fnmatch(domain, pattern):
+                return True
+        return False
+
+
+broker_config = BrokerConfig()
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/handlers/__init__.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/handlers/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/handlers/http.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/handlers/http.py
new file mode 100644
index 000000000..2f6690291
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/handlers/http.py
@@ -0,0 +1,13 @@
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+async def handle_http_request(params: dict) -> dict:
+    return {
+        "status": "error",
+        "error": {
+            "code": "NOT_IMPLEMENTED",
+            "message": "HTTP proxy not implemented in Phase 1",
+        },
+    }
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/handlers/ping.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/handlers/ping.py
new file mode 100644
index 000000000..11bb1e484
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/handlers/ping.py
@@ -0,0 +1,60 @@
+import asyncio
+import logging
+
+from ..config import broker_config
+
+logger = logging.getLogger(__name__)
+
+
+async def handle_ping(params: dict) -> dict:
+    domain = params.get("domain")
+    count = params.get("count", 3)
+    timeout = params.get("timeout", 5)
+
+    if not domain:
+        return {
+            "status": "error",
+            "error": {
+                "code": "INVALID_PARAMS",
+                "message": "Missing 'domain' parameter",
+            },
+        }
+
+    if not broker_config.is_ping_domain_allowed(domain):
+        return {
+            "status": "error",
+            "error": {
+                "code": "POLICY_DENIED",
+                "message": f"Domain not in ping allowlist: {domain}",
+            },
+        }
+
+    cmd = ["ping", "-c", str(int(count)), "-t", str(int(timeout)), domain]
+    logger.info("Executing: %s", " ".join(cmd))
+
+    try:
+        proc = await asyncio.create_subprocess_exec(
+            *cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=int(timeout) + 10)
+
+        return {
+            "status": "ok",
+            "result": {
+                "exit_code": proc.returncode,
+                "stdout": stdout.decode("utf-8", errors="replace"),
+                "stderr": stderr.decode("utf-8", errors="replace"),
+                "domain": domain,
+                "count": count,
+                "timeout": timeout,
+            },
+        }
+    except asyncio.TimeoutError:
+        return {
+            "status": "error",
+            "error": {"code": "PING_TIMEOUT", "message": f"Ping to {domain} timed out"},
+        }
+    except Exception as e:
+        return {"status": "error", "error": {"code": "PING_FAILED", "message": str(e)}}
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/ipc_server.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/ipc_server.py
new file mode 100644
index 000000000..4fcdb70f0
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/ipc_server.py
@@ -0,0 +1,109 @@
+import asyncio
+import json
+import logging
+import os
+import struct
+from pathlib import Path
+
+from .config import broker_config
+from .handlers.http import handle_http_request
+from .handlers.ping import handle_ping
+
+logger = logging.getLogger(__name__)
+
+HEADER_FORMAT = ">I"
+HEADER_SIZE = 4
+
+# Method -> async handler. `heartbeat` and `configure` are handled inline in
+# _handle_connection (they need direct access to the connection/config), so they
+# are not registered here.
+HANDLERS = {
+    "ping": handle_ping,
+    "http_request": handle_http_request,
+}
+
+
+async def _recv_message(reader: asyncio.StreamReader) -> dict | None:
+    try:
+        header = await reader.readexactly(HEADER_SIZE)
+    except asyncio.IncompleteReadError:
+        return None
+    length = struct.unpack(HEADER_FORMAT, header)[0]
+    payload = await reader.readexactly(length)
+    message: dict = json.loads(payload.decode("utf-8"))
+    return message
+
+
+async def _send_message(writer: asyncio.StreamWriter, msg: dict) -> None:
+    payload = json.dumps(msg).encode("utf-8")
+    header = struct.pack(HEADER_FORMAT, len(payload))
+    writer.write(header + payload)
+    await writer.drain()
+
+
+async def _handle_connection(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
+    peer = writer.get_extra_info("peername")
+    logger.info("Agent connected: %s", peer)
+
+    try:
+        while True:
+            msg = await _recv_message(reader)
+            if msg is None:
+                logger.info("Agent disconnected")
+                break
+
+            request_id = msg.get("id", "unknown")
+            method = msg.get("method", "")
+            params = msg.get("params", {})
+
+            logger.info("Request %s: method=%s", request_id, method)
+
+            if method == "heartbeat":
+                response = {"id": request_id, "status": "ok", "result": {"alive": True}}
+            elif method == "configure":
+                updated = broker_config.update(
+                    allowed_ping_domains=params.get("allowed_ping_domains"),
+                    allowed_domains=params.get("allowed_domains"),
+                )
+                response = {
+                    "id": request_id,
+                    "status": "ok",
+                    "result": {"updated": updated},
+                }
+            elif method in HANDLERS:
+                handler = HANDLERS[method]
+                result = await handler(params)
+                response = {"id": request_id, **result}
+            else:
+                response = {
+                    "id": request_id,
+                    "status": "error",
+                    "error": {
+                        "code": "UNKNOWN_METHOD",
+                        "message": f"Unknown method: {method}",
+                    },
+                }
+
+            await _send_message(writer, response)
+    except Exception as e:
+        logger.exception("Error handling connection: %s", e)
+    finally:
+        writer.close()
+        await writer.wait_closed()
+
+
+async def start_server(socket_path: str):
+    path = Path(socket_path)
+    if path.exists():
+        path.unlink()
+    path.parent.mkdir(parents=True, exist_ok=True)
+
+    server = await asyncio.start_unix_server(_handle_connection, path=socket_path)
+    # Intentional: the socket is created by the (root) broker, but the untrusted
+    # agent runs as UID 1000 and must be able to connect to it. Cross-UID access
+    # to this IPC socket is core to the sandbox design.
+    os.chmod(socket_path, 0o777)  # nosec B103
+    logger.info("Broker IPC server listening on %s", socket_path)
+
+    async with server:
+        await server.serve_forever()
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/main.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/main.py
new file mode 100644
index 000000000..ec5c40773
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/broker/src/main.py
@@ -0,0 +1,24 @@
+import asyncio
+import logging
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+from .ipc_server import start_server
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+def main():
+    socket_path = os.environ.get("IPC_SOCKET_PATH", "/tmp/ipc/broker.sock")
+    logger.info("Starting broker, socket=%s", socket_path)
+    asyncio.run(start_server(socket_path))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/cleanup.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/cleanup.py
new file mode 100644
index 000000000..1ec449d15
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/cleanup.py
@@ -0,0 +1,98 @@
+"""
+Tear down the Egress-Controlled Code Execution demo.
+
+Reads ``runtime_config.json`` (written by ``deploy.py``), stops the sandbox and
+broker containers (mirroring the supervisor's own shutdown order), then deletes
+the AgentCore Runtime so it stops incurring cost.
+
+Pass ``--delete-ecr`` to also delete the three ECR repositories that hold the
+images, and ``--delete-role`` to delete the IAM execution role — but only if
+``deploy.py`` created it (a role you supplied via ``ROLE_ARN`` is left alone).
+
+Usage:
+    python cleanup.py [--delete-ecr] [--delete-role]
+"""
+
+import json
+import os
+import sys
+import uuid
+
+import boto3
+
+CONFIG_FILE = "runtime_config.json"
+ECR_REPO = "egress-coding-execution"
+
+
+def main():
+    delete_ecr = "--delete-ecr" in sys.argv[1:]
+    delete_role = "--delete-role" in sys.argv[1:]
+
+    try:
+        with open(CONFIG_FILE) as f:
+            config = json.load(f)
+    except FileNotFoundError:
+        print(f"Error: {CONFIG_FILE} not found. Nothing to clean up.")
+        sys.exit(1)
+
+    region = config["region"]
+    runtime_id = config["runtime_id"]
+    runtime_arn = config["runtime_arn"]
+    control = boto3.client("bedrock-agentcore-control", region_name=region)
+    runtime = boto3.client("bedrock-agentcore", region_name=region)
+
+    print(f"Cleaning up runtime {runtime_id} (region {region})\n")
+
+    # Best-effort: stop the containers via the supervisor before deleting the
+    # runtime. A fresh session id is fine — these commands are idempotent.
+    session_id = f"egress-coding-exec-cleanup-{uuid.uuid4().hex}"
+    for command in ("stop_agent", "stop_broker"):
+        try:
+            runtime.invoke_agent_runtime(
+                agentRuntimeArn=runtime_arn,
+                contentType="application/json",
+                accept="application/json",
+                runtimeSessionId=session_id,
+                payload=json.dumps({"command": command, "params": {}}).encode("utf-8"),
+            )
+            print(f"  {command}: requested")
+        except Exception as e:  # noqa: BLE001 - cleanup is best-effort
+            print(f"  {command}: skipped ({e})")
+
+    try:
+        control.delete_agent_runtime(agentRuntimeId=runtime_id)
+        print("✓ Runtime delete requested (transitions to DELETING)")
+    except Exception as e:  # noqa: BLE001
+        print(f"  Warning: delete_agent_runtime failed: {e}")
+
+    if delete_ecr:
+        ecr = boto3.client("ecr", region_name=region)
+        for component in ("supervisor", "broker", "agent"):
+            repo = f"{ECR_REPO}/{component}"
+            try:
+                ecr.delete_repository(repositoryName=repo, force=True)
+                print(f"✓ Deleted ECR repo {repo}")
+            except Exception as e:  # noqa: BLE001
+                print(f"  Warning: could not delete {repo}: {e}")
+
+    if delete_role:
+        role_name = config.get("role_name")
+        if not role_name:
+            print("  Skipping role deletion (deploy.py did not create the role)")
+        else:
+            iam = boto3.client("iam")
+            try:
+                for p in iam.list_role_policies(RoleName=role_name).get("PolicyNames", []):
+                    iam.delete_role_policy(RoleName=role_name, PolicyName=p)
+                iam.delete_role(RoleName=role_name)
+                print(f"✓ Deleted IAM role {role_name}")
+            except Exception as e:  # noqa: BLE001
+                print(f"  Warning: could not delete role {role_name}: {e}")
+
+    if os.path.exists(CONFIG_FILE):
+        os.remove(CONFIG_FILE)
+    print("\n✓ Cleanup complete")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/deploy.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/deploy.py
new file mode 100644
index 000000000..d88e4d484
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/deploy.py
@@ -0,0 +1,223 @@
+"""
+Deploy the Egress-Controlled Code Execution sample to AgentCore Runtime.
+
+Creates (or reuses) an AgentCore Runtime from the pre-built ``supervisor`` image,
+waits until it is READY, and writes ``runtime_config.json`` for ``invoke.py`` /
+``cleanup.py`` to consume.
+
+Prerequisites (this script does NOT build images — that needs Docker + buildx +
+QEMU for linux/arm64; see the README "Building and pushing the images" section):
+  * The three images pushed to ECR: ``supervisor``, ``broker``, ``agent``.
+  * An IAM execution role AgentCore can assume (trust ``bedrock-agentcore.amazonaws.com``)
+    with permissions to pull from ECR and write CloudWatch Logs.
+
+Usage:
+    python deploy.py
+"""
+
+import json
+import os
+import sys
+import time
+
+import boto3
+from boto3.session import Session
+
+# --- Parameters (edit for your own account) --------------------------------
+ROLE_NAME = "EgressCodingExecutionRole"  # IAM execution role name...
+ROLE_ARN = None  # ...or set the full ARN here to override ROLE_NAME
+ECR_REPO = "egress-coding-execution"  # base ECR repo (holds supervisor/broker/agent)
+IMAGE_TAG = "latest"  # image tag to use for all three images
+RUNTIME_NAME = "egress_coding_execution_demo"  # AgentCore Runtime name (created or reused)
+
+CONFIG_FILE = "runtime_config.json"
+
+session = Session()
+REGION = session.region_name or "us-east-1"
+ACCOUNT_ID = session.client("sts").get_caller_identity()["Account"]  # never hardcode
+
+ECR_BASE = f"{ACCOUNT_ID}.dkr.ecr.{REGION}.amazonaws.com/{ECR_REPO}"
+SUPERVISOR_IMAGE = f"{ECR_BASE}/supervisor:{IMAGE_TAG}"
+BROKER_IMAGE = f"{ECR_BASE}/broker:{IMAGE_TAG}"
+AGENT_IMAGE = f"{ECR_BASE}/agent:{IMAGE_TAG}"
+
+control = boto3.client("bedrock-agentcore-control", region_name=REGION)
+
+
+def check_images_exist():
+    """Fail fast with a clear message if any of the three images is missing."""
+    ecr = boto3.client("ecr", region_name=REGION)
+    for component in ("supervisor", "broker", "agent"):
+        repo = f"{ECR_REPO}/{component}"
+        try:
+            ecr.describe_images(repositoryName=repo, imageIds=[{"imageTag": IMAGE_TAG}])
+        except Exception as e:  # noqa: BLE001 - surface any ECR lookup failure clearly
+            print(
+                f"✗ Image {repo}:{IMAGE_TAG} not found in ECR ({e}).\n"
+                "  Build and push the three images first — see the README "
+                '"Building and pushing the images" section.'
+            )
+            sys.exit(1)
+    print("✓ All three images present in ECR")
+
+
+def create_execution_role():
+    """Create (or reuse) the AgentCore execution role, returning its ARN.
+
+    Unlike the LLM-calling samples, this role grants no ``bedrock:InvokeModel`` —
+    the supervisor only needs to pull the three images from ECR and write logs.
+    """
+    iam = boto3.client("iam")
+    trust = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Effect": "Allow",
+                "Principal": {"Service": "bedrock-agentcore.amazonaws.com"},
+                "Action": "sts:AssumeRole",
+                "Condition": {
+                    "StringEquals": {"aws:SourceAccount": ACCOUNT_ID},
+                    "ArnLike": {"aws:SourceArn": f"arn:aws:bedrock-agentcore:{REGION}:{ACCOUNT_ID}:*"},
+                },
+            }
+        ],
+    }
+    policy = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Sid": "ECRPull",
+                "Effect": "Allow",
+                "Action": [
+                    "ecr:GetAuthorizationToken",
+                    "ecr:BatchGetImage",
+                    "ecr:GetDownloadUrlForLayer",
+                    "ecr:BatchCheckLayerAvailability",
+                ],
+                "Resource": "*",
+            },
+            {
+                "Sid": "Logs",
+                "Effect": "Allow",
+                "Action": [
+                    "logs:CreateLogGroup",
+                    "logs:CreateLogStream",
+                    "logs:PutLogEvents",
+                    "logs:DescribeLogStreams",
+                ],
+                "Resource": "*",
+            },
+        ],
+    }
+    created = False
+    try:
+        role_arn = iam.create_role(
+            RoleName=ROLE_NAME,
+            AssumeRolePolicyDocument=json.dumps(trust),
+            Description="Execution role for the Egress-Controlled Code Execution sample",
+        )["Role"]["Arn"]
+        created = True
+        print(f"✓ Created IAM role: {role_arn}")
+    except iam.exceptions.EntityAlreadyExistsException:
+        role_arn = f"arn:aws:iam::{ACCOUNT_ID}:role/{ROLE_NAME}"
+        print(f"✓ Reusing existing IAM role: {role_arn}")
+
+    # Attach the policy first, THEN wait — CreateAgentRuntime validates the role's
+    # ECR permissions, so the policy (not just the role) must have propagated.
+    iam.put_role_policy(
+        RoleName=ROLE_NAME,
+        PolicyName="ece-ecr-logs",
+        PolicyDocument=json.dumps(policy),
+    )
+    if created:
+        print("  waiting for the new role + policy to propagate ...")
+        time.sleep(15)
+    return role_arn
+
+
+def find_runtime(name):
+    """Return (agentRuntimeId, agentRuntimeArn) for an existing runtime, or None."""
+    kwargs = {}
+    while True:
+        resp = control.list_agent_runtimes(**kwargs)
+        for rt in resp.get("agentRuntimes", []):
+            if rt.get("agentRuntimeName") == name:
+                return rt["agentRuntimeId"], rt["agentRuntimeArn"]
+        token = resp.get("nextToken")
+        if not token:
+            return None
+        kwargs = {"nextToken": token}
+
+
+def wait_until_ready(runtime_id, timeout=600, interval=15):
+    """Poll GetAgentRuntime until READY (or raise on a terminal/failed state)."""
+    deadline = time.time() + timeout
+    last_status = None
+    while time.time() < deadline:
+        resp = control.get_agent_runtime(agentRuntimeId=runtime_id)
+        status = resp["status"]
+        if status != last_status:
+            print(f"  status: {status}")
+            last_status = status
+        if status == "READY":
+            return
+        if "FAILED" in status or status in ("DELETING",):
+            reason = resp.get("failureReason", "(no reason provided)")
+            raise RuntimeError(f"Runtime entered {status}: {reason}")
+        time.sleep(interval)
+    raise TimeoutError(f"Runtime not READY within {timeout}s (last: {last_status})")
+
+
+def main():
+    os.chdir(os.path.dirname(os.path.abspath(__file__)))
+
+    print(f"Account   : {ACCOUNT_ID}")
+    print(f"Region    : {REGION}")
+    print(f"Supervisor: {SUPERVISOR_IMAGE}\n")
+
+    check_images_exist()
+
+    # Use an explicitly provided role ARN, otherwise create/reuse ours. We record
+    # whether we own the role so cleanup only deletes a role this script created.
+    role_created_by_us = ROLE_ARN is None
+    role_arn = ROLE_ARN or create_execution_role()
+
+    existing = find_runtime(RUNTIME_NAME)
+    if existing:
+        runtime_id, runtime_arn = existing
+        print(f"✓ Reusing existing runtime '{RUNTIME_NAME}': {runtime_id}")
+    else:
+        print(f"Creating runtime '{RUNTIME_NAME}' ...")
+        created = control.create_agent_runtime(
+            agentRuntimeName=RUNTIME_NAME,
+            agentRuntimeArtifact={"containerConfiguration": {"containerUri": SUPERVISOR_IMAGE}},
+            roleArn=role_arn,
+            networkConfiguration={"networkMode": "PUBLIC"},
+            description="Egress-Controlled Code Execution supervisor (sandboxed untrusted code)",
+        )
+        runtime_id = created["agentRuntimeId"]
+        runtime_arn = created["agentRuntimeArn"]
+        print(f"✓ Created runtime: {runtime_id}")
+
+    print("\nWaiting for runtime to become READY ...")
+    wait_until_ready(runtime_id)
+    print("✓ Runtime is READY")
+
+    with open(CONFIG_FILE, "w") as f:
+        json.dump(
+            {
+                "region": REGION,
+                "runtime_id": runtime_id,
+                "runtime_arn": runtime_arn,
+                "broker_image": BROKER_IMAGE,
+                "agent_image": AGENT_IMAGE,
+                "role_name": ROLE_NAME if role_created_by_us else None,
+            },
+            f,
+            indent=2,
+        )
+    print(f"\n✓ Deployment complete! Wrote {CONFIG_FILE}. Test with: python invoke.py")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/images/architecture.png b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/images/architecture.png
new file mode 100644
index 000000000..99c54af75
Binary files /dev/null and b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/images/architecture.png differ
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/invoke.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/invoke.py
new file mode 100644
index 000000000..cc9a8667b
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/invoke.py
@@ -0,0 +1,122 @@
+"""
+Run the Egress-Controlled Code Execution demo end to end.
+
+Reads ``runtime_config.json`` (written by ``deploy.py``) and drives the full
+supervisor -> agent -> broker -> egress path in one session:
+
+    1. start_broker      (trusted egress proxy)
+    2. start_agent       (untrusted sandbox, no network of its own)
+    3. configure_broker  (set the egress allowlist at runtime)
+    4. status            (both containers running)
+    5. ping_domain ALLOW (amazon.com     -> permitted, ping executes)
+    6. ping_domain DENY  (aws.amazon.com -> blocked by the broker allowlist)
+
+The allow-vs-deny contrast is the security boundary this sample demonstrates. Both
+domains are Amazon's: the allowlist entry ``amazon.com`` is an exact match, so the
+subdomain ``aws.amazon.com`` is denied — showing the policy is a strict allowlist, not a
+substring/suffix match.
+
+Usage:
+    python invoke.py
+"""
+
+import json
+import sys
+import uuid
+
+import boto3
+
+CONFIG_FILE = "runtime_config.json"
+
+# Allowlist applied at runtime via configure_broker (dynamic egress policy).
+# Exact-match only (fnmatch); use a glob like "*.amazon.com" to allow subdomains.
+ALLOWED_PING_DOMAINS = ["amazon.com"]
+
+
+def load_config():
+    try:
+        with open(CONFIG_FILE) as f:
+            return json.load(f)
+    except FileNotFoundError:
+        print(f"Error: {CONFIG_FILE} not found. Run: python deploy.py")
+        sys.exit(1)
+
+
+def make_invoker(runtime, runtime_arn, session_id):
+    """Return an ``invoke(command, params)`` bound to one runtime session."""
+
+    def invoke(command, params=None, show=True):
+        body = {"command": command, "params": params or {}}
+        resp = runtime.invoke_agent_runtime(
+            agentRuntimeArn=runtime_arn,
+            contentType="application/json",
+            accept="application/json",
+            runtimeSessionId=session_id,
+            payload=json.dumps(body).encode("utf-8"),
+        )
+        result = json.loads(resp["response"].read().decode("utf-8"))
+        if show:
+            print(f"\n$ {command} {json.dumps(params or {})}")
+            print(json.dumps(result, indent=2))
+        return result
+
+    return invoke
+
+
+def main():
+    config = load_config()
+    region = config["region"]
+    runtime = boto3.client("bedrock-agentcore", region_name=region)
+
+    # One session id (>= 33 chars), reused so every command hits the same
+    # supervisor session and its running containers.
+    session_id = f"egress-coding-exec-demo-{uuid.uuid4().hex}"
+    assert len(session_id) >= 33, "runtimeSessionId must be at least 33 characters"
+    print(f"Session id: {session_id}")
+
+    invoke = make_invoker(runtime, config["runtime_arn"], session_id)
+
+    # 1. Start the broker (trusted egress proxy)
+    resp = invoke("start_broker", {"image_uri": config["broker_image"]})
+    assert resp.get("status") == "ok", f"start_broker failed: {resp}"
+
+    # 2. Start the agent (untrusted sandbox)
+    resp = invoke("start_agent", {"image_uri": config["agent_image"]})
+    assert resp.get("status") == "ok", f"start_agent failed: {resp}"
+
+    # 3. Configure the broker's egress allowlist (at runtime, no restart)
+    resp = invoke("configure_broker", {"allowed_ping_domains": ALLOWED_PING_DOMAINS})
+    assert resp.get("status") == "ok", f"configure_broker failed: {resp}"
+    assert "allowed_ping_domains" in resp.get("result", {}).get("updated", []), resp
+
+    # 4. Status — both containers should be running
+    resp = invoke("status")
+    assert resp.get("broker") == "running", f"broker not running: {resp}"
+    assert resp.get("agent") == "running", f"agent not running: {resp}"
+
+    # 5. ALLOW: broker lets the request through and the ping executes.
+    #    We do NOT assert ICMP reachability (exit_code 0): inside the microVM the
+    #    echo replies may not complete, which is a network-path artifact, not a
+    #    policy result.
+    allow = invoke("ping_domain", {"domain": "amazon.com", "count": 3, "timeout": 5})
+    assert allow.get("status") == "ok", f"expected ok, got: {allow}"
+    allow_result = allow.get("result", {})
+    assert "not in ping allowlist" not in allow_result.get("stderr", ""), f"unexpectedly denied: {allow_result}"
+    assert "PING amazon.com" in allow_result.get("stdout", ""), f"ping did not execute: {allow_result}"
+    print("\nALLOW: broker PERMITTED the request and ping executed against amazon.com.")
+
+    # 6. DENY: aws.amazon.com is also Amazon's, but the allowlist entry "amazon.com"
+    #    is an exact match, so the subdomain is blocked before any ping runs. The
+    #    agent wraps the POLICY_DENIED error into a ping-style result (status stays
+    #    "ok", exit_code -1, stderr carries the denial).
+    deny = invoke("ping_domain", {"domain": "aws.amazon.com", "count": 3, "timeout": 5})
+    assert deny.get("status") == "ok", f"unexpected envelope: {deny}"
+    deny_result = deny.get("result", {})
+    assert "not in ping allowlist" in deny_result.get("stderr", ""), f"expected a policy denial, got: {deny_result}"
+    print("DENY: broker BLOCKED the ping to aws.amazon.com (not in the allowlist).")
+
+    print("\n✓ Demo complete. Tear down with: python cleanup.py")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/requirements-dev.txt b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/requirements-dev.txt
new file mode 100644
index 000000000..666f0248d
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/requirements-dev.txt
@@ -0,0 +1,2 @@
+-r requirements.txt
+pytest>=7.4
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/requirements.txt b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/requirements.txt
new file mode 100644
index 000000000..63d05e023
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/requirements.txt
@@ -0,0 +1 @@
+boto3>=1.38.0
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/shared/ipc.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/shared/ipc.py
new file mode 100644
index 000000000..589a3e2c3
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/shared/ipc.py
@@ -0,0 +1,35 @@
+import json
+import socket
+import struct
+from typing import Optional
+
+HEADER_SIZE = 4
+HEADER_FORMAT = ">I"
+
+
+def send_message(sock: socket.socket, msg: dict) -> None:
+    payload = json.dumps(msg).encode("utf-8")
+    header = struct.pack(HEADER_FORMAT, len(payload))
+    sock.sendall(header + payload)
+
+
+def recv_message(sock: socket.socket) -> Optional[dict]:
+    header = _recv_exact(sock, HEADER_SIZE)
+    if header is None:
+        return None
+    (length,) = struct.unpack(HEADER_FORMAT, header)
+    payload = _recv_exact(sock, length)
+    if payload is None:
+        return None
+    message: dict = json.loads(payload.decode("utf-8"))
+    return message
+
+
+def _recv_exact(sock: socket.socket, n: int) -> Optional[bytes]:
+    buf = bytearray()
+    while len(buf) < n:
+        chunk = sock.recv(n - len(buf))
+        if not chunk:
+            return None
+        buf.extend(chunk)
+    return bytes(buf)
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/requirements.txt b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/requirements.txt
new file mode 100644
index 000000000..1b449a3bf
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/requirements.txt
@@ -0,0 +1,2 @@
+bedrock-agentcore>=0.1.0
+boto3>=1.38.0
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/__init__.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/containerd.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/containerd.py
new file mode 100644
index 000000000..29e8ecf95
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/containerd.py
@@ -0,0 +1,94 @@
+import logging
+import os
+import subprocess
+
+logger = logging.getLogger(__name__)
+
+CONTAINERD_SOCKET = os.environ.get("CONTAINERD_SOCKET", "/run/containerd/containerd.sock")
+
+
+def pull_image(image_uri: str, ecr_token: str) -> None:
+    cmd = [
+        "ctr",
+        "-a",
+        CONTAINERD_SOCKET,
+        "images",
+        "pull",
+        "--user",
+        f"AWS:{ecr_token}",
+        image_uri,
+    ]
+    logger.info("Pulling image: %s", image_uri)
+    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
+    if result.returncode != 0:
+        raise RuntimeError(f"Image pull failed: {result.stderr}")
+    logger.info("Image pulled successfully: %s", image_uri)
+
+
+def start_container(ctr_args: list) -> None:
+    logger.info("Starting container: %s", " ".join(ctr_args))
+    result = subprocess.run(ctr_args, capture_output=True, text=True, timeout=60)
+    if result.returncode != 0:
+        raise RuntimeError(f"Container start failed: {result.stderr}")
+    logger.info("Container started successfully")
+
+
+def stop_container(container_name: str) -> None:
+    kill_cmd = ["ctr", "-a", CONTAINERD_SOCKET, "tasks", "kill", container_name]
+    logger.info("Killing container task: %s", container_name)
+    subprocess.run(kill_cmd, capture_output=True, text=True, timeout=30)
+
+    task_del_cmd = ["ctr", "-a", CONTAINERD_SOCKET, "tasks", "delete", container_name]
+    logger.info("Deleting task: %s", container_name)
+    subprocess.run(task_del_cmd, capture_output=True, text=True, timeout=30)
+
+    delete_cmd = [
+        "ctr",
+        "-a",
+        CONTAINERD_SOCKET,
+        "containers",
+        "delete",
+        container_name,
+    ]
+    logger.info("Deleting container: %s", container_name)
+    result = subprocess.run(delete_cmd, capture_output=True, text=True, timeout=30)
+    if result.returncode != 0:
+        logger.warning("Container delete may have failed: %s", result.stderr)
+
+
+def get_container_status(container_name: str) -> str:
+    cmd = ["ctr", "-a", CONTAINERD_SOCKET, "tasks", "ls"]
+    result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
+    if result.returncode != 0:
+        return "unknown"
+
+    for line in result.stdout.strip().split("\n"):
+        parts = line.split()
+        if parts and parts[0] == container_name:
+            return parts[2].lower() if len(parts) > 2 else "exists"
+
+    return "not_started"
+
+
+def list_containers() -> dict:
+    """Return the raw containerd view from inside the runtime microVM.
+
+    Captures `ctr tasks ls` (running tasks + PIDs + state) and
+    `ctr containers ls` (all created containers + their images).
+    """
+    tasks = subprocess.run(
+        ["ctr", "-a", CONTAINERD_SOCKET, "tasks", "ls"],
+        capture_output=True,
+        text=True,
+        timeout=10,
+    )
+    containers = subprocess.run(
+        ["ctr", "-a", CONTAINERD_SOCKET, "containers", "ls"],
+        capture_output=True,
+        text=True,
+        timeout=10,
+    )
+    return {
+        "tasks_ls": tasks.stdout if tasks.returncode == 0 else tasks.stderr,
+        "containers_ls": (containers.stdout if containers.returncode == 0 else containers.stderr),
+    }
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/ecr.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/ecr.py
new file mode 100644
index 000000000..83c4ecce7
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/ecr.py
@@ -0,0 +1,15 @@
+import base64
+import logging
+
+import boto3
+
+logger = logging.getLogger(__name__)
+
+
+def get_ecr_token(region: str = "us-east-1") -> str:
+    client = boto3.client("ecr", region_name=region)
+    response = client.get_authorization_token()
+    auth_data = response["authorizationData"][0]
+    token = base64.b64decode(auth_data["authorizationToken"]).decode("utf-8")
+    # Token format is "AWS:<password>"
+    return token.split(":", 1)[1]
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/main.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/main.py
new file mode 100644
index 000000000..252a26a39
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/main.py
@@ -0,0 +1,204 @@
+import json
+import logging
+import os
+import sys
+from pathlib import Path
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+from bedrock_agentcore.runtime import BedrockAgentCoreApp
+
+from .containerd import (
+    get_container_status,
+    list_containers,
+    pull_image,
+    start_container,
+    stop_container,
+)
+from .ecr import get_ecr_token
+from .profiles import IPC_DIR, WORKSPACE_DIR, get_agent_ctr_args, get_broker_ctr_args
+from .task_dispatch import TaskDispatcher
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [supervisor] %(levelname)s: %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+BROKER_CONTAINER = "broker"
+AGENT_CONTAINER = "agent"
+
+dispatcher = TaskDispatcher(socket_path=os.path.join(IPC_DIR, "supervisor.sock"))
+
+Path(IPC_DIR).mkdir(parents=True, exist_ok=True)
+Path(WORKSPACE_DIR).mkdir(parents=True, exist_ok=True)
+dispatcher.start()
+logger.info("Supervisor ready")
+
+app = BedrockAgentCoreApp()
+
+
+@app.entrypoint
+async def handle(payload):
+    data = json.loads(payload) if isinstance(payload, str) else payload
+    command = data.get("command")
+    params = data.get("params", {})
+
+    if command == "start_broker":
+        return await _start_broker(params)
+    elif command == "start_agent":
+        return await _start_agent(params)
+    elif command == "ping_domain":
+        return await _ping_domain(params)
+    elif command == "configure_broker":
+        return await _configure_broker(params)
+    elif command == "stop_broker":
+        return await _stop_broker()
+    elif command == "stop_agent":
+        return await _stop_agent()
+    elif command == "status":
+        return _status()
+    elif command == "list_containers":
+        return _list_containers()
+    else:
+        return {"status": "error", "message": f"Unknown command: {command}"}
+
+
+def _status():
+    return {
+        "status": "ok",
+        "broker": get_container_status(BROKER_CONTAINER),
+        "agent": get_container_status(AGENT_CONTAINER),
+    }
+
+
+def _list_containers():
+    return {"status": "ok", "result": list_containers()}
+
+
+async def _start_broker(params: dict):
+    try:
+        image_uri = params["image_uri"]
+        env = params.get("env")
+
+        region = os.environ.get("AWS_REGION", "us-east-1")
+        token = get_ecr_token(region)
+        pull_image(image_uri, token)
+
+        ctr_args = get_broker_ctr_args(BROKER_CONTAINER, image_uri, env=env)
+        start_container(ctr_args)
+
+        return {"status": "ok", "message": "Broker started"}
+    except Exception as e:
+        logger.exception("Failed to start broker")
+        return {"status": "error", "message": str(e)}
+
+
+async def _start_agent(params: dict):
+    try:
+        image_uri = params["image_uri"]
+
+        region = os.environ.get("AWS_REGION", "us-east-1")
+        token = get_ecr_token(region)
+        pull_image(image_uri, token)
+
+        ctr_args = get_agent_ctr_args(AGENT_CONTAINER, image_uri)
+        start_container(ctr_args)
+
+        if not dispatcher.wait_for_agent(timeout=60):
+            raise RuntimeError("Agent did not connect within 60 seconds")
+
+        return {"status": "ok", "message": "Agent started and connected"}
+    except Exception as e:
+        logger.exception("Failed to start agent")
+        return {"status": "error", "message": str(e)}
+
+
+async def _ping_domain(params: dict):
+    broker_status = get_container_status(BROKER_CONTAINER)
+    agent_status = get_container_status(AGENT_CONTAINER)
+
+    if broker_status != "running":
+        return {
+            "status": "error",
+            "message": f"Broker not running (status: {broker_status})",
+        }
+    if agent_status != "running":
+        return {
+            "status": "error",
+            "message": f"Agent not running (status: {agent_status})",
+        }
+
+    domain = params.get("domain")
+    if not domain:
+        return {"status": "error", "message": "Missing 'domain' parameter"}
+
+    result = dispatcher.dispatch_task(
+        "ping_domain",
+        {
+            "domain": domain,
+            "count": params.get("count", 3),
+            "timeout": params.get("timeout", 5),
+        },
+    )
+
+    if result and result.get("status") == "ok":
+        return {"status": "ok", "result": result.get("result", {})}
+    else:
+        error = result.get("error", {}) if result else {}
+        return {"status": "error", "message": error.get("message", "Ping task failed")}
+
+
+async def _configure_broker(params: dict):
+    broker_status = get_container_status(BROKER_CONTAINER)
+    if broker_status != "running":
+        return {
+            "status": "error",
+            "message": f"Broker not running (status: {broker_status})",
+        }
+
+    import socket
+    import uuid
+
+    from shared.ipc import recv_message, send_message
+
+    broker_socket_path = os.path.join(IPC_DIR, "broker.sock")
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    try:
+        sock.connect(broker_socket_path)
+        msg = {
+            "id": str(uuid.uuid4()),
+            "method": "configure",
+            "params": {
+                "allowed_ping_domains": params.get("allowed_ping_domains"),
+                "allowed_domains": params.get("allowed_domains"),
+            },
+        }
+        send_message(sock, msg)
+        response = recv_message(sock)
+        return response if response else {"status": "error", "message": "No response from broker"}
+    except Exception as e:
+        logger.exception("Failed to configure broker")
+        return {"status": "error", "message": str(e)}
+    finally:
+        sock.close()
+
+
+async def _stop_agent():
+    try:
+        stop_container(AGENT_CONTAINER)
+        return {"status": "ok", "message": "Agent stopped"}
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+
+async def _stop_broker():
+    try:
+        stop_container(BROKER_CONTAINER)
+        return {"status": "ok", "message": "Broker stopped"}
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+
+if __name__ == "__main__":
+    app.run()
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/profiles.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/profiles.py
new file mode 100644
index 000000000..6472a21ed
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/profiles.py
@@ -0,0 +1,55 @@
+import os
+from typing import Optional
+
+CONTAINERD_SOCKET = os.environ.get("CONTAINERD_SOCKET", "/run/containerd/containerd.sock")
+IPC_DIR = "/run/containerd/sandbox-ipc"
+WORKSPACE_DIR = "/run/containerd/sandbox-workspace"
+
+
+def get_broker_ctr_args(container_name: str, image_uri: str, env: Optional[dict] = None) -> list:
+    args = [
+        "ctr",
+        "-a",
+        CONTAINERD_SOCKET,
+        "run",
+        "-d",
+        "--net-host",
+        "--mount",
+        f"type=bind,src={IPC_DIR},dst=/tmp/ipc,options=rbind:rw",
+        "--mount",
+        f"type=bind,src={WORKSPACE_DIR},dst=/tmp/workspace,options=rbind:rw",
+        "--env",
+        "IPC_SOCKET_PATH=/tmp/ipc/broker.sock",
+        "--env",
+        "WORKSPACE_PATH=/tmp/workspace",
+    ]
+
+    if env:
+        for key, value in env.items():
+            args.extend(["--env", f"{key}={value}"])
+
+    args.extend([image_uri, container_name])
+    return args
+
+
+def get_agent_ctr_args(container_name: str, image_uri: str) -> list:
+    args = [
+        "ctr",
+        "-a",
+        CONTAINERD_SOCKET,
+        "run",
+        "-d",
+        "--mount",
+        f"type=bind,src={IPC_DIR},dst=/tmp/ipc,options=rbind:rw",
+        "--mount",
+        f"type=bind,src={WORKSPACE_DIR},dst=/tmp/workspace,options=rbind:rw",
+        "--env",
+        "BROKER_SOCKET_PATH=/tmp/ipc/broker.sock",
+        "--env",
+        "SUPERVISOR_SOCKET_PATH=/tmp/ipc/supervisor.sock",
+        "--env",
+        "WORKSPACE_PATH=/tmp/workspace",
+        image_uri,
+        container_name,
+    ]
+    return args
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/task_dispatch.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/task_dispatch.py
new file mode 100644
index 000000000..de86a2333
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/supervisor/src/task_dispatch.py
@@ -0,0 +1,99 @@
+import logging
+import os
+import socket
+import sys
+import threading
+import uuid
+from pathlib import Path
+from typing import Optional
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
+
+from shared.ipc import recv_message, send_message
+
+logger = logging.getLogger(__name__)
+
+SUPERVISOR_SOCKET_PATH = os.environ.get("SUPERVISOR_SOCKET_PATH", "/run/containerd/sandbox-ipc/supervisor.sock")
+
+
+class TaskDispatcher:
+    def __init__(self, socket_path: Optional[str] = None):
+        self._socket_path = socket_path or SUPERVISOR_SOCKET_PATH
+        self._server_sock: Optional[socket.socket] = None
+        self._agent_conn: Optional[socket.socket] = None
+        self._lock = threading.Lock()
+
+    def start(self):
+        path = Path(self._socket_path)
+        if path.exists():
+            path.unlink()
+        path.parent.mkdir(parents=True, exist_ok=True)
+
+        self._server_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        self._server_sock.bind(self._socket_path)
+        self._server_sock.listen(1)
+        # Intentional: the socket is created by the (root) supervisor, but the
+        # untrusted agent runs as UID 1000 and must be able to connect to it.
+        # Cross-UID access to this IPC socket is core to the sandbox design.
+        os.chmod(self._socket_path, 0o777)  # nosec B103
+        logger.info("Task dispatch socket listening at %s", self._socket_path)
+
+    def wait_for_agent(self, timeout: float = 60) -> bool:
+        if self._server_sock is None:
+            raise RuntimeError("Dispatcher not started; call start() first")
+        self._server_sock.settimeout(timeout)
+        try:
+            self._agent_conn, addr = self._server_sock.accept()
+            logger.info("Agent connected to task dispatch socket")
+            return True
+        except socket.timeout:
+            logger.error("Timed out waiting for agent to connect")
+            return False
+
+    def dispatch_task(self, method: str, params: dict, timeout: float = 660) -> dict:
+        with self._lock:
+            if not self._agent_conn:
+                return {
+                    "status": "error",
+                    "error": {"code": "NO_AGENT", "message": "Agent not connected"},
+                }
+
+            request_id = str(uuid.uuid4())
+            msg = {"id": request_id, "method": method, "params": params}
+
+            self._agent_conn.settimeout(timeout)
+            try:
+                send_message(self._agent_conn, msg)
+                response = recv_message(self._agent_conn)
+                return (
+                    response
+                    if response
+                    else {
+                        "status": "error",
+                        "error": {
+                            "code": "NO_RESPONSE",
+                            "message": "Agent disconnected",
+                        },
+                    }
+                )
+            except socket.timeout:
+                return {
+                    "status": "error",
+                    "error": {"code": "TIMEOUT", "message": "Agent task timed out"},
+                }
+            except Exception as e:
+                return {
+                    "status": "error",
+                    "error": {"code": "DISPATCH_ERROR", "message": str(e)},
+                }
+
+    def stop(self):
+        if self._agent_conn:
+            self._agent_conn.close()
+            self._agent_conn = None
+        if self._server_sock:
+            self._server_sock.close()
+            self._server_sock = None
+        path = Path(self._socket_path)
+        if path.exists():
+            path.unlink()
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/tests/conftest.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/tests/conftest.py
new file mode 100644
index 000000000..d74d60349
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/tests/conftest.py
@@ -0,0 +1,16 @@
+"""Pytest path setup.
+
+The sample is a multi-image project (not an installed package), so the source
+directories are made importable here the same way the containers do it via
+PYTHONPATH=/app. This lets the tests import `shared.ipc` and the broker sources.
+"""
+
+import os
+import sys
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+# Sample root (for `shared.ipc`) and the broker package root (for `src.config`).
+for path in (ROOT, os.path.join(ROOT, "broker")):
+    if path not in sys.path:
+        sys.path.insert(0, path)
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/tests/unit/test_broker_config.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/tests/unit/test_broker_config.py
new file mode 100644
index 000000000..c70577b0f
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/tests/unit/test_broker_config.py
@@ -0,0 +1,49 @@
+"""Unit tests for the broker allowlist policy (broker/src/config.py)."""
+
+from src.config import BrokerConfig
+
+
+def test_exact_domain_match_allowed():
+    cfg = BrokerConfig()
+    cfg.update(allowed_ping_domains=["google.com", "amazon.com"])
+    assert cfg.is_ping_domain_allowed("google.com") is True
+    assert cfg.is_ping_domain_allowed("amazon.com") is True
+
+
+def test_domain_not_in_allowlist_is_denied():
+    cfg = BrokerConfig()
+    cfg.update(allowed_ping_domains=["google.com"])
+    assert cfg.is_ping_domain_allowed("example.com") is False
+
+
+def test_glob_wildcard_matching():
+    """fnmatch glob patterns (e.g. *.amazon.com) are honored."""
+    cfg = BrokerConfig()
+    cfg.update(allowed_ping_domains=["*.amazon.com"])
+    assert cfg.is_ping_domain_allowed("aws.amazon.com") is True
+    assert cfg.is_ping_domain_allowed("docs.amazon.com") is True
+    assert cfg.is_ping_domain_allowed("amazon.com") is False  # no subdomain
+
+
+def test_empty_allowlist_denies_everything():
+    cfg = BrokerConfig()
+    assert cfg.is_ping_domain_allowed("google.com") is False
+
+
+def test_update_reports_which_lists_changed():
+    cfg = BrokerConfig()
+    updated = cfg.update(allowed_ping_domains=["a.com"], allowed_domains=["b.com"])
+    assert set(updated) == {"allowed_ping_domains", "allowed_domains"}
+
+    # Only the ping list this time.
+    updated = cfg.update(allowed_ping_domains=["c.com"])
+    assert updated == ["allowed_ping_domains"]
+
+
+def test_allowlist_getters_return_copies():
+    """Mutating a returned list must not affect internal state."""
+    cfg = BrokerConfig()
+    cfg.update(allowed_ping_domains=["google.com"])
+    returned = cfg.allowed_ping_domains
+    returned.append("evil.com")
+    assert "evil.com" not in cfg.allowed_ping_domains
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/tests/unit/test_ipc.py b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/tests/unit/test_ipc.py
new file mode 100644
index 000000000..9f56826a3
--- /dev/null
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/12-egress-coding-execution/tests/unit/test_ipc.py
@@ -0,0 +1,55 @@
+"""Unit tests for the shared length-prefixed JSON IPC framing."""
+
+import socket
+import struct
+
+from shared.ipc import HEADER_SIZE, recv_message, send_message
+
+
+def test_round_trip_simple_message():
+    """A message sent on one end is received identically on the other."""
+    a, b = socket.socketpair()
+    try:
+        msg = {"id": "abc-123", "method": "ping", "params": {"domain": "google.com"}}
+        send_message(a, msg)
+        assert recv_message(b) == msg
+    finally:
+        a.close()
+        b.close()
+
+
+def test_round_trip_preserves_unicode_and_nesting():
+    """Nested structures and non-ASCII content survive the round trip."""
+    a, b = socket.socketpair()
+    try:
+        msg = {"result": {"stdout": "café ✓ 日本", "nested": [1, 2, {"x": True}]}}
+        send_message(a, msg)
+        assert recv_message(b) == msg
+    finally:
+        a.close()
+        b.close()
+
+
+def test_recv_returns_none_on_closed_connection():
+    """recv_message returns None (not an exception) when the peer closed."""
+    a, b = socket.socketpair()
+    a.close()
+    try:
+        assert recv_message(b) is None
+    finally:
+        b.close()
+
+
+def test_wire_format_is_big_endian_length_prefix():
+    """The 4-byte header is a big-endian uint32 of the JSON payload length."""
+    a, b = socket.socketpair()
+    try:
+        send_message(a, {"k": "v"})
+        header = b.recv(HEADER_SIZE)
+        (length,) = struct.unpack(">I", header)
+        payload = b.recv(length)
+        assert length == len(payload)
+        assert payload == b'{"k": "v"}'
+    finally:
+        a.close()
+        b.close()
diff --git a/01-features/02-host-your-agent/01-runtime/03-advanced/README.md b/01-features/02-host-your-agent/01-runtime/03-advanced/README.md
index 3525034f2..5d61173d2 100644
--- a/01-features/02-host-your-agent/01-runtime/03-advanced/README.md
+++ b/01-features/02-host-your-agent/01-runtime/03-advanced/README.md
@@ -10,14 +10,14 @@ Beyond basic agent and tool hosting, AgentCore runtime provides advanced capabil
 | 02 | [Session Management](02-session-management/) | Maintain context across invocations, session isolation | deploy/invoke/cleanup |
 | 03 | [Bidirectional Streaming](03-bidirectional-streaming/) | WebSocket-based real-time voice agents | Reference (Docker-based) |
 | 04 | [Async Agents](04-async-agents/) | Long-running background tasks with async task tracking | Notebooks |
-| 04b | [Bidirectional Streaming (WebRTC)](04-bidirectional-streaming-webrtc/) | Voice agents with Kinesis Video Streams | Reference (Docker-based) |
 | 05 | [Execute Commands](05-execute-commands/) | Run shell commands inside runtime sessions | deploy/invoke/cleanup |
 | 06 | [Multi-Agent](06-multi-agent/) | Orchestrate multiple agents across runtimes | deploy/invoke/cleanup |
-| 06b | [Persistent Filesystems](06-persistent-filesystems/) | Persist files across session stop/resume cycles | deploy/invoke/cleanup |
+| 07 | [Persistent Filesystems](07-persistent-filesystems/) | Persist files across session stop/resume cycles | deploy/invoke/cleanup |
 | 08 | [Connect to VPC Resources](08-connect-to-vpc-resources/) | Deploy agents in your VPC for private resource access | CDK (TypeScript) |
-| 08b | [MCP End-to-End](08-mcp-e2e/) | Full MCP feature set: tools, resources, prompts, sampling, elicitation, progress | Notebooks |
-| 09 | [MCP Dynamic Client Registration](09-mcp-dynamic-client-registration/) | Auth0 OAuth + DCR for MCP server authentication | deploy/invoke/cleanup |
-| 10 | [Middleware Support](10-middleware-support/) | Starlette middleware for observability and error handling | deploy/invoke/cleanup |
+| 09 | [MCP End-to-End](09-mcp-e2e/) | Full MCP feature set: tools, resources, prompts, sampling, elicitation, progress | Notebooks |
+| 10 | [MCP Dynamic Client Registration](10-mcp-dynamic-client-registration/) | Auth0 OAuth + DCR for MCP server authentication | deploy/invoke/cleanup |
+| 11 | [Middleware Support](11-middleware-support/) | Starlette middleware for observability and error handling | deploy/invoke/cleanup |
+| 12 | [Egress-Controlled Code Execution](12-egress-coding-execution/) | Sandbox untrusted code in the microVM with broker-mediated, allowlisted egress | Reference (Docker-based) |
 
 ## Tutorial Formats
 
diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md
index 13e076748..c75cd98fe 100644
--- a/CONTRIBUTORS.md
+++ b/CONTRIBUTORS.md
@@ -122,3 +122,5 @@
 - JobRamos (jobdram)
 - Will Matos (wilmatos)
 - Senthil Mohan (skmohan)
+- Fabio Balancin (balancin)
+- Varun Gunda (vvargu)