Dedupe streaming tokens on a sequence id

Streaming is at-least-once per activity attempt: a retried node re-runs
and re-publishes its writes, so subscribers can see duplicate tokens.
Tag each token chunk with a monotonic seq and dedupe on it in the client
and test, demonstrating the documented idempotent-consumer pattern
instead of implicitly assuming exactly-once delivery.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: brianstrauch

langgraph_plugin/graph_api/streaming/run_workflow.py

@@ -22,12 +22,19 @@ async def main() -> None:
 
     # Subscribe to all topics on the workflow's stream and demultiplex on topic.
     ws = WorkflowStreamClient.create(client, handle.id)
+    # Streaming is at-least-once per activity attempt, so a retried node may
+    # re-publish tokens. Dedupe on the chunk's seq to consume idempotently.
+    seen_tokens: set[int] = set()
     async for item in ws.subscribe(
         from_offset=0,
         result_type=dict,
         poll_cooldown=timedelta(milliseconds=50),
     ):
         if item.topic == "tokens":
+            seq = item.data["seq"]
+            if seq in seen_tokens:
+                continue  # duplicate from a node retry; already consumed.
+            seen_tokens.add(seq)
             print(item.data["token"], end="", flush=True)
         elif item.topic == "progress":
             if item.data.get("done"):

langgraph_plugin/graph_api/streaming/workflow.py

@@ -36,11 +36,18 @@ async def outline(state: State) -> dict[str, str]:
 
 
 async def write_story(state: State) -> dict[str, str]:
-    """Write the story, emitting each word as a token via the stream writer."""
+    """Write the story, emitting each word as a token via the stream writer.
+
+    Streaming is at-least-once per activity attempt: if this node retries
+    (transient failure, worker restart) it re-runs from scratch and re-publishes
+    its writes, so subscribers may see the same token twice. Each chunk therefore
+    carries a monotonic ``seq`` so consumers can dedupe idempotently. A retry
+    re-emits the same ``seq`` values, letting the client drop the duplicates.
+    """
     writer = get_stream_writer()
     words = f"{state['story']} Once upon a time, there was {state['topic']}.".split()
-    for word in words:
-        writer({"token": word + " "})
+    for seq, word in enumerate(words):
+        writer({"seq": seq, "token": word + " "})
     return {"story": " ".join(words)}
 
 

tests/langgraph_plugin/streaming_test.py

@@ -51,10 +51,14 @@ async def test_streaming_graph_api(client: Client) -> None:
 
         result = await handle.result()
 
-    # Tokens reassemble into the final story.
+    # Tokens reassemble into the final story. Streaming is at-least-once per
+    # activity attempt, so dedupe on seq (keeping first-seen order) before
+    # reassembling, exactly as an idempotent consumer would.
     assert tokens, "expected at least one token"
-    assert all("token" in t for t in tokens)
-    assembled = "".join(t["token"] for t in tokens).strip()
+    assert all("seq" in t and "token" in t for t in tokens)
+    seen: set[int] = set()
+    deduped = [t for t in tokens if not (t["seq"] in seen or seen.add(t["seq"]))]
+    assembled = "".join(t["token"] for t in deduped).strip()
     assert assembled == result
 
     # Workflow-side astream publish: one chunk per node, in order.