fix(check-collections): qualify check paths with collection kind to ensure uniqueness (#2792)

* refactor(check): rename path->relative_path, full_path->check_path

* fix(test): complete path->relative_path / full_path->check_path rename across unit tests

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix(check-selector): match --check-paths against full check_path (fixes data-standard partial runs)

* feat(check-selector): add source=, collection=/standard=, relative_path= filter fields

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(check-path): emit option-3 format type.id:relative for non-contract collections

Non-contract check collections (e.g. data standards) now emit
checkPath as "{wire_source}.{collection_id}:{relative_path}" (option 3)
instead of the old "{collection_id}.{relative_path}" format. Contracts
stay bare (checkPath == relative_path). Adds a delimiter-safety guard
in base.py verify() that rejects collection_id values containing '.'
or ':' (reserved check-path delimiters). Updates all stub check_path
values in selector and alignment tests to the option-3 format.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* test(check-path): migrate data-standard stubs to option-3 format; doc + name accuracy

* test+guard: self-updating supported-fields test, wire_source delimiter guard, negative coverage

* fix(check-collections): surface isolated parse/verify errors instead of failing silently

A data standard with an invalid `name:` (e.g. `test.standard`) exited 3 while
printing nothing at all. The name guard was firing correctly — the message just
never reached the user.

`build_error_result` is the isolation boundary for `abort_on_first_error=False`
callers: it set `status=ERROR` (driving exit 3), stored the exception on
`result.error`, set `log_records=None`, and never logged. The only reader of
`result.error` is the Cloud `mark_scan_as_failed` path, gated on a scan_id — so
a local run surfaced nothing.

Not specific to the dot: every construct/verify-time error was silent this way.
It hits data standards always (api.py pins `abort_on_first_error=False`) and any
multi-file contract run (`abort_on_first_error=(len(sources) == 1)`). Only the
single-contract path re-raised, which is why it went unnoticed.

Log the exception where it is isolated, captured into the placeholder's own
`Logs` so it is attributed to the failing file rather than to whichever sibling
happens to be the active capture target (constructing a Logs leaves it active,
so after the construct phase that is the last-constructed collection — a bare
emit would file the error under an innocent sibling and ship it in that
sibling's Cloud log payload). `log_records` now carries the error line, so an
errored placeholder is self-diagnosing instead of an empty payload.

Before: `Exiting with code 3` and nothing else.
After:  `🚨 data-standard 'name' must not contain '.' or ':': got 'test.standard', in <file>`

Tests: two regression tests pinning that isolated parse and verify failures are
logged. Full unit suite green (994 passed, 2 skipped).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* style(check-selector): black formatting for SUPPORTED_FIELDS

CI runs pre-commit --all-files; this file was left unformatted by an
earlier commit on this branch.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* docs(check-path): correct comments left describing the pre-option-3 wire format

Three comments still documented the old "{collection_id}.{relative_path}" prefix
and routing via firstSegmentOf(checkPath) — a backend function this change
deletes. They now describe the option-3 form
"{wire_source}.{collection_id}:{relative_path}" and how the backend parses it.

Comments only; no behaviour change.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: m1n0

soda-core/src/soda_core/check_collections/base.py

@@ -26,7 +26,7 @@
 from soda_core.common.env_config_helper import EnvConfigHelper
 from soda_core.common.exceptions import SodaCoreException, get_exception_stacktrace
 from soda_core.common.logging_constants import Emoticons, ExtraKeys, soda_logger
-from soda_core.common.logs import Location, Logs
+from soda_core.common.logs import Location, Logs, preserve_active_logs
 from soda_core.common.metadata_types import SamplerType
 from soda_core.common.soda_cloud_converter import map_sampler_type_from_dto
 from soda_core.common.soda_cloud_dto import DatasetConfigurationDTO
@@ -801,6 +801,16 @@ def verify(self) -> CheckCollectionResult:
                 f"collection_id (used to prefix checkPath for backend routing)."
             )
 
+        if self.collection_id and (("." in self.collection_id) or (":" in self.collection_id)):
+            raise ValueError(
+                f"collection_id {self.collection_id!r} must not contain '.' or ':' (reserved check-path delimiters)"
+            )
+
+        if self.wire_source and (("." in self.wire_source) or (":" in self.wire_source)):
+            raise ValueError(
+                f"wire_source {self.wire_source!r} must not contain '.' or ':' (reserved check-path delimiters)"
+            )
+
         if self.data_source_impl and self.soda_config.is_running_on_runner:
             self.data_source_impl.switch_warehouse(self.compute_warehouse, contract_impl=self)
         data_source: Optional[DataSource] = None
@@ -846,7 +856,7 @@ def verify(self) -> CheckCollectionResult:
                 # Evaluate the checks
                 for check_impl in self.all_check_impls:
                     if check_impl.skip:
-                        logger.info(f"Skipping evaluation of check at path '{check_impl.path}'")
+                        logger.info(f"Skipping evaluation of check at path '{check_impl.relative_path}'")
                         check_result: CheckResult = CheckResult(
                             check=check_impl._build_check_info(), outcome=CheckOutcome.EXCLUDED
                         )
@@ -1046,6 +1056,30 @@ def build_error_result(
         Used by per-item isolation in ``execute_check_collections``.
         """
         now = datetime.now(tz=timezone.utc)
+        # Surface the failure instead of only stashing it on ``result.error``.
+        # This is the isolation boundary for ``abort_on_first_error=False``
+        # callers — always data standards, and any multi-file contract run.
+        # They never re-raise, and the only other reader of ``result.error`` is
+        # the Cloud ``mark_scan_as_failed`` path (gated on a scan_id), so
+        # without this a local run exits non-zero having printed nothing.
+        #
+        # Capture into this placeholder's OWN Logs rather than emitting bare:
+        # constructing a Logs leaves it active, so after the construct phase the
+        # active target is the LAST-constructed collection. A bare emit here
+        # would file this file's error under an unrelated sibling — and ship it
+        # inside that sibling's Cloud log payload. ``preserve_active_logs``
+        # stops the throwaway target leaking past this call.
+        # No emoticon here: the console formatter already prefixes ERROR records
+        # with one. Yaml parse errors also already carry their own
+        # ", in <file>[line,column]" suffix, so only name the source when the
+        # message does not locate itself.
+        source_description = getattr(yaml_source, "file_path", None) or getattr(yaml_source, "description", None)
+        message = str(exception) or type(exception).__name__
+        locates_itself = bool(source_description) and str(source_description) in message
+        location = f", in {source_description}" if source_description and not locates_itself else ""
+        with preserve_active_logs():
+            error_logs = Logs()
+            logger.error(f"{message}{location}")
         # Invariant: this placeholder Contract is never uploaded to Soda Cloud.
         # ``build_error_result`` is only invoked when the YAML failed to parse
         # before a real ``Contract`` could be constructed; the result it
@@ -1074,7 +1108,10 @@ def build_error_result(
             measurements=[],
             check_results=[],
             sending_results_to_soda_cloud_failed=False,
-            log_records=None,
+            # Carries the error line captured above, so the placeholder is
+            # self-diagnosing (a FAILED scan in Cloud no longer ships an empty
+            # log payload) instead of leaving ``result.error`` the only record.
+            log_records=error_logs.get_log_records(),
             post_processing_stages=[],
         )
         result.error = exception
@@ -1218,7 +1255,7 @@ def _verify_check_sources_aligned(self, verification_result: "CheckCollectionRes
         # ``result.get_errors()`` with no re-collection.
         for check_result in offending:
             logger.error(
-                f"Source mismatch — check '{check_result.check.full_path}' has "
+                f"Source mismatch — check '{check_result.check.check_path}' has "
                 f"source={check_result.check.source!r} but parent collection "
                 f"declares wire_source={self.wire_source!r}. Skipping Cloud upload to avoid "
                 f"a backend-side source-mismatch failure on the whole batch."

soda-core/src/soda_core/common/soda_cloud.py

@@ -1714,15 +1714,16 @@ def _build_check_result_cloud_dict(
 ) -> dict:
     return {
         "identities": {"vc1": check_result.check.identity},
-        # Wire path: ``Check.full_path`` is the yaml-internal ``path`` for
-        # subtypes that emit byte-identical historical paths, and
-        # ``"{collection_id}.{path}"`` for subtypes that need a prefix so
-        # the backend's ``firstSegmentOf(checkPath)`` can match the
-        # subtype's identifier. Selector matching still uses ``check.path``.
-        # Falls back to ``path`` when ``full_path`` is empty (the
+        # Wire path: ``Check.check_path`` is the yaml-internal ``relative_path``
+        # for subtypes that emit byte-identical historical paths (contracts),
+        # and the option-3 form
+        # ``"{wire_source}.{collection_id}:{relative_path}"`` for subtypes that
+        # need a prefix, so the backend can parse out the subtype's identifier.
+        # Selector matching still uses ``check.relative_path``.
+        # Falls back to ``relative_path`` when ``check_path`` is empty (the
         # dataclass default) — protects external constructors of
-        # ``Check(...)`` that don't set ``full_path`` explicitly.
-        "checkPath": check_result.check.full_path or check_result.check.path,
+        # ``Check(...)`` that don't set ``check_path`` explicitly.
+        "checkPath": check_result.check.check_path or check_result.check.relative_path,
         "name": check_result.check.name,
         "type": "generic",
         "checkType": check_result.check.type,

soda-core/src/soda_core/contracts/contract_verification.py

@@ -306,25 +306,26 @@ class Check:
 
     i.e. "after" CheckImpl has been executed and check result is being created.
 
-    ``path`` is the YAML-internal stripped path (e.g.
+    ``relative_path`` is the YAML-internal stripped path (e.g.
     ``"columns.email.checks.missing"``). It is what selector matching and
     user-facing messages use; it is byte-identical to the value emitted by
     every prior contract verification.
 
-    ``full_path`` is the wire path emitted to Soda Cloud as ``checkPath``.
-    For contracts (``wire_source == "soda-contract"``) it equals ``path``.
-    For non-contract subtypes (data standards, ...) it is prefixed with
-    ``"{collection_id}.{path}"`` so the backend's
-    ``firstSegmentOf(checkPath)`` filter matches the
-    ``DataStandard.name``. Splitting the two means the prefix never leaks
-    into selector matching or error display.
+    ``check_path`` is the wire path emitted to Soda Cloud as ``checkPath``.
+    For contracts (``wire_source == "soda-contract"``) it equals
+    ``relative_path``. For non-contract subtypes (data standards, ...) it is
+    the option-3 form ``"{wire_source}.{collection_id}:{relative_path}"``, which
+    the backend parses (split on the first ':', then the ref on the first '.')
+    to recover the collection id and match it against ``DataStandard.name``.
+    Splitting the two means the prefix never leaks into selector matching or
+    error display.
     """
 
     column_name: Optional[str]
     type: str
     qualifier: Optional[str]
     name: Optional[str]
-    path: str
+    relative_path: str
     identity: str
     definition: str
     contract_file_line: int
@@ -333,13 +334,13 @@ class Check:
     attributes: Optional[dict[str, Any]]
     location: Optional[Location]
     # Wire-path emitted to Soda Cloud as ``checkPath``. For contracts it
-    # equals ``path``; for non-contract subtypes the engine prefixes it
-    # with ``"{collection_id}.{path}"`` so the backend's
-    # ``firstSegmentOf(checkPath)`` filter matches ``DataStandard.name``.
-    # Defaults to ``""`` so external callers that construct ``Check(...)``
-    # without specifying it still work — emission code falls back to
-    # ``path`` when ``full_path`` is empty.
-    full_path: str = ""
+    # equals ``relative_path``; for non-contract subtypes it is the option-3
+    # form ``"{wire_source}.{collection_id}:{relative_path}"``, which the
+    # backend parses to recover the collection id and match
+    # ``DataStandard.name``. Defaults to ``""`` so external callers that
+    # construct ``Check(...)`` without specifying it still work — emission code
+    # falls back to ``relative_path`` when ``check_path`` is empty.
+    check_path: str = ""
     # Per-check source override. ``None`` (the default) means: emit the
     # enclosing ``CheckCollectionImpl.wire_source`` as the wire ``"source"``
     # field. A non-None value is exercised by future extensions that emit

soda-core/src/soda_core/contracts/impl/check_selector.py

@@ -18,7 +18,18 @@ class CheckSelector:
     - Different fields: AND (all groups must match)
     """
 
-    SUPPORTED_FIELDS = {"type", "name", "column", "path", "qualifier"}
+    SUPPORTED_FIELDS = {
+        "type",
+        "name",
+        "column",
+        "path",
+        "relative_path",
+        "check_path",
+        "qualifier",
+        "source",
+        "collection",
+        "standard",
+    }
     ATTRIBUTES_PREFIX = "attributes."
 
     def __init__(self, field: str, value: str, raw: str):
@@ -71,10 +82,10 @@ def parse_all(cls, expressions: Optional[list[str]]) -> list[CheckSelector]:
 
     @classmethod
     def from_check_paths(cls, check_paths: Optional[list[str]]) -> list[CheckSelector]:
-        """Convert --check-paths values to path selectors."""
+        """Convert --check-paths values to full check_path selectors (the wire form the FE sends)."""
         if not check_paths:
             return []
-        return [cls(field="path", value=path, raw=f"path={path}") for path in check_paths]
+        return [cls(field="check_path", value=p, raw=f"check_path={p}") for p in check_paths]
 
     def matches(self, check_impl) -> bool:
         """Returns True if the given CheckImpl matches this selector."""
@@ -99,10 +110,16 @@ def _get_check_value(self, check_impl) -> Optional[str | list[str]]:
             return check_impl.name
         elif self.field == "column":
             return check_impl.column_impl.column_yaml.name if check_impl.column_impl else None
-        elif self.field == "path":
-            return check_impl.path
+        elif self.field in ("path", "relative_path"):
+            return check_impl.relative_path
+        elif self.field == "check_path":
+            return check_impl.check_path
         elif self.field == "qualifier":
             return check_impl.check_yaml.qualifier
+        elif self.field == "source":
+            return check_impl.contract_impl.wire_source
+        elif self.field in ("collection", "standard"):
+            return check_impl.contract_impl.collection_id
         elif self.field.startswith(self.ATTRIBUTES_PREFIX):
             attr_key = self.field[len(self.ATTRIBUTES_PREFIX) :]
             attr_value = check_impl.attributes.get(attr_key)

soda-core/src/soda_core/contracts/impl/contract_verification_impl.py

@@ -1143,7 +1143,7 @@ def column_expression(self) -> Optional[SqlExpressionStr | COLUMN]:
     }
 
     @property
-    def path(self) -> str:
+    def relative_path(self) -> str:
         parts: list[str] = []
 
         column_name = self.column_impl.column_yaml.name if self.column_impl else None
@@ -1161,35 +1161,44 @@ def path(self) -> str:
         return ".".join(parts)
 
     @property
-    def full_path(self) -> str:
+    def check_path(self) -> str:
         """Wire path emitted to Soda Cloud as ``checkPath``.
 
-        For contracts (``wire_source == "soda-contract"``) this is identical
-        to ``self.path`` — byte-identical to today's emission. For
-        non-contract subtypes it is prefixed with
-        ``"{collection_id}.{path}"`` so the backend's
-        ``firstSegmentOf(checkPath)`` filter can match the subtype's
-        identifier.
+        Format (option 3): ``"{wire_source}.{collection_id}:{relative_path}"``
+        using exactly one ``:`` as the delimiter between the ``{type}.{id}``
+        prefix and the ``{relative}`` path.
 
-        Selector matching uses ``self.path`` (not ``full_path``) so the
-        prefix never leaks into ``--check-selector`` matching.
+        - Contracts (``wire_source == "soda-contract"``): bare
+          ``self.relative_path``, byte-identical to today's emission.
+        - Non-contract subtypes (e.g. data standards): the full option-3
+          prefix ``f"{wire_source}.{collection_id}:{relative_path}"``.
+        - Defensive fallback (no ``collection_id``): bare ``self.relative_path``.
+
+        The ``type`` (wire_source) and ``id`` (collection_id) segments must not
+        contain ``.`` or ``:``. ``wire_source`` is delimiter-safe **by
+        convention** (hardcoded subclass literals, never user input); the
+        ``base.py`` guard in ``verify()`` enforces this for ``collection_id``
+        (which does come from user-supplied config).
+
+        Selector matching uses ``self.relative_path`` (not ``check_path``) so
+        the prefix never leaks into ``--check-selector`` matching.
         """
         # ``contract_impl`` is the back-ref to the enclosing
         # ``CheckCollectionImpl`` (name preserved during the rename slice;
         # rename deferred to a follow-up). The wire_source check matches
         # ``ContractImpl.wire_source`` literally so any non-contract
         # subtype automatically opts into prefixing.
         if self.contract_impl.wire_source == "soda-contract":
-            return self.path
+            return self.relative_path
         collection_id: Optional[str] = self.contract_impl.collection_id
         # Non-contract subtypes MUST declare collection_id: the
         # ``CheckCollectionImpl.verify()`` guard raises before this property
         # is reached when collection_id is missing, but we defensively fall
         # back to the bare path so dataclass-build callers can still
         # instantiate a Check during error paths.
         if not collection_id:
-            return self.path
-        return f"{collection_id}.{self.path}"
+            return self.relative_path
+        return f"{self.contract_impl.wire_source}.{collection_id}:{self.relative_path}"
 
     def _get_name_with_default(self, check_yaml: CheckYaml) -> str:
         if isinstance(check_yaml.name, str):
@@ -1222,8 +1231,8 @@ def _build_check_info(self) -> Check:
             type=self.type,
             qualifier=self.check_yaml.qualifier,
             name=self.name,
-            path=self.path,
-            full_path=self.full_path,
+            relative_path=self.relative_path,
+            check_path=self.check_path,
             identity=self.identity,
             definition=self._build_definition(),
             column_name=self.column_impl.column_yaml.name if self.column_impl else None,

soda-tests/tests/unit/test_alignment_guard.py

@@ -68,8 +68,8 @@ def _make_check(*, source):
         type="missing",
         qualifier=None,
         name="No missing values",
-        path="columns.email.checks.missing",
-        full_path="my_pii_standard.columns.email.checks.missing",
+        relative_path="columns.email.checks.missing",
+        check_path="data-standard.my_pii_standard:columns.email.checks.missing",
         identity="abc",
         definition="...",
         contract_file_line=1,

soda-tests/tests/unit/test_check_collections_base.py

@@ -9,6 +9,7 @@
 
 from __future__ import annotations
 
+import logging
 from datetime import datetime, timezone
 from typing import Optional
 
@@ -228,6 +229,35 @@ def test_execute_check_collections_handles_parse_failures():
     assert isinstance(session_result.results[0].error, RuntimeError)
 
 
+def test_execute_check_collections_logs_isolated_parse_failure(caplog):
+    """An isolated parse failure must be LOGGED, not only stored on ``result.error``.
+
+    Regression guard: ``build_error_result`` used to stash the exception on
+    ``result.error`` with ``log_records=None`` and never emit it. The only
+    reader of ``result.error`` is the Soda Cloud ``mark_scan_as_failed``
+    path (gated on ``soda_scan_id``), so a local run — every
+    ``abort_on_first_error=False`` caller, i.e. always for data standards
+    and any multi-file contract run — exited 3 in total silence.
+    """
+    caplog.set_level(logging.ERROR)
+    sources = [_LabelledSource("bad", kind=_RAISING_KIND)]
+    session_result = execute_check_collections(yaml_sources=sources, data_source_impl=None)
+
+    assert session_result.results[0].status is CheckCollectionStatus.ERROR
+    assert "parse failure for bad" in caplog.text
+
+
+def test_execute_check_collections_logs_isolated_verify_failure(caplog):
+    """Same contract for a verify()-time failure under per-item isolation."""
+    caplog.set_level(logging.ERROR)
+    exc = RuntimeError("verify exploded")
+    sources = [_LabelledSource("a", raise_on_verify=exc)]
+    session_result = execute_check_collections(yaml_sources=sources, data_source_impl=None)
+
+    assert session_result.results[0].status is CheckCollectionStatus.ERROR
+    assert "verify exploded" in caplog.text
+
+
 def test_execute_check_collections_records_error_for_unknown_kind():
     """An unknown ``kind:`` produces an ERROR-status placeholder result;
     other items in the batch still run.