"""The turn-cap checkpoint-park bundle (goal 28 task D2).

The daemon's FIRST turn-cap hit pushes `suspended` with reason
`turn_cap`; the web parks the task with the one-shot wrap-up notice
(immediately resume-offerable — no wake source needed) and stamps
`turn_cap_parked` so the resuming assignment carries the flag. The
next cap hit arrives as the ordinary failed runaway push and ends the
task for good.
"""
from harness import *  # noqa: F401,F403 — the shared wire helpers


def run(ctx):
    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token

    print("== goal 28 D2: the turn-cap checkpoint-park ==")
    fake.ask_on_register = False
    fake.hold_state = True

    task_id = create_queued(auth, {
        "project_id": "1", "title": "Runaway: parks at the first turn cap",
        "priority": "2"})
    ask_status, assignment = fake.ask_now()
    check(ask_status == 200 and int(assignment["task"]["id"]) == task_id,
          "the ask claims the queued task")
    fake.state = "working"
    check(fake.push_results("working") == 200, "the working push opens the round")

    # The first cap hit: a suspended transition carrying reason turn_cap
    # (the summary names the cap the way the daemon words it).
    fake.state = "suspended"
    check(fake.push_results("suspended",
                            summary="turn cap park: 400 assistant turns without "
                                    "reaching a terminal state (cap 400) — parked "
                                    "for wrap-up, not failed",
                            reason="turn_cap") == 200,
          "the turn-cap park transition lands")

    conn = db()
    row = conn.execute(
        "select status, pending_resume_notice, turn_cap_parked from tasks where id = ?",
        (task_id,)).fetchone()
    check(row is not None and row[0] == "suspended",
          "the task row parks suspended (not failed)")
    check(row is not None and row[1] is not None and "WRAP UP NOW" in row[1]
          and "stopped for good" in row[1],
          "the one-shot wrap-up notice is set on the row")
    check(row is not None and row[2] in (1, True),
          "the lineage flag turn_cap_parked is stamped")

    # The park is IMMEDIATELY resume-offerable (the notice is the lift):
    # the very next ask hands the SAME task back with the wrap-up as its
    # resume_notice and the turn_cap_parked flag riding the assignment.
    resume_status, resumed = fake.ask_now()
    check(resume_status == 200 and int(resumed["task"]["id"]) == task_id,
          "the parked task is immediately offerable again")
    check("Turn cap park" in (resumed.get("resume_notice") or ""),
          "the wrap-up notice rides the resuming assignment")
    check(resumed.get("turn_cap_parked") is True,
          "the assignment carries turn_cap_parked")

    # The notice is one-shot: consumed by the allocating assignment.
    conn = db()
    row = conn.execute(
        "select pending_resume_notice from tasks where id = ?", (task_id,)).fetchone()
    check(row is not None and row[0] is None,
          "the one-shot notice is consumed by the allocation")
    # The lineage flag SURVIVES the resume (a second cap hit must still
    # be final), and rides the fresh working round's next assignment.
    conn = db()
    row = conn.execute(
        "select turn_cap_parked from tasks where id = ?", (task_id,)).fetchone()
    check(row is not None and row[0] in (1, True),
          "the lineage flag survives the resume")

    # The resumed run reaches the cap again: the ordinary failed runaway
    # push — with turn_cap_parked on the row, this is the final kill.
    fake.state = "working"
    check(fake.push_results("working") == 200, "the resumed round opens")
    fake.state = "failed"
    check(fake.push_results("failed",
                            summary="runaway session: a resumed run reached the turn "
                                    "cap again — 400 assistant turns without a terminal "
                                    "state, cap 400 (SADDLE_TURN_CAP overrides it), "
                                    "aborting like the degeneration path") == 200,
          "the second cap hit's failed push lands")
    wait_for(lambda: task_state(auth, task_id).get("status") == "failed"
             or False, timeout=30, what="the runaway task failed for good")