Add inbox activity cards: cardsvc, store, events, API and producers #8

Merged
weichen merged 1 commits from feature/inbox-r2 into main 2026-09-30 06:52:49 +00:00
11 changed files with 648 additions and 40 deletions
+6
View File
@@ -16,6 +16,7 @@ import (
"time"
"agent-system-backend/internal/bus"
"agent-system-backend/internal/cardsvc"
"agent-system-backend/internal/intentgate"
"agent-system-backend/internal/pgbus"
"agent-system-backend/internal/plugins"
@@ -101,7 +102,12 @@ func Run(opt Options) error {
mgr.SetRetriever(pipe)
mgr.SetIndexer(pipe)
// 收件箱任务卡:事件走会话事件管道(持久化/seq/重放);关闭即 T0 基线。
cardSvc := cardsvc.New(st, mgr.EmitCardEvent)
mgr.SetCards(cardSvc)
srv := server.New(st, mgr, reg, profSvc, setSvc, pluginHost)
srv.Cards = cardSvc
srv.Retriever = pipe
srv.Indexer = pipe
+175
View File
@@ -0,0 +1,175 @@
// Package cardsvc 是收件箱任务卡的服务层:创建/状态迁移/置顶/弃置/动作裁决,
// 每次变更持久化并经事件通道发布 card.* 投影(服务端权威,前端是投影)。
// 确认动作的语义回执由调用方写进对话时间线(resolved 事件即审计锚点)。
package cardsvc
import (
"errors"
"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")
)
type Service struct {
st *store.Store
emit Emitter
now func() time.Time
}
func New(st *store.Store, emit Emitter) *Service {
return &Service{st: st, emit: emit, now: time.Now}
}
// Create 补全标识与时间戳、持久化并发布 card.created。ref 为投影引用
// (HITL 投影卡 = "hitl:<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
}
if card.ConversationID != "" && s.emit != nil {
if err := s.emit(card.ConversationID, card.SourceRunID, core.EvCardCreated, core.CardPayload{Card: card}); err != nil {
return card, err
}
}
return card, nil
}
// 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 保持原值。
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
}
if card.ConversationID != "" && s.emit != nil {
return s.emit(card.ConversationID, card.SourceRunID, core.EvCardUpdated, core.CardPayload{Card: card})
}
return nil
}
// SetStatus 状态迁移:更新并发布 card.status。
func (s *Service) SetStatus(owner, id string, status core.CardStatus, tone, note string) (core.ActivityCard, error) {
c, err := s.st.GetCard(owner, id)
if err != nil {
return c, err
}
c.Status = status
if tone != "" {
c.Tone = tone
}
if err := s.Update(owner, c); err != nil {
return c, err
}
if c.ConversationID != "" && s.emit != nil {
if err := s.emit(c.ConversationID, c.SourceRunID, core.EvCardStatus, core.CardStatusPayload{CardID: c.ID, Status: status, Tone: c.Tone, Note: note}); err != nil {
return c, err
}
}
return c, nil
}
// SetPinned 置顶/取消(收件箱"不沉没"语义)。
func (s *Service) SetPinned(owner, id string, pinned bool) (core.ActivityCard, error) {
c, err := s.st.GetCard(owner, id)
if err != nil {
return c, err
}
c.Pinned = pinned
return c, s.Update(owner, c)
}
// 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, err := s.st.GetCard(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.Update(owner, c); err != nil {
return c, "", err
}
if c.ConversationID != "" && s.emit != nil {
if err := s.emit(c.ConversationID, c.SourceRunID, core.EvCardResolved, core.CardResolvedPayload{CardID: c.ID, ActionID: actionID, Answer: freeText}); err != nil {
return c, "", err
}
}
return c, label, 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, err := s.st.GetCard(owner, id)
if err != nil {
return err
}
if c.Status.Terminal() {
return nil // 已决(双现任一侧先到)
}
c.Status = core.CardResolved
if err := s.Update(owner, c); err != 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
}
+134
View File
@@ -0,0 +1,134 @@
package cardsvc
import (
"errors"
"sync"
"testing"
core "agent-system-core"
"agent-system-backend/internal/store"
)
func openTest(t *testing.T) (*store.Store, *Service, func() []core.StreamEvent) {
t.Helper()
st, err := store.OpenSQLite(t.TempDir() + "/test.db")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { st.Close() })
var mu sync.Mutex
var events []core.StreamEvent
emit := func(convID, runID string, typ core.StreamEventType, payload any) error {
mu.Lock()
events = append(events, core.StreamEvent{ConversationID: convID, Type: typ})
mu.Unlock()
return nil
}
return st, New(st, emit), func() []core.StreamEvent {
mu.Lock()
defer mu.Unlock()
return append([]core.StreamEvent(nil), events...)
}
}
func sampleCard() core.ActivityCard {
return core.ActivityCard{
ProjectID: "prj_1", ConversationID: "cnv_1", Kind: core.CardKindConfirm,
Title: "相关性确认", Status: core.CardActionRequired, Tone: "warn",
Actions: []core.UIAction{{ID: "keep", Label: "相关 · 入库"}, {ID: "drop", Label: "无关 · 仅归档"}},
}
}
func TestCardLifecycle(t *testing.T) {
st, svc, evsFn := openTest(t)
if _, err := st.CreateProject(store.OwnerDefault, "p"); err != nil {
t.Fatal(err)
}
card, err := svc.Create(store.OwnerDefault, sampleCard(), "")
if err != nil {
t.Fatal(err)
}
if card.ID == "" || card.CreatedAt.IsZero() {
t.Fatal("create must fill id/timestamps")
}
// 未裁决卡不可再点不存在动作
if _, _, err := svc.Action(store.OwnerDefault, card.ID, "nonexistent", ""); !errors.Is(err, ErrNoAction) {
t.Fatalf("unknown action: %v", err)
}
got, label, err := svc.Action(store.OwnerDefault, card.ID, "keep", "")
if err != nil || label != "相关 · 入库" || got.Status != core.CardResolved {
t.Fatalf("action: %v %q %s", err, label, got.Status)
}
// 已决卡幂等拒绝重复裁决
if _, _, err := svc.Action(store.OwnerDefault, card.ID, "keep", ""); !errors.Is(err, ErrNotAwaiting) {
t.Fatalf("re-action: %v", err)
}
evs := evsFn()
if len(evs) < 2 || evs[0].Type != core.EvCardCreated || evs[len(evs)-1].Type != core.EvCardResolved {
t.Fatalf("events: %+v", evs)
}
}
func TestPinDismissAndList(t *testing.T) {
st, svc, _ := openTest(t)
if _, err := st.CreateProject(store.OwnerDefault, "p"); err != nil {
t.Fatal(err)
}
a, _ := svc.Create(store.OwnerDefault, core.ActivityCard{ProjectID: "prj_1", ConversationID: "cnv_1", Kind: core.CardKindStatus, Title: "A"}, "")
b, _ := svc.Create(store.OwnerDefault, core.ActivityCard{ProjectID: "prj_1", ConversationID: "cnv_1", Kind: core.CardKindStatus, Title: "B"}, "")
if _, err := svc.SetPinned(store.OwnerDefault, a.ID, true); err != nil {
t.Fatal(err)
}
got, _ := svc.Get(store.OwnerDefault, a.ID)
if !got.Pinned {
t.Fatal("pin failed")
}
if _, err := svc.Dismiss(store.OwnerDefault, b.ID); err != nil {
t.Fatal(err)
}
gotB, _ := svc.Get(store.OwnerDefault, b.ID)
if !gotB.Status.Terminal() {
t.Fatal("dismiss should reach terminal state")
}
list, err := svc.List(store.OwnerDefault, "cnv_1")
if err != nil || len(list) != 2 {
t.Fatalf("list: %v %d", err, len(list))
}
// 租户隔离:他人不可见
if other, err := svc.List("someone-else", "cnv_1"); err != nil || len(other) != 0 {
t.Fatalf("owner isolation broken: %v %d", err, len(other))
}
}
func TestResolveByRefIdempotent(t *testing.T) {
st, svc, _ := openTest(t)
prj, err := st.CreateProject(store.OwnerDefault, "p")
if err != nil {
t.Fatal(err)
}
conv, err := st.CreateConversation(store.OwnerDefault, prj.ID, "c", "")
if err != nil {
t.Fatal(err)
}
card := sampleCard()
card.ProjectID, card.ConversationID = prj.ID, conv.ID
proj, err := svc.Create(store.OwnerDefault, card, "hitl:x1")
if err != nil {
t.Fatal(err)
}
if err := svc.ResolveByRef(store.OwnerDefault, "hitl:x1", "approve", "同意"); err != nil {
t.Fatal(err)
}
if err := svc.ResolveByRef(store.OwnerDefault, "hitl:x1", "approve", "again"); err != nil {
t.Fatalf("second resolve must be idempotent no-op: %v", err)
}
got, _ := svc.Get(store.OwnerDefault, proj.ID)
if got.Status != core.CardResolved {
t.Fatal("should be resolved")
}
// 未知 ref:静默
if err := svc.ResolveByRef(store.OwnerDefault, "hitl:none", "", ""); err != nil {
t.Fatalf("unknown ref: %v", err)
}
}
+73 -13
View File
@@ -10,13 +10,14 @@ import (
"sync"
"time"
core "agent-system-core"
"agent-system-backend/internal/bus"
"agent-system-backend/internal/cardsvc"
"agent-system-backend/internal/profiles"
"agent-system-backend/internal/runtime"
"agent-system-backend/internal/settings"
"agent-system-backend/internal/store"
"agent-system-backend/internal/tools"
core "agent-system-core"
)
type Manager struct {
@@ -29,10 +30,11 @@ type Manager struct {
subRunner runtime.Runner // subagent runner(LLM 轻任务)
mock runtime.Runner // 会话脚本回放(演示/测试与 mock 模式)
mockSub runtime.Runner // 子代理脚本回放
gate IntentGate // 可选意图闸门(nil = 未启用,行为回退)
events *eventBuffer // 事件批处理(写放大治理;seq 权威仍在 store)
retriever core.Retriever // 可选混合检索(nil = store 全文)
indexer core.DocumentIndexer // 可选索引入口(nil = store 整文档单块)
gate IntentGate
cards *cardsvc.Service // 可选意图闸门(nil = 未启用,行为回退)
events *eventBuffer // 事件批处理(写放大治理;seq 权威仍在 store)
retriever core.Retriever // 可选混合检索(nil = store 全文)
indexer core.DocumentIndexer // 可选索引入口(nil = store 整文档单块)
mu sync.Mutex
active map[string]*runHandle
@@ -83,6 +85,15 @@ type IntentGate interface {
// SetGate 注入意图闸门(装配期一次;插件意图 overlay 经 pluginhost 回调重建)。
func (m *Manager) SetGate(g IntentGate) { m.gate = g }
// SetCards 注入收件箱卡服务(nil = 不产卡;T0 退回基线仍成立)。
func (m *Manager) SetCards(s *cardsvc.Service) { m.cards = s }
// EmitCardEvent 供服务层(cardsvc 投影)复用会话事件管道(持久化 + seq + 扇出)。
func (m *Manager) EmitCardEvent(conversationID, runID string, typ core.StreamEventType, payload any) error {
_, err := m.emit(conversationID, runID, typ, payload)
return err
}
// SetRetriever 注入混合检索(rag 管线;nil = store 全文检索)。
func (m *Manager) SetRetriever(r core.Retriever) { m.retriever = r }
@@ -104,10 +115,10 @@ type runCtx struct {
subHint string
}
func (c *runCtx) Context() context.Context { return c.ctx }
func (c *runCtx) Run() core.Run { return c.run }
func (c *runCtx) Profile() core.AgentProfile { return c.prof }
func (c *runCtx) Settings() core.EffectiveSettings { return c.eff }
func (c *runCtx) Context() context.Context { return c.ctx }
func (c *runCtx) Run() core.Run { return c.run }
func (c *runCtx) Profile() core.AgentProfile { return c.prof }
func (c *runCtx) Settings() core.EffectiveSettings { return c.eff }
func (c *runCtx) History() []core.Message {
hist, _ := c.mgr.st.LastMessages(c.run.ConversationID, 200)
return hist
@@ -147,6 +158,26 @@ func (c *runCtx) AskHITL(card core.HITLCard) (core.HITLResponse, error) {
if _, err := c.mgr.emit(c.run.ConversationID, c.run.ID, core.EvHITLRequest, card); err != nil {
return core.HITLResponse{}, err
}
// 收件箱投影卡(双现):待确认进收件箱不沉没;ref 联动使任一侧响应即双侧销项。
if c.mgr.cards != nil {
proj := core.ActivityCard{
ProjectID: c.run.ProjectID, ConversationID: c.run.ConversationID, SourceRunID: c.run.ID,
Kind: core.CardKindConfirm, Title: card.Question, Status: core.CardActionRequired,
Tone: map[bool]string{true: "danger", false: "warn"}[card.Blocked],
}
if card.Blocked {
proj.Body = append(proj.Body, core.UINode{Type: core.UIMarkdown, Markdown: "⚠ 阻断:" + card.BlockReason})
}
for _, ch := range card.Choices {
proj.Actions = append(proj.Actions, core.UIAction{ID: ch.ID, Label: ch.Label, Style: map[bool]string{true: "primary", false: "default"}[ch.Recommended]})
}
if card.AllowFreeInput {
proj.Body = append(proj.Body, core.UINode{Type: core.UIMarkdown, Markdown: "_可在输入框自由输入回应。_"})
}
if _, err := c.mgr.cards.Create(c.run.Owner, proj, "hitl:"+card.ID); err != nil {
return core.HITLResponse{}, err
}
}
select {
case ans := <-c.h.hitlCh:
return ans.resp, ans.err
@@ -177,6 +208,26 @@ func (c *runCtx) UpdateTodos(l core.TodoList) error {
return err
}
// CreateCard / SetCardStatus:run 内产卡入口(摄取/编排/工作流步骤;mock 演示)。
// 收件箱关闭(cards==nil)时为 no-op——T0 退回基线不受影响。
func (c *runCtx) CreateCard(card core.ActivityCard) (core.ActivityCard, error) {
if c.mgr.cards == nil {
return card, nil
}
card.ProjectID = c.run.ProjectID
card.ConversationID = c.run.ConversationID
card.SourceRunID = c.run.ID
return c.mgr.cards.Create(c.run.Owner, card, "")
}
func (c *runCtx) SetCardStatus(id string, status core.CardStatus, tone, note string) error {
if c.mgr.cards == nil {
return nil
}
_, err := c.mgr.cards.SetStatus(c.run.Owner, id, status, tone, note)
return err
}
func (c *runCtx) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) {
art := core.Artifact{ProjectID: c.run.ProjectID, Owner: c.run.Owner, Name: name, Kind: kind, MIME: mime, SourceRun: c.run.ID}
if err := c.mgr.st.CreateArtifact(&art, content); err != nil {
@@ -259,7 +310,9 @@ type toolCtx struct{ rc *runCtx }
func (t toolCtx) ProjectID() string { return t.rc.run.ProjectID }
func (t toolCtx) ConversationID() string { return t.rc.run.ConversationID }
func (t toolCtx) Retrieve(q core.RetrievalQuery) ([]core.RetrievalHit, error) { return t.rc.Retrieve(q) }
func (t toolCtx) Retrieve(q core.RetrievalQuery) ([]core.RetrievalHit, error) {
return t.rc.Retrieve(q)
}
func (t toolCtx) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) {
return t.rc.WriteArtifact(name, kind, mime, content)
}
@@ -431,9 +484,9 @@ func (m *Manager) CancelRun(runID string) error {
// ---- HITL 解析(三分支路由,4.5)----
// ResolveHITL 处理用户对卡片的响应:
// 1) 选项命中 → 回填挂起 run;
// 2) 自由输入以 "/" 开头(命令)→ 取消挂起 run,返回转轨标志由 server 层发送新消息;
// 3) 其他自由输入 → 作为澄清回答回填。
// 1. 选项命中 → 回填挂起 run;
// 2. 自由输入以 "/" 开头(命令)→ 取消挂起 run,返回转轨标志由 server 层发送新消息;
// 3. 其他自由输入 → 作为澄清回答回填。
func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (pivotText string, pivoted bool, err error) {
card, err := m.st.GetHITLCard(owner, cardID)
if err != nil {
@@ -474,6 +527,9 @@ func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (piv
_ = m.CancelRun(run.ID)
m.st.ResolveHITLCard(owner, cardID, core.HITLResolvedPayload{CardID: core.ID(cardID), Kind: "cancelled", Answer: answer})
m.emit(run.ConversationID, run.ID, core.EvHITLResolved, core.HITLResolvedPayload{CardID: core.ID(cardID), Kind: "cancelled", Answer: answer})
if m.cards != nil {
_ = m.cards.ResolveByRef(owner, "hitl:"+cardID, "", answer)
}
_ = pivotReason
return trimmed, true, nil
}
@@ -483,6 +539,10 @@ func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (piv
return "", false, err
}
m.emit(run.ConversationID, run.ID, core.EvHITLResolved, resolution)
// 投影卡同步销项(双现;幂等)。
if m.cards != nil {
_ = m.cards.ResolveByRef(owner, "hitl:"+cardID, resp.ChoiceID, answer)
}
m.mu.Lock()
h, ok := m.active[run.ID]
+26 -1
View File
@@ -5,8 +5,8 @@ import (
"strings"
"time"
core "agent-system-core"
"agent-system-backend/internal/squadadapter"
core "agent-system-core"
)
func firstNonEmpty0(vals ...string) string {
@@ -100,6 +100,31 @@ func (m *MockRunner) Execute(rc RunCtx) error {
{ID: "s3", Text: "汇总输出", Status: "done"},
}})
case strings.Contains(user, "#upload"):
// 收件箱卡片流演示(提案 T2):摄取卡生命周期 + 冲突对比卡(待处理)。
addText("📎 已上传 临床前动物试验方案-草案.pdf(演示)——上传即构建记忆:先给相关性结论,解析进度在右栏卡片。")
card, _ := rc.CreateCard(core.ActivityCard{Kind: core.CardKindProgress, Title: "临床前动物试验方案-草案.pdf", Status: core.CardActive, Tone: "info",
Body: []core.UINode{{Type: core.UIProgress, Progress: &core.UIProgressSpec{Label: "解析中", Value: 0.25}}}})
rc.Emit(core.EvThinkingDelta, core.ThinkingDeltaPayload{Index: 0, Text: "记忆构建管道:相关性 ✓(命中:试验样本量、随访周期)→ 拆解 → 打标签 → 切片入库……"})
parts = append(parts, core.Part{Type: core.PartThinking, Text: "记忆构建管道:相关性 ✓ → 拆解 → 打标签 → 切片入库", Process: true})
card.Body = []core.UINode{
{Type: core.UIProgress, Progress: &core.UIProgressSpec{Label: "切片 24/24", Value: 1, Tone: "ok"}},
{Type: core.UIKV, KV: [][2]string{{"快通道", "✓ 38s"}, {"慢通道", "✓ 3m12s"}, {"追溯候选边", "6 条"}}},
{Type: core.UIList, Title: "标签", Items: []string{"验证确认(跨阶段引用)", "technical ×1", "regulatory ×1"}},
}
card.Status = core.CardDone
card.Tone = "ok"
_ = rc.SetCardStatus(card.ID, core.CardDone, "ok", "")
_ = rc.Emit(core.EvCardUpdated, core.CardPayload{Card: card})
cmp, _ := rc.CreateCard(core.ActivityCard{Kind: core.CardKindCompare, Title: "项目执行周期不一致 · 目标已基线", Status: core.CardActionRequired, Tone: "danger",
Body: []core.UINode{{Type: core.UICompare, Compare: &core.UICompareSpec{
Left: core.UICompareSide{Title: "基线 V1.2 · 已评审", Tone: "danger", Nodes: []core.UINode{{Type: core.UIMarkdown, Markdown: "12 个月(立项到注册)"}}},
Right: core.UICompareSide{Title: "新资料 · 待确认", Tone: "ok", Nodes: []core.UINode{{Type: core.UIMarkdown, Markdown: "18 个月(含临床试验)"}}},
}}},
Actions: []core.UIAction{{ID: "escalate", Label: "发起变更流程", Style: "primary"}, {ID: "keep", Label: "维持原样", Style: "danger"}},
})
parts = append(parts, core.Part{Type: core.PartCard, Card: &core.CardPart{Kind: "cardlink", Data: []core.UINode{{Type: core.UIMarkdown, Markdown: fmt.Sprintf("↗ 解析完成见卡片;冲突处理卡 `card:%s` 已进右栏收件箱(待处理)", cmp.ID)}}}})
case strings.Contains(user, "#doc"):
rc.Emit(core.EvThinkingDelta, core.ThinkingDeltaPayload{Index: 0, Text: "准备生成文档……"})
parts = append(parts, core.Part{Type: core.PartThinking, Text: "准备生成文档……", Process: true})
+3
View File
@@ -30,6 +30,9 @@ type RunCtx interface {
Tools() []core.ToolDesc
// Subtask 返回子代理任务描述(仅 subagent run 有值)。
Subtask() string
// CreateCard / SetCardStatus:产收件箱任务卡(收件箱关闭时为 no-op)。
CreateCard(card core.ActivityCard) (core.ActivityCard, error)
SetCardStatus(id string, status core.CardStatus, tone, note string) error
// SetContextObservation 记录本次运行的上下文组装观测(随 run.completed 下发)。
SetContextObservation(o core.ContextObservation)
}
+107 -23
View File
@@ -15,7 +15,7 @@ import (
"strings"
"time"
core "agent-system-core"
"agent-system-backend/internal/cardsvc"
"agent-system-backend/internal/ingest"
"agent-system-backend/internal/plugins"
"agent-system-backend/internal/profiles"
@@ -25,21 +25,23 @@ import (
"agent-system-backend/internal/store"
"agent-system-backend/internal/tools"
"agent-system-backend/web"
core "agent-system-core"
)
type Server struct {
st *store.Store
Auth *Authenticator // API key 认证(nil/未启用 = 默认租户;组合根注入)
Retriever core.Retriever // 混合检索(nil = store 全文;组合根注入)
Indexer core.DocumentIndexer // 检索索引入口(nil = 整文档单块;组合根注入)
Limiter ratelimit.Limiter // 租户限流(nil = 不限;组合根注入)
LimitLookup func() int // 当前限流值(次/分;<=0 不限;组合根缓存注入)
mgr *runmgr.Manager
reg *tools.Registry
prof *profiles.Service
set *settings.Service
plugins *plugins.Host
static fs.FS
st *store.Store
Auth *Authenticator // API key 认证(nil/未启用 = 默认租户;组合根注入)
Retriever core.Retriever // 混合检索(nil = store 全文;组合根注入)
Indexer core.DocumentIndexer // 检索索引入口(nil = 整文档单块;组合根注入)
Limiter ratelimit.Limiter // 租户限流(nil = 不限;组合根注入)
LimitLookup func() int // 当前限流值(次/分;<=0 不限;组合根缓存注入)
mgr *runmgr.Manager
Cards *cardsvc.Service // 收件箱任务卡(nil = 不启用;组合根注入)
reg *tools.Registry
prof *profiles.Service
set *settings.Service
plugins *plugins.Host
static fs.FS
// Subscribe 注入 bus 订阅(事件在线转发;重放走 store 游标)。
Subscribe func(conversationID string) (<-chan core.StreamEvent, func())
}
@@ -72,6 +74,12 @@ func (s *Server) Handler() http.Handler {
// ---- HITL ----
mux.HandleFunc("POST /api/hitl/{id}/resolve", s.resolveHITL)
// ---- 收件箱任务卡(提案 T2)----
mux.HandleFunc("GET /api/conversations/{id}/cards", s.listCards)
mux.HandleFunc("POST /api/cards/{id}/actions/{actionId}", s.cardAction)
mux.HandleFunc("POST /api/cards/{id}/pin", s.cardPin)
mux.HandleFunc("POST /api/cards/{id}/dismiss", s.cardDismiss)
// ---- 事件流(SSE)----
mux.HandleFunc("GET /api/events", s.events)
@@ -205,7 +213,6 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) {
}
}
func (s *Server) serveStatic(w http.ResponseWriter, r *http.Request) {
path := strings.TrimPrefix(r.URL.Path, "/")
if path == "" {
@@ -242,7 +249,9 @@ func (s *Server) listProjects(w http.ResponseWriter, r *http.Request) {
}
func (s *Server) createProject(w http.ResponseWriter, r *http.Request) {
var req struct{ Name string `json:"name"` }
var req struct {
Name string `json:"name"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil || strings.TrimSpace(req.Name) == "" {
http.Error(w, "name required", http.StatusBadRequest)
return
@@ -254,8 +263,8 @@ func (s *Server) createProject(w http.ResponseWriter, r *http.Request) {
func (s *Server) patchProject(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
var req struct {
Name *string `json:"name"`
Settings map[string]any `json:"settings"`
Name *string `json:"name"`
Settings map[string]any `json:"settings"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
@@ -457,6 +466,53 @@ func (s *Server) resolveHITL(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]any{"resolved": true, "pivoted": pivoted, "pivotText": pivotText})
}
// ---- 收件箱任务卡 ----
func (s *Server) listCards(w http.ResponseWriter, r *http.Request) {
owner := s.owner(r)
if _, err := s.st.GetConversation(owner, r.PathValue("id")); err != nil {
writeStoreErr(w, err)
return
}
cards, err := s.Cards.List(owner, r.PathValue("id"))
writeJSONOr(w, map[string]any{"cards": cards}, err)
}
// cardAction 处理卡上按钮:resolved + 回执消息进对话时间线(确认即留痕)。
func (s *Server) cardAction(w http.ResponseWriter, r *http.Request) {
owner := s.owner(r)
var body struct {
FreeText string `json:"freeText,omitempty"`
}
_ = json.NewDecoder(r.Body).Decode(&body)
card, label, err := s.Cards.Action(owner, r.PathValue("id"), r.PathValue("actionId"), body.FreeText)
if err != nil {
writeStoreErr(w, err)
return
}
// 回执消息:确认结论写进对话时间线(含所点按钮语义),审计锚点为 card.resolved 事件。
if card.ConversationID != "" {
msg := core.Message{ID: store.NewID("msg"), ConversationID: card.ConversationID, Role: core.RoleAssistant, CreatedAt: time.Now().UTC()}
msg.Parts = []core.Part{{Type: core.PartText, Text: fmt.Sprintf("✓ 已按「%s」处理「%s」——结论已记录并留痕(card:%s)", label, card.Title, card.ID)}}
_ = s.st.AppendMessage(&msg)
}
writeJSON(w, http.StatusOK, map[string]any{"card": card})
}
func (s *Server) cardPin(w http.ResponseWriter, r *http.Request) {
var body struct {
Pinned bool `json:"pinned"`
}
_ = json.NewDecoder(r.Body).Decode(&body)
card, err := s.Cards.SetPinned(s.owner(r), r.PathValue("id"), body.Pinned)
writeJSONOr(w, map[string]any{"card": card}, err)
}
func (s *Server) cardDismiss(w http.ResponseWriter, r *http.Request) {
card, err := s.Cards.Dismiss(s.owner(r), r.PathValue("id"))
writeJSONOr(w, map[string]any{"card": card}, err)
}
// ---- 上传与工件 ----
func (s *Server) upload(w http.ResponseWriter, r *http.Request) {
@@ -484,7 +540,35 @@ func (s *Server) upload(w http.ResponseWriter, r *http.Request) {
writeStoreErr(w, err)
return
}
// 摄取卡(SR-A5 界面化):解析过程显式可见,失败不静默——原始文件已存工件可取回。
// v1 边界:上传为项目级事件,卡片按项目持久化(conversationId 空);会话内演示见 mock #upload。
ingestCard := core.ActivityCard{
ProjectID: projectID, Kind: core.CardKindProgress, Title: header.Filename,
Status: core.CardActive, Tone: "info",
Body: []core.UINode{{Type: core.UIProgress, Progress: &core.UIProgressSpec{Label: "解析中", Value: 0.3}}},
}
if s.Cards != nil {
ingestCard, _ = s.Cards.Create(owner, ingestCard, "")
}
doc := ingest.Parse(art.Name, mime, content)
if s.Cards != nil {
ingestCard.Status = core.CardDone
ingestCard.Tone = "ok"
body := []core.UINode{
{Type: core.UIKV, KV: [][2]string{{"解析器", doc.Meta.Parser}, {"内容块", fmt.Sprint(len(doc.Blocks))}}},
{Type: core.UIActions, Actions: []core.UIAction{{ID: "open", Label: "📄 打开工件", Command: ""}}},
}
if len(doc.Warnings) > 0 {
ingestCard.Status = core.CardActionRequired
ingestCard.Tone = "warn"
for _, wn := range doc.Warnings {
body = append(body, core.UINode{Type: core.UIMarkdown, Markdown: "⚠ " + wn})
}
body = append(body, core.UINode{Type: core.UIMarkdown, Markdown: "_解析不完整不静默:原始文件已永久保留,可取回或重试。_"})
}
ingestCard.Body = body
_ = s.Cards.Update(owner, ingestCard)
}
// 文本可解析的进检索索引(管线分块+全文+可选向量)。
if plain := doc.PlainText(); plain != "" && len(plain) < 200000 {
if s.Indexer != nil {
@@ -659,13 +743,13 @@ func (s *Server) deleteUserTool(w http.ResponseWriter, r *http.Request) {
// ---- 插件 ----
type pluginView struct {
Manifest core.PluginManifest `json:"manifest"`
Panels []core.PluginPanel `json:"panels"`
Manifest core.PluginManifest `json:"manifest"`
Panels []core.PluginPanel `json:"panels"`
Cards map[string][]core.UINode `json:"cards"`
Workflows []core.WorkflowDef `json:"workflows"`
Intents []json.RawMessage `json:"intents"`
Prompts map[string]string `json:"prompts"`
Settings []core.SettingDesc `json:"settings"`
Workflows []core.WorkflowDef `json:"workflows"`
Intents []json.RawMessage `json:"intents"`
Prompts map[string]string `json:"prompts"`
Settings []core.SettingDesc `json:"settings"`
}
func (s *Server) listPlugins(w http.ResponseWriter, r *http.Request) {
+28 -1
View File
@@ -5,7 +5,7 @@
// - RoleRegistry → 内置小队角色(规则规划器自主编排四阶段骨架)
// - Planner → squad RulePlanner(确定性;LLM 规划器为后续档)
// - BackendResolver→ 自定义 "agent-system" 后端工厂:每步执行 = rc.SpawnSubagent
// (子 Run 进任务树,走既有 mock/LLM 子代理执行器与事件)
// (子 Run 进任务树,走既有 mock/LLM 子代理执行器与事件)
// - ProgressSink → 步骤事件映射 plan.update / agent.progress
// - HITL(挂起) → 引擎 Pending HumanRequest ↔ rc.AskHITL 卡片;RespondDecision 回填
package squadadapter
@@ -36,12 +36,29 @@ type Ctx interface {
WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error)
}
// CardSink 是可选的产卡能力(提案 T2 编排卡):runCtx 实现之;测试桩无需实现。
type CardSink interface {
CreateCard(card core.ActivityCard) (core.ActivityCard, error)
SetCardStatus(id string, status core.CardStatus, tone, note string) error
}
// Orchestrate 在一次 conversation run 内驱动 squad 引擎完成目标。
// 返回摘要文本(由 runner 落为消息部件);error = 编排失败。
func Orchestrate(rc Ctx, dataDir, goalText string) (string, error) {
if strings.TrimSpace(goalText) == "" {
return "", fmt.Errorf("squad: goal text is empty")
}
// 编排卡(可选能力):开始 → 执行中 → 结果/失败显式终态;无 CardSink 时静默跳过。
var cardID string
if cs, ok := rc.(CardSink); ok {
title := goalText
if len(title) > 48 {
title = title[:48] + "…"
}
if card, err := cs.CreateCard(core.ActivityCard{Kind: core.CardKindProgress, Title: "小队编排 · " + title, Status: core.CardActive, Tone: "info"}); err == nil {
cardID = card.ID
}
}
runDir := filepath.Join(dataDir, "squad", rc.Run().ID)
registry := squad.NewRoleRegistry()
@@ -114,8 +131,18 @@ func Orchestrate(rc Ctx, dataDir, goalText string) (string, error) {
}
switch res.Status {
case squad.RunCompleted:
if cardID != "" {
if cs, ok := rc.(CardSink); ok {
_ = cs.SetCardStatus(cardID, core.CardDone, "ok", summary)
}
}
return summary, nil
default:
if cardID != "" {
if cs, ok := rc.(CardSink); ok {
_ = cs.SetCardStatus(cardID, core.CardFailed, "danger", res.Error)
}
}
return summary, fmt.Errorf("squad: run %s status %s: %s", res.RunID, res.Status, res.Error)
}
}
+81
View File
@@ -0,0 +1,81 @@
package store
import (
"database/sql"
"encoding/json"
"errors"
"fmt"
"time"
core "agent-system-core"
)
// 收件箱任务卡持久化:payload 存 core.ActivityCard 整卡 JSON(卡片是可变状态对象,
// 服务端权威);status/pinned/ref 冗余列支撑过滤、排序与 HITL 投影联动。
// SaveCard 插入或更新整卡(upsert)。
func (s *Store) SaveCard(owner string, c *core.ActivityCard, ref string) error {
payload, err := json.Marshal(c)
if err != nil {
return fmt.Errorf("save card %s: %w", c.ID, err)
}
_, err = s.db.Exec(s.q(`INSERT INTO activity_cards (id, project_id, conversation_id, ref, owner, payload, status, pinned, created_at, updated_at)
VALUES (?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(id) DO UPDATE SET payload=excluded.payload, status=excluded.status, pinned=excluded.pinned, ref=CASE WHEN excluded.ref='' THEN activity_cards.ref ELSE excluded.ref END, updated_at=excluded.updated_at`),
c.ID, c.ProjectID, c.ConversationID, ref, owner, string(payload), string(c.Status), boolInt(c.Pinned), c.CreatedAt.Format(time.RFC3339Nano), c.UpdatedAt.Format(time.RFC3339Nano))
if err != nil {
return fmt.Errorf("save card %s: %w", c.ID, err)
}
return nil
}
func (s *Store) GetCard(owner, id string) (core.ActivityCard, error) {
row := s.db.QueryRow(s.q(`SELECT payload FROM activity_cards WHERE id=? AND owner=?`), id, owner)
var payload string
if err := row.Scan(&payload); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return core.ActivityCard{}, ErrNotFound
}
return core.ActivityCard{}, err
}
var c core.ActivityCard
if err := json.Unmarshal([]byte(payload), &c); err != nil {
return core.ActivityCard{}, fmt.Errorf("card %s payload: %w", id, err)
}
return c, nil
}
func (s *Store) ListCards(owner, conversationID string) ([]core.ActivityCard, error) {
rows, err := s.db.Query(s.q(`SELECT payload FROM activity_cards WHERE conversation_id=? AND owner=? ORDER BY updated_at DESC`), conversationID, owner)
if err != nil {
return nil, fmt.Errorf("list cards: %w", err)
}
defer rows.Close()
var out []core.ActivityCard
for rows.Next() {
var payload string
if err := rows.Scan(&payload); err != nil {
return nil, err
}
var c core.ActivityCard
if err := json.Unmarshal([]byte(payload), &c); err != nil {
return nil, fmt.Errorf("card payload: %w", err)
}
out = append(out, c)
}
return out, rows.Err()
}
// CardIDByRef 按引用取卡 ID(HITL 投影联动:"hitl:<id>";最新一条)。
func (s *Store) CardIDByRef(owner, ref string) (string, error) {
row := s.db.QueryRow(s.q(`SELECT id FROM activity_cards WHERE ref=? AND owner=? ORDER BY updated_at DESC LIMIT 1`), ref, owner)
var id string
err := row.Scan(&id)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return "", ErrNotFound
}
return "", err
}
return id, nil
}
+13
View File
@@ -202,6 +202,19 @@ func init() {
migration{version: 5, name: "knowledge_vectors",
up: migV5Up,
down: map[string][]string{"sqlite": migV5Down, "postgres": migV5Down}},
migration{version: 6, name: "activity_cards",
up: map[string][]string{"sqlite": {
// 收件箱任务卡(core.ActivityCard):payload 存整卡 JSON;
// status/pinned/ref 冗余列供过滤、排序与 HITL 投影联动(提案 T2)。
`CREATE TABLE IF NOT EXISTS activity_cards (id TEXT PRIMARY KEY, project_id TEXT NOT NULL, conversation_id TEXT NOT NULL DEFAULT '', ref TEXT NOT NULL DEFAULT '', owner TEXT NOT NULL DEFAULT '', payload TEXT NOT NULL, status TEXT NOT NULL, pinned INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL)`,
`CREATE INDEX IF NOT EXISTS idx_cards_conv ON activity_cards(conversation_id, updated_at)`,
`CREATE INDEX IF NOT EXISTS idx_cards_ref ON activity_cards(ref)`,
}, "postgres": {
`CREATE TABLE IF NOT EXISTS activity_cards (id TEXT PRIMARY KEY, project_id TEXT NOT NULL, conversation_id TEXT NOT NULL DEFAULT '', ref TEXT NOT NULL DEFAULT '', owner TEXT NOT NULL DEFAULT '', payload TEXT NOT NULL, status TEXT NOT NULL, pinned INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL)`,
`CREATE INDEX IF NOT EXISTS idx_cards_conv ON activity_cards(conversation_id, updated_at)`,
`CREATE INDEX IF NOT EXISTS idx_cards_ref ON activity_cards(ref)`,
}},
down: map[string][]string{"sqlite": {`DROP TABLE IF EXISTS activity_cards`}, "postgres": {`DROP TABLE IF EXISTS activity_cards`}}},
)
}
+2 -2
View File
@@ -56,8 +56,8 @@ func TestMigrationUpgradesExistingDatabase(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if len(got) != 5 {
t.Fatalf("expected versions [1..5], got %v", got)
if len(got) != 6 {
t.Fatalf("expected versions [1..6], got %v", got)
}
// 老库形态(v2 前):events 全局自增、settings 单列主键。
if err := st.MigrateDown(1); err != nil {