fix: keep leftover recipe history envelope after a mid-put crash

src/hyperloom/inference_optimizer/tests/test_local_recipe_store.py

@@ -227,6 +227,94 @@ def test_put_recipe_second_call_archives_prior_and_bumps_version(
     assert live["best_throughput"] == 2000.0
 
 
+def test_put_recipe_keeps_existing_history_envelope_when_snapshot_matches(
+    tmp_path: Path,
+) -> None:
+    """A leftover ``history/v{live}`` from a crash between the two renames
+    must not be overwritten by the next successful put."""
+    store = LocalRecipeStore(root=tmp_path)
+    cid = _cid()
+    store.put_recipe(
+        canonical_id=cid,
+        best_throughput=1000.0,
+        provenance={"source": "first", "generator": "ut"},
+    )
+    store.put_recipe(
+        canonical_id=cid,
+        best_throughput=2000.0,
+        provenance={"source": "second", "generator": "ut"},
+    )
+    live = store.get_recipe(canonical_id=cid)
+    assert live is not None and live["version"] == 2
+    archive_path = store._history_version_path(cid, 2)
+    crashed = {
+        "canonical_id": cid,
+        "version": 2,
+        "archived_at": "2026-01-01T00:00:00.000000+00:00",
+        "replaced_by": {"source": "crashed", "generator": "ut"},
+        "snapshot": dict(live),
+    }
+    archive_path.parent.mkdir(parents=True, exist_ok=True)
+    archive_path.write_text(json.dumps(crashed, indent=2, sort_keys=True), encoding="utf-8")
+
+    third = store.put_recipe(
+        canonical_id=cid,
+        best_throughput=3000.0,
+        provenance={"source": "third", "generator": "ut"},
+    )
+    assert third["version"] == 3
+    live = store.get_recipe(canonical_id=cid)
+    assert live is not None
+    assert live["version"] == 3
+    assert live["best_throughput"] == 3000.0
+    leftover = json.loads(archive_path.read_text(encoding="utf-8"))
+    assert leftover["replaced_by"] == {"source": "crashed", "generator": "ut"}
+    assert leftover["archived_at"] == "2026-01-01T00:00:00.000000+00:00"
+    assert leftover["snapshot"]["best_throughput"] == 2000.0
+    v1 = store.get_recipe(canonical_id=cid, version=1)
+    assert v1 is not None and v1["best_throughput"] == 1000.0
+
+
+@pytest.mark.parametrize("corrupt_body", ["{ truncated", "[]"])
+def test_put_recipe_rewrites_unreadable_history_and_still_advances(
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+    corrupt_body: str,
+) -> None:
+    """A leftover history file that cannot be parsed must not block later puts."""
+    store = LocalRecipeStore(root=tmp_path)
+    cid = _cid()
+    store.put_recipe(
+        canonical_id=cid,
+        best_throughput=1000.0,
+        provenance={"source": "first", "generator": "ut"},
+    )
+    store.put_recipe(
+        canonical_id=cid,
+        best_throughput=2000.0,
+        provenance={"source": "second", "generator": "ut"},
+    )
+    archive_path = store._history_version_path(cid, 2)
+    archive_path.parent.mkdir(parents=True, exist_ok=True)
+    archive_path.write_text(corrupt_body, encoding="utf-8")
+
+    with caplog.at_level("WARNING"):
+        third = store.put_recipe(
+            canonical_id=cid,
+            best_throughput=3000.0,
+            provenance={"source": "third", "generator": "ut"},
+        )
+    assert third["version"] == 3
+    live = store.get_recipe(canonical_id=cid)
+    assert live is not None
+    assert live["version"] == 3
+    assert live["best_throughput"] == 3000.0
+    rewritten = json.loads(archive_path.read_text(encoding="utf-8"))
+    assert rewritten["replaced_by"] == {"source": "third", "generator": "ut"}
+    assert rewritten["snapshot"]["best_throughput"] == 2000.0
+    assert any("unreadable history" in rec.message for rec in caplog.records)
+
+
 def test_put_recipe_counts_report_pre_and_post_write_sizes(
     tmp_path: Path,
 ) -> None:

src/hyperloom/orchestrator/knowledge/recipe_kb/local_store.py

@@ -428,6 +428,12 @@ def put_recipe(
     ) -> dict[str, Any]:
         """Atomically upsert a recipe row and archive the prior live version.
 
+        If ``history/v{N}.json`` already holds a snapshot of the current live
+        row, that archive is kept (crash between the two renames) and only
+        live is advanced. The kept envelope records the write that created
+        the archive, which may not have finished; the completing write's
+        provenance is on the live row.
+
         Returns ``{"canonical_id", "version", "created", "prior_counts",
         "counts"}``. The two count maps are the sizes of each list-valued
         knowledge field before and after the write; audit consumers diff them
@@ -448,26 +454,51 @@ def put_recipe(
 
             if not created:
                 # Archive prior live before overwrite; ``replaced_by`` carries
-                # the triggering write's provenance for audit.
+                # the triggering write's provenance for audit. If history for
+                # this live version already holds the same snapshot, a prior
+                # put crashed after that rename — keep the existing envelope
+                # instead of clobbering ``replaced_by``.
                 archive_path = self._history_version_path(
                     canonical_id,
                     prior_version,
                 )
-                archive_payload: dict[str, Any] = {
-                    "canonical_id": canonical_id,
-                    "version": prior_version,
-                    "archived_at": now,
-                    "replaced_by": dict(provenance or {}),
-                    "snapshot": dict(live) if isinstance(live, dict) else {},
-                }
-                atomic_write_json(
-                    archive_path,
-                    archive_payload,
-                    indent=2,
-                    sort_keys=True,
-                    make_parents=True,
-                    fsync=True,
-                )
+                try:
+                    existing_archive = _read_json(archive_path)
+                except LocalRecipeStoreError as exc:
+                    log.warning(
+                        "put_recipe: unreadable history v%s at %s (%s); rewriting it",
+                        prior_version,
+                        archive_path,
+                        exc,
+                    )
+                    existing_archive = None
+                snapshot = dict(live) if isinstance(live, dict) else {}
+                if (
+                    isinstance(existing_archive, dict)
+                    and isinstance(existing_archive.get("snapshot"), dict)
+                    and existing_archive["snapshot"] == snapshot
+                ):
+                    log.debug(
+                        "put_recipe: keeping existing history v%s at %s (crash residue)",
+                        prior_version,
+                        archive_path,
+                    )
+                else:
+                    archive_payload: dict[str, Any] = {
+                        "canonical_id": canonical_id,
+                        "version": prior_version,
+                        "archived_at": now,
+                        "replaced_by": dict(provenance or {}),
+                        "snapshot": snapshot,
+                    }
+                    atomic_write_json(
+                        archive_path,
+                        archive_payload,
+                        indent=2,
+                        sort_keys=True,
+                        make_parents=True,
+                        fsync=True,
+                    )
 
             # Build payload via ``Recipe.from_dict`` so dataclass instances and
             # dicts both round-trip into the same on-disk shape.