Merge pull request #138 from valueonag/int

Int

env_dev.env

@@ -82,6 +82,8 @@ TEAMSBOT_BROWSER_BOT_URL = https://cae-poweron-shared.redwater-53d21339.switzerl
 # Debug Configuration
 APP_DEBUG_CHAT_WORKFLOW_ENABLED = True
 APP_DEBUG_CHAT_WORKFLOW_DIR = D:/Athi/Local/Web/poweron/local/debug
+APP_DEBUG_ACCOUNTING_SYNC_ENABLED = True
+APP_DEBUG_ACCOUNTING_SYNC_DIR = D:/Athi/Local/Web/poweron/local/debug/sync
 
 # Manadate Pre-Processing Servers
 PREPROCESS_ALTHAUS_CHAT_SECRET = DEV_ENC:Z0FBQUFBQnBudkpGbEphQ3ZUMlFMQ2EwSGpoSE9NNzRJNTJtaGk1N0RGakdIYnVVeVFHZmF5OXB3QTVWLVNaZk9wNkhfQkZWRnVwRGRxem9iRzJIWXdpX1NIN2FwSExfT3c9PQ==

env_int.env

@@ -80,6 +80,8 @@ TEAMSBOT_BROWSER_BOT_URL = https://cae-poweron-shared.redwater-53d21339.switzerl
 # Debug Configuration
 APP_DEBUG_CHAT_WORKFLOW_ENABLED = FALSE
 APP_DEBUG_CHAT_WORKFLOW_DIR = ./test-chat
+APP_DEBUG_ACCOUNTING_SYNC_ENABLED = FALSE
+APP_DEBUG_ACCOUNTING_SYNC_DIR = ./debug/sync
 
 # Manadate Pre-Processing Servers
 PREPROCESS_ALTHAUS_CHAT_SECRET = INT_ENC:Z0FBQUFBQnBaSnM4UkNBelhvckxCQUVjZm94N3BZUDcxaEMyckE2dm1lRVhqODhrWU1SUjNXZ3dQZlVJOWhveXFkZXpobW5xT0NneGZ2SkNUblFmYXd0WTBYNTl3UmRnSWc9PQ==

env_prod.env

@@ -81,6 +81,8 @@ TEAMSBOT_BROWSER_BOT_URL = https://cae-poweron-shared.redwater-53d21339.switzerl
 # Debug Configuration
 APP_DEBUG_CHAT_WORKFLOW_ENABLED = FALSE
 APP_DEBUG_CHAT_WORKFLOW_DIR = ./test-chat
+APP_DEBUG_ACCOUNTING_SYNC_ENABLED = FALSE
+APP_DEBUG_ACCOUNTING_SYNC_DIR = ./debug/sync
 
 # Manadate Pre-Processing Servers
 PREPROCESS_ALTHAUS_CHAT_SECRET = PROD_ENC:Z0FBQUFBQnBaSnM4RVRmYW5IelNIbklTUDZIMEoycEN4ZFF0YUJoWWlUTUh2M0dhSXpYRXcwVkRGd1VieDNsYkdCRlpxMUR5Rjk1RDhPRkE5bmVtc2VDMURfLW9QNkxMVHN0M1JhbU9sa3JHWmdDZnlHS3BQRVBGTERVMHhXOVdDOWVqNkhfSUQyOHo=

modules/connectors/connectorDbPostgre.py

@@ -1396,6 +1396,192 @@ class DatabaseConnector:
             self.connection.rollback()
             return False
 
+    def recordCreateBulk(
+        self, model_class: type, records: List[Union[Dict[str, Any], BaseModel]]
+    ) -> int:
+        """Bulk-insert many records in a single transaction.
+
+        Use this instead of calling recordCreate() in a tight loop when importing
+        large datasets (>100 rows). Performance gain is roughly two orders of
+        magnitude because:
+          - one network round-trip via execute_values() instead of N
+          - one COMMIT instead of N
+          - initial ID is registered once for the whole batch instead of every row
+
+        Returns the number of rows successfully inserted. Caller is responsible
+        for catching exceptions; on any error the transaction is rolled back so
+        the table stays consistent (all-or-nothing).
+        """
+        if not records:
+            return 0
+
+        table = model_class.__name__
+        if not self._ensureTableExists(model_class):
+            raise ValueError(f"Table {table} does not exist")
+
+        fields = _get_model_fields(model_class)
+        columns = ["id"] + [f for f in fields.keys() if f != "id"]
+        modelFields = model_class.model_fields
+
+        effectiveUserId = _current_user_id.get()
+        if effectiveUserId is None:
+            effectiveUserId = self.userId
+        currentTime = getUtcTimestamp()
+
+        normalised: List[Dict[str, Any]] = []
+        for raw in records:
+            if isinstance(raw, BaseModel):
+                rec = raw.model_dump()
+            elif isinstance(raw, dict):
+                rec = raw.copy()
+            else:
+                raise ValueError("Bulk record must be a Pydantic model or dictionary")
+            if "id" not in rec or not rec["id"]:
+                rec["id"] = str(uuid.uuid4())
+            createdTs = rec.get("sysCreatedAt")
+            if createdTs is None or createdTs == 0 or createdTs == 0.0:
+                rec["sysCreatedAt"] = currentTime
+                if effectiveUserId:
+                    rec["sysCreatedBy"] = effectiveUserId
+            elif not rec.get("sysCreatedBy") and effectiveUserId:
+                rec["sysCreatedBy"] = effectiveUserId
+            rec["sysModifiedAt"] = currentTime
+            if effectiveUserId:
+                rec["sysModifiedBy"] = effectiveUserId
+            normalised.append(rec)
+
+        rows = [self._coerceRowForInsert(rec, columns, fields, modelFields) for rec in normalised]
+
+        col_names = ", ".join([f'"{c}"' for c in columns])
+        updates = ", ".join(
+            [f'"{c}" = EXCLUDED."{c}"' for c in columns[1:]
+             if c not in ("sysCreatedAt", "sysCreatedBy")]
+        )
+        sql = (
+            f'INSERT INTO "{table}" ({col_names}) VALUES %s '
+            f'ON CONFLICT ("id") DO UPDATE SET {updates}'
+        )
+
+        try:
+            self._ensure_connection()
+            with self.connection.cursor() as cursor:
+                psycopg2.extras.execute_values(cursor, sql, rows, page_size=500)
+            self.connection.commit()
+        except Exception as e:
+            logger.error(f"Bulk insert into {table} failed (n={len(rows)}): {e}")
+            try:
+                self.connection.rollback()
+            except Exception:
+                pass
+            raise
+
+        if self.getInitialId(model_class) is None and normalised:
+            self._registerInitialId(table, normalised[0]["id"])
+            logger.info(f"Registered initial ID {normalised[0]['id']} for table {table}")
+
+        return len(rows)
+
+    def _coerceRowForInsert(
+        self,
+        record: Dict[str, Any],
+        columns: List[str],
+        fields: Dict[str, str],
+        modelFields: Dict[str, Any],
+    ) -> tuple:
+        """Convert one record dict to a positional tuple matching `columns`.
+
+        Mirrors the per-column coercion logic in `_save_record` so that bulk and
+        single inserts produce identical on-disk values (timestamps as floats,
+        enums as strings, vectors as pgvector text, JSONB as JSON strings).
+        """
+        import json as _json
+        out = []
+        for col in columns:
+            value = record.get(col)
+            if col in ("sysCreatedAt", "sysModifiedAt") and value is not None:
+                if isinstance(value, str):
+                    try:
+                        value = float(value)
+                    except Exception:
+                        pass
+            elif hasattr(value, "value"):
+                value = value.value
+            elif col in fields and _isVectorType(fields[col]) and value is not None:
+                if isinstance(value, list):
+                    value = f"[{','.join(str(v) for v in value)}]"
+            elif col in fields and fields[col] == "JSONB" and value is not None:
+                if isinstance(value, (dict, list)):
+                    value = _json.dumps(value)
+                elif isinstance(value, str):
+                    try:
+                        _json.loads(value)
+                    except (ValueError, TypeError):
+                        value = _json.dumps(value)
+                elif hasattr(value, "model_dump"):
+                    value = _json.dumps(value.model_dump())
+                else:
+                    value = _json.dumps(value)
+            out.append(value)
+        return tuple(out)
+
+    def recordDeleteWhere(
+        self, model_class: type, recordFilter: Dict[str, Any]
+    ) -> int:
+        """Delete all records matching a simple equality filter, in one statement.
+
+        Replaces the N+1 pattern `for r in getRecordset(...): recordDelete(r.id)`.
+        Returns the number of rows actually deleted. If the table holds the
+        initial ID and that row gets deleted, the initial ID registration is
+        cleared so the next insert can re-register a fresh one.
+        """
+        if not recordFilter:
+            raise ValueError("recordDeleteWhere requires a non-empty recordFilter (refusing to truncate)")
+
+        table = model_class.__name__
+        if not self._ensureTableExists(model_class):
+            return 0
+
+        fields = _get_model_fields(model_class)
+        clauses: List[str] = []
+        params: List[Any] = []
+        for key, val in recordFilter.items():
+            if key not in fields and key != "id":
+                raise ValueError(f"recordDeleteWhere: unknown column {table}.{key}")
+            clauses.append(f'"{key}" = %s')
+            params.append(val)
+        whereSql = " AND ".join(clauses)
+
+        initialId = self.getInitialId(model_class)
+        try:
+            self._ensure_connection()
+            with self.connection.cursor() as cursor:
+                if initialId is not None:
+                    cursor.execute(
+                        f'SELECT 1 FROM "{table}" WHERE "id" = %s AND ' + whereSql,
+                        [initialId, *params],
+                    )
+                    initialIsAffected = cursor.fetchone() is not None
+                else:
+                    initialIsAffected = False
+
+                cursor.execute(f'DELETE FROM "{table}" WHERE ' + whereSql, params)
+                deleted = cursor.rowcount or 0
+            self.connection.commit()
+        except Exception as e:
+            logger.error(f"Bulk delete from {table} failed (filter={recordFilter}): {e}")
+            try:
+                self.connection.rollback()
+            except Exception:
+                pass
+            raise
+
+        if deleted and initialIsAffected:
+            self._removeInitialId(table)
+            logger.info(f"Initial ID for table {table} cleared (bulk-delete removed it)")
+        if deleted:
+            logger.info(f"recordDeleteWhere: deleted {deleted} rows from {table} where {recordFilter}")
+        return deleted
+
     def getInitialId(self, model_class: type) -> Optional[str]:
         """Returns the initial ID for a table."""
         table = model_class.__name__

