package run

import (
	"fmt"
	"go/ast"
	"go/parser"
	"go/token"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"testing"

	"github.com/stretchr/testify/require"
)

// This file replaces the earlier self-mirroring literal producer list with a
// mechanical AST/typed bijection: it independently discovers the event kinds the
// real producers emit by parsing their source, then proves the per-kind public
// schema registry covers every produced kind (and nothing stale), that kernel /
// dispatcher failure producers are marked failure-capable (and nothing that is
// not failure-produced is), and that the concrete reviewed field corrections
// hold. A new producer kind, a changed failure classification, or a stale schema
// entry fails this test without any hand-maintained list to drift.

// producerEventKind records one producer-emitted event kind and whether it is
// produced through a failure path (runkernel.FailureEvent or
// failureEffectEvent), which requires the schema to be failure-capable.
type producerEventKind struct {
	kind           string
	failureCapable bool
}

// discoverProducerEventKinds parses the workflow runtime producer packages and
// returns the exhaustive set of event kinds they emit. It is the mechanical
// source of truth; there is NO hand-written literal list to keep in sync.
func discoverProducerEventKinds(t *testing.T) []producerEventKind {
	t.Helper()
	seen := map[string]bool{}
	var result []producerEventKind
	record := func(kind string, failureCapable bool) {
		trimmed := strings.TrimSpace(kind)
		if trimmed == "" {
			return
		}
		if !seen[trimmed] {
			seen[trimmed] = true
			result = append(result, producerEventKind{kind: trimmed, failureCapable: failureCapable})
			return
		}
		if failureCapable {
			for index := range result {
				if result[index].kind == trimmed {
					result[index].failureCapable = true
				}
			}
		}
	}

	scanDir(t, "internal/workflow/run", func(file *ast.File, consts map[string]string) {
		ast.Inspect(file, func(node ast.Node) bool {
			switch typed := node.(type) {
			case *ast.CompositeLit:
				// Event{Kind: "..."} and []Event{{Kind: "..."}, ...} (elided
				// elements in a typed slice).
				switch compositeLitTypeName(typed) {
				case "Event":
					if kind := compositeLitFieldString(typed, "Kind", consts); kind != "" {
						record(kind, compositeLitHasFailureFields(typed))
					}
				case "EventSlice":
					for _, element := range typed.Elts {
						if inner, ok := element.(*ast.CompositeLit); ok {
							if kind := compositeLitFieldString(inner, "Kind", consts); kind != "" {
								record(kind, compositeLitHasFailureFields(inner))
							}
						}
					}
				}
			case *ast.CallExpr:
				// FailureEvent("kind", ...) is the deterministic reducer failure path.
				if ident, ok := typed.Fun.(*ast.Ident); ok && ident.Name == "FailureEvent" {
					if kind := firstCallStringArg(typed, consts); kind != "" {
						record(kind, true)
					}
				}
			}
			return true
		})
	})

	scanDir(t, "internal/workflow/orchestrator", func(file *ast.File, consts map[string]string) {
		ast.Inspect(file, func(node ast.Node) bool {
			switch typed := node.(type) {
			case *ast.CompositeLit:
				// EventDraft{Type: "..."} (plain or store-qualified) and
				// []EventDraft{{Type: "..."}, ...} (elided slice elements), plus
				// runkernel.Event{Kind: ...} emitted by the engine/callback layers.
				switch compositeLitTypeName(typed) {
				case "EventDraft":
					if kind := compositeLitFieldString(typed, "Type", consts); kind != "" {
						record(kind, compositeLitHasFailureFields(typed))
					}
				case "EventDraftSlice":
					for _, element := range typed.Elts {
						if inner, ok := element.(*ast.CompositeLit); ok {
							if kind := compositeLitFieldString(inner, "Type", consts); kind != "" {
								record(kind, compositeLitHasFailureFields(inner))
							}
						}
					}
				case "Event":
					if kind := compositeLitFieldString(typed, "Kind", consts); kind != "" {
						record(kind, compositeLitHasFailureFields(typed))
					}
				case "EventSlice":
					for _, element := range typed.Elts {
						if inner, ok := element.(*ast.CompositeLit); ok {
							if kind := compositeLitFieldString(inner, "Kind", consts); kind != "" {
								record(kind, compositeLitHasFailureFields(inner))
							}
						}
					}
				}
			case *ast.CallExpr:
				// effectEvent("kind", ...) / failureEffectEvent("kind", ...)
				if ident, ok := typed.Fun.(*ast.Ident); ok {
					switch ident.Name {
					case "effectEvent":
						if kind := firstCallStringArg(typed, consts); kind != "" {
							record(kind, false)
						}
					case "failureEffectEvent":
						if kind := firstCallStringArg(typed, consts); kind != "" {
							record(kind, true)
						}
					}
				}
			}
			return true
		})
	})

	// The store layer also emits durable run events (e.g. run_canceled).
	scanDir(t, "internal/store/workflow/run", func(file *ast.File, consts map[string]string) {
		ast.Inspect(file, func(node ast.Node) bool {
			if typed, ok := node.(*ast.CompositeLit); ok {
				switch compositeLitTypeName(typed) {
				case "EventDraft":
					if kind := compositeLitFieldString(typed, "Type", consts); kind != "" {
						record(kind, compositeLitHasFailureFields(typed))
					}
				case "EventDraftSlice":
					for _, element := range typed.Elts {
						if inner, ok := element.(*ast.CompositeLit); ok {
							if kind := compositeLitFieldString(inner, "Type", consts); kind != "" {
								record(kind, compositeLitHasFailureFields(inner))
							}
						}
					}
				}
			}
			return true
		})
	})

	// The invocation-acceptance ledger entry run_accepted is produced in
	// internal/store/workflow/invocation (not the workflowrun store dir) through
	// `Type: domain.RunEventAccepted`, a cross-package constant the AST scanner
	// cannot resolve. It is reviewed here as a first-class non-failure store
	// event kind so the public schema bijection stays exact.
	for _, kind := range reviewedStoreEventKindsOutsideWorkflowrun {
		record(kind, false)
	}

	sort.Slice(result, func(i, j int) bool { return result[i].kind < result[j].kind })
	return result
}

