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

fix(intentflow): keep plan step receipts durable under PG events

Browse Source 
Normalize empty TEXT[] binds, isolate step-changed aggregates from plan
revision uniqueness, propagate AdvanceEligible errors, and emit
taskrun.completed to the registered notify consumer so finalize no longer
rolls back live receipts.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-16 10:42:20 +00:00
co-authored by Cursor
parent ad42f83e5b
commit aac42936bc
5 changed files with 109 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/app/intentflow/emit.go
+18 -1
View File
@@ -83,10 +83,27 @@ func (p *Plans) emitFact(ctx context.Context, plan *Plan, stepID string, now tim
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: "intent_plan", AggregateID: plan.PlanID, AggregateVersion: plan.Revision,
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,
internal/app/intentflow/pgsink_advance_test.go
+74
View File
@@ -0,0 +1,74 @@
package intentflow_test
import (
"context"
"testing"
"time"
"chat-agent-system/internal/app/intentflow"
"chat-agent-system/internal/app/uow"
"chat-agent-system/internal/db"
"chat-agent-system/internal/events"
"github.com/jackc/pgx/v5/pgxpool"
)
func TestAdvanceEligible_WithPGSinkWritesReceipt(t *testing.T) {
if testing.Short() {
t.Skip("short")
}
dsn := db.SkipIfNoTestDatabaseURL(t)
ctx := context.Background()
pool, err := pgxpool.New(ctx, dsn)
if err != nil {
t.Fatal(err)
}
t.Cleanup(pool.Close)
owner := intentflow.NewFakeActionOwner()
plans := intentflow.NewPlans(uow.NewPG(pool), intentflow.NewRepository(), intentflow.Options{
Owner: owner,
Events: events.PGSink{},
Now: func() time.Time { return time.Date(2026, 9, 16, 12, 0, 0, 0, time.UTC) },
})
prin := intentflow.Principal{UserID: 1, Environment: "test", ProjectID: 992, SessionID: "sess-adv-sink2"}
created, err := plans.CreatePlan(ctx, intentflow.CreateCommand{
Principal: prin, ClientTurnID: "adv-sink-2", RequestDigest: "d-sink2", WritePlan: true,
ExecutionMode: intentflow.ModeLive,
Steps: []intentflow.StepDraft{
{ActionKey: "regulation.query", ActionDigest: "q", ExecutionMode: intentflow.ModeLive},
},
})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
_, _ = pool.Exec(context.Background(), `DELETE FROM intent_plan_decisions WHERE plan_id=$1`, created.PlanID)
_, _ = pool.Exec(context.Background(), `DELETE FROM intent_plan_steps WHERE plan_id=$1`, created.PlanID)
_, _ = pool.Exec(context.Background(), `DELETE FROM intent_plans WHERE plan_id=$1`, created.PlanID)
_, _ = pool.Exec(context.Background(), `DELETE FROM event_deliveries WHERE event_id LIKE $1`, "intent-plan:"+created.PlanID+"%")
_, _ = pool.Exec(context.Background(), `DELETE FROM event_envelopes WHERE aggregate_id=$1`, created.PlanID)
})
confirmed, err := plans.DecidePlan(ctx, intentflow.DecisionCommand{
Principal: prin, PlanID: created.PlanID, ExpectedRevision: created.Revision,
DecisionNonce: "n-sink2", DecisionDigest: "dd-sink2", Kind: intentflow.DecisionConfirm,
})
if err != nil {
t.Fatal(err)
}
claimed, err := plans.ClaimStep(ctx, "w-sink", confirmed.PlanID, confirmed.Steps[0].StepID)
if err != nil || claimed == nil {
t.Fatalf("claim: %v %+v", err, claimed)
}
t.Logf("claimed fence=%d status=%s", claimed.FencingToken, claimed.Step.Status)
view, err := plans.ExecuteClaimed(ctx, claimed)
t.Logf("execute view=%+v err=%v", view, err)
if err != nil {
t.Fatalf("execute: %v", err)
}
var status, ref string
_ = pool.QueryRow(ctx, `SELECT status, result_ref FROM intent_plan_steps WHERE step_id=$1`, confirmed.Steps[0].StepID).Scan(&status, &ref)
t.Logf("db status=%s ref=%s", status, ref)
if status != "succeeded" {
t.Fatalf("want succeeded got %s", status)
}
}
internal/app/intentflow/repository.go
+11 -3
View File
@@ -77,13 +77,14 @@ func insertStep(ctx context.Context, tx pgx.Tx, s Step) error {
if len(slots) == 0 {
slots = []byte("{}")
}
dependsOn := textArrayOrEmpty(s.DependsOn)
_, err := tx.Exec(ctx, `
INSERT INTO intent_plan_steps (
step_id, plan_id, ordinal, action_key, slots, depends_on, status, expected_resource_version,
action_digest, idempotency_key, attempt, result_ref, result_digest, safe_summary,
lease_owner, lease_until, fencing_token, version, execution_mode, created_at, updated_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21)`,
s.StepID, s.PlanID, s.Ordinal, s.ActionKey, slots, s.DependsOn, string(s.Status),
s.StepID, s.PlanID, s.Ordinal, s.ActionKey, slots, dependsOn, string(s.Status),
s.ExpectedResourceVersion, s.ActionDigest, s.IdempotencyKey, s.Attempt, s.ResultRef,
s.ResultDigest, s.SafeSummary, s.LeaseOwner, s.LeaseUntil, s.FencingToken, s.Version,
string(s.ExecutionMode), s.CreatedAt, s.UpdatedAt,
@@ -208,7 +209,7 @@ INSERT INTO intent_plan_decisions (
exclude_unselected, decision_digest, outcome, created_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`,
d.DecisionID, d.PlanID, d.Nonce, d.PlanRevision, d.ActorUserID, string(d.Kind),
d.StepIDs, d.ExcludeUnselected, d.DecisionDigest, d.Outcome, d.CreatedAt,
textArrayOrEmpty(d.StepIDs), d.ExcludeUnselected, d.DecisionDigest, d.Outcome, d.CreatedAt,
)
if isUniqueViolation(err) {
return uow.ErrDigestConflict
@@ -256,7 +257,7 @@ SET status=$1, depends_on=$2, expected_resource_version=$3, action_digest=$4,
lease_owner=$10, lease_until=$11, fencing_token=$12, version=$13, execution_mode=$14,
updated_at=$15, slots=$16
WHERE step_id=$17 AND plan_id=$18`,
string(s.Status), s.DependsOn, s.ExpectedResourceVersion, s.ActionDigest,
string(s.Status), textArrayOrEmpty(s.DependsOn), s.ExpectedResourceVersion, s.ActionDigest,
s.IdempotencyKey, s.Attempt, s.ResultRef, s.ResultDigest, s.SafeSummary,
s.LeaseOwner, s.LeaseUntil, s.FencingToken, s.Version, string(s.ExecutionMode),
s.UpdatedAt, jsonbOrNil(s.Slots), s.StepID, planID,
@@ -473,6 +474,13 @@ func jsonbOrNil(raw []byte) []byte {
return raw
}
func textArrayOrEmpty(vals []string) []string {
if vals == nil {
return []string{}
}
return vals
}
func statusStrings(in []PlanStatus) []string {
out := make([]string, len(in))
for i, s := range in {
internal/app/intentflow/turn.go
+5 -1
View File
@@ -151,7 +151,11 @@ func (p *Plans) AdvanceEligible(ctx context.Context, workerID, planID string) (*
last = got
}
if last == nil {
return p.GetPlan(ctx, prin, planID)
view, getErr := p.GetPlan(ctx, prin, planID)
if firstErr != nil {
return view, firstErr
}
return view, getErr
}
return last, firstErr
}
internal/taskrun/finalize.go
+1 -1
View File
@@ -448,7 +448,7 @@ func (c *Commands) appendTerminalEvent(ctx context.Context, record *RunRecord, r
OccurredAt: time.Now().UTC(),
Payload: body,
}
return (events.PGSink{}).Append(ctx, env, []string{"projection", "fav"})
return (events.PGSink{}).Append(ctx, env, []string{events.ConsumerNotify})
}
func (c *Commands) stageControlledDoc(ctx context.Context, record *RunRecord, result *bizdev.TaskRunResult, status string, man *RunManifest) error {