feat: Improve session metrics upsert logic and add unit tests for source field handling

- Enhanced the `upsert_session_metrics_from_csv_mapped` function to skip EAV writes when a `source_field` is provided, ensuring canonical handling of activity logs.
- Updated the docstring to clarify the behavior regarding `source_field` and EAV logic.
- Added unit tests to validate the new behavior, ensuring correct functionality when `source_field` is set or not set during CSV imports.

backend/data_layer/activity_session_metrics.py

@@ -256,6 +256,10 @@ def upsert_session_metrics_from_csv_mapped(
 
     Kernfelder (Datum, Start, Distanz, HF, …) schreibt der Executor nach activity_log;
     hier keine doppelten EAV-Zeilen für dieselben Registry-Keys.
+
+    Ist ``training_parameters.source_field`` gesetzt, ist die zugehörige ``activity_log``-Spalte
+    kanonisch (wie beim Lesen in ``merge_column_backed_and_eav_metrics``) — kein EAV-Schreiben,
+    auch wenn der Parameter-Key vom Registry-Schlüssel abweicht.
     """
     cur.execute(
         "SELECT profile_id FROM activity_log WHERE id = %s",
@@ -274,6 +278,9 @@ def upsert_session_metrics_from_csv_mapped(
             continue
         if pkey in ACTIVITY_MODULE_REGISTRY_FIELD_KEYS:
             continue
+        sf_raw = spec.get("source_field")
+        if sf_raw is not None and str(sf_raw).strip():
+            continue
         tid = spec["training_parameter_id"]
         dt = spec["data_type"]
         rules = _validation_rules_dict(spec["validation_rules"])
@@ -308,9 +315,9 @@ def merge_column_backed_and_eav_metrics(
     eav_metrics: Sequence[Dict[str, Any]],
 ) -> List[Dict[str, Any]]:
     """
-    Effektive Metrikliste: Pro Schema-Parameter mit source_field gilt activity_log als kanonisch, wenn
+    Effektive Metrikliste: Pro Schema-Parameter mit ``source_field`` hat ``activity_log`` Vorrang, wenn
-    die Spalte befüllt und koerzierbar ist; sonst Fallback EAV. Reine EAV-Parameter (ohne Spalte oder
+    die Spalte befüllt und koerzierbar ist; sonst EAV, sonst Legacy-Spalte (EAV-primär). Eine Semantik
-    leere Spalte) kommen aus EAV. Verhindert doppelte Semantik ohne Schreib-Sync.
+    erscheint nur einmal — konsistent mit CSV-EAV-Upsert (dort kein Schreiben bei gesetztem ``source_field``).
     """
     eav_by_key = {m["key"]: m for m in eav_metrics}
     merged: List[Dict[str, Any]] = []

backend/tests/test_activity_session_metrics.py

@@ -11,6 +11,7 @@ from data_layer.activity_session_metrics import (
     merge_column_backed_and_eav_metrics,
     merge_parameter_schema_rows,
     resolve_activity_attribute_schema,
+    upsert_session_metrics_from_csv_mapped,
     _row_value_tuple,
     _validate_single_value,
 )
@@ -305,3 +306,74 @@ def test_merge_eav_primary_falls_back_to_legacy_hr_min_column():
     assert len(out) == 1
     assert out[0]["key"] == "min_hr"
     assert out[0]["value"] == 88
+
+
+@patch("data_layer.activity_session_metrics.resolve_activity_attribute_schema")
+def test_upsert_csv_skips_eav_when_source_field_maps_activity_log(mock_schema):
+    """Parameter mit source_field: kanonisch activity_log — kein doppeltes EAV (z. B. avg_hr → hr_avg)."""
+    mock_schema.return_value = [
+        {
+            "key": "avg_hr",
+            "training_parameter_id": 42,
+            "data_type": "integer",
+            "validation_rules": {"min": 30, "max": 220},
+            "source_field": "hr_avg",
+        }
+    ]
+
+    class Cur:
+        def __init__(self):
+            self.asm_inserts = 0
+
+        def execute(self, sql, params=None):
+            if "INSERT INTO activity_session_metrics" in sql:
+                self.asm_inserts += 1
+
+        def fetchone(self):
+            return {"profile_id": "00000000-0000-0000-0000-000000000001"}
+
+    cur = Cur()
+    upsert_session_metrics_from_csv_mapped(
+        cur,
+        "00000000-0000-0000-0000-000000000001",
+        "00000000-0000-0000-0000-000000000002",
+        {"avg_hr": 130},
+        "cardio",
+        1,
+    )
+    assert cur.asm_inserts == 0
+
+
+@patch("data_layer.activity_session_metrics.resolve_activity_attribute_schema")
+def test_upsert_csv_writes_eav_when_no_source_field(mock_schema):
+    mock_schema.return_value = [
+        {
+            "key": "custom_note",
+            "training_parameter_id": 99,
+            "data_type": "string",
+            "validation_rules": {},
+            "source_field": None,
+        }
+    ]
+
+    class Cur:
+        def __init__(self):
+            self.asm_inserts = 0
+
+        def execute(self, sql, params=None):
+            if "INSERT INTO activity_session_metrics" in sql:
+                self.asm_inserts += 1
+
+        def fetchone(self):
+            return {"profile_id": "00000000-0000-0000-0000-000000000001"}
+
+    cur = Cur()
+    upsert_session_metrics_from_csv_mapped(
+        cur,
+        "00000000-0000-0000-0000-000000000001",
+        "00000000-0000-0000-0000-000000000002",
+        {"custom_note": "x"},
+        "cardio",
+        1,
+    )
+    assert cur.asm_inserts == 1