Feature/api tests

.github/workflows/ci.yml

@@ -101,3 +101,42 @@ jobs:
         run: >-
           pytest test/integration -rs --continue-on-collection-errors
           --timeout=120 --timeout-method=thread
+  api:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.13"
+          cache: pip
+
+      # The radical dependencies must not come from naive PyPI resolution:
+      # PyPI's rhapsody-py 0.4.0 lacks `rhapsody.backends.execution.orbit`
+      # -- the backend `digitaltwin.service` imports -- which was the
+      # collection-blocking ModuleNotFoundError this workflow documented.
+      # Pre-installed pinned, so resolving digitaltwin's requirements
+      # keeps them.  asyncflow 0.5.1 carries the non-main-thread fix;
+      # orbit is floored, not pinned, since 0.7.0 (dispatcher dialect).
+      - name: Install pinned radical dependencies
+        run: >-
+          pip install "radical.asyncflow==0.5.1" "radical.orbit>=0.7"
+          "rhapsody-py[telemetry] @ git+https://github.com/radical-cybertools/rhapsody@e491cd2"
+          "rose @ git+https://github.com/radical-cybertools/ROSE@64330d9cb43c3e13ca67daf0d8ae84a2ae6c3f17"
+
+      # src/ layout: pytest tests the *installed* package, never the
+      # working tree.  Always install before testing -- a stale install
+      # will silently pass against old code.
+      - name: Install digitaltwin (test + service extras)
+        run: pip install ".[test,service,learn]" pytest-timeout
+
+      # --continue-on-collection-errors: one broken test module (see the
+      # PR description) must not hide the rest of the suite behind an
+      # "Interrupted" abort. --timeout guards against a hanging test
+      # taking down the whole job; it turns a hang into a failed test
+      # with a traceback.
+      - name: Run api tests
+        run: >-
+          pytest test/api_test -rs --continue-on-collection-errors
+          --timeout=120 --timeout-method=thread

src/digitaltwin/components.py

@@ -389,6 +389,7 @@ def __init__(self, name: str, hard: bool = True) -> None:
         self.count_hard = 0
         self.count_soft = 0
         self.set_soft = False
+        self.recv_soft = 0
 
     def __str__(self) -> str:
         return self.name
@@ -423,15 +424,18 @@ async def put(self, in_data: TypedData):
         dtype = in_data.dtype
         if not (self.dtypes[dtype]):
             # soft. just store the result
+            if len(self.previous[dtype]) == 0:
+                self.recv_soft += 1
+
             if self.previous_retain.get(dtype, True):
                 self.previous[dtype] = [in_data.data]
                 self.previous_retain[dtype] = False
             else:
                 self.previous[dtype].append(in_data.data)
-            if not self.set_soft:
+
+            if self.count_hard == 0 and not self.set_soft:
                 self.set_soft = True
-                for _ in range(self.count_soft):
-                    self._update.release()
+                self._update.release()
             return
 
         def predicate():
@@ -487,15 +491,23 @@ async def run(self):
             self.condition.notify_all()
             self.condition.release()
 
-            for i in range(self.count_hard + self.count_soft):
+            # The update will only fire until ALL the soft items gets something.
+            if self.recv_soft < self.count_soft:
+                await asyncio.sleep(0.01)
+                continue
+
+            # did all hard vals update.
+            for i in range(self.count_hard):
+                await self._update.acquire()
+
+            if self.count_hard == 0:
                 await self._update.acquire()
 
             self.set_soft = False
             for dtype in self.dtypes:
                 if self.dtypes[dtype]:
                     continue
                 # emit on any soft barriers
