diff --git a/.golangci.yaml b/.golangci.yaml
index a9a0658..72ed2f9 100644
--- a/.golangci.yaml
+++ b/.golangci.yaml
@@ -63,7 +63,7 @@ linters:
     - iface # Detect the incorrect use of interfaces, helping developers avoid interface pollution.
     - importas # Enforces consistent import aliases.
     #- inamedparam # Reports interfaces with unnamed method parameters.
-    - interfacebloat # Checks the number of methods in an interface
+    #- interfacebloat # Checks the number of methods in an interface
     - intrange # finds places where for loops could make use of an integer range
     - iotamixing # checks that iota is not mixed with other constants
     #- ireturn # Accept Interfaces, Return Concrete Types
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0e94a7b..73e18a8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.9.0] - 2026-06-25
+
+### Added
+
+- `workflows`: Added `client.Workflows.Delete` and `client.Workflows.UnpublishRelease`.
+- `workflows`: Added `job.WithClusterSlugs(...)` to filter jobs by cluster slug (any task in that cluster).
+
+### Fixed
+
+- `workflows`: Fixed `PollingTaskRunner` retry handling so request-shaped `NextTask` and `TaskFailed` RPC errors do not cause endless retry loops.
+
 ## [0.8.0] - 2026-06-05
 
 ### Added
@@ -128,7 +139,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added support for Tilebox Observability, including logging and tracing helpers.
 - Added examples for using the library.
 
-[Unreleased]: https://github.com/tilebox/tilebox-go/compare/v0.8.0...HEAD
+[Unreleased]: https://github.com/tilebox/tilebox-go/compare/v0.9.0...HEAD
+[0.9.0]: https://github.com/tilebox/tilebox-go/compare/v0.8.0...v0.9.0
 [0.8.0]: https://github.com/tilebox/tilebox-go/compare/v0.7.1...v0.8.0
 [0.7.1]: https://github.com/tilebox/tilebox-go/compare/v0.7.0...v0.7.1
 [0.7.0]: https://github.com/tilebox/tilebox-go/compare/v0.6.0...v0.7.0
