fix(actions): fail unexpandable reusable workflow callers and decouple the job emitter's cross-run processing (#38565)

## Changes

### 1. Handle reusable workflow expansion failures

If a reusable workflow caller job is invalid (e.g. uses a workflow with
syntax error), the job emitter should mark it as failed instead of
retrying.

Related:
https://github.com/go-gitea/gitea/pull/38518#discussion_r3608882095

### 2. No longer process concurrent run inline

**Before**: If a run(R1)'s status change unlocks another blocked run(R2)
via concurrency group, the job emitter will process R2 in R1's
transaction.

**Current**: No longer process R2 inline and emit the ID of R2 to let
another pass process it.

models/actions/run_job.go

@@ -333,6 +333,14 @@ func GetDirectChildJobsByParent(ctx context.Context, parentJob *ActionRunJob) (A
 	return jobs, nil
 }
 
+// DeleteDirectChildJobsByParent deletes the direct child jobs of a parent job.
+func DeleteDirectChildJobsByParent(ctx context.Context, parentJob *ActionRunJob) error {
+	_, err := db.GetEngine(ctx).
+		Where("run_id=? AND parent_job_id=?", parentJob.RunID, parentJob.ID).
+		Delete(new(ActionRunJob))
+	return err
+}
+
 // CollectAllDescendantJobs returns every job in `allJobs` that lives under parent's subtree (recursively), excluding `parent` itself
 func CollectAllDescendantJobs(parent *ActionRunJob, allJobs []*ActionRunJob) []*ActionRunJob {
 	parents := map[int64]bool{parent.ID: true}

services/actions/job_emitter.go

@@ -15,6 +15,7 @@ import (
 	"gitea.dev/modules/log"
 	"gitea.dev/modules/queue"
 	"gitea.dev/modules/setting"
+	"gitea.dev/modules/timeutil"
 	"gitea.dev/modules/util"
 
 	"xorm.io/builder"
@@ -96,7 +97,7 @@ func checkJobsByRunID(ctx context.Context, runID int64) error {
 			continue
 		}
 		if err := EmitJobsIfReadyByRun(rid); err != nil {
-			log.Error("re-emit run %d after caller expansion: %v", rid, err)
+			log.Error("re-emit run %d: %v", rid, err)
 		}
 	}
 	NotifyWorkflowJobsAndRunsStatusUpdate(ctx, result.CancelledJobs)
@@ -147,57 +148,66 @@ func createCommitStatusesForJobsByRun(ctx context.Context, jobs []*actions_model
 	return nil
 }
 
-// findBlockedRunIDByConcurrency finds a blocked concurrent run in a repo and returns 0 when there is no blocked run.
-func findBlockedRunIDByConcurrency(ctx context.Context, repoID int64, concurrencyGroup string) (int64, error) {
+// findConcurrencyWaiterToWake returns a run (other than excludeRunID) blocked on the group that can be woken now
+func findConcurrencyWaiterToWake(ctx context.Context, repoID, excludeRunID int64, concurrencyGroup string) (int64, error) {
 	if concurrencyGroup == "" {
 		return 0, nil
 	}
+
+	// The slot should be free before any waiter can proceed.
+	holderAttempts, holderJobs, err := actions_model.GetConcurrentRunAttemptsAndJobs(ctx, repoID, concurrencyGroup, []actions_model.Status{actions_model.StatusRunning, actions_model.StatusCancelling})
+	if err != nil {
+		return 0, fmt.Errorf("find concurrency-group holders: %w", err)
+	}
+	if len(holderAttempts) > 0 || len(holderJobs) > 0 {
+		return 0, nil
+	}
+
 	cAttempts, cJobs, err := actions_model.GetConcurrentRunAttemptsAndJobs(ctx, repoID, concurrencyGroup, []actions_model.Status{actions_model.StatusBlocked})
 	if err != nil {
-		return 0, fmt.Errorf("find concurrent runs and jobs: %w", err)
+		return 0, fmt.Errorf("find blocked concurrent runs: %w", err)
 	}
-
-	if len(cAttempts) > 0 {
-		return cAttempts[0].RunID, nil
+	for _, a := range cAttempts {
+		if a.RunID != excludeRunID {
+			return a.RunID, nil
+		}
 	}
-	if len(cJobs) > 0 {
-		return cJobs[0].RunID, nil
+	for _, j := range cJobs {
+		if j.RunID != excludeRunID {
+			return j.RunID, nil
+		}
 	}
-
 	return 0, nil
 }
 
-func checkBlockedConcurrentRun(ctx context.Context, repoID, runID int64) (*jobsCheckResult, error) {
-	concurrentRun, err := actions_model.GetRunByRepoAndID(ctx, repoID, runID)
-	if err != nil {
-		return nil, fmt.Errorf("get run %d: %w", runID, err)
-	}
-	if concurrentRun.NeedApproval {
-		return &jobsCheckResult{}, nil
-	}
-
-	return checkJobsOfCurrentRunAttempt(ctx, concurrentRun)
-}
-
-// checkRunConcurrency rechecks runs blocked by concurrency that may become unblocked after the current run releases a workflow-level or job-level concurrency group.
-// RunIDsToReEmit propagates from inner checkJobsOfCurrentRunAttempt calls; see that function's doc.
+// checkRunConcurrency wakes a run blocked by concurrency that may become runnable now that
+// the current run's activity may have freed a workflow-level or job-level concurrency group.
 func checkRunConcurrency(ctx context.Context, run *actions_model.ActionRun) (*jobsCheckResult, error) {
 	result := &jobsCheckResult{}
 	checkedConcurrencyGroup := make(container.Set[string])
 
 	collect := func(concurrencyGroup string) error {
-		concurrentRunID, err := findBlockedRunIDByConcurrency(ctx, run.RepoID, concurrencyGroup)
-		if err != nil {
-			return fmt.Errorf("find blocked run by concurrency: %w", err)
-		}
-		if concurrentRunID > 0 {
-			r, err := checkBlockedConcurrentRun(ctx, run.RepoID, concurrentRunID)
-			if err != nil {
-				return err
-			}
-			result.merge(r)
-		}
 		checkedConcurrencyGroup.Add(concurrencyGroup)
+
+		// Exclude run.ID: this run's own jobs are resolved by checkJobsOfCurrentRunAttempt, no need to re-emit.
+		concurrentRunID, err := findConcurrencyWaiterToWake(ctx, run.RepoID, run.ID, concurrencyGroup)
+		if err != nil {
+			return err
+		}
+		if concurrentRunID == 0 {
+			return nil
+		}
+
+		concurrentRun, err := actions_model.GetRunByRepoAndID(ctx, run.RepoID, concurrentRunID)
+		if err != nil {
+			return fmt.Errorf("get concurrent run %d: %w", concurrentRunID, err)
+		}
+		// A run awaiting approval is not advanced by concurrency; ApproveRuns emits it once approved.
+		if concurrentRun.NeedApproval {
+			return nil
+		}
+
+		result.RunIDsToReEmit = append(result.RunIDsToReEmit, concurrentRunID)
 		return nil
 	}
 
@@ -282,10 +292,26 @@ func checkJobsOfCurrentRunAttempt(ctx context.Context, run *actions_model.Action
 				switch status {
 				case actions_model.StatusWaiting:
 					if err := expandReusableWorkflowCaller(ctx, run, attempt, job, vars); err != nil {
-						return fmt.Errorf("trigger caller-ready %d: %w", job.ID, err)
+						// Terminal expansion failure (an invalid/unresolvable reusable workflow): fail this caller.
+						log.Warn("caller %d cannot be expanded: %v", job.ID, err)
+						job.Status = actions_model.StatusFailure
+						job.Stopped = timeutil.TimeStampNow()
+						if n, uerr := actions_model.UpdateRunJob(ctx, job, builder.Eq{"status": actions_model.StatusBlocked, "is_expanded": false}, "status", "stopped"); uerr != nil {
+							return fmt.Errorf("mark unexpandable caller %d failed: %w", job.ID, uerr)
+						} else if n == 1 {
+							log.Warn("unexpandable caller %d has been marked as failed", job.ID)
+							result.UpdatedJobs = append(result.UpdatedJobs, job)
+							// Re-emit so the failed caller's dependents get resolved on the next pass.
+							expandedAnyCaller = true
+						} else {
+							// A concurrent writer advanced the caller; restore the in-memory state.
+							log.Warn("unexpandable caller %d has been advanced by a concurrent writer, not marking it failed", job.ID)
+							job.Status = actions_model.StatusBlocked
+							job.Stopped = 0
+						}
+					} else {
+						expandedAnyCaller = true
 					}
-					// expandReusableWorkflowCaller inserts children as Blocked. They need a follow-up resolver pass.
-					expandedAnyCaller = true
 				case actions_model.StatusSkipped:
 					job.Status = actions_model.StatusSkipped
 					if _, err := actions_model.UpdateRunJob(ctx, job, nil, "status"); err != nil {
@@ -477,7 +503,7 @@ type jobsCheckResult struct {
 	UpdatedJobs []*actions_model.ActionRunJob
 	// CancelledJobs are jobs cancelled by job-level concurrency while preparing to start.
 	CancelledJobs []*actions_model.ActionRunJob
-	// RunIDsToReEmit are runs whose newly expanded reusable workflow callers need another resolver pass.
+	// RunIDsToReEmit are runs that need another resolver pass in their own transaction.
 	RunIDsToReEmit []int64
 }
 

services/actions/job_emitter_test.go

@@ -203,7 +203,7 @@ func Test_checkRunConcurrency_NoDuplicateConcurrencyGroupCheck(t *testing.T) {
 	assert.NoError(t, unittest.PrepareTestDatabase())
 	ctx := t.Context()
 
-	// Run A: the triggering run of attempt A
+	// Run A: the triggering run of attempt A. It is done, so it no longer holds "test-cg", which is what lets checkRunConcurrency wake the blocked waiter.
 	runA := &actions_model.ActionRun{
 		RepoID:        4,
 		OwnerID:       1,
@@ -211,16 +211,16 @@ func Test_checkRunConcurrency_NoDuplicateConcurrencyGroupCheck(t *testing.T) {
 		WorkflowID:    "test.yml",
 		Index:         9901,
 		Ref:           "refs/heads/main",
-		Status:        actions_model.StatusRunning,
+		Status:        actions_model.StatusSuccess,
 	}
 	assert.NoError(t, db.Insert(ctx, runA))
 
-	// Attempt A: an attempt of run A with concurrency group "test-cg"
+	// Attempt A: a done attempt of run A with concurrency group "test-cg"
 	runAAttempt := &actions_model.ActionRunAttempt{
 		RepoID:           4,
 		RunID:            runA.ID,
 		Attempt:          1,
-		Status:           actions_model.StatusRunning,
+		Status:           actions_model.StatusSuccess,
 		ConcurrencyGroup: "test-cg",
 	}
 	assert.NoError(t, db.Insert(ctx, runAAttempt))
@@ -283,9 +283,11 @@ func Test_checkRunConcurrency_NoDuplicateConcurrencyGroupCheck(t *testing.T) {
 	result, err := checkRunConcurrency(ctx, runA)
 	assert.NoError(t, err)
 
-	if assert.Len(t, result.Jobs, 1) {
-		assert.Equal(t, jobBBlocked.ID, result.Jobs[0].ID)
+	// "test-cg" is free, so the single blocked waiter (run B) is collected for re-emit.
+	if assert.Len(t, result.RunIDsToReEmit, 1) {
+		assert.Equal(t, runB.ID, result.RunIDsToReEmit[0])
 	}
+	assert.Empty(t, result.Jobs)
 }
 
 // Test_checkJobsOfCurrentRunAttempt_RunLevelConcurrencyKeepsJobsBlocked verifies that
@@ -352,3 +354,82 @@ jobs:
 	refreshed := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: blockedJob.ID})
 	assert.Equal(t, actions_model.StatusBlocked, refreshed.Status)
 }
+
+// Test_checkRunConcurrency_HeldGroupDoesNotWake verifies that only an unoccupied concurrency group can wake up a blocked run/job.
+func Test_checkRunConcurrency_HeldGroupDoesNotWake(t *testing.T) {
+	assert.NoError(t, unittest.PrepareTestDatabase())
+	ctx := t.Context()
+
+	// Run A holds "test-cg": its attempt is still running.
+	runA := &actions_model.ActionRun{
+		RepoID: 4, OwnerID: 1, TriggerUserID: 1, WorkflowID: "test.yml",
+		Index: 9911, Ref: "refs/heads/main", Status: actions_model.StatusRunning,
+	}
+	assert.NoError(t, db.Insert(ctx, runA))
+	runAAttempt := &actions_model.ActionRunAttempt{
+		RepoID: 4, RunID: runA.ID, Attempt: 1, Status: actions_model.StatusRunning, ConcurrencyGroup: "test-cg",
+	}
+	assert.NoError(t, db.Insert(ctx, runAAttempt))
+	_, err := db.Exec(ctx, "UPDATE `action_run` SET latest_attempt_id = ? WHERE id = ?", runAAttempt.ID, runA.ID)
+	assert.NoError(t, err)
+
+	// Run B is blocked on the same group.
+	runB := &actions_model.ActionRun{
+		RepoID: 4, OwnerID: 1, TriggerUserID: 1, WorkflowID: "test.yml",
+		Index: 9912, Ref: "refs/heads/main", Status: actions_model.StatusBlocked,
+	}
+	assert.NoError(t, db.Insert(ctx, runB))
+	runBAttempt := &actions_model.ActionRunAttempt{
+		RepoID: 4, RunID: runB.ID, Attempt: 1, Status: actions_model.StatusBlocked, ConcurrencyGroup: "test-cg",
+	}
+	assert.NoError(t, db.Insert(ctx, runBAttempt))
+	_, err = db.Exec(ctx, "UPDATE `action_run` SET latest_attempt_id = ? WHERE id = ?", runBAttempt.ID, runB.ID)
+	assert.NoError(t, err)
+
+	runA, _, _ = db.GetByID[actions_model.ActionRun](ctx, runA.ID)
+	result, err := checkRunConcurrency(ctx, runA)
+	assert.NoError(t, err)
+
+	// The group is held by run A, so run B must not be woken; A will wake it when it releases the group.
+	assert.Empty(t, result.RunIDsToReEmit)
+}
+
+// Test_findConcurrencyWaiterToWake covers the finder's contract: it skips the run being processed (excludeRunID),
+// returns another blocked waiter when the group is free, and returns 0 while the group is still held.
+func Test_findConcurrencyWaiterToWake(t *testing.T) {
+	assert.NoError(t, unittest.PrepareTestDatabase())
+	ctx := t.Context()
+
+	const repoID int64 = 4
+	seed := func(index int64, group string, status actions_model.Status) *actions_model.ActionRun {
+		run := &actions_model.ActionRun{
+			RepoID: repoID, OwnerID: 1, TriggerUserID: 1, WorkflowID: "test.yml",
+			Index: index, Ref: "refs/heads/main", Status: status,
+		}
+		assert.NoError(t, db.Insert(ctx, run))
+		assert.NoError(t, db.Insert(ctx, &actions_model.ActionRunAttempt{
+			RepoID: repoID, RunID: run.ID, Attempt: 1, Status: status, ConcurrencyGroup: group,
+		}))
+		return run
+	}
+
+	// Free group "excl-cg" with two blocked runs: excluding self returns the other waiter, not self.
+	self := seed(99701, "excl-cg", actions_model.StatusBlocked)
+	other := seed(99702, "excl-cg", actions_model.StatusBlocked)
+	id, err := findConcurrencyWaiterToWake(ctx, repoID, self.ID, "excl-cg")
+	assert.NoError(t, err)
+	assert.Equal(t, other.ID, id)
+
+	// Free group "solo-cg" with only self blocked: excluding it leaves no waiter.
+	solo := seed(99703, "solo-cg", actions_model.StatusBlocked)
+	id, err = findConcurrencyWaiterToWake(ctx, repoID, solo.ID, "solo-cg")
+	assert.NoError(t, err)
+	assert.Equal(t, int64(0), id)
+
+	// Held group "held-cg" (a running holder) has a blocked waiter, but nothing is woken while held.
+	seed(99704, "held-cg", actions_model.StatusRunning)
+	seed(99705, "held-cg", actions_model.StatusBlocked)
+	id, err = findConcurrencyWaiterToWake(ctx, repoID, 0, "held-cg")
+	assert.NoError(t, err)
+	assert.Equal(t, int64(0), id)
+}

services/actions/reusable_workflow.go

@@ -5,6 +5,7 @@ package actions
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
 
@@ -236,30 +237,33 @@ func expandReusableWorkflowCaller(ctx context.Context, run *actions_model.Action
 		return fmt.Errorf("build call payload: %w", err)
 	}
 
-	// 8. Insert direct children of this caller.
-	existingChildren, err := actions_model.GetDirectChildJobsByParent(ctx, caller)
-	if err != nil {
-		return fmt.Errorf("get existing children of caller %d: %w", caller.ID, err)
-	}
-	if len(existingChildren) > 0 {
-		// Should not happen - child jobs cannot be expanded before the caller gets ready
-		return fmt.Errorf("invariant violation: caller %d has %d pre-existing children", caller.ID, len(existingChildren))
-	}
-	if err := insertCallerChildren(ctx, run, attempt, caller, content, contentSourceRepoID, contentSourceCommitSHA, vars, workflowCallInputs); err != nil {
-		return err
-	}
-
-	// 9. Update caller-related cols.
-	caller.CallPayload = string(callPayload)
+	// 8. Claim the expansion by flipping is_expanded false->true BEFORE inserting any children.
+	// Two concurrent expanders serialize on this row: exactly one winner matches (n==1) and owns the expansion.
+	// Children are only ever inserted by the claim winner, so no duplicate child rows can arise.
 	caller.IsExpanded = true
-	n, err := actions_model.UpdateRunJob(ctx, caller,
+	n, err := actions_model.UpdateRunJob(ctx, caller, builder.And(
 		builder.Eq{"is_expanded": false},
-		"call_secrets", "reusable_workflow_content", "call_payload", "is_expanded")
+		builder.In("status", actions_model.StatusBlocked, actions_model.StatusWaiting),
+	), "is_expanded")
 	if err != nil {
-		return fmt.Errorf("commit caller %d expansion: %w", caller.ID, err)
+		caller.IsExpanded = false // the claim was not established
+		return fmt.Errorf("claim caller %d expansion: %w", caller.ID, err)
 	}
 	if n == 0 {
-		return fmt.Errorf("caller %d already expanded by another writer", caller.ID)
+		// Another writer won the expansion, or the caller has been moved to a terminal status (e.g. failed/cancelled).
+		return nil
+	}
+
+	// 9. We own the expansion: insert the direct children.
+	if err := insertCallerChildren(ctx, run, attempt, caller, content, contentSourceRepoID, contentSourceCommitSHA, vars, workflowCallInputs); err != nil {
+		// On failure, undo the partial expansion so an error return always leaves the caller unexpanded and childless.
+		return errors.Join(err, undoExpansion(ctx, caller))
+	}
+
+	// 10. Persist the remaining caller metadata (the row is already ours via the claim above).
+	caller.CallPayload = string(callPayload)
+	if _, err := actions_model.UpdateRunJob(ctx, caller, nil, "call_secrets", "reusable_workflow_content", "call_payload"); err != nil {
+		return errors.Join(fmt.Errorf("persist caller %d expansion metadata: %w", caller.ID, err), undoExpansion(ctx, caller))
 	}
 	return nil
 }
@@ -375,3 +379,16 @@ func ResolveUses(ctx context.Context, uses string) (*jobparser.UsesRef, error) {
 	}
 	return ref, nil
 }
+
+// undoExpansion rolls back a partial expansion owned by the current transaction:
+// it removes the inserted children and releases the is_expanded claim itself.
+func undoExpansion(ctx context.Context, caller *actions_model.ActionRunJob) error {
+	if err := actions_model.DeleteDirectChildJobsByParent(ctx, caller); err != nil {
+		return fmt.Errorf("delete children of caller %d: %w", caller.ID, err)
+	}
+	caller.IsExpanded = false
+	if _, err := actions_model.UpdateRunJob(ctx, caller, nil, "is_expanded"); err != nil {
+		return fmt.Errorf("release caller %d expansion claim: %w", caller.ID, err)
+	}
+	return nil
+}

services/actions/reusable_workflow_test.go

@@ -206,3 +206,34 @@ func TestResolveUses(t *testing.T) {
 		assert.ErrorContains(t, err, "must point to this Gitea instance")
 	})
 }
+
+func TestUndoExpansion(t *testing.T) {
+	require.NoError(t, unittest.PrepareTestDatabase())
+	ctx := t.Context()
+
+	// A claimed caller with two children inserted by the aborted expansion, plus a sibling that must survive.
+	caller := &actions_model.ActionRunJob{
+		RunID: 991, RepoID: 4, OwnerID: 1, JobID: "caller", Name: "caller",
+		Status: actions_model.StatusBlocked, IsReusableCaller: true, IsExpanded: true,
+	}
+	require.NoError(t, db.Insert(ctx, caller))
+	for _, jobID := range []string{"child1", "child2"} {
+		require.NoError(t, db.Insert(ctx, &actions_model.ActionRunJob{
+			RunID: 991, RepoID: 4, OwnerID: 1, JobID: jobID, Name: jobID,
+			Status: actions_model.StatusBlocked, ParentJobID: caller.ID,
+		}))
+	}
+	sibling := &actions_model.ActionRunJob{
+		RunID: 991, RepoID: 4, OwnerID: 1, JobID: "sibling", Name: "sibling",
+		Status: actions_model.StatusBlocked,
+	}
+	require.NoError(t, db.Insert(ctx, sibling))
+
+	require.NoError(t, undoExpansion(ctx, caller))
+
+	assert.Equal(t, 0, unittest.GetCount(t, &actions_model.ActionRunJob{ParentJobID: caller.ID}))
+	assert.False(t, caller.IsExpanded)
+	refreshed := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: caller.ID})
+	assert.False(t, refreshed.IsExpanded)
+	unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: sibling.ID})
+}