-
                 # drain previous in reverse append order
                 self.output_queues[dtype].put_nowait(
                     WindowedTypeData(

src/digitaltwin/runtime.py

@@ -13,6 +13,7 @@
 from collections import defaultdict, deque
 from contextvars import ContextVar
 from dataclasses import dataclass, field
+import sys
 from typing import cast
 
 try:
@@ -1018,6 +1019,7 @@ def _record_error(self, exc: BaseException | str):
             error,
             exc_info=exc if isinstance(exc, BaseException) else None,
         )
+        print(f"twin component failed: {error}", file=sys.stderr)
 
         if self.state is RuntimeState.FAILED:
             # the cause is already recorded, and its teardown is running
@@ -1282,6 +1284,8 @@ async def _internal_agent_inference(self, in_data: TypedData, req_dtype: DataTyp
                     return TypedData(NULL_DTYPE, None)
                 return answer
 
+        raise ValueError("No component found")
+
     # add a barrier
     def add_barrier(self, barrier: Barrier) -> None:
         """Register a synchronization barrier.

test/api_test/README.md

@@ -0,0 +1,48 @@
+# API test coverage mapping
+
+This maps the `api_test/conftest.py` checklist of DT framework promises to the
+existing numbered demos under `test/` (00-11, 100), as a starting point for
+writing real pytests that check the framework does what it promises, not just
+that individual functions don't crash. These are not unit tests. 
+
+## Checklist → Demo coverage
+
+| Checklist item | Covered by | Confidence |
+|---|---|---|
+| **ADD_INPUT** | 01, 04 (`runtime.add_input` binds external sensor channel) | Solid, but only via manual two-terminal run |
+| **ADD_TASK - persistent** | 02, 03, 05, 06, 07, 08, 09, 10, 11 | Solid |
+| **ADD_TASK - non-persistent** | Every demo's `data_sink` | Solid but incidental — nothing asserts non-persistence specifically |
+| **ADD_INVESTIGATOR - input callback** | 03, 06 (gregory/nilakantha/monte_carlo), 100 | Solid |
+| **ADD_INVESTIGATOR - inference task** | Nearly all demos | Solid |
+| **ADD_INVESTIGATOR - inference task update** | 02, 03, 04, 05, 06, 100 (`publish_new_model`) | Solid |
+| **ADD_AGENT - Model selector task** | 04, 05, 06, 09, 11, 100 | Solid |
+| **ADD_AGENT - Model publish task** | 06, 100 (`model_publish_cb` override) | Solid |
+| **ADD_AGENT - filter input task** | none | **Zero coverage** — `ON_FILTERED_INPUT`/`ON_FILTERED_OUTPUT` exist in runtime.py but no demo subscribes to them |
+| **ADD_AGENT - inter-agent inference** | 100 only (`get_inference` chained through profiler→endpoint) | Solid but entangled with the profiler harness |
+| **ADD_AGENT - Model selector update** | 04, 05, 06, 09, 11, 100 | Solid |
+| **ADD_AGENT - Multi Investigators** | 05, 06, 11 | Solid |
+| **ADD_BARRIER - Hard barrier** | 07 — but the hard-barrier block is **commented out** in `run_me.py` | **Effectively uncovered** |
+| **ADD_BARRIER - Soft default barrier** | 07 (a/b/c soft dtypes) | Solid |
+| **ADD_BARRIER - Hard(slow)/soft(fast)** | 07 mixes delays but doesn't isolate/assert this pairing | Gap |
+| **ADD_BARRIER - Soft(fast)/hard(slow)** | Same — happens incidentally in 07, never asserted | Gap |
+| **ADD_DATA_JOIN - Data Join** | 08 (real join), 09 (trivial single-dtype no-op join) | Solid |
+| **ADD_DATA_SPLIT - task** | 09, 10 | Solid |
+| **ADD_DATA_SPLIT - a None** | none — no demo's split ever returns a fully-None result | **Gap** |
+| **ADD_DATA_SPLIT - one None, one Item** | 10 (`HighLow`) | Solid |
+| **ADD_DATA_SPLIT - both items** | none — no split ever returns two live `TypedData` in one call | **Gap** |
+
+## Gaps (zero real coverage today)
+
+1. **Filter input task** (`ON_FILTERED_INPUT`/`ON_FILTERED_OUTPUT`) — mechanism exists, nothing exercises it.
+2. **Hard barrier** — only example is commented out in `07-barrier/run_me.py`.
+3. **Hard/soft speed-pairing semantics** — 07 runs a 5-sensor mix but never isolates or asserts either ordering.
+4. **Data split → all-None result**.
+5. **Data split → both outputs populated simultaneously**.
+
+## Behaviors seen in demos but not on the checklist
+
+- Basic lifecycle (start/stop/redeploy progression across 00→01→02)
+- Shared sub-tasks across investigators (`11-shared-sim`: `register_shared_subtask`/`get_shared_subtask`/`call_shared_subtask`)
+- Remote/distributed orchestration (`09-remote`: `RemoteDTOrchestrator`, `runtime.package`, `register_user_modules`)
+- Resource-aware model selection (`100`) — really a richer version of "inter-agent inference"
+- Windowed data reads underlying the soft barrier (`WindowDataType`/`WindowedTypeData`)