test(docflow): prove three production chains share one Postgres transaction

Isolated-library coverage checks approved/candidate/impact envelopes, empty-env rollback, and zero FAV staging rows. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-15 08:52:48 +00:00
parent 504058debf
commit 0f2057765a
3 changed files with 283 additions and 1 deletions
@@ -0,0 +1,66 @@
+package docflow
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/events"
+)
+
+type recSink struct {
+	envs []events.Envelope
+	cons [][]string
+	err  error
+}
+
+func (s *recSink) Append(_ context.Context, env events.Envelope, consumers []string) error {
+	if s.err != nil {
+		return s.err
+	}
+	if err := events.ValidateEnvelope(env); err != nil {
+		return err
+	}
+	if err := events.AssertLiveConsumers(env.EventName, consumers); err != nil {
+		return err
+	}
+	s.envs = append(s.envs, env)
+	s.cons = append(s.cons, consumers)
+	return nil
+}
+
+func TestEmitFailClosedMissingProject(t *testing.T) {
+	sink := &recSink{}
+	cmds := &Commands{Events: sink, Environment: "dev"}
+	err := cmds.emit(context.Background(), &docgov.DocumentRecord{DocID: "d", ProjectID: 0}, events.EventApproved, "e1", "k", "document_approval", "d", 1, Principal{UserID: 1}, []byte(`{"doc_id":"d"}`))
+	if !errors.Is(err, events.ErrInvalidScope) {
+		t.Fatalf("missing project: %v", err)
+	}
+	if len(sink.envs) != 0 {
+		t.Fatal("must not append")
+	}
+	if err := cmds.emit(context.Background(), &docgov.DocumentRecord{DocID: "d", ProjectID: 9}, events.EventApproved, "e1", "k", "document_approval", "d", 1, Principal{UserID: 1}, []byte(`{"doc_id":"d","project_id":9,"task_key":"intake"}`)); err != nil {
+		t.Fatal(err)
+	}
+	if len(sink.envs) != 1 || sink.envs[0].ProjectID != 9 || sink.envs[0].Environment != "dev" {
+		t.Fatalf("%+v", sink.envs)
+	}
+	for _, c := range sink.cons[0] {
+		if c == events.ConsumerFAV {
+			t.Fatal("fav consumer")
+		}
+	}
+}
+
+func TestEmitRejectsFAVConsumers(t *testing.T) {
+	sink := &recSink{}
+	cmds := &Commands{Events: sink, Environment: "dev"}
+	if err := cmds.Events.Append(context.Background(), events.Envelope{
+		EventID: "e", EventName: events.EventVersionCommitted, SchemaVersion: events.EnvelopeSchema,
+		Environment: "dev", ProjectID: 9, AggregateID: "d", PayloadDigest: events.Digest([]byte(`{"doc_id":"d"}`)),
+		Producer: "docflow", Payload: []byte(`{"doc_id":"d"}`),
+	}, []string{events.ConsumerFAV}); err == nil {
+		t.Fatal("fav must fail")
+	}
+}
@@ -0,0 +1,216 @@
+package docflow_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/app/docflow"
+	"chat-agent-system/internal/app/uow"
+	"chat-agent-system/internal/db"
+	"chat-agent-system/internal/docgov"
+	"chat-agent-system/internal/docrevision"
+	"chat-agent-system/internal/events"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+func init() {
+	if _, err := os.Stat("deploy/postgres/goose"); err == nil {
+		return
+	}
+	for _, cand := range []string{"../../..", "../../../.."} {
+		if fi, err := os.Stat(cand + "/deploy/postgres/goose"); err == nil && fi.IsDir() {
+			_ = os.Chdir(cand)
+			return
+		}
+	}
+}
+
+func e0aPool(t *testing.T) (*pgxpool.Pool, context.Context) {
+	t.Helper()
+	dsn := db.SkipIfNoTestDatabaseURL(t)
+	db.RefuseSharedEnvDSN(t, dsn)
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(pool.Close)
+	if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+	return pool, ctx
+}
+
+func TestE0A_ThreeChainsSameTxAndFailClosed_PG(t *testing.T) {
+	if testing.Short() {
+		t.Skip("E0A production chains need isolated postgres")
+	}
+	pool, ctx := e0aPool(t)
+	suffix := fmt.Sprintf("%d", time.Now().UnixNano())
+	var projectID int64
+	if err := pool.QueryRow(ctx, `INSERT INTO projects (code, name, status) VALUES ($1,$2,'active') RETURNING id`,
+		"e0a-"+suffix, "e0a").Scan(&projectID); err != nil {
+		t.Fatal(err)
+	}
+	docID := "e0a_" + suffix
+	matID := "e0am_" + suffix
+	failID := "e0af_" + suffix
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM event_deliveries WHERE event_id LIKE $1 OR event_id LIKE $2 OR event_id LIKE $3`, "document:"+docID+"%", "document:"+matID+"%", "document:"+failID+"%")
+		_, _ = pool.Exec(context.Background(), `DELETE FROM event_envelopes WHERE aggregate_id IN ($1,$2,$3) OR aggregate_id LIKE $4`, docID, matID, failID, matID+":%")
+		_, _ = pool.Exec(context.Background(), `DELETE FROM document_change_events WHERE doc_id IN ($1,$2,$3)`, docID, matID, failID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM document_versions WHERE doc_id IN ($1,$2,$3)`, docID, matID, failID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM document_snapshots WHERE doc_id IN ($1,$2,$3)`, docID, matID, failID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM document_records WHERE doc_id IN ($1,$2,$3)`, docID, matID, failID)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id=$1`, projectID)
+	})
+
+	gov := docgov.NewGovernanceEngine(pool)
+	rev := docrevision.NewService(pool)
+	rev.ImpactResolver = func(_ context.Context, _ string) []docrevision.DocImpact {
+		return []docrevision.DocImpact{{DocID: "peer_" + suffix, ImpactType: "depends", Reason: "e0a-fixture"}}
+	}
+	for _, id := range []string{docID, matID, failID} {
+		if _, err := gov.CreateOrGetDocument(ctx, id, projectID, "intake", "t", "V0.5_feasibility", "ws/"+id, 1, "A", 2, "R", 3, "P"); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := gov.SubmitForReview(ctx, docID, 1, docgov.RoleAuthor, "A"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := gov.ReviewApprove(ctx, docID, 2, docgov.RoleReviewer, "R", "ok"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := gov.SubmitForReview(ctx, failID, 1, docgov.RoleAuthor, "A"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := gov.ReviewApprove(ctx, failID, 2, docgov.RoleReviewer, "R", "ok"); err != nil {
+		t.Fatal(err)
+	}
+
+	cmds := &docflow.Commands{
+		UoW: uow.NewPG(pool), Gov: gov, Rev: rev, Auth: docflow.DutyAuthorizer(),
+		Events: events.PGSink{}, Environment: "dev",
+	}
+	doc, err := cmds.Approve(ctx, docflow.Principal{UserID: 3, Role: docgov.RoleApprover, Name: "P"}, docflow.DecisionCommand{DocID: docID})
+	if err != nil || doc == nil {
+		t.Fatalf("approve: %v %+v", err, doc)
+	}
+
+	var envelopes, deliveries, fav int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_envelopes WHERE aggregate_id=$1 AND event_name=$2`, docID, events.EventApproved).Scan(&envelopes); err != nil {
+		t.Fatal(err)
+	}
+	if envelopes != 1 {
+		t.Fatalf("approved envelopes=%d", envelopes)
+	}
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_deliveries d JOIN event_envelopes e ON e.event_id=d.event_id WHERE e.aggregate_id=$1 AND d.consumer='workflow_advance'`, docID).Scan(&deliveries); err != nil {
+		t.Fatal(err)
+	}
+	if deliveries != 1 {
+		t.Fatalf("approved workflow_advance deliveries=%d", deliveries)
+	}
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_deliveries d JOIN event_envelopes e ON e.event_id=d.event_id WHERE e.aggregate_id IN ($1,$2) AND d.consumer='fav'`, docID, matID).Scan(&fav); err != nil {
+		t.Fatal(err)
+	}
+	if fav != 0 {
+		t.Fatalf("fav deliveries=%d", fav)
+	}
+	var staging int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM artifact_payload_staging`).Scan(&staging); err != nil {
+		t.Fatal(err)
+	}
+	if staging != 0 {
+		t.Fatalf("fav staging rows=%d", staging)
+	}
+
+	bad := &docflow.Commands{
+		UoW: uow.NewPG(pool), Gov: gov, Rev: rev, Auth: docflow.DutyAuthorizer(),
+		Events: events.PGSink{}, Environment: "",
+	}
+	if _, err := bad.Approve(ctx, docflow.Principal{UserID: 3, Role: docgov.RoleApprover, Name: "P"}, docflow.DecisionCommand{DocID: failID}); !errors.Is(err, events.ErrInvalidScope) {
+		t.Fatalf("empty env want scope, got %v", err)
+	}
+	var failState string
+	if err := pool.QueryRow(ctx, `SELECT current_state FROM document_records WHERE doc_id=$1`, failID).Scan(&failState); err != nil {
+		t.Fatal(err)
+	}
+	if failState != string(docgov.StateApproving) {
+		t.Fatalf("fail-closed rolled back domain? state=%s", failState)
+	}
+	var failEnv int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_envelopes WHERE aggregate_id=$1`, failID).Scan(&failEnv); err != nil {
+		t.Fatal(err)
+	}
+	if failEnv != 0 {
+		t.Fatalf("fail-closed left envelopes=%d", failEnv)
+	}
+
+	if _, err := rev.OpenTemplateVersion(ctx, matID, "# base\n\n法规引用 GB9706\n", docrevision.ActorSnapshot{}, projectID, "ws/"+matID); err != nil {
+		t.Fatal(err)
+	}
+	mat, err := cmds.MaterializeAgent(ctx, docflow.Principal{UserID: 1}, docrevision.AgentMaterializeRequest{
+		DocID: matID, RunID: "run-" + suffix, OutputMarkdown: "# base\n\npeer edit\n",
+		ProjectID: projectID, WorkDir: "ws/" + matID, ActorKind: docrevision.ActorAgentSquad, AutoCommitUserID: 1,
+	})
+	if err != nil && mat == nil {
+		t.Fatal(err)
+	}
+	if mat != nil && mat.Candidate != nil && mat.Candidate.Status == docrevision.CandidateNeedsConfirmation {
+		if _, err := cmds.AgentCommit(ctx, docflow.Principal{UserID: 1}, docflow.AgentCommitCommand{
+			DocID: matID, CandidateID: mat.Candidate.ID,
+			Decision: docrevision.CommitAgentDecision{Decision: "commit", Confirmed: true, ActorUserID: 1},
+		}); err != nil {
+			t.Fatalf("agent commit: %v", err)
+		}
+	}
+	var ready, committed, impact int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_envelopes WHERE aggregate_id=$1 AND event_name=$2`, matID, events.EventCandidateReady).Scan(&ready); err != nil {
+		t.Fatal(err)
+	}
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_envelopes WHERE aggregate_id=$1 AND event_name=$2`, matID, events.EventVersionCommitted).Scan(&committed); err != nil {
+		t.Fatal(err)
+	}
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_envelopes WHERE event_name=$1 AND aggregate_id LIKE $2`, events.EventDownstreamImpact, matID+":%").Scan(&impact); err != nil {
+		t.Fatal(err)
+	}
+	if ready+committed == 0 {
+		t.Fatalf("candidate-ready=%d version_committed=%d mat=%+v", ready, committed, mat)
+	}
+	if impact < 1 {
+		t.Fatalf("downstream-impact envelopes=%d (ready=%d committed=%d)", impact, ready, committed)
+	}
+}
+
+func TestE0A_InventoryClassifiesDirtyScope_PG(t *testing.T) {
+	if testing.Short() {
+		t.Skip("inventory pg")
+	}
+	pool, ctx := e0aPool(t)
+	unit := uow.NewPG(pool)
+	env := events.Envelope{EventID: "dirty-scope-" + fmt.Sprintf("%d", time.Now().UnixNano()), EventName: events.EventApproved, SchemaVersion: 1}
+	row := events.ClassifyEnvelope(env, events.ConsumerNotify)
+	if row.Classification != events.ClassificationQuarantine {
+		t.Fatalf("%+v", row)
+	}
+	err := unit.Within(ctx, func(scope uow.TxScope) error {
+		ctx := uow.WithScope(ctx, scope)
+		return events.PersistInventory(ctx, row, "ops", "e0a-ticket")
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var class string
+	if err := pool.QueryRow(ctx, `SELECT classification FROM event_inventory_decisions WHERE event_id=$1`, env.EventID).Scan(&class); err != nil {
+		t.Fatal(err)
+	}
+	if class != events.ClassificationQuarantine {
+		t.Fatalf("%s", class)
+	}
+	_, _ = pool.Exec(ctx, `DELETE FROM event_inventory_decisions WHERE event_id=$1`, env.EventID)
+}
@@ -60,7 +60,7 @@ func TestOutboxAndStagingSameTxRollback(t *testing.T) {
 	unit := uow.NewPG(pool)
 	err = unit.Within(ctx, func(scope uow.TxScope) error {
 		ctx = uow.WithScope(ctx, scope)
-		if err := (events.PGSink{}).Append(ctx, env, []string{"projection", "fav"}); err != nil {
+		if err := (events.PGSink{}).Append(ctx, env, []string{"projection", "notify"}); err != nil {
 			return err
 		}
 		ing := &artifactvault.Ingest{Env: "dev"}