weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(intentflow): emit plan facts onto Track E outbox

Browse Source 
Append awaiting/step/completed envelopes in the command UoW and project them without changing dispatcher leases.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-16 00:46:09 +00:00
co-authored by Cursor
parent cf063f5727
commit c36d891b05
7 changed files with 510 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/events/main.go
+5 -3
View File
@@ -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})
internal/app/app.go
+3 -1
View File
@@ -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()
internal/app/intentflow/emit.go
+122
View File
@@ -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
}
internal/app/intentflow/emit_test.go
+106
View File
@@ -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)
}
}
internal/app/intentflow/plans.go
+21 -19
View File
@@ -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
internal/app/intentflow/projection.go
+243
View File
@@ -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)
}
}
internal/app/intentflow/testdata/intent-plan-awaiting-v1.json
Vendored
+10
View File
@@ -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"
}