// reviewedStoreEventKindsOutsideWorkflowrun is the small reviewed set of store
// event kinds emitted outside internal/store/workflow/run (the scanned dir) whose
// Type is a cross-package constant the AST scanner cannot resolve. run_accepted
// is the invocation-acceptance ledger entry (sequence 1) written by
// internal/store/workflow/invocation.
var reviewedStoreEventKindsOutsideWorkflowrun = []string{"run_accepted"}

// scanDir parses every non-test .go file under dir (relative to the repository
// root, resolved from the package working directory), collects its const string
// declarations, and calls visit for each.
func scanDir(t *testing.T, dir string, visit func(*ast.File, map[string]string)) {
	t.Helper()
	root := repoRootForTest(t)
	absDir := filepath.Join(root, dir)
	entries, err := os.ReadDir(absDir)
	if err != nil {
		t.Fatalf("read producer dir %s: %v", dir, err)
	}
	for _, entry := range entries {
		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".go") || strings.HasSuffix(entry.Name(), "_test.go") {
			continue
		}
		path := filepath.Join(absDir, entry.Name())
		file, err := parser.ParseFile(token.NewFileSet(), path, nil, 0)
		if err != nil {
			t.Fatalf("parse producer %s: %v", path, err)
		}
		visit(file, fileStringConsts(file))
	}
}

// fileStringConsts returns the string-valued constants declared in one file,
// keyed by identifier name, so composite-literal `Kind: initialRunStartedEvent`
// can resolve the producer kind.
func fileStringConsts(file *ast.File) map[string]string {
	consts := map[string]string{}
	for _, declaration := range file.Decls {
		gen, ok := declaration.(*ast.GenDecl)
		if !ok || gen.Tok != token.CONST {
			continue
		}
		for _, spec := range gen.Specs {
			valueSpec, ok := spec.(*ast.ValueSpec)
			if !ok {
				continue
			}
			for index, name := range valueSpec.Names {
				if index >= len(valueSpec.Values) {
					continue
				}
				if lit, ok := valueSpec.Values[index].(*ast.BasicLit); ok && lit.Kind == token.STRING {
					if unquoted, err := strconv.Unquote(lit.Value); err == nil {
						consts[name.Name] = unquoted
					}
				}
			}
		}
	}
	return consts
}

// repoRootForTest walks up from the package working directory to the repository
// root (the directory containing go.mod).
func repoRootForTest(t *testing.T) string {
	t.Helper()
	dir, err := os.Getwd()
	if err != nil {
		t.Fatalf("getwd: %v", err)
	}
	for {
		if _, statErr := os.Stat(filepath.Join(dir, "go.mod")); statErr == nil {
			return dir
		}
		parent := filepath.Dir(dir)
		if parent == dir {
			t.Fatalf("repository root (go.mod) not found from %s", dir)
		}
		dir = parent
	}
}

// compositeLitTypeName returns the base type name of a composite literal, for
// `Event{...}` (Ident), `runstore.EventDraft{...}` (SelectorExpr), and
// `[]Event{...}` / `[]runstore.EventDraft{...}` (ArrayType), mapping the
// slice forms to "<Base>Slice" so the scanner can walk their elided elements.
func compositeLitTypeName(lit *ast.CompositeLit) string {
	switch typed := lit.Type.(type) {
	case *ast.Ident:
		return typed.Name
	case *ast.SelectorExpr:
		return typed.Sel.Name
	case *ast.ArrayType:
		switch element := typed.Elt.(type) {
		case *ast.Ident:
			return element.Name + "Slice"
		case *ast.SelectorExpr:
			return element.Sel.Name + "Slice"
		}
	}
	return ""
}

// compositeLitFieldString returns the literal string of a composite-literal
// field value (resolving a const identifier), or "" when the field is absent or
// not a resolvable string.
func compositeLitFieldString(lit *ast.CompositeLit, field string, consts map[string]string) string {
	for _, element := range lit.Elts {
		kv, ok := element.(*ast.KeyValueExpr)
		if !ok {
			continue
		}
		key, ok := kv.Key.(*ast.Ident)
		if !ok || key.Name != field {
			continue
		}
		if value, ok := kv.Value.(*ast.BasicLit); ok && value.Kind == token.STRING {
			if unquoted, err := strconv.Unquote(value.Value); err == nil {
				return unquoted
			}
		}
		if ident, ok := kv.Value.(*ast.Ident); ok {
			if resolved, ok := consts[ident.Name]; ok {
				return resolved
			}
		}
	}
	return ""
}

