intake review fixes: idempotent ledger add, inclusive ms checkpoint, FIFO promotion

- ledger_add_for_post adopts an existing open item carrying post:<id>
  before adding (crash between add and state save no longer duplicates)
- intake window is now inclusive (ts >= cp, init cp = newest+1) with the
  post-id map as the dedupe, so same-millisecond posts can't be lost to a
  crash between them
- newcomers can't jump the queued backlog: loop order is track -> promote
  -> ingest, and ingest queues when a backlog exists even if a slot is free
- documented the kanban --idempotency-key crash-recovery contract

Co-Authored-By: Claude Fable 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01YECpkAwQUwgu7NVy91R8fW

bridge.py

@@ -325,10 +325,17 @@ def kanban(*args: str, timeout: int = 60) -> str:
 
 
 def ledger_add_for_post(post: dict) -> str:
-    """Create a pipeline ledger item for a Pulse post; return its FULL id."""
+    """Create (or adopt) a pipeline ledger item for a Pulse post; return its
+    FULL id. Idempotent: a crash after `shre-items add` but before the state
+    save must not create a second item on retry, so existing open items are
+    searched for this post's marker first."""
+    marker = f"post:{post['id']}"
+    for item in open_items():
+        if marker in (item.get("detail") or ""):
+            return item["id"]
     content = (post.get("content") or "").strip()
     title = " ".join(content.split())[:80] or "(untitled pulse post)"
-    detail = f"{content}\n\n[pulse-intake] post:{post['id']} " \
+    detail = f"{content}\n\n[pulse-intake] {marker} " \
              f"author:{post.get('user_name') or post.get('user_id')}"
     out = shre_items("add", "--kind", "pipeline", "--stage", "queued",
                      "--title", title, "--detail", detail,
@@ -349,6 +356,10 @@ def intake_start_task(state: dict, post_id: str, entry: dict) -> None:
     """Create the kanban task for an intake entry and acknowledge on the post."""
     content = entry.get("content", "")
     title = " ".join(content.split())[:80] or "(pulse task)"
+    # Crash recovery leans on the CLI's documented idempotency contract:
+    # "If a non-archived task with this key exists, its id is returned
+    # instead of creating a duplicate" — so re-calling after a crash between
+    # create and save_state adopts the existing task rather than forking one.
     raw = kanban("create", title,
                  "--body", f"{content}\n\n(from Pulse post {post_id}, "
                            f"ledger {entry['ledger_id'][:8]})",
@@ -375,17 +386,20 @@ def intake_ingest(state: dict) -> None:
                       f"{ensure_channel(state)}/messages?limit=100") or []
     cp = int(state.get("intake_checkpoint") or 0)
     if cp == 0:
-        # First intake run: start from the newest existing post; never ingest
-        # history.
-        state["intake_checkpoint"] = max(
+        # First intake run: start strictly after the newest existing post;
+        # never ingest history. (+1 because the window below is INCLUSIVE.)
+        state["intake_checkpoint"] = 1 + max(
             [int(m.get("created_at") or 0) for m in msgs] or [0])
         save_state(state)
         return
+    # Inclusive window (ts >= cp) so two posts sharing one millisecond can't
+    # lose the second to a crash between them; the intake map (keyed by post
+    # id) is the dedupe, mirroring the outbound seen-set discipline.
     candidates = []
     for m in msgs:
         ts = int(m.get("created_at") or 0)
         content = (m.get("content") or "").strip()
-        if (ts <= cp or m.get("thread_id") or m["id"] in state["intake"]
+        if (ts < cp or m.get("thread_id") or m["id"] in state["intake"]
                 or (m.get("user_name") or "").lower() in INTAKE_IGNORE_USERS
                 or (m.get("user_id") or "").startswith("agent:")
                 or content.startswith("🤖") or "ledger:" in content):
@@ -403,7 +417,8 @@ def intake_ingest(state: dict) -> None:
                                          int(post["created_at"]))
         save_state(state)
         log(f"intake {post['id'][:8]} -> ledger {ledger_id[:8]}")
-        if intake_active_count(state) < INTAKE_MAX_CONCURRENT:
+        backlog = any(e.get("queued") for e in state["intake"].values())
+        if not backlog and intake_active_count(state) < INTAKE_MAX_CONCURRENT:
             intake_start_task(state, post["id"], entry)
         else:
             entry["queued"] = True
@@ -543,7 +558,9 @@ def main() -> int:
         except Exception as e:
             log(f"ERROR poll: {e}")
         if INTAKE_ENABLED:
-            for step in (intake_ingest, intake_track, intake_promote):
+            # track first (frees slots), then promote the queued backlog
+            # (FIFO), then ingest new posts — so newcomers can't jump ahead.
+            for step in (intake_track, intake_promote, intake_ingest):
                 try:
                     step(state)
                 except Exception as e: