"""API tests via FastAPI TestClient (in-process, fast).""" def test_health(client): r = client.get("/health") assert r.status_code == 200 assert r.json()["status"] == "ok" def test_submit_minimal(client): r = client.post( "/submit", json={ "cmd": ["bash", "-c", "echo hi"], "cwd": "/tmp", "project": "project-a", }, ) assert r.status_code == 200, r.text body = r.json() assert body["id"] > 0 assert body["state"] == "queued" assert body["priority"] == 55 # project-a default from fixture def test_submit_persists_submitted_via_cli(client): """#51: submitted_via='cli' on the request body must round-trip through the broker DB so observability queries can count CLI vs MCP traffic.""" r = client.post( "/submit", json={ "cmd": ["bash", "-c", "x"], "cwd": "/tmp", "project": "project-a", "submitted_via": "cli", }, ) assert r.status_code == 200 job_id = r.json()["id"] r2 = client.get(f"/jobs/{job_id}") assert r2.json()["submitted_via"] == "cli" def test_submit_persists_submitted_via_mcp(client): """#51 mirror: submitted_via='mcp' round-trips identically.""" r = client.post( "/submit", json={ "cmd": ["bash", "-c", "x"], "cwd": "/tmp", "project": "project-a", "submitted_via": "mcp", }, ) assert r.status_code == 200 job_id = r.json()["id"] r2 = client.get(f"/jobs/{job_id}") assert r2.json()["submitted_via"] == "mcp" def test_submit_rejects_unknown_submitted_via(client): """submitted_via is constrained to the cli|mcp Literal — strings outside the set are rejected at the Pydantic boundary.""" r = client.post( "/submit", json={ "cmd": ["bash", "-c", "x"], "cwd": "/tmp", "project": "project-a", "submitted_via": "garbage", }, ) assert r.status_code == 422 def test_submit_omits_submitted_via_persists_null(client): """Old clients that don't pass submitted_via must still work; the column stays NULL so historical queries can distinguish 'pre-#51' from cli/mcp.""" r = client.post( "/submit", json={"cmd": ["bash", "-c", "x"], "cwd": "/tmp", "project": "project-a"}, ) assert r.status_code == 200 job_id = r.json()["id"] r2 = client.get(f"/jobs/{job_id}") assert r2.json()["submitted_via"] is None def test_submit_with_profile_applies_resources(client): r = client.post( "/submit", json={ "cmd": ["bash", "train.sh"], "cwd": "/tmp", "project": "project-b", "profile": "gpu-heavy", "priority_delta": 5, }, ) assert r.status_code == 200 body = r.json() assert body["priority"] == 85 # 80 + 5 job_id = body["id"] r2 = client.get(f"/jobs/{job_id}") assert r2.status_code == 200 j = r2.json() assert j["vram_gb"] == 28 assert j["ram_gb"] == 22 assert j["cpus"] == 8 def test_list_jobs(client): for i in range(3): client.post( "/submit", json={ "cmd": ["echo", str(i)], "cwd": "/tmp", "project": "project-a", }, ) r = client.get("/jobs") assert r.status_code == 200 assert len(r.json()) == 3 def test_submit_unknown_profile_404(client): r = client.post( "/submit", json={ "cmd": ["echo", "x"], "cwd": "/tmp", "project": "project-a", "profile": "nonexistent-profile", }, ) assert r.status_code == 404 def test_classify_unknown_cmd_not_heavy(client): r = client.post("/classify", json={"cmd": "ls -la"}) assert r.status_code == 200 assert r.json()["heavy"] is False def test_classify_known_heavy_cmd(client): r = client.post("/classify", json={"cmd": "bash train_lora_v5.sh"}) assert r.status_code == 200 body = r.json() assert body["heavy"] is True assert body["rule_id"] == "sdxl-lora-train" assert body["suggest_profile"] == "gpu-heavy" def test_list_workers_empty_before_any_heartbeat(client): r = client.get("/workers") assert r.status_code == 200 assert r.json() == [] def test_list_workers_returns_registered_worker(client): client.post( "/heartbeat", json={ "host": "desktop", "host_aliases": ["desktop-vm"], "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, "arch": "x86_64", "os": "linux", "gpu": True, "tags": ["python3", "R", "cuda", "wsl"], }, ) r = client.get("/workers") assert r.status_code == 200 body = r.json() assert len(body) == 1 w = body[0] assert w["host"] == "desktop" assert w["host_aliases"] == ["desktop-vm"] assert w["state"] == "online" assert w["gpu"] is True assert "R" in w["tags"] assert w["last_heartbeat"].endswith("+00:00") # A heartbeat that omits the slot fields reads as single-slot, idle. assert w["max_concurrent"] == 1 assert w["running"] == 0 def test_list_workers_surfaces_slot_usage(client): client.post( "/heartbeat", json={ "host": "desktop", "free_vram_gb": 9.1, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 6, "gpu": True, "max_concurrent": 3, "running": 2, }, ) w = client.get("/workers").json()[0] assert w["max_concurrent"] == 3 assert w["running"] == 2 def test_delete_worker_404_when_unknown(client): r = client.delete("/workers/nosuchhost") assert r.status_code == 404 def test_delete_worker_409_when_online(client): client.post( "/heartbeat", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ) r = client.delete("/workers/desktop") assert r.status_code == 409 assert "online" in r.json()["detail"].lower() # Row still present listing = client.get("/workers").json() assert any(w["host"] == "desktop" for w in listing) def test_delete_offline_worker_succeeds(client): from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Worker client.post( "/heartbeat", json={ "host": "ghosthost", "free_vram_gb": 0.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 4.0, "idle_cpus": 2, }, ) engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Worker) .where(Worker.host == "ghosthost") .values(state="offline", last_heartbeat=datetime.now(UTC) - timedelta(minutes=10)) ) r = client.delete("/workers/ghosthost") assert r.status_code == 200 body = r.json() assert body["ok"] is True assert body["deleted"] == "ghosthost" listing = client.get("/workers").json() assert all(w["host"] != "ghosthost" for w in listing) def test_heartbeat_registers_worker(client): r = client.post( "/heartbeat", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ) assert r.status_code == 200 def test_next_job_returns_highest_priority_fitting(client): # Submit 2 jobs; desktop worker should get the priority-80 one (project-b) client.post( "/submit", json={ "cmd": ["echo", "1"], "cwd": "/tmp", "project": "project-a", "profile": "small", "host_pin": "any", }, ) client.post( "/submit", json={ "cmd": ["echo", "2"], "cwd": "/tmp", "project": "project-b", "profile": "small", "host_pin": "any", }, ) # Heartbeat client.post( "/heartbeat", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ) r = client.post( "/next-job", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ) assert r.status_code == 200 body = r.json() assert body is not None assert body["project"] == "project-b" # priority 80 beats 55 assert body["state"] == "assigned" assert body["worker"] == "desktop" def test_next_job_empty_queue_returns_null(client): client.post( "/heartbeat", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ) r = client.post( "/next-job", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ) assert r.status_code == 200 assert r.json() is None def test_next_job_longpoll_times_out_returns_null(client): """M4: an async long-poll on an empty queue must return None near wait_s without holding a threadpool thread. Here it should return in ~0.4s, not hang and not return early.""" import time as _time client.post( "/heartbeat", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ) t0 = _time.monotonic() r = client.post( "/next-job", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, "wait_s": 0.4, }, ) elapsed = _time.monotonic() - t0 assert r.status_code == 200 assert r.json() is None assert 0.3 < elapsed < 3.0, f"long-poll should wait ~wait_s, took {elapsed:.2f}s" def test_next_job_longpoll_wakes_on_submit(client): """M4: a submit that lands while a worker is long-polling must WAKE the poll (via the cross-thread asyncio.Event), returning the job well before the wait_s deadline and before the _LONGPOLL_RECHECK_S backstop — proving the wait suspends on the loop rather than parking a thread until recheck.""" import threading import time as _time client.post( "/heartbeat", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ) result: dict = {} def _poll(): t0 = _time.monotonic() r = client.post( "/next-job", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, "wait_s": 8.0, }, ) result["elapsed"] = _time.monotonic() - t0 result["body"] = r.json() poller = threading.Thread(target=_poll) poller.start() _time.sleep(0.4) # let the poll start and suspend on the wake event client.post("/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"}) poller.join(timeout=5.0) assert not poller.is_alive(), "long-poll did not return after submit" assert result["body"] is not None and result["body"]["state"] == "assigned" # Woken by the submit, not by the 10s recheck backstop. assert result["elapsed"] < 5.0, f"wake was slow ({result['elapsed']:.2f}s) — recheck, not wake?" def test_append_log_and_complete(client): sub = client.post( "/submit", json={ "cmd": ["echo", "x"], "cwd": "/tmp", "project": "project-a", "profile": "small", "host_pin": "any", }, ).json() client.post( "/heartbeat", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ) claim = client.post( "/next-job", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ).json() assert claim["id"] == sub["id"] r = client.post(f"/jobs/{sub['id']}/log", content=b"hello world\n") assert r.status_code == 200 assert r.json()["bytes"] == len(b"hello world\n") r = client.post( f"/jobs/{sub['id']}/complete", json={"exit_code": 0, "final_state": "completed"}, ) assert r.status_code == 200 assert r.json()["state"] == "completed" j = client.get(f"/jobs/{sub['id']}").json() assert j["state"] == "completed" assert j["exit_code"] == 0 def test_signal_poll(client): sub = client.post( "/submit", json={ "cmd": ["sleep", "60"], "cwd": "/tmp", "project": "project-a", "profile": "small", "host_pin": "any", }, ).json() client.post( "/heartbeat", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ) client.post( "/next-job", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ) r = client.get(f"/jobs/{sub['id']}/signal") assert r.status_code == 200 assert r.json()["signal"] is None client.post(f"/jobs/{sub['id']}/cancel") r2 = client.get(f"/jobs/{sub['id']}/signal") assert r2.json()["signal"] == "cancel" def test_list_projects(client): r = client.get("/projects") assert r.status_code == 200 assert r.json()["project-b"]["priority"] == 80 def test_set_project_priority(client): r = client.post("/projects/project-b", json={"priority": 90}) assert r.status_code == 200 assert r.json()["projects"]["project-b"]["priority"] == 90 assert r.json()["project"] == "project-b" def test_nudge_project_priority(client): r = client.post("/projects/project-b/nudge", json={"delta": 5}) assert r.status_code == 200 assert r.json()["projects"]["project-b"]["priority"] == 85 assert r.json()["project"] == "project-b" def test_setting_a_priority_by_a_differently_spelled_name_reprices_the_same_project(client): """Otherwise the write path mints a SECOND entry that folds onto the first, creating exactly the ambiguity the read path then has to warn about and refuse to resolve — so the fix would leave its own mechanism reachable.""" r = client.post("/projects/PROJECT-B", json={"priority": 90}) assert r.status_code == 200 body = r.json() assert body["projects"]["project-b"]["priority"] == 90 assert "PROJECT-B" not in body["projects"], ( f"minted a second colliding entry: {sorted(body['projects'])}" ) # The typed name is absent from the table, so the response has to NAME the # key it wrote or no caller can find it. Asserting only the absence above is # what let the CLI ship indexing by the typed name (KeyError, post-write). assert body["project"] == "project-b" def test_nudging_by_a_differently_spelled_name_moves_the_same_project(client): r = client.post("/projects/Project_B/nudge", json={"delta": 5}) assert r.status_code == 200 body = r.json() assert body["projects"]["project-b"]["priority"] == 85 assert "Project_B" not in body["projects"], ( f"minted a second colliding entry: {sorted(body['projects'])}" ) assert body["project"] == "project-b" def test_an_unmatched_name_still_creates_a_new_project(client): """Positive control: folding must not stop a genuinely new project from being registered under the name its owner chose.""" r = client.post("/projects/a-brand-new-project", json={"priority": 70}) assert r.status_code == 200 assert r.json()["projects"]["a-brand-new-project"]["priority"] == 70 assert r.json()["project"] == "a-brand-new-project" def test_set_project_priority_malformed_payload_is_422_not_500(client): """A missing or non-integer `priority` is a client error: pydantic validation returns 422, not a handler KeyError/ValueError -> 500 (audit LOW).""" assert client.post("/projects/project-b", json={}).status_code == 422 assert client.post("/projects/project-b", json={"priority": "high"}).status_code == 422 assert client.post("/projects/project-b", json={"priority": None}).status_code == 422 def test_nudge_project_priority_malformed_payload_is_422_not_500(client): assert client.post("/projects/project-b/nudge", json={}).status_code == 422 assert client.post("/projects/project-b/nudge", json={"delta": "lots"}).status_code == 422 def test_reload_reloads_projects(client, sample_projects_yaml): sample_projects_yaml.write_text("""projects: project-b: { priority: 99 } _default: { priority: 40 } """) r = client.post("/reload") assert r.status_code == 200 g = client.get("/projects").json() assert g["project-b"]["priority"] == 99 def test_submit_with_requires_persists_and_returns(client): r = client.post( "/submit", json={ "cmd": ["echo", "hi"], "cwd": "/tmp", "project": "project-a", "requires": {"arch": "x86_64", "gpu": True, "needs": ["cuda"]}, }, ) assert r.status_code == 200 job_id = r.json()["id"] got = client.get(f"/jobs/{job_id}").json() assert got["requires"] == { "arch": "x86_64", "os": "any", "gpu": True, "needs": ["cuda"], "idempotent": False, } def test_heartbeat_persists_capabilities(client): r = client.post( "/heartbeat", json={ "host": "rpi4", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 3.5, "idle_cpus": 4, "arch": "arm64", "os": "linux", "gpu": False, "tags": ["python3", "always-on"], "host_aliases": [], }, ) assert r.status_code == 200 r = client.post( "/submit", json={ "cmd": ["uname", "-m"], "cwd": "/tmp", "project": "project-a", "requires": {"arch": "arm64"}, }, ) job_id = r.json()["id"] r = client.post( "/next-job", json={ "host": "rpi4", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 3.5, "idle_cpus": 4, }, ) assert r.status_code == 200 body = r.json() assert body is not None assert body["id"] == job_id def test_next_job_uses_persisted_worker_capabilities(client): client.post( "/heartbeat", json={ "host": "server", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 24, "idle_cpus": 12, "arch": "x86_64", "os": "linux", "gpu": False, "tags": ["R", "python3"], "host_aliases": [], }, ) r = client.post( "/submit", json={ "cmd": ["Rscript", "x.R"], "cwd": "/tmp", "project": "project-b", "requires": {"needs": ["R"]}, }, ) job_id = r.json()["id"] r = client.post( "/next-job", json={ "host": "server", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 24, "idle_cpus": 12, }, ) assert r.json()["id"] == job_id def test_orphan_sweeper_reclaims_after_timeout(client): """A job assigned to a worker whose heartbeat went silent >5min must be re-queued.""" from datetime import UTC, datetime, timedelta from sqlalchemy import update # Worker heartbeats once client.post( "/heartbeat", json={ "host": "ghost", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, "arch": "x86_64", "os": "linux", "gpu": False, "tags": [], "host_aliases": [], }, ) # Submit + claim r = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", }, ) job_id = r.json()["id"] claim = client.post( "/next-job", json={ "host": "ghost", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, }, ) assert claim.json()["id"] == job_id assert claim.json()["state"] == "assigned" # Fast-forward: manually backdate the worker's heartbeat >5min from jobd.db import Worker engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Worker) .where(Worker.host == "ghost") .values(last_heartbeat=datetime.now(UTC) - timedelta(minutes=6)) ) # Trigger sweeper manually (exposed via private test hook) client.app.state.sweep_once() got = client.get(f"/jobs/{job_id}").json() assert got["state"] == "queued" assert got["worker"] is None def test_sweeper_reclaim_honors_pending_cancel(client): """A2 (audit 2026-07-05, supersedes the M1 contract this test used to pin): a job assigned to a dead worker with a pending user CANCEL must land CANCELLED — not be requeued with the signal silently cleared, which re-ran a job the user was already told was being cancelled. (M1's clear-on-requeue still holds for non-cancel signals: see test_sweeper_reclaim_still_clears_preempt_signal.)""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Worker client.post( "/heartbeat", json={ "host": "ghost", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, "arch": "x86_64", "os": "linux", "gpu": False, "tags": [], "host_aliases": [], }, ) job_id = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"} ).json()["id"] client.post( "/next-job", json={ "host": "ghost", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, }, ) # A cancel while ASSIGNED stamps a pending 'cancel' signal (worker not seen yet). client.post(f"/jobs/{job_id}/cancel") assert client.get(f"/jobs/{job_id}/signal").json()["signal"] == "cancel" # Worker goes silent past the reclaim threshold; sweep re-queues the job. engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Worker) .where(Worker.host == "ghost") .values(last_heartbeat=datetime.now(UTC) - timedelta(minutes=6)) ) client.app.state.sweep_once() got = client.get(f"/jobs/{job_id}").json() assert got["state"] == "cancelled", got assert got["finished_at"] is not None assert client.get(f"/jobs/{job_id}/signal").json()["signal"] is None def test_sweeper_reclaim_still_clears_preempt_signal(client): """M1: a pending non-cancel signal (preempt) on a dead worker's ASSIGNED job is cleared by the requeue so the fresh worker it re-dispatches to doesn't read it on the first /signal poll and kill the re-run.""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Job, Worker client.post( "/heartbeat", json={ "host": "ghost", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, "arch": "x86_64", "os": "linux", "gpu": False, "tags": [], "host_aliases": [], }, ) job_id = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"} ).json()["id"] client.post( "/next-job", json={ "host": "ghost", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, }, ) engine = client.app.state.engine with engine.begin() as conn: conn.execute(update(Job).where(Job.id == job_id).values(signal="preempt")) conn.execute( update(Worker) .where(Worker.host == "ghost") .values(last_heartbeat=datetime.now(UTC) - timedelta(minutes=6)) ) client.app.state.sweep_once() got = client.get(f"/jobs/{job_id}").json() assert got["state"] == "queued", got assert client.get(f"/jobs/{job_id}/signal").json()["signal"] is None def test_sweeper_reclaim_does_not_clobber_concurrent_complete(client, monkeypatch): """F1 (review finding): the sweeper's ASSIGNED reclaim is a CAS on WHERE state=ASSIGNED, so a /complete that a briefly-revived worker commits between the sweep's SELECT and its write is NOT clobbered back to QUEUED (which would lose the result and re-dispatch → double execution). The race is injected deterministically: right as the reclaim's teardown write (_requeue_or_honor_cancel, whose WHERE carries the same state=ASSIGNED guard) runs, we flip the row to COMPLETED (as a concurrent /complete would), then delegate to the real guard — which must match 0 rows and leave it completed. """ from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.broker import sweeper as sweeper_mod from jobd.db import Job, Worker from jobd.models import JobState client.post( "/heartbeat", json={ "host": "ghost", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, "arch": "x86_64", "os": "linux", "gpu": False, "tags": [], "host_aliases": [], }, ) job_id = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"} ).json()["id"] client.post( "/next-job", json={ "host": "ghost", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, }, ) engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Worker) .where(Worker.host == "ghost") .values(last_heartbeat=datetime.now(UTC) - timedelta(minutes=10)) ) # The sweeper resolves the teardown helper from its own module # (jobd.broker.sweeper), not jobd.app, so patch it there. real_teardown = sweeper_mod._requeue_or_honor_cancel def racing_teardown(session, jid, expected, *, now, **requeue_values): # Fire once, on the ASSIGNED reclaim write: simulate a /complete # landing in the same window by flipping the row COMPLETED first. if expected == (JobState.ASSIGNED,): monkeypatch.setattr(sweeper_mod, "_requeue_or_honor_cancel", real_teardown) session.execute( update(Job).where(Job.id == jid).values(state=JobState.COMPLETED.value, exit_code=0) ) return real_teardown(session, jid, expected, now=now, **requeue_values) monkeypatch.setattr(sweeper_mod, "_requeue_or_honor_cancel", racing_teardown) client.app.state.sweep_once() # The concurrent completion must survive — not be overwritten to queued. assert client.get(f"/jobs/{job_id}").json()["state"] == "completed" def _start_job_on_worker(client, host, job_json): """Helper: heartbeat a worker, submit a job, claim it, and POST /started so the job reaches RUNNING. Returns the job id.""" client.post( "/heartbeat", json={ "host": host, "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, "arch": "x86_64", "os": "linux", "gpu": False, "tags": [], "host_aliases": [], }, ) r = client.post("/submit", json=job_json) assert r.status_code == 200, r.text job_id = r.json()["id"] claim = client.post( "/next-job", json={ "host": host, "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, }, ) assert claim.json()["id"] == job_id started = client.post(f"/jobs/{job_id}/started") assert started.json()["state"] == "running", started.text return job_id def test_running_job_on_dead_worker_orphaned_and_cascades(client): """CRIT-2: a non-idempotent RUNNING job whose worker goes silent >5min is transitioned to ORPHANED (not left RUNNING forever), and its queued dependents are cascade-cancelled so /wait can return.""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Worker parent_id = _start_job_on_worker( client, "ghost", {"cmd": ["true"], "cwd": "/tmp", "project": "project-a"} ) # A child depending on the parent (default policy: needs parent COMPLETED). child = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "depends_on": [parent_id], }, ) child_id = child.json()["id"] engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Worker) .where(Worker.host == "ghost") .values(last_heartbeat=datetime.now(UTC) - timedelta(minutes=6)) ) client.app.state.sweep_once() parent = client.get(f"/jobs/{parent_id}").json() assert parent["state"] == "orphaned", parent assert parent["termination_reason"] == "worker_died" assert parent["finished_at"] is not None child_got = client.get(f"/jobs/{child_id}").json() assert child_got["state"] == "cancelled", child_got def test_running_idempotent_job_on_dead_worker_requeued(client): """CRIT-2: an idempotent RUNNING job is safe to re-run, so a dead worker (>90s for idempotent) requeues it rather than orphaning it.""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Worker job_id = _start_job_on_worker( client, "ghost", { "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "requires": {"idempotent": True}, }, ) engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Worker) .where(Worker.host == "ghost") .values(last_heartbeat=datetime.now(UTC) - timedelta(seconds=120)) ) client.app.state.sweep_once() got = client.get(f"/jobs/{job_id}").json() assert got["state"] == "queued", got assert got["worker"] is None assert got["started_at"] is None def test_running_job_past_wall_clock_orphaned_despite_live_worker(client): """RUNNING reaper wall-clock backstop: a job whose worker is still heartbeating but whose max_wall_s + grace has elapsed is orphaned with termination_reason='wall_clock_exceeded' — not left RUNNING forever. Mechanism: worker-side enforcement (SIGTERM at max_wall_s in job_worker.poll_signals) only lives in the worker's per-job monitor. If the worker crashes mid-run and restarts within DEAD_WORKER_SECONDS, the new worker has no memory of the job, never enforces its wall clock, and never posts /complete; the broker sees a live worker and leaves the job RUNNING forever. Regression for stranded job 1364 (desktop host-power crash 2026-06-03).""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Job, Worker job_id = _start_job_on_worker( client, "alive", {"cmd": ["true"], "cwd": "/tmp", "project": "project-a", "max_wall_s": 900}, ) # A queued dependent so we also prove the wall-clock orphan cascades. child = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "depends_on": [job_id], }, ) child_id = child.json()["id"] engine = client.app.state.engine with engine.begin() as conn: # Worker is HEALTHY (fresh heartbeat) — the dead-worker reaper would NOT # fire. Only the wall-clock backstop should. conn.execute( update(Worker).where(Worker.host == "alive").values(last_heartbeat=datetime.now(UTC)) ) # Job started far past max_wall_s + grace (naive UTC, as stored). conn.execute( update(Job) .where(Job.id == job_id) .values(started_at=datetime.now(UTC).replace(tzinfo=None) - timedelta(hours=4)) ) client.app.state.sweep_once() got = client.get(f"/jobs/{job_id}").json() assert got["state"] == "orphaned", got assert got["termination_reason"] == "wall_clock_exceeded", got assert got["finished_at"] is not None child_got = client.get(f"/jobs/{child_id}").json() assert child_got["state"] == "cancelled", child_got def test_running_job_within_wall_clock_left_running(client): """Anti-vacuity for the wall-clock backstop: a RUNNING job on a live worker that is still well within max_wall_s is left RUNNING (the backstop must not reap healthy in-progress jobs).""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Job, Worker job_id = _start_job_on_worker( client, "alive", {"cmd": ["true"], "cwd": "/tmp", "project": "project-a", "max_wall_s": 900}, ) engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Worker).where(Worker.host == "alive").values(last_heartbeat=datetime.now(UTC)) ) conn.execute( update(Job) .where(Job.id == job_id) .values(started_at=datetime.now(UTC).replace(tzinfo=None) - timedelta(seconds=60)) ) client.app.state.sweep_once() got = client.get(f"/jobs/{job_id}").json() assert got["state"] == "running", got def test_submitted_env_redacted_on_reads_real_on_dispatch(client): """Submitted env is delivered to the claiming worker via /next-job with its REAL values, but every observability read surface (the /submit response and GET /jobs/{id}) masks the values to a placeholder — keys stay visible so an operator can see WHICH vars are set without the plaintext leaking to any token-holder or into MCP/agent context (audit 2026-07-01, LOW-Sec --env).""" client.post( "/heartbeat", json={ "host": "w1", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, "arch": "x86_64", "os": "linux", "gpu": False, "tags": [], "host_aliases": [], }, ) r = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "env": {"FOO": "bar", "BAZ": "qux"}, }, ) assert r.status_code == 200, r.text # /submit response: keys present, values masked (no plaintext echoed back). assert r.json()["env"] == {"FOO": "***", "BAZ": "***"} job_id = r.json()["id"] # GET /jobs/{id} and GET /jobs: same masking. assert client.get(f"/jobs/{job_id}").json()["env"] == {"FOO": "***", "BAZ": "***"} listed = next(j for j in client.get("/jobs").json() if j["id"] == job_id) assert listed["env"] == {"FOO": "***", "BAZ": "***"} # /next-job: the worker claim gets the REAL values so the job actually runs. claim = client.post( "/next-job", json={ "host": "w1", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, }, ) assert claim.json()["env"] == {"FOO": "bar", "BAZ": "qux"} def test_unmatcheable_job_gets_warning(client): """Job requiring arm64 with no arm64 worker advertising gets a warning. With #43 submit_preflight, this fires at submit time instead of 60s later in the sweeper. The sweeper-path remains the fallback when capabilities change AFTER submit (worker goes offline) — covered by other tests. """ client.post( "/heartbeat", json={ "host": "desktop", "free_vram_gb": 10, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, "arch": "x86_64", "os": "linux", "gpu": True, "tags": [], "host_aliases": [], }, ) r = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "requires": {"arch": "arm64"}, }, ) body = r.json() assert body["warning"] is not None assert "unsatisfiable" in body["warning"] assert "arm64" in body["warning"] def test_no_requires_job_never_gets_warning_on_empty_fleet(client): """A job with no requires block is matcheable by definition — empty fleet != mismatch.""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Job # No heartbeats — empty fleet r = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"}, ) job_id = r.json()["id"] engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Job) .where(Job.id == job_id) .values(submitted_at=datetime.now(UTC) - timedelta(seconds=90)) ) client.app.state.sweep_once() assert client.get(f"/jobs/{job_id}").json()["warning"] is None def test_warning_clears_when_matching_worker_appears(client): """Warning set on unmatcheable job clears once a capable worker heartbeats.""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Job client.post( "/heartbeat", json={ "host": "x86box", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, "arch": "x86_64", "os": "linux", "gpu": False, "tags": [], "host_aliases": [], }, ) r = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "requires": {"arch": "arm64"}, }, ) job_id = r.json()["id"] engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Job) .where(Job.id == job_id) .values(submitted_at=datetime.now(UTC) - timedelta(seconds=90)) ) client.app.state.sweep_once() assert client.get(f"/jobs/{job_id}").json()["warning"] is not None # arm64 worker joins the fleet client.post( "/heartbeat", json={ "host": "pi4", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 3, "idle_cpus": 4, "arch": "arm64", "os": "linux", "gpu": False, "tags": [], "host_aliases": [], }, ) client.app.state.sweep_once() assert client.get(f"/jobs/{job_id}").json()["warning"] is None def test_orphan_sweeper_idempotent_reclaims_at_90s(client): """A job with requires.idempotent=true reclaims after 90s, not 5min.""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Worker client.post( "/heartbeat", json={ "host": "ghost-idem", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, "arch": "x86_64", "os": "linux", "gpu": False, "tags": [], "host_aliases": [], }, ) r = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "requires": {"idempotent": True}, }, ) job_id = r.json()["id"] claim = client.post( "/next-job", json={ "host": "ghost-idem", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, }, ) assert claim.json()["id"] == job_id engine = client.app.state.engine # 2 minutes silent: past idempotent cutoff (90s) but under the 5min default with engine.begin() as conn: conn.execute( update(Worker) .where(Worker.host == "ghost-idem") .values(last_heartbeat=datetime.now(UTC) - timedelta(seconds=120)) ) client.app.state.sweep_once() got = client.get(f"/jobs/{job_id}").json() assert got["state"] == "queued" assert got["worker"] is None # ---------- depends_on job chaining ---------- def _submit(client, project="project-a", cmd=None, depends_on=None, any_exit=False): body = { "cmd": cmd or ["true"], "cwd": "/tmp", "project": project, } if depends_on is not None: body["depends_on"] = depends_on if any_exit: body["depends_on_any_exit"] = True r = client.post("/submit", json=body) return r def _heartbeat(client, host="w1"): client.post( "/heartbeat", json={ "host": host, "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, "arch": "x86_64", "os": "linux", "gpu": False, "tags": [], "host_aliases": [], }, ) def _next_job(client, host="w1"): return client.post( "/next-job", json={ "host": host, "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, }, ) def test_submit_unknown_parent_rejected(client): r = _submit(client, depends_on=[9999]) assert r.status_code == 400 assert "parent" in r.text.lower() or "depends_on" in r.text.lower() def test_child_not_dispatched_while_parent_queued(client): parent = _submit(client).json() child = _submit(client, depends_on=[parent["id"]]).json() _heartbeat(client) got = _next_job(client).json() assert got["id"] == parent["id"] got2 = _next_job(client).json() assert got2 is None or got2 == {} row = client.get(f"/jobs/{child['id']}").json() assert row["state"] == "queued" def test_child_dispatchable_after_parent_completes(client): parent = _submit(client).json() child = _submit(client, depends_on=[parent["id"]]).json() _heartbeat(client) _next_job(client) client.post(f"/jobs/{parent['id']}/complete", json={"exit_code": 0}) claim = _next_job(client).json() assert claim["id"] == child["id"] def test_parent_failure_cancels_child(client): parent = _submit(client).json() child = _submit(client, depends_on=[parent["id"]]).json() _heartbeat(client) _next_job(client) client.post( f"/jobs/{parent['id']}/complete", json={"exit_code": 1, "final_state": "failed"}, ) row = client.get(f"/jobs/{child['id']}").json() assert row["state"] == "cancelled" assert row.get("warning") def test_parent_cancelled_queued_cancels_child(client): parent = _submit(client).json() child = _submit(client, depends_on=[parent["id"]]).json() client.post(f"/jobs/{parent['id']}/cancel") row = client.get(f"/jobs/{child['id']}").json() assert row["state"] == "cancelled" def test_parent_failure_cascade_is_transitive(client): """A<-B<-C: when A fails via /complete, the cascade must cancel B AND transitively C (audit 2026-07-01 H2 — the old single-level cascade cancelled only the direct child B and stranded C in QUEUED forever).""" a = _submit(client).json() b = _submit(client, depends_on=[a["id"]]).json() c = _submit(client, depends_on=[b["id"]]).json() _heartbeat(client) _next_job(client) client.post(f"/jobs/{a['id']}/complete", json={"exit_code": 1, "final_state": "failed"}) assert client.get(f"/jobs/{b['id']}").json()["state"] == "cancelled" assert client.get(f"/jobs/{c['id']}").json()["state"] == "cancelled" def test_submit_rejects_default_dep_on_terminal_failed_parent(client): """audit 2026-07-01 H2: a default-policy dep on an ALREADY failed-side terminal parent can never be satisfied (the parent won't reach COMPLETED) and no future transition fires the cascade, so the child would strand in QUEUED forever. Reject at submit (400) instead.""" parent = _submit(client).json() _heartbeat(client) _next_job(client) client.post(f"/jobs/{parent['id']}/complete", json={"exit_code": 1, "final_state": "failed"}) assert client.get(f"/jobs/{parent['id']}").json()["state"] == "failed" r = _submit(client, depends_on=[parent["id"]]) assert r.status_code == 400, r.text assert "failed-side terminal" in r.text def test_submit_allows_any_exit_dep_on_terminal_parent(client): """The submit-time reject applies only to default-policy deps: an any-exit child is satisfied by any terminal parent, so it must still be accepted and become dispatchable immediately.""" parent = _submit(client).json() _heartbeat(client) _next_job(client) client.post(f"/jobs/{parent['id']}/complete", json={"exit_code": 1, "final_state": "failed"}) r = _submit(client, depends_on=[parent["id"]], any_exit=True) assert r.status_code == 200, r.text child = r.json() claim = _next_job(client).json() assert claim["id"] == child["id"] def test_submit_allows_default_dep_on_completed_parent(client): """A default-policy dep on an already-COMPLETED parent is satisfied, not rejected — the reject targets only failed-side terminals.""" parent = _submit(client).json() _heartbeat(client) _next_job(client) client.post(f"/jobs/{parent['id']}/complete", json={"exit_code": 0, "final_state": "completed"}) r = _submit(client, depends_on=[parent["id"]]) assert r.status_code == 200, r.text def test_depends_on_any_exit_allows_failed_parent(client): parent = _submit(client).json() child = _submit(client, depends_on=[parent["id"]], any_exit=True).json() _heartbeat(client) _next_job(client) client.post( f"/jobs/{parent['id']}/complete", json={"exit_code": 2, "final_state": "failed"}, ) row = client.get(f"/jobs/{child['id']}").json() assert row["state"] == "queued" claim = _next_job(client).json() assert claim["id"] == child["id"] def test_parent_preempted_cancels_child(client): """#35: PREEMPTED is treated as a failure-side terminal for the depends_on cascade — queued children without any_exit get cancelled, same as parent FAILED/CANCELLED. Aligns with the shipped path A semantics ("preempt = job dies, project rerun later"): a downstream job whose input never materialized cannot run. If a future user wants the chain to wait through preempt instead, they set `depends_on_any_exit=True` (covered by the next test).""" parent = _submit(client).json() child = _submit(client, depends_on=[parent["id"]]).json() _heartbeat(client) _next_job(client) client.post(f"/jobs/{parent['id']}/started") client.post( f"/jobs/{parent['id']}/complete", json={"exit_code": -15, "final_state": "preempted"}, ) parent_after = client.get(f"/jobs/{parent['id']}").json() assert parent_after["state"] == "preempted" row = client.get(f"/jobs/{child['id']}").json() assert row["state"] == "cancelled" assert "parent_failed" in (row.get("warning") or "") assert str(parent["id"]) in (row.get("warning") or "") def test_parent_preempted_any_exit_keeps_child_queued(client): """#35: depends_on_any_exit=True opts a child out of the preempt-cascade; the child stays QUEUED and becomes runnable once the parent reaches any terminal. Use this when you'd rather have the pipeline limp forward on partial output than re-submit the chain.""" parent = _submit(client).json() child = _submit(client, depends_on=[parent["id"]], any_exit=True).json() _heartbeat(client) _next_job(client) client.post(f"/jobs/{parent['id']}/started") client.post( f"/jobs/{parent['id']}/complete", json={"exit_code": -15, "final_state": "preempted"}, ) row = client.get(f"/jobs/{child['id']}").json() assert row["state"] == "queued" claim = _next_job(client).json() assert claim["id"] == child["id"] def test_output_endpoint_returns_tail(client): """Streaming logs via /log should be retrievable via /output.""" job = _submit(client).json() # Simulate worker streaming chunks client.post(f"/jobs/{job['id']}/log", content=b"line 1\n") client.post(f"/jobs/{job['id']}/log", content=b"line 2\n") client.post(f"/jobs/{job['id']}/log", content=b"line 3\n") r = client.get(f"/jobs/{job['id']}/output") assert r.status_code == 200 body = r.json() assert body["tail"] == "line 1\nline 2\nline 3\n" assert body["size_bytes"] == 21 assert body["truncated"] is False def test_output_endpoint_truncates_to_tail_window(client): """tail=5 on a 21-byte file returns the last 5 bytes + truncated=True.""" job = _submit(client).json() client.post(f"/jobs/{job['id']}/log", content=b"line 1\nline 2\nline 3\n") r = client.get(f"/jobs/{job['id']}/output", params={"tail": 5}) assert r.status_code == 200 body = r.json() assert body["tail"] == "ine 3\n"[-5:] assert body["size_bytes"] == 21 assert body["returned_bytes"] == 5 assert body["truncated"] is True def test_output_endpoint_empty_when_no_log(client): """Job exists but worker never streamed → size_bytes=0, empty tail.""" job = _submit(client).json() r = client.get(f"/jobs/{job['id']}/output") assert r.status_code == 200 body = r.json() assert body["size_bytes"] == 0 assert body["tail"] == "" def test_output_endpoint_404_on_unknown_job(client): r = client.get("/jobs/99999/output") assert r.status_code == 404 def test_complete_auto_derives_failed_from_nonzero_rc(client): """If the worker omits final_state, nonzero rc → failed, zero → completed. Matters for the depends_on cascade, which keys off terminal state.""" job_ok = _submit(client).json() job_bad = _submit(client).json() _heartbeat(client) _next_job(client) _next_job(client) # worker omits final_state client.post(f"/jobs/{job_ok['id']}/complete", json={"exit_code": 0}) client.post(f"/jobs/{job_bad['id']}/complete", json={"exit_code": 1}) assert client.get(f"/jobs/{job_ok['id']}").json()["state"] == "completed" assert client.get(f"/jobs/{job_bad['id']}").json()["state"] == "failed" def test_complete_clears_pending_signal(client): """After worker posts /complete, job.signal must be None so a future reclaim/retry doesn't see a stale cancel signal.""" job = _submit(client).json() _heartbeat(client) _next_job(client) client.post(f"/jobs/{job['id']}/cancel") assert client.get(f"/jobs/{job['id']}/signal").json()["signal"] == "cancel" client.post( f"/jobs/{job['id']}/complete", json={"exit_code": -15, "final_state": "cancelled"}, ) assert client.get(f"/jobs/{job['id']}/signal").json()["signal"] is None def test_job_info_reports_pending_signal(client): """GET /jobs/{id} carries job.signal through its lifecycle: None while running, 'cancel' once a cancel is queued, None again after /complete. JobInfo used to omit the column, so every status read reported no pending signal even while the worker was about to SIGTERM the job.""" job = _submit(client).json() _heartbeat(client) _next_job(client) client.post(f"/jobs/{job['id']}/started") before = client.get(f"/jobs/{job['id']}").json() assert "signal" in before and before["signal"] is None client.post(f"/jobs/{job['id']}/cancel") assert client.get(f"/jobs/{job['id']}").json()["signal"] == "cancel" listed = {j["id"]: j for j in client.get("/jobs").json()} assert listed[job["id"]]["signal"] == "cancel" client.post( f"/jobs/{job['id']}/complete", json={"exit_code": -15, "final_state": "cancelled"}, ) assert client.get(f"/jobs/{job['id']}").json()["signal"] is None def test_job_info_reports_pending_preempt(client): """The same field carries 'preempt' for a preempted preemptible job.""" _heartbeat(client) job = client.post( "/submit", json={"cmd": ["sleep", "999"], "cwd": "/tmp", "project": "project-a", "preemptible": True}, ).json() _next_job(client) client.post(f"/jobs/{job['id']}/started") assert client.post(f"/jobs/{job['id']}/preempt").status_code == 200 assert client.get(f"/jobs/{job['id']}").json()["signal"] == "preempt" def test_complete_is_idempotent_on_terminal_job(client): """P3.2: a second /complete on an already-terminal job is a no-op — the recorded outcome is NOT overwritten (worker retries / late posts can't clobber a completed/cancelled job).""" job = _submit(client).json() _heartbeat(client) _next_job(client) client.post(f"/jobs/{job['id']}/started") r1 = client.post( f"/jobs/{job['id']}/complete", json={"exit_code": 0, "final_state": "completed"}, ) assert r1.json()["state"] == "completed" # A second, contradictory /complete must not change anything. r2 = client.post( f"/jobs/{job['id']}/complete", json={"exit_code": 1, "final_state": "failed"}, ) assert r2.status_code == 200 assert r2.json()["state"] == "completed" assert r2.json()["exit_code"] == 0 got = client.get(f"/jobs/{job['id']}").json() assert got["state"] == "completed" and got["exit_code"] == 0 def test_complete_rejects_invalid_final_state(client): """P3.2: a final_state that isn't a valid terminal state is rejected (400) rather than stored as a non-enum string that breaks later reads.""" job = _submit(client).json() _heartbeat(client) _next_job(client) client.post(f"/jobs/{job['id']}/started") bogus = client.post( f"/jobs/{job['id']}/complete", json={"exit_code": 0, "final_state": "definitely-not-a-state"}, ) assert bogus.status_code == 400 # A real JobState that isn't terminal is also rejected. nonterminal = client.post( f"/jobs/{job['id']}/complete", json={"exit_code": 0, "final_state": "running"}, ) assert nonterminal.status_code == 400 # The job is untouched — still running. assert client.get(f"/jobs/{job['id']}").json()["state"] == "running" def test_complete_rejects_stale_worker(client): """M2: after a partition reclaim + re-dispatch, the ORIGINAL worker (still running the workload) must not be able to terminal-ize the job that now belongs to a different worker. A /complete whose X-Jobd-Worker header doesn't match job.worker is refused (409); the job stays running.""" job = _submit(client).json() _heartbeat(client, host="w1") _next_job(client, host="w1") # job.worker = w1 client.post(f"/jobs/{job['id']}/started", headers={"X-Jobd-Worker": "w1"}) stale = client.post( f"/jobs/{job['id']}/complete", json={"exit_code": 0, "final_state": "completed"}, headers={"X-Jobd-Worker": "w2"}, # a different, stale worker ) assert stale.status_code == 409, stale.text assert client.get(f"/jobs/{job['id']}").json()["state"] == "running" # The owning worker completes it normally. ok = client.post( f"/jobs/{job['id']}/complete", json={"exit_code": 0, "final_state": "completed"}, headers={"X-Jobd-Worker": "w1"}, ) assert ok.status_code == 200, ok.text assert client.get(f"/jobs/{job['id']}").json()["state"] == "completed" def test_complete_without_worker_header_still_accepted(client): """Backward compatibility: a pre-header worker sends no X-Jobd-Worker, so the stale-worker check is skipped and /complete behaves as before.""" job = _submit(client).json() _heartbeat(client, host="w1") _next_job(client, host="w1") r = client.post( f"/jobs/{job['id']}/complete", json={"exit_code": 0, "final_state": "completed"} ) assert r.status_code == 200, r.text assert client.get(f"/jobs/{job['id']}").json()["state"] == "completed" def test_log_rejects_stale_worker(client): """M2: a stale worker's /log chunk must not interleave into the log file of the run that now owns the job.""" job = _submit(client).json() _heartbeat(client, host="w1") _next_job(client, host="w1") stale = client.post( f"/jobs/{job['id']}/log", content=b"stale bytes", headers={"X-Jobd-Worker": "w2"} ) assert stale.status_code == 409, stale.text ok = client.post( f"/jobs/{job['id']}/log", content=b"real bytes", headers={"X-Jobd-Worker": "w1"} ) assert ok.status_code == 200, ok.text out = client.get(f"/jobs/{job['id']}/output").json() assert out["tail"] == "real bytes" def test_job_info_exposes_session_id(client): r = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "session_id": "sess-abc-123", }, ) assert r.status_code == 200 assert r.json()["session_id"] == "sess-abc-123" got = client.get(f"/jobs/{r.json()['id']}").json() assert got["session_id"] == "sess-abc-123" def _hb_wsl(client, host, aliases): client.post( "/heartbeat", json={ "host": host, "host_aliases": aliases, "free_vram_gb": 0.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 16.0, "idle_cpus": 4, "arch": "x86_64", "os": "linux", "gpu": False, "tags": ["wsl"], }, ) def _submit_mnt_c(client, host_pin): return client.post( "/submit", json={ "cmd": ["true"], "cwd": "/mnt/c/Users/dev/some/project", "project": "project-a", "host_pin": host_pin, }, ) def test_submit_rejects_mnt_c_cwd_without_a_single_host(client): """/mnt/c/ differs per WSL host → a pin that can route to more than one host (unpinned, or a pool alias) is a 400, not queued.""" _hb_wsl(client, "laptop", ["any", "any-gpu"]) _hb_wsl(client, "desktop", ["any", "any-gpu"]) for pin in ("any", "any-gpu"): r = _submit_mnt_c(client, pin) assert r.status_code == 400, pin assert "/mnt/c/" in r.text # positive control, same broker state: one named host is accepted assert _submit_mnt_c(client, "desktop").status_code == 200 def test_submit_allows_mnt_c_cwd_pinned_to_any_single_host(client): # No host name is special: the old rule accepted only laptop/MSI/any-laptop. _hb_wsl(client, "laptop", ["any"]) _hb_wsl(client, "desktop", ["any"]) for pin in ("laptop", "desktop"): assert _submit_mnt_c(client, pin).status_code == 200, pin def test_cancel_running_sets_signal_not_state(client): """POST /cancel on a claimed job must set job.signal='cancel' and leave state=assigned — the worker completes the transition after SIGTERMing the child.""" job = _submit(client).json() _heartbeat(client) claim = _next_job(client).json() assert claim["id"] == job["id"] r = client.post(f"/jobs/{job['id']}/cancel") assert r.status_code == 200 row = r.json() # state unchanged — worker must finish the job before it flips assert row["state"] in ("assigned", "running") sig = client.get(f"/jobs/{job['id']}/signal").json() assert sig["signal"] == "cancel" # worker reports the terminal state once the child has exited client.post( f"/jobs/{job['id']}/complete", json={"exit_code": -15, "final_state": "cancelled"}, ) row = client.get(f"/jobs/{job['id']}").json() assert row["state"] == "cancelled" assert row["exit_code"] == -15 def test_cancel_queued_loses_race_to_claim_falls_through_to_signal(client, monkeypatch): """H3 cancel-vs-claim TOCTOU. If a worker's atomic queued->assigned claim lands between cancel_job's read (which saw QUEUED) and its guarded write, the guarded UPDATE matches 0 rows. cancel must NOT clobber the claim to CANCELLED (which would strand the worker running a job the user was told was cancelled, with no kill signal). It must fall through to the signal path: state stays assigned, signal='cancel'. The race is injected deterministically: the first _cas_state call (the queued->cancelled CAS) flips the row to ASSIGNED via a real update and reports 0 rows, exactly as a concurrent claim committing mid-request would. """ from jobd.broker.routes import jobs as jobs_routes_mod from jobd.models import JobState job = _submit(client).json() real_cas = jobs_routes_mod._cas_state calls = {"n": 0} def racing_cas(session, job_id, expected, **values): calls["n"] += 1 if calls["n"] == 1 and values.get("state") == JobState.CANCELLED: # Simulate the matcher's claim winning: queued -> assigned, and # report that our guarded cancel matched nothing. real_cas(session, job_id, (JobState.QUEUED,), state=JobState.ASSIGNED, worker="w1") return 0 return real_cas(session, job_id, expected, **values) monkeypatch.setattr(jobs_routes_mod, "_cas_state", racing_cas) r = client.post(f"/jobs/{job['id']}/cancel") assert r.status_code == 200 # Must NOT have been clobbered to cancelled. assert r.json()["state"] in ("assigned", "running"), r.json() # The cancel still lands — as a signal the worker honors. sig = client.get(f"/jobs/{job['id']}/signal").json() assert sig["signal"] == "cancel" def test_started_transitions_assigned_to_running(client): """Worker POST /started flips assigned -> running so callers can observe the live in-flight phase. Without this, jobs spend their entire run in ASSIGNED and the MCP cancel signal_sent synthesis can't distinguish queued-cancel from SIGTERM-cancel.""" job = _submit(client).json() _heartbeat(client) _next_job(client) assert client.get(f"/jobs/{job['id']}").json()["state"] == "assigned" r = client.post(f"/jobs/{job['id']}/started") assert r.status_code == 200 assert r.json()["state"] == "running" def test_started_is_idempotent_and_no_op_on_terminal(client): """Two /started POSTs leave the job in running. /started after /complete is a no-op (state stays terminal). The worker may retry /started on a transient network error; the broker must not re-open a finished job.""" job = _submit(client).json() _heartbeat(client) _next_job(client) client.post(f"/jobs/{job['id']}/started") client.post(f"/jobs/{job['id']}/started") assert client.get(f"/jobs/{job['id']}").json()["state"] == "running" client.post(f"/jobs/{job['id']}/complete", json={"exit_code": 0}) client.post(f"/jobs/{job['id']}/started") after = client.get(f"/jobs/{job['id']}").json() assert after["state"] == "completed" def test_started_404_on_unknown_job(client): r = client.post("/jobs/9999999/started") assert r.status_code == 404 def test_submit_fast_path_profile_stamps_flag(client): """Profiles declaring fast_path: true must propagate to the job so the worker can skip heavy-run wrapping. Without this, the flag lives only in profiles.yaml and nobody reads it.""" r = client.post( "/submit", json={ "cmd": ["echo", "hi"], "cwd": "/tmp", "project": "project-a", "profile": "small", }, ) assert r.status_code == 200 assert r.json()["fast_path"] is True def test_submit_non_fast_path_profile_leaves_flag_false(client): r = client.post( "/submit", json={ "cmd": ["bash", "train.sh"], "cwd": "/tmp", "project": "project-a", "profile": "gpu-heavy", }, ) assert r.status_code == 200 assert r.json()["fast_path"] is False def test_submit_no_profile_defaults_fast_path_false(client): r = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"}, ) assert r.json()["fast_path"] is False def test_submit_fast_path_request_field_honored_without_profile(client): """fast_path=true in the submit body must propagate to the job even when no profile is named. Without this, every fast-path caller is forced to define a profile just to flip the flag — which is what jobd-mcp ran into.""" r = client.post( "/submit", json={ "cmd": ["echo", "x"], "cwd": "/tmp", "project": "project-a", "fast_path": True, }, ) assert r.status_code == 200 assert r.json()["fast_path"] is True def test_submit_fast_path_request_field_overrides_profile_false(client): """An explicit fast_path=true on the request beats a profile that says fast_path=false. Submit-time signal is more specific than the profile.""" r = client.post( "/submit", json={ "cmd": ["bash", "train.sh"], "cwd": "/tmp", "project": "project-a", "profile": "gpu-heavy", "fast_path": True, }, ) assert r.status_code == 200 assert r.json()["fast_path"] is True def test_submit_fast_path_request_field_can_disable_profile_true(client): """The reverse: fast_path=false at submit beats profile.fast_path=true.""" r = client.post( "/submit", json={ "cmd": ["echo", "hi"], "cwd": "/tmp", "project": "project-a", "profile": "small", "fast_path": False, }, ) assert r.status_code == 200 assert r.json()["fast_path"] is False def test_submit_max_wall_and_idle_timeout_round_trip(client): """The two watchdog fields persist through submit and come back on read. Worker-side enforcement is covered separately in test_worker.""" r = client.post( "/submit", json={ "cmd": ["echo", "hi"], "cwd": "/tmp", "project": "project-a", "max_wall_s": 3600, "idle_timeout_s": 300, }, ) assert r.status_code == 200 body = r.json() assert body["max_wall_s"] == 3600 assert body["idle_timeout_s"] == 300 assert body["termination_reason"] is None def test_submit_rejects_negative_timeouts(client): """Pydantic ge=1 constraint must reject zero/negative — there's no sensible meaning to 'timeout in 0 seconds'.""" r = client.post( "/submit", json={ "cmd": ["echo"], "cwd": "/tmp", "project": "project-a", "max_wall_s": 0, }, ) assert r.status_code == 422 def test_submit_omitted_timeouts_default_to_null(client): """When neither timeout is specified, both come back null. Worker reads null as 'no per-job override; fall through to env default if any'.""" r = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"}, ) body = r.json() assert body["max_wall_s"] is None assert body["idle_timeout_s"] is None def test_complete_persists_termination_reason(client): """Worker can pass termination_reason in the /complete payload; it persists and surfaces in JobInfo. Watchdog kills set this so callers can distinguish 'failed because exit 1' from 'failed because hung'.""" r = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"}, ) job_id = r.json()["id"] # Need to dispatch first so it's in a transitional state, then complete. # Easier: hand-flip via an existing path — POST /complete directly works # for any job state in this codebase (see complete_job). r2 = client.post( f"/jobs/{job_id}/complete", json={"exit_code": -15, "final_state": "failed", "termination_reason": "idle_timeout"}, ) assert r2.status_code == 200 body = r2.json() assert body["termination_reason"] == "idle_timeout" assert body["state"] == "failed" def test_submit_fast_path_omitted_falls_through_to_profile(client): """When fast_path isn't specified at submit, the profile's value still wins. Backward-compat with existing callers that rely on profile-derived flag.""" r = client.post( "/submit", json={ "cmd": ["echo", "hi"], "cwd": "/tmp", "project": "project-a", "profile": "small", }, ) assert r.status_code == 200 assert r.json()["fast_path"] is True def test_heartbeat_persists_mount_roots_and_workers_returns_them(client): client.post( "/heartbeat", json={ "host": "desktop-vm", "free_vram_gb": 20, "unregistered_vram_gb": 0, "free_ram_gb": 30, "idle_cpus": 8, "arch": "x86_64", "os": "linux", "gpu": True, "tags": [], "host_aliases": [], "mount_roots": ["/home", "/mnt/d", "/tmp"], }, ) r = client.get("/workers") assert r.status_code == 200 rows = [w for w in r.json() if w["host"] == "desktop-vm"] assert rows and rows[0]["mount_roots"] == ["/home", "/mnt/d", "/tmp"] def test_next_job_skips_job_with_cwd_outside_worker_mount_roots(client): """Worker that advertises mount_roots must not receive jobs whose cwd isn't under any of them — otherwise the job fails at subprocess spawn with cwd-does-not-exist.""" client.post( "/heartbeat", json={ "host": "desktop-vm", "free_vram_gb": 20, "unregistered_vram_gb": 0, "free_ram_gb": 30, "idle_cpus": 8, "host_aliases": [], "mount_roots": ["/home", "/tmp"], }, ) # A second worker DOES cover /mnt/d so the submit is routable (the realistic # fleet state). The submit-time cwd_routability probe hard-denies an any-pin # cwd that NO known worker covers; here laptop covers it, so submit succeeds # and we exercise the /next-job filter for the non-covering desktop-vm. client.post( "/heartbeat", json={ "host": "laptop", "free_vram_gb": 20, "unregistered_vram_gb": 0, "free_ram_gb": 30, "idle_cpus": 8, "host_aliases": [], "mount_roots": ["/mnt/d"], }, ) r = client.post( "/submit", json={ "cmd": ["ls"], "cwd": "/mnt/d/some/project", "project": "project-a", "host_pin": "any", }, ) assert r.status_code == 200 got = client.post( "/next-job", json={ "host": "desktop-vm", "free_vram_gb": 20, "unregistered_vram_gb": 0, "free_ram_gb": 30, "idle_cpus": 8, }, ).json() assert got is None or got == {} def test_next_job_empty_mount_roots_disables_filter(client): """A worker that doesn't advertise mount_roots (older binary) gets the old behavior: no filtering. Backwards-compat guard — don't starve the pre-2026-04-24 fleet.""" client.post( "/heartbeat", json={ "host": "legacy-worker", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, "host_aliases": [], }, ) r = client.post( "/submit", json={ "cmd": ["ls"], "cwd": "/some/unusual/path", "project": "project-a", }, ) job_id = r.json()["id"] got = client.post( "/next-job", json={ "host": "legacy-worker", "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, }, ).json() assert got is not None and got["id"] == job_id def test_next_job_mixed_fleet_routes_by_mount_root(client): """Two workers with disjoint roots; each only claims jobs under its own root.""" for host, roots in (("host-a", ["/home"]), ("host-b", ["/mnt/d"])): client.post( "/heartbeat", json={ "host": host, "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, "host_aliases": [], "mount_roots": roots, }, ) job_home = client.post( "/submit", json={"cmd": ["ls"], "cwd": "/home/user/proj", "project": "demo-project"}, ).json() job_d = client.post( "/submit", json={"cmd": ["ls"], "cwd": "/mnt/d/data", "project": "project-a"}, ).json() def _poll(host): return client.post( "/next-job", json={ "host": host, "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, }, ).json() got_a = _poll("host-a") assert got_a is not None and got_a["id"] == job_home["id"] got_b = _poll("host-b") assert got_b is not None and got_b["id"] == job_d["id"] def test_fanin_requires_all_parents_complete(client): p1 = _submit(client).json() p2 = _submit(client).json() child = _submit(client, depends_on=[p1["id"], p2["id"]]).json() _heartbeat(client) first = _next_job(client).json() client.post(f"/jobs/{first['id']}/complete", json={"exit_code": 0}) second = _next_job(client).json() assert second["id"] in (p1["id"], p2["id"]) assert second["id"] != child["id"] client.post(f"/jobs/{second['id']}/complete", json={"exit_code": 0}) third = _next_job(client).json() assert third["id"] == child["id"] # ---------- scheduler-awareness warnings (#38 blocked-by-load, #39 single-slot) ---------- def _heartbeat_caps(client, host: str, **kwargs): """Heartbeat with default linux/x86_64/no-gpu unless overridden.""" body = { "host": host, "free_vram_gb": 0, "unregistered_vram_gb": 0, "free_ram_gb": 8, "idle_cpus": 4, "arch": "x86_64", "os": "linux", "gpu": False, "tags": [], "host_aliases": [], } body.update(kwargs) client.post("/heartbeat", json=body) def test_submit_warns_when_only_eligible_worker_is_busy(client): """#39: one host satisfies caps + host_pin AND has a job in flight → warning.""" _heartbeat_caps(client, "solo") first = client.post( "/submit", json={"cmd": ["sleep", "60"], "cwd": "/tmp", "project": "project-a"}, ).json() # Claim it so 'solo' is busy claimed = _next_job(client, host="solo").json() assert claimed["id"] == first["id"] second = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"}, ).json() assert second["warning"] is not None assert "will queue behind" in second["warning"] assert f"job {first['id']}" in second["warning"] assert "solo" in second["warning"] def test_submit_no_warning_when_multiple_eligible_workers(client): """#39: with two eligible hosts, even if one is busy, no single-slot warning.""" _heartbeat_caps(client, "host-a") _heartbeat_caps(client, "host-b") first = client.post( "/submit", json={"cmd": ["sleep", "60"], "cwd": "/tmp", "project": "project-a"}, ).json() _next_job(client, host="host-a") # busy 'host-a' second = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"}, ).json() assert second["warning"] is None assert first["id"] != second["id"] def test_submit_no_warning_when_only_worker_is_idle(client): """#39: one host but idle → no warning (the matcher will pick it next poll).""" _heartbeat_caps(client, "solo") j = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"}, ).json() assert j["warning"] is None def test_submit_preflight_unknown_host_pin_typo(client): """#43: `--host desktop-vm` (typo for `desktop`) gets a submit-time warning, not a 60s-later sweeper warning. Catches the silent-queue-forever class.""" _heartbeat_caps(client, "desktop") _heartbeat_caps(client, "laptop") r = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "host_pin": "desktop-vm", }, ) body = r.json() assert body["warning"] is not None assert "desktop-vm" in body["warning"] assert "unsatisfiable host_pin" in body["warning"] def test_submit_preflight_unmet_needs_tag(client): """#43: `--needs nonexistent-tag` flagged at submit, not 60s later.""" _heartbeat_caps(client, "desktop", tags=["python3"]) r = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "requires": {"needs": ["nonexistent-tag"]}, }, ) body = r.json() assert body["warning"] is not None assert "nonexistent-tag" in body["warning"] def test_submit_preflight_offline_worker_satisfies_pin(client): """A worker that's been seen but is offline still represents fleet capability — don't false-positive a typo when desktop is rebooting.""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Worker _heartbeat_caps(client, "desktop") # Mark offline by aging last_heartbeat past OFFLINE_AFTER_SECONDS. engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Worker) .where(Worker.host == "desktop") .values( state="offline", last_heartbeat=datetime.now(UTC) - timedelta(seconds=600), ) ) r = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "host_pin": "desktop", }, ) body = r.json() # Either no warning, or only "will queue behind" / contention warning — # never an "unsatisfiable host_pin" claim against a known offline worker. if body["warning"] is not None: assert "unsatisfiable host_pin" not in body["warning"] def test_submit_preflight_silent_when_satisfiable(client): _heartbeat_caps(client, "desktop", tags=["cuda-32gb"], gpu=True, free_vram_gb=24) r = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "requires": {"gpu": True, "needs": ["cuda-32gb"]}, }, ) body = r.json() if body["warning"] is not None: assert "unsatisfiable" not in body["warning"] def test_submit_warning_respects_host_pin(client): """#39: a host_pin'd job sees only matching hosts as eligible.""" _heartbeat_caps(client, "host-a") _heartbeat_caps(client, "host-b") # Pin to host-a, busy host-a first = client.post( "/submit", json={ "cmd": ["sleep", "60"], "cwd": "/tmp", "project": "project-a", "host_pin": "host-a", }, ).json() _next_job(client, host="host-a") pinned = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "host_pin": "host-a", }, ).json() assert pinned["warning"] is not None assert "will queue behind" in pinned["warning"] assert "host-a" in pinned["warning"] assert first["id"] != pinned["id"] # --- #42 Fix A: GPU contention warning --- def test_submit_gpu_contention_warning_fires_when_all_gpu_workers_saturated(client): """A `--gpu` job submitted when every GPU worker has foreign processes holding most VRAM gets a warning that names the saturated hosts.""" _heartbeat_caps( client, "desktop", gpu=True, free_vram_gb=0.5, unregistered_vram_gb=31.5, tags=["cuda"], ) _heartbeat_caps( client, "laptop", gpu=True, free_vram_gb=1.0, unregistered_vram_gb=23.0, tags=["cuda"], ) j = client.post( "/submit", json={ "cmd": ["nvidia-smi"], "cwd": "/tmp", "project": "project-a", "requires": {"gpu": True}, }, ).json() assert j["warning"] is not None assert "GPU contention" in j["warning"] assert "desktop" in j["warning"] assert "laptop" in j["warning"] def test_submit_gpu_contention_silent_when_one_worker_has_headroom(client): """At least one GPU host with VRAM headroom → no contention warning (the matcher will route there).""" _heartbeat_caps( client, "desktop", gpu=True, free_vram_gb=0.5, unregistered_vram_gb=31.5, tags=["cuda"], ) _heartbeat_caps( client, "laptop", gpu=True, free_vram_gb=11.0, unregistered_vram_gb=0.0, tags=["cuda"], ) j = client.post( "/submit", json={ "cmd": ["nvidia-smi"], "cwd": "/tmp", "project": "project-a", "requires": {"gpu": True}, }, ).json() assert j["warning"] is None def test_resident_foreign_model_does_not_close_the_card(client): """#144 end to end: a worker whose card has 8.5 GB free next to a 22.9 GB resident model (NVML free already excludes it) takes a 4 GB GPU job, with no contention warning. Control: an 8 GB job, queued first at a higher priority, still doesn't fit (8 > 8.5 - 1) and stays queued.""" ad = {"free_vram_gb": 8.5, "unregistered_vram_gb": 22.9, "free_ram_gb": 8, "idle_cpus": 4} _heartbeat_caps(client, "ollama-box", gpu=True, tags=["cuda"], **ad) def submit(vram, prio): return client.post( "/submit", json={ "cmd": ["nvidia-smi"], "cwd": "/tmp", "project": "project-a", "requires": {"gpu": True}, "vram_gb": vram, "priority": prio, }, ).json() big = submit(8, 90) small = submit(4, 50) assert small["warning"] is None got = client.post("/next-job", json={"host": "ollama-box", **ad}) assert got.status_code == 200 and got.json()["id"] == small["id"] assert client.get(f"/jobs/{big['id']}").json()["state"] == "queued" def test_submit_gpu_contention_silent_when_no_gpu_required(client): """A CPU-only job is unaffected by GPU foreign-process contention.""" _heartbeat_caps( client, "desktop", gpu=True, free_vram_gb=0.5, unregistered_vram_gb=31.5, tags=["cuda"], ) j = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"}, ).json() assert j["warning"] is None def test_sweep_blocked_warning_after_5min_with_nonpreemptible_blocker(client): """#38: queued >5min with all eligible workers running non-preemptible jobs.""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Job _heartbeat_caps(client, "solo") blocker = client.post( "/submit", json={ "cmd": ["sleep", "9999"], "cwd": "/tmp", "project": "project-a", "preemptible": False, }, ).json() _next_job(client, host="solo") # claim it queued = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "scheduling_timeout_s": 7 * 24 * 3600, }, ).json() # Backdate submitted_at to >5min ago engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Job) .where(Job.id == queued["id"]) .values(submitted_at=datetime.now(UTC) - timedelta(seconds=400)) ) client.app.state.sweep_once() got = client.get(f"/jobs/{queued['id']}").json() assert got["warning"] is not None assert got["warning"].startswith("blocked: ") assert f"job {blocker['id']}" in got["warning"] assert "preempt" in got["warning"] def test_sweep_auto_preempts_when_blocker_is_preemptible(client): """#8 path A: a preemptible blocker on the only eligible worker is auto-preempted in favor of a higher-priority queued job past the queue-age threshold. The blocker's `signal` flips to 'preempt' and the queued job stays warning-free for that path (no nonpreemptible-blocker string).""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd import app as app_mod from jobd.db import Job _heartbeat_caps(client, "solo") blocker = client.post( "/submit", json={ "cmd": ["sleep", "9999"], "cwd": "/tmp", "project": "project-a", "preemptible": True, }, ).json() _next_job(client, host="solo") client.post(f"/jobs/{blocker['id']}/started") queued = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "scheduling_timeout_s": 7 * 24 * 3600, }, ).json() engine = client.app.state.engine runtime_floor = app_mod.AUTO_PREEMPT_MIN_RUNTIME_SECONDS now = datetime.now(UTC) with engine.begin() as conn: conn.execute( update(Job) .where(Job.id == blocker["id"]) .values( priority=30, started_at=now - timedelta(seconds=runtime_floor + 60), ) ) conn.execute( update(Job) .where(Job.id == queued["id"]) .values(priority=90, submitted_at=now - timedelta(seconds=400)) ) client.app.state.sweep_once() sig = client.get(f"/jobs/{blocker['id']}/signal").json() assert sig["signal"] == "preempt" blocker_after = client.get(f"/jobs/{blocker['id']}").json() assert blocker_after["warning"] is not None assert blocker_after["warning"].startswith("auto-preempted in favor of job ") assert str(queued["id"]) in blocker_after["warning"] queued_after = client.get(f"/jobs/{queued['id']}").json() if queued_after["warning"] is not None: assert not queued_after["warning"].startswith("blocked: ") def test_sweep_no_auto_preempt_when_blocker_runtime_below_floor(client): """Blocker started ASSIGNED) and the worker's `/started` POST (state -> RUNNING), there's a real-world window. The endpoint must accept that window: signal=preempt is queued so the worker sees it on its first poll after subprocess launch.""" _heartbeat(client) job = client.post( "/submit", json={ "cmd": ["sleep", "999"], "cwd": "/tmp", "project": "project-a", "preemptible": True, }, ).json() _next_job(client) # QUEUED -> ASSIGNED, NO /started pre = client.get(f"/jobs/{job['id']}").json() assert pre["state"] == "assigned" r = client.post(f"/jobs/{job['id']}/preempt") assert r.status_code == 200 assert r.json()["state"] == "assigned" # state untouched; signal queued sig = client.get(f"/jobs/{job['id']}/signal").json() assert sig["signal"] == "preempt" def test_preempt_blockers_signals_lower_priority_blocker(client): """`POST /jobs/{id}/preempt-blockers` finds a preemptible blocker on the queued job's eligible workers and signals it immediately — no queue-age or runtime guards.""" from sqlalchemy import update from jobd.db import Job _heartbeat_caps(client, "solo") blocker = client.post( "/submit", json={ "cmd": ["sleep", "9999"], "cwd": "/tmp", "project": "project-a", "preemptible": True, }, ).json() _next_job(client, host="solo") client.post(f"/jobs/{blocker['id']}/started") queued = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"}, ).json() engine = client.app.state.engine with engine.begin() as conn: conn.execute(update(Job).where(Job.id == blocker["id"]).values(priority=30)) conn.execute(update(Job).where(Job.id == queued["id"]).values(priority=90)) r = client.post(f"/jobs/{queued['id']}/preempt-blockers") assert r.status_code == 200 body = r.json() assert body["signaled"] == blocker["id"] assert body["worker"] == "solo" sig = client.get(f"/jobs/{blocker['id']}/signal").json() assert sig["signal"] == "preempt" blocker_after = client.get(f"/jobs/{blocker['id']}").json() assert "(manual)" in (blocker_after["warning"] or "") def test_preempt_blockers_409_when_not_queued(client): """Refuses with 409 when target job is not in queued state.""" _heartbeat(client) job = client.post( "/submit", json={ "cmd": ["sleep", "999"], "cwd": "/tmp", "project": "project-a", "preemptible": True, }, ).json() _next_job(client) client.post(f"/jobs/{job['id']}/started") r = client.post(f"/jobs/{job['id']}/preempt-blockers") assert r.status_code == 409 assert "queued" in r.json()["detail"] def test_preempt_blockers_no_candidate_returns_reason(client): """If no preemptible blocker at lower priority qualifies, returns `signaled=null` with a reason explaining why.""" from sqlalchemy import update from jobd.db import Job _heartbeat_caps(client, "solo") blocker = client.post( "/submit", json={ "cmd": ["sleep", "9999"], "cwd": "/tmp", "project": "project-a", "preemptible": True, }, ).json() _next_job(client, host="solo") client.post(f"/jobs/{blocker['id']}/started") queued = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"}, ).json() # Make the queued job's priority NOT exceed the blocker's engine = client.app.state.engine with engine.begin() as conn: conn.execute(update(Job).where(Job.id == blocker["id"]).values(priority=80)) conn.execute(update(Job).where(Job.id == queued["id"]).values(priority=50)) r = client.post(f"/jobs/{queued['id']}/preempt-blockers") assert r.status_code == 200 body = r.json() assert body["signaled"] is None assert "lower priority" in body["reason"] def test_preempt_blockers_force_overrides_priority_guard(client): """`force=true` drops the priority guard, letting an operator preempt blockers at equal-or-higher priority.""" from sqlalchemy import update from jobd.db import Job _heartbeat_caps(client, "solo") blocker = client.post( "/submit", json={ "cmd": ["sleep", "9999"], "cwd": "/tmp", "project": "project-a", "preemptible": True, }, ).json() _next_job(client, host="solo") client.post(f"/jobs/{blocker['id']}/started") queued = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"}, ).json() engine = client.app.state.engine with engine.begin() as conn: conn.execute(update(Job).where(Job.id == blocker["id"]).values(priority=80)) conn.execute(update(Job).where(Job.id == queued["id"]).values(priority=50)) r = client.post(f"/jobs/{queued['id']}/preempt-blockers", params={"force": "true"}) assert r.status_code == 200 body = r.json() assert body["signaled"] == blocker["id"] def test_list_jobs_warnings_only_filters(client): """`GET /jobs?warnings_only=true` returns only jobs with non-null warning. Jobs without warnings drop out of the result.""" from sqlalchemy import update from jobd.db import Job _heartbeat(client) no_warn = _submit(client).json() with_warn = _submit(client).json() engine = client.app.state.engine with engine.begin() as conn: conn.execute(update(Job).where(Job.id == with_warn["id"]).values(warning="test marker")) all_rows = client.get("/jobs").json() ids_all = {j["id"] for j in all_rows} assert {no_warn["id"], with_warn["id"]} <= ids_all only = client.get("/jobs", params={"warnings_only": "true"}).json() ids_only = {j["id"] for j in only} assert with_warn["id"] in ids_only assert no_warn["id"] not in ids_only def test_sweep_clears_blocked_warning_when_blocker_finishes(client): """#38: once the non-preemptible blocker completes, the queue-age warning clears.""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Job _heartbeat_caps(client, "solo") blocker = client.post( "/submit", json={ "cmd": ["sleep", "9999"], "cwd": "/tmp", "project": "project-a", "preemptible": False, }, ).json() _next_job(client, host="solo") queued = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "scheduling_timeout_s": 7 * 24 * 3600, }, ).json() engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Job) .where(Job.id == queued["id"]) .values(submitted_at=datetime.now(UTC) - timedelta(seconds=400)) ) client.app.state.sweep_once() assert client.get(f"/jobs/{queued['id']}").json()["warning"].startswith("blocked: ") # Blocker finishes client.post(f"/jobs/{blocker['id']}/complete", json={"exit_code": 0}) client.app.state.sweep_once() got = client.get(f"/jobs/{queued['id']}").json() assert got["warning"] is None or not got["warning"].startswith("blocked: ") def test_sweep_preserves_will_queue_behind_warning(client): """The matcheable-clear path must not wipe submit-time 'will queue behind' warnings.""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd.db import Job _heartbeat_caps(client, "solo") client.post( "/submit", json={"cmd": ["sleep", "60"], "cwd": "/tmp", "project": "project-a"}, ) _next_job(client, host="solo") queued = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"}, ).json() assert "will queue behind" in queued["warning"] # Push past UNMATCHEABLE threshold but well under the 5min blocked threshold. engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Job) .where(Job.id == queued["id"]) .values(submitted_at=datetime.now(UTC) - timedelta(seconds=90)) ) client.app.state.sweep_once() got = client.get(f"/jobs/{queued['id']}").json() assert got["warning"] is not None assert "will queue behind" in got["warning"] def _read_events(tmp_path) -> list[dict]: import json p = tmp_path / "logs" / "events.jsonl" if not p.exists(): return [] return [json.loads(line) for line in p.read_text().splitlines() if line.strip()] def test_auto_preempt_emits_jsonl_event_from_sweeper(client, tmp_path): """#32: when the sweeper auto-preempts, an `auto_preempt` event with source='sweeper' lands in events.jsonl with both job IDs, the worker, and queue-age.""" from datetime import UTC, datetime, timedelta from sqlalchemy import update from jobd import app as app_mod from jobd.db import Job _heartbeat_caps(client, "solo") blocker = client.post( "/submit", json={ "cmd": ["sleep", "9999"], "cwd": "/tmp", "project": "project-a", "preemptible": True, }, ).json() _next_job(client, host="solo") client.post(f"/jobs/{blocker['id']}/started") queued = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/tmp", "project": "project-a", "scheduling_timeout_s": 7 * 24 * 3600, }, ).json() engine = client.app.state.engine runtime_floor = app_mod.AUTO_PREEMPT_MIN_RUNTIME_SECONDS now = datetime.now(UTC) with engine.begin() as conn: conn.execute( update(Job) .where(Job.id == blocker["id"]) .values(priority=30, started_at=now - timedelta(seconds=runtime_floor + 60)) ) conn.execute( update(Job) .where(Job.id == queued["id"]) .values(priority=90, submitted_at=now - timedelta(seconds=400)) ) client.app.state.sweep_once() events = [e for e in _read_events(tmp_path) if e["event"] == "auto_preempt"] assert len(events) == 1, events e = events[0] assert e["source"] == "broker" assert e["job_id"] == blocker["id"] assert e["project"] == "project-a" p = e["payload"] assert p["cause"] == "sweeper" assert p["queued_job"] == queued["id"] assert p["worker"] == "solo" assert p["queued_priority"] == 90 assert p["candidate_priority"] == 30 assert p["queue_age_s"] >= 400 def test_preempt_blockers_emits_jsonl_event(client, tmp_path): """#32: manual `preempt-blockers` escalation also writes an `auto_preempt` event, distinguished by source='manual' (or 'manual_force' with --force).""" from sqlalchemy import update from jobd.db import Job _heartbeat_caps(client, "solo") blocker = client.post( "/submit", json={ "cmd": ["sleep", "9999"], "cwd": "/tmp", "project": "project-a", "preemptible": True, }, ).json() _next_job(client, host="solo") client.post(f"/jobs/{blocker['id']}/started") queued = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"} ).json() engine = client.app.state.engine with engine.begin() as conn: conn.execute(update(Job).where(Job.id == blocker["id"]).values(priority=30)) conn.execute(update(Job).where(Job.id == queued["id"]).values(priority=90)) client.post(f"/jobs/{queued['id']}/preempt-blockers") client.post( f"/jobs/{queued['id']}/preempt-blockers", params={"force": "true"} ) # second call: candidate already signaled, returns no candidate; no event events = [e for e in _read_events(tmp_path) if e["event"] == "auto_preempt"] assert len(events) == 1, events assert events[0]["source"] == "broker" assert events[0]["job_id"] == blocker["id"] assert events[0]["project"] == "project-a" assert events[0]["payload"]["cause"] == "manual" assert events[0]["payload"]["queued_job"] == queued["id"] assert events[0]["payload"]["worker"] == "solo" def test_checkpoint_complete_endpoint_emits_event(client, tmp_path): """#37: workers POST /jobs/{id}/checkpoint-complete after observing the user's `jobd-checkpoint-complete` token in stdout during a preempt. The endpoint appends a `checkpoint_complete` event to events.jsonl and makes no DB state changes.""" submit = client.post( "/submit", json={"cmd": ["echo", "hi"], "cwd": "/tmp", "project": "project-a"}, ).json() job_id = submit["id"] r = client.post(f"/jobs/{job_id}/checkpoint-complete") assert r.status_code == 200, r.text assert r.json() == {"ok": True} events = [e for e in _read_events(tmp_path) if e["event"] == "checkpoint_complete"] assert len(events) == 1, events assert events[0]["source"] == "broker" assert events[0]["job_id"] == job_id assert events[0]["project"] == "project-a" def test_checkpoint_complete_endpoint_404_unknown_job(client): r = client.post("/jobs/9999999/checkpoint-complete") assert r.status_code == 404 assert "no such job" in r.json()["detail"] def test_refuse_admission_requeues_and_emits_event(client, tmp_path): """#41c: a worker that pulled a job and observed live GPU contention posts /refuse-admission. Broker reverts state QUEUED, clears worker + started_at, and writes an `admission_blocked` event.""" client.post( "/heartbeat", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ) submitted = client.post( "/submit", json={"cmd": ["echo", "hi"], "cwd": "/tmp", "project": "project-b"}, ).json() job_id = submitted["id"] pulled = client.post( "/next-job", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, }, ).json() assert pulled is not None assert pulled["state"] == "assigned" assert pulled["worker"] == "desktop" r = client.post( f"/jobs/{job_id}/refuse-admission", json={ "required_gb": 8.0, "free_gb": 1.5, "foreign_pids": [4242, 4243], "foreign_vram_gb": 22.5, }, ) assert r.status_code == 200, r.text info = r.json() assert info["state"] == "queued" assert info["worker"] is None assert info["started_at"] is None events = [e for e in _read_events(tmp_path) if e["event"] == "admission_blocked"] assert len(events) == 1, events e = events[0] assert e["source"] == "broker" assert e["job_id"] == job_id assert e["project"] == "project-b" p = e["payload"] assert p["worker"] == "desktop" assert p["required_gb"] == 8.0 assert p["free_gb"] == 1.5 assert p["foreign_pids"] == [4242, 4243] assert p["foreign_vram_gb"] == 22.5 def test_refuse_admission_409_when_not_assigned(client): """Job is queued (never pulled) → /refuse-admission returns 409.""" submitted = client.post( "/submit", json={"cmd": ["echo", "hi"], "cwd": "/tmp", "project": "project-b"}, ).json() job_id = submitted["id"] r = client.post( f"/jobs/{job_id}/refuse-admission", json={"required_gb": 8.0, "free_gb": 1.5}, ) assert r.status_code == 409 assert "queued" in r.json()["detail"] assert "must be 'assigned'" in r.json()["detail"] def test_refuse_admission_404_unknown_job(client): r = client.post( "/jobs/9999999/refuse-admission", json={"required_gb": 8.0, "free_gb": 1.5}, ) assert r.status_code == 404 assert "no such job" in r.json()["detail"] def test_submit_vram_gb_explicit_value_persists_to_job_row(client): """#41d end-to-end: `vram_gb` on JobSubmit lands on the Job row and surfaces through JobInfo. Without the field on JobSubmit, Pydantic silently dropped it and the matcher fell back to 0/tier-tag/floor.""" info = client.post( "/submit", json={ "cmd": ["echo", "ok"], "cwd": "/tmp", "project": "project-b", "vram_gb": 12.0, }, ).json() assert info["vram_gb"] == 12.0 def test_submit_vram_gb_zero_uses_profile_default(client, tmp_path): """vram_gb omitted (default 0) — profile spec wins if one is selected.""" info = client.post( "/submit", json={ "cmd": ["echo", "ok"], "cwd": "/tmp", "project": "project-b", }, ).json() assert info["vram_gb"] == 0.0 def _register_worker(client, host, mount_roots, host_aliases=None): """Register/heartbeat a worker advertising the given mount_roots.""" return client.post( "/heartbeat", json={ "host": host, "free_vram_gb": 20, "unregistered_vram_gb": 0, "free_ram_gb": 30, "idle_cpus": 8, "host_aliases": host_aliases or [], "mount_roots": mount_roots, }, ) def test_submit_hard_denies_pinned_cwd_without_mount_root(client): # Register a worker whose mount_roots do NOT cover the cwd, pinned to it. _register_worker(client, host="desktop", mount_roots=["/home", "/tmp"]) r = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/mnt/d/data/x", "project": "project-a", "host_pin": "desktop", }, ) assert r.status_code == 400 assert "/mnt/d/data/x" in r.text def test_submit_hard_denies_any_pin_uncovered_cwd(client): # any-pin cwd under no worker's mount_roots is unroutable everywhere -> 400, # not a silent QUEUED sit (B never fires because no worker claims it). _register_worker(client, host="broker-host", mount_roots=["/home", "/tmp"]) r = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/scratch/run1", "project": "project-a", "host_pin": "any", }, ) assert r.status_code == 400 assert "/scratch/run1" in r.text def test_job_orm_has_excluded_workers_column(): """B: the per-job cwd-refusal exclusion set column exists on the ORM.""" from jobd.db import Job assert hasattr(Job, "excluded_workers_json") def _submit_and_assign(client, cwd, worker, mount_roots=None, host_pin="any"): """Submit a job and claim it on `worker` via /next-job -> returns job id (ASSIGNED).""" mr = mount_roots if mount_roots is not None else ["/home"] r = client.post( "/submit", json={ "cmd": ["true"], "cwd": cwd, "project": "project-a", "host_pin": host_pin, }, ) assert r.status_code == 200, r.text jid = r.json()["id"] got = client.post( "/next-job", json={ "host": worker, "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, "mount_roots": mr, }, ).json() assert got is not None and got["id"] == jid, f"claim failed: {got}" return jid def test_refuse_admission_cwd_missing_records_exclusion_and_requeues(client): _register_worker(client, host="laptop", mount_roots=["/home"]) _register_worker(client, host="desktop", mount_roots=["/home"]) cwd = "/home/u/p/.claude/worktrees/wt" jid = _submit_and_assign(client, cwd=cwd, worker="desktop") r = client.post(f"/jobs/{jid}/refuse-admission", json={"reason": "cwd_missing", "cwd": cwd}) assert r.status_code == 200, r.text assert r.json()["state"] == "queued" # laptop remains -> re-queued # desktop is excluded: a poll as desktop must not re-offer the job got = client.post( "/next-job", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, "mount_roots": ["/home"], }, ).json() assert got is None or got["id"] != jid # laptop still gets it got2 = client.post( "/next-job", json={ "host": "laptop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, "mount_roots": ["/home"], }, ).json() assert got2 is not None and got2["id"] == jid def test_refuse_admission_cwd_missing_terminal_when_no_eligible_left(client): _register_worker(client, host="desktop", mount_roots=["/home"]) cwd = "/home/u/p/wt" jid = _submit_and_assign(client, cwd=cwd, worker="desktop") r = client.post(f"/jobs/{jid}/refuse-admission", json={"reason": "cwd_missing", "cwd": cwd}) assert r.status_code == 200, r.text info = r.json() assert info["state"] == "failed" assert info["termination_reason"] == "cwd_unreachable" def test_refuse_admission_cwd_unreachable_cascades_to_children(client): """audit 2026-07-01 H2: when a job fails cwd_unreachable (no eligible worker advertises its cwd), its default-policy dependents must cascade-cancel — the old path failed the parent without a cascade and stranded the child.""" _register_worker(client, host="desktop", mount_roots=["/home"]) cwd = "/home/u/p/wt3" parent_id = _submit_and_assign(client, cwd=cwd, worker="desktop") child = client.post( "/submit", json={ "cmd": ["true"], "cwd": "/home/u/p/child", "project": "project-a", "depends_on": [parent_id], }, ).json() r = client.post( f"/jobs/{parent_id}/refuse-admission", json={"reason": "cwd_missing", "cwd": cwd} ) assert r.status_code == 200, r.text assert r.json()["termination_reason"] == "cwd_unreachable" row = client.get(f"/jobs/{child['id']}").json() assert row["state"] == "cancelled", row assert "parent_failed" in (row.get("warning") or "") def test_refuse_admission_gpu_contention_unchanged(client): _register_worker(client, host="desktop", mount_roots=["/home"]) jid = _submit_and_assign(client, cwd="/home/u/p", worker="desktop") r = client.post(f"/jobs/{jid}/refuse-admission", json={"required_gb": 20.0, "free_gb": 2.0}) assert r.status_code == 200, r.text assert r.json()["state"] == "queued" def test_refuse_admission_rejects_stale_worker(client): """M2 completeness (review finding): a worker that isn't the job's current owner must not be able to requeue/exclude it via /refuse-admission — after a partition-reclaim + re-dispatch, the original worker's delayed refusal would otherwise wipe the new owner's active claim. Foreign X-Jobd-Worker -> 409, job untouched; the owner's own refusal still works.""" _register_worker(client, host="desktop", mount_roots=["/home"]) jid = _submit_and_assign(client, cwd="/home/u/p", worker="desktop") # job.worker=desktop stale = client.post( f"/jobs/{jid}/refuse-admission", json={"required_gb": 20.0, "free_gb": 2.0}, headers={"X-Jobd-Worker": "ghost-worker"}, ) assert stale.status_code == 409, stale.text assert client.get(f"/jobs/{jid}").json()["state"] == "assigned" # not clobbered ok = client.post( f"/jobs/{jid}/refuse-admission", json={"required_gb": 20.0, "free_gb": 2.0}, headers={"X-Jobd-Worker": "desktop"}, ) assert ok.status_code == 200, ok.text assert ok.json()["state"] == "queued" def test_next_job_skips_excluded_worker(client): _register_worker(client, host="laptop", mount_roots=["/home"]) _register_worker(client, host="desktop", mount_roots=["/home"]) cwd = "/home/u/p/wt2" jid = _submit_and_assign(client, cwd=cwd, worker="desktop") client.post(f"/jobs/{jid}/refuse-admission", json={"reason": "cwd_missing", "cwd": cwd}) got = client.post( "/next-job", json={ "host": "desktop", "free_vram_gb": 30.0, "unregistered_vram_gb": 0.0, "free_ram_gb": 28.0, "idle_cpus": 10, "mount_roots": ["/home"], }, ).json() # Only one job exists and desktop is excluded -> desktop gets nothing. assert got is None def test_wait_streams_log_in_bounded_slices_without_utf8_corruption(client, tmp_path, monkeypatch): """LOW: /wait must read the backlog in bounded slices (not f.read() the whole file into one allocation) and must not corrupt a multibyte UTF-8 char that lands on a slice boundary.""" from sqlalchemy import update from jobd.broker.routes import jobs as jobs_routes_mod from jobd.db import Job from jobd.models import JobState # Force a tiny slice so the 4-byte emoji straddles a read boundary and the # 20-byte log is delivered across several slices. /wait resolves the # constant from the routes module's globals (Stage-3 split). monkeypatch.setattr(jobs_routes_mod, "WAIT_STREAM_CHUNK_BYTES", 8) job_id = client.post( "/submit", json={"cmd": ["bash", "-c", "x"], "cwd": "/tmp", "project": "project-a"}, ).json()["id"] # Emoji bytes span offsets 6..9, so with an 8-byte slice its four bytes are # split across slice 1 (offsets 0-7) and slice 2 (offsets 8-15). content = "ABCDEF" + "\U0001f600" + "GHIJKLMNOP" log_file = tmp_path / "logs" / f"{job_id}.log" log_file.parent.mkdir(parents=True, exist_ok=True) log_file.write_bytes(content.encode("utf-8")) # Flip to a terminal state so the generator drains and returns immediately # instead of polling. engine = client.app.state.engine with engine.begin() as conn: conn.execute( update(Job).where(Job.id == job_id).values(state=JobState.COMPLETED.value, exit_code=0) ) resp = client.get(f"/wait/{job_id}") assert resp.status_code == 200 raw = resp.text # Reassemble the log payload (content has no newlines -> one data line per # slice). The trailing terminal event's JSON data is also a data: line, so # assert the log content is a contiguous substring rather than equal. payload = "".join( line[len("data:") :].lstrip(" ") for line in raw.splitlines() if line.startswith("data:") ) assert content in payload # full backlog delivered, emoji intact assert "�" not in payload # no replacement char -> boundary handled assert raw.count("event: log") >= 2 # delivered across multiple slices assert '"state": "completed"' in raw or '"state":"completed"' in raw def test_complete_rejects_non_int_exit_code(client): """/complete takes a typed CompletePayload (audit 2026-07-05): an untyped dict used to let a malformed worker payload store a string exit_code in the int column; now it's a 422 at the Pydantic boundary.""" job_id = client.post( "/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a"} ).json()["id"] r = client.post(f"/jobs/{job_id}/complete", json={"exit_code": "not-an-int"}) assert r.status_code == 422 # And the job is untouched. assert client.get(f"/jobs/{job_id}").json()["state"] == "queued"