// compositeLitHasFailureFields reports whether a producer Event/EventDraft
// composite literal embeds a failure contract field (Code/Stage/Path/Problem/
// Hint/Retryable/Diagnostics/WorkItem/Truncation/Failure), making the kind
// failure-capable even though it was not emitted through FailureEvent/
// failureEffectEvent.
func compositeLitHasFailureFields(lit *ast.CompositeLit) bool {
	for _, element := range lit.Elts {
		kv, ok := element.(*ast.KeyValueExpr)
		if !ok {
			continue
		}
		key, ok := kv.Key.(*ast.Ident)
		if !ok {
			continue
		}
		switch key.Name {
		case "Code", "Stage", "Path", "Problem", "Hint", "Retryable", "Diagnostics", "WorkItem", "Truncation", "Failure":
			return true
		}
	}
	return false
}

// firstCallStringArg returns the first string-literal (or const) argument of a
// call, or "".
func firstCallStringArg(call *ast.CallExpr, consts map[string]string) string {
	if len(call.Args) == 0 {
		return ""
	}
	if value, ok := call.Args[0].(*ast.BasicLit); ok && value.Kind == token.STRING {
		if unquoted, err := strconv.Unquote(value.Value); err == nil {
			return unquoted
		}
	}
	if ident, ok := call.Args[0].(*ast.Ident); ok {
		if resolved, ok := consts[ident.Name]; ok {
			return resolved
		}
	}
	return ""
}

// producerKindSchemaBijectionViolations runs the exact kind bijection between
// the mechanically discovered producer kinds and a schema registry snapshot,
// returning a violation list (empty = the bijection holds).
func producerKindSchemaBijectionViolations(produced []producerEventKind, schemas map[string]eventEnvelopeSchema) []string {
	var violations []string
	producedSet := map[string]bool{}
	for _, p := range produced {
		producedSet[p.kind] = true
	}

	// Every producer-emitted kind must have a closed schema.
	for _, p := range produced {
		schema, ok := schemas[p.kind]
		if !ok {
			violations = append(violations, fmt.Sprintf("producer event kind %q has no closed public schema", p.kind))
			continue
		}
		if p.failureCapable && !schema.failureAllowed {
			violations = append(violations, fmt.Sprintf("failure-produced kind %q must be failure-capable", p.kind))
		}
	}

	// Every schema entry must be producer-emitted (stale schema entries fail).
	for kind := range schemas {
		if !producedSet[kind] {
			violations = append(violations, fmt.Sprintf("schema kind %q is not emitted by a producer (stale entry)", kind))
		}
	}
	return violations
}

// TestProducerKindSchemaBijection proves the per-kind schema registry is the
// exact bijection of producer-emitted event kinds: every producer kind has a
// schema (no silent `{}` projection), every schema entry is producer-emitted (no
// stale registry drift), and every kernel/dispatcher failure-produced kind is
// failure-capable (never a lost failure contract). The producer side is derived
// mechanically from source, not mirrored from the schema.
func TestProducerKindSchemaBijection(t *testing.T) {
	require.Empty(t, producerKindSchemaBijectionViolations(discoverProducerEventKinds(t), eventEnvelopeSchemas))
}

// TestProducerKindSchemaBijectionMutationFixtures proves the kind bijection is
// executable: a stale schema kind or a dropped producer kind produces
// violations.
func TestProducerKindSchemaBijectionMutationFixtures(t *testing.T) {
	produced := discoverProducerEventKinds(t)

	// Stale schema kind: a schema entry with no producer fails.
	stale := cloneEnvelopeSchemas()
	stale["invented_kind"] = eventEnvelopeSchema{context: contextFields("kind")}
	require.NotEmpty(t, producerKindSchemaBijectionViolations(produced, stale),
		"a stale schema kind must fail the kind bijection")

	// Dropped producer kind: removing a produced kind's schema fails.
	dropped := cloneEnvelopeSchemas()
	delete(dropped, "message_sent")
	require.NotEmpty(t, producerKindSchemaBijectionViolations(produced, dropped),
		"a produced kind without a schema must fail the kind bijection")
}