modules/features/trustee/accounting/accountingDataSync.py

@@ -2,16 +2,27 @@
 # All rights reserved.
 """Orchestrates importing accounting data from external systems into TrusteeData* tables.
 
-Flow: load config → resolve connector → fetch data → clear old records → write new records → compute balances.
+Flow per phase:
+  1. async fetch from external system (HTTP, awaits cleanly on the event loop)
+  2. await asyncio.to_thread(...) for the DB write phase, so the heavy
+     synchronous psycopg2 calls do NOT block the FastAPI event loop
+  3. inside the worker thread we use bulk delete / bulk insert (single
+     transaction per phase) instead of N+1 single-row operations
+
+Why this matters: a typical accounting sync is ~10k-100k rows. The old
+implementation called ``recordCreate`` row-by-row on the event loop, which
+froze every other request (chat, health-check, etc.) for minutes. See
+``local/notes/changelog.txt`` for the original incident analysis.
 """
 
+import asyncio
 import json as _json
 import logging
 import os
 import time
 from collections import defaultdict
 from pathlib import Path
-from typing import Dict, Any, List, Optional
+from typing import Callable, Dict, Any, List, Optional, Type
 
 from .accountingConnectorBase import BaseAccountingConnector
 from .accountingRegistry import _getAccountingRegistry
@@ -19,30 +30,57 @@ from .accountingRegistry import _getAccountingRegistry
 logger = logging.getLogger(__name__)
 
 
-_DEBUG_SYNC_DIR = Path("D:/Athi/Local/Web/poweron/local/debug/sync")
+_HEARTBEAT_EVERY = 500
 
 
-def _debugSyncDir() -> Path:
+def _isDebugDumpEnabled() -> bool:
-    _DEBUG_SYNC_DIR.mkdir(parents=True, exist_ok=True)
+    """Whether to write raw connector payloads to disk for offline inspection.
-    return _DEBUG_SYNC_DIR
 
-
+    Controlled by ``APP_DEBUG_ACCOUNTING_SYNC_ENABLED``. Default False so that
-def _isDebugEnabled() -> bool:
+    INT/PROD never spend disk/IO/RAM on dumping 7-figure JSON files. Mirrors
+    the existing ``APP_DEBUG_CHAT_WORKFLOW_ENABLED`` pattern.
+    """
     try:
         from modules.shared.configuration import APP_CONFIG
-        return APP_CONFIG.get("APP_LOGGING_FILE_ENABLED", False) is True or str(APP_CONFIG.get("APP_LOGGING_FILE_ENABLED", "")).lower() == "true"
+        raw = APP_CONFIG.get("APP_DEBUG_ACCOUNTING_SYNC_ENABLED", False)
+        if isinstance(raw, bool):
+            return raw
+        return str(raw).strip().lower() in {"true", "1", "yes", "on"}
     except Exception:
         return False
 
 
-def _dumpSyncData(tag: str, rows: list):
+def _resolveDebugDumpDir() -> Optional[Path]:
-    """Write raw connector data to a timestamped JSON file in local/debug/sync/."""
+    """Resolve the debug dump directory. Returns None if dumping is disabled
-    if not _isDebugEnabled():
+    or the path could not be created."""
+    if not _isDebugDumpEnabled():
+        return None
+    try:
+        from modules.shared.configuration import APP_CONFIG
+        configured = APP_CONFIG.get("APP_DEBUG_ACCOUNTING_SYNC_DIR", None)
+        if not configured:
+            return None
+        path = Path(str(configured))
+        if not path.is_absolute():
+            gatewayDir = Path(__file__).resolve().parents[4]
+            path = gatewayDir / configured
+        path.mkdir(parents=True, exist_ok=True)
+        return path
+    except Exception as ex:
+        logger.warning(f"Could not resolve debug dump dir: {ex}")
+        return None
+
+
+def _dumpSyncData(tag: str, rows: list) -> None:
+    """Write raw connector data to a timestamped JSON file. No-op unless
+    ``APP_DEBUG_ACCOUNTING_SYNC_ENABLED`` is true AND the configured dir
+    resolves."""
+    dumpDir = _resolveDebugDumpDir()
+    if dumpDir is None:
         return
     try:
-        d = _debugSyncDir()
         ts = time.strftime("%Y%m%d-%H%M%S")
-        path = d / f"{ts}_{tag}.json"
+        path = dumpDir / f"{ts}_{tag}.json"
         serializable = []
         for r in rows:
             if isinstance(r, dict):
@@ -71,11 +109,25 @@ class AccountingDataSync:
         mandateId: str,
         dateFrom: Optional[str] = None,
         dateTo: Optional[str] = None,
