344 lines
10 KiB
Go
344 lines
10 KiB
Go
package store
|
|
|
|
import (
|
|
core "agent-system-core"
|
|
"bytes"
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"time"
|
|
)
|
|
|
|
type ArtifactWitness struct {
|
|
ArtifactID string
|
|
UpdatedAt string
|
|
}
|
|
type TaskProjectionWrite struct {
|
|
SourceWitnesses []ArtifactWitness
|
|
Owner string
|
|
Card core.ActivityCard
|
|
Ref string
|
|
ExpectedCardVersion int
|
|
ExpectedTaskRevision int64
|
|
Channels []string
|
|
Buffered []core.StreamEvent
|
|
// AccountActor, when set, commits the projection under that actor's final
|
|
// effective-account fence; it must equal Owner.
|
|
AccountActor string
|
|
AllowImplicitDefault bool
|
|
}
|
|
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
|
|
foundConversation := false
|
|
for _, channel := range channels {
|
|
if channel == c.ConversationID {
|
|
foundConversation = true
|
|
}
|
|
}
|
|
if !foundConversation {
|
|
return nil, fmt.Errorf("task.scope: canonical stream required")
|
|
}
|
|
if w.AccountActor != "" {
|
|
if w.AccountActor != w.Owner {
|
|
return nil, ErrNotFound
|
|
}
|
|
var events []core.StreamEvent
|
|
err := s.withEffectiveAccountWrite(w.AccountActor, w.AllowImplicitDefault, func(tx *sql.Tx, _ Account) error {
|
|
if err := s.cardResourcesForWrite(tx, w.AccountActor, c.ProjectID, c.ConversationID); err != nil {
|
|
return err
|
|
}
|
|
var err error
|
|
events, err = s.commitTaskProjectionTx(tx, w, c, task, channels)
|
|
return err
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return events, nil
|
|
}
|
|
tx, err := s.db.Begin()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback()
|
|
events, err := s.commitTaskProjectionTx(tx, w, c, task, channels)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err = tx.Commit(); err != nil {
|
|
return nil, err
|
|
}
|
|
return events, nil
|
|
}
|
|
|
|
func (s *Store) commitTaskProjectionTx(tx *sql.Tx, w TaskProjectionWrite, c core.ActivityCard, task core.TaskSnapshot, channels []string) ([]core.StreamEvent, error) {
|
|
c.Task = &task
|
|
var err error
|
|
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 c.Status != old.Status && c.Status != core.CardDismissed && c.Status != core.CardResolved {
|
|
return nil, ErrConflict
|
|
}
|
|
}
|
|
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 task.InputRevision != old.Task.InputRevision || task.ParentRunID != old.Task.ParentRunID || task.SourceMessageID != old.Task.SourceMessageID {
|
|
return nil, ErrConflict
|
|
}
|
|
if !viewOnly {
|
|
c.Pinned = old.Pinned
|
|
if old.Status == core.CardDismissed || old.Status == core.CardResolved {
|
|
c.Status = old.Status
|
|
}
|
|
}
|
|
}
|
|
c.Body = task.SummaryBody()
|
|
c.SourceRunID = task.RunID
|
|
c.Title = task.Title
|
|
// Serialize source-version checks with the artifact overwrite transaction.
|
|
sort.Slice(w.SourceWitnesses, func(i, j int) bool { return w.SourceWitnesses[i].ArtifactID < w.SourceWitnesses[j].ArtifactID })
|
|
for _, src := range w.SourceWitnesses {
|
|
if err = s.dialect.lockScope(tx, w.Owner, "artifact:"+src.ArtifactID); err != nil {
|
|
return nil, err
|
|
}
|
|
var project, version string
|
|
if err = tx.QueryRow(s.q(`SELECT project_id,updated_at FROM artifacts WHERE id=? AND owner=?`), src.ArtifactID, w.Owner).Scan(&project, &version); err != nil {
|
|
return nil, err
|
|
}
|
|
if project != c.ProjectID || version != src.UpdatedAt {
|
|
return nil, ErrConflict
|
|
}
|
|
}
|
|
var events []core.StreamEvent
|
|
if !viewOnly && task.ExecutionEpoch > 0 {
|
|
if err = s.lockAdmission(tx); err != nil {
|
|
return nil, err
|
|
}
|
|
var instance, until string
|
|
var epoch int64
|
|
if err = tx.QueryRow(s.q(`SELECT instance_id,epoch,lease_until FROM conversation_executions WHERE conversation_id=?`), "task:"+task.TaskID).Scan(&instance, &epoch, &until); err != nil {
|
|
return nil, err
|
|
}
|
|
if instance != task.OwnerInstanceID || epoch != task.ExecutionEpoch || (!task.Terminal() && until <= nowISO()) {
|
|
return nil, ErrFenced
|
|
}
|
|
}
|
|
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 {
|
|
var base int64
|
|
if base, err = s.appendEventsInTx(tx, ch, batch, "", ""); err != nil {
|
|
return nil, err
|
|
}
|
|
assignEventSequences(batch, base)
|
|
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
|
|
}
|
|
}
|
|
if err = task.Validate(); 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}
|
|
var base int64
|
|
if base, err = s.appendEventsInTx(tx, ch, batch, "", ""); err != nil {
|
|
return nil, err
|
|
}
|
|
assignEventSequences(batch, base)
|
|
events = append(events, batch...)
|
|
}
|
|
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
|
|
}
|
|
|
|
type TaskCardRecord struct {
|
|
Owner string
|
|
Card core.ActivityCard
|
|
}
|
|
|
|
func (s *Store) TaskCards() ([]TaskCardRecord, error) {
|
|
rows, err := s.db.Query(`SELECT owner,payload FROM activity_cards`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []TaskCardRecord
|
|
for rows.Next() {
|
|
var r TaskCardRecord
|
|
var raw string
|
|
if err = rows.Scan(&r.Owner, &raw); err != nil {
|
|
return nil, err
|
|
}
|
|
if err = json.Unmarshal([]byte(raw), &r.Card); err != nil {
|
|
return nil, err
|
|
}
|
|
if r.Card.Task != nil {
|
|
out = append(out, r)
|
|
}
|
|
}
|
|
return out, rows.Err()
|
|
}
|