[cherry-pick][2.58.0][core][taskEvents out of GCS][8/n] Reroute state head to task events head for state APIs. (#65160)

[elliot-barn]

Cherry-pick of #65160 (merge commit 12436ea) into releases/2.58.0.

This PR builds on #65141, so this branch is stacked on the #65141 cherry-pick (#65303) and must merge after it — until then the diff shown here includes both commits. The pick applied cleanly on top of that branch, and the patch-id matches the original merge commit exactly.

The intermediate 7/n (#65158) that this PR builds on needs no separate cherry-pick — it was merged into the #65141 PR branch rather than into master, so its content already arrives via #65303.

Not a duplicate: no existing open PR against releases/2.58.0 contains this change. Cherry-picked with AI assistance (Claude Code); pre-commit hooks ran on the changed files and passed.

🤖 Generated with Claude Code

[gemini-code-assist]

Code Review

This pull request implements the migration of task events from GCS to the dashboard head. It introduces a new TaskEventsHead subprocess module, an in-memory TaskEventStorage with priority-based garbage collection, and a TaskEventManager to reconcile task states against GCS worker and job lifecycle events. Additionally, it adds an HTTP publisher client to send events from the aggregator agent to the dashboard head and updates the State API to query from this new source when enabled. The review feedback highlights a few critical issues: a missing aiohttp import in the publisher client that would cause a NameError on connection failures, potential CPU-exhausting infinite loops in the GCS subscription polling error paths, and a double-counting bug in the dropped task attempts metric.

[gemini-code-assist]

The module aiohttp is referenced in this file (e.g., aiohttp.ClientConnectionError on line 398) but is never imported. This will lead to a NameError when a connection error or timeout occurs. Please import aiohttp at the top of the file.

Suggested change

	import asyncio
	import asyncio
	import aiohttp

[gemini-code-assist]

If subscriber.poll raises an exception repeatedly (e.g., due to a persistent GCS connection issue), this loop will spin infinitely without any delay, leading to 100% CPU exhaustion and log flooding. Adding a short sleep (e.g., await asyncio.sleep(1)) in the except block will prevent this.

Suggested change

	except Exception:
	logger.exception("Failed handling worker-death notifications.")
	except Exception:
	logger.exception("Failed handling worker-death notifications.")
	await asyncio.sleep(1)

[gemini-code-assist]

If subscriber.poll raises an exception repeatedly (e.g., due to a persistent GCS connection issue), this loop will spin infinitely without any delay, leading to 100% CPU exhaustion and log flooding. Adding a short sleep (e.g., await asyncio.sleep(1)) in the except block will prevent this.

Suggested change

	except Exception:
	logger.exception("Failed handling job-finished notifications.")
	except Exception:
	logger.exception("Failed handling job-finished notifications.")
	await asyncio.sleep(1)

[gemini-code-assist]

The metric STAT_TOTAL_ATTEMPTS_DROPPED is incremented here, but if attempt in self._primary_index is true, _remove_task_attempt(attempt) is called, which increments the same metric again. This leads to double-counting of dropped task attempts. We should only increment it here if the attempt is not already in the index.

Suggested change

	self._stats[STAT_TOTAL_ATTEMPTS_DROPPED] += 1
	if attempt in self._primary_index:
	self._remove_task_attempt(attempt)
	if attempt in self._primary_index:
	self._remove_task_attempt(attempt)
	else:
	self._stats[STAT_TOTAL_ATTEMPTS_DROPPED] += 1

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 5a90b44. Configure here.}

[cursor]

Unclear PR description

Low Severity

⚠️ This PR needs a clearer title and/or description.

To help reviewers, please ensure your PR includes:

• Title: A concise summary of the change
• Description:
  • What problem does this solve?
  • How does this PR solve it?
  • Any relevant context for reviewers such as:
    • Why is the problem important to solve?
    • Why was this approach chosen over others?

See this list of PRs as examples for PRs that have gone above and beyond:

• [Core] Introduce local port service discovery  #59613
• [Core] Improve Large-Scale Resource View Synchronization Through Sync Message Batching #57641
• Remove node observability information from hot path of core components #56474
• [core][rdt] Support out-of-order actors by extracting metadata when creating #59610
• [core] fix open leak for plasma store memory (shm/fallback) by workers #52622

This violates the Clear PR Descriptions and Titles rule: the description covers cherry-pick stacking logistics but does not explain the problem being solved or how the change fixes it.

 

^{Triggered by project rule: Bugbot Rules}

^{Reviewed by Cursor Bugbot for commit 5a90b44. Configure here.}