| 71 | |
| 72 | |
| 73 | class Fixture: |
| 74 | def __init__(self, binary, root): |
| 75 | self.binary = str(binary) |
| 76 | self.root = root |
| 77 | self.env = os.environ.copy() |
| 78 | self.env["NO_COLOR"] = "1" |
| 79 | # The foreground owner is the only process this fixture owns and stops. |
| 80 | self.owner = None |
| 81 | self.owner_log = tempfile.TemporaryFile() |
| 82 | |
| 83 | def start(self): |
| 84 | self.run("init", str(self.root)) |
| 85 | canonical = str((self.root / ".atomic").resolve()) |
| 86 | digest = blake3.blake3(canonical.encode()).hexdigest()[:24] |
| 87 | self.endpoint = f"/tmp/atomic-owner-{digest}.sock" |
| 88 | self.owner = subprocess.Popen( |
| 89 | [self.binary, "agent", "database-owner", "serve", "--repository", str(self.root)], |
| 90 | stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=self.owner_log, |
| 91 | env=self.env) |
| 92 | deadline = time.monotonic() + 15 |
| 93 | while True: |
| 94 | try: |
| 95 | self.rpc("Ping") |
| 96 | break |
| 97 | except (OSError, RuntimeError): |
| 98 | if self.owner.poll() is not None or time.monotonic() > deadline: |
| 99 | raise RuntimeError("owner failed to start") |
| 100 | time.sleep(.025) |
| 101 | self.hook("session-start", {"session_id": SESSION}) |
| 102 | |
| 103 | def run(self, *args, payload=None, timeout=45): |
| 104 | result = subprocess.run([self.binary, *args, "--no-color"], cwd=self.root, |
| 105 | input=payload, stdout=subprocess.PIPE, stderr=subprocess.PIPE, |
| 106 | timeout=timeout, env=self.env) |
| 107 | if result.returncode: |
| 108 | raise RuntimeError(f"{' '.join(args)}: {result.stderr.decode(errors='replace')}") |
| 109 | return result |
| 110 | |
| 111 | def hook(self, verb, payload): |
| 112 | result = self.run("agent", "hooks", "claude-code", verb, "--foreground", |
| 113 | payload=encoded(payload)) |
| 114 | if result.stderr: |
| 115 | raise RuntimeError(result.stderr.decode(errors="replace")) |
| 116 | return result |
| 117 | |
| 118 | def rpc(self, request): |
| 119 | # Match the owner's versioned JSON frame and one connection per RPC. |
| 120 | frame = encoded({"version": 1, "request_id": "perf", "request": request}) |
| 121 | with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as stream: |
| 122 | stream.settimeout(45) |
| 123 | stream.connect(self.endpoint) |
| 124 | stream.sendall(struct.pack(">I", len(frame)) + frame) |
| 125 | |
| 126 | def read_exact(size): |
| 127 | data = bytearray() |
| 128 | while len(data) < size: |
| 129 | chunk = stream.recv(size - len(data)) |
| 130 | if not chunk: |