// TestConcreteReviewedEventFieldCorrections pins the exact reviewed field-shape
// corrections: the producer-emitted fields for parallel/fan-out/loop/approval
// events and the non-failure EffectResult contract. These are the concrete
// Review defects closed end-to-end.
func TestConcreteReviewedEventFieldCorrections(t *testing.T) {
	schemaFor := func(kind string) eventEnvelopeSchema {
		schema := eventKindSchema(kind)
		require.NotNil(t, schema, "kind %q must have a schema", kind)
		return *schema
	}

	// parallel_opened: producer emits {activationId, scopeId}.
	parallel := schemaFor("parallel_opened")
	require.Contains(t, parallel.context, "activationId", "parallel_opened must carry the producer activationId")
	require.Contains(t, parallel.context, "scopeId")
	require.False(t, parallel.failureAllowed)

	// fan_out_opened: producer emits {activationId, nodeId, scopeId}.
	fanOutOpened := schemaFor("fan_out_opened")
	require.Contains(t, fanOutOpened.context, "nodeId", "fan_out_opened must carry the producer nodeId")
	require.Contains(t, fanOutOpened.context, "activationId")
	require.Contains(t, fanOutOpened.context, "scopeId")
	require.False(t, fanOutOpened.failureAllowed)

	// fan_out_resolved: producer emits {scopeId, port}.
	fanOutResolved := schemaFor("fan_out_resolved")
	require.Contains(t, fanOutResolved.context, "port", "fan_out_resolved must carry the producer port")
	require.Contains(t, fanOutResolved.context, "scopeId")
	require.False(t, fanOutResolved.failureAllowed)

	// loop_broken: producer emits {activationId, scopeId} with NO failure.
	loopBroken := schemaFor("loop_broken")
	require.Contains(t, loopBroken.context, "activationId")
	require.Contains(t, loopBroken.context, "scopeId")
	require.False(t, loopBroken.failureAllowed, "loop_broken is a plain event, never failure-capable")

	// approval_decided: producer emits {effectId, activationId, nodeId,
	// notificationId, decision, decidedByMemberId} — never notificationRevision
	// or completedAt (non-producer fields).
	approvalDecided := schemaFor("approval_decided")
	for _, field := range []string{"effectId", "activationId", "nodeId", "notificationId", "decision", "decidedByMemberId"} {
		require.Contains(t, approvalDecided.context, field, "approval_decided must carry producer field %q", field)
	}
	require.NotContains(t, approvalDecided.context, "notificationRevision", "approval_decided must not invent notificationRevision")
	require.NotContains(t, approvalDecided.context, "completedAt", "approval_decided must not invent completedAt")

	// EffectResult non-failure projection must match orchestrator.EffectResult
	// exactly: no invented kind/status/retryAt/receiptRecovery fields.
	for _, invented := range []string{"kind", "status", "retryAt", "receiptRecovery"} {
		_, present := effectResultNonFailureContext[invented]
		require.False(t, present, "EffectResult non-failure projection must not invent %q", invented)
	}
	for _, real := range []string{"port", "messageId", "notificationId", "notificationRevision", "completedAt", "assigneeMemberId", "decidedByMemberId", "callbackResponderAgentId", "wakeAgentIds", "deliveryScope", "threadScope", "workScope"} {
		_, present := effectResultNonFailureContext[real]
		require.True(t, present, "EffectResult non-failure projection must keep real field %q", real)
	}

	// The authoritative Plan port grammar rejects a built-in fixed port
	// vocabulary: there is no eventEnvelopePortValues, and ports project through
	// the Plan grammar.
	require.NotContains(t, []string{"completed", "failed", "timed_out", "approved", "rejected"},
		"eventEnvelopePortValues", "dead fixed port vocabulary must be removed")
}

// producerFailureCapabilityViolations proves every schema entry marked
// failure-capable is either (a) produced through a deterministic failure path
// (runkernel.FailureEvent or failureEffectEvent), or (b) a hand-built EventDraft
// kind whose producer embeds a failure contract in the payload
// (reviewedHandBuiltFailureCapableKinds). A failure-capable marker for any other
// kind is stale drift and fails.
func producerFailureCapabilityViolations(produced []producerEventKind, schemas map[string]eventEnvelopeSchema) []string {
	var violations []string
	failureProduced := map[string]bool{}
	producedSet := map[string]bool{}
	for _, p := range produced {
		producedSet[p.kind] = true
		if p.failureCapable {
			failureProduced[p.kind] = true
		}
	}
	for kind, schema := range schemas {
		if !schema.failureAllowed {
			continue
		}
		if !producedSet[kind] {
			violations = append(violations, fmt.Sprintf("failure-capable kind %q is not producer-emitted", kind))
			continue
		}
		if failureProduced[kind] {
			continue
		}
		if !reviewedHandBuiltFailureCapableKinds[kind] {
			violations = append(violations, fmt.Sprintf(
				"failure-capable kind %q is not failure-produced (FailureEvent/failureEffectEvent) nor a reviewed hand-built failure EventDraft", kind))
		}
	}
	return violations
}

// TestProducerKindBijectionNoStaleFailureCapable proves every schema entry
// marked failure-capable is either (a) produced through a deterministic failure
// path (runkernel.FailureEvent or failureEffectEvent), or (b) a hand-built
// EventDraft kind whose producer embeds a failure contract in the payload
// (reviewedHandBuiltFailureCapableKinds). A failure-capable marker for any other
// kind is stale drift and fails.
func TestProducerKindBijectionNoStaleFailureCapable(t *testing.T) {
	require.Empty(t, producerFailureCapabilityViolations(discoverProducerEventKinds(t), eventEnvelopeSchemas))
}

// TestProducerKindBijectionNoStaleFailureCapableMutationFixture proves the
// failure-capability check is executable: a schema that marks a
// non-failure-producing kind failure-capable fails.
func TestProducerKindBijectionNoStaleFailureCapableMutationFixture(t *testing.T) {
	produced := discoverProducerEventKinds(t)
	drifted := cloneEnvelopeSchemas()
	drifted["message_sent"] = eventEnvelopeSchema{context: contextFields("kind"), failureAllowed: true}
	require.NotEmpty(t, producerFailureCapabilityViolations(produced, drifted),
		"a non-failure-producing kind marked failure-capable must fail")
}