+        progressCb: Optional[Callable[[int, Optional[str]], None]] = None,
     ) -> Dict[str, Any]:
         """Run a full data import for a feature instance.
 
-        Returns a summary dict with counts per entity and any errors.
+        Returns a summary dict with counts per entity and any errors. All heavy
+        DB work is offloaded to a worker thread via ``asyncio.to_thread`` so
+        the event loop remains responsive for other requests.
+
+        ``progressCb(percent, message)`` -- if provided, called at every phase
+        boundary so the UI poll on ``GET /api/jobs/{jobId}`` shows real
+        movement instead of jumping from 10 % to 100 %. Safe to omit.
         """
+        def _progress(pct: int, msg: str) -> None:
+            if progressCb is None:
+                return
+            try:
+                progressCb(pct, msg)
+            except Exception as ex:
+                logger.warning(f"progressCb failed at {pct}%: {ex}")
         from modules.features.trustee.datamodelFeatureTrustee import (
             TrusteeAccountingConfig,
             TrusteeDataAccount,
@@ -109,9 +161,8 @@ class AccountingDataSync:
         encryptedConfig = cfgRecord.get("encryptedConfig", "")
 
         try:
-            import json
             plainJson = decryptValue(encryptedConfig)
-            connConfig = json.loads(plainJson) if plainJson else {}
+            connConfig = _json.loads(plainJson) if plainJson else {}
         except Exception as e:
             summary["errors"].append(f"Failed to decrypt config: {e}")
             return summary
@@ -122,97 +173,87 @@ class AccountingDataSync:
             return summary
 
         scope = {"featureInstanceId": featureInstanceId, "mandateId": mandateId}
-        logger.info(f"AccountingDataSync starting for {featureInstanceId}, connector={connectorType}, dateFrom={dateFrom}, dateTo={dateTo}")
+        logger.info(
+            f"AccountingDataSync starting for {featureInstanceId}, "
+            f"connector={connectorType}, dateFrom={dateFrom}, dateTo={dateTo}"
+        )
         fetchedAccountNumbers: list = []
 
-        # 1) Chart of accounts
+        # ---- Phase 1: Chart of accounts ----
+        # Progress budget: 15-30 %. The fetch (15 %) is usually a single
+        # snappy API call; the persist step (30 %) is bulk-insert and finishes
+        # in <100 ms even for thousands of rows.
         try:
+            _progress(15, "Lade Kontenplan...")
             charts = await connector.getChartOfAccounts(connConfig)
             _dumpSyncData("accounts", charts)
             fetchedAccountNumbers = [acc.accountNumber for acc in charts if acc.accountNumber]
-            self._clearTable(TrusteeDataAccount, featureInstanceId)
+            _progress(25, f"Speichere {len(charts)} Konten...")
-            for acc in charts:
+            written = await asyncio.to_thread(
-                self._if.db.recordCreate(TrusteeDataAccount, {
+                self._persistAccounts, charts, scope, featureInstanceId, TrusteeDataAccount
-                    "accountNumber": acc.accountNumber,
+            )
-                    "label": acc.label,
+            summary["accounts"] = written
-                    "accountType": acc.accountType or "",
+            _progress(30, f"{written} Konten gespeichert.")
-                    "currency": "CHF",
-                    "isActive": True,
-                    **scope,
-                })
-            summary["accounts"] = len(charts)
         except Exception as e:
             logger.error(f"Import accounts failed: {e}", exc_info=True)
             summary["errors"].append(f"Accounts: {e}")
 
-        # 2) Journal entries + lines (pass already-fetched chart to avoid redundant API call)
+        # ---- Phase 2: Journal entries + lines ----
+        # Progress budget: 35-65 %. Usually the longest phase; the external
+        # API often paginates per account, so the fetch alone can take 30+ s.
         try:
-            rawEntries = await connector.getJournalEntries(connConfig, dateFrom=dateFrom, dateTo=dateTo, accountNumbers=fetchedAccountNumbers or None)
+            _progress(35, "Lade Journaleintraege vom Buchhaltungssystem...")
+            rawEntries = await connector.getJournalEntries(
+                connConfig,
+                dateFrom=dateFrom,
+                dateTo=dateTo,
+                accountNumbers=fetchedAccountNumbers or None,
+            )
             _dumpSyncData("journalEntries", rawEntries)
-            self._clearTable(TrusteeDataJournalEntry, featureInstanceId)
+            _progress(60, f"Speichere {len(rawEntries)} Buchungssaetze...")
-            self._clearTable(TrusteeDataJournalLine, featureInstanceId)
+            entriesCount, linesCount = await asyncio.to_thread(
-            lineCount = 0
+                self._persistJournal, rawEntries, scope, featureInstanceId,
-            for raw in rawEntries:
+                TrusteeDataJournalEntry, TrusteeDataJournalLine,
-                import uuid
+            )
-                entryId = str(uuid.uuid4())
+            summary["journalEntries"] = entriesCount
-                self._if.db.recordCreate(TrusteeDataJournalEntry, {
+            summary["journalLines"] = linesCount
-                    "id": entryId,
+            _progress(65, f"{entriesCount} Saetze + {linesCount} Buchungszeilen gespeichert.")
-                    "externalId": raw.get("externalId"),
-                    "bookingDate": raw.get("bookingDate"),
-                    "reference": raw.get("reference"),
-                    "description": raw.get("description", ""),
-                    "currency": raw.get("currency", "CHF"),
-                    "totalAmount": float(raw.get("totalAmount", 0)),
-                    **scope,
-                })
-                for line in (raw.get("lines") or []):
-                    self._if.db.recordCreate(TrusteeDataJournalLine, {
-                        "journalEntryId": entryId,
-                        "accountNumber": line.get("accountNumber", ""),
-                        "debitAmount": float(line.get("debitAmount", 0)),
-                        "creditAmount": float(line.get("creditAmount", 0)),
-                        "currency": line.get("currency", "CHF"),
-                        "taxCode": line.get("taxCode"),
-                        "costCenter": line.get("costCenter"),
-                        "description": line.get("description", ""),
-                        **scope,
-                    })
-                    lineCount += 1
-            summary["journalEntries"] = len(rawEntries)
-            summary["journalLines"] = lineCount
         except Exception as e:
-            logger.error(f"Import journal entries failed: {e}")
+            logger.error(f"Import journal entries failed: {e}", exc_info=True)
             summary["errors"].append(f"Journal entries: {e}")
 
-        # 3) Contacts (customers + vendors)
+        # ---- Phase 3: Contacts (customers + vendors) ----
+        # Progress budget: 70-85 %. Two quick API calls + one bulk-insert.
         try:
-            self._clearTable(TrusteeDataContact, featureInstanceId)
+            _progress(70, "Lade Kunden...")
-            contactCount = 0
-
             customers = await connector.getCustomers(connConfig)
             _dumpSyncData("customers", customers)
-            for c in customers:
+            _progress(78, "Lade Lieferanten...")
-                self._if.db.recordCreate(TrusteeDataContact, self._mapContact(c, "customer", scope))
-                contactCount += 1
-
             vendors = await connector.getVendors(connConfig)
             _dumpSyncData("vendors", vendors)
-            for v in vendors:
+            _progress(82, f"Speichere {len(customers) + len(vendors)} Kontakte...")
-                self._if.db.recordCreate(TrusteeDataContact, self._mapContact(v, "vendor", scope))
+            contactCount = await asyncio.to_thread(
-                contactCount += 1
+                self._persistContacts, customers, vendors, scope,
-
+                featureInstanceId, TrusteeDataContact,
+            )
             summary["contacts"] = contactCount
+            _progress(85, f"{contactCount} Kontakte gespeichert.")
         except Exception as e:
             logger.error(f"Import contacts failed: {e}", exc_info=True)
             summary["errors"].append(f"Contacts: {e}")
 
-        # 4) Compute account balances from journal lines
+        # ---- Phase 4: Compute account balances ----
+        # Progress budget: 90-95 %. Pure DB aggregation, no external calls.
         try:
-            self._clearTable(TrusteeDataAccountBalance, featureInstanceId)
+            _progress(90, "Berechne Kontensaldi...")
-            balanceCount = self._computeBalances(featureInstanceId, mandateId)
+            balanceCount = await asyncio.to_thread(
+                self._persistBalances, featureInstanceId, mandateId,
+                TrusteeDataJournalEntry, TrusteeDataJournalLine, TrusteeDataAccountBalance,
+            )
             summary["accountBalances"] = balanceCount
+            _progress(95, f"{balanceCount} Saldi berechnet.")
         except Exception as e:
-            logger.error(f"Compute balances failed: {e}")
+            logger.error(f"Compute balances failed: {e}", exc_info=True)
             summary["errors"].append(f"Balances: {e}")
 
         cfgId = cfgRecord.get("id")
@@ -255,6 +296,210 @@ class AccountingDataSync:
         )
         return summary
 
