Make batch workloads first-class

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/k8s/workload.go

@@ -1,6 +1,7 @@
 package k8s
 
 import (
+	"context"
 	"fmt"
 
 	corev1 "k8s.io/api/core/v1"
@@ -45,6 +46,28 @@ 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 nil, fmt.Errorf("job %s/%s has no pod selector", namespace, name)
+		}
+		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) {
 	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) {

internal/mcp/tools_workloads.go

@@ -12,7 +12,9 @@ import (
 	"time"
 
 	"github.com/modelcontextprotocol/go-sdk/mcp"
+	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 
 	"github.com/skyhook-io/radar/internal/k8s"
 	aicontext "github.com/skyhook-io/radar/pkg/ai/context"
@@ -36,7 +38,7 @@ type manageCronJobInput struct {
 }
 
 type getWorkloadLogsInput struct {
-	Kind      string `json:"kind,omitempty" jsonschema:"workload kind: deployment, statefulset, or daemonset. Defaults to deployment when omitted."`
+	Kind      string `json:"kind,omitempty" jsonschema:"workload kind: deployment, statefulset, daemonset, job, or workflow. Defaults to deployment when omitted."`
 	Namespace string `json:"namespace" jsonschema:"workload namespace"`
 	Name      string `json:"name" jsonschema:"workload name"`
 	Container string `json:"container,omitempty" jsonschema:"specific container name, defaults to all containers"`
@@ -175,7 +177,7 @@ func handleManageCronJob(ctx context.Context, req *mcp.CallToolRequest, input ma
 func handleGetWorkloadLogs(ctx context.Context, req *mcp.CallToolRequest, input getWorkloadLogsInput) (*mcp.CallToolResult, any, error) {
 	kind := normalizeWorkloadLogsKind(input.Kind)
 	if kind == "" {
-		return nil, nil, fmt.Errorf("invalid kind %q: must be deployment, statefulset, or daemonset", input.Kind)
+		return nil, nil, fmt.Errorf("invalid kind %q: must be deployment, statefulset, daemonset, job, or workflow", input.Kind)
 	}
 
 	if !checkNamespaceAccess(ctx, input.Namespace) {
@@ -201,11 +203,14 @@ func handleGetWorkloadLogs(ctx context.Context, req *mcp.CallToolRequest, input
 	// Get pods matching the workload
 	pods := cache.GetPodsForWorkload(input.Namespace, selector)
 	if len(pods) == 0 {
-		return toJSONResult(map[string]any{
+		empty := describeMCPWorkloadLogEmpty(ctx, kind, input.Namespace, input.Name)
+		response := map[string]any{
 			"workload": fmt.Sprintf("%s/%s/%s", kind, input.Namespace, input.Name),
 			"pods":     0,
-			"logs":     "no pods found for this workload",
-		})
+			"logs":     empty.Message,
+		}
+		addMCPWorkloadLogEmptyMetadata(response, empty)
+		return toJSONResult(response)
 	}
 
 	tailLines := int64(100)
@@ -277,6 +282,135 @@ func handleGetWorkloadLogs(ctx context.Context, req *mcp.CallToolRequest, input
 	return toJSONResult(resp)
 }
 
+type mcpWorkloadLogEmptyMetadata struct {
+	Reason  string
+	Message string
+	Command string
+}
+
+func describeMCPWorkloadLogEmpty(ctx context.Context, kind, namespace, name string) mcpWorkloadLogEmptyMetadata {
+	switch kind {
+	case "jobs":
+		return describeMCPJobLogEmpty(namespace, name)
+	case "workflows":
+		return describeMCPWorkflowLogEmpty(ctx, namespace, name)
+	default:
+		return mcpWorkloadLogEmptyMetadata{
+			Reason:  "no-pods",
+			Message: "no pods found for this workload",
+		}
+	}
+}
+
+func describeMCPJobLogEmpty(namespace, name string) mcpWorkloadLogEmptyMetadata {
+	metadata := mcpWorkloadLogEmptyMetadata{
+		Reason:  "no-pods",
+		Message: "No pods found for this Job yet. Check scheduling, admission, or controller events.",
+		Command: "kubectl logs job/" + name + " -n " + namespace,
+	}
+	cache := k8s.GetResourceCache()
+	if cache == nil || cache.Jobs() == nil {
+		return metadata
+	}
+	job, err := cache.Jobs().Jobs(namespace).Get(name)
+	if err != nil {
+		return metadata
+	}
+	applyMCPTerminalJobEmptyState(&metadata, job, namespace, name)
+	return metadata
+}
+
+func applyMCPTerminalJobEmptyState(metadata *mcpWorkloadLogEmptyMetadata, job *batchv1.Job, namespace, name string) {
+	if _, complete := mcpJobCondition(job, batchv1.JobComplete); !complete {
+		if _, failed := mcpJobCondition(job, batchv1.JobFailed); !failed {
+			return
+		}
+	}
+	metadata.Reason = "pods-gone"
+	metadata.Message = "This Job has finished, but its pods are no longer present in Kubernetes. If logs were retained externally, use your logging system or try kubectl logs job/" + name + " -n " + namespace + "."
+}
+
+func mcpJobCondition(job *batchv1.Job, conditionType batchv1.JobConditionType) (batchv1.JobCondition, bool) {
+	for _, condition := range job.Status.Conditions {
+		if condition.Type == conditionType && condition.Status == corev1.ConditionTrue {
+			return condition, true
+		}
+	}
+	return batchv1.JobCondition{}, false
+}
+
+func describeMCPWorkflowLogEmpty(ctx context.Context, namespace, name string) mcpWorkloadLogEmptyMetadata {
+	metadata := mcpWorkloadLogEmptyMetadata{
+		Reason:  "no-pods",
+		Message: "No Workflow pods found yet. Check scheduling, admission, or controller events.",
+		Command: "argo logs " + name + " -n " + namespace,
+	}
+	cache := k8s.GetResourceCache()
+	if cache == nil {
+		return metadata
+	}
+	workflow, err := cache.GetDynamicWithGroup(ctx, "Workflow", namespace, name, "argoproj.io")
+	if err != nil {
+		return metadata
+	}
+	applyMCPTerminalWorkflowEmptyState(&metadata, workflow.Object, namespace, name)
+	return metadata
+}
+
+func applyMCPTerminalWorkflowEmptyState(metadata *mcpWorkloadLogEmptyMetadata, workflow map[string]any, namespace, name string) {
+	phase, _, _ := unstructured.NestedString(workflow, "status", "phase")
+	if !mcpWorkflowTerminal(phase) {
+		return
+	}
+	metadata.Reason = "pods-gone"
+	if mcpWorkflowArchiveLogsConfigured(workflow) {
+		metadata.Message = "This Workflow has finished and its pods are no longer present. Archived logs appear to be enabled; use the configured Argo or logging UI, or try argo logs " + name + " -n " + namespace + "."
+	} else {
+		metadata.Message = "This Workflow has finished and its pods are no longer present. Argo may have garbage-collected them; Kubernetes pod logs are no longer available here."
+	}
+}
+
+func mcpWorkflowTerminal(phase string) bool {
+	switch phase {
+	case "Succeeded", "Failed", "Error":
+		return true
+	default:
+		return false
+	}
+}
+
+func mcpWorkflowArchiveLogsConfigured(obj map[string]any) bool {
+	if archiveLogs, ok, _ := unstructured.NestedBool(obj, "spec", "archiveLogs"); ok && archiveLogs {
+		return true
+	}
+	templates, ok, _ := unstructured.NestedSlice(obj, "spec", "templates")
+	if !ok {
+		return false
+	}
+	for _, template := range templates {
+		templateMap, ok := template.(map[string]any)
+		if !ok {
+			continue
+		}
+		if archiveLogs, ok, _ := unstructured.NestedBool(templateMap, "archiveLocation", "archiveLogs"); ok && archiveLogs {
+			return true
+		}
+	}
+	return false
+}
+
+func addMCPWorkloadLogEmptyMetadata(response map[string]any, metadata mcpWorkloadLogEmptyMetadata) {
+	if metadata.Reason != "" {
+		response["emptyReason"] = metadata.Reason
+	}
+	if metadata.Message != "" {
+		response["emptyMessage"] = metadata.Message
+	}
+	if metadata.Command != "" {
+		response["command"] = metadata.Command
+	}
+}
+
 // schedulingBlockerWarnings detects when a restart won't accomplish what the
 // agent likely wants: if the workload currently has Pending pods blocked on
 // scheduling or post-bind (CNI/volume) issues, a rolling restart just creates
@@ -568,5 +702,12 @@ func normalizeWorkloadLogsKind(kind string) string {
 	if strings.TrimSpace(kind) == "" {
 		return "deployments"
 	}
-	return normalizeWorkloadKind(kind)
+	switch strings.ToLower(kind) {
+	case "job", "jobs":
+		return "jobs"
+	case "workflow", "workflows":
+		return "workflows"
+	default:
+		return normalizeWorkloadKind(kind)
+	}
 }

internal/mcp/tools_workloads_test.go

@@ -0,0 +1,84 @@
+package mcp
+
+import (
+	"strings"
+	"testing"
+
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestApplyMCPTerminalJobEmptyState(t *testing.T) {
+	metadata := mcpWorkloadLogEmptyMetadata{
+		Reason:  "no-pods",
+		Message: "No pods found for this Job yet.",
+		Command: "kubectl logs job/nightly -n ci",
+	}
+	job := &batchv1.Job{
+		Status: batchv1.JobStatus{
+			Conditions: []batchv1.JobCondition{
+				{
+					Type:   batchv1.JobComplete,
+					Status: corev1.ConditionTrue,
+				},
+			},
+		},
+	}
+
+	applyMCPTerminalJobEmptyState(&metadata, job, "ci", "nightly")
+
+	if metadata.Reason != "pods-gone" {
+		t.Fatalf("Reason = %q, want pods-gone", metadata.Reason)
+	}
+	if !strings.Contains(metadata.Message, "finished") || !strings.Contains(metadata.Message, "kubectl logs job/nightly -n ci") {
+		t.Fatalf("Message = %q, want terminal kubectl guidance", metadata.Message)
+	}
+	if metadata.Command != "kubectl logs job/nightly -n ci" {
+		t.Fatalf("Command = %q", metadata.Command)
+	}
+}
+
+func TestApplyMCPTerminalWorkflowEmptyStateWithArchiveLogs(t *testing.T) {
+	metadata := mcpWorkloadLogEmptyMetadata{
+		Reason:  "no-pods",
+		Message: "No Workflow pods found yet.",
+		Command: "argo logs nightly -n ci",
+	}
+	workflow := map[string]any{
+		"status": map[string]any{
+			"phase": "Succeeded",
+		},
+		"spec": map[string]any{
+			"archiveLogs": true,
+		},
+	}
+
+	applyMCPTerminalWorkflowEmptyState(&metadata, workflow, "ci", "nightly")
+
+	if metadata.Reason != "pods-gone" {
+		t.Fatalf("Reason = %q, want pods-gone", metadata.Reason)
+	}
+	if !strings.Contains(metadata.Message, "Archived logs") || !strings.Contains(metadata.Message, "argo logs nightly -n ci") {
+		t.Fatalf("Message = %q, want archived-log guidance", metadata.Message)
+	}
+	if metadata.Command != "argo logs nightly -n ci" {
+		t.Fatalf("Command = %q", metadata.Command)
+	}
+}
+
+func TestMCPJobConditionIgnoresRetryCounters(t *testing.T) {
+	job := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "retrying",
+			Namespace: "ci",
+		},
+		Status: batchv1.JobStatus{
+			Failed: 1,
+		},
+	}
+
+	if _, ok := mcpJobCondition(job, batchv1.JobFailed); ok {
+		t.Fatal("mcpJobCondition treated failed counter as terminal condition")
+	}
+}