Remove unused validation methods from telemetry_adapter

Removed several validation methods and their associated logic from telemetry_adapter.py, streamlining the telemetry input processing.

Author: BryceWDesign

src/ix_autonomy_assurance_case_runtime/telemetry_adapter.py

@@ -384,390 +384,3 @@ def normalize(self, telemetry_input: TelemetryNormalizationInput) -> TelemetryAd
             captured_at_utc=telemetry_input.captured_at_utc,
             received_at_utc=telemetry_input.received_at_utc,
             payload=telemetry_input.payload,
-            freshness_status=freshness_status,
-            quality_flags=quality_flags,
-            replay_record_id=telemetry_input.replay_record_id,
-        )
-        decision = (
-            TelemetryAdapterDecision.ACCEPTED
-            if envelope.can_support_runtime_evaluation()
-            else TelemetryAdapterDecision.DEGRADED
-        )
-        return TelemetryAdapterReport(
-            input_id=telemetry_input.input_id,
-            decision=decision,
-            findings=findings,
-            envelope=envelope,
-        )
-
-    def _validate_source_and_schema(
-        self,
-        telemetry_input: TelemetryNormalizationInput,
-        source: TelemetrySource | None,
-        schema: TelemetrySchema | None,
-    ) -> tuple[TelemetryAdapterFinding, ...]:
-        """Validate source and schema catalog references."""
-
-        findings: list[TelemetryAdapterFinding] = []
-        if source is None:
-            findings.append(
-                TelemetryAdapterFinding(
-                    finding_id=f"input-{telemetry_input.input_id}-missing-source",
-                    severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                    message="Telemetry input references a missing source.",
-                    source_id=telemetry_input.source_id,
-                )
-            )
-        if schema is None:
-            findings.append(
-                TelemetryAdapterFinding(
-                    finding_id=f"input-{telemetry_input.input_id}-missing-schema",
-                    severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                    message="Telemetry input references a missing schema.",
-                    schema_id=telemetry_input.schema_id,
-                )
-            )
-        if source is not None and schema is not None and not source.supports_schema(
-            schema.schema_id
-        ):
-            findings.append(
-                TelemetryAdapterFinding(
-                    finding_id=f"source-{source.source_id}-does-not-support-{schema.schema_id}",
-                    severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                    message="Telemetry source is not allowed to emit the requested schema.",
-                    source_id=source.source_id,
-                    schema_id=schema.schema_id,
-                    quality_flag=TelemetryQualityFlag.SCHEMA_MISMATCH,
-                )
-            )
-        return tuple(findings)
-
-    def _validate_replay_boundary(
-        self,
-        telemetry_input: TelemetryNormalizationInput,
-        source: TelemetrySource,
-    ) -> tuple[tuple[TelemetryAdapterFinding, ...], tuple[TelemetryQualityFlag, ...]]:
-        """Validate replay metadata and return findings plus quality flags."""
-
-        findings: list[TelemetryAdapterFinding] = []
-        flags: list[TelemetryQualityFlag] = []
-        needs_replay_record = (
-            source.kind is TelemetrySourceKind.LOG_REPLAY
-            or telemetry_input.replay_record_id is not None
-        )
-        if source.kind is TelemetrySourceKind.LOG_REPLAY:
-            flags.append(TelemetryQualityFlag.REPLAY_ONLY)
-        if not needs_replay_record:
-            return (), ()
-
-        if telemetry_input.replay_record_id is None:
-            findings.append(
-                TelemetryAdapterFinding(
-                    finding_id=f"source-{source.source_id}-missing-replay-record",
-                    severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                    message="Replay telemetry source requires a replay record ID.",
-                    source_id=source.source_id,
-                )
-            )
-            return tuple(findings), tuple(flags)
-
-        replay_record = self._catalog.replay_record_by_id(telemetry_input.replay_record_id)
-        if replay_record is None:
-            severity = (
-                TelemetryAdapterFindingSeverity.BLOCKER
-                if self._policy.require_known_replay_record
-                else TelemetryAdapterFindingSeverity.WARNING
-            )
-            findings.append(
-                TelemetryAdapterFinding(
-                    finding_id=f"replay-{telemetry_input.replay_record_id}-missing",
-                    severity=severity,
-                    message="Telemetry input references a replay record not present in catalog.",
-                    source_id=source.source_id,
-                    replay_record_id=telemetry_input.replay_record_id,
-                    quality_flag=TelemetryQualityFlag.REPLAY_ONLY,
-                )
-            )
-            flags.append(TelemetryQualityFlag.REPLAY_ONLY)
-            return tuple(findings), tuple(flags)
-
-        flags.append(TelemetryQualityFlag.REPLAY_ONLY)
-        if replay_record.source_id != telemetry_input.source_id:
-            findings.append(
-                TelemetryAdapterFinding(
-                    finding_id=f"replay-{replay_record.replay_record_id}-source-mismatch",
-                    severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                    message="Replay record source does not match telemetry input source.",
-                    source_id=telemetry_input.source_id,
-                    replay_record_id=replay_record.replay_record_id,
-                )
-            )
-        if replay_record.schema_id != telemetry_input.schema_id:
-            findings.append(
-                TelemetryAdapterFinding(
-                    finding_id=f"replay-{replay_record.replay_record_id}-schema-mismatch",
-                    severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                    message="Replay record schema does not match telemetry input schema.",
-                    schema_id=telemetry_input.schema_id,
-                    replay_record_id=replay_record.replay_record_id,
-                    quality_flag=TelemetryQualityFlag.SCHEMA_MISMATCH,
-                )
-            )
-
-        return tuple(findings), tuple(flags)
-
-    def _validate_timestamp_posture(
-        self,
-        telemetry_input: TelemetryNormalizationInput,
-    ) -> tuple[
-        tuple[TelemetryAdapterFinding, ...],
-        TelemetryFreshnessStatus,
-        tuple[TelemetryQualityFlag, ...],
-    ]:
-        """Validate telemetry capture and receive timestamp posture."""
-
-        findings: list[TelemetryAdapterFinding] = []
-        flags: list[TelemetryQualityFlag] = []
-        captured_at = telemetry_input.captured_at
-        received_at = telemetry_input.received_at
-
-        if captured_at > received_at:
-            findings.append(
-                TelemetryAdapterFinding(
-                    finding_id=f"input-{telemetry_input.input_id}-future-captured",
-                    severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                    message="Telemetry captured_at_utc is after received_at_utc.",
-                    quality_flag=TelemetryQualityFlag.FUTURE_TIMESTAMP,
-                )
-            )
-            return (
-                tuple(findings),
-                TelemetryFreshnessStatus.FUTURE_DATED,
-                (TelemetryQualityFlag.FUTURE_TIMESTAMP,),
-            )
-
-        latency_seconds = int((received_at - captured_at).total_seconds())
-        if latency_seconds > self._policy.max_latency_seconds:
-            findings.append(
-                TelemetryAdapterFinding(
-                    finding_id=f"input-{telemetry_input.input_id}-stale",
-                    severity=TelemetryAdapterFindingSeverity.WARNING,
-                    message=(
-                        "Telemetry latency exceeds adapter freshness policy and cannot "
-                        "support acceptance-oriented evaluation without degradation."
-                    ),
-                    quality_flag=TelemetryQualityFlag.STALE_TIMESTAMP,
-                )
-            )
-            flags.append(TelemetryQualityFlag.STALE_TIMESTAMP)
-            return tuple(findings), TelemetryFreshnessStatus.STALE, tuple(flags)
-
-        return tuple(findings), TelemetryFreshnessStatus.CURRENT, tuple(flags)
-
-    def _validate_payload_schema(
-        self,
-        telemetry_input: TelemetryNormalizationInput,
-        schema: TelemetrySchema,
-    ) -> tuple[tuple[TelemetryAdapterFinding, ...], tuple[TelemetryQualityFlag, ...]]:
-        """Validate telemetry payload against schema fields."""
-
-        findings: list[TelemetryAdapterFinding] = []
-        flags: list[TelemetryQualityFlag] = []
-        for required_field_name in schema.required_field_names():
-            if required_field_name not in telemetry_input.payload:
-                findings.append(
-                    TelemetryAdapterFinding(
-                        finding_id=(
-                            f"input-{telemetry_input.input_id}-missing-field-"
-                            f"{required_field_name}"
-                        ),
-                        severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                        message="Telemetry payload is missing a required schema field.",
-                        schema_id=schema.schema_id,
-                        field_name=required_field_name,
-                        quality_flag=TelemetryQualityFlag.MISSING_FIELD,
-                    )
-                )
-                flags.append(TelemetryQualityFlag.MISSING_FIELD)
-
-        schema_field_names = {field.field_name for field in schema.fields}
-        for field_name, value in telemetry_input.payload.items():
-            field = schema.field_by_name(field_name)
-            if field is None:
-                findings.append(
-                    TelemetryAdapterFinding(
-                        finding_id=f"input-{telemetry_input.input_id}-unexpected-field-{field_name}",
-                        severity=TelemetryAdapterFindingSeverity.WARNING,
-                        message="Telemetry payload includes a field not declared by the schema.",
-                        schema_id=schema.schema_id,
-                        field_name=field_name,
-                    )
-                )
-                continue
-            field_findings, field_flags = _validate_field_value(
-                telemetry_input=telemetry_input,
-                schema=schema,
-                field=field,
-                value=value,
-            )
-            findings.extend(field_findings)
-            flags.extend(field_flags)
-
-        if not schema_field_names:
-            findings.append(
-                TelemetryAdapterFinding(
-                    finding_id=f"schema-{schema.schema_id}-has-no-fields",
-                    severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                    message="Telemetry schema has no fields.",
-                    schema_id=schema.schema_id,
-                    quality_flag=TelemetryQualityFlag.SCHEMA_MISMATCH,
-                )
-            )
-            flags.append(TelemetryQualityFlag.SCHEMA_MISMATCH)
-
-        return tuple(findings), tuple(flags)
-
-    def _validate_source_trust(
-        self,
-        source: TelemetrySource,
-    ) -> tuple[tuple[TelemetryAdapterFinding, ...], tuple[TelemetryQualityFlag, ...]]:
-        """Validate telemetry source trust posture."""
-
-        if source.can_support_acceptance() or (
-            source.kind is TelemetrySourceKind.LOG_REPLAY
-            and source.trust_level.rank >= TelemetryTrustLevel.MODERATE.rank
-        ):
-            return (), ()
-
-        severity = (
-            TelemetryAdapterFindingSeverity.BLOCKER
-            if self._policy.reject_untrusted_sources
-            else TelemetryAdapterFindingSeverity.WARNING
-        )
-        return (
-            (
-                TelemetryAdapterFinding(
-                    finding_id=f"source-{source.source_id}-trust-{source.trust_level.value}",
-                    severity=severity,
-                    message=(
-                        "Telemetry source trust level cannot support acceptance-oriented "
-                        "evaluation without degradation."
-                    ),
-                    source_id=source.source_id,
-                    quality_flag=TelemetryQualityFlag.SOURCE_UNTRUSTED,
-                ),
-            ),
-            (TelemetryQualityFlag.SOURCE_UNTRUSTED,),
-        )
-
-
-def _validate_field_value(
-    telemetry_input: TelemetryNormalizationInput,
-    schema: TelemetrySchema,
-    field: TelemetrySchemaField,
-    value: Any,
-) -> tuple[tuple[TelemetryAdapterFinding, ...], tuple[TelemetryQualityFlag, ...]]:
-    """Validate one payload field value against its schema field definition."""
-
-    findings: list[TelemetryAdapterFinding] = []
-    flags: list[TelemetryQualityFlag] = []
-    if not _matches_field_type(field, value):
-        findings.append(
-            TelemetryAdapterFinding(
-                finding_id=f"input-{telemetry_input.input_id}-field-{field.field_name}-type",
-                severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                message=(
-                    f"Telemetry field {field.field_name!r} does not match expected type "
-                    f"{field.field_type.value!r}."
-                ),
-                schema_id=schema.schema_id,
-                field_name=field.field_name,
-                quality_flag=TelemetryQualityFlag.SCHEMA_MISMATCH,
-            )
-        )
-        flags.append(TelemetryQualityFlag.SCHEMA_MISMATCH)
-        return tuple(findings), tuple(flags)
-
-    if field.field_type is TelemetryFieldType.ENUM and value not in field.allowed_values:
-        findings.append(
-            TelemetryAdapterFinding(
-                finding_id=f"input-{telemetry_input.input_id}-field-{field.field_name}-enum",
-                severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                message=f"Telemetry enum field {field.field_name!r} has unsupported value.",
-                schema_id=schema.schema_id,
-                field_name=field.field_name,
-                quality_flag=TelemetryQualityFlag.SCHEMA_MISMATCH,
-            )
-        )
-        flags.append(TelemetryQualityFlag.SCHEMA_MISMATCH)
-
-    if field.field_type.is_numeric():
-        numeric_value = float(value)
-        if field.minimum_value is not None and numeric_value < field.minimum_value:
-            findings.append(
-                TelemetryAdapterFinding(
-                    finding_id=f"input-{telemetry_input.input_id}-field-{field.field_name}-below-min",
-                    severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                    message=f"Telemetry numeric field {field.field_name!r} is below minimum.",
-                    schema_id=schema.schema_id,
-                    field_name=field.field_name,
-                    quality_flag=TelemetryQualityFlag.OUT_OF_RANGE,
-                )
-            )
-            flags.append(TelemetryQualityFlag.OUT_OF_RANGE)
-        if field.maximum_value is not None and numeric_value > field.maximum_value:
-            findings.append(
-                TelemetryAdapterFinding(
-                    finding_id=f"input-{telemetry_input.input_id}-field-{field.field_name}-above-max",
-                    severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                    message=f"Telemetry numeric field {field.field_name!r} is above maximum.",
-                    schema_id=schema.schema_id,
-                    field_name=field.field_name,
-                    quality_flag=TelemetryQualityFlag.OUT_OF_RANGE,
-                )
-            )
-            flags.append(TelemetryQualityFlag.OUT_OF_RANGE)
-
-    if field.field_type is TelemetryFieldType.TIMESTAMP:
-        try:
-            _parse_utc_timestamp(str(value), f"telemetry field {field.field_name!r}")
-        except ContractValueError:
-            findings.append(
-                TelemetryAdapterFinding(
-                    finding_id=f"input-{telemetry_input.input_id}-field-{field.field_name}-timestamp",
-                    severity=TelemetryAdapterFindingSeverity.BLOCKER,
-                    message=f"Telemetry timestamp field {field.field_name!r} is invalid.",
-                    schema_id=schema.schema_id,
-                    field_name=field.field_name,
-                    quality_flag=TelemetryQualityFlag.SCHEMA_MISMATCH,
-                )
-            )
-            flags.append(TelemetryQualityFlag.SCHEMA_MISMATCH)
-
-    return tuple(findings), tuple(flags)
-
-
-def _matches_field_type(field: TelemetrySchemaField, value: Any) -> bool:
-    """Return whether a payload value matches the expected telemetry field type."""
-
-    if field.field_type is TelemetryFieldType.STRING:
-        return isinstance(value, str)
-    if field.field_type is TelemetryFieldType.INTEGER:
-        return isinstance(value, int) and not isinstance(value, bool)
-    if field.field_type is TelemetryFieldType.FLOAT:
-        return isinstance(value, (int, float)) and not isinstance(value, bool)
-    if field.field_type is TelemetryFieldType.BOOLEAN:
-        return isinstance(value, bool)
-    if field.field_type is TelemetryFieldType.ENUM:
-        return isinstance(value, str)
-    if field.field_type is TelemetryFieldType.TIMESTAMP:
-        return isinstance(value, str)
-    return False
-
-
-def _dedupe_quality_flags(
-    quality_flags: tuple[TelemetryQualityFlag, ...],
-) -> tuple[TelemetryQualityFlag, ...]:
-    """Return quality flags in first-seen order without duplicates."""
-
-    return tuple(dict.fromkeys(quality_flags))