fix: ensure ResponseComplete hook always executes (kubernetes-sigs/gateway-api-inference-extension#2064)

This guarantees request/response symmetry to prevent capacity leaks in
stateful plugins (e.g., Concurrency Detector).

Previously, errors during JSON marshaling, client disconnects, or split
streaming chunks could cause the `ResponseComplete` hook to be skipped.

Changes:
- Add `defer` safety block to trigger completion on errors/disconnects.
- Move streaming completion trigger to the authoritative `EndOfStream`
  signal rather than relying on body content parsing.

Author: LukeAVanDrie

pkg/epp/handlers/response.go

@@ -97,10 +97,6 @@ func (s *StreamingServer) HandleResponseBodyModelStreaming(ctx context.Context,
 			cachedToken = reqCtx.Usage.PromptTokenDetails.CachedTokens
 		}
 		metrics.RecordPromptCachedTokens(reqCtx.IncomingModelName, reqCtx.TargetModelName, cachedToken)
-		_, err := s.director.HandleResponseBodyComplete(ctx, reqCtx)
-		if err != nil {
-			logger.Error(err, "error in HandleResponseBodyComplete")
-		}
 	}
 }
 

pkg/epp/handlers/server.go

@@ -160,6 +160,17 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 		if reqCtx.RequestRunning {
 			metrics.DecRunningRequests(reqCtx.IncomingModelName)
 		}
+
+		// If we scheduled a pod (TargetPod != nil) but never marked the response  as complete (e.g. error, disconnect,
+		// panic), force the completion hooks to run.
+		if reqCtx.TargetPod != nil && !reqCtx.ResponseComplete {
+			// Use a fresh context as the request context might be canceled (Client Disconnect).
+			// We only need logging from the original context.
+			cleanupCtx := log.IntoContext(context.Background(), logger)
+			if _, err := s.director.HandleResponseBodyComplete(cleanupCtx, reqCtx); err != nil {
+				logger.Error(err, "error in HandleResponseBodyComplete")
+			}
+		}
 	}(err, reqCtx)
 
 	for {
@@ -270,6 +281,10 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 				s.HandleResponseBodyModelStreaming(ctx, reqCtx, responseText)
 				if v.ResponseBody.EndOfStream {
 					loggerTrace.Info("stream completed")
+					reqCtx.ResponseComplete = true
+					if _, err := s.director.HandleResponseBodyComplete(ctx, reqCtx); err != nil {
+						logger.Error(err, "error in HandleResponseBodyComplete")
+					}
 
 					reqCtx.ResponseCompleteTimestamp = time.Now()
 					metrics.RecordRequestLatencies(ctx, reqCtx.IncomingModelName, reqCtx.TargetModelName, reqCtx.RequestReceivedTimestamp, reqCtx.ResponseCompleteTimestamp)

pkg/epp/requestcontrol/plugins.go

@@ -52,7 +52,12 @@ type ResponseStreaming interface {
 	ResponseStreaming(ctx context.Context, request *types.LLMRequest, response *Response, targetPod *backend.Pod)
 }
 
-// ResponseComplete is called by the director after the complete response is sent.
+// ResponseComplete is called by the director when the request lifecycle terminates.
+// This occurs after a response is fully sent, OR if the request fails/disconnects after a pod was scheduled.
+//
+// Plugins should assume this is the final cleanup hook for a request.
+//
+// TODO: Consider passing an error or success bool; however, this is a breaking change and is deffered for now.
 type ResponseComplete interface {
 	plugins.Plugin
 	ResponseComplete(ctx context.Context, request *types.LLMRequest, response *Response, targetPod *backend.Pod)