// reviewedHandBuiltFailureCapableKinds is the small reviewed set of event kinds
// whose producer embeds a failure contract in a hand-built EventDraft payload
// (the AST cannot mechanically infer failure capability from a plain EventDraft
// literal). The callback completion event carries a failure when the callback
// result failed. A NEW failureAllowed kind outside this set and the
// failure-produced set fails the bijection.
var reviewedHandBuiltFailureCapableKinds = map[string]bool{
	"message_callback_completed": true,
}

// producerFieldDiscovery is the mechanical per-kind emitted-field discovery.
// For kernel `Event{...}` literals the Go struct field names map to the JSON
// payload keys via eventGoFieldToJSONKey; for dispatch events the producer-owned
// ProducerEventFieldSets registry (internal/workflow/orchestrator) is the exact
// declared field set (including the base effect context and kind-specific
// extras); the store-layer EventDraft kinds whose payload is a variable are
// declared in reviewedStoreEventDraftFieldSets (the AST cannot see a map
// variable). The schema context (minus the always-present `kind`) must be the
// EXACT bijection of this discovered set for every schema entry.
type producerFieldDiscovery struct {
	kind   string
	fields map[string]bool
}

// eventGoFieldToJSONKey maps a producer Event struct field to its JSON payload
// key. Unknown fields (Code/Stage/Path/Problem/Hint/Retryable/Diagnostics/
// WorkItem/Truncation) are the failure namespace and handled separately.
func eventGoFieldToJSONKey(field string) string {
	switch field {
	case "ActivationID":
		return "activationId"
	case "NodeID":
		return "nodeId"
	case "AgentID":
		return "agentId"
	case "Port":
		return "port"
	case "ScopeID":
		return "scopeId"
	case "Kind":
		return "kind"
	}
	return ""
}

// discoverProducerEventFieldSets mechanically derives the emitted non-kind
// field set per producer event kind from the ACTUAL producer source: kernel
// Event struct literals and FailureEvent call arguments, plus the orchestrator
// dispatch event builders (effectEvent/failureEffectEvent call args and their
// extra map keys). It does NOT union any hand-maintained producer registry into
// discovery, so a synchronized registry+schema edit that invents a field (for
// example scopeId on run_failed, which no real call site emits) fails the
// schema bijection because the mechanical truth from the call sites disagrees.
// Genuinely dynamic producer variants (store EventDraft payloads and the
// callback/approval EventDraft payloads built from map/JSON variables the AST
// cannot resolve) are modeled explicitly in the reviewed sets below.
func discoverProducerEventFieldSets(t *testing.T) map[string]producerFieldDiscovery {
	t.Helper()
	discovered := map[string]producerFieldDiscovery{}
	record := func(kind string, fields map[string]bool) {
		if existing, ok := discovered[kind]; ok {
			for field := range fields {
				existing.fields[field] = true
			}
			discovered[kind] = existing
			return
		}
		discovered[kind] = producerFieldDiscovery{kind: kind, fields: fields}
	}

	// Kernel Event literals: mechanical field discovery from the Go struct.
	scanDir(t, "internal/workflow/run", func(file *ast.File, consts map[string]string) {
		ast.Inspect(file, func(node ast.Node) bool {
			switch typed := node.(type) {
			case *ast.CompositeLit:
				switch compositeLitTypeName(typed) {
				case "Event", "EventSlice":
					collectEventLiteralFields(t, typed, consts, record)
				}
			case *ast.CallExpr:
				// FailureEvent("kind", failure, scopeID, activationID, nodeID)
				// emits scopeId/activationId/nodeId (only when the arg is
				// non-empty) plus the failure namespace.
				if ident, ok := typed.Fun.(*ast.Ident); ok && ident.Name == "FailureEvent" {
					if kind := firstCallStringArg(typed, consts); kind != "" {
						record(kind, failureEventEmittedFields(typed))
					}
				}
			}
			return true
		})
	})

	// orchestrator producers: kernel Event literals and FailureEvent calls in the
	// engine/callback layers, plus the effectEvent/failureEffectEvent dispatch
	// builders. All discovered mechanically from the actual call sites.
	scanDir(t, "internal/workflow/orchestrator", func(file *ast.File, consts map[string]string) {
		ast.Inspect(file, func(node ast.Node) bool {
			switch typed := node.(type) {
			case *ast.CompositeLit:
				switch compositeLitTypeName(typed) {
				case "Event", "EventSlice":
					collectEventLiteralFields(t, typed, consts, record)
				}
			case *ast.CallExpr:
				if ident, ok := typed.Fun.(*ast.Ident); ok {
					switch ident.Name {
					case "FailureEvent":
						if kind := firstCallStringArg(typed, consts); kind != "" {
							record(kind, failureEventEmittedFields(typed))
						}
					case "effectEvent":
						if kind := firstCallStringArg(typed, consts); kind != "" {
							record(kind, dispatchEventEmittedFields(typed, false))
						}
					case "failureEffectEvent":
						if kind := firstCallStringArg(typed, consts); kind != "" {
							record(kind, dispatchEventEmittedFields(typed, true))
						}
					}
				}
			}
			return true
		})
	})

	// Genuinely dynamic producer variants: store-layer EventDraft kinds whose
	// payload is a map/JSON variable, and the callback/approval EventDraft
	// payloads built from helper/map writes the AST cannot resolve. These are
	// reviewed against the schema explicitly; they cannot drift mechanically.
	for kind, fields := range reviewedStoreEventDraftFieldSets {
		record(kind, fields)
	}
	for kind, fields := range reviewedDispatchEventDraftFieldSets {
		record(kind, fields)
	}
	return discovered
}

