fix(fava): serialize source mutations, share HTTP client, validate BQL input

Fava-client hardening cluster (CODE-REVIEW-2026-06 #7, #14, #15, #19
+ libra-#23, libra-#53):

- New FavaClient.transform_source_line does the whole
  read-checksum-modify-write under the global write lock and maps
  Fava 409/412 to ChecksumConflictError. The approve and reject
  endpoints used to do this dance with raw httpx and no lock — two
  concurrent mutations raced each other and every other ledger
  writer (libra-#23). They now route through the new method and
  translate conflicts to HTTP 409.
- update_entry_source / delete_entry raise ChecksumConflictError on
  409/412 instead of leaking raw HTTPStatusError.
- One shared httpx.AsyncClient per FavaClient (12 per-call
  instantiations removed — no more TCP handshake per request);
  closed via libra_stop. Health probes keep their 2s timeout
  per-request.
- Account names/patterns are validated against ^[A-Za-z0-9:_-]+$
  before interpolation into BQL string literals.
- The posting amount regexes are consolidated into module-level
  compiled patterns, all decimal-tolerant — the old integer-only
  SATS pattern silently dropped decimal-SATS postings (Fava's @@->@
  normalisation emits them) from balances.
- add-account no longer verifies its own write with a second
  serialized get_all_accounts round-trip (libra-#53):
  sync_single_account_from_beancount grows an assume_exists path.

New test: concurrent approve+reject must both land (was
lost-update/412 before the lock).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

__init__.py

@@ -30,6 +30,17 @@ def libra_stop():
         except Exception as ex:
             logger.warning(ex)
 
+    # Close the Fava client's shared HTTP connection pool. libra_stop is
+    # synchronous, so schedule the close; if the loop is already gone the
+    # sockets die with the process anyway.
+    from .fava_client import _fava_client
+
+    if _fava_client is not None:
+        try:
+            asyncio.get_event_loop().create_task(_fava_client.aclose())
+        except Exception as ex:
+            logger.warning(f"Could not close Fava HTTP client: {ex}")
+
 
 def libra_start():
     """Initialize Libra extension background tasks"""

account_sync.py

@@ -285,7 +285,11 @@ async def sync_accounts_from_beancount(force_full_sync: bool = False) -> dict:
     return stats
 
 
-async def sync_single_account_from_beancount(account_name: str) -> bool:
+async def sync_single_account_from_beancount(
+    account_name: str,
+    description: Optional[str] = None,
+    assume_exists: bool = False,
+) -> bool:
     """
     Sync a single account from Beancount to Libra DB.
 
@@ -294,6 +298,13 @@ async def sync_single_account_from_beancount(account_name: str) -> bool:
 
     Args:
         account_name: Hierarchical account name (e.g., "Expenses:Food")
+        description: Description for the Libra DB row (only used with
+            assume_exists — otherwise read from Beancount metadata)
+        assume_exists: Skip the Fava existence lookup. Pass when the
+            caller just wrote the Open directive itself — verifying via
+            a second serialized get_all_accounts round-trip doubles the
+            latency of every account create for no information gain
+            (libra-#53).
 
     Returns:
         True if account was created/updated, False if it already existed or failed
@@ -306,6 +317,22 @@ async def sync_single_account_from_beancount(account_name: str) -> bool:
         logger.debug(f"Account already exists: {account_name}")
         return False
 
+    if assume_exists:
+        try:
+            await create_account(
+                CreateAccount(
+                    name=account_name,
+                    account_type=infer_account_type_from_name(account_name),
+                    description=description,
+                    user_id=extract_user_id_from_account_name(account_name),
+                )
+            )
+            logger.info(f"Created account (writer-asserted): {account_name}")
+            return True
+        except Exception as e:
+            logger.error(f"Failed to sync account {account_name}: {e}")
+            return False
+
     # Get from Beancount
     fava = get_fava_client()
     try:

beancount_format.py

@@ -115,13 +115,18 @@ def format_balance(
     account: str,
     amount: int,
     currency: str = "SATS"
-) -> str:
+) -> Dict[str, Any]:
     """
-    Format a balance assertion directive for Beancount.
+    Format a balance assertion directive for Fava's JSON API.
 
     Balance assertions verify that an account has an expected balance on a specific date.
     They are checked automatically by Beancount when the file is loaded.
 
+    Fava's `deserialise` (fava/serialisation.py) expects
+    `{"t": "Balance", "amount": {"number", "currency"}, ...}` — the
+    previous source-string return 500'd on every assertion create
+    (libra-#39).
+
     Args:
         date_val: Date of the balance assertion
         account: Account name (e.g., "Assets:Bitcoin:Lightning")
@@ -129,15 +134,15 @@ def format_balance(
         currency: Currency code (default: "SATS")
 
     Returns:
-        Beancount balance directive as a string
-
-    Example:
-        >>> format_balance(date(2025, 11, 10), "Assets:Bitcoin:Lightning", 1500000, "SATS")
-        '2025-11-10 balance Assets:Bitcoin:Lightning  1500000 SATS'
+        Fava API Balance entry dict ready for `fava.add_entry`.
     """
-    date_str = date_val.strftime('%Y-%m-%d')
-    # Two spaces between account and amount (Beancount convention)
-    return f"{date_str} balance {account}  {amount} {currency}"
+    return {
+        "t": "Balance",
+        "date": date_val.strftime('%Y-%m-%d'),
+        "account": account,
+        "amount": {"number": str(amount), "currency": currency},
+        "meta": {},
+    }
 
 
 def format_posting_with_cost(
@@ -252,9 +257,10 @@ def format_posting_at_average_cost(
         amount_str = f"{amount_sats} SATS {{{cost_currency}}}"
         logger.info(f"format_posting_at_average_cost: Generated amount_str='{amount_str}' with cost_currency='{cost_currency}'")
     else:
-        # No cost
-        amount_str = f"{amount_sats} SATS {{}}"
-        logger.warning(f"format_posting_at_average_cost: cost_currency is None, using empty cost basis")
+        # No cost basis — omit the braces entirely. Empty "{}" is not
+        # valid Beancount syntax and fails to parse on ledger load.
+        amount_str = f"{amount_sats} SATS"
+        logger.warning(f"format_posting_at_average_cost: cost_currency is None, omitting cost basis")
 
     posting_meta = metadata or {}
 
@@ -304,6 +310,25 @@ def format_posting_simple(
     }
 
 
+def fiat_rate_metadata(amount_sats: int, fiat_amount: Decimal) -> Dict[str, str]:
+    """Exchange-rate metadata (sats per fiat unit, fiat per BTC) as exact
+    Decimal strings.
+
+    These values become the cost-basis record for the entry, so they must
+    not carry float drift — CLAUDE.md mandates Decimal for all fiat math.
+
+    Returns:
+        {"fiat_rate": "<sats per fiat unit>", "btc_rate": "<fiat per BTC>"}
+    """
+    if amount_sats <= 0 or fiat_amount <= 0:
+        return {"fiat_rate": "0", "btc_rate": "0"}
+    fiat_rate = (Decimal(amount_sats) / fiat_amount).quantize(Decimal("0.000001"))
+    btc_rate = (
+        fiat_amount / Decimal(amount_sats) * Decimal(100_000_000)
+    ).quantize(Decimal("0.01"))
+    return {"fiat_rate": str(fiat_rate), "btc_rate": str(btc_rate)}
+
+
 def format_expense_entry(
     user_id: str,
     expense_account: str,
@@ -718,15 +743,18 @@ def format_net_settlement_entry(
     entry_date: date,
     payment_hash: Optional[str] = None,
     reference: Optional[str] = None,
-    settled_entry_links: Optional[List[str]] = None
+    settled_entry_links: Optional[List[str]] = None,
+    credit_account: Optional[str] = None,
+    credit_overflow_fiat: Decimal = Decimal(0),
 ) -> Dict[str, Any]:
     """
     Format a net settlement payment entry (user paying net balance).
 
-    Creates a three-posting transaction:
+    Creates a three- to four-posting transaction:
     1. Lightning payment in SATS with @@ total price notation
     2. Clear receivables in EUR
     3. Clear payables in EUR
+    4. Credit overflow when the payment exceeds what it clears
 
     Example:
         Assets:Bitcoin:Lightning        565251 SATS @@ 517.00 EUR
@@ -734,25 +762,61 @@ def format_net_settlement_entry(
         Liabilities:Payable:User          38.00 EUR
         = 517 - 555 + 38 = 0 ✓
 
+    Constraint enforced inline (same contract as
+    `format_fiat_net_settlement_entry`):
+        net_fiat_amount = total_receivable_fiat - total_payable_fiat
+                          + credit_overflow_fiat
+
     Args:
         user_id: User ID
         payment_account: Payment account (e.g., "Assets:Bitcoin:Lightning")
         receivable_account: User's receivable account
         payable_account: User's payable account
         amount_sats: SATS amount paid
-        net_fiat_amount: Net fiat amount (receivable - payable)
-        total_receivable_fiat: Total receivables to clear
-        total_payable_fiat: Total payables to clear
+        net_fiat_amount: Fiat value of the payment being recorded
+        total_receivable_fiat: Receivables cleared by this payment
+        total_payable_fiat: Payables cleared by this payment
         fiat_currency: Currency (EUR, USD)
         description: Payment description
         entry_date: Date of payment
         payment_hash: Lightning payment hash
         reference: Optional reference
         settled_entry_links: List of expense/receivable links being settled (e.g., ["exp-abc123", "rcv-def456"])
+        credit_account: User's credit account receiving overflow (required
+            when credit_overflow_fiat > 0)
+        credit_overflow_fiat: Payment excess beyond what it clears, absorbed
+            as a liability libra owes the user going forward
 
     Returns:
         Fava API entry dict
+
+    Raises:
+        ValueError: if any amount is negative, or the payment doesn't
+            balance against what it clears — an unbalanced settlement
+            must never reach the ledger.
     """
+    for label, value in (
+        ("net_fiat_amount", net_fiat_amount),
+        ("total_receivable_fiat", total_receivable_fiat),
+        ("total_payable_fiat", total_payable_fiat),
+        ("credit_overflow_fiat", credit_overflow_fiat),
+    ):
+        if value < 0:
+            raise ValueError(f"{label} must be non-negative; got {value}")
+
+    expected_payment = (
+        total_receivable_fiat - total_payable_fiat + credit_overflow_fiat
+    )
+    if abs(net_fiat_amount - expected_payment) > Decimal("0.01"):
+        raise ValueError(
+            f"net_fiat_amount {net_fiat_amount} does not match expected "
+            f"{expected_payment} (= receivable {total_receivable_fiat} "
+            f"- payable {total_payable_fiat} + credit {credit_overflow_fiat}); "
+            f"refusing to write an unbalanced settlement"
+        )
+    if credit_overflow_fiat > 0 and not credit_account:
+        raise ValueError("credit_account required when credit_overflow_fiat > 0")
+
     # Build postings for net settlement
     # Note: We use @@ (total price) syntax for cleaner formatting, but Fava's API
     # will convert this to @ (per-unit price) with a long decimal when writing to file.
@@ -761,20 +825,26 @@ def format_net_settlement_entry(
     postings = [
         {
             "account": payment_account,
-            "amount": f"{abs(amount_sats)} SATS @@ {abs(net_fiat_amount):.2f} {fiat_currency}",
+            "amount": f"{abs(amount_sats)} SATS @@ {net_fiat_amount:.2f} {fiat_currency}",
             "meta": {"payment-hash": payment_hash} if payment_hash else {}
         },
         {
             "account": receivable_account,
-            "amount": f"-{abs(total_receivable_fiat):.2f} {fiat_currency}",
+            "amount": f"-{total_receivable_fiat:.2f} {fiat_currency}",
             "meta": {"sats-equivalent": str(abs(amount_sats))}
         },
         {
             "account": payable_account,
-            "amount": f"{abs(total_payable_fiat):.2f} {fiat_currency}",
+            "amount": f"{total_payable_fiat:.2f} {fiat_currency}",
             "meta": {}
         }
     ]
+    if credit_overflow_fiat > 0:
+        postings.append({
+            "account": credit_account,
+            "amount": f"-{credit_overflow_fiat:.2f} {fiat_currency}",
+            "meta": {}
+        })
 
     entry_meta = {
         "user-id": user_id,

core/validation.py

@@ -5,7 +5,7 @@ Comprehensive validation following Beancount's plugin system approach,
 but implemented as simple functions that can be called directly.
 """
 
-from decimal import Decimal
+from decimal import Decimal, InvalidOperation
 from typing import Any, Dict, List, Optional
 
 
@@ -278,11 +278,14 @@ def validate_metadata(
             }
         )
 
-    # Validate fiat amount is valid Decimal
+    # Validate fiat amount is valid Decimal. InvalidOperation is what
+    # Decimal actually raises on garbage input ("abc") — it is not a
+    # ValueError subclass, so without it the raw exception leaked to
+    # callers (libra-#38).
     if has_fiat_amount:
         try:
             Decimal(str(metadata["fiat_amount"]))
-        except (ValueError, TypeError) as e:
+        except (ValueError, TypeError, InvalidOperation) as e:
             raise ValidationError(
                 f"Invalid fiat_amount: {metadata['fiat_amount']}",
                 {"error": str(e)}

crud.py

@@ -1696,3 +1696,73 @@ async def check_user_has_role_permission(
                 return True
 
     return False
+
+
+# =============================================================================
+# PROCESSED PAYMENTS (Lightning payment idempotency gate)
+# =============================================================================
+# The Fava-side duplicate checks are read-then-write races; this table's
+# primary key on payment_hash makes exactly one claimant win. Shared by the
+# background invoice listener (tasks.on_invoice_paid) and the client-driven
+# /record-payment endpoint.
+
+
+async def claim_payment(payment_hash: str) -> bool:
+    """Atomically claim a Lightning payment for recording.
+
+    Returns True when this caller owns the claim; False when the payment
+    is already recorded or another coroutine is recording it right now.
+    """
+    result = await db.execute(
+        """
+        INSERT INTO processed_payments (payment_hash, status)
+        VALUES (:payment_hash, 'processing')
+        ON CONFLICT (payment_hash) DO NOTHING
+        """,
+        {"payment_hash": payment_hash},
+    )
+    return result.rowcount == 1
+
+
+async def get_processed_payment(payment_hash: str) -> Optional[dict]:
+    row = await db.fetchone(
+        "SELECT payment_hash, status, entry_id FROM processed_payments"
+        " WHERE payment_hash = :payment_hash",
+        {"payment_hash": payment_hash},
+    )
+    return dict(row) if row else None
+
+
+async def mark_payment_done(payment_hash: str, entry_id: Optional[str] = None) -> None:
+    await db.execute(
+        """
+        UPDATE processed_payments SET status = 'done', entry_id = :entry_id
+        WHERE payment_hash = :payment_hash
+        """,
+        {"payment_hash": payment_hash, "entry_id": entry_id},
+    )
+
+
+async def release_payment_claim(payment_hash: str) -> None:
+    """Compensating delete after a failed recording, so redelivery retries.
+
+    Only removes an in-flight claim — a 'done' row is permanent.
+    """
+    await db.execute(
+        "DELETE FROM processed_payments"
+        " WHERE payment_hash = :payment_hash AND status = 'processing'",
+        {"payment_hash": payment_hash},
+    )
+
+
+async def clear_stale_payment_claims() -> int:
+    """Drop 'processing' claims left behind by a previous process life.
+
+    A live claim only exists inside a running coroutine, so anything
+    still 'processing' at listener startup belongs to a crashed or
+    restarted process and would otherwise block that payment forever.
+    """
+    result = await db.execute(
+        "DELETE FROM processed_payments WHERE status = 'processing'"
+    )
+    return result.rowcount

fava_client.py

@@ -20,7 +20,8 @@ See: https://github.com/beancount/fava/blob/main/src/fava/json_api.py
 import asyncio
 import re
 import httpx
-from typing import Any, Dict, List, Optional
+from contextlib import asynccontextmanager
+from typing import Any, AsyncIterator, Callable, Dict, List, Optional
 from decimal import Decimal
 from datetime import date, datetime
 from loguru import logger
@@ -44,6 +45,34 @@ def _infer_target_file(account_name: str) -> str:
     return "accounts/chart.beancount"
 
 
+# Posting amount-string patterns shared by the balance parsers. Fava's
+# @@ → @ normalisation can emit decimal SATS values, so every SATS group
+# must tolerate decimals (the old integer-only pattern silently dropped
+# those postings from balances).
+_TOTAL_PRICE_RE = re.compile(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@@\s+(-?[\d.]+)\s+SATS$')
+_UNIT_PRICE_RE = re.compile(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@\s+([\d.]+)\s+SATS$')
+_FIAT_AMOUNT_RE = re.compile(r'^(-?[\d.]+)\s+([A-Z]{3})$')
+_SATS_AMOUNT_RE = re.compile(r'^(-?[\d.]+)\s+SATS')
+
+
+def _sats_to_int(value: str) -> int:
+    """Parse a (possibly decimal) SATS amount string to whole sats."""
+    return int(Decimal(value))
+
+
+# Account names/patterns are interpolated into BQL string literals; restrict
+# them to Beancount account characters so caller-supplied input can't break
+# out of the quoted literal.
+_BQL_ACCOUNT_RE = re.compile(r'^[A-Za-z0-9:_-]+$')
+
+
+def _validate_bql_account(value: str) -> str:
+    """Validate a value bound for interpolation into a BQL string literal."""
+    if not _BQL_ACCOUNT_RE.match(value):
+        raise ValueError(f"Invalid account name for BQL query: {value!r}")
+    return value
+
+
 def _escape_beancount_string(value: str) -> str:
     """Escape a value for safe inclusion in a Beancount string literal.
 
@@ -136,6 +165,27 @@ class FavaClient:
         self._main_dir_cache: Optional[str] = None
         self._main_dir_lock = asyncio.Lock()
 
+        # Shared HTTP client, created lazily on first use. One client
+        # means one connection pool instead of a TCP handshake per call.
+        self._http: Optional[httpx.AsyncClient] = None
+
+    @asynccontextmanager
+    async def _client(self) -> AsyncIterator[httpx.AsyncClient]:
+        """Yield the shared HTTP client (lazily created).
+
+        Kept as a context manager so call sites read the same as the
+        per-call clients they replace; the client itself is NOT closed on
+        exit — call `aclose()` at extension shutdown.
+        """
+        if self._http is None or self._http.is_closed:
+            self._http = httpx.AsyncClient(timeout=self.timeout)
+        yield self._http
+
+    async def aclose(self) -> None:
+        """Close the shared HTTP client (extension shutdown)."""
+        if self._http is not None and not self._http.is_closed:
+            await self._http.aclose()
+
     async def _resolve_target_file(self, target_file: str) -> str:
         """
         Turn a relative include path into the absolute path fava expects.
@@ -160,7 +210,7 @@ class FavaClient:
         if self._main_dir_cache is None:
             async with self._main_dir_lock:
                 if self._main_dir_cache is None:
-                    async with httpx.AsyncClient(timeout=self.timeout) as client:
+                    async with self._client() as client:
                         resp = await client.get(f"{self.base_url}/options")
                         resp.raise_for_status()
                         main_file = resp.json()["data"]["beancount_options"]["filename"]
@@ -236,7 +286,7 @@ class FavaClient:
         # Acquire global write lock to serialize ledger modifications
         async with self._write_lock:
             try:
-                async with httpx.AsyncClient(timeout=self.timeout) as client:
+                async with self._client() as client:
                     response = await client.put(
                         f"{self.base_url}/add_entries",
                         json={"entries": [entry]},
@@ -350,10 +400,11 @@ class FavaClient:
 
         # Use sum(weight) for SATS and sum(number) for fiat
         # Note: BQL doesn't support != operator, so use flag = '*' to exclude pending
+        _validate_bql_account(account_name)
         query = f"SELECT sum(number), sum(weight) WHERE account = '{account_name}' AND flag = '*'"
 
         try:
-            async with httpx.AsyncClient(timeout=self.timeout) as client:
+            async with self._client() as client:
                 response = await client.get(
                     f"{self.base_url}/query",
                     params={"query_string": query}
@@ -446,14 +497,14 @@ class FavaClient:
                 import re
 
                 # Try total price notation: "50.00 EUR @@ 50000 SATS"
-                total_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@@\s+(-?\d+)\s+SATS$', amount_str)
+                total_price_match = _TOTAL_PRICE_RE.match(amount_str)
                 # Try per-unit price notation: "50.00 EUR @ 1000.5 SATS"
-                unit_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@\s+([\d.]+)\s+SATS$', amount_str)
+                unit_price_match = _UNIT_PRICE_RE.match(amount_str)
 
                 if total_price_match:
                     fiat_amount = Decimal(total_price_match.group(1))
                     fiat_currency = total_price_match.group(2)
-                    sats_amount = int(total_price_match.group(3))
+                    sats_amount = _sats_to_int(total_price_match.group(3))
 
                     if fiat_currency not in fiat_balances:
                         fiat_balances[fiat_currency] = Decimal(0)
@@ -480,8 +531,8 @@ class FavaClient:
                     accounts_dict[account_name]["sats"] += sats_amount
 
                 # Try simple fiat format: "50.00 EUR" (check metadata for sats)
-                elif re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str):
-                    fiat_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str)
+                elif _FIAT_AMOUNT_RE.match(amount_str):
+                    fiat_match = _FIAT_AMOUNT_RE.match(amount_str)
                     if fiat_match and fiat_match.group(2) in ('EUR', 'USD', 'GBP'):
                         fiat_amount = Decimal(fiat_match.group(1))
                         fiat_currency = fiat_match.group(2)
@@ -502,9 +553,9 @@ class FavaClient:
 
                 else:
                     # Old format: SATS with cost/price notation - extract SATS amount
-                    sats_match = re.match(r'^(-?\d+)\s+SATS', amount_str)
+                    sats_match = _SATS_AMOUNT_RE.match(amount_str)
                     if sats_match:
-                        sats_amount = int(sats_match.group(1))
+                        sats_amount = _sats_to_int(sats_match.group(1))
                         total_sats += sats_amount
 
                         # Track per account
@@ -603,14 +654,14 @@ class FavaClient:
                 import re
 
                 # Try total price notation: "50.00 EUR @@ 50000 SATS"
-                total_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@@\s+(-?\d+)\s+SATS$', amount_str)
+                total_price_match = _TOTAL_PRICE_RE.match(amount_str)
                 # Try per-unit price notation: "50.00 EUR @ 1000.5 SATS"
-                unit_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@\s+([\d.]+)\s+SATS$', amount_str)
+                unit_price_match = _UNIT_PRICE_RE.match(amount_str)
 
                 if total_price_match:
                     fiat_amount = Decimal(total_price_match.group(1))
                     fiat_currency = total_price_match.group(2)
-                    sats_amount = int(total_price_match.group(3))
+                    sats_amount = _sats_to_int(total_price_match.group(3))
 
                     if fiat_currency not in user_data[user_id]["fiat_balances"]:
                         user_data[user_id]["fiat_balances"][fiat_currency] = Decimal(0)
@@ -629,8 +680,8 @@ class FavaClient:
                     user_data[user_id]["balance"] += sats_amount
 
                 # Try simple fiat format: "50.00 EUR" (check metadata for sats)
-                elif re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str):
-                    fiat_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str)
+                elif _FIAT_AMOUNT_RE.match(amount_str):
+                    fiat_match = _FIAT_AMOUNT_RE.match(amount_str)
                     if fiat_match and fiat_match.group(2) in ('EUR', 'USD', 'GBP'):
                         fiat_amount = Decimal(fiat_match.group(1))
                         fiat_currency = fiat_match.group(2)
@@ -648,9 +699,9 @@ class FavaClient:
 
                 else:
                     # Old format: SATS with cost/price notation
-                    sats_match = re.match(r'^(-?\d+)\s+SATS', amount_str)
+                    sats_match = _SATS_AMOUNT_RE.match(amount_str)
                     if sats_match:
-                        sats_amount = int(sats_match.group(1))
+                        sats_amount = _sats_to_int(sats_match.group(1))
                         user_data[user_id]["balance"] += sats_amount
 
                     # Extract fiat from cost syntax or metadata (backward compatibility)
@@ -683,9 +734,12 @@ class FavaClient:
             True if Fava responds, False otherwise
         """
         try:
-            async with httpx.AsyncClient(timeout=2.0) as client:
+            async with self._client() as client:
+                # Health probes stay fast regardless of the configured
+                # request timeout.
                 response = await client.get(
-                    f"{self.base_url}/changed"
+                    f"{self.base_url}/changed",
+                    timeout=2.0
                 )
                 return response.status_code == 200
         except Exception as e:
@@ -721,12 +775,13 @@ class FavaClient:
         """
         # Build Beancount query
         if account_pattern:
+            _validate_bql_account(account_pattern)
             query = f"SELECT * WHERE account ~ '{account_pattern}' ORDER BY date DESC LIMIT {limit}"
         else:
             query = f"SELECT * ORDER BY date DESC LIMIT {limit}"
 
         try:
-            async with httpx.AsyncClient(timeout=self.timeout) as client:
+            async with self._client() as client:
                 response = await client.get(
                     f"{self.base_url}/query",
                     params={"query_string": query}
@@ -807,7 +862,7 @@ class FavaClient:
             https://beancount.github.io/docs/beancount_query_language.html
         """
         try:
-            async with httpx.AsyncClient(timeout=self.timeout) as client:
+            async with self._client() as client:
                 response = await client.get(
                     f"{self.base_url}/query",
                     params={"query_string": query_string}
@@ -1341,7 +1396,7 @@ class FavaClient:
             # (BQL's SELECT DISTINCT account only returns accounts with postings)
             account_names: set[str] = set()
 
-            async with httpx.AsyncClient(timeout=self.timeout) as client:
+            async with self._client() as client:
                 for endpoint in ("balance_sheet", "income_statement"):
                     try:
                         response = await client.get(f"{self.base_url}/{endpoint}")
@@ -1440,7 +1495,7 @@ class FavaClient:
                 params["time"] = f"{cutoff_date.isoformat()} - {today.isoformat()}"
                 logger.info(f"Querying journal for last {days} days (from {cutoff_date})")
 
-            async with httpx.AsyncClient(timeout=self.timeout) as client:
+            async with self._client() as client:
                 response = await client.get(f"{self.base_url}/journal", params=params)
                 response.raise_for_status()
                 result = response.json()
@@ -1482,7 +1537,7 @@ class FavaClient:
             sha256sum = context["sha256sum"]
         """
         try:
-            async with httpx.AsyncClient(timeout=self.timeout) as client:
+            async with self._client() as client:
                 response = await client.get(
                     f"{self.base_url}/context",
                     params={"entry_hash": entry_hash}
@@ -1529,7 +1584,7 @@ class FavaClient:
         # Acquire global write lock to serialize ledger modifications
         async with self._write_lock:
             try:
-                async with httpx.AsyncClient(timeout=self.timeout) as client:
+                async with self._client() as client:
                     response = await client.put(
                         f"{self.base_url}/source_slice",
                         json={
@@ -1544,11 +1599,78 @@ class FavaClient:
 
             except httpx.HTTPStatusError as e:
                 logger.error(f"Fava update error: {e.response.status_code} - {e.response.text}")
+                if e.response.status_code in (409, 412):
+                    raise ChecksumConflictError(
+                        f"Entry {entry_hash} changed concurrently"
+                    ) from e
                 raise
             except httpx.RequestError as e:
                 logger.error(f"Fava connection error: {e}")
                 raise
 
+    async def transform_source_line(
+        self,
+        filename: str,
+        lineno: int,
+        transform: Callable[[str], str],
+    ) -> bool:
+        """Atomically read-modify-write one line of a ledger source file.
+
+        Holds the global write lock across the whole read-modify-write, so
+        another writer can't slip in between the checksum read and the
+        write (libra-#23: the approve/reject endpoints used to do this
+        read-then-write with raw httpx and no lock).
+
+        The transform receives the current line and returns the new one;
+        returning it unchanged skips the write.
+
+        Returns:
+            True when the line was changed and written, False on a no-op.
+
+        Raises:
+            ValueError: lineno is outside the file.
+            ChecksumConflictError: an out-of-process writer changed the
+                file between read and write (409/412 from Fava).
+        """
+        async with self._write_lock:
+            async with self._client() as client:
+                response = await client.get(
+                    f"{self.base_url}/source",
+                    params={"filename": filename},
+                )
+                response.raise_for_status()
+                source_data = response.json()["data"]
+                sha256sum = source_data["sha256sum"]
+                lines = source_data["source"].split("\n")
+
+                idx = lineno - 1
+                if idx < 0 or idx >= len(lines):
+                    raise ValueError(f"Line {lineno} not found in {filename}")
+
+                new_line = transform(lines[idx])
+                if new_line == lines[idx]:
+                    return False
+                lines[idx] = new_line
+
+                try:
+                    update = await client.put(
+                        f"{self.base_url}/source",
+                        json={
+                            "file_path": filename,
+                            "source": "\n".join(lines),
+                            "sha256sum": sha256sum,
+                        },
+                        headers={"Content-Type": "application/json"},
+                    )
+                    update.raise_for_status()
+                except httpx.HTTPStatusError as e:
+                    if e.response.status_code in (409, 412):
+                        raise ChecksumConflictError(
+                            f"{filename} changed concurrently"
+                        ) from e
+                    raise
+                return True
+
     async def delete_entry(self, entry_hash: str, sha256sum: str) -> str:
         """
         Delete an entry from the Beancount file.
@@ -1571,7 +1693,7 @@ class FavaClient:
         # Acquire global write lock to serialize ledger modifications
         async with self._write_lock:
             try:
-                async with httpx.AsyncClient(timeout=self.timeout) as client:
+                async with self._client() as client:
                     response = await client.delete(
                         f"{self.base_url}/source_slice",
                         params={
@@ -1585,6 +1707,10 @@ class FavaClient:
 
             except httpx.HTTPStatusError as e:
                 logger.error(f"Fava delete error: {e.response.status_code} - {e.response.text}")
+                if e.response.status_code in (409, 412):
+                    raise ChecksumConflictError(
+                        f"Entry {entry_hash} changed concurrently"
+                    ) from e
                 raise
             except httpx.RequestError as e:
                 logger.error(f"Fava connection error: {e}")
@@ -1664,7 +1790,7 @@ class FavaClient:
             # Acquire global write lock to serialize ledger modifications
             async with self._write_lock:
                 try:
-                    async with httpx.AsyncClient(timeout=self.timeout) as client:
+                    async with self._client() as client:
                         # Step 1: Get current source file (fresh read on each attempt)
                         response = await client.get(
                             f"{self.base_url}/source",
@@ -1817,7 +1943,7 @@ class FavaClient:
             # Query 1: Get all original expense/receivable entries for this user
             # These are entries with the expense-entry or receivable-entry tag
             original_query = f"""
-                SELECT date, narration, account, number, weight, links,
+                SELECT date, narration, account, number, currency, weight, links,
                        any_meta('entry-id') as entry_id
                 WHERE account ~ '{account_pattern}'
                   AND '{entry_tag}' IN tags
@@ -1851,7 +1977,10 @@ class FavaClient:
             entries_by_link: Dict[str, Dict[str, Any]] = {}
 
             for row in original_result["rows"]:
-                date_val, narration, account, number, weight, links, entry_id = row
+                (
+                    date_val, narration, account, number, currency,
+                    weight, links, entry_id,
+                ) = row
 
                 # Skip if no links
                 if not links or not isinstance(links, list):
@@ -1875,9 +2004,11 @@ class FavaClient:
                 if entry_link in entries_by_link:
                     continue
 
-                # Parse amounts
-                fiat_amount = abs(float(number)) if number else 0.0
-                fiat_currency = "EUR"  # Default, could be extracted from posting
+                # Parse amounts. The posting's real currency matters: callers
+                # net these totals per currency, and the old hardcoded "EUR"
+                # let USD (or SATS-only) entries be summed as if they were EUR.
+                fiat_amount = str(abs(Decimal(str(number)))) if number else "0"
+                fiat_currency = currency or "EUR"
 
                 # Parse SATS from weight column
                 sats_amount = 0

migrations.py

@@ -34,9 +34,33 @@ Original migration sequence (Nov 2025):
 - m014: Removed legacy equity accounts (MemberEquity, RetainedEarnings)
 - m015: Converted entry_lines to single amount field
 - m016: Dropped journal_entries and entry_lines tables (Fava integration)
+
+IDEMPOTENCY CONTRACT:
+Every statement here must be a silent no-op on re-run. The migration
+version bump lands in the core LNbits DB (`dbversions`) while the DDL
+lands in `ext_libra` — the two writes are not atomic. If the version
+bump fails after the DDL commits, the whole migration re-runs on next
+boot; a bare CREATE/ALTER/INSERT then crashes the extension until
+manual dbversions surgery.
 """
 
 
+async def _alter_add_column_safe(db, sql: str) -> None:
+    """ALTER TABLE ADD COLUMN that swallows duplicate-column errors.
+
+    Neither SQLite nor Postgres supports ADD COLUMN IF NOT EXISTS
+    portably, so re-runs are made no-ops by swallowing the error both
+    backends raise for an existing column.
+    """
+    try:
+        await db.execute(sql)
+    except Exception as exc:
+        msg = str(exc).lower()
+        if "duplicate column" in msg or "already exists" in msg:
+            return
+        raise
+
+
 async def m001_initial(db):
     """
     Initial Libra database schema (squashed from m001-m016).
@@ -63,7 +87,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE accounts (
+        CREATE TABLE IF NOT EXISTS accounts (
             id TEXT PRIMARY KEY,
             name TEXT NOT NULL UNIQUE,
             account_type TEXT NOT NULL,
@@ -76,13 +100,13 @@ async def m001_initial(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_accounts_user_id ON accounts (user_id);
+        CREATE INDEX IF NOT EXISTS idx_accounts_user_id ON accounts (user_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_accounts_type ON accounts (account_type);
+        CREATE INDEX IF NOT EXISTS idx_accounts_type ON accounts (account_type);
         """
     )
 
@@ -93,7 +117,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE extension_settings (
+        CREATE TABLE IF NOT EXISTS extension_settings (
             id TEXT NOT NULL PRIMARY KEY,
             libra_wallet_id TEXT,
             fava_url TEXT NOT NULL DEFAULT 'http://localhost:3333',
@@ -111,7 +135,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE user_wallet_settings (
+        CREATE TABLE IF NOT EXISTS user_wallet_settings (
             id TEXT NOT NULL PRIMARY KEY,
             user_wallet_id TEXT,
             updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
@@ -126,7 +150,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE manual_payment_requests (
+        CREATE TABLE IF NOT EXISTS manual_payment_requests (
             id TEXT PRIMARY KEY,
             user_id TEXT NOT NULL,
             amount INTEGER NOT NULL,
@@ -143,14 +167,14 @@ async def m001_initial(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_manual_payment_requests_user_id
+        CREATE INDEX IF NOT EXISTS idx_manual_payment_requests_user_id
         ON manual_payment_requests (user_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_manual_payment_requests_status
+        CREATE INDEX IF NOT EXISTS idx_manual_payment_requests_status
         ON manual_payment_requests (status);
         """
     )
@@ -163,7 +187,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE balance_assertions (
+        CREATE TABLE IF NOT EXISTS balance_assertions (
             id TEXT PRIMARY KEY,
             date TIMESTAMP NOT NULL,
             account_id TEXT NOT NULL,
@@ -188,21 +212,21 @@ async def m001_initial(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_balance_assertions_account_id
+        CREATE INDEX IF NOT EXISTS idx_balance_assertions_account_id
         ON balance_assertions (account_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_balance_assertions_status
+        CREATE INDEX IF NOT EXISTS idx_balance_assertions_status
         ON balance_assertions (status);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_balance_assertions_date
+        CREATE INDEX IF NOT EXISTS idx_balance_assertions_date
         ON balance_assertions (date);
         """
     )
@@ -216,7 +240,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE user_equity_status (
+        CREATE TABLE IF NOT EXISTS user_equity_status (
             user_id TEXT PRIMARY KEY,
             is_equity_eligible BOOLEAN NOT NULL DEFAULT FALSE,
             equity_account_name TEXT,
@@ -230,7 +254,7 @@ async def m001_initial(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_user_equity_status_eligible
+        CREATE INDEX IF NOT EXISTS idx_user_equity_status_eligible
         ON user_equity_status (is_equity_eligible)
         WHERE is_equity_eligible = TRUE;
         """
@@ -245,7 +269,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE account_permissions (
+        CREATE TABLE IF NOT EXISTS account_permissions (
             id TEXT PRIMARY KEY,
             user_id TEXT NOT NULL,
             account_id TEXT NOT NULL,
@@ -262,7 +286,7 @@ async def m001_initial(db):
     # Index for looking up permissions by user
     await db.execute(
         """
-        CREATE INDEX idx_account_permissions_user_id
+        CREATE INDEX IF NOT EXISTS idx_account_permissions_user_id
         ON account_permissions (user_id);
         """
     )
@@ -270,7 +294,7 @@ async def m001_initial(db):
     # Index for looking up permissions by account
     await db.execute(
         """
-        CREATE INDEX idx_account_permissions_account_id
+        CREATE INDEX IF NOT EXISTS idx_account_permissions_account_id
         ON account_permissions (account_id);
         """
     )
@@ -278,7 +302,7 @@ async def m001_initial(db):
     # Composite index for checking specific user+account permissions
     await db.execute(
         """
-        CREATE INDEX idx_account_permissions_user_account
+        CREATE INDEX IF NOT EXISTS idx_account_permissions_user_account
         ON account_permissions (user_id, account_id);
         """
     )
@@ -286,7 +310,7 @@ async def m001_initial(db):
     # Index for finding permissions by type
     await db.execute(
         """
-        CREATE INDEX idx_account_permissions_type
+        CREATE INDEX IF NOT EXISTS idx_account_permissions_type
         ON account_permissions (permission_type);
         """
     )
@@ -294,7 +318,7 @@ async def m001_initial(db):
     # Index for finding expired permissions
     await db.execute(
         """
-        CREATE INDEX idx_account_permissions_expires
+        CREATE INDEX IF NOT EXISTS idx_account_permissions_expires
         ON account_permissions (expires_at)
         WHERE expires_at IS NOT NULL;
         """
@@ -320,6 +344,7 @@ async def m001_initial(db):
             f"""
             INSERT INTO accounts (id, name, account_type, description, created_at)
             VALUES (:id, :name, :type, :description, {db.timestamp_now})
+            ON CONFLICT (name) DO NOTHING
             """,
             {
                 "id": str(uuid.uuid4()),
@@ -342,17 +367,18 @@ async def m002_add_account_is_active(db):
 
     Default: All existing accounts are marked as active (TRUE).
     """
-    await db.execute(
+    await _alter_add_column_safe(
+        db,
         """
         ALTER TABLE accounts
         ADD COLUMN is_active BOOLEAN NOT NULL DEFAULT TRUE
-        """
+        """,
     )
 
     # Create index for faster queries filtering by is_active
     await db.execute(
         """
-        CREATE INDEX idx_accounts_is_active ON accounts (is_active)
+        CREATE INDEX IF NOT EXISTS idx_accounts_is_active ON accounts (is_active)
         """
     )
 
@@ -372,17 +398,18 @@ async def m003_add_account_is_virtual(db):
 
     Default: All existing accounts are real (is_virtual = FALSE).
     """
-    await db.execute(
+    await _alter_add_column_safe(
+        db,
         """
         ALTER TABLE accounts
         ADD COLUMN is_virtual BOOLEAN NOT NULL DEFAULT FALSE
-        """
+        """,
     )
 
     # Create index for faster queries filtering by is_virtual
     await db.execute(
         """
-        CREATE INDEX idx_accounts_is_virtual ON accounts (is_virtual)
+        CREATE INDEX IF NOT EXISTS idx_accounts_is_virtual ON accounts (is_virtual)
         """
     )
 
@@ -402,6 +429,7 @@ async def m003_add_account_is_virtual(db):
             f"""
             INSERT INTO accounts (id, name, account_type, description, is_active, is_virtual, created_at)
             VALUES (:id, :name, :type, :description, TRUE, TRUE, {db.timestamp_now})
+            ON CONFLICT (name) DO NOTHING
             """,
             {
                 "id": str(uuid.uuid4()),
@@ -438,7 +466,7 @@ async def m004_add_rbac_tables(db):
 
     await db.execute(
         f"""
-        CREATE TABLE roles (
+        CREATE TABLE IF NOT EXISTS roles (
             id TEXT PRIMARY KEY,
             name TEXT NOT NULL UNIQUE,
             description TEXT,
@@ -451,13 +479,13 @@ async def m004_add_rbac_tables(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_roles_name ON roles (name);
+        CREATE INDEX IF NOT EXISTS idx_roles_name ON roles (name);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_roles_is_default ON roles (is_default)
+        CREATE INDEX IF NOT EXISTS idx_roles_is_default ON roles (is_default)
         WHERE is_default = TRUE;
         """
     )
@@ -469,7 +497,7 @@ async def m004_add_rbac_tables(db):
 
     await db.execute(
         f"""
-        CREATE TABLE role_permissions (
+        CREATE TABLE IF NOT EXISTS role_permissions (
             id TEXT PRIMARY KEY,
             role_id TEXT NOT NULL,
             account_id TEXT NOT NULL,
@@ -484,19 +512,19 @@ async def m004_add_rbac_tables(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_role_permissions_role_id ON role_permissions (role_id);
+        CREATE INDEX IF NOT EXISTS idx_role_permissions_role_id ON role_permissions (role_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_role_permissions_account_id ON role_permissions (account_id);
+        CREATE INDEX IF NOT EXISTS idx_role_permissions_account_id ON role_permissions (account_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_role_permissions_type ON role_permissions (permission_type);
+        CREATE INDEX IF NOT EXISTS idx_role_permissions_type ON role_permissions (permission_type);
         """
     )
 
@@ -507,7 +535,7 @@ async def m004_add_rbac_tables(db):
 
     await db.execute(
         f"""
-        CREATE TABLE user_roles (
+        CREATE TABLE IF NOT EXISTS user_roles (
             id TEXT PRIMARY KEY,
             user_id TEXT NOT NULL,
             role_id TEXT NOT NULL,
@@ -522,19 +550,19 @@ async def m004_add_rbac_tables(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_user_roles_user_id ON user_roles (user_id);
+        CREATE INDEX IF NOT EXISTS idx_user_roles_user_id ON user_roles (user_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_user_roles_role_id ON user_roles (role_id);
+        CREATE INDEX IF NOT EXISTS idx_user_roles_role_id ON user_roles (role_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_user_roles_expires ON user_roles (expires_at)
+        CREATE INDEX IF NOT EXISTS idx_user_roles_expires ON user_roles (expires_at)
         WHERE expires_at IS NOT NULL;
         """
     )
@@ -542,7 +570,7 @@ async def m004_add_rbac_tables(db):
     # Composite index for checking specific user+role assignments
     await db.execute(
         """
-        CREATE INDEX idx_user_roles_user_role ON user_roles (user_id, role_id);
+        CREATE INDEX IF NOT EXISTS idx_user_roles_user_role ON user_roles (user_id, role_id);
         """
     )
 
@@ -586,6 +614,7 @@ async def m004_add_rbac_tables(db):
             f"""
             INSERT INTO roles (id, name, description, is_default, created_by, created_at)
             VALUES (:id, :name, :description, :is_default, :created_by, {db.timestamp_now})
+            ON CONFLICT (name) DO NOTHING
             """,
             {
                 "id": str(uuid.uuid4()),
@@ -595,3 +624,30 @@ async def m004_add_rbac_tables(db):
                 "created_by": "system",  # System-created default roles
             },
         )
+
+
+async def m005_add_processed_payments(db):
+    """
+    Local idempotency gate for Lightning payment recording.
+
+    The Fava-side duplicate check (`add_entry_idempotent`, journal-link
+    scan) is a read-then-write race: the background invoice listener and
+    the client-driven /record-payment endpoint can both pass the "not
+    present" check for the same payment_hash and both insert. The
+    primary key on payment_hash makes exactly one claimant win.
+
+    status lifecycle: 'processing' (claimed, write in flight) → 'done'
+    (entry recorded). Failed claims are deleted so redelivery retries;
+    'processing' rows from a crashed process are cleared at listener
+    startup.
+    """
+    await db.execute(
+        f"""
+        CREATE TABLE IF NOT EXISTS processed_payments (
+            payment_hash TEXT PRIMARY KEY,
+            status TEXT NOT NULL DEFAULT 'processing',
+            entry_id TEXT,
+            created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
+        );
+        """
+    )

tasks.py

@@ -179,12 +179,31 @@ async def wait_for_paid_invoices():
     This ensures payments are recorded even if the user closes their browser
     before the payment is detected by client-side polling.
     """
+    from .crud import clear_stale_payment_claims
+
     invoice_queue = Queue()
     register_invoice_listener(invoice_queue, "ext_libra")
 
+    # Claims from a previous process life can't be live anymore — clear
+    # them so those payments aren't blocked forever.
+    cleared = await clear_stale_payment_claims()
+    if cleared:
+        logger.warning(
+            f"[LIBRA] Cleared {cleared} stale in-flight payment claim(s) "
+            "from a previous run"
+        )
+
     while True:
         payment = await invoice_queue.get()
-        await on_invoice_paid(payment)
+        try:
+            await on_invoice_paid(payment)
+        except Exception:
+            # One bad payment must not kill the listener for the rest of
+            # the process lifetime; its claim was released, so redelivery
+            # can retry it.
+            logger.exception(
+                f"[LIBRA] Failed to record payment {payment.payment_hash}"
+            )
 
 
 async def on_invoice_paid(payment: Payment) -> None:
@@ -210,8 +229,20 @@ async def on_invoice_paid(payment: Payment) -> None:
         logger.warning(f"Libra invoice {payment.payment_hash} missing user_id in metadata")
         return
 
+    from .crud import claim_payment, mark_payment_done, release_payment_claim
     from .fava_client import get_fava_client
 
+    # Local idempotency gate: exactly one claimant (this listener or the
+    # /record-payment endpoint) gets to record a given payment_hash. The
+    # Fava-side idempotent write below stays as a second layer for
+    # entries recorded before this table existed.
+    if not await claim_payment(payment.payment_hash):
+        logger.info(
+            f"Payment {payment.payment_hash} already recorded or being "
+            "recorded; skipping"
+        )
+        return
+
     fava = get_fava_client()
 
     # Use idempotency key based on payment hash - this ensures duplicate
@@ -245,26 +276,36 @@ async def on_invoice_paid(payment: Payment) -> None:
 
             if not fiat_currency or not fiat_amount:
                 logger.error(f"Payment {payment.payment_hash} missing fiat currency/amount metadata")
+                await release_payment_claim(payment.payment_hash)
                 return
 
-            # Get user's current balance to determine receivables and payables
+            # Get user's current balance to determine what this payment clears
             balance = await fava.get_user_balance(user_id)
             fiat_balances = balance.get("fiat_balances", {})
             total_fiat_balance = fiat_balances.get(fiat_currency, Decimal(0))
 
-            # Determine receivables and payables based on balance
-            # Positive balance = user owes libra (receivable)
-            # Negative balance = libra owes user (payable)
-            if total_fiat_balance > 0:
-                # User owes libra
-                total_receivable = total_fiat_balance
-                total_payable = Decimal(0)
-            else:
-                # Libra owes user
-                total_receivable = Decimal(0)
-                total_payable = abs(total_fiat_balance)
+            # Settle only what this payment covers. The balance is already
+            # net (positive = user owes libra); a partial payment clears
+            # that much receivable, and any excess — or the whole payment
+            # when nothing is owed — becomes credit libra owes the user.
+            # (Previously partial payments cleared the FULL balance against
+            # a smaller payment, shipping unbalanced postings.)
+            tolerance = Decimal("0.01")
+            open_receivable = (
+                total_fiat_balance if total_fiat_balance > 0 else Decimal(0)
+            )
+            total_receivable = min(open_receivable, fiat_amount)
+            total_payable = Decimal(0)
+            credit_overflow = fiat_amount - total_receivable
+            if credit_overflow < tolerance:
+                # Absorb sub-cent rounding into the receivable leg.
+                credit_overflow = Decimal(0)
+                total_receivable = fiat_amount
 
-            logger.info(f"Settlement: {fiat_amount} {fiat_currency} (Receivable: {total_receivable}, Payable: {total_payable})")
+            logger.info(
+                f"Settlement: {fiat_amount} {fiat_currency} "
+                f"(clears receivable: {total_receivable}, credit: {credit_overflow})"
+            )
 
             # Get account names
             user_receivable = await get_or_create_user_account(
@@ -273,22 +314,35 @@ async def on_invoice_paid(payment: Payment) -> None:
             user_payable = await get_or_create_user_account(
                 user_id, AccountType.LIABILITY, "Accounts Payable"
             )
+            user_credit = None
+            if credit_overflow > 0:
+                user_credit = await get_or_create_user_account(
+                    user_id, AccountType.LIABILITY, "Credit"
+                )
             lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning")
             if not lightning_account:
                 logger.error("Lightning account 'Assets:Bitcoin:Lightning' not found")
+                await release_payment_claim(payment.payment_hash)
                 return
 
-            # Query for unsettled entries to link this settlement back to them
-            # Net settlement can settle both expenses and receivables
+            # Link the source entries this settlement reconciles — but only
+            # when the payment clears the full open balance. On a partial
+            # payment we can't know which entries are covered, and linking
+            # them would make get_unsettled_entries_bql treat them as
+            # settled. Only same-currency entries qualify either way.
             settled_links = []
-            try:
-                unsettled_expenses = await fava.get_unsettled_entries_bql(user_id, "expense")
-                settled_links.extend([e["link"] for e in unsettled_expenses if e.get("link")])
-                unsettled_receivables = await fava.get_unsettled_entries_bql(user_id, "receivable")
-                settled_links.extend([e["link"] for e in unsettled_receivables if e.get("link")])
-            except Exception as e:
-                logger.warning(f"Could not query unsettled entries for settlement links: {e}")
-                # Continue without links - settlement will still be recorded
+            if open_receivable > 0 and fiat_amount + tolerance >= open_receivable:
+                try:
+                    unsettled_expenses = await fava.get_unsettled_entries_bql(user_id, "expense")
+                    unsettled_receivables = await fava.get_unsettled_entries_bql(user_id, "receivable")
+                    settled_links.extend(
+                        e["link"]
+                        for e in unsettled_expenses + unsettled_receivables
+                        if e.get("link") and e.get("fiat_currency") == fiat_currency
+                    )
+                except Exception as e:
+                    logger.warning(f"Could not query unsettled entries for settlement links: {e}")
+                    # Continue without links - settlement will still be recorded
 
             # Format as net settlement transaction
             entry = format_net_settlement_entry(
@@ -305,7 +359,9 @@ async def on_invoice_paid(payment: Payment) -> None:
                 entry_date=datetime.now().date(),
                 payment_hash=payment.payment_hash,
                 reference=payment.payment_hash,
-                settled_entry_links=settled_links if settled_links else None
+                settled_entry_links=settled_links if settled_links else None,
+                credit_account=user_credit.name if user_credit else None,
+                credit_overflow_fiat=credit_overflow,
             )
 
             # Submit to Fava using idempotent method to prevent duplicates
@@ -324,6 +380,11 @@ async def on_invoice_paid(payment: Payment) -> None:
                     f"{result.get('data', 'Unknown')}"
                 )
 
+            await mark_payment_done(payment.payment_hash, idempotency_key)
+
         except Exception as e:
             logger.error(f"Error recording Libra payment {payment.payment_hash}: {e}")
+            # Release the claim so redelivery (or the /record-payment
+            # endpoint) can retry this payment.
+            await release_payment_claim(payment.payment_hash)
             raise

tests/test_migrations.py

@@ -0,0 +1,110 @@
+"""Migration idempotency tests.
+
+The migration version bump lands in the core LNbits DB (`dbversions`)
+while the DDL lands in `ext_libra` — the two writes are not atomic. If
+the bump fails after the DDL commits, the whole migration re-runs on
+next boot, so every statement must be a silent no-op on re-run instead
+of crashing with `duplicate column` / `table exists` / UNIQUE
+violations (which bricks the extension until manual dbversions
+surgery).
+
+These tests run the full migration chain twice against a fresh SQLite
+database — the second pass simulates the lost-version-bump re-run.
+"""
+import importlib
+import re
+from uuid import uuid4
+
+import pytest
+
+from lnbits.db import Database
+
+pytestmark = pytest.mark.anyio
+
+
+def _module(name: str):
+    """Import a libra submodule under whichever path the active LNbits layout
+    uses (default `lnbits.extensions.libra` or bare `libra`)."""
+    for prefix in ("lnbits.extensions.libra", "libra"):
+        try:
+            return importlib.import_module(f"{prefix}.{name}")
+        except ModuleNotFoundError:
+            continue
+    raise ModuleNotFoundError(f"libra.{name}: tried both import paths")
+
+
+migrations = _module("migrations")
+
+# Same discovery as lnbits.core.helpers.run_migration: m### prefix,
+# module definition order.
+_MIGRATION_RE = re.compile(r"^m(\d\d\d)_")
+MIGRATION_FUNCTIONS = [
+    fn for name, fn in vars(migrations).items() if _MIGRATION_RE.match(name)
+]
+
+# Tables the chain must leave behind — one probe row read per table
+# proves both existence and queryability after a double run.
+EXPECTED_TABLES = [
+    "accounts",
+    "extension_settings",
+    "user_wallet_settings",
+    "manual_payment_requests",
+    "balance_assertions",
+    "user_equity_status",
+    "account_permissions",
+    "roles",
+    "role_permissions",
+    "user_roles",
+    "processed_payments",
+]
+
+
+async def _run_all_migrations(db: Database) -> None:
+    async with db.connect() as conn:
+        for migrate in MIGRATION_FUNCTIONS:
+            await migrate(conn)
+
+
+async def _seed_counts(db: Database) -> dict:
+    async with db.connect() as conn:
+        accounts = await conn.fetchall("SELECT id, name FROM accounts")
+        roles = await conn.fetchall("SELECT id, name FROM roles")
+    return {
+        "account_names": sorted(r["name"] for r in accounts),
+        "account_ids": sorted(r["id"] for r in accounts),
+        "role_names": sorted(r["name"] for r in roles),
+        "role_ids": sorted(r["id"] for r in roles),
+    }
+
+
+async def test_migrations_rerun_is_noop():
+    """Full chain twice: second run must not raise and not re-seed."""
+    db = Database(f"ext_libra_migtest_{uuid4().hex[:8]}")
+
+    await _run_all_migrations(db)
+    first = await _seed_counts(db)
+
+    # Simulate the lost dbversions bump: everything runs again.
+    await _run_all_migrations(db)
+    second = await _seed_counts(db)
+
+    # Seeds must not duplicate (names) and must not be replaced (ids).
+    assert second == first
+    assert first["account_names"], "seed accounts missing after migration"
+    assert "Employee" in first["role_names"]
+
+    # Every table exists and is queryable after the double run.
+    async with db.connect() as conn:
+        for table in EXPECTED_TABLES:
+            await conn.fetchall(f"SELECT * FROM {table} LIMIT 1")  # noqa: S608
+
+
+async def test_single_migration_rerun_is_noop():
+    """Each migration individually survives an immediate re-run (the
+    version bump fails right after that one migration committed)."""
+    db = Database(f"ext_libra_migtest_{uuid4().hex[:8]}")
+
+    async with db.connect() as conn:
+        for migrate in MIGRATION_FUNCTIONS:
+            await migrate(conn)
+            await migrate(conn)  # re-run before "bumping" to the next

tests/test_payment_idempotency.py

@@ -0,0 +1,281 @@
+"""Lightning payment idempotency — the `processed_payments` claim gate.
+
+The background invoice listener (`tasks.on_invoice_paid`) and the
+client-driven `POST /record-payment` endpoint can both fire for the
+same `payment_hash` (queue redelivery after restart, webhook + poller).
+The Fava-side duplicate checks are read-then-write races; the local
+`processed_payments` primary key makes exactly one claimant win.
+
+These tests bypass invoice generation (blocked by libra/issues/40) by
+delivering synthetic paid `Payment` objects straight to
+`on_invoice_paid` and by inserting paid payment rows via the LNbits
+core crud for the endpoint tests.
+"""
+import asyncio
+import importlib
+from uuid import uuid4
+
+import pytest
+
+from lnbits.core.crud.payments import create_payment
+from lnbits.core.models.payments import CreatePayment, Payment, PaymentState
+
+from .helpers import list_user_entries, post_receivable
+
+pytestmark = pytest.mark.anyio
+
+
+def _module(name: str):
+    """Import a libra submodule under whichever path the active LNbits layout
+    uses (default `lnbits.extensions.libra` or bare `libra`)."""
+    for prefix in ("lnbits.extensions.libra", "libra"):
+        try:
+            return importlib.import_module(f"{prefix}.{name}")
+        except ModuleNotFoundError:
+            continue
+    raise ModuleNotFoundError(f"libra.{name}: tried both import paths")
+
+
+tasks = _module("tasks")
+libra_crud = _module("crud")
+
+
+def _paid_payment(
+    wallet_id: str,
+    user_id: str,
+    *,
+    fiat_amount: str = "100.00",
+    fiat_currency: str = "EUR",
+    sats: int = 100_000,
+) -> Payment:
+    payment_hash = uuid4().hex + uuid4().hex[:32]
+    return Payment(
+        checking_id=payment_hash,
+        payment_hash=payment_hash,
+        wallet_id=wallet_id,
+        amount=sats * 1000,
+        fee=0,
+        bolt11="lnbcfake",
+        status=PaymentState.SUCCESS,
+        extra={
+            "tag": "libra",
+            "user_id": user_id,
+            "fiat_currency": fiat_currency,
+            "fiat_amount": fiat_amount,
+        },
+    )
+
+
+async def _setup_receivable(
+    client, super_user_headers, configured_user, standard_accounts,
+    amount: str = "100.00",
+):
+    user, wallet = configured_user
+    await post_receivable(
+        client,
+        super_user_headers=super_user_headers,
+        user_id=user.id,
+        amount=amount,
+        description=f"Idempotency setup {uuid4().hex[:6]}",
+        revenue_account=standard_accounts["revenue_rent"]["name"],
+    )
+    # Force a Fava reload before downstream balance reads (see #37).
+    await list_user_entries(client, wallet_inkey=wallet.inkey)
+    return user, wallet
+
+
+async def _entries_with_link(client, wallet_inkey: str, link: str) -> list:
+    payload = await list_user_entries(client, wallet_inkey=wallet_inkey)
+    return [
+        e for e in payload["entries"] if link in (e.get("links") or [])
+    ]
+
+
+# ---------------------------------------------------------------------------
+# on_invoice_paid — the background listener path
+# ---------------------------------------------------------------------------
+
+
+async def test_double_delivery_records_exactly_once(
+    client, super_user_headers, configured_user, standard_accounts
+):
+    """Same payment delivered twice (queue redelivery) → one ledger entry."""
+    user, wallet = await _setup_receivable(
+        client, super_user_headers, configured_user, standard_accounts
+    )
+    payment = _paid_payment(wallet.id, user.id)
+
+    await tasks.on_invoice_paid(payment)
+    await tasks.on_invoice_paid(payment)
+
+    link = f"ln-{payment.payment_hash[:16]}"
+    assert len(await _entries_with_link(client, wallet.inkey, link)) == 1
+
+    row = await libra_crud.get_processed_payment(payment.payment_hash)
+    assert row is not None and row["status"] == "done"
+
+
+async def test_failed_recording_releases_claim_and_retry_succeeds(
+    client, super_user_headers, configured_user, standard_accounts, monkeypatch
+):
+    """A Fava failure mid-write must not permanently block the payment."""
+    user, wallet = await _setup_receivable(
+        client, super_user_headers, configured_user, standard_accounts
+    )
+    payment = _paid_payment(wallet.id, user.id)
+
+    fava_client = _module("fava_client")
+    fava = fava_client.get_fava_client()
+
+    async def _boom(*args, **kwargs):
+        raise RuntimeError("fava down")
+
+    monkeypatch.setattr(fava, "add_entry_idempotent", _boom)
+    with pytest.raises(RuntimeError):
+        await tasks.on_invoice_paid(payment)
+    monkeypatch.undo()
+
+    # Claim released → nothing recorded, retry allowed.
+    assert await libra_crud.get_processed_payment(payment.payment_hash) is None
+
+    await tasks.on_invoice_paid(payment)
+    row = await libra_crud.get_processed_payment(payment.payment_hash)
+    assert row is not None and row["status"] == "done"
+    link = f"ln-{payment.payment_hash[:16]}"
+    assert len(await _entries_with_link(client, wallet.inkey, link)) == 1
+
+
+async def test_listener_survives_poison_payment_and_clears_stale_claims(
+    client, super_user_headers, configured_user, standard_accounts, monkeypatch
+):
+    """One bad payment must not kill the listener; stale 'processing'
+    claims from a previous process life are cleared at startup."""
+    user, wallet = await _setup_receivable(
+        client, super_user_headers, configured_user, standard_accounts
+    )
+
+    # A claim left behind by a "crashed" previous run.
+    stale_hash = uuid4().hex + uuid4().hex[:32]
+    assert await libra_crud.claim_payment(stale_hash)
+
+    captured: dict = {}
+    monkeypatch.setattr(
+        tasks,
+        "register_invoice_listener",
+        lambda queue, name: captured.update(queue=queue),
+    )
+
+    listener = asyncio.create_task(tasks.wait_for_paid_invoices())
+    try:
+        for _ in range(50):
+            if "queue" in captured:
+                break
+            await asyncio.sleep(0.05)
+        assert "queue" in captured, "listener never registered its queue"
+
+        poison = _paid_payment(wallet.id, user.id, fiat_amount="not-a-number")
+        good = _paid_payment(wallet.id, user.id)
+        captured["queue"].put_nowait(poison)
+        captured["queue"].put_nowait(good)
+
+        row = None
+        for _ in range(100):
+            row = await libra_crud.get_processed_payment(good.payment_hash)
+            if row and row["status"] == "done":
+                break
+            await asyncio.sleep(0.1)
+        assert row is not None and row["status"] == "done", (
+            "good payment was not recorded after the poison payment"
+        )
+    finally:
+        listener.cancel()
+
+    # Startup cleared the stale claim; the poison payment's claim was
+    # released on failure so redelivery could retry it.
+    assert await libra_crud.get_processed_payment(stale_hash) is None
+    assert await libra_crud.get_processed_payment(poison.payment_hash) is None
+
+
+# ---------------------------------------------------------------------------
+# POST /record-payment — the client-driven path
+# ---------------------------------------------------------------------------
+
+
+async def _insert_paid_payment_row(wallet_id: str, user_id: str) -> str:
+    payment_hash = uuid4().hex + uuid4().hex[:32]
+    await create_payment(
+        checking_id=payment_hash,
+        data=CreatePayment(
+            wallet_id=wallet_id,
+            payment_hash=payment_hash,
+            bolt11="lnbcfake",
+            amount_msat=100_000_000,
+            memo="idempotency test",
+            extra={
+                "tag": "libra",
+                "user_id": user_id,
+                "fiat_currency": "EUR",
+                "fiat_amount": "100.00",
+            },
+        ),
+        status=PaymentState.SUCCESS,
+    )
+    return payment_hash
+
+
+async def test_record_payment_conflicts_while_in_flight(
+    client, super_user_headers, configured_user, standard_accounts
+):
+    user, wallet = await _setup_receivable(
+        client, super_user_headers, configured_user, standard_accounts
+    )
+    payment_hash = await _insert_paid_payment_row(wallet.id, user.id)
+
+    # Another claimant (e.g. the background listener) is mid-recording.
+    assert await libra_crud.claim_payment(payment_hash)
+
+    r = await client.post(
+        "/libra/api/v1/record-payment",
+        headers={"X-Api-Key": wallet.inkey},
+        json={"payment_hash": payment_hash},
+    )
+    assert r.status_code == 409, r.text
+
+    # Once that claimant finishes, a replay reports "already recorded"
+    # instead of writing a second entry.
+    await libra_crud.mark_payment_done(payment_hash, f"ln-{payment_hash[:16]}")
+    r = await client.post(
+        "/libra/api/v1/record-payment",
+        headers={"X-Api-Key": wallet.inkey},
+        json={"payment_hash": payment_hash},
+    )
+    assert r.status_code == 200, r.text
+    assert "already recorded" in r.json()["message"].lower()
+
+
+async def test_record_payment_records_once_then_replays_safely(
+    client, super_user_headers, configured_user, standard_accounts
+):
+    user, wallet = await _setup_receivable(
+        client, super_user_headers, configured_user, standard_accounts
+    )
+    payment_hash = await _insert_paid_payment_row(wallet.id, user.id)
+
+    r = await client.post(
+        "/libra/api/v1/record-payment",
+        headers={"X-Api-Key": wallet.inkey},
+        json={"payment_hash": payment_hash},
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["message"] == "Payment recorded successfully"
+
+    r = await client.post(
+        "/libra/api/v1/record-payment",
+        headers={"X-Api-Key": wallet.inkey},
+        json={"payment_hash": payment_hash},
+    )
+    assert r.status_code == 200, r.text
+    assert "already recorded" in r.json()["message"].lower()
+
+    link = f"ln-{payment_hash[:16]}"
+    assert len(await _entries_with_link(client, wallet.inkey, link)) == 1

tests/test_reconciliation_api.py

@@ -18,19 +18,6 @@ from uuid import uuid4
 import pytest
 
 
-# Tests that try to actually create + check an assertion all hit issue #39:
-# `format_balance` returns a Beancount source string but `fava.add_entry`
-# expects a dict, so Fava 500s on every assertion-create call. The contract
-# violation is on libra's side; mark these strict-xfail so they go green
-# automatically once #39 lands and the format_balance return shape is fixed.
-ASSERTION_CREATE_BROKEN = pytest.mark.xfail(
-    reason="libra/issues/39 — POST /assertions submits a Beancount source string "
-    "to Fava's JSON API and 500s. Drop this marker when the format_balance "
-    "return type is changed to a dict.",
-    strict=True,
-)
-
-
 # ---------------------------------------------------------------------------
 # helpers (local — assertion endpoints don't have wrapper helpers yet)
 # ---------------------------------------------------------------------------
@@ -58,7 +45,6 @@ async def _create_assertion(
 # ---------------------------------------------------------------------------
 
 
-@ASSERTION_CREATE_BROKEN
 @pytest.mark.anyio
 async def test_assertion_against_empty_account_passes(
     client, super_user_headers, standard_accounts,
@@ -79,7 +65,6 @@ async def test_assertion_against_empty_account_passes(
     assert body.get("difference_sats", 0) == 0
 
 
-@ASSERTION_CREATE_BROKEN
 @pytest.mark.anyio
 async def test_assertion_with_wrong_balance_returns_409(
     client, super_user_headers, standard_accounts,
@@ -102,7 +87,6 @@ async def test_assertion_with_wrong_balance_returns_409(
     assert detail.get("difference_sats") == 999_999 or detail.get("difference_sats") == -999_999
 
 
-@ASSERTION_CREATE_BROKEN
 @pytest.mark.anyio
 async def test_assertion_with_tolerance_accepts_small_diff(
     client, super_user_headers, standard_accounts,
@@ -119,7 +103,6 @@ async def test_assertion_with_tolerance_accepts_small_diff(
     assert r.json().get("status") == "passed"
 
 
-@ASSERTION_CREATE_BROKEN
 @pytest.mark.anyio
 async def test_list_assertions_returns_created(
     client, super_user_headers, standard_accounts,
@@ -145,7 +128,6 @@ async def test_list_assertions_returns_created(
     assert assertion_id in ids, f"created assertion {assertion_id} missing from list {ids}"
 
 
-@ASSERTION_CREATE_BROKEN
 @pytest.mark.anyio
 async def test_get_assertion_by_id(
     client, super_user_headers, standard_accounts,
@@ -167,7 +149,6 @@ async def test_get_assertion_by_id(
     assert r.json().get("id") == assertion_id
 
 
-@ASSERTION_CREATE_BROKEN
 @pytest.mark.anyio
 async def test_recheck_assertion_via_check_endpoint(
     client, super_user_headers, standard_accounts,
@@ -190,7 +171,6 @@ async def test_recheck_assertion_via_check_endpoint(
     assert r.json().get("status") == "passed"
 
 
-@ASSERTION_CREATE_BROKEN
 @pytest.mark.anyio
 async def test_delete_assertion_removes_it(
     client, super_user_headers, standard_accounts,

tests/test_settlement_api.py

@@ -10,6 +10,7 @@ Underpay without explicit entry-picks returns 400 with diff details so
 the operator can either pay the exact net or specify `settled_entry_links`.
 """
 import importlib
+from decimal import Decimal
 from uuid import uuid4
 
 import pytest
@@ -268,10 +269,12 @@ async def test_underpay_without_explicit_links_returns_400(
     assert r.status_code == 400, f"expected 400, got {r.status_code}: {r.text}"
     payload = r.json().get("detail")
     assert isinstance(payload, dict), f"expected structured detail, got {payload!r}"
-    assert payload.get("cash_paid") == 30.0
-    assert payload.get("net_obligation") == 100.0
-    assert payload.get("receivable_total") == 100.0
-    assert payload.get("payable_total") == 0.0
+    # Amounts are exact Decimal strings (not floats) so the operator can
+    # act on them without precision loss.
+    assert Decimal(payload.get("cash_paid")) == Decimal("30.00")
+    assert Decimal(payload.get("net_obligation")) == Decimal("100.00")
+    assert Decimal(payload.get("receivable_total")) == Decimal("100.00")
+    assert Decimal(payload.get("payable_total")) == Decimal("0")
 
 
 @pytest.mark.anyio

tests/test_unit.py

@@ -553,11 +553,6 @@ def test_validate_metadata_fiat_amount_without_currency_raises():
         val.validate_metadata({"fiat_amount": "10.00"})
 
 
-@pytest.mark.xfail(
-    reason="libra/issues/38 — except clause doesn't catch decimal.InvalidOperation, "
-    "so the raw exception leaks instead of becoming ValidationError. Flip when fixed.",
-    strict=True,
-)
 def test_validate_metadata_fiat_amount_invalid_decimal_raises():
     with pytest.raises(val.ValidationError) as exc:
         val.validate_metadata({"fiat_amount": "not-a-number", "fiat_currency": "EUR"})
@@ -570,3 +565,111 @@ def test_validate_metadata_both_present_passes():
 
 def test_validate_metadata_neither_present_passes():
     val.validate_metadata({"source": "api"})
+
+
+# ---------------------------------------------------------------------------
+# format_net_settlement_entry — balance guard + credit overflow
+# ---------------------------------------------------------------------------
+
+
+def _net_settlement(**overrides):
+    kwargs = dict(
+        user_id="abc12345",
+        payment_account="Assets:Bitcoin:Lightning",
+        receivable_account="Assets:Receivable:User-abc12345",
+        payable_account="Liabilities:Payable:User-abc12345",
+        amount_sats=565251,
+        net_fiat_amount=Decimal("517.00"),
+        total_receivable_fiat=Decimal("555.00"),
+        total_payable_fiat=Decimal("38.00"),
+        fiat_currency="EUR",
+        description="test settlement",
+        entry_date=date(2026, 7, 12),
+        payment_hash="ff" * 32,
+    )
+    kwargs.update(overrides)
+    return bf.format_net_settlement_entry(**kwargs)
+
+
+def test_net_settlement_balanced_passes():
+    entry = _net_settlement()
+    amounts = [p["amount"] for p in entry["postings"]]
+    assert any("@@ 517.00 EUR" in a for a in amounts)
+    assert "-555.00 EUR" in amounts
+    assert "38.00 EUR" in amounts
+
+
+def test_net_settlement_unbalanced_partial_payment_raises():
+    # Payment of 300 can't clear a 555 receivable net of 38 payable —
+    # this is the pre-fix partial-payment shape that shipped unbalanced
+    # postings to the ledger.
+    with pytest.raises(ValueError, match="unbalanced"):
+        _net_settlement(net_fiat_amount=Decimal("300.00"))
+
+
+def test_net_settlement_negative_amount_raises():
+    with pytest.raises(ValueError, match="non-negative"):
+        _net_settlement(total_receivable_fiat=Decimal("-1.00"))
+
+
+def test_net_settlement_credit_overflow_adds_leg():
+    entry = _net_settlement(
+        net_fiat_amount=Decimal("600.00"),
+        credit_account="Liabilities:Credit:User-abc12345",
+        credit_overflow_fiat=Decimal("83.00"),
+    )
+    amounts = [p["amount"] for p in entry["postings"]]
+    assert "-83.00 EUR" in amounts
+
+
+def test_net_settlement_credit_overflow_without_account_raises():
+    with pytest.raises(ValueError, match="credit_account"):
+        _net_settlement(
+            net_fiat_amount=Decimal("600.00"),
+            credit_overflow_fiat=Decimal("83.00"),
+        )
+
+
+# ---------------------------------------------------------------------------
+# format_posting_at_average_cost — no empty cost braces
+# ---------------------------------------------------------------------------
+
+
+def test_average_cost_posting_without_currency_omits_braces():
+    posting = bf.format_posting_at_average_cost(
+        account="Assets:Receivable:User-abc", amount_sats=-996896,
+    )
+    assert posting["amount"] == "-996896 SATS"
+    assert "{" not in posting["amount"]
+
+
+def test_average_cost_posting_with_currency_keeps_braces():
+    posting = bf.format_posting_at_average_cost(
+        account="Assets:Receivable:User-abc",
+        amount_sats=-996896,
+        cost_currency="EUR",
+    )
+    assert posting["amount"] == "-996896 SATS {EUR}"
+
+
+# ---------------------------------------------------------------------------
+# fiat_rate_metadata — Decimal-exact rate strings
+# ---------------------------------------------------------------------------
+
+
+def test_fiat_rate_metadata_is_exact():
+    meta = bf.fiat_rate_metadata(107419, Decimal("100.00"))
+    assert meta["fiat_rate"] == "1074.190000"
+    assert meta["btc_rate"] == "93093.40"
+    # No float artifacts like 1074.1899999999998
+    Decimal(meta["fiat_rate"])
+    Decimal(meta["btc_rate"])
+
+
+def test_fiat_rate_metadata_zero_amounts():
+    assert bf.fiat_rate_metadata(0, Decimal("100.00")) == {
+        "fiat_rate": "0", "btc_rate": "0",
+    }
+    assert bf.fiat_rate_metadata(1000, Decimal("0")) == {
+        "fiat_rate": "0", "btc_rate": "0",
+    }

tests/test_void_reject_api.py

@@ -210,3 +210,69 @@ async def test_double_reject_returns_404_on_second_call(
     assert r.status_code in (200, 404), (
         f"second reject should be deterministic, got {r.status_code}: {r.text}"
     )
+
+
+@pytest.mark.anyio
+async def test_concurrent_approve_and_reject_are_serialized(
+    client, super_user_headers, configured_user, standard_accounts,
+):
+    """Two mutations of the same ledger source file fired concurrently must
+    BOTH land. Before libra-#23 each endpoint did its own read-modify-write
+    with raw httpx and no lock, so one writer overwrote the other's change
+    (or 412'd on the stale checksum). Now both route through
+    FavaClient.transform_source_line under the global write lock.
+    """
+    import asyncio
+
+    _, wallet = configured_user
+    approve_tag = f"conc-approve-{uuid4().hex[:6]}"
+    reject_tag = f"conc-reject-{uuid4().hex[:6]}"
+
+    posted = {}
+    for tag in (approve_tag, reject_tag):
+        posted[tag] = await post_expense(
+            client,
+            wallet_inkey=wallet.inkey,
+            user_wallet_id=wallet.id,
+            amount="10.00",
+            currency="EUR",
+            description=tag,
+            expense_account=standard_accounts["expense_food"]["name"],
+        )
+
+    # Force a Fava reload so the approve/reject lookups see both fresh
+    # pending entries (see #37).
+    await list_user_entries(client, wallet_inkey=wallet.inkey)
+
+    r_approve, r_reject = await asyncio.gather(
+        client.post(
+            f"/libra/api/v1/entries/{posted[approve_tag]['id']}/approve",
+            headers=super_user_headers,
+        ),
+        client.post(
+            f"/libra/api/v1/entries/{posted[reject_tag]['id']}/reject",
+            headers=super_user_headers,
+        ),
+    )
+    assert r_approve.status_code == 200, f"approve: {r_approve.text}"
+    assert r_reject.status_code == 200, f"reject: {r_reject.text}"
+
+    # Both mutations must be visible: one entry voided, the other cleared
+    # (a cleared entry no longer matches the pending-only reject lookup).
+    listing = await list_user_entries(client, wallet_inkey=wallet.inkey)
+    entries = listing.get("entries", [])
+    rejected = next(
+        (e for e in entries if reject_tag in (e.get("description") or "")), None,
+    )
+    assert rejected is not None and "voided" in rejected.get("tags", []), (
+        f"rejected entry lost its #voided tag: {rejected}"
+    )
+
+    second_reject = await client.post(
+        f"/libra/api/v1/entries/{posted[approve_tag]['id']}/reject",
+        headers=super_user_headers,
+    )
+    assert second_reject.status_code == 404, (
+        "approved entry should no longer match the pending-only reject "
+        f"lookup, got {second_reject.status_code}"
+    )

views_api.py

@@ -13,6 +13,7 @@ from lnbits.decorators import (
 )
 from lnbits.utils.exchange_rates import allowed_currencies, fiat_amount_as_satoshis
 
+from .beancount_format import fiat_rate_metadata
 from .crud import (
     approve_manual_payment_request,
     check_balance_assertion,
@@ -1074,8 +1075,7 @@ async def api_create_expense_entry(
         metadata = {
             "fiat_currency": data.currency.upper(),
             "fiat_amount": str(data.amount.quantize(Decimal("0.001"))),  # Store as string with 3 decimal places
-            "fiat_rate": float(amount_sats) / float(data.amount) if data.amount > 0 else 0,
-            "btc_rate": float(data.amount) / float(amount_sats) * 100_000_000 if amount_sats > 0 else 0,
+            **fiat_rate_metadata(amount_sats, data.amount),
         }
 
     # Get or create expense account
@@ -1272,8 +1272,7 @@ async def api_create_income_entry(
     metadata = {
         "fiat_currency": fiat_currency,
         "fiat_amount": str(data.amount.quantize(Decimal("0.001"))),
-        "fiat_rate": float(amount_sats) / float(data.amount) if data.amount > 0 else 0,
-        "btc_rate": float(data.amount) / float(amount_sats) * 100_000_000 if amount_sats > 0 else 0,
+        **fiat_rate_metadata(amount_sats, data.amount),
     }
 
     # Submit to Fava
@@ -1371,8 +1370,7 @@ async def api_create_receivable_entry(
         metadata = {
             "fiat_currency": data.currency.upper(),
             "fiat_amount": str(data.amount.quantize(Decimal("0.001"))),  # Store as string with 3 decimal places
-            "fiat_rate": float(amount_sats) / float(data.amount) if data.amount > 0 else 0,
-            "btc_rate": float(data.amount) / float(amount_sats) * 100_000_000 if amount_sats > 0 else 0,
+            **fiat_rate_metadata(amount_sats, data.amount),
         }
 
     # Get or create revenue account
@@ -1760,15 +1758,10 @@ async def api_generate_payment_invoice(
                 proportion = Decimal(data.amount) / Decimal(total_sat_balance)
                 invoice_fiat_amount = abs(total_fiat_balance) * proportion
 
-                # Calculate fiat rate (sats per fiat unit)
-                fiat_rate = float(data.amount) / float(invoice_fiat_amount) if invoice_fiat_amount > 0 else 0
-                btc_rate = float(invoice_fiat_amount) / float(data.amount) * 100_000_000 if data.amount > 0 else 0
-
                 invoice_extra.update({
                     "fiat_currency": fiat_currency,
                     "fiat_amount": str(invoice_fiat_amount.quantize(Decimal("0.001"))),
-                    "fiat_rate": fiat_rate,
-                    "btc_rate": btc_rate,
+                    **fiat_rate_metadata(data.amount, invoice_fiat_amount),
                 })
 
     logger.info(f"Invoice extra metadata: {invoice_extra}")
@@ -1850,90 +1843,130 @@ async def api_record_payment(
 
     try:
         async with httpx.AsyncClient(timeout=5.0) as client:
-            # Get recent entries from Fava's journal endpoint
+            # Get recent entries from Fava's journal endpoint. base_url
+            # already ends in /api — the previous "/api/journal" path
+            # 404'd, so this duplicate check silently never ran.
             response = await client.get(
-                f"{fava.base_url}/api/journal",
+                f"{fava.base_url}/journal",
                 params={"time": ""}  # Get all entries
             )
+            response.raise_for_status()
+            response_data = response.json()
+            entries = response_data.get('entries', [])
 
-            if response.status_code == 200:
-                response_data = response.json()
-                entries = response_data.get('entries', [])
-
-                # Check if any entry has our payment link
-                for entry in entries:
-                    entry_links = entry.get('links', [])
-                    if link_to_find in entry_links:
-                        # Payment already recorded, return existing entry
-                        balance_data = await fava.get_user_balance_bql(target_user_id)
-                        return {
-                            "journal_entry_id": f"fava-exists-{data.payment_hash[:16]}",
-                            "new_balance": balance_data["balance"],
-                            "message": "Payment already recorded",
-                        }
-    except Exception as e:
+            # Check if any entry has our payment link
+            for entry in entries:
+                entry_links = entry.get('links', [])
+                if link_to_find in entry_links:
+                    # Payment already recorded, return existing entry
+                    balance_data = await fava.get_user_balance_bql(target_user_id)
+                    return {
+                        "journal_entry_id": f"fava-exists-{data.payment_hash[:16]}",
+                        "new_balance": balance_data["balance"],
+                        "message": "Payment already recorded",
+                    }
+    except httpx.HTTPError as e:
+        # Fail CLOSED: if Fava can't confirm the payment isn't already
+        # recorded, refuse to write — proceeding on a transient blip is
+        # how double entries happen. The client can simply retry.
         logger.warning(f"Could not check Fava for duplicate payment: {e}")
-        # Continue anyway - Fava/Beancount will catch duplicate if it exists
-
-    # Convert amount from millisatoshis to satoshis
-    amount_sats = payment.amount // 1000
-
-    # Extract fiat metadata from invoice (if present)
-    fiat_currency = None
-    fiat_amount = None
-    if payment.extra and isinstance(payment.extra, dict):
-        logger.info(f"Payment.extra contents: {payment.extra}")
-        fiat_currency = payment.extra.get("fiat_currency")
-        fiat_amount_str = payment.extra.get("fiat_amount")
-        if fiat_amount_str:
-            from decimal import Decimal
-            fiat_amount = Decimal(str(fiat_amount_str))
-
-    logger.info(f"Extracted fiat metadata - currency: {fiat_currency}, amount: {fiat_amount}")
-
-    # Get user's receivable account (what user owes)
-    user_receivable = await get_or_create_user_account(
-        target_user_id, AccountType.ASSET, "Accounts Receivable"
-    )
-
-    # Get lightning account
-    lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning")
-    if not lightning_account:
         raise HTTPException(
-            status_code=HTTPStatus.NOT_FOUND, detail="Lightning account not found"
+            status_code=HTTPStatus.SERVICE_UNAVAILABLE,
+            detail="Cannot verify payment duplicate status; try again shortly",
         )
 
-    # Get unsettled receivable entries to link to this settlement
-    unsettled = await fava.get_unsettled_entries_bql(target_user_id, "receivable")
-    settled_links = [e["link"] for e in unsettled if e.get("link")]
-
-    # Format payment entry and submit to Fava
-    entry = format_payment_entry(
-        user_id=target_user_id,
-        payment_account=lightning_account.name,
-        payable_or_receivable_account=user_receivable.name,
-        amount_sats=amount_sats,
-        description=f"Lightning payment from user {target_user_id[:8]}",
-        entry_date=datetime.now().date(),
-        is_payable=False,  # User paying libra (receivable settlement)
-        fiat_currency=fiat_currency,
-        fiat_amount=fiat_amount,
-        payment_hash=data.payment_hash,
-        reference=data.payment_hash,
-        settled_entry_links=settled_links
+    # Local idempotency gate shared with the background invoice listener
+    # (tasks.on_invoice_paid): exactly one claimant records a payment_hash.
+    from .crud import (
+        claim_payment,
+        get_processed_payment,
+        mark_payment_done,
+        release_payment_claim,
     )
 
-    logger.info(f"Formatted payment entry: {entry}")
+    if not await claim_payment(data.payment_hash):
+        existing = await get_processed_payment(data.payment_hash)
+        if existing and existing["status"] == "done":
+            balance_data = await fava.get_user_balance_bql(target_user_id)
+            return {
+                "journal_entry_id": existing.get("entry_id")
+                or f"fava-exists-{data.payment_hash[:16]}",
+                "new_balance": balance_data["balance"],
+                "message": "Payment already recorded",
+            }
+        raise HTTPException(
+            status_code=HTTPStatus.CONFLICT,
+            detail="Payment is being recorded; check balance shortly",
+        )
 
-    # Submit to Fava
-    result = await fava.add_entry(entry)
-    logger.info(f"Payment entry submitted to Fava: {result.get('data', 'Unknown')}")
+    # Convert amount from millisatoshis to satoshis
+    try:
+        amount_sats = payment.amount // 1000
+
+        # Extract fiat metadata from invoice (if present)
+        fiat_currency = None
+        fiat_amount = None
+        if payment.extra and isinstance(payment.extra, dict):
+            logger.info(f"Payment.extra contents: {payment.extra}")
+            fiat_currency = payment.extra.get("fiat_currency")
+            fiat_amount_str = payment.extra.get("fiat_amount")
+            if fiat_amount_str:
+                from decimal import Decimal
+                fiat_amount = Decimal(str(fiat_amount_str))
+
+        logger.info(f"Extracted fiat metadata - currency: {fiat_currency}, amount: {fiat_amount}")
+
+        # Get user's receivable account (what user owes)
+        user_receivable = await get_or_create_user_account(
+            target_user_id, AccountType.ASSET, "Accounts Receivable"
+        )
+
+        # Get lightning account
+        lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning")
+        if not lightning_account:
+            raise HTTPException(
+                status_code=HTTPStatus.NOT_FOUND, detail="Lightning account not found"
+            )
+
+        # Get unsettled receivable entries to link to this settlement
+        unsettled = await fava.get_unsettled_entries_bql(target_user_id, "receivable")
+        settled_links = [e["link"] for e in unsettled if e.get("link")]
+
+        # Format payment entry and submit to Fava
+        entry = format_payment_entry(
+            user_id=target_user_id,
+            payment_account=lightning_account.name,
+            payable_or_receivable_account=user_receivable.name,
+            amount_sats=amount_sats,
+            description=f"Lightning payment from user {target_user_id[:8]}",
+            entry_date=datetime.now().date(),
+            is_payable=False,  # User paying libra (receivable settlement)
+            fiat_currency=fiat_currency,
+            fiat_amount=fiat_amount,
+            payment_hash=data.payment_hash,
+            reference=data.payment_hash,
+            settled_entry_links=settled_links
+        )
+
+        logger.info(f"Formatted payment entry: {entry}")
+
+        # Submit to Fava
+        result = await fava.add_entry(entry)
+        logger.info(f"Payment entry submitted to Fava: {result.get('data', 'Unknown')}")
+
+        entry_id = f"ln-{data.payment_hash[:16]}"
+        await mark_payment_done(data.payment_hash, entry_id)
+    except BaseException:
+        # Release the claim so a retry (client or background listener)
+        # can record this payment.
+        await release_payment_claim(data.payment_hash)
+        raise
 
     # Get updated balance from Fava
     balance_data = await fava.get_user_balance_bql(target_user_id)
 
     return {
-        "journal_entry_id": f"fava-{datetime.now().timestamp()}",
+        "journal_entry_id": entry_id,
         "new_balance": balance_data["balance"],
         "message": "Payment recorded successfully",
     }
@@ -2024,6 +2057,19 @@ async def api_settle_receivable(
         unsettled_payables = await fava.get_unsettled_entries_bql(data.user_id, "expense")
         unsettled_receivables = await fava.get_unsettled_entries_bql(data.user_id, "receivable")
 
+        # Net only entries denominated in the settlement currency — summing
+        # mixed currencies as if they were one silently mis-states the net
+        # obligation and links entries this settlement doesn't actually clear.
+        settle_currency = data.currency.upper()
+        unsettled_payables = [
+            e for e in unsettled_payables
+            if e.get("fiat_currency") == settle_currency
+        ]
+        unsettled_receivables = [
+            e for e in unsettled_receivables
+            if e.get("fiat_currency") == settle_currency
+        ]
+
         payable_total = sum(
             (Decimal(str(e["fiat_amount"])) for e in unsettled_payables),
             Decimal(0),
@@ -2067,10 +2113,10 @@ async def api_settle_receivable(
                         "net to clear all open entries, or pass "
                         "`settled_entry_links` to settle a specific subset."
                     ),
-                    "cash_paid": float(cash_paid),
-                    "net_obligation": float(net_obligation),
-                    "receivable_total": float(receivable_total),
-                    "payable_total": float(payable_total),
+                    "cash_paid": str(cash_paid),
+                    "net_obligation": str(net_obligation),
+                    "receivable_total": str(receivable_total),
+                    "payable_total": str(payable_total),
                     "currency": data.currency.upper(),
                 },
             )
@@ -2857,8 +2903,7 @@ async def api_approve_expense_entry(
 
     This updates the transaction in the Beancount file via Fava API.
     """
-    import httpx
-    from .fava_client import get_fava_client
+    from .fava_client import ChecksumConflictError, get_fava_client
 
     fava = get_fava_client()
 
@@ -2893,57 +2938,29 @@ async def api_approve_expense_entry(
             detail="Entry metadata missing filename or lineno"
         )
 
-    # 3. Get the source file from Fava
-    async with httpx.AsyncClient(timeout=fava.timeout) as client:
-        response = await client.get(
-            f"{fava.base_url}/source",
-            params={"filename": filename}
-        )
-        response.raise_for_status()
-        source_data = response.json()["data"]
+    # 3. Flip the flag under FavaClient's write lock — the whole
+    # read-modify-write is atomic against every other ledger writer.
+    old_pattern = f"{date_str} !"
 
-        sha256sum = source_data["sha256sum"]
-        source = source_data["source"]
-        lines = source.split('\n')
-
-        # 4. Find and modify the entry at the specified line
-        # Line numbers are 1-indexed, list is 0-indexed
-        entry_line_idx = lineno - 1
-
-        if entry_line_idx >= len(lines):
+    def _approve(line: str) -> str:
+        if old_pattern not in line:
             raise HTTPException(
                 status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
-                detail=f"Line {lineno} not found in source file"
+                detail=f"Line {lineno} does not contain expected pattern '{old_pattern}'. Found: {line}"
             )
+        return line.replace(old_pattern, f"{date_str} *", 1)
 
-        entry_line = lines[entry_line_idx]
-
-        # Check if the line contains the pending flag pattern
-        old_pattern = f"{date_str} !"
-        if old_pattern not in entry_line:
-            raise HTTPException(
-                status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
-                detail=f"Line {lineno} does not contain expected pattern '{old_pattern}'. Found: {entry_line}"
-            )
-
-        # Replace the flag
-        new_pattern = f"{date_str} *"
-        new_line = entry_line.replace(old_pattern, new_pattern, 1)
-        lines[entry_line_idx] = new_line
-
-        # 5. Write back the modified source
-        new_source = '\n'.join(lines)
-
-        update_response = await client.put(
-            f"{fava.base_url}/source",
-            json={
-                "file_path": filename,
-                "source": new_source,
-                "sha256sum": sha256sum
-            },
-            headers={"Content-Type": "application/json"}
+    try:
+        await fava.transform_source_line(filename, lineno, _approve)
+    except ValueError as e:
+        raise HTTPException(
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e)
+        )
+    except ChecksumConflictError:
+        raise HTTPException(
+            status_code=HTTPStatus.CONFLICT,
+            detail="Ledger changed concurrently; retry the approval",
         )
-        update_response.raise_for_status()
 
     logger.info(f"Entry {entry_id} approved (flag changed to *)")
 
@@ -2966,8 +2983,7 @@ async def api_reject_expense_entry(
     Adds #voided tag for audit trail while keeping the '!' flag.
     Voided transactions are excluded from balances but preserved in the ledger.
     """
-    import httpx
-    from .fava_client import get_fava_client
+    from .fava_client import ChecksumConflictError, get_fava_client
 
     fava = get_fava_client()
 
@@ -3002,50 +3018,26 @@ async def api_reject_expense_entry(
             detail="Entry metadata missing filename or lineno"
         )
 
-    # 3. Get the source file from Fava
-    async with httpx.AsyncClient(timeout=fava.timeout) as client:
-        response = await client.get(
-            f"{fava.base_url}/source",
-            params={"filename": filename}
+    # 3. Add the #voided tag under FavaClient's write lock — the whole
+    # read-modify-write is atomic against every other ledger writer.
+    def _void(line: str) -> str:
+        if "#voided" in line:
+            return line  # already voided — no-op
+        return line.rstrip() + ' #voided'
+
+    try:
+        changed = await fava.transform_source_line(filename, lineno, _void)
+    except ValueError as e:
+        raise HTTPException(
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e)
         )
-        response.raise_for_status()
-        source_data = response.json()["data"]
-
-        sha256sum = source_data["sha256sum"]
-        source = source_data["source"]
-        lines = source.split('\n')
-
-        # 4. Find and modify the entry at the specified line - add #voided tag
-        entry_line_idx = lineno - 1
-
-        if entry_line_idx >= len(lines):
-            raise HTTPException(
-                status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
-                detail=f"Line {lineno} not found in source file"
-            )
-
-        entry_line = lines[entry_line_idx]
-
-        # Add #voided tag if not already present
-        if "#voided" not in entry_line:
-            # Add #voided tag to the transaction line
-            new_line = entry_line.rstrip() + ' #voided'
-            lines[entry_line_idx] = new_line
-
-            # 5. Write back the modified source
-            new_source = '\n'.join(lines)
-
-            update_response = await client.put(
-                f"{fava.base_url}/source",
-                json={
-                    "file_path": filename,
-                    "source": new_source,
-                    "sha256sum": sha256sum
-                },
-                headers={"Content-Type": "application/json"}
-            )
-            update_response.raise_for_status()
-            logger.info(f"Entry {entry_id} rejected (added #voided tag)")
+    except ChecksumConflictError:
+        raise HTTPException(
+            status_code=HTTPStatus.CONFLICT,
+            detail="Ledger changed concurrently; retry the rejection",
+        )
+    if changed:
+        logger.info(f"Entry {entry_id} rejected (added #voided tag)")
 
     return {
         "message": f"Entry {entry_id} rejected (marked as voided)",
@@ -3774,8 +3766,14 @@ async def api_admin_add_chart_account(
             "already_existed": True,
         }
 
-    # Mirror into libra DB so permissions / metadata layer sees it.
-    synced = await sync_single_account_from_beancount(payload.name)
+    # Mirror into libra DB so permissions / metadata layer sees it. We just
+    # wrote the Open directive ourselves, so skip the verification
+    # round-trip through Fava (libra-#53).
+    synced = await sync_single_account_from_beancount(
+        payload.name,
+        description=payload.description,
+        assume_exists=True,
+    )
 
     return {
         "success": True,