feat(ip-pool): pre-assign static IPs to build VMs via VMX guestinfo

Eliminates VMware-Tools IP-detect polling (observed: 39–177 s variance on
Linux host B7 benchmark).  When [ip_pool] is configured the orchestrator:

  1. Allocates a slot IP from a JSON-backed pool (fcntl-locked on Linux)
  2. Injects guestinfo.ip-assignment / netmask / gateway into the cloned VMX
     before vmrun start
  3. Skips _wait_for_ip entirely — uses the pre-assigned IP immediately
  4. Releases the slot in _destroy_clone (all paths: success/failure/interrupt)

New modules / changes:
  src/ci_orchestrator/ip_pool.py      IpSlotPool + _file_lock context manager
  src/ci_orchestrator/config.py       IpPoolConfig dataclass; [ip_pool] TOML
  src/ci_orchestrator/commands/job.py _inject_guestinfo_ip helper; pool wiring
  tests/python/test_ip_pool.py        12 unit tests (acquire/release/persist)
  tests/python/test_config.py         ip_pool config parsing tests
  tests/python/test_commands_job.py   pool path + _inject_guestinfo_ip tests
  config.example.toml                 [ip_pool] section (commented example)

Coverage 90.02%; mypy --strict clean.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

tests/python/test_commands_job.py

@@ -531,3 +531,110 @@ def test_job_wait_for_ip_returns_none_on_backend_error(
         job_module._wait_for_ip(_Boom(), VmHandle("x"), deadline=0.5, poll=0.001)  # type: ignore[arg-type]
         is None
     )
+
+
+# ── _inject_guestinfo_ip ─────────────────────────────────────────────────────
+
+
+def test_inject_guestinfo_ip_appends_lines(tmp_path: Path) -> None:
+    vmx = tmp_path / "test.vmx"
+    vmx.write_text('guestOS = "ubuntu-64"\n', encoding="utf-8")
+    job_module._inject_guestinfo_ip(vmx, "192.168.79.201", "255.255.255.0", "192.168.79.2")
+    content = vmx.read_text()
+    assert 'guestinfo.ip-assignment = "192.168.79.201"' in content
+    assert 'guestinfo.netmask = "255.255.255.0"' in content
+    assert 'guestinfo.gateway = "192.168.79.2"' in content
+    assert 'guestOS = "ubuntu-64"' in content
+
+
+def test_inject_guestinfo_ip_replaces_existing(tmp_path: Path) -> None:
+    vmx = tmp_path / "test.vmx"
+    vmx.write_text(
+        'guestOS = "ubuntu-64"\n'
+        'guestinfo.ip-assignment = "192.168.79.100"\n',
+        encoding="utf-8",
+    )
+    job_module._inject_guestinfo_ip(vmx, "192.168.79.201", "255.255.255.0", "")
+    content = vmx.read_text()
+    assert "192.168.79.100" not in content
+    assert 'guestinfo.ip-assignment = "192.168.79.201"' in content
+
+
+def test_inject_guestinfo_ip_omits_empty_gateway(tmp_path: Path) -> None:
+    vmx = tmp_path / "test.vmx"
+    vmx.write_text('guestOS = "ubuntu-64"\n', encoding="utf-8")
+    job_module._inject_guestinfo_ip(vmx, "192.168.79.201", "255.255.255.0", "")
+    assert "guestinfo.gateway" not in vmx.read_text()
+
+
+# ── job with ip_pool ─────────────────────────────────────────────────────────
+
+
+def test_job_with_ip_pool_uses_preassigned_ip(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, template_vmx: Path
+) -> None:
+    """When ip_pool is configured the job skips _wait_for_ip and uses the pool IP."""
+    from ci_orchestrator.config import Config, IpPoolConfig
+    from ci_orchestrator.ip_pool import IpSlotPool
+
+    backend = _FakeBackend(running_after=1)
+    calls = _patch_common(monkeypatch, backend)
+
+    pool_addresses = ["192.168.79.201"]
+    lease_file = tmp_path / "ip-pool.json"
+    pool_cfg = IpPoolConfig(
+        addresses=pool_addresses,
+        netmask="255.255.255.0",
+        gateway="192.168.79.2",
+        lease_file=lease_file,
+    )
+
+    # Patch load_config to return a config with ip_pool set.
+    real_cfg = job_module.load_config()
+    patched_cfg = Config(
+        paths=real_cfg.paths,
+        backend=real_cfg.backend,
+        vmrun_path=real_cfg.vmrun_path,
+        ssh_key_path=real_cfg.ssh_key_path,
+        guest_cred_target=real_cfg.guest_cred_target,
+        ip_pool=pool_cfg,
+    )
+    monkeypatch.setattr(job_module, "load_config", lambda: patched_cfg)
+
+    base_clone = tmp_path / "build-vms"
+    base_artifact = tmp_path / "artifacts"
+
+    result = CliRunner().invoke(
+        cli, _common_args(template_vmx, clone_base=base_clone, artifact_base=base_artifact)
+    )
+    assert result.exit_code == 0, result.output
+
+    # backend.get_ip was NOT called — IP came from pool.
+    assert "get_ip" not in backend.calls
+
+    # The pre-assigned IP appears in output.
+    assert "192.168.79.201" in result.output
+    assert "pre-assigned" in result.output
+
+    # Pool slot was released after job.
+    released = IpSlotPool(pool_addresses, "255.255.255.0", "", lease_file)._read_leases()
+    assert released["192.168.79.201"] is None
+
+    # Build was invoked with the pool IP.
+    assert calls["linux_build"][0]["ip_address"] == "192.168.79.201"
+
+    # VMX injection confirmed via log output (clone dir removed after job).
+    assert "pre-assigned" in result.output
+
+
+def test_destroy_clone_releases_pool_ip(tmp_path: Path) -> None:
+    from ci_orchestrator.ip_pool import IpSlotPool
+
+    lease = tmp_path / "pool.json"
+    pool = IpSlotPool(["10.0.0.1"], "255.255.255.0", "", lease)
+    ip = pool.acquire("owner")
+
+    job_module._destroy_clone(None, None, None, pool=pool, pool_ip=ip)
+
+    leases = pool._read_leases()
+    assert leases["10.0.0.1"] is None

