Add control-plane reporting for tracked runs (flyte run --tracked) (#1356)

## Overview

Tracked runs: the SDK keeps orchestrating everything locally while
reporting run state — actions, attempts, inputs/outputs, HTML reports —
to the control plane, so local runs appear live in the console. Builds
on the IDL in flyteorg/flyte#7737 and the `TrackedRunService` rename
(released as flyteidl2 2.0.40, which this PR pins).

### How it works

- **RemoteRunReporter** — a third `RunRecorder` sink alongside the TUI
tracker and SQLite store. The recorder observer contract is synchronous
and fires from async controller code *and* the threads that run sync
tasks, so the sink is a sync enqueue facade over async I/O: `record_*`
calls capture a fully-computed event under a lock (per-attempt monotonic
versions; attempts start at 1 per the server contract) and a dedicated
background event loop batches `TrackedRunService.ReportActions` calls
and performs the signed-URL uploads (the same background-loop shape as
`flyte.syncify`). Terminal events trigger a bounded flush barrier at run
end. Reporting is best-effort and never fails or hangs the run (unless
strict mode is on).
- **Data path routes like everything else**: artifacts upload via
`DataProxyService.CreateUploadLocation` signed PUT URLs with the
`tracked-runs/<run>/<action>[/<attempt>]` filename_root scheme, routed
by `ClusterService.SelectCluster(OPERATION_TRACKED_RUN_DATA)` — an org
with a directly-reachable data plane cluster uploads straight to that
cluster's store; an org without one is served by the control plane's own
store. The reporter stamps the routing cluster onto reported attempt
events so reads later route to the same store. Inputs upload before
CreateRun (deterministic path, no URI on the event); outputs/report URIs
attach to the terminal event after their uploads complete. Reports also
flush live mid-run so the report tab updates while the run executes.
- **Opt-in surface** (one naming family, anchored on the flag):
  - `flyte run --local --tracked [--tracked-strict]`
- config: `local.tracked` / `local.tracked_strict` (section mirrors
`local.persistence`)
- `flyte.init(local_tracked=..., local_tracked_strict=...)`; `flyte
create config --local-tracked`
  - runcontext: `flyte.with_runcontext(mode="local", tracked=True)`
- Requires project/domain; degrades gracefully with one warning when no
client is initialized. `--tracked-strict` fails the run loudly on the
first reporting failure (for debugging reporting itself). SIGINT/SIGTERM
abort-flushes in-flight actions as ABORTED. `Run.url` points at the
console tracked-runs page. Falsy outputs (`0`, `""`, `[]`, `False`)
report correctly (presence, not truthiness). File/Dir raw data never
uploads — only the three metadata artifacts (inputs.pb / outputs.pb /
report.html) reach the dataproxy.

The deck feature (`@env.task(report=True)`, `flyte.report`) and the
`@trace` decorator are unrelated and untouched.

## Test Plan

- 688 unit tests green on the released flyteidl2 pin (reporter lifecycle
incl. retries/ordering/flush barrier/failure isolation/strict mode,
upload helper, client protocol, CLI flags, config plumbing).
- Live E2E against three environments running a backend that implements
`TrackedRunService`:
- a single-process development stack: full lifecycle, watches, replay
idempotency.
- a hosted control plane with a directly-reachable data plane cluster:
runs + inputs/outputs + reports render in the console; artifacts land in
the cluster's metadata store, with the routing cluster recorded on
attempts.
- a hosted control plane where artifacts are served by the control
plane's own store — the fallback leg of the routed data path, verified
end to end on the released flyteidl2 pin.

## Rollout Plan

The feature is opt-in per run/config and requires a control plane that
implements `TrackedRunService`; nothing changes for users who don't pass
`--tracked`.

## Rollback Plan

Revert; the feature is client-side and opt-in.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Signed-off-by: Haytham Abuelfutuh <haytham@afutuh.com>

Author: EngHabu

pyproject.toml

@@ -214,7 +214,6 @@ mypy_path = ["src"]
 
 [tool.uv]
 exclude-newer = "5 days"
-
 # Opt flyteorg packages out of the global recency cutoff so freshly published
 # flyteidl2/flyte_controller_base releases can be picked up immediately.
 [tool.uv.exclude-newer-package]

src/flyte/_initialize.py

@@ -43,6 +43,8 @@ class CommonInit:
     source_config_path: Optional[Path] = None  # Only used for documentation
     sync_local_sys_paths: bool = True
     local_persistence: bool = False
+    local_tracked: bool = False
+    local_tracked_strict: bool = False
 
 
 @dataclass(init=True, kw_only=True, repr=True, eq=True, frozen=True)
@@ -233,6 +235,8 @@ async def init(
     sync_local_sys_paths: bool = True,
     load_plugin_type_transformers: bool = True,
     local_persistence: bool = False,
+    local_tracked: bool = False,
+    local_tracked_strict: bool = False,
 ) -> None:
     """
     Initialize the Flyte system with the given configuration. This method should be called before any other Flyte
@@ -280,6 +284,11 @@ async def init(
         load_plugin_type_transformers: If enabled (default True), load the type transformer plugins registered under
             the "flyte.plugins.types" entry point group.
         local_persistence: Whether to enable SQLite persistence for local run metadata (default: False).
+        local_tracked: Whether to report tracked run state to the Flyte control plane
+            (default: False). Requires an initialized client and a configured project/domain.
+        local_tracked_strict: Strict tracked-run reporting for debugging (default: False). Any
+            reporting failure fails the run loudly instead of being logged and swallowed. Only takes
+            effect when reporting is enabled.
         disable_keyring: Disable storage of tokens in local keyring.
 
     Returns:
@@ -341,6 +350,8 @@ async def init(
             source_config_path=source_config_path,
             sync_local_sys_paths=sync_local_sys_paths,
             local_persistence=local_persistence,
+            local_tracked=local_tracked,
+            local_tracked_strict=local_tracked_strict,
         )
 
         logger.info(f"Flyte initialized with config: {_init_config}")
@@ -435,6 +446,8 @@ async def init_from_config(
         source_config_path=cfg_path,
         sync_local_sys_paths=sync_local_sys_paths,
         local_persistence=cfg.local.persistence,
+        local_tracked=cfg.local.tracked,
+        local_tracked_strict=cfg.local.tracked_strict,
         # disable_keyring is threaded outside _platform_to_client_kwargs
         # because the helper output is also spread into
         # create_remote_controller from init_in_cluster, and the
@@ -781,6 +794,22 @@ def is_persistence_enabled() -> bool:
     return cfg.local_persistence
 
 
+def is_local_tracked_enabled() -> bool:
+    """Check if reporting tracked run state to the control plane is enabled."""
+    cfg = _get_init_config()
+    if cfg is None:
+        return False
+    return cfg.local_tracked
+
+
+def is_local_tracked_strict() -> bool:
+    """Check if strict tracked-run reporting (fail the run on any reporting failure) is enabled."""
+    cfg = _get_init_config()
+    if cfg is None:
+        return False
+    return cfg.local_tracked_strict
+
+
 def initialize_in_cluster() -> None:
     """
     Initialize the system for in-cluster execution. This is a placeholder function and does not perform any actions.

src/flyte/_internal/controllers/_local_controller.py

@@ -221,6 +221,8 @@ async def submit(self, _task: TaskTemplate, *args, **kwargs) -> Any:
             short_name=_task.short_name if _task.short_name != _task.name else None,
             parent_id=parent_id,
             inputs=native_inputs,
+            proto_inputs=inputs.proto_inputs,
+            task=_task,
             output_path=sub_action_output_path,
             has_report=_task.report,
             cache_enabled=cache_enabled,
@@ -374,6 +376,8 @@ async def get_action_outputs(
                 task_name=func_name,
                 parent_id=task_action.name,
                 inputs=native_inputs,
+                proto_inputs=converted_inputs.proto_inputs,
+                trace_interface=_interface,
                 output_path=action_output_path,
             )
 
@@ -406,7 +410,8 @@ async def record_trace(self, info: TraceInfo):
             self._recorder.record_failure(action_id=info.action.name, error=str(info.error))
         else:
             converted_outputs = None
-            if info.interface.outputs and info.output:
+            # Presence, not truthiness: falsy results (0, "", [], False) are real outputs.
+            if info.interface.outputs and info.output is not None:
                 _ctx = ctx.new_in_driver_literal_conversion(True) if ctx.is_task_context() else nullcontext()
                 with _ctx:
                     converted_outputs = await convert.convert_from_native_to_outputs(

src/flyte/_persistence/_recorder.py

@@ -4,8 +4,9 @@
 
 
 class RunRecorder:
-    """Unified proxy that delegates recording events to the TUI tracker and/or
-    the SQLite persistence layer (RunStore).
+    """Unified proxy that delegates recording events to the TUI tracker, the SQLite
+    persistence layer (RunStore) and/or the control-plane reporter
+    (`RemoteRunReporter`).
 
     The controller only talks to this single object — no more interleaved
     `if tracker` / `if persist` conditionals.
@@ -19,20 +20,25 @@ def __init__(
         tracker: Any | None = None,
         persist: bool = False,
         run_name: str | None = None,
+        reporter: Any | None = None,
     ) -> None:
         self._tracker = tracker
         self._persist = persist and run_name is not None
         self._run_name: str = run_name or ""
+        self._reporter = reporter
 
     @property
     def is_active(self) -> bool:
         """True if at least one recording backend is enabled."""
-        return self._tracker is not None or self._persist
+        return self._tracker is not None or self._persist or self._reporter is not None
 
     def get_action(self, action_id: str) -> Any:
-        """Delegate to the tracker, or return None when tracker is absent."""
+        """Delegate to the tracker (or the reporter when no tracker is attached), or
+        return None when neither backend knows the action."""
         if self._tracker is not None:
             return self._tracker.get_action(action_id)
+        if self._reporter is not None:
+            return self._reporter.get_action(action_id)
         return None
 
     # ------------------------------------------------------------------
@@ -74,6 +80,9 @@ def record_start(
         parent_id: str | None = None,
         short_name: str | None = None,
         inputs: dict | None = None,
+        proto_inputs: Any = None,
+        task: Any = None,
+        trace_interface: Any = None,
         output_path: str | None = None,
         has_report: bool = False,
         cache_enabled: bool = False,
@@ -121,6 +130,26 @@ def record_start(
                 log_links=log_links,
             )
 
+        if self._reporter is not None:
+            # The reporter needs the raw proto inputs (when available) so it can
+            # offload the action's inputs.pb, and the task template / trace interface
+            # so the reported spec carries a typed interface (the console gates I/O
+            # rendering on it).
+            self._reporter.record_start(
+                action_id=action_id,
+                task_name=task_name,
+                parent_id=parent_id,
+                proto_inputs=proto_inputs,
+                task=task,
+                trace_interface=trace_interface,
+                output_path=output_path,
+                has_report=bool(has_report),
+                group=group,
+                cache_enabled=cache_enabled,
+                cache_hit=cache_hit,
+                disable_run_cache=disable_run_cache,
+            )
+
     def record_complete(self, *, action_id: str, outputs: Any = None) -> None:
         # Convert outputs to a display representation once, so both backends
         # receive the same pre-formatted data.
@@ -140,6 +169,10 @@ def record_complete(self, *, action_id: str, outputs: Any = None) -> None:
                 outputs=repr(display) if display is not None else None,
             )
 
+        if self._reporter is not None:
+            # The reporter needs the raw outputs (proto wrapper), not the display form.
+            self._reporter.record_complete(action_id=action_id, outputs=outputs)
+
     @staticmethod
     def _to_display(outputs: Any) -> Any:
         """Convert raw outputs to a display-friendly representation.
@@ -173,6 +206,9 @@ def record_failure(self, *, action_id: str, error: str) -> None:
                 error=error,
             )
 
+        if self._reporter is not None:
+            self._reporter.record_failure(action_id=action_id, error=error)
+
     def record_attempt_start(self, *, action_id: str, attempt_num: int) -> None:
         if self._tracker is not None and hasattr(self._tracker, "record_attempt_start"):
             self._tracker.record_attempt_start(action_id=action_id, attempt_num=attempt_num)
@@ -186,6 +222,9 @@ def record_attempt_start(self, *, action_id: str, attempt_num: int) -> None:
                 attempt_num=attempt_num,
             )
 
+        if self._reporter is not None:
+            self._reporter.record_attempt_start(action_id=action_id, attempt_num=attempt_num)
+
     def record_attempt_complete(self, *, action_id: str, attempt_num: int, outputs: Any = None) -> None:
         display: Any = None
         if outputs is not None:
@@ -208,6 +247,9 @@ def record_attempt_complete(self, *, action_id: str, attempt_num: int, outputs:
                 outputs=repr(display) if display is not None else None,
             )
 
+        if self._reporter is not None:
+            self._reporter.record_attempt_complete(action_id=action_id, attempt_num=attempt_num, outputs=outputs)
+
     def record_attempt_failure(self, *, action_id: str, attempt_num: int, error: str) -> None:
         if self._tracker is not None and hasattr(self._tracker, "record_attempt_failure"):
             self._tracker.record_attempt_failure(
@@ -226,8 +268,11 @@ def record_attempt_failure(self, *, action_id: str, attempt_num: int, error: str
                 error=error,
             )
 
+        if self._reporter is not None:
+            self._reporter.record_attempt_failure(action_id=action_id, attempt_num=attempt_num, error=error)
+
     # ------------------------------------------------------------------
-    # Root "a0" action (called by _run.py — persistence only)
+    # Root "a0" action (called by _run.py — persistence + reporter only)
     # ------------------------------------------------------------------
 
     def record_root_start(self, *, task_name: str) -> None:
@@ -241,6 +286,9 @@ def record_root_start(self, *, task_name: str) -> None:
                 parent_id=None,
             )
 
+        if self._reporter is not None:
+            self._reporter.record_root_start(task_name=task_name)
+
     def record_root_complete(self) -> None:
         if self._persist:
             from flyte._persistence._run_store import RunStore
@@ -250,6 +298,9 @@ def record_root_complete(self) -> None:
                 action_name="a0",
             )
 
+        if self._reporter is not None:
+            self._reporter.record_root_complete()
+
     def record_root_failure(self, *, error: str) -> None:
         if self._persist:
             from flyte._persistence._run_store import RunStore
@@ -260,6 +311,9 @@ def record_root_failure(self, *, error: str) -> None:
                 error=error,
             )
 
+        if self._reporter is not None:
+            self._reporter.record_root_failure(error=error)
+
     # ------------------------------------------------------------------
     # Static helpers
     # ------------------------------------------------------------------