uploader: clarify runtime data flow

Distinguish HTTP outcomes from file results, give resource and telemetry state domain-specific names, and replace string-based report aggregation with explicit fields. Remove stale rule commentary while preserving runtime and report contracts.

Validated with the 449-test Python suite, all 199 //tools tests, compileall, template lint, buildifier, and git diff checks.

Author: tonyredondo

tools/core/test_optimization_uploader.bzl

@@ -32,22 +32,10 @@ Why generated launchers/configuration:
 Developer navigation:
 - Starlark test helpers (manifest/CODEOWNERS parsing parity) near the top.
 - `_uploader_impl` builds both script templates and wires rule outputs.
-- Script templates contain the runtime behavior for discovery, enrichment,
-  uploads, retries, and locking.
+- `uploader_py` owns current runtime behavior; legacy generated scripts remain
+  here only as a rollout fallback.
 """
 
-# Usage pattern:
-#   bazel test //... || test_status=$?; test_status=${test_status:-0}; DD_API_KEY="$DD_API_KEY" DD_SITE="$DD_SITE" bazel run //:dd_upload_payloads; exit $test_status
-#
-# Key features:
-# - Discovers all test.outputs/ directories in bazel-testlogs automatically
-# - Supports sharded tests (shard_N_of_M/) and retries (run_N_of_M/)
-# - Uploads test payloads to CI Test Cycle intake
-# - Uploads coverage payloads to Code Coverage intake
-# - Deletes payloads after successful upload (unless DD_TEST_OPTIMIZATION_KEEP_PAYLOADS=1)
-# - Uses workspace-level lock to prevent concurrent uploaders
-# - Enriches payloads with context.json metadata
-
 load(
     "//tools/core:common_utils.bzl",
     "RULES_VERSION",

tools/core/uploader_py/file_worker.py

@@ -339,22 +339,23 @@ def _process_test(
             f"uploading test chunk {chunk.index}/{len(requests)} "
             f"({chunk.size_bytes} bytes)",
         )
-        result = transport.post_json(
+        http_result = transport.post_json(
             runtime.endpoints.test_url,
             headers,
             request.body,
             content_encoding=request.content_encoding,
         )
-        requests_attempted += result.attempts
-        retries += result.retries
+        requests_attempted += http_result.attempts
+        retries += http_result.retries
         _debug(
             runtime,
             task,
-            f"test chunk {chunk.index} completed after {result.attempts} attempt(s)",
+            f"test chunk {chunk.index} completed after "
+            f"{http_result.attempts} attempt(s)",
         )
-        if not result.succeeded:
+        if not http_result.succeeded:
             failure_code, failure_message = _http_failure(
-                result,
+                http_result,
                 payload_type=task.payload_type,
                 payload_limit_context=(
                     f"chunk={chunk.index}/{len(requests)} "
@@ -407,16 +408,18 @@ def _enrich_and_validate_test(
         warnings.append(sidecar_warning)
     repo_key = payload_repo_key(bazel_metadata)
     context_selection = runtime.context_plan.select(repo_key)
+    if sidecar_warning:
+        sidecar_state = "invalid"
+    elif bazel_metadata is not None:
+        sidecar_state = "loaded"
+    else:
+        sidecar_state = "absent"
     _debug(
         runtime,
         task,
         "Bazel sidecar=%s context_repo=%s context_selected=%s context_warning=%s"
         % (
-            "invalid"
-            if sidecar_warning
-            else "loaded"
-            if bazel_metadata is not None
-            else "absent",
+            sidecar_state,
             repr(repo_key[:256]) if repo_key is not None else "none",
             "yes" if context_selection.values is not None else "no",
             context_selection.warning_code or "none",
@@ -621,26 +624,26 @@ def _process_coverage(
         )
         return FileResult(status=FileStatus.SUCCEEDED, **coverage_result_fields)
 
-    result = transport.post_prepared_multipart(
+    http_result = transport.post_prepared_multipart(
         runtime.endpoints.coverage_url,
         headers,
         prepared,
     )
     _debug(
         runtime,
         task,
-        f"coverage request completed after {result.attempts} attempt(s)",
+        f"coverage request completed after {http_result.attempts} attempt(s)",
     )
-    if not result.succeeded:
+    if not http_result.succeeded:
         failure_code, failure_message = _http_failure(
-            result,
+            http_result,
             payload_type=task.payload_type,
         )
         return FileResult(
             status=FileStatus.FAILED,
-            requests_attempted=result.attempts,
+            requests_attempted=http_result.attempts,
             requests_failed=1,
-            retries=result.retries,
+            retries=http_result.retries,
             failure_code=failure_code,
             failure_message=failure_message,
             **coverage_result_fields,
@@ -650,9 +653,9 @@ def _process_coverage(
     _debug(runtime, task, f"coverage cleanup completed source_deleted={deleted}")
     return FileResult(
         status=FileStatus.SUCCEEDED,
-        requests_attempted=result.attempts,
+        requests_attempted=http_result.attempts,
         requests_succeeded=1,
-        retries=result.retries,
+        retries=http_result.retries,
         source_deleted=deleted,
         warning_codes=(cleanup_warning,) if cleanup_warning else (),
         **coverage_result_fields,
@@ -702,7 +705,9 @@ def _process_telemetry(
             failure_message=metadata_failure,
         )
 
-    requests = [_TelemetryRequest(source_body, _telemetry_headers(runtime, payload))]
+    requests: list[_TelemetryRequest] = [
+        _TelemetryRequest(source_body, _telemetry_headers(runtime, payload))
+    ]
     if directive.create_synthetic:
         synthetic = _build_synthetic_telemetry(payload, directive, warnings)
         if synthetic is not None:
@@ -758,21 +763,22 @@ def _process_telemetry(
     requests_succeeded = 0
     retries = 0
     for request in requests:
-        result = transport.post_json(
+        http_result = transport.post_json(
             runtime.endpoints.telemetry_url,
             request.headers,
             request.body,
         )
-        requests_attempted += result.attempts
-        retries += result.retries
+        requests_attempted += http_result.attempts
+        retries += http_result.retries
         _debug(
             runtime,
             task,
-            f"telemetry request completed after {result.attempts} attempt(s)",
+            "telemetry request completed after "
+            f"{http_result.attempts} attempt(s)",
         )
-        if not result.succeeded:
+        if not http_result.succeeded:
             failure_code, failure_message = _http_failure(
-                result,
+                http_result,
                 payload_type=task.payload_type,
             )
             return FileResult(
@@ -1047,12 +1053,12 @@ def _cleanup_source(path: Path, keep_payloads: bool) -> tuple[bool, str | None]:
 
 
 def _http_failure(
-    result: HttpResult,
+    http_result: HttpResult,
     *,
     payload_type: PayloadType,
     payload_limit_context: str | None = None,
 ) -> tuple[str, str]:
-    if result.status_code == 413:
+    if http_result.status_code == 413:
         if payload_type is PayloadType.TEST and payload_limit_context:
             return (
                 "payload_limit_contract_mismatch",
@@ -1063,9 +1069,9 @@ def _http_failure(
             f"HTTP 413 for unsplit {payload_type.value} payload; "
             f"{payload_type.value} splitting is not supported",
         )
-    if result.status_code is not None:
-        return "upload_http_error", f"HTTP {result.status_code}"
-    return "upload_transport_error", result.transport_error or "transport error"
+    if http_result.status_code is not None:
+        return "upload_http_error", f"HTTP {http_result.status_code}"
+    return "upload_transport_error", http_result.transport_error or "transport error"
 
 
 def _event_count(payload: Mapping[str, Any]) -> int:

tools/core/uploader_py/freshness.py

@@ -28,7 +28,7 @@
 from .config import UploaderConfig
 from .discovery import DiscoveryResult, ScanRoot
 from .json_utils import strict_json_loads
-from .models import FileResult, FileStatus, PayloadType
+from .models import FileResult, FileStatus, FileTask, PayloadType
 
 
 class FreshnessError(RuntimeError):
@@ -251,7 +251,7 @@ def filter_discovery_for_freshness(
     selected_outputs = []
     task_label_by_output: dict[str, str] = {}
     skipped: list[str] = []
-    tasks_by_output: dict[str, list[object]] = {}
+    tasks_by_output: dict[str, list[FileTask]] = {}
     for task in discovery.tasks:
         tasks_by_output.setdefault(task.output_key or "", []).append(task)
 

tools/core/uploader_py/reporting.py

@@ -182,20 +182,20 @@ def statistics(self) -> dict[str, Any]:
                 "source_files_split": sum(
                     int(result.chunks_created > 1) for result in self.results
                 ),
-                "chunks_created": _sum(self.results, "chunks_created"),
-                "chunks_uploaded": _sum(self.results, "chunks_uploaded"),
-                "chunks_failed": _sum(self.results, "chunks_failed"),
+                "chunks_created": sum(result.chunks_created for result in self.results),
+                "chunks_uploaded": sum(result.chunks_uploaded for result in self.results),
+                "chunks_failed": sum(result.chunks_failed for result in self.results),
                 "oversized_single_events": sum(
                     int(result.failure_code == "single_event_exceeds_payload_limit")
                     for result in self.results
                 ),
             },
             "requests": {
-                "planned": _sum(self.results, "requests_planned"),
-                "attempted": _sum(self.results, "requests_attempted"),
-                "succeeded": _sum(self.results, "requests_succeeded"),
-                "failed": _sum(self.results, "requests_failed"),
-                "retries": _sum(self.results, "retries"),
+                "planned": sum(result.requests_planned for result in self.results),
+                "attempted": sum(result.requests_attempted for result in self.results),
+                "succeeded": sum(result.requests_succeeded for result in self.results),
+                "failed": sum(result.requests_failed for result in self.results),
+                "retries": sum(result.retries for result in self.results),
             },
             "warnings": dict(sorted(warning_codes.items())),
             "failures": dict(sorted(failure_codes.items())),
@@ -211,8 +211,8 @@ def human_lines(self) -> tuple[str, ...]:
         concurrency = stats["concurrency"]
 
         def type_value(name: str) -> str:
-            values = payload_types[name]
-            return f"{values['succeeded']}/{values['failed']}/{values['skipped']}"
+            counts = payload_types[name]
+            return f"{counts['succeeded']}/{counts['failed']}/{counts['skipped']}"
 
         return (
             (
@@ -438,7 +438,7 @@ def _legacy_payload_type_counts(
             if result.status is FileStatus.SUCCEEDED
         )
     else:
-        telemetry_processed = _sum(telemetry, "requests_succeeded")
+        telemetry_processed = sum(result.requests_succeeded for result in telemetry)
     telemetry_failed = sum(
         (
             result.requests_failed
@@ -501,16 +501,14 @@ def _result_reason(
 def _payload_type_counts(
     results: tuple[FileResult, ...], payload_type: PayloadType
 ) -> dict[str, int]:
-    selected = tuple(result for result in results if result.payload_type is payload_type)
+    type_results = tuple(
+        result for result in results if result.payload_type is payload_type
+    )
     return {
-        status.value: _status_count(selected, status)
+        status.value: _status_count(type_results, status)
         for status in FileStatus
     }
 
 
 def _status_count(results: Iterable[FileResult], status: FileStatus) -> int:
     return sum(int(result.status is status) for result in results)
-
-
-def _sum(results: Iterable[FileResult], field_name: str) -> int:
-    return sum(int(getattr(result, field_name)) for result in results)

tools/core/uploader_py/resources.py

@@ -58,44 +58,46 @@ def load_resources(
     """Resolve contexts, telemetry facts, and schema once with warning fallback."""
     warnings: list[str] = []
     override_path = _existing_file(inputs.context_override)
-    override_context = _load_json_object(override_path) if override_path else None
-    if inputs.context_override is not None and override_context is None:
+    override_values = _load_json_object(override_path) if override_path else None
+    if inputs.context_override is not None and override_values is None:
         warnings.append("context_override_invalid")
 
     context_records: list[ContextRecord] = []
-    primary_path: Path | None = None
-    runtime_override_enabled = override_context is not None
-    if override_context is not None:
-        context_records.append(ContextRecord.create("__runtime_override__", override_context))
-        primary_path = override_path
+    primary_context_path: Path | None = None
+    runtime_override_enabled = override_values is not None
+    if override_values is not None:
+        context_records.append(
+            ContextRecord.create("__runtime_override__", override_values)
+        )
+        primary_context_path = override_path
     else:
-        manifest = _resolve_optional(resolver, inputs.context_manifest_paths)
-        if inputs.context_manifest_paths and manifest is None:
+        context_manifest = _resolve_optional(resolver, inputs.context_manifest_paths)
+        if inputs.context_manifest_paths and context_manifest is None:
             warnings.append("context_manifest_unresolved")
-        if manifest is not None:
+        if context_manifest is not None:
             seen_repo_keys: set[str] = set()
             for repo_key, short_path, artifact_path in _context_manifest_entries(
-                manifest,
+                context_manifest,
                 warnings,
             ):
                 normalized_repo_key = repo_key.rsplit("+", 1)[-1]
                 if normalized_repo_key in seen_repo_keys:
                     warnings.append("context_manifest_duplicate_repo")
                     continue
-                resolved = _resolve_optional(
+                context_path = _resolve_optional(
                     resolver,
                     (artifact_path, short_path),
                 )
-                context = _load_json_object(resolved) if resolved else None
-                if context is None:
+                context_values = _load_json_object(context_path) if context_path else None
+                if context_values is None:
                     warnings.append("context_entry_invalid")
                     continue
                 seen_repo_keys.add(normalized_repo_key)
                 context_records.append(
-                    ContextRecord.create(normalized_repo_key, context)
+                    ContextRecord.create(normalized_repo_key, context_values)
                 )
-                if primary_path is None:
-                    primary_path = resolved
+                if primary_context_path is None:
+                    primary_context_path = context_path
 
     primary_context_record = context_records[0] if context_records else None
     primary_context = (
@@ -120,13 +122,13 @@ def load_resources(
             warnings,
             "telemetry_facts_manifest_invalid",
         ):
-            resolved = _resolve_optional(resolver, (artifact_path, short_path))
-            if resolved is None:
+            facts_path = _resolve_optional(resolver, (artifact_path, short_path))
+            if facts_path is None:
                 warnings.append("telemetry_facts_entry_unresolved")
                 continue
-            telemetry_facts_paths.append(resolved)
-    if runtime_override_enabled and primary_path is not None:
-        sibling = primary_path.parent / "telemetry_facts.json"
+            telemetry_facts_paths.append(facts_path)
+    if runtime_override_enabled and primary_context_path is not None:
+        sibling = primary_context_path.parent / "telemetry_facts.json"
         if sibling.is_file():
             telemetry_facts_paths.append(sibling.resolve())
     telemetry_facts_paths = list(dict.fromkeys(sorted(telemetry_facts_paths)))
@@ -138,7 +140,7 @@ def load_resources(
     return LoadedResources(
         context_plan=context_plan,
         primary_context=primary_context,
-        primary_context_path=primary_path,
+        primary_context_path=primary_context_path,
         telemetry_facts_paths=tuple(telemetry_facts_paths),
         schema=schema,
         warning_codes=tuple(dict.fromkeys(warnings)),