Add owner-side ODS flow control and simplify the BC demo

agent_context/MAP.yaml

@@ -430,6 +430,7 @@ topics:
     DepthSidecarManager, depth_meta.json, fragment_id, finalize, expert action schema, qpos_slice, qvel_slice,
     validate_dataset, create_split_manifest, recover_recording, episode_uuid, language_persistent,
     SharedLanguageRegistry, resolve_language, configure_ods, ODSReuseProfileCfg,
+    ODSFlowController, ODSFlowControllerCfg, get_flow_status, ODS 供数控制,
     episode_sample_budget, OnlineDataBudgetExhaustedError, ODS 自动配置, BatchSource, SequenceSpecCfg,
     OnlineDataSource, TensorDataSource, LeRobotDataSource, OfflineDataset, SequenceDataset,
     DataUnavailableError, validate_data_schema, data_schema_fingerprint, ChunkBC, SimpleBCPickPlace, 数据集加载, 离线加载]
@@ -438,6 +439,7 @@ topics:
   - embodichain/data_pipeline/engine/data.py
   - embodichain/data_pipeline/engine/language.py
   - embodichain/data_pipeline/configuration.py
+  - embodichain/data_pipeline/runtime.py
   - embodichain/data_pipeline/datasets/online_data.py
   - embodichain/data_pipeline/datasets/sequence.py
   - embodichain/data_pipeline/datasets/sources.py

agent_context/topics/data-pipeline/data-pipeline.md

@@ -11,6 +11,7 @@ belongs to [data-assets](../data-assets/data-assets.md).
 |---|---|
 | Online worker and sample selection | `embodichain/data_pipeline/engine/data.py`: `OnlineDataEngine` |
 | Throughput-derived settings and protected reuse profiles | `embodichain/data_pipeline/configuration.py`: `configure_ods`; sampler enforcement in `engine/data.py` |
+| Explicit owner-side refill, waits and pacing | `embodichain/data_pipeline/runtime.py`: `ODSFlowController`; compact READY-only flow status in `engine/data.py` |
 | PyTorch iteration | `embodichain/data_pipeline/datasets/`: `sequence_data.py` owns `SequenceDataset`; `online_data.py` preserves `OnlineDataset`; `OfflineDataset` is an alias |
 | Shared sequence contract and sources | `embodichain/data_pipeline/datasets/sequence.py`, `sources.py`, `_sampling.py` |
 | Bounded numeric LeRobot loading | `embodichain/data_pipeline/datasets/lerobot.py`: `LeRobotDataSource`, explicit field mapping |
@@ -82,7 +83,9 @@ action contracts independently of the expert-controller schema; see
 Distinguish a terminal worker failure from a sampling request with no eligible
 unlocked window. The latter raises `DataUnavailableError` (a `RuntimeError`)
 immediately without itself setting `FAILED` or triggering a refill. Datasets
-propagate both kinds of exception without an automatic retry loop.
+propagate both kinds of exception without an automatic retry loop. Applications
+can opt into `ODSFlowController` around their manual engine for bounded retries,
+refill and plan-based pacing; start/stop and training stay application-owned.
 
 ## Persistence contract
 

agent_context/topics/data-pipeline/online-sampling.md

@@ -51,6 +51,39 @@
    so DataLoader can collate; batch mode re-samples a dynamic chunk per yield. Multi-worker
    readers sample shared memory independently.
 
