# Take Model Audit (CP-7 Phase 1)

Generated: 2026-04-28T10:59:19Z
Predecessor: post-cp6-workflow-object-model | Rollback: pre-cp7-take-model

## What CP-7 introduces
- pipeline/core/take.py — Take + Beat + Scene dataclasses
- Take wraps one Workflow; Take.execute runs + compresses step status
- Beat groups Takes; new_take + select_primary
- Scene = thin wrapper around list[Beat]
- JSON round-trip for all three
- NO persistence / NO production migration / NO eval / NO DIRECTOR subtypes

## Existing take/beat survey + CP-6 hand-off verification
Machine-readable: cp7_phase1_take_candidates.json + cp7_phase1_handoff_verification.json

## Frozen contracts (do not touch)
dispatch.py / receipts.py / dispatch_context.py / registry.py / workflow.py /
execution/step_runner.py / step_types.py / provider adapters / sidecar provenance.prompt (CP-3 lock)

## Tag verification
```
4c012ab5946d3a4ce3064d8c38ba5f4f5bfa9706
c55696bbbb7ee9b128755d76d1157e6f7fe1a507
0f3893b24fdf75beb25212f2065792bddff118e6
```

---

## Post-CP-7 summary

Phase 7 ships documentation only (no code, no tests). This section freezes the
shipped surface and records what was deliberately deferred.

### Shipped schema (frozen at CP-7)

All three classes live in `recoil/pipeline/core/take.py` and are re-exported
from `pipeline.core` via `pipeline/core/__init__.py`.

```
TakeStatus      = Literal["pending", "running", "succeeded", "failed", "partial"]
PrimaryStrategy = Literal["first_success", "manual", "score"]
```

**`Take` (dataclass)**
- `take_id: str` — caller-assigned, unique within a Beat.
- `take_index: int` — auto-assigned by `Beat.new_take` (0-based, append order).
- `workflow: Workflow` — exactly one CP-6 Workflow. Take wraps Workflow 1:1.
- `status: TakeStatus = "pending"` — set to `"running"` for the duration of
  `execute()`, then compressed to `succeeded` / `failed` / `partial` based on
  the underlying step statuses.
- `created_at: str` (ISO-8601 UTC) — `field(default_factory=utc_now_iso8601)`.
- `take_metadata: dict[str, Any]` — free-form caller metadata.

`Take.execute(*, context, pre_step=None, post_step=None, on_failure=None)` runs
the wrapped Workflow via `Workflow.run` and compresses step status into the
take-level `status` field. Hooks (`pre_step` / `post_step` / `on_failure`)
pass through to `Workflow.run` unchanged. There are NO take-level hooks in
CP-7 (no `pre_take` / `post_take`).

Status compression rule (`_compress_step_status`):
- all steps `"succeeded"` → `"succeeded"`
- no step `"succeeded"` → `"failed"`
- mixed (some succeeded, some failed/skipped) → `"partial"`

**`Beat` (dataclass)**
- `beat_id: str`
- `takes: list[Take]` — append-only ordered list.
- `primary_take_id: Optional[str]` — set by `select_primary` or by caller
  under `strategy="manual"`.
- `beat_metadata: dict[str, Any]`
- `created_at: str` (ISO-8601 UTC).

Methods:
- `Beat.new_take(workflow, *, take_id=None, take_metadata=None) -> Take` —
  constructs a `Take` with auto-assigned `take_index = len(self.takes)`,
  appends to `self.takes`, returns it. Re-attempts always create a new
  Take; existing Takes are never mutated.
- `Beat.select_primary(strategy="first_success") -> Optional[Take]` — sets
  and returns `primary_take_id`'s Take. `"first_success"` picks the first
  take whose `status == "succeeded"`. `"manual"` is a no-op (caller already
  set `primary_take_id`). `"score"` raises `NotImplementedError` (gated on
  CP-9 eval primitives).
- `Beat.primary_take -> Optional[Take]` — property lookup by
  `primary_take_id`.

**`Scene` (dataclass)**
- `scene_id: str`
- `beats: list[Beat]`
- `scene_metadata: dict[str, Any]`
- `created_at: str` (ISO-8601 UTC).

Scene is a thin grouping of Beats — dataclass + `to_dict` / `from_dict`
serialization only. No execution methods.

All three classes JSON round-trip via `to_dict` / `from_dict`.
`from_dict(to_dict(obj))` is value-equal to `obj` for a fully-populated
Take, Beat, or Scene (verified in `test_take_serialization.py` and
`test_take_beat_poc::test_take_beat_poc_round_trip_serialization`).

CP-7 is in-memory only. No disk persistence is shipped.

### Phase 5 POC tests

`recoil/pipeline/core/tests/test_take_beat_poc.py` ships 6 POC tests that
collectively prove the Take/Beat re-attempt idiom over a 2-step
keyframe→video Workflow:

1. `test_take_beat_poc_happy_path` — single Take with all steps succeeding
   yields `Take.status == "succeeded"` and `Beat.select_primary()` picks
   it as primary.
2. `test_take_beat_poc_first_take_fails_second_succeeds` — re-attempt
   semantics: when take_0 fails, a fresh take_1 is created via
   `Beat.new_take`, `select_primary("first_success")` skips the failed
   take and picks take_1.
3. `test_take_beat_poc_partial_take_not_chosen` — `"first_success"`
   strategy ignores `partial` Takes; only `succeeded` qualifies.
4. `test_take_beat_poc_re_attempt_idiom` — verifies that re-attempting
   does NOT mutate the prior Take (immutable history); each attempt is
   a new Take appended to `Beat.takes` with monotonic `take_index`.