tests/python/test_config.py

@@ -107,3 +107,41 @@ def test_toml_provides_ssh_key_path_and_guest_cred_target(tmp_path: Path) -> Non
     cfg = load_config(config_path=toml, env={})
     assert cfg.ssh_key_path == key
     assert cfg.guest_cred_target == "MyTarget"
+
+
+def test_toml_ip_pool_parsed(tmp_path: Path) -> None:
+    lease = tmp_path / "ip-pool.json"
+    toml = tmp_path / "config.toml"
+    toml.write_text(
+        "[ip_pool]\n"
+        'addresses = ["192.168.79.201", "192.168.79.202"]\n'
+        'netmask   = "255.255.255.0"\n'
+        'gateway   = "192.168.79.2"\n'
+        f'lease_file = "{lease.as_posix()}"\n',
+        encoding="utf-8",
+    )
+    cfg = load_config(config_path=toml, env={})
+    assert cfg.ip_pool is not None
+    assert cfg.ip_pool.addresses == ["192.168.79.201", "192.168.79.202"]
+    assert cfg.ip_pool.netmask == "255.255.255.0"
+    assert cfg.ip_pool.gateway == "192.168.79.2"
+    assert cfg.ip_pool.lease_file == lease
+
+
+def test_toml_ip_pool_default_lease_file(tmp_path: Path) -> None:
+    toml = tmp_path / "config.toml"
+    toml.write_text(
+        "[ip_pool]\n"
+        'addresses = ["192.168.79.201"]\n',
+        encoding="utf-8",
+    )
+    cfg = load_config(config_path=toml, env={})
+    assert cfg.ip_pool is not None
+    assert cfg.ip_pool.lease_file.name == "ip-pool.json"
+
+
+def test_ip_pool_absent_when_not_configured(tmp_path: Path) -> None:
+    toml = tmp_path / "config.toml"
+    toml.write_text("[backend]\ntype = \"workstation\"\n", encoding="utf-8")
+    cfg = load_config(config_path=toml, env={})
+    assert cfg.ip_pool is None

tests/python/test_ip_pool.py

