# BUILD_SPEC — Console v2 Product Completion

**Generated:** 2026-05-12 (revised by Opus 4.7 after Sonnet draft + audit)
**Input:** `consultations/recoil/console-v2-product-completion-2026-05-12/SYNTHESIS.md`
**Detail level:** max
**Visual design:** none (backend + one prompt_compiler patch)
**Phases:** 5 (0, 1, 1.5b, 2, 3)

---

## Synthesis Deviations (read before harness dispatch)

This BUILD_SPEC deviates from SYNTHESIS in three deliberate ways:

1. **SYNTHESIS Phase 0 SSOT-dual-route migration: REJECTED.** SYNTHESIS decision #8 prescribes killing `/api/chat/proposals/{id}/approve` and migrating to `mutation_routes.py`. Codebase audit (2026-05-12) shows these are two separate proposal systems with incompatible ID formats (UUID hex vs `prop_NNN`) and different storage backends. Migrating breaks ProposalTray. This spec instead wires the executor into the existing `proposals_routes.py:approve_proposal()` path, replacing the 501 with real execution. A1/A2/A3 cleanup items in SYNTHESIS Phase 0 are also verified already done in the codebase.

2. **SYNTHESIS Phase 1.5a (Edit Prompt UI Button): CUT.** SYNTHESIS justified it as a hedge against parent-take-mcp not being ready. parent-take-mcp merged 2026-05-12. The MCP `create_proposal` tool (Phase 1.5b) is the proposal-origination surface. No "beat detail view" exists in Console v2 and JT does not want one built.

3. **SYNTHESIS Phase 4 (remaining ProposalKinds): DEFERRED to orchestrated follow-on.** Phase 4 contained "TBD" markers (`RetryStrategyEditProposal` location, `MultiBeatDirectiveProposal` diff schema). Maximum-detail specs don't ship with TBDs. After this build verifies green, a separate orchestration round will: read what shipped, run a 1-round Opus consult on remaining kinds, generate a Phase 4 BUILD_SPEC, dispatch. See "Post-build orchestration" at the bottom of this file.

4. **`shot["prompt_override"]` read-path added to Phase 1.** SYNTHESIS Risk #1 called this out as CRITICAL silent-no-op risk. Audit (2026-05-12) confirmed `recoil/core/prompt_compiler.py:compile()` never reads `shot["prompt_override"]`. Phase 1 now wires the read path as an early-return bypass at the top of `compile()`. Video pipeline (`recoil/pipeline/_lib/prompt_engine.py`) is unchanged in this build — flagged for follow-on.

---

## Dependency Graph

```
Phase 0 (Smoke Test Foundation + Kind Field): none
Phase 1 (PromptRewrite Executor + Compiler Read Path): depends_on 0
Phase 1.5b (create_proposal MCP Tool): depends_on 1 (parent-take-mcp already merged)
Phase 2 (ParamTweak Executor): depends_on 1
Phase 3 (ScriptEdit Executor): depends_on 1
```

---

## CODEBASE STATE AT SPEC TIME (read before each phase)

**Already done — do NOT re-implement:**
- A1 (import cycle): `recoil/api/schemas/_base.py` exists. `engine.py` and `system_status.py` both import from it. DONE.
- A2 (was_pending): `proposal_dispatch.py` already accepts `was_pending: bool` explicitly. DONE.
- A3 (tree walkers): `focus_walker.ts` exists at `src/lib/focus_walker.ts`. `App.tsx` and `use_breadcrumb.ts` already import `findFocusPath` from it. DONE.

**Active proposal lifecycle (verify before editing):**
- Real proposals (chat-originated): stored at `~/.recoil/proposals/<project>/<uuid_hex>.json`
- Created via `POST /api/chat/proposals` → `proposals_routes.py:create_proposal()`
- Approved via `POST /api/chat/proposals/<id>/approve` → `proposals_routes.py:approve_proposal()` (currently returns 501)
- Listed via `GET /api/chat/proposals/<project>` → `proposals_routes.py:list_proposals()`
- Frontend: `ProposalTray.tsx` calls all three routes above

**Fixture proposals (queue inspector):**
- Managed by `mutation_routes.py` at `/api/proposals/{id}/approve|reject|defer`
- Use short IDs (prop_001, prop_002, etc.) from `stub_routes._pending_items()`
- These are NOT the ProposalTray proposals — do not confuse

**beats.py adapter — key methods:**
- `get_shot_dict(beat_id, project_id=None) → Optional[dict]` — returns dict only, no path
- `set_primary`, `toggle_circled`, `reject_take` — all use `_find_shot_for_take(take_id, project_id)` → `(path, shot, raw_take, idx)` then call `atomic_write_json(path, shot)`
- New method `set_prompt_override` must follow the same (path, shot) → atomic_write pattern

**BUS API:**
- `BUS.emit_sync(severity, scope, summary, detail=None, payload=None)` — sync path, safe from non-async code
- `BUS._reset_for_tests()` — clears ring buffer for test isolation
- `BUS.history(after_id=None) → list[EngineEvent]` — returns ring buffer contents
- Valid severities: "success", "info", "warning", "error", "fallback"

**Test infrastructure:**
- Tests in `recoil/api/tests/` use `TestClient(app)` as context manager
- Pattern: `with TestClient(app) as c: yield c`
- Import pattern: `from recoil.api.eventbus import BUS`, `from recoil.api.main import app`
- `from recoil.api.stub_routes import _reset_acted_for_tests` — required before each test
- NEVER assert SSE stream in tests — TestClient deadlocks on EventSourceResponse (documented in test_sse_routes.py)
- Always assert BUS.history() instead of SSE

---

## Phase 0: Smoke Test Foundation + Kind Field

depends_on: none

**Objective:** Add `kind` field to proposal creation so the executor dispatcher can route by ProposalKind. Create smoke test infrastructure. Both are prerequisites for Phase 1.

**Why `kind` field is needed:** The current `_ProposalCreate` schema has no `kind` field. The executor in Phase 1 needs to know whether an approved proposal is a `PromptRewriteProposal`, `ParameterChangeProposal`, etc. Without storing `kind` at creation time, the dispatcher must guess from the diff structure (fragile).

### Files to create

**`recoil/api/tests/test_smoke_integration.py`** — new test file, flat alongside existing tests