+    # ===== Sync persistence helpers (run inside asyncio.to_thread) =====
+
+    def _persistAccounts(self, charts: list, scope: Dict[str, Any],
+                         featureInstanceId: str, modelAccount: Type) -> int:
+        """Bulk-replace chart of accounts for one feature instance."""
+        t0 = time.time()
+        self._bulkClear(modelAccount, featureInstanceId)
+        rows = [{
+            "accountNumber": acc.accountNumber,
+            "label": acc.label,
+            "accountType": acc.accountType or "",
+            "currency": "CHF",
+            "isActive": True,
+            **scope,
+        } for acc in charts]
+        n = self._bulkCreate(modelAccount, rows)
+        logger.info(f"Persisted {n} accounts for {featureInstanceId} in {time.time() - t0:.1f}s")
+        return n
+
+    def _persistJournal(self, rawEntries: list, scope: Dict[str, Any],
+                        featureInstanceId: str, modelEntry: Type, modelLine: Type) -> tuple:
+        """Bulk-replace journal entries + journal lines for one feature instance.
+
+        We pre-build the line rows in memory keyed by the freshly minted entryId
+        so a single ``execute_values`` call can persist all of them.
+        """
+        import uuid as _uuid
+        t0 = time.time()
+        self._bulkClear(modelEntry, featureInstanceId)
+        self._bulkClear(modelLine, featureInstanceId)
+
+        entryRows: List[Dict[str, Any]] = []
+        lineRows: List[Dict[str, Any]] = []
+        for raw in rawEntries:
+            entryId = str(_uuid.uuid4())
+            entryRows.append({
+                "id": entryId,
+                "externalId": raw.get("externalId"),
+                "bookingDate": raw.get("bookingDate"),
+                "reference": raw.get("reference"),
+                "description": raw.get("description", ""),
+                "currency": raw.get("currency", "CHF"),
+                "totalAmount": float(raw.get("totalAmount", 0)),
+                **scope,
+            })
+            for line in (raw.get("lines") or []):
+                lineRows.append({
+                    "journalEntryId": entryId,
+                    "accountNumber": line.get("accountNumber", ""),
+                    "debitAmount": float(line.get("debitAmount", 0)),
+                    "creditAmount": float(line.get("creditAmount", 0)),
+                    "currency": line.get("currency", "CHF"),
+                    "taxCode": line.get("taxCode"),
+                    "costCenter": line.get("costCenter"),
+                    "description": line.get("description", ""),
+                    **scope,
+                })
+            if len(entryRows) % _HEARTBEAT_EVERY == 0:
+                logger.info(
+                    f"Journal build progress: {len(entryRows)}/{len(rawEntries)} entries, "
+                    f"{len(lineRows)} lines so far"
+                )
+
+        entriesCount = self._bulkCreate(modelEntry, entryRows)
+        linesCount = self._bulkCreate(modelLine, lineRows)
+        logger.info(
+            f"Persisted {entriesCount} entries + {linesCount} lines for "
+            f"{featureInstanceId} in {time.time() - t0:.1f}s"
+        )
+        return entriesCount, linesCount
+
+    def _persistContacts(self, customers: list, vendors: list, scope: Dict[str, Any],
+                         featureInstanceId: str, modelContact: Type) -> int:
+        """Bulk-replace contacts (customers + vendors) for one feature instance."""
+        t0 = time.time()
+        self._bulkClear(modelContact, featureInstanceId)
+        rows = [self._mapContact(c, "customer", scope) for c in customers]
+        rows += [self._mapContact(v, "vendor", scope) for v in vendors]
+        n = self._bulkCreate(modelContact, rows)
+        logger.info(f"Persisted {n} contacts for {featureInstanceId} in {time.time() - t0:.1f}s")
+        return n
+
+    def _persistBalances(self, featureInstanceId: str, mandateId: str,
+                         modelEntry: Type, modelLine: Type, modelBalance: Type) -> int:
+        """Re-aggregate journal lines into monthly + annual balances."""
+        t0 = time.time()
+        self._bulkClear(modelBalance, featureInstanceId)
+
+        entries = self._if.db.getRecordset(
+            modelEntry, recordFilter={"featureInstanceId": featureInstanceId},
+        ) or []
+        entryDates: Dict[str, str] = {}
+        for e in entries:
+            eid = e.get("id") if isinstance(e, dict) else getattr(e, "id", None)
+            bdate = e.get("bookingDate") if isinstance(e, dict) else getattr(e, "bookingDate", None)
+            if eid and bdate:
+                entryDates[eid] = bdate
+
+        lines = self._if.db.getRecordset(
+            modelLine, recordFilter={"featureInstanceId": featureInstanceId},
+        ) or []
+
+        buckets: Dict[tuple, Dict[str, float]] = defaultdict(lambda: {"debit": 0.0, "credit": 0.0})
+        for ln in lines:
+            if isinstance(ln, dict):
+                jeid = ln.get("journalEntryId", "")
+                accNo = ln.get("accountNumber", "")
+                debit = float(ln.get("debitAmount", 0))
+                credit = float(ln.get("creditAmount", 0))
+            else:
+                jeid = getattr(ln, "journalEntryId", "")
+                accNo = getattr(ln, "accountNumber", "")
+                debit = float(getattr(ln, "debitAmount", 0))
+                credit = float(getattr(ln, "creditAmount", 0))
+
+            bdate = entryDates.get(jeid, "")
+            if not accNo or not bdate:
+                continue
+            parts = bdate.split("-")
+            if len(parts) < 2:
+                continue
+            try:
+                year = int(parts[0])
+                month = int(parts[1])
+            except ValueError:
+                continue
+
+            buckets[(accNo, year, month)]["debit"] += debit
+            buckets[(accNo, year, month)]["credit"] += credit
+            buckets[(accNo, year, 0)]["debit"] += debit
+            buckets[(accNo, year, 0)]["credit"] += credit
+
+        scope = {"featureInstanceId": featureInstanceId, "mandateId": mandateId}
+        rows = [{
+            "accountNumber": accNo,
+            "periodYear": year,
+            "periodMonth": month,
+            "openingBalance": 0.0,
+            "debitTotal": round(totals["debit"], 2),
+            "creditTotal": round(totals["credit"], 2),
+            "closingBalance": round(totals["debit"] - totals["credit"], 2),
+            "currency": "CHF",
+            **scope,
+        } for (accNo, year, month), totals in buckets.items()]
+        n = self._bulkCreate(modelBalance, rows)
+        logger.info(
+            f"Persisted {n} balances for {featureInstanceId} in {time.time() - t0:.1f}s"
+        )
+        return n
+
+    # ===== Low-level bulk helpers =====
+
+    def _bulkClear(self, model: Type, featureInstanceId: str) -> int:
+        """Delete every row for this feature instance in a single statement."""
+        try:
+            return self._if.db.recordDeleteWhere(
+                model, {"featureInstanceId": featureInstanceId}
+            )
+        except AttributeError:
+            # Backwards-compatible path if the connector hasn't been upgraded
+            # yet. Logs a warning so we notice in dev/CI.
+            logger.warning(
+                "DatabaseConnector.recordDeleteWhere missing — falling back to slow per-row delete for %s",
+                model.__name__,
+            )
+            records = self._if.db.getRecordset(
+                model, recordFilter={"featureInstanceId": featureInstanceId}
+            ) or []
+            count = 0
+            for r in records:
+                rid = r.get("id") if isinstance(r, dict) else getattr(r, "id", None)
+                if rid:
+                    try:
+                        self._if.db.recordDelete(model, rid)
+                        count += 1
+                    except Exception:
+                        pass
+            return count
+
+    def _bulkCreate(self, model: Type, rows: List[Dict[str, Any]]) -> int:
+        """Insert all rows in a single transaction. Falls back to per-row
+        insert only if the connector lacks ``recordCreateBulk`` (legacy)."""
+        if not rows:
+            return 0
+        try:
+            return self._if.db.recordCreateBulk(model, rows)
+        except AttributeError:
+            logger.warning(
+                "DatabaseConnector.recordCreateBulk missing — falling back to slow per-row insert for %s",
+                model.__name__,
+            )
+            n = 0
+            for i, r in enumerate(rows, start=1):
+                try:
+                    self._if.db.recordCreate(model, r)
+                    n += 1
+                except Exception as ex:
+                    logger.warning(f"Per-row insert failed at {i}/{len(rows)}: {ex}")
+                if i % _HEARTBEAT_EVERY == 0:
+                    logger.info(f"Per-row insert progress: {i}/{len(rows)} rows ({model.__name__})")
+            return n
+
+    # ===== Field helpers =====
+
     @staticmethod
     def _safeStr(val: Any) -> str:
         """Convert a value to a safe string for DB storage, collapsing nested dicts/lists."""
