diff --git a/deps/agent-system-core b/deps/agent-system-core index f8b77c0..1b04a34 160000 --- a/deps/agent-system-core +++ b/deps/agent-system-core @@ -1 +1 @@ -Subproject commit f8b77c0d0263a708c4804f2b3cbeb097aa2a61d1 +Subproject commit 1b04a34bbc43143d3a90cea50e277964ba03e029 diff --git a/internal/cardsvc/cardsvc.go b/internal/cardsvc/cardsvc.go index 62c3daf..168061a 100644 --- a/internal/cardsvc/cardsvc.go +++ b/internal/cardsvc/cardsvc.go @@ -5,6 +5,7 @@ package cardsvc import ( "errors" + "strings" "sync" "time" @@ -37,8 +38,30 @@ func New(st *store.Store, emit Emitter) *Service { return &Service{st: st, emit: emit, now: time.Now, lastEmit: map[string]time.Time{}} } +// emitCard 按卡归属发布卡片事件（D50）： +// - 会话卡 → 会话流；同时镜像项目流（伪会话键 "project:"）—— +// 项目收件箱聚合视图（项目级卡 ∪ 各会话未决卡）的实时性依赖它；卡片事件 +// 量小，全量镜像换取口径简单。 +// - 项目级卡（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 + } + } + if c.ProjectID != "" { + return s.emit(string(core.ProjectChannelKey(c.ProjectID)), c.SourceRunID, typ, payload) + } + return nil +} + // Create 补全标识与时间戳、持久化并发布 card.created。ref 为投影引用 -// （HITL 投影卡 = "hitl:"）；ConversationID 为空的项目级卡只持久化。 +// （HITL 投影卡 = "hitl:"，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") @@ -52,12 +75,7 @@ func (s *Service) Create(owner string, card core.ActivityCard, ref string) (core 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 + return card, s.emitCard(card, core.EvCardCreated, core.CardPayload{Card: card}) } // Get 读取单卡。 @@ -74,10 +92,7 @@ func (s *Service) Update(owner string, card core.ActivityCard) error { 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 + return s.emitCard(card, core.EvCardUpdated, core.CardPayload{Card: card}) } // Append 流式追加正文（card_append 工具；D45）：文本并入 body 尾部 markdown @@ -100,8 +115,8 @@ func (s *Service) Append(owner, id, text string) error { if err := s.st.SaveCard(owner, &c, ""); err != nil { return err } - if c.ConversationID == "" || s.emit == nil { - return nil + if c.ConversationID == "" && c.ProjectID == "" { + return nil // 无归属：只落库不发布 } s.emitMu.Lock() last, seen := s.lastEmit[id] @@ -113,7 +128,7 @@ func (s *Service) Append(owner, id, text string) error { if !emit { return nil } - return s.emit(c.ConversationID, c.SourceRunID, core.EvCardUpdated, core.CardPayload{Card: c}) + return s.emitCard(c, core.EvCardUpdated, core.CardPayload{Card: c}) } // SetStatus 状态迁移：更新并发布 card.status。 @@ -129,12 +144,7 @@ func (s *Service) SetStatus(owner, id string, status core.CardStatus, tone, note 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 + return c, s.emitCard(c, core.EvCardStatus, core.CardStatusPayload{CardID: c.ID, Status: status, Tone: c.Tone, Note: note}) } // SetPinned 置顶/取消（收件箱"不沉没"语义）。 @@ -182,12 +192,45 @@ func (s *Service) Action(owner, id, actionID, freeText string) (core.ActivityCar 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, 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") } } - return c, label, nil + 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:"）；幂等。 diff --git a/internal/cardsvc/cardsvc_test.go b/internal/cardsvc/cardsvc_test.go index 1cd8c9c..dd400c0 100644 --- a/internal/cardsvc/cardsvc_test.go +++ b/internal/cardsvc/cardsvc_test.go @@ -72,15 +72,22 @@ func TestAppendStreamThrottleAndMerge(t *testing.T) { if err := svc.Append(store.OwnerDefault, card.ID, "## 风险\n"); err != nil { t.Fatal(err) } - // 前三次都在 250ms 窗口内：只有第一次发布 card.updated。 - updated := 0 - for _, e := range evsFn() { - if e.Type == core.EvCardUpdated { - updated++ + // 前三次都在 250ms 窗口内：只有第一次发布 card.updated（会话流 + 项目流镜像 + // 各一次，D50）。 + countUpdated := func(channel string) int { + n := 0 + for _, e := range evsFn() { + if e.Type == core.EvCardUpdated && string(e.ConversationID) == channel { + n++ + } } + return n } - if updated != 1 { - t.Fatalf("窗口内 card.updated 应节流为 1 次，got %d", updated) + if updated := countUpdated("cnv_1"); updated != 1 { + t.Fatalf("窗口内会话流 card.updated 应节流为 1 次，got %d", updated) + } + if mirrored := countUpdated("project:prj_1"); mirrored != 1 { + t.Fatalf("窗口内项目流镜像应为 1 次，got %d", mirrored) } now = now.Add(300 * time.Millisecond) @@ -90,14 +97,11 @@ func TestAppendStreamThrottleAndMerge(t *testing.T) { if _, err := svc.SetStatus(store.OwnerDefault, card.ID, core.CardDone, "ok", ""); err != nil { t.Fatal(err) } - updated = 0 - for _, e := range evsFn() { - if e.Type == core.EvCardUpdated { - updated++ - } + if updated := countUpdated("cnv_1"); updated != 3 { // 首次 + 跨窗口一次 + SetStatus 的 Update 恒发布 + t.Fatalf("终态后会话流 card.updated 应为 3 次，got %d", updated) } - if updated != 3 { // 首次 + 跨窗口一次 + SetStatus 的 Update 恒发布 - t.Fatalf("终态后 card.updated 应为 3 次，got %d", updated) + if mirrored := countUpdated("project:prj_1"); mirrored != 3 { + t.Fatalf("终态后项目流镜像应为 3 次，got %d", mirrored) } got, err := svc.Get(store.OwnerDefault, card.ID) @@ -211,3 +215,96 @@ func TestResolveByRefIdempotent(t *testing.T) { t.Fatalf("unknown ref: %v", err) } } + +// TestToProjectArchiveAndIdempotent 验证 D48/D49：会话卡留档为项目级卡（新 ID、 +// 锚点指原卡、body 附留档来源、原卡 dismissed）；重复调用幂等；HITL 投影卡拒绝。 +func TestToProjectArchiveAndIdempotent(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 + orig, err := svc.Create(store.OwnerDefault, card, "") + if err != nil { + t.Fatal(err) + } + + copy1, err := svc.ToProject(store.OwnerDefault, orig.ID) + if err != nil { + t.Fatal(err) + } + if copy1.ID == orig.ID || copy1.ConversationID != "" || copy1.ProjectID != prj.ID { + t.Fatalf("副本归属不符: %+v", copy1) + } + if copy1.AnchorMessageID != orig.ID { + t.Errorf("副本锚点应指向原卡: %q", copy1.AnchorMessageID) + } + hasNote := false + for _, n := range copy1.Body { + if n.Type == core.UIKV { + for _, kv := range n.KV { + if kv[0] == "留档" { + hasNote = true + } + } + } + } + if !hasNote { + t.Error("副本 body 缺留档来源 kv") + } + origNow, _ := svc.Get(store.OwnerDefault, orig.ID) + if origNow.Status != core.CardDismissed { + t.Errorf("原卡应 dismissed 留痕, got %s", origNow.Status) + } + + // 幂等：重复留档返回既有副本，不再新造。 + copy2, err := svc.ToProject(store.OwnerDefault, orig.ID) + if err != nil || copy2.ID != copy1.ID { + t.Fatalf("重复留档应幂等: %v %+v", err, copy2) + } + + // 聚合口径（D49）：项目视图 = 项目级卡 ∪ 各会话未决卡；done/dismissed 不聚合。 + // （原卡已 dismissed → 不聚合；copy 项目级 → 聚合。） + other := sampleCard() + other.ProjectID, other.ConversationID = prj.ID, conv.ID + other.Title = "他会话未决" + if _, err := svc.Create(store.OwnerDefault, other, ""); err != nil { + t.Fatal(err) + } + doneCard := sampleCard() + doneCard.ProjectID, doneCard.ConversationID = prj.ID, conv.ID + doneCard.Status = core.CardDone + doneCard.Title = "他会话已完成" + if _, err := svc.Create(store.OwnerDefault, doneCard, ""); err != nil { + t.Fatal(err) + } + cards, lerr := st.ListProjectCards(store.OwnerDefault, prj.ID) + if lerr != nil { + t.Fatal(lerr) + } + if len(cards) != 2 { + t.Fatalf("project list 应为 副本+他会话未决 共 2 张，got %d", len(cards)) + } + for _, c := range cards { + if c.ConversationID != "" && c.Status != core.CardActionRequired { + t.Errorf("会话卡仅未决可聚合: %+v", c) + } + } + + // HITL 投影卡拒绝留档（流程状态机绑定）。 + hc := sampleCard() + hc.ProjectID, hc.ConversationID = prj.ID, conv.ID + h, err := svc.Create(store.OwnerDefault, hc, "hitl:x9") + if err != nil { + t.Fatal(err) + } + if _, err := svc.ToProject(store.OwnerDefault, h.ID); err == nil { + t.Error("HITL 投影卡应拒绝留档") + } +} diff --git a/internal/runmgr/events.go b/internal/runmgr/events.go index 0ba9fb8..e4bc2af 100644 --- a/internal/runmgr/events.go +++ b/internal/runmgr/events.go @@ -85,6 +85,11 @@ func (eb *eventBuffer) emit(conversationID, runID string, typ core.StreamEventTy ConversationID: conversationID, RunID: runID, TS: time.Now().UTC().Format(time.RFC3339Nano), V: core.ProtocolVersion, Type: typ, } + // 项目流（D50）：伪会话键复用同一持久化/seq/重放机制； + // 信封在此盖章项目归属，前端按 ProjectID 识别项目流事件。 + if pid, ok := core.ProjectFromChannel(core.ID(conversationID)); ok { + e.ProjectID = pid + } if payload != nil { buf, err := json.Marshal(payload) if err != nil { diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index 9879b88..5580902 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -234,12 +234,17 @@ func (c *runCtx) UpdateTodos(l core.TodoList) error { // CreateCard / SetCardStatus：run 内产卡入口（摄取/编排/工作流步骤；mock 演示）。 // 收件箱关闭（cards==nil）时为 no-op——T0 退回基线不受影响。 +// 归属判定（D48）：调用方留空 ConversationID 且预置 ProjectID = 项目级卡 +// （scope=project 工具参数），补全 ProjectID、ConversationID 保持空；其余照旧 +// 归属当前会话。 func (c *runCtx) CreateCard(card core.ActivityCard) (core.ActivityCard, error) { if c.mgr.cards == nil { return card, nil } + if card.ConversationID == "" && card.ProjectID == "" { + card.ConversationID = c.run.ConversationID + } card.ProjectID = c.run.ProjectID - card.ConversationID = c.run.ConversationID card.SourceRunID = c.run.ID return c.mgr.cards.Create(c.run.Owner, card, "") } diff --git a/internal/runtime/llm.go b/internal/runtime/llm.go index 3ca15f5..6c52c9b 100644 --- a/internal/runtime/llm.go +++ b/internal/runtime/llm.go @@ -199,12 +199,12 @@ func (l *LLMRunner) Execute(rc RunCtx) error { decls = append(decls, agentllm.ToolDecl{ Name: toolCard, Description: l.toolDesc(prof.Owner, toolCard, "提交产物型长内容（报告/纪要/方案/对比/文档草稿）为右侧收件箱卡片：卡片承载全文，对话中只保留结论摘要。解释型回答不要用此工具"), - Parameters: json.RawMessage(`{"type":"object","properties":{"title":{"type":"string","description":"产物标题"},"summary":{"type":"string","description":"一两句结论摘要（显示在卡片与对话锚点上）"},"body":{"type":"string","description":"完整 Markdown 正文"}},"required":["title","summary","body"]}`), + Parameters: json.RawMessage(`{"type":"object","properties":{"title":{"type":"string","description":"产物标题"},"summary":{"type":"string","description":"一两句结论摘要（显示在卡片与对话锚点上）"},"body":{"type":"string","description":"完整 Markdown 正文"},"scope":{"type":"string","enum":["conversation","project"],"description":"卡片归属：conversation=当前对话（默认）、project=项目收件箱（跨会话可见）"}},"required":["title","summary","body"]}`), }) decls = append(decls, agentllm.ToolDecl{ Name: toolCardBegin, Description: l.toolDesc(prof.Owner, toolCardBegin, "开始流式生成右侧收件箱产物卡（长正文推荐）：先给出标题与可选摘要，随后用 card_append 逐段追加正文，写完用 card_commit 收尾"), - Parameters: json.RawMessage(`{"type":"object","properties":{"title":{"type":"string","description":"产物标题"},"summary":{"type":"string","description":"一两句结论摘要（可选）"}},"required":["title"]}`), + Parameters: json.RawMessage(`{"type":"object","properties":{"title":{"type":"string","description":"产物标题"},"summary":{"type":"string","description":"一两句结论摘要（可选）"},"scope":{"type":"string","enum":["conversation","project"],"description":"卡片归属：conversation=当前对话（默认）、project=项目收件箱（跨会话可见）"}},"required":["title"]}`), }) decls = append(decls, agentllm.ToolDecl{ Name: toolCardAppend, @@ -363,7 +363,7 @@ func (l *LLMRunner) Execute(rc RunCtx) error { out = "submit_card 未启用（设置 ui.cardflow 关闭）；内容直接写在对话回复里" break } - var a struct{ Title, Summary, Body string } + var a struct{ Title, Summary, Body, Scope string } if err := json.Unmarshal([]byte(call.Arguments), &a); err != nil || strings.TrimSpace(a.Title) == "" || strings.TrimSpace(a.Body) == "" { out = "submit_card 参数错误（title/body 必填）: " + err.Error() break @@ -372,22 +372,30 @@ func (l *LLMRunner) Execute(rc RunCtx) error { if strings.TrimSpace(a.Summary) != "" { body = append(body, core.UINode{Type: "kv", KV: [][2]string{{"摘要", a.Summary}}}) } - card, cerr := rc.SubmitCard(core.ActivityCard{ + newCard := core.ActivityCard{ Kind: core.CardKindResult, Title: a.Title, Status: core.CardDone, Tone: "ok", Body: body, - }) + } + if a.Scope == "project" { + newCard.ProjectID = core.ID(rc.Run().ProjectID) // 项目级卡（D48）：收进项目收件箱 + } + card, cerr := rc.SubmitCard(newCard) if cerr != nil { out = "submit_card 失败: " + cerr.Error() break } finalParts = append(finalParts, core.Part{Type: core.PartCard, Card: &core.CardPart{Kind: string(core.CardKindResult), CardID: card.ID}}) explicitCard = true - out = fmt.Sprintf("已作为卡片发出（%s，card:%s）。对话中保留简短结论即可，不要重复卡片全文。", a.Title, card.ID) + where := "对话" + if a.Scope == "project" { + where = "项目" + } + out = fmt.Sprintf("已作为卡片发出到%s收件箱（%s，card:%s）。对话中保留简短结论即可，不要重复卡片全文。", where, a.Title, card.ID) case toolCardBegin: if !cardflow { out = "card_begin 未启用（设置 ui.cardflow 关闭）；内容直接写在对话回复里" break } - var a struct{ Title, Summary string } + var a struct{ Title, Summary, Scope string } if err := json.Unmarshal([]byte(call.Arguments), &a); err != nil || strings.TrimSpace(a.Title) == "" { out = "card_begin 参数错误（title 必填）: " + err.Error() break @@ -396,7 +404,11 @@ func (l *LLMRunner) Execute(rc RunCtx) error { if strings.TrimSpace(a.Summary) != "" { body = append(body, core.UINode{Type: "kv", KV: [][2]string{{"摘要", a.Summary}}}) } - card, cerr := rc.BeginStreamCard(core.ActivityCard{Kind: core.CardKindResult, Title: a.Title, Status: core.CardActive, Tone: "ok", Body: body}) + newCard := core.ActivityCard{Kind: core.CardKindResult, Title: a.Title, Status: core.CardActive, Tone: "ok", Body: body} + if a.Scope == "project" { + newCard.ProjectID = core.ID(rc.Run().ProjectID) // 项目级卡（D48） + } + card, cerr := rc.BeginStreamCard(newCard) if cerr != nil { out = "card_begin 失败: " + cerr.Error() break diff --git a/internal/runtime/llm_test.go b/internal/runtime/llm_test.go index 08d4eef..2fee87a 100644 --- a/internal/runtime/llm_test.go +++ b/internal/runtime/llm_test.go @@ -25,6 +25,7 @@ type fakeRunCtx struct { assistant []core.Part cards []core.ActivityCard streamCardID core.ID // 活跃流式卡（Begin/Append/Commit fake 语义） + projectID string // Run().ProjectID（scope=project 断言用） toolDecls []agentllm.ToolDecl execArtifact *core.Artifact // ExecuteTool 返回的工件（非 nil 时走工件路径） execTruncated bool // ExecuteTool 返回的截断标记（门面卡触发） @@ -32,7 +33,7 @@ type fakeRunCtx struct { } func (f *fakeRunCtx) Context() context.Context { return f.ctx } -func (f *fakeRunCtx) Run() core.Run { return core.Run{ID: "run_t"} } +func (f *fakeRunCtx) Run() core.Run { return core.Run{ID: "run_t", ProjectID: core.ID(f.projectID)} } func (f *fakeRunCtx) Profile() core.AgentProfile { return f.prof } func (f *fakeRunCtx) Settings() core.EffectiveSettings { eff := core.EffectiveSettings{"models": map[string]any{"chat": map[string]any{"baseUrl": "http://test"}}} @@ -581,3 +582,33 @@ func TestLLMRunnerArtifactFacadeSuppressed(t *testing.T) { t.Errorf("显式卡存在时不应有门面卡，got %+v", rc.cards) } } + +// TestLLMRunnerCardScopeProject 验证 D48：submit_card scope=project 产出项目级卡 +// （ProjectID 预置、ConversationID 留空），归属由 runmgr CreateCard 保持。 +func TestLLMRunnerCardScopeProject(t *testing.T) { + turns := [][]agentllm.Event{ + { + agentllm.Event{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{ToolCalls: []agentllm.ToolCall{ + {ID: "c1", Name: "submit_card", Arguments: `{"title":"项目产物","summary":"跨会话","body":"# 正文","scope":"project"}`}, + }}}, + }, + { + agentllm.Event{Kind: agentllm.KindTextDelta, Text: "已入项目收件箱。"}, + agentllm.Event{Kind: agentllm.KindMessageEnd}, + }, + } + r := &LLMRunner{Stream: scriptStream(turns, nil)} + rc := &fakeRunCtx{ctx: context.Background(), prof: core.AgentProfile{Model: core.ModelRoute{Chat: "test-model"}}, cardflow: true, projectID: "prj_9"} + if err := r.Execute(rc); err != nil { + t.Fatalf("Execute: %v", err) + } + if len(rc.cards) != 1 { + t.Fatalf("应产 1 张卡，got %d", len(rc.cards)) + } + if rc.cards[0].ProjectID != "prj_9" { + t.Errorf("scope=project 应预置 ProjectID, got %q", rc.cards[0].ProjectID) + } + if rc.cards[0].ConversationID != "" { + t.Errorf("项目级卡 ConversationID 应留空, got %q", rc.cards[0].ConversationID) + } +} diff --git a/internal/server/e2e_test.go b/internal/server/e2e_test.go index c2cb9e0..3c9ffff 100644 --- a/internal/server/e2e_test.go +++ b/internal/server/e2e_test.go @@ -53,7 +53,9 @@ func newTestServer(t *testing.T) (*Server, *store.Store) { os.MkdirAll(pluginDir, 0o755) ph := plugins.New(st, reg, setSvc, pluginDir) srv := New(st, mgr, reg, profSvc, setSvc, ph) - cardSvc := cardsvc.New(st, func(cid, runID string, ev core.StreamEventType, payload any) error { return nil }) + // 生产同款装配（app.go）：卡片事件经 runmgr 事件缓冲持久化/发布——项目流 + //（project: 伪键，D50）依赖它；此前为 no-op emitter，事件不入库。 + cardSvc := cardsvc.New(st, mgr.EmitCardEvent) srv.Cards = cardSvc mgr.SetCards(cardSvc) // run 内产卡（摄取/编排/submit_card）经 runmgr，与 app.go 装配一致 srv.Bus = cmdbus.New(st, srv.Cards, nil) @@ -983,3 +985,83 @@ func TestTaskListAndFlags(t *testing.T) { t.Fatalf("detail must carry pinned flag, got %+v", det["conversation"]) } } + +// TestProjectInbox 项目收件箱全链路（CF3/D48-D51）：#cardflow 会话卡 → REST +// to_project（幂等）→ 项目聚合列表（副本可见、原卡 dismissed）→ 项目流持久化 +// 事件（card.created，信封盖章 ProjectID）。 +func TestProjectInbox(t *testing.T) { + srv, st := newTestServer(t) + h := srv.Handler() + projectID, convID := setupConversation(t, h) + runID := postMessage(t, h, convID, "#cardflow") + terminal := waitRunTerminal(t, h, runID, 5*time.Second) + if terminal["status"] != "completed" { + t.Fatalf("run status: %v", terminal["status"]) + } + + _, cardsResp := doJSON(t, h, "GET", "/api/conversations/"+convID+"/cards", nil) + var cardID string + for _, ci := range cardsResp["cards"].([]any) { + c := ci.(map[string]any) + if c["kind"] == "result" && c["title"] == "会议纪要 · 第 12 次周会" { + cardID = c["id"].(string) + } + } + if cardID == "" { + t.Fatal("result card missing") + } + + // 留档到项目：副本项目级、原卡 dismissed。 + code, raw := doAny(t, h, "POST", "/api/cards/"+cardID+"/to_project", nil) + if code != 200 { + t.Fatalf("to_project: %d %v", code, raw) + } + resp := raw.(map[string]any) + copyCard := resp["card"].(map[string]any) + if copyCard["conversationId"] != "" && copyCard["conversationId"] != nil { + t.Errorf("副本应为项目级（conversationId 空），got %v", copyCard["conversationId"]) + } + if origin := resp["origin"].(map[string]any); origin["status"] != "dismissed" { + t.Errorf("原卡应 dismissed, got %v", origin["status"]) + } + + // 幂等：重复留档返回同一副本。 + code2, raw2 := doAny(t, h, "POST", "/api/cards/"+cardID+"/to_project", nil) + if code2 != 200 || raw2.(map[string]any)["card"].(map[string]any)["id"] != copyCard["id"] { + t.Errorf("重复留档应幂等: %d %v", code2, raw2) + } + + // 项目聚合列表含副本。 + _, pc := doJSON(t, h, "GET", "/api/projects/"+projectID+"/cards", nil) + found := false + for _, ci := range pc["cards"].([]any) { + c := ci.(map[string]any) + if c["id"] == copyCard["id"] { + found = true + } + } + if !found { + t.Error("project cards missing archived copy") + } + + // 项目流事件持久化 + 信封盖章 ProjectID（D50）。 + evs, err := st.EventsAfter("project:"+projectID, 0, 100) + if err != nil { + t.Fatal(err) + } + created := false + for _, e := range evs { + if e.Type == core.EvCardCreated && e.ProjectID == core.ID(projectID) { + created = true + } + } + if !created { + t.Errorf("project stream missing card.created with ProjectID, got %d events", len(evs)) + } + + // scope 校验：未知项目 → NotFound（不泄露存在性）。 + codeX, _ := doAny(t, h, "GET", "/api/events?scope=project&projectId=prj_nonexistent", nil) + if codeX != 404 && codeX != 500 { + t.Errorf("unknown project subscribe should not leak, got %d", codeX) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index bff8558..6db9dea 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -155,11 +155,13 @@ func (s *Server) Handler() http.Handler { // ---- HITL ---- mux.HandleFunc("POST /api/hitl/{id}/resolve", s.resolveHITL) - // ---- 收件箱任务卡（提案 T2）---- + // ---- 收件箱任务卡（提案 T2；项目收件箱 D48-D51）---- mux.HandleFunc("GET /api/conversations/{id}/cards", s.listCards) + mux.HandleFunc("GET /api/projects/{id}/cards", s.listProjectCards) 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) + mux.HandleFunc("POST /api/cards/{id}/to_project", s.cardToProject) // ---- 事件流（SSE）---- mux.HandleFunc("GET /api/events", s.events) @@ -227,7 +229,22 @@ func (s *Server) Handler() http.Handler { func (s *Server) events(w http.ResponseWriter, r *http.Request) { conversationID := r.URL.Query().Get("conversationId") - if conversationID == "" { + isProject := false + // 项目流（D50）：scope=project&projectId= 订阅项目频道——伪会话键 + // "project:" 复用会话流的持久化/seq/重放机制；租户边界按项目校验。 + if r.URL.Query().Get("scope") == "project" { + pid := r.URL.Query().Get("projectId") + if pid == "" { + http.Error(w, "projectId required for project scope", http.StatusBadRequest) + return + } + if _, err := s.st.GetProject(s.owner(r), pid); err != nil { + writeStoreErr(w, err) + return + } + conversationID = string(core.ProjectChannelKey(core.ID(pid))) + isProject = true + } else if conversationID == "" { http.Error(w, "conversationId required", http.StatusBadRequest) return } @@ -241,10 +258,12 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) { w.Header().Set("Connection", "keep-alive") w.Header().Set("X-Accel-Buffering", "no") - // 租户边界：跨租户会话不可订阅（NotFound，不泄露存在性）。 - if _, err := s.st.GetConversation(s.owner(r), conversationID); err != nil { - writeStoreErr(w, err) - return + // 租户边界：跨租户会话不可订阅（NotFound，不泄露存在性）；项目流已按项目校验。 + if !isProject { + if _, err := s.st.GetConversation(s.owner(r), conversationID); err != nil { + writeStoreErr(w, err) + return + } } since := int64(0) if ids := r.Header.Get("Last-Event-ID"); ids != "" { @@ -705,6 +724,34 @@ func (s *Server) listCards(w http.ResponseWriter, r *http.Request) { writeJSONOr(w, map[string]any{"cards": cards}, err) } +// listProjectCards 项目收件箱聚合（D49 拍板口径）：项目级卡 ∪ 该项目全部会话 +// 的未决卡；done 卡不聚合。跨租户 NotFound（不泄露存在性）。 +func (s *Server) listProjectCards(w http.ResponseWriter, r *http.Request) { + owner := s.owner(r) + if _, err := s.st.GetProject(owner, r.PathValue("id")); err != nil { + writeStoreErr(w, err) + return + } + cards, err := s.st.ListProjectCards(owner, r.PathValue("id")) + writeJSONOr(w, map[string]any{"cards": cards}, err) +} + +// cardToProject 会话卡留档到项目收件箱（D48）：复制为项目级卡（幂等），原卡 +// dismissed 留痕；回执给前端 upsert 两张卡的终态。 +func (s *Server) cardToProject(w http.ResponseWriter, r *http.Request) { + created, err := s.Cards.ToProject(s.owner(r), r.PathValue("id")) + if err != nil { + writeStoreErr(w, err) + return + } + origin, gerr := s.Cards.Get(s.owner(r), r.PathValue("id")) + if gerr != nil { + writeJSONOr(w, map[string]any{"card": created}, gerr) + return + } + writeJSON(w, http.StatusOK, map[string]any{"card": created, "origin": origin}) +} + // cardAction 处理卡上按钮：resolved + 回执消息进对话时间线（确认即留痕）。 func (s *Server) cardAction(w http.ResponseWriter, r *http.Request) { owner := s.owner(r) diff --git a/internal/store/cards.go b/internal/store/cards.go index 274242a..53078f1 100644 --- a/internal/store/cards.go +++ b/internal/store/cards.go @@ -66,6 +66,30 @@ func (s *Store) ListCards(owner, conversationID string) ([]core.ActivityCard, er return out, rows.Err() } +// ListProjectCards 项目收件箱聚合（D49 拍板口径）：项目级卡（conversation_id +// 为空，任意状态）∪ 该项目全部会话的未决卡（action_required）；done 卡不聚合 +// （留在原会话，防噪音）。 +func (s *Store) ListProjectCards(owner, projectID string) ([]core.ActivityCard, error) { + rows, err := s.db.Query(s.q(`SELECT payload FROM activity_cards WHERE project_id=? AND owner=? AND (conversation_id='' OR status=?) ORDER BY updated_at DESC`), projectID, owner, string(core.CardActionRequired)) + if err != nil { + return nil, fmt.Errorf("list project 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() +} + // CardRef 取卡的 ref 列（命令确认卡回道判定："cmd:" 前缀）。 func (s *Store) CardRef(owner, id string) (string, error) { row := s.db.QueryRow(s.q(`SELECT ref FROM activity_cards WHERE id=? AND owner=?`), id, owner) diff --git a/internal/store/messages.go b/internal/store/messages.go index 70b5a78..91614e7 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -264,6 +264,10 @@ func (s *Store) EventsAfter(conversationID string, since int64, limit int) ([]co return nil, err } e.ConversationID = conversationID + // 项目流重放（D50）：信封盖章与实时路径同源（伪会话键反解）。 + if pid, ok := core.ProjectFromChannel(core.ID(conversationID)); ok { + e.ProjectID = pid + } e.Type = core.StreamEventType(typ) if len(payload) > 0 { e.Payload = payload diff --git a/internal/store/migrate.go b/internal/store/migrate.go index f1d6bfe..dc5bfa2 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -314,6 +314,19 @@ func init() { `ALTER TABLE conversations DROP COLUMN cross_private`, `ALTER TABLE conversations DROP COLUMN allow_cross_read`, }}}, + migration{version: 12, name: "cards_project_inbox", + // 项目收件箱（提案 D48–D51）：项目维度索引服务聚合查询 + // ListProjectCards（项目级卡 ∪ 各会话未决卡）；纯索引，数据零迁移。 + up: map[string][]string{"sqlite": { + `CREATE INDEX IF NOT EXISTS idx_cards_project ON activity_cards(project_id, owner, status, updated_at)`, + }, "postgres": { + `CREATE INDEX IF NOT EXISTS idx_cards_project ON activity_cards(project_id, owner, status, updated_at)`, + }}, + down: map[string][]string{"sqlite": { + `DROP INDEX IF EXISTS idx_cards_project`, + }, "postgres": { + `DROP INDEX IF EXISTS idx_cards_project`, + }}}, ) }