diff --git a/lib/crewai/src/crewai/telemetry/tracing/last_run.py b/lib/crewai/src/crewai/telemetry/tracing/last_run.py index c98181496..f47385180 100644 --- a/lib/crewai/src/crewai/telemetry/tracing/last_run.py +++ b/lib/crewai/src/crewai/telemetry/tracing/last_run.py @@ -59,7 +59,11 @@ def record_last_run( amp_base_url: str | None, ) -> Path | None: """Write the record atomically; the path, or None when recording is off - or the write failed. Never raises — a run is never failed by this.""" + or the write failed. Never raises — a run is never failed by this. + + "Last" means the run that FINISHED last: a record already there for a run + that finished later is kept, so two crews finishing together in one + project leave the newer one whichever writer gets to the file last.""" if not recording_enabled(): return None record: dict[str, Any] = { @@ -82,7 +86,13 @@ def record_last_run( try: with os.fdopen(descriptor, "w", encoding="utf-8") as handle: handle.write(json.dumps(record, indent=2) + "\n") - os.replace(temporary, path) + if _newer_than(record, read_last_run(path.parent.parent)): + os.replace(temporary, path) + else: + logger.debug( + "A run that finished later is already recorded in %s", path + ) + os.unlink(temporary) except OSError: with contextlib.suppress(OSError): os.unlink(temporary) @@ -97,6 +107,20 @@ def record_last_run( return path +def _newer_than(record: dict[str, Any], existing: dict[str, Any] | None) -> bool: + """Is RECORD the later-finished run? A missing or unreadable existing record, + or one without a comparable time, never wins over the run just finished.""" + if not existing or existing.get("execution_id") == record["execution_id"]: + return True + ours = record["finished_at"] or record["recorded_at"] + theirs = existing.get("finished_at") or existing.get("recorded_at") + if not isinstance(theirs, str) or not isinstance(ours, str): + return True + return ( + ours >= theirs + ) # both ISO 8601 in UTC with the same precision: text order is time order + + def read_last_run(directory: Path | None = None) -> dict[str, Any] | None: """The record, or None when there is none or it cannot be read.""" path = last_run_path(directory) diff --git a/lib/crewai/tests/telemetry/test_last_run.py b/lib/crewai/tests/telemetry/test_last_run.py index ec515fec0..645ccde34 100644 --- a/lib/crewai/tests/telemetry/test_last_run.py +++ b/lib/crewai/tests/telemetry/test_last_run.py @@ -82,3 +82,26 @@ def test_a_failed_write_leaves_no_temporary_file(project, monkeypatch): monkeypatch.setattr(last_run.os, "replace", lambda src, dst: (_ for _ in ()).throw(OSError("disk full"))) assert last_run.record_last_run(execution_id="x", tier=None, started_at_ns=None, finished_at_ns=None, amp_base_url=None) is None assert list((project / ".crewai").iterdir()) == [] + + +def test_the_run_that_finished_last_stays_recorded_whichever_writer_comes_last(project): + """Two crews finish together; the OLDER run's writer gets to the file after the newer one did.""" + second = 1_000_000_000 + base = 1_758_240_000 * second + last_run.record_last_run(execution_id="newer", tier="ephemeral", started_at_ns=base, finished_at_ns=base + 30 * second, amp_base_url=None) + kept = last_run.record_last_run(execution_id="older", tier="ephemeral", started_at_ns=base, finished_at_ns=base + 10 * second, amp_base_url=None) + written = last_run.read_last_run(project) + assert kept == project / ".crewai" / "last_run.json" + assert written is not None and written["execution_id"] == "newer" + assert [child.name for child in (project / ".crewai").iterdir()] == ["last_run.json"] # the loser's temporary file is gone + + # The same run recorded again (a refreshed grant) and a run that finished later both replace it. + last_run.record_last_run(execution_id="newer", tier="authenticated", started_at_ns=base, finished_at_ns=base + 30 * second, amp_base_url=None) + assert last_run.read_last_run(project)["tier"] == "authenticated" + last_run.record_last_run(execution_id="newest", tier="ephemeral", started_at_ns=base, finished_at_ns=base + 40 * second, amp_base_url=None) + assert last_run.read_last_run(project)["execution_id"] == "newest" + + # A record without a comparable time never blocks the run just finished. + (project / ".crewai" / "last_run.json").write_text(json.dumps({"execution_id": "legacy"}), encoding="utf-8") + last_run.record_last_run(execution_id="fresh", tier=None, started_at_ns=None, finished_at_ns=None, amp_base_url=None) + assert last_run.read_last_run(project)["execution_id"] == "fresh"