From e220ca8b3a82c19b56d7b4b12548dd631cca8ea5 Mon Sep 17 00:00:00 2001 From: Lukas Bindreiter Date: Mon, 19 Oct 2026 15:22:32 +0200 Subject: [PATCH] Workflow scoped task identifiers --- apis/workflows/v1/core.proto | 31 +++++++++++++++++++++++++------ apis/workflows/v1/job.proto | 16 ++++++++++++++++ apis/workflows/v1/task.proto | 8 ++++++-- apis/workflows/v1/worker.proto | 9 ++++++++- apis/workflows/v1/workflows.proto | 17 ++++++++++++++--- 5 files changed, 69 insertions(+), 12 deletions(-) diff --git a/apis/workflows/v1/core.proto b/apis/workflows/v1/core.proto index 055424a..1fc350c 100644 --- a/apis/workflows/v1/core.proto +++ b/apis/workflows/v1/core.proto @@ -143,9 +143,10 @@ enum TaskState { TASK_STATE_FAILED_OPTIONAL = 6; } -// An identifier for a task. -message TaskIdentifier { - // A unique name of a task (unique within a namespace). +// Identifies a task implementation before an execution scope is selected. +// Release definitions and worker discovery use this type. Runners bind it to a workflow or advertise it unscoped. +message UnboundTaskIdentifier { + // The local task name. string name = 1 [(buf.validate.field).string = { min_bytes: 1 max_bytes: 256 @@ -157,9 +158,27 @@ message TaskIdentifier { }]; } -// A list of task identifiers. -message TaskIdentifiers { - repeated TaskIdentifier identifiers = 1; +// A list of task identifiers without execution scope bindings. +message UnboundTaskIdentifiers { + repeated UnboundTaskIdentifier identifiers = 1; +} + +// Identifies a task implementation by name, version and by (optional) workflow scope. +message TaskIdentifier { + // The task name within its workflow scope, or within the unscoped task space. + string name = 1 [(buf.validate.field).string = { + min_bytes: 1 + max_bytes: 256 + }]; + // Version of the task. + string version = 2 [(buf.validate.field).string = { + min_bytes: 1 + pattern: "^v(\\d+)\\.(\\d+)$" + }]; + // The immutable ID of the workflow that scopes this task. + // In task submissions and execution messages, omission means unscoped, never inheritance of a parent's scope. + // Executable tasks returned by the API carry the resolved scope. + tilebox.v1.ID workflow_id = 3 [features.field_presence = EXPLICIT]; } // A list of tasks. diff --git a/apis/workflows/v1/job.proto b/apis/workflows/v1/job.proto index a273d56..bed2686 100644 --- a/apis/workflows/v1/job.proto +++ b/apis/workflows/v1/job.proto @@ -76,6 +76,9 @@ message TaskSummary { string cluster_slug = 12; // Whether failure of the task is optional, either directly or through an optional ancestor. bool optional = 13; + // The remote reference of the task's workflow, owner-local or namespace-qualified as in Workflow.reference. + // Absent for unscoped tasks or deleted workflows. Omission does not change the task's stored execution scope. + string workflow_reference = 14 [features.field_presence = EXPLICIT]; } // JobTaskChildrenPrefetch configures the child pages included with the requested task page. @@ -198,6 +201,19 @@ message QueryFilters { // Filter jobs by their clusters (whether they have tasks on certain clusters). repeated string cluster_slugs = 7; + + // Filter jobs having at least one task scoped to any of these workflow references. + // Empty means no workflow filter. Unscoped tasks never match. + repeated string workflow_references = 8 [ + (buf.validate.field).repeated.items.string.min_len = 1, + (buf.validate.field).repeated.items.string.max_len = 201, + (buf.validate.field).repeated.items.string.pattern = "^([a-z0-9]+(-[a-z0-9]+)*:)?[A-Za-z0-9-]{1,100}$", + (buf.validate.field).repeated.items.cel = { + id: "workflow_reference.namespace_length" + message: "workflow namespace must be at most 100 characters" + expression: "!this.contains(':') || this.split(':')[0].size() <= 100" + } + ]; } // QueryJobsRequest requests a list of jobs. diff --git a/apis/workflows/v1/task.proto b/apis/workflows/v1/task.proto index 62a568c..387bace 100644 --- a/apis/workflows/v1/task.proto +++ b/apis/workflows/v1/task.proto @@ -24,7 +24,9 @@ message NextTaskRequest { message NextTaskToRun { // The cluster that this task runner is running on. string cluster_slug = 1 [(buf.validate.field).string.min_len = 1]; - // The task identifiers that this task runner can run. + // The concrete task addresses this runner can execute. Scoped capabilities include workflow_id; + // omission advertises only an unscoped address. During the migration/introduction phase of scoped tasks, + // runners advertise legacy unscoped aliases as separate entries. repeated TaskIdentifier identifiers = 2; } @@ -36,12 +38,14 @@ message ComputedTask { // If not set, the display message specified upon task submission will be kept. string display = 2; // A list of sub-tasks that the just computed task spawned. + // Workflow scope must be explicitly set in the sub-task identifiers, omission means explicitly unscoped. + // Cross-workflow child submissions are also allowed. TaskSubmissions sub_tasks = 5; // A list of progress updates that the computed task wants to report. repeated Progress progress_updates = 4; // A list of sub-tasks that the just computed task spawned. This field is deprecated and will be removed in the - // future. Kept around for backwards compatibility. + // future. Kept around for backwards compatibility. Workflow scope follows the same rules as sub_tasks. repeated SingleTaskSubmission legacy_sub_tasks = 3 [ deprecated = true, // we support more than that due to grouping and tree spawning, but we rather want to encourage users diff --git a/apis/workflows/v1/worker.proto b/apis/workflows/v1/worker.proto index 0fb1d35..47d052f 100644 --- a/apis/workflows/v1/worker.proto +++ b/apis/workflows/v1/worker.proto @@ -61,7 +61,9 @@ message ExecuteTaskResponse { // WorkerService defines the RPC interface between a runner and a a worker runtime that can execute workflow tasks. service WorkerService { // ListRegisteredTasks is called by the runner to discover the task identifiers that a worker runtime can execute. - rpc ListRegisteredTasks(google.protobuf.Empty) returns (workflows.v1.TaskIdentifiers); + // These are local task definitions without workflow IDs. The managed runner binds them to its deployed workflow + // and adds any enabled legacy aliases when advertising capabilities to the task service. + rpc ListRegisteredTasks(google.protobuf.Empty) returns (workflows.v1.UnboundTaskIdentifiers); // InitializeWorker is called by the runner to initialize a worker runtime with some known metadata about the // runner. This allows us to add those metadata to any logs and traces emitted by the worker. @@ -71,6 +73,11 @@ service WorkerService { // task, and then respond with the result of the task execution, which can either be a computed task, or a failed // task. If the worker runtime crashes or becomes unreachable during task execution, the runner will treat the task // as failed with an unknown error. + // The runner preserves the executing task's scope. Workers dispatch by local name/version and resolve child scopes + // before returning them, defaulting to the executing task's scope rather than the release's workflow ID. + // During migration, the managed runner fills missing child workflow IDs from the executing task's scope for legacy + // workers, preserving explicit IDs. An explicit unscoped child of a scoped task cannot be expressed through this + // adapter; it must be removed before supporting that override. The task service does not perform inheritance. rpc ExecuteTask(workflows.v1.Task) returns (ExecuteTaskResponse) {} // WatchLogs streams structured runtime logs to the local runner, independently of API log exports and raw diff --git a/apis/workflows/v1/workflows.proto b/apis/workflows/v1/workflows.proto index 76954f9..c5bc9a0 100644 --- a/apis/workflows/v1/workflows.proto +++ b/apis/workflows/v1/workflows.proto @@ -253,12 +253,19 @@ message CreateWorkflowRequest { features.field_presence = EXPLICIT, (buf.validate.field).string.max_len = 1000000 ]; + // Whether managed runners must also advertise unscoped task aliases for this workflow. + // Omission defaults to true for older clients. Set false explicitly to create a scoped-only workflow. + // Scoped task addresses are available regardless of this value. Needed during the migration to scoped task addresses, + // and this flag will be retired once all deployed runners support scoped tasks. + bool legacy_task_aliases = 5 [features.field_presence = EXPLICIT]; } // Workflow represents a logical grouping for a set of tasks. It can have multiple releases, each comprising a different // set of tasks. message Workflow { - reserved 1; // reserved for workflow ID, which is currently not exposed in the API but may be added in the future. + // The immutable workflow ID used to scope task identifiers. Only used to resolve TaskIdentifier.workflow_id for + // scoped tasks in the runner. Not used for user-facing addressing, which is always by workflow reference. + tilebox.v1.ID id = 1; // Allocated remote address: owner-local, or namespace:reference for another namespace's workflow. // Public listings always qualify references. Pass this value to workflow_reference request fields. string reference = 2; @@ -272,6 +279,9 @@ message Workflow { repeated WorkflowRelease releases = 5; // additional presentation metadata for public workflows, to display in the public workflow gallery. WorkflowPresentation presentation = 7; + // Whether managed runners must advertise unscoped task aliases in addition to workflow-scoped addresses. + // Unscoped task aliases must not conflict with other deployed task addresses on the same cluster. + bool legacy_task_aliases = 10 [features.field_presence = EXPLICIT]; } // WorkflowPresentation is an extended metadata message for a workflow, providing additional presentation properties @@ -308,8 +318,9 @@ message ReleaseContent { // A fingerprint of the release content, which is a hash of the task identifiers, files, runner object path, // command override, and build config. Computed by the client for idempotent builds. string fingerprint = 1 [(buf.validate.field).string.pattern = "^[a-f0-9]{64}$"]; - // List of task identifiers included in the release artifact. - repeated TaskIdentifier tasks = 2 [(buf.validate.field).repeated.min_items = 1]; + // Unbound task definitions included in the release artifact, without workflow IDs in both requests and responses. + // The managed runner binds them to the owning workflow when advertising capabilities; release content stays unbound. + repeated UnboundTaskIdentifier tasks = 2 [(buf.validate.field).repeated.min_items = 1]; // List of files/directories included in the release artifact. repeated Path files = 3; // The python module/object path to the runner instance, e.g. "my_module.my_runner:runner".