replay: daemon-global entry dedup — ingestion keys on (saddle, entry id)

The per-session in-memory entry cursor accepted outbox redeliveries under
any FRESH session id (the post-mortem's 286 duplicates after session
churn) and forgot everything on a web restart. Ingestion now claims each
pushed entry id per saddle in the persistent saddle_entry_dedup table
(unique pair, inserted before the entry stores): a redelivery under any
session binding — restart, churn, mis-keyed send — is consumed silently,
and only freshly-claimed entries store and project.

src/Api/SaddleApi.vala

@@ -1074,8 +1074,11 @@ namespace ClankerSaddle.Api {
      * ```
      *
      * Entries keep today's OutputEntry shape and store like the deleted
-     * /output poll path (the daemon's global entry ids make redeliveries
-     * idempotent). The state transitions drive the same bookkeeping the
+     * /output poll path; plan 13 B1 makes the daemon's global entry
+     * ids a DAEMON-GLOBAL (saddle, id) dedup persisted in the database
+     * — a redelivery under any session id (restart, churn, a mis-keyed
+     * send) is consumed silently, and only freshly-claimed entries
+     * project. The state transitions drive the same bookkeeping the
      * /status poll switch ran: `blocked` raises the block payload as an
      * answerable `task_blocks` row (the pulse bump makes the question
      * appear on live pages) and parks the saddle; `suspended` parks for
@@ -1126,7 +1129,7 @@ namespace ClankerSaddle.Api {
                 var response = new Json.Object();
                 if (body.has_member("entries")
                         && body.get_member("entries").get_node_type() == Json.NodeType.ARRAY) {
-                    var stored = yield saddles.store_pushed_entries(session_id,
+                    var stored = yield saddles.store_pushed_entries((!)saddle, session_id,
                         body.get_array_member("entries"));
                     yield saddles.apply_entry_events(((!)session).task_id, stored, tasks);
                 }

src/Data/Entities.vala

@@ -246,6 +246,28 @@ namespace ClankerSaddle.Data {
         }
     }
 
+    /**
+     * One daemon-global entry id already ingested for a saddle (plan
+     * 13 B1): the (saddle, entry id) pair is the replay-integrity key.
+     * An outbox redelivery under ANY session id — a web restart, session
+     * churn, a mis-keyed send — finds its row already here and is
+     * consumed silently; the pair survives daemon restarts because it
+     * lives in the database, not in memory.
+     */
+    public class SaddleEntryDedup : Object {
+
+        public int64 id { get; set; }
+        public int64 saddle_id { get; set; }
+        public int64 entry_id { get; set; }
+
+        public static void entity_mapping(EntityMapperBuilder<SaddleEntryDedup> cfg) {
+            cfg.table("saddle_entry_dedup")
+                .column<int64?>("id", o => o.id, (o, v) => o.id = v)
+                .column<int64?>("saddle_id", o => o.saddle_id, (o, v) => o.saddle_id = v)
+                .column<int64?>("entry_id", o => o.entry_id, (o, v) => o.entry_id = v);
+        }
+    }
+
     /**
      * A numbered entry captured during a {@link Session} — a bash
      * command/output pair or a conversation-level event (comment, block,

src/Data/Migrations/M0001_Initial.vala

@@ -15,8 +15,10 @@ namespace ClankerSaddle.Data.Migrations {
      * and the draft-lifecycle columns (`status` defaults to
      * 'draft'); saddle registrations with git identity
      * and the per-saddle token; the session + command + raw-log audit
-     * trail; rules and their attachments; project notes; capability
-     * tags; and the plan-11 notification tables for task E
+     * trail; the plan-13 daemon-global entry dedup
+     * (`saddle_entry_dedup`); rules and their attachments; project
+     * notes; capability tags; and the plan-11 notification tables for
+     * task E
      * (`push_subscriptions`, `task_notify_prefs` — included now so
      * D+E needs no migration).
      */
@@ -324,6 +326,25 @@ namespace ClankerSaddle.Data.Migrations {
                 t.column<int64?>("duration_ms");
             });
 
+            // Plan 13 B1: the daemon-global entry dedup — every pushed
+            // entry id this saddle has already ingested. The unique
+            // (saddle_id, entry_id) pair is the replay-integrity key: a
+            // redelivered entry under ANY session id (web restart,
+            // session churn, a mis-keyed send) finds its row here and is
+            // consumed silently.
+            b.create_table("saddle_entry_dedup", (t) => {
+                t.column<int64?>("id")
+                    .auto_increment()
+                    .primary_key();
+                t.column<int64?>("saddle_id")
+                    .not_null()
+                    .references("saddles", "id")
+                    .indexed();
+                t.column<int64?>("entry_id")
+                    .not_null();
+                t.unique("uq_saddle_entry_dedup_pair", {"saddle_id", "entry_id"});
+            });
+
             b.create_table("api_tokens", (t) => {
                 t.column<int64?>("id")
                     .auto_increment()
@@ -501,6 +522,7 @@ namespace ClankerSaddle.Data.Migrations {
         }
 
         public override void down(MigrationBuilder b) {
+            b.drop_table("saddle_entry_dedup");
             b.drop_table("goal_events");
             b.drop_table("goal_blocks");
             b.drop_table("goal_comments");

src/Services/SaddlePollState.vala

@@ -4,9 +4,7 @@ using Inversion;
 namespace ClankerSaddle.Services {
 
     /**
-     * Process-wide saddle poll bookkeeping: the last daemon-side output
-     * entry id stored per session (so redelivered results pushes never
-     * duplicate `session_commands` rows), the block answers and human
+     * Process-wide saddle poll bookkeeping: the block answers and human
      * comments already handed to a saddle (each delivered exactly once;
      * a restart re-delivers and the daemon deduplicates), plus the
      * plan-10 ask/wake liveness bookkeeping: the outcome of a saddle's
@@ -17,6 +15,11 @@ namespace ClankerSaddle.Services {
      * (the staleness probe's timer — one exceptional probe per silent
      * episode, reset by the next push).
      *
+     * The plan-13 entry-ingestion dedup deliberately does NOT live
+     * here: it is daemon-global and persistent (the
+     * `saddle_entry_dedup` table), so it survives restarts and session
+     * churn — see SaddleService.store_pushed_entries.
+     *
      * A singleton by necessity: {@link SaddleService} is request-scoped,
      * so this object is where the cross-request cursors live. Ids are
      * stringified: the Invercargill collections cannot key on value types.
@@ -27,7 +30,6 @@ namespace ClankerSaddle.Services {
         private Dictionary<string, string> probe_uptime = new Dictionary<string, string>();
         private Dictionary<string, string> probe_activity = new Dictionary<string, string>();
         private Dictionary<string, string> probe_contact = new Dictionary<string, string>();
-        private Dictionary<string, string> last_stored_output_id = new Dictionary<string, string>();
         private HashSet<string> delivered_answers = new HashSet<string>();
         private HashSet<string> delivered_comments = new HashSet<string>();
         private HashSet<string> resumed_notified = new HashSet<string>();
@@ -40,22 +42,6 @@ namespace ClankerSaddle.Services {
         private HashSet<string> stale_probed = new HashSet<string>();
         private Mutex state_lock = Mutex();
 
-        /** The last output entry id stored for `session_id` (0 when none). */
-        public int64 since_for(int64 session_id) {
-            state_lock.lock();
-            string stored = "";
-            bool found = last_stored_output_id.try_get(session_id.to_string(), out stored);
-            state_lock.unlock();
-            return found ? int64.parse(stored) : 0;
-        }
-
-        /** Records the new high-water output entry id for `session_id`. */
-        public void store_since(int64 session_id, int64 entry_id) {
-            state_lock.lock();
-            last_stored_output_id.set(session_id.to_string(), entry_id.to_string());
-            state_lock.unlock();
-        }
-
         /**
          * Records a daemon /status observation for the saddles-page
          * health panel: reported state, uptime, last-activity epoch and

src/Services/SaddleService.vala

@@ -1475,19 +1475,20 @@ namespace ClankerSaddle.Services {
         /**
          * Stores a pushed batch of round entries (`POST
          * /api/saddle/results`, plan 10 phase 2) as `session_commands`
-         * rows — the mapping the deleted `GET /output` poller ran. The
-         * daemon's global entry ids (today's OutputEntry shape keeps
-         * them) drive the per-session cursor in {@link
-         * SaddlePollState}: a redelivered batch (an outbox retry after
-         * a lost response) skips the entries the cursor already
-         * covers, so at-least-once pushes store exactly once. Entries
-         * without an id always store. Returns the entries stored,
-         * oldest first, for {@link apply_entry_events}.
+         * rows — the mapping the deleted `GET /output` poller ran. Plan
+         * 13 B1: the dedup key is the DAEMON-GLOBAL (saddle, entry id)
+         * pair ({@link SaddleEntryDedup}), persisted in the database so
+         * it survives web restarts and ignores session churn entirely —
+         * a redelivered entry is consumed silently under whatever
+         * session id the payload carries, and a push may only CREATE
+         * rows for entries whose ids have not already landed on this
+         * saddle. Entries without an id always store. Returns the
+         * entries stored, oldest first, for {@link apply_entry_events}
+         * — the deduped entry stream every projection keys on (B2).
          */
-        public async Vector<Json.Object> store_pushed_entries(int64 session_id,
+        public async Vector<Json.Object> store_pushed_entries(Saddle saddle, int64 session_id,
                 Json.Array entries) throws Error {
             var stored = new Vector<Json.Object>();
-            int64 since = poll_state.since_for(session_id);
             foreach (var element in entries.get_elements()) {
                 if (element.get_node_type() != Json.NodeType.OBJECT) {
                     continue;
@@ -1496,7 +1497,9 @@ namespace ClankerSaddle.Services {
                 int64 entry_id = entry.has_member("id")
                     && entry.get_member("id").get_value_type() == typeof(int64)
                     ? entry.get_int_member("id") : 0;
-                if (entry_id > 0 && entry_id <= since) {
+                if (entry_id > 0 && !(yield claim_entry_id(saddle.id, entry_id))) {
+                    // A redelivery under any session binding — consumed,
+                    // never duplicated, never projected.
                     continue;
                 }
                 var command = new SessionCommand() {
@@ -1519,17 +1522,38 @@ namespace ClankerSaddle.Services {
                 };
                 yield db.insert_async<SessionCommand>(command);
                 stored.add(entry);
-                if (entry_id > 0) {
-                    // Advance the persisted cursor per row: a failure
-                    // mid-pass must not re-store the entries already
-                    // inserted on the daemon's retry.
-                    since = entry_id;
-                    poll_state.store_since(session_id, since);
-                }
             }
             return stored;
         }
 
+        /**
+         * Claims `(saddle_id, entry_id)` for ingestion: true when the
+         * id is fresh on this saddle, false when it already landed (an
+         * outbox redelivery — consumed silently regardless of the
+         * payload's session binding). The dedup row is inserted BEFORE
+         * the entry stores, so a mid-pass failure can never re-project
+         * an entry the daemon retries (exactly-once projections beat
+         * at-least-once storage).
+         */
+        private async bool claim_entry_id(int64 saddle_id, int64 entry_id) throws Error {
+            try {
+                yield db.insert_async<SaddleEntryDedup>(new SaddleEntryDedup() {
+                    saddle_id = saddle_id,
+                    entry_id = entry_id
+                });
+                return true;
+            } catch (Error insert_error) {
+                var existing = yield db.query<SaddleEntryDedup>()
+                    .where(expr("saddle_id == $0 && entry_id == $1",
+                        elem<int64?>(saddle_id), elem<int64?>(entry_id)))
+                    .first_async();
+                if (existing == null) {
+                    throw insert_error;
+                }
+                return false;
+            }
+        }
+
         /**
          * Applies a pushed state transition (`POST /api/saddle/results`,
          * plan 10 phase 2) — the same task/saddle bookkeeping the

src/main.vala

@@ -36,6 +36,7 @@ int main(string[] args) {
         db.add_entity<Saddle>(Saddle.entity_mapping);
         db.add_entity<Session>(Session.entity_mapping);
         db.add_entity<SessionCommand>(SessionCommand.entity_mapping);
+        db.add_entity<SaddleEntryDedup>(SaddleEntryDedup.entity_mapping);
         db.add_entity<ApiToken>(ApiToken.entity_mapping);
         db.add_entity<Rule>(Rule.entity_mapping);
         db.add_entity<RuleProject>(RuleProject.entity_mapping);

test/web_stack_test.py

@@ -2400,7 +2400,7 @@ def bundle_dispatch(ctx):
     check(status == 200, "redelivered entries push accepted (at-least-once)")
     stored = command_rows(raw_session)
     check(len([r for r in stored if r[0] == "sed -n 1p README.md"]) == 1,
-          "redelivered entries store exactly once (entry-id cursor)")
+          "redelivered entries store exactly once (daemon-global dedup)")
 
     # The retired daemon routes 404 (phase 2 deleted /output, /log
     # and /start serving) — and the web app never calls them.
@@ -5822,8 +5822,140 @@ def bundle_goals(ctx):
     check(wait_task_done(auth, probe_id, timeout=60) is not None,
           "the probe task completes after its resume")
 
+
+# =============================================================================
+# 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 bundle_replay(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
+    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)")
+
+    # Drain: nothing queued remains for the shard's shutdown.
+    check(queued_count() == 0, "the replay shard drains its queue")
+
 BUNDLES_ORDER = ("pages", "dispatch", "subtasks", "lineage",
-                 "allocation", "lifecycle", "answers", "push", "goals")
+                 "allocation", "lifecycle", "answers", "push", "goals",
+                 "replay")
 
 
 def run_bundle(name):