uploader: clarify worker and reporting models

Use domain-specific names for telemetry, result, resource, and legacy-report state. Remove a redundant report wrapper and unused telemetry property while preserving configuration precedence and output contracts.

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

Author: tonyredondo

tools/core/uploader_py/config.py

@@ -161,6 +161,15 @@ def parse_uploader_config(
     """Resolve generated config, environment, and CLI using public precedence."""
     env = dict(os.environ if environ is None else environ)
     args = _parser().parse_args(list(argv))
+
+    def text_option(
+        cli_value: str | None,
+        variable: str,
+        default: str = "",
+    ) -> str:
+        """Prefer an explicit CLI value to its environment fallback."""
+        return cli_value if cli_value is not None else env.get(variable) or default
+
     config_path = Path(args.config)
     rule = load_rule_config(config_path)
     invocation_cwd = Path(cwd or Path.cwd()).absolute()
@@ -207,25 +216,20 @@ def parse_uploader_config(
     if args.validate_enrichment and not args.dry_run:
         raise ConfigError("--validate-enrichment requires --dry-run")
 
-    freshness_source_value = (
-        args.freshness_source
-        if args.freshness_source is not None
-        else env.get("DD_TEST_OPTIMIZATION_FRESHNESS_SOURCE") or "auto"
+    freshness_source_value = text_option(
+        args.freshness_source,
+        "DD_TEST_OPTIMIZATION_FRESHNESS_SOURCE",
+        "auto",
     )
     freshness_source = _choice(
         "--freshness-source/DD_TEST_OPTIMIZATION_FRESHNESS_SOURCE",
         freshness_source_value,
         VALID_FRESHNESS_SOURCES,
     )
-    new_freshness_mode = (
-        args.freshness_mode
-        if args.freshness_mode is not None
-        else env.get("DD_TEST_OPTIMIZATION_FRESHNESS_MODE") or None
-    )
-    legacy_freshness_mode = (
-        args.execution_log_mode
-        if args.execution_log_mode is not None
-        else env.get("DD_TEST_OPTIMIZATION_EXECUTION_LOG_MODE") or None
+    new_freshness_mode = text_option(args.freshness_mode, "DD_TEST_OPTIMIZATION_FRESHNESS_MODE")
+    legacy_freshness_mode = text_option(
+        args.execution_log_mode,
+        "DD_TEST_OPTIMIZATION_EXECUTION_LOG_MODE",
     )
     freshness_mode = _choice(
         "--freshness-mode/DD_TEST_OPTIMIZATION_FRESHNESS_MODE",
@@ -235,40 +239,39 @@ def parse_uploader_config(
     if args.allow_cached_payload_uploads:
         freshness_mode = "disabled"
 
-    artifact_source_value = (
-        args.artifact_source
-        if args.artifact_source is not None
-        else env.get("DD_TEST_OPTIMIZATION_ARTIFACT_SOURCE") or "local"
+    artifact_source_value = text_option(
+        args.artifact_source,
+        "DD_TEST_OPTIMIZATION_ARTIFACT_SOURCE",
+        "local",
     )
     artifact_source = _choice(
         "--artifact-source/DD_TEST_OPTIMIZATION_ARTIFACT_SOURCE",
         artifact_source_value,
         VALID_ARTIFACT_SOURCES,
     )
-    remote_artifacts_value = (
-        args.remote_artifacts
-        if args.remote_artifacts is not None
-        else env.get("DD_TEST_OPTIMIZATION_REMOTE_ARTIFACTS") or "disabled"
+    remote_artifacts_value = text_option(
+        args.remote_artifacts,
+        "DD_TEST_OPTIMIZATION_REMOTE_ARTIFACTS",
+        "disabled",
     )
     remote_artifacts = _choice(
         "--remote-artifacts/DD_TEST_OPTIMIZATION_REMOTE_ARTIFACTS",
         remote_artifacts_value,
         VALID_REMOTE_ARTIFACT_MODES,
     )
-    downloader_timeout_value = (
-        args.bep_artifact_downloader_timeout_sec
-        if args.bep_artifact_downloader_timeout_sec is not None
-        else env.get("DD_TEST_OPTIMIZATION_BEP_ARTIFACT_DOWNLOADER_TIMEOUT_SEC") or "300"
+    downloader_timeout_value = text_option(
+        args.bep_artifact_downloader_timeout_sec,
+        "DD_TEST_OPTIMIZATION_BEP_ARTIFACT_DOWNLOADER_TIMEOUT_SEC",
+        "300",
     )
     downloader_timeout = _positive_decimal(
         "--bep-artifact-downloader-timeout-sec",
         downloader_timeout_value,
     )
 
-    staging_text = (
-        args.artifact_staging_dir
-        if args.artifact_staging_dir is not None
-        else env.get("DD_TEST_OPTIMIZATION_ARTIFACT_STAGING_DIR", "")
+    staging_text = text_option(
+        args.artifact_staging_dir,
+        "DD_TEST_OPTIMIZATION_ARTIFACT_STAGING_DIR",
     )
     artifact_staging_dir = (
         Path(staging_text)
@@ -284,21 +287,17 @@ def parse_uploader_config(
         bep_json_values.append(environment_bep)
     bep_json_values.extend(args.bep_json)
 
-    expected_enriched_tags = tuple(args.expected_enriched_tag) or DEFAULT_EXPECTED_ENRICHED_TAGS
-    report_text = (
-        args.report_json
-        if args.report_json is not None
-        else env.get("DD_TEST_OPTIMIZATION_UPLOADER_REPORT_JSON", "")
+    expected_enriched_tags = (
+        tuple(args.expected_enriched_tag) or DEFAULT_EXPECTED_ENRICHED_TAGS
     )
-    execution_log_text = (
-        args.execution_log_json
-        if args.execution_log_json is not None
-        else env.get("DD_TEST_OPTIMIZATION_EXECUTION_LOG_JSON", "")
+    report_text = text_option(args.report_json, "DD_TEST_OPTIMIZATION_UPLOADER_REPORT_JSON")
+    execution_log_text = text_option(
+        args.execution_log_json,
+        "DD_TEST_OPTIMIZATION_EXECUTION_LOG_JSON",
     )
-    downloader_text = (
-        args.bep_artifact_downloader
-        if args.bep_artifact_downloader is not None
-        else env.get("DD_TEST_OPTIMIZATION_BEP_ARTIFACT_DOWNLOADER", "")
+    downloader_text = text_option(
+        args.bep_artifact_downloader,
+        "DD_TEST_OPTIMIZATION_BEP_ARTIFACT_DOWNLOADER",
     )
 
     return UploaderConfig(

tools/core/uploader_py/coordinator.py

@@ -39,6 +39,8 @@
 
 @dataclass(frozen=True)
 class CoordinatorSettings:
+    """Worker-facing configuration with lifecycle-only options removed."""
+
     workspace: Path
     workers: int
     dry_run: bool

tools/core/uploader_py/file_worker.py

@@ -111,6 +111,14 @@ class _TestRequest:
     content_encoding: str | None = None
 
 
+@dataclass(frozen=True)
+class _TelemetryRequest:
+    """One telemetry body paired with the headers derived from that body."""
+
+    body: Path
+    headers: dict[str, str]
+
+
 def common_headers(
     runtime: WorkerRuntime,
     payload: Mapping[str, Any] | None = None,
@@ -294,7 +302,7 @@ def _process_test(
         warnings,
     )
 
-    base = dict(
+    test_result_fields = dict(
         task_id=task.task_id,
         source_path=task.display_path,
         payload_type=task.payload_type,
@@ -317,7 +325,7 @@ def _process_test(
         return FileResult(
             status=FileStatus.SUCCEEDED,
             warning_codes=_unique_codes(warnings),
-            **base,
+            **test_result_fields,
         )
 
     requests_attempted = 0
@@ -366,7 +374,7 @@ def _process_test(
                 warning_codes=_unique_codes(warnings),
                 failure_code=failure_code,
                 failure_message=failure_message,
-                **base,
+                **test_result_fields,
             )
         requests_succeeded += 1
 
@@ -383,7 +391,7 @@ def _process_test(
         retries=retries,
         source_deleted=deleted,
         warning_codes=_unique_codes(final_warnings),
-        **base,
+        **test_result_fields,
     )
 
 
@@ -572,7 +580,7 @@ def _process_coverage(
     except OSError as exc:
         return _failed(task, "coverage_payload_read_failed", type(exc).__name__)
 
-    base = dict(
+    coverage_result_fields = dict(
         task_id=task.task_id,
         source_path=task.display_path,
         payload_type=task.payload_type,
@@ -611,7 +619,7 @@ def _process_coverage(
             task,
             f"dry-run prepared {prepared.content_length}-byte coverage multipart body",
         )
-        return FileResult(status=FileStatus.SUCCEEDED, **base)
+        return FileResult(status=FileStatus.SUCCEEDED, **coverage_result_fields)
 
     result = transport.post_prepared_multipart(
         runtime.endpoints.coverage_url,
@@ -635,7 +643,7 @@ def _process_coverage(
             retries=result.retries,
             failure_code=failure_code,
             failure_message=failure_message,
-            **base,
+            **coverage_result_fields,
         )
 
     deleted, cleanup_warning = _cleanup_source(task.source_path, runtime.keep_payloads)
@@ -647,7 +655,7 @@ def _process_coverage(
         retries=result.retries,
         source_deleted=deleted,
         warning_codes=(cleanup_warning,) if cleanup_warning else (),
-        **base,
+        **coverage_result_fields,
     )
 
 
@@ -694,7 +702,7 @@ def _process_telemetry(
             failure_message=metadata_failure,
         )
 
-    bodies: list[tuple[Path, dict[str, Any]]] = [(source_body, payload)]
+    requests = [_TelemetryRequest(source_body, _telemetry_headers(runtime, payload))]
     if directive.create_synthetic:
         synthetic = _build_synthetic_telemetry(payload, directive, warnings)
         if synthetic is not None:
@@ -704,58 +712,56 @@ def _process_telemetry(
             )
             synthetic_path = task_directory / "telemetry_synthetic.json"
             synthetic_path.write_bytes(_compact_json_line(synthetic))
-            bodies.append((synthetic_path, synthetic))
-    prepared_headers = tuple(
-        _telemetry_headers(runtime, body_payload)
-        for _body_path, body_payload in bodies
-    )
-    _debug(runtime, task, f"prepared {len(bodies)} telemetry request(s)")
+            requests.append(
+                _TelemetryRequest(synthetic_path, _telemetry_headers(runtime, synthetic))
+            )
+    _debug(runtime, task, f"prepared {len(requests)} telemetry request(s)")
     if _debug_enabled(runtime):
-        for index, (body_path, _body_payload) in enumerate(bodies, start=1):
+        for index, request in enumerate(requests, start=1):
             try:
-                body_bytes = body_path.stat().st_size
+                body_bytes = request.body.stat().st_size
             except OSError as exc:
                 _debug(
                     runtime,
                     task,
-                    f"telemetry request={index}/{len(bodies)} size unavailable: "
+                    f"telemetry request={index}/{len(requests)} size unavailable: "
                     f"{type(exc).__name__}",
                 )
             else:
                 _debug(
                     runtime,
                     task,
-                    f"telemetry request={index}/{len(bodies)} bytes={body_bytes}",
+                    f"telemetry request={index}/{len(requests)} bytes={body_bytes}",
                 )
 
-    base = dict(
+    telemetry_result_fields = dict(
         task_id=task.task_id,
         source_path=task.display_path,
         payload_type=task.payload_type,
-        requests_planned=len(bodies),
+        requests_planned=len(requests),
     )
     if runtime.dry_run:
-        for (body_path, _body_payload), headers in zip(bodies, prepared_headers):
+        for request in requests:
             prepare_json_request(
                 runtime.endpoints.telemetry_url,
-                headers,
-                body_path,
+                request.headers,
+                request.body,
             )
         _debug(runtime, task, "dry-run completed telemetry preparation without network")
         return FileResult(
             status=FileStatus.SUCCEEDED,
             warning_codes=_unique_codes(warnings),
-            **base,
+            **telemetry_result_fields,
         )
 
     requests_attempted = 0
     requests_succeeded = 0
     retries = 0
-    for (body_path, _body_payload), headers in zip(bodies, prepared_headers):
+    for request in requests:
         result = transport.post_json(
             runtime.endpoints.telemetry_url,
-            headers,
-            body_path,
+            request.headers,
+            request.body,
         )
         requests_attempted += result.attempts
         retries += result.retries
@@ -778,7 +784,7 @@ def _process_telemetry(
                 warning_codes=_unique_codes(warnings),
                 failure_code=failure_code,
                 failure_message=failure_message,
-                **base,
+                **telemetry_result_fields,
             )
         requests_succeeded += 1
 
@@ -793,7 +799,7 @@ def _process_telemetry(
         retries=retries,
         source_deleted=deleted,
         warning_codes=_unique_codes(warnings),
-        **base,
+        **telemetry_result_fields,
     )
 
 

tools/core/uploader_py/freshness.py

@@ -7,7 +7,7 @@
 """Pre-worker BEP staging and freshness selection.
 
 The doctor remains the source of truth for the non-trivial BEP and artifact
-carrier formats.  This module loads that runtime once, snapshots its result in
+carrier formats. This module loads that runtime once, snapshots its result in
 immutable uploader models, and keeps all filesystem and policy decisions out
 of worker threads.
 """
@@ -48,6 +48,8 @@ class RemoteOutput:
 
 @dataclass(frozen=True)
 class FreshnessPlan:
+    """Immutable authorization snapshot used by discovery and postflight checks."""
+
     selected_source: str = "none"
     eligibility_enabled: bool = False
     eligible_outputs: frozenset[tuple[str, str]] = frozenset()
@@ -62,6 +64,8 @@ class FreshnessPlan:
 
 @dataclass(frozen=True)
 class FreshnessPreparation:
+    """Freshness plan plus any staged artifacts owned by this invocation."""
+
     plan: FreshnessPlan
     scan_roots: tuple[ScanRoot, ...]
     staged_roots: tuple[Path, ...] = ()
@@ -175,6 +179,7 @@ def prepare_freshness(
         except SystemExit as exc:
             raise FreshnessError(_doctor_failure(doctor, exc)) from exc
         except BaseException:
+            # Staging must not survive interrupts or unexpected doctor errors.
             if staged and staging_base is not None:
                 doctor._cleanup_staged_bep_run_roots(
                     staged,
@@ -196,6 +201,7 @@ def prepare_freshness(
             },
         )
     except BaseException:
+        # Planning failures must obey the same ownership cleanup contract.
         if staged and doctor is not None and staging_base is not None:
             doctor._cleanup_staged_bep_run_roots(
                 staged,