+## Explicit owner-side flow control
+
+`runtime.py` owns opt-in `ODSFlowController` around a caller-managed manual
+engine. Its creating process/thread calls the controller; the application still
+starts/stops the engine and owns training, evaluation and resource accounting.
+An optional `ODSConfiguration` must match engine cache, producer count, quotas
+and batch size. A plan or production limit requires an explicit positive
+`gym_config.num_envs` dividing cache size; unplanned/unlimited legacy engines
+retain their existing layout support.
+With a plan, `prepare_batch()` schedules quota-threshold refill, bounded quota
+waits and batch-start pacing; a pause does not accumulate catch-up credits.
+Without a plan it only polls, leaving round/frozen refill decisions explicit.
+The optional accepted-episode limit covers initial fill and whole-cache refills;
+partial-cache remainder is not produced, and direct engine requests bypass it.
+Already pending generations are adopted without a new engine request or refill
+callback; only requests made through the controller receive request timestamps.
+
+Controller waits drain lifecycle events and check health/caller deadlines.
+`sample_context()` retries only live no-window/quota exceptions within a total
+bound; Source/Dataset behavior stays immediate. Generation waits can accept an
+absolute monotonic deadline that includes earlier work. Callback interval/refill
+times use absolute host monotonic seconds; the application aligns its own clock.
+The controller does not restore source or pacing state. The
+[configuration guide](../../../docs/source/features/ods/configuration.md#apply-the-plan)
+owns the integration example and callback details.
+
+`engine.check_health()` is READY-only and checks existing worker errors/liveness
+without sampling/refill. `get_flow_status()` returns compact generation and
+optional quota counters without per-row audit records or payload clones.
+Producer counters and quota totals use separate lock snapshots. Unlocked quota
+uses the same valid/continuity and write-window test as the full audit, but is a
+coarse scheduling bound; exact context/filter eligibility stays with sampling.
+
 ## Optional publication accounting
 
 `OnlineDataEngineCfg.lifecycle_event_capacity` enables a bounded shared log;
@@ -79,8 +112,9 @@ quota from no valid windows and terminal worker failure. It is immediate and
 does not partially spend a batch. `get_sampling_budget_snapshot()` returns
 retained identities, budgets, delivered/remaining counts and lifetime delivery;
 clean stopped engines remain auditable. The owner decides when to refill, wait
-with a bound, or stop. The fixed publication budget is not an adaptive replay
-retention policy and does not restore old sources after process restart.
+with a bound, or stop, directly or through the optional flow controller. The
+fixed publication budget is not an adaptive replay retention policy and does
+not restore old sources after process restart.
 
 `configuration.py` derives bounded budgets, cache/refill settings and pacing
 targets from measured accepted episodes/s and training chunks/s. It reads an

docs/source/api_reference/embodichain/embodichain.data_pipeline.engine.rst

@@ -44,6 +44,12 @@ require a DatasetManager.
 ``request_refill()`` and observe publication with ``wait_for_generation()``.
 The default remains sample-threshold refill. ``get_stats()`` returns detached
 counters and excludes the active write window from valid-frame counts.
+``check_health()`` requires READY and surfaces worker failures without sampling
+or refill. ``get_flow_status()`` supplies compact generation, accepted-episode
+and optional quota counters for frequent scheduling checks. Its producer and
+quota counters use separate lock snapshots; unlocked quota is a coarse bound
+that does not apply every sequence/filter requirement. Use
+``get_sampling_budget_snapshot()`` for the retained-source audit.
 ``shutdown_timeout`` is the finite, positive graceful-close wait in seconds
 (default ``5.0``). Increasing it allows longer worker teardown; it does not
 change forced terminate/kill waits or their terminal failure semantics.

docs/source/api_reference/embodichain/embodichain.data_pipeline.rst

@@ -22,6 +22,7 @@ storage for LeRobot datasets on Python 3.10--3.12).
       depth_video
       engine
       recording
+      runtime
 
 Datasets
 --------
@@ -58,6 +59,31 @@ Online Data Engine
 
       data
 
+Owner-side flow control
+-----------------------
+
+The application starts and stops its manual-refill engine. The optional
+controller applies a measured plan, drains publication events, checks health,
+and bounds generation/data/quota waits. A plan or production limit requires an
+explicit producer count dividing cache size. Sources and Datasets retain
+immediate sampling/error behavior. See :doc:`/features/ods/configuration` for
+matching engine settings, callback clocks and an integration example.
+
+.. automodule:: embodichain.data_pipeline.runtime
+
+.. autosummary::
+
+   embodichain.data_pipeline.runtime.ODSFlowControllerCfg
+   embodichain.data_pipeline.runtime.ODSFlowController
+
+.. autoclass:: embodichain.data_pipeline.runtime.ODSFlowControllerCfg
+   :members:
+   :show-inheritance:
+
+.. autoclass:: embodichain.data_pipeline.runtime.ODSFlowController
+   :members:
+   :show-inheritance:
+
 Recording integrity
 -------------------
 

