From 31a8a5c3767778f29c04d80024ed13f6fa345ab9 Mon Sep 17 00:00:00 2001 From: Amir Fathi Date: Sat, 5 Sep 2026 13:29:45 +0000 Subject: [PATCH] fix #442: don't flag an in-flight parallel duplicate as crashed ActivityLogGuard.begin() had no way to tell a batch mate still running (execute_parallel dispatches all calls before any of them completes) apart from a row that survived a crash and restart. Track in-flight idem_keys per guard instance: a restart creates a fresh, empty set, so the crash-recovery path still fires exactly as before. Signed-off-by: Amir Fathi --- app/triggers/activity_log.py | 24 ++++++++++++++++++++++++ tests/test_activity_log.py | 25 +++++++++++++++++++++++++ 2 files changed, 49 insertions(+) diff --git a/app/triggers/activity_log.py b/app/triggers/activity_log.py index a7d878a2..9c0ee3c9 100644 --- a/app/triggers/activity_log.py +++ b/app/triggers/activity_log.py @@ -240,6 +240,10 @@ class ActivityLogGuard: def __init__(self, log: ActivityLog): self._log = log + # idem_keys with an INTENT this process recorded and hasn't yet + # complete()'d, so begin() can tell a live batch mate apart from a + # crash-then-restart, which a fresh empty set never carries. + self._in_flight: set[str] = set() def begin( self, @@ -254,6 +258,7 @@ def begin( # Fresh attempt, or deliberate retake after a failed/uncertain # run — record intent BEFORE the side effect. self._log.record_intent(idem_key, action_name, session_id) + self._in_flight.add(idem_key) return GuardDecision(proceed=True, idem_key=idem_key) if row["status"] == STATUS_DONE: @@ -265,6 +270,7 @@ def begin( done_at = 0.0 if time.time() - done_at > DONE_DEDUP_WINDOW_SECONDS: self._log.record_intent(idem_key, action_name, session_id) + self._in_flight.add(idem_key) return GuardDecision(proceed=True, idem_key=idem_key) # This exact side effect just completed — return its stored # output instead of doing it again. @@ -281,6 +287,23 @@ def begin( logger.info(f"[ActivityLog] Skipped duplicate {action_name} ({idem_key})") return GuardDecision(proceed=False, idem_key=idem_key, stored_output=stored) + if idem_key in self._in_flight: + # A batch mate is still executing this exact call, not crashed; + # refuse without touching the row so it can record the outcome. + logger.info( + f"[ActivityLog] {action_name} ({idem_key}) already in " + f"flight in this batch, refusing the duplicate" + ) + return GuardDecision( + proceed=False, + idem_key=idem_key, + note=( + f"{action_name} is already running with these exact " + f"inputs. Wait for it to finish instead of calling it " + f"again." + ), + ) + # Stale INTENT: a previous attempt was interrupted between starting # the side effect and recording its outcome. Surface once; the next # identical attempt (if the LLM/user decides to retry) is allowed. @@ -306,6 +329,7 @@ def complete( status: str, outputs: Optional[Dict[str, Any]], ) -> None: + self._in_flight.discard(idem_key) ledger_status = STATUS_DONE if status == "success" else STATUS_FAILED provider_ref = None diff --git a/tests/test_activity_log.py b/tests/test_activity_log.py index aec4a12f..5cd392f5 100644 --- a/tests/test_activity_log.py +++ b/tests/test_activity_log.py @@ -78,6 +78,31 @@ def test_crash_window_surfaces_uncertainty_once_then_allows_retry(self, tmp_path assert d3.proceed assert log2.get(d1.idem_key)["status"] == "INTENT" + def test_parallel_duplicate_in_same_process_is_not_flagged_as_crashed( + self, tmp_path + ): + log, guard = make_guard(tmp_path) + d1 = guard.begin("send_gmail", INPUTS, "task1") + assert d1.proceed + # A batch mate with identical inputs, dispatched by execute_parallel + # before d1's complete() is reached: not a crash, still running. + d2 = guard.begin("send_gmail", INPUTS, "task1") + assert not d2.proceed + assert d2.stored_output is None + assert "MAY" not in d2.note + assert "already running" in d2.note + assert log.get(d1.idem_key)["status"] == "INTENT" + + guard.complete( + d1.idem_key, "success", {"status": "success", "message_id": "msg-1"} + ) + + # Once complete()'d, a later identical call follows the ordinary + # DONE-dedup path, not the in-flight one. + d3 = guard.begin("send_gmail", INPUTS, "task1") + assert not d3.proceed + assert d3.stored_output["_idempotent_replay"] is True + def test_failed_run_can_be_retried(self, tmp_path): log, guard = make_guard(tmp_path) d1 = guard.begin("send_gmail", INPUTS, "task1")