clanker-saddle/test/bundles/replay.py

"""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: the daemon-global entry dedup and the
# idempotent terminal/narrative projections. Replays the live-failure
# post-mortem shapes directly against the push channels: an outbox
# redelivery arriving under the WRONG session id after churn (and after
# a web restart), and repeated failure pushes that must still project
# exactly once. Includes this shard's one deliberate app restart (the
# dedup must survive it — it is database state, not memory).
# =============================================================================

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 session_commands where session_id = ?",
                (session,)).fetchone()[0]
        finally:
            conn.close()

    def dedup_rows():
        conn = db()
        try:
            return conn.execute("select count(*) from saddle_entry_dedup").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 13 B1: entry dedup is daemon-global, not 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 command entries the daemon already
    # delivered once (their ids landed with the source session).
    burst = [
        {"id": 5001, "command": "make test", "output": "ok", "exit_code": 0,
         "type": "command", "timestamp": "2026-09-11T00:00:00Z"},
        {"id": 5002, "command": "grep -r flange src/", "output": "src/flange.vala",
         "exit_code": 0, "type": "command", "timestamp": "2026-09-11T00:00:01Z"},
    ]
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": src_session, "state": "done",
                            "entries": burst})
    check(status == 200 and rows_for(src_session) >= 2,
          "the entries batch stores under its session")
    stored_once = rows_for(src_session)
    dedup_after_first = dedup_rows()

    # Redelivery under the SAME session: consumed, nothing new.
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": src_session, "state": "done",
                            "entries": burst})
    check(status == 200 and rows_for(src_session) == stored_once
          and dedup_rows() == dedup_after_first,
          "a same-session redelivery stores nothing new (consumed silently)")

    # Session churn: a brand-new session id on the same saddle replays
    # the SAME entry ids — the post-mortem's mis-keyed outbox send.
    # The entries' ids already populated another session, so the push
    # may not create rows here either.
    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, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": churn_session, "state": "done",
                            "entries": burst})
    check(status == 200 and rows_for(churn_session) == 0,
          "replay-under-wrong-session stores nothing new (daemon-global dedup)")
    check(dedup_rows() == dedup_after_first,
          "the consumed redelivery claims no fresh dedup rows")

    # A fresh entry id under the churned session DOES store — the dedup
    # keys on the entry id, never on the session.
    fresh = [{"id": 5003, "command": "git status", "output": "clean",
              "exit_code": 0, "type": "command", "timestamp": "2026-09-11T00:00:02Z"}]
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": churn_session, "state": "done",
                            "entries": fresh})
    check(status == 200 and rows_for(churn_session) == 1
          and dedup_rows() == dedup_after_first + 1,
          "a fresh entry id stores under any session of its saddle")

    # Web restart: the dedup survives (database state, not memory).
    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, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": churn_session, "state": "done",
                            "entries": burst + fresh})
    check(status == 200 and rows_for(churn_session) == 1
          and rows_for(src_session) == stored_once,
          "a post-restart redelivery still stores nothing (dedup is persistent)")

    print("== plan 13 B2: duplicate failed entries project once (task) ==")
    # The post-mortem's honest burst: the same failure entry redelivered
    # under a churned session AND a second, DISTINCT failure entry id
    # in the same session — one transition, one comment, ever.
    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)
    first_fail = [{"id": 6001, "command": "saddle failure",
                   "output": "bash -n rejected the script", "exit_code": 1,
                   "type": "failed", "timestamp": "2026-09-11T00:01:00Z"}]
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": burst_session, "state": "failed",
                            "entries": first_fail,
                            "summary": "bash -n rejected the script"})
    check(status == 200
          and task_state(auth, burst_task)["status"] == "failed",
          "the failure entry + state pair 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: a DISTINCT failure entry id in the same session, plus
    # both ids redelivered under a churned session — nothing re-runs.
    second_fail = [{"id": 6002, "command": "saddle failure",
                    "output": "bash -n rejected the script", "exit_code": 1,
                    "type": "failed", "timestamp": "2026-09-11T00:01:01Z"}]
    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",
                                "entries": first_fail + second_fail,
                                "summary": "bash -n rejected the script"})
        check(status == 200,
              "the burst 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 entries 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]
    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()

    goal_fail = [{"id": 7001, "command": "saddle failure",
                  "output": "the harness collapsed", "exit_code": 1,
                  "type": "failed", "timestamp": "2026-09-11T00:02:00Z"},
                 {"id": 7002, "command": "saddle failure",
                  "output": "the harness collapsed", "exit_code": 1,
                  "type": "failed", "timestamp": "2026-09-11T00:02:01Z"}]
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": run_session, "state": "failed",
                            "entries": goal_fail, "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,
          "a two-entry failure burst records exactly one goal-failed event")

    # Redeliveries — same ids under the run's session AND under 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",
                                "entries": goal_fail,
                                "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 13 B4: the comment projection stores the body ==")
    # The heredoc shape: the command is the pipe invocation, the OUTPUT
    # is the body. The projections must carry the body — the command
    # echo never becomes comment text, and the literal `saddle
    # comment` string never renders outside the terminal.
    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]
    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."
    comment_entry = [{"id": 8001, "command": "cat << 'EOF' | saddle comment",
                      "output": comment_body, "exit_code": 0, "type": "comment",
                      "timestamp": "2026-09-11T00:03:00Z"}]
    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                           {"session_id": comment_session, "state": "suspended",
                            "entries": comment_entry})
    check(status == 200, "the comment entry push is accepted")
    conn = db()
    task_comment_bodies = [r[0] for r in conn.execute(
        "select body from task_comments where task_id = ?", (comment_run,))]
    tables = {r[0] for r in conn.execute(
        "select name from sqlite_master where type = 'table'").fetchall()}
    conn.close()
    check(task_comment_bodies == [comment_body],
          "the run task's comment stores the body, not the command echo")
    check("goal_comments" not in tables,
          "the goal_comments table is gone (the run task's rows are the thread)")
    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 comment body")
    check(state["comments"] and state["comments"][0]["body"] == comment_body
          and any(e["kind"] == "comment"
                  and e["text"].startswith("The flange weld holds.")
                  for e in state["events"]),
          "the goal slot carries the body in its comments and 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")
    check(page.index('stm-markdown="comment.body"')
          < page.index('stm-text="comment.quote"'),
          "the comment card shows the body before the quote")

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