diff --git a/.changes/unreleased/Feature-20260717-000000.yaml b/.changes/unreleased/Feature-20260717-000000.yaml
new file mode 100644
index 0000000..cc64e75
--- /dev/null
+++ b/.changes/unreleased/Feature-20260717-000000.yaml
@@ -0,0 +1,3 @@
+kind: Feature
+body: Emit account/job metadata as Kubernetes pod labels (`opslevel.com/account-id`, `opslevel.com/job-id`, `opslevel.com/mode`, `opslevel.com/job-type`, plus a short-form `accountID`) so job pods can be filtered by account/job/mode without parsing the pod name
+time: 2026-07-17T00:00:00.000000Z
diff --git a/src/pkg/k8s.go b/src/pkg/k8s.go
index dbbf6aa..464c62e 100644
--- a/src/pkg/k8s.go
+++ b/src/pkg/k8s.go
@@ -111,6 +111,16 @@ func NewJobRunner(runnerId string, path string) *JobRunner {
 	}
 }
 
+// getRunnerJobVariable returns the value of the job variable with the given key, if present.
+func getRunnerJobVariable(configs []opslevel.RunnerJobVariable, key string) string {
+	for _, config := range configs {
+		if config.Key == key {
+			return config.Value
+		}
+	}
+	return ""
+}
+
 // getPodEnv returns the env vars to inject into a container for the given
 // scope. Variables with no Scope set are visible to every container; variables
 // with a Scope are only visible to containers running in that scope.
@@ -372,6 +382,36 @@ func (s *JobRunner) Run(ctx context.Context, job opslevel.RunnerJob, stdout, std
 			Outcome: opslevel.RunnerJobOutcomeEnumFailed,
 		}
 	}
+	// Descriptive labels for observability (e.g. DataDog podLabelsAsTags), so
+	// running pods can be filtered by account/job/mode without parsing the pod
+	// name. Added after building the selector so they stay out of the PDB
+	// selector, which the instance label already makes unique.
+	//
+	// Each label is stamped in both its prefixed long form (stable, ideal for
+	// DataDog podLabelsAsTags) and a short form for terse CLI filtering
+	// (e.g. `kubectl get pods -l accountID=XXX`).
+
+	// app.kubernetes.io/name is the well-known Kubernetes label for the name of
+	// the application; every job pod is an instance of "opslevel-job".
+	labels["app.kubernetes.io/name"] = "opslevel-job"
+	// mode is the runner backend the job came from: "api" or "faktory".
+	labels["opslevel.com/mode"] = viper.GetString("mode")
+	labels["mode"] = viper.GetString("mode")
+	// job-id is the OpsLevel job identifier (one value per job).
+	labels["opslevel.com/job-id"] = id
+	labels["jobID"] = id
+	// Variable keys arrive uppercased/sanitized by the OpsLevel API (see
+	// RunnerJobType#format_key_as_valid_env_var_name), so match the uppercase key.
+	if accountId := getRunnerJobVariable(job.Variables, "ACCOUNT_ID"); accountId != "" {
+		// account-id is the OpsLevel account the job belongs to.
+		labels["opslevel.com/account-id"] = accountId
+		labels["accountID"] = accountId
+	}
+	if jobType := getRunnerJobVariable(job.Variables, "JOB_TYPE"); jobType != "" {
+		// job-type is the job's kind / template slug (e.g. "repo_grep_v1").
+		labels["opslevel.com/job-type"] = jobType
+		labels["jobType"] = jobType
+	}
 	// TODO: manage pods based on image for re-use?
 	cfgMap, err := s.CreateConfigMap(ctx, s.getConfigMapObject(identifier, job))
 	if err != nil {
diff --git a/src/pkg/k8s_test.go b/src/pkg/k8s_test.go
index f7322e5..76d1338 100644
--- a/src/pkg/k8s_test.go
+++ b/src/pkg/k8s_test.go
@@ -193,6 +193,50 @@ func TestGetPodEnv_FiltersByScope(t *testing.T) {
 	autopilot.Equals(t, []string{"BOTH", "MAIN_ONLY"}, mainKeys)
 }
 
+func TestGetRunnerJobVariable_ReturnsMatchingValue(t *testing.T) {
+	// Arrange
+	// The OpsLevel API uppercases/sanitizes variable keys before sending them,
+	// so the runner matches the uppercase key.
+	vars := []opslevel.RunnerJobVariable{
+		{Key: "FOO", Value: "bar"},
+		{Key: "ACCOUNT_ID", Value: "acct-123"},
+	}
+
+	// Act
+	value := getRunnerJobVariable(vars, "ACCOUNT_ID")
+
+	// Assert
+	autopilot.Equals(t, "acct-123", value)
+}
+
+func TestGetRunnerJobVariable_ReturnsEmptyWhenMissing(t *testing.T) {
+	// Arrange
+	vars := []opslevel.RunnerJobVariable{
+		{Key: "FOO", Value: "bar"},
+	}
+
+	// Act
+	value := getRunnerJobVariable(vars, "ACCOUNT_ID")
+
+	// Assert
+	autopilot.Equals(t, "", value)
+}
+
+func TestGetRunnerJobVariable_IsCaseSensitive(t *testing.T) {
+	// Arrange
+	// Guards the bug where the runner looked up "account_id" but the API sends
+	// "ACCOUNT_ID"; the lowercase key must NOT match.
+	vars := []opslevel.RunnerJobVariable{
+		{Key: "ACCOUNT_ID", Value: "acct-123"},
+	}
+
+	// Act
+	value := getRunnerJobVariable(vars, "account_id")
+
+	// Assert
+	autopilot.Equals(t, "", value)
+}
+
 func TestGetPodObject_NoInitCommands(t *testing.T) {
 	// Arrange
 	runner := &JobRunner{