Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
a32ea2b
feat(scheduler): GPU arbiter — VRAM-accounted admission + queue + evi…
hognek Jul 6, 2026
9714de0
fix(scheduler): preempt running GPU work on eviction, not just cancel…
hognek Jul 7, 2026
1277abe
fix(scheduler): clear lease-release ownership in eviction vs. completion
hognek Jul 7, 2026
74b1185
fix(tests): remove test_gpu_arbiter.py — uses #1689-style API incompa…
hognek Jul 7, 2026
815cca8
fix(scheduler): close TOCTOU race in GPU admission with in-flight VRA…
hognek Jul 7, 2026
eb377f6
fix(scheduler): move lease claim inside try/finally + subtract reserv…
hognek Jul 7, 2026
91fda18
fix(scheduler): non-blocking drain + eviction-to-make-room in GPU arb…
hognek Jul 7, 2026
b616b84
fix(scheduler): use _reserve_and_check in post-eviction re-admit path
hognek Jul 7, 2026
752d422
fix(scheduler): add _running_lock coverage to eviction paths in GpuAr…
hognek Jul 7, 2026
eec8e5a
fix(scheduler): wire GpuArbiter with real VRAM probe in app.py
hognek Jul 7, 2026
82e7611
fix(scheduler): add _running_lock coverage to eviction paths
hognek Jul 7, 2026
4da4cd1
feat(cluster): worker auto-update with graceful drain, pause, install…
hognek Jul 6, 2026
55ece87
fix(cluster): use exact worker-name compare in lease drain matching
hognek Jul 7, 2026
bf43f4e
feat(scheduler): pause/resume GPU queue + hardware-aware LLM admission
hognek Jul 7, 2026
2983400
fix(tests): add missing await to _evict_task calls in test_gpu_arbite…
hognek Jul 7, 2026
814e5fd
fix(cluster): make drain_worker/cancel_drain async, hold _lease_lock …
hognek Jul 7, 2026
b6e17e6
fix(cluster): move worker.status='offline' inside _lease_lock; return…
hognek Jul 7, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -124,3 +124,4 @@ docs/STATUS.md
data/pending-restart.json
# Stray dev screenshot artifacts
desktop-initial.png
venv/
118 changes: 118 additions & 0 deletions tests/test_cluster.py
Original file line number Diff line number Diff line change
Expand Up @@ -259,3 +259,121 @@ async def test_router_returns_none_for_no_workers(self):
data, worker_name = await router.route_request("chat", "POST", "/v1/chat/completions", {})
assert data is None
assert worker_name is None


@pytest.mark.asyncio
class TestWorkerDrain:
"""Tests for taOS #890 — worker auto-update with graceful drain."""

async def test_drain_worker_graceful_sets_draining_status(self):
"""Graceful drain sets status to 'draining', leaves leases intact."""
mgr = ClusterManager()
await mgr.register_worker(_make_worker("gpu-box", capabilities=["chat"]))

result = await mgr.drain_worker("gpu-box", graceful=True)
assert result["worker"] == "gpu-box"
assert result["previous_status"] == "online"
assert result["status"] == "draining"
assert result["released_leases"] == 0
assert mgr.get_worker("gpu-box").status == "draining"

async def test_drain_worker_force_releases_leases_and_marks_offline(self):
"""Force drain releases all leases and marks worker offline."""
mgr = ClusterManager()
await mgr.register_worker(_make_worker("gpu-box", capabilities=["chat"]))

# Add a lease for this worker
from tinyagentos.cluster.worker_protocol import GpuLease
import time
lease = GpuLease(
lease_id="l_test1",
resource_id="gpu-box:gpu-cuda-0",
caller="test",
expires_at=time.time() + 60,
required_vram_mb=0,
)
mgr._leases["l_test1"] = lease

result = await mgr.drain_worker("gpu-box", graceful=False)
assert result["released_leases"] == 1
assert result["status"] == "offline"
assert mgr.get_worker("gpu-box").status == "offline"
assert "l_test1" not in mgr._leases

async def test_drain_worker_unknown_returns_error(self):
"""Draining a non-existent worker returns error dict."""
mgr = ClusterManager()
result = await mgr.drain_worker("nonexistent")
assert "error" in result
assert result["worker"] == "nonexistent"

async def test_cancel_drain_returns_worker_to_online(self):
"""Cancel drain returns a draining worker back to online."""
mgr = ClusterManager()
await mgr.register_worker(_make_worker("gpu-box", capabilities=["chat"]))
await mgr.drain_worker("gpu-box", graceful=True)
assert mgr.get_worker("gpu-box").status == "draining"

result = await mgr.cancel_drain("gpu-box")
assert result["worker"] == "gpu-box"
assert result["status"] == "online"
assert mgr.get_worker("gpu-box").status == "online"

async def test_cancel_drain_only_works_on_draining(self):
"""Cancel drain returns error for non-draining workers."""
mgr = ClusterManager()
await mgr.register_worker(_make_worker("gpu-box"))

result = await mgr.cancel_drain("gpu-box")
assert "error" in result
assert "not draining" in result["error"]

async def test_cancel_drain_unknown_returns_error(self):
"""Cancel drain on unknown worker returns error."""
mgr = ClusterManager()
result = await mgr.cancel_drain("nonexistent")
assert "error" in result

async def test_draining_workers_excluded_from_routing(self):
"""Draining workers should not receive new tasks."""
mgr = ClusterManager()
await mgr.register_worker(_make_worker("online-gpu", capabilities=["chat"], load=0.1))
await mgr.register_worker(_make_worker("draining-gpu", capabilities=["chat"], load=0.0))
await mgr.drain_worker("draining-gpu", graceful=True)

result = mgr.get_workers_for_capability("chat")
assert len(result) == 1
assert result[0].name == "online-gpu"

async def test_draining_workers_excluded_from_catalog(self):
"""Draining workers are excluded from the aggregate catalog."""
mgr = ClusterManager()
online = _make_worker("online", capabilities=["chat"])
online.backends = [{"name": "b1", "type": "ollama", "url": "u", "capabilities": ["chat"], "models": [{"name": "m1"}]}]
draining = _make_worker("draining", capabilities=["chat"])
draining.backends = [{"name": "b2", "type": "ollama", "url": "u", "capabilities": ["chat"], "models": [{"name": "m2"}]}]
await mgr.register_worker(online)
await mgr.register_worker(draining)
await mgr.drain_worker("draining", graceful=True)

out = mgr.aggregate_catalog()
assert [w["name"] for w in out["workers"]] == ["online"]
assert [m["name"] for m in out["models"]] == ["m1"]

async def test_draining_workers_excluded_from_lease_claim(self):
"""Lease claims should fail for draining workers."""
mgr = ClusterManager()
await mgr.register_worker(_make_worker("draining-gpu", url="http://draining:9000"))

# Set up worker with VRAM info
worker = mgr.get_worker("draining-gpu")
worker.free_vram_mb = 8000

await mgr.drain_worker("draining-gpu", graceful=True)

lease = await mgr.claim_lease(
resource_id="draining-gpu:gpu-cuda-0",
caller="test",
ttl_seconds=30,
)
assert lease is None
Loading
Loading