```python
"""Integration smoke test — validates the full HTTP → disk → BUS path.

Uses TestClient(app) so the test runs in-process without requiring the
LaunchAgent or uvicorn to be active. NEVER consume the SSE stream —
TestClient deadlocks on EventSourceResponse. Assert BUS.history() instead.
"""
from __future__ import annotations

import json
import tempfile
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from recoil.api.eventbus import BUS
from recoil.api.main import app
from recoil.api.stub_routes import _reset_acted_for_tests


@pytest.fixture
def client():
    """TestClient bound to lifespan so BUS binds to the test loop."""
    _reset_acted_for_tests()
    BUS._reset_for_tests()
    with TestClient(app) as c:
        yield c


def test_system_status_returns_200(client: TestClient) -> None:
    """API starts and system-status endpoint responds."""
    r = client.get("/api/system-status")
    assert r.status_code == 200
    data = r.json()
    assert "api" in data or "status" in data or "schema_version" in data


def test_create_proposal_returns_ok(client: TestClient) -> None:
    """POST /api/chat/proposals creates a proposal with the new kind field."""
    body = {
        "target": "beat:TEST_BEAT_001",
        "title": "Smoke test proposal",
        "est_cost_usd": 0.01,
        "est_time": "1s",
        "kind": "PromptRewriteProposal",
        "diff": [{"kind": "rewrite", "after": "New prompt text here"}],
    }
    r = client.post("/api/chat/proposals", json=body)
    assert r.status_code == 200, r.text
    data = r.json()
    assert data.get("ok") is True
    assert "id" in data
    # Kind field must round-trip: list proposals and verify kind stored
    proposal_id = data["id"]
    list_r = client.get("/api/chat/proposals/default")
    assert list_r.status_code == 200
    proposals = list_r.json()
    match = next((p for p in proposals if p["id"] == proposal_id), None)
    assert match is not None, f"proposal {proposal_id} not in list"
    assert match.get("kind") == "PromptRewriteProposal"
```

### Files to modify

**`recoil/api/proposals_routes.py`**

**Change 1: Add `kind` to `_ProposalCreate`.**

In the `_ProposalCreate` class (around line 243), add:
```python
class _ProposalCreate(BaseModel):
    target: str
    title: str
    est_cost_usd: float
    est_time: str
    diff: list[_DiffEntry] = Field(default_factory=list)
    project: Optional[str] = None
    session_id: Optional[str] = None
    kind: Optional[str] = None  # ADD THIS LINE — ProposalKind string
```

**Change 2: Store `kind` in the proposal JSON.**

In `create_proposal()`, in the `doc = { ... }` dict (around line 339), add `"kind": body.kind` alongside the other fields:
```python
doc = {
    "schema_version": SCHEMA_VERSION,
    "id": pid,
    "project": project_id,
    "title": body.title,
    "target": body.target,
    "diff": [d.model_dump(exclude_none=True) for d in body.diff],
    "est_cost_usd": body.est_cost_usd,
    "est_time": body.est_time,
    "kind": body.kind,  # ADD THIS LINE
    "status": "pending",
    "created_at": _now_iso(),
}
```

**Change 3: Include `kind` in the list response.**

In `list_proposals()`, in the `out.append({ ... })` dict (around line 383), add `"kind": doc.get("kind")`:
```python
out.append({
    "id": doc.get("id"),
    "title": doc.get("title"),
    "target": doc.get("target"),
    "diff": doc.get("diff", []),
    "est_cost_usd": doc.get("est_cost_usd"),
    "est_time": doc.get("est_time"),
    "status": doc.get("status"),
    "kind": doc.get("kind"),  # ADD THIS LINE
})
```

### Scope boundary

Do NOT modify `approve_proposal()` — Phase 1 does that.
Do NOT modify any route paths.
Do NOT create the executor — Phase 1 does that.

### Validation

```bash
cd /Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS && \
PYTHONPATH=/Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS \
  pytest recoil/api/tests/test_smoke_integration.py -v && \
echo "Phase 0 OK"
```

Also verify syntax:
```bash
cd /Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS && \
python3 -c "import ast; ast.parse(open('recoil/api/proposals_routes.py').read()); print('syntax OK')" && \
grep -n '"kind"' recoil/api/proposals_routes.py | head -5
```

---

## Phase 1: PromptRewrite Executor

depends_on: 0

**Objective:** Wire the first real ProposalKind executor end-to-end. When a `PromptRewriteProposal` is approved: (a) write `shot["prompt_override"]` to disk, (b) ensure `recoil/core/prompt_compiler.py:compile()` reads that field as a full-prompt bypass at generation time, (c) emit a BUS event, (d) replace the 501 response with 200 + execution result.

**Architecture:** The executor lives in `recoil/api/executors/`. The dispatcher lives in `proposals_routes.py`'s `approve_proposal()` — this is the right home because `proposals_routes.py` owns the full proposal lifecycle for disk-based proposals. Do NOT wire into `mutation_routes.py` — that handles fixture proposals only.

**Read-path wire (NEW):** `prompt_compiler.py:compile()` at `recoil/core/prompt_compiler.py:1179` currently assembles a prompt from 10 layers and ignores `shot["prompt_override"]`. This phase adds an early-return bypass at the top of `compile()`: if `shot.get("prompt_override")` is truthy, return that as the prompt directly. This matches the existing `generation_config.prompt_override` semantic in `recoil/pipeline/tools/build_coverage_passes.py:58`.

**Scoped out of this build:** `recoil/pipeline/_lib/prompt_engine.py` (video / multi-modal builders) reads `prompt_data`, `routing_data`, etc. — not `shot["prompt_override"]`. Video-pipeline override wiring is a separate follow-on. PromptRewriteProposal in this build affects keyframe (`prompt_compiler.compile`) prompts only. Flag this scope boundary in the BUS event payload.

### Files to create

**`recoil/api/executors/__init__.py`**

```python
"""Proposal executors — one module per ProposalKind.

Each executor exports a single `execute(**kwargs) -> dict` function.
The dispatcher in proposals_routes.py calls the right executor based
on the stored `kind` field of the approved proposal.
"""
```

**`recoil/api/executors/prompt_rewrite.py`**

