fix(EVO-558): QA fixes — FNV hash, signature placement, A2A adapter hardening

- Fix FNV hash to guarantee positive int64 (mask high bit)
- Prepend signature to first segment only instead of all parts
- Harden A2A adapter response parsing and error handling
- Update pipeline service for multi-platform bot delegation
- Update tests to match corrected behavior

Author: gomessguii

.env.example

@@ -1,6 +1,5 @@
 LISTEN_ADDR=:8080
 REDIS_URL=redis://localhost:6379
 BOT_RUNTIME_SECRET=change_me_in_production
-AI_PROCESSOR_URL=http://ai-processor:8000
-AI_PROCESSOR_API_KEY=your_api_key
+AI_PROCESSOR_BASE_URL=http://ai-processor:8000
 AI_CALL_TIMEOUT_SECONDS=30

cmd/server/main.go

@@ -51,7 +51,7 @@ func main() {
 	debounce := debounceService.NewDebounceEngine(pipelineRepo)
 
 	// Step 6: AI adapter
-	aiAdapter := aiService.NewAIAdapter(cfg.AIProcessorURL, cfg.AIProcessorAPIKey, cfg.AICallTimeoutSeconds)
+	aiAdapter := aiService.NewAIAdapter(cfg.AIProcessorBaseURL, cfg.AICallTimeoutSeconds)
 
 	// Step 7: dispatch engine
 	dispatch := dispatchService.NewDispatchEngine()

internal/config/config.go

@@ -10,8 +10,7 @@ type Config struct {
 	ListenAddr           string
 	RedisURL             string
 	BotRuntimeSecret     string
-	AIProcessorURL       string
-	AIProcessorAPIKey    string
+	AIProcessorBaseURL   string
 	AICallTimeoutSeconds int
 }
 
@@ -28,11 +27,7 @@ func Load() (*Config, error) {
 	if err != nil {
 		return nil, err
 	}
-	aiProcessorURL, err := mustGetEnv("AI_PROCESSOR_URL")
-	if err != nil {
-		return nil, err
-	}
-	aiProcessorAPIKey, err := mustGetEnv("AI_PROCESSOR_API_KEY")
+	aiProcessorBaseURL, err := mustGetEnv("AI_PROCESSOR_BASE_URL")
 	if err != nil {
 		return nil, err
 	}
@@ -45,8 +40,7 @@ func Load() (*Config, error) {
 		ListenAddr:           listenAddr,
 		RedisURL:             redisURL,
 		BotRuntimeSecret:     botRuntimeSecret,
-		AIProcessorURL:       aiProcessorURL,
-		AIProcessorAPIKey:    aiProcessorAPIKey,
+		AIProcessorBaseURL:   aiProcessorBaseURL,
 		AICallTimeoutSeconds: aiCallTimeout,
 	}, nil
 }

pkg/ai/model/a2a.go

@@ -1,17 +1,61 @@
 package model
 
-// A2ARequest is the envelope sent to AI Processor via HTTP POST.
-// ContactID and ConversationID are internal tracking fields (json:"-") used for
-// structured logging; they are never serialised into the HTTP payload.
+// A2ARequest carries the data needed to call AI Processor via JSON-RPC 2.0.
+// Fields are NOT serialised directly — the adapter builds the wire format.
 type A2ARequest struct {
-	Message        string `json:"message"` // aggregated buffer content (FR-15)
-	ContactID      int64  `json:"-"`
-	ConversationID int64  `json:"-"`
+	AgentBotID     string // used to construct URL path: /api/v1/a2a/{AgentBotID}
+	ContactID      int64  // used for userId in JSON-RPC params
+	ConversationID int64  // used for contextId in JSON-RPC params
+	ApiKey         string // used for X-API-Key header (per-event auth)
+	Message        string // aggregated buffer content (FR-15)
 }
 
-// A2AResponse is the raw JSON response from AI Processor.
+// jsonRPCRequest is the JSON-RPC 2.0 envelope sent to AI Processor.
+type JSONRPCRequest struct {
+	JSONRPC string         `json:"jsonrpc"`
+	ID      string         `json:"id"`
+	Method  string         `json:"method"`
+	Params  JSONRPCParams  `json:"params"`
+}
+
+type JSONRPCParams struct {
+	ContextID string         `json:"contextId"`
+	UserID    string         `json:"userId"`
+	Message   JSONRPCMessage `json:"message"`
+	Metadata  map[string]any `json:"metadata"`
+}
+
+type JSONRPCMessage struct {
+	Role  string         `json:"role"`
+	Parts []JSONRPCPart  `json:"parts"`
+}
+
+type JSONRPCPart struct {
+	Type string `json:"type"`
+	Text string `json:"text,omitempty"`
+}
+
+// A2AResponse is the JSON-RPC 2.0 response from AI Processor.
 type A2AResponse struct {
-	Content string `json:"content"`
+	Result *A2AResult `json:"result"`
+}
+
+type A2AResult struct {
+	Artifacts []A2AArtifact `json:"artifacts"`
+	Message   *A2AMessage   `json:"message"`
+}
+
+type A2AArtifact struct {
+	Parts []A2APart `json:"parts"`
+}
+
+type A2AMessage struct {
+	Parts []A2APart `json:"parts"`
+}
+
+type A2APart struct {
+	Type string `json:"type"`
+	Text string `json:"text"`
 }
 
 // NormalizedResponse is the internal format after parsing A2AResponse.

pkg/ai/service/ai_adapter.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"log/slog"
 	"net/http"
+	"strings"
 	"time"
 
 	brtErrors "github.com/EvolutionAPI/evo-bot-runtime/internal/errors"
@@ -18,24 +19,23 @@ import (
 // maxResponseBytes caps the AI Processor response body to prevent OOM on oversized payloads.
 const maxResponseBytes = 1 << 20 // 1 MiB
 
-// AIAdapter calls the AI Processor via A2A protocol.
+// AIAdapter calls the AI Processor via A2A protocol (JSON-RPC 2.0).
 // Swap the backend by providing a different implementation at main.go wiring.
 type AIAdapter interface {
 	Call(ctx context.Context, req *model.A2ARequest) (*model.NormalizedResponse, error)
 }
 
 type aiAdapter struct {
-	url         string
-	apiKey      string
+	baseURL     string
 	timeoutSecs int
 	client      *http.Client
 }
 
 // NewAIAdapter constructs the adapter. Returns interface (GEAR R03).
-func NewAIAdapter(url, apiKey string, timeoutSecs int) AIAdapter {
+// baseURL is the AI Processor base URL without path (e.g. http://ai-processor:8000).
+func NewAIAdapter(baseURL string, timeoutSecs int) AIAdapter {
 	return &aiAdapter{
-		url:         url,
-		apiKey:      apiKey,
+		baseURL:     strings.TrimRight(baseURL, "/"),
 		timeoutSecs: timeoutSecs,
 		client:      &http.Client{},
 	}
@@ -45,24 +45,41 @@ func (a *aiAdapter) Call(ctx context.Context, req *model.A2ARequest) (*model.Nor
 	start := time.Now()
 
 	// Wrap with timeout — inner timeout, outer ctx for pipeline cancellation.
-	// Error discrimination order matters:
-	//   1. ctx.Err() first  → pipeline cancellation (outer context, set by PipelineService.Cancel)
-	//   2. timeoutCtx.Err() → AI timeout (inner context, set by WithTimeout)
-	//   3. default           → generic HTTP error
 	timeoutCtx, cancel := context.WithTimeout(ctx, time.Duration(a.timeoutSecs)*time.Second)
 	defer cancel()
 
-	body, err := json.Marshal(req)
+	// Build per-event URL: {baseURL}/api/v1/a2a/{agent_bot_id}
+	url := fmt.Sprintf("%s/api/v1/a2a/%s", a.baseURL, req.AgentBotID)
+
+	// Build JSON-RPC 2.0 envelope
+	rpcReq := model.JSONRPCRequest{
+		JSONRPC: "2.0",
+		ID:      fmt.Sprintf("%d:%d", req.ContactID, req.ConversationID),
+		Method:  "message/send",
+		Params: model.JSONRPCParams{
+			ContextID: fmt.Sprintf("%d", req.ConversationID),
+			UserID:    fmt.Sprintf("%d", req.ContactID),
+			Message: model.JSONRPCMessage{
+				Role: "user",
+				Parts: []model.JSONRPCPart{
+					{Type: "text", Text: req.Message},
+				},
+			},
+			Metadata: map[string]any{},
+		},
+	}
+
+	body, err := json.Marshal(rpcReq)
 	if err != nil {
 		return nil, fmt.Errorf("pipeline.ai.marshal: %w", err)
 	}
 
-	httpReq, err := http.NewRequestWithContext(timeoutCtx, http.MethodPost, a.url, bytes.NewReader(body))
+	httpReq, err := http.NewRequestWithContext(timeoutCtx, http.MethodPost, url, bytes.NewReader(body))
 	if err != nil {
 		return nil, fmt.Errorf("pipeline.ai.new_request: %w", err)
 	}
 	httpReq.Header.Set("Content-Type", "application/json")
-	httpReq.Header.Set("Authorization", "Bearer "+a.apiKey)
+	httpReq.Header.Set("X-API-Key", req.ApiKey)
 
 	resp, err := a.client.Do(httpReq)
 	if err != nil {
@@ -90,11 +107,40 @@ func (a *aiAdapter) Call(ctx context.Context, req *model.A2ARequest) (*model.Nor
 		return nil, fmt.Errorf("pipeline.ai.decode: %w", err)
 	}
 
+	content := extractResponseText(&a2aResp)
+
 	slog.Info("pipeline.ai.http.completed",
 		"contact_id", req.ContactID,
 		"conversation_id", req.ConversationID,
 		"duration_ms", time.Since(start).Milliseconds(),
 	)
 
-	return &model.NormalizedResponse{Content: a2aResp.Content}, nil
+	return &model.NormalizedResponse{Content: content}, nil
+}
+
+// extractResponseText extracts the text content from the A2A JSON-RPC response.
+// Tries result.artifacts[0].parts[0].text first, then result.message.parts[0].text.
+func extractResponseText(resp *model.A2AResponse) string {
+	if resp.Result == nil {
+		return ""
+	}
+	// Try artifacts first (primary response format)
+	if len(resp.Result.Artifacts) > 0 {
+		for _, artifact := range resp.Result.Artifacts {
+			for _, part := range artifact.Parts {
+				if part.Text != "" {
+					return part.Text
+				}
+			}
+		}
+	}
+	// Fallback to message format
+	if resp.Result.Message != nil {
+		for _, part := range resp.Result.Message.Parts {
+			if part.Text != "" {
+				return part.Text
+			}
+		}
+	}
+	return ""
 }