feat: add LocalRunService for reporting locally-orchestrated runs

buf.yaml

@@ -19,11 +19,21 @@ lint:
       - flyteidl2/cacheservice/cacheservice.proto
       - flyteidl2/cacheservice/v2/cacheservice.proto
       - flyteidl2/project/project_service.proto
+      # LocalRunService deliberately reuses RunService request/response messages so clients and
+      # UIs work against local runs with only a service swap. The local-run dataproxy RPCs reuse
+      # the corresponding dataplane RPCs' messages for the same reason.
+      - flyteidl2/dataproxy/dataproxy_service.proto
+      - flyteidl2/workflow/local_run_service.proto
+      - flyteidl2/workflow/run_service.proto
     RPC_REQUEST_STANDARD_NAME:
       - flyteidl2/cacheservice/cacheservice.proto
       - flyteidl2/cacheservice/v2/cacheservice.proto
+      - flyteidl2/dataproxy/dataproxy_service.proto
+      - flyteidl2/workflow/local_run_service.proto
     RPC_RESPONSE_STANDARD_NAME:
       - flyteidl2/cacheservice/cacheservice.proto
       - flyteidl2/cacheservice/v2/cacheservice.proto
+      - flyteidl2/dataproxy/dataproxy_service.proto
+      - flyteidl2/workflow/local_run_service.proto
     SERVICE_SUFFIX:
       - flyteidl2/datacatalog/datacatalog.proto

flyteidl2/dataproxy/dataproxy_service.proto

@@ -21,6 +21,10 @@ enum ArtifactType {
   ARTIFACT_TYPE_REPORT = 1;
   // ARTIFACT_TYPE_CODE_BUNDLE refers to the code bundle (tarball) generated for an action.
   ARTIFACT_TYPE_CODE_BUNDLE = 2;
+  // ARTIFACT_TYPE_INPUTS refers to an action's serialized inputs (inputs.pb).
+  ARTIFACT_TYPE_INPUTS = 3;
+  // ARTIFACT_TYPE_OUTPUTS refers to an action's serialized outputs (outputs.pb).
+  ARTIFACT_TYPE_OUTPUTS = 4;
 }
 
 // DataProxyService provides an interface for managing data uploads and downloads.
@@ -40,6 +44,22 @@ service DataProxyService {
 
   // Stream logs for an action attempt.
   rpc TailLogs(TailLogsRequest) returns (stream TailLogsResponse) {}
+
+  // UploadMetadata generates a signed URL for uploading a local run's metadata artifact
+  // (inputs.pb / outputs.pb / report.html) directly to the control plane's storage backend.
+  // Local runs only; never routes to a dataplane.
+  rpc UploadMetadata(UploadMetadataRequest) returns (CreateUploadLocationResponse) {}
+
+  // Get input and output data for an action of a local run, served directly from the control
+  // plane's storage backend. Local runs only; never routes to a dataplane.
+  rpc GetLocalActionData(GetActionDataRequest) returns (GetActionDataResponse) {
+    option idempotency_level = NO_SIDE_EFFECTS;
+  }
+
+  // CreateLocalDownloadLink generates signed URL(s) for downloading an artifact of a local run
+  // action attempt directly from the control plane's storage backend. Local runs only; never
+  // routes to a dataplane.
+  rpc CreateLocalDownloadLink(CreateLocalDownloadLinkRequest) returns (CreateDownloadLinkResponse) {}
 }
 
 // CreateUploadLocationRequest specifies the request for the CreateUploadLocation API.
@@ -233,6 +253,67 @@ message TailLogsRequest {
   optional string connector_endpoint = 4;
 }
 
