diff --git a/cmd/events/main.go b/cmd/events/main.go index 3102625c..f9d8ebfe 100644 --- a/cmd/events/main.go +++ b/cmd/events/main.go @@ -14,6 +14,7 @@ import ( "chat-agent-system/internal/app/docadvance" "chat-agent-system/internal/app/docnotify" "chat-agent-system/internal/app/docproject" + "chat-agent-system/internal/app/intentflow" "chat-agent-system/internal/config" "chat-agent-system/internal/db" "chat-agent-system/internal/docgov" @@ -78,9 +79,10 @@ func main() { _ = disp.Register(events.ConsumerWorkflowAdvance, (&docadvance.EventConsumer{ Inst: instSvc, Docs: gov, PW: pwSvc, Notify: notify, Env: cfg.Env, }).HandleLease) - _ = disp.Register(events.ConsumerProjection, (&docproject.EventConsumer{ - Docs: gov, Env: cfg.Env, - }).HandleLease) + _ = disp.Register(events.ConsumerProjection, intentflow.MuxProjection( + (&docproject.EventConsumer{Docs: gov, Env: cfg.Env}).HandleLease, + (&intentflow.ProjectionConsumer{Env: cfg.Env}).HandleLease, + )) mux := http.NewServeMux() mux.HandleFunc("GET /health", func(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, map[string]any{"status": "ok", "env": cfg.Env, "owner": owner}) diff --git a/internal/app/app.go b/internal/app/app.go index 9002082d..25bdce9f 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -192,7 +192,9 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { if pool != nil { rt.Docflow.UoW = uow.NewPG(pool) rt.Docflow.Events = events.PGSink{} - rt.IntentPlans = intentflow.NewPlans(rt.Docflow.UoW, intentflow.NewRepository(), intentflow.Options{}) + rt.IntentPlans = intentflow.NewPlans(rt.Docflow.UoW, intentflow.NewRepository(), intentflow.Options{ + Events: events.PGSink{}, + }) } svc.SetAgentMaterializer(rt.Docflow) rt.bindDocumentNotify() diff --git a/internal/app/intentflow/emit.go b/internal/app/intentflow/emit.go new file mode 100644 index 00000000..cda3d0e4 --- /dev/null +++ b/internal/app/intentflow/emit.go @@ -0,0 +1,122 @@ +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) + env := events.Envelope{ + EventID: eventID, EventName: name, SchemaVersion: events.EnvelopeSchema, + Environment: plan.Principal.Environment, ProjectID: plan.Principal.ProjectID, + AggregateType: "intent_plan", AggregateID: plan.PlanID, AggregateVersion: plan.Revision, + 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 +} diff --git a/internal/app/intentflow/emit_test.go b/internal/app/intentflow/emit_test.go new file mode 100644 index 00000000..fe4a4387 --- /dev/null +++ b/internal/app/intentflow/emit_test.go @@ -0,0 +1,106 @@ +package intentflow_test + +import ( + "context" + "encoding/json" + "errors" + "os" + "path/filepath" + "runtime" + "testing" + + "chat-agent-system/internal/app/intentflow" + "chat-agent-system/internal/events" +) + +func TestCreatePlanEmitsAwaitingInput(t *testing.T) { + sink := &intentflow.MemorySink{} + proj := intentflow.NewProjector() + now, opts, _ := testClock() + opts.Events = sink + p := intentflow.NewPlans(&intentflow.SerialUoW{}, intentflow.NewMemoryStore(), opts) + ctx := context.Background() + view, err := p.CreatePlan(ctx, draft("digest-a")) + if err != nil { + t.Fatal(err) + } + envs := sink.All() + if len(envs) != 1 || envs[0].EventName != events.EventIntentPlanAwaiting { + t.Fatalf("want one awaiting_input, got %#v", envs) + } + if envs[0].Producer != "app/intentflow" { + t.Fatalf("producer %s", envs[0].Producer) + } + var payload map[string]any + if err := json.Unmarshal(envs[0].Payload, &payload); err != nil { + t.Fatal(err) + } + for _, bad := range []string{"prompt", "slots", "permissions", "result", "input"} { + if _, ok := payload[bad]; ok { + t.Fatalf("sensitive field %s leaked", bad) + } + } + if err := proj.ApplyEnvelope(envs[0]); err != nil { + t.Fatal(err) + } + if err := proj.ApplyEnvelope(envs[0]); err != nil { + t.Fatal(err) + } + sh, ok := proj.VisibleShadow(principal(), view.PlanID) + if !ok || sh.Revision != 1 { + t.Fatalf("shadow %+v ok=%v", sh, ok) + } + other := principal() + other.UserID = 99 + if _, ok := proj.VisibleShadow(other, view.PlanID); ok { + t.Fatal("cross-user shadow must be hidden") + } + got, err := p.GetPlan(ctx, principal(), view.PlanID) + if err != nil || got == nil { + t.Fatalf("SSOT GET must work without card success: %v", err) + } + _ = now +} + +func TestProjectionOnlyForwardAndForgedRejected(t *testing.T) { + proj := intentflow.NewProjector() + newer := events.Envelope{ + EventID: "e2", EventName: events.EventIntentPlanAwaiting, SchemaVersion: events.EnvelopeSchema, + Environment: "dev", ProjectID: 42, AggregateType: "intent_plan", AggregateID: "pln_x", + AggregateVersion: 2, Producer: "app/intentflow", Payload: []byte(`{"plan_id":"pln_x","revision":2,"status":"executing","environment":"dev","project_id":42,"user_id":7,"session_id":"sess-u4","summary_ref":""}`), + } + newer.PayloadDigest = events.Digest(newer.Payload) + older := newer + older.EventID = "e1" + older.AggregateVersion = 1 + older.Payload = []byte(`{"plan_id":"pln_x","revision":1,"status":"awaiting_confirmation","environment":"dev","project_id":42,"user_id":7,"session_id":"sess-u4","summary_ref":""}`) + older.PayloadDigest = events.Digest(older.Payload) + if err := proj.ApplyEnvelope(newer); err != nil { + t.Fatal(err) + } + if err := proj.ApplyEnvelope(older); err != nil { + t.Fatal(err) + } + sh, _ := proj.Shadow("pln_x") + if sh.Revision != 2 || sh.Status != "executing" { + t.Fatalf("low revision must not overwrite: %+v", sh) + } + forged := newer + forged.EventID = "e3" + forged.Producer = "forged" + forged.PayloadDigest = events.Digest(forged.Payload) + if err := proj.ApplyEnvelope(forged); !errors.Is(err, events.ErrIntegrityFailure) { + t.Fatalf("forged producer: %v", err) + } +} + +func TestAwaitingPayloadGolden(t *testing.T) { + _, file, _, _ := runtime.Caller(0) + raw, err := os.ReadFile(filepath.Join(filepath.Dir(file), "testdata", "intent-plan-awaiting-v1.json")) + if err != nil { + t.Fatal(err) + } + if err := intentflow.AssertPlanPayload(events.EventIntentPlanAwaiting, raw); err != nil { + t.Fatal(err) + } +} diff --git a/internal/app/intentflow/plans.go b/internal/app/intentflow/plans.go index 552ec322..28c9bcd3 100644 --- a/internal/app/intentflow/plans.go +++ b/internal/app/intentflow/plans.go @@ -19,17 +19,19 @@ const ( // Plans is the owner implementation of Service plus worker/recovery methods. type Plans struct { - uow uow.UnitOfWork - store Store - owner ActionOwner - now func() time.Time - newID func(prefix string) string - lease time.Duration - ttl time.Duration + uow uow.UnitOfWork + store Store + owner ActionOwner + events EventSink + now func() time.Time + newID func(prefix string) string + lease time.Duration + ttl time.Duration } type Options struct { Owner ActionOwner + Events EventSink Now func() time.Time NewID func(prefix string) string Lease time.Duration @@ -53,7 +55,7 @@ func NewPlans(unit uow.UnitOfWork, store Store, opt Options) *Plans { if ttl <= 0 { ttl = defaultPlanTTL } - return &Plans{uow: unit, store: store, owner: opt.Owner, now: now, newID: newID, lease: lease, ttl: ttl} + return &Plans{uow: unit, store: store, owner: opt.Owner, events: opt.Events, now: now, newID: newID, lease: lease, ttl: ttl} } func randomID(prefix string) string { @@ -89,7 +91,7 @@ func (p *Plans) GetPlan(ctx context.Context, principal Principal, planID string) return nil, err } var view *PlanView - err := p.uow.Within(ctx, func(scope uow.TxScope) error { + err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error { plan, err := p.store.GetPlan(ctx, scope, planID) if err != nil { return err @@ -108,7 +110,7 @@ func (p *Plans) GetActivePlan(ctx context.Context, principal Principal) (*PlanVi return nil, err } var view *PlanView - err := p.uow.Within(ctx, func(scope uow.TxScope) error { + err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error { plan, err := p.store.GetActive(ctx, scope, principal) if err != nil { return err @@ -155,7 +157,7 @@ func (p *Plans) CreatePlan(ctx context.Context, in CreateCommand) (*PlanView, er } var view *PlanView - err := p.uow.Within(ctx, func(scope uow.TxScope) error { + err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error { existing, err := p.store.GetByTurn(ctx, scope, in.Principal.Environment, in.Principal.SessionID, in.ClientTurnID) if err != nil { return err @@ -244,7 +246,7 @@ func (p *Plans) CreatePlan(ctx context.Context, in CreateCommand) (*PlanView, er return err } view = toView(plan) - return nil + return p.emitFact(ctx, plan, "", now) }) return view, err } @@ -258,7 +260,7 @@ func (p *Plans) DecidePlan(ctx context.Context, in DecisionCommand) (*PlanView, } now := p.now() var view *PlanView - err := p.uow.Within(ctx, func(scope uow.TxScope) error { + err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error { plan, err := p.store.GetPlanForUpdate(ctx, scope, in.PlanID) if err != nil { return err @@ -357,7 +359,7 @@ func (p *Plans) DecidePlan(ctx context.Context, in DecisionCommand) (*PlanView, plan.Status = to plan.Revision = newRev view = toView(plan) - return nil + return p.emitFact(ctx, plan, "", now) }) return view, err } @@ -421,7 +423,7 @@ func (p *Plans) ClaimStep(ctx context.Context, workerID, planID, stepID string) now := p.now() until := now.Add(p.lease) var out *ClaimedStep - err := p.uow.Within(ctx, func(scope uow.TxScope) error { + err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error { plan, err := p.store.GetPlanForUpdate(ctx, scope, planID) if err != nil { return err @@ -493,7 +495,7 @@ func (p *Plans) RecoverReceipt(ctx context.Context, claimed *ClaimedStep) (*Plan func (p *Plans) writeReceipt(ctx context.Context, claimed *ClaimedStep, rec Receipt) (*PlanView, error) { now := p.now() var view *PlanView - err := p.uow.Within(ctx, func(scope uow.TxScope) error { + err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error { ok, err := p.store.WriteStepReceipt(ctx, scope, claimed.WorkerID, claimed.FencingToken, Step{ PlanID: claimed.Plan.PlanID, StepID: claimed.Step.StepID, @@ -512,14 +514,14 @@ func (p *Plans) writeReceipt(ctx context.Context, claimed *ClaimedStep, rec Rece return err } view = toView(plan) - return nil + return p.emitFact(ctx, plan, claimed.Step.StepID, now) }) return view, err } func (p *Plans) ExpireDue(ctx context.Context) (int, error) { var n int - err := p.uow.Within(ctx, func(scope uow.TxScope) error { + err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error { var err error n, err = p.store.ExpireWaiting(ctx, scope, p.now()) return err @@ -529,7 +531,7 @@ func (p *Plans) ExpireDue(ctx context.Context) (int, error) { func (p *Plans) PurgeClosed(ctx context.Context, olderThan time.Time) (int, error) { var n int - err := p.uow.Within(ctx, func(scope uow.TxScope) error { + err := p.withTx(ctx, func(ctx context.Context, scope uow.TxScope) error { var err error n, err = p.store.PurgeClosed(ctx, scope, olderThan) return err diff --git a/internal/app/intentflow/projection.go b/internal/app/intentflow/projection.go new file mode 100644 index 00000000..ff44139c --- /dev/null +++ b/internal/app/intentflow/projection.go @@ -0,0 +1,243 @@ +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) + } +} diff --git a/internal/app/intentflow/testdata/intent-plan-awaiting-v1.json b/internal/app/intentflow/testdata/intent-plan-awaiting-v1.json new file mode 100644 index 00000000..c1aaa3db --- /dev/null +++ b/internal/app/intentflow/testdata/intent-plan-awaiting-v1.json @@ -0,0 +1,10 @@ +{ + "plan_id": "pln_golden", + "revision": 1, + "status": "awaiting_confirmation", + "environment": "dev", + "project_id": 42, + "user_id": 7, + "session_id": "sess-u4", + "summary_ref": "safe" +}