399 lines
14 KiB
Go
399 lines
14 KiB
Go
// Package cardsvc 是收件箱任务卡的服务层：创建/状态迁移/置顶/弃置/动作裁决，
|
||
// 每次变更持久化并经事件通道发布 card.* 投影（服务端权威，前端是投影）。
|
||
// 确认动作的语义回执由调用方写进对话时间线（resolved 事件即审计锚点）。
|
||
package cardsvc
|
||
|
||
import (
|
||
"errors"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"agent-system-backend/internal/store"
|
||
core "agent-system-core"
|
||
)
|
||
|
||
// Emitter 由宿主注入（runmgr 事件缓冲包装）：卡片事件走与会话事件同一管道，
|
||
// 获得持久化、seq 排序、断线重放与终态对账（Stream 契约）。
|
||
type Emitter func(conversationID, runID string, typ core.StreamEventType, payload any) error
|
||
|
||
// AccountCheckedEmitter persists an actor-attributed event through its own
|
||
// final account-checked Store transaction. The card write and event emission
|
||
// are separate commits; the application wires this to its guarded event sink.
|
||
type AccountCheckedEmitter func(actor, conversationID, runID string, typ core.StreamEventType, payload any) error
|
||
|
||
var (
|
||
ErrNotFound = store.ErrNotFound // 语义对齐存储层：跨租户/不存在一律 NotFound
|
||
ErrNoAction = errors.New("card: action not found")
|
||
ErrNotAwaiting = errors.New("card: not awaiting action")
|
||
ErrGuardedEmitterRequired = errors.New("card: account-checked emitter required")
|
||
)
|
||
|
||
// cardAppendWindow 是流式追加（Append）的 card.updated 发布节流窗口（D45）。
|
||
const cardAppendWindow = 250 * time.Millisecond
|
||
|
||
type Service struct {
|
||
TaskCommitter func(store.TaskProjectionWrite) ([]core.StreamEvent, error)
|
||
st *store.Store
|
||
emit Emitter
|
||
accountCheckedEmit AccountCheckedEmitter
|
||
projectSharedEmit ProjectSharedEmitter
|
||
// MirrorConvCards EXT-12：会话卡是否镜像进项目公共流（缺省 true=现状全量
|
||
// 镜像；false 时会话卡只进本人会话流，项目级卡不受影响）。
|
||
MirrorConvCards bool
|
||
now func() time.Time
|
||
emitMu sync.Mutex
|
||
lastEmit map[string]time.Time // 流式追加的最近发布时刻（按卡）
|
||
}
|
||
|
||
func New(st *store.Store, emit Emitter) *Service {
|
||
return &Service{st: st, emit: emit, now: time.Now, lastEmit: map[string]time.Time{}, MirrorConvCards: true}
|
||
}
|
||
|
||
func (s *Service) SetAccountCheckedEmitter(emit AccountCheckedEmitter) {
|
||
s.accountCheckedEmit = emit
|
||
}
|
||
|
||
// ProjectSharedEmitter is an internal trusted publication adapter. It is never
|
||
// bound to model arguments or exposed as a mandatory extension capability.
|
||
type ProjectSharedEmitter func(owner, producer, channel, runID string, typ core.StreamEventType, payload any) error
|
||
|
||
func (s *Service) SetProjectSharedEmitter(emit ProjectSharedEmitter) { s.projectSharedEmit = emit }
|
||
|
||
// PublishProjectCard explicitly publishes a project-level structural snapshot.
|
||
// Ordinary Create/Update remain private unless this entry issues a new proof.
|
||
func (s *Service) PublishProjectCard(owner, producer string, card core.ActivityCard, ref string) (core.ActivityCard, error) {
|
||
if s.projectSharedEmit == nil || owner == "" || producer == "" || card.ProjectID == "" || card.ConversationID != "" {
|
||
return card, errors.New("card: trusted project publisher required")
|
||
}
|
||
if card.ID != "" {
|
||
current, err := s.st.GetCard(owner, card.ID)
|
||
if err != nil {
|
||
return card, err
|
||
}
|
||
if current.ProjectID != card.ProjectID || current.ConversationID != "" {
|
||
return card, store.ErrNotFound
|
||
}
|
||
} else {
|
||
card.ID = store.NewID("card")
|
||
}
|
||
now := s.now().UTC()
|
||
card.CreatedAt = now
|
||
card.UpdatedAt = now
|
||
if card.Status == "" {
|
||
card.Status = core.CardActive
|
||
}
|
||
if err := s.st.SaveCard(owner, &card, ref); err != nil {
|
||
return card, err
|
||
}
|
||
err := s.projectSharedEmit(owner, producer, string(core.ProjectChannelKey(card.ProjectID)), card.SourceRunID, core.EvCardCreated, core.CardPayload{Card: card})
|
||
return card, err
|
||
}
|
||
|
||
// emitCard 按卡归属发布卡片事件（D50）：
|
||
// - 会话卡 → 会话流；同时镜像项目流（伪会话键 "project:<projectId>"）——
|
||
// 项目收件箱聚合视图（项目级卡 ∪ 各会话未决卡）的实时性依赖它；卡片事件
|
||
// 量小，全量镜像换取口径简单。
|
||
// - 项目级卡（ConversationID 空、ProjectID 非空）→ 只发布项目流。
|
||
// - 无归属（收件箱关闭期的纯持久化）不发布。
|
||
func (s *Service) emitCard(c core.ActivityCard, typ core.StreamEventType, payload any) error {
|
||
if s.emit == nil || (c.ConversationID == "" && c.ProjectID == "") {
|
||
return nil
|
||
}
|
||
if c.ConversationID != "" {
|
||
if err := s.emit(c.ConversationID, c.SourceRunID, typ, payload); err != nil {
|
||
return err
|
||
}
|
||
// EXT-12 最小开关：会话卡不镜像进项目公共流（不发布即不持久，
|
||
// 重放天然不含会话卡）；项目级卡（无会话归属）不受影响。
|
||
if !s.MirrorConvCards {
|
||
return nil
|
||
}
|
||
}
|
||
if c.ProjectID != "" {
|
||
return s.emit(string(core.ProjectChannelKey(c.ProjectID)), c.SourceRunID, typ, payload)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Create 补全标识与时间戳、持久化并发布 card.created。ref 为投影引用
|
||
// （HITL 投影卡 = "hitl:<id>"，to_project 留档副本 = "to_project:<原卡ID>"）；
|
||
// ConversationID 为空的项目级卡发布到项目流。
|
||
func (s *Service) Create(owner string, card core.ActivityCard, ref string) (core.ActivityCard, error) {
|
||
if card.Task != nil {
|
||
return card, store.ErrConflict
|
||
}
|
||
if card.ID != "" {
|
||
if existing, err := s.st.GetCard(owner, card.ID); err == nil && existing.Task != nil {
|
||
return card, store.ErrConflict
|
||
}
|
||
}
|
||
if card.ID == "" {
|
||
card.ID = store.NewID("card")
|
||
}
|
||
now := s.now().UTC()
|
||
card.CreatedAt = now
|
||
card.UpdatedAt = now
|
||
if card.Status == "" {
|
||
card.Status = core.CardActive
|
||
}
|
||
if err := s.st.SaveCard(owner, &card, ref); err != nil {
|
||
return card, err
|
||
}
|
||
return card, s.emitCard(card, core.EvCardCreated, core.CardPayload{Card: card})
|
||
}
|
||
|
||
// Get 读取单卡。
|
||
func (s *Service) Get(owner, id string) (core.ActivityCard, error) {
|
||
c, err := s.st.GetCard(owner, id)
|
||
if err == nil {
|
||
c = s.present(owner, c)
|
||
}
|
||
return c, err
|
||
}
|
||
|
||
// List 按会话列出（updated_at 降序；置顶与待办排序是前端关注点）。
|
||
func (s *Service) List(owner, conversationID string) ([]core.ActivityCard, error) {
|
||
cards, err := s.st.ListCards(owner, conversationID)
|
||
if err == nil {
|
||
cards = s.PresentCards(owner, cards)
|
||
}
|
||
return cards, err
|
||
}
|
||
|
||
// PresentCards applies the read-time projections to cards read elsewhere
|
||
// (for example a cursor snapshot), in place.
|
||
func (s *Service) PresentCards(owner string, cards []core.ActivityCard) []core.ActivityCard {
|
||
for i := range cards {
|
||
cards[i] = s.present(owner, cards[i])
|
||
}
|
||
return cards
|
||
}
|
||
|
||
// present applies every read-time projection: resolved HITL answers and
|
||
// task source visibility.
|
||
func (s *Service) present(owner string, c core.ActivityCard) core.ActivityCard {
|
||
return s.VisibleTaskCard(owner, s.presentResolved(owner, c))
|
||
}
|
||
|
||
// Update 保存整卡变更（流式 Body 增量等）并发布 card.updated；ref 保持原值。
|
||
// 系统镜像路径（同会话 run 的顺序行为）沿用盲写；用户态裁决走 UpdateVersioned。
|
||
func (s *Service) Update(owner string, card core.ActivityCard) error {
|
||
if card.Task == nil {
|
||
if old, e := s.st.GetCard(owner, card.ID); e == nil && old.Task != nil {
|
||
return store.ErrConflict
|
||
}
|
||
}
|
||
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
|
||
}
|
||
return s.emitCard(card, core.EvCardUpdated, core.CardPayload{Card: card})
|
||
}
|
||
|
||
// UpdateVersioned 乐观并发保存（U3，KTD7）：版本不匹配透传 store.ErrConflict，
|
||
// HTTP 层映射 409；客户端重读卡后重试。
|
||
func (s *Service) UpdateVersioned(owner string, card core.ActivityCard, expectedVersion int) error {
|
||
if card.Task == nil {
|
||
if current, err := s.st.GetCard(owner, card.ID); err == nil && current.Task != nil {
|
||
return store.ErrConflict
|
||
}
|
||
}
|
||
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
|
||
}
|
||
return s.emitCard(card, core.EvCardUpdated, core.CardPayload{Card: card})
|
||
}
|
||
|
||
// Append 流式追加正文（card_append 工具；D45）：文本并入 body 尾部 markdown
|
||
// 节点（尾节点非 markdown 则新增），逐次持久化；card.updated 按 cardAppendWindow
|
||
// 节流——窗口内的追加只落库不发布（投影语义：事件携带整卡，后续事件自然覆盖
|
||
// 省略的中间投影）。终态发布不受节流影响（SetStatus/Update 恒发布）。
|
||
func (s *Service) Append(owner, id, text string) error {
|
||
c, err := s.st.GetCard(owner, id)
|
||
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
|
||
} else {
|
||
c.Body = append(c.Body, core.UINode{Type: core.UIMarkdown, Markdown: text})
|
||
}
|
||
}
|
||
c.UpdatedAt = s.now().UTC()
|
||
if err := s.st.SaveCard(owner, &c, ""); err != nil {
|
||
return err
|
||
}
|
||
if c.ConversationID == "" && c.ProjectID == "" {
|
||
return nil // 无归属：只落库不发布
|
||
}
|
||
s.emitMu.Lock()
|
||
last, seen := s.lastEmit[id]
|
||
emit := !seen || s.now().Sub(last) >= cardAppendWindow
|
||
if emit {
|
||
s.lastEmit[id] = s.now()
|
||
}
|
||
s.emitMu.Unlock()
|
||
if !emit {
|
||
return nil
|
||
}
|
||
return s.emitCard(c, core.EvCardUpdated, core.CardPayload{Card: c})
|
||
}
|
||
|
||
// SetStatus 状态迁移：更新并发布 card.status。
|
||
func (s *Service) SetStatus(owner, id string, status core.CardStatus, tone, note string) (core.ActivityCard, error) {
|
||
c, ver, err := s.st.GetCardVersion(owner, id)
|
||
if err != nil {
|
||
return c, err
|
||
}
|
||
c.Status = status
|
||
if tone != "" {
|
||
c.Tone = tone
|
||
}
|
||
if err := s.UpdateVersioned(owner, c, ver); err != nil {
|
||
return c, err
|
||
}
|
||
if c.Task != nil {
|
||
return s.Get(owner, c.ID)
|
||
}
|
||
return c, s.emitCard(c, core.EvCardStatus, core.CardStatusPayload{CardID: c.ID, Status: status, Tone: c.Tone, Note: note})
|
||
}
|
||
|
||
// SetPinned 置顶/取消（收件箱"不沉没"语义）。
|
||
func (s *Service) SetPinned(owner, id string, pinned bool) (core.ActivityCard, error) {
|
||
c, ver, err := s.st.GetCardVersion(owner, id)
|
||
if err != nil {
|
||
return c, err
|
||
}
|
||
c.Pinned = pinned
|
||
if e := s.UpdateVersioned(owner, c, ver); e != nil {
|
||
return c, e
|
||
}
|
||
return s.Get(owner, c.ID)
|
||
}
|
||
|
||
// Dismiss 关闭卡（非确认语义，留痕）。
|
||
func (s *Service) Dismiss(owner, id string) (core.ActivityCard, error) {
|
||
return s.SetStatus(owner, id, core.CardDismissed, "", "dismissed by user")
|
||
}
|
||
|
||
// Action 处理卡上按钮点击：仅 action_required 卡可裁决；结果 resolved 并发布
|
||
// card.resolved（ActionID 即按钮语义）。
|
||
func (s *Service) Action(owner, id, actionID, freeText string) (core.ActivityCard, string, error) {
|
||
c, ver, err := s.st.GetCardVersion(owner, id)
|
||
if err != nil {
|
||
return c, "", err
|
||
}
|
||
if c.Status != core.CardActionRequired {
|
||
return c, "", ErrNotAwaiting
|
||
}
|
||
label := actionID
|
||
found := false
|
||
for _, a := range c.Actions {
|
||
if a.ID == actionID {
|
||
label = a.Label
|
||
found = true
|
||
break
|
||
}
|
||
}
|
||
switch {
|
||
case found:
|
||
case actionID == "" && freeText != "":
|
||
label = freeText // 自由输入语义
|
||
default:
|
||
return c, "", ErrNoAction
|
||
}
|
||
c.Status = core.CardResolved
|
||
recordAnswer(&c, actionID, freeText)
|
||
if err := s.UpdateVersioned(owner, c, ver); err != nil {
|
||
return c, "", err
|
||
}
|
||
return c, label, s.emitCard(c, core.EvCardResolved, core.CardResolvedPayload{CardID: c.ID, ActionID: actionID, Answer: freeText})
|
||
}
|
||
|
||
// ToProject 把会话卡留档为项目级卡（D48/D49）：复制为新卡（ConversationID 空，
|
||
// AnchorMessageID 指向原卡，body 附留档来源），原卡 dismissed 留痕——不移动
|
||
// （会话时间线不可变）。ref "to_project:<原卡ID>" 保证幂等（重复调用返回既有
|
||
// 副本，不重复置态）。HITL/命令投影卡（ref 前缀 hitl:/cmd:）拒绝留档——它们
|
||
// 与流程状态机绑定，dismiss 会破坏销项语义。
|
||
func (s *Service) ToProject(owner, id string) (core.ActivityCard, error) {
|
||
c, err := s.st.GetCard(owner, id)
|
||
if err != nil {
|
||
return c, err
|
||
}
|
||
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")
|
||
}
|
||
}
|
||
ref := "to_project:" + c.ID
|
||
if existID, gerr := s.st.CardIDByRef(owner, ref); gerr == nil && existID != "" {
|
||
return s.st.GetCard(owner, existID) // 幂等：已留档
|
||
}
|
||
dup := c
|
||
dup.ID = store.NewID("card")
|
||
dup.ConversationID = ""
|
||
dup.AnchorMessageID = core.ID(c.ID)
|
||
dup.Pinned = false
|
||
dup.Body = append(append([]core.UINode{}, c.Body...), core.UINode{Type: core.UIKV, KV: [][2]string{{"留档", "自会话卡 " + c.ID}}})
|
||
created, err := s.Create(owner, dup, ref)
|
||
if err != nil {
|
||
return created, err
|
||
}
|
||
if _, err := s.SetStatus(owner, c.ID, core.CardDismissed, "info", "已留档到项目"); err != nil {
|
||
return created, err
|
||
}
|
||
return created, nil
|
||
}
|
||
|
||
// ResolveByRef 按引用解析投影卡（HITL resolve 时同步销项："hitl:<id>"）；幂等。
|
||
func (s *Service) ResolveByRef(owner, ref, actionID, answer string) error {
|
||
id, err := s.st.CardIDByRef(owner, ref)
|
||
if err != nil {
|
||
if errors.Is(err, store.ErrNotFound) {
|
||
return nil // 无投影卡（历史数据/非确认卡）——静默跳过
|
||
}
|
||
return err
|
||
}
|
||
c, ver, err := s.st.GetCardVersion(owner, id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if c.Status.Terminal() {
|
||
return nil // 已决（双现任一侧先到）
|
||
}
|
||
c.Status = core.CardResolved
|
||
recordAnswer(&c, actionID, answer)
|
||
if err := s.UpdateVersioned(owner, c, ver); err != nil {
|
||
if errors.Is(err, store.ErrConflict) {
|
||
return nil // 双现任一侧先到（版本竞争败者）：不重试，卡以先到侧为准
|
||
}
|
||
return err
|
||
}
|
||
if c.ConversationID != "" && s.emit != nil {
|
||
return s.emit(c.ConversationID, c.SourceRunID, core.EvCardResolved, core.CardResolvedPayload{CardID: c.ID, ActionID: actionID, Answer: answer})
|
||
}
|
||
return nil
|
||
}
|