diff --git a/tests/dragon_ci/README.md b/tests/dragon_ci/README.md
new file mode 100644
index 0000000..47ca491
--- /dev/null
+++ b/tests/dragon_ci/README.md
@@ -0,0 +1,89 @@
+# Dragon CI tests
+
+Standalone pre-release contract tests for the Dragon runtime. The suite has
+**no dependency on Rhapsody** — only `dragon`, `pytest`, `pytest-asyncio`,
+`pytest-timeout`, `cloudpickle`, and the standard library. Its purpose is to
+fail loudly when a Dragon release breaks any API surface that Rhapsody (or
+similar downstream stacks built on the V3 `Batch` workflow) depends on.
+
+## What's covered
+
+Each test pins one observable contract: symbol name, callable signature,
+attribute presence, or a short behavioral round-trip. Failures should point
+the Dragon team at the specific contract that changed.
+
+**Happy-path contract files** (expected to pass):
+
+| File                            | Subsystem                                            |
+|---------------------------------|------------------------------------------------------|
+| `test_batch_lifecycle.py`       | `Batch` ctor kwargs, attrs, close/join/terminate    |
+| `test_batch_function.py`        | `Batch.function()` + `results_ddict` 5-tuple shape  |
+| `test_batch_process.py`         | `Batch.process()` + stdio-capture pitfalls          |
+| `test_batch_job.py`             | `Batch.job()` (wire + opt-in real MPI launch)       |
+| `test_process_template.py`      | `ProcessTemplate` ctor, attrs, `Popen` consts       |
+| `test_policy.py`                | `Policy` ctor, `Distribution`/`Placement` enums     |
+| `test_machine_system.py`        | `System().nnodes`, `hostname_policies`              |
+| `test_process_group.py`         | native `ProcessGroup` lifecycle                     |
+| `test_queue_event.py`           | `native.queue.Queue` + `native.event.Event`         |
+| `test_ddict.py`                 | `DDict` core ops                                    |
+| `test_telemetry_collector.py`   | `dragon.telemetry.collector` + `AccVendor`          |
+| `test_async_in_batch.py`        | async-coroutine wrapping pattern (V3)               |
+| `test_worker_failure_modes.py`  | caught-exception and `sys.exit` reporting paths     |
+
+**Known-Dragon-bug pins** (expected to FAIL via pytest-timeout until Dragon fixes them):
+
+| File                                       | Bug pinned                                                |
+|--------------------------------------------|-----------------------------------------------------------|
+| `test_ddict_unknown_key_blocks.py`         | `DDict[unknown_key]` blocks instead of raising `KeyError` when `wait_for_keys=True` |
+| `test_sigkill_in_worker_hangs.py`          | SIGKILL'd worker is undetected; `task.get()` blocks forever; `timeout=` kwarg is ignored on the blocking DDict read |
+| `test_unpickleable_function_hangs.py`      | Worker dies during cloudpickle unpickling without diagnostic — same silent-hang as SIGKILL |
+
+These three live in their own files because `pytest-timeout`'s `method="thread"`
+leaks the wedged test thread, which holds Dragon resources and breaks any
+subsequent test in the same pytest invocation. One bug per file keeps things
+clean.
+
+## Running
+
+The suite must run under the Dragon launcher because Dragon's multiprocessing
+backend is required even for in-process tests.
+
+```
+dragon python -m pytest -c tests/dragon_ci/pytest.ini \
+    --rootdir=tests/dragon_ci tests/dragon_ci/ -v
+```
+
+The `-c` and `--rootdir` flags keep pytest from picking up the parent
+`tests/conftest.py` (which imports Rhapsody).
+
+## Skip markers
+
+Tests auto-skip what the environment can't support:
+
+| Marker                          | Active when                                            |
+|---------------------------------|--------------------------------------------------------|
+| `requires_multi_node`           | `System().nnodes >= 2`                                 |
+| `requires_gpu`                  | `identify_gpu()` reports at least one device          |
+| `requires_mpi`                  | `DRAGON_CI_HAS_PMI=1` is set (opt-in)                  |
+| `requires_telemetry_collector`  | `dragon.telemetry.collector` is importable             |
+
+MPI launches are opt-in because a failing PMIx init aborts the entire Dragon
+runtime — not just the calling test — and so cannot be probed safely.
+
+## Adding tests
+
+- One contract per test. Prefer parametrized signature/attribute checks for
+  enumerations of kwargs or methods.
+- For callables passed to `Batch.function()` / `ProcessTemplate(target=…)`,
+  define the function in `_dragon_ci_helpers.py`, **not** in the test file.
+  Top-level callables defined in `test_*.py` files cannot be re-imported by
+  Dragon workers (pytest module names aren't on `PYTHONPATH`); the worker
+  fails silently to unpickle and `task.get()` blocks forever. The conftest
+  injects this directory into `PYTHONPATH` so the helpers module IS visible
+  to workers.
+- Use the `batch` session fixture when possible. Use `fresh_batch` only for
+  tests that intentionally kill workers — those leave the Batch unrecoverable.
+- New "documents a Dragon bug" tests should each live in their own file,
+  use `@pytest.mark.timeout(N, method="thread")`, and contain only the one
+  test (see the existing `*_hangs.py` / `*_blocks.py` files as templates).
+- Do **not** import `rhapsody` from anywhere in this directory.
diff --git a/tests/dragon_ci/_dragon_ci_helpers.py b/tests/dragon_ci/_dragon_ci_helpers.py
new file mode 100644
index 0000000..65762d2
--- /dev/null
+++ b/tests/dragon_ci/_dragon_ci_helpers.py
@@ -0,0 +1,78 @@
+"""Top-level helpers used by tests.
+
+Must live in a regular importable module (not in a ``test_*.py`` file) so
+Dragon's workers can re-import it via cloudpickle's by-reference path. See
+the conftest module docstring for the full story.
+"""
+
+import asyncio
+import os
+import signal
+import sys
+import time
+
+# --- Batch.function targets ----------------------------------------------
+
+
+def add(a, b):
+    return a + b
+
+
+def kwfn(x, *, mult=1):
+    return x * mult
+
+
+def slow_double(x, delay):
+    time.sleep(delay)
+    return x * 2
+
+
+def print_and_return(value, msg):
+    print(msg)
+    print("err msg", file=sys.stderr)
+    return value
+
+
+def raise_value_error(message):
+    raise ValueError(message)
+
+
+# --- async dispatch (test_async_in_batch.py) -----------------------------
+
+
+async def async_double(x):
+    await asyncio.sleep(0)
+    return x * 2
+
+
+def async_run_shim(*args, **kwargs):
+    """Mirror Rhapsody's wrapper in dragon.py V3."""
+    return asyncio.run(async_double(*args, **kwargs))
+
+
+# --- ProcessGroup workers (test_process_group.py, test_queue_event.py) ---
+
+
+def pg_worker(queue, shutdown_event):
+    """Push one identity message, then loop until shutdown_event is set."""
+    queue.put({"pid": os.getpid(), "host": os.uname().nodename}, timeout=10.0)
+    while not shutdown_event.wait(timeout=0.05):
+        pass
+
+
+# --- worker failure-mode probes (test_worker_failure_modes.py) -----------
+
+
+def fn_assert_false(_x):
+    # Raise explicitly rather than `assert False` so the probe still fires under
+    # `python -O` (which strips asserts); the test expects this AssertionError.
+    raise AssertionError("intentional dragon_ci probe")
+
+
+def fn_sys_exit_one(_x):
+    sys.exit(1)
+
+
+def fn_kill_self(_x):
+    """Abnormal worker termination — no chance for Dragon's exit-detector."""
+    os.kill(os.getpid(), signal.SIGKILL)
diff --git a/tests/dragon_ci/_offpath/marker_helper.py b/tests/dragon_ci/_offpath/marker_helper.py
new file mode 100644
index 0000000..19fc43d
--- /dev/null
+++ b/tests/dragon_ci/_offpath/marker_helper.py
@@ -0,0 +1,10 @@
+"""Helper module deliberately placed in a subdirectory that the conftest does NOT add to
+``PYTHONPATH``.
+
+Tests that want to reproduce the runtime
+``sys.path.insert`` regression import from here.
+"""
+
+
+def offpath_add(a, b):
+    return a + b
diff --git a/tests/dragon_ci/conftest.py b/tests/dragon_ci/conftest.py
new file mode 100644
index 0000000..e05cb89
--- /dev/null
+++ b/tests/dragon_ci/conftest.py
@@ -0,0 +1,196 @@
+"""Pytest configuration for the Dragon CI contract suite.
+
+Run with::
+
+    dragon python -m pytest -c tests/dragon_ci/pytest.ini \
+        --rootdir=tests/dragon_ci tests/dragon_ci/
+
+The ``-c`` / ``--rootdir`` flags keep pytest from picking up the parent
+``tests/conftest.py`` (which imports Rhapsody).
+
+Worker-importability note
+-------------------------
+cloudpickle pickles top-level functions **by reference** (just the
+``module.qualname`` pair). Dragon workers re-import the module by that
+name, and their ``sys.path`` is NOT inherited from the parent — only the
+``PYTHONPATH`` env var is. So a helper that the parent can import only
+because of a runtime ``sys.path.insert`` will fail in the worker, the
+result is never written to the DDict, and ``task.get()`` blocks forever
+(see ``test_worker_failure_modes.py``).
+
+To avoid the trap, this conftest puts the suite directory on both
+``sys.path`` (for the parent) and ``PYTHONPATH`` (for workers). All
+callables passed to Batch tasks must live in ``_dragon_ci_helpers``,
+not in a ``test_*.py`` file.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+
+import pytest
+import pytest_asyncio
+
+# --- Make _dragon_ci_helpers importable from workers ----------------------
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+if _HERE not in sys.path:
+    sys.path.insert(0, _HERE)
+# Only append a separator when PYTHONPATH is already populated; a trailing
+# separator (empty entry) is implicitly treated as the cwd, which we don't want.
+_existing_pp = os.environ.get("PYTHONPATH")
+os.environ["PYTHONPATH"] = f"{_HERE}{os.pathsep}{_existing_pp}" if _existing_pp else _HERE
+
+
+# --- Dragon must be importable -------------------------------------------
+
+pytest.importorskip("dragon", reason="run under `dragon python`")
+
+
+# --- Environment detection (single pass at import time) ------------------
+
+
+def _safe(fn, default):
+    try:
+        return fn()
+    except Exception:
+        return default
+
+
+def _detect_nnodes() -> int:
+    from dragon.native.machine import System
+
+    return int(System().nnodes)
+
+
+def _detect_ngpus() -> int:
+    from dragon.telemetry.collector import identify_gpu
+
+    _vendor, count = identify_gpu()
+    # AMD's identify_gpu returns a list; normalize to length.
+    return len(count) if isinstance(count, list) else int(count or 0)
+
+
+def _has_telemetry_collector() -> bool:
+    import dragon.telemetry.collector  # noqa: F401
+
+    return True
+
+
+_NNODES = _safe(_detect_nnodes, 0)
+_NGPUS = _safe(_detect_ngpus, 0)
+_HAS_TELEM = _safe(_has_telemetry_collector, False)
+# PMI/PMIx launch failures abort the entire Dragon runtime, so we cannot
+# probe at runtime. Gate on explicit opt-in instead.
+_HAS_PMI = os.environ.get("DRAGON_CI_HAS_PMI", "").lower() not in ("", "0", "false")
+
+
+# --- Early-abort gate: must run under `dragon python` --------------------
+
+
+def pytest_configure(config):
+    """Abort early if not running under the Dragon launcher.
+
+    Without the gate, ``pytest tests/dragon_ci`` collects fine (dragon is
+    importable from the venv) but each test fails individually with cryptic
+    ``DRAGON_MEMORY_ERRNO`` channel-attach errors. The gate gives a single
+    actionable message instead.
+
+    Fast path: scan for any ``DRAGON_*`` env var — the launcher exports
+    several of them. Slow path: if env vars are present but the runtime
+    is unreachable (e.g. a stale dragon session), the behavioural probe
+    catches that.
+    """
+    dragon_envs = [k for k in os.environ if k.startswith("DRAGON_")]
+    if not dragon_envs:
+        pytest.exit(
+            "Dragon CI suite must run under the Dragon launcher:\n"
+            "    dragon python -m pytest -c tests/dragon_ci/pytest.ini "
+            "--rootdir=tests/dragon_ci tests/dragon_ci/",
+            returncode=2,
+        )
+    try:
+        _detect_nnodes()
+    except Exception as exc:
+        pytest.exit(
+            f"DRAGON_* env vars present ({len(dragon_envs)} found) but the "
+            f"Dragon runtime is unreachable — likely a stale dragon session. "
+            f"Re-launch with `dragon python -m pytest ...`.\nProbe error: {exc!r}",
+            returncode=2,
+        )
+
+
+# --- Auto-skip markers ----------------------------------------------------
+
+# (marker_name, predicate-that-must-be-true-to-run, reason)
+_SKIP_RULES = (
+    ("requires_multi_node", _NNODES >= 2, f"requires >= 2 Dragon nodes (have {_NNODES})"),
+    ("requires_gpu", _NGPUS >= 1, "requires at least one GPU"),
+    ("requires_mpi", _HAS_PMI, "set DRAGON_CI_HAS_PMI=1 to enable MPI/PMIx tests"),
+    ("requires_telemetry_collector", _HAS_TELEM, "dragon.telemetry.collector not importable"),
+)
+
+
+def pytest_collection_modifyitems(config, items):
+    for marker, ok, reason in _SKIP_RULES:
+        if ok:
+            continue
+        skip = pytest.mark.skip(reason=reason)
+        for item in items:
+            if marker in item.keywords:
+                item.add_marker(skip)
+
+
+# --- Batch fixtures -------------------------------------------------------
+
+
+def _new_batch():
+    """Construct a Batch with Dragon's required mp start method set."""
+    import multiprocessing as mp
+
+    from dragon.workflows.batch import Batch
+
+    try:
+        if mp.get_start_method(allow_none=True) != "dragon":
+            mp.set_start_method("dragon", force=True)
+    except RuntimeError:
+        pass
+    return Batch(disable_telem=True)
+
+
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
+async def batch():
+    """One Batch shared across the session.
+
+    Async-fixture so construction happens inside an asyncio loop, mirroring
+    Rhapsody's own V3 tests. Tests must NOT kill workers or call ``close()``
+    on this Batch — use ``fresh_batch`` for that.
+    """
+    b = _new_batch()
+    try:
+        yield b
+    finally:
+        try:
+            b.join(timeout=30.0)
+        except Exception:
+            pass
+
+
+@pytest_asyncio.fixture(loop_scope="session")
+async def fresh_batch():
+    """A throwaway Batch for tests that intentionally kill workers — the session ``batch`` cannot
+    recover once a worker is SIGKILL'd.
+
+    Skips the graceful ``join()`` because dead workers make it block for
+    the full timeout; ``terminate()`` can also raise ``DragonUserCodeError``
+    on dead workers — both are swallowed.
+    """
+    b = _new_batch()
+    try:
+        yield b
+    finally:
+        try:
+            b.terminate()
+        except Exception:
+            pass
diff --git a/tests/dragon_ci/pytest.ini b/tests/dragon_ci/pytest.ini
new file mode 100644
index 0000000..4656f0a
--- /dev/null
+++ b/tests/dragon_ci/pytest.ini
@@ -0,0 +1,14 @@
+[pytest]
+testpaths = .
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*
+addopts = --tb=short -ra
+asyncio_mode = auto
+asyncio_default_fixture_loop_scope = session
+asyncio_default_test_loop_scope = session
+markers =
+    requires_multi_node: needs an allocation with >= 2 Dragon nodes
+    requires_gpu: needs at least one GPU visible to Dragon's telemetry collector
+    requires_mpi: needs DRAGON_CI_HAS_PMI=1 (real MPI launch)
+    requires_telemetry_collector: needs dragon.telemetry.collector importable
diff --git a/tests/dragon_ci/test_async_in_batch.py b/tests/dragon_ci/test_async_in_batch.py
new file mode 100644
index 0000000..71a560c
--- /dev/null
+++ b/tests/dragon_ci/test_async_in_batch.py
@@ -0,0 +1,38 @@
+"""Contract: async-coroutine target dispatch via ``Batch.function``.
+
+Rhapsody V3's ``build_task`` detects ``asyncio.iscoroutinefunction(target)``
+and wraps the coroutine in a ``asyncio.run(...)`` shim before passing it to
+``batch.function``. This test pins that pattern, and also verifies the
+underlying primitives Rhapsody depends on:
+
+- ``asyncio.iscoroutinefunction`` recognises the user's function,
+- ``Batch.function`` accepts a plain (non-coroutine) callable,
+- a coroutine result reaches ``batch.results_ddict[uid]`` unchanged after
+  the wrapper applies ``asyncio.run``.
+
+Helpers live in ``_dragon_ci_helpers`` — see the conftest module docstring
+for why callables passed to Batch tasks cannot be defined in this file.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from _dragon_ci_helpers import async_double  # noqa: E402
+from _dragon_ci_helpers import async_run_shim  # noqa: E402
+from dragon.workflows.batch import Batch
+
+
+def test_iscoroutinefunction_detects_async_def():
+    assert asyncio.iscoroutinefunction(async_double)
+    assert not asyncio.iscoroutinefunction(async_run_shim)
+
+
+def test_async_result_reaches_results_ddict(batch: Batch):
+    """The shimmed coroutine result must surface as a plain int in the DDict."""
+    task = batch.function(async_run_shim, 21)
+    assert task.get(timeout=60.0) == 42
+
+    result, _tb, raised, _stdout, _stderr = batch.results_ddict[task.uid]
+    assert raised is False
+    assert result == 42
diff --git a/tests/dragon_ci/test_batch_function.py b/tests/dragon_ci/test_batch_function.py
new file mode 100644
index 0000000..35ca79a
--- /dev/null
+++ b/tests/dragon_ci/test_batch_function.py
@@ -0,0 +1,89 @@
+"""Contract: ``Batch.function()`` mode and the results-DDict tuple shape.
+
+Rhapsody's V3 backend reads results directly from the Batch-owned DDict::
+
+    result, tb, raised, stdout, stderr = batch.results_ddict[task.uid]
+
+so the tuple layout, the truthiness of ``raised``, and the types of
+``stdout``/``stderr`` are all part of the contract. ``Function.get()`` is the
+documented public alternative and is also pinned.
+
+Helpers live in ``_dragon_ci_helpers`` — see the conftest module docstring
+for why callables passed to Batch tasks cannot be defined in this file.
+"""
+
+from __future__ import annotations
+
+import pytest
+from _dragon_ci_helpers import add  # noqa: E402  (must be importable from workers)
+from _dragon_ci_helpers import kwfn  # noqa: E402  (must be importable from workers)
+from _dragon_ci_helpers import print_and_return  # noqa: E402  (must be importable from workers)
+from _dragon_ci_helpers import raise_value_error  # noqa: E402  (must be importable from workers)
+from _dragon_ci_helpers import slow_double  # noqa: E402  (must be importable from workers)
+from dragon.workflows.batch import Batch
+from dragon.workflows.batch.batch import Function
+from dragon.workflows.batch.batch import TaskNotReadyError
+
+
+def test_function_returns_function_handle(batch: Batch):
+    """``batch.function()`` returns a ``Function`` with ``.uid`` and ``.get``."""
+    task = batch.function(add, 2, 3)
+    assert isinstance(task, Function)
+    assert task.uid and callable(task.get)
+    assert task.get(timeout=60.0) == 5
+
+
+def test_function_get_reraises_exception(batch: Batch):
+    task = batch.function(raise_value_error, "intentional failure")
+    with pytest.raises(ValueError, match="intentional failure"):
+        task.get(timeout=60.0)
+
+
+def test_function_supports_kwargs(batch: Batch):
+    """``Batch.function`` must forward ``**kwargs`` to the target."""
+    assert batch.function(kwfn, 6, mult=7).get(timeout=60.0) == 42
+
+
+def test_function_get_non_blocking_raises_when_not_ready(batch: Batch):
+    """``Function.get(block=False)`` raises ``TaskNotReadyError`` pre-completion."""
+    task = batch.function(slow_double, 21, 1.0)
+    with pytest.raises(TaskNotReadyError):
+        task.get(block=False)
+    assert task.get(timeout=60.0) == 42
+
+
+def test_results_ddict_five_tuple_on_success(batch: Batch):
+    """Rhapsody unpacks exactly five fields.
+
+    Order and shape are the contract.
+    """
+    task = batch.function(add, 1, 1)
+    task.get(timeout=60.0)
+    entry = batch.results_ddict[task.uid]
+    assert isinstance(entry, tuple) and len(entry) == 5, (
+        f"results_ddict tuple shape changed: {entry!r}"
+    )
+    result, tb, raised, stdout, stderr = entry
+    assert result == 2
+    assert tb is None and raised is False
+    assert isinstance(stdout, str) and isinstance(stderr, str)
+
+
+def test_results_ddict_five_tuple_on_failure(batch: Batch):
+    """On exception: ``raised=True``, ``result`` holds the exception, ``tb`` is str."""
+    task = batch.function(raise_value_error, "boom")
+    with pytest.raises(ValueError):
+        task.get(timeout=60.0)
+    result, tb, raised, stdout, stderr = batch.results_ddict[task.uid]
+    assert raised is True
+    assert isinstance(result, BaseException)
+    assert isinstance(tb, str) and "ValueError" in tb
+    assert isinstance(stdout, str) and isinstance(stderr, str)
+
+
+def test_function_captures_stdout_and_stderr(batch: Batch):
+    task = batch.function(print_and_return, 7, "hello-from-function")
+    assert task.get(timeout=60.0) == 7
+    _r, _tb, _raised, stdout, stderr = batch.results_ddict[task.uid]
+    assert "hello-from-function" in stdout
+    assert "err msg" in stderr
diff --git a/tests/dragon_ci/test_batch_job.py b/tests/dragon_ci/test_batch_job.py
new file mode 100644
index 0000000..2dc7fcb
--- /dev/null
+++ b/tests/dragon_ci/test_batch_job.py
@@ -0,0 +1,57 @@
+"""Contract: ``Batch.job()`` — multi-rank / MPI launch mode.
+
+Rhapsody V3 builds jobs as a list of ``(nranks, ProcessTemplate)`` tuples.
+Wire-only checks always run; actual launches are guarded by ``requires_mpi``
+(opt-in via ``DRAGON_CI_HAS_PMI=1``) because a failing PMIx init aborts
+the whole Dragon runtime, not just the test.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import pytest
+from dragon.infrastructure.facts import PMIBackend
+from dragon.native.process import ProcessTemplate
+from dragon.workflows.batch import Batch
+from dragon.workflows.batch.batch import Job
+
+
+@pytest.mark.parametrize("kwarg", ["process_templates", "name", "timeout", "pmi"])
+def test_batch_job_kwarg(kwarg):
+    assert kwarg in inspect.signature(Batch.job).parameters, (
+        f"Batch.job({kwarg}=) removed — Rhapsody depends on it"
+    )
+
+
+def test_pmi_backend_has_pmix():
+    """Rhapsody passes ``PMIBackend.PMIX`` as the cross-vendor portable backend."""
+    assert "PMIX" in {m.name for m in PMIBackend}
+
+
+@pytest.mark.requires_mpi
+@pytest.mark.parametrize("nranks", [1, 2])
+def test_batch_job_launch(batch: Batch, nranks):
+    """A real PMIx launch must produce a ``Job`` handle and complete cleanly.
+
+    Skips on hosts where PMIx is unavailable (documented failure mode is
+    ``RuntimeError: Unable to initialize PMIx server`` from
+    ``DragonPMIxJob.__init__``).
+    """
+    try:
+        job = batch.job(
+            [(nranks, ProcessTemplate("/bin/true", args=()))],
+            name=f"dragon-ci-{nranks}rank",
+            pmi=PMIBackend.PMIX,
+        )
+        job.get(timeout=120.0)
+    except RuntimeError as exc:
+        # Only skip for a genuine PMIx-unavailable host; let other RuntimeErrors
+        # (and any TypeError/AttributeError/etc.) surface as real failures.
+        if "pmi" in str(exc).lower():
+            pytest.skip(f"PMIX launch unavailable on this host: {exc!r}")
+        raise
+
+    assert isinstance(job, Job)
+    result, tb, raised, _stdout, _stderr = batch.results_ddict[job.uid]
+    assert raised is False, f"{nranks}-rank job failed: result={result!r} tb={tb!r}"
diff --git a/tests/dragon_ci/test_batch_lifecycle.py b/tests/dragon_ci/test_batch_lifecycle.py
new file mode 100644
index 0000000..6dcf857
--- /dev/null
+++ b/tests/dragon_ci/test_batch_lifecycle.py
@@ -0,0 +1,70 @@
+"""Contract: ``dragon.workflows.batch.Batch`` construction, attributes, teardown.
+
+Rhapsody's V3 backend constructs ``Batch`` with the kwargs below, reads
+``num_workers`` / ``num_managers`` for logging, and shuts down via
+``close()`` then ``join(timeout=...)`` with ``terminate()`` as a fallback.
+``fence()`` is exposed to user code as a barrier.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import pytest
+from dragon.workflows.batch import Batch
+
+_BATCH_CTOR_KWARGS = (
+    "num_nodes",
+    "pool_nodes",
+    "disable_telem",
+    "scheduler_workers",
+    "results_ddict_mem",
+)
+_BATCH_METHODS = ("function", "process", "job", "fence", "close", "join", "terminate")
+
+
+@pytest.mark.parametrize("kwarg", _BATCH_CTOR_KWARGS)
+def test_batch_ctor_kwarg(kwarg):
+    assert kwarg in inspect.signature(Batch.__init__).parameters, (
+        f"Batch.__init__ no longer accepts {kwarg!r}"
+    )
+
+
+@pytest.mark.parametrize("method", _BATCH_METHODS)
+def test_batch_method_present(method):
+    assert callable(getattr(Batch, method, None)), f"Batch.{method}() removed"
+
+
+def test_batch_num_workers_and_num_managers(batch):
+    """Rhapsody logs ``batch.num_workers`` and ``batch.num_managers`` at startup."""
+    assert isinstance(batch.num_workers, int) and batch.num_workers >= 1
+    assert isinstance(batch.num_managers, int) and batch.num_managers >= 1
+
+
+def test_batch_results_ddict_supports_lookup(batch):
+    """Rhapsody reads ``batch.results_ddict[task.uid]`` directly."""
+    rd = batch.results_ddict
+    assert hasattr(rd, "__contains__") and hasattr(rd, "__getitem__")
+
+
+def test_batch_fence_callable_on_idle_batch(batch):
+    """Rhapsody exposes ``Batch.fence()`` to user code as a barrier.
+
+    On an idle Batch it must complete without raising.
+    """
+    batch.fence()
+
+
+def test_batch_close_join_terminate_lifecycle():
+    """A throwaway Batch must accept close()+join()+terminate() without raising.
+
+    Combined into one test so we only pay one Batch startup. Note ``close()``
+    is deprecated as of Dragon 0.14 (no-op); only ``join()`` is real teardown.
+    """
+    b = Batch(disable_telem=True)
+    try:
+        b.close()
+        b.join(timeout=30.0)
+    finally:
+        # Always terminate so a raising/timed-out join() can't leak the Batch.
+        b.terminate()
diff --git a/tests/dragon_ci/test_batch_process.py b/tests/dragon_ci/test_batch_process.py
new file mode 100644
index 0000000..1038129
--- /dev/null
+++ b/tests/dragon_ci/test_batch_process.py
@@ -0,0 +1,82 @@
+"""Contract: ``Batch.process()`` mode and ``ProcessTemplate`` plumbing.
+
+Rhapsody V3 launches subprocess tasks via::
+
+    batch.process(ProcessTemplate(target, args=(...), cwd=..., policy=...))
+
+Key contract pinned by these tests: **a bare ``ProcessTemplate`` (no
+``stdout=Popen.PIPE``) does NOT capture child stdout/stderr into the
+``results_ddict`` 5-tuple — those slots come back as empty strings.**
+Rhapsody works around this by wrapping the command in a shell script that
+redirects to files (see the V3 ``capture_stdio`` code path).
+"""
+
+from __future__ import annotations
+
+import pytest
+from dragon.native.process import ProcessTemplate
+from dragon.workflows.batch import Batch
+
+
+def test_process_returns_five_tuple(batch: Batch):
+    """A trivial subprocess task completes and yields the 5-tuple shape."""
+    task = batch.process(ProcessTemplate("/bin/true", args=()))
+    entry = batch.results_ddict[task.uid]
+    assert isinstance(entry, tuple) and len(entry) == 5, (
+        f"results_ddict shape changed for batch.process: {entry!r}"
+    )
+    _result, _tb, raised, _stdout, _stderr = entry
+    assert raised is False
+
+
+def test_process_default_stdio_is_empty(batch: Batch):
+    """Without ``stdout=Popen.PIPE``, the stdout/stderr slots are empty strings.
+
+    If Dragon ever starts capturing by default, Rhapsody's ``capture_stdio``
+    workaround can be retired — this test will fail then.
+    """
+    task = batch.process(ProcessTemplate("/bin/echo", args=("hello-process",)))
+    _r, _tb, _raised, stdout, stderr = batch.results_ddict[task.uid]
+    assert stdout == "" and stderr == "", (
+        f"Dragon may now capture stdio by default: stdout={stdout!r} stderr={stderr!r}"
+    )
+
+
+def test_process_template_args_none_raises_typeerror(batch: Batch):
+    """``ProcessTemplate(...).args`` defaults to None, which makes ``Batch.process`` raise
+    ``TypeError: 'NoneType' object is not iterable``.
+
+    Pass ``args=()`` to avoid it.
+    """
+    with pytest.raises(TypeError, match="NoneType"):
+        batch.process(ProcessTemplate("/bin/true"))
+
+
+def test_process_non_zero_exit_is_visible(batch: Batch):
+    """``/bin/false`` (exit 1) must surface failure through the 5-tuple."""
+    task = batch.process(ProcessTemplate("/bin/false", args=()))
+    result, tb, raised, _stdout, _stderr = batch.results_ddict[task.uid]
+    # Dragon may store the failure as ``raised=True`` or as a non-zero result.
+    failure_visible = raised or bool(tb) or result not in (None, 0, True)
+    assert failure_visible, (
+        "Non-zero exit produced a clean-looking tuple — cannot distinguish failure"
+    )
+
+
+def test_process_capture_stdio_via_shell_redirect(batch: Batch, tmp_path):
+    """Pins Rhapsody's ``capture_stdio`` workaround.
+
+    A wrapper bash script redirects its own stdout/stderr to files, then runs the real command.
+    Dragon launches the bash; the files end up with the captured output regardless of Dragon's PIPE
+    behaviour.
+    """
+    stdout_path = tmp_path / "task.stdout"
+    script_path = tmp_path / "task.sh"
+    script_path.write_text(f"#!/usr/bin/bash\n/bin/echo 'stdout-from-wrapped' 1>{stdout_path}\n")
+    script_path.chmod(0o755)
+
+    task = batch.process(ProcessTemplate("/bin/bash", args=(str(script_path),)))
+    _ = batch.results_ddict[task.uid]  # wait for completion
+
+    assert stdout_path.exists(), "wrapper script did not produce stdout file"
+    assert "stdout-from-wrapped" in stdout_path.read_text()
diff --git a/tests/dragon_ci/test_ddict.py b/tests/dragon_ci/test_ddict.py
new file mode 100644
index 0000000..b0bd057
--- /dev/null
+++ b/tests/dragon_ci/test_ddict.py
@@ -0,0 +1,97 @@
+"""Contract: ``dragon.data.ddict.ddict.DDict`` core surface.
+
+Rhapsody V3 reads ``batch.results_ddict``; downstream users construct
+DDicts directly via the kwargs and methods checked below.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import pytest
+from dragon.data.ddict.ddict import DDict
+
+_DDICT_KWARGS = (
+    "managers_per_node",
+    "n_nodes",
+    "total_mem",
+    "working_set_size",
+    "wait_for_keys",
+    "policy",
+    "timeout",
+)
+_DDICT_METHODS = ("pput", "bput", "clear", "destroy", "detach", "keys", "items", "freeze")
+
+
+@pytest.mark.parametrize("kwarg", _DDICT_KWARGS)
+def test_ddict_ctor_kwarg(kwarg):
+    assert kwarg in inspect.signature(DDict.__init__).parameters, (
+        f"DDict.__init__ no longer accepts {kwarg!r}"
+    )
+
+
+@pytest.mark.parametrize("method", _DDICT_METHODS)
+def test_ddict_method_present(method):
+    assert callable(getattr(DDict, method, None)), f"DDict.{method}() removed"
+
+
+@pytest.fixture(scope="module")
+def _shared_ddict():
+    """One DDict built once per module — tear down on exit."""
+    d = DDict(
+        n_nodes=1,
+        managers_per_node=1,
+        total_mem=4 * 1024 * 1024,
+        wait_for_keys=False,
+        working_set_size=1,
+    )
+    try:
+        yield d
+    finally:
+        d.destroy()
+
+
+@pytest.fixture
+def ddict(_shared_ddict):
+    """Per-test view on the shared DDict, cleared each call."""
+    _shared_ddict.clear()
+    return _shared_ddict
+
+
+def test_ddict_setitem_getitem_contains(ddict):
+    ddict["a"] = 1
+    assert "a" in ddict and ddict["a"] == 1
+
+
+def test_ddict_pput(ddict):
+    ddict.pput("k", {"nested": [1, 2, 3]})
+    assert ddict["k"] == {"nested": [1, 2, 3]}
+
+
+def test_ddict_delitem(ddict):
+    ddict["k"] = "v"
+    del ddict["k"]
+    assert "k" not in ddict
+
+
+def test_ddict_clear_removes_all(ddict):
+    ddict["x"] = 1
+    ddict["y"] = 2
+    ddict.clear()
+    assert "x" not in ddict and "y" not in ddict
+
+
+def test_ddict_keyerror_on_missing(ddict):
+    with pytest.raises(KeyError):
+        _ = ddict["missing"]
+
+
+def test_ddict_instance_has_is_frozen(ddict):
+    """``is_frozen`` lives on the instance (not on the class) in Dragon 0.14."""
+    assert hasattr(ddict, "is_frozen")
+
+
+def test_ddict_round_trip_complex_payload(ddict):
+    payload = {"list": [1, 2, 3], "tuple": (4, 5), "set": {6, 7}, "nested": {"a": [(0, "z")]}}
+    ddict["complex"] = payload
+    assert ddict["complex"] == payload
diff --git a/tests/dragon_ci/test_ddict_unknown_key_blocks.py b/tests/dragon_ci/test_ddict_unknown_key_blocks.py
new file mode 100644
index 0000000..b077348
--- /dev/null
+++ b/tests/dragon_ci/test_ddict_unknown_key_blocks.py
@@ -0,0 +1,56 @@
+"""Documented Dragon bug — **test disabled**.
+
+A DDict built with ``wait_for_keys=True`` (the setting Batch's results-DDict
+uses) **blocks indefinitely** when a never-written key is read, instead of
+raising ``KeyError``. Rhapsody's V3 monitor loop has a
+``try/except KeyError: continue`` polling pattern that depends on the
+expected KeyError-on-missing contract — when that contract isn't honored,
+the monitor loop deadlocks on the first never-arriving result.
+
+This is the silent-hang failure mode that prompted the entire test suite.
+
+The test below is **skipped** because letting it hit ``pytest.mark.timeout``
+produces a hard failure that doesn't compose with ``pytest.mark.xfail``
+(pytest-timeout raises a ``BaseException`` subclass that xfail does not
+catch). To check whether Dragon has fixed the bug, remove the
+``pytest.mark.skip`` decorator and run this file under
+``dragon python -m pytest`` — the test will either return cleanly (Dragon
+fixed it, re-enable permanently) or pytest-timeout will fire (still broken).
+"""
+
+import warnings
+
+import pytest
+from dragon.data.ddict.ddict import DDict
+
+warnings.warn(
+    "test_ddict_get_unknown_key_blocks_forever is DISABLED — Dragon bug "
+    "(DDict[unknown_key] blocks instead of raising KeyError when "
+    "wait_for_keys=True) is not actively checked. Re-enable to retest.",
+    UserWarning,
+    stacklevel=2,
+)
+
+
+@pytest.mark.skip(
+    reason="DISABLED — Dragon bug: DDict[unknown_key] blocks instead of raising "
+    "KeyError when wait_for_keys=True. Re-enable to check if fixed."
+)
+@pytest.mark.timeout(10, method="thread")
+def test_ddict_get_unknown_key_blocks_forever():
+    # wait_for_keys=True requires working_set_size > 1 (Dragon enforces it).
+    d = DDict(
+        n_nodes=1,
+        managers_per_node=1,
+        total_mem=4 * 1024 * 1024,
+        wait_for_keys=True,
+        working_set_size=2,
+    )
+    try:
+        with pytest.raises(KeyError):
+            _ = d["never-written"]
+    finally:
+        try:
+            d.destroy()
+        except Exception:
+            pass
diff --git a/tests/dragon_ci/test_machine_system.py b/tests/dragon_ci/test_machine_system.py
new file mode 100644
index 0000000..b7e4250
--- /dev/null
+++ b/tests/dragon_ci/test_machine_system.py
@@ -0,0 +1,45 @@
+"""Contract: ``dragon.native.machine.System`` shape and node enumeration.
+
+Rhapsody's telemetry adapter calls ``System().hostname_policies()`` to obtain
+one ``Policy`` per node, then pins one worker per Policy.
+"""
+
+from __future__ import annotations
+
+import pytest
+from dragon.infrastructure.policy import Policy
+from dragon.native.machine import System
+
+
+@pytest.mark.parametrize("attr", ["nnodes", "hostname_policies"])
+def test_system_attribute_present(attr):
+    """Rhapsody reads ``System().nnodes`` (V1 DDict sizing) and calls ``hostname_policies()``
+    (telemetry adapter, one worker per node)."""
+    assert hasattr(System(), attr), f"System.{attr} removed"
+
+
+def test_system_nnodes_is_int_ge_1():
+    assert isinstance(System().nnodes, int) and System().nnodes >= 1
+
+
+def test_hostname_policies_returns_one_policy_per_node():
+    s = System()
+    policies = s.hostname_policies()
+    assert isinstance(policies, list)
+    assert len(policies) == s.nnodes
+    assert all(isinstance(p, Policy) for p in policies)
+
+
+def test_hostname_policies_identify_each_node():
+    """Each per-node Policy must carry an identifier (host_name or host_id)."""
+    for p in System().hostname_policies():
+        assert p.host_name or p.host_id >= 0 or p.placement == Policy.Placement.HOST_NAME, (
+            f"hostname_policies returned a Policy with no node identifier: {p}"
+        )
+
+
+@pytest.mark.requires_multi_node
+def test_hostname_policies_are_distinct():
+    policies = System().hostname_policies()
+    keys = [(p.host_name, p.host_id) for p in policies]
+    assert len(set(keys)) == len(policies), f"duplicate identifiers: {keys}"
diff --git a/tests/dragon_ci/test_policy.py b/tests/dragon_ci/test_policy.py
new file mode 100644
index 0000000..20aacf1
--- /dev/null
+++ b/tests/dragon_ci/test_policy.py
@@ -0,0 +1,60 @@
+"""Contract: ``dragon.infrastructure.policy.Policy`` shape.
+
+Rhapsody V3 builds Policy instances with combinations of ``host_id``,
+``distribution``, ``placement``, ``host_name``, and ``gpu_affinity``.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import pytest
+from dragon.infrastructure.policy import Policy
+
+_POLICY_KWARGS = (
+    "placement",
+    "host_name",
+    "host_id",
+    "distribution",
+    "cpu_affinity",
+    "gpu_affinity",
+    "wait_mode",
+)
+
+
+@pytest.mark.parametrize("kwarg", _POLICY_KWARGS)
+def test_policy_ctor_kwarg(kwarg):
+    assert kwarg in inspect.signature(Policy.__init__).parameters, (
+        f"Policy.__init__ no longer accepts {kwarg!r}"
+    )
+
+
+@pytest.mark.parametrize("attr", _POLICY_KWARGS)
+def test_policy_attribute_present(attr):
+    assert hasattr(Policy(), attr), f"Policy().{attr} attribute removed"
+
+
+@pytest.mark.parametrize(
+    "enum, required",
+    [
+        (Policy.Distribution, ("BLOCK", "ROUNDROBIN", "DEFAULT")),
+        (Policy.Placement, ("DEFAULT", "HOST_NAME", "HOST_ID", "ANYWHERE", "LOCAL")),
+    ],
+)
+def test_policy_enum_members(enum, required):
+    members = {m.name for m in enum}
+    missing = set(required) - members
+    assert not missing, f"{enum.__name__} missing members: {missing}; got {members}"
+
+
+def test_policy_with_host_id_and_block_distribution():
+    p = Policy(host_id=0, distribution=Policy.Distribution.BLOCK)
+    assert p.host_id == 0
+    assert p.distribution == Policy.Distribution.BLOCK
+
+
+def test_policy_with_placement_host_name_and_gpu_affinity():
+    p = Policy(placement=Policy.Placement.HOST_NAME, host_name="x", gpu_affinity=[0, 1])
+    assert p.placement == Policy.Placement.HOST_NAME
+    assert p.host_name == "x"
+    assert p.gpu_affinity == [0, 1]
diff --git a/tests/dragon_ci/test_process_group.py b/tests/dragon_ci/test_process_group.py
new file mode 100644
index 0000000..bbb5a77
--- /dev/null
+++ b/tests/dragon_ci/test_process_group.py
@@ -0,0 +1,69 @@
+"""Contract: native ``ProcessGroup`` lifecycle.
+
+The telemetry adapter spawns one worker per node via::
+
+    grp = ProcessGroup(restart=False, pmi=None)
+    grp.add_process(nproc=1, template=ProcessTemplate(target=worker_fn, ...))
+    grp.init(); grp.start(); ...; grp.join(timeout=...); grp.close()
+
+and treats ``DragonUserCodeError`` as an expected exception class on
+graceful shutdown.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import pytest
+from _dragon_ci_helpers import pg_worker  # noqa: E402
+from dragon.native.event import Event
+from dragon.native.process import ProcessTemplate
+from dragon.native.process_group import DragonUserCodeError
+from dragon.native.process_group import ProcessGroup
+from dragon.native.queue import Queue
+
+
+@pytest.mark.parametrize("kwarg", ["restart", "pmi", "policy"])
+def test_process_group_ctor_kwarg(kwarg):
+    assert kwarg in inspect.signature(ProcessGroup.__init__).parameters, (
+        f"ProcessGroup.__init__ no longer accepts {kwarg!r}"
+    )
+
+
+@pytest.mark.parametrize("method", ["add_process", "init", "start", "join", "stop", "close"])
+def test_process_group_method_present(method):
+    assert callable(getattr(ProcessGroup, method, None)), f"ProcessGroup.{method}() removed"
+
+
+@pytest.mark.parametrize("kwarg", ["nproc", "template"])
+def test_process_group_add_process_kwarg(kwarg):
+    assert kwarg in inspect.signature(ProcessGroup.add_process).parameters, (
+        f"ProcessGroup.add_process no longer accepts {kwarg!r}"
+    )
+
+
+def test_dragon_user_code_error_importable():
+    """Rhapsody catches ``DragonUserCodeError`` during graceful shutdown."""
+    assert isinstance(DragonUserCodeError, type)
+    assert issubclass(DragonUserCodeError, BaseException)
+
+
+def test_process_group_lifecycle_smoke():
+    """Build, init, start, join, close a 1-worker group end-to-end."""
+    q = Queue(maxsize=10)
+    shutdown = Event()
+
+    grp = ProcessGroup(restart=False, pmi=None)
+    grp.add_process(nproc=1, template=ProcessTemplate(target=pg_worker, args=(q, shutdown)))
+    grp.init()
+    grp.start()
+
+    try:
+        msg = q.get(timeout=30.0)
+        assert msg.get("pid") and msg.get("host")
+    finally:
+        # Guarantee the worker is told to stop and the group is reaped even if
+        # q.get() times out or the assertion fails, else pg_worker loops forever.
+        shutdown.set()
+        grp.join(timeout=30.0)
+        grp.close()
diff --git a/tests/dragon_ci/test_process_template.py b/tests/dragon_ci/test_process_template.py
new file mode 100644
index 0000000..4354f9a
--- /dev/null
+++ b/tests/dragon_ci/test_process_template.py
@@ -0,0 +1,91 @@
+"""Contract: ``dragon.native.process.ProcessTemplate`` constructor and attributes.
+
+Rhapsody V3 builds ProcessTemplate objects with these kwargs::
+
+    ProcessTemplate(target, args=(...), kwargs={...}, cwd=..., env=...,
+                    policy=Policy(...), stdin=Popen.DEVNULL,
+                    stdout=Popen.PIPE, stderr=Popen.PIPE)
+
+and later introspects ``pt.cwd``, ``pt.policy``, ``pt.env`` and unpacks
+``pt.argdata`` via ``cloudpickle.loads`` to verify ``(target, args, kwargs)``
+were preserved.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import cloudpickle
+import pytest
+from dragon.infrastructure.policy import Policy
+from dragon.native.process import Popen
+from dragon.native.process import ProcessTemplate
+
+_TEMPLATE_KWARGS = (
+    "target",
+    "args",
+    "kwargs",
+    "cwd",
+    "env",
+    "stdin",
+    "stdout",
+    "stderr",
+    "policy",
+)
+
+
+@pytest.mark.parametrize("kwarg", _TEMPLATE_KWARGS)
+def test_process_template_ctor_kwarg(kwarg):
+    assert kwarg in inspect.signature(ProcessTemplate.__init__).parameters, (
+        f"ProcessTemplate.__init__ no longer accepts {kwarg!r}"
+    )
+
+
+@pytest.mark.parametrize("name", ["PIPE", "DEVNULL"])
+def test_popen_stdio_sentinel(name):
+    """Rhapsody uses ``Popen.PIPE`` and ``Popen.DEVNULL`` as stream sentinels."""
+    assert hasattr(Popen, name), f"Popen.{name} constant removed"
+
+
+def test_template_cwd_attribute():
+    pt = ProcessTemplate("/bin/echo", args=("x",), cwd="/tmp")
+    assert pt.cwd == "/tmp"
+
+
+def test_template_policy_identity_preserved():
+    """``pt.policy is policy`` — Rhapsody's own tests rely on the same identity."""
+    policy = Policy(gpu_affinity=[0, 1, 2, 3])
+    pt = ProcessTemplate("/bin/echo", args=("x",), policy=policy)
+    assert pt.policy is policy
+    assert pt.policy.gpu_affinity == [0, 1, 2, 3]
+
+
+def test_template_env_dict_preserved():
+    pt = ProcessTemplate("/usr/bin/env", args=(), env={"FOO": "BAR"})
+    assert pt.env.get("FOO") == "BAR"
+
+
+def test_template_stdio_sentinels_accepted():
+    """PIPE/DEVNULL must be acceptable to the stdin/stdout/stderr kwargs."""
+    ProcessTemplate(
+        "/bin/echo",
+        args=("x",),
+        stdin=Popen.DEVNULL,
+        stdout=Popen.PIPE,
+        stderr=Popen.PIPE,
+    )
+
+
+def test_template_argdata_round_trips_via_cloudpickle():
+    """Rhapsody round-trips ``(target, args, kwargs)`` out of ``pt.argdata``."""
+
+    def _target(x, y):
+        return x + y
+
+    pt = ProcessTemplate(_target, args=(1, 2), kwargs={"z": 3})
+    target, stored_args, stored_kwargs = cloudpickle.loads(pt.argdata)
+    # ``target`` is reconstituted across cloudpickle, so identity won't hold —
+    # compare behaviour instead.
+    assert target(1, 2) == _target(1, 2)
+    assert tuple(stored_args) == (1, 2)
+    assert stored_kwargs == {"z": 3}
diff --git a/tests/dragon_ci/test_queue_event.py b/tests/dragon_ci/test_queue_event.py
new file mode 100644
index 0000000..621a1b2
--- /dev/null
+++ b/tests/dragon_ci/test_queue_event.py
@@ -0,0 +1,72 @@
+"""Contract: ``dragon.native.queue.Queue`` and ``dragon.native.event.Event``.
+
+The telemetry adapter uses these as the cross-process result channel and
+shutdown signal: ``Queue(maxsize=)`` with ``put(timeout=)``/``get(timeout=)``,
+``Event`` with ``wait(timeout=)``/``set()``.
+"""
+
+from __future__ import annotations
+
+import inspect
+import time
+
+import pytest
+from _dragon_ci_helpers import pg_worker  # noqa: E402
+from dragon.native.event import Event
+from dragon.native.process import ProcessTemplate
+from dragon.native.process_group import ProcessGroup
+from dragon.native.queue import Queue
+
+
+def test_queue_ctor_accepts_maxsize():
+    assert "maxsize" in inspect.signature(Queue.__init__).parameters
+
+
+@pytest.mark.parametrize("method", ["put", "get", "close"])
+def test_queue_method_present(method):
+    assert callable(getattr(Queue, method, None)), f"Queue.{method}() removed"
+
+
+@pytest.mark.parametrize("method", ["wait", "set", "clear", "is_set"])
+def test_event_method_present(method):
+    assert callable(getattr(Event, method, None)), f"Event.{method}() removed"
+
+
+def test_queue_put_get_roundtrip_in_process():
+    q = Queue(maxsize=4)
+    q.put({"k": "v"}, timeout=5.0)
+    assert q.get(timeout=5.0) == {"k": "v"}
+
+
+def test_event_wait_returns_truthy_when_set():
+    e = Event()
+    assert not e.is_set()
+    e.set()
+    assert e.is_set()
+    assert e.wait(timeout=1.0)
+
+
+def test_event_wait_returns_falsy_on_timeout():
+    e = Event()
+    t0 = time.monotonic()
+    rv = e.wait(timeout=0.2)
+    assert not rv, f"Event.wait should be falsy on timeout, got {rv!r}"
+    assert time.monotonic() - t0 >= 0.15, "Event.wait returned before its timeout"
+
+
+def test_queue_event_cross_process_roundtrip():
+    """A ProcessGroup worker pushes to a Queue and observes an Event."""
+    q = Queue(maxsize=8)
+    ev = Event()
+
+    grp = ProcessGroup(restart=False, pmi=None)
+    grp.add_process(nproc=1, template=ProcessTemplate(target=pg_worker, args=(q, ev)))
+    grp.init()
+    grp.start()
+    try:
+        msg = q.get(timeout=30.0)
+        assert msg.get("pid") and msg.get("host")
+    finally:
+        ev.set()
+        grp.join(timeout=30.0)
+        grp.close()
diff --git a/tests/dragon_ci/test_sigkill_in_worker_hangs.py b/tests/dragon_ci/test_sigkill_in_worker_hangs.py
new file mode 100644
index 0000000..cf23693
--- /dev/null
+++ b/tests/dragon_ci/test_sigkill_in_worker_hangs.py
@@ -0,0 +1,44 @@
+"""Documented Dragon bug — **test disabled**.
+
+``SIGKILL`` inside a function task denies Dragon a clean exit signal, so
+the result-DDict entry is never written and the dispatcher has no way to
+learn the worker died. ``task.get()`` then blocks indefinitely — its
+``timeout=`` kwarg only governs manager selection, not the blocking
+DDict read.
+
+The test below is **skipped** because letting it hit ``pytest.mark.timeout``
+produces a hard failure that doesn't compose with ``pytest.mark.xfail``
+(pytest-timeout raises a ``BaseException`` subclass that xfail does not
+catch). To check whether Dragon has fixed the bug, remove the
+``pytest.mark.skip`` decorator and run this file under
+``dragon python -m pytest`` — the test will either return cleanly (Dragon
+fixed it, re-enable permanently) or pytest-timeout will fire (still broken).
+
+Lives in its own file because ``method="thread"`` leaks the test thread
+(still wedged inside Dragon's blocking read) and the leaked thread holds
+resources that break any subsequent test in the same pytest invocation.
+"""
+
+import warnings
+
+import pytest
+from _dragon_ci_helpers import fn_kill_self  # noqa: E402
+from dragon.workflows.batch import Batch
+
+warnings.warn(
+    "test_sigkill_in_worker_hangs is DISABLED — Dragon bug (abnormal worker "
+    "termination is not detected; task.get() blocks indefinitely and its "
+    "timeout= kwarg is not honored on the blocking DDict read) is not "
+    "actively checked. Re-enable to retest.",
+    UserWarning,
+    stacklevel=2,
+)
+
+
+@pytest.mark.skip(
+    reason="DISABLED — Dragon bug: SIGKILL'd worker is undetected, task.get() "
+    "blocks forever, timeout= is ignored. Re-enable to check if fixed."
+)
+@pytest.mark.timeout(5, method="thread")
+def test_sigkill_in_worker_hangs(fresh_batch: Batch):
+    fresh_batch.function(fn_kill_self, 1).get(timeout=2.0)
diff --git a/tests/dragon_ci/test_telemetry_collector.py b/tests/dragon_ci/test_telemetry_collector.py
new file mode 100644
index 0000000..30dbdfc
--- /dev/null
+++ b/tests/dragon_ci/test_telemetry_collector.py
@@ -0,0 +1,61 @@
+"""Contract: ``dragon.telemetry.collector`` + ``AccVendor``.
+
+Rhapsody's ``DragonTelemetryAdapter`` imports these helpers directly to
+collect per-node GPU metrics inside a Dragon ProcessGroup worker.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import pytest
+from dragon.infrastructure.gpu_desc import AccVendor
+from dragon.infrastructure.gpu_desc import find_accelerators
+from dragon.telemetry.collector import get_amd_metrics
+from dragon.telemetry.collector import get_intel_metrics
+from dragon.telemetry.collector import get_nvidia_metrics
+from dragon.telemetry.collector import identify_gpu
+
+# All tests here need dragon.telemetry.collector — auto-skip if unavailable.
+pytestmark = pytest.mark.requires_telemetry_collector
+
+
+@pytest.mark.parametrize(
+    "fn",
+    [identify_gpu, get_amd_metrics, get_intel_metrics, get_nvidia_metrics, find_accelerators],
+)
+def test_collector_symbol_callable(fn):
+    assert callable(fn), f"{fn} no longer callable"
+
+
+@pytest.mark.parametrize("name", ["NVIDIA", "AMD", "INTEL"])
+def test_acc_vendor_member(name):
+    assert name in {m.name for m in AccVendor}, f"AccVendor.{name} removed"
+
+
+def test_identify_gpu_signature_and_return_shape():
+    assert len(inspect.signature(identify_gpu).parameters) == 0, (
+        "identify_gpu() now takes arguments"
+    )
+    vendor, _count = identify_gpu()
+    if vendor is not None:
+        assert vendor in list(AccVendor), f"unknown vendor: {vendor!r}"
+
+
+def test_dragon_telemetry_module_importable():
+    """The bare ``dragon.telemetry`` import is used as the adapter gate."""
+    import dragon.telemetry  # noqa: F401
+
+
+@pytest.mark.requires_gpu
+def test_get_nvidia_metrics_returns_metric_value_dicts():
+    """Telemetry adapter expects ``[{metric: ..., value: ...}, ...]``."""
+    vendor, _count = identify_gpu()
+    if vendor != AccVendor.NVIDIA:
+        pytest.skip(f"NVIDIA GPU expected, got {vendor!r}")
+
+    metrics = get_nvidia_metrics(0, telemetry_level=3)
+    assert metrics, "get_nvidia_metrics returned empty for present GPU"
+    for m in metrics:
+        assert isinstance(m, dict)
+        assert "metric" in m and "value" in m, f"metric dict shape changed: {m!r}"
diff --git a/tests/dragon_ci/test_unpickleable_function_hangs.py b/tests/dragon_ci/test_unpickleable_function_hangs.py
new file mode 100644
index 0000000..eaf4800
--- /dev/null
+++ b/tests/dragon_ci/test_unpickleable_function_hangs.py
@@ -0,0 +1,62 @@
+"""Documented Dragon bug — **test disabled**.
+
+A helper module made importable in the parent only via runtime
+``sys.path.insert`` (not via ``PYTHONPATH``) cannot be re-imported by the
+worker. cloudpickle pickles the helper by reference, the worker dies
+during unpickling, and ``task.get()`` blocks forever — same root cause
+as the SIGKILL hang.
+
+This is the silent-hang failure mode pytest users hit when they define a
+helper at the top of a ``test_*.py`` file and pass it to
+``batch.function()``. The conftest works around it by adding this
+directory to ``PYTHONPATH``; the test below disables that workaround
+locally to reproduce the bug.
+
+The test is **skipped** because letting it hit ``pytest.mark.timeout``
+produces a hard failure that doesn't compose with ``pytest.mark.xfail``
+(pytest-timeout raises a ``BaseException`` subclass that xfail does not
+catch). To check whether Dragon has fixed the bug, remove the
+``pytest.mark.skip`` decorator and run this file under
+``dragon python -m pytest`` — the test will either return cleanly (Dragon
+fixed it, re-enable permanently) or pytest-timeout will fire (still broken).
+
+Lives in its own file for the same reason as the other ``*_hangs.py``
+files: ``method="thread"`` leaks the test thread which breaks
+subsequent tests in the same pytest invocation.
+"""
+
+import os
+import sys
+import warnings
+
+import pytest
+from dragon.workflows.batch import Batch
+
+warnings.warn(
+    "test_unpickleable_function_hangs is DISABLED — Dragon bug (worker dies "
+    "during cloudpickle unpickling without diagnostic; task.get() blocks "
+    "forever) is not actively checked. Re-enable to retest.",
+    UserWarning,
+    stacklevel=2,
+)
+
+
+@pytest.mark.skip(
+    reason="DISABLED — Dragon bug: worker silently dies during cloudpickle "
+    "unpickling; task.get() blocks forever. Re-enable to check if fixed."
+)
+@pytest.mark.timeout(5, method="thread")
+def test_unpickleable_function_hangs(fresh_batch: Batch):
+    here = os.path.dirname(os.path.abspath(__file__))
+    offpath = os.path.join(here, "_offpath")
+    assert offpath not in os.environ.get("PYTHONPATH", "").split(os.pathsep), (
+        "test setup error: _offpath ended up on PYTHONPATH"
+    )
+
+    sys.path.insert(0, offpath)
+    try:
+        from marker_helper import offpath_add  # noqa: PLC0415
+    finally:
+        sys.path.remove(offpath)
+
+    fresh_batch.function(offpath_add, 3, 4).get(timeout=2.0)
diff --git a/tests/dragon_ci/test_worker_failure_modes.py b/tests/dragon_ci/test_worker_failure_modes.py
new file mode 100644
index 0000000..eb354d9
--- /dev/null
+++ b/tests/dragon_ci/test_worker_failure_modes.py
@@ -0,0 +1,43 @@
+"""Contract: Dragon's worker-failure detection — the *clean* paths.
+
+Two failure modes Dragon DOES detect, surfaced through ``task.get()``:
+
+1. **Caught exception** in the user function → the exception instance and
+   its traceback land in the results-DDict 5-tuple (``raised=True``),
+   and ``task.get()`` re-raises it.
+2. **Clean process exit** (``sys.exit(N)``) → no result is written, but
+   Dragon's dispatcher notices and synthesises
+   ``RuntimeError("function worker exited without producing output: …")``.
+
+The silent-hang failure modes (abnormal worker death, unpickleable
+function) live in their own files because they leak threads that break
+subsequent tests in the same pytest invocation:
+
+- ``test_sigkill_in_worker_hangs.py``
+- ``test_unpickleable_function_hangs.py``
+- ``test_ddict_unknown_key_blocks.py``
+"""
+
+from __future__ import annotations
+
+import pytest
+from _dragon_ci_helpers import fn_assert_false  # noqa: E402
+from _dragon_ci_helpers import fn_sys_exit_one  # noqa: E402
+from dragon.workflows.batch import Batch
+
+
+def test_caught_exception_is_reraised_and_tuple_recorded(batch: Batch):
+    task = batch.function(fn_assert_false, 1)
+    with pytest.raises(AssertionError, match="intentional dragon_ci probe"):
+        task.get(timeout=10.0)
+    result, tb, raised, _stdout, _stderr = batch.results_ddict[task.uid]
+    assert raised is True
+    assert isinstance(result, AssertionError)
+    assert isinstance(tb, str) and "AssertionError" in tb
+
+
+def test_sys_exit_in_worker_raises_runtime_error(batch: Batch):
+    """Clean ``sys.exit`` surfaces as ``RuntimeError`` with that exact phrase."""
+    task = batch.function(fn_sys_exit_one, 1)
+    with pytest.raises(RuntimeError, match="exited without producing output"):
+        task.get(timeout=10.0)