```python
"""Executor for PromptRewriteProposal.

Writes shot["prompt_override"] = new_text to the shot JSON via the
beats adapter and emits a BUS event.

NOTE: prompt_override is stored on disk but NOT yet read by the prompt
compiler at generation time. This is a known integration gap — flagged
in the BUS event payload so operators can see it in the events drawer.
"""
from __future__ import annotations

import logging
from typing import Optional

from fastapi import HTTPException

from recoil.api.adapters import beats as beats_adapter
from recoil.api.eventbus import BUS

logger = logging.getLogger(__name__)

_SCOPE = "api/executors/prompt_rewrite"


def execute(beat_id: str, new_text: str, project_id: Optional[str] = None) -> dict:
    """Write new_text to shot["prompt_override"] on disk.

    Args:
        beat_id: Shot file stem (e.g. "EP001_SH02"). Must match a .json
                 file in projects/{project}/state/visual/shots/.
        new_text: The full replacement prompt text.
        project_id: Optional project slug. If None, all projects are scanned.

    Returns:
        {"shot_id": str, "prompt_override_set": True}

    Raises:
        HTTPException(404): If beat_id resolves to no shot file on disk.
    """
    try:
        result = beats_adapter.set_prompt_override(beat_id, new_text, project_id)
    except KeyError:
        BUS.emit_sync(
            severity="error",
            scope=_SCOPE,
            summary=f"prompt_rewrite_target_not_found: {beat_id}",
            payload={
                "beat_id": beat_id,
                "severity": "error",
                "note": "beat_id not found in any project shots directory",
            },
        )
        raise HTTPException(
            status_code=404,
            detail={
                "error": "beat_not_found",
                "beat_id": beat_id,
                "message": (
                    f"Cannot rewrite prompt: beat {beat_id!r} does not exist on disk. "
                    "Check that the project is loaded and the beat ID is correct."
                ),
            },
        )
    BUS.emit_sync(
        severity="success",
        scope=_SCOPE,
        summary=f"prompt_rewrite_applied: {beat_id}",
        payload={
            **result,
            "beat_id": beat_id,
            "new_text_length": len(new_text),
            "scope_note": (
                "prompt_override wired into recoil/core/prompt_compiler.py "
                "(keyframe path). Video pipeline "
                "(recoil/pipeline/_lib/prompt_engine.py) is unchanged in this "
                "build — video-side override is a separate follow-on."
            ),
        },
    )
    return result
```

### Files to modify

**`recoil/api/adapters/beats.py`**

Add `set_prompt_override` after the existing mutation helpers (after `reject_take`, before `__all__`):

```python
def set_prompt_override(
    beat_id: str, new_text: str, project_id: Optional[str] = None
) -> dict:
    """Write new_text to shot["prompt_override"] for the given beat.

    Follows the same atomic write pattern as set_primary / toggle_circled.
    Searches by beat_id (shot file stem), not by take_id.

    Raises KeyError if no shot file exists for beat_id.
    """
    validate_hierarchy_id("beat_id", beat_id)
    if project_id is not None:
        validate_project_id(project_id)
    candidates: list[str]
    if project_id:
        candidates = [project_id]
    else:
        root = projects_root()
        candidates = [p.name for p in root.iterdir() if p.is_dir()] if root.exists() else []
    for slug in candidates:
        path = _shots_dir(slug) / f"{beat_id}.json"
        if not path.exists():
            continue
        shot = _load_shot(path)
        if shot is None:
            continue
        shot["prompt_override"] = new_text
        atomic_write_json(path, shot)
        return {"shot_id": shot.get("shot_id") or path.stem, "prompt_override_set": True}
    raise KeyError(f"beat {beat_id!r} not found in any project")
```

Also add `set_prompt_override` to the `__all__` list at the bottom of beats.py.

**`recoil/core/prompt_compiler.py`** — wire the read path

Find the `compile()` function (around line 1179). Immediately after the docstring (which ends around line 1216 with `}`), and BEFORE the `if storyboard is None:` block, insert:

```python
    # ── prompt_override bypass ──────────────────────────────────────────
    # If shot has a prompt_override (set by approved PromptRewriteProposal),
    # return it directly as the full prompt — bypasses all layer assembly.
    # Matches generation_config.prompt_override semantic in build_coverage_passes.py.
    override = shot.get("prompt_override")
    if override:
        capabilities = MODEL_CAPABILITIES.get(model, MODEL_CAPABILITIES["z_image"])
        if capabilities["negative_prompt"]:
            negative = (project_config or {}).get(
                "negative_prompt", _DEFAULT_PROJECT_CONFIG["negative_prompt"]
            )
        else:
            negative = ""
        return {
            "prompt": override,
            "negative_prompt": negative,
            "prompt_hash": _prompt_hash(override, negative),
            "layers": {name: "" for name in LAYER_ORDER},
            "warnings": ["prompt_override active — layer assembly bypassed"],
            "model": model,
            "override_applied": True,
        }
```

Add an inline test to `recoil/core/test_prompt_compiler.py` (or wherever the existing tests live — grep first; if no test file exists, add to the smoke test):

```python
def test_compile_respects_prompt_override():
    """When shot has prompt_override set, compile() returns it directly."""
    from recoil.core.prompt_compiler import compile as compile_prompt
    shot = {
        "id": 1,
        "prompt_override": "A wide aerial shot at golden hour, two figures running across a salt flat.",
    }
    result = compile_prompt(shot=shot, breakdown={}, episode=1, model="z_image")
    assert result["prompt"] == shot["prompt_override"]
    assert result.get("override_applied") is True
    assert result["warnings"] == ["prompt_override active — layer assembly bypassed"]
    # All layers must be empty strings (not absent keys) for consistent downstream consumption.
    for layer_name in [
        "lora_triggers", "subject", "action_pose", "wardrobe_props", "environment",
        "lighting", "color_objects", "camera_lens", "film_style", "quality_guard",
    ]:
        assert result["layers"][layer_name] == ""
```

**`recoil/api/proposals_routes.py`**

Add imports at the top of the file (after existing imports):
```python
from recoil.api.executors import prompt_rewrite as _executor_prompt_rewrite
```

Replace the `approve_proposal` function body. The current function (lines ~400-440) returns 501 with `"approved_not_dispatched"`. Replace with:

