Add outcome-aware dataset fragments

agent_context/topics/env-framework/env-framework.md

@@ -13,6 +13,8 @@
 | `embodichain/lab/gym/envs/base_env.py` | `BaseEnv(gym.Env)` + `EnvCfg` — low-level env loop |
 | `embodichain/lab/gym/envs/types.py` | `ControllerAction` — owned controller-ready action boundary |
 | `embodichain/lab/gym/envs/embodied_env.py` | `EmbodiedEnv(BaseEnv)` + `EmbodiedEnvCfg` — modular task base class |
+| `embodichain/lab/gym/envs/demo.py` | Segment-aware demonstration execution, result, and persistence-mode contracts |
+| `embodichain/lab/scripts/run_env.py` | Offline collection retries and explicit dataset commit/abort boundaries |
 | `embodichain/lab/gym/utils/registration.py` | `@register_env` decorator + `REGISTERED_ENVS` registry + `make()` |
 | `embodichain/lab/gym/utils/gym_utils.py` | Gym config parsing and config-owned runtime registration |
 | `embodichain/lab/gym/utils/_component_composition.py` | Reusable physical environment, embodiment, and standalone scene resolution |
@@ -74,6 +76,8 @@ gym.Env
   dataset save, and manager resets.
 - Overrides `_update_sim_state()` to run event-manager `interval` mode.
 - Manages rollout buffer (expert or RL mode) via `_hook_after_sim_step()`.
+- Owns per-row demonstration metadata and dense segment acceptance, attempt,
+  and causal-continuity annotations.
 - `extensions` dict entries are set as attributes on both cfg and env instance.
 
 ### Action boundary (`types.py`, `embodied_env.py`)
@@ -109,6 +113,10 @@ gym.Env
   results.
 - `reset()` lets `BaseEnv` read that final mask before clearing the active
   bridge, preventing completed state from leaking into the next episode.
+- `DemoSegmentResult.successes` is the persisted segment acceptance authority.
+  `DemoExecutionCfg` keeps continuous episodes as the default and can instead
+  make each eligible natural segment an independent dataset fragment. This
+  mode does not restore state or resume after failure.
 - Environments without a Task Program keep the ordinary `BaseEnv.is_task_success()`
   behavior and task-specific overrides.
 
@@ -279,9 +287,9 @@ reset(options)
   ├── is_task_success() → save status before resetting
   ├── sim.reset_objects_state(env_ids, excluded_uids)
   ├── _initialize_episode(env_ids)
-  │     ├── dataset_manager.apply("save") for successful episodes (or an
-  │     │   explicit dataset-only ``commit_env_ids`` subset during a final
-  │     │   vector batch)
+  │     ├── dataset_manager.apply("save") for successful episodes, eligible
+  │     │   segment-fragment rows, or an explicit dataset-only
+  │     │   ``commit_env_ids`` subset during a final vector batch
   │     ├── event_manager.apply("reset", env_ids)
   │     ├── observation_manager.reset(env_ids)
   │     └── reward_manager.reset(env_ids)

agent_context/topics/manager-functor/manager-functor.md

