Persist task projections and replay cursors atomically
Why: Background work needs consistent cards, events, and user reading state. Evidence: SELF make evidence records SQLite rollback, projection, pin, and regression checks; PostgreSQL replay is pending the composition gate. Impact: Adds an optional atomic task path and a consistent card replay cursor without schema changes. Refs: Agent-System-Dev fast-slow-u2-u3 plan Task 3
This commit is contained in:
@@ -27,8 +27,9 @@ var (
|
||||
const cardAppendWindow = 250 * time.Millisecond
|
||||
|
||||
type Service struct {
|
||||
st *store.Store
|
||||
emit Emitter
|
||||
TaskCommitter func(store.TaskProjectionWrite) ([]core.StreamEvent, error)
|
||||
st *store.Store
|
||||
emit Emitter
|
||||
// MirrorConvCards EXT-12：会话卡是否镜像进项目公共流（缺省 true=现状全量
|
||||
// 镜像；false 时会话卡只进本人会话流，项目级卡不受影响）。
|
||||
MirrorConvCards bool
|
||||
@@ -97,6 +98,13 @@ func (s *Service) List(owner, conversationID string) ([]core.ActivityCard, error
|
||||
// Update 保存整卡变更（流式 Body 增量等）并发布 card.updated；ref 保持原值。
|
||||
// 系统镜像路径（同会话 run 的顺序行为）沿用盲写；用户态裁决走 UpdateVersioned。
|
||||
func (s *Service) Update(owner string, card core.ActivityCard) error {
|
||||
if card.Task != nil {
|
||||
_, ver, err := s.st.GetCardVersion(owner, card.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return s.UpdateVersioned(owner, card, ver)
|
||||
}
|
||||
card.UpdatedAt = s.now().UTC()
|
||||
if err := s.st.SaveCard(owner, &card, ""); err != nil {
|
||||
return err
|
||||
@@ -107,6 +115,9 @@ func (s *Service) Update(owner string, card core.ActivityCard) error {
|
||||
// UpdateVersioned 乐观并发保存（U3，KTD7）：版本不匹配透传 store.ErrConflict，
|
||||
// HTTP 层映射 409；客户端重读卡后重试。
|
||||
func (s *Service) UpdateVersioned(owner string, card core.ActivityCard, expectedVersion int) error {
|
||||
if card.Task != nil {
|
||||
return s.commitProjection(store.TaskProjectionWrite{Owner: owner, Card: card, ExpectedCardVersion: expectedVersion, ExpectedTaskRevision: card.Task.Revision, Channels: s.taskChannels(card)})
|
||||
}
|
||||
card.UpdatedAt = s.now().UTC()
|
||||
if err := s.st.SaveCardVersioned(owner, &card, "", expectedVersion); err != nil {
|
||||
return err
|
||||
@@ -123,6 +134,9 @@ func (s *Service) Append(owner, id, text string) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if c.Task != nil {
|
||||
return ErrNoAction
|
||||
}
|
||||
if text != "" {
|
||||
if n := len(c.Body); n > 0 && c.Body[n-1].Type == core.UIMarkdown {
|
||||
c.Body[n-1].Markdown += text
|
||||
@@ -227,6 +241,9 @@ func (s *Service) ToProject(owner, id string) (core.ActivityCard, error) {
|
||||
if c.ConversationID == "" {
|
||||
return c, errors.New("card: already project-level")
|
||||
}
|
||||
if c.Task != nil {
|
||||
return c, errors.New("card: keep the original task reference in its conversation")
|
||||
}
|
||||
if ref, rerr := s.st.CardRef(owner, c.ID); rerr == nil {
|
||||
if strings.HasPrefix(ref, "hitl:") || strings.HasPrefix(ref, "cmd:") {
|
||||
return c, errors.New("card: flow-bound card cannot be archived to project")
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
package cardsvc
|
||||
|
||||
import (
|
||||
"agent-system-backend/internal/store"
|
||||
core "agent-system-core"
|
||||
"time"
|
||||
)
|
||||
|
||||
func taskCardState(task core.TaskSnapshot) (core.CardStatus, string) {
|
||||
switch task.ExecutionState {
|
||||
case "completed":
|
||||
return core.CardDone, "ok"
|
||||
case "failed", "interrupted":
|
||||
return core.CardFailed, "danger"
|
||||
case "awaiting_user":
|
||||
return core.CardActionRequired, "warn"
|
||||
default:
|
||||
return core.CardActive, "info"
|
||||
}
|
||||
}
|
||||
func (s *Service) commitProjection(w store.TaskProjectionWrite) error {
|
||||
if s.TaskCommitter != nil {
|
||||
_, err := s.TaskCommitter(w)
|
||||
return err
|
||||
}
|
||||
_, err := s.st.CommitTaskProjection(w)
|
||||
return err
|
||||
}
|
||||
func (s *Service) taskChannels(card core.ActivityCard) []string {
|
||||
channels := []string{card.ConversationID}
|
||||
if s.MirrorConvCards && card.ProjectID != "" {
|
||||
channels = append(channels, string(core.ProjectChannelKey(card.ProjectID)))
|
||||
}
|
||||
return channels
|
||||
}
|
||||
func (s *Service) CommitTask(owner string, next core.TaskSnapshot, expectedRevision int64) error {
|
||||
c, version, err := s.st.GetCardVersion(owner, next.CardID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
c.Task = &next
|
||||
c.Body = next.SummaryBody()
|
||||
c.Status, c.Tone = taskCardState(next)
|
||||
c.UpdatedAt = time.Now().UTC()
|
||||
return s.commitProjection(store.TaskProjectionWrite{Owner: owner, Card: c, ExpectedCardVersion: version, ExpectedTaskRevision: expectedRevision, Channels: s.taskChannels(c)})
|
||||
}
|
||||
func (s *Service) CreateTask(owner string, card core.ActivityCard, ref string) (core.ActivityCard, error) {
|
||||
if card.Task == nil {
|
||||
return card, store.ErrConflict
|
||||
}
|
||||
card.Status, card.Tone = taskCardState(*card.Task)
|
||||
card.Body = card.Task.SummaryBody()
|
||||
card.CreatedAt = time.Now().UTC()
|
||||
card.UpdatedAt = card.CreatedAt
|
||||
if err := s.commitProjection(store.TaskProjectionWrite{Owner: owner, Card: card, Ref: ref, Channels: s.taskChannels(card)}); err != nil {
|
||||
return card, err
|
||||
}
|
||||
return s.st.GetCard(owner, card.ID)
|
||||
}
|
||||
@@ -0,0 +1,63 @@
|
||||
package cardsvc
|
||||
|
||||
import (
|
||||
core "agent-system-core"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestTaskProjectionPreservesUserReadingPreferences(t *testing.T) {
|
||||
st, svc, _ := openTest(t)
|
||||
c := core.ActivityCard{ID: "card-task", ProjectID: "p1", ConversationID: "c1", Title: "任务", Kind: core.CardKindProgress}
|
||||
c.Task = &core.TaskSnapshot{SchemaVersion: 1, TaskID: "t1", AttemptID: "a1", CardID: c.ID, ProjectID: "p1", ConversationID: "c1", OwnerInstanceID: "i1", StreamScope: "conversation:c1", Revision: 1, ExecutionState: "queued", OutcomeCompleteness: "none", DeliveryState: "pending", Title: "任务"}
|
||||
if _, e := svc.CreateTask("o", c, ""); e != nil {
|
||||
t.Fatal(e)
|
||||
}
|
||||
current, e := st.GetCard("o", c.ID)
|
||||
if e != nil {
|
||||
t.Fatal("task card not saved", e)
|
||||
}
|
||||
current.Pinned = true
|
||||
current.Status = core.CardDismissed
|
||||
if e = st.SaveCard("o", ¤t, ""); e != nil {
|
||||
t.Fatal(e)
|
||||
}
|
||||
next := *current.Task
|
||||
next.Revision = 2
|
||||
next.ExecutionState = "completed"
|
||||
next.Conclusion = "有用成果"
|
||||
if e = svc.CommitTask("o", next, 1); e != nil {
|
||||
t.Fatal(e)
|
||||
}
|
||||
got, e := st.GetCard("o", c.ID)
|
||||
if e != nil || !got.Pinned || got.Status != core.CardDismissed || got.Task.ExecutionState != "completed" {
|
||||
t.Fatal("work update stole user state", got, e)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskProjectionUserPinHasAtomicEventWatermark(t *testing.T) {
|
||||
st, svc, _ := openTest(t)
|
||||
c := core.ActivityCard{ID: "card-pin", ProjectID: "p1", ConversationID: "c1", Title: "任务", Kind: core.CardKindProgress}
|
||||
c.Task = &core.TaskSnapshot{SchemaVersion: 1, TaskID: "t1", AttemptID: "a1", CardID: c.ID, ProjectID: "p1", ConversationID: "c1", OwnerInstanceID: "i1", StreamScope: "conversation:c1", Revision: 1, ExecutionState: "queued", OutcomeCompleteness: "none", DeliveryState: "pending", Title: "任务"}
|
||||
if _, e := svc.CreateTask("o", c, ""); e != nil {
|
||||
t.Fatal(e)
|
||||
}
|
||||
if _, e := svc.SetPinned("o", c.ID, true); e != nil {
|
||||
t.Fatal(e)
|
||||
}
|
||||
snapshot, e := st.CardsAtCursor("o", "c1")
|
||||
if e != nil || snapshot.ReplaySeq != 2 || len(snapshot.Cards) != 1 || !snapshot.Cards[0].Pinned {
|
||||
t.Fatal("pin not atomically replayable", snapshot, e)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskProjectionCannotCreateSecondExecutionOwner(t *testing.T) {
|
||||
_, svc, _ := openTest(t)
|
||||
c := core.ActivityCard{ID: "card-archive", ProjectID: "p1", ConversationID: "c1", Title: "任务", Kind: core.CardKindProgress}
|
||||
c.Task = &core.TaskSnapshot{SchemaVersion: 1, TaskID: "t1", AttemptID: "a1", CardID: c.ID, ProjectID: "p1", ConversationID: "c1", OwnerInstanceID: "i1", StreamScope: "conversation:c1", Revision: 1, ExecutionState: "queued", OutcomeCompleteness: "none", DeliveryState: "pending", Title: "任务"}
|
||||
if _, e := svc.CreateTask("o", c, ""); e != nil {
|
||||
t.Fatal(e)
|
||||
}
|
||||
if _, e := svc.ToProject("o", c.ID); e == nil {
|
||||
t.Fatal("task was copied with a second execution owner")
|
||||
}
|
||||
}
|
||||
@@ -157,3 +157,40 @@ func (eb *eventBuffer) flushAll() {
|
||||
eb.flushConversation(id)
|
||||
}
|
||||
}
|
||||
|
||||
// commitTask keeps buffered records and the work projection in one transaction.
|
||||
func (eb *eventBuffer) commitTask(w store.TaskProjectionWrite) ([]core.StreamEvent, error) {
|
||||
eb.flushing.Lock()
|
||||
defer eb.flushing.Unlock()
|
||||
channels := append([]string(nil), w.Channels...)
|
||||
if len(channels) == 0 {
|
||||
channels = []string{w.Card.ConversationID}
|
||||
}
|
||||
eb.mu.Lock()
|
||||
for _, ch := range channels {
|
||||
w.Buffered = append(w.Buffered, eb.buf[ch]...)
|
||||
delete(eb.buf, ch)
|
||||
}
|
||||
eb.mu.Unlock()
|
||||
events, err := eb.store.CommitTaskProjection(w)
|
||||
if err != nil {
|
||||
eb.mu.Lock()
|
||||
for i := len(w.Buffered) - 1; i >= 0; i-- {
|
||||
e := w.Buffered[i]
|
||||
eb.buf[e.ConversationID] = append([]core.StreamEvent{e}, eb.buf[e.ConversationID]...)
|
||||
}
|
||||
eb.mu.Unlock()
|
||||
return nil, err
|
||||
}
|
||||
grouped := map[string][]core.StreamEvent{}
|
||||
for _, e := range events {
|
||||
eb.bus.Publish(e)
|
||||
grouped[e.ConversationID] = append(grouped[e.ConversationID], e)
|
||||
}
|
||||
if eb.onPersisted != nil {
|
||||
for ch, batch := range grouped {
|
||||
eb.onPersisted(ch, batch)
|
||||
}
|
||||
}
|
||||
return events, nil
|
||||
}
|
||||
|
||||
@@ -180,7 +180,12 @@ type IntentGate interface {
|
||||
func (m *Manager) SetGate(g IntentGate) { m.gate = g }
|
||||
|
||||
// SetCards 注入收件箱卡服务（nil = 不产卡；T0 退回基线仍成立）。
|
||||
func (m *Manager) SetCards(s *cardsvc.Service) { m.cards = s }
|
||||
func (m *Manager) SetCards(s *cardsvc.Service) {
|
||||
m.cards = s
|
||||
if s != nil {
|
||||
s.TaskCommitter = m.events.commitTask
|
||||
}
|
||||
}
|
||||
|
||||
// EmitCardEvent 供服务层（cardsvc 投影）复用会话事件管道（持久化 + seq + 扇出）。
|
||||
func (m *Manager) EmitCardEvent(conversationID, runID string, typ core.StreamEventType, payload any) error {
|
||||
|
||||
@@ -1144,8 +1144,8 @@ func (s *Server) listCards(w http.ResponseWriter, r *http.Request) {
|
||||
writeStoreErr(w, err)
|
||||
return
|
||||
}
|
||||
cards, err := s.Cards.List(owner, r.PathValue("id"))
|
||||
writeJSONOr(w, map[string]any{"cards": cards}, err)
|
||||
snapshot, err := s.st.CardsAtCursor(owner, r.PathValue("id"))
|
||||
writeJSONOr(w, snapshot, err)
|
||||
}
|
||||
|
||||
// listProjectCards 项目收件箱聚合（D49 拍板口径）：项目级卡 ∪ 该项目全部会话
|
||||
|
||||
@@ -274,6 +274,21 @@ func (s *Store) AppendEvents(conversationID string, evs []core.StreamEvent) (int
|
||||
return 0, err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
base, err := s.appendEventsTx(tx, conversationID, evs)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if err = tx.Commit(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return base, nil
|
||||
}
|
||||
|
||||
// appendEventsTx shares sequence allocation and transactional NOTIFY with projections.
|
||||
func (s *Store) appendEventsTx(tx *sql.Tx, conversationID string, evs []core.StreamEvent) (int64, error) {
|
||||
if len(evs) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
if err := s.dialect.lockConversation(tx, conversationID); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
@@ -306,9 +321,6 @@ func (s *Store) AppendEvents(conversationID string, evs []core.StreamEvent) (int
|
||||
return 0, err
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
for i := range evs {
|
||||
evs[i].Seq = base.Int64 + int64(i) + 1
|
||||
}
|
||||
|
||||
@@ -0,0 +1,226 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
core "agent-system-core"
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
)
|
||||
|
||||
type TaskProjectionWrite struct {
|
||||
Owner string
|
||||
Card core.ActivityCard
|
||||
Ref string
|
||||
ExpectedCardVersion int
|
||||
ExpectedTaskRevision int64
|
||||
Channels []string
|
||||
Buffered []core.StreamEvent
|
||||
}
|
||||
type CardsSnapshot struct {
|
||||
Cards []core.ActivityCard `json:"cards"`
|
||||
ReplaySeq int64 `json:"replaySeq"`
|
||||
}
|
||||
|
||||
func (s *Store) CommitTaskProjection(w TaskProjectionWrite) ([]core.StreamEvent, error) {
|
||||
c := w.Card
|
||||
if c.Task == nil || c.Task.CardID != c.ID || c.Task.ConversationID != c.ConversationID || c.Task.ProjectID != c.ProjectID {
|
||||
return nil, fmt.Errorf("task.scope: card binding mismatch")
|
||||
}
|
||||
task := *c.Task
|
||||
c.Task = &task
|
||||
if err := task.Validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
channels := append([]string(nil), w.Channels...)
|
||||
if len(channels) == 0 {
|
||||
channels = []string{c.ConversationID}
|
||||
}
|
||||
sort.Strings(channels)
|
||||
unique := channels[:0]
|
||||
for _, ch := range channels {
|
||||
if ch != c.ConversationID && ch != "project:"+c.ProjectID {
|
||||
return nil, fmt.Errorf("task.scope: foreign channel")
|
||||
}
|
||||
if len(unique) == 0 || unique[len(unique)-1] != ch {
|
||||
unique = append(unique, ch)
|
||||
}
|
||||
}
|
||||
channels = unique
|
||||
tx, err := s.db.Begin()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
for _, ch := range channels {
|
||||
if err = s.dialect.lockConversation(tx, ch); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
var oldPayload, oldOwner string
|
||||
var version int
|
||||
err = tx.QueryRow(s.q(`SELECT payload,owner,version FROM activity_cards WHERE id=?`), c.ID).Scan(&oldPayload, &oldOwner, &version)
|
||||
creating := errors.Is(err, sql.ErrNoRows)
|
||||
viewOnly := false
|
||||
if err != nil && !creating {
|
||||
return nil, err
|
||||
}
|
||||
if creating {
|
||||
if w.ExpectedCardVersion != 0 || w.ExpectedTaskRevision != 0 || task.Revision != 1 {
|
||||
return nil, ErrConflict
|
||||
}
|
||||
} else {
|
||||
if oldOwner != w.Owner {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
var old core.ActivityCard
|
||||
if err = json.Unmarshal([]byte(oldPayload), &old); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if version != w.ExpectedCardVersion || old.Task == nil || old.Task.Revision != w.ExpectedTaskRevision || (task.Revision != old.Task.Revision+1 && task.Revision != old.Task.Revision) || old.Task.TaskID != task.TaskID || old.Task.AttemptID != task.AttemptID || old.ConversationID != c.ConversationID || old.ProjectID != c.ProjectID || old.Task.OwnerInstanceID != task.OwnerInstanceID {
|
||||
return nil, ErrConflict
|
||||
}
|
||||
if task.Revision == old.Task.Revision {
|
||||
a, _ := json.Marshal(old.Task)
|
||||
b, _ := json.Marshal(task)
|
||||
if !bytes.Equal(a, b) {
|
||||
return nil, ErrConflict
|
||||
}
|
||||
viewOnly = true
|
||||
}
|
||||
if old.Task.Terminal() && old.Task.ExecutionState != task.ExecutionState {
|
||||
return nil, ErrConflict
|
||||
}
|
||||
if old.Task.RunID != "" && old.Task.RunID != task.RunID {
|
||||
return nil, ErrConflict
|
||||
}
|
||||
if task.ExecutionEpoch < old.Task.ExecutionEpoch {
|
||||
return nil, ErrConflict
|
||||
}
|
||||
if !viewOnly {
|
||||
c.Pinned = old.Pinned
|
||||
if old.Status == core.CardDismissed || old.Status == core.CardResolved {
|
||||
c.Status = old.Status
|
||||
}
|
||||
}
|
||||
}
|
||||
var events []core.StreamEvent
|
||||
for _, ch := range channels {
|
||||
var batch []core.StreamEvent
|
||||
for _, e := range w.Buffered {
|
||||
if e.ConversationID == ch {
|
||||
batch = append(batch, e)
|
||||
}
|
||||
}
|
||||
if len(batch) > 0 {
|
||||
if _, err = s.appendEventsTx(tx, ch, batch); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
events = append(events, batch...)
|
||||
}
|
||||
}
|
||||
if !viewOnly {
|
||||
if err = tx.QueryRow(s.q(`SELECT COALESCE(MAX(seq),0)+1 FROM events WHERE conversation_id=?`), c.ConversationID).Scan(&task.SourceSeq); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
raw, err := json.Marshal(c)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
c.UpdatedAt = now
|
||||
if c.CreatedAt.IsZero() {
|
||||
c.CreatedAt = now
|
||||
}
|
||||
raw, err = json.Marshal(c)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if creating {
|
||||
_, err = tx.Exec(s.q(`INSERT INTO activity_cards(id,project_id,conversation_id,ref,owner,payload,status,pinned,created_at,updated_at,version) VALUES(?,?,?,?,?,?,?,?,?,?,1)`), c.ID, c.ProjectID, c.ConversationID, w.Ref, w.Owner, string(raw), c.Status, boolInt(c.Pinned), c.CreatedAt.Format(time.RFC3339Nano), c.UpdatedAt.Format(time.RFC3339Nano))
|
||||
} else {
|
||||
var res sql.Result
|
||||
res, err = tx.Exec(s.q(`UPDATE activity_cards SET payload=?,status=?,pinned=?,updated_at=?,version=version+1 WHERE id=? AND owner=? AND version=?`), string(raw), c.Status, boolInt(c.Pinned), c.UpdatedAt.Format(time.RFC3339Nano), c.ID, w.Owner, version)
|
||||
if err == nil {
|
||||
n, _ := res.RowsAffected()
|
||||
if n != 1 {
|
||||
return nil, ErrConflict
|
||||
}
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
if isUniqueViolation(err) {
|
||||
return nil, ErrConflict
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
payload, err := json.Marshal(core.CardPayload{Card: c})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, ch := range channels {
|
||||
e := core.StreamEvent{ConversationID: ch, RunID: task.RunID, TS: now.Format(time.RFC3339Nano), V: core.ProtocolVersion, Type: core.EvCardUpdated, Payload: payload}
|
||||
if creating {
|
||||
e.Type = core.EvCardCreated
|
||||
}
|
||||
if pid, ok := core.ProjectFromChannel(core.ID(ch)); ok {
|
||||
e.ProjectID = pid
|
||||
}
|
||||
batch := []core.StreamEvent{e}
|
||||
if _, err = s.appendEventsTx(tx, ch, batch); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
events = append(events, batch...)
|
||||
}
|
||||
if err = tx.Commit(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return events, nil
|
||||
}
|
||||
func (s *Store) CardsAtCursor(owner, conversationID string) (CardsSnapshot, error) {
|
||||
opts := &sql.TxOptions{}
|
||||
if s.Dialect() == "postgres" {
|
||||
opts.Isolation = sql.LevelRepeatableRead
|
||||
opts.ReadOnly = true
|
||||
}
|
||||
tx, err := s.db.BeginTx(context.Background(), opts)
|
||||
if err != nil {
|
||||
return CardsSnapshot{}, err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
rows, err := tx.Query(s.q(`SELECT payload FROM activity_cards WHERE conversation_id=? AND owner=? ORDER BY updated_at DESC`), conversationID, owner)
|
||||
if err != nil {
|
||||
return CardsSnapshot{}, err
|
||||
}
|
||||
out := CardsSnapshot{Cards: []core.ActivityCard{}}
|
||||
for rows.Next() {
|
||||
var raw string
|
||||
if err = rows.Scan(&raw); err != nil {
|
||||
rows.Close()
|
||||
return out, err
|
||||
}
|
||||
var c core.ActivityCard
|
||||
if err = json.Unmarshal([]byte(raw), &c); err != nil {
|
||||
rows.Close()
|
||||
return out, err
|
||||
}
|
||||
out.Cards = append(out.Cards, c)
|
||||
}
|
||||
err = rows.Err()
|
||||
rows.Close()
|
||||
if err != nil {
|
||||
return out, err
|
||||
}
|
||||
if err = tx.QueryRow(s.q(`SELECT COALESCE(MAX(seq),0) FROM events WHERE conversation_id=?`), conversationID).Scan(&out.ReplaySeq); err != nil {
|
||||
return out, err
|
||||
}
|
||||
if err = tx.Commit(); err != nil {
|
||||
return out, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
@@ -0,0 +1,92 @@
|
||||
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.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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user