5. `test_take_beat_poc_round_trip_serialization` — full Beat (with
   multiple Takes containing executed Workflows + receipts) survives
   `to_dict` → `from_dict` round-trip with value equality.
6. `test_take_beat_poc_scene_groups_beats` — Scene groups multiple
   Beats; serialization round-trips through Scene → list[Beat] →
   list[list[Take]] without identity loss.

These six tests are the integration-level proof that Take + Beat compose
correctly over a real Workflow (not just unit-level dataclass behavior).
Unit-level coverage lives in `test_take.py`, `test_beat.py`,
`test_scene.py`, `test_take_execute.py`, `test_beat_primary.py`,
`test_take_serialization.py`, plus end-to-end `test_take_e2e_scenarios.py`.

### Deferred items (NOT shipped in CP-7)

CP-7 deliberately does not ship:

- **DIRECTOR step subtypes** — `CinematographyStep`, `StoryBeatStep`,
  `ShotConfigStep`, `ExpressionPoseStep`. Carried from CP-6 hand-off.
  Decision (subclassing vs. payload tagging) deferred to a later CP.
- **Disk persistence** — CP-7 is in-memory only. `to_dict` / `from_dict`
  exist for tests/debug, but no save/load to disk. Future persistence CP
  will pick a directory layout, sidecar format, and load semantics.
- **Take-level hooks** — `pre_take` / `post_take` are NOT implemented.
  Only step-level hooks (`pre_step` / `post_step` / `on_failure`) pass
  through to `Workflow.run`. Take-level hooks are deferred until a
  concrete caller needs them (likely CP-9 evals).
- **Score-based primary selection** — `Beat.select_primary(strategy="score")`
  raises `NotImplementedError`. Implementation gated on CP-9, which fills
  `eval_scores` on receipts and provides the primary-selection default
  replacement.
- **Production migration target** — CP-7 ships test-only POCs (Phase 5).
  No production callsite has been migrated to use Take/Beat. The
  migration target (likely `pipeline/orchestrator/pipeline.py` flat-runner
  sequences or a new shot-driver) is deferred to a dedicated post-CP-9
  CP, after eval primitives exist.

### Production migration candidates (carried from CP-6 hand-off — still deferred)

Listed verbatim from CP-6→CP-7 hand-off (`consultations/recoil/cp6-workflow-spec/CP7_HANDOFF.md`).
CP-7 makes no new architectural decisions on these — they remain open:

1. `pipeline/orchestrator/pipeline.py` flat-runner sequences (sequential
   keyframe→video pairs that fit the Take shape).
2. A new shot-driver that owns both keyframe and video phases for a
   single shot.
3. NOT `production_loop.py` — its keyframe and video dispatches are in
   separate methods called from separate orchestrator-loop iterations
   (verified during CP-6 spec review; not a sequential pair).
4. R2V / multi-shot bypass paths (CP-5 deferred): `dispatch_cli.py:893`
   (`runner.execute_pass()`), `:1576` (`runner.execute_wan_r2v()`),
   `:1281`/`:1364` (`runner.execute_multi_shot()`).
5. Bulk migration of remaining dispatch sites where shape fits (37
   dispatch sites cataloged in CP-6 Phase 1).

### Rollback procedure

CP-7 introduces only additive code (new file `pipeline/core/take.py` +
re-exports in `pipeline/core/__init__.py`) and 7 new test files. No frozen
contract was modified. To roll back:

```
git checkout pre-cp7-take-model -- recoil/pipeline/core/take.py recoil/pipeline/core/__init__.py
git rm recoil/pipeline/core/tests/test_take.py \
       recoil/pipeline/core/tests/test_beat.py \
       recoil/pipeline/core/tests/test_scene.py \
       recoil/pipeline/core/tests/test_take_execute.py \
       recoil/pipeline/core/tests/test_beat_primary.py \
       recoil/pipeline/core/tests/test_take_serialization.py \
       recoil/pipeline/core/tests/test_take_beat_poc.py
```

(`test_take_e2e_scenarios.py`, added in Phase 6, is also a new file and
should be `git rm`-ed alongside the seven above when rolling back the full
CP-7 surface.)

The rollback restores `pipeline/core/take.py` to its absent state at the
`pre-cp7-take-model` tag and removes the public re-exports from
`pipeline/core/__init__.py`. After rollback, run the full pytest suite to
confirm clean state (expected count: 1188 — the pre-CP-7 baseline).

### data-contracts.md edit — DEFERRED pending JT sign-off

Per spec scope boundary (`BUILD_SPEC_CP7.md` §"Scope boundary"), the
suggested `data-contracts.md` paragraph is NOT applied in Phase 7. It is
quoted here verbatim and will be carried into the CP-8 hand-off for JT
sign-off:

> **Take + Beat + Scene (CP-7).** A `Take` wraps exactly one `Workflow`
> (CP-6). `Take.execute` runs the workflow and compresses step status into
> a take-level status (`succeeded` / `failed` / `partial` / `pending` /
> `running`). A `Beat` groups multiple Takes for one logical shot;
> `Beat.new_take` auto-assigns take_index;
> `Beat.select_primary(strategy="first_success")` picks the primary.
> `Scene` is a thin grouping of Beats. CP-7 is in-memory only — JSON
> round-trip via `to_dict` / `from_dict`. No new sidecar provenance keys.
> Receipts continue to be addressed by `(workflow_id, workflow_step_id)`;
> CP-7 introduces no new dispatch-side identity. CP-8 adds audio Takes.
> CP-9 fills `eval_scores` and replaces the primary-selection default
> with score-based logic.

Apply when JT signs off.