@@ -286,84 +531,3 @@ class AccountingDataSync:
             "vatNumber": s(raw.get("vat_identifier") or raw.get("vatNumber") or ""),
             **scope,
         }
-
-    def _clearTable(self, model, featureInstanceId: str):
-        """Delete all records for this feature instance from a TrusteeData* table."""
-        records = self._if.db.getRecordset(model, recordFilter={"featureInstanceId": featureInstanceId})
-        for r in (records or []):
-            rid = r.get("id") if isinstance(r, dict) else getattr(r, "id", None)
-            if rid:
-                try:
-                    self._if.db.recordDelete(model, rid)
-                except Exception:
-                    pass
-
-    def _computeBalances(self, featureInstanceId: str, mandateId: str) -> int:
-        """Aggregate journal lines into monthly + annual account balances."""
-        from modules.features.trustee.datamodelFeatureTrustee import (
-            TrusteeDataJournalEntry,
-            TrusteeDataJournalLine,
-            TrusteeDataAccountBalance,
-        )
-
-        entries = self._if.db.getRecordset(
-            TrusteeDataJournalEntry,
-            recordFilter={"featureInstanceId": featureInstanceId},
-        ) or []
-        entryDates = {}
-        for e in entries:
-            eid = e.get("id") if isinstance(e, dict) else getattr(e, "id", None)
-            bdate = e.get("bookingDate") if isinstance(e, dict) else getattr(e, "bookingDate", None)
-            if eid and bdate:
-                entryDates[eid] = bdate
-
-        lines = self._if.db.getRecordset(
-            TrusteeDataJournalLine,
-            recordFilter={"featureInstanceId": featureInstanceId},
-        ) or []
-
-        # key: (accountNumber, year, month)
-        buckets: Dict[tuple, Dict[str, float]] = defaultdict(lambda: {"debit": 0.0, "credit": 0.0})
-        for ln in lines:
-            if isinstance(ln, dict):
-                jeid = ln.get("journalEntryId", "")
-                accNo = ln.get("accountNumber", "")
-                debit = float(ln.get("debitAmount", 0))
-                credit = float(ln.get("creditAmount", 0))
-            else:
-                jeid = getattr(ln, "journalEntryId", "")
-                accNo = getattr(ln, "accountNumber", "")
-                debit = float(getattr(ln, "debitAmount", 0))
-                credit = float(getattr(ln, "creditAmount", 0))
-
-            bdate = entryDates.get(jeid, "")
-            if not accNo or not bdate:
-                continue
-            parts = bdate.split("-")
-            if len(parts) < 2:
-                continue
-            year = int(parts[0])
-            month = int(parts[1])
-
-            buckets[(accNo, year, month)]["debit"] += debit
-            buckets[(accNo, year, month)]["credit"] += credit
-            buckets[(accNo, year, 0)]["debit"] += debit
-            buckets[(accNo, year, 0)]["credit"] += credit
-
-        count = 0
-        scope = {"featureInstanceId": featureInstanceId, "mandateId": mandateId}
-        for (accNo, year, month), totals in buckets.items():
-            closing = totals["debit"] - totals["credit"]
-            self._if.db.recordCreate(TrusteeDataAccountBalance, {
-                "accountNumber": accNo,
-                "periodYear": year,
-                "periodMonth": month,
-                "openingBalance": 0.0,
-                "debitTotal": round(totals["debit"], 2),
-                "creditTotal": round(totals["credit"], 2),
-                "closingBalance": round(closing, 2),
-                "currency": "CHF",
-                **scope,
-            })
-            count += 1
-        return count

modules/features/trustee/routeFeatureTrustee.py

@@ -1682,14 +1682,15 @@ async def _trusteeAccountingSyncJobHandler(job: Dict[str, Any], progressCb) -> D
     progressCb(5, "Initialisiere Import...")
     interface = getInterface(rootUser, mandateId=mandateId, featureInstanceId=instanceId)
     sync = AccountingDataSync(interface)
-    progressCb(10, "Lese Daten vom Buchhaltungssystem...")
+    progressCb(10, "Verbinde mit Buchhaltungssystem...")
     result = await sync.importData(
         featureInstanceId=instanceId,
         mandateId=mandateId,
         dateFrom=payload.get("dateFrom"),
         dateTo=payload.get("dateTo"),
+        progressCb=progressCb,
     )
-    progressCb(100, "Import abgeschlossen")
+    progressCb(100, "Import abgeschlossen.")
     return result
 
 
@@ -1902,6 +1903,20 @@ def _validateInstanceAdmin(instanceId: str, context: RequestContext) -> str:
     return mandateId
 
 