diff --git a/go.mod b/go.mod
index 5093f5b..31030f6 100644
--- a/go.mod
+++ b/go.mod
@@ -43,10 +43,10 @@ require (
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 // indirect
 	go.opentelemetry.io/otel/log v0.20.0 // indirect
-	golang.org/x/net v0.55.0 // indirect
-	golang.org/x/sys v0.45.0 // indirect
-	golang.org/x/text v0.37.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect
+	golang.org/x/net v0.56.0 // indirect
+	golang.org/x/sys v0.46.0 // indirect
+	golang.org/x/text v0.38.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20260622175928-b703f567277d // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260622175928-b703f567277d // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
diff --git a/go.sum b/go.sum
index 73a8b52..cc724e4 100644
--- a/go.sum
+++ b/go.sum
@@ -87,18 +87,18 @@ go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpu
 go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
-golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8=
-golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww=
-golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
-golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
-golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
-golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
+golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o=
+golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec=
+golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw=
+golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/text v0.38.0 h1:sXmwo9DwP3OK9EZ7PqAdaooSGozfl/3a6/xJcbzPRhE=
+golang.org/x/text v0.38.0/go.mod h1:YXZt3QhHUKYT53r2lLKFIVi6Ao1jdzrTR/KQ09qyxF4=
 gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
 gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
-google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8=
-google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
+google.golang.org/genproto/googleapis/api v0.0.0-20260622175928-b703f567277d h1:xr2lwHI91bn3UiXcnyzRMQjp2LRiM8wEHzwUaE0YhTs=
+google.golang.org/genproto/googleapis/api v0.0.0-20260622175928-b703f567277d/go.mod h1:O0ZOWSrfWfJ+Z5HbwZ+wNtHsg/vk1k2C/w67eww8PfQ=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260622175928-b703f567277d h1:mpAgMyM9vQHxycBlDq50y1VHpfSfVwzXvrQKtYbXuUY=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260622175928-b703f567277d/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
 google.golang.org/grpc v1.81.1 h1:VnnIIZ88UzOOKLukQi+ImGz8O1Wdp8nAGGnvOfEIWQQ=
 google.golang.org/grpc v1.81.1/go.mod h1:xGH9GfzOyMTGIOXBJmXt+BX/V0kcdQbdcuwQ/zNw42I=
 google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
diff --git a/internal/span/span.go b/internal/span/span.go
index b766fe0..8458304 100644
--- a/internal/span/span.go
+++ b/internal/span/span.go
@@ -51,9 +51,16 @@ func GetTraceParentOfCurrentSpan(ctx context.Context) string {
 	return traceparent
 }
 
-func StartJobSpan[Result any](ctx context.Context, tracer oteltrace.Tracer, spanName string, job *workflowsv1.Job, f func(ctx context.Context) (Result, error)) (Result, error) {
-	carrier := propagation.MapCarrier{"traceparent": job.GetTraceParent()}
-	ctx = propagator.Extract(ctx, carrier)
+// ContextWithTraceParent returns a context carrying the remote span context encoded in traceParent.
+// It extracts trace correlation only and does not start or record a new span.
+func ContextWithTraceParent(ctx context.Context, traceParent string) context.Context {
+	if traceParent == "" {
+		return ctx
+	}
 
-	return observability.WithSpanResult(ctx, tracer, spanName, f)
+	return propagator.Extract(ctx, propagation.MapCarrier{"traceparent": traceParent})
+}
+
+func StartJobSpan[Result any](ctx context.Context, tracer oteltrace.Tracer, spanName string, job *workflowsv1.Job, f func(ctx context.Context) (Result, error)) (Result, error) {
+	return observability.WithSpanResult(ContextWithTraceParent(ctx, job.GetTraceParent()), tracer, spanName, f)
 }
diff --git a/protogen/workflows/v1/job.pb.go b/protogen/workflows/v1/job.pb.go
index 69bd05b..5c94ef7 100644
--- a/protogen/workflows/v1/job.pb.go
+++ b/protogen/workflows/v1/job.pb.go
@@ -744,6 +744,7 @@ type QueryFilters struct {
 	xxx_hidden_States        []JobState             `protobuf:"varint,4,rep,packed,name=states,enum=workflows.v1.JobState"`
 	xxx_hidden_Name          string                 `protobuf:"bytes,5,opt,name=name"`
 	xxx_hidden_TaskStates    []TaskState            `protobuf:"varint,6,rep,packed,name=task_states,json=taskStates,enum=workflows.v1.TaskState"`
+	xxx_hidden_ClusterSlugs  []string               `protobuf:"bytes,7,rep,name=cluster_slugs,json=clusterSlugs"`
 	unknownFields            protoimpl.UnknownFields
 	sizeCache                protoimpl.SizeCache
 }
@@ -817,6 +818,13 @@ func (x *QueryFilters) GetTaskStates() []TaskState {
 	return nil
 }
 
+func (x *QueryFilters) GetClusterSlugs() []string {
+	if x != nil {
+		return x.xxx_hidden_ClusterSlugs
+	}
+	return nil
+}
+
 func (x *QueryFilters) SetTimeInterval(v *v1.TimeInterval) {
 	x.xxx_hidden_TimeInterval = v
 }
@@ -841,6 +849,10 @@ func (x *QueryFilters) SetTaskStates(v []TaskState) {
 	x.xxx_hidden_TaskStates = v
 }
 
+func (x *QueryFilters) SetClusterSlugs(v []string) {
+	x.xxx_hidden_ClusterSlugs = v
+}
+
 func (x *QueryFilters) HasTimeInterval() bool {
 	if x == nil {
 		return false
@@ -877,6 +889,8 @@ type QueryFilters_builder struct {
 	Name string
 	// Filter jobs by their tasks (whether they have tasks in a certain state).
 	TaskStates []TaskState
+	// Filter jobs by their clusters (whether they have tasks on certain clusters).
+	ClusterSlugs []string
 }
 
 func (b0 QueryFilters_builder) Build() *QueryFilters {
@@ -889,6 +903,7 @@ func (b0 QueryFilters_builder) Build() *QueryFilters {
 	x.xxx_hidden_States = b.States
 	x.xxx_hidden_Name = b.Name
 	x.xxx_hidden_TaskStates = b.TaskStates
+	x.xxx_hidden_ClusterSlugs = b.ClusterSlugs
 	return m0
 }
 
@@ -1365,7 +1380,7 @@ const file_workflows_v1_job_proto_rawDesc = "" +
 	"\x06job_id\x18\x01 \x01(\v2\x0e.tilebox.v1.IDB\x06\xbaH\x03\xc8\x01\x01R\x05jobId\x12B\n" +
 	"\x0erender_options\x18\x02 \x01(\v2\x1b.workflows.v1.RenderOptionsR\rrenderOptions\x128\n" +
 	"\x05theme\x18\x03 \x01(\x0e2\".workflows.v1.WorkflowDiagramThemeR\x05theme\x12(\n" +
-	"\x10include_job_name\x18\x04 \x01(\bR\x0eincludeJobName\"\xe9\x02\n" +
+	"\x10include_job_name\x18\x04 \x01(\bR\x0eincludeJobName\"\x8e\x03\n" +
 	"\fQueryFilters\x12=\n" +
 	"\rtime_interval\x18\x01 \x01(\v2\x18.tilebox.v1.TimeIntervalR\ftimeInterval\x127\n" +
 	"\vid_interval\x18\x02 \x01(\v2\x16.tilebox.v1.IDIntervalR\n" +
@@ -1374,7 +1389,8 @@ const file_workflows_v1_job_proto_rawDesc = "" +
 	"\x06states\x18\x04 \x03(\x0e2\x16.workflows.v1.JobStateR\x06states\x12\x1b\n" +
 	"\x04name\x18\x05 \x01(\tB\a\xbaH\x04r\x02\x18dR\x04name\x128\n" +
 	"\vtask_states\x18\x06 \x03(\x0e2\x17.workflows.v1.TaskStateR\n" +
-	"taskStates:#\xbaH \"\x1e\n" +
+	"taskStates\x12#\n" +
+	"\rcluster_slugs\x18\a \x03(\tR\fclusterSlugs:#\xbaH \"\x1e\n" +
 	"\rtime_interval\n" +
 	"\vid_interval\x10\x00\"\xbd\x01\n" +
 	"\x10QueryJobsRequest\x124\n" +
diff --git a/protogen/workflows/v1/workflows.pb.go b/protogen/workflows/v1/workflows.pb.go
index e6bdb21..d7fe18d 100644
--- a/protogen/workflows/v1/workflows.pb.go
+++ b/protogen/workflows/v1/workflows.pb.go
@@ -518,6 +518,109 @@ func (b0 GetWorkflowRequest_builder) Build() *GetWorkflowRequest {
 	return m0
 }
 
+// DeleteWorkflowRequest deletes an existing workflow.
+type DeleteWorkflowRequest struct {
+	state                   protoimpl.MessageState `protogen:"opaque.v1"`
+	xxx_hidden_WorkflowSlug string                 `protobuf:"bytes,1,opt,name=workflow_slug,json=workflowSlug"`
+	unknownFields           protoimpl.UnknownFields
+	sizeCache               protoimpl.SizeCache
+}
+
+func (x *DeleteWorkflowRequest) Reset() {
+	*x = DeleteWorkflowRequest{}
+	mi := &file_workflows_v1_workflows_proto_msgTypes[8]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *DeleteWorkflowRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*DeleteWorkflowRequest) ProtoMessage() {}
+
+func (x *DeleteWorkflowRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_workflows_v1_workflows_proto_msgTypes[8]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+func (x *DeleteWorkflowRequest) GetWorkflowSlug() string {
+	if x != nil {
+		return x.xxx_hidden_WorkflowSlug
+	}
+	return ""
+}
+
+func (x *DeleteWorkflowRequest) SetWorkflowSlug(v string) {
+	x.xxx_hidden_WorkflowSlug = v
+}
+
+type DeleteWorkflowRequest_builder struct {
+	_ [0]func() // Prevents comparability and use of unkeyed literals for the builder.
+
+	// The slug of the workflow to delete.
+	WorkflowSlug string
+}
+
+func (b0 DeleteWorkflowRequest_builder) Build() *DeleteWorkflowRequest {
+	m0 := &DeleteWorkflowRequest{}
+	b, x := &b0, m0
+	_, _ = b, x
+	x.xxx_hidden_WorkflowSlug = b.WorkflowSlug
+	return m0
+}
+
+// DeleteWorkflowResponse is the response to DeleteWorkflowRequest.
+type DeleteWorkflowResponse struct {
+	state         protoimpl.MessageState `protogen:"opaque.v1"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *DeleteWorkflowResponse) Reset() {
+	*x = DeleteWorkflowResponse{}
+	mi := &file_workflows_v1_workflows_proto_msgTypes[9]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *DeleteWorkflowResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*DeleteWorkflowResponse) ProtoMessage() {}
+
+func (x *DeleteWorkflowResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_workflows_v1_workflows_proto_msgTypes[9]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+type DeleteWorkflowResponse_builder struct {
+	_ [0]func() // Prevents comparability and use of unkeyed literals for the builder.
+
+}
+
+func (b0 DeleteWorkflowResponse_builder) Build() *DeleteWorkflowResponse {
+	m0 := &DeleteWorkflowResponse{}
+	b, x := &b0, m0
+	_, _ = b, x
+	return m0
+}
+
 // PublishWorkflowReleaseRequest publishes a new workflow release artifact, and associates it with a workflow.
 type PublishWorkflowReleaseRequest struct {
 	state                   protoimpl.MessageState `protogen:"opaque.v1"`
@@ -530,7 +633,7 @@ type PublishWorkflowReleaseRequest struct {
 
 func (x *PublishWorkflowReleaseRequest) Reset() {
 	*x = PublishWorkflowReleaseRequest{}
-	mi := &file_workflows_v1_workflows_proto_msgTypes[8]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[10]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -542,7 +645,7 @@ func (x *PublishWorkflowReleaseRequest) String() string {
 func (*PublishWorkflowReleaseRequest) ProtoMessage() {}
 
 func (x *PublishWorkflowReleaseRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_workflows_v1_workflows_proto_msgTypes[8]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[10]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -626,6 +729,135 @@ func (b0 PublishWorkflowReleaseRequest_builder) Build() *PublishWorkflowReleaseR
 	return m0
 }
 
+// UnpublishWorkflowReleaseRequest unpublishes a specific workflow release.
+type UnpublishWorkflowReleaseRequest struct {
+	state                   protoimpl.MessageState `protogen:"opaque.v1"`
+	xxx_hidden_WorkflowSlug string                 `protobuf:"bytes,1,opt,name=workflow_slug,json=workflowSlug"`
+	xxx_hidden_ReleaseId    *v1.ID                 `protobuf:"bytes,2,opt,name=release_id,json=releaseId"`
+	unknownFields           protoimpl.UnknownFields
+	sizeCache               protoimpl.SizeCache
+}
+
+func (x *UnpublishWorkflowReleaseRequest) Reset() {
+	*x = UnpublishWorkflowReleaseRequest{}
+	mi := &file_workflows_v1_workflows_proto_msgTypes[11]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *UnpublishWorkflowReleaseRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*UnpublishWorkflowReleaseRequest) ProtoMessage() {}
+
+func (x *UnpublishWorkflowReleaseRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_workflows_v1_workflows_proto_msgTypes[11]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+func (x *UnpublishWorkflowReleaseRequest) GetWorkflowSlug() string {
+	if x != nil {
+		return x.xxx_hidden_WorkflowSlug
+	}
+	return ""
+}
+
+func (x *UnpublishWorkflowReleaseRequest) GetReleaseId() *v1.ID {
+	if x != nil {
+		return x.xxx_hidden_ReleaseId
+	}
+	return nil
+}
+
+func (x *UnpublishWorkflowReleaseRequest) SetWorkflowSlug(v string) {
+	x.xxx_hidden_WorkflowSlug = v
+}
+
+func (x *UnpublishWorkflowReleaseRequest) SetReleaseId(v *v1.ID) {
+	x.xxx_hidden_ReleaseId = v
+}
+
+func (x *UnpublishWorkflowReleaseRequest) HasReleaseId() bool {
+	if x == nil {
+		return false
+	}
+	return x.xxx_hidden_ReleaseId != nil
+}
+
+func (x *UnpublishWorkflowReleaseRequest) ClearReleaseId() {
+	x.xxx_hidden_ReleaseId = nil
+}
+
+type UnpublishWorkflowReleaseRequest_builder struct {
+	_ [0]func() // Prevents comparability and use of unkeyed literals for the builder.
+
+	// The slug of the workflow containing the release to unpublish.
+	WorkflowSlug string
+	// The ID of the release to unpublish.
+	ReleaseId *v1.ID
+}
+
+func (b0 UnpublishWorkflowReleaseRequest_builder) Build() *UnpublishWorkflowReleaseRequest {
+	m0 := &UnpublishWorkflowReleaseRequest{}
+	b, x := &b0, m0
+	_, _ = b, x
+	x.xxx_hidden_WorkflowSlug = b.WorkflowSlug
+	x.xxx_hidden_ReleaseId = b.ReleaseId
+	return m0
+}
+
+// UnpublishWorkflowReleaseResponse is the response to UnpublishWorkflowReleaseRequest.
+type UnpublishWorkflowReleaseResponse struct {
+	state         protoimpl.MessageState `protogen:"opaque.v1"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *UnpublishWorkflowReleaseResponse) Reset() {
+	*x = UnpublishWorkflowReleaseResponse{}
+	mi := &file_workflows_v1_workflows_proto_msgTypes[12]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *UnpublishWorkflowReleaseResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*UnpublishWorkflowReleaseResponse) ProtoMessage() {}
+
+func (x *UnpublishWorkflowReleaseResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_workflows_v1_workflows_proto_msgTypes[12]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+type UnpublishWorkflowReleaseResponse_builder struct {
+	_ [0]func() // Prevents comparability and use of unkeyed literals for the builder.
+
+}
+
+func (b0 UnpublishWorkflowReleaseResponse_builder) Build() *UnpublishWorkflowReleaseResponse {
+	m0 := &UnpublishWorkflowReleaseResponse{}
+	b, x := &b0, m0
+	_, _ = b, x
+	return m0
+}
+
 // ListWorkflowsRequest lists all workflows.
 type ListWorkflowsRequest struct {
 	state         protoimpl.MessageState `protogen:"opaque.v1"`
@@ -635,7 +867,7 @@ type ListWorkflowsRequest struct {
 
 func (x *ListWorkflowsRequest) Reset() {
 	*x = ListWorkflowsRequest{}
-	mi := &file_workflows_v1_workflows_proto_msgTypes[9]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[13]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -647,7 +879,7 @@ func (x *ListWorkflowsRequest) String() string {
 func (*ListWorkflowsRequest) ProtoMessage() {}
 
 func (x *ListWorkflowsRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_workflows_v1_workflows_proto_msgTypes[9]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[13]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -680,7 +912,7 @@ type ListWorkflowsResponse struct {
 
 func (x *ListWorkflowsResponse) Reset() {
 	*x = ListWorkflowsResponse{}
-	mi := &file_workflows_v1_workflows_proto_msgTypes[10]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[14]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -692,7 +924,7 @@ func (x *ListWorkflowsResponse) String() string {
 func (*ListWorkflowsResponse) ProtoMessage() {}
 
 func (x *ListWorkflowsResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_workflows_v1_workflows_proto_msgTypes[10]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[14]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -742,7 +974,7 @@ type CreateWorkflowRequest struct {
 
 func (x *CreateWorkflowRequest) Reset() {
 	*x = CreateWorkflowRequest{}
-	mi := &file_workflows_v1_workflows_proto_msgTypes[11]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[15]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -754,7 +986,7 @@ func (x *CreateWorkflowRequest) String() string {
 func (*CreateWorkflowRequest) ProtoMessage() {}
 
 func (x *CreateWorkflowRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_workflows_v1_workflows_proto_msgTypes[11]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[15]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -817,7 +1049,7 @@ type Workflow struct {
 
 func (x *Workflow) Reset() {
 	*x = Workflow{}
-	mi := &file_workflows_v1_workflows_proto_msgTypes[12]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[16]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -829,7 +1061,7 @@ func (x *Workflow) String() string {
 func (*Workflow) ProtoMessage() {}
 
 func (x *Workflow) ProtoReflect() protoreflect.Message {
-	mi := &file_workflows_v1_workflows_proto_msgTypes[12]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[16]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -913,13 +1145,14 @@ type WorkflowRelease struct {
 	xxx_hidden_Artifact  *Artifact              `protobuf:"bytes,2,opt,name=artifact"`
 	xxx_hidden_Content   *ReleaseContent        `protobuf:"bytes,3,opt,name=content"`
 	xxx_hidden_CreatedAt *timestamppb.Timestamp `protobuf:"bytes,4,opt,name=created_at,json=createdAt"`
+	xxx_hidden_Clusters  *[]*Cluster            `protobuf:"bytes,5,rep,name=clusters"`
 	unknownFields        protoimpl.UnknownFields
 	sizeCache            protoimpl.SizeCache
 }
 
 func (x *WorkflowRelease) Reset() {
 	*x = WorkflowRelease{}
-	mi := &file_workflows_v1_workflows_proto_msgTypes[13]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[17]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -931,7 +1164,7 @@ func (x *WorkflowRelease) String() string {
 func (*WorkflowRelease) ProtoMessage() {}
 
 func (x *WorkflowRelease) ProtoReflect() protoreflect.Message {
-	mi := &file_workflows_v1_workflows_proto_msgTypes[13]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[17]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -970,6 +1203,15 @@ func (x *WorkflowRelease) GetCreatedAt() *timestamppb.Timestamp {
 	return nil
 }
 
+func (x *WorkflowRelease) GetClusters() []*Cluster {
+	if x != nil {
+		if x.xxx_hidden_Clusters != nil {
+			return *x.xxx_hidden_Clusters
+		}
+	}
+	return nil
+}
+
 func (x *WorkflowRelease) SetId(v *v1.ID) {
 	x.xxx_hidden_Id = v
 }
@@ -986,6 +1228,10 @@ func (x *WorkflowRelease) SetCreatedAt(v *timestamppb.Timestamp) {
 	x.xxx_hidden_CreatedAt = v
 }
 
+func (x *WorkflowRelease) SetClusters(v []*Cluster) {
+	x.xxx_hidden_Clusters = &v
+}
+
 func (x *WorkflowRelease) HasId() bool {
 	if x == nil {
 		return false
@@ -1037,6 +1283,7 @@ type WorkflowRelease_builder struct {
 	Artifact  *Artifact
 	Content   *ReleaseContent
 	CreatedAt *timestamppb.Timestamp
+	Clusters  []*Cluster
 }
 
 func (b0 WorkflowRelease_builder) Build() *WorkflowRelease {
@@ -1047,6 +1294,7 @@ func (b0 WorkflowRelease_builder) Build() *WorkflowRelease {
 	x.xxx_hidden_Artifact = b.Artifact
 	x.xxx_hidden_Content = b.Content
 	x.xxx_hidden_CreatedAt = b.CreatedAt
+	x.xxx_hidden_Clusters = &b.Clusters
 	return m0
 }
 
@@ -1062,7 +1310,7 @@ type Artifact struct {
 
 func (x *Artifact) Reset() {
 	*x = Artifact{}
-	mi := &file_workflows_v1_workflows_proto_msgTypes[14]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[18]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -1074,7 +1322,7 @@ func (x *Artifact) String() string {
 func (*Artifact) ProtoMessage() {}
 
 func (x *Artifact) ProtoReflect() protoreflect.Message {
-	mi := &file_workflows_v1_workflows_proto_msgTypes[14]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[18]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1150,7 +1398,7 @@ type ReleaseContent struct {
 
 func (x *ReleaseContent) Reset() {
 	*x = ReleaseContent{}
-	mi := &file_workflows_v1_workflows_proto_msgTypes[15]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[19]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -1162,7 +1410,7 @@ func (x *ReleaseContent) String() string {
 func (*ReleaseContent) ProtoMessage() {}
 
 func (x *ReleaseContent) ProtoReflect() protoreflect.Message {
-	mi := &file_workflows_v1_workflows_proto_msgTypes[15]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[19]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1274,7 +1522,7 @@ type Path struct {
 
 func (x *Path) Reset() {
 	*x = Path{}
-	mi := &file_workflows_v1_workflows_proto_msgTypes[16]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[20]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -1286,7 +1534,7 @@ func (x *Path) String() string {
 func (*Path) ProtoMessage() {}
 
 func (x *Path) ProtoReflect() protoreflect.Message {
-	mi := &file_workflows_v1_workflows_proto_msgTypes[16]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[20]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1363,7 +1611,7 @@ type DeployWorkflowReleaseRequest struct {
 
 func (x *DeployWorkflowReleaseRequest) Reset() {
 	*x = DeployWorkflowReleaseRequest{}
-	mi := &file_workflows_v1_workflows_proto_msgTypes[17]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[21]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -1375,7 +1623,7 @@ func (x *DeployWorkflowReleaseRequest) String() string {
 func (*DeployWorkflowReleaseRequest) ProtoMessage() {}
 
 func (x *DeployWorkflowReleaseRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_workflows_v1_workflows_proto_msgTypes[17]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[21]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1460,7 +1708,7 @@ type DeployWorkflowReleaseResponse struct {
 
 func (x *DeployWorkflowReleaseResponse) Reset() {
 	*x = DeployWorkflowReleaseResponse{}
-	mi := &file_workflows_v1_workflows_proto_msgTypes[18]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[22]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -1472,7 +1720,7 @@ func (x *DeployWorkflowReleaseResponse) String() string {
 func (*DeployWorkflowReleaseResponse) ProtoMessage() {}
 
 func (x *DeployWorkflowReleaseResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_workflows_v1_workflows_proto_msgTypes[18]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[22]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1547,7 +1795,7 @@ type UndeployWorkflowReleaseRequest struct {
 
 func (x *UndeployWorkflowReleaseRequest) Reset() {
 	*x = UndeployWorkflowReleaseRequest{}
-	mi := &file_workflows_v1_workflows_proto_msgTypes[19]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[23]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -1559,7 +1807,7 @@ func (x *UndeployWorkflowReleaseRequest) String() string {
 func (*UndeployWorkflowReleaseRequest) ProtoMessage() {}
 
 func (x *UndeployWorkflowReleaseRequest) ProtoReflect() protoreflect.Message {
-	mi := &file_workflows_v1_workflows_proto_msgTypes[19]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[23]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1644,7 +1892,7 @@ type UndeployWorkflowReleaseResponse struct {
 
 func (x *UndeployWorkflowReleaseResponse) Reset() {
 	*x = UndeployWorkflowReleaseResponse{}
-	mi := &file_workflows_v1_workflows_proto_msgTypes[20]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[24]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -1656,7 +1904,7 @@ func (x *UndeployWorkflowReleaseResponse) String() string {
 func (*UndeployWorkflowReleaseResponse) ProtoMessage() {}
 
 func (x *UndeployWorkflowReleaseResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_workflows_v1_workflows_proto_msgTypes[20]
+	mi := &file_workflows_v1_workflows_proto_msgTypes[24]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -1739,12 +1987,20 @@ const file_workflows_v1_workflows_proto_rawDesc = "" +
 	"\x14ListClustersResponse\x121\n" +
 	"\bclusters\x18\x01 \x03(\v2\x15.workflows.v1.ClusterR\bclusters\"B\n" +
 	"\x12GetWorkflowRequest\x12,\n" +
-	"\rworkflow_slug\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\fworkflowSlug\"\xc6\x01\n" +
+	"\rworkflow_slug\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\fworkflowSlug\"E\n" +
+	"\x15DeleteWorkflowRequest\x12,\n" +
+	"\rworkflow_slug\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\fworkflowSlug\"\x18\n" +
+	"\x16DeleteWorkflowResponse\"\xc6\x01\n" +
 	"\x1dPublishWorkflowReleaseRequest\x12,\n" +
 	"\rworkflow_slug\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\fworkflowSlug\x127\n" +
 	"\vartifact_id\x18\x02 \x01(\v2\x0e.tilebox.v1.IDB\x06\xbaH\x03\xc8\x01\x01R\n" +
 	"artifactId\x12>\n" +
-	"\acontent\x18\x03 \x01(\v2\x1c.workflows.v1.ReleaseContentB\x06\xbaH\x03\xc8\x01\x01R\acontent\"\x16\n" +
+	"\acontent\x18\x03 \x01(\v2\x1c.workflows.v1.ReleaseContentB\x06\xbaH\x03\xc8\x01\x01R\acontent\"\x86\x01\n" +
+	"\x1fUnpublishWorkflowReleaseRequest\x12,\n" +
+	"\rworkflow_slug\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\fworkflowSlug\x125\n" +
+	"\n" +
+	"release_id\x18\x02 \x01(\v2\x0e.tilebox.v1.IDB\x06\xbaH\x03\xc8\x01\x01R\treleaseId\"\"\n" +
+	" UnpublishWorkflowReleaseResponse\"\x16\n" +
 	"\x14ListWorkflowsRequest\"M\n" +
 	"\x15ListWorkflowsResponse\x124\n" +
 	"\tworkflows\x18\x01 \x03(\v2\x16.workflows.v1.WorkflowR\tworkflows\"V\n" +
@@ -1755,13 +2011,14 @@ const file_workflows_v1_workflows_proto_rawDesc = "" +
 	"\x04slug\x18\x02 \x01(\tR\x04slug\x12\x12\n" +
 	"\x04name\x18\x03 \x01(\tR\x04name\x12 \n" +
 	"\vdescription\x18\x04 \x01(\tR\vdescription\x129\n" +
-	"\breleases\x18\x05 \x03(\v2\x1d.workflows.v1.WorkflowReleaseR\breleasesJ\x04\b\x01\x10\x02\"\xd8\x01\n" +
+	"\breleases\x18\x05 \x03(\v2\x1d.workflows.v1.WorkflowReleaseR\breleasesJ\x04\b\x01\x10\x02\"\x8b\x02\n" +
 	"\x0fWorkflowRelease\x12\x1e\n" +
 	"\x02id\x18\x01 \x01(\v2\x0e.tilebox.v1.IDR\x02id\x122\n" +
 	"\bartifact\x18\x02 \x01(\v2\x16.workflows.v1.ArtifactR\bartifact\x126\n" +
 	"\acontent\x18\x03 \x01(\v2\x1c.workflows.v1.ReleaseContentR\acontent\x129\n" +
 	"\n" +
-	"created_at\x18\x04 \x01(\v2\x1a.google.protobuf.TimestampR\tcreatedAt\"Y\n" +
+	"created_at\x18\x04 \x01(\v2\x1a.google.protobuf.TimestampR\tcreatedAt\x121\n" +
+	"\bclusters\x18\x05 \x03(\v2\x15.workflows.v1.ClusterR\bclusters\"Y\n" +
 	"\bArtifact\x12\x1e\n" +
 	"\x02id\x18\x01 \x01(\v2\x0e.tilebox.v1.IDR\x02id\x12-\n" +
 	"\x06digest\x18\x02 \x01(\tB\x15\xbaH\x12r\x102\x0e^[a-f0-9]{64}$R\x06digest\"\x8a\x02\n" +
@@ -1790,7 +2047,7 @@ const file_workflows_v1_workflows_proto_rawDesc = "" +
 	"\rcluster_slugs\x18\x03 \x03(\tB\x0e\xbaH\v\x92\x01\b\b\x01\"\x04r\x02 \x01R\fclusterSlugs\"\x8d\x01\n" +
 	"\x1fUndeployWorkflowReleaseResponse\x127\n" +
 	"\arelease\x18\x01 \x01(\v2\x1d.workflows.v1.WorkflowReleaseR\arelease\x121\n" +
-	"\bclusters\x18\x02 \x03(\v2\x15.workflows.v1.ClusterR\bclusters2\x97\a\n" +
+	"\bclusters\x18\x02 \x03(\v2\x15.workflows.v1.ClusterR\bclusters2\xef\b\n" +
 	"\x10WorkflowsService\x12J\n" +
 	"\rCreateCluster\x12\".workflows.v1.CreateClusterRequest\x1a\x15.workflows.v1.Cluster\x12D\n" +
 	"\n" +
@@ -1799,85 +2056,97 @@ const file_workflows_v1_workflows_proto_rawDesc = "" +
 	"\fListClusters\x12!.workflows.v1.ListClustersRequest\x1a\".workflows.v1.ListClustersResponse\x12M\n" +
 	"\x0eCreateWorkflow\x12#.workflows.v1.CreateWorkflowRequest\x1a\x16.workflows.v1.Workflow\x12X\n" +
 	"\rListWorkflows\x12\".workflows.v1.ListWorkflowsRequest\x1a#.workflows.v1.ListWorkflowsResponse\x12G\n" +
-	"\vGetWorkflow\x12 .workflows.v1.GetWorkflowRequest\x1a\x16.workflows.v1.Workflow\x12d\n" +
-	"\x16PublishWorkflowRelease\x12+.workflows.v1.PublishWorkflowReleaseRequest\x1a\x1d.workflows.v1.WorkflowRelease\x12p\n" +
+	"\vGetWorkflow\x12 .workflows.v1.GetWorkflowRequest\x1a\x16.workflows.v1.Workflow\x12[\n" +
+	"\x0eDeleteWorkflow\x12#.workflows.v1.DeleteWorkflowRequest\x1a$.workflows.v1.DeleteWorkflowResponse\x12d\n" +
+	"\x16PublishWorkflowRelease\x12+.workflows.v1.PublishWorkflowReleaseRequest\x1a\x1d.workflows.v1.WorkflowRelease\x12y\n" +
+	"\x18UnpublishWorkflowRelease\x12-.workflows.v1.UnpublishWorkflowReleaseRequest\x1a..workflows.v1.UnpublishWorkflowReleaseResponse\x12p\n" +
 	"\x15DeployWorkflowRelease\x12*.workflows.v1.DeployWorkflowReleaseRequest\x1a+.workflows.v1.DeployWorkflowReleaseResponse\x12v\n" +
 	"\x17UndeployWorkflowRelease\x12,.workflows.v1.UndeployWorkflowReleaseRequest\x1a-.workflows.v1.UndeployWorkflowReleaseResponseB\xb9\x01\n" +
 	"\x10com.workflows.v1B\x0eWorkflowsProtoP\x01Z?github.com/tilebox/tilebox-go/protogen/workflows/v1;workflowsv1\xa2\x02\x03WXX\xaa\x02\fWorkflows.V1\xca\x02\fWorkflows\\V1\xe2\x02\x18Workflows\\V1\\GPBMetadata\xea\x02\rWorkflows::V1\x92\x03\x02\b\x02b\beditionsp\xe8\a"
 
-var file_workflows_v1_workflows_proto_msgTypes = make([]protoimpl.MessageInfo, 21)
+var file_workflows_v1_workflows_proto_msgTypes = make([]protoimpl.MessageInfo, 25)
 var file_workflows_v1_workflows_proto_goTypes = []any{
-	(*Cluster)(nil),                         // 0: workflows.v1.Cluster
-	(*CreateClusterRequest)(nil),            // 1: workflows.v1.CreateClusterRequest
-	(*GetClusterRequest)(nil),               // 2: workflows.v1.GetClusterRequest
-	(*DeleteClusterRequest)(nil),            // 3: workflows.v1.DeleteClusterRequest
-	(*DeleteClusterResponse)(nil),           // 4: workflows.v1.DeleteClusterResponse
-	(*ListClustersRequest)(nil),             // 5: workflows.v1.ListClustersRequest
-	(*ListClustersResponse)(nil),            // 6: workflows.v1.ListClustersResponse
-	(*GetWorkflowRequest)(nil),              // 7: workflows.v1.GetWorkflowRequest
-	(*PublishWorkflowReleaseRequest)(nil),   // 8: workflows.v1.PublishWorkflowReleaseRequest
-	(*ListWorkflowsRequest)(nil),            // 9: workflows.v1.ListWorkflowsRequest
-	(*ListWorkflowsResponse)(nil),           // 10: workflows.v1.ListWorkflowsResponse
-	(*CreateWorkflowRequest)(nil),           // 11: workflows.v1.CreateWorkflowRequest
-	(*Workflow)(nil),                        // 12: workflows.v1.Workflow
-	(*WorkflowRelease)(nil),                 // 13: workflows.v1.WorkflowRelease
-	(*Artifact)(nil),                        // 14: workflows.v1.Artifact
-	(*ReleaseContent)(nil),                  // 15: workflows.v1.ReleaseContent
-	(*Path)(nil),                            // 16: workflows.v1.Path
-	(*DeployWorkflowReleaseRequest)(nil),    // 17: workflows.v1.DeployWorkflowReleaseRequest
-	(*DeployWorkflowReleaseResponse)(nil),   // 18: workflows.v1.DeployWorkflowReleaseResponse
-	(*UndeployWorkflowReleaseRequest)(nil),  // 19: workflows.v1.UndeployWorkflowReleaseRequest
-	(*UndeployWorkflowReleaseResponse)(nil), // 20: workflows.v1.UndeployWorkflowReleaseResponse
-	(*v1.ID)(nil),                           // 21: tilebox.v1.ID
-	(*timestamppb.Timestamp)(nil),           // 22: google.protobuf.Timestamp
-	(*TaskIdentifier)(nil),                  // 23: workflows.v1.TaskIdentifier
+	(*Cluster)(nil),                          // 0: workflows.v1.Cluster
+	(*CreateClusterRequest)(nil),             // 1: workflows.v1.CreateClusterRequest
+	(*GetClusterRequest)(nil),                // 2: workflows.v1.GetClusterRequest
+	(*DeleteClusterRequest)(nil),             // 3: workflows.v1.DeleteClusterRequest
+	(*DeleteClusterResponse)(nil),            // 4: workflows.v1.DeleteClusterResponse
+	(*ListClustersRequest)(nil),              // 5: workflows.v1.ListClustersRequest
+	(*ListClustersResponse)(nil),             // 6: workflows.v1.ListClustersResponse
+	(*GetWorkflowRequest)(nil),               // 7: workflows.v1.GetWorkflowRequest
+	(*DeleteWorkflowRequest)(nil),            // 8: workflows.v1.DeleteWorkflowRequest
+	(*DeleteWorkflowResponse)(nil),           // 9: workflows.v1.DeleteWorkflowResponse
+	(*PublishWorkflowReleaseRequest)(nil),    // 10: workflows.v1.PublishWorkflowReleaseRequest
+	(*UnpublishWorkflowReleaseRequest)(nil),  // 11: workflows.v1.UnpublishWorkflowReleaseRequest
+	(*UnpublishWorkflowReleaseResponse)(nil), // 12: workflows.v1.UnpublishWorkflowReleaseResponse
+	(*ListWorkflowsRequest)(nil),             // 13: workflows.v1.ListWorkflowsRequest
+	(*ListWorkflowsResponse)(nil),            // 14: workflows.v1.ListWorkflowsResponse
+	(*CreateWorkflowRequest)(nil),            // 15: workflows.v1.CreateWorkflowRequest
+	(*Workflow)(nil),                         // 16: workflows.v1.Workflow
+	(*WorkflowRelease)(nil),                  // 17: workflows.v1.WorkflowRelease
+	(*Artifact)(nil),                         // 18: workflows.v1.Artifact
+	(*ReleaseContent)(nil),                   // 19: workflows.v1.ReleaseContent
+	(*Path)(nil),                             // 20: workflows.v1.Path
+	(*DeployWorkflowReleaseRequest)(nil),     // 21: workflows.v1.DeployWorkflowReleaseRequest
+	(*DeployWorkflowReleaseResponse)(nil),    // 22: workflows.v1.DeployWorkflowReleaseResponse
+	(*UndeployWorkflowReleaseRequest)(nil),   // 23: workflows.v1.UndeployWorkflowReleaseRequest
+	(*UndeployWorkflowReleaseResponse)(nil),  // 24: workflows.v1.UndeployWorkflowReleaseResponse
+	(*v1.ID)(nil),                            // 25: tilebox.v1.ID
+	(*timestamppb.Timestamp)(nil),            // 26: google.protobuf.Timestamp
+	(*TaskIdentifier)(nil),                   // 27: workflows.v1.TaskIdentifier
 }
 var file_workflows_v1_workflows_proto_depIdxs = []int32{
-	12, // 0: workflows.v1.Cluster.deployed_releases:type_name -> workflows.v1.Workflow
+	16, // 0: workflows.v1.Cluster.deployed_releases:type_name -> workflows.v1.Workflow
 	0,  // 1: workflows.v1.ListClustersResponse.clusters:type_name -> workflows.v1.Cluster
-	21, // 2: workflows.v1.PublishWorkflowReleaseRequest.artifact_id:type_name -> tilebox.v1.ID
-	15, // 3: workflows.v1.PublishWorkflowReleaseRequest.content:type_name -> workflows.v1.ReleaseContent
-	12, // 4: workflows.v1.ListWorkflowsResponse.workflows:type_name -> workflows.v1.Workflow
-	13, // 5: workflows.v1.Workflow.releases:type_name -> workflows.v1.WorkflowRelease
-	21, // 6: workflows.v1.WorkflowRelease.id:type_name -> tilebox.v1.ID
-	14, // 7: workflows.v1.WorkflowRelease.artifact:type_name -> workflows.v1.Artifact
-	15, // 8: workflows.v1.WorkflowRelease.content:type_name -> workflows.v1.ReleaseContent
-	22, // 9: workflows.v1.WorkflowRelease.created_at:type_name -> google.protobuf.Timestamp
-	21, // 10: workflows.v1.Artifact.id:type_name -> tilebox.v1.ID
-	23, // 11: workflows.v1.ReleaseContent.tasks:type_name -> workflows.v1.TaskIdentifier
-	16, // 12: workflows.v1.ReleaseContent.files:type_name -> workflows.v1.Path
-	16, // 13: workflows.v1.Path.children:type_name -> workflows.v1.Path
-	21, // 14: workflows.v1.DeployWorkflowReleaseRequest.release_id:type_name -> tilebox.v1.ID
-	13, // 15: workflows.v1.DeployWorkflowReleaseResponse.release:type_name -> workflows.v1.WorkflowRelease
-	0,  // 16: workflows.v1.DeployWorkflowReleaseResponse.clusters:type_name -> workflows.v1.Cluster
-	21, // 17: workflows.v1.UndeployWorkflowReleaseRequest.release_id:type_name -> tilebox.v1.ID
-	13, // 18: workflows.v1.UndeployWorkflowReleaseResponse.release:type_name -> workflows.v1.WorkflowRelease
-	0,  // 19: workflows.v1.UndeployWorkflowReleaseResponse.clusters:type_name -> workflows.v1.Cluster
-	1,  // 20: workflows.v1.WorkflowsService.CreateCluster:input_type -> workflows.v1.CreateClusterRequest
-	2,  // 21: workflows.v1.WorkflowsService.GetCluster:input_type -> workflows.v1.GetClusterRequest
-	3,  // 22: workflows.v1.WorkflowsService.DeleteCluster:input_type -> workflows.v1.DeleteClusterRequest
-	5,  // 23: workflows.v1.WorkflowsService.ListClusters:input_type -> workflows.v1.ListClustersRequest
-	11, // 24: workflows.v1.WorkflowsService.CreateWorkflow:input_type -> workflows.v1.CreateWorkflowRequest
-	9,  // 25: workflows.v1.WorkflowsService.ListWorkflows:input_type -> workflows.v1.ListWorkflowsRequest
-	7,  // 26: workflows.v1.WorkflowsService.GetWorkflow:input_type -> workflows.v1.GetWorkflowRequest
-	8,  // 27: workflows.v1.WorkflowsService.PublishWorkflowRelease:input_type -> workflows.v1.PublishWorkflowReleaseRequest
-	17, // 28: workflows.v1.WorkflowsService.DeployWorkflowRelease:input_type -> workflows.v1.DeployWorkflowReleaseRequest
-	19, // 29: workflows.v1.WorkflowsService.UndeployWorkflowRelease:input_type -> workflows.v1.UndeployWorkflowReleaseRequest
-	0,  // 30: workflows.v1.WorkflowsService.CreateCluster:output_type -> workflows.v1.Cluster
-	0,  // 31: workflows.v1.WorkflowsService.GetCluster:output_type -> workflows.v1.Cluster
-	4,  // 32: workflows.v1.WorkflowsService.DeleteCluster:output_type -> workflows.v1.DeleteClusterResponse
-	6,  // 33: workflows.v1.WorkflowsService.ListClusters:output_type -> workflows.v1.ListClustersResponse
-	12, // 34: workflows.v1.WorkflowsService.CreateWorkflow:output_type -> workflows.v1.Workflow
-	10, // 35: workflows.v1.WorkflowsService.ListWorkflows:output_type -> workflows.v1.ListWorkflowsResponse
-	12, // 36: workflows.v1.WorkflowsService.GetWorkflow:output_type -> workflows.v1.Workflow
-	13, // 37: workflows.v1.WorkflowsService.PublishWorkflowRelease:output_type -> workflows.v1.WorkflowRelease
-	18, // 38: workflows.v1.WorkflowsService.DeployWorkflowRelease:output_type -> workflows.v1.DeployWorkflowReleaseResponse
-	20, // 39: workflows.v1.WorkflowsService.UndeployWorkflowRelease:output_type -> workflows.v1.UndeployWorkflowReleaseResponse
-	30, // [30:40] is the sub-list for method output_type
-	20, // [20:30] is the sub-list for method input_type
-	20, // [20:20] is the sub-list for extension type_name
-	20, // [20:20] is the sub-list for extension extendee
-	0,  // [0:20] is the sub-list for field type_name
+	25, // 2: workflows.v1.PublishWorkflowReleaseRequest.artifact_id:type_name -> tilebox.v1.ID
+	19, // 3: workflows.v1.PublishWorkflowReleaseRequest.content:type_name -> workflows.v1.ReleaseContent
+	25, // 4: workflows.v1.UnpublishWorkflowReleaseRequest.release_id:type_name -> tilebox.v1.ID
+	16, // 5: workflows.v1.ListWorkflowsResponse.workflows:type_name -> workflows.v1.Workflow
+	17, // 6: workflows.v1.Workflow.releases:type_name -> workflows.v1.WorkflowRelease
+	25, // 7: workflows.v1.WorkflowRelease.id:type_name -> tilebox.v1.ID
+	18, // 8: workflows.v1.WorkflowRelease.artifact:type_name -> workflows.v1.Artifact
+	19, // 9: workflows.v1.WorkflowRelease.content:type_name -> workflows.v1.ReleaseContent
+	26, // 10: workflows.v1.WorkflowRelease.created_at:type_name -> google.protobuf.Timestamp
+	0,  // 11: workflows.v1.WorkflowRelease.clusters:type_name -> workflows.v1.Cluster
+	25, // 12: workflows.v1.Artifact.id:type_name -> tilebox.v1.ID
+	27, // 13: workflows.v1.ReleaseContent.tasks:type_name -> workflows.v1.TaskIdentifier
+	20, // 14: workflows.v1.ReleaseContent.files:type_name -> workflows.v1.Path
+	20, // 15: workflows.v1.Path.children:type_name -> workflows.v1.Path
+	25, // 16: workflows.v1.DeployWorkflowReleaseRequest.release_id:type_name -> tilebox.v1.ID
+	17, // 17: workflows.v1.DeployWorkflowReleaseResponse.release:type_name -> workflows.v1.WorkflowRelease
+	0,  // 18: workflows.v1.DeployWorkflowReleaseResponse.clusters:type_name -> workflows.v1.Cluster
+	25, // 19: workflows.v1.UndeployWorkflowReleaseRequest.release_id:type_name -> tilebox.v1.ID
+	17, // 20: workflows.v1.UndeployWorkflowReleaseResponse.release:type_name -> workflows.v1.WorkflowRelease
+	0,  // 21: workflows.v1.UndeployWorkflowReleaseResponse.clusters:type_name -> workflows.v1.Cluster
+	1,  // 22: workflows.v1.WorkflowsService.CreateCluster:input_type -> workflows.v1.CreateClusterRequest
+	2,  // 23: workflows.v1.WorkflowsService.GetCluster:input_type -> workflows.v1.GetClusterRequest
+	3,  // 24: workflows.v1.WorkflowsService.DeleteCluster:input_type -> workflows.v1.DeleteClusterRequest
+	5,  // 25: workflows.v1.WorkflowsService.ListClusters:input_type -> workflows.v1.ListClustersRequest
+	15, // 26: workflows.v1.WorkflowsService.CreateWorkflow:input_type -> workflows.v1.CreateWorkflowRequest
+	13, // 27: workflows.v1.WorkflowsService.ListWorkflows:input_type -> workflows.v1.ListWorkflowsRequest
+	7,  // 28: workflows.v1.WorkflowsService.GetWorkflow:input_type -> workflows.v1.GetWorkflowRequest
+	8,  // 29: workflows.v1.WorkflowsService.DeleteWorkflow:input_type -> workflows.v1.DeleteWorkflowRequest
+	10, // 30: workflows.v1.WorkflowsService.PublishWorkflowRelease:input_type -> workflows.v1.PublishWorkflowReleaseRequest
+	11, // 31: workflows.v1.WorkflowsService.UnpublishWorkflowRelease:input_type -> workflows.v1.UnpublishWorkflowReleaseRequest
+	21, // 32: workflows.v1.WorkflowsService.DeployWorkflowRelease:input_type -> workflows.v1.DeployWorkflowReleaseRequest
+	23, // 33: workflows.v1.WorkflowsService.UndeployWorkflowRelease:input_type -> workflows.v1.UndeployWorkflowReleaseRequest
+	0,  // 34: workflows.v1.WorkflowsService.CreateCluster:output_type -> workflows.v1.Cluster
+	0,  // 35: workflows.v1.WorkflowsService.GetCluster:output_type -> workflows.v1.Cluster
+	4,  // 36: workflows.v1.WorkflowsService.DeleteCluster:output_type -> workflows.v1.DeleteClusterResponse
+	6,  // 37: workflows.v1.WorkflowsService.ListClusters:output_type -> workflows.v1.ListClustersResponse
+	16, // 38: workflows.v1.WorkflowsService.CreateWorkflow:output_type -> workflows.v1.Workflow
+	14, // 39: workflows.v1.WorkflowsService.ListWorkflows:output_type -> workflows.v1.ListWorkflowsResponse
+	16, // 40: workflows.v1.WorkflowsService.GetWorkflow:output_type -> workflows.v1.Workflow
+	9,  // 41: workflows.v1.WorkflowsService.DeleteWorkflow:output_type -> workflows.v1.DeleteWorkflowResponse
+	17, // 42: workflows.v1.WorkflowsService.PublishWorkflowRelease:output_type -> workflows.v1.WorkflowRelease
+	12, // 43: workflows.v1.WorkflowsService.UnpublishWorkflowRelease:output_type -> workflows.v1.UnpublishWorkflowReleaseResponse
+	22, // 44: workflows.v1.WorkflowsService.DeployWorkflowRelease:output_type -> workflows.v1.DeployWorkflowReleaseResponse
+	24, // 45: workflows.v1.WorkflowsService.UndeployWorkflowRelease:output_type -> workflows.v1.UndeployWorkflowReleaseResponse
+	34, // [34:46] is the sub-list for method output_type
+	22, // [22:34] is the sub-list for method input_type
+	22, // [22:22] is the sub-list for extension type_name
+	22, // [22:22] is the sub-list for extension extendee
+	0,  // [0:22] is the sub-list for field type_name
 }
 
 func init() { file_workflows_v1_workflows_proto_init() }
@@ -1892,7 +2161,7 @@ func file_workflows_v1_workflows_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_workflows_v1_workflows_proto_rawDesc), len(file_workflows_v1_workflows_proto_rawDesc)),
 			NumEnums:      0,
-			NumMessages:   21,
+			NumMessages:   25,
 			NumExtensions: 0,
 			NumServices:   1,
 		},
diff --git a/protogen/workflows/v1/workflows_grpc.pb.go b/protogen/workflows/v1/workflows_grpc.pb.go
index 23bd0ae..ae30673 100644
--- a/protogen/workflows/v1/workflows_grpc.pb.go
+++ b/protogen/workflows/v1/workflows_grpc.pb.go
@@ -21,16 +21,18 @@ import (
 const _ = grpc.SupportPackageIsVersion9
 
 const (
-	WorkflowsService_CreateCluster_FullMethodName           = "/workflows.v1.WorkflowsService/CreateCluster"
-	WorkflowsService_GetCluster_FullMethodName              = "/workflows.v1.WorkflowsService/GetCluster"
-	WorkflowsService_DeleteCluster_FullMethodName           = "/workflows.v1.WorkflowsService/DeleteCluster"
-	WorkflowsService_ListClusters_FullMethodName            = "/workflows.v1.WorkflowsService/ListClusters"
-	WorkflowsService_CreateWorkflow_FullMethodName          = "/workflows.v1.WorkflowsService/CreateWorkflow"
-	WorkflowsService_ListWorkflows_FullMethodName           = "/workflows.v1.WorkflowsService/ListWorkflows"
-	WorkflowsService_GetWorkflow_FullMethodName             = "/workflows.v1.WorkflowsService/GetWorkflow"
-	WorkflowsService_PublishWorkflowRelease_FullMethodName  = "/workflows.v1.WorkflowsService/PublishWorkflowRelease"
-	WorkflowsService_DeployWorkflowRelease_FullMethodName   = "/workflows.v1.WorkflowsService/DeployWorkflowRelease"
-	WorkflowsService_UndeployWorkflowRelease_FullMethodName = "/workflows.v1.WorkflowsService/UndeployWorkflowRelease"
+	WorkflowsService_CreateCluster_FullMethodName            = "/workflows.v1.WorkflowsService/CreateCluster"
+	WorkflowsService_GetCluster_FullMethodName               = "/workflows.v1.WorkflowsService/GetCluster"
+	WorkflowsService_DeleteCluster_FullMethodName            = "/workflows.v1.WorkflowsService/DeleteCluster"
+	WorkflowsService_ListClusters_FullMethodName             = "/workflows.v1.WorkflowsService/ListClusters"
+	WorkflowsService_CreateWorkflow_FullMethodName           = "/workflows.v1.WorkflowsService/CreateWorkflow"
+	WorkflowsService_ListWorkflows_FullMethodName            = "/workflows.v1.WorkflowsService/ListWorkflows"
+	WorkflowsService_GetWorkflow_FullMethodName              = "/workflows.v1.WorkflowsService/GetWorkflow"
+	WorkflowsService_DeleteWorkflow_FullMethodName           = "/workflows.v1.WorkflowsService/DeleteWorkflow"
+	WorkflowsService_PublishWorkflowRelease_FullMethodName   = "/workflows.v1.WorkflowsService/PublishWorkflowRelease"
+	WorkflowsService_UnpublishWorkflowRelease_FullMethodName = "/workflows.v1.WorkflowsService/UnpublishWorkflowRelease"
+	WorkflowsService_DeployWorkflowRelease_FullMethodName    = "/workflows.v1.WorkflowsService/DeployWorkflowRelease"
+	WorkflowsService_UndeployWorkflowRelease_FullMethodName  = "/workflows.v1.WorkflowsService/UndeployWorkflowRelease"
 )
 
 // WorkflowsServiceClient is the client API for WorkflowsService service.
@@ -46,7 +48,9 @@ type WorkflowsServiceClient interface {
 	CreateWorkflow(ctx context.Context, in *CreateWorkflowRequest, opts ...grpc.CallOption) (*Workflow, error)
 	ListWorkflows(ctx context.Context, in *ListWorkflowsRequest, opts ...grpc.CallOption) (*ListWorkflowsResponse, error)
 	GetWorkflow(ctx context.Context, in *GetWorkflowRequest, opts ...grpc.CallOption) (*Workflow, error)
+	DeleteWorkflow(ctx context.Context, in *DeleteWorkflowRequest, opts ...grpc.CallOption) (*DeleteWorkflowResponse, error)
 	PublishWorkflowRelease(ctx context.Context, in *PublishWorkflowReleaseRequest, opts ...grpc.CallOption) (*WorkflowRelease, error)
+	UnpublishWorkflowRelease(ctx context.Context, in *UnpublishWorkflowReleaseRequest, opts ...grpc.CallOption) (*UnpublishWorkflowReleaseResponse, error)
 	DeployWorkflowRelease(ctx context.Context, in *DeployWorkflowReleaseRequest, opts ...grpc.CallOption) (*DeployWorkflowReleaseResponse, error)
 	UndeployWorkflowRelease(ctx context.Context, in *UndeployWorkflowReleaseRequest, opts ...grpc.CallOption) (*UndeployWorkflowReleaseResponse, error)
 }
@@ -129,6 +133,16 @@ func (c *workflowsServiceClient) GetWorkflow(ctx context.Context, in *GetWorkflo
 	return out, nil
 }
 
+func (c *workflowsServiceClient) DeleteWorkflow(ctx context.Context, in *DeleteWorkflowRequest, opts ...grpc.CallOption) (*DeleteWorkflowResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(DeleteWorkflowResponse)
+	err := c.cc.Invoke(ctx, WorkflowsService_DeleteWorkflow_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *workflowsServiceClient) PublishWorkflowRelease(ctx context.Context, in *PublishWorkflowReleaseRequest, opts ...grpc.CallOption) (*WorkflowRelease, error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
 	out := new(WorkflowRelease)
@@ -139,6 +153,16 @@ func (c *workflowsServiceClient) PublishWorkflowRelease(ctx context.Context, in
 	return out, nil
 }
 
+func (c *workflowsServiceClient) UnpublishWorkflowRelease(ctx context.Context, in *UnpublishWorkflowReleaseRequest, opts ...grpc.CallOption) (*UnpublishWorkflowReleaseResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(UnpublishWorkflowReleaseResponse)
+	err := c.cc.Invoke(ctx, WorkflowsService_UnpublishWorkflowRelease_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *workflowsServiceClient) DeployWorkflowRelease(ctx context.Context, in *DeployWorkflowReleaseRequest, opts ...grpc.CallOption) (*DeployWorkflowReleaseResponse, error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
 	out := new(DeployWorkflowReleaseResponse)
@@ -172,7 +196,9 @@ type WorkflowsServiceServer interface {
 	CreateWorkflow(context.Context, *CreateWorkflowRequest) (*Workflow, error)
 	ListWorkflows(context.Context, *ListWorkflowsRequest) (*ListWorkflowsResponse, error)
 	GetWorkflow(context.Context, *GetWorkflowRequest) (*Workflow, error)
+	DeleteWorkflow(context.Context, *DeleteWorkflowRequest) (*DeleteWorkflowResponse, error)
 	PublishWorkflowRelease(context.Context, *PublishWorkflowReleaseRequest) (*WorkflowRelease, error)
+	UnpublishWorkflowRelease(context.Context, *UnpublishWorkflowReleaseRequest) (*UnpublishWorkflowReleaseResponse, error)
 	DeployWorkflowRelease(context.Context, *DeployWorkflowReleaseRequest) (*DeployWorkflowReleaseResponse, error)
 	UndeployWorkflowRelease(context.Context, *UndeployWorkflowReleaseRequest) (*UndeployWorkflowReleaseResponse, error)
 	mustEmbedUnimplementedWorkflowsServiceServer()
@@ -206,9 +232,15 @@ func (UnimplementedWorkflowsServiceServer) ListWorkflows(context.Context, *ListW
 func (UnimplementedWorkflowsServiceServer) GetWorkflow(context.Context, *GetWorkflowRequest) (*Workflow, error) {
 	return nil, status.Error(codes.Unimplemented, "method GetWorkflow not implemented")
 }
+func (UnimplementedWorkflowsServiceServer) DeleteWorkflow(context.Context, *DeleteWorkflowRequest) (*DeleteWorkflowResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method DeleteWorkflow not implemented")
+}
 func (UnimplementedWorkflowsServiceServer) PublishWorkflowRelease(context.Context, *PublishWorkflowReleaseRequest) (*WorkflowRelease, error) {
 	return nil, status.Error(codes.Unimplemented, "method PublishWorkflowRelease not implemented")
 }
+func (UnimplementedWorkflowsServiceServer) UnpublishWorkflowRelease(context.Context, *UnpublishWorkflowReleaseRequest) (*UnpublishWorkflowReleaseResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "method UnpublishWorkflowRelease not implemented")
+}
 func (UnimplementedWorkflowsServiceServer) DeployWorkflowRelease(context.Context, *DeployWorkflowReleaseRequest) (*DeployWorkflowReleaseResponse, error) {
 	return nil, status.Error(codes.Unimplemented, "method DeployWorkflowRelease not implemented")
 }
@@ -362,6 +394,24 @@ func _WorkflowsService_GetWorkflow_Handler(srv interface{}, ctx context.Context,
 	return interceptor(ctx, in, info, handler)
 }
 
+func _WorkflowsService_DeleteWorkflow_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(DeleteWorkflowRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(WorkflowsServiceServer).DeleteWorkflow(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: WorkflowsService_DeleteWorkflow_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(WorkflowsServiceServer).DeleteWorkflow(ctx, req.(*DeleteWorkflowRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _WorkflowsService_PublishWorkflowRelease_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(PublishWorkflowReleaseRequest)
 	if err := dec(in); err != nil {
@@ -380,6 +430,24 @@ func _WorkflowsService_PublishWorkflowRelease_Handler(srv interface{}, ctx conte
 	return interceptor(ctx, in, info, handler)
 }
 
+func _WorkflowsService_UnpublishWorkflowRelease_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(UnpublishWorkflowReleaseRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(WorkflowsServiceServer).UnpublishWorkflowRelease(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: WorkflowsService_UnpublishWorkflowRelease_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(WorkflowsServiceServer).UnpublishWorkflowRelease(ctx, req.(*UnpublishWorkflowReleaseRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _WorkflowsService_DeployWorkflowRelease_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(DeployWorkflowReleaseRequest)
 	if err := dec(in); err != nil {
@@ -451,10 +519,18 @@ var WorkflowsService_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "GetWorkflow",
 			Handler:    _WorkflowsService_GetWorkflow_Handler,
 		},
+		{
+			MethodName: "DeleteWorkflow",
+			Handler:    _WorkflowsService_DeleteWorkflow_Handler,
+		},
 		{
 			MethodName: "PublishWorkflowRelease",
 			Handler:    _WorkflowsService_PublishWorkflowRelease_Handler,
 		},
+		{
+			MethodName: "UnpublishWorkflowRelease",
+			Handler:    _WorkflowsService_UnpublishWorkflowRelease_Handler,
+		},
 		{
 			MethodName: "DeployWorkflowRelease",
 			Handler:    _WorkflowsService_DeployWorkflowRelease_Handler,
diff --git a/protogen/workflows/v1/workflowsv1connect/workflows.connect.go b/protogen/workflows/v1/workflowsv1connect/workflows.connect.go
index 045cbb3..f9a0041 100644
--- a/protogen/workflows/v1/workflowsv1connect/workflows.connect.go
+++ b/protogen/workflows/v1/workflowsv1connect/workflows.connect.go
@@ -56,9 +56,15 @@ const (
 	// WorkflowsServiceGetWorkflowProcedure is the fully-qualified name of the WorkflowsService's
 	// GetWorkflow RPC.
 	WorkflowsServiceGetWorkflowProcedure = "/workflows.v1.WorkflowsService/GetWorkflow"
+	// WorkflowsServiceDeleteWorkflowProcedure is the fully-qualified name of the WorkflowsService's
+	// DeleteWorkflow RPC.
+	WorkflowsServiceDeleteWorkflowProcedure = "/workflows.v1.WorkflowsService/DeleteWorkflow"
 	// WorkflowsServicePublishWorkflowReleaseProcedure is the fully-qualified name of the
 	// WorkflowsService's PublishWorkflowRelease RPC.
 	WorkflowsServicePublishWorkflowReleaseProcedure = "/workflows.v1.WorkflowsService/PublishWorkflowRelease"
+	// WorkflowsServiceUnpublishWorkflowReleaseProcedure is the fully-qualified name of the
+	// WorkflowsService's UnpublishWorkflowRelease RPC.
+	WorkflowsServiceUnpublishWorkflowReleaseProcedure = "/workflows.v1.WorkflowsService/UnpublishWorkflowRelease"
 	// WorkflowsServiceDeployWorkflowReleaseProcedure is the fully-qualified name of the
 	// WorkflowsService's DeployWorkflowRelease RPC.
 	WorkflowsServiceDeployWorkflowReleaseProcedure = "/workflows.v1.WorkflowsService/DeployWorkflowRelease"
@@ -76,7 +82,9 @@ type WorkflowsServiceClient interface {
 	CreateWorkflow(context.Context, *connect.Request[v1.CreateWorkflowRequest]) (*connect.Response[v1.Workflow], error)
 	ListWorkflows(context.Context, *connect.Request[v1.ListWorkflowsRequest]) (*connect.Response[v1.ListWorkflowsResponse], error)
 	GetWorkflow(context.Context, *connect.Request[v1.GetWorkflowRequest]) (*connect.Response[v1.Workflow], error)
+	DeleteWorkflow(context.Context, *connect.Request[v1.DeleteWorkflowRequest]) (*connect.Response[v1.DeleteWorkflowResponse], error)
 	PublishWorkflowRelease(context.Context, *connect.Request[v1.PublishWorkflowReleaseRequest]) (*connect.Response[v1.WorkflowRelease], error)
+	UnpublishWorkflowRelease(context.Context, *connect.Request[v1.UnpublishWorkflowReleaseRequest]) (*connect.Response[v1.UnpublishWorkflowReleaseResponse], error)
 	DeployWorkflowRelease(context.Context, *connect.Request[v1.DeployWorkflowReleaseRequest]) (*connect.Response[v1.DeployWorkflowReleaseResponse], error)
 	UndeployWorkflowRelease(context.Context, *connect.Request[v1.UndeployWorkflowReleaseRequest]) (*connect.Response[v1.UndeployWorkflowReleaseResponse], error)
 }
@@ -134,12 +142,24 @@ func NewWorkflowsServiceClient(httpClient connect.HTTPClient, baseURL string, op
 			connect.WithSchema(workflowsServiceMethods.ByName("GetWorkflow")),
 			connect.WithClientOptions(opts...),
 		),
+		deleteWorkflow: connect.NewClient[v1.DeleteWorkflowRequest, v1.DeleteWorkflowResponse](
+			httpClient,
+			baseURL+WorkflowsServiceDeleteWorkflowProcedure,
+			connect.WithSchema(workflowsServiceMethods.ByName("DeleteWorkflow")),
+			connect.WithClientOptions(opts...),
+		),
 		publishWorkflowRelease: connect.NewClient[v1.PublishWorkflowReleaseRequest, v1.WorkflowRelease](
 			httpClient,
 			baseURL+WorkflowsServicePublishWorkflowReleaseProcedure,
 			connect.WithSchema(workflowsServiceMethods.ByName("PublishWorkflowRelease")),
 			connect.WithClientOptions(opts...),
 		),
+		unpublishWorkflowRelease: connect.NewClient[v1.UnpublishWorkflowReleaseRequest, v1.UnpublishWorkflowReleaseResponse](
+			httpClient,
+			baseURL+WorkflowsServiceUnpublishWorkflowReleaseProcedure,
+			connect.WithSchema(workflowsServiceMethods.ByName("UnpublishWorkflowRelease")),
+			connect.WithClientOptions(opts...),
+		),
 		deployWorkflowRelease: connect.NewClient[v1.DeployWorkflowReleaseRequest, v1.DeployWorkflowReleaseResponse](
 			httpClient,
 			baseURL+WorkflowsServiceDeployWorkflowReleaseProcedure,
@@ -157,16 +177,18 @@ func NewWorkflowsServiceClient(httpClient connect.HTTPClient, baseURL string, op
 
 // workflowsServiceClient implements WorkflowsServiceClient.
 type workflowsServiceClient struct {
-	createCluster           *connect.Client[v1.CreateClusterRequest, v1.Cluster]
-	getCluster              *connect.Client[v1.GetClusterRequest, v1.Cluster]
-	deleteCluster           *connect.Client[v1.DeleteClusterRequest, v1.DeleteClusterResponse]
-	listClusters            *connect.Client[v1.ListClustersRequest, v1.ListClustersResponse]
-	createWorkflow          *connect.Client[v1.CreateWorkflowRequest, v1.Workflow]
-	listWorkflows           *connect.Client[v1.ListWorkflowsRequest, v1.ListWorkflowsResponse]
-	getWorkflow             *connect.Client[v1.GetWorkflowRequest, v1.Workflow]
-	publishWorkflowRelease  *connect.Client[v1.PublishWorkflowReleaseRequest, v1.WorkflowRelease]
-	deployWorkflowRelease   *connect.Client[v1.DeployWorkflowReleaseRequest, v1.DeployWorkflowReleaseResponse]
-	undeployWorkflowRelease *connect.Client[v1.UndeployWorkflowReleaseRequest, v1.UndeployWorkflowReleaseResponse]
+	createCluster            *connect.Client[v1.CreateClusterRequest, v1.Cluster]
+	getCluster               *connect.Client[v1.GetClusterRequest, v1.Cluster]
+	deleteCluster            *connect.Client[v1.DeleteClusterRequest, v1.DeleteClusterResponse]
+	listClusters             *connect.Client[v1.ListClustersRequest, v1.ListClustersResponse]
+	createWorkflow           *connect.Client[v1.CreateWorkflowRequest, v1.Workflow]
+	listWorkflows            *connect.Client[v1.ListWorkflowsRequest, v1.ListWorkflowsResponse]
+	getWorkflow              *connect.Client[v1.GetWorkflowRequest, v1.Workflow]
+	deleteWorkflow           *connect.Client[v1.DeleteWorkflowRequest, v1.DeleteWorkflowResponse]
+	publishWorkflowRelease   *connect.Client[v1.PublishWorkflowReleaseRequest, v1.WorkflowRelease]
+	unpublishWorkflowRelease *connect.Client[v1.UnpublishWorkflowReleaseRequest, v1.UnpublishWorkflowReleaseResponse]
+	deployWorkflowRelease    *connect.Client[v1.DeployWorkflowReleaseRequest, v1.DeployWorkflowReleaseResponse]
+	undeployWorkflowRelease  *connect.Client[v1.UndeployWorkflowReleaseRequest, v1.UndeployWorkflowReleaseResponse]
 }
 
 // CreateCluster calls workflows.v1.WorkflowsService.CreateCluster.
@@ -204,11 +226,21 @@ func (c *workflowsServiceClient) GetWorkflow(ctx context.Context, req *connect.R
 	return c.getWorkflow.CallUnary(ctx, req)
 }
 
+// DeleteWorkflow calls workflows.v1.WorkflowsService.DeleteWorkflow.
+func (c *workflowsServiceClient) DeleteWorkflow(ctx context.Context, req *connect.Request[v1.DeleteWorkflowRequest]) (*connect.Response[v1.DeleteWorkflowResponse], error) {
+	return c.deleteWorkflow.CallUnary(ctx, req)
+}
+
 // PublishWorkflowRelease calls workflows.v1.WorkflowsService.PublishWorkflowRelease.
 func (c *workflowsServiceClient) PublishWorkflowRelease(ctx context.Context, req *connect.Request[v1.PublishWorkflowReleaseRequest]) (*connect.Response[v1.WorkflowRelease], error) {
 	return c.publishWorkflowRelease.CallUnary(ctx, req)
 }
 
+// UnpublishWorkflowRelease calls workflows.v1.WorkflowsService.UnpublishWorkflowRelease.
+func (c *workflowsServiceClient) UnpublishWorkflowRelease(ctx context.Context, req *connect.Request[v1.UnpublishWorkflowReleaseRequest]) (*connect.Response[v1.UnpublishWorkflowReleaseResponse], error) {
+	return c.unpublishWorkflowRelease.CallUnary(ctx, req)
+}
+
 // DeployWorkflowRelease calls workflows.v1.WorkflowsService.DeployWorkflowRelease.
 func (c *workflowsServiceClient) DeployWorkflowRelease(ctx context.Context, req *connect.Request[v1.DeployWorkflowReleaseRequest]) (*connect.Response[v1.DeployWorkflowReleaseResponse], error) {
 	return c.deployWorkflowRelease.CallUnary(ctx, req)
@@ -228,7 +260,9 @@ type WorkflowsServiceHandler interface {
 	CreateWorkflow(context.Context, *connect.Request[v1.CreateWorkflowRequest]) (*connect.Response[v1.Workflow], error)
 	ListWorkflows(context.Context, *connect.Request[v1.ListWorkflowsRequest]) (*connect.Response[v1.ListWorkflowsResponse], error)
 	GetWorkflow(context.Context, *connect.Request[v1.GetWorkflowRequest]) (*connect.Response[v1.Workflow], error)
+	DeleteWorkflow(context.Context, *connect.Request[v1.DeleteWorkflowRequest]) (*connect.Response[v1.DeleteWorkflowResponse], error)
 	PublishWorkflowRelease(context.Context, *connect.Request[v1.PublishWorkflowReleaseRequest]) (*connect.Response[v1.WorkflowRelease], error)
+	UnpublishWorkflowRelease(context.Context, *connect.Request[v1.UnpublishWorkflowReleaseRequest]) (*connect.Response[v1.UnpublishWorkflowReleaseResponse], error)
 	DeployWorkflowRelease(context.Context, *connect.Request[v1.DeployWorkflowReleaseRequest]) (*connect.Response[v1.DeployWorkflowReleaseResponse], error)
 	UndeployWorkflowRelease(context.Context, *connect.Request[v1.UndeployWorkflowReleaseRequest]) (*connect.Response[v1.UndeployWorkflowReleaseResponse], error)
 }
@@ -282,12 +316,24 @@ func NewWorkflowsServiceHandler(svc WorkflowsServiceHandler, opts ...connect.Han
 		connect.WithSchema(workflowsServiceMethods.ByName("GetWorkflow")),
 		connect.WithHandlerOptions(opts...),
 	)
+	workflowsServiceDeleteWorkflowHandler := connect.NewUnaryHandler(
+		WorkflowsServiceDeleteWorkflowProcedure,
+		svc.DeleteWorkflow,
+		connect.WithSchema(workflowsServiceMethods.ByName("DeleteWorkflow")),
+		connect.WithHandlerOptions(opts...),
+	)
 	workflowsServicePublishWorkflowReleaseHandler := connect.NewUnaryHandler(
 		WorkflowsServicePublishWorkflowReleaseProcedure,
 		svc.PublishWorkflowRelease,
 		connect.WithSchema(workflowsServiceMethods.ByName("PublishWorkflowRelease")),
 		connect.WithHandlerOptions(opts...),
 	)
+	workflowsServiceUnpublishWorkflowReleaseHandler := connect.NewUnaryHandler(
+		WorkflowsServiceUnpublishWorkflowReleaseProcedure,
+		svc.UnpublishWorkflowRelease,
+		connect.WithSchema(workflowsServiceMethods.ByName("UnpublishWorkflowRelease")),
+		connect.WithHandlerOptions(opts...),
+	)
 	workflowsServiceDeployWorkflowReleaseHandler := connect.NewUnaryHandler(
 		WorkflowsServiceDeployWorkflowReleaseProcedure,
 		svc.DeployWorkflowRelease,
@@ -316,8 +362,12 @@ func NewWorkflowsServiceHandler(svc WorkflowsServiceHandler, opts ...connect.Han
 			workflowsServiceListWorkflowsHandler.ServeHTTP(w, r)
 		case WorkflowsServiceGetWorkflowProcedure:
 			workflowsServiceGetWorkflowHandler.ServeHTTP(w, r)
+		case WorkflowsServiceDeleteWorkflowProcedure:
+			workflowsServiceDeleteWorkflowHandler.ServeHTTP(w, r)
 		case WorkflowsServicePublishWorkflowReleaseProcedure:
 			workflowsServicePublishWorkflowReleaseHandler.ServeHTTP(w, r)
+		case WorkflowsServiceUnpublishWorkflowReleaseProcedure:
+			workflowsServiceUnpublishWorkflowReleaseHandler.ServeHTTP(w, r)
 		case WorkflowsServiceDeployWorkflowReleaseProcedure:
 			workflowsServiceDeployWorkflowReleaseHandler.ServeHTTP(w, r)
 		case WorkflowsServiceUndeployWorkflowReleaseProcedure:
@@ -359,10 +409,18 @@ func (UnimplementedWorkflowsServiceHandler) GetWorkflow(context.Context, *connec
 	return nil, connect.NewError(connect.CodeUnimplemented, errors.New("workflows.v1.WorkflowsService.GetWorkflow is not implemented"))
 }
 
+func (UnimplementedWorkflowsServiceHandler) DeleteWorkflow(context.Context, *connect.Request[v1.DeleteWorkflowRequest]) (*connect.Response[v1.DeleteWorkflowResponse], error) {
+	return nil, connect.NewError(connect.CodeUnimplemented, errors.New("workflows.v1.WorkflowsService.DeleteWorkflow is not implemented"))
+}
+
 func (UnimplementedWorkflowsServiceHandler) PublishWorkflowRelease(context.Context, *connect.Request[v1.PublishWorkflowReleaseRequest]) (*connect.Response[v1.WorkflowRelease], error) {
 	return nil, connect.NewError(connect.CodeUnimplemented, errors.New("workflows.v1.WorkflowsService.PublishWorkflowRelease is not implemented"))
 }
 
+func (UnimplementedWorkflowsServiceHandler) UnpublishWorkflowRelease(context.Context, *connect.Request[v1.UnpublishWorkflowReleaseRequest]) (*connect.Response[v1.UnpublishWorkflowReleaseResponse], error) {
+	return nil, connect.NewError(connect.CodeUnimplemented, errors.New("workflows.v1.WorkflowsService.UnpublishWorkflowRelease is not implemented"))
+}
+
 func (UnimplementedWorkflowsServiceHandler) DeployWorkflowRelease(context.Context, *connect.Request[v1.DeployWorkflowReleaseRequest]) (*connect.Response[v1.DeployWorkflowReleaseResponse], error) {
 	return nil, connect.NewError(connect.CodeUnimplemented, errors.New("workflows.v1.WorkflowsService.DeployWorkflowRelease is not implemented"))
 }
diff --git a/workflows/v1/job/options.go b/workflows/v1/job/options.go
index 919d27c..c2df009 100644
--- a/workflows/v1/job/options.go
+++ b/workflows/v1/job/options.go
@@ -126,6 +126,8 @@ type QueryOptions struct {
 	Cursor *Cursor
 	// AutomationIDs filters jobs by the automations that submitted them.
 	AutomationIDs []uuid.UUID
+	// ClusterSlugs filters jobs by the clusters they were submitted to.
+	ClusterSlugs []string
 	// States filters jobs by their states.
 	States []State
 	// TaskStates filters jobs by the states of their tasks.
@@ -264,6 +266,14 @@ func WithAutomationIDs(automationIDs ...uuid.UUID) QueryOption {
 	})
 }
 
+// WithClusterSlugs specifies multiple cluster slugs to filter jobs by.
+// Only jobs that have tasks on any of the specified clusters will be returned.
+func WithClusterSlugs(clusterSlugs ...string) QueryOption {
+	return queryOptionFunc(func(cfg *QueryOptions) {
+		cfg.ClusterSlugs = append(cfg.ClusterSlugs, clusterSlugs...)
+	})
+}
+
 // WithJobStates filters jobs by their state.
 // Only jobs in any of the specified states will be returned.
 func WithJobStates(states ...State) QueryOption {
diff --git a/workflows/v1/job/options_test.go b/workflows/v1/job/options_test.go
index 1d7c0f9..d62523e 100644
--- a/workflows/v1/job/options_test.go
+++ b/workflows/v1/job/options_test.go
@@ -137,6 +137,16 @@ func Test_QueryOptions(t *testing.T) {
 				AutomationIDs: []uuid.UUID{automationID, anotherAutomationID},
 			},
 		},
+		{
+			name: "with cluster slugs",
+			options: []QueryOption{
+				WithClusterSlugs("default", "gpu"),
+				WithClusterSlugs("batch"),
+			},
+			want: QueryOptions{
+				ClusterSlugs: []string{"default", "gpu", "batch"},
+			},
+		},
 		{
 			name: "with job states",
 			options: []QueryOption{
diff --git a/workflows/v1/jobs.go b/workflows/v1/jobs.go
index 64768ef..488d3c8 100644
--- a/workflows/v1/jobs.go
+++ b/workflows/v1/jobs.go
@@ -243,6 +243,8 @@ type JobClient interface {
 	//   - job.WithTemporalExtent: specifies the time or ID interval for which jobs should be queried. (Optional)
 	//   - job.WithAutomationIDs: specifies the automation IDs to filter jobs by. Only jobs submitted by the specified
 	//  automation will be returned. (Optional)
+	//   - job.WithClusterSlugs: specifies the cluster slugs to filter jobs by. Only jobs submitted to the specified
+	//     clusters will be returned. (Optional)
 	//   - job.WithJobStates: specifies the job state to filter jobs by. Only jobs with the specified state will be returned. (Optional)
 	//   - job.WithTaskStates: specifies task states to filter jobs by. Only jobs with at least one task in one of the
 	//     specified states will be returned. (Optional)
@@ -261,6 +263,8 @@ type JobClient interface {
 	//   - job.WithTemporalExtent: specifies the time or ID interval for which jobs should be queried. (Optional)
 	//   - job.WithAutomationIDs: specifies the automation IDs to filter jobs by. Only jobs submitted by the specified
 	//  automation will be returned. (Optional)
+	//   - job.WithClusterSlugs: specifies the cluster slugs to filter jobs by. Only jobs submitted to the specified
+	//     clusters will be returned. (Optional)
 	//   - job.WithJobStates: specifies the job state to filter jobs by. Only jobs with the specified state will be returned. (Optional)
 	//   - job.WithTaskStates: specifies task states to filter jobs by. Only jobs with at least one task in one of the
 	//     specified states will be returned. (Optional)
@@ -572,6 +576,7 @@ func (c jobClient) queryPage(ctx context.Context, opts *job.QueryOptions) (*JobP
 		TimeInterval:  timeInterval,
 		IdInterval:    idInterval,
 		AutomationIds: automationIDs,
+		ClusterSlugs:  opts.ClusterSlugs,
 		States:        states,
 		TaskStates:    taskStates,
 		Name:          opts.Name,
diff --git a/workflows/v1/jobs_test.go b/workflows/v1/jobs_test.go
index 2ad64bd..9e90061 100644
--- a/workflows/v1/jobs_test.go
+++ b/workflows/v1/jobs_test.go
@@ -258,6 +258,18 @@ func Test_jobClient_Query_WithTaskStates(t *testing.T) {
 	assert.Equal(t, []workflowsv1.TaskState{workflowsv1.TaskState_TASK_STATE_RUNNING, workflowsv1.TaskState_TASK_STATE_FAILED}, service.queryFilters[0].GetTaskStates())
 }
 
+func Test_jobClient_Query_WithClusterSlugs(t *testing.T) {
+	service := &mockJobService{}
+	client := jobClient{service: service}
+
+	jobs, err := Collect(client.Query(context.Background(), job.WithClusterSlugs("default", "gpu")))
+	require.NoError(t, err)
+	assert.Empty(t, jobs)
+
+	require.Len(t, service.queryFilters, 1)
+	assert.Equal(t, []string{"default", "gpu"}, service.queryFilters[0].GetClusterSlugs())
+}
+
 func Test_jobClient_Query_WithLimit(t *testing.T) {
 	nextPage := tileboxv1.Pagination_builder{StartingAfter: tileboxv1.NewUUID(uuid.MustParse("01994da4-255e-740d-9df7-b8c1aa41c75b"))}.Build()
 	service := &mockJobService{
diff --git a/workflows/v1/polling_runner.go b/workflows/v1/polling_runner.go
index b2dbe28..0a1a696 100644
--- a/workflows/v1/polling_runner.go
+++ b/workflows/v1/polling_runner.go
@@ -10,13 +10,18 @@ import (
 	"sync/atomic"
 	"time"
 
+	"connectrpc.com/connect"
 	"github.com/google/uuid"
 	"github.com/samber/lo"
+	"github.com/tilebox/tilebox-go/internal/span"
+	obslogger "github.com/tilebox/tilebox-go/observability/logger"
 	workflowsv1 "github.com/tilebox/tilebox-go/protogen/workflows/v1"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
+const maxTaskFailedRetries = 3
+
 // TaskExecutor is responsible for actually executing a task and keeping track of which tasks can be executed.
 type TaskExecutor interface {
 	// TaskIdentifiers returns the current task capabilities of this executor.
@@ -108,12 +113,17 @@ func (r *PollingTaskRunner) InterruptActiveTask(ctx context.Context) error {
 	if activeTask == nil || activeTask.GetId() == nil || activeTask.GetId().AsUUID() == uuid.Nil {
 		return nil
 	}
-	return r.reportTaskFailed(ctx, activeTask.GetId().AsUUID(), activeTask.GetDisplay(), false, nil)
+	_, err := r.service.TaskFailed(ctx, activeTask.GetId().AsUUID(), activeTask.GetDisplay(), false, nil)
+	return err
 }
 
 func (r *PollingTaskRunner) run(ctx context.Context, stopWhenIdling bool) error {
 	var work *workflowsv1.NextTaskResponse
-	var lastExecutionResult *workflowsv1.ExecuteTaskResponse
+	pending := pendingReport{}
+	// currentTaskLogContext carries task_id and trace context for logs related to the task
+	// that is currently executing or whose result is pending. When no task result is pending,
+	// it is reset to the main runner context so callers never have to handle a nil context.
+	currentTaskLogContext := ctx
 
 	for {
 		if err := ctx.Err(); err != nil {
@@ -123,30 +133,36 @@ func (r *PollingTaskRunner) run(ctx context.Context, stopWhenIdling bool) error
 			return err
 		}
 
-		// check if we need to report a task failure:
-		if lastExecutionResult != nil && lastExecutionResult.GetFailedTask() != nil {
-			// the last task execution resulted in a task failure, so let's report that back to the orchestrator
-			pendingFailed := lastExecutionResult.GetFailedTask()
-			_, err := r.service.TaskFailed(ctx, pendingFailed.GetTaskId().AsUUID(), pendingFailed.GetDisplay(), pendingFailed.GetWasWorkflowError(), pendingFailed.GetProgressUpdates())
-			if err != nil {
-				r.logError(ctx, err, "failed to report task failure", slog.String("task_id", pendingFailed.GetTaskId().AsUUID().String()), slog.String("task_display", pendingFailed.GetDisplay()))
-				// a failure to report a task as failed typically means something is wrong with our API, so let's do one idle time before continuing with the normal runner loop
+		if pending.hasFailureToReport() {
+			var outcome reportOutcome
+			pending, outcome = r.reportPendingFailure(ctx, currentTaskLogContext, pending)
+			switch outcome {
+			case reportSucceeded:
+				currentTaskLogContext = ctx //nolint:fatcontext // Reset task-scoped log state to the root runner context.
+			case reportRetryNow:
+				continue
+			case reportRetryLater:
+				if r.idle(ctx, randomFallbackIdleDuration()) {
+					return nil
+				}
+				continue
+			case reportResetRunner:
+				work = nil
+				pending = pendingReport{}
+				currentTaskLogContext = ctx
 				if r.idle(ctx, randomFallbackIdleDuration()) {
 					return nil
 				}
-				// we did not clear lastExecutionResult yet, so the next loop will attempt to report the task as failed again
 				continue
 			}
-			lastExecutionResult = nil
-			work = nil
 		}
 
 		// we don't have a current task to work on, so let's request one
 		if work == nil || work.GetNextTask() == nil {
 			// while requesting a new task, in the same request we also report the result of the previous task if there is one
 			var lastComputedTask *workflowsv1.ComputedTask
-			if lastExecutionResult.GetComputedTask() != nil {
-				lastComputedTask = lastExecutionResult.GetComputedTask()
+			if pending.hasComputedTaskToReport() {
+				lastComputedTask = pending.result.GetComputedTask()
 			}
 
 			// check whether the runner should still request a new task
@@ -178,7 +194,6 @@ func (r *PollingTaskRunner) run(ctx context.Context, stopWhenIdling bool) error
 			}
 
 			// now let's log a debug message of what exactly we're doing with the NextTask request, depending on whether we have a lastComputedTask, and if we're a requesting a next task or not
-
 			switch {
 			case nextTaskToRun != nil && lastComputedTask != nil:
 				// we report a task result, and immediately request a new task
@@ -192,41 +207,75 @@ func (r *PollingTaskRunner) run(ctx context.Context, stopWhenIdling bool) error
 				r.logger.WarnContext(ctx, "unexpected NextTask request, when both computed_task and next_task is not set")
 			}
 
-			taskResponse, err := r.service.NextTask(ctx, lastComputedTask, nextTaskToRun)
-			if err != nil {
-				r.logError(ctx, err, "failed to request next task")
-				if r.idle(ctx, randomFallbackIdleDuration()) {
-					return nil
+			if lastComputedTask != nil {
+				var outcome reportOutcome
+				work, pending, outcome = r.reportPendingComputed(ctx, currentTaskLogContext, pending, nextTaskToRun)
+				switch outcome {
+				case reportSucceeded:
+					currentTaskLogContext = ctx
+				case reportRetryNow:
+					continue
+				case reportRetryLater:
+					if r.idle(ctx, randomFallbackIdleDuration()) {
+						return nil
+					}
+					continue
+				case reportResetRunner:
+					// during a next task request we should never get that, but just to be defensive in case of future
+					// changes
+					work = nil
+					pending = pendingReport{}
+					currentTaskLogContext = ctx
+					if r.idle(ctx, randomFallbackIdleDuration()) {
+						return nil
+					}
+					continue
 				}
-				continue
+			} else {
+				// work stealing:
+				taskResponse, err := r.service.NextTask(ctx, nil, nextTaskToRun)
+				if err != nil {
+					// easy retry case, since we don't have a task result to report anyway
+					logError(r.logger, ctx, err, "failed to request next task, will retry")
+					if r.idle(ctx, randomFallbackIdleDuration()) {
+						return nil
+					}
+					continue
+				}
+				work = taskResponse
 			}
-			lastExecutionResult = nil
-			work = taskResponse
 		}
 
 		if work != nil && work.GetNextTask() != nil {
 			task := work.GetNextTask()
 			work = nil
 			if isEmpty(task.GetId()) {
-				r.logError(ctx, nil, "got a task without an ID - skipping to the next task")
+				logError(r.logger, ctx, nil, "got a task without an ID - skipping to the next task")
 				continue
 			}
+			currentTaskLogContext = taskContextForLogs(ctx, task)
 			if task.GetRetryCount() > 0 {
-				r.logger.DebugContext(ctx, "retrying task", slog.String("task_id", task.GetId().AsUUID().String()), slog.String("task_display", task.GetDisplay()), slog.Int64("retry_count", task.GetRetryCount()))
+				r.logger.DebugContext(currentTaskLogContext, "retrying task", slog.String("task_display", task.GetDisplay()), slog.Int64("retry_count", task.GetRetryCount()))
 			} else {
-				r.logger.DebugContext(ctx, "executing task", slog.String("task_id", task.GetId().AsUUID().String()), slog.String("task_display", task.GetDisplay()))
+				r.logger.DebugContext(currentTaskLogContext, "executing task", slog.String("task_display", task.GetDisplay()))
 			}
 
 			response, err := r.executeTask(ctx, task)
 			if err != nil {
-				lastExecutionResult = workflowsv1.ExecuteTaskResponse_builder{FailedTask: r.failedTaskFromError(task, err)}.Build()
+				pending = pendingReport{
+					result: workflowsv1.ExecuteTaskResponse_builder{FailedTask: r.failedTaskFromError(task, err)}.Build(),
+				}
 				continue
 			}
 			if response.GetFailedTask() == nil && response.GetComputedTask() == nil {
-				lastExecutionResult = workflowsv1.ExecuteTaskResponse_builder{FailedTask: r.failedTaskFromError(task, errors.New("executor returned neither computed nor failed task"))}.Build()
+				pending = pendingReport{
+					result: workflowsv1.ExecuteTaskResponse_builder{FailedTask: r.failedTaskFromError(task, errors.New("executor returned neither computed nor failed task"))}.Build(),
+				}
 				continue
 			}
-			lastExecutionResult = response
+			pending = pendingReport{
+				result: response,
+			}
 			continue // don't idle, immediately return the result
 		}
 
@@ -287,9 +336,139 @@ func (r *PollingTaskRunner) failedTaskFromError(task *workflowsv1.Task, taskErro
 	}.Build()
 }
 
-func (r *PollingTaskRunner) reportTaskFailed(ctx context.Context, taskID uuid.UUID, display string, wasWorkflowError bool, progressUpdates []*workflowsv1.Progress) error {
-	_, err := r.service.TaskFailed(ctx, taskID, display, wasWorkflowError, progressUpdates)
-	return err
+type reportOutcome int
+
+const (
+	reportSucceeded reportOutcome = iota
+	reportRetryNow
+	reportRetryLater
+	reportResetRunner
+)
+
+type pendingReport struct {
+	result *workflowsv1.ExecuteTaskResponse
+
+	taskFailedAttempts   int
+	taskFailedSimplified bool
+}
+
+func (p pendingReport) hasFailureToReport() bool {
+	return p.result != nil && p.result.HasFailedTask()
+}
+
+func (p pendingReport) hasComputedTaskToReport() bool {
+	return p.result != nil && p.result.HasComputedTask()
+}
+
+// reportPendingFailure owns the retry policy for TaskFailed RPCs. The first request-shaped error
+// returns a pending report with a simplified TaskFailedRequest, so the runner never resends the same
+// rejected user display/progress payload. After simplification, another request-shaped error or three
+// total failed attempts resets the runner state; retryable errors below that cap are retried after idling.
+func (r *PollingTaskRunner) reportPendingFailure(ctx, logCtx context.Context, pending pendingReport) (pendingReport, reportOutcome) {
+	failedTask := pending.result.GetFailedTask()
+	_, err := r.service.TaskFailed(ctx, failedTask.GetTaskId().AsUUID(), failedTask.GetDisplay(), failedTask.GetWasWorkflowError(), failedTask.GetProgressUpdates())
+	if err == nil {
+		return pendingReport{}, reportSucceeded
+	}
+
+	retryAttempt := pending.taskFailedAttempts + 1
+	resetRunnerState := retryAttempt >= maxTaskFailedRetries
+	simplifiedRequest := false
+	failedTaskToRetry := failedTask
+	taskFailedSimplified := pending.taskFailedSimplified
+	if !shouldRetryRPCAfterTimeout(err) {
+		if pending.taskFailedSimplified {
+			resetRunnerState = true
+		} else {
+			failedTaskToRetry = simplifiedTaskFailedRequest(failedTask, err)
+			taskFailedSimplified = true
+			simplifiedRequest = true
+		}
+	}
+
+	// we want to avoid modifying the input argument silently, so we construct a new result to return
+	pendingRetry := pendingReport{
+		result:               workflowsv1.ExecuteTaskResponse_builder{FailedTask: failedTaskToRetry}.Build(),
+		taskFailedAttempts:   retryAttempt,
+		taskFailedSimplified: taskFailedSimplified,
+	}
+
+	logError(r.logger, logCtx, err, "failed to report task failure back to Tilebox", slog.Int("retry_count", retryAttempt), slog.Int("max_retries", maxTaskFailedRetries), slog.Bool("request_simplified", taskFailedSimplified), slog.Bool("resetting_runner", resetRunnerState))
+	if simplifiedRequest && !resetRunnerState {
+		return pendingRetry, reportRetryNow
+	}
+	if resetRunnerState {
+		return pendingRetry, reportResetRunner
+	}
+	return pendingRetry, reportRetryLater
+}
+
+// reportPendingComputed reports a computed task through NextTask. If the combined
+// computed-result-plus-new-work request fails with a request-shaped error, it retries once without
+// the new-work half to avoid blaming task output for cluster/identifier/auth request failures. Only
+// payload-shape errors from an isolated computed-task report are converted into workflow failures.
+func (r *PollingTaskRunner) reportPendingComputed(ctx, logCtx context.Context, pending pendingReport, nextTaskToRun *workflowsv1.NextTaskToRun) (*workflowsv1.NextTaskResponse, pendingReport, reportOutcome) {
+	computedTask := pending.result.GetComputedTask()
+	taskResponse, err := r.service.NextTask(ctx, computedTask, nextTaskToRun)
+	if err == nil {
+		return taskResponse, pendingReport{}, reportSucceeded
+	}
+	if shouldRetryRPCAfterTimeout(err) {
+		logError(r.logger, logCtx, err, "failed to report computed task, will retry", slog.String("task_id", computedTask.GetId().AsUUID().String()), slog.String("task_display", computedTask.GetDisplay()))
+		return nil, pending, reportRetryLater
+	}
+
+	if nextTaskToRun != nil {
+		logError(r.logger, logCtx, err, "failed to report computed task and request next task due to request error, retrying without new work request", slog.String("task_id", computedTask.GetId().AsUUID().String()), slog.String("task_display", computedTask.GetDisplay()))
+		taskResponse, err = r.service.NextTask(ctx, computedTask, nil)
+		if err == nil {
+			return taskResponse, pendingReport{}, reportSucceeded
+		}
+		if shouldRetryRPCAfterTimeout(err) || !shouldFailComputedTaskForRPC(err) {
+			logError(r.logger, logCtx, err, "failed to report computed task without requesting next task, will retry", slog.String("task_id", computedTask.GetId().AsUUID().String()), slog.String("task_display", computedTask.GetDisplay()))
+			return nil, pending, reportRetryLater
+		}
+	}
+
+	if !shouldFailComputedTaskForRPC(err) {
+		logError(r.logger, logCtx, err, "failed to report computed task due to request error, will retry", slog.String("task_id", computedTask.GetId().AsUUID().String()), slog.String("task_display", computedTask.GetDisplay()))
+		return nil, pending, reportRetryLater
+	}
+
+	logError(r.logger, logCtx, err, "failed to report computed task due to invalid payload, will report task as failed", slog.String("task_id", computedTask.GetId().AsUUID().String()), slog.String("task_display", computedTask.GetDisplay()))
+
+	return nil, pendingReport{
+		result:               workflowsv1.ExecuteTaskResponse_builder{FailedTask: failedTaskFromComputedTaskRequestError(computedTask, err)}.Build(),
+		taskFailedAttempts:   0,
+		taskFailedSimplified: false,
+	}, reportRetryNow
+}
+
+func simplifiedTaskFailedRequest(failedTask *workflowsv1.TaskFailedRequest, err error) *workflowsv1.TaskFailedRequest {
+	return workflowsv1.TaskFailedRequest_builder{
+		TaskId:           failedTask.GetTaskId(),
+		Display:          failedTaskDisplay("", err),
+		WasWorkflowError: true,
+	}.Build()
+}
+
+func taskContextForLogs(ctx context.Context, task *workflowsv1.Task) context.Context {
+	if task == nil {
+		return ctx
+	}
+
+	if !isEmpty(task.GetId()) {
+		ctx = obslogger.ContextWithSlogAttributes(ctx, slog.String("task_id", task.GetId().AsUUID().String()))
+	}
+	return span.ContextWithTraceParent(ctx, task.GetJob().GetTraceParent())
+}
+
+func failedTaskFromComputedTaskRequestError(computedTask *workflowsv1.ComputedTask, err error) *workflowsv1.TaskFailedRequest {
+	return workflowsv1.TaskFailedRequest_builder{
+		TaskId:           computedTask.GetId(),
+		Display:          failedTaskDisplay(computedTask.GetDisplay(), err),
+		WasWorkflowError: true,
+	}.Build()
 }
 
 func (r *PollingTaskRunner) extendTaskLease(ctx context.Context, taskID uuid.UUID, initialLease, initialWait time.Duration) {
@@ -306,7 +485,7 @@ func (r *PollingTaskRunner) extendTaskLease(ctx context.Context, taskID uuid.UUI
 		r.logger.DebugContext(ctx, "extending task lease", slog.String("task_id", taskID.String()), slog.Duration("lease", lease), slog.Duration("wait", wait))
 		extension, err := r.service.ExtendTaskLease(ctx, taskID, 2*lease)
 		if err != nil {
-			r.logError(ctx, err, "failed to extend task lease", slog.String("task_id", taskID.String()))
+			logError(r.logger, ctx, err, "failed to extend task lease", slog.String("task_id", taskID.String()))
 			return
 		}
 		if extension.GetLease() == nil {
@@ -333,6 +512,43 @@ func randomFallbackIdleDuration() time.Duration {
 	return fallbackPollingInterval + rand.N(fallbackJitterInterval)
 }
 
+func shouldRetryRPCAfterTimeout(err error) bool {
+	switch connect.CodeOf(err) { //nolint:exhaustive // Only request-error codes are special-cased; all other codes retry.
+	case connect.CodeInvalidArgument,
+		connect.CodeNotFound,
+		connect.CodeAlreadyExists,
+		connect.CodeFailedPrecondition,
+		connect.CodeOutOfRange,
+		connect.CodeUnimplemented:
+		return false
+	}
+
+	switch status.Code(err) { //nolint:exhaustive // Only request-error codes are special-cased; all other codes retry.
+	case codes.InvalidArgument,
+		codes.NotFound,
+		codes.AlreadyExists,
+		codes.FailedPrecondition,
+		codes.OutOfRange,
+		codes.Unimplemented:
+		return false
+	}
+	return true
+}
+
+func shouldFailComputedTaskForRPC(err error) bool {
+	switch connect.CodeOf(err) { //nolint:exhaustive // Only payload-shape errors fail the computed task.
+	case connect.CodeInvalidArgument, connect.CodeOutOfRange:
+		return true
+	}
+
+	switch status.Code(err) { //nolint:exhaustive // Only payload-shape errors fail the computed task.
+	case codes.InvalidArgument, codes.OutOfRange:
+		return true
+	default:
+		return false
+	}
+}
+
 // idle for the given duration, and return true if idling was interrupted due to a context error (e.g. cancellation)
 func (r *PollingTaskRunner) idle(ctx context.Context, duration time.Duration) bool {
 	timer := time.NewTimer(duration)
@@ -345,10 +561,12 @@ func (r *PollingTaskRunner) idle(ctx context.Context, duration time.Duration) bo
 	}
 }
 
-func (r *PollingTaskRunner) logError(ctx context.Context, err error, msg string, args ...any) {
+func logError(logger *slog.Logger, ctx context.Context, err error, msg string, args ...any) {
 	switch {
 	case errors.Is(err, context.Canceled):
 		return
+	case connect.CodeOf(err) == connect.CodeCanceled:
+		return
 	case status.Code(err) == codes.Canceled:
 		return
 	}
@@ -356,5 +574,5 @@ func (r *PollingTaskRunner) logError(ctx context.Context, err error, msg string,
 	fields := make([]any, 0, len(args)+1)
 	fields = append(fields, slog.Any("error", err))
 	fields = append(fields, args...)
-	r.logger.ErrorContext(ctx, msg, fields...)
+	logger.ErrorContext(ctx, msg, fields...)
 }
diff --git a/workflows/v1/polling_runner_test.go b/workflows/v1/polling_runner_test.go
new file mode 100644
index 0000000..2f2dc44
--- /dev/null
+++ b/workflows/v1/polling_runner_test.go
@@ -0,0 +1,319 @@
+package workflows
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"testing"
+	"time"
+
+	"connectrpc.com/connect"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	tileboxv1 "github.com/tilebox/tilebox-go/protogen/tilebox/v1"
+	workflowsv1 "github.com/tilebox/tilebox-go/protogen/workflows/v1"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/durationpb"
+)
+
+func TestPollingTaskRunnerReportsExecutorReturnedFailedTaskAsIs(t *testing.T) {
+	taskID := uuid.New()
+	taskDisplay := "Python task"
+	mockNextTask := workflowsv1.Task_builder{
+		Id: tileboxv1.NewUUID(taskID),
+		Identifier: workflowsv1.TaskIdentifier_builder{
+			Name:    "python.Task",
+			Version: "v1.0",
+		}.Build(),
+		State:   workflowsv1.TaskState_TASK_STATE_RUNNING,
+		Display: &taskDisplay,
+		Lease: workflowsv1.TaskLease_builder{
+			Lease:                             durationpb.New(5 * time.Minute),
+			RecommendedWaitUntilNextExtension: durationpb.New(5 * time.Minute),
+		}.Build(),
+	}.Build()
+	progressUpdates := []*workflowsv1.Progress{
+		workflowsv1.Progress_builder{Label: "work", Total: 10, Done: 4}.Build(),
+	}
+	service := &mockMinimalTaskService{nextTask: mockNextTask}
+	executor := &failedResponseExecutor{
+		response: workflowsv1.ExecuteTaskResponse_builder{
+			FailedTask: workflowsv1.TaskFailedRequest_builder{
+				TaskId:           tileboxv1.NewUUID(taskID),
+				Display:          "Python classified failure",
+				WasWorkflowError: true,
+				ProgressUpdates:  progressUpdates,
+			}.Build(),
+		}.Build(),
+	}
+	runner := NewPollingTaskRunner(service, "default", executor, slog.Default())
+
+	require.NoError(t, runner.RunAll(context.Background()))
+
+	require.True(t, service.failed)
+	require.Equal(t, "Python classified failure", service.failedDisplay)
+	require.True(t, service.failedWasWorkflowError)
+	require.Equal(t, progressUpdates, service.failedProgressUpdates)
+}
+
+func TestPollingTaskRunnerReportsComputedTaskRequestErrorAsWorkflowFailure(t *testing.T) {
+	taskID := uuid.New()
+	taskDisplay := "Python task"
+	mockNextTask := workflowsv1.Task_builder{
+		Id: tileboxv1.NewUUID(taskID),
+		Identifier: workflowsv1.TaskIdentifier_builder{
+			Name:    "python.Task",
+			Version: "v1.0",
+		}.Build(),
+		State:   workflowsv1.TaskState_TASK_STATE_RUNNING,
+		Display: &taskDisplay,
+		Lease: workflowsv1.TaskLease_builder{
+			Lease:                             durationpb.New(5 * time.Minute),
+			RecommendedWaitUntilNextExtension: durationpb.New(5 * time.Minute),
+		}.Build(),
+	}.Build()
+	service := &requestErrorTaskService{
+		nextTask:                mockNextTask,
+		nextTaskComputedTaskErr: connect.NewError(connect.CodeInvalidArgument, errors.New("invalid computed task")),
+	}
+	executor := &failedResponseExecutor{
+		response: workflowsv1.ExecuteTaskResponse_builder{
+			ComputedTask: workflowsv1.ComputedTask_builder{
+				Id:      tileboxv1.NewUUID(taskID),
+				Display: taskDisplay,
+			}.Build(),
+		}.Build(),
+	}
+	runner := NewPollingTaskRunner(service, "default", executor, slog.Default())
+
+	require.NoError(t, runner.RunAll(context.Background()))
+
+	require.Len(t, service.taskFailedRequests, 1)
+	failedTask := service.taskFailedRequests[0]
+	assert.Equal(t, taskID, failedTask.GetTaskId().AsUUID())
+	assert.True(t, failedTask.GetWasWorkflowError())
+	assert.Contains(t, failedTask.GetDisplay(), taskDisplay)
+	assert.Contains(t, failedTask.GetDisplay(), "invalid computed task")
+	assert.Empty(t, failedTask.GetProgressUpdates())
+}
+
+func TestPollingTaskRunnerDoesNotFailComputedTaskWhenCombinedNextTaskRequestIsInvalid(t *testing.T) {
+	taskID := uuid.New()
+	taskDisplay := "Python task"
+	mockNextTask := workflowsv1.Task_builder{
+		Id: tileboxv1.NewUUID(taskID),
+		Identifier: workflowsv1.TaskIdentifier_builder{
+			Name:    "python.Task",
+			Version: "v1.0",
+		}.Build(),
+		State:   workflowsv1.TaskState_TASK_STATE_RUNNING,
+		Display: &taskDisplay,
+		Lease: workflowsv1.TaskLease_builder{
+			Lease:                             durationpb.New(5 * time.Minute),
+			RecommendedWaitUntilNextExtension: durationpb.New(5 * time.Minute),
+		}.Build(),
+	}.Build()
+	service := &requestErrorTaskService{
+		nextTask:                      mockNextTask,
+		nextTaskComputedAndRequestErr: connect.NewError(connect.CodeInvalidArgument, errors.New("invalid work request")),
+	}
+	executor := &failedResponseExecutor{
+		response: workflowsv1.ExecuteTaskResponse_builder{
+			ComputedTask: workflowsv1.ComputedTask_builder{
+				Id:      tileboxv1.NewUUID(taskID),
+				Display: taskDisplay,
+			}.Build(),
+		}.Build(),
+	}
+	runner := NewPollingTaskRunner(service, "default", executor, slog.Default())
+
+	require.NoError(t, runner.RunAll(context.Background()))
+
+	assert.Empty(t, service.taskFailedRequests)
+	require.Len(t, service.computedTasks, 1)
+	assert.Equal(t, taskID, service.computedTasks[0].GetId().AsUUID())
+}
+
+func TestPollingTaskRunnerRetriesTaskFailedRequestErrorOnceAsWorkflowFailure(t *testing.T) {
+	taskID := uuid.New()
+	taskDisplay := "Python task"
+	mockNextTask := workflowsv1.Task_builder{
+		Id: tileboxv1.NewUUID(taskID),
+		Identifier: workflowsv1.TaskIdentifier_builder{
+			Name:    "python.Task",
+			Version: "v1.0",
+		}.Build(),
+		State:   workflowsv1.TaskState_TASK_STATE_RUNNING,
+		Display: &taskDisplay,
+		Lease: workflowsv1.TaskLease_builder{
+			Lease:                             durationpb.New(5 * time.Minute),
+			RecommendedWaitUntilNextExtension: durationpb.New(5 * time.Minute),
+		}.Build(),
+	}.Build()
+	service := &requestErrorTaskService{
+		nextTask: mockNextTask,
+		taskFailedErrors: []error{
+			connect.NewError(connect.CodeInvalidArgument, errors.New("invalid failed task request")),
+		},
+	}
+	progressUpdates := []*workflowsv1.Progress{
+		workflowsv1.Progress_builder{Label: "work", Total: 10, Done: 4}.Build(),
+	}
+	executor := &failedResponseExecutor{
+		response: workflowsv1.ExecuteTaskResponse_builder{
+			FailedTask: workflowsv1.TaskFailedRequest_builder{
+				TaskId:           tileboxv1.NewUUID(taskID),
+				Display:          "user defined display",
+				WasWorkflowError: false,
+				ProgressUpdates:  progressUpdates,
+			}.Build(),
+		}.Build(),
+	}
+	runner := NewPollingTaskRunner(service, "default", executor, slog.Default())
+
+	require.NoError(t, runner.RunAll(context.Background()))
+
+	require.Len(t, service.taskFailedRequests, 2)
+	assert.Equal(t, "user defined display", service.taskFailedRequests[0].GetDisplay())
+	assert.False(t, service.taskFailedRequests[0].GetWasWorkflowError())
+	assert.Equal(t, progressUpdates, service.taskFailedRequests[0].GetProgressUpdates())
+	assert.NotContains(t, service.taskFailedRequests[1].GetDisplay(), "user defined display")
+	assert.Contains(t, service.taskFailedRequests[1].GetDisplay(), "invalid failed task request")
+	assert.True(t, service.taskFailedRequests[1].GetWasWorkflowError())
+	assert.Empty(t, service.taskFailedRequests[1].GetProgressUpdates())
+}
+
+func TestPollingTaskRunnerStopsRetryingTaskFailedAfterSecondRequestError(t *testing.T) {
+	taskID := uuid.New()
+	mockNextTask := workflowsv1.Task_builder{
+		Id: tileboxv1.NewUUID(taskID),
+		Identifier: workflowsv1.TaskIdentifier_builder{
+			Name:    "python.Task",
+			Version: "v1.0",
+		}.Build(),
+		State: workflowsv1.TaskState_TASK_STATE_RUNNING,
+		Lease: workflowsv1.TaskLease_builder{
+			Lease:                             durationpb.New(5 * time.Minute),
+			RecommendedWaitUntilNextExtension: durationpb.New(5 * time.Minute),
+		}.Build(),
+	}.Build()
+	service := &requestErrorTaskService{
+		nextTask: mockNextTask,
+		taskFailedErrors: []error{
+			connect.NewError(connect.CodeInvalidArgument, errors.New("invalid failed task request")),
+			connect.NewError(connect.CodeInvalidArgument, errors.New("still invalid")),
+		},
+	}
+	executor := &failedResponseExecutor{
+		response: workflowsv1.ExecuteTaskResponse_builder{
+			FailedTask: workflowsv1.TaskFailedRequest_builder{
+				TaskId:  tileboxv1.NewUUID(taskID),
+				Display: "user defined display",
+			}.Build(),
+		}.Build(),
+	}
+	runner := NewPollingTaskRunner(service, "default", executor, slog.Default())
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
+	defer cancel()
+
+	require.NoError(t, runner.RunAll(ctx))
+
+	require.Len(t, service.taskFailedRequests, 2)
+	assert.Equal(t, "user defined display", service.taskFailedRequests[0].GetDisplay())
+	assert.NotContains(t, service.taskFailedRequests[1].GetDisplay(), "user defined display")
+}
+
+func TestPollingTaskRunnerResetsTaskFailedAfterMaxRetryableErrors(t *testing.T) {
+	taskID := uuid.New()
+	service := &requestErrorTaskService{
+		taskFailedErrors: []error{
+			connect.NewError(connect.CodeUnavailable, errors.New("unavailable 1")),
+			connect.NewError(connect.CodeUnavailable, errors.New("unavailable 2")),
+			connect.NewError(connect.CodeUnavailable, errors.New("unavailable 3")),
+		},
+	}
+	runner := NewPollingTaskRunner(service, "default", &failedResponseExecutor{}, slog.Default())
+	pending := pendingReport{
+		result: workflowsv1.ExecuteTaskResponse_builder{
+			FailedTask: workflowsv1.TaskFailedRequest_builder{
+				TaskId:  tileboxv1.NewUUID(taskID),
+				Display: "task failed",
+			}.Build(),
+		}.Build(),
+	}
+
+	var outcome reportOutcome
+	pending, outcome = runner.reportPendingFailure(context.Background(), context.Background(), pending)
+	assert.Equal(t, reportRetryLater, outcome)
+	pending, outcome = runner.reportPendingFailure(context.Background(), context.Background(), pending)
+	assert.Equal(t, reportRetryLater, outcome)
+	_, outcome = runner.reportPendingFailure(context.Background(), context.Background(), pending)
+	assert.Equal(t, reportResetRunner, outcome)
+	require.Len(t, service.taskFailedRequests, maxTaskFailedRetries)
+}
+
+type failedResponseExecutor struct {
+	response *workflowsv1.ExecuteTaskResponse
+}
+
+func (e *failedResponseExecutor) TaskIdentifiers() []*workflowsv1.TaskIdentifier {
+	return []*workflowsv1.TaskIdentifier{
+		workflowsv1.TaskIdentifier_builder{Name: "python.Task", Version: "v1.0"}.Build(),
+	}
+}
+
+func (e *failedResponseExecutor) ExecuteTask(context.Context, *workflowsv1.Task) (*workflowsv1.ExecuteTaskResponse, error) {
+	return e.response, nil
+}
+
+type requestErrorTaskService struct {
+	computedTasks                 []*workflowsv1.ComputedTask
+	nextTask                      *workflowsv1.Task
+	nextTaskComputedAndRequestErr error
+	nextTaskComputedTaskErr       error
+	taskFailedErrors              []error
+	taskFailedRequests            []*workflowsv1.TaskFailedRequest
+}
+
+var _ TaskService = &requestErrorTaskService{}
+
+func (s *requestErrorTaskService) NextTask(_ context.Context, computedTask *workflowsv1.ComputedTask, nextTaskToRun *workflowsv1.NextTaskToRun) (*workflowsv1.NextTaskResponse, error) {
+	if computedTask != nil && nextTaskToRun != nil && s.nextTaskComputedAndRequestErr != nil {
+		return nil, s.nextTaskComputedAndRequestErr
+	}
+	if computedTask != nil && s.nextTaskComputedTaskErr != nil {
+		return nil, s.nextTaskComputedTaskErr
+	}
+	if computedTask != nil {
+		s.computedTasks = append(s.computedTasks, computedTask)
+	}
+	if s.nextTask != nil {
+		response := workflowsv1.NextTaskResponse_builder{NextTask: proto.CloneOf(s.nextTask)}.Build()
+		s.nextTask = nil
+		return response, nil
+	}
+	return workflowsv1.NextTaskResponse_builder{}.Build(), nil
+}
+
+func (s *requestErrorTaskService) TaskFailed(_ context.Context, taskID uuid.UUID, display string, wasWorkflowError bool, progressUpdates []*workflowsv1.Progress) (*workflowsv1.TaskStateResponse, error) {
+	s.taskFailedRequests = append(s.taskFailedRequests, workflowsv1.TaskFailedRequest_builder{
+		TaskId:           tileboxv1.NewUUID(taskID),
+		Display:          display,
+		WasWorkflowError: wasWorkflowError,
+		ProgressUpdates:  progressUpdates,
+	}.Build())
+	if len(s.taskFailedErrors) > 0 {
+		err := s.taskFailedErrors[0]
+		s.taskFailedErrors = s.taskFailedErrors[1:]
+		return nil, err
+	}
+	return workflowsv1.TaskStateResponse_builder{State: workflowsv1.TaskState_TASK_STATE_FAILED}.Build(), nil
+}
+
+func (s *requestErrorTaskService) ExtendTaskLease(context.Context, uuid.UUID, time.Duration) (*workflowsv1.TaskLease, error) {
+	return workflowsv1.TaskLease_builder{
+		Lease:                             durationpb.New(5 * time.Minute),
+		RecommendedWaitUntilNextExtension: durationpb.New(5 * time.Minute),
+	}.Build(), nil
+}
diff --git a/workflows/v1/runner_test.go b/workflows/v1/runner_test.go
index 3963fda..c453406 100644
--- a/workflows/v1/runner_test.go
+++ b/workflows/v1/runner_test.go
@@ -391,60 +391,6 @@ func TestTaskRunner_RunForeverReportsComputedTaskOnce(t *testing.T) {
 	assert.Equal(t, mockNextTask.GetId(), mockTaskClient.computedTasks[0].GetId())
 }
 
-func TestPollingTaskRunnerReportsExecutorReturnedFailedTaskAsIs(t *testing.T) {
-	taskID := uuid.New()
-	taskDisplay := "Python task"
-	mockNextTask := workflowsv1.Task_builder{
-		Id: tileboxv1.NewUUID(taskID),
-		Identifier: workflowsv1.TaskIdentifier_builder{
-			Name:    "python.Task",
-			Version: "v1.0",
-		}.Build(),
-		State:   workflowsv1.TaskState_TASK_STATE_RUNNING,
-		Display: &taskDisplay,
-		Lease: workflowsv1.TaskLease_builder{
-			Lease:                             durationpb.New(5 * time.Minute),
-			RecommendedWaitUntilNextExtension: durationpb.New(5 * time.Minute),
-		}.Build(),
-	}.Build()
-	progressUpdates := []*workflowsv1.Progress{
-		workflowsv1.Progress_builder{Label: "work", Total: 10, Done: 4}.Build(),
-	}
-	service := &mockMinimalTaskService{nextTask: mockNextTask}
-	executor := &failedResponseExecutor{
-		response: workflowsv1.ExecuteTaskResponse_builder{
-			FailedTask: workflowsv1.TaskFailedRequest_builder{
-				TaskId:           tileboxv1.NewUUID(taskID),
-				Display:          "Python classified failure",
-				WasWorkflowError: true,
-				ProgressUpdates:  progressUpdates,
-			}.Build(),
-		}.Build(),
-	}
-	runner := NewPollingTaskRunner(service, "default", executor, slog.Default())
-
-	require.NoError(t, runner.RunAll(context.Background()))
-
-	require.True(t, service.failed)
-	require.Equal(t, "Python classified failure", service.failedDisplay)
-	require.True(t, service.failedWasWorkflowError)
-	require.Equal(t, progressUpdates, service.failedProgressUpdates)
-}
-
-type failedResponseExecutor struct {
-	response *workflowsv1.ExecuteTaskResponse
-}
-
-func (e *failedResponseExecutor) TaskIdentifiers() []*workflowsv1.TaskIdentifier {
-	return []*workflowsv1.TaskIdentifier{
-		workflowsv1.TaskIdentifier_builder{Name: "python.Task", Version: "v1.0"}.Build(),
-	}
-}
-
-func (e *failedResponseExecutor) ExecuteTask(context.Context, *workflowsv1.Task) (*workflowsv1.ExecuteTaskResponse, error) {
-	return e.response, nil
-}
-
 func Test_withTaskExecutionContextRoundtrip(t *testing.T) {
 	tracer := noop.NewTracerProvider().Tracer("")
 	service := mockTaskService{}
diff --git a/workflows/v1/service.go b/workflows/v1/service.go
index f9868ca..c62a9ac 100644
--- a/workflows/v1/service.go
+++ b/workflows/v1/service.go
@@ -360,7 +360,9 @@ type WorkflowService interface {
 	CreateWorkflow(ctx context.Context, name, description string) (*workflowsv1.Workflow, error)
 	ListWorkflows(ctx context.Context) (*workflowsv1.ListWorkflowsResponse, error)
 	GetWorkflow(ctx context.Context, slug string) (*workflowsv1.Workflow, error)
+	DeleteWorkflow(ctx context.Context, slug string) error
 	PublishWorkflowRelease(ctx context.Context, workflowSlug string, artifactID uuid.UUID, content *ReleaseContent) (*workflowsv1.WorkflowRelease, error)
+	UnpublishWorkflowRelease(ctx context.Context, workflowSlug string, releaseID uuid.UUID) error
 	DeployWorkflowRelease(ctx context.Context, workflowSlug string, releaseID uuid.UUID, clusterSlugs []string) (*workflowsv1.DeployWorkflowReleaseResponse, error)
 	UndeployWorkflowRelease(ctx context.Context, workflowSlug string, releaseID uuid.UUID, clusterSlugs []string) (*workflowsv1.UndeployWorkflowReleaseResponse, error)
 }
@@ -467,6 +469,19 @@ func (s *workflowService) GetWorkflow(ctx context.Context, slug string) (*workfl
 	})
 }
 
+func (s *workflowService) DeleteWorkflow(ctx context.Context, slug string) error {
+	return observability.WithSpan(ctx, s.tracer, "workflows/workflows/delete", func(ctx context.Context) error {
+		_, err := s.workflowClient.DeleteWorkflow(ctx, connect.NewRequest(workflowsv1.DeleteWorkflowRequest_builder{
+			WorkflowSlug: slug,
+		}.Build()))
+		if err != nil {
+			return fmt.Errorf("failed to delete workflow: %w", err)
+		}
+
+		return nil
+	})
+}
+
 func (s *workflowService) PublishWorkflowRelease(ctx context.Context, workflowSlug string, artifactID uuid.UUID, content *ReleaseContent) (*workflowsv1.WorkflowRelease, error) {
 	return observability.WithSpanResult(ctx, s.tracer, "workflows/workflows/publish_release", func(ctx context.Context) (*workflowsv1.WorkflowRelease, error) {
 		protoContent, err := releaseContentToProto(content)
@@ -487,6 +502,20 @@ func (s *workflowService) PublishWorkflowRelease(ctx context.Context, workflowSl
 	})
 }
 
+func (s *workflowService) UnpublishWorkflowRelease(ctx context.Context, workflowSlug string, releaseID uuid.UUID) error {
+	return observability.WithSpan(ctx, s.tracer, "workflows/workflows/unpublish_release", func(ctx context.Context) error {
+		_, err := s.workflowClient.UnpublishWorkflowRelease(ctx, connect.NewRequest(workflowsv1.UnpublishWorkflowReleaseRequest_builder{
+			WorkflowSlug: workflowSlug,
+			ReleaseId:    tileboxv1.NewUUID(releaseID),
+		}.Build()))
+		if err != nil {
+			return fmt.Errorf("failed to unpublish workflow release: %w", err)
+		}
+
+		return nil
+	})
+}
+
 func (s *workflowService) DeployWorkflowRelease(ctx context.Context, workflowSlug string, releaseID uuid.UUID, clusterSlugs []string) (*workflowsv1.DeployWorkflowReleaseResponse, error) {
 	return observability.WithSpanResult(ctx, s.tracer, "workflows/workflows/deploy_release", func(ctx context.Context) (*workflowsv1.DeployWorkflowReleaseResponse, error) {
 		res, err := s.workflowClient.DeployWorkflowRelease(ctx, connect.NewRequest(workflowsv1.DeployWorkflowReleaseRequest_builder{
diff --git a/workflows/v1/workflows.go b/workflows/v1/workflows.go
index 0688073..2fbfabb 100644
--- a/workflows/v1/workflows.go
+++ b/workflows/v1/workflows.go
@@ -106,9 +106,15 @@ type WorkflowClient interface {
 	// Get returns a workflow by its slug.
 	Get(ctx context.Context, slug string) (*Workflow, error)
 
+	// Delete deletes a workflow by its slug.
+	Delete(ctx context.Context, slug string) error
+
 	// PublishRelease publishes a new immutable release for a workflow.
 	PublishRelease(ctx context.Context, workflowSlug string, artifactID uuid.UUID, content *ReleaseContent) (*WorkflowRelease, error)
 
+	// UnpublishRelease unpublishes a workflow release.
+	UnpublishRelease(ctx context.Context, workflowSlug string, releaseID uuid.UUID) error
+
 	// DeployRelease deploys a workflow release to clusters.
 	DeployRelease(ctx context.Context, workflowSlug string, releaseID uuid.UUID, clusterSlugs []string) (*WorkflowReleaseDeployment, error)
 
@@ -155,6 +161,10 @@ func (c workflowClient) Get(ctx context.Context, slug string) (*Workflow, error)
 	return protoToWorkflow(response), nil
 }
 
+func (c workflowClient) Delete(ctx context.Context, slug string) error {
+	return c.service.DeleteWorkflow(ctx, slug)
+}
+
 func (c workflowClient) PublishRelease(ctx context.Context, workflowSlug string, artifactID uuid.UUID, content *ReleaseContent) (*WorkflowRelease, error) {
 	response, err := c.service.PublishWorkflowRelease(ctx, workflowSlug, artifactID, content)
 	if err != nil {
@@ -164,6 +174,10 @@ func (c workflowClient) PublishRelease(ctx context.Context, workflowSlug string,
 	return protoToWorkflowRelease(response), nil
 }
 
+func (c workflowClient) UnpublishRelease(ctx context.Context, workflowSlug string, releaseID uuid.UUID) error {
+	return c.service.UnpublishWorkflowRelease(ctx, workflowSlug, releaseID)
+}
+
 func (c workflowClient) DeployRelease(ctx context.Context, workflowSlug string, releaseID uuid.UUID, clusterSlugs []string) (*WorkflowReleaseDeployment, error) {
 	response, err := c.service.DeployWorkflowRelease(ctx, workflowSlug, releaseID, clusterSlugs)
 	if err != nil {
diff --git a/workflows/v1/workflows_test.go b/workflows/v1/workflows_test.go
index 3fc53ff..268738b 100644
--- a/workflows/v1/workflows_test.go
+++ b/workflows/v1/workflows_test.go
@@ -70,6 +70,17 @@ func TestWorkflowClient_Get(t *testing.T) {
 	assert.Equal(t, "Agentic Workflow", workflow.Name)
 }
 
+func TestWorkflowClient_Delete(t *testing.T) {
+	ctx := context.Background()
+	service := &fakeWorkflowService{}
+	client := workflowClient{service: service}
+
+	err := client.Delete(ctx, "agentic-workflow")
+	require.NoError(t, err)
+
+	assert.Equal(t, "agentic-workflow", service.deleteWorkflowSlug)
+}
+
 func TestWorkflowClient_PublishRelease(t *testing.T) {
 	ctx := context.Background()
 	releaseID := uuid.New()
@@ -132,6 +143,19 @@ func TestWorkflowClient_PublishRelease(t *testing.T) {
 	assert.Equal(t, createdAt, release.CreatedAt)
 }
 
+func TestWorkflowClient_UnpublishRelease(t *testing.T) {
+	ctx := context.Background()
+	releaseID := uuid.New()
+	service := &fakeWorkflowService{}
+	client := workflowClient{service: service}
+
+	err := client.UnpublishRelease(ctx, "agentic-workflow", releaseID)
+	require.NoError(t, err)
+
+	assert.Equal(t, "agentic-workflow", service.unpublishWorkflowSlug)
+	assert.Equal(t, releaseID, service.unpublishReleaseID)
+}
+
 func TestWorkflowClient_DeployRelease(t *testing.T) {
 	ctx := context.Background()
 	releaseID := uuid.New()
@@ -245,14 +269,18 @@ type fakeWorkflowService struct {
 	undeployWorkflowReleaseResponse *workflowsv1.UndeployWorkflowReleaseResponse
 	err                             error
 
-	createName        string
-	createDescription string
-	getSlug           string
+	createName         string
+	createDescription  string
+	getSlug            string
+	deleteWorkflowSlug string
 
 	publishWorkflowSlug string
 	publishArtifactID   uuid.UUID
 	publishContent      *ReleaseContent
 
+	unpublishWorkflowSlug string
+	unpublishReleaseID    uuid.UUID
+
 	deployWorkflowSlug string
 	deployReleaseID    uuid.UUID
 	deployClusterSlugs []string
@@ -293,6 +321,11 @@ func (s *fakeWorkflowService) GetWorkflow(_ context.Context, slug string) (*work
 	return s.workflow, s.err
 }
 
+func (s *fakeWorkflowService) DeleteWorkflow(_ context.Context, slug string) error {
+	s.deleteWorkflowSlug = slug
+	return s.err
+}
+
 func (s *fakeWorkflowService) PublishWorkflowRelease(_ context.Context, workflowSlug string, artifactID uuid.UUID, content *ReleaseContent) (*workflowsv1.WorkflowRelease, error) {
 	s.publishWorkflowSlug = workflowSlug
 	s.publishArtifactID = artifactID
@@ -300,6 +333,12 @@ func (s *fakeWorkflowService) PublishWorkflowRelease(_ context.Context, workflow
 	return s.workflowRelease, s.err
 }
 
+func (s *fakeWorkflowService) UnpublishWorkflowRelease(_ context.Context, workflowSlug string, releaseID uuid.UUID) error {
+	s.unpublishWorkflowSlug = workflowSlug
+	s.unpublishReleaseID = releaseID
+	return s.err
+}
+
 func (s *fakeWorkflowService) DeployWorkflowRelease(_ context.Context, workflowSlug string, releaseID uuid.UUID, clusterSlugs []string) (*workflowsv1.DeployWorkflowReleaseResponse, error) {
 	s.deployWorkflowSlug = workflowSlug
 	s.deployReleaseID = releaseID