Agent-System-Backend/internal/store/task_projection_test.go

package store

import (
	core "agent-system-core"
	"errors"
	"testing"
	"time"
)

func projectionFixture(t *testing.T, st *Store) core.ActivityCard {
	t.Helper()
	p, e := st.CreateProject("o", "projection")
	if e != nil {
		t.Fatal(e)
	}
	conv, e := st.CreateConversation("o", p.ID, "projection", "")
	if e != nil {
		t.Fatal(e)
	}
	c := core.ActivityCard{ID: NewID("card"), ProjectID: p.ID, ConversationID: conv.ID, Kind: core.CardKindProgress, Title: "整理", Status: core.CardActive, CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC()}
	c.Task = &core.TaskSnapshot{SchemaVersion: 1, TaskID: "t1", AttemptID: "a1", CardID: c.ID, ProjectID: p.ID, ConversationID: conv.ID, OwnerInstanceID: "i1", Revision: 1, StreamScope: "conversation:" + conv.ID, ExecutionState: "queued", OutcomeCompleteness: "none", DeliveryState: "pending", Title: "整理"}
	c.Body = c.Task.SummaryBody()
	return c
}
func TestTaskProjectionCommitsCardAndBothStreams(t *testing.T) {
	st := openTestStore(t)
	c := projectionFixture(t, st)
	events, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: c, Channels: []string{c.ConversationID, "project:" + c.ProjectID}})
	if e != nil || len(events) != 2 {
		t.Fatal("missing atomic projection", events, e)
	}
	got, e := st.GetCard("o", c.ID)
	if e != nil || got.Task == nil || got.Task.SourceSeq != 1 {
		t.Fatal("card watermark not committed", e)
	}
	for _, channel := range []string{c.ConversationID, "project:" + c.ProjectID} {
		rows, e := st.EventsAfter(channel, 0, 20)
		if e != nil || len(rows) != 1 {
			t.Fatal("event missing", e)
		}
	}
	snapshot, e := st.CardsAtCursor("o", c.ConversationID)
	if e != nil || snapshot.ReplaySeq != 1 || len(snapshot.Cards) != 1 {
		t.Fatal("inconsistent snapshot", snapshot, e)
	}
}
func TestTaskProjectionRejectsOwnerOldRevisionAndLateState(t *testing.T) {
	st := openTestStore(t)
	c := projectionFixture(t, st)
	if _, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: c}); e != nil {
		t.Fatal(e)
	}
	next := c
	copy := *c.Task
	next.Task = &copy
	copy.Revision = 2
	copy.ExecutionState = "completed"
	next.Status = core.CardDone
	if _, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "other", Card: next, ExpectedCardVersion: 1, ExpectedTaskRevision: 1}); e == nil {
		t.Fatal("foreign write accepted")
	}
	if _, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: next, ExpectedCardVersion: 1, ExpectedTaskRevision: 1}); e != nil {
		t.Fatal(e)
	}
	copy.Revision = 3
	copy.ExecutionState = "running"
	copy.RunID = "run1"
	if _, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: next, ExpectedCardVersion: 2, ExpectedTaskRevision: 2}); !errors.Is(e, ErrConflict) {
		t.Fatal("terminal regressed", e)
	}
}
func TestTaskProjectionEventFailureRollsBackCard(t *testing.T) {
	st := openTestStore(t)
	c := projectionFixture(t, st)
	var injection = `CREATE TRIGGER reject_task_event BEFORE INSERT ON events BEGIN SELECT RAISE(ABORT,'event injection'); END`
	if st.Dialect() == "postgres" {
		if _, e := st.db.Exec(`CREATE FUNCTION reject_task_event_fn() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'event injection'; END $$`); e != nil {
			t.Fatal(e)
		}
		injection = `CREATE TRIGGER reject_task_event BEFORE INSERT ON events FOR EACH ROW EXECUTE FUNCTION reject_task_event_fn()`
	}
	if _, e := st.db.Exec(injection); e != nil {
		t.Fatal(e)
	}
	_, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: c})
	if e == nil {
		t.Fatal("injection did not fail")
	}
	if _, e = st.GetCard("o", c.ID); !errors.Is(e, ErrNotFound) {
		t.Fatal("card survived rolled-back event", e)
	}
}

func TestTaskProjectionRejectsChangedSourceWitness(t *testing.T) {
	st := openTestStore(t)
	c := projectionFixture(t, st)
	a := core.Artifact{Owner: "o", ProjectID: c.ProjectID, Name: "input.txt"}
	if e := st.CreateArtifact(&a, []byte("old")); e != nil {
		t.Fatal(e)
	}
	witness := ArtifactWitness{ArtifactID: a.ID, UpdatedAt: a.UpdatedAt.Format(time.RFC3339Nano)}
	if e := st.UpdateArtifactContent("o", a.ID, []byte("new")); e != nil {
		t.Fatal(e)
	}
	if _, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: c, SourceWitnesses: []ArtifactWitness{witness}}); !errors.Is(e, ErrConflict) {
		t.Fatal("stale source proof accepted", e)
	}
}

func TestTaskProjectionConcurrentCASHasOneWinner(t *testing.T) {
	st := openTestStore(t)
	c := projectionFixture(t, st)
	if _, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: c}); e != nil {
		t.Fatal(e)
	}
	ch := make(chan error, 2)
	for i := 0; i < 2; i++ {
		go func() {
			next := c
			tcopy := *c.Task
			tcopy.Revision = 2
			tcopy.Focus = "核对"
			next.Task = &tcopy
			_, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: next, ExpectedTaskRevision: 1, ExpectedCardVersion: 1})
			ch <- e
		}()
	}
	winners := 0
	for i := 0; i < 2; i++ {
		if e := <-ch; e == nil {
			winners++
		} else if !errors.Is(e, ErrConflict) {
			t.Fatal(e)
		}
	}
	if winners != 1 {
		t.Fatal("CAS winners", winners)
	}
}

func TestTaskProjectionRequiresCanonicalStreamAndRejectsFalseCompletion(t *testing.T) {
	st := openTestStore(t)
	c := projectionFixture(t, st)
	if _, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: c, Channels: []string{"project:" + c.ProjectID}}); e == nil {
		t.Fatal("projection has no canonical conversation event")
	}
	if _, e := st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: c}); e != nil {
		t.Fatal(e)
	}
	current, ver, e := st.GetCardVersion("o", c.ID)
	if e != nil {
		t.Fatal(e)
	}
	current.Status = core.CardDone
	if _, e = st.CommitTaskProjection(TaskProjectionWrite{Owner: "o", Card: current, ExpectedCardVersion: ver, ExpectedTaskRevision: current.Task.Revision}); e == nil {
		t.Fatal("view update forged completion")
	}
}