diff --git a/internal/agent/loop_finalize.go b/internal/agent/loop_finalize.go index 44bc40ab..d15c1bcd 100644 --- a/internal/agent/loop_finalize.go +++ b/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{ diff --git a/internal/agent/loop_pipeline_adapter.go b/internal/agent/loop_pipeline_adapter.go index 33778445..1bf544dc 100644 --- a/internal/agent/loop_pipeline_adapter.go +++ b/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{ diff --git a/internal/consolidation/episodic_worker.go b/internal/consolidation/episodic_worker.go index e03edc92..b4c7762d 100644 --- a/internal/consolidation/episodic_worker.go +++ b/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, diff --git a/internal/consolidation/workers_test.go b/internal/consolidation/workers_test.go index 14045dbb..17c60175 100644 --- a/internal/consolidation/workers_test.go +++ b/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) {