```python
@router.post("/chat/proposals/{proposal_id}/approve")
@_json_500_on_unhandled
def approve_proposal(proposal_id: str, body: _ProposalAction):
    """Mark approved on disk; dispatch executor if kind is wired.

    Phase 1: PromptRewriteProposal wired. All other kinds still return
    501 "approved_not_dispatched" until their executor is written.
    """
    project_id = body.project or "default"
    path = _proposal_path(project_id, proposal_id)
    captured: dict[str, Any] = {}

    def _approve(doc: dict) -> dict:
        if doc.get("status") != "pending":
            raise HTTPException(
                status_code=409,
                detail=f"proposal not pending: {doc.get('status')}",
            )
        captured["title"] = doc.get("title")
        captured["target"] = doc.get("target")
        captured["kind"] = doc.get("kind")
        captured["diff"] = doc.get("diff", [])
        captured["project"] = doc.get("project", project_id)
        doc["status"] = "approved"
        doc["approved_at"] = _now_iso()
        return doc

    _read_modify_write(path, _approve)

    kind = captured.get("kind")
    target = captured.get("target") or ""
    diff = captured.get("diff") or []
    proposal_project = captured.get("project") or project_id

    # ── PromptRewriteProposal dispatch ──────────────────────────────────
    if kind == "PromptRewriteProposal":
        # Target format: "beat:BEAT_ID"
        if not target.startswith("beat:"):
            return JSONResponse(
                status_code=422,
                content={
                    "error": "invalid_target",
                    "detail": f"PromptRewriteProposal target must start with 'beat:'; got {target!r}",
                    "proposal_id": proposal_id,
                },
            )
        beat_id = target[len("beat:"):]
        # Extract new_text from diff: first entry with "after" or "text" key.
        new_text = ""
        for entry in diff:
            new_text = entry.get("after") or entry.get("text") or ""
            if new_text:
                break
        if not new_text:
            return JSONResponse(
                status_code=422,
                content={
                    "error": "empty_new_text",
                    "detail": "PromptRewriteProposal diff contains no 'after' or 'text' value",
                    "proposal_id": proposal_id,
                },
            )
        # execute() raises HTTPException(404) if beat not found.
        result = _executor_prompt_rewrite.execute(
            beat_id=beat_id,
            new_text=new_text,
            project_id=proposal_project if proposal_project != "default" else None,
        )
        BUS.emit_sync(
            severity="success",
            scope=_BUS_SCOPE,
            summary=f"proposal approved + executed: {captured.get('title')}",
            payload={
                "id": proposal_id,
                "project": project_id,
                "kind": kind,
                "target": target,
                "status": "approved",
                "execution_result": result,
            },
        )
        return {"ok": True, "status": "executed", "result": result, "proposal_id": proposal_id}

    # ── Unimplemented kinds: honest 501 ─────────────────────────────────
    # Handles kind=None (pre-Phase-0 proposals that lack the field) and
    # any unimplemented kind string. Both fall through here intentionally —
    # no special None branch needed. kind=None in the summary tells operators
    # the proposal predates the kind field; a named kind tells them the
    # executor isn't built yet.
    BUS.emit_sync(
        severity="info",
        scope=_BUS_SCOPE,
        summary=f"proposal approved (no executor, kind={kind!r}): {captured.get('title')}",
        payload={
            "id": proposal_id,
            "project": project_id,
            "kind": kind,
            "target": target,
            "status": "approved_not_dispatched",
        },
    )
    return JSONResponse(
        status_code=501,
        content={
            "status": "approved_not_dispatched",
            "detail": (
                f"Proposal marked approved on disk. No executor for kind {kind!r} yet — "
                "run generation manually."
            ),
            "proposal_id": proposal_id,
            "project": project_id,
            "kind": kind,
            "target": target,
        },
    )
```

**`recoil/api/tests/test_smoke_integration.py`** — expand with PromptRewrite tests

Add to the existing file (do not replace — append new test functions):

```python
def test_prompt_rewrite_approve_writes_disk(
    client: TestClient, tmp_path: Path, monkeypatch
) -> None:
    """Approve a PromptRewriteProposal → shot JSON gets prompt_override field."""
    import json as _json
    from recoil.api.adapters import beats as _beats
    from recoil.core.paths import projects_root as _projects_root

    # Seed a minimal shot JSON in tmp_path under a fake project.
    fake_project = "smoke_test_project"
    shots_dir = tmp_path / fake_project / "state" / "visual" / "shots"
    shots_dir.mkdir(parents=True)
    beat_id = "SMOKE_SH01"
    shot_data = {"shot_id": beat_id, "takes": [], "status": "pending"}
    (shots_dir / f"{beat_id}.json").write_text(
        _json.dumps(shot_data), encoding="utf-8"
    )

    # Monkeypatch projects_root() → tmp_path so beats adapter finds the fake shot.
    monkeypatch.setattr(_beats, "_shots_dir", lambda pid: tmp_path / pid / "state" / "visual" / "shots")

    # Create a PromptRewriteProposal.
    create_body = {
        "target": f"beat:{beat_id}",
        "title": "Smoke: rewrite prompt",
        "est_cost_usd": 0.0,
        "est_time": "instant",
        "kind": "PromptRewriteProposal",
        "project": fake_project,
        "diff": [{"kind": "rewrite", "after": "A completely new prompt text."}],
    }
    cr = client.post("/api/chat/proposals", json=create_body)
    assert cr.status_code == 200, cr.text
    proposal_id = cr.json()["id"]

    # Approve the proposal.
    ar = client.post(
        f"/api/chat/proposals/{proposal_id}/approve",
        json={"project": fake_project},
    )
    assert ar.status_code == 200, ar.text
    result = ar.json()
    assert result.get("status") == "executed"
    assert result.get("ok") is True

    # Assert disk mutation: shot JSON now has prompt_override.
    shot_path = shots_dir / f"{beat_id}.json"
    updated = _json.loads(shot_path.read_text())
    assert updated.get("prompt_override") == "A completely new prompt text."

    # Assert BUS history has the prompt_rewrite_applied event.
    history = BUS.history()
    rewrite_events = [e for e in history if "prompt_rewrite_applied" in e.summary]
    assert len(rewrite_events) >= 1, f"No prompt_rewrite_applied event; history: {[e.summary for e in history]}"


def test_prompt_rewrite_bad_beat_id_returns_404(
    client: TestClient, tmp_path: Path, monkeypatch
) -> None:
    """Approve a PromptRewriteProposal with a nonexistent beat → 404 + BUS error event."""
    from recoil.api.adapters import beats as _beats

    # Monkeypatch to an empty directory so no shots exist.
    empty_shots = tmp_path / "empty" / "state" / "visual" / "shots"
    monkeypatch.setattr(_beats, "_shots_dir", lambda pid: empty_shots)

    create_body = {
        "target": "beat:NONEXISTENT_BEAT",
        "title": "Bad beat test",
        "est_cost_usd": 0.0,
        "est_time": "instant",
        "kind": "PromptRewriteProposal",
        "diff": [{"kind": "rewrite", "after": "irrelevant"}],
    }
    cr = client.post("/api/chat/proposals", json=create_body)
    proposal_id = cr.json()["id"]

    ar = client.post(
        f"/api/chat/proposals/{proposal_id}/approve",
        json={"project": "default"},
    )
    assert ar.status_code == 404, ar.text
    detail = ar.json().get("detail", {})
    assert detail.get("error") == "beat_not_found"
    assert "NONEXISTENT_BEAT" in str(detail)

    # Assert BUS error event.
    history = BUS.history()
    error_events = [e for e in history if e.severity == "error"]
    assert len(error_events) >= 1
    assert any("prompt_rewrite_target_not_found" in e.summary for e in error_events)
```

