"""W4b-2 unit tests — scene_detail_redo_service.redo_scene_detail_shot
projection-promote behavior.

Two acceptance gates:

1. **clean-promote**: when the post-redo manifest has zero shots with
   ``status == "contract_violation"`` (a clean redo), the service updates
   the step_run row to ``status='completed'`` + synchronized counts +
   ``error_message=NULL``. This unblocks the next
   ``scene_detail?mode=resume`` from re-running 60 shots through the LLM.

2. **dirty-no-promote**: when the post-redo manifest still has at least
   one contract_violation shot, the service preserves the prior step_run
   status (and error_message) and only touches ``updated_at``.

The heavy parts of the service (project config / step factory / context
loader / LLM analyze / checkpoint sync / atomic file write) are mocked so
the test exercises ONLY the projection-promote logic against a real
in-memory SQLite step_run table.
"""
from __future__ import annotations

import uuid
from typing import Any, Dict, List, Optional
from unittest.mock import MagicMock

import pytest
from sqlalchemy import create_engine, text as sql_text
from sqlalchemy.orm import Session


# step_run schema mirrors the production raw-SQL CREATE in
# app/core/database.py (the columns the redo service touches).
_STEP_RUN_SCHEMA = """
CREATE TABLE step_run (
    id TEXT PRIMARY KEY,
    project_id TEXT NOT NULL,
    episode_id TEXT NOT NULL,
    step_id TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending',
    run_id TEXT,
    resolved_model TEXT,
    input_hash TEXT,
    upstream_revision TEXT,
    prompt_version TEXT,
    applicable_count INTEGER,
    completed_count INTEGER DEFAULT 0,
    failed_count INTEGER DEFAULT 0,
    error_message TEXT,
    result_summary TEXT,
    started_at TEXT,
    completed_at TEXT,
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    UNIQUE(project_id, episode_id, step_id)
)
"""


@pytest.fixture
def db_with_partial_step_run():
    """in-memory SQLite + step_run row with status='partial'.

    Simulates the post-W4b S18-fix entry state: scene_detail step_run is
    partial because a prior contract_violation forced manifest to dirty.
    """
    engine = create_engine("sqlite:///:memory:")
    with engine.begin() as conn:
        conn.execute(sql_text(_STEP_RUN_SCHEMA))
        conn.execute(
            sql_text(
                "INSERT INTO step_run (id, project_id, episode_id, step_id, "
                "status, applicable_count, completed_count, failed_count, "
                "error_message, created_at, updated_at) "
                "VALUES (:id, 'p1', 'e1', 'scene_detail', 'partial', 61, "
                "60, 1, 'prior owned validation contract_violation', "
                "'2026-05-23T00:00:00+00:00', '2026-05-23T00:00:00+00:00')"
            ),
            {"id": str(uuid.uuid4())},
        )
    with Session(engine) as db:
        yield engine, db


def _build_valid_sentinel(t2i_prompt: str = "stub") -> Dict[str, Any]:
    """Synthesize a sentinel that passes assert_owned_sentinel_shape.

    Uses build_owned_sentinel with empty owned/usage so the resulting
    sentinel is the minimum valid shape: all 4 hashes are real
    16-char lowercase hex of empty inputs, validator=FULL,
    violations=[]. This keeps the redo-service test fixtures
    decoupled from the heavy ctx the production verify_completion
    consults — the lightweight verify only checks
    ``assert_owned_sentinel_shape`` which is pure.
    """
    from app.core.steps._owned_helpers import build_owned_sentinel

    return build_owned_sentinel(
        owned=[],
        camera_direction="",
        t2i_prompt=t2i_prompt,
        is_close_framing=False,
        violations=[],
        owned_object_usage=[],
    )


def _make_shot_result(
    *,
    scene_index: int,
    shot_index: int,
    contract_violation: bool = False,
    include_sentinel: bool = True,
    omit_sentinel: bool = False,
    empty_t2i_variations: bool = False,
) -> Dict[str, Any]:
    """Minimal shot result dict — only the fields the redo service reads.

    Args:
        scene_index / shot_index / contract_violation: as before.
        include_sentinel: when True (default) attach a valid
            owned_validation sentinel to each variation so the
            lightweight verify treats the shot as clean.
        omit_sentinel: when True, variation has no owned_validation —
            exercises the W4b-fixup verify "sentinel missing" reason.
        empty_t2i_variations: when True, t2i_variations = [] —
            exercises the W4b-fixup verify "empty t2i_variations" reason.
    """
    if empty_t2i_variations:
        tvars: List[Dict[str, Any]] = []
    else:
        variation: Dict[str, Any] = {"t2i_prompt": "stub"}
        if include_sentinel and not omit_sentinel:
            variation["owned_validation"] = _build_valid_sentinel("stub")
        tvars = [variation]
    result: Dict[str, Any] = {
        "scene_index": scene_index,
        "_shot_index": shot_index,
        "t2i_variations": tvars,
        "render_prompt_card_hash": "h" + str(scene_index) + "_" + str(shot_index),
    }
    if contract_violation:
        result["status"] = "contract_violation"
    return result