docs/source/features/ods/architecture.md

@@ -17,7 +17,7 @@ flowchart TB
     Stored["Resident tensors / LeRobot files"] --> Source
     Source --> Learner["Learner<br/>Direct sampling or SequenceDataset"]
     Learner --> Eval["Independent student evaluation"]
-    Host["Application owner<br/>configure_ods, pacing, refill and shutdown"]
+    Host["Application owner<br/>Plan, optional flow control and shutdown"]
     Host -.-> Engine
     Host -.-> Learner
 ```
@@ -40,6 +40,7 @@ that requests targeted collection from the producer.
 | Production and shared state | `engine/data.py`: lifecycle, worker health, publication, sampling, generation accounting, provenance and optional live delivery ledger. |
 | Language metadata | `engine/language.py`: append-only engine-local task/subtask dictionaries used by copied numeric language IDs. |
 | Planning | `configuration.py`: typed reuse/runtime inputs, measured-rate calculations and immutable `ODSConfiguration`. The host applies the plan. |
+| Owner-side flow control | `runtime.py`: opt-in `ODSFlowController` for manual refill, bounded waits, lifecycle draining and plan-based batch pacing. The application starts/stops the engine and owns training. |
 | Contract and schema | `datasets/sequence.py`, `schema.py`: `BatchSource`, `SequenceSpecCfg`, typed sampling errors, semantic declaration validation and schema fingerprints. |
 | Live/resident adaptation | `datasets/sources.py`: `OnlineDataSource` delegates to its caller-owned engine; `TensorDataSource` owns resident numeric data and its optional separate ledger. |
 | Numeric SDK loading | `datasets/lerobot.py`: `LeRobotDataSource`, explicit field mapping, bounded reads and a resident allocation limit. |
@@ -58,7 +59,8 @@ origin IDs also do not establish portable dataset content identity.
 **Live:** the owner starts the engine and waits for initial fill. The worker
 accepts task rollouts and publishes complete episodes. `OnlineDataSource`
 delegates to the engine's locked sampler and returns canonical owned contexts.
-The host schedules further fills and training; shutdown belongs to the creator.
+The host schedules further fills and training, optionally through
+`ODSFlowController`; shutdown belongs to the engine creator.
 Refill publishes batches of rows incrementally: readable rows can temporarily
 belong to different generations. A generation filter is explicit;
 `wait_for_generation()` reports completion of the entire requested fill.
@@ -84,7 +86,9 @@ training-efficiency advantage.
 
 - `configure_ods()` is a static planner. It does not run calibration, mutate a
   live engine, choose a learning-safe profile or invoke the optimizer. The
-  application enforces active/average pacing and planned refill thresholds.
+  application can explicitly select `ODSFlowController` to enforce the plan's
+  refill threshold and batch-start pacing with bounded waits. The controller
+  is creator-process/thread only and does not start or stop the engine.
 - `request_refill()` is owner-only and coalesces an outstanding request.
   `wait_for_generation()` waits for a complete fill. Optional bounded lifecycle
   logging must be drained during production and waits.
@@ -94,8 +98,9 @@ training-efficiency advantage.
   there is no distributed global quota across independent sources/runs.
 - `DataUnavailableError` means no eligible window exists;
   `DataBudgetExhaustedError` means eligible data cannot fund the requested batch.
-  Worker/read failures remain terminal errors. Dataset iteration adds no hidden
-  retry, refill or budget reset.
+  The opt-in controller retries only these transient live errors up to its
+  deadline while draining events and checking health. Worker/read failures
+  propagate. Dataset iteration adds no hidden retry, refill or budget reset.
 - Model/optimizer/RNG recovery belongs to the application. The demo recreates a
   live cache with a new run ID; exact replay-cache or protected offline-ledger
   recovery is not implemented.