### What already exists (from Phase 0)

- `recoil/api/tests/test_smoke_integration.py` — created in Phase 0 (basic tests)
- `proposals_routes.py` — `_ProposalCreate` has `kind` field, `create_proposal` stores `kind`, `list_proposals` returns `kind`

### Scope boundary

Do NOT touch `mutation_routes.py` — fixture proposals remain unchanged.
Do NOT modify any existing routes other than `approve_proposal` in `proposals_routes.py`.
Do NOT add error handling for scenarios that cannot happen in single-operator context.
The `monkeypatch.setattr(_beats, "_shots_dir", ...)` approach in the test is the correct test isolation strategy — it redirects the internal function rather than mocking at a higher level.

### Validation

```bash
cd /Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS && \
PYTHONPATH=/Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS \
  pytest recoil/api/tests/test_smoke_integration.py -v && \
python3 -c "
import ast
ast.parse(open('recoil/api/adapters/beats.py').read())
ast.parse(open('recoil/api/executors/prompt_rewrite.py').read())
ast.parse(open('recoil/api/proposals_routes.py').read())
ast.parse(open('recoil/core/prompt_compiler.py').read())
print('syntax OK')
" && \
grep -q 'set_prompt_override' recoil/api/adapters/beats.py && \
grep -q 'prompt_rewrite_applied' recoil/api/executors/prompt_rewrite.py && \
grep -q '_executor_prompt_rewrite' recoil/api/proposals_routes.py && \
grep -q 'prompt_override bypass\|override_applied' recoil/core/prompt_compiler.py && \
PYTHONPATH=/Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS python3 -c "
from recoil.core.prompt_compiler import compile as cp
r = cp(shot={'id': 1, 'prompt_override': 'TEST OVERRIDE STRING'}, breakdown={}, episode=1, model='z_image')
assert r['prompt'] == 'TEST OVERRIDE STRING', f'override not applied: {r[\"prompt\"]!r}'
assert r.get('override_applied') is True
print('compile() override bypass works')
" && \
echo "Phase 1 OK"
```

---

## Phase 1.5b: create_proposal MCP Tool

depends_on: 1, parent-take-mcp build merged

**GATE:** Do NOT start this phase until `parent-take-mcp` build is merged and `sync-machines.sh pull` has been run. The file `recoil/api/console_mcp_shim.py` is being modified by that build.

**Objective:** Add `create_proposal(kind, payload)` tool to the MCP shim so Claude in the embedded terminal can originate proposals that appear in ProposalTray.

### Files to modify

**`recoil/api/console_mcp_shim.py`**

First, read the current file to understand the existing `get_current_selection()` tool pattern. Then add `create_proposal` following the exact same registration pattern.

The tool should:
1. Accept `kind: str` (must be a valid ProposalKind) and `payload: dict`
2. POST to `/api/chat/proposals` internally using the httpx client (or requests, whichever the shim already uses)
3. Extract `beat_id` from `payload` (or call `get_current_selection()` to get it if absent)
4. Build the proposal body: `{target: "beat:{beat_id}", kind, diff: [{kind: "rewrite", after: payload.get("new_text", "")}], ...}`
5. Return the created proposal ID on success
6. CATCH `status_code == 422` and return the error as a readable string (not raise) — prevents MCP 422 black hole

Valid ProposalKinds:
```python
_VALID_KINDS = {
    "PromptRewriteProposal",
    "BeatInsertionProposal",
    "ParameterChangeProposal",
    "ScriptEditProposal",
    "MultiBeatDirectiveProposal",
    "ExtractCutawayProposal",
    "RefSwapProposal",
    "RetryStrategyEditProposal",
}
```

Validate `kind` against this set before POSTing — return readable error string if invalid.

### New tests to add to `recoil/api/tests/test_console_mcp_shim.py`

The existing test file uses `unittest.TestCase` with `unittest.mock.patch` against `urllib.request.urlopen`. The new tests follow that exact pattern. Append these classes to the file:

```python
class TestCreateProposal(unittest.TestCase):
    def test_create_proposal_posts_correct_body(self):
        """create_proposal sends POST /api/chat/proposals with correct body and returns ID."""
        api_response = {"ok": True, "id": "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4"}

        with patch("urllib.request.urlopen", return_value=_make_mock_resp(api_response)) as mock_open:
            result = shim.tool_create_proposal({
                "kind": "PromptRewriteProposal",
                "payload": {
                    "beat_id": "EP001_SH01",
                    "new_text": "Rewritten prompt.",
                    "project": "tartarus",
                },
            })

        req_obj = mock_open.call_args[0][0]
        self.assertEqual(req_obj.method, "POST")
        self.assertIn("/api/chat/proposals", req_obj.full_url)
        sent_body = json.loads(req_obj.data.decode("utf-8"))
        self.assertEqual(sent_body["kind"], "PromptRewriteProposal")
        self.assertEqual(sent_body["target"], "beat:EP001_SH01")
        self.assertEqual(sent_body["diff"][0]["after"], "Rewritten prompt.")
        self.assertEqual(sent_body["project"], "tartarus")
        self.assertEqual(result["id"], "a1b2c3d4e5f6a1b2c3d4e5f6a1b2c3d4")
        self.assertTrue(result["ok"])

    def test_create_proposal_invalid_kind_returns_error(self):
        """create_proposal with invalid kind returns error dict, does not raise."""
        result = shim.tool_create_proposal({
            "kind": "NotARealKind",
            "payload": {"beat_id": "EP001_SH01", "new_text": "text"},
        })
        self.assertIsInstance(result, dict)
        self.assertIn("error", result)
        self.assertIn("NotARealKind", result.get("error", ""))

    def test_create_proposal_via_jsonrpc_dispatch(self):
        """create_proposal is reachable via tools/call JSON-RPC method."""
        api_response = {"ok": True, "id": "deadbeefdeadbeefdeadbeefdeadbeef"}
        request = {
            "jsonrpc": "2.0",
            "id": 5,
            "method": "tools/call",
            "params": {
                "name": "create_proposal",
                "arguments": {
                    "kind": "PromptRewriteProposal",
                    "payload": {
                        "beat_id": "EP001_SH02",
                        "new_text": "New text here.",
                        "project": "tartarus",
                    },
                },
            },
        }
        with patch("urllib.request.urlopen", return_value=_make_mock_resp(api_response)):
            response = shim._handle_request(request)

        self.assertIsNotNone(response)
        result = response["result"]
        self.assertFalse(result["isError"])
        parsed = json.loads(result["content"][0]["text"])
        self.assertTrue(parsed["ok"])
        self.assertEqual(parsed["id"], "deadbeefdeadbeefdeadbeefdeadbeef")
```

