"""The replay bundle (moved verbatim from the pre-split web_stack_test.py;
see test/harness.py for the shared machinery and the runner)."""
from harness import *  # noqa: F401,F403 — the shared wire helpers


# =============================================================================
# Bundle: replay — plan 13 B1/B2 shapes, rebuilt on the plan-17 ledger:
# the idempotent events push (UNIQUE(session_id, seq) — database state,
# so it survives a web restart), the session-churn redelivery shape, and
# repeated failure pushes that must still project exactly once.
# =============================================================================

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

    def rows_for(session):
        conn = db()
        try:
            return conn.execute(
                "select count(*) from events where session_id = ?",
                (session,)).fetchone()[0]
        finally:
            conn.close()

    def open_session_on_fake(task_id):
        """A fixture open session row on the shard's saddle (the churn
        shape: a fresh web-side run row the daemon never really drove)."""
        conn = db()
        try:
            conn.execute(
                "insert into sessions (task_id, saddle_id, started_at)"
                " values (?, ?, strftime('%s','now'))", (task_id, fake_id))
            conn.commit()
            return conn.execute(
                "select max(id) from sessions where task_id = ?",
                (task_id,)).fetchone()[0]
        finally:
            conn.close()

    print("== plan 17 §7.2: event pushes are idempotent per session ==")
    # A first task runs to completion; its session is the replay source.
    first = create_queued(auth, {"project_id": "1", "title": "Replay source task",
                                 "priority": "2"})
    check(wait_task_done(auth, first, timeout=60) is not None,
          "the replay-source task completes")
    conn = db()
    src_session = conn.execute(
        "select max(id) from sessions where task_id = ?", (first,)).fetchone()[0]
    conn.close()

    # The outbox batch: two events the daemon already delivered once.
    burst = [
        {"seq": 5001, "kind": "tool_call", "actor": "model",
         "payload": {"turn": 7, "tool": "run_command",
                     "args": {"command": "make test"}}},
        {"seq": 5002, "kind": "tool_result", "actor": "harness",
         "payload": {"turn": 7, "stdout": "ok", "exit_code": 0}},
    ]
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": src_session, "events": burst})
    check(status == 200 and int(body.get("stored") or 0) == 2,
          "the event batch stores under its session")
    stored_once = rows_for(src_session)

    # Redelivery under the SAME session: consumed, nothing new.
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": src_session, "events": burst})
    check(status == 200 and int(body.get("stored") or 0) == 0
          and rows_for(src_session) == stored_once,
          "a same-session redelivery stores nothing new (consumed silently)")

    # Session churn: a mis-keyed outbox send under a brand-new session
    # id replays the SAME seqs. The ledger's idempotency key is
    # (session_id, seq) — the churned session is a different key, so
    # the events store under it (the daemon contract makes the session
    # id part of the key; the old daemon-global entry dedup is gone).
    conn = db()
    conn.execute(
        "insert into tasks (project_id, title, status, priority, created_at,"
        " completed_at) values (1, 'Churn target task', 'done', 5,"
        " strftime('%s','now'), strftime('%s','now'))")
    conn.commit()
    churn_task = conn.execute(
        "select max(id) from tasks where title = 'Churn target task'").fetchone()[0]
    conn.close()
    churn_session = open_session_on_fake(churn_task)
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": churn_session, "events": burst})
    check(status == 200 and int(body.get("stored") or 0) == 2
          and rows_for(churn_session) == 2,
          "a churned-session replay stores under the new key (per-session seq)")

    # Web restart: the idempotency survives (database constraint, not
    # memory) — the same-session redelivery still stores nothing.
    app.terminate()
    app.wait(timeout=10)
    app = start_app()
    auth = login("admin", "adminpass123")
    ctx.app = app  # teardown tracks the restarted process
    ctx.auth = auth
    # The restart wiped the in-memory ask bookkeeping; the daemon
    # re-establishes sleeping with a fresh ask (the real daemon's
    # idle-awake ask / 15-minute safety self-ask).
    fake.ask_now()
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": churn_session, "events": burst})
    check(status == 200 and int(body.get("stored") or 0) == 0
          and rows_for(churn_session) == 2,
          "a post-restart redelivery still stores nothing (persistent)")

    print("== plan 13 B2: duplicate failed pushes project once (task) ==")
    # The post-mortem's honest burst, rebuilt on the state channel: the
    # failed state redelivered repeatedly — one transition, one
    # summary, ever (terminal pushes are consumed silently).
    def agent_comments(task_id, body=None):
        conn = db()
        try:
            if body is None:
                return conn.execute(
                    "select count(*) from task_comments"
                    " where task_id = ? and author = 'agent'",
                    (task_id,)).fetchone()[0]
            return conn.execute(
                "select count(*) from task_comments"
                " where task_id = ? and author = 'agent' and body = ?",
                (task_id, body)).fetchone()[0]
        finally:
            conn.close()

    conn = db()
    conn.execute(
        "insert into tasks (project_id, title, status, priority, created_at)"
        " values (1, 'Burst failure task', 'working', 5, strftime('%s','now'))")
    conn.commit()
    burst_task = conn.execute(
        "select max(id) from tasks where title = 'Burst failure task'").fetchone()[0]
    conn.close()
    burst_session = open_session_on_fake(burst_task)
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": burst_session, "state": "failed",
                            "summary": "bash -n rejected the script"})
    check(status == 200
          and task_state(auth, burst_task)["status"] == "failed",
          "the failure state push fails the task")
    check(agent_comments(burst_task, "bash -n rejected the script") == 0
          and "bash -n rejected the script"
          in (task_state(auth, burst_task).get("summary") or ""),
          "the failure explanation lives ONCE — in tasks.summary, the terminal cache (S4: no duplicate agent comment)")

    # The burst: the same failure redelivered under the session AND a
    # churned session — nothing re-runs.
    churned_burst = open_session_on_fake(burst_task)
    for target in (burst_session, churned_burst):
        status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                               {"session_id": target, "state": "failed",
                                "summary": "bash -n rejected the script"})
        check(status == 200,
              "the failed-state push under %s is consumed, not rejected" % (
                  "its session" if target == burst_session else "a churned session"))
    check(task_state(auth, burst_task)["status"] == "failed"
          and agent_comments(burst_task) == 0
          and "bash -n rejected the script"
          in (task_state(auth, burst_task).get("summary") or ""),
          "duplicate failed pushes leave one transition, one summary, zero agent comments (S4)")

    print("== plan 13 B2: goal failure projects once ==")
    # Goal context: `saddle failure` fails the goal ONCE — the 115
    # goal-failed events of the live failure are structurally gone.
    def goal_failed_events(goal_id):
        # Plan 15 C3: the narrative derives at read time — the
        # failed-goal row IS the goal-failed record (one row, ever).
        return len([e for e in goal_state(goal_id)["events"]
                    if e["kind"] == "goal-failed"])

    act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "Burst goal", "brief": "Fails honestly.",
        "branch_name": "goal/burst"})
    conn = db()
    burst_goal = conn.execute(
        "select id from goals where title = 'Burst goal'").fetchone()[0]
    # (Parking guard: the scripted park needs a live wake source.)
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, mode, creator, created_at) values (1, ?, 'task',"
                 " 'Burst live worker', 'blocked', 5, 'implement', 'goal',"
                 " strftime('%s','now'))", (burst_goal,))
    conn.commit()
    conn.close()
    fake.goal_script = [("suspend",)]
    act(auth, goal_state(burst_goal)["start"], {})
    conn = db()
    run_task = conn.execute(
        "select id from tasks where goal_id = ? and kind = 'goal_run'",
        (burst_goal,)).fetchone()[0]
    conn.close()
    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                   or False, timeout=60, what="burst goal agent parks") is not None,
          "the burst goal's agent parks first")
    conn = db()
    run_session = conn.execute(
        "select max(id) from sessions where task_id = ?",
        (run_task,)).fetchone()[0]
    conn.close()

    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": run_session, "state": "failed",
                            "summary": "the harness collapsed"})
    check(status == 200 and goal_state(burst_goal)["status"] == "failed",
          "the goal agent's failure fails the goal")
    check(goal_failed_events(burst_goal) == 1,
          "the failure push records exactly one goal-failed event")

    # Redeliveries — under the run's session AND a churned session —
    # plus the bare failed state again: still one.
    churned_run = open_session_on_fake(run_task)
    for target in (run_session, churned_run):
        status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                               {"session_id": target, "state": "failed",
                                "summary": "the harness collapsed"})
        check(status == 200, "the goal-failure redelivery under %s is consumed"
              % ("its session" if target == run_session else "a churned session"))
    check(goal_state(burst_goal)["status"] == "failed"
          and goal_failed_events(burst_goal) == 1,
          "redelivered failures never re-run the goal failure (one event, ever)")
    check(task_state(auth, run_task)["status"] == "failed",
          "the goal-run task stays failed after the replays")
    conn = db()
    burst_blocks = conn.execute(
        "select count(*) from task_blocks where task_id = ?", (run_task,)
    ).fetchone()[0]
    conn.close()
    check(burst_blocks == 0,
          "the goal-failed path raises ZERO blocks (nothing pins the pile)")

    print("== plan 17: agent remarks ride the ledger as comment events ==")
    # The heredoc `cat << 'EOF' | saddle comment` shape: the daemon
    # emits a comment event carrying the body. The body is the record —
    # no command echo ever renders, and the goal page's current-activity
    # line reads the newest remark.
    act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "Comment goal", "brief": "Reports cleanly.",
        "branch_name": "goal/comment"})
    conn = db()
    comment_goal = conn.execute(
        "select id from goals where title = 'Comment goal'").fetchone()[0]
    # (Parking guard: the scripted park needs a live wake source.)
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, mode, creator, created_at) values (1, ?, 'task',"
                 " 'Comment live worker', 'blocked', 5, 'implement', 'goal',"
                 " strftime('%s','now'))", (comment_goal,))
    conn.commit()
    conn.close()
    fake.goal_script = [("suspend",)]
    act(auth, goal_state(comment_goal)["start"], {})
    conn = db()
    comment_run = conn.execute(
        "select id from tasks where goal_id = ? and kind = 'goal_run'",
        (comment_goal,)).fetchone()[0]
    conn.close()
    check(wait_for(lambda: task_state(auth, comment_run)["status"] == "suspended"
                   or False, timeout=60, what="comment goal agent parks") is not None,
          "the comment goal's agent parks")
    conn = db()
    comment_session = conn.execute(
        "select max(id) from sessions where task_id = ?",
        (comment_run,)).fetchone()[0]
    conn.close()

    comment_body = "The flange weld holds. First sentence carries the information."
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": comment_session,
                               "events": [{
                                   "seq": 8001, "kind": "comment",
                                   "actor": "model",
                                   "payload": {"body": comment_body}}]})
    check(status == 200 and int(body.get("stored") or 0) == 1,
          "the comment event stores on the ledger")
    conn = db()
    tables = {r[0] for r in conn.execute(
        "select name from sqlite_master where type = 'table'").fetchall()}
    comment_events = conn.execute(
        "select count(*) from events where scope_id = ? and kind = 'comment'"
        " and actor = 'model'",
        (comment_run,)).fetchone()[0]
    conn.close()
    check(comment_events == 1
          and "goal_comments" not in tables,
          "the ledger carries the remark (the goal_comments table is gone;"
          " system bookkeeping comments are separate)")
    narrative_comments = [e["text"] for e in goal_state(comment_goal)["events"]
                          if e["kind"] == "comment"]
    check(len(narrative_comments) == 1
          and narrative_comments[0].startswith("The flange weld holds.")
          and "saddle comment" not in narrative_comments[0]
          and "EOF" not in narrative_comments[0],
          "the derived narrative's comment rows carry the body text")

    state = goal_state(comment_goal)
    check(state["current_command"] == comment_body,
          "the goal page's current-activity line shows the remark body")
    check(any(e["kind"] == "comment"
              and e["text"].startswith("The flange weld holds.")
              for e in state["events"]),
          "the goal slot carries the remark in its narrative")
    page = http_get_page(auth, "/goals/%d" % comment_goal)
    check("saddle comment" not in page,
          "the rendered goal page never shows the command echo")
    check("event.kind == 'comment' ? ' footnote'" in page,
          "the narrative's comment events render with footnote styling")

    print("== P1/P2: event ledger reader indexes + batched ingestion ==")
    # P1 (i): a FRESH database (this shard's app migrated it at
    # startup) carries both M0003 reader indexes, and the
    # scope-filtered reader shapes actually use them — no full
    # 'SCAN events' for the page/cursor walks or the resume
    # coalescer's terminal-transition scan.
    def event_indexes():
        conn = db()
        try:
            return {r[0] for r in conn.execute(
                "select name from sqlite_master"
                " where type = 'index' and tbl_name = 'events'").fetchall()}
        finally:
            conn.close()

    def query_plan(sql, params=()):
        conn = db()
        try:
            return " ".join(r[3] for r in conn.execute(
                "explain query plan " + sql, params).fetchall())
        finally:
            conn.close()

    check({"idx_events_scope", "idx_events_terminal"} <= event_indexes(),
          "M0003 creates both event reader indexes on a fresh database")
    scope_plan = query_plan(
        "select id from events where scope_type = 'task' and scope_id = ?"
        " and id < ? order by id desc limit 30", (first, 2 ** 40))
    check("idx_events_scope" in scope_plan and "SCAN events" not in scope_plan,
          "the scope-filtered readers search idx_events_scope (no table scan)")
    terminal_plan = query_plan(
        "select id, created_at from events where kind = 'state_change'"
        " and scope_type = 'task' and scope_id = ?", (first,))
    check("idx_events_terminal" in terminal_plan
          and "SCAN events" not in terminal_plan,
          "the terminal-transition reader searches the partial idx_events_terminal")

    # P2 (iii): batched ingestion stays idempotent — a redelivered
    # batch stores 0 new rows and errors nothing, and a MIXED batch
    # (one duplicate + one new event) stores exactly the new row.
    batch_target = open_session_on_fake(first)
    one_event = [{"seq": 6001, "kind": "progress",
                  "payload": {"text": "batched"}}]
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": batch_target, "events": one_event})
    check(status == 200 and int(body.get("stored") or 0) == 1,
          "a fresh batched event stores one row")
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": batch_target,
                               "events": one_event + [
                                   {"seq": 6002, "kind": "progress",
                                    "payload": {"text": "second"}}]})
    check(status == 200 and int(body.get("stored") or 0) == 1,
          "a mixed redelivery batch stores only the new row, no error")
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": batch_target, "events": one_event})
    check(status == 200 and int(body.get("stored") or 0) == 0,
          "a redelivered batch stores 0 new rows and errors nothing")

    # P2 (iv): a batch containing one invalid event is refused WHOLE
    # — staging validates before any insert, so nothing partially
    # stores (the invalid event sits between two valid ones).
    conn = db()
    target_rows = conn.execute(
        "select count(*) from events where session_id = ?",
        (batch_target,)).fetchone()[0]
    conn.close()
    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
                              {"session_id": batch_target, "events": [
                                  {"seq": 6101, "kind": "progress",
                                   "payload": {}},
                                  {"seq": 0, "kind": "progress",
                                   "payload": {}},
                                  {"seq": 6102, "kind": "progress",
                                   "payload": {}}]})
    check(status == 400, "a batch with an invalid event is refused (HTTP 400)")
    conn = db()
    after_rows = conn.execute(
        "select count(*) from events where session_id = ?",
        (batch_target,)).fetchone()[0]
    conn.close()
    check(after_rows == target_rows,
          "the refused batch stores nothing (no partial storage)")

    # P1 (ii): a serial-2 database upgraded in place gains the
    # indexes — strip M0003's record and both indexes off this
    # shard's database, restart the app, and the startup migration
    # must reapply them over the existing ledger.
    conn = db()
    conn.execute("drop index if exists idx_events_scope")
    conn.execute("drop index if exists idx_events_terminal")
    conn.execute("delete from __migrations"
                 " where namespace = 'clanker-saddle' and serial = 3")
    conn.commit()
    conn.close()
    app.terminate()
    app.wait(timeout=10)
    app = start_app()
    auth = login("admin", "adminpass123")
    ctx.app = app
    ctx.auth = auth
    conn = db()
    serial3 = conn.execute(
        "select count(*) from __migrations"
        " where namespace = 'clanker-saddle' and serial = 3").fetchone()[0]
    kept_rows = conn.execute(
        "select count(*) from events where session_id = ?",
        (batch_target,)).fetchone()[0]
    conn.close()
    check({"idx_events_scope", "idx_events_terminal"} <= event_indexes()
          and serial3 == 1,
          "a serial-2 database upgraded in place gains both indexes")
    check(kept_rows == after_rows,
          "the in-place upgrade leaves the ledger's rows untouched")

    # Drain: nothing queued remains for the shard's shutdown.
    check(queued_count() == 0, "the replay shard drains its queue")