# Orchestration — the converged production path: plan shots → grouping → persisted
# Scene-of-Beats → per-Beat take loop → dispatch. EpisodeRunner is THE live entry
# (replaces ProductionLoop / run_episode.py / ClientSequenceRunner; ssot_manifest
# orchestration, canonical EpisodeRunner.run_episode_batches).
#
# Model: Scene → Beat → Take (scene_beat_take). A Scene is a thin grouping of Beats;
# a Beat groups Takes for one logical shot/batch and selects primary_take_id; a Take
# wraps exactly one Workflow (workflow_take_model). EpisodeRunner materializes this
# from shot batches and persists each Scene to
# _pipeline/state/orchestration/scenes/{episode}_{scene_id}.json.
#
# THE central divergence: the grouping STRATEGY REGISTRY forks both the persisted
# scene-id namespace (continuity→BATCH_NNN, oner→ONER_NNN, coverage→pass_id) AND the
# dispatch modality (r2v_multi vs video_i2v fallback). batch_selector re-maps a
# CONT/ONER selector back to the SAME persisted scene file the board/r2v read.
# Capability state lives in ssot_manifest.yaml (orchestration / scene_beat_take /
# workflow_take_model), not here.
subsystem: orchestration

capabilities:
  - id: orchestration
    reduced_tier: always
    risk_level: critical
  - id: scene_beat_take
    reduced_tier: always
    risk_level: high
  - id: workflow_take_model
    reduced_tier: auto
    risk_level: medium

entrypoints:
  - id: generate_cli
    file: recoil/pipeline/cli/generate.py
    symbol: main
    invoke: "python3 recoil/pipeline/cli/generate.py <project> <ep> --grouping {coverage|continuity|oner|auto} [--pass <id>] [--batch EP{ep}_{CONT|ONER}_{ord}] [--new-take] [--dry-run]"
    kind: cli
    summary: >
      Live production entry. Loads the plan, builds ExecutionStore + StepRunner,
      cold-starts LearningEngine→StrategyEngine, constructs EpisodeRunner, acquires a
      lockfile, then asyncio.run(run_episode_batches()). The --batch path BYPASSES
      coverage-pass loading + plan reassembly and dispatches a single persisted scene
      directly (not run_episode_batches).
    capabilities: [orchestration, scene_beat_take]
    produces: [persisted_scene, ops_log]
    reduced_tier: always
    risk_level: critical

  - id: episode_runner_module
    file: recoil/pipeline/orchestrator/episode_runner.py
    symbol: EpisodeRunner.run_episode_batches
    invoke: "from recoil.pipeline.orchestrator.episode_runner import EpisodeRunner; await EpisodeRunner(...).run_episode_batches(...)"
    kind: module
    summary: >
      Clusters shots into groups via the grouping registry, converts each Group to a
      Scene of Beats, runs Beats under an asyncio.Semaphore, performs stale-take
      recovery + phantom-success invalidation + current-ref fingerprint revalidation
      before dispatch, and persists each Scene through the guarded writer.
    capabilities: [orchestration, scene_beat_take, workflow_take_model]
    produces: [persisted_scene]
    reduced_tier: always
    risk_level: critical

  - id: reroll_route
    file: recoil/pipeline/api/routes/reroll.py
    symbol: reroll
    invoke: "POST /reroll {project, episode, batch_id: EP{ep}_{CONT|ONER}_{ord}}"
    kind: tool
    summary: >
      HTTP reroll surface. Parses the batch selector to the SAME persisted scene id
      as --batch (batch_selector), board-gate pre-scans BEFORE any mutating call,
      then prepare_beat_for_reroll clears a stale primary and dispatches one new take.
    capabilities: [orchestration, scene_beat_take]
    produces: [persisted_scene]
    reduced_tier: auto
    risk_level: high

