[yuecideng]

Consolidated into #828, which now targets main. All commits from this PR are already present in codex/ods-stream-design; implementation, CI fixes and remaining validation continue in #828. This PR is closed as superseded.

Description

Protected ODS consumers currently reproduce refill, quota-readiness, wait and pacing loops in application code. Add an opt-in ODSFlowController that applies an existing measured configuration in the owner process/thread, drains lifecycle events during bounded waits, propagates worker failures and respects whole-cache production ceilings. Source and Dataset sampling retain their immediate error contracts.

Add compact get_flow_status() and public check_health() APIs without constructing a full quota audit on each update. Migrate the BC demo and direct LeRobot collector to the controller, remove duplicate polling and obsolete learner branches, and move resource sampling alongside application accounting. The trainer is 226 lines shorter.

The application retains engine start/stop, optimizer, evaluation and checkpoint ownership. This applies a static measured plan; it does not calibrate throughput, qualify a safe reuse profile or provide distributed/replay recovery. Update ODS/API docs and project context while retaining the simple architecture diagram.

Refs #788. No new dependencies.

Type of change

• Enhancement (non-breaking change which improves an existing functionality)

Validation

• Conda py311: 571 passed across tests/data_pipeline, API-doc tooling and agent-context tests. Tests cover bounded waits, terminal errors, owner restrictions, refill races, quota limits and real forkserver production with a backpressured lifecycle log. Final focused runtime rerun: 66 passed (included in the 571 cases).
• 15 paired application scenarios / 30 complete runs against cd5589255: identical sampled batches, model/optimizer/RNG state, checkpoint save times/selection and stop reasons. Uses real CPU BC/source/checkpoint code with deterministic simulator/evaluator and clock substitutes; includes budget boundaries, resident/frozen paths, final supply exhaustion and cross-round refill.
• Real Franka simulation in py311 with the host's matching DexSim runtime: online control completed generation 2, 4 optimizer updates and 16 delivered chunks; cold offline collection wrote/reloaded 4 LeRobot episodes and completed one protected update. Both performed independent student evaluation and closed all owned workers without cleanup errors. These are engineering smoke checks with reduced scene counts and a synthetic online capacity packet, not learning-quality or throughput qualification. An initial invalid synthetic cache/production-cap pairing was correctly rejected before simulation; the corrected run passed and both attempts were retained.
• Black 26.3.1 (black .), API coverage 2792/2792, Python 3.10 syntax, git diff --check, context validation and a scoped ODS Sphinx build passed. The full repository suite and full-site documentation build were not run.

Checklist

• I have run the black . command to format the code base.
• I reviewed affected documentation and agent context and updated it where needed.
• Public API changes are reflected in the API docs (python docs/scripts/check_api_docs.py).
• I have added focused tests for the production contracts.
• Dependencies reviewed; no changes required.