Agent-System/internal/app/intentflow/emit.go

package intentflow

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"chat-agent-system/internal/app/uow"
	"chat-agent-system/internal/events"
)

// EventSink is the Track E outbox port. Production uses events.PGSink.
type EventSink interface {
	Append(ctx context.Context, env events.Envelope, consumers []string) error
}

// CardProjector is the ctxmgr-facing port. Command UoW must not call it;
// projection consumers project cards after the plan is committed.
type CardProjector interface {
	ProjectCard(ctx context.Context, card CardProjection) error
}

// CardProjection is the desensitized fact a chat card may show.
type CardProjection struct {
	Principal  Principal
	PlanID     string
	Revision   int64
	Status     PlanStatus
	SummaryRef string
}

type planFact struct {
	PlanID      string `json:"plan_id"`
	Revision    int64  `json:"revision"`
	Status      string `json:"status"`
	Environment string `json:"environment"`
	ProjectID   int64  `json:"project_id"`
	UserID      int64  `json:"user_id"`
	SessionID   string `json:"session_id"`
	SummaryRef  string `json:"summary_ref"`
	StepID      string `json:"step_id,omitempty"`
}

func (p *Plans) withTx(ctx context.Context, fn func(ctx context.Context, scope uow.TxScope) error) error {
	if p == nil || p.uow == nil {
		return fmt.Errorf("intentflow: unit of work required")
	}
	return p.uow.Within(ctx, func(scope uow.TxScope) error {
		return fn(uow.WithScope(ctx, scope), scope)
	})
}

func (p *Plans) emitFact(ctx context.Context, plan *Plan, stepID string, now time.Time) error {
	if p == nil || p.events == nil || plan == nil {
		return nil
	}
	name := events.EventIntentPlanAwaiting
	if stepID != "" {
		name = events.EventIntentPlanStepChanged
	} else {
		switch plan.Status {
		case PlanCompleted, PlanCancelled, PlanExpired, PlanFailed, PlanSuperseded:
			name = events.EventIntentPlanCompleted
		}
	}
	fact := planFact{
		PlanID:      plan.PlanID,
		Revision:    plan.Revision,
		Status:      string(plan.Status),
		Environment: plan.Principal.Environment,
		ProjectID:   plan.Principal.ProjectID,
		UserID:      plan.Principal.UserID,
		SessionID:   plan.Principal.SessionID,
		SummaryRef:  plan.SafeSummary,
		StepID:      stepID,
	}
	body, err := json.Marshal(fact)
	if err != nil {
		return err
	}
	if err := AssertPlanPayload(name, body); err != nil {
		return err
	}
	eventID := fmt.Sprintf("intent-plan:%s:%s:%d:%s", plan.PlanID, name, plan.Revision, stepID)
	aggType := "intent_plan"
	aggID := plan.PlanID
	aggVer := plan.Revision
	if stepID != "" {
		// Step facts must not collide with plan-revision unique (aggregate_type, id, version).
		aggType = "intent_plan_step"
		aggID = stepID
		aggVer = 1
		for _, s := range plan.Steps {
			if s.StepID == stepID {
				if s.Version > 0 {
					aggVer = s.Version
				}
				break
			}
		}
	}
	env := events.Envelope{
		EventID: eventID, EventName: name, SchemaVersion: events.EnvelopeSchema,
		Environment: plan.Principal.Environment, ProjectID: plan.Principal.ProjectID,
		AggregateType: aggType, AggregateID: aggID, AggregateVersion: aggVer,
		ActorRef: fmt.Sprintf("user:%d", plan.Principal.UserID), IdempotencyKey: eventID,
		PayloadDigest: events.Digest(body), Producer: "app/intentflow",
		OccurredAt: now, Payload: body,
	}
	return p.events.Append(ctx, env, events.LiveConsumers(name))
}

func AssertPlanPayload(eventName string, body []byte) error {
	var raw map[string]any
	if err := json.Unmarshal(body, &raw); err != nil {
		return err
	}
	allowed := map[string]struct{}{}
	for _, c := range events.StaticContracts() {
		if c.EventName != eventName {
			continue
		}
		for _, f := range c.PayloadFields {
			allowed[f] = struct{}{}
		}
	}
	for k := range raw {
		if _, ok := allowed[k]; !ok {
			return fmt.Errorf("intentflow: payload field %q not in %s contract", k, eventName)
		}
		for _, bad := range []string{"prompt", "slots", "permissions", "result", "input"} {
			if k == bad {
				return fmt.Errorf("intentflow: sensitive payload field %s", k)
			}
		}
	}
	return nil
}