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 }