[High] locals().get("transport") blinds the PG handler to early orchestration failures — the UN-3652 symptom can still occur.

transport is only bound as a local at L696 (transport = normalize_transport(...)), inside the try (starts L675). If the exception fires between L675 and L696 — e.g. in _get_callback_queue_name_api(), FairnessKey(...), or normalize_transport itself — then "transport" is not in locals(), this falls back to DEFAULT_WORKFLOW_TRANSPORT (celery), and is_pg_transport() returns False. For a genuine PG run that fails early the UI error publish is skipped and failure_counts is {}, leaving files perpetually "in progress" — exactly what this PR is meant to fix. The real transport is knowable from kwargs.get("transport") from the function's first line, so the "fail-closed to Celery" rationale doesn't hold here.

Secondary risk: reading via locals().get(...) defeats static analysis — a future rename of transport silently disables the whole PG path instead of raising NameError.

Fix: hoist transport above the try from the authoritative source and reference it directly (matching general/tasks.py, where transport is a normalized parameter), then drop this line and the duplicate assignment at L696.

[Medium] workflow_logger.log_error(...) is not None-guarded here — asymmetric with the api handler.

The api handler guards its logger (if error_logger:, api-deployment/tasks.py:828), but here workflow_logger.log_error(...) is called directly. workflow_logger is nullable and other sites in this file guard it (if workflow_logger:). If it is None, the resulting AttributeError is swallowed by the surrounding except and downgraded to a generic "Failed to publish error to UI logs" warning — the user gets no UI error and the cause looks like a transient publish hiccup. Guard it explicitly: if workflow_logger: workflow_logger.log_error(...).

[Medium] A failed status update silently drops the counter reconciliation, and this handler never re-raises.

The PR now routes the entire reconciliation (**failure_counts) through this update_workflow_execution_status call. If it raises, the except swallows it to a warning, so both the ERROR status and the successful=0/failed=total reconciliation are lost. Unlike the api handler, this path does not re-raise (it builds error_response and continues at L785), so there is no second chance for the failure to surface — the execution is left with total_files set and files perpetually "in progress", the exact UN-3652 state. Consider logging this at error (with an error id) rather than warning, since a failed terminal-status write is a real defect. (Distinct from the greptile P1 on the api handler, which is about the missing wrapper.)

[Low] Cache-resolved files are reported as failed.

Here total_files = len(files_to_send) includes file-history cache hits (cached_count = len(cached_results); cf. the "Send all files, not just non-cached ones" comment). pg_failure_file_counts returns successful=0, failed=total_files, so files already resolved from cache are counted as failed, inflating the "N failed" figure on the run. The in-progress=0 goal is still met, but the number is misleading. Consider computing successful_files = cached_count, failed_files = total_files - cached_count at this call site (the general path has no cache-on-send concept).

[Low] Return contract is stringly-coupled to a remote kwarg signature, with no test at the seam.

The three keys (total_files, successful_files, failed_files) must exactly match the parameter names of InternalAPIClient.update_workflow_execution_status because they are spread via **failure_counts. A rename/typo on either side surfaces only as a runtime TypeError, and only on the rare PG-failure path. The unit tests assert the literal dict but never that the keys are valid kwargs of the consumer, so they would stay green through a real break; dict[str, int] also can't express the successful + failed <= total invariant. Cheapest insurance: a regression test binding the result against the real signature, e.g. inspect.signature(InternalAPIClient.update_workflow_execution_status).bind_partial(execution_id=..., status=..., **counts). Optionally document the shape with a TypedDict (it still can't encode the arithmetic invariant, so prioritize the test). transport: str is correct as-is — the whole transport seam traffics in canonical strings.

[High — test coverage] The behavioral change of this PR is untested; these tests cover only the pure helper.

The new tests exercise pg_failure_file_counts (a dict literal behind an is_pg_transport gate), but the actual change — the two rewritten except handlers (error surfacing to UI, the locals().get fallback, **failure_counts → update_workflow_execution_status, the nested try/except) — has zero coverage. A regression there silently reintroduces the "files stuck in progress" bug.

Highest-value addition: a parametrized (PG vs Celery) unit test of each handler with a mocked api_client that forces the orchestration call to raise, asserting (PG) update_workflow_execution_status is called with status=ERROR, total_files=N, successful_files=0, failed_files=N, and (Celery) with no file-aggregate kwargs (proving "byte-identical" at the call site, not just in the helper). Also worth covering: the UI-logger raising must not prevent the status update / re-raise.

Minor: test_pg_counts_satisfy_serializer_constraints re-implements the serializer rule inline instead of invoking the real validator, and test_pg_marks_attempted_files_failed / test_result_zeroes_ui_in_progress largely restate the literal (they only fail if someone edits the dict, not if behavior breaks). The genuinely useful cases are the Celery/non-PG no-op tests.

[Low] Comment rot: orphan (B)/(C) labels and a hard-coded line reference.

• (C)/(B) (also in general/tasks.py and the test module docstring) reference subtasks of the external Jira plan — there is no (A), they are out of order, and nothing in the code defines them, so they become meaningless once the ticket closes. Drop the letters or renumber as self-contained (1)/(2) in logical order.
• ~L695 (L818) is already off by one (the assignment is at L696) and will drift further with any edit. Reference it symbolically: "assigned by the normalize_transport(...) call inside the try".

Also: this ~10-line rationale is near-duplicated in general/tasks.py:757-764 and already diverging ("bound above the try" vs "persisted at orchestration start"). Consider trimming both to a one-line pointer to the pg_failure_file_counts docstring, which already carries the full rationale. (The factual claims themselves — serializer constraint, total_files binding — were verified accurate.)