feat(v7-h06): job-pause/resume primitive

Closes V7-H06 (cppmega-mlx-nh1i) backend: in-process pause/resume
registry mirroring V5-G09 abort_token pattern but with a pause→resume
cycle.

cppmega_v4/runtime/job_control.py:
  - pause(token) / resume(token) / is_paused(token) — idempotent.
  - wait_while_paused(token, poll_s, max_wait_s) — bounded blocking.
  - reset() — test helper.

Tests (tests/v4/test_job_control.py): 6/6 — pause sets, resume
clears, pause idempotent, null/empty token never paused,
wait_while_paused returns when resume fires from another thread,
max_wait bound respected, distinct tokens independent.

stage_train wait_while_paused integration + UI Pause/Resume
buttons next to Cancel are V7-H06-UI follow-ups.

Author: apstenku123

cppmega_v4/runtime/job_control.py

@@ -0,0 +1,60 @@
+"""V7-H06: in-process job-pause/resume registry.
+
+Mirrors V5-G09 abort_token pattern (cppmega_v4.runner.stages.request_abort)
+but with a pause→resume cycle instead of a one-shot abort.
+
+The train loop polls is_paused(token) between steps; while paused it
+spins on a 50ms sleep without advancing optimizer state. resume(token)
+clears the flag. pause(token) sets it.
+"""
+
+from __future__ import annotations
+
+import time
+from threading import Lock
+
+_PAUSED_JOBS: set[str] = set()
+_LOCK = Lock()
+
+
+def pause(token: str) -> None:
+    """Mark a job as paused. Idempotent."""
+    if not token:
+        return
+    with _LOCK:
+        _PAUSED_JOBS.add(str(token))
+
+
+def resume(token: str) -> None:
+    """Clear the pause flag. Idempotent."""
+    with _LOCK:
+        _PAUSED_JOBS.discard(str(token))
+
+
+def is_paused(token: str | None) -> bool:
+    if not token:
+        return False
+    with _LOCK:
+        return str(token) in _PAUSED_JOBS
+
+
+def wait_while_paused(token: str | None,
+                       *, poll_s: float = 0.05,
+                       max_wait_s: float = 600.0) -> None:
+    """Block while the job is paused, polling every poll_s. Bounded by
+    max_wait_s to avoid deadlock if resume() never lands."""
+    if not token:
+        return
+    deadline = time.time() + max_wait_s
+    while is_paused(token) and time.time() < deadline:
+        time.sleep(poll_s)
+
+
+def reset() -> None:
+    """Test helper — clear the entire paused-jobs set."""
+    with _LOCK:
+        _PAUSED_JOBS.clear()
+
+
+__all__ = ["pause", "resume", "is_paused",
+           "wait_while_paused", "reset"]

tests/v4/test_job_control.py

@@ -0,0 +1,65 @@
+"""V7-H06: job pause/resume primitive tests."""
+
+from __future__ import annotations
+
+import threading
+import time
+
+import pytest
+
+from cppmega_v4.runtime import job_control as jc
+
+
+@pytest.fixture(autouse=True)
+def _reset():
+    jc.reset()
+    yield
+    jc.reset()
+
+
+def test_v7_h06_pause_sets_flag_resume_clears():
+    assert jc.is_paused("t1") is False
+    jc.pause("t1")
+    assert jc.is_paused("t1") is True
+    jc.resume("t1")
+    assert jc.is_paused("t1") is False
+
+
+def test_v7_h06_pause_is_idempotent():
+    jc.pause("t1")
+    jc.pause("t1")
+    jc.resume("t1")
+    assert jc.is_paused("t1") is False
+
+
+def test_v7_h06_null_or_empty_token_never_paused():
+    assert jc.is_paused(None) is False
+    assert jc.is_paused("") is False
+
+
+def test_v7_h06_wait_while_paused_returns_when_resumed():
+    jc.pause("job-X")
+
+    def _resumer():
+        time.sleep(0.15)
+        jc.resume("job-X")
+
+    threading.Thread(target=_resumer, daemon=True).start()
+    t0 = time.time()
+    jc.wait_while_paused("job-X", poll_s=0.02, max_wait_s=2.0)
+    elapsed = time.time() - t0
+    assert 0.10 <= elapsed < 2.0
+
+
+def test_v7_h06_wait_while_paused_bounded_by_max_wait():
+    jc.pause("stuck")
+    t0 = time.time()
+    jc.wait_while_paused("stuck", poll_s=0.02, max_wait_s=0.1)
+    elapsed = time.time() - t0
+    assert 0.08 < elapsed < 0.5
+
+
+def test_v7_h06_distinct_tokens_independent():
+    jc.pause("a")
+    assert jc.is_paused("a") is True
+    assert jc.is_paused("b") is False