Preserve replay across correction chains

jarca_ledger_mcp/application/assets_service.py

@@ -694,16 +694,37 @@ def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str
                 except (TypeError, json.JSONDecodeError):
                     stored_fingerprint = {"authority": existing_run["request_fingerprint"], "request": ""}
                 latest_basis_run = conn.execute(
-                    "SELECT correction_of_run_id FROM asset_snapshot_runs WHERE basis_date=? ORDER BY rowid DESC LIMIT 1",
+                    "SELECT id, correction_of_run_id FROM asset_snapshot_runs WHERE basis_date=? ORDER BY rowid DESC LIMIT 1",
                     (basis,),
                 ).fetchone()
-                authority_changed_after_correction = bool(
+                run_in_correction_chain = False
-                    latest_basis_run
+                seen_run_ids: set[str] = set()
-                    and latest_basis_run["correction_of_run_id"] == existing_run["id"]
+                chain_run = latest_basis_run
+                while chain_run and chain_run["id"] not in seen_run_ids:
+                    if chain_run["id"] == existing_run["id"]:
+                        run_in_correction_chain = True
+                        break
+                    seen_run_ids.add(chain_run["id"])
+                    parent_id = chain_run["correction_of_run_id"]
+                    chain_run = (
+                        conn.execute(
+                            "SELECT id, correction_of_run_id FROM asset_snapshot_runs WHERE id=?",
+                            (parent_id,),
+                        ).fetchone()
+                        if parent_id
+                        else None
+                    )
+                immutable_replay_run = bool(
+                    run_in_correction_chain
+                    and latest_basis_run
+                    and (
+                        latest_basis_run["id"] != existing_run["id"]
+                        or existing_run["correction_of_run_id"]
+                    )
                 )
                 if stored_fingerprint.get("request") == request_fingerprint and (
                     stored_fingerprint.get("authority") == authority_fingerprint
-                    or authority_changed_after_correction
+                    or immutable_replay_run
                 ):
                     replay_rows, replay_meta = _snapshot_rows_for_run(conn, existing_run["id"])
                     replay_items = [_public_evidence(item) for item in replay_meta.get("applied_evidence", [])]

tests/test_issue_12_review63_assets.py

@@ -94,6 +94,37 @@ def test_f002_correction_replay_preserves_full_initial_response(tmp_path):
     assert initial_replay == first
 
 
+def test_f006_initial_replay_survives_multiple_later_corrections(tmp_path):
+    db_path = str(tmp_path / "ledger.sqlite3")
+    service.create_manual_asset(
+        "현금", 100, "2026-06-30", "cash", db_path=db_path
+    )
+    asset_id = service.list_manual_assets(db_path=db_path)["manual_assets"][0]["id"]
+    first = service.close_asset_snapshot(
+        "2026-06-30", idempotency_key="multi-initial", db_path=db_path
+    )
+    assert first["success"] is True
+
+    service.update_manual_asset(asset_id, valuation_amount=110, db_path=db_path)
+    first_correction = service.close_asset_snapshot(
+        "2026-06-30", correction_reason="first correction",
+        idempotency_key="multi-correction-1", db_path=db_path,
+    )
+    assert first_correction["success"] is True
+
+    service.update_manual_asset(asset_id, valuation_amount=120, db_path=db_path)
+    second_correction = service.close_asset_snapshot(
+        "2026-06-30", correction_reason="second correction",
+        idempotency_key="multi-correction-2", db_path=db_path,
+    )
+    assert second_correction["success"] is True
+
+    replay = service.close_asset_snapshot(
+        "2026-06-30", idempotency_key="multi-initial", db_path=db_path
+    )
+    assert replay == first
+
+
 def test_f003_zero_row_conditional_runs_are_timeseries_authority_with_aliases(tmp_path):
     db_path = str(tmp_path / "ledger.sqlite3")
     service.create_ledger_account("미평가", db_path=db_path)