Make batch workloads first-class

Author: nadaverell

NOTES.md

@@ -0,0 +1,8 @@
+# Autodev Notes
+
+## 2026-07-10 - batch workload rebase over #1106
+
+- Assumption: PR #1106's redesigned application/workload scope model is the base design; batch support should extend the new Overview/Topology/History and workload-scope surfaces instead of restoring the older rail/top-band layout.
+- Decision: Keep operational Issues in the app Overview by embedding the existing Issues row renderer scoped to app workloads, not by creating a separate incident-summary component.
+- Decision: Preserve pod log health metadata from main and add Argo workflow step metadata as separate fields (`stepID`, `stepName`, `stepPhase`) so workflow context does not overwrite Kubernetes pod phase.
+- Decision: Add only additive shared UI props for batch Overview overrides and app Overview issue slots; do not remove or rename existing `@skyhook-io/k8s-ui` exports.

docs/mcp.md

@@ -194,7 +194,7 @@ Add to `~/.gemini/settings.json`:
 | `get_events` | Recent Kubernetes Warning events, deduplicated and sorted by recency. Filter by resource kind/name to scope. | `namespace` (optional), `limit` (optional, default 20, max 100), `kind` (optional), `name` (optional) |
 | `get_changes` | Recent meaningful changes from the Kubernetes cluster timeline plus native Helm release deployment/operation history (`source: helm`). Use to investigate what changed before an incident, including failed upgrades, rollbacks, and current Helm revisions. If the response includes `sourcesErrored`, treat it as partial data for those sources. Use `get_helm_release include=history,operations` for the full Helm revision trail. | `namespace` (optional), `kind` (optional), `name` (optional), `since` (optional, e.g. `1h`, `30m`; default `1h`), `limit` (optional, default 20, max 50) |
 | `get_pod_logs` | Filtered pod logs prioritizing errors/warnings, with secret redaction. Set `grep` for server-side filtering. | `namespace` (required), `name` (required), `container` (optional), `tail_lines` (optional, default 200), `grep` (optional) |
-| `get_workload_logs` | Aggregated, AI-filtered logs from all pods of a workload (Deployment, StatefulSet, DaemonSet) | `kind` (required), `namespace` (required), `name` (required), `container` (optional), `tail_lines` (optional, default 100 per pod), `grep` (optional) |
+| `get_workload_logs` | Aggregated, AI-filtered logs from all pods of a workload (Deployment, StatefulSet, DaemonSet, Job, Argo Workflow) | `kind` (required), `namespace` (required), `name` (required), `container` (optional), `tail_lines` (optional, default 100 per pod), `grep` (optional) |
 | `get_cluster_audit` | Static config posture — best-practice findings (Security / Reliability / Efficiency) with remediation. INDEPENDENT of operational health; for "what's broken right now?" use `issues`. | `namespace` (optional), `category` (optional), `severity` (optional) |
 | `list_packages` | Installed packages (Helm releases, label-managed workloads, CRDs, Argo Applications, Flux HelmReleases + Kustomizations) with source provenance, versions, and health, in one call. Response includes `sourceLegend` for the stable source codes. | `namespace` (optional), `source` (optional: `H`/`helm`, `L`/`labels`, `C`/`crds`, `A`/`argocd`, `F`/`fluxcd`), `chart` (optional substring) |
 | `list_helm_releases` | List Helm releases with status, resource health, storage namespace, Flux ownership, current `lastOperation`, and a capped `operations` trail when Helm history indicates failed upgrades, rollback-after-failure, rollbacks, or stuck pending operations. Use this first for Helm deployment debugging. | `namespace` (optional) |

internal/issues/category.go

@@ -13,6 +13,7 @@ type classifyInput struct {
 	APIGroup             string // the resource's API group (Issue.Group)
 	Kind                 string
 	Reason               string
+	Message              string
 	LastTerminatedReason string
 }
 
@@ -145,6 +146,9 @@ func classifyProblem(in classifyInput) issuesapi.Category {
 	case "Service port mismatch":
 		return issuesapi.CategoryMissingConfigRef
 	}
