|
| 1 | +"""Docker transport for runner containers. |
| 2 | +
|
| 3 | +The Protocol is the seam tests plug into: a hand-written double implements it |
| 4 | +without pulling in the Docker SDK. |
| 5 | +""" |
| 6 | + |
| 7 | +import base64 |
| 8 | +from collections.abc import AsyncIterator |
| 9 | +from typing import Protocol |
| 10 | + |
| 11 | +CLAUDE_RUNNER_IMAGE = "claude-runner:latest" |
| 12 | + |
| 13 | +CONTAINER_MEMORY_BYTES = 512 * 1024 * 1024 |
| 14 | +CONTAINER_NANO_CPUS = 1_000_000_000 |
| 15 | + |
| 16 | +# The entrypoint reads newline-terminated commands from this FIFO. |
| 17 | +_INPUT_FIFO = "/tmp/claude-input" |
| 18 | + |
| 19 | + |
| 20 | +class DockerClient(Protocol): |
| 21 | + """What the container module needs from a Docker daemon.""" |
| 22 | + |
| 23 | + async def create_container( |
| 24 | + self, |
| 25 | + image: str, |
| 26 | + environment: dict[str, str], |
| 27 | + volumes: list[str], |
| 28 | + labels: dict[str, str], |
| 29 | + ) -> str: |
| 30 | + """Create a container with stdin enabled and return its id.""" |
| 31 | + ... |
| 32 | + |
| 33 | + async def start_container(self, container_id: str) -> None: ... |
| 34 | + |
| 35 | + async def stop_container(self, container_id: str) -> None: ... |
| 36 | + |
| 37 | + async def remove_container(self, container_id: str, force: bool = False) -> None: ... |
| 38 | + |
| 39 | + # Not ``async def``: this returns an async iterator, it is not a coroutine |
| 40 | + # that resolves to one. Declaring it ``async`` made every implementation |
| 41 | + # fail the Protocol and would deadlock a double written to match it. |
| 42 | + def container_logs(self, container_id: str, follow: bool = False) -> AsyncIterator[str]: |
| 43 | + """Stream the container's stdout, line chunks as they arrive.""" |
| 44 | + ... |
| 45 | + |
| 46 | + async def write_to_container(self, container_id: str, data: str) -> None: |
| 47 | + """Write one message to the container's input FIFO.""" |
| 48 | + ... |
| 49 | + |
| 50 | + async def wait_container(self, container_id: str) -> int: |
| 51 | + """Block until the container exits; return its exit code.""" |
| 52 | + ... |
| 53 | + |
| 54 | + async def close(self) -> None: ... |
| 55 | + |
| 56 | + |
| 57 | +class AioDockerClient: |
| 58 | + """Production client wrapping aiodocker.""" |
| 59 | + |
| 60 | + def __init__(self) -> None: |
| 61 | + import aiodocker |
| 62 | + |
| 63 | + self._docker = aiodocker.Docker() |
| 64 | + |
| 65 | + async def create_container( |
| 66 | + self, |
| 67 | + image: str, |
| 68 | + environment: dict[str, str], |
| 69 | + volumes: list[str], |
| 70 | + labels: dict[str, str], |
| 71 | + ) -> str: |
| 72 | + container = await self._docker.containers.create( |
| 73 | + { |
| 74 | + "Image": image, |
| 75 | + "Env": [f"{k}={v}" for k, v in environment.items()], |
| 76 | + "Labels": labels, |
| 77 | + "OpenStdin": True, |
| 78 | + "HostConfig": { |
| 79 | + "Binds": volumes, |
| 80 | + "Memory": CONTAINER_MEMORY_BYTES, |
| 81 | + "NanoCPUs": CONTAINER_NANO_CPUS, |
| 82 | + "NetworkMode": "bridge", |
| 83 | + }, |
| 84 | + } |
| 85 | + ) |
| 86 | + return container.id |
| 87 | + |
| 88 | + async def start_container(self, container_id: str) -> None: |
| 89 | + container = await self._docker.containers.get(container_id) |
| 90 | + await container.start() |
| 91 | + |
| 92 | + async def stop_container(self, container_id: str) -> None: |
| 93 | + container = await self._docker.containers.get(container_id) |
| 94 | + await container.stop() |
| 95 | + |
| 96 | + async def remove_container(self, container_id: str, force: bool = False) -> None: |
| 97 | + container = await self._docker.containers.get(container_id) |
| 98 | + await container.delete(force=force) |
| 99 | + |
| 100 | + async def container_logs(self, container_id: str, follow: bool = False) -> AsyncIterator[str]: |
| 101 | + """Yield the container's stdout. |
| 102 | +
|
| 103 | + stdout only: Claude Code writes stream-json there, while stderr |
| 104 | + carries diagnostics that would duplicate events. aiodocker returns a |
| 105 | + list when not following and an async iterator when following, so the |
| 106 | + two cases are spelled out rather than passed a dynamic flag. |
| 107 | + """ |
| 108 | + container = await self._docker.containers.get(container_id) |
| 109 | + if follow: |
| 110 | + async for line in container.log(stdout=True, stderr=False, follow=True): |
| 111 | + yield line |
| 112 | + return |
| 113 | + for line in await container.log(stdout=True, stderr=False, follow=False): |
| 114 | + yield line |
| 115 | + |
| 116 | + async def write_to_container(self, container_id: str, data: str) -> None: |
| 117 | + """Deliver one message to the entrypoint's read loop. |
| 118 | +
|
| 119 | + The payload is base64-encoded so no shell metacharacter in user text |
| 120 | + can be interpreted, and newline-terminated so ``read`` returns it as a |
| 121 | + complete line. |
| 122 | + """ |
| 123 | + container = await self._docker.containers.get(container_id) |
| 124 | + encoded = base64.b64encode((data + "\n").encode()).decode() |
| 125 | + exec_obj = await container.exec(cmd=["sh", "-c", f"echo {encoded} | base64 -d > {_INPUT_FIFO}"]) |
| 126 | + await exec_obj.start(detach=True) |
| 127 | + |
| 128 | + async def wait_container(self, container_id: str) -> int: |
| 129 | + container = await self._docker.containers.get(container_id) |
| 130 | + result = await container.wait() |
| 131 | + return result["StatusCode"] |
| 132 | + |
| 133 | + async def close(self) -> None: |
| 134 | + await self._docker.close() |
0 commit comments