def _wire_service_internals(
    monkeypatch,
    *,
    existing_scenes: List[Dict[str, Any]],
    redo_result: Dict[str, Any],
    cp_dir: Any,
):
    """Patch the heavy collaborators of redo_scene_detail_shot.

    Returns: (atomic_write_log, sync_called_flag) so callers can assert.
    """
    import app.services.scene_detail_redo_service as svc

    # 1) _load_project_config — return a benign dict.
    monkeypatch.setattr(
        "app.services.step_execution_service._load_project_config",
        lambda db, pid: {"project_id": pid},
    )

    # 2) get_step_runner — return a Mock that masquerades as
    #    SceneDetailStep (isinstance check uses the real class).
    from app.core.steps.detail_steps import SceneDetailStep

    fake_step = MagicMock(spec=SceneDetailStep)
    fake_step._cp_dir = cp_dir
    fake_step._build_llm_inputs.return_value = ({"role": "system"}, {})
    fake_step._analyze_one.return_value = redo_result
    fake_step.load_checkpoint.return_value = {
        "schema_version": 7,
        "status": "partial",
        "applicable_count": len(existing_scenes),
        "completed_count": len(existing_scenes),
        "failed_count": 1,
        "data": {"scenes": list(existing_scenes)},
    }
    monkeypatch.setattr(
        "app.services.analysis_dispatch_service.get_step_runner",
        lambda *a, **kw: fake_step,
    )

    # 3) SceneContextLoader — load_all() returns a ctx with the matching
    #    segments + shot_scenes_map.
    fake_ctx = MagicMock()
    fake_ctx.segments = [
        {"scene_index": redo_result["scene_index"]},
    ]
    fake_ctx.shot_scenes_map = {
        redo_result["scene_index"]: [
            {"shot_index": redo_result["_shot_index"]},
        ],
    }
    fake_loader = MagicMock()
    fake_loader.load_all.return_value = fake_ctx
    monkeypatch.setattr(
        "app.core.steps.scene_context_loader.SceneContextLoader",
        lambda step: fake_loader,
    )

    # 4) atomic_write_json — capture writes so we can assert manifest shape.
    write_log: List[Dict[str, Any]] = []

    def fake_atomic_write(path, payload):
        write_log.append({"path": str(path), "payload": payload})

    monkeypatch.setattr(
        "app.core.checkpoint_io.atomic_write_json", fake_atomic_write
    )

    # 5) orchestrate_full_sync — no-op, return summary.
    sync_log: Dict[str, Any] = {"called": False}

    def fake_sync(project_id, episode_id, db, step_id=None):
        sync_log["called"] = True
        sync_log["args"] = (project_id, episode_id, step_id)
        return {"ok": 1}

    monkeypatch.setattr(svc, "orchestrate_full_sync", fake_sync)

    return write_log, sync_log


def _read_step_run(db: Session) -> Optional[Dict[str, Any]]:
    row = db.execute(
        sql_text(
            "SELECT status, completed_count, failed_count, applicable_count, "
            "error_message, completed_at, updated_at "
            "FROM step_run WHERE project_id='p1' AND episode_id='e1' "
            "AND step_id='scene_detail'"
        )
    ).fetchone()
    if row is None:
        return None
    return {
        "status": row[0],
        "completed_count": row[1],
        "failed_count": row[2],
        "applicable_count": row[3],
        "error_message": row[4],
        "completed_at": row[5],
        "updated_at": row[6],
    }


