fix(agent): use agentUUID instead of agent key in DomainEvent for epi… (#826)

* fix(agent): use agentUUID instead of agent key in DomainEvent for episodic worker

DomainEvent.AgentID was set to l.id (agent_key string like "goctech-leader")
instead of l.agentUUID.String() (UUID). This caused episodic worker to fail
with "invalid input syntax for type uuid" on every session.completed event,
breaking the entire 3-tier memory consolidation pipeline.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(consolidation): parse tenant/agent IDs safely in episodic worker

Replace uuid.MustParse with uuid.Parse so a malformed tenant_id or
agent_id in a DomainEvent no longer panics the worker goroutine. Parse
up front and return a descriptive error before touching the store, so
bad IDs surface with context instead of leaking into a raw PG error.

Add regression tests asserting Handle rejects non-UUID agent_id and
tenant_id with clear errors and skips creation — guards against the
l.id vs l.agentUUID mix-up fixed in the prior commit.

---------

Co-authored-by: tuannt23065 <tuannt23065@users.noreply.github.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: viettranx <viettranx@gmail.com>

internal/agent/loop_finalize.go

@@ -201,7 +201,7 @@ func (l *Loop) finalizeRun(
 		l.domainBus.Publish(eventbus.DomainEvent{
 			Type:     eventbus.EventSessionCompleted,
 			TenantID: l.tenantID.String(),
-			AgentID:  l.id,
+			AgentID:  l.agentUUID.String(),
 			UserID:   req.UserID,
 			SourceID: req.SessionKey,
 			Payload: &eventbus.SessionCompletedPayload{

internal/agent/loop_pipeline_adapter.go

@@ -160,7 +160,7 @@ func (l *Loop) buildPipelineDeps(req *RunRequest, bridgeRS *runState) pipeline.P
 				l.domainBus.Publish(eventbus.DomainEvent{
 					Type:     eventbus.EventSessionCompleted,
 					TenantID: l.tenantID.String(),
-					AgentID:  l.id,
+					AgentID:  l.agentUUID.String(),
 					UserID:   req.UserID,
 					SourceID: sessionKey,
 					Payload: &eventbus.SessionCompletedPayload{

internal/consolidation/episodic_worker.go

@@ -33,6 +33,17 @@ func (w *episodicWorker) Handle(ctx context.Context, event eventbus.DomainEvent)
 		return fmt.Errorf("episodic: unexpected payload type %T", event.Payload)
 	}
 
+	// Parse tenant/agent IDs up front so we fail fast on bad input instead of
+	// leaking into a PG error or panicking via uuid.MustParse later on.
+	tenantUUID, err := uuid.Parse(event.TenantID)
+	if err != nil {
+		return fmt.Errorf("episodic: invalid tenant_id %q: %w", event.TenantID, err)
+	}
+	agentUUID, err := uuid.Parse(event.AgentID)
+	if err != nil {
+		return fmt.Errorf("episodic: invalid agent_id %q: %w", event.AgentID, err)
+	}
+
 	// Build source_id for idempotency
 	sourceID := fmt.Sprintf("%s:%d", payload.SessionKey, payload.CompactionCount)
 	exists, err := w.store.ExistsBySourceID(ctx, event.AgentID, event.UserID, sourceID)
@@ -65,8 +76,8 @@ func (w *episodicWorker) Handle(ctx context.Context, event eventbus.DomainEvent)
 	expiresAt := time.Now().UTC().Add(90 * 24 * time.Hour)
 
 	ep := &store.EpisodicSummary{
-		TenantID:   uuid.MustParse(event.TenantID),
-		AgentID:    uuid.MustParse(event.AgentID),
+		TenantID:   tenantUUID,
+		AgentID:    agentUUID,
 		UserID:     event.UserID,
 		SessionKey: payload.SessionKey,
 		Summary:    summary,

internal/consolidation/workers_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -384,6 +385,66 @@ func TestEpisodicWorkerHandle_DuplicateSourceID(t *testing.T) {
 	}
 }
 
+// TestEpisodicWorkerHandle_NonUUIDAgentID guards the regression where Loop
+// published DomainEvent.AgentID as the agent key (e.g. "goctech-leader")
+// instead of l.agentUUID.String(). The episodic worker must reject such
+// events with a clear error — never panic, never leak a raw PG error.
+func TestEpisodicWorkerHandle_NonUUIDAgentID(t *testing.T) {
+	mockStore := &mockEpisodicStore{}
+	worker := &episodicWorker{store: mockStore}
+
+	ctx := context.Background()
+	event := eventbus.DomainEvent{
+		Type:     eventbus.EventSessionCompleted,
+		TenantID: uuid.New().String(),
+		AgentID:  "goctech-leader", // agent key, not a UUID
+		UserID:   "test-user",
+		Payload: &eventbus.SessionCompletedPayload{
+			SessionKey:      "session-123",
+			CompactionCount: 0,
+			Summary:         "Summary",
+		},
+	}
+
+	err := worker.Handle(ctx, event)
+	if err == nil {
+		t.Fatal("Expected error for non-UUID agent_id, got nil")
+	}
+	if !strings.Contains(err.Error(), "invalid agent_id") {
+		t.Errorf("Expected 'invalid agent_id' error, got: %v", err)
+	}
+	if len(mockStore.created) != 0 {
+		t.Errorf("Expected no episodic created on bad agent_id, got %d", len(mockStore.created))
+	}
+}
+
+// TestEpisodicWorkerHandle_NonUUIDTenantID mirrors the agent_id guard for tenant_id.
+func TestEpisodicWorkerHandle_NonUUIDTenantID(t *testing.T) {
+	mockStore := &mockEpisodicStore{}
+	worker := &episodicWorker{store: mockStore}
+
+	ctx := context.Background()
+	event := eventbus.DomainEvent{
+		Type:     eventbus.EventSessionCompleted,
+		TenantID: "not-a-uuid",
+		AgentID:  uuid.New().String(),
+		UserID:   "test-user",
+		Payload: &eventbus.SessionCompletedPayload{
+			SessionKey:      "session-123",
+			CompactionCount: 0,
+			Summary:         "Summary",
+		},
+	}
+
+	err := worker.Handle(ctx, event)
+	if err == nil {
+		t.Fatal("Expected error for non-UUID tenant_id, got nil")
+	}
+	if !strings.Contains(err.Error(), "invalid tenant_id") {
+		t.Errorf("Expected 'invalid tenant_id' error, got: %v", err)
+	}
+}
+
 // Test semantic worker
 
 func TestSemanticWorkerHandle_WithValidExtraction(t *testing.T) {