Add Flora inference examples

SOURCE_MANIFEST.sha256

@@ -1,5 +1,5 @@
 837c0aa8748126d2193799e7bec51d6550ca4ba07767591e66dcb370c3115e44  MODEL_PROVENANCE.json
-6c051cfaeac3bbce138ba1b95ff1e21c7f2812356c472e9a11cf4b9467099680  README.md
+88cb6731fc521a0404faed7c5045c3db77185341849bc4e0b6e60be3443f7d6c  README.md
 ac4d8abd2a14045b22151354a7493049d651ec10c0b019fbe4927805e45c4b81  adapter_config.json
 1d80fe544e4cd1a7ff6b0ea4be71e988972408dab81d4c8af72e3a628739d6c3  adapter_model.safetensors
 d28ca68768ac2ce0ad42eaaf54d5bac4f8a298f7d017f5a04f3866094c65a839  chat_template.jinja

batch_utils/CODE_MANIFEST.json

@@ -0,0 +1,173 @@
+{
+  "files": {
+    "batch_utils/README.md": {
+      "bytes": 4125,
+      "sha256": "f9a9dd0a67b914bb1304a53674c03deec9f6a9aaa580781a13d88b365284a65c"
+    },
+    "batch_utils/__init__.py": {
+      "bytes": 91,
+      "sha256": "50a358089004c69e4c83c41a55eb44c52179d3fa20be4d4c5b92fabbc8271ac4"
+    },
+    "batch_utils/__main__.py": {
+      "bytes": 78,
+      "sha256": "28de248c91c00642578a541827361af9b0e5e049e4c61dd6ec1f9dc78580c24e"
+    },
+    "batch_utils/api.py": {
+      "bytes": 3600,
+      "sha256": "ec5e17b5b02c20a40e52beab7d5ebfd49e32129ef03eb918790df720b0d97177"
+    },
+    "batch_utils/bootstrap.py": {
+      "bytes": 679,
+      "sha256": "83e49883136af2088038c6070303d0e89071a318e98252b40ff988230e491817"
+    },
+    "batch_utils/cli.py": {
+      "bytes": 4761,
+      "sha256": "9dc0256b45252d7512b1c69eb7eba3b681725d7c6109287dc3892ec9a9d668e1"
+    },
+    "batch_utils/config.py": {
+      "bytes": 503,
+      "sha256": "0967244955e8f128e03cc6a85973a41d1571f78acf4a9bc6f91e6682781f9891"
+    },
+    "batch_utils/media.py": {
+      "bytes": 5135,
+      "sha256": "e0ee696b63f3171d5163aadb960322f422183981c3768efd76c409976f3fda5b"
+    },
+    "batch_utils/models.json": {
+      "bytes": 612,
+      "sha256": "210e56e9ed46f61c6e18a974b78fccfb6a6aee3a6b0cb6cda5b5f774f934f536"
+    },
+    "batch_utils/models.py": {
+      "bytes": 11436,
+      "sha256": "92b3ab4b376f9a4dd3c7f75568746467f17a969430c63330667f2c6d2760f1fe"
+    },
+    "batch_utils/pipeline.py": {
+      "bytes": 11853,
+      "sha256": "588d4105197706da8acf3b591ba9671239505b53030f1d30cba0088ddad5432e"
+    },
+    "batch_utils/preprocessing.py": {
+      "bytes": 3311,
+      "sha256": "6bc9a2c7f08f92b4933f68c0fc674f938339eceeebdec186eef1514294b167ee"
+    },
+    "batch_utils/requirements.txt": {
+      "bytes": 163,
+      "sha256": "66d607ebfbd6525e005db045812453c4356d7df4e42e7b40b4994bd0e5e0cd51"
+    },
+    "batch_utils/vendor/MANIFEST.json": {
+      "bytes": 3225,
+      "sha256": "91ce5d8a5f13a583d48384c5de7f46696f5cd896ad2c1f3ce0eead9a1acae6d8"
+    },
+    "batch_utils/vendor/flora_unified/__init__.py": {
+      "bytes": 5,
+      "sha256": "9f56e761d79bfdb34304a012586cb04d16b435ef6130091a97702e559260a2f2"
+    },
+    "batch_utils/vendor/flora_unified/unified_input.py": {
+      "bytes": 5978,
+      "sha256": "2aa2b521191b68913b03d7582b7a3bcfad49d90c014a1a1bb7571b5e4f08d7ef"
+    },
+    "batch_utils/vendor/production_adapter/TEMPSEG_PROMPTS.json": {
+      "bytes": 6654,
+      "sha256": "66edd6fb43b7fa55cc6257beee5c42a1a4555ab9790320c31114d77bd07a05f1"
+    },
+    "batch_utils/vendor/production_adapter/adapter.py": {
+      "bytes": 7526,
+      "sha256": "083c6788b561d726a85d9f3f8f516593a9f6f76a43f315581c821871e3f82312"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/__init__.py": {
+      "bytes": 5,
+      "sha256": "9f56e761d79bfdb34304a012586cb04d16b435ef6130091a97702e559260a2f2"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/exact_frame_serving.py": {
+      "bytes": 2120,
+      "sha256": "f921e279b5d52f723b3e77e25441763683d05c6e9cd2f8ff204466a50255b346"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/schema.py": {
+      "bytes": 3489,
+      "sha256": "42c523e4db1a4da3fe0d0d578f3cf6e94aed7666c06e3f18081c209159de04b9"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/subtask_captioning/__init__.py": {
+      "bytes": 5,
+      "sha256": "9f56e761d79bfdb34304a012586cb04d16b435ef6130091a97702e559260a2f2"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/subtask_captioning/u2_contract.py": {
+      "bytes": 3511,
+      "sha256": "873476e78a9e3e222ae9e47fcb7a04c8c18ef518b2847c401a2e1cfb2292fe90"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/subtask_captioning/u2_worker.py": {
+      "bytes": 2526,
+      "sha256": "870641228b0be442613d520232d451047742ca2ea41fd0cd51dbf0a58ba567d2"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/__init__.py": {
+      "bytes": 5,
+      "sha256": "9f56e761d79bfdb34304a012586cb04d16b435ef6130091a97702e559260a2f2"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/contracts.py": {
+      "bytes": 3361,
+      "sha256": "8591d849b0b65fda6b21511557d4ee5eabd47f8332e2b6831f40d5bb472d978c"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/fine_subtask.py": {
+      "bytes": 77158,
+      "sha256": "6fbf8337bab8491148a78a822187d74397b7f0552700252451a18cd7c78c8e04"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/fine_temporal.py": {
+      "bytes": 66254,
+      "sha256": "32f2b99c97176f5e10934450da638577268c2e5d9977c2ea138ebf1b7eda58fa"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/fine_worker.py": {
+      "bytes": 1090,
+      "sha256": "8a31a77180c8d406113d4b494644153564a6f46c8a9b68ad7babb3cc1695edec"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/interface.py": {
+      "bytes": 996,
+      "sha256": "a5424b2cc3a4c879552dd49bae207a58fb9e751359f441c03af167a2e1879359"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/scope_scheduler.py": {
+      "bytes": 12654,
+      "sha256": "be95411b39124d83811fe74dcac2d911169ecf475a81b7460c64b19fbbc7e5db"
+    },
+    "batch_utils/vendor/production_adapter/core/flora_runtime/worker_transport.py": {
+      "bytes": 953,
+      "sha256": "30fb64869eb6a27ab96bf394c105f19fcb42590be4424c66b76b89e17631693a"
+    },
+    "batch_utils/vendor/production_gdn_fix.py": {
+      "bytes": 202,
+      "sha256": "a3f52604bfa2ec58901df63215df5e47d37e1d7974aa56d36a1e4efe8cad3fc6"
+    },
+    "batch_utils/vendor/pts_input_adapter.py": {
+      "bytes": 2340,
+      "sha256": "d784d6b77005eb6945dee334e4c9136fbededfa2049d5945244f198517984767"
+    },
+    "batch_utils/vendor/robot_caption_geometry.py": {
+      "bytes": 2739,
+      "sha256": "1c874d3c63359663276d31a3e9c3112a5982ad4000f44c808541b2471f5f7d2d"
+    },
+    "batch_utils/vendor/support/canonical_rgb.py": {
+      "bytes": 1010,
+      "sha256": "1e3230c60b45d8e017060d8a5dd2b5f48e2199192f09dcff4d04c05b4349e35a"
+    },
+    "batch_utils/vendor/support/composers.py": {
+      "bytes": 1263,
+      "sha256": "180b97a6a634979c1b0dbe720ed7a1abd818dc7ee31fdc4eeca7a8277e43d1b6"
+    },
+    "batch_utils/vendor/support/plans.py": {
+      "bytes": 3762,
+      "sha256": "d75b918804c9ec10f8642d362a54a74c3eb875abf22e8a60228a0d6b2cbb570e"
+    },
+    "batch_utils/vendor/support/u2_contract.py": {
+      "bytes": 3321,
+      "sha256": "50190914b719bbc9b8f2b53e52a6bd2b142a452e2fb1c5c897ea4c31123ad4dd"
+    },
+    "batch_utils/vendor/tempseg_fp32_loader.py": {
+      "bytes": 3419,
+      "sha256": "7ee9889f5aa5909abf3ecd1c34e3ac2f81a32a2098d94ae3ba9f33831e7a07c0"
+    },
+    "batch_utils/vendor/verify_gdn.py": {
+      "bytes": 2149,
+      "sha256": "96596ff2db4425f4dea71798aeabb581aa83e1ad8a878cfcb9efb8dbac80cda6"
+    },
+    "infer.py": {
+      "bytes": 171,
+      "sha256": "8cddbb43ba5b9d884e8f771b9c9e219ce2e0a3c1d71a94a75d02b1d3b730f93d"
+    }
+  },
+  "schema_version": "flora_inference_code_manifest_v1"
+}

batch_utils/README.md

@@ -0,0 +1,107 @@
+# FLORA Data Engine inference
+
+This directory contains the complete **TempSeg → Captioner** workflow, identical
+in both model repositories. Use either copy. TempSeg predicts action intervals;
+the Captioner describes those predicted intervals. The native windowing,
+adaptive retries, stitching, prompts, source timestamps, and camera layouts are
+included; no internal source checkout or running demo is required.
+
+## Setup
+
+Python 3.12 and a CUDA GPU with 80 GB VRAM are the validated configuration.
+The two models run sequentially. Download/cache space of at least 150 GB is
+recommended for the released FP32 Captioner, base model, and adapter.
+
+From a downloaded copy of either repository:
+
+```bash
+python -m pip install torch==2.11.0 torchvision==0.26.0 --index-url https://download.pytorch.org/whl/cu128
+python -m pip install -r batch_utils/requirements.txt
+hf auth login
+```
+
+Both private model repositories must be accessible to the logged-in account.
+`HF_TOKEN` is also supported; do not put a token in source code or command-line
+arguments. Weights are fetched automatically at the revisions in `models.json`.
+Set `FLORA_MODEL_CACHE` to choose the local model directory.
+
+The NVIDIA repositories are selected by default. To override them:
+
+```bash
+export FLORA_TEMPSEG_REPO="nvidia/Flora-TempSeg"
+export FLORA_CAPTIONER_REPO="nvidia/Flora-Captioner"
+```
+
+## Single video
+
+```bash
+python infer.py --video videos/human.mp4 --profile human --output human.json
+```
+
+Three synchronized robot cameras:
+
+```bash
+python infer.py --video videos/top.mp4 \
+  --left-wrist videos/left.mp4 --right-wrist videos/right.mp4 \
+  --profile yam --output yam.json
+```
+
+Use `--profile dex` for Dex or `--profile tianji` for Tianji inputs.
+YAM defaults to native 16-second temporal windows; use `--yam32` for the native
+32-second YAM profile. Human uses one video. Robot profiles accept one view or
+all three; `--yam32` requires all three. Wrist files must share the main video's
+relative frame timestamps, frame count, and duration. Inputs are not resampled.
+
+The validated input limits are 120 seconds, 256 MiB per file, 4096 pixels per
+side, 7200 frames, and at least 4 source FPS. Variable-rate video is supported
+by the single-view Human path. `--instruction` optionally supplies an episode
+task; leave it blank for the default workflow.
+
+CPU-only media validation performs no inference or model downloads:
+
+```bash
+python -m batch_utils --video videos/human.mp4 --profile human --validate-only
+```
+
+## JSONL jobs
+
+Each row specifies one job. Camera paths are relative to the JSONL file:
+
+```json
+{"id":"human01","video":"videos/human.mp4","profile":"human"}
+{"id":"robot01","video":"videos/top.mp4","left_wrist":"videos/left.mp4","right_wrist":"videos/right.mp4","profile":"yam","yam32":true}
+```
+
+```bash
+python infer.py --input-jsonl jobs.jsonl --output-dir results
+```
+
+Jobs and predicted segments run sequentially; JSONL is not GPU batch scheduling.
+Existing outputs are preserved unless `--overwrite` is explicit. An error stops
+the command; successfully written earlier jobs remain available.
+
+## Python usage
+
+```python
+from batch_utils import DataEnginePipeline
+
+engine = DataEnginePipeline()
+result = engine.annotate("videos/human.mp4", profile="human")
+for segment in result["segments"]:
+    print(segment["start_s"], segment["end_s"], segment["caption"])
+```
+
+Outputs contain clip-relative seconds, caption completion status, camera roles,
+input SHA-256, and model revisions. Native overlapping intervals are retained.
+`include_audit=True` / `--include-audit` adds temporal and caption-input records.
+The code preserves FP32 LoRA/GDN values and validates checkpoint manifests.
+Full manifest verification happens on first use in the process; reuse requires
+unchanged manifest bytes and all file identities/timestamps. FP32 parameter
+values are checked on every load. First-time downloads and verification are
+substantial; later jobs reuse the verified local files.
+
+Example output (illustrative):
+
+```json
+{"segments":[{"start_s":0.8,"end_s":3.2,"caption":"Pick up the cup with the right hand.","status":"complete"}]}
+```

batch_utils/__init__.py

@@ -0,0 +1,3 @@
+from .api import DataEnginePipeline
+__all__ = ['DataEnginePipeline']
+__version__ = '1.0.0'

batch_utils/__main__.py

@@ -0,0 +1,3 @@
+from .cli import main
+if __name__ == '__main__':
+    raise SystemExit(main())

batch_utils/api.py

@@ -0,0 +1,58 @@
+from pathlib import Path
+import math
+import threading
+from .config import MAX_UPLOAD_BYTES, MODELS
+_REQUEST_LOCK = threading.RLock()
+_PROFILES = {'human': 'Human', 'yam': 'YAM', 'dex': 'Dex', 'tianji': 'Tianji'}
+
+class DataEnginePipeline:
+
+    @staticmethod
+    def _inputs(video, profile, left_wrist, right_wrist, yam32, instruction):
+        from .media import InputError, probe, check_synchronized
+        if not isinstance(profile, str) or profile.lower() not in _PROFILES:
+            raise InputError('profile must be human, yam, dex, or tianji')
+        canonical = _PROFILES[profile.lower()]
+        if bool(left_wrist) != bool(right_wrist):
+            raise InputError('Provide both left and right wrist videos together.')
+        if canonical == 'Human' and left_wrist:
+            raise InputError('The human profile accepts one video.')
+        if not isinstance(yam32, bool):
+            raise InputError('yam32 must be true or false.')
+        if yam32 and (canonical != 'YAM' or not left_wrist):
+            raise InputError('yam32 requires YAM and all three camera views.')
+        if not isinstance(instruction, str) or len(instruction.strip()) > 2000:
+            raise InputError('instruction must be a string of at most 2000 characters.')
+        paths = {'main': video}
+        if left_wrist:
+            paths.update(left_wrist=left_wrist, right_wrist=right_wrist)
+        videos = {}
+        for role, value in paths.items():
+            if not isinstance(value, (str, Path)) or not str(value).strip():
+                raise InputError('Each camera path must be a nonempty local path.')
+            path = Path(value).expanduser().resolve(strict=True)
+            if not path.is_file() or not 0 < path.stat().st_size <= MAX_UPLOAD_BYTES:
+                raise InputError('Each input must be a nonempty local video of at most 256 MiB.')
+            videos[role] = probe(path)
+        check_synchronized(videos)
+        return (videos, canonical)
+
+    def inspect_inputs(self, video, *, profile='human', left_wrist=None, right_wrist=None, yam32=False, instruction=''):
+        videos, canonical = self._inputs(video, profile, left_wrist, right_wrist, yam32, instruction)
+        return {'status': 'inputs_validated', 'profile': canonical, 'yam32': yam32, 'duration_s': videos['main'].duration_s, 'model_revisions': MODELS, 'new_model_calls': 0, 'cameras': {role: {'sha256': v.sha256, 'source_fps': str(v.rate), 'frames': len(v.pts_us), 'size': list(v.size), 'constant_frame_rate': v.cfr, 'duration_s': v.duration_s} for role, v in videos.items()}}
+
+    def annotate(self, video, *, profile='human', left_wrist=None, right_wrist=None, yam32=False, instruction='', progress=None, include_audit=False):
+        import torch
+        if not torch.cuda.is_available():
+            raise RuntimeError('CUDA is required for inference. Use --validate-only to check inputs on CPU.')
+        from . import pipeline
+        with _REQUEST_LOCK:
+            videos, canonical = self._inputs(video, profile, left_wrist, right_wrist, yam32, instruction)
+            result = pipeline.annotate(videos, canonical, yam32, instruction, progress or (lambda *a, **k: None))
+        duration = result['duration_s']
+        for segment in result['segments']:
+            if not (math.isfinite(segment['start_s']) and math.isfinite(segment['end_s']) and (0 <= segment['start_s'] < segment['end_s'] <= duration + 0.001)):
+                raise RuntimeError('The pipeline returned an invalid caption interval.')
+        if not include_audit:
+            result.pop('pipeline', None)
+        return result

batch_utils/bootstrap.py

@@ -0,0 +1,18 @@
+import hashlib
+import json
+import sys
+from .config import VENDOR
+
+def initialize():
+    manifest = json.loads((VENDOR / 'MANIFEST.json').read_text())
+    for relative, expected in manifest.items():
+        path = VENDOR / relative
+        if not path.resolve().is_relative_to(VENDOR.resolve()):
+            raise RuntimeError('Invalid runtime manifest path')
+        if hashlib.sha256(path.read_bytes()).hexdigest() != expected:
+            raise RuntimeError(f'Production code changed: {relative}')
+    for path in [VENDOR, VENDOR / 'support', VENDOR / 'production_adapter']:
+        sys.path.insert(0, str(path))
+    import adapter
+    return adapter
+production = initialize()

batch_utils/cli.py

@@ -0,0 +1,81 @@
+from pathlib import Path
+import argparse
+import json
+import re
+import sys
+from .api import DataEnginePipeline
+
+def parser():
+    p = argparse.ArgumentParser(description='FLORA: TempSeg -> Captioner with native video preprocessing.')
+    source = p.add_mutually_exclusive_group(required=True)
+    source.add_argument('--video', type=Path, help='Main/egocentric local video.')
+    source.add_argument('--input-jsonl', type=Path, help='One video job per line; paths relative to this JSONL.')
+    p.add_argument('--profile', choices=['human', 'yam', 'dex', 'tianji'], default='human')
+    p.add_argument('--left-wrist', type=Path)
+    p.add_argument('--right-wrist', type=Path)
+    p.add_argument('--yam32', action='store_true', help='Use the native 32-second three-view YAM temporal profile.')
+    p.add_argument('--instruction', default='', help='Optional episode instruction; blank by default.')
+    p.add_argument('--output', type=Path, help='Single-video result; default annotations.json.')
+    p.add_argument('--output-dir', type=Path, help='Required with --input-jsonl for inference.')
+    p.add_argument('--validate-only', action='store_true', help='Check media/timestamps on CPU; no weight downloads or generation.')
+    p.add_argument('--include-audit', action='store_true', help='Include temporal/model-input audit records in the result.')
+    p.add_argument('--overwrite', action='store_true', help='Explicitly allow replacing existing output files.')
+    return p
+
+def jobs(args, p):
+    if args.video:
+        if args.output_dir:
+            p.error('--output-dir is for JSONL jobs; use --output for a single video')
+        return [(dict(video=args.video, profile=args.profile, left_wrist=args.left_wrist, right_wrist=args.right_wrist, yam32=args.yam32, instruction=args.instruction), args.output or Path('annotations.json'))]
+    if args.output or args.left_wrist or args.right_wrist or args.yam32 or args.instruction or (args.profile != 'human'):
+        p.error('Put per-video profile/cameras/instruction in JSONL rows.')
+    if not args.output_dir and (not args.validate_only):
+        p.error('--output-dir is required with --input-jsonl')
+    source = args.input_jsonl.expanduser().resolve(strict=True)
+    items, ids = ([], set())
+    allowed = {'id', 'video', 'profile', 'left_wrist', 'right_wrist', 'yam32', 'instruction'}
+    for line_number, line in enumerate(source.read_text().splitlines(), 1):
+        if not line.strip():
+            continue
+        row = json.loads(line)
+        if not isinstance(row, dict) or set(row) - allowed or 'video' not in row:
+            raise ValueError(f'Invalid JSONL job on line {line_number}.')
+        name = row.pop('id', f'video_{line_number:06d}')
+        if not isinstance(name, str) or not re.fullmatch('[A-Za-z0-9][A-Za-z0-9_.-]{0,95}', name) or name in ids:
+            raise ValueError(f'Invalid or duplicate job id on line {line_number}.')
+        ids.add(name)
+        for key in ['video', 'left_wrist', 'right_wrist']:
+            if row.get(key) is not None:
+                if not isinstance(row[key], str) or not row[key].strip():
+                    raise ValueError(f'Invalid camera path on JSONL line {line_number}.')
+                value = Path(row[key]).expanduser()
+                row[key] = value if value.is_absolute() else source.parent / value
+        items.append((row, (args.output_dir or Path('.')) / (name + '.json')))
+    if not items:
+        raise ValueError('The JSONL contains no video jobs.')
+    return items
+
+def main(argv=None):
+    p = parser()
+    args = p.parse_args(argv)
+    try:
+        items = jobs(args, p)
+        if not args.validate_only and (not args.overwrite):
+            for _, output in items:
+                if output.exists():
+                    raise FileExistsError(f'Output already exists: {output}. Use --overwrite to replace it.')
+        engine = DataEnginePipeline()
+        for kwargs, output in items:
+            if args.validate_only:
+                print(json.dumps(engine.inspect_inputs(**kwargs), allow_nan=False))
+                continue
+            result = engine.annotate(**kwargs, include_audit=args.include_audit, progress=lambda fraction, desc='': print(f'{fraction:.0%} {desc}', file=sys.stderr, flush=True))
+            payload = json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False) + '\n'
+            output.parent.mkdir(parents=True, exist_ok=True)
+            with output.open('w' if args.overwrite else 'x') as f:
+                f.write(payload)
+            print(json.dumps({'output': str(output), 'segments': len(result['segments'])}))
+        return 0
+    except (OSError, ValueError, RuntimeError) as error:
+        print(f'FLORA inference failed: {error}', file=sys.stderr)
+        return 1

batch_utils/config.py

@@ -0,0 +1,12 @@
+from pathlib import Path
+import json
+import os
+ROOT = Path(__file__).resolve().parent
+VENDOR = ROOT / 'vendor'
+MODELS = json.loads((ROOT / 'models.json').read_text())
+MODEL_CACHE = Path(os.environ.get('FLORA_MODEL_CACHE', Path.home() / '.cache' / 'flora' / 'models'))
+MAX_DURATION_S = 120
+MAX_UPLOAD_BYTES = 256 * 1024 * 1024
+MAX_DIMENSION = 4096
+MAX_FRAME_COUNT = 7200
+PROFILE_POOLS = {'Human': 'human', 'Human actions': 'human', 'YAM': 'yam', 'Dex': 'dex', 'Tianji': 'tianji', 'Robot actions': 'yam'}

batch_utils/media.py

@@ -0,0 +1,106 @@
+from dataclasses import dataclass
+from fractions import Fraction
+from pathlib import Path
+import hashlib
+import av
+from .config import MAX_DURATION_S, MAX_DIMENSION, MAX_FRAME_COUNT
+
+class InputError(ValueError):
+    pass
+
+@dataclass(frozen=True)
+class Video:
+    path: Path
+    sha256: str
+    rate: Fraction
+    pts_us: tuple
+    duration_us: int
+    size: tuple
+    cfr: bool
+
+    @property
+    def duration_s(self):
+        return self.duration_us / 1000000
+
+def probe(path):
+    with av.open(str(path), options={'err_detect': 'explode'}) as container:
+        if len(container.streams.video) != 1:
+            raise InputError('Use a file containing one video stream.')
+        stream = container.streams.video[0]
+        stream.codec_context.thread_count = 1
+        rate = stream.average_rate
+        if not rate or rate < 4:
+            raise InputError('The input needs at least 4 source frames per second.')
+        size = (stream.codec_context.width, stream.codec_context.height)
+        if min(size) <= 0 or max(size) > MAX_DIMENSION:
+            raise InputError('Video dimensions must be at most 4096 pixels per side.')
+        times, zero, last_length = ([], None, None)
+        for frame in container.decode(stream):
+            if frame.pts is None:
+                raise InputError('Every source frame needs a presentation timestamp.')
+            zero = frame.pts if zero is None else zero
+            t = Fraction(frame.pts - zero) * stream.time_base * 1000000
+            if times and t <= times[-1]:
+                raise InputError('Source timestamps must be strictly increasing.')
+            if t >= MAX_DURATION_S * 1000000 or len(times) >= MAX_FRAME_COUNT:
+                raise InputError(f'Video exceeds {MAX_DURATION_S:g} seconds or {MAX_FRAME_COUNT} frames.')
+            times.append(t)
+            last_length = Fraction(frame.duration) * stream.time_base * 1000000 if frame.duration else None
+        if len(times) < 2:
+            raise InputError('The video contains too few source frames.')
+        tail = last_length or Fraction(1000000, rate)
+        duration = round(times[-1] + tail)
+        if duration > MAX_DURATION_S * 1000000:
+            raise InputError(f'Video exceeds {MAX_DURATION_S:g} seconds.')
+        cfr = all((abs(t - Fraction(i * 1000000, rate)) <= 1 for i, t in enumerate(times)))
+    with path.open('rb') as f:
+        digest = hashlib.file_digest(f, 'sha256').hexdigest()
+    return Video(path, digest, rate, tuple(times), duration, size, cfr)
+
+def check_synchronized(videos):
+    main = videos['main']
+    if len(videos) == 3:
+        for role, video in videos.items():
+            if video.rate != main.rate or video.pts_us != main.pts_us or video.duration_us != main.duration_us:
+                raise InputError('Three camera files must have identical relative frame timestamps, frame counts and duration.')
+        if not main.cfr:
+            raise InputError('The three-view profile requires constant-rate synchronized source videos.')
+
+def decode_selected(video, wanted, canonical=False):
+    wanted = set(wanted)
+    result = {}
+    if not wanted:
+        return result
+    if min(wanted) < 0 or max(wanted) >= len(video.pts_us):
+        raise InputError('A requested source frame is outside the video.')
+    with av.open(str(video.path), options={'err_detect': 'explode'}) as container:
+        stream = container.streams.video[0]
+        stream.codec_context.thread_count = 1
+        zero = None
+        for i, frame in enumerate(container.decode(stream)):
+            zero = frame.pts if zero is None else zero
+            if i in wanted:
+                if Fraction(frame.pts - zero) * stream.time_base * 1000000 != video.pts_us[i]:
+                    raise InputError('Source frame timestamps changed while decoding.')
+                use_canonical = canonical and frame.format.name == 'yuv420p' and (frame.width % 16 == 0) and (frame.height % 2 == 0) and (int(frame.colorspace) in (2, 5, 6)) and (int(frame.color_range) in (0, 1))
+                if use_canonical:
+                    from canonical_rgb import canonical_rgb
+                    result[i] = canonical_rgb(frame)
+                else:
+                    result[i] = frame.to_image().convert('RGB')
+            if i >= max(wanted):
+                break
+    if set(result) != wanted:
+        raise InputError('A requested source frame could not be decoded.')
+    return result
+
+def frames_for(videos, wanted, mosaic='YAM'):
+    frames = {role: decode_selected(video, wanted, canonical=len(videos) == 3) for role, video in videos.items()}
+    if len(videos) == 1:
+        return frames['main']
+    from composers import compose_top_dominant_mosaic, compose_frame
+    if mosaic == 'YAM':
+        return {i: compose_top_dominant_mosaic(frames['main'][i], frames['left_wrist'][i], frames['right_wrist'][i]) for i in wanted}
+    if mosaic == 'Dex':
+        return {i: compose_frame(dict(ego=frames['main'][i], left_wrist=frames['left_wrist'][i], right_wrist=frames['right_wrist'][i]), ('ego', 'left_wrist', 'right_wrist')) for i in wanted}
+    raise InputError('Choose a supported camera layout.')

batch_utils/models.json

@@ -0,0 +1,18 @@
+{
+  "tempseg": {
+    "repo": "nvidia/Flora-TempSeg",
+    "revision": "114a1076f70eb53f4a58ac045505db3087e79347",
+    "sha256": "1d80fe544e4cd1a7ff6b0ea4be71e988972408dab81d4c8af72e3a628739d6c3",
+    "repo_env": "FLORA_TEMPSEG_REPO"
+  },
+  "captioner": {
+    "repo": "nvidia/Flora-Captioner",
+    "revision": "fef0bbfe11829ca7f3539d58da7a6a2c2f109dc6",
+    "source_manifest_identity": "2ab6cafd286df7f446bfb84fbc4af7c54773d5e890dd277ac5a0453d0d4fb778",
+    "repo_env": "FLORA_CAPTIONER_REPO"
+  },
+  "tempseg_base": {
+    "repo": "Qwen/Qwen3.5-9B",
+    "revision": "c202236235762e1c871ad0ccb60c8ee5ba337b9a"
+  }
+}

batch_utils/models.py