Also add `self.assertIn("create_proposal", tool_names)` to the existing `TestToolsList.test_tools_list` test.

### Validation

```bash
cd /Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS && \
python3 -c "
import ast; ast.parse(open('recoil/api/console_mcp_shim.py').read())
print('syntax OK')
" && \
grep -q 'create_proposal' recoil/api/console_mcp_shim.py && \
PYTHONPATH=/Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS \
  pytest recoil/api/tests/test_console_mcp_shim.py -v && \
echo "Phase 1.5b OK"
```

Also manual terminal test (required):
- In the embedded terminal, ask Claude to rewrite the prompt for the currently selected beat
- Proposal appears in ProposalTray

---

## Phase 2: ParamTweak Executor

depends_on: 1

**Objective:** Wire the `ParameterChangeProposal` kind. Requires writing `update_take_params()` in the beats adapter — this function does NOT exist anywhere in the codebase.

### Files to create

**`recoil/api/executors/param_tweak.py`**

```python
"""Executor for ParameterChangeProposal.

Writes params_delta fields into the target take's dict within the shot JSON.
"""
from __future__ import annotations

import logging
from typing import Optional

from fastapi import HTTPException

from recoil.api.adapters import beats as beats_adapter
from recoil.api.eventbus import BUS

logger = logging.getLogger(__name__)

_SCOPE = "api/executors/param_tweak"


def execute(take_id: str, params_delta: dict, project_id: Optional[str] = None) -> dict:
    """Merge params_delta into the target take's dict on disk.

    Args:
        take_id: The take to update (e.g. "EP001_SH02_T001").
        params_delta: Key-value pairs to merge into the take dict.
        project_id: Optional project slug.

    Returns:
        {"shot_id": str, "take_id": str, "params_updated": list[str]}

    Raises:
        HTTPException(404): If take_id resolves to no take on disk.
    """
    try:
        result = beats_adapter.update_take_params(take_id, params_delta, project_id)
    except KeyError:
        BUS.emit_sync(
            severity="error",
            scope=_SCOPE,
            summary=f"param_tweak_target_not_found: {take_id}",
            payload={"take_id": take_id, "severity": "error"},
        )
        raise HTTPException(
            status_code=404,
            detail={
                "error": "take_not_found",
                "take_id": take_id,
                "message": f"Cannot update params: take {take_id!r} not found on disk.",
            },
        )
    BUS.emit_sync(
        severity="success",
        scope=_SCOPE,
        summary=f"param_tweak_applied: {take_id}",
        payload={**result, "take_id": take_id, "params_delta_keys": list(params_delta.keys())},
    )
    return result
```

### Files to modify

**`recoil/api/adapters/beats.py`**

Add `update_take_params` after `set_prompt_override`, before `__all__`:

```python
def update_take_params(
    take_id: str, params_delta: dict, project_id: Optional[str] = None
) -> dict:
    """Merge params_delta into the take dict for take_id.

    Uses _find_shot_for_take to locate (path, shot, raw_take, idx).
    Updates raw_take in-place with params_delta keys, then atomic_write_json.

    Raises KeyError if take_id is not found.
    """
    path, shot, raw_take, _idx = _find_shot_for_take(take_id, project_id)
    updated_keys = []
    for key, value in params_delta.items():
        raw_take[key] = value
        updated_keys.append(key)
    atomic_write_json(path, shot)
    return {
        "shot_id": shot.get("shot_id") or path.stem,
        "take_id": take_id,
        "params_updated": updated_keys,
    }
```

Also add `update_take_params` to `__all__`.

**`recoil/api/proposals_routes.py`**

Add import:
```python
from recoil.api.executors import param_tweak as _executor_param_tweak
```

In `approve_proposal()`, add a branch for `ParameterChangeProposal` AFTER the `PromptRewriteProposal` block and BEFORE the unimplemented-kinds 501:

```python
# ── ParameterChangeProposal dispatch ────────────────────────────────
if kind == "ParameterChangeProposal":
    # Target format: "take:TAKE_ID"
    if not target.startswith("take:"):
        return JSONResponse(
            status_code=422,
            content={
                "error": "invalid_target",
                "detail": f"ParameterChangeProposal target must start with 'take:'; got {target!r}",
                "proposal_id": proposal_id,
            },
        )
    take_id = target[len("take:"):]
    # Extract params_delta: each diff entry with a "key" field becomes a param.
    params_delta = {}
    for entry in diff:
        key = entry.get("key")
        value = entry.get("after") or entry.get("text")
        if key and value is not None:
            params_delta[key] = value
    if not params_delta:
        return JSONResponse(
            status_code=422,
            content={
                "error": "empty_params_delta",
                "detail": "ParameterChangeProposal diff contains no key/after pairs",
                "proposal_id": proposal_id,
            },
        )
    result = _executor_param_tweak.execute(
        take_id=take_id,
        params_delta=params_delta,
        project_id=proposal_project if proposal_project != "default" else None,
    )
    BUS.emit_sync(
        severity="success",
        scope=_BUS_SCOPE,
        summary=f"proposal approved + executed: {captured.get('title')}",
        payload={"id": proposal_id, "kind": kind, "execution_result": result},
    )
    return {"ok": True, "status": "executed", "result": result, "proposal_id": proposal_id}
```