routes:
  - id: grouping_coverage
    capability: orchestration
    surface: coverage
    entry: recoil/pipeline/_lib/grouping.py::_assemble_coverage
    resolves_via: recoil/pipeline/orchestrator/coverage_planner.py::build_passes
    ref_kind: coverage_pass
    gated_by: grouping_strategy
    fallback_route: null
    risk_level: high

  - id: grouping_continuity
    capability: orchestration
    surface: continuity
    entry: recoil/pipeline/_lib/grouping.py::_assemble_continuity
    resolves_via: recoil/pipeline/_lib/scene_clusterer.py::cluster_shots_into_batches
    ref_kind: batch
    gated_by: grouping_strategy
    fallback_route: grouping_continuity_per_shot   # below_threshold batch → video_i2v
    risk_level: high

  - id: grouping_continuity_per_shot
    capability: orchestration
    surface: continuity
    entry: recoil/pipeline/_lib/grouping.py::_assemble_continuity
    resolves_via: recoil/pipeline/_lib/scene_clusterer.py::cluster_shots_into_batches
    ref_kind: per_shot_i2v          # Batch.below_threshold (len < min_batch_size=3)
    gated_by: grouping_strategy
    fallback_route: null
    risk_level: medium

  - id: grouping_oner
    capability: orchestration
    surface: oner
    entry: recoil/pipeline/_lib/grouping.py::_assemble_oner
    resolves_via: recoil/pipeline/_lib/grouping.py::_split_oner_scene_for_caps
    ref_kind: oner                  # 1 shot → video_i2v, >1 → r2v_multi
    gated_by: grouping_strategy
    fallback_route: null
    risk_level: high

  - id: batch_reroll_selector
    capability: scene_beat_take
    surface: reroll
    entry: recoil/pipeline/orchestrator/batch_selector.py::parse_batch_selector
    resolves_via: recoil/pipeline/orchestrator/batch_selector.py::verify_scene_grouping_metadata
    ref_kind: persisted_scene_id    # CONT→BATCH_NNN, ONER→ONER_NNN
    gated_by: null
    fallback_route: null
    risk_level: high

flags:
  - id: grouping_strategy
    defined: recoil/pipeline/_lib/grouping.py::get_grouping
    default: coverage
    forks_capability: orchestration
    gates_routes: [grouping_coverage, grouping_continuity, grouping_oner]
    summary: >
      --grouping {coverage|continuity|oner|solo|auto} selects a GROUPING_REGISTRY
      assembler. Each forks the persisted scene-id namespace AND the modality
      (coverage→pass_id/r2v_multi; continuity→BATCH_NNN/r2v_multi|video_i2v;
      oner→ONER_NNN). The default production strategy is coverage.

  - id: force_new_take
    defined: recoil/pipeline/cli/generate.py::main
    default: false
    forks_capability: scene_beat_take
    gates_routes: [batch_reroll_selector]
    summary: >
      --new-take requires exactly one r2v_multi batch target and reroll grouping
      derived from the primary take; --batch requires --new-take; --retry is
      incompatible with --batch.

loops:
  - id: beat_take_retry
    kind: retry
    capability: scene_beat_take
    trigger: "Take fails (or no primary_take yet) inside _dispatch_one_beat; loop while not beat.is_exhausted and not beat.approved and not strategy_exhausted."
    driver: recoil/pipeline/orchestrator/episode_runner.py::EpisodeRunner._dispatch_one_beat
    bound:
      max: 3
      source: "EpisodeRunner.max_takes (default 3; --max-takes CLI override propagates to existing scenes on resume)"
    exit:
      - "beat.approved or beat.is_exhausted (max_takes reached → max_takes_reached)"
      - "StrategyEngine strategy set exhausted (_strategy_exhausted)"
      - "BudgetExhaustedError (budget_guard.would_exceed)"
    retries_into: grouping_coverage
    reads_artifact: persisted_scene
    reduced_tier: always
    risk_level: high

  - id: strategy_retry
    kind: strategy
    capability: workflow_take_model
    trigger: "With a StrategyEngine injected, a failed Take is classified → StrategyEngine.select_and_apply → StrategyDiff mutates the next workflow. Without StrategyEngine retries are blind (same workflow)."
    driver: recoil/pipeline/orchestrator/episode_runner.py::EpisodeRunner._dispatch_one_beat
    bound:
      max: 3
      source: "shares max_takes with beat_take_retry; strategy set size also bounds it"
    exit:
      - "strategy set exhausted (_strategy_exhausted)"
      - "Take succeeds and becomes primary"
    retries_into: beat_take_retry
    reduced_tier: auto
    risk_level: medium

  - id: dispatch_retry_backoff
    kind: retry
    capability: orchestration
    trigger: "A dispatched shot fails; classify_failure decides retryable vs permanent, then queue with exponential backoff."
    driver: recoil/pipeline/orchestrator/retry_dispatcher.py::RetryDispatcher.dispatch
    bound:
      max: "per-shot attempt count vs policy max"
      source: "RetryDispatcher retry budget (per-shot attempts vs policy)"
    exit:
      - "classify_failure → permanent (returns None)"
      - "retry budget exhausted → permanent"
    reduced_tier: auto
    risk_level: medium

