Agent-System/internal/app/docflow/e0a_pg_test.go

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)
}