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

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

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