From ee5aefd5aca05599912d6a792c215583ed19ed5a Mon Sep 17 00:00:00 2001 From: Charan Rathore <180254320+charan-rathore@users.noreply.github.com> Date: Mon, 28 Sep 2026 23:41:14 +0530 Subject: [PATCH 1/3] Fix async gate for absorbed mid-turn notifications Refine handling of queued commands and removals in langfuse_hook. --- hooks/langfuse_hook.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/hooks/langfuse_hook.py b/hooks/langfuse_hook.py index e1dbe6d..68f24bd 100644 --- a/hooks/langfuse_hook.py +++ b/hooks/langfuse_hook.py @@ -1177,9 +1177,11 @@ def get_undelivered_queued_notification_ids(rows: List[Dict[str, Any]]) -> List[ A queued result already fills the launch entry's final_content, so the pending-agents check goes clean — yet the turn provably continues: the delivery row and Claude's follow-up response are still outstanding. - Queue remove rows carry no notification content and cannot be matched, so - a removed notification keeps the gate closed until the turn ends — the - safe direction (close-time emission is always correct). + A notification absorbed mid-turn appears as a queued_command attachment, + then leaves the queue in a remove row repeating its content. A matched + removal can no longer continue the turn, so it resolves the gate. Removes + without content cannot be matched; keep the gate closed for those older + row shapes until the turn ends. """ queued: List[str] = [] delivered = set() @@ -1189,7 +1191,7 @@ def get_undelivered_queued_notification_ids(rows: List[Dict[str, Any]]) -> List[ tool_use_id = get_tool_use_id_from_task_notification(row) if not tool_use_id: continue - if row.get("type") == "queue-operation": + if row.get("type") == "queue-operation" and row.get("operation") != "remove": queued.append(tool_use_id) else: delivered.add(tool_use_id) From 972174543de28366f3d73db0bc0092ed83385075 Mon Sep 17 00:00:00 2001 From: Charan Rathore <180254320+charan-rathore@users.noreply.github.com> Date: Mon, 28 Sep 2026 23:41:51 +0530 Subject: [PATCH 2/3] Add integration coverage for absorbed notification removal --- .../test_emit_new_turns_from_transcript.py | 63 +++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/tests/integration/test_emit_new_turns_from_transcript.py b/tests/integration/test_emit_new_turns_from_transcript.py index 3ae7f6a..0a5401e 100644 --- a/tests/integration/test_emit_new_turns_from_transcript.py +++ b/tests/integration/test_emit_new_turns_from_transcript.py @@ -250,3 +250,66 @@ def test_only_the_turn_root_span_is_marked_as_root( # Children explicitly opt out of the SDK's app-root auto-marking, so the # events view never lists them as root observations. assert all(o._otel_span.attributes.get("langfuse.internal.is_app_root") is False for o in children) + + +def test_background_bash_turn_ships_at_its_stop_when_notification_is_absorbed_mid_turn( + hook_module, + fake_langfuse, + isolated_hook_state, + tmp_path, +): + """Claude Code 2.1.281 enqueues a background Bash completion, attaches it + to the running turn and removes it with reason absorbed_mid_turn.""" + notification = ( + "\nb0hs0t91w\n" + "toolu_bg1\n" + "/tmp/tasks/b0hs0t91w.output\n" + "completed\n" + "Background command \"curl\" completed (exit code 0)\n" + "" + ) + rows = [ + {"type": "user", "uuid": "user-1", "timestamp": "2026-01-01T00:00:00.000Z", + "sessionId": "s", "message": {"role": "user", "content": "Trigger the request and debug it."}}, + {"type": "assistant", "uuid": "a-1", "timestamp": "2026-01-01T00:00:01.000Z", + "message": {"id": "msg-1", "role": "assistant", "model": "claude-test", + "content": [{"type": "tool_use", "id": "toolu_bg1", "name": "Bash", + "input": {"command": "curl -s https://example.test/", + "run_in_background": True}}]}}, + {"type": "user", "uuid": "tr-1", "timestamp": "2026-01-01T00:00:02.000Z", + "toolUseResult": {"stdout": "", "stderr": "", "interrupted": False, + "backgroundTaskId": "b0hs0t91w"}, + "message": {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_bg1", + "content": "Command running in background with ID: b0hs0t91w."}]}}, + {"type": "assistant", "uuid": "a-2", "timestamp": "2026-01-01T00:00:03.000Z", + "message": {"id": "msg-2", "role": "assistant", "model": "claude-test", + "content": [{"type": "tool_use", "id": "toolu_wait", "name": "mcp__debug__wait", + "input": {}}]}}, + {"type": "queue-operation", "operation": "enqueue", + "timestamp": "2026-01-01T00:00:04.000Z", "content": notification}, + {"type": "user", "uuid": "tr-2", "timestamp": "2026-01-01T00:00:05.000Z", + "message": {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_wait", + "content": "exited"}]}}, + {"type": "attachment", "uuid": "att-1", "timestamp": "2026-01-01T00:00:04.000Z", + "attachment": {"type": "queued_command", "prompt": notification, + "commandMode": "task-notification"}}, + {"type": "queue-operation", "operation": "remove", "reason": "absorbed_mid_turn", + "timestamp": "2026-01-01T00:00:05.500Z", "content": notification}, + {"type": "assistant", "uuid": "a-3", "timestamp": "2026-01-01T00:00:06.000Z", + "message": {"id": "msg-3", "role": "assistant", "model": "claude-test", + "content": [{"type": "text", "text": "Request finished; breakpoint hit."}]}}, + ] + transcript = tmp_path / "transcript.jsonl" + transcript.write_text("\n".join(json.dumps(r) for r in rows) + "\n", encoding="utf-8") + config = hook_module.LangfuseConfig("public", "secret", "https://example.test", "user-1") + + def roots(): + return [o for o in fake_langfuse.observations if o.name == "Conversational Turn"] + + # The only Stop must export the final output. There may be no next prompt + # or usable SessionEnd to close this turn. + hook_module.emit_new_turns_from_transcript(fake_langfuse, config, "session-bg", transcript) + assert len(roots()) == 1 + assert roots()[0].output == {"role": "assistant", "content": "Request finished; breakpoint hit."} + hook_module.emit_new_turns_from_transcript(fake_langfuse, config, "session-bg", transcript) + assert len(roots()) == 1 From da3a6ebae5eb25ed07b8b257da475823d5aec600 Mon Sep 17 00:00:00 2001 From: Charan Rathore <180254320+charan-rathore@users.noreply.github.com> Date: Mon, 28 Sep 2026 23:42:04 +0530 Subject: [PATCH 3/3] Add unit test for absorbed notification gate resolution --- tests/unit/test_emission_gate.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/unit/test_emission_gate.py b/tests/unit/test_emission_gate.py index b42fc99..c5d790f 100644 --- a/tests/unit/test_emission_gate.py +++ b/tests/unit/test_emission_gate.py @@ -28,3 +28,24 @@ def test_rows_without_notifications_pass_the_gate(hook_module): {"type": "queue-operation", "operation": "remove", "content": None}, ] assert hook_module.get_undelivered_queued_notification_ids(rows) == [] + + +def test_mid_turn_absorbed_notification_counts_as_delivered(hook_module): + # Claude Code 2.1.281: the attachment absorbs the notification and a + # content-bearing remove takes it out of the queue. + rows = [ + {"type": "queue-operation", "operation": "enqueue", "content": NOTIFICATION}, + {"type": "attachment", "attachment": {"type": "queued_command", "prompt": NOTIFICATION, + "commandMode": "task-notification"}}, + {"type": "queue-operation", "operation": "remove", "content": NOTIFICATION, + "reason": "absorbed_mid_turn"}, + ] + assert hook_module.get_undelivered_queued_notification_ids(rows) == [] + + +def test_removal_without_content_keeps_the_gate_closed(hook_module): + rows = [ + {"type": "queue-operation", "operation": "enqueue", "content": NOTIFICATION}, + {"type": "queue-operation", "operation": "remove", "content": None}, + ] + assert hook_module.get_undelivered_queued_notification_ids(rows) == ["toolu_X"]