**`recoil/api/tests/test_smoke_integration.py`**

Add ParamTweak smoke test (follow the same pattern as `test_prompt_rewrite_approve_writes_disk` in Phase 1 — seed a shot JSON with a take, approve a ParameterChangeProposal with `target="take:TAKE_ID"`, assert the take dict was updated).

### What already exists (from Phase 1)

- `recoil/api/executors/__init__.py`
- `recoil/api/executors/prompt_rewrite.py` (reference for pattern)
- `proposals_routes.py` with `approve_proposal` dispatch block and executor imports

### Scope boundary

Do NOT modify `_find_shot_for_take` — use it as-is.
Do NOT add locking for the read-modify-write race condition — accepted risk for single-operator MVP.

### Validation

```bash
cd /Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS && \
PYTHONPATH=/Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS \
  pytest recoil/api/tests/test_smoke_integration.py -v && \
python3 -c "
import ast
ast.parse(open('recoil/api/executors/param_tweak.py').read())
ast.parse(open('recoil/api/adapters/beats.py').read())
print('syntax OK')
" && \
grep -q 'update_take_params' recoil/api/adapters/beats.py && \
grep -q 'param_tweak_applied' recoil/api/executors/param_tweak.py && \
echo "Phase 2 OK"
```

---

## Phase 3: ScriptEdit Executor

depends_on: 1

### What already exists (from prior phases)

- **Executor pattern (Phase 1):** Each executor is a module at `recoil/api/executors/<name>.py` exporting `execute(**kwargs) -> dict`. The executor raises `HTTPException(404)` if the target is not found on disk, and emits BUS events on success/error via `BUS.emit_sync`.
- **Dispatch block in `proposals_routes.py`:** `approve_proposal()` reads the proposal from disk via `_read_modify_write`, captures `kind/target/diff/project`, then dispatches by `kind` string match. Each kind branch validates the target prefix, extracts parameters from `diff`, calls the executor, emits a BUS event, and returns `{ok, status: "executed", result}`. Unmatched kinds fall through to 501 "approved_not_dispatched".
- **Executor imports at top of `proposals_routes.py`:** `from recoil.api.executors import prompt_rewrite as _executor_prompt_rewrite` (Phase 1) and `from recoil.api.executors import param_tweak as _executor_param_tweak` (Phase 2, if already built).
- **`get_episode_id_for_beat` helper (this phase adds it):** Do NOT import the private `_derive_episode_id` from `beats.py` — it is an underscore-prefixed internal. Instead, this phase adds a public `get_episode_id_for_beat(beat_id, project_id) -> str | None` to `beats.py` (see beats.py section below) and imports that wrapper in `proposals_routes.py`.

**Objective:** Wire `ScriptEditProposal`. Write target is the episode script file in Fountain format. Requires locating the script file via the beat's episode/scene hierarchy.

### Files to create

**`recoil/api/executors/script_edit.py`**

```python
"""Executor for ScriptEditProposal.

Write target: episode script file in Fountain format. The script file
is located by resolving the proposal target's episode identifier.
"""
from __future__ import annotations

import logging
from pathlib import Path
from typing import Optional

from fastapi import HTTPException

from recoil.api.eventbus import BUS
from recoil.core.paths import projects_root

logger = logging.getLogger(__name__)

_SCOPE = "api/executors/script_edit"


def _find_script_path(episode_id: str, project_id: str) -> Optional[Path]:
    """Locate the Fountain script file for episode_id in project_id.

    Searches: projects/{project}/episodes/{episode_id}/script.fountain
    and       projects/{project}/episodes/{episode_id}/{episode_id}.fountain
    Returns None if not found.
    """
    ep_dir = projects_root() / project_id / "episodes" / episode_id
    for name in ("script.fountain", f"{episode_id}.fountain"):
        candidate = ep_dir / name
        if candidate.exists():
            return candidate
    return None


def execute(
    episode_id: str, new_script_text: str, project_id: Optional[str] = None
) -> dict:
    """Replace the script file contents for episode_id.

    Args:
        episode_id: Episode identifier (e.g. "ep_001").
        new_script_text: Full replacement Fountain script text.
        project_id: Optional project slug.

    Returns:
        {"episode_id": str, "script_path": str, "script_edit_applied": True}

    Raises:
        HTTPException(404): If script file not found.
        HTTPException(422): If project_id is None (required for ScriptEdit).
    """
    if not project_id:
        raise HTTPException(
            status_code=422,
            detail={"error": "project_id_required", "message": "ScriptEditProposal requires a project_id"},
        )
    script_path = _find_script_path(episode_id, project_id)
    if script_path is None:
        BUS.emit_sync(
            severity="error",
            scope=_SCOPE,
            summary=f"script_edit_target_not_found: {episode_id}",
            payload={"episode_id": episode_id, "project_id": project_id, "severity": "error"},
        )
        raise HTTPException(
            status_code=404,
            detail={
                "error": "script_not_found",
                "episode_id": episode_id,
                "project_id": project_id,
                "message": f"No script file found for episode {episode_id!r} in project {project_id!r}",
            },
        )
    # Atomic write: temp file + rename.
    import os
    import tempfile
    parent = script_path.parent
    fd, tmp = tempfile.mkstemp(dir=str(parent), suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(new_script_text)
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp, str(script_path))
    except Exception:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise
    BUS.emit_sync(
        severity="success",
        scope=_SCOPE,
        summary=f"script_edit_applied: {episode_id}",
        payload={
            "episode_id": episode_id,
            "project_id": project_id,
            "script_path": str(script_path),
            "new_text_length": len(new_script_text),
        },
    )
    return {
        "episode_id": episode_id,
        "script_path": str(script_path),
        "script_edit_applied": True,
    }
```

### Files to modify

**`recoil/api/adapters/beats.py`**

Add `get_episode_id_for_beat` after `get_shot_dict`, before the mutation helpers (`set_prompt_override`, `update_take_params`):