+	if isForbiddenMessage(in.Message) {
+		return issuesapi.CategoryRBACForbidden
+	}
 	switch in.Kind {
 	case "Pod":
 		if in.Reason == "OOMKilled" {
@@ -259,6 +263,12 @@ func classifyProblem(in classifyInput) issuesapi.Category {
 		}
 		return issuesapi.CategoryUnknown
 
+	case "Workflow", "CronWorkflow":
+		if strings.Contains(strings.ToLower(in.APIGroup), "argoproj.io") && isForbiddenMessage(in.Message) {
+			return issuesapi.CategoryRBACForbidden
+		}
+		return issuesapi.CategoryUnknown
+
 	case "Application":
 		// ArgoCD Application health/sync failure from DetectGitOpsProblems.
 		// Gate on group so a same-named CRD from another controller can't be
@@ -300,6 +310,11 @@ func classifyProblem(in classifyInput) issuesapi.Category {
 	return issuesapi.CategoryUnknown
 }
 
+func isForbiddenMessage(message string) bool {
+	m := strings.ToLower(message)
+	return strings.Contains(m, " is forbidden: user ") && strings.Contains(m, " cannot ")
+}
+
 // classifyGitOpsReason maps a GitOps detector/condition reason to a specific
 // GitOps failure category. Shared by the Argo + Flux arms of both Classify
 // (conditions) and classifyProblem (DetectGitOpsProblems) so the two paths

internal/issues/category_test.go

@@ -40,6 +40,7 @@ func TestClassify(t *testing.T) {
 		{"config error waiting", classifyInput{Source: SourceProblem, Kind: "Pod", Reason: "CreateContainerConfigError"}, issuesapi.CategoryContainerWaiting},
 		{"pending pod (non-scheduling)", classifyInput{Source: SourceProblem, Kind: "Pod", Reason: "Pending"}, issuesapi.CategoryContainerWaiting},
 		{"errored pod", classifyInput{Source: SourceProblem, Kind: "Pod", Reason: "Error"}, issuesapi.CategoryCrashLoop},
+		{"pod status rbac forbidden message", classifyInput{Source: SourceProblem, Kind: "Pod", Reason: "Completed", Message: `workflowtaskresults.argoproj.io is forbidden: User "system:serviceaccount:radar-batch-visual:default" cannot create resource "workflowtaskresults" in API group "argoproj.io" in the namespace "radar-batch-visual"`}, issuesapi.CategoryRBACForbidden},
 		{"high restart thrash", classifyInput{Source: SourceProblem, Kind: "Pod", Reason: "HighRestartCount"}, issuesapi.CategoryHighRestart},
 		{"liveness probe failed", classifyInput{Source: SourceProblem, Kind: "Pod", Reason: "LivenessProbeFailed"}, issuesapi.CategoryLivenessProbeFail},
 		{"liveness probe beats stale oom", classifyInput{Source: SourceProblem, Kind: "Pod", Reason: "LivenessProbeFailed", LastTerminatedReason: "OOMKilled"}, issuesapi.CategoryLivenessProbeFail},
@@ -100,6 +101,8 @@ func TestClassify(t *testing.T) {
 		{"job stuck active", classifyInput{Source: SourceProblem, Kind: "Job", Reason: "Running for 3h with no completions"}, issuesapi.CategoryJobFailed},
 		{"cronjob stale", classifyInput{Source: SourceProblem, Kind: "CronJob", Reason: "stale"}, issuesapi.CategoryCronJobFailed},
 		{"cronjob never scheduled", classifyInput{Source: SourceProblem, Kind: "CronJob", Reason: "never-scheduled"}, issuesapi.CategoryCronJobFailed},
+		{"argo workflow rbac forbidden", classifyInput{Source: SourceProblem, Kind: "Workflow", APIGroup: "argoproj.io", Reason: "Completed", Message: `workflowtaskresults.argoproj.io is forbidden: User "system:serviceaccount:radar-batch-visual:default" cannot create resource "workflowtaskresults" in API group "argoproj.io" in the namespace "radar-batch-visual"`}, issuesapi.CategoryRBACForbidden},
+		{"argo workflow completed without forbidden stays unknown", classifyInput{Source: SourceProblem, Kind: "Workflow", APIGroup: "argoproj.io", Reason: "Completed", Message: "child 'example' failed"}, issuesapi.CategoryUnknown},
 
 		// missing_ref
 		{"missing configmap", classifyInput{Source: SourceMissingRef, Kind: "Pod", Reason: "Missing ConfigMap"}, issuesapi.CategoryMissingConfigRef},

internal/issues/issues_test.go

@@ -141,6 +141,35 @@ func TestCompose_PopulatesCategoryAndGroup(t *testing.T) {
 	}
 }
 
+func TestCompose_ClassifiesArgoWorkflowForbiddenMessage(t *testing.T) {
+	p := &fakeProvider{
+		problems: []k8s.Detection{{
+			Kind:       "Pod",
+			Namespace:  "ns",
+			Name:       "daily-import-step-123",
+			Severity:   "critical",
+			Reason:     "Completed",
+			Message:    `workflowtaskresults.argoproj.io is forbidden: User "system:serviceaccount:ns:default" cannot create resource "workflowtaskresults" in API group "argoproj.io" in the namespace "ns"`,
+			OwnerKind:  "Workflow",
+			OwnerName:  "daily-import",
+			OwnerGroup: "argoproj.io",
+		}},
+	}
+	out := Compose(p, Filters{Grouped: true})
+	if len(out) != 1 {
+		t.Fatalf("got %d issues: %+v", len(out), out)
+	}
+	if out[0].Kind != "Workflow" || out[0].Name != "daily-import" || out[0].Group != "argoproj.io" {
+		t.Fatalf("subject = %s/%s/%s, want argoproj.io/Workflow/daily-import", out[0].Group, out[0].Kind, out[0].Name)
+	}
+	if out[0].Reason != "RBACForbidden" {
+		t.Fatalf("reason = %q, want RBACForbidden", out[0].Reason)
+	}
+	if out[0].Category != issuesapi.CategoryRBACForbidden || out[0].CategoryGroup != issuesapi.GroupSecurity {
+		t.Fatalf("category = %q/%q, want %q/%q", out[0].Category, out[0].CategoryGroup, issuesapi.CategoryRBACForbidden, issuesapi.GroupSecurity)
+	}
+}
+
 func TestCompose_GroupsMemberPodsUnderOwner(t *testing.T) {
 	// Two pods of the same Deployment failing the same way share one issue
 	// ID (the future collapse target); a third pod failing differently gets
@@ -2020,8 +2049,8 @@ func TestSymptomNamesSecret(t *testing.T) {
 	}{
 		{`references Secret "foo" which does not exist`, "prod", true},
 		{`MountVolume.SetUp failed: secret "foo" not found`, "prod", true},
-		{`secrets "foo" not found`, "prod", true},                  // plural kubelet path
-		{`couldn't find key tls.crt in Secret prod/foo`, "prod", true}, // namespaced missing-key
+		{`secrets "foo" not found`, "prod", true},                          // plural kubelet path
+		{`couldn't find key tls.crt in Secret prod/foo`, "prod", true},     // namespaced missing-key
 		{`references ConfigMap "foo" which does not exist`, "prod", false}, // same name, wrong kind
 		{`waiting on something else`, "prod", false},
 	}

internal/issues/normalize.go

@@ -48,14 +48,18 @@ func fromProblem(p k8s.Detection, now time.Time, source Source) Issue {
 		// issue looking fresh. AgeSeconds (resource age) is a stable lower bound.
 		since = now.Add(-time.Duration(p.AgeSeconds) * time.Second)
 	}
+	reason := p.Reason
+	if isForbiddenMessage(p.Message) {
+		reason = "RBACForbidden"
+	}
 	iss := Issue{
 		Severity:             sev,
 		Source:               source,
 		Kind:                 p.Kind,
 		Group:                resolveGroup(p.Group, p.Kind),
 		Namespace:            p.Namespace,
 		Name:                 p.Name,
-		Reason:               p.Reason,
+		Reason:               reason,
 		Message:              p.Message,
 		RawMessage:           p.RawMessage,
 		Cause:                p.Cause,
@@ -103,6 +107,7 @@ func classifyIssue(i *Issue) {
 		APIGroup:             i.Group,
 		Kind:                 i.Kind,
 		Reason:               i.Reason,
+		Message:              i.Message,
 		LastTerminatedReason: i.LastTerminatedReason,
 	})
 	i.CategoryGroup = issuesapi.GroupOf(i.Category)

internal/k8s/dynamic_cache.go

@@ -162,6 +162,8 @@ var supportedCRDFallbacks = []supportedCRDResource{
 	{Group: "argoproj.io", Versions: []string{"v1alpha1"}, Resource: "rollouts", Kind: "Rollout", Namespaced: true},
 	{Group: "argoproj.io", Versions: []string{"v1alpha1"}, Resource: "workflows", Kind: "Workflow", Namespaced: true},
 	{Group: "argoproj.io", Versions: []string{"v1alpha1"}, Resource: "cronworkflows", Kind: "CronWorkflow", Namespaced: true},
+	{Group: "argoproj.io", Versions: []string{"v1alpha1"}, Resource: "workflowtemplates", Kind: "WorkflowTemplate", Namespaced: true},
+	{Group: "argoproj.io", Versions: []string{"v1alpha1"}, Resource: "clusterworkflowtemplates", Kind: "ClusterWorkflowTemplate", Namespaced: false},
 	{Group: "cert-manager.io", Versions: []string{"v1"}, Resource: "certificates", Kind: "Certificate", Namespaced: true},
 	{Group: "cert-manager.io", Versions: []string{"v1"}, Resource: "issuers", Kind: "Issuer", Namespaced: true},
 	{Group: "cert-manager.io", Versions: []string{"v1"}, Resource: "clusterissuers", Kind: "ClusterIssuer", Namespaced: false},

internal/k8s/workload.go

@@ -1,6 +1,7 @@
 package k8s
 
 import (
+	"context"
 	"fmt"
 
 	corev1 "k8s.io/api/core/v1"
@@ -45,6 +46,30 @@ func GetWorkloadSelector(cache *ResourceCache, kind, namespace, name string) (*m
 		}
 		return ds.Spec.Selector, nil
 
+	case "job", "jobs":
+		lister := cache.Jobs()
+		if lister == nil {
+			return nil, fmt.Errorf("insufficient permissions to list jobs")
+		}
+		job, err := lister.Jobs(namespace).Get(name)
+		if err != nil {
+			return nil, fmt.Errorf("job %s/%s not found: %w", namespace, name, err)
+		}
+		if job.Spec.Selector == nil {
+			return &metav1.LabelSelector{
+				MatchLabels: map[string]string{"batch.kubernetes.io/job-name": name},
+			}, nil
+		}
+		return job.Spec.Selector, nil
+
+	case "workflow", "workflows":
+		if _, err := cache.GetDynamicWithGroup(context.Background(), "Workflow", namespace, name, "argoproj.io"); err != nil {
+			return nil, fmt.Errorf("workflow %s/%s not found: %w", namespace, name, err)
+		}
+		return &metav1.LabelSelector{
+			MatchLabels: map[string]string{"workflows.argoproj.io/workflow": name},
+		}, nil
+
 	default:
 		return nil, fmt.Errorf("unsupported workload kind: %s", kind)
 	}

internal/mcp/tools.go

@@ -450,7 +450,7 @@ func registerTools(server *mcp.Server, includeWrites bool) {
 	mcp.AddTool(server, &mcp.Tool{
 		Name: "get_workload_logs",
 		Description: "Get aggregated logs from all pods of a workload (Deployment, StatefulSet, " +
-			"or DaemonSet). Logs are collected from all matching pods concurrently, then " +
+			"DaemonSet, Job, or Argo Workflow). Logs are collected from all matching pods concurrently, then " +
 			"server-side filtered to errors, warnings, panics, and stack traces using " +
 			"deterministic regex patterns and deduplicated. Set grep for additional " +
 			"server-side filtering before that summary stage, like `kubectl logs | grep PATTERN`. " +

internal/mcp/tools_filter_test.go

@@ -699,6 +699,12 @@ func TestNormalizeWorkloadLogsKind_DefaultsToDeployment(t *testing.T) {
 	if got := normalizeWorkloadLogsKind("statefulset"); got != "statefulsets" {
 		t.Fatalf("statefulset workload-log kind = %q, want statefulsets", got)
 	}
+	if got := normalizeWorkloadLogsKind("job"); got != "jobs" {
+		t.Fatalf("job workload-log kind = %q, want jobs", got)
+	}
+	if got := normalizeWorkloadLogsKind("Workflow"); got != "workflows" {
+		t.Fatalf("Workflow workload-log kind = %q, want workflows", got)
+	}
 }
 
 func TestParseLogsSince(t *testing.T) {