@@ -46,9 +46,20 @@ At init, the manager resolves every `FunctorCfg.func` (string → callable or cl
 | Recorder | File | Behavior |
 |----------|------|----------|
 | `LeRobotRecorder` | `managers/datasets.py` | Synchronous. `__call__` runs convert + `add_frame` + `save_episode` inline, blocking `env.reset()`. Default; base class for the async variant. |
-| `AsyncLeRobotRecorder` | `managers/async_datasets.py` | Subclass. `__call__` clones the rollout-buffer slice (obs+actions) to CPU, enqueues it, and returns immediately. A single daemon worker thread drains the queue and runs the same `_save_single_episode` path. `finalize()` drains then calls `dataset.finalize()`. |
+| `AsyncLeRobotRecorder` | `managers/async_datasets.py` | Subclass. `__call__` clones the rollout-buffer slice (obs+actions) to CPU, enqueues it, and returns immediately. A single daemon worker thread drains the queue through the same `_persist_episode_payload` path. `finalize()` drains then calls `dataset.finalize()`. |
 
-**Save flow**: `env.step` writes each frame into `rollout_buffer` (`_hook_after_sim_step`). On truncation the caller does `env.reset(options={"save_data": True})` -> `_initialize_episode` -> `DatasetManager.apply("save", env_ids)`. For a final partial vector batch, `run-env` performs one full reset with `save_data=False` and `commit_env_ids`, so only selected **dataset** rows are persisted while whole-world reset events remain safe; camera and trajectory recorders retain their normal discard behavior. `DatasetFunctorCfg.save_failed_episodes=True` saves every env on every ordinary reset (not only successes). `env.close()` -> `dataset_manager.finalize()` flushes any remaining buffer.
+**Save flow**: `env.step` writes each frame into `rollout_buffer` (`_hook_after_sim_step`). On truncation the caller does `env.reset(options={"save_data": True})` -> `_initialize_episode` -> `DatasetManager.apply("save", env_ids)`. For a final partial vector batch, `run-env` performs one full reset with `save_data=False` and `commit_env_ids`, so only selected **dataset** rows are persisted while whole-world reset events remain safe; camera and trajectory recorders retain their normal discard behavior. `DatasetFunctorCfg.save_failed_episodes=True` saves every env on every ordinary reset (not only successes). `env.close()` -> `dataset_manager.finalize()` drains explicitly committed async work; it never commits the live rollout implicitly.
+
+`DemoExecutionCfg(mode="segment_fragments")` changes one buffered Task Program
+row into independent natural-segment payloads. Accepted segments are retained
+by default; failed segments require `save_failed_fragments=True`. Every
+fragment carries dense `segment_accepted`, `segment_attempt_id`, and
+`continuity_id` features plus Task Program provenance in the JSONL sidecar.
+Fragment commits are append-only: failure of a later fragment does not roll
+back earlier ones. The recorder-local deterministic `fragment_id` registry
+deduplicates same-run retries. A LeRobot commit followed by sidecar/depth
+failure is sticky and rejects retry rather than writing a duplicate. This is
+not a cross-process recovery journal.
 
 **Two independent speed levers** (both honor `image_writer_threads` / `image_writer_processes` in `params`, wired through to `LeRobotDataset.create()` -> lerobot `AsyncImageWriter`):
 - Opt A: `LeRobotRecorder` + `image_writer_threads=4` - per-frame PNG writes offloaded to a thread pool. ~2.5x faster, no background thread, bounded memory.
@@ -61,6 +72,8 @@ At init, the manager resolves every `FunctorCfg.func` (string → callable or cl
 **Correctness invariants for the async recorder** (do not break these when editing):
 - The buffer slice is **cloned in the caller thread** before enqueue - the worker must not hold a view into `rollout_buffer` (it is cleared/reused on reset).
 - **Single worker** only - `LeRobotDataset` is not thread-safe and FIFO order must be preserved for `episode_index`.
+- Duplicate fragment ids pass through `_persist_episode_payload` so sync and
+  async writers share the same idempotency rule.
 - `finalize()` must drain the queue before `dataset.finalize()`.
 - `__call__` accepts `**kwargs` because `DatasetManager.apply` passes `**functor_cfg.params` (includes construction-only params like `image_writer_threads`); `manager_base._resolve_common_functor_cfg` tolerates `**kwargs`.
 

agent_context/topics/task-programs/task-programs.md

@@ -198,6 +198,25 @@ Keep these boundaries separate:
 reward, reset, and persistence. Final success is published only after every
 segment lifecycle completes normally.
 
+## Demonstration outcome and persistence
+
+The bridge's accepted mask remains the sole segment-quality authority. The
+common demo executor records it in `DemoSegmentResult.successes`, classifies the
+first authoritative failure phase in `outcome_kinds`, and writes
+`segment_accepted`, `segment_attempt_id`, and `continuity_id` for every real
+rollout frame. Task Program segment metadata uses `task_program_id`; LeRobot
+fragment sidecars copy it to the provider-neutral `source_program_id` field.
+
+`DemoExecutionCfg` is collector-owned, not part of the Task Program language.
+Its default `continuous` mode preserves episode semantics. Its
+`segment_fragments` mode persists eligible naturally executed segments as
+independent LeRobot episodes, with failed fragments requiring explicit opt-in.
+Current execution remains fail-closed and every frame has `continuity_id == 0`;
+checkpoint capture, state restore, and suffix resume are deferred. Fragment
+commits are individually durable and recorder-locally idempotent by stable
+`fragment_id`, so a later write failure does not duplicate an earlier fragment
+on retry.
+
 ## Parallel and registered calls
 
 Parallel execution requires disjoint resource claims and runtime targets,
@@ -245,13 +264,16 @@ language.
 | Live simulation binding | `integrations/simulation/` |
 | Gym action/segment lifecycle | `gym/envs/task_program/bridge.py` |
 | Episode program selection/final success | `gym/envs/embodied_env.py` |
+| Outcome annotations and persistence mode | `gym/envs/demo.py`, `gym/envs/embodied_env.py` |
+| LeRobot fragment slicing/idempotency | `gym/envs/managers/datasets.py`, `async_datasets.py` |
 
 ## Focused validation
 
 ```bash
 pytest -q tests/lab/task_program
 pytest -q tests/gym/envs/task_program
 pytest -q tests/gym/envs/test_embodied_env_task_program.py
+pytest -q tests/gym/envs/test_demo.py tests/gym/envs/managers
 pytest -q tests/agents/mllm/test_task_program.py
 pytest -q tests/sim/atomic_actions
 python docs/scripts/check_api_docs.py