package intentflow import ( "context" "encoding/json" "strings" "sync" "chat-agent-system/internal/events" "github.com/jackc/pgx/v5" ) // MemorySink records envelopes for tests. It still enforces live contracts. type MemorySink struct { mu sync.Mutex Envs []events.Envelope } func (m *MemorySink) Append(ctx context.Context, env events.Envelope, consumers []string) error { if err := events.AssertLiveConsumers(env.EventName, consumers); err != nil { return err } if err := events.ValidateEnvelope(env); err != nil { return err } m.mu.Lock() defer m.mu.Unlock() m.Envs = append(m.Envs, env) return nil } func (m *MemorySink) All() []events.Envelope { m.mu.Lock() defer m.mu.Unlock() out := make([]events.Envelope, len(m.Envs)) copy(out, m.Envs) return out } // Shadow is the only-forward projection of a plan. It is not SSOT. type Shadow struct { PlanID string Revision int64 Status string StepID string Environment string ProjectID int64 UserID int64 SessionID string SummaryRef string } // Projector applies intent.plan.* facts idempotently and only forward. type Projector struct { mu sync.Mutex events map[string]struct{} shadow map[string]Shadow cards []CardProjection fail bool } func NewProjector() *Projector { return &Projector{events: map[string]struct{}{}, shadow: map[string]Shadow{}} } func (p *Projector) SetFail(v bool) { if p == nil { return } p.mu.Lock() p.fail = v p.mu.Unlock() } func (p *Projector) ProjectCard(ctx context.Context, card CardProjection) error { if p == nil { return nil } p.mu.Lock() defer p.mu.Unlock() if p.fail { return events.ErrReplayApply } p.cards = append(p.cards, card) return nil } func (p *Projector) Cards() []CardProjection { p.mu.Lock() defer p.mu.Unlock() out := make([]CardProjection, len(p.cards)) copy(out, p.cards) return out } func (p *Projector) Shadow(planID string) (Shadow, bool) { p.mu.Lock() defer p.mu.Unlock() s, ok := p.shadow[planID] return s, ok } func (p *Projector) ApplyEnvelope(env events.Envelope) error { if p == nil { return nil } if err := events.ValidateEnvelope(env); err != nil { return err } if !strings.HasPrefix(env.EventName, "intent.plan.") { return nil } fact, err := decodePlanFact(env.Payload) if err != nil { return err } p.mu.Lock() defer p.mu.Unlock() if _, ok := p.events[env.EventID]; ok { return nil } p.events[env.EventID] = struct{}{} cur, ok := p.shadow[fact.PlanID] if ok && fact.Revision < cur.Revision { return nil } p.shadow[fact.PlanID] = Shadow{ PlanID: fact.PlanID, Revision: fact.Revision, Status: fact.Status, StepID: fact.StepID, Environment: fact.Environment, ProjectID: fact.ProjectID, UserID: fact.UserID, SessionID: fact.SessionID, SummaryRef: fact.SummaryRef, } return nil } func (p *Projector) VisibleShadow(prin Principal, planID string) (Shadow, bool) { s, ok := p.Shadow(planID) if !ok { return Shadow{}, false } if s.Environment != prin.Environment || s.ProjectID != prin.ProjectID || s.SessionID != prin.SessionID || s.UserID != prin.UserID { return Shadow{}, false } return s, true } func decodePlanFact(body []byte) (planFact, error) { var fact planFact if err := json.Unmarshal(body, &fact); err != nil { return planFact{}, err } return fact, nil } // ProjectionConsumer applies intent.plan.* onto shadow tables inside the lease tx. type ProjectionConsumer struct { Env string Cards CardProjector } func (c *ProjectionConsumer) HandleLease(ctx context.Context, lease events.Lease, tx pgx.Tx) error { if c == nil { return events.ErrUnknownConsumer } principal := events.ServicePrincipal{ ID: "projection-" + c.Env, Environment: c.Env, Consumer: events.ConsumerProjection, Actions: []string{"projection.apply"}, } if err := events.AuthorizePrincipal(principal, lease.Environment, lease.ProjectID, "projection.apply"); err != nil { return events.ErrQuarantine } if !strings.HasPrefix(lease.EventName, "intent.plan.") { return nil } if err := events.ValidateEnvelope(lease.Envelope); err != nil { return events.ErrQuarantine } fact, err := decodePlanFact(lease.Envelope.Payload) if err != nil { return events.ErrQuarantine } if fact.UserID <= 0 || fact.SessionID == "" { return events.ErrQuarantine } tag, err := tx.Exec(ctx, ` INSERT INTO intent_plan_projections ( source_event_id, plan_id, revision, status, step_id, environment, project_id, user_id, session_id, summary_ref ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) ON CONFLICT (source_event_id) DO NOTHING`, lease.EventID, fact.PlanID, fact.Revision, fact.Status, fact.StepID, fact.Environment, fact.ProjectID, fact.UserID, fact.SessionID, fact.SummaryRef) if err != nil { return err } if tag.RowsAffected() == 0 { return nil } _, err = tx.Exec(ctx, ` INSERT INTO intent_plan_shadow ( plan_id, revision, status, step_id, environment, project_id, user_id, session_id, summary_ref, updated_at ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9, now()) ON CONFLICT (plan_id) DO UPDATE SET revision = EXCLUDED.revision, status = EXCLUDED.status, step_id = EXCLUDED.step_id, summary_ref = EXCLUDED.summary_ref, updated_at = now() WHERE intent_plan_shadow.revision <= EXCLUDED.revision`, fact.PlanID, fact.Revision, fact.Status, fact.StepID, fact.Environment, fact.ProjectID, fact.UserID, fact.SessionID, fact.SummaryRef) if err != nil { return err } if c.Cards != nil { if err := c.Cards.ProjectCard(ctx, CardProjection{ Principal: Principal{ UserID: fact.UserID, Environment: fact.Environment, ProjectID: fact.ProjectID, SessionID: fact.SessionID, }, PlanID: fact.PlanID, Revision: fact.Revision, Status: PlanStatus(fact.Status), SummaryRef: fact.SummaryRef, }); err != nil { return err } } return nil } // MuxProjection dispatches document vs intent-plan leases without changing claim semantics. func MuxProjection(doc events.Handler, plan events.Handler) events.Handler { return func(ctx context.Context, lease events.Lease, tx pgx.Tx) error { if strings.HasPrefix(lease.EventName, "intent.plan.") { if plan == nil { return nil } return plan(ctx, lease, tx) } if doc == nil { return nil } return doc(ctx, lease, tx) } }