fix(eventbus): add non-UUID AgentID publish-time guard

Publish-time observer warns (does not block) when DomainEvent.AgentID is
non-empty and fails uuid.Parse. Distinct log field `non_uuid_agent_id`
avoids colliding with existing telemetry that parses `agent_id` as UUID.

Safety net for PR #826-class regressions. Validator lives in bus.Publish
so all call sites are covered without per-publisher instrumentation.

Phase 1 Fix C of agent identity hardening.
This commit is contained in:
viettranx committed 2026-04-11 21:22:23 +07:00
1 parent 52f48e5ea6
commit de2bc201fa
3 files changed
+168

No files matched your search

+3
View File
@@ -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:
+33
View File
@@ -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,
)
}
}
+132
View File
@@ -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())
}
}