artifacts:
  - id: persisted_scene
    path: "_pipeline/state/orchestration/scenes/{episode}_{scene_id}.json   (e.g. ep_001_BATCH_004.json, ep_001_ONER_002.json, ep_001_<pass_id>.json)"
    written_by: [episode_runner_module, generate_cli, reroll_route]
    read_by: [batch_reroll_selector, board_ref_path, video_ref_path]
    schema: "Scene{beats[].beat_metadata: {scene_id, modality, grouping{strategy,ordinal}, shot|batch_shots[](CanonicalShot dicts w/ location_id+scene_index), inputs_fingerprint}}"
    reduced_tier: always
    risk_level: critical

# ops_log artifact declared once in retry_loops.topo.yaml (richer ndjson schema);
# this subsystem also writes batches_planned events to it (entrypoint `produces`).

divergences:
  - id: grouping_strategy_fork
    capability: orchestration
    forks: [grouping_coverage, grouping_continuity, grouping_oner]
    on_flag: grouping_strategy
    invariant: >
      The grouping STRATEGY chosen at run time forks BOTH the persisted scene-id
      namespace AND the dispatch modality, and the resulting {episode}_{scene_id}.json
      is the SSOT the board + r2v re-read. A scene persisted under one strategy MUST
      be re-derived (or --batch-targeted) under the SAME strategy: continuity scenes
      are named BATCH_NNN but carry grouping.strategy='continuity' (the CONT selector
      token), and the per-strategy ordinal must round-trip through
      verify_scene_grouping_metadata before any reroll spend.
    risk: >
      An engineer adding a grouping strategy or rerolling under the wrong --grouping
      can mismatch the selector→scene mapping (CONT→BATCH vs ONER→ONER), feed the
      wrong persisted beat to dispatch, or silently flip modality (r2v_multi↔video_i2v)
      and lose the [Xs-Ys] batch timestamp annotations.
    reduced_tier: always
    risk_level: critical

  - id: batch_modality_fork
    capability: scene_beat_take
    forks: [grouping_continuity, grouping_continuity_per_shot]
    on_flag: null
    invariant: >
      A continuity Batch with len(shots) < min_batch_size (3) is flagged
      below_threshold and MUST dispatch as per-shot video_i2v, not r2v_multi; the
      same is true for a 1-shot oner. cluster_shots_into_batches owns this decision —
      callers MUST NOT force a below-threshold batch onto r2v_multi.
    risk: >
      single_batch_from_shots deliberately DEFEATS the below_threshold/heuristic
      splitting (a coverage pass is one billed Flora run regardless of shot count),
      so the two clusterers diverge on whether to fall back. Mixing them — using
      cluster_shots_into_batches where a deliberate pass-as-batch was intended, or
      vice versa — silently changes batching, billing, and modality.
    reduced_tier: always
    risk_level: high

  - id: scene_version_append
    capability: orchestration
    forks: [grouping_coverage, batch_reroll_selector]
    invariant: >
      The sole RE-DERIVATION writer APPENDS a new structure-immutable
      ep_NNN_BATCH_NNN.v{N+1}.json body (a candidate version) in the per-batch
      ep_NNN_BATCH_NNN.versions.json manifest; it MUST NOT rewrite the shot STRUCTURE
      of any existing version and MUST NOT move active_version. In-place take/board
      STATUS updates to the ACTIVE version body remain allowed (the existing save_scene
      caller surface); only SceneVersionStore.conform/revert moves the pointer. Every
      prior version's structure stays intact and re-pointable.
    risk: >
      A re-derivation writer that OVERWRITES an existing version's shot structure, or
      any code that moves the pointer outside conform/revert, silently destroys
      approved shot structure — the exact loss this design prevents.
    reduced_tier: always
    risk_level: high

symbols:
  - id: episode_runner_dispatch_one_beat
    file: recoil/pipeline/orchestrator/episode_runner.py
    symbol: EpisodeRunner._dispatch_one_beat
    role: "The take-loop junction — owns max_takes, StrategyEngine mutation, budget gate, and primary_take selection; where Scene→Beat→Take becomes dispatch."
    reduced_tier: always
    risk_level: high

  - id: scene_from_group
    file: recoil/pipeline/orchestrator/episode_runner.py
    symbol: EpisodeRunner._scene_from_group
    role: "Group→Scene materialization: stamps beat_metadata.grouping + batch_shots (CanonicalShot dicts w/ location_id+scene_index) that the board/r2v and fingerprint revalidation read back."
    reduced_tier: always
    risk_level: high

  - id: get_grouping
    file: recoil/pipeline/_lib/grouping.py
    symbol: get_grouping
    role: "GROUPING_REGISTRY dispatch — the single seam that forks the persisted scene namespace + modality per strategy."
    reduced_tier: always
    risk_level: high
