diff --git a/docs/pipelines.md b/docs/pipelines.md
index a7f62ce0c0f..38c4ad46520 100644
--- a/docs/pipelines.md
+++ b/docs/pipelines.md
@@ -820,6 +820,44 @@ Overall, `PipelineRun` state transitioning is explained below for respective sce
 Please refer to the [table](pipelineruns.md#monitoring-execution-status) under Monitoring Execution Status to learn about
 what kind of events are triggered based on the `Pipelinerun` status.
 
+
+### Using Execution `Status` of `pipelineTask`
+
+Finally Task can utilize execution status of any of the `pipelineTasks` under `tasks` section using param:
+
+```yaml
+    finally:
+    - name: finaltask
+      params:
+        - name: task1Status
+          value: "$(tasks.task1.status)"
+      taskSpec:
+        params:
+          - name: task1Status
+        steps:
+          - image: ubuntu
+            name: print-task-status
+            script: |
+              if [ $(params.task1Status) == "Failed" ]
+              then
+                echo "Task1 has failed, continue processing the failure"
+              fi
+```
+
+This kind of variable can have any one of the values from the following table:
+
+| Status | Description |
+| ------- | -----------|
+| Succeeded | `taskRun` for the `pipelineTask` completed successfully |
+| Failed | `taskRun` for the `pipelineTask` completed with a failure or cancelled by the user |
+| None | the `pipelineTask` has been skipped or no execution information available for the `pipelineTask` |
+
+For an end-to-end example, see [`status` in a `PipelineRun`](../examples/v1beta1/pipelineruns/pipelinerun-task-execution-status.yaml).
+
+**Note:** `$(tasks.<pipelineTask>.status)` is instantiated and available at runtime and must be used as a param value
+as is without concatenating it with any other param or string, for example, this kind of usage is not validated/supported
+`task status is $(tasks.<pipelineTask>.status)`.
+
 ### Known Limitations
 
 ### Specifying `Resources` in Final Tasks
diff --git a/docs/variables.md b/docs/variables.md
index 61df9d6cc9a..84651369d7e 100644
--- a/docs/variables.md
+++ b/docs/variables.md
@@ -23,6 +23,7 @@ For instructions on using variable substitutions see the relevant section of [th
 | `context.pipelineRun.namespace` | The namespace of the `PipelineRun` that this `Pipeline` is running in. |
 | `context.pipelineRun.uid` | The uid of the `PipelineRun` that this `Pipeline` is running in. |
 | `context.pipeline.name` | The name of this `Pipeline` . |
+| `tasks.<pipelineTaskName>.status` | The execution status of the specified `pipelineTask`, only available in `finally` tasks. |
 
 
 ## Variables available in a `Task`
diff --git a/examples/v1beta1/pipelineruns/pipelinerun-task-execution-status.yaml b/examples/v1beta1/pipelineruns/pipelinerun-task-execution-status.yaml
new file mode 100644
index 00000000000..57cc38af235
--- /dev/null
+++ b/examples/v1beta1/pipelineruns/pipelinerun-task-execution-status.yaml
@@ -0,0 +1,44 @@
+kind: PipelineRun
+apiVersion: tekton.dev/v1beta1
+metadata:
+  generateName: pr-execution-status-
+spec:
+  serviceAccountName: 'default'
+  pipelineSpec:
+    tasks:
+    - name: task1 # successful task
+      taskSpec:
+        steps:
+        - image: ubuntu
+          name: hello
+          script: |
+            echo "Hello World!"
+    - name: task2 # skipped task
+      when:
+        - input: "true"
+          operator: "notin"
+          values: ["true"]
+      taskSpec:
+        steps:
+          - image: ubuntu
+            name: success
+            script: |
+              exit 0
+    finally:
+    - name: task3 # this task verifies the status of dag tasks, it fails if verification fails
+      params:
+        - name: task1Status
+          value: "$(tasks.task1.status)"
+        - name: task2Status
+          value: "$(tasks.task2.status)"
+      taskSpec:
+        params:
+          - name: task1Status
+          - name: task2Status
+        steps:
+          - image: alpine
+            name: verify-dag-task-status
+            script: |
+              if [[ $(params.task1Status) != "Succeeded" ||  $(params.task2Status) != "None" ]]; then
+                exit 1;
+              fi
diff --git a/pkg/apis/pipeline/v1beta1/pipeline_validation.go b/pkg/apis/pipeline/v1beta1/pipeline_validation.go
index 8a6351fcda3..fc1864f0d63 100644
--- a/pkg/apis/pipeline/v1beta1/pipeline_validation.go
+++ b/pkg/apis/pipeline/v1beta1/pipeline_validation.go
@@ -62,6 +62,7 @@ func (ps *PipelineSpec) Validate(ctx context.Context) (errs *apis.FieldError) {
 	errs = errs.Also(validatePipelineParameterVariables(ps.Tasks, ps.Params).ViaField("tasks"))
 	errs = errs.Also(validatePipelineParameterVariables(ps.Finally, ps.Params).ViaField("finally"))
 	errs = errs.Also(validatePipelineContextVariables(ps.Tasks))
+	errs = errs.Also(validateExecutionStatusVariables(ps.Tasks, ps.Finally))
 	// Validate the pipeline's workspaces.
 	errs = errs.Also(validatePipelineWorkspaces(ps.Workspaces, ps.Tasks, ps.Finally))
 	// Validate the pipeline's results
@@ -290,6 +291,40 @@ func validatePipelineContextVariables(tasks []PipelineTask) *apis.FieldError {
 	return errs.Also(validatePipelineContextVariablesInParamValues(paramValues, "context\\.pipeline", pipelineContextNames))
 }
 
+func validateExecutionStatusVariables(tasks []PipelineTask, finallyTasks []PipelineTask) (errs *apis.FieldError) {
+	// creating a list of pipelineTask names to validate tasks.<name>.status
+	pipelineRunTasksContextNames := sets.String{}
+	for idx, t := range tasks {
+		for _, param := range t.Params {
+			// validate dag pipeline tasks not accessing execution status of other pipeline task
+			if ps, ok := GetVarSubstitutionExpressionsForParam(param); ok {
+				for _, p := range ps {
+					if strings.HasPrefix(p, "tasks.") && strings.HasSuffix(p, ".status") {
+						errs = errs.Also(apis.ErrInvalidValue(fmt.Sprintf("pipeline tasks can not refer to execution status of any other pipeline task"),
+							"value").ViaFieldKey("params", param.Name).ViaFieldIndex("tasks", idx))
+					}
+				}
+			}
+		}
+		pipelineRunTasksContextNames.Insert(t.Name)
+	}
+
+	// validate finally tasks accessing execution status of a dag task specified in the pipeline
+	var paramValues []string
+	for _, t := range finallyTasks {
+		for _, param := range t.Params {
+			paramValues = append(paramValues, param.Value.StringVal)
+			paramValues = append(paramValues, param.Value.ArrayVal...)
+		}
+	}
+	for _, paramValue := range paramValues {
+		if strings.HasPrefix(stripVarSubExpression(paramValue), "tasks.") && strings.HasSuffix(stripVarSubExpression(paramValue), ".status") {
+			errs = errs.Also(substitution.ValidateVariablePS(paramValue, "tasks", "status", pipelineRunTasksContextNames).ViaField("value"))
+		}
+	}
+	return errs
+}
+
 func validatePipelineContextVariablesInParamValues(paramValues []string, prefix string, contextNames sets.String) (errs *apis.FieldError) {
 	for _, paramValue := range paramValues {
 		errs = errs.Also(substitution.ValidateVariableP(paramValue, prefix, contextNames).ViaField("value"))
diff --git a/pkg/apis/pipeline/v1beta1/pipeline_validation_test.go b/pkg/apis/pipeline/v1beta1/pipeline_validation_test.go
index d2fab8280a9..2498256710d 100644
--- a/pkg/apis/pipeline/v1beta1/pipeline_validation_test.go
+++ b/pkg/apis/pipeline/v1beta1/pipeline_validation_test.go
@@ -2116,6 +2116,80 @@ func TestContextInvalid(t *testing.T) {
 	}
 }
 
+func TestPipelineTasksExecutionStatus(t *testing.T) {
+	tests := []struct {
+		name          string
+		tasks         []PipelineTask
+		finalTasks    []PipelineTask
+		expectedError apis.FieldError
+	}{{
+		name: "valid string variable in finally accessing pipelineTask status",
+		tasks: []PipelineTask{{
+			Name: "foo",
+		}},
+		finalTasks: []PipelineTask{{
+			Name:    "bar",
+			TaskRef: &TaskRef{Name: "bar-task"},
+			Params: []Param{{
+				Name: "foo-status", Value: ArrayOrString{Type: ParamTypeString, StringVal: "$(tasks.foo.status)"},
+			}},
+		}},
+	}, {
+		name: "invalid string variable in dag task accessing pipelineTask status",
+		tasks: []PipelineTask{{
+			Name:    "foo",
+			TaskRef: &TaskRef{Name: "foo-task"},
+			Params: []Param{{
+				Name: "bar-status", Value: ArrayOrString{Type: ParamTypeString, StringVal: "$(tasks.bar.status)"},
+			}},
+		}},
+		expectedError: apis.FieldError{
+			Message: `invalid value: pipeline tasks can not refer to execution status of any other pipeline task`,
+			Paths:   []string{"tasks[0].params[bar-status].value"},
+		},
+	}, {
+		name: "invalid array variable in dag task accessing pipelineTask status",
+		tasks: []PipelineTask{{
+			Name:    "foo",
+			TaskRef: &TaskRef{Name: "foo-task"},
+			Params: []Param{{
+				Name: "bar-status", Value: ArrayOrString{Type: ParamTypeArray, ArrayVal: []string{"$(tasks.bar.status)"}},
+			}},
+		}},
+		expectedError: apis.FieldError{
+			Message: `invalid value: pipeline tasks can not refer to execution status of any other pipeline task`,
+			Paths:   []string{"tasks[0].params[bar-status].value"},
+		},
+	}, {
+		name: "invalid string variable in finally accessing missing pipelineTask status",
+		finalTasks: []PipelineTask{{
+			Name:    "bar",
+			TaskRef: &TaskRef{Name: "bar-task"},
+			Params: []Param{{
+				Name: "notask-status", Value: ArrayOrString{Type: ParamTypeString, StringVal: "$(tasks.notask.status)"},
+			}},
+		}},
+		expectedError: *apis.ErrGeneric(`non-existent variable in "$(tasks.notask.status)"`, "value"),
+	}}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := validateExecutionStatusVariables(tt.tasks, tt.finalTasks)
+			if len(tt.expectedError.Error()) == 0 {
+				if err != nil {
+					t.Errorf("Pipeline.validateExecutionStatusVariables() returned error for valid pipeline variable accessing execution status: %s: %v", tt.name, err)
+				}
+			} else {
+				if err == nil {
+					t.Errorf("Pipeline.validateExecutionStatusVariables() did not return error for invalid pipeline parameters accessing execution status: %s, %s", tt.name, tt.tasks[0].Params)
+				}
+				if d := cmp.Diff(tt.expectedError.Error(), err.Error(), cmpopts.IgnoreUnexported(apis.FieldError{})); d != "" {
+					t.Errorf("PipelineSpec.Validate() errors diff %s", diff.PrintWantGot(d))
+				}
+			}
+		})
+	}
+}
+
 func getTaskSpec() TaskSpec {
 	return TaskSpec{
 		Steps: []Step{{
diff --git a/pkg/reconciler/pipelinerun/pipelinerun.go b/pkg/reconciler/pipelinerun/pipelinerun.go
index ce2b6412492..3cec41a8450 100644
--- a/pkg/reconciler/pipelinerun/pipelinerun.go
+++ b/pkg/reconciler/pipelinerun/pipelinerun.go
@@ -581,7 +581,12 @@ func (c *Reconciler) runNextSchedulableTask(ctx context.Context, pr *v1beta1.Pip
 	pipelineRunFacts.ResetSkippedCache()
 
 	// GetFinalTasks only returns tasks when a DAG is complete
-	nextRprts = append(nextRprts, pipelineRunFacts.GetFinalTasks()...)
+	fnextRprts := pipelineRunFacts.GetFinalTasks()
+	if len(fnextRprts) != 0 {
+		// apply the runtime context just before creating taskRuns for final tasks in queue
+		resources.ApplyPipelineTaskContext(fnextRprts, pipelineRunFacts.GetPipelineTaskStatus(ctx))
+		nextRprts = append(nextRprts, fnextRprts...)
+	}
 
 	for _, rprt := range nextRprts {
 		if rprt == nil || rprt.Skip(pipelineRunFacts) {
diff --git a/pkg/reconciler/pipelinerun/pipelinerun_test.go b/pkg/reconciler/pipelinerun/pipelinerun_test.go
index 3cd461c2a11..8b9a04379c7 100644
--- a/pkg/reconciler/pipelinerun/pipelinerun_test.go
+++ b/pkg/reconciler/pipelinerun/pipelinerun_test.go
@@ -4548,6 +4548,103 @@ func TestReconcilePipeline_TaskSpecMetadata(t *testing.T) {
 	}
 }
 
+func TestReconciler_ReconcileKind_PipelineTaskContext(t *testing.T) {
+	names.TestingSeed()
+
+	pipelineName := "p-pipelinetask-status"
+	pipelineRunName := "pr-pipelinetask-status"
+
+	ps := []*v1beta1.Pipeline{tb.Pipeline(pipelineName, tb.PipelineNamespace("foo"), tb.PipelineSpec(
+		tb.PipelineTask("task1", "mytask"),
+		tb.FinalPipelineTask("finaltask", "finaltask",
+			tb.PipelineTaskParam("pipelineRun-tasks-task1", "$(tasks.task1.status)"),
+		),
+	))}
+
+	prs := []*v1beta1.PipelineRun{tb.PipelineRun(pipelineRunName, tb.PipelineRunNamespace("foo"),
+		tb.PipelineRunSpec(pipelineName, tb.PipelineRunServiceAccountName("test-sa")),
+	)}
+
+	ts := []*v1beta1.Task{
+		tb.Task("mytask", tb.TaskNamespace("foo")),
+		tb.Task("finaltask", tb.TaskNamespace("foo"),
+			tb.TaskSpec(
+				tb.TaskParam("pipelineRun-tasks-task1", v1beta1.ParamTypeString),
+			),
+		),
+	}
+
+	trs := []*v1beta1.TaskRun{
+		tb.TaskRun(pipelineRunName+"-task1-xxyyy",
+			tb.TaskRunNamespace("foo"),
+			tb.TaskRunOwnerReference("PipelineRun", pipelineRunName,
+				tb.OwnerReferenceAPIVersion("tekton.dev/v1beta1"),
+				tb.Controller, tb.BlockOwnerDeletion,
+			),
+			tb.TaskRunLabel("tekton.dev/pipeline", pipelineName),
+			tb.TaskRunLabel("tekton.dev/pipelineRun", pipelineRunName),
+			tb.TaskRunLabel("tekton.dev/pipelineTask", "task1"),
+			tb.TaskRunSpec(
+				tb.TaskRunTaskRef("mytask"),
+				tb.TaskRunServiceAccountName("test-sa"),
+			),
+			tb.TaskRunStatus(
+				tb.StatusCondition(
+					apis.Condition{
+						Type:   apis.ConditionSucceeded,
+						Status: corev1.ConditionFalse,
+						Reason: v1beta1.TaskRunReasonFailed.String(),
+					},
+				),
+			),
+		),
+	}
+
+	d := test.Data{
+		PipelineRuns: prs,
+		Pipelines:    ps,
+		Tasks:        ts,
+		TaskRuns:     trs,
+	}
+	prt := NewPipelineRunTest(d, t)
+	defer prt.Cancel()
+
+	_, clients := prt.reconcileRun("foo", pipelineRunName, []string{}, false)
+
+	expectedTaskRunName := pipelineRunName + "-finaltask-9l9zj"
+	expectedTaskRun := tb.TaskRun(expectedTaskRunName,
+		tb.TaskRunNamespace("foo"),
+		tb.TaskRunOwnerReference("PipelineRun", pipelineRunName,
+			tb.OwnerReferenceAPIVersion("tekton.dev/v1beta1"),
+			tb.Controller, tb.BlockOwnerDeletion,
+		),
+		tb.TaskRunLabel("tekton.dev/pipeline", pipelineName),
+		tb.TaskRunLabel("tekton.dev/pipelineRun", pipelineRunName),
+		tb.TaskRunLabel("tekton.dev/pipelineTask", "finaltask"),
+		tb.TaskRunSpec(
+			tb.TaskRunTaskRef("finaltask"),
+			tb.TaskRunServiceAccountName("test-sa"),
+			tb.TaskRunParam("pipelineRun-tasks-task1", "Failed"),
+		),
+	)
+	// Check that the expected TaskRun was created
+	actual, err := clients.Pipeline.TektonV1beta1().TaskRuns("foo").List(prt.TestAssets.Ctx, metav1.ListOptions{
+		LabelSelector: "tekton.dev/pipelineTask=finaltask,tekton.dev/pipelineRun=" + pipelineRunName,
+		Limit:         1,
+	})
+
+	if err != nil {
+		t.Fatalf("Failure to list TaskRun's %s", err)
+	}
+	if len(actual.Items) != 1 {
+		t.Fatalf("Expected 1 TaskRuns got %d", len(actual.Items))
+	}
+	actualTaskRun := actual.Items[0]
+	if d := cmp.Diff(&actualTaskRun, expectedTaskRun, ignoreResourceVersion); d != "" {
+		t.Errorf("expected to see TaskRun %v created. Diff %s", expectedTaskRunName, diff.PrintWantGot(d))
+	}
+}
+
 // NewPipelineRunTest returns PipelineRunTest with a new PipelineRun controller created with specified state through data
 // This PipelineRunTest can be reused for multiple PipelineRuns by calling reconcileRun for each pipelineRun
 func NewPipelineRunTest(data test.Data, t *testing.T) *PipelineRunTest {
diff --git a/pkg/reconciler/pipelinerun/resources/apply.go b/pkg/reconciler/pipelinerun/resources/apply.go
index 5c2f862b6c0..97177b7f812 100644
--- a/pkg/reconciler/pipelinerun/resources/apply.go
+++ b/pkg/reconciler/pipelinerun/resources/apply.go
@@ -87,6 +87,17 @@ func ApplyTaskResults(targets PipelineRunState, resolvedResultRefs ResolvedResul
 	}
 }
 
+//ApplyPipelineTaskContext replaces context variables referring to execution status with the specified status
+func ApplyPipelineTaskContext(state PipelineRunState, replacements map[string]string) {
+	for _, resolvedPipelineRunTask := range state {
+		if resolvedPipelineRunTask.PipelineTask != nil {
+			pipelineTask := resolvedPipelineRunTask.PipelineTask.DeepCopy()
+			pipelineTask.Params = replaceParamValues(pipelineTask.Params, replacements, nil)
+			resolvedPipelineRunTask.PipelineTask = pipelineTask
+		}
+	}
+}
+
 // ApplyWorkspaces replaces workspace variables in the given pipeline spec with their
 // concrete values.
 func ApplyWorkspaces(p *v1beta1.PipelineSpec, pr *v1beta1.PipelineRun) *v1beta1.PipelineSpec {
diff --git a/pkg/reconciler/pipelinerun/resources/apply_test.go b/pkg/reconciler/pipelinerun/resources/apply_test.go
index bf0554ad601..636130f3be1 100644
--- a/pkg/reconciler/pipelinerun/resources/apply_test.go
+++ b/pkg/reconciler/pipelinerun/resources/apply_test.go
@@ -882,4 +882,42 @@ func TestApplyTaskResultsToPipelineResults(t *testing.T) {
 			}
 		})
 	}
+
+}
+
+func TestApplyTaskRunContext(t *testing.T) {
+	r := map[string]string{
+		"tasks.task1.status": "succeeded",
+		"tasks.task3.status": "none",
+	}
+	state := PipelineRunState{{
+		PipelineTask: &v1beta1.PipelineTask{
+			Name:    "task4",
+			TaskRef: &v1beta1.TaskRef{Name: "task"},
+			Params: []v1beta1.Param{{
+				Name:  "task1",
+				Value: *v1beta1.NewArrayOrString("$(tasks.task1.status)"),
+			}, {
+				Name:  "task3",
+				Value: *v1beta1.NewArrayOrString("$(tasks.task3.status)"),
+			}},
+		},
+	}}
+	expectedState := PipelineRunState{{
+		PipelineTask: &v1beta1.PipelineTask{
+			Name:    "task4",
+			TaskRef: &v1beta1.TaskRef{Name: "task"},
+			Params: []v1beta1.Param{{
+				Name:  "task1",
+				Value: *v1beta1.NewArrayOrString("succeeded"),
+			}, {
+				Name:  "task3",
+				Value: *v1beta1.NewArrayOrString("none"),
+			}},
+		},
+	}}
+	ApplyPipelineTaskContext(state, r)
+	if d := cmp.Diff(expectedState, state); d != "" {
+		t.Fatalf("ApplyTaskRunContext() %s", diff.PrintWantGot(d))
+	}
 }
diff --git a/pkg/reconciler/pipelinerun/resources/pipelinerunresolution.go b/pkg/reconciler/pipelinerun/resources/pipelinerunresolution.go
index 5e01eb8f307..baacbd66f57 100644
--- a/pkg/reconciler/pipelinerun/resources/pipelinerunresolution.go
+++ b/pkg/reconciler/pipelinerun/resources/pipelinerunresolution.go
@@ -132,6 +132,18 @@ func (t ResolvedPipelineRunTask) IsStarted() bool {
 	return t.TaskRun != nil && t.TaskRun.Status.GetCondition(apis.ConditionSucceeded) != nil
 }
 
+// IsConditionStatusFalse returns true when a task has succeeded condition with status set to false
+// it includes task failed after retries are exhausted, cancelled tasks, and time outs
+func (t ResolvedPipelineRunTask) IsConditionStatusFalse() bool {
+	if t.IsStarted() {
+		if t.IsCustomTask() {
+			return t.Run.Status.GetCondition(apis.ConditionSucceeded).IsFalse()
+		}
+		return t.TaskRun.Status.GetCondition(apis.ConditionSucceeded).IsFalse()
+	}
+	return false
+}
+
 func (t *ResolvedPipelineRunTask) checkParentsDone(facts *PipelineRunFacts) bool {
 	stateMap := facts.State.ToMap()
 	node := facts.TasksGraph.Nodes[t.PipelineTask.Name]
diff --git a/pkg/reconciler/pipelinerun/resources/pipelinerunstate.go b/pkg/reconciler/pipelinerun/resources/pipelinerunstate.go
index 10382dc1cbc..a6a13721657 100644
--- a/pkg/reconciler/pipelinerun/resources/pipelinerunstate.go
+++ b/pkg/reconciler/pipelinerun/resources/pipelinerunstate.go
@@ -17,6 +17,7 @@ limitations under the License.
 package resources
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/tektoncd/pipeline/pkg/apis/pipeline/v1beta1"
@@ -28,6 +29,15 @@ import (
 	"knative.dev/pkg/apis"
 )
 
+const (
+	// ReasonUnknown indicates that the execution status of a pipelineTask is unknown
+	PipelineTaskStateNone = "None"
+	// PipelineTaskStatusPrefix is a prefix of the param representing execution state of pipelineTask
+	PipelineTaskStatusPrefix = "tasks."
+	// PipelineTaskStatusSuffix is a suffix of the param representing execution state of pipelineTask
+	PipelineTaskStatusSuffix = ".status"
+)
+
 // PipelineRunState is a slice of ResolvedPipelineRunTasks the represents the current execution
 // state of the PipelineRun.
 type PipelineRunState []*ResolvedPipelineRunTask
@@ -370,6 +380,31 @@ func (facts *PipelineRunFacts) GetSkippedTasks() []v1beta1.SkippedTask {
 	return skipped
 }
 
+// GetPipelineTaskStatus returns the status of a PipelineTask depending on its taskRun
+// the checks are implemented such that the finally tasks are requesting status of the dag tasks
+func (facts *PipelineRunFacts) GetPipelineTaskStatus(ctx context.Context) map[string]string {
+	// construct a map of tasks.<pipelineTask>.status and its state
+	tStatus := make(map[string]string)
+	for _, t := range facts.State {
+		if facts.isDAGTask(t.PipelineTask.Name) {
+			var s string
+			switch {
+			// execution status is Succeeded when a task has succeeded condition with status set to true
+			case t.IsSuccessful():
+				s = v1beta1.TaskRunReasonSuccessful.String()
+			// execution status is Failed when a task has succeeded condition with status set to false
+			case t.IsConditionStatusFalse():
+				s = v1beta1.TaskRunReasonFailed.String()
+			default:
+				// None includes skipped as well
+				s = PipelineTaskStateNone
+			}
+			tStatus[PipelineTaskStatusPrefix+t.PipelineTask.Name+PipelineTaskStatusSuffix] = s
+		}
+	}
+	return tStatus
+}
+
 // successfulOrSkippedTasks returns a list of the names of all of the PipelineTasks in state
 // which have successfully completed or skipped
 func (facts *PipelineRunFacts) successfulOrSkippedDAGTasks() []string {
diff --git a/pkg/reconciler/pipelinerun/resources/pipelinerunstate_test.go b/pkg/reconciler/pipelinerun/resources/pipelinerunstate_test.go
index 06fc000cb86..088d7d5b2df 100644
--- a/pkg/reconciler/pipelinerun/resources/pipelinerunstate_test.go
+++ b/pkg/reconciler/pipelinerun/resources/pipelinerunstate_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package resources
 
 import (
+	"context"
 	"fmt"
 	"testing"
 	"time"
@@ -1318,3 +1319,123 @@ func TestAdjustStartTime(t *testing.T) {
 		})
 	}
 }
+
+func TestPipelineRunFacts_GetPipelineTaskStatus(t *testing.T) {
+	tcs := []struct {
+		name           string
+		state          PipelineRunState
+		dagTasks       []v1beta1.PipelineTask
+		expectedStatus map[string]string
+	}{{
+		name:     "no-tasks-started",
+		state:    noneStartedState,
+		dagTasks: []v1beta1.PipelineTask{pts[0], pts[1]},
+		expectedStatus: map[string]string{
+			PipelineTaskStatusPrefix + pts[0].Name + PipelineTaskStatusSuffix: PipelineTaskStateNone,
+			PipelineTaskStatusPrefix + pts[1].Name + PipelineTaskStatusSuffix: PipelineTaskStateNone,
+		},
+	}, {
+		name:     "one-task-started",
+		state:    oneStartedState,
+		dagTasks: []v1beta1.PipelineTask{pts[0], pts[1]},
+		expectedStatus: map[string]string{
+			PipelineTaskStatusPrefix + pts[0].Name + PipelineTaskStatusSuffix: PipelineTaskStateNone,
+			PipelineTaskStatusPrefix + pts[1].Name + PipelineTaskStatusSuffix: PipelineTaskStateNone,
+		},
+	}, {
+		name:     "one-task-finished",
+		state:    oneFinishedState,
+		dagTasks: []v1beta1.PipelineTask{pts[0], pts[1]},
+		expectedStatus: map[string]string{
+			PipelineTaskStatusPrefix + pts[0].Name + PipelineTaskStatusSuffix: v1beta1.TaskRunReasonSuccessful.String(),
+			PipelineTaskStatusPrefix + pts[1].Name + PipelineTaskStatusSuffix: PipelineTaskStateNone,
+		},
+	}, {
+		name:     "one-task-failed",
+		state:    oneFailedState,
+		dagTasks: []v1beta1.PipelineTask{pts[0], pts[1]},
+		expectedStatus: map[string]string{
+			PipelineTaskStatusPrefix + pts[0].Name + PipelineTaskStatusSuffix: v1beta1.TaskRunReasonFailed.String(),
+			PipelineTaskStatusPrefix + pts[1].Name + PipelineTaskStatusSuffix: PipelineTaskStateNone,
+		},
+	}, {
+		name:     "all-finished",
+		state:    allFinishedState,
+		dagTasks: []v1beta1.PipelineTask{pts[0], pts[1]},
+		expectedStatus: map[string]string{
+			PipelineTaskStatusPrefix + pts[0].Name + PipelineTaskStatusSuffix: v1beta1.TaskRunReasonSuccessful.String(),
+			PipelineTaskStatusPrefix + pts[1].Name + PipelineTaskStatusSuffix: v1beta1.TaskRunReasonSuccessful.String(),
+		},
+	}, {
+		name: "task-with-when-expressions-passed",
+		state: PipelineRunState{{
+			PipelineTask: &pts[9],
+			TaskRunName:  "pr-guard-succeeded-task-not-started",
+			TaskRun:      nil,
+			ResolvedTaskResources: &resources.ResolvedTaskResources{
+				TaskSpec: &task.Spec,
+			},
+		}},
+		dagTasks: []v1beta1.PipelineTask{pts[9]},
+		expectedStatus: map[string]string{
+			PipelineTaskStatusPrefix + pts[9].Name + PipelineTaskStatusSuffix: PipelineTaskStateNone,
+		},
+	}, {
+		name: "tasks-when-expression-failed-and-task-skipped",
+		state: PipelineRunState{{
+			PipelineTask: &pts[10],
+			TaskRunName:  "pr-guardedtask-skipped",
+			ResolvedTaskResources: &resources.ResolvedTaskResources{
+				TaskSpec: &task.Spec,
+			},
+		}},
+		dagTasks: []v1beta1.PipelineTask{pts[10]},
+		expectedStatus: map[string]string{
+			PipelineTaskStatusPrefix + pts[10].Name + PipelineTaskStatusSuffix: PipelineTaskStateNone,
+		},
+	}, {
+		name: "when-expression-task-with-parent-started",
+		state: PipelineRunState{{
+			PipelineTask: &pts[0],
+			TaskRun:      makeStarted(trs[0]),
+			ResolvedTaskResources: &resources.ResolvedTaskResources{
+				TaskSpec: &task.Spec,
+			},
+		}, {
+			PipelineTask: &pts[11],
+			TaskRun:      nil,
+			ResolvedTaskResources: &resources.ResolvedTaskResources{
+				TaskSpec: &task.Spec,
+			},
+		}},
+		dagTasks: []v1beta1.PipelineTask{pts[0], pts[11]},
+		expectedStatus: map[string]string{
+			PipelineTaskStatusPrefix + pts[0].Name + PipelineTaskStatusSuffix:  PipelineTaskStateNone,
+			PipelineTaskStatusPrefix + pts[11].Name + PipelineTaskStatusSuffix: PipelineTaskStateNone,
+		},
+	}, {
+		name:     "task-cancelled",
+		state:    taskCancelled,
+		dagTasks: []v1beta1.PipelineTask{pts[4]},
+		expectedStatus: map[string]string{
+			PipelineTaskStatusPrefix + pts[4].Name + PipelineTaskStatusSuffix: PipelineTaskStateNone,
+		},
+	}}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			d, err := dag.Build(v1beta1.PipelineTaskList(tc.dagTasks), v1beta1.PipelineTaskList(tc.dagTasks).Deps())
+			if err != nil {
+				t.Fatalf("Unexpected error while buildig graph for DAG tasks %v: %v", tc.dagTasks, err)
+			}
+			facts := PipelineRunFacts{
+				State:           tc.state,
+				TasksGraph:      d,
+				FinalTasksGraph: &dag.Graph{},
+			}
+			s := facts.GetPipelineTaskStatus(context.Background())
+			if d := cmp.Diff(tc.expectedStatus, s); d != "" {
+				t.Fatalf("Test failed: %s Mismatch in pipelineTask execution state %s", tc.name, diff.PrintWantGot(d))
+			}
+		})
+	}
+}
diff --git a/pkg/substitution/substitution.go b/pkg/substitution/substitution.go
index feddb442a49..d63e01d9136 100644
--- a/pkg/substitution/substitution.go
+++ b/pkg/substitution/substitution.go
@@ -60,6 +60,22 @@ func ValidateVariableP(value, prefix string, vars sets.String) *apis.FieldError
 	return nil
 }
 
