From 827b816eebcb7895e2ddc595a9da4d8c2dbbb158 Mon Sep 17 00:00:00 2001 From: Nirav Patel Date: Sat, 22 Aug 2026 15:03:19 -0400 Subject: [PATCH] intake review fixes: idempotent ledger add, inclusive ms checkpoint, FIFO promotion - ledger_add_for_post adopts an existing open item carrying post: 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 Claude-Session: https://claude.ai/code/session_01YECpkAwQUwgu7NVy91R8fW --- bridge.py | 33 +++++++++++++++++++++++++-------- 1 file changed, 25 insertions(+), 8 deletions(-) diff --git a/bridge.py b/bridge.py index d6a40e2..6551636 100644 --- a/bridge.py +++ b/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: