Scoped task identifiers

CHANGELOG.md

@@ -7,8 +7,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- `tilebox-workflows`: Add `workflow` selection by reference or object to job submission and automation creation,
+  and `workflows` filters to job queries. Omitting scope preserves unscoped direct-runner behavior.
+- `tilebox-workflows`: Preserve task workflow IDs and inherit the executing parent's scope for children.
+  Keep child scope inheritance internal to the runner and task discovery and release content unbound.
+- `tilebox-workflows`: Expose workflow IDs, references, project slugs, and presence-aware legacy alias policies.
+  Retain `.slug` compatibility.
+
 ### Changed
 
+- `tilebox-workflows`: Always create scoped-only workflows with legacy task aliases disabled.
+- `tilebox-workflows`: Use `UnboundTaskIdentifier` for task metadata, runner registries, and release contents;
+  keep execution scope in `TaskIdentifier` and resolve each future task's scope before merging submissions.
+- `tilebox-workflows`: Rename the `Workflow` constructor field `slug` to `reference`; retain `.slug` as a read-only
+  alias. Order workflows by reference, name, and ID (missing IDs first), ignoring other metadata when sorting.
 - `tilebox-core`: New package to own shared Connect RPC infrastructure and protobuf-py messages, public
   `tilebox.query` interval/pagination types, and `tilebox.errors` exceptions with a common `TileboxError` base.
 - `tilebox-grpc`: Retain an empty compatibility package depending on `tilebox-core>=0.67.0` for one release.

tilebox-workflows/README.md