// collectEventLiteralFields extracts the non-kind JSON fields set on one Event
// composite literal (and the elements of an EventSlice literal).
func collectEventLiteralFields(t *testing.T, lit *ast.CompositeLit, consts map[string]string, record func(string, map[string]bool)) {
	t.Helper()
	if compositeLitTypeName(lit) == "EventSlice" {
		for _, element := range lit.Elts {
			if inner, ok := element.(*ast.CompositeLit); ok {
				collectEventLiteralFields(t, inner, consts, record)
			}
		}
		return
	}
	kind := compositeLitFieldString(lit, "Kind", consts)
	if kind == "" {
		return
	}
	fields := map[string]bool{}
	for _, element := range lit.Elts {
		kv, ok := element.(*ast.KeyValueExpr)
		if !ok {
			continue
		}
		key, ok := kv.Key.(*ast.Ident)
		if !ok {
			continue
		}
		if jsonKey := eventGoFieldToJSONKey(key.Name); jsonKey != "" && jsonKey != "kind" {
			fields[jsonKey] = true
		}
	}
	if len(fields) > 0 {
		record(kind, fields)
	}
}

// reviewedStoreEventDraftFieldSets is the small reviewed supplement for
// store-layer EventDraft kinds whose payload is built from a map/JSON variable
// that the AST literal scanner cannot see. The exact payload keys are reviewed
// against the schema here so they cannot silently drift.
var reviewedStoreEventDraftFieldSets = map[string]map[string]bool{
	"run_accepted":                          {"status": true},
	"run_started":                           {},
	"run_canceled":                          {"reason": true, "actor": true},
	"run_failed":                            {"code": true, "stage": true, "path": true, "problem": true, "hint": true, "retryable": true, "diagnostics": true, "truncation": true, "workItem": true},
	"effect_waiting_for_responder_capacity": {"effectId": true, "activationId": true, "nodeId": true, "kind": true, "waitingReason": true},
	"effect_responder_selected":             {"effectId": true, "activationId": true, "nodeId": true, "kind": true, "waitingReason": true},
	"effect_responder_promoted":             {"effectId": true, "activationId": true, "nodeId": true, "kind": true, "waitingReason": true},
}

// reviewedDispatchEventDraftFieldSets is the small reviewed supplement for
// orchestrator dispatch EventDraft kinds whose payload is built inside a helper
// (callbackCompletionEventPayload) or from a marshaled map literal
// (approval_decided), so the AST call-site scanner cannot see the payload keys
// at the effectEvent/failureEffectEvent boundary. The exact payload keys are
// reviewed against the schema here so they cannot silently drift.
var reviewedDispatchEventDraftFieldSets = map[string]map[string]bool{
	"message_callback_completed": {
		"effectId": true, "activationId": true, "nodeId": true, "agentId": true, "port": true,
	},
	"approval_decided": {
		"effectId": true, "activationId": true, "nodeId": true,
		"notificationId": true, "decision": true, "decidedByMemberId": true,
	},
}

// dispatchEventEmittedFields returns the base effect context field set plus the
// kind-specific extra map-literal keys passed to one effectEvent /
// failureEffectEvent dispatch producer call. The base effect payload always
// carries effectId/activationId/nodeId; the extra map literal keys are the
// kind-specific context fields (for example messageId/deliveryScope/threadScope
// on message_sent). failureEffectEvent calls additionally carry the shared
// failure namespace. `kind` is the intent-kind payload field handled by the
// always-present schema `kind` validator, so it is excluded here.
func dispatchEventEmittedFields(call *ast.CallExpr, failureCapable bool) map[string]bool {
	fields := map[string]bool{"effectId": true, "activationId": true, "nodeId": true}
	if failureCapable {
		for field := range failureNamespaceFields {
			fields[field] = true
		}
	}
	// effectEvent(kind, effect, request, extra) / failureEffectEvent(kind, effect,
	// request, result, extra): the extra map literal is the last argument.
	extraIndex := 3
	if failureCapable {
		extraIndex = 4
	}
	if extraIndex >= len(call.Args) {
		return fields
	}
	if lit, ok := call.Args[extraIndex].(*ast.CompositeLit); ok {
		for _, element := range lit.Elts {
			kv, ok := element.(*ast.KeyValueExpr)
			if !ok {
				continue
			}
			key, ok := kv.Key.(*ast.BasicLit)
			if !ok || key.Kind != token.STRING {
				continue
			}
			if name, err := strconv.Unquote(key.Value); err == nil {
				fields[name] = true
			}
		}
	}
	return fields
}