@@ -0,0 +1,117 @@
+"""Tests for ci_orchestrator.ip_pool — IpSlotPool acquire/release."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from ci_orchestrator.ip_pool import IpSlotPool
+
+
+def make_pool(tmp_path: Path, addresses: list[str] | None = None) -> IpSlotPool:
+    return IpSlotPool(
+        addresses=addresses or ["10.0.0.1", "10.0.0.2"],
+        netmask="255.255.255.0",
+        gateway="10.0.0.254",
+        lease_file=tmp_path / "ip-pool.json",
+    )
+
+
+# ── construction ────────────────────────────────────────────────────────────
+
+
+def test_empty_addresses_raises(tmp_path: Path) -> None:
+    with pytest.raises(ValueError, match="at least one"):
+        IpSlotPool([], "255.255.255.0", "", tmp_path / "pool.json")
+
+
+# ── acquire ─────────────────────────────────────────────────────────────────
+
+
+def test_acquire_returns_first_free_ip(tmp_path: Path) -> None:
+    pool = make_pool(tmp_path)
+    ip = pool.acquire("owner-1")
+    assert ip == "10.0.0.1"
+
+
+def test_acquire_second_slot(tmp_path: Path) -> None:
+    pool = make_pool(tmp_path)
+    pool.acquire("owner-1")
+    ip2 = pool.acquire("owner-2")
+    assert ip2 == "10.0.0.2"
+
+
+def test_acquire_writes_lease_file(tmp_path: Path) -> None:
+    pool = make_pool(tmp_path)
+    pool.acquire("owner-1")
+    leases = json.loads((tmp_path / "ip-pool.json").read_text())
+    assert leases["10.0.0.1"] == "owner-1"
+    assert leases["10.0.0.2"] is None
+
+
+def test_acquire_exhausted_raises(tmp_path: Path) -> None:
+    pool = make_pool(tmp_path, addresses=["10.0.0.1"])
+    pool.acquire("owner-1")
+    with pytest.raises(RuntimeError, match="exhausted"):
+        pool.acquire("owner-2", timeout=0.1)
+
+
+# ── release ──────────────────────────────────────────────────────────────────
+
+
+def test_release_frees_slot(tmp_path: Path) -> None:
+    pool = make_pool(tmp_path)
+    ip = pool.acquire("owner-1")
+    pool.release(ip)
+    leases = json.loads((tmp_path / "ip-pool.json").read_text())
+    assert leases[ip] is None
+
+
+def test_release_unknown_ip_is_noop(tmp_path: Path) -> None:
+    pool = make_pool(tmp_path)
+    pool.release("99.99.99.99")  # should not raise
+
+
+def test_release_and_reacquire(tmp_path: Path) -> None:
+    pool = make_pool(tmp_path, addresses=["10.0.0.1"])
+    ip = pool.acquire("owner-1")
+    pool.release(ip)
+    ip2 = pool.acquire("owner-2")
+    assert ip2 == "10.0.0.1"
+
+
+# ── persistence ──────────────────────────────────────────────────────────────
+
+
+def test_second_pool_instance_sees_leases(tmp_path: Path) -> None:
+    lease = tmp_path / "ip-pool.json"
+    p1 = IpSlotPool(["10.0.0.1", "10.0.0.2"], "255.255.255.0", "", lease)
+    p1.acquire("owner-1")
+
+    p2 = IpSlotPool(["10.0.0.1", "10.0.0.2"], "255.255.255.0", "", lease)
+    ip = p2.acquire("owner-2")
+    assert ip == "10.0.0.2"
+
+
+def test_corrupt_lease_file_treated_as_empty(tmp_path: Path) -> None:
+    lease = tmp_path / "ip-pool.json"
+    lease.write_text("NOT JSON", encoding="utf-8")
+    pool = make_pool(tmp_path)
+    ip = pool.acquire("owner-1")
+    assert ip == "10.0.0.1"
+
+
+def test_missing_lease_file_treated_as_empty(tmp_path: Path) -> None:
+    pool = make_pool(tmp_path)
+    ip = pool.acquire("owner-1")
+    assert ip == "10.0.0.1"
+
+
+def test_non_dict_lease_json_treated_as_empty(tmp_path: Path) -> None:
+    lease = tmp_path / "ip-pool.json"
+    lease.write_text("[1, 2, 3]", encoding="utf-8")  # valid JSON but not a dict
+    pool = make_pool(tmp_path)
+    ip = pool.acquire("owner-1")
+    assert ip == "10.0.0.1"