load(PR1): persist the interesting corpus as a campaign ConfigMap (DD-026)

deploy/e2e/e2e.sh

@@ -320,6 +320,18 @@ YAML
   echo "  (campaign coveragePct=${cpct:-<none>}; $(grep -c 'via corpusDir' "$dlog") value-file(s) resolved from corpus)"
   rm -f "$dlog"
 
+  # --- DD-026 PR 1: the run emits its interesting "replay corpus" as a campaign-owned ConfigMap ---
+  ccorpus="$($K -n "$NS" get closurejvmcampaign jpetstore-campaign -o jsonpath='{.status.corpusConfigMap}' 2>/dev/null || true)"
+  ccount=0; cowner2=""
+  if [ -n "$ccorpus" ]; then
+    ccount="$($K -n "$NS" get configmap "$ccorpus" -o jsonpath='{.data.corpus\.txt}' 2>/dev/null | grep -c '^/' || true)"
+    cowner2="$($K -n "$NS" get configmap "$ccorpus" -o jsonpath='{.metadata.ownerReferences[0].kind}' 2>/dev/null || true)"
+  fi
+  check "campaign emitted status.corpusConfigMap"            "echo '$ccorpus' | grep -q 'jpetstore-campaign-corpus-out'"
+  check "replay-corpus ConfigMap has route entries"          "[ '${ccount:-0}' -ge 1 ]"
+  check "replay-corpus ConfigMap owned by the campaign (GC)" "[ '$cowner2' = 'ClosureJVMCampaign' ]"
+  echo "  (corpusConfigMap=${ccorpus:-<none>}; ${ccount} route(s))"
+
   # --- P5b: the operator brought up a per-campaign dashboard and the driver pushed to it ---------
   say "Assert the per-campaign dashboard (P5b)"
   durl="$($K -n "$NS" get closurejvmcampaign jpetstore-campaign -o jsonpath='{.status.dashboardURL}' 2>/dev/null || true)"

deploy/helm/closurejvm-operator/crds/closurejvm.dev_closurejvmcampaigns.yaml

@@ -229,6 +229,12 @@ spec:
                 x-kubernetes-list-map-keys:
                 - type
                 x-kubernetes-list-type: map
+              corpusConfigMap:
+                description: |-
+                  CorpusConfigMap names a ConfigMap the operator emits at end-of-run holding the run's interesting
+                  "replay corpus" (the inputs that reached new coverage), for reproducibility, the dashboard corpus
+                  view, and load-mode replay (DD-026 PR 1). Owner-referenced to the campaign, so it GCs with it.
+                type: string
               coveragePct:
                 description: CoveragePct/Findings are read from the driver's end-of-run
                   summary (DD-025 §7a).

deploy/helm/closurejvm-operator/templates/rbac.yaml

@@ -20,7 +20,7 @@ rules:
     verbs: ["create", "delete", "get", "list", "watch"]
   - apiGroups: [""]
     resources: ["configmaps"]
-    verbs: ["get", "list", "watch"]
+    verbs: ["get", "list", "watch", "create", "update"]
   - apiGroups: [""]
     resources: ["pods"]
     verbs: ["get", "list", "watch"]

docs/LOAD-MODE-DESIGN.md

@@ -162,9 +162,14 @@ status:
 ## 6. Phased plan
 
 - [x] **PR 0 — this design note (DD-026).**
-- [ ] **PR 1 — producer.** Runner writes the interesting corpus out; operator captures it into a
-  campaign-owned ConfigMap and sets `status.corpusConfigMap`. Ships on its own (reproducibility +
-  dashboard corpus view). Adds no `load` behavior yet.
+- [x] **PR 1 — producer.** **Chosen transport: reuse the termination-message channel, not a sidecar.**
+  The driver splices a byte-capped (~3 KB, ≈top-N) `replayCorpus` array into the summary it already
+  writes to `/dev/termination-log`; the operator (which now has `configmaps: create;update`) reads it
+  back and materializes a campaign-owned `<campaign>-corpus-out` ConfigMap, setting
+  `status.corpusConfigMap`. This keeps the driver **credential-less** (no new pod SA/identity — the
+  costly path §3 flagged) and the top-N cap is the right load-replay semantics anyway. Trade-off: the
+  emitted corpus is bounded to what fits the ~4 KiB termination message; a larger transport
+  (sidecar/PVC) is a future extension if corpora need to be bigger.
 - [ ] **PR 2 — consumer.** `mode: load` + `driver.concurrency`; the load driver; `status.load` metrics
   + dashboard wiring.
 