// failureEventEmittedFields returns the context + failure-namespace field set
// emitted by one runkernel.FailureEvent call. The context fields scopeId/
// activationId/nodeId are only included when the corresponding call argument is
// a NON-EMPTY string literal or an identifier/expression (conservatively
// present); an explicit empty-string literal means the producer did not emit it.
func failureEventEmittedFields(call *ast.CallExpr) map[string]bool {
	fields := map[string]bool{
		"code": true, "stage": true, "path": true, "problem": true, "hint": true,
		"retryable": true, "diagnostics": true, "truncation": true, "workItem": true,
	}
	// Args: [0] kind, [1] failure, [2] scopeID, [3] activationID, [4] nodeID.
	contextArgs := []struct {
		field string
		index int
	}{
		{"scopeId", 2},
		{"activationId", 3},
		{"nodeId", 4},
	}
	for _, arg := range contextArgs {
		if arg.index >= len(call.Args) {
			continue
		}
		if lit, ok := call.Args[arg.index].(*ast.BasicLit); ok && lit.Kind == token.STRING {
			if unquoted, err := strconv.Unquote(lit.Value); err == nil && unquoted == "" {
				continue // explicit empty: not emitted
			}
		}
		fields[arg.field] = true
	}
	return fields
}

// failureNamespaceFields are projected through the shared typed failure
// projector, never through the context allowlist.
var failureNamespaceFields = map[string]bool{
	"code": true, "stage": true, "path": true, "problem": true, "hint": true,
	"retryable": true, "diagnostics": true, "truncation": true, "workItem": true,
}

// producerFieldSchemaBijectionViolations runs the two-way per-kind field
// bijection and returns a human-readable violation list (empty = the exact
// bijection holds). It is the executable invariant: a synchronized
// registry+schema edit that invents a field no real call site emits (for
// example scopeId on run_failed) fails the producer-emitted direction because
// the mechanical discovery from the call sites disagrees.
func producerFieldSchemaBijectionViolations(schemas map[string]eventEnvelopeSchema, discovered map[string]producerFieldDiscovery) []string {
	var violations []string
	for kind, schema := range schemas {
		produced := discovered[kind]
		// A kind-only schema ({kind} only) has no producer-emitted context fields
		// by definition; the discovery may legitimately carry no entry.
		if produced.fields == nil {
			if len(schema.context) != 1 {
				violations = append(violations, fmt.Sprintf("kind %q has no discovered producer fields but schema carries %d context fields", kind, len(schema.context)))
			}
			if _, hasKind := schema.context["kind"]; !hasKind {
				violations = append(violations, fmt.Sprintf("kind %q with no discovered fields must carry `kind`", kind))
			}
			continue
		}
		if len(produced.fields) == 0 && len(schema.context) != 1 {
			violations = append(violations, fmt.Sprintf("kind %q produced no fields but schema carries %d context fields", kind, len(schema.context)))
		}

		// Both directions: schema context (minus kind and failure-namespace) ==
		// produced context fields.
		for field := range produced.fields {
			if failureNamespaceFields[field] {
				if !schema.failureAllowed {
					violations = append(violations, fmt.Sprintf("failure-namespace field %q for kind %q requires failureAllowed", field, kind))
				}
				continue
			}
			if _, hasValidator := schema.context[field]; !hasValidator {
				violations = append(violations, fmt.Sprintf("producer-emitted context field %q for kind %q has no schema validator (silent drop)", field, kind))
			}
		}
		for field := range schema.context {
			if field == "kind" {
				continue
			}
			if failureNamespaceFields[field] {
				continue
			}
			if !produced.fields[field] {
				violations = append(violations, fmt.Sprintf("schema context field %q for kind %q is not producer-emitted (invented field)", field, kind))
			}
		}
	}
	return violations
}

// TestProducerFieldSchemaBijection proves the per-kind schema context field set
// is the EXACT bijection of the producer-emitted CONTEXT fields in BOTH
// directions: every producer-emitted context field has a schema validator (no
// silent drop), and every schema context field is producer-emitted (no invented
// field). The failure-namespace fields are projected through the shared typed
// projector (verified by TestProducerFieldSchemaBijectionFailureNamespace), not
// the context allowlist, so they are excluded from this comparison.
func TestProducerFieldSchemaBijection(t *testing.T) {
	discovered := discoverProducerEventFieldSets(t)
	require.Empty(t, producerFieldSchemaBijectionViolations(eventEnvelopeSchemas, discovered))
}

// TestProducerFieldSchemaBijectionMutationFixtures proves the two-way invariant
// is EXECUTABLE: a schema mutation that adds an invented field, drops a real
// producer field, adds a stale kind, or drops a producer kind produces
// violations, so a synchronized registry+schema edit cannot make production
// drift pass.
func TestProducerFieldSchemaBijectionMutationFixtures(t *testing.T) {
	discovered := discoverProducerEventFieldSets(t)

	// Invented field: scopeId on run_failed is not emitted by any real call site
	// (all FailureEvent run_failed call sites pass an empty scopeID arg), so a
	// schema that carries it must fail the producer-emitted direction.
	invented := cloneEnvelopeSchemas()
	invented["run_failed"] = eventEnvelopeSchema{
		context:        contextFields("kind", "activationId", "nodeId", "scopeId"),
		failureAllowed: true,
	}
	require.NotEmpty(t, producerFieldSchemaBijectionViolations(invented, discovered),
		"an invented run_failed.scopeId field must fail the producer-emitted bijection")

	// Missing field: dropping a real producer-emitted activationId from run_failed
	// fails the silent-drop direction.
	missing := cloneEnvelopeSchemas()
	missing["run_failed"] = eventEnvelopeSchema{
		context:        contextFields("kind", "nodeId"),
		failureAllowed: true,
	}
	require.NotEmpty(t, producerFieldSchemaBijectionViolations(missing, discovered),
		"a dropped producer field must fail the no-silent-drop bijection")

	// Invented field on a dispatch kind: scopeId is not a message_sent producer
	// field either.
	inventedDispatch := cloneEnvelopeSchemas()
	inventedDispatch["message_sent"] = eventEnvelopeSchema{context: contextFields(
		"effectId", "activationId", "nodeId", "kind", "messageId", "deliveryScope", "threadScope", "scopeId")}
	require.NotEmpty(t, producerFieldSchemaBijectionViolations(inventedDispatch, discovered),
		"an invented dispatch field must fail the producer-emitted bijection")
}