@@ -75,6 +75,41 @@ runner = client.runner(tasks=[MyFirstTask])
 runner.run_all()
 ```
 
+## Workflow-scoped tasks
+
+To submit tasks to a managed workflow, pass its remote reference or a `Workflow` object:
+
+```python
+jobs.submit("scoped-job", MyFirstTask(), workflow="my-workflow")
+
+workflow = client.workflows().find("my-workflow")
+jobs.submit("another-job", MyFirstTask(), workflow=workflow)
+jobs.query(workflows=[workflow, "tilebox:shared-workflow"])
+```
+
+The workflow's immutable UUID scopes each root task. A failed reference lookup raises an error without submitting.
+Omitting `workflow` keeps roots unscoped, including jobs submitted from inside another task. Existing direct runners
+continue to register and execute unscoped tasks.
+
+Task metadata, runner registries, and release contents use `UnboundTaskIdentifier` (name and version).
+Submissions and executions use `TaskIdentifier`, which also carries the optional workflow ID. Use `.bind(workflow_id)`
+to create an execution identifier and `.unbind()` to recover its implementation identifier.
+
+Children inherit the executing parent's scope. Execution through a legacy unscoped alias therefore produces unscoped
+children. The runner handles inheritance internally; `context.submit_subtask()` and `context.submit_subtasks()`
+do not expose workflow selection. Cross-workflow and explicitly unscoped child overrides are deferred until the CLI
+can distinguish scope-aware workers from workers that need its compatibility adapter.
+
+Both `create_cron_automation()` and `create_storage_event_automation()` accept the same `workflow` parameter as
+`jobs.submit()`. Workflow objects expose `id`, `reference`, `project_slug`, and `legacy_task_aliases`; `.slug` remains
+a read-only alias for the remote reference. When constructing a `Workflow` directly, use `reference=` instead of
+`slug=`. The project slug is not a unique remote selector. Workflows sort by reference, name, and ID, with missing
+IDs first; other metadata does not affect ordering.
+
+`client.workflows().create(name, summary)` always creates a scoped-only workflow; creating legacy workflows is not
+supported. Existing workflows still expose their alias policy. A `None` policy in a response means the server did not
+report that field, rather than reporting `False`.
+
 ## Concurrent worker execution
 
 A worker runtime can execute multiple tasks concurrently in one Python process. Each execution receives a newly

tilebox-workflows/tests/automations/test_client.py

@@ -1,6 +1,7 @@
 from unittest.mock import MagicMock
 from uuid import UUID, uuid4
 
+import pytest
 from hypothesis.stateful import Bundle, RuleBasedStateMachine, consumes, rule
 from hypothesis.strategies import lists
 from protobuf.wkt.google.protobuf.empty_pb import Empty
@@ -15,14 +16,20 @@
     AutomationPrototype,
     CronTrigger,
     StorageEventTrigger,
+    StorageLocation,
+    StorageType,
     TaskIdentifier,
+    Workflow,
     uuid_to_uuid_message,
 )
 from tilebox.workflows.v1.automation_connect import AutomationServiceClientSync
 from tilebox.workflows.v1.automation_pb import AutomationPrototype as AutomationPrototypeMessage
 from tilebox.workflows.v1.automation_pb import Automations, DeleteAutomationRequest
 from tilebox.workflows.v1.automation_pb import CronTrigger as CronTriggerMessage
 from tilebox.workflows.v1.automation_pb import StorageEventTrigger as StorageEventTriggerMessage
+from tilebox.workflows.v1.workflows_connect import WorkflowsServiceClientSync
+from tilebox.workflows.v1.workflows_pb import GetWorkflowRequest
+from tilebox.workflows.v1.workflows_pb import Workflow as WorkflowMessage
 
 
 class MockAutomationService:
@@ -77,6 +84,44 @@ def list_automations(self, req: Empty) -> Automations:
         return Automations(automations=list(self.automations.values()))
 
 
+@pytest.mark.parametrize("kind", ["cron", "storage"])
+@pytest.mark.parametrize("selection", ["reference", "object", "unscoped"])
+def test_automation_scope_survives_creation_and_reads(kind: str, selection: str) -> None:
+    class ScheduledTask(CronTask):
+        pass
+
+    class StorageTask(StorageEventTask):
+        pass
+
+    workflow = Workflow("owner:workflow-123", "Workflow", "", id=uuid4(), project_slug="workflow")
+
+    def find(request: GetWorkflowRequest) -> WorkflowMessage:
+        if request.workflow_reference != workflow.reference:
+            raise NotFoundError("Workflow not found")
+        return workflow.to_message()
+
+    workflow_service = MagicMock(spec=WorkflowsServiceClientSync)
+    workflow_service.get_workflow.side_effect = find
+    mock_service = MockAutomationService()
+    client = AutomationClient(MagicMock(spec=AutomationServiceClientSync, wraps=mock_service), workflow_service)
+    selected = workflow.reference if selection == "reference" else workflow if selection == "object" else None
+    if kind == "cron":
+        created = client.create_cron_automation("Scheduled", ScheduledTask(), "0 * * * *", workflow=selected)
+    else:
+        storage = StorageLocation(uuid4(), "test-bucket", StorageType.AWS_S3)
+        created = client.create_storage_event_automation(
+            "Storage", StorageTask(), (storage, "*.tif"), workflow=selected
+        )
+    expected = workflow.id if selected is not None else None
+    assert created.prototype.identifier.workflow_id == expected
+    assert client.find(created.id).prototype.identifier.workflow_id == expected
+    assert client.all()[0].prototype.identifier.workflow_id == expected
+    stored = mock_service.automations[created.id].prototype
+    assert stored is not None
+    assert stored.identifier is not None
+    assert stored.identifier.workflow_id == uuid_to_uuid_message(expected)
+
+
 class AutomationCRUDOperations(RuleBasedStateMachine):
     """
     A state machine that tests the CRUD operations of the Automations client.

tilebox-workflows/tests/jobs/test_client.py

@@ -18,6 +18,7 @@
     LogSeverity,
     LogSeverityLiteral,
     Span,
+    Workflow,
     uuid_message_to_uuid,
     uuid_to_uuid_message,
 )
@@ -40,6 +41,9 @@
     VisualizeJobRequest,
 )
 from tilebox.workflows.v1.telemetry_pb import PaginatedLogsData, PaginatedSpansData
+from tilebox.workflows.v1.workflows_connect import WorkflowsServiceClientSync
+from tilebox.workflows.v1.workflows_pb import GetWorkflowRequest
+from tilebox.workflows.v1.workflows_pb import Workflow as WorkflowMessage
 
 
 class DummyTask(Task):
@@ -190,8 +194,10 @@ class MockJobService:
     def __init__(self) -> None:
         self.jobs: dict[UUID, JobMessage] = {}
         self.query_requests: list[QueryJobsRequest] = []
+        self.submit_requests: list[SubmitJobRequest] = []
 
     def submit_job(self, req: SubmitJobRequest) -> JobMessage:
+        self.submit_requests.append(req)
         job_id = uuid4()
         job = JobMessage(
             id=uuid_to_uuid_message(job_id),
@@ -246,16 +252,83 @@ def query_jobs(self, req: QueryJobsRequest) -> QueryJobsResponse:
         return QueryJobsResponse(jobs=list(self.jobs.values()))
 
 
-def test_query_filters_by_clusters() -> None:
+@pytest.mark.parametrize("selection", ["reference", "object", "unscoped"])
+def test_submit_workflow_scope(selection: str) -> None:
+    class AnotherTask(Task):
+        def execute(self, context: ExecutionContext) -> None:
+            _ = context
+
+    workflow = Workflow("owner:workflow-123", "Workflow", "", id=uuid4(), project_slug="workflow")
+
+    def find(request: GetWorkflowRequest) -> WorkflowMessage:
+        if request.workflow_reference != workflow.reference:
+            raise NotFoundError("Workflow not found")
+        return workflow.to_message()
+
+    workflow_service = MagicMock(spec=WorkflowsServiceClientSync)
+    workflow_service.get_workflow.side_effect = find
+    mock_service = MockJobService()
+    client = JobClient(
+        MagicMock(spec=JobServiceClientSync, wraps=mock_service), MagicMock(), NoopWorkflowTracer(), workflow_service
+    )
+    selected = workflow.reference if selection == "reference" else workflow if selection == "object" else None
+    client.submit("scoped job", [DummyTask(), AnotherTask()], cluster=["a", "b"], workflow=selected)
+    submissions = mock_service.submit_requests[-1].tasks
+    assert submissions is not None
+    assert len(submissions.identifier_lookup) == 2
+    assert {uuid_message_to_optional_uuid(identifier.workflow_id) for identifier in submissions.identifier_lookup} == {
+        workflow.id if selected is not None else None
+    }
+    # A subsequent ordinary submission must not inherit a previous submission's scope.
+    client.submit("direct job", DummyTask())
+    direct = mock_service.submit_requests[-1].tasks
+    assert direct is not None
+    assert direct.identifier_lookup[0].workflow_id is None
+
+
+@pytest.mark.parametrize("selection", ["missing", "empty", "missing-id", "nil-id"])
+def test_submit_invalid_workflow_never_falls_back_to_unscoped(selection: str) -> None:
+    mock_service = MockJobService()
+    workflow_service = MagicMock(spec=WorkflowsServiceClientSync)
+    workflow_service.get_workflow.side_effect = NotFoundError("Workflow not found")
+    selected = {
+        "missing": "missing",
+        "empty": "",
+        "missing-id": Workflow("old", "Old", ""),
+        "nil-id": Workflow("invalid", "Invalid", "", id=UUID(int=0)),
+    }[selection]
+    client = JobClient(
+        MagicMock(spec=JobServiceClientSync, wraps=mock_service), MagicMock(), NoopWorkflowTracer(), workflow_service
+    )
+    with pytest.raises((NotFoundError, ValueError)):
+        client.submit("must not submit", DummyTask(), workflow=selected)
+    assert not mock_service.jobs
+    assert not mock_service.submit_requests
+
+
+@pytest.mark.parametrize(
+    ("workflows", "expected"),
+    [
+        (None, []),
+        ([], []),
+        ("owner:one", ["owner:one"]),
+        (Workflow("one-123", "One", "", project_slug="one"), ["one-123"]),
+        (["owner:one", Workflow("two-456", "Two", "", project_slug="two")], ["owner:one", "two-456"]),
+    ],
+)
+def test_query_filters_by_clusters_and_workflows(
+    workflows: Workflow | str | list[Workflow | str] | None, expected: list[str]
+) -> None:
     mock_service = MockJobService()
     service = MagicMock(spec=JobServiceClientSync, wraps=mock_service)
     job_client = JobClient(service, MagicMock(), NoopWorkflowTracer())
 
-    job_client.query((uuid4(), uuid4()), clusters=["cluster-a", "cluster-b"])
+    job_client.query((uuid4(), uuid4()), clusters=["cluster-a", "cluster-b"], workflows=workflows)
 
     filters = mock_service.query_requests[-1].filters
     assert filters is not None
     assert list(filters.cluster_slugs) == ["cluster-a", "cluster-b"]
+    assert list(filters.workflow_references) == expected
 
 
 @pytest.mark.parametrize("use_interval", [False, True])

tilebox-workflows/tests/runner/test_worker_concurrency.py

@@ -11,6 +11,7 @@
 import pytest
 from protobuf.wkt.google.protobuf.empty_pb import Empty
 
+from _tilebox.uuid import uuid_message_to_optional_uuid
 from tilebox.datasets.uuid import must_uuid_to_uuid_message
 from tilebox.workflows import ExecutionContext, Runner, Task
 from tilebox.workflows.cache import InMemoryCache, JobCache
@@ -23,6 +24,7 @@
 from tilebox.workflows.task import TaskMeta
 from tilebox.workflows.v1 import core_pb, worker_pb
 from tilebox.workflows.v1.worker_connect import WorkerServiceClientSync
+from tilebox.workflows.v1.workflows_pb import Workflow as WorkflowMessage
 
 
 def test_worker_executes_tasks_concurrently_with_isolated_execution_state(
@@ -127,6 +129,108 @@ async def execute(self, context: ExecutionContext) -> None:
     assert {attributes["task_id"] for attributes in log_attributes} == task_ids
 
 
+def test_worker_preserves_execution_scope_across_concurrent_descendants(  # noqa: PLR0915
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    scopes = [uuid4(), uuid4(), None]
+    barrier = threading.Barrier(len(scopes))
+
+    class Grandchild(Task):
+        def execute(self, context: ExecutionContext) -> None:
+            context.current_task.display = str(context.current_task.identifier.workflow_id)
+
+    class Child(Task):
+        def execute(self, context: ExecutionContext) -> None:
+            future = context.submit_subtask(Grandchild())
+            assert future.identifier().workflow_id == context.current_task.identifier.workflow_id
+
+    class Parent(Task):
+        def execute(self, context: ExecutionContext) -> None:
+            barrier.wait(timeout=5)
+            context.current_task.display = str(context.current_task.identifier.workflow_id)
+            futures = [context.submit_subtask(Child()), *context.submit_subtasks([Child(), Child()])]
+            assert all(
+                future.identifier().workflow_id == context.current_task.identifier.workflow_id for future in futures
+            )
+
+    fake_client = MagicMock()
+    fake_client._tracer = NoopWorkflowTracer()
+    fake_client._auth = {"url": "https://worker.example", "token": "worker-key"}
+    fake_client._task_logger = StructuredLogger(logging.getLogger("tilebox.workflows.tests.scoped-worker"))
+    monkeypatch.setenv("TILEBOX_RUNTIME_DIR", str(tmp_path))
+    monkeypatch.setenv("TILEBOX_RUNTIME_TOKEN", "test-token")
+    server = WorkerServer(Runner(tasks=[Parent, Child, Grandchild]), "127.0.0.1:0")
+    headers = {"authorization": "Bearer test-token"}
+
+    with (
+        patch("tilebox.workflows.runner.worker_service.Client", return_value=fake_client),
+        patch("tilebox.workflows.runner.worker_service.initialize_logging"),
+    ):
+        server.start()
+        try:
+            with WorkerServiceClientSync(f"http://127.0.0.1:{server._port}", timeout_ms=10000) as worker:
+                discovered = worker.list_registered_tasks(Empty(), headers=headers)
+                assert len(discovered.identifiers) == 3
+                assert all(
+                    isinstance(identifier, core_pb.UnboundTaskIdentifier) for identifier in discovered.identifiers
+                )
+                worker.initialize_worker(
+                    worker_pb.InitializeRunnerRequest(
+                        runner_id=must_uuid_to_uuid_message(uuid4()),
+                        # Release metadata must never override the executing task's scope, especially legacy aliases.
+                        workflow=WorkflowMessage(id=must_uuid_to_uuid_message(uuid4()), legacy_task_aliases=True),
+                    ),
+                    headers=headers,
+                )
+                job = _job()
+                parents = [_task_message(Parent(), job) for _ in scopes]
+                for parent, scope in zip(parents, scopes, strict=True):
+                    assert parent.identifier is not None
+                    parent.identifier.workflow_id = must_uuid_to_uuid_message(scope) if scope is not None else None
+                with ThreadPoolExecutor(max_workers=len(scopes)) as pool:
+                    futures = [pool.submit(worker.execute_task, parent, headers=headers) for parent in parents]
+                    responses = [future.result(timeout=10) for future in futures]
+                for response, scope in zip(responses, scopes, strict=True):
+                    computed = response.computed_task
+                    assert computed is not None
+                    assert computed.display == str(scope)
+                    children = computed.sub_tasks
+                    assert children is not None
+                    child_scopes = []
+                    for group in children.task_groups:
+                        for task_input, pointer in zip(group.inputs, group.identifier_pointers, strict=True):
+                            identifier = children.identifier_lookup[pointer]
+                            child_scope = uuid_message_to_optional_uuid(identifier.workflow_id)
+                            child_scopes.append(child_scope)
+                            child = core_pb.Task(
+                                id=must_uuid_to_uuid_message(uuid4()),
+                                identifier=identifier,
+                                input=task_input,
+                                job=job.to_message(),
+                            )
+                            result = worker.execute_task(child, headers=headers).computed_task
+                            assert result is not None
+                            assert result.sub_tasks is not None
+                            grandchildren = result.sub_tasks
+                            grandchild_identifier = grandchildren.identifier_lookup[0]
+                            assert uuid_message_to_optional_uuid(grandchild_identifier.workflow_id) == child_scope
+                            grandchild = core_pb.Task(
+                                id=must_uuid_to_uuid_message(uuid4()),
+                                identifier=grandchild_identifier,
+                                input=grandchildren.task_groups[0].inputs[0],
+                                job=job.to_message(),
+                            )
+                            final = worker.execute_task(grandchild, headers=headers).computed_task
+                            assert final is not None
+                            assert final.display == str(child_scope)
+                    assert child_scopes == [scope, scope, scope]
+                assert worker.list_registered_tasks(Empty(), headers=headers) == discovered
+                worker.shutdown_worker(Empty(), headers=headers)
+        finally:
+            server.shutdown()
+            server.wait()
+
+
 def test_lazy_storage_locations_are_loaded_once_during_concurrent_access() -> None:
     storage_location = MagicMock()
     storage_location.id = UUID(int=1)
@@ -180,7 +284,7 @@ def _job() -> Job:
 
 
 def _task_message(task: Task, job: Job) -> core_pb.Task:
-    identifier = TaskMeta.for_task(task).identifier
+    identifier = TaskMeta.for_task(task).identifier.bind()
     return TaskData(
         id=uuid4(),
         identifier=identifier,