Add workflow concurrency limit option (#6671)

Co-authored-by: Lauris B <lauris@nix.lv>
Co-authored-by: 6543 <6543@obermui.de>

cmd/server/openapi/docs.go

@@ -5715,6 +5715,18 @@ const docTemplate = `{
                 "agent_id": {
                     "type": "integer"
                 },
+                "concurrency_group": {
+                    "description": "ConcurrencyGroup identifies tasks that are limited against each other.\nIt is empty when no concurrency limit applies.",
+                    "type": "string"
+                },
+                "concurrency_limit": {
+                    "description": "ConcurrencyLimit is the maximum number of tasks sharing the same\nConcurrencyGroup that may run at once. A value \u003c= 0 means unlimited.",
+                    "type": "integer"
+                },
+                "created": {
+                    "description": "Created is the unix timestamp the task's pipeline was created at. It\ndefines the queue ordering across pipelines.",
+                    "type": "integer"
+                },
                 "dep_status": {
                     "type": "object",
                     "additionalProperties": {
@@ -6198,6 +6210,18 @@ const docTemplate = `{
                 "agent_name": {
                     "type": "string"
                 },
+                "concurrency_group": {
+                    "description": "ConcurrencyGroup identifies tasks that are limited against each other.\nIt is empty when no concurrency limit applies.",
+                    "type": "string"
+                },
+                "concurrency_limit": {
+                    "description": "ConcurrencyLimit is the maximum number of tasks sharing the same\nConcurrencyGroup that may run at once. A value \u003c= 0 means unlimited.",
+                    "type": "integer"
+                },
+                "created": {
+                    "description": "Created is the unix timestamp the task's pipeline was created at. It\ndefines the queue ordering across pipelines.",
+                    "type": "integer"
+                },
                 "dep_status": {
                     "type": "object",
                     "additionalProperties": {

docs/docs/20-usage/25-workflows.md

@@ -144,3 +144,45 @@ The same syntax works at the step level within a workflow: if a step uses `depen
 Some workflows don't need the source code, like creating a notification on failure.
 Read more about `skip_clone` at [pipeline syntax](./20-workflow-syntax.md#skip_clone)
 :::
+
+## Concurrency
+
+By default workflows run with no concurrency limit. Some workflows, however, must not run more than a given number of times at once. A typical example is a deployment workflow: running two deployments at the same time can cause race conditions or corrupt state. Cancelling the previous pipeline is often not an option either, since it could interrupt an ongoing deployment.
+
+The `concurrency` setting limits how many instances of a workflow may run at the same time. When the limit is reached, additional instances stay queued and start only once a running one has finished. Nothing is cancelled.
+
+```yaml title=".woodpecker/deploy.yaml"
+steps:
+  - name: deploy
+    image: debian:stable-slim
+    commands:
+      - echo deploying
+
+depends_on:
+  - test
+
+concurrency:
+  limit: 1
+```
+
+You can also use the shorthand form to only set the limit:
+
+```yaml
+concurrency: 1
+```
+
+### Ordering
+
+Queued workflows of the same group start in the order their pipelines were created, **not** in the order they become ready to run. This matters when a workflow depends on other workflows (via `depends_on`) whose duration varies: even if a later pipeline's checks finish first, its limited workflow will not overtake an earlier pipeline that is still waiting. This guarantees that, for example, deployments happen in commit order.
+
+### Groups
+
+By default, the limit applies per workflow within a repository. Different runs of the same workflow are limited against each other, while different workflows (and other repositories) are unaffected.
+
+Setting a `group` is optional. You can set a custom `group` to share a limit across workflows or to make the limit more specific. The group supports [environment variable substitution](./50-environment.md), so you can, for example, limit concurrency per branch or per deployment target:
+
+```yaml
+concurrency:
+  limit: 1
+  group: deploy-${CI_COMMIT_BRANCH}
+```

e2e/scenarios/concurrency_test.go

@@ -0,0 +1,173 @@
+// Copyright 2026 Woodpecker Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build test
+
+package scenarios
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"go.woodpecker-ci.org/woodpecker/v3/e2e/setup"
+	forge_types "go.woodpecker-ci.org/woodpecker/v3/server/forge/types"
+	"go.woodpecker-ci.org/woodpecker/v3/server/model"
+	"go.woodpecker-ci.org/woodpecker/v3/server/pipeline"
+)
+
+// Two workflows in one pipeline:
+//   - workflow A opts into concurrency limit 1 (fully serialized across runs)
+//   - workflow B opts into concurrency limit 2 (at most two in flight at once)
+//
+// The concurrency group defaults to the workflow name and is scoped per repo,
+// so every A instance across all pipeline runs shares one group, and likewise
+// for B. Each step sleeps so the workflows overlap in wall-clock time, making
+// the limit observable through the recorded start/finish timestamps.
+var (
+	concurrencyWorkflowA = []byte(`
+skip_clone: true
+concurrency: 1
+steps:
+  - name: work
+    image: dummy
+    commands:
+      - echo workflow-a
+    environment:
+      SLEEP: "2s"
+`)
+
+	concurrencyWorkflowB = []byte(`
+skip_clone: true
+concurrency:
+  limit: 2
+steps:
+  - name: work
+    image: dummy
+    commands:
+      - echo workflow-b
+    environment:
+      SLEEP: "2s"
+`)
+)
+
+// TestWorkflowConcurrencyLimit runs three rounds of a two-workflow pipeline on
+// an agent with six free slots — enough capacity that the concurrency limits,
+// not the agent, are the only thing constraining parallelism. It then inspects
+// the recorded workflow timings and asserts:
+//   - every pipeline succeeds,
+//   - workflow A (limit 1) never overlaps itself,
+//   - workflow B (limit 2) never exceeds two concurrent instances and does in
+//     fact reach two at some point (otherwise the test would not distinguish a
+//     working limit-2 from an accidental limit-1).
+func TestWorkflowConcurrencyLimit(t *testing.T) {
+	const rounds = 3
+
+	env := setup.StartServer(t.Context(), t, []*forge_types.FileMeta{
+		{Name: ".woodpecker/a.yaml", Data: concurrencyWorkflowA},
+		{Name: ".woodpecker/b.yaml", Data: concurrencyWorkflowB},
+	})
+
+	// Six slots: 3 rounds × 2 workflows = 6 workflows could all run at once
+	// if nothing limited them, so any serialization we observe is the
+	// concurrency limit at work, not slot starvation.
+	agent := setup.StartAgent(t, env.GRPCAddr, setup.WithCapacity(6))
+	setup.WaitForAgentRegistered(t, env.Store, agent)
+
+	// Trigger all rounds up front so they compete for the concurrency groups.
+	created := make([]*model.Pipeline, 0, rounds)
+	for i := range rounds {
+		p, err := pipeline.Create(t.Context(), env.Store, env.Fixtures.Repo, &model.Pipeline{
+			Event:  model.EventPush,
+			Branch: "main",
+			Commit: fmt.Sprintf("deadbeef%d", i),
+			Ref:    "refs/heads/main",
+			Author: env.Fixtures.Owner.Login,
+			Sender: env.Fixtures.Owner.Login,
+		})
+		require.NoErrorf(t, err, "create pipeline round %d", i)
+		require.NotNil(t, p)
+		created = append(created, p)
+	}
+
+	// Wait for every round to finish, collecting workflow timings by name.
+	byWorkflow := map[string][]wfInterval{}
+
+	for i, p := range created {
+		finished := setup.WaitForPipeline(t, env.Store, p.ID)
+		assert.Equalf(t, model.StatusSuccess, finished.Status, "round %d pipeline status", i)
+
+		workflows, err := env.Store.WorkflowGetTree(finished)
+		require.NoErrorf(t, err, "workflow tree round %d", i)
+
+		for _, wf := range workflows {
+			require.NotZerof(t, wf.Started, "round %d workflow %q has no start time", i, wf.Name)
+			require.NotZerof(t, wf.Finished, "round %d workflow %q has no finish time", i, wf.Name)
+			byWorkflow[wf.Name] = append(byWorkflow[wf.Name], wfInterval{wf.Started, wf.Finished})
+		}
+	}
+
+	require.Len(t, byWorkflow["a"], rounds, "expected one A workflow per round")
+	require.Len(t, byWorkflow["b"], rounds, "expected one B workflow per round")
+
+	maxConcurrentA := maxConcurrent(byWorkflow["a"])
+	maxConcurrentB := maxConcurrent(byWorkflow["b"])
+
+	assert.LessOrEqualf(t, maxConcurrentA, 1,
+		"workflow A has concurrency limit 1 but %d ran at once", maxConcurrentA)
+	assert.LessOrEqualf(t, maxConcurrentB, 2,
+		"workflow B has concurrency limit 2 but %d ran at once", maxConcurrentB)
+	assert.GreaterOrEqualf(t, maxConcurrentB, 2,
+		"workflow B has concurrency limit 2 but never ran more than %d at once — limit not exercised", maxConcurrentB)
+}
+
+// wfInterval is a workflow's [start, finish) wall-clock window in unix seconds.
+type wfInterval struct{ start, finish int64 }
+
+// maxConcurrent returns the largest number of intervals that overlap at any
+// instant. Intervals are treated as half-open [start, finish): a workflow that
+// finishes at the same second another starts is not counted as overlapping, so
+// back-to-back serialized runs report a max concurrency of 1.
+func maxConcurrent(intervals []wfInterval) int {
+	type event struct {
+		t     int64
+		delta int
+	}
+	events := make([]event, 0, len(intervals)*2)
+	for _, iv := range intervals {
+		events = append(events, event{iv.start, +1}, event{iv.finish, -1})
+	}
+	// Insertion sort by time; at an equal timestamp process finishes (-1)
+	// before starts (+1) so a touching boundary does not count as overlap.
+	for i := 1; i < len(events); i++ {
+		for j := i; j > 0; j-- {
+			a, b := events[j-1], events[j]
+			if a.t < b.t || (a.t == b.t && a.delta <= b.delta) {
+				break
+			}
+			events[j-1], events[j] = b, a
+		}
+	}
+
+	cur, best := 0, 0
+	for _, e := range events {
+		cur += e.delta
+		if cur > best {
+			best = cur
+		}
+	}
+	return best
+}

e2e/setup/agent.go

@@ -66,6 +66,10 @@ type agentConfig struct {
 
 	// orgID pins the agent to a specific organization. See WithOrgID.
 	orgID int64
+
+	// capacity is the number of workflows the agent runs in parallel
+	// (advertised to the server and number of runner goroutines).
+	capacity int
 }
 
 // WithHostname sets the agent's hostname label (default: "test-agent").
@@ -98,6 +102,13 @@ func WithOrgID(id int64) AgentOption {
 	return func(c *agentConfig) { c.orgID = id }
 }
 
+// WithCapacity sets how many workflows the agent runs in parallel. This is the
+// value advertised to the server (RegisterAgent capacity) and the number of
+// runner goroutines started. Defaults to AgentMaxWorkflows when unset or <= 0.
+func WithCapacity(n int) AgentOption {
+	return func(c *agentConfig) { c.capacity = n }
+}
+
 // StartAgent connects an in-process agent using the dummy backend to the gRPC
 // server at grpcAddr and returns an *AgentEnv whose AgentID is populated once
 // the agent has registered. Pass AgentOption values to configure labels, hostname,
@@ -114,10 +125,14 @@ func StartAgent(t *testing.T, grpcAddr string, opts ...AgentOption) *AgentEnv {
 		hostname:     "test-agent",
 		customLabels: make(map[string]string),
 		orgID:        model.IDNotSet, // global by default
+		capacity:     AgentMaxWorkflows,
 	}
 	for _, o := range opts {
 		o(cfg)
 	}
+	if cfg.capacity <= 0 {
+		cfg.capacity = AgentMaxWorkflows
+	}
 
 	env := &AgentEnv{name: cfg.hostname}
 
@@ -155,7 +170,7 @@ func StartAgent(t *testing.T, grpcAddr string, opts ...AgentOption) *AgentEnv {
 		Version:      version.String(),
 		Backend:      backend.Name(),
 		Platform:     engInfo.Platform,
-		Capacity:     AgentMaxWorkflows,
+		Capacity:     cfg.capacity,
 		CustomLabels: cfg.customLabels,
 	})
 	require.NoErrorf(t, err, "StartAgent(%s): register with server: %v", cfg.hostname, err)
@@ -192,11 +207,11 @@ func StartAgent(t *testing.T, grpcAddr string, opts ...AgentOption) *AgentEnv {
 	}
 
 	counter := &agent.State{
-		Polling:  AgentMaxWorkflows,
+		Polling:  cfg.capacity,
 		Metadata: make(map[string]agent.Info),
 	}
 
-	for i := range AgentMaxWorkflows {
+	for i := range cfg.capacity {
 		go func(slot int) {
 			runner := agent.NewRunner(client, filter, cfg.hostname, counter, backend)
 			log.Debug().Int("slot", slot).Str("hostname", cfg.hostname).Msg("test agent: runner started")

pipeline/frontend/builder/builder.go

@@ -162,10 +162,12 @@ func (b *PipelineBuilder) genItemForWorkflow(workflow *Workflow, axis matrix.Axi
 	}
 
 	item = &Item{
-		Workflow:  workflow,
-		Config:    ir,
-		Labels:    parsed.Labels,
-		DependsOn: parsed.DependsOn,
+		Workflow:         workflow,
+		Config:           ir,
+		Labels:           parsed.Labels,
+		DependsOn:        parsed.DependsOn,
+		ConcurrencyLimit: parsed.Concurrency.Limit,
+		ConcurrencyGroup: parsed.Concurrency.Group,
 		// TODO: remove in next major.
 		RunsOn: parsed.RunsOn, //nolint:staticcheck
 	}

pipeline/frontend/builder/types.go

@@ -22,11 +22,13 @@ import (
 )
 
 type Item struct {
-	Workflow  *Workflow
-	Labels    map[string]string
-	DependsOn constraint.DependsOn
-	RunsOn    []string
-	Config    *backend_types.Config
+	Workflow         *Workflow
+	Labels           map[string]string
+	DependsOn        constraint.DependsOn
+	RunsOn           []string
+	ConcurrencyLimit int
+	ConcurrencyGroup string
+	Config           *backend_types.Config
 }
 
 type Workflow struct {

pipeline/frontend/yaml/linter/schema/.woodpecker/test-concurrency-invalid.yaml

@@ -0,0 +1,8 @@
+steps:
+  deploy:
+    image: alpine
+    commands:
+      - echo deploying
+
+concurrency:
+  group: deploy-production

pipeline/frontend/yaml/linter/schema/.woodpecker/test-concurrency-shorthand.yaml

@@ -0,0 +1,7 @@
+steps:
+  deploy:
+    image: alpine
+    commands:
+      - echo deploying
+
+concurrency: 1

pipeline/frontend/yaml/linter/schema/.woodpecker/test-concurrency.yaml

@@ -0,0 +1,9 @@
+steps:
+  deploy:
+    image: alpine
+    commands:
+      - echo deploying
+
+concurrency:
+  limit: 1
+  group: deploy-production

pipeline/frontend/yaml/linter/schema/schema.json

@@ -42,6 +42,10 @@
       "description": "List of workflow dependencies. Accepts strings or objects with name and optional fields. Read more: https://woodpecker-ci.org/docs/usage/workflows#flow-control",
       "$ref": "#/definitions/depends_on_list"
     },
+    "concurrency": {
+      "description": "Limit how many instances of this workflow may run at the same time. Read more: https://woodpecker-ci.org/docs/usage/workflows#concurrency",
+      "$ref": "#/definitions/concurrency"
+    },
     "runs_on": {
       "type": "array",
       "description": "Deprecated: use `when.status` instead. Read more: https://woodpecker-ci.org/docs/usage/workflows#flow-control",
@@ -104,6 +108,31 @@
         }
       ]
     },
+    "concurrency": {
+      "description": "Limit how many instances of this workflow may run at the same time. Provide an integer for the limit, or an object to also set a custom group.",
+      "oneOf": [
+        {
+          "type": "integer",
+          "minimum": 1
+        },
+        {
+          "type": "object",
+          "additionalProperties": false,
+          "required": ["limit"],
+          "properties": {
+            "limit": {
+              "type": "integer",
+              "minimum": 1,
+              "description": "Maximum number of workflows sharing the same group that run at once."
+            },
+            "group": {
+              "type": "string",
+              "description": "Identifier shared by workflows that are limited against each other. Defaults to the workflow name. Environment variables are substituted, e.g. `deploy-${CI_COMMIT_BRANCH}`."
+            }
+          }
+        }
+      ]
+    },
     "clone": {
       "description": "Configures the clone step. Read more: https://woodpecker-ci.org/docs/usage/workflow-syntax#clone",
       "oneOf": [

pipeline/frontend/yaml/linter/schema/schema_test.go

@@ -122,6 +122,21 @@ func TestSchema(t *testing.T) {
 			testFile: ".woodpecker/test-kubernetes-backend-tolerations.yaml",
 			fail:     false,
 		},
+		{
+			name:     "Concurrency",
+			testFile: ".woodpecker/test-concurrency.yaml",
+			fail:     false,
+		},
+		{
+			name:     "Concurrency shorthand",
+			testFile: ".woodpecker/test-concurrency-shorthand.yaml",
+			fail:     false,
+		},
+		{
+			name:     "Concurrency invalid",
+			testFile: ".woodpecker/test-concurrency-invalid.yaml",
+			fail:     true,
+		},
 	}
 
 	for _, tt := range testTable {

pipeline/frontend/yaml/parse_test.go

@@ -57,6 +57,40 @@ func TestParse(t *testing.T) {
 		assert.Error(t, err)
 	})
 
+	t.Run("Should unmarshal concurrency object", func(t *testing.T) {
+		out, err := ParseString(`steps:
+  deploy:
+    image: alpine
+concurrency:
+  limit: 2
+  group: deploy
+`)
+		assert.NoError(t, err)
+		assert.Equal(t, 2, out.Concurrency.Limit)
+		assert.Equal(t, "deploy", out.Concurrency.Group)
+	})
+
+	t.Run("Should unmarshal concurrency shorthand", func(t *testing.T) {
+		out, err := ParseString(`steps:
+  deploy:
+    image: alpine
+concurrency: 1
+`)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, out.Concurrency.Limit)
+		assert.Empty(t, out.Concurrency.Group)
+	})
+
+	t.Run("Should default concurrency to disabled", func(t *testing.T) {
+		out, err := ParseString(`steps:
+  deploy:
+    image: alpine
+`)
+		assert.NoError(t, err)
+		assert.True(t, out.Concurrency.IsZero())
+		assert.Equal(t, 0, out.Concurrency.Limit)
+	})
+
 	t.Run("Should handle simple yaml anchors", func(t *testing.T) {
 		out, err := ParseString(simpleYamlAnchors)
 		assert.NoError(t, err)
@@ -210,6 +244,7 @@ labels:
 depends_on:
   - lint
   - test
+concurrency: 1
 `
 
 var simpleYamlAnchors = `
@@ -248,6 +283,9 @@ steps:
     environment:
       DRIVER: next
       PLATFORM: linux
+concurrency:
+  limit: 1
+  group: test
 `
 
 var sampleDeepYaml = `
@@ -305,6 +343,9 @@ func TestReSerialize(t *testing.T) {
       environment:
         DRIVER: next
         PLATFORM: linux
+concurrency:
+    limit: 1
+    group: test
 `, string(work1Bin))
 
 	work2, err := ParseString(sampleYaml)
@@ -357,6 +398,7 @@ labels:
 depends_on:
     - lint
     - test
+concurrency: 1
 `, string(workBin2))
 }
 

pipeline/frontend/yaml/types/concurrency.go

@@ -0,0 +1,93 @@
+// Copyright 2026 Woodpecker Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package types
+
+import (
+	"fmt"
+
+	"go.yaml.in/yaml/v4"
+)
+
+// Concurrency limits how many instances of a workflow may run at the same
+// time. It can be unmarshaled from:
+//   - an integer: `concurrency: 1` (limit only, default per-workflow group)
+//   - an object: `concurrency: {limit: 1, group: deploy}`
+type Concurrency struct {
+	// Limit is the maximum number of workflows sharing the same group that
+	// are allowed to run at the same time. A value <= 0 disables the limit.
+	Limit int `yaml:"limit,omitempty"`
+	// Group identifies which workflows are mutually limited. Workflows that
+	// resolve to the same group within a repository are serialized according
+	// to the limit. When empty the limit applies per workflow, so different
+	// runs of the same workflow are limited against each other.
+	Group string `yaml:"group,omitempty"`
+}
+
+// UnmarshalYAML implements the Unmarshaler interface. It inspects the YAML
+// node kind to decide how to decode, instead of speculatively decoding into an
+// int and falling back on error:
+//   - a scalar (`concurrency: 1`) sets only the limit
+//   - a mapping (`concurrency: {limit: 1, group: deploy}`) sets both fields
+func (c *Concurrency) UnmarshalYAML(value *yaml.Node) error {
+	if value.Kind == yaml.DocumentNode && len(value.Content) == 1 {
+		value = value.Content[0]
+	}
+	// resolve anchors/aliases so the kind switch sees the referenced node.
+	if value.Kind == yaml.AliasNode {
+		value = value.Alias
+	}
+
+	switch value.Kind {
+	// shorthand: `concurrency: <int>`
+	case yaml.ScalarNode:
+		var limit int
+		if err := value.Decode(&limit); err != nil {
+			return fmt.Errorf("failed to unmarshal concurrency limit: %w", err)
+		}
+		c.Limit = limit
+		return nil
+
+	// full form: `concurrency: {limit: <int>, group: <string>}`
+	case yaml.MappingNode:
+		// alias type avoids recursing into this UnmarshalYAML.
+		type concurrencyAlias Concurrency
+		var tmp concurrencyAlias
+		if err := value.Decode(&tmp); err != nil {
+			return fmt.Errorf("failed to unmarshal concurrency: %w", err)
+		}
+		*c = Concurrency(tmp)
+		return nil
+
+	default:
+		return fmt.Errorf("failed to unmarshal concurrency: expected an integer or a mapping, got %v", value.Kind)
+	}
+}
+
+// MarshalYAML implements the Marshaler interface. It mirrors UnmarshalYAML so
+// the config round-trips: when only a limit is set (no explicit group) it emits
+// the shorthand `concurrency: <int>`, otherwise the full `{limit, group}` form.
+func (c Concurrency) MarshalYAML() (any, error) {
+	if c.Group == "" {
+		return c.Limit, nil
+	}
+	// alias type avoids recursing into this MarshalYAML.
+	type concurrencyAlias Concurrency
+	return concurrencyAlias(c), nil
+}
+
+// IsZero treats a disabled (limit <= 0) concurrency as empty for omitempty.
+func (c Concurrency) IsZero() bool {
+	return c.Limit <= 0 && c.Group == ""
+}

pipeline/frontend/yaml/types/concurrency_test.go

@@ -0,0 +1,174 @@
+// Copyright 2026 Woodpecker Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package types
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.yaml.in/yaml/v4"
+)
+
+func TestUnmarshalConcurrency(t *testing.T) {
+	tests := []struct {
+		name     string
+		yaml     string
+		expected Concurrency
+	}{
+		{
+			name:     "shorthand integer",
+			yaml:     `concurrency: 3`,
+			expected: Concurrency{Limit: 3},
+		},
+		{
+			name:     "full form with limit and group",
+			yaml:     "concurrency:\n  limit: 2\n  group: deploy",
+			expected: Concurrency{Limit: 2, Group: "deploy"},
+		},
+		{
+			name:     "full form with limit only",
+			yaml:     "concurrency:\n  limit: 1",
+			expected: Concurrency{Limit: 1},
+		},
+		{
+			name:     "full form with group only",
+			yaml:     "concurrency:\n  group: deploy",
+			expected: Concurrency{Group: "deploy"},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			var parsed struct {
+				Concurrency Concurrency `yaml:"concurrency"`
+			}
+			err := yaml.Unmarshal([]byte(tc.yaml), &parsed)
+			assert.NoError(t, err)
+			assert.Equal(t, tc.expected, parsed.Concurrency)
+		})
+	}
+}
+
+func TestUnmarshalConcurrencyError(t *testing.T) {
+	var parsed struct {
+		Concurrency Concurrency `yaml:"concurrency"`
+	}
+	// a sequence is neither a valid shorthand int nor a valid object.
+	err := yaml.Unmarshal([]byte("concurrency:\n  - invalid"), &parsed)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "failed to unmarshal concurrency")
+}
+
+func TestConcurrencyIsZero(t *testing.T) {
+	tests := []struct {
+		name        string
+		concurrency Concurrency
+		expected    bool
+	}{
+		{name: "empty", concurrency: Concurrency{}, expected: true},
+		{name: "disabled limit", concurrency: Concurrency{Limit: 0}, expected: true},
+		{name: "negative limit", concurrency: Concurrency{Limit: -1}, expected: true},
+		{name: "with limit", concurrency: Concurrency{Limit: 1}, expected: false},
+		{name: "with group only", concurrency: Concurrency{Group: "deploy"}, expected: false},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.expected, tc.concurrency.IsZero())
+		})
+	}
+}
+
+func TestMarshalConcurrencyOmitsZero(t *testing.T) {
+	parsed := struct {
+		Concurrency Concurrency `yaml:"concurrency,omitempty"`
+	}{}
+	out, err := yaml.Marshal(parsed)
+	assert.NoError(t, err)
+	// IsZero makes an unset concurrency omitted from the output.
+	assert.NotContains(t, string(out), "concurrency")
+}
+
+func TestMarshalConcurrency(t *testing.T) {
+	tests := []struct {
+		name        string
+		concurrency Concurrency
+		expected    string
+	}{
+		{
+			name:        "shorthand integer when no group",
+			concurrency: Concurrency{Limit: 3},
+			expected:    "concurrency: 3\n",
+		},
+		{
+			name:        "full form when group is set",
+			concurrency: Concurrency{Limit: 2, Group: "deploy"},
+			expected:    "concurrency:\n    limit: 2\n    group: deploy\n",
+		},
+		{
+			name:        "group only omits the zero limit",
+			concurrency: Concurrency{Group: "deploy"},
+			expected:    "concurrency:\n    group: deploy\n",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			wrapped := struct {
+				Concurrency Concurrency `yaml:"concurrency"`
+			}{tc.concurrency}
+			out, err := yaml.Marshal(wrapped)
+			assert.NoError(t, err)
+			assert.Equal(t, tc.expected, string(out))
+		})
+	}
+}
+
+func TestConcurrencyRoundTrip(t *testing.T) {
+	tests := []struct {
+		name        string
+		concurrency Concurrency
+		// wantShorthand asserts whether the marshaled form is the bare
+		// integer (true) or the expanded object (false).
+		wantShorthand bool
+	}{
+		{name: "limit only", concurrency: Concurrency{Limit: 3}, wantShorthand: true},
+		{name: "limit and group", concurrency: Concurrency{Limit: 2, Group: "deploy"}, wantShorthand: false},
+		{name: "group only", concurrency: Concurrency{Group: "deploy"}, wantShorthand: false},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			type wrapper struct {
+				Concurrency Concurrency `yaml:"concurrency"`
+			}
+
+			out, err := yaml.Marshal(wrapper{tc.concurrency})
+			assert.NoError(t, err)
+
+			if tc.wantShorthand {
+				assert.NotContains(t, string(out), "limit:", "expected shorthand, got object form")
+				assert.NotContains(t, string(out), "group:")
+			} else {
+				assert.Contains(t, string(out), "group:")
+			}
+
+			var back wrapper
+			err = yaml.Unmarshal(out, &back)
+			assert.NoError(t, err)
+			assert.Equal(t, tc.concurrency, back.Concurrency, "value changed across marshal/unmarshal")
+		})
+	}
+}

pipeline/frontend/yaml/types/workflow.go

@@ -21,14 +21,15 @@ import (
 type (
 	// Workflow defines a workflow configuration.
 	Workflow struct {
-		When      constraint.When      `yaml:"when,omitempty"`
-		Workspace Workspace            `yaml:"workspace,omitempty"`
-		Clone     ContainerList        `yaml:"clone,omitempty"`
-		Steps     ContainerList        `yaml:"steps,omitempty"`
-		Services  ContainerList        `yaml:"services,omitempty"`
-		Labels    map[string]string    `yaml:"labels,omitempty"`
-		DependsOn constraint.DependsOn `yaml:"depends_on,omitempty"`
-		SkipClone bool                 `yaml:"skip_clone,omitempty"`
+		When        constraint.When      `yaml:"when,omitempty"`
+		Workspace   Workspace            `yaml:"workspace,omitempty"`
+		Clone       ContainerList        `yaml:"clone,omitempty"`
+		Steps       ContainerList        `yaml:"steps,omitempty"`
+		Services    ContainerList        `yaml:"services,omitempty"`
+		Labels      map[string]string    `yaml:"labels,omitempty"`
+		DependsOn   constraint.DependsOn `yaml:"depends_on,omitempty"`
+		Concurrency Concurrency          `yaml:"concurrency,omitempty"`
+		SkipClone   bool                 `yaml:"skip_clone,omitempty"`
 		// Deprecated: use when.status. TODO remove in next major.
 		RunsOn []string `yaml:"runs_on,omitempty"`
 	}

server/model/task.go

@@ -34,6 +34,15 @@ type Task struct {
 	AgentID      int64                  `json:"agent_id"     xorm:"'agent_id'"`
 	PipelineID   int64                  `json:"pipeline_id"  xorm:"'pipeline_id'"`
 	RepoID       int64                  `json:"repo_id"      xorm:"'repo_id'"`
+	// ConcurrencyLimit is the maximum number of tasks sharing the same
+	// ConcurrencyGroup that may run at once. A value <= 0 means unlimited.
+	ConcurrencyLimit int `json:"concurrency_limit" xorm:"NOT NULL DEFAULT 0 'concurrency_limit'"`
+	// ConcurrencyGroup identifies tasks that are limited against each other.
+	// It is empty when no concurrency limit applies.
+	ConcurrencyGroup string `json:"concurrency_group" xorm:"'concurrency_group'"`
+	// Created is the unix timestamp the task's pipeline was created at. It
+	// defines the queue ordering across pipelines.
+	Created int64 `json:"created" xorm:"NOT NULL DEFAULT 0 'created'"`
 } //	@name	Task
 
 // TableName return database table name for xorm.

server/pipeline/queue.go

@@ -19,6 +19,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"maps"
+	"time"
 
 	"go.woodpecker-ci.org/woodpecker/v3/pipeline/frontend/builder"
 	"go.woodpecker-ci.org/woodpecker/v3/rpc"
@@ -36,6 +37,12 @@ func queuePipeline(ctx context.Context, repo *model.Repo, activePipeline *model.
 			Labels:     make(map[string]string),
 			PipelineID: activePipeline.ID,
 			RepoID:     repo.ID,
+			Created:    activePipeline.Created,
+		}
+		// fall back to the current time if the pipeline has no creation
+		// timestamp, so the queue always has a defined ordering key.
+		if task.Created == 0 {
+			task.Created = time.Now().Unix()
 		}
 		maps.Copy(task.Labels, item.Labels)
 		err := task.ApplyLabelsFromRepo(repo)
@@ -46,6 +53,19 @@ func queuePipeline(ctx context.Context, repo *model.Repo, activePipeline *model.
 		task.RunOn = item.RunsOn
 		task.DepStatus = make(map[string]model.StatusValue)
 
+		// Set up the concurrency limit if the workflow opted in.
+		if item.ConcurrencyLimit > 0 {
+			task.ConcurrencyLimit = item.ConcurrencyLimit
+
+			// If no group assigned, each workflow is it's own unique group,
+			// else we use defined group unique per repo.
+			if item.ConcurrencyGroup == "" {
+				task.ConcurrencyGroup = fmt.Sprintf("%d/%s/", repo.ID, item.Workflow.Name)
+			} else {
+				task.ConcurrencyGroup = fmt.Sprintf("%d//%s", repo.ID, item.ConcurrencyGroup)
+			}
+		}
+
 		task.Data, err = json.Marshal(rpc.Workflow{
 			ID:      fmt.Sprint(item.Workflow.ID),
 			Config:  item.Config,

server/pipeline/queue_test.go

@@ -0,0 +1,125 @@
+// Copyright 2026 Woodpecker Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package pipeline
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	"go.woodpecker-ci.org/woodpecker/v3/pipeline/frontend/builder"
+	"go.woodpecker-ci.org/woodpecker/v3/server"
+	"go.woodpecker-ci.org/woodpecker/v3/server/model"
+	"go.woodpecker-ci.org/woodpecker/v3/server/pubsub/memory"
+	queue_mocks "go.woodpecker-ci.org/woodpecker/v3/server/queue/mocks"
+	"go.woodpecker-ci.org/woodpecker/v3/server/scheduler"
+)
+
+func TestQueuePipelineConcurrency(t *testing.T) {
+	repo := &model.Repo{ID: 7}
+	activePipeline := &model.Pipeline{ID: 42}
+
+	tests := []struct {
+		name          string
+		item          *builder.Item
+		expectedLimit int
+		expectedGroup string
+	}{
+		{
+			name: "no limit leaves concurrency unset",
+			item: &builder.Item{
+				Workflow: &builder.Workflow{ID: 1, Name: "build"},
+			},
+			expectedLimit: 0,
+			expectedGroup: "",
+		},
+		{
+			name: "explicit group is scoped to the repo",
+			item: &builder.Item{
+				Workflow:         &builder.Workflow{ID: 2, Name: "build"},
+				ConcurrencyLimit: 2,
+				ConcurrencyGroup: "deploy",
+			},
+			expectedLimit: 2,
+			expectedGroup: "7//deploy",
+		},
+		{
+			name: "empty group defaults to the workflow name",
+			item: &builder.Item{
+				Workflow:         &builder.Workflow{ID: 3, Name: "test"},
+				ConcurrencyLimit: 1,
+			},
+			expectedLimit: 1,
+			expectedGroup: "7/test/",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			var captured []*model.Task
+			mockQueue := queue_mocks.NewMockQueue(t)
+			mockQueue.On("PushAtOnce", mock.Anything, mock.Anything).
+				Run(func(args mock.Arguments) {
+					captured, _ = args.Get(1).([]*model.Task)
+				}).
+				Return(nil).Once()
+			server.Config.Services.Scheduler = scheduler.NewScheduler(mockQueue, memory.New())
+
+			err := queuePipeline(t.Context(), repo, activePipeline, []*builder.Item{tc.item})
+			require.NoError(t, err)
+			require.Len(t, captured, 1)
+
+			task := captured[0]
+			assert.Equal(t, tc.expectedLimit, task.ConcurrencyLimit)
+			assert.Equal(t, tc.expectedGroup, task.ConcurrencyGroup)
+		})
+	}
+}
+
+func TestQueuePipelineCreated(t *testing.T) {
+	repo := &model.Repo{ID: 7}
+	item := &builder.Item{Workflow: &builder.Workflow{ID: 1, Name: "build"}}
+
+	runOnce := func(t *testing.T, pipeline *model.Pipeline) *model.Task {
+		t.Helper()
+		var captured []*model.Task
+		mockQueue := queue_mocks.NewMockQueue(t)
+		mockQueue.On("PushAtOnce", mock.Anything, mock.Anything).
+			Run(func(args mock.Arguments) {
+				captured, _ = args.Get(1).([]*model.Task)
+			}).
+			Return(nil).Once()
+		server.Config.Services.Scheduler = scheduler.NewScheduler(mockQueue, memory.New())
+
+		err := queuePipeline(t.Context(), repo, pipeline, []*builder.Item{item})
+		require.NoError(t, err)
+		require.Len(t, captured, 1)
+		return captured[0]
+	}
+
+	t.Run("inherits the pipeline creation time", func(t *testing.T) {
+		task := runOnce(t, &model.Pipeline{ID: 42, Created: 1700000000})
+		assert.Equal(t, int64(1700000000), task.Created)
+	})
+
+	t.Run("falls back to now when the pipeline has no creation time", func(t *testing.T) {
+		before := time.Now().Unix()
+		task := runOnce(t, &model.Pipeline{ID: 42})
+		assert.GreaterOrEqual(t, task.Created, before)
+	})
+}

server/queue/fifo.go

@@ -319,6 +319,13 @@ func (q *fifo) assignToWorker() (*list.Element, *worker) {
 		task, _ := element.Value.(*model.Task)
 		log.Debug().Msgf("queue: trying to assign task: %v with deps %v", task.ID, task.Dependencies)
 
+		// skip tasks that would exceed their workflow concurrency limit, they
+		// stay pending and are retried on the next process tick.
+		if !q.canRunConcurrent(task) {
+			log.Debug().Msgf("queue: task %v deferred due to concurrency group %q", task.ID, task.ConcurrencyGroup)
+			continue
+		}
+
 		for worker := range q.workers {
 			matched, score := worker.filter(task)
 			if matched && score > bestScore {
@@ -335,6 +342,82 @@ func (q *fifo) assignToWorker() (*list.Element, *worker) {
 	return nil, nil
 }
 
+// canRunConcurrent reports whether the given task may currently start without
+// violating its workflow concurrency limit. Tasks without a limit always pass,
+// keeping the default scheduling behavior unchanged.
+//
+// Slots within a concurrency group are granted in creation order (earliest
+// pipeline first, by the task's Created timestamp, with the workflow name as a
+// deterministic tiebreaker) rather than in the order tasks become ready. This
+// guarantees that a later pipeline whose dependencies happen to finish faster
+// cannot overtake an earlier one that is still waiting.
+//
+// The ordering reservation only applies across pipelines. Within a single
+// pipeline, execution order is already defined by depends_on, and reserving a
+// slot for an earlier workflow could deadlock when it depends on a later
+// workflow that shares the same group (e.g. deploy.yaml depending on
+// test.yaml). Because dependencies never cross pipelines, restricting the
+// reservation to other pipelines keeps the ordering guarantee while making
+// such deadlocks impossible.
+//
+// Expects the queue to be locked by the caller.
+func (q *fifo) canRunConcurrent(task *model.Task) bool {
+	if task.ConcurrencyLimit <= 0 || task.ConcurrencyGroup == "" {
+		return true
+	}
+
+	group := task.ConcurrencyGroup
+
+	// count tasks of the same group that already occupy a running slot.
+	running := 0
+	for _, e := range q.running {
+		if e.item.ConcurrencyGroup == group {
+			running++
+		}
+	}
+	if running >= task.ConcurrencyLimit {
+		return false
+	}
+
+	// count not-yet-running members of the group from earlier pipelines. They
+	// have priority for the remaining slots, even if they are still waiting on
+	// their dependencies, so cross-pipeline ordering is preserved.
+	ahead := 0
+	countAhead := func(other *model.Task) {
+		if other.ConcurrencyGroup != group || other.ID == task.ID {
+			return
+		}
+		// only reserve order across pipelines; see the function doc above.
+		if other.PipelineID == task.PipelineID {
+			return
+		}
+		if taskOrderLess(other, task) {
+			ahead++
+		}
+	}
+	for element := q.pending.Front(); element != nil; element = element.Next() {
+		other, _ := element.Value.(*model.Task)
+		countAhead(other)
+	}
+	for element := q.waitingOnDeps.Front(); element != nil; element = element.Next() {
+		other, _ := element.Value.(*model.Task)
+		countAhead(other)
+	}
+
+	return running+ahead < task.ConcurrencyLimit
+}
+
+// taskOrderLess reports whether task a was instantiated before task b. Ordering
+// is by the Created timestamp (the pipeline creation time), with the workflow
+// name as a deterministic tiebreaker for tasks created within the same second.
+// The task ID is intentionally not used for ordering.
+func taskOrderLess(a, b *model.Task) bool {
+	if a.Created != b.Created {
+		return a.Created < b.Created
+	}
+	return a.Name < b.Name
+}
+
 func (q *fifo) resubmitExpiredPipelines() {
 	for taskID, taskState := range q.running {
 		if time.Now().After(taskState.deadline) {

server/queue/fifo_test.go

@@ -671,6 +671,228 @@ func TestFifoDependencies(t *testing.T) {
 	})
 }
 
+func TestFifoConcurrency(t *testing.T) {
+	ctx, cancel, q := setupTestQueue(t)
+	defer cancel(nil)
+
+	t.Run("limit serializes group in instantiation order", func(t *testing.T) {
+		// Lower Created == instantiated earlier. taskB is pushed first to
+		// prove the queue serializes by creation order, not by push/ready
+		// order. Distinct pipeline IDs model two pipelines of the same
+		// workflow. Ordering must not depend on the task ID, so taskA (the
+		// earlier one) deliberately has the higher ID.
+		taskA := &model.Task{ID: "200", PipelineID: 1, Created: 100, ConcurrencyGroup: "repo:deploy", ConcurrencyLimit: 1}
+		taskB := &model.Task{ID: "100", PipelineID: 2, Created: 200, ConcurrencyGroup: "repo:deploy", ConcurrencyLimit: 1}
+
+		assert.NoError(t, q.PushAtOnce(ctx, []*model.Task{taskB, taskA}))
+		waitForProcess()
+
+		got, err := q.Poll(ctx, 1, filterFnTrue)
+		assert.NoError(t, err)
+		assert.Equal(t, "200", got.ID) // earliest instantiated (lowest Created) runs first
+
+		waitForProcess()
+		info := q.Info(ctx)
+		assert.Len(t, info.Running, 1)
+		assert.Len(t, info.Pending, 1) // taskB deferred by concurrency limit
+
+		// taskB cannot be polled while taskA holds the only slot.
+		pollCtx, pollCancel := context.WithTimeout(ctx, 200*time.Millisecond)
+		_, err = q.Poll(pollCtx, 2, filterFnTrue)
+		pollCancel()
+		assert.Error(t, err)
+
+		// finishing taskA frees the slot for taskB.
+		assert.NoError(t, q.Done(ctx, got.ID, model.StatusSuccess))
+		waitForProcess()
+
+		got2, err := q.Poll(ctx, 2, filterFnTrue)
+		assert.NoError(t, err)
+		assert.Equal(t, "100", got2.ID)
+		assert.NoError(t, q.Done(ctx, got2.ID, model.StatusSuccess))
+		waitForProcess()
+	})
+
+	t.Run("preserves instantiation order over readiness", func(t *testing.T) {
+		// Pipeline 1's deploy (taskA) waits on its own slow check (dep), while
+		// pipeline 2's deploy (taskB) is immediately ready. taskB must still
+		// wait for the earlier taskA. Ordering is by Created, not ID, so taskA
+		// has the higher ID but the lower Created.
+		dep := &model.Task{ID: "50", PipelineID: 1, Created: 100}
+		taskA := &model.Task{
+			ID:               "200",
+			PipelineID:       1,
+			Created:          100,
+			ConcurrencyGroup: "repo:deploy2",
+			ConcurrencyLimit: 1,
+			Dependencies:     []string{"50"},
+			DepStatus:        make(map[string]model.StatusValue),
+			RunOn:            []string{"success", "failure"},
+		}
+		taskB := &model.Task{ID: "100", PipelineID: 2, Created: 200, ConcurrencyGroup: "repo:deploy2", ConcurrencyLimit: 1}
+
+		assert.NoError(t, q.PushAtOnce(ctx, []*model.Task{dep, taskA, taskB}))
+		waitForProcess()
+
+		info := q.Info(ctx)
+		assert.Equal(t, 1, info.Stats.WaitingOnDeps) // taskA waiting on dep
+
+		// only the dependency is runnable; taskB is held back behind taskA.
+		got, err := q.Poll(ctx, 1, filterFnTrue)
+		assert.NoError(t, err)
+		assert.Equal(t, "50", got.ID)
+
+		pollCtx, pollCancel := context.WithTimeout(ctx, 200*time.Millisecond)
+		_, err = q.Poll(pollCtx, 2, filterFnTrue)
+		pollCancel()
+		assert.Error(t, err, "taskB must not overtake the earlier taskA")
+
+		assert.NoError(t, q.Done(ctx, got.ID, model.StatusSuccess))
+		waitForProcess()
+
+		gotA, err := q.Poll(ctx, 1, filterFnTrue)
+		assert.NoError(t, err)
+		assert.Equal(t, "200", gotA.ID)
+		assert.NoError(t, q.Done(ctx, gotA.ID, model.StatusSuccess))
+		waitForProcess()
+
+		gotB, err := q.Poll(ctx, 2, filterFnTrue)
+		assert.NoError(t, err)
+		assert.Equal(t, "100", gotB.ID)
+		assert.NoError(t, q.Done(ctx, gotB.ID, model.StatusSuccess))
+		waitForProcess()
+	})
+
+	t.Run("limit greater than one allows parallelism", func(t *testing.T) {
+		t1 := &model.Task{ID: "300", PipelineID: 1, ConcurrencyGroup: "repo:build", ConcurrencyLimit: 2}
+		t2 := &model.Task{ID: "400", PipelineID: 2, ConcurrencyGroup: "repo:build", ConcurrencyLimit: 2}
+		t3 := &model.Task{ID: "500", PipelineID: 3, ConcurrencyGroup: "repo:build", ConcurrencyLimit: 2}
+
+		assert.NoError(t, q.PushAtOnce(ctx, []*model.Task{t1, t2, t3}))
+		waitForProcess()
+
+		g1, err := q.Poll(ctx, 1, filterFnTrue)
+		assert.NoError(t, err)
+		g2, err := q.Poll(ctx, 2, filterFnTrue)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []string{"300", "400"}, []string{g1.ID, g2.ID})
+
+		waitForProcess()
+		info := q.Info(ctx)
+		assert.Len(t, info.Running, 2)
+		assert.Len(t, info.Pending, 1) // third deferred until a slot frees
+
+		pollCtx, pollCancel := context.WithTimeout(ctx, 200*time.Millisecond)
+		_, err = q.Poll(pollCtx, 3, filterFnTrue)
+		pollCancel()
+		assert.Error(t, err)
+
+		assert.NoError(t, q.Done(ctx, g1.ID, model.StatusSuccess))
+		waitForProcess()
+		g3, err := q.Poll(ctx, 3, filterFnTrue)
+		assert.NoError(t, err)
+		assert.Equal(t, "500", g3.ID)
+		assert.NoError(t, q.Done(ctx, g2.ID, model.StatusSuccess))
+		assert.NoError(t, q.Done(ctx, g3.ID, model.StatusSuccess))
+		waitForProcess()
+	})
+
+	t.Run("same pipeline dependency does not deadlock", func(t *testing.T) {
+		// Within one pipeline, deploy.yaml (lower ID, alphabetically first)
+		// depends on test.yaml (higher ID), and both share a concurrency group.
+		// The ordering reservation must not treat the dependent deploy as
+		// "ahead" of its own dependency, otherwise neither can ever run.
+		deploy := &model.Task{
+			ID:               "100",
+			PipelineID:       1,
+			ConcurrencyGroup: "repo:ci",
+			ConcurrencyLimit: 1,
+			Dependencies:     []string{"200"},
+			DepStatus:        make(map[string]model.StatusValue),
+			RunOn:            []string{"success", "failure"},
+		}
+		test := &model.Task{
+			ID:               "200",
+			PipelineID:       1,
+			ConcurrencyGroup: "repo:ci",
+			ConcurrencyLimit: 1,
+		}
+
+		assert.NoError(t, q.PushAtOnce(ctx, []*model.Task{deploy, test}))
+		waitForProcess()
+
+		// test must be runnable even though deploy has a lower ID.
+		got, err := q.Poll(ctx, 1, filterFnTrue)
+		assert.NoError(t, err)
+		assert.Equal(t, "200", got.ID, "the dependency must not be starved by the dependent")
+		assert.NoError(t, q.Done(ctx, got.ID, model.StatusSuccess))
+		waitForProcess()
+
+		gotDeploy, err := q.Poll(ctx, 1, filterFnTrue)
+		assert.NoError(t, err)
+		assert.Equal(t, "100", gotDeploy.ID)
+		assert.NoError(t, q.Done(ctx, gotDeploy.ID, model.StatusSuccess))
+		waitForProcess()
+	})
+
+	t.Run("different groups do not block each other", func(t *testing.T) {
+		t1 := &model.Task{ID: "600", ConcurrencyGroup: "repo:a", ConcurrencyLimit: 1}
+		t2 := &model.Task{ID: "700", ConcurrencyGroup: "repo:b", ConcurrencyLimit: 1}
+
+		assert.NoError(t, q.PushAtOnce(ctx, []*model.Task{t1, t2}))
+		waitForProcess()
+
+		g1, err := q.Poll(ctx, 1, filterFnTrue)
+		assert.NoError(t, err)
+		g2, err := q.Poll(ctx, 2, filterFnTrue)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []string{"600", "700"}, []string{g1.ID, g2.ID})
+
+		assert.NoError(t, q.Done(ctx, g1.ID, model.StatusSuccess))
+		assert.NoError(t, q.Done(ctx, g2.ID, model.StatusSuccess))
+		waitForProcess()
+	})
+
+	t.Run("no limit keeps default behavior", func(t *testing.T) {
+		t1 := &model.Task{ID: "800"}
+		t2 := &model.Task{ID: "900"}
+
+		assert.NoError(t, q.PushAtOnce(ctx, []*model.Task{t1, t2}))
+		waitForProcess()
+
+		g1, err := q.Poll(ctx, 1, filterFnTrue)
+		assert.NoError(t, err)
+		g2, err := q.Poll(ctx, 2, filterFnTrue)
+		assert.NoError(t, err)
+		assert.ElementsMatch(t, []string{"800", "900"}, []string{g1.ID, g2.ID})
+
+		assert.NoError(t, q.Done(ctx, g1.ID, model.StatusSuccess))
+		assert.NoError(t, q.Done(ctx, g2.ID, model.StatusSuccess))
+		waitForProcess()
+	})
+
+	t.Run("task ordering uses Created with name tiebreak", func(t *testing.T) {
+		// earlier Created sorts first, regardless of ID.
+		assert.True(t, taskOrderLess(
+			&model.Task{ID: "999", Created: 100},
+			&model.Task{ID: "1", Created: 200},
+		))
+		assert.False(t, taskOrderLess(
+			&model.Task{ID: "1", Created: 200},
+			&model.Task{ID: "999", Created: 100},
+		))
+		// equal Created falls back to the workflow name, alphabetically.
+		assert.True(t, taskOrderLess(
+			&model.Task{ID: "2", Created: 100, Name: "alpha"},
+			&model.Task{ID: "1", Created: 100, Name: "beta"},
+		))
+		assert.False(t, taskOrderLess(
+			&model.Task{ID: "1", Created: 100, Name: "beta"},
+			&model.Task{ID: "2", Created: 100, Name: "alpha"},
+		))
+	})
+}
+
 func TestFifoLeaseManagement(t *testing.T) {
 	ctx, cancel, q := setupTestQueue(t)
 	defer cancel(nil)