+def _serializeRoleForApi(role) -> Dict[str, Any]:
+    """Dump a Role and resolve the multilingual ``description`` to a plain string.
+
+    The Role.description field is a ``TextMultilingual`` (``{xx, de, en, ...}``).
+    The frontend expects a plain string, so we resolve via the request language
+    here (same pattern as ``getQuickActions``). Without this the React tree
+    crashes with "Objects are not valid as a React child".
+    """
+    from modules.shared.i18nRegistry import resolveText
+    payload = role.model_dump()
+    payload["description"] = resolveText(payload.get("description"))
+    return payload
+
+
 @router.get("/{instanceId}/instance-roles", response_model=PaginatedResponse)
 @limiter.limit("30/minute")
 def get_instance_roles(
@@ -1921,7 +1936,7 @@ def get_instance_roles(
     roles = rootInterface.getRolesByFeatureCode("trustee", featureInstanceId=instanceId)
     
     return PaginatedResponse(
-        items=[r.model_dump() for r in roles],
+        items=[_serializeRoleForApi(r) for r in roles],
         pagination=None
     )
 
@@ -1947,7 +1962,7 @@ def get_instance_role(
     if str(role.featureInstanceId) != instanceId:
         raise HTTPException(status_code=404, detail=f"Role {roleId} not found in this instance")
     
-    return role.model_dump()
+    return _serializeRoleForApi(role)
 
 
 @router.get("/{instanceId}/instance-roles/{roleId}/rules", response_model=PaginatedResponse)

modules/interfaces/interfaceDbChat.py

@@ -219,6 +219,22 @@ class ChatObjects:
         # Everything else is an object
         return True
 
+    def _unwrapOptional(self, fieldType):
+        """Unwrap ``Optional[X]`` / ``Union[X, None]`` to ``X``.
+
+        The generic JSONB detection in ``_separateObjectFields`` checks
+        ``__origin__`` against ``(dict, list)``. For ``Optional[List[str]]``
+        the origin is ``Union``, so JSONB fields declared as ``Optional[...]``
+        would silently fall through to ``objectFields`` and be dropped on
+        write. Unwrapping the Optional first keeps the existing detection
+        intact while supporting nullable JSONB columns.
+        """
+        if getattr(fieldType, '__origin__', None) is Union:
+            nonNone = [a for a in getattr(fieldType, '__args__', ()) if a is not type(None)]
+            if len(nonNone) == 1:
+                return nonNone[0]
+        return fieldType
+
     def _separateObjectFields(self, model_class, data: Dict[str, Any]) -> tuple[Dict[str, Any], Dict[str, Any]]:
         """Separate simple fields from object fields based on Pydantic model structure."""
         simpleFields = {}
@@ -232,7 +248,7 @@ class ChatObjects:
             if fieldName in modelFields:
                 fieldInfo = modelFields[fieldName]
                 # Pydantic v2 only
-                fieldType = fieldInfo.annotation
+                fieldType = self._unwrapOptional(fieldInfo.annotation)
 
                 # Always route relational/object fields to object_fields for separate handling
                 # These fields are stored in separate normalized tables, not as JSONB

modules/routes/routeI18n.py

@@ -42,7 +42,11 @@ from modules.datamodels.datamodelNotification import NotificationType
 from modules.interfaces.interfaceDbManagement import getInterface as getMgmtInterface
 from modules.routes.routeNotifications import _createNotification
 from modules.shared.configuration import APP_CONFIG
-from modules.shared.i18nRegistry import _loadCache as _reloadI18nCache, apiRouteContext
+from modules.shared.i18nRegistry import (
+    _enforceSourcePlaceholders,
+    _loadCache as _reloadI18nCache,
+    apiRouteContext,
+)
 from modules.shared.timeUtils import getUtcTimestamp
 
 routeApiMsg = apiRouteContext("routeI18n")
@@ -248,7 +252,12 @@ async def _translateBatch(
             f"2. If the source is already in the target language, keep it (do not re-translate, "
             f"do not paraphrase).\n"
             f"3. KEEP the exact JSON keys from the input — do NOT translate or modify the keys.\n"
-            f"4. KEEP placeholders like {{variable}}, {{count}}, %s, %(name)s exactly as they are.\n"
+            f"4. PLACEHOLDERS ARE SACRED. Tokens of the form {{name}}, {{count}}, "
+            f"{{konten}}, {{anyWord}}, %s, %(name)s, %d MUST be copied character-for-"
+            f"character into the translation, EVEN IF the name inside the curly braces "
+            f"looks like a German or English word. Never translate, rename, reorder, "
+            f"add, or remove placeholders. Example: '{{konten}} Konten' translated to "
+            f"English MUST stay '{{konten}} accounts' — NEVER '{{accounts}} accounts'.\n"
             f"5. Preserve leading/trailing whitespace, punctuation and capitalisation pattern.\n"
             f"6. Answer ONLY with a JSON object mapping source-key -> translated value in "
             f"{targetLanguageLabel}. No markdown fences, no comments, no explanations.\n"
@@ -320,10 +329,31 @@ async def _translateBatch(
         if batchIdx < totalBatches - 1:
             await asyncio.sleep(_TRANSLATE_BATCH_PAUSE_S)
 
+    _enforcePlaceholdersOnBatch(result)
     _matchCapitalization(keysToTranslate, result)
     return result
 
 
+def _enforcePlaceholdersOnBatch(translations: Dict[str, str]) -> None:
+    """Ensure every translated value preserves the source key's placeholders.
+
+    See ``_enforceSourcePlaceholders`` for the detailed strategy. Mutates
+    ``translations`` in place; logs a warning per repaired key.
+    """
+    repaired = 0
+    for sourceKey, translatedValue in list(translations.items()):
+        fixed, changed = _enforceSourcePlaceholders(sourceKey, translatedValue)
+        if changed:
+            translations[sourceKey] = fixed
+            repaired += 1
+            logger.warning(
+                "i18n placeholder mismatch repaired: %r -> %r",
+                translatedValue, fixed,
+            )
+    if repaired:
+        logger.info("i18n batch: repaired placeholders in %d translations", repaired)
+
+
 def _matchCapitalization(originals: Dict[str, str], translations: Dict[str, str]) -> None:
     """Ensure translations preserve the capitalisation pattern of the original key."""
     for key, translated in translations.items():
@@ -857,6 +887,87 @@ async def sync_xx_master(
     return result
 
 
+def _repairLanguageSetPlaceholders(db, code: str, userId: Optional[str]) -> dict:
+    """Persistently fix placeholder mismatches in one language set.
+
+    Walks every entry, runs ``_enforceSourcePlaceholders(key, value)`` and
+    persists any changed values back to the row. Only saves if at least one
+    entry was modified.
+    """
+    rows = db.getRecordset(UiLanguageSet, recordFilter={"id": code})
+    if not rows:
+        raise HTTPException(status_code=404, detail=routeApiMsg("Sprachset nicht gefunden"))
+    row = dict(rows[0])
+    entries = _rowEntries(row)
+    repaired: List[Dict[str, str]] = []
+    for entry in entries:
+        key = entry.get("key", "")
+        val = entry.get("value", "")
+        fixed, changed = _enforceSourcePlaceholders(key, val)
+        if changed:
+            repaired.append({"key": key, "before": val, "after": fixed})
+            entry["value"] = fixed
+
+    if repaired:
+        row["entries"] = entries
+        if "keys" in row:
+            del row["keys"]
+        row["sysModifiedAt"] = getUtcTimestamp()
+        row["sysModifiedBy"] = userId
+        db.recordModify(UiLanguageSet, code, row)
+
+    return {
+        "code": code,
+        "checked": len(entries),
+        "repaired": len(repaired),
+        "examples": repaired[:10],
+    }
+
+
+@router.post("/sets/{code}/repair-placeholders")
+async def repair_language_set_placeholders(
+    code: str,
+    adminUser: User = Depends(requireSysAdmin),
+):
+    """SysAdmin: persistently restore placeholder tokens in one language set.
+
+    Use this once after the AI translator turned ``{konten}`` into
+    ``{accounts}`` (or similar). Compares each entry's value against its
+    German source key; where the placeholder *names* differ but the *count*
+    matches, restores the source names positionally. Safe and idempotent.
+    """
+    c = code.strip().lower()
+    if c == "xx":
+        raise HTTPException(status_code=400, detail=routeApiMsg("Das xx-Set hat keine Übersetzungen."))
+    db = getMgmtInterface(adminUser, mandateId=None).db
+    result = _repairLanguageSetPlaceholders(db, c, str(adminUser.id))
+    await _reloadI18nCache()
+    return result
+
+
+@router.post("/sets/repair-placeholders-all")
+async def repair_all_language_sets_placeholders(
+    adminUser: User = Depends(requireSysAdmin),
+):
+    """SysAdmin: persistently restore placeholder tokens in ALL language sets."""
+    db = getMgmtInterface(adminUser, mandateId=None).db
+    rows = db.getRecordset(UiLanguageSet)
+    summary: List[dict] = []
+    totalRepaired = 0
+    for row in rows:
+        code = row.get("id", "")
+        if not code or code == "xx":
+            continue
+        try:
+            res = _repairLanguageSetPlaceholders(db, code, str(adminUser.id))
+            summary.append(res)
+            totalRepaired += res["repaired"]
+        except HTTPException:
+            continue
+    await _reloadI18nCache()
+    return {"languages": len(summary), "totalRepaired": totalRepaired, "details": summary}
+
+
 @router.get("/sets/{code}/sync-diff")
 async def get_language_sync_diff(
     code: str,

modules/serviceCenter/services/serviceSubscription/mainServiceSubscription.py

@@ -246,6 +246,30 @@ class SubscriptionService:
         if not priceMapping or (not priceMapping.stripePriceIdUsers and not priceMapping.stripePriceIdInstances):
             raise ValueError(f"Stripe Price IDs not provisioned for plan {plan.planKey}")
 
+        # Defense in depth: if either of the persisted Stripe Price IDs has been
+        # archived in Stripe in the meantime (e.g. another environment's bootstrap
+        # rotated them on a shared Stripe account), the upcoming
+        # ``checkout.Session.create`` would fail with "The price specified is
+        # inactive". Trigger a one-shot bootstrap re-run to rotate inactive prices,
+        # then reload the mapping. This is idempotent and cheap when nothing
+        # changed.
+        if not self._areStripePricesActive(stripe, priceMapping):
+            logger.warning(
+                "Stripe Price(s) for plan %s are no longer active in Stripe — "
+                "running bootstrap to rotate.", plan.planKey,
+            )
+            try:
+                from modules.serviceCenter.services.serviceSubscription.stripeBootstrap import bootstrapStripePrices
+                bootstrapStripePrices()
+                priceMapping = getStripePricesForPlan(plan.planKey)
+            except Exception as ex:
+                logger.error("Inline Stripe bootstrap failed for plan %s: %s", plan.planKey, ex)
+            if not priceMapping or not self._areStripePricesActive(stripe, priceMapping):
+                raise ValueError(
+                    f"Stripe Price IDs for plan {plan.planKey} are inactive and "
+                    "could not be rotated automatically."
+                )
+
         stripeCustomerId = self._resolveStripeCustomer(mandateId)
         if not stripeCustomerId:
             raise ValueError(f"Could not resolve Stripe customer for mandate {mandateId}")
@@ -353,6 +377,27 @@ class SubscriptionService:
         except Exception as e:
             logger.error("Failed to clear stripeCustomerId for mandate %s: %s", mandateId, e)
 
+    def _areStripePricesActive(self, stripe, priceMapping) -> bool:
+        """Verify that every persisted Stripe Price ID for the plan is still
+        ``active`` in Stripe. ``stripe.Price.retrieve`` returns archived prices
+        too, so we must inspect the ``active`` flag explicitly. Returns True
+        only when ALL non-empty price IDs resolve to active prices."""
+        priceIds = [pid for pid in (
+            getattr(priceMapping, "stripePriceIdUsers", None),
+            getattr(priceMapping, "stripePriceIdInstances", None),
+        ) if pid]
+        if not priceIds:
+            return False
+        for pid in priceIds:
+            try:
+                price = stripe.Price.retrieve(pid)
+                if not bool(getattr(price, "active", False) if not isinstance(price, dict) else price.get("active")):
+                    return False
+            except Exception as ex:
+                logger.warning("Stripe Price %s could not be retrieved: %s", pid, ex)
+                return False
+        return True
+
     def _resolveStripeCustomer(self, mandateId: str) -> Optional[str]:
         try:
             from modules.interfaces.interfaceDbBilling import getInterface as getBillingInterface

modules/serviceCenter/services/serviceSubscription/stripeBootstrap.py

@@ -124,16 +124,6 @@ def _findExistingStripePrice(
     return None
 
 
-def _getStripePriceAmount(stripe, priceId: str) -> Optional[int]:
-    """Retrieve the unit_amount (in Rappen) of an existing Stripe Price."""
-    try:
-        from modules.shared.stripeClient import stripeToDict
-        price = stripeToDict(stripe.Price.retrieve(priceId))
-        return price.get("unit_amount") if price else None
-    except Exception:
-        return None
-
-
 def _reconcilePrice(
     stripe,
     productId: str,
@@ -143,25 +133,35 @@ def _reconcilePrice(
     nickname: str,
     intervalCount: int = 1,
 ) -> str:
-    """If the stored Stripe Price has a different amount, create a new one and deactivate the old."""
+    """If the stored Stripe Price has a different amount, is no longer active,
+    or has a different recurring interval, create/find a new active one and
+    deactivate the old."""
     from modules.shared.stripeClient import stripeToDict
 
     expectedCents = int(round(expectedCHF * 100))
-    actualCents = _getStripePriceAmount(stripe, oldPriceId)
+    actualCents: Optional[int] = None
     matchesRecurring = False
+    isActive = False
+    retrieveFailed = False
     try:
         raw = stripe.Price.retrieve(oldPriceId)
         pd = stripeToDict(raw)
+        actualCents = pd.get("unit_amount")
         matchesRecurring = _recurringMatches(pd.get("recurring") or {}, interval, intervalCount)
-    except Exception:
+        # Stripe.Price.retrieve returns archived prices too, so we MUST check
-        pass
+        # `active` explicitly. Subscription.create rejects inactive prices with
+        # "The price specified is inactive. This field only accepts active prices."
+        isActive = bool(pd.get("active"))
+    except Exception as ex:
+        retrieveFailed = True
+        logger.warning("Could not retrieve Stripe Price %s: %s", oldPriceId, ex)
 
-    if actualCents == expectedCents and matchesRecurring:
+    if not retrieveFailed and isActive and actualCents == expectedCents and matchesRecurring:
         return oldPriceId
 
     logger.warning(
-        "Price drift or recurring mismatch for %s: Stripe amount=%s Rappen (expected %s). Rotating price.",
+        "Rotating Stripe Price %s on product %s: active=%s amount=%s (expected %s) recurringMatches=%s retrieveFailed=%s.",
-        oldPriceId, actualCents, expectedCents,
+        oldPriceId, productId, isActive, actualCents, expectedCents, matchesRecurring, retrieveFailed,
     )
 
     existingMatch = _findExistingStripePrice(stripe, productId, expectedCents, interval, intervalCount)
@@ -221,8 +221,13 @@ def _archiveOtherRecurringPrices(
 
 
 def _validateStripeIdsExist(stripe, mapping: StripePlanPrice) -> bool:
-    """Quick check whether at least the stored product IDs still exist in Stripe.
+    """Quick check whether the stored Stripe product IDs still exist.
-    Returns False when running against a different Stripe account or after DB copy."""
+
+    Returns False when running against a different Stripe account or after a
+    DB copy from another environment. Price-level validation (active flag,
+    drift) is handled by ``_reconcilePrice``; we don't fail here on archived
+    prices, otherwise we'd needlessly re-provision products on every rotation.
+    """
     try:
         if mapping.stripeProductIdUsers:
             stripe.Product.retrieve(mapping.stripeProductIdUsers)

modules/shared/i18nRegistry.py

@@ -12,15 +12,59 @@ At runtime, t() returns the cached translation for the current request language.
 from __future__ import annotations
 
 import logging
+import re
 from contextvars import ContextVar
 from dataclasses import dataclass, field as dataclass_field
-from typing import Any, Dict, List, Optional, Type
+from typing import Any, Dict, List, Optional, Tuple, Type
 
 from pydantic import BaseModel
 
 logger = logging.getLogger(__name__)
 
 
+# Matches {placeholderName} tokens used by t(...) param substitution in the
+# frontend (LanguageContext._applyParams) and the gateway. Allows ASCII
+# identifiers and digits, no spaces.
+_PLACEHOLDER_PATTERN = re.compile(r"\{[A-Za-z_][A-Za-z0-9_]*\}")
+
+
+def _enforceSourcePlaceholders(sourceKey: str, translatedValue: str) -> Tuple[str, bool]:
+    """Repair a translated value so its placeholder tokens match the source key.
+
+    Background: AI translators occasionally translate the *names* of
+    placeholders even when instructed not to (e.g. ``{konten}`` -> ``{accounts}``).
+    The frontend then cannot substitute params and the user sees raw
+    ``{accounts}`` in the UI.
+
+    Strategy (positional, conservative):
+      - if the source has no placeholders -> nothing to do
+      - if source and translation have the same set of tokens -> nothing to do
+      - if both have the *same number* of tokens but different names -> swap
+        each translation token with the source token at the same position
+      - if counts differ -> leave the translation untouched (too risky to
+        guess; surfaced as a logger.warning by the caller if desired)
+
+    Returns ``(repairedValue, wasChanged)``.
+    """
+    if not sourceKey or not translatedValue:
+        return translatedValue, False
+    sourceTokens = _PLACEHOLDER_PATTERN.findall(sourceKey)
+    if not sourceTokens:
+        return translatedValue, False
+    valueTokens = _PLACEHOLDER_PATTERN.findall(translatedValue)
+    if not valueTokens:
+        return translatedValue, False
+    if sourceTokens == valueTokens:
+        return translatedValue, False
+    if len(sourceTokens) != len(valueTokens):
+        return translatedValue, False
+    parts = _PLACEHOLDER_PATTERN.split(translatedValue)
+    rebuilt = parts[0]
+    for idx, srcTok in enumerate(sourceTokens):
+        rebuilt += srcTok + parts[idx + 1]
+    return rebuilt, True
+
+
 def _extractRegistrySourceText(obj: Any) -> str:
     """Resolve a str or multilingual dict to one canonical registry key string."""
     if isinstance(obj, str):
@@ -492,6 +536,48 @@ def _registerNodeLabels():
     logger.info("i18n node labels: %d new keys (node.*/port.* context)", added)
 
 
+def _registerAccountingConnectorLabels():
+    """Register all accounting connector configField labels (label) at boot time.
+
+    Connector ``getRequiredConfigFields()`` is normally invoked lazily at first
+    request, which is too late for the boot-sync. We discover the connectors
+    here so their ``t()`` calls register the keys before they are written to the
+    ``xx`` set and AI-translated for every active language set.
+    """
+    added = 0
+    try:
+        from modules.features.trustee.accounting.accountingRegistry import _getAccountingRegistry
+    except ImportError:
+        logger.debug("i18n accounting connectors: registry not importable")
+        return
+
+    try:
+        registry = _getAccountingRegistry()
+    except Exception as e:
+        logger.warning("i18n accounting connectors: registry init failed: %s", e)
+        return
+
+    for connectorType, connector in (registry._connectors or {}).items():
+        try:
+            for field in connector.getRequiredConfigFields():
+                key = getattr(field, "label", "") or ""
+                if not isinstance(key, str) or not key:
+                    continue
+                if key not in _REGISTRY:
+                    _REGISTRY[key] = _I18nRegistryEntry(
+                        context=f"connector.accounting.{connectorType}",
+                        value="",
+                    )
+                    added += 1
+        except Exception as e:
+            logger.warning(
+                "i18n accounting connector %s: failed to read fields: %s",
+                connectorType, e,
+            )
+
+    logger.info("i18n accounting connector labels: %d new keys", added)
+
+
 def _registerDatamodelOptionLabels():
     """Register all frontend_options labels from Pydantic datamodels and subscription plans."""
     added = 0
@@ -568,6 +654,7 @@ async def _syncRegistryToDb():
     _registerServiceCenterLabels()
     _registerNodeLabels()
     _registerDatamodelOptionLabels()
+    _registerAccountingConnectorLabels()
 
     if not _REGISTRY:
         logger.info("i18n registry: no keys to sync (empty registry)")
@@ -668,6 +755,13 @@ async def _syncRegistryToDb():
 async def _loadCache():
     """Boot hook: load all UiLanguageSets into the in-memory cache.
 
+    Also persistently repairs placeholder mismatches in the DB:
+    if an entry's value has placeholder *names* that differ from the
+    source key (typical AI translation mishap, e.g. ``{konten}`` ->
+    ``{accounts}``), the source names are restored positionally and the
+    row is written back to the DB. Idempotent and safe -- only mutates
+    when the placeholder count matches and the names actually differ.
+
     After this, t() lookups are O(1) dict access with no DB calls.
     """
     from modules.datamodels.datamodelUiLanguage import UiLanguageSet
@@ -686,6 +780,8 @@ async def _loadCache():
     rows = db.getRecordset(UiLanguageSet)
     _CACHE.clear()
 
+    repairedTotal = 0
+    persistedLanguages = 0
     for row in rows:
         code = row.get("id", "")
         if code == "xx":
@@ -694,13 +790,49 @@ async def _loadCache():
         if not isinstance(entries, list):
             continue
         langDict: Dict[str, str] = {}
-        for e in entries:
+        repairedInLang = 0
-            key = e.get("key", "")
+        # Walk a mutable copy so we can write the corrected entries back to
-            val = e.get("value", "")
+        # the row without re-reading from the DB.
-            if key and val:
+        for entry in entries:
-                langDict[key] = val
+            key = entry.get("key", "")
+            val = entry.get("value", "")
+            if not key or not val:
+                continue
+            fixed, changed = _enforceSourcePlaceholders(key, val)
+            if changed:
+                entry["value"] = fixed
+                repairedInLang += 1
+            langDict[key] = fixed
         if langDict:
             _CACHE[code] = langDict
+        if repairedInLang:
+            repairedTotal += repairedInLang
+            try:
+                rowToSave = dict(row)
+                rowToSave["entries"] = entries
+                if "keys" in rowToSave:
+                    del rowToSave["keys"]
+                db.recordModify(UiLanguageSet, code, rowToSave)
+                persistedLanguages += 1
+                logger.info(
+                    "i18n boot repair: fixed and persisted %d placeholder mismatches in language '%s'",
+                    repairedInLang, code,
+                )
+            except Exception as ex:
+                # Persistence is best-effort -- the in-memory cache is
+                # already correct (langDict above contains the fixed
+                # values), so the UI works either way. Log and move on.
+                logger.warning(
+                    "i18n boot repair: in-memory fixed %d entries in '%s' but DB persist failed: %s",
+                    repairedInLang, code, ex,
+                )
 
-    logger.info("i18n cache loaded: %d languages, %d total keys",
+    logger.info(
-                len(_CACHE), sum(len(v) for v in _CACHE.values()))
+        "i18n cache loaded: %d languages, %d total keys%s",
+        len(_CACHE), sum(len(v) for v in _CACHE.values()),
+        (
+            f" (boot-repaired {repairedTotal} placeholders, "
+            f"persisted to {persistedLanguages} language sets)"
+            if repairedTotal else ""
+        ),
+    )