diff --git a/internal/eventbus/bus_impl.go b/internal/eventbus/bus_impl.go index 64ae246f..bbbb53a9 100644 --- a/internal/eventbus/bus_impl.go +++ b/internal/eventbus/bus_impl.go @@ -63,6 +63,9 @@ func (b *busImpl) Publish(event DomainEvent) { if b.draining.Load() { return } + // Publish-time observer: warns on non-UUID AgentID drift without blocking. + // See validate_agent_id.go for rationale (PR #826 regression safety net). + validateAgentID(event) select { case b.queue <- event: default: diff --git a/internal/eventbus/validate_agent_id.go b/internal/eventbus/validate_agent_id.go new file mode 100644 index 00000000..2ed21f08 --- /dev/null +++ b/internal/eventbus/validate_agent_id.go @@ -0,0 +1,33 @@ +package eventbus + +import ( + "log/slog" + + "github.com/google/uuid" +) + +// validateAgentID is a publish-time observer that logs a warning when a DomainEvent +// carries a non-UUID AgentID. It does NOT block the publish — observability only. +// +// Motivation: PR #826 fixed multiple call sites that passed agent_key strings +// (e.g. "goctech-leader") into DomainEvent.AgentID, silently corrupting downstream +// consumers. This helper acts as a safety net to catch any future drift BEFORE +// the event reaches a consumer that parses the field as a UUID. +// +// Log field name: `non_uuid_agent_id` (intentionally distinct from the standard +// `agent_id` field used elsewhere) to avoid collision with observability tooling +// that parses `agent_id` as a UUID. See red-team finding H6. +// +// See docs/agent-identity-conventions.md (Phase 6) for the convention. +func validateAgentID(event DomainEvent) { + if event.AgentID == "" { + return // legitimate team-owned, tenant-scoped, or anonymous event + } + if _, err := uuid.Parse(event.AgentID); err != nil { + slog.Warn("eventbus.non_uuid_agent_id", + "event_type", event.Type, + "non_uuid_agent_id", event.AgentID, + "source_id", event.SourceID, + ) + } +} diff --git a/internal/eventbus/validate_agent_id_test.go b/internal/eventbus/validate_agent_id_test.go new file mode 100644 index 00000000..732d9477 --- /dev/null +++ b/internal/eventbus/validate_agent_id_test.go @@ -0,0 +1,132 @@ +package eventbus + +import ( + "bytes" + "context" + "log/slog" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/google/uuid" +) + +// captureSlog swaps the default slog logger for one that writes to a buffer +// and returns a restore function plus a pointer to the buffer. +func captureSlog(t *testing.T) (*bytes.Buffer, func()) { + t.Helper() + var buf bytes.Buffer + handler := slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}) + prev := slog.Default() + slog.SetDefault(slog.New(handler)) + return &buf, func() { slog.SetDefault(prev) } +} + +func TestValidateAgentID_EmptyAgentID_NoWarn(t *testing.T) { + buf, restore := captureSlog(t) + defer restore() + + validateAgentID(DomainEvent{ + Type: EventRunCompleted, + SourceID: "r1", + AgentID: "", + }) + + if strings.Contains(buf.String(), "non_uuid_agent_id") { + t.Errorf("empty AgentID should not trigger warning, got log:\n%s", buf.String()) + } +} + +func TestValidateAgentID_ValidUUID_NoWarn(t *testing.T) { + buf, restore := captureSlog(t) + defer restore() + + validateAgentID(DomainEvent{ + Type: EventEpisodicCreated, + SourceID: "s1", + AgentID: uuid.New().String(), + }) + + if strings.Contains(buf.String(), "non_uuid_agent_id") { + t.Errorf("valid UUID should not trigger warning, got log:\n%s", buf.String()) + } +} + +func TestValidateAgentID_NonUUIDWarns(t *testing.T) { + buf, restore := captureSlog(t) + defer restore() + + validateAgentID(DomainEvent{ + Type: EventEpisodicCreated, + SourceID: "s1", + AgentID: "goctech-leader", // agent_key, not UUID + }) + + out := buf.String() + if !strings.Contains(out, "non_uuid_agent_id") { + t.Errorf("non-UUID AgentID should emit warning with distinct field, got log:\n%s", out) + } + if !strings.Contains(out, "goctech-leader") { + t.Errorf("warning should include the offending value, got log:\n%s", out) + } + if !strings.Contains(out, "eventbus.non_uuid_agent_id") { + t.Errorf("warning message should be eventbus.non_uuid_agent_id, got log:\n%s", out) + } +} + +func TestValidateAgentID_DistinctFieldName_NoCollision(t *testing.T) { + // H6 mitigation: the field name must NOT be `agent_id` (which downstream + // observability tooling parses as UUID) — it must be `non_uuid_agent_id`. + buf, restore := captureSlog(t) + defer restore() + + validateAgentID(DomainEvent{ + Type: EventRunCompleted, + SourceID: "r1", + AgentID: "bad-key", + }) + + out := buf.String() + if !strings.Contains(out, "non_uuid_agent_id=") { + t.Errorf("expected field name `non_uuid_agent_id=`, got log:\n%s", out) + } +} + +// TestBusPublish_InvokesValidator verifies the validator is wired into Publish, +// so every publish site is covered without per-caller instrumentation. +func TestBusPublish_InvokesValidator(t *testing.T) { + buf, restore := captureSlog(t) + defer restore() + + bus := newTestBus() + ctx := t.Context() + bus.Start(ctx) + defer func() { _ = bus.Drain(500 * time.Millisecond) }() + + var received atomic.Int32 + bus.Subscribe(EventRunCompleted, func(_ context.Context, _ DomainEvent) error { + received.Add(1) + return nil + }) + + // Publish with non-UUID AgentID — should still dispatch (observability only, not blocking) + bus.Publish(DomainEvent{ + Type: EventRunCompleted, + SourceID: "validator-test", + AgentID: "invalid-agent-key", + }) + + // Wait up to 200ms for the worker to dispatch. + deadline := time.Now().Add(200 * time.Millisecond) + for received.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + } + + if received.Load() != 1 { + t.Errorf("expected event to still be dispatched (non-blocking), got received=%d", received.Load()) + } + if !strings.Contains(buf.String(), "non_uuid_agent_id") { + t.Errorf("expected bus.Publish to invoke validateAgentID, got log:\n%s", buf.String()) + } +}