[ASIM] Fix replay timestamp and failure handling

.script/tests/asimParsersTest/ingestTimeParser/README.md

@@ -58,9 +58,15 @@ discover target
 ```
 
 Parser validation does not depend on raw-table replay. QTP source replay is
-optional: if its table, DCR, upload, or visibility stage fails, the report
-records a warning and blocks QTP/parity without failing the required parser
-validation stages.
+optional. If its table or DCR cannot be established, the report records a
+warning and blocks QTP/parity without failing the required parser validation
+stages. Once the raw DCR exists, upload or visibility failures fail the target;
+QTP execution and parity also fail the target when they run and fail.
+
+`qtpSourceReplay` persists the latest `observedSchema`, `observedRows`, and
+`observedCountsByTimestamp` from raw-table visibility checks. These remain
+available in `validation-result.json` when visibility detects unexpected,
+duplicate, or incomplete rows.
 
 QTP parity stores the complete `qtp_schema`, `qtp_rows`, `itp_schema`, and
 `itp_rows` under `outputValidation`. It derives `intersect_schema` from shared

.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/client.py

@@ -511,7 +511,10 @@ def wait_for_exact_rows(
                 raise HarnessError(
                     stage,
                     "Exact timestamp query returned unexpected timestamps.",
-                    details={"unexpected": unexpected},
+                    details={
+                        "unexpected": unexpected,
+                        "observedQueryTable": query_table,
+                    },
                 )
             observed_counts = {
                 timestamp: raw_observed_counts.get(timestamp, 0)
@@ -530,7 +533,10 @@ def wait_for_exact_rows(
                     stage,
                     "Exact timestamp slice contains more rows than "
                     "expected for one or more timestamps.",
-                    details={"overages": overages},
+                    details={
+                        "overages": overages,
+                        "observedQueryTable": query_table,
+                    },
                 )
             if observed_counts == expected_counts:
                 if stable_once:
@@ -546,6 +552,7 @@ def wait_for_exact_rows(
                     details={
                         "expectedCounts": expected_counts,
                         "observedCounts": observed_counts,
+                        "observedQueryTable": query_table,
                     },
                 )
             delay = self.poll_delays[

.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/common.py

@@ -17,6 +17,11 @@
 import requests
 import yaml
 
+from asim_ingestion.fixtures.timestamps import (
+    parse_kusto_datetime_ticks,
+    ticks_to_kusto_datetime,
+)
+
 HARNESS_RESULT_VERSION = 1
 SNAPSHOT_VERSION = 1
 
@@ -41,10 +46,6 @@
 
 IDENTIFIER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]*$")
 RESOURCE_NAME_RE = re.compile(r"[^a-z0-9-]+")
-KUSTO_TIMESTAMP_RE = re.compile(
-    r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{7}Z$"
-)
-
 REPLAY_EXCLUDED_COLUMNS = {
     "_BilledSize",
     "_IsBillable",
@@ -178,11 +179,14 @@ def _require_identifier(value: str, field: str) -> str:
 
 
 def _validate_timestamp(value: str) -> str:
-    if not KUSTO_TIMESTAMP_RE.fullmatch(value):
-        raise ValueError(
-            f"Expected exact Kusto timestamp with seven digits: {value}"
+    try:
+        return ticks_to_kusto_datetime(
+            parse_kusto_datetime_ticks(value)
         )
-    return value
+    except (TypeError, ValueError) as exc:
+        raise ValueError(
+            f"Expected a valid Kusto timestamp: {value}"
+        ) from exc
 
 
 def _compact_json_bytes(value: Any) -> bytes:
@@ -410,6 +414,70 @@ def _query_timestamp_counts(
     return dict(counts)
 
 
+def build_observed_query_rows(
+    query_table: Mapping[str, Any],
+    *,
+    stage: str,
+) -> dict[str, Any]:
+    columns = query_table.get("columns")
+    raw_rows = query_table.get("rows")
+    if not isinstance(columns, list) or not isinstance(raw_rows, list):
+        raise HarnessError(
+            stage, "Log Analytics result must contain columns and rows."
+        )
+    schema: list[dict[str, str]] = []
+    column_names: list[str] = []
+    for column in columns:
+        if not isinstance(column, Mapping):
+            raise HarnessError(
+                stage, "Invalid Log Analytics column metadata."
+            )
+        name = column.get("name")
+        column_type = column.get("type")
+        if not isinstance(name, str) or not isinstance(column_type, str):
+            raise HarnessError(
+                stage, "Invalid Log Analytics column metadata."
+            )
+        if name in column_names:
+            raise HarnessError(
+                stage,
+                f"Log Analytics returned duplicate column {name}.",
+            )
+        column_names.append(name)
+        schema.append(
+            {
+                "name": name,
+                "type": _normalize_dcr_type(column_type),
+            }
+        )
+    type_by_name = {
+        column["name"]: column["type"] for column in schema
+    }
+    rows: list[dict[str, Any]] = []
+    for raw_row in raw_rows:
+        if (
+            not isinstance(raw_row, list)
+            or len(raw_row) != len(column_names)
+        ):
+            raise HarnessError(stage, "Invalid Log Analytics row shape.")
+        row = {
+            name: _json_safe_value(value, type_by_name[name])
+            for name, value in zip(column_names, raw_row)
+        }
+        timestamp = row.get("TimeGenerated")
+        if isinstance(timestamp, str):
+            row["TimeGenerated"] = _validate_timestamp(timestamp)
+        rows.append(row)
+    return {
+        "schema": schema,
+        "rows": rows,
+        "countsByTimestamp": _query_timestamp_counts(
+            query_table,
+            stage,
+        ),
+    }
+
+
 def build_upload_batches(
     rows: Sequence[Mapping[str, Any]],
     max_bytes: int = 1_000_000,

.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/azure/runner.py

@@ -14,21 +14,62 @@
     _validate_timestamp,
     _write_json_no_clobber,
     build_fixture_snapshot,
+    build_observed_query_rows,
     build_resource_names,
     count_rows_by_timestamp,
 )
 
 
 def _error_payload(exc: HarnessError) -> dict[str, Any]:
+    details = exc.details
+    if isinstance(details, Mapping) and "observedQueryTable" in details:
+        details = dict(details)
+        details.pop("observedQueryTable", None)
     return {
         "type": type(exc).__name__,
         "stage": exc.stage,
         "message": str(exc),
         "statusCode": exc.status_code,
-        "details": exc.details,
+        "details": details,
     }
 
 
+def _record_qtp_observation(
+    qtp_replay: dict[str, Any],
+    query_table: Any,
+) -> None:
+    if not isinstance(query_table, Mapping):
+        return
+    observed = build_observed_query_rows(
+        query_table,
+        stage="raw-visibility",
+    )
+    qtp_replay["observedSchema"] = observed["schema"]
+    qtp_replay["observedRows"] = observed["rows"]
+    qtp_replay["observedCountsByTimestamp"] = observed[
+        "countsByTimestamp"
+    ]
+
+
+def _block_optional_parity(
+    output_validation: Mapping[str, Any],
+    *,
+    detail: str,
+    qtp_replay: Mapping[str, Any],
+) -> dict[str, Any]:
+    blocked = dict(output_validation)
+    blocked["queryTimeExecution"] = {
+        "status": "blocked",
+        "detail": detail,
+    }
+    blocked["parity"] = {
+        "status": "blocked",
+        "detail": detail,
+    }
+    blocked["qtpSourceReplay"] = qtp_replay
+    return blocked
+
+
 def _output_failure(
     output_validation: Mapping[str, Any],
 ) -> HarnessError | None:
@@ -143,6 +184,9 @@ def run(
                 ),
                 "inputStream": input_stream or None,
                 "error": None,
+                "observedSchema": [],
+                "observedRows": [],
+                "observedCountsByTimestamp": {},
             },
             "cleanup": cleanup,
             "telemetry": {
@@ -287,28 +331,60 @@ def run(
                             transform_kql="source",
                             output_stream=input_stream,
                         )
-                    result["rawDcr"] = raw_handle.to_dict()
-                    raw_counts = count_rows_by_timestamp(
-                        rows
-                    )
-                    with self._stage("raw-upload"):
-                        result["rawUpload"] = self.upload_rows(
-                            raw_handle,
-                            rows,
-                            reconcile_table=source_table,
-                            expected_counts_by_timestamp=raw_counts,
-                        )
-                    with self._stage("raw-visibility"):
-                        self.wait_for_exact_rows(
-                            source_table,
-                            timestamps,
-                            raw_counts,
-                            stage="raw-visibility",
-                        )
-                    qtp_replay["status"] = "passed"
                 except HarnessError as exc:
                     qtp_replay["status"] = "warning"
                     qtp_replay["error"] = _error_payload(exc)
+                else:
+                    result["rawDcr"] = raw_handle.to_dict()
+                    raw_counts = count_rows_by_timestamp(rows)
+                    try:
+                        with self._stage("raw-upload"):
+                            result["rawUpload"] = self.upload_rows(
+                                raw_handle,
+                                rows,
+                                reconcile_table=source_table,
+                                expected_counts_by_timestamp=raw_counts,
+                            )
+                        with self._stage("raw-visibility"):
+                            observed_query_table = (
+                                self.wait_for_exact_rows(
+                                    source_table,
+                                    timestamps,
+                                    raw_counts,
+                                    stage="raw-visibility",
+                                )
+                            )
+                        _record_qtp_observation(
+                            qtp_replay,
+                            observed_query_table,
+                        )
+                        qtp_replay["status"] = "passed"
+                    except HarnessError as exc:
+                        details = exc.details
+                        if isinstance(details, Mapping):
+                            _record_qtp_observation(
+                                qtp_replay,
+                                details.get("observedQueryTable"),
+                            )
+                            cleaned_details = dict(details)
+                            cleaned_details.pop(
+                                "observedQueryTable",
+                                None,
+                            )
+                            exc.details = cleaned_details
+                        qtp_replay["status"] = "failed"
+                        qtp_replay["error"] = _error_payload(exc)
+                        result["outputValidation"] = (
+                            _block_optional_parity(
+                                result["outputValidation"],
+                                detail=(
+                                    "Blocked by QTP raw ingestion "
+                                    "failure."
+                                ),
+                                qtp_replay=qtp_replay,
+                            )
+                        )
+                        raise
 
             if (
                 validation_runner is not None
@@ -317,22 +393,13 @@ def run(
                 with self._stage("output-validation"):
                     qtp_replay = result["qtpSourceReplay"]
                     if qtp_replay.get("status") == "warning":
-                        output_validation = dict(
-                            result["outputValidation"]
-                        )
-                        blocked_detail = (
-                            "Skipped because QTP source replay failed."
-                        )
-                        output_validation["queryTimeExecution"] = {
-                            "status": "blocked",
-                            "detail": blocked_detail,
-                        }
-                        output_validation["parity"] = {
-                            "status": "blocked",
-                            "detail": blocked_detail,
-                        }
-                        output_validation["qtpSourceReplay"] = (
-                            qtp_replay
+                        output_validation = _block_optional_parity(
+                            result["outputValidation"],
+                            detail=(
+                                "Skipped because QTP raw DCR "
+                                "creation failed."
+                            ),
+                            qtp_replay=qtp_replay,
                         )
                     else:
                         parity_validation = dict(

.script/tests/asimParsersTest/ingestTimeParser/asim_ingestion/runtime/queries.py

@@ -9,7 +9,10 @@
 
 import yaml
 
-from asim_ingestion.fixtures.timestamps import parse_kusto_datetime_ticks
+from asim_ingestion.fixtures.timestamps import (
+    parse_kusto_datetime_ticks,
+    ticks_to_kusto_datetime,
+)
 from asim_ingestion.parser.kql import split_kql_statements, split_pipeline
 
 MAX_MISMATCH_EXAMPLES = 10
@@ -253,11 +256,20 @@ def _parse_filter_counts(
     timestamps: Sequence[str],
 ) -> dict[str, int]:
     _, rows = _normalize_query_table(query_table, "filter")
-    expected_timestamps = set(timestamps)
+    expected_timestamps = {
+        ticks_to_kusto_datetime(
+            parse_kusto_datetime_ticks(str(timestamp))
+        )
+        for timestamp in timestamps
+    }
     counts = {timestamp: 0 for timestamp in expected_timestamps}
     for row in rows:
         timestamp = row.get("TimeGenerated")
         count = row.get("_AsimSelectedCount")
+        if isinstance(timestamp, str):
+            timestamp = ticks_to_kusto_datetime(
+                parse_kusto_datetime_ticks(timestamp)
+            )
         if not isinstance(timestamp, str) or timestamp not in counts:
             raise ValidationError(
                 "filter",