@@ -0,0 +1,189 @@
+from collections import Counter
+from pathlib import Path
+import gc
+import hashlib
+import json
+import os
+import threading
+import time
+from datetime import datetime, timezone
+from .config import MODELS, VENDOR, MODEL_CACHE
+_LOCK = threading.Lock()
+_MANIFEST_LOCK = threading.Lock()
+_VERIFIED_MANIFESTS = {}
+
+class NotReady(RuntimeError):
+    pass
+
+def verify_cuda_residency(model):
+    devices = Counter((str(p.device) for p in model.parameters()))
+    if not devices or any((not d.startswith('cuda:') for d in devices)):
+        raise NotReady('GPU memory is insufficient for this native Captioner configuration.')
+    return dict(devices)
+
+def runtime_event(event, **values):
+    print('FLORA_RUNTIME ' + json.dumps(dict(event=event, utc=datetime.now(timezone.utc).isoformat(), **values), sort_keys=True), flush=True)
+
+def ready():
+    if os.environ.get('FLORA_ENABLE_INFERENCE', '0') != '1':
+        return (False, 'Preview ready · live annotation is paused.')
+    import torch
+    if not torch.cuda.is_available():
+        return (False, 'Preview ready · GPU hardware is needed for annotation.')
+    if sum((torch.cuda.get_device_properties(i).total_memory for i in range(torch.cuda.device_count()))) < 75 * 1024 ** 3:
+        return (False, 'Annotation needs an 80 GB GPU or a validated multi-GPU configuration.')
+    if not os.environ.get('HF_TOKEN'):
+        return (False, 'Annotation is waiting for access to the private weights.')
+    return (True, 'Ready to annotate · one video at a time.')
+
+def snapshot(kind):
+    from huggingface_hub import snapshot_download
+    spec = dict(MODELS[kind])
+    spec['repo'] = os.environ.get(spec.get('repo_env', ''), spec['repo'])
+    if not spec['repo']:
+        raise ValueError('Set ' + spec['repo_env'] + ' to the released checkpoint repository.')
+    patterns = ['adapter_config.json', 'adapter_model.safetensors', 'config.json', 'generation_config.json', 'model.safetensors.index.json', 'model-*.safetensors', 'processor_config.json', 'tokenizer.json', 'tokenizer_config.json', 'chat_template.jinja', 'README.md', 'MODEL_PROVENANCE.json', 'SOURCE_MANIFEST.sha256']
+    return Path(snapshot_download(spec['repo'], revision=spec['revision'], token=os.environ.get('HF_TOKEN'), allow_patterns=patterns, local_dir=MODEL_CACHE / kind))
+
+def processor(kind, path):
+    from transformers import AutoProcessor
+    from transformers.models.qwen3_vl.video_processing_qwen3_vl import Qwen3VLVideoProcessor
+    proc = AutoProcessor.from_pretrained(path, local_files_only=True, use_fast=True, min_pixels=1024, max_pixels=589824, video_processor=Qwen3VLVideoProcessor.from_pretrained(path, local_files_only=True, fps=4 if kind == 'tempseg' else 8, max_frames=128 if kind == 'tempseg' else 40))
+    proc.chat_template = (path / 'chat_template.jinja').read_text()
+    return proc
+
+def validate_manifest(path):
+    with _MANIFEST_LOCK:
+        _validate_manifest(path)
+
+def _validate_manifest(path):
+    started = time.monotonic()
+    allowed_root = path.parent.parent if path.parent.name == 'snapshots' else path
+    manifest = (path / 'SOURCE_MANIFEST.sha256').read_bytes()
+    key = str(path.resolve())
+    manifest_sha = hashlib.sha256(manifest).hexdigest()
+    files = []
+
+    def fingerprint(source):
+        resolved = source.resolve()
+        if not resolved.is_relative_to(allowed_root.resolve()):
+            raise RuntimeError('Invalid model manifest')
+        stat = resolved.stat()
+        return (str(resolved), stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns, stat.st_ctime_ns)
+    for line in manifest.decode().splitlines():
+        if not line.strip():
+            continue
+        expected, filename = line.split(maxsplit=1)
+        relative = Path(filename.lstrip('*'))
+        if relative.is_absolute() or '..' in relative.parts:
+            raise RuntimeError('Invalid model manifest')
+        source = path / relative
+        if not source.resolve().is_relative_to(allowed_root.resolve()):
+            runtime_event('manifest_path_rejected', file=filename, snapshot=str(path), target=str(source.resolve()), allowed_root=str(allowed_root.resolve()))
+            raise RuntimeError('Invalid model manifest')
+        files.append((source, expected, fingerprint(source)))
+    before = tuple(((str(source), expected, stamp) for source, expected, stamp in files))
+    if _VERIFIED_MANIFESTS.get(key) == (manifest_sha, before):
+        runtime_event('manifest_verified_reuse', kind=path.name, files=len(files), payload_bytes_read=0, elapsed_s=round(time.monotonic() - started, 3))
+        return
+    _VERIFIED_MANIFESTS.pop(key, None)
+    for source, expected, _ in files:
+        with source.open('rb') as f:
+            if hashlib.file_digest(f, 'sha256').hexdigest() != expected:
+                raise RuntimeError('A model asset failed its integrity check')
+    after = tuple(((str(source), expected, fingerprint(source)) for source, expected, _ in files))
+    if before != after or manifest != (path / 'SOURCE_MANIFEST.sha256').read_bytes():
+        raise RuntimeError('A model asset changed during its integrity check')
+    _VERIFIED_MANIFESTS[key] = (manifest_sha, after)
+    runtime_event('manifest_full_verification', kind=path.name, files=len(files), payload_bytes_read=sum((stamp[3] for _, _, stamp in files)), elapsed_s=round(time.monotonic() - started, 3))
+
+def preflight_adapter_download():
+    path = snapshot('tempseg')
+    validate_manifest(path)
+    runtime_event('adapter_download_preflight_pass', revision=MODELS['tempseg']['revision'], adapter_sha256=MODELS['tempseg']['sha256'], directory=str(path))
+
+class Session:
+
+    def __init__(self):
+        self.model = self.proc = None
+        self.audit = {}
+
+    def load(self, kind):
+        import torch
+        previous_threads = torch.get_num_threads()
+        torch.set_num_threads(4)
+        load_started = time.monotonic()
+        from transformers import AutoModelForImageTextToText
+        from transformers.models.qwen3_5.modeling_qwen3_5 import Qwen3_5PreTrainedModel
+        from production_gdn_fix import _keep_native_gdn_fp32
+        from verify_gdn import verify
+        self.close()
+        runtime_event('model_load_start', kind=kind, cpu_threads=torch.get_num_threads(), previous_cpu_threads=previous_threads)
+        _keep_native_gdn_fp32(Qwen3_5PreTrainedModel)
+        path = snapshot(kind)
+        runtime_event('model_assets_ready', kind=kind, elapsed_s=round(time.monotonic() - load_started, 3))
+        validate_manifest(path)
+        runtime_event('model_assets_verified', kind=kind, elapsed_s=round(time.monotonic() - load_started, 3))
+        if kind == 'tempseg':
+            base_spec = MODELS['tempseg_base']
+            from huggingface_hub import snapshot_download
+            base = snapshot_download(base_spec['repo'], revision=base_spec['revision'], token=os.environ.get('HF_TOKEN'), allow_patterns=['*.safetensors', '*.json', '*.jinja'])
+            from tempseg_fp32_loader import load
+            records = {}
+            self.model = load({'base': base, 'export': str(path), 'adapter_sha256': MODELS[kind]['sha256']}, lambda name, value: records.__setitem__(name, value))
+            self.audit[kind] = records
+        else:
+            index = json.loads((path / 'model.safetensors.index.json').read_text())
+            extra = {k for k in index['weight_map'] if '_extra_state' in k}
+            memory = {i: f'{max(1, int(torch.cuda.get_device_properties(i).total_memory / 1024 ** 3) - 12)}GiB' for i in range(torch.cuda.device_count())}
+            self.model, info = AutoModelForImageTextToText.from_pretrained(path, local_files_only=True, dtype=torch.bfloat16, attn_implementation={'text_config': 'sdpa', 'vision_config': 'sdpa'}, device_map='balanced', max_memory=memory, output_loading_info=True)
+            if any((info.get(k) for k in ['missing_keys', 'mismatched_keys', 'error_msgs'])) or not set(info.get('unexpected_keys', [])) <= extra:
+                raise RuntimeError('The released Captioner did not load completely')
+            devices = verify_cuda_residency(self.model)
+            self.model.eval()
+            self.audit[kind] = {'gdn': verify(self.model, path), 'loading_info': info, 'parameter_devices': devices, 'parameter_dtypes': dict(Counter((str(p.dtype) for p in self.model.parameters())))}
+        runtime_event('model_parameters_verified', kind=kind, elapsed_s=round(time.monotonic() - load_started, 3))
+        self.proc = processor(kind, path)
+        if kind == 'tempseg':
+            adapter = self.audit[kind]['ADAPTER_VALUE_VERIFICATION.json']
+            gdn = self.audit[kind]['GDN_VALUE_VERIFICATION.json']
+            verification = dict(adapter_tensors=adapter['tensors'], adapter_values_exact=adapter['all_saved_values_exact_before_and_after_placement'], adapter_fp32=adapter['holders_fp32_before_assignment'], gdn_values_exact=gdn['all_values_equal_saved_checkpoint'], gdn_fp32=gdn['all_runtime_fp32'])
+        else:
+            gdn = self.audit[kind]['gdn']
+            verification = dict(gdn_values_exact=gdn['all_values_equal_saved_checkpoint'], gdn_fp32=gdn['all_runtime_fp32'], parameter_dtypes=self.audit[kind]['parameter_dtypes'])
+        runtime_event('model_ready', kind=kind, revision=MODELS[kind]['revision'], torch_version=torch.__version__, gpu=torch.cuda.get_device_name(0), allocated_gib=round(torch.cuda.memory_allocated() / 1024 ** 3, 3), **verification)
+        return (self.model, self.proc)
+
+    def close(self):
+        self.model = self.proc = None
+        gc.collect()
+        import torch
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+
+def generate(model, proc, inputs, budget):
+    import torch
+    inputs = dict(inputs)
+    inputs.pop('video_metadata', None)
+    n = inputs['input_ids'].shape[1]
+    if n + budget > 32768:
+        raise InputErrorForModel('The selected video exceeds the native model context budget.')
+    input_tensors = {k: {'dtype': str(v.dtype), 'shape': list(v.shape), 'sha256': hashlib.sha256(v.detach().cpu().contiguous().view(torch.uint8).numpy().tobytes()).hexdigest()} for k, v in inputs.items() if torch.is_tensor(v)}
+    input_tensor_digest = hashlib.sha256(json.dumps(input_tensors, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
+    device = next(model.parameters()).device
+    args = {k: v.to(device) if torch.is_tensor(v) else v for k, v in inputs.items()}
+    eos = proc.tokenizer.eos_token_id
+    started = time.monotonic()
+    runtime_event('generation_start', input_tokens=n, output_limit=budget, input_tensor_digest=input_tensor_digest, input_tensors=input_tensors)
+    with torch.inference_mode():
+        output = model.generate(**args, do_sample=False, num_beams=1, max_new_tokens=budget, eos_token_id=eos, pad_token_id=proc.tokenizer.pad_token_id, use_cache=True, max_time=300)
+    ids = output[0, n:].cpu().tolist()
+    stop = bool(ids) and ids[-1] in ([eos] if isinstance(eos, int) else eos)
+    text = proc.tokenizer.decode(ids, skip_special_tokens=True).strip()
+    runtime_event('generation_end', input_tokens=n, output_limit=budget, generated_tokens=len(ids), complete=stop and bool(text), elapsed_s=round(time.monotonic() - started, 3), peak_allocated_gib=round(torch.cuda.max_memory_allocated() / 1024 ** 3, 3))
+    if not stop or not text:
+        raise InputErrorForModel('The model did not return a complete output for this input.')
+    return text
+
+class InputErrorForModel(RuntimeError):
+    pass

batch_utils/pipeline.py

@@ -0,0 +1,169 @@
+import asyncio
+from fractions import Fraction
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+import time
+from .bootstrap import production as a
+from .config import MODELS, PROFILE_POOLS
+from .media import frames_for, check_synchronized, InputError
+from .models import Session, generate, runtime_event
+from .preprocessing import temporal_inputs, human_transport, instruction_suffix, three_view_caption_inputs
+
+class Predictor:
+
+    def __init__(self, videos, source, pool, bound, model, proc, instruction, progress):
+        self.videos, self.source, self.pool, self.bound = (videos, source, pool, bound)
+        self.model, self.proc, self.instruction, self.progress = (model, proc, instruction, progress)
+        self.calls = []
+
+    def predict(self, request):
+        from flora_runtime.video_segmentation.fine_subtask import FINE_RESPONSE_SCHEMA
+        if request['source']['source_id'] != self.source.source_id or request['model_binding'] != self.bound:
+            raise RuntimeError('Production request binding changed')
+        predictions = []
+        for payload in request['windows']:
+            window = a.window_from_payload(payload, self.source)
+            plan = a.temporal_frame_plan(window, list(self.videos['main'].pts_us))
+            mosaic = 'Dex' if self.pool in ('dex', 'tianji') else 'YAM'
+            images = frames_for(self.videos, set(plan['source_frame_indices']), mosaic=mosaic)
+            if self.pool == 'human':
+                images = human_transport(images)
+            frames = [images[i] for i in plan['source_frame_indices']]
+            inputs, prompt = temporal_inputs(frames, window, self.pool, self.proc, list(self.videos), self.instruction)
+            self.progress(0.25, desc=f'Finding actions · window {len(self.calls) + 1}')
+            text = generate(self.model, self.proc, inputs, 1024)
+            predictions.append(dict(scope_id=window.scope_id, prediction_json=text))
+            self.calls.append(dict(window=window.worker_payload(), frame_plan=plan, prompt_sha256=hashlib.sha256(prompt.encode()).hexdigest(), output=text))
+        return dict(schema_version=FINE_RESPONSE_SCHEMA, source_id=self.source.source_id, model_io_contract=self.bound['model_io_contract'], model_binding=self.bound, predictions=predictions)
+
+def segment(videos, pool, session, instruction, progress):
+    model, proc = session.load('tempseg')
+    main = videos['main']
+    source = a.AnnotationSource(source_id='upload_' + main.sha256[:24], location=str(main.path), duration_s=main.duration_s, fps=float(main.rate))
+    bound = dict(candidate_id='flora_final', checkpoint_sha256=MODELS['tempseg']['sha256'], model_io_contract=a.fine_window_profile(a.PROFILES[pool]).model_io_contract)
+    predictor = Predictor(videos, source, pool, bound, model, proc, instruction, progress)
+    if len(videos) == 3:
+        fine = a.FineSubtaskSegmenter(predictor=predictor, model_binding=bound, profile_id='robot_atomic_intent_v1', window_profile_id=a.PROFILES[pool], fail_on_unresolved=True, batch_concurrency=1, unresolved_stitch_tolerance_ticks=2, quality_policy=a.FineSegmentationQualityPolicy(drop_guard_only_segments=True))
+        engine = a.HierarchicalSubtaskSegmenter(coarse_segmenter=a.NoCoarse(), fine_segmenter=fine, fine_prior_mode='none', fine_failure_policy='adaptive_fine_then_conservative_completion', fine_output_coverage_policy='allow_gaps', adaptive_fine_retry_max_windows=32)
+    else:
+        engine = a.production_segmenter(predictor, bound, pool)
+    segments = asyncio.run(engine.segment_async(source))
+    return ([dict(start_s=s.span.start_s, end_s=s.span.end_s, metadata=dict(s.metadata)) for s in segments], predictor.calls)
+
+def single_plans(videos, segments, profile):
+    if profile in ('Human', 'Human actions'):
+        from pts_input_adapter import native_blocks, microseconds
+        from flora_runtime.subtask_captioning import u2_contract
+        main = videos['main']
+        pts = [float(t / 1000000) for t in main.pts_us]
+        plans, dispositions = ([], [])
+        for i, s in enumerate(segments):
+            record = dict(event_id=f'event_{i:04d}', event_index=i, start_s=s['start_s'], end_s=s['end_s'])
+            try:
+                blocks = native_blocks(pts, microseconds(s['start_s']), microseconds(s['end_s']), main.duration_us, main.rate, u2_contract)
+            except ValueError:
+                dispositions.append(dict(record, status='unsupported_target_frames'))
+                continue
+            plans.append(dict(record, blocks=blocks, source_fps_num=main.rate.numerator, source_fps_den=main.rate.denominator))
+            dispositions.append(dict(record, status='caption_input'))
+        return (plans, dispositions)
+    main = videos['main']
+    if not main.cfr:
+        raise InputError('Use the Human actions profile for videos with variable frame timestamps.')
+    return a.caption_plans(segments, dict(fps_num=main.rate.numerator, fps_den=main.rate.denominator, frame_count=len(main.pts_us)))
+
+def caption_single(videos, segments, profile, session, instruction, progress):
+    from flora_unified.unified_input import processor_inputs, prompt as native_prompt
+    from transformers.video_utils import VideoMetadata
+    plans, dispositions = single_plans(videos, segments, profile)
+    if not plans:
+        return ([dict(d, caption='', status=d['status']) for d in dispositions], [])
+    model, proc = session.load('captioner')
+    rows, audit = ({}, [])
+    with tempfile.TemporaryDirectory(prefix='flora-frames-') as temporary:
+        root = Path(temporary)
+        wanted = {i for plan in plans for block in plan['blocks'] for i in block['frame_indices']}
+        images = frames_for(videos, wanted)
+        receipts = {}
+        for i, image in images.items():
+            path = root / f'{i:012d}.png'
+            image.save(path, format='PNG')
+            receipts[i] = dict(path=path.name, png_sha256=hashlib.sha256(path.read_bytes()).hexdigest(), size=list(image.size))
+        for n, plan in enumerate(plans):
+            progress(0.55 + 0.4 * n / len(plans), desc=f'Writing captions · {n + 1}/{len(plans)}')
+            sample = a.caption_sample(plan, receipts, ['main'])
+            if instruction.strip():
+                from flora_unified.unified_input import load_frames
+                frames, _ = load_frames(sample, root)
+                text = proc.apply_chat_template([dict(role='user', content=native_prompt(sample, proc.video_token) + instruction_suffix(instruction))], tokenize=False, add_generation_prompt=True, enable_thinking=False)
+                metadata = [VideoMetadata(total_num_frames=len(f), fps=float(videos['main'].rate), frames_indices=list(b['block_local_timestamp_indices'])) for f, b in zip(frames, sample['blocks'], strict=True)]
+                inputs = proc(text=[text], videos=frames, video_metadata=metadata, do_sample_frames=False, return_metadata=True, return_tensors='pt')
+            else:
+                inputs, _, text = processor_inputs(sample, root, proc)
+            caption = generate(model, proc, inputs, 512)
+            rows[plan['event_id']] = dict(start_s=plan['start_s'], end_s=plan['end_s'], caption=caption, status='complete')
+            audit.append(dict(event_id=plan['event_id'], blocks=plan['blocks'], prompt_sha256=hashlib.sha256(text.encode()).hexdigest()))
+    return ([rows.get(d['event_id'], dict(start_s=d['start_s'], end_s=d['end_s'], caption='', status=d['status'])) for d in dispositions], audit)
+
+def caption_three(videos, segments, session, instruction, progress, profile='YAM'):
+    from plans import plans_for_episode
+    from robot_caption_geometry import normalized, source_frames
+    main = videos['main']
+    from math import lcm
+    scale = lcm(*(t.denominator for t in main.pts_us)) * 1000000
+    ticks = [int(t * scale / 1000000) for t in main.pts_us]
+    clock = dict(ticks=ticks, time_base=str(Fraction(1, scale)), fps=str(main.rate), duration_s=main.duration_s, first_pts=0)
+    group = 'Dex' if profile in ('Dex', 'Tianji') else 'YAM'
+    episode = dict(episode_id='upload', source_id=main.sha256, group=group, units=[])
+    prepared, rows, audit = ([], [], [])
+    for i, s in enumerate(segments):
+        unit = dict(target_id=f'event_{i:04d}', unit_id=f'prediction:{i}', native_scope_s=[s['start_s'], s['end_s']])
+        try:
+            plan = plans_for_episode(dict(episode, units=[unit]), {role: clock for role in videos})[0]
+            prepared.append((i, plan))
+        except AssertionError:
+            rows.append((i, dict(start_s=s['start_s'], end_s=s['end_s'], caption='', status='unsupported_target_frames')))
+    if prepared:
+        model, proc = session.load('captioner')
+        for n, (i, plan) in enumerate(prepared):
+            progress(0.55 + 0.4 * n / len(prepared), desc=f'Writing captions · {n + 1}/{len(prepared)}')
+            wanted = {j for block in plan['blocks'] for j in block['indices']}
+            images = frames_for(videos, wanted, mosaic=group)
+            geometry = normalized(plan, images)
+            inputs, text = three_view_caption_inputs(plan, source_frames(geometry), proc, instruction)
+            caption = generate(model, proc, inputs, 512)
+            rows.append((i, dict(start_s=plan['native_scope_s'][0], end_s=plan['native_scope_s'][1], caption=caption, status='complete')))
+            audit.append(dict(event_id=plan['target_id'], blocks=plan['blocks'], common_geometry=geometry['common_geometry'], camera_layout=group, prompt_sha256=hashlib.sha256(text.encode()).hexdigest()))
+    return ([row for _, row in sorted(rows)], audit)
+
+def annotate(videos, profile, long_windows, instruction, progress):
+    check_synchronized(videos)
+    instruction_suffix(instruction)
+    if profile not in PROFILE_POOLS:
+        raise InputError('Choose a supported action profile.')
+    if len(videos) == 3 and profile in ('Human', 'Human actions'):
+        raise InputError('Choose YAM, Dex or Tianji when uploading wrist cameras.')
+    if long_windows and len(videos) != 3:
+        raise InputError('The 32-second profile requires three synchronized camera views.')
+    if long_windows and profile not in ('YAM', 'Robot actions'):
+        raise InputError('The 32-second profile is available for YAM videos.')
+    pool = 'yam32' if long_windows else PROFILE_POOLS[profile]
+    start = time.monotonic()
+    session = Session()
+    try:
+        progress(0.1, desc='Loading the action segmenter')
+        segments, temporal_audit = segment(videos, pool, session, instruction, progress)
+        runtime_event('segmentation_complete', windows=len(temporal_audit), segments=len(segments))
+        progress(0.5, desc='Stitching finished · loading the captioner')
+        if len(videos) == 3:
+            captions, caption_audit = caption_three(videos, segments, session, instruction, progress, profile=profile)
+        else:
+            captions, caption_audit = caption_single(videos, segments, profile, session, instruction, progress)
+        for row, segment_row in zip(captions, segments, strict=True):
+            row['segmentation_provenance'] = segment_row['metadata'].get('segmentation_provenance', 'fine')
+        runtime_event('annotation_complete', segments=len(captions), complete_captions=sum((r['status'] == 'complete' for r in captions)), elapsed_s=round(time.monotonic() - start, 3))
+        return dict(schema_version='flora_demo_annotation_v1', duration_s=videos['main'].duration_s, segments=captions, elapsed_s=round(time.monotonic() - start, 2), source_clock='seconds relative to the first native source presentation timestamp', camera_roles=list(videos), profile=profile, model_revisions=MODELS, input_sha256={role: video.sha256 for role, video in videos.items()}, pipeline=dict(temporal_windows=temporal_audit, caption_inputs=caption_audit))
+    finally:
+        session.close()

batch_utils/preprocessing.py

@@ -0,0 +1,56 @@
+from fractions import Fraction
+import io
+import json
+from PIL import Image
+from .bootstrap import production as a
+
+def instruction_suffix(instruction):
+    value = instruction.strip()
+    if not value:
+        return ''
+    if len(value) > 2000:
+        raise ValueError('Keep the optional task instruction below 2000 characters.')
+    return '\n' + json.dumps({'episode_instruction': value}, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False)
+
+def human_transport(frames):
+    from flora_runtime.exact_frame_serving import encode_regularized_image
+    output = {}
+    for index, image in frames.items():
+        payload = encode_regularized_image(image, encoding='jpeg_q98_444')['payload']
+        with Image.open(io.BytesIO(payload)) as decoded:
+            output[index] = decoded.convert('RGB')
+    return output
+
+def temporal_inputs(frames, window, pool, proc, roles, instruction=''):
+    from transformers.video_utils import VideoMetadata
+    from flora_runtime.exact_frame_serving import regularized_resize_dims
+    profile = window.window_profile
+    if len(frames) != profile.slot_count:
+        raise ValueError('Incorrect native temporal slot count')
+    regularized = []
+    for image in frames:
+        image = image.convert('RGB')
+        height, width = regularized_resize_dims(height=image.height, width=image.width, min_pixels=256, max_pixels=40000 if len(roles) == 1 else 160000)
+        if image.size != (width, height):
+            image = image.resize((width, height), Image.Resampling.BICUBIC)
+        regularized.append(image)
+    prompt = a.temporal_prompt(window, pool) + instruction_suffix(instruction)
+    text = proc.apply_chat_template([dict(role='user', content=prompt.replace('<video>', proc.video_token))], tokenize=False, add_generation_prompt=True, enable_thinking=False)
+    inputs = proc(text=[text], videos=[regularized], video_metadata=[VideoMetadata(total_num_frames=profile.slot_count, fps=profile.slot_fps, frames_indices=list(range(profile.slot_count)))], do_sample_frames=False, return_metadata=True, return_tensors='pt')
+    inputs.pop('video_metadata', None)
+    return (inputs, text)
+
+def three_view_caption_inputs(plan, frames, proc, instruction=''):
+    from transformers.video_utils import VideoMetadata
+    from plans import HEADING, INSTRUCTION
+    time_base = Fraction(plan['time_base'])
+    content, videos, metadata = ([HEADING], [], [])
+    for block in plan['blocks']:
+        times = ', '.join((f'{float(t * time_base):.9f}' for t in block['pts_ticks']))
+        content.append(block['role'] + ':\nFrames are in this order; native source timestamps in seconds: [' + times + '].\n' + proc.video_token)
+        videos.append([frames[i] for i in block['indices']])
+        metadata.append(VideoMetadata(total_num_frames=len(block['indices']), fps=float(Fraction(plan['clock_fps'])), frames_indices=[i - block['source_start_frame'] for i in block['indices']]))
+    content.append(INSTRUCTION + instruction_suffix(instruction))
+    text = proc.apply_chat_template([dict(role='user', content=''.join(content))], tokenize=False, add_generation_prompt=True, enable_thinking=False)
+    result = proc(text=[text], videos=videos, video_metadata=metadata, do_sample_frames=False, return_tensors='pt', padding=True)
+    return (result, text)

batch_utils/requirements.txt

@@ -0,0 +1,10 @@
+torch==2.11.0
+torchvision==0.26.0
+huggingface_hub==1.33.0
+transformers==5.8.1
+peft==0.18.1
+accelerate==1.12.0
+safetensors==0.7.0
+av==16.0.1
+numpy<3
+Pillow>=11,<13

batch_utils/vendor/MANIFEST.json

@@ -0,0 +1,29 @@
+{
+  "flora_unified/__init__.py": "9f56e761d79bfdb34304a012586cb04d16b435ef6130091a97702e559260a2f2",
+  "flora_unified/unified_input.py": "2aa2b521191b68913b03d7582b7a3bcfad49d90c014a1a1bb7571b5e4f08d7ef",
+  "production_adapter/TEMPSEG_PROMPTS.json": "66edd6fb43b7fa55cc6257beee5c42a1a4555ab9790320c31114d77bd07a05f1",
+  "production_adapter/adapter.py": "083c6788b561d726a85d9f3f8f516593a9f6f76a43f315581c821871e3f82312",
+  "production_adapter/core/flora_runtime/__init__.py": "9f56e761d79bfdb34304a012586cb04d16b435ef6130091a97702e559260a2f2",
+  "production_adapter/core/flora_runtime/exact_frame_serving.py": "f921e279b5d52f723b3e77e25441763683d05c6e9cd2f8ff204466a50255b346",
+  "production_adapter/core/flora_runtime/schema.py": "42c523e4db1a4da3fe0d0d578f3cf6e94aed7666c06e3f18081c209159de04b9",
+  "production_adapter/core/flora_runtime/subtask_captioning/__init__.py": "9f56e761d79bfdb34304a012586cb04d16b435ef6130091a97702e559260a2f2",
+  "production_adapter/core/flora_runtime/subtask_captioning/u2_contract.py": "873476e78a9e3e222ae9e47fcb7a04c8c18ef518b2847c401a2e1cfb2292fe90",
+  "production_adapter/core/flora_runtime/subtask_captioning/u2_worker.py": "870641228b0be442613d520232d451047742ca2ea41fd0cd51dbf0a58ba567d2",
+  "production_adapter/core/flora_runtime/video_segmentation/__init__.py": "9f56e761d79bfdb34304a012586cb04d16b435ef6130091a97702e559260a2f2",
+  "production_adapter/core/flora_runtime/video_segmentation/contracts.py": "8591d849b0b65fda6b21511557d4ee5eabd47f8332e2b6831f40d5bb472d978c",
+  "production_adapter/core/flora_runtime/video_segmentation/fine_subtask.py": "6fbf8337bab8491148a78a822187d74397b7f0552700252451a18cd7c78c8e04",
+  "production_adapter/core/flora_runtime/video_segmentation/fine_temporal.py": "32f2b99c97176f5e10934450da638577268c2e5d9977c2ea138ebf1b7eda58fa",
+  "production_adapter/core/flora_runtime/video_segmentation/fine_worker.py": "8a31a77180c8d406113d4b494644153564a6f46c8a9b68ad7babb3cc1695edec",
+  "production_adapter/core/flora_runtime/video_segmentation/interface.py": "a5424b2cc3a4c879552dd49bae207a58fb9e751359f441c03af167a2e1879359",
+  "production_adapter/core/flora_runtime/video_segmentation/scope_scheduler.py": "be95411b39124d83811fe74dcac2d911169ecf475a81b7460c64b19fbbc7e5db",
+  "production_adapter/core/flora_runtime/worker_transport.py": "30fb64869eb6a27ab96bf394c105f19fcb42590be4424c66b76b89e17631693a",
+  "production_gdn_fix.py": "a3f52604bfa2ec58901df63215df5e47d37e1d7974aa56d36a1e4efe8cad3fc6",
+  "pts_input_adapter.py": "d784d6b77005eb6945dee334e4c9136fbededfa2049d5945244f198517984767",
+  "robot_caption_geometry.py": "1c874d3c63359663276d31a3e9c3112a5982ad4000f44c808541b2471f5f7d2d",
+  "support/canonical_rgb.py": "1e3230c60b45d8e017060d8a5dd2b5f48e2199192f09dcff4d04c05b4349e35a",
+  "support/composers.py": "180b97a6a634979c1b0dbe720ed7a1abd818dc7ee31fdc4eeca7a8277e43d1b6",
+  "support/plans.py": "d75b918804c9ec10f8642d362a54a74c3eb875abf22e8a60228a0d6b2cbb570e",
+  "support/u2_contract.py": "50190914b719bbc9b8f2b53e52a6bd2b142a452e2fb1c5c897ea4c31123ad4dd",
+  "tempseg_fp32_loader.py": "7ee9889f5aa5909abf3ecd1c34e3ac2f81a32a2098d94ae3ba9f33831e7a07c0",
+  "verify_gdn.py": "96596ff2db4425f4dea71798aeabb581aa83e1ad8a878cfcb9efb8dbac80cda6"
+}

batch_utils/vendor/flora_unified/__init__.py

@@ -0,0 +1 @@
+pass

batch_utils/vendor/flora_unified/unified_input.py

@@ -0,0 +1,89 @@
+from fractions import Fraction
+import hashlib
+import math
+from pathlib import Path
+from PIL import Image
+INTENT_INSTRUCTION = 'Write one concise, self-contained English instruction for the local operation being attempted in TARGET.\nUse a natural imperative, concrete objects and the working hand; include meaningful support by the other hand and necessary source or destination. Say left hand, right hand or both hands. For a handover, name the object and both the giving and receiving hands.\nInfer the local goal from the video sequence. Use BEFORE and AFTER only to disambiguate TARGET, not to add operations occurring outside it. A failed or repeated attempt retains its established instruction; do not replace the goal with an accident, a success claim or a description of mechanical adjustments. Do not invent an object, hand role or hidden goal.\nOutput only the instruction, on one line, without timestamps, execution-quality commentary or the block names.'
+
+def validate(sample):
+    blocks = sample['blocks']
+    roles = [b['role'] for b in blocks]
+    if roles not in [['TARGET'], ['BEFORE', 'TARGET'], ['TARGET', 'AFTER'], ['BEFORE', 'TARGET', 'AFTER']]:
+        raise ValueError('Missing TARGET, duplicate or reordered temporal role')
+    if sample['camera_count'] not in (1, 3):
+        raise ValueError('Unimplemented camera renderer')
+    for b in blocks:
+        start, end = (b['source_start_frame'], b['source_end_frame_exclusive'])
+        indices = b['block_local_timestamp_indices']
+        n = len(b['paths'])
+        fps = Fraction(b['clock_fps_num'], b['clock_fps_den'])
+        if fps <= 0 or end <= start or (not n) or (n != len(indices)) or (n != len(b['frames'])):
+            raise ValueError('Invalid exact frame plan')
+        if any((i < 0 or i >= end - start for i in indices)):
+            raise ValueError('Frame outside block clock')
+        if any((a > b for a, b in zip(indices, indices[1:]))):
+            raise ValueError('Non-monotone timestamps')
+        original = b.get('original_descriptor', {})
+        if 'frame_indices' in original and original['frame_indices'] != [start + i for i in indices]:
+            raise ValueError('Source frame ordinals disagree with block-local clock')
+        if len(set(indices)) != n and (not any(original.get('frame_cache_padding_mask', []))):
+            raise ValueError('Repeated frame requires explicit padding evidence')
+        if Fraction(b['effective_fps_num'], b['effective_fps_den']) != Fraction(n) * fps / (end - start):
+            raise ValueError('Effective sampling rate does not match block duration')
+        if n % 2:
+            raise ValueError('No implicit temporal pair padding')
+    expected = sample['request']['target']['native_span']
+    target = next((b for b in blocks if b['role'] == 'TARGET'))
+    if expected['start_frame'] is not None and (target['source_start_frame'], target['source_end_frame_exclusive']) != (expected['start_frame'], expected['end_frame_exclusive']):
+        raise ValueError('Accepted target span changed')
+    text = sample['request']['target']['text']
+    if hashlib.sha256(text.encode()).hexdigest() != sample['request']['target']['sha256']:
+        raise ValueError('Accepted target changed')
+
+def prompt(sample, video_token):
+    validate(sample)
+    heading = ''
+    if sample['camera_count'] == 3:
+        if sample['camera_roles'] != ['main', 'left_wrist', 'right_wrist']:
+            raise ValueError('Camera roles do not match the mosaic renderer')
+        heading = 'Each frame is a synchronized camera mosaic: main camera above, left wrist camera below left, and right wrist camera below right. All panels show the same physical time. Describe each physical operation once across the views.\n\n'
+    return heading + '\n\n'.join((b['role'] + ':\n' + video_token for b in sample['blocks'])) + '\n\n' + INTENT_INSTRUCTION
+
+def load_frames(sample, root, preserve_canvas=True):
+    validate(sample)
+    videos = []
+    sizes = []
+    max_pixels = 524288 if preserve_canvas and sample['camera_count'] == 3 else 262144
+    for b in sample['blocks']:
+        frames = []
+        shapes = []
+        for path, receipt in zip(b['paths'], b['frames'], strict=True):
+            p = Path(root) / path
+            if not p.resolve().is_relative_to(Path(root).resolve()):
+                raise ValueError('Unowned frame path')
+            raw = p.read_bytes()
+            if hashlib.sha256(raw).hexdigest() != receipt['png_sha256']:
+                raise ValueError('PNG changed')
+            with Image.open(p) as im:
+                im.load()
+                frame = im.convert('RGB')
+            if list(frame.size) != receipt['size']:
+                raise ValueError('Frame dimensions changed')
+            original = list(frame.size)
+            area = frame.width * frame.height
+            if area > max_pixels:
+                scale = math.sqrt(max_pixels / area)
+                frame = frame.resize((int(frame.width * scale), int(frame.height * scale)))
+            frames.append(frame)
+            shapes.append(dict(original=original, regularized=list(frame.size)))
+        videos.append(frames)
+        sizes.append(shapes)
+    return (videos, sizes)
+
+def processor_inputs(sample, root, processor, preserve_canvas=True):
+    from transformers.video_utils import VideoMetadata
+    videos, sizes = load_frames(sample, root, preserve_canvas)
+    metadata = [VideoMetadata(total_num_frames=len(frames), fps=float(Fraction(b['clock_fps_num'], b['clock_fps_den'])), frames_indices=list(b['block_local_timestamp_indices'])) for frames, b in zip(videos, sample['blocks'], strict=True)]
+    text = processor.apply_chat_template([dict(role='user', content=prompt(sample, processor.video_token))], tokenize=False, add_generation_prompt=True, enable_thinking=False)
+    inputs = processor(text=[text], videos=videos, video_metadata=metadata, do_sample_frames=False, return_metadata=True, return_tensors='pt')
+    return (inputs, sizes, text)

batch_utils/vendor/production_adapter/TEMPSEG_PROMPTS.json

@@ -0,0 +1,27 @@
+[
+  {
+    "pool": "dex",
+    "prompt": "<video>\nThe video contains 64 ordered temporal slots sampled at 4 Hz for one 16-second local view.\nThe emit core is 12 seconds with a 2-second guard on each side.\nSegment meaningful local operations. Acquisition and delivery are distinct when observable; retain distinct directional handovers. Ordinary contacts, adjustments and low-motion pauses within the same ongoing goal do not create new events. Keep clear failure/retry or execution-quality transitions as separate intervals even when their intended operation is unchanged. Represent genuine pure rest as an interval, not a gap.\nEdge padding repeats the nearest real sampled frame; ignore slots outside the valid tick envelope.\nTimes use 20 Hz local ticks in the half-open range [0,320).\nCanonical window envelope:\n{\"core_end_tick\":280,\"core_start_tick\":40,\"valid_end_tick\":320,\"valid_start_tick\":0}\nEmit only intervals with positive half-open intersection with [core_start_tick,core_end_tick). Clip intervals at valid video limits; lo/ro indicate continuation beyond the visible limit. lo=true requires s==valid_start_tick; ro=true requires e==valid_end_tick.\nReturn exactly one JSON object and no prose or Markdown, with only top-level key segments. Each row has keys exactly in order s,e,lo,ro. Sort rows by (s,e,lo,ro).",
+    "prompt_sha256": "a50644d790524e1d4c80a5ad154e940d9bb512b0816eb9d2703cf5721a2f1943"
+  },
+  {
+    "pool": "human",
+    "prompt": "<video>\nThe video contains 64 ordered temporal slots sampled at 4 Hz for one 16-second local view.\nThe emit core is 12 seconds with a 2-second guard on each side.\nEdge padding repeats the nearest real frame; ignore slots outside the valid tick envelope.\nTimes use 20 Hz local ticks in the half-open range [0,320).\nCanonical window envelope:\n{\"valid_start_tick\":0,\"valid_end_tick\":320,\"core_start_tick\":40,\"core_end_tick\":280}\nEmit only subtask intervals with positive half-open intersection with [core_start_tick,core_end_tick).\nA quantized boundary alias may instead end exactly at core_start_tick or start exactly at core_end_tick.\nlo=true requires s==valid_start_tick; ro=true requires e==valid_end_tick.\nReturn exactly one JSON object and no prose or Markdown.\nUse only the top-level key segments; each row's keys must appear exactly as s,e,lo,ro.\nPreserve concurrent or duplicate intervals as separate rows and sort rows by (s,e,lo,ro).\nAn empty list is allowed only when no interval has emit-core support.\n",
+    "prompt_sha256": "21521d5e3b854488f022f402cbb3c31137e2a1a8a114837b16e353c631bc350a"
+  },
+  {
+    "pool": "tianji",
+    "prompt": "<video>\nThe video contains 64 ordered temporal slots sampled at 4 Hz for one 16-second local view.\nThe emit core is 12 seconds with a 2-second guard on each side.\nEdge padding repeats the nearest real frame; ignore slots outside the valid tick envelope.\nTimes use 20 Hz local ticks in the half-open range [0,320).\nCanonical window envelope:\n{\"core_end_tick\":280,\"core_start_tick\":40,\"valid_end_tick\":320,\"valid_start_tick\":40}\nEmit only atomic intervals with positive half-open intersection with [core_start_tick,core_end_tick).\nA quantized boundary alias may instead end exactly at core_start_tick or start exactly at core_end_tick.\nlo=true requires s==valid_start_tick; ro=true requires e==valid_end_tick.\nReturn exactly one JSON object and no prose or Markdown.\nUse only the top-level key segments; each row's keys must appear exactly as s,e,lo,ro.\nPreserve concurrent or duplicate atomic intervals as separate rows and sort rows by (s,e,lo,ro).\nAn empty list is allowed only when no atomic interval has emit-core support.\n",
+    "prompt_sha256": "497e9adbd8ae4ffc76f54ee4f42bd20d597a44667709cee7fccf447c3f504243"
+  },
+  {
+    "pool": "yam32",
+    "prompt": "<video>\nThe video contains 128 ordered temporal slots sampled at 4 Hz for one 32-second local view.\nThe emit core is 24 seconds with a 4-second guard on each side.\nSegment meaningful local operations. Acquisition and delivery are distinct when observable; retain distinct directional handovers. Ordinary contacts, adjustments and low-motion pauses within the same ongoing goal do not create new events. Keep clear failure/retry or execution-quality transitions as separate intervals even when their intended operation is unchanged. Represent genuine pure rest as an interval, not a gap.\nEdge padding repeats the nearest real sampled frame; ignore slots outside the valid tick envelope.\nTimes use 20 Hz local ticks in the half-open range [0,640).\nCanonical window envelope:\n{\"core_end_tick\":560,\"core_start_tick\":80,\"valid_end_tick\":640,\"valid_start_tick\":0}\nEmit only intervals with positive half-open intersection with [core_start_tick,core_end_tick). Clip intervals at valid video limits; lo/ro indicate continuation beyond the visible limit. lo=true requires s==valid_start_tick; ro=true requires e==valid_end_tick.\nReturn exactly one JSON object and no prose or Markdown, with only top-level key segments. Each row has keys exactly in order s,e,lo,ro. Sort rows by (s,e,lo,ro).",
+    "prompt_sha256": "7c6913c78b3fddca0507d6f25534bb89797160ac6342b28f0117b17063ba116f"
+  },
+  {
+    "pool": "yam",
+    "prompt": "<video>\nThe video contains 64 ordered temporal slots sampled at 4 Hz for one 16-second local view.\nThe emit core is 12 seconds with a 2-second guard on each side.\nSegment meaningful local operations. Acquisition and delivery are distinct when observable; retain distinct directional handovers. Ordinary contacts, adjustments and low-motion pauses within the same ongoing goal do not create new events. Keep clear failure/retry or execution-quality transitions as separate intervals even when their intended operation is unchanged. Represent genuine pure rest as an interval, not a gap.\nEdge padding repeats the nearest real sampled frame; ignore slots outside the valid tick envelope.\nTimes use 20 Hz local ticks in the half-open range [0,320).\nCanonical window envelope:\n{\"core_end_tick\":280,\"core_start_tick\":40,\"valid_end_tick\":320,\"valid_start_tick\":0}\nEmit only intervals with positive half-open intersection with [core_start_tick,core_end_tick). Clip intervals at valid video limits; lo/ro indicate continuation beyond the visible limit. lo=true requires s==valid_start_tick; ro=true requires e==valid_end_tick.\nReturn exactly one JSON object and no prose or Markdown, with only top-level key segments. Each row has keys exactly in order s,e,lo,ro. Sort rows by (s,e,lo,ro).",
+    "prompt_sha256": "a50644d790524e1d4c80a5ad154e940d9bb512b0816eb9d2703cf5721a2f1943"
+  }
+]

batch_utils/vendor/production_adapter/adapter.py

@@ -0,0 +1,102 @@
+from dataclasses import asdict
+from fractions import Fraction
+import hashlib
+import json
+from pathlib import Path
+import re
+import sys
+sys.path.insert(0, str(Path(__file__).resolve().parent / 'core'))
+from flora_runtime.schema import AnnotationSource, TemporalSegment, TimeSpan
+from flora_runtime.video_segmentation.fine_subtask import FineSubtaskSegmenter, HierarchicalSubtaskSegmenter, FineSegmentationQualityPolicy
+from flora_runtime.video_segmentation.fine_temporal import FineWindow, fine_window_profile, TICK_US, HUMAN_FINE_WINDOW_PROFILE_ID, YAM_ATOMIC_LONG_FINE_WINDOW_PROFILE_ID, YAM_INTENT_32S_FINE_WINDOW_PROFILE_ID
+from flora_runtime.video_segmentation.fine_worker import _nearest_index
+from flora_runtime.subtask_captioning.u2_worker import _plan_blocks, U2WorkerError
+HERE = Path(__file__).resolve().parent
+TEMPLATES = {x['pool']: x for x in json.loads((HERE / 'TEMPSEG_PROMPTS.json').read_text())}
+PROFILES = {'human': HUMAN_FINE_WINDOW_PROFILE_ID, 'tianji': YAM_ATOMIC_LONG_FINE_WINDOW_PROFILE_ID, 'dex': YAM_ATOMIC_LONG_FINE_WINDOW_PROFILE_ID, 'yam': YAM_ATOMIC_LONG_FINE_WINDOW_PROFILE_ID, 'yam32': YAM_INTENT_32S_FINE_WINDOW_PROFILE_ID}
+
+def temporal_prompt(window, pool):
+    assert window.window_profile_id == PROFILES[pool]
+    row = TEMPLATES[pool]
+    template = row['prompt']
+    assert hashlib.sha256(template.encode()).hexdigest() == row['prompt_sha256']
+    pattern = '(Canonical window envelope:\\s*)(\\{[^{}]+\\})'
+    matches = list(re.finditer(pattern, template))
+    assert len(matches) == 1
+    match = matches[0]
+    keys = list(json.loads(match[2]))
+    assert set(keys) == {'valid_start_tick', 'valid_end_tick', 'core_start_tick', 'core_end_tick'}
+    envelope = json.dumps({k: getattr(window, k) for k in keys}, separators=(',', ':'))
+    return template[:match.start(2)] + envelope + template[match.end(2):]
+
+def window_from_payload(payload, source):
+    start = payload['requested_view_start_us']
+    fields = {k: payload[k] for k in ('window_index', 'requested_view_start_us', 'requested_view_end_us', 'valid_start_tick', 'valid_end_tick', 'core_start_tick', 'core_end_tick', 'scope_id', 'window_profile_id', 'coarse_spans')}
+    duration = round(source.duration_s * 1000000)
+    fields.update(source_id=source.source_id, video_end_us=duration, valid_start_us=max(0, start), valid_end_us=min(duration, payload['requested_view_end_us']), core_start_us=max(0, start + payload['core_start_tick'] * TICK_US), core_end_us=min(duration, start + payload['core_end_tick'] * TICK_US))
+    return FineWindow(**fields)
+
+class NoCoarse:
+
+    def segment(self, *_args, **_kwargs):
+        raise AssertionError('Unexpected coarse segmentation request')
+
+def production_segmenter(predictor, binding, pool):
+    assert binding['model_io_contract'] == fine_window_profile(PROFILES[pool]).model_io_contract
+    fine = FineSubtaskSegmenter(predictor=predictor, model_binding=binding, profile_id='human_ego_atomic_v0_2', window_profile_id=PROFILES[pool], fail_on_unresolved=True, batch_concurrency=1, unresolved_stitch_tolerance_ticks=2, quality_policy=FineSegmentationQualityPolicy.from_value({'drop_guard_only_segments': True, 'hard_max_overlap_depth': 3, 'hard_max_segments_per_window': 23, 'max_adjacent_matching_ambiguities': 0, 'require_event_when_coarse_present': True, 'schema_version': 'subtask_annotation_atomic_fine_quality_policy_v1', 'soft_max_segments_per_window': 12}))
+    return HierarchicalSubtaskSegmenter(coarse_segmenter=NoCoarse(), fine_segmenter=fine, fine_prior_mode='none', fine_failure_policy='adaptive_fine_then_conservative_completion', fine_output_coverage_policy='allow_gaps', adaptive_fine_retry_max_windows=32)
+
+def temporal_frame_plan(window, pts_us):
+    assert pts_us and pts_us[0] == 0 and all((a < b for a, b in zip(pts_us, pts_us[1:])))
+    assert pts_us[-1] < window.video_end_us
+    profile = window.window_profile
+    requested = [window.requested_view_start_us + i * profile.slot_spacing_us for i in range(profile.slot_count)]
+    pad = [not 0 <= t < window.video_end_us for t in requested]
+    indices = [None if padding else _nearest_index(t, pts_us, max_selection_error_us=profile.slot_spacing_us // 2) for t, padding in zip(requested, pad, strict=True)]
+    real = [i for i, v in enumerate(indices) if v is not None]
+    assert real
+    first, last = (real[0], real[-1])
+    indices[:first] = [indices[first]] * first
+    indices[last + 1:] = [indices[last]] * (len(indices) - last - 1)
+    assert all((v is not None for v in indices))
+    observed = [v for v, p in zip(indices, pad, strict=True) if not p]
+    assert len(set(observed)) == len(observed), 'Real temporal slots must not duplicate evidence'
+    return {'source_frame_indices': indices, 'padding_mask': pad, 'requested_us': requested, 'actual_pts_us': [float(pts_us[i]) for i in indices], 'model_clock_hz': profile.slot_fps, 'tick_us': TICK_US, 'max_real_error_us': float(max((abs(pts_us[i] - t) for i, t, p in zip(indices, requested, pad, strict=True) if not p)))}
+
+def caption_plans(segments, media):
+    sources = [TemporalSegment(TimeSpan(s['start_s'], s['end_s']), dict(s.get('metadata', {}))) for s in segments]
+    sources = HierarchicalSubtaskSegmenter._with_refreshed_u2_context(tuple(sources))
+    plans, dispositions = ([], [])
+    for i, s in enumerate(sources):
+        eid = s.metadata.get('fine_event_id') or f'event_{i:06d}'
+        record = {'event_id': eid, 'start_s': s.span.start_s, 'end_s': s.span.end_s, 'event_index': i}
+        context = s.metadata['u2_context']
+        try:
+            blocks = _plan_blocks(context, media)
+        except U2WorkerError as error:
+            if 'TARGET span has fewer than' not in str(error):
+                raise
+            dispositions.append(dict(record, status='unsupported_target_frames', reason=str(error)))
+            continue
+        serialized = [asdict(b) for b in blocks]
+        for b in serialized:
+            b['frame_indices'] = list(b['frame_indices'])
+        plans.append(dict(record, blocks=serialized, source_fps_num=media['fps_num'], source_fps_den=media['fps_den'], metadata=dict(s.metadata)))
+        dispositions.append(dict(record, status='caption_input'))
+    assert len({x['event_id'] for x in dispositions}) == len(dispositions)
+    return (plans, dispositions)
+
+def caption_sample(plan, frame_receipts, camera_roles):
+    blocks = []
+    clock = Fraction(plan['source_fps_num'], plan['source_fps_den'])
+    for b in plan['blocks']:
+        ids = b['frame_indices']
+        n = len(ids)
+        span = b['end_frame_exclusive'] - b['start_frame']
+        rate = Fraction(n) * clock / span
+        receipts = [frame_receipts[i] for i in ids]
+        blocks.append(dict(role=b['role'], paths=[r['path'] for r in receipts], frames=receipts, source_start_frame=b['start_frame'], source_end_frame_exclusive=b['end_frame_exclusive'], clock_fps_num=clock.numerator, clock_fps_den=clock.denominator, block_local_timestamp_indices=[i - b['start_frame'] for i in ids], effective_fps_num=rate.numerator, effective_fps_den=rate.denominator, original_descriptor={'frame_indices': ids}))
+    target = next((b for b in blocks if b['role'] == 'TARGET'))
+    text = 'REFERENCE_NOT_ADMITTED'
+    request = {'target': {'native_span': {'start_frame': target['source_start_frame'], 'end_frame_exclusive': target['source_end_frame_exclusive']}, 'text': text, 'sha256': hashlib.sha256(text.encode()).hexdigest()}}
+    return {'request': request, 'camera_count': len(camera_roles), 'camera_roles': camera_roles, 'blocks': blocks}

batch_utils/vendor/production_adapter/core/flora_runtime/__init__.py

@@ -0,0 +1 @@
+pass

batch_utils/vendor/production_adapter/core/flora_runtime/exact_frame_serving.py

@@ -0,0 +1,47 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+import hashlib
+import io
+import math
+from typing import Any
+PIXEL_HASH_SCHEMA = 'rgb_u8_width_height_v1'
+
+class ExactFrameServingError(ValueError):
+    pass
+
+def pixel_sha256(image: Any) -> str:
+    rgb = image.convert('RGB')
+    header = f'{PIXEL_HASH_SCHEMA}\x00RGB\x00{rgb.width}\x00{rgb.height}\x00'.encode('ascii')
+    return hashlib.sha256(header + rgb.tobytes()).hexdigest()
+
+def regularized_resize_dims(*, height: int, width: int, min_pixels: int=16 * 16, max_pixels: int=512 * 512) -> tuple[int, int]:
+    if min(height, width, min_pixels, max_pixels) <= 0 or min_pixels > max_pixels:
+        raise ExactFrameServingError('regularized resize arguments are invalid')
+    area = height * width
+    if area > max_pixels:
+        scale = math.sqrt(max_pixels / area)
+        return (int(height * scale), int(width * scale))
+    if area < min_pixels:
+        scale = math.sqrt(min_pixels / area)
+        return (int(height * scale), int(width * scale))
+    return (height, width)
+
+def regularize_image(image: Any) -> Any:
+    from PIL import Image
+    source = image.convert('RGB')
+    height, width = regularized_resize_dims(height=source.height, width=source.width)
+    if source.size == (width, height):
+        return source
+    return source.resize((width, height), resample=Image.Resampling.BICUBIC)
+
+def encode_regularized_image(image: Any, *, encoding: str) -> dict[str, Any]:
+    regularized = regularize_image(image)
+    stream = io.BytesIO()
+    if encoding == 'png_level6':
+        regularized.save(stream, format='PNG', optimize=False, compress_level=6)
+    elif encoding == 'jpeg_q98_444':
+        regularized.save(stream, format='JPEG', quality=98, subsampling=0, optimize=False)
+    else:
+        raise ExactFrameServingError(f'unsupported exact-frame cache encoding: {encoding}')
+    return {'payload': stream.getvalue(), 'width': regularized.width, 'height': regularized.height, 'rgb_pixel_sha256': pixel_sha256(regularized)}

batch_utils/vendor/production_adapter/core/flora_runtime/schema.py

@@ -0,0 +1,88 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+import math
+import os
+from typing import Any
+
+def _require_nonempty(value: str, name: str) -> None:
+    if not value.strip():
+        raise ValueError(f'{name} must not be empty')
+
+def _require_finite(value: float, name: str) -> None:
+    if not math.isfinite(value):
+        raise ValueError(f'{name} must be finite, got {value}')
+
+@dataclass(frozen=True)
+class VideoStream:
+    camera: str
+    location: str | os.PathLike[str]
+    fps: float | None = None
+    width: int | None = None
+    height: int | None = None
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        _require_nonempty(self.camera, 'camera')
+        location = os.fspath(self.location)
+        _require_nonempty(location, 'location')
+        object.__setattr__(self, 'location', location)
+        object.__setattr__(self, 'metadata', dict(self.metadata))
+        if self.fps is not None:
+            _require_finite(self.fps, 'fps')
+            if self.fps <= 0:
+                raise ValueError(f'fps must be positive, got {self.fps}')
+        for value, name in ((self.width, 'width'), (self.height, 'height')):
+            if value is not None and value <= 0:
+                raise ValueError(f'{name} must be positive, got {value}')
+
+@dataclass(frozen=True)
+class AnnotationSource:
+    source_id: str
+    location: str | os.PathLike[str]
+    video_streams: tuple[VideoStream, ...] = ()
+    duration_s: float | None = None
+    fps: float | None = None
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        _require_nonempty(self.source_id, 'source_id')
+        location = os.fspath(self.location)
+        _require_nonempty(location, 'location')
+        object.__setattr__(self, 'location', location)
+        object.__setattr__(self, 'video_streams', tuple(self.video_streams))
+        object.__setattr__(self, 'metadata', dict(self.metadata))
+        cameras = [stream.camera for stream in self.video_streams]
+        if len(cameras) != len(set(cameras)):
+            raise ValueError(f'video stream camera names must be unique, got {cameras}')
+        if self.duration_s is not None:
+            _require_finite(self.duration_s, 'duration_s')
+            if self.duration_s <= 0:
+                raise ValueError(f'duration_s must be positive, got {self.duration_s}')
+        if self.fps is not None:
+            _require_finite(self.fps, 'fps')
+            if self.fps <= 0:
+                raise ValueError(f'fps must be positive, got {self.fps}')
+
+@dataclass(frozen=True, order=True)
+class TimeSpan:
+    start_s: float
+    end_s: float
+
+    def __post_init__(self) -> None:
+        _require_finite(self.start_s, 'start_s')
+        _require_finite(self.end_s, 'end_s')
+        if self.start_s < 0:
+            raise ValueError(f'start_s must be non-negative, got {self.start_s}')
+        if self.end_s <= self.start_s:
+            raise ValueError(f'end_s must be greater than start_s, got [{self.start_s}, {self.end_s})')
+
+@dataclass(frozen=True)
+class TemporalSegment:
+    span: TimeSpan
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, 'metadata', dict(self.metadata))

batch_utils/vendor/production_adapter/core/flora_runtime/subtask_captioning/__init__.py

@@ -0,0 +1 @@
+pass

batch_utils/vendor/production_adapter/core/flora_runtime/subtask_captioning/u2_contract.py

@@ -0,0 +1,58 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+from fractions import Fraction
+TARGET_SAMPLE_FPS = 8
+TARGET_MIN_FRAMES = 8
+TARGET_MAX_FRAMES = 40
+TARGET_FRAME_FACTOR = 2
+CONTEXT_SAMPLE_FPS = 8
+CONTEXT_MIN_FRAMES = 4
+CONTEXT_MAX_FRAMES = 12
+CONTEXT_FRAME_FACTOR = 4
+MAX_TOTAL_FRAMES = 64
+SAMPLER_VERSION = 'duration_adaptive_uniform_bin_centers_nested_context_v2'
+
+class U2ContractError(ValueError):
+    pass
+
+def uniform_bin_centers(start: int, end_exclusive: int, count: int) -> list[int]:
+    length = end_exclusive - start
+    if count <= 0:
+        return []
+    if length < count:
+        raise U2ContractError(f'span [{start}, {end_exclusive}) has fewer than {count} frames')
+    result = [start + (2 * index + 1) * length // (2 * count) for index in range(count)]
+    if len(result) != len(set(result)) or result[0] < start or result[-1] >= end_exclusive:
+        raise AssertionError('uniform frame sampler violated its contract')
+    return result
+
+def duration_adaptive_sample(start: int, end_exclusive: int, *, source_fps_num: int, source_fps_den: int, sample_fps: int, min_frames: int, max_frames: int, frame_factor: int) -> tuple[list[int], dict[str, int | bool | str]]:
+    available = end_exclusive - start
+    if source_fps_num <= 0 or source_fps_den <= 0 or available <= 0:
+        raise U2ContractError('source FPS and frame span must be positive')
+    if min_frames % frame_factor or max_frames % frame_factor:
+        raise U2ContractError('adaptive frame bounds must align to frame factor')
+    duration = Fraction(available * source_fps_den, source_fps_num)
+    requested = min(Fraction(max_frames), max(Fraction(min_frames), duration * sample_fps))
+    desired_count = requested.numerator // (requested.denominator * frame_factor) * frame_factor
+    actual_count = min(desired_count, available // frame_factor * frame_factor)
+    if actual_count < frame_factor:
+        raise U2ContractError(f'span [{start}, {end_exclusive}) has fewer than {frame_factor} frames')
+    indices = uniform_bin_centers(start, end_exclusive, actual_count)
+    return (indices, {'policy': 'duration_fps_bounded_unique_frames_v1', 'sampler_version': SAMPLER_VERSION, 'source_fps_num': source_fps_num, 'source_fps_den': source_fps_den, 'sample_fps_num': sample_fps, 'sample_fps_den': 1, 'min_frames': min_frames, 'max_frames': max_frames, 'frame_factor': frame_factor, 'available_unique_source_frames': available, 'desired_frame_count': desired_count, 'actual_frame_count': actual_count, 'source_limited': actual_count < desired_count, 'selection': 'uniform_bin_centers'})
+
+def _ceil_fraction(value: Fraction) -> int:
+    return -(-value.numerator // value.denominator)
+
+def seconds_to_frame_span(start_sec: float | str, end_sec: float | str, *, source_fps_num: int, source_fps_den: int, source_frame_count: int) -> tuple[int, int]:
+    start = Fraction(str(start_sec))
+    end = Fraction(str(end_sec))
+    if start < 0 or end <= start:
+        raise U2ContractError(f'invalid seconds span [{start_sec}, {end_sec})')
+    fps = Fraction(source_fps_num, source_fps_den)
+    start_frame = max(0, min(source_frame_count, _ceil_fraction(start * fps)))
+    end_frame = max(0, min(source_frame_count, _ceil_fraction(end * fps)))
+    if end_frame <= start_frame:
+        raise U2ContractError(f'seconds span [{start_sec}, {end_sec}) maps to no source frames')
+    return (start_frame, end_frame)

batch_utils/vendor/production_adapter/core/flora_runtime/subtask_captioning/u2_worker.py

@@ -0,0 +1,44 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+from dataclasses import dataclass
+from typing import Any
+from flora_runtime.subtask_captioning.u2_contract import CONTEXT_FRAME_FACTOR, CONTEXT_MAX_FRAMES, CONTEXT_MIN_FRAMES, CONTEXT_SAMPLE_FPS, MAX_TOTAL_FRAMES, TARGET_FRAME_FACTOR, TARGET_MAX_FRAMES, TARGET_MIN_FRAMES, TARGET_SAMPLE_FPS, duration_adaptive_sample, seconds_to_frame_span
+_ROLES = ('BEFORE', 'TARGET', 'AFTER')
+
+class U2WorkerError(RuntimeError):
+    pass
+
+@dataclass(frozen=True)
+class _BlockPlan:
+    role: str
+    start_frame: int
+    end_frame_exclusive: int
+    frame_indices: tuple[int, ...]
+    sampling_receipt: dict[str, Any]
+
+def _plan_blocks(u2_context: dict[str, Any], media: dict[str, Any]) -> tuple[_BlockPlan, ...]:
+    plans: list[_BlockPlan] = []
+    for role, key in zip(_ROLES, ('before', 'target', 'after'), strict=True):
+        span = u2_context.get(key)
+        if span is None:
+            if role == 'TARGET':
+                raise U2WorkerError('U2 TARGET context is absent')
+            continue
+        if not isinstance(span, dict):
+            raise U2WorkerError(f'U2 {key} span must be an object')
+        start, end = seconds_to_frame_span(span.get('start_s'), span.get('end_s'), source_fps_num=media['fps_num'], source_fps_den=media['fps_den'], source_frame_count=media['frame_count'])
+        target = role == 'TARGET'
+        frame_factor = TARGET_FRAME_FACTOR if target else CONTEXT_FRAME_FACTOR
+        if end - start < frame_factor:
+            if not target:
+                continue
+            raise U2WorkerError(f'U2 {role} span has fewer than the required {frame_factor}-frame group')
+        indices, receipt = duration_adaptive_sample(start, end, source_fps_num=media['fps_num'], source_fps_den=media['fps_den'], sample_fps=TARGET_SAMPLE_FPS if target else CONTEXT_SAMPLE_FPS, min_frames=TARGET_MIN_FRAMES if target else CONTEXT_MIN_FRAMES, max_frames=TARGET_MAX_FRAMES if target else CONTEXT_MAX_FRAMES, frame_factor=frame_factor)
+        plans.append(_BlockPlan(role=role, start_frame=start, end_frame_exclusive=end, frame_indices=tuple(indices), sampling_receipt=dict(receipt)))
+    total = sum((len(plan.frame_indices) for plan in plans))
+    if total > MAX_TOTAL_FRAMES:
+        raise U2WorkerError(f'U2 frame budget {total} exceeds {MAX_TOTAL_FRAMES}')
+    return tuple(plans)
+if __name__ == '__main__':
+    raise SystemExit(main())

batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/__init__.py

@@ -0,0 +1 @@
+pass

batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/contracts.py

@@ -0,0 +1,73 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+import math
+from typing import Any, Literal
+from flora_runtime.schema import TimeSpan
+EventCompleteness = Literal['complete', 'left_censored', 'right_censored', 'both_censored', 'uncertain']
+
+def _require_nonempty(value: str, label: str) -> None:
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f'{label} must be non-empty')
+
+def _validate_confidence(value: float | None, label: str) -> None:
+    if value is None:
+        return
+    if isinstance(value, bool) or not isinstance(value, int | float):
+        raise TypeError(f'{label} must be a number in [0, 1]')
+    if not math.isfinite(float(value)) or not 0.0 <= float(value) <= 1.0:
+        raise ValueError(f'{label} must be finite and in [0, 1]')
+
+def span_contains(outer: TimeSpan, inner: TimeSpan, *, tolerance_s: float=1e-09) -> bool:
+    if tolerance_s < 0 or not math.isfinite(tolerance_s):
+        raise ValueError('tolerance_s must be finite and non-negative')
+    return outer.start_s <= inner.start_s + tolerance_s and outer.end_s + tolerance_s >= inner.end_s
+
+@dataclass(frozen=True)
+class ProcessingScope:
+    scope_id: str
+    source_id: str
+    core_span: TimeSpan
+    context_span: TimeSpan
+    proposal_confidence: float | None = None
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        _require_nonempty(self.scope_id, 'scope_id')
+        _require_nonempty(self.source_id, 'source_id')
+        if not span_contains(self.context_span, self.core_span):
+            raise ValueError('context_span must contain core_span')
+        _validate_confidence(self.proposal_confidence, 'proposal_confidence')
+        object.__setattr__(self, 'metadata', dict(self.metadata))
+
+@dataclass(frozen=True)
+class EventInterval:
+    event_id: str
+    source_id: str
+    span: TimeSpan
+    completeness: EventCompleteness = 'complete'
+    start_confidence: float | None = None
+    end_confidence: float | None = None
+    claim: str | None = None
+    source_scope_ids: tuple[str, ...] = ()
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        _require_nonempty(self.event_id, 'event_id')
+        _require_nonempty(self.source_id, 'source_id')
+        if self.completeness not in {'complete', 'left_censored', 'right_censored', 'both_censored', 'uncertain'}:
+            raise ValueError(f'unsupported completeness: {self.completeness!r}')
+        _validate_confidence(self.start_confidence, 'start_confidence')
+        _validate_confidence(self.end_confidence, 'end_confidence')
+        if self.claim is not None:
+            _require_nonempty(self.claim, 'claim')
+            object.__setattr__(self, 'claim', self.claim.strip())
+        scope_ids = tuple(self.source_scope_ids)
+        if len(scope_ids) != len(set(scope_ids)):
+            raise ValueError('source_scope_ids must be unique')
+        for scope_id in scope_ids:
+            _require_nonempty(scope_id, 'source_scope_id')
+        object.__setattr__(self, 'source_scope_ids', scope_ids)
+        object.__setattr__(self, 'metadata', dict(self.metadata))

batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/fine_subtask.py

@@ -0,0 +1,852 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+import asyncio
+from collections import Counter
+from collections.abc import Mapping, Sequence
+from dataclasses import asdict, dataclass, field, replace
+import hashlib
+import json
+import math
+from typing import Any
+from flora_runtime.schema import AnnotationSource, TemporalSegment, TimeSpan
+from flora_runtime.video_segmentation.contracts import EventInterval, ProcessingScope
+from flora_runtime.video_segmentation.fine_temporal import DEFAULT_FINE_WINDOW_PROFILE_ID, MAX_BOUNDED_ANCHOR_TICKS, STITCH_TOLERANCE_TICKS, TICK_US, FineFragment, FineSegmentationResult, FineTemporalError, FineWindow, StitchingPolicy, UnresolvedFineEvent, analyze_stitching_seams, ceil_div, fine_window_profile, parse_prediction_with_diagnostics, stitch_predictions, window_from_scope
+from flora_runtime.video_segmentation.interface import SegmentTopology, VideoSegmenter
+from flora_runtime.video_segmentation.scope_scheduler import FINE_DIRECTIONAL_RETRY_SCOPE_NAMESPACE, schedule_fine_scopes, schedule_recentered_fine_retry_scopes, seconds_to_us
+from flora_runtime.worker_transport import JsonPredictor, SubtaskAnnotationAdapterError, source_payload
+FINE_REQUEST_SCHEMA = 'subtask_annotation_atomic_fine_batch_request_v1'
+FINE_RESPONSE_SCHEMA = 'subtask_annotation_atomic_fine_batch_response_v1'
+FINE_QUALITY_POLICY_SCHEMA = 'subtask_annotation_atomic_fine_quality_policy_v1'
+PROVIDER_FREE_FINE_COMPLETION_POLICIES = {'adaptive_fine_then_conservative_completion'}
+ADAPTIVE_FINE_REPAIR_SCHEMA = 'subtask_annotation_adaptive_fine_repair_v1'
+ADAPTIVE_FINE_RETRY_MAX_WINDOWS = 4
+ADAPTIVE_FINE_RETRY_HARD_MAX_WINDOWS = 64
+CONSERVATIVE_FINE_COMPLETION_SCHEMA = 'subtask_annotation_conservative_fine_completion_v2'
+UNRESOLVED_LOCAL_STITCH_SCHEMA = 'subtask_annotation_unresolved_local_stitch_v1'
+SEGMENT_LOCAL_REPAIR_ENDPOINT_TOLERANCE_S = 0.5
+
+class FineSegmentationOutputError(SubtaskAnnotationAdapterError):
+
+    def __init__(self, message: str, *, diagnostics: Mapping[str, object], partial_result: FineSegmentationResult | None=None) -> None:
+        super().__init__(message)
+        self.diagnostics = dict(diagnostics)
+        self.partial_result = partial_result
+
+@dataclass(frozen=True)
+class FineSegmentationQualityPolicy:
+    soft_max_segments_per_window: int = 32
+    hard_max_segments_per_window: int = 32
+    hard_max_overlap_depth: int = 32
+    max_adjacent_matching_ambiguities: int = 0
+    drop_guard_only_segments: bool = False
+    require_event_when_coarse_present: bool = False
+
+    def __post_init__(self) -> None:
+        integer_fields = ('soft_max_segments_per_window', 'hard_max_segments_per_window', 'hard_max_overlap_depth', 'max_adjacent_matching_ambiguities')
+        for name in integer_fields:
+            value = getattr(self, name)
+            if isinstance(value, bool) or not isinstance(value, int) or value < 0:
+                raise ValueError(f'{name} must be a non-negative integer')
+        if self.soft_max_segments_per_window > self.hard_max_segments_per_window:
+            raise ValueError('soft_max_segments_per_window must not exceed the hard maximum')
+        if type(self.drop_guard_only_segments) is not bool:
+            raise TypeError('drop_guard_only_segments must be a boolean')
+        if type(self.require_event_when_coarse_present) is not bool:
+            raise TypeError('require_event_when_coarse_present must be a boolean')
+
+    @classmethod
+    def from_value(cls, value: object) -> FineSegmentationQualityPolicy:
+        if not isinstance(value, dict):
+            raise TypeError('fine quality policy must be an object')
+        expected = {'schema_version', 'soft_max_segments_per_window', 'hard_max_segments_per_window', 'hard_max_overlap_depth', 'max_adjacent_matching_ambiguities', 'drop_guard_only_segments', 'require_event_when_coarse_present'}
+        missing, unknown = (expected - set(value), set(value) - expected)
+        if missing or unknown:
+            raise ValueError(f'fine quality policy key mismatch; missing={sorted(missing)}, unknown={sorted(unknown)}')
+        if value['schema_version'] != FINE_QUALITY_POLICY_SCHEMA:
+            raise ValueError(f'fine quality policy schema must be {FINE_QUALITY_POLICY_SCHEMA!r}')
+        return cls(**{key: value[key] for key in expected - {'schema_version'}})
+
+    def to_dict(self) -> dict[str, object]:
+        return {'schema_version': FINE_QUALITY_POLICY_SCHEMA, 'soft_max_segments_per_window': self.soft_max_segments_per_window, 'hard_max_segments_per_window': self.hard_max_segments_per_window, 'hard_max_overlap_depth': self.hard_max_overlap_depth, 'max_adjacent_matching_ambiguities': self.max_adjacent_matching_ambiguities, 'drop_guard_only_segments': self.drop_guard_only_segments, 'require_event_when_coarse_present': self.require_event_when_coarse_present}
+
+def _span_payload(span: TimeSpan | None) -> dict[str, float] | None:
+    if span is None:
+        return None
+    return {'start_s': span.start_s, 'end_s': span.end_s}
+
+def _nearest_nonoverlapping_context(ordered: Sequence[EventInterval | TemporalSegment], index: int) -> tuple[TimeSpan | None, TimeSpan | None]:
+    target = ordered[index].span
+    before_candidates = [item.span for item in ordered if item.span.end_s <= target.start_s]
+    after_candidates = [item.span for item in ordered if item.span.start_s >= target.end_s]
+    before = max(before_candidates, key=lambda span: (span.end_s, span.start_s), default=None)
+    after = min(after_candidates, key=lambda span: (span.start_s, span.end_s), default=None)
+    return (before, after)
+
+def events_to_temporal_segments(events: Sequence[EventInterval], *, diagnostics: Mapping[str, object] | None=None, resolution: str='fine') -> tuple[TemporalSegment, ...]:
+    ordered = tuple(sorted(events, key=lambda item: (item.span, item.event_id)))
+    result: list[TemporalSegment] = []
+    for index, event in enumerate(ordered):
+        before, after = _nearest_nonoverlapping_context(ordered, index)
+        metadata = dict(event.metadata)
+        metadata.setdefault('segmentation_provenance', 'fine')
+        metadata.setdefault('segmentation_resolution', resolution)
+        event_identity = {'fine_event_id': event.event_id} if metadata['segmentation_resolution'] == 'fine' else {'segment_event_id': event.event_id}
+        metadata.update({**event_identity, 'source_scope_ids': list(event.source_scope_ids), 'segment_topology': SegmentTopology.CONCURRENT.value, 'u2_context': {'before': _span_payload(before), 'target': _span_payload(event.span), 'after': _span_payload(after)}})
+        if diagnostics is not None:
+            metadata['fine_segmentation_diagnostics'] = dict(diagnostics)
+        result.append(TemporalSegment(span=event.span, metadata=metadata))
+    return tuple(result)
+
+def _overlap_statistics(events: Sequence[EventInterval]) -> tuple[int, int]:
+    overlap_pairs = 0
+    for left_index, left in enumerate(events):
+        overlap_pairs += sum((min(left.span.end_s, right.span.end_s) > max(left.span.start_s, right.span.start_s) for right in events[left_index + 1:]))
+    points = [(item.span.start_s, 1) for item in events] + [(item.span.end_s, -1) for item in events]
+    depth = current = 0
+    for _, delta in sorted(points, key=lambda item: (item[0], item[1])):
+        current += delta
+        depth = max(depth, current)
+    return (overlap_pairs, depth)
+
+def _exact_duplicate_intervals(events: Sequence[EventInterval]) -> list[dict[str, object]]:
+    counts = Counter(((item.span.start_s, item.span.end_s) for item in events))
+    return [{'start_s': start_s, 'end_s': end_s, 'count': count} for (start_s, end_s), count in sorted(counts.items()) if count > 1]
+
+def _diagnostic_sha256(value: Mapping[str, object]) -> str:
+    payload = (json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False, separators=(',', ':')) + '\n').encode()
+    return hashlib.sha256(payload).hexdigest()
+
+def _inference_context(source: AnnotationSource, coarse: Sequence[TemporalSegment], *, fine_prior_mode: str) -> dict[str, object]:
+    timestamp_prior = {'schema_version': 'subtask_annotation_coarse_timestamp_prior_v1', 'source_id': source.source_id, 'spans': [{'start_s': segment.span.start_s, 'end_s': segment.span.end_s} for segment in coarse]}
+    return {'fine_prior_mode': fine_prior_mode, 'coarse_prior_count': len(coarse), 'coarse_prior_sha256': _diagnostic_sha256(timestamp_prior)}
+
+def _with_inference_context(diagnostics: Mapping[str, object], context: Mapping[str, object]) -> dict[str, object]:
+    result = dict(diagnostics)
+    result.update(context)
+    result['diagnostics_sha256'] = _diagnostic_sha256({key: value for key, value in result.items() if key != 'diagnostics_sha256'})
+    return result
+
+@dataclass(frozen=True)
+class _FineWindowBatchPrediction:
+    fragments_by_window: Mapping[int, tuple[FineFragment, ...]]
+    raw_segment_counts: tuple[int, ...]
+    emitted_segment_counts: tuple[int, ...]
+    guard_only_fragments: tuple[tuple[FineFragment, ...], ...]
+
+@dataclass(frozen=True)
+class _FineRetryPrediction:
+    candidates_by_window: Mapping[int, tuple[TemporalSegment, ...]]
+    diagnostics: Mapping[str, object]
+
+@dataclass(frozen=True)
+class FineSubtaskSegmenter:
+    predictor: JsonPredictor
+    model_binding: Mapping[str, Any]
+    profile_id: str = 'human_ego_atomic_v0_2'
+    window_profile_id: str = DEFAULT_FINE_WINDOW_PROFILE_ID
+    fail_on_unresolved: bool = True
+    batch_concurrency: int = 1
+    unresolved_stitch_tolerance_ticks: int = STITCH_TOLERANCE_TICKS
+    quality_policy: FineSegmentationQualityPolicy = field(default_factory=FineSegmentationQualityPolicy)
+    segment_topology = SegmentTopology.CONCURRENT
+
+    def __post_init__(self) -> None:
+        binding = dict(self.model_binding)
+        if not binding:
+            raise ValueError('model_binding must not be empty')
+        window_profile = fine_window_profile(self.window_profile_id)
+        declared_model_io_contract = binding.get('model_io_contract')
+        if declared_model_io_contract is not None and declared_model_io_contract != window_profile.model_io_contract:
+            raise ValueError('model_binding.model_io_contract must match the selected fine window profile')
+        if isinstance(self.batch_concurrency, bool) or not isinstance(self.batch_concurrency, int) or self.batch_concurrency <= 0:
+            raise ValueError('batch_concurrency must be a positive integer')
+        if isinstance(self.unresolved_stitch_tolerance_ticks, bool) or not isinstance(self.unresolved_stitch_tolerance_ticks, int) or (not STITCH_TOLERANCE_TICKS <= self.unresolved_stitch_tolerance_ticks <= MAX_BOUNDED_ANCHOR_TICKS):
+            raise ValueError(f'unresolved_stitch_tolerance_ticks must be an integer in [{STITCH_TOLERANCE_TICKS}, {MAX_BOUNDED_ANCHOR_TICKS}]')
+        object.__setattr__(self, 'model_binding', binding)
+
+    def _stitch_with_unresolved_local_candidates(self, windows: Sequence[FineWindow], fragments_by_window: Mapping[int, Sequence[FineFragment]]) -> FineSegmentationResult:
+        strict = stitch_predictions(windows, fragments_by_window)
+        tolerance_ticks = self.unresolved_stitch_tolerance_ticks
+        if not strict.unresolved or tolerance_ticks == STITCH_TOLERANCE_TICKS:
+            return strict
+        policy = StitchingPolicy.component_local_bounded_anchor(endpoint_tolerance_ticks=tolerance_ticks)
+        seam_diagnostics = analyze_stitching_seams(windows, fragments_by_window, policy=policy)
+        candidate_rows: list[dict[str, object]] = []
+        allowlist: set[tuple[int, int, int]] = set()
+        seams = seam_diagnostics.get('seams')
+        if not isinstance(seams, list):
+            raise FineTemporalError('component-local seam diagnostics are missing seam rows')
+        for seam_index, seam in enumerate(seams):
+            if not isinstance(seam, dict):
+                raise FineTemporalError('component-local seam diagnostic row must be an object')
+            local = seam.get('component_local_identity')
+            if not isinstance(local, dict):
+                raise FineTemporalError('component-local seam evidence is missing')
+            accepted_pairs = local.get('accepted_pairs')
+            if not isinstance(accepted_pairs, list):
+                raise FineTemporalError('component-local accepted-pair evidence is invalid')
+            for pair in accepted_pairs:
+                if not isinstance(pair, dict):
+                    raise FineTemporalError('component-local accepted pair must be an object')
+                left_id = pair.get('left_observation_id')
+                right_id = pair.get('right_observation_id')
+                if isinstance(left_id, bool) or not isinstance(left_id, int) or isinstance(right_id, bool) or (not isinstance(right_id, int)):
+                    raise FineTemporalError('component-local accepted pair identity is invalid')
+                allowlist.add((seam_index, left_id, right_id))
+                candidate_rows.append({'seam_index': seam_index, **pair})
+        if not allowlist:
+            return strict
+        relaxed = stitch_predictions(windows, fragments_by_window, policy=policy, component_pair_allowlist=allowlist)
+        if len(relaxed.unresolved) >= len(strict.unresolved):
+            raise FineTemporalError('component-local unresolved stitching did not reduce unresolved components')
+        unresolved_by_observation = {observation_id: component_index for component_index, component in enumerate(strict.unresolved) for observation_id in component.observation_ids}
+        resolved_by_observation = {observation_id: event for event in strict.events for observation_id in event.metadata.get('observation_ids', ()) if isinstance(observation_id, int)}
+        candidates: list[dict[str, object]] = []
+        seen_components: set[int] = set()
+        for row in candidate_rows:
+            partial_root = row.get('partial_component_root')
+            resolved_root = row.get('resolved_component_root')
+            if isinstance(partial_root, bool) or not isinstance(partial_root, int) or isinstance(resolved_root, bool) or (not isinstance(resolved_root, int)):
+                raise FineTemporalError('component-local candidate roots are invalid')
+            component_index = unresolved_by_observation.get(partial_root)
+            resolved_event = resolved_by_observation.get(resolved_root)
+            if component_index is None or resolved_event is None:
+                raise FineTemporalError('component-local candidate does not map to one strict unresolved and one strict resolved component')
+            if component_index in seen_components:
+                raise FineTemporalError('component-local candidate discovery produced multiple targets for one unresolved component')
+            seen_components.add(component_index)
+            candidates.append({'partial_component_index': component_index, 'resolved_event_id': resolved_event.event_id, 'resolved_event_span': _span_payload(resolved_event.span), 'pair': row})
+        local_stitch_diagnostics = {'schema_version': UNRESOLVED_LOCAL_STITCH_SCHEMA, 'mode': 'candidate_only_requires_adaptive_fine_confirmation_v1', 'strict_tolerance_ticks': STITCH_TOLERANCE_TICKS, 'candidate_tolerance_ticks': tolerance_ticks, 'tick_us': TICK_US, 'strict_unresolved_component_count': len(strict.unresolved), 'candidate_component_count': len(candidates), 'potential_remaining_unresolved_component_count': len(relaxed.unresolved), 'potential_absorbed_unresolved_component_count': len(strict.unresolved) - len(relaxed.unresolved), 'accepted_pair_count': len(allowlist), 'resolved_geometry_changed': False, 'semantic_quality_established': False, 'clean_acceptance_requires_adaptive_fine_confirmation': True, 'candidates': candidates}
+        return replace(strict, diagnostics={**dict(strict.diagnostics), 'unresolved_local_stitch': local_stitch_diagnostics})
+
+    def _assess_result(self, result: FineSegmentationResult, *, inference_arm: str, raw_segment_counts: Sequence[int], emitted_segment_counts: Sequence[int], guard_only_fragments: Sequence[Sequence[object]]) -> FineSegmentationResult:
+        overlap_pairs, overlap_depth = _overlap_statistics(result.events)
+        exact_duplicates = _exact_duplicate_intervals(result.events)
+        policy = self.quality_policy
+        window_profile = fine_window_profile(self.window_profile_id)
+        violations: list[dict[str, object]] = []
+        for index, count in enumerate(raw_segment_counts):
+            if count > policy.hard_max_segments_per_window:
+                violations.append({'code': 'hard_segment_count', 'window_index': index, 'observed': count, 'maximum': policy.hard_max_segments_per_window, 'failure_class': 'delivery_budget_exceeded'})
+        if overlap_depth > policy.hard_max_overlap_depth:
+            violations.append({'code': 'overlap_depth', 'observed': overlap_depth, 'maximum': policy.hard_max_overlap_depth, 'failure_class': 'delivery_budget_exceeded'})
+        if exact_duplicates:
+            violations.append({'code': 'exact_duplicate_intervals', 'observed': exact_duplicates, 'failure_class': 'structural_invalid'})
+        if result.adjacent_matching_ambiguities > policy.max_adjacent_matching_ambiguities:
+            violations.append({'code': 'stitch_ambiguity', 'observed': result.adjacent_matching_ambiguities, 'maximum': policy.max_adjacent_matching_ambiguities, 'failure_class': 'structural_invalid'})
+        if self.fail_on_unresolved and result.unresolved:
+            violations.append({'code': 'unresolved_components', 'observed': len(result.unresolved), 'maximum': 0, 'failure_class': 'structural_invalid'})
+        diagnostics: dict[str, object] = {'schema_version': 'subtask_annotation_atomic_fine_diagnostics_v1', 'inference_arm': inference_arm, 'window_profile_id': window_profile.profile_id, 'window_profile': window_profile.to_dict(), 'model_io_contract': window_profile.model_io_contract, 'target_granularity': window_profile.target_granularity, 'policy': policy.to_dict(), 'raw_segment_counts': list(raw_segment_counts), 'emitted_segment_counts': list(emitted_segment_counts), 'guard_only_dropped': [[asdict(fragment) for fragment in fragments] for fragments in guard_only_fragments], 'soft_limit_exceeded_windows': [index for index, count in enumerate(raw_segment_counts) if count > policy.soft_max_segments_per_window], 'resolved_event_count': len(result.events), 'unresolved_components': [asdict(item) for item in result.unresolved], 'adjacent_matching_ambiguities': result.adjacent_matching_ambiguities, 'overlap_pair_count': overlap_pairs, 'max_overlap_depth': overlap_depth, 'exact_duplicate_intervals': exact_duplicates, 'acceptance_failure_classes': sorted({str(item['failure_class']) for item in violations}), 'violations': violations}
+        if result.diagnostics:
+            diagnostics['stitch_diagnostics'] = dict(result.diagnostics)
+        diagnostics['diagnostics_sha256'] = _diagnostic_sha256(diagnostics)
+        assessed = replace(result, diagnostics=diagnostics)
+        if violations:
+            codes = ','.join((str(item['code']) for item in violations))
+            raise FineSegmentationOutputError(f'fine segmentation failed output acceptance gate: {codes}', diagnostics=diagnostics, partial_result=assessed)
+        return assessed
+
+    def _predict_scheduled_windows(self, source: AnnotationSource, windows: Sequence[FineWindow]) -> _FineWindowBatchPrediction:
+        return asyncio.run(self._predict_scheduled_windows_async(source, windows))
+
+    async def _predict_scheduled_windows_async(self, source: AnnotationSource, windows: Sequence[FineWindow]) -> _FineWindowBatchPrediction:
+        request = {'schema_version': FINE_REQUEST_SCHEMA, 'profile_id': self.profile_id, 'model_binding': dict(self.model_binding), 'source': source_payload(source), 'windows': [window.worker_payload() for window in windows], 'max_concurrency': min(self.batch_concurrency, len(windows))}
+        apredict = getattr(self.predictor, 'apredict', None)
+        if callable(apredict):
+            response = await apredict(request)
+        else:
+            response = await asyncio.to_thread(self.predictor.predict, request)
+        required_keys = {'schema_version', 'source_id', 'model_io_contract', 'model_binding', 'predictions'}
+        if set(response) != required_keys:
+            raise SubtaskAnnotationAdapterError('fine-segmentation response must contain exactly schema_version, source_id, model_io_contract, model_binding, predictions')
+        if response['schema_version'] != FINE_RESPONSE_SCHEMA:
+            raise SubtaskAnnotationAdapterError('fine-segmentation response schema mismatch')
+        if response['source_id'] != source.source_id:
+            raise SubtaskAnnotationAdapterError('fine-segmentation response identity mismatch')
+        if response['model_io_contract'] != fine_window_profile(self.window_profile_id).model_io_contract:
+            raise SubtaskAnnotationAdapterError('fine-segmentation model I/O contract mismatch')
+        if response['model_binding'] != self.model_binding:
+            raise SubtaskAnnotationAdapterError('fine-segmentation worker model_binding mismatch')
+        prediction_rows = response['predictions']
+        if not isinstance(prediction_rows, list) or len(prediction_rows) != len(windows):
+            raise SubtaskAnnotationAdapterError('fine-segmentation predictions must exactly cover every requested window')
+        expected_scope_ids = [window.scope_id for window in windows]
+        observed_scope_ids: list[str] = []
+        fragments_by_window = {}
+        raw_segment_counts: list[int] = []
+        emitted_segment_counts: list[int] = []
+        guard_only_fragments: list[Sequence[object]] = []
+        for window, prediction in zip(windows, prediction_rows, strict=True):
+            if not isinstance(prediction, dict) or set(prediction) != {'scope_id', 'prediction_json'}:
+                raise SubtaskAnnotationAdapterError('each fine-segmentation prediction must contain exactly scope_id and prediction_json')
+            observed_scope_ids.append(prediction['scope_id'])
+            if prediction['scope_id'] != window.scope_id:
+                raise SubtaskAnnotationAdapterError('fine-segmentation prediction order or scope differs')
+            prediction_json = prediction['prediction_json']
+            if not isinstance(prediction_json, str):
+                raise SubtaskAnnotationAdapterError('fine-segmentation prediction_json must be a string')
+            try:
+                parsed = parse_prediction_with_diagnostics(prediction_json, window, drop_guard_only=self.quality_policy.drop_guard_only_segments)
+            except FineTemporalError as exc:
+                raise FineSegmentationOutputError(f'fine-segmentation window {window.window_index} returned an invalid prediction: {exc}', diagnostics={'schema_version': 'subtask_annotation_atomic_fine_diagnostics_v1', 'stage': 'parse_prediction', 'window_index': window.window_index, 'error': str(exc)}) from exc
+            fragments_by_window[window.window_index] = parsed.fragments
+            raw_segment_counts.append(len(parsed.fragments) + len(parsed.guard_only_fragments))
+            emitted_segment_counts.append(len(parsed.fragments))
+            guard_only_fragments.append(parsed.guard_only_fragments)
+        if observed_scope_ids != expected_scope_ids or len(set(observed_scope_ids)) != len(observed_scope_ids):
+            raise SubtaskAnnotationAdapterError('fine-segmentation response scope coverage differs')
+        return _FineWindowBatchPrediction(fragments_by_window={index: tuple(fragments) for index, fragments in fragments_by_window.items()}, raw_segment_counts=tuple(raw_segment_counts), emitted_segment_counts=tuple(emitted_segment_counts), guard_only_fragments=tuple((tuple(items) for items in guard_only_fragments)))
+
+    def predict_repair_candidates(self, source: AnnotationSource, windows: Sequence[FineWindow], *, arm: str='adaptive_fine_repair') -> _FineRetryPrediction:
+        return asyncio.run(self.predict_repair_candidates_async(source, windows, arm=arm))
+
+    async def predict_repair_candidates_async(self, source: AnnotationSource, windows: Sequence[FineWindow], *, arm: str='adaptive_fine_repair') -> _FineRetryPrediction:
+        batch = await self._predict_scheduled_windows_async(source, windows)
+        hard_limit_violations = [{'code': 'hard_segment_count', 'window_index': window.window_index, 'observed': count, 'maximum': self.quality_policy.hard_max_segments_per_window, 'failure_class': 'delivery_budget_exceeded'} for window, count in zip(windows, batch.raw_segment_counts, strict=True) if count > self.quality_policy.hard_max_segments_per_window]
+        diagnostics: dict[str, object] = {'schema_version': ADAPTIVE_FINE_REPAIR_SCHEMA, 'stage': 'fine_retry_prediction', 'inference_arm': arm, 'worker_batch_call_count': 1, 'retry_window_count': len(windows), 'raw_segment_counts': list(batch.raw_segment_counts), 'emitted_segment_counts': list(batch.emitted_segment_counts), 'guard_only_dropped': [[asdict(fragment) for fragment in fragments] for fragments in batch.guard_only_fragments], 'violations': hard_limit_violations}
+        diagnostics['diagnostics_sha256'] = _diagnostic_sha256(diagnostics)
+        if hard_limit_violations:
+            raise FineSegmentationOutputError('adaptive fine retry exceeded the per-window segment budget', diagnostics=diagnostics)
+        candidates_by_window: dict[int, tuple[TemporalSegment, ...]] = {}
+        for window in windows:
+            candidates: list[TemporalSegment] = []
+            for fragment in batch.fragments_by_window[window.window_index]:
+                start_us = (window.global_tick_offset + fragment.s) * TICK_US
+                end_us = (window.global_tick_offset + fragment.e) * TICK_US
+                physical_start_us = max(window.valid_start_us, start_us)
+                physical_end_us = min(window.valid_end_us, end_us)
+                if physical_end_us <= physical_start_us:
+                    raise FineSegmentationOutputError('adaptive fine retry produced an empty physical source interval', diagnostics={**diagnostics, 'window_index': window.window_index, 'fragment': asdict(fragment), 'model_tick_span_us': {'start': start_us, 'end': end_us}, 'physical_valid_span_us': {'start': window.valid_start_us, 'end': window.valid_end_us}})
+                physical_clip = None
+                if physical_start_us != start_us or physical_end_us != end_us:
+                    physical_clip = {'contract': 'model_tick_span_intersect_physical_valid_span_v1', 'model_tick_span_us': {'start': start_us, 'end': end_us}, 'physical_valid_span_us': {'start': window.valid_start_us, 'end': window.valid_end_us}, 'clipped_start': physical_start_us != start_us, 'clipped_end': physical_end_us != end_us}
+                candidates.append(TemporalSegment(TimeSpan(physical_start_us / 1000000, physical_end_us / 1000000), {'scope_clipped_start': fragment.lo, 'scope_clipped_end': fragment.ro, 'fine_retry_scope_id': window.scope_id, 'fine_retry_window_index': window.window_index, **({'fine_retry_physical_clip': physical_clip} if physical_clip is not None else {})}))
+            candidates_by_window[window.window_index] = tuple(candidates)
+        return _FineRetryPrediction(candidates_by_window=candidates_by_window, diagnostics=diagnostics)
+
+    def segment_with_priors(self, source: AnnotationSource, coarse_proposals: Sequence[ProcessingScope], *, arm: str='no_prior_global_scan') -> FineSegmentationResult:
+        return asyncio.run(self.segment_with_priors_async(source, coarse_proposals, arm=arm))
+
+    async def segment_with_priors_async(self, source: AnnotationSource, coarse_proposals: Sequence[ProcessingScope], *, arm: str='no_prior_global_scan') -> FineSegmentationResult:
+        if source.duration_s is None:
+            raise SubtaskAnnotationAdapterError('fine segmentation requires exact source duration_s')
+        scopes = schedule_fine_scopes(source_id=source.source_id, duration_s=source.duration_s, coarse_proposals=coarse_proposals, arm=arm, window_profile_id=self.window_profile_id)
+        windows = tuple((window_from_scope(scope) for scope in scopes))
+        batch = await self._predict_scheduled_windows_async(source, windows)
+        result = self._stitch_with_unresolved_local_candidates(windows, batch.fragments_by_window)
+        return self._assess_result(result, inference_arm=arm, raw_segment_counts=batch.raw_segment_counts, emitted_segment_counts=batch.emitted_segment_counts, guard_only_fragments=batch.guard_only_fragments)
+
+    def segment(self, source: AnnotationSource) -> tuple[TemporalSegment, ...]:
+        return asyncio.run(self.segment_async(source))
+
+    async def segment_async(self, source: AnnotationSource) -> tuple[TemporalSegment, ...]:
+        result = await self.segment_with_priors_async(source, (), arm='no_prior_global_scan')
+        return events_to_temporal_segments(result.events, diagnostics=result.diagnostics)
+
+@dataclass(frozen=True)
+class _AcceptedFineRetry:
+    component_index: int
+    window: FineWindow
+    candidate: TemporalSegment
+    repaired_span: TimeSpan
+    repair_mode: str = 'boundary_completion'
+    confirmation_target_event_id: str | None = None
+    local_stitch_candidate: Mapping[str, object] | None = None
+
+@dataclass(frozen=True)
+class _MatchedFineRetryCandidate:
+    candidate: TemporalSegment
+    repaired_span: TimeSpan
+    repair_mode: str
+    local_stitch_candidate: Mapping[str, object] | None = None
+
+def _tick_seconds(value: int) -> float:
+    return value * TICK_US / 1000000
+
+def _same_event_geometry(left: TimeSpan, right: TimeSpan, *, tolerance_s: float=SEGMENT_LOCAL_REPAIR_ENDPOINT_TOLERANCE_S) -> bool:
+    return abs(left.start_s - right.start_s) <= tolerance_s and abs(left.end_s - right.end_s) <= tolerance_s
+
+def _unresolved_local_stitch_candidates(diagnostics: Mapping[str, object]) -> dict[int, dict[str, object]]:
+    stitch_diagnostics = diagnostics.get('stitch_diagnostics')
+    if not isinstance(stitch_diagnostics, dict):
+        return {}
+    local = stitch_diagnostics.get('unresolved_local_stitch')
+    if not isinstance(local, dict):
+        return {}
+    candidates = local.get('candidates')
+    if not isinstance(candidates, list):
+        raise SubtaskAnnotationAdapterError('unresolved-local stitch diagnostics contain an invalid candidate ledger')
+    by_component: dict[int, dict[str, object]] = {}
+    for candidate in candidates:
+        if not isinstance(candidate, dict):
+            raise SubtaskAnnotationAdapterError('unresolved-local stitch candidate must be an object')
+        component_index = candidate.get('partial_component_index')
+        event_id = candidate.get('resolved_event_id')
+        span = candidate.get('resolved_event_span')
+        if isinstance(component_index, bool) or not isinstance(component_index, int) or component_index < 0 or (not isinstance(event_id, str)) or (not event_id) or (not isinstance(span, dict)) or (set(span) != {'start_s', 'end_s'}):
+            raise SubtaskAnnotationAdapterError('unresolved-local stitch candidate binding is invalid')
+        if component_index in by_component:
+            raise SubtaskAnnotationAdapterError('unresolved-local stitch candidate ledger repeats a component')
+        by_component[component_index] = dict(candidate)
+    return by_component
+
+def _local_stitch_target_span(candidate: Mapping[str, object]) -> TimeSpan:
+    payload = candidate.get('resolved_event_span')
+    if not isinstance(payload, dict):
+        raise SubtaskAnnotationAdapterError('unresolved-local stitch candidate target span is missing')
+    start_s, end_s = (payload.get('start_s'), payload.get('end_s'))
+    if isinstance(start_s, bool) or not isinstance(start_s, int | float) or isinstance(end_s, bool) or (not isinstance(end_s, int | float)):
+        raise SubtaskAnnotationAdapterError('unresolved-local stitch candidate target span is invalid')
+    return TimeSpan(float(start_s), float(end_s))
+
+def _retry_confirms_local_stitch(retry_candidate: TemporalSegment, local_candidate: Mapping[str, object]) -> bool:
+    return _same_event_geometry(retry_candidate.span, _local_stitch_target_span(local_candidate), tolerance_s=STITCH_TOLERANCE_TICKS * TICK_US / 1000000)
+
+def _component_observation_span(component: UnresolvedFineEvent) -> TimeSpan:
+    return TimeSpan(_tick_seconds(min((item[0] for item in component.observation_spans_ticks))), _tick_seconds(max((item[1] for item in component.observation_spans_ticks))))
+
+def _endpoint_consensus_ticks(values: Sequence[int]) -> int:
+    ordered = sorted(values)
+    if not ordered:
+        raise SubtaskAnnotationAdapterError('cannot form endpoint consensus without evidence')
+    middle = len(ordered) // 2
+    if len(ordered) % 2:
+        return ordered[middle]
+    return (ordered[middle - 1] + ordered[middle] + 1) // 2
+
+def _conservative_component_completion_span(source: AnnotationSource, component: UnresolvedFineEvent) -> tuple[TimeSpan, str]:
+    if source.duration_s is None:
+        raise SubtaskAnnotationAdapterError('conservative fine completion requires exact source duration_s')
+    if not component.observation_spans_ticks:
+        raise SubtaskAnnotationAdapterError('conservative fine completion requires at least one observed fragment')
+    observed_start_tick = min((item[0] for item in component.observation_spans_ticks))
+    observed_end_tick = max((item[1] for item in component.observation_spans_ticks))
+    has_start = bool(component.closed_start_ticks)
+    has_end = bool(component.closed_end_ticks)
+    start_tick = _endpoint_consensus_ticks(component.closed_start_ticks) if has_start else observed_start_tick
+    end_tick = _endpoint_consensus_ticks(component.closed_end_ticks) if has_end else observed_end_tick
+    strategy = {(True, False): 'closed_start_plus_observed_censored_end_v1', (False, True): 'observed_censored_start_plus_closed_end_v1', (True, True): 'closed_endpoint_consensus_v1', (False, False): 'observed_fragment_envelope_v1'}[has_start, has_end]
+    start_s = max(0.0, _tick_seconds(start_tick))
+    end_s = min(source.duration_s, _tick_seconds(end_tick))
+    if end_s <= start_s:
+        start_s = max(0.0, _tick_seconds(observed_start_tick))
+        end_s = min(source.duration_s, _tick_seconds(observed_end_tick))
+        strategy = 'observed_fragment_envelope_after_endpoint_disagreement_v1'
+    if end_s <= start_s:
+        raise SubtaskAnnotationAdapterError('conservative fine completion produced an empty physical interval')
+    return (TimeSpan(start_s, end_s), strategy)
+
+def _component_boundary_completion_span(component: UnresolvedFineEvent, segment: TemporalSegment) -> TimeSpan | None:
+    has_start = bool(component.closed_start_ticks)
+    has_end = bool(component.closed_end_ticks)
+    if has_start == has_end:
+        return None
+    observation = _component_observation_span(component)
+    if has_start:
+        if segment.metadata.get('scope_clipped_end') is True:
+            return None
+        if not segment.span.start_s < observation.end_s <= segment.span.end_s + 1e-09:
+            return None
+        start_s = _tick_seconds(_endpoint_consensus_ticks(component.closed_start_ticks))
+        end_s = segment.span.end_s
+    else:
+        if segment.metadata.get('scope_clipped_start') is True:
+            return None
+        if not segment.span.start_s <= observation.start_s + 1e-09 < segment.span.end_s:
+            return None
+        start_s = segment.span.start_s
+        end_s = _tick_seconds(_endpoint_consensus_ticks(component.closed_end_ticks))
+    if end_s <= start_s:
+        return None
+    return TimeSpan(start_s, end_s)
+
+def _match_fine_retry_candidates(component_index: int, component: UnresolvedFineEvent, candidates: Sequence[TemporalSegment], local_stitch_candidates: Mapping[int, Mapping[str, object]]) -> tuple[tuple[_MatchedFineRetryCandidate, ...], tuple[dict[str, object], ...]]:
+    local_candidate = local_stitch_candidates.get(component_index)
+    matching: list[_MatchedFineRetryCandidate] = []
+    ignored: list[dict[str, object]] = []
+    for candidate in candidates:
+        if local_candidate is not None and _retry_confirms_local_stitch(candidate, local_candidate):
+            matching.append(_MatchedFineRetryCandidate(candidate=candidate, repaired_span=_local_stitch_target_span(local_candidate), repair_mode='unresolved_local_stitch_confirmed', local_stitch_candidate=local_candidate))
+            continue
+        repaired_span = _component_boundary_completion_span(component, candidate)
+        if repaired_span is not None:
+            matching.append(_MatchedFineRetryCandidate(candidate=candidate, repaired_span=repaired_span, repair_mode='boundary_completion'))
+            continue
+        ignored.append({'span': _span_payload(candidate.span), 'scope_clipped_start': candidate.metadata.get('scope_clipped_start'), 'scope_clipped_end': candidate.metadata.get('scope_clipped_end')})
+    return (tuple(matching), tuple(ignored))
+
+def _covering_span(left: TimeSpan, right: TimeSpan) -> TimeSpan:
+    return TimeSpan(min(left.start_s, right.start_s), max(left.end_s, right.end_s))
+
+def _unsupported_unresolved_component_indices(source: AnnotationSource, unresolved: Sequence[UnresolvedFineEvent]) -> list[int]:
+    return [index for index, component in enumerate(unresolved) if component.matching_ambiguous or (component.closed_start_ticks and component.closed_end_ticks) or (not component.closed_start_ticks and (not component.closed_end_ticks)) or (not component.closed_start_ticks and min((item[0] for item in component.observation_spans_ticks)) <= 0) or (not component.closed_end_ticks and source.duration_s is not None and (_tick_seconds(max((item[1] for item in component.observation_spans_ticks))) >= source.duration_s - 1e-09))]
+
+def _adaptive_fine_failure(exc: FineSegmentationOutputError, *, code: str, detail: Mapping[str, object]) -> FineSegmentationOutputError:
+    diagnostics = dict(exc.diagnostics)
+    failures = list(diagnostics.get('adaptive_fine_repair_failures', []))
+    failures.append({'code': code, **dict(detail)})
+    diagnostics['adaptive_fine_repair_failures'] = failures
+    diagnostics['diagnostics_sha256'] = _diagnostic_sha256({key: value for key, value in diagnostics.items() if key != 'diagnostics_sha256'})
+    return FineSegmentationOutputError(f'fine segmentation could not be repaired by bounded fine retry: {code}', diagnostics=diagnostics, partial_result=exc.partial_result)
+
+def _remaining_adaptive_fine_error(exc: FineSegmentationOutputError, result: FineSegmentationResult, *, repaired_events: tuple[EventInterval, ...], accepted_component_indices: Sequence[int], local_stitch_candidates: Mapping[int, Mapping[str, object]], deferred_failures: Sequence[Mapping[str, object]], repair_payload: Mapping[str, object], inference_context: Mapping[str, object]) -> FineSegmentationOutputError:
+    accepted = set(accepted_component_indices)
+    remaining_indices = [index for index in range(len(result.unresolved)) if index not in accepted]
+    if not remaining_indices:
+        raise SubtaskAnnotationAdapterError('remaining adaptive Fine error requires at least one unresolved component')
+    original_rows = exc.diagnostics.get('unresolved_components')
+    rows_are_aligned = isinstance(original_rows, list) and len(original_rows) == len(result.unresolved)
+    remaining_rows = [original_rows[index] for index in remaining_indices] if rows_are_aligned else []
+    repaired_rows = [original_rows[index] for index in sorted(accepted)] if rows_are_aligned else []
+    reduced_index = {old_index: new_index for new_index, old_index in enumerate(remaining_indices)}
+    remaining_local_candidates = []
+    for old_index in remaining_indices:
+        candidate = local_stitch_candidates.get(old_index)
+        if candidate is None:
+            continue
+        rebound = dict(candidate)
+        rebound['partial_component_index'] = reduced_index[old_index]
+        rebound['original_partial_component_index'] = old_index
+        remaining_local_candidates.append(rebound)
+    diagnostics = dict(exc.diagnostics)
+    failures = list(diagnostics.get('adaptive_fine_repair_failures', []))
+    failures.extend((dict(item) for item in deferred_failures))
+    diagnostics['adaptive_fine_repair_failures'] = failures
+    diagnostics['pre_repair_violations'] = list(diagnostics.get('violations', []))
+    diagnostics['repaired_unresolved_components'] = repaired_rows
+    diagnostics['unresolved_components'] = remaining_rows
+    violations = [dict(item) for item in diagnostics.get('violations', []) if isinstance(item, dict)]
+    for violation in violations:
+        if violation.get('code') == 'unresolved_components':
+            violation['observed'] = len(remaining_indices)
+    diagnostics['violations'] = violations
+    diagnostics['resolved_event_count'] = len(repaired_events)
+    diagnostics['overlap_pair_count'] = repair_payload['post_repair_overlap_pair_count']
+    diagnostics['max_overlap_depth'] = repair_payload['post_repair_max_overlap_depth']
+    diagnostics['exact_duplicate_intervals'] = repair_payload['post_repair_exact_duplicate_intervals']
+    diagnostics['adaptive_fine_repair'] = dict(repair_payload)
+    diagnostics['adaptive_fine_repair_sha256'] = _diagnostic_sha256(repair_payload)
+    stitch_diagnostics = dict(diagnostics.get('stitch_diagnostics', {}))
+    local_diagnostics = dict(stitch_diagnostics.get('unresolved_local_stitch', {}))
+    if local_diagnostics:
+        local_diagnostics.update({'accepted_pair_count': len(remaining_local_candidates), 'candidate_component_count': len(remaining_local_candidates), 'candidates': remaining_local_candidates, 'potential_absorbed_unresolved_component_count': len(remaining_local_candidates), 'potential_remaining_unresolved_component_count': len(remaining_indices) - len(remaining_local_candidates), 'strict_unresolved_component_count': len(remaining_indices)})
+        stitch_diagnostics['unresolved_local_stitch'] = local_diagnostics
+        diagnostics['stitch_diagnostics'] = stitch_diagnostics
+    diagnostics = _with_inference_context(diagnostics, inference_context)
+    partial_result = FineSegmentationResult(events=repaired_events, fragments=result.fragments, unresolved=tuple((result.unresolved[index] for index in remaining_indices)), adjacent_matching_ambiguities=result.adjacent_matching_ambiguities, diagnostics=diagnostics)
+    return FineSegmentationOutputError('bounded fine retry partially repaired the source; remaining components require conservative completion', diagnostics=diagnostics, partial_result=partial_result)
+
+@dataclass(frozen=True)
+class HierarchicalSubtaskSegmenter:
+    coarse_segmenter: VideoSegmenter
+    fine_segmenter: FineSubtaskSegmenter
+    fine_prior_mode: str = 'none'
+    fine_failure_policy: str = 'adaptive_fine_then_conservative_completion'
+    fine_output_coverage_policy: str = 'allow_gaps'
+    adaptive_fine_retry_max_windows: int = ADAPTIVE_FINE_RETRY_MAX_WINDOWS
+    segment_topology = SegmentTopology.CONCURRENT
+
+    def __post_init__(self):
+        if self.fine_prior_mode != 'none' or self.fine_failure_policy != 'adaptive_fine_then_conservative_completion' or self.fine_output_coverage_policy != 'allow_gaps':
+            raise ValueError('Use the released inference workflow configuration.')
+        if isinstance(self.adaptive_fine_retry_max_windows, bool) or not isinstance(self.adaptive_fine_retry_max_windows, int) or (not 1 <= self.adaptive_fine_retry_max_windows <= ADAPTIVE_FINE_RETRY_HARD_MAX_WINDOWS):
+            raise ValueError('Invalid adaptive inference retry budget')
+
+    def adaptive_fine_repair(self, source: AnnotationSource, exc: FineSegmentationOutputError, *, inference_context: Mapping[str, object] | None=None) -> tuple[TemporalSegment, ...]:
+        return asyncio.run(self.adaptive_fine_repair_async(source, exc, inference_context=inference_context))
+
+    async def adaptive_fine_repair_async(self, source: AnnotationSource, exc: FineSegmentationOutputError, *, inference_context: Mapping[str, object] | None=None) -> tuple[TemporalSegment, ...]:
+        if inference_context is None:
+            inference_context = _inference_context(source, (), fine_prior_mode='none')
+        result = exc.partial_result
+        violation_codes = [str(item.get('code')) for item in exc.diagnostics.get('violations', []) if isinstance(item, dict)]
+        if result is None or not result.unresolved or violation_codes != ['unresolved_components']:
+            raise _adaptive_fine_failure(exc, code='ineligible_fine_failure', detail={'violation_codes': violation_codes}) from exc
+        if source.duration_s is None:
+            raise _adaptive_fine_failure(exc, code='duration_unavailable', detail={}) from exc
+        progressive_provider_free = self.fine_failure_policy in PROVIDER_FREE_FINE_COMPLETION_POLICIES
+        window_profile = fine_window_profile(self.fine_segmenter.window_profile_id)
+        local_stitch_candidates = _unresolved_local_stitch_candidates(exc.diagnostics)
+        unsupported_components = _unsupported_unresolved_component_indices(source, result.unresolved)
+        if unsupported_components and (not progressive_provider_free):
+            raise _adaptive_fine_failure(exc, code='unsupported_unresolved_geometry', detail={'component_indices': unsupported_components}) from exc
+        if len(result.unresolved) > self.adaptive_fine_retry_max_windows and (not progressive_provider_free):
+            raise _adaptive_fine_failure(exc, code='retry_budget_exceeded', detail={'unresolved_component_count': len(result.unresolved), 'max_retry_windows': self.adaptive_fine_retry_max_windows}) from exc
+        deferred_failures: list[dict[str, object]] = []
+        unsupported_set = set(unsupported_components)
+        if unsupported_components:
+            deferred_failures.append({'code': 'unsupported_unresolved_geometry', 'component_indices': unsupported_components})
+        eligible_component_indices = [index for index in range(len(result.unresolved)) if index not in unsupported_set]
+        eligible_component_indices.sort(key=lambda index: (index not in local_stitch_candidates, index))
+        selected_component_indices = eligible_component_indices[:self.adaptive_fine_retry_max_windows]
+        budget_deferred_indices = eligible_component_indices[self.adaptive_fine_retry_max_windows:]
+        if budget_deferred_indices:
+            deferred_failures.append({'code': 'retry_budget_exceeded', 'unresolved_component_count': len(result.unresolved), 'max_retry_windows': self.adaptive_fine_retry_max_windows, 'deferred_component_indices': budget_deferred_indices, 'candidate_priority': 'unresolved_local_stitch_first_then_source_order_v1'})
+        if not selected_component_indices:
+            diagnostics = dict(exc.diagnostics)
+            failures = list(diagnostics.get('adaptive_fine_repair_failures', []))
+            failures.extend(deferred_failures)
+            diagnostics['adaptive_fine_repair_failures'] = failures
+            diagnostics['diagnostics_sha256'] = _diagnostic_sha256({key: value for key, value in diagnostics.items() if key != 'diagnostics_sha256'})
+            raise FineSegmentationOutputError('fine segmentation had no eligible bounded fine retry components', diagnostics=diagnostics, partial_result=result) from exc
+        component_edge_ticks = tuple(((component_index, max((item[1] for item in component.observation_spans_ticks)) if component.closed_start_ticks else min((item[0] for item in component.observation_spans_ticks))) for component_index in selected_component_indices for component in (result.unresolved[component_index],)))
+        try:
+            scopes = schedule_recentered_fine_retry_scopes(source_id=source.source_id, duration_s=source.duration_s, component_edge_ticks=component_edge_ticks, window_profile_id=window_profile.profile_id)
+            windows = tuple((window_from_scope(scope) for scope in scopes))
+        except Exception as geometry_error:
+            raise _adaptive_fine_failure(exc, code='invalid_retry_geometry', detail={'error_type': type(geometry_error).__name__, 'error': str(geometry_error)}) from geometry_error
+        try:
+            prediction = await self.fine_segmenter.predict_repair_candidates_async(source, windows)
+        except FineSegmentationOutputError as retry_error:
+            raise _adaptive_fine_failure(exc, code='retry_output_rejected', detail={'error': str(retry_error), 'retry_diagnostics': retry_error.diagnostics}) from retry_error
+        accepted_by_component: dict[int, _AcceptedFineRetry] = {}
+        retry_diagnostics_by_component: dict[int, dict[str, object]] = {}
+        incomplete_primary: list[tuple[int, int, dict[str, object]]] = []
+        for (component_index, edge_tick), window in zip(component_edge_ticks, windows, strict=True):
+            component = result.unresolved[component_index]
+            candidates = prediction.candidates_by_window[window.window_index]
+            matching_candidates, ignored_candidates = _match_fine_retry_candidates(component_index, component, candidates, local_stitch_candidates)
+            failure_detail = {'component_index': component_index, 'retry_scope_id': window.scope_id, 'retry_edge_tick': edge_tick, 'candidate_count': len(candidates), 'matching_candidate_count': len(matching_candidates), 'ignored_candidates': ignored_candidates, 'retry_prediction_diagnostics': dict(prediction.diagnostics)}
+            if len(matching_candidates) > 1:
+                if not progressive_provider_free:
+                    raise _adaptive_fine_failure(exc, code='ambiguous_component_coverage', detail=failure_detail)
+                deferred_failures.append({'code': 'ambiguous_component_coverage', **failure_detail})
+                continue
+            if not matching_candidates:
+                if not candidates:
+                    if not progressive_provider_free:
+                        raise _adaptive_fine_failure(exc, code='incomplete_component_coverage', detail=failure_detail)
+                    deferred_failures.append({'code': 'incomplete_component_coverage', **failure_detail})
+                    continue
+                incomplete_primary.append((component_index, edge_tick, failure_detail))
+                continue
+            match = matching_candidates[0]
+            accepted_by_component[component_index] = _AcceptedFineRetry(component_index=component_index, window=window, candidate=match.candidate, repaired_span=match.repaired_span, repair_mode=match.repair_mode, confirmation_target_event_id=str(match.local_stitch_candidate['resolved_event_id']) if match.local_stitch_candidate is not None else None, local_stitch_candidate=match.local_stitch_candidate)
+            retry_diagnostics_by_component[component_index] = {'component_index': component_index, 'retry_pass': 'centered', 'repair_mode': match.repair_mode, 'retry_scope_id': window.scope_id, 'retry_edge_tick': edge_tick, 'requested_view_start_s': window.requested_view_start_us / 1000000, 'requested_view_end_s': window.requested_view_end_us / 1000000, 'candidate_count': len(candidates), 'candidate_evidence_span': _span_payload(match.candidate.span), 'repaired_span': _span_payload(match.repaired_span), 'ignored_candidates': list(ignored_candidates), **({'unresolved_local_stitch_confirmation': dict(match.local_stitch_candidate)} if match.local_stitch_candidate is not None else {})}
+        secondary_windows: tuple[FineWindow, ...] = ()
+        secondary_prediction: _FineRetryPrediction | None = None
+        if incomplete_primary:
+            secondary_edges = tuple(((component_index, edge_tick) for component_index, edge_tick, _ in incomplete_primary))
+            secondary_offsets = {component_index: window_profile.left_guard_us // TICK_US if result.unresolved[component_index].closed_start_ticks else window_profile.core_ticks - window_profile.right_guard_us // TICK_US for component_index, _, _ in incomplete_primary}
+            try:
+                secondary_scopes = schedule_recentered_fine_retry_scopes(source_id=source.source_id, duration_s=source.duration_s, component_edge_ticks=secondary_edges, arm='adaptive_fine_directional_retry', edge_core_offset_ticks=secondary_offsets, scope_namespace=FINE_DIRECTIONAL_RETRY_SCOPE_NAMESPACE, window_profile_id=window_profile.profile_id)
+                secondary_windows = tuple((window_from_scope(scope) for scope in secondary_scopes))
+            except Exception as geometry_error:
+                raise _adaptive_fine_failure(exc, code='invalid_directional_retry_geometry', detail={'error_type': type(geometry_error).__name__, 'error': str(geometry_error), 'component_indices': [item[0] for item in incomplete_primary]}) from geometry_error
+            try:
+                secondary_prediction = await self.fine_segmenter.predict_repair_candidates_async(source, secondary_windows, arm='adaptive_fine_directional_retry')
+            except FineSegmentationOutputError as retry_error:
+                raise _adaptive_fine_failure(exc, code='directional_retry_output_rejected', detail={'error': str(retry_error), 'retry_diagnostics': retry_error.diagnostics}) from retry_error
+            for (component_index, edge_tick, primary_failure), window in zip(incomplete_primary, secondary_windows, strict=True):
+                component = result.unresolved[component_index]
+                candidates = secondary_prediction.candidates_by_window[window.window_index]
+                matching_candidates, ignored_candidates = _match_fine_retry_candidates(component_index, component, candidates, local_stitch_candidates)
+                if len(matching_candidates) != 1:
+                    failure = {'code': 'incomplete_component_coverage' if not matching_candidates else 'ambiguous_component_coverage', 'component_index': component_index, 'retry_pass': 'directional_missing_endpoint_context', 'retry_scope_id': window.scope_id, 'retry_edge_tick': edge_tick, 'candidate_count': len(candidates), 'matching_candidate_count': len(matching_candidates), 'ignored_candidates': list(ignored_candidates), 'primary_retry_failure': primary_failure, 'retry_prediction_diagnostics': dict(secondary_prediction.diagnostics)}
+                    if not progressive_provider_free:
+                        raise _adaptive_fine_failure(exc, code=str(failure.pop('code')), detail=failure)
+                    deferred_failures.append(failure)
+                    continue
+                match = matching_candidates[0]
+                accepted_by_component[component_index] = _AcceptedFineRetry(component_index=component_index, window=window, candidate=match.candidate, repaired_span=match.repaired_span, repair_mode=match.repair_mode, confirmation_target_event_id=str(match.local_stitch_candidate['resolved_event_id']) if match.local_stitch_candidate is not None else None, local_stitch_candidate=match.local_stitch_candidate)
+                retry_diagnostics_by_component[component_index] = {'component_index': component_index, 'retry_pass': 'directional_missing_endpoint_context', 'repair_mode': match.repair_mode, 'retry_scope_id': window.scope_id, 'retry_edge_tick': edge_tick, 'retry_edge_core_offset_tick': secondary_offsets[component_index], 'requested_view_start_s': window.requested_view_start_us / 1000000, 'requested_view_end_s': window.requested_view_end_us / 1000000, 'candidate_count': len(candidates), 'candidate_evidence_span': _span_payload(match.candidate.span), 'repaired_span': _span_payload(match.repaired_span), 'ignored_candidates': list(ignored_candidates), 'primary_retry_failure': primary_failure, **({'unresolved_local_stitch_confirmation': dict(match.local_stitch_candidate)} if match.local_stitch_candidate is not None else {})}
+        accepted = [accepted_by_component[index] for index in sorted(accepted_by_component)]
+        retry_diagnostics = [retry_diagnostics_by_component[index] for index in sorted(retry_diagnostics_by_component)]
+        combined_events = [replace(event, metadata={'segmentation_provenance': 'fine', **dict(event.metadata)}) for event in result.events]
+        retry_event_count_added = 0
+        suppressed_duplicate_count = 0
+        local_stitch_confirmation_count = 0
+        for repair in accepted:
+            if repair.repair_mode == 'unresolved_local_stitch_confirmed':
+                geometry_matches = [(0.0, index) for index, event in enumerate(combined_events) if event.event_id == repair.confirmation_target_event_id]
+                if len(geometry_matches) != 1 or repair.local_stitch_candidate is None:
+                    raise _adaptive_fine_failure(exc, code='invalid_local_stitch_confirmation_target', detail={'component_index': repair.component_index, 'target_event_id': repair.confirmation_target_event_id, 'target_match_count': len(geometry_matches)})
+            else:
+                geometry_matches = [(abs(event.span.start_s - repair.repaired_span.start_s) + abs(event.span.end_s - repair.repaired_span.end_s), index) for index, event in enumerate(combined_events) if _same_event_geometry(event.span, repair.repaired_span)]
+            if geometry_matches:
+                _, event_index = min(geometry_matches)
+                matched = combined_events[event_index]
+                metadata = dict(matched.metadata)
+                lineage = metadata.get('adaptive_fine_repair')
+                prior_lineage = dict(lineage) if isinstance(lineage, dict) else {}
+                component_indices = set(prior_lineage.get('component_indices', []))
+                component_indices.add(repair.component_index)
+                retry_scope_ids = set(prior_lineage.get('retry_scope_ids', []))
+                retry_scope_ids.add(repair.window.scope_id)
+                metadata.update({'segmentation_provenance': 'fine_repaired_by_fine_retry', 'adaptive_fine_repair': {'component_indices': sorted(component_indices), 'retry_scope_ids': sorted(retry_scope_ids), 'candidate_evidence_span': _span_payload(repair.candidate.span), 'duplicate_suppressed': True, 'repair_mode': repair.repair_mode}, **({'unresolved_local_stitch_confirmation': {'schema_version': UNRESOLVED_LOCAL_STITCH_SCHEMA, 'status': 'confirmed_by_adaptive_fine_retry', 'strict_confirmation_tolerance_ticks': STITCH_TOLERANCE_TICKS, 'retry_candidate_span': _span_payload(repair.candidate.span), 'candidate': dict(repair.local_stitch_candidate)}} if repair.local_stitch_candidate is not None else {})})
+                combined_events[event_index] = replace(matched, source_scope_ids=tuple(sorted({*matched.source_scope_ids, repair.window.scope_id})), metadata=metadata)
+                suppressed_duplicate_count += 1
+                local_stitch_confirmation_count += int(repair.repair_mode == 'unresolved_local_stitch_confirmed')
+                continue
+            candidate_metadata = {key: value for key, value in repair.candidate.metadata.items() if key not in {'scope_clipped_start', 'scope_clipped_end'}}
+            combined_events.append(EventInterval(event_id=f'{source.source_id}:adaptive-fine-repair:{retry_event_count_added:06d}', source_id=source.source_id, span=repair.repaired_span, source_scope_ids=(*sorted({*result.unresolved[repair.component_index].source_scope_ids, repair.window.scope_id}),), metadata={**candidate_metadata, 'segmentation_provenance': 'fine_repaired_by_fine_retry', 'adaptive_fine_repair': {'component_indices': [repair.component_index], 'retry_scope_ids': [repair.window.scope_id], 'candidate_evidence_span': _span_payload(repair.candidate.span), 'duplicate_suppressed': False, 'repair_mode': repair.repair_mode}}))
+            retry_event_count_added += 1
+        ordered_events = tuple(sorted(combined_events, key=lambda item: (item.span, item.event_id)))
+        overlap_pairs, overlap_depth = _overlap_statistics(ordered_events)
+        exact_duplicates = _exact_duplicate_intervals(ordered_events)
+        post_repair_violations: list[dict[str, object]] = []
+        if overlap_depth > self.fine_segmenter.quality_policy.hard_max_overlap_depth:
+            post_repair_violations.append({'code': 'overlap_depth', 'observed': overlap_depth, 'maximum': self.fine_segmenter.quality_policy.hard_max_overlap_depth, 'failure_class': 'delivery_budget_exceeded'})
+        if exact_duplicates:
+            post_repair_violations.append({'code': 'exact_duplicate_intervals', 'observed': exact_duplicates, 'failure_class': 'structural_invalid'})
+        if post_repair_violations:
+            raise _adaptive_fine_failure(exc, code='invalid_combined_geometry', detail={'combined_event_count': len(ordered_events), 'violations': post_repair_violations})
+        repair_payload: dict[str, object] = {'schema_version': ADAPTIVE_FINE_REPAIR_SCHEMA, 'scheduler_contract': 'adaptive_recentered_then_directional_' + window_profile.retry_scheduler_contract(directional=True).removeprefix('adaptive_directional_') if secondary_windows else window_profile.retry_scheduler_contract(directional=False), 'model_io_contract': window_profile.model_io_contract, 'window_profile_id': window_profile.profile_id, 'max_retry_windows': self.adaptive_fine_retry_max_windows, 'unresolved_component_count': len(result.unresolved), 'selected_component_count': len(selected_component_indices), 'selected_component_indices': selected_component_indices, 'selection_priority': 'unresolved_local_stitch_first_then_source_order_v1', 'budget_deferred_component_count': len(budget_deferred_indices), 'unsupported_deferred_component_count': len(unsupported_components), 'repaired_component_count': len(accepted), 'worker_batch_call_count': 1 + int(bool(secondary_windows)), 'retry_window_count': len(windows) + len(secondary_windows), 'centered_retry_window_count': len(windows), 'directional_retry_window_count': len(secondary_windows), 'fine_event_count_before': len(result.events), 'retry_event_count_added': retry_event_count_added, 'retry_duplicate_suppressed': suppressed_duplicate_count, 'unresolved_local_stitch_candidate_count': len(local_stitch_candidates), 'unresolved_local_stitch_confirmation_count': local_stitch_confirmation_count, 'combined_event_count': len(ordered_events), 'post_repair_overlap_pair_count': overlap_pairs, 'post_repair_max_overlap_depth': overlap_depth, 'post_repair_exact_duplicate_intervals': exact_duplicates, 'retry_prediction_diagnostics': dict(prediction.diagnostics), **({'directional_retry_prediction_diagnostics': dict(secondary_prediction.diagnostics)} if secondary_prediction is not None else {}), 'components': retry_diagnostics, 'component_failures': deferred_failures}
+        remaining_component_indices = [index for index in range(len(result.unresolved)) if index not in accepted_by_component]
+        repair_payload['remaining_component_count'] = len(remaining_component_indices)
+        repair_payload['partial_success'] = bool(accepted) and bool(remaining_component_indices)
+        if remaining_component_indices:
+            raise _remaining_adaptive_fine_error(exc, result, repaired_events=ordered_events, accepted_component_indices=tuple(accepted_by_component), local_stitch_candidates=local_stitch_candidates, deferred_failures=deferred_failures, repair_payload=repair_payload, inference_context=inference_context) from exc
+        diagnostics = dict(exc.diagnostics)
+        diagnostics['pre_repair_violations'] = list(diagnostics.get('violations', []))
+        diagnostics['repaired_unresolved_components'] = list(diagnostics.get('unresolved_components', []))
+        diagnostics['violations'] = []
+        diagnostics['unresolved_components'] = []
+        diagnostics['acceptance_failure_classes'] = []
+        diagnostics['resolved_event_count'] = len(ordered_events)
+        diagnostics['overlap_pair_count'] = overlap_pairs
+        diagnostics['max_overlap_depth'] = overlap_depth
+        diagnostics['exact_duplicate_intervals'] = exact_duplicates
+        diagnostics['adaptive_fine_repair'] = repair_payload
+        diagnostics['adaptive_fine_repair_sha256'] = _diagnostic_sha256(repair_payload)
+        diagnostics = _with_inference_context(diagnostics, inference_context)
+        return events_to_temporal_segments(ordered_events, diagnostics=diagnostics)
+
+    def _conservative_fine_completion(self, source: AnnotationSource, exc: FineSegmentationOutputError, *, inference_context: Mapping[str, object]) -> tuple[TemporalSegment, ...]:
+        result = exc.partial_result
+        violations = [dict(item) for item in exc.diagnostics.get('violations', []) if isinstance(item, dict)]
+        violation_codes = {str(item.get('code')) for item in violations}
+        if result is None or not result.unresolved or (not violation_codes) or (not violation_codes <= {'unresolved_components', 'stitch_ambiguity'}) or (source.duration_s is None):
+            raise exc
+        combined_events = [replace(event, metadata={'segmentation_provenance': 'fine', 'segmentation_resolution': 'fine', **dict(event.metadata)}) for event in result.events]
+        completion_records: list[dict[str, object]] = []
+        added_count = 0
+        suppressed_duplicate_count = 0
+        local_stitch_degraded_count = 0
+        local_stitch_candidates = _unresolved_local_stitch_candidates(exc.diagnostics)
+        for component_index, component in enumerate(result.unresolved):
+            local_candidate = local_stitch_candidates.get(component_index)
+            if local_candidate is not None:
+                target_event_id = str(local_candidate['resolved_event_id'])
+                target_matches = [index for index, event in enumerate(combined_events) if event.event_id == target_event_id]
+                if len(target_matches) != 1:
+                    raise SubtaskAnnotationAdapterError('unresolved-local degraded completion target is missing or ambiguous')
+                target_index = target_matches[0]
+                matched = combined_events[target_index]
+                strict_target_span = _local_stitch_target_span(local_candidate)
+                component_span, component_strategy = _conservative_component_completion_span(source, component)
+                completed_span = _covering_span(matched.span, component_span)
+                record = {'component_index': component_index, 'strategy': 'unconfirmed_unresolved_local_stitch_coverage_union_v2', 'observation_span': _span_payload(_component_observation_span(component)), 'component_completion_span': _span_payload(component_span), 'component_completion_strategy': component_strategy, 'strict_target_span': _span_payload(strict_target_span), 'pre_completion_target_span': _span_payload(matched.span), 'completed_span': _span_payload(completed_span), 'closed_start_ticks': list(component.closed_start_ticks), 'closed_end_ticks': list(component.closed_end_ticks), 'matching_ambiguous': component.matching_ambiguous, 'boundary_claim': 'observation_supported_union_of_strict_target_and_unresolved_component_v2', 'strict_target_geometry_expanded': completed_span != strict_target_span, 'candidate': dict(local_candidate), 'adaptive_fine_confirmation_failures': list(exc.diagnostics.get('adaptive_fine_repair_failures', [])), 'duplicate_suppressed': True}
+                metadata = dict(matched.metadata)
+                prior_evidence = metadata.get('unresolved_local_stitch_evidence', [])
+                evidence = list(prior_evidence) if isinstance(prior_evidence, list) else []
+                evidence.append(record)
+                metadata.update({'segmentation_provenance': 'fine_with_unconfirmed_unresolved_local_stitch', 'segmentation_resolution': 'fine_degraded', 'review_required': True, 'semantic_quality_established': False, 'unresolved_local_stitch_evidence': evidence})
+                combined_events[target_index] = replace(matched, span=completed_span, source_scope_ids=tuple(sorted({*matched.source_scope_ids, *component.source_scope_ids})), metadata=metadata)
+                suppressed_duplicate_count += 1
+                local_stitch_degraded_count += 1
+                completion_records.append(record)
+                continue
+            span, strategy = _conservative_component_completion_span(source, component)
+            record: dict[str, object] = {'component_index': component_index, 'strategy': strategy, 'observation_span': _span_payload(_component_observation_span(component)), 'completed_span': _span_payload(span), 'closed_start_ticks': list(component.closed_start_ticks), 'closed_end_ticks': list(component.closed_end_ticks), 'matching_ambiguous': component.matching_ambiguous, 'boundary_claim': 'observation_supported_conservative_extent_v1'}
+            exact_match_index = next((index for index, event in enumerate(combined_events) if event.span == span), None)
+            if exact_match_index is not None:
+                matched = combined_events[exact_match_index]
+                metadata = dict(matched.metadata)
+                prior_evidence = metadata.get('conservative_fine_completion_evidence', [])
+                evidence = list(prior_evidence) if isinstance(prior_evidence, list) else []
+                evidence.append(record)
+                metadata.update({'segmentation_provenance': 'fine_with_conservative_completion_evidence', 'segmentation_resolution': 'fine_degraded', 'review_required': True, 'semantic_quality_established': False, 'conservative_fine_completion_evidence': evidence})
+                combined_events[exact_match_index] = replace(matched, source_scope_ids=tuple(sorted({*matched.source_scope_ids, *component.source_scope_ids})), metadata=metadata)
+                record['duplicate_suppressed'] = True
+                suppressed_duplicate_count += 1
+            else:
+                record['duplicate_suppressed'] = False
+                combined_events.append(EventInterval(event_id=f'{source.source_id}:conservative-fine-completion:{component_index:06d}', source_id=source.source_id, span=span, source_scope_ids=tuple(sorted(component.source_scope_ids)), metadata={'segmentation_provenance': 'fine_conservative_completion', 'segmentation_resolution': 'fine_degraded', 'review_required': True, 'semantic_quality_established': False, 'boundary_claim': 'observation_supported_conservative_extent_v1', 'conservative_fine_completion': record}))
+                added_count += 1
+            completion_records.append(record)
+        ordered_events = tuple(sorted(combined_events, key=lambda item: (item.span, item.event_id)))
+        if not ordered_events:
+            raise exc
+        overlap_pairs, overlap_depth = _overlap_statistics(ordered_events)
+        exact_duplicates = _exact_duplicate_intervals(ordered_events)
+        if exact_duplicates:
+            raise SubtaskAnnotationAdapterError('conservative fine completion failed to suppress exact duplicate intervals')
+        completion_payload: dict[str, object] = {'schema_version': CONSERVATIVE_FINE_COMPLETION_SCHEMA, 'model_io_contract': fine_window_profile(self.fine_segmenter.window_profile_id).model_io_contract, 'window_profile_id': self.fine_segmenter.window_profile_id, 'terminal_contract': 'parsed_fine_evidence_only_no_coarse_provider_v1', 'boundary_claim': 'observation_supported_conservative_extent_v1', 'coarse_provider_called': False, 'additional_model_call_count': 0, 'max_retry_windows': self.adaptive_fine_retry_max_windows, 'fine_event_count_before': len(result.events), 'unresolved_component_count': len(result.unresolved), 'completion_event_count_added': added_count, 'completion_duplicate_suppressed': suppressed_duplicate_count, 'unresolved_local_stitch_candidate_count': len(local_stitch_candidates), 'unresolved_local_stitch_degraded_count': local_stitch_degraded_count, 'combined_event_count': len(ordered_events), 'post_completion_overlap_pair_count': overlap_pairs, 'post_completion_max_overlap_depth': overlap_depth, 'post_completion_exact_duplicate_intervals': exact_duplicates, 'components': completion_records}
+        diagnostics = dict(exc.diagnostics)
+        diagnostics['pre_completion_violations'] = violations
+        diagnostics['pre_completion_unresolved_components'] = list(diagnostics.get('unresolved_components', []))
+        diagnostics['violations'] = []
+        diagnostics['unresolved_components'] = []
+        diagnostics['acceptance_failure_classes'] = []
+        diagnostics['resolved_event_count'] = len(ordered_events)
+        diagnostics['overlap_pair_count'] = overlap_pairs
+        diagnostics['max_overlap_depth'] = overlap_depth
+        diagnostics['exact_duplicate_intervals'] = exact_duplicates
+        diagnostics['output_degraded'] = True
+        diagnostics['semantic_quality_established'] = False
+        diagnostics['conservative_fine_completion'] = completion_payload
+        diagnostics['conservative_fine_completion_sha256'] = _diagnostic_sha256(completion_payload)
+        diagnostics = _with_inference_context(diagnostics, inference_context)
+        return events_to_temporal_segments(ordered_events, diagnostics=diagnostics)
+
+    @staticmethod
+    def _bind_physical_source_envelope(source: AnnotationSource, segments: Sequence[TemporalSegment]) -> tuple[TemporalSegment, ...]:
+        if source.duration_s is None:
+            raise SubtaskAnnotationAdapterError('hierarchical fine segmentation requires exact source duration_s')
+        duration_us = seconds_to_us(source.duration_s, label='duration_s')
+        terminal_tick_us = ceil_div(duration_us, TICK_US) * TICK_US
+        bounded: list[TemporalSegment] = []
+        for segment in segments:
+            if segment.span.end_s <= source.duration_s + 1e-06:
+                bounded.append(segment)
+                continue
+            model_end_us = round(segment.span.end_s * 1000000)
+            if not math.isclose(segment.span.end_s, model_end_us / 1000000, abs_tol=1e-09) or model_end_us != terminal_tick_us or segment.span.start_s >= source.duration_s:
+                raise SubtaskAnnotationAdapterError(f'hierarchical segment for {source.source_id} ends at {segment.span.end_s}s beyond the physical source duration {source.duration_s}s and is not the terminal model tick')
+            metadata = dict(segment.metadata)
+            metadata['terminal_source_clip'] = {'contract': 'terminal_20hz_tick_to_physical_source_duration_v1', 'model_end_us': model_end_us, 'physical_end_us': duration_us}
+            bounded.append(TemporalSegment(TimeSpan(segment.span.start_s, source.duration_s), metadata))
+        ordered = tuple(sorted(bounded, key=lambda item: (item.span, repr(item.metadata))))
+        output: list[TemporalSegment] = []
+        for index, segment in enumerate(ordered):
+            before, after = _nearest_nonoverlapping_context(ordered, index)
+            metadata = dict(segment.metadata)
+            metadata['u2_context'] = {'before': _span_payload(before), 'target': _span_payload(segment.span), 'after': _span_payload(after)}
+            output.append(TemporalSegment(segment.span, metadata))
+        return tuple(output)
+
+    @staticmethod
+    def _with_refreshed_u2_context(segments: Sequence[TemporalSegment]) -> tuple[TemporalSegment, ...]:
+        ordered = tuple(sorted(segments, key=lambda item: (item.span, repr(item.metadata))))
+        output: list[TemporalSegment] = []
+        for index, segment in enumerate(ordered):
+            before, after = _nearest_nonoverlapping_context(ordered, index)
+            metadata = dict(segment.metadata)
+            metadata['u2_context'] = {'before': _span_payload(before), 'target': _span_payload(segment.span), 'after': _span_payload(after)}
+            output.append(TemporalSegment(segment.span, metadata))
+        return tuple(output)
+
+    def segment(self, source: AnnotationSource) -> tuple[TemporalSegment, ...]:
+        return asyncio.run(self.segment_async(source))
+
+    async def segment_async(self, source):
+        return self._bind_physical_source_envelope(source, await self._segment_impl(source))
+
+    async def _segment_impl(self, source):
+        inference_context = _inference_context(source, (), fine_prior_mode='none')
+        segment_with_priors_async = getattr(self.fine_segmenter, 'segment_with_priors_async', None)
+        try:
+            if callable(segment_with_priors_async):
+                result = await segment_with_priors_async(source, (), arm='no_prior_global_scan')
+            else:
+                result = await asyncio.to_thread(self.fine_segmenter.segment_with_priors, source, (), arm='no_prior_global_scan')
+        except FineSegmentationOutputError as exc:
+            contextual_error = FineSegmentationOutputError(str(exc), diagnostics=_with_inference_context(exc.diagnostics, inference_context), partial_result=exc.partial_result)
+            try:
+                return await self.adaptive_fine_repair_async(source, contextual_error, inference_context=inference_context)
+            except FineSegmentationOutputError as adaptive_error:
+                return self._conservative_fine_completion(source, adaptive_error, inference_context=inference_context)
+        diagnostics = _with_inference_context(result.diagnostics, inference_context)
+        return events_to_temporal_segments(result.events, diagnostics=diagnostics)

batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/fine_temporal.py

@@ -0,0 +1,987 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+from collections.abc import Collection, Iterable, Mapping, Sequence
+from dataclasses import dataclass, field
+from fractions import Fraction
+import json
+import math
+from flora_runtime.schema import TimeSpan
+from flora_runtime.video_segmentation.contracts import EventInterval, ProcessingScope
+TICK_US = 50000
+CORE_US = 6000000
+LEFT_GUARD_US = 1000000
+RIGHT_GUARD_US = 1000000
+SLOT_COUNT = 64
+SLOT_FPS = 8
+LEGACY_FINE_WINDOW_PROFILE_ID = 'atomic_1plus6plus1_64slots_8hz'
+HUMAN_FINE_WINDOW_PROFILE_ID = 'human_16s_4hz'
+YAM_ATOMIC_LONG_FINE_WINDOW_PROFILE_ID = 'yam_atomic_2plus12plus2_64slots_4hz'
+YAM_INTENT_32S_FINE_WINDOW_PROFILE_ID = 'yam_intent_4plus24plus4_128slots_4hz'
+DEFAULT_FINE_WINDOW_PROFILE_ID = LEGACY_FINE_WINDOW_PROFILE_ID
+MAX_SEGMENTS = 32
+STITCH_TOLERANCE_TICKS = 2
+MAX_BOUNDED_ANCHOR_TICKS = 4
+STRICT_TRACE_STITCHING_CONTRACT = 'strict_shared_trace_endpoint_v1'
+SHARED_TRACE_IOU_STITCHING_CONTRACT = 'shared_trace_iou_endpoint_v1'
+COMPONENT_LOCAL_IDENTITY_STITCHING_CONTRACT = 'component_local_identity_anchor_v1'
+COMPONENT_LOCAL_BOUNDED_ANCHOR_STITCHING_CONTRACT = 'component_local_bounded_anchor_certificate_v1'
+COMPONENT_LOCAL_STITCHING_CONTRACTS = frozenset({COMPONENT_LOCAL_IDENTITY_STITCHING_CONTRACT, COMPONENT_LOCAL_BOUNDED_ANCHOR_STITCHING_CONTRACT})
+STITCHING_CONTRACTS = {STRICT_TRACE_STITCHING_CONTRACT, SHARED_TRACE_IOU_STITCHING_CONTRACT, *COMPONENT_LOCAL_STITCHING_CONTRACTS}
+STITCHING_SEAM_DIAGNOSTICS_SCHEMA = 'data_engine_fine_stitching_seam_diagnostics_v1'
+WINDOW_SCHEMA = 'data_engine_atomic_fine_window_v1'
+MODEL_IO_CONTRACT = 'flora_inference_atomic_temporal_segmentation_v1'
+HUMAN_MODEL_IO_CONTRACT = 'flora_human_temporal_segmentation_v1'
+YAM_ATOMIC_LONG_MODEL_IO_CONTRACT = 'data_engine_yam_atomic_temporal_segmentation_16s4hz_v1'
+YAM_INTENT_32S_MODEL_IO_CONTRACT = 'data_engine_yam_intent_temporal_segmentation_32s4hz_v1'
+ComponentPairKey = tuple[int, int, int]
+
+class FineTemporalError(ValueError):
+    pass
+
+class FinePredictionParseError(FineTemporalError):
+    pass
+
+@dataclass(frozen=True)
+class FineWindowProfile:
+    profile_id: str
+    left_guard_us: int
+    core_us: int
+    right_guard_us: int
+    slot_count: int
+    slot_fps: int
+    model_io_contract: str
+    target_granularity: str
+
+    def __post_init__(self) -> None:
+        if not self.profile_id or not self.model_io_contract:
+            raise FineTemporalError('fine window profile identity and model contract are required')
+        if self.target_granularity not in {'atomic_event', 'coarse_subtask_run'}:
+            raise FineTemporalError('fine window target granularity is unsupported')
+        values = (self.left_guard_us, self.core_us, self.right_guard_us, self.slot_count, self.slot_fps)
+        if any((isinstance(value, bool) or not isinstance(value, int) or value <= 0 for value in values)):
+            raise FineTemporalError('fine window geometry and sampling values must be positive integers')
+        if self.view_us % TICK_US or self.core_us % TICK_US:
+            raise FineTemporalError('fine window geometry must align to the 20 Hz model clock')
+        if self.left_guard_us != self.right_guard_us:
+            raise FineTemporalError('fine window profiles currently require symmetric guards')
+        if 1000000 % self.slot_fps:
+            raise FineTemporalError('fine window sample rate must divide one second exactly')
+        if self.view_us * self.slot_fps != self.slot_count * 1000000:
+            raise FineTemporalError('fine window slots, duration, and sample rate disagree')
+
+    @property
+    def view_us(self) -> int:
+        return self.left_guard_us + self.core_us + self.right_guard_us
+
+    @property
+    def view_ticks(self) -> int:
+        return self.view_us // TICK_US
+
+    @property
+    def core_ticks(self) -> int:
+        return self.core_us // TICK_US
+
+    @property
+    def slot_spacing_us(self) -> int:
+        return 1000000 // self.slot_fps
+
+    @property
+    def scheduler_contract(self) -> str:
+        core = self._duration_token(self.core_us)
+        guard = self._duration_token(self.left_guard_us)
+        return f'fixed_{core}_core_{guard}_guards_global_grid_v1'
+
+    def retry_scheduler_contract(self, *, directional: bool) -> str:
+        prefix = 'adaptive_directional' if directional else 'adaptive_recentered'
+        core = self._duration_token(self.core_us)
+        guard = self._duration_token(self.left_guard_us)
+        geometry = f'{core}_core_{guard}_guards'
+        return f'{prefix}_{geometry}_v1'
+
+    @staticmethod
+    def _duration_token(value_us: int) -> str:
+        return f'{value_us // 1000000}s' if value_us % 1000000 == 0 else f'{value_us}us'
+
+    def to_dict(self) -> dict[str, int | str]:
+        return {'profile_id': self.profile_id, 'left_guard_us': self.left_guard_us, 'core_us': self.core_us, 'right_guard_us': self.right_guard_us, 'view_us': self.view_us, 'slot_count': self.slot_count, 'slot_fps': self.slot_fps, 'model_io_contract': self.model_io_contract, 'target_granularity': self.target_granularity}
+FINE_WINDOW_PROFILES: Mapping[str, FineWindowProfile] = {LEGACY_FINE_WINDOW_PROFILE_ID: FineWindowProfile(profile_id=LEGACY_FINE_WINDOW_PROFILE_ID, left_guard_us=LEFT_GUARD_US, core_us=CORE_US, right_guard_us=RIGHT_GUARD_US, slot_count=SLOT_COUNT, slot_fps=SLOT_FPS, model_io_contract=MODEL_IO_CONTRACT, target_granularity='atomic_event'), HUMAN_FINE_WINDOW_PROFILE_ID: FineWindowProfile(profile_id=HUMAN_FINE_WINDOW_PROFILE_ID, left_guard_us=2000000, core_us=12000000, right_guard_us=2000000, slot_count=64, slot_fps=4, model_io_contract=HUMAN_MODEL_IO_CONTRACT, target_granularity='coarse_subtask_run'), YAM_ATOMIC_LONG_FINE_WINDOW_PROFILE_ID: FineWindowProfile(profile_id=YAM_ATOMIC_LONG_FINE_WINDOW_PROFILE_ID, left_guard_us=2000000, core_us=12000000, right_guard_us=2000000, slot_count=64, slot_fps=4, model_io_contract=YAM_ATOMIC_LONG_MODEL_IO_CONTRACT, target_granularity='atomic_event'), YAM_INTENT_32S_FINE_WINDOW_PROFILE_ID: FineWindowProfile(profile_id=YAM_INTENT_32S_FINE_WINDOW_PROFILE_ID, left_guard_us=4000000, core_us=24000000, right_guard_us=4000000, slot_count=128, slot_fps=4, model_io_contract=YAM_INTENT_32S_MODEL_IO_CONTRACT, target_granularity='atomic_event')}
+
+def fine_window_profile(profile_id: str=DEFAULT_FINE_WINDOW_PROFILE_ID) -> FineWindowProfile:
+    try:
+        return FINE_WINDOW_PROFILES[profile_id]
+    except (KeyError, TypeError) as exc:
+        raise FineTemporalError(f'unknown fine window profile: {profile_id!r}') from exc
+
+@dataclass(frozen=True)
+class StitchingPolicy:
+    contract: str = STRICT_TRACE_STITCHING_CONTRACT
+    endpoint_tolerance_ticks: int = STITCH_TOLERANCE_TICKS
+    min_shared_trace_iou_numerator: int = 1
+    min_shared_trace_iou_denominator: int = 2
+
+    def __post_init__(self) -> None:
+        if self.contract not in STITCHING_CONTRACTS:
+            raise FineTemporalError(f'unknown stitching contract: {self.contract}')
+        integer_fields = ('endpoint_tolerance_ticks', 'min_shared_trace_iou_numerator', 'min_shared_trace_iou_denominator')
+        for field_name in integer_fields:
+            value = getattr(self, field_name)
+            if isinstance(value, bool) or not isinstance(value, int):
+                raise FineTemporalError(f'{field_name} must be an integer')
+        if self.endpoint_tolerance_ticks < 0:
+            raise FineTemporalError('endpoint_tolerance_ticks must be non-negative')
+        if self.contract == COMPONENT_LOCAL_BOUNDED_ANCHOR_STITCHING_CONTRACT and (not STITCH_TOLERANCE_TICKS < self.endpoint_tolerance_ticks <= MAX_BOUNDED_ANCHOR_TICKS):
+            raise FineTemporalError('bounded-anchor tolerance must be above strict and at most four ticks')
+        if not 0 < self.min_shared_trace_iou_numerator <= self.min_shared_trace_iou_denominator:
+            raise FineTemporalError('shared-trace IoU threshold must lie in (0,1]')
+
+    @classmethod
+    def strict(cls, tolerance_ticks: int=STITCH_TOLERANCE_TICKS) -> StitchingPolicy:
+        return cls(contract=STRICT_TRACE_STITCHING_CONTRACT, endpoint_tolerance_ticks=tolerance_ticks)
+
+    @classmethod
+    def shared_trace_iou(cls, *, endpoint_tolerance_ticks: int, numerator: int=1, denominator: int=2) -> StitchingPolicy:
+        return cls(contract=SHARED_TRACE_IOU_STITCHING_CONTRACT, endpoint_tolerance_ticks=endpoint_tolerance_ticks, min_shared_trace_iou_numerator=numerator, min_shared_trace_iou_denominator=denominator)
+
+    @classmethod
+    def component_local_identity(cls, *, endpoint_tolerance_ticks: int=4, numerator: int=3, denominator: int=4) -> StitchingPolicy:
+        if endpoint_tolerance_ticks <= STITCH_TOLERANCE_TICKS:
+            raise FineTemporalError('component-local candidate tolerance must exceed the strict tolerance')
+        return cls(contract=COMPONENT_LOCAL_IDENTITY_STITCHING_CONTRACT, endpoint_tolerance_ticks=endpoint_tolerance_ticks, min_shared_trace_iou_numerator=numerator, min_shared_trace_iou_denominator=denominator)
+
+    @classmethod
+    def component_local_bounded_anchor(cls, *, endpoint_tolerance_ticks: int=4, numerator: int=3, denominator: int=4) -> StitchingPolicy:
+        if not STITCH_TOLERANCE_TICKS < endpoint_tolerance_ticks <= MAX_BOUNDED_ANCHOR_TICKS:
+            raise FineTemporalError('bounded-anchor candidate tolerance must be above strict and at most four ticks')
+        return cls(contract=COMPONENT_LOCAL_BOUNDED_ANCHOR_STITCHING_CONTRACT, endpoint_tolerance_ticks=endpoint_tolerance_ticks, min_shared_trace_iou_numerator=numerator, min_shared_trace_iou_denominator=denominator)
+
+    def to_dict(self) -> dict[str, int | str]:
+        return {'contract': self.contract, 'endpoint_tolerance_ticks': self.endpoint_tolerance_ticks, 'min_shared_trace_iou_numerator': self.min_shared_trace_iou_numerator, 'min_shared_trace_iou_denominator': self.min_shared_trace_iou_denominator}
+
+def floor_div(value: int, divisor: int) -> int:
+    if divisor <= 0:
+        raise FineTemporalError('divisor must be positive')
+    return value // divisor
+
+def ceil_div(value: int, divisor: int) -> int:
+    if divisor <= 0:
+        raise FineTemporalError('divisor must be positive')
+    return -(-value // divisor)
+
+def round_half_up_div(value: int, divisor: int) -> int:
+    if divisor <= 0 or value < 0:
+        raise FineTemporalError('round_half_up_div requires non-negative input and positive divisor')
+    quotient, remainder = divmod(value, divisor)
+    return quotient + int(remainder * 2 >= divisor)
+
+@dataclass(frozen=True)
+class FineWindow:
+    source_id: str
+    window_index: int
+    video_end_us: int
+    requested_view_start_us: int
+    requested_view_end_us: int
+    valid_start_us: int
+    valid_end_us: int
+    core_start_us: int
+    core_end_us: int
+    valid_start_tick: int
+    valid_end_tick: int
+    core_start_tick: int
+    core_end_tick: int
+    scope_id: str
+    window_profile_id: str = DEFAULT_FINE_WINDOW_PROFILE_ID
+    coarse_spans: tuple[Mapping[str, object], ...] = ()
+
+    def __post_init__(self) -> None:
+        if not self.source_id or not self.scope_id or self.window_index < 0 or (self.video_end_us <= 0):
+            raise FineTemporalError('window identity, index, and duration are required')
+        profile = fine_window_profile(self.window_profile_id)
+        if self.requested_view_end_us - self.requested_view_start_us != profile.view_us:
+            raise FineTemporalError(f'requested view must match profile {profile.profile_id!r}')
+        if not self.requested_view_start_us <= self.valid_start_us < self.valid_end_us <= self.requested_view_end_us:
+            raise FineTemporalError('invalid visible-video envelope')
+        if not 0 <= self.valid_start_us < self.valid_end_us <= self.video_end_us:
+            raise FineTemporalError('valid envelope must lie inside the video')
+        if not self.valid_start_us <= self.core_start_us < self.core_end_us <= self.valid_end_us:
+            raise FineTemporalError('emit core must lie inside the valid envelope')
+        if not 0 <= self.valid_start_tick < self.valid_end_tick <= profile.view_ticks:
+            raise FineTemporalError('invalid visible tick envelope')
+        if not self.valid_start_tick <= self.core_start_tick < self.core_end_tick <= self.valid_end_tick:
+            raise FineTemporalError('invalid core tick envelope')
+        coarse_spans = tuple((dict(item) for item in self.coarse_spans))
+        if len(coarse_spans) > 64:
+            raise FineTemporalError('one view may contain at most 64 coarse spans')
+        previous: tuple[int, int, bool, bool] | None = None
+        for index, row in enumerate(coarse_spans):
+            if set(row) != {'s', 'e', 'lo', 'ro'}:
+                raise FineTemporalError(f'coarse span {index} must contain exactly s,e,lo,ro')
+            fragment = FineFragment(row['s'], row['e'], row['lo'], row['ro'])
+            if isinstance(fragment.s, bool) or isinstance(fragment.e, bool):
+                raise FineTemporalError(f'coarse span {index} ticks must be integers')
+            if not isinstance(fragment.s, int) or not isinstance(fragment.e, int):
+                raise FineTemporalError(f'coarse span {index} ticks must be integers')
+            if type(fragment.lo) is not bool or type(fragment.ro) is not bool:
+                raise FineTemporalError(f'coarse span {index} open flags must be booleans')
+            if not self.valid_start_tick <= fragment.s < fragment.e <= self.valid_end_tick:
+                raise FineTemporalError(f'coarse span {index} lies outside the valid envelope')
+            if fragment.lo and fragment.s != self.valid_start_tick:
+                raise FineTemporalError(f'coarse span {index} lo=true is not clipped to valid start')
+            if fragment.ro and fragment.e != self.valid_end_tick:
+                raise FineTemporalError(f'coarse span {index} ro=true is not clipped to valid end')
+            if previous is not None and fragment.sort_key <= previous:
+                raise FineTemporalError('coarse spans must be unique and in canonical order')
+            previous = fragment.sort_key
+        object.__setattr__(self, 'coarse_spans', coarse_spans)
+
+    @property
+    def global_tick_offset(self) -> int:
+        if self.requested_view_start_us % TICK_US:
+            raise FineTemporalError('requested view is not aligned to the 20 Hz model clock')
+        return self.requested_view_start_us // TICK_US
+
+    @property
+    def window_profile(self) -> FineWindowProfile:
+        return fine_window_profile(self.window_profile_id)
+
+    def worker_payload(self) -> dict[str, object]:
+        return {'schema_version': WINDOW_SCHEMA, 'model_io_contract': self.window_profile.model_io_contract, 'window_profile_id': self.window_profile_id, 'scope_id': self.scope_id, 'window_index': self.window_index, 'requested_view_start_us': self.requested_view_start_us, 'requested_view_end_us': self.requested_view_end_us, 'valid_start_tick': self.valid_start_tick, 'valid_end_tick': self.valid_end_tick, 'core_start_tick': self.core_start_tick, 'core_end_tick': self.core_end_tick, 'coarse_spans': [dict(item) for item in self.coarse_spans]}
+
+@dataclass(frozen=True)
+class FineFragment:
+    s: int
+    e: int
+    lo: bool
+    ro: bool
+
+    @property
+    def sort_key(self) -> tuple[int, int, bool, bool]:
+        return (self.s, self.e, self.lo, self.ro)
+
+def _reject_constant(value: str) -> None:
+    raise FinePredictionParseError(f'non-finite JSON constant is forbidden: {value}')
+
+def _object_without_duplicate_keys(pairs: list[tuple[str, object]]) -> dict[str, object]:
+    result: dict[str, object] = {}
+    for key, value in pairs:
+        if key in result:
+            raise FinePredictionParseError(f'duplicate JSON key is forbidden: {key}')
+        result[key] = value
+    return result
+
+def _validate_fragment_envelope(fragment: FineFragment, window: FineWindow) -> None:
+    if isinstance(fragment.s, bool) or isinstance(fragment.e, bool):
+        raise FinePredictionParseError('segment ticks must be integers, not booleans')
+    if not isinstance(fragment.s, int) or not isinstance(fragment.e, int):
+        raise FinePredictionParseError('segment ticks must be integers')
+    if type(fragment.lo) is not bool or type(fragment.ro) is not bool:
+        raise FinePredictionParseError('open flags must be JSON booleans')
+    if not window.valid_start_tick <= fragment.s < fragment.e <= window.valid_end_tick <= window.window_profile.view_ticks:
+        raise FinePredictionParseError('segment lies outside the valid tick envelope')
+    if fragment.lo and fragment.s != window.valid_start_tick:
+        raise FinePredictionParseError('lo=true requires clipping to valid_start_tick')
+    if fragment.ro and fragment.e != window.valid_end_tick:
+        raise FinePredictionParseError('ro=true requires clipping to valid_end_tick')
+
+def _has_emit_core_support(fragment: FineFragment, window: FineWindow) -> bool:
+    core_intersects = fragment.s < window.core_end_tick and fragment.e > window.core_start_tick
+    core_touches = fragment.e == window.core_start_tick or fragment.s == window.core_end_tick
+    return core_intersects or core_touches
+
+def _validate_fragment(fragment: FineFragment, window: FineWindow) -> None:
+    _validate_fragment_envelope(fragment, window)
+    if not _has_emit_core_support(fragment, window):
+        raise FinePredictionParseError('segment has no emit-core support')
+
+@dataclass(frozen=True)
+class ParsedFinePrediction:
+    fragments: tuple[FineFragment, ...]
+    guard_only_fragments: tuple[FineFragment, ...]
+
+def parse_prediction_with_diagnostics(payload: str, window: FineWindow, *, drop_guard_only: bool) -> ParsedFinePrediction:
+    if not isinstance(payload, str) or not payload:
+        raise FinePredictionParseError('prediction_json must be a non-empty string')
+    try:
+        value = json.loads(payload, object_pairs_hook=_object_without_duplicate_keys, parse_constant=_reject_constant)
+    except FinePredictionParseError:
+        raise
+    except (json.JSONDecodeError, TypeError, ValueError) as exc:
+        raise FinePredictionParseError('prediction is not one complete JSON object') from exc
+    if not isinstance(value, dict) or list(value) != ['segments']:
+        raise FinePredictionParseError('prediction must have exactly one top-level key: segments')
+    rows = value['segments']
+    if not isinstance(rows, list):
+        raise FinePredictionParseError('segments must be a list')
+    if len(rows) > MAX_SEGMENTS:
+        raise FinePredictionParseError(f'prediction contains more than {MAX_SEGMENTS} segments')
+    result: list[FineFragment] = []
+    guard_only: list[FineFragment] = []
+    previous: tuple[int, int, bool, bool] | None = None
+    for index, row in enumerate(rows):
+        if not isinstance(row, dict) or list(row) != ['s', 'e', 'lo', 'ro']:
+            raise FinePredictionParseError(f'segment {index}: keys must appear exactly as s,e,lo,ro')
+        fragment = FineFragment(row['s'], row['e'], row['lo'], row['ro'])
+        try:
+            _validate_fragment_envelope(fragment, window)
+        except FinePredictionParseError as exc:
+            raise FinePredictionParseError(f'segment {index}: {exc}') from exc
+        if previous is not None and fragment.sort_key < previous:
+            raise FinePredictionParseError('segments are not in canonical order')
+        previous = fragment.sort_key
+        if not _has_emit_core_support(fragment, window):
+            if not drop_guard_only:
+                raise FinePredictionParseError(f'segment {index}: segment has no emit-core support')
+            guard_only.append(fragment)
+            continue
+        result.append(fragment)
+    return ParsedFinePrediction(tuple(result), tuple(guard_only))
+
+@dataclass(frozen=True)
+class _Observation:
+    observation_id: int
+    window_index: int
+    row_ordinal: int
+    scope_id: str
+    start_tick: int
+    end_tick: int
+    lo: bool
+    ro: bool
+    valid_start_tick: int
+    valid_end_tick: int
+
+    @property
+    def has_closed_start(self) -> bool:
+        return not self.lo
+
+    @property
+    def has_closed_end(self) -> bool:
+        return not self.ro
+
+@dataclass(frozen=True)
+class UnresolvedFineEvent:
+    observation_ids: tuple[int, ...]
+    source_scope_ids: tuple[str, ...]
+    observation_spans_ticks: tuple[tuple[int, int], ...]
+    closed_start_ticks: tuple[int, ...]
+    closed_end_ticks: tuple[int, ...]
+    support_start_tick: int
+    support_end_tick: int
+    matching_ambiguous: bool
+
+@dataclass(frozen=True)
+class FineSegmentationResult:
+    events: tuple[EventInterval, ...]
+    fragments: tuple[EventInterval, ...]
+    unresolved: tuple[UnresolvedFineEvent, ...]
+    adjacent_matching_ambiguities: int
+    diagnostics: Mapping[str, object] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, 'diagnostics', dict(self.diagnostics))
+
+def _observations(window: FineWindow, fragments: Sequence[FineFragment], *, first_id: int) -> tuple[_Observation, ...]:
+    result: list[_Observation] = []
+    for ordinal, fragment in enumerate(fragments):
+        _validate_fragment(fragment, window)
+        offset = window.global_tick_offset
+        result.append(_Observation(observation_id=first_id + ordinal, window_index=window.window_index, row_ordinal=ordinal, scope_id=window.scope_id, start_tick=offset + fragment.s, end_tick=offset + fragment.e, lo=fragment.lo, ro=fragment.ro, valid_start_tick=offset + window.valid_start_tick, valid_end_tick=offset + window.valid_end_tick))
+    return tuple(result)
+
+def _shared_trace(item: _Observation, start: int, end: int) -> tuple[int, int] | None:
+    trace = (max(item.start_tick, start), min(item.end_tick, end))
+    return trace if trace[0] < trace[1] else None
+
+@dataclass(frozen=True)
+class _CandidateEvaluation:
+    cost: tuple[int, int] | None
+    rejection: str | None
+
+def _candidate_evaluation(left: _Observation, right: _Observation, overlap_start: int, overlap_end: int, policy: StitchingPolicy) -> _CandidateEvaluation:
+    left_trace = _shared_trace(left, overlap_start, overlap_end)
+    right_trace = _shared_trace(right, overlap_start, overlap_end)
+    if left_trace is None or right_trace is None:
+        return _CandidateEvaluation(None, 'no_shared_trace')
+    deltas = (abs(left_trace[0] - right_trace[0]), abs(left_trace[1] - right_trace[1]))
+    if policy.contract in {STRICT_TRACE_STITCHING_CONTRACT, *COMPONENT_LOCAL_STITCHING_CONTRACTS}:
+        if max(deltas) > policy.endpoint_tolerance_ticks:
+            return _CandidateEvaluation(None, 'shared_trace_endpoint_delta')
+        trace_cost = sum(deltas)
+    else:
+        intersection = max(0, min(left_trace[1], right_trace[1]) - max(left_trace[0], right_trace[0]))
+        union = max(left_trace[1], right_trace[1]) - min(left_trace[0], right_trace[0])
+        if union <= 0 or intersection * policy.min_shared_trace_iou_denominator < union * policy.min_shared_trace_iou_numerator:
+            return _CandidateEvaluation(None, 'shared_trace_iou')
+        trace_cost = (union - intersection) * 1000000 // union
+    closed_cost = 0
+    if left.has_closed_start and right.has_closed_start:
+        delta = abs(left.start_tick - right.start_tick)
+        if delta > policy.endpoint_tolerance_ticks:
+            return _CandidateEvaluation(None, 'closed_start_delta')
+        closed_cost += delta
+    elif left.has_closed_start and left.start_tick > right.valid_start_tick + policy.endpoint_tolerance_ticks or (right.has_closed_start and right.start_tick > left.valid_start_tick + policy.endpoint_tolerance_ticks):
+        return _CandidateEvaluation(None, 'closed_start_visible_to_censored_peer')
+    if left.has_closed_end and right.has_closed_end:
+        delta = abs(left.end_tick - right.end_tick)
+        if delta > policy.endpoint_tolerance_ticks:
+            return _CandidateEvaluation(None, 'closed_end_delta')
+        closed_cost += delta
+    elif left.has_closed_end and left.end_tick < right.valid_end_tick - policy.endpoint_tolerance_ticks or (right.has_closed_end and right.end_tick < left.valid_end_tick - policy.endpoint_tolerance_ticks):
+        return _CandidateEvaluation(None, 'closed_end_visible_to_censored_peer')
+    return _CandidateEvaluation((trace_cost, closed_cost), None)
+
+@dataclass(frozen=True)
+class _MatchState:
+    count: int = 0
+    trace_cost: int = 0
+    closed_cost: int = 0
+    pairings: tuple[tuple[tuple[int, int], ...], ...] = ((),)
+
+    @property
+    def key(self) -> tuple[int, int, int]:
+        return (-self.count, self.trace_cost, self.closed_cost)
+
+def _choose(states: Sequence[_MatchState]) -> _MatchState:
+    best_key = min((state.key for state in states))
+    optimal = [state for state in states if state.key == best_key]
+    pairings = tuple(sorted({pairing for state in optimal for pairing in state.pairings})[:2])
+    return _MatchState(-best_key[0], best_key[1], best_key[2], pairings)
+
+def _match_adjacent(left: Sequence[_Observation], right: Sequence[_Observation], left_window: FineWindow, right_window: FineWindow, policy: StitchingPolicy) -> tuple[tuple[tuple[int, int], ...], bool, tuple[tuple[tuple[int, int], ...], ...]]:
+    if right_window.window_index != left_window.window_index + 1:
+        raise FineTemporalError('stitching requires the complete ordered global window grid')
+    overlap_start = max(left_window.global_tick_offset + left_window.valid_start_tick, right_window.global_tick_offset + right_window.valid_start_tick)
+    overlap_end = min(left_window.global_tick_offset + left_window.valid_end_tick, right_window.global_tick_offset + right_window.valid_end_tick)
+    if overlap_start >= overlap_end:
+        raise FineTemporalError('adjacent windows have no shared valid-visible guard')
+    n, m = (len(left), len(right))
+    dp = [[_MatchState() for _ in range(m + 1)] for _ in range(n + 1)]
+    for i in range(1, n + 1):
+        dp[i][0] = dp[i - 1][0]
+    for j in range(1, m + 1):
+        dp[0][j] = dp[0][j - 1]
+    for i in range(1, n + 1):
+        for j in range(1, m + 1):
+            candidates = [dp[i - 1][j], dp[i][j - 1]]
+            evaluation = _candidate_evaluation(left[i - 1], right[j - 1], overlap_start, overlap_end, policy)
+            if evaluation.cost is not None:
+                prior = dp[i - 1][j - 1]
+                pair = (left[i - 1].observation_id, right[j - 1].observation_id)
+                candidates.append(_MatchState(prior.count + 1, prior.trace_cost + evaluation.cost[0], prior.closed_cost + evaluation.cost[1], tuple(((*pairing, pair) for pairing in prior.pairings))))
+            dp[i][j] = _choose(candidates)
+    result = dp[n][m]
+    return (result.pairings[0], len(result.pairings) > 1, result.pairings)
+
+class _UnionFind:
+
+    def __init__(self, values: Iterable[int]) -> None:
+        self.parent = {value: value for value in values}
+
+    def find(self, value: int) -> int:
+        parent = self.parent[value]
+        if parent != value:
+            self.parent[value] = self.find(parent)
+        return self.parent[value]
+
+    def union(self, left: int, right: int) -> None:
+        left_root, right_root = (self.find(left), self.find(right))
+        if left_root != right_root:
+            self.parent[max(left_root, right_root)] = min(left_root, right_root)
+
+def _median(values: Sequence[int]) -> int:
+    ordered = sorted(values)
+    if not ordered:
+        raise FineTemporalError('cannot form endpoint consensus from no values')
+    middle = len(ordered) // 2
+    if len(ordered) % 2:
+        return ordered[middle]
+    value = Fraction(ordered[middle - 1] + ordered[middle], 2)
+    return math.floor(value + Fraction(1, 2))
+
+@dataclass(frozen=True)
+class _ComponentState:
+    root: int
+    observation_ids: tuple[int, ...]
+    starts: tuple[int, ...]
+    ends: tuple[int, ...]
+
+    @property
+    def resolved(self) -> bool:
+        return bool(self.starts) and bool(self.ends) and (self.starts[-1] - self.starts[0] <= STITCH_TOLERANCE_TICKS) and (self.ends[-1] - self.ends[0] <= STITCH_TOLERANCE_TICKS) and (_median(self.ends) > _median(self.starts))
+
+    @property
+    def partial_endpoint(self) -> tuple[str, tuple[int, ...]] | None:
+        if bool(self.starts) == bool(self.ends):
+            return None
+        name, values = ('start', self.starts) if self.starts else ('end', self.ends)
+        if values[-1] - values[0] > STITCH_TOLERANCE_TICKS:
+            return None
+        return (name, values)
+
+@dataclass(frozen=True)
+class _LocalCandidate:
+    seam_index: int
+    left_observation_id: int
+    right_observation_id: int
+    resolved_root: int
+    partial_root: int
+    anchor_endpoint: str
+    anchor_tick: int
+    anchor_delta_ticks: int
+    trace_intersection_ticks: int
+    trace_union_ticks: int
+
+    @property
+    def pair(self) -> tuple[int, int]:
+        return (self.left_observation_id, self.right_observation_id)
+
+    def to_dict(self) -> dict[str, object]:
+        result = {'left_observation_id': self.left_observation_id, 'right_observation_id': self.right_observation_id, 'resolved_component_root': self.resolved_root, 'partial_component_root': self.partial_root, 'anchor_endpoint': self.anchor_endpoint, 'anchor_tick': self.anchor_tick, 'trace_intersection_ticks': self.trace_intersection_ticks, 'trace_union_ticks': self.trace_union_ticks}
+        if self.anchor_delta_ticks:
+            result['anchor_delta_ticks'] = self.anchor_delta_ticks
+        return result
+
+@dataclass(frozen=True)
+class _AssociationPlan:
+    base_pairs_by_seam: tuple[tuple[tuple[int, int], ...], ...]
+    pairs_by_seam: tuple[tuple[tuple[int, int], ...], ...]
+    ambiguous_by_seam: tuple[bool, ...]
+    competing_by_seam: tuple[tuple[tuple[tuple[int, int], ...], ...], ...]
+    component_consistency_tolerance_ticks: int
+    component_local_by_seam: tuple[Mapping[str, object], ...]
+    component_local_summary: Mapping[str, object] | None
+    component_local_geometry_by_seam: tuple[Mapping[tuple[int, int], tuple[int, int]], ...]
+
+def _component_states(observations: Sequence[_Observation], union_find: _UnionFind) -> dict[int, _ComponentState]:
+    grouped: dict[int, list[_Observation]] = {}
+    for observation in observations:
+        grouped.setdefault(union_find.find(observation.observation_id), []).append(observation)
+    return {root: _ComponentState(root=root, observation_ids=tuple(sorted((observation.observation_id for observation in component))), starts=tuple(sorted((observation.start_tick for observation in component if not observation.lo))), ends=tuple(sorted((observation.end_tick for observation in component if not observation.ro)))) for root, component in grouped.items()}
+
+def _resolved_component_geometry(states: Mapping[int, _ComponentState]) -> tuple[tuple[int, int], ...]:
+    return tuple(sorted(((_median(state.starts), _median(state.ends)) for state in states.values() if state.resolved)))
+
+def _trace_iou_counts(left: _Observation, right: _Observation, *, overlap_start: int, overlap_end: int) -> tuple[int, int]:
+    left_trace = _shared_trace(left, overlap_start, overlap_end)
+    right_trace = _shared_trace(right, overlap_start, overlap_end)
+    if left_trace is None or right_trace is None:
+        return (0, 0)
+    intersection = max(0, min(left_trace[1], right_trace[1]) - max(left_trace[0], right_trace[0]))
+    union = max(left_trace[1], right_trace[1]) - min(left_trace[0], right_trace[0])
+    return (intersection, union)
+
+def _identity_anchor(left: _ComponentState, right: _ComponentState, *, maximum_delta_ticks: int) -> tuple[tuple[int, int, str, int, int] | None, str | None]:
+    if left.resolved == right.resolved:
+        return (None, 'requires_one_resolved_one_partial')
+    resolved, partial = (left, right) if left.resolved else (right, left)
+    partial_endpoint = partial.partial_endpoint
+    if partial_endpoint is None:
+        return (None, 'partial_component_is_not_one_sided')
+    endpoint_name, values = partial_endpoint
+    anchor_values = resolved.starts if endpoint_name == 'start' else resolved.ends
+    anchor_tick = _median(anchor_values)
+    maximum_delta = max((abs(value - anchor_tick) for value in values))
+    if maximum_delta > maximum_delta_ticks:
+        return (None, 'partial_endpoint_is_not_exactly_anchored' if maximum_delta_ticks == 0 else 'partial_endpoint_exceeds_anchor_tolerance')
+    if maximum_delta == 0 and _median((*anchor_values, *values)) != anchor_tick:
+        return (None, 'partial_endpoint_would_move_resolved_geometry')
+    return ((resolved.root, partial.root, endpoint_name, anchor_tick, maximum_delta), None)
+
+def _monotone_with_pairs(pair: tuple[int, int], *, base_pairs: Sequence[tuple[int, int]], left: Sequence[_Observation], right: Sequence[_Observation]) -> bool:
+    left_position = {item.observation_id: index for index, item in enumerate(left)}
+    right_position = {item.observation_id: index for index, item in enumerate(right)}
+    left_index, right_index = (left_position[pair[0]], right_position[pair[1]])
+    return all((not (left_index < left_position[base_left] and right_index > right_position[base_right] or (left_index > left_position[base_left] and right_index < right_position[base_right])) for base_left, base_right in base_pairs))
+
+def _component_local_absorptions(ordered: Sequence[FineWindow], by_window: Mapping[int, Sequence[_Observation]], observations: Sequence[_Observation], base_pairs_by_seam: Sequence[Sequence[tuple[int, int]]], base_union: _UnionFind, policy: StitchingPolicy) -> tuple[tuple[tuple[tuple[int, int], ...], ...], tuple[Mapping[str, object], ...], Mapping[str, object], tuple[Mapping[tuple[int, int], tuple[int, int]], ...]]:
+    states = _component_states(observations, base_union)
+    strict_policy = StitchingPolicy.strict()
+    candidates: list[_LocalCandidate] = []
+    rejection_counts: dict[str, int] = {}
+    seam_rejections: list[dict[str, int]] = [{} for _ in range(max(0, len(ordered) - 1))]
+    funnel_counts: dict[str, int] = {}
+    seam_funnel_counts: list[dict[str, int]] = [{} for _ in range(max(0, len(ordered) - 1))]
+    relaxed_rejection_counts: dict[str, int] = {}
+    identity_trace_rejection_buckets: dict[str, int] = {}
+    anchor_delta_ticks_histogram: dict[str, int] = {}
+
+    def reject(seam_index: int, reason: str) -> None:
+        rejection_counts[reason] = rejection_counts.get(reason, 0) + 1
+        current = seam_rejections[seam_index]
+        current[reason] = current.get(reason, 0) + 1
+
+    def advance(seam_index: int, stage: str) -> None:
+        funnel_counts[stage] = funnel_counts.get(stage, 0) + 1
+        current = seam_funnel_counts[seam_index]
+        current[stage] = current.get(stage, 0) + 1
+
+    def increment(values: dict[str, int], name: str) -> None:
+        values[name] = values.get(name, 0) + 1
+    for seam_index in range(len(ordered) - 1):
+        left_window, right_window = (ordered[seam_index], ordered[seam_index + 1])
+        left, right = (by_window[seam_index], by_window[seam_index + 1])
+        overlap_start = max(left_window.global_tick_offset + left_window.valid_start_tick, right_window.global_tick_offset + right_window.valid_start_tick)
+        overlap_end = min(left_window.global_tick_offset + left_window.valid_end_tick, right_window.global_tick_offset + right_window.valid_end_tick)
+        base_pairs = tuple(base_pairs_by_seam[seam_index])
+        matched_left = {pair[0] for pair in base_pairs}
+        matched_right = {pair[1] for pair in base_pairs}
+        for left_item in left:
+            if left_item.observation_id in matched_left:
+                continue
+            for right_item in right:
+                if right_item.observation_id in matched_right:
+                    continue
+                advance(seam_index, 'unmatched_observation_pairs')
+                left_root = base_union.find(left_item.observation_id)
+                right_root = base_union.find(right_item.observation_id)
+                if left_root == right_root:
+                    advance(seam_index, 'same_component_pairs')
+                    continue
+                advance(seam_index, 'cross_component_pairs')
+                relaxed = _candidate_evaluation(left_item, right_item, overlap_start, overlap_end, policy)
+                if relaxed.cost is None:
+                    advance(seam_index, 'relaxed_infeasible_pairs')
+                    increment(relaxed_rejection_counts, str(relaxed.rejection or 'unknown'))
+                    continue
+                advance(seam_index, 'relaxed_feasible_pairs')
+                strict = _candidate_evaluation(left_item, right_item, overlap_start, overlap_end, strict_policy)
+                if strict.cost is not None:
+                    advance(seam_index, 'strict_feasible_pairs')
+                    reject(seam_index, 'strict_feasible_not_selected')
+                    continue
+                advance(seam_index, 'strict_infeasible_pairs')
+                intersection, union = _trace_iou_counts(left_item, right_item, overlap_start=overlap_start, overlap_end=overlap_end)
+                if union <= 0 or intersection * policy.min_shared_trace_iou_denominator < union * policy.min_shared_trace_iou_numerator:
+                    if union <= 0 or intersection <= 0:
+                        trace_bucket = 'zero'
+                    elif intersection * 4 < union:
+                        trace_bucket = 'gt_0_lt_0.25'
+                    elif intersection * 2 < union:
+                        trace_bucket = '0.25_to_lt_0.5'
+                    else:
+                        trace_bucket = '0.5_to_lt_0.75'
+                    increment(identity_trace_rejection_buckets, trace_bucket)
+                    reject(seam_index, 'identity_trace_iou')
+                    continue
+                advance(seam_index, 'identity_trace_pass_pairs')
+                maximum_anchor_delta_ticks = policy.endpoint_tolerance_ticks if policy.contract == COMPONENT_LOCAL_BOUNDED_ANCHOR_STITCHING_CONTRACT else 0
+                anchor, reason = _identity_anchor(states[left_root], states[right_root], maximum_delta_ticks=maximum_anchor_delta_ticks)
+                if anchor is None:
+                    if reason in {'partial_endpoint_is_not_exactly_anchored', 'partial_endpoint_exceeds_anchor_tolerance'}:
+                        left_state, right_state = (states[left_root], states[right_root])
+                        resolved, partial = (left_state, right_state) if left_state.resolved else (right_state, left_state)
+                        partial_endpoint = partial.partial_endpoint
+                        if partial_endpoint is None:
+                            raise FineTemporalError('exact-anchor rejection lacks a partial endpoint')
+                        endpoint_name, values = partial_endpoint
+                        anchor_values = resolved.starts if endpoint_name == 'start' else resolved.ends
+                        anchor_tick = _median(anchor_values)
+                        maximum_delta = max((abs(value - anchor_tick) for value in values))
+                        increment(anchor_delta_ticks_histogram, str(maximum_delta))
+                    reject(seam_index, str(reason))
+                    continue
+                advance(seam_index, 'identity_anchor_pass_pairs')
+                pair = (left_item.observation_id, right_item.observation_id)
+                if not _monotone_with_pairs(pair, base_pairs=base_pairs, left=left, right=right):
+                    reject(seam_index, 'crosses_strict_pairing')
+                    continue
+                advance(seam_index, 'monotone_pair_candidates')
+                resolved_root, partial_root, endpoint_name, anchor_tick, anchor_delta_ticks = anchor
+                candidates.append(_LocalCandidate(seam_index=seam_index, left_observation_id=pair[0], right_observation_id=pair[1], resolved_root=resolved_root, partial_root=partial_root, anchor_endpoint=endpoint_name, anchor_tick=anchor_tick, anchor_delta_ticks=anchor_delta_ticks, trace_intersection_ticks=intersection, trace_union_ticks=union))
+    by_component_pair: dict[tuple[int, int], list[_LocalCandidate]] = {}
+    for candidate in candidates:
+        key = (candidate.resolved_root, candidate.partial_root)
+        by_component_pair.setdefault(key, []).append(candidate)
+    unique_edges: list[_LocalCandidate] = []
+    for values in by_component_pair.values():
+        if len(values) != 1:
+            for candidate in values:
+                reject(candidate.seam_index, 'multiple_observation_edges')
+            continue
+        unique_edges.append(values[0])
+        advance(values[0].seam_index, 'unique_component_edge_pairs')
+    degrees: dict[int, int] = {}
+    for candidate in unique_edges:
+        degrees[candidate.resolved_root] = degrees.get(candidate.resolved_root, 0) + 1
+        degrees[candidate.partial_root] = degrees.get(candidate.partial_root, 0) + 1
+    reciprocal_unique: list[_LocalCandidate] = []
+    for candidate in unique_edges:
+        if degrees[candidate.resolved_root] != 1 or degrees[candidate.partial_root] != 1:
+            reject(candidate.seam_index, 'non_unique_component_identity')
+            continue
+        reciprocal_unique.append(candidate)
+        advance(candidate.seam_index, 'reciprocal_unique_pairs')
+    crossing: set[_LocalCandidate] = set()
+    by_seam_candidates: dict[int, list[_LocalCandidate]] = {}
+    for candidate in reciprocal_unique:
+        by_seam_candidates.setdefault(candidate.seam_index, []).append(candidate)
+    for seam_index, values in by_seam_candidates.items():
+        left, right = (by_window[seam_index], by_window[seam_index + 1])
+        for candidate in values:
+            peers = [peer.pair for peer in values if peer is not candidate]
+            if not _monotone_with_pairs(candidate.pair, base_pairs=peers, left=left, right=right):
+                crossing.add(candidate)
+    accepted = []
+    for candidate in reciprocal_unique:
+        if candidate in crossing:
+            reject(candidate.seam_index, 'crossing_local_candidates')
+            continue
+        accepted.append(candidate)
+        advance(candidate.seam_index, 'geometry_accepted_pairs')
+    accepted_by_seam: list[tuple[tuple[int, int], ...]] = []
+    geometry_by_seam: list[Mapping[tuple[int, int], tuple[int, int]]] = []
+    diagnostics_by_seam: list[Mapping[str, object]] = []
+    for seam_index in range(len(ordered) - 1):
+        seam_candidates = tuple((candidate for candidate in candidates if candidate.seam_index == seam_index))
+        seam_accepted = tuple((candidate for candidate in accepted if candidate.seam_index == seam_index))
+        accepted_by_seam.append(tuple(sorted((candidate.pair for candidate in seam_accepted))))
+        geometry_by_seam.append({candidate.pair: (_median(states[candidate.resolved_root].starts), _median(states[candidate.resolved_root].ends)) for candidate in seam_accepted})
+        diagnostics_by_seam.append({'candidate_pairs': [candidate.to_dict() for candidate in seam_candidates], 'accepted_pairs': [candidate.to_dict() for candidate in seam_accepted], 'rejection_counts': dict(sorted(seam_rejections[seam_index].items())), 'candidate_funnel': dict(sorted(seam_funnel_counts[seam_index].items()))})
+    summary = {'contract': policy.contract, 'strict_base_tolerance_ticks': STITCH_TOLERANCE_TICKS, 'candidate_tolerance_ticks': policy.endpoint_tolerance_ticks, 'minimum_identity_trace_iou': {'numerator': policy.min_shared_trace_iou_numerator, 'denominator': policy.min_shared_trace_iou_denominator}, 'candidate_pair_count': len(candidates), 'accepted_pair_count': len(accepted), 'absorbed_partial_component_count': len(accepted), 'resolved_event_count_change_by_contract': 0, 'resolved_event_geometry_change_by_contract': False, 'new_model_calls': 0, 'rejection_counts': dict(sorted(rejection_counts.items())), 'candidate_funnel': dict(sorted(funnel_counts.items())), 'candidate_funnel_details': {'relaxed_infeasible_rejection_counts': dict(sorted(relaxed_rejection_counts.items())), 'identity_trace_iou_rejection_buckets': dict(sorted(identity_trace_rejection_buckets.items())), 'anchor_maximum_delta_ticks_histogram': dict(sorted(anchor_delta_ticks_histogram.items(), key=lambda item: int(item[0])))}}
+    if policy.contract == COMPONENT_LOCAL_BOUNDED_ANCHOR_STITCHING_CONTRACT:
+        summary['anchor_contract'] = {'mode': 'bounded_certificate_only', 'maximum_delta_ticks': policy.endpoint_tolerance_ticks, 'resolved_geometry_authority': 'strict_resolved_component', 'explicit_allowlist_required_for_runtime': True}
+    return (tuple(accepted_by_seam), tuple(diagnostics_by_seam), summary, tuple(geometry_by_seam))
+
+def _association_plan(ordered: Sequence[FineWindow], by_window: Mapping[int, Sequence[_Observation]], observations: Sequence[_Observation], policy: StitchingPolicy, component_pair_allowlist: Collection[ComponentPairKey] | None=None) -> _AssociationPlan:
+    match_policy = StitchingPolicy.strict() if policy.contract in COMPONENT_LOCAL_STITCHING_CONTRACTS else policy
+    base_pairs: list[tuple[tuple[int, int], ...]] = []
+    ambiguous: list[bool] = []
+    competing: list[tuple[tuple[tuple[int, int], ...], ...]] = []
+    for index in range(len(ordered) - 1):
+        pairs, is_ambiguous, alternatives = _match_adjacent(by_window[index], by_window[index + 1], ordered[index], ordered[index + 1], match_policy)
+        base_pairs.append(pairs)
+        ambiguous.append(is_ambiguous)
+        competing.append(alternatives)
+    if policy.contract not in COMPONENT_LOCAL_STITCHING_CONTRACTS:
+        return _AssociationPlan(base_pairs_by_seam=tuple(base_pairs), pairs_by_seam=tuple(base_pairs), ambiguous_by_seam=tuple(ambiguous), competing_by_seam=tuple(competing), component_consistency_tolerance_ticks=policy.endpoint_tolerance_ticks, component_local_by_seam=tuple(({} for _ in base_pairs)), component_local_summary=None, component_local_geometry_by_seam=tuple(({} for _ in base_pairs)))
+    base_union = _UnionFind((item.observation_id for item in observations))
+    for pairs in base_pairs:
+        for left_id, right_id in pairs:
+            base_union.union(left_id, right_id)
+    accepted, diagnostics, summary, geometry_by_seam = _component_local_absorptions(ordered, by_window, observations, base_pairs, base_union, policy)
+    if component_pair_allowlist is not None:
+        available = {(seam_index, *pair) for seam_index, seam_pairs in enumerate(accepted) for pair in seam_pairs}
+        requested = set(component_pair_allowlist)
+        unknown = requested - available
+        if unknown:
+            raise FineTemporalError(f'component pair allowlist contains non-candidates: {sorted(unknown)}')
+        filtered = tuple((tuple((pair for pair in seam_pairs if (seam_index, *pair) in requested)) for seam_index, seam_pairs in enumerate(accepted)))
+        rejected_count = sum(map(len, accepted)) - sum(map(len, filtered))
+        filtered_diagnostics = []
+        for seam_index, row in enumerate(diagnostics):
+            current = dict(row)
+            current['accepted_pairs'] = [pair for pair in row['accepted_pairs'] if (seam_index, int(pair['left_observation_id']), int(pair['right_observation_id'])) in requested]
+            rejection_counts = dict(row['rejection_counts'])
+            seam_rejected = len(accepted[seam_index]) - len(filtered[seam_index])
+            if seam_rejected:
+                rejection_counts['caption_certificate'] = seam_rejected
+            current['rejection_counts'] = dict(sorted(rejection_counts.items()))
+            filtered_diagnostics.append(current)
+        accepted = filtered
+        geometry_by_seam = tuple(({pair: geometry for pair, geometry in seam_geometry.items() if (seam_index, *pair) in requested} for seam_index, seam_geometry in enumerate(geometry_by_seam)))
+        diagnostics = tuple(filtered_diagnostics)
+        summary = dict(summary)
+        summary['accepted_pair_count'] = sum(map(len, accepted))
+        summary['absorbed_partial_component_count'] = sum(map(len, accepted))
+        rejection_counts = dict(summary['rejection_counts'])
+        if rejected_count:
+            rejection_counts['caption_certificate'] = rejected_count
+        summary['rejection_counts'] = dict(sorted(rejection_counts.items()))
+    combined = tuple((tuple(sorted((*strict_pairs, *local_pairs))) for strict_pairs, local_pairs in zip(base_pairs, accepted, strict=True)))
+    return _AssociationPlan(base_pairs_by_seam=tuple(base_pairs), pairs_by_seam=combined, ambiguous_by_seam=tuple(ambiguous), competing_by_seam=tuple(competing), component_consistency_tolerance_ticks=STITCH_TOLERANCE_TICKS, component_local_by_seam=diagnostics, component_local_summary=summary, component_local_geometry_by_seam=geometry_by_seam)
+
+def _fragment_event(source_id: str, window: FineWindow, fragment: FineFragment, observation_id: int, row_ordinal: int) -> EventInterval:
+    start_us = max(0, (window.global_tick_offset + fragment.s) * TICK_US)
+    end_us = min(window.video_end_us, (window.global_tick_offset + fragment.e) * TICK_US)
+    if end_us <= start_us:
+        raise FineTemporalError('quantized local fragment is degenerate in source time')
+    completeness = {(False, False): 'complete', (True, False): 'left_censored', (False, True): 'right_censored', (True, True): 'both_censored'}[fragment.lo, fragment.ro]
+    return EventInterval(event_id=f'{source_id}:fragment:{observation_id:06d}', source_id=source_id, span=TimeSpan(start_us / 1000000, end_us / 1000000), completeness=completeness, source_scope_ids=(window.scope_id,), metadata={'window_index': window.window_index, 'row_ordinal': row_ordinal, 'observation_id': observation_id})
+
+def _observation_payload(item: _Observation) -> dict[str, int | bool | str]:
+    return {'observation_id': item.observation_id, 'window_index': item.window_index, 'row_ordinal': item.row_ordinal, 'scope_id': item.scope_id, 'start_tick': item.start_tick, 'end_tick': item.end_tick, 'lo': item.lo, 'ro': item.ro, 'valid_start_tick': item.valid_start_tick, 'valid_end_tick': item.valid_end_tick}
+
+def analyze_stitching_seams(windows: Sequence[FineWindow], fragments_by_window: Mapping[int, Sequence[FineFragment]], *, policy: StitchingPolicy | None=None, tolerance_ticks: int=STITCH_TOLERANCE_TICKS, component_pair_allowlist: Collection[ComponentPairKey] | None=None) -> dict[str, object]:
+    selected = policy or StitchingPolicy.strict(tolerance_ticks)
+    if policy is not None and tolerance_ticks != STITCH_TOLERANCE_TICKS:
+        raise FineTemporalError('pass either policy or a non-default tolerance_ticks, not both')
+    if component_pair_allowlist is not None and (policy is None or policy.contract not in COMPONENT_LOCAL_STITCHING_CONTRACTS):
+        raise FineTemporalError('component pair allowlist requires a component-local policy')
+    ordered = tuple(sorted(windows, key=lambda item: item.window_index))
+    if not ordered or [item.window_index for item in ordered] != list(range(len(ordered))):
+        raise FineTemporalError('windows must be one complete zero-based sequence')
+    if len({item.source_id for item in ordered}) != 1 or len({item.video_end_us for item in ordered}) != 1 or len({item.window_profile_id for item in ordered}) != 1:
+        raise FineTemporalError('all windows must belong to one source, duration, and profile')
+    unknown = set(fragments_by_window) - {item.window_index for item in ordered}
+    if unknown:
+        raise FineTemporalError(f'fragments supplied for unknown windows: {sorted(unknown)}')
+    by_window: dict[int, tuple[_Observation, ...]] = {}
+    observations: list[_Observation] = []
+    next_id = 0
+    for window in ordered:
+        current = _observations(window, tuple(fragments_by_window.get(window.window_index, ())), first_id=next_id)
+        next_id += len(current)
+        observations.extend(current)
+        by_window[window.window_index] = current
+    plan = _association_plan(ordered, by_window, observations, selected, component_pair_allowlist)
+    seams: list[dict[str, object]] = []
+    rejection_totals: dict[str, int] = {}
+    total_feasible = total_matched = total_ambiguous = 0
+    for index in range(len(ordered) - 1):
+        left_window, right_window = (ordered[index], ordered[index + 1])
+        left, right = (by_window[index], by_window[index + 1])
+        overlap_start = max(left_window.global_tick_offset + left_window.valid_start_tick, right_window.global_tick_offset + right_window.valid_start_tick)
+        overlap_end = min(left_window.global_tick_offset + left_window.valid_end_tick, right_window.global_tick_offset + right_window.valid_end_tick)
+        rejection_counts: dict[str, int] = {}
+        feasible_pairs: list[dict[str, object]] = []
+        for left_item in left:
+            for right_item in right:
+                evaluation = _candidate_evaluation(left_item, right_item, overlap_start, overlap_end, selected)
+                if evaluation.cost is None:
+                    reason = str(evaluation.rejection)
+                    rejection_counts[reason] = rejection_counts.get(reason, 0) + 1
+                    rejection_totals[reason] = rejection_totals.get(reason, 0) + 1
+                    continue
+                feasible_pairs.append({'left_observation_id': left_item.observation_id, 'right_observation_id': right_item.observation_id, 'trace_cost': evaluation.cost[0], 'closed_endpoint_cost': evaluation.cost[1]})
+        pairs = plan.pairs_by_seam[index]
+        ambiguous = plan.ambiguous_by_seam[index]
+        competing = plan.competing_by_seam[index]
+        matched_left = {item[0] for item in pairs}
+        matched_right = {item[1] for item in pairs}
+        total_feasible += len(feasible_pairs)
+        total_matched += len(pairs)
+        total_ambiguous += int(ambiguous)
+        seams.append({'left_window_index': left_window.window_index, 'right_window_index': right_window.window_index, 'overlap_start_tick': overlap_start, 'overlap_end_tick': overlap_end, 'left_observations': [_observation_payload(item) for item in left], 'right_observations': [_observation_payload(item) for item in right], 'feasible_pairs': feasible_pairs, 'matched_pairs': [list(item) for item in pairs], 'competing_optimal_pairings': [[list(item) for item in pairing] for pairing in competing], 'matching_ambiguous': ambiguous, 'unmatched_left_observation_ids': [item.observation_id for item in left if item.observation_id not in matched_left], 'unmatched_right_observation_ids': [item.observation_id for item in right if item.observation_id not in matched_right], 'rejection_counts': dict(sorted(rejection_counts.items())), 'component_local_identity': dict(plan.component_local_by_seam[index])})
+    return {'schema_version': STITCHING_SEAM_DIAGNOSTICS_SCHEMA, 'policy': selected.to_dict(), 'window_count': len(ordered), 'seam_count': len(seams), 'observation_count': next_id, 'feasible_pair_count': total_feasible, 'matched_pair_count': total_matched, 'ambiguous_seam_count': total_ambiguous, 'rejection_counts': dict(sorted(rejection_totals.items())), 'component_local_identity': dict(plan.component_local_summary) if plan.component_local_summary is not None else None, 'seams': seams}
+
+def stitch_predictions(windows: Sequence[FineWindow], fragments_by_window: Mapping[int, Sequence[FineFragment]], *, tolerance_ticks: int=STITCH_TOLERANCE_TICKS, policy: StitchingPolicy | None=None, component_pair_allowlist: Collection[ComponentPairKey] | None=None) -> FineSegmentationResult:
+    if policy is not None and tolerance_ticks != STITCH_TOLERANCE_TICKS:
+        raise FineTemporalError('pass either policy or a non-default tolerance_ticks, not both')
+    if component_pair_allowlist is not None and (policy is None or policy.contract not in COMPONENT_LOCAL_STITCHING_CONTRACTS):
+        raise FineTemporalError('component pair allowlist requires a component-local policy')
+    selected = policy or StitchingPolicy.strict(tolerance_ticks)
+    if selected.contract == COMPONENT_LOCAL_BOUNDED_ANCHOR_STITCHING_CONTRACT and component_pair_allowlist is None:
+        raise FineTemporalError('bounded-anchor runtime requires an explicit component pair allowlist')
+    ordered = tuple(sorted(windows, key=lambda item: item.window_index))
+    if not ordered or [item.window_index for item in ordered] != list(range(len(ordered))):
+        raise FineTemporalError('windows must be one complete zero-based sequence')
+    if len({item.source_id for item in ordered}) != 1 or len({item.video_end_us for item in ordered}) != 1 or len({item.window_profile_id for item in ordered}) != 1:
+        raise FineTemporalError('all windows must belong to one source, duration, and profile')
+    unknown = set(fragments_by_window) - {item.window_index for item in ordered}
+    if unknown:
+        raise FineTemporalError(f'fragments supplied for unknown windows: {sorted(unknown)}')
+    observations: list[_Observation] = []
+    by_window: dict[int, tuple[_Observation, ...]] = {}
+    fragments: list[EventInterval] = []
+    next_id = 0
+    for window in ordered:
+        rows = tuple(fragments_by_window.get(window.window_index, ()))
+        current = _observations(window, rows, first_id=next_id)
+        for offset, fragment in enumerate(rows):
+            fragments.append(_fragment_event(window.source_id, window, fragment, next_id + offset, offset))
+        next_id += len(current)
+        observations.extend(current)
+        by_window[window.window_index] = current
+    plan = _association_plan(ordered, by_window, observations, selected, component_pair_allowlist)
+    baseline_geometry: tuple[tuple[int, int], ...] | None = None
+    baseline_unresolved_count: int | None = None
+    if selected.contract in COMPONENT_LOCAL_STITCHING_CONTRACTS:
+        baseline_union = _UnionFind((item.observation_id for item in observations))
+        for pairs in plan.base_pairs_by_seam:
+            for left_id, right_id in pairs:
+                baseline_union.union(left_id, right_id)
+        baseline_states = _component_states(observations, baseline_union)
+        baseline_geometry = _resolved_component_geometry(baseline_states)
+        baseline_unresolved_count = sum((not state.resolved for state in baseline_states.values()))
+    union_find = _UnionFind((item.observation_id for item in observations))
+    ambiguity_ids: set[int] = set()
+    ambiguity_count = 0
+    for pairs, ambiguous, competing in zip(plan.pairs_by_seam, plan.ambiguous_by_seam, plan.competing_by_seam, strict=True):
+        if ambiguous:
+            ambiguity_count += 1
+            ambiguity_ids.update((value for pairing in competing for pair in pairing for value in pair))
+        for left_id, right_id in pairs:
+            union_find.union(left_id, right_id)
+    authoritative_geometry_by_root: dict[int, tuple[int, int]] = {}
+    for geometry_by_pair in plan.component_local_geometry_by_seam:
+        for (left_id, right_id), geometry in geometry_by_pair.items():
+            root = union_find.find(left_id)
+            if root != union_find.find(right_id):
+                raise FineTemporalError('component-local geometry pair was not associated')
+            prior = authoritative_geometry_by_root.get(root)
+            if prior is not None and prior != geometry:
+                raise FineTemporalError('component-local candidates disagree on authoritative geometry')
+            authoritative_geometry_by_root[root] = geometry
+    if baseline_geometry is not None and baseline_unresolved_count is not None:
+        final_states = _component_states(observations, union_find)
+        accepted_count = sum((len(pairs) - len(base_pairs) for pairs, base_pairs in zip(plan.pairs_by_seam, plan.base_pairs_by_seam, strict=True)))
+        effective_geometry = tuple(sorted((authoritative_geometry_by_root.get(root, (_median(state.starts), _median(state.ends))) for root, state in final_states.items() if root in authoritative_geometry_by_root or state.resolved)))
+        if effective_geometry != baseline_geometry:
+            raise FineTemporalError('component-local identity relaxation changed resolved event geometry')
+        if sum((root not in authoritative_geometry_by_root and (not state.resolved) for root, state in final_states.items())) != baseline_unresolved_count - accepted_count:
+            raise FineTemporalError('component-local identity relaxation changed unsupported component structure')
+    grouped: dict[int, list[_Observation]] = {}
+    for observation in observations:
+        grouped.setdefault(union_find.find(observation.observation_id), []).append(observation)
+    ambiguous_roots = {union_find.find(value) for value in ambiguity_ids}
+    resolved: list[tuple[TimeSpan, tuple[str, ...], dict[str, object]]] = []
+    unresolved: list[UnresolvedFineEvent] = []
+    video_end_us = ordered[0].video_end_us
+    windows_by_index = {item.window_index: item for item in ordered}
+    for root, component in sorted(grouped.items()):
+        starts = tuple(sorted((item.start_tick for item in component if item.has_closed_start)))
+        ends = tuple(sorted((item.end_tick for item in component if item.has_closed_end)))
+        observation_spans = tuple(((item.start_tick, item.end_tick) for item in sorted(component, key=lambda item: item.observation_id)))
+        scope_ids = tuple(dict.fromkeys((item.scope_id for item in sorted(component, key=lambda item: item.window_index))))
+        raw_consistent = bool(starts) and bool(ends) and (starts[-1] - starts[0] <= plan.component_consistency_tolerance_ticks) and (ends[-1] - ends[0] <= plan.component_consistency_tolerance_ticks)
+        authoritative_geometry = authoritative_geometry_by_root.get(root)
+        consistent = authoritative_geometry is not None or raw_consistent
+        if authoritative_geometry is not None:
+            start_tick, end_tick = authoritative_geometry
+            start_us = max(0, start_tick * TICK_US)
+            end_us = min(video_end_us, end_tick * TICK_US)
+        else:
+            start_us = max(0, _median(starts) * TICK_US) if starts else None
+            end_us = min(video_end_us, _median(ends) * TICK_US) if ends else None
+        if not consistent or start_us is None or end_us is None or (end_us <= start_us):
+            support_window_indices = {item.window_index for item in component}
+            if not starts:
+                support_window_indices.update((item.window_index - 1 for item in component if item.window_index > 0))
+            if not ends:
+                support_window_indices.update((item.window_index + 1 for item in component if item.window_index + 1 < len(ordered)))
+            if starts and ends and (not consistent):
+                support_window_indices.update((item.window_index - 1 for item in component if item.window_index > 0))
+                support_window_indices.update((item.window_index + 1 for item in component if item.window_index + 1 < len(ordered)))
+            support_windows = tuple((windows_by_index[index] for index in sorted(support_window_indices)))
+            support_start_tick = min((item.global_tick_offset + item.valid_start_tick for item in support_windows))
+            support_end_tick = max((item.global_tick_offset + item.valid_end_tick for item in support_windows))
+            unresolved.append(UnresolvedFineEvent(tuple(sorted((item.observation_id for item in component))), scope_ids, observation_spans, starts, ends, support_start_tick, support_end_tick, root in ambiguous_roots))
+            continue
+        resolved.append((TimeSpan(start_us / 1000000, end_us / 1000000), scope_ids, {'observation_ids': tuple(sorted((item.observation_id for item in component))), 'matching_ambiguous': root in ambiguous_roots, **({'component_local_geometry_authority': 'strict_resolved_component', 'raw_closed_endpoints_consistent': raw_consistent} if authoritative_geometry is not None and selected.contract == COMPONENT_LOCAL_BOUNDED_ANCHOR_STITCHING_CONTRACT else {})}))
+    resolved.sort(key=lambda item: (item[0].start_s, item[0].end_s, item[1]))
+    identified = tuple((EventInterval(event_id=f'{ordered[0].source_id}:fine:{index:06d}', source_id=ordered[0].source_id, span=span, source_scope_ids=scope_ids, metadata=metadata) for index, (span, scope_ids, metadata) in enumerate(resolved)))
+    return FineSegmentationResult(events=identified, fragments=tuple(fragments), unresolved=tuple(unresolved), adjacent_matching_ambiguities=ambiguity_count, diagnostics={'component_local_identity': dict(plan.component_local_summary)} if plan.component_local_summary is not None else {})
+
+def window_from_scope(scope: ProcessingScope) -> FineWindow:
+    metadata = dict(scope.metadata)
+    if metadata.get('schema_version') != WINDOW_SCHEMA:
+        raise FineTemporalError('processing scope is not a fine-window scope')
+    profile_id = metadata.get('window_profile_id', DEFAULT_FINE_WINDOW_PROFILE_ID)
+    profile = fine_window_profile(profile_id)
+    if metadata.get('model_io_contract') != profile.model_io_contract:
+        raise FineTemporalError('processing scope model I/O contract mismatch')
+    declared_profile = metadata.get('window_profile')
+    if declared_profile is not None and declared_profile != profile.to_dict():
+        raise FineTemporalError('processing scope fine window profile binding mismatch')
+    required = ('window_index', 'video_end_us', 'requested_view_start_us', 'requested_view_end_us', 'valid_start_us', 'valid_end_us', 'core_start_us', 'core_end_us', 'valid_start_tick', 'valid_end_tick', 'core_start_tick', 'core_end_tick')
+    values: dict[str, int] = {}
+    for key in required:
+        value = metadata.get(key)
+        if isinstance(value, bool) or not isinstance(value, int):
+            raise FineTemporalError(f'scope metadata {key} must be an integer')
+        values[key] = value
+    window = FineWindow(source_id=scope.source_id, scope_id=scope.scope_id, window_profile_id=profile.profile_id, coarse_spans=tuple(metadata.get('coarse_spans') or ()), **values)
+    expected_core = TimeSpan(window.core_start_us / 1000000, window.core_end_us / 1000000)
+    expected_context = TimeSpan(window.valid_start_us / 1000000, window.valid_end_us / 1000000)
+    if scope.core_span != expected_core or scope.context_span != expected_context:
+        raise FineTemporalError('float scope spans disagree with bound exact-us geometry')
+    return window

batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/fine_worker.py

@@ -0,0 +1,23 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+from bisect import bisect_left
+from collections.abc import Sequence
+from fractions import Fraction
+SLOT_SPACING_US = 125000
+
+class FineWorkerError(RuntimeError):
+    pass
+
+def _nearest_index(requested_us: int, times: Sequence[Fraction], *, max_selection_error_us: int=SLOT_SPACING_US // 2) -> int:
+    requested = Fraction(requested_us, 1)
+    position = bisect_left(times, requested)
+    candidates = [index for index in (position - 1, position) if 0 <= index < len(times)]
+    if not candidates:
+        raise FineWorkerError('no decoded frame is available for an in-video slot')
+    selected = min(candidates, key=lambda index: (abs(times[index] - requested), times[index], index))
+    if abs(times[selected] - requested) > max_selection_error_us:
+        raise FineWorkerError('nearest decoded frame is farther than half a profile slot')
+    return selected
+if __name__ == '__main__':
+    raise SystemExit(main())

batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/interface.py

@@ -0,0 +1,24 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+from collections.abc import Awaitable, Sequence
+from enum import Enum
+from typing import Protocol, runtime_checkable
+from flora_runtime.schema import AnnotationSource, TemporalSegment, TimeSpan
+
+class SegmentTopology(str, Enum):
+    EXCLUSIVE = 'exclusive_intervals'
+    CONCURRENT = 'concurrent_intervals'
+
+def segment_topology(segmenter: object) -> SegmentTopology:
+    value = getattr(segmenter, 'segment_topology', SegmentTopology.EXCLUSIVE)
+    try:
+        return SegmentTopology(value)
+    except ValueError as exc:
+        raise ValueError(f'segmenter declares unsupported segment_topology: {value!r}') from exc
+
+@runtime_checkable
+class VideoSegmenter(Protocol):
+
+    def segment(self, source: AnnotationSource) -> Sequence[TemporalSegment | TimeSpan] | Awaitable[Sequence[TemporalSegment | TimeSpan]]:
+        ...

batch_utils/vendor/production_adapter/core/flora_runtime/video_segmentation/scope_scheduler.py

@@ -0,0 +1,163 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+from collections.abc import Mapping, Sequence
+from decimal import Decimal, InvalidOperation
+from itertools import pairwise
+import math
+from flora_runtime.schema import TimeSpan
+from flora_runtime.video_segmentation.contracts import ProcessingScope
+from flora_runtime.video_segmentation.fine_temporal import DEFAULT_FINE_WINDOW_PROFILE_ID, TICK_US, WINDOW_SCHEMA, ceil_div, fine_window_profile, floor_div, round_half_up_div
+
+class ScopeSchedulerError(ValueError):
+    pass
+FINE_WINDOW_SCOPE_NAMESPACE = 'fine-window'
+FINE_RETRY_SCOPE_NAMESPACE = 'fine-retry'
+FINE_DIRECTIONAL_RETRY_SCOPE_NAMESPACE = 'fine-retry-directional'
+
+def seconds_to_us(value: float, *, label: str) -> int:
+    if isinstance(value, bool) or not isinstance(value, int | float):
+        raise ScopeSchedulerError(f'{label} must be numeric')
+    try:
+        parsed = Decimal(str(value)) * 1000000
+    except (InvalidOperation, ValueError) as exc:
+        raise ScopeSchedulerError(f'{label} is not valid decimal seconds') from exc
+    if not parsed.is_finite() or parsed != parsed.to_integral_value():
+        raise ScopeSchedulerError(f'{label} must resolve to integral microseconds')
+    result = int(parsed)
+    if result < 0:
+        raise ScopeSchedulerError(f'{label} must be non-negative')
+    return result
+
+def _quantized_prior(proposal: ProcessingScope, *, requested_start_us: int, valid_start_us: int, valid_end_us: int, valid_start_tick: int, valid_end_tick: int) -> dict[str, object] | None:
+    start_us = seconds_to_us(proposal.core_span.start_s, label=f'{proposal.scope_id}.start_s')
+    end_us = seconds_to_us(proposal.core_span.end_s, label=f'{proposal.scope_id}.end_s')
+    if start_us >= valid_end_us or end_us <= valid_start_us:
+        return None
+    visible_start = max(start_us, valid_start_us)
+    visible_end = min(end_us, valid_end_us)
+    lo, ro = (start_us < valid_start_us, end_us > valid_end_us)
+    start_delta, end_delta = (visible_start - requested_start_us, visible_end - requested_start_us)
+    s = valid_start_tick if lo else round_half_up_div(start_delta, TICK_US)
+    e = valid_end_tick if ro else round_half_up_div(end_delta, TICK_US)
+    if s >= e:
+        s, e = (floor_div(start_delta, TICK_US), ceil_div(end_delta, TICK_US))
+    if not valid_start_tick <= s < e <= valid_end_tick:
+        raise ScopeSchedulerError(f'proposal {proposal.scope_id} cannot be represented at 20 Hz')
+    return {'s': s, 'e': e, 'lo': lo, 'ro': ro}
+
+def schedule_fine_scopes(*, source_id: str, duration_s: float, coarse_proposals: Sequence[ProcessingScope]=(), arm: str='no_prior_global_scan', window_profile_id: str=DEFAULT_FINE_WINDOW_PROFILE_ID) -> tuple[ProcessingScope, ...]:
+    if not isinstance(source_id, str) or not source_id.strip():
+        raise ScopeSchedulerError('source_id must be non-empty')
+    if not isinstance(arm, str) or not arm.strip():
+        raise ScopeSchedulerError('arm must be non-empty')
+    try:
+        profile = fine_window_profile(window_profile_id)
+    except ValueError as exc:
+        raise ScopeSchedulerError(str(exc)) from exc
+    video_end_us = seconds_to_us(duration_s, label='duration_s')
+    if video_end_us <= 0:
+        raise ScopeSchedulerError('duration_s must be positive')
+    proposals = tuple(coarse_proposals)
+    proposal_ids = [proposal.scope_id for proposal in proposals]
+    if len(proposal_ids) != len(set(proposal_ids)):
+        raise ScopeSchedulerError('coarse proposal scope_ids must be unique')
+    if any((proposal.source_id != source_id for proposal in proposals)):
+        raise ScopeSchedulerError('every coarse proposal must belong to source_id')
+    for proposal in proposals:
+        proposal_end_us = seconds_to_us(proposal.core_span.end_s, label=f'{proposal.scope_id}.end_s')
+        if proposal_end_us > video_end_us:
+            raise ScopeSchedulerError(f'coarse proposal {proposal.scope_id} extends past source')
+    proposals = tuple(sorted(proposals, key=lambda item: (item.core_span, item.scope_id)))
+    scopes: list[ProcessingScope] = []
+    window_index = 0
+    while True:
+        core_start_us = window_index * profile.core_us
+        if core_start_us >= video_end_us:
+            break
+        core_end_us = min(core_start_us + profile.core_us, video_end_us)
+        requested_start_us = core_start_us - profile.left_guard_us
+        requested_end_us = requested_start_us + profile.view_us
+        valid_start_us = max(0, requested_start_us)
+        valid_end_us = min(video_end_us, requested_end_us)
+        valid_start_tick = floor_div(valid_start_us - requested_start_us, TICK_US)
+        valid_end_tick = ceil_div(valid_end_us - requested_start_us, TICK_US)
+        core_start_tick = floor_div(core_start_us - requested_start_us, TICK_US)
+        core_end_tick = ceil_div(core_end_us - requested_start_us, TICK_US)
+        prior_groups: dict[tuple[int, int, bool, bool], list[str]] = {}
+        for proposal in proposals:
+            row = _quantized_prior(proposal, requested_start_us=requested_start_us, valid_start_us=valid_start_us, valid_end_us=valid_end_us, valid_start_tick=valid_start_tick, valid_end_tick=valid_end_tick)
+            if row is not None:
+                key = (int(row['s']), int(row['e']), bool(row['lo']), bool(row['ro']))
+                prior_groups.setdefault(key, []).append(proposal.scope_id)
+        ordered_priors = sorted(prior_groups.items())
+        coarse_rows = [{'s': key[0], 'e': key[1], 'lo': key[2], 'ro': key[3]} for key, _ in ordered_priors]
+        visible_ids = [scope_id for _, scope_ids in ordered_priors for scope_id in scope_ids]
+        coarse_span_lineage = [{'row_index': index, 'scope_ids': scope_ids} for index, (_, scope_ids) in enumerate(ordered_priors)]
+        if len(coarse_rows) > 64:
+            raise ScopeSchedulerError('one fine view contains more than 64 coarse priors')
+        scope_id = f'{source_id}:{FINE_WINDOW_SCOPE_NAMESPACE}:{window_index:06d}'
+        metadata = {'schema_version': WINDOW_SCHEMA, 'model_io_contract': profile.model_io_contract, 'window_profile_id': profile.profile_id, 'window_profile': profile.to_dict(), 'scheduler_contract': profile.scheduler_contract, 'arm': arm, 'guidance': 'coarse_prior' if coarse_rows else 'no_prior_global_scan', 'coarse_spans': coarse_rows, 'coarse_scope_ids': visible_ids, 'coarse_span_lineage': coarse_span_lineage, 'window_index': window_index, 'video_end_us': video_end_us, 'requested_view_start_us': requested_start_us, 'requested_view_end_us': requested_end_us, 'valid_start_us': valid_start_us, 'valid_end_us': valid_end_us, 'core_start_us': core_start_us, 'core_end_us': core_end_us, 'valid_start_tick': valid_start_tick, 'valid_end_tick': valid_end_tick, 'core_start_tick': core_start_tick, 'core_end_tick': core_end_tick}
+        scopes.append(ProcessingScope(scope_id=scope_id, source_id=source_id, core_span=TimeSpan(core_start_us / 1000000, core_end_us / 1000000), context_span=TimeSpan(valid_start_us / 1000000, valid_end_us / 1000000), metadata=metadata))
+        window_index += 1
+    if not scopes:
+        raise ScopeSchedulerError('scope scheduling produced no windows')
+    if scopes[0].core_span.start_s != 0 or not math.isclose(scopes[-1].core_span.end_s, duration_s):
+        raise ScopeSchedulerError('fine emit cores do not cover the complete source')
+    for left, right in pairwise(scopes):
+        if left.core_span.end_s != right.core_span.start_s:
+            raise ScopeSchedulerError('fine emit cores are not an exact partition')
+    return tuple(scopes)
+
+def schedule_recentered_fine_retry_scopes(*, source_id: str, duration_s: float, component_edge_ticks: Sequence[tuple[int, int]], arm: str='adaptive_fine_repair', edge_core_offset_ticks: Mapping[int, int] | None=None, scope_namespace: str=FINE_RETRY_SCOPE_NAMESPACE, window_profile_id: str=DEFAULT_FINE_WINDOW_PROFILE_ID) -> tuple[ProcessingScope, ...]:
+    if not isinstance(source_id, str) or not source_id.strip():
+        raise ScopeSchedulerError('source_id must be non-empty')
+    if not isinstance(arm, str) or not arm.strip():
+        raise ScopeSchedulerError('arm must be non-empty')
+    try:
+        profile = fine_window_profile(window_profile_id)
+    except ValueError as exc:
+        raise ScopeSchedulerError(str(exc)) from exc
+    if not isinstance(scope_namespace, str) or not scope_namespace or any((not (character.isalnum() or character in '_-') for character in scope_namespace)):
+        raise ScopeSchedulerError('adaptive fine retry scope_namespace is invalid')
+    video_end_us = seconds_to_us(duration_s, label='duration_s')
+    if video_end_us <= 0:
+        raise ScopeSchedulerError('duration_s must be positive')
+    targets = tuple(component_edge_ticks)
+    if not targets:
+        raise ScopeSchedulerError('adaptive fine retry requires at least one component edge')
+    component_indices = [item[0] for item in targets]
+    if len(component_indices) != len(set(component_indices)):
+        raise ScopeSchedulerError('adaptive fine retry component indices must be unique')
+    offsets = dict(edge_core_offset_ticks or {})
+    if not set(offsets) <= set(component_indices):
+        raise ScopeSchedulerError('adaptive fine retry offsets contain an unknown component')
+    scopes: list[ProcessingScope] = []
+    for window_index, (component_index, edge_tick) in enumerate(targets):
+        if isinstance(component_index, bool) or not isinstance(component_index, int) or component_index < 0:
+            raise ScopeSchedulerError('adaptive fine retry component index must be non-negative')
+        if isinstance(edge_tick, bool) or not isinstance(edge_tick, int):
+            raise ScopeSchedulerError('adaptive fine retry edge tick must be an integer')
+        edge_core_offset_tick = offsets.get(component_index, floor_div(profile.core_us // 2, TICK_US))
+        if isinstance(edge_core_offset_tick, bool) or not isinstance(edge_core_offset_tick, int) or (not 0 <= edge_core_offset_tick <= profile.core_ticks):
+            raise ScopeSchedulerError('adaptive fine retry edge core offset must lie inside the emit core')
+        edge_us = edge_tick * TICK_US
+        if not 0 < edge_us < video_end_us:
+            raise ScopeSchedulerError('adaptive fine retry edge must be inside the source')
+        latest_full_core_start_us = floor_div(max(0, video_end_us - profile.core_us), TICK_US) * TICK_US
+        core_start_us = min(max(0, edge_us - edge_core_offset_tick * TICK_US), latest_full_core_start_us)
+        core_end_us = min(core_start_us + profile.core_us, video_end_us)
+        requested_start_us = core_start_us - profile.left_guard_us
+        requested_end_us = requested_start_us + profile.view_us
+        valid_start_us = max(0, requested_start_us)
+        valid_end_us = min(video_end_us, requested_end_us)
+        valid_start_tick = floor_div(valid_start_us - requested_start_us, TICK_US)
+        valid_end_tick = ceil_div(valid_end_us - requested_start_us, TICK_US)
+        core_start_tick = floor_div(core_start_us - requested_start_us, TICK_US)
+        core_end_tick = ceil_div(core_end_us - requested_start_us, TICK_US)
+        if not core_start_us <= edge_us <= core_end_us:
+            raise ScopeSchedulerError('adaptive fine retry edge does not lie in the emit core')
+        scope_id = f'{source_id}:{scope_namespace}:{window_index:06d}'
+        metadata = {'schema_version': WINDOW_SCHEMA, 'model_io_contract': profile.model_io_contract, 'window_profile_id': profile.profile_id, 'window_profile': profile.to_dict(), 'scheduler_contract': profile.retry_scheduler_contract(directional=component_index in offsets), 'arm': arm, 'guidance': 'no_prior_adaptive_retry', 'coarse_spans': [], 'coarse_scope_ids': [], 'coarse_span_lineage': [], 'window_index': window_index, 'retry_component_index': component_index, 'retry_edge_tick': edge_tick, 'retry_edge_core_offset_tick': edge_core_offset_tick, 'video_end_us': video_end_us, 'requested_view_start_us': requested_start_us, 'requested_view_end_us': requested_end_us, 'valid_start_us': valid_start_us, 'valid_end_us': valid_end_us, 'core_start_us': core_start_us, 'core_end_us': core_end_us, 'valid_start_tick': valid_start_tick, 'valid_end_tick': valid_end_tick, 'core_start_tick': core_start_tick, 'core_end_tick': core_end_tick}
+        scopes.append(ProcessingScope(scope_id=scope_id, source_id=source_id, core_span=TimeSpan(core_start_us / 1000000, core_end_us / 1000000), context_span=TimeSpan(valid_start_us / 1000000, valid_end_us / 1000000), metadata=metadata))
+    return tuple(scopes)

batch_utils/vendor/production_adapter/core/flora_runtime/worker_transport.py

@@ -0,0 +1,18 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+from collections.abc import Mapping
+from typing import Any, Protocol, runtime_checkable
+from flora_runtime.schema import AnnotationSource
+
+class SubtaskAnnotationAdapterError(RuntimeError):
+    pass
+
+@runtime_checkable
+class JsonPredictor(Protocol):
+
+    def predict(self, payload: Mapping[str, Any]) -> Mapping[str, Any]:
+        ...
+
+def source_payload(source: AnnotationSource) -> dict[str, Any]:
+    return {'source_id': source.source_id, 'location': str(source.location), 'duration_s': source.duration_s, 'fps': source.fps, 'metadata': dict(source.metadata), 'video_streams': [{'camera': stream.camera, 'location': str(stream.location), 'fps': stream.fps, 'width': stream.width, 'height': stream.height, 'metadata': dict(stream.metadata)} for stream in source.video_streams]}

batch_utils/vendor/production_gdn_fix.py

@@ -0,0 +1,2 @@
+def _keep_native_gdn_fp32(model_class: type) -> None:
+    model_class._keep_in_fp32_modules_strict = sorted(set(getattr(model_class, '_keep_in_fp32_modules_strict', None) or ()) | {'A_log', 'dt_bias'})

batch_utils/vendor/pts_input_adapter.py

@@ -0,0 +1,36 @@
+import bisect
+from decimal import Decimal, ROUND_HALF_EVEN
+
+def microseconds(value):
+    return int((Decimal(str(value)) * 1000000).to_integral_value(rounding=ROUND_HALF_EVEN))
+
+def frame_span(pts_s, start_us, end_us):
+    if type(start_us) is not int or type(end_us) is not int or start_us < 0 or (end_us <= start_us):
+        raise ValueError('invalid integer-microsecond source interval')
+    times = [microseconds(t) for t in pts_s]
+    if not times or any((a >= b for a, b in zip(times, times[1:]))):
+        raise ValueError('PTS not strictly increasing after microsecond serialization')
+    return (bisect.bisect_left(times, start_us), bisect.bisect_left(times, end_us))
+
+def native_blocks(pts_s, start_us, end_us, duration_us, rate, contract):
+    if end_us > duration_us:
+        raise ValueError('target beyond source')
+    windows = [('BEFORE', max(0, start_us - 2000000), start_us), ('TARGET', start_us, end_us), ('AFTER', end_us, min(duration_us, end_us + 2000000))]
+    blocks = []
+    for role, a, b in windows:
+        if a == b:
+            continue
+        lo, hi = frame_span(pts_s, a, b)
+        target = role == 'TARGET'
+        factor = contract.TARGET_FRAME_FACTOR if target else contract.CONTEXT_FRAME_FACTOR
+        if hi - lo < factor:
+            if target:
+                raise ValueError('target_has_fewer_than_native_frame_factor')
+            continue
+        ids, receipt = contract.duration_adaptive_sample(lo, hi, source_fps_num=rate.numerator, source_fps_den=rate.denominator, sample_fps=contract.TARGET_SAMPLE_FPS if target else contract.CONTEXT_SAMPLE_FPS, min_frames=contract.TARGET_MIN_FRAMES if target else contract.CONTEXT_MIN_FRAMES, max_frames=contract.TARGET_MAX_FRAMES if target else contract.CONTEXT_MAX_FRAMES, frame_factor=factor)
+        assert len(set(ids)) == len(ids) and all((lo <= i < hi for i in ids))
+        times = [microseconds(pts_s[i]) for i in ids]
+        assert all((a <= t < b for t in times))
+        blocks.append({'role': role, 'interval_us': [a, b], 'start_frame': lo, 'end_frame_exclusive': hi, 'frame_indices': ids, 'actual_pts_us': times, 'sampling_receipt': receipt, 'model_timestamp_clock': 'native ordinal/average FPS; exact source PTS retained separately'})
+    assert sum((len(b['frame_indices']) for b in blocks)) <= contract.MAX_TOTAL_FRAMES
+    return blocks

batch_utils/vendor/robot_caption_geometry.py

@@ -0,0 +1,60 @@
+import hashlib, io
+from PIL import Image
+PACKED_IMAGE_LIMIT_BYTES = 21_866_496
+SCALES = [1.0, 0.9, 0.8, 0.7, 0.6, 0.5, 0.4, 0.3, 0.2]
+
+def sha(b):
+    return hashlib.sha256(b).hexdigest()
+
+def png(im):
+    b = io.BytesIO()
+    im.save(b, format='PNG', compress_level=1, optimize=False)
+    return b.getvalue()
+
+def normalized(plan, frames, expected=None):
+    factors = [expected['common_geometry']['scale']] if expected else SCALES
+    for factor in factors:
+        blocks = []
+        count = 0
+        for block in plan['blocks']:
+            assets = []
+            for start in range(0, len(block['indices']), 2):
+                indices = block['indices'][start:start + 2]
+                originals = [frames[i] for i in indices]
+                w, h = originals[0].size
+                nw = round(w * factor)
+                nh = round(h * factor)
+                packed = Image.new('RGB', (nw * len(indices), nh))
+                rgb = []
+                for pos, image in enumerate(originals):
+                    if factor != 1:
+                        image = image.resize((nw, nh), Image.Resampling.LANCZOS)
+                    packed.paste(image, (pos * nw, 0))
+                    rgb.append(sha(image.tobytes()))
+                blob = png(packed)
+                count += len(blob) + (-len(blob) % 3)
+                assets.append(dict(indices=indices, size=list(packed.size), sha256=sha(blob), rgb_sha256=rgb, bytes=len(blob), _blob=blob))
+            blocks.append(dict(role=block['role'], images=assets))
+        if count <= PACKED_IMAGE_LIMIT_BYTES:
+            break
+    else:
+        raise RuntimeError('No common caption input geometry fits')
+    if expected:
+        for old, new in zip(expected['frame_blocks'], blocks):
+            assert old['role'] == new['role'] and len(old['images']) == len(new['images'])
+            for a, b in zip(old['images'], new['images']):
+                assert a['indices'] == b['indices'] and a['rgb_sha256'] == b['rgb_sha256'] and (a['size'] == b['size'])
+    return dict(plan, frame_blocks=blocks, common_geometry=dict(scale=factor, source_pixel_unchanged=factor == 1, aligned_image_bytes=count, image_size_limit_bytes=PACKED_IMAGE_LIMIT_BYTES, sampling_frames_removed=0))
+
+def source_frames(normalized):
+    out = {}
+    for block in normalized['frame_blocks']:
+        for a in block['images']:
+            image = Image.open(io.BytesIO(a['_blob'])).convert('RGB')
+            n = len(a['indices'])
+            w = image.width // n
+            for pos, i in enumerate(a['indices']):
+                frame = image.crop((pos * w, 0, (pos + 1) * w, image.height))
+                assert sha(frame.tobytes()) == a['rgb_sha256'][pos]
+                out[i] = frame
+    return out

batch_utils/vendor/support/canonical_rgb.py

@@ -0,0 +1,25 @@
+import numpy as np
+from PIL import Image
+
+def canonical_rgb(frame):
+    assert frame.format.name == 'yuv420p'
+    assert frame.width % 16 == 0 and frame.height % 2 == 0
+    assert int(frame.colorspace) in (2, 5, 6), ('unsupported colorspace', frame.colorspace)
+    assert int(frame.color_range) in (0, 1), ('unsupported range', frame.color_range)
+    planes = []
+    for p in frame.planes:
+        a = np.frombuffer(p, dtype=np.uint8).reshape(p.height, p.line_size)[:, :p.width].astype(np.int32)
+        planes.append(a)
+    y, u, v = planes
+    u = u.repeat(2, axis=0).repeat(2, axis=1)
+    v = v.repeat(2, axis=0).repeat(2, axis=1)
+
+    def q13(c):
+        return c * 8192 + 32768 >> 16
+    y = (y * 8 - 128) * q13(65536 * 255 // 219) >> 16
+    u = u * 8 - 1024
+    v = v * 8 - 1024
+    r = y + (v * q13(104597) >> 16)
+    g = y + (u * q13(-25675) >> 16) + (v * q13(-53279) >> 16)
+    b = y + (u * q13(132201) >> 16)
+    return Image.fromarray(np.clip(np.stack([r, g, b], axis=-1), 0, 255).astype(np.uint8))

batch_utils/vendor/support/composers.py

@@ -0,0 +1,26 @@
+from PIL import Image, ImageOps
+from typing import Any
+VIEWS = ('ego', 'left_wrist', 'right_wrist')
+CAPTION_TOP_SIZE = (768, 432)
+CAPTION_WRIST_SIZE = (384, 216)
+CAPTION_MOSAIC_WIDTH = 768
+CAPTION_MOSAIC_HEIGHT = 648
+
+def compose_frame(frames, views):
+    canvas = Image.new('RGB', (640, 720))
+    for view, (x, y, w, h) in zip(VIEWS, [(0, 0, 640, 480), (0, 480, 320, 240), (320, 480, 320, 240)]):
+        if view in views:
+            panel = ImageOps.contain(frames[view].convert('RGB'), (w, h), Image.Resampling.BICUBIC)
+            canvas.paste(panel, (x + (w - panel.width) // 2, y + (h - panel.height) // 2))
+    return canvas
+
+def compose_top_dominant_mosaic(top: Any, left: Any, right: Any) -> Any:
+    from PIL import Image
+    top_panel = top.convert('RGB').resize(CAPTION_TOP_SIZE, Image.Resampling.BICUBIC)
+    left_panel = left.convert('RGB').resize(CAPTION_WRIST_SIZE, Image.Resampling.BICUBIC)
+    right_panel = right.convert('RGB').resize(CAPTION_WRIST_SIZE, Image.Resampling.BICUBIC)
+    output = Image.new('RGB', (CAPTION_MOSAIC_WIDTH, CAPTION_MOSAIC_HEIGHT))
+    output.paste(top_panel, (0, 0))
+    output.paste(left_panel, (0, CAPTION_TOP_SIZE[1]))
+    output.paste(right_panel, (CAPTION_WRIST_SIZE[0], CAPTION_TOP_SIZE[1]))
+    return output

batch_utils/vendor/support/plans.py

@@ -0,0 +1,38 @@
+import bisect, hashlib
+from fractions import Fraction
+import u2_contract as U2
+ROLES = ('main', 'left_wrist', 'right_wrist')
+HEADING = 'Each frame is a synchronized camera mosaic: main camera above, left wrist camera below left, and right wrist camera below right. All panels show the same physical time. Describe each physical operation once across the views.\n\n'
+INSTRUCTION = 'Write one concise, self-contained English instruction for the local operation being attempted in TARGET.\nUse a natural imperative, concrete objects and the working hand; include meaningful support by the other hand and necessary source or destination. Say left hand, right hand or both hands. For a handover, name the object and both the giving and receiving hands.\nInfer the local goal from the video sequence. Use BEFORE and AFTER only to disambiguate TARGET, not to add operations occurring outside it. A failed or repeated attempt retains its established instruction; do not replace the goal with an accident, a success claim or a description of mechanical adjustments. Do not invent an object, hand role or hidden goal.\nOutput only the instruction, on one line, without timestamps, execution-quality commentary or the block names.'
+
+def plans_for_episode(ep, clocks):
+    clock = clocks['main']
+    pts = [float(t * Fraction(clock['time_base'])) for t in clock['ticks']]
+    fps = Fraction(clock['fps'])
+    plans = []
+    for unit in ep['units']:
+        start, end = unit['native_scope_s']
+        assert 0 <= start < end <= clock['duration_s'] + 1e-05
+        blocks = []
+        spans = [('BEFORE', max(0.0, start - 2.0), start), ('TARGET', start, end), ('AFTER', end, min(clock['duration_s'], end + 2.0))]
+        for role, a, b in spans:
+            lo, hi = (bisect.bisect_left(pts, a - 1e-09), bisect.bisect_left(pts, b - 1e-09))
+            hi = min(hi, len(pts))
+            factor = U2.TARGET_FRAME_FACTOR if role == 'TARGET' else U2.CONTEXT_FRAME_FACTOR
+            if hi - lo < factor:
+                if role != 'TARGET':
+                    continue
+                assert hi - lo == 1, (unit['target_id'], 'empty target')
+                indices = [lo, lo]
+                receipt = U2.repeated_single_target_receipt(lo, source_fps_num=fps.numerator, source_fps_den=fps.denominator)
+                padding = [False, True]
+            else:
+                target = role == 'TARGET'
+                indices, receipt = U2.duration_adaptive_sample(lo, hi, source_fps_num=fps.numerator, source_fps_den=fps.denominator, sample_fps=U2.TARGET_SAMPLE_FPS if target else U2.CONTEXT_SAMPLE_FPS, min_frames=U2.TARGET_MIN_FRAMES if target else U2.CONTEXT_MIN_FRAMES, max_frames=U2.TARGET_MAX_FRAMES if target else U2.CONTEXT_MAX_FRAMES, frame_factor=factor)
+                padding = [False] * len(indices)
+            blocks.append(dict(role=role, scope_s=[a, b], source_start_frame=lo, source_end_frame_exclusive=hi, indices=indices, pts_ticks=[clock['ticks'][i] for i in indices], padding_mask=padding, sampling_receipt=receipt))
+        assert any((b['role'] == 'TARGET' for b in blocks))
+        assert sum((len(b['indices']) for b in blocks)) <= U2.MAX_TOTAL_FRAMES
+        prompt = HEADING + '\n\n'.join((b['role'] + ':\n<video>' for b in blocks)) + '\n\n' + INSTRUCTION
+        plans.append(dict(target_id=unit['target_id'], unit_id=unit['unit_id'], episode_id=ep['episode_id'], source_id=ep['source_id'], group=ep['group'], native_scope_s=unit['native_scope_s'], camera_roles=list(ROLES), clock_fps=str(fps), time_base=clock['time_base'], blocks=blocks, prompt=prompt, prompt_sha256=hashlib.sha256(prompt.encode()).hexdigest(), context_policy='common_real_up_to_2s_each_side', source_clock_origin=clock['first_pts'], do_sample_frames=False))
+    return plans

batch_utils/vendor/support/u2_contract.py

@@ -0,0 +1,49 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+from fractions import Fraction
+TARGET_SAMPLE_FPS = 8
+TARGET_MIN_FRAMES = 8
+TARGET_MAX_FRAMES = 40
+TARGET_FRAME_FACTOR = 2
+CONTEXT_SAMPLE_FPS = 8
+CONTEXT_MIN_FRAMES = 4
+CONTEXT_MAX_FRAMES = 12
+CONTEXT_FRAME_FACTOR = 4
+MAX_TOTAL_FRAMES = 64
+SAMPLER_VERSION = 'duration_adaptive_uniform_bin_centers_nested_context_v2'
+SHORT_TARGET_POLICY = 'repeat_single_source_frame_to_temporal_pair_v1'
+
+class U2ContractError(ValueError):
+    pass
+
+def uniform_bin_centers(start: int, end_exclusive: int, count: int) -> list[int]:
+    length = end_exclusive - start
+    if count <= 0:
+        return []
+    if length < count:
+        raise U2ContractError(f'span [{start}, {end_exclusive}) has fewer than {count} frames')
+    result = [start + (2 * index + 1) * length // (2 * count) for index in range(count)]
+    if len(result) != len(set(result)) or result[0] < start or result[-1] >= end_exclusive:
+        raise AssertionError('uniform frame sampler violated its contract')
+    return result
+
+def duration_adaptive_sample(start: int, end_exclusive: int, *, source_fps_num: int, source_fps_den: int, sample_fps: int, min_frames: int, max_frames: int, frame_factor: int) -> tuple[list[int], dict[str, int | bool | str]]:
+    available = end_exclusive - start
+    if source_fps_num <= 0 or source_fps_den <= 0 or available <= 0:
+        raise U2ContractError('source FPS and frame span must be positive')
+    if min_frames % frame_factor or max_frames % frame_factor:
+        raise U2ContractError('adaptive frame bounds must align to frame factor')
+    duration = Fraction(available * source_fps_den, source_fps_num)
+    requested = min(Fraction(max_frames), max(Fraction(min_frames), duration * sample_fps))
+    desired_count = requested.numerator // (requested.denominator * frame_factor) * frame_factor
+    actual_count = min(desired_count, available // frame_factor * frame_factor)
+    if actual_count < frame_factor:
+        raise U2ContractError(f'span [{start}, {end_exclusive}) has fewer than {frame_factor} frames')
+    indices = uniform_bin_centers(start, end_exclusive, actual_count)
+    return (indices, {'policy': 'duration_fps_bounded_unique_frames_v1', 'sampler_version': SAMPLER_VERSION, 'source_fps_num': source_fps_num, 'source_fps_den': source_fps_den, 'sample_fps_num': sample_fps, 'sample_fps_den': 1, 'min_frames': min_frames, 'max_frames': max_frames, 'frame_factor': frame_factor, 'available_unique_source_frames': available, 'desired_frame_count': desired_count, 'actual_frame_count': actual_count, 'source_limited': actual_count < desired_count, 'selection': 'uniform_bin_centers'})
+
+def repeated_single_target_receipt(frame_index: int, *, source_fps_num: int, source_fps_den: int) -> dict[str, int | str]:
+    if frame_index < 0 or source_fps_num <= 0 or source_fps_den <= 0:
+        raise U2ContractError('single-target fallback received invalid source metadata')
+    return {'policy': SHORT_TARGET_POLICY, 'source_fps_num': source_fps_num, 'source_fps_den': source_fps_den, 'source_frame_index': frame_index, 'available_unique_source_frames': 1, 'actual_frame_count': TARGET_FRAME_FACTOR, 'repeated_frame_count': TARGET_FRAME_FACTOR}

batch_utils/vendor/tempseg_fp32_loader.py

@@ -0,0 +1,51 @@
+from collections import Counter
+import hashlib
+from pathlib import Path
+import types
+
+def digest(path):
+    with Path(path).open('rb') as stream:
+        return hashlib.file_digest(stream, 'sha256').hexdigest()
+
+def load(spec, save):
+    import torch
+    from transformers import AutoModelForImageTextToText
+    from transformers.models.qwen3_5.modeling_qwen3_5 import Qwen3_5PreTrainedModel
+    from safetensors.torch import load_file
+    from peft import PeftConfig, get_peft_model, get_peft_model_state_dict, set_peft_model_state_dict
+    from production_gdn_fix import _keep_native_gdn_fp32
+    from verify_gdn import verify
+    _keep_native_gdn_fp32(Qwen3_5PreTrainedModel)
+    base, info = AutoModelForImageTextToText.from_pretrained(spec['base'], local_files_only=True, dtype=torch.bfloat16, attn_implementation='sdpa', output_loading_info=True)
+    assert not any((info.get(k) for k in ['missing_keys', 'unexpected_keys', 'mismatched_keys', 'error_msgs'])), info
+    path = Path(spec['export']) / 'adapter_model.safetensors'
+    assert digest(path) == spec['adapter_sha256']
+    source = load_file(str(path), device='cpu')
+    assert len(source) == 496 and sum((v.numel() for v in source.values())) == 86556672
+    assert all((v.dtype == torch.float32 for v in source.values())), 'Expected corrected FP32 checkpoint'
+    model = get_peft_model(base, PeftConfig.from_pretrained(spec['export'], local_files_only=True), adapter_name='candidate', autocast_adapter_dtype=True)
+    initial = get_peft_model_state_dict(model, adapter_name='candidate')
+    assert set(initial) == set(source)
+    assert all((v.dtype == torch.float32 for v in initial.values())), 'PEFT must create FP32 adapter holders'
+    del initial
+    load_info = set_peft_model_state_dict(model, source, adapter_name='candidate')
+    assert not load_info.unexpected_keys, load_info
+    actual = get_peft_model_state_dict(model, adapter_name='candidate')
+    assert set(actual) == set(source)
+    assert all((actual[k].dtype == v.dtype and torch.equal(actual[k].detach().cpu(), v) for k, v in source.items())), 'Saved FP32 LoRA values must be loaded exactly'
+    del actual
+    model.to('cuda:0').eval()
+    actual = get_peft_model_state_dict(model, adapter_name='candidate')
+    assert all((actual[k].dtype == torch.float32 and torch.equal(actual[k].detach().cpu(), v) for k, v in source.items())), 'Final device placement must preserve saved FP32 values'
+    save('ADAPTER_VALUE_VERIFICATION.json', dict(tensors=496, elements=86556672, source_dtypes=dict(Counter((str(v.dtype) for v in source.values()))), holders_fp32_before_assignment=True, all_saved_values_exact_before_and_after_placement=True, adapter_sha256=spec['adapter_sha256'], merged=False))
+    del source, actual
+    gdn = verify(model, spec['base'], peft_prefix=True)
+    save('GDN_VALUE_VERIFICATION.json', gdn)
+    original_generate = model.generate
+
+    def generate_bf16(self, *args, **kwargs):
+        with torch.autocast(device_type='cuda', dtype=torch.bfloat16):
+            return original_generate(*args, **kwargs)
+    model.generate = types.MethodType(generate_bf16, model)
+    save('MODEL_LOADED.json', dict(adapter_persistent_dtype='torch.float32', compute_autocast='torch.bfloat16', gdn_values_exact=True, base_loading_info=info, parameter_dtypes=dict(Counter((str(p.dtype) for p in model.parameters()))), input_output_generation_policy_changed=False))
+    return model

batch_utils/vendor/verify_gdn.py

@@ -0,0 +1,36 @@
+from collections import Counter, defaultdict
+import hashlib, json
+from pathlib import Path
+
+def verify(model, path, *, peft_prefix=False):
+    import torch
+    from safetensors import safe_open
+    path = Path(path)
+    mapping = json.loads((path / 'model.safetensors.index.json').read_text())['weight_map']
+    wanted = {name for name in mapping if name.endswith(('.A_log', '.dt_bias'))}
+    assert wanted
+    actual = {}
+    for name, p in model.named_parameters():
+        if not name.endswith(('.A_log', '.dt_bias')):
+            continue
+        if peft_prefix:
+            assert name.startswith('base_model.model.')
+            name = name[len('base_model.model.'):]
+        assert name not in actual
+        actual[name] = p
+    assert set(actual) == wanted, {'missing': sorted(wanted - set(actual)), 'extra': sorted(set(actual) - wanted)}
+    groups = defaultdict(list)
+    for name in wanted:
+        groups[mapping[name]].append(name)
+    evidence = []
+    for filename, names in sorted(groups.items()):
+        with safe_open(str(path / filename), framework='pt', device='cpu') as source:
+            for name in sorted(names):
+                saved = source.get_tensor(name)
+                p = actual[name]
+                assert p.dtype == torch.float32, (name, str(p.dtype), 'must preserve FP32 before loading')
+                assert p.device.type == 'cuda', (name, 'must verify final device placement')
+                value = p.detach().cpu()
+                assert torch.equal(value, saved.float()), (name, 'checkpoint bits changed; upcasting rounded BF16 is insufficient')
+                evidence.append(dict(name=name, elements=p.numel(), source_dtype=str(saved.dtype), runtime_dtype=str(p.dtype), device=str(p.device), fp32_value_sha256=hashlib.sha256(value.contiguous().view(torch.uint8).numpy().tobytes()).hexdigest(), saved_value_exact=True))
+    return dict(status='PASS_NATIVE_GDN_FP32_CHECKPOINT_VALUES', tensors=len(evidence), source_dtypes=dict(Counter((r['source_dtype'] for r in evidence))), all_runtime_fp32=True, all_values_equal_saved_checkpoint=True, checked_after_final_device_placement=True, values=evidence)

infer.py

@@ -0,0 +1,5 @@
+"""Run the complete paired Data Engine workflow from either model repository."""
+from batch_utils.cli import main
+
+if __name__ == "__main__":
+    raise SystemExit(main())