From 0f2057765aee28fa3448e7198058515bc58aa4d5 Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 15 Sep 2026 08:52:48 +0000 Subject: [PATCH] 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 --- internal/app/docflow/commands_scope_test.go | 66 ++++++ internal/app/docflow/e0a_pg_test.go | 216 ++++++++++++++++++++ internal/app/docflow/same_tx_pg_test.go | 2 +- 3 files changed, 283 insertions(+), 1 deletion(-) create mode 100644 internal/app/docflow/commands_scope_test.go create mode 100644 internal/app/docflow/e0a_pg_test.go diff --git a/internal/app/docflow/commands_scope_test.go b/internal/app/docflow/commands_scope_test.go new file mode 100644 index 00000000..dd407e03 --- /dev/null +++ b/internal/app/docflow/commands_scope_test.go @@ -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") + } +} diff --git a/internal/app/docflow/e0a_pg_test.go b/internal/app/docflow/e0a_pg_test.go new file mode 100644 index 00000000..e818c6fc --- /dev/null +++ b/internal/app/docflow/e0a_pg_test.go @@ -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) +} diff --git a/internal/app/docflow/same_tx_pg_test.go b/internal/app/docflow/same_tx_pg_test.go index 20d02b8f..f4d4e2fa 100644 --- a/internal/app/docflow/same_tx_pg_test.go +++ b/internal/app/docflow/same_tx_pg_test.go @@ -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"}