ci / vet-and-test (pull_request) Successful in 49s
Why: Legacy project streams replayed upload filenames and parser metadata to newly authorized members even though the underlying cards remained owner-only. Evidence: SELF real HTTP red regressions exposed legacy replay, spoofed sharing claims, private-history pagination, and live delivery after membership removal. The affected packages, uncached full Backend suite, isolated PostgreSQL Store suite, and agent-system build pass after applying the event boundary. Impact: Registered lifecycle streams resolve trusted resource ownership and require an internal publication proof for cross-owner delivery. Event and proof commit in one transaction; v30 adds an additive SQLite/PostgreSQL sidecar without granting access to historical events. Seen and visible cursors remain separate, revoked members leave live streams, and unregistered base behavior remains compatible. The storage migration tool preserves the new proof rows. Refs: project-member-access-implementation-plan.md Task 1 / R5; authorized legacy SSE privacy follow-up; project-member-composed-legacy-project-history.json
320 lines
12 KiB
Go
320 lines
12 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
|
||
|
||
var (
|
||
ErrNotFound = store.ErrNotFound // 语义对齐存储层:跨租户/不存在一律 NotFound
|
||
ErrNoAction = errors.New("card: action not found")
|
||
ErrNotAwaiting = errors.New("card: not awaiting action")
|
||
)
|
||
|
||
// cardAppendWindow 是流式追加(Append)的 card.updated 发布节流窗口(D45)。
|
||
const cardAppendWindow = 250 * time.Millisecond
|
||
|
||
type Service struct {
|
||
st *store.Store
|
||
emit Emitter
|
||
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}
|
||
}
|
||
|
||
// 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.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) { return s.st.GetCard(owner, id) }
|
||
|
||
// List 按会话列出(updated_at 降序;置顶与待办排序是前端关注点)。
|
||
func (s *Service) List(owner, conversationID string) ([]core.ActivityCard, error) {
|
||
return s.st.ListCards(owner, conversationID)
|
||
}
|
||
|
||
// Update 保存整卡变更(流式 Body 增量等)并发布 card.updated;ref 保持原值。
|
||
// 系统镜像路径(同会话 run 的顺序行为)沿用盲写;用户态裁决走 UpdateVersioned。
|
||
func (s *Service) Update(owner string, card core.ActivityCard) error {
|
||
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 {
|
||
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 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
|
||
}
|
||
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
|
||
return c, s.UpdateVersioned(owner, c, ver)
|
||
}
|
||
|
||
// 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
|
||
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 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
|
||
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
|
||
}
|