From 9dee3b3d26a919068b59745fb0d6b4d23436c4ff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tu=C3=A2n?= Date: Sat, 11 Apr 2026 08:12:55 +0700 Subject: [PATCH] =?UTF-8?q?fix(agent):=20use=20agentUUID=20instead=20of=20?= =?UTF-8?q?agent=20key=20in=20DomainEvent=20for=20epi=E2=80=A6=20(#826)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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) * 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 Co-authored-by: Claude Opus 4.6 (1M context) Co-authored-by: viettranx --- internal/agent/loop_finalize.go | 2 +- internal/agent/loop_pipeline_adapter.go | 2 +- internal/consolidation/episodic_worker.go | 15 +++++- internal/consolidation/workers_test.go | 61 +++++++++++++++++++++++ 4 files changed, 76 insertions(+), 4 deletions(-) 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) {