def test_clean_redo_promotes_step_run_to_completed(
    db_with_partial_step_run, monkeypatch, tmp_path
):
    """Clean manifest after redo → step_run is promoted.

    Setup: existing manifest has 61 shots, one of which (S18_Shot12) is
    in contract_violation. Redo replaces that shot with a clean result
    (no ``status`` key). Service writes manifest as status='completed'
    + failed_count=0 and promotes step_run to status='completed'.
    """
    engine, db = db_with_partial_step_run

    existing_scenes: List[Dict[str, Any]] = [
        _make_shot_result(scene_index=i // 2 + 1, shot_index=i % 2 + 1)
        for i in range(60)
    ]
    # The 61st shot is the contract_violation one we will repair.
    existing_scenes.append(
        _make_shot_result(scene_index=18, shot_index=12, contract_violation=True)
    )
    # Clean replacement (no status key).
    redo_result = _make_shot_result(scene_index=18, shot_index=12)

    write_log, sync_log = _wire_service_internals(
        monkeypatch,
        existing_scenes=existing_scenes,
        redo_result=redo_result,
        cp_dir=tmp_path,
    )

    from app.services.scene_detail_redo_service import redo_scene_detail_shot

    ret = redo_scene_detail_shot(
        db=db,
        project_id="p1",
        episode_id="e1",
        scene_index=18,
        shot_index=12,
    )

    # Service return shape — W4b-2 + W4b-fixup additions.
    assert ret["ok"] is True
    assert ret["replaced"] is True
    assert ret["manifest_clean"] is True
    assert ret["dirty_count"] == 0
    assert ret["dirty_shot_indices"] == []
    assert ret["verify_clean"] is True
    assert ret["verify_reasons"] == []
    assert ret["checkpoint_sync_success"] is True
    assert ret["step_run_promoted"] is True
    assert ret["step_run_promoted_reason"] == ""

    # Manifest write reflects clean state.
    assert len(write_log) == 1
    manifest_payload = write_log[0]["payload"]
    assert manifest_payload["status"] == "completed"
    assert manifest_payload["failed_count"] == 0
    assert manifest_payload["completed_count"] == 61

    # checkpoint_sync was invoked.
    assert sync_log["called"] is True

    # step_run row promoted.
    row = _read_step_run(db)
    assert row is not None
    assert row["status"] == "completed"
    assert row["completed_count"] == 61
    assert row["failed_count"] == 0
    assert row["error_message"] is None
    assert row["completed_at"] is not None


def test_dirty_redo_preserves_step_run_status(
    db_with_partial_step_run, monkeypatch, tmp_path
):
    """Dirty manifest after redo → step_run status preserved.

    Setup: existing manifest has 60 clean shots + S18_Shot12
    contract_violation + S22_Shot1 contract_violation. Redo of S18_Shot12
    succeeds in producing a clean replacement, but S22_Shot1 remains
    dirty. Service must leave step_run.status='partial' and preserve
    error_message; only updated_at is refreshed.
    """
    engine, db = db_with_partial_step_run

    existing_scenes: List[Dict[str, Any]] = [
        _make_shot_result(scene_index=i // 2 + 1, shot_index=i % 2 + 1)
        for i in range(59)
    ]
    # Two dirty shots.
    existing_scenes.append(
        _make_shot_result(scene_index=18, shot_index=12, contract_violation=True)
    )
    existing_scenes.append(
        _make_shot_result(scene_index=22, shot_index=1, contract_violation=True)
    )

    # Redo of S18_Shot12 produces a clean shot — but S22_Shot1 is still
    # in the manifest as contract_violation, so manifest stays dirty.
    redo_result = _make_shot_result(scene_index=18, shot_index=12)

    write_log, sync_log = _wire_service_internals(
        monkeypatch,
        existing_scenes=existing_scenes,
        redo_result=redo_result,
        cp_dir=tmp_path,
    )

    # Snapshot step_run BEFORE the call so we can compare preservation.
    before = _read_step_run(db)
    assert before is not None
    assert before["status"] == "partial"
    assert before["error_message"] == "prior owned validation contract_violation"

    from app.services.scene_detail_redo_service import redo_scene_detail_shot

    ret = redo_scene_detail_shot(
        db=db,
        project_id="p1",
        episode_id="e1",
        scene_index=18,
        shot_index=12,
    )

    # Service return shape — dirty branch.
    assert ret["ok"] is True
    assert ret["replaced"] is True
    assert ret["manifest_clean"] is False
    assert ret["dirty_count"] == 1
    assert ret["dirty_shot_indices"] == [(22, 1)]
    assert ret["step_run_promoted"] is False
    # W4b-fixup: response includes the reason promote was skipped.
    assert "manifest dirty" in ret["step_run_promoted_reason"]
    # checkpoint_sync still ran; verify_clean is False because one shot
    # has status=contract_violation.
    assert ret["checkpoint_sync_success"] is True
    assert ret["verify_clean"] is False

    # Manifest top-level reflects dirty.
    assert len(write_log) == 1
    manifest_payload = write_log[0]["payload"]
    assert manifest_payload["status"] == "partial"
    assert manifest_payload["failed_count"] == 1
    assert manifest_payload["completed_count"] == 61

    # step_run row — status / error_message preserved; only updated_at moved.
    after = _read_step_run(db)
    assert after is not None
    assert after["status"] == "partial"
    assert after["error_message"] == "prior owned validation contract_violation"
    assert after["completed_at"] is None  # was never set
    # updated_at advanced (string comparison on ISO timestamps is OK).
    assert after["updated_at"] > before["updated_at"]


def test_clean_redo_clears_prior_error_message(
    db_with_partial_step_run, monkeypatch, tmp_path
):
    """Clean redo must NULL-out prior error_message.

    Specific guard: the prior partial row's error_message
    ("prior owned validation contract_violation") must not survive a
    clean promote, otherwise downstream UI would still flag the step
    as having a recent error.
    """
    engine, db = db_with_partial_step_run

    existing_scenes: List[Dict[str, Any]] = [
        _make_shot_result(scene_index=i + 1, shot_index=1)
        for i in range(60)
    ]
    existing_scenes.append(
        _make_shot_result(scene_index=18, shot_index=12, contract_violation=True)
    )
    redo_result = _make_shot_result(scene_index=18, shot_index=12)

    _wire_service_internals(
        monkeypatch,
        existing_scenes=existing_scenes,
        redo_result=redo_result,
        cp_dir=tmp_path,
    )

    from app.services.scene_detail_redo_service import redo_scene_detail_shot

    redo_scene_detail_shot(
        db=db,
        project_id="p1",
        episode_id="e1",
        scene_index=18,
        shot_index=12,
    )

    row = _read_step_run(db)
    assert row is not None
    assert row["error_message"] is None


# ---------------------------------------------------------------------------
# W4b-fixup IMPORTANT 1 — verify_clean must gate promote.
#
# Three negative tests: per-shot statuses are all clean (no
# contract_violation) AND the redo target shot itself is clean, BUT one
# OTHER shot in the manifest has a lightweight-verify-blocking defect:
#   - empty t2i_variations
#   - missing owned_validation sentinel
#   - sentinel shape invalid
# Expected: promote NOT fired, step_run.status preserved as 'partial',
# response surfaces verify_clean=False with a reasons list.
# ---------------------------------------------------------------------------


def test_verify_blocks_promote_when_other_shot_has_empty_t2i_variations(
    db_with_partial_step_run, monkeypatch, tmp_path
):
    """IMPORTANT 1: per-shot status all OK but one shot has empty
    t2i_variations → verify_clean=False → no promote."""
    engine, db = db_with_partial_step_run

    existing_scenes: List[Dict[str, Any]] = [
        _make_shot_result(scene_index=i + 1, shot_index=1)
        for i in range(60)
    ]
    # One stray shot with empty t2i_variations (e.g. an aborted prior
    # _analyze_one that left only the envelope). NOT marked as
    # contract_violation — status field absent. This is exactly the
    # "manifest looks per-shot-clean but verify must catch it" case.
    existing_scenes.append(
        _make_shot_result(
            scene_index=99, shot_index=1, empty_t2i_variations=True
        )
    )
    # The redo target is a different shot — clean replacement.
    redo_result = _make_shot_result(scene_index=18, shot_index=12)
    existing_scenes.append(
        _make_shot_result(scene_index=18, shot_index=12, contract_violation=True)
    )

    _wire_service_internals(
        monkeypatch,
        existing_scenes=existing_scenes,
        redo_result=redo_result,
        cp_dir=tmp_path,
    )

    before = _read_step_run(db)
    assert before is not None
    assert before["status"] == "partial"

    from app.services.scene_detail_redo_service import redo_scene_detail_shot

    ret = redo_scene_detail_shot(
        db=db,
        project_id="p1",
        episode_id="e1",
        scene_index=18,
        shot_index=12,
    )

    # manifest itself is "clean" per per-shot status (S18 repaired, S99
    # never had status=contract_violation), but verify catches the empty
    # t2i_variations.
    assert ret["manifest_clean"] is True
    assert ret["verify_clean"] is False
    assert any("empty t2i_variations" in r for r in ret["verify_reasons"])
    assert ret["step_run_promoted"] is False
    assert "lightweight verify failed" in ret["step_run_promoted_reason"]

    # step_run preserved.
    after = _read_step_run(db)
    assert after is not None
    assert after["status"] == "partial"
    assert (
        after["error_message"] == "prior owned validation contract_violation"
    )


def test_verify_blocks_promote_when_variation_lacks_owned_validation(
    db_with_partial_step_run, monkeypatch, tmp_path
):
    """IMPORTANT 1: a variation is missing owned_validation sentinel
    → verify_clean=False → no promote."""
    engine, db = db_with_partial_step_run

    existing_scenes: List[Dict[str, Any]] = [
        _make_shot_result(scene_index=i + 1, shot_index=1)
        for i in range(59)
    ]
    # One shot's variation has no owned_validation key at all.
    existing_scenes.append(
        _make_shot_result(
            scene_index=99, shot_index=1, omit_sentinel=True
        )
    )
    existing_scenes.append(
        _make_shot_result(scene_index=18, shot_index=12, contract_violation=True)
    )

    redo_result = _make_shot_result(scene_index=18, shot_index=12)

    _wire_service_internals(
        monkeypatch,
        existing_scenes=existing_scenes,
        redo_result=redo_result,
        cp_dir=tmp_path,
    )

    from app.services.scene_detail_redo_service import redo_scene_detail_shot

    ret = redo_scene_detail_shot(
        db=db,
        project_id="p1",
        episode_id="e1",
        scene_index=18,
        shot_index=12,
    )

    assert ret["manifest_clean"] is True
    assert ret["verify_clean"] is False
    assert any(
        "owned_validation missing" in r for r in ret["verify_reasons"]
    )
    assert ret["step_run_promoted"] is False

    after = _read_step_run(db)
    assert after is not None
    assert after["status"] == "partial"


def test_verify_blocks_promote_when_sentinel_shape_invalid(
    db_with_partial_step_run, monkeypatch, tmp_path
):
    """IMPORTANT 1: a variation has owned_validation present but shape
    invalid (missing required field) → verify_clean=False → no promote."""
    engine, db = db_with_partial_step_run

    existing_scenes: List[Dict[str, Any]] = [
        _make_shot_result(scene_index=i + 1, shot_index=1)
        for i in range(59)
    ]
    # Shot with a malformed sentinel (missing the required violations field
    # — assert_owned_sentinel_shape will raise).
    bad_shot = _make_shot_result(scene_index=99, shot_index=1)
    bad_sentinel = dict(_build_valid_sentinel("stub"))
    bad_sentinel.pop("violations")
    bad_shot["t2i_variations"][0]["owned_validation"] = bad_sentinel
    existing_scenes.append(bad_shot)
    existing_scenes.append(
        _make_shot_result(scene_index=18, shot_index=12, contract_violation=True)
    )

    redo_result = _make_shot_result(scene_index=18, shot_index=12)

    _wire_service_internals(
        monkeypatch,
        existing_scenes=existing_scenes,
        redo_result=redo_result,
        cp_dir=tmp_path,
    )

    from app.services.scene_detail_redo_service import redo_scene_detail_shot

    ret = redo_scene_detail_shot(
        db=db,
        project_id="p1",
        episode_id="e1",
        scene_index=18,
        shot_index=12,
    )

    assert ret["manifest_clean"] is True
    assert ret["verify_clean"] is False
    assert any(
        "owned_validation shape invalid" in r for r in ret["verify_reasons"]
    )
    assert ret["step_run_promoted"] is False


# ---------------------------------------------------------------------------
# W4b-fixup IMPORTANT 2 — checkpoint_sync_success AND rowcount==1 must
# both gate promote.
# ---------------------------------------------------------------------------


def test_sync_failure_blocks_promote(
    db_with_partial_step_run, monkeypatch, tmp_path
):
    """IMPORTANT 2: orchestrate_full_sync raises → promote NOT fired
    even though manifest_clean and verify_clean are True."""
    engine, db = db_with_partial_step_run

    existing_scenes: List[Dict[str, Any]] = [
        _make_shot_result(scene_index=i + 1, shot_index=1)
        for i in range(60)
    ]
    existing_scenes.append(
        _make_shot_result(scene_index=18, shot_index=12, contract_violation=True)
    )
    redo_result = _make_shot_result(scene_index=18, shot_index=12)

    _wire_service_internals(
        monkeypatch,
        existing_scenes=existing_scenes,
        redo_result=redo_result,
        cp_dir=tmp_path,
    )

    # Override orchestrate_full_sync to raise.
    import app.services.scene_detail_redo_service as svc

    def boom(project_id, episode_id, db, step_id=None):
        raise RuntimeError("sync boom")

    monkeypatch.setattr(svc, "orchestrate_full_sync", boom)

    from app.services.scene_detail_redo_service import redo_scene_detail_shot

    ret = redo_scene_detail_shot(
        db=db,
        project_id="p1",
        episode_id="e1",
        scene_index=18,
        shot_index=12,
    )

    assert ret["manifest_clean"] is True
    assert ret["verify_clean"] is True
    assert ret["checkpoint_sync_success"] is False
    assert ret["checkpoint_sync"] == {"error": "sync boom"}
    assert ret["step_run_promoted"] is False
    assert "checkpoint_sync failed" in ret["step_run_promoted_reason"]

    after = _read_step_run(db)
    assert after is not None
    assert after["status"] == "partial"


def test_missing_step_run_row_yields_promoted_false_with_reason(
    monkeypatch, tmp_path
):
    """IMPORTANT 2: UPDATE rowcount == 0 (step_run row absent) →
    promote function returns False with a reason mentioning row-not-
    found, response surfaces step_run_promoted=False."""
    # in-memory SQLite WITHOUT a pre-seeded scene_detail row.
    engine = create_engine("sqlite:///:memory:")
    with engine.begin() as conn:
        conn.execute(sql_text(_STEP_RUN_SCHEMA))
        # Insert a row for a DIFFERENT step_id so the UNIQUE constraint
        # exists but the redo target's UPDATE finds 0 rows.
        conn.execute(
            sql_text(
                "INSERT INTO step_run (id, project_id, episode_id, "
                "step_id, status, created_at, updated_at) VALUES "
                "(:id, 'p1', 'e1', 'shot_extract', 'completed', "
                "'2026-05-23T00:00:00+00:00', '2026-05-23T00:00:00+00:00')"
            ),
            {"id": str(uuid.uuid4())},
        )
    db = Session(engine)
    try:
        existing_scenes: List[Dict[str, Any]] = [
            _make_shot_result(scene_index=i + 1, shot_index=1)
            for i in range(60)
        ]
        existing_scenes.append(
            _make_shot_result(
                scene_index=18, shot_index=12, contract_violation=True
            )
        )
        redo_result = _make_shot_result(scene_index=18, shot_index=12)

        _wire_service_internals(
            monkeypatch,
            existing_scenes=existing_scenes,
            redo_result=redo_result,
            cp_dir=tmp_path,
        )

        from app.services.scene_detail_redo_service import (
            redo_scene_detail_shot,
        )

        ret = redo_scene_detail_shot(
            db=db,
            project_id="p1",
            episode_id="e1",
            scene_index=18,
            shot_index=12,
        )

        # Preconditions all met, but UPDATE rowcount==0 because there's
        # no scene_detail step_run row.
        assert ret["manifest_clean"] is True
        assert ret["verify_clean"] is True
        assert ret["checkpoint_sync_success"] is True
        assert ret["step_run_promoted"] is False
        assert "row not found" in ret["step_run_promoted_reason"]
        assert "rowcount=0" in ret["step_run_promoted_reason"]
    finally:
        db.close()


# ── 2026-08-26 Codex 2차 재리뷰 IMPORTANT ────────────────────────────


def test_도는_중이면_touch_를_했다고_보고하지_않는다():
    """★0줄인데 「touched」라고 적으면 안 한 일을 했다고 보고하는 것이다."""
    from unittest.mock import MagicMock

    from app.services import scene_detail_redo_service as svc

    db = MagicMock()
    db.execute.return_value.rowcount = 0      # scene_detail 이 도는 중

    out = svc._touch_step_run_updated_at(
        db=db, project_id="p1", episode_id="e1", reason="선행 조건 미달")

    assert out["promoted"] is False
    assert "skipped" in out["reason"], (
        f"아무것도 안 바꿨는데 그렇게 안 적었다: {out['reason']}")


def test_한_줄을_바꿨으면_그대로_보고한다():
    """양성 대조 — 정상 경로에서는 사유가 그대로 남아야 한다."""
    from unittest.mock import MagicMock

    from app.services import scene_detail_redo_service as svc

    db = MagicMock()
    db.execute.return_value.rowcount = 1

    out = svc._touch_step_run_updated_at(
        db=db, project_id="p1", episode_id="e1", reason="선행 조건 미달")

    assert out == {"promoted": False, "reason": "선행 조건 미달"}