+// UploadMetadataRequest specifies the request for the UploadMetadata API. The data proxy service
+// derives the storage path from the authenticated org and the target action, following the local
+// run metadata layout: <prefix>/<org>/<project>/<domain>/<run>/<action>/inputs.pb for inputs and
+// <prefix>/<org>/<project>/<domain>/<run>/<action>/<attempt>/{outputs.pb,report.html} otherwise.
+message UploadMetadataRequest {
+  // ArtifactType is the type of artifact to upload. INPUTS targets an action; OUTPUTS and REPORT
+  // target an action attempt.
+  ArtifactType artifact_type = 1 [(buf.validate.field).enum = {
+    in: [
+      1,
+      3,
+      4
+    ]
+  }];
+
+  // Target identifies the action (INPUTS) or action attempt (OUTPUTS / REPORT) the artifact
+  // belongs to.
+  oneof target {
+    option (buf.validate.oneof).required = true;
+
+    common.ActionIdentifier action_id = 2;
+
+    common.ActionAttemptIdentifier action_attempt_id = 3;
+  }
+
+  // ContentMD5 restricts the upload location to the specific MD5 provided.
+  // +required
+  bytes content_md5 = 4 [(buf.validate.field).bytes.len = 16];
+
+  // ContentLength specifies the size of the content to be uploaded in bytes. Validated against
+  // the platform's maximum upload size.
+  // +optional
+  int64 content_length = 5;
+
+  // ExpiresIn defines the requested expiration duration for the generated URL. The request will
+  // be rejected if this exceeds the platform's configured maximum.
+  // +optional. The default value comes from the global config.
+  google.protobuf.Duration expires_in = 6;
+
+  // If true, the data proxy will add content_md5 to the Signed URL requirements, forcing clients
+  // to send this checksum with the object. Required for data integrity on backends like GCP.
+  bool add_content_md5_metadata = 7;
+}
+
+// CreateLocalDownloadLinkRequest specifies the request for the CreateLocalDownloadLink API.
+message CreateLocalDownloadLinkRequest {
+  // ArtifactType is the type of artifact to download.
+  // +required
+  ArtifactType artifact_type = 1 [(buf.validate.field).enum = {
+    not_in: [0]
+  }];
+
+  // ActionAttemptId identifies the local run action attempt whose artifact is to be downloaded.
+  common.ActionAttemptIdentifier action_attempt_id = 2 [(buf.validate.field).required = true];
+
+  // ExpiresIn defines the requested expiration duration for the generated URLs. The request will
+  // be rejected if this exceeds the platform's configured maximum.
+  // +optional. The default value comes from the global config.
+  google.protobuf.Duration expires_in = 3;
+}
+
 // Reponse message for tailing logs.
 message TailLogsResponse {
   // A batch of logs.

flyteidl2/workflow/local_run_service.proto

@@ -0,0 +1,149 @@
+syntax = "proto3";
+
+package flyteidl2.workflow;
+
+import "buf/validate/validate.proto";
+import "flyteidl2/common/identifier.proto";
+import "flyteidl2/common/run.proto";
+import "flyteidl2/task/common.proto";
+import "flyteidl2/task/run.proto";
+import "flyteidl2/task/task_definition.proto";
+import "flyteidl2/workflow/run_definition.proto";
+import "flyteidl2/workflow/run_service.proto";
+import "google/protobuf/timestamp.proto";
+import "google/rpc/status.proto";
+
+option go_package = "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/workflow";
+
+// LocalRunService manages runs that are orchestrated OUTSIDE the platform — typically on a user's
+// machine — where the client executes actions itself and only reports their state here. Local runs
+// are tracked separately from platform-orchestrated runs and never involve a dataplane.
+//
+// The read/watch surface deliberately mirrors RunService (same request/response messages) so that
+// clients and UIs built against RunService work against local runs with only a service swap.
+service LocalRunService {
+  // Register a new local run. The server creates the root action ("a0") in the reported state and
+  // returns the resolved run. If a run name is not provided, the server generates one.
+  rpc CreateRun(CreateLocalRunRequest) returns (CreateRunResponse) {}
+
+  // Report state for one or more actions of a local run. Creates actions on first report and
+  // updates them on subsequent reports. Reports are idempotent: an event with an
+  // (attempt, version, phase) tuple that was already recorded is acknowledged as success.
+  rpc ReportActions(ReportLocalActionsRequest) returns (ReportLocalActionsResponse) {}
+
+  // Get detailed information about a local run.
+  rpc GetRunDetails(GetRunDetailsRequest) returns (GetRunDetailsResponse) {
+    option idempotency_level = NO_SIDE_EFFECTS;
+  }
+
+  // Stream detailed information updates about a local run. The call will terminate when the run
+  // reaches a terminal phase.
+  rpc WatchRunDetails(WatchRunDetailsRequest) returns (stream WatchRunDetailsResponse) {}
+
+  // Get detailed information about an action of a local run.
+  rpc GetActionDetails(GetActionDetailsRequest) returns (GetActionDetailsResponse) {
+    option idempotency_level = NO_SIDE_EFFECTS;
+  }
+
+  // Stream detailed information updates about an action of a local run. The call will terminate
+  // when the action reaches a terminal phase.
+  rpc WatchActionDetails(WatchActionDetailsRequest) returns (stream WatchActionDetailsResponse) {}
+
+  // List local runs based on the provided filter criteria.
+  rpc ListRuns(ListRunsRequest) returns (ListRunsResponse) {
+    option idempotency_level = NO_SIDE_EFFECTS;
+  }
+
+  // Stream updates for local runs based on the provided filter criteria.
+  rpc WatchRuns(WatchRunsRequest) returns (stream WatchRunsResponse) {}
+
+  // List all actions for a given local run.
+  rpc ListActions(ListActionsRequest) returns (ListActionsResponse) {
+    option idempotency_level = NO_SIDE_EFFECTS;
+  }
+
+  // Stream updates for actions of a given local run.
+  rpc WatchActions(WatchActionsRequest) returns (stream WatchActionsResponse) {}
+
+  // Abort a local run: mark the run and all of its non-terminal actions ABORTED on the server.
+  // The platform cannot stop the local orchestrator; subsequent reports against aborted actions
+  // are rejected. Aborting an already-terminal run is a no-op acknowledged as success.
+  rpc AbortRun(AbortRunRequest) returns (AbortRunResponse) {}
+}
+
+// Request message for creating a local run.
+message CreateLocalRunRequest {
+  oneof id {
+    option (buf.validate.oneof).required = true;
+
+    // The user provided run id.
+    common.RunIdentifier run_id = 1;
+
+    // The project id for this run. Run name will be generated.
+    common.ProjectIdentifier project_id = 2;
+  }
+
+  // The task this run executes.
+  oneof task {
+    option (buf.validate.oneof).required = true;
+
+    // The task id to use.
+    task.TaskIdentifier task_id = 3;
+
+    // The task spec to use.
+    task.TaskSpec task_spec = 4;
+  }
+
+  // Reference to the run's inputs, previously uploaded via DataProxyService.UploadMetadata.
+  optional common.OffloadedInputData offloaded_input_data = 5;
+
+  // The run spec to use. Only client-relevant fields are honored; dataplane scheduling fields
+  // (queue, cluster) are ignored for local runs.
+  task.RunSpec run_spec = 6;
+
+  // User-defined labels attached to this run at creation time.
+  map<string, string> labels = 7;
+
+  // Time the local run actually started. If unset, the server defaults it to the current time.
+  google.protobuf.Timestamp run_start_time = 8;
+}
+
+// A single action state report within a local run.
+message LocalActionUpdate {
+  // The event describing this state transition. Carries the action id, attempt (starting at 1),
+  // phase, monotonically increasing version, timestamps, error info and output references
+  // (output_uri / report_uri under the control plane's storage).
+  ActionEvent event = 1 [(buf.validate.field).required = true];
+
+  // Name of the parent action if this is a nested action. Empty for the root action. Only
+  // honored on the first report of an action.
+  string parent_name = 2;
+
+  // Group this action belongs to, if applicable. Only honored on the first report of an action.
+  string group = 3;
+
+  // The action's spec. Should be provided on the first report of an action; ignored afterwards.
+  oneof spec {
+    TaskAction task = 4;
+    TraceAction trace = 5;
+  }
+
+  // Rolled-up status of the action (attempts count, start/end time).
+  ActionStatus status = 6;
+}
+
+// Request message for reporting local action state.
+message ReportLocalActionsRequest {
+  // The run these action reports belong to. Every update's action id must reference this run.
+  common.RunIdentifier run_id = 1 [(buf.validate.field).required = true];
+
+  // The action state reports.
+  repeated LocalActionUpdate updates = 2 [(buf.validate.field).repeated.min_items = 1];
+}
+
+// Response message for reporting local action state.
+message ReportLocalActionsResponse {
+  // Per-update results, parallel to the request's updates. Duplicate (already recorded) events
+  // are reported as success.
+  repeated google.rpc.Status statuses = 1;
+}

flyteidl2/workflow/run_definition.proto

@@ -477,6 +477,10 @@ enum RunSource {
   RUN_SOURCE_WEB = 1;
   RUN_SOURCE_CLI = 2;
   RUN_SOURCE_SCHEDULE_TRIGGER = 3;
+
+  // The run is orchestrated outside the platform (e.g. on a user's machine) and its state is
+  // reported via LocalRunService.
+  RUN_SOURCE_LOCAL = 4;
 }
 
 // TaskGroup represents a group of runs for a specific task.