// cloneEnvelopeSchemas returns a deep-enough copy of the global schema registry
// (context maps copied) so mutation fixtures can edit a snapshot without
// disturbing the shared registry.
func cloneEnvelopeSchemas() map[string]eventEnvelopeSchema {
	cloned := make(map[string]eventEnvelopeSchema, len(eventEnvelopeSchemas))
	for kind, schema := range eventEnvelopeSchemas {
		contextCopy := make(map[string]func(any) (any, bool), len(schema.context))
		for field, validator := range schema.context {
			contextCopy[field] = validator
		}
		schemaCopy := schema
		schemaCopy.context = contextCopy
		cloned[kind] = schemaCopy
	}
	return cloned
}

// producerFieldSchemaBijectionFailureNamespaceViolations proves the failure
// namespace stays on the shared typed projector boundary: no failure-namespace
// field may be declared in any schema's context allowlist (which would bypass or
// duplicate the shared projector), and every failure-capable kind's produced
// field set must be able to reach the projector (the schema carries
// failureAllowed). This is the executable two-way failure-namespace invariant.
func producerFieldSchemaBijectionFailureNamespaceViolations(schemas map[string]eventEnvelopeSchema, discovered map[string]producerFieldDiscovery) []string {
	var violations []string
	for kind, schema := range schemas {
		// Direction 1: no failure-namespace field may be allowlisted in the
		// context map. A synchronized registry+schema edit that moves a failure
		// field into the context allowlist (or invents one there) fails.
		for field := range schema.context {
			if failureNamespaceFields[field] {
				violations = append(violations, fmt.Sprintf("failure-namespace field %q must not be allowlisted in kind %q context (shared projector boundary)", field, kind))
			}
		}
		// Direction 2: when the producer emits failure fields for a kind, the
		// schema must be failure-capable so the shared projector is reachable.
		if produced := discovered[kind]; produced.fields != nil {
			for field := range produced.fields {
				if failureNamespaceFields[field] && !schema.failureAllowed {
					violations = append(violations, fmt.Sprintf("producer-emitted failure-namespace field %q for kind %q requires failureAllowed", field, kind))
				}
			}
		}
	}
	return violations
}

// TestProducerFieldSchemaBijectionFailureNamespace proves the exact failure
// namespace stays on the shared typed projector boundary for every schema kind
// (both directions): no failure field leaks into a context allowlist, and every
// producer-emitted failure field reaches a failure-capable schema.
func TestProducerFieldSchemaBijectionFailureNamespace(t *testing.T) {
	discovered := discoverProducerEventFieldSets(t)
	require.Empty(t, producerFieldSchemaBijectionFailureNamespaceViolations(eventEnvelopeSchemas, discovered))
}

// TestProducerFieldSchemaBijectionFailureNamespaceMutationFixtures proves the
// failure-namespace boundary is executable: a schema that allowlists a
// failure-namespace field fails, and a non-failure-capable schema for a
// failure-producing kind fails.
func TestProducerFieldSchemaBijectionFailureNamespaceMutationFixtures(t *testing.T) {
	discovered := discoverProducerEventFieldSets(t)

	// Allowlist leak: moving code/stage into the context allowlist bypasses the
	// shared projector boundary and must fail. Mutate the context map directly
	// because contextFields() only accepts shared validators (failure fields have
	// none).
	leaked := cloneEnvelopeSchemas()
	leakedRunFailed := eventEnvelopeSchema{
		context:        contextFields("kind", "activationId", "nodeId"),
		failureAllowed: true,
	}
	leakedRunFailed.context["code"] = envelopeFieldCancelReason
	leakedRunFailed.context["stage"] = envelopeFieldCancelReason
	leaked["run_failed"] = leakedRunFailed
	require.NotEmpty(t, producerFieldSchemaBijectionFailureNamespaceViolations(leaked, discovered),
		"a failure-namespace field moved into a context allowlist must fail")

	// Non-failure-capable schema for a failure-producing kind must fail.
	nonCapable := cloneEnvelopeSchemas()
	nonCapable["effect_failed"] = eventEnvelopeSchema{
		context: contextFields("effectId", "activationId", "nodeId", "kind"),
	}
	require.NotEmpty(t, producerFieldSchemaBijectionFailureNamespaceViolations(nonCapable, discovered),
		"a non-failure-capable schema for a failure-producing kind must fail")
}