fix(source): preserve original custom_id on cancellation

internal/processor/worker/source_planfile.go

@@ -64,11 +64,11 @@ func NewPlanFileSource(cfg PlanFileSourceConfig) *PlanFileSource {
 	}
 }
 
-// Produce sends one item per plan entry to the channel. After context
-// cancellation, it skips I/O (ReadAt + Unmarshal) but still sends a
-// minimal item so the dispatcher's drain loop can account for it.
-// This avoids both silent entry drops and unnecessary I/O during shutdown.
-func (s *PlanFileSource) Produce(ctx context.Context, outgoingRequestCh chan<- pipeline.RequestItem) error {
+// Produce sends one item per plan entry to the channel. It always reads the
+// input line so each item retains the original custom_id: cancel / expire
+// drain still needs that identity in the error file even when inference is
+// skipped. Context cancellation is handled by the dispatcher drain path.
+func (s *PlanFileSource) Produce(_ context.Context, outgoingRequestCh chan<- pipeline.RequestItem) error {
 	defer close(outgoingRequestCh)
 
 	for safeModelID, modelID := range s.modelMap.SafeToModel {
@@ -79,15 +79,6 @@ func (s *PlanFileSource) Produce(ctx context.Context, outgoingRequestCh chan<- p
 		}
 
 		for _, entry := range entries {
-			if ctx.Err() != nil {
-				reqID := fmt.Sprintf("batch_req_%s", uuid.NewString())
-				outgoingRequestCh <- pipeline.RequestItem{
-					RequestID: reqID,
-					CustomID:  reqID,
-					ModelID:   modelID,
-				}
-				continue
-			}
 			item, err := s.readEntry(entry, modelID)
 			if err != nil {
 				return err

internal/processor/worker/source_planfile_test.go

@@ -569,9 +569,10 @@ func TestPlanFileSource_Produce_BadPlanFile(t *testing.T) {
 
 // TestPlanFileSource_Produce_CancellationProducesAllEntries verifies that when
 // the context is cancelled mid-produce, ALL plan entries still reach the output
-// channel — either as normal items (produced before cancellation) or as items
-// that the downstream dispatcher can drain as cancelled. If entries are silently
-// dropped, completed + failed < total and the job's output files are incomplete.
+// channel with their original custom_id so the dispatcher can drain them as
+// cancelled without losing OpenAI request identity. If entries are dropped or
+// get synthetic custom_ids, completed + failed may match total while clients
+// cannot match error-file rows to their input.
 func TestPlanFileSource_Produce_CancellationProducesAllEntries(t *testing.T) {
 	const totalRequests = 10
 	dir := t.TempDir()
@@ -616,7 +617,7 @@ func TestPlanFileSource_Produce_CancellationProducesAllEntries(t *testing.T) {
 	resolver := inference.NewSingleClientResolver(client)
 	defer func() { _ = resolver.Close() }()
 
-	// Cancel the context immediately so the source hits ctx.Err() early.
+	// Cancel before Produce; identity must still come from the input lines.
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
 
@@ -632,20 +633,24 @@ func TestPlanFileSource_Produce_CancellationProducesAllEntries(t *testing.T) {
 	out := make(chan pipeline.RequestItem, totalRequests+1)
 	_ = source.Produce(ctx, out)
 
-	var produced, skippedIO int
+	seen := make(map[string]bool, totalRequests)
 	for item := range out {
-		produced++
 		if item.CustomID == item.RequestID {
-			skippedIO++
+			t.Errorf("custom_id %q equals request_id: cancel path must keep original input custom_id", item.CustomID)
 		}
+		if seen[item.CustomID] {
+			t.Errorf("duplicate custom_id %q", item.CustomID)
+		}
+		seen[item.CustomID] = true
 	}
 
-	if produced != totalRequests {
-		t.Fatalf("produced %d items, want %d: source dropped %d entries on cancellation",
-			produced, totalRequests, totalRequests-produced)
+	if len(seen) != totalRequests {
+		t.Fatalf("produced %d unique custom_ids, want %d", len(seen), totalRequests)
 	}
-	if skippedIO != totalRequests {
-		t.Fatalf("skippedIO %d items, want %d: source should skip I/O after cancellation",
-			skippedIO, totalRequests)
+	for i := range totalRequests {
+		want := fmt.Sprintf("c-%d", i)
+		if !seen[want] {
+			t.Errorf("missing custom_id %q after cancelled Produce", want)
+		}
 	}
 }

scripts/dev-deploy-dispatcher.sh

@@ -64,8 +64,11 @@ else
 fi
 
 step "Loading dispatcher image into Kind cluster '${KIND_CLUSTER_NAME}'..."
-if [[ "${CONTAINER_TOOL}" == "docker" ]]; then
-    kind load docker-image "${DISPATCHER_IMAGE}" --name "${KIND_CLUSTER_NAME}"
+if docker exec "${KIND_CLUSTER_NAME}-control-plane" ctr --namespace=k8s.io images list -q 2>/dev/null | grep -q "^${DISPATCHER_IMAGE}$"; then
+    log "Image already present in Kind node, skipping load"
+elif [[ "${CONTAINER_TOOL}" == "docker" ]]; then
+    docker save "${DISPATCHER_IMAGE}" | docker exec -i "${KIND_CLUSTER_NAME}-control-plane" \
+        ctr --namespace=k8s.io images import --snapshotter=overlayfs -
 else
     ${CONTAINER_TOOL} save "${DISPATCHER_IMAGE}" | kind load image-archive /dev/stdin --name "${KIND_CLUSTER_NAME}"
 fi