Agent-System/internal/events/registry.go

package events

import (
	"encoding/json"
	"fmt"
	"strings"
)

const (
	ConsumerNotify          = "notify"
	ConsumerProjection      = "projection"
	ConsumerWorkflowAdvance = "workflow_advance"
	ConsumerFAV             = "fav"

	EventCandidateReady        = "document.agent_candidate_ready"
	EventVersionCommitted      = "document.agent_version_committed"
	EventDownstreamImpact      = "document.downstream_impact_detected"
	EventChangeReportDegraded  = "document.change_report_degraded"
	EventApproved              = "document.approved"
	EventTaskRunCompleted      = "taskrun.completed"
	EventTaskRunCancelled      = "taskrun.cancelled"
	EventFAVArtifactStaged     = "fav.artifact_staged"
	EventIntentPlanAwaiting    = "intent.plan.awaiting_input"
	EventIntentPlanStepChanged = "intent.plan.step_changed"
	EventIntentPlanCompleted   = "intent.plan.completed"
	EventDocProposalRequired  = "document.proposal_required"
	EventDocUnclaimedEdit     = "document.unclaimed_edit_applied"
)

// Contract is the static allowlist row for one event name.
type Contract struct {
	EventName     string
	SchemaVersion int
	Enabled       bool
	Producers     []string
	Consumers     []string
	PayloadFields []string
}

// StaticContracts is the runtime copy of module-boundaries event sections.
// TestRegistryParityWithYAML fails if this drifts from the YAML.
func StaticContracts() []Contract {
	return []Contract{
		{EventName: EventCandidateReady, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "candidate_id", "risk"}},
		{EventName: EventVersionCommitted, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events", "app/docproject"}, Consumers: []string{ConsumerProjection, ConsumerNotify}, PayloadFields: []string{"doc_id", "version_no"}},
		{EventName: EventDownstreamImpact, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "impacted_doc_id", "version_no"}},
		{EventName: EventChangeReportDegraded, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "report_id"}},
		{EventName: EventApproved, SchemaVersion: 1, Enabled: true, Producers: []string{"app/docflow", "events", "app/docadvance"}, Consumers: []string{ConsumerWorkflowAdvance, ConsumerNotify}, PayloadFields: []string{"doc_id", "project_id", "task_key"}},
		{EventName: EventTaskRunCompleted, SchemaVersion: 1, Enabled: true, Producers: []string{"taskrun"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"run_id"}},
		{EventName: EventTaskRunCancelled, SchemaVersion: 1, Enabled: true, Producers: []string{"taskrun"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"run_id"}},
		{EventName: EventFAVArtifactStaged, SchemaVersion: 1, Enabled: false, Producers: []string{"app/docflow"}, Consumers: []string{ConsumerFAV}, PayloadFields: []string{"doc_id", "snapshot_id"}},
		{EventName: EventIntentPlanAwaiting, SchemaVersion: 1, Enabled: true, Producers: []string{"app/intentflow", "events"}, Consumers: []string{ConsumerProjection, ConsumerNotify}, PayloadFields: []string{"plan_id", "revision", "status", "environment", "project_id", "user_id", "session_id", "summary_ref"}},
		{EventName: EventIntentPlanStepChanged, SchemaVersion: 1, Enabled: true, Producers: []string{"app/intentflow", "events"}, Consumers: []string{ConsumerProjection, ConsumerNotify}, PayloadFields: []string{"plan_id", "revision", "step_id", "status", "environment", "project_id", "user_id", "session_id", "summary_ref"}},
		{EventName: EventIntentPlanCompleted, SchemaVersion: 1, Enabled: true, Producers: []string{"app/intentflow", "events"}, Consumers: []string{ConsumerProjection, ConsumerNotify}, PayloadFields: []string{"plan_id", "revision", "status", "environment", "project_id", "user_id", "session_id", "summary_ref"}},
		{EventName: EventDocProposalRequired, SchemaVersion: 1, Enabled: true, Producers: []string{"docgov", "bizdev", "app/supplementimpact", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "proposal_id", "project_id"}},
		{EventName: EventDocUnclaimedEdit, SchemaVersion: 1, Enabled: true, Producers: []string{"docgov", "bizdev", "app/supplementimpact", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "project_id"}},
	}
}

func contractByName(name string) (Contract, bool) {
	for _, c := range StaticContracts() {
		if c.EventName == name {
			return c, true
		}
	}
	return Contract{}, false
}

func containsName(list []string, want string) bool {
	for _, n := range list {
		if n == want {
			return true
		}
	}
	return false
}

// AssertLiveConsumers rejects disabled (FAV) and unregistered consumers.
func AssertLiveConsumers(eventName string, consumers []string) error {
	c, ok := contractByName(eventName)
	if !ok {
		return fmt.Errorf("%w: %s", ErrUnregistered, eventName)
	}
	if !c.Enabled {
		return fmt.Errorf("%w: %s", ErrDisabledConsumer, eventName)
	}
	for _, cons := range consumers {
		if cons == ConsumerFAV || strings.EqualFold(cons, "fav") {
			return fmt.Errorf("%w: fav", ErrDisabledConsumer)
		}
		if !containsName(c.Consumers, cons) {
			return fmt.Errorf("%w: %s for %s", ErrUnregistered, cons, eventName)
		}
	}
	return nil
}

// LiveConsumers returns enabled consumers for a producer append.
func LiveConsumers(eventName string) []string {
	c, ok := contractByName(eventName)
	if !ok || !c.Enabled {
		return nil
	}
	out := make([]string, 0, len(c.Consumers))
	for _, n := range c.Consumers {
		if n == ConsumerFAV {
			continue
		}
		out = append(out, n)
	}
	return out
}

// Registry is the runtime handler table. It must be a subset of the static list.
type Registry struct {
	handlers map[string]struct{}
}

func NewRegistry() *Registry {
	return &Registry{handlers: map[string]struct{}{}}
}

func (r *Registry) Register(consumer string) error {
	if r == nil {
		return ErrUnknownConsumer
	}
	if consumer == ConsumerFAV || strings.EqualFold(consumer, "fav") {
		return fmt.Errorf("%w: fav", ErrDisabledConsumer)
	}
	allowed := false
	for _, c := range StaticContracts() {
		if containsName(c.Consumers, consumer) && c.Enabled {
			allowed = true
			break
		}
	}
	if !allowed {
		return fmt.Errorf("%w: %s", ErrUnregistered, consumer)
	}
	r.handlers[consumer] = struct{}{}
	return nil
}

// ManifestJSON is the read-only registry dump. Runtime handlers must be a subset.
func ManifestJSON() ([]byte, error) {
	type row struct {
		EventName     string   `json:"event_name"`
		SchemaVersion int      `json:"schema_version"`
		Enabled       bool     `json:"enabled"`
		Producers     []string `json:"producers"`
		Consumers     []string `json:"consumers"`
		PayloadFields []string `json:"payload_fields"`
	}
	out := make([]row, 0, len(StaticContracts()))
	for _, c := range StaticContracts() {
		out = append(out, row{c.EventName, c.SchemaVersion, c.Enabled, c.Producers, c.Consumers, c.PayloadFields})
	}
	return json.Marshal(out)
}