operator/api/v1alpha1/closurejvmcampaign_types.go

@@ -119,6 +119,11 @@ type ClosureJVMCampaignStatus struct {
 	CoveragePct string `json:"coveragePct,omitempty"`
 	// +optional
 	Findings int32 `json:"findings,omitempty"`
+	// CorpusConfigMap names a ConfigMap the operator emits at end-of-run holding the run's interesting
+	// "replay corpus" (the inputs that reached new coverage), for reproducibility, the dashboard corpus
+	// view, and load-mode replay (DD-026 PR 1). Owner-referenced to the campaign, so it GCs with it.
+	// +optional
+	CorpusConfigMap string `json:"corpusConfigMap,omitempty"`
 	// +optional
 	StartTime *metav1.Time `json:"startTime,omitempty"`
 	// +optional

operator/config/crd/bases/closurejvm.dev_closurejvmcampaigns.yaml

@@ -229,6 +229,12 @@ spec:
                 x-kubernetes-list-map-keys:
                 - type
                 x-kubernetes-list-type: map
+              corpusConfigMap:
+                description: |-
+                  CorpusConfigMap names a ConfigMap the operator emits at end-of-run holding the run's interesting
+                  "replay corpus" (the inputs that reached new coverage), for reproducibility, the dashboard corpus
+                  view, and load-mode replay (DD-026 PR 1). Owner-referenced to the campaign, so it GCs with it.
+                type: string
               coveragePct:
                 description: CoveragePct/Findings are read from the driver's end-of-run
                   summary (DD-025 §7a).

operator/config/rbac/role.yaml

@@ -79,8 +79,10 @@ rules:
   resources:
   - configmaps
   verbs:
+  - create
   - get
   - list
+  - update
   - watch
 - apiGroups:
   - ""

operator/internal/controller/closurejvmcampaign_controller.go

@@ -22,6 +22,7 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"strings"
 	"time"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -62,7 +63,7 @@ type ClosureJVMCampaignReconciler struct {
 //+kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;delete
 //+kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;delete
 //+kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch
-//+kubebuilder:rbac:groups=core,resources=configmaps,verbs=get;list;watch
+//+kubebuilder:rbac:groups=core,resources=configmaps,verbs=get;list;watch;create;update
 
 func (r *ClosureJVMCampaignReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	l := log.FromContext(ctx)
@@ -200,11 +201,28 @@ func (r *ClosureJVMCampaignReconciler) Reconcile(ctx context.Context, req ctrl.R
 
 	// --- aggregate the Job's outcome -----------------------------------------------------------
 	if job.Status.Succeeded > 0 {
-		campaign.Status.Phase = closurejvmv1alpha1.CampaignCompleted
 		if s := r.readDriverSummary(ctx, &campaign); s != nil {
 			campaign.Status.CoveragePct = fmt.Sprintf("%.1f", s.Exploration.Coverage.Pct)
 			campaign.Status.Findings = s.Exploration.Corpus
+			// Persist the interesting "replay corpus" as a campaign-owned ConfigMap (DD-026 PR 1) —
+			// for reproducibility, the dashboard corpus view, and load-mode replay. The driver stays
+			// credential-less: it wrote the corpus into its summary (termination message); the operator,
+			// which already holds the RBAC, materializes the ConfigMap here. Emit BEFORE flipping to a
+			// terminal phase: a terminal campaign never reconciles again (top-of-func guard), so a
+			// transient failure here would otherwise permanently forfeit the corpus. On failure, stay
+			// Running and retry next reconcile (the create-or-update is idempotent).
+			if len(s.ReplayCorpus) > 0 {
+				if err := r.emitCorpusConfigMap(ctx, &campaign, s.ReplayCorpus); err != nil {
+					l.Error(err, "emitting replay-corpus ConfigMap; will retry")
+					campaign.Status.Phase = closurejvmv1alpha1.CampaignRunning
+					if uerr := r.Status().Update(ctx, &campaign); uerr != nil {
+						return ctrl.Result{}, uerr
+					}
+					return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
+				}
+			}
 		}
+		campaign.Status.Phase = closurejvmv1alpha1.CampaignCompleted
 		now := metav1.Now()
 		campaign.Status.CompletionTime = &now
 		meta.SetStatusCondition(&campaign.Status.Conditions, metav1.Condition{
@@ -231,14 +249,52 @@ func (r *ClosureJVMCampaignReconciler) Reconcile(ctx context.Context, req ctrl.R
 	return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
 }
 
-// driverSummary is the subset of StatusReporter.snapshotJson the campaign surfaces.
+// driverSummary is the subset of the driver's end-of-run summary the campaign surfaces.
 type driverSummary struct {
 	Exploration struct {
 		Corpus   int32 `json:"corpus"`
 		Coverage struct {
 			Pct float64 `json:"pct"`
 		} `json:"coverage"`
 	} `json:"exploration"`
+	// ReplayCorpus is the capped set of interesting inputs the run fired (DD-026 PR 1); the operator
+	// materializes it into status.corpusConfigMap.
+	ReplayCorpus []string `json:"replayCorpus"`
+}
+
+const corpusOutKey = "corpus.txt"
+
+func corpusConfigMapName(c *closurejvmv1alpha1.ClosureJVMCampaign) string {
+	return c.Name + "-corpus-out"
+}
+
+// emitCorpusConfigMap materializes the run's replay corpus into a campaign-owned ConfigMap (one key,
+// newline-joined) and records it in status.corpusConfigMap. Create-or-update (idempotent on requeue).
+func (r *ClosureJVMCampaignReconciler) emitCorpusConfigMap(ctx context.Context, c *closurejvmv1alpha1.ClosureJVMCampaign, corpus []string) error {
+	cm := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{Name: corpusConfigMapName(c), Namespace: c.Namespace},
+		Data:       map[string]string{corpusOutKey: strings.Join(corpus, "\n") + "\n"},
+	}
+	if err := controllerutil.SetControllerReference(c, cm, r.Scheme); err != nil {
+		return err
+	}
+	var existing corev1.ConfigMap
+	switch err := r.Get(ctx, types.NamespacedName{Namespace: cm.Namespace, Name: cm.Name}, &existing); {
+	case apierrors.IsNotFound(err):
+		if cerr := r.Create(ctx, cm); cerr != nil && !apierrors.IsAlreadyExists(cerr) {
+			return cerr
+		}
+	case err != nil:
+		return err
+	default:
+		existing.Data = cm.Data
+		existing.OwnerReferences = cm.OwnerReferences
+		if uerr := r.Update(ctx, &existing); uerr != nil {
+			return uerr
+		}
+	}
+	c.Status.CorpusConfigMap = cm.Name
+	return nil
 }
 
 // readDriverSummary finds the driver Job's pod and parses the summary the runner wrote to its

operator/internal/controller/closurejvmcampaign_controller_test.go

@@ -120,6 +120,7 @@ var _ = Describe("ClosureJVMCampaign Controller (P5a)", func() {
 			// remove it explicitly to keep specs isolated.
 			&appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: campaignName + "-dashboard", Namespace: namespace}},
 			&corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: campaignName + "-dashboard", Namespace: namespace}},
+			&corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: campaignName + "-corpus-out", Namespace: namespace}},
 		} {
 			_ = k8sClient.Delete(ctx, o)
 		}
@@ -220,6 +221,44 @@ var _ = Describe("ClosureJVMCampaign Controller (P5a)", func() {
 		Expect(got.Status.CompletionTime).NotTo(BeNil())
 	})
 
+	It("emits the replay corpus as a campaign-owned ConfigMap on completion (DD-026 PR 1)", func() {
+		Expect(k8sClient.Create(ctx, newTargetDeploy())).To(Succeed())
+		makeInjectedTarget()
+		Expect(k8sClient.Create(ctx, newCampaign())).To(Succeed())
+		_, err := reconcileOnce()
+		Expect(err).NotTo(HaveOccurred())
+
+		job := &batchv1.Job{}
+		Expect(k8sClient.Get(ctx, jobKey, job)).To(Succeed())
+		job.Status.Succeeded = 1
+		Expect(k8sClient.Status().Update(ctx, job)).To(Succeed())
+		pod := &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{Name: campaignName + "-driver-c", Namespace: namespace,
+				Labels: map[string]string{"closurejvm.dev/campaign": campaignName}},
+			Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "driver", Image: runnerImg}}},
+		}
+		Expect(k8sClient.Create(ctx, pod)).To(Succeed())
+		Expect(k8sClient.Get(ctx, types.NamespacedName{Name: pod.Name, Namespace: namespace}, pod)).To(Succeed())
+		pod.Status.ContainerStatuses = []corev1.ContainerStatus{{
+			Name: "driver",
+			State: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{
+				Message: `{"exploration":{"corpus":2,"coverage":{"pct":11.0}},"replayCorpus":["/actions/Catalog.action","/actions/Cart.action?add=1"]}`}},
+		}}
+		Expect(k8sClient.Status().Update(ctx, pod)).To(Succeed())
+
+		_, err = reconcileOnce()
+		Expect(err).NotTo(HaveOccurred())
+
+		got := &closurejvmv1alpha1.ClosureJVMCampaign{}
+		Expect(k8sClient.Get(ctx, campaignKey, got)).To(Succeed())
+		Expect(got.Status.CorpusConfigMap).To(Equal(campaignName + "-corpus-out"))
+
+		cm := &corev1.ConfigMap{}
+		Expect(k8sClient.Get(ctx, types.NamespacedName{Name: campaignName + "-corpus-out", Namespace: namespace}, cm)).To(Succeed())
+		Expect(cm.Data["corpus.txt"]).To(Equal("/actions/Catalog.action\n/actions/Cart.action?add=1\n"))
+		Expect(cm.OwnerReferences).To(ContainElement(HaveField("Name", campaignName)))
+	})
+
 	It("surfaces a failed initContainer's reason in campaign status (review #24)", func() {
 		Expect(k8sClient.Create(ctx, newTargetDeploy())).To(Succeed())
 		makeInjectedTarget()

runner/coverage/CoverageGuidedRun.java

@@ -139,6 +139,9 @@ public static void main(String[] args) throws Exception {
         // Start the corpus as the full seed set so every seeded endpoint is exercised, rather than
         // relying on random discovery to stumble onto them.
         List<String> corpus = new ArrayList<>(seeds);
+        // Publish the live corpus so the end-of-run summary can emit a capped "replay corpus" the
+        // operator persists (DD-026 PR 1). Same list object, populated as the run finds new coverage.
+        lastCorpus = corpus;
         long best = 0, total = 0;
 
         // Alternate session epochs: sign on for a stretch (so account/cart/order handlers run
@@ -174,7 +177,9 @@ public static void main(String[] args) throws Exception {
                 total = lastCoverageTotal;   // else a sequence-only run reports coverage=N/0
                 if (coveredAfterSeq > best) {
                     best = coveredAfterSeq;
-                    corpus.add(sequence.get(sequence.size() - 1));
+                    // Synchronized: the summary shutdown hook may snapshot this list concurrently on an
+                    // external SIGTERM (kubectl delete pod / eviction) before the deadline is hit.
+                    synchronized (corpus) { corpus.add(sequence.get(sequence.size() - 1)); }
                     StatusReporter.recordSaved("Coverage");
                 }
                 continue;
@@ -212,7 +217,7 @@ public static void main(String[] args) throws Exception {
             total = lastCoverageTotal;
             if (covered > best) {
                 best = covered;
-                corpus.add(input);
+                synchronized (corpus) { corpus.add(input); } // see the sequence branch above
                 StatusReporter.recordSaved("Coverage");
             }
         }
@@ -236,16 +241,111 @@ static long parseDurationMillis(String s) {
         return (long) (Double.parseDouble(num.trim()) * mult);
     }
 
-    /** Write the end-of-run metrics (StatusReporter's snapshot JSON) to {@code path} for the operator. */
-    private static void writeSummary(String path) {
+    /** The run's live corpus, published for the end-of-run summary's replay-corpus emission (DD-026). */
+    static volatile List<String> lastCorpus; // package-private for the combined-size test
+
+    /** Absolute upper bound on the replay-corpus bytes (also overridable); the effective budget is the
+     *  smaller of this and what's left of the ~4 KiB termination message after the metrics JSON. */
+    static final int REPLAY_CORPUS_MAX_BYTES =
+            Integer.getInteger("closurejvm.corpus.out.maxBytes", 3000);
+
+    /** Keep the whole summary (metrics + corpus) safely under kubelet's 4096-byte termination cap. */
+    private static final int TERMINATION_MSG_BUDGET = 3900;
+
+    /**
+     * Write the end-of-run summary (StatusReporter's metrics JSON, plus a capped {@code replayCorpus}
+     * array of the interesting inputs) to {@code path} for the operator to read back (DD-025 §7a,
+     * DD-026 PR 1). The corpus is best-effort and is built in a separate try so a corpus failure (e.g.
+     * a shutdown-time hiccup) can never take down the metrics summary the operator depends on.
+     */
+    static void writeSummary(String path) { // package-private for testing
+        String snap;
+        try {
+            snap = StatusReporter.snapshotJson(); // a complete JSON object: {...}
+        } catch (Exception e) {
+            return; // no metrics to write
+        }
+        String corpusJson = "[]";
+        try {
+            // Defensive copy under the list's own monitor — the main loop's synchronized add()s and this
+            // snapshot can't interleave, so no ConcurrentModificationException on a SIGTERM race.
+            List<String> lc = lastCorpus;
+            List<String> snapshot;
+            if (lc != null) {
+                synchronized (lc) { snapshot = new ArrayList<>(lc); }
+            } else {
+                snapshot = java.util.Collections.emptyList();
+            }
+            // Budget the corpus to what fits the termination message alongside the actual metrics size.
+            int snapBytes = snap.getBytes(StandardCharsets.UTF_8).length;
+            int overhead = ",\"replayCorpus\":".length(); // the closing '}' is already counted in snap
+            int budget = Math.min(REPLAY_CORPUS_MAX_BYTES, TERMINATION_MSG_BUDGET - snapBytes - overhead);
+            corpusJson = replayCorpusJson(snapshot, Math.max(2, budget)); // >=2 so "[]" always fits
+        } catch (Exception ignored) {
+            corpusJson = "[]"; // corpus is best-effort; fall through and still write the metrics
+        }
         try {
-            java.nio.file.Files.write(Paths.get(path),
-                    StatusReporter.snapshotJson().getBytes(StandardCharsets.UTF_8));
+            // Splice "replayCorpus":[...] before the closing brace; the operator ignores the field if it
+            // doesn't parse it, and materializes status.corpusConfigMap if it does.
+            String merged = snap.endsWith("}")
+                    ? snap.substring(0, snap.length() - 1) + ",\"replayCorpus\":" + corpusJson + "}"
+                    : snap;
+            java.nio.file.Files.write(Paths.get(path), merged.getBytes(StandardCharsets.UTF_8));
         } catch (Exception ignored) {
             // never let summary-writing break the run's exit
         }
     }
 
+    /**
+     * A JSON array of distinct corpus inputs, added in order until the encoded UTF-8 size would exceed
+     * {@code maxBytes} (a true byte budget — the termination message is byte-limited). Package-private
+     * for testing.
+     */
+    static String replayCorpusJson(List<String> corpus, int maxBytes) {
+        StringBuilder sb = new StringBuilder("[");
+        int bytes = 1 + 1; // '[' + ']'
+        if (corpus != null) {
+            java.util.LinkedHashSet<String> seen = new java.util.LinkedHashSet<>(corpus);
+            boolean first = true;
+            for (String entry : seen) {
+                String enc = jsonString(entry);
+                int encBytes = enc.getBytes(StandardCharsets.UTF_8).length + (first ? 0 : 1); // + comma
+                if (bytes + encBytes > maxBytes) {
+                    break;
+                }
+                if (!first) {
+                    sb.append(',');
+                }
+                sb.append(enc);
+                bytes += encBytes;
+                first = false;
+            }
+        }
+        return sb.append(']').toString();
+    }
+
+    /** Minimal JSON string encoder (quotes + escapes) — avoids a JSON dependency for one field. */
+    private static String jsonString(String s) {
+        StringBuilder b = new StringBuilder("\"");
+        for (int i = 0; i < s.length(); i++) {
+            char c = s.charAt(i);
+            switch (c) {
+                case '"':  b.append("\\\""); break;
+                case '\\': b.append("\\\\"); break;
+                case '\n': b.append("\\n");  break;
+                case '\r': b.append("\\r");  break;
+                case '\t': b.append("\\t");  break;
+                default:
+                    if (c < 0x20) {
+                        b.append(String.format("\\u%04x", (int) c));
+                    } else {
+                        b.append(c);
+                    }
+            }
+        }
+        return b.append('"').toString();
+    }
+
     /**
      * Cookie jar for the current session "epoch". JPetStore keeps the signed-on account in the
      * HTTP session, so without carrying JSESSIONID every authenticated handler sees a null account