+func ValidateVariablePS(value, prefix string, suffix string, vars sets.String) *apis.FieldError {
+	if vs, present := extractVariablesFromString(value, prefix); present {
+		for _, v := range vs {
+			v = strings.TrimSuffix(v, suffix)
+			if !vars.Has(v) {
+				return &apis.FieldError{
+					Message: fmt.Sprintf("non-existent variable in %q", value),
+					// Empty path is required to make the `ViaField`, … work
+					Paths: []string{""},
+				}
+			}
+		}
+	}
+	return nil
+}
+
 // Verifies that variables matching the relevant string expressions do not reference any of the names present in vars.
 func ValidateVariableProhibited(name, value, prefix, locationName, path string, vars sets.String) *apis.FieldError {
 	if vs, present := extractVariablesFromString(value, prefix); present {
diff --git a/pkg/substitution/substitution_test.go b/pkg/substitution/substitution_test.go
index d3cd4627548..801de32222c 100644
--- a/pkg/substitution/substitution_test.go
+++ b/pkg/substitution/substitution_test.go
@@ -116,6 +116,39 @@ func TestValidateVariables(t *testing.T) {
 	}
 }
 
+func TestValidateVariablePS(t *testing.T) {
+	type args struct {
+		paramValue string
+		vars       sets.String
+	}
+	for _, tc := range []struct {
+		name          string
+		paramValue    string
+		vars          sets.String
+		expectedError *apis.FieldError
+	}{{
+		name:          "valid pipeline task in variable",
+		paramValue:    "--flag=$(tasks.task1.status)",
+		vars:          sets.NewString("task1"),
+		expectedError: nil,
+	}, {
+		name:       "undefined pipeline task",
+		paramValue: "--flag=$(tasks.task1.status)",
+		vars:       sets.NewString("foo"),
+		expectedError: &apis.FieldError{
+			Message: `non-existent variable in "--flag=$(tasks.task1.status)"`,
+			Paths:   []string{""},
+		},
+	}} {
+		t.Run(tc.name, func(t *testing.T) {
+			got := substitution.ValidateVariablePS(tc.paramValue, "tasks", "status", tc.vars)
+			if d := cmp.Diff(got, tc.expectedError, cmp.AllowUnexported(apis.FieldError{})); d != "" {
+				t.Errorf("ValidateVariablePS() error did not match expected error for %s: %s", tc.name, diff.PrintWantGot(d))
+			}
+		})
+	}
+}
+
 func TestApplyReplacements(t *testing.T) {
 	type args struct {
 		input        string
diff --git a/test/pipelinefinally_test.go b/test/pipelinefinally_test.go
index 40f0274b8b6..7cd2549d479 100644
--- a/test/pipelinefinally_test.go
+++ b/test/pipelinefinally_test.go
@@ -60,6 +60,11 @@ func TestPipelineLevelFinally_OneDAGTaskFailed_Failure(t *testing.T) {
 		t.Fatalf("Failed to create final Task: %s", err)
 	}
 
+	finalTaskWithStatus := getTaskVerifyingStatus(t, namespace)
+	if _, err := c.TaskClient.Create(ctx, finalTaskWithStatus, metav1.CreateOptions{}); err != nil {
+		t.Fatalf("Failed to create final Task checking executing status: %s", err)
+	}
+
 	pipeline := getPipeline(t,
 		namespace,
 		map[string]string{
@@ -72,6 +77,28 @@ func TestPipelineLevelFinally_OneDAGTaskFailed_Failure(t *testing.T) {
 		},
 		map[string]string{
 			"finaltask1": finalTask.Name,
+			"finaltask2": finalTaskWithStatus.Name,
+		},
+		map[string][]v1beta1.Param{
+			"finaltask2": {{
+				Name: "dagtask1-status",
+				Value: v1beta1.ArrayOrString{
+					Type:      "string",
+					StringVal: "$(tasks.dagtask1.status)",
+				},
+			}, {
+				Name: "dagtask2-status",
+				Value: v1beta1.ArrayOrString{
+					Type:      "string",
+					StringVal: "$(tasks.dagtask2.status)",
+				},
+			}, {
+				Name: "dagtask3-status",
+				Value: v1beta1.ArrayOrString{
+					Type:      "string",
+					StringVal: "$(tasks.dagtask3.status)",
+				},
+			}},
 		},
 	)
 	if _, err := c.PipelineClient.Create(ctx, pipeline, metav1.CreateOptions{}); err != nil {
@@ -115,6 +142,27 @@ func TestPipelineLevelFinally_OneDAGTaskFailed_Failure(t *testing.T) {
 				t.Errorf("Error waiting for TaskRun to succeed: %v", err)
 			}
 			finalTaskStartTime = taskrunItem.Status.StartTime
+		case n == "finaltask2":
+			if err := WaitForTaskRunState(ctx, c, taskrunItem.Name, TaskRunSucceed(taskrunItem.Name), "TaskRunSuccess"); err != nil {
+				t.Errorf("Error waiting for TaskRun to succeed: %v", err)
+			}
+			for _, p := range taskrunItem.Spec.Params {
+				switch param := p.Name; param {
+				case "dagtask1-status":
+					if p.Value.StringVal != v1beta1.TaskRunReasonFailed.String() {
+						t.Errorf("Task param \"%s\" is set to \"%s\", expected it to resolve to \"%s\"", param, p.Value.StringVal, v1beta1.TaskRunReasonFailed.String())
+					}
+				case "dagtask2-status":
+					if p.Value.StringVal != v1beta1.TaskRunReasonSuccessful.String() {
+						t.Errorf("Task param \"%s\" is set to \"%s\", expected it to resolve to \"%s\"", param, p.Value.StringVal, v1beta1.TaskRunReasonSuccessful.String())
+					}
+
+				case "dagtask3-status":
+					if p.Value.StringVal != resources.PipelineTaskStateNone {
+						t.Errorf("Task param \"%s\" is set to \"%s\", expected it to resolve to \"%s\"", param, p.Value.StringVal, resources.PipelineTaskStateNone)
+					}
+				}
+			}
 		default:
 			t.Fatalf("TaskRuns were not found for both final and dag tasks")
 		}
@@ -152,6 +200,7 @@ func TestPipelineLevelFinally_OneFinalTaskFailed_Failure(t *testing.T) {
 		map[string]string{
 			"finaltask1": finalTask.Name,
 		},
+		map[string][]v1beta1.Param{},
 	)
 	if _, err := c.PipelineClient.Create(ctx, pipeline, metav1.CreateOptions{}); err != nil {
 		t.Fatalf("Failed to create Pipeline: %s", err)
@@ -215,7 +264,7 @@ func isSkipped(t *testing.T, taskRunName string, conds duckv1beta1.Conditions) b
 	return false
 }
 
-func getTaskDef(n, namespace, script string) *v1beta1.Task {
+func getTaskDef(n, namespace, script string, params []v1beta1.ParamSpec) *v1beta1.Task {
 	return &v1beta1.Task{
 		ObjectMeta: metav1.ObjectMeta{Name: n, Namespace: namespace},
 		Spec: v1beta1.TaskSpec{
@@ -223,20 +272,33 @@ func getTaskDef(n, namespace, script string) *v1beta1.Task {
 				Container: corev1.Container{Image: "alpine"},
 				Script:    script,
 			}},
+			Params: params,
 		},
 	}
 }
 
 func getSuccessTask(t *testing.T, namespace string) *v1beta1.Task {
-	return getTaskDef(helpers.ObjectNameForTest(t), namespace, "exit 0")
+	return getTaskDef(helpers.ObjectNameForTest(t), namespace, "exit 0", []v1beta1.ParamSpec{})
 }
 
 func getFailTask(t *testing.T, namespace string) *v1beta1.Task {
-	return getTaskDef(helpers.ObjectNameForTest(t), namespace, "exit 1")
+	return getTaskDef(helpers.ObjectNameForTest(t), namespace, "exit 1", []v1beta1.ParamSpec{})
 }
 
 func getDelaySuccessTask(t *testing.T, namespace string) *v1beta1.Task {
-	return getTaskDef(helpers.ObjectNameForTest(t), namespace, "sleep 5; exit 0")
+	return getTaskDef(helpers.ObjectNameForTest(t), namespace, "sleep 5; exit 0", []v1beta1.ParamSpec{})
+}
+
+func getTaskVerifyingStatus(t *testing.T, namespace string) *v1beta1.Task {
+	params := []v1beta1.ParamSpec{{
+		Name: "dagtask1-status",
+	}, {
+		Name: "dagtask2-status",
+	}, {
+		Name: "dagtask3-status",
+	}}
+	script := "exit 0"
+	return getTaskDef(helpers.ObjectNameForTest(t), namespace, script, params)
 }
 
 func getCondition(t *testing.T, namespace string) *v1alpha1.Condition {
@@ -251,7 +313,7 @@ func getCondition(t *testing.T, namespace string) *v1alpha1.Condition {
 	}
 }
 
-func getPipeline(t *testing.T, namespace string, ts map[string]string, c map[string]string, f map[string]string) *v1beta1.Pipeline {
+func getPipeline(t *testing.T, namespace string, ts map[string]string, c map[string]string, f map[string]string, p map[string][]v1beta1.Param) *v1beta1.Pipeline {
 	var pt []v1beta1.PipelineTask
 	var fpt []v1beta1.PipelineTask
 	for k, v := range ts {
@@ -270,6 +332,7 @@ func getPipeline(t *testing.T, namespace string, ts map[string]string, c map[str
 		fpt = append(fpt, v1beta1.PipelineTask{
 			Name:    k,
 			TaskRef: &v1beta1.TaskRef{Name: v},
+			Params:  p[k],
 		})
 	}
 	pipeline := &v1beta1.Pipeline{