```python
def get_episode_id_for_beat(
    beat_id: str, project_id: Optional[str] = None
) -> str | None:
    """Resolve episode_id for a beat by loading its shot file.

    Public wrapper — use this instead of importing _derive_episode_id directly.
    """
    validate_hierarchy_id("beat_id", beat_id)
    if project_id is not None:
        validate_project_id(project_id)
    if project_id:
        candidates = [project_id]
    else:
        root = projects_root()
        candidates = [p.name for p in root.iterdir() if p.is_dir()] if root.exists() else []
    for slug in candidates:
        path = _shots_dir(slug) / f"{beat_id}.json"
        if not path.exists():
            continue
        shot = _load_shot(path)
        if shot is None:
            continue
        return _derive_episode_id(shot, path, slug)
    return None
```

Also add `"get_episode_id_for_beat"` to the `__all__` list.

**`recoil/api/proposals_routes.py`**

Add imports:
```python
from recoil.api.executors import script_edit as _executor_script_edit
from recoil.api.adapters.beats import get_episode_id_for_beat as _get_episode_id
```

In `approve_proposal()`, add `ScriptEditProposal` branch (after ParameterChangeProposal, before 501):

```python
# ── ScriptEditProposal dispatch ──────────────────────────────────────
if kind == "ScriptEditProposal":
    # Target format: "episode:EPISODE_ID" or "beat:BEAT_ID" (episode inferred)
    episode_id = None
    if target.startswith("episode:"):
        episode_id = target[len("episode:"):]
    elif target.startswith("beat:"):
        beat_id = target[len("beat:"):]
        # Derive episode_id from beat file via the public helper (never import _derive_episode_id directly)
        episode_id = _get_episode_id(
            beat_id, proposal_project if proposal_project != "default" else None
        )
    if not episode_id:
        return JSONResponse(
            status_code=422,
            content={"error": "episode_id_unresolvable", "target": target, "proposal_id": proposal_id},
        )
    new_script = ""
    for entry in diff:
        new_script = entry.get("after") or entry.get("text") or ""
        if new_script:
            break
    result = _executor_script_edit.execute(
        episode_id=episode_id,
        new_script_text=new_script,
        project_id=proposal_project if proposal_project != "default" else None,
    )
    return {"ok": True, "status": "executed", "result": result, "proposal_id": proposal_id}
```

### Validation

```bash
cd /Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS && \
python3 -c "
import ast
ast.parse(open('recoil/api/executors/script_edit.py').read())
print('syntax OK')
" && \
grep -q 'ScriptEditProposal' recoil/api/proposals_routes.py && \
grep -q 'script_edit_applied' recoil/api/executors/script_edit.py && \
grep -q 'get_episode_id_for_beat' recoil/api/adapters/beats.py && \
grep -q '_get_episode_id' recoil/api/proposals_routes.py && \
PYTHONPATH=/Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS \
  pytest recoil/api/tests/test_smoke_integration.py -v && \
echo "Phase 3 OK"
```

---

## Post-build orchestration: Phase 4 (remaining ProposalKinds) — DO NOT DISPATCH IN THIS BUILD

**Phase 4 is intentionally NOT in this BUILD_SPEC.** Do not implement it during this harness run. It will be specified by a separate orchestration round after Phases 0–3 verify green:

1. Read this build's commits + final `build-log-{session}.md`
2. Run a 1-round Opus consult: "Given the wired executor pattern in `recoil/api/executors/` (PromptRewrite, ParamTweak, ScriptEdit), specify executors for the remaining 5 ProposalKinds: BeatInsertion, MultiBeatDirective, ExtractCutaway, RefSwap, RetryStrategyEdit. For each: write target file, target shape, adapter method signature, validation criteria. Flag any kind that needs its own spec round."
3. Generate `BUILD_SPEC_PHASE_4.md` from the consult output
4. Dispatch as a separate harness run on Studio

The harness that runs THIS spec does NOT touch Phase 4 work. Phase 4 dispatch is triggered by the orchestrating session monitoring this build, only after Phases 0–3 are confirmed green (tests pass, no blocked phases, debug loop clean).

**Why deferred:** Phase 4 in SYNTHESIS contained "location TBD" for RetryStrategyEditProposal and "audit the diff schema before implementing" for MultiBeatDirectiveProposal. A maximum-detail spec doesn't ship with TBDs. Once Phases 0–3 land and the executor pattern is proven in real code, the remaining kinds can be specified with that pattern as a concrete reference.

---

## Verification Checklist (run after all phases)

```bash
# 1. All tests pass
cd /Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS && \
PYTHONPATH=/Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS \
  pytest recoil/api/tests/ -v --tb=short

# 2. TypeScript clean
cd /Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS/recoil/console-v2 && pnpm typecheck

# 3. All executor files exist and pass syntax
python3 -c "
import ast, pathlib
for f in pathlib.Path('recoil/api/executors').glob('*.py'):
    ast.parse(f.read_text())
    print(f'OK: {f}')
"

# 4. Executor registry (all wired kinds)
grep -c 'PromptRewriteProposal\|ParameterChangeProposal\|ScriptEditProposal' \
  recoil/api/proposals_routes.py

# 5. prompt_override read path wired in compiler
grep -q 'prompt_override bypass\|override_applied' recoil/core/prompt_compiler.py && \
  echo "compiler read-path wired"

# 6. End-to-end: write override, compile, verify bypass
PYTHONPATH=/Users/joeturnerlin/Dropbox/CLAUDE_PROJECTS python3 -c "
from recoil.core.prompt_compiler import compile as cp
r = cp(shot={'id': 1, 'prompt_override': 'TEST'}, breakdown={}, episode=1, model='z_image')
assert r['prompt'] == 'TEST' and r.get('override_applied') is True
print('end-to-end override bypass works')
"
```

---

## Known Integration Gaps

1. **Video pipeline does not read `shot["prompt_override"]`.** `recoil/pipeline/_lib/prompt_engine.py` (multi-modal builders for video / audio / lipsync) reads `prompt_data`, `routing_data`, etc. — not `shot["prompt_override"]`. PromptRewriteProposal in this build affects the keyframe path (`recoil/core/prompt_compiler.py:compile()`) only. Wiring video-pipeline override is a separate follow-on.

2. **`update_take_params` modifies raw take dict fields.** These fields may or may not be read by the generation pipeline. Verify per field before using in production.

3. **Edit Prompt origination is terminal-only in this build.** Console v2 has no beat-detail UI surface to mount a button on. JT does not want one built. Proposal origination flows through the embedded terminal via the MCP `create_proposal` tool (Phase 1.5b). When a chrome-level prompt-edit affordance is wanted, it'll be its own consultation + spec.