tests/integration/actions_reusable_workflow_test.go

@@ -574,6 +574,49 @@ jobs:
 			assert.Equal(t, 0, unittest.GetCount(t, &actions_model.ActionRun{RepoID: repo.ID}))
 		})
 
+		t.Run("Nested caller with missing callee fails instead of blocking", func(t *testing.T) {
+			// When the expansion hits a terminal error (e.g. missing callee), the emitter must fail the caller and let the run finish as failed, not retry the expansion forever.
+			apiRepo := createActionsTestRepo(t, user2Token, "nested-caller-missing-callee", false)
+			repo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: apiRepo.ID})
+
+			runner := newMockRunner()
+			runner.registerAsRepoRunner(t, repo.OwnerName, repo.Name, "mock-runner", []string{"ubuntu-latest"}, false)
+
+			createRepoWorkflowFile(t, user2, user2Token, repo, ".gitea/workflows/caller.yaml",
+				`name: Caller
+on: push
+jobs:
+  plain_job:
+    runs-on: ubuntu-latest
+    steps:
+      - run: echo 'job'
+  bad_caller:
+    needs: plain_job
+    uses: ./.gitea/workflows/does-not-exist.yml
+`)
+
+			// plain_job runs first; bad_caller is Blocked on needs and is NOT expanded at creation.
+			plainTask := runner.fetchTask(t)
+			_, plainJob, run := getTaskAndJobAndRunByTaskID(t, plainTask.Id)
+			assert.Equal(t, "plain_job", plainJob.JobID)
+			badCallerPre := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{RunID: run.ID, JobID: "bad_caller"})
+			assert.Equal(t, actions_model.StatusBlocked, badCallerPre.Status)
+			assert.False(t, badCallerPre.IsExpanded)
+
+			runner.execTask(t, plainTask, &mockTaskOutcome{result: runnerv1.Result_RESULT_SUCCESS})
+
+			// The emitter now tries to expand bad_caller, hits the missing callee, and fails the caller.
+			badCaller := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: badCallerPre.ID})
+			assert.Equal(t, actions_model.StatusFailure, badCaller.Status)
+			// No children were inserted (the terminal error precedes the child inserts).
+			assert.Equal(t, 0, unittest.GetCount(t, &actions_model.ActionRunJob{ParentJobID: badCallerPre.ID}))
+
+			finalRun := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRun{ID: run.ID})
+			assert.Equal(t, actions_model.StatusFailure, finalRun.Status)
+
+			runner.fetchNoTask(t) // no task scheduled for the failed caller; the run is not stuck
+		})
+
 		t.Run("Fork PR with secrets: inherit does not leak base repo secrets", func(t *testing.T) {
 			// user2 owns the base repo, configures a secret, and registers a reusable workflow that declares a required secret.
 			// The caller workflow uses `secrets: inherit`.